> ## 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. 4: Make it durable

> Move click-row writes onto a queue, then broadcast link events to independent subscribers.

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

In Chapter 3 every redirect triggers `link::record_click` directly, so the database write runs on
the redirect's hot path and a slow write slows the redirect. In this chapter you move that work onto
a **queue** so redirects return immediately, then publish link events on **durable topics** that
independent subscribers receive: a Python analytics worker and a cache refresher, both decoupled
from the `link` worker.

## Add the workers

This chapter uses `queue` and a new Python `analytics` worker. Uncomment the Ch. 4 block in
`worker-compose.yaml`:

```yaml worker-compose.yaml theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
queue:
  worker: package://queue
  version: "0.21.11"
  config_name: queue
  working_dir: .
  config_override:
    queue_configs:
      clicks:
        type: standard
        max_retries: 5
        concurrency: 5
analytics:
  worker: path://./analytics
  start_after: [database, queue]
```

Also uncomment the `analytics` database in the Ch. 3 `database` block. `analytics` is a second
database: it goes under `databases:`, at the same indentation as `primary:`, not inside `primary:`
or its `pool:`. The whole block then reads:

```yaml worker-compose.yaml theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
database:
  worker: package://database
  version: "0.5.13"
  config_name: database
  working_dir: .
  config_override:
    databases:
      primary:
        pool:
          acquire_timeout_ms: 5000
          idle_timeout_ms: 30000
          max: 10
        url: sqlite:./data/iii.db
      analytics:
        url: sqlite:./data/analytics.db
```

If `database::query db=analytics` later answers `UNKNOWN_DB`, the `analytics` entry is nested at
the wrong level.

## Make redirects fast with a queue

A queue holds work that is accepted now and run later. The `clicks` queue you uncommented above is
the one `link::record_click` uses.

> [!NOTE]
>
> Queue names are references to the queue and do not place any restrictions on what can be put into
> a given queue.

You already wrote `link::record_click` in Chapter 3, where `http::redirect` triggers it directly.
Nothing about that function needs to change to make it queuable. You only change how it's triggered
with a `TriggerAction`.

**First import `TriggerAction`** into `link/src/index.ts`:

```typescript {1} src/index.ts theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
import { InvocationError, registerWorker, TriggerAction } from "iii-sdk";
import { Logger } from "@iii-dev/helpers/observability";
```

Then add an `action` to the existing `link::record_click` call in `http::redirect` so the `queue`
worker enqueues it instead of running it inline:

```typescript src/index.ts {6} theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
worker.registerFunction("http::redirect", async (req) => {
  // ...unchanged from the Chapter 3 http::redirect: the code, link::resolve and the 404...
  await worker.trigger({
    function_id: "link::record_click",
    payload: { code, clicked_at: new Date().toISOString() },
    action: TriggerAction.Enqueue({ queue: "clicks" }),
  });
  return { status_code: 302, headers: { Location: url } };
});
```

The redirect now returns as soon as the click is accepted onto the queue. `link::record_click`
drains the queue in the background, with retries and a dead-letter queue if a write keeps failing.

## Broadcast events with durable topics

A queue delivers each message to one consumer. When several unrelated parts of the system need to
react to the same event, publish it on a **topic** instead. The `queue` worker serves topics too:
`iii::durable::publish` sends a message to a topic, and every function bound to that topic with a
`durable:subscriber` trigger receives its own copy.

> [!NOTE]
>
> Topic messages are durable: each subscriber's copy is kept until it is handled, retried when the
> handler fails, and moved to a dead-letter queue when it keeps failing. Delivery is at least once,
> so make subscribers safe to run twice for the same message.
>
> For notifications that nobody needs to replay, such as a live click counter, a worker can declare
> its own trigger type instead. You'll build one in Chapter 5.

Here we'll publish an event when a link is created, and another when a link is updated. Later
we'll make an `analytics` worker that collects data from these events. We don't have link updating
functionality yet, so we'll add that and an HTTP endpoint for it too.

### Publish on link.created

Publish an event whenever a link is created. Replace `link::create` with this version: it keeps the
creation time in a `created_at` variable, writes it with the row, and after `state::set` publishes
it on the `link.created` topic along with the code and URL. A subscriber then knows when the link
was created, even when its copy arrives later. A taken code throws before the publish, so it
publishes nothing:

