// The market side: offerings -> request (vm -> rfp -> submitRfp) -> bids -> policy -> accept.
// No submitBid: this requester serves no XRPC, so bids are read from bidder repos.
use abi::records::temp::compute::Vm;
use abi::records::temp::market::{Accept, EvaluatePolicy, Rfp, SubmitAccept, SubmitRfp, evaluate_policy};
use abi::repo::create_record;
use abi::{AtUri, Call, Fan, Handle, LexiconType, On, Run, TypedStrongRef};
use serde::{Deserialize, Serialize};
use serde_json::{Map, Value, json};

use crate::calls::{attest, bid_uri, created, read, signed, typed};
use crate::ids::{CallId, Mode};
use crate::input::{Args, MAX_PEERS, Out, Route, Step, bidders, did, list, optional_route, peer, routes, strong};
use crate::nsid::{BID, BIDS_FREE, COMPUTE_VM, EVALUATE_POLICY, OFFERING, STRONG_REF, SUBMIT_ACCEPT, SUBMIT_RFP, USER_DATA};

const DEFAULT_DISK: &str = "50G";

/// compute.vm's mem and disk are sizes, "<n>M" or "<n>G" (a bidder refuses anything else after it
/// accepts); a number is GiB.
/// An atproto datetime: YYYY-MM-DDTHH:MM:SS, optional fraction, then Z or ±HH:MM.
fn datetime(t: &str) -> bool {
    let b = t.as_bytes();
    let digits = |r: std::ops::Range<usize>| r.clone().all(|i| b.get(i).is_some_and(u8::is_ascii_digit));
    let head = b.len() >= 20 && digits(0..4) && b[4] == b'-' && digits(5..7) && b[7] == b'-' && digits(8..10) && b[10] == b'T' && digits(11..13) && b[13] == b':' && digits(14..16) && b[16] == b':' && digits(17..19);
    let mut i = 19;
    if head && b[i] == b'.' {
        i += 1;
        while b.get(i).is_some_and(u8::is_ascii_digit) {
            i += 1;
        }
    }
    head && (&t[i..] == "Z" || (t.len() == i + 6 && (b[i] == b'+' || b[i] == b'-') && digits(i + 1..i + 3) && b[i + 3] == b':' && digits(i + 4..i + 6)))
}

pub fn size(value: &Value, key: &str) -> Result<Value, String> {
    let text = match value {
        Value::Number(n) => n.as_u64().filter(|n| *n > 0).map(|n| format!("{n}G")),
        Value::String(s) => s.strip_suffix(['M', 'G']).filter(|n| !n.is_empty() && n.bytes().all(|b| b.is_ascii_digit()) && n.len() <= 7).map(|_| s.clone()),
        _ => None,
    };
    text.map(Value::String).ok_or(format!("{key} must be a size like 4G or 512M, or a number of GiB"))
}

// ---- offerings: bidder endpoints (old: runComputeContract step 4) ----------------------------

#[derive(Default, Serialize, Deserialize)]
pub struct Offerings {
    bidders: Vec<String>,
    reads: Fan,
}

pub fn offers_compute(offering: &Value) -> bool {
    offering["appliesTo"].as_array().is_some_and(|a| a.contains(&json!(COMPUTE_VM))) && offering["endpoints"].as_array().is_some_and(|e| !e.is_empty()) && offering["withdrawnAt"].is_null()
}

