//! An append-only, structured record of every write atgc sends to a PDS.
//!
//! # Why this exists
//!
//! [`crate::logging::oauth`] answers "why did my session stop working". This one
//! answers the other question an incident asks, which nothing could answer
//! before: **what did atgc actually write to my repository, and did it land?**
//!
//! Every durable thing atgc does is a record in the user's own PDS — a pull
//! request and each of its rounds, a repo, an SSH key, a comment, a vote, a
//! feedback report — and a stack reconcile rewrites a dozen of them in one
//! atomic batch. When one of those goes wrong the visible symptom is a long
//! way from the cause: a pull that shows an old round, a stack whose
//! `dependentOn` chain points at a record that was never rewritten, an appview
//! index that disagrees with the PDS. Reconstructing what was sent meant
//! re-running the command with `--debug` and hoping it misbehaved twice.
//!
//! So: one line per write leaving the process, one line per answer coming
//! back, with the collection, the record key, the compare-and-swap
//! precondition that was or was not sent, and the at-uri and CID the PDS
//! returned. `--dry-run` writes nothing and therefore appears here as nothing,
//! which is the honest rendering of it.
//!
//! # Where it observes from
//!
//! At the transport, exactly as [`crate::logging::oauth::LoggedHttpClient`] does,
//! and for the same reason spelled out there: [`LoggedPdsClient`] wraps the
//! HTTP client every authenticated request goes through, so no call site can
//! forget to log and no future one can be added without being logged. `pr
//! create`, `stack resubmit`, `key add`, `repo edit`, `report` and everything
//! after them are covered without a line of code in any of them.
//!
//! It also means what is recorded is what went on the wire rather than what a
//! caller believed it was sending, which is the distinction that mattered in
//! the OAuth incident and will matter here the first time a record is
//! serialized differently from how it reads in source.
//!
//! One consequence worth knowing before it surprises you: a DPoP-protected
//! PDS answers a request with a fresh nonce demand (`401`,
//! `use_dpop_nonce`) and jacquard retries. So a single logical write can
//! appear as a refusal immediately followed by an identical request that
//! succeeds. That is the truth of the exchange, not a bug in the log.
//!
//! # Format
//!
//! One JSON object per line at `~/.config/atgc/pds.jsonl`, mode 0600,
//! `ATGC_PDS_LOG` to move it and `ATGC_PDS_LOG=0` to switch it off. The
//! envelope, the size cap, the rotation and the invocation id are
//! [`crate::logging::file`]'s and are identical to the OAuth log's, so a line from
//! each can be laid beside the other and joined on `inv`.
//!
//! # What is never written
//!
//! **Record content.** Not a pull request's body, not a repo description, not
//! a comment, not the bytes of a patch or an image. A record is often the
//! user's own draft prose and a log is not the place for it. What is kept
//! instead is enough to identify a write without reproducing it: the
//! collection, the record key, the serialized size, and [`Fp`] of the value —
//! eight characters of a SHA-256, which is how "this resubmit uploaded the
//! same patch as the last one" gets answered without either patch being on
//! disk twice.
//!
//! Every field that could hold content is typed [`Fp`], so writing the real
//! thing there does not compile, and the extraction functions in this file
//! ([`request_of`], [`result_of`]) are the complete set of places that so
//! much as look at a body. Response bodies are not stored either: a refusal
//! keeps the XRPC `error` and `message`, both clipped, and nothing else.

use crate::logging::file::{Fp, Log, MAX_FIELD, clip};
use serde::{Deserialize, Serialize};
use std::borrow::Cow;

/// This log: `~/.config/atgc/pds.jsonl`, `ATGC_PDS_LOG` to move or disable
/// it.
pub static LOG: Log = Log::new("pds.jsonl", "ATGC_PDS_LOG");

