noetl-tools 3.5.0

//! Spool backends (RFC §8.3) — the durable buffer that holds payload bytes
//! while a downstream is unavailable.
//!
//! Two backends ship here:
//!
//! - [`NatsObjectBackend`] — wraps the NATS Object Store the `nats` tool
//!   already speaks (`object_put` / `object_get` / `object_list` /
//!   `object_delete`). The in-cluster default: NATS is already deployed, so
//!   no external bucket credential is needed and the live outage proof runs
//!   against it.
//! - [`LocalDiskBackend`] — one JSON file per item under a directory; the
//!   CLI / dev backend (RFC Mode/Phase 6 reuses it).
//!
//! `gcs` / `s3` backends (the tenant object store via keychain-auth) are
//! the same trait with an HTTP/SDK body; they are feature-gated and tracked
//! separately — the trait is the seam so adding them is additive.
//!
//! Every backend stores items keyed by [`super::item::SpoolItem::object_key`]
//! (a zero-padded `recv_seq` prefix), so [`SpoolBackend::list`] returns them
//! in **receive order** — the cheap path for `ordering: global`.

use async_trait::async_trait;

use crate::error::ToolError;

use super::item::SpoolItem;

/// One stored item's metadata, returned by [`SpoolBackend::list`] in
/// receive order (lexical by object key == numeric by `recv_seq`).
#[derive(Debug, Clone, PartialEq)]
pub struct SpoolMeta {
    /// Backend object key ([`SpoolItem::object_key`]).
    pub key: String,
    /// Stored byte size — summed for the retention/cost gauge.
    pub size: u64,
}

/// A durable store-and-forward buffer. Implementations hold whatever
/// connection / handle they need; the trait is pure CRUD over keyed items.
#[async_trait]
pub trait SpoolBackend: Send + Sync {
    /// Backend kind name for events / metrics (`nats_object`, `local_disk`).
    fn kind(&self) -> &'static str;

    /// Persist one item. Idempotent on the object key — re-writing the same
    /// key (a redelivered message spooled twice) overwrites rather than
    /// duplicating, so the spool inherits the source's at-least-once
    /// guarantee without double-storing.
    async fn put(&self, item: &SpoolItem) -> Result<(), ToolError>;

    /// List every stored item's metadata in receive order.
    async fn list(&self) -> Result<Vec<SpoolMeta>, ToolError>;

    /// Fetch + decode one item by key.
    async fn get(&self, key: &str) -> Result<SpoolItem, ToolError>;

    /// Delete one item by key (GC after a successful drain).
    async fn delete(&self, key: &str) -> Result<(), ToolError>;

    /// Total bytes currently stored — the live value for the
    /// `noetl_subscription_spool_bytes` gauge + the `max_bytes` ceiling.
    async fn total_bytes(&self) -> Result<u64, ToolError> {
        Ok(self.list().await?.iter().map(|m| m.size).sum())
    }

    /// Number of items currently stored.
    async fn len(&self) -> Result<usize, ToolError> {
        Ok(self.list().await?.len())
    }

    /// True when the spool is empty.
    async fn is_empty(&self) -> Result<bool, ToolError> {
        Ok(self.len().await? == 0)
    }
}

// ---------------------------------------------------------------------------
// local_disk backend
// ---------------------------------------------------------------------------

/// One JSON file per item under `dir`. The CLI / dev backend.
#[derive(Debug, Clone)]
pub struct LocalDiskBackend {
    dir: std::path::PathBuf,
}

impl LocalDiskBackend {
    /// Open (creating if absent) the spool directory.
    pub async fn open(dir: impl Into<std::path::PathBuf>) -> Result<Self, ToolError> {
        let dir = dir.into();
        tokio::fs::create_dir_all(&dir).await.map_err(|e| {
            ToolError::Io(format!("spool dir '{}' create failed: {e}", dir.display()))
        })?;
        Ok(Self { dir })
    }

    fn path_for(&self, key: &str) -> std::path::PathBuf {
        self.dir.join(format!("{key}.json"))
    }
}

