Add a marketing tool API with marketer keys and audience queries

CHANGELOG.md

@@ -2,6 +2,13 @@
 
 ## Unreleased
 
+- added: A marketing push endpoint, `POST /marketing/push`. One call sends a message to a single device or login (a test), to an explicit list of device ids, or to the audience a location filter resolves to: include/exclude lists of countries, cities and regions, where the lines within a list are alternatives, the lists narrow each other, and an empty country include means every country. Region and city rules are accepted only with exactly one included country, since those values are stored without their country. With `dryRun: true` it resolves the audience and reports it without sending, so the caller can review the list and then send exactly those ids. Sends to a list stream a progress log, and each send may carry an optional Edge deep link (`url`) that the app opens after login, accepted only from an allow-list of buy, sell, swap, exchange, plugin (without a query), promotion, fund-account and earn links since the app follows any link it can parse, and an optional `label` that the server logs beside the campaign id. Tests obey the marketing opt-out like any other send.
+- added: `GET /marketing/countries`, the countries devices are located in under the caller's targeted keys, with counts, so a caller can offer exactly the names an audience filter matches.
+- added: A `marketer` flag and a `targetApiKeys` list on API keys. The marketing endpoint requires the flag (or `admin`) and only reaches devices registered under the listed keys, so the api keys baked into the apps cannot call it.
+- fixed: Recognize Firebase's current "token not registered" error, so devices whose app has been uninstalled get their tokens cleared instead of being retried on every send forever.
+- fixed: Skip devices registered with an empty token instead of sending them to Firebase, which rejected every one with "Exactly one of topic, token or condition is required".
+- fixed: Log the device and error when a push fails. The details were being passed to Pino in the wrong argument order, so every failure logged as a bare "Unknown error".
+
 ## 2.5.0 (2025-05-08)
 
 - changed: Move API keys to a separate settings document.

pushServerConfig.sample.json

@@ -0,0 +1,7 @@
+{
+  "listenHost": "127.0.0.1",
+  "listenPort": 8008,
+  "amqpUri": "amqp://username:password@localhost:5672",
+  "couchUri": "http://username:password@localhost:5984",
+  "currentCluster": ""
+}

src/daemons/publishDaemon.ts

@@ -9,6 +9,7 @@ import {
   getDevicesByLoginId
 } from '../db/couchDevices'
 import { DbConnections } from '../db/dbConnections'
+import { isUnregisteredToken } from '../util/firebaseErrors'
 import { logger } from '../util/logger'
 import { asRabbitMessage, SendableMessage } from '../util/pushSender'
 import { runDaemon } from './runDaemon'
@@ -73,7 +74,10 @@ async function sendToDevice(
 
   if (message.isMarketing && ignoreMarketing) return
   if (message.isPriceChange && ignorePriceChanges) return
-  if (apiKey == null || deviceToken == null) return
+  // Some devices register with an empty string for a token rather than none
+  // at all. Firebase rejects those with "Exactly one of topic, token or
+  // condition is required", so treat them as missing:
+  if (apiKey == null || deviceToken == null || deviceToken === '') return
 
   const sender = await getSender(connections, apiKey)
   if (sender == null) return
@@ -88,12 +92,14 @@ async function sendToDevice(
       data: message.data ?? {}
     })
   } catch (error) {
-    if (String(error).includes('not a valid FCM registration token')) {
+    if (isUnregisteredToken(error)) {
       logger.info(`Disabling device: ${deviceId}`)
       deviceRow.device.deviceToken = undefined
       await deviceRow.save()
     } else {
-      logger.info('Unknown error', { deviceId, error })
+      // Pino takes the details first and the message second. Passing them the
+      // other way around silently drops them:
+      logger.info({ deviceId, error: String(error) }, 'Unknown error')
     }
   }
 }

src/db/couchApiKeys.ts

