cortexkit-bus-inmemory 0.3.2

use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use async_trait::async_trait;
use cortexkit_bus_trait::{
    map_connection_failure, map_credential_sign_failure, terminally_dispose, BackendKind, BusError,
    ClaimOutcome, ConformanceBackend, ConnectionFailure, ContentDigest, CredentialSignFailure,
    DurableOwner, Headers, PropertyOutcome, PropertySpec, Register, RegisterUpdate, RegisterWatch,
    Stream, StreamCursor, WatchEvent, WorkQueue,
};

use crate::{BackendEvent, InMemoryBus, InMemoryConfig, WorkQueueConfig};

pub struct InMemoryConformance;

#[async_trait]
impl ConformanceBackend for InMemoryConformance {
    fn backend_kind(&self) -> BackendKind {
        BackendKind::InMemory
    }

    async fn probe(&self, property: PropertySpec) -> PropertyOutcome {
        let result = match property.name {
            "publish_ack_monotonic_sequence" => publish_ack_sequence().await,
            "consumer_replay_within_process_lifetime" => replay_from_cursor().await,
            "stream_delivery_count_rises_on_each_nak" => stream_delivery_count().await,
            "stream_term_is_never_redelivered" => stream_term().await,
            "stream_in_progress_extends_ack_wait" => stream_in_progress().await,
            "register_revision" => register_revision().await,
            "register_absent_vs_empty" => register_absent_vs_empty().await,
            "watch_snapshot_then_updates_with_connection_flag" => register_watch().await,
            "work_redelivery_count" => work_redelivery().await,
            "work_cap_exhaustion_and_dead_letter" => cap_exhaustion().await,
            "full_work_queue_names_exhausted_limit" => full_queue().await,
            "error_taxonomy_and_boundary_mapping" => error_taxonomy().await,
            other => Err(format!("in-memory has no parity probe named {other}")),
        };
        match result {
            Ok(()) => PropertyOutcome::Passed,
            Err(reason) => PropertyOutcome::Failed { reason },
        }
    }
}

async fn publish_ack_sequence() -> Result<(), String> {
    let bus = InMemoryBus::new(InMemoryConfig::default());
    let first = bus
        .publish("ck.box.room.a.post", "one", digest("one"), Headers::new())
        .await
        .map_err(debug)?;
    let second = bus
        .publish("ck.box.room.a.post", "two", digest("two"), Headers::new())
        .await
        .map_err(debug)?;
    require(
        !first.duplicate()
            && !second.duplicate()
            && second.stream_sequence() > first.stream_sequence(),
        "publish sequence did not increase",
    )
}

async fn replay_from_cursor() -> Result<(), String> {
    let bus =
        InMemoryBus::new(InMemoryConfig::default().with_durable(DurableOwner::Agent("agent")));
    let receipt = bus
        .publish(
            "ck.box.peer.agent.s.deliver",
            "m1",
            digest("m1"),
            Headers::new(),
        )
        .await
        .map_err(debug)?;
    require(!receipt.duplicate(), "unexpected duplicate publish")?;
    let expected = bus
        .consumer(DurableOwner::Agent("agent"))
        .await
        .map_err(debug)?
        .next()
        .await
        .map_err(debug)?
        .ok_or("first cursor saw no message")?;
    let mut reopened = bus
        .consumer(DurableOwner::Agent("agent"))
        .await
        .map_err(debug)?;
    let replayed = reopened
        .next()
        .await
        .map_err(debug)?
        .ok_or("reopened cursor saw no replay")?;
    require(
        expected.stream_seq == replayed.stream_seq,
        "cursor did not replay",
    )?;
    reopened.ack().await.map_err(debug)
}

const STREAM_SUBJECT: &str = "ck.box.peer.agent.s.deliver";

