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

Switch repository

Loading…

// The contract's last step: find the signals `signal` wrote for its receipt (a webrtc.connect
// shell, a webrtc:// fetch to a kept VM's shell) at their computed keys, then delete them and every
// record the run created.
use abi::repo::delete_record;
use abi::{AtUri, Call};
use serde_json::{Value, json};

use crate::calls::read;
use crate::contract::{Contracting, Phase, finish};
use crate::guest::{STREAM_MAX, Stream};
use crate::ids::CallId;
use crate::input::tags;

pub fn cleanup(c: &mut Contracting) {
    let s = &mut c.state;
    if *c.args.arg("cleanup") == false || s.created.is_empty() {
        return finish(c);
    }
    s.phase = Phase::Cleanup;
    let Some(receipt) = s.receipt.as_ref().map(|r| r.uri.clone()) else { return delete(c) };
    // A kept VM's release (resume) has no tags of its own: the default ones.
    let tags = if s.tags.is_empty() { tags(&Value::Null).unwrap_or_default() } else { s.tags.clone() };
    s.signals = Stream::own(&tags);
    if s.signals.is_empty() {
        return delete(c);
    }
    let reads: Vec<_> = s.signals.iter().enumerate().map(|(i, stream)| (CallId::Page(i), signal_read(c.args.me(), stream, &receipt))).collect();
    c.out.fan_out(&mut s.deletes, reads);
}

fn signal_read(me: &str, stream: &Stream, receipt: &str) -> Call {
    read(&stream.signal(me, receipt).0)
}

// One read of a signal stream: a found payload and its event are the run's to delete and the next
// index is read; a missing one ends that stream.
pub fn signals(c: &mut Contracting, i: usize, result: Result<Value, String>) {
    let s = &mut c.state;
    let receipt = s.receipt.as_ref().map(|r| r.uri.clone()).unwrap_or_default();
    let found = result.as_ref().is_ok_and(|o| o["value"].is_object());
    if let (Some(stream), true) = (s.signals.get_mut(i), found) {
        let (payload, event) = stream.signal(c.args.me(), &receipt);
        for uri in [payload.to_string(), event.to_string()] {
            if !s.created.contains(&uri) {
                s.created.push(uri);
            }
        }
        stream.next += 1;
        if stream.next < STREAM_MAX * 4 {
            let call = signal_read(c.args.me(), stream, &receipt);
            c.out.fan_out(&mut s.deletes, [(CallId::Page(i), call)]);
        }
    }
    if s.deletes.land(&CallId::Page(i), result).is_some() {
        delete(c);
    }
}

fn delete(c: &mut Contracting) {
    let s = &mut c.state;
    let mut deletes = Vec::new();
    for (k, uri) in s.created.iter().enumerate().rev() {
        match uri.parse::<AtUri>() {
            Ok(uri) => deletes.push((CallId::Delete(k), delete_record(&uri))),
            Err(_) => s.errors.push(format!("cannot delete {}: not an at:// record URI", json!(uri))),
        }
    }
    if deletes.is_empty() {
        return finish(c);
    }
    c.out.fan_out(&mut s.deletes, deletes);
}

pub fn deleted(c: &mut Contracting, k: usize, result: Result<Value, String>) {
    let s = &mut c.state;
    match &result {
        Ok(_) => s.deleted.extend(s.created.get(k).cloned()),
        // A record this host's managed_records ledger does not hold is refused: noted, not fatal.
        Err(e) => s.errors.push(format!("delete {}: {e}", s.created.get(k).map(String::as_str).unwrap_or_default())),
    }
    if s.deletes.land(&CallId::Delete(k), result).is_some() {
        finish(c);
    }
}