julien/remanso-jetstream
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: julien:7875d24d48f17eb8d4f9bb386b236730ff407708
Branches Tags
julien:main
...
compare: julien:49608cff201b3f2acb80dccd470820b62b370970
Branches Tags
julien:main

5 Commits
7875d24d48 ... 49608cff20

Author SHA1 Message Date
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
10 changed files with 426 additions and 318 deletions
Expand all files Collapse all files

290
jetstream.ts
View File

@@ -5,11 +5,12 @@ import {
getWebhooksByDidAndVerb, getWebhooksByDidAndVerb,
saveCursor, saveCursor,
upsertNote, upsertNote,
type WebhookVerb,
} from "./src/data/db.ts"; } from "./src/data/db.ts";
import { Note } from "./src/data/note.ts"; import { Note } from "./src/data/note.ts";
import { log } from "./src/log.ts"; import { log } from "./src/log.ts";
import { ensureIndex, indexNote, removeNote } from "./src/search/opensearch.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> => { const safeIndex = async (note: Note): Promise<void> => {
try { try {
@@ -27,190 +28,127 @@ const safeRemove = async (did: string, rkey: string): Promise<void> => {
} }
}; };
globalThis.addEventListener("unhandledrejection", (e) => { const bulkDeps: BulkDeps = {
log("[jetstream] unhandled rejection:", e.reason); getSubs: getWebhooksByDidAndVerb,
}); dispatch: dispatchAll,
globalThis.addEventListener("error", (e) => {
log("[jetstream] uncaught error:", e.error);
});
type WebhookTarget = {
method: string;
url: string;
token?: string;
}; };
const dispatchAll = async ( if (import.meta.main) {
webhooks: WebhookTarget[], globalThis.addEventListener("unhandledrejection", (e) => {
payload: Record<string, unknown>, log("[jetstream] unhandled rejection:", e.reason);
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(); globalThis.addEventListener("error", (e) => {
log(`[jetstream] starting with cursor: ${cursor ?? "none"}`); log("[jetstream] uncaught error:", e.error);
});
const jetstream = new Jetstream({ const cursor = getCursor();
wantedCollections: ["space.remanso.note"], log(`[jetstream] starting with cursor: ${cursor ?? "none"}`);
cursor: cursor ? Number(cursor) : undefined,
endpoint: "https://jetstream2.fr.hose.cam/subscribe", 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) => {
try {
const {
did,
commit: { rkey, record },
} = event;
log(`[jetstream] creating ${did}/${rkey}...`);
const note = record as unknown as Omit<Note, "did" | "rkey">;
upsertNote({ did, rkey, ...note });
log(`[jetstream] create ${did}/${rkey}: ${note.title}`);
await Promise.allSettled([
fireWebhooks(
did,
"create",
{ event: "create", did, rkey, ...note },
getWebhooksByDidAndVerb,
),
safeIndex({ did, rkey, ...note }),
]);
queueBulkCreate(did, { rkey, ...note }, bulkDeps);
} catch (error) {
log(`[jetstream] error on create:`, error);
}
});
jetstream.onUpdate("space.remanso.note", async (event) => {
try {
const {
did,
commit: { rkey, record },
} = event;
log(`[jetstream] updating ${did}/${rkey}...`);
const note = record as unknown as Omit<Note, "did" | "rkey">;
upsertNote({ did, rkey, ...note });
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 },
getWebhooksByDidAndVerb,
),
safeIndex({ did, rkey, ...note }),
]);
queueBulkCreate(did, { rkey, ...note }, bulkDeps);
} catch (error) {
log(`[jetstream] error on update:`, error);
}
});
jetstream.onDelete("space.remanso.note", async (event) => {
try {
const {
did,
commit: { rkey },
} = event;
log(`[jetstream] deleting ${did}/${rkey}...`);
deleteNote({ did, rkey });
log(`[jetstream] delete ${did}/${rkey}`);
await Promise.allSettled([
fireWebhooks(
did,
"delete",
{ event: "delete", did, rkey },
getWebhooksByDidAndVerb,
),
safeRemove(did, rkey),
]);
} catch (error) {
log(`[jetstream] error on delete:`, error);
}
});
jetstream.on("open", () => {
log("[jetstream] connected");
});
jetstream.on("close", () => {
log("[jetstream] connection closed");
});
jetstream.on("error", (error) => {
log("[jetstream] connection closed with error", error);
});
jetstream.onCreate("space.remanso.note", async (event) => {
try { try {
const { await ensureIndex();
did,
commit: { rkey, record },
} = event;
log(`[jetstream] creating ${did}/${rkey}...`);
const note = record as unknown as Omit<Note, "did" | "rkey">;
upsertNote({ did, rkey, ...note });
log(`[jetstream] create ${did}/${rkey}: ${note.title}`);
await Promise.allSettled([
fireWebhooks(did, "create", { event: "create", did, rkey, ...note }),
safeIndex({ did, rkey, ...note }),
]);
queueBulkCreate(did, { rkey, ...note });
} catch (error) { } catch (error) {
log(`[jetstream] error on create:`, error); log("[opensearch] ensureIndex failed at startup:", error);
} }
});
jetstream.onUpdate("space.remanso.note", async (event) => { log("[jetstream] launching");
try { jetstream.start();
const {
did,
commit: { rkey, record },
} = event;
log(`[jetstream] updating ${did}/${rkey}...`);
const note = record as unknown as Omit<Note, "did" | "rkey">;
upsertNote({ did, rkey, ...note });
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 }),
safeIndex({ did, rkey, ...note }),
]);
queueBulkCreate(did, { rkey, ...note });
} catch (error) {
log(`[jetstream] error on update:`, error);
}
});
jetstream.onDelete("space.remanso.note", async (event) => { setInterval(() => {
try { if (jetstream.cursor) {
const { saveCursor(jetstream.cursor);
did, }
commit: { rkey }, }, 5000);
} = event;
log(`[jetstream] deleting ${did}/${rkey}...`);
deleteNote({ did, rkey });
log(`[jetstream] delete ${did}/${rkey}`);
await Promise.allSettled([
fireWebhooks(did, "delete", { event: "delete", did, rkey }),
safeRemove(did, rkey),
]);
} catch (error) {
log(`[jetstream] error on delete:`, error);
}
});
jetstream.on("open", () => {
log("[jetstream] connected");
});
jetstream.on("close", () => {
log("[jetstream] connection closed");
});
jetstream.on("error", (error) => {
log("[jetstream] connection closed with error", error);
});
try {
await ensureIndex();
} catch (error) {
log("[opensearch] ensureIndex failed at startup:", error);
} }
log("[jetstream] launching");
jetstream.start();
setInterval(() => {
if (jetstream.cursor) {
saveCursor(jetstream.cursor);
}
}, 5000);

