1// @effect-diagnostics cryptoRandomUUID:off globalConsole:off - vendored upstream protocol, not an Effect service.
2import { HttpError, unavailable } from './errors.ts';
3import { GRACE_SECONDS, LEASE_SECONDS } from './limits.ts';
4import { measured, type Observations } from './observations.ts';
5
6export interface Scope {
7 scope_id: string;
8 endpoint: string;
9 repository_id: string;
10 repository_owner_id: string;
11 branch: string;
12 policy_version: number;
13 writes_enabled: number;
14 // 1 when `pull_request` runs whose PR targets `branch` may also publish.
15 pull_requests: number;
16}
17export interface Generation {
18 generation_id: string;
19 scope_id: string;
20 value_object: string;
21 blob_object: string;
22 blob_id: string | null;
23 value_size: number;
24 blob_size: number;
25}
26export interface Selection extends Generation {
27 kind: 'exact' | 'fallback';
28 key: number[];
29}
30
31// Preserve empty and non-UTF-8 keys as BLOBs, without text or hash conversions.
32export function binary(bytes: Uint8Array): ArrayBuffer {
33 return bytes.slice().buffer;
34}
35
36export async function getScope(db: D1Database, id: string, stats?: Observations): Promise<Scope> {
37 let scope: Scope | null;
38 try {
39 const { results } = await measured(
40 db
41 .prepare(`SELECT s.*, (s.writes_enabled AND d.writes_enabled) AS writes_enabled
42 FROM scopes s, deployment d WHERE s.scope_id = ? AND s.enabled = 1 AND d.enabled = 1`)
43 .bind(id)
44 .all<Scope>(),
45 stats,
46 );
47 scope = results[0] ?? null;
48 } catch {
49 unavailable();
50 }
51 if (!scope) throw new HttpError(404, 'unknown_scope');
52 return scope;
53}
54
55export async function selectEntry(
56 db: D1Database,
57 scope: string,
58 key: Uint8Array,
59 secondary: Uint8Array,
60 stats?: Observations,
61): Promise<Selection | null> {
62 // Both branches use indexed identities in one SQLite snapshot.
63 try {
64 const { results } = await measured(
65 db
66 .prepare(`
67 SELECT g.*, e.key, 'exact' AS kind, 0 AS priority
68 FROM entries e JOIN generations g ON g.generation_id = e.generation_id
69 JOIN scopes s ON s.scope_id = e.scope_id CROSS JOIN deployment d
70 WHERE e.scope_id = ?1 AND e.key = ?2 AND g.state = 'ready' AND g.expires_at > unixepoch()
71 AND s.enabled = 1 AND d.enabled = 1
72 UNION ALL
73 SELECT g.*, e.key, 'fallback' AS kind, 1 AS priority
74 FROM associations a JOIN entries e ON e.scope_id = a.scope_id AND e.key = a.target_key
75 JOIN generations g ON g.generation_id = e.generation_id
76 JOIN scopes s ON s.scope_id = e.scope_id CROSS JOIN deployment d
77 WHERE a.scope_id = ?1 AND a.secondary_key = ?3 AND g.state = 'ready' AND g.expires_at > unixepoch()
78 AND s.enabled = 1 AND d.enabled = 1
79 ORDER BY priority LIMIT 1`)
80 .bind(scope, binary(key), binary(secondary))
81 .all<Selection>(),
82 stats,
83 );
84 return results[0] ?? null;
85 } catch {
86 unavailable();
87 }
88}
89
90export async function selectBlob(
91 db: D1Database,
92 scope: string,
93 blob: string,
94 stats?: Observations,
95): Promise<Generation | null> {
96 try {
97 const { results } = await measured(
98 db
99 .prepare(`SELECT g.* FROM generations g JOIN scopes s ON s.scope_id = g.scope_id CROSS JOIN deployment d
100 WHERE g.scope_id = ? AND g.blob_id = ? AND s.enabled = 1 AND d.enabled = 1
101 AND ((g.state = 'ready' AND g.expires_at > unixepoch()) OR (g.state = 'retired' AND g.gc_after > unixepoch()))`)
102 .bind(scope, blob)
103 .all<Generation>(),
104 stats,
105 );
106 return results[0] ?? null;
107 } catch {
108 unavailable();
109 }
110}
111
112export async function reserve(
113 db: D1Database,
114 scope: Scope,
115 tokenExp: number,
116 bytes: number,
117 stats?: Observations,
118): Promise<Generation> {
119 const id = crypto.randomUUID();
120 const prefix = `${scope.scope_id}/${id}`;
121 try {
122 await measured(
123 db
124 .prepare(`INSERT INTO generations
125 (generation_id, scope_id, state, policy_version, token_exp, lease_until, gc_after, value_object, blob_object, charged_bytes)
126 VALUES (?, ?, 'uploading', ?, ?, unixepoch() + ?, unixepoch() + ?, ?, ?, ?)`)
127 .bind(
128 id,
129 scope.scope_id,
130 scope.policy_version,
131 tokenExp,
132 LEASE_SECONDS,
133 LEASE_SECONDS + GRACE_SECONDS,
134 `${prefix}/value`,
135 `${prefix}/blob`,
136 bytes,
137 )
138 .run(),
139 stats,
140 );
141 } catch {
142 unavailable();
143 }
144 return {
145 generation_id: id,
146 scope_id: scope.scope_id,
147 value_object: `${prefix}/value`,
148 blob_object: `${prefix}/blob`,
149 blob_id: null,
150 value_size: 0,
151 blob_size: 0,
152 };
153}
154
155export async function recordMultipart(
156 db: D1Database,
157 generation: Generation,
158 uploadId: string,
159 stats?: Observations,
160): Promise<void> {
161 const result = await measured(
162 db
163 .prepare(
164 `UPDATE generations SET upload_id = ? WHERE generation_id = ? AND state = 'uploading' AND lease_until > unixepoch()`,
165 )
166 .bind(uploadId, generation.generation_id)
167 .run(),
168 stats,
169 );
170 if (result.meta.changes !== 1) unavailable();
171}
172
173export async function publish(
174 db: D1Database,
175 generation: Generation,
176 key: Uint8Array,
177 secondary: Uint8Array,
178 stats?: Observations,
179): Promise<void> {
180 const actual = generation.value_size + generation.blob_size;
181 let results: D1Result[];
182 try {
183 results = await measured(
184 db.batch([
185 db
186 .prepare(`UPDATE generations SET state = 'ready', key = ?, secondary_key = ?, blob_id = ?,
187 value_size = ?, blob_size = ?, charged_bytes = ?,
188 expires_at = unixepoch() + (SELECT retention_seconds FROM scopes WHERE scope_id = generations.scope_id),
189 gc_after = unixepoch() + (SELECT retention_seconds FROM scopes WHERE scope_id = generations.scope_id)
190 WHERE generation_id = ? AND state = 'uploading' AND lease_until > unixepoch() AND token_exp > unixepoch()
191 AND charged_bytes >= ? AND EXISTS (
192 SELECT 1 FROM scopes s, deployment d WHERE s.scope_id = generations.scope_id
193 AND s.policy_version = generations.policy_version AND s.enabled = 1 AND s.writes_enabled = 1
194 AND d.enabled = 1 AND d.writes_enabled = 1 AND s.charged_bytes <= s.byte_limit AND d.charged_bytes <= d.byte_limit
195 ) RETURNING generation_id`)
196 .bind(
197 binary(key),
198 binary(secondary),
199 generation.blob_id,
200 generation.value_size,
201 generation.blob_size,
202 actual,
203 generation.generation_id,
204 actual,
205 ),
206 ]),
207 stats,
208 );
209 } catch {
210 unavailable();
211 }
212 if (results[0]?.results.length !== 1) throw new HttpError(503, 'publication_guard');
213}
214
215export async function abandon(db: D1Database, id: string): Promise<void> {
216 // Keep charges during the grace period for R2 operations that finish late.
217 await db
218 .prepare(`UPDATE generations SET lease_until = min(lease_until, unixepoch()), gc_after = min(gc_after, unixepoch() + ?)
219 WHERE generation_id = ? AND state = 'uploading'`)
220 .bind(GRACE_SECONDS, id)
221 .run();
222}
223
224export async function cleanup(env: Env): Promise<void> {
225 const limit = Number(env.GC_BATCH_SIZE);
226 if (!Number.isInteger(limit) || limit < 1 || limit > 256)
227 throw new Error('Invalid GC batch size');
228 const claim = crypto.randomUUID();
229 const claimed =
230 await env.INDEX.prepare(`UPDATE generations SET state = 'deleting', gc_claim = ?, gc_after = unixepoch() + 600
231 WHERE generation_id IN (SELECT generation_id FROM generations WHERE gc_after <= unixepoch() ORDER BY gc_after LIMIT ?)
232 RETURNING generation_id, value_object, blob_object, upload_id`)
233 .bind(claim, limit)
234 .all<{
235 generation_id: string;
236 value_object: string;
237 blob_object: string;
238 upload_id: string | null;
239 }>();
240 let deleted = 0;
241 for (const generation of claimed.results) {
242 try {
243 if (generation.upload_id)
244 await env.ARTIFACTS.resumeMultipartUpload(
245 generation.blob_object,
246 generation.upload_id,
247 ).abort();
248 await env.ARTIFACTS.delete([generation.value_object, generation.blob_object]);
249 // FK deletion only removes entries still selecting this generation.
250 await env.INDEX.prepare(
251 `DELETE FROM generations WHERE generation_id = ? AND state = 'deleting' AND gc_claim = ?`,
252 )
253 .bind(generation.generation_id, claim)
254 .run();
255 deleted++;
256 } catch {
257 console.warn(
258 JSON.stringify({
259 operation: 'cleanup',
260 error: 'deletion_failed',
261 generation: generation.generation_id,
262 }),
263 );
264 }
265 }
266 await cleanAssociations(env.INDEX, limit);
267 console.log(JSON.stringify({ operation: 'cleanup', claimed: claimed.results.length, deleted }));
268}
269
270async function cleanAssociations(db: D1Database, limit: number): Promise<void> {
271 // Cursor bounds scanning as well as deletion, even when all associations are live.
272 const batch = await db
273 .prepare(`SELECT scope_id, secondary_key FROM associations
274 WHERE (scope_id, secondary_key) > (SELECT scope_id, secondary_key FROM maintenance WHERE id = 1)
275 ORDER BY scope_id, secondary_key LIMIT ?`)
276 .bind(limit)
277 .all<{ scope_id: string; secondary_key: number[] }>();
278 const last = batch.results.at(-1);
279 await db.batch([
280 db
281 .prepare(`DELETE FROM associations WHERE (scope_id, secondary_key) IN (
282 SELECT scope_id, secondary_key FROM associations
283 WHERE (scope_id, secondary_key) > (SELECT scope_id, secondary_key FROM maintenance WHERE id = 1)
284 ORDER BY scope_id, secondary_key LIMIT ?)
285 AND NOT EXISTS (SELECT 1 FROM entries e WHERE e.scope_id = associations.scope_id AND e.key = associations.target_key)`)
286 .bind(limit),
287 db
288 .prepare('UPDATE maintenance SET scope_id = ?, secondary_key = ? WHERE id = 1')
289 .bind(last?.scope_id ?? '', binary(new Uint8Array(last?.secondary_key ?? []))),
290 ]);
291}