cratefield-module-webhooks 0.5.0

//! The delivery contract, end to end against a real SQLite database: the
//! event-type filter and subject isolation, the 2xx path, the 5xx
//! retry-then-dead-letter path, replay (including its refusal when the
//! endpoint is gone), the event types no header could carry, and the
//! signature a receiver verifies.
//!
//! The ports are the same ones a deployment gets: a `Clock` the test can
//! move forward (so a retried row actually becomes due again — a frozen
//! clock would claim each row exactly once and prove nothing about the
//! second attempt), and an `HttpClient` that records the full request —
//! headers included, because the signature lives in one — and answers from
//! a script.

#![allow(clippy::missing_panics_doc)]
// Interior mutability here records test observations — a clock a test can
// move, a scripted provider's queue, the requests a delivery made. It is
// not request state (ADR 0007); the scoped allow follows the policy in the
// workspace `clippy.toml`, as `cratefield-testing`'s own fakes do.
#![allow(clippy::disallowed_types)]

use std::fmt::Write as _;
use std::sync::{Arc, Mutex};

use async_trait::async_trait;
use bytes::Bytes;
use cratefield_core::{
    Clock, Config, Database, HttpClient, HttpError, MapConfig, Module, ModuleContext,
    PersonalDataCatalog, Port, Ports, ScheduledBudget, TemplateRegistry, Venture,
};
use cratefield_module_webhooks::{
    Delivery, DrainReport, EndpointError, PublishError, Published, Webhooks,
};
use cratefield_testing::TestHarness;
use hmac::{Hmac, KeyInit, Mac};
use http::{Request, Response};
use serde_json::{Value, json};
use sha2::Sha256;
use time::OffsetDateTime;

const NOW_UNIX: i64 = 1_800_000_000;
/// The same instant as [`NOW_UNIX`], in the stored spelling: every stored
/// timestamp should read as a time, never as an epoch, and the two must
/// agree or a queued row sits in the future of the clock reading it.
const NOW_ISO: &str = "2027-01-15T08:00:00Z";

const ALICE: &str = "acct-alice";
const BOB: &str = "acct-bob";

// ---------------------------------------------------------------------------
// The kit

struct Kit {
    harness: TestHarness,
    module: Webhooks,
    clock: TestClock,
    http: CaptureClient,
}

impl Kit {
    /// A harness with the webhooks module mounted, its `Clock` port the
    /// movable [`TestClock`] and its `HttpClient` port the recording
    /// [`CaptureClient`], which answers from `script` (statuses, in
    /// order) and then always 200.
    fn mount(max_attempts: u32, script: &[u16]) -> Self {
        Self::mount_module(Webhooks::new().max_attempts(max_attempts), script)
    }

    /// [`Kit::mount`] for a module built some other way, a configured one.
    fn mount_module(module: Webhooks, script: &[u16]) -> Self {
        let clock = TestClock::at(NOW_UNIX);
        let http = CaptureClient::scripted(script);
        let harness = TestHarness::with_ports(vec![Box::new(module.clone())], |ports| {
            ports.clock = Some(Arc::new(clock.clone()));
            ports.http = Some(Arc::new(http.clone()));
        });
        Self {
            harness,
            module,
            clock,
            http,
        }
    }

    fn db(&self) -> Arc<dyn Database> {
        Arc::clone(&self.harness.db)
    }

    /// Every port a scheduled invocation would carry, wired to this kit —
    /// before a test pokes a hole in one to prove the drain refuses.
    fn wired_ports(&self, config: &Arc<dyn Config>) -> Ports {
        let mut ports = Ports::with_config(Arc::clone(config));
        ports.db = Some(Arc::clone(&self.harness.db));
        ports.clock = Some(Arc::new(self.clock.clone()));
        ports.http = Some(Arc::new(self.http.clone()));
        ports
    }

    /// The context a scheduled invocation would carry: this kit's
    /// database, movable clock and recording client.
    fn context(&self, config: Arc<dyn Config>) -> ModuleContext {
        let ports = self.wired_ports(&config);
        self.context_over(ports, config)
    }

    /// A context over given ports — the escape hatch the port-hole tests
    /// use to hand the drain a context with one port gone.
    fn context_over(&self, ports: Ports, config: Arc<dyn Config>) -> ModuleContext {
        ModuleContext {
            ports,
            config,
            events: self.harness.harness.events().clone(),
            templates: Arc::new(TemplateRegistry::default()),
            venture: Arc::new(Venture::new("test-venture", "test.example")),
            unprotected_writes_accepted: false,
            ui_mounted: false,
            personal_data: Arc::new(PersonalDataCatalog::default()),
            scheduled: Arc::new(ScheduledBudget::unbounded()),
        }
    }

    /// The context a scheduled invocation would carry with `budget` as its
    /// `ctx.scheduled` — the same wiring as [`Kit::context`], with the
    /// budget the runtime would hand this module for the tick.
    fn context_with_budget(&self, budget: Arc<ScheduledBudget>) -> ModuleContext {
        let config = Kit::config();
        let mut ctx = self.context_over(self.wired_ports(&config), config);
        ctx.scheduled = budget;
        ctx
    }

