Relay Operations

use swarm_authority_examples::observation;
#[path = "../relay_directory.rs"]
mod relay_directory;
use chio_core_types::Keypair;
use chio_federation::pheromone_gossip::*;
use chio_pheromone::*;
use chio_pheromone_relay::*;

use std::sync::Arc;
use swarm_authority_examples::now_ms;

#[path = "../relay_receiver.rs"]
mod relay_receiver;
use relay_receiver::Receiver;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    if std::env::args().nth(1).as_deref() == Some("setup") {
        return setup();
    }
    let sender = Keypair::generate();
    let recipient = Keypair::generate();
    let passport = Keypair::generate();
    // Reserve an available loopback address, then close it for the outage case.
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
    let address = listener.local_addr()?;
    drop(listener);
    let endpoint = format!("http://{address}");
    let receiver_document = relay_directory::directory(&sender, endpoint.clone());
    let sender_document = relay_directory::client_directory(&recipient, endpoint);
    let receiver_directory = PeerDirectory::from_document(receiver_document.clone(), now_ms())?;
    let sender_directory = PeerDirectory::from_document(sender_document.clone(), now_ms())?;
    let run_dir = format!("relay-run-{}", now_ms());
    std::fs::create_dir(&run_dir)?;
    std::fs::write(
        format!("{run_dir}/sender-directory.json"),
        serde_json::to_vec_pretty(&sender_document)?,
    )?;
    std::fs::write(
        format!("{run_dir}/receiver-directory.json"),
        serde_json::to_vec_pretty(&receiver_document)?,
    )?;
    let outbox = SqlitePheromoneRelayStore::open(format!("{run_dir}/sender.sqlite"))?;
    let inbox = Arc::new(SqlitePheromoneRelayStore::open(format!(
        "{run_dir}/receiver.sqlite"
    ))?);
    let deposit = sign_deposit(observation::body(&passport), &passport)?;
    let batch = PheromoneGossipBatch {
        schema: PHEROMONE_GOSSIP_BATCH_SCHEMA.into(),
        recipient_kernel_id: "did:chio:buyer-kernel".into(),
        treaty_id: "treaty:buyer-llamaworks:support-ops".into(),
        flushed_at_unix_ms: now_ms(),
        frames: vec![PheromoneDepositGossip {
            schema: PHEROMONE_GOSSIP_SCHEMA.into(),
            deposit,
            origin_kernel_id: "did:chio:llamaworks".into(),
            gossiping_peer_kernel_id: "did:chio:llamaworks".into(),
            treaty_id: "treaty:buyer-llamaworks:support-ops".into(),
            ts_unix_ms: now_ms(),
            transit_chain: None,
        }],
    };
    outbox.enqueue_batch(
        "did:chio:llamaworks",
        "did:chio:buyer-kernel",
        &batch.treaty_id,
        &batch,
        now_ms(),
    )?;
    let unavailable = deliver_due_batches(
        &outbox,
        sender_directory.clone(),
        sender.clone(),
        "did:chio:llamaworks",
        now_ms(),
        1,
    )
    .await?;
    assert_eq!(unavailable.retried, 1);
    println!("receiver offline: {} retry queued", unavailable.retried);
    // Advance the injected scheduler clock past backoff. This does not sleep or
    // change the machine clock; production schedulers wait until work is due.
    let retry_at = now_ms() + 180_000;
    let mut context = observation::live_context(&passport, &recipient);
    for policy in &mut context.scarcity_policies {
        policy.window_end_unix_ms = now_ms() + 3_600_000;
        policy.window_id = scarcity_window_id(policy, &batch.treaty_id)?;
        policy.policy_sha256 = scarcity_policy_sha256(policy)?;
    }
    let observations = Arc::new(InMemoryPheromoneSubstrate::new());
    let receiver = Arc::new(Receiver {
        observations: observations.clone(),
        context,
    });
    let service = PheromoneRelayService::new(
        PheromoneRelayConfig {
            local_kernel_id: "did:chio:buyer-kernel".into(),
            profile: RelayProfile::LocalDev,
            now_unix_ms: retry_at,
            freshness_window_ms: 60_000,
            max_body_bytes: 256_000,
            use_system_clock: false,
            operator_token: None,
            report_dir: None,
        },
        receiver_directory,
        receiver,
        inbox,
    );
    let listener = tokio::net::TcpListener::bind(address).await?;
    let server = tokio::spawn(service.serve(listener));
    let delivered = deliver_due_batches(
        &outbox,
        sender_directory,
        sender,
        "did:chio:llamaworks",
        retry_at,
        1,
    )
    .await?;
    server.abort();
    assert_eq!(delivered.delivered, 1, "{delivered:?}");
    let admitted = observations.query_deposits(&DepositQuery {
        subject_class: None,
        treaty_id: None,
    })?;
    assert_eq!(admitted.len(), 1);
    println!(
        "receiver online: {} batch delivered; {} signed observation stored",
        delivered.delivered,
        admitted.len()
    );
    println!("evidence: {run_dir}");
    Ok(())
}

