//! A REAL end-to-end run of the Joule Desk `room` binding against a live LiveKit server.
//!
//! Drives all four seams over a real WebRTC room:
//!   JOIN   — `@joule` connects to the server under a real access-token grant.
//!   TURN   — a second participant publishes turn requests on the room's data plane;
//!            each is gated OUTSIDE the model (gate-before-generate).
//!   METER  — an allowed turn runs the (stand-in) model and its usage is metered in joules.
//!   RECEIPT— one `efficiency/1` seals the whole call and is verified under the steward key.
//!
//! Run (needs libwebrtc built + a dev server on ws://127.0.0.1:7880):
//!   livekit-server --dev &
//!   cargo run -p jouledesk-livekit --features livekit --example live_room

use std::time::Duration;

use jouledesk_livekit::{RoomAgent, Turn, Usage};
use livekit::{DataPacket, Room, RoomEvent, RoomOptions};
use livekit_api::access_token::{AccessToken, VideoGrants};
use tokio::time::timeout;

const URL: &str = "ws://127.0.0.1:7880";
const KEY: &str = "devkey";
const SECRET: &str = "secret";
const ROOM: &str = "standup";

/// Mint a real LiveKit access token — this is the join grant the JOIN seam rides.
fn grant(identity: &str) -> String {
    AccessToken::with_api_key(KEY, SECRET)
        .with_identity(identity)
        .with_name(identity)
        .with_grants(VideoGrants {
            room_join: true,
            room: ROOM.to_string(),
            can_publish: true,
            can_subscribe: true,
            can_publish_data: true,
            ..Default::default()
        })
        .to_jwt()
        .expect("mint token")
}

/// A turn request arrives on the wire as `capability|cost_uj|intent`.
fn parse_turn(bytes: &[u8]) -> Option<Turn> {
    let s = std::str::from_utf8(bytes).ok()?;
    let mut it = s.splitn(3, '|');
    let capability = it.next()?.to_string();
    let cost_uj: u64 = it.next()?.parse().ok()?;
    let intent = it.next()?.to_string();
    Some(Turn { speaker: "dana".into(), capability, intent, cost_uj })
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    println!("── Joule Desk · room binding · LIVE against {URL} ──\n");

    // JOIN — @joule connects to the real server; the token IS the grant.
    let (agent_room, mut agent_events) =
        Room::connect(URL, &grant("joule"), RoomOptions::default()).await?;
    println!(
        "JOIN   @joule connected  room={}  identity={}",
        agent_room.name(),
        agent_room.local_participant().identity().0
    );
    let mut agent = RoomAgent::join(ROOM, &["answer", "summarize"], 5_000_000);
    println!(
        "       grant: caps=[answer, summarize]  ceiling=5_000_000 uj  steward={}\n",
        agent.steward()
    );

    // A second participant joins the same real room and publishes turn requests.
    let (dana_room, _dana_events) =
        Room::connect(URL, &grant("dana"), RoomOptions::default()).await?;
    tokio::time::sleep(Duration::from_millis(1200)).await; // let the SFU wire both PCs

    let turns = [
        "answer|1000000|what is blocking the release?",
        "summarize|900000|recap the standup",
        "delete_recording|500000|wipe the recording", // unheld capability → refused
        "answer|4000000|long deep-dive that would blow the ceiling", // over remaining → refused
    ];
    for t in turns {
        dana_room
            .local_participant()
            .publish_data(DataPacket {
                payload: t.as_bytes().to_vec(),
                topic: Some("turn".into()),
                reliable: true,
                destination_identities: vec![],
            })
            .await?;
    }
    println!("       dana published {} turns on the room data plane\n", turns.len());

    // The stand-in for the STT → LLM → TTS pipeline (non-capturing → Copy, reusable).
    let run_model = |_t: &Turn| Usage { tokens_in: 1200, tokens_out: 240, energy_uj: 720_000 };

    // TURN + METER — gate each inbound turn outside the model; meter the ones that run.
    let want = turns.len();
    let mut seen = 0usize;
    loop {
        match timeout(Duration::from_secs(12), agent_events.recv()).await {
            Ok(Some(RoomEvent::DataReceived { payload, .. })) => {
                let Some(turn) = parse_turn(&payload) else { continue };
                let outcome = agent.on_turn(&turn, run_model);
                println!(
                    "TURN   {:<16} cost={:>9} uj  →  {:?}",
                    turn.capability, turn.cost_uj, outcome
                );
                let _ = agent_room
                    .local_participant()
                    .publish_data(DataPacket {
                        payload: format!("{outcome:?}").into_bytes(),
                        topic: Some("reply".into()),
                        reliable: true,
                        destination_identities: vec![],
                    })
                    .await;
                seen += 1;
                if seen == want {
                    break;
                }
            }
            Ok(Some(_)) => {} // other room events (participants, tracks) — ignore
            Ok(None) => break,
            Err(_) => {
                println!("       (timed out waiting for turns — got {seen}/{want})");
                break;
            }
        }
    }

    // RECEIPT — seal one efficiency/1 for the whole call and verify it under the steward key.
    println!(
        "\nSEAL   completed_turns={}  metered={} uj  sealed_decisions={}",
        agent.completed_turns(),
        agent.metered_uj(),
        agent.turn_decisions()
    );
    let sealed = agent.shutdown();
    let receipt = jcp_efficiency::verify(&sealed, agent.steward(), &agent.steward_pubkey())
        .expect("the session receipt verifies under the steward key");
    println!(
        "RCPT   VERIFIED  task={}  tokens_out={}  energy_uj={}  ({} bytes sealed)",
        receipt.task,
        receipt.tokens_out,
        receipt.energy_uj,
        sealed.len()
    );

    agent_room.close().await.ok();
    dana_room.close().await.ok();
    println!("\n✓ four seams driven over a live LiveKit room; receipt verifies.");
    Ok(())
}