1import * as Cloudflare from "@/Cloudflare/index.ts";
2import * as Deferred from "effect/Deferred";
3import * as Effect from "effect/Effect";
4import * as Schema from "effect/Schema";
5import * as Stream from "effect/Stream";
6import { Greeting, Rejected, SocketRpcs } from "./rpcs.ts";
7
8export class SocketObject extends Cloudflare.RpcDurableObject<SocketObject>()(
9 "SocketObject",
10 { schema: SocketRpcs },
11) {}
12
13export const SocketObjectLive = SocketObject.make(
14 Effect.gen(function* () {
15 const state = yield* Cloudflare.DurableObjectState;
16 return Effect.gen(function* () {
17 const boots = ((yield* state.storage.get<number>("boots")) ?? 0) + 1;
18 yield* state.storage.put("boots", boots);
19 const opened: string[] = [];
20 const closed: string[] = [];
21 const cleanupStarted: string[] = [];
22 const cleanups = new Map<string, Deferred.Deferred<void>>();
23 return SocketRpcs.toLayer({
24 greet: ({ name }) =>
25 Effect.succeed(new Greeting({ message: `Hello, ${name}!` })),
26 increment: () =>
27 state.storage
28 .transaction(
29 Effect.gen(function* () {
30 const count =
31 ((yield* state.storage.get<number>("count")) ?? 0) + 1;
32 yield* state.storage.put("count", count);
33 return count;
34 }),
35 )
36 .pipe(Effect.orDie),
37 reject: () =>
38 Effect.fail(new Rejected({ message: "rejected by handler" })),
39 numbers: ({ count }) => Stream.range(1, count),
40 watch: ({ key }) =>
41 Stream.unwrap(
42 Effect.gen(function* () {
43 yield* state.storage
44 .transaction(
45 Effect.gen(function* () {
46 const invocations =
47 (yield* state.storage.get<Record<string, number>>(
48 "invocations",
49 )) ?? {};
50 yield* state.storage.put("invocations", {
51 ...invocations,
52 [key]: (invocations[key] ?? 0) + 1,
53 });
54 }),
55 )
56 .pipe(Effect.orDie);
57 yield* Effect.sync(() => opened.push(key));
58 yield* Effect.addFinalizer(() =>
59 Effect.sync(() => closed.push(key)),
60 );
61 return Stream.concat(
62 Stream.make(1),
63 Stream.fromEffect(Effect.never),
64 );
65 }),
66 ),
67 cleanup: ({ key, waitForDisconnect }) =>
68 Effect.gen(function* () {
69 const release = yield* Deferred.make<void>();
70 yield* Effect.addFinalizer(() =>
71 Effect.gen(function* () {
72 yield* Effect.sync(() => cleanupStarted.push(key));
73 // No timer or I/O keeps the activation alive while cleanup waits.
74 yield* Deferred.await(release);
75 yield* state.storage
76 .transaction(
77 Effect.gen(function* () {
78 const completed =
79 (yield* state.storage.get<Record<string, number>>(
80 "cleanupCompleted",
81 )) ?? {};
82 yield* state.storage.put("cleanupCompleted", {
83 ...completed,
84 [key]: boots,
85 });
86 }),
87 )
88 .pipe(Effect.orDie);
89 yield* Effect.sync(() => cleanups.delete(key));
90 }),
91 );
92 yield* Effect.sync(() => {
93 cleanups.set(key, release);
94 opened.push(key);
95 });
96 return yield* waitForDisconnect
97 ? Effect.never
98 : Effect.succeed(boots);
99 }),
100 releaseCleanup: ({ key }) =>
101 Effect.suspend(() => {
102 const release = cleanups.get(key);
103 return release === undefined
104 ? Effect.succeed(false)
105 : Deferred.succeed(release, undefined);
106 }),
107 invalidateSocketSerialization: Effect.fn(function* () {
108 const sockets = yield* state.getWebSockets("alchemy:rpc");
109 const attachmentSchema = Schema.Struct({
110 __alchemyRpcWebSocket: Schema.Struct({
111 version: Schema.Literal(1),
112 clientId: Schema.Number,
113 pending: Schema.Boolean,
114 serialization: Schema.String,
115 }),
116 });
117 for (const socket of sockets) {
118 yield* Effect.sync(() => {
119 const attachment = socket.deserializeAttachment<unknown>();
120 if (!Schema.is(attachmentSchema)(attachment)) {
121 throw new Error("Missing RPC WebSocket attachment");
122 }
123 if (attachment.__alchemyRpcWebSocket.pending) {
124 throw new Error("Cannot change serialization during an RPC");
125 }
126 socket.serializeAttachment({
127 ...attachment,
128 __alchemyRpcWebSocket: {
129 ...attachment.__alchemyRpcWebSocket,
130 serialization: "application/x-incompatible-rpc",
131 },
132 });
133 });
134 }
135 return sockets.length;
136 }),
137 abort: Effect.fn(function* () {
138 // Native abort discards buffered storage writes.
139 yield* state.storage.sync();
140 yield* state.abort("RPC WebSocket test abort", { retryAlarm: false });
141 }),
142 stats: Effect.fn(function* () {
143 return {
144 boots,
145 count: (yield* state.storage.get<number>("count")) ?? 0,
146 opened: [...opened],
147 closed: [...closed],
148 invocations:
149 (yield* state.storage.get<Record<string, number>>(
150 "invocations",
151 )) ?? {},
152 cleanupStarted: [...cleanupStarted],
153 cleanupCompleted:
154 (yield* state.storage.get<Record<string, number>>(
155 "cleanupCompleted",
156 )) ?? {},
157 };
158 }),
159 });
160 });
161 }),
162);