headless_lms_credit_registration/use_cases/enrolment_discovery/
mod.rs

1//! The `enrolment-discovery` phase: who the study registry says is on the course.
2//!
3//! Each iteration fetches the enrolment lists of the course codes that are due, those with people
4//! waiting first, in batches as large as the registry takes; every module on a code shares its
5//! list. One list wakes the registrations of people we already have a link for and, while account
6//! linking is switched on and `link-emails` is not paused, claims a linking email for everybody
7//! else, which the `link-emails` phase is made due at once to send. When to fetch a code is
8//! [`headless_lms_models::credit_registration_roster_schedules`].
9
10mod listing;
11mod reconcile;
12
13pub(crate) use reconcile::{list_unlinked, list_unlinked_enrolled_before};
14
15use headless_lms_models::course_module_suotar_configurations::ModuleToList;
16use headless_lms_models::credit_registration_phase_state;
17use headless_lms_models::credit_registration_roster_schedules::{
18    RosterSchedule, ScheduleSelection, ensure_rows, get_modules_by_code, get_schedules,
19};
20use headless_lms_utils::prelude::{DateTime, Utc};
21use sqlx::{PgConnection, PgPool};
22use uuid::Uuid;
23
24use crate::error::CreditRegistrationResult;
25use crate::phase::CreditRegistrationPhase;
26use crate::registry::{CourseCode, RegistryOperation, RosterCode, StudyRegistry};
27use crate::workflow::Counts;
28use headless_lms_models::credit_registrations::RegistrationScope;
29
30use listing::fetch_course_roster;
31
32/// One code of a listing request, with the modules that share its roster.
33struct CodeListing {
34    code: RosterCode,
35    modules: Vec<ModuleToList>,
36}
37
38/// With account linking off (`account_linking_since` `None`), a roster still wakes linked students'
39/// registrations, and only the mails are left out.
40pub(crate) async fn run<R: StudyRegistry>(
41    pool: &PgPool,
42    scope: &RegistrationScope,
43    account_linking_since: Option<DateTime<Utc>>,
44    registry: &mut R,
45) -> CreditRegistrationResult<Counts> {
46    // The limiter counts roster requests, so the limit is how many may go out.
47    let request_limit = registry.allowance(RegistryOperation::ListCourseRoster);
48    if request_limit == 0 {
49        return Ok(Counts::default());
50    }
51    let mut conn = pool.acquire().await?;
52    // Claims made during a pause would all go out at once on resume. Nobody counts as waiting
53    // meanwhile either, so rungs and presses are left for a fetch that can claim mail.
54    let mailing_since = if credit_registration_phase_state::is_paused(
55        &mut conn,
56        CreditRegistrationPhase::LinkEmails.as_str(),
57    )
58    .await?
59    {
60        None
61    } else {
62        account_linking_since
63    };
64    let due = load_due_roster_codes(&mut conn, scope.course_id, mailing_since).await?;
65    let planned = plan_roster_requests(due, request_limit, registry.roster_request_size());
66    let requests = load_listing_modules(&mut conn, scope.course_id, planned).await?;
67    drop(conn);
68
69    let mut counts = Counts::default();
70    let mut claimed_mail = false;
71    for request in requests {
72        let fetched = fetch_course_roster(pool, registry, &request, mailing_since).await?;
73        counts += fetched.counts;
74        claimed_mail |= fetched.claimed_mail;
75    }
76    // A scoped run writes nothing to the phase-state row.
77    if claimed_mail && scope.is_unscoped() {
78        let mut conn = pool.acquire().await?;
79        credit_registration_phase_state::run_now(
80            &mut conn,
81            CreditRegistrationPhase::LinkEmails.as_str(),
82        )
83        .await?;
84    }
85    Ok(counts)
86}
87
88/// The codes whose enrolment lists are due, those with people waiting first, then by when each
89/// fell due.
90async fn load_due_roster_codes(
91    conn: &mut PgConnection,
92    course_id: Option<Uuid>,
93    account_linking_since: Option<DateTime<Utc>>,
94) -> CreditRegistrationResult<Vec<RosterCode>> {
95    let now = Utc::now();
96    ensure_rows(conn, course_id).await?;
97    let mut due: Vec<RosterSchedule> = get_schedules(
98        conn,
99        course_id,
100        ScheduleSelection::DueCandidates,
101        account_linking_since,
102    )
103    .await?
104    .into_iter()
105    .filter(|schedule| schedule.is_due(now))
106    .collect();
107    due.sort_by_key(|schedule| (schedule.waiting_count == 0, schedule.next_fetch_at(now)));
108    Ok(due
109        .into_iter()
110        .filter_map(|schedule| {
111            Some(RosterCode {
112                course_code: CourseCode::parse(&schedule.course_code)?,
113                is_fetched_alone: schedule.is_fetched_alone,
114            })
115        })
116        .collect())
117}
118
119/// Groups due codes into requests in the order given: a code fetched alone in one of its own, the
120/// rest in batches of `request_size`. Keeps the first `request_limit`.
121fn plan_roster_requests(
122    due: Vec<RosterCode>,
123    request_limit: usize,
124    request_size: usize,
125) -> Vec<Vec<RosterCode>> {
126    let mut requests: Vec<Vec<RosterCode>> = Vec::new();
127    let mut open_batch: Option<usize> = None;
128    for code in due {
129        if code.is_fetched_alone {
130            requests.push(vec![code]);
131            continue;
132        }
133        match open_batch {
134            Some(index) if requests[index].len() < request_size => requests[index].push(code),
135            _ => {
136                open_batch = Some(requests.len());
137                requests.push(vec![code]);
138            }
139        }
140    }
141    requests.truncate(request_limit);
142    requests
143}
144
145/// Each planned request's codes, with the modules that share each code's roster.
146async fn load_listing_modules(
147    conn: &mut PgConnection,
148    course_id: Option<Uuid>,
149    planned: Vec<Vec<RosterCode>>,
150) -> CreditRegistrationResult<Vec<Vec<CodeListing>>> {
151    let codes: Vec<String> = planned
152        .iter()
153        .flatten()
154        .map(|code| code.course_code.as_str().to_string())
155        .collect();
156    let mut modules_by_code = get_modules_by_code(conn, course_id, &codes).await?;
157    Ok(planned
158        .into_iter()
159        .map(|request| {
160            request
161                .into_iter()
162                .map(|code| CodeListing {
163                    modules: modules_by_code
164                        .remove(code.course_code.as_str())
165                        .unwrap_or_default(),
166                    code,
167                })
168                .collect()
169        })
170        .collect())
171}
172
173#[cfg(test)]
174mod tests {
175    use super::*;
176
177    fn roster_codes(alone: &[&str], batched: usize) -> Vec<RosterCode> {
178        let alone = alone.iter().map(|alone_code| RosterCode {
179            course_code: CourseCode::parse(alone_code).expect("a code"),
180            is_fetched_alone: true,
181        });
182        let batched = (0..batched).map(|index| RosterCode {
183            course_code: CourseCode::parse(&format!("B{index}")).expect("a code"),
184            is_fetched_alone: false,
185        });
186        alone.chain(batched).collect()
187    }
188
189    fn planned_sizes(requests: &[Vec<RosterCode>]) -> Vec<usize> {
190        requests.iter().map(Vec::len).collect()
191    }
192
193    #[test]
194    fn roster_codes_fetched_alone_go_in_requests_of_their_own_and_the_rest_in_full_batches() {
195        let due = roster_codes(&["A1", "A2"], 51);
196        let requests = plan_roster_requests(due, usize::MAX, 50);
197        assert_eq!(planned_sizes(&requests), [1, 1, 50, 1]);
198        assert_eq!(requests[1][0].course_code.as_str(), "A2");
199        assert_eq!(requests[3][0].course_code.as_str(), "B50");
200    }
201
202    #[test]
203    fn roster_planning_keeps_the_first_requests_up_to_the_limit() {
204        let requests = plan_roster_requests(roster_codes(&["A1"], 51), 2, 50);
205        assert_eq!(planned_sizes(&requests), [1, 50]);
206    }
207
208    #[test]
209    fn a_roster_probe_sends_one_code() {
210        let requests = plan_roster_requests(roster_codes(&[], 60), 1, 1);
211        assert_eq!(planned_sizes(&requests), [1]);
212    }
213}
headless_lms_credit_registration/use_cases/enrolment_discovery/mod.rs

headless_lms_credit_registration/use_cases/enrolment_discovery/
mod.rs
