liminal-server 0.24.9

//! App-credential placement is durable, idempotent and attempted once per boot.

use super::tap_placement::{PlacementBook, PlacementResult};
use crate::server::services_boot::{registry_dir, tap_placement_journal_path};
use crate::services::ServiceEffectError;
use crate::services::lifecycle::{Lifecycle, ServiceRecord, ServiceState};
use crate::services::lys_http::{LysAnswer, LysHttp};
use std::sync::{Arc, Mutex};

type TestResult = Result<(), Box<dyn std::error::Error>>;

fn registry_fixture(store: &std::path::Path) -> Result<ServiceRecord, Box<dyn std::error::Error>> {
    let record = ServiceRecord {
        id: "svc-studio".to_owned(),
        name: "studio".to_owned(),
        template: Some("studio".to_owned()),
        purpose: None,
        contracts: std::collections::BTreeMap::new(),
        builder: None,
        builder_grants: None,
        builder_bound: false,
        approved_by: Vec::new(),
        responsible: None,
        plan: crate::services::lifecycle::plan::Plan::default(),
        approvals: Vec::new(),
        subjects: vec!["jobs.studio".to_owned()],
        state: ServiceState::Asleep,
        created_by: "holder".to_owned(),
        transitions: vec![],
    };
    let dir = registry_dir(store);
    std::fs::create_dir_all(&dir)?;
    std::fs::write(dir.join("svc-studio.json"), serde_json::to_vec(&record)?)?;
    Ok(record)
}

fn boot_registry(store: &std::path::Path) -> Result<Arc<Lifecycle>, Box<dyn std::error::Error>> {
    Ok(
        Lifecycle::open(&[], registry_dir(store), "ablative".to_owned(), None)
            .map_err(|error| error.to_string())?,
    )
}

#[test]
fn tap_placement_journal_survives_the_services_registry_second_boot() -> TestResult {
    let dir = tempfile::tempdir()?;
    let store = dir.path().join("store");
    let record = registry_fixture(&store)?;
    let path = tap_placement_journal_path(&store, "ablative")?;
    let first = boot_registry(&store)?;
    assert_eq!(first.record(&record.id)?, record);
    let http = Http::new(200, serde_json::Value::Null);
    let mut book = PlacementBook::open(&path, "ablative")?;
    for subject in [
        "demo.hello",
        "embed.bge",
        "jobs.studio",
        "liminal.cost",
        "liminal.events",
    ] {
        assert_eq!(
            book.attempt(subject, &http, "app-credential")?,
            PlacementResult::Placed
        );
    }
    drop(book);
    drop(first);
    let path = tap_placement_journal_path(&store, "ablative")?;
    let second = boot_registry(&store)?;
    assert_eq!(second.record(&record.id)?, record);
    assert!(path.is_file());
    assert_eq!(path, dir.path().join("store-console-tap-placements.json"));
    assert_ne!(path.parent(), Some(registry_dir(&store).as_path()));
    let mut restarted = PlacementBook::open(&path, "ablative")?;
    for subject in [
        "demo.hello",
        "embed.bge",
        "jobs.studio",
        "liminal.cost",
        "liminal.events",
    ] {
        assert_eq!(
            restarted.attempt(subject, &http, "app-credential")?,
            PlacementResult::Skipped
        );
    }
    assert_eq!(http.calls()?, 5);
    Ok(())
}

#[test]
fn tap_placement_migrates_legacy_journal_before_registry_boot() -> TestResult {
    let dir = tempfile::tempdir()?;
    for has_current in [false, true] {
        let store = dir
            .path()
            .join(if has_current { "both" } else { "old-only" });
        let record = registry_fixture(&store)?;
        let path = tap_placement_journal_path(&store, "ablative")?;
        let http = Http::new(200, serde_json::Value::Null);
        let offline = Http::new(200, serde_json::Value::Null);
        *offline.answer.lock().map_err(|_| "answer poisoned")? =
            Err(ServiceEffectError("offline".to_owned()));
        if has_current {
            let mut current = PlacementBook::open(&path, "ablative")?;
            current.attempt("jobs.new", &http, "app-credential")?;
            assert!(
                current
                    .attempt("jobs.studio", &offline, "app-credential")
                    .is_err()
            );
        }
        let legacy = registry_dir(&store).join("console-tap-placements.json");
        let mut old = PlacementBook::open(&legacy, "ablative")?;
        old.attempt("jobs.studio", &http, "app-credential")?;
        assert!(old.attempt("jobs.old", &offline, "app-credential").is_err());
        drop(old);
        let path = tap_placement_journal_path(&store, "ablative")?;
        let registry = boot_registry(&store)?;
        assert_eq!(registry.record(&record.id)?, record);
        assert!(!legacy.exists());
        let mut migrated = PlacementBook::open(&path, "ablative")?;
        assert_eq!(migrated.pending(), vec!["jobs.old".to_owned()]);
        assert_eq!(
            migrated.attempt("jobs.studio", &http, "app-credential")?,
            PlacementResult::Skipped
        );
        if has_current {
            assert_eq!(
                migrated.attempt("jobs.new", &http, "app-credential")?,
                PlacementResult::Skipped
            );
        }
        assert_eq!(http.calls()?, 1 + usize::from(has_current));
        assert_eq!(tap_placement_journal_path(&store, "ablative")?, path);
    }
    Ok(())
}