/// How many operations of an `applyWrites` batch are described individually.
///
/// A stack reconcile is the batch that matters and runs to a handful of ops;
/// beyond that the per-op detail stops paying for the line budget it costs,
/// and the count is kept instead. Eight is what fits: the width test below
/// is the authority, and it failed at twelve.
const MAX_BATCH_OPS: usize = 8;

/// Cap for the identifier-shaped fields: DIDs, NSIDs, record keys, CIDs,
/// at-uris, XRPC error names.
///
/// [`MAX_FIELD`] is the cap for a field that could say anything, and every
/// one of these could not: the longest of them in practice is a CID at about
/// 60 characters and the shortest is a 13-character record key. 128 is two
/// to ten times any real value, which is the right amount of room for a
/// field whose shape is known — and the difference matters, because a
/// `write_request` carries six of them at once plus a batch, and six times
/// 512 does not fit in one atomic append.
const MAX_ID: usize = 128;

/// Cap for a field inside one op of a batch, where the budget is divided by
/// [`MAX_BATCH_OPS`].
///
/// Same argument again, tighter: the two fields this applies to are a
/// collection NSID (`sh.tangled.repo.pull` is 20 characters) and a record
/// key (13).
const MAX_OP_ID: usize = 64;

/// The PDS write methods, and the complete set of XRPC calls this log
/// records.
///
/// Named rather than matched by string at the call site so that the reader
/// can colour and align them, and so that adding a method is a change the
/// compiler asks about. Everything else atgc sends — `getRecord`,
/// `listRecords`, `describeRepo`, `getServiceAuth`, and every `sh.tangled.*`
/// call to a knot — is a read or is not the PDS, and is not recorded here.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Op {
    Create,
    Put,
    Delete,
    ApplyWrites,
    UploadBlob,
}

impl Op {
    /// The XRPC method this op is, as the last path segment of its endpoint.
    fn from_nsid(nsid: &str) -> Option<Op> {
        match nsid {
            "com.atproto.repo.createRecord" => Some(Op::Create),
            "com.atproto.repo.putRecord" => Some(Op::Put),
            "com.atproto.repo.deleteRecord" => Some(Op::Delete),
            "com.atproto.repo.applyWrites" => Some(Op::ApplyWrites),
            "com.atproto.repo.uploadBlob" => Some(Op::UploadBlob),
            _ => None,
        }
    }
}

/// Which record a write is about.
///
/// Nested rather than three flat fields because it is the same triple on
/// every event and a reader renders it as one thing — `sh.tangled.repo.pull/
/// 3msve3b` against a DID. `jq '.target.collection'` reaches it.
///
/// All three are optional because `uploadBlob` has none of them: a blob is
/// addressed by its own CID afterwards and belongs to the authenticated
/// account, which the request body never names.
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
pub struct Target {
    /// The DID whose repository is being written to.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub repo: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub collection: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub rkey: Option<String>,
}

impl Target {
    pub fn is_empty(&self) -> bool {
        self.repo.is_none() && self.collection.is_none() && self.rkey.is_none()
    }
}

/// What an `applyWrites` batch does, one op at a time.
///
/// The batch is the case the per-record events cannot describe: a stack
/// reconcile updates some pulls, creates others and deletes the dropped ones
/// in a single commit, and "applyWrites, 7 ops" is not enough to tell a
/// reconcile that did the right thing from one that did not.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BatchOp {
    pub action: BatchAction,
    pub collection: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub rkey: Option<String>,
    /// The record's content, as a fingerprint and never otherwise.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub value: Option<Fp>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum BatchAction {
    Create,
    Update,
    Delete,
    /// A `$type` in the batch this build does not know. Recorded rather than
    /// dropped, because a batch whose ops do not add up is worth seeing.
    Unknown,
}