    fn config() -> Arc<dyn Config> {
        Arc::new(MapConfig::default())
    }

    async fn create_endpoint(&self, subject: &str, url: &str, filters: &[&str]) -> String {
        self.module
            .create_endpoint(&*self.db(), subject, url, filters, NOW_ISO)
            .await
            .expect("the endpoint registers")
            .endpoint_id
    }

    /// `publish` + commit, the way a caller with its own transaction does
    /// it: the statements go into the caller's batch, never their own.
    async fn publish(&self, subject: &str, event_type: &str, data: Value) -> Published {
        let published = self
            .module
            .publish(&*self.db(), subject, event_type, &data, NOW_ISO)
            .await
            .expect("the event fans out");
        self.harness
            .db
            .batch_atomic(&published.clone().into_statements())
            .await
            .expect("the batch commits");
        published
    }

    async fn drain(&self) -> DrainReport {
        self.drain_under(Kit::config()).await
    }

    /// A pass whose `ctx.config` is `config`: the alias comes from here.
    async fn drain_under(&self, config: Arc<dyn Config>) -> DrainReport {
        self.module
            .drain_with(&self.context(config))
            .await
            .expect("the drain runs")
    }

    /// A pass under an explicit budget, the way a scheduled invocation
    /// runs: `budget` is the module's `ctx.scheduled`.
    async fn drain_within(&self, budget: &Arc<ScheduledBudget>) -> DrainReport {
        self.module
            .drain_with(&self.context_with_budget(Arc::clone(budget)))
            .await
            .expect("the drain runs")
    }

    async fn deliveries(&self, subject: &str) -> Vec<Delivery> {
        self.module
            .deliveries(&*self.db(), subject, 100)
            .await
            .expect("the log reads")
    }
}

// ---------------------------------------------------------------------------
// A clock the test moves

#[derive(Clone)]
struct TestClock(Arc<Mutex<OffsetDateTime>>);

impl TestClock {
    fn at(unix: i64) -> Self {
        Self(Arc::new(Mutex::new(
            OffsetDateTime::from_unix_timestamp(unix).expect("in range"),
        )))
    }

    fn advance(&self, seconds: i64) {
        let mut now = self.0.lock().expect("clock");
        *now = now.saturating_add(time::Duration::seconds(seconds));
    }
}

impl Clock for TestClock {
    fn now(&self) -> OffsetDateTime {
        *self.0.lock().expect("clock")
    }
}

// ---------------------------------------------------------------------------
// An HttpClient that records the whole request and answers from a script

#[derive(Clone, Default)]
struct CaptureClient {
    inner: Arc<Mutex<CaptureInner>>,
}

#[derive(Default)]
struct CaptureInner {
    script: Vec<u16>,
    answered: usize,
    captured: Vec<Captured>,
}

#[derive(Clone)]
struct Captured {
    uri: String,
    headers: Vec<(String, String)>,
    body: Vec<u8>,
}

impl CaptureClient {
    fn scripted(statuses: &[u16]) -> Self {
        Self {
            inner: Arc::new(Mutex::new(CaptureInner {
                script: statuses.to_vec(),
                ..CaptureInner::default()
            })),
        }
    }

    fn captured(&self) -> Vec<Captured> {
        self.inner.lock().expect("http lock").captured.clone()
    }
}

#[async_trait]
impl HttpClient for CaptureClient {
    async fn send(&self, request: Request<Bytes>) -> Result<Response<Bytes>, HttpError> {
        let (parts, body) = request.into_parts();
        let mut inner = self.inner.lock().expect("http lock");
        inner.captured.push(Captured {
            uri: parts.uri.to_string(),
            headers: parts
                .headers
                .iter()
                .map(|(name, value)| {
                    (
                        name.as_str().to_owned(),
                        value.to_str().unwrap_or_default().to_owned(),
                    )
                })
                .collect(),
            body: body.to_vec(),
        });
        // The script, then a 200 forever: the tests that care about
        // failures script every answer they expect to see.
        let status = if inner.answered < inner.script.len() {
            let status = inner.script[inner.answered];
            inner.answered += 1;
            status
        } else {
            200
        };
        Ok(Response::builder()
            .status(status)
            .body(Bytes::new())
            .expect("a bare status is a valid response"))
    }
}

// ---------------------------------------------------------------------------
// Assertion helpers

fn header<'a>(captured: &'a Captured, name: &str) -> &'a str {
    match captured
        .headers
        .iter()
        .find(|(key, _)| key.eq_ignore_ascii_case(name))
    {
        Some((_, value)) => value.as_str(),
        None => panic!("no {name} header on the delivery"),
    }
}

fn envelope(captured: &Captured) -> Value {
    serde_json::from_slice(&captured.body).expect("the body is the JSON envelope")
}

/// One delivery per URL, in the order they were made.
fn captures_to<'a>(captured: &'a [Captured], uri: &str) -> Vec<&'a Captured> {
    captured.iter().filter(|c| c.uri == uri).collect()
}

