Observe

import { registerWorker, type Observed } from "iii-sdk";

const iii = registerWorker(process.env.III_URL ?? "ws://localhost:49134");

iii.registerFunction("alerts::payment-failed", async (event: Observed<{ amount: number }>) => {
  console.log(event.function_id, event.error?.code, event.input?.amount, event.caller?.worker_name);
  return null; // ignored for auto delivery
});

const failures = iii.observe(
  {
    function_id: "payments::charge",
    on: "error",
    filters: { "input.amount": { $gte: 500 } },
    capture: ["input", "error", "caller"],
  },
  "alerts::payment-failed",
  { max_in_flight: 4 },
);
await failures.ready(); // rejects with ObserveRejection (error.code) on error or disposed
// later: await failures.dispose()

iii.register_function("alerts::payment-failed", lambda event: print(event["error"]["code"]))

failures = iii.observe(
    {"function_id": "payments::charge", "on": "error",
     "filters": {"input.amount": {"$gte": 500}}, "capture": ["input", "error", "caller"]},
    "alerts::payment-failed",
    {"max_in_flight": 4},
)
failures.ready()  # blocks; in async code: await failures.ready_async()

use iii_sdk::{Capture, Error, ObserverOptions, Observed, On, RegisterFunction, Source};
use serde_json::{json, Value};

iii.register_function(
    "alerts::payment-failed",
    RegisterFunction::new(|event: Observed<Value>| {
        println!("{} failed: {:?}", event.function_id, event.error);
        Ok::<_, Error>(())
    }),
);
let failures = iii.observe_with(
    Source::function("payments::charge")
        .on(On::Error)
        .filters(json!({ "input.amount": { "$gte": 500 } }))
        .capture([Capture::Input, Capture::Error, Capture::Caller]),
    "alerts::payment-failed",
    ObserverOptions::default().max_in_flight(4),
);
failures.ready().await?;

client.RegisterFunction("alerts::payment-failed", func(ctx context.Context, data json.RawMessage) (any, error) {
	var ev iii.Observed
	if err := json.Unmarshal(data, &ev); err != nil {
		return nil, err
	}
	log.Printf("%s failed: %s", ev.FunctionID, ev.Error.Code)
	return nil, nil
})
failures, err := client.Observe(iii.Source{
	FunctionID: "payments::charge",
	On:         iii.OnError,
	Filters:    iii.Filters{"input.amount": iii.Op{"$gte": 500}},
	Capture:    []iii.Capture{iii.CaptureInput, iii.CaptureError, iii.CaptureCaller},
}, "alerts::payment-failed", iii.ObserverOptions{MaxInFlight: 4})
if err == nil {
	err = failures.Ready(ctx) // *iii.ObserveRejectionError carries the code
}

{ "caller.parent.worker_name": { "$ne": "search-indexer" },
  "input.items": { "$any": { "@.sku": { "$prefix": "VIP-" } } } }

import { type Observer, ObserveRejection } from "iii-sdk";

const schedules = new Map<string, Set<Observer>>(); // expression -> observers

const onTick = iii.registerFunction(
  "cron::on-tick",
  async (tick: { expression: string; scheduled_time: string }) => {
    for (const obs of schedules.get(tick.expression) ?? []) obs.emit(tick); // false: filters said no
    return null;
  },
  {
    observe: {
      delivery: "owner",
      args_format: { type: "object", required: ["expression"], properties: { expression: { type: "string" } } },
      legacy_trigger_type: "cron", // optional: old registerTrigger({ type: 'cron' }) consumers keep working
      onObserverAdded(obs) {
        const expression = obs.args.expression as string;
        if (!isValidCron(expression)) throw new ObserveRejection("INVALID_CRON_EXPRESSION", expression); // your validator
        if (!schedules.has(expression)) schedules.set(expression, new Set());
        schedules.get(expression)!.add(obs); // index before the ack
      },
      onObserverRemoved(obs) {
        schedules.get(obs.args.expression as string)?.delete(obs);
      },
    },
  },
);
// your scheduler fires: iii.trigger({ function_id: "cron::on-tick", payload: tick }) (runs in-process)
// onTick.isObserved() is one load: skip building events nobody receives

