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

//! 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).

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::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 rand::RngCore;
use serde::Deserialize;
use serde_json::json;
use std::collections::{HashMap, HashSet};
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 votable version, 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 votable.
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"));
        }
        // Held back by an objection: no new vote, on the server or on the chain (ADR 0010).
        if !st.votable(p) {
            return Err(AppError::Conflict("objection"));
        }
        out.push(Cast { proposal_id: p.id.clone(), revision: p.revision, choice: parse_choice(l.choice.as_deref())? });
    }
    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(())
}

/// `POST /api/docs/{id}/votes` — a grouped vote in a managed group: one write.
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?;
    if trustin_access::vote_way(&st.group.to_group()) == VoteWay::Chain {
        return Err(AppError::Conflict("chain_only"));
    }
    if !docs::may_vote(&state, &user, &st.group).await? {
        return Err(AppError::Forbidden);
    }
    let casts = casts(&st, &b.lines)?;
    write_server_votes(&state, &user.id, &casts).await?;
    let entered = adoption::run(&state, &st.doc.id).await?;
    Ok(Json(json!({ "count": casts.len(), "entered": entered })))
}

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

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 FROM ballots";

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

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.cfg.vote.as_ref().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 (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.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>,
    }
    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,
                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, "voter": r.voter, "lines": r.lines, "createdAt": r.created_at, "settledAt": r.settled_at,
            })
        })
        .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,
        "block": row.block,
        "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,
        "voter": voter, "entered": entered, "expiresAt": row.expires_at,
    }))
}

/// 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<()> {
    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) => {
            confirm(state, row, found, identity.index).await
        }
        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");
    }
    Ok(())
}

/// The ballot counts: written for the identity of the signer, line by line,
/// where no more recent ballot of that identity decided already.
async fn confirm(state: &AppState, row: &BallotRow, found: &Found, index: u32) -> AppResult<()> {
    let voter = format!("idty:{index}");
    let position = found.comment.position;
    let now = util::now();
    {
        let _one = CONFIRM.lock().await;
        let mut tx = state.db.begin().await?;
        // 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(&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(&voter)
        .bind(found.name.clone().unwrap_or_else(|| voter.clone()))
        .bind(&found.signer)
        .bind(admin)
        .bind(&now)
        .execute(&mut *tx)
        .await?;
        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(&voter)
        .bind(&found.signer)
        .bind(i64::from(position.block))
        .bind(i64::from(position.event))
        .bind(&found.event_id)
        .bind(&now)
        .bind(&row.id)
        .execute(&mut *tx)
        .await?;
        if won.rows_affected() != 1 {
            // Settled in the meantime: confirming twice changes nothing.
            tx.rollback().await?;
            return Ok(());
        }
        let lines: Vec<(String, i64, String)> =
            sqlx::query_as("SELECT proposal_id, revision, choice FROM ballot_lines WHERE ballot_id = ?")
                .bind(&row.id)
                .fetch_all(&mut *tx)
                .await?;
        for (proposal_id, revision, choice) in lines {
            let others: Vec<(i64, i64)> = sqlx::query_as(
                "SELECT b.block, b.event_index 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 = ? AND b.id <> ?
                    AND b.block IS NOT NULL AND b.event_index IS NOT NULL",
            )
            .bind(&voter)
            .bind(&proposal_id)
            .bind(&row.id)
            .fetch_all(&mut *tx)
            .await?;
            let others: Vec<ChainPosition> = others
                .into_iter()
                .filter_map(|(b, e)| {
                    Some(ChainPosition { block: u32::try_from(b).ok()?, event: u32::try_from(e).ok()? })
                })
                .collect();
            if !ballot::decides(position, &others) {
                continue;
            }
            match Choice::parse(&choice) {
                Some(Choice::None) => {
                    sqlx::query("DELETE FROM votes WHERE proposal_id = ? AND identity_id = ?")
                        .bind(&proposal_id)
                        .bind(&voter)
                        .execute(&mut *tx)
                        .await?;
                }
                Some(c) => {
                    sqlx::query(
                        "INSERT INTO votes (proposal_id, identity_id, choice, revision, cast_at, source, block, event_index, ballot_id)
                         VALUES (?, ?, ?, ?, ?, 'chain', ?, ?, ?)
                         ON CONFLICT(proposal_id, identity_id) DO UPDATE SET choice = excluded.choice,
                           revision = excluded.revision, cast_at = excluded.cast_at, source = 'chain',
                           block = excluded.block, event_index = excluded.event_index, ballot_id = excluded.ballot_id",
                    )
                    .bind(&proposal_id)
                    .bind(&voter)
                    .bind(c.as_str())
                    .bind(revision)
                    .bind(&now)
                    .bind(i64::from(position.block))
                    .bind(i64::from(position.event))
                    .bind(&row.id)
                    .execute(&mut *tx)
                    .await?;
                }
                None => tracing::warn!(ballot = %row.id, %choice, "unknown choice in a ballot line"),
            }
        }
        tx.commit().await?;
    }
    tracing::info!(ballot = %row.id, block = position.block, event = position.event, "ballot confirmed");
    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?;
    Ok(())
}

/// Lets go of one pending ballot not seen on the chain in time.
async fn let_go(state: &AppState, row: &BallotRow) -> AppResult<()> {
    sqlx::query("UPDATE ballots SET status = 'expired', settled_at = ? WHERE id = ? AND status = 'pending'")
        .bind(util::now())
        .bind(&row.id)
        .execute(&state.db)
        .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 done = sqlx::query(
        "UPDATE ballots SET status = 'expired', settled_at = ? WHERE status = 'pending' AND expires_at <= ?",
    )
    .bind(util::now())
    .bind(util::in_seconds(-GRACE_SECS))
    .execute(&state.db)
    .await?;
    Ok(done.rows_affected())
}

/// Purges the ballots let go long enough ago, lines and all. 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 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?;
    let rows = sqlx::query_as::<_, BallotRow>(&format!(
        "{BALLOT_SELECT} WHERE status = 'pending'
          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 yet counted, says of one version: pending on
/// the chain, or refused, or let go.
#[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,
}

/// 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
/// (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,
    }
    let rows = sqlx::query_as::<_, Row>(
        "SELECT l.proposal_id, b.id, b.status, b.reason, l.choice 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 b = MyBallot { id: r.id, status: r.status, reason: r.reason, choice: r.choice };
        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);
            }
            _ => {
                latest.insert(r.proposal_id, b);
            }
        }
    }
    latest.extend(pending);
    Ok(latest)
}

/// 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);
    let (open, reason) = match way {
        VoteWay::Server => {
            let open = trustin_access::votes_in(&id, &g, granted);
            (open, (!open).then_some("members_only"))
        }
        VoteWay::Chain => match (&state.cfg.vote, trustin_access::asks_ballot(&id, &g, granted)) {
            (None, _) => (false, Some("chain_off")),
            (Some(_), false) => (false, Some("members_only")),
            (Some(_), true) => (true, None),
        },
    };
    let chain = match (way, &state.cfg.vote) {
        (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,
    };
    Ok(json!({ "way": way, "open": open, "reason": reason, "chain": chain }))
}