//! The vote, grouped: the whole document, a section or an article in one
//! gesture (Yvv, 2026-10-30: « Je peux voter tout d'un coup, ou une section
//! entière, ou un article avec son titre, ou séparément »).
//!
//! - In a **managed group** (essai, Axiom…), the votes are written on the
//!   server in one transaction, as before: the test group stays easy to try.
//! - Where the electorate is the **Ğ1 web of trust** (Monnaie libre), a vote is
//!   one transaction on the chain, signed by the person's wallet (ADR 0008):
//!   the server seals a **pending ballot** (`ld_engine::ballot`), the page has
//!   it signed and sent, and the server **confirms** it by reading the indexer:
//!   the remark, the transfer of the amount to the vote account in the same
//!   transaction, the signer's identity and its membership — now. The vote
//!   counts for the identity `idty:<index>` of the signer, created if need be;
//!   the session that asked for the ballot sees it as « ton vote ».
//! - **The last one stands**, per proposal and per identity, in the order of
//!   the chain: a ballot confirmed late never overrides a more recent one.
//!
//! A background task reads the indexer for pending ballots every 10 seconds,
//! at most 50 a pass, the least recently read first: every pending ballot is
//! read at least every ⌈n / 50⌉ passes, whether a page waits for it or not (a
//! Ğecko code). A ballot is read one last time at its end (24 hours), then let
//! go; the adoption pass runs after each confirmation. Ballots let go are purged
//! after a while (a day for those the wallet never signed, a week for the others).
//!
//! **The buffer** (ADR 0012, Yvv 2026-11-02: « je vois ce que j'ai fait sans
//! attendre », with a control « a posteriori » of the integrity of the vote): a
//! pending ballot is an **intention**, seen at once by the person (« ton vote »)
//! and in the totals shown, never counted for adoption; the chain brings its
//! **proof**; the **reconciliation** (`ld_engine::buffer::reconcile`) keeps one
//! voice per identity and per version, the last in the order of the chain, and
//! sets aside what is never proven in time, refused, overtaken — or a *for*
//! proven while an objection holds its version back. What is set aside is told
//! in the box « Messages ».
//!
//! **The simulation**: while the chain is closed (no `LD_VOTE_ACCOUNT`), a group
//! of the chain votes anyway, without a wallet — through the same buffer: a
//! sealed ballot, simulated, proven at once by the simulated confirmer, for the
//! identity of the session.

use crate::AppState;
use crate::adoption::{self, Entered};
use crate::auth::{IdentityRow, User};
use crate::docs::{self, DocState};
use crate::error::{AppError, AppResult};
use crate::indexer::{self, Found, Reading};
use crate::messages::{self, Topic};
use crate::util;
use axum::Json;
use axum::extract::{Path, State};
use axum::http::{HeaderValue, header};
use axum::response::{IntoResponse, Response};
use ld_engine::address;
use ld_engine::ballot::{self, ChainPosition, Choice, Expected, Verdict};
use ld_engine::buffer::{self, Outcome, Proven, Stage};
use rand::RngCore;
use serde::Deserialize;
use serde_json::json;
use sqlx::SqliteConnection;
use std::collections::{HashMap, HashSet};
use std::sync::atomic::Ordering;
use std::time::Duration;
use tokio::sync::Mutex;
use trustin_access::VoteWay;

/// Every article of a long document, with room to spare.
const MAX_LINES: usize = 200;
/// Ballots one identity may hold at once: those pending, and those let go within
/// the last hour (`DELETE`, the wallet did not sign) — whatever the session.
const MAX_PENDING: i64 = 20;
const DROPPED_WINDOW_SECS: i64 = 3600;
/// A ballot not seen on the chain within this time is let go.
pub const EXPIRES_HOURS: i64 = 24;
/// At its end, a ballot no indexer could be read for — or whose comment waits
/// for a final block — is kept and read again, this long at most.
const GRACE_SECS: i64 = 3600;
/// Let go, a ballot is purged, lines and all: a day after the wallet did not sign
/// it, a week after it was not seen in time (under the article, it says so meanwhile).
const PURGE_DROPPED_SECS: i64 = 24 * 3600;
const PURGE_EXPIRED_SECS: i64 = 7 * 24 * 3600;
/// The page asks often; the indexer is read at most this often per ballot.
const RECHECK_SECS: i64 = 3;
const WATCH_EVERY: Duration = Duration::from_secs(10);
/// Pending ballots read per pass of the background task, the least recently read first.
const PER_PASS: i64 = 50;
/// Ballots a history shows, the most recent first.
const HISTORY: i64 = 200;

/// One confirmation at a time: « the last one stands » reads the other ballots of
/// the identity, then writes; two at once could both read before either writes.
static CONFIRM: Mutex<()> = Mutex::const_new(());

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct LineBody {
    proposal_id: String,
    /// `for`, `against`, or null to withdraw.
    choice: Option<String>,
}

#[derive(Deserialize)]
pub struct VotesBody {
    lines: Vec<LineBody>,
}

#[derive(Deserialize)]
pub struct BallotBody {
    lines: Vec<LineBody>,
    /// The account that will sign, in any SS58 form, when the page knows it.
    author: Option<String>,
}

/// A line checked against the document: a version open to its choice, its revision now.
pub(crate) struct Cast {
    pub proposal_id: String,
    pub revision: i64,
    pub choice: Choice,
}

fn parse_choice(raw: Option<&str>) -> AppResult<Choice> {
    match raw {
        None | Some("none") => Ok(Choice::None),
        Some(c) => Choice::parse(c).ok_or(AppError::Invalid("choice")),
    }
}

/// The lines of a vote, read against the document: at least one, each version
/// once, each one of this document and open to its choice — a version held back
/// by an objection takes against or a withdrawal, never a *for* (`DocState::admits`).
/// One line refused refuses them all.
fn casts(st: &DocState, lines: &[LineBody]) -> AppResult<Vec<Cast>> {
    if lines.is_empty() || lines.len() > MAX_LINES {
        return Err(AppError::Invalid("lines"));
    }
    let mut seen = HashSet::new();
    let mut out = Vec::with_capacity(lines.len());
    for l in lines {
        if !seen.insert(l.proposal_id.as_str()) {
            return Err(AppError::Invalid("lines"));
        }
        let p = st.proposals.iter().find(|p| p.id == l.proposal_id).ok_or(AppError::Invalid("lines"))?;
        if !matches!(p.status.as_str(), "open" | "in_force") {
            return Err(AppError::Conflict("not_votable"));
        }
        let choice = parse_choice(l.choice.as_deref())?;
        // Held back by an objection: no new *for*, on the server or on the chain
        // (ADR 0010, Yvv 2026-11-01: « Retirer ou "contre" »).
        if !st.admits(p, choice) {
            return Err(AppError::Conflict("objection"));
        }
        out.push(Cast { proposal_id: p.id.clone(), revision: p.revision, choice });
    }
    Ok(out)
}

