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

refactor(db): lazy-init connection and add test seam

Browse Source 
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.
This commit is contained in:
Julien Calixte
2026-06-07 22:04:12 +02:00
parent dc6daf3f54
commit 6760c434b4
2 changed files with 56 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
scripts/backfill-opensearch.ts
View File

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

71
src/data/db.ts
View File

@@ -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<
{ journal_mode: string }
>();
console.log(`[db] journal_mode=${row.journal_mode}, busy_timeout=10000`);
} catch (e) {
console.error("[db] failed to set PRAGMAs:", e);
}
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) {
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,