1/**
2 * In-process harness for the repo object store: a `SqlClient` over
3 * `bun:sqlite` (the same DDL the Durable Object runs) and an in-memory
4 * `BlobStoreShape` with ranged reads. Lets the emitter, compaction, and
5 * closure code be unit-tested without a Worker.
6 */
7import { Database } from "bun:sqlite";
8import * as Effect from "effect/Effect";
9import * as Stream from "effect/Stream";
10import type { BlobBody, BlobMeta, BlobStoreShape } from "@/Git/BlobStore.ts";
11import { BlobStoreError } from "@/Git/BlobStore.ts";
12import {
13 chunk,
14 MAX_IN_PARAMS,
15 placeholders,
16 REPO_DDL,
17 type SqlClient,
18} from "@/Git/Store/Sql.ts";
19import { StoreError } from "@/Git/Protocol/Store.ts";
20
21type Value = string | number | ArrayBuffer | Uint8Array | null;
22
23const bind = (v: Value): string | number | Uint8Array | null =>
24 v instanceof ArrayBuffer ? new Uint8Array(v) : v;
25
26/** Rows come back with BLOB columns as ArrayBuffer, like the DO. */
27const normalize = (row: Record<string, unknown>) => {
28 for (const key of Object.keys(row)) {
29 const v = row[key];
30 if (v instanceof Uint8Array) {
31 row[key] = v.buffer.slice(v.byteOffset, v.byteOffset + v.byteLength);
32 }
33 }
34 return row;
35};
36
37class RollbackSignal {
38 constructor(readonly error: unknown) {}
39}
40
41export const makeTestSqlClient = (): SqlClient & { readonly db: Database } => {
42 const db = new Database(":memory:");
43 for (const ddl of REPO_DDL) db.run(ddl);
44
45 const exec = (query: string, bindings: ReadonlyArray<Value>) => {
46 // Durable Object SQLite: at most 100 bound parameters per statement.
47 if (bindings.length > MAX_IN_PARAMS) {
48 throw new Error(
49 `too many SQL variables (${bindings.length} > ${MAX_IN_PARAMS}): SQLITE_ERROR`,
50 );
51 }
52 const stmt = db.query(query);
53 const isRead = /^\s*(SELECT|WITH|PRAGMA)/i.test(query);
54 let rowsWritten = 0;
55 let rows: Array<Record<string, unknown>> = [];
56 if (isRead || /RETURNING/i.test(query)) {
57 rows = stmt.all(...bindings.map(bind)) as Array<Record<string, unknown>>;
58 } else {
59 // `run` reports `changes`, which is what the DO cursor's rowsWritten is.
60 rowsWritten = (stmt.run(...bindings.map(bind)) as { changes: number })
61 .changes;
62 }
63 const out = rows.map(normalize);
64 return {
65 toArray: () => out,
66 one: () => {
67 if (out.length !== 1)
68 throw new Error(`expected 1 row, got ${out.length}`);
69 return out[0];
70 },
71 raw: () => out.map((r) => Object.values(r)),
72 rowsWritten,
73 rowsRead: out.length,
74 columnNames: out.length > 0 ? Object.keys(out[0]!) : [],
75 [Symbol.iterator]: () => out[Symbol.iterator](),
76 };
77 };
78 const raw = {
79 exec: (query: string, ...bindings: Array<Value>) => exec(query, bindings),
80 } as unknown as SqlClient["raw"];
81
82 const allWith = <Row extends Record<string, unknown>>(
83 query: string,
84 bindings: ReadonlyArray<Value>,
85 ) =>
86 Effect.try({
87 try: () => exec(query, bindings).toArray() as Array<Row>,
88 catch: (cause) =>
89 new StoreError({
90 reason: `sql failed (${query.slice(0, 120)}): ${cause instanceof Error ? cause.message : String(cause)}`,
91 }),
92 });
93
94 return {
95 db,
96 raw,
97 all: (query, ...bindings) => allWith(query, bindings as Array<Value>),
98 first: (query, ...bindings) =>
99 allWith(query, bindings as Array<Value>).pipe(
100 Effect.map((rows) => (rows.length > 0 ? rows[0] : undefined)),
101 ) as never,
102 run: (query, ...bindings) =>
103 allWith(query, bindings as Array<Value>).pipe(Effect.asVoid),
104 inChunks: (makeQuery, items, options) =>
105 Effect.gen(function* () {
106 const prefix = options?.prefix ?? [];
107 const suffix = options?.suffix ?? [];
108 const size = Math.max(
109 1,
110 Math.min(
111 options?.chunkSize ?? MAX_IN_PARAMS,
112 MAX_IN_PARAMS - prefix.length - suffix.length,
113 ),
114 );
115 const rows: Array<never> = [];
116 for (const part of chunk(items, size)) {
117 const batch = yield* allWith(makeQuery(placeholders(part.length)), [
118 ...prefix,
119 ...part,
120 ...suffix,
121 ] as Array<Value>);
122 for (const row of batch) rows.push(row as never);
123 }
124 return rows;
125 }) as never,
126 transactionSync: (body) =>
127 Effect.try({
128 try: () => {
129 db.run("BEGIN");
130 try {
131 const result = body(raw, (error) => {
132 throw new RollbackSignal(error);
133 });
134 db.run("COMMIT");
135 return result;
136 } catch (error) {
137 db.run("ROLLBACK");
138 throw error;
139 }
140 },
141 catch: (cause) =>
142 cause instanceof RollbackSignal
143 ? (cause.error as never)
144 : new StoreError({
145 reason: `transactionSync failed: ${cause instanceof Error ? cause.message : String(cause)}`,
146 }),
147 }) as never,
148 };
149};
150
151export interface MemoryBlobStore extends BlobStoreShape {
152 readonly objects: Map<string, Uint8Array>;
153 /** Every `get`, with its range — for asserting window fetch counts. */
154 readonly gets: Array<{ key: string; offset?: number; length?: number }>;
155}
156
157export const makeMemoryBlobStore = (): MemoryBlobStore => {
158 const objects = new Map<string, Uint8Array>();
159 const uploads = new Map<string, Map<number, Uint8Array>>();
160 const putPart = (uploadId: string, partNumber: number, part: Uint8Array) =>
161 Effect.suspend(() => {
162 const stored = uploads.get(uploadId);
163 if (stored === undefined) {
164 return Effect.fail(
165 new BlobStoreError({ reason: `no upload ${uploadId}` }),
166 );
167 }
168 stored.set(partNumber, Uint8Array.from(part));
169 return Effect.succeed({ partNumber, etag: `etag-${partNumber}` });
170 });
171
172 const gets: MemoryBlobStore["gets"] = [];
173 const body = (bytes: Uint8Array): BlobBody => ({
174 size: bytes.length,
175 bytes: Effect.succeed(bytes),
176 stream: Stream.succeed(bytes),
177 });
178 return {
179 objects,
180 gets,
181 get: (key, range) =>
182 Effect.sync(() => {
183 gets.push({ key, offset: range?.offset, length: range?.length });
184 const whole = objects.get(key);
185 if (whole === undefined) return null;
186 if (range === undefined) return body(whole);
187 return body(
188 whole.subarray(
189 range.offset,
190 Math.min(range.offset + range.length, whole.length),
191 ),
192 );
193 }),
194 put: (key, data) =>
195 Effect.gen(function* () {
196 if (data instanceof Uint8Array) {
197 objects.set(key, Uint8Array.from(data));
198 return;
199 }
200 const parts = Array.from(
201 yield* Stream.runCollect(
202 data as Stream.Stream<Uint8Array, BlobStoreError>,
203 ),
204 );
205 const total = parts.reduce((n, p) => n + p.length, 0);
206 const out = new Uint8Array(total);
207 let at = 0;
208 for (const p of parts) {
209 out.set(p, at);
210 at += p.length;
211 }
212 objects.set(key, out);
213 }) as never,
214 head: (key) =>
215 Effect.sync((): BlobMeta | null => {
216 const o = objects.get(key);
217 return o === undefined ? null : { key, size: o.length };
218 }),
219 multipart: (key) =>
220 Effect.sync(() => {
221 const uploadId = `${key}#${uploads.size + 1}`;
222 uploads.set(uploadId, new Map());
223 return {
224 uploadId,
225 uploadPart: (partNumber, part) => putPart(uploadId, partNumber, part),
226 complete: (parts) =>
227 Effect.suspend(() => {
228 const stored = uploads.get(uploadId);
229 if (stored === undefined) {
230 return Effect.fail(
231 new BlobStoreError({ reason: `no upload ${uploadId}` }),
232 );
233 }
234 // R2's rule, checked as R2 checks it: every part but the
235 // last (in the list as given) must be the same size.
236 const listed = parts.map((p) => stored.get(p.partNumber));
237 if (listed.some((p) => p === undefined)) {
238 return Effect.fail(
239 new BlobStoreError({ reason: "complete: unknown part" }),
240 );
241 }
242 const sizes = listed.map((p) => p!.length);
243 for (let i = 0; i < sizes.length - 1; i++) {
244 if (sizes[i] !== sizes[0]) {
245 return Effect.fail(
246 new BlobStoreError({
247 reason:
248 "Your proposed upload is smaller than the minimum allowed object size",
249 }),
250 );
251 }
252 }
253 const total = sizes.reduce((n, x) => n + x, 0);
254 const out = new Uint8Array(total);
255 let at = 0;
256 for (const p of listed) {
257 out.set(p!, at);
258 at += p!.length;
259 }
260 objects.set(key, out);
261 uploads.delete(uploadId);
262 return Effect.void;
263 }),
264 abort: Effect.sync(() => {
265 uploads.delete(uploadId);
266 }),
267 };
268 }),
269 uploadPart: (_key, uploadId, partNumber, part) =>
270 putPart(uploadId, partNumber, part),
271 delete: (keys) =>
272 Effect.sync(() => {
273 for (const key of typeof keys === "string" ? [keys] : keys) {
274 objects.delete(key);
275 }
276 }),
277 list: (prefix) =>
278 Stream.fromIterable(
279 Array.from(objects.entries())
280 .filter(([k]) => k.startsWith(prefix ?? ""))
281 .map(([key, o]): BlobMeta => ({ key, size: o.length })),
282 ),
283 } as MemoryBlobStore;
284};