fix(jetstream): use ReturnType<typeof setTimeout> for bulk timer

setTimeout returns an opaque Timeout type in newer Deno
releases, so the previous timer: number annotation fails type
checking in CI. Using ReturnType<typeof setTimeout> is portable
across Deno versions and runtimes.

jetstream.ts

@@ -5,11 +5,12 @@ import {
   getWebhooksByDidAndVerb,
   saveCursor,
   upsertNote,
-  type WebhookVerb,
 } from "./src/data/db.ts";
 import { Note } from "./src/data/note.ts";
 import { log } from "./src/log.ts";
 import { ensureIndex, indexNote, removeNote } from "./src/search/opensearch.ts";
+import { dispatchAll, fireWebhooks } from "./src/jetstream/webhooks.ts";
+import { type BulkDeps, queueBulkCreate } from "./src/jetstream/bulk.ts";
 
 const safeIndex = async (note: Note): Promise<void> => {
   try {
@@ -27,6 +28,12 @@ const safeRemove = async (did: string, rkey: string): Promise<void> => {
   }
 };
 
+const bulkDeps: BulkDeps = {
+  getSubs: getWebhooksByDidAndVerb,
+  dispatch: dispatchAll,
+};
+
+if (import.meta.main) {
   globalThis.addEventListener("unhandledrejection", (e) => {
     log("[jetstream] unhandled rejection:", e.reason);
   });
@@ -35,91 +42,6 @@ globalThis.addEventListener("error", (e) => {
     log("[jetstream] uncaught error:", e.error);
   });
 
-type WebhookTarget = {
-  method: string;
-  url: string;
-  token?: string;
-};
-
-const dispatchAll = async (
-  webhooks: WebhookTarget[],
-  payload: Record<string, unknown>,
-  label: string,
-): Promise<void> => {
-  if (webhooks.length === 0) return;
-  const results = await Promise.allSettled(
-    webhooks.map(({ method, url, token }) => {
-      const hasBody = method !== "GET" && method !== "HEAD";
-      return fetch(url, {
-        method,
-        headers: {
-          ...(hasBody ? { "Content-Type": "application/json" } : {}),
-          ...(token ? { Authorization: `Bearer ${token}` } : {}),
-        },
-        body: hasBody ? JSON.stringify(payload) : undefined,
-      });
-    }),
-  );
-  for (const result of results) {
-    if (result.status === "rejected") {
-      log(`[jetstream] ${label} webhook error:`, result.reason);
-    }
-  }
-};
-
-const fireWebhooks = async (
-  did: string,
-  verb: WebhookVerb,
-  payload: Record<string, unknown>,
-): Promise<void> => {
-  const webhooks = getWebhooksByDidAndVerb(did, verb);
-  await dispatchAll(webhooks, payload, `${verb} ${did}`);
-};
-
-const BULK_CREATE_DEBOUNCE_MS = 500;
-
-type BulkBuffer = {
-  records: Record<string, unknown>[];
-  timer: number;
-};
-const bulkBuffers = new Map<string, BulkBuffer>();
-
-const flushBulkCreate = async (did: string): Promise<void> => {
-  const buffer = bulkBuffers.get(did);
-  if (!buffer) return;
-  bulkBuffers.delete(did);
-  const webhooks = getWebhooksByDidAndVerb(did, "bulk-create");
-  log(`[jetstream] bulk-create ${did}: ${buffer.records.length} record(s)`);
-  await dispatchAll(
-    webhooks,
-    { event: "bulk-create", did, records: buffer.records },
-    `bulk-create ${did}`,
-  );
-};
-
-// Buffered records are not persisted: if jetstream restarts mid-window, those
-// `bulk-create` notifications are lost. Subscribers reconcile on cold start
-// because the underlying notes are already saved to the `note` table.
-const queueBulkCreate = (
-  did: string,
-  record: Record<string, unknown>,
-): void => {
-  const existing = bulkBuffers.get(did);
-  if (existing) {
-    clearTimeout(existing.timer);
-    existing.records.push(record);
-    existing.timer = setTimeout(
-      () => flushBulkCreate(did),
-      BULK_CREATE_DEBOUNCE_MS,
-    );
-    return;
-  }
-  bulkBuffers.set(did, {
-    records: [record],
-    timer: setTimeout(() => flushBulkCreate(did), BULK_CREATE_DEBOUNCE_MS),
-  });
-};
-
   const cursor = getCursor();
   log(`[jetstream] starting with cursor: ${cursor ?? "none"}`);
 
@@ -140,10 +62,15 @@ jetstream.onCreate("space.remanso.note", async (event) => {
       upsertNote({ did, rkey, ...note });
       log(`[jetstream] create ${did}/${rkey}: ${note.title}`);
       await Promise.allSettled([
-      fireWebhooks(did, "create", { event: "create", did, rkey, ...note }),
+        fireWebhooks(
+          did,
+          "create",
+          { event: "create", did, rkey, ...note },
+          getWebhooksByDidAndVerb,
+        ),
         safeIndex({ did, rkey, ...note }),
       ]);
-    queueBulkCreate(did, { rkey, ...note });
+      queueBulkCreate(did, { rkey, ...note }, bulkDeps);
     } catch (error) {
       log(`[jetstream] error on create:`, error);
     }
@@ -161,10 +88,15 @@ jetstream.onUpdate("space.remanso.note", async (event) => {
       log(`[jetstream] update ${did}/${rkey}: ${note.title}`);
       // Updates fold into the `create` verb — subscribers reconcile by (did, rkey).
       await Promise.allSettled([
-      fireWebhooks(did, "create", { event: "create", did, rkey, ...note }),
+        fireWebhooks(
+          did,
+          "create",
+          { event: "create", did, rkey, ...note },
+          getWebhooksByDidAndVerb,
+        ),
         safeIndex({ did, rkey, ...note }),
       ]);
-    queueBulkCreate(did, { rkey, ...note });
+      queueBulkCreate(did, { rkey, ...note }, bulkDeps);
     } catch (error) {
       log(`[jetstream] error on update:`, error);
     }
@@ -180,7 +112,12 @@ jetstream.onDelete("space.remanso.note", async (event) => {
       deleteNote({ did, rkey });
       log(`[jetstream] delete ${did}/${rkey}`);
       await Promise.allSettled([
-      fireWebhooks(did, "delete", { event: "delete", did, rkey }),
+        fireWebhooks(
+          did,
+          "delete",
+          { event: "delete", did, rkey },
+          getWebhooksByDidAndVerb,
+        ),
         safeRemove(did, rkey),
       ]);
     } catch (error) {
@@ -214,3 +151,4 @@ setInterval(() => {
       saveCursor(jetstream.cursor);
     }
   }, 5000);
+}

scripts/backfill-opensearch.ts

@@ -9,7 +9,7 @@
 // Requires OPENSEARCH_URL to be set; falls back to a no-op otherwise.
 // SQLite path follows the same SQLITE_PATH convention as the server/jetstream.
 
-import { db } from "../src/data/db.ts";
+import { closeDb, listDistinctNoteDids } from "../src/data/db.ts";
 import { resolvePds } from "../src/auth/verify.ts";
 import { ensureIndex, indexNote } from "../src/search/opensearch.ts";
 import { log } from "../src/log.ts";
@@ -100,14 +100,12 @@ const main = async () => {
     Deno.exit(1);
   }
 
-  const rows = db.prepare("SELECT DISTINCT did FROM note ORDER BY did").all<
-    { did: string }
-  >();
-  log(`[backfill] ${rows.length} DID(s) to process`);
+  const dids = listDistinctNoteDids();
+  log(`[backfill] ${dids.length} DID(s) to process`);
 
   let totalIndexed = 0;
   let totalFailed = 0;
-  for (const { did } of rows) {
+  for (const did of dids) {
     try {
       const { indexed, failed } = await backfillDid(did);
       log(`[backfill] ${did}: indexed=${indexed} failed=${failed}`);
@@ -119,7 +117,7 @@ const main = async () => {
     }
   }
   log(`[backfill] done. indexed=${totalIndexed} failed=${totalFailed}`);
-  db.close();
+  closeDb();
 };
 
 await main();

server.ts

@@ -19,13 +19,25 @@ type AuthCtx = {
   response: { status: number; body: unknown };
 };
 
+let authenticator: typeof authenticateRequest = authenticateRequest;
+
+export const _setAuthenticatorForTest = (
+  fn: typeof authenticateRequest,
+): void => {
+  authenticator = fn;
+};
+
+export const _resetAuthenticatorForTest = (): void => {
+  authenticator = authenticateRequest;
+};
+
 const requireDidOwnership = async (
   ctx: AuthCtx,
   did: string,
 ): Promise<boolean> => {
   let verifiedDid: string;
   try {
-    verifiedDid = await authenticateRequest(
+    verifiedDid = await authenticator(
       ctx.request.headers.get("Authorization"),
     );
   } catch {
@@ -41,7 +53,8 @@ const requireDidOwnership = async (
   return true;
 };
 
-const ADMIN_DIDS = new Set(
+const adminDids = (): Set<string> =>
+  new Set(
     (Deno.env.get("ADMIN_DIDS") ?? "")
       .split(",")
       .map((d) => d.trim())
@@ -51,7 +64,7 @@ const ADMIN_DIDS = new Set(
 const requireAdmin = async (ctx: AuthCtx): Promise<boolean> => {
   let verifiedDid: string;
   try {
-    verifiedDid = await authenticateRequest(
+    verifiedDid = await authenticator(
       ctx.request.headers.get("Authorization"),
     );
   } catch {
@@ -59,7 +72,7 @@ const requireAdmin = async (ctx: AuthCtx): Promise<boolean> => {
     ctx.response.body = { error: "Unauthorized" };
     return false;
   }
-  if (!ADMIN_DIDS.has(verifiedDid)) {
+  if (!adminDids().has(verifiedDid)) {
     ctx.response.status = 403;
     ctx.response.body = { error: "Admin only" };
     return false;
@@ -260,6 +273,7 @@ router.delete("/:did/webhooks/:id", async (ctx) => {
 //   ctx.response.status = 204;
 // })
 
+export const createApp = (): Application => {
   const app = new Application();
 
   app.use(async (ctx, next) => {
@@ -282,5 +296,11 @@ app.use(async (ctx, next) => {
   app.use(router.routes());
   app.use(router.allowedMethods());
 
+  return app;
+};
+
+if (import.meta.main) {
+  const app = createApp();
   log("[server] listening on port 8080");
   app.listen({ port: 8080 });
+}

src/auth/verify.ts

@@ -7,7 +7,7 @@ interface DidDocument {
   service?: { id: string; serviceEndpoint: string }[];
 }
 
-function decodeJwtPayload(token: string): JwtPayload {
+export function decodeJwtPayload(token: string): JwtPayload {
   const parts = token.split(".");
   if (parts.length !== 3) throw new Error("Invalid JWT format");
   const payload = parts[1].replace(/-/g, "+").replace(/_/g, "/");

src/data/db.ts

@@ -1,18 +1,39 @@
 import { Database } from "@db/sqlite";
 import type { Note } from "./note.ts";
 
-export const db = new Database(Deno.env.get("SQLITE_PATH") ?? "notes.db");
+let _db: Database | null = null;
 
+const openDefaultDb = (): Database => {
+  const handle = new Database(Deno.env.get("SQLITE_PATH") ?? "notes.db");
   try {
-  db.exec("PRAGMA busy_timeout=10000");
-  db.exec("PRAGMA journal_mode=WAL");
-  const [row] = db.prepare("PRAGMA journal_mode").all<
+    handle.exec("PRAGMA busy_timeout=10000");
+    handle.exec("PRAGMA journal_mode=WAL");
+    const [row] = handle.prepare("PRAGMA journal_mode").all<
       { journal_mode: string }
     >();
     console.log(`[db] journal_mode=${row.journal_mode}, busy_timeout=10000`);
   } catch (e) {
     console.error("[db] failed to set PRAGMAs:", e);
   }
+  return handle;
+};
+
+const getDb = (): Database => _db ??= openDefaultDb();
+
+export const _setDbForTest = (handle: Database): void => {
+  _db = handle;
+};
+
+export const _resetDbForTest = (): void => {
+  _db = null;
+};
+
+export const closeDb = (): void => {
+  if (_db) {
+    _db.close();
+    _db = null;
+  }
+};
 
 type NoteRow = {
   did: string;
@@ -23,6 +44,7 @@ type NoteRow = {
 };
 
 export const getNotes = (cursor?: string, limit = 20) => {
+  const db = getDb();
   const notes = cursor
     ? db.prepare(
       "SELECT did, rkey, title, publishedAt, createdAt, language FROM note WHERE discoverable = 1 AND rkey < ? ORDER BY rkey DESC LIMIT ?",
@@ -38,6 +60,7 @@ export const getNotes = (cursor?: string, limit = 20) => {
 };
 
 export const getNotesByDid = (did: string, cursor?: string, limit = 20) => {
+  const db = getDb();
   const notes = cursor
     ? db.prepare(
       "SELECT did, rkey, title, publishedAt, createdAt, language FROM note WHERE discoverable = 1 AND did = ? AND rkey < ? ORDER BY rkey DESC LIMIT ?",
@@ -54,6 +77,7 @@ export const getNotesByDid = (did: string, cursor?: string, limit = 20) => {
 
 export const getNotesByDids = (dids: string[], cursor?: string, limit = 20) => {
   if (dids.length === 0) return { notes: [] };
+  const db = getDb();
   const placeholders = dids.map(() => "?").join(", ");
   const notes = cursor
     ? db.prepare(
@@ -70,18 +94,18 @@ export const getNotesByDids = (dids: string[], cursor?: string, limit = 20) => {
 };
 
 export const deleteNote = ({ did, rkey }: { did: string; rkey: string }) => {
-  db.exec("DELETE FROM note WHERE did = ? AND rkey = ?", did, rkey);
+  getDb().exec("DELETE FROM note WHERE did = ? AND rkey = ?", did, rkey);
 };
 
 export const getCursor = (): string | undefined => {
-  const row = db.prepare(
+  const row = getDb().prepare(
     "SELECT value FROM state WHERE key = 'cursor'",
   ).get<{ value: string }>();
   return row?.value;
 };
 
 export const saveCursor = (cursor: number) => {
-  db.exec(
+  getDb().exec(
     "INSERT OR REPLACE INTO state (key, value) VALUES ('cursor', ?)",
     String(cursor),
   );
@@ -101,6 +125,7 @@ type WebhookSubscriptionRow = {
 export const addWebhookSubscription = (
   { did, method, url, token, verb }: Omit<WebhookSubscriptionRow, "id">,
 ): WebhookSubscriptionRow => {
+  const db = getDb();
   db.exec(
     "INSERT INTO webhook_subscription (did, method, url, token, verb) VALUES (?, ?, ?, ?, ?)",
     did,
@@ -116,13 +141,13 @@ export const addWebhookSubscription = (
 };
 
 export const deleteWebhooksByDid = (did: string): void => {
-  db.exec("DELETE FROM webhook_subscription WHERE did = ?", did);
+  getDb().exec("DELETE FROM webhook_subscription WHERE did = ?", did);
 };
 
 export const deleteWebhookById = (
   { did, id }: { did: string; id: number },
 ): boolean => {
-  const result = db.prepare(
+  const result = getDb().prepare(
     "DELETE FROM webhook_subscription WHERE did = ? AND id = ?",
   ).run(did, id);
   return result > 0;
@@ -131,13 +156,13 @@ export const deleteWebhookById = (
 export const listWebhooksByDid = (
   did: string,
 ): Omit<WebhookSubscriptionRow, "token">[] => {
-  return db.prepare(
+  return getDb().prepare(
     "SELECT id, did, method, url, verb FROM webhook_subscription WHERE did = ? ORDER BY id DESC",
   ).all<Omit<WebhookSubscriptionRow, "token">>(did);
 };
 
 export const listAllWebhooks = (): Omit<WebhookSubscriptionRow, "token">[] => {
-  return db.prepare(
+  return getDb().prepare(
     "SELECT id, did, method, url, verb FROM webhook_subscription ORDER BY did, id",
   ).all<Omit<WebhookSubscriptionRow, "token">>();
 };
@@ -146,14 +171,20 @@ export const getWebhooksByDidAndVerb = (
   did: string,
   verb: WebhookVerb,
 ): WebhookSubscriptionRow[] => {
-  return db.prepare(
+  return getDb().prepare(
     "SELECT id, did, method, url, token, verb FROM webhook_subscription WHERE did = ? AND verb = ? ORDER BY id DESC LIMIT 10",
   ).all<WebhookSubscriptionRow>(did, verb);
 };
 
+export const listDistinctNoteDids = (): string[] => {
+  return getDb().prepare(
+    "SELECT DISTINCT did FROM note ORDER BY did",
+  ).all<{ did: string }>().map((row) => row.did);
+};
+
 export const upsertNote = (note: Note) => {
   const now = new Date().toISOString();
-  db.exec(
+  getDb().exec(
     `
     INSERT INTO note (
       title,

src/jetstream/bulk.ts

@@ -0,0 +1,66 @@
+import { log } from "../log.ts";
+import type { GetSubsByVerb, WebhookTarget } from "./webhooks.ts";
+
+export const BULK_CREATE_DEBOUNCE_MS = 500;
+
+export type BulkDeps = {
+  getSubs: GetSubsByVerb;
+  dispatch: (
+    webhooks: WebhookTarget[],
+    payload: Record<string, unknown>,
+    label: string,
+  ) => Promise<void>;
+  debounceMs?: number;
+};
+
+type BulkBuffer = {
+  records: Record<string, unknown>[];
+  timer: ReturnType<typeof setTimeout>;
+};
+
+const bulkBuffers = new Map<string, BulkBuffer>();
+
+export const flushBulkCreate = async (
+  did: string,
+  deps: BulkDeps,
+): Promise<void> => {
+  const buffer = bulkBuffers.get(did);
+  if (!buffer) return;
+  bulkBuffers.delete(did);
+  const webhooks = deps.getSubs(did, "bulk-create");
+  log(`[jetstream] bulk-create ${did}: ${buffer.records.length} record(s)`);
+  await deps.dispatch(
+    webhooks,
+    { event: "bulk-create", did, records: buffer.records },
+    `bulk-create ${did}`,
+  );
+};
+
+// Buffered records are not persisted: if jetstream restarts mid-window, those
+// `bulk-create` notifications are lost. Subscribers reconcile on cold start
+// because the underlying notes are already saved to the `note` table.
+export const queueBulkCreate = (
+  did: string,
+  record: Record<string, unknown>,
+  deps: BulkDeps,
+): void => {
+  const ms = deps.debounceMs ?? BULK_CREATE_DEBOUNCE_MS;
+  const existing = bulkBuffers.get(did);
+  if (existing) {
+    clearTimeout(existing.timer);
+    existing.records.push(record);
+    existing.timer = setTimeout(() => flushBulkCreate(did, deps), ms);
+    return;
+  }
+  bulkBuffers.set(did, {
+    records: [record],
+    timer: setTimeout(() => flushBulkCreate(did, deps), ms),
+  });
+};
+
+export const _resetBulkBuffersForTest = (): void => {
+  for (const buffer of bulkBuffers.values()) {
+    clearTimeout(buffer.timer);
+  }
+  bulkBuffers.clear();
+};

src/jetstream/webhooks.ts

@@ -0,0 +1,49 @@
+import type { WebhookVerb } from "../data/db.ts";
+import { log } from "../log.ts";
+
+export type WebhookTarget = {
+  method: string;
+  url: string;
+  token?: string;
+};
+
+export const dispatchAll = async (
+  webhooks: WebhookTarget[],
+  payload: Record<string, unknown>,
+  label: string,
+): Promise<void> => {
+  if (webhooks.length === 0) return;
+  const results = await Promise.allSettled(
+    webhooks.map(({ method, url, token }) => {
+      const hasBody = method !== "GET" && method !== "HEAD";
+      return fetch(url, {
+        method,
+        headers: {
+          ...(hasBody ? { "Content-Type": "application/json" } : {}),
+          ...(token ? { Authorization: `Bearer ${token}` } : {}),
+        },
+        body: hasBody ? JSON.stringify(payload) : undefined,
+      });
+    }),
+  );
+  for (const result of results) {
+    if (result.status === "rejected") {
+      log(`[jetstream] ${label} webhook error:`, result.reason);
+    }
+  }
+};
+
+export type GetSubsByVerb = (
+  did: string,
+  verb: WebhookVerb,
+) => WebhookTarget[];
+
+export const fireWebhooks = async (
+  did: string,
+  verb: WebhookVerb,
+  payload: Record<string, unknown>,
+  getSubs: GetSubsByVerb,
+): Promise<void> => {
+  const webhooks = getSubs(did, verb);
+  await dispatchAll(webhooks, payload, `${verb} ${did}`);
+};

src/migrations/apply.ts

@@ -0,0 +1,83 @@
+import type { Database } from "@db/sqlite";
+
+export const applyMigrations = (db: Database): void => {
+  db.exec(`
+    CREATE TABLE IF NOT EXISTS note (
+      title TEXT NOT NULL,
+      publishedAt DATETIME,
+      createdAt DATETIME NOT NULL,
+      did TEXT NOT NULL,
+      rkey TEXT NOT NULL,
+      discoverable INTEGER NOT NULL DEFAULT 1,
+      PRIMARY KEY (did, rkey)
+    );
+  `);
+
+  try {
+    db.exec(
+      `ALTER TABLE note ADD COLUMN listed INTEGER NOT NULL DEFAULT 1;`,
+    );
+  } catch {
+    // Column already exists — no-op
+  }
+
+  try {
+    db.exec(
+      `ALTER TABLE note ADD COLUMN language STRING;`,
+    );
+  } catch {
+    // Column already exists — no-op
+  }
+
+  try {
+    db.exec(`ALTER TABLE note RENAME COLUMN listed TO discoverable;`);
+  } catch {
+    // Column already renamed or doesn't exist — no-op
+  }
+
+  db.exec(`
+    CREATE TABLE IF NOT EXISTS state (
+      key TEXT PRIMARY KEY,
+      value TEXT NOT NULL
+    );
+  `);
+
+  db.exec(`
+    CREATE TABLE IF NOT EXISTS webhook_subscription (
+      id     INTEGER PRIMARY KEY,
+      did    TEXT NOT NULL,
+      method TEXT NOT NULL,
+      url    TEXT NOT NULL
+    );
+  `);
+
+  db.exec(`
+    CREATE INDEX IF NOT EXISTS idx_webhook_subscription_did
+      ON webhook_subscription(did);
+  `);
+
+  try {
+    db.exec(`ALTER TABLE webhook_subscription ADD COLUMN token TEXT;`);
+  } catch {
+    // Column already exists — no-op
+  }
+
+  try {
+    db.exec(
+      `ALTER TABLE webhook_subscription ADD COLUMN verb TEXT NOT NULL DEFAULT 'create';`,
+    );
+    db.exec(`
+      INSERT INTO webhook_subscription (did, method, url, token, verb)
+      SELECT did, method, url, token, 'delete'
+      FROM webhook_subscription
+      WHERE verb = 'create';
+    `);
+  } catch {
+    // Column already exists — backfill already happened on a previous run.
+  }
+
+  db.exec(`
+    CREATE INDEX IF NOT EXISTS idx_webhook_subscription_did_verb
+      ON webhook_subscription(did, verb);
+  `);
+};

src/migrations/init.ts

@@ -1,83 +1,6 @@
-import { db } from "../data/db.ts";
-
-db.exec(`
-  CREATE TABLE IF NOT EXISTS note (
-    title TEXT NOT NULL,
-    publishedAt DATETIME,
-    createdAt DATETIME NOT NULL,
-    did TEXT NOT NULL,
-    rkey TEXT NOT NULL,
-    discoverable INTEGER NOT NULL DEFAULT 1,
-    PRIMARY KEY (did, rkey)
-  );
-`);
-
-try {
-  db.exec(
-    `ALTER TABLE note ADD COLUMN listed INTEGER NOT NULL DEFAULT 1;`,
-  );
-} catch {
-  // Column already exists — no-op
-}
-
-try {
-  db.exec(
-    `ALTER TABLE note ADD COLUMN language STRING;`,
-  );
-} catch {
-  // Column already exists — no-op
-}
-
-try {
-  db.exec(`ALTER TABLE note RENAME COLUMN listed TO discoverable;`);
-} catch {
-  // Column already renamed or doesn't exist — no-op
-}
-
-db.exec(`
-  CREATE TABLE IF NOT EXISTS state (
-    key TEXT PRIMARY KEY,
-    value TEXT NOT NULL
-  );
-`);
-
-db.exec(`
-  CREATE TABLE IF NOT EXISTS webhook_subscription (
-    id     INTEGER PRIMARY KEY,
-    did    TEXT NOT NULL,
-    method TEXT NOT NULL,
-    url    TEXT NOT NULL
-  );
-`);
-
-db.exec(`
-  CREATE INDEX IF NOT EXISTS idx_webhook_subscription_did
-    ON webhook_subscription(did);
-`);
-
-try {
-  db.exec(`ALTER TABLE webhook_subscription ADD COLUMN token TEXT;`);
-} catch {
-  // Column already exists — no-op
-}
-
-try {
-  db.exec(
-    `ALTER TABLE webhook_subscription ADD COLUMN verb TEXT NOT NULL DEFAULT 'create';`,
-  );
-  db.exec(`
-    INSERT INTO webhook_subscription (did, method, url, token, verb)
-    SELECT did, method, url, token, 'delete'
-    FROM webhook_subscription
-    WHERE verb = 'create';
-  `);
-} catch {
-  // Column already exists — backfill already happened on a previous run.
-}
-
-db.exec(`
-  CREATE INDEX IF NOT EXISTS idx_webhook_subscription_did_verb
-    ON webhook_subscription(did, verb);
-`);
+import { Database } from "@db/sqlite";
+import { applyMigrations } from "./apply.ts";
 
+const db = new Database(Deno.env.get("SQLITE_PATH") ?? "notes.db");
+applyMigrations(db);
 db.close();

src/search/opensearch.ts

@@ -139,7 +139,7 @@ export type SearchOptions = {
   limit: number;
 };
 
-const decodeCursor = (cursor: string): [number, string] | undefined => {
+export const decodeCursor = (cursor: string): [number, string] | undefined => {
   try {
     const parsed = JSON.parse(atob(cursor));
     if (
@@ -157,10 +157,10 @@ const decodeCursor = (cursor: string): [number, string] | undefined => {
   return undefined;
 };
 
-const encodeCursor = (score: number, rkey: string): string =>
+export const encodeCursor = (score: number, rkey: string): string =>
   btoa(JSON.stringify([score, rkey]));
 
-const firstSnippet = (
+export const firstSnippet = (
   highlight: Record<string, string[]> | undefined,
   source: { content?: string; title?: string },
 ): string => {