pushkin-core 0.2.1

//! Peer board (spec §9): persisted, restart-surviving coordination in
//! `SQLite`. Every operation is scoped by a run ID so concurrent runs never
//! intersect. Verbs: register / status / peers / broadcast / send / read
//! (auto-cursor per recipient), plus claim / release on paths — the
//! pre-write gate consults claims to deny cross-agent edits. Embedded
//! `SQLite` only (spec §10); schema is a versioned, append-friendly step.

use globset::Glob;
use rusqlite::{params, Connection};
use std::path::Path;
use thiserror::Error;

#[derive(Debug, Error)]
pub enum BoardError {
    #[error("board storage error: {0}")]
    Storage(#[from] rusqlite::Error),
    #[error("invalid claim glob '{glob}': {message}")]
    BadGlob { glob: String, message: String },
}

/// One agent's registration row.
#[derive(Debug)]
pub struct Peer {
    pub agent: String,
    pub status: Option<String>,
}

/// One message delivered to a reader.
#[derive(Debug)]
pub struct Message {
    pub id: i64,
    pub from_agent: String,
    pub body: String,
}

/// An active path claim.
#[derive(Debug)]
pub struct Claim {
    pub agent: String,
    pub path_glob: String,
}

pub struct Board {
    conn: Connection,
    run_id: String,
}

impl Board {
    /// Opens (creating as needed) the board for one run.
    ///
    /// # Errors
    /// Returns `BoardError` when the database cannot be opened or migrated.
    pub fn open(path: &Path, run_id: &str) -> Result<Self, BoardError> {
        let conn = Connection::open(path)?;
        conn.pragma_update(None, "journal_mode", "WAL")?;
        // Versioned, append-only migration steps (AGENTS.md SQLite rule).
        // Timestamps: epoch ms UTC, one convention for every board table.
        conn.execute_batch(
            "CREATE TABLE IF NOT EXISTS board_schema (version INTEGER NOT NULL);
             CREATE TABLE IF NOT EXISTS agents (
               run_id TEXT NOT NULL,
               agent  TEXT NOT NULL,
               status TEXT,
               registered_at_ms INTEGER NOT NULL,
               PRIMARY KEY (run_id, agent)
             );
             CREATE TABLE IF NOT EXISTS messages (
               id INTEGER PRIMARY KEY AUTOINCREMENT,
               run_id TEXT NOT NULL,
               from_agent TEXT NOT NULL,
               to_agent TEXT,
               body TEXT NOT NULL,
               sent_at_ms INTEGER NOT NULL
             );
             CREATE TABLE IF NOT EXISTS cursors (
               run_id TEXT NOT NULL,
               agent  TEXT NOT NULL,
               last_read_id INTEGER NOT NULL,
               PRIMARY KEY (run_id, agent)
             );
             CREATE TABLE IF NOT EXISTS claims (
               run_id TEXT NOT NULL,
               agent  TEXT NOT NULL,
               path_glob TEXT NOT NULL,
               claimed_at_ms INTEGER NOT NULL,
               PRIMARY KEY (run_id, path_glob)
             );",
        )?;
        let versions: i64 =
            conn.query_row("SELECT COUNT(*) FROM board_schema", [], |row| row.get(0))?;
        if versions == 0 {
            conn.execute("INSERT INTO board_schema (version) VALUES (1)", [])?;
        }
        Ok(Self {
            conn,
            run_id: run_id.to_owned(),
        })
    }

    /// Registers (or re-registers, idempotently) an agent in this run.
    ///
    /// # Errors
    /// Returns `BoardError` on storage failure.
    pub fn register(&self, agent: &str) -> Result<(), BoardError> {
        self.conn.execute(
            "INSERT INTO agents (run_id, agent, status, registered_at_ms)
             VALUES (?1, ?2, NULL, ?3)
             ON CONFLICT (run_id, agent) DO NOTHING",
            params![self.run_id, agent, now_ms()],
        )?;
        Ok(())
    }

    /// Updates an agent's free-text status line.
    ///
    /// # Errors
    /// Returns `BoardError` on storage failure.
    pub fn set_status(&self, agent: &str, status: &str) -> Result<(), BoardError> {
        self.register(agent)?;
        self.conn.execute(
            "UPDATE agents SET status = ?3 WHERE run_id = ?1 AND agent = ?2",
            params![self.run_id, agent, status],
        )?;
        Ok(())
    }

    /// All agents in this run except `agent` itself.
    ///
    /// # Errors
    /// Returns `BoardError` on storage failure.
    pub fn peers(&self, agent: &str) -> Result<Vec<Peer>, BoardError> {
        let mut statement = self.conn.prepare(
            "SELECT agent, status FROM agents
             WHERE run_id = ?1 AND agent != ?2 ORDER BY agent",
        )?;
        let peers = statement
            .query_map(params![self.run_id, agent], |row| {
                Ok(Peer {
                    agent: row.get(0)?,
                    status: row.get(1)?,
                })
            })?
            .collect::<Result<Vec<_>, _>>()?;
        Ok(peers)
    }
}

impl Board {
    /// Queues a message to every current peer (`to_agent` NULL = broadcast).
    ///
    /// # Errors
    /// Returns `BoardError` on storage failure.
    pub fn broadcast(&self, from_agent: &str, body: &str) -> Result<(), BoardError> {
        self.conn.execute(
            "INSERT INTO messages (run_id, from_agent, to_agent, body, sent_at_ms)
             VALUES (?1, ?2, NULL, ?3, ?4)",
            params![self.run_id, from_agent, body, now_ms()],
        )?;
        Ok(())
    }