/// Writes votes on the server, all or none.
pub(crate) async fn write_server_votes(state: &AppState, identity: &str, casts: &[Cast]) -> AppResult<()> {
    let mut tx = state.db.begin().await?;
    let now = util::now();
    for c in casts {
        match c.choice {
            Choice::None => {
                sqlx::query("DELETE FROM votes WHERE proposal_id = ? AND identity_id = ?")
                    .bind(&c.proposal_id)
                    .bind(identity)
                    .execute(&mut *tx)
                    .await?;
            }
            choice => {
                sqlx::query(
                    "INSERT INTO votes (proposal_id, identity_id, choice, revision, cast_at, source) VALUES (?, ?, ?, ?, ?, 'server')
                     ON CONFLICT(proposal_id, identity_id) DO UPDATE SET choice = excluded.choice,
                       revision = excluded.revision, cast_at = excluded.cast_at, source = 'server',
                       block = NULL, event_index = NULL, ballot_id = NULL",
                )
                .bind(&c.proposal_id)
                .bind(identity)
                .bind(choice.as_str())
                .bind(c.revision)
                .bind(&now)
                .execute(&mut *tx)
                .await?;
            }
        }
    }
    tx.commit().await?;
    Ok(())
}

/// How a vote of this session is written by the server in a group now.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum ServerWay {
    /// A managed group: the server is the authority, the vote is recorded at once.
    Direct,
    /// A group of the chain while the chain is closed: a simulated ballot, proven
    /// by the simulated confirmer (ADR 0012).
    Simulated,
}

/// Where the vote goes through the chain and the chain is open, never on the
/// server (`409 chain_only`): a ballot to sign.
pub(crate) fn server_way(state: &AppState, group: &docs::GroupRow) -> AppResult<ServerWay> {
    match trustin_access::vote_way(&group.to_group()) {
        VoteWay::Server => Ok(ServerWay::Direct),
        VoteWay::Chain if state.chain().is_some() => Err(AppError::Conflict("chain_only")),
        VoteWay::Chain => Ok(ServerWay::Simulated),
    }
}

/// May this session write its vote on the server, this way? trustinWallet's rules:
/// `votes_in` (a managed group), `casts_simulated` (the simulation).
pub(crate) async fn may_write(
    state: &AppState,
    user: &IdentityRow,
    group: &docs::GroupRow,
    way: ServerWay,
) -> AppResult<bool> {
    let granted = docs::granted(state, &group.slug, &user.id).await?;
    let (id, g) = (user.to_identity(), group.to_group());
    Ok(match way {
        ServerWay::Direct => trustin_access::votes_in(&id, &g, granted),
        ServerWay::Simulated => trustin_access::casts_simulated(&id, &g, granted, state.chain().is_some()),
    })
}

/// Writes the votes of a session on the server — directly, or in simulation — and
/// runs the adoption pass. Returns what entered into force.
pub(crate) async fn write(
    state: &AppState,
    user: &IdentityRow,
    st: &DocState,
    way: ServerWay,
    casts: &[Cast],
) -> AppResult<Vec<Entered>> {
    match way {
        ServerWay::Direct => {
            write_server_votes(state, &user.id, casts).await?;
            adoption::run(state, &st.doc.id).await
        }
        ServerWay::Simulated => simulate(state, &user.id, st, casts).await,
    }
}

/// `POST /api/docs/{id}/votes` — a grouped vote written by the server: one write
/// in a managed group, one simulated ballot while the chain is closed.
pub async fn vote_many(
    State(state): State<AppState>,
    User(user): User,
    Path(doc_id): Path<String>,
    Json(b): Json<VotesBody>,
) -> AppResult<Json<serde_json::Value>> {
    let st = docs::load_state(&state, &doc_id).await?;
    docs::require_open(&state, &user, &st.group).await?;
    let way = server_way(&state, &st.group)?;
    if !may_write(&state, &user, &st.group, way).await? {
        return Err(AppError::Forbidden);
    }
    let casts = casts(&st, &b.lines)?;
    let entered = write(&state, &user, &st, way, &casts).await?;
    Ok(Json(json!({ "count": casts.len(), "entered": entered, "simulated": way == ServerWay::Simulated })))
}

/// The simulated confirmer (ADR 0012): while the chain is closed, a vote goes
/// through the same buffer as a vote of the chain — a sealed ballot, simulated
/// (no account, no amount, no wallet), proven at once for the identity of the
/// session, at block 0, then reconciled. Returns what entered into force.
pub(crate) async fn simulate(
    state: &AppState,
    requester: &str,
    st: &DocState,
    casts: &[Cast],
) -> AppResult<Vec<Entered>> {
    let lines: Vec<ballot::Line> = casts
        .iter()
        .map(|c| ballot::Line {
            proposal_id: c.proposal_id.clone(),
            revision: u32::try_from(c.revision).unwrap_or(u32::MAX),
            choice: c.choice,
        })
        .collect();
    let salt = salt();
    let sealed = ballot::seal(&st.doc.id, None, &lines, &salt).map_err(|_| AppError::Invalid("lines"))?;
    let id = util::id();
    let mut tx = state.db.begin().await?;
    sqlx::query(
        "INSERT INTO ballots (id, doc_id, requested_by, author, salt, digest, remark_hash, account, amount, status,
           created_at, expires_at, mode)
         VALUES (?, ?, ?, NULL, ?, ?, ?, '', 0, 'pending', ?, ?, 'simulated')",
    )
    .bind(&id)
    .bind(&st.doc.id)
    .bind(requester)
    .bind(&salt)
    .bind(&sealed.digest)
    .bind(&sealed.remark_hash)
    .bind(util::now())
    .bind(util::in_seconds(EXPIRES_HOURS * 3600))
    .execute(&mut *tx)
    .await?;
    for c in casts {
        sqlx::query("INSERT INTO ballot_lines (ballot_id, proposal_id, revision, choice) VALUES (?, ?, ?, ?)")
            .bind(&id)
            .bind(&c.proposal_id)
            .bind(c.revision)
            .bind(c.choice.as_str())
            .execute(&mut *tx)
            .await?;
    }
    tx.commit().await?;
    let row = load(state, &id).await?;
    settle_simulated(state, &row).await
}

/// Proves a simulated ballot: the voice of the session that asked for it.
pub async fn settle_simulated(state: &AppState, row: &BallotRow) -> AppResult<Vec<Entered>> {
    settle(state, row, Proof { voter: row.requested_by.clone(), position: None, found: None }).await
}

#[derive(Deserialize)]
pub struct DevChainBody {
    open: bool,
}

/// `POST /api/dev/vote-chain` — tests only, with the dev login: closes (or opens
/// again) the chain of this server, to try the simulation (ADR 0012).
pub async fn dev_chain(
    State(state): State<AppState>,
    Json(b): Json<DevChainBody>,
) -> AppResult<Json<serde_json::Value>> {
    if !state.cfg.dev_login {
        return Err(AppError::NotFound);
    }
    state.chain_closed.store(!b.open, Ordering::Relaxed);
    Ok(Json(json!({ "open": state.chain().is_some() })))
}

