1import { LIVE_OBJECTS } from "../Store/ObjectStore.ts";
2/**
3 * The fork alarm job (DESIGN.md §2.3 Fork).
4 *
5 * A fork streams a row snapshot from the parent Repo DO over an RPC
6 * `Stream` and copies it in batches:
7 *
8 * - `config` — only the fork-relevant keys (`default_branch`,
9 * `description`); identity keys (repoId/owner/name/created_at/status)
10 * belong to the fork itself.
11 * - `refs`, `commits`, `commit_parents` — copied verbatim.
12 * - `objects` — `location='row'` rows carry their `zdata` BLOB;
13 * `location='r2'` rows are copied **by reference**: the `r2_key` column
14 * carries the parent's full key, which is immutable and shared
15 * (DESIGN.md §3.2). The parent's R2 prefix is retained while the
16 * Registry reports `fork_count > 0`.
17 *
18 * Every insert is `INSERT OR IGNORE` keyed on the natural primary key, so
19 * the job is idempotent and alarm-re-armable: a crashed run resumes by
20 * re-streaming from the start and skipping already-copied rows.
21 */
22import * as Effect from "effect/Effect";
23import { Base64 } from "effect/encoding";
24import * as Option from "effect/Option";
25import * as Result from "effect/Result";
26import * as Stream from "effect/Stream";
27import { StoreError } from "../Protocol/Store.ts";
28import type {
29 CommitParentRow,
30 CommitRow,
31 ObjectRow,
32 RefRow,
33 SqlClient,
34} from "../Store/Sql.ts";
35
36// ─────────────────────────────────────────────────────────────────────────────
37// Snapshot wire shapes
38// ─────────────────────────────────────────────────────────────────────────────
39
40/**
41 * A snapshotted `objects` row. `zdata` is `null` when `location != 'row'`.
42 *
43 * NOTE: `zdata` crosses the RPC boundary as **base64**, not raw bytes. The
44 * snapshot is a `Stream` of plain objects, which the RPC layer serializes
45 * as JSONL — a `Uint8Array` inside one of those objects would arrive as
46 * `{"0":31,"1":139,...}` and silently corrupt every forked object.
47 */
48export interface SnapshotObjectRow {
49 readonly oid: string;
50 readonly type: number;
51 readonly size: number;
52 readonly zsize: number;
53 readonly location: string;
54 /** base64 of `zlib(content)`, or `null` when the bytes live in R2. */
55 readonly zdata: string | null;
56 /** Full R2 key — may point into the PARENT repo's prefix (shared). */
57 readonly r2_key: string | null;
58 readonly pack_id: string | null;
59 readonly pack_offset: number | null;
60}
61
62/**
63 * One batch of snapshotted rows from a single table. The union member is
64 * discriminated by `table`.
65 */
66export type SnapshotChunk =
67 | {
68 readonly table: "config";
69 readonly rows: ReadonlyArray<{
70 readonly key: string;
71 readonly value: string;
72 }>;
73 }
74 | {
75 readonly table: "refs";
76 readonly rows: ReadonlyArray<{
77 readonly name: string;
78 readonly oid: string;
79 }>;
80 }
81 | {
82 readonly table: "objects";
83 readonly rows: ReadonlyArray<SnapshotObjectRow>;
84 }
85 | {
86 readonly table: "commits";
87 readonly rows: ReadonlyArray<{
88 readonly oid: string;
89 readonly tree: string;
90 readonly gen: number;
91 readonly commit_time: number;
92 }>;
93 }
94 | {
95 readonly table: "commit_parents";
96 readonly rows: ReadonlyArray<{
97 readonly oid: string;
98 readonly parent: string;
99 readonly ord: number;
100 }>;
101 };
102
103/** Decodes a snapshot row's base64 `zdata` back into bytes. */
104const decodeZData = (
105 row: SnapshotObjectRow,
106): Effect.Effect<Uint8Array, StoreError> => {
107 const decoded = Base64.decode(row.zdata ?? "");
108 return Result.isSuccess(decoded)
109 ? Effect.succeed(decoded.success)
110 : Effect.fail(
111 new StoreError({ reason: `fork: corrupt zdata for ${row.oid}` }),
112 );
113};
114
115/** Config keys that transfer from parent to fork. */
116const FORKED_CONFIG_KEYS = ["default_branch", "description"] as const;
117
118/** Rows per page for small-row tables. */
119const PAGE_SIZE = 400;
120/** Rows per page for `objects` (each `zdata` BLOB may be up to 1 MiB). */
121const OBJECTS_PAGE_SIZE = 8;
122
123// ─────────────────────────────────────────────────────────────────────────────
124// Parent side: produce the snapshot stream
125// ─────────────────────────────────────────────────────────────────────────────
126
127/**
128 * Streams a point-in-time row snapshot of one repo (the parent side of a
129 * fork). Backs the Repo DO's `snapshotRows` RPC method — DO RPC methods may
130 * return `Stream`s, and the caller (the fork's alarm job) sees chunks live.
131 *
132 * Staged (in-flight push) objects are excluded; keyed pagination keeps
133 * per-batch memory bounded. The stream is not transactional across pages —
134 * a concurrent push may add rows mid-snapshot; forks take whatever refs
135 * page they observe, which is the same semantics as `git clone` during a
136 * push.
137 */
138export const snapshotStream = (
139 sql: SqlClient,
140): Stream.Stream<SnapshotChunk, StoreError> => {
141 const config: Stream.Stream<SnapshotChunk, StoreError> = Stream.fromEffect(
142 Effect.gen(function* () {
143 const rows = yield* sql.all<{ key: string; value: string }>(
144 `SELECT key, value FROM config WHERE key IN (${FORKED_CONFIG_KEYS.map(() => "?").join(", ")})`,
145 ...FORKED_CONFIG_KEYS,
146 );
147 return { table: "config", rows } satisfies SnapshotChunk;
148 }),
149 );
150
151 const refs: Stream.Stream<SnapshotChunk, StoreError> = Stream.paginate(
152 "",
153 (after: string) =>
154 Effect.gen(function* () {
155 const rows = yield* sql.all<RefRow>(
156 `SELECT name, oid FROM refs WHERE name > ? ORDER BY name LIMIT ?`,
157 after,
158 PAGE_SIZE,
159 );
160 const chunk: SnapshotChunk = { table: "refs", rows };
161 const next =
162 rows.length < PAGE_SIZE
163 ? Option.none<string>()
164 : Option.some(rows[rows.length - 1]!.name);
165 return [rows.length > 0 ? [chunk] : [], next] as const;
166 }),
167 );
168
169 const objects: Stream.Stream<SnapshotChunk, StoreError> = Stream.paginate(
170 "",
171 (after: string) =>
172 Effect.gen(function* () {
173 const rows = yield* sql.all<ObjectRow>(
174 `SELECT oid, type, size, zsize, location, zdata, r2_key, pack_id, pack_offset, staged_push
175 FROM objects WHERE oid > ? AND ${LIVE_OBJECTS} ORDER BY oid LIMIT ?`,
176 after,
177 OBJECTS_PAGE_SIZE,
178 );
179 const chunk: SnapshotChunk = {
180 table: "objects",
181 rows: rows.map((row): SnapshotObjectRow => ({
182 oid: row.oid,
183 type: row.type,
184 size: row.size,
185 zsize: row.zsize,
186 location: row.location,
187 zdata:
188 row.zdata === null
189 ? null
190 : Base64.encode(new Uint8Array(row.zdata)),
191 r2_key: row.r2_key,
192 pack_id: row.pack_id,
193 pack_offset: row.pack_offset,
194 })),
195 };
196 const next =
197 rows.length < OBJECTS_PAGE_SIZE
198 ? Option.none<string>()
199 : Option.some(rows[rows.length - 1]!.oid);
200 return [rows.length > 0 ? [chunk] : [], next] as const;
201 }),
202 );
203
204 const commits: Stream.Stream<SnapshotChunk, StoreError> = Stream.paginate(
205 "",
206 (after: string) =>
207 Effect.gen(function* () {
208 const rows = yield* sql.all<CommitRow>(
209 `SELECT oid, tree, gen, commit_time FROM commits WHERE oid > ? ORDER BY oid LIMIT ?`,
210 after,
211 PAGE_SIZE,
212 );
213 const chunk: SnapshotChunk = { table: "commits", rows };
214 const next =
215 rows.length < PAGE_SIZE
216 ? Option.none<string>()
217 : Option.some(rows[rows.length - 1]!.oid);
218 return [rows.length > 0 ? [chunk] : [], next] as const;
219 }),
220 );
221
222 const commitParents: Stream.Stream<SnapshotChunk, StoreError> =
223 Stream.paginate(["", ""] as readonly [string, string], (after) =>
224 Effect.gen(function* () {
225 const rows = yield* sql.all<CommitParentRow>(
226 `SELECT oid, parent, ord FROM commit_parents
227 WHERE (oid > ? OR (oid = ? AND parent > ?)) ORDER BY oid, parent LIMIT ?`,
228 after[0],
229 after[0],
230 after[1],
231 PAGE_SIZE,
232 );
233 const chunk: SnapshotChunk = { table: "commit_parents", rows };
234 const last = rows[rows.length - 1];
235 const next =
236 rows.length < PAGE_SIZE || last === undefined
237 ? Option.none<readonly [string, string]>()
238 : Option.some([last.oid, last.parent] as const);
239 return [rows.length > 0 ? [chunk] : [], next] as const;
240 }),
241 );
242
243 // Objects before refs: a partially-copied fork must never advertise a ref
244 // whose closure is missing (the job is re-run from scratch anyway, but
245 // this ordering makes even a crashed fork's DB internally consistent).
246 return config.pipe(
247 Stream.concat(objects),
248 Stream.concat(commits),
249 Stream.concat(commitParents),
250 Stream.concat(refs),
251 );
252};
253
254// ─────────────────────────────────────────────────────────────────────────────
255// Fork side: apply the snapshot
256// ─────────────────────────────────────────────────────────────────────────────
257
258/** Zero-copy-when-possible Uint8Array → ArrayBuffer for BLOB binding. */
259const toArrayBuffer = (bytes: Uint8Array): ArrayBuffer =>
260 bytes.byteOffset === 0 &&
261 bytes.byteLength === bytes.buffer.byteLength &&
262 bytes.buffer instanceof ArrayBuffer
263 ? bytes.buffer
264 : (bytes.buffer.slice(
265 bytes.byteOffset,
266 bytes.byteOffset + bytes.byteLength,
267 ) as ArrayBuffer);
268
269/**
270 * Applies one snapshot chunk to the fork's own SQLite. All inserts are
271 * `INSERT OR IGNORE` on the natural primary key — re-running the whole
272 * snapshot after a crash converges without duplicating rows.
273 */
274export const applySnapshotChunk = Effect.fn(function* (
275 sql: SqlClient,
276 chunk: SnapshotChunk,
277) {
278 switch (chunk.table) {
279 case "config": {
280 for (const row of chunk.rows) {
281 yield* sql.run(
282 `INSERT INTO config (key, value) VALUES (?, ?)
283 ON CONFLICT (key) DO UPDATE SET value = excluded.value`,
284 row.key,
285 row.value,
286 );
287 }
288 return;
289 }
290 case "refs": {
291 for (const row of chunk.rows) {
292 yield* sql.run(
293 `INSERT OR IGNORE INTO refs (name, oid) VALUES (?, ?)`,
294 row.name,
295 row.oid,
296 );
297 }
298 return;
299 }
300 case "objects": {
301 for (const row of chunk.rows) {
302 yield* sql.run(
303 `INSERT OR IGNORE INTO objects (oid, type, size, zsize, location, zdata, r2_key, pack_id, pack_offset, staged_push)
304 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL)`,
305 row.oid,
306 row.type,
307 row.size,
308 row.zsize,
309 row.location,
310 row.zdata === null ? null : toArrayBuffer(yield* decodeZData(row)),
311 row.r2_key,
312 row.pack_id,
313 row.pack_offset,
314 );
315 }
316 return;
317 }
318 case "commits": {
319 for (const row of chunk.rows) {
320 yield* sql.run(
321 `INSERT OR IGNORE INTO commits (oid, tree, gen, commit_time) VALUES (?, ?, ?, ?)`,
322 row.oid,
323 row.tree,
324 row.gen,
325 row.commit_time,
326 );
327 }
328 return;
329 }
330 case "commit_parents": {
331 for (const row of chunk.rows) {
332 yield* sql.run(
333 `INSERT OR IGNORE INTO commit_parents (oid, parent, ord) VALUES (?, ?, ?)`,
334 row.oid,
335 row.parent,
336 row.ord,
337 );
338 }
339 return;
340 }
341 }
342});
343
344/** Dependencies of {@link runForkJob}. */
345export interface ForkJobOptions {
346 /** The fork's own SQL client. */
347 readonly sql: SqlClient;
348 /**
349 * The parent's snapshot stream — `parentStub.snapshotRows()` from the
350 * fork DO's alarm handler. Error type is widened to `unknown` because
351 * the stream crosses the RPC boundary.
352 */
353 readonly snapshot: Stream.Stream<SnapshotChunk, unknown>;
354}
355
356/**
357 * Drains the parent snapshot into the fork's SQLite. Returns the number of
358 * chunks applied. Idempotent; the caller flips `status` to `ready` (and
359 * clears the jobs row) on success, or re-arms the alarm on failure.
360 */
361export const runForkJob = (
362 options: ForkJobOptions,
363): Effect.Effect<number, StoreError> =>
364 options.snapshot.pipe(
365 Stream.mapError(
366 (error) =>
367 new StoreError({
368 reason: `fork snapshot stream failed: ${String(error)}`,
369 }),
370 ),
371 Stream.runFoldEffect(
372 () => 0,
373 (count, chunk) =>
374 applySnapshotChunk(options.sql, chunk).pipe(Effect.as(count + 1)),
375 ),
376 );