Use Functions & Triggers

</Tab>
<Tab title="Python">
```python title="math_add.py"
import os
import uuid

from iii import Logger, register_worker

iii = register_worker(os.environ.get("III_URL", "ws://localhost:49134"))


def add(data):
    logger = Logger()
    result = data["a"] + data["b"]
    id = str(uuid.uuid4())
    iii.trigger({"function_id": "state::set", "payload": {"scope": "math", "key": id, "value": result}})
    logger.info("Math add completed", {"id": id, "result": result})
    return {"id": id, "result": result}


iii.register_function({"id": "math::add"}, add)

# Triggerable from any other function or worker
iii.trigger({"function_id": "math::add", "payload": {"a": 2, "b": 3}})

    iii.trigger(TriggerRequest { function_id: "state::set".into(), payload: json!({ "scope": "math", "key": id, "value": result }), action: None, timeout_ms: None }).await?;

    logger.info("Math add completed", Some(json!({ "id": id, "result": result })));
    Ok(json!({ "id": id, "result": result }))
  }
},

</Tab>
</Tabs>

## HTTP-Invoked Functions

Instead of passing a handler, you can register an external HTTP endpoint as a function. The engine
makes the HTTP call when the function is triggered — no client-side HTTP code needed. This is useful
for delegating work to external services like webhooks, serverless functions, or third-party APIs.

<Warning title="Engine module required">
HTTP-invoked functions require `HttpFunctionsModule` to be enabled in your engine config.
See the [engine configuration guide](/how-to/configure-engine) for details.
</Warning>

Pass an `HttpInvocationConfig` as the second argument to `registerFunction` instead of a handler:

<Tabs>
<Tab title="Node / TypeScript">
```typescript title="http-invoked.ts"
import { registerWorker } from 'iii-sdk';

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

iii.registerFunction(
  {
    id: 'notifications::send',
    description: 'POST notification to Service Provider webhook',
  },
  {
    url: 'https://hooks.provider.example.com/notify',
    method: 'POST',
    timeout_ms: 5000,
    headers: { 'X-Service': 'iii-worker' },
    auth: {
      type: 'bearer',
      token_key: 'PROVIDER_API_TOKEN',
    },
  },
);

// Triggerable from any other function or worker
await iii.trigger({ function_id: 'notifications::send', payload: { channel: '#alerts', text: 'Deploy succeeded' } });

</Tab>
<Tab title="Rust">
```rust title="http_invoked.rs"
use iii_sdk::{register_worker, InitOptions, HttpAuthConfig, HttpInvocationConfig, HttpMethod, RegisterFunctionMessage, TriggerRequest};
use std::collections::HashMap;

let iii = register_worker("ws://localhost:49134", InitOptions::default());

let mut headers = HashMap::new();
headers.insert("X-Service".to_string(), "iii-worker".to_string());

iii.register_function(
  RegisterFunctionMessage {
    id: "notifications::send".into(),
    description: Some("POST notification to Service Provider webhook".into()),
    request_format: None, response_format: None, metadata: None, invocation: None,
  },
  HttpInvocationConfig {
    url: "https://hooks.provider.example.com/notify".to_string(),
    method: HttpMethod::Post,
    timeout_ms: Some(5000),
    headers,
    auth: Some(HttpAuthConfig::Bearer {
      token_key: "PROVIDER_API_TOKEN".to_string(),
    }),
  },
);

iii.trigger(TriggerRequest { function_id: "notifications::send".into(), payload: json!({ "channel": "#alerts", "text": "Deploy succeeded" }), action: None, timeout_ms: None }).await?;

</Tab>
<Tab title="Rust">
```rust title="trigger.rs"
let iii_clone = iii.clone();
iii.register_function(
  RegisterFunctionMessage { id: "math::calculate".into(), description: None, request_format: None, response_format: None, metadata: None, invocation: None },
  move |input| {
    let iii = iii_clone.clone();
    async move {
      let result = iii.trigger(TriggerRequest { function_id: "math::add".into(), payload: json!({"a": input["a"], "b": input["b"]}), action: None, timeout_ms: None }).await?;
      println!("{:?}", result); // {"result": 5}
      Ok(result)
    }
  },
);

</Tab>
<Tab title="Python">
```python title="trigger_void.py"
from iii import TriggerAction


def calculate_async(data):
    iii.trigger({"function_id": "math::add", "payload": {"a": data["a"], "b": data["b"]}, "action": TriggerAction.Void()})


iii.register_function({"id": "math::calculate-async"}, calculate_async)

</Tab>
</Tabs>

### Enqueue — `trigger` with `TriggerAction.Enqueue({ queue })`

Enqueue work to a named queue for async processing. The engine acknowledges with `{ messageReceiptId }` when the job is accepted. The target function receives the payload when a worker processes it. Requires queues defined in `iii-config.yaml` — see [Use Queues](/how-to/use-queues).

<Tabs>
<Tab title="Node / TypeScript">
```typescript title="trigger-enqueue.ts"
import { TriggerAction } from 'iii-sdk'

