headless_lms_server/domain/credit_registration_phases/
mod.rs

1//! The thirteen credit-registration pipeline phases and the one-iteration dispatcher.
2//!
3//! Both the worker loops and the test tick endpoint go through [`run_phase_once`], so a phase cannot
4//! behave differently depending on who ran it.
5
6pub mod breaker;
7mod config_validation;
8mod enrolment_discovery;
9mod import;
10mod ledger_snapshot;
11mod link_emails;
12pub mod linking_mail_resend;
13mod product_token_refresh;
14mod resolve_enrolments;
15mod retention_sweep;
16mod student_notifications;
17mod verify;
18pub mod worker_loop;
19
20use headless_lms_base::error::backend_error::BackendError;
21use headless_lms_models::credit_registration_events::{
22    CreditRegistrationEventKind, scrub_text, suotar_exchange_details,
23};
24use headless_lms_models::credit_registrations::{
25    CreditRegistration, CreditRegistrationState, Transition,
26};
27use headless_lms_models::email_templates::{
28    EmailTemplateType, get_generic_email_template_by_type_and_language,
29};
30use headless_lms_models::library::credit_registration::backoff::next_attempt_at;
31use headless_lms_models::library::credit_registration::classification::is_retryable_transient_wire_code;
32use headless_lms_models::library::credit_registration::legacy_mirror::{
33    LEGACY_MIRROR_LIMIT, mirror_successes_to_legacy_ledger,
34};
35use headless_lms_models::library::credit_registration::materialize::{
36    GRADE_IMPROVEMENT_LIMIT, MATERIALIZE_LIMIT, ensure_registration_rows_for_eligible_completions,
37    start_re_attempts_for_improved_grades,
38};
39use headless_lms_models::library::credit_registration::outcomes::{
40    Outcome, RowFacts, request_level_outcome,
41};
42use headless_lms_models::library::credit_registration::preconditions::{
43    PRECONDITIONS_LIMIT, recompute_preconditions,
44};
45use headless_lms_models::{credit_registration_phase_state, credit_registrations};
46use headless_lms_models::{
47    credit_registration_phase_state::PhaseRunOutcome, verified_student_numbers,
48};
49use headless_lms_utils::error::util_error::{SuotarErrorVariant, UtilError, UtilErrorType};
50use headless_lms_utils::prelude::Utc;
51use headless_lms_utils::services::suotar::{
52    ListedPerson, SuotarBatchResponse, SuotarClient, SuotarEndpoint, SuotarItemStatus,
53    SuotarResponseItem,
54};
55use sqlx::{Connection, PgConnection, PgPool};
56use std::collections::{BTreeSet, HashMap};
57use std::future::Future;
58use std::pin::Pin;
59use uuid::Uuid;
60
61/// Which rows one iteration may touch.
62pub use headless_lms_models::credit_registrations::RegistrationScope as PhaseScope;
63
64/// A pipeline phase. [`CreditRegistrationPhase::as_str`] is canonical: it is
65/// `credit_registration_phase_state.phase`, the tick endpoint's `?phase=` and the audit log's
66/// `target_phase`.
67#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
68pub enum CreditRegistrationPhase {
69    Materialize,
70    Preconditions,
71    ResolveEnrolments,
72    Import,
73    Verify,
74    LegacyMirror,
75    StudentNotifications,
76    EnrolmentDiscovery,
77    LinkEmails,
78    ProductTokenRefresh,
79    ConfigValidation,
80    RetentionSweep,
81    LedgerSnapshot,
82}
83
84impl CreditRegistrationPhase {
85    /// Every phase, in pipeline order.
86    pub const ALL: [Self; 13] = [
87        Self::Materialize,
88        Self::Preconditions,
89        Self::ResolveEnrolments,
90        Self::Import,
91        Self::Verify,
92        Self::LegacyMirror,
93        Self::StudentNotifications,
94        Self::EnrolmentDiscovery,
95        Self::LinkEmails,
96        Self::ProductTokenRefresh,
97        Self::ConfigValidation,
98        Self::RetentionSweep,
99        Self::LedgerSnapshot,
100    ];
101
102    /// The phases `run-registrar-tick` runs, in pipeline order. Not every `credit-registrar` phase:
103    /// `legacy-mirror` and `student-notifications` are driven explicitly by specs that need them.
104    pub const REGISTRAR_TICK_SEQUENCE: [Self; 5] = [
105        Self::Materialize,
106        Self::Preconditions,
107        Self::ResolveEnrolments,
108        Self::Import,
109        Self::Verify,
110    ];
111
112    pub fn as_str(self) -> &'static str {
113        match self {
114            Self::Materialize => "materialize",
115            Self::Preconditions => "preconditions",
116            Self::ResolveEnrolments => "resolve-enrolments",
117            Self::Import => "import",
118            Self::Verify => "verify",
119            Self::LegacyMirror => "legacy-mirror",
120            Self::StudentNotifications => "student-notifications",
121            Self::EnrolmentDiscovery => "enrolment-discovery",
122            Self::LinkEmails => "link-emails",
123            Self::ProductTokenRefresh => "product-token-refresh",
124            Self::ConfigValidation => "config-validation",
125            Self::RetentionSweep => "retention-sweep",
126            Self::LedgerSnapshot => "ledger-snapshot",
127        }
128    }
129
130    pub fn from_phase_name(name: &str) -> Option<Self> {
131        Self::ALL.into_iter().find(|phase| phase.as_str() == name)
132    }
133
134    /// Which worker process owns the phase's loop.
135    pub fn process_name(self) -> &'static str {
136        match self {
137            Self::Materialize
138            | Self::Preconditions
139            | Self::ResolveEnrolments
140            | Self::Import
141            | Self::Verify
142            | Self::LegacyMirror
143            | Self::StudentNotifications => "credit-registrar",
144            Self::EnrolmentDiscovery
145            | Self::LinkEmails
146            | Self::ProductTokenRefresh
147            | Self::ConfigValidation
148            | Self::RetentionSweep
149            | Self::LedgerSnapshot => "suotar-syncer",
150        }
151    }
152
153    /// Whether the phase talks to the study registry, and so shares the circuit breaker with the
154    /// other such phases of its own worker process (`breaker::BREAKERS` is process-local, not
155    /// shared between `credit-registrar` and `suotar-syncer`).
156    pub fn calls_study_registry(self) -> bool {
157        matches!(
158            self,
159            Self::ResolveEnrolments
160                | Self::Import
161                | Self::Verify
162                | Self::EnrolmentDiscovery
163                | Self::ProductTokenRefresh
164        )
165    }
166
167    /// The ledger states this phase is the one to move a row out of.
168    ///
169    /// The Workers tab's "queue depth it is responsible for", and the depth the failing-phase alert
170    /// asks about before calling a quiet phase wedged. Empty for the phases whose work is not a
171    /// ledger state at all: `materialize`'s queue is completions with no row yet, and the syncer's
172    /// phases work on course modules. Narrower than what `preconditions` may claim, which is every
173    /// non-terminal row: these are the states nothing else advances.
174    pub fn owned_states(self) -> &'static [CreditRegistrationState] {
175        match self {
176            Self::Preconditions => &[
177                CreditRegistrationState::Pending,
178                CreditRegistrationState::NoUsableEnrolment,
179                CreditRegistrationState::FailedRetryable,
180                CreditRegistrationState::Blocked,
181            ],
182            Self::ResolveEnrolments => &[
183                CreditRegistrationState::ReadyToSubmit,
184                CreditRegistrationState::ResolvingEnrolment,
185            ],
186            Self::Import => &[
187                CreditRegistrationState::CheckingEnrolment,
188                CreditRegistrationState::Submitting,
189            ],
190            Self::Verify => &[
191                CreditRegistrationState::AwaitingVerification,
192                CreditRegistrationState::SubmissionUncertain,
193            ],
194            _ => &[],
195        }
196    }
197
198    pub fn scope_support(self) -> ScopeSupport {
199        match self {
200            Self::Preconditions
201            | Self::ResolveEnrolments
202            | Self::Import
203            | Self::Verify
204            | Self::LegacyMirror
205            | Self::StudentNotifications => ScopeSupport::LEDGER,
206            // No ledger row exists yet, so there is no registration id to narrow on.
207            Self::Materialize => ScopeSupport {
208                course: true,
209                user: true,
210                registration_ids: false,
211            },
212            // These reach their rows through the course module, which has no user dimension: a
213            // roster, a product token and a module configuration are facts about a course, not
214            // about one of our accounts.
215            Self::EnrolmentDiscovery
216            | Self::LinkEmails
217            | Self::ProductTokenRefresh
218            | Self::ConfigValidation => ScopeSupport {
219                course: true,
220                user: false,
221                registration_ids: false,
222            },
223            // Sweeps whole tables by age; there is nothing in them to narrow on.
224            Self::RetentionSweep => ScopeSupport::NONE,
225            // Counts every row in the ledger for the day; a scoped run would write that as if it
226            // were everyone's snapshot.
227            Self::LedgerSnapshot => ScopeSupport::NONE,
228        }
229    }
230}
231
232/// Which of the scope's dimensions a phase's claim query can apply. Declared rather than assumed,
233/// so a phase added later cannot quietly ignore a scope and sweep the whole database.
234#[derive(Debug, Clone, Copy, PartialEq, Eq)]
235pub struct ScopeSupport {
236    pub course: bool,
237    pub user: bool,
238    pub registration_ids: bool,
239}
240
241impl ScopeSupport {
242    pub const NONE: Self = Self {
243        course: false,
244        user: false,
245        registration_ids: false,
246    };
247    /// The phases that claim ledger rows, which carry all three keys themselves.
248    pub const LEDGER: Self = Self {
249        course: true,
250        user: true,
251        registration_ids: true,
252    };
253
254    fn covers(self, scope: &PhaseScope) -> bool {
255        let requested_unsupported = (scope.course_id.is_some() && !self.course)
256            || (scope.user_id.is_some() && !self.user)
257            || (!scope.credit_registration_ids.is_empty() && !self.registration_ids);
258        !requested_unsupported
259    }
260}
261
262/// What one dispatch attempt did.
263#[derive(Debug, Clone, PartialEq)]
264pub enum PhaseTick {
265    Ran(PhaseRunOutcome),
266    /// The phase legitimately did nothing; not counted as a failure.
267    Skipped(PhaseSkipReason),
268    /// The scope names something this phase cannot narrow on; refused rather than run wide.
269    ScopeNotSupported,
270}
271
272#[derive(Debug, Clone, Copy, PartialEq, Eq)]
273pub enum PhaseSkipReason {
274    Paused,
275    CircuitBreakerOpen,
276}
277
278/// Everything a phase iteration needs from its caller: the worker loop or the test tick endpoint.
279pub struct PhaseContext<'a> {
280    pub pool: &'a PgPool,
281    pub suotar_client: &'a SuotarClient,
282    /// Shortens the circuit breaker's cooldown to something a test can wait out.
283    pub test_mode: bool,
284    /// Goes into the audit log's `worker_name` alongside the phase.
285    pub caller: &'a str,
286    /// Absolute base for links in queued mail, which outlive the process that wrote them.
287    pub base_url: &'a str,
288    /// Read by `enrolment-discovery` for the email-match fast track, whose enabled flag doubles as
289    /// its kill switch.
290    pub suotar_conf: &'a headless_lms_base::config::SuotarConfiguration,
291}
292
293impl<'a> PhaseContext<'a> {
294    pub(crate) fn worker_name(&self, phase: CreditRegistrationPhase) -> String {
295        worker_name(self.caller, phase)
296    }
297
298    /// Builds a context from the application configuration, the shape every construction site
299    /// starts from.
300    pub fn from_app(
301        pool: &'a PgPool,
302        suotar_client: &'a SuotarClient,
303        app_conf: &'a headless_lms_base::config::ApplicationConfiguration,
304        caller: &'a str,
305    ) -> Self {
306        Self {
307            pool,
308            suotar_client,
309            test_mode: app_conf.test_mode,
310            caller,
311            base_url: &app_conf.base_url,
312            suotar_conf: &app_conf.suotar_configuration,
313        }
314    }
315}
316
317/// The audit log's `worker_name`, which the database caps at 64 characters.
318pub(crate) fn worker_name(caller: &str, phase: CreditRegistrationPhase) -> String {
319    format!("{caller}/{}", phase.as_str())
320}
321
322/// Both statements that create ledger rows, bounded apart from each other. Together in one phase so
323/// the Workers tab's row-creation counter accounts for every row the pipeline invented.
324async fn run_materialize(
325    ctx: &PhaseContext<'_>,
326    scope: &PhaseScope,
327) -> anyhow::Result<PhaseRunOutcome> {
328    let mut conn = ctx.pool.acquire().await?;
329    let created =
330        ensure_registration_rows_for_eligible_completions(&mut conn, scope, MATERIALIZE_LIMIT)
331            .await?;
332    let re_attempted =
333        start_re_attempts_for_improved_grades(&mut conn, scope, GRADE_IMPROVEMENT_LIMIT).await?;
334    Ok(PhaseRunOutcome::processed(created + re_attempted))
335}
336
337/// Database-only, so it keeps running while the study registry is unreachable.
338async fn run_preconditions(
339    ctx: &PhaseContext<'_>,
340    scope: &PhaseScope,
341) -> anyhow::Result<PhaseRunOutcome> {
342    let mut conn = ctx.pool.acquire().await?;
343    let moved = recompute_preconditions(&mut conn, scope, PRECONDITIONS_LIMIT).await?;
344    Ok(PhaseRunOutcome::processed(moved))
345}
346
347async fn run_legacy_mirror(
348    ctx: &PhaseContext<'_>,
349    scope: &PhaseScope,
350) -> anyhow::Result<PhaseRunOutcome> {
351    let mut conn = ctx.pool.acquire().await?;
352    let mirrored = mirror_successes_to_legacy_ledger(&mut conn, scope, LEGACY_MIRROR_LIMIT).await?;
353    Ok(PhaseRunOutcome::processed(mirrored))
354}
355
356/// Runs exactly one iteration of one phase. The match below is the only place a phase
357/// implementation is registered; it is exhaustive over [`CreditRegistrationPhase`], so a variant
358/// added there without a dispatch arm here fails to compile.
359pub async fn run_phase_once(
360    ctx: &PhaseContext<'_>,
361    phase: CreditRegistrationPhase,
362    scope: &PhaseScope,
363) -> anyhow::Result<PhaseTick> {
364    // Before the pause check: a caller whose narrowing cannot be honoured must not be told it ran.
365    if !phase.scope_support().covers(scope) {
366        return Ok(PhaseTick::ScopeNotSupported);
367    }
368    let mut conn = ctx.pool.acquire().await?;
369    if credit_registration_phase_state::is_paused(&mut conn, phase.as_str()).await? {
370        return Ok(PhaseTick::Skipped(PhaseSkipReason::Paused));
371    }
372    // A scoped run writes nothing to the phase-state row: that row describes the workers, and a
373    // test's traffic in it would make a dead worker look alive to the heartbeat alert.
374    let bookkeeping = scope.is_unscoped();
375    // Before the breaker check, unlike the pause above, which health.rs excludes from the staleness
376    // alert by itself. A cooldown is a worker deliberately waiting, not a worker that died, and
377    // skipping the heartbeat through it would raise a critical alert within a tick or two.
378    if bookkeeping {
379        credit_registration_phase_state::heartbeat(&mut conn, phase.as_str()).await?;
380    }
381    let breaker_key = breaker::ScopeKey::of(scope);
382    if phase.calls_study_registry() && breaker::is_open(&breaker_key) {
383        // Only these stop: an outage must not stall the database-only phases.
384        return Ok(PhaseTick::Skipped(PhaseSkipReason::CircuitBreakerOpen));
385    }
386    drop(conn);
387
388    let body: Pin<Box<dyn Future<Output = anyhow::Result<PhaseRunOutcome>> + '_>> = match phase {
389        CreditRegistrationPhase::Materialize => Box::pin(run_materialize(ctx, scope)),
390        CreditRegistrationPhase::Preconditions => Box::pin(run_preconditions(ctx, scope)),
391        CreditRegistrationPhase::ResolveEnrolments => Box::pin(resolve_enrolments::run(ctx, scope)),
392        CreditRegistrationPhase::Import => Box::pin(import::run(ctx, scope)),
393        CreditRegistrationPhase::Verify => Box::pin(verify::run(ctx, scope)),
394        CreditRegistrationPhase::LegacyMirror => Box::pin(run_legacy_mirror(ctx, scope)),
395        CreditRegistrationPhase::StudentNotifications => {
396            Box::pin(student_notifications::run(ctx, scope))
397        }
398        CreditRegistrationPhase::EnrolmentDiscovery => {
399            Box::pin(enrolment_discovery::run(ctx, scope))
400        }
401        CreditRegistrationPhase::LinkEmails => Box::pin(link_emails::run(ctx, scope)),
402        CreditRegistrationPhase::ProductTokenRefresh => {
403            Box::pin(product_token_refresh::run(ctx, scope))
404        }
405        CreditRegistrationPhase::ConfigValidation => Box::pin(config_validation::run(ctx, scope)),
406        CreditRegistrationPhase::RetentionSweep => Box::pin(retention_sweep::run(ctx, scope)),
407        CreditRegistrationPhase::LedgerSnapshot => Box::pin(ledger_snapshot::run(ctx, scope)),
408    };
409
410    let exchanges_before = ctx.suotar_client.exchange_count();
411    let outcome = match body.await {
412        Ok(outcome) => outcome,
413        Err(error) => PhaseRunOutcome {
414            items_processed: 0,
415            items_failed: 0,
416            error: Some(scrub_text(&format!("{error:#}"))),
417        },
418    };
419    if let Some(error) = &outcome.error {
420        error!(
421            "Credit registration phase {} failed: {error}",
422            phase.as_str()
423        );
424    }
425    // An iteration that never sent a request says nothing about whether the study registry is up,
426    // so it must neither count against the breaker nor clear a run of failures. Phases share one
427    // breaker, and an empty queue is the common case: without this, a phase with nothing to do
428    // resets the counter every tick and the breaker never opens during an outage.
429    let reached_study_registry = ctx.suotar_client.exchange_count() > exchanges_before;
430    if phase.calls_study_registry() && reached_study_registry {
431        if outcome.error.is_some() {
432            if breaker::record_failure(&breaker_key, breaker::cooldown(ctx.test_mode)) {
433                warn!(
434                    "Pausing the study registry phases for {:?} after {} consecutive failures.",
435                    breaker::cooldown(ctx.test_mode),
436                    breaker::MAX_CONSECUTIVE_SUOTAR_FAILURES
437                );
438            }
439        } else {
440            breaker::record_success(&breaker_key);
441        }
442    }
443    if bookkeeping {
444        let mut conn = ctx.pool.acquire().await?;
445        credit_registration_phase_state::record_run(&mut conn, phase.as_str(), &outcome).await?;
446    }
447    Ok(PhaseTick::Ran(outcome))
448}
449
450/// One template lookup per type and language per iteration rather than per mail. `None` means no
451/// template exists, which a mail phase reports rather than failing the batch it was found in.
452#[derive(Default)]
453pub(crate) struct TemplateCache(HashMap<(EmailTemplateType, String), Option<Uuid>>);
454
455impl TemplateCache {
456    pub(crate) async fn id_for(
457        &mut self,
458        conn: &mut PgConnection,
459        template_type: EmailTemplateType,
460        language: &str,
461    ) -> anyhow::Result<Option<Uuid>> {
462        let key = (template_type, language.to_string());
463        if let Some(id) = self.0.get(&key) {
464            return Ok(*id);
465        }
466        let found =
467            match get_generic_email_template_by_type_and_language(conn, template_type, language)
468                .await
469            {
470                Ok(template) => Some(template.id),
471                Err(error)
472                    if matches!(
473                        error.error_type(),
474                        headless_lms_models::ModelErrorType::RecordNotFound
475                    ) =>
476                {
477                    None
478                }
479                Err(error) => return Err(error.into()),
480            };
481        self.0.insert(key, found);
482        Ok(found)
483    }
484}
485
486/// A phase whose whole body is "claim rows, look up each one's template, skip it if the template is
487/// missing, otherwise queue a mail". `link-emails` and `student-notifications` are its only two
488/// shapes; [`run_mail_queue_phase`] is the loop they share.
489pub(crate) trait MailQueuePhase {
490    type Item;
491    /// Per-run state [`queue`](Self::queue) may want across items, the way [`TemplateCache`] is
492    /// kept across items already. `()` for a phase that needs none.
493    type Cache: Default;
494
495    async fn claim(conn: &mut PgConnection, scope: &PhaseScope) -> anyhow::Result<Vec<Self::Item>>;
496
497    fn template_type(item: &Self::Item) -> EmailTemplateType;
498    fn language(item: &Self::Item) -> String;
499
500    /// Inserts the delivery and records it on the claimed item, given the template the caller
501    /// already resolved.
502    async fn queue(
503        ctx: &PhaseContext<'_>,
504        conn: &mut PgConnection,
505        item: &Self::Item,
506        template_id: Uuid,
507        cache: &mut Self::Cache,
508    ) -> anyhow::Result<()>;
509
510    /// One entry of the missing-templates report, e.g. the language alone or a type-and-language
511    /// pair, depending on whether the phase has more than one template type.
512    fn missing_template_label(template_type: EmailTemplateType, language: &str) -> String;
513
514    /// The fixed lead-in of the missing-templates error message.
515    fn missing_templates_error_prefix() -> &'static str;
516}
517
518/// Claims, resolves templates for, and queues mail for one iteration of a [`MailQueuePhase`]. A mail
519/// with no template is skipped rather than failing the iteration: the batch is one transaction, so an
520/// error would roll back every mail that could be queued, and the claimed rows stay claimable.
521pub(crate) async fn run_mail_queue_phase<P: MailQueuePhase>(
522    ctx: &PhaseContext<'_>,
523    scope: &PhaseScope,
524) -> anyhow::Result<PhaseRunOutcome> {
525    let mut conn = ctx.pool.acquire().await?;
526    let mut tx = conn.begin().await?;
527    let claimed = P::claim(&mut tx, scope).await?;
528    let mut templates = TemplateCache::default();
529    let mut cache = P::Cache::default();
530    let mut missing_templates: BTreeSet<String> = BTreeSet::new();
531    let mut skipped = 0;
532    for item in &claimed {
533        let template_type = P::template_type(item);
534        let language = P::language(item);
535        let Some(template_id) = templates.id_for(&mut tx, template_type, &language).await? else {
536            missing_templates.insert(P::missing_template_label(template_type, &language));
537            skipped += 1;
538            continue;
539        };
540        P::queue(ctx, &mut tx, item, template_id, &mut cache).await?;
541    }
542    tx.commit().await?;
543
544    Ok(PhaseRunOutcome {
545        items_processed: i32::try_from(claimed.len()).unwrap_or(i32::MAX),
546        items_failed: skipped,
547        error: (!missing_templates.is_empty()).then(|| {
548            format!(
549                "{} {}.",
550                P::missing_templates_error_prefix(),
551                missing_templates.into_iter().collect::<Vec<_>>().join(", ")
552            )
553        }),
554    })
555}
556
557/// What one iteration of a [`SuotarBatchPhase`] settled before it sent anything.
558pub(crate) struct Prepared<Row, Item> {
559    /// The rows the batch is built from, each with the request item it became.
560    pub sendable: Vec<(Row, Item)>,
561    /// Rows the preflight already wrote a decision for, so no answer is owed for them.
562    pub decided: i32,
563    /// How many of `decided` ended up carrying an error code.
564    pub failed: i32,
565}
566
567impl<Row, Item> Default for Prepared<Row, Item> {
568    fn default() -> Self {
569        Self {
570            sendable: Vec::new(),
571            decided: 0,
572            failed: 0,
573        }
574    }
575}
576
577/// A phase whose iteration is "claim rows, decide in one transaction what may be asked, send one
578/// batch, write one answer per row". `import`, `resolve-enrolments`, and each of `verify`'s two
579/// flows; [`run_suotar_batch_phase`] is the loop they share, and the only place the transaction
580/// shape, the moved-on skipping and the counters are written down.
581pub(crate) trait SuotarBatchPhase {
582    /// A row to send for, with whatever its preflight read alongside it.
583    type Row;
584    /// The request item, which is also what the audit log records as sent.
585    type Item: serde::Serialize;
586    /// The endpoint's per-item result body.
587    type Result;
588
589    /// The iteration's error when every item came back transiently unavailable.
590    const ALL_TRANSIENT_ERROR: &'static str;
591
592    /// Claims rows and decides what may be asked about them. Whatever has to be true before the
593    /// request leaves is written here, in the caller's transaction.
594    async fn prepare(
595        &mut self,
596        ctx: &PhaseContext<'_>,
597        conn: &mut PgConnection,
598        scope: &PhaseScope,
599    ) -> anyhow::Result<Prepared<Self::Row, Self::Item>>;
600
601    fn registration(row: &Self::Row) -> &CreditRegistration;
602
603    /// What the row was addressed as, and so how its answer is found again.
604    fn request_item_id(row: &Self::Row) -> String;
605
606    /// The student number this row's request carried, where it carried one: a number the registry
607    /// rejects may only cost the link it was sent under.
608    fn sent_student_number(_row: &Self::Row) -> Option<&str> {
609        None
610    }
611
612    async fn send(
613        &self,
614        ctx: &PhaseContext<'_>,
615        rows: &[Self::Row],
616        items: Vec<Self::Item>,
617    ) -> Result<SuotarBatchResponse<Self::Result>, UtilError>;
618
619    /// Applies one answer, or the absence of one, to its row. Returns whether the row ended up in a
620    /// failure state; errors with `PreconditionFailed` if another writer moved the row meanwhile.
621    async fn apply(
622        &self,
623        conn: &mut PgConnection,
624        row: &Self::Row,
625        item: Option<&SuotarResponseItem<Self::Result>>,
626        event: OutcomeEvent<'_>,
627    ) -> anyhow::Result<bool>;
628
629    /// What one row gets when the study registry rejected the whole request.
630    async fn apply_request_rejection(
631        &self,
632        conn: &mut PgConnection,
633        row: &Self::Row,
634        request: &serde_json::Value,
635        error: &UtilError,
636    ) -> anyhow::Result<bool>;
637}
638
639/// Runs one iteration of a [`SuotarBatchPhase`].
640///
641/// `items_processed` counts the rows this iteration wrote a decision for: the preflight's included,
642/// the ones another writer had moved on before the answer could be applied excluded.
643/// `items_failed` counts how many of those ended up carrying an error code.
644pub(crate) async fn run_suotar_batch_phase<P: SuotarBatchPhase>(
645    phase: &mut P,
646    ctx: &PhaseContext<'_>,
647    scope: &PhaseScope,
648) -> anyhow::Result<PhaseRunOutcome> {
649    let mut conn = ctx.pool.acquire().await?;
650    let mut tx = conn.begin().await?;
651    let prepared = phase.prepare(ctx, &mut tx, scope).await?;
652    tx.commit().await?;
653    // Held only for the claim; the Suotar call below can pin it for the whole request timeout.
654    drop(conn);
655
656    let mut processed = prepared.decided;
657    let mut items_failed = prepared.failed;
658    if prepared.sendable.is_empty() {
659        return Ok(PhaseRunOutcome {
660            items_processed: processed,
661            items_failed,
662            error: None,
663        });
664    }
665    let (rows, items): (Vec<_>, Vec<_>) = prepared.sendable.into_iter().unzip();
666    let requests = requests_json(&items);
667
668    let response = match phase.send(ctx, &rows, items).await {
669        Ok(response) => response,
670        Err(error) => {
671            let mut conn = ctx.pool.acquire().await?;
672            for (row, request) in rows.iter().zip(requests.iter()) {
673                let applied = phase
674                    .apply_request_rejection(&mut conn, row, request, &error)
675                    .await;
676                count_applied(
677                    applied,
678                    P::registration(row),
679                    &mut processed,
680                    &mut items_failed,
681                )?;
682            }
683            return Ok(PhaseRunOutcome {
684                items_processed: processed,
685                items_failed,
686                error: Some(scrub_text(error.message())),
687            });
688        }
689    };
690
691    let mut conn = ctx.pool.acquire().await?;
692    for (row, request) in rows.iter().zip(requests.iter()) {
693        let request_item_id = P::request_item_id(row);
694        let response_json = response_item_json(&response.raw_response, &request_item_id);
695        let event = OutcomeEvent {
696            suotar_api_call_id: response.call_id,
697            request: Some(request),
698            response: response_json.as_ref(),
699            sent_student_number: P::sent_student_number(row),
700            ..OutcomeEvent::default()
701        };
702        let applied = phase
703            .apply(&mut conn, row, response.item(&request_item_id), event)
704            .await;
705        count_applied(
706            applied,
707            P::registration(row),
708            &mut processed,
709            &mut items_failed,
710        )?;
711    }
712
713    Ok(PhaseRunOutcome {
714        items_processed: processed,
715        items_failed,
716        error: every_item_failed_transiently(&response).then(|| P::ALL_TRANSIENT_ERROR.to_string()),
717    })
718}
719
720/// Counts one written row, or skips one that had already moved on. Skipped rather than propagated:
721/// the row belongs to whoever moved it, and aborting would leave the rest of the batch in the state
722/// the preflight wrote, which no phase claims again.
723fn count_applied(
724    applied: anyhow::Result<bool>,
725    row: &CreditRegistration,
726    processed: &mut i32,
727    items_failed: &mut i32,
728) -> anyhow::Result<()> {
729    match applied {
730        Ok(failed) => {
731            *processed += 1;
732            *items_failed += i32::from(failed);
733            Ok(())
734        }
735        Err(error) if row_moved_on(&error) => {
736            warn!(
737                "Credit registration {} moved on while the study registry answered; leaving it. {error:#}",
738                row.id
739            );
740            Ok(())
741        }
742        Err(error) => Err(error),
743    }
744}
745
746/// Templates are stored per language and courses carry a locale. The course's language, not the
747/// recipient's: the linking mail's recipient may have no account here, and an account records no UI
748/// language to prefer.
749pub(crate) fn template_language(course_language_code: &str) -> String {
750    course_language_code
751        .split(['-', '_'])
752        .next()
753        .unwrap_or(course_language_code)
754        .to_lowercase()
755}
756
757/// Every address the study registry holds for a listed person, in the order it lists them; which
758/// one they read is not something we can know.
759pub(crate) fn listed_person_addresses(person: &ListedPerson) -> Vec<String> {
760    [
761        Some(person.primary_email.clone()),
762        person.secondary_email.clone(),
763    ]
764    .into_iter()
765    .flatten()
766    .filter(|address| !address.trim().is_empty())
767    .collect()
768}
769
770/// The request bodies as sent, kept alongside the typed items so a rejected batch can pair each row
771/// with what was actually asked of it for the audit log.
772pub(crate) fn requests_json<T: serde::Serialize>(items: &[T]) -> Vec<serde_json::Value> {
773    items
774        .iter()
775        .map(|item| serde_json::to_value(item).unwrap_or_default())
776        .collect()
777}
778
779/// The response item for one request item, read from the raw body rather than rebuilt from the
780/// typed value, so the audit trail holds what actually arrived.
781pub(crate) fn response_item_json(
782    raw_response: &serde_json::Value,
783    request_item_id: &str,
784) -> Option<serde_json::Value> {
785    raw_response
786        .as_array()?
787        .iter()
788        .find(|item| item.get("requestItemId").and_then(|id| id.as_str()) == Some(request_item_id))
789        .cloned()
790}
791
792/// Applies one decided outcome to one row, with the exchange that produced it.
793///
794/// `expected_from_state` guards against writing back a decision made from a row snapshot that an
795/// `await` (an external call, or just the gap since claiming) has let go stale: pass the state the
796/// phase itself put the row in before that `await`, or `Some(registration.state)` when the phase
797/// never moved the row before its own `await`.
798pub(crate) async fn apply_outcome(
799    conn: &mut PgConnection,
800    registration: &CreditRegistration,
801    outcome: &Outcome,
802    event: OutcomeEvent<'_>,
803    expected_from_state: Option<CreditRegistrationState>,
804) -> anyhow::Result<()> {
805    // Only if the request carried this number: a student who linked a working one while the request
806    // was out must not lose the link they just made.
807    if outcome.drop_verified_student_number
808        && let Some(linked) =
809            verified_student_numbers::get_by_user_id(conn, registration.user_id).await?
810        && event.sent_student_number == Some(linked.student_number.as_str())
811    {
812        verified_student_numbers::soft_delete(conn, linked.id).await?;
813    }
814    if outcome.increment_submit_retry_count {
815        credit_registrations::increment_submit_retry_count(conn, registration.id).await?;
816    }
817    credit_registrations::transition(
818        conn,
819        registration.id,
820        &Transition {
821            error_message: event.error_message.map(scrub_text),
822            event_kind: CreditRegistrationEventKind::SuotarResponse,
823            event_message: event.message.map(str::to_string),
824            suotar_api_call_id: event.suotar_api_call_id,
825            event_details: Some(suotar_exchange_details(event.request, event.response)),
826            ..outcome_transition(outcome, expected_from_state)
827        },
828    )
829    .await?;
830    Ok(())
831}
832
833/// The ledger write one decided outcome asks for, without the audit half [`apply_outcome`] adds.
834/// For the paths that decide an outcome without an exchange to record.
835pub(crate) fn outcome_transition(
836    outcome: &Outcome,
837    expected_from_state: Option<CreditRegistrationState>,
838) -> Transition {
839    Transition {
840        error_code: outcome.error_code,
841        needs_admin_attention: outcome.needs_admin_attention,
842        expected_from_state,
843        next_attempt_at: outcome
844            .delay_secs
845            .map(|delay_secs| next_attempt_at(Utc::now(), delay_secs)),
846        ..Transition::to(outcome.to_state)
847    }
848}
849
850/// Whether the error is `transition` refusing to write because another writer moved the row since
851/// the snapshot the decision was made from.
852///
853/// A phase that hits this on one row of a batch must skip that row and carry on: the row belongs to
854/// whoever moved it, and aborting the loop would leave every row after it in the state the phase's
855/// own preflight wrote, with no phase claiming that state again.
856pub(crate) fn row_moved_on(error: &anyhow::Error) -> bool {
857    error
858        .downcast_ref::<headless_lms_models::ModelError>()
859        .is_some_and(|error| {
860            matches!(
861                error.error_type(),
862                headless_lms_models::ModelErrorType::PreconditionFailed
863            )
864        })
865}
866
867/// Whether an outcome counts against the iteration's `items_failed`: an error code is a failed
868/// item, so a verify poll answered `notRegistered` is not one.
869pub(crate) fn counts_as_failed(outcome: &Outcome) -> bool {
870    outcome.error_code.is_some()
871}
872
873/// The scheduling history one outcome decision needs from a row.
874pub(crate) fn row_facts(row: &CreditRegistration) -> RowFacts {
875    RowFacts {
876        now: Utc::now(),
877        first_failed_at: row.first_failed_at,
878        submit_retry_count: row.submit_retry_count,
879        verify_attempt_count: row.verify_attempt_count,
880        submitted_at: row.submitted_at,
881    }
882}
883
884/// Whether the whole batch came back saying "not now", so the worker stops burning calls. A batch
885/// with one good item is a success: something moved.
886pub(crate) fn every_item_failed_transiently<R>(response: &SuotarBatchResponse<R>) -> bool {
887    !response.items.is_empty()
888        && response.items.iter().all(|item| {
889            item.status == SuotarItemStatus::Error && is_retryable_transient_wire_code(&item.code)
890        })
891}
892
893/// Applies one request-level outcome to one row of a batch the study registry rejected whole.
894/// Returns whether the row ended up carrying an error code.
895///
896/// `expected_from_state` is the state the phase's own preflight put every row in, since the rows
897/// were read before that transition and are stale by the time this runs.
898pub(crate) async fn apply_request_level_outcome(
899    conn: &mut PgConnection,
900    endpoint: SuotarEndpoint,
901    row: &CreditRegistration,
902    request: &serde_json::Value,
903    error: &UtilError,
904    expected_from_state: CreditRegistrationState,
905) -> anyhow::Result<bool> {
906    let outcome = request_level_outcome(endpoint, suotar_error_variant(error), &row_facts(row));
907    apply_outcome(
908        conn,
909        row,
910        &outcome,
911        OutcomeEvent {
912            message: Some("The study registry rejected the whole request."),
913            error_message: Some(error.message()),
914            request: Some(request),
915            ..OutcomeEvent::default()
916        },
917        Some(expected_from_state),
918    )
919    .await?;
920    Ok(counts_as_failed(&outcome))
921}
922
923/// A failure that never reached the study registry is safe to send again; everything else may have
924/// been acted on. Anything that is not a client error was raised before the request was built.
925fn suotar_error_variant(error: &UtilError) -> SuotarErrorVariant {
926    match error.error_type() {
927        UtilErrorType::SuotarClientError(variant) => *variant,
928        _ => SuotarErrorVariant::TransportNotDelivered,
929    }
930}
931
932/// The audit half of applying an outcome. Both bodies are scrubbed on the way into the event row.
933#[derive(Default)]
934pub(crate) struct OutcomeEvent<'a> {
935    /// The student number this row's request actually carried, which may no longer be the linked
936    /// one by the time the answer is applied.
937    pub sent_student_number: Option<&'a str>,
938    pub message: Option<&'a str>,
939    /// Persisted on the ledger row, so it is scrubbed before it is written.
940    pub error_message: Option<&'a str>,
941    pub suotar_api_call_id: Option<Uuid>,
942    pub request: Option<&'a serde_json::Value>,
943    pub response: Option<&'a serde_json::Value>,
944}
945
946#[cfg(test)]
947mod tests {
948    use headless_lms_models::credit_registration_phase_state::PHASES;
949
950    use super::*;
951
952    /// A mismatch with the seeded `credit_registration_phase_state` rows makes a tick or a
953    /// heartbeat silently target a row that does not exist.
954    #[test]
955    fn phase_names_match_the_seeded_rows() {
956        let from_enum: Vec<&str> = CreditRegistrationPhase::ALL
957            .iter()
958            .map(|phase| phase.as_str())
959            .collect();
960        assert_eq!(from_enum, PHASES);
961        assert_eq!(from_enum.len(), 13);
962    }
963
964    /// A phase that cannot honour the narrowing it was handed has to say so, or a caller that
965    /// believes it narrowed the run gets a silently wrong answer.
966    #[test]
967    fn a_phase_refuses_a_scope_it_cannot_apply() {
968        let ids = PhaseScope {
969            credit_registration_ids: vec![Uuid::new_v4()],
970            ..PhaseScope::default()
971        };
972        assert!(
973            !CreditRegistrationPhase::Materialize
974                .scope_support()
975                .covers(&ids)
976        );
977        assert!(CreditRegistrationPhase::Import.scope_support().covers(&ids));
978        assert!(
979            !CreditRegistrationPhase::RetentionSweep
980                .scope_support()
981                .covers(&PhaseScope::for_course(Uuid::new_v4()))
982        );
983    }
984
985    #[test]
986    fn the_audit_name_says_who_ran_the_phase() {
987        for caller in ["credit-registrar", "run-tick"] {
988            for phase in CreditRegistrationPhase::ALL {
989                let name = worker_name(caller, phase);
990                assert!(name.starts_with(caller));
991                assert!(name.ends_with(phase.as_str()));
992                assert!(name.len() <= 64, "{name}");
993            }
994        }
995    }
996
997    #[test]
998    fn a_locale_narrows_to_the_language_the_templates_are_stored_under() {
999        assert_eq!(template_language("fi-FI"), "fi");
1000        assert_eq!(template_language("en_US"), "en");
1001        assert_eq!(template_language("en"), "en");
1002    }
1003
1004    #[test]
1005    fn a_response_item_is_found_by_its_request_item_id() {
1006        let raw = serde_json::json!([
1007            { "requestItemId": "cr-1", "status": "ok", "code": "sent" },
1008            { "requestItemId": "cr-2", "status": "error", "code": "sisuTimeout" },
1009        ]);
1010        assert_eq!(
1011            response_item_json(&raw, "cr-2").and_then(|item| item
1012                .get("code")
1013                .and_then(|code| code.as_str().map(str::to_string))),
1014            Some("sisuTimeout".to_string())
1015        );
1016        assert_eq!(response_item_json(&raw, "cr-9"), None);
1017    }
1018}
headless_lms_server/domain/credit_registration_phases/mod.rs

headless_lms_server/domain/credit_registration_phases/
mod.rs
