From 209b346d16820472894c60fa462f9c70d16e14ab Mon Sep 17 00:00:00 2001 From: Jose Olarte III Date: Fri, 28 Aug 2026 16:39:22 +0800 Subject: [PATCH] Add a user-level alertSearch scheduler that runs each authorization-batch user once per pass, with an in-flight guard, without changing FCM wakeup. --- CHANGELOG.md | 5 + README.md | 4 +- package.json | 2 +- src/alertSearch/index.ts | 10 ++ src/alertSearch/scheduler.test.ts | 193 +++++++++++++++++++++++++++++ src/alertSearch/scheduler.ts | 117 +++++++++++++++++ src/db/alertAuthorizationSqlite.ts | 13 ++ src/index.ts | 2 + 8 files changed, 344 insertions(+), 2 deletions(-) create mode 100644 src/alertSearch/scheduler.test.ts create mode 100644 src/alertSearch/scheduler.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 8273f15..b115a30 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,11 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [0.1.10] - 2026.08.28 +### Added +- Dedicated user-level alertSearch scheduler (`startAlertSearchScheduler`) that calls `runDailyAlertSearch` once per authorization-batch user, with a process-local in-flight guard; independent of FCM `WAKEUP_PING` + + ## [0.1.9] - 2026.08.28 ### Added - `runDailyAlertSearch` selects today's unused delegated JWT from the stored batch timezone, runs `runAlertSearchCycle`, and consumes that JWT only when both Endorser and Partner complete (`success` or `empty`) diff --git a/README.md b/README.md index 2e9bc4d..df3c432 100644 --- a/README.md +++ b/README.md @@ -75,7 +75,9 @@ The **delegated** JWT is sent as `Authorization: Bearer`. Pass independent `endo `loadAlertSearchCursors` / `retrieveAlertSearch` / `advanceAlertSearchCursors` (or `runAlertSearchCycle`) persist those bounds per user DID in SQLite. Cursors advance only after a complete `success` retrieval (not `empty`, `pagination`, or errors). A Partner page of 50 rows that share the oldest `updatedAt` is `pagination` because exclusive `beforeDate` cannot drain timestamp ties. -`runDailyAlertSearch(userId, now?)` is a callable daily orchestrator (not a cron job). It uses the latest batch's stored IANA timezone to pick today's unused delegated JWT, runs `runAlertSearchCycle` with that JWT, and marks that specific JWT consumed only when both Endorser and Partner complete (`success` or `empty`, including both empty). Pagination or source failures leave the JWT unused so the same day can be retried. An invalid stored timezone is an error; there is no fallback to the server timezone. This is not wired to the existing FCM scheduler. +`runDailyAlertSearch(userId, now?)` uses the latest batch's stored IANA timezone to pick today's unused delegated JWT, runs `runAlertSearchCycle` with that JWT, and marks that specific JWT consumed only when both Endorser and Partner complete (`success` or `empty`, including both empty). Pagination or source failures leave the JWT unused so the same day can be retried. An invalid stored timezone is an error; there is no fallback to the server timezone. + +`startAlertSearchScheduler()` (started from `src/index.ts` next to the FCM scheduler) is a **separate** user-level job. It lists distinct `userId`s from `alert_authorization_batches` and calls `runDailyAlertSearch` once per user. It does not use `fcm_registrations`, does not call `sendPushToDevice`, and does not change `WAKEUP_PING`. A process-local in-flight flag skips a tick if a pass is still running. No notification content is built from alertSearch results yet. ## Storage diff --git a/package.json b/package.json index 42bd605..3d7333b 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "notification-wakeup-service", - "version": "0.1.9", + "version": "0.1.10", "private": true, "type": "module", "packageManager": "pnpm@11.4.0", diff --git a/src/alertSearch/index.ts b/src/alertSearch/index.ts index 31aabef..20981b0 100644 --- a/src/alertSearch/index.ts +++ b/src/alertSearch/index.ts @@ -48,3 +48,13 @@ export type { DailyAlertSearchCycleInput, DailyAlertSearchResult, } from "./daily.js"; +export { + ALERT_SEARCH_SCHEDULER_INTERVAL_MS, + runAlertSearchSchedulerPass, + startAlertSearchScheduler, + stopAlertSearchScheduler, +} from "./scheduler.js"; +export type { + AlertSearchSchedulerPassInput, + AlertSearchSchedulerPassResult, +} from "./scheduler.js"; diff --git a/src/alertSearch/scheduler.test.ts b/src/alertSearch/scheduler.test.ts new file mode 100644 index 0000000..e968287 --- /dev/null +++ b/src/alertSearch/scheduler.test.ts @@ -0,0 +1,193 @@ +import assert from "node:assert/strict"; +import { readFileSync } from "node:fs"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { afterEach, beforeEach, describe, it } from "node:test"; +import { alertAuthorizationDb } from "../db/alertAuthorizationSqlite.js"; +import { db as fcmDb } from "../db/fcmTokensSqlite.js"; +import { closeDatabase } from "../db/sqlite.js"; +import type { DailyAlertSearchResult } from "./daily.js"; +import { + isAlertSearchSchedulerPassInFlight, + resetAlertSearchSchedulerPassGuard, + runAlertSearchSchedulerPass, + startAlertSearchScheduler, + stopAlertSearchScheduler, +} from "./scheduler.js"; + +const USER_A = "did:ethr:0xusera"; +const USER_B = "did:ethr:0xuserb"; + +function stubDailyResult(userId: string): DailyAlertSearchResult { + return { + userId, + localDay: null, + batchId: null, + jwtSequence: null, + endorserOutcome: null, + partnerOutcome: null, + completed: false, + consumed: false, + }; +} + +async function seedBatch(userId: string, batchId: string) { + await alertAuthorizationDb.replaceUnusedBatch({ + userId, + batchId, + timezone: "America/Denver", + jwts: [ + { + sequence: 1, + day: "2026-08-28", + jwt: `jwt-${userId}`, + nbf: 1, + exp: 2, + }, + ], + }); +} + +async function seedDevice(userId: string, deviceId: string, fcmToken: string) { + await fcmDb.upsert({ + userId, + deviceId, + fcmToken, + platform: "ios", + updatedAt: new Date(), + }); +} + +describe("alertSearch scheduler pass", () => { + let dir: string; + let previousDataDir: string | undefined; + + beforeEach(async () => { + previousDataDir = process.env.NOTIFY_DATA_DIR; + dir = await mkdtemp(path.join(tmpdir(), "alert-search-sched-")); + process.env.NOTIFY_DATA_DIR = dir; + closeDatabase(); + stopAlertSearchScheduler(); + resetAlertSearchSchedulerPassGuard(); + }); + + afterEach(async () => { + stopAlertSearchScheduler(); + resetAlertSearchSchedulerPassGuard(); + closeDatabase(); + if (previousDataDir === undefined) { + delete process.env.NOTIFY_DATA_DIR; + } else { + process.env.NOTIFY_DATA_DIR = previousDataDir; + } + await rm(dir, { recursive: true, force: true }); + }); + + it("runs alertSearch once for a user with multiple FCM devices", async () => { + await seedBatch(USER_A, "batch-a"); + await seedDevice(USER_A, "device-1", "token-1"); + await seedDevice(USER_A, "device-2", "token-2"); + const ran: string[] = []; + const result = await runAlertSearchSchedulerPass({ + runDaily: async (userId) => { + ran.push(userId); + return stubDailyResult(userId); + }, + }); + assert.equal(result.skipped, false); + assert.deepEqual(ran, [USER_A]); + assert.equal(result.attempted, 1); + }); + + it("runs alertSearch once per eligible user", async () => { + await seedBatch(USER_A, "batch-a"); + await seedBatch(USER_B, "batch-b"); + const ran: string[] = []; + const result = await runAlertSearchSchedulerPass({ + runDaily: async (userId) => { + ran.push(userId); + return stubDailyResult(userId); + }, + }); + assert.deepEqual(ran, [USER_A, USER_B]); + assert.equal(result.attempted, 2); + assert.equal(result.failed, 0); + }); + + it("skips a tick while an alertSearch pass is in flight", async () => { + await seedBatch(USER_A, "batch-a"); + let release!: () => void; + const blocked = new Promise((resolve) => { + release = resolve; + }); + const first = runAlertSearchSchedulerPass({ + runDaily: async (userId) => { + await blocked; + return stubDailyResult(userId); + }, + }); + while (!isAlertSearchSchedulerPassInFlight()) { + await Promise.resolve(); + } + const second = await runAlertSearchSchedulerPass({ + runDaily: async () => { + throw new Error("second pass should not run daily"); + }, + }); + assert.equal(second.skipped, true); + assert.equal(second.attempted, 0); + release(); + const firstResult = await first; + assert.equal(firstResult.skipped, false); + assert.equal(firstResult.attempted, 1); + }); + + it("continues other users when one runDailyAlertSearch fails", async () => { + await seedBatch(USER_A, "batch-a"); + await seedBatch(USER_B, "batch-b"); + const ran: string[] = []; + const result = await runAlertSearchSchedulerPass({ + runDaily: async (userId) => { + ran.push(userId); + if (userId === USER_A) throw new Error("boom"); + return stubDailyResult(userId); + }, + }); + assert.deepEqual(ran, [USER_A, USER_B]); + assert.equal(result.failed, 1); + assert.equal(result.attempted, 2); + }); + + it("does not run a pass on start, and a second start is a no-op", async () => { + await seedBatch(USER_A, "batch-a"); + assert.equal(startAlertSearchScheduler(), true); + assert.equal(startAlertSearchScheduler(), false); + await new Promise((resolve) => setTimeout(resolve, 25)); + assert.equal(isAlertSearchSchedulerPassInFlight(), false); + stopAlertSearchScheduler(); + }); +}); + +describe("alertSearch scheduler isolation from FCM", () => { + it("does not call sendPushToDevice or read fcm_registrations", () => { + const alertSched = readFileSync( + path.join(process.cwd(), "src/alertSearch/scheduler.ts"), + "utf8" + ); + assert.equal(alertSched.includes("sendPushToDevice"), false); + assert.equal(alertSched.includes("fcmTokensSqlite"), false); + assert.equal(alertSched.includes("WAKEUP_PING"), false); + }); + + it("leaves the FCM scheduler on sendPushToDevice only", () => { + const fcmSched = readFileSync( + path.join(process.cwd(), "src/scheduler.ts"), + "utf8" + ); + assert.equal(fcmSched.includes("sendPushToDevice"), true); + assert.equal(fcmSched.includes("runDailyAlertSearch"), false); + assert.equal(fcmSched.includes("runAlertSearchSchedulerPass"), false); + assert.equal(fcmSched.includes("startAlertSearchScheduler"), false); + }); +}); diff --git a/src/alertSearch/scheduler.ts b/src/alertSearch/scheduler.ts new file mode 100644 index 0000000..3ed7c7f --- /dev/null +++ b/src/alertSearch/scheduler.ts @@ -0,0 +1,117 @@ +import { alertAuthorizationDb } from "../db/alertAuthorizationSqlite.js"; +import { errorMessage, formatElapsedMs } from "../util/formatElapsed.js"; +import { + runDailyAlertSearch, + type DailyAlertSearchResult, +} from "./daily.js"; + +/** Independent of the FCM wakeup interval; does not share that timer. */ +export const ALERT_SEARCH_SCHEDULER_INTERVAL_MS = 5 * 60 * 1000; + +export type AlertSearchUserRunner = ( + userId: string +) => Promise; + +export type AlertSearchSchedulerPassInput = { + listUserIds?: () => Promise; + runDaily?: AlertSearchUserRunner; +}; + +export type AlertSearchSchedulerPassResult = { + skipped: boolean; + userIds: string[]; + attempted: number; + failed: number; +}; + +let intervalId: ReturnType | undefined; +let passInFlight = false; + +export function isAlertSearchSchedulerPassInFlight(): boolean { + return passInFlight; +} + +/** + * One user-oriented alertSearch pass. Skips if a pass is already running. + * Does not send FCM or read fcm_registrations. + */ +export async function runAlertSearchSchedulerPass( + input: AlertSearchSchedulerPassInput = {} +): Promise { + if (passInFlight) { + console.log("[AlertSearchScheduler] Pass skipped (already in flight)"); + return { skipped: true, userIds: [], attempted: 0, failed: 0 }; + } + + passInFlight = true; + const passStarted = Date.now(); + console.log("[AlertSearchScheduler] Pass started"); + + try { + const listUserIds = + input.listUserIds ?? + (() => alertAuthorizationDb.listDistinctUserIds()); + const runDaily = input.runDaily ?? runDailyAlertSearch; + const userIds = await listUserIds(); + let failed = 0; + + for (const userId of userIds) { + try { + await runDaily(userId); + } catch (err) { + failed += 1; + console.error( + "[AlertSearchScheduler] User failed", + userId + ":", + errorMessage(err) + ); + } + } + + console.log( + "[AlertSearchScheduler] Pass completed in", + formatElapsedMs(Date.now() - passStarted) + ",", + `attempted ${userIds.length}, failed ${failed}` + ); + return { + skipped: false, + userIds, + attempted: userIds.length, + failed, + }; + } catch (err) { + console.error( + "[AlertSearchScheduler] Pass failed in", + formatElapsedMs(Date.now() - passStarted) + ":", + errorMessage(err) + ); + throw err; + } finally { + passInFlight = false; + } +} + +/** + * Starts a dedicated interval. Does not run a pass immediately (same as FCM). + * Calling twice is a no-op. Independent of the FCM device wakeup timer. + */ +export function startAlertSearchScheduler(): boolean { + if (intervalId !== undefined) return false; + + intervalId = setInterval(() => { + void runAlertSearchSchedulerPass(); + }, ALERT_SEARCH_SCHEDULER_INTERVAL_MS); + return true; +} + +export function stopAlertSearchScheduler(): void { + if (intervalId !== undefined) { + clearInterval(intervalId); + intervalId = undefined; + } +} + +/** Test helper: drop the in-flight flag after an interrupted pass. */ +export function resetAlertSearchSchedulerPassGuard(): void { + passInFlight = false; +} diff --git a/src/db/alertAuthorizationSqlite.ts b/src/db/alertAuthorizationSqlite.ts index 7d4d69c..6f96f8e 100644 --- a/src/db/alertAuthorizationSqlite.ts +++ b/src/db/alertAuthorizationSqlite.ts @@ -216,6 +216,19 @@ export const alertAuthorizationDb = { return row.n; }, + async listDistinctUserIds(): Promise { + const rows = getDatabase() + .prepare( + ` + SELECT DISTINCT user_id + FROM alert_authorization_batches + ORDER BY user_id + ` + ) + .all() as { user_id: string }[]; + return rows.map((row) => row.user_id); + }, + async getLatestBatch( userId: string ): Promise { diff --git a/src/index.ts b/src/index.ts index 705b2a4..1ddeac7 100644 --- a/src/index.ts +++ b/src/index.ts @@ -4,6 +4,7 @@ import express from "express"; import "./services/firebase.js"; import { debugRouter } from "./routes/debug.js"; import { notificationsRouter } from "./routes/notifications.js"; +import { startAlertSearchScheduler } from "./alertSearch/scheduler.js"; import { startScheduler } from "./scheduler.js"; const app = express(); @@ -28,6 +29,7 @@ app.use("/notifications", notificationsRouter); app.use("/debug", debugRouter); startScheduler(); +startAlertSearchScheduler(); app.listen(port, () => { console.log("* Running backend");