org_gdocs/

push.rs

//! P7 — the `push` orchestration: publish the org body's projection to its linked
//! Google Doc and write machine state back into the file.
//!
//! This is the imperative shell (EI-4). It delegates every decision to the pure
//! cores — `CUSTOM_ID`s ([`crate::custom_id`]), projection ([`crate::project`]),
//! sync state ([`crate::syncstate`]), comments ([`crate::comments_meta`]),
//! region writeback ([`crate::orgfile`]) — and only sequences the effects
//! (network, then file content) between them.
//!
//! Guarantees:
//!
//! - **DI-2** — the body is reproduced byte-for-byte except the sanctioned
//!   `:CUSTOM_ID:` insertions (and the tool-owned `#+GDOC_*` header keywords).
//! - **DI-9 (refined)** — when the body *does* change, the push fully replaces it
//!   (clear then re-insert, never diffed). But a push whose projection fingerprint
//!   ([`crate::project::Projection::fingerprint`]) matches the one recorded in
//!   `** Sync State` **skips the body write entirely** — a full-replace deletes the
//!   text existing Google comments anchor to, orphaning them ("original content
//!   deleted"), so re-pushing unchanged content (e.g. just to post a reply or
//!   resolve a comment) must not touch the body. Operator-approved relaxation.
//! - **DI-3** — all state is read from and written to the file content; no sidecar.
//!
//! IO (reading the file, writing it back, the clock) stays at the binary edge;
//! [`push`] takes the file content and the timestamp and returns the new content.

use kb::parser::parse_document;

use crate::comments_meta::{self, CommentState};
use crate::custom_id::ensure_section_ids;
use crate::envelope;
use crate::error::{Error, Result};
use crate::google::client::GoogleClient;
use crate::google::docs::{self, DocumentRef};
use crate::orgfile;
use crate::project;
use crate::sexp::Sexp;
use crate::syncstate::{PostedReply, SyncState};

/// The tool-owned metadata heading that opens the machine region.
const METADATA_HEADING: &str = "* GDOC_METADATA :noexport:";

/// The result of a push: the document reference, what changed, and the new file
/// content the caller must persist.
#[derive(Debug, Clone)]
pub struct PushOutcome {
    /// The (possibly newly created) target document.
    pub document: DocumentRef,
    /// Whether the document was created by this push.
    pub created: bool,
    /// Number of anchored elements projected (position-map size).
    pub element_count: usize,
    /// Comment ids that were resolved in Google this push.
    pub resolved: Vec<String>,
    /// Number of operator-authored replies posted to Google this push.
    pub replied: usize,
    /// Whether the document body was (re)written this push. `false` when the
    /// projection was unchanged since the last push, so the full-replace was
    /// skipped to preserve existing comment anchors.
    pub body_updated: bool,
    /// The new file content to write back (body preserved, machine region rebuilt).
    pub new_content: String,
}

