// agents-service: the agents XRPC service on the Rust host.
//
//   serve.json        the ingress module registers a did:key with xrpc.fedproxy.com and forwards
//                     com.fedfork.agents.{launchAgent,getAgent,shutdownAgent} to the handlers below
//   handlers          check the caller's atproto service auth (atlibs_host::serviceauth: aud = this
//                     service, lxm = the method, iss in ATLIBS_AGENTS_ISSUERS), then run
//   composition.json  agents + requester + allow-bidders, as the signed-in account, each instance held to
//                     its own module's session-record scope (records/<session>.json beside the module)
//
// flags (the operator account, explicit; no account is read from the environment):
//   --app-password <handle>     the account the service runs as (required)
//   --password-file <path|->    its app password from a file, or `-` for stdin (first line, trimmed)
//   --password-env <NAME>       its app password from the env var NAMED here (never in argv)
//   --pds <url>                 createSession service (default https://bsky.social)
//   exactly one of --password-file / --password-env.
// env: ATLIBS_SERVER_STATE, ATLIBS_IROH_KEY (as atlibs-host),
//      ATLIBS_AGENTS_ISSUERS (optional: only these DIDs, or `did:<method>:*` prefixes, may call; default
//      anyone who signs in), ATLIBS_AGENTS_USER_DATA (required: a
//      user_data file, `node ../atlibs-iroh/examples/iroh.mjs user-data $ATLIBS_IROH_KEY.pub`), ATLIBS_AGENTS_BIDDERS
//      (default bidders), ATLIBS_RELAY (https://xrpc.fedproxy.com), ATLIBS_AGENTS_LIFETIME_MS (serve this
//      long, default until Ctrl-C), ATLIBS_AGENTS_MAX_PER_CALLER (launches a caller may have at a time,
//      default 2), ATLIBS_AGENTS_MAX_LAUNCHES (launches the service runs at a time, all callers: default 8),
//      ATLIBS_REPO (where modules/, records/ and target/ are; default this checkout; the requester and
//      policy-open modules come from ../atlibs-socialweb-computer-requester, ingress from ../atlibs-fedproxy,
//      iroh-user-data (a hook default of the requester's user-data) from ../atlibs-iroh).
use std::collections::{HashMap, HashSet};
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::Duration;

use atlibs_host::compose::{self, CallFor, Composition, Handler, Info};
use atlibs_host::crypto::ingress_host;
use atlibs_host::host::{DidkeyCall, Fetch, HostCall, Session};
use atlibs_host::oauth::ScopeCall;
use atlibs_host::transport::{TransportCall, Transports};
use atlibs_host::run::{Call, Ledger, Module, Options, Request, Sub};
use atlibs_host::serviceauth::{Auth, Replay, Resolve, resolver, verify};
use atlibs_host::store::{Store, uuid};
use atlibs_host::wasm::{WALL_TIME_MS, Wasm};
use atlibs_host::{BoxFut, Signal, now_ms, pretty};
use serde_json::{Map, Value, json};
use tokio::sync::{oneshot, watch};

const LAUNCH: &str = "com.fedfork.agents.launch";
const LAUNCH_AGENT: &str = "com.fedfork.agents.launchAgent";
const GET_AGENT: &str = "com.fedfork.agents.getAgent";
const SHUTDOWN_AGENT: &str = "com.fedfork.agents.shutdownAgent";
// What a caller may choose; user_data, the default bidders and the guest command
// (the agents module's default, `atirc agent run`) are the service's.
const CLIENT_FIELDS: [&str; 5] = ["name", "via", "bidders", "exp_s", "tag"];
// The agent token's life when the caller names none; the agents module caps exp_s at 24 h. The
// service shuts a launch down when it ends.
const DEFAULT_EXP_S: u64 = 7200;
// Launches a caller may have at a time (ATLIBS_AGENTS_MAX_PER_CALLER).
const MAX_PER_CALLER: usize = 2;
// Launches the service runs at a time, all callers together (ATLIBS_AGENTS_MAX_LAUNCHES): the
// service account pays for every VM, and a new identity is cheap.
const MAX_LAUNCHES: usize = 8;
// The relay answers 502 after 30 s; a slower shutdown answers "stopping" and keeps going.
const SHUTDOWN_WAIT: Duration = Duration::from_secs(25);