const URL_A: &str = "https://alice-a.test/hooks";
const URL_B: &str = "https://alice-b.test/hooks";
const URL_C: &str = "https://bob.test/hooks";

// ---------------------------------------------------------------------------

#[pollster::test]
async fn the_filter_decides_who_hears_an_event_and_no_subject_hears_anothers() {
    let kit = Kit::mount(5, &[]);
    let filtered = kit.create_endpoint(ALICE, URL_A, &["order.paid"]).await;
    // A second later, so the two creations carry distinct timestamps and
    // "creation order" is a property of the rows, not of the id tiebreak
    // that only decides rows stored in the same second.
    let catch_all = kit
        .module
        .create_endpoint(&*kit.db(), ALICE, URL_B, &[], "2027-01-15T08:00:01Z")
        .await
        .expect("the endpoint registers")
        .endpoint_id;
    kit.create_endpoint(BOB, URL_C, &[]).await;

    // The list never hands secrets back: the type carries none, and both
    // of Alice's endpoints are hers, in creation order.
    let endpoints = kit
        .module
        .endpoints(&*kit.db(), ALICE)
        .await
        .expect("the list reads");
    assert_eq!(
        endpoints.iter().map(|e| e.id.as_str()).collect::<Vec<_>>(),
        [filtered.as_str(), catch_all.as_str()]
    );

    // A refunded order passes only the catch-all: the filtered endpoint
    // said `order.paid` and this is not one.
    assert_eq!(
        kit.publish(ALICE, "order.refunded", json!({ "n": 1 }))
            .await
            .endpoints,
        1
    );
    // A paid order reaches both of Alice's endpoints — and only hers.
    assert_eq!(
        kit.publish(ALICE, "order.paid", json!({ "n": 2 }))
            .await
            .endpoints,
        2
    );
    // Bob's event fans out to exactly one endpoint: his own.
    assert_eq!(
        kit.publish(BOB, "order.paid", json!({ "n": 3 }))
            .await
            .endpoints,
        1
    );

    let report = kit.drain().await;
    assert_eq!(report.claimed, 4);
    assert_eq!(report.delivered, 4);
    assert_eq!(report.retried, 0);
    assert_eq!(report.dead_lettered, 0);

    // Per endpoint, per subject: the filtered endpoint saw one delivery,
    // Alice's catch-all two, and every envelope's subject is its own
    // endpoint's subject — Bob's event never touched Alice's URL.
    let captured = kit.http.captured();
    assert_eq!(captures_to(&captured, URL_A).len(), 1);
    assert_eq!(captures_to(&captured, URL_B).len(), 2);
    assert_eq!(captures_to(&captured, URL_C).len(), 1);
    for c in &captured {
        let body = envelope(c);
        let expected_subject = if c.uri == URL_C { BOB } else { ALICE };
        assert_eq!(body["subject"], expected_subject, "{}", c.uri);
    }

    // And each subject's log is its own.
    assert_eq!(kit.deliveries(ALICE).await.len(), 3);
    assert_eq!(kit.deliveries(BOB).await.len(), 1);
}

#[pollster::test]
async fn a_2xx_delivery_is_logged_with_its_status_and_completes() {
    let kit = Kit::mount(5, &[]);
    kit.create_endpoint(ALICE, URL_A, &[]).await;
    let published = kit
        .publish(ALICE, "order.paid", json!({ "invoice": "in_1" }))
        .await;

    let report = kit.drain().await;
    assert_eq!(report.claimed, 1);
    assert_eq!(report.delivered, 1);

    let log = kit.deliveries(ALICE).await;
    assert_eq!(log.len(), 1, "one attempt, logged once");
    let entry = &log[0];
    assert_eq!(entry.endpoint_id.len(), 26, "an endpoint was named");
    assert_eq!(entry.event_id, published.event_id);
    assert_eq!(entry.event_type, "order.paid");
    assert_eq!(entry.attempt, 1);
    assert_eq!(entry.status_code, Some(200));
    assert_eq!(entry.error, None);

    // Completed: a second pass claims nothing.
    assert_eq!(kit.drain().await.claimed, 0);
}

