//! The dead-letter hold for remote changes an apply pass could not land.
//!
//! Before this existed, `apply` folded every unapplied row into one `skipped`
//! counter, `sync` dropped that counter on the floor, and the pull cursor
//! advanced regardless. The server never sends an entry twice, so the row was
//! gone: device A had it, device B did not, and both reported a clean sync.
//!
//! The hold makes the cursor safe to advance. Anything the apply could not write
//! is stored here **as it came off the wire**, not as the conflict layer resolved
//! it, so a retry re-enters [`resolve_pull`](super::hlc::resolve_pull) and is
//! judged against the local state of the moment rather than replaying a decision
//! made against a database that has since moved.
//!
//! Two states, and the difference is whether trying again could ever help:
//!
//! - **Deferred** is retryable: an unknown table (clears when the client is
//!   upgraded) or a constraint violation (clears when the missing parent lands).
//!   Retried automatically at the top of every pull, up to [`MAX_ATTEMPTS`].
//! - **Rejected** is not: a payload with no object, no reconstructable primary
//!   key, no insertable columns. Identical bytes fail identically, so it is held
//!   for display only. A deferred entry that exhausts its attempts is promoted
//!   here rather than retried forever.
//!
//! Rows the schema *intends* to drop (an `exclude_where` predicate,
//! `DeleteMode::Ignore`) are not failures and never reach this table.

use std::collections::HashSet;

use rusqlite::{Connection, OptionalExtension};

use super::apply::{ApplyOutcome, Unapplied};
use crate::error::Result;
use crate::ids::DeviceId;
use crate::types::{ChangeEntry, PulledChange};

/// Retries before a deferred entry is promoted to rejected.
///
/// A constraint violation whose parent genuinely never arrives would otherwise
/// be retried on every pull for the life of the install, and the held set would
/// grow without bound. Five is enough for the ordering cases this exists for (a
/// late parent lands on the next pull, not the fifth) and short enough that a
/// permanently-broken entry stops costing work.
pub const MAX_ATTEMPTS: i64 = 5;

/// DDL for the hold, shared by [`SyncSchema::migration_sql`](super::schema::SyncSchema::migration_sql)
/// and the per-connection upgrade in [`db::ensure_scope_schema`](super::db::ensure_scope_schema),
/// so an existing install gets the table on its next connection open rather than
/// waiting for the app to re-run its migration.
///
/// One row per (scope, table, row_id): a later change to a row that is already
/// held replaces the payload and keeps the attempt count, so a row that keeps
/// failing does not accumulate copies.
pub(crate) const DEFERRED_DDL: &str = "\
CREATE TABLE IF NOT EXISTS sync_deferred (
    scope      TEXT    NOT NULL DEFAULT '',
    table_name TEXT    NOT NULL,
    row_id     TEXT    NOT NULL,
    cause      TEXT    NOT NULL,
    state      TEXT    NOT NULL DEFAULT 'deferred',
    attempts   INTEGER NOT NULL DEFAULT 0,
    seq        INTEGER NOT NULL DEFAULT 0,
    device_id  TEXT    NOT NULL DEFAULT '',
    entry      TEXT    NOT NULL,
    first_seen TEXT    NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
    last_seen  TEXT    NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
    PRIMARY KEY (scope, table_name, row_id)
);
";

/// What a held entry is waiting on.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum HoldState {
    /// Retryable: retried at the top of each pull until [`MAX_ATTEMPTS`].
    Deferred,
    /// Unretryable, or out of attempts. Held for display only.
    Rejected,
}

impl HoldState {
    fn as_str(self) -> &'static str {
        match self {
            Self::Deferred => "deferred",
            Self::Rejected => "rejected",
        }
    }
}

/// One entry in the hold, as listed for a human.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct HeldEntry {
    /// Scope the change belongs to: `""` for personal, otherwise the group id.
    pub scope: String,
    /// Table the change targets.
    pub table: String,
    /// Wire row id of the change.
    pub row_id: String,
    /// Why the apply could not land it.
    pub cause: String,
    /// Whether a retry could still help.
    pub state: HoldState,
    /// Retries spent so far.
    pub attempts: i64,
    /// When the entry was first held (RFC 3339, UTC).
    pub first_seen: String,
    /// The held change's row payload, so a consumer can name the row in terms
    /// its user recognises. `None` for a delete, or a payload that no longer
    /// parses.
    pub payload: Option<serde_json::Value>,
}

