1// Keeps the web node's copy of cache.sqlite in sync with the source of
2// truth. The SOT serves a server-sent-events feed at /db/events; this
3// module holds it open and pulls /db on every beacon. The server emits a
4// beacon on connect, so boot, reconnect, and missed-while-disconnected all
5// resolve through the same path — there is no polling and no TTL.
6//
7// Pulls are ETag-guarded (no-change is a tiny 304; the etag is persisted so
8// a reboot does not re-download an unchanged database), land in a temp
9// file, and are copied into .clover/cache.sqlite with sqlite's backup. The
10// file stays the same, so every open connection reads the new rows, and the
11// incremental graph sees the change like any other write.
12const console = log.scoped("sync");
13
14const token = process.env.CLOVER_SOT_KEY ?? null;
15const enabled = process.env.CLOVER_DB_SYNC !== "0" && token != null;
16const sotUrl = new URL(process.env.CLOVER_SOT_URL ?? "https://db.paperclover.net")
17 .toString().replace(/\/+$/g, "");
18if (!enabled) {
19 console.warn(
20 "database sync disabled"
21 + (token ? " (CLOVER_DB_SYNC=0)" : " (no CLOVER_SOT_KEY)"),
22 );
23}
24
25let inFlight: Promise<void> | null = null;
26
27export function revalidate(): Promise<void> {
28 return inFlight ??= revalidateInner().finally(() => {
29 inFlight = null;
30 });
31}
32
33async function revalidateInner(): Promise<void> {
34 const { location } = mediaFiles.db;
35 const res = await fetch(`${sotUrl}/db`, {
36 headers: {
37 Authorization: token!,
38 ...etag() ? { "If-None-Match": etag()! } : null,
39 },
40 signal: AbortSignal.timeout(120_000),
41 });
42 if (res.status === 304) return;
43 if (!res.ok || !res.body) {
44 throw new Error(`source of truth responded ${res.status} ${res.statusText}`);
45 }
46
47 // another pull can be landing at once, from this process's last copy of
48 // this code or from another process
49 const tmp = `${location}.${crypto.randomUUID()}.tmp`;
50 try {
51 await stream.promises.pipeline(
52 stream.Readable.fromWeb(res.body as any),
53 fs.createWriteStream(tmp),
54 );
55 using snapshot = new sqlite.DatabaseSync(tmp, { readOnly: true });
56 await sqlite.backup(snapshot, location);
57 } finally {
58 await fsp.rm(tmp, { force: true });
59 }
60 saveEtag(res.headers.get("ETag"));
61 console.info(`database updated (${string.formatByteSize(Number(res.headers.get("Content-Length") ?? "0"))})`);
62}
63
64// -- the persisted etag, stored beside the database --
65let cachedEtag: string | null | undefined;
66function etagPath() {
67 return mediaFiles.db.location + ".etag";
68}
69function etag(): string | null {
70 if (cachedEtag === undefined) {
71 try {
72 cachedEtag = fs.readFileSync(etagPath(), "utf-8").trim() || null;
73 } catch {
74 cachedEtag = null;
75 }
76 }
77 return cachedEtag;
78}
79function saveEtag(value: string | null) {
80 cachedEtag = value;
81 try {
82 fs.writeFileSync(etagPath(), value ?? "");
83 } catch (err) {
84 console.warn("could not persist database etag:", err);
85 }
86}
87
88// -- the subscription --
89// a hand-rolled sse reader over fetch (EventSource cannot send the
90// Authorization header). any event block triggers a revalidate; a failed
91// revalidate aborts the connection so the reconnect path retries it.
92async function subscribeLoop(stop: AbortSignal) {
93 let backoff = 1000;
94 while (!stop.aborted) {
95 const ctrl = new AbortController();
96 let watchdog: NodeJS.Timeout | null = null;
97 try {
98 const res = await fetch(`${sotUrl}/db/events`, {
99 headers: { Authorization: token! },
100 signal: AbortSignal.any([stop, ctrl.signal]),
101 });
102 if (!res.ok || !res.body) {
103 throw new Error(`${res.status} ${res.statusText}`);
104 }
105 console.info(`subscribed to database changes from ${sotUrl}`);
106 backoff = 1000;
107
108 // the server keepalives every 25s; 90s of silence means the
109 // connection died without a FIN
110 const armWatchdog = () => {
111 if (watchdog) clearTimeout(watchdog);
112 watchdog = setTimeout(
113 () => ctrl.abort(new Error("no keepalive")),
114 90_000,
115 );
116 watchdog.unref();
117 };
118 armWatchdog();
119 const reader = res.body.getReader();
120 const decoder = new TextDecoder();
121 let buffer = "";
122 while (true) {
123 const { value, done } = await reader.read();
124 if (done) break;
125 armWatchdog();
126 buffer += decoder.decode(value, { stream: true });
127 const blocks = buffer.split("\n\n");
128 buffer = blocks.pop()!;
129 const changed = blocks.some((block) =>
130 block.split("\n").some((line) => line.startsWith("event:") || line.startsWith("data:"))
131 );
132 if (changed) {
133 void revalidate().catch((err) => {
134 console.error("revalidate failed:", err);
135 ctrl.abort(new Error("revalidate failed"));
136 });
137 }
138 }
139 throw new Error("stream ended");
140 } catch (err: any) {
141 if (stop.aborted) return;
142 console.warn(
143 `database subscription lost (${err?.message ?? err}); `
144 + `retrying in ${Math.round(backoff / 1000)}s`,
145 );
146 } finally {
147 if (watchdog) clearTimeout(watchdog);
148 ctrl.abort();
149 }
150 await new Promise<void>((resolve) => setTimeout(resolve, backoff).unref());
151 backoff = Math.min(backoff * 2, 60_000);
152 }
153}
154
155if (enabled) {
156 // reloading this code subscribes again, so the old subscription ends
157 const stop = new AbortController();
158 incremental.onDispose(import.meta, () => stop.abort());
159 void subscribeLoop(stop.signal);
160}
161
162import * as fs from "node:fs";
163import * as fsp from "node:fs/promises";
164import * as sqlite from "node:sqlite";
165import * as stream from "node:stream";
166
167import * as mediaFiles from "#src/file-viewer/models/media-files.ts";
168
169import * as incremental from "@clo/lib/incremental";
170import * as log from "@clo/lib/log";
171import * as string from "@clo/lib/string";