iii.registerFunction({ id: 'orders::create' }, async (input) => {
  const order = { id: crypto.randomUUID(), ...input }
  const result = await iii.trigger({
    function_id: 'orders::process-order',
    payload: order,
    action: TriggerAction.Enqueue({ queue: 'payment' }),
  })
  return { orderId: order.id, messageReceiptId: result.messageReceiptId }
})

</Tab>
<Tab title="Rust">
```rust title="trigger_enqueue.rs"
use iii_sdk::{RegisterFunctionMessage, TriggerAction, TriggerRequest};

let iii_clone = iii.clone();
iii.register_function(
  RegisterFunctionMessage { id: "orders::create".into(), description: None, request_format: None, response_format: None, metadata: None, invocation: None },
  move |input| {
    let iii = iii_clone.clone();
    async move {
      let order_id = uuid::Uuid::new_v4().to_string();
      let order = json!({ "id": order_id, "items": input["items"], "total": input["total"] });
      let result = iii.trigger(TriggerRequest {
        function_id: "orders::process-order".into(),
        payload: order,
        action: Some(TriggerAction::Enqueue { queue: "payment".into() }),
        timeout_ms: None,
      }).await?;
      Ok(json!({ "orderId": order_id, "messageReceiptId": result["messageReceiptId"] }))
    }
  },
);

</Tab>
<Tab title="Python">
```python title="http_trigger.py"
def multiply(req):
    logger = Logger()
    a, b = req["body"]["a"], req["body"]["b"]
    result = a * b
    logger.info("Math multiply", {"a": a, "b": b, "result": result})
    return {
        "status_code": 200,
        "body": {"result": result},
        "headers": {"Content-Type": "application/json"},
    }


iii.register_function({"id": "math::multiply"}, multiply)

iii.register_trigger({
    "type": "http",
    "function_id": "math::multiply",
    "config": {"api_path": "/math/multiply", "http_method": "POST"},
})

</Tab>
</Tabs>

#### Cron

This example aggregates all stored math results (from `math::add` above) every 30 minutes.

<Tabs>
<Tab title="Node / TypeScript">
```typescript title="cron-trigger.ts"
iii.registerFunction({ id: 'math::aggregation' }, async () => {
  const logger = new Logger();
  const results = await iii.trigger({ function_id: 'state::list', payload: { scope: 'math' } });
  const values = results.filter((r) => typeof r === 'number');
  const sum = values.reduce((a, b) => a + b, 0);
  const aggregation = { count: values.length, sum, average: values.length ? sum / values.length : 0 };
  logger.info('Math aggregation completed', aggregation);
  return aggregation;
});

iii.registerTrigger({
  type: 'cron',
  function_id: 'math::aggregation',
  config: { expression: '0 */30 * * * *' }, // every 30 minutes
});

</Tab>
<Tab title="Rust">
```rust title="cron_trigger.rs"
let iii_clone = iii.clone();
iii.register_function(
  RegisterFunctionMessage { id: "math::aggregation".into(), description: None, request_format: None, response_format: None, metadata: None, invocation: None },
  move |_| {
    let iii = iii_clone.clone();
    async move {
      let logger = Logger::new();
      let results = iii.trigger(TriggerRequest { function_id: "state::list".into(), payload: json!({ "scope": "math" }), action: None, timeout_ms: None }).await?;
      let values: Vec<i64> = results.as_array()
        .map(|arr| arr.iter().filter_map(|r| r.as_i64()).collect())
        .unwrap_or_default();
      let sum: i64 = values.iter().sum();
      let count = values.len();
      let avg = sum as f64 / count as f64;
      logger.info("Math aggregation completed", Some(json!({ "count": count, "sum": sum, "average": avg })));
      Ok(json!({ "count": count, "sum": sum, "average": avg }))
    }
  },
);

iii.register_trigger(RegisterTriggerInput {
  trigger_type: "cron".into(),
  function_id: "math::aggregation".into(),
  config: json!({ "expression": "0 */30 * * * *" }), // every 30 minutes
})?;

