//! Push: send local changes to the server and apply what it says (M10.7c).
//!
//! Three sources feed a push: rows flagged `dirty` (created or edited locally), rows
//! in `pending_deletes` (permanently deleted locally — see `local::schema` v2 for why
//! a delete needs its own record), and files attached on this device that are still
//! waiting to go up (`attachments.uploaded = 0`, schema v10).
//!
//! Sync is **whole-note**: an upsert carries the client's full current state, not a
//! patch (docs/sync.md). The server resolves conflicts last-write-wins by the client's
//! `edited_at`, snapshotting anything it overwrites into the note's version history.

use rusqlite::{params, Connection};
use serde::{Deserialize, Serialize};

use super::blobs::BlobStore;
use super::client::{self, UploadError};
use super::state;
use crate::local::models::{access, entity};
use crate::local::Db;

/// The server rejects a batch larger than this (`MAX_PUSH` in `sync.py`).
const BATCH: usize = 500;

/// Backstop: a batch whose results never clear `dirty` would loop forever.
const MAX_BATCHES: usize = 10_000;

#[derive(Debug, Clone, Default, Serialize, PartialEq, Eq)]
pub struct PushSummary {
    pub batches: usize,
    pub sent: usize,
    pub created: usize,
    pub applied: usize,
    /// The server had a newer edit and kept it. Not a failure — the local row stops
    /// being dirty and the following pull adopts the server's version.
    pub kept: usize,
    pub noop: usize,
    /// Still dirty, and surfaced: these need a human (a duplicate label name is the
    /// realistic case). Silently retrying forever would be the wrong shape.
    pub rejected: usize,
    pub errors: Vec<String>,
    /// Files attached on this device that reached the server this cycle.
    pub uploaded: usize,
    /// Files that didn't. Their reasons are in `errors`; one the server refused for
    /// good also carries its reason on the attachment and isn't tried again.
    pub upload_failed: usize,
}

impl PushSummary {
    fn absorb(&mut self, other: PushSummary) {
        self.batches += other.batches;
        self.sent += other.sent;
        self.created += other.created;
        self.applied += other.applied;
        self.kept += other.kept;
        self.noop += other.noop;
        self.rejected += other.rejected;
        self.errors.extend(other.errors);
        self.uploaded += other.uploaded;
        self.upload_failed += other.upload_failed;
    }
}

// --- outgoing shapes ---------------------------------------------------------

/// A change's operation on the wire.
const UPSERT: &str = "upsert";
const DELETE: &str = "delete";

/// One entry in the `changes` array. Notes and labels share the envelope; serde skips
/// the fields that don't apply, so the server sees exactly the shape docs/sync.md
/// describes for each entity.
#[derive(Debug, Default, Serialize)]
pub struct Change {
    pub entity: &'static str,
    pub id: String,
    pub op: &'static str,
    pub edited_at: String,

    #[serde(skip_serializing_if = "Option::is_none")]
    pub body: Option<String>,
    /// A LABEL's colour. A note has none since M315, so a note change leaves this
    /// `None` and the key never reaches the wire.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub color: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub pinned: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub archived: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub trashed: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub remind_at: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub recurrence: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub position: Option<i64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub label_ids: Option<Vec<String>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub created_at: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub name: Option<String>,
    /// When this account last pinned, archived or moved a note someone else owns
    /// (#5176): `pinned`, `archived` and `position` on such a note are its own, and
    /// the server weighs them against this rather than `edited_at`, which stays the
    /// time of the text.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub state_at: Option<String>,
}

impl Change {
    fn delete(entity: &'static str, id: String, edited_at: String) -> Self {
        Change {
            entity,
            id,
            op: DELETE,
            edited_at,
            ..Default::default()
        }
    }
}

// --- incoming results --------------------------------------------------------

#[derive(Debug, Deserialize)]
struct PushResponse {
    #[serde(default)]
    results: Vec<PushResult>,
}

#[derive(Debug, Clone, Deserialize)]
pub struct PushResult {
    #[serde(default)]
    pub id: Option<String>,
    #[serde(default)]
    pub entity: Option<String>,
    #[serde(default)]
    pub status: String,
    #[serde(default)]
    pub sync_revision: Option<i64>,
    #[serde(default)]
    pub error: Option<String>,
}

// --- collecting --------------------------------------------------------------

/// What the server takes beyond the base protocol, read from the features it
/// advertises. What it doesn't take stays queued here, untouched, until a server
/// that does: an older one would refuse it, or worse, accept it and keep nothing.
#[derive(Debug, Clone, Copy, Default)]
pub struct Accepts {
    /// Attachment and preview removals, and file uploads (`attachment_sync`). An
    /// older server would answer "unknown entity" and the removal would read as a
    /// failure.
    pub attachment_sync: bool,
    /// This account's own pin, archive and position on a note someone else owns
    /// (`shared_state`, #5176).
    pub shared_state: bool,
}

impl Accepts {
    /// Everything this client can send.
    pub const ALL: Accepts = Accepts {
        attachment_sync: true,
        shared_state: true,
    };
}

