1/**
2 * The "source of truth" server is the canonical storage for
3 * paper clover's files. This is technically needed because
4 * the VPS she uses can only store about 20gb of content, where
5 * the contents of /file is about 48gb as of writing; only a
6 * limited amount of data can be cached.
7 *
8 * What's great about this system is it also allows scaling the
9 * website up into multiple servers, if that is ever desired.
10 * When that happens, mutations to "q+a" will be moved here,
11 * and the SQLite synchronization mechanism will apply to both
12 * of those databases.
13 *
14 * An alternative to this would have been to use a SMB client,
15 * but I've read that the systems used for caching don't work
16 * like the HTTP Cache-Control header, where you can say a file
17 * is valid for up to a certain amount of time. If we seriously
18 * need cache busts (paper clover does not), the proper way
19 * would be to push a message to all VPS nodes instead of
20 * checking upstream if a file changed every time.
21 * @module
22 */
23const app = new Hono();
24export default app;
25export const port = Number(process.env.PORT ?? 4000);
26
27const token = UNWRAP(process.env.CLOVER_SOT_KEY);
28
29if (!fs.existsSync(rawFileRoot)) {
30 throw new Error(`${rawFileRoot} does not exist`);
31}
32if (!fs.existsSync(derivedFileRoot)) {
33 throw new Error(`${derivedFileRoot} does not exist`);
34}
35
36/**
37 * `node run source-of-truth` boots the full source of truth: the indexer
38 * service (file watcher, periodic sweeps, processors) plus this http server.
39 * set CLOVER_INDEXER=0 for a serve-only process.
40 */
41export async function main() {
42 if (process.env.CLOVER_INDEXER !== "0") {
43 service = indexer.start();
44 }
45 const server = await http.serve({
46 respond: async (request) => app.fetch(request),
47 port,
48 });
49 console.info(`source of truth live at ${server.url}`);
50 const close = () => void server.close().then(() => process.exit(0));
51 process.on("SIGINT", close);
52 process.on("SIGTERM", close);
53}
54let service: indexer.Service | null = null;
55
56// Re-use file descriptors if the same file is being read twice.
57const fds = new Map<string, Awaitable<{ fd: number; refs: number }>>();
58
59app.get("/", (c) => {
60 return c.text("everything you love, loves you back.");
61});
62
63app.get("/file/*", async (c) => {
64 const fullQuery = c.req.path.slice("/file".length);
65 const [filePath, derivedAsset, ...invalid] = fullQuery.split(":/");
66 ASSERT(filePath != null);
67 if (invalid.length > 0) return c.notFound();
68 if (filePath.length <= 1) return c.notFound();
69 const permission = permissions.getByPrefix(filePath);
70 if (permission !== 0 && c.req.header("Authorization") !== token) {
71 return c.json({ error: "invalid authorization header" }, 401);
72 }
73 const file = mediaFiles.getByPath(filePath);
74 if (!file || file.kind === "directory") return c.notFound();
75 // derived asset names are plain relative paths under the hash directory
76 if (
77 derivedAsset
78 && derivedAsset.split("/").some((part) => part === "" || part === "." || part === "..")
79 ) {
80 return c.notFound();
81 }
82 const fullPath = derivedAsset
83 ? path.join(derivedFileRoot, file.hash, derivedAsset)
84 : path.join(rawFileRoot, file.path);
85 let handle: { fd: number; refs: number } | null = null;
86 try {
87 handle = await fds.get(fullPath) ?? null;
88 if (!handle) {
89 const promise = openFile(fullPath, "r")
90 .then((fd) => ({ fd, refs: 0 }))
91 .catch((err) => {
92 fds.delete(fullPath);
93 throw err;
94 });
95 fds.set(fullPath, promise);
96 fds.set(fullPath, handle = await promise);
97 }
98 handle.refs += 1;
99 } catch (err: any) {
100 if (err.code === "ENOENT") return c.notFound();
101 throw err;
102 }
103 const { size } = await fstat(handle.fd);
104 let start = 0;
105 let end = Infinity;
106 const ranges = http.parseRangeHeader(c.req.header("Range") ?? "", size);
107 if (ranges?.[0] && ranges.length === 1) {
108 ({ start, end } = ranges[0]);
109 c.status(206);
110 c.header("Content-Range", `bytes ${start}-${end}/${size}`);
111 }
112 c.header("Content-Length", String(Math.min(end, size - 1) - start + 1));
113 const nodeStream = fs.createReadStream(fullPath, {
114 fd: handle.fd,
115 // Positioned, or requests sharing this descriptor would share its offset.
116 start,
117 end,
118 fs: {
119 close: util.callbackify(async () => {
120 ASSERT(handle);
121 if ((handle.refs -= 1) <= 0) {
122 fds.delete(fullPath);
123 await closeFile(handle.fd);
124 }
125 handle = null;
126 }),
127 read: fsCallbacks.read,
128 },
129 });
130 c.header("Content-Type", mime.for(fullPath));
131 return c.body(stream.Readable.toWeb(nodeStream) as ReadableStream);
132});
133
134// tell web nodes to pull, after the database changed some other way than
135// through the indexer
136app.post("/reload", async (c) => {
137 if (c.req.header("Authorization") !== token) {
138 return c.json({ error: "invalid authorization header" }, 401);
139 }
140 indexer.emitDbChange();
141 return c.body(null, 204);
142});
143
144// a tiny server-sent-events feed of database changes. consumers (web nodes,
145// dev servers) hold this open and pull /db whenever a beacon arrives; the
146// connection itself is the subscription, so the source of truth never needs
147// a list of its consumers. events carry the generation, but clients can
148// treat any event as "something changed" — /db pulls are ETag-guarded.
149app.get("/db/events", (c) => {
150 if (c.req.header("Authorization") !== token) {
151 return c.json({ error: "invalid authorization header" }, 401);
152 }
153 const encoder = new TextEncoder();
154 let cleanup: (() => void) | null = null;
155 const stream = new ReadableStream<Uint8Array>({
156 start(controller) {
157 const send = (text: string) => controller.enqueue(encoder.encode(text));
158 // an event on connect makes clients revalidate immediately, covering
159 // anything they missed while disconnected
160 send(`retry: 5000\n\n`);
161 send(`event: change\ndata: ${indexer.generation}\n\n`);
162 const unsubscribe = indexer.subscribeDbChange((generation) => {
163 send(`event: change\ndata: ${generation}\n\n`);
164 });
165 // keepalive comments so proxies and clients hold the connection open
166 const keepalive = setInterval(() => send(`: keepalive\n\n`), 25_000);
167 keepalive.unref();
168 cleanup = () => {
169 unsubscribe[Symbol.dispose]();
170 clearInterval(keepalive);
171 };
172 },
173 cancel() {
174 cleanup?.();
175 },
176 });
177 c.header("Content-Type", "text/event-stream");
178 c.header("Cache-Control", "no-store");
179 return c.body(stream);
180});
181
182// live progress of all indexing operations, as a clover progress stream.
183// each request snapshots the current tree state and then follows along;
184// connect any time with `node run tail-progress`.
185app.get("/progress", (c) => {
186 if (c.req.header("Authorization") !== token) {
187 return c.json({ error: "invalid authorization header" }, 401);
188 }
189 c.header("Content-Type", progress.contentType);
190 return c.body(progress.encodeByteStream(indexer.root) as ReadableStream);
191});
192
193app.get("/studio/stats", (c) => {
194 if (c.req.header("Authorization") !== token) return c.json({ error: "invalid authorization header" }, 401);
195 return c.json(studio.stats(service));
196});
197
198app.get("/studio/activity", (c) => {
199 if (c.req.header("Authorization") !== token) return c.json({ error: "invalid authorization header" }, 401);
200 return c.json(studio.activity(indexer.root));
201});
202
203// download a consistent snapshot of cache.sqlite. web nodes call this with
204// `If-None-Match` on a stale-while-revalidate schedule, and immediately
205// when pinged at /internal/db-changed.
206app.get("/db", async (c) => {
207 if (c.req.header("Authorization") !== token) {
208 return c.json({ error: "invalid authorization header" }, 401);
209 }
210 const snap = await getDbSnapshot();
211 if (c.req.header("If-None-Match") === snap.etag) {
212 return c.body(null, 304);
213 }
214 c.header("Content-Type", "application/vnd.sqlite3");
215 c.header("Content-Length", String(snap.size));
216 c.header("ETag", snap.etag);
217 return c.body(
218 stream.Readable.toWeb(fs.createReadStream(snap.path)) as ReadableStream,
219 );
220});
221
222// trigger a sweep (or a subtree scan with {"path": "/2025"}). progress is
223// visible on /progress; this returns immediately.
224app.post("/scan", async (c) => {
225 if (c.req.header("Authorization") !== token) {
226 return c.json({ error: "invalid authorization header" }, 401);
227 }
228 if (!service) return c.json({ error: "indexer is disabled" }, 409);
229 const body = await c.req.json().catch(() => ({}));
230 const subPath = typeof body.path === "string" ? body.path : undefined;
231 void service.sweep("manual", subPath);
232 return c.json({ ok: true }, 202);
233});
234
235// -- database snapshotting --
236// `VACUUM INTO` writes a compact, transactionally-consistent copy that does
237// not depend on -wal sidecar files. regenerated lazily when the indexer
238// reports changes, at most every 30 seconds.
239interface DbSnapshot {
240 gen: number;
241 time: number;
242 path: string;
243 etag: string;
244 size: number;
245}
246let snapshot: DbSnapshot | null = null;
247let snapshotInFlight: Promise<DbSnapshot> | null = null;
248
249function getDbSnapshot() {
250 const gen = indexer.generation;
251 // a stale-generation snapshot is never served: subscribers pull exactly
252 // once per beacon, so a 304 here would leave them out of date until the
253 // ttl backstop. beacons are debounced upstream, which bounds how often
254 // the vacuum can actually run.
255 if (snapshot && snapshot.gen === gen && fs.existsSync(snapshot.path)) {
256 return Promise.resolve(snapshot);
257 }
258 return snapshotInFlight ??= (async () => {
259 try {
260 const dir = process.env.CLOVER_DB ?? ".clover";
261 const file = path.join(dir, "db-snapshot.sqlite");
262 await fs.rm(file, { force: true });
263 mediaFiles.db.run(sql`vacuum into ${file}`);
264 const [hash, { size }] = await Promise.all([
265 hashFile(new Path(file)),
266 fs.stat(file),
267 ]);
268 return snapshot = {
269 gen,
270 time: Date.now(),
271 path: file,
272 etag: `"${hash}"`,
273 size,
274 };
275 } finally {
276 snapshotInFlight = null;
277 }
278 })();
279}
280
281const openFile = util.promisify(fsCallbacks.open);
282const closeFile = util.promisify(fsCallbacks.close);
283const fstat = util.promisify(fsCallbacks.fstat);
284
285import * as fs from "#sitegen/fs";
286import { Path } from "#sitegen/path";
287import { hashFile } from "#src/file-viewer/indexer/scan.ts";
288import * as indexer from "#src/file-viewer/indexer/service.ts";
289import * as studio from "#src/file-viewer/indexer/studio.ts";
290import * as mediaFiles from "#src/file-viewer/models/media-files.ts";
291import * as permissions from "#src/file-viewer/models/permissions.ts";
292import { ASSERT, UNWRAP } from "@clo/lib/assert";
293import * as http from "@clo/lib/http";
294import * as mime from "@clo/lib/mime";
295import * as progress from "@clo/lib/progress";
296import { sql } from "drizzle-orm";
297import { Hono } from "hono";
298import * as fsCallbacks from "node:fs";
299import * as path from "node:path";
300import * as stream from "node:stream";
301import * as util from "node:util";
302import { derivedFileRoot, rawFileRoot } from "./file-viewer/paths.ts";