liminal-server 0.25.4

use super::*;

#[test]
fn a_conversation_opens_only_with_call_on_its_subject() -> Result<(), Failure> {
    let grants = Arc::new(Grants::default());
    let door = Arc::new(Door::default());
    let (runtime, mut state) = admitted(&grants, &door)?;

    let refusal = conversation_refusal(open(&runtime, &mut state, "orders"))?;
    assert!(
        refusal.contains(r#""action":"call""#) && refusal.contains(r#""id":"orders""#),
        "the refusal names the subject and the action by name: {refusal}"
    );
    assert_eq!(door.opens()?, Vec::new(), "a refused open spawns nothing");
    assert!(!state.conversations.contains_key(&CONVERSATION));

    grants.grant("orders", HotAction::Call)?;
    assert!(matches!(
        open(&runtime, &mut state, "orders"),
        FrameAction::NoResponse
    ));
    assert_eq!(door.opens()?, vec![(CONVERSATION, "orders".to_owned())]);
    assert!(state.conversations.contains_key(&CONVERSATION));
    Ok(())
}

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

    // The grant goes between open and message: the message is its own
    // decision on the conversation's retained subject, never the open's.
    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!(door.messages()?, 0, "a refused message is never forwarded");

    grants.grant("orders", HotAction::Call)?;
    assert!(matches!(
        message(&runtime, &mut state),
        FrameAction::NoResponse
    ));
    assert_eq!(door.messages()?, 1);
    Ok(())
}