// ---------------------------------------------------------------------------
// Ballots of the chain
// ---------------------------------------------------------------------------

#[derive(Clone, Debug, sqlx::FromRow)]
pub struct BallotRow {
    pub id: String,
    pub doc_id: String,
    pub requested_by: String,
    pub author: Option<String>,
    pub digest: String,
    pub remark_hash: String,
    pub account: String,
    pub amount: i64,
    pub status: String,
    pub reason: Option<String>,
    pub voter_id: Option<String>,
    pub block: Option<i64>,
    pub entered: Option<String>,
    pub created_at: String,
    pub expires_at: String,
    pub checked_at: Option<String>,
    /// `chain`, or `simulated` (ADR 0012); read as `chain` when absent.
    #[sqlx(default)]
    pub mode: String,
    #[sqlx(default)]
    pub settled_at: Option<String>,
}

const BALLOT_SELECT: &str = "SELECT id, doc_id, requested_by, author, digest, remark_hash, account, amount, status,
    reason, voter_id, block, entered, created_at, expires_at, checked_at, mode, settled_at FROM ballots";

impl BallotRow {
    pub fn comment(&self) -> String {
        format!("{}{}", ballot::PREFIX, self.digest)
    }

    /// A ballot of the simulation, proven by the simulated confirmer.
    pub fn simulated(&self) -> bool {
        self.mode == "simulated"
    }

    pub fn mode(&self) -> &'static str {
        if self.simulated() { "simulated" } else { "chain" }
    }

    /// Where it sits on the chain: never for a simulated ballot (block 0 orders it only).
    pub fn block_shown(&self) -> Option<i64> {
        self.block.filter(|_| !self.simulated())
    }
}

fn salt() -> String {
    let mut bytes = [0u8; 16];
    rand::rng().fill_bytes(&mut bytes);
    hex::encode(bytes)
}

/// `POST /api/docs/{id}/ballots` — seal a ballot for the wallet to sign.
pub async fn create(
    State(state): State<AppState>,
    User(user): User,
    Path(doc_id): Path<String>,
    Json(b): Json<BallotBody>,
) -> AppResult<Json<serde_json::Value>> {
    let st = docs::load_state(&state, &doc_id).await?;
    docs::require_open(&state, &user, &st.group).await?;
    let group = st.group.to_group();
    if trustin_access::vote_way(&group) == VoteWay::Server {
        return Err(AppError::Conflict("server_vote"));
    }
    let granted = docs::granted(&state, &group.slug, &user.id).await?;
    if !trustin_access::asks_ballot(&user.to_identity(), &group, granted) {
        return Err(AppError::Forbidden);
    }
    let chain = state.chain().ok_or(AppError::Conflict("chain_vote_off"))?;
    let casts = casts(&st, &b.lines)?;
    let author = match b.author.as_deref().map(str::trim).filter(|a| !a.is_empty()) {
        Some(a) => Some(address::to_g1(a).ok_or(AppError::Invalid("author"))?),
        None => None,
    };
    // Pending, or let go within the hour: letting a ballot go does not free its place at once.
    let held: i64 = sqlx::query_scalar(
        "SELECT COUNT(*) FROM ballots WHERE requested_by = ? AND mode = 'chain'
            AND (status = 'pending' OR (reason = 'dropped' AND settled_at > ?))",
    )
    .bind(&user.id)
    .bind(util::in_seconds(-DROPPED_WINDOW_SECS))
    .fetch_one(&state.db)
    .await?;
    if held >= MAX_PENDING {
        return Err(AppError::Conflict("too_many_ballots"));
    }
    let lines: Vec<ballot::Line> = casts
        .iter()
        .map(|c| ballot::Line {
            proposal_id: c.proposal_id.clone(),
            revision: u32::try_from(c.revision).unwrap_or(u32::MAX),
            choice: c.choice,
        })
        .collect();
    let salt = salt();
    let sealed = ballot::seal(&st.doc.id, author.as_deref(), &lines, &salt).map_err(|_| AppError::Invalid("lines"))?;
    let id = util::id();
    let now = util::now();
    let expires = util::in_seconds(EXPIRES_HOURS * 3600);
    let mut tx = state.db.begin().await?;
    sqlx::query(
        "INSERT INTO ballots (id, doc_id, requested_by, author, salt, digest, remark_hash, account, amount, status,
           created_at, expires_at)
         VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'pending', ?, ?)",
    )
    .bind(&id)
    .bind(&st.doc.id)
    .bind(&user.id)
    .bind(&author)
    .bind(&salt)
    .bind(&sealed.digest)
    .bind(&sealed.remark_hash)
    .bind(&chain.account)
    .bind(i64::try_from(chain.amount).unwrap_or(i64::MAX))
    .bind(&now)
    .bind(&expires)
    .execute(&mut *tx)
    .await?;
    for c in &casts {
        sqlx::query("INSERT INTO ballot_lines (ballot_id, proposal_id, revision, choice) VALUES (?, ?, ?, ?)")
            .bind(&id)
            .bind(&c.proposal_id)
            .bind(c.revision)
            .bind(c.choice.as_str())
            .execute(&mut *tx)
            .await?;
    }
    tx.commit().await?;
    Ok(Json(json!({ "ballot": {
        "id": id, "status": "pending", "comment": sealed.comment, "account": chain.account,
        "amount": chain.amount, "lines": casts.len(), "author": author, "expiresAt": expires,
    } })))
}

async fn load(state: &AppState, id: &str) -> AppResult<BallotRow> {
    sqlx::query_as::<_, BallotRow>(&format!("{BALLOT_SELECT} WHERE id = ?"))
        .bind(id)
        .fetch_optional(&state.db)
        .await?
        .ok_or(AppError::NotFound)
}

/// `GET /api/ballots/{id}` — where a ballot stands, for the session that asked
/// for it. A pending ballot is looked for on the indexer at most every few seconds.
pub async fn status(
    State(state): State<AppState>,
    User(user): User,
    Path(id): Path<String>,
) -> AppResult<Json<serde_json::Value>> {
    let mut row = load(&state, &id).await?;
    if row.requested_by != user.id {
        return Err(AppError::NotFound);
    }
    if row.status == "pending"
        && (row.simulated() || row.checked_at.as_ref().is_none_or(|c| *c < util::in_seconds(-RECHECK_SECS)))
    {
        check(&state, &row).await?;
        row = load(&state, &id).await?;
    }
    Ok(Json(json!({ "ballot": view(&state, &row).await? })))
}

/// `DELETE /api/ballots/{id}` — the wallet refused to sign, or could not pay:
/// nothing left for the chain, so the ballot is let go at once rather than shown
/// as waiting for a day. Only a pending ballot, only by the session that asked.
pub async fn drop(
    State(state): State<AppState>,
    User(user): User,
    Path(id): Path<String>,
) -> AppResult<Json<serde_json::Value>> {
    let row = load(&state, &id).await?;
    if row.requested_by != user.id {
        return Err(AppError::NotFound);
    }
    sqlx::query(
        "UPDATE ballots SET status = 'expired', reason = 'dropped', settled_at = ? WHERE id = ? AND status = 'pending'",
    )
    .bind(util::now())
    .bind(&id)
    .execute(&state.db)
    .await?;
    let row = load(&state, &id).await?;
    Ok(Json(json!({ "ballot": view(&state, &row).await? })))
}