/// Publish `content`'s projection to its linked doc, returning the rewritten file.
///
/// `default_title` is used to title a newly created doc when the file has no
/// `#+TITLE:`. `now` is the caller-supplied push timestamp (clock at the edge).
///
/// # Errors
///
/// Returns a [`crate::error::Error`] on org-parse failure, any Google API failure,
/// or a malformed existing machine region.
pub async fn push(
    client: &GoogleClient,
    content: &str,
    default_title: &str,
    now: &str,
) -> Result<PushOutcome> {
    let (body, machine) = orgfile::split(content);
    let machine = machine.unwrap_or("");

    // 1. Sanctioned body mutation: ensure every heading has a CUSTOM_ID (P2).
    let body_with_ids = ensure_section_ids(body);

    // 2. Parse the body read-only (P1/kb) and project it (P3).
    let document =
        parse_document(&body_with_ids).map_err(|err| Error::OrgParse(err.to_string()))?;
    let projection = project::project(&document);
    let element_count = projection.positions.len();
    let fingerprint = projection.fingerprint();
    let existing = SyncState::parse_block(machine)?;

    // 3. Create-or-get the doc.
    let (document_ref, created) = if let Some(id) = orgfile::read_keyword(content, "GDOC_ID") {
        (DocumentRef::from_id(id), false)
    } else {
        let title =
            orgfile::read_keyword(content, "TITLE").unwrap_or_else(|| default_title.to_owned());
        (client.create_document(&title).await?, true)
    };

    // 4. Full-replace the body (P4a, DI-9) ONLY for a new doc or when the
    //    projection changed since the last push. Skipping an unchanged re-push
    //    avoids deleting the text that existing Google comments anchor to — a
    //    full-replace would orphan every anchored comment as "original content
    //    deleted" (operator-approved DI-9 relaxation: full-replace on change,
    //    no-op on no change).
    let body_updated = created || existing.projection_hash.as_deref() != Some(fingerprint.as_str());
    if body_updated {
        let batch =
            full_replace_batch(client, &document_ref.id, created, projection.requests).await?;
        client.batch_update(&document_ref.id, batch).await?;
    }

    // 5. Resolve comments the operator marked DONE (P6 read + P4b update, DI-4).
    let resolved = resolve_done(client, &document_ref.id, machine).await;

    // 6. Post operator-authored replies not already posted (P6 read + Drive
    //    replies.create), tracking posted state in the sync state for idempotence.
    let (mut state, replied) = post_replies(client, &document_ref.id, machine, existing).await;

    // 7. Write machine state back, body preserved (P5/P6/P1). The position map and
    //    projection fingerprint are this push's; collaborators and posted replies
    //    carry over.
    state.positions = projection.positions;
    state.projection_hash = Some(fingerprint);
    let new_content = write_back(&body_with_ids, machine, &document_ref, &state, now);

    Ok(PushOutcome {
        document: document_ref,
        created,
        element_count,
        resolved,
        replied,
        body_updated,
        new_content,
    })
}

/// Post every operator-authored reply not already recorded as posted, isolating
/// failures (a failed post is simply retried next push). Returns the updated sync
/// state (collaborators preserved, newly-posted replies appended) and the count
/// posted this run.
async fn post_replies(
    client: &GoogleClient,
    document_id: &str,
    machine: &str,
    mut state: SyncState,
) -> (SyncState, usize) {
    let mut replied = 0;
    for reply in comments_meta::pending_replies(machine) {
        if state.reply_posted(&reply.comment_id, &reply.content) {
            continue;
        }
        if client
            .create_reply(document_id, &reply.comment_id, &reply.content)
            .await
            .is_ok()
        {
            state.posted_replies.push(PostedReply {
                comment_id: reply.comment_id,
                content: reply.content,
            });
            replied += 1;
        }
    }
    (state, replied)
}

/// Build the batch: for an existing doc, clear the body first (DI-9); then the
/// projection requests. Fetching the end index also enforces DI-6 (single tab).
async fn full_replace_batch(
    client: &GoogleClient,
    document_id: &str,
    created: bool,
    requests: Vec<google_docs1::api::Request>,
) -> Result<Vec<google_docs1::api::Request>> {
    let mut batch = Vec::with_capacity(requests.len() + 1);
    if !created {
        let end_index = client.document_end_index(document_id).await?;
        if let Some(delete) = docs::delete_body_request(end_index) {
            batch.push(delete);
        }
    }
    batch.extend(requests);
    Ok(batch)
}

/// Resolve every comment whose heading is `DONE`, isolating failures; return the
/// ids that were resolved.
async fn resolve_done(client: &GoogleClient, document_id: &str, machine: &str) -> Vec<String> {
    let done: Vec<String> = comments_meta::parse_entries(machine)
        .into_iter()
        .filter(|entry| entry.state == CommentState::Done)
        .map(|entry| entry.id)
        .collect();
    client
        .resolve_comments(document_id, &done)
        .await
        .into_iter()
        .filter_map(|(id, outcome)| outcome.is_ok().then_some(id))
        .collect()
}