#[test]
fn a_worker_registration_is_decided_on_its_task_queue_then_every_namespace() -> Result<(), Failure>
{
    let grants = Arc::new(Grants::default());
    let door = Arc::new(Door::default());
    let (runtime, mut state) = admitted(&grants, &door)?;
    grants.grant("orders", HotAction::Serve)?;
    grants.grant("invoices", HotAction::Serve)?;
    let register = |state: &mut ConnectionProcessState| {
        apply_frame(
            PID,
            &runtime,
            state,
            Frame::WorkerRegister {
                flags: 0,
                registration: WorkerRegistration {
                    namespaces: vec!["orders".to_owned(), "invoices".to_owned()],
                    task_queue: "findings".to_owned(),
                    node: None,
                    activity_types: Vec::new(),
                    identity: "fixture-holder".to_owned(),
                    activities: Vec::new(),
                },
            },
        )
    };

    let FrameAction::Respond(Frame::WorkerRegisterAck {
        outcome: WorkerRegisterOutcome::Rejected { reason },
        ..
    }) = register(&mut state)
    else {
        return Err("a worker serving a task queue its pass lacks was not refused".into());
    };
    assert!(
        reason.contains(r#""action":"serve""#) && reason.contains(r#""id":"findings""#),
        "the refusal names the task queue the pass lacks: {reason}"
    );
    assert!(reason.contains(r#""resource":"liminal.subject:findings""#));
    assert_eq!(
        runtime.stored_registration_for_tests(PID)?,
        None,
        "a refused registration is never stored"
    );

    assert_eq!(runtime.serving("findings"), None);
    assert!(state.worker_flows.is_none());
    grants.grant("findings", HotAction::Serve)?;
    assert!(matches!(
        register(&mut state),
        FrameAction::Respond(Frame::WorkerRegisterAck {
            outcome: WorkerRegisterOutcome::Accepted,
            ..
        })
    ));
    assert_eq!(runtime.serving("findings"), Some(PID));
    grants.revoke("invoices", HotAction::Serve)?;
    let FrameAction::Respond(Frame::WorkerRegisterAck {
        outcome: WorkerRegisterOutcome::Rejected { reason },
        ..
    }) = register(&mut state)
    else {
        return Err("a granted task queue bypassed a missing namespace grant".into());
    };
    assert!(
        reason.contains(r#""action":"serve""#) && reason.contains(r#""id":"invoices""#),
        "the namespace is still decided after the task queue: {reason}"
    );
    Ok(())
}

/// A subscription inbox that never delivers: the signal under test is decided
/// before anything is pulled from it.
#[derive(Debug)]
struct Empty;

impl SubscriptionResource for Empty {
    fn unsubscribe(self: Box<Self>) -> Result<(), ServerError> {
        Ok(())
    }

    fn try_next(&mut self) -> Option<liminal::envelope::Envelope> {
        None
    }

    fn has_pending(&self) -> bool {
        false
    }
}

/// Installs one subscription on `subject`, delivering on `STREAM`, with one
/// issued delivery `ISSUED` in its pressure ledger for a signal to reference.
fn subscribed(state: &mut ConnectionProcessState, subject: &str) {
    let mut subscription = ConnectionSubscription::new(
        SUBSCRIPTION,
        SchemaId::new([1; SchemaId::WIRE_LEN]),
        Box::new(Empty),
    );
    subscription.set_channel(subject.to_owned());
    subscription.set_stream_id(STREAM);
    subscription.set_pressure_ledger_depth(4);
    subscription.issued_delivery(ISSUED);
    state.subscriptions.insert(SUBSCRIPTION, subscription);
}

/// Whether the subscription's ledger still holds `ISSUED`: a granted signal
/// consumes it, a refused one leaves it.
fn ledger_holds_issued(state: &mut ConnectionProcessState) -> Result<bool, Failure> {
    let subscription = state
        .subscriptions
        .get_mut(&SUBSCRIPTION)
        .ok_or("the subscription under test is gone")?;
    Ok(subscription.match_pressure(&ISSUED.to_string(), true))
}

fn accept(runtime: &ConnectionRuntime, state: &mut ConnectionProcessState) -> FrameAction {
    apply_frame(
        PID,
        runtime,
        state,
        Frame::Accept {
            flags: 0,
            stream_id: STREAM,
            referenced_message_id: MessageId::new(ISSUED.to_string()),
        },
    )
}

#[test]
fn a_backpressure_signal_is_decided_on_its_subscriptions_subject() -> Result<(), Failure> {
    let grants = Arc::new(Grants::default());
    let door = Arc::new(Door::default());
    let (runtime, mut state) = admitted(&grants, &door)?;
    subscribed(&mut state, "orders");

    // No refusal frame exists for a signal: the refusal is that nothing
    // happens, and the ledger entry the signal referenced is still there.
    assert!(matches!(
        accept(&runtime, &mut state),
        FrameAction::NoResponse
    ));
    assert!(
        ledger_holds_issued(&mut state)?,
        "a refused signal consumes nothing from the ledger"
    );

    grants.grant("orders", HotAction::Subscribe)?;
    assert!(matches!(
        accept(&runtime, &mut state),
        FrameAction::NoResponse
    ));
    assert!(
        !ledger_holds_issued(&mut state)?,
        "a granted signal is matched against the ledger and consumes its entry"
    );
    Ok(())
}

fn reply(
    runtime: &ConnectionRuntime,
    state: &mut ConnectionProcessState,
    correlation_id: u64,
) -> FrameAction {
    apply_frame(
        PID,
        runtime,
        state,
        Frame::PushReply {
            flags: 0,
            stream_id: STREAM,
            correlation_id,
            payload: b"answer".to_vec(),
        },
    )
}

#[test]
fn a_push_reply_is_decided_on_the_subject_its_push_was_made_on() -> Result<(), Failure> {
    let grants = Arc::new(Grants::default());
    let door = Arc::new(Door::default());
    let (runtime, mut state) = admitted(&grants, &door)?;

    // The subject is recorded at push time; the reply is refused on it and the
    // pusher reads the refusal by type, with the subject and action named.
    let awaiter = runtime.push_for_tests(PID, Some("orders"))?;
    assert!(matches!(
        reply(&runtime, &mut state, awaiter.correlation_id()),
        FrameAction::NoResponse
    ));
    match awaiter.receive(Duration::from_millis(50)) {
        Err(ServerError::PushReplyRefused {
            correlation_id,
            reason,
        }) => {
            assert_eq!(correlation_id, awaiter.correlation_id());
            assert!(
                reason.contains(r#""action":"serve""#) && reason.contains(r#""id":"orders""#),
                "the refusal names the pushed subject and the action it lacked: {reason}"
            );
        }
        other => return Err(format!("the refused reply was not read by type: {other:?}").into()),
    }

    // A push made with no subject cannot be decided: refused by name, so the
    // pusher learns the binding was never recorded rather than waiting.
    let unbound = runtime.push_for_tests(PID, None)?;
    assert!(matches!(
        reply(&runtime, &mut state, unbound.correlation_id()),
        FrameAction::NoResponse
    ));
    match unbound.receive(Duration::from_millis(50)) {
        Err(ServerError::PushReplyRefused { reason, .. }) => {
            assert!(reason.contains("no subject binding"), "{reason}");
        }
        other => return Err(format!("the unbound reply was not refused: {other:?}").into()),
    }

    grants.grant("orders", HotAction::Serve)?;
    let granted = runtime.push_for_tests(PID, Some("orders"))?;
    assert!(matches!(
        reply(&runtime, &mut state, granted.correlation_id()),
        FrameAction::NoResponse
    ));
    assert_eq!(
        granted.receive(Duration::from_millis(50))?,
        b"answer".to_vec(),
        "a granted reply is delivered byte-exact"
    );
    Ok(())
}

const MAPPED: u64 = 70;
const UNMAPPED: u64 = 71;

/// A participant service that places conversation `MAPPED` under `orders` and
/// nothing else; its dispatch is never reached by a decided-and-refused
/// request, so `handle` is an error.
#[derive(Debug)]
struct Mapped;

impl ParticipantSemanticHandler for Mapped {
    fn handle(
        &self,
        _context: ParticipantConnectionContext,
        _conversations: &mut ParticipantConnectionConversations,
        _request: ClientRequest,
    ) -> Result<ServerValue, ParticipantSemanticError> {
        Err(ParticipantSemanticError::Internal {
            message: "the participant handler is not a route under test".to_owned(),
        })
    }

    fn conversation_subject(&self, conversation_id: u64) -> Option<String> {
        (conversation_id == MAPPED).then(|| "orders".to_owned())
    }
}

fn admission(conversation_id: u64) -> ClientRequest {
    ClientRequest::RecordAdmission(RecordAdmission {
        conversation_id,
        participant_id: 2,
        capability_generation: Generation::ONE,
        record_admission_attempt_token: RecordAdmissionAttemptToken::new([0xA7; 16]),
        payload: vec![1, 2, 3],
    })
}

#[test]
fn a_participant_request_is_decided_on_its_conversations_subject() -> Result<(), Failure> {
    let grants = Arc::new(Grants::default());
    let door = Arc::new(Door::default());
    let (runtime, state) = admitted(&grants, &door)?;

    // A conversation no authority places under a subject is refused by name,
    // never decided on a subject guessed from its number.
    let unmapped = participant_access_refusal(&runtime, &state, &Mapped, &admission(UNMAPPED))
        .ok_or("an unmapped conversation was not refused")?;
    assert!(
        unmapped.contains("no subject mapping for conversation 71"),
        "{unmapped}"
    );

    // A mapped conversation is decided with the route's action on its subject.
    let refusal = participant_access_refusal(&runtime, &state, &Mapped, &admission(MAPPED))
        .ok_or("a record admission without publish on orders was not refused")?;
    assert!(
        refusal.contains(r#""action":"publish""#) && refusal.contains(r#""id":"orders""#),
        "the refusal names the conversation's subject and the route's action: {refusal}"
    );
    grants.grant("orders", HotAction::Publish)?;
    assert_eq!(
        participant_access_refusal(&runtime, &state, &Mapped, &admission(MAPPED)),
        None,
        "a granted admission proceeds to dispatch"
    );

    // A recovery batch is decided on every conversation it names before any
    // effect: one unmapped conversation refuses the whole batch.
    let batch = ClientRequest::ObserverRecovery(ObserverRecoveryHandshake {
        observer_refusals: vec![
            ObserverRefusal {
                conversation_id: MAPPED,
                refused_epoch: 1,
            },
            ObserverRefusal {
                conversation_id: UNMAPPED,
                refused_epoch: 1,
            },
        ],
    });
    grants.grant("orders", HotAction::Subscribe)?;
    let batch_refusal = participant_access_refusal(&runtime, &state, &Mapped, &batch)
        .ok_or("a batch naming an unmapped conversation was not refused")?;
    assert!(batch_refusal.contains("conversation 71"), "{batch_refusal}");
    Ok(())
}

/// ACCESS-004 R1 and R2 on a Lys pass: the admitted holder is the
/// connection's verified holder, so a worker registers under that name and
/// is refused by name under any other, with nothing stored.
#[test]
fn a_lys_holder_registers_under_its_own_name_and_no_other() -> Result<(), Failure> {
    let grants = Arc::new(Grants::default());
    let door = Arc::new(Door::default());
    let (runtime, mut state) = admitted(&grants, &door)?;
    let holder = state
        .verified_holder
        .as_ref()
        .ok_or("a Lys Connect left no verified holder")?;
    assert_eq!(holder.participant(), b"fixture-holder");
    let register = |state: &mut ConnectionProcessState, identity: &str| {
        apply_frame(
            PID,
            &runtime,
            state,
            Frame::WorkerRegister {
                flags: 0,
                registration: WorkerRegistration {
                    namespaces: Vec::new(),
                    task_queue: "findings".to_owned(),
                    node: None,
                    activity_types: Vec::new(),
                    identity: identity.to_owned(),
                    activities: Vec::new(),
                },
            },
        )
    };

    let FrameAction::Respond(Frame::WorkerRegisterAck {
        outcome: WorkerRegisterOutcome::Rejected { reason },
        ..
    }) = register(&mut state, "archie")
    else {
        return Err("a Lys holder registered under another name".into());
    };
    assert_eq!(reason, "registration identity is not the verified holder");
    assert!(!reason.contains("fixture-holder"), "{reason}");
    assert_eq!(runtime.stored_registration_for_tests(PID)?, None);

    grants.grant("findings", HotAction::Serve)?;
    assert!(matches!(
        register(&mut state, "fixture-holder"),
        FrameAction::Respond(Frame::WorkerRegisterAck {
            outcome: WorkerRegisterOutcome::Accepted,
            ..
        })
    ));
    Ok(())
}

// ---- ACCESS-003 R1: the service request route ----

#[path = "../worker_admission_tests.rs"]
mod worker_admission_tests;