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

# Observable functions

> Decide what happens when the functions your worker hosts are observed: set things up for each observer, match observers to calls, and answer the caller.

> [!NOTE]
>
> In 0.26, a trigger type is an observable function: `registerTriggerType` becomes
> `registerFunction` with the hooks on this page. To move a 0.25 worker, see [Upgrading from 0.25.x
> to 0.26.x](https://iii.dev/docs/next/upgrading/from-0-25-x.md).

Every function your worker registers can be observed unless you specify otherwise (see
[Registration options](#registration-options)). By default, iii runs each observer's target after
each successful call of the function, with what the observer captures: the call's input, the
handler's return value, and the caller unless its `capture` says otherwise (the call's metadata only
when `capture` names it; see [Observers](https://iii.dev/docs/next/using-iii/observers.md)). Your hooks receive the call's
metadata whatever the observers capture, so route on it in `onObservationMatch` or
`onObservationDeliver`. This page describes
how to manage this: what the observers of its functions receive, which calls reach which observer,
what each observer sets up, and what the caller gets back. To observe a function, see
[Observers](https://iii.dev/docs/next/using-iii/observers.md).

## Hooks

Your worker defines function observation with four hooks, two are observer specific, and two are
specific to each individual observation. Every hook is optional. Without it, iii's default applies;
setting it overrides that default for the function, or, set at the worker level, for every function
the worker hosts.

| Hook | Default, without the hook | Setting the hook overrides it |
| - | - | - |
| `onObserverAdded` | Every observer is accepted as soon as it is added. | Your set-up runs first; the observer is accepted when it returns, rejected if it throws. |
| `onObserverRemoved` | Nothing runs. | Your teardown runs. |
| `onObservationMatch` | Every call is observed by every observer. | Your hook selects which observers get each call. |
| `onObservationDeliver` | iii runs each matched observer's target; the caller gets the handler's return value. | You deliver the call yourself and decide what the caller gets back. |

Each observer's `capture` and filters still decide what its target receives, whichever hooks are
set, unless `onObservationDeliver` runs the targets itself.

Both observer lifecycle hooks can be set for one function or for every function the worker hosts,
and both levels run. When an observer is added, the worker's hook runs, then the function's. When an
observer is removed, the function's hook runs, then the worker's. For each function the SDK runs the
observer lifecycle hooks one at a time, in the order the engine announced the observers; hooks of
different functions run concurrently.

### Observer lifecycle

These hooks run once per observer, when it starts and stops observing the function.

#### `onObserverAdded(obs)`

By default, every observer is accepted as soon as it is added. Setting this hook overrides that: it
runs before the observer is accepted. Use it to set something up from the observer's args. When it
returns, the observer becomes active and receives the function's calls. Throw to reject the observer
instead. A hook that runs past 30 seconds (`observers.owner_ack_timeout_ms`) fails the observer with
`OWNER_TIMEOUT`; a removed hook that runs past it is only logged.

> [tab] Node / TypeScript
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> import { ObserveRejection } from "iii-sdk";
> import { Pool } from "pg";
>
> const pools = new Map([["primary", new Pool({ connectionString: process.env.PRIMARY_DATABASE_URL })]]);
>
> worker.onObserverAdded("database::on-row-changed", async (obs) => {
>   const { db, table } = obs.args as { db: string; table: string };
>   const pool = pools.get(db);
>   if (!pool) throw new ObserveRejection("UNKNOWN_DB", `no database '${db}'`);
>   await pool.query(
>     `CREATE OR REPLACE TRIGGER iii_${table}_changed AFTER INSERT OR UPDATE OR DELETE ON ${table}
>      FOR EACH ROW EXECUTE FUNCTION iii_notify_row_changed()`,
>   );
> });
> ```

> [tab] Python
>
> ```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> import os
>
> from iii import ObserveRejection
> from psycopg_pool import ConnectionPool
>
> pools = {"primary": ConnectionPool(os.environ["PRIMARY_DATABASE_URL"])}
>
> @worker.on_observer_added("database::on-row-changed")
> def install_trigger(obs):
>     db, table = obs.args["db"], obs.args["table"]
>     pool = pools.get(db)
>     if pool is None:
>         raise ObserveRejection("UNKNOWN_DB", f"no database '{db}'")
>     with pool.connection() as conn:
>         conn.execute(
>             f"CREATE OR REPLACE TRIGGER iii_{table}_changed AFTER INSERT OR UPDATE OR DELETE ON {table} "
>             "FOR EACH ROW EXECUTE FUNCTION iii_notify_row_changed()"
>         )
> ```

> [tab] Rust
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> use std::collections::HashMap;
> use std::sync::Arc;
>
> use iii_sdk::{ObserveRejection, Observer};
> use serde::Deserialize;
> use sqlx::PgPool;
>
> #[derive(Deserialize)]
> struct RowChangedArgs {
>     db: String,
>     table: String,
> }
>
> let pools: Arc<HashMap<String, PgPool>> = Arc::new(HashMap::from([(
>     "primary".to_string(),
>     PgPool::connect(&std::env::var("PRIMARY_DATABASE_URL")?).await?,
> )]));
>
> worker.on_observer_added_for("database::on-row-changed", move |obs: Observer| {
>     let pools = pools.clone();
>     async move {
>         let args: RowChangedArgs = obs.args()?;
>         let Some(pool) = pools.get(&args.db) else {
>             return Err(ObserveRejection::new("UNKNOWN_DB", format!("no database '{}'", args.db)));
>         };
>         sqlx::query(&format!(
>             "CREATE OR REPLACE TRIGGER iii_{0}_changed AFTER INSERT OR UPDATE OR DELETE ON {0} \
>              FOR EACH ROW EXECUTE FUNCTION iii_notify_row_changed()",
>             args.table
>         ))
>         .execute(pool)
>         .await
>         .map_err(|e| ObserveRejection::new("DB_ERROR", e.to_string()))?;
>         Ok(())
>     }
> });
> ```

> [tab] Go
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> import (
>     "context"
>     "database/sql"
>     "fmt"
>     "os"
>
>     _ "github.com/jackc/pgx/v5/stdlib"
> )
>
> primary, _ := sql.Open("pgx", os.Getenv("PRIMARY_DATABASE_URL"))
> pools := map[string]*sql.DB{"primary": primary}
>
> type rowChangedArgs struct {
>     DB    string `json:"db"`
>     Table string `json:"table"`
> }
>
> worker.OnObserverAddedFor("database::on-row-changed", func(ctx context.Context, obs *iii.Observer) error {
>     var args rowChangedArgs
>     if err := obs.DecodeArgs(&args); err != nil {
>         return err
>     }
>     pool, ok := pools[args.DB]
>     if !ok {
>         return iii.ObserveRejection("UNKNOWN_DB", "no database '"+args.DB+"'")
>     }
>     _, err := pool.ExecContext(ctx, fmt.Sprintf(
>         "CREATE OR REPLACE TRIGGER iii_%[1]s_changed AFTER INSERT OR UPDATE OR DELETE ON %[1]s "+
>             "FOR EACH ROW EXECUTE FUNCTION iii_notify_row_changed()", args.Table))
>     return err
> })
> ```

#### `onObserverRemoved(obs)`

Runs after the engine has removed the observer and delivery to it has stopped. Use it to perform
your own teardown. It runs once for each accepted observer, when the consumer disposes it, when
`max_fires` or expiry ends it, when the observing worker disconnects, and after a late rejection.
For an observer that is never accepted it runs only at the levels whose `onObserverAdded` resolved.

> [!NOTE]
>
> If `onObserverRemoved` throws an error it will cause the error to be logged but it will not
> prevent observer removal. Observer removal happens at the engine level.

> [tab] Node / TypeScript
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> import { Pool } from "pg";
>
> const pools = new Map([["primary", new Pool({ connectionString: process.env.PRIMARY_DATABASE_URL })]]);
>
> worker.onObserverRemoved("database::on-row-changed", async (obs) => {
>   const { db, table } = obs.args as { db: string; table: string };
>   await pools.get(db)?.query(`DROP TRIGGER IF EXISTS iii_${table}_changed ON ${table}`);
> });
> ```

> [tab] Python
>
> ```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> import os
>
> from psycopg_pool import ConnectionPool
>
> pools = {"primary": ConnectionPool(os.environ["PRIMARY_DATABASE_URL"])}
>
> @worker.on_observer_removed("database::on-row-changed")
> def drop_trigger(obs):
>     db, table = obs.args["db"], obs.args["table"]
>     pool = pools.get(db)
>     if pool is None:
>         return
>     with pool.connection() as conn:
>         conn.execute(f"DROP TRIGGER IF EXISTS iii_{table}_changed ON {table}")
> ```

> [tab] Rust
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> use std::collections::HashMap;
> use std::sync::Arc;
>
> use iii_sdk::Observer;
> use serde::Deserialize;
> use sqlx::PgPool;
>
> #[derive(Deserialize)]
> struct RowChangedArgs {
>     db: String,
>     table: String,
> }
>
> let pools: Arc<HashMap<String, PgPool>> = Arc::new(HashMap::from([(
>     "primary".to_string(),
>     PgPool::connect(&std::env::var("PRIMARY_DATABASE_URL")?).await?,
> )]));
>
> worker.on_observer_removed_for("database::on-row-changed", move |obs: Observer| {
>     let pools = pools.clone();
>     async move {
>         let Ok(args) = obs.args::<RowChangedArgs>() else {
>             return;
>         };
>         if let Some(pool) = pools.get(&args.db) {
>             let _ = sqlx::query(&format!("DROP TRIGGER IF EXISTS iii_{0}_changed ON {0}", args.table))
>                 .execute(pool)
>                 .await;
>         }
>     }
> });
> ```

> [tab] Go
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> import (
>     "context"
>     "database/sql"
>     "fmt"
>     "os"
>
>     _ "github.com/jackc/pgx/v5/stdlib"
> )
>
> primary, _ := sql.Open("pgx", os.Getenv("PRIMARY_DATABASE_URL"))
> pools := map[string]*sql.DB{"primary": primary}
>
> type rowChangedArgs struct {
>     DB    string `json:"db"`
>     Table string `json:"table"`
> }
>
> worker.OnObserverRemovedFor("database::on-row-changed", func(ctx context.Context, obs *iii.Observer) {
>     var args rowChangedArgs
>     if obs.DecodeArgs(&args) != nil {
>         return
>     }
>     if pool, ok := pools[args.DB]; ok {
>         pool.ExecContext(ctx, fmt.Sprintf("DROP TRIGGER IF EXISTS iii_%[1]s_changed ON %[1]s", args.Table))
>     }
> })
> ```

#### Set a hook for one function or for every function

A hook set on the worker runs for every function the worker hosts. A hook set on one function runs
for that function only.

> [tab] Node / TypeScript
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> import { Client } from "pg";
>
> // every function this worker hosts: hold one LISTEN connection while anything is observed
> let listener: Client | undefined;
> let observed = 0;
>
> worker.onObserverAdded(async () => {
>   if (observed++ > 0) return;
>   listener = new Client({ connectionString: process.env.PRIMARY_DATABASE_URL });
>   await listener.connect();
>   await listener.query("LISTEN iii_row_changed");
> });
>
> worker.onObserverRemoved(async () => {
>   if (--observed > 0) return;
>   await listener?.end();
>   listener = undefined;
> });
> ```

> [tab] Python
>
> ```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> import os
>
> import psycopg
>
> # every function this worker hosts: hold one LISTEN connection while anything is observed
> listener: psycopg.Connection | None = None
> observed = 0
>
> @worker.on_observer_added
> def open_listener(obs):
>     global listener, observed
>     observed += 1
>     if observed > 1:
>         return
>     listener = psycopg.connect(os.environ["PRIMARY_DATABASE_URL"], autocommit=True)
>     listener.execute("LISTEN iii_row_changed")
>
> @worker.on_observer_removed
> def close_listener(obs):
>     global listener, observed
>     observed -= 1
>     if observed > 0:
>         return
>     listener.close()
>     listener = None
> ```

> [tab] Rust
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> use std::sync::atomic::{AtomicUsize, Ordering};
> use std::sync::Arc;
>
> use iii_sdk::{ObserveRejection, Observer};
> use sqlx::postgres::PgListener;
> use tokio::sync::Mutex;
>
> // every function this worker hosts: hold one LISTEN connection while anything is observed
> let listener: Arc<Mutex<Option<PgListener>>> = Arc::new(Mutex::new(None));
> let observed = Arc::new(AtomicUsize::new(0));
>
> let (slot, count) = (listener.clone(), observed.clone());
> worker.on_observer_added(move |_obs: Observer| {
>     let (slot, count) = (slot.clone(), count.clone());
>     async move {
>         if count.fetch_add(1, Ordering::SeqCst) > 0 {
>             return Ok(());
>         }
>         let url = std::env::var("PRIMARY_DATABASE_URL").unwrap_or_default();
>         let mut conn = PgListener::connect(&url)
>             .await
>             .map_err(|e| ObserveRejection::new("DB_ERROR", e.to_string()))?;
>         conn.listen("iii_row_changed")
>             .await
>             .map_err(|e| ObserveRejection::new("DB_ERROR", e.to_string()))?;
>         *slot.lock().await = Some(conn);
>         Ok(())
>     }
> });
>
> let (slot, count) = (listener.clone(), observed.clone());
> worker.on_observer_removed(move |_obs: Observer| {
>     let (slot, count) = (slot.clone(), count.clone());
>     async move {
>         if count.fetch_sub(1, Ordering::SeqCst) > 1 {
>             return;
>         }
>         slot.lock().await.take();
>     }
> });
> ```

> [tab] Go
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> import (
>     "context"
>     "os"
>     "sync"
>
>     "github.com/jackc/pgx/v5"
> )
>
> // every function this worker hosts: hold one LISTEN connection while anything is observed
> var (
>     mu       sync.Mutex
>     listener *pgx.Conn
>     observed int
> )
>
> worker.OnObserverAdded(func(ctx context.Context, obs *iii.Observer) error {
>     mu.Lock()
>     defer mu.Unlock()
>     observed++
>     if observed > 1 {
>         return nil
>     }
>     conn, err := pgx.Connect(ctx, os.Getenv("PRIMARY_DATABASE_URL"))
>     if err != nil {
>         return err
>     }
>     if _, err := conn.Exec(ctx, "LISTEN iii_row_changed"); err != nil {
>         return err
>     }
>     listener = conn
>     return nil
> })
>
> worker.OnObserverRemoved(func(ctx context.Context, obs *iii.Observer) {
>     mu.Lock()
>     defer mu.Unlock()
>     observed--
>     if observed > 0 {
>         return
>     }
>     listener.Close(ctx)
>     listener = nil
> })
> ```

For one function, pass its id to the same pair: The hook names in each SDK:

| Hook | Node / Browser | Python | Rust | Go |
| - | - | - | - | - |
| Added, every function | `onObserverAdded(hook)` | `on_observer_added(hook)` | `on_observer_added(hook)` | `OnObserverAdded(hook)` |
| Added, one function | `onObserverAdded(id, hook)` | `on_observer_added(id, hook)` | `on_observer_added_for(id, hook)` | `OnObserverAddedFor(id, hook)` |
| Removed, every function | `onObserverRemoved(hook)` | `on_observer_removed(hook)` | `on_observer_removed(hook)` | `OnObserverRemoved(hook)` |
| Removed, one function | `onObserverRemoved(id, hook)` | `on_observer_removed(id, hook)` | `on_observer_removed_for(id, hook)` | `OnObserverRemovedFor(id, hook)` |
| Match, one function | `onObservationMatch(id, hook)` | `on_observation_match(id, hook)` | `on_observation_match_for(id, rule)` | `OnObservationMatchFor(id, rule)` |
| Deliver, one function | `onObservationDeliver(id, hook)` | `on_observation_deliver(id, hook)` | `on_observation_deliver_for(id, hook)` | `OnObservationDeliverFor(id, hook)` |

The `FunctionRef` has the same methods with the id filled in: `onObserverAdded(hook)` and its
siblings in Node, Python and Rust (`on_observer_added(hook)`), and `OnObserverAdded(hook)` in Go. In
Node, Rust and Go each method returns its receiver, so calls chain. In Python each returns the hook,
so it also works as a decorator, and `on_observer_added("id")` without a hook is a decorator
factory. Added, removed and deliver hooks may be sync or async in Node and Python. A match predicate
and key functions are sync, since they run on every call. In Rust a hook is an async closure or a
type that implements `AddedHook`, `RemovedHook` or `DeliverHook`.

A match hook is a predicate or a pair of key functions (see
[Match observers to calls](#match-observers-to-calls)):

| SDK | Predicate | Keys |
| - | - | - |
| Node / Browser | `onObservationMatch(id, (obs, call) => …)` | `onObservationMatch(id, { observer, call })` |
| Python | `on_observation_match(id, predicate)` | `on_observation_match(id, observer=…, call=…)` |
| Rust | `Match::predicate(f)` | `Match::keys(observer_keys, call_keys)` |
| Go | `iii.MatchFunc(f)` | `iii.MatchKeys(observerKeys, callKeys)` |

Pass `null` (`None` in Python, `nil` in Go) to remove a hook. Rust removes hooks with `clear_*`
methods: `clear_observer_added()`, `clear_observer_removed()`, `clear_observation_match()` and
`clear_observation_deliver()` on the `FunctionRef`; `clear_observer_added_for(id)`,
`clear_observer_removed_for(id)`, `clear_observation_match_for(id)` and
`clear_observation_deliver_for(id)` on the client; and `clear_observer_added()` and
`clear_observer_removed()` on the client for the worker-wide hooks.

Hooks are stored by function id. They stay in place across `unregister()` and a later registration
of the same id, and they can be set before the function is registered. Once the worker's connection
is registered, the SDK logs a warning for each hook id the worker has not registered, which catches
a mistyped id.

> [tab] Node / TypeScript
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> import { Client } from "pg";
>
> let listener: Client | undefined;
> let observed = 0;
>
> worker.onObserverAdded("database::on-row-changed", async (obs) => {
>   if (observed++ > 0) return;
>   listener = new Client({ connectionString: process.env.PRIMARY_DATABASE_URL });
>   await listener.connect();
>   await listener.query("LISTEN iii_row_changed");
> });
>
> worker.onObserverRemoved("database::on-row-changed", async (obs) => {
>   if (--observed > 0) return;
>   await listener?.end();
>   listener = undefined;
> });
> ```

> [tab] Python
>
> ```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> import os
>
> import psycopg
>
> listener: psycopg.Connection | None = None
> observed = 0
>
> @worker.on_observer_added("database::on-row-changed")
> def open_listener(obs):
>     global listener, observed
>     observed += 1
>     if observed > 1:
>         return
>     listener = psycopg.connect(os.environ["PRIMARY_DATABASE_URL"], autocommit=True)
>     listener.execute("LISTEN iii_row_changed")
>
> @worker.on_observer_removed("database::on-row-changed")
> def close_listener(obs):
>     global listener, observed
>     observed -= 1
>     if observed > 0:
>         return
>     listener.close()
>     listener = None
> ```

> [tab] Rust
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> use std::sync::atomic::{AtomicUsize, Ordering};
> use std::sync::Arc;
>
> use iii_sdk::{ObserveRejection, Observer};
> use sqlx::postgres::PgListener;
> use tokio::sync::Mutex;
>
> let listener: Arc<Mutex<Option<PgListener>>> = Arc::new(Mutex::new(None));
> let observed = Arc::new(AtomicUsize::new(0));
>
> let (slot, count) = (listener.clone(), observed.clone());
> worker.on_observer_added_for("database::on-row-changed", move |_obs: Observer| {
>     let (slot, count) = (slot.clone(), count.clone());
>     async move {
>         if count.fetch_add(1, Ordering::SeqCst) > 0 {
>             return Ok(());
>         }
>         let url = std::env::var("PRIMARY_DATABASE_URL").unwrap_or_default();
>         let mut conn = PgListener::connect(&url)
>             .await
>             .map_err(|e| ObserveRejection::new("DB_ERROR", e.to_string()))?;
>         conn.listen("iii_row_changed")
>             .await
>             .map_err(|e| ObserveRejection::new("DB_ERROR", e.to_string()))?;
>         *slot.lock().await = Some(conn);
>         Ok(())
>     }
> });
>
> let (slot, count) = (listener.clone(), observed.clone());
> worker.on_observer_removed_for("database::on-row-changed", move |_obs: Observer| {
>     let (slot, count) = (slot.clone(), count.clone());
>     async move {
>         if count.fetch_sub(1, Ordering::SeqCst) > 1 {
>             return;
>         }
>         slot.lock().await.take();
>     }
> });
> ```

> [tab] Go
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> import (
>     "context"
>     "os"
>     "sync"
>
>     "github.com/jackc/pgx/v5"
> )
>
> var (
>     mu       sync.Mutex
>     listener *pgx.Conn
>     observed int
> )
>
> worker.OnObserverAddedFor("database::on-row-changed", func(ctx context.Context, obs *iii.Observer) error {
>     mu.Lock()
>     defer mu.Unlock()
>     observed++
>     if observed > 1 {
>         return nil
>     }
>     conn, err := pgx.Connect(ctx, os.Getenv("PRIMARY_DATABASE_URL"))
>     if err != nil {
>         return err
>     }
>     if _, err := conn.Exec(ctx, "LISTEN iii_row_changed"); err != nil {
>         return err
>     }
>     listener = conn
>     return nil
> })
>
> worker.OnObserverRemovedFor("database::on-row-changed", func(ctx context.Context, obs *iii.Observer) {
>     mu.Lock()
>     defer mu.Unlock()
>     observed--
>     if observed > 0 {
>         return
>     }
>     listener.Close(ctx)
>     listener = nil
> })
> ```

A hook for one function is set by id (see [Hooks](#hooks)) or chained on the `FunctionRef` that
`registerFunction` returns. Each method returns its receiver, so calls chain. Added, removed and
deliver hooks can be sync or async. A match predicate is sync, since it runs on every call. In Rust
a hook is an async closure or a type that implements `AddedHook`, `RemovedHook` or `DeliverHook`.
Implement these traits with native `async fn`, without `#[async_trait]`: their futures must be
`Send` and may borrow `self` across `.await`. See
[Implement Rust hooks with native async fn](https://iii.dev/docs/next/upgrading/from-0-25-x.md#implement-rust-hooks-with-native-async-fn).

Removing a hook returns the function to its default (see [Hooks](#hooks)).

Hooks are stored by function id. They stay in place across `unregister()` and a later registration
of the same id, and they can be set before the function is registered. Once the worker's connection
is registered, the SDK logs a warning for each hook id the worker has not registered, which catches
a mistyped id.

#### Reject an observer

An observer is accepted when every `onObserverAdded` that applies to it resolves. The SDK then
acknowledges it to the engine, and the observer becomes `active`. So its database trigger exists
before the first call it must receive.

An `onObserverAdded` that throws rejects the observer. A thrown `ObserveRejection(code, message)`
sets the code the consumer's `ready()` fails with; any other error is `OBSERVE_REJECTED`.

> [tab] Node / TypeScript
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> import { ObserveRejection } from "iii-sdk";
>
> worker.onObserverAdded("database::on-row-changed", async (obs) => {
>   const { db } = obs.args as { db: string };
>   // accept: a hook that returns accepts the observer
>   if (db === "primary") return;
>   // reject: a thrown ObserveRejection sets the code ready() fails with
>   throw new ObserveRejection("UNKNOWN_DB", `no database '${db}'`);
> });
> ```

> [tab] Python
>
> ```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> from iii import ObserveRejection
>
> @worker.on_observer_added("database::on-row-changed")
> def accept_primary(obs):
>     db = obs.args["db"]
>     # accept: a hook that returns accepts the observer
>     if db == "primary":
>         return
>     # reject: a raised ObserveRejection sets the code ready() fails with
>     raise ObserveRejection("UNKNOWN_DB", f"no database '{db}'")
> ```

> [tab] Rust
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> use iii_sdk::{ObserveRejection, Observer};
>
> worker.on_observer_added_for("database::on-row-changed", move |obs: Observer| async move {
>     let db = obs.args_value()["db"].as_str().unwrap_or_default().to_string();
>     // accept: Ok accepts the observer
>     if db == "primary" {
>         return Ok(());
>     }
>     // reject: Err sets the code ready() fails with
>     Err(ObserveRejection::new("UNKNOWN_DB", format!("no database '{db}'")))
> });
> ```

> [tab] Go
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> worker.OnObserverAddedFor("database::on-row-changed", func(ctx context.Context, obs *iii.Observer) error {
>     db, _ := obs.Args["db"].(string)
>     // accept: returning nil accepts the observer
>     if db == "primary" {
>         return nil
>     }
>     // reject: the returned error sets the code ready() fails with
>     return iii.ObserveRejection("UNKNOWN_DB", "no database '"+db+"'")
> })
> ```

#### No replay of past executions

iii does not store observation history. A hook reacts only to the executions of its function that
happen after the hook is registered, and an observer receives only the calls that happen after it is
accepted.

#### Reject an observer later

An observer that the function accepted can become invalid later. For example, a configuration reload
can remove the resource that the observer watches. `obs.reject(code, message)` stops delivery to the
observer immediately, and the SDK runs `onObserverRemoved` next, connected or not. The engine moves
the observer to `error` with that code. If the connection is lost, the SDK sends the rejection again
after it reconnects.

> [tab] Node / TypeScript
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> import { Pool } from "pg";
>
> type RowChange = { db: string; table: string; op: "insert" | "update" | "delete"; row: unknown };
>
> const pools = new Map([["primary", new Pool({ connectionString: process.env.PRIMARY_DATABASE_URL })]]);
> const rowChanged = worker.registerFunction(
>   "database::on-row-changed",
>   (change: RowChange) => change,
>   {
>     private: true,
>     // the args an observer of this function passes
>     args_format: {
>       type: "object",
>       properties: { db: { type: "string" }, table: { type: "string" }, op: { type: "string" } },
>       required: ["db", "table", "op"],
>     },
>   },
> );
>
> function rejectDroppedDatabases() {
>   for (const obs of rowChanged.observers()) {
>     if (!pools.has(obs.args.db as string)) {
>       obs.reject("UNKNOWN_DB", `no database '${obs.args.db}'`);
>     }
>   }
> }
> ```

> [tab] Python
>
> ```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> import os
>
> from psycopg_pool import ConnectionPool
>
> pools = {"primary": ConnectionPool(os.environ["PRIMARY_DATABASE_URL"])}
> # the args an observer of this function passes
> ROW_ARGS = {
>     "type": "object",
>     "properties": {"db": {"type": "string"}, "table": {"type": "string"}, "op": {"type": "string"}},
>     "required": ["db", "table", "op"],
> }
>
> row_changed = worker.register_function(
>     "database::on-row-changed", lambda change: change, private=True, args_format=ROW_ARGS
> )
>
> def reject_dropped_databases():
>     for obs in row_changed.observers():
>         db = obs.args["db"]
>         if db not in pools:
>             obs.reject("UNKNOWN_DB", f"no database '{db}'")
> ```

> [tab] Rust
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> use std::collections::HashMap;
>
> use iii_sdk::{Error, RegisterFunction};
> use schemars::JsonSchema; // schemars 0.8, the version iii-sdk uses
> use serde::{Deserialize, Serialize};
> use serde_json::Value;
> use sqlx::PgPool;
>
> #[derive(Serialize, Deserialize, JsonSchema)]
> struct RowChange {
>     db: String,
>     table: String,
>     op: String,
>     row: Value,
> }
>
> let pools: HashMap<String, PgPool> = HashMap::from([(
>     "primary".to_string(),
>     PgPool::connect(&std::env::var("PRIMARY_DATABASE_URL")?).await?,
> )]);
>
> #[derive(JsonSchema)]
> struct RowChangeArgs {
>     db: String,
>     table: String,
>     op: String,
> }
>
> let row_changed = worker.register_function(
>     "database::on-row-changed",
>     RegisterFunction::new(|change: RowChange| Ok::<_, Error>(change))
>         .private()
>         // the args an observer of this function passes
>         .args_format::<RowChangeArgs>(),
> );
>
> fn reject_dropped_databases(
>     row_changed: &iii_sdk::runtime::FunctionRef,
>     pools: &HashMap<String, PgPool>,
> ) {
>     for obs in row_changed.observers() {
>         let db = obs.args_value()["db"].as_str().unwrap_or_default().to_string();
>         if !pools.contains_key(&db) {
>             obs.reject("UNKNOWN_DB", format!("no database '{db}'"));
>         }
>     }
> }
> ```

> [tab] Go
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> import (
>     "database/sql"
>     "encoding/json"
>     "os"
>
>     _ "github.com/jackc/pgx/v5/stdlib"
> )
>
> type rowChange struct {
>     DB    string          `json:"db"`
>     Table string          `json:"table"`
>     Op    string          `json:"op"`
>     Row   json.RawMessage `json:"row"`
> }
>
> primary, _ := sql.Open("pgx", os.Getenv("PRIMARY_DATABASE_URL"))
> pools := map[string]*sql.DB{"primary": primary}
>
> rowChanged, _ := worker.RegisterFunction("database::on-row-changed", iii.Identity, iii.FunctionOptions{
>     Private: true,
>     // the args an observer of this function passes
>     ArgsFormat: json.RawMessage(`{"type":"object","properties":{"db":{"type":"string"},"table":{"type":"string"},"op":{"type":"string"}},"required":["db","table","op"]}`),
> })
>
> rejectDroppedDatabases := func() {
>     for _, obs := range rowChanged.Observers() {
>         db, _ := obs.Args["db"].(string)
>         if _, ok := pools[db]; !ok {
>             obs.Reject("UNKNOWN_DB", "no database '"+db+"'")
>         }
>     }
> }
> ```

`FunctionRef.observers()` returns the active observers of the function.

### Observation lifecycle

These hooks run per call. An observation is one call of the function reaching, or being considered
for, one observer.

> [!NOTE]
>
> An observer's args and the call settle the choice in
> [`onObservationMatch`](#match-observers-to-calls). Choose in `onObservationDeliver` when the
> choice compares observers with each other, or when making it produces part of the payload.

#### `onObservationMatch(obs, call)`

By default, every call is observed by every observer. Setting this hook overrides that: it selects,
for one call, which observers get it, as a predicate that returns a boolean. iii then applies each
matched observer's `capture` and filters as usual.

> [tab] Node / TypeScript
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> worker.onObservationMatch("database::on-row-changed", (obs, call) =>
>   // one observation: this observer, this call
>   obs.args.table === (call.input as { table: string }).table,
> );
> ```

> [tab] Python
>
> ```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> worker.on_observation_match(
>     "database::on-row-changed",
>     # one observation: this observer, this call
>     lambda obs, call: obs.args["table"] == call.input["table"],
> )
> ```

> [tab] Rust
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> use iii_sdk::Match;
>
> worker.on_observation_match_for(
>     "database::on-row-changed",
>     // one observation: this observer, this call
>     Match::predicate(|obs, call| obs.args_value()["table"] == call.input["table"]),
> );
> ```

> [tab] Go
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> import "encoding/json"
>
> type rowChange struct {
>     Table string `json:"table"`
> }
>
> worker.OnObservationMatchFor("database::on-row-changed", iii.MatchFunc(
>     // one observation: this observer, this call
>     func(obs *iii.Observer, call *iii.Call) bool {
>         var change rowChange
>         _ = json.Unmarshal(call.Input, &change)
>         return obs.Args["table"] == change.Table
>     },
> ))
> ```

#### `onObservationDeliver(call, observers)`

By default, iii carries out each observation: it runs the target of every matched observer whose
capture and filters match the call, and gives the caller the handler's return value. Setting this
hook overrides that: you deliver the call yourself and decide what is returned to the caller.

> [tab] Node / TypeScript
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> type RowChange = {
>   db: string;
>   table: string;
>   op: "insert" | "update" | "delete";
>   row: Record<string, unknown>;
> };
>
> worker.onObservationDeliver("database::on-row-changed", async (call, observers) => {
>   const change = call.input as RowChange;
>   // the observer that names this table, before one that names every table
>   const observer =
>     observers.find((obs) => obs.args.table === change.table) ??
>     observers.find((obs) => obs.args.table === "*");
>   if (!observer || !observer.matches(call)) return { delivered: false };
>   return worker.call({ observer, payload: change }); // the caller gets the target's result
> });
> ```

> [tab] Python
>
> ```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> @worker.on_observation_deliver("database::on-row-changed")
> async def deliver(call, observers):
>     change = call.input
>     # the observer that names this table, before one that names every table
>     observer = next((obs for obs in observers if obs.args["table"] == change["table"]), None)
>     if observer is None:
>         observer = next((obs for obs in observers if obs.args["table"] == "*"), None)
>     if observer is None or not observer.matches(call):
>         return {"delivered": False}
>     return await worker.call_async({"observer": observer, "payload": change})  # the target's result
> ```

> [tab] Rust
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> use iii_sdk::protocol::CallRequest;
> use iii_sdk::{Call, Observer};
> use serde_json::json;
>
> let iii = worker.clone();
> worker.on_observation_deliver_for("database::on-row-changed", move |call: Call, observers: Vec<Observer>| {
>     let iii = iii.clone();
>     async move {
>         let table = call.input["table"].as_str().unwrap_or_default().to_string();
>         // the observer that names this table, before one that names every table
>         let observer = observers
>             .iter()
>             .find(|obs| obs.args_value()["table"] == table)
>             .or_else(|| observers.iter().find(|obs| obs.args_value()["table"] == "*"));
>         let Some(observer) = observer else {
>             return Ok(json!({ "delivered": false }));
>         };
>         if !observer.matches(&call) {
>             return Ok(json!({ "delivered": false }));
>         }
>         iii.call(CallRequest::observer(observer, call.input.clone())).await // the target's result
>     }
> });
> ```

> [tab] Go
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> import (
>     "context"
>     "encoding/json"
> )
>
> type rowChange struct {
>     DB    string          `json:"db"`
>     Table string          `json:"table"`
>     Op    string          `json:"op"`
>     Row   json.RawMessage `json:"row"`
> }
>
> worker.OnObservationDeliverFor("database::on-row-changed", func(ctx context.Context, call *iii.Call, observers []*iii.Observer) (any, error) {
>     var change rowChange
>     if err := json.Unmarshal(call.Input, &change); err != nil {
>         return nil, err
>     }
>     var observer *iii.Observer
>     for _, obs := range observers {
>         // the observer that names this table, before one that names every table
>         if obs.Args["table"] == change.Table {
>             observer = obs
>             break
>         }
>         if observer == nil && obs.Args["table"] == "*" {
>             observer = obs
>         }
>     }
>     if observer == nil || !observer.Matches(call) {
>         return map[string]bool{"delivered": false}, nil
>     }
>     return worker.Call(ctx, iii.CallRequest{Observer: observer, Payload: call.Input}) // the target's result
> })
> ```

#### Match observers to calls

`onObservationMatch` selects, for each call, which observers get it, so a function can route its
calls by the args of its observers. The predicate receives the observer and the call, and returns a
boolean: `true` when that observer gets this call, `false` when it does not. The SDK calls it once
for each active observer, before that observer's capture and filters; a predicate that throws counts as
no match for that observer. iii does everything else as usual: each observer's capture, filters
and policy still apply.

> [tab] Node / TypeScript
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> worker.onObservationMatch(
>   "database::on-row-changed",
>   (obs, call) => obs.args.db === call.input.db && obs.args.table === call.input.table,
> );
> ```

> [tab] Python
>
> ```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> worker.on_observation_match(
>     "database::on-row-changed",
>     lambda obs, call: obs.args["db"] == call.input["db"] and obs.args["table"] == call.input["table"],
> )
> ```

> [tab] Rust
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> use iii_sdk::Match;
>
> worker.on_observation_match_for(
>     "database::on-row-changed",
>     Match::predicate(|obs, call| {
>         let args = obs.args_value();
>         args["db"] == call.input["db"] && args["table"] == call.input["table"]
>     }),
> );
> ```

> [tab] Go
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> import "encoding/json"
>
> type rowChange struct {
>     DB    string `json:"db"`
>     Table string `json:"table"`
> }
>
> worker.OnObservationMatchFor("database::on-row-changed", iii.MatchFunc(func(obs *iii.Observer, call *iii.Call) bool {
>     var change rowChange
>     _ = json.Unmarshal(call.Input, &change)
>     return obs.Args["db"] == change.DB && obs.Args["table"] == change.Table
> }))
> ```

## Producing events with observers

Function observation is not limited to input/output. Functions can use the same observer pattern to
produce events during execution. Do it with a second function that this worker hosts, and call it
with the event as the payload. That second function works as the event source:

* Its handler returns its input, so one call is one event.
* Each call runs the observers of the event source.
* The producing function gets the handler's return value, so it keeps running.
* `private: true` keeps the calls to this worker: other workers can observe the event source, and
  only this worker can call it (see [Private functions](#private-functions)).

`FunctionRef.isObserved()` returns true when the event source has at least 1 active observer. It has
minimal overhead, so it can be used to skip events that are not being observed.

> [tab] Node / TypeScript
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> type RowChange = {
>   db: string;
>   table: string;
>   op: "insert" | "update" | "delete";
>   row: Record<string, unknown>;
> };
>
> const rowChanged = worker.registerFunction(
>   "database::on-row-changed",
>   (change: RowChange) => change,
>   {
>     private: true,
>     // the args an observer of this function passes
>     args_format: {
>       type: "object",
>       properties: { db: { type: "string" }, table: { type: "string" }, op: { type: "string" } },
>       required: ["db", "table", "op"],
>     },
>   },
> );
>
> // called for each NOTIFY on the worker's LISTEN connection
> async function publishRowChange(db: string, table: string, op: RowChange["op"], row: RowChange["row"]) {
>   if (!rowChanged.isObserved()) return; // no observers: skip the call
>   await worker.call({ function_id: "database::on-row-changed", payload: { db, table, op, row } });
> }
>
> // from any handler of this worker, for each changed row
> await publishRowChange("primary", "orders", "update", { id: 42, status: "paid" });
> ```

> [tab] Python
>
> ```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> # the args an observer of this function passes
> ROW_ARGS = {
>     "type": "object",
>     "properties": {"db": {"type": "string"}, "table": {"type": "string"}, "op": {"type": "string"}},
>     "required": ["db", "table", "op"],
> }
>
> row_changed = worker.register_function(
>     "database::on-row-changed", lambda change: change, private=True, args_format=ROW_ARGS
> )
>
> # called for each NOTIFY on the worker's LISTEN connection
> def publish_row_change(db, table, op, row):
>     if not row_changed.is_observed():
>         return  # no observers: skip the call
>     worker.call({
>         "function_id": "database::on-row-changed",
>         "payload": {"db": db, "table": table, "op": op, "row": row},
>     })
>
> # from any handler of this worker, for each changed row
> publish_row_change("primary", "orders", "update", {"id": 42, "status": "paid"})
> ```

> [tab] Rust
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> use iii_sdk::protocol::CallRequest;
> use iii_sdk::{runtime::FunctionRef, Error, IIIClient, RegisterFunction};
> use schemars::JsonSchema; // schemars 0.8, the version iii-sdk uses
> use serde::{Deserialize, Serialize};
> use serde_json::{json, Value};
>
> #[derive(Serialize, Deserialize, JsonSchema)]
> struct RowChange {
>     db: String,
>     table: String,
>     op: String,
>     row: Value,
>     // ... the row values
> }
>
> fn row_key(db: &str, table: &str, op: &str) -> String {
>     format!("{db}|{table}|{op}")
> }
>
> // the keys an observer listens for
> fn observer_keys(obs: &Observer) -> Keys<'_> {
>     let args: RowChangedArgs = obs.args().expect("validated against args_format");
>     let table = args.table.unwrap_or_else(|| "*".into());
>     let ops = args.ops.unwrap_or_else(|| vec!["*".into()]);
>     ops.iter().map(|op| row_key(&args.db, &table, op)).collect::<Vec<_>>().into()
> }
>
> // the keys a call is
> fn call_keys(call: &Call) -> Keys<'_> {
>     let field = |name: &str| call.input[name].as_str().unwrap_or_default().to_owned();
>     let (db, table, op) = (field("db"), field("table"), field("op"));
>     [table.as_str(), "*"]
>         .into_iter()
>         .flat_map(|t| [op.as_str(), "*"].map(|o| row_key(&db, t, o)))
>         .collect::<Vec<_>>()
>         .into()
> }
>
> /// The database connections, their config, and the capture installed per table.
> #[derive(Clone)]
> struct NativeCapture { /* ... */ }
> impl AddedHook for NativeCapture {
>     async fn on_added(&self, obs: Observer) -> Result<(), ObserveRejection> {
>         let args: RowChangedArgs = obs.args()?;
>         if !self.config.has(&args.db) {
>             let message = format!("no database '{}'", args.db);
>             return Err(ObserveRejection::new("UNKNOWN_DB", message));
>         }
>         if self.config.native(&args.db) {
>             self.install(&args.db, args.table.as_deref()).await; // CREATE TRIGGER ... pg_notify(...)
>         }
>         Ok(())
>     }
> }
>
> #[derive(JsonSchema)]
> struct RowChangeArgs {
>     db: String,
>     table: String,
>     op: String,
> impl RemovedHook for NativeCapture {
>     async fn on_removed(&self, obs: Observer) {
>         if let Ok(args) = obs.args::<RowChangedArgs>() {
>             // drops the database trigger with its last observer
>             self.release(&args.db, args.table.as_deref()).await;
>         }
>     }
> }
>
> let row_changed = worker.register_function(
>     "database::on-row-changed",
>     RegisterFunction::new(|change: RowChange| Ok::<_, Error>(change))
>         .private()
>         // the args an observer of this function passes
>         .args_format::<RowChangeArgs>(),
> );
>
> // called for each NOTIFY on the worker's LISTEN connection
> async fn publish_row_change(
>     worker: &IIIClient,
>     row_changed: &FunctionRef,
>     db: &str,
>     table: &str,
>     op: &str,
>     row: Value,
> ) -> Result<(), Error> {
>     if !row_changed.is_observed() {
>         return Ok(()); // no observers: skip the call
>     }
>     let change = json!({ "db": db, "table": table, "op": op, "row": row });
>     worker.call(CallRequest::new("database::on-row-changed", change)).await?;
>     Ok(())
> }
>
> // from any handler of this worker, for each changed row
> publish_row_change(&worker, &row_changed, "primary", "orders", "update", json!({ "id": 42, "status": "paid" }))
>     .await?;
> ```

> [tab] Go
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> import (
>     "context"
>     "encoding/json"
> )
>
> type rowChange struct {
>     DB    string          `json:"db"`
>     Table string          `json:"table"`
>     Op    string          `json:"op"`
>     Row   json.RawMessage `json:"row"`
> }
>
> rowChanged, _ := worker.RegisterFunction("database::on-row-changed", iii.Identity, iii.FunctionOptions{
>     Private: true,
>     // the args an observer of this function passes
>     ArgsFormat: json.RawMessage(`{"type":"object","properties":{"db":{"type":"string"},"table":{"type":"string"},"op":{"type":"string"}},"required":["db","table","op"]}`),
> })
>
> // called for each NOTIFY on the worker's LISTEN connection
> publishRowChange := func(ctx context.Context, change rowChange) error {
>     if !rowChanged.IsObserved() {
>         return nil // no observers: skip the call
>     }
>     payload, _ := json.Marshal(change)
>     _, err := worker.Call(ctx, iii.CallRequest{FunctionID: "database::on-row-changed", Payload: payload})
>     return err
> }
>
> // from any handler of this worker, for each changed row
> _ = publishRowChange(ctx, rowChange{DB: "primary", Table: "orders", Op: "update", Row: json.RawMessage(`{"id":42,"status":"paid"}`)})
> ```

Because the worker calls its own function, the call runs in-process
([local dispatch](https://iii.dev/docs/next/using-iii/observers.md#local-dispatch)) and reaches the engine only when an
observer's filters match.

### Send an observer its initial state

A source sends one observer something of its own, such as its current state when the observer is
added or an event it missed, by calling through that observer. The call runs that observer's target
and no other.

`obs.accepted()` resolves once the SDK has acknowledged the observer. A call made after it resolves
reaches an active observer, since the engine handles a connection's messages in order, as long as
the observer's target is registered: an observer whose target is not registered yet stays `pending`,
and the call fails with `OBSERVER_NOT_ACTIVE`. The observer becomes active after its
`onObserverAdded` returns, so the hook starts a task that waits for `accepted()` and then makes the
call. `accepted()` fails for an observer rejected first, such as one whose `onObserverAdded` throws,
and the task then sends nothing.

The target receives the payload as given, with the observer's registration metadata. Send what your
function's targets expect.

> [tab] Node / TypeScript
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> import { CallAction } from "iii-sdk";
> import { Pool } from "pg";
>
> const pools = new Map([["primary", new Pool({ connectionString: process.env.PRIMARY_DATABASE_URL })]]);
>
> worker.onObserverAdded("database::on-row-changed", (obs) => {
>   // send the table's row count once the observer is active
>   void obs
>     .accepted()
>     .then(async () => {
>       const { db, table } = obs.args as { db: string; table: string };
>       const pool = pools.get(db);
>       const counted = pool ? await pool.query(`SELECT count(*)::int AS n FROM ${table}`) : undefined;
>       await worker.call({
>         observer: obs,
>         payload: { table, rows: counted ? counted.rows[0].n : 0 },
>         action: CallAction.Void(),
>       });
>     })
>     .catch(() => {
>       // a rejected observer gets nothing
>     });
> });
> ```

> [tab] Python
>
> ```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> import asyncio
> import os
>
> from iii import CallAction, ObserveRejection
> from psycopg_pool import ConnectionPool
>
> pools = {"primary": ConnectionPool(os.environ["PRIMARY_DATABASE_URL"])}
>
> async def send_row_count(obs):
>     try:
>         await obs.accepted_async()
>     except ObserveRejection:
>         return  # a rejected observer gets nothing
>     db, table = obs.args["db"], obs.args["table"]
>     rows = 0
>     pool = pools.get(db)
>     if pool is not None:
>         with pool.connection() as conn:
>             rows = conn.execute(f"SELECT count(*) FROM {table}").fetchone()[0]
>     await worker.call_async({
>         "observer": obs,
>         "payload": {"table": table, "rows": rows},
>         "action": CallAction.Void(),
>     })
>
> @worker.on_observer_added("database::on-row-changed")
> async def on_added(obs):
>     # send the table's row count once the observer is active
>     asyncio.create_task(send_row_count(obs))
> ```

> [tab] Rust
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> use std::collections::HashMap;
> use std::sync::Arc;
>
> use iii_sdk::{CallAction, CallRequest, Observer};
> use serde_json::json;
> use sqlx::PgPool;
>
> let pools: Arc<HashMap<String, PgPool>> = Arc::new(HashMap::from([(
>     "primary".to_string(),
>     PgPool::connect(&std::env::var("PRIMARY_DATABASE_URL")?).await?,
> )]));
>
> let iii = worker.clone();
> worker.on_observer_added_for("database::on-row-changed", move |obs: Observer| {
>     let (iii, pools) = (iii.clone(), pools.clone());
>     async move {
>         // send the table's row count once the observer is active
>         tokio::spawn(async move {
>             if obs.accepted().await.is_err() {
>                 return; // a rejected observer gets nothing
>             }
>             let args = obs.args_value();
>             let (db, table) = (
>                 args["db"].as_str().unwrap_or_default(),
>                 args["table"].as_str().unwrap_or_default(),
>             );
>             let rows = match pools.get(db) {
>                 Some(pool) => sqlx::query_scalar::<_, i64>(&format!("SELECT count(*) FROM {table}"))
>                     .fetch_one(pool)
>                     .await
>                     .unwrap_or(0),
>                 None => 0,
>             };
>             let payload = json!({ "table": table, "rows": rows });
>             let _ = iii
>                 .call(CallRequest::observer(&obs, payload).action(CallAction::Void))
>                 .await;
>         });
>         Ok(())
>     }
> });
> ```

> [tab] Go
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> import (
>     "context"
>     "database/sql"
>     "encoding/json"
>     "os"
>
>     _ "github.com/jackc/pgx/v5/stdlib"
> )
>
> primary, _ := sql.Open("pgx", os.Getenv("PRIMARY_DATABASE_URL"))
> pools := map[string]*sql.DB{"primary": primary}
>
> worker.OnObserverAddedFor("database::on-row-changed", func(ctx context.Context, obs *iii.Observer) error {
>     // send the table's row count once the observer is active
>     go func() {
>         if obs.Accepted(context.Background()) != nil {
>             return // a rejected observer gets nothing
>         }
>         db, _ := obs.Args["db"].(string)
>         table, _ := obs.Args["table"].(string)
>         rows := 0
>         if pool, ok := pools[db]; ok {
>             _ = pool.QueryRowContext(context.Background(), "SELECT count(*) FROM "+table).Scan(&rows)
>         }
>         payload, _ := json.Marshal(map[string]any{"table": table, "rows": rows})
>         worker.Call(context.Background(), iii.CallRequest{
>             Observer: obs,
>             Payload:  payload,
>             Action:   iii.VoidAction(),
>         })
>     }()
>     return nil
> })
> ```

## Call a target through its observer

iii delivers a call to every matched observer and does not wait for the targets. When the worker
that hosts a function must reach one observer on its own, it makes a call that names that observer.
The call runs that observer's target and resolves to the target's result.

In one line, the call is:

> [tab] Node / TypeScript
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> const response = await worker.call({ observer: obs, payload: request }); // the target's result
> ```

> [tab] Python
>
> ```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> response = worker.call({"observer": obs, "payload": request})  # the target's result
> ```

> [tab] Rust
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> let response = iii.call(CallRequest::observer(&obs, request)).await?; // the target's result
> use iii_sdk::protocol::CallRequest;
> use iii_sdk::{
>     AddedHook, Call, DeliverHook, Error, ObserveRejection, Observer, RegisterFunction,
>     RemovedHook,
> };
> use serde_json::{json, Value};
>
> /// The route table, shared by its hooks, and the client to call targets with.
> #[derive(Clone)]
> struct Routes { /* table, worker: IIIClient */ }
> impl AddedHook for Routes {
>     async fn on_added(&self, obs: Observer) -> Result<(), ObserveRejection> {
>         self.add(obs) // HTTP_ROUTE_CONFLICT
>     }
> }
> impl RemovedHook for Routes {
>     async fn on_removed(&self, obs: Observer) {
>         self.remove(obs.id());
>     }
> }
> impl DeliverHook for Routes {
>     async fn on_deliver(&self, call: Call, _observers: Vec<Observer>) -> Result<Value, Error> {
>         let request: HttpRequest = serde_json::from_value(call.input.clone())
>             .map_err(|e| Error::Handler(e.to_string()))?;
>         let Some(route) = self.matches(&request.method, &request.path) else {
>             return Ok(json!({ "status_code": 404 }));
>         };
>         if !route.obs.matches(&call) {
>             return Ok(json!({ "status_code": 422 }));
>         }
>         run_middleware(&route, &request).await?;
>         let payload = json!(HttpRequest { path_params: route.params.clone(), ..request });
>         self.worker.call(CallRequest::observer(&route.obs, payload)).await
>     }
> }
>
> // by id, before registering, so the hooks see every observer
> worker.on_observer_added_for("http::on-request", routes.clone());
> worker.on_observer_removed_for("http::on-request", routes.clone());
> worker.on_observation_deliver_for("http::on-request", routes.clone());
>
> worker.register_function(
>     "http::on-request",
>     RegisterFunction::new(|request: HttpRequest| Ok::<_, Error>(request))
>         .args_format::<HttpRouteArgs>()
>         .private(),
> );
>
> // each request is a call of http::on-request, and its result is the response
> let request = json!(to_http_request(req));
> let response = worker.call(CallRequest::new("http::on-request", request)).await?;
> ```

> [tab] Go
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> response, err := worker.Call(ctx, iii.CallRequest{Observer: obs, Payload: request}) // the target's result
> ```

The http worker does this for every request. `http::on-request` returns its input, one observer with
`{ method, path }` args declares one route, and the deliver hook picks the route of the request and
calls through that route's observer. The target's return value is the response. The full flow:

> [tab] Node / TypeScript
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> import { ObserveRejection, registerWorker } from "iii-sdk";
>
> const worker = registerWorker("ws://127.0.0.1:49134", { workerName: "http" });
>
> type RouteArgs = { method: string; path: string };
> const routes = new Map<string, string>(); // "GET /orders/:id" -> observer id
> const routeKey = (args: RouteArgs) => `${args.method} ${args.path}`;
>
> // one call is one request: the handler returns its input
> worker.registerFunction("http::on-request", (request: unknown) => request, {
>   args_format: {
>     type: "object",
>     properties: { method: { type: "string" }, path: { type: "string" } },
>     required: ["method", "path"],
>   },
>   private: true,
> });
>
> worker.onObserverAdded("http::on-request", (obs) => {
>   const key = routeKey(obs.args as RouteArgs);
>   if (routes.has(key)) throw new ObserveRejection("HTTP_ROUTE_CONFLICT", `${key} is taken`);
>   routes.set(key, obs.id);
> });
>
> worker.onObserverRemoved("http::on-request", (obs) => {
>   for (const [key, id] of routes) if (id === obs.id) routes.delete(key);
> });
>
> worker.onObservationDeliver("http::on-request", async (call, observers) => {
>   const request = call.input as RouteArgs;
>   const obs = observers.find((o) => o.id === routes.get(routeKey(request)));
>   if (!obs) return { status_code: 404, body: "NOT_FOUND" };
>   if (!obs.matches(call)) return { status_code: 422, body: "FILTERS_NOT_MET" };
>   return worker.call({ observer: obs, payload: call.input }); // the target's result
> });
>
> // the server: one call per request, and its result is the HTTP response
> const response = await worker.call({
>   function_id: "http::on-request",
>   payload: { method: "GET", path: "/orders/:id" },
> });
> ```

> [tab] Python
>
> ```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> import os
>
> from iii import ObserveRejection, register_worker
>
> worker = register_worker(os.environ["III_URL"])
>
> routes: dict[str, str] = {}  # "GET /orders/:id" -> observer id
>
> def route_key(args):
>     return f"{args['method']} {args['path']}"
>
> ROUTE_ARGS = {
>     "type": "object",
>     "properties": {"method": {"type": "string"}, "path": {"type": "string"}},
>     "required": ["method", "path"],
> }
>
> # one call is one request: the handler returns its input
> worker.register_function("http::on-request", lambda request: request, args_format=ROUTE_ARGS, private=True)
>
> @worker.on_observer_added("http::on-request")
> def add_route(obs):
>     key = route_key(obs.args)
>     if key in routes:
>         raise ObserveRejection("HTTP_ROUTE_CONFLICT", f"{key} is taken")
>     routes[key] = obs.id
>
> @worker.on_observer_removed("http::on-request")
> def remove_route(obs):
>     for key, obs_id in list(routes.items()):
>         if obs_id == obs.id:
>             del routes[key]
>
> @worker.on_observation_deliver("http::on-request")
> def deliver(call, observers):
>     obs = next((o for o in observers if o.id == routes.get(route_key(call.input))), None)
>     if obs is None:
>         return {"status_code": 404, "body": "NOT_FOUND"}
>     if not obs.matches(call):
>         return {"status_code": 422, "body": "FILTERS_NOT_MET"}
>     return worker.call({"observer": obs, "payload": call.input})  # the target's result
>
> # the server: one call per request, and its result is the HTTP response
> response = worker.call(
>     {"function_id": "http::on-request", "payload": {"method": "GET", "path": "/orders/:id"}}
> )
> ```

> [tab] Rust
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> use std::collections::HashMap;
> use std::sync::Arc;
>
> use iii_sdk::{
>     AddedHook, Call, CallRequest, DeliverHook, Error, IIIClient, ObserveRejection, Observer,
>     RegisterFunction, RemovedHook,
> };
> use serde_json::{json, Value};
> use tokio::sync::RwLock;
>
> #[derive(serde::Deserialize, schemars::JsonSchema)]
> struct RouteArgs {
>     method: String,
>     path: String,
> }
>
> struct Hooks {
>     routes: Arc<RwLock<HashMap<String, String>>>, // "GET /orders/:id" -> observer id
>     iii: Arc<IIIClient>,
> }
> impl AddedHook for Hooks {
> impl AddedHook for Subscribers {
>     async fn on_added(&self, obs: Observer) -> Result<(), ObserveRejection> {
>         let args: RouteArgs = obs
>             .args()
>             .map_err(|e| ObserveRejection::new("INVALID_ARGS", e.to_string()))?;
>         let key = format!("{} {}", args.method, args.path);
>         let mut routes = self.routes.write().await;
>         if routes.contains_key(&key) {
>             return Err(ObserveRejection::new(
>                 "HTTP_ROUTE_CONFLICT",
>                 format!("{key} is taken"),
>             ));
>         }
>         routes.insert(key, obs.id().to_string());
>         Ok(())
>     }
> }
> impl RemovedHook for Hooks {
> impl RemovedHook for Subscribers {
>     async fn on_removed(&self, obs: Observer) {
>         self.routes.write().await.retain(|_, id| id != obs.id());
>     }
> }
> impl DeliverHook for Hooks {
> impl DeliverHook for Subscribers {
>     async fn on_deliver(&self, call: Call, observers: Vec<Observer>) -> Result<Value, Error> {
>         let key = format!(
>             "{} {}",
>             call.input["method"].as_str().unwrap_or_default(),
>             call.input["path"].as_str().unwrap_or_default()
>         );
>         let id = self.routes.read().await.get(&key).cloned();
>         let Some(obs) = id.and_then(|id| observers.into_iter().find(|o| o.id() == id)) else {
>             return Ok(json!({ "status_code": 404, "body": "NOT_FOUND" }));
>         };
>         if !obs.matches(&call) {
>             return Ok(json!({ "status_code": 422, "body": "FILTERS_NOT_MET" }));
>         }
>         // the route target's result is the response
>         self.iii.call(CallRequest::observer(&obs, call.input)).await
>     }
> }
>
> // start-up: the hooks are set before the function registers, so they run for every observer
> let hooks = Hooks {
>     routes: Arc::new(RwLock::new(HashMap::new())),
>     iii: iii.clone(),
> };
> iii.on_observer_added_for("http::on-request", hooks.clone());
> iii.on_observer_removed_for("http::on-request", hooks.clone());
> iii.on_observation_deliver_for("http::on-request", hooks);
> iii.register_function(
>     "http::on-request",
>     RegisterFunction::new(|request: Value| Ok::<_, Error>(request))
>         .args_format::<RouteArgs>()
>         .private(),
> );
>
> // the server: one call per request, and its result is the HTTP response
> let response = iii
>     .call(CallRequest::new(
>         "http::on-request",
>         json!({ "method": "GET", "path": "/orders/:id" }),
>     ))
>     .await?;
> ```

> [tab] Go
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> package main
>
> import (
>     "context"
>     "encoding/json"
>
>     iii "github.com/iii-hq/iii/sdk/packages/go/iii"
> )
>
> type routeArgs struct {
>     Method string `json:"method"`
>     Path   string `json:"path"`
> }
>
> func main() {
>     worker := iii.RegisterWorker(iii.DefaultEngineURL)
>     routes := map[string]string{} // "GET /orders/:id" -> observer id
>     key := func(a routeArgs) string { return a.Method + " " + a.Path }
>
>     // one call is one request: the handler returns its input
>     _, _ = worker.RegisterFunction("http::on-request", iii.Identity, iii.RegisterFunctionOptions{
>         ArgsFormat: json.RawMessage(`{"type":"object","properties":{"method":{"type":"string"},"path":{"type":"string"}},"required":["method","path"]}`),
>         Private:    true,
>     })
>
>     worker.OnObserverAddedFor("http::on-request", func(ctx context.Context, obs *iii.Observer) error {
>         var args routeArgs
>         if err := obs.DecodeArgs(&args); err != nil {
>             return err
>         }
>         k := key(args)
>         if _, taken := routes[k]; taken {
>             return iii.ObserveRejection("HTTP_ROUTE_CONFLICT", k+" is taken")
>         }
>         routes[k] = obs.ID
>         return nil
>     })
>
>     worker.OnObserverRemovedFor("http::on-request", func(ctx context.Context, obs *iii.Observer) {
>         for k, id := range routes {
>             if id == obs.ID {
>                 delete(routes, k)
>             }
>         }
>     })
>
>     worker.OnObservationDeliverFor("http::on-request", func(ctx context.Context, call *iii.Call, observers []*iii.Observer) (any, error) {
>         var req routeArgs
>         _ = json.Unmarshal(call.Input, &req)
>         for _, obs := range observers {
>             if obs.ID != routes[key(req)] {
>                 continue
>             }
>             if !obs.Matches(call) {
>                 return map[string]any{"status_code": 422, "body": "FILTERS_NOT_MET"}, nil
>             }
>             // the route target's result is the response
>             return worker.Call(ctx, iii.CallRequest{Observer: obs, Payload: call.Input})
>         }
>         return map[string]any{"status_code": 404, "body": "NOT_FOUND"}, nil
>     })
>
>     // the server: one call per request, and its result is the HTTP response
>     response, err := worker.Call(context.Background(), iii.CallRequest{
>         FunctionID: "http::on-request",
>         Payload:    json.RawMessage(`{"method":"GET","path":"/orders/:id"}`),
>     })
>     _ = response
>     _ = err
> }
> ```

Only the worker that hosts the observer's function may make such a call. The observer's policy,
metadata and the loop guard still apply.

## Registration options

`registerFunction` takes the options that describe a function to the engine; the
[SDK references](https://iii.dev/docs/next/reference/sdk-node.md) list them. Three of them change what observation means:

### Private functions

An event source is observed by everyone and called only by its own worker. A call from another
worker would be a forged request or row change carrying the source's identity. `private: true`
expresses that:

* The engine accepts a call of a private function only from the connection that hosts it. Any other
  connection, and the engine itself, gets `FORBIDDEN`. A call through an observer is the observing
  worker's call, and an activation is the engine's.
* A private function is observable, and its observers receive every call its worker makes.
* A private function cannot be an observer's target (`FORBIDDEN` at `observe`).
* A worker's call to its own private function passes the input and the result by reference. No other
  worker can read either value, so the SDK copies neither, and the handler and hooks treat both as
  read-only. With `NODE_ENV=development`, the Node SDK freezes them, so a write throws. An
  observer's target receives its own copy, serialized when the call matches.

Any function can be private; event sources are the common case.

### Prevent function observation

Some functions must stay unobserved, for example because their input carries credentials. Register
them with `observable: false`. Every `observe` of them fails with `NOT_OBSERVABLE`.

### One host per function

A function id has one host per namespace, the worker that registers it and runs its handler (see
[Namespaces](https://iii.dev/docs/next/using-iii/namespaces.md)). Observer hooks therefore run once per observer, in that
host.

When the host disconnects, the observers of its functions go `pending`. The next worker that
registers the function receives those observers and runs its hooks for each one: the added hook, or
the worker-level pair, so the setup runs again in the new host. An observer becomes active again
once its hook returns. Run one instance of a worker that hosts an event source, and let the platform
restart it: stop the old instance before the new one starts (`strategy: Recreate` in Kubernetes).

## The `Observer` object

Hooks get an `Observer`: the record of what the consumer asked for. Who observes, the observer's
metadata and its policy stay with the engine. The object stays the same for the full life of the
observer, so the worker can keep it, for example in a map.

| Member | Meaning |
| - | - |
| `id` | The observer id. |
| `function_id` | The observed function. |
| `args` | The observer's args, already validated against `args_format` by the engine. |
| `filters` | The observer's filters in [normal form](https://iii.dev/docs/next/reference/filters.md#normal-form). |
| `target` | The target function id. |
| `capture` | Which parts of the call its payload carries, and with them which calls the observer gets (see [`capture`](../using-iii/observers#capture)). |
| `matches(call)` | Evaluates the observer's capture and filters on a call. The outcomes that fire it come from the engine's `fires_on`, or, from an engine without `observe.capture`, from the deprecated legacy `on`. |
| `accepted()` | Resolves once the SDK has acknowledged the observer. See [Send an observer its initial state](#send-an-observer-its-initial-state). |
| `reject(code, message)` | Late rejection of an active observer. See [Reject an observer later](#reject-an-observer-later). |

To reject an observer from `onObserverAdded`, throw `new ObserveRejection(code, message)`.

## The `Call` object

`onObservationMatch` and `onObservationDeliver` receive the call:

| Field | Meaning |
| - | - |
| `function_id` | The function that was called. |
| `invocation_id` | The id of the call. |
| `status` | `success` or `error`. |
| `input` | The call's input. |
| `output` | The handler's return value: what observers observe. |
| `error` | `{ code, message }`, what the handler threw. |
| `caller` | Who made the call. See [Caller and causality](https://iii.dev/docs/next/using-iii/observers.md#caller-and-causality). |
| `metadata` | The call's metadata. |


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