async fn stream_delivery_count() -> Result<(), String> {
    let bus =
        InMemoryBus::new(InMemoryConfig::default().with_durable(DurableOwner::Agent("agent")));
    let receipt = bus
        .publish(STREAM_SUBJECT, "m1", digest("m1"), Headers::new())
        .await
        .map_err(debug)?;
    require(!receipt.duplicate(), "unexpected duplicate publish")?;
    let mut cursor = bus
        .consumer(DurableOwner::Agent("agent"))
        .await
        .map_err(debug)?;
    for expected in 1..=3 {
        let delivery = cursor
            .next()
            .await
            .map_err(debug)?
            .ok_or("a nak'd message was not redelivered")?;
        require(
            delivery.delivery_count == expected,
            &format!(
                "delivery {expected} reported count {}",
                delivery.delivery_count
            ),
        )?;
        if expected < 3 {
            cursor.nak(Duration::ZERO).await.map_err(debug)?;
        }
    }
    cursor.ack().await.map_err(debug)
}

async fn stream_term() -> Result<(), String> {
    let bus =
        InMemoryBus::new(InMemoryConfig::default().with_durable(DurableOwner::Agent("agent")));
    for id in ["m1", "m2"] {
        let receipt = bus
            .publish(STREAM_SUBJECT, id, digest(id), Headers::new())
            .await
            .map_err(debug)?;
        require(!receipt.duplicate(), "unexpected duplicate publish")?;
    }
    let mut cursor = bus
        .consumer(DurableOwner::Agent("agent"))
        .await
        .map_err(debug)?;
    let first = cursor
        .next()
        .await
        .map_err(debug)?
        .ok_or("first message was not delivered")?;
    require(first.message.id == "m1", "first delivery was not m1")?;
    cursor.term(Some("body_absent")).await.map_err(debug)?;
    let second = cursor
        .next()
        .await
        .map_err(debug)?
        .ok_or("the message after a term was not delivered")?;
    require(
        second.message.id == "m2" && second.delivery_count == 1,
        "a terminated message was redelivered",
    )?;
    cursor.ack().await.map_err(debug)?;
    let mut reopened = bus
        .consumer(DurableOwner::Agent("agent"))
        .await
        .map_err(debug)?;
    require(
        reopened.next().await.map_err(debug)?.is_none(),
        "a terminated message was redelivered to a reopened cursor",
    )?;
    require(
        bus.events().iter().any(|event| {
            matches!(
                event,
                BackendEvent::StreamTerminated { stream_seq, reason: Some(reason) }
                    if *stream_seq == first.stream_seq && reason == "body_absent"
            )
        }),
        "the termination and its reason were not recorded",
    )
}

async fn stream_in_progress() -> Result<(), String> {
    let ack_wait = Duration::from_millis(200);
    let clock = Arc::new(Mutex::new(Instant::now()));
    let mut bus = InMemoryBus::new(
        InMemoryConfig::default()
            .with_durable(DurableOwner::Agent("agent"))
            .with_stream_ack_wait(ack_wait),
    );
    // Only this bus's delivery clock advances, so scheduler delays cannot
    // carry the probe past the extended lease or affect other backends.
    bus.stream_clock = {
        let clock = Arc::clone(&clock);
        Arc::new(move || *clock.lock().expect("stream probe clock poisoned"))
    };
    for id in ["m1", "m2"] {
        let receipt = bus
            .publish(STREAM_SUBJECT, id, digest(id), Headers::new())
            .await
            .map_err(debug)?;
        require(!receipt.duplicate(), "unexpected duplicate publish")?;
    }
    let mut slow = bus
        .consumer(DurableOwner::Agent("agent"))
        .await
        .map_err(debug)?;
    let mut other = bus
        .consumer(DurableOwner::Agent("agent"))
        .await
        .map_err(debug)?;

    // Control: without a progress report the message comes back to another
    // cursor once the ack wait runs out.
    slow.next()
        .await
        .map_err(debug)?
        .ok_or("m1 was not delivered")?;
    *clock.lock().expect("stream probe clock poisoned") += ack_wait * 3 / 2;
    let redelivered = other
        .next()
        .await
        .map_err(debug)?
        .ok_or("m1 was not redelivered after the ack wait")?;
    require(
        redelivered.message.id == "m1" && redelivered.delivery_count == 2,
        "the ack-wait redelivery was not m1's second delivery",
    )?;
    other.ack().await.map_err(debug)?;
    let _ = slow.ack().await;

    // A progress report inside the window restarts the ack wait, so the
    // message is not redelivered past the original deadline.
    slow.next()
        .await
        .map_err(debug)?
        .ok_or("m2 was not delivered")?;
    *clock.lock().expect("stream probe clock poisoned") += ack_wait * 3 / 5;
    slow.in_progress().await.map_err(debug)?;
    *clock.lock().expect("stream probe clock poisoned") += ack_wait * 3 / 5;
    require(
        other.next().await.map_err(debug)?.is_none(),
        "a message reported in progress was redelivered past the original ack wait",
    )?;
    slow.ack().await.map_err(debug)?;
    require(
        other.next().await.map_err(debug)?.is_none(),
        "an acknowledged message was redelivered",
    )
}

