orch/store/
jobs.rs

1use super::*;
2use serde::{Deserialize, Serialize};
3
4/// Explicit column list for `SELECT` queries on the `job_state` table.
5///
6/// Using `SELECT *` with `sqlx-sqlite`'s prepared statement cache can cause OOB
7/// panics when the schema changes (new migration columns) because the cached
8/// column metadata may not match the current table structure. Explicit columns
9/// prevent this mismatch.
10const JOB_STATE_COLS: &str = "repo, job_id, last_run, last_task_status, active_task_id";
11
12/// Runtime state for a scheduled job, stored in SQLite (not in .orch.yml).
13#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow)]
14pub struct JobState {
15    pub repo: String,
16    pub job_id: String,
17    pub last_run: Option<String>,
18    pub last_task_status: Option<String>,
19    pub active_task_id: Option<String>,
20}
21
22impl TaskStore {
23    pub async fn get_job_state(
24        &self,
25        repo: &str,
26        job_id: &str,
27    ) -> anyhow::Result<Option<JobState>> {
28        let sql = format!(
29            "SELECT {} FROM job_state WHERE repo = ? AND job_id = ?",
30            JOB_STATE_COLS
31        );
32        let row: Option<JobState> = sqlx::query_as(&sql)
33            .bind(repo)
34            .bind(job_id)
35            .fetch_optional(&self.pool)
36            .await?;
37        Ok(row)
38    }
39
40    /// Upsert runtime state for a job.
41    pub async fn upsert_job_state(&self, state: &JobState) -> anyhow::Result<()> {
42        sqlx::query(
43            "INSERT INTO job_state (repo, job_id, last_run, last_task_status, active_task_id)
44         VALUES (?, ?, ?, ?, ?)
45         ON CONFLICT(repo, job_id) DO UPDATE SET
46           last_run = excluded.last_run,
47           last_task_status = excluded.last_task_status,
48           active_task_id = excluded.active_task_id",
49        )
50        .bind(&state.repo)
51        .bind(&state.job_id)
52        .bind(&state.last_run)
53        .bind(&state.last_task_status)
54        .bind(&state.active_task_id)
55        .execute(&self.pool)
56        .await?;
57        Ok(())
58    }
59
60    /// List all job states for a repo.
61    pub async fn list_job_states(&self, repo: &str) -> anyhow::Result<Vec<JobState>> {
62        let sql = format!("SELECT {} FROM job_state WHERE repo = ?", JOB_STATE_COLS);
63        let rows: Vec<JobState> = sqlx::query_as(&sql)
64            .bind(repo)
65            .fetch_all(&self.pool)
66            .await?;
67        Ok(rows)
68    }
69
70    // ---------------------------------------------------------------
71    // Channel Subscriptions
72    // ---------------------------------------------------------------
73
74    /// Subscribe a channel/thread to notifications for a project.
75    pub async fn subscribe_channel(
76        &self,
77        channel: &str,
78        thread_id: &str,
79        repo: &str,
80        topic_id: Option<&str>,
81    ) -> anyhow::Result<()> {
82        let topic = topic_id.unwrap_or("");
83        sqlx::query(
84        "INSERT OR IGNORE INTO channel_subscriptions (channel, thread_id, repo, topic_id) VALUES (?, ?, ?, ?)",
85    )
86        .bind(channel)
87        .bind(thread_id)
88        .bind(repo)
89        .bind(topic)
90        .execute(&self.pool)
91        .await?;
92        Ok(())
93    }
94
95    /// Unsubscribe a channel/thread from a project's notifications.
96    pub async fn unsubscribe_channel(
97        &self,
98        channel: &str,
99        thread_id: &str,
100        repo: &str,
101    ) -> anyhow::Result<()> {
102        sqlx::query(
103            "DELETE FROM channel_subscriptions WHERE channel = ? AND thread_id = ? AND repo = ?",
104        )
105        .bind(channel)
106        .bind(thread_id)
107        .bind(repo)
108        .execute(&self.pool)
109        .await?;
110        Ok(())
111    }
112
113    /// List repos that a channel/thread is subscribed to.
114    #[cfg(test)]
115    pub async fn list_channel_subscriptions(
116        &self,
117        channel: &str,
118        thread_id: &str,
119    ) -> anyhow::Result<Vec<String>> {
120        let rows: Vec<(String,)> = sqlx::query_as(
121            "SELECT repo FROM channel_subscriptions WHERE channel = ? AND thread_id = ?",
122        )
123        .bind(channel)
124        .bind(thread_id)
125        .fetch_all(&self.pool)
126        .await?;
127        Ok(rows.into_iter().map(|r| r.0).collect())
128    }
129
130    /// List all subscriptions (channel, thread_id, topic_id) for a repo.
131    pub async fn list_subscribers_for_repo(
132        &self,
133        repo: &str,
134    ) -> anyhow::Result<Vec<(String, String, String)>> {
135        let rows: Vec<(String, String, String)> = sqlx::query_as(
136            "SELECT channel, thread_id, topic_id FROM channel_subscriptions WHERE repo = ?",
137        )
138        .bind(repo)
139        .fetch_all(&self.pool)
140        .await?;
141        Ok(rows)
142    }
143}
orch/store/jobs.rs

orch/store/
jobs.rs