/// Each bidder's compute offering (getRecord at self) as { did, endpoints }; pass them as request's bidders.
pub fn offerings(run: Run<Args>) -> Result<Handle, String> {
    let (mut step, on) = Step::<Offerings>::new(run)?;
    step.args.signed_in()?;
    match on {
        On::Start => {
            let bidders = bidders(&step.args)?;
            let reads = bidders.iter().enumerate().map(|(i, bidder)| (CallId::Offerings(i), read(&AtUri::new(bidder, OFFERING, "self"))));
            step.out.fan_out(&mut step.state.reads, reads);
            step.state.bidders = bidders;
        }
        On::Result(id @ CallId::Offerings(_), result) => {
            let Some(results) = step.state.reads.land(&id, result) else { return step.handle() };
            let (mut endpoints, mut errors) = (Vec::<Value>::new(), Vec::new());
            for (bidder, result) in step.state.bidders.iter().zip(results) {
                match result {
                    Ok(output) => {
                        if let Some(peer) = Some(&output["value"]).filter(|v| offers_compute(v)).and_then(|v| peer(bidder, v)) {
                            if !endpoints.iter().any(|e| e["did"] == peer["did"]) {
                                endpoints.push(peer);
                            }
                        }
                    }
                    Err(error) => errors.push(json!({ "bidder": bidder, "error": error })),
                }
            }
            step.out.finish(json!({ "offerings": endpoints, "errors": errors }));
        }
        _ => {}
    }
    step.handle()
}

// ---- request: vm -> rfp -> submitRfp ------------------------------------------------------------

/// [userData hook ->] createRecord vm -> attest rfp -> createRecord rfp, the rfp naming the vm, the input's
/// policies and onBehalfOf. Records it creates go on `kept`. No user_data: the hook builds it from shell,
/// iroh, webrtc_ssh, ssh_key and tunnels (default: the user-data module).
#[derive(Default, Serialize, Deserialize)]
#[serde(default, rename_all = "camelCase")]
pub struct RfpChain {
    policies: Vec<TypedStrongRef>,
    on_behalf_of: Option<TypedStrongRef>,
    unsigned: Value,
    pub user_data: String,
    pub vm: Option<TypedStrongRef>,
    pub rfp: Option<TypedStrongRef>,
}

impl RfpChain {
    // disk defaults like socialweb-computer-kcp compute-request-xrpc (DEFAULT_VM_DISK).
    pub fn start(args: &Args, on_behalf_of: Option<TypedStrongRef>, out: &mut Out) -> Result<Self, String> {
        let policies = match args.arg("policies") {
            Value::Null => vec![],
            Value::Array(list) => list.iter().map(|p| strong(p, "policies")).collect::<Result<_, _>>()?,
            _ => return Err("policies must be a list of strong refs".into()),
        };
        let mut chain = Self { policies, on_behalf_of, ..Self::default() };
        match args.arg("user_data") {
            Value::Null => {
                let input: Map<String, Value> = ["shell", "iroh", "webrtc_ssh", "ssh_key", "tunnels"].into_iter().filter(|k| !args.arg(k).is_null()).map(|k| (k.to_string(), args.arg(k).clone())).collect();
                out.call(CallId::UserData, Call::new("", USER_DATA).input(Value::Object(input)));
            }
            value => {
                chain.user_data = value.as_str().filter(|s| !s.trim().is_empty()).ok_or("user_data must be cloud-init text, or omitted")?.into();
                chain.create_vm(args, out)?;
            }
        }
        Ok(chain)
    }

    fn create_vm(&self, args: &Args, out: &mut Out) -> Result<(), String> {
        let role = args.arg("role").as_str().map(str::trim).filter(|s| !s.is_empty()).unwrap_or("compute");
        // what the bidder runs, always said: a bidder serving vm and deno must not have to guess
        let kind = match args.arg("kind").as_str().unwrap_or("vm") {
            kind @ ("vm" | "deno") => kind,
            _ => return Err("kind must be vm or deno".into()),
        };
        let given = |key: &str| Some(args.arg(key).clone()).filter(|v| !v.is_null());
        let cpus = given("cpus").map(|c| c.as_u64().filter(|n| (1..=256).contains(n)).map(|n| json!(n)).ok_or("cpus must be an integer, 1 to 256")).transpose()?;
        let vm = Vm {
            cpus,
            mem: given("mem").map(|m| size(&m, "mem")).transpose()?,
            disk: given("disk").map_or(Ok(json!(DEFAULT_DISK)), |d| size(&d, "disk"))?,
            network: given("network"),
            location: None,
            role: role.into(),
            user_data: self.user_data.clone(),
            created_at: args.created_at()?,
        };
        let mut record = typed(vm);
        record["kind"] = json!(kind);
        // when the bidder ends the VM (it ends one without, a day after the accept)
        if let Some(at) = given("expires_at") {
            record["expiresAt"] = json!(at.as_str().filter(|t| datetime(t)).ok_or("expires_at must be an RFC 3339 datetime, e.g. 2026-11-10T22:00:00Z")?);
        }
        out.call(CallId::Vm, create_record(args.me(), Vm::lexicon_type(), None, record));
        Ok(())
    }