// ---- compositions --------------------------------------------------------------------------

// Where a module key's modules/<key>/module.json may be: this repo, then the sibling checkouts that own the
// modules the compositions borrow (requester, policy-open; ingress; iroh-user-data). Each has its own records/ and target/.
const SIBLINGS: [&str; 3] = ["../atlibs-socialweb-computer-requester", "../atlibs-fedproxy", "../atlibs-iroh"];

/// The checkout holding modules/<key>: `repo` or a sibling of it.
fn root_of(repo: &Path, key: &str) -> Result<PathBuf, String> {
    std::iter::once(repo.to_path_buf())
        .chain(SIBLINGS.iter().map(|s| repo.join(s)))
        .find(|r| r.join(format!("modules/{key}/module.json")).is_file())
        .ok_or_else(|| format!("{key}: no modules/{key}/module.json in {} or {}", repo.display(), SIBLINGS.join(", ")))
}

/// A module's session record as (rkey, scope): records/<session>.json in its checkout (the CLI's module_of
/// reads the same file); None when module.json names no session.
fn scope_of(root: &Path, spec: &Value) -> Result<Option<(String, String)>, String> {
    let Some(session) = spec["session"].as_str() else { return Ok(None) };
    let path = root.join(format!("records/{session}.json"));
    let record: Value = serde_json::from_slice(&std::fs::read(&path).map_err(|e| format!("{}: {e}", path.display()))?).map_err(|e| format!("{}: {e}", path.display()))?;
    let scope = record["clientMetadata"]["scope"].as_str().ok_or_else(|| format!("{}: no clientMetadata.scope", path.display()))?;
    Ok(Some((session.to_string(), scope.to_string())))
}

/// A composition file under modules/agents with each instance's module (modules/<key>/module.json,
/// target/wasm32-unknown-unknown/release/<key>.wasm and its session's records/<session>.json, from root_of),
/// hook defaults (the requester's policy) included (compose::load), checked against the transports the
/// service serves.
async fn load(repo: &Path, file: &str, provides: &HashSet<String>) -> Result<Composition, String> {
    let manifest: Value = serde_json::from_slice(&std::fs::read(repo.join("modules/agents").join(file)).map_err(|e| format!("{file}: {e}"))?).map_err(|e| format!("{file}: {e}"))?;
    let repo = repo.to_path_buf();
    let loader: compose::ModuleLoader = Arc::new(move |key: String| {
        let repo = repo.clone();
        Box::pin(async move {
            let read = |p: PathBuf| std::fs::read(&p).map_err(|e| format!("{}: {e}", p.display()));
            if !atlibs_host::run::module_key_ok(&key) {
                return Err(format!("{key}: not a module key"));
            }
            let root = root_of(&repo, &key)?;
            let spec: Value = serde_json::from_slice(&read(root.join(format!("modules/{key}/module.json")))?).map_err(|e| format!("{key}: {e}"))?;
            let path = root.join(format!("target/wasm32-unknown-unknown/release/{}.wasm", key.replace('-', "_")));
            let mut module = Module::from_manifest(read(path.clone())?, &spec)?;
            // Identity by path, as the CLI: the composition's ledger owner (and so which records shutdown may
            // delete) stays the same across rebuilds of the wasm.
            module.id = Some(std::fs::canonicalize(&path).unwrap_or(path).display().to_string());
            module.key = Some(key);
            module.scope = scope_of(&root, &spec)?;
            Ok(module)
        })
    });
    let comp = compose::load(&manifest, &loader).await?;
    compose::check(&comp, provides)?;
    Ok(comp)
}