/// Everything atgc can observe about a write to a PDS.
///
/// Serialized with `#[serde(tag = "event")]`, flat, for the same `jq`
/// reasons [`crate::logging::oauth::Event`] gives — and deserialized through this
/// same enum by [`crate::cmd::logs::pds`], so a variant added here is a compile
/// error in the reader rather than a line that renders as nothing.
///
/// # These variants are public
///
/// `atgc logs pds --json` prints the PDS write log's lines verbatim, so
/// every variant name below — as `serde` renames it, `snake_case` into the
/// `event` field — and every field name in it is part of the CLI's output
/// contract. `docs/output.md` says so, and the stability rules there apply:
/// a new variant or a new field is a `feat`, a rename or a removal is a
/// breaking `!`. Rename one here and somebody's `jq` filter stops matching.
#[derive(Debug, Serialize, Deserialize)]
#[serde(tag = "event", rename_all = "snake_case")]
pub enum Event {
    /// First line of every invocation, carrying the same facts as the OAuth
    /// log's own head line and stamped with the same `inv`.
    ///
    /// Written even by a command that turns out to send no writes at all.
    /// "This ran and wrote nothing" is an answer, and a log that only has
    /// lines when something happened cannot give it.
    Invocation {
        pid: u32,
        user: Option<String>,
        cwd: Option<String>,
        version: Cow<'static, str>,
        subcommand: String,
        rotated_from_bytes: Option<u64>,
    },

    /// A write left the process. No answer yet.
    ///
    /// Paired with exactly one of the three outcomes below, by `inv` and by
    /// order. An unpaired `write_request` is itself the finding: the process
    /// died, or is still hanging, between sending and hearing back.
    WriteRequest {
        op: Op,
        endpoint: String,
        #[serde(skip_serializing_if = "Target::is_empty", default)]
        target: Target,
        /// The `swapRecord` precondition, when the caller sent one: the CID
        /// the record was read at. A CID is public and short, so it is kept
        /// verbatim — matching it against the `cid` of an earlier
        /// `write_applied` is how a lost compare-and-swap is traced to the
        /// write that won.
        #[serde(skip_serializing_if = "Option::is_none")]
        swap_record: Option<String>,
        /// The `swapCommit` precondition: the whole repository's commit.
        #[serde(skip_serializing_if = "Option::is_none")]
        swap_commit: Option<String>,
        /// Bytes of the request body as sent.
        bytes: usize,
        /// The record's content as a fingerprint, never as content.
        #[serde(skip_serializing_if = "Option::is_none")]
        value: Option<Fp>,
        /// `applyWrites` only: what the batch does, up to
        /// [`MAX_BATCH_OPS`] of it.
        #[serde(skip_serializing_if = "Option::is_none")]
        writes: Option<Vec<BatchOp>>,
        /// How many ops the batch had beyond the ones described.
        #[serde(skip_serializing_if = "Option::is_none")]
        writes_omitted: Option<usize>,
    },

    /// The PDS accepted the write.
    WriteApplied {
        op: Op,
        #[serde(skip_serializing_if = "Target::is_empty", default)]
        target: Target,
        http_status: u16,
        /// The at-uri of the record written, when the answer names one.
        #[serde(skip_serializing_if = "Option::is_none")]
        uri: Option<String>,
        /// The CID the record landed at — the value a later `swapRecord`
        /// must present, and the join between this log and the record now in
        /// the PDS.
        #[serde(skip_serializing_if = "Option::is_none")]
        cid: Option<String>,
        /// `applyWrites` only: how many results came back.
        #[serde(skip_serializing_if = "Option::is_none")]
        results: Option<usize>,
        elapsed_ms: u64,
    },

    /// The PDS refused the write.
    ///
    /// `error` is the XRPC error name — `InvalidSwap`, `RecordNotFound`,
    /// `InvalidRequest` — which is the field worth filtering on;
    /// `AuthenticationRequired` and a `use_dpop_nonce` message together are
    /// the ordinary nonce handshake rather than a failure.
    ///
    /// The response body is deliberately not kept whole. `error` and
    /// `message` are the parts a PDS puts real information in, and a body
    /// echoed verbatim is the one place a record's own content could
    /// reappear in this file after all the care taken to keep it out.
    WriteRefused {
        op: Op,
        #[serde(skip_serializing_if = "Target::is_empty", default)]
        target: Target,
        http_status: u16,
        #[serde(skip_serializing_if = "Option::is_none")]
        error: Option<String>,
        #[serde(skip_serializing_if = "Option::is_none")]
        message: Option<String>,
        elapsed_ms: u64,
    },