```typescript {4-6,17,25-28} src/index.ts theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
worker.registerFunction("link::create", async (payload: { url: string; code?: string }) => {
  const code = payload.code ?? makeCode();
  const url = /^https?:\/\//i.test(payload.url) ? payload.url : `https://${payload.url}`;
  // One timestamp for the row and the event, so subscribers see when the link
  // was created, not when their copy arrived.
  const created_at = new Date().toISOString();
  // ON CONFLICT DO NOTHING turns a taken code into zero inserted rows instead
  // of a database error, so link::create can say so in its own words.
  const { affected_rows } = await worker.trigger<
    { db: string; sql: string; params: string[] },
    { affected_rows: number }
  >({
    function_id: "database::execute",
    payload: {
      db: DB,
      sql: "INSERT INTO links (code, url, created_at) VALUES (?, ?, ?) ON CONFLICT(code) DO NOTHING",
      params: [code, url, created_at],
    },
  });
  if (affected_rows === 0) throw new Error(`code \`${code}\` is already taken`);
  await worker.trigger({
    function_id: "state::set",
    payload: { scope: "links", key: code, value: { url } },
  });
  await worker.trigger({
    function_id: "iii::durable::publish",
    payload: { topic: "link.created", data: { code, url, created_at } },
  });
  logger.info("link created", { code, url });
  return { code, url };
});
```

### Publish on link.updated

Now add an update path to the `link` worker so a link's target can change, and announce it.

First, the domain function: it updates the database row and publishes a `link.updated` event on its
own topic, the same way. Place it at the end of `link/src/index.ts`:

```typescript src/index.ts theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
worker.registerFunction("link::update", async (payload: { code: string; url: string }) => {
  const url = /^https?:\/\//i.test(payload.url) ? payload.url : `https://${payload.url}`;
  await worker.trigger({
    function_id: "database::execute",
    payload: {
      db: DB,
      sql: "UPDATE links SET url = ? WHERE code = ?",
      params: [url, payload.code],
    },
  });
  await worker.trigger({
    function_id: "iii::durable::publish",
    payload: { topic: "link.updated", data: { code: payload.code, url } },
  });
  return { code: payload.code, url };
});
```

### Expose link updating via HTTP

Then connect `link::update` with an HTTP handler that validates input and calls the domain function:

```typescript src/index.ts theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
worker.registerFunction("http::update", async (req) => {
  const code = req.path_params.code;
  const url = req.body?.url;
  if (!url) {
    return {
      status_code: 400,
      body: { error: 'missing "url"' },
      headers: { "Content-Type": "application/json" },
    };
  }
  const link = await worker.trigger<{ code: string; url: string }, { code: string; url: string }>({
    function_id: "link::update",
    payload: { code, url },
  });
  return { status_code: 200, body: link, headers: { "Content-Type": "application/json" } };
});
```

And finally the trigger that binds `http::update` to `PUT /links/:code`:

```typescript src/index.ts theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
worker.registerTrigger({
  type: "http",
  function_id: "http::update",
  config: { api_path: "/links/:code", http_method: "PUT" },
});
```

This last bit of code isn't about topics, but you'll need it for trying out your new feature at the
end of the chapter.

## Add reactive state: Keep the cache correct without coupling

`link::update` changes the database but not the state cache, so a query could serve stale data.
Rather than handle refreshing the state cache inside `link::update`, subscribe to the `link.updated`
topic with a `durable:subscriber` trigger:

> [!NOTE]
>
> **Why durable matters here.** This cache refresher must receive every update: a dropped event
> would leave the cache pointing at a stale URL. A durable topic keeps the message until the
> subscriber handles it, and retries it when the handler fails. Use a durable topic whenever a
> missed event would corrupt state.

```typescript src/index.ts theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
worker.registerFunction("link::on_link_updated", async (data: { code: string; url: string }) => {
  await worker.trigger({
    function_id: "state::set",
    payload: { scope: "links", key: data.code, value: { url: data.url } },
  });
});

worker.registerTrigger({
  type: "durable:subscriber",
  function_id: "link::on_link_updated",
  config: { topic: "link.updated" },
});
```

## Create an analytics worker in Python

Queues and events are useful within a single worker but also between workers. Thus far we've been
writing all of our code in TypeScript. However workers are not restricted to specific languages or
runtimes. So this time we'll create an analytics worker in Python to count links.

### Adding a second worker

We already referenced the analytics worker in `worker-compose.yaml`, now let's give it some code.

The `analytics` directory holds a Python worker, with an `analytics/src/main.py` entrypoint and an
`iii.worker.yaml` manifest of its own:

```yaml analytics/iii.worker.yaml theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
name: analytics
runtime:
  # Base OCI image used as the worker rootfs when virtualized.
  base_image: docker.io/iiidev/python:latest
scripts:
  install: python3 -m venv .venv && .venv/bin/pip install -r requirements.txt
  # src/main.py ships fully commented. Python runs a comment-only file, prints
  # nothing and exits 0, so the worker would vanish without a word: refuse to
  # start until the Ch. 4 block is uncommented. -u keeps the startup line
  # visible in the Compose logs instead of buffered.
  start: >-
    grep -q '^[^#[:space:]]' src/main.py
    || { echo 'analytics: src/main.py is still fully commented; uncomment its Ch. 4 block, then restart the analytics container' >&2; exit 1; };
    exec .venv/bin/python -u src/main.py
