Persist snooze until and reopen it with a periodic sweep (CL-7208)

apps/hub/src/inbox-unsnooze-sweep.ts

@@ -0,0 +1,132 @@
+// A minimal periodic loop that reopens a snoozed inbox item once its
+// `until` has passed (CL-7208) — mirroring `credential-expiry-sweep.ts`
+// and `routine-scheduler.ts`, the only other periodic loops in this hub,
+// rather than standing up a new generic scheduling primitive. Neither
+// `@corbits/notify`'s `createNotifyDispatcher` (shaped around sink
+// delivery — attempts, backoff, a `SinkRegistry` — not a status flip) nor
+// the routine scheduler (`RoutineStore` is routine-domain-specific) fits
+// "reopen a mailbox row when its snooze timer elapses" without bending
+// their semantics, so this repeats the same small `setInterval` + own
+// due-scan shape those two already use.
+//
+// `claimAndReopenSnooze` (in `@corbits/inbox`) does the actual claim: a
+// single transaction that deletes the due snooze row and flips the
+// message back to `open` together, so a throw here rolls back both and
+// leaves the row for the next tick to retry rather than orphaning it —
+// the mail-then-claim lesson CL-7209 applied to the credential-expiry
+// sweep, applied here to the claim itself.
+import { getLogger } from "@intx/log";
+import { reportError } from "@corbits/error-sink";
+import {
+  claimAndReopenSnooze,
+  findDueSnoozes,
+  type DueSnooze,
+} from "@corbits/inbox";
+import type { MailboxDb, MailboxEventBus } from "@corbits/mailbox";
+
+export type InboxUnsnoozeSweepStore = {
+  findDueSnoozes(now: Date): Promise<readonly DueSnooze[]>;
+  /** Atomically claims one due snooze row and reopens its message if it's
+   * still snoozed. Returns whether it actually reopened something — see
+   * `claimAndReopenSnooze`'s own doc comment for the false cases. */
+  claimAndReopen(row: DueSnooze, now: Date): Promise<boolean>;
+};
+
+export function createDrizzleInboxUnsnoozeSweepStore(
+  db: MailboxDb,
+): InboxUnsnoozeSweepStore {
+  return {
+    findDueSnoozes: (now) => findDueSnoozes(db, now),
+    claimAndReopen: (row, now) => claimAndReopenSnooze(db, row, now),
+  };
+}
+
+export type InboxUnsnoozeSweepDeps = {
+  store: InboxUnsnoozeSweepStore;
+  bus: Pick<MailboxEventBus, "publish">;
+  /** Injectable for deterministic tests; defaults to `Date.now`-backed wall time. */
+  now?: () => Date;
+};
+
+const POLL_INTERVAL_MS = 60 * 1000;
+const publishLog = getLogger(["hub", "inbox-unsnooze-sweep"]);
+
+function publishReopened(
+  bus: Pick<MailboxEventBus, "publish">,
+  row: DueSnooze,
+): void {
+  // Best-effort, matching every other mailbox event publish in this repo
+  // (see packages/inbox/src/routes.ts's own `publish` helper): a bus
+  // failure here must never turn an already-committed reopen into a
+  // reported sweep failure.
+  try {
+    bus.publish(
+      { tenantId: row.tenantId, principalId: row.principalId },
+      { type: "mailbox", id: row.messageId, op: "enrich" },
+    );
+  } catch (error) {
+    publishLog.error(
+      "mailbox reopen event publish failed for {id} on tenant {tenantId}, principal {principalId}: {error}",
+      {
+        id: row.messageId,
+        tenantId: row.tenantId,
+        principalId: row.principalId,
+        error: error instanceof Error ? error.message : String(error),
+      },
+    );
+  }
+}
+
+/**
+ * One sweep: reopen every snooze due at `at`. Exported (rather than kept
+ * as a closure) so a test can drive a single, deterministic pass without
+ * waiting on `setInterval`.
+ */
+export async function tickInboxUnsnoozeSweep(
+  deps: Pick<InboxUnsnoozeSweepDeps, "store" | "bus">,
+  at: Date,
+): Promise<void> {
+  const due = await deps.store.findDueSnoozes(at);
+  for (const row of due) {
+    try {
+      const reopened = await deps.store.claimAndReopen(row, at);
+      if (reopened) publishReopened(deps.bus, row);
+    } catch (error) {
+      reportError(error, {
+        operation: "inbox_unsnooze_sweep",
+        tenantId: row.tenantId,
+        extra: { messageId: row.messageId },
+      });
+      // The claim is transactional (delete + reopen together): a throw
+      // means neither happened, so the row is still there and the next
+      // tick retries it instead of leaving the item snoozed forever.
+    }
+  }
+}
+
+export function createInboxUnsnoozeSweep(deps: InboxUnsnoozeSweepDeps) {
+  const now = deps.now ?? (() => new Date());
+  let tickInFlight = false;
+
+  async function tick(): Promise<void> {
+    if (tickInFlight) return;
+    tickInFlight = true;
+    try {
+      await tickInboxUnsnoozeSweep({ store: deps.store, bus: deps.bus }, now());
+    } catch (err) {
+      reportError(err, { operation: "inbox_unsnooze_sweep_tick" });
+    } finally {
+      tickInFlight = false;
+    }
+  }
+
+  const interval = setInterval(() => void tick(), POLL_INTERVAL_MS);
+  if (typeof interval.unref === "function") interval.unref();
+
+  return {
+    tick,
+    stop(): void {
+      clearInterval(interval);
+    },
+  };
+}