/// Everything waiting to go up, oldest edit first so a truncated batch still makes
/// forward progress in a sensible order.
pub fn collect(conn: &Connection, limit: usize, accepts: Accepts) -> rusqlite::Result<Vec<Change>> {
    let mut out = Vec::new();
    collect_deletes(conn, &mut out, limit, accepts.attachment_sync)?;
    if out.len() < limit {
        collect_labels(conn, &mut out, limit)?;
    }
    if out.len() < limit {
        collect_notes(conn, &mut out, limit, accepts.shared_state)?;
    }
    Ok(out)
}

fn collect_deletes(
    conn: &Connection,
    out: &mut Vec<Change>,
    limit: usize,
    attachment_sync: bool,
) -> rusqlite::Result<()> {
    // Filtered in SQL rather than skipped below, so held-back removals can't use up
    // the LIMIT and starve the deletes that can go.
    let mut stmt = conn.prepare(
        "SELECT entity, id, deleted_at FROM pending_deletes
          WHERE entity IN ('note', 'label')
             OR (?2 AND entity IN ('attachment', 'preview'))
          ORDER BY deleted_at LIMIT ?1",
    )?;
    let rows = stmt.query_map(params![limit as i64, attachment_sync], |r| {
        Ok((
            r.get::<_, String>(0)?,
            r.get::<_, String>(1)?,
            r.get::<_, String>(2)?,
        ))
    })?;
    for row in rows {
        let (entity, id, deleted_at) = row?;
        // Only these exist on the wire; anything else is a bug in a writer, and
        // shipping it would earn a rejection that no retry could clear.
        let entity: &'static str = match entity.as_str() {
            entity::NOTE => entity::NOTE,
            entity::LABEL => entity::LABEL,
            entity::ATTACHMENT => entity::ATTACHMENT,
            entity::PREVIEW => entity::PREVIEW,
            _ => continue,
        };
        out.push(Change::delete(entity, id, deleted_at));
    }
    Ok(())
}

fn collect_labels(conn: &Connection, out: &mut Vec<Change>, limit: usize) -> rusqlite::Result<()> {
    let remaining = limit.saturating_sub(out.len());
    let mut stmt = conn.prepare(
        "SELECT id, name, color, updated_at FROM labels
          WHERE dirty = 1 ORDER BY updated_at LIMIT ?1",
    )?;
    let rows = stmt.query_map(params![remaining as i64], |r| {
        Ok(Change {
            entity: entity::LABEL,
            id: r.get(0)?,
            op: UPSERT,
            name: Some(r.get(1)?),
            color: Some(r.get(2)?),
            edited_at: r.get(3)?,
            ..Default::default()
        })
    })?;
    for row in rows {
        out.push(row?);
    }
    Ok(())
}

fn collect_notes(
    conn: &Connection,
    out: &mut Vec<Change>,
    limit: usize,
    shared_state: bool,
) -> rusqlite::Result<()> {
    let remaining = limit.saturating_sub(out.len());
    let ids: Vec<String> = {
        // A note shared with us to view has only this account's own pin, archive and
        // place to send, and only to a server that keeps them. Without one, a dirty
        // view note has nothing the server would take, and the next pull puts the
        // server's copy back over it.
        let mut stmt = conn.prepare(
            "SELECT id FROM notes
              WHERE dirty = 1
                AND (permission <> 'view' OR (?2 AND state_at IS NOT NULL))
              ORDER BY updated_at LIMIT ?1",
        )?;
        let rows = stmt.query_map(params![remaining as i64, shared_state], |r| {
            r.get::<_, String>(0)
        })?;
        rows.collect::<rusqlite::Result<Vec<String>>>()?
    };
    for id in ids {
        out.push(note_change(conn, &id, shared_state)?);
    }
    Ok(())
}

/// The note's own columns. A named struct rather than a twelve-wide tuple so the
/// field-to-column mapping stays readable at the call site.
struct NoteRow {
    body: String,
    position: i64,
    pinned: bool,
    archived: bool,
    trashed: bool,
    remind_at: Option<String>,
    recurrence: Option<String>,
    created_at: String,
    updated_at: String,
    /// `owner`, or `edit` or `view` for a note someone shared with us (#5175).
    permission: String,
    /// When this account last pinned, archived or moved it, if it isn't ours (#5176).
    state_at: Option<String>,
}

fn note_row(conn: &Connection, id: &str) -> rusqlite::Result<NoteRow> {
    conn.query_row(
        "SELECT body, position, pinned, archived, trashed,
                remind_at, recurrence, created_at, updated_at, permission, state_at
           FROM notes WHERE id = ?1",
        params![id],
        |r| {
            Ok(NoteRow {
                body: r.get(0)?,
                position: r.get(1)?,
                pinned: r.get::<_, i64>(2)? != 0,
                archived: r.get::<_, i64>(3)? != 0,
                trashed: r.get::<_, i64>(4)? != 0,
                remind_at: r.get(5)?,
                recurrence: r.get(6)?,
                created_at: r.get(7)?,
                updated_at: r.get(8)?,
                permission: r.get(9)?,
                state_at: r.get(10)?,
            })
        },
    )
}