```

### Subscribe to link.created events

The scaffold ships `analytics/src/main.py` with this code commented out: uncomment its Ch. 4 block
(or replace the file with the code below). It binds a `durable:subscriber` trigger to the
`link.created` topic and records every new short link under the UTC day of the event's
`created_at`.

Delivery is at least once, so the same event can arrive twice. Instead of adding 1 to a counter,
the worker keeps one row per link in `link_events`, keyed by `code`, and writes it with a single
`INSERT ... ON CONFLICT(code) DO NOTHING`: a redelivered event finds its row and changes nothing.
The per-day counts come from `daily_link_counts`, a view that counts those rows by day:

> [accordion] analytics/src/main.py
>
> ```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> import os
> import time
> from datetime import datetime, timezone
>
> from iii import register_worker, InitOptions
> from iii_helpers.observability import Logger
>
> worker = register_worker(
>     os.environ.get("III_URL", "ws://localhost:49134"),
>     InitOptions(worker_name="analytics"),
> )
> logger = Logger()
>
> DB = "analytics"
>
> SCHEMA = [
>     # One row per link. Delivery is at least once, so a redelivered event
>     # finds its row already there and changes nothing.
>     "CREATE TABLE IF NOT EXISTS link_events (code TEXT PRIMARY KEY, day TEXT NOT NULL)",
>     # The per-day counts are computed from those rows, never stored.
>     "CREATE VIEW IF NOT EXISTS daily_link_counts AS "
>     "SELECT day, COUNT(*) AS count FROM link_events GROUP BY day",
> ]
>
> def ensure_schema() -> None:
>     """The analytics worker owns its own table, in its own database.
>
>     The database worker may register a moment after analytics, so retry until it
>     answers instead of crashing on the first call.
>     """
>     for attempt in range(1, 31):
>         try:
>             for sql in SCHEMA:
>                 worker.trigger(
>                     {"function_id": "database::execute", "payload": {"db": DB, "sql": sql}}
>                 )
>             return
>         except Exception:
>             if attempt >= 30:
>                 raise
>             time.sleep(1)
>
> def on_link_created(data: dict) -> dict:
>     """Runs whenever link publishes `link.created`. Records the link's day."""
>     # The day the link was created (UTC), not the day this copy arrived.
>     created_at = datetime.fromisoformat(data["created_at"].replace("Z", "+00:00"))
>     day = created_at.astimezone(timezone.utc).strftime("%Y-%m-%d")
>     worker.trigger(
>         {
>             "function_id": "database::execute",
>             "payload": {
>                 "db": DB,
>                 "sql": "INSERT INTO link_events (code, day) VALUES (?, ?) "
>                 "ON CONFLICT(code) DO NOTHING",
>                 "params": [data["code"], day],
>             },
>         }
>     )
>     logger.info(f"recorded new link {data['code']} for {day}")
>     return {"ok": True}
>
> ensure_schema()
>
> worker.register_function("analytics::on_link_created", on_link_created)
> worker.register_trigger(
>     {
>         "type": "durable:subscriber",
>         "function_id": "analytics::on_link_created",
>         "config": {"topic": "link.created"},
>     }
> )
>
> print("Analytics worker started")
> ```

Analytics keeps its rows in its own database, so the `link` worker never has to know it exists.
That is the `analytics` entry you uncommented at the start of the chapter, alongside the `primary`
one from Chapter 3.

Restart Compose so it picks up the change:

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

### See it work

Create five links, follow one a few times, and change its target:

```bash theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
# Make some new links
for n in $(seq 1 5); do
  curl -s -X POST http://127.0.0.1:3111/links \
    -H 'Content-Type: application/json' -d "{\"url\":\"https://iii.dev/$n\",\"code\":\"analyticslink$n\"}"
done
```

The view counts each new link once, under the day it was created:

```bash theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
iii trigger database::query db=analytics sql="SELECT day, count FROM daily_link_counts"
```

```json theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
{ "rows": [{ "day": "2026-05-27", "count": 5 }], "row_count": 1 }
```

Now follow one of them a few times (each click goes through the queue instead of a blocking write),
then change its target. `PUT /links/:code` publishes `link.updated` durably and the subscriber refreshes
the cache, so `link::resolve` returns the new URL right away:

```bash theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
for _ in 1 2 3; do curl -s -o /dev/null http://127.0.0.1:3111/s/analyticslink1; done
curl -s -X PUT http://127.0.0.1:3111/links/analyticslink1 \
  -H 'Content-Type: application/json' -d '{"url":"https://iii.dev/updated"}'
iii trigger link::resolve code=analyticslink1
```

```json theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
{ "url": "https://iii.dev/updated" }
```

## Conclusion

Redirects no longer wait on a database write: click rows ride a queue, drained in the background.
Link events fan out over durable topics to a Python analytics counter and a cache refresher, and the
`link` worker does not know either of them exists. Next, in
[Ch. 5: Stream live clicks](https://iii.dev/docs/tutorials/linkly/streaming.md), you broadcast every click in real time
from a dedicated `click-streamer` worker.


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