//! Entering into force — automatic, as soon as a proposal reaches its //! threshold (Yvv, 2026-10-21: « Dès qu'une proposition atteint le score, elle //! remonte en vigueur »). The human gesture is the vote itself; the engine //! only counts. Every entry is recorded with its numbers and queued for //! publication to the texts repository. use crate::AppState; use crate::docs::{self, DocRow, DocState}; use crate::error::AppResult; use crate::util; use ld_engine::compose::{self, ProposalKind, UnitState}; use serde::Serialize; use std::time::Duration; use tokio::sync::Mutex; /// One adoption pass at a time per process: SQLite has a single writer, and /// two concurrent votes must not both promote rivals of the same unit. static PASS: Mutex<()> = Mutex::const_new(()); #[derive(Debug, Clone, Serialize)] #[serde(rename_all = "camelCase")] pub struct Entered { pub proposal_id: String, pub unit_id: String, pub code: String, pub kind: &'static str, } pub async fn enqueue_publication(state: &AppState, doc: &DocRow, reason: &str) -> AppResult<()> { if doc.publish_path.is_none() { return Ok(()); } let now = util::now(); sqlx::query("INSERT INTO outbox (doc_id, reason, created_at, next_at) VALUES (?, ?, ?, ?)") .bind(&doc.id) .bind(reason) .bind(&now) .bind(&now) .execute(&state.db) .await?; state.publish_wake.notify_one(); Ok(()) } /// Seeds a minima that later reach their own threshold are marked confirmed. async fn confirm_seeds(state: &AppState, st: &DocState) -> AppResult<()> { for unit in &st.units { let Some(fid) = &unit.in_force_id else { continue }; let Some(p) = st.proposals.iter().find(|p| &p.id == fid) else { continue }; if p.entered_at.is_some() || p.confirmed_at.is_some() { continue; } if st.evaluation(unit, p).is_some_and(|e| e.reached) { sqlx::query("UPDATE proposals SET confirmed_at = ? WHERE id = ? AND confirmed_at IS NULL") .bind(util::now()) .bind(&p.id) .execute(&state.db) .await?; } } Ok(()) } pub async fn run(state: &AppState, doc_id: &str) -> AppResult> { let _pass = PASS.lock().await; let st = docs::load_state(state, doc_id).await?; confirm_seeds(state, &st).await?; let units = st.unit_states(); let winners = compose::adoptions(&units, &st.proposal_states(), st.electorate, st.doc.vote_floor) .map_err(|_| crate::error::AppError::Invalid("electorate"))?; let mut entered = Vec::new(); for id in winners { let Some(p) = st.proposals.iter().find(|p| p.id == id) else { continue }; let Some(unit) = st.units.iter().find(|u| u.id == p.unit_id) else { continue }; let us = units.iter().find(|u| u.id == unit.id); let tally = st.tally(&p.id); let eval = st.evaluation(unit, p); let params = us.map(|u| compose::params_for(u, p.kind(), st.doc.vote_floor)); let now = util::now(); let mut tx = state.db.begin().await?; // The text that enters is the text that was counted: a revision that landed // since the state was read stops it (the next pass will judge it again). let unchanged = sqlx::query( "UPDATE proposals SET updated_at = updated_at WHERE id = ? AND revision = ? AND title = ? AND body = ? AND status = ?", ) .bind(&p.id) .bind(p.revision) .bind(&p.title) .bind(&p.body) .bind(&p.status) .execute(&mut *tx) .await?; if unchanged.rows_affected() != 1 { continue; } let kind = match p.kind() { ProposalKind::Text | ProposalKind::Remove => { // The displaced version goes back among the proposals, with its votes. if let Some(old) = &unit.in_force_id { sqlx::query("UPDATE proposals SET status = 'open', left_at = ?, updated_at = ? WHERE id = ?") .bind(&now) .bind(&now) .bind(old) .execute(&mut *tx) .await?; } if p.kind() == ProposalKind::Text { sqlx::query("UPDATE proposals SET status = 'in_force', entered_at = ?, left_at = NULL, updated_at = ? WHERE id = ?") .bind(&now) .bind(&now) .bind(&p.id) .execute(&mut *tx) .await?; sqlx::query("UPDATE units SET in_force_id = ? WHERE id = ?") .bind(&p.id) .bind(&unit.id) .execute(&mut *tx) .await?; "text" } else { sqlx::query("UPDATE proposals SET status = 'applied', entered_at = ?, updated_at = ? WHERE id = ?") .bind(&now) .bind(&now) .bind(&p.id) .execute(&mut *tx) .await?; sqlx::query("UPDATE units SET in_force_id = NULL WHERE id = ?") .bind(&unit.id) .execute(&mut *tx) .await?; "remove" } } ProposalKind::Inertia(target) => { sqlx::query("UPDATE proposals SET status = 'applied', entered_at = ?, updated_at = ? WHERE id = ?") .bind(&now) .bind(&now) .bind(&p.id) .execute(&mut *tx) .await?; sqlx::query("UPDATE units SET inertia = ? WHERE id = ?") .bind(target.as_str()) .bind(&unit.id) .execute(&mut *tx) .await?; "inertia" } }; sqlx::query( "INSERT INTO adoptions (proposal_id, unit_id, doc_id, kind, votes_for, votes_against, electorate, required, params, at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", ) .bind(&p.id) .bind(&unit.id) .bind(&st.doc.id) .bind(kind) .bind(i64::from(tally.for_votes)) .bind(i64::from(tally.against)) .bind(i64::from(st.electorate)) .bind(eval.map_or(0, |e| i64::from(e.required))) .bind(serde_json::to_string(¶ms).unwrap_or_default()) .bind(&now) .execute(&mut *tx) .await?; tx.commit().await?; tracing::info!(doc = %st.doc.id, unit = %unit.code, proposal = %p.id, kind, "entered into force"); entered.push(Entered { proposal_id: p.id.clone(), unit_id: unit.id.clone(), code: unit.code.clone(), kind }); } if !entered.is_empty() { let codes: Vec<&str> = entered.iter().map(|e| e.code.as_str()).collect(); enqueue_publication(state, &st.doc, &format!("entrée en vigueur : {}", codes.join(", "))).await?; } Ok(entered) } /// The electorate changes (the web of trust grows or shrinks): every few /// minutes, every document is passed through the engine again. pub async fn sweeper(state: AppState) { let mut tick = tokio::time::interval(Duration::from_secs(300)); loop { tick.tick().await; let ids: Vec = match sqlx::query_scalar("SELECT id FROM documents").fetch_all(&state.db).await { Ok(ids) => ids, Err(e) => { tracing::warn!(error = %e, "sweeper"); continue; } }; for id in ids { if let Err(e) = run(&state, &id).await { tracing::warn!(doc = %id, error = %e, "sweeper pass"); } } } } /// Unit states, re-exported for tests. pub fn unit_states(st: &DocState) -> Vec { st.unit_states() }