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

# Upgrading from 0.24.x to 0.25.x

> Nothing breaks: triggers keep working. Review the behavior changes, then move registerTrigger to iii.observe and registerTriggerType to observable functions where the source allows it.

0.25 makes every function observable. A trigger type becomes a function that knows who observes it,
and a trigger becomes an observer created with `iii.observe(source, target, options?)`. The old
trigger APIs keep working in 0.25, so most projects upgrade without code changes. The exceptions are
Rust builds that treat warnings as errors (`-D warnings`), which need `#[allow(deprecated)]` on the
calls they keep, and code that calls `unregisterTriggerType`, which now takes effect. Read the
[behavior changes](#behavior-changes) first, then move to the new APIs at your own pace.

> [!NOTE]
>
> The http, cron, queue, state, stream, pubsub, and database workers still publish trigger types in
> 0.25. Keep using `registerTrigger` for them; see [Providers not migrated
> yet](#providers-not-migrated-yet).

## What changes for existing workers

* **The old APIs keep working.** 0.24 SDKs run against a 0.25 engine unchanged (except that an
  `unregisterTriggerType` now [takes effect](#unregistertriggertype-now-takes-effect)), and their
  functions are observable. 0.25 SDKs run against a 0.24 engine too: `observe` fails with
  `UNSUPPORTED`, observe hooks never run, and calls are not dispatched locally.

* **`registerTrigger` and `registerTriggerType` are deprecated** in every SDK and will be removed in
  a future release. They keep working through the trigger type's provider or, once the provider
  moved to a function, through the engine's [legacy shim](#keep-old-consumers-working).

* **One runtime warning per process per API.** The first call logs, through the SDK's logger (each
  SDK names its own API):

  ```text theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
  registerTrigger is deprecated and will be removed in a future release. Use iii.observe(source,
  target) instead. Migration guide: https://iii.dev/docs/upgrading/from-0-24-x
  ```

  `registerTriggerType` logs the equivalent message pointing to `registerFunction` with an `observe`
  descriptor. The engine logs deprecated registrations at `debug` only.

* **Language markers.** TypeScript marks the APIs `@deprecated`, Python emits a `DeprecationWarning`
  at the call site, Go marks them `Deprecated:` (reported by linters such as staticcheck), and Rust
  marks the methods `#[deprecated(since = "0.25.0")]`.

> [!WARNING]
>
> Rust crates built with `#![deny(warnings)]` or `-D warnings` fail to compile on calls to
> `register_trigger`, `register_trigger_type`, or `unregister_trigger_type`. Put
> `#[allow(deprecated)]` on the calls you keep for now (for example bindings to the http or cron
> worker), or migrate them.

## Behavior changes

### Local dispatch

A worker that calls a function it registered itself now runs the handler in-process once the engine
confirmed the registration, instead of sending the call through the engine. Results, error shapes,
and timeouts are the same. The differences:

* **No 16 MiB message limit** on that path, because nothing is serialized. A test that expects
  `payload_too_large` on a call to a function in the same worker must force the remote path.
* **No engine-side span or call metrics** for the local call. The SDK still records its own
  `execute` span, and reports the number of local invocations every 10 seconds
  (`localinvocationstats`), which the engine adds to `iii.invocations.total` with
  `dispatch="local"`.
* **Value semantics.** The handler receives a copy of the payload and the caller a copy of the
  result, as over the wire.
* **Rust:** a local call runs on the caller's task, and a sync handler (`RegisterFunction::new`)
  runs inline on it, where the call's timeout cannot interrupt it. For a handler that can block, use
  an async handler (`RegisterFunction::new_async`) or set `local_dispatch(false)` on it.
* **Recursion:** a local call nested more than 64 local calls deep goes through the engine. In the
  browser while offline, local calls run up to 256 levels deep, and a deeper one fails at once with
  `LOCAL_DEPTH_EXCEEDED`.
* **Python:** a sync handler called locally runs on a thread of its own and honors the caller's
  timeout. Whatever a local handler raises, `BaseException`s such as `KeyboardInterrupt` and
  `SystemExit` included, reaches the caller as an `InvocationError` with code `invocation_failed`
  and the original exception as its `__cause__`.
* **Disconnects:** once confirmed, local dispatch keeps serving calls to the worker's own functions
  while the connection is down.

Local dispatch is off for a session with an RBAC `middleware_function_id`, or one that may not
invoke the function, because a local call would bypass those checks. To keep the old behavior for a
function, a call, or the whole engine:

| Scope | Node / browser | Python | Rust | Go |
| - | - | - | - | - |
| One function | `{ localDispatch: false }` | `local_dispatch=False` | `RegisterFunction::new(f).local_dispatch(false)` | `RegisterFunctionOptions{LocalDispatch: &off}` |
| One call | `dispatch: "remote"` | `"dispatch": "remote"` | `TriggerRequest::new(id, payload).dispatch(Dispatch::Remote)` | `TriggerRequest{Dispatch: iii.DispatchRemote}` |
| Whole engine | `observers.local_dispatch: false` in the [engine config](/docs/next/using-iii/configuration.md#observers) | | | |

### `unregisterTriggerType` now takes effect

Every SDK has always sent `unregistertriggertype`, but engines before 0.25 ignored it. The 0.25
engine removes the trigger type when the sending worker is its provider (and ignores it otherwise).
Its triggers then move to another provider of the type, are taken by a function that declares it as
`legacy_trigger_type`, or wait until a provider registers again. If your worker calls
`unregisterTriggerType` and expected the type to stay, remove the call.

### Smaller changes

* **Legacy shim results.** For a `registerTrigger` that the
  [legacy shim](#keep-old-consumers-working) turns into an observer, the engine sends one
  `TriggerRegistrationResult`, also on success. The SDKs ignore the success result.
* **`RegisterFunction` frames** omit `request_format` and `response_format` when absent instead of
  sending `null`.
* **Loop guard.** Every invocation carries an observe depth; activations deeper than
  `observers.max_observe_depth` (8) are dropped. Workers on 0.24 SDKs do not forward the depth, so
  through them the engine estimates it from their activations still in flight.
* **Observe hooks that hang.** An `onObserverAdded` hook still running after 30 seconds fails the
  observer with `OWNER_TIMEOUT`, and the next hook of the same function may start while it runs.
* **Rust SDK:**
  * public protocol structs gained fields and `Message` gained variants, so struct literals of them
    and exhaustive `match`es on `Message` need updating (prefer the constructors, such as
    `RegisterTriggerInput::new` and `TriggerRequest::new`);
  * the connection state turns `Reconnecting` as soon as the link drops;
  * a panicking handler now returns an `invocation_failed` error to the caller instead of leaving it
    to time out;
  * an `invocation_id` of `None` is omitted from frames instead of sent as `null`.
* **Browser SDK:**
  * `trigger` now sends `metadata`, `traceparent`, and `baggage`, and handlers receive the metadata
    as their second argument;
  * the `unregistertriggertype` frame no longer includes `description`;
  * while the page is offline, functions it hosts run locally up to 256 nested calls, and a deeper
    call fails with `LOCAL_DEPTH_EXCEEDED`.

## Transition overview

| 0.24 | 0.25 |
| - | - |
| `registerTrigger({ type, function_id, config, metadata })` | `observe({ function_id: <source function>, args, filters }, <target>, { metadata })` |
| trigger `type` | the source function, for example `engine::on-functions-changed` |
| trigger `config` | `args` (owner-delivered sources) or `filters` (auto-delivered sources) |
| `condition_function_id` | `filters` for conditions you can state over the event; a condition that needs code stays on `registerTrigger` or moves into the target |
| trigger `function_id` | the target (second argument) |
| trigger `metadata` | `options.metadata`, delivered as the activation's metadata sidecar |
| `trigger.unregister()` | `handle.dispose()` |
| a `TriggerRegistrationResult` error | the handle's `error` state; `ready()` rejects with the code |
| `registerTriggerType({ id, description }, handler)` | `registerFunction(id, handler, { description, observe: { delivery: "owner", args_format, legacy_trigger_type } })` |
| `trigger_request_format` (Python, Rust, Go) | `observe.args_format` |
| `call_request_format` (Python, Rust, Go) | the function's `request_format` |
| `TriggerHandler.registerTrigger(config)` | `observe.onObserverAdded(obs)` |
| `TriggerHandler.unregisterTrigger(config)` | `observe.onObserverRemoved(obs)` |
| `config.id`, `config.function_id`, `config.config` | `obs.id`, `obs.target`, `obs.args` (`obs.legacy.config` for old consumers) |
| `config.metadata` | not visible to the owner: it stays with the engine and reaches the target |
| firing: `trigger({ function_id: binding.function_id, payload })` per binding | `obs.emit(event)` (fire-and-forget) or `obs.call(payload)` (request/response) per observer |
| `unregisterTriggerType(...)` | `functionRef.unregister()` |
| `TriggerTypeRef.registerTrigger` / `.registerFunction` helpers | `registerFunction` plus `observe` |
| Rust `TriggerHandler`, `RegisterTriggerType`, `RegisterTriggerInput` | `ObserveHandler`, `RegisterFunction::observe(Observe::owner()..)`, `Source::function(..)` |
| `engine::registered-triggers::list` (trigger instances) | `engine::observers::list` |

## Move `registerTrigger` to `iii.observe`

Do this for triggers whose source is a function: any regular function, the
[engine-hosted event sources](/docs/next/using-iii/observers.md#engine-hosted-event-sources), or a provider
that moved to an observable function. Register the target first, then observe. With
`deliver: "input"` the target receives the same payload its trigger handler received, so the handler
stays as it is.

### Engine triggers

The engine's own trigger types have source functions in 0.25:

| Trigger type | Observe |
| - | - |
| `engine::functions-available` | `engine::on-functions-changed` |
| `engine::workers-available` | `engine::on-workers-changed` |
| `configuration` | `configuration::on-updated`, filter on `input.id` and `input.event_type` |
| `log` | `observability::on-log`, filter on `input.severity_text` |
| `trace` | `observability::on-span` |

**Node / TypeScript**

**Before (0.24)**

```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
worker.registerTrigger({
  type: "engine::functions-available",
  function_id: "catalog::refresh",
  config: {},
});
```

**After (0.25)**

```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
const refresh = worker.observe(
  { function_id: "engine::on-functions-changed", deliver: "input" },
  "catalog::refresh",
);
await refresh.ready();
```

**Python**

**Before (0.24)**

```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
worker.register_trigger({
    "type": "engine::functions-available",
    "function_id": "catalog::refresh",
    "config": {},
})
```

**After (0.25)**

```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
refresh = worker.observe(
    {"function_id": "engine::on-functions-changed", "deliver": "input"},
    "catalog::refresh",
)
refresh.ready()  # or: await refresh.ready_async()
```

**Rust**

**Before (0.24)**

```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
use iii_sdk::protocol::RegisterTriggerInput;
use serde_json::json;

worker.register_trigger(RegisterTriggerInput::new(
    "engine::functions-available",
    "catalog::refresh",
    json!({}),
))?;
```

**After (0.25)**

```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
use iii_sdk::{Deliver, Source};

let refresh = worker.observe(
    Source::function("engine::on-functions-changed").deliver(Deliver::Input),
    "catalog::refresh",
);
refresh.ready().await?;
```

**Go**

**Before (0.24)**

```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
client.RegisterTrigger("catalog-refresh", "engine::functions-available", "catalog::refresh", nil)
```

**After (0.25)**

```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
refresh, err := client.Observe(iii.Source{
    FunctionID: "engine::on-functions-changed",
    Deliver:    iii.DeliverInput,
}, "catalog::refresh")
if err == nil {
    err = refresh.Ready(ctx)
}
```

### Triggers with config and metadata

For a provider that moved to an owner-delivered function, the trigger's `config` becomes the
observer's `args` and its `metadata` the observer's `metadata`. The example uses the `timer`
provider from the next section.

**Node / TypeScript**

**Before (0.24)**

```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
const trigger = worker.registerTrigger({
  type: "timer",
  function_id: "reports::refresh",
  config: { every_ms: 60_000 },
  metadata: { team: "reports" },
});
// later
trigger.unregister();
```

**After (0.25)**

```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
const observer = worker.observe(
  { function_id: "timer::on-tick", args: { every_ms: 60_000 }, deliver: "input" },
  "reports::refresh",
  { metadata: { team: "reports" } },
);
// later
await observer.dispose();
```

**Python**

**Before (0.24)**

```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
trigger = worker.register_trigger({
    "type": "timer",
    "function_id": "reports::refresh",
    "config": {"every_ms": 60000},
    "metadata": {"team": "reports"},
})
# later
trigger.unregister()
```

**After (0.25)**

```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
observer = worker.observe(
    {"function_id": "timer::on-tick", "args": {"every_ms": 60000}, "deliver": "input"},
    "reports::refresh",
    {"metadata": {"team": "reports"}},
)
# later
observer.dispose()
```

**Rust**

**Before (0.24)**

```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
use iii_sdk::protocol::RegisterTriggerInput;
use serde_json::json;

let trigger = worker.register_trigger(
    RegisterTriggerInput::new("timer", "reports::refresh", json!({ "every_ms": 60_000 }))
        .with_metadata(json!({ "team": "reports" })),
)?;
// later
trigger.unregister();
```

**After (0.25)**

```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
use iii_sdk::{Deliver, ObserverOptions, Source};
use serde_json::json;

let observer = worker.observe_with(
    Source::function("timer::on-tick")
        .args(json!({ "every_ms": 60_000 }))
        .deliver(Deliver::Input),
    "reports::refresh",
    ObserverOptions::default().metadata(json!({ "team": "reports" })),
);
// later
observer.dispose().await?;
```

**Go**

**Before (0.24)**

```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
client.RegisterTrigger("reports-refresh", "timer", "reports::refresh",
    json.RawMessage(`{"every_ms":60000}`), json.RawMessage(`{"team":"reports"}`))
```

**After (0.25)**

```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
observer, err := client.Observe(iii.Source{
    FunctionID: "timer::on-tick",
    Args:       iii.Args{"every_ms": 60000},
    Deliver:    iii.DeliverInput,
}, "reports::refresh", iii.ObserverOptions{Metadata: json.RawMessage(`{"team":"reports"}`)})
// later
observer.Dispose(ctx)
```

Filters cover conditions you can state over the event. For example, a `condition_function_id` that
let only gold customers through becomes `filters: { "input.customer_tier": { "$eq": "gold" } }`,
evaluated without an extra function call; see the [filter language](/docs/next/reference/filters.md). A
condition that needs code (a lookup, an auth check) has no filter form: keep that trigger on
`registerTrigger`, which works in 0.25, or move the check into the target function.

> [!NOTE]
>
> Browser workers use the same `observe` call as Node. A browser handler knows its invocation only
> until its first `await`; wrap later calls in `worker.withInvocation(inv, fn)` so they carry the
> caller and the loop guard's depth. While the page is offline, its own functions run locally up to
> 256 nested calls, then fail with `LOCAL_DEPTH_EXCEEDED`. See [Caller and
> causality](/docs/next/using-iii/observers.md#caller-and-causality).

## Move a trigger type to an observable function

A trigger type becomes a function with an `observe` descriptor. Its owner keeps the same job: set
something up per observer, and hand each event to the observers it belongs to. The engine now stores
the bindings, validates their args, applies their filters and policy, and replays them when your
worker reconnects.

1. Pick a function id for the event, by convention `<worker>::on-<event>`.
2. Set `delivery: "owner"` and describe the config your trigger type accepted as
   `observe.args_format`. If the trigger type declared schemas (Python, Rust, Go), move
   `trigger_request_format` to `observe.args_format` and `call_request_format` to the function's
   `request_format`.
3. Move `registerTrigger(config)` to `onObserverAdded(obs)` and `unregisterTrigger(config)` to
   `onObserverRemoved(obs)`. Read `obs.args` instead of `config.config`.
4. Replace `trigger({ function_id: binding.function_id, payload })` with `obs.emit(event)`, or with
   `obs.call(payload)` when the result matters (an HTTP response, a queue acknowledgement).
5. Declare `legacy_trigger_type` with the old type id, and stop registering the trigger type, so old
   consumers keep working ([below](#keep-old-consumers-working)).

The example moves a `timer` trigger type, which fires each binding every `every_ms` milliseconds.

**Node / TypeScript**

**Before (0.24)**

```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
const bindings = new Map<string, ReturnType<typeof setInterval>>(); // trigger id → its timer

worker.registerTriggerType<{ every_ms: number }>(
  { id: "timer", description: "Fires every every_ms milliseconds" },
  {
    async registerTrigger({ id, function_id, config }) {
      const handle = setInterval(() => {
        worker
          .trigger({ function_id, payload: { every_ms: config.every_ms, fired_at_ms: Date.now() } })
          .catch((err) => console.error(err));
      }, config.every_ms);
      bindings.set(id, handle);
    },
    async unregisterTrigger({ id }) {
      clearInterval(bindings.get(id));
      bindings.delete(id);
    },
  },
);
```

**After (0.25)**

```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
import { ObserveRejection, type Observer } from "iii-sdk";

type TimerArgs = { every_ms: number };
const timers = new Map<string, ReturnType<typeof setInterval>>(); // observer id → its timer

// An observer made from an old registerTrigger has its config in obs.legacy.
const argsOf = (obs: Observer) => (obs.legacy ? obs.legacy.config : obs.args) as TimerArgs;

worker.registerFunction(
  "timer::on-tick",
  async (tick: { every_ms: number; fired_at_ms: number }) => tick,
  {
    description: "Fires every `every_ms` milliseconds. Observe it with args { every_ms }.",
    observe: {
      delivery: "owner",
      args_format: {
        type: "object",
        additionalProperties: false,
        required: ["every_ms"],
        properties: { every_ms: { type: "integer", minimum: 100 } },
      },
      legacy_trigger_type: "timer",
      onObserverAdded(obs) {
        const { every_ms } = argsOf(obs);
        if (!Number.isInteger(every_ms) || every_ms < 100) {
          // args are checked by the engine, legacy configs are not
          throw new ObserveRejection("INVALID_TIMER_CONFIG", "every_ms must be an integer of at least 100");
        }
        const handle = setInterval(() => obs.emit({ every_ms, fired_at_ms: Date.now() }), every_ms);
        timers.set(obs.id, handle);
      },
      onObserverRemoved(obs) {
        clearInterval(timers.get(obs.id));
        timers.delete(obs.id);
      },
    },
  },
);
```

**Python**

**Before (0.24)**

```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
import threading
import time

from iii.trigger import TriggerConfig, TriggerHandler

bindings: dict[str, threading.Event] = {}  # trigger id → stop flag

class TimerHandler(TriggerHandler):
    async def register_trigger(self, config: TriggerConfig) -> None:
        every_ms = config.config["every_ms"]
        stop = bindings[config.id] = threading.Event()

        def run() -> None:
            while not stop.wait(every_ms / 1000):
                worker.trigger({
                    "function_id": config.function_id,
                    "payload": {"every_ms": every_ms, "fired_at_ms": int(time.time() * 1000)},
                })

        threading.Thread(target=run, daemon=True).start()

    async def unregister_trigger(self, config: TriggerConfig) -> None:
        stop = bindings.pop(config.id, None)
        if stop:
            stop.set()

worker.register_trigger_type(
    {"id": "timer", "description": "Fires every every_ms milliseconds"},
    TimerHandler(),
)
```

**After (0.25)**

```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
import threading
import time

from iii import ObserveHooks, ObserveRejection

timers: dict[str, threading.Event] = {}  # observer id → stop flag

def args_of(obs) -> dict:
    # An observer made from an old register_trigger has its config in obs.legacy.
    return obs.legacy["config"] if obs.legacy else obs.args

class TimerHooks(ObserveHooks):
    delivery = "owner"
    args_format = {
        "type": "object",
        "additionalProperties": False,
        "required": ["every_ms"],
        "properties": {"every_ms": {"type": "integer", "minimum": 100}},
    }
    legacy_trigger_type = "timer"

    def on_observer_added(self, obs):
        every_ms = args_of(obs).get("every_ms")
        if not isinstance(every_ms, int) or every_ms < 100:
            # args are checked by the engine, legacy configs are not
            raise ObserveRejection("INVALID_TIMER_CONFIG", "every_ms must be an integer of at least 100")
        stop = timers[obs.id] = threading.Event()

        def run() -> None:
            while not stop.wait(every_ms / 1000):
                obs.emit({"every_ms": every_ms, "fired_at_ms": int(time.time() * 1000)})

        threading.Thread(target=run, daemon=True).start()

    def on_observer_removed(self, obs):
        stop = timers.pop(obs.id, None)
        if stop:
            stop.set()

def on_tick(tick: dict) -> dict:
    return tick

worker.register_function("timer::on-tick", on_tick, observe=TimerHooks())
```

**Rust**

**Before (0.24)**

```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use iii_sdk::protocol::TriggerRequest;
use iii_sdk::trigger::{TriggerConfig, TriggerHandler};
use iii_sdk::{Error, IIIClient, RegisterTriggerType};
use serde_json::json;

#[derive(Clone)]
struct TimerHandler {
    iii: IIIClient,
    tasks: Arc<Mutex<HashMap<String, tokio::task::JoinHandle<()>>>>, // trigger id → its task
}

#[async_trait::async_trait]
impl TriggerHandler for TimerHandler {
    async fn register_trigger(&self, config: TriggerConfig) -> Result<(), Error> {
        let every_ms = config.config["every_ms"].as_u64().unwrap_or(1_000);
        let (iii, function_id) = (self.iii.clone(), config.function_id.clone());
        let task = tokio::spawn(async move {
            let mut interval = tokio::time::interval(Duration::from_millis(every_ms));
            loop {
                interval.tick().await;
                let fired_at_ms = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_millis() as u64;
                let tick = json!({ "every_ms": every_ms, "fired_at_ms": fired_at_ms });
                let _ = iii.trigger(TriggerRequest::new(function_id.clone(), tick)).await;
            }
        });
        self.tasks.lock().unwrap().insert(config.id, task);
        Ok(())
    }

    async fn unregister_trigger(&self, config: TriggerConfig) -> Result<(), Error> {
        if let Some(task) = self.tasks.lock().unwrap().remove(&config.id) {
            task.abort();
        }
        Ok(())
    }
}

worker.register_trigger_type(RegisterTriggerType::new(
    "timer",
    "Fires every every_ms milliseconds",
    TimerHandler { iii: worker.clone(), tasks: Default::default() },
));
```

**After (0.25)**

```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use iii_sdk::{Error, Observe, ObserveHandler, ObserveRejection, Observer, RegisterFunction};
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use serde_json::json;

#[derive(Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
struct TimerArgs {
    every_ms: u64,
}

#[derive(Serialize, Deserialize, JsonSchema)]
struct Tick {
    every_ms: u64,
    fired_at_ms: u64,
}

#[derive(Clone, Default)]
struct Timers(Arc<Mutex<HashMap<String, tokio::task::JoinHandle<()>>>>); // observer id → its task

#[async_trait::async_trait]
impl ObserveHandler for Timers {
    async fn on_observer_added(&self, obs: &Observer) -> Result<(), ObserveRejection> {
        // An observer made from an old register_trigger has its config in legacy().
        let TimerArgs { every_ms } = match obs.legacy() {
            Some(legacy) => serde_json::from_value(legacy.config.clone())
                .map_err(|e| ObserveRejection::new("INVALID_TIMER_CONFIG", e.to_string()))?,
            None => obs.args()?,
        };
        let observer = obs.clone();
        let task = tokio::spawn(async move {
            let mut interval = tokio::time::interval(Duration::from_millis(every_ms));
            loop {
                interval.tick().await;
                let fired_at_ms = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_millis() as u64;
                let _ = observer.emit(&json!({ "every_ms": every_ms, "fired_at_ms": fired_at_ms }));
            }
        });
        self.0.lock().unwrap().insert(obs.id().to_string(), task);
        Ok(())
    }

    async fn on_observer_removed(&self, obs: &Observer) {
        if let Some(task) = self.0.lock().unwrap().remove(obs.id()) {
            task.abort();
        }
    }
}

worker.register_function(
    "timer::on-tick",
    RegisterFunction::new(|tick: Tick| Ok::<_, Error>(tick))
        .description("Fires every `every_ms` milliseconds. Observe it with args { every_ms }.")
        .observe(
            Observe::owner()
                .args_format::<TimerArgs>()
                .legacy_trigger_type("timer")
                .handler(Timers::default()),
        ),
);
```

**Go**

**Before (0.24)**

```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
type timerHandler struct {
    client *iii.Client
    mu     sync.Mutex
    stops  map[string]chan struct{} // trigger id → stop
}

func (h *timerHandler) RegisterTrigger(ctx context.Context, cfg iii.TriggerConfig) error {
    var config struct {
        EveryMs int64 `json:"every_ms"`
    }
    if err := json.Unmarshal(cfg.Config, &config); err != nil {
        return err
    }
    stop := make(chan struct{})
    h.mu.Lock()
    h.stops[cfg.ID] = stop
    h.mu.Unlock()
    go func() {
        ticker := time.NewTicker(time.Duration(config.EveryMs) * time.Millisecond)
        defer ticker.Stop()
        for {
            select {
            case <-stop:
                return
            case now := <-ticker.C:
                data, _ := json.Marshal(map[string]int64{"every_ms": config.EveryMs, "fired_at_ms": now.UnixMilli()})
                h.client.Trigger(context.Background(), iii.TriggerRequest{FunctionID: cfg.FunctionID, Data: data, Namespace: cfg.Namespace})
            }
        }
    }()
    return nil
}

func (h *timerHandler) UnregisterTrigger(ctx context.Context, cfg iii.TriggerConfig) error {
    h.mu.Lock()
    defer h.mu.Unlock()
    if stop, ok := h.stops[cfg.ID]; ok {
        close(stop)
        delete(h.stops, cfg.ID)
    }
    return nil
}

client.RegisterTriggerType("timer", "Fires every every_ms milliseconds",
    &timerHandler{client: client, stops: map[string]chan struct{}{}})
```

**After (0.25)**

```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
var (
    mu    sync.Mutex
    stops = map[string]chan struct{}{} // observer id → stop
)

func everyMs(obs *iii.Observer) (int64, error) {
    var args struct {
        EveryMs int64 `json:"every_ms"`
    }
    var err error
    if obs.Legacy != nil {
        err = json.Unmarshal(obs.Legacy.Config, &args) // an old RegisterTrigger: its config is the args
    } else {
        err = obs.DecodeArgs(&args)
    }
    if err == nil && args.EveryMs < 100 {
        // args are checked by the engine, legacy configs are not
        err = iii.ObserveRejection("INVALID_TIMER_CONFIG", "every_ms must be at least 100")
    }
    return args.EveryMs, err
}

func registerTimer(client *iii.Client) error {
    return client.RegisterFunction("timer::on-tick", func(ctx context.Context, data json.RawMessage) (any, error) {
        return data, nil
    }, iii.RegisterFunctionOptions{Observe: &iii.ObserveHooks{
        Delivery:          iii.DeliveryOwner,
        ArgsFormat:        json.RawMessage(`{"type":"object","additionalProperties":false,"required":["every_ms"],"properties":{"every_ms":{"type":"integer","minimum":100}}}`),
        LegacyTriggerType: "timer",
        OnObserverAdded: func(ctx context.Context, obs *iii.Observer) error {
            every, err := everyMs(obs)
            if err != nil {
                return err
            }
            stop := make(chan struct{})
            mu.Lock()
            stops[obs.ID] = stop
            mu.Unlock()
            go func() {
                ticker := time.NewTicker(time.Duration(every) * time.Millisecond)
                defer ticker.Stop()
                for {
                    select {
                    case <-stop:
                        return
                    case now := <-ticker.C:
                        obs.Emit(context.Background(), map[string]int64{"every_ms": every, "fired_at_ms": now.UnixMilli()})
                    }
                }
            }()
            return nil
        },
        OnObserverRemoved: func(ctx context.Context, obs *iii.Observer) {
            mu.Lock()
            defer mu.Unlock()
            if stop, ok := stops[obs.ID]; ok {
                close(stop)
                delete(stops, obs.ID)
            }
        },
    }})
}
```

The "after" versions keep one timer per observer, like the original. To share one timer per interval
between all its observers, emit from the function's own handler, as in
[Observable functions](/docs/next/creating-workers/observable-functions.md#example-a-timer-that-emits).

### Keep old consumers working

`legacy_trigger_type: "timer"` lets workers that still call `registerTrigger({ type: "timer" })`
keep working with no change, as long as no worker registers the `timer` trigger type the old way:

* The engine turns such a trigger into an observer of your function, owned by the worker that
  registered it. Its id is the trigger id, it has no args and no filters, its metadata is the
  trigger's, and it uses `deliver: "input"`, so the bound function receives the event you emit,
  exactly as it received the fired payload before.
* Your `onObserverAdded` hook sees `obs.legacy = { trigger_type, config }`. The engine does not
  check the config against `args_format`, so validate it as your old `TriggerHandler` did. Accept or
  reject it like any observer; the old worker receives one `TriggerRegistrationResult` with the
  outcome.
* `unregisterTrigger` disposes the observer.
* A worker that registers the trigger type the old way always wins for new registrations. Remove the
  `registerTriggerType` call in the same release that adds the function.
* Only workers on a 0.25 SDK can stand in for a trigger type.

See [Accept old triggers](/docs/next/creating-workers/observable-functions.md#accept-old-triggers) for the
lookup rules.

## Providers not migrated yet

The http, cron, queue (`durable:subscriber`), state, stream, pubsub (`subscribe`), and database
workers live in the workers repository and still register trigger types in 0.25. Until they ship
source functions (such as `http::on-request`, `cron::on-tick`, or `database::on-row-changed`):

* keep using `registerTrigger` to bind functions to them; the deprecation warning is informational,
  and Rust code needs `#[allow(deprecated)]` on those calls;
* their documentation, including the config shapes at [workers.iii.dev](https://workers.iii.dev),
  stays valid;
* once a provider ships its function with `legacy_trigger_type`, your existing `registerTrigger`
  calls keep working through the engine, and you can move them to `iii.observe` at your own pace.

What you can observe in 0.25: any regular function, the
[engine-hosted event sources](/docs/next/using-iii/observers.md#engine-hosted-event-sources), and providers
that already moved to [observable functions](/docs/next/creating-workers/observable-functions.md).

## Migration checklist

* [ ] Upgrade the engine and the SDKs; nothing else is required to keep running
* [ ] Check functions that call themselves for [local dispatch](#local-dispatch) differences (large
  payloads, blocking Rust sync handlers, tests that expect engine spans)
* [ ] Remove `unregisterTriggerType` calls that expected the type to stay registered
* [ ] Add `#[allow(deprecated)]` to Rust trigger calls you keep, if warnings are errors
* [ ] Move triggers on engine events (`engine::functions-available`, `engine::workers-available`,
  `configuration`, `log`, `trace`) to `iii.observe`
* [ ] Move your own trigger types to observable functions with `legacy_trigger_type`
* [ ] Keep `registerTrigger` for the http, cron, queue, state, stream, pubsub, and database workers

## Result

The project runs on 0.25 with its triggers unchanged. Functions you observe deliver to their targets
through the engine's observers, calls a worker makes to its own functions run in-process, and
trigger types you moved keep serving the workers that still use `registerTrigger`.

> [!NOTE]
>
> For the observer model, see [Observers](/docs/next/using-iii/observers.md). For the provider side, see
> [Observable functions](/docs/next/creating-workers/observable-functions.md). For filter syntax, see the
> [filter language reference](/docs/next/reference/filters.md).


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