techne-sdk 0.3.3

//! A stock client for Bluesky's Tap (feature `tap`): verified sync, delivered as events to a sink you
//! give it, each frame acked once what it delivered is durable.
//!
//! [Tap](https://github.com/bluesky-social/indigo/tree/main/cmd/tap) is the Sync 1.1 reference
//! consumer. It reads a relay's (or a PDS's) `com.atproto.sync.subscribeRepos`, verifies each
//! commit's signature against the repo's DID document, the MST operations against `prevData`, and
//! identity; it backfills every repo it tracks from the repo's own PDS (`com.atproto.sync.getRepo`,
//! the commit signature checked) and holds that repo's live events until the backfill is out. What a
//! reader gets on `WS /channel` is JSON, one event per text frame, filtered to the collections Tap was
//! started with (`TAP_COLLECTION_FILTERS`):
//!
//! ```text
//! {"id":4,"type":"record","record":{"live":false,"did":…,"rev":…,"collection":…,"rkey":…,
//!   "action":"create"|"update"|"delete","record":{…}?,"cid":…?}}
//! {"id":1,"type":"identity","identity":{"did":…,"handle":…,"is_active":true,"status":"active"}}
//! ```
//!
//! (indigo `cmd/tap/types.go`, `MarshallableEvt`.)
//!
//! **Acks.** Tap runs with acks on (its default): a frame stays in Tap's outbox until the reader sends
//! `{"type":"ack","id":N}`, and an unacked frame is sent again after `TAP_RETRY_TIMEOUT` or after a
//! restart; an acked frame is never sent again. So this client acks a frame only once every event it
//! maps to is DURABLE, not merely handed over: a reader's store may commit asynchronously, and a
//! commit its database has not flushed is lost in a crash. Handled frames wait for one durability
//! barrier ([`Progress::save`], which must return only once every earlier write is on disk) and are
//! acked after it, together ([`run_connection`]). A sink or barrier error acks nothing more and drops
//! the connection; the lane reconnects and Tap delivers the frames again. So delivery is at least once
//! and there is no cursor of ours: Tap keeps it (the upstream firehose cursor and the outbox). The
//! barrier's row ([`durable_key`]) is telemetry. A reader must make a repeat harmless (order by repo
//! revision, a repeat a no-op).
//!
//! **Order.** Per repo, Tap sends a live event (`live: true`) only after every earlier event of that
//! repo is acked, and sends nothing else of that repo until the live one is acked; historical events
//! (`live: false`, a backfill or a resync) may be in flight together. A newly tracked repo's history is
//! therefore delivered before any of its live events. This client handles frames one at a time, in the
//! order they arrive, and acks them in that order after their barrier, so it never weakens that order.
//! Because Tap holds a repo's next live event until this ack, the barrier runs as soon as no further
//! frame is already waiting, never on a timer alone.
//!
//! **Mapping** to [`FirehoseEvent`]: `create`/`update` → [`CommitOp::Upsert`], `delete` →
//! [`CommitOp::Delete`]; an `identity` frame → an [`AccountEvent`] (`is_active`) and, when it carries
//! a handle, an [`IdentityEvent`]. Tap folds the firehose's `#account` into `identity` frames
//! (`status`: active, deactivated, takendown, suspended, deleted). Tap frames carry no time; `time_us`
//! is the repo revision's TID time (telemetry only: order by revision).
//!
//! One Tap serves one reader: Tap shards frames across the WebSockets connected to it, so two readers
//! on one Tap would each see part of the stream. Each reader runs its own Tap (or Taps, one per
//! upstream: Tap reads exactly one relay or PDS), or its own lane of a fan-out.
//!
//! Ported from the platform's `regenos_indexer::tap` (P20.7): the event types, [`EventSink`] and
//! [`Progress`] keep the platform's names, fields and variants, so a reader switches with its `use`s.

use std::collections::HashSet;
use std::fmt;
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use anyhow::Context as _;
use async_trait::async_trait;
use futures_util::{SinkExt, StreamExt};
use serde::Deserialize;
use serde_json::Value;

// ── The events ──────────────────────────────────────────────────────────────────────────────────

/// What a commit did to a record.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CommitOp {
    /// A record was created or updated.
    Upsert,
    /// A record was deleted. The firehose omits the cid on a delete, so a delete keys only on the
    /// natural key.
    Delete,
}

/// A repo-commit event: a record was created/updated or deleted in some account's repo.
#[derive(Debug, Clone, PartialEq)]
pub struct CommitEvent {
    /// The repo (account) DID.
    pub did: String,
    /// The collection NSID.
    pub collection: String,
    /// The record key.
    pub rkey: String,
    /// Create/update vs delete.
    pub op: CommitOp,
    /// The record body (absent on delete).
    pub record: Option<Value>,
    /// The record CID (absent on delete).
    pub cid: Option<String>,
    /// The repository revision of the commit.
    pub rev: String,
    /// Event time in microseconds since the Unix epoch.
    pub time_us: i64,
}

/// An account-status event: `active=false` when the account is deactivated, taken down, suspended
/// or deleted.
#[derive(Debug, Clone, PartialEq)]
pub struct AccountEvent {
    /// The account DID.
    pub did: String,
    /// Whether the account is active.
    pub active: bool,
    /// The account was deleted (`status: deleted`), not only deactivated or taken down.
    pub deleted: bool,
    /// Event time in microseconds since the Unix epoch.
    pub time_us: i64,
}

/// An identity event (handle change).
#[derive(Debug, Clone, PartialEq)]
pub struct IdentityEvent {
    /// The account DID.
    pub did: String,
    /// The new handle, if present in the event.
    pub handle: Option<String>,
    /// Event time in microseconds since the Unix epoch.
    pub time_us: i64,
}

