Count Stripe calls and finish activity backfill once

scripts/mutation/equivalent-mutants.txt

@@ -1036,8 +1036,8 @@ src/shared/cache-registry.ts:126:47  ?? → ||   # get(): Set|undefined — a Se
 
 # Stripe request null checks: both values are typed as string|null, never undefined,
 # so strict and loose null equality agree for every possible input.
-src/shared/stripe/request.ts:89:17  === → ==   # retryAfter is string|null; both comparisons are true only for null
-src/shared/stripe/request.ts:151:15  === → ==   # Headers.get returns string|null; both comparisons are true only for null
+src/shared/stripe/request.ts:90:17  === → ==   # retryAfter is string|null; both comparisons are true only for null
+src/shared/stripe/request.ts:152:15  === → ==   # Headers.get returns string|null; both comparisons are true only for null
 
 # A header part without "=" has no value. Its key cannot affect timestamp or
 # signature state: a timestamp assignment remains undefined and a v1 signature

src/shared/db/activity-log-backfill.ts

@@ -7,21 +7,16 @@
  * with the env key, re-encrypt under the owner's public key, write it back.
  * Only the public key is needed (no password), so it runs unattended.
  *
- * It is resumable without a cursor: a converted row no longer matches the
- * `enc:` prefix, so each batch shrinks the remaining set. A batch that finds
- * nothing leaves the task disabled until more legacy data exists.
+ * It is resumable without a row cursor: a converted row no longer matches the
+ * `enc:` prefix, so each batch shrinks the remaining set. The maintenance task
+ * stores a completion checkpoint after the final batch, avoiding future scans.
  */
 
 /* jscpd:ignore-start */
 import { decrypt, ENCRYPTION_PREFIX } from "#shared/crypto/encryption.ts";
 import { encryptWithOwnerKey } from "#shared/crypto/keys.ts";
 import type { EnvKeyEncrypted } from "#shared/crypto/sealed.ts";
-import {
-  executeBatch,
-  queryAll,
-  rowExists,
-  update,
-} from "#shared/db/client.ts";
+import { executeBatch, queryAll, update } from "#shared/db/client.ts";
 import { ACTIVITY_LOG_BACKFILL_BATCH } from "#shared/limits.ts";
 import { logDebug } from "#shared/logger.ts";
 
@@ -32,6 +27,8 @@ import { logDebug } from "#shared/logger.ts";
 // legacy env-key ciphertext.
 type LegacyRow = { id: number; message: EnvKeyEncrypted };
 
