1import * as Cloudflare from "@/Cloudflare";
2import * as Drizzle from "@/Drizzle/Cloudflare.ts";
3import type { RuntimeContext } from "@/RuntimeContext.ts";
4import * as Cause from "effect/Cause";
5import { sql } from "drizzle-orm";
6import * as Context from "effect/Context";
7import * as Deferred from "effect/Deferred";
8import * as Effect from "effect/Effect";
9import * as Exit from "effect/Exit";
10import * as Fiber from "effect/Fiber";
11import * as Scheduler from "effect/Scheduler";
12// The exact artifacts `drizzle-kit generate` emits for
13// `driver: "durable-sqlite"` — a `migrations.js` that imports each
14// migration's `.sql` file as a text module. Bare `.sql` imports resolve
15// via the bundler's default text module types (see Bundle.ts), the same
16// way Wrangler's `Text` rules handle them.
17import migrations from "./drizzle/migrations.js";
18import { posts, relations, users } from "./schema.ts";
19
20class TransactionMarker extends Context.Service<TransactionMarker, string>()(
21 "DrizzleTransactionMarker",
22) {}
23
24export class DrizzleClockObject extends Cloudflare.DurableObject<DrizzleClockObject>()(
25 "DrizzleClockObject",
26 Effect.succeed(Effect.succeed({ wait: () => Effect.sleep("100 millis") })),
27) {}
28
29export class DrizzleUsersObject extends Cloudflare.DurableObject<DrizzleUsersObject>()(
30 "DrizzleUsersObject",
31 Effect.gen(function* () {
32 const clocks = yield* DrizzleClockObject;
33 return Effect.gen(function* () {
34 const state = yield* Cloudflare.DurableObjectState;
35 // Opens drizzle over this instance's SQLite storage — with the
36 // relational schema — and applies the generated migrations before
37 // any request touches the db.
38 const db = yield* Drizzle.DurableObject({ migrations, relations });
39
40 return {
41 clockName: () => Effect.sync(() => state.id.toString()),
42 sqliteClock: () =>
43 clocks.getByName(state.id.toString()).wait().pipe(Effect.as("ready")),
44 sqliteGate: (view: boolean) =>
45 Effect.gen(function* () {
46 const context = yield* Effect.context<RuntimeContext>();
47 const clock = clocks.getByName(state.id.toString());
48 // Native timers must be registered in the real enclosing input gate.
49 return yield* Effect.promise(() =>
50 state.raw.blockConcurrencyWhile(() => {
51 let outerRan = false;
52 let finished = false;
53 let finishedBeforeClear: boolean | undefined;
54 let release: Promise<void> | undefined;
55 const outer = setTimeout(() => {
56 outerRan = true;
57 }, 0);
58 return Effect.runPromise(
59 Effect.gen(function* () {
60 const original = db.$client;
61 const client = view
62 ? original.withoutTransforms()
63 : original;
64 const callerScheduler = yield* Scheduler.Scheduler;
65 const enteredAfterOuter = yield* client.withTransaction(
66 Effect.gen(function* () {
67 const enteredAfterOuter = outerRan;
68 // The independent DO completion can clear a blocked parent timer.
69 yield* Effect.sync(() => {
70 release = Effect.runPromise(
71 clock.wait().pipe(Effect.provideContext(context)),
72 ).then(() => {
73 finishedBeforeClear = finished;
74 clearTimeout(outer);
75 });
76 });
77 yield* client`SELECT 1`;
78 yield* Effect.yieldNow;
79 return enteredAfterOuter;
80 }),
81 );
82 const restoredScheduler = yield* Scheduler.Scheduler;
83 return {
84 enteredAfterOuter,
85 sameTransaction:
86 client.withTransaction === original.withTransaction,
87 samePermit: client.reserve === original.reserve,
88 sameTransactionContext:
89 client.transactionService ===
90 original.transactionService,
91 restoredScheduler: restoredScheduler === callerScheduler,
92 };
93 }).pipe(Effect.provideContext(context)),
94 )
95 .then((value) => {
96 finished = true;
97 return release?.then(() => ({
98 ...value,
99 finishedBeforeClear,
100 }));
101 })
102 .finally(() => clearTimeout(outer));
103 }),
104 );
105 }),
106 sqliteRollback: () =>
107 Effect.gen(function* () {
108 const client = db.$client;
109 const view = client.withoutTransforms();
110 const finalized: string[] = [];
111 const callerScope = yield* Effect.scope;
112 const scopedBody = (value: string) =>
113 Effect.gen(function* () {
114 const marker = yield* TransactionMarker;
115 const scope = yield* Effect.scope;
116 yield* Effect.addFinalizer(() =>
117 Effect.sync(() => {
118 finalized.push(
119 `${marker}:${value}:${scope !== callerScope}`,
120 );
121 }),
122 );
123 yield* db.insert(users).values({ name: value });
124 });
125 const failed = yield* client
126 .withTransaction(
127 scopedBody("failed").pipe(
128 Effect.andThen(Effect.fail("rollback")),
129 Effect.scoped,
130 ),
131 )
132 .pipe(Effect.exit);
133 const defect = yield* client
134 .withTransaction(
135 scopedBody("defect").pipe(
136 Effect.andThen(Effect.die("rollback defect")),
137 Effect.scoped,
138 ),
139 )
140 .pipe(Effect.exit);
141 const started = yield* Deferred.make<void>();
142 const release = yield* Deferred.make<void>();
143 const writer = yield* Effect.forkChild(
144 client.withTransaction(
145 scopedBody("interrupted").pipe(
146 Effect.andThen(Deferred.succeed(started, undefined)),
147 Effect.andThen(Deferred.await(release)),
148 Effect.scoped,
149 ),
150 ),
151 );
152 yield* Deferred.await(started);
153 const reader = yield* Effect.forkChild(
154 view`SELECT name FROM users`,
155 { startImmediately: true },
156 );
157 const waitingForPermit = yield* Effect.sync(
158 () => reader.pollUnsafe() === undefined,
159 );
160 yield* Fiber.interrupt(writer);
161 const interrupted = yield* Fiber.await(writer);
162 const rowsAfterRollback = yield* Fiber.join(reader);
163 yield* view.withTransaction(
164 scopedBody("committed").pipe(Effect.scoped),
165 );
166 const finalRows = yield* view`SELECT name FROM users`;
167 return {
168 failed:
169 Exit.isFailure(failed) &&
170 failed.cause.reasons.some(
171 (reason) =>
172 Cause.isFailReason(reason) && reason.error === "rollback",
173 ),
174 defect:
175 Exit.isFailure(defect) &&
176 defect.cause.reasons.some(
177 (reason) =>
178 Cause.isDieReason(reason) &&
179 reason.defect === "rollback defect",
180 ),
181 interrupted:
182 Exit.isFailure(interrupted) &&
183 Cause.hasInterruptsOnly(interrupted.cause),
184 waitingForPermit,
185 rowsAfterRollback,
186 finalRows,
187 finalized,
188 callerScopePreserved:
189 (yield* Effect.scope) === callerScope &&
190 callerScope.state._tag !== "Closed",
191 };
192 }).pipe(Effect.provideService(TransactionMarker, "caller")),
193 addUser: (name: string) =>
194 db
195 .insert(users)
196 .values({ name })
197 .returning()
198 .pipe(Effect.map((rows) => rows[0]!.id)),
199 addPost: (userId: number, title: string) =>
200 db.insert(posts).values({ userId, title }).pipe(Effect.asVoid),
201 listUsers: () =>
202 db
203 .select()
204 .from(users)
205 .pipe(Effect.map((rows) => rows.map((row) => row.name))),
206 // Relational query through the `relations` config — proves the
207 // schema/relationships flow through Drizzle.DurableObject's types.
208 listUsersWithPosts: () =>
209 db.query.users.findMany({ with: { posts: true } }).pipe(
210 Effect.map((rows) =>
211 rows.map((row) => ({
212 name: row.name,
213 posts: row.posts.map((post) => post.title),
214 })),
215 ),
216 ),
217 // A deliberately failing query, recovered with a typed catch —
218 // proves per-operation errors are tagged and catchable.
219 queryMissingTable: () =>
220 Effect.gen(function* () {
221 return yield* db.run(sql`SELECT * FROM missing_table`).pipe(
222 Effect.as("unexpected success"),
223 Effect.catchTag("EffectDrizzleQueryError", (error) =>
224 Effect.succeed(`caught:${error._tag}`),
225 ),
226 );
227 }),
228 };
229 });
230 }),
231) {}