/// One event from the stream, independent of the wire format.
#[derive(Debug, Clone, PartialEq)]
pub enum FirehoseEvent {
    /// A repo-commit (record create/update/delete).
    Commit(CommitEvent),
    /// An account-status change.
    Account(AccountEvent),
    /// An identity (handle) change.
    Identity(IdentityEvent),
}

impl FirehoseEvent {
    /// The event time in microseconds, regardless of variant.
    pub fn time_us(&self) -> i64 {
        match self {
            FirehoseEvent::Commit(e) => e.time_us,
            FirehoseEvent::Account(e) => e.time_us,
            FirehoseEvent::Identity(e) => e.time_us,
        }
    }
}

/// Where a lane delivers its events.
#[async_trait]
pub trait EventSink: Send + Sync {
    /// Take one event. An error drops the connection without acking the frame, so Tap sends it again.
    async fn handle(&self, event: FirehoseEvent) -> anyhow::Result<()>;
    /// Take one event of a hinted frame (a fan-out's fast lane, [`TapRecord::hinted`]): its author is
    /// waiting on it. Handled like any other unless the reader puts that work first.
    async fn handle_hinted(&self, event: FirehoseEvent) -> anyhow::Result<()> {
        self.handle(event).await
    }
    /// A line for the progress log.
    fn progress(&self) -> String {
        String::new()
    }
}

/// Saved progress by key. For a Tap lane, `save` is the durability barrier before an ack: it must
/// return only once every write the sink made before it is durable (a synchronous commit on the
/// sink's own database does that).
#[async_trait]
pub trait Progress: Send + Sync {
    /// The saved progress under `key`.
    async fn load(&self, key: &str) -> anyhow::Result<Option<String>>;
    /// Save `value` under `key`.
    async fn save(&self, key: &str, value: &str) -> anyhow::Result<()>;
}

// ── Frames ──────────────────────────────────────────────────────────────────────────────────────

/// The record half of a Tap `record` frame (`cmd/tap/types.go`, `RecordEvt`).
#[derive(Debug, Clone, PartialEq, Deserialize)]
pub struct TapRecord {
    /// `true` when the event came over the firehose, `false` for backfill or resync.
    pub live: bool,
    pub did: String,
    pub rev: String,
    pub collection: String,
    pub rkey: String,
    /// `create`, `update` or `delete`.
    pub action: String,
    /// The record body (absent on a delete, and when Tap could not decode the block).
    #[serde(default)]
    pub record: Option<Value>,
    #[serde(default)]
    pub cid: Option<String>,
    /// Not Tap's: a fan-out's fast lane. The record's author told the platform about it, and the
    /// platform fetched and verified it: someone is waiting on it.
    #[serde(default)]
    pub hinted: bool,
}

/// The identity half of a Tap `identity` frame (`IdentityEvt`).
#[derive(Debug, Clone, PartialEq, Deserialize)]
pub struct TapIdentity {
    pub did: String,
    /// Empty when Tap has not resolved one.
    #[serde(default)]
    pub handle: String,
    pub is_active: bool,
    /// active, deactivated, takendown, suspended or deleted.
    #[serde(default)]
    pub status: String,
}

/// One frame from Tap's `/channel`.
#[derive(Debug, Clone, PartialEq)]
pub enum TapFrame {
    Record { id: u64, record: TapRecord },
    Identity { id: u64, identity: TapIdentity },
    /// A frame type this reader does not know (acked and skipped, so a newer Tap cannot wedge a repo).
    Unknown { id: u64, kind: String },
}

impl TapFrame {
    /// The outbox id to ack.
    pub fn id(&self) -> u64 {
        match self {
            TapFrame::Record { id, .. } | TapFrame::Identity { id, .. } | TapFrame::Unknown { id, .. } => *id,
        }
    }

    /// Whether this is a live event (`record.live`); identity frames count as live.
    pub fn is_live(&self) -> bool {
        match self {
            TapFrame::Record { record, .. } => record.live,
            TapFrame::Identity { .. } | TapFrame::Unknown { .. } => true,
        }
    }
}

#[derive(Deserialize)]
struct WireFrame {
    #[serde(rename = "type")]
    kind: String,
    #[serde(default)]
    record: Option<TapRecord>,
    #[serde(default)]
    identity: Option<TapIdentity>,
}

/// Why a frame could not be read.
#[derive(Debug)]
pub enum FrameError {
    /// Not JSON, or no numeric `id`: there is nothing to ack.
    Unreadable(String),
    /// A known type whose body does not parse; carries the id so the frame can still be acked.
    Malformed { id: u64, kind: String, reason: String },
}

impl fmt::Display for FrameError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            FrameError::Unreadable(reason) => write!(f, "unreadable Tap frame: {reason}"),
            FrameError::Malformed { id, kind, reason } => write!(f, "malformed Tap {kind} frame {id}: {reason}"),
        }
    }
}

impl std::error::Error for FrameError {}

/// Parse one text frame.
pub fn parse_frame(text: &str) -> Result<TapFrame, FrameError> {
    let raw: Value = serde_json::from_str(text).map_err(|e| FrameError::Unreadable(e.to_string()))?;
    let id = raw["id"].as_u64().ok_or_else(|| FrameError::Unreadable("no numeric id".into()))?;
    let kind = raw["type"].as_str().unwrap_or_default().to_owned();
    let malformed = |reason: String| FrameError::Malformed { id, kind: kind.clone(), reason };
    let wire: WireFrame = serde_json::from_value(raw).map_err(|e| malformed(e.to_string()))?;
    match wire.kind.as_str() {
        "record" => {
            let record = wire.record.ok_or_else(|| malformed("no `record` object".into()))?;
            Ok(TapFrame::Record { id, record })
        }
        "identity" => {
            let identity = wire.identity.ok_or_else(|| malformed("no `identity` object".into()))?;
            Ok(TapFrame::Identity { id, identity })
        }
        _ => Ok(TapFrame::Unknown { id, kind: wire.kind }),
    }
}