    pub fn step(&mut self, args: &Args, id: CallId, result: Result<Value, String>, out: &mut Out, kept: &mut Vec<String>) -> Result<Option<TypedStrongRef>, String> {
        match id {
            CallId::UserData => {
                let output = result.map_err(|e| format!("{USER_DATA}: {e}"))?;
                let user_data = output["user_data"].as_str().filter(|s| !s.trim().is_empty());
                self.user_data = user_data.ok_or(format!("user_data is required: no {USER_DATA} hook built it"))?.into();
                self.create_vm(args, out)?;
            }
            CallId::Vm => {
                let vm = created(result)?;
                kept.push(vm.uri.clone());
                let rfp = Rfp {
                    domain: "compute".into(),
                    payload: vm.clone(),
                    submit_bid: None,
                    policies: Some(self.policies.clone()).filter(|p| !p.is_empty()),
                    on_behalf_of: self.on_behalf_of.clone(),
                    created_at: args.created_at()?,
                };
                self.unsigned = typed(rfp);
                out.call(CallId::AttestRfp, attest(args.me(), Some(&self.unsigned)));
                self.vm = Some(vm);
            }
            CallId::AttestRfp => out.call(CallId::Rfp, create_record(args.me(), Rfp::lexicon_type(), None, signed(result, &self.unsigned)?)),
            CallId::Rfp => {
                let rfp = created(result)?;
                kept.push(rfp.uri.clone());
                self.rfp = Some(rfp.clone());
                return Ok(Some(rfp));
            }
            _ => {}
        }
        Ok(None)
    }
}

pub fn submit_rfp(rfp: &TypedStrongRef) -> SubmitRfp {
    SubmitRfp { rfp_uri: rfp.uri.clone(), rfp_cid: rfp.cid.clone() }
}

#[derive(Default, Serialize, Deserialize)]
pub struct Request {
    bidders: Vec<Route>,
    chain: RfpChain,
    submits: Fan,
}

pub fn request(run: Run<Args>) -> Result<Handle, String> {
    let (mut step, on) = Step::<Request>::new(run)?;
    step.args.signed_in()?;
    match on {
        On::Start => {
            let bidders = routes(step.args.arg("bidders"), "bidders")?;
            let on_behalf_of = match step.args.arg("onBehalfOf") {
                Value::Null => None,
                value => Some(strong(value, "onBehalfOf")?),
            };
            step.state.chain = RfpChain::start(&step.args, on_behalf_of, &mut step.out)?;
            step.state.bidders = bidders;
        }
        On::Result(id @ (CallId::UserData | CallId::Vm | CallId::AttestRfp | CallId::Rfp), result) => {
            let Request { bidders, chain, submits } = &mut step.state;
            let Some(rfp) = chain.step(&step.args, id, result, &mut step.out, &mut vec![])? else { return step.handle() };
            if bidders.is_empty() {
                step.out.finish(json!({ "vm": chain.vm, "rfp": rfp, "submitted": [] }));
            } else {
                let input = submit_rfp(&rfp);
                step.out.fan_out(submits, bidders.iter().enumerate().map(|(i, bidder)| (CallId::SubmitRfp(i), bidder.call(SUBMIT_RFP, &input))));
            }
        }
        On::Result(id @ CallId::SubmitRfp(_), result) => {
            let Request { bidders, chain, submits } = &mut step.state;
            let Some(results) = submits.land(&id, result) else { return step.handle() };
            let submitted: Vec<Value> = bidders.iter().zip(results).map(|(bidder, result)| match result {
                Ok(output) => json!({ "bidder": bidder, "ok": true, "output": output }),
                Err(error) => json!({ "bidder": bidder, "ok": false, "error": error }),
            }).collect();
            step.out.finish(json!({ "vm": chain.vm, "rfp": chain.rfp, "submitted": submitted }));
        }
        _ => {}
    }
    step.handle()
}

