feat(pds): com.atproto.sync.subscribeRepos — lokaler Firehose

Bisher erreichten eigene Records die AppView nur über den Best-Effort-Push
/internal/ingest-commit. Ging der verloren (AppView kurz weg, Netzwerk-
fehler), war der Post dauerhaft weg: der öffentliche Jetstream kennt diese
PDS nicht, es gab also keinen zweiten Weg.

Jeder Commit schreibt sein Event in derselben Transaktion nach
firehose_events. Damit kann es keinen Commit ohne Event geben — und keine
Sequenz ohne Commit.

Die seq muss lückenfrei sein, sonst ist sie als Cursor wertlos: BIGSERIAL
vergibt Nummern bei INSERT, nicht bei COMMIT, also können zwei Schreiber 5
und 6 ziehen und in umgekehrter Reihenfolge sichtbar werden — ein Leser
dazwischen sieht 6, merkt sich das und erfährt von 5 nie. Ein globaler
pg_advisory_xact_lock unmittelbar vor dem INSERT erzwingt Commit-Reihenfolge
== seq-Reihenfolge. Er wird nach dem per-Repo-FOR-UPDATE genommen, überall in
derselben Reihenfolge, also ohne Deadlock-Risiko. Preis: das Ende jeder
schreibenden Transaktion ist global serialisiert; das steht im Modulkopf.

Der WebSocket-Handler abonniert den Broadcast, *bevor* er die Datenbank
liest, und filtert Live-Events auf seq > Wasserstand. Aus einem Rennen wird
so eine Dublette, die sich filtern lässt, statt einer Lücke, die es nicht
gibt. Ein zu langsamer Consumer bekommt #info/OutdatedCursor und fällt auf
den DB-Replay zurück, statt getrennt zu werden — die Events sind durabel,
also ist der Rückfall verlustfrei.

Frame-Hülle ist konformes DAG-CBOR mit Tag-42-Links (neues Modul dag_cbor,
aus car.rs herausgezogen statt dupliziert). Die Blöcke darin behalten die
Konvention dieses Repos: CIDs als Strings. Ein fremder Consumer liest die
Frames, scheitert aber an den Blockinhalten — das zu ändern hieße, jede CID
im System zu ändern, inklusive der did:plc-Ableitung. Steht so im Modulkopf.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013HC9HLrUU1LNwkzp8nkDLX

Cargo.lock