/// The ack Tap expects for frame `id` (`cmd/tap/types.go`, `WsResponse`).
pub fn ack_message(id: u64) -> String {
    format!(r#"{{"type":"ack","id":{id}}}"#)
}

/// The events one frame stands for, in the order they are handled. Empty for a frame there is
/// nothing to ingest from (an unknown type, an unknown action).
pub fn events(frame: &TapFrame) -> Vec<FirehoseEvent> {
    match frame {
        TapFrame::Record { record: r, .. } => {
            let op = match r.action.as_str() {
                "create" | "update" => CommitOp::Upsert,
                "delete" => CommitOp::Delete,
                other => {
                    tracing::warn!(action = other, did = %r.did, "a Tap record frame with an unknown action; skipped");
                    return Vec::new();
                }
            };
            let upsert = op == CommitOp::Upsert;
            vec![FirehoseEvent::Commit(CommitEvent {
                did: r.did.clone(),
                collection: r.collection.clone(),
                rkey: r.rkey.clone(),
                op,
                record: if upsert { r.record.clone() } else { None },
                cid: if upsert { r.cid.clone() } else { None },
                rev: r.rev.clone(),
                time_us: tid_time_us(&r.rev).unwrap_or_else(now_us),
            })]
        }
        TapFrame::Identity { identity: i, .. } => {
            let time_us = now_us();
            let mut out = vec![FirehoseEvent::Account(AccountEvent { did: i.did.clone(), active: i.is_active, deleted: i.status == "deleted", time_us })];
            // An empty handle means Tap has none resolved: never clear a known one for that.
            if !i.handle.is_empty() {
                out.push(FirehoseEvent::Identity(IdentityEvent { did: i.did.clone(), handle: Some(i.handle.clone()), time_us }));
            }
            out
        }
        TapFrame::Unknown { kind, id } => {
            tracing::warn!(%kind, id, "a Tap frame of a type this reader does not know; acked and skipped");
            Vec::new()
        }
    }
}

/// The time in a TID (a repo revision), microseconds since the Unix epoch; `None` when `tid` is not
/// one.
pub fn tid_time_us(tid: &str) -> Option<i64> {
    const ALPHABET: &[u8] = b"234567abcdefghijklmnopqrstuvwxyz";
    if tid.len() != 13 {
        return None;
    }
    let mut n: u64 = 0;
    for c in tid.bytes() {
        let digit = ALPHABET.iter().position(|&a| a == c)? as u64;
        n = n.checked_mul(32)?.checked_add(digit)?;
    }
    if n >> 63 != 0 {
        return None;
    }
    Some((n >> 10) as i64)
}

fn now_us() -> i64 {
    SystemTime::now().duration_since(UNIX_EPOCH).map(|d| d.as_micros() as i64).unwrap_or_default()
}

// ── Connections ─────────────────────────────────────────────────────────────────────────────────

/// What a connection delivered: a text frame (Tap's JSON), or a control frame (a pong, a ping), which
/// counts only as a sign of life.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Incoming {
    Text(String),
    Control,
}

/// A connection to one Tap's `/channel`: frames in, acks and keepalive pings out.
#[async_trait]
pub trait TapConnection: Send {
    /// The next frame; `None` when the connection closed. Must be cancel-safe: it runs under a
    /// timeout ([`run_connection`]'s keepalive).
    async fn next_frame(&mut self) -> Option<anyhow::Result<Incoming>>;
    /// Ack frame `id`.
    async fn ack(&mut self, id: u64) -> anyhow::Result<()>;
    /// Send a keepalive ping (a live server answers with a pong).
    async fn ping(&mut self) -> anyhow::Result<()>;
}

/// How a Tap lane keeps its connection honest and comes back: after `ping_every` with no frame it
/// pings; after `idle_timeout` with no frame at all (a pong counts) it drops the connection as dead;
/// connecting is bounded by `connect_timeout`; reconnects back off from `backoff_min`, doubling to
/// `backoff_max`, and a connection that acked a frame resets the backoff. A dropped or silent
/// connection never leaves the lane dead: Tap resends whatever was not acked.
///
/// It also paces the acks ([`run_connection`]): handled frames wait for one durability barrier and
/// are acked together, at the latest once `ack_max_frames` are waiting or the oldest has waited
/// `ack_max_wait`, and at once whenever no further frame is already there to read.
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct TapKeepalive {
    pub ping_every: Duration,
    pub idle_timeout: Duration,
    pub connect_timeout: Duration,
    pub backoff_min: Duration,
    pub backoff_max: Duration,
    /// At most this many handled frames wait unacked for the barrier. Keep it below a fan-out's
    /// per-reader window, so a reader behind a fan-out never stalls it.
    pub ack_max_frames: usize,
    /// The longest a handled frame waits for its ack while frames keep arriving.
    pub ack_max_wait: Duration,
}

impl Default for TapKeepalive {
    fn default() -> Self {
        Self {
            ping_every: Duration::from_secs(30),
            idle_timeout: Duration::from_secs(90),
            connect_timeout: Duration::from_secs(30),
            backoff_min: Duration::from_secs(1),
            backoff_max: Duration::from_secs(60),
            ack_max_frames: 256,
            ack_max_wait: Duration::from_millis(500),
        }
    }
}

impl TapKeepalive {
    /// The wait before the next connection after `failures` connections in a row that acked nothing.
    pub fn backoff(&self, failures: u32) -> Duration {
        self.backoff_min.saturating_mul(2u32.saturating_pow(failures.min(16))).min(self.backoff_max)
    }
}

/// What one connection did.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub struct TapStats {
    /// Frames acked after their events were handled.
    pub acked: u64,
    /// Of those, historical (`live: false`) record frames.
    pub historical: u64,
    /// Of those, live record frames.
    pub live: u64,
    /// Identity frames.
    pub identity: u64,
    /// Frames acked without ingest: a collection this reader does not want, a malformed frame, an
    /// unknown type or action (logged).
    pub skipped: u64,
}

