decision/crates/ld-server/src/publish.rs

//! Publication of the texts in force to the git repository that client
//! applications read to update themselves (Yvv, 2026-10-21: « Le tout est
//! automatique poussé sur le dépôt git exploité par les applis clientes »).
//!
//! Every adoption, typo fix or translation of a text in force queues an
//! outbox row. The worker renders the document — one Markdown file per
//! language plus a JSON file — commits it in the server's own clone and pushes
//! with a deploy key. A failure is retried with backoff; the vote never waits
//! for the network.

use crate::AppState;
use crate::docs::{self, DocState};
use crate::util;
use serde_json::json;
use std::path::{Path, PathBuf};
use std::time::Duration;
use tokio::process::Command;

const IDLE: Duration = Duration::from_secs(60);
const MAX_BACKOFF_SECS: i64 = 6 * 3600;

/// The deploy key of the server: generated on first start, never leaves the
/// data volume. Its public half is shown to access administrators, who add it
/// as a deploy key (write) on the texts repository.
pub async fn ensure_key(state: &AppState) -> anyhow::Result<()> {
    let Some(publish) = &state.cfg.publish else { return Ok(()) };
    if publish.key.exists() {
        return Ok(());
    }
    if let Some(dir) = publish.key.parent() {
        tokio::fs::create_dir_all(dir).await?;
    }
    let out = Command::new("ssh-keygen")
        .args(["-q", "-t", "ed25519", "-N", "", "-C", "libredecision-publication", "-f"])
        .arg(&publish.key)
        .output()
        .await?;
    anyhow::ensure!(out.status.success(), "ssh-keygen: {}", String::from_utf8_lossy(&out.stderr).trim());
    tracing::info!(key = %publish.key.display(), "publication key generated");
    Ok(())
}

pub fn public_key(state: &AppState) -> Option<String> {
    let publish = state.cfg.publish.as_ref()?;
    let mut path = publish.key.clone().into_os_string();
    path.push(".pub");
    std::fs::read_to_string(path).ok().map(|k| k.trim().to_owned())
}

pub async fn worker(state: AppState) {
    loop {
        // A pass first: what was queued before the start goes out without waiting.
        if state.cfg.publish.is_some()
            && let Err(e) = pass(&state).await
        {
            tracing::warn!(error = %e, "publication pass");
        }
        tokio::select! {
            () = state.publish_wake.notified() => {}
            () = tokio::time::sleep(IDLE) => {}
        }
    }
}

#[derive(sqlx::FromRow)]
struct Due {
    id: i64,
    doc_id: String,
    attempts: i64,
}

/// One publication pass, now (tests, and the worker).
pub async fn pass(state: &AppState) -> anyhow::Result<()> {
    let due = sqlx::query_as::<_, Due>(
        "SELECT id, doc_id, attempts FROM outbox WHERE done_at IS NULL AND next_at <= ? ORDER BY id",
    )
    .bind(util::now())
    .fetch_all(&state.db)
    .await?;
    let mut docs_due: Vec<String> = due.iter().map(|d| d.doc_id.clone()).collect();
    docs_due.dedup();
    for doc_id in docs_due {
        let rows: Vec<&Due> = due.iter().filter(|d| d.doc_id == doc_id).collect();
        let last = rows.iter().map(|d| d.id).max().unwrap_or(0);
        let attempts = rows.iter().map(|d| d.attempts).max().unwrap_or(0);
        match publish_doc(state, &doc_id).await {
            Ok(sha) => {
                sqlx::query("UPDATE outbox SET done_at = ?, commit_sha = ?, last_error = NULL WHERE doc_id = ? AND done_at IS NULL AND id <= ?")
                    .bind(util::now())
                    .bind(&sha)
                    .bind(&doc_id)
                    .bind(last)
                    .execute(&state.db)
                    .await?;
                tracing::info!(doc = %doc_id, commit = sha.as_deref().unwrap_or("unchanged"), "published");
            }
            Err(e) => {
                let wait = (60_i64 << attempts.min(12)).min(MAX_BACKOFF_SECS);
                sqlx::query("UPDATE outbox SET attempts = attempts + 1, last_error = ?, next_at = ? WHERE doc_id = ? AND done_at IS NULL")
                    .bind(format!("{e:#}").chars().take(500).collect::<String>())
                    .bind(util::in_seconds(wait))
                    .bind(&doc_id)
                    .execute(&state.db)
                    .await?;
                tracing::warn!(doc = %doc_id, error = %e, retry_in = wait, "publication failed");
            }
        }
    }
    Ok(())
}