fn note_change(conn: &Connection, id: &str, shared_state: bool) -> rusqlite::Result<Change> {
    let row = note_row(conn, id)?;

    // Someone else's note: its text if it was shared to edit, and this account's own
    // pin, archive and place once they have been changed here, to a server that keeps
    // them. Trash, reminders and labels are the owner's, and this account's labels
    // were never on it.
    if row.permission != access::OWNER {
        let state_at = row.state_at.filter(|_| shared_state);
        let own = state_at.is_some();
        return Ok(Change {
            entity: entity::NOTE,
            id: id.to_string(),
            op: UPSERT,
            edited_at: row.updated_at,
            body: (row.permission == access::EDIT).then_some(row.body),
            pinned: own.then_some(row.pinned),
            archived: own.then_some(row.archived),
            position: own.then_some(row.position),
            state_at,
            ..Default::default()
        });
    }

    // MANUAL memberships only. Tag-sourced ones (`via_tag = 1`) are re-derived by the
    // server from the body; sending them as label_ids would convert them into manual
    // assignments that no longer disappear when the #tag is removed from the text.
    let label_ids = {
        let mut stmt =
            conn.prepare("SELECT label_id FROM note_labels WHERE note_id = ?1 AND via_tag = 0")?;
        let rows = stmt.query_map(params![id], |r| r.get::<_, String>(0))?;
        rows.collect::<rusqlite::Result<Vec<String>>>()?
    };

    Ok(Change {
        entity: entity::NOTE,
        id: id.to_string(),
        op: UPSERT,
        // The local `updated_at` IS the client's edit time, which is what the
        // server's last-write-wins comparison runs against.
        edited_at: row.updated_at,
        body: Some(row.body),
        pinned: Some(row.pinned),
        archived: Some(row.archived),
        trashed: Some(row.trashed),
        remind_at: row.remind_at,
        recurrence: row.recurrence,
        position: Some(row.position),
        label_ids: Some(label_ids),
        created_at: Some(row.created_at),
        ..Default::default()
    })
}

// --- applying results --------------------------------------------------------

/// Fold one batch's results back into the local store, atomically.
pub fn apply_results(
    conn: &Connection,
    sent: &[Change],
    results: &[PushResult],
) -> rusqlite::Result<PushSummary> {
    let tx = conn.unchecked_transaction()?;
    let mut summary = PushSummary {
        batches: 1,
        sent: sent.len(),
        ..Default::default()
    };
    // The server answers positionally, one result per change. Zip rather than trust
    // the echoed id: a rejected malformed entry may carry no id at all.
    let mut lowest_kept: Option<i64> = None;

    for (change, result) in sent.iter().zip(results.iter()) {
        match result.status.as_str() {
            "created" | "applied" => {
                clear_dirty(&tx, change, result.sync_revision)?;
                if result.status == "created" {
                    summary.created += 1;
                } else {
                    summary.applied += 1;
                }
                if change.op == DELETE {
                    forget_pending_delete(&tx, change)?;
                }
            }
            "noop" => {
                // The server had nothing to do — typically a delete for a row it
                // never saw (created and deleted while offline).
                clear_dirty(&tx, change, result.sync_revision)?;
                forget_pending_delete(&tx, change)?;
                summary.noop += 1;
            }
            "kept" => {
                // The server's version is newer. Stop being dirty — re-pushing would
                // lose to the same comparison forever — and let the next pull bring
                // the server's copy down.
                clear_dirty(&tx, change, None)?;
                if change.op == DELETE {
                    // Our delete lost to a newer server edit; the note lives on, and
                    // the pull will restore it locally. Drop the tombstone so we
                    // don't keep trying to delete a note the user has since edited.
                    forget_pending_delete(&tx, change)?;
                }
                if let Some(revision) = result.sync_revision {
                    lowest_kept = Some(lowest_kept.map_or(revision, |c: i64| c.min(revision)));
                }
                summary.kept += 1;
            }
            _ => {
                // "rejected" and anything unrecognized: leave the row dirty so it is
                // retried, and surface the reason. A duplicate label name is the
                // realistic case and only a human can resolve it.
                summary.rejected += 1;
                let reason = result
                    .error
                    .clone()
                    .unwrap_or_else(|| result.status.clone());
                summary
                    .errors
                    .push(format!("{} {}: {reason}", change.entity, change.id));
            }
        }
    }

    // A `kept` result means the server holds a version we have not seen. Normally its
    // revision is above our cursor and the next pull fetches it anyway. If it is NOT
    // — which happens when a skewed clock makes a genuinely later local edit look
    // older — rewind so that note is re-fetched. Without this the local edit is
    // dropped from sync and the stale copy stays on screen with nothing marking it.
    if let Some(revision) = lowest_kept {
        let current = state::read(&tx)?.last_cursor;
        if revision <= current {
            state::set_cursor(&tx, (revision - 1).max(0))?;
        }
    }

    tx.commit()?;
    Ok(summary)
}