async fn register_revision() -> Result<(), String> {
    let bus = InMemoryBus::new(InMemoryConfig::default());
    let first = bus.put("module", b"one".to_vec()).await.map_err(debug)?;
    let second = bus.put("module", b"two".to_vec()).await.map_err(debug)?;
    let read = bus.get("module").await.map_err(debug)?;
    require(second > first, "register revision did not increase")?;
    require(read.revision == second, "get returned the wrong revision")
}

async fn register_absent_vs_empty() -> Result<(), String> {
    let bus = InMemoryBus::new(InMemoryConfig::default());
    bus.put("empty", Vec::new()).await.map_err(debug)?;
    let empty = bus.get("empty").await.map_err(debug)?;
    require(empty.value.is_empty(), "empty value was not preserved")?;
    require(
        matches!(bus.get("missing").await, Err(BusError::Absent { .. })),
        "missing key was not Absent",
    )
}

async fn register_watch() -> Result<(), String> {
    let bus = InMemoryBus::new(InMemoryConfig::default());
    bus.put("module.a", b"v1".to_vec()).await.map_err(debug)?;
    let mut watch = bus.watch("module.").await.map_err(debug)?;
    require(
        !watch.has_snapshot(),
        "watch claimed a snapshot before delivery",
    )?;
    let snapshot = watch.next().await.map_err(debug)?;
    require(
        matches!(snapshot, WatchEvent::Snapshot { ref entries, .. } if entries.len() == 1),
        "watch did not deliver the complete initial snapshot",
    )?;
    require(
        watch.has_snapshot(),
        "watch did not record snapshot receipt",
    )?;
    bus.put("module.a", b"v2".to_vec()).await.map_err(debug)?;
    require(
        matches!(
            watch.next().await.map_err(debug)?,
            WatchEvent::Update(RegisterUpdate::Put(_))
        ),
        "watch did not distinguish an update",
    )
}

async fn work_redelivery() -> Result<(), String> {
    let bus = queue_bus(3, 4_096);
    let receipt = bus
        .publish(
            "ck.box.effect.a.s.intent",
            "m1",
            digest("m1"),
            Headers::new(),
        )
        .await
        .map_err(debug)?;
    require(!receipt.duplicate(), "unexpected duplicate publish")?;
    let first = item(bus.claim().await.map_err(debug)?)?;
    bus.nak(first.token, Duration::ZERO).await.map_err(debug)?;
    let second = item(bus.claim().await.map_err(debug)?)?;
    require(
        first.delivery_count == 1,
        "first delivery count was not one",
    )?;
    require(second.delivery_count == 2, "redelivery count was not two")
}

async fn cap_exhaustion() -> Result<(), String> {
    let bus = queue_bus(1, 4_096);
    let receipt = bus
        .publish(
            "ck.box.effect.a.s.intent",
            "m1",
            digest("m1"),
            Headers::new(),
        )
        .await
        .map_err(debug)?;
    require(!receipt.duplicate(), "unexpected duplicate publish")?;
    let first = item(bus.claim().await.map_err(debug)?)?;
    bus.nak(first.token, Duration::ZERO).await.map_err(debug)?;
    let exhausted = match bus.claim().await.map_err(debug)? {
        ClaimOutcome::MaxDeliveriesExceeded(exhausted) => exhausted,
        other => {
            return Err(format!(
                "cap did not report MaxDeliveriesExceeded: {other:?}"
            ))
        }
    };
    let dead_stream = InMemoryBus::new(InMemoryConfig::default());
    let queue = DeadLetterObservedQueue {
        queue: &bus,
        dead_stream: &dead_stream,
    };
    let receipt = terminally_dispose(
        &dead_stream,
        &queue,
        "box",
        &exhausted,
        digest("dead-record"),
    )
    .await
    .map_err(debug)?;
    require(!receipt.duplicate(), "dead letter was not stored")?;
    require(
        bus.events()
            .iter()
            .any(|event| matches!(event, BackendEvent::Terminated { .. })),
        "exhausted delivery was not terminated",
    )
}