// ---- bids ---------------------------------------------------------------------------------------

#[derive(Default, Serialize, Deserialize)]
pub struct Bids {
    rfp: String,
    bidders: Vec<String>,
    reads: Fan,
}

/// A free bid's payload is (a strong ref to) a market.bids.free.
pub fn is_free(payload: &Value) -> bool {
    let nsid = match payload["$type"].as_str() {
        Some(nsid) if nsid != STRONG_REF => Some(nsid.to_string()),
        _ => payload["uri"].as_str().and_then(|u| u.parse::<AtUri>().ok()).map(|u| u.collection),
    };
    nsid.as_deref() == Some(BIDS_FREE)
}

/// A bid URI (at://did/market.bid/rkey, rkey = submitRfp's `name`) is read exactly; a bidder DID's
/// bid is read at the key it gives a bid on this rfp {uri, cid} (bid_uri).
/// Winner as requester-xrpc picks it: the first free bid, else the lowest payload.cost.
pub fn bids(run: Run<Args>) -> Result<Handle, String> {
    let (mut step, on) = Step::<Bids>::new(run)?;
    step.args.signed_in()?;
    match on {
        On::Start => {
            let rfp_ref = strong(step.args.arg("rfp"), "rfp")?;
            let rfp = rfp_ref.uri.clone();
            let sources: Vec<String> = list(step.args.arg("bidders"), "bidders")?.into_iter().chain(list(step.args.arg("bids"), "bids")?).collect();
            if sources.is_empty() {
                return Err("bidders or bids is required".into());
            }
            if sources.len() > MAX_PEERS {
                return Err(format!("at most {MAX_PEERS} bidders and bids"));
            }
            let (mut reads, mut bidders) = (Vec::new(), Vec::new());
            for (i, source) in sources.iter().enumerate() {
                let (bidder, call) = if source.starts_with("at://") {
                    let uri = source.parse::<AtUri>().ok().filter(|u| u.collection == BID).ok_or("bids must be market.bid at:// URIs")?;
                    (did(&json!(uri.repo), "bids")?, read(&uri))
                } else {
                    let bidder = did(&json!(source), "bidders")?;
                    let call = read(&bid_uri(&bidder, &rfp_ref));
                    (bidder, call)
                };
                reads.push((CallId::Bid(i), call));
                bidders.push(bidder);
            }
            step.out.fan_out(&mut step.state.reads, reads);
            step.state.rfp = rfp;
            step.state.bidders = bidders;
        }
        On::Result(id @ CallId::Bid(_), result) => {
            let Bids { rfp, bidders, reads } = &mut step.state;
            let Some(results) = reads.land(&id, result) else { return step.handle() };
            let (mut bids, mut errors) = (Vec::new(), Vec::new());
            for (bidder, result) in bidders.iter().zip(results) {
                let output = match result {
                    Ok(output) => output,
                    Err(error) => {
                        errors.push(json!({ "bidder": bidder, "error": error }));
                        continue;
                    }
                };
                for record in [&output].into_iter().filter(|r| r["value"]["rfp"]["uri"] == **rfp) {
                    let payload = &record["value"]["payload"];
                    bids.push(json!({
                        "bidder": bidder,
                        "uri": record["uri"],
                        "cid": record["cid"],
                        "free": is_free(payload),
                        "cost": payload["cost"],
                        "submitAccept": record["value"]["submitAccept"],
                    }));
                }
            }
            let cost = |b: &Value| b["cost"].as_f64().unwrap_or(f64::INFINITY);
            let winner = bids.iter().find(|b| b["free"] == true).or_else(|| bids.iter().reduce(|best, b| if cost(b) < cost(best) { b } else { best })).cloned();
            step.out.finish(json!({ "rfp": rfp, "bids": bids, "winner": winner, "errors": errors }));
        }
        _ => {}
    }
    step.handle()
}

