1import type * as cf from "@cloudflare/workers-types";
2import * as Context from "effect/Context";
3import * as Effect from "effect/Effect";
4import type { RuntimeContext } from "../../RuntimeContext.ts";
5import {
6 fromDurableObjectStorage,
7 type DurableObjectStorage,
8} from "./DurableObjectStorage.ts";
9import { fromWebSocket, type WebSocket } from "./WebSocket.ts";
10
11/**
12 * Options for {@link DurableObjectState.abort}.
13 *
14 * `retryAlarm` defaults to `true`: an alarm interrupted by `abort` is
15 * retried after the isolate resets. Set it to `false` to cancel that
16 * alarm instead.
17 */
18export type DurableObjectAbortOptions = cf.DurableObjectAbortOptions;
19
20export class DurableObjectState extends Context.Service<
21 DurableObjectState,
22 {
23 readonly id: cf.DurableObjectId;
24 readonly storage: DurableObjectStorage;
25 container?: cf.Container;
26 /**
27 * Run an Effect in the background without blocking the current event,
28 * keeping the Durable Object alive until it settles. The Effect runs with
29 * the caller's full context (services, tracing), and the resulting
30 * promise is registered with workerd's `state.waitUntil`.
31 */
32 waitUntil<A, E, R>(
33 effect: Effect.Effect<A, E, R>,
34 ): Effect.Effect<void, never, R | RuntimeContext>;
35 /**
36 * The raw workerd DurableObjectState, for interop with async APIs.
37 */
38 readonly raw: cf.DurableObjectState;
39 /**
40 * Run `callback` while workerd holds every other event on this object.
41 * The callback runs with the caller's full context (services, tracing),
42 * as `waitUntil` does, so a service provided to the calling fiber is
43 * visible inside the gate. A defect in the callback rejects the gate
44 * and workerd resets the object, which is the platform's contract.
45 */
46 blockConcurrencyWhile<T, R = never>(
47 callback: () => Effect.Effect<T, never, R>,
48 ): Effect.Effect<T, never, R | RuntimeContext>;
49 acceptWebSocket(
50 ws: WebSocket,
51 tags?: string[],
52 ): Effect.Effect<void, never, RuntimeContext>;
53 getWebSockets(
54 tag?: string,
55 ): Effect.Effect<WebSocket[], never, RuntimeContext>;
56 setWebSocketAutoResponse(
57 maybeReqResp?: cf.WebSocketRequestResponsePair,
58 ): Effect.Effect<void, never, RuntimeContext>;
59 getWebSocketAutoResponse(): Effect.Effect<
60 cf.WebSocketRequestResponsePair | null,
61 never,
62 RuntimeContext
63 >;
64 getWebSocketAutoResponseTimestamp(
65 ws: cf.WebSocket,
66 ): Effect.Effect<Date | null, never, RuntimeContext>;
67 setHibernatableWebSocketEventTimeout(
68 timeoutMs?: number,
69 ): Effect.Effect<void, never, RuntimeContext>;
70 getHibernatableWebSocketEventTimeout(): Effect.Effect<
71 number | null,
72 never,
73 RuntimeContext
74 >;
75 getTags(ws: cf.WebSocket): Effect.Effect<string[], never, RuntimeContext>;
76 /**
77 * Forcibly reset this Durable Object. A JavaScript `Error` with the
78 * given message is logged and cannot be caught in application code.
79 *
80 * By default an in-progress alarm retries after the reset. Pass
81 * `{ retryAlarm: false }` to stop it instead — for example an `alarm`
82 * handler that deletes storage so the constructor does not recreate it.
83 */
84 abort(
85 reason?: string,
86 options?: DurableObjectAbortOptions,
87 ): Effect.Effect<void, never, RuntimeContext>;
88 }
89>()("Cloudflare.DurableObjectState") {}
90
91export const fromDurableObjectState = (
92 state: cf.DurableObjectState,
93): DurableObjectState["Service"] => ({
94 id: state.id,
95 container: state.container,
96 storage: fromDurableObjectStorage(state.storage),
97 raw: state,
98 waitUntil: <A, E, R>(effect: Effect.Effect<A, E, R>) =>
99 Effect.gen(function* () {
100 const context = yield* Effect.context<R>();
101 // Register the promise with workerd un-awaited — waitUntil extends the
102 // event's lifetime without blocking the caller.
103 yield* Effect.sync(() =>
104 state.waitUntil(
105 Effect.runPromise(effect.pipe(Effect.provide(context))),
106 ),
107 );
108 }),
109 blockConcurrencyWhile: <T, R = never>(
110 callback: () => Effect.Effect<T, never, R>,
111 ) =>
112 Effect.gen(function* () {
113 const context = yield* Effect.context<R>();
114 // The failure is typed away as before: a rejected gate is the
115 // platform resetting the object, not a value a caller handles.
116 return yield* Effect.promise(() =>
117 state.blockConcurrencyWhile(() =>
118 Effect.runPromise(callback().pipe(Effect.provide(context))),
119 ),
120 );
121 }),
122 acceptWebSocket: (ws: WebSocket, tags?: string[]) =>
123 Effect.sync(() => state.acceptWebSocket(ws.ws, tags)),
124 getWebSockets: (tag?: string) =>
125 Effect.sync(() => state.getWebSockets(tag).map(fromWebSocket)),
126 setWebSocketAutoResponse: (maybeReqResp?: cf.WebSocketRequestResponsePair) =>
127 Effect.sync(() => state.setWebSocketAutoResponse(maybeReqResp)),
128 getWebSocketAutoResponse: () =>
129 Effect.sync(() => state.getWebSocketAutoResponse()),
130 getWebSocketAutoResponseTimestamp: (ws: cf.WebSocket) =>
131 Effect.sync(() => state.getWebSocketAutoResponseTimestamp(ws)),
132 setHibernatableWebSocketEventTimeout: (timeoutMs?: number) =>
133 Effect.sync(() => state.setHibernatableWebSocketEventTimeout(timeoutMs)),
134 getHibernatableWebSocketEventTimeout: () =>
135 Effect.sync(() => state.getHibernatableWebSocketEventTimeout()),
136 getTags: (ws: cf.WebSocket) => Effect.sync(() => state.getTags(ws)),
137 abort: (reason?: string, options?: DurableObjectAbortOptions) =>
138 Effect.sync(() => state.abort(reason, options)),
139});