#[derive(Debug)]
struct Http {
    calls: Mutex<Vec<(String, String, serde_json::Value)>>,
    answer: Mutex<Result<LysAnswer, ServiceEffectError>>,
    called: std::sync::Condvar,
}

impl Http {
    fn new(status: u16, body: serde_json::Value) -> Self {
        Self {
            calls: Mutex::new(vec![]),
            answer: Mutex::new(Ok(LysAnswer { status, body })),
            called: std::sync::Condvar::new(),
        }
    }
    fn wait_for_calls(&self, count: usize) -> Result<(), Box<dyn std::error::Error>> {
        let calls = self.calls.lock().map_err(|_| "calls poisoned")?;
        drop(
            self.called
                .wait_while(calls, |calls| calls.len() < count)
                .map_err(|_| "calls poisoned")?,
        );
        Ok(())
    }

    fn calls(&self) -> Result<usize, Box<dyn std::error::Error>> {
        Ok(self.calls.lock().map_err(|_| "calls poisoned")?.len())
    }
}

impl LysHttp for Http {
    fn get(&self, _: &str, _: &str) -> Result<LysAnswer, ServiceEffectError> {
        Err(ServiceEffectError("GET was not requested".to_owned()))
    }
    fn post(
        &self,
        bearer: &str,
        path: &str,
        body: &serde_json::Value,
    ) -> Result<LysAnswer, ServiceEffectError> {
        self.calls
            .lock()
            .map_err(|_| ServiceEffectError("calls poisoned".to_owned()))?
            .push((bearer.to_owned(), path.to_owned(), body.clone()));
        self.called.notify_all();
        match &*self
            .answer
            .lock()
            .map_err(|_| ServiceEffectError("answer poisoned".to_owned()))?
        {
            Ok(answer) => Ok(answer.clone()),
            Err(error) => Err(ServiceEffectError(error.to_string())),
        }
    }
}