@@ -1,4 +1,4 @@
-import { asBoolean, asObject, asOptional, asString } from 'cleaners'
+import { asArray, asBoolean, asObject, asOptional, asString } from 'cleaners'
 import {
   asCouchDoc,
   asMaybeNotFoundError,
@@ -32,7 +32,9 @@ export const asCouchApiKey = asCouchDoc(
   asObject<CouchApiKey>({
     appId: asString,
     admin: asBoolean,
-    adminsdk: asOptional(asFirebaseAdminKey)
+    adminsdk: asOptional(asFirebaseAdminKey),
+    marketer: asOptional(asBoolean, false),
+    targetApiKeys: asOptional(asArray(asString), () => [])
   })
 )
 

src/db/couchDevices.ts

@@ -25,6 +25,10 @@ import { asBase64 } from '../types/pushCleaners'
 import { Device } from '../types/pushTypes'
 import { DbConnections } from './dbConnections'
 
+// Couch rejects an `_all_docs` request with no keys, and very large key lists
+// make for unwieldy requests, so batch lookups at this size:
+const FETCH_BATCH_SIZE = 500
+
 /**
  * A device returned from the database.
  * To make changes, edit the `device` object, then call `save`.
@@ -117,12 +121,45 @@ const locationByRegionDesign = makeJsDesign('locationByRegion', ({ emit }) => ({
   reduce: '_count'
 }))
 
+/**
+ * Looks up devices by API key, then by IP location.
+ *
+ * Between them the key and the row value carry everything an audience query
+ * needs: the location in the key, and the fields that decide whether the
+ * device can be sent to in the value, so the query never has to fetch the
+ * documents themselves. Devices with no location are absent from this view,
+ * and are therefore unreachable by location targeting.
+ *
+ * The map runs inside Couch's view server, which has no `??` helper (sucrase
+ * compiles it to one), so the null checks are spelled out as ternaries.
+ */
+const apiKeyLocationDesign = makeJsDesign('apiKeyLocation', ({ emit }) => ({
+  map: function (doc) {
+    if (doc.apiKey == null || doc.location == null) return
+    emit(
+      [
+        doc.apiKey,
+        doc.location.country != null ? doc.location.country : '',
+        doc.location.region != null ? doc.location.region : '',
+        doc.location.city != null ? doc.location.city : ''
+      ],
+      {
+        deviceToken: doc.deviceToken,
+        ignoreMarketing: doc.ignoreMarketing === true,
+        visited: doc.visited
+      }
+    )
+  },
+  reduce: '_count'
+}))
+
 export const couchDevicesSetup: DatabaseSetup = {
   name: 'push-devices',
   documents: {
     '_design/loginId': loginIdDesign,
     '_design/locationByCity': locationByCityDesign,
-    '_design/locationByRegion': locationByRegionDesign
+    '_design/locationByRegion': locationByRegionDesign,
+    '_design/apiKeyLocation': apiKeyLocationDesign
   }
 }
 
@@ -183,6 +220,32 @@ export const countDevicesByLocation = async (
   })
 }
 
+/**
+ * Counts the located devices under an API key, by country, straight from the
+ * view's reduce, so it answers in milliseconds however large the key is. The
+ * names are exactly what ip-api stored, which is what an audience filter
+ * matches on. These are registration counts before the token and opt-out
+ * checks, so they run above what an audience query returns.
+ */
+export async function countDevicesByCountry(
+  connections: DbConnections,
+  apiKey: string
+): Promise<Array<{ country: string; count: number }>> {
+  const db = connections.couch.use(couchDevicesSetup.name)
+  const response = await db.view('apiKeyLocation', 'apiKeyLocation', {
+    reduce: true,
+    group_level: 2,
+    start_key: [apiKey],
+    end_key: [apiKey, 'zzzzzz']
+  })
+  return response.rows.flatMap(row => {
+    const key = asMaybe(asArray(asString))(row.key)
+    const count = asMaybe(asNumber)(row.value)
+    if (key == null || count == null) return []
+    return [{ country: key[1] ?? '', count }]
+  })
+}
+
 /**
  * Looks up a device by its id.
  * If the device does not exist in the database, creates a fresh row.
@@ -217,6 +280,45 @@ export async function getDeviceById(
   return makeDeviceRow(db, asCouchDevice(raw))
 }
 
+/** One batch of a by-id lookup, along with how far through the ids we are. */
+export interface DeviceBatch {
+  deviceRows: DeviceRow[]
+  /** How many of the requested ids have been read so far. */
+  idsRead: number
+}
+
+/**
+ * Looks up many devices at once, skipping any ids that are missing, deleted, or
+ * unreadable. Ids are de-duplicated, and the lookup runs in batches to keep
+ * individual Couch requests small.
+ *
+ * This yields per batch rather than returning everything at once, so a caller
+ * working through hundreds of thousands of ids can report progress instead of
+ * going quiet for the whole lookup.
+ */
+export async function* streamDeviceBatchesByIds(
+  connections: DbConnections,
+  deviceIds: string[]
+): AsyncIterableIterator<DeviceBatch> {
+  const db = connections.couch.use(couchDevicesSetup.name)
+  const unique = [...new Set(deviceIds)]
+
+  for (let i = 0; i < unique.length; i += FETCH_BATCH_SIZE) {
+    const keys = unique.slice(i, i + FETCH_BATCH_SIZE)
+    const response = await db.fetch({ keys })
+
+    const deviceRows: DeviceRow[] = []
+    for (const row of response.rows) {
+      // Missing and deleted ids come back as rows without a document:
+      if ('error' in row || row.doc == null) continue
+      const couchDevice = asMaybe(asCouchDevice)(row.doc)
+      if (couchDevice == null) continue
+      deviceRows.push(makeDeviceRow(db, couchDevice))
+    }
+    yield { deviceRows, idsRead: Math.min(i + keys.length, unique.length) }
+  }
+}
+
 /**
  * Finds all the devices that have logged into this account.
  */