fn clear_dirty(conn: &Connection, change: &Change, revision: Option<i64>) -> rusqlite::Result<()> {
    // A delete has no local row left to update.
    if change.op == DELETE {
        return Ok(());
    }
    let table = match change.entity {
        entity::LABEL => "labels",
        _ => "notes",
    };
    match revision {
        Some(rev) => conn.execute(
            &format!("UPDATE {table} SET dirty = 0, sync_revision = ?2 WHERE id = ?1"),
            params![change.id, rev],
        )?,
        None => conn.execute(
            &format!("UPDATE {table} SET dirty = 0 WHERE id = ?1"),
            params![change.id],
        )?,
    };
    Ok(())
}

fn forget_pending_delete(conn: &Connection, change: &Change) -> rusqlite::Result<()> {
    if change.op != DELETE {
        return Ok(());
    }
    conn.execute(
        "DELETE FROM pending_deletes WHERE entity = ?1 AND id = ?2",
        params![change.entity, change.id],
    )?;
    Ok(())
}

/// True when anything is waiting to go up. Cheap enough to call before a cycle.
pub fn has_pending(conn: &Connection) -> rusqlite::Result<bool> {
    // One definition of "pending": the fingerprint's. Counting where a LIMIT 1 would
    // do costs nothing on a local store, and two lists of the same four predicates
    // would drift.
    Ok(pending_fingerprint(conn)?.is_some())
}

/// A value that changes whenever the set of pending changes does, and stays put
/// while it doesn't. `None` when nothing is pending.
///
/// For a background syncer deciding whether there is anything NEW to send.
/// `has_pending` cannot answer that: a change the server rejected stays dirty on
/// purpose (it needs a person), so "something is pending" stays true after every
/// cycle, and a loop keyed on it would resend the same rejected row forever. Any
/// local write moves a count or stamps a later `updated_at`, so it moves this too.
pub fn pending_fingerprint(conn: &Connection) -> rusqlite::Result<Option<String>> {
    let fingerprint: String = conn.query_row(
        "SELECT (SELECT COUNT(*) || ':' || IFNULL(MAX(updated_at), '') FROM notes WHERE dirty = 1)
             || '|' || (SELECT COUNT(*) || ':' || IFNULL(MAX(updated_at), '') FROM labels WHERE dirty = 1)
             || '|' || (SELECT COUNT(*) || ':' || IFNULL(MAX(deleted_at), '') FROM pending_deletes)
             || '|' || (SELECT COUNT(*) FROM attachments WHERE uploaded = 0 AND upload_error IS NULL)",
        [],
        |r| r.get(0),
    )?;
    Ok((fingerprint != "0:|0:|0:|0").then_some(fingerprint))
}

/// A file attached on this device, ready to go up.
#[derive(Debug, PartialEq, Eq)]
pub struct PendingUpload {
    pub note_id: String,
    pub id: String,
    pub filename: String,
    pub mime: String,
    pub sha256: String,
}

/// Files waiting to upload whose note the server already holds. A note still dirty
/// after the push (its change was rejected) keeps its files back too: the server files
/// an attachment under its note, and would answer 404 for one it doesn't have.
pub fn pending_uploads(conn: &Connection) -> rusqlite::Result<Vec<PendingUpload>> {
    let mut stmt = conn.prepare(
        "SELECT a.note_id, a.id, IFNULL(a.filename, 'file'), a.mime, a.sha256
           FROM attachments a JOIN notes n ON n.id = a.note_id
          WHERE a.uploaded = 0 AND a.upload_error IS NULL AND a.sha256 IS NOT NULL
            AND n.dirty = 0
          ORDER BY a.note_id, a.position",
    )?;
    let rows = stmt.query_map([], |r| {
        Ok(PendingUpload {
            note_id: r.get(0)?,
            id: r.get(1)?,
            filename: r.get(2)?,
            mime: r.get(3)?,
            sha256: r.get(4)?,
        })
    })?;
    rows.collect()
}

/// Record what became of one upload.
fn settle_upload(
    conn: &Connection,
    id: &str,
    result: &Result<(), UploadError>,
) -> rusqlite::Result<()> {
    match result {
        Ok(()) => conn.execute(
            "UPDATE attachments SET uploaded = 1, upload_error = NULL WHERE id = ?1",
            params![id],
        )?,
        Err(UploadError::Refused(reason)) => conn.execute(
            "UPDATE attachments SET upload_error = ?2 WHERE id = ?1",
            params![id, reason],
        )?,
        Err(UploadError::Retry(_)) => 0,
    };
    Ok(())
}