// ---------------------------------------------------------------------------
// The receipt: the ballot itself, for the one who asked — never anyone else
// ---------------------------------------------------------------------------

/// A title as stored (`{lang: text}`), as JSON; a bare text if it is not a map.
fn title_json(raw: &str) -> serde_json::Value {
    serde_json::from_str::<serde_json::Value>(raw).ok().filter(|v| v.is_object()).unwrap_or_else(|| json!(raw))
}

/// `GET /api/ballots` — the history of the person's ballots, the most recent
/// first: those of its identity only, and not those the wallet never signed.
pub async fn mine(State(state): State<AppState>, User(user): User) -> AppResult<Json<serde_json::Value>> {
    #[derive(sqlx::FromRow)]
    struct Row {
        id: String,
        doc_id: String,
        group_slug: String,
        slug: String,
        title: String,
        digest: String,
        status: String,
        reason: Option<String>,
        block: Option<i64>,
        voter: Option<String>,
        lines: i64,
        created_at: String,
        settled_at: Option<String>,
        mode: String,
    }
    let rows = sqlx::query_as::<_, Row>(&format!(
        "SELECT b.id, b.doc_id, d.group_slug, d.slug, d.title, b.digest, b.status, b.reason, b.block, b.mode,
                i.display_name AS voter, (SELECT COUNT(*) FROM ballot_lines l WHERE l.ballot_id = b.id) AS lines,
                b.created_at, b.settled_at
           FROM ballots b JOIN documents d ON d.id = b.doc_id LEFT JOIN identities i ON i.id = b.voter_id
          WHERE b.requested_by = ? AND IFNULL(b.reason, '') <> 'dropped'
          ORDER BY b.created_at DESC, b.rowid DESC LIMIT {HISTORY}"
    ))
    .bind(&user.id)
    .fetch_all(&state.db)
    .await?;
    let list: Vec<serde_json::Value> = rows
        .into_iter()
        .map(|r| {
            json!({
                "id": r.id, "doc": { "id": r.doc_id, "group": r.group_slug, "slug": r.slug, "title": title_json(&r.title) },
                "comment": format!("{}{}", ballot::PREFIX, r.digest), "status": r.status, "reason": r.reason,
                "block": r.block.filter(|_| r.mode != "simulated"), "voter": r.voter, "lines": r.lines,
                "createdAt": r.created_at, "settledAt": r.settled_at,
                "mode": if r.mode == "simulated" { "simulated" } else { "chain" },
            })
        })
        .collect();
    Ok(Json(json!({ "ballots": list })))
}

/// `GET /api/ballots/{id}/receipt` — the ballot itself, to download: its lines,
/// its salt, the canonical text they make, its seal and where it stands. With
/// it, the person recomputes the seal and finds it on the chain; nobody else
/// ever gets it (the chain shows who voted, never what).
pub async fn receipt(State(state): State<AppState>, User(user): User, Path(id): Path<String>) -> AppResult<Response> {
    #[derive(sqlx::FromRow)]
    struct Extra {
        salt: String,
        signer: Option<String>,
        event_id: Option<String>,
        settled_at: Option<String>,
        title: String,
    }
    #[derive(sqlx::FromRow)]
    struct LineRow {
        proposal_id: String,
        revision: i64,
        choice: String,
        code: String,
        title: Option<String>,
    }
    let row = load(&state, &id).await?;
    if row.requested_by != user.id {
        return Err(AppError::NotFound);
    }
    let extra = sqlx::query_as::<_, Extra>(
        "SELECT b.salt, b.signer, b.event_id, b.settled_at, d.title FROM ballots b JOIN documents d ON d.id = b.doc_id
          WHERE b.id = ?",
    )
    .bind(&id)
    .fetch_one(&state.db)
    .await?;
    let lines = sqlx::query_as::<_, LineRow>(
        "SELECT l.proposal_id, l.revision, l.choice, u.code, p.title FROM ballot_lines l
           JOIN proposals p ON p.id = l.proposal_id JOIN units u ON u.id = p.unit_id
          WHERE l.ballot_id = ? ORDER BY l.proposal_id",
    )
    .bind(&id)
    .fetch_all(&state.db)
    .await?;
    let sealed: Vec<ballot::Line> = lines
        .iter()
        .filter_map(|l| {
            Some(ballot::Line {
                proposal_id: l.proposal_id.clone(),
                revision: u32::try_from(l.revision).ok()?,
                choice: Choice::parse(&l.choice)?,
            })
        })
        .collect();
    let canonical = ballot::canonical(&row.doc_id, row.author.as_deref(), &sealed, &extra.salt)
        .map_err(|e| anyhow::anyhow!("ballot {id} does not read back: {e:?}"))?;
    let seal = ballot::seal(&row.doc_id, row.author.as_deref(), &sealed, &extra.salt)
        .map_err(|e| anyhow::anyhow!("ballot {id} does not read back: {e:?}"))?;
    if seal.digest != row.digest || sealed.len() != lines.len() {
        return Err(anyhow::anyhow!("ballot {id}: its lines no longer make its seal").into());
    }
    let voter: Option<String> = match &row.voter_id {
        Some(v) => {
            sqlx::query_scalar("SELECT display_name FROM identities WHERE id = ?")
                .bind(v)
                .fetch_optional(&state.db)
                .await?
        }
        None => None,
    };
    let body = json!({
        "format": "ld1",
        "id": row.id,
        "doc": row.doc_id,
        "docTitle": title_json(&extra.title),
        "author": row.author,
        "lines": lines.iter().map(|l| json!({
            "proposalId": l.proposal_id, "revision": l.revision, "choice": l.choice, "article": l.code, "title": l.title,
        })).collect::<Vec<_>>(),
        "salt": extra.salt,
        "canonical": canonical,
        "digest": seal.digest,
        "comment": seal.comment,
        "remarkHash": seal.remark_hash,
        "account": row.account,
        "amount": row.amount,
        "status": row.status,
        "reason": row.reason,
        "mode": row.mode(),
        "block": row.block_shown(),
        "event": extra.event_id,
        "signer": extra.signer,
        "voter": voter,
        "createdAt": row.created_at,
        "settledAt": extra.settled_at,
    });
    let day = row.created_at.get(..10).unwrap_or("vote");
    let name = format!("recu-vote-{day}-{}.json", row.id.get(..8).unwrap_or(&row.id));
    let mut res = Json(body).into_response();
    let headers = res.headers_mut();
    if let Ok(v) = HeaderValue::from_str(&format!("attachment; filename=\"{name}\"")) {
        headers.insert(header::CONTENT_DISPOSITION, v);
    }
    headers.insert(header::CACHE_CONTROL, HeaderValue::from_static("no-store"));
    Ok(res)
}

