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.
refactor(search,auth): export pure helpers for direct testing
2026-06-07 22:14:03 +02:00 · 2026-06-07 22:06:49 +02:00 · 2026-06-07 22:06:19 +02:00 · 2026-06-07 22:05:12 +02:00 · 2026-06-07 22:04:12 +02:00 · 2026-06-07 22:01:57 +02:00
10 changed files with 426 additions and 318 deletions
--- a/jetstream.ts
+++ b/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,109 +28,30 @@ const safeRemove = async (did: string, rkey: string): Promise<void> => {
  }
 };

-globalThis.addEventListener("unhandledrejection", (e) => {
+const bulkDeps: BulkDeps = {
+  getSubs: getWebhooksByDidAndVerb,
+  dispatch: dispatchAll,
+};
+
+if (import.meta.main) {
+  globalThis.addEventListener("unhandledrejection", (e) => {
    log("[jetstream] unhandled rejection:", e.reason);
-});
+  });

-globalThis.addEventListener("error", (e) => {
+  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 cursor = getCursor();
+  log(`[jetstream] starting with cursor: ${cursor ?? "none"}`);

-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"}`);
-
-const jetstream = new Jetstream({
+  const jetstream = new Jetstream({
    wantedCollections: ["space.remanso.note"],
    cursor: cursor ? Number(cursor) : undefined,
    endpoint: "https://jetstream2.fr.hose.cam/subscribe",
-});
+  });

-jetstream.onCreate("space.remanso.note", async (event) => {
+  jetstream.onCreate("space.remanso.note", async (event) => {
    try {
      const {
        did,
@@ -140,16 +62,21 @@ 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);
    }
-});
+  });

-jetstream.onUpdate("space.remanso.note", async (event) => {
+  jetstream.onUpdate("space.remanso.note", async (event) => {
    try {
      const {
        did,
@@ -161,16 +88,21 @@ 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);
    }
-});
+  });

-jetstream.onDelete("space.remanso.note", async (event) => {
+  jetstream.onDelete("space.remanso.note", async (event) => {
    try {
      const {
        did,
@@ -180,37 +112,43 @@ 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) {
      log(`[jetstream] error on delete:`, error);
    }
-});
+  });

-jetstream.on("open", () => {
+  jetstream.on("open", () => {
    log("[jetstream] connected");
-});
+  });

-jetstream.on("close", () => {
+  jetstream.on("close", () => {
    log("[jetstream] connection closed");
-});
+  });

-jetstream.on("error", (error) => {
+  jetstream.on("error", (error) => {
    log("[jetstream] connection closed with error", error);
-});
+  });

-try {
+  try {
    await ensureIndex();
-} catch (error) {
+  } catch (error) {
    log("[opensearch] ensureIndex failed at startup:", error);
-}
+  }

-log("[jetstream] launching");
-jetstream.start();
+  log("[jetstream] launching");
+  jetstream.start();

-setInterval(() => {
+  setInterval(() => {
    if (jetstream.cursor) {
      saveCursor(jetstream.cursor);
    }
-}, 5000);
+  }, 5000);
+}
--- a/scripts/backfill-opensearch.ts
+++ b/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();
--- a/server.ts
+++ b/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,17 +53,18 @@ 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())
      .filter(Boolean),
-);
+  );

 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,9 +273,10 @@ router.delete("/:did/webhooks/:id", async (ctx) => {
 //   ctx.response.status = 204;
 // })

-const app = new Application();
+export const createApp = (): Application => {
+  const app = new Application();

-app.use(async (ctx, next) => {
+  app.use(async (ctx, next) => {
    ctx.response.headers.set("Access-Control-Allow-Origin", "*");
    ctx.response.headers.set(
      "Access-Control-Allow-Methods",
@@ -277,10 +291,16 @@ app.use(async (ctx, next) => {
      return;
    }
    await next();
-});
+  });

-app.use(router.routes());
-app.use(router.allowedMethods());
+  app.use(router.routes());
+  app.use(router.allowedMethods());

-log("[server] listening on port 8080");
-app.listen({ port: 8080 });
+  return app;
+};
+
+if (import.meta.main) {
+  const app = createApp();
+  log("[server] listening on port 8080");
+  app.listen({ port: 8080 });
+}
--- a/src/auth/verify.ts
+++ b/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, "/");
--- a/src/data/db.ts
+++ b/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;

-try {
-  db.exec("PRAGMA busy_timeout=10000");
-  db.exec("PRAGMA journal_mode=WAL");
-  const [row] = db.prepare("PRAGMA journal_mode").all<
+const openDefaultDb = (): Database => {
+  const handle = new Database(Deno.env.get("SQLITE_PATH") ?? "notes.db");
+  try {
+    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) {
+  } 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,
--- a/src/jetstream/bulk.ts
+++ b/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();
+};
--- a/src/jetstream/webhooks.ts
+++ b/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}`);
+};
--- a/src/migrations/apply.ts
+++ b/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);
+  `);
+};
--- a/src/migrations/init.ts
+++ b/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();
--- a/src/search/opensearch.ts
+++ b/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 => {
Author	SHA1	Message	Date
Julien Calixte	62a55fd3e4	fix(jetstream): use ReturnType<typeof setTimeout> for bulk timer All checks were successful CI / check (push) Successful in 14s Details 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.	2026-06-07 22:14:03 +02:00
Julien Calixte	49608cff20	refactor(search,auth): export pure helpers for direct testing encodeCursor/decodeCursor/firstSnippet (opensearch) and decodeJwtPayload (verify) become public exports so tests can hit them without round-tripping through fetch.	2026-06-07 22:06:49 +02:00
Julien Calixte	01048c753e	refactor(jetstream): extract webhook dispatch and bulk-create modules dispatchAll, fireWebhooks, and the bulk-create debounce buffer move out of jetstream.ts into src/jetstream/{webhooks,bulk}.ts. fireWebhooks and the bulk helpers now take their DB lookup and dispatch as parameters so tests can inject stubs. Top-level Jetstream construction, event handlers, ensureIndex call, and the cursor-saving interval are gated behind import.meta.main so importing jetstream.ts no longer opens a WebSocket.	2026-06-07 22:06:19 +02:00
Julien Calixte	8964f66184	refactor(server): extract createApp and add auth/admin test seams createApp() builds the Application without listening, so tests can call app.handle() directly. The listen call is gated behind import.meta.main so importing server.ts no longer boots a port. ADMIN_DIDS is read inside requireAdmin on each call (was frozen at module load), and a mutable authenticator indirection lets tests stub authenticateRequest via _setAuthenticatorForTest.	2026-06-07 22:05:12 +02:00
Julien Calixte	6760c434b4	refactor(db): lazy-init connection and add test seam The Database handle is now constructed on first use and held in a module-level slot. New _setDbForTest and _resetDbForTest exports let tests swap in a :memory: handle without rewriting callers. closeDb() replaces the previous db.close() pattern used by short-lived scripts. Adds listDistinctNoteDids() to keep the backfill script from poking the raw connection.	2026-06-07 22:04:12 +02:00
Julien Calixte	dc6daf3f54	refactor(migrations): extract applyMigrations as callable function	2026-06-07 22:01:57 +02:00