#[pollster::test]
async fn a_5xx_retries_with_backoff_then_dead_letters_after_max_attempts() {
    // Three attempts allowed, three 500s scripted: the delivery exhausts
    // exactly its budget.
    let kit = Kit::mount(3, &[500, 500, 500]);
    kit.create_endpoint(ALICE, URL_A, &[]).await;
    let published = kit
        .publish(ALICE, "order.paid", json!({ "invoice": "in_2" }))
        .await;

    // First pass: one attempt, logged with the status, put back with the
    // first backoff.
    let report = kit.drain().await;
    assert_eq!(report.claimed, 1);
    assert_eq!(report.retried, 1);
    assert_eq!(kit.deliveries(ALICE).await.len(), 1);

    // The backoff holds: with the clock unmoved the row is not due, so
    // the second pass claims nothing — a 30s-old failure is not retried
    // 0s later.
    assert_eq!(kit.drain().await.claimed, 0);

    // +30s, the first backoff: attempt two, another 500, another backoff —
    // the doubling means this one is 60s out.
    kit.clock.advance(30);
    assert_eq!(kit.drain().await.retried, 1);
    assert_eq!(kit.deliveries(ALICE).await.len(), 2);

    kit.clock.advance(59);
    assert_eq!(kit.drain().await.claimed, 0, "59s into a 60s backoff");

    // +60s total: the last allowed attempt fails and the row dead-letters.
    kit.clock.advance(1);
    let report = kit.drain().await;
    assert_eq!(report.claimed, 1);
    assert_eq!(report.dead_lettered, 1);

    // The log holds every attempt, in order, all with the status.
    let log = kit.deliveries(ALICE).await;
    assert_eq!(log.len(), 3);
    assert_eq!(
        log.iter().map(|d| d.attempt).collect::<Vec<_>>(),
        [3, 2, 1],
        "newest attempt first"
    );
    assert!(log.iter().all(|d| d.status_code == Some(500)));

    // The dead letter is the event, not a summary of it.
    let letters = kit
        .module
        .dead_letters(&*kit.db(), ALICE, 10)
        .await
        .expect("the dead letters read");
    assert_eq!(letters.len(), 1);
    let letter = &letters[0];
    assert_eq!(letter.event_id, published.event_id);
    assert_eq!(letter.attempts, 3);
    assert_eq!(letter.reason, "attempts_exhausted");
    assert_eq!(letter.status_code, Some(500));
    assert!(
        letter.last_error.contains("500"),
        "the receiver's answer is preserved: {}",
        letter.last_error
    );

    // Given up means out of the queue: nothing left to claim.
    assert_eq!(kit.drain().await.claimed, 0);
}

#[pollster::test]
async fn replay_sends_a_dead_letter_again_and_resets_the_attempts() {
    let kit = Kit::mount(2, &[500, 500]);
    kit.create_endpoint(ALICE, URL_A, &[]).await;
    let published = kit.publish(ALICE, "order.paid", json!({ "n": 7 })).await;
    kit.drain().await;
    kit.clock.advance(30);
    kit.drain().await;
    assert_eq!(
        kit.module
            .dead_letters(&*kit.db(), ALICE, 10)
            .await
            .expect("read")
            .len(),
        1
    );

    // Replay re-enqueues the event — attempts reset to zero — and removes
    // the letter, in one batch. The receiver now answers 200 (the script
    // is exhausted), so the drain delivers. The clock moves first, so the
    // replayed attempt is logged strictly after the two it follows.
    kit.clock.advance(1);
    let letter_id = kit
        .module
        .dead_letters(&*kit.db(), ALICE, 10)
        .await
        .expect("read")[0]
        .id
        .clone();
    assert!(
        kit.module
            .replay(&*kit.db(), ALICE, &letter_id, NOW_ISO)
            .await
            .expect("the replay commits")
    );
    assert!(
        !kit.module
            .replay(&*kit.db(), ALICE, &letter_id, NOW_ISO)
            .await
            .expect("the second replay is an ordinary no"),
        "a replayed letter is gone; replaying it again is false, not an error"
    );
    assert_eq!(
        kit.module
            .dead_letters(&*kit.db(), ALICE, 10)
            .await
            .expect("read")
            .len(),
        0
    );

    let report = kit.drain().await;
    assert_eq!(report.claimed, 1);
    assert_eq!(report.delivered, 1);

    // The log keeps the whole history: the two failed attempts, then the
    // replayed delivery as attempt one of its (new) life.
    let log = kit.deliveries(ALICE).await;
    assert_eq!(log.len(), 3);
    assert_eq!(log[0].attempt, 1, "the replayed delivery restarts at one");
    assert_eq!(log[0].status_code, Some(200));

    // And it is the same event, not a copy with a new identity.
    let captured = kit.http.captured();
    let last = captured.last().expect("a delivery was made");
    let body = envelope(last);
    assert_eq!(body["id"], published.event_id.as_str());
    assert_eq!(body["data"]["n"], 7);

    // A replay names a letter of *this subject*: another subject's id (or
    // any invented one) is an ordinary `false`.
    assert!(
        !kit.module
            .replay(&*kit.db(), BOB, &letter_id, NOW_ISO)
            .await
            .expect("read")
    );
}

