johnandersen777.bsky.social/atlibs-socialweb-computer-requester

Switch repository

Loading…

// A signed payload under a market.event for the bidder: vm.delete (release), or a guest event the
// bidder forwards to the VM's guest agent (signal: a webrtc offer or an age key).
use abi::records::temp::compute::events::guest::{PublicKey, SignalWebRtc};
use abi::records::temp::market::{Event, SubmitEvent};
use abi::repo::create_record;
use abi::{Handle, LexiconType, On, Run, TypedStrongRef};
use serde::{Deserialize, Serialize};
use serde_json::Value;

use crate::calls::{attest, created, read, signed, typed};
use crate::guest::{STREAM_MAX, Stream};
use crate::ids::CallId;
use crate::input::{Args, Out, Route, Step, optional_route, strong, tag};
use crate::nsid::SUBMIT_EVENT;

#[derive(Serialize)]
pub struct Posted {
    pub payload: Option<TypedStrongRef>,
    pub event: Option<TypedStrongRef>,
    pub submitted: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub error: Option<String>,
}

/// Sign and create the payload (its collection is its $type), then sign and create the market.event
/// naming it under the receipt, then submitEvent when there is a route.
#[derive(Default, Serialize, Deserialize)]
#[serde(default, rename_all = "camelCase")]
pub struct PostChain {
    receipt: Option<TypedStrongRef>,
    submit_event: Option<Route>,
    unsigned: Value,
    signed: Value,
    pub payload: Option<TypedStrongRef>,
    pub event: Option<TypedStrongRef>,
    rkeys: Option<(String, String)>,
}

impl PostChain {
    /// rkeys: the payload's and the event's (a signal's computed keys), else TIDs.
    pub fn start(args: &Args, receipt: TypedStrongRef, submit_event: Option<Route>, payload: Value, rkeys: Option<(String, String)>, out: &mut Out) -> Self {
        out.call(CallId::AttestPayload, attest(args.me(), Some(&payload)));
        Self { receipt: Some(receipt), submit_event, unsigned: payload, rkeys, ..Self::default() }
    }

    pub fn step(&mut self, args: &Args, id: CallId, result: Result<Value, String>, out: &mut Out, kept: &mut Vec<String>) -> Result<Option<Posted>, String> {
        let me = args.me();
        match id {
            CallId::AttestPayload => {
                let payload = signed(result, &self.unsigned)?;
                let collection = self.unsigned["$type"].as_str().ok_or("bad state")?;
                out.call(CallId::CreatePayload, create_record(me, collection, self.rkeys.as_ref().map(|k| k.0.as_str()), payload));
            }
            CallId::CreatePayload => {
                let payload = created(result)?;
                kept.push(payload.uri.clone());
                let receipt = self.receipt.clone().ok_or("bad state")?;
                self.unsigned = typed(Event { receipt, payload: payload.clone(), created_at: args.created_at()? });
                out.call(CallId::AttestEvent, attest(me, Some(&self.unsigned)));
                self.payload = Some(payload);
            }
            CallId::AttestEvent => {
                self.signed = signed(result, &self.unsigned)?;
                out.call(CallId::CreateEvent, create_record(me, Event::lexicon_type(), self.rkeys.as_ref().map(|k| k.1.as_str()), self.signed.clone()));
            }
            CallId::CreateEvent => {
                let event = created(result)?;
                kept.push(event.uri.clone());
                self.event = Some(event.clone());
                let Some(route) = &self.submit_event else { return Ok(Some(self.posted(false, None))) };
                out.call(CallId::SubmitEvent, route.call(SUBMIT_EVENT, SubmitEvent { uri: event.uri.clone(), cid: event.cid.clone(), record: self.signed.clone() }));
            }
            CallId::SubmitEvent => return Ok(Some(self.posted(result.is_ok(), result.err()))),
            _ => {}
        }
        Ok(None)
    }

    fn posted(&self, submitted: bool, error: Option<String>) -> Posted {
        Posted { payload: self.payload.clone(), event: self.event.clone(), submitted, error }
    }
}

/// The contract's own release and cleanup (contract::resume).
pub fn release(run: Run<Args>) -> Result<Handle, String> {
    crate::contract::resume(run)
}

/// `offer` is age armored ciphertext of {sdp, replyTo}, sealed in the client; the
/// module never sees plaintext or a private key. The signal is the next of its stream (receipt, tag,
/// kind) at a computed key: its own reads find n, then the payload and event are created there.
#[derive(Default, Serialize, Deserialize)]
#[serde(default, rename_all = "camelCase")]
pub struct Signal {
    stream: Stream,
    receipt: Option<TypedStrongRef>,
    payload: Value,
    chain: PostChain,
}

pub fn signal(run: Run<Args>) -> Result<Handle, String> {
    let (mut step, on) = Step::<Signal>::new(run)?;
    step.args.signed_in()?;
    match on {
        On::Start => {
            let args = &step.args;
            let receipt = strong(args.arg("receipt"), "receipt")?;
            let tag = tag(args.arg("tag"))?;
            let created_at = args.created_at()?;
            let (payload, collection, kind) = match (args.arg("offer").as_str(), args.arg("key").as_str()) {
                (Some(c), None) if c.starts_with("-----BEGIN AGE ENCRYPTED FILE-----") && c.len() <= 16384 => {
                    (typed(SignalWebRtc { kind: "offer".into(), tag: tag.clone(), ciphertext: c.into(), created_at }), "guest.signalWebRTC", "offer")
                }
                (None, Some(k)) if k.starts_with("age1") && k.len() <= 256 => {
                    (typed(PublicKey { kind: "age-x25519".into(), tag: tag.clone(), key: k.into(), created_at }), "guest.publicKey", "age-x25519")
                }
                _ => return Err("one of offer (age armored, at most 16384 bytes) or key (age1...) is required".into()),
            };
            step.state = Signal { stream: Stream::new(collection, &tag, kind), receipt: Some(receipt), payload, ..Signal::default() };
            next(&mut step);
        }
        On::Result(CallId::Page(_), result) => {
            let s = &mut step.state;
            if result.is_ok_and(|o| o["value"].is_object()) && s.stream.next + 1 < STREAM_MAX * 4 {
                s.stream.next += 1;
                next(&mut step);
            } else {
                let receipt = s.receipt.clone().ok_or("bad state")?;
                let (payload, event) = s.stream.signal(step.args.me(), &receipt.uri);
                let submit_event = optional_route(step.args.arg("submitEvent"), "submitEvent")?;
                s.chain = PostChain::start(&step.args, receipt, submit_event, s.payload.take(), Some((payload.rkey, event.rkey)), &mut step.out);
            }
        }
        On::Result(id, result) => {
            if let Some(posted) = step.state.chain.step(&step.args, id, result, &mut step.out, &mut vec![])? {
                step.out.finish(posted);
            }
        }
        _ => {}
    }
    step.handle()
}

fn next(step: &mut Step<Signal>) {
    let s = &step.state;
    let receipt = s.receipt.as_ref().map(|r| r.uri.as_str()).unwrap_or_default();
    step.out.call(CallId::Page(s.stream.next), read(&s.stream.signal(step.args.me(), receipt).0));
}