chore(workers): warn at boot on legacy NULL consecutive_failures_by_kind rows

Post-v1.4.47 the legacy column was dropped; rows that still carry
consecutiveFailuresByKind: null alert two strikes later than they
should because max([1]) = 1 vs max(N+1, 1) = N+1 >= threshold.
The probe is fire-and-forget at startup, surfaces a single Wide
Event warning with the row count, idempotent across reboots.
Closes M1 in v1.4.48 backlog.

src/lib/jobs/__tests__/integration-status-null-probe.test.ts

@@ -0,0 +1,75 @@
+/**
+ * Unit coverage for the v1.4.48 M1 worker-boot probe that flags
+ * legacy `integration_statuses` rows whose JSON failure bucket is
+ * still NULL.
+ *
+ * The contract under test:
+ *   - emits a single `prisma.integrationStatus.count` against the
+ *     `consecutiveFailuresByKind: { equals: null }` predicate;
+ *   - when the count is > 0, the wrapping Wide Event is elevated to
+ *     `warn` and the count is annotated under
+ *     `meta.integration_status_null_buckets`;
+ *   - when the count is 0, the event stays at `info` and no warning
+ *     is added.
+ */
+import { describe, it, expect, vi } from "vitest";
+import type { PrismaClient } from "@/generated/prisma/client";
+import { WideEventBuilder } from "@/lib/logging/event-builder";
+import { eventStorage } from "@/lib/logging/context";
+import { probeIntegrationStatusNullBuckets } from "../integration-status-null-probe";
+
+function makePrismaMock(count: number) {
+  return {
+    $queryRaw: vi.fn().mockResolvedValue([{ count: BigInt(count) }]),
+  } as unknown as PrismaClient;
+}
+
+async function runWithEvent<T>(
+  fn: (evt: WideEventBuilder) => Promise<T>,
+): Promise<{ evt: WideEventBuilder; value: T }> {
+  const evt = new WideEventBuilder("background");
+  const value = await eventStorage.run(evt, () => fn(evt));
+  return { evt, value };
+}
+
+describe("probeIntegrationStatusNullBuckets", () => {
+  it("queries with the right predicate and returns the count", async () => {
+    const prisma = makePrismaMock(3);
+    const { value } = await runWithEvent(() =>
+      probeIntegrationStatusNullBuckets(prisma),
+    );
+
+    expect(value).toEqual({ count: 3 });
+    const queryRaw = prisma.$queryRaw as ReturnType<typeof vi.fn>;
+    expect(queryRaw).toHaveBeenCalledTimes(1);
+    const sql = (queryRaw.mock.calls[0][0] as TemplateStringsArray).join("");
+    expect(sql).toMatch(/consecutive_failures_by_kind\s+IS\s+NULL/i);
+    expect(sql).toMatch(/integration_statuses/);
+  });
+
+  it("elevates the event to warn and annotates the count when > 0", async () => {
+    const prisma = makePrismaMock(2);
+    const { evt } = await runWithEvent(() =>
+      probeIntegrationStatusNullBuckets(prisma),
+    );
+    evt.finish();
+    const json = evt.toJSON();
+
+    expect(json.level).toBe("warn");
+    expect(json.warnings?.[0]).toMatch(/2 row\(s\) still carry consecutiveFailuresByKind=NULL/);
+    expect(json.meta?.integration_status_null_buckets).toBe(2);
+  });
+
+  it("stays at info and emits no warning when count is 0", async () => {
+    const prisma = makePrismaMock(0);
+    const { evt } = await runWithEvent(() =>
+      probeIntegrationStatusNullBuckets(prisma),
+    );
+    evt.finish();
+    const json = evt.toJSON();
+
+    expect(json.level).toBe("info");
+    expect(json.warnings).toBeUndefined();
+    expect(json.meta?.integration_status_null_buckets).toBeUndefined();
+  });
+});

src/lib/jobs/integration-status-null-probe.ts