/// How much is being held for a scope, for a sync-status surface.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct HoldCounts {
    /// Entries still awaiting an automatic retry.
    pub deferred: u64,
    /// Entries that will not be retried.
    pub rejected: u64,
}

impl HoldCounts {
    /// Total held entries, whatever their state.
    pub fn total(self) -> u64 {
        self.deferred + self.rejected
    }
}

/// Identity of a held row: `(table, row_id)`. Enough to match an apply outcome
/// against the batch it came from, since a resolved entry keeps both.
pub(crate) type RowKey = (String, String);

pub(crate) fn key_of(entry: &ChangeEntry) -> RowKey {
    (entry.table.clone(), entry.row_id.clone())
}

fn key_of_unapplied(row: &Unapplied) -> RowKey {
    (row.table.clone(), row.row_id.clone())
}

/// Load the entries eligible for an automatic retry, as they were pulled.
///
/// Returned in `seq` order so a held batch keeps its original server ordering
/// relative to itself; the caller puts them in front of the newly pulled batch.
pub fn load_retryable(conn: &Connection, scope: &str) -> Result<Vec<PulledChange>> {
    let mut stmt = conn.prepare(
        "SELECT entry, device_id, seq FROM sync_deferred \
         WHERE scope = ?1 AND state = 'deferred' AND attempts < ?2 \
         ORDER BY seq",
    )?;
    let rows = stmt.query_map(rusqlite::params![scope, MAX_ATTEMPTS], |r| {
        Ok((
            r.get::<_, String>(0)?,
            r.get::<_, String>(1)?,
            r.get::<_, i64>(2)?,
        ))
    })?;

    let mut out = Vec::new();
    for row in rows {
        let (entry_json, device, seq) = row?;
        // A payload that will not deserialize is a held entry we can never retry
        // (a downgrade, or a corrupted row). Skip it rather than failing the pull;
        // it stays in the table and stays visible.
        let Ok(entry) = serde_json::from_str::<ChangeEntry>(&entry_json) else {
            tracing::warn!("held entry could not be deserialized, skipping retry");
            continue;
        };
        // The originating device only feeds conflict detection ("did this come
        // from me"); an unparseable one degrades to nil, which is never a live
        // device, rather than failing the retry.
        let device_id =
            uuid::Uuid::parse_str(&device).map_or_else(|_| DeviceId::nil(), DeviceId::new);
        out.push(PulledChange {
            entry,
            device_id,
            seq,
            // A held entry already passed the peer storage gate on the pull that
            // held it, and the hold stores the change rather than the envelope, so
            // there is no stamp to re-read and nothing left to re-check.
            storage_version: None,
        });
    }
    Ok(out)
}

const LIST_COLUMNS: &str =
    "scope, table_name, row_id, cause, state, attempts, first_seen, entry FROM sync_deferred";

fn read_held(r: &rusqlite::Row<'_>) -> rusqlite::Result<HeldEntry> {
    let state: String = r.get(4)?;
    Ok(HeldEntry {
        scope: r.get(0)?,
        table: r.get(1)?,
        row_id: r.get(2)?,
        cause: r.get(3)?,
        state: if state == "rejected" {
            HoldState::Rejected
        } else {
            HoldState::Deferred
        },
        attempts: r.get(5)?,
        first_seen: r.get(6)?,
        // The payload is handed back so a consumer can label the row with
        // something a person recognises (a task's title) instead of a wire row
        // id. A payload that will not parse degrades to None, never an error.
        payload: serde_json::from_str::<ChangeEntry>(&r.get::<_, String>(7)?)
            .ok()
            .and_then(|e| e.data),
    })
}

/// List everything held for a scope, newest first, for a UI surface.
pub fn list(conn: &Connection, scope: &str) -> Result<Vec<HeldEntry>> {
    let mut stmt = conn.prepare(&format!(
        "SELECT {LIST_COLUMNS} WHERE scope = ?1 ORDER BY last_seen DESC"
    ))?;
    collect(stmt.query_map([scope], read_held)?)
}

