1// Uncompressed files are read directly from the media store root. Derivied
2// assets like compressed files, optimized images, and streamable video are
3// stored in the `derived` folder. After scanning, the derived assets are
4// uploaded into the store (storage1/clofi-derived dataset on NAS). Since
5// multiple files can share the same hash, the number of references is
6// tracked, and the derived content is only produced once. This means if a
7// file is deleted, it should only decrement a reference count; deleting it
8// once all references are removed.
9
10const db = getDatabase("cache.sqlite");
11
12const roots = sqliteTable("derived_roots", {
13 id: integer().primaryKey({ autoIncrement: true }),
14 /** milliseconds */
15 date: integer().notNull(),
16 /** the hash of the files it derives from, and the name of what it holds */
17 key: text().notNull().unique(),
18}, (t) => [index("derived_roots_key").on(t.key)]);
19const files = sqliteTable("derived_files", {
20 id: integer().primaryKey({ autoIncrement: true }),
21 /** a path in the root's directory */
22 file: text().notNull(),
23 root: integer().notNull().references(() => roots.id, { onDelete: "cascade" }),
24 /** bytes */
25 size: integer().notNull(),
26 /** sha1 */
27 hash: text().notNull(),
28}, (t) => [index("derived_files_root").on(t.root)]);
29const refs = sqliteTable("derived_refs", {
30 id: integer().primaryKey({ autoIncrement: true }),
31 file: integer().notNull().references(() => mediaFiles.table.id, { onDelete: "cascade" }),
32 root: integer().notNull().references(() => roots.id, { onDelete: "cascade" }),
33}, (t) => [
34 unique().on(t.file, t.root),
35 index("derived_refs_file").on(t.file),
36 index("derived_refs_root").on(t.root),
37]);
38db.migrate(roots, 1);
39db.migrate(files, 1);
40db.migrate(refs, 1);
41
42// derived assets are written directly into the derived file store. files are
43// produced into a `tmp.`-prefixed sibling directory and renamed into place,
44// so a crash never leaves a partially-written asset at its final path.
45export const workDir = Path.resolve(derivedFileRoot);
46
47let ongoing = new Map<string, Promise<number>>();
48
49/** produce a derived */
50export async function produce(
51 { mediaFile: file, node, subkey, cores, producer }: {
52 mediaFile: MediaFile;
53 node: progress.Node;
54 subkey: string;
55 cores: number;
56 producer: (root: Path) => Promise<void>;
57 },
58) {
59 const key = `${file.hash}/${subkey}`;
60 const current = ongoing.get(key);
61 let root: number | null = null;
62 if (current) {
63 root = await current;
64 } else {brk: {
65 root = db.select({ id: roots.id }).from(roots).where(eq(roots.key, key)).get()?.id ?? null;
66 if (root) break brk;
67 const { promise, resolve, reject } = Promise.withResolvers<number>();
68 ongoing.set(key, promise);
69 // the rejection is rethrown to this caller; concurrent callers await
70 // through `ongoing`, but when there are none the rejection would
71 // otherwise be unobserved
72 promise.catch(() => {});
73 const finalDir = workDir.join(key);
74 const tmp = workDir.join(`${file.hash}/tmp.${subkey}`);
75 try {
76 node.hidden = true;
77 const filesWithStats = await queue.run({
78 cores,
79 async run() {
80 node.hidden = false;
81 await tmp.delete({ recursive: true, force: true });
82 await tmp.makeOrEmptyDir();
83 await producer(tmp);
84 const sha1queue = new queue.PriorityQueue(cores);
85 return await Promise.all(
86 (await tmp.readDir({ recursive: true }))
87 .filter((path) => !path.base.startsWith("tmp."))
88 .map(async (path) =>
89 sha1queue.run({
90 async run() {
91 using _ = node.start("hash " + path.relative(tmp));
92 const stats = await path.stat();
93 if (stats.isDirectory()) {
94 return [path, stats.size, null] as const;
95 }
96 const hash = await new Promise<string>(
97 (resolve, reject) => {
98 const reader = fs.createReadStream(path.toString());
99 reader.on("error", reject);
100
101 const hasher = crypto.createHash("sha1").setEncoding(
102 "hex",
103 );
104 hasher.on("error", reject);
105 hasher.on("readable", () => resolve(hasher.read()));
106
107 reader.pipe(hasher);
108 },
109 );
110 return [path, stats.size, hash] as const;
111 },
112 })
113 ),
114 );
115 },
116 });
117 // move into the final location before recording rows; readers only
118 // discover the asset through the database, so this is safe.
119 await finalDir.delete({ recursive: true, force: true });
120 await fsp.rename(tmp.toString(), finalDir.toString());
121
122 root = db.transaction((tx) => {
123 const { id } = UNWRAP(tx.insert(roots).values({ key, date: Date.now() }).returning({ id: roots.id }).get());
124 for (const [path, size, hash] of filesWithStats) {
125 if (hash === null) continue;
126 tx.insert(files).values({ root: id, file: path.relative(tmp), size, hash }).run();
127 }
128 return id;
129 });
130 resolve(root);
131 } catch (e) {
132 if (root) deleteRoot({ id: root });
133 // remove partial output from disk: the tmp dir if the producer died,
134 // or the renamed final dir if the database writes failed (e.g. the
135 // source file was deleted while this asset was being produced)
136 await tmp.delete({ recursive: true, force: true }).catch(() => {});
137 await finalDir.delete({ recursive: true, force: true }).catch(() => {});
138 reject(e);
139 throw e;
140 } finally {
141 ongoing.delete(key);
142 }
143 }}
144 ASSERT(root != null);
145 const ref = { root, file: file.id };
146 db.insert(refs).values(ref).onConflictDoNothing({ target: [refs.file, refs.root] }).run();
147 return {
148 [Symbol.dispose]: () => {
149 db.delete(refs).where(and(eq(refs.root, ref.root), eq(refs.file, ref.file))).run();
150 },
151 };
152}
153
154/** a derived asset of a file, by its path in the root that holds it */
155export function get(file: MediaFile, subPath: string) {
156 const row = db.select({ size: files.size, hash: files.hash, date: roots.date, key: roots.key })
157 .from(refs)
158 .innerJoin(files, eq(files.root, refs.root))
159 .innerJoin(roots, eq(roots.id, files.root))
160 .where(and(eq(refs.file, file.id), eq(files.file, subPath)))
161 .get();
162 if (!row) return null;
163 return { path: row.key + "/" + subPath, size: row.size, hash: row.hash, date: new Date(row.date) };
164}
165
166/** roots no file refers to anymore */
167export function findOrphanedRoots() {
168 return db.select().from(roots).leftJoin(refs, eq(refs.root, roots.id)).where(isNull(refs.id)).all()
169 .map((row) => row.derived_roots);
170}
171
172export function deleteRoot(root: { id: number }) {
173 db.delete(roots).where(eq(roots.id, root.id)).run();
174}
175
176/** remove an orphaned root's files from the derived store */
177export async function deleteRootFiles(root: { key: string }) {
178 const dir = workDir.join(root.key);
179 await dir.delete({ recursive: true, force: true });
180 // reclaim the per-hash parent directory once its last root is gone
181 // (rmdir refuses non-empty directories, which is exactly what we want)
182 await fsp.rmdir(UNWRAP(dir.parent).toString()).catch(() => {});
183}
184
185/**
186 * delete `tmp.*` directories left behind by crashed producers. only removes
187 * directories untouched for a day, to never race an ongoing producer.
188 *
189 * `tmp.*` FILES are never touched: encode intermediates inside completed
190 * roots (`av1-au/tmp.av1.mp4`, ...) are intentionally named that way to be
191 * excluded from `derived_files`, but the dash producer reads them across
192 * root directories, so they must persist.
193 */
194export async function cleanAbandonedTmp(): Promise<number> {
195 let cleaned = 0;
196 const dayAgo = Date.now() - 24 * 60 * 60 * 1000;
197 for (const hashDir of await workDir.readDir().catch(() => [])) {
198 for (const sub of await hashDir.readDir().catch(() => [])) {
199 if (!sub.base.startsWith("tmp.")) continue;
200 try {
201 const stat = await sub.stat();
202 if (!stat.isDirectory()) continue;
203 if (stat.mtime.getTime() > dayAgo) continue;
204 await sub.delete({ recursive: true, force: true });
205 cleaned += 1;
206 } catch {}
207 }
208 }
209 return cleaned;
210}
211
212import { getDatabase } from "#sitegen/drizzle";
213import { Path } from "#sitegen/path";
214import { ASSERT, UNWRAP } from "@clo/lib/assert";
215import * as progress from "@clo/lib/progress";
216import * as queue from "@clo/lib/queue";
217import { and, eq, isNull } from "drizzle-orm";
218import { index, integer, sqliteTable, text, unique } from "drizzle-orm/sqlite-core";
219import * as crypto from "node:crypto";
220import * as fs from "node:fs";
221import * as fsp from "node:fs/promises";
222import { derivedFileRoot } from "../paths.ts";
223import * as mediaFiles from "./media-files.ts";
224import type { MediaFile } from "./media-files.ts";