1/**
2 * Repo / Registry DO SQLite schema (DESIGN.md §3.1) and typed query helpers.
3 *
4 * Exposes:
5 *
6 * - the exact DDL as idempotent init effects ({@link initRepoSchema},
7 * {@link initRegistrySchema}) — run in the DO's outer init under
8 * `blockConcurrencyWhile` semantics,
9 * - typed row interfaces mirroring each table,
10 * - a small {@link SqlClient} over the raw synchronous `cf.SqlStorage`
11 * surface: `all`/`first`/`run`, the chunked-`IN` helper (≤ 100 bound
12 * params per statement), and the {@link SqlClient.transactionSync}
13 * wrapper every ref mutation goes through (DESIGN.md §3.4).
14 *
15 * All helpers operate on the raw synchronous SQLite surface (`cursor
16 * .toArray()` is synchronous inside a DO) wrapped in `Effect.try`, so SQL
17 * failures surface as typed {@link StoreError}s instead of defects.
18 */
19import type * as cf from "@cloudflare/workers-types";
20import type {
21 DurableObjectState,
22 SqlStorageValue,
23} from "../../Cloudflare/Workers/index.ts";
24import * as Effect from "effect/Effect";
25import { StoreError } from "../Protocol/Store.ts";
26
27// ─────────────────────────────────────────────────────────────────────────────
28// DDL (exact, DESIGN.md §3.1)
29// ─────────────────────────────────────────────────────────────────────────────
30
31/**
32 * Current Repo DO schema version, stored in the `config` table under
33 * {@link SCHEMA_VERSION_KEY} for future migrations.
34 */
35export const SCHEMA_VERSION = 2;
36
37/**
38 * The `config` key holding {@link SCHEMA_VERSION}.
39 */
40export const SCHEMA_VERSION_KEY = "schema_version";
41
42/**
43 * The Repo DO DDL, one statement per element, applied in order. Every
44 * statement is idempotent (`CREATE TABLE IF NOT EXISTS`), so re-running init
45 * after a crash or on every cold start is safe.
46 */
47export const REPO_DDL: ReadonlyArray<string> = [
48 // config: repoId, owner, name, default_branch, description, created_at,
49 // schema_version, status ('ready'|'importing'|'forking'|'deleting'),
50 // read_only ('0'|'1'), fork_of (parent repoId or absent)
51 `CREATE TABLE IF NOT EXISTS config (key TEXT PRIMARY KEY, value TEXT NOT NULL)`,
52 // all refs, full names. HEAD is virtual: symref to config.default_branch.
53 `CREATE TABLE IF NOT EXISTS refs (
54 name TEXT PRIMARY KEY,
55 oid TEXT NOT NULL
56) WITHOUT ROWID`,
57 // object index + loose bytes. zdata is zlib(content) WITHOUT the loose
58 // header, so pack emission is: varint(type,size) + zdata, verbatim.
59 `CREATE TABLE IF NOT EXISTS objects (
60 oid TEXT PRIMARY KEY,
61 type INTEGER NOT NULL,
62 size INTEGER NOT NULL,
63 zsize INTEGER NOT NULL,
64 location TEXT NOT NULL DEFAULT 'row',
65 zdata BLOB,
66 r2_key TEXT,
67 pack_id TEXT,
68 pack_offset INTEGER,
69 staged_push TEXT
70) WITHOUT ROWID`,
71 `CREATE INDEX IF NOT EXISTS idx_objects_staged ON objects(staged_push) WHERE staged_push IS NOT NULL`,
72 // commit graph for closure/negotiation walks (populated at ingest)
73 `CREATE TABLE IF NOT EXISTS commits (
74 oid TEXT PRIMARY KEY,
75 tree TEXT NOT NULL,
76 gen INTEGER NOT NULL,
77 commit_time INTEGER NOT NULL
78) WITHOUT ROWID`,
79 `CREATE TABLE IF NOT EXISTS commit_parents (
80 oid TEXT NOT NULL,
81 parent TEXT NOT NULL,
82 ord INTEGER NOT NULL,
83 PRIMARY KEY (oid, parent)
84) WITHOUT ROWID`,
85 // crash-safety bookkeeping for pushes (GC'd by alarm)
86 `CREATE TABLE IF NOT EXISTS pushes (
87 push_id TEXT PRIMARY KEY,
88 started_at INTEGER NOT NULL,
89 state TEXT NOT NULL
90) WITHOUT ROWID`,
91 // async job bookkeeping (import/fork/delete-purge progress; one row per kind)
92 `CREATE TABLE IF NOT EXISTS jobs (
93 kind TEXT PRIMARY KEY,
94 state TEXT NOT NULL,
95 detail TEXT,
96 updated_at INTEGER NOT NULL
97) WITHOUT ROWID`,
98 // pull requests: base/head are live branch refs — the rows store intent +
99 // lifecycle, never a diff snapshot. Rows are never deleted (close, don't
100 // delete — numbers stay dense and stable); `merge_commit` is set iff
101 // state='merged' (the head tip for a fast-forward, the two-parent merge
102 // commit otherwise).
103 `CREATE TABLE IF NOT EXISTS pulls (
104 number INTEGER PRIMARY KEY,
105 title TEXT NOT NULL,
106 body TEXT,
107 base_ref TEXT NOT NULL,
108 head_ref TEXT NOT NULL,
109 state TEXT NOT NULL DEFAULT 'open',
110 created_at INTEGER NOT NULL,
111 updated_at INTEGER NOT NULL,
112 merged_at INTEGER,
113 merge_commit TEXT
114) WITHOUT ROWID`,
115 `CREATE INDEX IF NOT EXISTS idx_pulls_state ON pulls(state, number)`,
116];
117
118/**
119 * Additive column migrations for Repo tables created by an earlier schema
120 * version (mirrors {@link REGISTRY_MIGRATIONS}). `CREATE TABLE IF NOT
121 * EXISTS` leaves an existing table untouched, so any future `pulls` (or
122 * other) column addition goes here — never into the `CREATE TABLE`.
123 * Re-running is safe because "duplicate column name" is swallowed.
124 */
125export const REPO_MIGRATIONS: ReadonlyArray<string> = [];
126
127/**
128 * The Registry DO DDL (DESIGN.md §3.1, bottom): the `owner/name → repoId`
129 * authority, listing surface, and fork-lineage ledger.
130 */
131export const REGISTRY_DDL: ReadonlyArray<string> = [
132 `CREATE TABLE IF NOT EXISTS repos (
133 owner TEXT NOT NULL,
134 name TEXT NOT NULL,
135 repo_id TEXT NOT NULL UNIQUE,
136 description TEXT,
137 fork_of TEXT,
138 fork_count INTEGER NOT NULL DEFAULT 0,
139 created_at INTEGER NOT NULL,
140 deleted_at INTEGER,
141 -- Denormalised display fields, pushed here by the Repo DO whenever they
142 -- change (DESIGN.md section 15, bottleneck 7). The DO remains the source
143 -- of truth; these exist so listing repos does not have to wake one
144 -- Durable Object per listed row just to render a page.
145 default_branch TEXT NOT NULL DEFAULT 'main',
146 read_only INTEGER NOT NULL DEFAULT 0,
147 is_public INTEGER NOT NULL DEFAULT 0,
148 status TEXT NOT NULL DEFAULT 'ready',
149 PRIMARY KEY (owner, name)
150) WITHOUT ROWID`,
151];
152
153// ─────────────────────────────────────────────────────────────────────────────
154// Typed row interfaces
155// ─────────────────────────────────────────────────────────────────────────────
156
157/** One `config` row. */
158export interface ConfigRow extends Record<string, SqlStorageValue> {
159 readonly key: string;
160 readonly value: string;
161}
162
163/** One `refs` row: a full ref name and its 40-hex tip oid. */
164export interface RefRow extends Record<string, SqlStorageValue> {
165 readonly name: string;
166 readonly oid: string;
167}
168
169/**
170 * One full `objects` row. `zdata` BLOBs come back from workerd SQLite as
171 * `ArrayBuffer`s; `location` is `'row' | 'r2' | 'pack'` (pack = v1.1);
172 * `staged_push` is `NULL` for live objects, else the staging push id.
173 */
174export interface ObjectRow extends Record<string, SqlStorageValue> {
175 readonly oid: string;
176 readonly type: number;
177 readonly size: number;
178 readonly zsize: number;
179 readonly location: string;
180 readonly zdata: ArrayBuffer | null;
181 readonly r2_key: string | null;
182 readonly pack_id: string | null;
183 readonly pack_offset: number | null;
184 readonly staged_push: string | null;
185}
186
187/** The metadata projection of an `objects` row (no byte payload columns). */
188export interface ObjectMetaRow extends Record<string, SqlStorageValue> {
189 readonly oid: string;
190 readonly type: number;
191 readonly size: number;
192 readonly zsize: number;
193 readonly location: string;
194}
195
196/** One `commits` row (commit-graph node). */
197export interface CommitRow extends Record<string, SqlStorageValue> {
198 readonly oid: string;
199 readonly tree: string;
200 readonly gen: number;
201 readonly commit_time: number;
202}
203
204/** One `commit_parents` row (commit-graph edge, `ord` = parent position). */
205export interface CommitParentRow extends Record<string, SqlStorageValue> {
206 readonly oid: string;
207 readonly parent: string;
208 readonly ord: number;
209}
210
211/** One `pushes` row (`state` is `'staging' | 'committed'`). */
212export interface PushRow extends Record<string, SqlStorageValue> {
213 readonly push_id: string;
214 readonly started_at: number;
215 readonly state: string;
216}
217
218/** One `pulls` row (`state` is `'open' | 'closed' | 'merged'`). */
219export interface PullRow extends Record<string, SqlStorageValue> {
220 readonly number: number;
221 readonly title: string;
222 readonly body: string | null;
223 readonly base_ref: string;
224 readonly head_ref: string;
225 readonly state: string;
226 readonly created_at: number;
227 readonly updated_at: number;
228 readonly merged_at: number | null;
229 readonly merge_commit: string | null;
230}
231
232/** One `jobs` row (`kind` is `'import' | 'fork' | 'purge' | 'gc'`). */
233export interface JobRow extends Record<string, SqlStorageValue> {
234 readonly kind: string;
235 readonly state: string;
236 readonly detail: string | null;
237 readonly updated_at: number;
238}
239
240/** One Registry `repos` row. */
241export interface RegistryRepoRow extends Record<string, SqlStorageValue> {
242 readonly owner: string;
243 readonly name: string;
244 readonly repo_id: string;
245 readonly description: string | null;
246 readonly fork_of: string | null;
247 readonly fork_count: number;
248 readonly created_at: number;
249 readonly deleted_at: number | null;
250 /** Denormalised from the Repo DO — see the DDL comment. */
251 readonly default_branch: string;
252 readonly read_only: number;
253 readonly is_public: number;
254 readonly status: string;
255}
256
257// ─────────────────────────────────────────────────────────────────────────────
258// Chunked-IN helpers
259// ─────────────────────────────────────────────────────────────────────────────
260
261/**
262 * The maximum number of bound parameters per statement (DESIGN.md: chunked
263 * `IN` lists ≤ 100 params).
264 */
265export const MAX_IN_PARAMS = 100;
266
267/**
268 * Splits `items` into chunks of at most `size` elements (default
269 * {@link MAX_IN_PARAMS}).
270 */
271export const chunk = <T>(
272 items: ReadonlyArray<T>,
273 size: number = MAX_IN_PARAMS,
274): Array<ReadonlyArray<T>> => {
275 const out: Array<ReadonlyArray<T>> = [];
276 for (let i = 0; i < items.length; i += size) {
277 out.push(items.slice(i, i + size));
278 }
279 return out;
280};
281
282/**
283 * Builds an `IN`-list placeholder string: `placeholders(3)` → `"?, ?, ?"`.
284 */
285export const placeholders = (count: number): string =>
286 Array.from({ length: count }, () => "?").join(", ");
287
288// ─────────────────────────────────────────────────────────────────────────────
289// SqlClient
290// ─────────────────────────────────────────────────────────────────────────────
291
292/**
293 * Options for {@link SqlClient.inChunks}: fixed bindings placed before /
294 * after each chunk's `IN`-list params, and an optional chunk-size override.
295 */
296export interface InChunksOptions {
297 /** Bindings placed before the chunk params in each statement. */
298 readonly prefix?: ReadonlyArray<SqlStorageValue> | undefined;
299 /** Bindings placed after the chunk params in each statement. */
300 readonly suffix?: ReadonlyArray<SqlStorageValue> | undefined;
301 /** Max chunk params per statement (capped so total ≤ {@link MAX_IN_PARAMS}). */
302 readonly chunkSize?: number | undefined;
303}
304
305/**
306 * A thin, typed, Effect-wrapped client over a Durable Object's synchronous
307 * SQLite surface. Construct with {@link makeSqlClient}. All methods fail with
308 * {@link StoreError} on SQL errors (never defects).
309 */
310export interface SqlClient {
311 /** The raw synchronous Cloudflare SQL surface (for advanced callers). */
312 readonly raw: cf.SqlStorage;
313 /** Runs a query and returns every row. */
314 readonly all: <Row extends Record<string, SqlStorageValue>>(
315 query: string,
316 ...bindings: Array<SqlStorageValue>
317 ) => Effect.Effect<Array<Row>, StoreError>;
318 /** Runs a query and returns the first row, or `undefined` when empty. */
319 readonly first: <Row extends Record<string, SqlStorageValue>>(
320 query: string,
321 ...bindings: Array<SqlStorageValue>
322 ) => Effect.Effect<Row | undefined, StoreError>;
323 /** Runs a statement for its side effects, consuming any cursor fully. */
324 readonly run: (
325 query: string,
326 ...bindings: Array<SqlStorageValue>
327 ) => Effect.Effect<void, StoreError>;
328 /**
329 * Runs `makeQuery(placeholders)` once per ≤ 100-param chunk of `items` and
330 * concatenates the rows. Row order across chunks follows input order of the
331 * chunks, but callers must not rely on intra-query ordering.
332 */
333 readonly inChunks: <Row extends Record<string, SqlStorageValue>>(
334 makeQuery: (placeholders: string) => string,
335 items: ReadonlyArray<SqlStorageValue>,
336 options?: InChunksOptions,
337 ) => Effect.Effect<Array<Row>, StoreError>;
338 /**
339 * Runs `body` inside `DurableObjectStorage.transactionSync` — atomic
340 * against mid-event failure, with the DO's input/output gates guaranteeing
341 * no interleaving between statements (DESIGN.md §3.4). The body MUST be
342 * fully synchronous (no awaits, no Effects).
343 *
344 * `rollback(error)` aborts the transaction (every statement is rolled
345 * back) and surfaces `error` on the typed error channel — this is the CAS
346 * abort path for `atomic` pushes. Any other thrown value also rolls back
347 * and surfaces as a {@link StoreError}.
348 */
349 readonly transactionSync: <A, E = never>(
350 body: (sql: cf.SqlStorage, rollback: (error: E) => never) => A,
351 ) => Effect.Effect<A, StoreError | E>;
352}
353
354const errorMessage = (cause: unknown): string =>
355 cause instanceof Error ? cause.message : String(cause);
356
357/** Internal control-flow signal for {@link SqlClient.transactionSync}. */
358class RollbackSignal {
359 constructor(readonly error: unknown) {}
360}
361
362/**
363 * Creates a {@link SqlClient} over a Durable Object's state handle. Plain
364 * factory (not a Layer) so `RepoObject.ts` / `RegistryObject.ts` can wire it
365 * inside their inner (per-instance) effects.
366 */
367export const makeSqlClient = (
368 state: DurableObjectState["Service"],
369): SqlClient => {
370 const sql = state.storage.sql.raw;
371 const storage = state.raw.storage;
372
373 const allWith = <Row extends Record<string, SqlStorageValue>>(
374 query: string,
375 bindings: ReadonlyArray<SqlStorageValue>,
376 ): Effect.Effect<Array<Row>, StoreError> =>
377 Effect.try({
378 try: () => sql.exec<Row>(query, ...bindings).toArray(),
379 catch: (cause) =>
380 new StoreError({
381 reason: `sql failed (${query.slice(0, 120)}): ${errorMessage(cause)}`,
382 }),
383 });
384
385 return {
386 raw: sql,
387 all: (query, ...bindings) => allWith(query, bindings),
388 first: <Row extends Record<string, SqlStorageValue>>(
389 query: string,
390 ...bindings: Array<SqlStorageValue>
391 ) =>
392 allWith<Row>(query, bindings).pipe(
393 Effect.map((rows) => (rows.length > 0 ? rows[0] : undefined)),
394 ),
395 run: (query, ...bindings) => allWith(query, bindings).pipe(Effect.asVoid),
396 inChunks: <Row extends Record<string, SqlStorageValue>>(
397 makeQuery: (ph: string) => string,
398 items: ReadonlyArray<SqlStorageValue>,
399 options?: InChunksOptions,
400 ): Effect.Effect<Array<Row>, StoreError> =>
401 Effect.gen(function* () {
402 const prefix = options?.prefix ?? [];
403 const suffix = options?.suffix ?? [];
404 const fixed = prefix.length + suffix.length;
405 const size = Math.max(
406 1,
407 Math.min(options?.chunkSize ?? MAX_IN_PARAMS, MAX_IN_PARAMS - fixed),
408 );
409 const rows: Array<Row> = [];
410 for (const part of chunk(items, size)) {
411 const batch = yield* allWith<Row>(
412 makeQuery(placeholders(part.length)),
413 [...prefix, ...part, ...suffix],
414 );
415 for (const row of batch) {
416 rows.push(row);
417 }
418 }
419 return rows;
420 }),
421 transactionSync: <A, E = never>(
422 body: (sql: cf.SqlStorage, rollback: (error: E) => never) => A,
423 ) =>
424 Effect.try({
425 try: () =>
426 storage.transactionSync(() =>
427 body(sql, (error: E): never => {
428 throw new RollbackSignal(error);
429 }),
430 ),
431 catch: (thrown): StoreError | E =>
432 thrown instanceof RollbackSignal
433 ? (thrown.error as E)
434 : new StoreError({
435 reason: `transactionSync failed: ${errorMessage(thrown)}`,
436 }),
437 }),
438 };
439};
440
441// ─────────────────────────────────────────────────────────────────────────────
442// Schema init
443// ─────────────────────────────────────────────────────────────────────────────
444
445/**
446 * Idempotently applies the Repo DO DDL ({@link REPO_DDL}) and seeds the
447 * `schema_version` config row. Run in the Repo DO's outer init effect under
448 * `blockConcurrencyWhile` semantics.
449 */
450export const initRepoSchema = Effect.fn(function* (sql: SqlClient) {
451 for (const statement of REPO_DDL) {
452 yield* sql.run(statement);
453 }
454 for (const statement of REPO_MIGRATIONS) {
455 // Already-applied migrations fail with "duplicate column name" — that
456 // is the success case on every run after the first.
457 yield* sql.run(statement).pipe(Effect.ignore);
458 }
459 yield* sql.run(
460 `INSERT INTO config (key, value) VALUES (?, ?) ON CONFLICT (key) DO NOTHING`,
461 SCHEMA_VERSION_KEY,
462 String(SCHEMA_VERSION),
463 );
464});
465
466/**
467 * Idempotently applies the Registry DO DDL ({@link REGISTRY_DDL}). Run in
468 * the Registry DO's outer init effect.
469 */
470/**
471 * Additive column migrations for Registry tables created by an earlier
472 * version. `CREATE TABLE IF NOT EXISTS` leaves an existing table untouched,
473 * so new columns must be added explicitly; re-running is safe because
474 * "duplicate column name" is swallowed.
475 */
476const REGISTRY_MIGRATIONS: ReadonlyArray<string> = [
477 `ALTER TABLE repos ADD COLUMN default_branch TEXT NOT NULL DEFAULT 'main'`,
478 `ALTER TABLE repos ADD COLUMN read_only INTEGER NOT NULL DEFAULT 0`,
479 `ALTER TABLE repos ADD COLUMN is_public INTEGER NOT NULL DEFAULT 0`,
480 `ALTER TABLE repos ADD COLUMN status TEXT NOT NULL DEFAULT 'ready'`,
481];
482
483export const initRegistrySchema = Effect.fn(function* (sql: SqlClient) {
484 for (const statement of REGISTRY_DDL) {
485 yield* sql.run(statement);
486 }
487 for (const statement of REGISTRY_MIGRATIONS) {
488 // Already-applied migrations fail with "duplicate column name" — that
489 // is the success case on every run after the first.
490 yield* sql.run(statement).pipe(Effect.ignore);
491 }
492});