    /// Queues a directed message.
    ///
    /// # Errors
    /// Returns `BoardError` on storage failure.
    pub fn send(&self, from_agent: &str, to_agent: &str, body: &str) -> Result<(), BoardError> {
        self.conn.execute(
            "INSERT INTO messages (run_id, from_agent, to_agent, body, sent_at_ms)
             VALUES (?1, ?2, ?3, ?4, ?5)",
            params![self.run_id, from_agent, to_agent, body, now_ms()],
        )?;
        Ok(())
    }

    /// Unread messages for `agent` (broadcasts + directed), advancing the
    /// auto-cursor so a second read never redelivers.
    ///
    /// # Errors
    /// Returns `BoardError` on storage failure.
    pub fn read(&self, agent: &str) -> Result<Vec<Message>, BoardError> {
        let cursor: i64 = self
            .conn
            .query_row(
                "SELECT last_read_id FROM cursors WHERE run_id = ?1 AND agent = ?2",
                params![self.run_id, agent],
                |row| row.get(0),
            )
            .unwrap_or(0);
        let mut statement = self.conn.prepare(
            "SELECT id, from_agent, body FROM messages
             WHERE run_id = ?1 AND id > ?2
               AND from_agent != ?3
               AND (to_agent IS NULL OR to_agent = ?3)
             ORDER BY id",
        )?;
        let messages = statement
            .query_map(params![self.run_id, cursor, agent], |row| {
                Ok(Message {
                    id: row.get(0)?,
                    from_agent: row.get(1)?,
                    body: row.get(2)?,
                })
            })?
            .collect::<Result<Vec<_>, _>>()?;
        if let Some(last) = messages.last() {
            self.conn.execute(
                "INSERT INTO cursors (run_id, agent, last_read_id) VALUES (?1, ?2, ?3)
                 ON CONFLICT (run_id, agent) DO UPDATE SET last_read_id = ?3",
                params![self.run_id, agent, last.id],
            )?;
        }
        Ok(messages)
    }

    /// Claims a path glob for `agent`. Idempotent for the holder; a claim
    /// held by another agent in this run is a loud conflict.
    ///
    /// # Errors
    /// Returns `BoardError::BadGlob` for an invalid glob, `Storage` for
    /// conflicts (constraint violation) and other database failures.
    pub fn claim(&self, agent: &str, path_glob: &str) -> Result<(), BoardError> {
        Glob::new(path_glob).map_err(|error| BoardError::BadGlob {
            glob: path_glob.to_owned(),
            message: error.to_string(),
        })?;
        self.register(agent)?;
        let changed = self.conn.execute(
            "INSERT INTO claims (run_id, agent, path_glob, claimed_at_ms)
             VALUES (?1, ?2, ?3, ?4)
             ON CONFLICT (run_id, path_glob) DO NOTHING",
            params![self.run_id, agent, path_glob, now_ms()],
        )?;
        if changed == 0 {
            let holder: String = self.conn.query_row(
                "SELECT agent FROM claims WHERE run_id = ?1 AND path_glob = ?2",
                params![self.run_id, path_glob],
                |row| row.get(0),
            )?;
            if holder != agent {
                return Err(BoardError::Storage(rusqlite::Error::SqliteFailure(
                    rusqlite::ffi::Error::new(rusqlite::ffi::SQLITE_CONSTRAINT),
                    Some(format!("path already claimed by {holder}")),
                )));
            }
        }
        Ok(())
    }

    /// Releases a claim; only the holder can release.
    ///
    /// # Errors
    /// Returns `BoardError` on storage failure.
    pub fn release(&self, agent: &str, path_glob: &str) -> Result<bool, BoardError> {
        let changed = self.conn.execute(
            "DELETE FROM claims WHERE run_id = ?1 AND agent = ?2 AND path_glob = ?3",
            params![self.run_id, agent, path_glob],
        )?;
        Ok(changed > 0)
    }

    /// Active claims in this run.
    ///
    /// # Errors
    /// Returns `BoardError` on storage failure.
    pub fn claims(&self) -> Result<Vec<Claim>, BoardError> {
        let mut statement = self
            .conn
            .prepare("SELECT agent, path_glob FROM claims WHERE run_id = ?1 ORDER BY path_glob")?;
        let claims = statement
            .query_map(params![self.run_id], |row| {
                Ok(Claim {
                    agent: row.get(0)?,
                    path_glob: row.get(1)?,
                })
            })?
            .collect::<Result<Vec<_>, _>>()?;
        Ok(claims)
    }

    /// The holder of a claim covering `file`, when that holder is not
    /// `agent` — i.e. the gate-side question "is this write blocked?".
    ///
    /// # Errors
    /// Returns `BoardError` on storage failure.
    pub fn blocking_holder(&self, agent: &str, file: &str) -> Result<Option<String>, BoardError> {
        for claim in self.claims()? {
            if claim.agent != agent && glob_matches(&claim.path_glob, file) {
                return Ok(Some(claim.agent));
            }
        }
        Ok(None)
    }
}

/// Whether `glob` matches `file`; an invalid glob matches nothing.
/// Public so CLI-side policy layers share one matching semantics with
/// the board and waivers instead of growing a second glob dependency.
#[must_use]
pub fn glob_matches(glob: &str, file: &str) -> bool {
    Glob::new(glob).is_ok_and(|g| g.compile_matcher().is_match(file))
}

fn now_ms() -> i64 {
    i64::try_from(
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map_or(0, |duration| duration.as_millis()),
    )
    .unwrap_or(i64::MAX)
}