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

@@ -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`)

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
 

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",

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";

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<void>((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);
+  });
+});

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<DailyAlertSearchResult>;
+
+export type AlertSearchSchedulerPassInput = {
+  listUserIds?: () => Promise<string[]>;
+  runDaily?: AlertSearchUserRunner;
+};
+
+export type AlertSearchSchedulerPassResult = {
+  skipped: boolean;
+  userIds: string[];
+  attempted: number;
+  failed: number;
+};
+
+let intervalId: ReturnType<typeof setInterval> | 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<AlertSearchSchedulerPassResult> {
+  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;
+}

src/db/alertAuthorizationSqlite.ts

@@ -216,6 +216,19 @@ export const alertAuthorizationDb = {
     return row.n;
   },
 
+  async listDistinctUserIds(): Promise<string[]> {
+    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<AlertAuthorizationBatchRecord | undefined> {

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");