liminal-server 0.25.4

use super::*;

/// Lys's HTTP as the service bindings see it: every POST answered 200 and
/// recorded by route, so a seat start is proven at the request Lys received.
#[derive(Debug, Default)]
struct ServiceLys {
    posts: Mutex<Vec<String>>,
}

impl crate::services::lys_http::LysHttp for ServiceLys {
    fn get(
        &self,
        _bearer: &str,
        path: &str,
    ) -> Result<crate::services::lys_http::LysAnswer, crate::services::ServiceEffectError> {
        Err(crate::services::ServiceEffectError(format!(
            "fixture has no answer for GET {path}"
        )))
    }

    fn post(
        &self,
        _bearer: &str,
        path: &str,
        _body: &serde_json::Value,
    ) -> Result<crate::services::lys_http::LysAnswer, crate::services::ServiceEffectError> {
        self.posts
            .lock()
            .map_err(|_| crate::services::ServiceEffectError("poisoned post recorder".to_owned()))?
            .push(path.to_owned());
        Ok(crate::services::lys_http::LysAnswer {
            status: 200,
            body: serde_json::json!({}),
        })
    }
}

/// The publish path behind the key sender: records every subject published
/// on, so a refused key send is proven to have published nothing.
#[derive(Debug, Default)]
struct ServicePublishes {
    published: Mutex<Vec<String>>,
}

impl crate::services::keys::ServicePublisher for ServicePublishes {
    fn publish(
        &self,
        subject: &str,
        _envelope: &MessageEnvelope,
        _operation: &str,
    ) -> Result<(), crate::services::ServiceEffectError> {
        self.published
            .lock()
            .map_err(|_| {
                crate::services::ServiceEffectError("poisoned publish recorder".to_owned())
            })?
            .push(subject.to_owned());
        Ok(())
    }
}

fn install_services(
    runtime: &ConnectionRuntime,
) -> Result<(Arc<ServiceLys>, Arc<ServicePublishes>), Failure> {
    let lys = Arc::new(ServiceLys::default());
    let sink = Arc::new(ServicePublishes::default());
    let http: Arc<dyn crate::services::lys_http::LysHttp> = lys.clone();
    let publisher: Arc<dyn crate::services::keys::ServicePublisher> = sink.clone();
    let bindings = crate::services::production::ServiceBindings::new(
        &http,
        "machine-bearer",
        "mac-1".to_owned(),
        &publisher,
    );
    runtime
        .install_service_bindings(Arc::new(bindings))
        .map_err(|_| "service bindings were already installed")?;
    Ok((lys, sink))
}

fn service_message(
    runtime: &ConnectionRuntime,
    state: &mut ConnectionProcessState,
    body: &str,
) -> FrameAction {
    apply_frame(
        PID,
        runtime,
        state,
        Frame::ConversationMessage {
            flags: liminal::protocol::CONVERSATION_REPLY_REQUESTED_FLAG,
            stream_id: STREAM,
            conversation_id: CONVERSATION,
            envelope: MessageEnvelope::new(
                SchemaId::new([3; SchemaId::WIRE_LEN]),
                CausalContext::independent(),
                body.as_bytes().to_vec(),
            ),
        },
    )
}

/// The JSON body a service answered on its conversation.
fn service_reply(action: FrameAction) -> Result<serde_json::Value, Failure> {
    match action {
        FrameAction::Respond(Frame::ConversationMessage {
            flags: liminal::protocol::CONVERSATION_REPLY_REQUESTED_FLAG,
            stream_id: STREAM,
            conversation_id: CONVERSATION,
            envelope,
        }) => Ok(serde_json::from_slice(&envelope.payload)?),
        other => Err(format!("the service did not reply on its conversation: {other:?}").into()),
    }
}

/// The refusal a service request answered, carrying the service reason code.
fn service_refusal(action: FrameAction) -> Result<String, Failure> {
    match action {
        FrameAction::Respond(Frame::ConversationError {
            conversation_id: CONVERSATION,
            stream_id: STREAM,
            reason_code: crate::server::connection::service_route::SERVICE_REFUSED_CODE,
            message: Some(message),
            ..
        }) => Ok(message),
        other => Err(
            format!("the service request was not refused on its conversation: {other:?}").into(),
        ),
    }
}

