techne-sdk 0.3.8

//! The stock Tap client (`techne_sdk::tap`), driven the way the platform's indexer tests drive its own
//! Tap source (ported, P20.7 G4): acks every `ack_max_frames`, a failed commit acks nothing, malformed
//! versus unreadable frames, a silent connection, and recorded frames replayed.
//!
//! `fixtures/tap/frames.jsonl` holds frames recorded on 2026-10-11 from a Tap built at the platform's
//! pinned indigo commit (patched to signal multiple upstreams), acks on, reading
//! relay1.us-east.bsky.network for repos added with `/repos/add`. They are verbatim except that the
//! DIDs of the third-party accounts and one handle were replaced (`did:plc:tapfixture…`,
//! `liker.example.com`); the calendar repo is a public test event. On that run a newly added repo got
//! its identity frame, then all of its historical frames, then its first live frame: history before
//! live is Tap's guarantee, and the tests below check that this client keeps it (handling frames in
//! arrival order, acking them only after their commits are made durable).

#![cfg(feature = "tap")]

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

use async_trait::async_trait;
use techne_sdk::tap::{
    events, parse_frame, run_connection, CommitOp, EventSink, FirehoseEvent, Incoming, Progress, TapConnection, TapFrame, TapKeepalive, TapStats,
};

const FRAMES: &str = include_str!("fixtures/tap/frames.jsonl");
const CALENDAR_EVENT: &str = "community.lexicon.calendar.event";

fn recorded() -> Vec<String> {
    FRAMES.lines().filter(|l| !l.trim().is_empty()).map(str::to_owned).collect()
}

/// The DAG-CBOR CID (v1, sha2-256) of a JSON record body with no links or bytes.
#[cfg(feature = "crypto")]
fn cid_for_json(body: &serde_json::Value) -> String {
    use ipld_core::cid::{multihash::Multihash, Cid};
    use ipld_core::ipld::Ipld;
    use sha2::Digest as _;
    let ipld: Ipld = serde_json::from_value(body.clone()).unwrap();
    let bytes = serde_ipld_dagcbor::to_vec(&ipld).unwrap();
    let digest = sha2::Sha256::digest(&bytes);
    Cid::new_v1(0x71, Multihash::wrap(0x12, &digest).unwrap()).to_string()
}

#[test]
fn recorded_frames_map_to_the_reader_s_events() {
    let frames: Vec<TapFrame> = recorded().iter().map(|l| parse_frame(l).expect("every recorded frame parses")).collect();
    assert_eq!(frames.len(), 8);

    // An identity frame: the account's status, then its handle.
    let evs = events(&frames[0]);
    assert!(matches!(&evs[0], FirehoseEvent::Account(a) if a.active && a.did == "did:plc:4jsg4522wmbohqveapunbj36"));
    assert!(matches!(&evs[1], FirehoseEvent::Identity(i) if i.handle.as_deref() == Some("feld.pod.exppad.com")));

    // A backfilled record: an upsert with its body, CID and the backfill commit's revision.
    let TapFrame::Record { record, .. } = &frames[1] else { panic!("frame 2 is a record") };
    assert!(!record.live, "a backfill frame is historical");
    let evs = events(&frames[1]);
    let FirehoseEvent::Commit(c) = &evs[0] else { panic!("a commit") };
    assert_eq!((c.op, c.collection.as_str(), c.rkey.as_str(), c.rev.as_str()), (CommitOp::Upsert, CALENDAR_EVENT, "test", "3ll5at2p33c2j"));
    // Tap's JSON body re-encodes to the CID the repo committed, so a reader's CID check holds.
    #[cfg(feature = "crypto")]
    assert_eq!(cid_for_json(c.record.as_ref().unwrap()), c.cid.clone().unwrap());
    // time_us comes from the revision's TID (telemetry): 3ll5at2p33c2j is 2025-03.
    assert!(c.time_us > 1_740_000_000_000_000 && c.time_us < 1_745_000_000_000_000, "{}", c.time_us);

    // A live delete: no body, no CID.
    let del = frames.iter().find(|f| matches!(f, TapFrame::Record { record, .. } if record.action == "delete")).unwrap();
    assert!(del.is_live());
    let FirehoseEvent::Commit(c) = &events(del)[0] else { panic!("a commit") };
    assert_eq!((c.op, c.record.is_none(), c.cid.is_none()), (CommitOp::Delete, true, true));

    // A live create.
    let FirehoseEvent::Commit(c) = &events(&frames[7])[0] else { panic!("a commit") };
    assert!(frames[7].is_live());
    assert_eq!(c.op, CommitOp::Upsert);
    // (Its body names other accounts' DIDs, replaced in the fixture, so its CID no longer matches it.)
    assert!(c.record.is_some() && c.cid.is_some());
}