#[async_trait]
impl SpoolBackend for LocalDiskBackend {
    fn kind(&self) -> &'static str {
        "local_disk"
    }

    async fn put(&self, item: &SpoolItem) -> Result<(), ToolError> {
        let path = self.path_for(&item.object_key());
        tokio::fs::write(&path, item.to_bytes()).await.map_err(|e| {
            ToolError::Io(format!("spool write '{}' failed: {e}", path.display()))
        })
    }

    async fn list(&self) -> Result<Vec<SpoolMeta>, ToolError> {
        let mut rd = match tokio::fs::read_dir(&self.dir).await {
            Ok(rd) => rd,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
            Err(e) => return Err(ToolError::Io(format!("spool list failed: {e}"))),
        };
        let mut metas = Vec::new();
        while let Some(entry) = rd
            .next_entry()
            .await
            .map_err(|e| ToolError::Io(format!("spool list entry failed: {e}")))?
        {
            let name = entry.file_name().to_string_lossy().into_owned();
            let Some(key) = name.strip_suffix(".json") else {
                continue;
            };
            let size = entry
                .metadata()
                .await
                .map(|m| m.len())
                .unwrap_or(0);
            metas.push(SpoolMeta { key: key.to_string(), size });
        }
        metas.sort_by(|a, b| a.key.cmp(&b.key));
        Ok(metas)
    }

    async fn get(&self, key: &str) -> Result<SpoolItem, ToolError> {
        let path = self.path_for(key);
        let bytes = tokio::fs::read(&path).await.map_err(|e| {
            ToolError::Io(format!("spool read '{}' failed: {e}", path.display()))
        })?;
        SpoolItem::from_bytes(&bytes)
    }

    async fn delete(&self, key: &str) -> Result<(), ToolError> {
        let path = self.path_for(key);
        match tokio::fs::remove_file(&path).await {
            Ok(()) => Ok(()),
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()), // already gone — idempotent
            Err(e) => Err(ToolError::Io(format!(
                "spool delete '{}' failed: {e}",
                path.display()
            ))),
        }
    }
}

// ---------------------------------------------------------------------------
// nats_object backend
// ---------------------------------------------------------------------------

/// Wraps a NATS Object Store bucket — the in-cluster default. Reuses the
/// same `async-nats` Object Store ops the `nats` tool exposes.
#[derive(Clone)]
pub struct NatsObjectBackend {
    store: async_nats::jetstream::object_store::ObjectStore,
    bucket: String,
}

impl NatsObjectBackend {
    /// Open (creating if absent) the Object Store bucket on `js`.
    pub async fn open(
        js: &async_nats::jetstream::Context,
        bucket: &str,
    ) -> Result<Self, ToolError> {
        // Try to open; create on first use so the runtime is self-bootstrapping
        // (ops doesn't have to pre-provision the bucket).
        let store = match js.get_object_store(bucket).await {
            Ok(s) => s,
            Err(_) => js
                .create_object_store(async_nats::jetstream::object_store::Config {
                    bucket: bucket.to_string(),
                    description: Some("NoETL subscription spool (RFC #90 Phase 4)".to_string()),
                    ..Default::default()
                })
                .await
                .map_err(|e| {
                    ToolError::ExecutionFailed(format!(
                        "spool object store bucket '{bucket}' open/create failed: {e}"
                    ))
                })?,
        };
        Ok(Self {
            store,
            bucket: bucket.to_string(),
        })
    }
}

#[async_trait]
impl SpoolBackend for NatsObjectBackend {
    fn kind(&self) -> &'static str {
        "nats_object"
    }

    async fn put(&self, item: &SpoolItem) -> Result<(), ToolError> {
        let key = item.object_key();
        let meta = async_nats::jetstream::object_store::ObjectMetadata {
            name: key.clone(),
            description: Some(item.spool_ref()),
            ..Default::default()
        };
        let mut reader = std::io::Cursor::new(item.to_bytes());
        self.store.put(meta, &mut reader).await.map_err(|e| {
            ToolError::ExecutionFailed(format!(
                "spool object_put '{key}' to '{}' failed: {e}",
                self.bucket
            ))
        })?;
        Ok(())
    }

    async fn list(&self) -> Result<Vec<SpoolMeta>, ToolError> {
        use futures::StreamExt;
        let mut stream = self.store.list().await.map_err(|e| {
            ToolError::ExecutionFailed(format!("spool object_list '{}' failed: {e}", self.bucket))
        })?;
        let mut metas = Vec::new();
        while let Some(item) = stream.next().await {
            match item {
                Ok(info) if !info.deleted => {
                    metas.push(SpoolMeta {
                        key: info.name,
                        size: info.size as u64,
                    });
                }
                Ok(_) => {} // tombstone
                Err(e) => tracing::warn!(bucket = %self.bucket, "spool list entry error: {e}"),
            }
        }
        metas.sort_by(|a, b| a.key.cmp(&b.key));
        Ok(metas)
    }

    async fn get(&self, key: &str) -> Result<SpoolItem, ToolError> {
        use tokio::io::AsyncReadExt;
        let mut object = self.store.get(key).await.map_err(|e| {
            ToolError::ExecutionFailed(format!("spool object_get '{key}' failed: {e}"))
        })?;
        let mut buf = Vec::new();
        object.read_to_end(&mut buf).await.map_err(|e| {
            ToolError::ExecutionFailed(format!("spool object_get '{key}' read failed: {e}"))
        })?;
        SpoolItem::from_bytes(&buf)
    }

    async fn delete(&self, key: &str) -> Result<(), ToolError> {
        self.store.delete(key).await.map_err(|e| {
            ToolError::ExecutionFailed(format!("spool object_delete '{key}' failed: {e}"))
        })
    }
}