async fn git(dir: &Path, state: &AppState, args: &[&str]) -> anyhow::Result<String> {
    let publish = state.cfg.publish.as_ref().ok_or_else(|| anyhow::anyhow!("publication not configured"))?;
    let known_hosts = state.cfg.data_dir.join("keys/known_hosts");
    let ssh = format!(
        "ssh -i {} -o IdentitiesOnly=yes -o BatchMode=yes -o StrictHostKeyChecking=accept-new -o UserKnownHostsFile={}",
        publish.key.display(),
        known_hosts.display()
    );
    let out = Command::new("git")
        .args([
            "-c",
            &format!("user.name={}", publish.author_name),
            "-c",
            &format!("user.email={}", publish.author_email),
        ])
        .args(args)
        .current_dir(dir)
        .env("GIT_SSH_COMMAND", ssh)
        .env("GIT_TERMINAL_PROMPT", "0")
        .output()
        .await?;
    if !out.status.success() {
        anyhow::bail!("git {}: {}", args.first().unwrap_or(&""), String::from_utf8_lossy(&out.stderr).trim());
    }
    Ok(String::from_utf8_lossy(&out.stdout).trim().to_owned())
}

/// The server's own clone: a cache of the remote, realigned on it before each
/// commit so a push never needs a merge. Returns the clone and the commit the
/// remote branch points to, if it exists.
async fn checkout(state: &AppState) -> anyhow::Result<(PathBuf, Option<String>)> {
    let publish = state.cfg.publish.as_ref().ok_or_else(|| anyhow::anyhow!("publication not configured"))?;
    let dir = state.cfg.data_dir.join("textes");
    if !dir.join(".git").exists() {
        tokio::fs::create_dir_all(&dir).await?;
        git(&dir, state, &["init", "-q"]).await?;
        git(&dir, state, &["remote", "add", "origin", &publish.remote]).await?;
    }
    git(&dir, state, &["remote", "set-url", "origin", &publish.remote]).await?;
    let has_remote_branch = git(&dir, state, &["ls-remote", "--heads", "origin", &publish.branch]).await?;
    let remote_head = has_remote_branch.split_whitespace().next().map(str::to_owned);
    if remote_head.is_none() {
        git(&dir, state, &["checkout", "-q", "-f", "-B", &publish.branch]).await?;
    } else {
        // This clone is a cache: whatever a failed pass left is re-rendered from the database.
        git(&dir, state, &["fetch", "-q", "origin", &publish.branch]).await?;
        git(&dir, state, &["checkout", "-q", "-f", "-B", &publish.branch, &format!("origin/{}", publish.branch)])
            .await?;
    }
    Ok((dir, remote_head))
}

async fn publish_doc(state: &AppState, doc_id: &str) -> anyhow::Result<Option<String>> {
    let st = docs::load_state(state, doc_id).await.map_err(|e| anyhow::anyhow!("{e}"))?;
    let Some(path) = st.doc.publish_path.clone() else { return Ok(None) };
    let rendered = render(state, &st).await?;
    let (dir, remote_head) = checkout(state).await?;
    let branch = state.cfg.publish.as_ref().map(|p| p.branch.clone()).unwrap_or_else(|| "main".into());
    let target = dir.join(&path);
    tokio::fs::create_dir_all(&target).await?;
    for (name, content) in &rendered {
        tokio::fs::write(target.join(name), content).await?;
    }
    let readme = dir.join("README.md");
    if !readme.exists() {
        tokio::fs::write(&readme, README).await?;
    }
    git(&dir, state, &["add", "-A", "--", &path, "README.md"]).await?;
    let staged = git(&dir, state, &["diff", "--cached", "--name-only"]).await?;
    if staged.is_empty() {
        // Nothing new — but a commit the remote never received still goes out: a
        // first push refused (a key without write access), or a new, empty
        // repository after a change of remote.
        let Ok(head) = git(&dir, state, &["rev-parse", "--verify", "-q", "HEAD"]).await else { return Ok(None) };
        if remote_head.as_deref() == Some(head.as_str()) {
            return Ok(None);
        }
        git(&dir, state, &["push", "-q", "origin", &format!("HEAD:{branch}")]).await?;
        return Ok(Some(head));
    }
    let reasons: Vec<String> =
        sqlx::query_scalar("SELECT reason FROM outbox WHERE doc_id = ? AND done_at IS NULL ORDER BY id")
            .bind(doc_id)
            .fetch_all(&state.db)
            .await?;
    let title = util::parse_lang(&st.doc.title).get(&st.doc.source_lang).cloned().unwrap_or_default();
    let mut message = format!("{title} — {}", reasons.last().cloned().unwrap_or_else(|| "mise à jour".into()));
    if reasons.len() > 1 {
        message.push_str("\n\n");
        for r in &reasons {
            message.push_str(&format!("- {r}\n"));
        }
    }
    git(&dir, state, &["commit", "-q", "-m", &message]).await?;
    git(&dir, state, &["push", "-q", "origin", &format!("HEAD:{branch}")]).await?;
    Ok(Some(git(&dir, state, &["rev-parse", "HEAD"]).await?))
}

