tangled-cli 0.1.0

//! Pull-request operations: PDS records, patch blobs, knot merges.

use std::io::{Read, Write};

use anyhow::{anyhow, Result};
use flate2::read::GzDecoder;
use flate2::write::GzEncoder;
use flate2::Compression;
use tangled_api::lexicon::com::atproto::{repo as atproto_repo, sync};
use tangled_api::lexicon::sh::tangled::repo as tangled_repo;
use tangled_api::types::BlobRef;

use super::auth::PdsAuth;
use super::types::{Pull, PullRecord};
use super::{http, service_auth, uri_rkey};

const PULL_COLLECTION: &str = "sh.tangled.repo.pull";

fn gzip_text(text: &str) -> Result<Vec<u8>> {
    let mut encoder = GzEncoder::new(Vec::new(), Compression::default());
    encoder.write_all(text.as_bytes())?;
    Ok(encoder.finish()?)
}

fn gunzip_text(bytes: &[u8]) -> Result<String> {
    let mut decoder = GzDecoder::new(bytes);
    let mut out = String::new();
    decoder.read_to_string(&mut out)?;
    Ok(out)
}

/// Timestamp format used by web-created pull records: whole seconds with a
/// numeric timezone offset (not `Z`).
pub(crate) fn pull_created_at_now() -> String {
    chrono::Local::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, false)
}

pub async fn list_pulls(
    pds_base: &str,
    author_did: &str,
    target_repo_at_uri: Option<&str>,
    auth: &PdsAuth,
) -> Result<Vec<PullRecord>> {
    let params = atproto_repo::list_records::Params {
        repo: author_did.to_string(),
        collection: PULL_COLLECTION.to_string(),
        limit: Some(100),
        cursor: None,
        reverse: None,
    };
    let rb = atproto_repo::list_records(http(), pds_base, &params);
    let res: atproto_repo::list_records::Output = auth.send(rb).await?;
    let mut out = vec![];
    for it in res.records {
        let pull: Pull = serde_json::from_value(it.value)?;
        if let Some(target) = target_repo_at_uri {
            if pull.target_repo() != target {
                continue;
            }
        }
        out.push(PullRecord {
            author_did: author_did.to_string(),
            rkey: uri_rkey(&it.uri).unwrap_or_default(),
            pull,
        });
    }
    Ok(out)
}

pub async fn get_pull_record(
    pds_base: &str,
    author_did: &str,
    rkey: &str,
    auth: &PdsAuth,
) -> Result<Pull> {
    let params = atproto_repo::get_record::Params {
        repo: author_did.to_string(),
        collection: PULL_COLLECTION.to_string(),
        rkey: rkey.to_string(),
        cid: None,
    };
    let rb = atproto_repo::get_record(http(), pds_base, &params);
    let out: atproto_repo::get_record::Output = auth.send(rb).await?;
    Ok(serde_json::from_value(out.value)?)
}

async fn upload_gzip_blob(
    pds_base: &str,
    auth: &PdsAuth,
    bytes: Vec<u8>,
) -> Result<BlobRef> {
    #[derive(serde::Deserialize)]
    struct Res {
        blob: BlobRef,
    }
    let rb = atproto_repo::upload_blob(
        http(),
        pds_base,
        bytes.into(),
        "application/gzip",
    );
    let res: Res = auth.send(rb).await?;
    Ok(res.blob)
}

#[allow(clippy::too_many_arguments)]
pub async fn create_pull(
    pds_base: &str,
    auth: &PdsAuth,
    author_did: &str,
    target_repo_ref: &str,
    target_branch: &str,
    source_repo_ref: Option<&str>,
    source_branch: &str,
    patch: &str,
    title: &str,
    body: Option<&str>,
) -> Result<String> {
    let now = pull_created_at_now();
    let record = if target_repo_ref.starts_with("did:") {
        // Current shape: gzipped patch stored as a blob in rounds[].
        let blob = upload_gzip_blob(pds_base, auth, gzip_text(patch)?).await?;
        let mut target = tangled_repo::pull::Target {
            repo: target_repo_ref.to_string(),
            branch: target_branch.to_string(),
            ..Default::default()
        };
        // The web app also writes repoDid inside target; keep parity.
        target
            .extra
            .insert("repoDid".to_string(), serde_json::json!(target_repo_ref));
        let record = tangled_repo::pull::Record {
            r#type: Some(PULL_COLLECTION.to_string()),
            target,
            source: Some(tangled_repo::pull::Source {
                branch: source_branch.to_string(),
                repo: source_repo_ref.map(str::to_string),
                ..Default::default()
            }),
            title: title.to_string(),
            body: body.map(str::to_string),
            rounds: vec![tangled_repo::pull::Round {
                patch_blob: blob,
                created_at: now.clone(),
                ..Default::default()
            }],
            created_at: now,
            ..Default::default()
        };
        serde_json::to_value(&record)?
    } else {
        // Legacy shape: inline patch and target repo as an at-uri.
        let mut record = serde_json::json!({
            "target": {
                "repo": target_repo_ref,
                "branch": target_branch,
            },
            "title": title,
            "patch": patch,
            "createdAt": now,
        });
        if let Some(body) = body {
            record["body"] = serde_json::json!(body);
        }
        record
    };

    let input = atproto_repo::create_record::Input {
        repo: author_did.to_string(),
        collection: PULL_COLLECTION.to_string(),
        rkey: None,
        validate: Some(false),
        record,
        swap_commit: None,
    };
    let rb = atproto_repo::create_record(http(), pds_base, &input);
    let out: atproto_repo::create_record::Output = auth.send(rb).await?;
    uri_rkey(&out.uri).ok_or_else(|| anyhow!("missing rkey in pull uri"))
}

