1/**
2 * `Git.RegistryD1` — the repo index on a D1 database.
3 *
4 * An interchangeable implementation of {@link RegistryStore}: same
5 * contract as {@link RegistryDurableObject}, different scaling shape.
6 * The Durable Object is one single-threaded object in one region for the
7 * whole service; D1 replicates reads, and `owner/name` resolution plus
8 * `WHERE owner = ?` listing are exactly what a relational store is built
9 * for. Writes (create, delete, summary refresh) are rare by comparison.
10 *
11 * Uniqueness moves with it. The DO gets a compare-and-swap from its own
12 * single-threaded `transactionSync`; here it comes from the schema —
13 * `PRIMARY KEY (owner, name)` plus an insert guarded by `WHERE NOT
14 * EXISTS`, which SQLite evaluates atomically within the statement. A
15 * losing racer sees `changes === 0` and reports `RepoAlreadyExists`,
16 * with no error-string sniffing.
17 *
18 * ### Choosing an index
19 * **Example:** D1 instead of the default Durable Object
20 * ```typescript
21 * const RepoIndex = Cloudflare.D1.Database("RepoIndex");
22 *
23 * const GitLive = Git.ApiLive.pipe(
24 * Layer.provide(Git.ApiHandlersLive),
25 * Layer.provide(Authentication.layer),
26 * Layer.provide(Git.ReposDurableObject),
27 * Layer.provide(Git.RegistryD1(RepoIndex)),
28 * Layer.provide(Git.BlobStoreR2(GitObjects)),
29 * );
30 * ```
31 *
32 * @layer
33 * @provides Git.RegistryStore
34 * @product Git
35 */
36import type * as runtime from "@cloudflare/workers-types";
37import * as Effect from "effect/Effect";
38import * as Layer from "effect/Layer";
39import * as Cloudflare from "../Cloudflare/index.ts";
40import type { RuntimeContext } from "../RuntimeContext.ts";
41import type { Database } from "../Cloudflare/D1/Database.ts";
42import type { QueryDatabaseClient } from "../Cloudflare/D1/QueryDatabase.ts";
43import { RepoAlreadyExists, ValidationError } from "./Api.ts";
44import { StoreError } from "./Protocol/Store.ts";
45import {
46 DEFAULT_LIST_LIMIT,
47 RegistryStore,
48 RESERVED_OWNERS,
49 decodeCursor,
50 encodeCursor,
51 toEntry,
52 ulid,
53 type CreateRepoInput,
54 type ListReposInput,
55 type ListReposResult,
56 type RegistryEntry,
57 type RegistryShape,
58 type RepoSummary,
59} from "./RegistryObject.ts";
60import { REGISTRY_DDL, type RegistryRepoRow } from "./Store/Sql.ts";
61
62/**
63 * D1's executors put failures in the defect channel (their error channel
64 * is `never`), so every call is wrapped to surface as a typed
65 * {@link StoreError} like the rest of the service.
66 */
67const d1 = <A>(
68 what: string,
69 effect: Effect.Effect<A, never, RuntimeContext>,
70): Effect.Effect<A, StoreError, RuntimeContext> =>
71 Effect.catchCause(effect, (cause) =>
72 Effect.fail(new StoreError({ reason: `registry ${what}: ${cause}` })),
73 );
74
75/**
76 * The repo index on D1. The database must already exist; the table is
77 * created idempotently on first use from the same DDL the Durable Object
78 * applies, so the two backends stay schema-identical.
79 */
80export const RegistryD1 = (database: Database): Layer.Layer<RegistryStore> =>
81 Layer.effect(
82 RegistryStore,
83 Effect.gen(function* () {
84 const db: QueryDatabaseClient =
85 yield* Cloudflare.D1.QueryDatabase(database);
86
87 // The DDL is shared with the DO (Store/Sql.ts) — one definition, so
88 // the backends can never drift apart.
89 const ready = Effect.suspend(() =>
90 Effect.forEach(REGISTRY_DDL, (statement) =>
91 d1("schema", db.prepare(statement).run()),
92 ),
93 ).pipe(Effect.asVoid, Effect.cached);
94 const schema = yield* ready;
95
96 const rows = <T>(result: runtime.D1Result<T>): ReadonlyArray<T> =>
97 (result.results ?? []) as ReadonlyArray<T>;
98
99 const shape: RegistryShape = {
100 createRepo: Effect.fn(function* (input: CreateRepoInput) {
101 const owner = input.owner.toLowerCase();
102 const name = input.name.toLowerCase();
103 if (RESERVED_OWNERS.has(owner)) {
104 return yield* new ValidationError({
105 message: `owner name '${owner}' is reserved`,
106 });
107 }
108 yield* schema;
109 const repoId = yield* ulid();
110 const createdAt = Date.now();
111
112 // Guarded insert: SQLite evaluates the NOT EXISTS and the write
113 // atomically, so this is the CAS the DO gets from its single
114 // thread. `changes === 0` means someone already holds the name
115 // (a soft-deleted row still does, until purge frees it).
116 const inserted = yield* d1(
117 "createRepo",
118 db
119 .prepare(
120 `INSERT INTO repos
121 (owner, name, repo_id, description, is_public, fork_of, fork_count, created_at, deleted_at)
122 SELECT ?, ?, ?, ?, ?, ?, 0, ?, NULL
123 WHERE NOT EXISTS (
124 SELECT 1 FROM repos WHERE owner = ? AND name = ?
125 )`,
126 )
127 .bind(
128 owner,
129 name,
130 repoId,
131 input.description ?? null,
132 input.public === true ? 1 : 0,
133 input.forkOf ?? null,
134 createdAt,
135 owner,
136 name,
137 )
138 .run(),
139 );
140 if ((inserted.meta?.changes ?? 0) === 0) {
141 return yield* new RepoAlreadyExists({ owner, repo: name });
142 }
143 if (input.forkOf !== undefined) {
144 // Deliberately not batched with the insert: the fork count is
145 // only ever READ derived from live `fork_of` rows (see
146 // `bumpForkCount` with delta 0), so a crash between the two
147 // cannot strand a purge.
148 yield* d1(
149 "recordFork",
150 db
151 .prepare(
152 `UPDATE repos SET fork_count = fork_count + 1 WHERE repo_id = ?`,
153 )
154 .bind(input.forkOf)
155 .run(),
156 );
157 }
158 return {
159 owner,
160 name,
161 repoId,
162 defaultBranch: "main",
163 readOnly: false,
164 public: input.public === true,
165 status: "ready",
166 description: input.description ?? null,
167 forkOf: input.forkOf ?? null,
168 forkCount: 0,
169 createdAt,
170 deletedAt: null,
171 } satisfies RegistryEntry;
172 }),
173
174 resolve: Effect.fn(function* (owner: string, name: string) {
175 yield* schema;
176 // Soft-deleted rows ARE returned (with `deletedAt` set) so the
177 // Worker can report `status: "deleting"` rather than a 404 that
178 // would race a re-create against the purge.
179 const row = yield* d1(
180 "resolve",
181 db
182 .prepare(`SELECT * FROM repos WHERE owner = ? AND name = ?`)
183 .bind(owner.toLowerCase(), name.toLowerCase())
184 .first<RegistryRepoRow>(),
185 );
186 return row === null ? undefined : toEntry(row);
187 }),
188
189 list: Effect.fn(function* (input: ListReposInput) {
190 yield* schema;
191 const limit = Math.max(
192 1,
193 Math.min(input.limit ?? DEFAULT_LIST_LIMIT, 100),
194 );
195 const after =
196 input.cursor === undefined ? undefined : decodeCursor(input.cursor);
197 const conditions: Array<string> = ["deleted_at IS NULL"];
198 const bindings: Array<string | number> = [];
199 if (input.publicOnly === true) {
200 conditions.push("is_public = 1");
201 }
202 if (input.owner !== undefined) {
203 conditions.push("owner = ?");
204 bindings.push(input.owner.toLowerCase());
205 }
206 if (after !== undefined) {
207 conditions.push("(owner > ? OR (owner = ? AND name > ?))");
208 bindings.push(after[0], after[0], after[1]);
209 }
210 const page = yield* d1(
211 "list",
212 db
213 .prepare(
214 `SELECT * FROM repos WHERE ${conditions.join(" AND ")}
215 ORDER BY owner, name LIMIT ?`,
216 )
217 .bind(...bindings, limit + 1)
218 .all<RegistryRepoRow>(),
219 );
220 const all = rows(page);
221 const hasMore = all.length > limit;
222 const items = hasMore ? all.slice(0, limit) : all;
223 const last = items[items.length - 1];
224 return {
225 items: items.map(toEntry),
226 nextCursor:
227 hasMore && last !== undefined
228 ? encodeCursor(last.owner, last.name)
229 : null,
230 hasMore,
231 } satisfies ListReposResult;
232 }),
233
234 recordFork: (parentRepoId) => shape.bumpForkCount(parentRepoId, 1),
235
236 bumpForkCount: Effect.fn(function* (repoId: string, delta: number) {
237 yield* schema;
238 if (delta !== 0) {
239 yield* d1(
240 "bumpForkCount",
241 db
242 .prepare(
243 `UPDATE repos SET fork_count = MAX(0, fork_count + ?) WHERE repo_id = ?`,
244 )
245 .bind(delta, repoId)
246 .run(),
247 );
248 const row = yield* d1(
249 "bumpForkCount read",
250 db
251 .prepare(`SELECT fork_count FROM repos WHERE repo_id = ?`)
252 .bind(repoId)
253 .first<{ fork_count: number }>(),
254 );
255 return row?.fork_count ?? 0;
256 }
257 // Delta 0 DERIVES the count from live fork rows rather than the
258 // stored column, so the purge fork-pin keeps working after the
259 // parent's own row is removed (matching the DO exactly).
260 const row = yield* d1(
261 "forkCount",
262 db
263 .prepare(`SELECT COUNT(*) AS n FROM repos WHERE fork_of = ?`)
264 .bind(repoId)
265 .first<{ n: number }>(),
266 );
267 return row?.n ?? 0;
268 }),
269
270 markDeleted: Effect.fn(function* (repoId: string) {
271 yield* schema;
272 yield* d1(
273 "markDeleted",
274 db
275 .prepare(
276 `UPDATE repos SET deleted_at = ? WHERE repo_id = ? AND deleted_at IS NULL`,
277 )
278 .bind(Date.now(), repoId)
279 .run(),
280 );
281 }),
282
283 updateSummary: Effect.fn(function* (
284 repoId: string,
285 summary: RepoSummary,
286 ) {
287 yield* schema;
288 yield* d1(
289 "updateSummary",
290 db
291 .prepare(
292 `UPDATE repos SET default_branch = ?, read_only = ?, is_public = ?, status = ?
293 WHERE repo_id = ?`,
294 )
295 .bind(
296 summary.defaultBranch,
297 summary.readOnly ? 1 : 0,
298 summary.public ? 1 : 0,
299 summary.status,
300 repoId,
301 )
302 .run(),
303 );
304 }),
305
306 removeRow: Effect.fn(function* (repoId: string) {
307 yield* schema;
308 const row = yield* d1(
309 "removeRow read",
310 db
311 .prepare(`SELECT * FROM repos WHERE repo_id = ?`)
312 .bind(repoId)
313 .first<RegistryRepoRow>(),
314 );
315 if (row === null) {
316 return; // idempotent — purge re-runs are expected
317 }
318 // One batch: the row goes and the fork parent is decremented
319 // together, or neither happens.
320 const statements = [
321 db.prepare(`DELETE FROM repos WHERE repo_id = ?`).bind(repoId),
322 ...(row.fork_of !== null
323 ? [
324 db
325 .prepare(
326 `UPDATE repos SET fork_count = MAX(0, fork_count - 1) WHERE repo_id = ?`,
327 )
328 .bind(row.fork_of),
329 ]
330 : []),
331 ];
332 yield* d1("removeRow", db.batch(statements));
333 }),
334 };
335
336 return shape;
337 }),
338 ).pipe(Layer.provide(Cloudflare.D1.QueryDatabaseBinding)) as never;