async fn view(state: &AppState, row: &BallotRow) -> AppResult<serde_json::Value> {
    let voter: Option<String> = match &row.voter_id {
        Some(v) => {
            sqlx::query_scalar("SELECT display_name FROM identities WHERE id = ?")
                .bind(v)
                .fetch_optional(&state.db)
                .await?
        }
        None => None,
    };
    let entered: serde_json::Value =
        row.entered.as_deref().and_then(|e| serde_json::from_str(e).ok()).unwrap_or_else(|| json!([]));
    Ok(json!({
        "id": row.id, "status": row.status, "reason": row.reason, "comment": row.comment(),
        "account": row.account, "amount": row.amount, "author": row.author, "block": row.block_shown(),
        "voter": voter, "entered": entered, "expiresAt": row.expires_at, "mode": row.mode(),
    }))
}

/// Reads the indexer for one pending ballot and settles it if the chain has spoken
/// — in a final block. At its end, a ballot is read one last time, then let go.
pub async fn check(state: &AppState, row: &BallotRow) -> AppResult<()> {
    if row.simulated() {
        return settle_simulated(state, row).await.map(|_| ());
    }
    let now = util::now();
    let due = row.expires_at <= now;
    let past_grace = row.expires_at <= util::in_seconds(-GRACE_SECS);
    let (Some((_, account)), Some(amount)) = (address::decode(&row.account), u64::try_from(row.amount).ok()) else {
        return Err(anyhow::anyhow!("ballot {} has an unreadable account or amount", row.id).into());
    };
    let expected =
        Expected { account, amount, author: row.author.as_deref().and_then(address::decode).map(|(_, k)| k) };
    let comment = row.comment();
    // Only the comments that carry exactly the remark of the ballot.
    let judged = |list: &[Found]| {
        let mine: Vec<ballot::Comment> =
            list.iter().filter(|f| f.remark == comment).map(|f| f.comment.clone()).collect();
        ballot::judge(&mine, &expected)
    };
    let filter = indexer::filter(&row.account, amount, row.author.as_deref());
    let reading = indexer::comments(state, &row.remark_hash, &filter, |list| judged(list).is_some()).await;
    sqlx::query("UPDATE ballots SET checked_at = ? WHERE id = ?").bind(&now).bind(&row.id).execute(&state.db).await?;
    let (list, unfinal) = match reading {
        Reading::Found { comments, unfinal } => (comments, unfinal),
        Reading::Silent => {
            tracing::debug!(ballot = %row.id, "no indexer answered: the ballot stays pending");
            if past_grace {
                let_go(state, row).await?;
            }
            return Ok(());
        }
    };
    let Some((position, verdict)) = judged(&list) else {
        // Nothing final settles it. At its end: let go, unless a comment waits for its block to be final.
        if due && (!unfinal || past_grace) {
            let_go(state, row).await?;
        }
        return Ok(());
    };
    let Some(found) = list.iter().find(|f| f.comment.position == position && f.remark == comment) else {
        return Ok(());
    };
    match verdict {
        // The right to vote is checked now, at confirmation, by trustinWallet's rule.
        Verdict::Paid { identity } if trustin_access::chain_vote_counts(identity.member) => {
            let proof = Proof {
                voter: format!("idty:{}", identity.index),
                position: Some(found.comment.position),
                found: Some(found),
            };
            settle(state, row, proof).await.map(|_| ())
        }
        Verdict::Paid { .. } => refuse(state, row, found, "not_member").await,
        other => refuse(state, row, found, other.reason().unwrap_or("refused")).await,
    }
}

async fn refuse(state: &AppState, row: &BallotRow, found: &Found, reason: &str) -> AppResult<()> {
    let done = sqlx::query(
        "UPDATE ballots SET status = 'refused', reason = ?, signer = ?, block = ?, event_index = ?, event_id = ?,
           settled_at = ? WHERE id = ? AND status = 'pending'",
    )
    .bind(reason)
    .bind(&found.signer)
    .bind(i64::from(found.comment.position.block))
    .bind(i64::from(found.comment.position.event))
    .bind(&found.event_id)
    .bind(util::now())
    .bind(&row.id)
    .execute(&state.db)
    .await?;
    if done.rows_affected() == 1 {
        tracing::info!(ballot = %row.id, reason, "ballot refused");
        tell(state, &row.id, &row.requested_by, &row.doc_id, "ballot_refused").await?;
    }
    Ok(())
}

/// Tells the session that asked for a ballot what became of it (ADR 0012): a
/// message of its box, without gesture — one votes again in the text.
async fn tell(state: &AppState, ballot_id: &str, requester: &str, doc_id: &str, kind: &str) -> AppResult<()> {
    messages::notify(state, requester, doc_id, Topic::Ballot, ballot_id, kind, 1, None).await
}

/// The proof of a ballot: whose voice it is, and where it sits on the chain.
pub struct Proof<'a> {
    /// `idty:<index>` of the signer; for a simulated ballot, the identity of the
    /// session that asked (there is no wallet).
    pub voter: String,
    /// `None`: a simulated ballot — block 0, at the next rank of the simulation.
    pub position: Option<ChainPosition>,
    /// What the chain said of the comment (account, name, event); none in simulation.
    pub found: Option<&'a Found>,
}