// ---------------------------------------------------------------------------
// gcs backend
// ---------------------------------------------------------------------------

/// Google Cloud Storage spool backend — the durable buffer for the
/// out-of-cluster Cloud Run runtime (RFC #90 Phase 5) and any in-cluster
/// runtime that prefers an object store over NATS.
///
/// Reuses the crate's existing dependencies: [`crate::auth::GcpAuth`] for
/// Application Default Credentials (Workload Identity on Cloud Run, the
/// gcloud ADC file locally, or `GOOGLE_APPLICATION_CREDENTIALS`) and
/// `reqwest` against the GCS JSON API — **no new dependency, no gRPC**, the
/// same shape the `pubsub` source backend uses.
///
/// One bucket holds many subscriptions' spools, separated by `prefix`
/// (e.g. `subscriptions/orders/spool/` for the live buffer and
/// `subscriptions/orders/dlq/` for the dead-letter sibling). Items are
/// stored under `{prefix}{object_key}` where `object_key` is the
/// zero-padded `recv_seq` (no slashes), so a `prefix`-scoped list returns
/// them in receive order — the cheap path for `ordering: global`.
///
/// On Cloud Run the credential is the runtime service account via Workload
/// Identity — "already-in-place trust" per `execution-model.md` (no key
/// file, no keychain hop). The keychain-alias path for a *tenant-owned*
/// external bucket is a future extension (the config carries the alias; ADC
/// is the platform-bucket default).
#[cfg(feature = "gcs")]
#[derive(Clone)]
pub struct GcsBackend {
    client: reqwest::Client,
    /// `None` when pointed at a no-auth emulator (fake-gcs-server); else the
    /// ADC token provider.
    auth: Option<crate::auth::GcpAuth>,
    bucket: String,
    /// Object-name prefix (ends with `/` unless empty) so one bucket serves
    /// many subscriptions + the live/dlq split.
    prefix: String,
    /// API base URL with no trailing slash (`https://storage.googleapis.com`
    /// for real GCS; an emulator base for tests).
    endpoint: String,
}

#[cfg(feature = "gcs")]
impl GcsBackend {
    /// Open the GCS spool backend against the real API using ADC.
    ///
    /// `prefix` is normalized to end with `/` (unless empty). The bucket is
    /// assumed to already exist (ops provisions it — GCS bucket creation is
    /// a project-admin op, not a runtime op).
    pub async fn open(bucket: &str, prefix: &str) -> Result<Self, ToolError> {
        Ok(Self {
            client: reqwest::Client::new(),
            auth: Some(crate::auth::GcpAuth::new()),
            bucket: bucket.to_string(),
            prefix: Self::norm_prefix(prefix),
            endpoint: "https://storage.googleapis.com".to_string(),
        })
    }

    /// Open against an explicit endpoint (a fake-gcs-server emulator) with an
    /// optional ADC provider — the seam the integration test + dev recipe use.
    pub fn with_endpoint(bucket: &str, prefix: &str, endpoint: &str, use_adc: bool) -> Self {
        Self {
            client: reqwest::Client::new(),
            auth: use_adc.then(crate::auth::GcpAuth::new),
            bucket: bucket.to_string(),
            prefix: Self::norm_prefix(prefix),
            endpoint: endpoint.trim_end_matches('/').to_string(),
        }
    }