/// In a script: the stream pauses here (nothing is there to read for a moment), then goes on.
const PAUSE: &str = "<pause>";

/// A connection scripted with frames, logging each ack into the shared log.
struct Scripted {
    frames: std::vec::IntoIter<String>,
    log: Arc<Mutex<Vec<String>>>,
}

#[async_trait]
impl TapConnection for Scripted {
    async fn next_frame(&mut self) -> Option<anyhow::Result<Incoming>> {
        if self.frames.as_slice().first().map(String::as_str) == Some(PAUSE) {
            // Like Tap holding a repo's next live event: nothing more comes until what was sent is
            // acked. A reader that waited for more before acking would hang here (the tests bound it).
            // The marker is taken only once the wait is over, so a cancelled call waits again.
            while !self.log.lock().unwrap().last().is_some_and(|l| l.starts_with("ack ")) {
                tokio::time::sleep(Duration::from_millis(5)).await;
            }
            self.frames.next();
        }
        self.frames.next().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(())
    }
}

/// The reader's durability barrier, logged: `save <key> <newest frame id>`.
struct Barrier {
    log: 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.log.lock().unwrap().push(format!("save {key} {value}"));
        Ok(())
    }
}

fn barrier(log: &Arc<Mutex<Vec<String>>>) -> Barrier {
    Barrier { log: log.clone() }
}

const KEY: &str = "tap:reader@ws://tap/channel";

/// [`run_connection`] with no collection filter, the default keepalive and a barrier logging into
/// `log` (the same log the scripted connection's acks and the sink's commits go to, so their order is
/// one sequence).
async fn run(conn: &mut dyn TapConnection, sink: &dyn EventSink, log: &Arc<Mutex<Vec<String>>>, stats: &mut TapStats) -> anyhow::Result<()> {
    run_connection(conn, sink, &barrier(log), KEY, None, &TapKeepalive::default(), stats).await
}

fn logged(ev: &FirehoseEvent) -> String {
    match ev {
        FirehoseEvent::Commit(c) => format!("commit {}/{}", c.did, c.rkey),
        FirehoseEvent::Account(a) => format!("account {}", a.did),
        FirehoseEvent::Identity(i) => format!("identity {}", i.did),
    }
}

/// A sink that logs each event it commits; it fails (commits nothing) on the `fail_at`-th event.
struct Logging {
    log: Arc<Mutex<Vec<String>>>,
    seen: Mutex<usize>,
    fail_at: Option<usize>,
}

fn logging(log: &Arc<Mutex<Vec<String>>>, fail_at: Option<usize>) -> Logging {
    Logging { log: log.clone(), seen: Mutex::new(0), fail_at }
}

#[async_trait]
impl EventSink for Logging {
    async fn handle(&self, event: FirehoseEvent) -> anyhow::Result<()> {
        let mut seen = self.seen.lock().unwrap();
        *seen += 1;
        if Some(*seen) == self.fail_at {
            anyhow::bail!("the store is down");
        }
        self.log.lock().unwrap().push(logged(&event));
        Ok(())
    }
}

fn scripted(frames: Vec<String>, log: &Arc<Mutex<Vec<String>>>) -> Scripted {
    Scripted { frames: frames.into_iter(), log: log.clone() }
}