fn setup() -> Result<(), Box<dyn std::error::Error>> {
    use std::fs::{self, OpenOptions};
    use std::io::Write;
    let directory = std::path::Path::new("relay-state");
    fs::create_dir(directory)?;
    let sender = Keypair::generate();
    let receiver = Keypair::generate();
    let passport = Keypair::generate();
    let endpoint = "http://127.0.0.1:18080".to_string();
    let sender_directory = relay_directory::client_directory(&receiver, endpoint.clone());
    let receiver_directory = relay_directory::directory(&sender, endpoint);
    for (name, key) in [
        ("sender", &sender),
        ("receiver", &receiver),
        ("passport", &passport),
    ] {
        let mut options = OpenOptions::new();
        options.write(true).create_new(true);
        #[cfg(unix)]
        {
            use std::os::unix::fs::OpenOptionsExt;
            options.mode(0o600);
        }
        let mut file = options.open(directory.join(format!("{name}-key.json")))?;
        file.write_all(&serde_json::to_vec(&serde_json::json!({"kernelId": if name == "receiver" { "did:chio:buyer-kernel" } else { "did:chio:llamaworks" }, "seedHex":key.seed_hex()}))?)?;
    }
    fs::write(
        directory.join("sender-directory.json"),
        serde_json::to_vec_pretty(&sender_directory)?,
    )?;
    fs::write(
        directory.join("receiver-directory.json"),
        serde_json::to_vec_pretty(&receiver_directory)?,
    )?;
    let body = observation::body(&passport);
    let batch = PheromoneGossipBatch {
        schema: PHEROMONE_GOSSIP_BATCH_SCHEMA.into(),
        recipient_kernel_id: "did:chio:buyer-kernel".into(),
        treaty_id: body.treaty_scope[0].clone(),
        flushed_at_unix_ms: now_ms(),
        frames: vec![PheromoneDepositGossip {
            schema: PHEROMONE_GOSSIP_SCHEMA.into(),
            treaty_id: body.treaty_scope[0].clone(),
            origin_kernel_id: body.kernel_id.clone(),
            gossiping_peer_kernel_id: body.kernel_id.clone(),
            deposit: sign_deposit(body, &passport)?,
            ts_unix_ms: now_ms(),
            transit_chain: None,
        }],
    };
    fs::write(
        directory.join("batch.json"),
        serde_json::to_vec_pretty(&batch)?,
    )?;
    let store = SqlitePheromoneRelayStore::open(directory.join("sender.sqlite"))?;
    store.enqueue_batch(
        "did:chio:llamaworks",
        "did:chio:buyer-kernel",
        &batch.treaty_id,
        &batch,
        now_ms(),
    )?;
    println!("relay-state: fresh keys, two directories, one signed observation queued");
    Ok(())
}

