Observable functions

import threading
import time

from iii import ObserveHooks

timers: dict[int, dict] = {}  # every_ms → {"observers": {id: Observer}, "stop": Event}
lock = threading.Lock()

def run_timer(every_ms: int, stop: threading.Event) -> None:
    while not stop.wait(every_ms / 1000):
        worker.trigger({
            "function_id": "timer::on-tick",
            "payload": {"every_ms": every_ms, "fired_at_ms": int(time.time() * 1000)},
        })

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

    def on_observer_added(self, obs):
        every_ms = obs.args["every_ms"]
        with lock:
            timer = timers.get(every_ms)
            if timer is None:
                # First observer of this interval: start one timer for all of them.
                stop = threading.Event()
                threading.Thread(target=run_timer, args=(every_ms, stop), daemon=True).start()
                timer = timers[every_ms] = {"observers": {}, "stop": stop}
            timer["observers"][obs.id] = obs

    def on_observer_removed(self, obs):
        every_ms = obs.args["every_ms"]
        with lock:
            timer = timers.get(every_ms)
            if timer is None:
                return
            timer["observers"].pop(obs.id, None)
            if not timer["observers"]:
                # Last observer of this interval: stop its timer.
                timer["stop"].set()
                del timers[every_ms]

def on_tick(tick: dict) -> None:
    with lock:
        timer = timers.get(tick["every_ms"])
        observers = list(timer["observers"].values()) if timer else []
    for obs in observers:
        obs.emit(tick)  # False when the observer's filters reject the tick

worker.register_function(
    "timer::on-tick",
    on_tick,
    description="Fires every `every_ms` milliseconds. Observe it with args { every_ms }.",
    observe=TimerHooks(),
)

use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

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

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

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

/// every_ms → the observers of that interval and the task that fires it
#[derive(Clone)]
struct Timers {
    iii: IIIClient,
    by_interval: Arc<Mutex<HashMap<u64, (Vec<Observer>, tokio::task::JoinHandle<()>)>>>,
}

#[async_trait::async_trait]
impl ObserveHandler for Timers {
    async fn on_observer_added(&self, obs: &Observer) -> Result<(), ObserveRejection> {
        let TimerArgs { every_ms } = obs.args()?; // validated by the engine
        let mut timers = self.by_interval.lock().unwrap();
        let (observers, _) = timers.entry(every_ms).or_insert_with(|| {
            // First observer of this interval: start one task for all of them.
            let iii = self.iii.clone();
            let task = tokio::spawn(async move {
                let mut interval = tokio::time::interval(Duration::from_millis(every_ms));
                loop {
                    interval.tick().await;
                    let fired_at_ms = SystemTime::now()
                        .duration_since(UNIX_EPOCH)
                        .map(|d| d.as_millis() as u64)
                        .unwrap_or_default();
                    let tick = json!({ "every_ms": every_ms, "fired_at_ms": fired_at_ms });
                    let _ = iii.trigger(TriggerRequest::new("timer::on-tick", tick)).await;
                }
            });
            (Vec::new(), task)
        });
        observers.push(obs.clone());
        Ok(())
    }

    async fn on_observer_removed(&self, obs: &Observer) {
        self.by_interval.lock().unwrap().retain(|_, (observers, task)| {
            observers.retain(|o| o.id() != obs.id());
            if observers.is_empty() {
                task.abort(); // last observer of this interval: stop its task
            }
            !observers.is_empty()
        });
    }
}

let timers = Timers { iii: worker.clone(), by_interval: Default::default() };

worker.register_function(
    "timer::on-tick",
    RegisterFunction::new({
        let timers = timers.clone();
        move |tick: Tick| {
            let observers = timers
                .by_interval
                .lock()
                .unwrap()
                .get(&tick.every_ms)
                .map(|(observers, _)| observers.clone())
                .unwrap_or_default();
            for obs in observers {
                obs.emit(&tick)?; // Ok(false) when the observer's filters reject the tick
            }
            Ok::<_, Error>(())
        }
    })
    .description("Fires every `every_ms` milliseconds. Observe it with args { every_ms }.")
    .observe(Observe::owner().args_format::<TimerArgs>().handler(timers.clone())),
);

