Skip to content

Commit b501f2e

Browse files
ImriKochWixclaude
ImriKochWix
and
claude
committed
feat(actors): room-handle API — actors.Name(id).connect() then subscribe/send
Replaces the flat subscribe(roomId, cb) / send(roomId, msg) surface. A room handle IS the connection: connect() opens the socket (required, idempotent), subscribe() registers one of N independent listeners (returns a per-listener unsubscribe), send() throws before connect, close() tears down socket + heartbeat + all listeners. Fixes the old wart where a second subscribe to the same room closed the first socket. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
1 parent e370e5f commit b501f2e

3 files changed

Lines changed: 284 additions & 143 deletions

File tree

‎src/modules/actors.ts‎

Lines changed: 118 additions & 114 deletions
Original file line numberDiff line numberDiff line change
@@ -1,13 +1,11 @@
11
import PartySocket from "partysocket";
2+
import type {
3+
ActorConnectOptions,
4+
ActorRoom,
5+
ActorSubscription,
6+
} from "./actors.types.js";
27

3-
// Module-level map: "ActorName:instanceId" → active socket
4-
const activeSockets = new Map<string, PartySocket>();
5-
6-
function socketKey(actorName: string, instanceId: string) {
7-
return `${actorName}:${instanceId}`;
8-
}
9-
10-
export function createActorsModule(config: {
8+
interface ActorsConfig {
119
appId: string;
1210
/** Current user access token, if authenticated. Rides the WS query (same
1311
* pattern as the entities socket) so the platform proxy can authenticate the
@@ -19,120 +17,126 @@ export function createActorsModule(config: {
1917
dispatcherWsUrl: string;
2018
/** WebSocket implementation for runtimes without a global one (Node < 22). */
2119
webSocketImpl?: unknown;
22-
}) {
23-
return new Proxy({} as Record<string, ActorClient>, {
24-
get(_, actorName: string) {
25-
return {
26-
subscribe(
27-
instanceId: string,
28-
callback: (data: unknown) => void,
29-
options?: { id?: string },
30-
): ActorSubscription {
31-
const key = socketKey(actorName, instanceId);
32-
// close existing if any
33-
activeSockets.get(key)?.close();
20+
}
3421

35-
// Connection id: caller-supplied (stable — reuse across reconnects/tabs as
36-
// you see fit) or auto-generated per subscription. PartySocket sends it
37-
// as ?_pk=; the platform proxy validates it and the actor sees this exact
38-
// value as conn.id, stable across reconnects.
39-
const connId = options?.id ?? crypto.randomUUID();
22+
// Heartbeat / half-open detection. PartySocket only reconnects on a browser
23+
// close/error event, so a silently-dead connection (TCP alive, no data — common
24+
// behind proxies/LBs) hangs until the OS idle timeout (~60s). Ping periodically
25+
// and force a reconnect if nothing comes back within DEAD_MS.
26+
const PING_MS = 1_000;
27+
const DEAD_MS = 3_000;
4028

41-
// No pre-connect token mint: the platform proxy authenticates the
42-
// connection itself, exactly like a backend-function call. `handler`
43-
// carries the case-preserved actor name (the `party` path segment is
44-
// lowercased by PartySocket). query as fn: re-read on every (re)connect
45-
// so a login/logout between reconnects is picked up.
46-
const ws = new PartySocket({
47-
host: config.dispatcherWsUrl,
48-
party: actorName,
49-
room: instanceId,
50-
id: connId,
51-
...(config.webSocketImpl ? { WebSocket: config.webSocketImpl as any } : {}),
52-
query: () => {
53-
const token = config.getAuthToken();
54-
return {
55-
app_id: config.appId,
56-
handler: actorName,
57-
...(token ? { token } : {}),
58-
...(config.functionsVersion ? { fv: config.functionsVersion } : {}),
59-
};
60-
},
61-
});
29+
class Room {
30+
private ws: PartySocket | null = null;
31+
private readonly listeners = new Set<(data: unknown) => void>();
32+
private heartbeat: ReturnType<typeof setInterval> | null = null;
33+
private connId: string | null = null;
6234

63-
activeSockets.set(key, ws);
35+
constructor(
36+
private readonly actorName: string,
37+
private readonly instanceId: string,
38+
private readonly config: ActorsConfig,
39+
) {}
6440

65-
// Heartbeat / half-open detection. PartySocket only reconnects on a
66-
// browser close/error event, so a silently-dead connection (TCP alive,
67-
// no data — common behind proxies/LBs) hangs until the OS idle timeout
68-
// (~60s). We ping periodically and force a reconnect if nothing comes
69-
// back within DEAD_MS, cutting detection from ~60s to a few seconds.
70-
const PING_MS = 1_000;
71-
const DEAD_MS = 3_000;
72-
let lastMsg = Date.now();
73-
const bumpAlive = () => { lastMsg = Date.now(); };
41+
get id(): string {
42+
if (!this.connId) {
43+
throw new Error(`${this.actorName}:${this.instanceId}: connect() before reading id`);
44+
}
45+
return this.connId;
46+
}
7447

75-
ws.addEventListener("open", bumpAlive);
76-
ws.addEventListener("message", (ev) => {
77-
bumpAlive();
78-
let data: unknown;
79-
try {
80-
data = JSON.parse(ev.data);
81-
} catch {
82-
return; // ignore malformed
83-
}
84-
// Swallow platform messages — never surface them to the app.
85-
const msgType = data && typeof data === "object" ? (data as { type?: unknown }).type : undefined;
86-
if (msgType === "__pong") return;
87-
callback(data);
88-
});
48+
connect(options?: ActorConnectOptions): this {
49+
if (this.ws) return this;
8950

90-
const heartbeat = setInterval(() => {
91-
if (Date.now() - lastMsg > DEAD_MS) {
92-
bumpAlive(); // avoid a reconnect storm while the new socket comes up
93-
ws.reconnect();
94-
return;
95-
}
96-
try {
97-
ws.send(JSON.stringify({ type: "__ping" }));
98-
} catch {
99-
// socket not open; the watchdog above will force a reconnect
100-
}
101-
}, PING_MS);
51+
// The client picks its own conn id; it becomes _pk → the actor's conn.id.
52+
const connId = options?.id ?? crypto.randomUUID();
53+
this.connId = connId;
10254

103-
return {
104-
id: connId, // the connection id (same value the actor sees as conn.id)
105-
unsubscribe() {
106-
clearInterval(heartbeat);
107-
activeSockets.delete(key);
108-
ws.close();
109-
},
110-
};
111-
},
112-
send(instanceId: string, data: unknown) {
113-
const key = socketKey(actorName, instanceId);
114-
const ws = activeSockets.get(key);
115-
if (!ws) throw new Error(`No active subscription for ${actorName}:${instanceId}`);
116-
ws.send(JSON.stringify(data));
117-
},
118-
};
119-
},
120-
});
121-
}
55+
const ws = new PartySocket({
56+
host: this.config.dispatcherWsUrl,
57+
party: this.actorName,
58+
room: this.instanceId,
59+
id: connId,
60+
...(this.config.webSocketImpl ? { WebSocket: this.config.webSocketImpl as any } : {}),
61+
// Re-read on every (re)connect so a login/logout between reconnects is
62+
// picked up. The platform proxy authenticates the connection itself —
63+
// no pre-connect token mint.
64+
query: () => {
65+
const token = this.config.getAuthToken();
66+
return {
67+
app_id: this.config.appId,
68+
handler: this.actorName,
69+
...(token ? { token } : {}),
70+
...(this.config.functionsVersion ? { fv: this.config.functionsVersion } : {}),
71+
};
72+
},
73+
});
74+
this.ws = ws;
12275

123-
/** Handle for an active actor subscription. */
124-
interface ActorSubscription {
125-
/** This connection's id — the same value the actor receives as `conn.id`. */
126-
id: string;
127-
/** Close the subscription and its underlying socket. */
128-
unsubscribe(): void;
76+
let lastMsg = Date.now();
77+
const bumpAlive = () => { lastMsg = Date.now(); };
78+
ws.addEventListener("open", bumpAlive);
79+
ws.addEventListener("message", (ev) => {
80+
bumpAlive();
81+
let data: unknown;
82+
try {
83+
data = JSON.parse(ev.data);
84+
} catch {
85+
return; // ignore malformed
86+
}
87+
const msgType = data && typeof data === "object" ? (data as { type?: unknown }).type : undefined;
88+
if (msgType === "__pong") return; // platform message — never surface it
89+
for (const listener of this.listeners) listener(data);
90+
});
91+
92+
this.heartbeat = setInterval(() => {
93+
if (Date.now() - lastMsg > DEAD_MS) {
94+
bumpAlive(); // avoid a reconnect storm while the new socket comes up
95+
ws.reconnect();
96+
return;
97+
}
98+
try {
99+
ws.send(JSON.stringify({ type: "__ping" }));
100+
} catch {
101+
// socket not open; the watchdog above will force a reconnect
102+
}
103+
}, PING_MS);
104+
105+
return this;
106+
}
107+
108+
subscribe(callback: (data: unknown) => void): ActorSubscription {
109+
if (!this.ws) {
110+
throw new Error(`${this.actorName}:${this.instanceId}: connect() before subscribe()`);
111+
}
112+
this.listeners.add(callback);
113+
return {
114+
unsubscribe: () => { this.listeners.delete(callback); },
115+
};
116+
}
117+
118+
send(data: unknown): void {
119+
if (!this.ws) {
120+
throw new Error(`${this.actorName}:${this.instanceId}: connect() before send()`);
121+
}
122+
this.ws.send(JSON.stringify(data));
123+
}
124+
125+
close(): void {
126+
if (this.heartbeat) {
127+
clearInterval(this.heartbeat);
128+
this.heartbeat = null;
129+
}
130+
this.listeners.clear();
131+
this.ws?.close();
132+
this.ws = null;
133+
}
129134
}
130135

131-
interface ActorClient {
132-
subscribe(
133-
instanceId: string,
134-
callback: (data: unknown) => void,
135-
options?: { id?: string },
136-
): ActorSubscription;
137-
send(instanceId: string, data: unknown): void;
136+
export function createActorsModule(config: ActorsConfig) {
137+
return new Proxy({} as Record<string, (instanceId: string) => ActorRoom>, {
138+
get(_, actorName: string) {
139+
return (instanceId: string) => new Room(actorName, instanceId, config) as unknown as ActorRoom;
140+
},
141+
});
138142
}

‎src/modules/actors.types.ts‎

Lines changed: 44 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -39,48 +39,63 @@ type ToServerFor<N extends string> = N extends keyof ActorRegistry
3939
: unknown
4040
: unknown;
4141

42-
/**
43-
* Client for a single named Actor.
44-
* Typed automatically when the actor is registered in {@link ActorRegistry}.
45-
*/
46-
export interface ActorClient<N extends string = string> {
42+
/** Options for {@link ActorRoom.connect}. */
43+
export interface ActorConnectOptions {
4744
/**
48-
* Open a WebSocket subscription. Returns a {@link ActorSubscription} with the
49-
* connection `id` (same value the actor sees as `conn.id`) and an `unsubscribe()` method.
50-
*
51-
* Pass `options.id` to control the connection id (e.g. a stable per-tab id so a
52-
* reconnect reuses the same server-side connection); omit it for an auto-generated
53-
* per-connection id.
45+
* The connection id — becomes the actor's `conn.id`. Supply a stable value
46+
* (e.g. persisted per tab) so a reconnect reuses the same server-side
47+
* identity; omit for an auto-generated per-connection id.
5448
*/
55-
subscribe(
56-
instanceId: string,
57-
callback: (data: ToClientFor<N>) => void,
58-
options?: { id?: string },
59-
): ActorSubscription;
60-
61-
/** Send a message over the open socket. Throws if not subscribed. */
62-
send(instanceId: string, data: ToServerFor<N>): void;
49+
id?: string;
6350
}
6451

65-
/** Handle for an active actor subscription. */
52+
/** Handle for one listener registered via {@link ActorRoom.subscribe}. */
6653
export interface ActorSubscription {
67-
/** This connection's id — the same value the actor receives as `conn.id`. */
68-
id: string;
69-
/** Close the subscription and its underlying socket. */
54+
/** Remove this listener; other listeners and the socket stay live. */
7055
unsubscribe(): void;
7156
}
7257

58+
/**
59+
* A single actor room. Obtained from {@link ActorClient} (`actors.MyActor(id)`)
60+
* and made live with {@link connect}. The handle IS the connection: one socket,
61+
* any number of {@link subscribe} listeners.
62+
*/
63+
export interface ActorRoom<N extends string = string> {
64+
/** The connection id (the value the actor sees as `conn.id`). Throws before {@link connect}. */
65+
readonly id: string;
66+
67+
/** Open the WebSocket (required before subscribe/send). Idempotent; returns this. */
68+
connect(options?: ActorConnectOptions): this;
69+
70+
/** Register a message listener. Multiple are allowed; returns a per-listener unsubscribe. */
71+
subscribe(callback: (data: ToClientFor<N>) => void): ActorSubscription;
72+
73+
/** Send a message. Throws before {@link connect}; buffered by the socket until open. */
74+
send(data: ToServerFor<N>): void;
75+
76+
/** Tear down the socket, heartbeat, and all listeners. */
77+
close(): void;
78+
}
79+
80+
/**
81+
* Client for a single named Actor — call it with a room id to get an
82+
* {@link ActorRoom}. Typed automatically when the actor is registered in
83+
* {@link ActorRegistry}.
84+
*/
85+
export interface ActorClient<N extends string = string> {
86+
(instanceId: string): ActorRoom<N>;
87+
}
88+
7389
/**
7490
* The actors module provides access to Cloudflare Durable Object-backed
7591
* Actors deployed by the Base44 platform.
7692
*
77-
* Actor names are accessed as dynamic properties on this module:
7893
* ```typescript
79-
* const sub = await base44.actors.MyActor.subscribe("room-1", (msg) => {
80-
* console.log(msg); // typed if MyActor is in ActorRegistry
81-
* });
82-
* const { id, unsubscribe } = sub;
83-
* unsubscribe();
94+
* const room = base44.actors.MyActor("room-1").connect();
95+
* const sub = room.subscribe((msg) => console.log(msg)); // typed via ActorRegistry
96+
* room.send({ type: "message", text: "hi" });
97+
* sub.unsubscribe();
98+
* room.close();
8499
* ```
85100
*/
86101
export type ActorsModule = {

0 commit comments

Comments
 (0)