apps/hub/src/index.ts

@@ -331,6 +331,10 @@ import {
   createCredentialExpirySweep,
   createDrizzleCredentialExpirySweepStore,
 } from "./credential-expiry-sweep";
+import {
+  createDrizzleInboxUnsnoozeSweepStore,
+  createInboxUnsnoozeSweep,
+} from "./inbox-unsnooze-sweep";
 
 import { type } from "arktype";
 import { betterAuth } from "better-auth";
@@ -2547,6 +2551,14 @@ export async function createHub(config: HubConfig) {
     },
   });
 
+  // Reopen a snoozed inbox item once its `until` has passed (CL-7208) — a
+  // light periodic sweep over `@corbits/inbox`'s own snooze table, on the
+  // same mailboxDb/mailboxBus every other mailbox consumer here shares.
+  const inboxUnsnoozeSweep = createInboxUnsnoozeSweep({
+    store: createDrizzleInboxUnsnoozeSweepStore(mailboxDb),
+    bus: mailboxBus,
+  });
+
   // Shared `FoldedRunsDeps` for every one-shot Myra prompt below (routine
   // drafting, agent-definition drafting): a real one-shot inference call
   // that launches a folded run, awaits its single reply, and tears the run
@@ -3496,6 +3508,7 @@ export async function createHub(config: HubConfig) {
       chatOrchestrator.dispose();
       routineScheduler.stop();
       credentialExpirySweep.stop();
+      inboxUnsnoozeSweep.stop();
       benchProvisioner.stop();
       await insightsUsage.close();
       await insightsLatency.close();

apps/hub/test/inbox-unsnooze-sweep.test.ts

@@ -0,0 +1,100 @@
+// The sweep's own orchestration — find due snoozes, claim-and-reopen each,
+// publish only on an actual reopen, and never drop a row on a claim
+// failure — against an in-memory fake store. Which rows are due, and the
+// transactional delete+reopen claim itself, are `@corbits/inbox`'s own
+// concern (`findDueSnoozes`/`claimAndReopenSnooze`); this only checks the
+// tick loop calls them correctly and behaves on what comes back.
+import { describe, expect, test } from "bun:test";
+import type { MailboxEvent, MailboxEventScope } from "@corbits/mailbox";
+import {
+  tickInboxUnsnoozeSweep,
+  type InboxUnsnoozeSweepStore,
+} from "../src/inbox-unsnooze-sweep";
+import type { DueSnooze } from "@corbits/inbox";
+
+function row(overrides: Partial<DueSnooze> = {}): DueSnooze {
+  return {
+    tenantId: "tnt_1",
+    principalId: "prn_1",
+    messageId: "msg_1",
+    ...overrides,
+  };
+}
+
+function fakeBus() {
+  const published: { scope: MailboxEventScope; event: MailboxEvent }[] = [];
+  return {
+    published,
+    bus: {
+      publish(scope: MailboxEventScope, event: MailboxEvent) {
+        published.push({ scope, event });
+      },
+    },
+  };
+}
+
+describe("tickInboxUnsnoozeSweep", () => {
+  test("reopens every due row and publishes a reopened event for each", async () => {
+    const due = [row({ messageId: "msg_1" }), row({ messageId: "msg_2" })];
+    const claimed: DueSnooze[] = [];
+    const store: InboxUnsnoozeSweepStore = {
+      findDueSnoozes: async () => due,
+      claimAndReopen: async (candidate) => {
+        claimed.push(candidate);
+        return true;
+      },
+    };
+    const { bus, published } = fakeBus();
+
+    await tickInboxUnsnoozeSweep({ store, bus }, new Date());
+
+    expect(claimed.map((c) => c.messageId)).toEqual(["msg_1", "msg_2"]);
+    expect(published).toHaveLength(2);
+    expect(published[0]?.event).toEqual({
+      type: "mailbox",
+      id: "msg_1",
+      op: "enrich",
+    });
+  });
+
+  test("does not publish when claimAndReopen reports it reopened nothing", async () => {
+    const store: InboxUnsnoozeSweepStore = {
+      findDueSnoozes: async () => [row()],
+      // False: another replica already claimed it, or the message had
+      // already left `snoozed` — either way, no event to publish.
+      claimAndReopen: async () => false,
+    };
+    const { bus, published } = fakeBus();
+
+    await tickInboxUnsnoozeSweep({ store, bus }, new Date());
+
+    expect(published).toHaveLength(0);
+  });
+
+  test("a throw from claimAndReopen is swallowed per-row and does not publish", async () => {
+    const due = [row({ messageId: "msg_bad" }), row({ messageId: "msg_ok" })];
+    const claimed: string[] = [];
+    const store: InboxUnsnoozeSweepStore = {
+      findDueSnoozes: async () => due,
+      claimAndReopen: async (candidate) => {
+        claimed.push(candidate.messageId);
+        if (candidate.messageId === "msg_bad") {
+          throw new Error("connection reset mid-claim");
+        }
+        return true;
+      },
+    };
+    const { bus, published } = fakeBus();
+
+    // Must not throw out of the tick — one bad row shouldn't abort the rest.
+    await tickInboxUnsnoozeSweep({ store, bus }, new Date());
+
+    expect(claimed).toEqual(["msg_bad", "msg_ok"]);
+    // Only the row that actually reopened publishes; the thrown claim is
+    // left for the next tick to retry (claimAndReopenSnooze's transaction
+    // rolls back on a throw, so the snooze row is still there) rather than
+    // being dropped.
+    expect(published).toHaveLength(1);
+    expect(published[0]?.event.id).toBe("msg_ok");
+  });
+});

packages/inbox/package.json

@@ -7,7 +7,8 @@
   "type": "module",
   "exports": {
     ".": "./src/index.ts",
-    "./client": "./src/client.ts"
+    "./client": "./src/client.ts",
+    "./migrations": "./src/migrations.ts"
   },
   "scripts": {
     "typecheck": "tsc --noEmit",
@@ -20,13 +21,14 @@
     "@intx/hub-api": "workspace:*",
     "@intx/log": "0.3.0",
     "arktype": "catalog:",
-    "hono": "catalog:"
+    "drizzle-orm": "catalog:",
+    "hono": "catalog:",
+    "postgres": "catalog:"
   },
   "devDependencies": {
     "@intx/db": "workspace:*",
     "@intx/hub-common": "0.3.0",
     "@types/bun": "catalog:",
-    "postgres": "catalog:",
     "typescript": "catalog:"
   }
 }

packages/inbox/src/index.ts

@@ -28,7 +28,14 @@ export {
   type CreateWorkbenchMailboxDeliveryOpts,
 } from "./delivery";
 export {
+  applyInboxMigrations,
   applyMailboxMigrations,
+  type ApplyInboxMigrationsReport,
   type ApplyMailboxMigrationsReport,
 } from "./migrations";
 export { createInboxRoutes, type CreateInboxRoutesDeps } from "./routes";
+export {
+  claimAndReopenSnooze,
+  findDueSnoozes,
+  type DueSnooze,
+} from "./snooze-store";