@@ -343,3 +445,100 @@ export async function* streamDevicesByLocation(
     yield makeDeviceRow(db, couchDevice)
   }
 }
+
+/**
+ * Builds a CouchDB start key of [apiKey, country, region, city], stopping at the
+ * first missing location field so the range stays a valid prefix.
+ */
+function makeApiKeyLocationStartKey(
+  apiKey: string,
+  location: { country?: string; region?: string; city?: string }
+): string[] {
+  const key = [apiKey]
+  for (const part of [location.country, location.region, location.city]) {
+    if (part == null) break
+    key.push(part)
+  }
+  return key
+}
+
+/**
+ * What the `apiKeyLocation` view stores alongside each device, so that an
+ * audience query can work from the index alone.
+ */
+export interface DeviceSummary {
+  deviceId: string
+  apiKey: string
+  country: string
+  region: string
+  city: string
+  deviceToken: string | undefined
+  ignoreMarketing: boolean
+  visited: Date
+}
+
+const asViewValue = asObject({
+  deviceToken: asOptional(asString),
+  ignoreMarketing: asOptional(asBoolean, false),
+  visited: asOptional(asDate, () => new Date(0))
+})
+
+// Rows to pull per request while paging through the view. Measured against
+// the production index: pages of 2,048 rows scanned at about 9,000 rows a
+// second and pages of 20,000 at about 18,000, with the bytes per row the
+// same, so the larger page halves a whole-country audience query. Each page
+// is about 6 MB of JSON, which the server parses comfortably.
+const VIEW_PAGE_SIZE = 20_000
+
+/**
+ * Streams a summary of every device registered under an API key, optionally
+ * narrowed by location. With no country at all this walks the key's whole
+ * slice of the index, every country included.
+ *
+ * This reads the view rows only. Fetching the documents would mean pulling
+ * hundreds of megabytes to read a handful of fields, since the view spans a
+ * whole country and the location filters are applied afterwards.
+ */
+export async function* streamDeviceSummariesByApiKeyLocation(
+  connections: DbConnections,
+  apiKey: string,
+  location: { country?: string; region?: string; city?: string }
+): AsyncIterableIterator<DeviceSummary> {
+  const db = connections.couch.use(couchDevicesSetup.name)
+  const startKey = makeApiKeyLocationStartKey(apiKey, location)
+  const endKey = [...startKey, 'zzzzzz']
+
+  let params: object = { start_key: startKey }
+  while (true) {
+    const response = await db.view('apiKeyLocation', 'apiKeyLocation', {
+      reduce: false,
+      include_docs: false,
+      limit: VIEW_PAGE_SIZE,
+      end_key: endKey,
+      ...params
+    })
+    const { rows } = response
+    if (rows.length === 0) return
+
+    for (const row of rows) {
+      const value = asMaybe(asViewValue)(row.value)
+      if (value == null) continue
+      // The location lives in the key, not the value:
+      const key = asMaybe(asArray(asString))(row.key) ?? []
+      yield {
+        ...value,
+        deviceId: row.id,
+        apiKey,
+        country: key[1] ?? '',
+        region: key[2] ?? '',
+        city: key[3] ?? ''
+      }
+    }
+
+    if (rows.length < VIEW_PAGE_SIZE) return
+    // Resume after the last row, which needs the doc id to break ties between
+    // devices sharing a location:
+    const last = rows[rows.length - 1]
+    params = { start_key: last.key, start_key_doc_id: last.id, skip: 1 }
+  }
+}

src/server/index.ts

@@ -5,6 +5,7 @@ import { makeExpressRoute } from 'serverlet/express'
 import { setupDatabases } from '../db/couchSetup'
 import { makeConnections, serverConfig } from '../serverConfig'
 import { logger } from '../util/logger'
+import { makeMarketingToolRouter } from './marketing/marketingTool'
 import { withLogging } from './middleware/withLogging'
 import { allRoutes } from './urls'
 
@@ -23,6 +24,11 @@ async function main(): Promise<void> {
   // Set up Express:
   const app = express()
   app.enable('trust proxy')
+  // The marketing push API, mounted before the Serverlet catch-all. It comes
+  // ahead of the body parser below because it installs its own with a larger
+  // limit, and whichever parser runs first wins:
+  app.use('/marketing', makeMarketingToolRouter(connections))
+
   app.use(express.json({ limit: '1mb' }))
   app.use('/', makeExpressRoute(server))