</Tab>
<Tab title="Python">
```python title="state_trigger.py"
from iii import HttpInvocationConfig

iii.register_function(
    {"id": "orders::webhook"},
    HttpInvocationConfig(url=os.environ["WEBHOOK_URL"], method="POST", timeout_ms=5000),
)

iii.register_trigger({
    "type": "state",
    "function_id": "orders::webhook",
    "config": {"scope": "orders", "key": "status"},
})

</Tab>
</Tabs>

{/* <Info title="Queue uses enqueue">
  Queue messaging uses `enqueue` — a built-in function, no trigger registration needed:
  Node `trigger({ function_id: 'enqueue', payload: { topic: 'user.created', data: {...} } })`,
  Python `trigger({'function_id': 'enqueue', 'payload': {'topic': 'user.created', 'data': {...}}})`,
  Rust `trigger(TriggerRequest::new("enqueue", json!({"topic": "user.created", "data": {...}})))`.
  See the [Queue module](/modules/module-queue) for details.
</Info> */}

## Trigger Types

| Type | Fires when | Config fields | Module |
|------|-----------|---------------|--------|
| `http` | HTTP request received | `api_path`, `http_method` | HTTP |
| `cron` | Schedule fires | `expression` | Cron |
| `queue` | Message published to a topic | `topic` | Queue |
| `subscribe` | PubSub message on a topic | `topic` | PubSub |
| `state` | State value changes | `scope`, `key` | State |
| `stream` | Stream value changes | `stream_name`, `group_id`, `item_id` | Stream |
| `stream:join` | Client connects to stream | — | Stream |
| `stream:leave` | Client disconnects from stream | — | Stream |
| `log` | Log entry emitted | `level` | Observability |

## Cross-language triggering

Any Function can be Triggered anywhere regardless of language. The engine handles serialization and routing:

<Tabs>
<Tab title="Node / TypeScript">
```typescript title="cross-language.ts"
iii.registerFunction({ id: 'math::double' }, async (input) => {
  const logger = new Logger();
  // Trigger a function that might be implemented in Python, Rust, or any other language
  const result = await iii.trigger({ function_id: 'math::add', payload: { a: input.value, b: input.value } })
  logger.info('Result', result) // { result: 10 } if input.value was 5
  return result
})

</Tab>
<Tab title="Rust">
```rust title="cross_language.rs"
let iii_clone = iii.clone();
iii.register_function(
  RegisterFunctionMessage { id: "math::double".into(), description: None, request_format: None, response_format: None, metadata: None, invocation: None },
  move |input| {
    let iii = iii_clone.clone();
    async move {
      // Trigger a function that might be implemented in Node, Python, or any other language
      let value = input["value"].as_i64().unwrap_or(0);
      let result = iii.trigger(TriggerRequest { function_id: "math::add".into(), payload: json!({"a": value, "b": value}), action: None, timeout_ms: None }).await?;
      println!("{:?}", result); // {"result": 10} if value was 5
      Ok(result)
    }
  },
);

</Tab>
<Tab title="Python">
```python title="unregister.py"
def create_order(data):
    return {"status_code": 201, "body": {"id": "123", "item": data["body"]["item"]}}


fn_ref = iii.register_function({"id": "orders::create"}, create_order)

trigger = iii.register_trigger({
    "type": "http",
    "function_id": "orders::create",
    "config": {"api_path": "/orders", "http_method": "POST"},
})

fn_ref.unregister()
trigger.unregister()

</Tab>
</Tabs>

## registerWorker, Function, and Trigger Registration is Synchronous

The `registerWorker()` call returns immediately rather than returning a Promise. This is intentional.
Connection establishment happens asynchronously in the background.

This design avoids requiring developers to wrap initialization in an async function or await the SDK before
registering functions and triggers. Without this, every function and trigger definition would need to
wait on initialization, adding boilerplate to every file.

The trade-off: if code calls `shutdown()` immediately after `registerWorker()`, the connection may not yet
be established, resulting in an error like `WebSocket was closed before the connection was established`.
In practice this rarely matters — most applications register functions, respond to triggers, and run indefinitely.

## Function IDs

Function IDs use a `namespace::name` convention but can be any arbitrary string. iii conventions recommend
following this rule but the iii engine does not enforce it.

<Warning title="iii prefix">
The `iii::` prefix is reserved for internal engine functions. Function IDs cannot start with `iii::`.
</Warning>

## Built-in Functions
{/* TODO: Break these out into modules once those docs are ready and then have short higher level
mentions with one or two noted functions each and then link to the detailed docs */}

The engine provides built-in functions that can be triggered the same way any other Function would be triggered. These handle common operations like state management, streaming, messaging, and observability.

### State

Persistent key-value storage with scoped namespaces.