/// The ballot is proven: confirmed for its voter, then each of its versions is
/// reconciled (`reconcile`). A *for* proven while an objection holds its
/// version back is set aside, and told (ADR 0012, revising 0010: it no longer
/// counts « as it was signed »). Proving twice changes nothing. Returns what
/// entered into force.
async fn settle(state: &AppState, row: &BallotRow, proof: Proof<'_>) -> AppResult<Vec<Entered>> {
    let now = util::now();
    // The versions an objection holds back now (read before the transaction: one writer).
    let held = docs::load_state(state, &row.doc_id).await?.blocked;
    let mut set_aside = false;
    let position = {
        let _one = CONFIRM.lock().await;
        let mut tx = state.db.begin().await?;
        if let Some(found) = proof.found {
            // The identity of the signer, as the chain has it now: a member (just
            // checked), with its name and its account. Undone with the rest if the
            // ballot was settled in the meantime.
            let admin = state.cfg.admin_g1_ids.contains(&proof.voter);
            sqlx::query(
                "INSERT INTO identities (id, kind, display_name, address, g1_member, admin, created_at)
                 VALUES (?, 'g1', ?, ?, 1, ?, ?)
                 ON CONFLICT(id) DO UPDATE SET display_name = excluded.display_name, address = excluded.address,
                   g1_member = 1, admin = MAX(identities.admin, excluded.admin)",
            )
            .bind(&proof.voter)
            .bind(found.name.clone().unwrap_or_else(|| proof.voter.clone()))
            .bind(&found.signer)
            .bind(admin)
            .bind(&now)
            .execute(&mut *tx)
            .await?;
        }
        let position = match proof.position {
            Some(p) => p,
            None => {
                let last: i64 = sqlx::query_scalar(
                    "SELECT IFNULL(MAX(event_index), 0) FROM ballots WHERE mode = 'simulated' AND status = 'confirmed'",
                )
                .fetch_one(&mut *tx)
                .await?;
                ChainPosition { block: 0, event: u32::try_from(last.saturating_add(1)).unwrap_or(u32::MAX) }
            }
        };
        let won = sqlx::query(
            "UPDATE ballots SET status = 'confirmed', voter_id = ?, signer = ?, block = ?, event_index = ?,
               event_id = ?, settled_at = ? WHERE id = ? AND status = 'pending'",
        )
        .bind(&proof.voter)
        .bind(proof.found.map(|f| &f.signer))
        .bind(i64::from(position.block))
        .bind(i64::from(position.event))
        .bind(proof.found.map(|f| &f.event_id))
        .bind(&now)
        .bind(&row.id)
        .execute(&mut *tx)
        .await?;
        if won.rows_affected() != 1 {
            // Settled in the meantime: proving twice changes nothing.
            tx.rollback().await?;
            return Ok(Vec::new());
        }
        let lines: Vec<(String, String)> =
            sqlx::query_as("SELECT proposal_id, choice FROM ballot_lines WHERE ballot_id = ?")
                .bind(&row.id)
                .fetch_all(&mut *tx)
                .await?;
        for (proposal_id, choice) in &lines {
            let Some(choice) = Choice::parse(choice) else {
                tracing::warn!(ballot = %row.id, %choice, "unknown choice in a ballot line");
                continue;
            };
            if buffer::discarded_by_objection(choice, held.contains(proposal_id)) {
                sqlx::query("UPDATE ballot_lines SET outcome = ? WHERE ballot_id = ? AND proposal_id = ?")
                    .bind(Outcome::Objection.as_str())
                    .bind(&row.id)
                    .bind(proposal_id)
                    .execute(&mut *tx)
                    .await?;
                set_aside = true;
            }
        }
        for (proposal_id, _) in &lines {
            reconcile(&mut tx, &proof.voter, proposal_id, &now).await?;
        }
        tx.commit().await?;
        position
    };
    tracing::info!(ballot = %row.id, mode = row.mode(), block = position.block, event = position.event, "ballot proven");
    let entered: Vec<Entered> = adoption::run(state, &row.doc_id).await?;
    sqlx::query("UPDATE ballots SET entered = ? WHERE id = ?")
        .bind(serde_json::to_string(&entered).unwrap_or_else(|_| "[]".into()))
        .bind(&row.id)
        .execute(&state.db)
        .await?;
    if set_aside {
        tell(state, &row.id, &row.requested_by, &row.doc_id, "ballot_objection").await?;
    }
    Ok(entered)
}

/// The reconciliation of one identity on one version (ADR 0012): every proven
/// line of it, read again; one voice, the last in the order of the chain
/// (`ld_engine::buffer::reconcile`), whatever the order the proofs came in. Each
/// line is told what became of it; the voice that stands is written in `votes`
/// — or, every line set aside, the vote of before is left as it is.
pub async fn reconcile(conn: &mut SqliteConnection, voter: &str, proposal_id: &str, now: &str) -> AppResult<()> {
    #[derive(sqlx::FromRow)]
    struct Row {
        ballot_id: String,
        mode: String,
        block: Option<i64>,
        event_index: Option<i64>,
        revision: i64,
        choice: String,
        outcome: Option<String>,
    }
    let rows = sqlx::query_as::<_, Row>(
        "SELECT b.id AS ballot_id, b.mode, b.block, b.event_index, l.revision, l.choice, l.outcome
           FROM ballots b JOIN ballot_lines l ON l.ballot_id = b.id
          WHERE b.status = 'confirmed' AND b.voter_id = ? AND l.proposal_id = ?
          ORDER BY b.block, b.event_index, b.created_at, b.rowid",
    )
    .bind(voter)
    .bind(proposal_id)
    .fetch_all(&mut *conn)
    .await?;
    let mut kept: Vec<(&Row, Proven)> = Vec::with_capacity(rows.len());
    for r in &rows {
        let position = r
            .block
            .zip(r.event_index)
            .and_then(|(b, e)| Some(ChainPosition { block: u32::try_from(b).ok()?, event: u32::try_from(e).ok()? }));
        let (Some(position), Some(choice)) = (position, Choice::parse(&r.choice)) else {
            tracing::warn!(ballot = %r.ballot_id, "a proven line that does not read back");
            continue;
        };
        let held = r.outcome.as_deref() == Some(Outcome::Objection.as_str());
        kept.push((r, Proven { position, choice, held }));
    }
    let lines: Vec<Proven> = kept.iter().map(|(_, p)| *p).collect();
    let done = buffer::reconcile(&lines);
    for ((r, _), outcome) in kept.iter().zip(&done.outcomes) {
        if r.outcome.as_deref() != Some(outcome.as_str()) {
            sqlx::query("UPDATE ballot_lines SET outcome = ? WHERE ballot_id = ? AND proposal_id = ?")
                .bind(outcome.as_str())
                .bind(&r.ballot_id)
                .bind(proposal_id)
                .execute(&mut *conn)
                .await?;
        }
    }
    let Some((r, line)) = done.standing.and_then(|i| kept.get(i)) else { return Ok(()) };
    match line.choice {
        Choice::None => {
            sqlx::query("DELETE FROM votes WHERE proposal_id = ? AND identity_id = ?")
                .bind(proposal_id)
                .bind(voter)
                .execute(&mut *conn)
                .await?;
        }
        c => {
            // Only when it changes: the same ballot standing again keeps its row as it is.
            sqlx::query(
                "INSERT INTO votes (proposal_id, identity_id, choice, revision, cast_at, source, block, event_index, ballot_id)
                 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
                 ON CONFLICT(proposal_id, identity_id) DO UPDATE SET choice = excluded.choice,
                   revision = excluded.revision, cast_at = excluded.cast_at, source = excluded.source,
                   block = excluded.block, event_index = excluded.event_index, ballot_id = excluded.ballot_id
                 WHERE votes.ballot_id IS NOT excluded.ballot_id",
            )
            .bind(proposal_id)
            .bind(voter)
            .bind(c.as_str())
            .bind(r.revision)
            .bind(now)
            .bind(if r.mode == "simulated" { "simulated" } else { "chain" })
            .bind(i64::from(line.position.block))
            .bind(i64::from(line.position.event))
            .bind(&r.ballot_id)
            .execute(&mut *conn)
            .await?;
        }
    }
    Ok(())
}

/// Lets go of one pending ballot not seen on the chain in time.
async fn let_go(state: &AppState, row: &BallotRow) -> AppResult<()> {
    let done = sqlx::query("UPDATE ballots SET status = 'expired', settled_at = ? WHERE id = ? AND status = 'pending'")
        .bind(util::now())
        .bind(&row.id)
        .execute(&state.db)
        .await?;
    if done.rows_affected() == 1 {
        tell(state, &row.id, &row.requested_by, &row.doc_id, "ballot_expired").await?;
    }
    Ok(())
}