/// Passes calls through and hands over the launch record's {uri, cid} as soon as the agents
/// module's createRecord of it returns: the launch id, before the VM.
struct Launched {
    inner: Arc<dyn Call>,
    tx: Mutex<Option<oneshot::Sender<Value>>>,
}

impl Call for Launched {
    fn call(&self, nsid: String, request: Request) -> BoxFut<'_, Result<Option<Value>, String>> {
        Box::pin(async move {
            let launch = nsid == "com.atproto.repo.createRecord" && request.input.as_ref().is_some_and(|i| i["collection"] == LAUNCH);
            let out = self.inner.call(nsid, request).await;
            if let (true, Ok(Some(created))) = (launch, &out) {
                if let Some(tx) = self.tx.lock().unwrap().take() {
                    let _ = tx.send(created.clone());
                }
            }
            out
        })
    }
    fn subscribe(&self, nsid: String, request: Request) -> BoxFut<'_, Result<Sub, String>> {
        self.inner.subscribe(nsid, request)
    }
}

// A launch this process started and has not shut down: whose it is, and whether its run ended.
struct Launch {
    caller: String,
    done: watch::Receiver<bool>,
}

pub struct Service {
    pub comp: Arc<Composition>,
    pub wasm: Arc<Wasm>,
    pub ledger: Option<Arc<dyn Ledger>>,
    /// by instance name: the signed-in account's call (HostCall + attest) held to that instance's own
    /// module's session scope (ScopeCall, as the CLI's Host::call_for), then the transports (iroh)
    calls: HashMap<String, Arc<dyn Call>>,
    pub did: String,
    pub defaults: Map<String, Value>,
    pub auth: Auth,
    pub resolve: Resolve,
    // service-auth tokens already used: each is good for one request
    replay: Replay,
    pub max_per_caller: usize,
    pub max_launches: usize,
    pub shutdown_wait: Duration,
    // ponytail: in memory, keyed by launch rkey (a slot key while the record is being made); a
    // restart forgets the caps and lifetimes of earlier launches (their tokens still expire).
    launches: Mutex<HashMap<String, Launch>>,
}

fn client_error(e: String) -> String {
    if e.split_once(' ').is_some_and(|(code, _)| code.len() == 3 && code.parse::<u16>().is_ok()) { e } else { format!("400 InvalidRequest: {e}") }
}

// exp_s as a number (form fields arrive as strings); None when it is not one (the module refuses it).
fn exp_s_of(body: &Value) -> Option<u64> {
    match body.get("exp_s") {
        None | Some(Value::Null) => Some(DEFAULT_EXP_S),
        Some(v) => v.as_u64().or_else(|| v.as_str().and_then(|s| s.trim().parse().ok())),
    }
}

impl Service {
    #[allow(clippy::too_many_arguments)]
    /// `account`: the signed-in account's unscoped call; each instance gets it only through its own scope.
    pub fn new(comp: Composition, wasm: Arc<Wasm>, ledger: Option<Arc<dyn Ledger>>, account: Arc<dyn Call>, transports: Transports, did: String, defaults: Map<String, Value>, auth: Auth, resolve: Resolve) -> Arc<Self> {
        let calls = comp
            .modules
            .iter()
            .map(|(name, module)| {
                let scoped: Arc<dyn Call> = Arc::new(ScopeCall { module: module.key.clone().unwrap_or_else(|| name.clone()), session: module.scope.clone(), inner: account.clone() });
                let listed = module.calls.as_ref().is_some_and(|c| !c.is_empty());
                (name.clone(), Arc::new(TransportCall { transports: transports.clone(), listed, inner: scoped }) as Arc<dyn Call>)
            })
            .collect();
        Arc::new(Self { comp: Arc::new(comp), wasm, ledger, calls, did, defaults, auth, resolve, replay: Replay::default(), max_per_caller: MAX_PER_CALLER, max_launches: MAX_LAUNCHES, shutdown_wait: SHUTDOWN_WAIT, launches: Mutex::default() })
    }

