headless_lms_models/
credit_registration_phase_state.rs

1//! Per-phase heartbeat and control for the credit registration pipeline.
2//!
3//! One row per phase, seeded by migration and thereafter only ever updated.
4use utoipa::ToSchema;
5
6use crate::prelude::*;
7
8/// Canonical phase names, used verbatim as the `phase` value, in the test tick endpoint, the
9/// dashboard and the audit log.
10pub const PHASES: &[&str] = &[
11    "materialize",
12    "preconditions",
13    "resolve-enrolments",
14    "import",
15    "verify",
16    "legacy-mirror",
17    "student-notifications",
18    "enrolment-discovery",
19    "link-emails",
20    "product-token-refresh",
21    "config-validation",
22    "retention-sweep",
23];
24
25#[derive(Debug, Serialize, Deserialize, PartialEq, Clone, ToSchema)]
26pub struct CreditRegistrationPhaseState {
27    pub id: Uuid,
28    pub created_at: DateTime<Utc>,
29    pub updated_at: DateTime<Utc>,
30    pub deleted_at: Option<DateTime<Utc>>,
31    pub phase: String,
32    pub process_name: String,
33    pub expected_interval_secs: i32,
34    pub last_heartbeat_at: Option<DateTime<Utc>>,
35    pub last_run_started_at: Option<DateTime<Utc>>,
36    pub last_run_finished_at: Option<DateTime<Utc>>,
37    pub last_success_at: Option<DateTime<Utc>>,
38    pub next_run_at: Option<DateTime<Utc>>,
39    pub items_processed_last_run: Option<i32>,
40    pub items_failed_last_run: Option<i32>,
41    pub consecutive_failures: i32,
42    pub last_error: Option<String>,
43    pub paused_at: Option<DateTime<Utc>>,
44    pub paused_by_user_id: Option<Uuid>,
45    pub pause_reason: Option<String>,
46}
47
48#[derive(Debug, Clone, PartialEq, Default)]
49pub struct PhaseRunOutcome {
50    pub items_processed: i32,
51    pub items_failed: i32,
52    /// `None` on success. Scrub before passing.
53    pub error: Option<String>,
54}
55
56pub async fn get_all(conn: &mut PgConnection) -> ModelResult<Vec<CreditRegistrationPhaseState>> {
57    let res = sqlx::query_as!(
58        CreditRegistrationPhaseState,
59        r#"
60SELECT *
61FROM credit_registration_phase_state
62WHERE deleted_at IS NULL
63ORDER BY process_name,
64  phase
65        "#,
66    )
67    .fetch_all(conn)
68    .await?;
69    Ok(res)
70}
71
72pub async fn get_by_phase(
73    conn: &mut PgConnection,
74    phase: &str,
75) -> ModelResult<CreditRegistrationPhaseState> {
76    let res = sqlx::query_as!(
77        CreditRegistrationPhaseState,
78        r#"
79SELECT *
80FROM credit_registration_phase_state
81WHERE phase = $1
82  AND deleted_at IS NULL
83        "#,
84        phase
85    )
86    .fetch_one(conn)
87    .await?;
88    Ok(res)
89}
90
91/// Written every iteration, work or not, so idle and wedged stay distinguishable.
92pub async fn heartbeat(conn: &mut PgConnection, phase: &str) -> ModelResult<()> {
93    sqlx::query!(
94        r#"
95UPDATE credit_registration_phase_state
96SET last_heartbeat_at = now(),
97  last_run_started_at = now()
98WHERE phase = $1
99  AND deleted_at IS NULL
100        "#,
101        phase
102    )
103    .execute(conn)
104    .await?;
105    Ok(())
106}
107
108/// Closes out an iteration. Only a success moves `last_success_at`, so a wedged phase stays
109/// distinguishable from a quiet one.
110pub async fn record_run(
111    conn: &mut PgConnection,
112    phase: &str,
113    outcome: &PhaseRunOutcome,
114) -> ModelResult<()> {
115    sqlx::query!(
116        r#"
117UPDATE credit_registration_phase_state
118SET last_run_finished_at = now(),
119  items_processed_last_run = $2,
120  items_failed_last_run = $3,
121  last_success_at = CASE
122    WHEN $4::text IS NULL THEN now()
123    ELSE last_success_at
124  END,
125  consecutive_failures = CASE
126    WHEN $4::text IS NULL THEN 0
127    ELSE consecutive_failures + 1
128  END,
129  last_error = $4
130WHERE phase = $1
131  AND deleted_at IS NULL
132        "#,
133        phase,
134        outcome.items_processed,
135        outcome.items_failed,
136        outcome.error,
137    )
138    .execute(conn)
139    .await?;
140    Ok(())
141}
142
143pub async fn is_paused(conn: &mut PgConnection, phase: &str) -> ModelResult<bool> {
144    let paused = sqlx::query_scalar!(
145        r#"
146SELECT paused_at IS NOT NULL AS "paused!"
147FROM credit_registration_phase_state
148WHERE phase = $1
149  AND deleted_at IS NULL
150        "#,
151        phase
152    )
153    .fetch_one(conn)
154    .await?;
155    Ok(paused)
156}
157
158pub async fn pause(
159    conn: &mut PgConnection,
160    phase: &str,
161    paused_by_user_id: Uuid,
162    pause_reason: Option<&str>,
163) -> ModelResult<()> {
164    sqlx::query!(
165        r#"
166UPDATE credit_registration_phase_state
167SET paused_at = now(),
168  paused_by_user_id = $2,
169  pause_reason = $3
170WHERE phase = $1
171  AND deleted_at IS NULL
172        "#,
173        phase,
174        paused_by_user_id,
175        pause_reason,
176    )
177    .execute(conn)
178    .await?;
179    Ok(())
180}
181
182pub async fn resume(conn: &mut PgConnection, phase: &str) -> ModelResult<()> {
183    sqlx::query!(
184        r#"
185UPDATE credit_registration_phase_state
186SET paused_at = NULL,
187  paused_by_user_id = NULL,
188  pause_reason = NULL
189WHERE phase = $1
190  AND deleted_at IS NULL
191        "#,
192        phase
193    )
194    .execute(conn)
195    .await?;
196    Ok(())
197}
198
199/// Makes the phase due now; the phase loop picks it up on its next `next_run_at` check.
200pub async fn run_now(conn: &mut PgConnection, phase: &str) -> ModelResult<()> {
201    sqlx::query!(
202        r#"
203UPDATE credit_registration_phase_state
204SET next_run_at = now()
205WHERE phase = $1
206  AND deleted_at IS NULL
207        "#,
208        phase
209    )
210    .execute(conn)
211    .await?;
212    Ok(())
213}
214
215pub async fn set_next_run_at(
216    conn: &mut PgConnection,
217    phase: &str,
218    next_run_at: DateTime<Utc>,
219) -> ModelResult<()> {
220    sqlx::query!(
221        r#"
222UPDATE credit_registration_phase_state
223SET next_run_at = $2
224WHERE phase = $1
225  AND deleted_at IS NULL
226        "#,
227        phase,
228        next_run_at,
229    )
230    .execute(conn)
231    .await?;
232    Ok(())
233}
headless_lms_models/credit_registration_phase_state.rs

headless_lms_models/
credit_registration_phase_state.rs
