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    "ledger-snapshot",
24];
25
26#[derive(Debug, Serialize, Deserialize, PartialEq, Clone, ToSchema)]
27pub struct CreditRegistrationPhaseState {
28    pub id: Uuid,
29    pub created_at: DateTime<Utc>,
30    pub updated_at: DateTime<Utc>,
31    pub deleted_at: Option<DateTime<Utc>>,
32    pub phase: String,
33    pub process_name: String,
34    pub expected_interval_secs: i32,
35    pub last_heartbeat_at: Option<DateTime<Utc>>,
36    pub last_run_started_at: Option<DateTime<Utc>>,
37    pub last_run_finished_at: Option<DateTime<Utc>>,
38    pub last_success_at: Option<DateTime<Utc>>,
39    pub next_run_at: Option<DateTime<Utc>>,
40    pub items_processed_last_run: Option<i32>,
41    pub items_failed_last_run: Option<i32>,
42    pub consecutive_failures: i32,
43    pub last_error: Option<String>,
44    pub paused_at: Option<DateTime<Utc>>,
45    pub paused_by_user_id: Option<Uuid>,
46    pub pause_reason: Option<String>,
47}
48
49#[derive(Debug, Clone, PartialEq, Default)]
50pub struct PhaseRunOutcome {
51    pub items_processed: i32,
52    pub items_failed: i32,
53    /// `None` on success. Scrub before passing.
54    pub error: Option<String>,
55}
56
57impl PhaseRunOutcome {
58    /// A clean iteration that moved `count` rows; saturating, so an over-large sweep never reaches
59    /// the dashboard as negative throughput.
60    pub fn processed(count: i64) -> Self {
61        Self {
62            items_processed: count.try_into().unwrap_or(i32::MAX),
63            items_failed: 0,
64            error: None,
65        }
66    }
67}
68
69pub async fn get_all(conn: &mut PgConnection) -> ModelResult<Vec<CreditRegistrationPhaseState>> {
70    let res = sqlx::query_as!(
71        CreditRegistrationPhaseState,
72        r#"
73SELECT *
74FROM credit_registration_phase_state
75WHERE deleted_at IS NULL
76ORDER BY process_name,
77  phase
78        "#,
79    )
80    .fetch_all(conn)
81    .await?;
82    Ok(res)
83}
84
85pub async fn get_by_phase(
86    conn: &mut PgConnection,
87    phase: &str,
88) -> ModelResult<CreditRegistrationPhaseState> {
89    let res = sqlx::query_as!(
90        CreditRegistrationPhaseState,
91        r#"
92SELECT *
93FROM credit_registration_phase_state
94WHERE phase = $1
95  AND deleted_at IS NULL
96        "#,
97        phase
98    )
99    .fetch_one(conn)
100    .await?;
101    Ok(res)
102}
103
104/// Written every iteration, work or not, so idle and wedged stay distinguishable.
105pub async fn heartbeat(conn: &mut PgConnection, phase: &str) -> ModelResult<()> {
106    sqlx::query!(
107        r#"
108UPDATE credit_registration_phase_state
109SET last_heartbeat_at = now(),
110  last_run_started_at = now()
111WHERE phase = $1
112  AND deleted_at IS NULL
113        "#,
114        phase
115    )
116    .execute(conn)
117    .await?;
118    Ok(())
119}
120
121/// Closes out an iteration. Only a success moves `last_success_at`, so a wedged phase stays
122/// distinguishable from a quiet one.
123pub async fn record_run(
124    conn: &mut PgConnection,
125    phase: &str,
126    outcome: &PhaseRunOutcome,
127) -> ModelResult<()> {
128    sqlx::query!(
129        r#"
130UPDATE credit_registration_phase_state
131SET last_run_finished_at = now(),
132  items_processed_last_run = $2,
133  items_failed_last_run = $3,
134  last_success_at = CASE
135    WHEN $4::text IS NULL THEN now()
136    ELSE last_success_at
137  END,
138  consecutive_failures = CASE
139    WHEN $4::text IS NULL THEN 0
140    ELSE consecutive_failures + 1
141  END,
142  last_error = $4
143WHERE phase = $1
144  AND deleted_at IS NULL
145        "#,
146        phase,
147        outcome.items_processed,
148        outcome.items_failed,
149        outcome.error,
150    )
151    .execute(conn)
152    .await?;
153    Ok(())
154}
155
156pub async fn is_paused(conn: &mut PgConnection, phase: &str) -> ModelResult<bool> {
157    let paused = sqlx::query_scalar!(
158        r#"
159SELECT paused_at IS NOT NULL AS "paused!"
160FROM credit_registration_phase_state
161WHERE phase = $1
162  AND deleted_at IS NULL
163        "#,
164        phase
165    )
166    .fetch_one(conn)
167    .await?;
168    Ok(paused)
169}
170
171pub async fn pause(
172    conn: &mut PgConnection,
173    phase: &str,
174    paused_by_user_id: Uuid,
175    pause_reason: Option<&str>,
176) -> ModelResult<()> {
177    sqlx::query!(
178        r#"
179UPDATE credit_registration_phase_state
180SET paused_at = now(),
181  paused_by_user_id = $2,
182  pause_reason = $3
183WHERE phase = $1
184  AND deleted_at IS NULL
185        "#,
186        phase,
187        paused_by_user_id,
188        pause_reason,
189    )
190    .execute(conn)
191    .await?;
192    Ok(())
193}
194
195pub async fn resume(conn: &mut PgConnection, phase: &str) -> ModelResult<()> {
196    sqlx::query!(
197        r#"
198UPDATE credit_registration_phase_state
199SET paused_at = NULL,
200  paused_by_user_id = NULL,
201  pause_reason = NULL
202WHERE phase = $1
203  AND deleted_at IS NULL
204        "#,
205        phase
206    )
207    .execute(conn)
208    .await?;
209    Ok(())
210}
211
212/// Makes the phase due now; the phase loop picks it up on its next `next_run_at` check.
213pub async fn run_now(conn: &mut PgConnection, phase: &str) -> ModelResult<()> {
214    sqlx::query!(
215        r#"
216UPDATE credit_registration_phase_state
217SET next_run_at = now()
218WHERE phase = $1
219  AND deleted_at IS NULL
220        "#,
221        phase
222    )
223    .execute(conn)
224    .await?;
225    Ok(())
226}
227
228pub async fn set_next_run_at(
229    conn: &mut PgConnection,
230    phase: &str,
231    next_run_at: DateTime<Utc>,
232) -> ModelResult<()> {
233    sqlx::query!(
234        r#"
235UPDATE credit_registration_phase_state
236SET next_run_at = $2
237WHERE phase = $1
238  AND deleted_at IS NULL
239        "#,
240        phase,
241        next_run_at,
242    )
243    .execute(conn)
244    .await?;
245    Ok(())
246}
headless_lms_models/credit_registration_phase_state.rs

headless_lms_models/
credit_registration_phase_state.rs