    /// The write never reached a PDS: DNS, TLS, connection, timeout.
    WriteTransportError {
        op: Op,
        #[serde(skip_serializing_if = "Target::is_empty", default)]
        target: Target,
        error: String,
        elapsed_ms: u64,
    },

    /// A record that could not be appended atomically, replaced by this.
    Oversize { of: String, bytes: usize },
}

/// Open the log and record the invocation. Called once, from `main`.
pub fn init(args: &[String]) {
    LOG.open();
    let Some(head) = LOG.head(args) else { return };
    emit(Event::Invocation {
        pid: head.pid,
        user: head.user,
        cwd: head.cwd,
        version: Cow::Borrowed(head.version),
        subcommand: head.subcommand,
        rotated_from_bytes: head.rotated_from_bytes,
    });
}

/// Append one event. Never fails, never blocks a command.
pub fn emit(event: Event) {
    LOG.emit(event, |of, bytes| Event::Oversize { of, bytes });
}

// ===========================================================================
// Reading a request and a response
// ===========================================================================

/// What a write request says about itself, before it is sent.
///
/// Returned as a pair so that the outcome events can repeat the op and the
/// target without parsing the body a second time.
pub struct Observed {
    pub op: Op,
    pub target: Target,
    pub request: Event,
}

/// The NSID a `/xrpc/<nsid>` path names.
fn nsid_of(path: &str) -> Option<&str> {
    path.rsplit_once("/xrpc/").map(|(_, nsid)| nsid)
}

/// Read a request body, if this request is a PDS write at all.
///
/// `None` for everything else on this transport — reads, knot calls, DID
/// documents, the token endpoint — which is what keeps this log to its name.
pub fn request_of(request: &http::Request<Vec<u8>>) -> Option<Observed> {
    if request.method() != http::Method::POST {
        return None;
    }
    let op = Op::from_nsid(nsid_of(request.uri().path())?)?;
    let endpoint = clip(&request.uri().to_string(), MAX_ID * 2);
    let body = request.body();

    // A blob has no JSON to read and no record to name: it is bytes, and the
    // only thing worth keeping about bytes is how many and which ones.
    if op == Op::UploadBlob {
        return Some(Observed {
            op,
            target: Target::default(),
            request: Event::WriteRequest {
                op,
                endpoint,
                target: Target::default(),
                swap_record: None,
                swap_commit: None,
                bytes: body.len(),
                value: Some(Fp::of_bytes(body)),
                writes: None,
                writes_omitted: None,
            },
        });
    }

    // Anything that fails to parse is still a write, and saying so with an
    // empty target beats dropping the line: a body this file cannot read is
    // exactly the case somebody will want to see.
    let json: serde_json::Value = serde_json::from_slice(body).unwrap_or(serde_json::Value::Null);
    let target = Target {
        repo: field(&json, "repo", MAX_ID),
        collection: field(&json, "collection", MAX_ID),
        rkey: field(&json, "rkey", MAX_ID),
    };
    let (writes, writes_omitted) = match op {
        Op::ApplyWrites => batch_ops(&json),
        _ => (None, None),
    };
    Some(Observed {
        op,
        target: target.clone(),
        request: Event::WriteRequest {
            op,
            endpoint,
            target,
            swap_record: field(&json, "swapRecord", MAX_ID),
            swap_commit: field(&json, "swapCommit", MAX_ID),
            bytes: body.len(),
            // The record itself, reduced to eight characters, here and
            // nowhere else.
            value: json.get("record").map(fingerprint),
            writes,
            writes_omitted,
        },
    })
}

