Ch. 5: Stream live clicks

Real-time delivery means pushing data to whoever is listening the moment it changes, like a live click counter on a dashboard. On iii that is what observing a function does: worker.observe(a, b) runs b each time a runs. A live click feed is a function that runs once per click, and anyone who wants to hear about clicks observes it, the same way you observed http::on-request and queue::on-message in earlier chapters. You’ll give the feed to a dedicated click-streamer worker so the link worker stays focused on links. A notification is not storage. The clicks already live in the clicks table from Chapter 3, so a listener that joins late reads the current total with a query, then counts the notifications that follow.

Earlier versions of this chapter used iii-stream and pubsub. If your project still has them, see Migrate from iii-stream and pubsub.

Add the worker

The click-streamer worker owns the live click feed. Uncomment the Ch. 5 block under containers:

worker-compose.yaml

click-streamer:
  worker: path://./click-streamer

Host the click feed in `click-streamer`

The feed is click-streamer::on-click, a function that runs once per recorded click and returns that click. What a function returns is what its observers receive, so every listener gets { id, code, clicked_at } for each click. click-streamer runs the feed each time link::record_click runs, by observing it. Create click-streamer/src/index.ts:

click-streamer/src/index.ts

import { ObserveRejection, registerWorker } from "iii-sdk";
import { Logger } from "@iii-dev/helpers/observability";

const worker = registerWorker(process.env.III_URL ?? "ws://localhost:49134", {
  workerName: "click-streamer",
});
const logger = new Logger();

type Click = { id: number; code: string; clicked_at: string };

// The live feed. Each call is one click, and the click it returns is what
// every listener receives.
const onClick = worker.registerFunction(
  "click-streamer::on-click",
  async (click: Click): Promise<Click> => ({
    id: click.id,
    code: click.code,
    clicked_at: click.clicked_at,
  }),
  {
    description: "Runs once per recorded click. Observers receive { id, code, clicked_at }.",
    // Only click-streamer calls the feed. Any worker may observe it.
    private: true,
  },
);

// iii accepts every observer by default. Cap the listeners: a throw rejects
// the new observer, and the worker that asked sees the error.
const MAX_LISTENERS = 256;
onClick.onObserverAdded(() => {
  if (onClick.observers().length >= MAX_LISTENERS) {
    throw new ObserveRejection(
      "TOO_MANY_LISTENERS",
      "click-streamer::on-click: too many listeners",
    );
  }
});

// Runs each time link::record_click runs, with the click it returned as `output`.
worker.registerFunction("click-streamer::announce", async ({ output: click }: { output: Click }) => {
  await worker.call({ function_id: "click-streamer::on-click", payload: click });
  return null;
});
worker.observe(
  { function_id: "link::record_click", capture: ["output"] },
  "click-streamer::announce",
);

logger.info("click-streamer ready");

Each part of that file has one job:

The feed returns the click. click-streamer::on-click builds the click from the three fields a listener needs and returns it. Its observers receive that value.
Only click-streamer calls the feed. private: true makes the engine accept calls of click-streamer::on-click from its own worker only, so no other worker can announce a click that never happened. Other workers can still observe it, which matters in Chapter 7, when a browser tab observes it.
The listeners are capped. onObserverAdded runs before iii accepts a new observer of the feed, and overrides iii’s default of accepting every one: past 256 listeners it throws, and the observer is rejected.
announce connects the feed to link. It observes link::record_click with capture: ["output"], so it runs after each successful call of link::record_click and receives what that call returned as the payload’s output. It then calls the feed with the click.

Return the click from `link`

click-streamer::announce receives what link::record_click returns, so have it return the click it recorded. database::execute returns the new row’s id as last_insert_id; it travels with the click so listeners can tell a new click from one they have already counted. Replace link::record_click in link/src/index.ts:

link/src/index.ts

worker.registerFunction(
  "link::record_click",
  async (payload: { code: string; clicked_at: string }) => {
    const { last_insert_id } = await worker.call<
      { db: string; sql: string; params: string[] },
      { last_insert_id: string | null }
    >({
      function_id: "database::execute",
      payload: {
        db: DB,
        sql: "INSERT INTO clicks (code, clicked_at) VALUES (?, ?)",
        params: [payload.code, payload.clicked_at],
      },
    });
    // The row is committed. Observers of link::record_click receive this click.
    return { id: Number(last_insert_id), code: payload.code, clicked_at: payload.clicked_at };
  },
);