const README: &str = "# Textes en vigueur — libreDecision\n\n\
Ce dépôt est écrit par libreDecision (https://decision.librodrome.org) : chaque fois qu'un article \
entre en vigueur par le vote, qu'une coquille est corrigée ou qu'une traduction change, le texte en \
vigueur est republié ici. Les applications clientes le lisent pour se mettre à jour.\n\n\
- `<groupe>/<document>/fr.md` : le texte qui fait foi (langue source) ;\n\
- `en.md`, `es.md` : traductions éditoriales, sans vote ;\n\
- `document.json` : le même contenu, structuré, avec l'état du vote de chaque article.\n\n\
Ne pas modifier à la main : le prochain passage de libreDecision réécrit ces fichiers.\n";

/// The date of the last change to what is published (an entry into force, a
/// typo fix, a translation) — never "now", so that publishing an unchanged
/// document changes nothing and commits nothing.
async fn last_change(state: &AppState, doc_id: &str) -> anyhow::Result<String> {
    let at: Option<String> = sqlx::query_scalar(
        "SELECT MAX(d) FROM (
           SELECT MAX(at) AS d FROM adoptions WHERE doc_id = ?1
           UNION ALL SELECT MAX(p.updated_at) FROM proposals p JOIN units u ON u.in_force_id = p.id WHERE u.doc_id = ?1
           UNION ALL SELECT MAX(t.updated_at) FROM translations t JOIN units u ON u.in_force_id = t.proposal_id WHERE u.doc_id = ?1
           UNION ALL SELECT created_at FROM documents WHERE id = ?1)",
    )
    .bind(doc_id)
    .fetch_one(&state.db)
    .await?;
    Ok(at.unwrap_or_else(util::now))
}

/// On start, every published document is queued once more: a change that
/// entered into force just before a restart, or a new rendering format, reaches
/// the repository. Unchanged documents produce no commit. What was waiting
/// after failures is tried again at once (a key just added, a new remote).
pub async fn resync(state: &AppState) -> anyhow::Result<()> {
    if state.cfg.publish.is_none() {
        return Ok(());
    }
    retry_now(state).await?;
    let now = util::now();
    sqlx::query(
        "INSERT INTO outbox (doc_id, reason, created_at, next_at)
         SELECT d.id, 'reprise au démarrage', ?1, ?1 FROM documents d
         WHERE d.publish_path IS NOT NULL
           AND NOT EXISTS (SELECT 1 FROM outbox o WHERE o.doc_id = d.id AND o.done_at IS NULL)",
    )
    .bind(&now)
    .execute(&state.db)
    .await?;
    state.publish_wake.notify_one();
    Ok(())
}

/// Everything still waiting goes out at the next pass, backoff forgotten.
pub async fn retry_now(state: &AppState) -> anyhow::Result<()> {
    sqlx::query("UPDATE outbox SET next_at = ?, attempts = 0 WHERE done_at IS NULL")
        .bind(util::now())
        .execute(&state.db)
        .await?;
    state.publish_wake.notify_one();
    Ok(())
}