/// A replay whose endpoint is gone must refuse and leave the letter in
/// place: consuming it would hand the event to a drain that drops
/// undeliverable rows, destroying the event for good.
#[pollster::test]
async fn a_replay_into_a_deleted_endpoint_refuses_and_keeps_the_letter() {
    let kit = Kit::mount(1, &[500]);
    let endpoint_id = kit.create_endpoint(ALICE, URL_A, &[]).await;
    kit.publish(ALICE, "order.paid", json!({ "n": 1 })).await;
    // One attempt allowed, one 500 scripted: the row dead-letters.
    assert_eq!(kit.drain().await.dead_lettered, 1);

    assert!(
        kit.module
            .delete_endpoint(&*kit.db(), ALICE, &endpoint_id)
            .await
            .expect("the delete commits")
    );

    let letter_id = kit
        .module
        .dead_letters(&*kit.db(), ALICE, 10)
        .await
        .expect("read")[0]
        .id
        .clone();
    assert!(
        !kit.module
            .replay(&*kit.db(), ALICE, &letter_id, NOW_ISO)
            .await
            .expect("the replay commits"),
        "no endpoint to deliver to, so the replay refuses"
    );
    assert_eq!(
        kit.module
            .dead_letters(&*kit.db(), ALICE, 10)
            .await
            .expect("read")
            .len(),
        1,
        "the letter stays put"
    );
    assert_eq!(
        kit.drain().await.claimed,
        0,
        "and nothing was enqueued behind the refusal"
    );
}

/// A type no HTTP header could carry is refused at the door — publish and
/// filter registration alike — so nothing undeliverable is ever enqueued
/// to burn its retry budget.
#[pollster::test]
async fn unusable_event_types_are_refused_at_publish_and_at_registration() {
    let kit = Kit::mount(5, &[]);
    kit.create_endpoint(ALICE, URL_A, &[]).await;

    let error = kit
        .module
        .publish(&*kit.db(), ALICE, "two tokens", &json!({}), NOW_ISO)
        .await
        .expect_err("whitespace makes the type undeliverable");
    assert!(
        matches!(error, PublishError::InvalidEventType(ref t) if t == "two tokens"),
        "{error}"
    );

    let error = kit
        .module
        .create_endpoint(
            &*kit.db(),
            ALICE,
            URL_B,
            &["order.paid", "bad type"],
            NOW_ISO,
        )
        .await
        .expect_err("a filter naming an undeliverable type is refused");
    assert!(
        matches!(error, EndpointError::InvalidEventType(ref t) if t == "bad type"),
        "{error}"
    );
    // The refused endpoint was not inserted: only the catch-all is listed.
    assert_eq!(
        kit.module
            .endpoints(&*kit.db(), ALICE)
            .await
            .expect("read")
            .len(),
        1
    );
}

#[pollster::test]
async fn the_signature_header_verifies_against_the_endpoints_secret() {
    let kit = Kit::mount(5, &[]);
    let secret = kit
        .module
        .create_endpoint(&*kit.db(), ALICE, URL_A, &["order.paid"], NOW_ISO)
        .await
        .expect("the endpoint registers")
        .secret;
    let published = kit
        .publish(ALICE, "order.paid", json!({ "invoice": "in_3" }))
        .await;
    kit.drain().await;

    let captured = kit.http.captured();
    assert_eq!(captured.len(), 1);
    let delivery = &captured[0];

    // The delivery names the event twice over: in the envelope and in the
    // headers, so a receiver can dedupe without parsing the body.
    assert_eq!(header(delivery, "Cratefield-Event-Id"), published.event_id);
    assert_eq!(header(delivery, "Cratefield-Event-Type"), "order.paid");
    assert_eq!(header(delivery, "Content-Type"), "application/json");

    // The receiver's side of the contract: split the header, recompute the
    // MAC over `{t}.{body}` with the secret from registration, compare.
    let signature = header(delivery, "Cratefield-Signature");
    let (t, v1) = signature
        .split_once(",v1=")
        .map(|(t, v1)| (t.trim_start_matches("t="), v1))
        .expect("the header shape");
    let expected = {
        let mut mac = Hmac::<Sha256>::new_from_slice(secret.as_bytes()).expect("any key length");
        mac.update(t.as_bytes());
        mac.update(b".");
        mac.update(&delivery.body);
        mac.finalize().into_bytes()
    };
    assert_eq!(hex(&expected), v1, "the delivery is authentic");

    // ...and the binding is real: a tampered body (or a re-stamped one)
    // does not verify.
    let tampered = {
        let mut mac = Hmac::<Sha256>::new_from_slice(secret.as_bytes()).expect("any key length");
        mac.update(t.as_bytes());
        mac.update(b".");
        mac.update(b"{\"invoice\":\"in_4\"}");
        mac.finalize().into_bytes()
    };
    assert_ne!(hex(&tampered), v1);
}

/// Lowercase hex, the encoding the signature header carries.
fn hex(bytes: &[u8]) -> String {
    let mut out = String::with_capacity(bytes.len() * 2);
    for byte in bytes {
        // Writing to a String cannot fail.
        let _ = write!(out, "{byte:02x}");
    }
    out
}