/// Handled frames waiting for the durability barrier before they are acked.
#[derive(Default)]
struct PendingAcks {
    ids: Vec<u64>,
    /// How many of `ids` count in [`TapStats::acked`] (a malformed frame is acked but not counted).
    counted: u64,
    /// Whether any of them wrote through the sink: only then does the ack need a barrier.
    wrote: bool,
    /// When the oldest of them was handled.
    since: Option<tokio::time::Instant>,
}

impl PendingAcks {
    fn push(&mut self, id: u64, counted: bool, wrote: bool) {
        self.ids.push(id);
        self.counted += u64::from(counted);
        self.wrote |= wrote;
        self.since.get_or_insert_with(tokio::time::Instant::now);
    }

    fn is_empty(&self) -> bool {
        self.ids.is_empty()
    }

    fn due(&self, keepalive: &TapKeepalive) -> bool {
        self.ids.len() >= keepalive.ack_max_frames.max(1) || self.since.is_some_and(|s| s.elapsed() >= keepalive.ack_max_wait)
    }

    /// The barrier, then the acks. The barrier is `durable.save(key, <newest id>)`, which returns only
    /// once every event these frames mapped to is durable, so Tap hears of none before. A failed
    /// barrier acks nothing.
    async fn flush(&mut self, conn: &mut dyn TapConnection, durable: &dyn Progress, key: &str, stats: &mut TapStats) -> anyhow::Result<()> {
        let batch = std::mem::take(self);
        let Some(&newest) = batch.ids.last() else { return Ok(()) };
        if batch.wrote {
            durable.save(key, &newest.to_string()).await.map_err(|e| e.context("tap: the durability barrier before the ack"))?;
        }
        for id in batch.ids {
            conn.ack(id).await?;
        }
        stats.acked += batch.counted;
        Ok(())
    }

    /// [`Self::flush`] on the way out with `error`: the frames already handled are still acked when
    /// the barrier holds (so a frame that keeps failing does not hold back those before it).
    async fn ack_handled_before(&mut self, error: anyhow::Error, conn: &mut dyn TapConnection, durable: &dyn Progress, key: &str, stats: &mut TapStats) -> anyhow::Error {
        if let Err(flush) = self.flush(conn, durable, key, stats).await {
            tracing::warn!(error = %format!("{flush:#}"), "tap: the frames handled before the failure were not acked (Tap sends them again)");
        }
        error
    }
}

/// Handle every frame `conn` delivers until it closes: hand each frame's events to `sink` in order,
/// then ack it once they are durable. A sink error returns WITHOUT acking that frame, so Tap delivers
/// it again. A frame that cannot be read at all (no id) is logged and left unacked; a malformed frame
/// with an id is acked and counted as skipped, so one bad frame cannot hold its repo forever. A record
/// of a collection outside `wanted` (when given: the reader's own list, whatever filter the Tap runs
/// with) is acked and skipped. A connection silent past `keepalive.idle_timeout` (pings unanswered) is
/// an error: the caller reconnects.
///
/// **Durable before acked.** Handled frames wait, and are acked only after `durable.save(durable_key,
/// <newest id>)`, which must flush every write before it (none is needed when the waiting frames wrote
/// nothing). The barrier and acks run whenever no next frame is already there to read (so a repo's next
/// live event, which Tap holds until this ack, is never kept waiting on a timer), and at the latest
/// after `keepalive.ack_max_frames` frames or `keepalive.ack_max_wait`. Frames are still handled one at
/// a time in arrival order; acks go out in that order.
pub async fn run_connection(
    conn: &mut dyn TapConnection,
    sink: &dyn EventSink,
    durable: &dyn Progress,
    durable_key: &str,
    wanted: Option<&HashSet<String>>,
    keepalive: &TapKeepalive,
    stats: &mut TapStats,
) -> anyhow::Result<()> {
    let mut last = tokio::time::Instant::now();
    let mut pending = PendingAcks::default();
    loop {
        let next = if pending.is_empty() {
            next_text(conn, keepalive, &mut last).await
        } else {
            // Only a frame already there (by the runtime's next turn, which also takes in what the
            // socket received): when the stream pauses, the waiting frames are acked now. Cancelling
            // `next_text` here is safe (`TapConnection::next_frame` is cancel-safe).
            match tokio::time::timeout(Duration::ZERO, next_text(conn, keepalive, &mut last)).await {
                Ok(next) => next,
                Err(_) => {
                    pending.flush(conn, durable, durable_key, stats).await?;
                    // The barrier's time is ours too, not the connection's silence.
                    last = tokio::time::Instant::now();
                    continue;
                }
            }
        };
        let text = match next {
            None => return pending.flush(conn, durable, durable_key, stats).await,
            Some(Ok(text)) => text,
            Some(Err(e)) => return Err(pending.ack_handled_before(e, conn, durable, durable_key, stats).await),
        };
        match parse_frame(&text) {
            Ok(frame) => {
                let unwanted = match (&frame, wanted) {
                    (TapFrame::Record { record, .. }, Some(w)) => !w.contains(&record.collection),
                    _ => false,
                };
                let evs = if unwanted { Vec::new() } else { events(&frame) };
                let ingest = !evs.is_empty();
                let hinted = matches!(&frame, TapFrame::Record { record, .. } if record.hinted);
                for ev in evs {
                    let handled = if hinted { sink.handle_hinted(ev).await } else { sink.handle(ev).await };
                    if let Err(e) = handled {
                        return Err(pending.ack_handled_before(e, conn, durable, durable_key, stats).await);
                    }
                }
                pending.push(frame.id(), true, ingest);
                match &frame {
                    TapFrame::Record { record, .. } if ingest && record.live => stats.live += 1,
                    TapFrame::Record { .. } if ingest => stats.historical += 1,
                    TapFrame::Record { .. } => stats.skipped += 1,
                    TapFrame::Identity { .. } => stats.identity += 1,
                    TapFrame::Unknown { .. } => stats.skipped += 1,
                }
            }
            Err(FrameError::Malformed { id, kind, reason }) => {
                tracing::error!(id, %kind, %reason, "a malformed Tap frame; acked and skipped (its data is lost)");
                pending.push(id, false, false);
                stats.skipped += 1;
            }
            Err(e @ FrameError::Unreadable(_)) => {
                tracing::error!(error = %e, frame = %text.chars().take(200).collect::<String>(), "an unreadable Tap frame; not acked");
            }
        }
        // Handling time is ours, not the connection's silence.
        last = tokio::time::Instant::now();
        if pending.due(keepalive) {
            pending.flush(conn, durable, durable_key, stats).await?;
            last = tokio::time::Instant::now();
        }
    }
}

