work with direct actors

Author: talge-a11y

src/client.ts

@@ -20,7 +20,11 @@ import type {
   CreateClientOptions,
 } from "./client.types.js";
 import { createAnalyticsModule } from "./modules/analytics.js";
-import { createActorsModule, resolveActorsHost } from "./modules/actors.js";
+import {
+  createActorsModule,
+  resolveActorsHost,
+  type ActorConnectionCredentials,
+} from "./modules/actors.js";
 
 // Re-export client types
 export type { Base44Client, CreateClientConfig, CreateClientOptions };
@@ -143,6 +147,16 @@ export function createClient(config: CreateClientConfig): Base44Client {
     interceptResponses: false,
   });
 
+  // Dedicated client for actor connection-token mints: no onError (a legacy
+  // actor answers every mint with an expected 409 before the proxy fallback,
+  // which must not reach the app's error handler — the actors module forwards
+  // genuine failures itself via onMintError) and no constructor token
+  // (auth is per-request so a login/logout is picked up on every reconnect).
+  const actorsAxiosClient = createAxiosClient({
+    baseURL: `${serverUrl}/api`,
+    headers,
+  });
+
   const userAuthModule = createAuthModule(
     axiosClient,
     functionsAxiosClient,
@@ -167,14 +181,33 @@ export function createClient(config: CreateClientConfig): Base44Client {
 
   const actorsModule = createActorsModule({
     appId,
-    // serverUrl is often relative/empty (same-origin app); PartySocket needs an
-    // absolute host, so fall back to the page origin.
+    // serverUrl is often relative/empty (same-origin app); the proxy-fallback
+    // URL needs an absolute host, so fall back to the page origin.
     host: resolveActorsHost(
       serverUrl,
       typeof window !== "undefined" ? window.location?.origin : undefined,
     ),
     functionsVersion,
     getAuthToken: () => token || getAccessToken(),
+    mintConnectionToken: async (actorName, room, connectionId) => {
+      const authToken = token || getAccessToken();
+      return await actorsAxiosClient.post<unknown, ActorConnectionCredentials>(
+        `/apps/${appId}/actors/${encodeURIComponent(actorName)}/connection-token`,
+        { room, connection_id: connectionId },
+        {
+          headers: {
+            ...(authToken ? { Authorization: `Bearer ${authToken}` } : {}),
+            // The mint endpoint resolves draft vs published from this header;
+            // only the functions axios clients send it by default.
+            ...(functionsVersion
+              ? { "Base44-Functions-Version": functionsVersion }
+              : {}),
+          },
+        },
+      );
+    },
+    transport: options?.actorsTransport,
+    onMintError: options?.onError,
   });
 
   const userModules = {

src/client.types.ts

@@ -19,8 +19,20 @@ import type { ActorsModule } from "./modules/actors.types.js";
 export interface CreateClientOptions {
   /**
    * Optional error handler that will be called whenever an API error occurs.
+   *
+   * Also receives {@link ActorsModule | actors} connection failures. Errors
+   * are usually {@linkcode Base44Error} instances — check `error.status`.
    */
   onError?: (error: Error) => void;
+  /**
+   * Forces the actors transport. `"auto"` (default) connects directly to the
+   * actor and falls back to the platform proxy when the app's actors don't
+   * support direct connections; `"proxy"` always uses the platform proxy
+   * (ops rollback — no connection-token calls); `"direct"` disables the
+   * fallback (validation environments).
+   * @internal
+   */
+  actorsTransport?: "auto" | "proxy" | "direct";
 }
 
 /**

src/modules/actors.ts

@@ -1,38 +1,91 @@
-import PartySocket from "partysocket";
+import { WebSocket as ReconnectingWebSocket } from "partysocket";
 import type {
   ActorConnectOptions,
   ActorRef,
   Connection as ConnectionType,
   ActorSubscription,
 } from "./actors.types.js";
 
+/** Credentials minted by the platform for one direct actor connection. */
+export interface ActorConnectionCredentials {
+  /** Direct actor endpoint, already carrying `?_pk=<connectionId>`. */
+  websocket_url: string;
+  /** Short-lived JWT bound to (app, actor, room, connectionId); appended to
+   * the URL as `token=` since browsers can't set WebSocket headers. */
+  token: string;
+}
+
 interface ActorsConfig {
   appId: string;
-  /** Current user access token, if authenticated. Rides the WS query so the
-   * platform proxy can authenticate the connection; anonymous connects omit it. */
+  /** Current user access token, if authenticated. Rides the WS query on the
+   * proxy-fallback path so the platform proxy can authenticate the connection;
+   * anonymous connects omit it. */
   getAuthToken(): string | null | undefined;
   /** Same semantics as function calls: editors with a non-prod version get the
    * draft actor script; everyone else gets the published one. */
   functionsVersion?: string;
-  /** Absolute host PartySocket dials (it strips the scheme and connects wss, ws
+  /** Absolute host for the proxy-fallback URL (scheme is swapped to wss, ws
    * for localhost). Resolved by {@link resolveActorsHost}. */
   host: string;
+  /** Mints a direct-connect credential for one (actor, room, connection).
+   * Called per connection attempt: the token's expiry is checked at upgrade,
+   * so every reconnect needs a fresh one. */
+  mintConnectionToken(
+    actorName: string,
+    room: string,
+    connectionId: string,
+  ): Promise<ActorConnectionCredentials>;
+  /** @internal Ops escape hatch: "proxy" never mints (legacy path only),
+   * "direct" never falls back. Default "auto". */
+  transport?: "auto" | "proxy" | "direct";
+  /** Called when a mint fails for a reason other than the expected
+   * direct→proxy fallback (which recovers by itself). Wired to the client's
+   * `options.onError`. */
+  onMintError?: (error: Error) => void;
 }
 
-// Heartbeat / half-open detection: PartySocket only reconnects on a close/error
+// Heartbeat / half-open detection: the socket only reconnects on a close/error
 // event, so ping periodically and force a reconnect if nothing returns in DEAD_MS.
 const PING_MS = 1_000;
 const DEAD_MS = 3_000;
 
+// Mint responses that mean "direct can't serve this connection, the proxy can":
+// 409 = legacy-family actor script, 503 = direct connections not provisioned,
+// 422 = no principal (e.g. anonymous outside a browser) or an id/room only the
+// proxy's looser validation accepts. The proxy serves migrated actors too, so
+// falling back is always safe.
+const PROXY_FALLBACK_STATUSES = new Set([409, 422, 503]);
+
+// Mint responses no retry can fix (bad request / forbidden / not found): the
+// connection closes instead of re-minting forever; a fresh connect() re-probes.
+// 401 is deliberately absent — the auth token is re-read on every attempt, so a
+// login recovers on the next retry. Disjoint from PROXY_FALLBACK_STATUSES.
+const TERMINAL_MINT_STATUSES = new Set([400, 403, 404]);
+
+/** The mint's rejection can be anything; a `Base44Error` carries a numeric
+ * `.status` (absent for network failures). */
+function mintErrorStatus(err: unknown): number | undefined {
+  const status =
+    err && typeof err === "object"
+      ? (err as { status?: unknown }).status
+      : undefined;
+  return typeof status === "number" ? status : undefined;
+}
+
+function toError(err: unknown): Error {
+  return err instanceof Error ? err : new Error(String(err));
+}
+
 /**
  * A live connection to an actor instance. Only obtainable from
  * {@link ActorRef.connect}, so `subscribe`/`send` are always valid — the socket
  * exists for this object's whole lifetime.
  */
 class Connection {
-  private readonly ws: PartySocket;
+  private readonly ws: ReconnectingWebSocket;
   private readonly listeners = new Set<(data: unknown) => void>();
   private heartbeat: ReturnType<typeof setInterval> | null = null;
+  private closed = false;
   /** The client-chosen conn id — becomes _pk → the actor's conn.id. */
   readonly id: string;
 
@@ -45,22 +98,62 @@ class Connection {
   ) {
     this.id = options?.id ?? crypto.randomUUID();
 
-    const ws = new PartySocket({
-      host: config.host,
-      party: actorName,
-      room: instanceId,
-      id: this.id,
-      // Re-read on every (re)connect so a login/logout is picked up.
-      query: () => {
-        const token = config.getAuthToken();
-        return {
-          app_id: config.appId,
-          handler: actorName,
-          ...(token ? { token } : {}),
-          ...(config.functionsVersion ? { fv: config.functionsVersion } : {}),
-        };
-      },
-    });
+    // Direct-first with proxy fallback, decided per connection attempt. Once a
+    // mint answers with a fallback status the choice is sticky for this
+    // socket's lifetime (a fresh connect() after close() probes direct again,
+    // picking up actors migrated in the meantime). Any other mint failure
+    // rejects, which ReconnectingWebSocket retries with backoff — except the
+    // terminal statuses, which close this connection for good.
+    let useProxy = config.transport === "proxy";
+    const urlProvider = async (): Promise<string> => {
+      if (this.closed) throw new Error("Actor connection is closed");
+      if (!useProxy) {
+        try {
+          const { websocket_url, token } = await config.mintConnectionToken(
+            actorName,
+            instanceId,
+            this.id,
+          );
+          const sep = websocket_url.includes("?") ? "&" : "?";
+          return `${websocket_url}${sep}token=${encodeURIComponent(token)}`;
+        } catch (err) {
+          const status = mintErrorStatus(err);
+          const isFallback =
+            config.transport !== "direct" &&
+            status !== undefined &&
+            PROXY_FALLBACK_STATUSES.has(status);
+          if (!isFallback) {
+            if (status !== undefined && TERMINAL_MINT_STATUSES.has(status)) {
+              // close() before notifying: ws.close() stops the redial the
+              // rethrow below would otherwise schedule, and a handler that
+              // immediately calls connect() gets a clean new connection.
+              this.close();
+            }
+            // Reported from here because the socket's error event only
+            // preserves `err.message`, never `.status`.
+            try {
+              config.onMintError?.(toError(err));
+            } catch {
+              // an app handler must not break the dial loop or mask `err`
+            }
+            throw err;
+          }
+          useProxy = true;
+        }
+      }
+      // Rebuilt per attempt so a login/logout is picked up on reconnect.
+      return buildProxyActorUrl(
+        config.host,
+        actorName,
+        instanceId,
+        this.id,
+        config.appId,
+        config.getAuthToken(),
+        config.functionsVersion,
+      );
+    };
+
+    const ws = new ReconnectingWebSocket(urlProvider);
     this.ws = ws;
 
     let lastMsg = Date.now();
@@ -82,7 +175,10 @@ class Connection {
     this.heartbeat = setInterval(() => {
       if (Date.now() - lastMsg > DEAD_MS) {
         bumpAlive(); // avoid a reconnect storm while the new socket comes up
-        ws.reconnect();
+        // Only kick a half-open socket (OPEN but silent). When it isn't open
+        // the socket is already redialing with backoff, and reconnect() would
+        // reset that backoff into a mint call every DEAD_MS.
+        if (ws.readyState === ws.OPEN) ws.reconnect();
         return;
       }
       try {
@@ -103,10 +199,14 @@ class Connection {
   }
 
   send(data: unknown): void {
+    // after close() the socket would buffer forever (unbounded enqueue)
+    if (this.closed) return;
     this.ws.send(JSON.stringify(data));
   }
 
   close(): void {
+    if (this.closed) return;
+    this.closed = true;
     if (this.heartbeat) {
       clearInterval(this.heartbeat);
       this.heartbeat = null;
@@ -140,10 +240,45 @@ function makeActorRef(
 }
 
 /**
- * Absolute host for the actor WebSocket. PartySocket needs an absolute host and
- * can't resolve a relative/empty `serverUrl` (same-origin apps use a relative
- * `/api`, so `serverUrl` is often `""`), so fall back to the page origin.
- * PartySocket handles the scheme (https→wss, ws for localhost).
+ * The legacy platform-proxy URL, byte-for-byte what PartySocket built before
+ * the direct path existed: same scheme swap (including its localhost-needs-a-
+ * port quirk), case-preserved party segment, `_pk` first in the query. The
+ * `handler` param is load-bearing — the proxy reads it for the actor name.
+ */
+export function buildProxyActorUrl(
+  rawHost: string,
+  actorName: string,
+  instanceId: string,
+  connectionId: string,
+  appId: string,
+  token: string | null | undefined,
+  functionsVersion?: string,
+): string {
+  let host = rawHost.replace(/^(http|https|ws|wss):\/\//, "");
+  if (host.endsWith("/")) host = host.slice(0, -1);
+  const insecure =
+    host.startsWith("localhost:") ||
+    host.startsWith("127.0.0.1:") ||
+    host.startsWith("192.168.") ||
+    host.startsWith("10.") ||
+    (host.startsWith("172.") &&
+      host.split(".")[1] >= "16" &&
+      host.split(".")[1] <= "31") ||
+    host.startsWith("[::ffff:7f00:1]:");
+  const query = new URLSearchParams([
+    ["_pk", connectionId],
+    ["app_id", appId],
+    ["handler", actorName],
+  ]);
+  if (token) query.append("token", token);
+  if (functionsVersion) query.append("fv", functionsVersion);
+  return `${insecure ? "ws" : "wss"}://${host}/parties/${actorName}/${instanceId}?${query}`;
+}
+
+/**
+ * Absolute host for the proxy-fallback actor URL. A relative/empty `serverUrl`
+ * can't be dialed (same-origin apps use a relative `/api`, so `serverUrl` is
+ * often `""`), so fall back to the page origin.
  */
 export function resolveActorsHost(serverUrl: string, browserOrigin?: string): string {
   return serverUrl && !serverUrl.startsWith("/") ? serverUrl : browserOrigin ?? serverUrl;

src/modules/actors.types.ts

@@ -67,10 +67,15 @@ export interface Connection<N extends string = string> {
   /** Register a message listener. Multiple are allowed; returns a per-listener unsubscribe. */
   subscribe(callback: (data: ToClientFor<N>) => void): ActorSubscription;
 
-  /** Send a message. Buffered by the socket until it's open. */
+  /** Send a message. Buffered by the socket until it's open; dropped after
+   * {@link close}. */
   send(data: ToServerFor<N>): void;
 
-  /** Tear down the socket, heartbeat, and all listeners. */
+  /**
+   * Tear down the socket, heartbeat, and all listeners. Safe to call more
+   * than once. A connection also closes itself when it fails permanently —
+   * see {@link ActorRef.connect}.
+   */
   close(): void;
 }
 
@@ -79,7 +84,15 @@ export interface Connection<N extends string = string> {
  * {@link connect} to open the socket and get a {@link Connection}.
  */
 export interface ActorRef<N extends string = string> {
-  /** Open the WebSocket and return the {@link Connection}. Idempotent. */
+  /**
+   * Open the WebSocket and return the {@link Connection}. Idempotent while the
+   * connection is open.
+   *
+   * A connection that fails permanently (for example, the actor doesn't exist
+   * or the caller isn't allowed to connect) closes itself and reports the
+   * error to the client's `onError` handler. Call `connect()` again after
+   * fixing the cause to get a fresh {@link Connection}, and re-subscribe.
+   */
   connect(options?: ActorConnectOptions): Connection<N>;
 }