#[pollster::test]
async fn an_unset_alias_sends_exactly_one_signature_header() {
    let kit = Kit::mount(5, &[]);
    kit.create_endpoint(ALICE, URL_A, &[]).await;
    kit.publish(ALICE, "order.paid", json!({ "n": 1 })).await;
    kit.drain().await;

    let captured = kit.http.captured();
    assert_eq!(captured.len(), 1);
    let delivery = &captured[0];
    for name in [
        "Content-Type",
        "Cratefield-Signature",
        "Cratefield-Event-Id",
        "Cratefield-Event-Type",
    ] {
        header(delivery, name);
    }
    let mut names: Vec<&str> = delivery
        .headers
        .iter()
        .map(|(name, _)| name.as_str())
        .collect();
    names.sort_unstable();
    assert_eq!(
        names,
        [
            "content-type",
            "cratefield-event-id",
            "cratefield-event-type",
            "cratefield-signature",
        ],
        "no alias header appears"
    );
    assert_eq!(kit.module.signature_headers(), ["Cratefield-Signature"]);
}

#[pollster::test]
async fn a_configured_alias_carries_the_same_signature_under_both_names() {
    let config = MapConfig::from_pairs([("WEBHOOKS_SIGNATURE_ALIAS", "Owlpost-Signature")]);
    assert!(Webhooks::new().validate_config(&config).is_ok());
    let module = Webhooks::from_config(&config).expect("a valid alias starts");
    assert_eq!(
        module.signature_headers(),
        ["Cratefield-Signature", "Owlpost-Signature"]
    );

    let kit = Kit::mount_module(module, &[]);
    let secret = kit
        .module
        .create_endpoint(&*kit.db(), ALICE, URL_A, &[], NOW_ISO)
        .await
        .expect("the endpoint registers")
        .secret;
    kit.publish(ALICE, "order.paid", json!({ "n": 1 })).await;
    kit.drain_under(Arc::new(config)).await;

    let captured = kit.http.captured();
    assert_eq!(captured.len(), 1);
    let delivery = &captured[0];
    let signature = header(delivery, "Cratefield-Signature");
    assert_eq!(header(delivery, "Owlpost-Signature"), signature);
    assert_eq!(
        delivery.headers.len(),
        5,
        "the alias adds exactly one header"
    );

    // A receiver verifies the one MAC under either name.
    let (t, v1) = signature
        .split_once(",v1=")
        .map(|(t, v1)| (t.trim_start_matches("t="), v1))
        .expect("the header shape");
    let mut mac = Hmac::<Sha256>::new_from_slice(secret.as_bytes()).expect("any key length");
    mac.update(t.as_bytes());
    mac.update(b".");
    mac.update(&delivery.body);
    assert_eq!(
        hex(&mac.finalize().into_bytes()),
        v1,
        "the delivery is authentic"
    );
}

/// A bad alias is refused by `from_config` and by `validate_config`, the two
/// places a deployment's config is checked, and the problem names the key.
#[test]
fn a_malformed_signature_alias_is_refused_at_startup() {
    for bad in [
        "   ",
        " Owlpost-Signature",
        "Owlpost Signature",
        "Owlpost:Signature",
        "Owlpost/Signature",
        "héllo",
    ] {
        let config = MapConfig::from_pairs([("WEBHOOKS_SIGNATURE_ALIAS", bad)]);
        let error = Webhooks::from_config(&config).expect_err("not an HTTP header token");
        assert!(
            error.to_string().contains("WEBHOOKS_SIGNATURE_ALIAS"),
            "{bad:?}: {error}"
        );
        assert!(Webhooks::new().validate_config(&config).is_err(), "{bad:?}");
    }
}

/// The alias may not be a header the module already sends, whatever its case.
#[test]
fn a_signature_alias_naming_an_engine_header_is_refused() {
    for sent in [
        "Cratefield-Signature",
        "cratefield-signature",
        "CONTENT-TYPE",
        "Cratefield-Event-Id",
        "cratefield-event-type",
    ] {
        let config = MapConfig::from_pairs([("WEBHOOKS_SIGNATURE_ALIAS", sent)]);
        assert!(
            Webhooks::from_config(&config).is_err(),
            "{sent} is already sent"
        );
        assert!(Webhooks::new().validate_config(&config).is_err(), "{sent}");
    }
}

/// The transport owns these: an alias may not name one, whatever its case.
#[test]
fn a_signature_alias_naming_a_transport_header_is_refused() {
    for owned in [
        "Host",
        "content-length",
        "Transfer-Encoding",
        "Connection",
        "TE",
        "Trailer",
        "Upgrade",
        "Keep-Alive",
        "Proxy-Authorization",
        "proxy-connection",
    ] {
        let config = MapConfig::from_pairs([("WEBHOOKS_SIGNATURE_ALIAS", owned)]);
        let error = Webhooks::from_config(&config).expect_err("the transport owns it");
        assert!(
            error.to_string().contains("WEBHOOKS_SIGNATURE_ALIAS"),
            "{owned}: {error}"
        );
        assert!(Webhooks::new().validate_config(&config).is_err(), "{owned}");
    }
}

/// An empty value is the same as an unset one, not a malformed alias.
#[test]
fn an_empty_signature_alias_means_unset() {
    let config = MapConfig::from_pairs([("WEBHOOKS_SIGNATURE_ALIAS", "")]);
    let module = Webhooks::from_config(&config).expect("empty is unset");
    assert_eq!(module.signature_headers(), ["Cratefield-Signature"]);
    assert!(Webhooks::new().validate_config(&config).is_ok());
}