/// Read an answer: which of the three outcomes it is, and what it said.
pub fn result_of(
    op: Op,
    target: Target,
    response: &http::Response<Vec<u8>>,
    elapsed_ms: u64,
) -> Event {
    let http_status = response.status().as_u16();
    let json: serde_json::Value =
        serde_json::from_slice(response.body()).unwrap_or(serde_json::Value::Null);
    if response.status().is_success() {
        return Event::WriteApplied {
            op,
            target,
            http_status,
            uri: field(&json, "uri", MAX_ID),
            // `createRecord` and `putRecord` answer with `cid`; `uploadBlob`
            // answers with a blob whose `ref` holds the same thing under
            // `$link`. Both are the address of what landed.
            cid: field(&json, "cid", MAX_ID).or_else(|| {
                json.get("blob")
                    .and_then(|b| b.get("ref"))
                    .and_then(|r| r.get("$link"))
                    .and_then(|l| l.as_str())
                    .map(|s| clip(s, MAX_ID))
            }),
            results: json.get("results").and_then(|r| r.as_array()).map(Vec::len),
            elapsed_ms,
        };
    }
    Event::WriteRefused {
        op,
        target,
        http_status,
        error: field(&json, "error", MAX_ID),
        message: field(&json, "message", MAX_FIELD),
        elapsed_ms,
    }
}

/// A string field of a JSON object, clipped to `max`.
///
/// Only ever called with the names of structural fields — `repo`,
/// `collection`, `rkey`, `swapRecord`, `swapCommit`, `uri`, `cid`, `error`,
/// `message`. Keeping the extraction to one narrow function is what makes
/// "no record content is written" checkable by reading its call sites.
fn field(json: &serde_json::Value, name: &str, max: usize) -> Option<String> {
    json.get(name)
        .and_then(|v| v.as_str())
        .map(|s| clip(s, max))
}

/// A JSON value's content as a fingerprint.
///
/// Serialized first, so that the digest is of the value and not of a
/// formatting of it — two records that differ only in key order fingerprint
/// alike, which is what a caller comparing "did I upload this already" wants.
fn fingerprint(value: &serde_json::Value) -> Fp {
    Fp::of(&serde_json::to_string(value).unwrap_or_default())
}

/// The ops of an `applyWrites` body, bounded.
fn batch_ops(json: &serde_json::Value) -> (Option<Vec<BatchOp>>, Option<usize>) {
    let Some(writes) = json.get("writes").and_then(|w| w.as_array()) else {
        return (None, None);
    };
    let ops: Vec<BatchOp> = writes
        .iter()
        .take(MAX_BATCH_OPS)
        .map(|w| BatchOp {
            action: match w.get("$type").and_then(|t| t.as_str()) {
                Some("com.atproto.repo.applyWrites#create") => BatchAction::Create,
                Some("com.atproto.repo.applyWrites#update") => BatchAction::Update,
                Some("com.atproto.repo.applyWrites#delete") => BatchAction::Delete,
                _ => BatchAction::Unknown,
            },
            collection: field(w, "collection", MAX_OP_ID).unwrap_or_default(),
            rkey: field(w, "rkey", MAX_OP_ID),
            value: w.get("value").map(fingerprint),
        })
        .collect();
    let omitted = writes.len().saturating_sub(ops.len());
    (Some(ops), (omitted > 0).then_some(omitted))
}

// ===========================================================================
// Instrumentation
// ===========================================================================

/// An HTTP client that records every PDS write on its way past.
///
/// Wraps [`crate::logging::oauth::LoggedHttpClient` ](crate::logging::oauth::LoggedHttpClient)
/// rather than replacing it: the two observers watch for different things on
/// the same traffic, and nesting them keeps each one's promise about what it
/// records auditable on its own. This is the outer layer, so it sees the
/// request exactly as jacquard built it.
///
/// Headers are never read. The DPoP proof is a header and a signed assertion
/// bound to this request; the authorization header carries the access token.
/// Neither is touched here and neither ever will be.
#[derive(Debug, Clone)]
pub struct LoggedPdsClient(pub crate::logging::oauth::LoggedHttpClient);

