headless_lms_credit_registration/use_cases/import/
claim.rs

1//! Claiming the rows an import iteration sends: which of them may go out, and the submission each
2//! one carries.
3
4use headless_lms_models::course_module_completion_registered_to_study_registries::completion_ids_registered_by_a_registrar;
5use headless_lms_models::credit_registration_events::{
6    self, CreditRegistrationEventKind, NewCreditRegistrationEvent,
7};
8use headless_lms_models::credit_registrations::{
9    AdminAttention, CreditRegistration, CreditRegistrationState, claim_due_for_import,
10    schedule_next_attempt, set_needs_admin_attention,
11};
12use headless_lms_models::library::credit_registration::backoff::SUBMIT_MAX_BACKOFF;
13use headless_lms_models::library::credit_registration::grade_mapping::{
14    MappedGrade, is_known_grade,
15};
16use headless_lms_models::library::credit_registration::outcomes::{
17    UnaskedMove, duplicate_of_other_registrar, incomplete_payload, invalid_payload_field,
18    submitting, unknown_grade,
19};
20use headless_lms_models::library::credit_registration::scrub::scrub_text;
21use headless_lms_utils::prelude::Utc;
22use secrecy::ExposeSecret;
23use sqlx::{Connection, PgConnection};
24use uuid::Uuid;
25
26use crate::error::{CreditRegistrationError, CreditRegistrationResult};
27use crate::error_reports::ErrorReporter;
28use crate::registry::{AttainmentSubmission, CourseCode, Credits, StudentNumber};
29use crate::use_cases::batch_flow::{BatchFlowContext, Prepared};
30use crate::workflow::{Applied, Claimed, ClaimedRegistration, write_unasked_move};
31
32/// Claims at most `limit` rows due for import, moves each one it may send to `submitting`, and
33/// settles the rest where they stand.
34pub(super) async fn claim_import_candidates(
35    ctx: &BatchFlowContext<'_>,
36    conn: &mut PgConnection,
37    limit: usize,
38) -> CreditRegistrationResult<Prepared<(), AttainmentSubmission>> {
39    let claimed =
40        claim_due_for_import(conn, ctx.scope, i64::try_from(limit).unwrap_or(i64::MAX)).await?;
41    // Registrars only, not our own mirror rows: a grade improvement is deliberately a second
42    // submission for the same completion.
43    let already_registered = completion_ids_registered_by_a_registrar(
44        conn,
45        &claimed
46            .iter()
47            .map(|row| row.course_module_completion_id)
48            .collect::<Vec<_>>(),
49    )
50    .await?;
51
52    let mut prepared = Prepared::new();
53    for row in claimed {
54        let (unasked, submission) = match preflight_submission(&row, &already_registered) {
55            Preflight::Send(submission) => (submitting(), Some(submission)),
56            Preflight::Settle(unasked) => (unasked, None),
57        };
58        // The claim holds the row's lock, so the guard only confirms the state it read.
59        let claim = ClaimedRegistration::left_in_place(row);
60        // One row's database error must not roll back the whole claim, which would leave the
61        // same row at the head of the next one.
62        let mut savepoint = conn.begin().await?;
63        match write_unasked_move(&mut savepoint, &claim, unasked).await {
64            Ok(applied) => {
65                savepoint.commit().await?;
66                match (submission, applied) {
67                    (Some(submission), Applied::Written { .. }) => {
68                        let mut registration = claim.into_registration();
69                        // The move just restamped it in the database; the answer's first verify
70                        // poll is timed from this send, not from an earlier one.
71                        registration.submitted_at = Some(Utc::now());
72                        prepared.send(
73                            Claimed {
74                                claim: ClaimedRegistration::moved_to(
75                                    registration,
76                                    CreditRegistrationState::Submitting,
77                                ),
78                                extra: (),
79                            },
80                            submission,
81                        )
82                    }
83                    (_, applied) => prepared.record_applied(claim.id(), applied),
84                }
85            }
86            Err(error) => {
87                savepoint.rollback().await?;
88                hold_back(&ctx.errors, conn, claim.registration(), &error).await?;
89                prepared.record_failed();
90            }
91        }
92    }
93    Ok(prepared)
94}
95
96/// What the preflight made of one claimed row.
97enum Preflight {
98    /// Moved to `submitting`, and the submission goes in the batch.
99    Send(AttainmentSubmission),
100    /// Settled where the move says, without a request.
101    Settle(UnaskedMove),
102}
103
104fn preflight_submission(row: &CreditRegistration, already_registered: &[Uuid]) -> Preflight {
105    if already_registered.contains(&row.course_module_completion_id) {
106        debug!(
107            credit_registration_id = %row.id,
108            "Another registrar already registered this completion; marking duplicate"
109        );
110        return Preflight::Settle(duplicate_of_other_registrar());
111    }
112    match build_submission(row) {
113        Ok(submission) => Preflight::Send(submission),
114        Err(problem) => {
115            match &problem {
116                Unsendable::Incomplete => {}
117                Unsendable::UnknownGrade => {
118                    warn!(
119                        credit_registration_id = %row.id,
120                        "Credit registration's frozen grade is not one Sisu accepts; needs admin attention"
121                    );
122                }
123                Unsendable::Invalid(field) => {
124                    warn!(
125                        credit_registration_id = %row.id,
126                        field = *field,
127                        "Sisu does not accept a required field; needs admin attention"
128                    );
129                }
130            }
131            Preflight::Settle(problem.unasked_move())
132        }
133    }
134}
135
136/// Parks a row the preflight could not write, flagged for an admin, so the rest of the claim still
137/// goes out.
138async fn hold_back(
139    errors: &ErrorReporter<'_>,
140    conn: &mut PgConnection,
141    row: &CreditRegistration,
142    error: &CreditRegistrationError,
143) -> CreditRegistrationResult<()> {
144    error!(
145        credit_registration_id = %row.id,
146        error = ?error,
147        "Could not prepare credit registration for import; holding it back"
148    );
149    errors
150        .report(
151            &error.to_string(),
152            Some(format!("{error:?}")),
153            serde_json::json!({ "credit_registration_id": row.id }),
154        )
155        .await;
156    set_needs_admin_attention(conn, row.id, AdminAttention::Raise).await?;
157    schedule_next_attempt(conn, row.id, Utc::now() + SUBMIT_MAX_BACKOFF).await?;
158    credit_registration_events::insert(
159        conn,
160        &NewCreditRegistrationEvent {
161            message: Some(format!(
162                "Import could not prepare this row, so it was held back: {}",
163                scrub_text(&error.to_string())
164            )),
165            ..NewCreditRegistrationEvent::new(row.id, CreditRegistrationEventKind::RetryScheduled)
166        },
167    )
168    .await?;
169    Ok(())
170}
171
172/// A frozen snapshot that cannot be sent. Suotar validates every item before acting on any, so one
173/// it would refuse takes the rest of the batch down with it.
174enum Unsendable {
175    Incomplete,
176    UnknownGrade,
177    /// A field Suotar requires to be non-empty, or `credits`, which it requires to be finite.
178    Invalid(&'static str),
179}
180
181impl Unsendable {
182    fn unasked_move(&self) -> UnaskedMove {
183        match self {
184            Self::Incomplete => incomplete_payload(),
185            Self::UnknownGrade => unknown_grade(),
186            Self::Invalid(field) => invalid_payload_field(field),
187        }
188    }
189}
190
191fn required_field<'a>(field: &'static str, value: &'a str) -> Result<&'a str, Unsendable> {
192    let value = value.trim();
193    if value.is_empty() {
194        Err(Unsendable::Invalid(field))
195    } else {
196        Ok(value)
197    }
198}
199
200/// Builds the submission from the frozen snapshot, or says why the row cannot go into a batch.
201fn build_submission(row: &CreditRegistration) -> Result<AttainmentSubmission, Unsendable> {
202    let (
203        Some(student_number),
204        Some(course_code),
205        Some(enrolment_id),
206        Some(attained_at),
207        Some(attainment_language),
208        Some(grade_scale_id),
209        Some(grade_id),
210        Some(credits),
211    ) = (
212        row.student_number.as_ref().map(ExposeSecret::expose_secret),
213        row.uh_course_code.as_deref(),
214        row.selected_enrolment_id.as_deref(),
215        row.attained_at,
216        row.attainment_language.as_deref(),
217        row.grade_scale_id.as_deref(),
218        row.grade_id.as_deref(),
219        row.credits,
220    )
221    else {
222        return Err(Unsendable::Incomplete);
223    };
224    let student_number = required_field("studentNumber", student_number)?;
225    let course_code = CourseCode::parse(course_code).ok_or(Unsendable::Invalid("courseCode"))?;
226    let enrolment_id = required_field("enrolmentId", enrolment_id)?;
227    let attainment_language = required_field("attainmentLanguage", attainment_language)?;
228    let grade = MappedGrade {
229        grade_scale_id: required_field("gradeScaleId", grade_scale_id)?.to_string(),
230        grade_id: required_field("gradeId", grade_id)?.to_string(),
231    };
232    let credits = Credits::from_stored(credits).ok_or(Unsendable::Invalid("credits"))?;
233    // Suotar would refuse it as `invalidGradeForGradeScale`; refused here, the row fails on our
234    // mapping without a round trip.
235    if !is_known_grade(&grade) {
236        return Err(Unsendable::UnknownGrade);
237    }
238    Ok(AttainmentSubmission {
239        student_number: StudentNumber::new(student_number),
240        course_code,
241        enrolment_id: enrolment_id.to_string(),
242        attained_at,
243        attainment_language: attainment_language.to_string(),
244        grade,
245        credits,
246    })
247}
headless_lms_credit_registration/use_cases/import/claim.rs

headless_lms_credit_registration/use_cases/import/
claim.rs