/// Observes the separate dead-letter stream at termination time, so the probe
/// still detects a termination that incorrectly precedes the publish.
struct DeadLetterObservedQueue<'a> {
    queue: &'a InMemoryBus,
    dead_stream: &'a InMemoryBus,
}

#[async_trait]
impl WorkQueue for DeadLetterObservedQueue<'_> {
    async fn claim(&self) -> cortexkit_bus_trait::BusResult<ClaimOutcome> {
        self.queue.claim().await
    }

    async fn ack(
        &self,
        token: cortexkit_bus_trait::DeliveryToken,
    ) -> cortexkit_bus_trait::BusResult<()> {
        self.queue.ack(token).await
    }

    async fn nak(
        &self,
        token: cortexkit_bus_trait::DeliveryToken,
        delay: Duration,
    ) -> cortexkit_bus_trait::BusResult<()> {
        self.queue.nak(token, delay).await
    }

    async fn term(
        &self,
        token: cortexkit_bus_trait::DeliveryToken,
    ) -> cortexkit_bus_trait::BusResult<()> {
        if !self.dead_stream.events().iter().any(|event| {
            matches!(event, BackendEvent::Published { subject, .. } if subject == "ck.box.effect.dead")
        }) {
            return Err(BusError::denied("ck.box.effect.dead", "dead-letter publish did not precede term"));
        }
        self.queue.term(token).await
    }
}

async fn full_queue() -> Result<(), String> {
    let bus = queue_bus(5, 150);
    let receipt = bus
        .publish(
            "ck.box.effect.a.s.intent",
            "first",
            digest("first"),
            Headers::new(),
        )
        .await
        .map_err(debug)?;
    require(!receipt.duplicate(), "unexpected duplicate publish")?;
    let error = bus
        .publish(
            "ck.box.effect.a.s.intent",
            "second",
            digest("second"),
            Headers::new(),
        )
        .await
        .expect_err("second publish should exceed max-bytes");
    require(
        matches!(error, BusError::Unavailable { exhausted_limit: Some(ref limit), .. } if limit == "max_bytes"),
        "full queue did not name max_bytes in Unavailable",
    )
}

async fn error_taxonomy() -> Result<(), String> {
    let denied_subject = "ck.box.room.denied.post";
    let bus = InMemoryBus::new(
        InMemoryConfig::default()
            .deny_subject(denied_subject)
            .with_work_queue(WorkQueueConfig {
                subject: "ck.box.effect.a.s.intent".into(),
                max_bytes: 4_096,
                max_deliveries: 5,
                max_nak_delay: Duration::from_millis(1),
            }),
    );
    require(
        matches!(bus.publish(denied_subject, "m", digest("m"), Headers::new()).await, Err(BusError::Denied { subject, .. }) if subject == denied_subject),
        "fixture denial was not Denied",
    )?;
    require(
        matches!(bus.get("missing").await, Err(BusError::Absent { .. })),
        "missing register key was not Absent",
    )?;
    let receipt = bus
        .publish(
            "ck.box.effect.a.s.intent",
            "work",
            digest("work"),
            Headers::new(),
        )
        .await
        .map_err(debug)?;
    require(!receipt.duplicate(), "unexpected duplicate publish")?;
    let claimed = item(bus.claim().await.map_err(debug)?)?;
    require(
        matches!(
            bus.nak(claimed.token, Duration::from_secs(1)).await,
            Err(BusError::Clamped {
                retry_after: Some(_),
                ..
            })
        ),
        "oversized delay was not reported Clamped",
    )?;
    require(
        matches!(
            map_connection_failure(ConnectionFailure::WriteOnSeveredSocket),
            BusError::Unavailable { .. }
        ),
        "severed write was not Unavailable",
    )?;
    require(
        matches!(
            map_connection_failure(ConnectionFailure::ServerClosedThenReauthenticationRefused {
                subject: denied_subject.into(),
                reason: "revoked".into()
            }),
            BusError::Denied { .. }
        ),
        "reconnect refusal was not Denied",
    )?;
    require(
        matches!(
            map_credential_sign_failure(CredentialSignFailure::RecordMissing {
                record: "nats.module.g1.e0".into()
            }),
            BusError::Absent { .. }
        ),
        "missing signing record was not Absent",
    )
}