@@ -0,0 +1,49 @@
+/**
+ * v1.4.48 M1 — boot-time probe for legacy `IntegrationStatus` rows that
+ * still carry `consecutiveFailuresByKind: NULL`.
+ *
+ * v1.4.47 W1 dropped the single-column `consecutiveFailures` integer in
+ * favour of the per-kind JSON bucket. The alert-ladder reads
+ * `Math.max(...Object.values(buckets))` against the threshold — but a
+ * row that survives with the JSON column still NULL deterministically
+ * counts as `max([1]) = 1` on its first failure (the fall-back signal
+ * the writer emits when the row is absent), so the operator notices
+ * two strikes later than they did pre-v1.4.47 where the legacy column
+ * back-stopped the maths.
+ *
+ * Operator judgement ("every active row has been written under
+ * v1.4.43+, so the column is non-null") is not a code invariant. This
+ * probe is the runtime version of that invariant: one count query at
+ * worker boot, a single Wide-Event warning if any such row survives.
+ * Fire-and-forget — a probe failure must never block boot. Idempotent
+ * across reboots.
+ */
+import type { PrismaClient } from "@/generated/prisma/client";
+import { annotate, getEvent } from "@/lib/logging/context";
+
+export interface IntegrationStatusNullProbeResult {
+  count: number;
+}
+
+type PrismaCountClient = Pick<PrismaClient, "$queryRaw">;
+
+export async function probeIntegrationStatusNullBuckets(
+  prisma: PrismaCountClient,
+): Promise<IntegrationStatusNullProbeResult> {
+  // Prisma's typed `count` rejects a literal `null` against a `Json?`
+  // column without going through `Prisma.JsonNullValueFilter`. The raw
+  // `IS NULL` is the simpler shape and matches the backlog spec.
+  const rows = await prisma.$queryRaw<Array<{ count: bigint }>>`
+    SELECT COUNT(*)::bigint AS count
+      FROM integration_statuses
+     WHERE consecutive_failures_by_kind IS NULL
+  `;
+  const count = Number(rows[0]?.count ?? 0);
+  if (count > 0) {
+    getEvent()?.addWarning(
+      `[integration-status-null-probe] ${count} row(s) still carry consecutiveFailuresByKind=NULL; alert ladder will trip two strikes late until they are rewritten`,
+    );
+    annotate({ meta: { integration_status_null_buckets: count } });
+  }
+  return { count };
+}

src/lib/jobs/reminder-worker.ts

@@ -98,6 +98,7 @@ import {
 } from "@/lib/measurements/drain-per-sample-cumulative";
 import { expireStaleInUseItems } from "@/lib/medications/inventory/service";
 import { rotateLegacyMoodLogSecrets } from "@/lib/moodlog-secret";
+import { probeIntegrationStatusNullBuckets } from "@/lib/jobs/integration-status-null-probe";
 import { deleteMessage } from "@/lib/telegram";
 import { decrypt, encrypt } from "@/lib/crypto";
 import { syncMoodLogEntries } from "@/lib/moodlog/sync";
@@ -1762,6 +1763,20 @@ export async function startReminderWorker() {
     workerLog("error", "Failed to reconcile orphan ImportJob rows", err);
   }
 
+  // v1.4.48 M1 — boot probe for legacy `integration_statuses` rows
+  // that still carry `consecutive_failures_by_kind = NULL`. After
+  // v1.4.47 dropped the single-column fallback, such rows alert two
+  // strikes later than they did pre-upgrade. The probe is a single
+  // count query + Wide-Event warning if any survive; fire-and-forget
+  // so a probe failure never blocks worker boot.
+  try {
+    await withBackgroundEvent("worker.boot.integration_status_null_probe", async () => {
+      await probeIntegrationStatusNullBuckets(getWorkerPrisma());
+    });
+  } catch (err) {
+    workerLog("error", "integration-status-null-probe failed", err);
+  }
+
   // Schedule recurring cron jobs
   const schedules: [string, string][] = [
     [QUEUE_NAME, CHECK_INTERVAL_CRON],