headless_lms_credit_registration/use_cases/resolve_enrolments/
mod.rs

1//! The `resolve-enrolments` phase: which enrolment the attainment belongs to, and what we will send.
2//!
3//! Ends with the payload frozen and the row queued for import in `checking_enrolment`, never
4//! `submitting`: that state means a request may be in flight, and is the import phase's to write.
5//!
6//! The row spends the Suotar round trip itself in `resolving_enrolment`, not `checking_enrolment`:
7//! `import`'s claim query reads the latter, and the row's own claim lock is gone as soon as the
8//! preflight transaction commits. Landing in a state `import` does not claim keeps a second tick of
9//! `import` from sending a request before the enrolment this one resolves is known.
10//!
11//! A row parked in `no_usable_enrolment` is checked where it stands instead, under
12//! [`claim_enrolment_checks`], so a check that finds nothing leaves it there with only its schedule
13//! and last check time moved.
14//!
15//! Each iteration first looks up the Sisu person for links that lack one; see [`persons`].
16
17mod enrolments;
18mod persons;
19
20use headless_lms_models::credit_registrations::{
21    BatchMove, CreditRegistration, CreditRegistrationState, claim_enrolment_checks,
22    restamp_resolving_enrolment, transition_batch,
23};
24use headless_lms_models::library::credit_registration::outcomes::resolving_enrolment;
25use headless_lms_utils::prelude::Utc;
26use sqlx::PgConnection;
27use uuid::Uuid;
28
29use crate::error::CreditRegistrationResult;
30use crate::registry::StudyRegistry;
31use crate::use_cases::batch_flow::{BatchFlowContext, run_registry_batch_flow};
32use crate::workflow::{ClaimedRegistration, Counts};
33
34use enrolments::ResolveEnrolments;
35use persons::ResolvePersonIds;
36
37pub(crate) async fn run<R: StudyRegistry>(
38    ctx: &BatchFlowContext<'_>,
39    registry: &mut R,
40) -> CreditRegistrationResult<Counts> {
41    let mut counts = run_registry_batch_flow::<ResolvePersonIds, _>(ctx, registry).await?;
42    counts += run_registry_batch_flow::<ResolveEnrolments, _>(ctx, registry).await?;
43    Ok(counts)
44}
45
46/// Which kind of lookup a claimed row is on, read from the state it was claimed in.
47#[derive(Debug, Clone, Copy, PartialEq, Eq)]
48enum Lookup {
49    /// On its way to a first resolve, from `ready_to_submit`.
50    FirstResolve,
51    /// Parked in `no_usable_enrolment`, and checked where it stands.
52    ParkedCheck,
53}
54
55impl Lookup {
56    fn of(row: &CreditRegistration) -> Self {
57        if row.state == CreditRegistrationState::NoUsableEnrolment {
58            Self::ParkedCheck
59        } else {
60            Self::FirstResolve
61        }
62    }
63
64    /// The state the row waits out the call in, which the answer's write expects to find.
65    fn in_flight_state(self) -> CreditRegistrationState {
66        match self {
67            Self::FirstResolve => CreditRegistrationState::ResolvingEnrolment,
68            Self::ParkedCheck => CreditRegistrationState::NoUsableEnrolment,
69        }
70    }
71}
72
73/// Claims `row` for its lookup, expecting the in-flight state [`hold_in_flight`] writes in the
74/// claim's transaction.
75fn claim_for_lookup(row: CreditRegistration) -> ClaimedRegistration {
76    let in_flight = Lookup::of(&row).in_flight_state();
77    ClaimedRegistration::moved_to(row, in_flight)
78}
79
80/// Keeps the claimed rows from being claimed again, or imported, while their lookups are out. In the
81/// claim's transaction, whose lock makes each move's guard a confirmation of the state it read.
82async fn hold_in_flight<'a>(
83    conn: &mut PgConnection,
84    claims: impl IntoIterator<Item = &'a ClaimedRegistration>,
85) -> CreditRegistrationResult<()> {
86    let now = Utc::now();
87    let (first_resolves, parked_ids) = split_by_lookup(claims);
88    let moves: Vec<BatchMove> = first_resolves
89        .into_iter()
90        .map(|claim| BatchMove {
91            id: claim.id(),
92            transition: resolving_enrolment().transition(Some(claim.registration().state), now),
93        })
94        .collect();
95    transition_batch(conn, &moves).await?;
96    claim_enrolment_checks(conn, &parked_ids).await?;
97    Ok(())
98}
99
100/// Restarts the recovery grace of the rows a split still holds before each resent half, so a split
101/// that outlasts it does not see them recovered, and their answers discarded, meanwhile.
102async fn keep_lookups_in_flight<'a>(
103    conn: &mut PgConnection,
104    claims: impl IntoIterator<Item = &'a ClaimedRegistration>,
105) -> CreditRegistrationResult<()> {
106    let (first_resolves, parked_ids) = split_by_lookup(claims);
107    let resolving_ids: Vec<Uuid> = first_resolves.iter().map(|claim| claim.id()).collect();
108    restamp_resolving_enrolment(conn, &resolving_ids).await?;
109    claim_enrolment_checks(conn, &parked_ids).await?;
110    Ok(())
111}
112
113/// First resolves, which wait in `resolving_enrolment`, apart from parked checks, which stay where
114/// they are under a check claim.
115fn split_by_lookup<'a>(
116    claims: impl IntoIterator<Item = &'a ClaimedRegistration>,
117) -> (Vec<&'a ClaimedRegistration>, Vec<Uuid>) {
118    let mut first_resolves = Vec::new();
119    let mut parked_ids = Vec::new();
120    for claim in claims {
121        match Lookup::of(claim.registration()) {
122            Lookup::FirstResolve => first_resolves.push(claim),
123            Lookup::ParkedCheck => parked_ids.push(claim.id()),
124        }
125    }
126    (first_resolves, parked_ids)
127}
headless_lms_credit_registration/use_cases/resolve_enrolments/mod.rs

headless_lms_credit_registration/use_cases/resolve_enrolments/
mod.rs