    // One root run of composition.json's agents instance; the run's error is Err.
    async fn run(&self, entry: &str, mut input: Map<String, Value>, launched: Option<oneshot::Sender<Value>>) -> Result<Value, String> {
        input.insert("did".into(), json!(self.did));
        let agents: Arc<dyn Call> = Arc::new(Launched { inner: self.calls["agents"].clone(), tx: Mutex::new(launched) });
        let (calls, comp) = (self.calls.clone(), self.comp.clone());
        let call_for: CallFor = Arc::new(move |name: &str| if name == "agents" { Some(agents.clone()) } else { calls.get(name).cloned() });
        let ran = compose::run_compose(comp, self.wasm.clone(), self.ledger.clone(), "agents", entry, input, call_for, HashMap::new(), None, Options::default()).await?;
        match ran.get("error").and_then(Value::as_str) {
            Some(e) => Err(e.to_string()),
            None => Ok(ran["result"].clone()),
        }
    }

    async fn launch(self: Arc<Self>, caller: &str, body: &Value) -> Result<Value, String> {
        let mut input = self.defaults.clone();
        input.insert("caller".into(), json!(caller));
        for key in CLIENT_FIELDS {
            if let Some(v) = body.get(key).filter(|v| !v.is_null()) {
                input.insert(key.into(), v.clone());
            }
        }
        let exp_s = exp_s_of(body);
        if let Some(exp_s) = exp_s {
            input.insert("exp_s".into(), json!(exp_s));
        }
        let (done_tx, done_rx) = watch::channel(false);
        // A slot for this launch, taken before any record so concurrent launches cannot pass the cap.
        let slot = format!("slot {}", uuid());
        {
            let mut launches = self.launches.lock().unwrap();
            if launches.values().filter(|l| l.caller == caller).count() >= self.max_per_caller {
                return Err(format!("429 RateLimitExceeded: at most {} launches at a time; shut one down first", self.max_per_caller));
            }
            if launches.len() >= self.max_launches {
                return Err(format!("429 RateLimitExceeded: the service runs at most {} launches at a time; try again later", self.max_launches));
            }
            launches.insert(slot.clone(), Launch { caller: caller.into(), done: done_rx });
        }
        let (tx, rx) = oneshot::channel();
        let me = self.clone();
        let task = tokio::spawn(async move {
            let ran = me.run("launch", input, Some(tx)).await;
            let _ = done_tx.send(true);
            if let Err(e) = &ran {
                eprintln!("agents-service: launch: {e}");
            }
            ran
        });
        let Ok(created) = rx.await else {
            // No launch record: the run ended before it (bad input).
            self.launches.lock().unwrap().remove(&slot);
            return Err(client_error(task.await.map_err(|e| e.to_string())?.err().unwrap_or_else(|| "the launch made no record".into())));
        };
        let uri = created["uri"].as_str().unwrap_or_default().to_string();
        let rkey = uri.rsplit('/').next().unwrap_or_default().to_string();
        eprintln!("agents-service: launching {uri}");
        {
            let mut launches = self.launches.lock().unwrap();
            if let Some(launch) = launches.remove(&slot) {
                launches.insert(rkey.clone(), launch);
            }
        }
        // A launch that did not get to running tore itself down: it no longer counts. One that runs
        // is shut down when its agent token expires.
        let me = self.clone();
        let (who, key) = (caller.to_string(), rkey.clone());
        let lifetime = Duration::from_secs(exp_s.unwrap_or(DEFAULT_EXP_S));
        tokio::spawn(async move {
            let started = tokio::time::Instant::now();
            if !matches!(task.await, Ok(Ok(v)) if v["status"] == "running") {
                me.launches.lock().unwrap().remove(&key);
                return;
            }
            tokio::time::sleep_until(started + lifetime).await;
            me.reap(&who, &key).await;
        });
        Ok(json!({ "rkey": rkey, "uri": uri, "status": "launching" }))
    }