/// What administrators see of the publication in trustinWallet.
pub async fn status(state: &AppState) -> anyhow::Result<serde_json::Value> {
    let pending: i64 =
        sqlx::query_scalar("SELECT COUNT(*) FROM outbox WHERE done_at IS NULL").fetch_one(&state.db).await?;
    let error: Option<String> = sqlx::query_scalar(
        "SELECT last_error FROM outbox WHERE done_at IS NULL AND last_error IS NOT NULL ORDER BY id DESC LIMIT 1",
    )
    .fetch_optional(&state.db)
    .await?;
    let last: Option<(String, Option<String>)> = sqlx::query_as(
        "SELECT done_at, commit_sha FROM outbox WHERE done_at IS NOT NULL AND commit_sha IS NOT NULL
         ORDER BY done_at DESC LIMIT 1",
    )
    .fetch_optional(&state.db)
    .await?;
    Ok(json!({
        "pending": pending,
        "lastError": error,
        "lastPublishedAt": last.as_ref().map(|(at, _)| at.clone()),
        "lastCommit": last.and_then(|(_, sha)| sha),
    }))
}

/// Markdown per language + JSON. Only the units in force, in order.
pub async fn render(state: &AppState, st: &DocState) -> anyhow::Result<Vec<(String, String)>> {
    let titles = util::parse_lang(&st.doc.title);
    let src = st.doc.source_lang.clone();
    let in_force: Vec<_> = st
        .units
        .iter()
        .filter_map(|u| {
            let fid = u.in_force_id.as_ref()?;
            st.proposals.iter().find(|p| &p.id == fid).map(|p| (u, p))
        })
        .collect();
    let refs: Vec<&docs::ProposalRow> = in_force.iter().map(|(_, p)| *p).collect();
    let tr = docs::translations_for(state, &refs).await.map_err(|e| anyhow::anyhow!("{e}"))?;
    let updated = last_change(state, &st.doc.id).await?;
    let source_url = format!("{}/g/{}/docs/{}", state.cfg.public_url, st.group.slug, st.doc.slug);
    let mut files = Vec::new();
    for lang in util::LANGS {
        let title = titles.get(lang).or_else(|| titles.get(&src)).cloned().unwrap_or_default();
        let mut md = format!(
            "---\ndocument: {}\nlangue: {lang}\nsource: {source_url}\nmise_a_jour: {updated}\n{}---\n\n# {title}\n",
            st.doc.slug,
            if lang == src { "fait_foi: oui\n".to_owned() } else { "traduction: editoriale\n".to_owned() },
        );
        for (_, p) in &in_force {
            let t = tr.get(&p.id).and_then(|m| m.get(lang));
            let (ut, ub, fallback) = match t {
                Some(t) if lang != src => (
                    t["title"].as_str().unwrap_or(&p.title).to_owned(),
                    t["body"].as_str().unwrap_or(&p.body).to_owned(),
                    t["stale"].as_bool().unwrap_or(false),
                ),
                _ => (p.title.clone(), p.body.clone(), lang != src),
            };
            md.push_str(&format!("\n## {ut}\n\n{ub}\n"));
            if fallback {
                md.push_str(&format!(
                    "\n<!-- {lang} : traduction à revoir ou absente, texte source affiché si absente -->\n"
                ));
            }
        }
        files.push((format!("{lang}.md"), md));
    }
    let units: Vec<_> = in_force
        .iter()
        .map(|(u, p)| {
            let mut title = serde_json::Map::new();
            let mut body = serde_json::Map::new();
            title.insert(src.clone(), json!(p.title));
            body.insert(src.clone(), json!(p.body));
            if let Some(m) = tr.get(&p.id) {
                for (lang, t) in m {
                    title.insert(lang.clone(), t["title"].clone());
                    body.insert(lang.clone(), t["body"].clone());
                }
            }
            let tally = st.tally(&p.id);
            json!({
                "code": u.code, "title": title, "body": body, "inertia": u.inertia,
                "status": if p.entered_at.is_some() { "adopted" } else if p.confirmed_at.is_some() { "confirmed" } else { "seed-minimal" },
                "enteredAt": p.entered_at, "votes": { "for": tally.for_votes, "against": tally.against },
            })
        })
        .collect();
    let json = json!({
        "document": st.doc.slug, "group": st.group.slug, "title": titles, "sourceLang": src,
        "source": source_url, "updatedAt": updated, "electorate": st.electorate,
        "voteFloor": st.doc.vote_floor, "units": units,
    });
    files.push(("document.json".into(), format!("{}\n", serde_json::to_string_pretty(&json)?)));
    Ok(files)
}