@@ -51,6 +51,8 @@ dependencies = [
  "axum",
  "base64",
  "chrono",
+ "ciborium",
+ "cid",
  "dotenvy",
  "futures",
  "p256",
@@ -60,6 +62,7 @@ dependencies = [
  "serde_json",
  "sqlx",
  "tokio",
+ "tokio-tungstenite",
  "tower",
  "tower-http",
  "tracing",
@@ -287,6 +290,7 @@ checksum = "edca88bc138befd0323b20752846e6587272d3b03b0343c8ea28a6f819e6e71f"
 dependencies = [
  "async-trait",
  "axum-core",
+ "base64",
  "bytes",
  "futures-util",
  "http",
@@ -305,8 +309,10 @@ dependencies = [
  "serde_json",
  "serde_path_to_error",
  "serde_urlencoded",
+ "sha1",
  "sync_wrapper",
  "tokio",
+ "tokio-tungstenite",
  "tower",
  "tower-layer",
  "tower-service",
@@ -1885,6 +1891,7 @@ dependencies = [
  "ciborium",
  "cid",
  "dotenvy",
+ "futures",
  "hex",
  "k256",
  "p256",
@@ -1895,6 +1902,7 @@ dependencies = [
  "sha2",
  "sqlx",
  "tokio",
+ "tokio-tungstenite",
  "tower",
  "tower-http",
  "tracing",

crates/pds-server/Cargo.toml

@@ -16,7 +16,7 @@ path = "src/main.rs"
 [dependencies]
 tokio = { workspace = true }
 dotenvy = { workspace = true }
-axum = { workspace = true }
+axum = { workspace = true, features = ["ws"] }
 tower = { workspace = true }
 tower-http = { workspace = true }
 serde = { workspace = true }
@@ -39,6 +39,7 @@ hex = { workspace = true }
 rand = { workspace = true }
 uuid = { workspace = true }
 bytes = { workspace = true }
+futures = { workspace = true }
 cid = { workspace = true }
 k256 = { workspace = true }
 p256 = { workspace = true }
@@ -56,3 +57,6 @@ sha2 = { workspace = true }
 hex = { workspace = true }
 sqlx = { workspace = true }
 at-crypto = { workspace = true }
+futures = { workspace = true }
+tokio-tungstenite = { workspace = true }
+chrono = { workspace = true }

crates/pds-server/src/dag_cbor.rs

@@ -0,0 +1,557 @@
+//! A small, self-contained DAG-CBOR encoder + decoder.
+//!
+//! Why this exists
+//!
+//! `ciborium` (the CBOR crate the rest of the workspace uses) speaks plain
+//! CBOR through serde. It has no notion of an IPLD *CID link*, which DAG-CBOR
+//! encodes as the IANA-registered tag `42` wrapping a byte string whose first
+//! byte is the multibase-identity prefix `0x00` followed by the binary CID.
+//! Serde has no representation for a CBOR tag, so `ciborium` silently encodes
+//! `cid::Cid` as a newtype struct instead — which is *not* DAG-CBOR and which
+//! no atproto consumer can read.
+//!
+//! `car.rs` already hand-rolled the handful of primitives needed for the CAR
+//! v1 header (`{version, roots: [<tag 42 link>]}`). The firehose frames need
+//! exactly the same primitives plus a couple more (arrays of maps, nullable
+//! links, byte strings, i64). Rather than write the encoder twice, both
+//! callers now go through this module.
+//!
+//! ## What "correct DAG-CBOR" means here
+//!
+//! * Map keys are text strings, sorted in the DAG-CBOR canonical order:
+//!   shorter keys first, then bytewise-ascending within a length. This is the
+//!   ordering `@ipld/dag-cbor` (and therefore the reference atproto
+//!   implementation) emits, so a frame produced here is byte-identical to one
+//!   produced by a Typescript PDS for the same logical value.
+//! * Integers use the shortest possible head. Byte and text strings likewise.
+//! * A CID link is `tag(42) || bytes(0x00 || <cid.to_bytes()>)`. The leading
+//!   `0x00` is the multibase identity prefix mandated by the DAG-CBOR spec for
+//!   binary CIDs; forgetting it is the single most common interop bug, so the
+//!   decoder asserts on it too.
+//! * Floats are deliberately *not* supported. DAG-CBOR permits them but
+//!   nothing in this codebase emits one, and accepting them would mean
+//!   deciding on a canonical float encoding we would never exercise.
+//!
+//! ## Scope
+//!
+//! This is not a general CBOR library. It handles definite-length items only
+//! (DAG-CBOR forbids indefinite lengths anyway) and rejects everything it does
+//! not understand rather than guessing. It is used for the *envelope* of
+//! things — CAR headers and firehose frames — never for repo blocks, which
+//! are produced by `at_repo`/`at_crypto` with their own (see
+//! `crate::firehose`) conventions.
+
+use anyhow::{anyhow, bail, Result};
+use cid::Cid;
+
+/// The IPLD CID-link tag. See <https://ipld.io/specs/codecs/dag-cbor/spec/>.
+pub const CID_TAG: u64 = 42;
+
+/// A decoded (or to-be-encoded) DAG-CBOR value.
+///
+/// `Link` is kept distinct from `Bytes` so a round-trip through
+/// [`decode`] / [`encode`] preserves the tag rather than flattening a link
+/// into an anonymous byte string.
+#[derive(Debug, Clone, PartialEq)]
+pub enum Value {
+    Null,
+    Bool(bool),
+    /// Signed integer. CBOR major types 0 (non-negative) and 1 (negative).
+    Int(i64),
+    Bytes(Vec<u8>),
+    Text(String),
+    Array(Vec<Value>),
+    /// Map with text keys. Insertion order is irrelevant — [`encode`] sorts
+    /// into the canonical DAG-CBOR order, and [`decode`] returns keys in the
+    /// order they appeared on the wire.
+    Map(Vec<(String, Value)>),
+    /// An IPLD CID link — `tag(42)` wrapping the identity-prefixed CID bytes.
+    Link(Cid),
+}
+
+impl Value {
+    /// Convenience: build a `Map` from an iterator of pairs.
+    pub fn map<I, K>(pairs: I) -> Value
+    where
+        I: IntoIterator<Item = (K, Value)>,
+        K: Into<String>,
+    {
+        Value::Map(pairs.into_iter().map(|(k, v)| (k.into(), v)).collect())
+    }
+
+    /// Convenience: a text value from anything string-ish.
+    pub fn text(s: impl Into<String>) -> Value {
+        Value::Text(s.into())
+    }
+
+    /// Look up a key in a `Map`. Returns `None` for a non-map or a
+    /// missing key. Used by the tests and by frame consumers.
+    pub fn get(&self, key: &str) -> Option<&Value> {
+        match self {
+            Value::Map(entries) => entries.iter().find(|(k, _)| k == key).map(|(_, v)| v),
+            _ => None,
+        }
+    }
+
+    pub fn as_i64(&self) -> Option<i64> {
+        match self {
+            Value::Int(i) => Some(*i),
+            _ => None,
+        }
+    }
+
+    pub fn as_str(&self) -> Option<&str> {
+        match self {
+            Value::Text(s) => Some(s.as_str()),
+            _ => None,
+        }
+    }
+
+    pub fn as_bool(&self) -> Option<bool> {
+        match self {
+            Value::Bool(b) => Some(*b),
+            _ => None,
+        }
+    }
+
+    pub fn as_bytes(&self) -> Option<&[u8]> {
+        match self {
+            Value::Bytes(b) => Some(b.as_slice()),
+            _ => None,
+        }
+    }
+
+    pub fn as_array(&self) -> Option<&[Value]> {
+        match self {
+            Value::Array(a) => Some(a.as_slice()),
+            _ => None,
+        }
+    }
+
+    pub fn as_link(&self) -> Option<&Cid> {
+        match self {
+            Value::Link(c) => Some(c),
+            _ => None,
+        }
+    }
+
+    pub fn is_null(&self) -> bool {
+        matches!(self, Value::Null)
+    }
+}
+
+// -- encoding ---------------------------------------------------------------
+
+/// Write a CBOR head: the 3-bit major type plus the argument, using the
+/// shortest encoding that fits.
+///
+/// Public because `car.rs` builds its length-prefixed sections around the same
+/// primitive and there is no reason to have two copies.
+pub fn write_head(out: &mut Vec<u8>, major: u8, n: u64) {
+    let m = (major & 0x07) << 5;
+    if n < 24 {
+        out.push(m | n as u8);
+    } else if n <= u8::MAX as u64 {
+        out.push(m | 24);
+        out.push(n as u8);
+    } else if n <= u16::MAX as u64 {
+        out.push(m | 25);
+        out.extend_from_slice(&(n as u16).to_be_bytes());
+    } else if n <= u32::MAX as u64 {
+        out.push(m | 26);
+        out.extend_from_slice(&(n as u32).to_be_bytes());
+    } else {
+        out.push(m | 27);
+        out.extend_from_slice(&n.to_be_bytes());
+    }
+}
+
+/// Append a CBOR text string (major type 3).
+pub fn write_text(out: &mut Vec<u8>, s: &str) {
+    write_head(out, 3, s.len() as u64);
+    out.extend_from_slice(s.as_bytes());
+}
+
+/// Append a CBOR byte string (major type 2).
+pub fn write_bytes(out: &mut Vec<u8>, b: &[u8]) {
+    write_head(out, 2, b.len() as u64);
+    out.extend_from_slice(b);
+}
+
+/// Append a CID as a DAG-CBOR link: `tag(42) || bytes(0x00 || cid)`.
+///
+/// The `0x00` is the multibase identity prefix. Binary CIDs inside DAG-CBOR
+/// always carry it; the textual form (`bafy…`) never does.
+pub fn write_link(out: &mut Vec<u8>, cid: &Cid) {
+    write_head(out, 6, CID_TAG);
+    let raw = cid.to_bytes();
+    let mut prefixed = Vec::with_capacity(raw.len() + 1);
+    prefixed.push(0x00);
+    prefixed.extend_from_slice(&raw);
+    write_bytes(out, &prefixed);
+}
+
+/// DAG-CBOR canonical map-key order: shorter keys sort first; equal-length
+/// keys sort bytewise ascending.
+///
+/// This is RFC 7049's "canonical CBOR" rule, which DAG-CBOR inherited and
+/// which `@ipld/dag-cbor` implements. (RFC 8949 later switched the *core*
+/// deterministic profile to plain bytewise ordering, but DAG-CBOR did not
+/// follow — using 8949's rule here would produce frames that differ from the
+/// reference implementation's for keys like `"op"` vs `"t"`.)
+fn canonical_key_cmp(a: &str, b: &str) -> std::cmp::Ordering {
+    a.len()
+        .cmp(&b.len())
+        .then_with(|| a.as_bytes().cmp(b.as_bytes()))
+}
+
+/// Encode a value into `out`.
+pub fn encode_into(out: &mut Vec<u8>, value: &Value) {
+    match value {
+        Value::Null => out.push(0xF6),
+        Value::Bool(false) => out.push(0xF4),
+        Value::Bool(true) => out.push(0xF5),
+        Value::Int(i) => {
+            if *i >= 0 {
+                write_head(out, 0, *i as u64);
+            } else {
+                // CBOR major type 1 stores -1-n, so n = -(i+1). Computed on
+                // i64 via `i128` to stay correct at `i64::MIN`, where
+                // `-(i + 1)` would overflow.
+                let n = (-((*i as i128) + 1)) as u64;
+                write_head(out, 1, n);
+            }
+        }
+        Value::Bytes(b) => write_bytes(out, b),
+        Value::Text(s) => write_text(out, s),
+        Value::Array(items) => {
+            write_head(out, 4, items.len() as u64);
+            for item in items {
+                encode_into(out, item);
+            }
+        }
+        Value::Map(entries) => {
+            let mut sorted: Vec<&(String, Value)> = entries.iter().collect();
+            sorted.sort_by(|a, b| canonical_key_cmp(&a.0, &b.0));
+            write_head(out, 5, sorted.len() as u64);
+            for (k, v) in sorted {
+                write_text(out, k);
+                encode_into(out, v);
+            }
+        }
+        Value::Link(cid) => write_link(out, cid),
+    }
+}
+
+/// Encode a value to a fresh `Vec<u8>`.
+///
+/// `#[allow(dead_code)]`: the server itself always appends into an existing
+/// buffer via [`encode_into`] (a frame is two values in one allocation), so
+/// this convenience wrapper is exercised only by the tests that assert on
+/// exact byte sequences. It is kept because the decoder half needs a matching
+/// encoder half to be testable at all.
+#[allow(dead_code)]
+pub fn encode(value: &Value) -> Vec<u8> {
+    let mut out = Vec::new();
+    encode_into(&mut out, value);
+    out
+}
+
+// -- decoding ---------------------------------------------------------------
+
+// -- Why a decoder lives in a server that only encodes ----------------------
+//
+// The PDS never parses a firehose frame in production — it writes them. The
+// decoder exists so the frame *contract* can be tested from the outside: a
+// unit test that only checks "the encoder produced these bytes" locks in
+// whatever the encoder happens to do, including its bugs. Decoding the bytes
+// back and asserting on the structure is what actually verifies that a tag-42
+// link is a link and not a byte string, that the two frame halves are
+// separable, and that a `null` `since` is `null` rather than absent.
+//
+// The integration test uses it for the same reason from the client side, and
+// `crates/appview` builds its consumer against the same shape.
+
+/// Decode exactly one value, requiring it to consume the whole input.
+#[allow(dead_code)]
+pub fn decode(bytes: &[u8]) -> Result<Value> {
+    let (v, used) = decode_one(bytes)?;
+    if used != bytes.len() {
+        bail!(
+            "trailing bytes after DAG-CBOR value: consumed {used} of {}",
+            bytes.len()
+        );
+    }
+    Ok(v)
+}
+
+/// Decode one value from the front of `bytes`, returning it along with the
+/// number of bytes consumed.
+///
+/// This is the entry point the firehose frame reader needs: a frame is two
+/// concatenated DAG-CBOR values (header then body) with no length prefix
+/// between them, so the only way to find the body is to decode the header and
+/// see where it ended.
+#[allow(dead_code)]
+pub fn decode_one(bytes: &[u8]) -> Result<(Value, usize)> {
+    decode_at(bytes, 0).map(|(v, end)| (v, end))
+}
+
+#[allow(dead_code)]
+fn decode_at(bytes: &[u8], offset: usize) -> Result<(Value, usize)> {
+    let (major, arg, mut p) = read_head(bytes, offset)?;
+    match major {
+        0 => {
+            let i = i64::try_from(arg).map_err(|_| anyhow!("CBOR uint {arg} exceeds i64"))?;
+            Ok((Value::Int(i), p))
+        }
+        1 => {
+            let v = -(arg as i128) - 1;
+            let i = i64::try_from(v).map_err(|_| anyhow!("CBOR nint {v} exceeds i64"))?;
+            Ok((Value::Int(i), p))
+        }
+        2 => {
+            let end = p + arg as usize;
+            if end > bytes.len() {
+                bail!("CBOR byte string exceeds input");
+            }
+            Ok((Value::Bytes(bytes[p..end].to_vec()), end))
+        }
+        3 => {
+            let end = p + arg as usize;
+            if end > bytes.len() {
+                bail!("CBOR text string exceeds input");
+            }
+            let s = std::str::from_utf8(&bytes[p..end])
+                .map_err(|e| anyhow!("invalid UTF-8 in CBOR text: {e}"))?;
+            Ok((Value::Text(s.to_string()), end))
+        }
+        4 => {
+            let mut items = Vec::with_capacity(arg.min(1024) as usize);
+            for _ in 0..arg {
+                let (v, next) = decode_at(bytes, p)?;
+                items.push(v);
+                p = next;
+            }
+            Ok((Value::Array(items), p))
+        }
+        5 => {
+            let mut entries = Vec::with_capacity(arg.min(1024) as usize);
+            for _ in 0..arg {
+                let (k, next) = decode_at(bytes, p)?;
+                p = next;
+                let key = match k {
+                    Value::Text(s) => s,
+                    other => bail!("DAG-CBOR map keys must be text, got {other:?}"),
+                };
+                let (v, next) = decode_at(bytes, p)?;
+                p = next;
+                entries.push((key, v));
+            }
+            Ok((Value::Map(entries), p))
+        }
+        6 => {
+            if arg != CID_TAG {
+                bail!("unsupported CBOR tag {arg}; DAG-CBOR allows only 42");
+            }
+            let (inner, end) = decode_at(bytes, p)?;
+            let raw = match inner {
+                Value::Bytes(b) => b,
+                other => bail!("CBOR tag 42 must wrap a byte string, got {other:?}"),
+            };
+            let stripped = raw
+                .split_first()
+                .filter(|(first, _)| **first == 0x00)
+                .map(|(_, rest)| rest)
+                .ok_or_else(|| {
+                    anyhow!("CID link missing the 0x00 multibase identity prefix")
+                })?;
+            let cid = Cid::read_bytes(stripped)
+                .map_err(|e| anyhow!("invalid CID inside tag 42: {e}"))?;
+            Ok((Value::Link(cid), end))
+        }
+        7 => match arg {
+            20 => Ok((Value::Bool(false), p)),
+            21 => Ok((Value::Bool(true), p)),
+            22 => Ok((Value::Null, p)),
+            // 23 is `undefined`, 25/26/27 are floats. DAG-CBOR forbids
+            // `undefined`; floats are out of scope (see the module header).
+            other => bail!("unsupported CBOR simple/float value {other}"),
+        },
+        other => bail!("unsupported CBOR major type {other}"),
+    }
+}
+
+/// Read a CBOR head at `offset`, returning `(major, argument, next_offset)`.
+///
+/// Public so `car.rs` can drive its own (deliberately non-conformant, see
+/// there) header parser off the same primitive instead of keeping a second
+/// copy.
+pub fn read_head(bytes: &[u8], offset: usize) -> Result<(u8, u64, usize)> {
+    let first = *bytes
+        .get(offset)
+        .ok_or_else(|| anyhow!("CBOR read past end of input at {offset}"))?;
+    let major = first >> 5;
+    let low = first & 0x1f;
+    let (arg, extra) = match low {
+        0..=23 => (low as u64, 0usize),
+        24 => (read_uint(bytes, offset + 1, 1)?, 1),
+        25 => (read_uint(bytes, offset + 1, 2)?, 2),
+        26 => (read_uint(bytes, offset + 1, 4)?, 4),
+        27 => (read_uint(bytes, offset + 1, 8)?, 8),
+        // 28..=30 are reserved; 31 is the indefinite-length marker, which
+        // DAG-CBOR forbids outright.
+        other => bail!("invalid or indefinite CBOR head 0x{other:02x}"),
+    };
+    Ok((major, arg, offset + 1 + extra))
+}
+
+fn read_uint(bytes: &[u8], offset: usize, width: usize) -> Result<u64> {
+    if offset + width > bytes.len() {
+        bail!("truncated CBOR integer of width {width}");
+    }
+    let mut n: u64 = 0;
+    for b in &bytes[offset..offset + width] {
+        n = (n << 8) | *b as u64;
+    }
+    Ok(n)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use at_crypto::cid::cid_for_cbor;
+
+    fn round_trip(v: Value) {
+        let bytes = encode(&v);
+        let back = decode(&bytes).expect("decode");
+        // Maps come back in canonical (encoded) order, so compare the
+        // re-encoding rather than the structure for map-bearing values.
+        assert_eq!(encode(&back), bytes, "re-encode must be stable");
+    }
+
+    #[test]
+    fn scalars_round_trip() {
+        for v in [
+            Value::Null,
+            Value::Bool(true),
+            Value::Bool(false),
+            Value::Int(0),
+            Value::Int(23),
+            Value::Int(24),
+            Value::Int(255),
+            Value::Int(256),
+            Value::Int(65_535),
+            Value::Int(65_536),
+            Value::Int(i64::MAX),
+            Value::Int(-1),
+            Value::Int(-24),
+            Value::Int(-1000),
+            Value::Int(i64::MIN),
+            Value::Text("hello".into()),
+            Value::Text(String::new()),
+            Value::Bytes(vec![1, 2, 3]),
+            Value::Bytes(Vec::new()),
+        ] {
+            let bytes = encode(&v);
+            assert_eq!(decode(&bytes).unwrap(), v, "round trip of {v:?}");
+        }
+    }
+
+    #[test]
+    fn int_heads_are_shortest_form() {
+        assert_eq!(encode(&Value::Int(1)), vec![0x01]);
+        assert_eq!(encode(&Value::Int(24)), vec![0x18, 24]);
+        assert_eq!(encode(&Value::Int(-1)), vec![0x20]);
+        assert_eq!(encode(&Value::Int(-25)), vec![0x38, 24]);
+    }
+
+    #[test]
+    fn cid_link_carries_tag_42_and_identity_prefix() {
+        let cid = cid_for_cbor(b"a block").unwrap();
+        let bytes = encode(&Value::Link(cid));
+        // 0xD8 0x2A == tag(42) in two-byte form.
+        assert_eq!(&bytes[0..2], &[0xD8, 0x2A]);
+        // Then a byte string whose first content byte is the 0x00 prefix.
+        let (_major, len, p) = read_head(&bytes, 2).unwrap();
+        assert_eq!(len as usize, cid.to_bytes().len() + 1);
+        assert_eq!(bytes[p], 0x00);
+        assert_eq!(decode(&bytes).unwrap(), Value::Link(cid));
+    }
+
+    #[test]
+    fn link_without_identity_prefix_is_rejected() {
+        let cid = cid_for_cbor(b"x").unwrap();
+        let mut bytes = Vec::new();
+        write_head(&mut bytes, 6, CID_TAG);
+        // Deliberately omit the leading 0x00.
+        write_bytes(&mut bytes, &cid.to_bytes());
+        let e = decode(&bytes).unwrap_err().to_string();
+        assert!(e.contains("identity prefix"), "got: {e}");
+    }
+
+    #[test]
+    fn map_keys_are_sorted_length_first() {
+        // The firehose header is exactly this shape, and the reference
+        // implementation emits `t` before `op` because it is shorter.
+        let v = Value::map([("op", Value::Int(1)), ("t", Value::text("#commit"))]);
+        let bytes = encode(&v);
+        assert_eq!(bytes[0], 0xA2, "map(2)");
+        assert_eq!(bytes[1], 0x61, "text(1)");
+        assert_eq!(bytes[2], b't');
+        // …and `op` follows after the "#commit" value.
+        let decoded = decode(&bytes).unwrap();
+        match &decoded {
+            Value::Map(entries) => {
+                assert_eq!(entries[0].0, "t");
+                assert_eq!(entries[1].0, "op");
+            }
+            other => panic!("expected map, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn nested_structures_round_trip() {
+        let cid = cid_for_cbor(b"nested").unwrap();
+        round_trip(Value::map([
+            ("seq", Value::Int(42)),
+            ("commit", Value::Link(cid)),
+            (
+                "ops",
+                Value::Array(vec![Value::map([
+                    ("action", Value::text("create")),
+                    ("path", Value::text("app.twi.post/3l")),
+                    ("cid", Value::Link(cid)),
+                ])]),
+            ),
+            ("blobs", Value::Array(vec![])),
+            ("since", Value::Null),
+        ]));
+    }
+
+    #[test]
+    fn decode_one_stops_at_the_value_boundary() {
+        // Two concatenated values — exactly how a firehose frame is laid out.
+        let mut buf = encode(&Value::map([("op", Value::Int(1))]));
+        let header_len = buf.len();
+        buf.extend_from_slice(&encode(&Value::map([("seq", Value::Int(7))])));
+        let (header, used) = decode_one(&buf).unwrap();
+        assert_eq!(used, header_len);
+        assert_eq!(header.get("op").and_then(Value::as_i64), Some(1));
+        let body = decode(&buf[used..]).unwrap();
+        assert_eq!(body.get("seq").and_then(Value::as_i64), Some(7));
+    }
+
+    #[test]
+    fn indefinite_length_is_rejected() {
+        // 0x9F == array(*) — legal CBOR, illegal DAG-CBOR.
+        assert!(decode(&[0x9F, 0x01, 0xFF]).is_err());
+    }
+
+    #[test]
+    fn trailing_bytes_are_rejected() {
+        let mut bytes = encode(&Value::Int(1));
+        bytes.push(0x01);
+        assert!(decode(&bytes).is_err());
+    }
+}

crates/pds-server/src/firehose.rs

@@ -0,0 +1,946 @@
+//! `com.atproto.sync.subscribeRepos` — the event log, the frame codec and the
+//! in-process broadcast channel.
+//!
+//! # What this is for
+//!
+//! Before this module the PDS emitted no firehose. A locally created record
+//! reached the AppView through exactly one channel: the best-effort HTTP push
+//! in [`crate::appview_push`], a detached `tokio::spawn` whose failure branch
+//! logs "jetstream will replay". For records that only exist on this PDS
+//! there *is* no Jetstream to replay them, so a lost push meant the post was
+//! never indexed and nothing would ever notice. The firehose replaces that
+//! hope with a durable, ordered log: every repo write appends one row inside
+//! the same transaction as the commit, and a consumer can ask for everything
+//! after a cursor at any later time.
+//!
+//! # Frame format
+//!
+//! One WebSocket **binary** message is two DAG-CBOR values written back to
+//! back with nothing between them — a header, then a body. There is no length
+//! prefix; the reader decodes the header and continues the body at the offset
+//! where the header ended (that is what [`crate::dag_cbor::decode_one`] is
+//! for).
+//!
+//! ```text
+//! regular:  {"op": 1, "t": "#commit"}  {"seq": …, "repo": …, …}
+//!           {"op": 1, "t": "#info"}    {"name": …, "message": …}
+//! error:    {"op": -1}                 {"error": "<Name>", "message": "<Text>"}
+//! ```
+//!
+//! The `#commit` body carries:
+//!
+//! | field     | type                                                    |
+//! |-----------|---------------------------------------------------------|
+//! | `seq`     | int — the cursor value for this event                   |
+//! | `rebase`  | bool — always `false` (we never rebase a repo)          |
+//! | `tooBig`  | bool — always `false` (see the size note below)         |
+//! | `repo`    | text — the DID                                          |
+//! | `commit`  | **CID link (tag 42)** — the new commit block            |
+//! | `rev`     | text — the new commit's revision                        |
+//! | `since`   | text or null — the previous commit's revision           |
+//! | `blocks`  | byte string — a CAR v1 file, commit block as root       |
+//! | `ops`     | array of `{action, path, cid}`; `cid` is a link or null  |
+//! | `blobs`   | array — always empty (blob refs live inside the record) |
+//! | `time`    | text — RFC 3339, when the event was appended            |
+//!
+//! Map keys are emitted in DAG-CBOR canonical order (shortest first, then
+//! bytewise), so the bytes match what a reference atproto implementation
+//! would produce for the same logical value.
+//!
+//! # Deliberate deviation from the atproto spec
+//!
+//! **The frame envelope is conformant. The blocks inside `blocks` are not.**
+//!
+//! This repository encodes CIDs *inside* commit blocks as CBOR text strings
+//! rather than as DAG-CBOR links with tag 42 — see `at_repo::commit` and
+//! `at_crypto::signing::sign_dag_cbor`. That convention predates this module
+//! and is load-bearing: the block bytes determine every CID in the system,
+//! including the `did:plc:` derivation, so changing it would re-address every
+//! repo in the database. It is explicitly out of scope here.
+//!
+//! The consequence, stated plainly: a foreign atproto consumer can connect,
+//! parse every frame, read `seq` / `repo` / `rev` / `ops`, and follow the
+//! stream. It will then fail when it tries to *validate* the payload — the
+//! CAR in `blocks` parses fine and the block CIDs hash correctly over their
+//! own bytes, but decoding a commit block as DAG-CBOR yields `"prev"` and
+//! `"data"` as strings where the spec demands links, and MST traversal
+//! against a stock implementation will not work. The AppView in this
+//! workspace reads the frames with the same conventions this crate writes
+//! them, which is why it works there.
+//!
+//! `tooBig` is therefore always `false`: it exists so a producer can say "the
+//! diff was too large, go fetch the repo yourself", and we never make that
+//! call — every commit here is one record change, and its CAR is small.
+//!
+//! # Lagging consumers
+//!
+//! The broadcast channel is bounded ([`FIREHOSE_CHANNEL_CAPACITY`]). A
+//! consumer that reads slower than the PDS writes will eventually be lapped,
+//! and `tokio::sync::broadcast` reports that as `RecvError::Lagged(n)`.
+//!
+//! The write path must never wait on a reader, so the channel cannot be made
+//! blocking. When a reader lags we send it an `#info` / `OutdatedCursor`
+//! frame and **fall back to the database replay** from the last sequence it
+//! actually received, then resume live. We do not disconnect it: the events
+//! are durable in `firehose_events`, so the fallback is lossless, whereas
+//! dropping the socket would force the client to reconnect and perform
+//! exactly the same replay after two extra round trips. The only thing a
+//! disconnect would buy is protection against a client that lags forever, and
+//! that is handled separately by capping consecutive recoveries
+//! ([`MAX_LAG_RECOVERIES`]) before closing with an error frame.
+//!
+//! # Retention
+//!
+//! Nothing prunes `firehose_events`. It grows by one row per repo write, each
+//! carrying the CAR of that commit's new blocks. That is unbounded, and this
+//! deployment has no retention job — an operator who wants one has to add it.
+//! Pruning is safe by design: the cursor handshake compares the requested
+//! cursor against the oldest surviving row and answers a too-old cursor with
+//! `#info` / `OutdatedCursor` followed by a replay from the oldest row that
+//! still exists, rather than pretending the gap is not there.
+
+use crate::car::CarWriter;
+use crate::dag_cbor::{encode_into, Value};
+use anyhow::{anyhow, Result};
+use at_crypto::cid::cid_from_multihash_bytes;
+use chrono::{DateTime, SecondsFormat, Utc};
+use cid::Cid;
+use serde_json::json;
+use std::sync::Arc;
+use tokio::sync::broadcast;
+
+/// How many events the live broadcast channel buffers per subscriber before
+/// the slowest one starts reporting `Lagged`.
+///
+/// 1024 is chosen so a consumer can stall for the length of a garbage
+/// collection or a slow network write without falling back to the database,
+/// while the memory ceiling stays bounded: the channel holds `Arc`s, so the
+/// cost is one CAR blob per queued event, shared across all subscribers.
+pub const FIREHOSE_CHANNEL_CAPACITY: usize = 1024;
+
+/// How many times in a row a single connection may be rescued from a lag
+/// before we give up and close it. A client that cannot keep up even with a
+/// database replay in between is not going to start; at that point the honest
+/// answer is an error frame rather than an endless catch-up loop that burns
+/// queries on its behalf.
+pub const MAX_LAG_RECOVERIES: u32 = 5;
+
+/// Rows returned per replay query. Bounded so a client reconnecting with
+/// `cursor=0` after a long uptime streams the backlog in chunks instead of
+/// materialising the whole table (and every CAR in it) in memory at once.
+pub const REPLAY_PAGE_SIZE: i64 = 200;
+
+/// The advisory-lock key that serialises `firehose_events` INSERTs. Any
+/// constant works as long as every writer uses the same one; this is
+/// `"fhose"` read as ASCII, which makes it recognisable in `pg_locks`.
+///
+/// What it costs: the lock is global, not per-repo, so the tail of every
+/// repo write — INSERT plus COMMIT — is serialised across all accounts.
+/// That is deliberate (a per-repo lock would order each repo's events but
+/// not the shared `seq` a consumer paginates on), and it bounds write
+/// throughput to how fast Postgres can commit one small INSERT at a time.
+/// If that ever becomes the ceiling, the fix is a different sequence
+/// design — a per-repo cursor, or handing out `seq` from a single writer
+/// task — not a weaker lock: a gap in `seq` is silent data loss for every
+/// consumer replaying from a cursor.
+pub const FIREHOSE_ADVISORY_LOCK_KEY: i64 = 0x66_68_6f_73_65;
+
+// -- ops --------------------------------------------------------------------
+
+/// What a single repo operation did to one record.
+///
+/// `Update` exists separately from `Create` because a consumer that keeps a
+/// materialised view needs to know whether to insert or replace; the MST
+/// itself does not distinguish them, so the write path resolves it by looking
+/// the key up before writing.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum RepoOpAction {
+    Create,
+    Update,
+    Delete,
+}
+
+impl RepoOpAction {
+    pub fn as_str(self) -> &'static str {
+        match self {
+            RepoOpAction::Create => "create",
+            RepoOpAction::Update => "update",
+            RepoOpAction::Delete => "delete",
+        }
+    }
+
+    pub fn parse(s: &str) -> Result<Self> {
+        match s {
+            "create" => Ok(RepoOpAction::Create),
+            "update" => Ok(RepoOpAction::Update),
+            "delete" => Ok(RepoOpAction::Delete),
+            other => Err(anyhow!("unknown repo op action `{other}`")),
+        }
+    }
+}
+
+/// One entry of a commit frame's `ops` array.
+///
+/// `path` is `"<collection>/<rkey>"` — the MST key, not an `at://` URI. That
+/// is what the wire format specifies, and it is also what the MST is actually
+/// keyed by, so there is no reassembly step on either side.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct RepoOp {
+    pub action: RepoOpAction,
+    pub path: String,
+    /// The record value's CID for `create` / `update`; `None` for `delete`,
+    /// where there is no resulting value to point at.
+    pub cid: Option<Cid>,
+}
+
+impl RepoOp {
+    pub fn create(collection: &str, rkey: &str, cid: Cid) -> Self {
+        Self {
+            action: RepoOpAction::Create,
+            path: format!("{collection}/{rkey}"),
+            cid: Some(cid),
+        }
+    }
+
+    pub fn update(collection: &str, rkey: &str, cid: Cid) -> Self {
+        Self {
+            action: RepoOpAction::Update,
+            path: format!("{collection}/{rkey}"),
+            cid: Some(cid),
+        }
+    }
+
+    pub fn delete(collection: &str, rkey: &str) -> Self {
+        Self {
+            action: RepoOpAction::Delete,
+            path: format!("{collection}/{rkey}"),
+            cid: None,
+        }
+    }
+
+    /// Pick `create` or `update` from whether the key already existed.
+    pub fn put(collection: &str, rkey: &str, cid: Cid, existed: bool) -> Self {
+        if existed {
+            Self::update(collection, rkey, cid)
+        } else {
+            Self::create(collection, rkey, cid)
+        }
+    }
+
+    /// JSON shape stored in `firehose_events.ops`. The CID is a string here
+    /// (JSONB has no link type); it becomes a tag-42 link again on the wire.
+    pub fn to_json(&self) -> serde_json::Value {
+        json!({
+            "action": self.action.as_str(),
+            "path": self.path,
+            "cid": self.cid.map(|c| c.to_string()),
+        })
+    }
+
+    pub fn from_json(v: &serde_json::Value) -> Result<Self> {
+        let action = RepoOpAction::parse(
+            v.get("action")
+                .and_then(|a| a.as_str())
+                .ok_or_else(|| anyhow!("op missing `action`"))?,
+        )?;
+        let path = v
+            .get("path")
+            .and_then(|p| p.as_str())
+            .ok_or_else(|| anyhow!("op missing `path`"))?
+            .to_string();
+        let cid = match v.get("cid") {
+            None | Some(serde_json::Value::Null) => None,
+            Some(serde_json::Value::String(s)) => {
+                Some(s.parse::<Cid>().map_err(|e| anyhow!("op cid: {e}"))?)
+            }
+            Some(other) => return Err(anyhow!("op `cid` must be a string or null, got {other}")),
+        };
+        Ok(Self { action, path, cid })
+    }
+
+    fn to_dag_cbor(&self) -> Value {
+        Value::map([
+            ("action", Value::text(self.action.as_str())),
+            ("path", Value::text(&self.path)),
+            (
+                "cid",
+                match self.cid {
+                    Some(c) => Value::Link(c),
+                    None => Value::Null,
+                },
+            ),
+        ])
+    }
+}
+
+// -- events -----------------------------------------------------------------
+
+/// One row of `firehose_events`, ready to be framed.
+///
+/// Both the live path (built at write time) and the replay path (read back
+/// from Postgres) produce this exact struct, which is what makes a replayed
+/// frame byte-identical to the live one — including `time`, which comes from
+/// the stored `created_at` rather than from the clock at send time.
+#[derive(Debug, Clone)]
+pub struct FirehoseEvent {
+    pub seq: i64,
+    pub did: String,
+    pub rev: String,
+    pub since: Option<String>,
+    pub commit: Cid,
+    /// CAR v1 file: the commit block as root, plus the blocks this commit
+    /// newly created.
+    pub blocks: Vec<u8>,
+    pub ops: Vec<RepoOp>,
+    pub time: DateTime<Utc>,
+}
+
+impl FirehoseEvent {
+    /// The `#commit` body as a DAG-CBOR value.
+    pub fn to_body(&self) -> Value {
+        Value::map([
+            ("seq", Value::Int(self.seq)),
+            ("rebase", Value::Bool(false)),
+            ("tooBig", Value::Bool(false)),
+            ("repo", Value::text(&self.did)),
+            ("commit", Value::Link(self.commit)),
+            ("rev", Value::text(&self.rev)),
+            (
+                "since",
+                match &self.since {
+                    Some(s) => Value::text(s),
+                    None => Value::Null,
+                },
+            ),
+            ("blocks", Value::Bytes(self.blocks.clone())),
+            (
+                "ops",
+                Value::Array(self.ops.iter().map(RepoOp::to_dag_cbor).collect()),
+            ),
+            ("blobs", Value::Array(Vec::new())),
+            (
+                "time",
+                Value::text(self.time.to_rfc3339_opts(SecondsFormat::Micros, true)),
+            ),
+        ])
+    }
+
+    /// The full binary WebSocket payload: `#commit` header then body.
+    pub fn to_frame(&self) -> Vec<u8> {
+        frame(&header_value("#commit"), &self.to_body())
+    }
+}
+
+/// Build the CAR that goes into a commit event's `blocks` field.
+///
+/// Root is the commit block; the remaining entries are the blocks this commit
+/// newly wrote (MST nodes and record values). Blocks that already existed in
+/// the repo are deliberately left out — that is the whole point of a diff
+/// stream, and a consumer that needs an ancestor block asks
+/// `com.atproto.sync.getBlocks` for it.
+pub fn build_blocks_car(
+    commit_cid: Cid,
+    commit_block: &[u8],
+    new_blocks: &[(Cid, Vec<u8>)],
+) -> Vec<u8> {
+    let mut w = CarWriter::new();
+    w.append(commit_cid, commit_block);
+    for (cid, data) in new_blocks {
+        w.append(*cid, data);
+    }
+    w.finish(&[commit_cid])
+}
+
+// -- frame encoding ---------------------------------------------------------
+
+/// `{"op": 1, "t": "<t>"}` — the header of a regular frame.
+fn header_value(t: &str) -> Value {
+    Value::map([("op", Value::Int(1)), ("t", Value::text(t))])
+}
+
+/// Concatenate a header and a body into one binary WebSocket payload.
+fn frame(header: &Value, body: &Value) -> Vec<u8> {
+    let mut out = Vec::new();
+    encode_into(&mut out, header);
+    encode_into(&mut out, body);
+    out
+}
+
+/// An `#info` frame: `{"op":1,"t":"#info"}` + `{"name":…,"message":…}`.
+///
+/// Informational, not fatal — the stream continues after it. We send it when
+/// a cursor is older than anything we still have, and when a live subscriber
+/// lagged and is being put back on the database replay.
+pub fn encode_info_frame(name: &str, message: &str) -> Vec<u8> {
+    frame(
+        &header_value("#info"),
+        &Value::map([("name", Value::text(name)), ("message", Value::text(message))]),
+    )
+}
+
+/// An error frame: `{"op":-1}` + `{"error":…,"message":…}`.
+///
+/// Terminal — the server closes the socket right after sending it.
+pub fn encode_error_frame(error: &str, message: &str) -> Vec<u8> {
+    frame(
+        &Value::map([("op", Value::Int(-1))]),
+        &Value::map([
+            ("error", Value::text(error)),
+            ("message", Value::text(message)),
+        ]),
+    )
+}
+
+// -- broadcast --------------------------------------------------------------
+
+/// The in-process fan-out from the write path to connected subscribers.
+///
+/// Cloneable and cheap: it is a `broadcast::Sender` plus nothing. Events are
+/// wrapped in an `Arc` so a burst of subscribers does not multiply the CAR
+/// blobs.
+#[derive(Clone)]
+pub struct Firehose {
+    tx: broadcast::Sender<Arc<FirehoseEvent>>,
+}
+
+impl Default for Firehose {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl Firehose {
+    pub fn new() -> Self {
+        let (tx, _rx) = broadcast::channel(FIREHOSE_CHANNEL_CAPACITY);
+        Self { tx }
+    }
+
+    /// Publish an event to every live subscriber.
+    ///
+    /// Never blocks and never fails in a way the caller must handle: with no
+    /// subscribers the send returns `Err`, which is the normal state of a PDS
+    /// nobody is watching. The event is already durable in Postgres by the
+    /// time we get here, so a dropped broadcast costs a consumer nothing
+    /// beyond having to replay by cursor.
+    pub fn publish(&self, event: FirehoseEvent) {
+        let _ = self.tx.send(Arc::new(event));
+    }
+
+    pub fn subscribe(&self) -> broadcast::Receiver<Arc<FirehoseEvent>> {
+        self.tx.subscribe()
+    }
+
+    /// Number of live subscribers. Used for logging / the healthz surface.
+    #[allow(dead_code)]
+    pub fn subscriber_count(&self) -> usize {
+        self.tx.receiver_count()
+    }
+}
+
+// -- persistence ------------------------------------------------------------
+
+/// Append one event inside an open transaction and return it with its
+/// assigned `seq` and `created_at`.
+///
+/// The advisory lock taken first is what makes the sequence usable as a
+/// cursor: without it two concurrent writers can be assigned seq 5 and 6 and
+/// commit in the other order, so a reader polling in between sees 6, records
+/// it as its cursor, and never learns about 5. Holding
+/// `pg_advisory_xact_lock` from just before the INSERT until COMMIT forces
+/// commit order to match seq order. It is taken *after* the per-repo
+/// `SELECT … FOR UPDATE` in [`crate::routes::helpers::apply_repo_write`], and
+/// every writer takes the two in that same order, so the pair cannot deadlock.
+pub async fn insert_event_in_tx(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    did: &str,
+    rev: &str,
+    since: Option<&str>,
+    commit: Cid,
+    blocks: Vec<u8>,
+    ops: Vec<RepoOp>,
+) -> Result<FirehoseEvent, sqlx::Error> {
+    sqlx::query("SELECT pg_advisory_xact_lock($1)")
+        .bind(FIREHOSE_ADVISORY_LOCK_KEY)
+        .execute(&mut **tx)
+        .await?;
+
+    let ops_json = serde_json::Value::Array(ops.iter().map(RepoOp::to_json).collect());
+
+    let (seq, created_at): (i64, DateTime<Utc>) = sqlx::query_as(
+        r#"INSERT INTO firehose_events (did, rev, since, commit_cid, blocks, ops)
+           VALUES ($1, $2, $3, $4, $5, $6)
+           RETURNING seq, created_at"#,
+    )
+    .bind(did)
+    .bind(rev)
+    .bind(since)
+    .bind(commit.to_bytes())
+    .bind(&blocks)
+    .bind(&ops_json)
+    .fetch_one(&mut **tx)
+    .await?;
+
+    Ok(FirehoseEvent {
+        seq,
+        did: did.to_string(),
+        rev: rev.to_string(),
+        since: since.map(|s| s.to_string()),
+        commit,
+        blocks,
+        ops,
+        time: created_at,
+    })
+}
+
+/// The `(min_seq, max_seq)` currently in the table, or `None` when it is
+/// empty. Used by the cursor handshake to tell "from the future" apart from
+/// "too old to still have".
+pub async fn seq_bounds(db: &sqlx::PgPool) -> Result<Option<(i64, i64)>, sqlx::Error> {
+    let row: (Option<i64>, Option<i64>) =
+        sqlx::query_as("SELECT MIN(seq), MAX(seq) FROM firehose_events")
+            .fetch_one(db)
+            .await?;
+    Ok(match row {
+        (Some(min), Some(max)) => Some((min, max)),
+        _ => None,
+    })
+}
+
+/// Read up to [`REPLAY_PAGE_SIZE`] events with `seq > after`, oldest first.
+pub async fn load_events_after(
+    db: &sqlx::PgPool,
+    after: i64,
+    limit: i64,
+) -> Result<Vec<FirehoseEvent>> {
+    let rows: Vec<(
+        i64,
+        String,
+        String,
+        Option<String>,
+        Vec<u8>,
+        Vec<u8>,
+        serde_json::Value,
+        DateTime<Utc>,
+    )> = sqlx::query_as(
+        r#"SELECT seq, did, rev, since, commit_cid, blocks, ops, created_at
+             FROM firehose_events
+            WHERE seq > $1
+            ORDER BY seq ASC
+            LIMIT $2"#,
+    )
+    .bind(after)
+    .bind(limit)
+    .fetch_all(db)
+    .await
+    .map_err(|e| anyhow!("firehose_events replay read: {e}"))?;
+
+    rows.into_iter()
+        .map(|(seq, did, rev, since, commit_cid, blocks, ops, created_at)| {
+            let commit = cid_from_multihash_bytes(&commit_cid)
+                .map_err(|e| anyhow!("firehose_events.commit_cid at seq {seq}: {e}"))?;
+            let ops = ops
+                .as_array()
+                .ok_or_else(|| anyhow!("firehose_events.ops at seq {seq} is not an array"))?
+                .iter()
+                .map(RepoOp::from_json)
+                .collect::<Result<Vec<_>>>()?;
+            Ok(FirehoseEvent {
+                seq,
+                did,
+                rev,
+                since,
+                commit,
+                blocks,
+                ops,
+                time: created_at,
+            })
+        })
+        .collect()
+}
+
+// -- cursor handshake -------------------------------------------------------
+
+/// What the connection handler should do with the cursor the client sent.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum CursorPlan {
+    /// No cursor: send live events only, nothing from the log.
+    LiveOnly,
+    /// Replay everything after `from`, then go live.
+    Replay { from: i64 },
+    /// The requested cursor predates the oldest row we still have. Warn with
+    /// an `#info` frame, then replay from `from` (the oldest surviving row
+    /// minus one) so the client at least gets everything that does exist.
+    OutdatedCursor { from: i64, message: String },
+    /// The cursor names an event that has not happened. This is a client bug
+    /// (or a cursor from a different server's log), and continuing would
+    /// silently strand it — so it is a terminal error frame.
+    FutureCursor { message: String },
+}
+
+/// Decide what to do with `cursor` given the log's current `(min, max)`.
+///
+/// Split out from the socket handler so the boundary conditions are testable
+/// without a database or a WebSocket.
+///
+/// Semantics of the cursor: it is the seq of the last event the client
+/// *already has*, so a replay yields `seq > cursor`. `cursor = 0` therefore
+/// means "everything", and `cursor = max` means "nothing yet, just go live" —
+/// which is a valid, empty replay rather than a future cursor.
+pub fn plan_cursor(cursor: Option<i64>, bounds: Option<(i64, i64)>) -> CursorPlan {
+    let cursor = match cursor {
+        None => return CursorPlan::LiveOnly,
+        Some(c) => c,
+    };
+    if cursor < 0 {
+        return CursorPlan::FutureCursor {
+            message: format!("cursor {cursor} is negative"),
+        };
+    }
+    let (min, max) = match bounds {
+        // An empty log accepts only cursor 0 ("give me everything, there is
+        // nothing"). Anything else refers to an event we never had.
+        None => {
+            return if cursor == 0 {
+                CursorPlan::Replay { from: 0 }
+            } else {
+                CursorPlan::FutureCursor {
+                    message: format!("cursor {cursor} is ahead of an empty log"),
+                }
+            };
+        }
+        Some(b) => b,
+    };
+    if cursor > max {
+        return CursorPlan::FutureCursor {
+            message: format!("cursor {cursor} is ahead of the latest event {max}"),
+        };
+    }
+    // `cursor >= min - 1` means the next event the client wants (cursor + 1)
+    // is still on disk. Below that, rows have been pruned and the client has
+    // a hole it can never fill.
+    if cursor < min - 1 {
+        return CursorPlan::OutdatedCursor {
+            from: min - 1,
+            message: format!(
+                "cursor {cursor} predates the oldest retained event {min}; \
+                 resuming from {min} — events {}..{} are gone",
+                cursor + 1,
+                min - 1
+            ),
+        };
+    }
+    CursorPlan::Replay { from: cursor }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::dag_cbor::{decode, decode_one};
+    use at_crypto::cid::cid_for_cbor;
+
+    fn sample_event() -> FirehoseEvent {
+        let commit = cid_for_cbor(b"commit block").unwrap();
+        let value = cid_for_cbor(b"record value").unwrap();
+        FirehoseEvent {
+            seq: 7,
+            did: "did:plc:alice".into(),
+            rev: "3lxxxxxxxx2".into(),
+            since: Some("3lxxxxxxxx1".into()),
+            commit,
+            blocks: build_blocks_car(
+                commit,
+                b"commit block",
+                &[(value, b"record value".to_vec())],
+            ),
+            ops: vec![RepoOp::create("app.twi.post", "3lrkey", value)],
+            time: DateTime::parse_from_rfc3339("2026-09-10T12:00:00.123456Z")
+                .unwrap()
+                .with_timezone(&Utc),
+        }
+    }
+
+    // -- frame encoding ----------------------------------------------------
+
+    #[test]
+    fn commit_frame_header_then_body() {
+        let ev = sample_event();
+        let bytes = ev.to_frame();
+        let (header, used) = decode_one(&bytes).unwrap();
+        assert_eq!(header.get("op").and_then(Value::as_i64), Some(1));
+        assert_eq!(header.get("t").and_then(Value::as_str), Some("#commit"));
+        // Everything after the header is exactly one more value — no padding,
+        // no length prefix.
+        let body = decode(&bytes[used..]).unwrap();
+        assert_eq!(body.get("seq").and_then(Value::as_i64), Some(7));
+        assert_eq!(
+            body.get("repo").and_then(Value::as_str),
+            Some("did:plc:alice")
+        );
+        assert_eq!(body.get("rev").and_then(Value::as_str), Some("3lxxxxxxxx2"));
+        assert_eq!(
+            body.get("since").and_then(Value::as_str),
+            Some("3lxxxxxxxx1")
+        );
+        assert_eq!(body.get("rebase").and_then(Value::as_bool), Some(false));
+        assert_eq!(body.get("tooBig").and_then(Value::as_bool), Some(false));
+        assert!(body.get("blobs").unwrap().as_array().unwrap().is_empty());
+        assert_eq!(
+            body.get("time").and_then(Value::as_str),
+            Some("2026-09-10T12:00:00.123456Z")
+        );
+    }
+
+    /// The header's exact bytes, locked in. The AppView builds its reader
+    /// against this, and canonical key ordering means `t` precedes `op` —
+    /// which is easy to get wrong and produces a frame a strict DAG-CBOR
+    /// decoder rejects.
+    #[test]
+    fn commit_header_has_the_exact_expected_bytes() {
+        let bytes = sample_event().to_frame();
+        assert_eq!(
+            &bytes[..15],
+            &[
+                0xA2, // map(2)
+                0x61, b't', // text(1) "t"
+                0x67, b'#', b'c', b'o', b'm', b'm', b'i', b't', // text(7) "#commit"
+                0x62, b'o', b'p', // text(2) "op"
+                0x01, // 1
+            ]
+        );
+    }
+
+    #[test]
+    fn info_and_error_headers_have_the_exact_expected_bytes() {
+        let info = encode_info_frame("OutdatedCursor", "x");
+        assert_eq!(
+            &info[..12],
+            &[0xA2, 0x61, b't', 0x65, b'#', b'i', b'n', b'f', b'o', 0x62, b'o', b'p']
+        );
+        assert_eq!(info[12], 0x01);
+        // Error: map(1) { "op": -1 }. -1 is major type 1 with argument 0.
+        let err = encode_error_frame("FutureCursor", "x");
+        assert_eq!(&err[..5], &[0xA1, 0x62, b'o', b'p', 0x20]);
+    }
+
+    #[test]
+    fn commit_cid_is_a_tag_42_link_not_a_string() {
+        // The whole point of the hand-rolled encoder: `ciborium` would have
+        // written this as a string or a newtype struct.
+        let ev = sample_event();
+        let bytes = ev.to_frame();
+        let (_h, used) = decode_one(&bytes).unwrap();
+        let body = decode(&bytes[used..]).unwrap();
+        assert_eq!(body.get("commit").and_then(Value::as_link), Some(&ev.commit));
+        // Locate the tag bytes directly, to prove it is not the decoder being
+        // generous: 0xD8 0x2A is tag(42).
+        assert!(
+            bytes.windows(2).any(|w| w == [0xD8, 0x2A]),
+            "frame must contain a tag-42 head"
+        );
+    }
+
+    #[test]
+    fn ops_encode_action_path_and_link() {
+        let ev = sample_event();
+        let bytes = ev.to_frame();
+        let (_h, used) = decode_one(&bytes).unwrap();
+        let body = decode(&bytes[used..]).unwrap();
+        let ops = body.get("ops").unwrap().as_array().unwrap();
+        assert_eq!(ops.len(), 1);
+        assert_eq!(ops[0].get("action").and_then(Value::as_str), Some("create"));
+        assert_eq!(
+            ops[0].get("path").and_then(Value::as_str),
+            Some("app.twi.post/3lrkey")
+        );
+        assert!(ops[0].get("cid").unwrap().as_link().is_some());
+    }
+
+    #[test]
+    fn delete_op_has_a_null_cid() {
+        let mut ev = sample_event();
+        ev.ops = vec![RepoOp::delete("app.bsky.feed.like", "3lrkey")];
+        let bytes = ev.to_frame();
+        let (_h, used) = decode_one(&bytes).unwrap();
+        let body = decode(&bytes[used..]).unwrap();
+        let ops = body.get("ops").unwrap().as_array().unwrap();
+        assert_eq!(ops[0].get("action").and_then(Value::as_str), Some("delete"));
+        assert!(ops[0].get("cid").unwrap().is_null());
+    }
+
+    #[test]
+    fn first_commit_has_a_null_since() {
+        let mut ev = sample_event();
+        ev.since = None;
+        let bytes = ev.to_frame();
+        let (_h, used) = decode_one(&bytes).unwrap();
+        let body = decode(&bytes[used..]).unwrap();
+        assert!(body.get("since").unwrap().is_null());
+    }
+
+    #[test]
+    fn blocks_is_a_parsable_car_rooted_at_the_commit() {
+        let ev = sample_event();
+        let bytes = ev.to_frame();
+        let (_h, used) = decode_one(&bytes).unwrap();
+        let body = decode(&bytes[used..]).unwrap();
+        let car = body.get("blocks").and_then(Value::as_bytes).unwrap();
+        let (header, blocks) = crate::car::parse(car).unwrap();
+        assert_eq!(header.version, 1);
+        assert_eq!(header.roots, vec![ev.commit]);
+        assert_eq!(blocks.len(), 2, "commit block + one new record block");
+        assert_eq!(blocks[0].cid, ev.commit);
+        assert_eq!(blocks[0].data, b"commit block");
+    }
+
+    #[test]
+    fn info_frame_round_trips() {
+        let bytes = encode_info_frame("OutdatedCursor", "resuming from 12");
+        let (header, used) = decode_one(&bytes).unwrap();
+        assert_eq!(header.get("op").and_then(Value::as_i64), Some(1));
+        assert_eq!(header.get("t").and_then(Value::as_str), Some("#info"));
+        let body = decode(&bytes[used..]).unwrap();
+        assert_eq!(
+            body.get("name").and_then(Value::as_str),
+            Some("OutdatedCursor")
+        );
+        assert_eq!(
+            body.get("message").and_then(Value::as_str),
+            Some("resuming from 12")
+        );
+    }
+
+    #[test]
+    fn error_frame_uses_op_minus_one_and_carries_no_t() {
+        let bytes = encode_error_frame("FutureCursor", "cursor 99 is ahead");
+        let (header, used) = decode_one(&bytes).unwrap();
+        assert_eq!(header.get("op").and_then(Value::as_i64), Some(-1));
+        assert!(
+            header.get("t").is_none(),
+            "an error header carries op only"
+        );
+        let body = decode(&bytes[used..]).unwrap();
+        assert_eq!(
+            body.get("error").and_then(Value::as_str),
+            Some("FutureCursor")
+        );
+        assert_eq!(
+            body.get("message").and_then(Value::as_str),
+            Some("cursor 99 is ahead")
+        );
+    }
+
+    // -- ops derivation ----------------------------------------------------
+
+    #[test]
+    fn put_picks_create_or_update_from_prior_existence() {
+        let cid = cid_for_cbor(b"v").unwrap();
+        assert_eq!(
+            RepoOp::put("c", "r", cid, false).action,
+            RepoOpAction::Create
+        );
+        assert_eq!(
+            RepoOp::put("c", "r", cid, true).action,
+            RepoOpAction::Update
+        );
+    }
+
+    #[test]
+    fn op_path_is_collection_slash_rkey() {
+        let cid = cid_for_cbor(b"v").unwrap();
+        assert_eq!(
+            RepoOp::create("app.bsky.feed.repost", "3lk", cid).path,
+            "app.bsky.feed.repost/3lk"
+        );
+    }
+
+    #[test]
+    fn op_json_round_trips_through_the_jsonb_shape() {
+        let cid = cid_for_cbor(b"v").unwrap();
+        for op in [
+            RepoOp::create("app.twi.post", "a", cid),
+            RepoOp::update("app.bsky.actor.profile", "self", cid),
+            RepoOp::delete("app.bsky.graph.follow", "b"),
+        ] {
+            let back = RepoOp::from_json(&op.to_json()).unwrap();
+            assert_eq!(back, op);
+        }
+    }
+
+    #[test]
+    fn op_json_rejects_an_unknown_action() {
+        let v = json!({"action": "rebase", "path": "a/b", "cid": null});
+        assert!(RepoOp::from_json(&v).is_err());
+    }
+
+    // -- cursor edge cases -------------------------------------------------
+
+    #[test]
+    fn no_cursor_is_live_only() {
+        assert_eq!(plan_cursor(None, Some((1, 10))), CursorPlan::LiveOnly);
+        assert_eq!(plan_cursor(None, None), CursorPlan::LiveOnly);
+    }
+
+    #[test]
+    fn cursor_zero_replays_everything() {
+        assert_eq!(
+            plan_cursor(Some(0), Some((1, 10))),
+            CursorPlan::Replay { from: 0 }
+        );
+    }
+
+    #[test]
+    fn cursor_at_the_head_is_an_empty_replay_not_an_error() {
+        // The client is fully caught up. Replaying `seq > 10` yields nothing
+        // and it goes straight live — that must not be a FutureCursor.
+        assert_eq!(
+            plan_cursor(Some(10), Some((1, 10))),
+            CursorPlan::Replay { from: 10 }
+        );
+    }
+
+    #[test]
+    fn cursor_past_the_head_is_a_future_cursor() {
+        assert!(matches!(
+            plan_cursor(Some(11), Some((1, 10))),
+            CursorPlan::FutureCursor { .. }
+        ));
+    }
+
+    #[test]
+    fn negative_cursor_is_a_future_cursor() {
+        assert!(matches!(
+            plan_cursor(Some(-1), Some((1, 10))),
+            CursorPlan::FutureCursor { .. }
+        ));
+    }
+
+    #[test]
+    fn empty_log_accepts_zero_and_refuses_anything_else() {
+        assert_eq!(plan_cursor(Some(0), None), CursorPlan::Replay { from: 0 });
+        assert!(matches!(
+            plan_cursor(Some(1), None),
+            CursorPlan::FutureCursor { .. }
+        ));
+    }
+
+    #[test]
+    fn cursor_exactly_one_below_the_oldest_row_is_still_exact() {
+        // min = 5 means seq 5 is the oldest surviving event. A client whose
+        // cursor is 4 wants 5 next — nothing is missing.
+        assert_eq!(
+            plan_cursor(Some(4), Some((5, 10))),
+            CursorPlan::Replay { from: 4 }
+        );
+    }
+
+    #[test]
+    fn cursor_below_the_pruned_window_is_outdated() {
+        match plan_cursor(Some(2), Some((5, 10))) {
+            CursorPlan::OutdatedCursor { from, message } => {
+                assert_eq!(from, 4, "resume so the next delivered event is 5");
+                assert!(message.contains('5'), "message should name the gap: {message}");
+            }
+            other => panic!("expected OutdatedCursor, got {other:?}"),
+        }
+    }
+}

crates/pds-server/src/main.rs

@@ -1,5 +1,7 @@
 mod appview_push;
 mod car;
+mod dag_cbor;
+mod firehose;
 mod jwt_issuer;
 mod keys;
 mod password;
@@ -146,6 +148,13 @@ pub fn router(state: AppState) -> Router {
             "/xrpc/com.atproto.sync.getBlob",
             get(routes::blob::get_blob),
         )
+        // The firehose. A WebSocket upgrade arrives as a plain GET, so this
+        // is a normal `get` route whose handler happens to return an
+        // upgrade response.
+        .route(
+            "/xrpc/com.atproto.sync.subscribeRepos",
+            get(routes::subscribe_repos::subscribe_repos),
+        )
         .route(
             "/blob/:cid",
             get(routes::blob::get_blob_by_cid),

crates/pds-server/src/routes/feed.rs

@@ -17,7 +17,10 @@
 //! removed from the MST, a new commit is signed, the AppView is
 //! told to drop the row, and we return the new commit CID + rev.
 
-use crate::routes::helpers::{apply_repo_write, err, lookup_handle, to_sqlx_error, RepoWriteOutcome};
+use crate::firehose::RepoOp;
+use crate::routes::helpers::{
+    apply_repo_write, err, lookup_handle, to_sqlx_error, RepoWriteOutcome, RepoWriteResult,
+};
 use at_repo::blockstore::Blockstore;
 use crate::routes::types::ErrorBody;
 use crate::state::AppState;
@@ -189,11 +192,16 @@ fn build_like_record(req: &CreateLikeReq) -> Result<Value, (StatusCode, Json<Err
 /// Postgres transaction with `SELECT … FOR UPDATE` on the user's
 /// `repos` row. Concurrent writers for the same DID now serialise
 /// behind the row lock instead of clobbering each other.)
+///
+/// Returns the full [`RepoWriteResult`] rather than just the commit: the
+/// firehose event that went into the same transaction carries the sequence
+/// number, which the handlers log so an operator can line a write up against
+/// what a subscriber received.
 async fn apply_and_commit<F>(
     state: &AppState,
     did: &str,
     f: F,
-) -> Result<at_repo::commit::Commit, (StatusCode, Json<ErrorBody>)>
+) -> Result<RepoWriteResult, (StatusCode, Json<ErrorBody>)>
 where
     F: for<'b> FnOnce(
         &'b mut at_repo::repo::Repo<at_repo::blockstore::MemoryBlockstore>,
@@ -201,7 +209,7 @@ where
         Box<dyn std::future::Future<Output = Result<RepoWriteOutcome, sqlx::Error>> + Send + 'b>,
     >,
 {
-    apply_repo_write(state, did, f).await.map(|o| o.commit)
+    apply_repo_write(state, did, f).await
 }
 
 // -- handlers ---------------------------------------------------------------
@@ -275,7 +283,7 @@ pub async fn create_like(
     let push_rkey = rkey.clone();
     let push_handle_str: Option<String> = lookup_handle(&state, &did).await;
 
-    let commit = apply_and_commit(&state, &did, move |repo| {
+    let write = apply_and_commit(&state, &did, move |repo| {
         let value_cid = value_cid;
         let rkey = rkey;
         let record_buf = record_buf;
@@ -297,16 +305,21 @@ pub async fn create_like(
                 commit,
                 head_cid_bytes,
                 head_commit_bytes,
+                // Always a create: the rkey is a freshly minted TID, so it
+                // cannot collide with an existing entry.
+                ops: vec![RepoOp::create(LIKE_COLLECTION, &rkey, value_cid)],
             })
         })
     })
     .await?;
 
+    let commit = write.commit;
     info!(
         collection = LIKE_COLLECTION,
         rkey = %push_rkey,
         cid = %value_cid,
         commit = %commit.cid,
+        seq = write.event.seq,
         "like created"
     );
 
@@ -366,10 +379,21 @@ pub async fn delete_record(
     // `Repo::delete_record` is idempotent at the MST level (returns
     // an unchanged tree if the key isn't present), so we always
     // sign a new commit — the spec says 200 on a no-op delete.
-    let commit = apply_and_commit(&state, &did, move |repo| {
+    let write = apply_and_commit(&state, &did, move |repo| {
         let collection = collection;
         let rkey = rkey;
         Box::pin(async move {
+            // Report the op only when there was something to remove.
+            // `delete_record` is idempotent — deleting a missing rkey signs
+            // an unchanged tree — and announcing a delete for a record that
+            // never existed would make a consumer drop a row it may legitimately
+            // hold under a different rkey, or (worse) log a phantom deletion
+            // on every retry of a duplicate unlike.
+            let existed = repo
+                .get_record(&collection, &rkey)
+                .await
+                .map_err(to_sqlx_error)?
+                .is_some();
             repo.delete_record(&collection, &rkey)
                 .await
                 .map_err(to_sqlx_error)?;
@@ -380,15 +404,22 @@ pub async fn delete_record(
                 commit,
                 head_cid_bytes,
                 head_commit_bytes,
+                ops: if existed {
+                    vec![RepoOp::delete(&collection, &rkey)]
+                } else {
+                    Vec::new()
+                },
             })
         })
     })
     .await?;
 
+    let commit = write.commit;
     info!(
         collection = %push_collection,
         rkey = %push_rkey,
         commit = %commit.cid,
+        seq = write.event.seq,
         "record deleted"
     );
 

crates/pds-server/src/routes/helpers.rs

@@ -12,6 +12,7 @@
 //!   writers for the same DID can't trample each other's MST updates
 //!   (Phase 5b review C1).
 
+use crate::firehose::{self, FirehoseEvent, RepoOp};
 use crate::routes::types::ErrorBody;
 use crate::state::AppState;
 use at_crypto::cid::cid_from_multihash_bytes;
@@ -24,6 +25,7 @@ use cid::Cid;
 use k256::ecdsa::SigningKey;
 use k256::SecretKey;
 use sqlx::Postgres;
+use std::collections::HashSet;
 use std::sync::Arc;
 
 /// Load every block belonging to `did` from the `repo_blocks` table into a
@@ -184,14 +186,31 @@ pub fn to_sqlx_error(e: anyhow::Error) -> sqlx::Error {
 // transaction commits or rolls back, so concurrent writers serialise
 // behind the holder rather than racing on the head_commit column.
 
-/// Result of a successful repo write: the new signed commit, the CID
+/// What the closure handed to [`apply_repo_write`] returns: the new signed
-/// pointing at the freshly-written head block, and the new revision
+/// commit, the CID pointing at the freshly-written head block, and the record
-/// string. Callers use the commit for AppView ingest pushes.
+/// operations the closure performed.
+///
+/// `ops` is not derivable from the commit — the MST stores the resulting
+/// tree, not the edit that produced it, and it cannot tell a create from an
+/// update at all. Only the closure knows what it did, so it says so, and the
+/// firehose event is built from that. Every write path must fill this in
+/// truthfully: an empty `ops` produces a commit frame that tells the AppView
+/// "something changed, guess what".
 #[derive(Debug, Clone)]
 pub struct RepoWriteOutcome {
     pub commit: at_repo::commit::Commit,
     pub head_cid_bytes: Vec<u8>,
     pub head_commit_bytes: Vec<u8>,
+    pub ops: Vec<RepoOp>,
+}
+
+/// What [`apply_repo_write`] returns to the route handler: the commit (used
+/// for the response body and the AppView push) plus the firehose event that
+/// was appended in the same transaction and has already been broadcast.
+#[derive(Debug, Clone)]
+pub struct RepoWriteResult {
+    pub commit: at_repo::commit::Commit,
+    pub event: FirehoseEvent,
 }
 
 /// Apply a write to the user's repo under a row-level lock on the
@@ -218,11 +237,27 @@ pub struct RepoWriteOutcome {
 /// write the row), but the transaction stays open until after the
 /// `UPDATE`. If the closure or `UPDATE` fails, the transaction rolls
 /// back and no head pointer or block row changes are visible.
+///
+/// ## The firehose event rides in the same transaction
+///
+/// Between step 7 and the COMMIT we append one row to `firehose_events`
+/// (see [`crate::firehose`]). It has to be the *same* transaction, not a
+/// follow-up write: if the event were appended afterwards, a crash in the
+/// window between the two would leave a repo whose head has moved but whose
+/// event log never mentions it — and since a consumer's cursor only ever
+/// moves forward, that commit would be invisible to every subscriber
+/// permanently. Sharing the transaction makes "the head moved" and "an event
+/// exists for it" one atomic fact. Conversely, a rollback discards both, so
+/// no subscriber ever sees an event for a commit that did not happen.
+///
+/// The broadcast to live subscribers happens *after* `COMMIT`, for the same
+/// reason in reverse: a subscriber must never receive an event that a
+/// rollback then erases.
 pub async fn apply_repo_write<F>(
     state: &AppState,
     did: &str,
     f: F,
-) -> Result<RepoWriteOutcome, (StatusCode, Json<ErrorBody>)>
+) -> Result<RepoWriteResult, (StatusCode, Json<ErrorBody>)>
 where
     F: for<'b> FnOnce(
         &'b mut Repo<MemoryBlockstore>,
@@ -240,8 +275,12 @@ where
 
     // 2. Take the row-level write lock. Postgres parks competing
     //    transactions here until we COMMIT/ROLLBACK.
-    let head_row: Option<(Vec<u8>, Vec<u8>, Option<Vec<u8>>)> = sqlx::query_as(
+    // `rev` comes along because it is the *previous* commit's revision,
-        "SELECT head_cid, head_commit, prev_commit
+    // which the firehose frame publishes as `since` — a consumer uses it to
+    // notice that it skipped an intermediate commit. It has to be read here,
+    // under the lock, before the UPDATE below overwrites it.
+    let head_row: Option<(Vec<u8>, Vec<u8>, Option<Vec<u8>>, String)> = sqlx::query_as(
+        "SELECT head_cid, head_commit, prev_commit, rev
            FROM repos
           WHERE did = $1
           FOR UPDATE",
@@ -257,8 +296,8 @@ where
         )
     })?;
 
-    let (head_cid_blob, head_commit_blob) = match head_row {
+    let (head_cid_blob, head_commit_blob, prev_rev) = match head_row {
-        Some(r) => (r.0, r.1),
+        Some(r) => (r.0, r.1, r.3),
         None => {
             return Err((
                 StatusCode::NOT_FOUND,
@@ -354,6 +393,26 @@ where
     //    to run additional queries under the row lock, we'd extend
     //    this helper to also hand out a `&mut PgConnection` (which
     //    doesn't have the lifetime headache of `&mut Transaction`).
+    //    Snapshot the CIDs the repo already had *before* the closure runs.
+    //    Diffing against this afterwards is what tells us which blocks are
+    //    new in this commit — the firehose CAR carries only those, because a
+    //    diff stream that re-sent the whole repo on every post would be
+    //    useless. The snapshot is taken here, after the head block re-seed
+    //    above, so the existing head commit does not look new.
+    let blocks_before: HashSet<Cid> = blockstore
+        .list()
+        .await
+        .map_err(|e| {
+            err(
+                StatusCode::INTERNAL_SERVER_ERROR,
+                "InternalServerError",
+                format!("blockstore list: {e:#}"),
+            )
+        })?
+        .into_iter()
+        .map(|(cid, _)| cid)
+        .collect();
+
     let outcome: RepoWriteOutcome = f(&mut repo).await.map_err(|e| {
         err(
             StatusCode::INTERNAL_SERVER_ERROR,
@@ -409,6 +468,51 @@ where
         )
     })?;
 
+    // 8. Append the firehose event. Same transaction as everything above —
+    //    see the "rides in the same transaction" note on this function.
+    //
+    //    The CAR carries the commit block as its root plus every block that
+    //    was not in the repo when we started: the new MST nodes and the new
+    //    record value. Blocks that already existed are omitted; a consumer
+    //    that needs an ancestor asks `com.atproto.sync.getBlocks` for it.
+    let new_blocks: Vec<(Cid, Vec<u8>)> = all_blocks
+        .iter()
+        .filter(|(cid, _)| !blocks_before.contains(*cid) && **cid != outcome.commit.cid)
+        .map(|(cid, bytes)| (*cid, bytes.clone()))
+        .collect();
+    let blocks_car = firehose::build_blocks_car(
+        outcome.commit.cid,
+        &outcome.head_commit_bytes,
+        &new_blocks,
+    );
+
+    // A repo whose head was the all-zero sentinel had no previous commit, so
+    // there is no previous revision to report — `since` is null rather than
+    // the `"0"` placeholder `createAccount` seeds the row with.
+    let since: Option<&str> = if is_zero_blob(&head_cid_blob) {
+        None
+    } else {
+        Some(prev_rev.as_str())
+    };
+
+    let event = firehose::insert_event_in_tx(
+        &mut tx,
+        did,
+        &outcome.commit.rev,
+        since,
+        outcome.commit.cid,
+        blocks_car,
+        outcome.ops.clone(),
+    )
+    .await
+    .map_err(|e| {
+        err(
+            StatusCode::INTERNAL_SERVER_ERROR,
+            "InternalServerError",
+            format!("firehose_events insert: {e}"),
+        )
+    })?;
+
     tx.commit().await.map_err(|e| {
         err(
             StatusCode::INTERNAL_SERVER_ERROR,
@@ -417,7 +521,18 @@ where
         )
     })?;
 
-    Ok(outcome)
+    // 9. Only now, with the commit durable, hand the event to live
+    //    subscribers. Publishing is non-blocking and cannot fail in a way
+    //    that matters: a slow subscriber is dealt with on its own side (see
+    //    the lag policy in `crate::firehose`), and with no subscribers at all
+    //    the send is a no-op. The row is on disk either way, so nothing is
+    //    lost if this reaches nobody.
+    state.firehose.publish(event.clone());
+
+    Ok(RepoWriteResult {
+        commit: outcome.commit,
+        event,
+    })
 }
 
 /// Persist every block in `blocks` into `repo_blocks` using the open

crates/pds-server/src/routes/mod.rs

@@ -5,5 +5,6 @@ pub mod helpers;
 pub mod identity;
 pub mod profile;
 pub mod repo;
+pub mod subscribe_repos;
 pub mod sync;
 pub mod types;

crates/pds-server/src/routes/profile.rs

@@ -16,6 +16,7 @@
 //! overwrite the corresponding fields. Best-effort push to the
 //! AppView follows so the `profiles` cache reflects the new avatar /
 //! display name / bio without waiting for the Jetstream replay.
+use crate::firehose::RepoOp;
 use crate::jwt_issuer;
 use crate::routes::helpers::{
     apply_repo_write, err, load_head_commit, load_signing_key, load_user_blockstore,
@@ -98,6 +99,9 @@ pub async fn set_profile(
 
     // Fetch the existing record, if any.
     let existing = read_profile_record(&state, &did).await?;
+    // Remembered before `existing` is consumed by the merge — the firehose op
+    // needs to know whether this is the first profile write for the account.
+    let existing_present = existing.is_some();
 
     // For any blob CIDs in the request, look up the real
     // `mime_type` / `size` from the `blobs` table — and verify
@@ -152,6 +156,12 @@ pub async fn set_profile(
     };
 
     let next_for_block = next.clone();
+    // `existing` was read before the merge above: a profile record that was
+    // already there makes this an `update` on the firehose, a first-ever
+    // `setProfile` a `create`. The rkey is the fixed `self`, so this is the
+    // one write path where updates are the common case rather than the
+    // exception.
+    let profile_existed = existing_present;
     let outcome = apply_repo_write(&state, &did, move |repo| {
         let value_cid = value_cid;
         let next_for_block = next_for_block;
@@ -175,6 +185,12 @@ pub async fn set_profile(
                 commit,
                 head_cid_bytes,
                 head_commit_bytes,
+                ops: vec![RepoOp::put(
+                    "app.bsky.actor.profile",
+                    "self",
+                    value_cid,
+                    profile_existed,
+                )],
             })
         })
     })
@@ -183,6 +199,7 @@ pub async fn set_profile(
     info!(
         did = %did,
         cid = %outcome.commit.cid,
+        seq = outcome.event.seq,
         "profile record created"
     );
 

crates/pds-server/src/routes/repo.rs

@@ -1,3 +1,4 @@
+use crate::firehose::RepoOp;
 use crate::routes::helpers::{
     apply_repo_write, err, lookup_handle, to_sqlx_error, RepoWriteOutcome,
 };
@@ -110,6 +111,17 @@ pub async fn create_record(
         let record_buf = record_buf;
         let collection = collection;
         Box::pin(async move {
+            // Ask the MST whether the key is already there *before* writing.
+            // The firehose distinguishes `create` from `update` and the tree
+            // itself cannot: after `put_record` both look identical. A
+            // caller-supplied `rkey` (rather than the generated TID) is the
+            // case that actually hits this — an overwrite of an existing
+            // record must not be announced as a create.
+            let existed = repo
+                .get_record(&collection, &rkey)
+                .await
+                .map_err(to_sqlx_error)?
+                .is_some();
             // Repo assumes the value block is already in the
             // blockstore — that's the caller's responsibility.
             repo.blockstore
@@ -127,23 +139,34 @@ pub async fn create_record(
                 commit,
                 head_cid_bytes,
                 head_commit_bytes,
+                ops: vec![RepoOp::put(&collection, &rkey, value_cid, existed)],
             })
         })
     })
     .await?;
 
     let uri = format!("at://{did}/{push_coll}/{push_rkey}");
+    let seq = outcome.event.seq;
     let commit = outcome.commit;
-    info!(uri = %uri, cid = %value_cid, commit = %commit.cid, "record created");
+    info!(
+        uri = %uri,
+        cid = %value_cid,
+        commit = %commit.cid,
+        seq,
+        "record created"
+    );
 
     // 10. Best-effort push to the AppView's `/internal/ingest-commit`.
     //     We send the full record value (not just the CID) because the
     //     AppView's indexer reads `embed` and `reply` off it.
     //
     //     **Spawned** (not awaited) so a transient AppView outage never
-    //     blocks the user's write response. If the push fails, the
+    //     blocks the user's write response. Losing the push is no longer
-    //     global Jetstream feed will eventually replay the commit to
+    //     terminal: the same commit was appended to `firehose_events` in
-    //     the AppView.
+    //     the write transaction above (see `seq` in the log line), so an
+    //     AppView that reconnects to `com.atproto.sync.subscribeRepos`
+    //     with its cursor picks it up. The push is now purely a latency
+    //     optimisation, not the only delivery path.
     tokio::spawn(async move {
         if let Err(e) = push_handle
             .push_create(

crates/pds-server/src/routes/subscribe_repos.rs

@@ -0,0 +1,301 @@
+//! `GET /xrpc/com.atproto.sync.subscribeRepos` — the firehose WebSocket.
+//!
+//! The frame format, the deviation from the atproto spec, the lag policy and
+//! the retention story all live in the module header of [`crate::firehose`];
+//! this file is only the socket.
+//!
+//! ## The handshake, and why it is ordered the way it is
+//!
+//! ```text
+//!   1. subscribe to the live broadcast          <-- BEFORE any DB read
+//!   2. read (min, max) from firehose_events
+//!   3. decide what the cursor means             <-- firehose::plan_cursor
+//!   4. drain the DB replay, remembering the highest seq sent
+//!   5. forward live events with seq > that high-water mark
+//! ```
+//!
+//! Step 1 has to come first. If we read the database and *then* subscribed,
+//! an event committed in between would be in neither: too late for the replay
+//! query, too early for the receiver. Subscribing first turns that race into
+//! a duplicate instead of a gap — the event is both replayed from the table
+//! and sitting in the channel — and a duplicate is something we can filter,
+//! which is what the high-water mark in step 5 does.
+//!
+//! The filter is exact rather than approximate because the write path
+//! serialises `firehose_events` inserts under an advisory lock (see
+//! [`crate::firehose::insert_event_in_tx`]): if the replay query saw seq `N`,
+//! then every seq below `N` is already committed and was also seen. So
+//! "everything the replay covered" is precisely "seq <= N", and every event
+//! that arrives on the channel afterwards has seq > N. No gap, no duplicate,
+//! at the handover.
+//!
+//! Without a cursor there is no replay at all and the high-water mark stays
+//! at zero: a `broadcast::Receiver` only ever yields messages sent after it
+//! was created, so "live only" needs no filtering.
+
+use crate::firehose::{
+    self, CursorPlan, FirehoseEvent, MAX_LAG_RECOVERIES, REPLAY_PAGE_SIZE,
+};
+use crate::state::AppState;
+use axum::extract::ws::{CloseFrame, Message, WebSocket, WebSocketUpgrade};
+use axum::extract::{Query, State};
+use axum::response::Response;
+use serde::Deserialize;
+use tokio::sync::broadcast::error::RecvError;
+use tracing::{debug, info, warn};
+
+use futures::{SinkExt, StreamExt};
+
+#[derive(Debug, Deserialize)]
+pub struct SubscribeQuery {
+    /// The seq of the last event the client already has. Everything with a
+    /// larger seq is replayed before the live stream starts. Absent means
+    /// "live only".
+    pub cursor: Option<i64>,
+}
+
+/// The upgrade handler. Everything interesting happens in [`run`].
+pub async fn subscribe_repos(
+    State(state): State<AppState>,
+    Query(q): Query<SubscribeQuery>,
+    ws: WebSocketUpgrade,
+) -> Response {
+    ws.on_upgrade(move |socket| run(socket, state, q.cursor))
+}
+
+/// Drive one subscriber for the life of its connection.
+async fn run(socket: WebSocket, state: AppState, cursor: Option<i64>) {
+    // 1. Subscribe first — see the ordering note in the module header.
+    let mut rx = state.firehose.subscribe();
+
+    let (mut sink, mut stream) = socket.split();
+
+    // A firehose subscriber sends nothing after the upgrade, but we still
+    // have to read the socket: that is the only way a Close frame (or a
+    // client that vanishes without sending data) is noticed while we are
+    // parked waiting for an event that may not come for hours. The reader
+    // task does nothing but detect the end of the connection and say so.
+    let (dead_tx, mut dead_rx) = tokio::sync::oneshot::channel::<()>();
+    tokio::spawn(async move {
+        while let Some(msg) = stream.next().await {
+            match msg {
+                Ok(Message::Close(_)) | Err(_) => break,
+                // Ping/Pong are handled by axum itself; anything else a
+                // client sends on this endpoint is meaningless and ignored
+                // rather than treated as an error.
+                Ok(_) => continue,
+            }
+        }
+        let _ = dead_tx.send(());
+    });
+
+    // 2./3. Work out what the cursor asks for.
+    let bounds = match firehose::seq_bounds(&state.db).await {
+        Ok(b) => b,
+        Err(e) => {
+            warn!(error = %e, "subscribeRepos: firehose_events bounds read failed");
+            let _ = send_error(&mut sink, "InternalServerError", "event log unavailable").await;
+            return;
+        }
+    };
+
+    let mut replay_from = match firehose::plan_cursor(cursor, bounds) {
+        CursorPlan::LiveOnly => None,
+        CursorPlan::Replay { from } => Some(from),
+        CursorPlan::OutdatedCursor { from, message } => {
+            // Not fatal: the client keeps its connection and gets everything
+            // we still have. It is told about the hole so it can decide
+            // whether to backfill some other way.
+            info!(cursor = ?cursor, %message, "subscribeRepos: outdated cursor");
+            if sink
+                .send(Message::Binary(firehose::encode_info_frame(
+                    "OutdatedCursor",
+                    &message,
+                )))
+                .await
+                .is_err()
+            {
+                return;
+            }
+            Some(from)
+        }
+        CursorPlan::FutureCursor { message } => {
+            // Fatal. Continuing would leave the client waiting for events
+            // that will be numbered below its cursor and therefore filtered
+            // out forever — silence is the worst possible answer here.
+            info!(cursor = ?cursor, %message, "subscribeRepos: future cursor");
+            let _ = send_error(&mut sink, "FutureCursor", &message).await;
+            return;
+        }
+    };
+
+    info!(
+        cursor = ?cursor,
+        replay_from = ?replay_from,
+        subscribers = state.firehose.subscriber_count(),
+        "subscribeRepos: client connected"
+    );
+
+    // 4. Drain the replay. `high_water` ends up as the last seq the client
+    //    has been given, which is exactly the boundary the live filter needs.
+    let mut high_water: i64 = 0;
+    if let Some(from) = replay_from.take() {
+        match replay(&mut sink, &state, from, &mut dead_rx).await {
+            Ok(last) => high_water = last,
+            // The socket died mid-replay, or the log became unreadable.
+            // Either way there is nothing left to do for this connection.
+            Err(()) => return,
+        }
+    }
+
+    // 5. Live.
+    let mut lag_recoveries: u32 = 0;
+    loop {
+        tokio::select! {
+            _ = &mut dead_rx => {
+                debug!("subscribeRepos: client closed");
+                return;
+            }
+            recv = rx.recv() => match recv {
+                Ok(event) => {
+                    if event.seq <= high_water {
+                        // Already delivered by the replay. This is the
+                        // duplicate the subscribe-first ordering trades the
+                        // gap for.
+                        continue;
+                    }
+                    if !send_event(&mut sink, &event).await {
+                        return;
+                    }
+                    high_water = event.seq;
+                }
+                Err(RecvError::Lagged(skipped)) => {
+                    lag_recoveries += 1;
+                    warn!(
+                        skipped,
+                        attempt = lag_recoveries,
+                        high_water,
+                        "subscribeRepos: subscriber lagged; falling back to the database replay"
+                    );
+                    if lag_recoveries > MAX_LAG_RECOVERIES {
+                        // See the lag policy in `crate::firehose`: we rescue a
+                        // slow client, repeatedly, but not forever.
+                        let _ = send_error(
+                            &mut sink,
+                            "ConsumerTooSlow",
+                            "consumer fell behind repeatedly; reconnect with a cursor",
+                        )
+                        .await;
+                        return;
+                    }
+                    let message = format!(
+                        "consumer lagged by {skipped} events; resuming from seq {high_water} \
+                         via the durable log"
+                    );
+                    if sink
+                        .send(Message::Binary(firehose::encode_info_frame(
+                            "OutdatedCursor",
+                            &message,
+                        )))
+                        .await
+                        .is_err()
+                    {
+                        return;
+                    }
+                    // Nothing is lost: every event is in `firehose_events`,
+                    // so re-reading from the high-water mark is the same
+                    // stream the channel dropped. The advisory-lock ordering
+                    // guarantees the new high-water mark is again an exact
+                    // boundary for the live filter.
+                    match replay(&mut sink, &state, high_water, &mut dead_rx).await {
+                        Ok(last) => high_water = last.max(high_water),
+                        Err(()) => return,
+                    }
+                }
+                Err(RecvError::Closed) => {
+                    // Only happens at process shutdown, when the AppState
+                    // (and with it the sender) is dropped.
+                    debug!("subscribeRepos: broadcast channel closed");
+                    return;
+                }
+            }
+        }
+    }
+}
+
+/// Stream every event with `seq > from` out of the database, page by page.
+///
+/// Returns the highest seq actually sent (or `from` when there was nothing to
+/// send), or `Err(())` when the connection or the database gave out — in
+/// which case the caller should drop the connection.
+///
+/// Paging matters: a client reconnecting with `cursor=0` after a long uptime
+/// would otherwise pull the whole table, CARs and all, into memory at once.
+/// It also gives the loop a natural place to notice a client that closed the
+/// socket halfway through a large backfill.
+async fn replay(
+    sink: &mut futures::stream::SplitSink<WebSocket, Message>,
+    state: &AppState,
+    from: i64,
+    dead_rx: &mut tokio::sync::oneshot::Receiver<()>,
+) -> Result<i64, ()> {
+    let mut cursor = from;
+    loop {
+        if dead_rx.try_recv().is_ok() {
+            return Err(());
+        }
+        let page = match firehose::load_events_after(&state.db, cursor, REPLAY_PAGE_SIZE).await {
+            Ok(p) => p,
+            Err(e) => {
+                warn!(error = %format!("{e:#}"), "subscribeRepos: replay read failed");
+                let _ = send_error(sink, "InternalServerError", "event log read failed").await;
+                return Err(());
+            }
+        };
+        if page.is_empty() {
+            return Ok(cursor);
+        }
+        for event in &page {
+            if !send_event(sink, event).await {
+                return Err(());
+            }
+            cursor = event.seq;
+        }
+        // A short page means we reached the end of the log.
+        if (page.len() as i64) < REPLAY_PAGE_SIZE {
+            return Ok(cursor);
+        }
+    }
+}
+
+/// Send one `#commit` frame. Returns `false` when the socket is gone.
+async fn send_event(
+    sink: &mut futures::stream::SplitSink<WebSocket, Message>,
+    event: &FirehoseEvent,
+) -> bool {
+    sink.send(Message::Binary(event.to_frame())).await.is_ok()
+}
+
+/// Send a terminal error frame and close the socket.
+///
+/// The close is explicit (rather than just dropping the sink) so a client
+/// distinguishes "the server said no" from "the connection broke".
+async fn send_error(
+    sink: &mut futures::stream::SplitSink<WebSocket, Message>,
+    error: &str,
+    message: &str,
+) -> bool {
+    if sink
+        .send(Message::Binary(firehose::encode_error_frame(error, message)))
+        .await
+        .is_err()
+    {
+        return false;
+    }
+    sink.send(Message::Close(Some(CloseFrame {
+        code: axum::extract::ws::close_code::NORMAL,
+        reason: error.to_string().into(),
+    })))
+    .await
+    .is_ok()
+}

crates/pds-server/src/state.rs

@@ -1,4 +1,5 @@
 use crate::appview_push::AppViewPushClient;
+use crate::firehose::Firehose;
 use at_blob::S3BlobStore;
 use at_identity::plc::PlcClient;
 use at_lexicon::{Lex, LexRegistry};
@@ -16,6 +17,14 @@ pub struct AppState {
     pub blockstore: Arc<MemoryBlockstore>,
     pub plc: PlcClient,
     pub appview: AppViewPushClient,
+    /// Live fan-out for `com.atproto.sync.subscribeRepos`.
+    ///
+    /// Lives on the shared state rather than in the route module because the
+    /// *write* paths publish into it — `routes::helpers::apply_repo_write`
+    /// hands every committed event over here — while the WebSocket handler
+    /// only subscribes. Cloning `AppState` clones the sender, which is the
+    /// intended way to reach it from a handler.
+    pub firehose: Firehose,
 }
 
 impl AppState {
@@ -26,9 +35,9 @@ impl AppState {
             Lex::from_json(include_str!("../../../lexicons/app/twi/post.json")).unwrap(),
         );
         // AT-Protocol standard collections: only the records the user
-        // might legitimately create server-side (feed.like + feed.repost).
+        // might legitimately create server-side (feed.like, feed.repost,
-        // The full atproto collection library is out of scope — for
+        // graph.follow). The full atproto collection library is out of
-        // anything else, callers pass `validate: false` in the
+        // scope — for anything else, callers pass `validate: false` in the
         // createRecord body.
         lex.lexicons.insert(
             "app.bsky.feed.like".to_string(),
@@ -38,6 +47,17 @@ impl AppState {
             "app.bsky.feed.repost".to_string(),
             Lex::from_json(include_str!("../../../lexicons/app/bsky/feed/repost.json")).unwrap(),
         );
+        // Follow record. Its absence was a real outage: the desktop
+        // client creates follows through `createRecord`, which validates
+        // by default, so every follow came back
+        // `unknown lexicon: app.bsky.graph.follow` — the button could
+        // never have worked. `subject` is a bare DID string here, not a
+        // strongRef like like/repost use, matching what the client sends
+        // and what the AppView's `follow_subject_did` reads.
+        lex.lexicons.insert(
+            "app.bsky.graph.follow".to_string(),
+            Lex::from_json(include_str!("../../../lexicons/app/bsky/graph/follow.json")).unwrap(),
+        );
         // Profile record — avatar/banner/display name/description.
         // Validates the createRecord body when the Tauri client calls
         // its setProfile command. Other fields stay optional so a
@@ -63,6 +83,7 @@ impl AppState {
             blockstore: Arc::new(MemoryBlockstore::new()),
             plc: PlcClient::new(plc_url),
             appview,
+            firehose: Firehose::new(),
         }
     }
 }

crates/pds-server/tests/firehose_integration.rs

@@ -0,0 +1,677 @@
+//! Integration tests for `com.atproto.sync.subscribeRepos`.
+//!
+//! Same contract as the other integration suites in this crate: they talk to a
+//! PDS listening on `127.0.0.1:2583` and **fail open** — if nothing answers
+//! `/healthz`, the test prints a note and returns green rather than failing a
+//! developer's `cargo test` on a machine with no server running. Start the
+//! server (`./target/debug/pds-server` with `.env` sourced) to actually
+//! exercise them.
+//!
+//! ## Why the frames are decoded by hand here
+//!
+//! `pds-server` is a binary, so a test crate cannot import its `dag_cbor`
+//! module — and that is a feature, not a limitation. These tests are the
+//! *consumer* side of the wire contract, and a consumer that reuses the
+//! producer's encoder proves nothing: it would happily agree with a frame
+//! that no other implementation can read. The decoder below is written from
+//! the spec (tag 42, identity prefix, length-first map keys) and knows
+//! nothing about how the server produced the bytes.
+
+use serde_json::{json, Value};
+use std::collections::BTreeMap;
+use std::time::Duration;
+
+use futures::{SinkExt, StreamExt};
+use tokio_tungstenite::tungstenite::Message;
+
+const PDS_URL: &str = "http://127.0.0.1:2583";
+const PDS_WS: &str = "ws://127.0.0.1:2583";
+
+// -- harness ---------------------------------------------------------------
+
+fn http() -> reqwest::Client {
+    reqwest::Client::builder()
+        .timeout(Duration::from_secs(5))
+        .build()
+        .unwrap()
+}
+
+async fn wait_for_pds() -> bool {
+    let c = http();
+    for _ in 0..20 {
+        if let Ok(r) = c.get(format!("{PDS_URL}/healthz")).send().await {
+            if r.status().is_success() {
+                return true;
+            }
+        }
+        tokio::time::sleep(Duration::from_millis(250)).await;
+    }
+    false
+}
+
+async fn fresh_user(prefix: &str) -> (reqwest::Client, String, String) {
+    let c = http();
+    let handle = format!(
+        "{}_{}.maarcadetweet.local",
+        prefix,
+        uuid::Uuid::new_v4().simple()
+    );
+    let acc: Value = c
+        .post(format!("{PDS_URL}/xrpc/com.atproto.server.createAccount"))
+        .json(&json!({"handle": handle, "password": "hunter2hunter2"}))
+        .send()
+        .await
+        .unwrap()
+        .json()
+        .await
+        .unwrap();
+    let did = acc["did"].as_str().expect("createAccount did").to_string();
+    let jwt = acc["access_jwt"].as_str().expect("access_jwt").to_string();
+    (c, did, jwt)
+}
+
+async fn create_post(c: &reqwest::Client, did: &str, jwt: &str, text: &str) -> Value {
+    c.post(format!("{PDS_URL}/xrpc/com.atproto.repo.createRecord"))
+        .bearer_auth(jwt)
+        .json(&json!({
+            "repo": did,
+            "collection": "app.twi.post",
+            "record": { "text": text, "createdAt": "2026-09-10T12:00:00Z" },
+        }))
+        .send()
+        .await
+        .unwrap()
+        .json()
+        .await
+        .unwrap()
+}
+
+type Socket = tokio_tungstenite::WebSocketStream<
+    tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
+>;
+
+async fn subscribe(cursor: Option<i64>) -> Socket {
+    let url = match cursor {
+        Some(c) => format!("{PDS_WS}/xrpc/com.atproto.sync.subscribeRepos?cursor={c}"),
+        None => format!("{PDS_WS}/xrpc/com.atproto.sync.subscribeRepos"),
+    };
+    let (socket, _resp) = tokio_tungstenite::connect_async(&url)
+        .await
+        .unwrap_or_else(|e| panic!("subscribeRepos connect to {url}: {e}"));
+    socket
+}
+
+/// Read the next **binary** message, or `None` on timeout.
+///
+/// Text messages would be a protocol violation on this endpoint and are
+/// asserted against rather than skipped.
+async fn next_frame(socket: &mut Socket) -> Option<Vec<u8>> {
+    let deadline = Duration::from_secs(10);
+    loop {
+        match tokio::time::timeout(deadline, socket.next()).await {
+            Err(_) => return None,
+            Ok(None) => return None,
+            Ok(Some(Ok(Message::Binary(b)))) => return Some(b),
+            Ok(Some(Ok(Message::Ping(_)))) | Ok(Some(Ok(Message::Pong(_)))) => continue,
+            Ok(Some(Ok(Message::Close(_)))) => return None,
+            Ok(Some(Ok(other))) => panic!("subscribeRepos sent a non-binary frame: {other:?}"),
+            Ok(Some(Err(e))) => panic!("subscribeRepos socket error: {e}"),
+        }
+    }
+}
+
+/// Give the server a moment to finish `on_upgrade` and actually subscribe to
+/// the broadcast channel before we trigger a write.
+///
+/// The TCP handshake completing does not mean the handler has run. Without
+/// this the test would occasionally write before the subscription exists and
+/// then wait for a frame that was published to nobody. (The *cursor* replay
+/// path is the real fix for that race in production; the live-only test is
+/// deliberately testing the raceable path, so it waits.)
+async fn settle() {
+    tokio::time::sleep(Duration::from_millis(300)).await;
+}
+
+// -- an independent DAG-CBOR reader ----------------------------------------
+
+#[derive(Debug, Clone, PartialEq)]
+enum Cbor {
+    Null,
+    Bool(bool),
+    Int(i64),
+    Bytes(Vec<u8>),
+    Text(String),
+    Array(Vec<Cbor>),
+    Map(BTreeMap<String, Cbor>),
+    /// tag(42) + bytes(0x00 || cid) — the binary CID is kept raw and
+    /// re-encoded to a `bafy…` string on demand, so the test never depends on
+    /// the server's own CID formatting.
+    Link(Vec<u8>),
+}
+
+impl Cbor {
+    fn get(&self, key: &str) -> &Cbor {
+        match self {
+            Cbor::Map(m) => m
+                .get(key)
+                .unwrap_or_else(|| panic!("missing key `{key}` in {self:?}")),
+            other => panic!("not a map: {other:?}"),
+        }
+    }
+    fn opt(&self, key: &str) -> Option<&Cbor> {
+        match self {
+            Cbor::Map(m) => m.get(key),
+            _ => None,
+        }
+    }
+    fn int(&self) -> i64 {
+        match self {
+            Cbor::Int(i) => *i,
+            other => panic!("not an int: {other:?}"),
+        }
+    }
+    fn text(&self) -> &str {
+        match self {
+            Cbor::Text(s) => s,
+            other => panic!("not text: {other:?}"),
+        }
+    }
+    fn bool(&self) -> bool {
+        match self {
+            Cbor::Bool(b) => *b,
+            other => panic!("not a bool: {other:?}"),
+        }
+    }
+    fn bytes(&self) -> &[u8] {
+        match self {
+            Cbor::Bytes(b) => b,
+            other => panic!("not bytes: {other:?}"),
+        }
+    }
+    fn array(&self) -> &[Cbor] {
+        match self {
+            Cbor::Array(a) => a,
+            other => panic!("not an array: {other:?}"),
+        }
+    }
+    /// The link's CID rendered as a base32 `bafy…` string, for comparison
+    /// against what the XRPC JSON responses return.
+    fn link_cid(&self) -> String {
+        match self {
+            Cbor::Link(raw) => cid::Cid::read_bytes(&raw[..])
+                .expect("tag-42 payload must be a valid binary CID")
+                .to_string(),
+            other => panic!("not a link: {other:?}"),
+        }
+    }
+}
+
+fn read_head(b: &[u8], p: usize) -> (u8, u64, usize) {
+    let first = b[p];
+    let major = first >> 5;
+    let low = first & 0x1f;
+    let (arg, extra) = match low {
+        0..=23 => (low as u64, 0usize),
+        24 => (b[p + 1] as u64, 1),
+        25 => (u16::from_be_bytes([b[p + 1], b[p + 2]]) as u64, 2),
+        26 => (
+            u32::from_be_bytes([b[p + 1], b[p + 2], b[p + 3], b[p + 4]]) as u64,
+            4,
+        ),
+        27 => {
+            let mut n = 0u64;
+            for i in 0..8 {
+                n = (n << 8) | b[p + 1 + i] as u64;
+            }
+            (n, 8)
+        }
+        other => panic!("indefinite or reserved CBOR head 0x{other:02x} — illegal in DAG-CBOR"),
+    };
+    (major, arg, p + 1 + extra)
+}
+
+fn decode_at(b: &[u8], p: usize) -> (Cbor, usize) {
+    let (major, arg, mut p) = read_head(b, p);
+    match major {
+        0 => (Cbor::Int(arg as i64), p),
+        1 => (Cbor::Int(-(arg as i64) - 1), p),
+        2 => {
+            let end = p + arg as usize;
+            (Cbor::Bytes(b[p..end].to_vec()), end)
+        }
+        3 => {
+            let end = p + arg as usize;
+            (
+                Cbor::Text(std::str::from_utf8(&b[p..end]).unwrap().to_string()),
+                end,
+            )
+        }
+        4 => {
+            let mut items = Vec::new();
+            for _ in 0..arg {
+                let (v, next) = decode_at(b, p);
+                items.push(v);
+                p = next;
+            }
+            (Cbor::Array(items), p)
+        }
+        5 => {
+            let mut m = BTreeMap::new();
+            let mut prev_key: Option<String> = None;
+            for _ in 0..arg {
+                let (k, next) = decode_at(b, p);
+                p = next;
+                let key = k.text().to_string();
+                // Canonical DAG-CBOR order: shorter keys first, then
+                // bytewise. Asserted here because a consumer written against
+                // a strict codec (cborg's `dag-cbor` decoder, for one) will
+                // reject a frame whose keys are out of order.
+                if let Some(prev) = &prev_key {
+                    let ordered = (prev.len(), prev.as_bytes()) < (key.len(), key.as_bytes());
+                    assert!(ordered, "map keys out of canonical order: {prev:?} then {key:?}");
+                }
+                prev_key = Some(key.clone());
+                let (v, next) = decode_at(b, p);
+                p = next;
+                m.insert(key, v);
+            }
+            (Cbor::Map(m), p)
+        }
+        6 => {
+            assert_eq!(arg, 42, "DAG-CBOR permits only tag 42");
+            let (inner, next) = decode_at(b, p);
+            let raw = match inner {
+                Cbor::Bytes(v) => v,
+                other => panic!("tag 42 must wrap bytes, got {other:?}"),
+            };
+            assert_eq!(
+                raw.first(),
+                Some(&0x00),
+                "a binary CID link must start with the 0x00 multibase identity prefix"
+            );
+            (Cbor::Link(raw[1..].to_vec()), next)
+        }
+        7 => match arg {
+            20 => (Cbor::Bool(false), p),
+            21 => (Cbor::Bool(true), p),
+            22 => (Cbor::Null, p),
+            other => panic!("unsupported CBOR simple value {other}"),
+        },
+        other => panic!("unsupported CBOR major type {other}"),
+    }
+}
+
+/// Split one binary frame into its header and body values, and assert that
+/// the two together consume the whole message — a frame with trailing bytes
+/// would silently desynchronise a streaming consumer.
+fn parse_frame(bytes: &[u8]) -> (Cbor, Cbor) {
+    let (header, after_header) = decode_at(bytes, 0);
+    let (body, end) = decode_at(bytes, after_header);
+    assert_eq!(end, bytes.len(), "frame must be exactly two CBOR values");
+    (header, body)
+}
+
+/// Minimal CAR v1 reader: returns the root CIDs and the block CIDs, both as
+/// `bafy…` strings.
+fn parse_car(bytes: &[u8]) -> (Vec<String>, Vec<String>) {
+    fn varint(b: &[u8], p: &mut usize) -> u64 {
+        let (mut v, mut shift) = (0u64, 0u32);
+        loop {
+            let byte = b[*p];
+            *p += 1;
+            v |= ((byte & 0x7f) as u64) << shift;
+            if byte & 0x80 == 0 {
+                return v;
+            }
+            shift += 7;
+        }
+    }
+    let mut p = 0usize;
+    let header_len = varint(bytes, &mut p) as usize;
+    let header = &bytes[p..p + header_len];
+    p += header_len;
+
+    // The CAR header is a CBOR map; walk it with the same head reader. Note
+    // that this server's CAR header tags its roots *without* the 0x00
+    // identity prefix (a documented deviation in `car.rs`), so the roots are
+    // read as plain tagged byte strings rather than through `decode_at`.
+    let mut hp = 0usize;
+    let (major, n, next) = read_head(header, hp);
+    assert_eq!(major, 5, "CAR header must be a map");
+    hp = next;
+    let mut roots = Vec::new();
+    for _ in 0..n {
+        let (m, len, next) = read_head(header, hp);
+        assert_eq!(m, 3);
+        hp = next;
+        let key = std::str::from_utf8(&header[hp..hp + len as usize]).unwrap().to_string();
+        hp += len as usize;
+        if key == "version" {
+            let (m, v, next) = read_head(header, hp);
+            assert_eq!(m, 0);
+            assert_eq!(v, 1, "CAR must be v1");
+            hp = next;
+        } else if key == "roots" {
+            let (m, count, next) = read_head(header, hp);
+            assert_eq!(m, 4);
+            hp = next;
+            for _ in 0..count {
+                let (m, tag, next) = read_head(header, hp);
+                assert_eq!((m, tag), (6, 42), "root must be a tag-42 link");
+                hp = next;
+                let (m, len, next) = read_head(header, hp);
+                assert_eq!(m, 2);
+                hp = next;
+                // A DAG-CBOR link wraps `0x00 || <binary CID>`; the 0x00 is
+                // the multibase identity prefix and is not part of the CID.
+                // Assert on it rather than skipping it silently — this
+                // reader stands in for a foreign consumer, and dropping the
+                // check would let the header regress unnoticed.
+                let raw = &header[hp..hp + len as usize];
+                assert_eq!(
+                    raw.first(),
+                    Some(&0x00),
+                    "CAR root link must carry the multibase identity prefix"
+                );
+                roots.push(cid::Cid::read_bytes(&raw[1..]).unwrap().to_string());
+                hp += len as usize;
+            }
+        } else {
+            panic!("unexpected CAR header key {key}");
+        }
+    }
+
+    let mut blocks = Vec::new();
+    while p < bytes.len() {
+        let section_len = varint(bytes, &mut p) as usize;
+        let section = &bytes[p..p + section_len];
+        let cid = cid::Cid::read_bytes(section).unwrap();
+        blocks.push(cid.to_string());
+        p += section_len;
+    }
+    (roots, blocks)
+}
+
+// -- tests -----------------------------------------------------------------
+
+/// A live subscriber receives a `#commit` frame for a record created after it
+/// connected, and every field of that frame says what it should.
+#[tokio::test]
+async fn live_subscriber_receives_a_commit_frame() {
+    if !wait_for_pds().await {
+        eprintln!("pds not running, skipping");
+        return;
+    }
+    let (c, did, jwt) = fresh_user("fhlive").await;
+
+    let mut socket = subscribe(None).await;
+    settle().await;
+
+    let created = create_post(&c, &did, &jwt, "hello firehose").await;
+    let record_cid = created["cid"].as_str().expect("createRecord cid").to_string();
+    let commit_cid = created["commit"]["cid"]
+        .as_str()
+        .expect("createRecord commit.cid")
+        .to_string();
+    let commit_rev = created["commit"]["rev"].as_str().unwrap().to_string();
+    let uri = created["uri"].as_str().unwrap().to_string();
+    let rkey = uri.rsplit('/').next().unwrap().to_string();
+
+    // The account was created moments ago and has never written before, so
+    // the first frame we see for it is this post's. Other accounts may be
+    // writing concurrently, so filter by DID rather than taking frame 1.
+    let (header, body) = loop {
+        let bytes = next_frame(&mut socket)
+            .await
+            .expect("expected a #commit frame within the timeout");
+        let (header, body) = parse_frame(&bytes);
+        if header.opt("t").map(|t| t.text()) == Some("#commit")
+            && body.get("repo").text() == did
+        {
+            break (header, body);
+        }
+    };
+
+    assert_eq!(header.get("op").int(), 1, "regular frames carry op = 1");
+    assert_eq!(header.get("t").text(), "#commit");
+
+    assert!(body.get("seq").int() > 0, "seq must be a real cursor value");
+    assert_eq!(body.get("repo").text(), did);
+    assert_eq!(body.get("rev").text(), commit_rev);
+    assert_eq!(body.get("commit").link_cid(), commit_cid);
+    assert!(!body.get("rebase").bool());
+    assert!(!body.get("tooBig").bool());
+    assert!(body.get("blobs").array().is_empty());
+    // First-ever commit on a brand new repo — there is no predecessor.
+    assert_eq!(*body.get("since"), Cbor::Null, "first commit has since = null");
+    // RFC 3339 with a Z offset.
+    let time = body.get("time").text();
+    assert!(
+        chrono_like_rfc3339(time),
+        "time must be RFC 3339, got {time}"
+    );
+
+    // ops
+    let ops = body.get("ops").array();
+    assert_eq!(ops.len(), 1, "one record write means one op");
+    assert_eq!(ops[0].get("action").text(), "create");
+    assert_eq!(ops[0].get("path").text(), format!("app.twi.post/{rkey}"));
+    assert_eq!(ops[0].get("cid").link_cid(), record_cid);
+
+    // blocks: a CAR rooted at the commit, containing the commit block and the
+    // new record block.
+    let (roots, blocks) = parse_car(body.get("blocks").bytes());
+    assert_eq!(roots, vec![commit_cid.clone()], "CAR root is the commit");
+    assert!(
+        blocks.contains(&commit_cid),
+        "CAR must carry the commit block itself; got {blocks:?}"
+    );
+    assert!(
+        blocks.contains(&record_cid),
+        "CAR must carry the new record block; got {blocks:?}"
+    );
+
+    let _ = socket.send(Message::Close(None)).await;
+}
+
+/// The same event, fetched again from the durable log with a cursor, is
+/// byte-identical to the live frame.
+///
+/// Byte-identity is the strong form of the claim and the one that matters: a
+/// consumer that deduplicates by hashing frames, or that verifies a signature
+/// over them, must not see two different representations of one event.
+#[tokio::test]
+async fn cursor_replay_reproduces_the_live_frame_exactly() {
+    if !wait_for_pds().await {
+        eprintln!("pds not running, skipping");
+        return;
+    }
+    let (c, did, jwt) = fresh_user("fhreplay").await;
+
+    let mut live = subscribe(None).await;
+    settle().await;
+    create_post(&c, &did, &jwt, "replay me").await;
+
+    let live_bytes = loop {
+        let bytes = next_frame(&mut live)
+            .await
+            .expect("expected a live #commit frame");
+        let (header, body) = parse_frame(&bytes);
+        if header.opt("t").map(|t| t.text()) == Some("#commit")
+            && body.get("repo").text() == did
+        {
+            break bytes;
+        }
+    };
+    let (_h, live_body) = parse_frame(&live_bytes);
+    let seq = live_body.get("seq").int();
+    let _ = live.send(Message::Close(None)).await;
+
+    // Reconnect asking for everything after the event *before* ours, so the
+    // replay's first matching frame is the one we just saw.
+    let mut replayed = subscribe(Some(seq - 1)).await;
+    let replay_bytes = loop {
+        let bytes = next_frame(&mut replayed)
+            .await
+            .expect("expected the event to come back from the replay");
+        let (header, body) = parse_frame(&bytes);
+        if header.opt("t").map(|t| t.text()) == Some("#commit")
+            && body.get("seq").int() == seq
+        {
+            break bytes;
+        }
+    };
+
+    assert_eq!(
+        replay_bytes, live_bytes,
+        "a replayed frame must be byte-identical to the live one"
+    );
+    let _ = replayed.send(Message::Close(None)).await;
+}
+
+/// Handing over from replay to live loses nothing and duplicates nothing.
+///
+/// Connect with a cursor at the current head (so the replay is empty), then
+/// write twice: both events must arrive, in order, exactly once each.
+#[tokio::test]
+async fn replay_to_live_handover_has_no_gap_and_no_duplicate() {
+    if !wait_for_pds().await {
+        eprintln!("pds not running, skipping");
+        return;
+    }
+    let (c, did, jwt) = fresh_user("fhhandover").await;
+
+    // Establish where the log currently ends by writing one event and reading
+    // its seq off the live stream.
+    let mut probe = subscribe(None).await;
+    settle().await;
+    create_post(&c, &did, &jwt, "probe").await;
+    let head_seq = loop {
+        let bytes = next_frame(&mut probe).await.expect("probe frame");
+        let (header, body) = parse_frame(&bytes);
+        if header.opt("t").map(|t| t.text()) == Some("#commit")
+            && body.get("repo").text() == did
+        {
+            break body.get("seq").int();
+        }
+    };
+    let _ = probe.send(Message::Close(None)).await;
+
+    // Now reconnect at that exact cursor: nothing to replay, straight to live.
+    let mut socket = subscribe(Some(head_seq)).await;
+    settle().await;
+
+    create_post(&c, &did, &jwt, "after handover one").await;
+    create_post(&c, &did, &jwt, "after handover two").await;
+
+    let mut seen: Vec<i64> = Vec::new();
+    while seen.len() < 2 {
+        let bytes = next_frame(&mut socket)
+            .await
+            .expect("expected both post-handover frames");
+        let (header, body) = parse_frame(&bytes);
+        if header.opt("t").map(|t| t.text()) != Some("#commit") {
+            continue;
+        }
+        let seq = body.get("seq").int();
+        assert!(
+            seq > head_seq,
+            "the cursor said we already had seq {head_seq}; got {seq} again"
+        );
+        if body.get("repo").text() == did {
+            assert!(!seen.contains(&seq), "event {seq} delivered twice");
+            seen.push(seq);
+        }
+    }
+    assert_eq!(seen.len(), 2);
+    assert!(seen[0] < seen[1], "events must arrive in seq order: {seen:?}");
+    let _ = socket.send(Message::Close(None)).await;
+}
+
+/// A cursor past the end of the log is a terminal error frame, not silence.
+#[tokio::test]
+async fn future_cursor_gets_an_error_frame() {
+    if !wait_for_pds().await {
+        eprintln!("pds not running, skipping");
+        return;
+    }
+    let mut socket = subscribe(Some(i64::MAX / 2)).await;
+    let bytes = next_frame(&mut socket)
+        .await
+        .expect("expected an error frame for a future cursor");
+    let (header, body) = parse_frame(&bytes);
+    assert_eq!(header.get("op").int(), -1, "error frames carry op = -1");
+    assert!(
+        header.opt("t").is_none(),
+        "an error header has no `t`, only `op`"
+    );
+    assert_eq!(body.get("error").text(), "FutureCursor");
+    assert!(
+        !body.get("message").text().is_empty(),
+        "the error should say what went wrong"
+    );
+}
+
+/// A record deleted through `deleteRecord` produces a `delete` op with a null
+/// CID — the one op shape that is not a link.
+#[tokio::test]
+async fn delete_produces_a_delete_op_with_a_null_cid() {
+    if !wait_for_pds().await {
+        eprintln!("pds not running, skipping");
+        return;
+    }
+    let (c, did, jwt) = fresh_user("fhdelete").await;
+    let created = create_post(&c, &did, &jwt, "to be deleted").await;
+    let rkey = created["uri"]
+        .as_str()
+        .unwrap()
+        .rsplit('/')
+        .next()
+        .unwrap()
+        .to_string();
+
+    let mut socket = subscribe(None).await;
+    settle().await;
+
+    let resp = c
+        .post(format!("{PDS_URL}/xrpc/com.atproto.repo.deleteRecord"))
+        .bearer_auth(&jwt)
+        .json(&json!({
+            "repo": did,
+            "collection": "app.twi.post",
+            "rkey": rkey,
+        }))
+        .send()
+        .await
+        .unwrap();
+    assert!(resp.status().is_success(), "deleteRecord: {:?}", resp.status());
+
+    let (_header, body) = loop {
+        let bytes = next_frame(&mut socket).await.expect("expected a delete frame");
+        let (header, body) = parse_frame(&bytes);
+        if header.opt("t").map(|t| t.text()) == Some("#commit")
+            && body.get("repo").text() == did
+        {
+            break (header, body);
+        }
+    };
+
+    let ops = body.get("ops").array();
+    assert_eq!(ops.len(), 1);
+    assert_eq!(ops[0].get("action").text(), "delete");
+    assert_eq!(ops[0].get("path").text(), format!("app.twi.post/{rkey}"));
+    assert_eq!(*ops[0].get("cid"), Cbor::Null, "a delete has no resulting CID");
+
+    // The second commit on this repo, so `since` is the previous revision.
+    assert!(
+        matches!(body.get("since"), Cbor::Text(_)),
+        "a follow-up commit must name its predecessor's rev, got {:?}",
+        body.get("since")
+    );
+    let _ = socket.send(Message::Close(None)).await;
+}
+
+/// Shape check for the frame's `time`: RFC 3339, UTC, with a `Z` suffix.
+fn chrono_like_rfc3339(s: &str) -> bool {
+    chrono::DateTime::parse_from_rfc3339(s).is_ok() && s.ends_with('Z')
+}

migrations/pds/0003_firehose.sql

@@ -0,0 +1,112 @@
+-- PDS database schema 0003: the firehose event log.
+--
+-- Why
+--
+-- Until now the PDS produced no `com.atproto.sync.subscribeRepos` stream at
+-- all. The only way a local record reached the AppView was the best-effort
+-- HTTP push in `appview_push.rs` — a fire-and-forget `tokio::spawn` that is
+-- explicitly documented as "the Jetstream replay will catch up". There is no
+-- Jetstream replay for records that only exist on this PDS, so a dropped push
+-- meant the post was simply never indexed. Nothing retried it, and nothing
+-- could: the commit lived in `repos` / `repo_blocks` but there was no ordered
+-- log of *what changed* for a consumer to walk.
+--
+-- This table is that log. Every repo write appends exactly one row, in the
+-- same transaction as the head-pointer update, so the sequence and the repo
+-- head can never disagree. A consumer that reconnects with a cursor replays
+-- from here; a consumer that is live gets the same rows pushed over a
+-- broadcast channel.
+--
+-- Column choices
+--
+--   seq         BIGSERIAL PRIMARY KEY — the cursor. It has to be a single
+--               monotonically increasing integer because that is what the
+--               `subscribeRepos` wire contract hands the client and takes
+--               back as `?cursor=`. BIGSERIAL (not an `(timestamp, id)`
+--               keyset like the AppView's notifications table) because the
+--               protocol's cursor is opaque-but-numeric and clients compare
+--               it with `>`.
+--
+--               Sequence values are handed out at INSERT time, which by
+--               itself does NOT guarantee that they become *visible* in seq
+--               order — two transactions can grab 5 and 6 and commit in the
+--               opposite order, leaving a reader that polls in between with a
+--               gap it would never fill. The write path therefore takes
+--               `pg_advisory_xact_lock` on a fixed key immediately before
+--               this INSERT (see `routes::helpers::apply_repo_write`), which
+--               serialises the tail of every firehose-writing transaction so
+--               commit order == seq order. That is what makes "give me
+--               everything with seq > N" an exact, gap-free replay rather
+--               than a best guess.
+--
+--   did         the repo the event belongs to. Not a FK to `users(did)`:
+--               the log outlives the account. If a user is deleted we still
+--               want consumers that are mid-replay to see the events that
+--               already happened rather than have the rows cascade out from
+--               under their cursor.
+--
+--   rev         the new commit's revision (TID string), mirrored from
+--               `repos.rev`. Goes out as the frame's `rev`.
+--
+--   since       the *previous* commit's rev, or NULL for the first commit on
+--               a repo. The frame's `since` field; a consumer uses it to
+--               detect that it missed an intermediate commit.
+--
+--   commit_cid  BYTEA holding the raw binary CID of the new commit, stored
+--               the same way `repos.head_cid` stores it so the two are
+--               directly comparable with `=` and no text/binary conversion
+--               is needed to join them.
+--
+--   blocks      BYTEA holding a complete CAR v1 file: the commit block as the
+--               root plus every block this commit newly created (MST nodes
+--               and record values). Stored pre-serialised rather than
+--               reassembled from `repo_blocks` at read time because the
+--               *diff* — which blocks were new for this particular commit —
+--               is only knowable at write time. Recomputing it later would
+--               mean diffing two MST snapshots on every replayed event.
+--
+--   ops         JSONB array of `{action, path, cid}`, the same objects that
+--               go into the frame's `ops` field. JSONB rather than a child
+--               table because it is always read as a whole, is never queried
+--               by content, and a child table would need its own ordering
+--               column to reproduce the array faithfully.
+--
+--   created_at  when the event was appended. This is what the frame's `time`
+--               field carries, so a replayed frame is byte-identical to the
+--               live one that was broadcast at commit time — a consumer that
+--               deduplicates by hashing frames does not see two different
+--               frames for one event.
+--
+-- Retention: there is none
+-- ------------------------
+-- Nothing prunes this table. It grows by one row per repo write, and each row
+-- carries a CAR of the commit's new blocks (a few hundred bytes for a plain
+-- post, more when a record is large). At the volume this deployment sees that
+-- is fine for a long time, but it is unbounded, and an operator who wants a
+-- bound has to add one. Deleting the oldest rows is safe: a client whose
+-- cursor points before the surviving range gets an `#info`/`OutdatedCursor`
+-- frame and resumes from the oldest row that still exists. See the module
+-- header of `crates/pds-server/src/firehose.rs`.
+
+CREATE TABLE IF NOT EXISTS firehose_events (
+    seq            BIGSERIAL PRIMARY KEY,
+    did            TEXT NOT NULL,
+    rev            TEXT NOT NULL,
+    since          TEXT,
+    commit_cid     BYTEA NOT NULL,
+    blocks         BYTEA NOT NULL,
+    ops            JSONB NOT NULL,
+    created_at     TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+-- Cursor replay is `WHERE seq > $1 ORDER BY seq LIMIT $2`, which the
+-- BIGSERIAL primary key's own index already serves — no second index for
+-- that, on purpose: an extra index on `seq` would be pure write amplification
+-- on the hottest path in this table.
+--
+-- What the PK does *not* serve is "replay one repo", which is how an operator
+-- re-drives a single account into the AppView after an ingest bug, and how
+-- `getRepo`-style backfills are debugged. `(did, seq)` covers that and keeps
+-- the per-repo scan in seq order.
+CREATE INDEX IF NOT EXISTS firehose_events_did_seq_idx
+    ON firehose_events (did, seq);