    // Shuts a launch down at the end of its life, unless it already was.
    async fn reap(self: Arc<Self>, caller: &str, rkey: &str) {
        if !self.launches.lock().unwrap().contains_key(rkey) {
            return;
        }
        eprintln!("agents-service: {rkey} reached its lifetime; shutting down");
        if let Err(e) = self.clone().shutdown(caller, &json!({ "rkey": rkey })).await {
            eprintln!("agents-service: shutting down {rkey}: {e}");
        }
    }

    async fn shutdown(self: Arc<Self>, caller: &str, body: &Value) -> Result<Value, String> {
        let rkey = body["rkey"].as_str().filter(|r| !r.is_empty()).ok_or("400 InvalidRequest: rkey is required")?.to_string();
        // Another caller's launch in this process is refused at once (the module checks the record
        // for the rest); one still starting finishes first, so its VM is in the record.
        let running = match self.launches.lock().unwrap().get(&rkey) {
            Some(l) if l.caller != caller => return Err(format!("403 Forbidden: launch {rkey} belongs to another caller")),
            Some(l) => Some(l.done.clone()),
            None => None,
        };
        let input = Map::from_iter([("rkey".into(), json!(rkey)), ("delete".into(), json!(body["delete"] == true)), ("caller".into(), json!(caller))]);
        let me = self.clone();
        let key = rkey.clone();
        let task = tokio::spawn(async move {
            if let Some(mut done) = running {
                let _ = done.wait_for(|d| *d).await;
            }
            let out = me.run("shutdown", input, None).await;
            if out.is_ok() {
                me.launches.lock().unwrap().remove(&key);
            }
            out
        });
        match tokio::time::timeout(self.shutdown_wait, task).await {
            Ok(joined) => joined.map_err(|e| e.to_string())?.map_err(client_error),
            Err(_) => Ok(json!({ "rkey": rkey, "status": "stopping" })),
        }
    }

    /// One forwarded request {method, params, body, authorization}: errors are "<status> <Name>: <msg>".
    pub async fn handle(self: Arc<Self>, nsid: &str, request: Value) -> Result<Value, String> {
        let iss = verify(&self.auth, request["authorization"].as_str(), nsid, now_ms() / 1000, &self.resolve, &self.replay)
            .await
            .map_err(|e| format!("401 AuthenticationRequired: {e}"))?;
        let method = request["method"].as_str().unwrap_or_default();
        let (want, ok) = if nsid == GET_AGENT { ("GET", method == "GET") } else { ("POST", method == "POST") };
        if !ok {
            return Err(format!("405 InvalidRequest: {nsid} takes {want}"));
        }
        eprintln!("agents-service: {nsid} by {iss}");
        match nsid {
            LAUNCH_AGENT => self.launch(&iss, &request["body"]).await,
            SHUTDOWN_AGENT => self.shutdown(&iss, &request["body"]).await,
            GET_AGENT => {
                let rkey = request["params"]["rkey"].as_str().filter(|r| !r.is_empty()).ok_or("400 InvalidRequest: rkey is required")?;
                self.run("status", Map::from_iter([("rkey".into(), json!(rkey)), ("caller".into(), json!(iss))]), None).await.map_err(client_error)
            }
            _ => Err(format!("404 MethodNotImplemented: {nsid}")),
        }
    }

    pub fn handlers(self: &Arc<Self>) -> HashMap<String, Handler> {
        [("launchAgent", LAUNCH_AGENT), ("getAgent", GET_AGENT), ("shutdownAgent", SHUTDOWN_AGENT)]
            .into_iter()
            .map(|(name, nsid)| {
                let me = self.clone();
                let handler: Handler = Arc::new(move |input: Value, _info: Info| {
                    let me = me.clone();
                    Box::pin(async move { me.handle(nsid, input).await })
                });
                (name.to_string(), handler)
            })
            .collect()
    }
}

// ---- main ----------------------------------------------------------------------------------