fn queue_bus(max_deliveries: u32, max_bytes: usize) -> InMemoryBus {
    InMemoryBus::new(InMemoryConfig::default().with_work_queue(WorkQueueConfig {
        subject: "ck.box.effect.a.s.intent".into(),
        max_bytes,
        max_deliveries,
        max_nak_delay: Duration::from_secs(30),
    }))
}

fn item(outcome: ClaimOutcome) -> Result<cortexkit_bus_trait::WorkItem, String> {
    match outcome {
        ClaimOutcome::Item(item) => Ok(item),
        other => Err(format!("expected work item, got {other:?}")),
    }
}

fn digest(value: &str) -> ContentDigest {
    ContentDigest::of_bytes(value.as_bytes())
}

fn require(condition: bool, reason: &str) -> Result<(), String> {
    condition.then_some(()).ok_or_else(|| reason.into())
}

fn debug(error: impl std::fmt::Debug) -> String {
    format!("{error:?}")
}

#[cfg(test)]
mod tests {
    use super::*;
    use cortexkit_bus_trait::{
        classify_properties, run_conformance, PropertyApplicability, PropertySet, SuiteBug,
        CONFORMANCE_PROPERTIES,
    };

    struct WeakenedNatsFixture;

    #[async_trait]
    impl ConformanceBackend for WeakenedNatsFixture {
        fn backend_kind(&self) -> BackendKind {
            BackendKind::Nats
        }

        async fn probe(&self, property: PropertySpec) -> PropertyOutcome {
            match property.set {
                PropertySet::Parity => InMemoryConformance.probe(property).await,
                PropertySet::NatsOnly => PropertyOutcome::Failed {
                    reason: "fixture deliberately has only process-local semantics".into(),
                },
            }
        }
    }

    #[tokio::test]
    async fn in_memory_passes_every_parity_property() {
        let report = run_conformance(&InMemoryConformance, PropertySet::Parity)
            .await
            .expect("parity is applicable to in-memory");

        assert_eq!(report.property_set_name, "parity");
        for (property, outcome) in &report.outcomes {
            assert_eq!(outcome, &PropertyOutcome::Passed, "{}", property.name);
        }
        assert!(report.passed());
    }

    #[tokio::test]
    async fn asserting_nats_only_against_in_memory_is_a_suite_bug() {
        let result = run_conformance(&InMemoryConformance, PropertySet::NatsOnly).await;
        assert_eq!(result, Err(SuiteBug::NatsOnlyAssertedAgainstInMemory));
    }

    #[tokio::test(start_paused = true)]
    async fn in_progress_probe_does_not_wait_for_wall_clock() {
        stream_in_progress().await.unwrap();
    }

    #[tokio::test]
    async fn weakened_backend_passes_real_parity_probes_but_fails_nats_only() {
        let parity = run_conformance(&WeakenedNatsFixture, PropertySet::Parity)
            .await
            .expect("parity run is valid");
        let nats_only = run_conformance(&WeakenedNatsFixture, PropertySet::NatsOnly)
            .await
            .expect("NATS-only run is valid for a NATS-labelled fixture");

        assert!(parity.passed());
        assert!(!nats_only.passed());
    }

    #[test]
    fn every_nats_only_property_is_not_applicable_not_passed() {
        let classifications = classify_properties(BackendKind::InMemory);
        let nats_only = classifications
            .iter()
            .filter(|row| row.property.set == PropertySet::NatsOnly)
            .collect::<Vec<_>>();

        assert_eq!(
            nats_only.len(),
            CONFORMANCE_PROPERTIES
                .iter()
                .filter(|property| property.set == PropertySet::NatsOnly)
                .count()
        );
        assert!(nats_only.iter().all(|row| matches!(
            row.applicability,
            PropertyApplicability::NotApplicable { .. }
        )));
    }
}