1// @effect-diagnostics nodeBuiltinImport:off - real SQLite exercises concurrent D1 claims and accounting.
2import * as NodeFS from "node:fs";
3import * as NodeSqlite from "node:sqlite";
4import * as Effect from "effect/Effect";
5import { expect, it } from "@effect/vitest";
6import { cleanup } from "../vendor/vite-plus-remote-cache-cloudflare/src/database.ts";
7import * as Cache from "./Cache.ts";
8import * as Cleanup from "./Cleanup.ts";
9
10it.effect("cleans three full disjoint batches within the D1 invocation budget", () =>
11 Effect.gen(function* () {
12 const sqlite = new NodeSqlite.DatabaseSync(":memory:");
13 try {
14 sqlite.exec(
15 NodeFS.readFileSync(
16 new URL(
17 "../vendor/vite-plus-remote-cache-cloudflare/migrations/0001_cache.sql",
18 import.meta.url,
19 ),
20 "utf8",
21 ),
22 );
23 sqlite.exec(`
24 INSERT INTO scopes (scope_id, endpoint, repository, repository_id, repository_owner_id, branch)
25 VALUES ('fixture', 'https://cache.example', 'owner/repo', '1', '2', 'refs/heads/main');
26 WITH RECURSIVE ids(id) AS (SELECT 0 UNION ALL SELECT id+1 FROM ids WHERE id<1023)
27 INSERT INTO generations (generation_id, scope_id, state, policy_version, token_exp,
28 lease_until, gc_after, value_object, blob_object, charged_bytes)
29 SELECT 'g'||id, 'fixture', 'retired', 1, unixepoch()+3600, 0, 0,
30 'value/'||id, 'blob/'||id, 1 FROM ids;
31 INSERT INTO generations (generation_id, scope_id, state, policy_version, token_exp,
32 lease_until, gc_after, value_object, blob_object, charged_bytes)
33 VALUES ('live', 'fixture', 'ready', 1, unixepoch()+3600, 0, unixepoch()+3600,
34 'live/value', 'live/blob', 1);
35 `);
36 let queries = 0;
37 const claims: string[][] = [];
38 const deleted = new Set<string>();
39 const query = () => {
40 if (++queries > 1000) throw new Error("D1 query budget exceeded");
41 };
42 const prepare = (sql: string) => {
43 let args: NodeSqlite.SQLInputValue[] = [];
44 const statement = {
45 bind: (...values: NodeSqlite.SQLInputValue[]) => {
46 args = values.map((value) =>
47 value instanceof ArrayBuffer ? new Uint8Array(value) : value,
48 );
49 return statement;
50 },
51 all: async () => {
52 query();
53 const results = sqlite.prepare(sql).all(...args);
54 if (sql.startsWith("UPDATE generations"))
55 claims.push(results.map((row) => String(row.generation_id)));
56 return { results };
57 },
58 run: async () => {
59 query();
60 if (sql.startsWith("DELETE FROM generations")) {
61 const row = sqlite
62 .prepare(
63 "SELECT value_object, blob_object FROM generations WHERE generation_id = ?",
64 )
65 .get(args[0]!);
66 expect(deleted.has(String(row!.value_object))).toBe(true);
67 expect(deleted.has(String(row!.blob_object))).toBe(true);
68 }
69 return sqlite.prepare(sql).run(...args);
70 },
71 };
72 return statement;
73 };
74 const env = {
75 GC_BATCH_SIZE: String(Cleanup.batchSize),
76 INDEX: {
77 prepare,
78 batch: async (statements: ReturnType<typeof prepare>[]) =>
79 Promise.all(statements.map((statement) => statement.run())),
80 },
81 ARTIFACTS: {
82 delete: async (keys: string[]) => {
83 for (const key of keys) {
84 expect(deleted.has(key)).toBe(false);
85 deleted.add(key);
86 }
87 },
88 },
89 } as unknown as Parameters<typeof cleanup>[0];
90 yield* Cleanup.run().pipe(
91 Effect.provideService(Cache.Cache, {
92 handle: () => Effect.die("unused"),
93 cleanup: Effect.promise(() => cleanup(env)),
94 }),
95 );
96 expect(claims.map((claim) => claim.length)).toEqual([256, 256, 256]);
97 expect(new Set(claims.flat()).size).toBe(768);
98 expect(queries).toBeLessThanOrEqual(1000);
99 expect(deleted.size).toBe(1536);
100 expect(deleted.has("live/value")).toBe(false);
101 expect(sqlite.prepare("SELECT charged_bytes FROM scopes").get()).toEqual({
102 charged_bytes: 257,
103 });
104 expect(sqlite.prepare("SELECT count(*) AS n FROM generations").get()).toEqual({ n: 257 });
105 } finally {
106 sqlite.close();
107 }
108 }),
109);