#[test]
fn tap_placement_uses_app_credential_and_exact_lys_route_once_across_restart() -> TestResult {
    let dir = tempfile::tempdir()?;
    let store = dir.path().join("store");
    let record = registry_fixture(&store)?;
    let path = tap_placement_journal_path(&store, "ablative")?;
    let first = boot_registry(&store)?;
    assert_eq!(first.record(&record.id)?, record);
    let http = Http::new(
        200,
        serde_json::json!({"placed": {"child": "liminal.subject:console.tap.jobs.studio", "parent": "liminal.network:ablative", "restricted": false}}),
    );
    let mut book = PlacementBook::open(&path, "ablative")?;
    assert_eq!(
        book.attempt("jobs.studio", &http, "app-credential")?,
        PlacementResult::Placed
    );
    assert_eq!(
        book.attempt("jobs.studio", &http, "app-credential")?,
        PlacementResult::Skipped
    );
    let calls = http.calls.lock().map_err(|_| "calls poisoned")?;
    assert_eq!(calls.len(), 1);
    assert_eq!(calls[0].0, "app-credential");
    assert_eq!(calls[0].1, "apps/liminal/placements");
    assert_eq!(
        calls[0].2["child"],
        serde_json::json!({"kind": "liminal.subject", "id": "console.tap.jobs.studio"})
    );
    assert_eq!(
        calls[0].2["parent"],
        serde_json::json!({"kind": "liminal.network", "id": "ablative"})
    );
    assert_eq!(calls[0].2["operation"].as_str().map(str::len), Some(35));
    drop(calls);
    drop(book);
    drop(first);
    let second = boot_registry(&store)?;
    assert_eq!(second.record(&record.id)?, record);
    let mut restarted = PlacementBook::open(&path, "ablative")?;
    assert_eq!(
        restarted.attempt("jobs.studio", &http, "app-credential")?,
        PlacementResult::Skipped
    );
    assert_eq!(http.calls()?, 1);
    assert!(restarted.pending().is_empty());
    drop(restarted);
    let hook_http = Arc::new(Http::new(200, serde_json::Value::Null));
    let placements = super::tap_placement::TapPlacements::start(
        &path,
        "ablative",
        hook_http.clone(),
        "app-credential".to_owned(),
    )?;
    let now = std::time::Instant::now();
    let hub = super::ConsoleHub::new(
        "local",
        super::TopologyModel::new(
            super::ThisDevice {
                id: "local".to_owned(),
                liminal: "1".to_owned(),
                started: now,
                depot: false,
            },
            [],
            [],
            [],
            [],
        ),
        super::wall::WallBase::fixed(1_791_881_244_000, now),
    )?;
    hub.install_placements(placements.clone())?;
    for _ in 0..2 {
        hub.emit(super::TopologyChange::Service {
            subject: "jobs.hook".to_owned(),
            present: true,
        })?;
    }
    let built_in = super::tap_placement::BUILT_IN.len();
    hook_http.wait_for_calls(built_in + 1)?;
    drop(hub);
    drop(placements);
    let calls = taps_after_own(&hook_http)?;
    assert_eq!(calls.len(), 1, "the actual line event places its tap once");
    assert_eq!(calls[0].0, "app-credential");
    assert_eq!(calls[0].1, "apps/liminal/placements");
    assert_eq!(calls[0].2["child"]["id"], "console.tap.jobs.hook");
    Ok(())
}

#[test]
fn tap_placement_existing_parent_is_success_without_a_new_placement_log() -> TestResult {
    let dir = tempfile::tempdir()?;
    let mut book = PlacementBook::open(&dir.path().join("taps.json"), "ablative")?;
    let http = Http::new(
        400,
        serde_json::json!({"refusal": "placement_invalid", "reason": "placement_invalid: the resource is already placed in a parent"}),
    );
    assert_eq!(
        book.attempt("jobs.studio", &http, "app-credential")?,
        PlacementResult::Existing
    );
    assert_eq!(
        book.attempt("jobs.studio", &http, "app-credential")?,
        PlacementResult::Skipped
    );
    assert!(book.pending().is_empty());
    let refusal = Http::new(
        400,
        serde_json::json!({"refusal": "placement_invalid", "reason": "placement_invalid: wrong parent kind"}),
    );
    assert!(
        book.attempt("jobs.other", &refusal, "app-credential")
            .is_err(),
        "other placement refusals remain errors"
    );
    Ok(())
}

#[test]
fn tap_placement_unreachable_keeps_tap_serving_and_retries_only_next_boot() -> TestResult {
    let dir = tempfile::tempdir()?;
    let path = dir.path().join("taps.json");
    let http = Http::new(200, serde_json::Value::Null);
    *http.answer.lock().map_err(|_| "answer poisoned")? =
        Err(ServiceEffectError("lys_could_not_be_asked".to_owned()));
    let mut book = PlacementBook::open(&path, "ablative")?;
    assert!(
        book.attempt("jobs.studio", &http, "app-credential")
            .is_err()
    );
    assert_eq!(book.pending(), vec!["jobs.studio".to_owned()]);
    assert_eq!(
        book.attempt("jobs.studio", &http, "app-credential")?,
        PlacementResult::Skipped
    );
    assert_eq!(http.calls()?, 1);
    let started = std::time::Instant::now();
    let hub = super::ConsoleHub::new(
        "local",
        super::TopologyModel::new(
            super::ThisDevice {
                id: "local".to_owned(),
                liminal: "1".to_owned(),
                started,
                depot: false,
            },
            [],
            [],
            [],
            [],
        ),
        super::wall::WallBase::fixed(1_791_881_244_000, started),
    )?;
    hub.attach("console.tap.jobs.studio")?;
    assert_eq!(
        hub.attached_subjects()?,
        vec!["console.tap.jobs.studio".to_owned()]
    );
    drop(book);
    let mut restarted = PlacementBook::open(&path, "ablative")?;
    *http.answer.lock().map_err(|_| "answer poisoned")? = Ok(LysAnswer {
        status: 200,
        body: serde_json::Value::Null,
    });
    assert_eq!(
        restarted.attempt("jobs.studio", &http, "app-credential")?,
        PlacementResult::Placed
    );
    assert_eq!(http.calls()?, 2);
    assert!(restarted.pending().is_empty());
    Ok(())
}