/// A module built with `new()` still sends the alias its drain config names:
/// the wire follows the config the pass is given, not the constructor.
#[pollster::test]
async fn a_new_module_still_sends_the_alias_the_drain_config_names() {
    let kit = Kit::mount(5, &[]);
    let config: Arc<dyn Config> = Arc::new(MapConfig::from_pairs([(
        "WEBHOOKS_SIGNATURE_ALIAS",
        "Owlpost-Signature",
    )]));
    kit.create_endpoint(ALICE, URL_A, &[]).await;
    kit.publish(ALICE, "order.paid", json!({ "n": 1 })).await;
    kit.drain_under(config).await;

    let captured = kit.http.captured();
    assert_eq!(captured.len(), 1);
    let delivery = &captured[0];
    assert_eq!(
        header(delivery, "Owlpost-Signature"),
        header(delivery, "Cratefield-Signature"),
        "both names carry the one MAC"
    );
    assert_eq!(delivery.headers.len(), 5);
}

/// An alias that is unusable when the pass runs sends nothing, and leaves the
/// queued delivery where it was for a later, correct pass.
#[pollster::test]
async fn a_bad_alias_at_drain_time_sends_nothing_and_keeps_the_queue() {
    let kit = Kit::mount(5, &[]);
    kit.create_endpoint(ALICE, URL_A, &[]).await;
    kit.publish(ALICE, "order.paid", json!({ "n": 1 })).await;

    let bad: Arc<dyn Config> = Arc::new(MapConfig::from_pairs([(
        "WEBHOOKS_SIGNATURE_ALIAS",
        "Host",
    )]));
    let error = kit
        .module
        .drain_with(&kit.context(bad))
        .await
        .expect_err("the pass refuses the alias");
    assert!(
        error.to_string().contains("WEBHOOKS_SIGNATURE_ALIAS"),
        "{error}"
    );
    assert!(kit.http.captured().is_empty(), "nothing was sent");

    kit.drain().await;
    assert_eq!(kit.http.captured().len(), 1, "the row was still queued");
}

#[pollster::test]
async fn a_410_gone_never_retries() {
    let kit = Kit::mount(5, &[410]);
    kit.create_endpoint(ALICE, URL_A, &[]).await;
    kit.publish(ALICE, "order.paid", json!({ "n": 1 })).await;

    let report = kit.drain().await;
    assert_eq!(report.claimed, 1);
    assert_eq!(report.dead_lettered, 1);
    assert_eq!(report.retried, 0, "the endpoint said never again");

    let letters = kit
        .module
        .dead_letters(&*kit.db(), ALICE, 10)
        .await
        .expect("read");
    assert_eq!(letters.len(), 1);
    assert_eq!(letters[0].reason, "rejected");
    assert_eq!(letters[0].status_code, Some(410));
    assert_eq!(letters[0].attempts, 1, "one attempt, no budget burned");
}

#[pollster::test]
async fn the_config_override_tightens_the_attempt_budget() {
    // The compiled setting is 3; the deployment says 1. One 500 is then a
    // dead letter, not a retry — `WEBHOOKS_MAX_ATTEMPTS` is read by the
    // drain, through the context a scheduled invocation carries.
    let kit = Kit::mount(3, &[500]);
    kit.create_endpoint(ALICE, URL_A, &[]).await;
    kit.publish(ALICE, "order.paid", json!({ "n": 1 })).await;

    let config = Arc::new(MapConfig::from_pairs([("WEBHOOKS_MAX_ATTEMPTS", "1")]));
    let report = kit
        .module
        .drain_with(&kit.context(config))
        .await
        .expect("the drain runs");
    assert_eq!(report.dead_lettered, 1);
    assert_eq!(report.retried, 0);

    let letters = kit
        .module
        .dead_letters(&*kit.db(), ALICE, 10)
        .await
        .expect("read");
    assert_eq!(letters[0].attempts, 1);
}