/// Reassemble the file: body (with `CUSTOM_ID`s) + tool keywords + a regenerated
/// machine region (the given Sync State, existing Active Comments preserved
/// verbatim).
fn write_back(
    body_with_ids: &str,
    machine: &str,
    document: &DocumentRef,
    sync: &SyncState,
    now: &str,
) -> String {
    // Active Comments: preserve existing subtree verbatim (no new comments on push).
    let active = comments_meta::render_section(machine, &[]);
    let region = format!("{METADATA_HEADING}\n{}{active}", sync.render_block());

    let with_keywords = write_keywords(body_with_ids, document, now);
    orgfile::replace_metadata(&with_keywords, &region)
}

/// Upsert the tool-owned header keywords into the body.
fn write_keywords(body: &str, document: &DocumentRef, now: &str) -> String {
    let with_id = orgfile::upsert_keyword(body, "GDOC_ID", &document.id);
    let with_url = orgfile::upsert_keyword(&with_id, "GDOC_URL", &document.url);
    orgfile::upsert_keyword(&with_url, "GDOC_LAST_PUSH", now)
}

/// Build the success envelope for a completed push (A5).
#[must_use]
pub fn envelope(outcome: &PushOutcome) -> Sexp {
    let resolved = outcome
        .resolved
        .iter()
        .map(|id| Sexp::string(id.clone()))
        .collect();
    envelope::ok(
        "push",
        vec![
            ("document-id", Sexp::string(outcome.document.id.clone())),
            ("document-url", Sexp::string(outcome.document.url.clone())),
            ("created", Sexp::symbol(envelope::flag(outcome.created))),
            (
                "elements",
                Sexp::int(i64::try_from(outcome.element_count).unwrap_or(i64::MAX)),
            ),
            ("resolved", Sexp::list(resolved)),
            (
                "replied",
                Sexp::int(i64::try_from(outcome.replied).unwrap_or(i64::MAX)),
            ),
            (
                "body-updated",
                Sexp::symbol(envelope::flag(outcome.body_updated)),
            ),
        ],
    )
}

#[cfg(test)]
mod tests {
    use super::push;
    use crate::custom_id::ensure_section_ids;
    use crate::google::client::GoogleClient;
    use crate::orgfile;
    use mockito::{Matcher, Server};

    const NOW: &str = "2026-06-10T00:00:00+00:00";

    fn test_client(server: &Server) -> GoogleClient {
        let mut client = GoogleClient::new("test-token".to_owned()).expect("client builds");
        client.set_base_url(&server.url());
        client
    }

