notification-wakeup-service/test/alertSearch/scheduler.test.ts

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 "../../src/db/alertAuthorizationSqlite.js";
import { db as fcmDb } from "../../src/db/fcmTokensSqlite.js";
import { closeDatabase } from "../../src/db/sqlite.js";
import type { DailyAlertSearchResult } from "../../src/alertSearch/daily.js";
import {
  isAlertSearchSchedulerPassInFlight,
  resetAlertSearchSchedulerPassGuard,
  runAlertSearchSchedulerPass,
  startAlertSearchScheduler,
  stopAlertSearchScheduler,
} from "../../src/alertSearch/scheduler.js";
import type { AlertSearchDigest } from "../../src/alertSearch/digest.js";

const USER_A = "did:ethr:0xusera";
const USER_B = "did:ethr:0xuserb";

function stubDailyResult(userId: string): DailyAlertSearchResult {
  return {
    userId,
    utcDay: null,
    batchId: null,
    jwtSequence: null,
    endorserOutcome: null,
    partnerOutcome: null,
    completed: false,
    consumed: false,
    digest: null,
  };
}

function eligibleDaily(userId: string): DailyAlertSearchResult {
  const digest: AlertSearchDigest = {
    completed: true,
    hasUpdates: true,
    totalCount: 3,
    counts: {
      claims: 3,
      personalPlanContributions: 0,
      trackedPlanUpdates: 0,
      trackedPlanClaims: 0,
      plansNearby: 0,
      profilesNearby: 0,
    },
    records: {
      claims: [],
      personalPlanContributions: [],
      trackedPlanUpdates: [],
      trackedPlanClaims: [],
      plansNearby: [],
      profilesNearby: [],
    },
    endorser: { outcome: "success", completed: true },
    partner: { outcome: "empty", completed: true },
  };
  return {
    userId,
    utcDay: "2026-08-14",
    batchId: "batch-1",
    jwtSequence: 1,
    endorserOutcome: "success",
    partnerOutcome: "empty",
    completed: true,
    consumed: true,
    digest,
  };
}

/** The day every seeded JWT belongs to, and an instant inside it. */
const DAY = "2026-08-28";
const NOW = new Date(`${DAY}T12:00:00.000Z`);