fn env(key: &str) -> Option<String> {
    std::env::var(key).ok().filter(|v| !v.trim().is_empty())
}

const USAGE: &str = "usage: agents-service --app-password <handle> (--password-file <path|-> | --password-env <NAME>) [--pds <url>]";

/// The operator account: --app-password <handle> with exactly one password source.
// No Debug: the password is never printed.
struct Operator {
    handle: String,
    password: String,
    pds: String,
}

impl Operator {
    /// `read(path)` reads a password file (`-` = stdin); `var(name)` reads the named env var.
    fn parse(args: &[String], read: impl Fn(&str) -> Result<String, String>, var: impl Fn(&str) -> Option<String>) -> Result<Operator, String> {
        let (mut handle, mut file, mut name, mut pds) = (None, None, None, None);
        let mut it = args.iter();
        while let Some(flag) = it.next() {
            let slot = match flag.as_str() {
                "--app-password" => &mut handle,
                "--password-file" => &mut file,
                "--password-env" => &mut name,
                "--pds" => &mut pds,
                _ => return Err(format!("unknown argument {flag}\n{USAGE}")),
            };
            let value = it.next().filter(|v| !v.is_empty()).ok_or(format!("{flag} needs a value\n{USAGE}"))?;
            if slot.replace(value.clone()).is_some() {
                return Err(format!("{flag} given twice\n{USAGE}"));
            }
        }
        let handle = handle.ok_or(format!("--app-password <handle> is required (the operator account)\n{USAGE}"))?;
        let password = match (file, name) {
            (Some(path), None) => read(&path).map_err(|e| format!("--password-file {path}: {e}"))?.lines().next().unwrap_or("").trim().to_string(),
            (None, Some(name)) => var(&name).ok_or(format!("--password-env {name}: the variable is unset"))?.trim().to_string(),
            _ => return Err(format!("--app-password needs exactly one of --password-file / --password-env\n{USAGE}")),
        };
        if password.is_empty() {
            return Err("the app password is empty".into());
        }
        Ok(Operator { handle, password, pds: pds.unwrap_or_else(|| "https://bsky.social".into()) })
    }
}

fn read_password(path: &str) -> Result<String, String> {
    if path == "-" {
        let mut line = String::new();
        std::io::stdin().read_line(&mut line).map_err(|e| e.to_string())?;
        return Ok(line);
    }
    std::fs::read_to_string(path).map_err(|e| e.to_string())
}