/// The next text frame from `conn`, keeping it honest: after `keepalive.ping_every` with nothing it
/// pings; past `keepalive.idle_timeout` with no frame at all (a pong counts; `last` is when the last
/// one came) it is an error. `None` when the connection closed.
pub async fn next_text(conn: &mut dyn TapConnection, keepalive: &TapKeepalive, last: &mut tokio::time::Instant) -> Option<anyhow::Result<String>> {
    loop {
        let next = match tokio::time::timeout(keepalive.ping_every, conn.next_frame()).await {
            Ok(next) => next,
            Err(_) => {
                if last.elapsed() >= keepalive.idle_timeout {
                    return Some(Err(anyhow::anyhow!("tap: no frame for {:?}; the connection is dead", last.elapsed())));
                }
                match tokio::time::timeout(keepalive.ping_every, conn.ping()).await {
                    Ok(Ok(())) => {}
                    Ok(Err(e)) => return Some(Err(e)),
                    Err(_) => return Some(Err(anyhow::anyhow!("tap ping: not sent within {:?}", keepalive.ping_every))),
                }
                continue;
            }
        };
        let item = next?;
        *last = tokio::time::Instant::now();
        match item {
            Ok(Incoming::Text(text)) => return Some(Ok(text)),
            Ok(Incoming::Control) => continue,
            Err(e) => return Some(Err(e)),
        }
    }
}

/// The `/channel` WebSocket URL of a Tap given as `ws://host:2480`, `http://host:2480`, or a full
/// URL with a path (`…/channel`, or a fan-out lane `…/channel/<upstream>/<reader>`), kept as is.
pub fn channel_url(base: &str) -> String {
    let base = base.trim().trim_end_matches('/');
    let base = if let Some(rest) = base.strip_prefix("http://") {
        format!("ws://{rest}")
    } else if let Some(rest) = base.strip_prefix("https://") {
        format!("wss://{rest}")
    } else {
        base.to_owned()
    };
    // A URL that already names a channel is taken as is (`…/channel`, or a fan-out lane
    // `…/channel/<upstream>/<reader>`); anything else gets `/channel` (a Tap under a path prefix too).
    let path = base.split_once("://").map_or(base.as_str(), |(_, r)| r).split_once('/').map_or("", |(_, p)| p);
    if path == "channel" || path.ends_with("/channel") || path.starts_with("channel/") || path.contains("/channel/") {
        base
    } else {
        format!("{base}/channel")
    }
}

/// The Tap URLs a reader follows, from a `TAP_URLS`-style list (comma or space separated), each as its
/// [`channel_url`], once.
pub fn tap_urls(raw: Option<&str>) -> Vec<String> {
    let mut out: Vec<String> = Vec::new();
    for u in raw.unwrap_or_default().split(|c: char| c == ',' || c.is_whitespace()).filter(|s| !s.is_empty()) {
        let u = channel_url(u);
        if !out.contains(&u) {
            out.push(u);
        }
    }
    out
}

type WsStream = tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>;

struct WsConnection(WsStream);

#[async_trait]
impl TapConnection for WsConnection {
    async fn next_frame(&mut self) -> Option<anyhow::Result<Incoming>> {
        use tokio_tungstenite::tungstenite::Message;
        let msg = match self.0.next().await? {
            Ok(m) => m,
            Err(e) => return Some(Err(anyhow::anyhow!("tap read: {e}"))),
        };
        match msg {
            Message::Text(text) => Some(Ok(Incoming::Text(text.as_str().to_owned()))),
            Message::Close(_) => None,
            // Pings (answered by the library), pongs; binary frames are not part of Tap's protocol.
            _ => Some(Ok(Incoming::Control)),
        }
    }

    async fn ack(&mut self, id: u64) -> anyhow::Result<()> {
        self.0
            .send(tokio_tungstenite::tungstenite::Message::text(ack_message(id)))
            .await
            .map_err(|e| anyhow::anyhow!("tap ack {id}: {e}"))
    }

    async fn ping(&mut self) -> anyhow::Result<()> {
        self.0
            .send(tokio_tungstenite::tungstenite::Message::Ping("keepalive".into()))
            .await
            .map_err(|e| anyhow::anyhow!("tap ping: {e}"))
    }
}

/// Connect to the Tap (or a Tap fan-out lane) at `url` (`ws://` or `wss://`), within `timeout`.
pub async fn connect(url: &str, timeout: Duration) -> anyhow::Result<impl TapConnection> {
    match tokio::time::timeout(timeout, tokio_tungstenite::connect_async(url)).await {
        Ok(Ok((ws, _resp))) => Ok(WsConnection(ws)),
        Ok(Err(e)) => Err(anyhow::anyhow!("tap connect: {e}")),
        Err(_) => Err(anyhow::anyhow!("tap connect: no answer within {timeout:?}")),
    }
}

/// The progress key under which a lane on `url` records the newest frame it made durable (its
/// barrier, [`run_connection`]): `tap:<reader>@<url>`, so two readers sharing a database keep
/// separate rows. Telemetry only: nothing resumes from it.
pub fn durable_key(reader: &str, url: &str) -> String {
    format!("tap:{reader}@{url}")
}

