# pubsub

> pubsub is deprecated (pubsub) and will be removed in an upcoming release. Behavior is unchanged for now. Migration guide: https://github.com/iii-hq/iii/blob/6901a589dd39d9c34f9d5b6bd8bc82170ee21548/docs/next/upgrading/migrate-from-streams.mdx — Topic-based publish/subscribe messaging; registers the `subscribe` trigger type and the `publish` function.

| field | value |
|-------|-------|
| version | 0.21.12 |
| type | binary |
| license | Apache-2.0 |
| repo | https://github.com/iii-hq/workers |
| supported_targets | x86_64-apple-darwin, aarch64-apple-darwin, x86_64-pc-windows-msvc, aarch64-pc-windows-msvc, x86_64-unknown-linux-gnu, aarch64-unknown-linux-gnu, x86_64-unknown-linux-musl, armv7-unknown-linux-gnueabihf |
| author | iii |

## installation

```sh
iii trigger compose::add worker=pubsub@0.21.12
```

## configuration

```yaml
- adapter:
    name: local
```

## dependencies

- `configuration` @ `latest`

## readme

# pubsub

> **Deprecated:** pubsub is deprecated (pubsub) and will be removed in an upcoming release. Behavior is unchanged for now. Migration guide: https://iii.dev/docs/upgrading/migrate-from-streams
>
> This covers the `publish` function, the `subscribe` trigger type and the
> `pubsub` configuration entry. For new code, use a worker-owned trigger type
> for fire-and-forget fan-out, or the `queue` worker (`iii::durable::publish` +
> `durable:subscriber`) when delivery must not be lost. While deprecated, this
> worker logs a warning (target `iii::deprecation`) once at startup, on
> `publish` calls and on `subscribe` registrations (not per delivered event),
> at most once every 10 minutes per entry point and caller. Payloads and topic
> names are never logged. Traffic on topic `stream.events` (the engine's
> internal iii-stream bridge) does not warn.

Topic-based publish/subscribe messaging. Registers the `subscribe` trigger
type and the `publish` service function. This standalone worker replaces the
legacy built-in pub/sub service. Any function bound to a `subscribe` trigger on a topic receives
every event published to that topic; `publish` fans an event's `data` out to
all of a topic's subscribers as fire-and-forget invocations.

## Install

```bash
iii trigger compose::add worker=pubsub
```

`iii trigger compose::add` resolves the worker and its dependencies, writes
exact declarations to `worker-compose.yaml`, and reconciles the Compose project.

## Quickstart

Bind a function to the `subscribe` trigger type with a `topic`, then call the
`publish` function with `{ topic, data }`:

```rust
use iii_sdk::protocol::{RegisterTriggerInput, TriggerRequest};
use iii_sdk::{InitOptions, RegisterFunction, errors::Error, register_worker};
use serde_json::{json, Value};

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let iii = register_worker("ws://localhost:49134", InitOptions::default());

    iii.register_function(
        "orders::on_created",
        RegisterFunction::new_async(|data: Value| async move {
            println!("order event: {data}");
            Ok::<Value, Error>(Value::Null)
        }),
    );

    iii.register_trigger(RegisterTriggerInput {
        trigger_type: iii_pubsub::TRIGGER_TYPE.to_string(), // "subscribe"
        function_id: "orders::on_created".into(),
        config: json!({ "topic": "orders" }),
        metadata: None,
    })?;

    // Any caller (function, worker, or the stream bridge) publishes with:
    iii.trigger(TriggerRequest {
        function_id: "publish".into(),
        payload: json!({ "topic": "orders", "data": { "id": 42 } }),
        action: None,
        timeout_ms: Some(5000),
    })
    .await?;

    tokio::signal::ctrl_c().await?;
    Ok(())
}
```

The subscriber receives the **raw** `data` value (no envelope). Delivery is
fire-and-forget: `publish` returns a null result once dispatched and does not
wait for or report subscriber outcomes. Publishing to a topic with no
subscribers is a silent no-op; publishing with an empty topic fails with a
`topic_not_set` error.

## Trigger type

This worker always registers the `subscribe` trigger type. Bind a function to
it with:

| Field | Required | Default | Description |
|---|---|---|---|
| `topic` | yes | — | Topic to subscribe to. A missing or empty topic registers nothing (a warning is logged). |
| `condition_function_id` | no | — | Accepted for schema parity with the builtin; **never evaluated**. |

## The `publish` function