from iii import ObserveHooks, ObserveRejection

schedules: dict[str, set] = {}  # expression -> observers

def on_tick(tick):
    for obs in list(schedules.get(tick["expression"], ())):
        obs.emit(tick)  # False when the observer's filters reject the tick

def added(obs):  # sync or async; raise ObserveRejection(code, message) to refuse
    schedules.setdefault(obs.args["expression"], set()).add(obs)

def removed(obs):
    schedules.get(obs.args["expression"], set()).discard(obs)

iii.register_function("cron::on-tick", on_tick, observe=ObserveHooks(
    delivery="owner",
    args_format={"type": "object", "required": ["expression"],
                 "properties": {"expression": {"type": "string"}}},
    on_observer_added=added,
    on_observer_removed=removed,
))

use iii_sdk::{Error, Observe, ObserveHandler, ObserveRejection, Observer, RegisterFunction};
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use std::{collections::HashMap, sync::{Arc, Mutex}};

#[derive(Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
struct CronArgs { expression: String }

#[derive(Serialize, Deserialize, JsonSchema)]
struct CronTick { expression: String, scheduled_time: String }

#[derive(Clone, Default)]
struct Schedules(Arc<Mutex<HashMap<String, Vec<Observer>>>>); // expression -> observers

#[async_trait::async_trait] // ObserveHandler needs the async-trait crate
impl ObserveHandler for Schedules {
    async fn on_observer_added(&self, obs: &Observer) -> Result<(), ObserveRejection> {
        let CronArgs { expression } = obs.args()?; // validated by the engine
        self.0.lock().unwrap().entry(expression).or_default().push(obs.clone());
        Ok(())
    }
    async fn on_observer_removed(&self, obs: &Observer) {
        for list in self.0.lock().unwrap().values_mut() { list.retain(|o| o.id() != obs.id()); }
    }
}

let schedules = Schedules::default();
iii.register_function(
    "cron::on-tick",
    RegisterFunction::new({
        let schedules = schedules.clone();
        move |tick: CronTick| {
            let observers = schedules.0.lock().unwrap().get(&tick.expression).cloned();
            for obs in observers.unwrap_or_default() { obs.emit(&tick)?; } // Ok(false): filters said no
            Ok::<_, Error>(())
        }
    })
    .observe(Observe::owner().args_format::<CronArgs>().handler(schedules)),
);

// onTick is the function's handler; index is your expression -> observers map.
client.RegisterFunction("cron::on-tick", onTick, iii.RegisterFunctionOptions{Observe: &iii.ObserveHooks{
	Delivery:   iii.DeliveryOwner,
	ArgsFormat: json.RawMessage(`{"type":"object","required":["expression"],"properties":{"expression":{"type":"string"}}}`),
	OnObserverAdded: func(ctx context.Context, obs *iii.Observer) error {
		expr, _ := obs.Args["expression"].(string)
		return index.Add(expr, obs) // or iii.ObserveRejection("INVALID_CRON_EXPRESSION", "...")
	},
	OnObserverRemoved: func(ctx context.Context, obs *iii.Observer) { index.Remove(obs) },
}})
// inside onTick: ok, err := obs.Emit(ctx, tick); result, err := obs.Call(ctx, payload)

iii.observe({ function_id: "engine::on-workers-changed", deliver: "input" }, "ops::on-worker");
iii.observe(
  { function_id: "configuration::on-updated", deliver: "input",
    filters: { "input.id": { $eq: "orders" }, "input.event_type": { $eq: "configuration:updated" } } },
  "orders::reload",
);
// observability::* does not resolve to default automatically: name it from a namespaced worker
iii.observe(
  { function_id: "observability::on-log", namespace: "default", deliver: "input",
    filters: { "input.severity_text": { $eq: "ERROR" } } },
  "alerts::log",
);