Realtime

Send small live messages from your functions to the browser while a run goes on. The usual use is progress for a long job: the page shows each step as it happens.

The code is in examples/ts/realtime/ and examples/go/realtime/main.go.

How it works

your function ──publish──────────────────► engine ──WebSocket──► browser
your server   ──"a token for this user"──► engine
browser       ──connect with the token───► engine

• A channel is a name you pick, like deployment:dep_123. Channels belong to your app: another app's channel with the same name is another channel.
• A topic is a name inside a channel, like progress or logs.
• A token lets one browser read some topics of one channel. Your server asks the engine for it, after your own check that the user may see that channel. The engine does not know your users.

Publish from a function

TypeScript

const buildSite = flow.createFunction(
  { id: "build-site" },
  { event: "site/build-requested" },
  async ({ event, step }) => {
    const channel = `deployment:${event.data.deploymentId}`;

    // A failed publish must not fail the step: log it and go on.
    const progress = (pct: number, label: string) =>
      flow.realtime
        .publish({ channel, topic: "progress", data: { pct, label } })
        .catch((err) => console.warn("realtime publish failed", err));

    await step.run("render", async () => {
      await progress(10, "Rendering");
      return render(event.data.deploymentId);
    });
    await step.run("upload", async () => {
      await progress(60, "Uploading");
      return upload(event.data.deploymentId);
    });

    // After the last step, this runs only on the run's last call. So it is sent
    // once, unless that call fails and the engine tries it again.
    await progress(100, "Done");
  },
);

Go

// Build is the data of a site/build-requested event.
type Build struct {
	DeploymentID string `json:"deploymentId"`
}

// Progress is what the browser gets.
type Progress struct {
	Pct   int    `json:"pct"`
	Label string `json:"label"`
}

func addBuildSite(client *flow.Client) error {
	_, err := flow.CreateFunction(client,
		flow.FunctionOpts{ID: "build-site"},
		flow.EventTrigger("site/build-requested", nil),
		func(ctx context.Context, in flow.Input[Build]) (any, error) {
			id := in.Event.Data.DeploymentID
			channel := "deployment:" + id

			// A failed publish must not fail the step: log it and go on.
			progress := func(ctx context.Context, pct int, label string) {
				if err := client.Realtime().Publish(ctx, channel, "progress", Progress{pct, label}); err != nil {
					log.Printf("realtime publish failed: %v", err)
				}
			}

			if _, err := step.Run(ctx, "render", func(ctx context.Context) (bool, error) {
				progress(ctx, 10, "Rendering")
				return true, render(ctx, id)
			}); err != nil {
				return nil, err
			}
			if _, err := step.Run(ctx, "upload", func(ctx context.Context) (bool, error) {
				progress(ctx, 60, "Uploading")
				return true, upload(ctx, id)
			}); err != nil {
				return nil, err
			}

			// After the last step, this runs only on the run's last call. So it is
			// sent once, unless that call fails and the engine tries it again.
			progress(ctx, 100, "Done")
			return nil, nil
		},
	)
	return err
}

Publishing uses the app's signing key (FLOW_SIGNING_KEY, then FLOW_SIGNING_KEY_FALLBACK while you rotate). On Workers, pass the handler's env: flow.realtime.publish(message, { env }).

Give the browser a token

Add a route to your own server. Check the user first, then ask the engine:

TypeScript

// Your own route. Check that this user may see this deployment first: the engine
// does not know your users.
app.get("/api/deployments/:id/realtime-token", async (req, res) => {
  if (!(await userMaySee(req.get("Authorization"), req.params.id))) {
    res.status(403).json({ error: "forbidden" });
    return;
  }
  const token = await flow.realtime.token({
    channel: `deployment:${req.params.id}`,
    topics: ["progress"],
  });
  res.json(token);
});

Go

// realtimeToken is your own route. Check that this user may see this deployment
// first: the engine does not know your users.
func realtimeToken(client *flow.Client) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		id := r.PathValue("id")
		if !userMaySee(r, id) {
			http.Error(w, `{"error":"forbidden"}`, http.StatusForbidden)
			return
		}
		token, err := client.Realtime().Token(r.Context(), "deployment:"+id, "progress")
		if err != nil {
			http.Error(w, err.Error(), http.StatusBadGateway)
			return
		}
		w.Header().Set("Content-Type", "application/json")
		_ = json.NewEncoder(w).Encode(token)
	}
}

The token looks like this (shortened):

{
  "token": "eyJ2IjoyLCJ3cyI6...",
  "url": "ws://127.0.0.1:8787/v1/realtime/connect",
  "channel": "deployment:dep_42",
  "topics": ["progress"],
  "expiresAt": 1793042255886
}

url is the engine's WebSocket address, so the browser connects to the right engine with no change in your code.

Subscribe in the browser

Without React, with subscribe from @clevix/flow/realtime. It has no server code in it.