/// List everything held across every scope, newest first.
///
/// A device syncs its personal scope plus one per group, and a row held in a
/// group scope is just as lost as one held in the personal scope, so a status
/// surface that covered only personal would hide the failures this exists to
/// show.
pub fn list_all(conn: &Connection) -> Result<Vec<HeldEntry>> {
    let mut stmt = conn.prepare(&format!("SELECT {LIST_COLUMNS} ORDER BY last_seen DESC"))?;
    collect(stmt.query_map([], read_held)?)
}

fn collect<I>(rows: I) -> Result<Vec<HeldEntry>>
where
    I: Iterator<Item = rusqlite::Result<HeldEntry>>,
{
    let mut out = Vec::new();
    for row in rows {
        out.push(row?);
    }
    Ok(out)
}

fn read_tally(r: &rusqlite::Row<'_>) -> rusqlite::Result<(String, i64)> {
    Ok((r.get(0)?, r.get(1)?))
}

/// Held counts for a scope.
pub fn counts(conn: &Connection, scope: &str) -> Result<HoldCounts> {
    let mut stmt =
        conn.prepare("SELECT state, COUNT(*) FROM sync_deferred WHERE scope = ?1 GROUP BY state")?;
    tally(stmt.query_map([scope], read_tally)?)
}

/// Held counts across every scope. See [`list_all`].
pub fn counts_all(conn: &Connection) -> Result<HoldCounts> {
    let mut stmt = conn.prepare("SELECT state, COUNT(*) FROM sync_deferred GROUP BY state")?;
    tally(stmt.query_map([], read_tally)?)
}

fn tally<I>(rows: I) -> Result<HoldCounts>
where
    I: Iterator<Item = rusqlite::Result<(String, i64)>>,
{
    let mut out = HoldCounts::default();
    for row in rows {
        let (state, n) = row?;
        let n = u64::try_from(n).unwrap_or(0);
        match state.as_str() {
            "rejected" => out.rejected += n,
            _ => out.deferred += n,
        }
    }
    Ok(out)
}

/// Clear one held entry, by row identity. Returns whether a row was removed.
///
/// The retry path calls this when an entry finally lands; a UI can call it to
/// discard something the user has decided to abandon.
pub fn clear(conn: &Connection, scope: &str, table: &str, row_id: &str) -> Result<bool> {
    let n = conn.execute(
        "DELETE FROM sync_deferred WHERE scope = ?1 AND table_name = ?2 AND row_id = ?3",
        rusqlite::params![scope, table, row_id],
    )?;
    Ok(n > 0)
}

/// Reset a rejected entry so the next pull retries it once more.
///
/// This is the per-row Retry affordance: a rejected entry is not retried
/// automatically, but a human who has fixed the cause (upgraded the client,
/// restored the missing parent by hand) can put it back in the queue.
pub fn requeue(conn: &Connection, scope: &str, table: &str, row_id: &str) -> Result<bool> {
    let n = conn.execute(
        "UPDATE sync_deferred SET state = 'deferred', attempts = 0 \
         WHERE scope = ?1 AND table_name = ?2 AND row_id = ?3",
        rusqlite::params![scope, table, row_id],
    )?;
    Ok(n > 0)
}

/// Reconcile the hold with the outcome of an apply pass.
///
/// `batch` is every entry the pass was given, keyed by row, as pulled. `retried`
/// is the subset that came out of the hold. For each row:
///
/// - unapplied again, and it was retried: spend an attempt, promoting to rejected
///   at [`MAX_ATTEMPTS`];
/// - unapplied for the first time: hold it;
/// - retried and no longer unapplied: it landed (or the schema filtered it), so
///   clear it.
pub(crate) fn settle(
    conn: &Connection,
    scope: &str,
    outcome: &ApplyOutcome,
    batch: &std::collections::HashMap<RowKey, PulledChange>,
    retried: &HashSet<RowKey>,
) -> Result<()> {
    let mut still_failing: HashSet<RowKey> = HashSet::new();

    for row in &outcome.deferred {
        let key = key_of_unapplied(row);
        still_failing.insert(key.clone());
        if retried.contains(&key) {
            spend_attempt(conn, scope, row)?;
        } else {
            hold(conn, scope, row, HoldState::Deferred, batch.get(&key))?;
        }
    }

    for row in &outcome.rejected {
        let key = key_of_unapplied(row);
        still_failing.insert(key.clone());
        // A reject never earns another automatic attempt, whether it is new or a
        // retry that failed the same way again.
        hold(conn, scope, row, HoldState::Rejected, batch.get(&key))?;
    }

    for key in retried {
        if !still_failing.contains(key) {
            clear(conn, scope, &key.0, &key.1)?;
        }
    }
    Ok(())
}