/// What the log holds for `frames` handled as one batch: every frame's events, then ONE save naming
/// the newest frame, then the acks.
fn batch(frames: &[String]) -> Vec<String> {
    let frames: Vec<TapFrame> = frames.iter().map(|l| parse_frame(l).unwrap()).collect();
    let mut out: Vec<String> = frames.iter().flat_map(events).map(|e| logged(&e)).collect();
    out.push(format!("save {KEY} {}", frames.last().unwrap().id()));
    out.extend(frames.iter().map(|f| format!("ack {}", f.id())));
    out
}

/// Recorded frames replayed: no frame is acked before a save that came after every commit it mapped
/// to. Frames already waiting to be read share one save; when the stream pauses, what was handled is
/// saved and acked at once (Tap holds a repo's next live event until then, as the scripted pause does).
#[tokio::test]
async fn recorded_frames_are_acked_only_after_a_save_that_follows_their_commits() {
    let log = Arc::new(Mutex::new(Vec::new()));
    let sink = logging(&log, None);
    let mut stats = TapStats::default();
    let frames = recorded();
    let mut script = frames[..3].to_vec();
    script.push(PAUSE.to_owned());
    script.extend_from_slice(&frames[3..]);
    tokio::time::timeout(Duration::from_secs(10), run(&mut scripted(script, &log), &sink, &log, &mut stats))
        .await
        .expect("the frames before the pause were acked without waiting for more")
        .unwrap();
    let mut expected = batch(&frames[..3]);
    expected.extend(batch(&frames[3..]));
    assert_eq!(log.lock().unwrap().clone(), expected);
    assert_eq!(stats, TapStats { acked: 8, historical: 4, live: 2, identity: 2, skipped: 0 });

    // The same rule over the whole log, whatever the batching: each ack comes after a save that
    // comes after the last commit of that ack's frame.
    let log = log.lock().unwrap().clone();
    let position = |what: &str| log.iter().position(|l| l == what).unwrap_or_else(|| panic!("{what} is not in the log"));
    for line in &frames {
        let frame = parse_frame(line).unwrap();
        let ack = position(&format!("ack {}", frame.id()));
        let save = log[..ack].iter().rposition(|l| l.starts_with("save ")).expect("a save before every ack");
        for ev in events(&frame) {
            let commit = position(&logged(&ev));
            assert!(commit < save, "frame {}'s commit is in the save before its ack", frame.id());
        }
    }
}

/// A busy stream is still acked in bounded batches: at most `ack_max_frames` frames wait.
#[tokio::test]
async fn a_busy_stream_is_acked_every_ack_max_frames() {
    let log = Arc::new(Mutex::new(Vec::new()));
    let sink = logging(&log, None);
    let keepalive = TapKeepalive { ack_max_frames: 3, ..TapKeepalive::default() };
    let frames = recorded();
    run_connection(&mut scripted(frames.clone(), &log), &sink, &barrier(&log), KEY, None, &keepalive, &mut TapStats::default()).await.unwrap();
    let mut expected = batch(&frames[..3]);
    expected.extend(batch(&frames[3..6]));
    expected.extend(batch(&frames[6..]));
    assert_eq!(log.lock().unwrap().clone(), expected);
}

#[tokio::test]
async fn a_failed_commit_acks_nothing_and_the_redelivered_frame_is_taken() {
    let log = Arc::new(Mutex::new(Vec::new()));
    // Events: frame 2 = account + identity (events 1, 2), frame 4 = commit (event 3): it fails.
    let sink = logging(&log, Some(3));
    let mut stats = TapStats::default();
    let err = run(&mut scripted(recorded(), &log), &sink, &log, &mut stats).await.unwrap_err();
    assert!(format!("{err:#}").contains("the store is down"), "{err:#}");
    assert_eq!(
        log.lock().unwrap().clone(),
        vec![
            "account did:plc:4jsg4522wmbohqveapunbj36".to_owned(),
            "identity did:plc:4jsg4522wmbohqveapunbj36".to_owned(),
            format!("save {KEY} 2"),
            "ack 2".to_owned(),
        ],
        "frame 4 failed to commit, so it was not acked and nothing after it was read; frame 2 before it was made durable and acked"
    );
    assert_eq!(stats.acked, 1);
    // Tap sends the unacked frame again on the next connection (with whatever follows).
    log.lock().unwrap().clear();
    let mut stats = TapStats::default();
    run(&mut scripted(recorded()[1..].to_vec(), &log), &sink, &log, &mut stats).await.unwrap();
    assert_eq!(log.lock().unwrap().clone(), batch(&recorded()[1..]));
    assert_eq!(stats.acked, 7);
}