type tick struct {
    EveryMs   int64 `json:"every_ms"`
    FiredAtMs int64 `json:"fired_at_ms"`
}

type timer struct {
    observers map[string]*iii.Observer
    stop      chan struct{}
}

var (
    mu     sync.Mutex
    timers = map[int64]*timer{} // every_ms → observers and their timer
)

func runTimer(client *iii.Client, everyMs int64, stop chan struct{}) {
    ticker := time.NewTicker(time.Duration(everyMs) * time.Millisecond)
    defer ticker.Stop()
    for {
        select {
        case <-stop:
            return
        case now := <-ticker.C:
            data, _ := json.Marshal(tick{EveryMs: everyMs, FiredAtMs: now.UnixMilli()})
            client.Trigger(context.Background(), iii.TriggerRequest{FunctionID: "timer::on-tick", Data: data})
        }
    }
}

func everyMs(obs *iii.Observer) int64 {
    var args struct {
        EveryMs int64 `json:"every_ms"`
    }
    _ = obs.DecodeArgs(&args) // validated by the engine
    return args.EveryMs
}

func registerTimer(client *iii.Client) error {
    return client.RegisterFunction("timer::on-tick", func(ctx context.Context, data json.RawMessage) (any, error) {
        var t tick
        if err := json.Unmarshal(data, &t); err != nil {
            return nil, err
        }
        mu.Lock()
        var observers []*iii.Observer
        if tm := timers[t.EveryMs]; tm != nil {
            for _, obs := range tm.observers {
                observers = append(observers, obs)
            }
        }
        mu.Unlock()
        for _, obs := range observers {
            obs.Emit(ctx, t) // false when the observer's filters reject the tick
        }
        return nil, nil
    }, iii.RegisterFunctionOptions{Observe: &iii.ObserveHooks{
        Delivery:   iii.DeliveryOwner,
        ArgsFormat: json.RawMessage(`{"type":"object","additionalProperties":false,"required":["every_ms"],"properties":{"every_ms":{"type":"integer","minimum":100}}}`),
        OnObserverAdded: func(ctx context.Context, obs *iii.Observer) error {
            every := everyMs(obs)
            mu.Lock()
            defer mu.Unlock()
            tm := timers[every]
            if tm == nil {
                // First observer of this interval: start one timer for all of them.
                tm = &timer{observers: map[string]*iii.Observer{}, stop: make(chan struct{})}
                timers[every] = tm
                go runTimer(client, every, tm.stop)
            }
            tm.observers[obs.ID] = obs
            return nil
        },
        OnObserverRemoved: func(ctx context.Context, obs *iii.Observer) {
            every := everyMs(obs)
            mu.Lock()
            defer mu.Unlock()
            if tm := timers[every]; tm != nil {
                delete(tm.observers, obs.ID)
                if len(tm.observers) == 0 {
                    // Last observer of this interval: stop its timer.
                    close(tm.stop)
                    delete(timers, every)
                }
            }
        },
    }})
}

from iii import ObserveHooks, ObserveRejection

commands: dict = {}  # command → Observer

class CommandHooks(ObserveHooks):
    delivery = "owner"
    args_format = {
        "type": "object",
        "additionalProperties": False,
        "required": ["command"],
        "properties": {"command": {"type": "string"}},
    }

    def on_observer_added(self, obs):
        command = obs.args["command"]
        if command in commands:
            raise ObserveRejection("COMMAND_TAKEN", f"/{command} already has a handler")
        commands[command] = obs

    def on_observer_removed(self, obs):
        if commands.get(obs.args["command"]) is obs:
            del commands[obs.args["command"]]

