> ## Documentation Index
> Fetch the complete documentation index at: https://iii.dev/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# 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.

> [!NOTE]
>
> Earlier versions of this chapter used `iii-stream` and `pubsub`. If your project still has them,
> see [Migrate from iii-stream and pubsub](https://iii.dev/docs/0-25-0/upgrading/migrate-from-streams.md).

## Add the worker

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

```yaml worker-compose.yaml theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
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`:

```typescript click-streamer/src/index.ts theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
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`:

```typescript link/src/index.ts {4-7,15-20} theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
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 };
  },
);
```

> [!NOTE]
>
> 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.

```typescript link/src/index.ts theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
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:

```bash theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
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:

```typescript click-streamer/watch.ts theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
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:

```bash theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
cd click-streamer && npm install && npx tsx watch.ts
```

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

```bash theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
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:

```text theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
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:

```bash theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
iii trigger link::click_summary
```

```json theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
{ "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](https://iii.dev/docs/tutorials/linkly/channels.md), you bulk-load links from a CSV in
a single streamed upload.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.