Registered under the bare id **`publish`** (matching the builtin exactly, so
existing callers and the engine's stream bridge keep working). Input:

| Field | Description |
|---|---|
| `topic` | Topic to publish to. Empty → `topic_not_set` error. |
| `data` | JSON payload delivered verbatim to each subscriber. |

Returns a null result on success.

## Configuration

| Field | Default | Description |
|---|---|---|
| `adapter.name` | `local` | Backend: `local` (in-process broadcast) or `redis` (cross-instance Redis Pub/Sub). |
| `adapter.config.redis_url` | `redis://localhost:6379` | Redis connection URL (only for the `redis` adapter). |

Configuration is owned by the `configuration` worker — edit it from the
console (**Configuration → Workers → pubsub**) or seed it once via
`--config <file>.yaml` on first boot. The adapter **hot-swaps at runtime**: a
change rebuilds the backend, re-subscribes the live subscriptions onto it
*before* the swap (no delivery gap — a brief double-delivery overlap is
accepted for fire-and-forget pub/sub), then tears down the previous backend.
A build failure keeps the previous backend and config.

### Requires removing the legacy built-in pub/sub service

The legacy built-in pub/sub service also owns the `subscribe` trigger type and
the `publish` function. Two owners of the same trigger type on one engine
collide — whichever registers last wins — so this worker requires it
to be absent: omit it from the engine's `config.yaml` (a config that doesn't
list a worker won't run it).

On boot, this worker queries the engine for connected workers and refuses to
start with a clear error if the legacy built-in is still active, so a stale config
fails loudly instead of silently racing the built-in worker for ownership of
`subscribe`/`publish`.

## Parity vs builtin

| Behavior | Builtin | This worker |
|---|---|---|
| Trigger type | `subscribe` | same |
| Trigger config | `{topic}` (+ advertised `condition_function_id`, never evaluated) | same (still never evaluated) |
| Service functions | `publish` (bare id — the `#[service]` prefix is discarded by the macro) | `publish` (exact same id) |
| `publish` input | `{topic, data}` | same |
| `publish` success result | null | same |
| `publish` empty topic | error `topic_not_set` / "Topic is not set" | same (code prefixes the message: `topic_not_set: Topic is not set`) |
| Delivery payload | raw `data`, no envelope | same |
| Delivery semantics | fire-and-forget, results ignored, no retries | same |
| No subscribers | silent no-op | same |
| Subscribe with missing/empty topic | warn + register nothing (Ok) | same |
| Backends | local (in-process), redis | same |
| Redis: subscriptions per topic per instance | one (second warns, dropped) | same |
| Local unsubscribe | drops the ENTIRE topic entry (bug: kills co-subscribers) | removes only the given id (**deliberate fix**) |
| Adapter hot-swap | gated build-first, resubscribe-before-swap | same |
| Config entry id | legacy engine configuration | `pubsub` (new entry, seeded on first boot) |
| Config schema shape | hand-built `oneOf` union (local/redis) | plain derive — accepted values identical; explicit Console form |

## skills

This version ships 1 skill file for agents, also browsable on the worker page:

- [`SKILL.md`](https://workers.iii.dev/workers/pubsub?tab=skills&skill=SKILL.md&version=0.21.12)

### SKILL.md

# pubsub

> **Deprecated:** pubsub is deprecated (pubsub) and will be removed in an upcoming release. Behavior is unchanged for now. Migration guide: https://iii.dev/docs/upgrading/migrate-from-streams
>
> Applies to `publish`, the `subscribe` trigger type and the `pubsub` configuration entry. Do not start new work on it: use a worker-owned trigger type for fire-and-forget fan-out, or the `queue` worker (`iii::durable::publish` + `durable:subscriber`) when every event must be processed. The worker logs rate-limited deprecation warnings (target `iii::deprecation`); topic `stream.events`, used by the engine's iii-stream bridge, is exempt.

The `pubsub` worker is topic-based publish/subscribe messaging. Publish an event to a named topic with the `publish` function and every registered `subscribe` trigger whose `topic` matches is invoked with the raw payload — no envelope, no persistence, no retries. It is fire-and-forget broadcast: subscribers receive each event as it arrives, and a subscriber that is offline simply misses it.

Install it with `iii trigger compose::add worker=pubsub`. It replaces the former in-engine pub/sub service: the trigger type (`subscribe`) and the function id (`publish`) are unchanged, so existing application code keeps working — only the owner moved from an in-process builtin to this standalone worker.

The worker exposes one callable function (`publish`, registered with the bare function id `"publish"` — no namespace prefix) and one trigger type (`subscribe`). Two adapters: `local` (default; in-memory broadcast; only delivers to subscribers in this engine process; no external dependency) and `redis` (`redis_url: ${REDIS_URL:redis://localhost:6379}`; uses Redis Pub/Sub so events propagate across multiple engine instances). The adapter is set in the `pubsub` configuration entry and hot-swaps at runtime.

## When to Use

- Real-time notifications consumers may miss without consequence — UI live updates, ephemeral signals, telemetry mirroring.
- Loose coupling where an event source should not know about its consumers.
- Cross-instance fan-out of ephemeral events (configure the `redis` adapter).

## Boundaries

- No persistence, ordering, or retries — a subscriber that is down misses events permanently. Use the `queue` worker (topic mode, `durable:subscriber`) when every consumer must process every event.
- Topic matching is exact string equality; there are no wildcards or hierarchical patterns. Register one trigger per topic.
- An empty topic is rejected: `publish` returns `topic_not_set`, and a `subscribe` trigger registered with an empty topic never fires.
- For key/value or stream change reactivity, use the `state` or `stream` worker instead — `pubsub` carries discrete events, not state.

## Functions

- `publish` — broadcast an event to a topic; every `subscribe` trigger on that exact topic receives the raw `data`. Empty topic returns `topic_not_set`.

## Reactive triggers

Bind a `subscribe` trigger when a function should run every time `publish` broadcasts to a configured topic. The handler receives the raw `data` value from the publish call directly — no envelope. Multiple triggers can subscribe to the same topic; each gets an independent copy (true fan-out).

Reach for it when:

- A domain event from one part of the system should kick off side effects elsewhere without the publisher knowing the consumers.
- You want cross-instance fan-out via the `redis` adapter.

If subscribers must reliably process every event with retries and dead-letter handling, use the `queue` worker's `durable:subscriber` instead — `subscribe` is fire-and-forget.

### How to bind

1. Register a handler: `iii.registerFunction('notifications::on-order-shipped', handler)`.
2. Register the trigger:

```typescript
iii.registerTrigger({
  type: 'subscribe',
  function_id: 'notifications::on-order-shipped',
  config: {
    topic: 'orders.shipped',  // required, non-empty. Exact match only — no wildcards.
  },
})
```

The handler receives the published `data` unchanged, with no `topic` field; if one handler serves multiple topics, embed the topic inside `data` at publish time. The handler's return value is ignored and there is no retry.

For the published payload shape, call `iii get function info` on `publish` or the handler function id.

## api reference

```json
{
  "functions": [
    {
      "description": "Publish an event to a pubsub topic",
      "metadata": {},
      "name": "publish",
      "request_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "description": "Input of the `publish` function — exact field parity with the builtin (engine/src/workers/pubsub/pubsub.rs:78-84).",
        "properties": {
          "data": {
            "description": "JSON payload delivered to each subscriber."
          },
          "topic": {
            "description": "Topic to publish to. Subscribers registered for this topic receive the event.",
            "type": "string"
          }
        },
        "required": [
          "data",
          "topic"
        ],
        "title": "PubSubInput",
        "type": "object"
      },
      "response_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "description": "Success result for `publish`.\n\nThis intentionally serializes as JSON `null` for builtin parity while still giving the SDK a concrete JsonSchema response type for registry publish.",
        "title": "PubSubPublishResponse",
        "type": "null"
      }
    },
    {
      "description": "Internal: reload pubsub configuration from the authoritative store on change.",
      "metadata": {
        "internal": true
      },
      "name": "pubsub::on-config-change",
      "request_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "title": "ConfigChangeRequest",
        "type": "object"
      },
      "response_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "properties": {
          "ok": {
            "type": "boolean"
          }
        },
        "required": [
          "ok"
        ],
        "title": "ConfigChangeAck",
        "type": "object"
      }
    }
  ],
  "triggers": [
    {
      "description": "Subscribe to a topic",
      "invocation_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "description": "Advertised trigger config schema — field parity with the engine's `SubscribeTriggerConfig` (engine/src/trigger_formats.rs:135-141).",
        "properties": {
          "condition_function_id": {
            "description": "Condition function ID accepted for schema parity with the builtin trigger; never evaluated.",
            "type": [
              "string",
              "null"
            ]
          },
          "topic": {
            "description": "Topic to subscribe to",
            "type": "string"
          }
        },
        "required": [
          "topic"
        ],
        "title": "SubscribeTriggerSpec",
        "type": "object"
      },
      "metadata": {},
      "name": "subscribe",
      "return_schema": {}
    }
  ]
}
```