//! Push: send local changes to the server and apply what it says (M10.7c).
//!
//! Two sources feed a push: rows flagged `dirty` (created or edited locally) and rows
//! in `pending_deletes` (permanently deleted locally — see `local::schema` v2 for why
//! a delete needs its own record).
//!
//! 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, OptionalExtension};
use serde::{Deserialize, Serialize};

use super::client;
use super::state;
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>,
}

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);
    }
}

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

/// 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, 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 title: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub body: Option<String>,
    #[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 items: Option<Vec<ItemOut>>,
    #[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>,
}

impl Change {
    fn delete(entity: &'static str, id: String, edited_at: String) -> Self {
        Change {
            entity,
            id,
            op: "delete",
            edited_at,
            title: None,
            body: None,
            color: None,
            pinned: None,
            archived: None,
            trashed: None,
            remind_at: None,
            recurrence: None,
            position: None,
            items: None,
            label_ids: None,
            created_at: None,
            name: None,
        }
    }
}

#[derive(Debug, Serialize)]
pub struct ItemOut {
    pub text: String,
    pub checked: bool,
}

// --- 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 --------------------------------------------------------------

/// 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) -> rusqlite::Result<Vec<Change>> {
    let mut out = Vec::new();
    collect_deletes(conn, &mut out, limit)?;
    if out.len() < limit {
        collect_labels(conn, &mut out, limit)?;
    }
    if out.len() < limit {
        collect_notes(conn, &mut out, limit)?;
    }
    Ok(out)
}

fn collect_deletes(conn: &Connection, out: &mut Vec<Change>, limit: usize) -> rusqlite::Result<()> {
    let mut stmt = conn.prepare(
        "SELECT entity, id, deleted_at FROM pending_deletes ORDER BY deleted_at LIMIT ?1",
    )?;
    let rows = stmt.query_map(params![limit as i64], |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 'note' and 'label' exist on the wire; anything else is a bug in a
        // writer, and shipping it would earn a blanket rejection for the batch.
        let entity: &'static str = match entity.as_str() {
            "note" => "note",
            "label" => "label",
            _ => 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: "label",
            id: r.get(0)?,
            op: "upsert",
            name: Some(r.get(1)?),
            color: Some(r.get(2)?),
            edited_at: r.get(3)?,
            title: None,
            body: None,
            pinned: None,
            archived: None,
            trashed: None,
            remind_at: None,
            recurrence: None,
            position: None,
            items: None,
            label_ids: None,
            created_at: None,
        })
    })?;
    for row in rows {
        out.push(row?);
    }
    Ok(())
}

fn collect_notes(conn: &Connection, out: &mut Vec<Change>, limit: usize) -> rusqlite::Result<()> {
    let remaining = limit.saturating_sub(out.len());
    let ids: Vec<String> = {
        let mut stmt =
            conn.prepare("SELECT id FROM notes WHERE dirty = 1 ORDER BY updated_at LIMIT ?1")?;
        let rows = stmt.query_map(params![remaining as i64], |r| r.get::<_, String>(0))?;
        rows.collect::<rusqlite::Result<Vec<String>>>()?
    };
    for id in ids {
        out.push(note_change(conn, &id)?);
    }
    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 {
    title: Option<String>,
    body: String,
    color: String,
    position: i64,
    pinned: bool,
    archived: bool,
    trashed: bool,
    remind_at: Option<String>,
    recurrence: Option<String>,
    created_at: String,
    updated_at: String,
}

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

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

    let items = {
        let mut stmt = conn.prepare(
            "SELECT text, checked FROM checklist_items WHERE note_id = ?1 ORDER BY position",
        )?;
        let rows = stmt.query_map(params![id], |r| {
            Ok(ItemOut {
                text: r.get(0)?,
                checked: r.get::<_, i64>(1)? != 0,
            })
        })?;
        rows.collect::<rusqlite::Result<Vec<ItemOut>>>()?
    };

    // 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: "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,
        title: row.title,
        body: Some(row.body),
        color: Some(row.color),
        pinned: Some(row.pinned),
        archived: Some(row.archived),
        trashed: Some(row.trashed),
        remind_at: row.remind_at,
        recurrence: row.recurrence,
        position: Some(row.position),
        items: Some(items),
        label_ids: Some(label_ids),
        created_at: Some(row.created_at),
        name: None,
    })
}

// --- 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 {
        "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> {
    let pending: Option<i64> = conn
        .query_row(
            "SELECT 1 FROM notes WHERE dirty = 1
              UNION ALL SELECT 1 FROM labels WHERE dirty = 1
              UNION ALL SELECT 1 FROM pending_deletes LIMIT 1",
            [],
            |r| r.get(0),
        )
        .optional()?;
    Ok(pending.is_some())
}

/// Send everything pending, in batches, applying each batch's results before the
/// next is collected.
pub async fn run(db: &Db, base_url: &str, token: &str) -> Result<PushSummary, String> {
    let mut total = PushSummary::default();

    loop {
        let batch = {
            let conn = db.0.lock().map_err(|e| e.to_string())?;
            collect(&conn, BATCH).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.0.lock().map_err(|e| e.to_string())?;
            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 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)",
        total.sent,
        total.created,
        total.applied,
        total.kept,
        total.noop,
        total.rejected
    );
    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::schema;
    use crate::local::store;

    fn db() -> Connection {
        let conn = Connection::open_in_memory().expect("in-memory db");
        schema::migrate(&conn).expect("migrate");
        conn
    }

    fn seed_note(conn: &Connection, id: &str, dirty: i64) {
        conn.execute(
            "INSERT INTO notes (id, title, body, color, position, pinned, archived,
                                trashed, created_at, updated_at, sync_revision, dirty)
                  VALUES (?1, 'T', 'B', 'default', 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).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).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).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).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).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).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).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).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).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).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 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}"
        );
    }
}