/// Follow one Tap forever: connect, [`run_connection`] (acking after `durable.save(durable_key, …)`),
/// and on a close, an error or a silent connection reconnect with backoff ([`TapKeepalive`]). Frames
/// left unacked are sent again by Tap.
pub async fn follow_tap(
    url: String,
    wanted: Option<Arc<HashSet<String>>>,
    sink: Arc<dyn EventSink>,
    durable: Arc<dyn Progress>,
    durable_key: String,
    keepalive: TapKeepalive,
) {
    let mut failures = 0u32;
    loop {
        let mut stats = TapStats::default();
        let outcome = match connect(&url, keepalive.connect_timeout).await {
            Ok(mut conn) => {
                tracing::info!(%url, "following tap");
                run_connection(&mut conn, sink.as_ref(), durable.as_ref(), &durable_key, wanted.as_deref(), &keepalive, &mut stats).await
            }
            Err(error) => Err(error),
        };
        if stats.acked > 0 {
            failures = 0;
        }
        let wait = keepalive.backoff(failures);
        failures = failures.saturating_add(1);
        match outcome {
            Ok(()) => tracing::warn!(%url, ?stats, wait_ms = wait.as_millis() as u64, progress = %sink.progress(), "tap connection closed; reconnecting"),
            Err(error) => tracing::error!(%url, ?stats, wait_ms = wait.as_millis() as u64, error = %format!("{error:#}"),
                "tap lane lost its connection; reconnecting (Tap resends what was not acked)"),
        }
        tokio::time::sleep(wait).await;
    }
}