#[test]
fn a_seat_starts_on_an_outright_call_grant_through_the_service_route() -> Result<(), Failure> {
    let grants = Arc::new(Grants::default());
    let door = Arc::new(Door::default());
    grants.grant("seats.archie", HotAction::Call)?;
    let (runtime, mut state) = admitted(&grants, &door)?;
    let (lys, publishes) = install_services(&runtime)?;
    assert!(matches!(
        open(&runtime, &mut state, "seats.archie"),
        FrameAction::NoResponse
    ));

    let reply = service_reply(service_message(
        &runtime,
        &mut state,
        r#"{"operation":"op-1","payload":{}}"#,
    ))?;
    assert_eq!(reply["outcome"], "acted", "the seat start acted: {reply}");
    assert!(
        reply["grant"].is_string(),
        "the reply names the grant: {reply}"
    );
    assert_eq!(
        *lys.posts.lock().map_err(|_| "poisoned")?,
        vec!["agents/archie/start-command".to_owned()],
        "the seat was started through Lys once, on the configured machine"
    );
    assert!(
        publishes
            .published
            .lock()
            .map_err(|_| "poisoned")?
            .is_empty()
    );
    assert_eq!(door.messages()?, 0, "the request never reached the channel");
    Ok(())
}

#[test]
fn a_session_key_send_is_decided_live_and_sends_no_key_when_refused() -> Result<(), Failure> {
    let grants = Arc::new(Grants::default());
    let door = Arc::new(Door::default());
    // `call` on the subject opens the conversation; the send itself is the
    // deliberate decision, which this pass has no live answer for.
    grants.grant("sessions.s1.keys", HotAction::Call)?;
    let (runtime, mut state) = admitted(&grants, &door)?;
    let (lys, publishes) = install_services(&runtime)?;
    assert!(matches!(
        open(&runtime, &mut state, "sessions.s1.keys"),
        FrameAction::NoResponse
    ));

    let refusal = service_refusal(service_message(
        &runtime,
        &mut state,
        r#"{"operation":"op-2","payload":{"keys":"sealed"}}"#,
    ))?;
    assert!(
        refusal.contains(r#""id":"sessions.s1.keys""#) && refusal.contains(r#""action":"call""#),
        "the refusal names the resource and the action: {refusal}"
    );
    assert!(
        publishes
            .published
            .lock()
            .map_err(|_| "poisoned")?
            .is_empty(),
        "no key was published on a refused send"
    );
    assert!(lys.posts.lock().map_err(|_| "poisoned")?.is_empty());
    assert_eq!(door.messages()?, 0);

    grants.allow_open("sessions.s1.keys", AccessAction::Call)?;
    let reply = service_reply(service_message(
        &runtime,
        &mut state,
        r#"{"operation":"op-3","payload":{"keys":"sealed"}}"#,
    ))?;
    assert_eq!(
        reply["outcome"], "acted",
        "an outright live answer sends once: {reply}"
    );
    assert_eq!(
        *publishes.published.lock().map_err(|_| "poisoned")?,
        vec!["sessions.s1.keys".to_owned()]
    );
    Ok(())
}

#[test]
fn a_service_request_without_bindings_or_operation_is_refused_by_name() -> Result<(), Failure> {
    let grants = Arc::new(Grants::default());
    let door = Arc::new(Door::default());
    grants.grant("alarms", HotAction::Call)?;
    let (runtime, mut state) = admitted(&grants, &door)?;
    assert!(matches!(
        open(&runtime, &mut state, "alarms"),
        FrameAction::NoResponse
    ));
    let refusal = service_refusal(service_message(
        &runtime,
        &mut state,
        r#"{"operation":"op-4"}"#,
    ))?;
    assert!(
        refusal.contains("no service bindings are installed"),
        "refused by name before any decision: {refusal}"
    );

    install_services(&runtime)?;
    let refusal = service_refusal(service_message(&runtime, &mut state, r#"{"payload":{}}"#))?;
    assert!(
        refusal.contains("names no operation id"),
        "a request without an operation does nothing, by name: {refusal}"
    );
    assert_eq!(door.messages()?, 0);
    Ok(())
}

/// SRV-027 R2 (acceptance 8): the Lys `Call` decision precedes the routed
/// push. With a worker registered on the subject, a message under a pass
/// that lacks `Call` is refused by the existing decision before the directory
/// is read: 0 lookups, 0 pushes, 0 channel messages. With `Call` granted the
/// message takes the routed path (1 lookup, still 0 channel messages); this
/// fixture's runtime has no scheduler, so the push is refused by name there
/// and that refusal reaches the caller on its conversation.
#[test]
fn lys_call_decision_precedes_the_push() -> Result<(), Failure> {
    const WORKER: u64 = 12;
    let grants = Arc::new(Grants::default());
    let door = Arc::new(Door::default());
    let (runtime, mut state) = admitted(&grants, &door)?;
    runtime.register_for_tests(WORKER)?;
    runtime.set_registration(
        WORKER,
        WorkerRegistration {
            namespaces: Vec::new(),
            task_queue: "orders".to_owned(),
            node: None,
            activity_types: Vec::new(),
            identity: "orders-worker".to_owned(),
            activities: Vec::new(),
        },
    )?;
    grants.grant("orders", HotAction::Call)?;
    assert!(matches!(
        open(&runtime, &mut state, "orders"),
        FrameAction::NoResponse
    ));

    grants.revoke("orders", HotAction::Call)?;
    let refusal = conversation_refusal(message(&runtime, &mut state))?;
    assert!(
        refusal.contains(r#""action":"call""#) && refusal.contains(r#""id":"orders""#),
        "{refusal}"
    );
    assert_eq!(
        runtime.directory_lookups(),
        0,
        "the directory is not read before the Call decision"
    );
    assert_eq!(runtime.pending_push_count(), 0, "nothing was pushed");
    assert_eq!(door.messages()?, 0, "a refused message reaches no channel");

    grants.grant("orders", HotAction::Call)?;
    let routed = message(&runtime, &mut state);
    assert_eq!(
        runtime.directory_lookups(),
        1,
        "the decision passed and the directory was read once"
    );
    assert_eq!(
        door.messages()?,
        0,
        "a served subject never reaches the channel"
    );
    let words = conversation_refusal(routed)?;
    assert!(
        words.contains("cannot route to connection process 12"),
        "the fixture has no scheduler, so the push is refused by name: {words}"
    );
    Ok(())
}

/// Places every service, refusing nothing.
#[derive(Debug)]
struct Placed;

impl crate::services::lifecycle::Placer for Placed {
    fn place(
        &self,
        _operation: &str,
        _child: (&str, &str),
        _parent: (&str, &str),
    ) -> Result<(), String> {
        Ok(())
    }
}

/// A registry of the `jobs` template with each named service placed at its
/// millisecond.
fn jobs_placed(
    dir: &std::path::Path,
    placed: &[(&str, u64)],
) -> Result<Arc<crate::services::lifecycle::Lifecycle>, Failure> {
    let lifecycle = crate::services::lifecycle::Lifecycle::open(
        &[crate::config::types::ServiceTemplateConfig {
            name: "jobs".to_owned(),
            command: std::path::PathBuf::from("/bin/sh"),
            args: vec!["-c".to_owned(), "exit 0".to_owned()],
            subjects: vec!["jobs".to_owned()],
            env: Vec::new(),
            stop_timeout_ms: 2_000,
        }],
        dir.to_path_buf(),
        "net".to_owned(),
        Some(Arc::new(Placed)),
    )
    .map_err(|error| format!("the registry opens: {error:?}"))?;
    for (name, at_ms) in placed {
        lifecycle
            .create("jobs", name, "a", &format!("op-{name}"), *at_ms)
            .map_err(|error| format!("{error:?}"))?;
    }
    Ok(lifecycle)
}

#[test]
fn a_call_on_a_service_placed_after_the_pass_is_asked_of_lys_live() -> Result<(), Failure> {
    // the pass was issued at 1,000 s; jobs.late was placed after it, jobs.early before
    let grants = Arc::new(Grants {
        issued_at: Some(1_000),
        ..Grants::default()
    });
    let door = Arc::new(Door::default());
    let (runtime, mut state) = admitted(&grants, &door)?;
    let dir = tempfile::tempdir()?;
    let lifecycle = jobs_placed(dir.path(), &[("early", 999_999), ("late", 1_000_000)])?;
    assert!(runtime.install_lifecycle_for_tests(lifecycle));
    for subject in ["jobs.early", "jobs.late", "elsewhere"] {
        grants.allow_open(subject, HotAction::Call.into())?;
    }

    // placed before the pass: its offline refusal stands, Lys is not asked
    let refused = conversation_refusal(open(&runtime, &mut state, "jobs.early"))?;
    assert!(refused.contains("jobs.early"), "{refused}");
    // no service holds the subject: refused offline as before
    let refused = conversation_refusal(open(&runtime, &mut state, "elsewhere"))?;
    assert!(refused.contains("elsewhere"), "{refused}");
    assert!(door.opens()?.is_empty());
    // placed after the pass: Lys answers outright, and the conversation opens
    let opened = open(&runtime, &mut state, "jobs.late");
    assert!(
        !matches!(
            opened,
            FrameAction::Respond(Frame::ConversationError { .. })
        ),
        "{opened:?}"
    );
    assert_eq!(door.opens()?, vec![(CONVERSATION, "jobs.late".to_owned())]);
    Ok(())
}

#[test]
fn a_late_service_lys_refuses_live_is_refused_with_the_offline_words() -> Result<(), Failure> {
    let grants = Arc::new(Grants {
        issued_at: Some(1_000),
        ..Grants::default()
    });
    let door = Arc::new(Door::default());
    let (runtime, mut state) = admitted(&grants, &door)?;
    let dir = tempfile::tempdir()?;
    let lifecycle = jobs_placed(dir.path(), &[("late", 1_000_000)])?;
    assert!(runtime.install_lifecycle_for_tests(lifecycle));
    let refused = conversation_refusal(open(&runtime, &mut state, "jobs.late"))?;
    assert!(refused.contains("jobs.late"), "{refused}");
    assert!(door.opens()?.is_empty());
    Ok(())
}