William/v2 server

README.md

@@ -4,6 +4,8 @@
 
 This server sends push notifications to Edge client apps. It contains an HTTP server that clients can use to register for notifications, and a background process that checks for price changes and actually sends the messages.
 
+The docs folder has can find [an example of how to use the v2 API](./docs/demo.ts).
+
 ## Setup
 
 This server requires a working copies of Node.js, Yarn, PM2, and CouchDB. We also recommend using Caddy to terminate SSL connections.

docs/demo.ts

@@ -0,0 +1,113 @@
+import { asJSON, asMaybe, asObject, asString, uncleaner } from 'cleaners'
+import fetch from 'node-fetch'
+import { base64 } from 'rfc4648'
+
+import {
+  asDeviceUpdatePayload,
+  asLoginUpdatePayload,
+  asPushRequestBody
+} from '../src/types/pushApiTypes'
+
+// We are going to use uncleaners to type-check our payloads:
+const wasPushRequestBody = uncleaner(asPushRequestBody)
+const wasDeviceUpdatePayload = uncleaner(asDeviceUpdatePayload)
+const wasLoginUpdatePayload = uncleaner(asLoginUpdatePayload)
+
+/**
+ * Failed requests usually return this as their body.
+ */
+const asErrorBody = asJSON(
+  asObject({
+    error: asString
+  })
+)
+
+const server = 'http://127.0.0.1:8001'
+const apiKey = 'demo-api-key'
+const deviceId = 'example-device'
+// Non-existing account for demo purposes:
+const loginId = base64.parse('O8vz73AA76MejE3bCh6t6F2Lr7TLfQVB2+Tm0sC35tc=')
+
+/**
+ * All push server HTTP methods use "POST" with JSON.
+ */
+async function postJson(uri: string, body: unknown): Promise<unknown> {
+  console.log(JSON.stringify(body, null, 1))
+  const response = await fetch(uri, {
+    body: JSON.stringify(body),
+    headers: {
+      accept: 'application/json',
+      'content-type': 'application/json'
+    },
+    method: 'POST'
+  })
+  if (!response.ok) {
+    const error = asMaybe(asErrorBody)(await response.text())
+    let message = `POST ${uri} returned ${response.status}`
+    if (error != null) message += `: ${error.error}`
+    throw new Error(message)
+  }
+  return await response.json()
+}
+
+async function main(): Promise<void> {
+  // Create a device:
+  await postJson(
+    `${server}/v2/device/update/`,
+    wasPushRequestBody({
+      apiKey,
+      deviceId,
+      data: wasDeviceUpdatePayload({ loginIds: [loginId] })
+    })
+  )
+  console.log(`Updated device "${deviceId}"`)
+
+  // Grab the device status:
+  console.log(
+    await postJson(
+      `${server}/v2/device/`,
+      wasPushRequestBody({ apiKey, deviceId })
+    )
+  )
+
+  // Subscribe the user to a price change:
+  await postJson(
+    `${server}/v2/login/update/`,
+    wasPushRequestBody({
+      apiKey,
+      deviceId,
+      loginId,
+      data: wasLoginUpdatePayload({
+        createEvents: [
+          {
+            eventId: 'demo-event',
+            pushMessage: {
+              title: 'Example title',
+              body: 'Example body',
+              data: { what: 'happened' }
+            },
+            trigger: {
+              type: 'price-level',
+              currencyPair: 'BTC-USD',
+              aboveRate: 50000
+            }
+          }
+        ]
+      })
+    })
+  )
+  console.log(`Updated login "${base64.stringify(loginId)}"`)
+
+  // Grab the login status:
+  console.log(
+    await postJson(
+      `${server}/v2/login/`,
+      wasPushRequestBody({ apiKey, deviceId, loginId })
+    )
+  )
+}
+
+main().catch(error => {
+  console.error(String(error))
+  process.exitCode = 1
+})

package.json