async def on_command(req: dict) -> dict:
    obs = commands.get(req["command"])
    if obs is None:
        return {"reply": f"unknown command /{req['command']}"}
    return await obs.call_async(req)  # the target's result is this function's result

worker.register_function(
    "chat::on-command",
    on_command,
    description="One invocation per chat command. Observe it with args { command }.",
    observe=CommandHooks(),
)

use std::collections::HashMap;
use std::sync::{Arc, Mutex};

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

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

#[derive(Serialize, Deserialize, JsonSchema)]
struct CommandRequest {
    command: String,
    text: String,
    user: String,
}

#[derive(Serialize, Deserialize, JsonSchema)]
struct CommandReply {
    reply: String,
}

/// command → observer
#[derive(Clone, Default)]
struct Commands(Arc<Mutex<HashMap<String, Observer>>>);

#[async_trait::async_trait]
impl ObserveHandler for Commands {
    async fn on_observer_added(&self, obs: &Observer) -> Result<(), ObserveRejection> {
        let CommandArgs { command } = obs.args()?;
        let mut commands = self.0.lock().unwrap();
        if commands.contains_key(&command) {
            return Err(ObserveRejection::new(
                "COMMAND_TAKEN",
                format!("/{command} already has a handler"),
            ));
        }
        commands.insert(command, obs.clone());
        Ok(())
    }

    async fn on_observer_removed(&self, obs: &Observer) {
        self.0.lock().unwrap().retain(|_, o| o.id() != obs.id());
    }
}

let commands = Commands::default();

worker.register_function(
    "chat::on-command",
    RegisterFunction::new_async({
        let commands = commands.clone();
        move |req: CommandRequest| {
            let obs = commands.0.lock().unwrap().get(&req.command).cloned();
            async move {
                match obs {
                    // the target's result is this function's result
                    Some(obs) => obs.call::<CommandReply>(&req).await,
                    None => Ok(CommandReply { reply: format!("unknown command /{}", req.command) }),
                }
            }
        }
    })
    .description("One invocation per chat command. Observe it with args { command }.")
    .observe(Observe::owner().args_format::<CommandArgs>().handler(commands.clone())),
);

var (
    cmdMu    sync.Mutex
    commands = map[string]*iii.Observer{} // command → observer
)

func commandOf(obs *iii.Observer) string {
    command, _ := obs.Args["command"].(string) // required by ArgsFormat
    return command
}

func registerCommands(client *iii.Client) error {
    return client.RegisterFunction("chat::on-command", func(ctx context.Context, data json.RawMessage) (any, error) {
        var req struct {
            Command string `json:"command"`
        }
        if err := json.Unmarshal(data, &req); err != nil {
            return nil, err
        }
        cmdMu.Lock()
        obs := commands[req.Command]
        cmdMu.Unlock()
        if obs == nil {
            return map[string]string{"reply": "unknown command /" + req.Command}, nil
        }
        return obs.Call(ctx, data) // the target's result is this function's result
    }, iii.RegisterFunctionOptions{Observe: &iii.ObserveHooks{
        Delivery:   iii.DeliveryOwner,
        ArgsFormat: json.RawMessage(`{"type":"object","additionalProperties":false,"required":["command"],"properties":{"command":{"type":"string"}}}`),
        OnObserverAdded: func(ctx context.Context, obs *iii.Observer) error {
            command := commandOf(obs)
            cmdMu.Lock()
            defer cmdMu.Unlock()
            if _, taken := commands[command]; taken {
                return iii.ObserveRejection("COMMAND_TAKEN", "/"+command+" already has a handler")
            }
            commands[command] = obs
            return nil
        },
        OnObserverRemoved: func(ctx context.Context, obs *iii.Observer) {
            command := commandOf(obs)
            cmdMu.Lock()
            defer cmdMu.Unlock()
            if commands[command] == obs {
                delete(commands, command)
            }
        },
    }})
}