/// Every Tap in `urls`, forever, each as its own task, keeping only records of `wanted` collections
/// (empty: whatever the Tap sends). Each lane acks only after `durable.save` under
/// [`durable_key`]`(reader, url)`: `durable` must be the reader's own database (the one `sink` writes
/// to), whose save commits synchronously. Returns only when a lane task ends (a panic), with that
/// error. With no URL it waits forever.
pub async fn follow_taps(urls: Vec<String>, wanted: &[String], sink: Arc<dyn EventSink>, durable: Arc<dyn Progress>, reader: &str) -> anyhow::Result<()> {
    if urls.is_empty() {
        tracing::warn!("no Tap configured: no live ingest");
        std::future::pending::<()>().await;
    }
    let wanted = (!wanted.is_empty()).then(|| Arc::new(wanted.iter().cloned().collect::<HashSet<String>>()));
    let mut tasks = tokio::task::JoinSet::new();
    for url in urls {
        let (sink, wanted, durable, key) = (sink.clone(), wanted.clone(), durable.clone(), durable_key(reader, &url));
        tasks.spawn(async move {
            follow_tap(url.clone(), wanted, sink, durable, key, TapKeepalive::default()).await;
            anyhow::bail!("the tap lane {url} stopped")
        });
    }
    let error = match tasks.join_next().await {
        Some(Ok(Ok(()))) => anyhow::anyhow!("a task that runs for the life of the process returned"),
        Some(Ok(Err(error))) => error,
        Some(Err(join)) => anyhow::Error::new(join).context("a task panicked or was cancelled"),
        None => anyhow::anyhow!("no task was running"),
    };
    Err(error).context("a Tap lane ended")
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::Mutex;

    #[test]
    fn channel_urls_take_any_base_form() {
        assert_eq!(channel_url("ws://tap:2480"), "ws://tap:2480/channel");
        assert_eq!(channel_url("http://tap:2480/"), "ws://tap:2480/channel");
        assert_eq!(channel_url("https://tap.example"), "wss://tap.example/channel");
        assert_eq!(channel_url("ws://t:1/channel"), "ws://t:1/channel");
        assert_eq!(channel_url("ws://f:2480/channel/public/indexer"), "ws://f:2480/channel/public/indexer");
        assert_eq!(channel_url("wss://x.example/tap"), "wss://x.example/tap/channel");
        assert_eq!(tap_urls(Some("ws://a:2480, http://a:2480/channel ws://b:2480")), vec!["ws://a:2480/channel", "ws://b:2480/channel"]);
        assert!(tap_urls(None).is_empty());
    }

    #[test]
    fn the_ack_is_tap_s_ws_response() {
        let v: Value = serde_json::from_str(&ack_message(42)).unwrap();
        assert_eq!(v, serde_json::json!({"type": "ack", "id": 42}));
    }

    #[test]
    fn an_identity_with_no_handle_never_clears_one() {
        let f = parse_frame(r#"{"id":7,"type":"identity","identity":{"did":"did:plc:x","handle":"","is_active":false,"status":"takendown"}}"#).unwrap();
        let evs = events(&f);
        assert_eq!(evs.len(), 1);
        assert!(matches!(&evs[0], FirehoseEvent::Account(a) if !a.active && a.did == "did:plc:x"));
    }

    #[test]
    fn unknown_types_and_actions_ingest_nothing() {
        let f = parse_frame(r#"{"id":9,"type":"sync","sync":{}}"#).unwrap();
        assert_eq!(f, TapFrame::Unknown { id: 9, kind: "sync".into() });
        assert!(events(&f).is_empty());
        let f = parse_frame(r#"{"id":10,"type":"record","record":{"live":true,"did":"d","rev":"r","collection":"c","rkey":"k","action":"rename"}}"#).unwrap();
        assert!(events(&f).is_empty());
        assert!(matches!(parse_frame("{\"type\":\"record\"}"), Err(FrameError::Unreadable(_))));
        assert!(matches!(parse_frame(r#"{"id":3,"type":"record"}"#), Err(FrameError::Malformed { id: 3, .. })));
    }

    #[test]
    fn a_record_maps_to_a_commit_timed_by_its_revision() {
        let f = parse_frame(r#"{"id":4,"type":"record","record":{"live":false,"did":"did:plc:a","rev":"3m2aaaaaaaaa2","collection":"c.x","rkey":"k","action":"create","record":{"n":1},"cid":"bafy"}}"#).unwrap();
        assert!(!f.is_live());
        let evs = events(&f);
        let FirehoseEvent::Commit(c) = &evs[0] else { panic!("{evs:?}") };
        assert_eq!((c.op, c.record.clone(), c.cid.as_deref()), (CommitOp::Upsert, Some(serde_json::json!({"n": 1})), Some("bafy")));
        assert_eq!(Some(c.time_us), tid_time_us("3m2aaaaaaaaa2"));
        // A delete carries no body or cid even when the frame has them.
        let f = parse_frame(r#"{"id":5,"type":"record","record":{"live":true,"did":"d","rev":"3m2aaaaaaaab2","collection":"c","rkey":"k","action":"delete","record":{},"cid":"x"}}"#).unwrap();
        assert!(matches!(&events(&f)[0], FirehoseEvent::Commit(c) if c.op == CommitOp::Delete && c.record.is_none() && c.cid.is_none()));
        // A deleted account with a handle: an account event (deleted) then the identity.
        let f = parse_frame(r#"{"id":6,"type":"identity","identity":{"did":"d","handle":"h.test","is_active":false,"status":"deleted"}}"#).unwrap();
        let evs = events(&f);
        assert!(matches!(&evs[0], FirehoseEvent::Account(a) if a.deleted && !a.active));
        assert!(matches!(&evs[1], FirehoseEvent::Identity(i) if i.handle.as_deref() == Some("h.test")));
    }

    #[test]
    fn backoff_doubles_to_its_cap() {
        let k = TapKeepalive::default();
        assert_eq!((k.backoff(0), k.backoff(1), k.backoff(3), k.backoff(99)), (Duration::from_secs(1), Duration::from_secs(2), Duration::from_secs(8), Duration::from_secs(60)));
    }

    /// A sink recording what it got (and whether hinted), failing on a chosen rkey.
    #[derive(Default)]
    struct Recorder {
        log: Arc<Mutex<Vec<String>>>,
        fail_rkey: Option<String>,
    }

    impl Recorder {
        fn line(event: &FirehoseEvent) -> String {
            match event {
                FirehoseEvent::Commit(c) => format!("commit {} {:?}", c.rkey, c.op),
                FirehoseEvent::Account(a) => format!("account {} {}", a.did, a.active),
                FirehoseEvent::Identity(i) => format!("identity {} {}", i.did, i.handle.clone().unwrap_or_default()),
            }
        }
    }

    #[async_trait]
    impl EventSink for Recorder {
        async fn handle(&self, event: FirehoseEvent) -> anyhow::Result<()> {
            if matches!(&event, FirehoseEvent::Commit(c) if Some(&c.rkey) == self.fail_rkey.as_ref()) {
                anyhow::bail!("the sink refused");
            }
            self.log.lock().unwrap().push(Self::line(&event));
            Ok(())
        }
        async fn handle_hinted(&self, event: FirehoseEvent) -> anyhow::Result<()> {
            self.log.lock().unwrap().push(format!("hinted {}", Self::line(&event)));
            Ok(())
        }
    }

    /// A [`Progress`] writing its saves into a shared log.
    struct Barrier(Arc<Mutex<Vec<String>>>);

    #[async_trait]
    impl Progress for Barrier {
        async fn load(&self, _key: &str) -> anyhow::Result<Option<String>> {
            Ok(None)
        }
        async fn save(&self, key: &str, value: &str) -> anyhow::Result<()> {
            self.0.lock().unwrap().push(format!("save {key}={value}"));
            Ok(())
        }
    }

    /// A scripted connection: its frames, then a close; acks go into the shared log.
    struct Scripted {
        frames: std::collections::VecDeque<String>,
        log: Arc<Mutex<Vec<String>>>,
    }

    #[async_trait]
    impl TapConnection for Scripted {
        async fn next_frame(&mut self) -> Option<anyhow::Result<Incoming>> {
            self.frames.pop_front().map(|f| Ok(Incoming::Text(f)))
        }
        async fn ack(&mut self, id: u64) -> anyhow::Result<()> {
            self.log.lock().unwrap().push(format!("ack {id}"));
            Ok(())
        }
        async fn ping(&mut self) -> anyhow::Result<()> {
            Ok(())
        }
    }

    fn record_frame(id: u64, collection: &str, rkey: &str, hinted: bool) -> String {
        serde_json::json!({"id": id, "type": "record", "record": {"live": true, "did": "did:plc:a", "rev": "3m2aaaaaaaaa2",
            "collection": collection, "rkey": rkey, "action": "create", "record": {}, "cid": "bafy", "hinted": hinted}})
        .to_string()
    }

    #[tokio::test]
    async fn the_barrier_comes_before_the_acks_and_a_refused_frame_is_not_acked() {
        let log = Arc::new(Mutex::new(Vec::new()));
        let frames = [
            record_frame(1, "c.want", "a", false),
            record_frame(2, "c.other", "b", false),
            "not json".to_string(),
            r#"{"id":3,"type":"record"}"#.to_string(),
            record_frame(4, "c.want", "h", true),
            record_frame(5, "c.want", "boom", false),
            record_frame(6, "c.want", "never", false),
        ];
        let mut conn = Scripted { frames: frames.into_iter().collect(), log: log.clone() };
        let sink = Recorder { log: log.clone(), fail_rkey: Some("boom".into()) };
        let wanted: HashSet<String> = ["c.want".to_string()].into();
        let mut stats = TapStats::default();
        let err = run_connection(&mut conn, &sink, &Barrier(log.clone()), "k", Some(&wanted), &TapKeepalive::default(), &mut stats).await.unwrap_err();
        assert_eq!(err.to_string(), "the sink refused");
        // Every frame is waiting when the refusal comes (the scripted frames are all already there), so
        // the ones before it go out after one barrier; the refused one and those after it are not acked.
        assert_eq!(
            *log.lock().unwrap(),
            ["commit a Upsert", "hinted commit h Upsert", "save k=4", "ack 1", "ack 2", "ack 3", "ack 4"].map(String::from).to_vec()
        );
        assert_eq!(stats, TapStats { acked: 3, historical: 0, live: 2, identity: 0, skipped: 2 });
    }

    #[tokio::test]
    async fn frames_that_wrote_nothing_need_no_barrier() {
        let log = Arc::new(Mutex::new(Vec::new()));
        let mut conn = Scripted { frames: [r#"{"id":9,"type":"sync"}"#.to_string()].into_iter().collect(), log: log.clone() };
        let mut stats = TapStats::default();
        run_connection(&mut conn, &Recorder::default(), &Barrier(log.clone()), "k", None, &TapKeepalive::default(), &mut stats).await.unwrap();
        assert_eq!(*log.lock().unwrap(), vec!["ack 9".to_string()]);
    }

    /// A Tap on a local port: sends `frames` to the first connection, collects the acks it gets until
    /// it has one per frame, then closes. Returns its address and the acks.
    async fn fake_tap(frames: Vec<String>) -> (std::net::SocketAddr, tokio::sync::oneshot::Receiver<Vec<Value>>) {
        use tokio_tungstenite::tungstenite::Message;
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let (tx, rx) = tokio::sync::oneshot::channel();
        tokio::spawn(async move {
            let (tcp, _) = listener.accept().await.unwrap();
            let mut ws = tokio_tungstenite::accept_async(tcp).await.unwrap();
            for f in &frames {
                ws.send(Message::text(f.clone())).await.unwrap();
            }
            let mut acks = Vec::new();
            while acks.len() < frames.len() {
                match ws.next().await {
                    Some(Ok(Message::Text(t))) => acks.push(serde_json::from_str::<Value>(t.as_str()).unwrap()),
                    Some(Ok(_)) => {}
                    _ => break,
                }
            }
            let _ = ws.close(None).await;
            let _ = tx.send(acks);
        });
        (addr, rx)
    }

    fn two_frames() -> Vec<String> {
        vec![
            r#"{"id":11,"type":"record","record":{"live":false,"did":"did:plc:a","rev":"3m2aaaaaaaaa2","collection":"c.x","rkey":"r1","action":"create","record":{"n":1},"cid":"bafy"}}"#.into(),
            r#"{"id":12,"type":"identity","identity":{"did":"did:plc:a","handle":"a.test","is_active":true,"status":"active"}}"#.into(),
        ]
    }

    /// The two frames' events in order; the barrier may run once (after both) or twice (when frame 12
    /// was not there yet), but always last after frame 12.
    fn assert_delivered_then_saved(log: &[String], key: &str) {
        let events: Vec<&str> = log.iter().map(String::as_str).filter(|l| !l.starts_with("save ")).collect();
        assert_eq!(events, ["commit r1 Upsert", "account did:plc:a true", "identity did:plc:a a.test"], "{log:?}");
        assert_eq!(log.last().map(String::as_str), Some(format!("save {key}=12").as_str()), "{log:?}");
    }

    #[tokio::test]
    async fn run_connection_over_a_websocket_delivers_then_acks() {
        let (addr, acks) = fake_tap(two_frames()).await;
        let log = Arc::new(Mutex::new(Vec::new()));
        let mut conn = connect(&format!("ws://{addr}/channel"), Duration::from_secs(5)).await.unwrap();
        let mut stats = TapStats::default();
        let sink = Recorder { log: log.clone(), fail_rkey: None };
        tokio::time::timeout(Duration::from_secs(10), run_connection(&mut conn, &sink, &Barrier(log.clone()), "tap:t@x", None, &TapKeepalive::default(), &mut stats))
            .await
            .expect("the connection ends when the Tap closes")
            .unwrap();
        assert_delivered_then_saved(&log.lock().unwrap(), "tap:t@x");
        assert_eq!(acks.await.unwrap(), vec![serde_json::json!({"type": "ack", "id": 11}), serde_json::json!({"type": "ack", "id": 12})]);
        assert_eq!(stats, TapStats { acked: 2, historical: 1, live: 0, identity: 1, skipped: 0 });
    }

    /// `wss://` builds its TLS config from the one provider the crate enables (ring) instead of
    /// panicking for want of a process-wide one: a peer that is not TLS is a plain connect error.
    #[tokio::test]
    async fn wss_needs_no_installed_crypto_provider() {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(async move {
            let (tcp, _) = listener.accept().await.unwrap();
            drop(tcp);
        });
        let refused = connect(&format!("wss://{addr}/channel"), Duration::from_secs(5)).await;
        assert!(refused.is_err());
    }

    #[tokio::test]
    async fn follow_tap_delivers_and_acks() {
        let (addr, acks) = fake_tap(two_frames()).await;
        let log = Arc::new(Mutex::new(Vec::new()));
        let sink: Arc<dyn EventSink> = Arc::new(Recorder { log: log.clone(), fail_rkey: None });
        let durable: Arc<dyn Progress> = Arc::new(Barrier(log.clone()));
        let wanted = Some(Arc::new(HashSet::from(["c.x".to_string()])));
        let lane = tokio::spawn(follow_tap(channel_url(&format!("http://{addr}")), wanted, sink, durable, durable_key("t", "x"), TapKeepalive::default()));
        let acks = tokio::time::timeout(Duration::from_secs(10), acks).await.expect("the Tap got its acks").unwrap();
        lane.abort();
        assert_eq!(acks, vec![serde_json::json!({"type": "ack", "id": 11}), serde_json::json!({"type": "ack", "id": 12})]);
        assert_delivered_then_saved(&log.lock().unwrap(), "tap:t@x");
    }
}