#[tokio::test]
async fn a_malformed_frame_is_acked_and_skipped_and_an_unreadable_one_is_not_acked() {
    let log = Arc::new(Mutex::new(Vec::new()));
    let sink = logging(&log, None);
    let frames = vec![
        r#"{"id":5,"type":"record","record":{"live":"yes"}}"#.to_owned(),
        "not json".to_owned(),
        r#"{"id":6,"type":"newthing","newthing":{}}"#.to_owned(),
    ];
    let mut stats = TapStats::default();
    run(&mut scripted(frames, &log), &sink, &log, &mut stats).await.unwrap();
    // Nothing was written, so there is nothing to make durable: no save.
    assert_eq!(log.lock().unwrap().clone(), vec!["ack 5", "ack 6"]);
    assert_eq!(stats, TapStats { acked: 1, skipped: 2, ..TapStats::default() });
}

/// A connection that goes quiet: nothing to read until a ping, then (if `answers`) a pong per ping.
/// After `close_after` pings it closes.
struct Quiet {
    pongs: usize,
    pings: usize,
    answers: bool,
    close_after: usize,
}

#[async_trait]
impl TapConnection for Quiet {
    async fn next_frame(&mut self) -> Option<anyhow::Result<Incoming>> {
        if self.pings >= self.close_after {
            return None;
        }
        if self.pongs > 0 {
            self.pongs -= 1;
            return Some(Ok(Incoming::Control));
        }
        std::future::pending().await
    }
    async fn ack(&mut self, _id: u64) -> anyhow::Result<()> {
        Ok(())
    }
    async fn ping(&mut self) -> anyhow::Result<()> {
        self.pings += 1;
        if self.answers {
            self.pongs += 1;
        }
        Ok(())
    }
}

fn fast() -> TapKeepalive {
    TapKeepalive {
        ping_every: Duration::from_millis(20),
        idle_timeout: Duration::from_millis(100),
        connect_timeout: Duration::from_secs(1),
        backoff_min: Duration::from_millis(10),
        backoff_max: Duration::from_millis(80),
        ..TapKeepalive::default()
    }
}

/// A half-open connection (the server gone, no FIN) is noticed and dropped, so the lane reconnects
/// instead of waiting forever; a quiet connection that answers its pings is kept.
#[tokio::test]
async fn a_silent_connection_is_dropped_and_a_quiet_live_one_is_kept() {
    let log = Arc::new(Mutex::new(Vec::new()));
    let sink = logging(&log, None);
    let mut dead = Quiet { pongs: 0, pings: 0, answers: false, close_after: usize::MAX };
    let err = tokio::time::timeout(Duration::from_secs(5), run_connection(&mut dead, &sink, &barrier(&log), KEY, None, &fast(), &mut TapStats::default()))
        .await
        .expect("a dead connection does not hang the lane")
        .unwrap_err();
    assert!(format!("{err:#}").contains("the connection is dead"), "{err:#}");
    assert!(dead.pings >= 2, "it pinged before giving up: {}", dead.pings);

    // 12 pings at 20 ms is well past the 100 ms idle limit: the pongs are what keep it.
    let mut live = Quiet { pongs: 0, pings: 0, answers: true, close_after: 12 };
    run_connection(&mut live, &sink, &barrier(&log), KEY, None, &fast(), &mut TapStats::default()).await.expect("kept until it closed");
    assert_eq!(live.pings, 12);
    assert!(log.lock().unwrap().is_empty(), "no frames, so no commits, saves or acks: {:?}", log.lock().unwrap());
}