const stop = subscribe<{ pct: number; label: string }>({
  // Called before every connect: ask your own route for a new token.
  refreshToken: async () => {
    const res = await fetch(`${base}/api/deployments/${deploymentId}/realtime-token`, {
      headers: { Authorization: "Bearer your-session" },
    });
    if (!res.ok) throw new Error(`token route answered ${res.status}`);
    return (await res.json()) as RealtimeToken;
  },
  onMessage: (m) => {
    console.log(`${m.topic}: ${m.data.pct}% ${m.data.label}`);
    if (m.data.pct === 100) stop();
  },
  onState: (state) => console.log("state:", state), // connecting, open, closed, error
});

With React, with the useRealtime hook from @clevix/flow/react:

type Progress = { pct: number; label: string };

export function DeployProgress({ deploymentId, building }: { deploymentId: string; building: boolean }) {
  const { latest, state, error } = useRealtime<Progress>({
    channel: `deployment:${deploymentId}`, // a new channel drops old messages and connects again
    enabled: building, // false closes the socket
    refreshToken: async () => {
      const res = await fetch(`/api/deployments/${deploymentId}/realtime-token`);
      if (!res.ok) throw new Error(`token route answered ${res.status}`);
      return (await res.json()) as RealtimeToken;
    },
  });

  if (error) return <p>Live updates are off: {error.message}</p>;
  if (!latest) return <p>{state === "open" ? "Waiting for the build..." : "Connecting..."}</p>;
  return (
    <p>
      {latest.data.label}: {latest.data.pct}%
    </p>
  );
}

We ran the server and browser.ts (Node has a global WebSocket) against a local engine, with both the TypeScript and the Go server. Both printed:

state: connecting
state: open
progress: 10% Rendering
progress: 60% Uploading
progress: 100% Done

• The client asks your route for a new token before every connect.
• After a dropped connection it connects again, waiting 1 s, then longer, up to 30 s. It sends a ping every 30 s.
• useRealtime keeps the last 100 messages (bufferSize), and closes the socket when enabled is false or the component goes away. When channel changes, it drops its messages and connects again.
• If refreshToken gives back something that is not a token, the client reports an error state and tries again later.

Subscribe from a server

A server can follow channels too, for example to pass events between your own servers. flow.realtime.subscribe makes its own token (with the signing key) before every connect, so you need no token route:

// It makes its own token before every connect, so you need no token route.
const stop = flow.realtime.subscribe<{ pct: number; label: string }>({
  channels: ["deployment:dep_1", "deployment:dep_2"], // or channel: "deployment:dep_1"
  topics: ["progress"],
  onMessage: (m) => console.log(`${m.channel} ${m.data.pct}% ${m.data.label}`),
  onState: (state) => {
    // Messages sent while it was not open are lost: read them from your own store.
    if (state === "open") void catchUp();
  },
});

• It connects to the engine at baseUrl (FLOW_BASE_URL), not at the host in the token's url, which can be wrong behind a proxy or in wrangler dev.
• It needs a WebSocket: Node 22 and later have one. Elsewhere, pass one as WebSocket.
• It throws at once with no channel or no topic. It reconnects the same way as the browser client, and stops when you call the function it gave back.
• Messages can be lost. Nothing is stored, so a message sent while the socket reconnects (1 s, then longer, up to 30 s), or at the 1-hour renewal, never arrives. Use realtime as a hint, and read what you missed from your own store, for example each time onState says open.

Many channels on one socket

A token can be for 1 to 100 channels: pass channels instead of channel to flow.realtime.token or flow.realtime.subscribe. The topics apply to every channel, and each message still has its channel. A server that follows many channels then holds one socket, not one per channel. The browser subscribe takes such a token as it is.

Publish many at once

flow.realtime.publishMany sends 1 to 100 messages in one call. The engine checks them all first: one bad message fails the call, and none is sent.

await flow.realtime.publishMany([
  { channel: "deployment:dep_1", topic: "progress", data: { pct: 50, label: "Half" } },
  { channel: "deployment:dep_2", topic: "progress", data: { pct: 10, label: "Start" } },
]);

What to expect

• Messages are hints. A message goes only to browsers connected right now. Nothing is stored; a browser that connects later does not get older messages. Keep the real state in your app, and let the page read it (on load, for example).
• A publish can be sent more than once. publish is a plain call, not a step. Inside a step, it is sent again when that step is retried. Outside a step, it is sent again on every call: a publish before three steps goes out four times. Put it inside a step, or after the last step. After the last step, it runs only on the run's last call: once, unless that call fails and the engine tries it again.
• No secrets in messages. Anyone with a token for the channel sees every message on its topics.
• Do not let a publish fail your run. In Go, log the error and go on. In TypeScript, publish throws: catch it, as the example does.

Limits

What	Limit
One publish (the whole body)	64 KB
publishMany	1 to 100 messages, the body up to 1 MB, each data up to 64 KB
Channels per token	1 to 100 (channels), up to 8,000 characters of names in all
Channel name	1-256 characters: letters, digits, : _ - .
Topic name	1-64 characters: letters, digits, _ - .
Topics per token	1 to 10
Token	must be used to connect within 1 minute
A connection	up to 1 hour; then the client gets a new token and connects again
Browsers on one channel	1,000
Open sockets per token	1

The engine must have realtime turned on (REALTIME_TOKEN_SECRET). Without it, tokens fail with 503 realtime is not set up on this engine.