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    "config-validation",
21    "retention-sweep",
22    "ledger-snapshot",
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/// What one iteration of a phase did, as its phase-state row records it.
49#[derive(Debug, Clone, PartialEq, Default)]
50pub struct PhaseRunOutcome {
51    pub items_processed: i32,
52    /// Processed rows that only await something outside the pipeline; not in `items_failed`, and
53    /// not stored.
54    pub items_waiting: i32,
55    pub items_failed: i32,
56    /// `None` on success. Scrub before passing.
57    pub error: Option<String>,
58}
59
60pub async fn get_all(conn: &mut PgConnection) -> ModelResult<Vec<CreditRegistrationPhaseState>> {
61    let res = sqlx::query_as!(
62        CreditRegistrationPhaseState,
63        r#"
64SELECT *
65FROM credit_registration_phase_state
66WHERE deleted_at IS NULL
67ORDER BY process_name,
68  phase
69        "#,
70    )
71    .fetch_all(conn)
72    .await?;
73    Ok(res)
74}
75
76pub async fn get_by_phase(
77    conn: &mut PgConnection,
78    phase: &str,
79) -> ModelResult<CreditRegistrationPhaseState> {
80    let res = sqlx::query_as!(
81        CreditRegistrationPhaseState,
82        r#"
83SELECT *
84FROM credit_registration_phase_state
85WHERE phase = $1
86  AND deleted_at IS NULL
87        "#,
88        phase
89    )
90    .fetch_one(conn)
91    .await?;
92    Ok(res)
93}
94
95/// Written every iteration, work or not, so idle and wedged stay distinguishable.
96pub async fn heartbeat(conn: &mut PgConnection, phase: &str) -> ModelResult<()> {
97    sqlx::query!(
98        r#"
99UPDATE credit_registration_phase_state
100SET last_heartbeat_at = now(),
101  last_run_started_at = now()
102WHERE phase = $1
103  AND deleted_at IS NULL
104        "#,
105        phase
106    )
107    .execute(conn)
108    .await?;
109    Ok(())
110}
111
112/// Keeps a long iteration from reading as a dead worker; unlike [`heartbeat`], leaves
113/// `last_run_started_at` at the iteration's start.
114pub async fn keep_alive(conn: &mut PgConnection, phase: &str) -> ModelResult<()> {
115    sqlx::query!(
116        r#"
117UPDATE credit_registration_phase_state
118SET last_heartbeat_at = now()
119WHERE phase = $1
120  AND deleted_at IS NULL
121        "#,
122        phase
123    )
124    .execute(conn)
125    .await?;
126    Ok(())
127}
128
129/// Closes out an iteration. Only a success moves `last_success_at`, so a wedged phase stays
130/// distinguishable from a quiet one.
131pub async fn record_run(
132    conn: &mut PgConnection,
133    phase: &str,
134    outcome: &PhaseRunOutcome,
135) -> ModelResult<()> {
136    sqlx::query!(
137        r#"
138UPDATE credit_registration_phase_state
139SET last_run_finished_at = now(),
140  items_processed_last_run = $2,
141  items_failed_last_run = $3,
142  last_success_at = CASE
143    WHEN $4::text IS NULL THEN now()
144    ELSE last_success_at
145  END,
146  consecutive_failures = CASE
147    WHEN $4::text IS NULL THEN 0
148    ELSE consecutive_failures + 1
149  END,
150  last_error = $4
151WHERE phase = $1
152  AND deleted_at IS NULL
153        "#,
154        phase,
155        outcome.items_processed,
156        outcome.items_failed,
157        outcome.error,
158    )
159    .execute(conn)
160    .await?;
161    Ok(())
162}
163
164/// Closes out an iteration that sent nothing on purpose, counting it as productive so a held-back
165/// queue does not read as wedged. Unlike [`record_run`], leaves the failure streak and last error
166/// as they were.
167pub async fn record_deliberate_idle_run(conn: &mut PgConnection, phase: &str) -> ModelResult<()> {
168    sqlx::query!(
169        r#"
170UPDATE credit_registration_phase_state
171SET last_run_finished_at = now(),
172  last_success_at = now()
173WHERE phase = $1
174  AND deleted_at IS NULL
175        "#,
176        phase
177    )
178    .execute(conn)
179    .await?;
180    Ok(())
181}
182
183pub async fn is_paused(conn: &mut PgConnection, phase: &str) -> ModelResult<bool> {
184    let paused = sqlx::query_scalar!(
185        r#"
186SELECT paused_at IS NOT NULL AS "paused!"
187FROM credit_registration_phase_state
188WHERE phase = $1
189  AND deleted_at IS NULL
190        "#,
191        phase
192    )
193    .fetch_one(conn)
194    .await?;
195    Ok(paused)
196}
197
198pub async fn pause(
199    conn: &mut PgConnection,
200    phase: &str,
201    paused_by_user_id: Uuid,
202    pause_reason: Option<&str>,
203) -> ModelResult<()> {
204    sqlx::query!(
205        r#"
206UPDATE credit_registration_phase_state
207SET paused_at = now(),
208  paused_by_user_id = $2,
209  pause_reason = $3
210WHERE phase = $1
211  AND deleted_at IS NULL
212        "#,
213        phase,
214        paused_by_user_id,
215        pause_reason,
216    )
217    .execute(conn)
218    .await?;
219    Ok(())
220}
221
222pub async fn resume(conn: &mut PgConnection, phase: &str) -> ModelResult<()> {
223    sqlx::query!(
224        r#"
225UPDATE credit_registration_phase_state
226SET paused_at = NULL,
227  paused_by_user_id = NULL,
228  pause_reason = NULL
229WHERE phase = $1
230  AND deleted_at IS NULL
231        "#,
232        phase
233    )
234    .execute(conn)
235    .await?;
236    Ok(())
237}
238
239/// Makes the phase due now; the phase loop picks it up on its next `next_run_at` check.
240pub async fn run_now(conn: &mut PgConnection, phase: &str) -> ModelResult<()> {
241    sqlx::query!(
242        r#"
243UPDATE credit_registration_phase_state
244SET next_run_at = now()
245WHERE phase = $1
246  AND deleted_at IS NULL
247        "#,
248        phase
249    )
250    .execute(conn)
251    .await?;
252    Ok(())
253}
254
255pub async fn set_next_run_at(
256    conn: &mut PgConnection,
257    phase: &str,
258    next_run_at: DateTime<Utc>,
259) -> ModelResult<()> {
260    sqlx::query!(
261        r#"
262UPDATE credit_registration_phase_state
263SET next_run_at = $2
264WHERE phase = $1
265  AND deleted_at IS NULL
266        "#,
267        phase,
268        next_run_at,
269    )
270    .execute(conn)
271    .await?;
272    Ok(())
273}
headless_lms_models/credit_registration_phase_state.rs

headless_lms_models/
credit_registration_phase_state.rs