use chio_core_types::Keypair;
use chio_pheromone_relay::*;
use swarm_authority_examples::now_ms;
pub fn directory(sender: &Keypair, endpoint: String) -> PeerDirectoryDocument {
    PeerDirectoryDocument {
        schema: PHEROMONE_PEER_DIRECTORY_SCHEMA.to_string(),
        local_kernel_id: "did:chio:buyer-kernel".to_string(),
        issued_at_unix_ms: now_ms() - 1_000,
        expires_at_unix_ms: now_ms() + 3_600_000,
        peers: vec![PeerDirectoryEntry {
            kernel_id: "did:chio:llamaworks".to_string(),
            public_key: sender.public_key(),
            endpoint,
            treaty_subscriptions: vec!["treaty:buyer-llamaworks:support-ops".to_string()],
            relay_role: RelayRole::Origin,
            allowed_subject_class_namespaces: vec!["dev.chio.support".to_string()],
            accepted_ladder_refs: vec![RelayLadderRef {
                ladder_manifest_id: "ladder:llamaworks:support:v1".to_string(),
                ladder_manifest_sha256: "a".repeat(64),
                expires_at_unix_ms: now_ms() + 3_600_000,
            }],
            max_batch_frames: 8,
            max_catchup_frames: 16,
            max_catchup_bytes: 64_000,
        }],
    }
}

pub fn client_directory(recipient: &Keypair, endpoint: String) -> PeerDirectoryDocument {
    PeerDirectoryDocument {
        schema: PHEROMONE_PEER_DIRECTORY_SCHEMA.to_string(),
        local_kernel_id: "did:chio:llamaworks".to_string(),
        issued_at_unix_ms: now_ms() - 1_000,
        expires_at_unix_ms: now_ms() + 3_600_000,
        peers: vec![PeerDirectoryEntry {
            kernel_id: "did:chio:buyer-kernel".to_string(),
            public_key: recipient.public_key(),
            endpoint,
            treaty_subscriptions: vec!["treaty:buyer-llamaworks:support-ops".to_string()],
            relay_role: RelayRole::Receiver,
            allowed_subject_class_namespaces: vec!["dev.chio.support".to_string()],
            accepted_ladder_refs: vec![RelayLadderRef {
                ladder_manifest_id: "ladder:buyer:support:v1".to_string(),
                ladder_manifest_sha256: "b".repeat(64),
                expires_at_unix_ms: now_ms() + 3_600_000,
            }],
            max_batch_frames: 8,
            max_catchup_frames: 16,
            max_catchup_bytes: 64_000,
        }],
    }
}