#[derive(Clone)]
struct Log(Arc<Mutex<Vec<u8>>>);
impl std::io::Write for Log {
    fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
        self.0
            .lock()
            .map_err(|_| std::io::Error::other("log poisoned"))?
            .extend_from_slice(bytes);
        Ok(bytes.len())
    }
    fn flush(&mut self) -> std::io::Result<()> {
        Ok(())
    }
}

#[test]
fn tap_placement_logs_info_only_on_the_first_new_placement() -> TestResult {
    let dir = tempfile::tempdir()?;
    let store = dir.path().join("store");
    let record = registry_fixture(&store)?;
    let path = tap_placement_journal_path(&store, "ablative")?;
    let log = Log(Arc::new(Mutex::new(vec![])));
    let writer = log.clone();
    let subscriber = tracing_subscriber::fmt()
        .without_time()
        .with_ansi(false)
        .with_writer(move || writer.clone())
        .finish();
    let http = Http::new(200, serde_json::Value::Null);
    let mut first_boot_log_bytes = 0;
    tracing::subscriber::with_default(subscriber, || -> TestResult {
        let first = boot_registry(&store)?;
        assert_eq!(first.record(&record.id)?, record);
        let mut book = PlacementBook::open(&path, "ablative")?;
        book.attempt("jobs.studio", &http, "app-credential")?;
        book.attempt("jobs.studio", &http, "app-credential")?;
        drop(book);
        drop(first);
        first_boot_log_bytes = log.0.lock().map_err(|_| "log poisoned")?.len();
        let second = boot_registry(&store)?;
        assert_eq!(second.record(&record.id)?, record);
        PlacementBook::open(&path, "ablative")?.attempt("jobs.studio", &http, "app-credential")?;
        Ok(())
    })?;
    let bytes = log.0.lock().map_err(|_| "log poisoned")?;
    let text = std::str::from_utf8(&bytes)?;
    assert_eq!(text.matches("TAP_SUBJECT_PLACED").count(), 1);
    assert!(text.contains("INFO"));
    assert!(text.contains("console.tap.jobs.studio"));
    assert!(!text.contains("app-credential"));
    let second_boot =
        std::str::from_utf8(bytes.get(first_boot_log_bytes..).ok_or("log boundary")?)?;
    assert_eq!(second_boot.matches("TAP_SUBJECT_PLACED").count(), 0);
    drop(bytes);
    assert_eq!(http.calls()?, 1);
    Ok(())
}

#[test]
fn the_exchanges_own_subjects_are_placed_once_across_restart() -> TestResult {
    let dir = tempfile::tempdir()?;
    let path = dir.path().join("taps.json");
    let first = Arc::new(Http::new(200, serde_json::Value::Null));
    drop(super::tap_placement::TapPlacements::start(
        &path,
        "ablative",
        first.clone(),
        "app-credential".to_owned(),
    )?);
    assert_eq!(first.calls()?, super::tap_placement::BUILT_IN.len());
    let second = Arc::new(Http::new(200, serde_json::Value::Null));
    drop(super::tap_placement::TapPlacements::start(
        &path,
        "ablative",
        second.clone(),
        "app-credential".to_owned(),
    )?);
    assert_eq!(second.calls()?, 0, "a restart asks Lys again");
    Ok(())
}

type Call = (String, String, serde_json::Value);

/// The tap placements a hook saw, once the exchange's own subjects are checked to have been
/// placed as themselves, each once, never as taps.
fn taps_after_own(http: &Http) -> Result<Vec<Call>, Box<dyn std::error::Error>> {
    let all = http.calls.lock().map_err(|_| "calls poisoned")?.clone();
    let (taps, own): (Vec<Call>, Vec<Call>) = all.into_iter().partition(|call| {
        call.2["child"]["id"]
            .as_str()
            .is_some_and(|id| id.starts_with("console."))
    });
    let own: Vec<_> = own
        .iter()
        .filter_map(|call| call.2["child"]["id"].as_str())
        .collect();
    assert_eq!(own, super::tap_placement::BUILT_IN);
    Ok(taps)
}