/// Returns the patch text of a pull: inline for legacy records, otherwise
/// downloads and gunzips the latest round's patch blob.
pub async fn pull_patch(
    pds_base: &str,
    author_did: &str,
    pull: &Pull,
    auth: &PdsAuth,
) -> Result<String> {
    if !pull.patch.is_empty() {
        return Ok(pull.patch.clone());
    }
    let latest_round = pull
        .rounds
        .last()
        .ok_or_else(|| anyhow!("pull has no patch or rounds"))?;
    let params = sync::get_blob::Params {
        did: author_did.to_string(),
        cid: latest_round.patch_blob.reference.link.clone(),
    };
    let rb = sync::get_blob(http(), pds_base, &params);
    let bytes = auth.send_bytes(rb).await?;
    gunzip_text(&bytes)
}

pub async fn comment_pull(
    pds_base: &str,
    auth: &PdsAuth,
    author_did: &str,
    pull_at: &str,
    body: &str,
) -> Result<String> {
    let input = atproto_repo::create_record::Input {
        repo: author_did.to_string(),
        collection: "sh.tangled.repo.pull.comment".to_string(),
        rkey: None,
        validate: Some(false),
        record: serde_json::json!({
            "pull": pull_at,
            "body": body,
            "createdAt": super::now_rfc3339(),
        }),
        swap_commit: None,
    };
    let rb = atproto_repo::create_record(http(), pds_base, &input);
    let out: atproto_repo::create_record::Output = auth.send(rb).await?;
    uri_rkey(&out.uri)
        .ok_or_else(|| anyhow!("missing rkey in pull comment uri"))
}

pub async fn merge_pull(
    knot_base: &str,
    pull_did: &str,
    pull_rkey: &str,
    repo_did: &str,
    repo_name: &str,
    pds_base: &str,
    auth: &PdsAuth,
) -> Result<()> {
    // Fetch the pull to get patch and target branch
    let pull = get_pull_record(pds_base, pull_did, pull_rkey, auth).await?;

    let token = service_auth::mint(
        pds_base,
        auth,
        knot_base,
        tangled_repo::merge::NSID,
    )
    .await?;
    let patch = pull_patch(pds_base, pull_did, &pull, auth).await?;

    let input = tangled_repo::merge::Input {
        did: repo_did.to_string(),
        name: repo_name.to_string(),
        patch,
        branch: pull.target_branch().to_string(),
        commit_message: Some(pull.title.clone()),
        commit_body: if pull.body.is_empty() {
            None
        } else {
            Some(pull.body.clone())
        },
        author_name: None,
        author_email: None,
    };
    let rb = tangled_repo::merge(http(), knot_base, &input).bearer_auth(&token);
    let _: serde_json::Value = super::xrpc_send(rb).await?;
    Ok(())
}

#[cfg(test)]
mod tests {
    use mockito::{Matcher, Server};
    use serde_json::json;

    use super::*;

    #[tokio::test]
    async fn create_pull_uploads_gzipped_patch_blob_for_repo_did_targets() {
        let mut pds = Server::new_async().await;
        let _upload = pds
            .mock("POST", "/xrpc/com.atproto.repo.uploadBlob")
            .match_header("authorization", "Bearer access")
            .match_header("content-type", "application/gzip")
            .with_status(200)
            .with_body(
                json!({
                    "blob": {
                        "$type": "blob",
                        "ref": {"$link": "bafkpatch"},
                        "mimeType": "application/gzip",
                        "size": 42
                    }
                })
                .to_string(),
            )
            .create_async()
            .await;

        let _create_pull = pds
            .mock("POST", "/xrpc/com.atproto.repo.createRecord")
            .match_body(Matcher::PartialJson(json!({
                "repo": "did:plc:author",
                "collection": "sh.tangled.repo.pull",
                "record": {
                    "$type": "sh.tangled.repo.pull",
                    "target": {
                        "repo": "did:plc:repo",
                        "branch": "main",
                        "repoDid": "did:plc:repo"
                    },
                    "source": {
                        "branch": "feature",
                        "repo": "did:plc:source"
                    },
                    "rounds": [{
                        "patchBlob": {
                            "ref": {"$link": "bafkpatch"},
                            "mimeType": "application/gzip"
                        }
                    }]
                }
            })))
            .with_status(200)
            .with_body(
                json!({
                    "uri": "at://did:plc:author/sh.tangled.repo.pull/3pull",
                    "cid": "bafycid"
                })
                .to_string(),
            )
            .create_async()
            .await;

        let auth = PdsAuth::Bearer("access".to_string());
        let rkey = create_pull(
            &pds.url(),
            &auth,
            "did:plc:author",
            "did:plc:repo",
            "main",
            Some("did:plc:source"),
            "feature",
            "diff --git a/a b/a\n",
            "A change",
            None,
        )
        .await
        .expect("current pull records should store gzipped patch blobs");

        assert_eq!(rkey, "3pull");
    }

    #[test]
    fn pull_created_at_uses_numeric_timezone_offset() {
        let timestamp = pull_created_at_now();

        assert!(
            !timestamp.ends_with('Z'),
            "web-created Tangled pull records use numeric timezone offsets, not Z: {}",
            timestamp
        );
        assert!(
            !timestamp.contains('.'),
            "pull timestamps should use whole seconds like web-created records: {}",
            timestamp
        );
        assert!(
            timestamp
                .as_bytes()
                .get(timestamp.len().saturating_sub(6))
                .is_some_and(|ch| *ch == b'+' || *ch == b'-'),
            "expected RFC3339 numeric offset suffix, got {}",
            timestamp
        );
        chrono::DateTime::parse_from_rfc3339(&timestamp)
            .expect("timestamp should remain valid RFC3339");
    }
}