use async_trait::async_trait;
use chio_core_types::crypto::{canonical_json_bytes, sha256_hex};
use chio_federation::pheromone_gossip::*;
use chio_pheromone::*;
use chio_pheromone_relay::*;
use chio_pheromone_runtime::{
    PheromoneBatchOutcome, PheromoneFrameReport, PheromoneReceiveReport,
    PHEROMONE_RECEIVE_REPORT_SCHEMA,
};
use std::sync::Arc;
// A bounded local origin-to-receiver host: one frame per request, no forwarding.
// It authenticates the relay request through PheromoneRelayService, then validates
// the passport signature, receiver policy, observation nonce, and capacity here.
pub struct Receiver {
    pub observations: Arc<InMemoryPheromoneSubstrate>,
    pub context: PheromoneValidationContext,
}
#[async_trait]
impl RelayBatchReceiver for Receiver {
    async fn receive_batch(
        &self,
        batch: PheromoneGossipBatch,
        sender: String,
        now: u64,
    ) -> Result<PheromoneReceiveReport, PheromoneRelayError> {
        if batch.frames.len() != 1 {
            return Err(PheromoneRelayError::SenderMismatch(
                "this host requires exactly one origin frame".into(),
            ));
        }
        let frame = &batch.frames[0];
        if frame.origin_kernel_id != sender
            || frame.gossiping_peer_kernel_id != sender
            || frame.transit_chain.is_some()
        {
            return Err(PheromoneRelayError::SenderMismatch(
                "origin-only receiver".into(),
            ));
        }
        let mut context = self.context.clone();
        context.now_unix_ms = now;
        let result = self.observations.deposit(frame.deposit.clone(), &context);
        let accepted = result.is_ok();
        let (code, detail) = match result {
            Ok(()) => ("accepted".into(), "passport observation admitted".into()),
            Err(error) => (error.code().into(), error.to_string()),
        };
        Ok(PheromoneReceiveReport {
            schema: PHEROMONE_RECEIVE_REPORT_SCHEMA.into(),
            accepted,
            batch_outcome: if accepted {
                PheromoneBatchOutcome::Accepted
            } else {
                PheromoneBatchOutcome::Rejected
            },
            accepted_frame_count: u64::from(accepted),
            rejected_frame_count: u64::from(!accepted),
            batch_sha256: sha256_hex(
                &canonical_json_bytes(&batch)
                    .map_err(|e| PheromoneRelayError::SenderMismatch(e.to_string()))?,
            ),
            recipient_kernel_id: batch.recipient_kernel_id.clone(),
            authenticated_sender_kernel_id: sender,
            received_at_unix_ms: now,
            frames: vec![PheromoneFrameReport {
                frame_index: 0,
                accepted,
                code,
                detail,
                deposit_nonce: Some(frame.deposit.body.nonce.clone()),
            }],
        })
    }
}

use swarm_authority_examples::observation;
#[path = "../relay_receiver.rs"]
mod relay_receiver;
use chio_core_types::Keypair;
use chio_pheromone::*;
use chio_pheromone_relay::*;
use std::{path::Path, sync::Arc};
use swarm_authority_examples::now_ms;
fn key(root: &Path, name: &str) -> Result<Keypair, Box<dyn std::error::Error>> {
    let json: serde_json::Value =
        serde_json::from_slice(&std::fs::read(root.join(format!("{name}-key.json")))?)?;
    Ok(Keypair::from_seed_hex(
        json["seedHex"].as_str().ok_or("missing seedHex")?,
    )?)
}
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let root = std::path::PathBuf::from(
        std::env::args()
            .nth(1)
            .unwrap_or_else(|| "relay-state".into()),
    );
    let passport = key(&root, "passport")?;
    let receiver = key(&root, "receiver")?;
    let document: PeerDirectoryDocument =
        serde_json::from_slice(&std::fs::read(root.join("receiver-directory.json"))?)?;
    let directory = PeerDirectory::from_document(document, now_ms())?;
    let mut context = observation::live_context(&passport, &receiver);
    // This receiver owns the window and passport-admission configuration.
    for policy in &mut context.scarcity_policies {
        policy.window_start_unix_ms = now_ms() - 3_600_000;
        policy.window_end_unix_ms = now_ms() + 3_600_000;
        policy.window_id = scarcity_window_id(policy, &policy.treaty_scope[0])?;
        policy.policy_sha256 = scarcity_policy_sha256(policy)?;
    }
    let service = PheromoneRelayService::new(
        PheromoneRelayConfig {
            local_kernel_id: "did:chio:buyer-kernel".into(),
            profile: RelayProfile::LocalDev,
            now_unix_ms: now_ms(),
            freshness_window_ms: 60_000,
            max_body_bytes: 256_000,
            use_system_clock: true,
            operator_token: None,
            report_dir: Some(root.join("reports")),
        },
        directory,
        Arc::new(relay_receiver::Receiver {
            observations: Arc::new(InMemoryPheromoneSubstrate::new()),
            context,
        }),
        Arc::new(SqlitePheromoneRelayStore::open(
            root.join("receiver.sqlite"),
        )?),
    );
    let listener = tokio::net::TcpListener::bind("127.0.0.1:18080").await?;
    println!("receiver listening: http://127.0.0.1:18080");
    service.serve(listener).await?;
    Ok(())
}