/// Insert or refresh a held row.
fn hold(
    conn: &Connection,
    scope: &str,
    row: &Unapplied,
    state: HoldState,
    pulled: Option<&PulledChange>,
) -> Result<()> {
    let Some(pulled) = pulled else {
        // Every unapplied row is one the caller handed us, so this cannot happen
        // in the pull path. Refuse to hold a row with no payload rather than
        // write an entry no retry could ever use.
        tracing::warn!(
            table = %row.table,
            row_id = %row.row_id,
            "unapplied row has no pulled entry to hold; not recorded"
        );
        return Ok(());
    };
    let entry = serde_json::to_string(&pulled.entry)?;
    conn.execute(
        "INSERT INTO sync_deferred \
             (scope, table_name, row_id, cause, state, seq, device_id, entry) \
         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8) \
         ON CONFLICT(scope, table_name, row_id) DO UPDATE SET \
             cause = excluded.cause, \
             state = excluded.state, \
             seq = excluded.seq, \
             device_id = excluded.device_id, \
             entry = excluded.entry, \
             last_seen = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')",
        rusqlite::params![
            scope,
            row.table,
            row.row_id,
            row.cause,
            state.as_str(),
            pulled.seq,
            pulled.device_id.to_string(),
            entry,
        ],
    )?;
    Ok(())
}

/// Spend one of a held entry's attempts, promoting it to rejected at the cap.
fn spend_attempt(conn: &Connection, scope: &str, row: &Unapplied) -> Result<()> {
    conn.execute(
        "UPDATE sync_deferred SET \
             attempts = attempts + 1, \
             cause = ?4, \
             state = CASE WHEN attempts + 1 >= ?5 THEN 'rejected' ELSE state END, \
             last_seen = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') \
         WHERE scope = ?1 AND table_name = ?2 AND row_id = ?3",
        rusqlite::params![scope, row.table, row.row_id, row.cause, MAX_ATTEMPTS],
    )?;
    Ok(())
}