@@ -10,12 +10,13 @@
   "scripts": {
     "build": "sucrase -q -t typescript,imports -d ./lib ./src",
     "clean": "rimraf lib",
+    "cli": "node -r sucrase/register src/cli/cli.ts",
+    "demo": "node -r sucrase/register docs/demo.ts",
     "fix": "yarn-deduplicate && eslint . --fix",
     "lint": "eslint .",
     "precommit": "lint-staged && npm-run-all types prepare",
     "prepare": "husky install && npm-run-all clean build",
     "start-server": "node -r sucrase/register src/server/index.ts",
-    "start-price": "node -r sucrase/register src/price-script/index.ts",
     "types": "tsc",
     "verify": "npm-run-all lint types prepare"
   },
@@ -25,15 +26,18 @@
   "dependencies": {
     "@pm2/io": "^4.3.5",
     "axios": "^0.21.2",
+    "base-x": "^3.0.8",
     "body-parser": "^1.20.0",
     "cleaner-config": "^0.1.7",
     "cleaners": "^0.3.12",
+    "clipanion": "^3.2.0-rc.11",
     "cors": "^2.8.5",
     "edge-server-tools": "^0.2.15",
     "express": "^4.18.1",
     "firebase-admin": "^8.12.1",
     "nano": "^9.0.5",
     "node-fetch": "^2.6.7",
+    "rfc4648": "^1.5.2",
     "serverlet": "^0.1.1"
   },
   "devDependencies": {

pm2.json

@@ -5,12 +5,6 @@
       "error_file": "/var/log/pm2/pushErrors.log",
       "out_file": "/var/log/pm2/pushServer.log",
       "script": "./lib/server/index.js"
-    },
-    {
-      "name": "priceDaemon",
-      "error_file": "/var/log/pm2/priceErrors.log",
-      "out_file": "/var/log/pm2/priceDaemon.log",
-      "script": "./lib/price-script/index.js"
     }
   ]
 }

src/cli/cli.ts

@@ -0,0 +1,37 @@
+import { Builtins, Cli } from 'clipanion'
+import nano from 'nano'
+
+import packageJson from '../../package.json'
+import { setupDatabases } from '../db/couchSetup'
+import { serverConfig } from '../serverConfig'
+import { ServerContext } from './cliTools'
+import { GetDevice } from './commands/getDevice'
+import { MigrateDevices } from './commands/migrateDevices'
+
+async function main(): Promise<void> {
+  const connection = nano(serverConfig.couchUri)
+  await setupDatabases(connection, true)
+
+  const context: ServerContext = {
+    ...Cli.defaultContext,
+    connection
+  }
+
+  const cli = new Cli({
+    binaryLabel: 'push-server',
+    binaryName: 'yarn cli',
+    binaryVersion: packageJson.version
+  })
+
+  cli.register(Builtins.HelpCommand)
+  cli.register(Builtins.VersionCommand)
+
+  // Our commands:
+  cli.register(GetDevice)
+  cli.register(MigrateDevices)
+
+  const args = process.argv.slice(2)
+  await cli.runExit(args, context)
+}
+
+main().catch(error => console.error(error))

src/cli/cliTools.ts

@@ -0,0 +1,56 @@
+import { asString, Cleaner } from 'cleaners'
+import { BaseContext } from 'clipanion'
+import { ServerScope } from 'nano'
+import { base16, base64 } from 'rfc4648'
+import { Writable } from 'stream'
+
+export interface ServerContext extends BaseContext {
+  connection: ServerScope
+}
+
+export const asNumberString: Cleaner<number> = raw => {
+  const out = Number(asString(raw))
+  if (Number.isNaN(out)) throw new TypeError('Expected a number string')
+  return out
+}
+
+export function prettify(data: unknown): string {
+  return JSON.stringify(
+    data,
+    (key, value) =>
+      value instanceof Uint8Array
+        ? /_hex$/.test(key)
+          ? base16.stringify(value)
+          : base64.stringify(value)
+        : value,
+    1
+  )
+}
+
+/**
+ * For long-running database tasks,
+ * print a heartbeat to stderr every few seconds.
+ */
+export function makeHeartbeat(
+  stderr: Writable,
+  opts: { logSeconds?: number } = {}
+): (item?: string) => void {
+  const { logSeconds = 10 } = opts
+
+  const start = Date.now()
+  let nextLog = start + 1000 * logSeconds
+
+  let count = 0
+  return item => {
+    ++count
+    const now = Date.now()
+    if (now < nextLog) return
+
+    nextLog += 1000 * logSeconds
+    const seconds = (now - start) / 1000
+    let out = `${seconds.toFixed(2)}s, ${count} rows`
+    if (item != null) out += `, ${item}`
+    out += '\n'
+    stderr.write(out)
+  }
+}