    async fn mock_create(server: &mut Server, id: &str) -> mockito::Mock {
        server
            .mock("POST", "/v1/documents")
            .match_query(Matcher::Any)
            .with_status(200)
            .with_header("content-type", "application/json")
            .with_body(format!(r#"{{"documentId":"{id}"}}"#))
            .create_async()
            .await
    }

    async fn mock_batch(server: &mut Server, id: &str) -> mockito::Mock {
        server
            .mock("POST", format!("/v1/documents/{id}:batchUpdate").as_str())
            .match_query(Matcher::Any)
            .with_status(200)
            .with_header("content-type", "application/json")
            .with_body(format!(r#"{{"documentId":"{id}"}}"#))
            .create_async()
            .await
    }

    #[tokio::test]
    async fn push_creates_doc_and_rebuilds_machine_region() {
        let mut server = Server::new_async().await;
        let create = mock_create(&mut server, "DOC1").await;
        let batch = mock_batch(&mut server, "DOC1").await;

        let content = "#+TITLE: Spec\n* Intro\nHello world.\n";
        let outcome = push(&test_client(&server), content, "fallback", NOW)
            .await
            .expect("push succeeds");

        assert!(outcome.created);
        assert_eq!(outcome.element_count, 2); // heading + paragraph
        let written = &outcome.new_content;
        assert!(written.contains("Hello world.\n"));
        assert!(written.contains(":CUSTOM_ID: sec-intro\n"));
        assert!(written.contains("#+GDOC_ID: DOC1\n"));
        assert!(written.contains("#+GDOC_URL: https://docs.google.com/document/d/DOC1/edit\n"));
        assert!(written.contains(&format!("#+GDOC_LAST_PUSH: {NOW}\n")));
        assert!(written.contains("* GDOC_METADATA :noexport:\n** Sync State\n"));
        assert!(written.contains("(pos \"sec-intro\" 1 heading)"));
        assert!(written.contains("** Active Comments\n"));

        create.assert_async().await;
        batch.assert_async().await;
    }

    #[tokio::test]
    async fn push_preserves_body_prose_byte_for_byte() {
        let mut server = Server::new_async().await;
        let _create = mock_create(&mut server, "DOC1").await;
        let _batch = mock_batch(&mut server, "DOC1").await;

        let content = "* Intro\nHello world.\n\n* Details\nMore text.\n";
        let outcome = push(&test_client(&server), content, "fallback", NOW)
            .await
            .expect("push succeeds");

        // The body equals the original + sanctioned CUSTOM_IDs + tool keywords, and
        // nothing else: no reflow, no kb canonicalization (DI-2).
        let expected = {
            let body = ensure_section_ids(content);
            let body = orgfile::upsert_keyword(&body, "GDOC_ID", "DOC1");
            let body = orgfile::upsert_keyword(
                &body,
                "GDOC_URL",
                "https://docs.google.com/document/d/DOC1/edit",
            );
            orgfile::upsert_keyword(&body, "GDOC_LAST_PUSH", NOW)
        };
        let pushed_body = orgfile::body(&outcome.new_content);
        assert_eq!(
            pushed_body.trim_end_matches('\n'),
            expected.trim_end_matches('\n')
        );
    }

    #[tokio::test]
    async fn push_existing_doc_full_replaces_without_create() {
        let mut server = Server::new_async().await;
        // GET end index (existing doc) — also enforces DI-6.
        let get = server
            .mock("GET", "/v1/documents/EXISTING")
            .match_query(Matcher::Any)
            .with_status(200)
            .with_header("content-type", "application/json")
            .with_body(r#"{"body":{"content":[{"endIndex":50}]}}"#)
            .create_async()
            .await;
        // The batch must clear the body first (DI-9 full-replace).
        let batch = server
            .mock("POST", "/v1/documents/EXISTING:batchUpdate")
            .match_query(Matcher::Any)
            .match_body(Matcher::Regex("deleteContentRange".to_owned()))
            .with_status(200)
            .with_header("content-type", "application/json")
            .with_body(r#"{"documentId":"EXISTING"}"#)
            .create_async()
            .await;

        let content = "#+GDOC_ID: EXISTING\n#+TITLE: Spec\n* Intro\nHi.\n";
        let outcome = push(&test_client(&server), content, "fallback", NOW)
            .await
            .expect("push succeeds");

        assert!(!outcome.created);
        // No create mock registered: a create call would 404 and fail the push.
        get.assert_async().await;
        batch.assert_async().await;
    }

    #[tokio::test]
    async fn push_resolves_done_comments() {
        let mut server = Server::new_async().await;
        let _create = mock_create(&mut server, "DOC1").await;
        let _batch = mock_batch(&mut server, "DOC1").await;
        let resolve = server
            .mock("PATCH", "/files/DOC1/comments/CDONE")
            .match_query(Matcher::Any)
            .match_body(Matcher::PartialJsonString(
                r#"{"resolved":true}"#.to_owned(),
            ))
            .with_status(200)
            .with_header("content-type", "application/json")
            .with_body(r#"{"id":"CDONE","resolved":true}"#)
            .create_async()
            .await;

        let content = "#+GDOC_ID: DOC1\n* Intro\nHi.\n\n\
            * GDOC_METADATA :noexport:\n** Sync State\n#+begin_src emacs-lisp\n(gdoc-sync-state 1 (positions) (collaborators))\n#+end_src\n\
            ** Active Comments\n*** DONE Alice: fixed\n:PROPERTIES:\n:COMMENT_ID: CDONE\n:END:\n";
        // Existing doc id present, so create is not called; GET end index is needed.
        let _get = server
            .mock("GET", "/v1/documents/DOC1")
            .match_query(Matcher::Any)
            .with_status(200)
            .with_header("content-type", "application/json")
            .with_body(r#"{"body":{"content":[{"endIndex":10}]}}"#)
            .create_async()
            .await;

        let outcome = push(&test_client(&server), content, "fallback", NOW)
            .await
            .expect("push succeeds");

        assert_eq!(outcome.resolved, vec!["CDONE".to_owned()]);
        // The DONE comment heading is preserved (push resolves in Google; `clean`
        // removes it later).
        assert!(outcome.new_content.contains(":COMMENT_ID: CDONE\n"));
        resolve.assert_async().await;
    }

    #[tokio::test]
    async fn unchanged_reprojection_skips_the_full_replace() {
        let mut server = Server::new_async().await;
        // Both are expected exactly once — the second push must NOT touch the body.
        let get = server
            .mock("GET", "/v1/documents/DOC1")
            .match_query(Matcher::Any)
            .with_status(200)
            .with_header("content-type", "application/json")
            .with_body(r#"{"body":{"content":[{"endIndex":20}]}}"#)
            .expect(1)
            .create_async()
            .await;
        let batch = server
            .mock("POST", "/v1/documents/DOC1:batchUpdate")
            .match_query(Matcher::Any)
            .with_status(200)
            .with_header("content-type", "application/json")
            .with_body(r#"{"documentId":"DOC1"}"#)
            .expect(1)
            .create_async()
            .await;

        // Existing doc, no stored fingerprint yet → first push full-replaces.
        let content = "#+GDOC_ID: DOC1\n* Intro\nHello world.\n";
        let first = push(&test_client(&server), content, "fallback", NOW)
            .await
            .expect("first push succeeds");
        assert!(first.body_updated);
        assert!(first.new_content.contains("(projection-hash "));

        // Re-push the rewritten file unchanged: the fingerprint matches, so the
        // body write (GET + batchUpdate) is skipped and comment anchors survive.
        let second = push(&test_client(&server), &first.new_content, "fallback", NOW)
            .await
            .expect("second push succeeds");
        assert!(!second.body_updated);

        get.assert_async().await; // exactly once — first push only
        batch.assert_async().await;
    }

    #[tokio::test]
    async fn push_posts_pending_replies_then_records_them_for_idempotence() {
        let mut server = Server::new_async().await;
        let _create = mock_create(&mut server, "DOC1").await;
        let _batch = mock_batch(&mut server, "DOC1").await;
        let _get = server
            .mock("GET", "/v1/documents/DOC1")
            .match_query(Matcher::Any)
            .with_status(200)
            .with_header("content-type", "application/json")
            .with_body(r#"{"body":{"content":[{"endIndex":10}]}}"#)
            .create_async()
            .await;
        // The reply post is expected exactly once across two pushes (idempotent).
        let reply = server
            .mock("POST", "/files/DOC1/comments/C1/replies")
            .match_query(Matcher::Any)
            .match_body(Matcher::PartialJsonString(
                r#"{"content":"Clarified."}"#.to_owned(),
            ))
            .with_status(200)
            .with_header("content-type", "application/json")
            .with_body(r#"{"id":"R1"}"#)
            .expect(1)
            .create_async()
            .await;

        let content = "#+GDOC_ID: DOC1\n* Intro\nHi.\n\n\
            * GDOC_METADATA :noexport:\n** Sync State\n#+begin_src emacs-lisp\n(gdoc-sync-state 1 (positions) (collaborators))\n#+end_src\n\
            ** Active Comments\n*** TODO Alice: please clarify\n:PROPERTIES:\n:COMMENT_ID: C1\n:END:\n**** REPLY\nClarified.\n";

        let first = push(&test_client(&server), content, "fallback", NOW)
            .await
            .expect("first push succeeds");
        assert_eq!(first.replied, 1);
        // The posted reply is recorded in the regenerated sync state.
        assert!(
            first
                .new_content
                .contains("(posted-replies (reply \"C1\" \"Clarified.\")")
        );

        // A second push over the rewritten file must NOT re-post (mock expects 1).
        let second = push(&test_client(&server), &first.new_content, "fallback", NOW)
            .await
            .expect("second push succeeds");
        assert_eq!(second.replied, 0);
        reply.assert_async().await;
    }
}