+export const ACTIVITY_LOG_BACKFILL_COMPLETE = "complete";
+
 /**
  * Re-encrypt one batch of legacy env-key rows to the owner key. Returns the
  * number of rows converted (0 when none remain). All the rewrites land in a
@@ -64,16 +61,12 @@ export const backfillActivityLogBatch = async (
   return rows.length;
 };
 
-export const hasLegacyActivityLog = (): Promise<boolean> =>
-  rowExists("SELECT 1 FROM activity_log WHERE message LIKE ? LIMIT 1", [
-    `${ENCRYPTION_PREFIX}%`,
-  ]);
-
 export const runActivityLogBackfill = async (
   publicKey: string,
-): Promise<void> => {
+): Promise<number> => {
   const converted = await backfillActivityLogBatch(publicKey);
   if (converted > 0) {
     logDebug("Backfill", `activity_log: re-encrypted ${converted} rows`);
   }
+  return converted;
 };

src/shared/maintenance/registry.ts

@@ -1,25 +1,33 @@
-import { hasLegacyActivityLog } from "#shared/db/activity-log-backfill.ts";
+import { ACTIVITY_LOG_BACKFILL_COMPLETE } from "#shared/db/activity-log-backfill.ts";
 import { settings } from "#shared/db/settings.ts";
 import {
+  ACTIVITY_LOG_BACKFILL_BATCH,
   ACTIVITY_LOG_BACKFILL_INTERVAL_MS,
   PRUNE_INTERVAL_MS,
 } from "#shared/limits.ts";
-import { defineMaintenanceTasks } from "#shared/maintenance/definition.ts";
+import {
+  defineMaintenanceTasks,
+  type MaintenanceTaskCheck,
+} from "#shared/maintenance/definition.ts";
 import { CONFIG_KEYS } from "#shared/settings/keys.ts";
 
 const FAILURE_RETRY_MS = 5 * 60 * 1000;
 
+const alwaysEnabled = (
+  settingsKeys: readonly string[],
+): MaintenanceTaskCheck => ({
+  enabled: () => true,
+  maxDatabaseCalls: 0,
+  maxExternalCalls: 0,
+  settingsKeys,
+});
+
 export const MAINTENANCE_TASKS = defineMaintenanceTasks([
   {
-    check: {
-      enabled: () => true,
-      maxDatabaseCalls: 0,
-      maxExternalCalls: 0,
-      settingsKeys: [
-        CONFIG_KEYS.AUTO_PURGE_ORPHANS,
-        CONFIG_KEYS.ORPHAN_PURGE_RETENTION,
-      ],
-    },
+    check: alwaysEnabled([
+      CONFIG_KEYS.AUTO_PURGE_ORPHANS,
+      CONFIG_KEYS.ORPHAN_PURGE_RETENTION,
+    ]),
     deadlineMs: 15_000,
     failureRetryIntervalMs: FAILURE_RETRY_MS,
     intervalMs: PRUNE_INTERVAL_MS,
@@ -35,24 +43,24 @@ export const MAINTENANCE_TASKS = defineMaintenanceTasks([
     wakePolicy: "organic_safe",
   },
   {
-    check: {
-      enabled: async () =>
-        Boolean(settings.publicKey) && hasLegacyActivityLog(),
-      maxDatabaseCalls: 1,
-      maxExternalCalls: 0,
-      settingsKeys: [CONFIG_KEYS.PUBLIC_KEY],
-    },
+    check: alwaysEnabled([CONFIG_KEYS.PUBLIC_KEY]),
     deadlineMs: 10_000,
     failureRetryIntervalMs: ACTIVITY_LOG_BACKFILL_INTERVAL_MS,
     intervalMs: ACTIVITY_LOG_BACKFILL_INTERVAL_MS,
     maxDatabaseCalls: 2,
     maxExternalCalls: 0,
     name: "activity_log_backfill",
-    run: async () => {
+    run: async ({ checkpoint, requestFollowUp, setCheckpoint }) => {
+      if (checkpoint === ACTIVITY_LOG_BACKFILL_COMPLETE) return;
       const { runActivityLogBackfill } = await import(
         "#shared/db/activity-log-backfill.ts"
       );
-      await runActivityLogBackfill(settings.publicKey);
+      const converted = await runActivityLogBackfill(settings.publicKey);
+      if (converted < ACTIVITY_LOG_BACKFILL_BATCH) {
+        setCheckpoint(ACTIVITY_LOG_BACKFILL_COMPLETE);
+      } else {
+        requestFollowUp();
+      }
     },
     wakePolicy: "organic_safe",
   },

src/shared/stripe/request.ts

@@ -1,6 +1,7 @@
 import * as v from "valibot";
 import { namedError } from "#shared/named-error.ts";
 import { delay } from "#shared/now.ts";
+import { countExternalSubrequest } from "#shared/subrequest-budget.ts";
 import { encodeStripeForm, type StripeFormValue } from "./form.ts";
 import { parseStripeErrorBody } from "./schemas.ts";
 
@@ -288,6 +289,7 @@ export const createStripeRequest = (
 
     async function attempt(retry: number): Promise<T> {
       let response: Response;
+      countExternalSubrequest("Stripe API request");
       try {
         response = await config.fetch(url, {
           ...(method === "POST" ? { body: encoded } : {}),

test/shared/db/activity-log-backfill.test.ts

@@ -5,11 +5,10 @@ import { ENCRYPTION_PREFIX } from "#shared/crypto/encryption.ts";
 import { HYBRID_PREFIX } from "#shared/crypto/keys.ts";
 import {
   backfillActivityLogBatch,
-  hasLegacyActivityLog,
   runActivityLogBackfill,
 } from "#shared/db/activity-log-backfill.ts";
 import { getAllActivityLog, logActivity } from "#shared/db/activityLog.ts";
-import { execute } from "#shared/db/client.ts";
+import { execute, queryOne } from "#shared/db/client.ts";
 import { settings } from "#shared/db/settings.ts";
 import { setSuppressDebugLogs } from "#shared/log-settings.ts";
 import { MAINTENANCE_TASKS } from "#shared/maintenance/registry.ts";
@@ -20,10 +19,11 @@ import {
   rawActivityMessage,
 } from "#test-utils/activity-log.ts";
 import { describeWithEnv } from "#test-utils/db.ts";
+import { recordQueries } from "#test-utils/record-queries.ts";
 import { withTestSession } from "#test-utils/session.ts";
 
 const captureBackfillLogs = async (
-  operation: () => Promise<void>,
+  operation: () => Promise<unknown>,
 ): Promise<string[]> => {
   setSuppressDebugLogs(false);
   const debugStub = stub(console, "debug");
@@ -116,12 +116,11 @@ describeWithEnv("db > activity log backfill", { db: true }, () => {
     expect(logs).toEqual([]);
   });
 
-  test("reports whether legacy work remains", async () => {
-    expect(await hasLegacyActivityLog()).toBe(false);
+  test("returns the size of each remaining batch", async () => {
+    expect(await runActivityLogBackfill(settings.publicKey)).toBe(0);
     await insertLegacyActivity("legacy");
-    expect(await hasLegacyActivityLog()).toBe(true);
-    await runActivityLogBackfill(settings.publicKey);
-    expect(await hasLegacyActivityLog()).toBe(false);
+    expect(await runActivityLogBackfill(settings.publicKey)).toBe(1);
+    expect(await runActivityLogBackfill(settings.publicKey)).toBe(0);
   });
 
   test("runs the legacy backfill through the maintenance registry", async () => {
@@ -132,6 +131,31 @@ describeWithEnv("db > activity log backfill", { db: true }, () => {
     expect((await rawActivityMessage(id)).startsWith(HYBRID_PREFIX)).toBe(true);
   });
 
+  test("a completed task does not scan the activity log again", async () => {
+    await maintenance.run(MAINTENANCE_TASKS);
+    expect(
+      await queryOne<{ checkpoint: string }>(
+        "SELECT checkpoint FROM maintenance_tasks WHERE name = 'activity_log_backfill'",
+      ),
+    ).toEqual({ checkpoint: "complete" });
+    await execute(
+      "UPDATE maintenance_tasks SET next_run_at = 0 WHERE name = 'activity_log_backfill'",
+    );
+    const queries: string[] = [];
+    const restore = recordQueries(queries);
+    try {
+      await maintenance.run(MAINTENANCE_TASKS);
+    } finally {
+      restore();
+    }
+
+    expect(
+      queries.filter((sql) =>
+        sql.includes("FROM activity_log WHERE message LIKE"),
+      ),
+    ).toEqual([]);
+  });
+
   test("surfaces a corrupt legacy message to the task runner", async () => {
     await execute(
       "INSERT INTO activity_log (message, created, listing_id, attendee_id) VALUES (?, ?, NULL, NULL)",

test/shared/maintenance/registry.test.ts

@@ -1,15 +1,20 @@
 import { expect } from "@std/expect";
 import { it as test } from "@std/testing/bdd";
-import { hasLegacyActivityLog } from "#shared/db/activity-log-backfill.ts";
-import { settings } from "#shared/db/settings.ts";
+import { ENCRYPTION_PREFIX } from "#shared/crypto/encryption.ts";
+import { HYBRID_PREFIX } from "#shared/crypto/keys.ts";
+import { executeBatch } from "#shared/db/client.ts";
 import {
+  ACTIVITY_LOG_BACKFILL_BATCH,
   MAINTENANCE_PRUNE_BATCH,
   PRUNE_UNUSED_STRINGS_RETENTION_MS,
 } from "#shared/limits.ts";
 import type { MaintenanceTaskDeclaration } from "#shared/maintenance/definition.ts";
 import { MAINTENANCE_TASKS } from "#shared/maintenance/registry.ts";
-import { nowMs } from "#shared/now.ts";
-import { insertLegacyActivity } from "#test-utils/activity-log.ts";
+import { nowIso, nowMs } from "#shared/now.ts";
+import {
+  insertLegacyActivity,
+  rawActivityMessage,
+} from "#test-utils/activity-log.ts";
 import { describeWithEnv } from "#test-utils/db.ts";
 import {
   insertLoginAttempt,
@@ -26,6 +31,7 @@ const taskNamed = (name: string): MaintenanceTaskDeclaration => {
 const runTask = (
   task: MaintenanceTaskDeclaration,
   requestFollowUp: () => void = () => {},
+  setCheckpoint: (checkpoint: string | null) => void = () => {},
 ): void | Promise<void> =>
   task.run({
     budget: {
@@ -34,7 +40,7 @@ const runTask = (
     checkpoint: null,
     deadline: Date.now() + 10_000,
     requestFollowUp,
-    setCheckpoint: () => {},
+    setCheckpoint,
   });
 
 describeWithEnv("maintenance registry", { db: true }, () => {
@@ -63,7 +69,7 @@ describeWithEnv("maintenance registry", { db: true }, () => {
       {
         check: {
           enabled: undefined,
-          maxDatabaseCalls: 1,
+          maxDatabaseCalls: 0,
           maxExternalCalls: 0,
           settingsKeys: ["public_key"],
         },
@@ -104,27 +110,41 @@ describeWithEnv("maintenance registry", { db: true }, () => {
   });
 
   test("the activity task enables and drains legacy rows", async () => {
-    await insertLegacyActivity("registry legacy");
+    const id = await insertLegacyActivity("registry legacy");
     const task = taskNamed("activity_log_backfill");
 
     expect(await task.check.enabled()).toBe(true);
     await runTask(task);
 
-    expect(await hasLegacyActivityLog()).toBe(false);
+    expect((await rawActivityMessage(id)).startsWith(HYBRID_PREFIX)).toBe(true);
   });
 
-  test("the activity task stays disabled when no legacy rows remain", async () => {
-    expect(await taskNamed("activity_log_backfill").check.enabled()).toBe(
-      false,
-    );
+  test("the activity task stays available to preserve its checkpoint", async () => {
+    expect(await taskNamed("activity_log_backfill").check.enabled()).toBe(true);
   });
 
-  test("the activity task stays disabled without an owner public key", async () => {
-    await insertLegacyActivity("legacy without owner key");
-    settings.setForTest({ public_key: "" });
+  test("the activity task requests a follow-up after a full batch", async () => {
+    const firstId = await insertLegacyActivity("full registry batch");
+    const message = await rawActivityMessage(firstId);
+    expect(message.startsWith(ENCRYPTION_PREFIX)).toBe(true);
+    await executeBatch(
+      Array.from({ length: ACTIVITY_LOG_BACKFILL_BATCH - 1 }, () => ({
+        args: [message, nowIso()],
+        sql: "INSERT INTO activity_log (message, created, listing_id, attendee_id) VALUES (?, ?, NULL, NULL)",
+      })),
+    );
+    let followUps = 0;
+    const checkpoints: (string | null)[] = [];
 
-    expect(await taskNamed("activity_log_backfill").check.enabled()).toBe(
-      false,
+    await runTask(
+      taskNamed("activity_log_backfill"),
+      () => {
+        followUps += 1;
+      },
+      (checkpoint) => checkpoints.push(checkpoint),
     );
+
+    expect(followUps).toBe(1);
+    expect(checkpoints).toEqual([]);
   });
 });

test/shared/maintenance/runner.test.ts

@@ -168,6 +168,31 @@ describeWithEnv("maintenance runner", { db: true }, () => {
     expect(calls).toEqual([]);
   });
 
+  test("a disabled task is removed without running", async () => {
+    await execute(
+      "INSERT INTO maintenance_tasks (name, next_run_at) VALUES ('disabled', 0)",
+    );
+    const tasks = defineMaintenanceTasks([
+      declaration(
+        "disabled",
+        () => {
+          throw new Error("disabled task ran");
+        },
+        {
+          check: { enabled: () => false },
+        },
+      ),
+    ]);
+
+    await runWithSubrequestBudget(() => maintenance.run(tasks));
+
+    expect(
+      await queryOne(
+        "SELECT name FROM maintenance_tasks WHERE name = 'disabled'",
+      ),
+    ).toBeNull();
+  });
+
   test("the default scheduled wake includes scheduled-only work", async () => {
     const calls: string[] = [];
     const tasks = defineMaintenanceTasks([