async fn serve(operator: Operator, signal: Signal) -> Result<Value, String> {
    let need = |key: &str| env(key).ok_or(format!("{key} is required"));
    let repo = env("ATLIBS_REPO").map(PathBuf::from).unwrap_or_else(|| Path::new(env!("CARGO_MANIFEST_DIR")).join(".."));
    let state_dir = env("ATLIBS_SERVER_STATE").map(PathBuf::from).unwrap_or_else(|| PathBuf::from(std::env::var("HOME").unwrap_or_default()).join(".atlibs-server"));
    let relay = env("ATLIBS_RELAY").unwrap_or_else(|| "https://xrpc.fedproxy.com".into());
    let issuers: Vec<String> = env("ATLIBS_AGENTS_ISSUERS").unwrap_or_default().split([' ', ',']).filter(|s| s.starts_with("did:")).map(String::from).collect();
    let user_data = std::fs::read_to_string(need("ATLIBS_AGENTS_USER_DATA")?).map_err(|e| format!("ATLIBS_AGENTS_USER_DATA: {e}"))?;

    let store = Store::open(&state_dir, "atlibs-server")?;
    let client = reqwest::Client::builder().user_agent("atlibs-agents-service").build().map_err(|e| e.to_string())?;
    let fetch: Arc<dyn Fetch> = Arc::new(client.clone());
    let session = Session::login(&operator.handle, Some(operator.password), &operator.pds, fetch.clone(), Some(store.clone())).await?;
    let iroh_key = env("ATLIBS_IROH_KEY").map(PathBuf::from).unwrap_or_else(|| state_dir.join("iroh_ed25519"));
    let account: Arc<dyn Call> = Arc::new(HostCall { fetch: fetch.clone(), session: Some(session.clone()), attest: Some(store.clone()) });
    let transports = vec![atlibs_iroh::transport(iroh_key)];
    let provides = atlibs_host::transport::provides(&transports);

    // The service's did:key: the ingress identity and the aud callers mint service auth for.
    let serve_path = repo.join("modules/agents/serve.json");
    let key_id = format!("{}#ingress", std::fs::canonicalize(&serve_path).unwrap_or(serve_path).display());
    let key = store.didkey(&key_id).await?;
    let did_key = key.did.clone();
    let host = ingress_host(&did_key, &relay);
    eprintln!("agents-service for {} at https://{host}/xrpc/{{{LAUNCH_AGENT},{GET_AGENT},{SHUTDOWN_AGENT}}} (service DID {did_key})", session.did());
    let ingress: Arc<dyn Call> = Arc::new(DidkeyCall { key, services: vec![relay.clone()], fetch: fetch.clone(), inner: Arc::new(HostCall { fetch, session: None, attest: None }), store: Some(store.clone()) });

    let mut defaults = Map::from_iter([("user_data".into(), json!(user_data)), ("via".into(), json!("iroh"))]);
    if let Some(bidders) = env("ATLIBS_AGENTS_BIDDERS") {
        defaults.insert("bidders".into(), json!(bidders.split([' ', ',']).filter(|s| !s.is_empty()).collect::<Vec<_>>()));
    }
    let wasm = Wasm::new(WALL_TIME_MS);
    let ledger = Some(store.ledger("managed_records"));
    let auth = Auth { audiences: vec![did_key.clone(), format!("did:web:{host}")], issuers };
    let mut service = Service::new(load(&repo, "composition.json", &provides).await?, wasm.clone(), ledger.clone(), account, transports, session.did(), defaults, auth, resolver(client));
    if let Some(n) = env("ATLIBS_AGENTS_MAX_PER_CALLER") {
        Arc::get_mut(&mut service).unwrap().max_per_caller = n.parse().map_err(|_| "ATLIBS_AGENTS_MAX_PER_CALLER must be a number")?;
    }
    if let Some(n) = env("ATLIBS_AGENTS_MAX_LAUNCHES") {
        Arc::get_mut(&mut service).unwrap().max_launches = n.parse().map_err(|_| "ATLIBS_AGENTS_MAX_LAUNCHES must be a number")?;
    }

    let mut input = Map::from_iter([("relay".into(), json!(relay))]);
    if let Some(ms) = env("ATLIBS_AGENTS_LIFETIME_MS") {
        input.insert("lifetimeMs".into(), json!(ms));
    }
    let call_for: CallFor = Arc::new(move |name: &str| (name == "ingress").then(|| ingress.clone()));
    let serve = Arc::new(load(&repo, "serve.json", &provides).await?);
    compose::run_compose(serve, wasm, ledger, "ingress", "serve", input, call_for, service.handlers(), None, Options { signal: Some(signal), ..Default::default() }).await
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    if let Err(e) = atlibs_host::wasm::concurrency() {
        eprintln!("{e}");
        std::process::exit(1);
    }
    let args: Vec<String> = std::env::args().skip(1).collect();
    let operator = match Operator::parse(&args, read_password, |name| std::env::var(name).ok()) {
        Ok(operator) => operator,
        Err(e) => {
            eprintln!("{e}");
            std::process::exit(2);
        }
    };
    let signal = Signal::new();
    let stop = signal.clone();
    tokio::spawn(async move {
        if tokio::signal::ctrl_c().await.is_ok() {
            stop.abort("aborted");
        }
    });
    match serve(operator, signal).await {
        Ok(ran) => println!("{}", pretty(&ran)),
        Err(e) => {
            eprintln!("{e}");
            std::process::exit(1);
        }
    }
}

#[cfg(test)]
mod tests;