// ---- policy: the hook (module.json hooks), kcp's Verdict for a counterparty ------

/// The hook's input (kcp participant.PolicyInput, the requester's perspective): scope mode names the
/// bidder and the rfp (kcp accept.go bidAllowedWithPolicy), full mode adds the bid (runner.go:547).
pub fn evaluate_policy(me: &str, mode: Mode, rfp: &TypedStrongRef, bidder: &str, bid: Option<evaluate_policy::Record>) -> Call {
    let input = EvaluatePolicy {
        self_did: me.into(),
        perspective: Some("requester".into()),
        mode: Some(mode.to_string()),
        counterparty_did: Some(bidder.into()),
        subject_did: Some(bidder.into()),
        root_requester_did: Some(me.into()),
        rfp: Some(evaluate_policy::Record { uri: rfp.uri.clone(), cid: Some(rfp.cid.clone()), value: None }),
        bid,
        accept: None,
    };
    Call::new("", EVALUATE_POLICY).input(json!(input))
}

/// Anything but allow: true is a deny, with its messages; a failed call is one too, and null (a hook
/// nothing serves: the record lost its default) is "no verdict" (fail closed).
pub fn denied(verdict: &Result<Value, String>) -> Option<String> {
    match verdict {
        Ok(verdict) if verdict["allow"] == true => None,
        Ok(verdict) => {
            let msgs: Vec<&str> = verdict["violations"].as_array().into_iter().flatten().filter_map(|v| v["msg"].as_str()).collect();
            Some(if msgs.is_empty() { "no verdict".into() } else { msgs.join("; ") })
        }
        Err(error) => Some(error.clone()),
    }
}

// ---- accept: attest -> accept -> submitAccept ---------------------------------------------------

#[derive(Clone, Serialize, Deserialize)]
pub struct Receipt {
    pub uri: String,
    pub cid: Value,
    #[serde(rename = "submitEvent")]
    pub submit_event: Value,
}

impl Receipt {
    pub fn of(record: &Value) -> Self {
        Receipt { uri: record["uri"].as_str().unwrap_or_default().into(), cid: record["cid"].clone(), submit_event: record["value"]["submitEvent"].clone() }
    }
}

#[derive(Serialize)]
pub struct Accepted {
    pub accept: TypedStrongRef,
    pub receipt: Option<Receipt>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub submitted: Option<Value>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub error: Option<String>,
}

/// attest accept -> createRecord accept -> submitAccept. Without submitAccept the bidder finds the
/// accept on the firehose (read the receipt later); the accept exists by submitAccept, so a failed
/// one is reported, not raised.
#[derive(Default, Serialize, Deserialize)]
#[serde(default, rename_all = "camelCase")]
pub struct AcceptChain {
    unsigned: Value,
    submit_accept: Option<Route>,
    pub accept: Option<TypedStrongRef>,
}

impl AcceptChain {
    pub fn start(args: &Args, rfp: TypedStrongRef, bid: TypedStrongRef, submit_accept: Option<Route>, out: &mut Out) -> Result<Self, String> {
        let unsigned = typed(Accept { rfp, bid, payload: None, submit_event: None, created_at: args.created_at()? });
        out.call(CallId::AttestAccept, attest(args.me(), Some(&unsigned)));
        Ok(Self { unsigned, submit_accept, accept: None })
    }