/// A backlog larger than one tick's allowance is delivered across ticks,
/// each tick spending no more subrequests than its budget allows — the
/// "N due items per tick" contract of ADR 0023, charged one per outbound
/// `POST`.
#[pollster::test]
async fn a_backlog_drains_across_ticks_within_the_subrequest_budget() {
    const QUEUED: usize = 10;
    /// Delivery attempts (subrequests) one tick's budget allows.
    const PER_TICK: u32 = 3;
    let per_tick = usize::try_from(PER_TICK).expect("three fits a usize");

    let kit = Kit::mount(5, &[]);
    kit.create_endpoint(ALICE, URL_A, &[]).await;
    for n in 0..QUEUED {
        kit.publish(ALICE, "order.paid", json!({ "n": n })).await;
    }

    // A fresh budget per tick, the runtime's contract. ceil(10 / 3) = 4
    // ticks deliver the lot.
    let mut delivered = 0;
    for tick in 0..QUEUED.div_ceil(per_tick) {
        let budget = Arc::new(ScheduledBudget::new(None, Some(PER_TICK)));
        let before = kit.http.captured().len();
        let report = kit.drain_within(&budget).await;
        let sent = kit.http.captured().len() - before;
        assert!(sent <= per_tick, "tick {tick} sent {sent} POSTs");
        assert!(report.claimed <= per_tick, "tick {tick}: {report:?}");
        assert_eq!(
            budget.spent(),
            u32::try_from(report.claimed).expect("small"),
            "one subrequest per attempt, and never an overdraft"
        );
        assert_eq!(report.delivered, report.claimed, "every answer was a 200");
        assert_eq!(
            report.deferred, 0,
            "the budget capped the claim, nothing else"
        );
        delivered += report.delivered;
    }

    // Everything went out, exactly once: one POST per queued delivery, one
    // log row each, all first attempts.
    assert_eq!(delivered, QUEUED);
    assert_eq!(kit.http.captured().len(), QUEUED, "one POST per delivery");
    let log = kit.deliveries(ALICE).await;
    assert_eq!(log.len(), QUEUED);
    assert!(
        log.iter().all(|d| d.attempt == 1),
        "each row was delivered on its first attempt"
    );

    // Nothing waits behind the budget any more: a fresh tick finds nothing.
    let budget = Arc::new(ScheduledBudget::new(None, Some(PER_TICK)));
    let report = kit.drain_within(&budget).await;
    assert_eq!(report.claimed, 0);
    assert_eq!(report.deferred, 0);
}

/// A budget with nothing left delivers nothing and leaves the queue for
/// the next tick — whichever bound exhausted it — rather than draining it
/// anyway.
#[pollster::test]
async fn an_exhausted_budget_delivers_nothing_and_leaves_the_queue_intact() {
    let kit = Kit::mount(5, &[]);
    kit.create_endpoint(ALICE, URL_A, &[]).await;
    kit.publish(ALICE, "order.paid", json!({ "n": 1 })).await;

    // No subrequests to spend.
    let empty = Arc::new(ScheduledBudget::new(None, Some(0)));
    let report = kit.drain_within(&empty).await;
    assert_eq!(report.claimed, 0);
    assert_eq!(
        report.deferred, 0,
        "nothing was claimed, so nothing was released"
    );
    assert_eq!(kit.http.captured().len(), 0, "not one POST was made");
    assert_eq!(empty.spent(), 0);

    // The deadline already passed: the other way to arrive exhausted.
    let expired = Arc::new(ScheduledBudget::new(
        Some(OffsetDateTime::from_unix_timestamp(NOW_UNIX - 1).expect("in range")),
        None,
    ));
    assert_eq!(kit.drain_within(&expired).await.claimed, 0);
    assert_eq!(kit.http.captured().len(), 0);

    // Neither pass leased the row out from under the next: it is still due,
    // and a fresh budget delivers it exactly once.
    let budget = Arc::new(ScheduledBudget::new(None, Some(3)));
    let report = kit.drain_within(&budget).await;
    assert_eq!(report.claimed, 1);
    assert_eq!(report.delivered, 1);
    assert_eq!(kit.deliveries(ALICE).await.len(), 1, "logged exactly once");
}

/// The module needs a database, an HTTP client and a clock; a context
/// without one cannot start a drain, and says so rather than panicking —
/// or, for the clock, rather than silently falling back to the wall clock
/// a test never pinned.
async fn drain_without(kit: &Kit, absent: &str) -> String {
    let config: Arc<dyn Config> = Arc::new(MapConfig::default());
    let mut ports = kit.wired_ports(&config);
    match absent {
        "db" => ports.db = None,
        "HttpClient" => ports.http = None,
        "Clock" => ports.clock = None,
        other => panic!("no such port: {other}"),
    }
    kit.module
        .drain_with(&kit.context_over(ports, config))
        .await
        .expect_err("a missing port refuses to start a drain")
        .to_string()
}

#[pollster::test]
async fn a_drain_without_the_http_port_refuses_to_start() {
    let kit = Kit::mount(5, &[]);
    let error = drain_without(&kit, "HttpClient").await;
    assert!(
        error.contains("HttpClient"),
        "the error names the missing port: {error}"
    );
}

#[pollster::test]
async fn a_drain_without_the_clock_port_refuses_to_start() {
    let kit = Kit::mount(5, &[]);
    let error = drain_without(&kit, "Clock").await;
    assert!(
        error.contains("Clock"),
        "the error names the missing port: {error}"
    );
}

#[pollster::test]
async fn a_drain_without_the_db_port_refuses_to_start() {
    let kit = Kit::mount(5, &[]);
    let error = drain_without(&kit, "db").await;
    assert!(error.contains("Database"), "{error}");
}

/// The port set a deployment must resolve for this module to build at all.
#[test]
fn the_module_declares_the_ports_it_needs() {
    assert_eq!(
        Webhooks::new().requires(),
        [Port::Db, Port::HttpClient, Port::Clock]
    );
}