Observable functions

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.

Every function your worker registers can be observed unless you specify otherwise (see 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). 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.

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.

Node / TypeScript
Python
Rust
Go

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()`,
  );
});

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()"
        )

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(())
    }
});

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.

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.

Node / TypeScript
Python
Rust
Go

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}`);
});

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}")

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;
        }
    }
});

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.

Node / TypeScript
Python
Rust
Go

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;
});

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

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();
    }
});

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):

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.

Node / TypeScript
Python
Rust
Go

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;
});

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

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();
    }
});

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) 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. Removing a hook returns the function to its default (see 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.

Node / TypeScript
Python
Rust
Go

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}'`);
});

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}'")

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}'")))
});

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.

Node / TypeScript
Python
Rust
Go

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}'`);
    }
  }
}

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}'")

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}'"));
        }
    }
}

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.

An observer’s args and the call settle the choice in onObservationMatch. 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.

Node / TypeScript
Python
Rust
Go

worker.onObservationMatch("database::on-row-changed", (obs, call) =>
  // one observation: this observer, this call
  obs.args.table === (call.input as { table: string }).table,
);

worker.on_observation_match(
    "database::on-row-changed",
    # one observation: this observer, this call
    lambda obs, call: obs.args["table"] == call.input["table"],
)

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"]),
);

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.

Node / TypeScript
Python
Rust
Go

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
});

@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

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
    }
});

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.

Node / TypeScript
Python
Rust
Go

worker.onObservationMatch(
  "database::on-row-changed",
  (obs, call) => obs.args.db === call.input.db && obs.args.table === call.input.table,
);

worker.on_observation_match(
    "database::on-row-changed",
    lambda obs, call: obs.args["db"] == call.input["db"] and obs.args["table"] == call.input["table"],
)

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"]
    }),
);

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

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.

Node / TypeScript
Python
Rust
Go

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" });

# 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"})

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?;

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

Node / TypeScript
Python
Rust
Go

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
    });
});

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))

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(())
    }
});

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:

Node / TypeScript
Python
Rust
Go

const response = await worker.call({ observer: obs, payload: request }); // the target's result

response = worker.call({"observer": obs, "payload": request})  # the target's result

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?;

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:

Node / TypeScript
Python
Rust
Go

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" },
});

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"}}
)

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?;

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 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). 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.
`target`	The target function id.
`capture`	Which parts of the call its payload carries, and with them which calls the observer gets (see `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.
`reject(code, message)`	Late rejection of an active observer. See 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.
`metadata`	The call’s metadata.

Getting Started

Using iii

Understanding iii

Creating Workers

Upgrading

Observable functions

Hooks

Observer lifecycle

`onObserverAdded(obs)`

`onObserverRemoved(obs)`

Set a hook for one function or for every function

Reject an observer

No replay of past executions

Reject an observer later

Observation lifecycle

`onObservationMatch(obs, call)`

`onObservationDeliver(call, observers)`

Match observers to calls

Producing events with observers

Send an observer its initial state

Call a target through its observer

Registration options

Private functions

Prevent function observation

One host per function

The `Observer` object

The `Call` object

​Hooks

​Observer lifecycle

​onObserverAdded(obs)

​onObserverRemoved(obs)

​Set a hook for one function or for every function

​Reject an observer

​No replay of past executions

​Reject an observer later

​Observation lifecycle

​onObservationMatch(obs, call)

​onObservationDeliver(call, observers)

​Match observers to calls

​Producing events with observers

​Send an observer its initial state

​Call a target through its observer

​Registration options

​Private functions

​Prevent function observation

​One host per function

​The Observer object

​The Call object

Hooks

Observer lifecycle

`onObserverAdded(obs)`

`onObserverRemoved(obs)`

Set a hook for one function or for every function

Reject an observer

No replay of past executions

Reject an observer later

Observation lifecycle

`onObservationMatch(obs, call)`

`onObservationDeliver(call, observers)`

Match observers to calls

Producing events with observers

Send an observer its initial state

Call a target through its observer

Registration options

Private functions

Prevent function observation

One host per function

The `Observer` object

The `Call` object