The link worker stays focused on links: it records the click and returns it, and click-streamer reacts to that on its own.

A live counter is best effort. iii hands each click to the listeners without waiting for them, so the click write never waits on a slow listener, and a click a listener misses is not replayed. The listener’s next read of the total corrects it.

Read the current total

A listener that connects now has missed every earlier click, and notifications are never replayed. Give it a query to start from: add link::click_summary to link/src/index.ts. It returns the number of clicks and the highest click id, so a listener knows which notifications are already included in the total.

link/src/index.ts

worker.registerFunction("link::click_summary", async () => {
  const { rows } = await worker.call<
    { db: string; sql: string },
    { rows: Array<{ total: number; last_id: number | null }> }
  >({
    function_id: "database::query",
    payload: { db: DB, sql: "SELECT COUNT(*) AS total, MAX(id) AS last_id FROM clicks" },
  });
  return { total: rows[0]?.total ?? 0, last_id: rows[0]?.last_id ?? 0 };
});

Restart Compose so it picks up the change:

iii call compose::restart

The browser you build in Chapter 7 observes click-streamer::on-click and counts clicks live.

See it work

A listener is any worker that observes click-streamer::on-click. The scaffold ships a small one that prints each click, commented out in click-streamer/watch.ts: uncomment its Ch. 5 block (a fully commented file runs, prints nothing and exits). The code:

click-streamer/watch.ts

import { registerWorker } from "iii-sdk";

const worker = registerWorker(process.env.III_URL ?? "ws://127.0.0.1:49134", {
  workerName: "click-watcher",
  // The project's namespace, where click-streamer::on-click and link::* live.
  namespace: process.env.III_NAMESPACE || "default",
});

type Click = { id: number; code: string; clicked_at: string };

worker.registerFunction("click-watcher::print", async ({ output: click }: { output: Click }) => {
  console.log(`click #${click.id}: ${click.code} at ${click.clicked_at}`);
  return null;
});

// Observe first, then read the total, so no click falls between the two.
// capture: ["output"] hands click-watcher::print the click the feed returned, as `output`.
await worker
  .observe({ function_id: "click-streamer::on-click", capture: ["output"] }, "click-watcher::print")
  .ready();

const summary = await worker.call<Record<string, never>, { total: number; last_id: number }>({
  function_id: "link::click_summary",
  payload: {},
});
console.log(`watching clicks, ${summary.total} recorded so far`);

ready() resolves once the observer is active, so every click after that line reaches the watcher. Run it from the click-streamer folder and leave it running:

cd click-streamer && npm install && npx tsx watch.ts

In a second terminal, create a few links and follow each:

for n in $(seq 1 3); do
  curl -s -X POST http://127.0.0.1:3111/links \
    -H 'Content-Type: application/json' -d "{\"url\":\"https://iii.dev\",\"code\":\"stream-$n\"}"
  curl -s -o /dev/null "http://127.0.0.1:3111/s/stream-$n"
done

Running the loop again is fine: the links already exist, so POST /links answers 409 with {"error":"code `stream-1` is already taken"}, and each redirect still records a new click. Change the numbers in seq to create new links. Each redirect shows up in the watcher as soon as its click row is written:

watching clicks, 0 recorded so far
click #1: stream-1 at 2026-05-27T10:15:02.114Z
click #2: stream-2 at 2026-05-27T10:15:02.169Z
click #3: stream-3 at 2026-05-27T10:15:02.221Z

A listener that wants one link’s clicks adds a filter, and iii runs its target only for the clicks that match:

worker.observe(
  {
    function_id: "click-streamer::on-click",
    filters: { "input.code": { $eq: "stream-1" } },
    capture: ["output"],
  },
  "click-watcher::print",
);

The total is always one query away:

iii call link::click_summary

{ "total": 3, "last_id": 3 }

Stop the watcher with Ctrl+C. Its observer is suspended when the watcher disconnects, and iii removes it once the grace period (30 seconds by default) ends without a reconnect.

Conclusion

Linkly now pushes every click to its listeners in real time: click-streamer observes link::record_click and runs its click-streamer::on-click feed once per click, and every listener observes the feed. The clicks table stays the source of truth behind link::click_summary. Next, in Ch. 6: Move bulk data with channels, you bulk-load links from a CSV in a single streamed upload.

​Add the worker

​Host the click feed in click-streamer

​Return the click from link

​Read the current total

​See it work

​Conclusion

Add the worker

Host the click feed in `click-streamer`

Return the click from `link`

Read the current total

See it work

Conclusion