/// Lets go of the pending ballots past their end and the grace after it: those
/// the indexer could not be read for in time.
pub async fn expire(state: &AppState) -> AppResult<u64> {
    let gone: Vec<(String, String, String)> = sqlx::query_as(
        "UPDATE ballots SET status = 'expired', settled_at = ?
          WHERE status = 'pending' AND mode = 'chain' AND expires_at <= ?
         RETURNING id, requested_by, doc_id",
    )
    .bind(util::now())
    .bind(util::in_seconds(-GRACE_SECS))
    .fetch_all(&state.db)
    .await?;
    for (id, requester, doc_id) in &gone {
        tell(state, id, requester, doc_id, "ballot_expired").await?;
    }
    Ok(u64::try_from(gone.len()).unwrap_or(u64::MAX))
}

/// Purges the ballots let go long enough ago, lines and all, and the messages
/// that told of them. A ballot let go never counted: nothing of a vote is lost.
pub async fn purge(state: &AppState) -> AppResult<u64> {
    const WHICH: &str = "SELECT id FROM ballots WHERE status = 'expired' AND settled_at IS NOT NULL
        AND ((reason = 'dropped' AND settled_at < ?) OR settled_at < ?)";
    let (dropped, expired) = (util::in_seconds(-PURGE_DROPPED_SECS), util::in_seconds(-PURGE_EXPIRED_SECS));
    let mut tx = state.db.begin().await?;
    sqlx::query(&format!("DELETE FROM messages WHERE topic = 'ballot' AND subject_id IN ({WHICH})"))
        .bind(&dropped)
        .bind(&expired)
        .execute(&mut *tx)
        .await?;
    sqlx::query(&format!("DELETE FROM ballot_lines WHERE ballot_id IN ({WHICH})"))
        .bind(&dropped)
        .bind(&expired)
        .execute(&mut *tx)
        .await?;
    let done = sqlx::query(&format!("DELETE FROM ballots WHERE id IN ({WHICH})"))
        .bind(&dropped)
        .bind(&expired)
        .execute(&mut *tx)
        .await?;
    tx.commit().await?;
    Ok(done.rows_affected())
}

/// One pass of the background task: let go and purge, then read the indexer for
/// the pending ballots, the least recently read first (never read, then the
/// oldest reading). Returns how many were read.
pub async fn sweep(state: &AppState) -> AppResult<usize> {
    expire(state).await?;
    purge(state).await?;
    // The simulated confirmer: a simulated ballot is proven in its own request; one
    // left waiting (a stop in between) is proven here.
    let simulated =
        sqlx::query_as::<_, BallotRow>(&format!("{BALLOT_SELECT} WHERE status = 'pending' AND mode = 'simulated'"))
            .fetch_all(&state.db)
            .await?;
    for row in &simulated {
        if let Err(e) = settle_simulated(state, row).await {
            tracing::warn!(ballot = %row.id, error = %e, "simulated ballot");
        }
    }
    let rows = sqlx::query_as::<_, BallotRow>(&format!(
        "{BALLOT_SELECT} WHERE status = 'pending' AND mode = 'chain'
          ORDER BY checked_at IS NOT NULL, checked_at, created_at, rowid LIMIT {PER_PASS}"
    ))
    .fetch_all(&state.db)
    .await?;
    for row in &rows {
        if let Err(e) = check(state, row).await {
            tracing::warn!(ballot = %row.id, error = %e, "ballot check");
        }
    }
    Ok(rows.len())
}

pub async fn watcher(state: AppState) {
    let mut tick = tokio::time::interval(WATCH_EVERY);
    loop {
        tick.tick().await;
        if let Err(e) = sweep(&state).await {
            tracing::warn!(error = %e, "ballots");
        }
    }
}

// ---------------------------------------------------------------------------
// « Ton vote »
// ---------------------------------------------------------------------------

/// A vote of the person, on one version: hers, or cast by a ballot she asked for.
#[derive(Clone, Debug, sqlx::FromRow)]
pub struct MyVote {
    pub proposal_id: String,
    pub choice: String,
    pub revision: i64,
    pub source: String,
}

/// The votes `me` sees as « ton vote » in a document: those of its own identity,
/// and those a ballot it asked for cast on the chain (for the identity of the
/// wallet that signed). Only those: a session never sees what another session of
/// the same Ğ1 identity voted. The more recent on the chain comes last and wins.
pub async fn my_votes(state: &AppState, me: &str, doc_id: &str) -> AppResult<HashMap<String, MyVote>> {
    let rows = sqlx::query_as::<_, MyVote>(
        "SELECT v.proposal_id, v.choice, v.revision, v.source FROM votes v
           JOIN proposals p ON p.id = v.proposal_id JOIN units u ON u.id = p.unit_id
          WHERE u.doc_id = ?
            AND (v.identity_id = ? OR v.ballot_id IN (SELECT id FROM ballots WHERE requested_by = ? AND status = 'confirmed'))
          ORDER BY IFNULL(v.block, 0), IFNULL(v.event_index, 0), v.cast_at",
    )
    .bind(doc_id)
    .bind(me)
    .bind(me)
    .fetch_all(&state.db)
    .await?;
    Ok(rows.into_iter().map(|v| (v.proposal_id.clone(), v)).collect())
}

/// What a ballot of the person, not counted, says of one version: pending on the
/// chain (an intention, ADR 0012), refused, let go — or its *for* set aside by an
/// objection (`refused`, reason `objection`).
#[derive(Clone, Debug, serde::Serialize)]
#[serde(rename_all = "camelCase")]
pub struct MyBallot {
    pub id: String,
    pub status: String,
    pub reason: Option<String>,
    pub choice: String,
    /// The revision the line votes.
    #[serde(skip)]
    pub revision: i64,
}

/// Per version: the most recent pending ballot of `me` asked after its last
/// confirmed one; else its most recent ballot if that one was refused or let go,
/// or proven with its line set aside by an objection (a confirmed one shows as its
/// vote). A ballot the wallet never signed (dropped) says nothing; one asked
/// before a confirmed vote says nothing either.
pub async fn my_ballots(state: &AppState, me: &str, doc_id: &str) -> AppResult<HashMap<String, MyBallot>> {
    #[derive(sqlx::FromRow)]
    struct Row {
        proposal_id: String,
        id: String,
        status: String,
        reason: Option<String>,
        choice: String,
        revision: i64,
        outcome: Option<String>,
    }
    let rows = sqlx::query_as::<_, Row>(
        "SELECT l.proposal_id, b.id, b.status, b.reason, l.choice, l.revision, l.outcome FROM ballots b
           JOIN ballot_lines l ON l.ballot_id = b.id
          WHERE b.doc_id = ? AND b.requested_by = ? AND IFNULL(b.reason, '') <> 'dropped'
          ORDER BY b.created_at, b.rowid",
    )
    .bind(doc_id)
    .bind(me)
    .fetch_all(&state.db)
    .await?;
    let mut latest: HashMap<String, MyBallot> = HashMap::new();
    let mut pending: HashMap<String, MyBallot> = HashMap::new();
    for r in rows {
        let set_aside = r.status == "confirmed" && r.outcome.as_deref() == Some(Outcome::Objection.as_str());
        let b = MyBallot {
            id: r.id,
            status: if set_aside { "refused".into() } else { r.status },
            reason: if set_aside { Some(Outcome::Objection.as_str().into()) } else { r.reason },
            choice: r.choice,
            revision: r.revision,
        };
        match b.status.as_str() {
            // A confirmed vote: what was asked before it says nothing more.
            "confirmed" => {
                latest.remove(&r.proposal_id);
                pending.remove(&r.proposal_id);
            }
            "pending" => {
                pending.insert(r.proposal_id, b);
            }
            _ => {
                if set_aside {
                    pending.remove(&r.proposal_id);
                }
                latest.insert(r.proposal_id, b);
            }
        }
    }
    latest.extend(pending);
    Ok(latest)
}

