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

//! p̃ — the usual participation of each group, the anchor of the five levels
//! of inertia (ADR 0007, `ld_engine::inertia`).
//!
//! What is measured: the proposals of the group (every document, every kind,
//! modules included) that hold at least one vote on their current revision, the
//! last one cast within the past year — those whose status is named here, and
//! only those: `open` (under vote), `in_force` (in force, still under vote) and
//! `applied` (a removal or a change of inertia that passed: a decision taken).
//! Drafts (nobody votes on them) and proposals withdrawn by their author (their
//! votes can no longer be taken back) stay out, as would any status to come
//! until it is named. For each, T is the number of those
//! votes **at the time of measuring** — the same T the formula sees, since the
//! vote is permanent. The median of these T, and how many proposals it rests
//! on, are kept in `settings` under `participation:<group>`.
//!
//! When: once a day (at start, then by the sweeper), never at each vote — a
//! threshold that moved with every vote cast elsewhere could not be read. W
//! moves on its own (the web of trust, every six hours): p̃ = T̃ / W follows it
//! at once, so the scale still asks s_k of the same usual number of voters.

use crate::AppState;
use crate::util;
use serde::{Deserialize, Serialize};
use time::OffsetDateTime;
use time::format_description::well_known::Rfc3339;

/// Only proposals voted on within this many days count.
pub const WINDOW_DAYS: i64 = 365;
/// A measure older than this is taken again.
pub const EVERY_HOURS: i64 = 24;

const KEY: &str = "participation:";

/// What was measured for a group.
#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct Measure {
    /// The median number of voters, when at least one proposal was voted on.
    pub median: Option<f64>,
    /// Proposals it rests on.
    pub sample: u32,
    pub measured_at: String,
}

/// The numbers of voters of the group's proposals voted on since `since`, for
/// the statuses counted only (the list is explicit: see the head of this file).
pub async fn sample(state: &AppState, group: &str, since: &str) -> sqlx::Result<Vec<u32>> {
    let rows: Vec<i64> = sqlx::query_scalar(
        "SELECT COUNT(*) FROM votes v
           JOIN proposals p ON p.id = v.proposal_id
           JOIN units u ON u.id = p.unit_id
           JOIN documents d ON d.id = u.doc_id
          WHERE d.group_slug = ? AND p.status IN ('open', 'in_force', 'applied') AND v.revision = p.revision
            AND (v.source <> 'simulated' OR ?)
          GROUP BY v.proposal_id
         HAVING MAX(v.cast_at) >= ?",
    )
    .bind(group)
    // The simulated votes weigh here as they weigh in adoption (ADR 0012).
    .bind(state.vote_rules().adopts(ld_engine::buffer::Stage::Simulated))
    .bind(since)
    .fetch_all(&state.db)
    .await?;
    Ok(rows.into_iter().map(|n| u32::try_from(n).unwrap_or(u32::MAX)).collect())
}

/// Measures the group now.
pub async fn measure(state: &AppState, group: &str) -> sqlx::Result<Measure> {
    let now = OffsetDateTime::now_utc();
    let since = util::stamp(now - time::Duration::days(WINDOW_DAYS));
    let votes = sample(state, group, &since).await?;
    Ok(Measure {
        median: ld_engine::inertia::median(&votes),
        sample: u32::try_from(votes.len()).unwrap_or(u32::MAX),
        measured_at: util::stamp(now),
    })
}

/// The last measure kept for the group, if any.
pub async fn stored(state: &AppState, group: &str) -> sqlx::Result<Option<Measure>> {
    let raw: Option<String> = sqlx::query_scalar("SELECT value FROM settings WHERE key = ?")
        .bind(format!("{KEY}{group}"))
        .fetch_optional(&state.db)
        .await?;
    Ok(raw.and_then(|v| serde_json::from_str(&v).ok()))
}

async fn keep(state: &AppState, group: &str, m: &Measure) -> anyhow::Result<()> {
    sqlx::query(
        "INSERT INTO settings (key, value, updated_at) VALUES (?, ?, ?)
         ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at",
    )
    .bind(format!("{KEY}{group}"))
    .bind(serde_json::to_string(m)?)
    .bind(&m.measured_at)
    .execute(&state.db)
    .await?;
    Ok(())
}

fn is_stale(m: &Measure, now: OffsetDateTime) -> bool {
    OffsetDateTime::parse(&m.measured_at, &Rfc3339)
        .map_or(true, |at| now - at >= time::Duration::hours(EVERY_HOURS) || at > now)
}

/// Measures again every group whose measure is missing or a day old (all of
/// them with `force`). Returns the groups measured.
pub async fn refresh(state: &AppState, force: bool) -> anyhow::Result<Vec<String>> {
    let groups: Vec<String> =
        sqlx::query_scalar("SELECT slug FROM groups ORDER BY position").fetch_all(&state.db).await?;
    let now = OffsetDateTime::now_utc();
    let mut done = Vec::new();
    for group in groups {
        let due = force || stored(state, &group).await?.is_none_or(|m| is_stale(&m, now));
        if !due {
            continue;
        }
        let m = measure(state, &group).await?;
        keep(state, &group, &m).await?;
        tracing::info!(%group, median = ?m.median, sample = m.sample, "usual participation measured");
        done.push(group);
    }
    Ok(done)
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn a_measure_is_taken_again_after_a_day() {
        let now = OffsetDateTime::now_utc();
        let at =
            |hours: i64| Measure { measured_at: util::stamp(now - time::Duration::hours(hours)), ..Measure::default() };
        assert!(!is_stale(&at(1), now));
        assert!(!is_stale(&at(23), now));
        assert!(is_stale(&at(24), now));
        assert!(is_stale(&at(-2), now), "a measure from the future is taken again");
        assert!(is_stale(&Measure::default(), now), "unreadable");
    }
}