async function seedBatch(
  userId: string,
  batchId: string,
  notify?: { hour: number; minute: number }
) {
  await alertAuthorizationDb.replaceUnusedBatch({
    userId,
    batchId,
    notifyHourUtc: notify?.hour,
    notifyMinuteUtc: notify?.minute,
    jwts: [
      {
        sequence: 1,
        day: DAY,
        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({
      now: NOW,
      concurrency: 1,
      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({
      now: NOW,
      concurrency: 1,
      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("defers a user whose notify hour has not arrived", async () => {
    await seedBatch(USER_A, "batch-a", { hour: 18, minute: 0 });
    await seedBatch(USER_B, "batch-b");
    const ran: string[] = [];
    const result = await runAlertSearchSchedulerPass({
      now: NOW,
      concurrency: 1,
      runDaily: async (userId) => {
        ran.push(userId);
        return stubDailyResult(userId);
      },
    });
    assert.deepEqual(ran, [USER_B]);
    assert.equal(result.deferred, 1);
    assert.equal(result.attempted, 1);
  });

  it("runs that user once the hour passes", async () => {
    await seedBatch(USER_A, "batch-a", { hour: 18, minute: 0 });
    const ran: string[] = [];
    const result = await runAlertSearchSchedulerPass({
      now: new Date(`${DAY}T18:05:00.000Z`),
      concurrency: 1,
      runDaily: async (userId) => {
        ran.push(userId);
        return stubDailyResult(userId);
      },
    });
    assert.deepEqual(ran, [USER_A]);
    assert.equal(result.deferred, 0);
  });

  it("does not list a user whose day is already spent", async () => {
    await seedBatch(USER_A, "batch-a");
    const jwt = await alertAuthorizationDb.getUnusedForDay(USER_A, DAY);
    assert.ok(jwt);
    await alertAuthorizationDb.consumeUnusedJwt({ id: jwt.id, userId: USER_A });

    const ran: string[] = [];
    const result = await runAlertSearchSchedulerPass({
      now: NOW,
      concurrency: 1,
      runDaily: async (userId) => {
        ran.push(userId);
        return stubDailyResult(userId);
      },
    });
    assert.deepEqual(ran, []);
    assert.equal(result.attempted, 0);
    assert.equal(result.deferred, 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({
      now: NOW,
      concurrency: 1,
      runDaily: async (userId) => {
        await blocked;
        return stubDailyResult(userId);
      },
    });
    while (!isAlertSearchSchedulerPassInFlight()) {
      await Promise.resolve();
    }
    const second = await runAlertSearchSchedulerPass({
      now: NOW,
      concurrency: 1,
      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({
      now: NOW,
      concurrency: 1,
      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();
  });

  it("notifies once for an eligible digest and not for a non-eligible result", async () => {
    await seedBatch(USER_A, "batch-a");
    await seedBatch(USER_B, "batch-b");
    const notified: string[] = [];
    await runAlertSearchSchedulerPass({
      now: NOW,
      concurrency: 1,
      runDaily: async (userId) =>
        userId === USER_A ? eligibleDaily(userId) : stubDailyResult(userId),
      notify: async (result) => {
        if (result.digest?.hasUpdates) notified.push(result.userId);
      },
    });
    assert.deepEqual(notified, [USER_A]);
  });

  it("does not run alertSearch twice for two FCM devices on the same user", 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 notifyCalls: number[] = [];
    await runAlertSearchSchedulerPass({
      now: NOW,
      concurrency: 1,
      runDaily: async (userId) => {
        ran.push(userId);
        return eligibleDaily(userId);
      },
      notify: async () => {
        notifyCalls.push(1);
      },
    });
    assert.deepEqual(ran, [USER_A]);
    assert.equal(notifyCalls.length, 1);
  });

  it("does not notify again after the day's JWT was consumed", async () => {
    await seedBatch(USER_A, "batch-a");
    const notifies: Array<DailyAlertSearchResult["digest"]> = [];
    await runAlertSearchSchedulerPass({
      now: NOW,
      concurrency: 1,
      runDaily: async (userId) => eligibleDaily(userId),
      notify: async (result) => {
        notifies.push(result.digest);
      },
    });
    await runAlertSearchSchedulerPass({
      now: NOW,
      concurrency: 1,
      runDaily: async (userId) => stubDailyResult(userId),
      notify: async (result) => {
        notifies.push(result.digest);
      },
    });
    assert.equal(notifies.length, 2);
    assert.equal(notifies[0]?.hasUpdates, true);
    assert.equal(notifies[1], null);
  });

  it("does not suppress a later notification after an incomplete run", async () => {
    await seedBatch(USER_A, "batch-a");
    const eligible: boolean[] = [];
    const first = eligibleDaily(USER_A);
    const incompleteDigest = first.digest;
    assert.ok(incompleteDigest);
    await runAlertSearchSchedulerPass({
      now: NOW,
      concurrency: 1,
      runDaily: async () => ({
        ...first,
        completed: false,
        consumed: false,
        digest: {
          ...incompleteDigest,
          completed: false,
          hasUpdates: false,
        },
      }),
      notify: async (result) => {
        eligible.push(Boolean(result.consumed && result.digest?.hasUpdates));
      },
    });
    await runAlertSearchSchedulerPass({
      now: NOW,
      concurrency: 1,
      runDaily: async (userId) => eligibleDaily(userId),
      notify: async (result) => {
        eligible.push(Boolean(result.consumed && result.digest?.hasUpdates));
      },
    });
    assert.deepEqual(eligible, [false, true]);
  });

  it("keeps the daily result when notification throws", async () => {
    await seedBatch(USER_A, "batch-a");
    let dailyConsumed = false;
    const result = await runAlertSearchSchedulerPass({
      now: NOW,
      concurrency: 1,
      runDaily: async (userId) => {
        const daily = eligibleDaily(userId);
        dailyConsumed = daily.consumed;
        return daily;
      },
      notify: async () => {
        throw new Error("fcm down");
      },
    });
    assert.equal(result.failed, 0);
    assert.equal(dailyConsumed, true);
  });
});

describe("alertSearch scheduler isolation from FCM", () => {
  it("does not call sendPushToDevice or WAKEUP_PING", () => {
    const alertSched = readFileSync(
      path.join(process.cwd(), "src/alertSearch/scheduler.ts"),
      "utf8"
    );
    assert.equal(alertSched.includes("sendPushToDevice"), 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);
  });

  it("leaves WAKEUP_PING last_notified_at logic on sendPushToDevice", () => {
    const push = readFileSync(
      path.join(process.cwd(), "src/services/pushService.ts"),
      "utf8"
    );
    assert.equal(push.includes('type: "WAKEUP_PING"'), true);
    assert.equal(push.includes("sendAlertSearchPushToDevice"), true);
  });
});