    pub fn step(&mut self, args: &Args, id: CallId, result: Result<Value, String>, out: &mut Out, kept: &mut Vec<String>) -> Result<Option<Accepted>, String> {
        let accepted = |accept: &TypedStrongRef| Accepted { accept: accept.clone(), receipt: None, submitted: None, error: None };
        match id {
            CallId::AttestAccept => out.call(CallId::Accept, create_record(args.me(), Accept::lexicon_type(), None, signed(result, &self.unsigned)?)),
            CallId::Accept => {
                let accept = created(result)?;
                kept.push(accept.uri.clone());
                self.accept = Some(accept.clone());
                let Some(route) = &self.submit_accept else { return Ok(Some(accepted(&accept))) };
                out.call(CallId::SubmitAccept, route.call(SUBMIT_ACCEPT, SubmitAccept { accept_uri: accept.uri.clone(), accept_cid: accept.cid.clone() }));
            }
            CallId::SubmitAccept => {
                let accept = accepted(self.accept.as_ref().ok_or("bad state")?);
                return Ok(Some(match result {
                    // kcp bidders answer {ok, id} and mint the receipt later: find it with `receipt`.
                    Ok(output) if !output["uri"].is_string() => Accepted { submitted: Some(output), ..accept },
                    Ok(output) => Accepted { receipt: Some(Receipt { uri: output["uri"].as_str().unwrap_or_default().into(), cid: output["cid"].clone(), submit_event: output["submitEvent"].clone() }), ..accept },
                    Err(error) => Accepted { error: Some(error), ..accept },
                }));
            }
            _ => {}
        }
        Ok(None)
    }
}

#[derive(Default, Serialize, Deserialize)]
#[serde(default, rename_all = "camelCase")]
pub struct AcceptBid {
    rfp: Option<TypedStrongRef>,
    bid: Option<TypedStrongRef>,
    submit_accept: Option<Route>,
    chain: AcceptChain,
}

/// The policy hook in full mode first: the requester accepts no bid its policy denies (kcp runner.go:547).
pub fn accept(run: Run<Args>) -> Result<Handle, String> {
    let (mut step, on) = Step::<AcceptBid>::new(run)?;
    step.args.signed_in()?;
    match on {
        On::Start => {
            let rfp = strong(step.args.arg("rfp"), "rfp")?;
            let bid = strong(step.args.arg("bid"), "bid")?;
            let submit_accept = optional_route(step.args.arg("submitAccept"), "submitAccept")?;
            let bidder = bid.uri.parse::<AtUri>().map(|u| u.repo).unwrap_or_default();
            let record = evaluate_policy::Record { uri: bid.uri.clone(), cid: Some(bid.cid.clone()), value: None };
            step.out.call(CallId::AcceptPolicy, evaluate_policy(step.args.me(), Mode::Full, &rfp, &bidder, Some(record)));
            step.state = AcceptBid { rfp: Some(rfp), bid: Some(bid), submit_accept, chain: AcceptChain::default() };
        }
        On::Result(CallId::AcceptPolicy, verdict) => {
            let AcceptBid { rfp, bid, submit_accept, chain } = &mut step.state;
            let (rfp, bid) = (rfp.clone().ok_or("bad state")?, bid.clone().ok_or("bad state")?);
            if let Some(why) = denied(&verdict) {
                return Err(format!("bid {} denied by policy (full): {why}", bid.uri));
            }
            *chain = AcceptChain::start(&step.args, rfp, bid, submit_accept.take(), &mut step.out)?;
        }
        On::Result(id, result) => {
            if let Some(accepted) = step.state.chain.step(&step.args, id, result, &mut step.out, &mut vec![])? {
                step.out.finish(accepted);
            }
        }
        _ => {}
    }
    step.handle()
}