orch/store/
jobs.rs

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

orch/store/
jobs.rs