/// Send the bytes of every file attached here whose note has landed.
///
/// One file failing never fails the cycle, the same as a download: the notes have
/// already gone, and one unreachable or oversized file must not hold up every sync
/// after it. A refusal is recorded on the attachment instead, and a passing failure
/// is simply tried again next cycle.
async fn upload_pending(
    db: &Db,
    blobs: &BlobStore,
    base_url: &str,
    token: &str,
) -> Result<PushSummary, String> {
    let wanted = {
        let conn = db.conn()?;
        pending_uploads(&conn).map_err(|e| e.to_string())?
    };
    let mut summary = PushSummary::default();
    for upload in wanted {
        let result = match blobs.read(&upload.sha256) {
            Some(bytes) => {
                client::upload_attachment(
                    base_url,
                    token,
                    &upload.note_id,
                    &upload.id,
                    &upload.filename,
                    &upload.mime,
                    &upload.sha256,
                    bytes,
                )
                .await
            }
            // Nothing to send and nothing that could bring it back: the file was
            // only ever on this device.
            None => Err(UploadError::Refused(
                "the file's bytes are missing from this device".to_string(),
            )),
        };
        {
            let conn = db.conn()?;
            settle_upload(&conn, &upload.id, &result).map_err(|e| e.to_string())?;
        }
        match result {
            Ok(()) => summary.uploaded += 1,
            Err(UploadError::Refused(reason)) | Err(UploadError::Retry(reason)) => {
                log::warn!("attachment {}: {reason}", upload.id);
                summary.upload_failed += 1;
                summary
                    .errors
                    .push(format!("{} didn't upload: {reason}", upload.filename));
            }
        }
    }
    Ok(summary)
}

/// Send everything pending, in batches, applying each batch's results before the
/// next is collected — then the files, once their notes are there to hold them.
///
/// `accepts`: what the server advertises (see [`Accepts`]). Without
/// `attachment_sync`, uploads wait too.
pub async fn run(
    db: &Db,
    blobs: &BlobStore,
    base_url: &str,
    token: &str,
    accepts: Accepts,
) -> Result<PushSummary, String> {
    let mut total = PushSummary::default();

    loop {
        let batch = {
            let conn = db.conn()?;
            collect(&conn, BATCH, accepts).map_err(|e| e.to_string())?
        };
        if batch.is_empty() {
            break;
        }

        let raw = client::push_changes(base_url, token, &batch).await?;
        let results = parse_results(&raw)?;

        let applied = {
            let conn = db.conn()?;
            apply_results(&conn, &batch, &results).map_err(|e| e.to_string())?
        };
        // Everything rejected clears nothing, so the same batch would be collected
        // again forever. Stop and report instead.
        let progressed = applied.rejected < applied.sent;
        total.absorb(applied);

        if !progressed {
            break;
        }
        if total.batches >= MAX_BATCHES {
            return Err(format!(
                "Stopped after {MAX_BATCHES} push batches without draining the queue."
            ));
        }
    }

    if accepts.attachment_sync {
        total.absorb(upload_pending(db, blobs, base_url, token).await?);
    }

    if total.rejected > 0 {
        log::warn!(
            "push: {} change(s) rejected by the server: {}",
            total.rejected,
            total.errors.join("; ")
        );
    }
    log::info!(
        "push complete: {} sent ({} created, {} applied, {} kept, {} noop, {} rejected), \
         {} file(s) uploaded, {} not",
        total.sent,
        total.created,
        total.applied,
        total.kept,
        total.noop,
        total.rejected,
        total.uploaded,
        total.upload_failed
    );
    Ok(total)
}

