1// Which processors have run on which files. `processors` has a row for each
2// processor the code declares, with its version, and removing one from the
3// code removes its rows. `file_processors` has a row for each run that
4// finished: a file is pending for a processor without one, or with an older
5// version. failures keep their error, and the next sweep retries them.
6const db = getDatabase("cache.sqlite");
7
8export const statuses = { done: 1, failed: 2 } as const;
9export type Status = keyof typeof statuses;
10
11const processors = sqliteTable("processors", {
12 id: integer().primaryKey({ autoIncrement: true }),
13 name: text().notNull().unique(),
14 version: integer().notNull(),
15});
16const fileProcessors = sqliteTable("file_processors", {
17 file: integer().notNull().references(() => mediaFiles.table.id, { onDelete: "cascade" }),
18 processor: integer().notNull().references(() => processors.id, { onDelete: "cascade" }),
19 version: integer().notNull(),
20 status: sqliteNumericEnum(statuses)().notNull(),
21 /** milliseconds */
22 updated: integer().notNull(),
23 error: text(),
24}, (t) => [
25 primaryKey({ columns: [t.file, t.processor] }),
26 index("file_processors_processor").on(t.processor),
27]);
28db.migrate(processors, 1);
29db.migrate(fileProcessors, 1);
30
31/** the ids of the processors the code declares, by name, after removing the ones it no longer does */
32export function syncRegistry(defs: readonly { name: string; version: number }[]): Map<string, number> {
33 return db.transaction((tx) => {
34 const ids = new Map<string, number>();
35 for (const { name, version } of defs) {
36 const row = tx.insert(processors).values({ name, version })
37 .onConflictDoUpdate({ target: processors.name, set: { version } })
38 .returning({ id: processors.id }).get();
39 ids.set(name, UNWRAP(row).id);
40 }
41 tx.delete(processors).where(notInArray(processors.name, defs.map((def) => def.name))).run();
42 return ids;
43 });
44}
45
46/** what a file's processors finished, by processor id */
47export function getStates(file: number): Map<number, { version: number; status: Status }> {
48 const rows = db.select().from(fileProcessors).where(eq(fileProcessors.file, file)).all();
49 return new Map(rows.map((row) => [row.processor, { version: row.version, status: row.status }]));
50}
51
52export function recordResult(
53 file: number,
54 processor: number,
55 version: number,
56 status: Status,
57 error: string | null = null,
58) {
59 const row = { file, processor, version, status, updated: Date.now(), error };
60 db.insert(fileProcessors).values(row)
61 .onConflictDoUpdate({ target: [fileProcessors.file, fileProcessors.processor], set: row })
62 .run();
63}
64
65/** a file's contents changed, so every processor runs on it again */
66export function invalidateFile(file: number) {
67 db.delete(fileProcessors).where(eq(fileProcessors.file, file)).run();
68}
69
70import { getDatabase, sqliteNumericEnum } from "#sitegen/drizzle";
71import { UNWRAP } from "@clo/lib/assert";
72import { eq, notInArray } from "drizzle-orm";
73import { index, integer, primaryKey, sqliteTable, text } from "drizzle-orm/sqlite-core";
74import * as mediaFiles from "./media-files.ts";