/// The latest intention of a session on a version, still waiting for its proof.
pub struct Intent {
    pub requester: String,
    pub proposal_id: String,
    pub revision: i64,
    pub choice: Choice,
}

/// The intentions of a document (ADR 0012): for each session and each version,
/// its most recent pending ballot of the chain asked after its last confirmed one
/// on that version — the rule of `my_ballots`, for everyone.
pub async fn intents(state: &AppState, doc_id: &str) -> AppResult<Vec<Intent>> {
    #[derive(sqlx::FromRow)]
    struct Row {
        requested_by: String,
        proposal_id: String,
        revision: i64,
        choice: String,
    }
    let rows = sqlx::query_as::<_, Row>(
        "SELECT b.requested_by, l.proposal_id, l.revision, l.choice FROM ballots b
           JOIN ballot_lines l ON l.ballot_id = b.id
          WHERE b.doc_id = ? AND b.status = 'pending' AND b.mode = 'chain'
            AND NOT EXISTS (SELECT 1 FROM ballots c JOIN ballot_lines cl ON cl.ballot_id = c.id
                             WHERE c.requested_by = b.requested_by AND cl.proposal_id = l.proposal_id
                               AND c.status = 'confirmed'
                               AND (c.created_at > b.created_at OR (c.created_at = b.created_at AND c.rowid > b.rowid)))
          ORDER BY b.created_at, b.rowid",
    )
    .bind(doc_id)
    .fetch_all(&state.db)
    .await?;
    let mut latest: HashMap<(String, String), Intent> = HashMap::new();
    for r in rows {
        let Some(choice) = Choice::parse(&r.choice) else { continue };
        let key = (r.requested_by.clone(), r.proposal_id.clone());
        latest.insert(
            key,
            Intent { requester: r.requested_by, proposal_id: r.proposal_id, revision: r.revision, choice },
        );
    }
    Ok(latest.into_values().collect())
}

/// « Ton vote » as the person sees it now (ADR 0012): its latest gesture — an
/// intention still waiting for the chain — or the vote that counts for it.
pub struct Seen {
    pub choice: Option<String>,
    /// `recorded`, `pending`, `simulated`, or none.
    pub state: Option<&'static str>,
    /// It votes a revision gone by: « to reconfirm ».
    pub stale: bool,
    /// Proven on the chain.
    pub chain: bool,
}

pub fn seen(revision: i64, vote: Option<&MyVote>, ballot: Option<&MyBallot>) -> Seen {
    if let Some(b) = ballot.filter(|b| b.status == "pending") {
        let choice = (b.choice != Choice::None.as_str()).then(|| b.choice.clone());
        return Seen {
            stale: choice.is_some() && b.revision != revision,
            choice,
            state: Some(Stage::Pending.as_str()),
            chain: false,
        };
    }
    match vote {
        Some(v) => Seen {
            choice: Some(v.choice.clone()),
            state: Some(Stage::of_source(&v.source).as_str()),
            stale: v.revision != revision,
            chain: v.source == "chain",
        },
        None => Seen { choice: None, state: None, stale: false, chain: false },
    }
}

/// A ballot, for a message of the box of the session that asked for it (ADR
/// 0012): what became of it, line by line. No gesture: one votes again in the text.
pub async fn for_message(state: &AppState, viewer: &IdentityRow, id: &str) -> AppResult<Option<serde_json::Value>> {
    #[derive(sqlx::FromRow)]
    struct LineRow {
        proposal_id: String,
        unit_id: String,
        code: String,
        title: String,
        choice: String,
        outcome: Option<String>,
    }
    let row = sqlx::query_as::<_, BallotRow>(&format!("{BALLOT_SELECT} WHERE id = ?"))
        .bind(id)
        .fetch_optional(&state.db)
        .await?;
    let Some(row) = row.filter(|r| r.requested_by == viewer.id) else { return Ok(None) };
    let lines = sqlx::query_as::<_, LineRow>(
        "SELECT l.proposal_id, p.unit_id, u.code, p.title, l.choice, l.outcome FROM ballot_lines l
           JOIN proposals p ON p.id = l.proposal_id JOIN units u ON u.id = p.unit_id
          WHERE l.ballot_id = ? ORDER BY u.position, u.id",
    )
    .bind(id)
    .fetch_all(&state.db)
    .await?;
    Ok(Some(json!({
        "id": row.id, "status": row.status, "reason": row.reason, "mode": row.mode(),
        "createdAt": row.created_at, "settledAt": row.settled_at,
        "lines": lines.iter().map(|l| json!({
            "proposalId": l.proposal_id, "unitId": l.unit_id, "code": l.code, "title": l.title,
            "choice": l.choice, "outcome": l.outcome,
        })).collect::<Vec<_>>(),
    })))
}

/// How this session votes in a group, and whether it may now.
pub async fn access(state: &AppState, user: &IdentityRow, group: &docs::GroupRow) -> AppResult<serde_json::Value> {
    let g = group.to_group();
    let granted = docs::granted(state, &g.slug, &user.id).await?;
    let id = user.to_identity();
    let way = trustin_access::vote_way(&g);
    // While the chain is closed, a group of the chain votes in simulation: the vote
    // leaves the page by a write of the server, marked as simulated (ADR 0012).
    let simulated = way == VoteWay::Chain && state.chain().is_none();
    let open = match way {
        VoteWay::Server => trustin_access::votes_in(&id, &g, granted),
        VoteWay::Chain if simulated => trustin_access::casts_simulated(&id, &g, granted, false),
        VoteWay::Chain => trustin_access::asks_ballot(&id, &g, granted),
    };
    let reason = (!open).then_some("members_only");
    let chain = match (way, state.chain()) {
        (VoteWay::Chain, Some(c)) => json!({
            "account": c.account, "amount": c.amount, "nodes": c.nodes,
            "test": c.test_node.is_some(), "testNode": c.test_node, "expiresHours": EXPIRES_HOURS,
        }),
        _ => serde_json::Value::Null,
    };
    let leaves = if simulated { VoteWay::Server } else { way };
    Ok(json!({ "way": leaves, "open": open, "simulated": simulated, "reason": reason, "chain": chain }))
}