headless_lms_credit_registration/use_cases/enrolment_discovery/
reconcile.rs

1//! What one code's roster does to the modules on it: waking linked students' registrations and
2//! claiming linking mails for everybody else.
3
4use std::collections::{BTreeSet, HashMap, HashSet};
5
6use headless_lms_models::course_module_suotar_configurations::mark_listing_succeeded;
7use headless_lms_models::credit_registration_roster_schedules::{
8    AccountLinkingCodeCounters, record_linking_counters,
9};
10use headless_lms_models::library::credit_registration::account_linking::{
11    DiscoveredPerson, LinkingMailClaim, claim_linking_mails_batch, listed_addresses, mail_address,
12    registerable_completion_courses_by_account_email,
13};
14use headless_lms_models::library::credit_registration::enrolment_checks::{
15    RosterEnrolee, wake_for_roster_listing,
16};
17use headless_lms_models::library::credit_registration::study_registry::RosterPerson;
18use headless_lms_models::verified_student_numbers::{self, VerifiedStudentNumber};
19use headless_lms_utils::prelude::{DateTime, Utc};
20use headless_lms_utils::secret_string::expose_option;
21use secrecy::ExposeSecret;
22use sqlx::PgConnection;
23use uuid::Uuid;
24
25use super::CodeListing;
26use crate::error::CreditRegistrationResult;
27
28/// Wakes the linked students' registrations on every module of one code's roster and claims a
29/// linking mail for everybody else, once per course on the code: mails and links are per course.
30/// Returns the linking mails claimed, for the caller's roster-fetch summary.
31pub(super) async fn reconcile_roster(
32    conn: &mut PgConnection,
33    listing: &CodeListing,
34    people: &[RosterPerson],
35    account_linking_since: Option<DateTime<Utc>>,
36) -> CreditRegistrationResult<i32> {
37    let distinct = distinct_people(people);
38    let linked_rows = load_linked_accounts(conn, &distinct).await?;
39    let linked = LinkedAccounts::new(&linked_rows);
40    let enrolees = roster_enrolees(people, &linked);
41    if !enrolees.is_empty() {
42        for module in &listing.modules {
43            wake_for_roster_listing(conn, module.course_module_id, &enrolees).await?;
44        }
45    }
46    let mut mailed_count = 0;
47    if let Some(since) = account_linking_since {
48        let course_ids: BTreeSet<Uuid> = listing
49            .modules
50            .iter()
51            .map(|module| module.course_id)
52            .collect();
53        let mut considered = enrolled_since(&distinct, since);
54        let unlinked_early = unlinked_enrolled_before(&distinct, &linked, since);
55        let late_completer_courses =
56            late_completer_courses(conn, &unlinked_early, &course_ids, since).await?;
57        considered.extend(unlinked_early.iter().filter(|person| {
58            late_completer_courses.contains_key(person.person_id.expose_secret())
59        }));
60        let counters = claim_linking_mails(
61            conn,
62            &considered,
63            &linked,
64            &course_ids,
65            &late_completer_courses,
66        )
67        .await?;
68        mailed_count = counters.mailed_count;
69        let unlinked_before = unlinked_early.len() - late_completer_courses.len();
70        record_linking_counters(
71            conn,
72            listing.code.course_code.as_str(),
73            &counters,
74            i32::try_from(unlinked_before).unwrap_or(i32::MAX),
75        )
76        .await?;
77    }
78    for module in &listing.modules {
79        mark_listing_succeeded(
80            conn,
81            module.course_module_id,
82            account_linking_since.is_some(),
83        )
84        .await?;
85    }
86    Ok(mailed_count)
87}
88
89/// A person enrolled on several realisations of the code is listed once per realisation; keeps the
90/// most recent enrolment of each, one with no enrolment time counting as the oldest.
91fn distinct_people(people: &[RosterPerson]) -> Vec<&RosterPerson> {
92    let mut kept: Vec<&RosterPerson> = Vec::new();
93    let mut index_by_person: HashMap<&str, usize> = HashMap::new();
94    for person in people {
95        match index_by_person.get(person.person_id.expose_secret()) {
96            Some(&index) => {
97                if person.enrolled_at() > kept[index].enrolled_at() {
98                    kept[index] = person;
99                }
100            }
101            None => {
102                index_by_person.insert(person.person_id.expose_secret(), kept.len());
103                kept.push(person);
104            }
105        }
106    }
107    kept
108}
109
110/// For each of `early` (unlinked, enrolled before `since`) whose Sisu primary or secondary address is
111/// the account email of someone with a registerable completion since `since`, keyed by Sisu person
112/// id, the courses that completion is on: their completion is why they are worth mailing now.
113async fn late_completer_courses(
114    conn: &mut PgConnection,
115    early: &[&RosterPerson],
116    course_ids: &BTreeSet<Uuid>,
117    since: DateTime<Utc>,
118) -> CreditRegistrationResult<HashMap<String, BTreeSet<Uuid>>> {
119    let addresses_by_person: Vec<(&RosterPerson, Vec<String>)> = early
120        .iter()
121        .map(|&person| {
122            let addresses = listed_addresses(person).map(str::to_lowercase).collect();
123            (person, addresses)
124        })
125        .collect();
126    let mut addresses: Vec<String> = addresses_by_person
127        .iter()
128        .flat_map(|(_, addresses)| addresses.iter().cloned())
129        .collect();
130    addresses.sort_unstable();
131    addresses.dedup();
132    if addresses.is_empty() {
133        return Ok(HashMap::new());
134    }
135    let course_ids: Vec<Uuid> = course_ids.iter().copied().collect();
136    let courses_by_email =
137        registerable_completion_courses_by_account_email(conn, &course_ids, since, &addresses)
138            .await?;
139    Ok(addresses_by_person
140        .into_iter()
141        .filter_map(|(person, addresses)| {
142            let courses: BTreeSet<Uuid> = addresses
143                .iter()
144                .filter_map(|address| courses_by_email.get(address))
145                .flatten()
146                .copied()
147                .collect();
148            (!courses.is_empty()).then(|| (person.person_id.expose_secret().to_owned(), courses))
149        })
150        .collect())
151}
152
153/// The people the linking mails consider by enrolment time: those enrolled at or after `since`.
154/// The rest are mailed only through [`late_completer_courses`].
155fn enrolled_since<'a>(people: &[&'a RosterPerson], since: DateTime<Utc>) -> Vec<&'a RosterPerson> {
156    people
157        .iter()
158        .copied()
159        .filter(|person| {
160            person
161                .enrolled_at()
162                .is_some_and(|enrolled_at| enrolled_at >= since)
163        })
164        .collect()
165}
166
167/// The people [`enrolled_since`] leaves out, enrolled before `since` or at no known time, whom no
168/// account is linked to.
169fn unlinked_enrolled_before<'a>(
170    people: &[&'a RosterPerson],
171    linked: &LinkedAccounts<'_>,
172    since: DateTime<Utc>,
173) -> Vec<&'a RosterPerson> {
174    people
175        .iter()
176        .copied()
177        .filter(|person| {
178            person
179                .enrolled_at()
180                .is_none_or(|enrolled_at| enrolled_at < since)
181                && !linked.is_linked(person)
182        })
183        .collect()
184}
185
186/// [`unlinked_enrolled_before`] over a whole roster, each person once with their latest enrolment.
187pub(crate) async fn list_unlinked_enrolled_before(
188    conn: &mut PgConnection,
189    people: &[RosterPerson],
190    since: DateTime<Utc>,
191) -> CreditRegistrationResult<Vec<RosterPerson>> {
192    let distinct = distinct_people(people);
193    let linked_rows = load_linked_accounts(conn, &distinct).await?;
194    let linked = LinkedAccounts::new(&linked_rows);
195    Ok(unlinked_enrolled_before(&distinct, &linked, since)
196        .into_iter()
197        .cloned()
198        .collect())
199}
200
201/// Everyone on a roster whom no account is linked to, whenever they enrolled, each once with their
202/// latest enrolment.
203pub(crate) async fn list_unlinked(
204    conn: &mut PgConnection,
205    people: &[RosterPerson],
206) -> CreditRegistrationResult<Vec<RosterPerson>> {
207    let distinct = distinct_people(people);
208    let linked_rows = load_linked_accounts(conn, &distinct).await?;
209    let linked = LinkedAccounts::new(&linked_rows);
210    Ok(distinct
211        .into_iter()
212        .filter(|person| !linked.is_linked(person))
213        .cloned()
214        .collect())
215}
216
217/// The accounts already linked to someone on the roster, by Sisu person id or student number.
218async fn load_linked_accounts(
219    conn: &mut PgConnection,
220    people: &[&RosterPerson],
221) -> CreditRegistrationResult<Vec<VerifiedStudentNumber>> {
222    let person_ids: Vec<String> = people
223        .iter()
224        .map(|person| person.person_id.expose_secret().to_owned())
225        .collect();
226    let student_numbers: Vec<String> = people
227        .iter()
228        .map(|person| person.student_number.expose_secret().to_owned())
229        .collect();
230    let mut linked = verified_student_numbers::get_by_sisu_person_ids(conn, &person_ids).await?;
231    // A study_registry link has no person id until resolve-person-ids reaches it.
232    let linked_ids: HashSet<_> = linked.iter().map(|row| row.id).collect();
233    let linked_by_number = verified_student_numbers::get_by_student_numbers(conn, &student_numbers)
234        .await?
235        .into_iter()
236        .filter(|row| !linked_ids.contains(&row.id))
237        .collect::<Vec<_>>();
238    linked.extend(linked_by_number);
239    Ok(linked)
240}
241
242/// The accounts linked to someone on one code's roster, indexed once for every module on it.
243struct LinkedAccounts<'a> {
244    rows: &'a [VerifiedStudentNumber],
245    person_ids: HashSet<&'a str>,
246    student_numbers: HashSet<&'a str>,
247}
248
249impl<'a> LinkedAccounts<'a> {
250    fn new(rows: &'a [VerifiedStudentNumber]) -> Self {
251        Self {
252            rows,
253            person_ids: rows
254                .iter()
255                .filter_map(|row| expose_option(&row.sisu_person_id))
256                .collect(),
257            student_numbers: rows
258                .iter()
259                .map(|row| row.student_number.expose_secret())
260                .collect(),
261        }
262    }
263
264    /// Whether some account is linked to `person`, by Sisu person id or student number.
265    fn is_linked(&self, person: &RosterPerson) -> bool {
266        self.person_ids.contains(person.person_id.expose_secret())
267            || self
268                .student_numbers
269                .contains(person.student_number.expose_secret())
270    }
271}
272
273/// The linked accounts on the roster, each with every enrolment id the registry lists them under.
274fn roster_enrolees(people: &[RosterPerson], linked: &LinkedAccounts<'_>) -> Vec<RosterEnrolee> {
275    let mut ids_by_person_id: HashMap<&str, Vec<String>> = HashMap::new();
276    let mut person_id_by_student_number: HashMap<&str, &str> = HashMap::new();
277    for person in people {
278        let person_id = person.person_id.expose_secret();
279        person_id_by_student_number.insert(person.student_number.expose_secret(), person_id);
280        let ids = ids_by_person_id.entry(person_id).or_default();
281        if let Some(id) = person
282            .enrolment
283            .as_ref()
284            .and_then(|enrolment| enrolment.id.clone())
285        {
286            ids.push(id);
287        }
288    }
289    linked
290        .rows
291        .iter()
292        .filter_map(|row| {
293            let person_id = expose_option(&row.sisu_person_id)
294                .filter(|person_id| ids_by_person_id.contains_key(person_id))
295                .or_else(|| {
296                    person_id_by_student_number
297                        .get(row.student_number.expose_secret())
298                        .copied()
299                })?;
300            Some(RosterEnrolee {
301                user_id: row.user_id,
302                enrolment_ids: ids_by_person_id.get(person_id).cloned().unwrap_or_default(),
303            })
304        })
305        .collect()
306}
307
308/// Claims a linking mail for everyone in `people` we hold no link for, on each of `course_ids` or,
309/// for those in `courses_of_late_completers`, only on the courses their completion is on. Counts
310/// each person once by the strongest claim any course got for them.
311async fn claim_linking_mails(
312    conn: &mut PgConnection,
313    people: &[&RosterPerson],
314    linked: &LinkedAccounts<'_>,
315    course_ids: &BTreeSet<Uuid>,
316    courses_of_late_completers: &HashMap<String, BTreeSet<Uuid>>,
317) -> CreditRegistrationResult<AccountLinkingCodeCounters> {
318    let (mut counters, candidates) = linking_candidates(people, linked);
319    let mut strongest = vec![LinkingMailClaim::NoAddress; candidates.len()];
320    for &course_id in course_ids {
321        let on_course: Vec<usize> = (0..candidates.len())
322            .filter(|&index| {
323                courses_of_late_completers
324                    .get(candidates[index].person_id.expose_secret())
325                    .is_none_or(|courses| courses.contains(&course_id))
326            })
327            .collect();
328        if on_course.is_empty() {
329            continue;
330        }
331        let discovered: Vec<DiscoveredPerson> = on_course
332            .iter()
333            .map(|&index| DiscoveredPerson::listed(candidates[index], course_id))
334            .collect();
335        let claims = claim_linking_mails_batch(conn, &discovered).await?;
336        for (&index, claim) in on_course.iter().zip(claims) {
337            strongest[index] = strongest[index].max(claim);
338        }
339    }
340    for claim in strongest {
341        match claim {
342            LinkingMailClaim::Claimed => counters.mailed_count += 1,
343            LinkingMailClaim::SuppressedByDedup => counters.suppressed_by_dedup_count += 1,
344            LinkingMailClaim::SuppressedByRateCap => counters.suppressed_by_rate_cap_count += 1,
345            LinkingMailClaim::NoAddress => {}
346        }
347    }
348    Ok(counters)
349}
350
351/// The people on a roster a linking mail may go to, and the counters of those it may not: already
352/// linked, or with no address to mail.
353fn linking_candidates<'a>(
354    people: &[&'a RosterPerson],
355    linked: &LinkedAccounts<'_>,
356) -> (AccountLinkingCodeCounters, Vec<&'a RosterPerson>) {
357    let mut counters = AccountLinkingCodeCounters {
358        listed_person_count: i32::try_from(people.len()).unwrap_or(i32::MAX),
359        ..AccountLinkingCodeCounters::default()
360    };
361    let mut candidates = Vec::new();
362    for &person in people {
363        if linked.is_linked(person) {
364            counters.already_linked_count += 1;
365        } else if mail_address(person).is_none() {
366            // The only genuinely unreachable population, and the reason it has a counter of its own.
367            counters.no_address_count += 1;
368        } else {
369            candidates.push(person);
370        }
371    }
372    (counters, candidates)
373}
374
375#[cfg(test)]
376mod tests {
377    use chrono::{TimeDelta, Utc};
378    use headless_lms_models::library::credit_registration::study_registry::RosterEnrolment;
379    use headless_lms_models::secret::DbSecret;
380    use headless_lms_models::verified_student_numbers::StudentNumberVerificationMethod;
381    use secrecy::SecretString;
382    use uuid::Uuid;
383
384    use super::*;
385
386    fn person(person_id: &str, student_number: &str, email: Option<&str>) -> RosterPerson {
387        RosterPerson {
388            student_number: SecretString::from(student_number),
389            person_id: SecretString::from(person_id),
390            first_names: None,
391            last_name: None,
392            primary_email: email.map(SecretString::from),
393            secondary_email: None,
394            enrolment: None,
395        }
396    }
397
398    fn enrolled(person: RosterPerson, enrolment_id: &str, days_ago: i64) -> RosterPerson {
399        RosterPerson {
400            enrolment: Some(RosterEnrolment {
401                id: Some(enrolment_id.to_string()),
402                course_unit_realisation_id: None,
403                state: None,
404                enrolment_date_time: Some(Utc::now() - TimeDelta::days(days_ago)),
405            }),
406            ..person
407        }
408    }
409
410    fn link(student_number: &str, person_id: Option<&str>) -> VerifiedStudentNumber {
411        let now = Utc::now();
412        VerifiedStudentNumber {
413            id: Uuid::new_v4(),
414            created_at: now,
415            updated_at: now,
416            deleted_at: None,
417            user_id: Uuid::new_v4(),
418            student_number: DbSecret::new(student_number),
419            sisu_person_id: person_id.map(DbSecret::new),
420            first_names: None,
421            last_name: None,
422            verified_at: now,
423            verified_via: StudentNumberVerificationMethod::EmailedLink,
424            verified_via_email: None,
425            linked_by_user_id: None,
426            link_reason: None,
427            verified_from_course_id: None,
428        }
429    }
430
431    #[test]
432    fn a_person_on_several_realisations_is_kept_once_with_the_latest_enrolment() {
433        let people = [
434            enrolled(person("p1", "1", None), "old", 10),
435            person("p2", "2", None),
436            enrolled(person("p1", "1", None), "new", 1),
437            enrolled(person("p1", "1", None), "older", 20),
438        ];
439        let kept: Vec<_> = distinct_people(&people)
440            .iter()
441            .map(|person| {
442                (
443                    person.person_id.expose_secret().to_string(),
444                    person
445                        .enrolment
446                        .as_ref()
447                        .and_then(|enrolment| enrolment.id.clone()),
448                )
449            })
450            .collect();
451        assert_eq!(
452            kept,
453            [
454                ("p1".to_string(), Some("new".to_string())),
455                ("p2".to_string(), None)
456            ]
457        );
458    }
459
460    #[test]
461    fn a_linked_account_wakes_with_every_enrolment_it_is_listed_under() {
462        let people = [
463            enrolled(person("p1", "1", None), "e1", 2),
464            enrolled(person("p1", "1", None), "e2", 1),
465            enrolled(person("p2", "2", None), "e3", 1),
466        ];
467        let by_person_id = link("1", Some("p1"));
468        let by_number_only = link("2", None);
469        let unlisted = link("9", Some("p9"));
470        let links = [by_person_id.clone(), by_number_only.clone(), unlisted];
471        let enrolees = roster_enrolees(&people, &LinkedAccounts::new(&links));
472        let woken: Vec<_> = enrolees
473            .iter()
474            .map(|enrolee| (enrolee.user_id, enrolee.enrolment_ids.clone()))
475            .collect();
476        assert_eq!(
477            woken,
478            [
479                (
480                    by_person_id.user_id,
481                    vec!["e1".to_string(), "e2".to_string()]
482                ),
483                (by_number_only.user_id, vec!["e3".to_string()]),
484            ]
485        );
486    }
487
488    #[test]
489    fn only_unlinked_people_enrolled_since_linking_and_with_an_address_are_mailed() {
490        let people = [
491            enrolled(person("p1", "1", Some("linked-by-id@example.com")), "e1", 1),
492            enrolled(
493                person("p2", "2", Some("linked-by-number@example.com")),
494                "e2",
495                1,
496            ),
497            enrolled(person("p3", "3", None), "e3", 1),
498            enrolled(person("p4", "4", Some("  ")), "e4", 1),
499            enrolled(person("p5", "5", Some("new@example.com")), "e5", 1),
500            enrolled(person("p6", "6", Some("early@example.com")), "e6", 30),
501            person("p7", "7", Some("unknown-date@example.com")),
502        ];
503        let listed: Vec<&RosterPerson> = people.iter().collect();
504        let links = [link("other", Some("p1")), link("2", None)];
505        let considered = enrolled_since(&listed, Utc::now() - TimeDelta::days(10));
506        let (counters, candidates) = linking_candidates(&considered, &LinkedAccounts::new(&links));
507        assert_eq!(
508            counters,
509            AccountLinkingCodeCounters {
510                listed_person_count: 5,
511                already_linked_count: 2,
512                no_address_count: 2,
513                ..AccountLinkingCodeCounters::default()
514            }
515        );
516        assert_eq!(candidates.len(), 1);
517        assert_eq!(candidates[0].person_id.expose_secret(), "p5");
518    }
519}
headless_lms_credit_registration/use_cases/enrolment_discovery/reconcile.rs

headless_lms_credit_registration/use_cases/enrolment_discovery/
reconcile.rs