    fn norm_prefix(prefix: &str) -> String {
        if prefix.is_empty() || prefix.ends_with('/') {
            prefix.to_string()
        } else {
            format!("{prefix}/")
        }
    }

    /// Full object name for a bare key.
    fn name_for(&self, key: &str) -> String {
        format!("{}{}", self.prefix, key)
    }

    /// Percent-encode an object name for use in a URL path segment — GCS
    /// requires the full name (slashes included) encoded in the `o/{name}`
    /// path. Encodes everything outside the unreserved set.
    fn enc_path(name: &str) -> String {
        let mut out = String::with_capacity(name.len() * 3);
        for b in name.bytes() {
            match b {
                b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'.' | b'_' | b'~' => {
                    out.push(b as char)
                }
                _ => out.push_str(&format!("%{b:02X}")),
            }
        }
        out
    }

    /// Resolve the `Authorization` header value, if auth is configured.
    async fn auth_header(&self) -> Result<Option<String>, ToolError> {
        match &self.auth {
            Some(gcp) => {
                let token = gcp
                    .get_token(&["https://www.googleapis.com/auth/devstorage.read_write"])
                    .await?;
                Ok(Some(format!("Bearer {token}")))
            }
            None => Ok(None),
        }
    }
}

#[cfg(feature = "gcs")]
#[async_trait]
impl SpoolBackend for GcsBackend {
    fn kind(&self) -> &'static str {
        "gcs"
    }

    async fn put(&self, item: &SpoolItem) -> Result<(), ToolError> {
        let name = self.name_for(&item.object_key());
        let url = format!("{}/upload/storage/v1/b/{}/o", self.endpoint, self.bucket);
        let mut req = self
            .client
            .post(&url)
            .query(&[("uploadType", "media"), ("name", name.as_str())])
            .header("Content-Type", "application/json")
            .body(item.to_bytes());
        if let Some(auth) = self.auth_header().await? {
            req = req.header("Authorization", auth);
        }
        let resp = req.send().await.map_err(|e| {
            ToolError::ExecutionFailed(format!("spool gcs put '{name}' failed: {e}"))
        })?;
        if !resp.status().is_success() {
            let status = resp.status();
            let body = resp.text().await.unwrap_or_default();
            return Err(ToolError::ExecutionFailed(format!(
                "spool gcs put '{name}' to '{}' returned {status}: {body}",
                self.bucket
            )));
        }
        Ok(())
    }

    async fn list(&self) -> Result<Vec<SpoolMeta>, ToolError> {
        let url = format!("{}/storage/v1/b/{}/o", self.endpoint, self.bucket);
        let mut metas = Vec::new();
        let mut page_token: Option<String> = None;
        loop {
            let mut query: Vec<(&str, String)> = vec![("prefix", self.prefix.clone())];
            if let Some(tok) = &page_token {
                query.push(("pageToken", tok.clone()));
            }
            let mut req = self.client.get(&url).query(&query);
            if let Some(auth) = self.auth_header().await? {
                req = req.header("Authorization", auth);
            }
            let resp = req.send().await.map_err(|e| {
                ToolError::ExecutionFailed(format!("spool gcs list '{}' failed: {e}", self.bucket))
            })?;
            if !resp.status().is_success() {
                let status = resp.status();
                let body = resp.text().await.unwrap_or_default();
                return Err(ToolError::ExecutionFailed(format!(
                    "spool gcs list '{}' returned {status}: {body}",
                    self.bucket
                )));
            }
            let page: GcsListResponse = resp.json().await.map_err(|e| {
                ToolError::Json(format!("spool gcs list decode failed: {e}"))
            })?;
            for obj in page.items {
                // Strip the prefix so SpoolMeta.key is the bare object_key the
                // engine orders/gets/deletes by.
                let Some(key) = obj.name.strip_prefix(&self.prefix) else {
                    continue;
                };
                if key.is_empty() {
                    continue; // the prefix "directory" placeholder, if any
                }
                let size = obj.size.parse::<u64>().unwrap_or(0);
                metas.push(SpoolMeta {
                    key: key.to_string(),
                    size,
                });
            }
            match page.next_page_token {
                Some(tok) if !tok.is_empty() => page_token = Some(tok),
                _ => break,
            }
        }
        metas.sort_by(|a, b| a.key.cmp(&b.key));
        Ok(metas)
    }

    async fn get(&self, key: &str) -> Result<SpoolItem, ToolError> {
        let name = self.name_for(key);
        let url = format!(
            "{}/storage/v1/b/{}/o/{}",
            self.endpoint,
            self.bucket,
            Self::enc_path(&name)
        );
        let mut req = self.client.get(&url).query(&[("alt", "media")]);
        if let Some(auth) = self.auth_header().await? {
            req = req.header("Authorization", auth);
        }
        let resp = req.send().await.map_err(|e| {
            ToolError::ExecutionFailed(format!("spool gcs get '{name}' failed: {e}"))
        })?;
        if !resp.status().is_success() {
            let status = resp.status();
            let body = resp.text().await.unwrap_or_default();
            return Err(ToolError::ExecutionFailed(format!(
                "spool gcs get '{name}' returned {status}: {body}"
            )));
        }
        let bytes = resp.bytes().await.map_err(|e| {
            ToolError::ExecutionFailed(format!("spool gcs get '{name}' read failed: {e}"))
        })?;
        SpoolItem::from_bytes(&bytes)
    }

    async fn delete(&self, key: &str) -> Result<(), ToolError> {
        let name = self.name_for(key);
        let url = format!(
            "{}/storage/v1/b/{}/o/{}",
            self.endpoint,
            self.bucket,
            Self::enc_path(&name)
        );
        let mut req = self.client.delete(&url);
        if let Some(auth) = self.auth_header().await? {
            req = req.header("Authorization", auth);
        }
        let resp = req.send().await.map_err(|e| {
            ToolError::ExecutionFailed(format!("spool gcs delete '{name}' failed: {e}"))
        })?;
        // 404 == already gone → idempotent, like the other backends.
        if resp.status().is_success() || resp.status() == reqwest::StatusCode::NOT_FOUND {
            Ok(())
        } else {
            let status = resp.status();
            let body = resp.text().await.unwrap_or_default();
            Err(ToolError::ExecutionFailed(format!(
                "spool gcs delete '{name}' returned {status}: {body}"
            )))
        }
    }
}