src/cli/commands/getDevice.ts

@@ -0,0 +1,28 @@
+import { asString } from 'cleaners'
+import { Command, Option } from 'clipanion'
+
+import { getDeviceById } from '../../db/couchDevices'
+import { prettify, ServerContext } from '../cliTools'
+
+export class GetDevice extends Command<ServerContext> {
+  static paths = [['get-device']]
+  static usage = { description: "Shows a device's status" }
+
+  deviceId = Option.String({ name: 'deviceId', required: true })
+
+  async execute(): Promise<number> {
+    const { connection, stderr, stdout } = this.context
+    const deviceId = asString(this.deviceId)
+
+    const now = new Date()
+
+    const deviceRow = await getDeviceById(connection, deviceId, now)
+    if (!deviceRow.exists) {
+      stderr.write(`No device ${deviceId}\n`)
+      return 1
+    }
+    stdout.write(prettify(deviceRow.device))
+
+    return 0
+  }
+}

src/cli/commands/migrateDevices.ts

@@ -0,0 +1,71 @@
+import { Command } from 'clipanion'
+import { asMaybeNotFoundError, viewToStream } from 'edge-server-tools'
+import { base64 } from 'rfc4648'
+
+import { getDeviceById } from '../../db/couchDevices'
+import { syncedSettings } from '../../db/couchSettings'
+import { asLegacyDevice } from '../../models/Device'
+import { asLegacyUser } from '../../models/User'
+import { base58 } from '../../util/base58'
+import { verifyData } from '../../util/verifyData'
+import { makeHeartbeat, ServerContext } from '../cliTools'
+
+export class MigrateDevices extends Command<ServerContext> {
+  static paths = [['migrate-devices']]
+  static usage = { description: 'Migrate v1 devices to the v2 database' }
+
+  async execute(): Promise<number> {
+    const { connection, stderr, stdout } = this.context
+    const legacyUserDb = connection.use('db_user_settings')
+    const legacyDeviceDb = connection.use('db_devices')
+
+    const now = new Date()
+    const heartbeat = makeHeartbeat(stdout)
+
+    for await (const raw of viewToStream(
+      async params => await legacyUserDb.list(params)
+    )) {
+      try {
+        const clean = asLegacyUser(raw)
+        const loginId = base58.parse(clean.id)
+        const { devices } = clean.doc
+
+        for (const deviceId of Object.keys(devices)) {
+          if (!devices[deviceId]) continue
+
+          const raw = await legacyDeviceDb.get(deviceId).catch(error => {
+            if (asMaybeNotFoundError(error) != null) return
+            throw error
+          })
+          if (raw == null) continue
+          const clean = asLegacyDevice(raw)
+
+          const deviceRow = await getDeviceById(connection, deviceId, now)
+          const { device } = deviceRow
+          if (deviceRow.exists) {
+            // Add the user to the list:
+            if (device.loginIds.find(row => verifyData(loginId, row)) == null) {
+              device.loginIds = [...device.loginIds, loginId]
+              await deviceRow.save()
+            }
+          } else {
+            // Create the device:
+            device.deviceToken = clean.doc.tokenId
+            if (clean.doc.appId === '') {
+              device.apiKey = syncedSettings.doc.apiKeys[0].apiKey
+            }
+            device.loginIds = [loginId]
+            await deviceRow.save()
+          }
+        }
+
+        heartbeat(base64.stringify(loginId))
+      } catch (error) {
+        const id: string = (raw as any)._id
+        stderr.write(`Could not migrate user ${id} ${String(error)}\n`)
+      }
+    }
+
+    return 0
+  }
+}