impl jacquard::common::http_client::HttpClient for LoggedPdsClient {
    type Error = reqwest::Error;

    async fn send_http(
        &self,
        request: http::Request<Vec<u8>>,
    ) -> Result<http::Response<Vec<u8>>, Self::Error> {
        let observed = request_of(&request);
        let Some(observed) = observed else {
            return self.0.send_http(request).await;
        };
        let Observed {
            op,
            target,
            request: event,
        } = observed;
        emit(event);
        let started = std::time::Instant::now();
        let result = self.0.send_http(request).await;
        let elapsed_ms = started.elapsed().as_millis().min(u64::MAX as u128) as u64;
        match &result {
            Ok(response) => emit(result_of(op, target, response, elapsed_ms)),
            Err(e) => emit(Event::WriteTransportError {
                op,
                target,
                error: clip(&e.to_string(), MAX_FIELD),
                elapsed_ms,
            }),
        }
        result
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::logging::file::{MAX_LINE, line_for_test};

    fn post(nsid: &str, body: serde_json::Value) -> http::Request<Vec<u8>> {
        http::Request::post(format!("https://pds.example/xrpc/{nsid}"))
            .body(serde_json::to_vec(&body).unwrap())
            .unwrap()
    }

    /// The filter that keeps this log to its name. Reads, knot calls and the
    /// token endpoint share this transport and must produce nothing.
    #[test]
    fn only_pds_writes_are_observed() {
        assert!(request_of(&post("com.atproto.repo.putRecord", serde_json::json!({}))).is_some());
        assert!(request_of(&post("com.atproto.repo.listRecords", serde_json::json!({}))).is_none());
        assert!(request_of(&post("sh.tangled.repo.delete", serde_json::json!({}))).is_none());
        let get = http::Request::get("https://pds.example/xrpc/com.atproto.repo.getRecord")
            .body(Vec::new())
            .unwrap();
        assert!(request_of(&get).is_none());
    }

    /// The promise of this file, checked rather than asserted: a record's
    /// content must not appear in the line that describes writing it.
    #[test]
    fn a_records_content_never_reaches_the_line() {
        let secret = "a private draft nobody has published yet";
        let observed = request_of(&post(
            "com.atproto.repo.putRecord",
            serde_json::json!({
                "repo": "did:plc:abc",
                "collection": "sh.tangled.repo.pull",
                "rkey": "3msve3bfn452b",
                "swapRecord": "bafyreiaaa",
                "record": {"title": secret, "body": secret},
            }),
        ))
        .expect("a putRecord is observed");
        let line = String::from_utf8(line_for_test(&observed.request)).unwrap();
        assert!(!line.contains(secret), "{line}");
        assert!(!line.contains("draft"), "{line}");
        // What survives is enough to find the record without reproducing it.
        assert!(line.contains("sh.tangled.repo.pull"), "{line}");
        assert!(line.contains("3msve3bfn452b"), "{line}");
        assert!(line.contains("bafyreiaaa"), "{line}");
    }

    /// Two writes of the same value fingerprint alike and two of different
    /// values do not — the whole reason a fingerprint is kept at all.
    #[test]
    fn the_fingerprint_distinguishes_writes_without_revealing_them() {
        let of = |title: &str| {
            let observed = request_of(&post(
                "com.atproto.repo.putRecord",
                serde_json::json!({"record": {"title": title}}),
            ))
            .unwrap();
            match observed.request {
                Event::WriteRequest { value, .. } => value.unwrap(),
                other => panic!("came back as {other:?}"),
            }
        };
        assert_eq!(of("same"), of("same"));
        assert_ne!(of("same"), of("different"));
    }

    /// A blob has no record to name; its bytes are the only identity it has
    /// before the PDS answers with a CID.
    #[test]
    fn a_blob_is_recorded_by_size_and_digest() {
        let request = http::Request::post("https://pds.example/xrpc/com.atproto.repo.uploadBlob")
            .body(b"gzipped patch bytes".to_vec())
            .unwrap();
        let observed = request_of(&request).expect("uploadBlob is a write");
        match observed.request {
            Event::WriteRequest {
                op, bytes, value, ..
            } => {
                assert_eq!(op, Op::UploadBlob);
                assert_eq!(bytes, 19);
                assert_eq!(value, Some(Fp::of("gzipped patch bytes")));
            }
            other => panic!("came back as {other:?}"),
        }
    }

    /// A stack reconcile is the batch this exists for: the ops have to be
    /// legible one by one, or "applyWrites, 7 ops" is all anybody ever knows.
    #[test]
    fn a_batch_describes_its_ops_and_says_what_it_left_out() {
        let op = |i: usize, kind: &str| {
            serde_json::json!({
                "$type": format!("com.atproto.repo.applyWrites#{kind}"),
                "collection": "sh.tangled.repo.pull",
                "rkey": format!("rkey{i}"),
                "value": {"title": format!("pull {i}")},
            })
        };
        let writes: Vec<_> = (0..MAX_BATCH_OPS + 3).map(|i| op(i, "update")).collect();
        let observed = request_of(&post(
            "com.atproto.repo.applyWrites",
            serde_json::json!({"repo": "did:plc:abc", "writes": writes}),
        ))
        .unwrap();
        match observed.request {
            Event::WriteRequest {
                writes,
                writes_omitted,
                ..
            } => {
                let writes = writes.expect("a batch lists its ops");
                assert_eq!(writes.len(), MAX_BATCH_OPS);
                assert_eq!(writes[0].action, BatchAction::Update);
                assert_eq!(writes[0].rkey.as_deref(), Some("rkey0"));
                assert_eq!(writes_omitted, Some(3));
            }
            other => panic!("came back as {other:?}"),
        }
    }

    /// The two answers, and the fields a reader filters on.
    #[test]
    fn an_answer_is_read_for_where_the_record_landed_or_why_it_did_not() {
        let ok = http::Response::builder()
            .status(200)
            .body(
                serde_json::to_vec(&serde_json::json!({
                    "uri": "at://did:plc:abc/sh.tangled.repo.pull/3msve3bfn452b",
                    "cid": "bafyreibbb",
                }))
                .unwrap(),
            )
            .unwrap();
        match result_of(Op::Put, Target::default(), &ok, 12) {
            Event::WriteApplied { uri, cid, .. } => {
                assert!(uri.unwrap().ends_with("3msve3bfn452b"));
                assert_eq!(cid.as_deref(), Some("bafyreibbb"));
            }
            other => panic!("came back as {other:?}"),
        }

        let refused = http::Response::builder()
            .status(400)
            .body(
                serde_json::to_vec(&serde_json::json!({
                    "error": "InvalidSwap",
                    "message": "Record was at a different CID",
                }))
                .unwrap(),
            )
            .unwrap();
        match result_of(Op::Put, Target::default(), &refused, 12) {
            Event::WriteRefused { error, message, .. } => {
                assert_eq!(error.as_deref(), Some("InvalidSwap"));
                assert!(message.unwrap().contains("different CID"));
            }
            other => panic!("came back as {other:?}"),
        }
    }

    /// A blob upload answers with the CID under `blob.ref.$link` rather than
    /// at the top level, and that is still where the bytes landed.
    #[test]
    fn a_blobs_cid_is_found_where_upload_blob_puts_it() {
        let ok = http::Response::builder()
            .status(200)
            .body(
                serde_json::to_vec(&serde_json::json!({
                    "blob": {"$type": "blob", "ref": {"$link": "bafkreiccc"}, "size": 19},
                }))
                .unwrap(),
            )
            .unwrap();
        match result_of(Op::UploadBlob, Target::default(), &ok, 3) {
            Event::WriteApplied { cid, .. } => assert_eq!(cid.as_deref(), Some("bafkreiccc")),
            other => panic!("came back as {other:?}"),
        }
    }

    /// A body this build cannot read is still a write, and losing the line
    /// would hide exactly the case worth seeing.
    #[test]
    fn an_unreadable_body_is_still_recorded() {
        let request = http::Request::post("https://pds.example/xrpc/com.atproto.repo.putRecord")
            .body(b"not json at all".to_vec())
            .unwrap();
        let observed = request_of(&request).expect("still a write");
        match observed.request {
            Event::WriteRequest { target, bytes, .. } => {
                assert!(target.is_empty());
                assert_eq!(bytes, 15);
            }
            other => panic!("came back as {other:?}"),
        }
    }

    /// The writer and the reader share one description of the format.
    #[test]
    fn an_event_survives_a_round_trip_through_json() {
        let event = Event::WriteApplied {
            op: Op::ApplyWrites,
            target: Target {
                repo: Some("did:plc:abc".into()),
                collection: Some("sh.tangled.repo.pull".into()),
                rkey: None,
            },
            http_status: 200,
            uri: None,
            cid: None,
            results: Some(7),
            elapsed_ms: 91,
        };
        let line = line_for_test(&event);
        match serde_json::from_slice::<Event>(&line).expect("the writer's own line parses") {
            Event::WriteApplied {
                op,
                target,
                results,
                ..
            } => {
                assert_eq!(op, Op::ApplyWrites);
                assert_eq!(target.repo.as_deref(), Some("did:plc:abc"));
                assert_eq!(results, Some(7));
            }
            other => panic!("came back as {other:?}"),
        }
    }

    /// Every variant, at its widest, has to fit in one atomic append — the
    /// full batch included, which is what bounds [`MAX_BATCH_OPS`].
    #[test]
    fn the_widest_events_fit_within_one_atomic_write() {
        let long = "y".repeat(MAX_FIELD);
        let target = Target {
            repo: Some(clip(&long, MAX_ID)),
            collection: Some(clip(&long, MAX_ID)),
            rkey: Some(clip(&long, MAX_ID)),
        };
        let events = vec![
            Event::WriteRequest {
                op: Op::ApplyWrites,
                endpoint: clip(&long, MAX_ID * 2),
                target: target.clone(),
                swap_record: Some(clip(&long, MAX_ID)),
                swap_commit: Some(clip(&long, MAX_ID)),
                bytes: usize::MAX,
                value: Some(Fp::of(&long)),
                writes: Some(
                    (0..MAX_BATCH_OPS)
                        .map(|_| BatchOp {
                            action: BatchAction::Update,
                            collection: clip(&long, MAX_OP_ID),
                            rkey: Some(clip(&long, MAX_OP_ID)),
                            value: Some(Fp::of(&long)),
                        })
                        .collect(),
                ),
                writes_omitted: Some(usize::MAX),
            },
            Event::WriteRefused {
                op: Op::Put,
                target: target.clone(),
                http_status: 400,
                error: Some(clip(&long, MAX_ID)),
                message: Some(clip(&long, MAX_FIELD)),
                elapsed_ms: u64::MAX,
            },
            Event::WriteApplied {
                op: Op::Create,
                target: target.clone(),
                http_status: 200,
                uri: Some(clip(&long, MAX_ID)),
                cid: Some(clip(&long, MAX_ID)),
                results: Some(usize::MAX),
                elapsed_ms: u64::MAX,
            },
            Event::Invocation {
                pid: u32::MAX,
                user: Some(clip(&long, MAX_FIELD)),
                cwd: Some(clip(&long, MAX_FIELD)),
                version: Cow::Borrowed("0.1.0"),
                subcommand: "stack resubmit".into(),
                rotated_from_bytes: Some(u64::MAX),
            },
        ];
        for event in events {
            let line = line_for_test(&event);
            assert!(line.len() < MAX_LINE, "{event:?} is {} bytes", line.len());
        }
    }
}