Ch. 5: Stream live clicks

Push clicks to listeners in real time from a dedicated click-streamer worker that owns a trigger type, and read the running total with a query.

Real-time delivery means pushing data to whoever is listening the moment it changes, like a live click counter on a dashboard. On iii there is no special channel for this: a worker declares its own trigger type, and anyone who wants to hear about clicks binds a function to it, the same way you bound functions to http and durable:subscriber in earlier chapters. You’ll give that job 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

Own a trigger type in click-streamer

A trigger type has a provider: the worker that registered it. When another worker binds a function to the type, the engine calls the provider’s registerTrigger with that binding, and calls unregisterTrigger when the binding goes away or its worker disconnects. The provider keeps its own table of bindings and invokes their functions when something happens.

Create click-streamer/src/index.ts:

click-streamer/src/index.ts

import { registerWorker, TriggerAction } from "iii-sdk";
import type { TriggerConfig } from "iii-sdk/trigger";
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 };
type ClickFilter = { code?: string };

// Every function bound to click-streamer::click, keyed by trigger id. The cap
// keeps a flood of listeners from growing it without limit.
const MAX_LISTENERS = 256;
const listeners = new Map<string, TriggerConfig<ClickFilter>>();

worker.registerTriggerType<ClickFilter>(
  {
    id: "click-streamer::click",
    description: "Fires after a click is recorded. Config: { code? }. Payload: { id, code, clicked_at }.",
  },
  {
    async registerTrigger(binding) {
      // Throwing rejects the binding, and the worker that asked sees the error.
      const code = binding.config?.code;
      if (code !== undefined && typeof code !== "string") {
        throw new Error("click-streamer::click: code must be a string");
      }
      if (!listeners.has(binding.id) && listeners.size >= MAX_LISTENERS) {
        throw new Error("click-streamer::click: too many listeners");
      }
      listeners.set(binding.id, binding);
    },
    async unregisterTrigger(binding) {
      // Also called when the listening worker disconnects.
      listeners.delete(binding.id);
    },
  },
);

worker.registerFunction("click-streamer::broadcast", async (click: Click) => {
  // Copy the fields: the engine stamps bookkeeping fields such as
  // _caller_worker_id on every delivery.
  const event: Click = { id: click.id, code: click.code, clicked_at: click.clicked_at };
  let delivered = 0;
  for (const binding of listeners.values()) {
    // A binding with a code, even "", gets only that code's clicks; one
    // without a code gets every click.
    if (binding.config?.code !== undefined && binding.config.code !== event.code) continue;
    try {
      await worker.trigger({
        function_id: binding.function_id,
        // Deliver where the listener registered, with its metadata.
        namespace: binding.namespace,
        metadata: binding.metadata,
        payload: event,
        action: TriggerAction.Void(),
      });
      delivered++;
    } catch (err) {
      logger.warn("click delivery failed", { trigger: binding.id, error: String(err) });
    }
  }
  return { delivered };
});

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

A few things in there are the provider’s job, because nobody else can do them:

• Validate the filter. A listener can pass { code } to hear about one link only, and leave it out to hear every click. A code that is present always filters, even "". Anything that is not a string is rejected at bind time instead of failing later.
• Bound the fan-out. The table has a cap, and each delivery uses TriggerAction.Void(), so a slow listener never holds up the others.
• Deliver to the listener’s namespace. Passing binding.namespace matters in Chapter 7, when a browser tab binds from its own namespace.

Announce clicks from link

Now have link::record_click hand each click to click-streamer::broadcast, after the row is written. 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.trigger<
      { 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. Announce it, without waiting on the listeners.
    worker.trigger({
      function_id: "click-streamer::broadcast",
      payload: { id: Number(last_insert_id), code: payload.code, clicked_at: payload.clicked_at },
      action: TriggerAction.Void(),
    });
    return { recorded: true };
  },
);

In the new code above we didn’t use await and set the action to TriggerAction.Void(). This causes the function to return immediately before the broadcast completes. A live counter is best effort: if a notification is lost, the listener’s next read of the total corrects it, so the click write never waits on the listeners.

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.trigger<
    { 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 trigger compose::restart

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

See it work

A listener is any worker that binds a function to the trigger type. 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::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 (click: Click) => {
  console.log(`click #${click.id}: ${click.code} at ${click.clicked_at}`);
  return null;
});

// Bind first, then read the total, so no click falls between the two.
worker.registerTrigger({
  type: "click-streamer::click",
  function_id: "click-watcher::print",
  config: {},
});

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

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

The total is always one query away:

iii trigger link::click_summary

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

Stop the watcher with Ctrl+C. Its binding goes away with it, and click-streamer drops it from its table.

Conclusion

Linkly now pushes every click to its listeners in real time through a dedicated click-streamer worker that owns the click-streamer::click trigger type, while 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.