/// GCS JSON-API object-list response (the fields the spool needs).
#[cfg(feature = "gcs")]
#[derive(serde::Deserialize)]
struct GcsListResponse {
    #[serde(default)]
    items: Vec<GcsObject>,
    #[serde(rename = "nextPageToken")]
    next_page_token: Option<String>,
}

#[cfg(feature = "gcs")]
#[derive(serde::Deserialize)]
struct GcsObject {
    name: String,
    /// GCS reports object size as a decimal string.
    #[serde(default)]
    size: String,
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::tools::source::PolledMessage;

    fn item(seq: u64, id: &str, data: serde_json::Value) -> SpoolItem {
        let msg = PolledMessage {
            id: id.to_string(),
            data,
            headers: serde_json::Map::new(),
            attributes: serde_json::Value::Null,
            metadata: serde_json::Value::Null,
            ack_id: None,
        };
        SpoolItem::new("subscriptions/t", "nats", msg, None, seq, None, "default", "circuit_open", seq)
    }

    #[tokio::test]
    async fn local_disk_put_list_get_delete_roundtrip() {
        let tmp = std::env::temp_dir().join(format!("noetl-spool-test-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&tmp);
        let backend = LocalDiskBackend::open(&tmp).await.unwrap();

        assert!(backend.is_empty().await.unwrap());

        backend.put(&item(2, "b", serde_json::json!({"v": 2}))).await.unwrap();
        backend.put(&item(1, "a", serde_json::json!({"v": 1}))).await.unwrap();
        backend.put(&item(3, "c", serde_json::json!({"v": 3}))).await.unwrap();

        let metas = backend.list().await.unwrap();
        assert_eq!(metas.len(), 3);
        // list must be in receive order despite insert order
        let got = backend.get(&metas[0].key).await.unwrap();
        assert_eq!(got.recv_seq, 1);
        assert_eq!(backend.get(&metas[2].key).await.unwrap().recv_seq, 3);

        assert!(backend.total_bytes().await.unwrap() > 0);

        backend.delete(&metas[0].key).await.unwrap();
        assert_eq!(backend.len().await.unwrap(), 2);
        // delete is idempotent
        backend.delete(&metas[0].key).await.unwrap();

        let _ = std::fs::remove_dir_all(&tmp);
    }

    #[tokio::test]
    async fn local_disk_list_missing_dir_is_empty() {
        let tmp = std::env::temp_dir().join(format!("noetl-spool-missing-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&tmp);
        let backend = LocalDiskBackend { dir: tmp.clone() };
        // dir never created
        assert_eq!(backend.list().await.unwrap().len(), 0);
    }

    #[tokio::test]
    async fn local_disk_overwrite_is_idempotent_on_key() {
        let tmp = std::env::temp_dir().join(format!("noetl-spool-idem-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&tmp);
        let backend = LocalDiskBackend::open(&tmp).await.unwrap();
        // same recv_seq + id → same object key → overwrite, not duplicate
        backend.put(&item(7, "same", serde_json::json!(1))).await.unwrap();
        backend.put(&item(7, "same", serde_json::json!(2))).await.unwrap();
        assert_eq!(backend.len().await.unwrap(), 1);
        let _ = std::fs::remove_dir_all(&tmp);
    }

    // ---- gcs backend ----

    #[cfg(feature = "gcs")]
    #[test]
    fn gcs_prefix_is_normalized_to_trailing_slash() {
        assert_eq!(GcsBackend::norm_prefix("subscriptions/orders"), "subscriptions/orders/");
        assert_eq!(GcsBackend::norm_prefix("subscriptions/orders/"), "subscriptions/orders/");
        assert_eq!(GcsBackend::norm_prefix(""), "");
    }

    #[cfg(feature = "gcs")]
    #[test]
    fn gcs_name_for_joins_prefix_and_key() {
        let b = GcsBackend::with_endpoint("bkt", "sub/spool", "http://x", false);
        assert_eq!(b.name_for("00000000000000000001-abc"), "sub/spool/00000000000000000001-abc");
    }

    #[cfg(feature = "gcs")]
    #[test]
    fn gcs_enc_path_encodes_slashes_and_reserved() {
        // slashes in the prefix must be percent-encoded for the o/{name} path
        assert_eq!(GcsBackend::enc_path("a/b-c.d_e~f"), "a%2Fb-c.d_e~f");
        assert_eq!(GcsBackend::enc_path("k=1&v"), "k%3D1%26v");
        // unreserved set passes through untouched
        assert_eq!(GcsBackend::enc_path("AZaz09-._~"), "AZaz09-._~");
    }

    /// Live round-trip against a real GCS bucket. Gated on
    /// `NOETL_GCS_TEST_BUCKET` (set to a bucket the ambient ADC can write).
    /// Run: `NOETL_GCS_TEST_BUCKET=my-bucket cargo test --features gcs gcs_live -- --ignored --nocapture`.
    #[cfg(feature = "gcs")]
    #[tokio::test]
    #[ignore]
    async fn gcs_live_put_list_get_delete_roundtrip() {
        let Ok(bucket) = std::env::var("NOETL_GCS_TEST_BUCKET") else {
            eprintln!("skipping: NOETL_GCS_TEST_BUCKET unset");
            return;
        };
        let prefix = format!("noetl-spool-test/{}", std::process::id());
        let backend = GcsBackend::open(&bucket, &prefix).await.unwrap();

        assert!(backend.is_empty().await.unwrap(), "test prefix must start empty");

        backend.put(&item(2, "b", serde_json::json!({"v": 2}))).await.unwrap();
        backend.put(&item(1, "a", serde_json::json!({"v": 1}))).await.unwrap();
        backend.put(&item(3, "c", serde_json::json!({"v": 3}))).await.unwrap();

        let metas = backend.list().await.unwrap();
        assert_eq!(metas.len(), 3, "all three items listed");
        // list is in receive order despite insert order
        assert_eq!(backend.get(&metas[0].key).await.unwrap().recv_seq, 1);
        assert_eq!(backend.get(&metas[2].key).await.unwrap().recv_seq, 3);
        assert!(backend.total_bytes().await.unwrap() > 0);

        // payload integrity survives the round-trip
        let got = backend.get(&metas[0].key).await.unwrap();
        assert_eq!(got.message_id, "a");

        backend.delete(&metas[0].key).await.unwrap();
        assert_eq!(backend.len().await.unwrap(), 2);
        // delete is idempotent (404 → Ok)
        backend.delete(&metas[0].key).await.unwrap();

        // clean up the remaining items so reruns start clean
        for m in backend.list().await.unwrap() {
            backend.delete(&m.key).await.unwrap();
        }
        assert!(backend.is_empty().await.unwrap());
    }
}