1import type * as cf from "@cloudflare/workers-types";
2import * as Context from "effect/Context";
3import * as Data from "effect/Data";
4import * as Effect from "effect/Effect";
5import * as Exit from "effect/Exit";
6import * as Fiber from "effect/Fiber";
7import * as Scheduler from "effect/Scheduler";
8import * as Stream from "effect/Stream";
9import type { RuntimeContext } from "../../RuntimeContext.ts";
10import { flushDurableObjectAlarm } from "./DurableObjectAlarmStorage.ts";
11import {
12 ActiveStorageTransactions,
13 type ActiveStorageTransaction,
14} from "./DurableObjectTransactionContext.ts";
15
16/** A native transaction failure or invalid use of a transaction's owner. */
17export class DurableObjectStorageError extends Data.TaggedError(
18 "DurableObjectStorageError",
19)<{
20 readonly operation: string;
21 readonly message: string;
22 readonly cause?: unknown;
23}> {}
24
25// Pre-existing sibling fibers do not inherit the transaction's context.
26const activeStorageTransactions = new WeakMap<
27 cf.DurableObjectStorage,
28 ActiveStorageTransaction
29>();
30
31const checkTransactionOwner = (
32 transaction: ActiveStorageTransaction,
33 allowRolledBack = false,
34) =>
35 Effect.withFiber((fiber) =>
36 transaction.active &&
37 transaction.owner === fiber &&
38 (allowRolledBack || !transaction.rolledBack)
39 ? Effect.void
40 : Effect.fail(
41 new DurableObjectStorageError({
42 operation: "transaction",
43 message: !transaction.active
44 ? "The storage transaction callback has already finished"
45 : transaction.owner !== fiber
46 ? "A storage transaction cannot be used by another fiber"
47 : "The storage transaction has been rolled back",
48 }),
49 ),
50 );
51
52/** @internal Joins an owned transaction, or starts a native transaction. */
53function withStorageTransaction<A, E, R>(
54 storage: cf.DurableObjectStorage,
55 effect: Effect.Effect<A, E, R>,
56): Effect.Effect<A, E | DurableObjectStorageError, R>;
57function withStorageTransaction<A, E, R>(
58 storage: cf.DurableObjectStorage,
59 closure: (txn: DurableObjectTransaction) => Effect.Effect<A, E, R>,
60): Effect.Effect<A, E | DurableObjectStorageError, R>;
61function withStorageTransaction<A, E, R>(
62 storage: cf.DurableObjectStorage,
63 body:
64 | Effect.Effect<A, E, R>
65 | ((txn: DurableObjectTransaction) => Effect.Effect<A, E, R>),
66): Effect.Effect<A, E | DurableObjectStorageError, R> {
67 return Effect.gen(function* () {
68 const transactions = yield* ActiveStorageTransactions;
69 const existing =
70 transactions.get(storage) ?? activeStorageTransactions.get(storage);
71 const evaluate = (transaction: ActiveStorageTransaction) =>
72 Effect.suspend(() =>
73 typeof body === "function"
74 ? body(
75 makeDurableObjectTransaction(
76 transaction.transaction,
77 checkTransactionOwner(transaction).pipe(Effect.orDie),
78 checkTransactionOwner(transaction, true).pipe(
79 Effect.andThen(() =>
80 Effect.sync(() => {
81 if (!transaction.rolledBack) {
82 transaction.transaction.rollback();
83 transaction.rolledBack = true;
84 }
85 }),
86 ),
87 Effect.orDie,
88 ),
89 flushDurableObjectAlarm(storage),
90 ),
91 )
92 : body,
93 );
94
95 if (existing !== undefined) {
96 yield* checkTransactionOwner(existing);
97 return yield* evaluate(existing);
98 }
99
100 const context = yield* Effect.context<R>();
101 // Native input gates can block timers; fiber yields must use microtasks.
102 const scheduler = new Scheduler.MixedScheduler("sync");
103
104 return yield* Effect.scoped(
105 Effect.gen(function* () {
106 let cancelled = false;
107 let callbackFiber: Fiber.Fiber<A, E> | undefined;
108 let native: Promise<A> | undefined;
109 let failure: Exit.Failure<A, E> | undefined;
110 let currentTransaction: ActiveStorageTransaction | undefined;
111 const release = () => {
112 if (currentTransaction !== undefined) {
113 currentTransaction.active = false;
114 if (activeStorageTransactions.get(storage) === currentTransaction) {
115 activeStorageTransactions.delete(storage);
116 }
117 }
118 };
119
120 yield* Effect.addFinalizer(
121 Effect.fn(function* () {
122 cancelled = true;
123 if (callbackFiber !== undefined) {
124 yield* Fiber.interrupt(callbackFiber);
125 }
126 const settlement = native;
127 if (settlement !== undefined) {
128 // The callback must finish before waiting for native rollback.
129 yield* Effect.promise(() =>
130 settlement.then(
131 () => undefined,
132 () => undefined,
133 ),
134 );
135 }
136 }),
137 );
138
139 return yield* Effect.callback<A, E | DurableObjectStorageError>(
140 (resume) => {
141 const reject = (cause: unknown) => {
142 release();
143 resume(
144 failure !== undefined && cause === failure
145 ? Effect.failCause(failure.cause)
146 : Effect.fail(
147 new DurableObjectStorageError({
148 operation: "transaction",
149 message: "The native storage transaction failed",
150 cause,
151 }),
152 ),
153 );
154 };
155
156 try {
157 native = storage.transaction((txn) => {
158 if (cancelled) {
159 throw new DurableObjectStorageError({
160 operation: "transaction",
161 message: "The storage transaction caller was interrupted",
162 });
163 }
164 const transaction: ActiveStorageTransaction = {
165 transaction: txn,
166 owner: undefined,
167 active: true,
168 rolledBack: false,
169 alarmTablesEnsured: false,
170 alarmDirty: false,
171 };
172 currentTransaction = transaction;
173 activeStorageTransactions.set(storage, transaction);
174 const callbackContext = Context.add(
175 context,
176 ActiveStorageTransactions,
177 new Map(transactions).set(storage, transaction),
178 );
179
180 return new Promise<A>((resolve, rejectCallback) => {
181 callbackFiber = Effect.runForkWith(callbackContext)(
182 Effect.withFiber((fiber) => {
183 transaction.owner = fiber;
184 return evaluate(transaction).pipe(
185 Effect.scoped,
186 Effect.tap(() => flushDurableObjectAlarm(storage)),
187 );
188 }),
189 { scheduler },
190 );
191 callbackFiber.addObserver((exit) => {
192 transaction.active = false;
193 if (Exit.isSuccess(exit)) {
194 resolve(exit.value);
195 } else {
196 failure = exit;
197 rejectCallback(exit);
198 }
199 });
200 });
201 });
202 native.then((value) => {
203 release();
204 resume(Effect.succeed(value));
205 }, reject);
206 } catch (cause) {
207 reject(cause);
208 }
209 },
210 );
211 }),
212 );
213 });
214}
215
216// ---------------------------------------------------------------------------
217// SqlStorage — Effect-native wrapper around cf.SqlStorage
218// ---------------------------------------------------------------------------
219
220export type SqlStorageValue = cf.SqlStorageValue;
221
222export interface SqlCursor<
223 T extends Record<string, SqlStorageValue>,
224> extends Stream.Stream<T> {
225 next(): Effect.Effect<
226 { done?: false; value: T } | { done: true; value?: never },
227 never,
228 RuntimeContext
229 >;
230 toArray(): Effect.Effect<T[], never, RuntimeContext>;
231 one(): Effect.Effect<T, never, RuntimeContext>;
232 raw<U extends SqlStorageValue[]>(): Stream.Stream<U, never, RuntimeContext>;
233 readonly columnNames: string[];
234 readonly rowsRead: Effect.Effect<number, never, RuntimeContext>;
235 readonly rowsWritten: Effect.Effect<number, never, RuntimeContext>;
236}
237
238export interface SqlStorage {
239 /**
240 * The raw underlying Cloudflare SqlStorage binding.
241 *
242 * Use this when you need direct access for libraries that already support
243 * Cloudflare Durable Object SQLite storage.
244 */
245 readonly raw: cf.SqlStorage;
246 exec<T extends Record<string, SqlStorageValue>>(
247 query: string,
248 ...bindings: any[]
249 ): Effect.Effect<SqlCursor<T>, never, RuntimeContext>;
250 readonly databaseSize: number;
251}
252
253const fromSqlCursor = <T extends Record<string, SqlStorageValue>>(
254 cursor: cf.SqlStorageCursor<T>,
255): SqlCursor<T> => {
256 const stream = Stream.fromIterableEffect(Effect.sync(() => cursor));
257 return Object.assign(stream, {
258 next: () => Effect.sync(() => cursor.next()),
259 toArray: () => Effect.sync(() => cursor.toArray()),
260 one: () => Effect.sync(() => cursor.one()),
261 raw: <U extends SqlStorageValue[]>() =>
262 Stream.fromIterableEffect(Effect.sync(() => cursor.raw<U>())),
263 get columnNames() {
264 return cursor.columnNames;
265 },
266 rowsRead: Effect.sync(() => cursor.rowsRead),
267 rowsWritten: Effect.sync(() => cursor.rowsWritten),
268 }) as SqlCursor<T>;
269};
270
271const fromSqlStorage = (
272 sql: cf.SqlStorage,
273 checkOwner: Effect.Effect<void>,
274): SqlStorage => ({
275 raw: sql,
276 exec: <T extends Record<string, SqlStorageValue>>(
277 query: string,
278 ...bindings: any[]
279 ): Effect.Effect<SqlCursor<T>> =>
280 Effect.andThen(
281 checkOwner,
282 Effect.sync(() => fromSqlCursor(sql.exec<T>(query, ...bindings))),
283 ),
284 get databaseSize() {
285 return sql.databaseSize;
286 },
287});
288
289// ---------------------------------------------------------------------------
290// DurableObjectTransaction
291// ---------------------------------------------------------------------------
292
293export interface DurableObjectTransaction {
294 get<T = unknown>(
295 key: string,
296 options?: cf.DurableObjectGetOptions,
297 ): Effect.Effect<T | undefined, never, RuntimeContext>;
298 get<T = unknown>(
299 keys: string[],
300 options?: cf.DurableObjectGetOptions,
301 ): Effect.Effect<Map<string, T>, never, RuntimeContext>;
302 list<T = unknown>(
303 options?: cf.DurableObjectListOptions,
304 ): Effect.Effect<Map<string, T>, never, RuntimeContext>;
305 put<T>(
306 key: string,
307 value: T,
308 options?: cf.DurableObjectPutOptions,
309 ): Effect.Effect<void, never, RuntimeContext>;
310 put<T>(
311 entries: Record<string, T>,
312 options?: cf.DurableObjectPutOptions,
313 ): Effect.Effect<void, never, RuntimeContext>;
314 delete(
315 key: string,
316 options?: cf.DurableObjectPutOptions,
317 ): Effect.Effect<boolean, never, RuntimeContext>;
318 delete(
319 keys: string[],
320 options?: cf.DurableObjectPutOptions,
321 ): Effect.Effect<number, never, RuntimeContext>;
322 rollback(): Effect.Effect<void, never, RuntimeContext>;
323 getAlarm(
324 options?: cf.DurableObjectGetAlarmOptions,
325 ): Effect.Effect<number | null, never, RuntimeContext>;
326 setAlarm(
327 scheduledTime: number | Date,
328 options?: cf.DurableObjectSetAlarmOptions,
329 ): Effect.Effect<void, never, RuntimeContext>;
330 deleteAlarm(
331 options?: cf.DurableObjectSetAlarmOptions,
332 ): Effect.Effect<void, never, RuntimeContext>;
333}
334
335// ---------------------------------------------------------------------------
336// DurableObjectStorage
337// ---------------------------------------------------------------------------
338
339export interface DurableObjectStorage {
340 get<T = unknown>(
341 key: string,
342 options?: cf.DurableObjectGetOptions,
343 ): Effect.Effect<T | undefined, never, RuntimeContext>;
344 get<T = unknown>(
345 keys: string[],
346 options?: cf.DurableObjectGetOptions,
347 ): Effect.Effect<Map<string, T>, never, RuntimeContext>;
348 list<T = unknown>(
349 options?: cf.DurableObjectListOptions,
350 ): Effect.Effect<Map<string, T>, never, RuntimeContext>;
351 put<T>(
352 key: string,
353 value: T,
354 options?: cf.DurableObjectPutOptions,
355 ): Effect.Effect<void, never, RuntimeContext>;
356 put<T>(
357 entries: Record<string, T>,
358 options?: cf.DurableObjectPutOptions,
359 ): Effect.Effect<void, never, RuntimeContext>;
360 delete(
361 key: string,
362 options?: cf.DurableObjectPutOptions,
363 ): Effect.Effect<boolean, never, RuntimeContext>;
364 delete(
365 keys: string[],
366 options?: cf.DurableObjectPutOptions,
367 ): Effect.Effect<number, never, RuntimeContext>;
368 deleteAll(
369 options?: cf.DurableObjectPutOptions,
370 ): Effect.Effect<void, never, RuntimeContext>;
371 /**
372 * Run an Effect inside a native storage transaction with the caller's context.
373 * On SQLite-backed storage, SQL, KV, and alarm writes to this storage commit
374 * together. Typed failures, defects, and interruption roll back the transaction.
375 * Interruption waits for the callback fiber and native rollback to finish.
376 * Nested calls on the same fiber join the active transaction; transaction
377 * handles cannot be used by another fiber or after the callback finishes.
378 */
379 transaction<A, E = never, R = never>(
380 effect: Effect.Effect<A, E, R>,
381 ): Effect.Effect<A, E | DurableObjectStorageError, R | RuntimeContext>;
382 transaction<A, E = never, R = never>(
383 closure: (txn: DurableObjectTransaction) => Effect.Effect<A, E, R>,
384 ): Effect.Effect<A, E | DurableObjectStorageError, R | RuntimeContext>;
385 getAlarm(
386 options?: cf.DurableObjectGetAlarmOptions,
387 ): Effect.Effect<number | null, never, RuntimeContext>;
388 setAlarm(
389 scheduledTime: number | Date,
390 options?: cf.DurableObjectSetAlarmOptions,
391 ): Effect.Effect<void, never, RuntimeContext>;
392 deleteAlarm(
393 options?: cf.DurableObjectSetAlarmOptions,
394 ): Effect.Effect<void, never, RuntimeContext>;
395 sync(): Effect.Effect<void, never, RuntimeContext>;
396 sql: SqlStorage;
397 kv: cf.SyncKvStorage;
398 getCurrentBookmark(): Effect.Effect<string, never, RuntimeContext>;
399 getBookmarkForTime(
400 timestamp: number | Date,
401 ): Effect.Effect<string, never, RuntimeContext>;
402 onNextSessionRestoreBookmark(
403 bookmark: string,
404 ): Effect.Effect<string, never, RuntimeContext>;
405}
406
407// ---------------------------------------------------------------------------
408// Constructors from raw Cloudflare types
409// ---------------------------------------------------------------------------
410
411export const fromDurableObjectTransaction = (
412 txn: cf.DurableObjectTransaction,
413): DurableObjectTransaction => makeDurableObjectTransaction(txn, Effect.void);
414
415const makeDurableObjectTransaction = (
416 txn: cf.DurableObjectTransaction,
417 checkOwner: Effect.Effect<void>,
418 rollback = Effect.andThen(
419 checkOwner,
420 Effect.sync(() => txn.rollback()),
421 ),
422 flushAlarm: Effect.Effect<void> = Effect.void,
423): DurableObjectTransaction => {
424 const use = <A>(effect: Effect.Effect<A>) =>
425 Effect.andThen(checkOwner, effect);
426 const useAlarm = <A>(effect: Effect.Effect<A>) =>
427 use(Effect.andThen(flushAlarm, effect));
428
429 return {
430 get: ((
431 keyOrKeys: string | string[],
432 options?: cf.DurableObjectGetOptions,
433 ) => use(Effect.promise(() => txn.get(keyOrKeys as any, options)))) as any,
434 list: (options?: cf.DurableObjectListOptions) =>
435 use(Effect.promise(() => txn.list(options))),
436 put: ((
437 keyOrEntries: string | Record<string, unknown>,
438 valueOrOptions?: unknown,
439 maybeOptions?: cf.DurableObjectPutOptions,
440 ) =>
441 use(
442 typeof keyOrEntries === "string"
443 ? Effect.promise(() =>
444 txn.put(keyOrEntries, valueOrOptions, maybeOptions),
445 )
446 : Effect.promise(() =>
447 txn.put(
448 keyOrEntries,
449 valueOrOptions as cf.DurableObjectPutOptions | undefined,
450 ),
451 ),
452 )) as any,
453 delete: ((
454 keyOrKeys: string | string[],
455 options?: cf.DurableObjectPutOptions,
456 ) =>
457 use(Effect.promise(() => txn.delete(keyOrKeys as any, options)))) as any,
458 rollback: () => rollback,
459 getAlarm: (options?: cf.DurableObjectGetAlarmOptions) =>
460 useAlarm(Effect.promise(() => txn.getAlarm(options))),
461 setAlarm: (
462 scheduledTime: number | Date,
463 options?: cf.DurableObjectSetAlarmOptions,
464 ) => useAlarm(Effect.promise(() => txn.setAlarm(scheduledTime, options))),
465 deleteAlarm: (options?: cf.DurableObjectSetAlarmOptions) =>
466 useAlarm(Effect.promise(() => txn.deleteAlarm(options))),
467 };
468};
469
470export const fromDurableObjectStorage = (
471 storage: cf.DurableObjectStorage,
472): DurableObjectStorage => {
473 const checkOwner = Effect.withFiber((fiber) => {
474 const transaction =
475 fiber.getRef(ActiveStorageTransactions).get(storage) ??
476 activeStorageTransactions.get(storage);
477 return transaction === undefined
478 ? Effect.void
479 : checkTransactionOwner(transaction).pipe(Effect.orDie);
480 });
481 const use = <A>(effect: Effect.Effect<A>) =>
482 Effect.andThen(checkOwner, effect);
483 const useAlarm = <A>(effect: Effect.Effect<A>) =>
484 use(Effect.andThen(flushDurableObjectAlarm(storage), effect));
485
486 return {
487 get: ((
488 keyOrKeys: string | string[],
489 options?: cf.DurableObjectGetOptions,
490 ) =>
491 use(Effect.promise(() => storage.get(keyOrKeys as any, options)))) as any,
492 list: (options?: cf.DurableObjectListOptions) =>
493 use(Effect.promise(() => storage.list(options))),
494 put: ((
495 keyOrEntries: string | Record<string, unknown>,
496 valueOrOptions?: unknown,
497 maybeOptions?: cf.DurableObjectPutOptions,
498 ) =>
499 use(
500 typeof keyOrEntries === "string"
501 ? Effect.promise(() =>
502 storage.put(keyOrEntries, valueOrOptions, maybeOptions),
503 )
504 : Effect.promise(() =>
505 storage.put(
506 keyOrEntries,
507 valueOrOptions as cf.DurableObjectPutOptions | undefined,
508 ),
509 ),
510 )) as any,
511 delete: ((
512 keyOrKeys: string | string[],
513 options?: cf.DurableObjectPutOptions,
514 ) =>
515 use(
516 Effect.promise(() => storage.delete(keyOrKeys as any, options)),
517 )) as any,
518 deleteAll: (options?: cf.DurableObjectPutOptions) =>
519 use(Effect.promise(() => storage.deleteAll(options))),
520 transaction: <A, E, R>(
521 body:
522 | Effect.Effect<A, E, R>
523 | ((txn: DurableObjectTransaction) => Effect.Effect<A, E, R>),
524 ) =>
525 withStorageTransaction(storage, (txn) =>
526 typeof body === "function" ? body(txn) : body,
527 ),
528 getAlarm: (options?: cf.DurableObjectGetAlarmOptions) =>
529 useAlarm(Effect.promise(() => storage.getAlarm(options))),
530 setAlarm: (
531 scheduledTime: number | Date,
532 options?: cf.DurableObjectSetAlarmOptions,
533 ) =>
534 useAlarm(Effect.promise(() => storage.setAlarm(scheduledTime, options))),
535 deleteAlarm: (options?: cf.DurableObjectSetAlarmOptions) =>
536 useAlarm(Effect.promise(() => storage.deleteAlarm(options))),
537 sync: () => use(Effect.promise(() => storage.sync())),
538 sql: fromSqlStorage(storage.sql, checkOwner),
539 kv: storage.kv,
540 getCurrentBookmark: () =>
541 use(Effect.promise(() => storage.getCurrentBookmark())),
542 getBookmarkForTime: (timestamp: number | Date) =>
543 use(Effect.promise(() => storage.getBookmarkForTime(timestamp))),
544 onNextSessionRestoreBookmark: (bookmark: string) =>
545 use(Effect.promise(() => storage.onNextSessionRestoreBookmark(bookmark))),
546 };
547};