/// Parse the server's reply. Kept next to the shapes it produces.
pub fn parse_results(raw: &str) -> Result<Vec<PushResult>, String> {
    let parsed: PushResponse =
        serde_json::from_str(raw).map_err(|e| format!("Couldn't read the push response: {e}"))?;
    Ok(parsed.results)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::local::store;

    fn db() -> Connection {
        crate::local::memory_conn().expect("in-memory db")
    }

    fn seed_note(conn: &Connection, id: &str, dirty: i64) {
        conn.execute(
            "INSERT INTO notes (id, body, position, pinned, archived,
                                trashed, created_at, updated_at, sync_revision, dirty)
                  VALUES (?1, 'B', 0, 0, 0, 0,
                          '2026-07-26T00:00:00.000Z', '2026-07-26T00:00:00.000Z', 3, ?2)",
            params![id, dirty],
        )
        .expect("seed note");
    }

    fn ok(status: &str, revision: Option<i64>) -> PushResult {
        PushResult {
            id: None,
            entity: None,
            status: status.to_string(),
            sync_revision: revision,
            error: None,
        }
    }

    fn dirty_count(conn: &Connection) -> i64 {
        conn.query_row("SELECT COUNT(*) FROM notes WHERE dirty = 1", [], |r| {
            r.get(0)
        })
        .expect("count")
    }

    #[test]
    fn collects_only_dirty_notes() {
        let conn = db();
        seed_note(&conn, "clean", 0);
        seed_note(&conn, "dirty", 1);
        let batch = collect(&conn, 100, Accepts::ALL).expect("collect");
        assert_eq!(batch.len(), 1);
        assert_eq!(batch[0].id, "dirty");
        assert_eq!(batch[0].op, "upsert");
    }

    #[test]
    fn sends_only_manual_label_memberships() {
        // Tag-sourced labels are re-derived server-side. Sending them as label_ids
        // would convert them to manual assignments that survive removing the #tag.
        let conn = db();
        seed_note(&conn, "n1", 1);
        for (id, name, via_tag) in [("manual", "Manual", 0), ("tagged", "Tagged", 1)] {
            conn.execute(
                "INSERT INTO labels (id, name, color, created_at, updated_at, dirty)
                      VALUES (?1, ?2, 'default', '2026-01-01', '2026-01-01', 0)",
                params![id, name],
            )
            .expect("seed label");
            conn.execute(
                "INSERT INTO note_labels (note_id, label_id, via_tag) VALUES ('n1', ?1, ?2)",
                params![id, via_tag],
            )
            .expect("seed membership");
        }
        let batch = collect(&conn, 100, Accepts::ALL).expect("collect");
        let note = batch.iter().find(|c| c.entity == "note").expect("note");
        assert_eq!(note.label_ids.as_deref(), Some(&["manual".to_string()][..]));
    }

    #[test]
    fn a_local_delete_becomes_a_delete_change() {
        let conn = db();
        seed_note(&conn, "n1", 0);
        store::delete_forever(&conn, "n1").expect("delete");
        let batch = collect(&conn, 100, Accepts::ALL).expect("collect");
        assert_eq!(batch.len(), 1);
        assert_eq!(batch[0].op, "delete");
        assert_eq!(batch[0].entity, "note");
        assert_eq!(batch[0].id, "n1");
    }

    #[test]
    fn applied_clears_dirty_and_records_the_revision() {
        let conn = db();
        seed_note(&conn, "n1", 1);
        let batch = collect(&conn, 100, Accepts::ALL).expect("collect");
        apply_results(&conn, &batch, &[ok("applied", Some(42))]).expect("apply");
        assert_eq!(dirty_count(&conn), 0);
        let rev: i64 = conn
            .query_row("SELECT sync_revision FROM notes WHERE id = 'n1'", [], |r| {
                r.get(0)
            })
            .expect("revision");
        assert_eq!(rev, 42);
    }

    #[test]
    fn kept_clears_dirty_so_it_is_not_pushed_forever() {
        // The server has a newer edit. Re-pushing would lose the same comparison
        // every time; the following pull adopts the server's version instead.
        let conn = db();
        seed_note(&conn, "n1", 1);
        let batch = collect(&conn, 100, Accepts::ALL).expect("collect");
        let summary = apply_results(&conn, &batch, &[ok("kept", Some(99))]).expect("apply");
        assert_eq!(summary.kept, 1);
        assert_eq!(dirty_count(&conn), 0);
    }

    #[test]
    fn kept_rewinds_the_cursor_when_the_server_version_is_already_behind_it() {
        // Clock skew: a genuinely later local edit can look older, so the server
        // keeps its copy at a revision we have ALREADY consumed. Without a rewind the
        // next pull skips it and the stale local copy stays on screen silently.
        let conn = db();
        seed_note(&conn, "n1", 1);
        state::set_cursor(&conn, 100).expect("cursor");
        let batch = collect(&conn, 100, Accepts::ALL).expect("collect");
        apply_results(&conn, &batch, &[ok("kept", Some(40))]).expect("apply");
        assert_eq!(state::read(&conn).expect("state").last_cursor, 39);
    }

    #[test]
    fn kept_leaves_the_cursor_alone_when_the_server_version_is_ahead() {
        let conn = db();
        seed_note(&conn, "n1", 1);
        state::set_cursor(&conn, 10).expect("cursor");
        let batch = collect(&conn, 100, Accepts::ALL).expect("collect");
        apply_results(&conn, &batch, &[ok("kept", Some(40))]).expect("apply");
        assert_eq!(
            state::read(&conn).expect("state").last_cursor,
            10,
            "the pending pull already covers it"
        );
    }

    #[test]
    fn rejected_stays_dirty_and_is_reported() {
        let conn = db();
        seed_note(&conn, "n1", 1);
        let batch = collect(&conn, 100, Accepts::ALL).expect("collect");
        let mut bad = ok("rejected", None);
        bad.error = Some("name in use".into());
        let summary = apply_results(&conn, &batch, &[bad]).expect("apply");
        assert_eq!(summary.rejected, 1);
        assert_eq!(dirty_count(&conn), 1, "a rejected change must be retried");
        assert!(summary.errors[0].contains("name in use"));
    }

    #[test]
    fn an_acknowledged_delete_drops_its_tombstone() {
        let conn = db();
        seed_note(&conn, "n1", 0);
        store::delete_forever(&conn, "n1").expect("delete");
        let batch = collect(&conn, 100, Accepts::ALL).expect("collect");
        apply_results(&conn, &batch, &[ok("applied", Some(7))]).expect("apply");
        assert!(!has_pending(&conn).expect("pending"));
    }

    #[test]
    fn a_noop_delete_also_drops_its_tombstone() {
        // Created and deleted entirely offline: the server never saw it.
        let conn = db();
        seed_note(&conn, "n1", 1);
        store::delete_forever(&conn, "n1").expect("delete");
        let batch = collect(&conn, 100, Accepts::ALL).expect("collect");
        apply_results(&conn, &batch, &[ok("noop", None)]).expect("apply");
        assert!(!has_pending(&conn).expect("pending"));
    }

    #[test]
    fn merging_labels_marks_the_affected_notes_dirty() {
        // The membership change only reaches the server through the note itself.
        let conn = db();
        seed_note(&conn, "n1", 0);
        for (id, name) in [("src", "Source"), ("dst", "Target")] {
            conn.execute(
                "INSERT INTO labels (id, name, color, created_at, updated_at, dirty)
                      VALUES (?1, ?2, 'default', '2026-01-01', '2026-01-01', 0)",
                params![id, name],
            )
            .expect("seed label");
        }
        conn.execute(
            "INSERT INTO note_labels (note_id, label_id, via_tag) VALUES ('n1', 'src', 0)",
            [],
        )
        .expect("seed membership");

        store::merge_labels(&conn, "src", "dst").expect("merge");
        assert_eq!(dirty_count(&conn), 1, "the note's label set changed");
    }

    #[test]
    fn a_note_shared_to_edit_sends_its_text_and_nothing_else() {
        let conn = db();
        seed_note(&conn, "n1", 1);
        conn.execute("UPDATE notes SET permission = 'edit' WHERE id = 'n1'", [])
            .unwrap();
        let changes = collect(&conn, 10, Accepts::ALL).unwrap();
        assert_eq!(changes.len(), 1);
        let wire = serde_json::to_value(&changes[0]).unwrap();
        let mut keys: Vec<&str> = wire
            .as_object()
            .unwrap()
            .keys()
            .map(String::as_str)
            .collect();
        keys.sort_unstable();
        assert_eq!(keys, ["body", "edited_at", "entity", "id", "op"]);
    }

    #[test]
    fn a_note_shared_to_view_sends_only_this_accounts_own_state() {
        let conn = db();
        seed_note(&conn, "n1", 1);
        conn.execute("UPDATE notes SET permission = 'view' WHERE id = 'n1'", [])
            .unwrap();
        // Dirty with nothing of ours changed on it: nothing to send.
        assert!(collect(&conn, 10, Accepts::ALL).unwrap().is_empty());

        conn.execute(
            "UPDATE notes SET pinned = 1, position = 4,
                              state_at = '2026-07-27T00:00:00.000Z' WHERE id = 'n1'",
            [],
        )
        .unwrap();
        let changes = collect(&conn, 10, Accepts::ALL).unwrap();
        let wire = serde_json::to_value(&changes[0]).unwrap();
        assert!(wire.get("body").is_none(), "a view share sends no text");
        assert_eq!(wire["pinned"], true);
        assert_eq!(wire["archived"], false);
        assert_eq!(wire["position"], 4);
        assert_eq!(wire["state_at"], "2026-07-27T00:00:00.000Z");
        assert_eq!(wire["edited_at"], "2026-07-26T00:00:00.000Z");

        // A server that doesn't keep them gets nothing, and it waits here.
        let older = Accepts {
            shared_state: false,
            ..Accepts::ALL
        };
        assert!(collect(&conn, 10, older).unwrap().is_empty());
    }

    #[test]
    fn has_pending_is_false_on_a_clean_store() {
        let conn = db();
        seed_note(&conn, "n1", 0);
        assert!(!has_pending(&conn).expect("pending"));
    }

    #[test]
    fn parse_results_reads_the_documented_shape() {
        let results = parse_results(
            r#"{"results":[{"id":"a","entity":"note","status":"created","sync_revision":44},
                           {"id":"b","entity":"label","status":"rejected","error":"name in use"}]}"#,
        )
        .expect("parse");
        assert_eq!(results.len(), 2);
        assert_eq!(results[0].status, "created");
        assert_eq!(results[1].error.as_deref(), Some("name in use"));
    }

    #[test]
    fn a_delete_change_serializes_without_note_fields() {
        let change = Change::delete("note", "n1".into(), "2026-07-26T00:00:00.000Z".into());
        let json = serde_json::to_string(&change).expect("serialize");
        assert!(json.contains("\"op\":\"delete\""), "got {json}");
        assert!(
            !json.contains("body"),
            "a delete carries no content: {json}"
        );
    }

    #[test]
    fn the_pending_fingerprint_moves_only_when_the_pending_set_does() {
        let conn = db();
        assert_eq!(pending_fingerprint(&conn).unwrap(), None);

        seed_note(&conn, "n1", 1);
        let one = pending_fingerprint(&conn).unwrap().expect("one dirty note");
        // Asking again changes nothing: a rejected row left dirty is not news.
        assert_eq!(pending_fingerprint(&conn).unwrap(), Some(one.clone()));

        seed_note(&conn, "n2", 1);
        let two = pending_fingerprint(&conn)
            .unwrap()
            .expect("two dirty notes");
        assert_ne!(one, two, "another pending note is new work");

        conn.execute(
            "UPDATE notes SET updated_at = '2026-07-27T00:00:00.000Z' WHERE id = 'n1'",
            [],
        )
        .unwrap();
        assert_ne!(
            pending_fingerprint(&conn).unwrap(),
            Some(two),
            "a fresh edit is new work"
        );

        conn.execute("UPDATE notes SET dirty = 0", []).unwrap();
        assert_eq!(pending_fingerprint(&conn).unwrap(), None);
    }

    fn queued_upload(conn: &Connection, id: &str, note_id: &str, error: Option<&str>) {
        conn.execute(
            "INSERT INTO attachments (id, note_id, url, filename, mime, sha256, uploaded, upload_error)
                  VALUES (?1, ?2, '/x', 'f.txt', 'text/plain', 'h', 0, ?3)",
            params![id, note_id, error],
        )
        .expect("queued upload");
    }

    fn upload_ids(conn: &Connection) -> Vec<String> {
        pending_uploads(conn)
            .expect("uploads")
            .into_iter()
            .map(|u| u.id)
            .collect()
    }

    #[test]
    fn removals_of_files_and_previews_wait_for_a_server_that_takes_them() {
        let conn = db();
        store::record_pending_delete(&conn, "attachment", "a1").expect("tombstone");
        store::record_pending_delete(&conn, "preview", "p1").expect("tombstone");
        store::record_pending_delete(&conn, "note", "n9").expect("tombstone");

        // An older server would answer "unknown entity" to each of them.
        let older: Vec<_> = collect(&conn, 100, Accepts::default())
            .expect("collect")
            .iter()
            .map(|c| c.entity)
            .collect();
        assert_eq!(older, vec!["note"]);

        let mut newer: Vec<_> = collect(&conn, 100, Accepts::ALL)
            .expect("collect")
            .iter()
            .map(|c| (c.entity, c.op))
            .collect();
        newer.sort();
        assert_eq!(
            newer,
            vec![
                ("attachment", "delete"),
                ("note", "delete"),
                ("preview", "delete")
            ]
        );
    }

    #[test]
    fn a_removed_attachment_stays_removed_through_a_whole_cycle() {
        use crate::sync::{pull, wire};
        let conn = db();
        let server_note = |revision: i64, attachments: Vec<wire::Attachment>| wire::Note {
            body: "a note".into(),
            attachments,
            ..wire::sample_note("n1", revision)
        };
        let page = |note: wire::Note, cursor: i64| wire::ChangesPage {
            notes: vec![note],
            labels: vec![],
            cursor,
            has_more: false,
            revoked: vec![],
        };
        let a1 = wire::Attachment {
            id: "a1".into(),
            url: "/x".into(),
            filename: None,
            mime: "image/png".into(),
            size: None,
            sha256: None,
        };
        pull::apply_page(&conn, &page(server_note(1, vec![a1]), 1)).expect("first pull");

        store::delete_attachment(&conn, "n1", "a1").expect("remove");

        // Push: the removal and the touched note go up, and the server applies both.
        let batch = collect(&conn, 100, Accepts::ALL).expect("collect");
        let results: Vec<PushResult> = batch
            .iter()
            .map(|c| ok("applied", (c.entity == "note").then_some(5)))
            .collect();
        apply_results(&conn, &batch, &results).expect("results");
        assert!(
            !has_pending(&conn).expect("pending"),
            "the tombstone is settled"
        );

        // Pull: the server's note now comes without it.
        pull::apply_page(&conn, &page(server_note(6, vec![]), 6)).expect("second pull");
        let left: i64 = conn
            .query_row("SELECT COUNT(*) FROM attachments", [], |r| r.get(0))
            .expect("count");
        assert_eq!(left, 0);
    }

    #[test]
    fn a_file_uploads_only_once_its_note_is_on_the_server() {
        let conn = db();
        seed_note(&conn, "landed", 0);
        seed_note(&conn, "unsent", 1);
        queued_upload(&conn, "a", "landed", None);
        queued_upload(&conn, "b", "unsent", None);
        queued_upload(&conn, "c", "landed", Some("file is too large (max 25 MB)"));
        assert_eq!(upload_ids(&conn), vec!["a"]);
    }

    #[test]
    fn a_refused_upload_leaves_the_queue_and_a_passing_failure_stays_in_it() {
        let conn = db();
        seed_note(&conn, "n", 0);
        queued_upload(&conn, "big", "n", None);
        queued_upload(&conn, "flaky", "n", None);
        assert!(
            pending_fingerprint(&conn).expect("fp").is_some(),
            "uploads are pending work"
        );

        let refused = Err(UploadError::Refused("file is too large (max 25 MB)".into()));
        settle_upload(&conn, "big", &refused).expect("settle");
        settle_upload(&conn, "flaky", &Err(UploadError::Retry("offline".into()))).expect("settle");
        assert_eq!(
            upload_ids(&conn),
            vec!["flaky"],
            "the refusal is not resent every cycle"
        );
        assert!(has_pending(&conn).expect("pending"));

        settle_upload(&conn, "flaky", &Ok(())).expect("settle");
        assert!(upload_ids(&conn).is_empty());
        assert!(!has_pending(&conn).expect("pending"));
        assert_eq!(pending_fingerprint(&conn).expect("fp"), None);
        let reason: Option<String> = conn
            .query_row(
                "SELECT upload_error FROM attachments WHERE id = 'big'",
                [],
                |r| r.get(0),
            )
            .expect("row");
        assert_eq!(
            reason.as_deref(),
            Some("file is too large (max 25 MB)"),
            "shown on the file"
        );
    }
}