| Function | Purpose | Parameters |
|----------|---------|------------|
| `state::set` | Store a value | `scope`, `key`, `value` |
| `state::get` | Retrieve a value | `scope`, `key` |
| `state::delete` | Remove a value | `scope`, `key` |
| `state::update` | Atomic update with operations | `scope`, `key`, `ops` |
| `state::list` | List all values in a scope | `scope` |
| `state::list_groups` | List all state scopes | — |

### Stream

Real-time data streams with hierarchical organization.

| Function | Purpose | Parameters |
|----------|---------|------------|
| `stream::set` | Set a value in a stream | `stream_name`, `group_id`, `item_id`, `data` |
| `stream::get` | Get a value from a stream | `stream_name`, `group_id`, `item_id` |
| `stream::delete` | Delete a value from a stream | `stream_name`, `group_id`, `item_id` |
| `stream::update` | Atomic update with operations | `stream_name`, `group_id`, `item_id`, `ops` |
| `stream::list` | List items in a group | `stream_name`, `group_id` |
| `stream::list_groups` | List groups in a stream | `stream_name` |
| `stream::list_all` | List all streams | — |
| `stream::send` | Send event to subscribers | `stream_name`, `group_id`, `id`, `event_type`, `data` |

### Queue

Durable message queue for async processing.

| Function | Purpose | Parameters |
|----------|---------|------------|
| `enqueue` | Enqueue a message | `topic`, `data` |

### PubSub

Publish-subscribe messaging for event broadcasting.

| Function | Purpose | Parameters |
|----------|---------|------------|
| `publish` | Publish an event to subscribers | `topic`, `data` |

### Engine

Internal engine functions for introspection and management.

<Info title="Engine module is always loaded">
The Engine module is mandatory and always loaded.
</Info>

| Function | Purpose | Parameters |
|----------|---------|------------|
| `engine::functions::list` | List all registered functions | `include_internal` (optional, default: `false`) |
| `engine::workers::list` | List all workers with metrics | `worker_id` (optional) |
| `engine::triggers::list` | List all triggers | `include_internal` (optional, default: `false`) |
| `engine::workers::register` | Register worker metadata (internal worker call) | `_caller_worker_id`, `runtime` (optional), `version` (optional), `name` (optional), `os` (optional), `telemetry` (optional) |
| `engine::channels::create` | Create a streaming channel pair | `buffer_size` (optional) |

### Observability

Logging, tracing, and metrics. The Observability module is disabled by default and must be enabled in engine config.

| Function | Purpose | Parameters |
|----------|---------|------------|
| `engine::log::info` | Log info message | `message`, `data` (optional), `service_name` (optional), `trace_id` (optional), `span_id` (optional) |
| `engine::log::warn` | Log warning message | `message`, `data` (optional), `service_name` (optional), `trace_id` (optional), `span_id` (optional) |
| `engine::log::error` | Log error message | `message`, `data` (optional), `service_name` (optional), `trace_id` (optional), `span_id` (optional) |
| `engine::log::debug` | Log debug message | `message`, `data` (optional), `service_name` (optional), `trace_id` (optional), `span_id` (optional) |
| `engine::log::trace` | Log trace message | `message`, `data` (optional), `service_name` (optional), `trace_id` (optional), `span_id` (optional) |
| `engine::baggage::get` | Get baggage item | `key` |
| `engine::baggage::set` | Set baggage item | `key`, `value` |
| `engine::baggage::get_all` | Get all baggage items | — |
| `engine::traces::list` | List stored traces | `trace_id` (optional), `service_name` (optional), `name` (optional), `status` (optional), `offset` (optional), `limit` (optional), `min_duration_ms` (optional), `max_duration_ms` (optional), `start_time` (optional), `end_time` (optional), `sort_by` (optional), `sort_order` (optional), `attributes` (optional), `include_internal` (optional) |
| `engine::traces::tree` | Get trace tree | `trace_id` |
| `engine::traces::clear` | Clear stored traces | — |
| `engine::metrics::list` | List current metrics | `start_time` (optional), `end_time` (optional), `metric_name` (optional), `aggregate_interval` (optional) |
| `engine::logs::list` | List stored logs | `start_time` (optional), `end_time` (optional), `trace_id` (optional), `span_id` (optional), `severity_min` (optional), `severity_text` (optional), `offset` (optional), `limit` (optional) |
| `engine::logs::clear` | Clear stored logs | — |
| `engine::health::check` | System health status | — |
| `engine::alerts::list` | List alert states | — |
| `engine::alerts::evaluate` | Manually evaluate alerts | — |
| `engine::rollups::list` | Get pre-aggregated metrics | `start_time` (optional), `end_time` (optional), `level` (optional), `metric_name` (optional) |
| `engine::sampling::rules` | Get sampling rules | — |