/// Whether a scope holds anything at all, the cheap check the pull loop uses to
/// decide if an empty batch is worth a retry pass.
pub(crate) fn has_retryable(conn: &Connection, scope: &str) -> Result<bool> {
    let found = conn
        .query_row(
            "SELECT 1 FROM sync_deferred \
             WHERE scope = ?1 AND state = 'deferred' AND attempts < ?2 LIMIT 1",
            rusqlite::params![scope, MAX_ATTEMPTS],
            |_| Ok(()),
        )
        .optional()?;
    Ok(found.is_some())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::types::{ChangeOp, hlc_legacy_floor};
    use std::collections::HashMap;

    fn db() -> Connection {
        let conn = Connection::open_in_memory().unwrap();
        conn.execute_batch(DEFERRED_DDL).unwrap();
        conn
    }

    fn pulled(table: &str, row_id: &str, seq: i64) -> PulledChange {
        PulledChange {
            storage_version: None,
            entry: ChangeEntry {
                table: table.into(),
                op: ChangeOp::Insert,
                row_id: row_id.into(),
                timestamp: chrono::Utc::now(),
                hlc: hlc_legacy_floor(),
                data: Some(serde_json::json!({"id": row_id})),
                extra: serde_json::Map::default(),
            },
            device_id: DeviceId::nil(),
            seq,
        }
    }

    fn unapplied(table: &str, row_id: &str) -> Unapplied {
        Unapplied {
            table: table.into(),
            row_id: row_id.into(),
            cause: "constraint violation".into(),
        }
    }

    fn batch(entries: &[PulledChange]) -> HashMap<RowKey, PulledChange> {
        entries
            .iter()
            .map(|p| (key_of(&p.entry), p.clone()))
            .collect()
    }

    fn deferred_outcome(rows: Vec<Unapplied>) -> ApplyOutcome {
        ApplyOutcome {
            deferred: rows,
            ..ApplyOutcome::default()
        }
    }

    #[test]
    fn a_new_deferred_row_is_held_with_its_pulled_entry() {
        let conn = db();
        let p = pulled("child", "c1", 7);
        settle(
            &conn,
            "",
            &deferred_outcome(vec![unapplied("child", "c1")]),
            &batch(&[p]),
            &HashSet::new(),
        )
        .unwrap();

        let held = load_retryable(&conn, "").unwrap();
        assert_eq!(held.len(), 1);
        assert_eq!(held[0].entry.row_id, "c1");
        assert_eq!(held[0].seq, 7, "the wire seq is preserved for ordering");
        assert_eq!(counts(&conn, "").unwrap().deferred, 1);
    }

    #[test]
    fn a_retry_that_lands_clears_the_hold() {
        let conn = db();
        let p = pulled("child", "c1", 7);
        let b = batch(&[p]);
        settle(
            &conn,
            "",
            &deferred_outcome(vec![unapplied("child", "c1")]),
            &b,
            &HashSet::new(),
        )
        .unwrap();

        let retried: HashSet<RowKey> = [("child".to_string(), "c1".to_string())]
            .into_iter()
            .collect();
        settle(&conn, "", &ApplyOutcome::default(), &b, &retried).unwrap();

        assert!(load_retryable(&conn, "").unwrap().is_empty());
        assert_eq!(counts(&conn, "").unwrap().total(), 0);
    }

    #[test]
    fn attempts_are_capped_and_the_entry_is_promoted_to_rejected() {
        let conn = db();
        let p = pulled("child", "c1", 7);
        let b = batch(&[p]);
        let outcome = deferred_outcome(vec![unapplied("child", "c1")]);
        settle(&conn, "", &outcome, &b, &HashSet::new()).unwrap();

        let retried: HashSet<RowKey> = [("child".to_string(), "c1".to_string())]
            .into_iter()
            .collect();
        for _ in 0..MAX_ATTEMPTS {
            settle(&conn, "", &outcome, &b, &retried).unwrap();
        }

        assert!(
            load_retryable(&conn, "").unwrap().is_empty(),
            "a capped entry is no longer retried"
        );
        let listed = list(&conn, "").unwrap();
        assert_eq!(listed.len(), 1, "but it stays visible");
        assert_eq!(listed[0].state, HoldState::Rejected);
        assert_eq!(listed[0].attempts, MAX_ATTEMPTS);
        assert!(!has_retryable(&conn, "").unwrap());
    }

    #[test]
    fn requeue_gives_a_rejected_entry_one_more_run() {
        let conn = db();
        let p = pulled("child", "c1", 7);
        settle(
            &conn,
            "",
            &ApplyOutcome {
                rejected: vec![unapplied("child", "c1")],
                ..ApplyOutcome::default()
            },
            &batch(&[p]),
            &HashSet::new(),
        )
        .unwrap();
        assert!(load_retryable(&conn, "").unwrap().is_empty());

        assert!(requeue(&conn, "", "child", "c1").unwrap());
        assert_eq!(load_retryable(&conn, "").unwrap().len(), 1);
    }

    #[test]
    fn holds_are_partitioned_by_scope() {
        let conn = db();
        let p = pulled("child", "c1", 7);
        let b = batch(&[p]);
        let outcome = deferred_outcome(vec![unapplied("child", "c1")]);
        settle(&conn, "", &outcome, &b, &HashSet::new()).unwrap();
        settle(&conn, "group-a", &outcome, &b, &HashSet::new()).unwrap();

        assert_eq!(counts(&conn, "").unwrap().deferred, 1);
        assert_eq!(counts(&conn, "group-a").unwrap().deferred, 1);
        assert!(clear(&conn, "", "child", "c1").unwrap());
        assert_eq!(counts(&conn, "").unwrap().deferred, 0);
        assert_eq!(
            counts(&conn, "group-a").unwrap().deferred,
            1,
            "clearing one scope leaves the other alone"
        );
    }

    /// An outcome holding both kinds at once, for the count arithmetic.
    fn mixed_outcome(deferred: Vec<Unapplied>, rejected: Vec<Unapplied>) -> ApplyOutcome {
        ApplyOutcome {
            rejected,
            deferred,
            ..ApplyOutcome::default()
        }
    }

    #[test]
    fn hold_counts_total_adds_the_two_states() {
        let conn = db();
        // Two of one and three of the other: any arithmetic other than a sum
        // lands somewhere else (a difference underflows, a product gives six).
        let entries: Vec<PulledChange> = (0..5)
            .map(|i| pulled("child", &format!("c{i}"), i))
            .collect();
        settle(
            &conn,
            "",
            &mixed_outcome(
                vec![unapplied("child", "c0"), unapplied("child", "c1")],
                vec![
                    unapplied("child", "c2"),
                    unapplied("child", "c3"),
                    unapplied("child", "c4"),
                ],
            ),
            &batch(&entries),
            &HashSet::new(),
        )
        .unwrap();

        let counts = counts(&conn, "").unwrap();
        assert_eq!(counts.deferred, 2);
        assert_eq!(counts.rejected, 3);
        assert_eq!(counts.total(), 5);
        assert_eq!(list(&conn, "").unwrap().len(), 5);
    }

    #[test]
    fn the_all_scope_reads_see_every_scope_at_once() {
        // A row held in a group scope is as lost as one held in the personal
        // scope, so the status surface reads across both. Different counts per
        // scope, so neither read can pass by looking at one of them twice.
        let conn = db();
        let personal = pulled("child", "p1", 1);
        let group = [pulled("child", "g1", 2), pulled("child", "g2", 3)];
        settle(
            &conn,
            "",
            &deferred_outcome(vec![unapplied("child", "p1")]),
            &batch(&[personal]),
            &HashSet::new(),
        )
        .unwrap();
        settle(
            &conn,
            "group-a",
            &mixed_outcome(
                vec![unapplied("child", "g1")],
                vec![unapplied("child", "g2")],
            ),
            &batch(&group),
            &HashSet::new(),
        )
        .unwrap();

        let all = counts_all(&conn).unwrap();
        assert_eq!(all.deferred, 2, "the personal one plus the group's");
        assert_eq!(all.rejected, 1);
        assert_eq!(all.total(), 3);

        let listed = list_all(&conn).unwrap();
        assert_eq!(listed.len(), 3);
        let mut scopes: Vec<&str> = listed.iter().map(|e| e.scope.as_str()).collect();
        scopes.sort_unstable();
        scopes.dedup();
        assert_eq!(scopes, ["", "group-a"], "both scopes are represented");

        // The per-scope reads still see only their own, which is what makes the
        // pair of reads worth having.
        assert_eq!(counts(&conn, "").unwrap().total(), 1);
        assert_eq!(list(&conn, "group-a").unwrap().len(), 2);
    }

    #[test]
    fn clear_and_requeue_report_whether_the_row_was_there() {
        // Both return the rows-affected of their statement, and a caller uses it
        // to tell "retried" from "there was nothing to retry".
        let conn = db();
        settle(
            &conn,
            "",
            &ApplyOutcome {
                rejected: vec![unapplied("child", "c1")],
                ..ApplyOutcome::default()
            },
            &batch(&[pulled("child", "c1", 7)]),
            &HashSet::new(),
        )
        .unwrap();

        assert!(
            !requeue(&conn, "", "child", "absent").unwrap(),
            "no such row id"
        );
        assert!(
            !requeue(&conn, "", "other-scope", "c1").unwrap(),
            "no such table"
        );
        assert!(
            !clear(&conn, "elsewhere", "child", "c1").unwrap(),
            "no such scope"
        );
        assert_eq!(
            counts(&conn, "").unwrap().total(),
            1,
            "none of that touched the held row"
        );

        assert!(requeue(&conn, "", "child", "c1").unwrap());
        assert!(clear(&conn, "", "child", "c1").unwrap());
        assert!(
            !clear(&conn, "", "child", "c1").unwrap(),
            "the second clear finds nothing left"
        );
        assert_eq!(counts(&conn, "").unwrap().total(), 0);
    }

    #[test]
    fn a_repeat_failure_replaces_the_payload_without_duplicating_the_row() {
        let conn = db();
        let first = pulled("child", "c1", 7);
        let second = pulled("child", "c1", 9);
        let outcome = deferred_outcome(vec![unapplied("child", "c1")]);
        settle(&conn, "", &outcome, &batch(&[first]), &HashSet::new()).unwrap();
        settle(&conn, "", &outcome, &batch(&[second]), &HashSet::new()).unwrap();

        let held = load_retryable(&conn, "").unwrap();
        assert_eq!(held.len(), 1);
        assert_eq!(held[0].seq, 9, "the newer entry wins");
    }
}