headless_lms_server/controllers/main_frontend/credit_registration_admin/
dashboard.rs

1//! The Overview tab, the Suotar health panel and phase pause/resume/run-now controls.
2
3use std::collections::HashMap;
4
5use headless_lms_models::credit_registration_admin_actions::{
6    CreditRegistrationAdminAction, CreditRegistrationAdminActionTarget, GLOBAL_ADMIN_ROLE,
7    NewCreditRegistrationAdminAction,
8};
9use headless_lms_models::credit_registration_phase_state;
10use headless_lms_models::credit_registrations::{
11    self, CreditRegistrationErrorCode, CreditRegistrationErrorCodeCount, CreditRegistrationState,
12    OldestNonTerminalRegistration, StuckRegistrationCount,
13};
14use headless_lms_models::library::credit_registration::PendingReasonCounts;
15use headless_lms_models::suotar_api_calls::{
16    self, SuotarEndpoint, SuotarEndpointStanding as SuotarEndpointStandingRow,
17    SuotarEndpointStatsForWindow,
18};
19use utoipa::ToSchema;
20
21use crate::domain::credit_registration::health::{
22    CreditRegistrationHealth, evaluate, is_heartbeat_late, stuck_thresholds,
23};
24use crate::domain::credit_registration_phases::CreditRegistrationPhase;
25use crate::domain::credit_registration_phases::breaker::{
26    MAX_CONSECUTIVE_SUOTAR_FAILURES, ScopeKey, snapshot,
27};
28use crate::prelude::*;
29
30use super::{authorize_credit_registration_admin, required_reason};
31
32const THROUGHPUT_DAYS: i64 = 30;
33
34const ENDPOINT_STATS_WINDOWS_SECS: [i64; 3] = [60 * 60, 24 * 60 * 60, 7 * 24 * 60 * 60];
35
36#[derive(Debug, Serialize, Deserialize, PartialEq, Clone, ToSchema)]
37pub struct CreditRegistrationStateTotal {
38    pub state: CreditRegistrationState,
39    pub count: i64,
40}
41
42#[derive(Debug, Serialize, Deserialize, PartialEq, Clone, ToSchema)]
43pub struct CreditRegistrationErrorCodeTotal {
44    pub error_code: CreditRegistrationErrorCode,
45    /// Rows the pipeline is still working on.
46    pub in_flight_count: i64,
47    /// Rows that ended on this code.
48    pub terminal_failure_count: i64,
49}
50
51#[derive(Debug, Serialize, Deserialize, PartialEq, Clone, ToSchema)]
52pub struct CreditRegistrationOldestNonTerminal {
53    pub credit_registration_id: Uuid,
54    pub state: CreditRegistrationState,
55    pub state_entered_at: DateTime<Utc>,
56    /// Computed server-side: a page comparing its own clock against a server timestamp misjudges
57    /// this on a skewed client, the same reason `seconds_since_heartbeat` is computed here too.
58    pub seconds_in_state: i64,
59}
60
61#[derive(Debug, Serialize, Deserialize, PartialEq, Clone, ToSchema)]
62pub struct CreditRegistrationThroughputBucket {
63    pub day: DateTime<Utc>,
64    pub registered_count: i64,
65    /// `duplicate` and `not_improved`: the credit exists, and we did not put it there.
66    pub other_success_count: i64,
67    pub failed_count: i64,
68}
69
70#[derive(Debug, Serialize, Deserialize, PartialEq, Clone, ToSchema)]
71pub struct CreditRegistrationStuckTotal {
72    pub state: CreditRegistrationState,
73    pub count: i64,
74    pub severely_stuck_count: i64,
75    pub oldest_state_entered_at: Option<DateTime<Utc>>,
76}
77
78/// Where one study registry endpoint stands, over all time.
79#[derive(Debug, Serialize, Deserialize, PartialEq, Clone, ToSchema)]
80pub struct SuotarEndpointStanding {
81    pub endpoint: SuotarEndpoint,
82    pub last_success_at: Option<DateTime<Utc>>,
83    pub last_failure_at: Option<DateTime<Utc>>,
84    pub consecutive_failures: i64,
85}
86
87/// The circuit breaker as this web process holds it. The global key only — a narrowed run gets its own
88/// — and the counters live in process memory, so this says whether this server would currently skip a
89/// study registry call, not whether the workers would.
90#[derive(Debug, Serialize, Deserialize, PartialEq, Clone, ToSchema)]
91pub struct CreditRegistrationCircuitBreakerState {
92    pub open: bool,
93    pub consecutive_failures: i64,
94    pub open_for_secs: Option<i64>,
95    pub trips_after_consecutive_failures: i64,
96}
97
98/// One pipeline phase's heartbeat, written by the worker loops and by unscoped runs only, never by a
99/// narrowed one. Returned by the pause/resume/run-now actions; the Workers tab lists
100/// `CreditRegistrationPhaseRow` instead, which is wider.
101#[derive(Debug, Serialize, Deserialize, PartialEq, Clone, ToSchema)]
102pub struct CreditRegistrationPhaseStatus {
103    pub phase: String,
104    pub process_name: String,
105    pub expected_interval_secs: i32,
106    pub last_heartbeat_at: Option<DateTime<Utc>>,
107    pub last_success_at: Option<DateTime<Utc>>,
108    pub last_run_finished_at: Option<DateTime<Utc>>,
109    pub items_processed_last_run: Option<i32>,
110    pub items_failed_last_run: Option<i32>,
111    pub consecutive_failures: i32,
112    pub paused_at: Option<DateTime<Utc>>,
113    pub pause_reason: Option<String>,
114    /// No implementation is registered for the phase yet, so it has never reported and will not.
115    pub implemented: bool,
116    /// Computed server-side: a page comparing its own clock against a server timestamp misjudges this
117    /// on a skewed client.
118    pub seconds_since_heartbeat: Option<i64>,
119    /// `seconds_since_heartbeat > expected_interval_secs * health.thresholds.phase_heartbeat_interval_multiplier`.
120    /// Always `false` while paused or never heartbeated.
121    pub heartbeat_late: bool,
122}
123
124#[derive(Debug, Serialize, Deserialize, PartialEq, Clone, ToSchema)]
125pub struct CreditRegistrationOverview {
126    pub health: CreditRegistrationHealth,
127    pub counts_by_state: Vec<CreditRegistrationStateTotal>,
128    /// The `pending` depth split by what each row is waiting on, which the ledger does not store.
129    pub pending_by_reason: PendingReasonCounts,
130    pub error_codes: Vec<CreditRegistrationErrorCodeTotal>,
131    pub needs_admin_attention_count: i64,
132    pub oldest_non_terminal: Option<CreditRegistrationOldestNonTerminal>,
133    pub throughput: Vec<CreditRegistrationThroughputBucket>,
134    pub throughput_days: i64,
135    pub stuck: Vec<CreditRegistrationStuckTotal>,
136    pub endpoints: Vec<SuotarEndpointStanding>,
137    pub circuit_breaker: CreditRegistrationCircuitBreakerState,
138}
139
140#[derive(Debug, Serialize, Deserialize, PartialEq, Clone, ToSchema)]
141pub struct SuotarEndpointWindowStats {
142    pub endpoint: SuotarEndpoint,
143    pub call_count: i64,
144    pub failed_call_count: i64,
145    pub in_flight_count: i64,
146    pub ok_item_count: i64,
147    pub error_item_count: i64,
148    pub p50_duration_ms: Option<i32>,
149    pub p95_duration_ms: Option<i32>,
150    pub last_success_at: Option<DateTime<Utc>>,
151    pub last_failure_at: Option<DateTime<Utc>>,
152    /// The registry's own request-level code, an identifier rather than prose.
153    pub last_request_level_error_code: Option<String>,
154}
155
156#[derive(Debug, Serialize, Deserialize, PartialEq, Clone, ToSchema)]
157pub struct SuotarHealthWindow {
158    pub window_secs: i64,
159    pub endpoints: Vec<SuotarEndpointWindowStats>,
160}
161
162#[derive(Debug, Serialize, Deserialize, PartialEq, Clone, ToSchema)]
163pub struct SuotarHealth {
164    pub windows: Vec<SuotarHealthWindow>,
165}
166
167#[derive(Debug, Deserialize, ToSchema)]
168pub struct AdminPausePhasePayload {
169    pub reason: String,
170}
171
172#[derive(Debug, Deserialize, ToSchema)]
173pub struct AdminPhaseActionPayload {
174    pub reason: Option<String>,
175}
176
177/**
178GET `/api/v0/main-frontend/credit-registration-admin/overview` - Everything the Overview tab and the
179alert banner render, in one request so the tiles cannot contradict each other.
180*/
181#[instrument(skip(pool))]
182#[utoipa::path(
183    get,
184    path = "/overview",
185    operation_id = "getCreditRegistrationOverview",
186    tag = "credit-registration-admin",
187    responses(
188        (status = 200, description = "Counts, throughput, phase heartbeats and the active alerts", body = CreditRegistrationOverview)
189    )
190)]
191pub async fn get_credit_registration_overview(
192    user: AuthUser,
193    pool: web::Data<PgPool>,
194) -> ControllerResult<web::Json<CreditRegistrationOverview>> {
195    let mut conn = pool.acquire().await?;
196    let token = authorize_credit_registration_admin(&mut conn, user.id).await?;
197
198    let stuck_rows = credit_registrations::count_stuck(&mut conn, &stuck_thresholds()).await?;
199    let depths = credit_registrations::count_by_state(&mut conn).await?;
200    let health = evaluate(&mut conn, &stuck_rows, &depths).await?;
201    let counts_by_state = depths
202        .iter()
203        .map(|&(state, count)| CreditRegistrationStateTotal { state, count })
204        .collect();
205    let pending_by_reason = credit_registrations::count_pending_by_reason(&mut conn).await?;
206    let error_codes = credit_registrations::count_by_error_code(&mut conn)
207        .await?
208        .into_iter()
209        .map(to_error_code_total)
210        .collect();
211    let needs_admin_attention_count =
212        credit_registrations::count_needing_admin_attention(&mut conn).await?;
213    let oldest_non_terminal = credit_registrations::get_oldest_non_terminal(&mut conn)
214        .await?
215        .map(|row| to_oldest_non_terminal(row, Utc::now()));
216    let throughput = credit_registrations::get_throughput_by_day(
217        &mut conn,
218        Utc::now() - chrono::Duration::days(THROUGHPUT_DAYS),
219    )
220    .await?
221    .into_iter()
222    .map(|row| CreditRegistrationThroughputBucket {
223        day: row.day,
224        registered_count: row.registered_count,
225        other_success_count: row.other_success_count,
226        failed_count: row.failed_count,
227    })
228    .collect();
229    let stuck = stuck_rows.into_iter().map(to_stuck_total).collect();
230    let endpoints = suotar_api_calls::get_endpoint_standings(&mut conn)
231        .await?
232        .into_iter()
233        .map(to_endpoint_standing)
234        .collect();
235
236    token.authorized_ok(web::Json(CreditRegistrationOverview {
237        health,
238        counts_by_state,
239        pending_by_reason,
240        error_codes,
241        needs_admin_attention_count,
242        oldest_non_terminal,
243        throughput,
244        throughput_days: THROUGHPUT_DAYS,
245        stuck,
246        endpoints,
247        circuit_breaker: circuit_breaker_state(),
248    }))
249}
250
251/**
252GET `/api/v0/main-frontend/credit-registration-admin/suotar-health` - Per-endpoint call counts,
253success rates and latency percentiles over an hour, a day and a week.
254*/
255#[instrument(skip(pool))]
256#[utoipa::path(
257    get,
258    path = "/suotar-health",
259    operation_id = "getSuotarHealth",
260    tag = "credit-registration-admin",
261    responses(
262        (status = 200, description = "Study registry traffic per endpoint and window", body = SuotarHealth)
263    )
264)]
265pub async fn get_suotar_health(
266    user: AuthUser,
267    pool: web::Data<PgPool>,
268) -> ControllerResult<web::Json<SuotarHealth>> {
269    let mut conn = pool.acquire().await?;
270    let token = authorize_credit_registration_admin(&mut conn, user.id).await?;
271
272    let mut by_window: HashMap<i64, Vec<SuotarEndpointWindowStats>> = HashMap::new();
273    for row in
274        suotar_api_calls::get_endpoint_stats_for_windows(&mut conn, &ENDPOINT_STATS_WINDOWS_SECS)
275            .await?
276    {
277        by_window
278            .entry(row.window_secs)
279            .or_default()
280            .push(to_endpoint_window_stats_for_window(row));
281    }
282    let windows = ENDPOINT_STATS_WINDOWS_SECS
283        .into_iter()
284        .map(|window_secs| SuotarHealthWindow {
285            window_secs,
286            endpoints: by_window.remove(&window_secs).unwrap_or_default(),
287        })
288        .collect();
289
290    token.authorized_ok(web::Json(SuotarHealth { windows }))
291}
292
293/**
294POST `/api/v0/main-frontend/credit-registration-admin/phases/{phase}/pause` - Pauses one phase: the
295worker loop skips it on every tick until it is resumed.
296*/
297#[instrument(skip(pool, payload))]
298#[utoipa::path(
299    post,
300    path = "/phases/{phase}/pause",
301    operation_id = "adminPausePhase",
302    tag = "credit-registration-admin",
303    params(("phase" = String, Path, description = "One of the twelve canonical phase names")),
304    request_body = AdminPausePhasePayload,
305    responses(
306        (status = 200, description = "The phase's status after pausing", body = CreditRegistrationPhaseStatus),
307        (status = 422, description = "No reason given, or not one of the canonical phase names")
308    )
309)]
310pub async fn admin_pause_phase(
311    user: AuthUser,
312    pool: web::Data<PgPool>,
313    phase: web::Path<String>,
314    payload: web::Json<AdminPausePhasePayload>,
315) -> ControllerResult<web::Json<CreditRegistrationPhaseStatus>> {
316    let mut conn = pool.acquire().await?;
317    let token = authorize_credit_registration_admin(&mut conn, user.id).await?;
318
319    let phase = require_known_phase(&phase)?;
320    let reason = required_reason(&payload.reason)?;
321
322    let mut tx = conn.begin().await?;
323    credit_registration_phase_state::pause(&mut tx, phase, user.id, Some(reason)).await?;
324    models::credit_registration_admin_actions::record(
325        &mut tx,
326        &NewCreditRegistrationAdminAction {
327            target_phase: Some(phase.to_string()),
328            reason: Some(reason.to_string()),
329            ..NewCreditRegistrationAdminAction::new(
330                CreditRegistrationAdminAction::PausePhase,
331                CreditRegistrationAdminActionTarget::Phase,
332                user.id,
333                GLOBAL_ADMIN_ROLE,
334            )
335        },
336    )
337    .await?;
338    tx.commit().await?;
339
340    token.authorized_ok(web::Json(one_phase_status(&mut conn, phase).await?))
341}
342
343/**
344POST `/api/v0/main-frontend/credit-registration-admin/phases/{phase}/resume` - Resumes one paused
345phase.
346*/
347#[instrument(skip(pool, payload))]
348#[utoipa::path(
349    post,
350    path = "/phases/{phase}/resume",
351    operation_id = "adminResumePhase",
352    tag = "credit-registration-admin",
353    params(("phase" = String, Path, description = "One of the twelve canonical phase names")),
354    request_body = AdminPhaseActionPayload,
355    responses(
356        (status = 200, description = "The phase's status after resuming", body = CreditRegistrationPhaseStatus),
357        (status = 422, description = "Not one of the canonical phase names")
358    )
359)]
360pub async fn admin_resume_phase(
361    user: AuthUser,
362    pool: web::Data<PgPool>,
363    phase: web::Path<String>,
364    payload: web::Json<AdminPhaseActionPayload>,
365) -> ControllerResult<web::Json<CreditRegistrationPhaseStatus>> {
366    let mut conn = pool.acquire().await?;
367    let token = authorize_credit_registration_admin(&mut conn, user.id).await?;
368
369    let phase = require_known_phase(&phase)?;
370
371    let mut tx = conn.begin().await?;
372    credit_registration_phase_state::resume(&mut tx, phase).await?;
373    models::credit_registration_admin_actions::record(
374        &mut tx,
375        &NewCreditRegistrationAdminAction {
376            target_phase: Some(phase.to_string()),
377            reason: payload.reason.clone(),
378            ..NewCreditRegistrationAdminAction::new(
379                CreditRegistrationAdminAction::ResumePhase,
380                CreditRegistrationAdminActionTarget::Phase,
381                user.id,
382                GLOBAL_ADMIN_ROLE,
383            )
384        },
385    )
386    .await?;
387    tx.commit().await?;
388
389    token.authorized_ok(web::Json(one_phase_status(&mut conn, phase).await?))
390}
391
392/**
393POST `/api/v0/main-frontend/credit-registration-admin/phases/{phase}/run-now` - Makes one phase due
394immediately: the worker loop picks it up on its next tick instead of waiting out `next_run_at`.
395*/
396#[instrument(skip(pool, payload))]
397#[utoipa::path(
398    post,
399    path = "/phases/{phase}/run-now",
400    operation_id = "adminRunPhaseNow",
401    tag = "credit-registration-admin",
402    params(("phase" = String, Path, description = "One of the twelve canonical phase names")),
403    request_body = AdminPhaseActionPayload,
404    responses(
405        (status = 200, description = "The phase's status after being made due", body = CreditRegistrationPhaseStatus),
406        (status = 422, description = "Not one of the canonical phase names")
407    )
408)]
409pub async fn admin_run_phase_now(
410    user: AuthUser,
411    pool: web::Data<PgPool>,
412    phase: web::Path<String>,
413    payload: web::Json<AdminPhaseActionPayload>,
414) -> ControllerResult<web::Json<CreditRegistrationPhaseStatus>> {
415    let mut conn = pool.acquire().await?;
416    let token = authorize_credit_registration_admin(&mut conn, user.id).await?;
417
418    let phase = require_known_phase(&phase)?;
419
420    let mut tx = conn.begin().await?;
421    credit_registration_phase_state::run_now(&mut tx, phase).await?;
422    models::credit_registration_admin_actions::record(
423        &mut tx,
424        &NewCreditRegistrationAdminAction {
425            target_phase: Some(phase.to_string()),
426            reason: payload.reason.clone(),
427            ..NewCreditRegistrationAdminAction::new(
428                CreditRegistrationAdminAction::RunPhaseNow,
429                CreditRegistrationAdminActionTarget::Phase,
430                user.id,
431                GLOBAL_ADMIN_ROLE,
432            )
433        },
434    )
435    .await?;
436    tx.commit().await?;
437
438    token.authorized_ok(web::Json(one_phase_status(&mut conn, phase).await?))
439}
440
441/// Resolves a path segment to the spelling `credit_registration_phase_state` stores, refusing anything
442/// that is not a canonical phase name.
443fn require_known_phase(phase: &str) -> Result<&'static str, ControllerError> {
444    CreditRegistrationPhase::from_phase_name(phase)
445        .map(CreditRegistrationPhase::as_str)
446        .ok_or_else(|| {
447            controller_err!(
448                BadRequest,
449                "Not one of the canonical phase names.".to_string()
450            )
451        })
452}
453
454/// One phase's status, so a pause/resume/run-now response shows the effect without a second request.
455async fn one_phase_status(
456    conn: &mut PgConnection,
457    phase: &str,
458) -> Result<CreditRegistrationPhaseStatus, ControllerError> {
459    let row = credit_registration_phase_state::get_by_phase(conn, phase).await?;
460    Ok(to_phase_status(row, Utc::now()))
461}
462
463fn to_phase_status(
464    row: credit_registration_phase_state::CreditRegistrationPhaseState,
465    now: DateTime<Utc>,
466) -> CreditRegistrationPhaseStatus {
467    let seconds_since_heartbeat = row.last_heartbeat_at.map(|at| (now - at).num_seconds());
468    let heartbeat_late = is_heartbeat_late(
469        row.last_heartbeat_at,
470        row.expected_interval_secs,
471        row.paused_at,
472        now,
473    );
474    CreditRegistrationPhaseStatus {
475        implemented: CreditRegistrationPhase::from_phase_name(&row.phase).is_some(),
476        phase: row.phase,
477        process_name: row.process_name,
478        expected_interval_secs: row.expected_interval_secs,
479        last_heartbeat_at: row.last_heartbeat_at,
480        last_success_at: row.last_success_at,
481        last_run_finished_at: row.last_run_finished_at,
482        items_processed_last_run: row.items_processed_last_run,
483        items_failed_last_run: row.items_failed_last_run,
484        consecutive_failures: row.consecutive_failures,
485        paused_at: row.paused_at,
486        pause_reason: row.pause_reason,
487        seconds_since_heartbeat,
488        heartbeat_late,
489    }
490}
491
492fn circuit_breaker_state() -> CreditRegistrationCircuitBreakerState {
493    let state = snapshot(&ScopeKey::Global);
494    CreditRegistrationCircuitBreakerState {
495        open: state.open,
496        consecutive_failures: i64::from(state.consecutive_failures),
497        open_for_secs: state.open_for_secs.map(|secs| secs as i64),
498        trips_after_consecutive_failures: i64::from(MAX_CONSECUTIVE_SUOTAR_FAILURES),
499    }
500}
501
502fn to_error_code_total(row: CreditRegistrationErrorCodeCount) -> CreditRegistrationErrorCodeTotal {
503    CreditRegistrationErrorCodeTotal {
504        error_code: row.error_code,
505        in_flight_count: row.in_flight_count,
506        terminal_failure_count: row.terminal_failure_count,
507    }
508}
509
510fn to_oldest_non_terminal(
511    row: OldestNonTerminalRegistration,
512    now: DateTime<Utc>,
513) -> CreditRegistrationOldestNonTerminal {
514    CreditRegistrationOldestNonTerminal {
515        credit_registration_id: row.id,
516        state: row.state,
517        seconds_in_state: (now - row.state_entered_at).num_seconds(),
518        state_entered_at: row.state_entered_at,
519    }
520}
521
522fn to_stuck_total(row: StuckRegistrationCount) -> CreditRegistrationStuckTotal {
523    CreditRegistrationStuckTotal {
524        state: row.state,
525        count: row.count,
526        severely_stuck_count: row.severely_stuck_count,
527        oldest_state_entered_at: row.oldest_state_entered_at,
528    }
529}
530
531fn to_endpoint_standing(row: SuotarEndpointStandingRow) -> SuotarEndpointStanding {
532    SuotarEndpointStanding {
533        endpoint: row.endpoint,
534        last_success_at: row.last_success_at,
535        last_failure_at: row.last_failure_at,
536        consecutive_failures: row.consecutive_failures,
537    }
538}
539
540fn to_endpoint_window_stats_for_window(
541    row: SuotarEndpointStatsForWindow,
542) -> SuotarEndpointWindowStats {
543    SuotarEndpointWindowStats {
544        endpoint: row.endpoint,
545        call_count: row.call_count,
546        failed_call_count: row.failed_call_count,
547        in_flight_count: row.in_flight_count,
548        ok_item_count: row.ok_item_count,
549        error_item_count: row.error_item_count,
550        p50_duration_ms: row.p50_duration_ms,
551        p95_duration_ms: row.p95_duration_ms,
552        last_success_at: row.last_success_at,
553        last_failure_at: row.last_failure_at,
554        last_request_level_error_code: row.last_request_level_error_code,
555    }
556}
557
558pub fn _add_routes(cfg: &mut ServiceConfig) {
559    cfg.route("/overview", web::get().to(get_credit_registration_overview))
560        .route("/suotar-health", web::get().to(get_suotar_health))
561        .route("/phases/{phase}/pause", web::post().to(admin_pause_phase))
562        .route("/phases/{phase}/resume", web::post().to(admin_resume_phase))
563        .route(
564            "/phases/{phase}/run-now",
565            web::post().to(admin_run_phase_now),
566        );
567}
headless_lms_server/controllers/main_frontend/credit_registration_admin/dashboard.rs

headless_lms_server/controllers/main_frontend/credit_registration_admin/
dashboard.rs