12
scripts/backfill-opensearch.ts
View File

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

80
server.ts
View File

@@ -19,13 +19,25 @@ type AuthCtx = {
response: { status: number; body: unknown }; 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 ( const requireDidOwnership = async (
ctx: AuthCtx, ctx: AuthCtx,
did: string, did: string,
): Promise<boolean> => { ): Promise<boolean> => {
let verifiedDid: string; let verifiedDid: string;
try { try {
verifiedDid = await authenticateRequest( verifiedDid = await authenticator(
ctx.request.headers.get("Authorization"), ctx.request.headers.get("Authorization"),
); );
} catch { } catch {
@@ -41,17 +53,18 @@ const requireDidOwnership = async (
return true; return true;
}; };
const ADMIN_DIDS = new Set( const adminDids = (): Set<string> =>
(Deno.env.get("ADMIN_DIDS") ?? "") new Set(
.split(",") (Deno.env.get("ADMIN_DIDS") ?? "")
.map((d) => d.trim()) .split(",")
.filter(Boolean), .map((d) => d.trim())
); .filter(Boolean),
);
const requireAdmin = async (ctx: AuthCtx): Promise<boolean> => { const requireAdmin = async (ctx: AuthCtx): Promise<boolean> => {
let verifiedDid: string; let verifiedDid: string;
try { try {
verifiedDid = await authenticateRequest( verifiedDid = await authenticator(
ctx.request.headers.get("Authorization"), ctx.request.headers.get("Authorization"),
); );
} catch { } catch {
@@ -59,7 +72,7 @@ const requireAdmin = async (ctx: AuthCtx): Promise<boolean> => {
ctx.response.body = { error: "Unauthorized" }; ctx.response.body = { error: "Unauthorized" };
return false; return false;
} }
if (!ADMIN_DIDS.has(verifiedDid)) { if (!adminDids().has(verifiedDid)) {
ctx.response.status = 403; ctx.response.status = 403;
ctx.response.body = { error: "Admin only" }; ctx.response.body = { error: "Admin only" };
return false; return false;
@@ -260,27 +273,34 @@ router.delete("/:did/webhooks/:id", async (ctx) => {
// ctx.response.status = 204; // 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-Origin", "*");
ctx.response.headers.set( ctx.response.headers.set(
"Access-Control-Allow-Methods", "Access-Control-Allow-Methods",
"GET, POST, PUT, DELETE, OPTIONS", "GET, POST, PUT, DELETE, OPTIONS",
); );
ctx.response.headers.set( ctx.response.headers.set(
"Access-Control-Allow-Headers", "Access-Control-Allow-Headers",
"Content-Type, Authorization", "Content-Type, Authorization",
); );
if (ctx.request.method === "OPTIONS") { if (ctx.request.method === "OPTIONS") {
ctx.response.status = 204; ctx.response.status = 204;
return; return;
} }
await next(); await next();
}); });
app.use(router.routes()); app.use(router.routes());
app.use(router.allowedMethods()); app.use(router.allowedMethods());
log("[server] listening on port 8080"); return app;
app.listen({ port: 8080 }); };
if (import.meta.main) {
const app = createApp();
log("[server] listening on port 8080");
app.listen({ port: 8080 });
}

2
src/auth/verify.ts
View File

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

71
src/data/db.ts
View File

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

66
src/jetstream/bulk.ts Normal file
View File

@@ -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: number;
};
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();
};

49
src/jetstream/webhooks.ts Normal file
View File

@@ -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}`);
};

83
src/migrations/apply.ts Normal file
View File

@@ -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);
`);
};

85
src/migrations/init.ts
View File

@@ -1,83 +1,6 @@
import { db } from "../data/db.ts"; import { Database } from "@db/sqlite";
import { applyMigrations } from "./apply.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);
`);
const db = new Database(Deno.env.get("SQLITE_PATH") ?? "notes.db");
applyMigrations(db);
db.close(); db.close();

6
src/search/opensearch.ts
View File

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