From f6c7d28429301a08826e7830f8c2f3ccf036116c Mon Sep 17 00:00:00 2001 From: William Swanson Date: Thu, 28 Jul 2022 15:11:21 -0700 Subject: [PATCH 1/8] Implement v2 databases and HTTP endpoints --- package.json | 1 + src/db/couchDevices.ts | 178 +++++++++++++++ src/db/couchPushEvents.ts | 333 ++++++++++++++++++++++++++++ src/db/couchSetup.ts | 6 +- src/server/middleware/withDevice.ts | 59 +++++ src/server/routes/deviceRoutes.ts | 70 ++++++ src/server/routes/loginRoutes.ts | 50 +++++ src/server/urls.ts | 15 ++ src/types/pushApiTypes.ts | 149 +++++++++++++ src/types/pushCleaners.ts | 87 ++++++++ src/types/pushTypes.ts | 103 +++++++++ src/types/requestTypes.ts | 9 + yarn.lock | 5 + 13 files changed, 1064 insertions(+), 1 deletion(-) create mode 100644 src/db/couchDevices.ts create mode 100644 src/db/couchPushEvents.ts create mode 100644 src/server/middleware/withDevice.ts create mode 100644 src/server/routes/deviceRoutes.ts create mode 100644 src/server/routes/loginRoutes.ts create mode 100644 src/types/pushApiTypes.ts create mode 100644 src/types/pushCleaners.ts diff --git a/package.json b/package.json index 187f2bc..0ff884b 100644 --- a/package.json +++ b/package.json @@ -34,6 +34,7 @@ "firebase-admin": "^8.12.1", "nano": "^9.0.5", "node-fetch": "^2.6.7", + "rfc4648": "^1.5.2", "serverlet": "^0.1.1" }, "devDependencies": { diff --git a/src/db/couchDevices.ts b/src/db/couchDevices.ts new file mode 100644 index 0000000..53f45c0 --- /dev/null +++ b/src/db/couchDevices.ts @@ -0,0 +1,178 @@ +import { + asArray, + asBoolean, + asDate, + asObject, + asOptional, + asString, + uncleaner +} from 'cleaners' +import { + asCouchDoc, + asMaybeConflictError, + asMaybeNotFoundError, + CouchDoc, + DatabaseSetup, + makeJsDesign +} from 'edge-server-tools' +import { DocumentScope, ServerScope } from 'nano' +import { base64 } from 'rfc4648' + +import { asBase64 } from '../types/pushCleaners' +import { Device } from '../types/pushTypes' + +/** + * A device returned from the database. + * To make changes, edit the `device` object, then call `save`. + */ +export interface DeviceRow { + device: Device + exists: boolean + save: () => Promise +} + +type CouchDevice = Omit + +/** + * A single phone or other devIce, as stored in Couch. + * The document ID is the deviceId. + */ +export const asCouchDevice = asCouchDoc( + asObject({ + created: asDate, + + // Status: + apiKey: asOptional(asString), + deviceToken: asOptional(asString), + ignorePriceChanges: asOptional(asBoolean, false), + loginIds: asArray(asBase64), + visited: asDate + }) +) +const wasCouchDevice = uncleaner(asCouchDevice) + +/** + * Looks up devices that contain a particular login. + */ +const loginIdDesign = makeJsDesign('loginId', ({ emit }) => ({ + map: function (doc) { + for (let i = 0; i < doc.loginIds.length; ++i) { + emit(doc.loginIds[i], null) + } + } +})) + +export const couchDevicesSetup: DatabaseSetup = { + name: 'push-devices', + documents: { + '_design/loginId': loginIdDesign + } +} + +/** + * Looks up a device by its id. + * If the device does not exist in the database, creates a fresh row. + */ +export async function getDeviceById( + connection: ServerScope, + deviceId: string, + date: Date +): Promise { + const db = connection.use(couchDevicesSetup.name) + + const emptyDevice = { + id: deviceId, + doc: { + created: date, + apiKey: undefined, + deviceToken: undefined, + ignorePriceChanges: false, + loginIds: [], + visited: date + } + } + if (deviceId === '') return makeDeviceRow(db, emptyDevice) + + const raw = await db.get(deviceId).catch(error => { + if (asMaybeNotFoundError(error) != null) return + throw error + }) + + if (raw == null) return makeDeviceRow(db, emptyDevice) + return makeDeviceRow(db, asCouchDevice(raw)) +} + +/** + * Finds all the devices that have logged into this account. + */ +export async function getDevicesByLoginId( + connection: ServerScope, + loginId: Uint8Array +): Promise { + const db = connection.use(couchDevicesSetup.name) + const response = await db.view('loginId', 'loginId', { + include_docs: true, + key: base64.stringify(loginId) + }) + return response.rows.map(row => makeDeviceRow(db, asCouchDevice(row.doc))) +} + +function makeDeviceRow( + db: DocumentScope, + clean: CouchDoc +): DeviceRow { + const device = { ...clean.doc, deviceId: clean.id } + let rev = clean.rev + let base = { ...device } + + return { + device, + + get exists(): boolean { + return rev != null + }, + + async save(): Promise { + while (true) { + // Write to the database: + const doc: CouchDoc = { + doc: device, + id: clean.id, + rev + } + const response = await db.insert(wasCouchDevice(doc)).catch(error => { + if (asMaybeConflictError(error) == null) throw error + }) + + // If that worked, the merged document is now the latest: + if (response?.ok === true) { + rev = response.rev + base = { ...device } + return + } + + // Something went wrong, so grab the latest remote document: + const raw = await db.get(doc.id) + const remote = asCouchDevice(raw) + rev = remote.rev + + // If we don't have local edits, take the remote field: + if (device.apiKey === base.apiKey) { + device.apiKey = remote.doc.apiKey + } + if (device.deviceToken === base.deviceToken) { + device.deviceToken = remote.doc.deviceToken + } + if (device.ignorePriceChanges === base.ignorePriceChanges) { + device.ignorePriceChanges = remote.doc.ignorePriceChanges + } + if (device.loginIds === base.loginIds) { + device.loginIds = remote.doc.loginIds + } + if (device.visited < remote.doc.visited) { + device.visited = remote.doc.visited + } + } + } + } +} diff --git a/src/db/couchPushEvents.ts b/src/db/couchPushEvents.ts new file mode 100644 index 0000000..ce273f6 --- /dev/null +++ b/src/db/couchPushEvents.ts @@ -0,0 +1,333 @@ +import { + asArray, + asDate, + asEither, + asNull, + asNumber, + asObject, + asOptional, + asString, + uncleaner +} from 'cleaners' +import { + asCouchDoc, + asMaybeConflictError, + CouchDoc, + DatabaseSetup, + makeJsDesign, + viewToStream +} from 'edge-server-tools' +import { DocumentScope, ServerScope } from 'nano' +import { base64 } from 'rfc4648' + +import { NewPushEvent } from '../types/pushApiTypes' +import { + asBase64, + asBroadcastTx, + asPushEventState, + asPushMessage, + asPushTrigger +} from '../types/pushCleaners' +import { PushEvent } from '../types/pushTypes' + +/** + * An event returned from the database. + * To make changes, edit the `event` object, then call `save`. + */ +export interface PushEventRow { + event: PushEvent + save: () => Promise +} + +type CouchPushEvent = Omit + +/** + * A push event, as stored in Couch. + * The document ID is the creation date. + */ +export const asCouchPushEvent = asCouchDoc( + asObject({ + eventId: asString, // Not the document id! + deviceId: asOptional(asString), + loginId: asOptional(asBase64), + + // Event: + broadcastTxs: asOptional(asArray(asBroadcastTx)), + pushMessage: asOptional(asPushMessage), + trigger: asPushTrigger, + + // Status: + broadcastTxErrors: asOptional(asArray(asEither(asString, asNull))), + pushMessageEmits: asOptional(asNumber), + pushMessageFails: asOptional(asNumber), + state: asPushEventState, + triggered: asOptional(asDate) + }) +) +const wasCouchPushEvent = uncleaner(asCouchPushEvent) + +/** + * Looks up events attached to devices. + */ +const deviceIdDesign = makeJsDesign('deviceId', ({ emit }) => ({ + map: function (doc) { + if (doc.deviceId == null) return + if (doc.state === 'cancelled') return + if (doc.state === 'hidden') return + emit(doc.deviceId, null) + } +})) + +/** + * Looks up events attached to logins. + */ +const loginIdDesign = makeJsDesign('loginId', ({ emit }) => ({ + map: function (doc) { + if (doc.loginId == null) return + if (doc.state === 'cancelled') return + if (doc.state === 'hidden') return + emit(doc.loginId, null) + } +})) + +/** + * Looks up active address-balance events. + */ +const addressBalanceDesign = makeJsDesign('address-balance', ({ emit }) => ({ + map: function (doc) { + if (doc.trigger == null) return + if (doc.trigger.type !== 'address-balance') return + if (doc.state !== 'waiting') return + emit(doc._id, null) + } +})) + +/** + * Looks up active price-related events. + */ +const priceDesign = makeJsDesign('price', ({ emit }) => ({ + map: function (doc) { + if (doc.trigger == null) return + const type = doc.trigger.type + if (type !== 'price-change' && type !== 'price-level') return + if (doc.state !== 'waiting') return + emit(doc._id, null) + } +})) + +/** + * Looks up active price-related events. + */ +const txConfirmDesign = makeJsDesign('tx-confirm', ({ emit }) => ({ + map: function (doc) { + if (doc.trigger == null) return + if (doc.trigger.type !== 'tx-confirm') return + if (doc.state !== 'waiting') return + emit(doc._id, null) + } +})) + +export const couchEventsSetup: DatabaseSetup = { + name: 'push-events', + + documents: { + '_design/address-balance': addressBalanceDesign, + '_design/deviceId': deviceIdDesign, + '_design/loginId': loginIdDesign, + '_design/price': priceDesign, + '_design/tx-confirm': txConfirmDesign + } +} + +export async function addEvent( + connection: ServerScope, + event: PushEvent, + created: Date +): Promise { + const db = connection.use(couchEventsSetup.name) + try { + await db.insert( + wasCouchPushEvent({ + doc: event, + id: created.toISOString() + }) + ) + } catch (error) { + if (asMaybeConflictError(error) == null) throw error + await addEvent(connection, event, new Date(created.valueOf() + 1)) + } +} + +export async function adjustEvents( + connection: ServerScope, + opts: { + date: Date + deviceId?: string + loginId?: Uint8Array + createEvents?: NewPushEvent[] + removeEvents?: string[] + } +): Promise { + const { date, deviceId, loginId, createEvents = [], removeEvents = [] } = opts + + // Load existing events: + const eventRows = + deviceId != null + ? await getEventsByDeviceId(connection, deviceId) + : loginId != null + ? await getEventsByLoginId(connection, loginId) + : [] + + // Remove events from the array: + const removeSet = new Set(removeEvents) + for (const event of createEvents) removeSet.add(event.eventId) + const out: PushEvent[] = eventRows + .map(row => row.event) + .filter(event => !removeSet.has(event.eventId)) + + // Perform the deletion on the database: + for (const row of eventRows) { + if (!removeSet.has(row.event.eventId)) continue + if (row.event.state === 'waiting') row.event.state = 'cancelled' + else row.event.state = 'hidden' + await row.save() + } + + // Add new events: + for (const create of createEvents) { + const event: PushEvent = { + ...create, + created: date, + deviceId, + loginId, + state: 'waiting' + } + await addEvent(connection, event, date) + out.push(event) + } + + return out +} + +export async function getEventsByDeviceId( + connection: ServerScope, + deviceId: string +): Promise { + const db = connection.use(couchEventsSetup.name) + const response = await db.view('deviceId', 'deviceId', { + include_docs: true, + key: deviceId + }) + return response.rows.map(row => + makePushEventRow(db, asCouchPushEvent(row.doc)) + ) +} + +export async function getEventsByLoginId( + connection: ServerScope, + loginId: Uint8Array +): Promise { + const db = connection.use(couchEventsSetup.name) + const response = await db.view('loginId', 'loginId', { + include_docs: true, + key: base64.stringify(loginId) + }) + return response.rows.map(row => + makePushEventRow(db, asCouchPushEvent(row.doc)) + ) +} + +export async function* streamAddressBalanceEvents( + connection: ServerScope +): AsyncIterableIterator { + const db = connection.use(couchEventsSetup.name) + const stream = viewToStream(async params => { + return await db.view('address-balance', 'address-balance', params) + }) + for await (const raw of stream) { + yield makePushEventRow(db, asCouchPushEvent(raw)) + } +} + +export async function* streamPriceEvents( + connection: ServerScope +): AsyncIterableIterator { + const db = connection.use(couchEventsSetup.name) + const stream = viewToStream(async params => { + return await db.view('price', 'price', params) + }) + for await (const raw of stream) { + yield makePushEventRow(db, asCouchPushEvent(raw)) + } +} + +export async function* streamTxConfirmEvents( + connection: ServerScope +): AsyncIterableIterator { + const db = connection.use(couchEventsSetup.name) + const stream = viewToStream(async params => { + return await db.view('tx-confirm', 'tx-confirm', params) + }) + for await (const raw of stream) { + yield makePushEventRow(db, asCouchPushEvent(raw)) + } +} + +function makePushEventRow( + db: DocumentScope, + clean: CouchDoc +): PushEventRow { + const event = { ...clean.doc, created: new Date(clean.id) } + let rev = clean.rev + let base = { ...event } + + return { + event, + + async save(): Promise { + while (true) { + // Write to the database: + const doc: CouchDoc = { + doc: event, + id: clean.id, + rev + } + const response = await db + .insert(wasCouchPushEvent(doc)) + .catch(error => { + if (asMaybeConflictError(error) == null) throw error + }) + + // If that worked, the merged document is now the latest: + if (response?.ok === true) { + rev = response.rev + base = { ...event } + return + } + + // Something went wrong, so grab the latest remote document: + const raw = await db.get(doc.id) + const remote = asCouchPushEvent(raw) + rev = remote.rev + + // If we don't have local edits, take the remote field: + if (event.broadcastTxErrors === base.broadcastTxErrors) { + event.broadcastTxErrors = remote.doc.broadcastTxErrors + } + if ( + event.pushMessageEmits === base.pushMessageEmits && + event.pushMessageFails === base.pushMessageFails + ) { + event.pushMessageEmits = remote.doc.pushMessageEmits + event.pushMessageFails = remote.doc.pushMessageFails + } + if (event.state === base.state) { + event.state = remote.doc.state + } + if (event.triggered === base.triggered) { + event.triggered = remote.doc.triggered + } + } + } + } +} diff --git a/src/db/couchSetup.ts b/src/db/couchSetup.ts index d158e23..452e31b 100644 --- a/src/db/couchSetup.ts +++ b/src/db/couchSetup.ts @@ -7,6 +7,8 @@ import { ServerScope } from 'nano' import { serverConfig } from '../serverConfig' import { couchApiKeysSetup } from './couchApiKeys' +import { couchDevicesSetup } from './couchDevices' +import { couchEventsSetup } from './couchPushEvents' import { settingsSetup, syncedReplicators } from './couchSettings' // --------------------------------------------------------------------------- @@ -43,8 +45,10 @@ export async function setupDatabases( await setupDatabase(connection, settingsSetup, options) await Promise.all([ setupDatabase(connection, couchApiKeysSetup, options), - setupDatabase(connection, thresholdsSetup, options), + setupDatabase(connection, couchDevicesSetup, options), + setupDatabase(connection, couchEventsSetup, options), setupDatabase(connection, devicesSetup, options), + setupDatabase(connection, thresholdsSetup, options), setupDatabase(connection, usersSetup, options) ]) } diff --git a/src/server/middleware/withDevice.ts b/src/server/middleware/withDevice.ts new file mode 100644 index 0000000..ded4b2f --- /dev/null +++ b/src/server/middleware/withDevice.ts @@ -0,0 +1,59 @@ +import { Serverlet } from 'serverlet' + +import { getApiKeyByKey } from '../../db/couchApiKeys' +import { getDeviceById } from '../../db/couchDevices' +import { asPushRequestBody } from '../../types/pushApiTypes' +import { DbRequest, DeviceRequest } from '../../types/requestTypes' +import { errorResponse } from '../../types/responseTypes' +import { checkPayload } from '../../util/checkPayload' + +/** + * Parses the request payload and looks up the device. + * Legacy routes do not use this one. + */ +export const withDevice = + (server: Serverlet): Serverlet => + async request => { + const { connection, date, log, req } = request + + // Parse the common request body: + const checkedBody = checkPayload(asPushRequestBody, req.body) + if (checkedBody.error != null) return checkedBody.error + const body = checkedBody.clean + + // Look up the key in the database: + const apiKey = await log.debugTime( + 'getApiKeyByKey', + getApiKeyByKey(connection, body.apiKey) + ) + if (apiKey == null) { + return errorResponse('Incorrect API key', { status: 401 }) + } + + // Look up the device in the database, or get a dummy row: + const deviceRow = await log.debugTime( + 'getDeviceById', + getDeviceById(connection, body.deviceId, date) + ) + if (body.apiKey != null) { + deviceRow.device.apiKey = body.apiKey + } + if (body.deviceToken != null) { + deviceRow.device.deviceToken = body.deviceToken + } + deviceRow.device.visited = date + + // Pass that along: + const result = await server({ + ...request, + apiKey, + deviceRow, + loginId: body.loginId, + payload: body.data + }) + + // Flush any changes (such as the visited date): + await deviceRow.save() + + return result + } diff --git a/src/server/routes/deviceRoutes.ts b/src/server/routes/deviceRoutes.ts new file mode 100644 index 0000000..aeb1185 --- /dev/null +++ b/src/server/routes/deviceRoutes.ts @@ -0,0 +1,70 @@ +import { uncleaner } from 'cleaners' + +import { adjustEvents, getEventsByDeviceId } from '../../db/couchPushEvents' +import { + asDevicePayload, + asDeviceUpdatePayload +} from '../../types/pushApiTypes' +import { jsonResponse } from '../../types/responseTypes' +import { checkPayload } from '../../util/checkPayload' +import { withDevice } from '../middleware/withDevice' + +const wasDevicePayload = uncleaner(asDevicePayload) + +/** + * POST /v2/device + */ +export const deviceFetchRoute = withDevice(async request => { + const { + connection, + deviceRow: { device } + } = request + + const eventRows = await getEventsByDeviceId(connection, device.deviceId) + + return jsonResponse( + wasDevicePayload({ + events: eventRows.map(row => row.event), + ignorePriceChanges: device.ignorePriceChanges, + loginIds: device.loginIds + }) + ) +}) + +/** + * POST /v2/device/update + */ +export const deviceUpdateRoute = withDevice(async request => { + const { + connection, + date, + deviceRow: { device }, + payload + } = request + + const checked = checkPayload(asDeviceUpdatePayload, payload) + if (checked.error != null) return checked.error + const { loginIds, createEvents, ignorePriceChanges, removeEvents } = + checked.clean + + if (ignorePriceChanges != null) { + device.ignorePriceChanges = ignorePriceChanges + } + if (loginIds != null) { + device.loginIds = loginIds + } + const events = await adjustEvents(connection, { + date, + deviceId: device.deviceId, + createEvents, + removeEvents + }) + + return jsonResponse( + wasDevicePayload({ + events, + ignorePriceChanges: device.ignorePriceChanges, + loginIds: device.loginIds + }) + ) +}) diff --git a/src/server/routes/loginRoutes.ts b/src/server/routes/loginRoutes.ts new file mode 100644 index 0000000..73702b0 --- /dev/null +++ b/src/server/routes/loginRoutes.ts @@ -0,0 +1,50 @@ +import { uncleaner } from 'cleaners' + +import { adjustEvents, getEventsByLoginId } from '../../db/couchPushEvents' +import { asLoginPayload, asLoginUpdatePayload } from '../../types/pushApiTypes' +import { errorResponse, jsonResponse } from '../../types/responseTypes' +import { checkPayload } from '../../util/checkPayload' +import { withDevice } from '../middleware/withDevice' + +const wasLoginPayload = uncleaner(asLoginPayload) + +/** + * POST /v2/login + */ +export const loginFetchRoute = withDevice(async request => { + const { connection, loginId } = request + + if (loginId == null) { + return errorResponse('No login provided', { status: 400 }) + } + + const eventRows = await getEventsByLoginId(connection, loginId) + return jsonResponse( + wasLoginPayload({ + events: eventRows.map(row => row.event) + }) + ) +}) + +/** + * POST /v2/login/update + */ +export const loginUpdateRoute = withDevice(async request => { + const { connection, date, payload, loginId } = request + + if (loginId == null) { + return errorResponse('No login provided', { status: 400 }) + } + + const checked = checkPayload(asLoginUpdatePayload, payload) + if (checked.error != null) return checked.error + const { createEvents, removeEvents } = checked.clean + + const events = await adjustEvents(connection, { + date, + loginId, + createEvents, + removeEvents + }) + return jsonResponse(wasLoginPayload({ events })) +}) diff --git a/src/server/urls.ts b/src/server/urls.ts index 1cab56f..994733c 100644 --- a/src/server/urls.ts +++ b/src/server/urls.ts @@ -3,6 +3,7 @@ import { pickMethod, pickPath, Serverlet } from 'serverlet' import { DbRequest } from '../types/requestTypes' import { errorResponse, jsonResponse } from '../types/responseTypes' import { withLegacyApiKey } from './middleware/withLegacyApiKey' +import { deviceFetchRoute, deviceUpdateRoute } from './routes/deviceRoutes' import { attachUserV1Route, enableCurrencyV1Route, @@ -12,6 +13,7 @@ import { registerDeviceV1Route, toggleStateV1Route } from './routes/legacyRoutes' +import { loginFetchRoute, loginUpdateRoute } from './routes/loginRoutes' import { sendNotificationV1Route } from './routes/notificationRoute' const missingRoute: Serverlet = request => @@ -46,6 +48,19 @@ const urls: { [path: string]: Serverlet } = { '/v1/user/notifications/[0-9A-Za-z]+/?': pickMethod({ GET: withLegacyApiKey(fetchCurrencyV1Route), PUT: withLegacyApiKey(enableCurrencyV1Route) + }), + + '/v2/device/?': pickMethod({ + POST: deviceFetchRoute + }), + '/v2/device/update/?': pickMethod({ + POST: deviceUpdateRoute + }), + '/v2/login/?': pickMethod({ + POST: loginFetchRoute + }), + '/v2/login/update/?': pickMethod({ + POST: loginUpdateRoute }) } export const allRoutes: Serverlet = pickPath(urls, missingRoute) diff --git a/src/types/pushApiTypes.ts b/src/types/pushApiTypes.ts new file mode 100644 index 0000000..7e1cd64 --- /dev/null +++ b/src/types/pushApiTypes.ts @@ -0,0 +1,149 @@ +import { + asArray, + asBoolean, + asDate, + asEither, + asNull, + asNumber, + asObject, + asOptional, + asString, + asUnknown, + Cleaner +} from 'cleaners' + +import { + asBase64, + asBroadcastTx, + asPushEventState, + asPushMessage, + asPushTrigger +} from './pushCleaners' +import { BroadcastTx, PushEvent, PushMessage, PushTrigger } from './pushTypes' + +// --------------------------------------------------------------------------- +// Request types +// --------------------------------------------------------------------------- + +/** + * All v2 requests use this request body. + */ +export interface PushRequestBody { + // The request payload: + data?: unknown + + // Who is making the request: + apiKey: string + deviceId: string + deviceToken?: string + + // For logins: + loginId?: Uint8Array +} + +/** + * Template for creating new push events. + */ +export interface NewPushEvent { + readonly eventId: string + readonly broadcastTxs?: BroadcastTx[] + readonly pushMessage?: PushMessage + readonly trigger: PushTrigger +} + +/** + * PUSH /v2/device/update payload. + */ +export interface DeviceUpdatePayload { + createEvents?: NewPushEvent[] + removeEvents?: string[] + + ignorePriceChanges?: boolean + loginIds?: Uint8Array[] +} + +/** + * PUSH /v2/login/update payload. + */ +export interface LoginUpdatePayload { + createEvents?: NewPushEvent[] + removeEvents?: string[] +} + +// --------------------------------------------------------------------------- +// Request cleaners +// --------------------------------------------------------------------------- + +export const asPushRequestBody: Cleaner = asObject({ + // The request payload: + data: asUnknown, + + // Who is making the request: + apiKey: asString, + deviceId: asString, + deviceToken: asOptional(asString), + + // For logins: + loginId: asOptional(asBase64) +}) + +export const asNewPushEvent: Cleaner = asObject({ + eventId: asString, + broadcastTxs: asOptional(asArray(asBroadcastTx)), + pushMessage: asOptional(asPushMessage), + trigger: asPushTrigger +}) + +export const asDeviceUpdatePayload: Cleaner = asObject({ + createEvents: asOptional(asArray(asNewPushEvent), []), + removeEvents: asOptional(asArray(asString), []), + + ignorePriceChanges: asOptional(asBoolean), + loginIds: asOptional(asArray(asBase64)) +}) + +export const asLoginUpdatePayload: Cleaner = asObject({ + createEvents: asOptional(asArray(asNewPushEvent), []), + removeEvents: asOptional(asArray(asString), []) +}) + +// --------------------------------------------------------------------------- +// Response cleaners +// --------------------------------------------------------------------------- + +/** + * A push event returned from a query. + */ +export const asPushEventStatus: Cleaner< + Omit +> = asObject({ + eventId: asString, + + broadcastTxs: asOptional(asArray(asBroadcastTx)), + pushMessage: asOptional(asPushMessage), + trigger: asPushTrigger, + + // Status: + broadcastTxErrors: asOptional(asArray(asEither(asString, asNull))), + pushMessageEmits: asOptional(asNumber), // Number of devices we sent to + pushMessageFails: asOptional(asNumber), // Number of devices that failed + pushMessageError: asOptional(asString), + state: asPushEventState, + triggered: asOptional(asDate) +}) + +/** + * POST /v2/device response payload. + */ +export const asDevicePayload = asObject({ + events: asArray(asPushEventStatus), + ignorePriceChanges: asBoolean, + loginIds: asArray(asBase64) +}) + +/** + * POST /v2/login response payload. + */ +export const asLoginPayload = asObject({ + events: asArray(asPushEventStatus) +}) diff --git a/src/types/pushCleaners.ts b/src/types/pushCleaners.ts new file mode 100644 index 0000000..911913f --- /dev/null +++ b/src/types/pushCleaners.ts @@ -0,0 +1,87 @@ +import { + asCodec, + asEither, + asNumber, + asObject, + asOptional, + asString, + asTuple, + asValue, + Cleaner +} from 'cleaners' +import { base64 } from 'rfc4648' + +import { + AddressBalanceTrigger, + BroadcastTx, + PriceChangeTrigger, + PriceLevelTrigger, + PushEventState, + PushMessage, + PushTrigger, + TxConfirmTrigger +} from './pushTypes' + +export const asBase64 = asCodec( + raw => base64.parse(asString(raw)), + clean => base64.stringify(clean) +) + +export const asAddressBalanceTrigger: Cleaner = asObject( + { + type: asValue('address-balance'), + pluginId: asString, + tokenId: asOptional(asString), + address: asString, + aboveAmount: asOptional(asString), // Satoshis or Wei or such + belowAmount: asOptional(asString) // Satoshis or Wei or such + } +) + +export const asPriceChangeTrigger: Cleaner = asObject({ + type: asValue('price-change'), + pluginId: asString, + currencyPair: asString, // From our rates server + directions: asOptional(asTuple(asString, asString)), + dailyChange: asOptional(asNumber), // Percentage + hourlyChange: asOptional(asNumber) // Percentage +}) + +export const asPriceLevelTrigger: Cleaner = asObject({ + type: asValue('price-level'), + currencyPair: asString, // From our rates server + aboveRate: asOptional(asNumber), + belowRate: asOptional(asNumber) +}) + +export const asTxConfirmTrigger: Cleaner = asObject({ + type: asValue('tx-confirm'), + pluginId: asString, + confirmations: asNumber, + txid: asString +}) + +export const asPushTrigger: Cleaner = asEither( + asAddressBalanceTrigger, + asPriceChangeTrigger, + asPriceLevelTrigger, + asTxConfirmTrigger +) + +export const asBroadcastTx: Cleaner = asObject({ + pluginId: asString, + rawTx: asBase64 +}) + +export const asPushMessage: Cleaner = asObject({ + title: asOptional(asString), + body: asOptional(asString), + data: asOptional(asObject(asString)) +}) + +export const asPushEventState: Cleaner = asValue( + 'waiting', + 'cancelled', + 'triggered', + 'hidden' +) diff --git a/src/types/pushTypes.ts b/src/types/pushTypes.ts index 979fcb8..3acc270 100644 --- a/src/types/pushTypes.ts +++ b/src/types/pushTypes.ts @@ -28,3 +28,106 @@ export interface ApiKey { admin: boolean adminsdk?: FirebaseAdminKey } + +/** + * An app installed on a single phone. + * + * This the in-memory format, independent of the database. + */ +export interface Device { + readonly created: Date + readonly deviceId: string + + // Settings: + apiKey: string | undefined // Which app to send to? + deviceToken: string | undefined + ignorePriceChanges: boolean + loginIds: Uint8Array[] + visited: Date +} + +// +// Events that devices or logins may subscribe to. +// + +export interface AddressBalanceTrigger { + readonly type: 'address-balance' + readonly pluginId: string + readonly tokenId?: string + readonly address: string + readonly aboveAmount?: string // Satoshis or Wei or such + readonly belowAmount?: string // Satoshis or Wei or such +} + +export interface PriceChangeTrigger { + readonly type: 'price-change' + readonly currencyPair: string // From our rates server + readonly directions?: [string, string] // ['up', 'down'] in user's language + readonly dailyChange?: number // Percentage + readonly hourlyChange?: number // Percentage +} + +export interface PriceLevelTrigger { + readonly type: 'price-level' + readonly currencyPair: string // From our rates server + readonly aboveRate?: number + readonly belowRate?: number +} + +export interface TxConfirmTrigger { + readonly type: 'tx-confirm' + readonly pluginId: string + readonly confirmations: number + readonly txid: string +} + +export type PushTrigger = + | AddressBalanceTrigger + | PriceChangeTrigger + | PriceLevelTrigger + | TxConfirmTrigger + +/** + * Broadcasts a transaction to a blockchain. + */ +export interface BroadcastTx { + readonly pluginId: string + readonly rawTx: Uint8Array // asBase64 +} + +/** + * Sends a push notification. + */ +export interface PushMessage { + readonly title?: string + readonly body?: string + readonly data?: { [key: string]: string } // JSON to push to device +} + +export type PushEventState = + | 'waiting' // Waiting for the trigger + | 'cancelled' // Removed before the trigger happened + | 'triggered' // The trigger and effects are done + | 'hidden' // Removed after being triggered + +/** + * Combines a trigger with an action. + * This the in-memory format, independent of the database. + */ +export interface PushEvent { + readonly created: Date + readonly eventId: string // From the client, not globally unique + readonly deviceId?: string + readonly loginId?: Uint8Array + + readonly broadcastTxs?: BroadcastTx[] + readonly pushMessage?: PushMessage + readonly trigger: PushTrigger + + // Mutable state: + broadcastTxErrors?: Array // For ones that fail + pushMessageEmits?: number // Number of devices we sent to + pushMessageFails?: number // Number of devices that failed + state: PushEventState + triggered?: Date // When did we see the trigger? +} diff --git a/src/types/requestTypes.ts b/src/types/requestTypes.ts index 22d9c40..5ee8738 100644 --- a/src/types/requestTypes.ts +++ b/src/types/requestTypes.ts @@ -1,6 +1,7 @@ import { ServerScope } from 'nano' import { ExpressRequest } from 'serverlet/express' +import { DeviceRow } from '../db/couchDevices' import { ApiKey } from './pushTypes' export interface Logger { @@ -27,3 +28,11 @@ export interface ApiRequest extends DbRequest { readonly json: unknown readonly query: unknown } + +export interface DeviceRequest extends DbRequest { + readonly payload: unknown + + readonly apiKey: ApiKey + readonly deviceRow: DeviceRow + readonly loginId?: Uint8Array +} diff --git a/yarn.lock b/yarn.lock index f33be48..82a1f30 100644 --- a/yarn.lock +++ b/yarn.lock @@ -3369,6 +3369,11 @@ reusify@^1.0.4: resolved "https://registry.yarnpkg.com/reusify/-/reusify-1.0.4.tgz#90da382b1e126efc02146e90845a88db12925d76" integrity sha512-U9nH88a3fc/ekCF1l0/UP1IosiuIjyTh7hBvXVMHYgVcfGvt897Xguj2UOLDeI5BG2m7/uwyaLVT6fbtCwTyzw== +rfc4648@^1.5.2: + version "1.5.2" + resolved "https://registry.yarnpkg.com/rfc4648/-/rfc4648-1.5.2.tgz#cf5dac417dd83e7f4debf52e3797a723c1373383" + integrity sha512-tLOizhR6YGovrEBLatX1sdcuhoSCXddw3mqNVAcKxGJ+J0hFeJ+SjeWCv5UPA/WU3YzWPPuCVYgXBKZUPGpKtg== + rfdc@^1.3.0: version "1.3.0" resolved "https://registry.yarnpkg.com/rfdc/-/rfdc-1.3.0.tgz#d0b7c441ab2720d05dc4cf26e01c89631d9da08b" From 0500a44c95904a64010ba0b7645fe6f75e017738 Mon Sep 17 00:00:00 2001 From: William Swanson Date: Fri, 5 Aug 2022 14:05:04 -0700 Subject: [PATCH 2/8] Add a demo app --- README.md | 2 + docs/demo.ts | 113 +++++++++++++++++++++++++++++++++++++++++++++++++++ package.json | 1 + 3 files changed, 116 insertions(+) create mode 100644 docs/demo.ts diff --git a/README.md b/README.md index 0cb7b9d..9649323 100644 --- a/README.md +++ b/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. diff --git a/docs/demo.ts b/docs/demo.ts new file mode 100644 index 0000000..b13fb2d --- /dev/null +++ b/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 { + 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 { + // 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 +}) diff --git a/package.json b/package.json index 0ff884b..3da565f 100644 --- a/package.json +++ b/package.json @@ -10,6 +10,7 @@ "scripts": { "build": "sucrase -q -t typescript,imports -d ./lib ./src", "clean": "rimraf lib", + "demo": "node -r sucrase/register docs/demo.ts", "fix": "yarn-deduplicate && eslint . --fix", "lint": "eslint .", "precommit": "lint-staged && npm-run-all types prepare", From e4f9a03fe9efb88fb569e70f55a3b1052c7de30d Mon Sep 17 00:00:00 2001 From: William Swanson Date: Mon, 8 Aug 2022 10:32:26 -0700 Subject: [PATCH 3/8] Add a CLI tool --- package.json | 4 ++- src/cli/cli.ts | 35 ++++++++++++++++++++++ src/cli/cliTools.ts | 56 +++++++++++++++++++++++++++++++++++ src/cli/commands/getDevice.ts | 28 ++++++++++++++++++ yarn.lock | 12 ++++++++ 5 files changed, 134 insertions(+), 1 deletion(-) create mode 100644 src/cli/cli.ts create mode 100644 src/cli/cliTools.ts create mode 100644 src/cli/commands/getDevice.ts diff --git a/package.json b/package.json index 3da565f..531b09b 100644 --- a/package.json +++ b/package.json @@ -10,13 +10,14 @@ "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", + "start-server": "node -r sucrase/register src/server/index.ts", "types": "tsc", "verify": "npm-run-all lint types prepare" }, @@ -29,6 +30,7 @@ "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", diff --git a/src/cli/cli.ts b/src/cli/cli.ts new file mode 100644 index 0000000..190d552 --- /dev/null +++ b/src/cli/cli.ts @@ -0,0 +1,35 @@ +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' + +async function main(): Promise { + 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) + + const args = process.argv.slice(2) + await cli.runExit(args, context) +} + +main().catch(error => console.error(error)) diff --git a/src/cli/cliTools.ts b/src/cli/cliTools.ts new file mode 100644 index 0000000..24bc681 --- /dev/null +++ b/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 = 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) + } +} diff --git a/src/cli/commands/getDevice.ts b/src/cli/commands/getDevice.ts new file mode 100644 index 0000000..e4909f9 --- /dev/null +++ b/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 { + static paths = [['get-device']] + static usage = { description: "Shows a device's status" } + + deviceId = Option.String({ name: 'deviceId', required: true }) + + async execute(): Promise { + 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 + } +} diff --git a/yarn.lock b/yarn.lock index 82a1f30..b468352 100644 --- a/yarn.lock +++ b/yarn.lock @@ -863,6 +863,13 @@ cli-truncate@^2.1.0: slice-ansi "^3.0.0" string-width "^4.2.0" +clipanion@^3.2.0-rc.11: + version "3.2.0-rc.11" + resolved "https://registry.yarnpkg.com/clipanion/-/clipanion-3.2.0-rc.11.tgz#765661c9aeda8ecc14035a61a4b19b361f9a5400" + integrity sha512-fugY+N5uPop31VDYhjTc31DwPjCCLx6kmvdlFTf8fztpOxwplopiZr1XSHSA2qNmfpcXlJZKJsXMkxvXmdzK7g== + dependencies: + typanion "^3.8.0" + co@^4.6.0: version "4.6.0" resolved "https://registry.yarnpkg.com/co/-/co-4.6.0.tgz#6ea6bdf3d853ae54ccb8e47bfa0bf3f9031fb184" @@ -3892,6 +3899,11 @@ tsutils@^3.21.0: dependencies: tslib "^1.8.1" +typanion@^3.8.0: + version "3.9.0" + resolved "https://registry.yarnpkg.com/typanion/-/typanion-3.9.0.tgz#071a31a0f81c3c31226e190d0a6513ff1c8ae1a3" + integrity sha512-7yPk67IIquhKQcUXOBM27vDuGmZf6oJbEmzgVfDniHCkT6+z4JnKY85nKqbstoec8Kp7hD06TP3Kc98ij43PIg== + type-check@^0.4.0, type-check@~0.4.0: version "0.4.0" resolved "https://registry.yarnpkg.com/type-check/-/type-check-0.4.0.tgz#07b8203bfa7056c0657050e3ccd2c37730bab8f1" From 1a14b6944bb63b9902913896e9419797c3248503 Mon Sep 17 00:00:00 2001 From: William Swanson Date: Mon, 8 Aug 2022 09:51:12 -0700 Subject: [PATCH 4/8] Add base58 utilities --- package.json | 1 + src/util/base58.ts | 14 ++++++++++++++ yarn.lock | 7 +++++++ 3 files changed, 22 insertions(+) create mode 100644 src/util/base58.ts diff --git a/package.json b/package.json index 531b09b..b3d8752 100644 --- a/package.json +++ b/package.json @@ -27,6 +27,7 @@ "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", diff --git a/src/util/base58.ts b/src/util/base58.ts new file mode 100644 index 0000000..59401f3 --- /dev/null +++ b/src/util/base58.ts @@ -0,0 +1,14 @@ +import baseX from 'base-x' + +const base58Codec = baseX( + '123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz' +) + +export const base58 = { + parse(text: string): Uint8Array { + return base58Codec.decode(text) + }, + stringify(data: Uint8Array | number[]): string { + return base58Codec.encode(data) + } +} diff --git a/yarn.lock b/yarn.lock index b468352..36ee93a 100644 --- a/yarn.lock +++ b/yarn.lock @@ -741,6 +741,13 @@ balanced-match@^1.0.0: resolved "https://registry.yarnpkg.com/balanced-match/-/balanced-match-1.0.2.tgz#e83e3a7e3f300b34cb9d87f615fa0cbf357690ee" integrity sha512-3oSeUO0TMV67hN1AmbXsK4yaqU7tjiHlbxRDZOpH0KW9+CeX4bRAaX0Anxt0tx2MrpRpWwQaPwIlISEJhYU5Pw== +base-x@^3.0.8: + version "3.0.9" + resolved "https://registry.yarnpkg.com/base-x/-/base-x-3.0.9.tgz#6349aaabb58526332de9f60995e548a53fe21320" + integrity sha512-H7JU6iBHTal1gp56aKoaa//YUxEaAOUiydvrV/pILqIHXTtqxSkATOnDA2u+jZ/61sD+L/412+7kzXRtWukhpQ== + dependencies: + safe-buffer "^5.0.1" + base64-js@^1.3.0: version "1.3.1" resolved "https://registry.yarnpkg.com/base64-js/-/base64-js-1.3.1.tgz#58ece8cb75dd07e71ed08c736abc5fac4dbf8df1" From 473e5b8153083e0a57f86f40e9709235e7e8a2bb Mon Sep 17 00:00:00 2001 From: William Swanson Date: Mon, 8 Aug 2022 09:53:15 -0700 Subject: [PATCH 5/8] Add the `verifyData` utility --- src/util/verifyData.ts | 12 ++++++++++++ 1 file changed, 12 insertions(+) create mode 100644 src/util/verifyData.ts diff --git a/src/util/verifyData.ts b/src/util/verifyData.ts new file mode 100644 index 0000000..07f8980 --- /dev/null +++ b/src/util/verifyData.ts @@ -0,0 +1,12 @@ +/** + * Compares two byte arrays without data-dependent branches. + * Returns true if they match. + */ +export function verifyData(a: Uint8Array, b: Uint8Array): boolean { + const length = a.length + if (length !== b.length) return false + + let out = 0 + for (let i = 0; i < length; ++i) out |= a[i] ^ b[i] + return out === 0 +} From 479a274ecaa5b48f92b14c4cad4426c736001238 Mon Sep 17 00:00:00 2001 From: William Swanson Date: Mon, 8 Aug 2022 09:54:03 -0700 Subject: [PATCH 6/8] Keep the user lists updated in the legacy routes We are going to shut down the v1 routes, but we need the loginId list for security. --- src/server/routes/legacyRoutes.ts | 28 ++++++++++++++++++++++++++-- 1 file changed, 26 insertions(+), 2 deletions(-) diff --git a/src/server/routes/legacyRoutes.ts b/src/server/routes/legacyRoutes.ts index fec954d..07a4ef7 100644 --- a/src/server/routes/legacyRoutes.ts +++ b/src/server/routes/legacyRoutes.ts @@ -9,11 +9,14 @@ import { } from 'cleaners' import { Serverlet } from 'serverlet' +import { getDeviceById } from '../../db/couchDevices' import { Device } from '../../models/Device' import { User } from '../../models/User' import { ApiRequest } from '../../types/requestTypes' import { errorResponse, jsonResponse } from '../../types/responseTypes' +import { base58 } from '../../util/base58' import { checkPayload } from '../../util/checkPayload' +import { verifyData } from '../../util/verifyData' /** * The GUI names this `registerDevice`, and calls it at boot. @@ -23,7 +26,7 @@ import { checkPayload } from '../../util/checkPayload' * Response body: unused */ export const registerDeviceV1Route: Serverlet = async request => { - const { json, log, query } = request + const { apiKey, connection, date, json, log, query } = request const checkedQuery = checkPayload(asRegisterDeviceQuery, query) if (checkedQuery.error != null) return checkedQuery.error @@ -43,6 +46,16 @@ export const registerDeviceV1Route: Serverlet = async request => { log(`Device registered.`) } + // Update the v2 device: + { + const deviceRow = await getDeviceById(connection, deviceId, date) + const { device } = deviceRow + device.apiKey = apiKey.apiKey + device.deviceToken = clean.tokenId + device.visited = date + await deviceRow.save() + } + return jsonResponse(device) } @@ -79,7 +92,7 @@ export const fetchStateV1Route: Serverlet = async request => { * Response body: unused */ export const attachUserV1Route: Serverlet = async request => { - const { log, query } = request + const { connection, date, log, query } = request const checkedQuery = checkPayload(asAttachUserQuery, query) if (checkedQuery.error != null) return checkedQuery.error @@ -96,6 +109,17 @@ export const attachUserV1Route: Serverlet = async request => { log(`Successfully attached device "${deviceId}" to user "${userId}"`) + // Update the v2 device: + { + const deviceRow = await getDeviceById(connection, deviceId, date) + const { device } = deviceRow + const loginId = base58.parse(userId) + if (device.loginIds.find(row => verifyData(loginId, row)) == null) { + device.loginIds = [...device.loginIds, loginId] + await deviceRow.save() + } + } + return jsonResponse(user) } From 80336269135d2c640cf02a22af31fb7c5d296d33 Mon Sep 17 00:00:00 2001 From: William Swanson Date: Mon, 8 Aug 2022 12:00:28 -0700 Subject: [PATCH 7/8] Add a CLI command to migrate legacy users --- src/cli/cli.ts | 2 + src/cli/commands/migrateDevices.ts | 71 ++++++++++++++++++++++++++++++ src/models/Device.ts | 2 + src/models/User.ts | 2 + 4 files changed, 77 insertions(+) create mode 100644 src/cli/commands/migrateDevices.ts diff --git a/src/cli/cli.ts b/src/cli/cli.ts index 190d552..95b8509 100644 --- a/src/cli/cli.ts +++ b/src/cli/cli.ts @@ -6,6 +6,7 @@ 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 { const connection = nano(serverConfig.couchUri) @@ -27,6 +28,7 @@ async function main(): Promise { // Our commands: cli.register(GetDevice) + cli.register(MigrateDevices) const args = process.argv.slice(2) await cli.runExit(args, context) diff --git a/src/cli/commands/migrateDevices.ts b/src/cli/commands/migrateDevices.ts new file mode 100644 index 0000000..c7ee0ee --- /dev/null +++ b/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 { + static paths = [['migrate-devices']] + static usage = { description: 'Migrate v1 devices to the v2 database' } + + async execute(): Promise { + 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 + } +} diff --git a/src/models/Device.ts b/src/models/Device.ts index eb9a331..e541f8d 100644 --- a/src/models/Device.ts +++ b/src/models/Device.ts @@ -1,4 +1,5 @@ import { asNumber, asObject, asOptional, asString } from 'cleaners' +import { asCouchDoc } from 'edge-server-tools' import Nano from 'nano' import { serverConfig } from '../serverConfig' @@ -15,6 +16,7 @@ const asDevice = asObject({ edgeVersion: asString, edgeBuildNumber: asNumber }) +export const asLegacyDevice = asCouchDoc(asDevice) export class Device extends Base implements ReturnType { public static table = dbDevices diff --git a/src/models/User.ts b/src/models/User.ts index 917ad58..2abae06 100644 --- a/src/models/User.ts +++ b/src/models/User.ts @@ -3,6 +3,7 @@ /* eslint-disable @typescript-eslint/strict-boolean-expressions */ import { asBoolean, asMap, asObject, asOptional } from 'cleaners' +import { asCouchDoc } from 'edge-server-tools' import Nano from 'nano' import { serverConfig } from '../serverConfig' @@ -27,6 +28,7 @@ const asUser = asObject({ devices: asUserDevices, notifications: asUserNotifications }) +export const asLegacyUser = asCouchDoc(asUser) export interface INotificationsEnabledViewResponse { devices: ReturnType From a9dff1f17c6bca39cb08bfa277c3674deca77f9c Mon Sep 17 00:00:00 2001 From: William Swanson Date: Tue, 9 Aug 2022 16:28:58 -0700 Subject: [PATCH 8/8] Remove the legacy price script --- package.json | 1 - pm2.json | 6 - src/db/couchSettings.ts | 14 +- src/db/couchSetup.ts | 3 - src/models/CurrencyThreshold.ts | 58 -------- src/price-script/checkPriceChanges.ts | 177 ----------------------- src/price-script/fetchThresholdPrices.ts | 101 ------------- src/price-script/index.ts | 62 -------- src/price-script/prices.ts | 57 -------- 9 files changed, 1 insertion(+), 478 deletions(-) delete mode 100644 src/models/CurrencyThreshold.ts delete mode 100644 src/price-script/checkPriceChanges.ts delete mode 100644 src/price-script/fetchThresholdPrices.ts delete mode 100644 src/price-script/index.ts delete mode 100644 src/price-script/prices.ts diff --git a/package.json b/package.json index b3d8752..49cda2c 100644 --- a/package.json +++ b/package.json @@ -16,7 +16,6 @@ "lint": "eslint .", "precommit": "lint-staged && npm-run-all types prepare", "prepare": "husky install && npm-run-all clean build", - "start-price": "node -r sucrase/register src/price-script/index.ts", "start-server": "node -r sucrase/register src/server/index.ts", "types": "tsc", "verify": "npm-run-all lint types prepare" diff --git a/pm2.json b/pm2.json index 8cccd78..3310dc0 100644 --- a/pm2.json +++ b/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" } ] } diff --git a/src/db/couchSettings.ts b/src/db/couchSettings.ts index 467bfdf..4c1d613 100644 --- a/src/db/couchSettings.ts +++ b/src/db/couchSettings.ts @@ -1,11 +1,4 @@ -import { - asArray, - asBoolean, - asMaybe, - asNumber, - asObject, - asString -} from 'cleaners' +import { asArray, asBoolean, asMaybe, asObject, asString } from 'cleaners' import { asReplicatorSetupDocument, DatabaseSetup, @@ -26,13 +19,8 @@ const asSettings = asObject({ [] ), - // Default thresholds: - defaultAnomaly: asMaybe(asNumber, 90), - defaultHours: asMaybe(asObject(asNumber), { '1': 3, '24': 10 }), - // Mode toggles: debugLogs: asMaybe(asBoolean, false), - priceCheckInMinutes: asMaybe(asNumber, 5), // Other services we rely on: slackUri: asMaybe(asString, '') diff --git a/src/db/couchSetup.ts b/src/db/couchSetup.ts index 452e31b..17702b1 100644 --- a/src/db/couchSetup.ts +++ b/src/db/couchSetup.ts @@ -15,8 +15,6 @@ import { settingsSetup, syncedReplicators } from './couchSettings' // Databases // --------------------------------------------------------------------------- -const thresholdsSetup: DatabaseSetup = { name: 'db_currency_thresholds' } - const devicesSetup: DatabaseSetup = { name: 'db_devices' } const usersSetup: DatabaseSetup = { @@ -48,7 +46,6 @@ export async function setupDatabases( setupDatabase(connection, couchDevicesSetup, options), setupDatabase(connection, couchEventsSetup, options), setupDatabase(connection, devicesSetup, options), - setupDatabase(connection, thresholdsSetup, options), setupDatabase(connection, usersSetup, options) ]) } diff --git a/src/models/CurrencyThreshold.ts b/src/models/CurrencyThreshold.ts deleted file mode 100644 index c7c067b..0000000 --- a/src/models/CurrencyThreshold.ts +++ /dev/null @@ -1,58 +0,0 @@ -import { asBoolean, asMap, asNumber, asObject, asOptional } from 'cleaners' -import Nano from 'nano' - -import { serverConfig } from '../serverConfig' -import { Base } from './base' - -const nanoDb = Nano(serverConfig.couchUri) -const dbCurrencyThreshold = nanoDb.db.use('db_currency_thresholds') - -const asThreshold = asObject({ - custom: asOptional(asNumber), - lastUpdated: asNumber, - price: asNumber -}) -const asThresholds = asMap(asThreshold) - -interface ICurrencyThreshold { - disabled?: boolean - anomaly?: number - thresholds: ReturnType -} - -const asCurrencyThreshold = asObject({ - disabled: asOptional(asBoolean), - anomaly: asOptional(asNumber), - thresholds: asThresholds -}) - -export class CurrencyThreshold extends Base implements ICurrencyThreshold { - public static table = dbCurrencyThreshold - public static asType = asCurrencyThreshold - - public disabled?: boolean - public anomaly?: number - public thresholds!: ReturnType - - public static async fromCode( - currencyCode: string - ): Promise { - const threshold = new CurrencyThreshold(undefined, currencyCode) - return await threshold.save() - } - - public async update( - hours: string | number, - timestamp: number, - price: number - ): Promise { - const threshold = this.thresholds[hours] ?? { - lastUpdated: 0, - price: 0 - } - threshold.lastUpdated = timestamp - threshold.price = price - this.thresholds[hours] = threshold - return (await this.save()) as CurrencyThreshold - } -} diff --git a/src/price-script/checkPriceChanges.ts b/src/price-script/checkPriceChanges.ts deleted file mode 100644 index 81e134f..0000000 --- a/src/price-script/checkPriceChanges.ts +++ /dev/null @@ -1,177 +0,0 @@ -import io from '@pm2/io' -import { MetricType } from '@pm2/io/build/main/services/metrics' - -import { syncedSettings } from '../db/couchSettings' -import { CurrencyThreshold } from '../models/CurrencyThreshold' -import { Device } from '../models/Device' -import { User } from '../models/User' -import { PushResult, PushSender } from '../util/pushSender' -import { fetchThresholdPrice } from './fetchThresholdPrices' - -// Firebase Messaging API limits batch messages to 500 -const NOTIFICATION_LIMIT = 500 -const MANGO_FIND_LIMIT = 200 -const THRESHOLD_FETCH_LIMIT = 200 - -export interface NotificationPriceChange { - currencyCode: string - now: string - before: string - hourChange: string - priceNow: number - priceBefore: number - priceChange: number -} - -export async function checkPriceChanges(sender: PushSender): Promise { - // Sends a notification to devices about a price change - async function sendNotification( - thresholdPrice: NotificationPriceChange, - deviceTokens: string[] - ): Promise { - const { currencyCode, hourChange, priceChange, priceNow } = thresholdPrice - - const direction = priceChange > 0 ? 'up' : 'down' - const symbol = priceChange > 0 ? '+' : '' - const time = Number(hourChange) === 1 ? '1 hour' : `${hourChange} hours` - - const title = 'Price Alert' - const displayPrice = formatDisplayPrice(priceNow) - const body = `${currencyCode} is ${direction} ${symbol}${priceChange}% to $${displayPrice} in the last ${time}.` - const data = {} - - return await sender.send(title, body, deviceTokens, data) - } - - // Fetch list of threshold items and their prices - const thresholds = await CurrencyThreshold.where({ - selector: { - $or: [{ disabled: { $exists: false } }, { disabled: false }] - }, - limit: THRESHOLD_FETCH_LIMIT - }) - // Fetch default values to use if otherwise not defined - const { defaultAnomaly, defaultHours } = syncedSettings.doc - for (const threshold of thresholds) { - const currencyCode = threshold._id - const anomalyPercent = threshold.anomaly ?? defaultAnomaly - for (const hours in defaultHours) { - const priceData = await fetchThresholdPrice( - threshold, - hours, - defaultHours[hours], - anomalyPercent - ) - if (priceData == null) continue - - const { rows: usersDevices } = await User.devicesByCurrencyHours( - currencyCode, - hours - ) - // Skip if no users registered to currency - if (usersDevices.length === 0) continue - - const deviceIds: string[] = [] - for (const { value: userDevices } of usersDevices) { - for (const deviceId in userDevices) { - deviceIds.push(deviceId) - } - } - const tokenGenerator = deviceTokenGenerator(deviceIds) - let done = false - let successCount = 0 - let failureCount = 0 - while (!done) { - const next = await tokenGenerator.next() - done = next.done ?? false - - if (next.value != null) { - // Send notification to user about price change - try { - const response = await sendNotification(priceData, next.value) - successCount += response.successCount - failureCount += response.failureCount - } catch (err: any) { - io.notifyError(err, { - custom: { - message: - 'Failed to batch send notifications for currency hours threshold change', - currencyCode, - hours - } - }) - } - } - } - - const idPostfix = `${currencyCode}:${hours}` - const namePostfix = `Notifications For ${currencyCode} - ${hours} Hour` - io.metrics([ - { - type: MetricType.metric, - id: `notifications:success:${idPostfix}`, - name: `Successful ${namePostfix}`, - value: () => successCount - }, - { - type: MetricType.metric, - id: `notifications:failure:${idPostfix}`, - name: `Failed ${namePostfix}`, - value: () => failureCount - } - ]) - } - } -} - -async function* deviceTokenGenerator( - deviceIds: string[] -): AsyncGenerator { - const tokenSet: Set = new Set() - let tokens: string[] = [] - let bookmark: string | undefined - let done = false - while (!done) { - const response = await Device.table.find({ - bookmark, - selector: { - _id: { - $in: deviceIds - }, - tokenId: { - $exists: true, - $ne: null - } - }, - fields: ['tokenId'], - limit: MANGO_FIND_LIMIT - }) - bookmark = response.bookmark - - for (const { tokenId } of response.docs) { - if (tokenId == null || tokenSet.has(tokenId)) continue - - tokenSet.add(tokenId) - tokens.push(tokenId) - - if (tokens.length === NOTIFICATION_LIMIT) { - yield tokens - tokens = [] - } - } - - if (response.docs.length < MANGO_FIND_LIMIT) { - done = true - } - } - - return tokens -} - -// Set decimal place to 2 significant digits -function formatDisplayPrice(priceNow: number): number { - const numSplit = priceNow.toString().split('.') - const sigIndex = numSplit[1].search(/[1-9]/) - numSplit[1] = numSplit[1].substring(0, sigIndex + 2) - return Number(numSplit.join('.')) -} diff --git a/src/price-script/fetchThresholdPrices.ts b/src/price-script/fetchThresholdPrices.ts deleted file mode 100644 index 7b3e250..0000000 --- a/src/price-script/fetchThresholdPrices.ts +++ /dev/null @@ -1,101 +0,0 @@ -import io from '@pm2/io' - -import { CurrencyThreshold } from '../models/CurrencyThreshold' -import { NotificationPriceChange } from './checkPriceChanges' -import { getPrice } from './prices' - -const SLEEP_TIMEOUT = 1000 // in milliseconds - -type Counter = ReturnType -const processMetrics: { [id: string]: Counter | undefined } = {} - -async function sleep(ms = SLEEP_TIMEOUT): Promise { - return await new Promise(resolve => setTimeout(resolve, ms)) -} -export async function fetchThresholdPrice( - currencyThreshold: CurrencyThreshold, - hours: string | number, - defaultPercent: number, - anomalyPercent: number -): Promise { - const { disabled = false } = currencyThreshold - if (disabled) return - - const currencyCode = currencyThreshold._id - let priceNow: number - try { - await sleep() - priceNow = await getPrice(currencyCode, 'USD') - } catch { - return - } - - const hoursAgo = Date.now() - Number(hours) * 60 * 60 * 1000 - let threshold = currencyThreshold.thresholds[hours] - if (threshold == null) { - threshold = { - custom: undefined, - lastUpdated: 0, - price: 0 - } - } - - const before = - threshold.lastUpdated === 0 || hoursAgo > threshold.lastUpdated - ? hoursAgo - : threshold.lastUpdated - - let priceBefore - try { - priceBefore = await getPrice(currencyCode, 'USD', before) - } catch { - return - } - - const priceChange = parseFloat( - ((100 * (priceNow - priceBefore)) / priceBefore).toFixed(2) - ) - const now = new Date().toISOString() - const priceData: NotificationPriceChange = { - currencyCode, - now, - before: new Date(before).toISOString(), - hourChange: hours.toString(), - priceNow, - priceBefore, - priceChange - } - console.log(priceData) - - if (Math.abs(priceChange) >= anomalyPercent) { - io.notifyError(new Error('Rates Server Anomaly Price Change Detected'), { - custom: priceData - }) - - await currencyThreshold.save('disabled', true) - - return - } - - const percent = threshold.custom ?? defaultPercent - if (Math.abs(priceChange) >= percent) { - await currencyThreshold - .update(hours, Date.parse(now), priceNow) - .catch(err => { - console.error(`Could not update ${currencyCode} threshold data.`) - console.error(err) - }) - - const counterId = `threshold:crossed:${currencyCode}:${hours}` - let counter = processMetrics[counterId] - if (counter == null) { - counter = processMetrics[counterId] = io.counter({ - id: counterId, - name: `Threshold Crossed for ${currencyCode} - ${hours} Hour` - }) - } - counter.inc() - - return priceData - } -} diff --git a/src/price-script/index.ts b/src/price-script/index.ts deleted file mode 100644 index 1f3df7e..0000000 --- a/src/price-script/index.ts +++ /dev/null @@ -1,62 +0,0 @@ -import io from '@pm2/io' -import { makePeriodicTask } from 'edge-server-tools' -import nano from 'nano' - -import { getApiKeyByKey } from '../db/couchApiKeys' -import { syncedSettings } from '../db/couchSettings' -import { setupDatabases } from '../db/couchSetup' -import { serverConfig } from '../serverConfig' -import { makePushSender } from '../util/pushSender' -import { checkPriceChanges } from './checkPriceChanges' - -const runCounter = io.counter({ - id: 'price:script:counter', - name: 'Price Script Runner Count' -}) - -async function main(): Promise { - const { couchUri } = serverConfig - - // Set up databases: - const connection = nano(couchUri) - await setupDatabases(connection) - - if (syncedSettings.doc.apiKeys.length === 0) { - throw new Error('No partner apiKeys') - } - - // Read the API keys from settings: - const senders = await Promise.all( - syncedSettings.doc.apiKeys.map(async partner => { - const apiKey = await getApiKeyByKey(connection, partner.apiKey) - if (apiKey == null) { - throw new Error(`Cannot find API key ${partner.apiKey}`) - } - return await makePushSender(apiKey) - }) - ) - - // Check the prices every few minutes: - const task = makePeriodicTask( - async () => { - runCounter.inc() - - for (const sender of senders) { - await checkPriceChanges(sender) - } - }, - 60 * 1000 * syncedSettings.doc.priceCheckInMinutes, - { - onError(error) { - // eslint-disable-next-line @typescript-eslint/no-unnecessary-type-assertion - io.notifyError(error as any) - } - } - ) - task.start() -} - -main().catch(error => { - console.error(error) - process.exit(1) -}) diff --git a/src/price-script/prices.ts b/src/price-script/prices.ts deleted file mode 100644 index 26c4cac..0000000 --- a/src/price-script/prices.ts +++ /dev/null @@ -1,57 +0,0 @@ -import io from '@pm2/io' -import axios from 'axios' -import { asNumber } from 'cleaners' - -const TIMEOUT = 10000 // in milliseconds - -const rates = axios.create({ - baseURL: `https://rates1.edge.app/v1/exchangeRate`, - timeout: TIMEOUT -}) - -export async function getPriceChange( - base: string, - quote: string -): Promise { - const today = Date.now() - const yesterday = today - 1000 * 60 * 60 * 24 - - const todayPrice = await getPrice(base, quote, today) - const yesterdayPrice = await getPrice(base, quote, yesterday) - return (todayPrice - yesterdayPrice) / yesterdayPrice -} - -export async function getPrice( - base: string, - quote: string, - at?: number -): Promise { - let dateString: string = '' - if (at != null) { - dateString = `&date=${new Date(at).toISOString()}` - } - - try { - const { - data: { exchangeRate } - } = await rates.get(`?currency_pair=${base}_${quote}${dateString}`) - const rate = asNumber(parseFloat(exchangeRate)) - if (!/^\d+(\.\d+)?/.test(exchangeRate.toString())) { - throw new Error(`${base}/${quote} rate given was ${String(exchangeRate)}`) - } - return rate - } catch (err: any) { - // @ts-expect-error - const lookupDate = new Date(at).toISOString() - io.notifyError(err, { - custom: { - base, - quote, - lookupDate - } - }) - const reason: string = err.response.data.error - console.log(`Cannot fetch prices for ${base}/${quote} - ${reason}`) - throw err - } -}