headless_lms_credit_registration/use_cases/verify/
poll.rs

1//! Polling what became of the submissions that have an attainment id to poll by.
2
3use chrono::{DateTime, Utc};
4use headless_lms_models::credit_registrations::{
5    AdminAttention, VerifyFlow, mark_partially_registered, reset_for_resubmission,
6    set_needs_admin_attention,
7};
8use headless_lms_models::library::credit_registration::outcomes::{
9    Outcome, verify_inconclusive_outcome,
10};
11use sqlx::{Connection, PgConnection};
12
13use super::decide::{
14    PollAnswer, decide_poll, not_registered_decision, partially_registered_decision,
15};
16use super::lease::{Leased, VerifyAttempt, attempt_facts, claim_and_lease};
17use crate::error::CreditRegistrationResult;
18use crate::registry::{AttainmentId, ExchangeAudit, VerificationAnswer, VerificationRequest};
19use crate::use_cases::batch_flow::{BatchFlowContext, Prepared, RegistryBatchFlow};
20use crate::workflow::{
21    Applied, RefusalPolicy, write_decision, write_decision_committing_if_written,
22};
23
24const STUCK_WITHOUT_ATTAINMENT_ID_MESSAGE: &str =
25    "Credit registration is awaiting verification with no submitted attainment id";
26
27/// Polls the rows that have something to poll by.
28pub(super) struct VerifyPoll;
29
30impl RegistryBatchFlow for VerifyPoll {
31    type Extra = VerifyAttempt;
32    type Request = VerificationRequest;
33
34    const ALL_UNAVAILABLE_ERROR: &'static str = "Every verify poll came back unavailable.";
35    /// Deliberately not the shared request-level outcome: a failure to ask proves nothing was or
36    /// was not created, and moving the row towards `failed_retryable` would let an admin resubmit
37    /// it. The iteration still reports the refusal, and the gate still records it.
38    const REFUSAL: RefusalPolicy<VerifyAttempt> = RefusalPolicy::KeepWaiting {
39        outcome: still_polling,
40        message: "Could not verify this submission this time.",
41    };
42
43    /// A row stuck without a submitted attainment id is leased like the rest, so it is not claimed
44    /// again every iteration, but never sent; it is flagged for an admin and counted as failed.
45    async fn claim(
46        ctx: &BatchFlowContext<'_>,
47        conn: &mut PgConnection,
48        limit: usize,
49    ) -> CreditRegistrationResult<Prepared<VerifyAttempt, VerificationRequest>> {
50        let mut prepared = Prepared::new();
51        for poll in claim_and_lease(ctx, conn, VerifyFlow::Poll, limit).await? {
52            let row = poll.claim.registration();
53            let Some(submitted_attainment_id) = row.submitted_attainment_id.clone() else {
54                // Reported on every lease, which backs off as the submission ages; the admin flag
55                // can't mark it as reported, since an earlier move may already have raised it.
56                error!(
57                    credit_registration_id = %row.id,
58                    "Credit registration is awaiting verification with no submitted attainment id; stuck"
59                );
60                ctx.errors
61                    .report(
62                        STUCK_WITHOUT_ATTAINMENT_ID_MESSAGE,
63                        None,
64                        serde_json::json!({ "credit_registration_id": row.id }),
65                    )
66                    .await;
67                if !row.needs_admin_attention {
68                    set_needs_admin_attention(conn, row.id, AdminAttention::Raise).await?;
69                }
70                prepared.record_failed();
71                continue;
72            };
73            let request = VerificationRequest {
74                submitted_attainment_id: AttainmentId::new(submitted_attainment_id),
75            };
76            prepared.send(poll, request);
77        }
78        Ok(prepared)
79    }
80
81    async fn apply_answer(
82        conn: &mut PgConnection,
83        poll: &Leased,
84        answer: Option<&VerificationAnswer>,
85        audit: &ExchangeAudit,
86    ) -> CreditRegistrationResult<Applied> {
87        let claim = &poll.claim;
88        let facts = attempt_facts(poll, Utc::now());
89        let row_error = answer.and_then(|answer| answer.error_message.as_deref());
90        match decide_poll(claim.registration().state, answer, &facts) {
91            PollAnswer::Decided(decision) => {
92                write_decision(conn, claim, (*decision).with_row_error(row_error), audit).await
93            }
94            // Kept even if the row moved on: when a poll first saw the partial registration is a
95            // fact about Sisu, not about this move.
96            PollAnswer::PartiallyRegistered => {
97                let partially_registered_at = mark_partially_registered(conn, claim.id()).await?;
98                let decision = partially_registered_decision(&facts, partially_registered_at)
99                    .with_row_error(row_error);
100                write_decision(conn, claim, decision, audit).await
101            }
102            // The reset rolls back with a row that moved on.
103            PollAnswer::NotRegistered => {
104                let mut tx = conn.begin().await?;
105                let reimport_count = reset_for_resubmission(&mut tx, claim.id()).await?;
106                let decision =
107                    not_registered_decision(&facts, reimport_count).with_row_error(row_error);
108                write_decision_committing_if_written(tx, claim, decision, audit).await
109            }
110        }
111    }
112}
113
114fn still_polling(poll: &Leased, now: DateTime<Utc>) -> Outcome {
115    verify_inconclusive_outcome(poll.claim.registration().state, &attempt_facts(poll, now))
116}
headless_lms_credit_registration/use_cases/verify/poll.rs

headless_lms_credit_registration/use_cases/verify/
poll.rs
