hatchery_server/
tasks.rs

1//! P6: minimal task manager. Tasks live in SQLite under the server data dir. An operator
2//! creates and assigns them; a node sees (and progresses) only the tasks assigned to it, and a
3//! task can be assigned only to a node whose ledger entry holds the `tasks` grant (trusted by
4//! default). The hub enforces that; this module is storage and state rules.
5
6use crate::error::ServerError;
7use rusqlite::{params, Connection};
8use serde::Serialize;
9use std::path::Path;
10
11#[derive(Clone, Debug, Serialize, Eq, PartialEq)]
12pub struct Task {
13    pub id: i64,
14    pub title: String,
15    pub body: String,
16    pub assignee: Option<String>,
17    /// open | assigned | in_progress | done | cancelled
18    pub status: String,
19    pub created_at_ms: u64,
20    pub updated_at_ms: u64,
21}
22
23pub struct TaskStore {
24    conn: Connection,
25}
26
27fn unix_ms() -> u64 {
28    std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_millis() as u64
29}
30
31fn row(r: &rusqlite::Row<'_>) -> rusqlite::Result<Task> {
32    Ok(Task {
33        id: r.get(0)?,
34        title: r.get(1)?,
35        body: r.get(2)?,
36        assignee: r.get(3)?,
37        status: r.get(4)?,
38        created_at_ms: r.get::<_, i64>(5)? as u64,
39        updated_at_ms: r.get::<_, i64>(6)? as u64,
40    })
41}
42
43fn db(e: rusqlite::Error) -> ServerError {
44    ServerError::sub("tasks", e.to_string())
45}
46
47impl TaskStore {
48    pub fn open(data_dir: &Path) -> Result<Self, ServerError> {
49        let dir = data_dir.join("tasks");
50        std::fs::create_dir_all(&dir)?;
51        Self::from_conn(Connection::open(dir.join("tasks.db")).map_err(db)?)
52    }
53
54    pub fn in_memory() -> Result<Self, ServerError> {
55        Self::from_conn(Connection::open_in_memory().map_err(db)?)
56    }
57
58    fn from_conn(conn: Connection) -> Result<Self, ServerError> {
59        conn.execute_batch(
60            "PRAGMA journal_mode=WAL;
61             CREATE TABLE IF NOT EXISTS tasks (
62               id INTEGER PRIMARY KEY AUTOINCREMENT,
63               title TEXT NOT NULL, body TEXT NOT NULL, assignee TEXT,
64               status TEXT NOT NULL, created_at_ms INTEGER NOT NULL, updated_at_ms INTEGER NOT NULL);",
65        )
66        .map_err(db)?;
67        Ok(Self { conn })
68    }
69
70    pub fn create(&self, title: &str, body: &str, assignee: Option<&str>) -> Result<Task, ServerError> {
71        let now = unix_ms() as i64;
72        let status = if assignee.is_some() { "assigned" } else { "open" };
73        self.conn
74            .execute("INSERT INTO tasks (title, body, assignee, status, created_at_ms, updated_at_ms) VALUES (?1,?2,?3,?4,?5,?5)", params![title, body, assignee, status, now])
75            .map_err(db)?;
76        self.get(self.conn.last_insert_rowid())?.ok_or_else(|| ServerError::sub("tasks", "lost row"))
77    }
78
79    pub fn get(&self, id: i64) -> Result<Option<Task>, ServerError> {
80        let mut stmt = self.conn.prepare("SELECT id,title,body,assignee,status,created_at_ms,updated_at_ms FROM tasks WHERE id=?1").map_err(db)?;
81        let mut rows = stmt.query_map(params![id], row).map_err(db)?;
82        rows.next().transpose().map_err(db)
83    }
84
85    pub fn list(&self, assignee: Option<&str>, open_only: bool) -> Result<Vec<Task>, ServerError> {
86        let mut stmt = self
87            .conn
88            .prepare("SELECT id,title,body,assignee,status,created_at_ms,updated_at_ms FROM tasks WHERE (?1 IS NULL OR assignee=?1) AND (?2=0 OR status IN ('open','assigned','in_progress')) ORDER BY id")
89            .map_err(db)?;
90        let rows = stmt.query_map(params![assignee, open_only as i64], row).map_err(db)?;
91        rows.collect::<Result<_, _>>().map_err(db)
92    }
93
94    pub fn assign(&self, id: i64, assignee: &str) -> Result<bool, ServerError> {
95        let n = self
96            .conn
97            .execute("UPDATE tasks SET assignee=?2, status='assigned', updated_at_ms=?3 WHERE id=?1 AND status IN ('open','assigned')", params![id, assignee, unix_ms() as i64])
98            .map_err(db)?;
99        Ok(n == 1)
100    }
101
102    /// Operator cancel, or the assignee moving its own task along (`in_progress`, `done`).
103    pub fn set_status(&self, id: i64, status: &str, only_assignee: Option<&str>) -> Result<bool, ServerError> {
104        let n = self
105            .conn
106            .execute(
107                "UPDATE tasks SET status=?2, updated_at_ms=?3 WHERE id=?1 AND status NOT IN ('done','cancelled') AND (?4 IS NULL OR assignee=?4)",
108                params![id, status, unix_ms() as i64, only_assignee],
109            )
110            .map_err(db)?;
111        Ok(n == 1)
112    }
113}
114
115#[cfg(test)]
116mod tests {
117    use super::*;
118
119    #[test]
120    fn lifecycle_and_ownership() {
121        let s = TaskStore::in_memory().unwrap();
122        let t = s.create("build", "run it", None).unwrap();
123        assert_eq!(t.status, "open");
124        assert!(s.assign(t.id, "alpha").unwrap());
125        assert_eq!(s.list(Some("alpha"), true).unwrap().len(), 1);
126        assert!(s.list(Some("beta"), true).unwrap().is_empty());
127        // Only the assignee moves its own task.
128        assert!(!s.set_status(t.id, "done", Some("beta")).unwrap());
129        assert!(s.set_status(t.id, "in_progress", Some("alpha")).unwrap());
130        assert!(s.set_status(t.id, "done", Some("alpha")).unwrap());
131        // Finished is final; no reassign.
132        assert!(!s.set_status(t.id, "open", None).unwrap());
133        assert!(!s.assign(t.id, "beta").unwrap());
134        assert!(s.list(Some("alpha"), true).unwrap().is_empty());
135        assert_eq!(s.list(None, false).unwrap().len(), 1);
136    }
137
138    #[test]
139    fn survives_reopen() {
140        let dir = tempfile::tempdir().unwrap();
141        TaskStore::open(dir.path()).unwrap().create("a", "b", Some("alpha")).unwrap();
142        assert_eq!(TaskStore::open(dir.path()).unwrap().list(Some("alpha"), true).unwrap().len(), 1);
143    }
144}
hatchery_server/tasks.rs

hatchery_server/
tasks.rs
