headless_lms_server/controllers/mock_suotar/
store.rs

1//! The only Redis-aware part of the mock: key layout, generations, the per-request working set,
2//! the write-back pipeline and the call log.
3//!
4//! Its own connection rather than the cache wrapper, whose failures are silent no-ops: a component
5//! tests assert against has to fail loudly.
6
7use std::collections::{BTreeMap, HashMap};
8use std::sync::RwLock;
9
10use anyhow::{Context, anyhow};
11use itertools::Itertools;
12use redis::{AsyncCommands, aio::ConnectionManager};
13use serde::de::DeserializeOwned;
14use tokio::sync::{Mutex, OnceCell};
15
16use crate::prelude::*;
17
18use super::faults::Fault;
19use super::world::{
20    CourseCode, MockAttainment, MockCourseUnit, MockEnrolment, MockPerson, MockProductAccessToken,
21    MockSubmission, RecordedCall, StudentNumber, WorkingSet, WorldDefaults, WorldWrite,
22    person_course_key,
23};
24
25const GENERATION_KEY: &str = "ms:generation";
26const GENERATION_SEQ_KEY: &str = "ms:seq:generation";
27
28const META: &str = "meta";
29const PERSONS: &str = "persons";
30const COURSE_UNITS: &str = "courseUnits";
31const ENROLMENTS: &str = "enrolments";
32const ATTAINMENTS: &str = "attainments";
33const SUBMISSIONS: &str = "submissions";
34const PRODUCT_TOKENS: &str = "productTokens";
35const IDX_ENROLMENTS_BY_PERSON: &str = "idx:enrolmentsByPerson";
36const IDX_ENROLMENTS_BY_REALISATION: &str = "idx:enrolmentsByRealisation";
37const IDX_ATTAINMENTS_BY_PERSON_COURSE: &str = "idx:attainmentsByPersonCourse";
38const IDX_SUBMISSIONS_BY_PERSON_COURSE: &str = "idx:submissionsByPersonCourse";
39const IDX_OWNER_KEYS: &str = "idx:ownerKeys";
40const FAULTS: &str = "faults";
41const FAULTS_REMAINING: &str = "faults:remaining";
42const CALLS: &str = "calls";
43const SEQ_CALL: &str = "seq:call";
44const SEQ_PERSON: &str = "seq:person";
45const SEQ_FAULT: &str = "seq:fault";
46
47/// Closed by design: cleaning up a superseded generation is one `DEL` over these names, never a
48/// keyspace scan.
49const PREFIXED_KEYS: [&str; 18] = [
50    META,
51    PERSONS,
52    COURSE_UNITS,
53    ENROLMENTS,
54    ATTAINMENTS,
55    SUBMISSIONS,
56    PRODUCT_TOKENS,
57    IDX_ENROLMENTS_BY_PERSON,
58    IDX_ENROLMENTS_BY_REALISATION,
59    IDX_ATTAINMENTS_BY_PERSON_COURSE,
60    IDX_SUBMISSIONS_BY_PERSON_COURSE,
61    IDX_OWNER_KEYS,
62    FAULTS,
63    FAULTS_REMAINING,
64    CALLS,
65    SEQ_CALL,
66    SEQ_PERSON,
67    SEQ_FAULT,
68];
69
70/// What one `HMGET` answers with: a slot per requested field, empty where the field is absent.
71type Fields = Vec<Option<String>>;
72
73/// A whole world, as installed under one generation. Indexes are derived from the entities rather
74/// than part of it, so a caller cannot desynchronise them.
75#[derive(Debug, Clone, Default)]
76pub struct World {
77    pub defaults: WorldDefaults,
78    pub persons: BTreeMap<StudentNumber, MockPerson>,
79    pub course_units: BTreeMap<CourseCode, MockCourseUnit>,
80    pub enrolments: BTreeMap<String, MockEnrolment>,
81    pub attainments: BTreeMap<String, MockAttainment>,
82    pub submissions: BTreeMap<String, MockSubmission>,
83    pub product_tokens: BTreeMap<String, MockProductAccessToken>,
84}
85
86#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
87#[serde(rename_all = "camelCase")]
88pub struct OwnerKeys {
89    pub student_numbers: Vec<String>,
90    pub course_codes: Vec<String>,
91    pub product_ids: Vec<String>,
92}
93
94/// Everything a request needs before it may look at the body.
95#[derive(Debug, Clone)]
96pub struct Preamble {
97    pub generation: String,
98    pub defaults: WorldDefaults,
99    /// False when the index holds no world under this generation, which is what an external flush
100    /// looks like.
101    pub defaults_present: bool,
102    pub db_generation: Option<String>,
103    /// In arm order, which is precedence.
104    pub faults: Vec<Fault>,
105    /// A hint that saves a draw on a long-spent fault. Never the decision: the draw at the match is.
106    pub remaining: HashMap<String, i64>,
107}
108
109#[derive(Debug, Clone, Default, Serialize, Deserialize)]
110#[serde(rename_all = "camelCase")]
111pub struct WorldCounts {
112    pub persons: usize,
113    pub course_units: usize,
114    pub enrolments: usize,
115    pub attainments: usize,
116    pub submissions: usize,
117    pub product_tokens: usize,
118    pub faults_armed: usize,
119    pub faults_spent: usize,
120    pub call_log_len: usize,
121}
122
123pub struct MockSuotarStore {
124    client: redis::Client,
125    connection: OnceCell<ConnectionManager>,
126    /// Cached because one server process owns the index; re-read only when a prefixed read comes back
127    /// empty.
128    generation: RwLock<Option<String>>,
129    install_lock: Mutex<()>,
130}
131
132impl std::fmt::Debug for MockSuotarStore {
133    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
134        f.debug_struct("MockSuotarStore").finish()
135    }
136}
137
138impl MockSuotarStore {
139    /// Swaps in the mock's own database index so a flush touches nothing of the cache's. Connects on
140    /// first use, so an unreachable Redis is a per-request error rather than a cached success.
141    pub fn new(redis_url: &str, db_index: i64) -> anyhow::Result<Self> {
142        Ok(Self {
143            client: redis::Client::open(database_url(redis_url, db_index)?)
144                .context("failed to build the mock Suotar Redis client")?,
145            connection: OnceCell::new(),
146            generation: RwLock::new(None),
147            install_lock: Mutex::new(()),
148        })
149    }
150
151    async fn conn(&self) -> anyhow::Result<ConnectionManager> {
152        let manager = self
153            .connection
154            .get_or_try_init(|| async { ConnectionManager::new(self.client.clone()).await })
155            .await
156            .context("the mock Suotar could not reach Redis")?;
157        Ok(manager.clone())
158    }
159
160    fn cached_generation(&self) -> Option<String> {
161        self.generation.read().ok().and_then(|g| g.clone())
162    }
163
164    fn cache_generation(&self, generation: Option<String>) {
165        if let Ok(mut cached) = self.generation.write() {
166            *cached = generation;
167        }
168    }
169
170    pub async fn live_generation(&self) -> anyhow::Result<Option<String>> {
171        if let Some(generation) = self.cached_generation() {
172            return Ok(Some(generation));
173        }
174        let mut conn = self.conn().await?;
175        let generation: Option<String> = conn.get(GENERATION_KEY).await?;
176        self.cache_generation(generation.clone());
177        Ok(generation)
178    }
179
180    /// Flips the generation pointer last, so no request sees a half-installed world and a push needs
181    /// nothing cleared before it.
182    pub async fn install_world(
183        &self,
184        world: &World,
185        db_generation: Option<&str>,
186    ) -> anyhow::Result<String> {
187        let previous = self.live_generation().await?;
188        let mut conn = self.conn().await?;
189        let sequence: i64 = conn.incr(GENERATION_SEQ_KEY, 1).await?;
190        let generation = format!("g{sequence}");
191
192        let mut pipe = redis::pipe();
193        pipe.atomic();
194        pipe.hset(
195            key(&generation, META),
196            "defaults",
197            serde_json::to_string(&world.defaults)?,
198        )
199        .ignore();
200        pipe.hset(
201            key(&generation, META),
202            "installedAt",
203            Utc::now().to_rfc3339(),
204        )
205        .ignore();
206        if let Some(db_generation) = db_generation {
207            pipe.hset(key(&generation, META), "dbGeneration", db_generation)
208                .ignore();
209        }
210        write_entity_hash(&mut pipe, &generation, PERSONS, &world.persons)?;
211        write_entity_hash(&mut pipe, &generation, COURSE_UNITS, &world.course_units)?;
212        write_entity_hash(&mut pipe, &generation, ENROLMENTS, &world.enrolments)?;
213        write_entity_hash(&mut pipe, &generation, ATTAINMENTS, &world.attainments)?;
214        write_entity_hash(&mut pipe, &generation, SUBMISSIONS, &world.submissions)?;
215        write_entity_hash(
216            &mut pipe,
217            &generation,
218            PRODUCT_TOKENS,
219            &world.product_tokens,
220        )?;
221        write_derived_indexes(&mut pipe, &generation, world)?;
222        pipe.set(GENERATION_KEY, &generation).ignore();
223        pipe.query_async::<()>(&mut conn).await?;
224
225        self.cache_generation(Some(generation.clone()));
226
227        if let Some(previous) = previous.filter(|previous| previous != &generation) {
228            let mut cleanup = redis::pipe();
229            for name in PREFIXED_KEYS {
230                cleanup.del(key(&previous, name)).ignore();
231            }
232            cleanup.query_async::<()>(&mut conn).await?;
233        }
234        Ok(generation)
235    }
236
237    /// Serialised so a burst of first requests against an empty index does not each mint a generation.
238    pub async fn install_if_absent(
239        &self,
240        world: &World,
241        db_generation: Option<&str>,
242    ) -> anyhow::Result<String> {
243        let _guard = self.install_lock.lock().await;
244        self.cache_generation(None);
245        if let Some(generation) = self.live_generation().await?
246            && self.has_world(&generation).await?
247        {
248            return Ok(generation);
249        }
250        self.install_world(world, db_generation).await
251    }
252
253    async fn has_world(&self, generation: &str) -> anyhow::Result<bool> {
254        let mut conn = self.conn().await?;
255        let present: bool = conn.hexists(key(generation, META), "defaults").await?;
256        Ok(present)
257    }
258
259    /// Safe because the index is the mock's alone; the next contract request builds the world lazily.
260    pub async fn flush(&self) -> anyhow::Result<()> {
261        let mut conn = self.conn().await?;
262        redis::cmd("FLUSHDB").query_async::<()>(&mut conn).await?;
263        self.cache_generation(None);
264        Ok(())
265    }
266
267    pub async fn preamble(&self, generation: &str) -> anyhow::Result<Preamble> {
268        let mut conn = self.conn().await?;
269        let (meta, faults, remaining): (
270            HashMap<String, String>,
271            HashMap<String, String>,
272            HashMap<String, i64>,
273        ) = redis::pipe()
274            .hgetall(key(generation, META))
275            .hgetall(key(generation, FAULTS))
276            .hgetall(key(generation, FAULTS_REMAINING))
277            .query_async(&mut conn)
278            .await?;
279
280        let defaults = match meta.get("defaults") {
281            Some(raw) => {
282                serde_json::from_str(raw).context("stored world defaults are unreadable")?
283            }
284            None => WorldDefaults::default(),
285        };
286        let mut faults: Vec<Fault> = faults
287            .values()
288            .map(|raw| serde_json::from_str::<Fault>(raw))
289            .collect::<Result<_, _>>()
290            .context("a stored fault is unreadable")?;
291        faults.sort_by_key(|fault| fault.seq);
292
293        Ok(Preamble {
294            generation: generation.to_string(),
295            defaults_present: meta.contains_key("defaults"),
296            defaults,
297            db_generation: meta.get("dbGeneration").cloned(),
298            faults,
299            remaining,
300        })
301    }
302
303    pub async fn load_persons(
304        &self,
305        generation: &str,
306        student_numbers: &[String],
307    ) -> anyhow::Result<BTreeMap<String, MockPerson>> {
308        let mut conn = self.conn().await?;
309        hmget_json(&mut conn, &key(generation, PERSONS), student_numbers).await
310    }
311
312    pub async fn load_product_tokens(
313        &self,
314        generation: &str,
315        product_ids: &[String],
316    ) -> anyhow::Result<BTreeMap<String, MockProductAccessToken>> {
317        let mut conn = self.conn().await?;
318        hmget_json(&mut conn, &key(generation, PRODUCT_TOKENS), product_ids).await
319    }
320
321    /// Two pipelined round trips whatever the batch size: the keyed hashes, then the entities they
322    /// point at.
323    pub async fn load_for_person_course(
324        &self,
325        generation: &str,
326        student_numbers: &[String],
327        course_codes: &[String],
328    ) -> anyhow::Result<WorkingSet> {
329        let mut conn = self.conn().await?;
330        let person_course_keys: Vec<String> = student_numbers
331            .iter()
332            .flat_map(|student_number| {
333                course_codes
334                    .iter()
335                    .map(move |course_code| person_course_key(student_number, course_code))
336            })
337            .collect();
338
339        let mut first = redis::pipe();
340        push_hmget(&mut first, &key(generation, PERSONS), student_numbers);
341        push_hmget(&mut first, &key(generation, COURSE_UNITS), course_codes);
342        push_hmget(
343            &mut first,
344            &key(generation, IDX_ENROLMENTS_BY_PERSON),
345            student_numbers,
346        );
347        push_hmget(
348            &mut first,
349            &key(generation, IDX_ATTAINMENTS_BY_PERSON_COURSE),
350            &person_course_keys,
351        );
352        push_hmget(
353            &mut first,
354            &key(generation, IDX_SUBMISSIONS_BY_PERSON_COURSE),
355            &person_course_keys,
356        );
357        let (persons, course_units, enrolments_by_person, attainment_ids, submission_ids): (
358            Fields,
359            Fields,
360            Fields,
361            Fields,
362            Fields,
363        ) = first.query_async(&mut conn).await?;
364
365        let mut working = WorkingSet {
366            persons: zip_json(student_numbers, persons)?,
367            course_units: zip_json(course_codes, course_units)?,
368            enrolments_by_person: zip_json(student_numbers, enrolments_by_person)?,
369            attainments_by_person_course: zip_json(&person_course_keys, attainment_ids)?,
370            submissions_by_person_course: zip_json(&person_course_keys, submission_ids)?,
371            ..Default::default()
372        };
373
374        let enrolment_ids = flatten(working.enrolments_by_person.values());
375        let attainment_ids = flatten(working.attainments_by_person_course.values());
376        let submission_ids = flatten(working.submissions_by_person_course.values());
377
378        let mut second = redis::pipe();
379        push_hmget(&mut second, &key(generation, ENROLMENTS), &enrolment_ids);
380        push_hmget(&mut second, &key(generation, ATTAINMENTS), &attainment_ids);
381        push_hmget(&mut second, &key(generation, SUBMISSIONS), &submission_ids);
382        let (enrolments, attainments, submissions): (Fields, Fields, Fields) =
383            second.query_async(&mut conn).await?;
384
385        working.enrolments = zip_json(&enrolment_ids, enrolments)?;
386        working.attainments = zip_json(&attainment_ids, attainments)?;
387        working.submissions = zip_json(&submission_ids, submissions)?;
388        Ok(working)
389    }
390
391    /// Verify's body carries only submitted attainment ids, so the persons behind them come second.
392    pub async fn load_for_verify(
393        &self,
394        generation: &str,
395        submitted_attainment_ids: &[String],
396    ) -> anyhow::Result<WorkingSet> {
397        let mut conn = self.conn().await?;
398        let submissions: BTreeMap<String, MockSubmission> = hmget_json(
399            &mut conn,
400            &key(generation, SUBMISSIONS),
401            submitted_attainment_ids,
402        )
403        .await?;
404        let student_numbers: Vec<String> = submissions
405            .values()
406            .map(|submission| submission.student_number.clone())
407            .unique()
408            .collect();
409        let person_course_keys: Vec<String> = submissions
410            .values()
411            .map(|submission| {
412                person_course_key(&submission.student_number, &submission.course_code)
413            })
414            .unique()
415            .collect();
416        let persons = hmget_json(&mut conn, &key(generation, PERSONS), &student_numbers).await?;
417        // `register` (reached via `ripen`) appends to this index and commits the field back whole,
418        // so leaving it unloaded here would wipe every attainment already indexed for the pair.
419        let attainments_by_person_course = hmget_json(
420            &mut conn,
421            &key(generation, IDX_ATTAINMENTS_BY_PERSON_COURSE),
422            &person_course_keys,
423        )
424        .await?;
425        Ok(WorkingSet {
426            persons,
427            submissions,
428            attainments_by_person_course,
429            ..Default::default()
430        })
431    }
432
433    pub async fn load_for_list_by_course(
434        &self,
435        generation: &str,
436        course_codes: &[String],
437    ) -> anyhow::Result<WorkingSet> {
438        let mut conn = self.conn().await?;
439        let course_units: BTreeMap<String, MockCourseUnit> =
440            hmget_json(&mut conn, &key(generation, COURSE_UNITS), course_codes).await?;
441        let realisation_ids: Vec<String> = course_units
442            .values()
443            .flat_map(|unit| unit.realisations.iter().map(|r| r.id.clone()))
444            .unique()
445            .collect();
446        let enrolments_by_realisation: BTreeMap<String, Vec<String>> = hmget_json(
447            &mut conn,
448            &key(generation, IDX_ENROLMENTS_BY_REALISATION),
449            &realisation_ids,
450        )
451        .await?;
452        let enrolment_ids = flatten(enrolments_by_realisation.values());
453        let enrolments: BTreeMap<String, MockEnrolment> =
454            hmget_json(&mut conn, &key(generation, ENROLMENTS), &enrolment_ids).await?;
455        let student_numbers: Vec<String> = enrolments
456            .values()
457            .map(|enrolment| enrolment.student_number.clone())
458            .unique()
459            .collect();
460        let persons = hmget_json(&mut conn, &key(generation, PERSONS), &student_numbers).await?;
461        Ok(WorkingSet {
462            persons,
463            course_units,
464            enrolments,
465            enrolments_by_realisation,
466            ..Default::default()
467        })
468    }
469
470    /// The one write of a request: changed entities plus its call-log entry, in one atomic pipeline.
471    pub async fn commit(
472        &self,
473        generation: &str,
474        working: &WorkingSet,
475        call: &RecordedCall,
476        call_log_capacity: usize,
477    ) -> anyhow::Result<()> {
478        let mut conn = self.conn().await?;
479        let mut pipe = redis::pipe();
480        pipe.atomic();
481        for write in &working.writes {
482            match write {
483                WorldWrite::UpsertSubmission(id) => {
484                    let submission = working
485                        .submissions
486                        .get(id)
487                        .ok_or_else(|| anyhow!("write names a submission the working set lost"))?;
488                    pipe.hset(
489                        key(generation, SUBMISSIONS),
490                        id,
491                        serde_json::to_string(submission)?,
492                    )
493                    .ignore();
494                }
495                WorldWrite::UpsertAttainment(id) => {
496                    let attainment = working
497                        .attainments
498                        .get(id)
499                        .ok_or_else(|| anyhow!("write names an attainment the working set lost"))?;
500                    pipe.hset(
501                        key(generation, ATTAINMENTS),
502                        id,
503                        serde_json::to_string(attainment)?,
504                    )
505                    .ignore();
506                }
507                WorldWrite::IndexSubmission {
508                    student_number,
509                    course_code,
510                    ..
511                } => {
512                    let field = person_course_key(student_number, course_code);
513                    let ids = working
514                        .submissions_by_person_course
515                        .get(&field)
516                        .cloned()
517                        .unwrap_or_default();
518                    pipe.hset(
519                        key(generation, IDX_SUBMISSIONS_BY_PERSON_COURSE),
520                        field,
521                        serde_json::to_string(&ids)?,
522                    )
523                    .ignore();
524                }
525                WorldWrite::IndexAttainment {
526                    student_number,
527                    course_code,
528                    ..
529                } => {
530                    let field = person_course_key(student_number, course_code);
531                    let ids = working
532                        .attainments_by_person_course
533                        .get(&field)
534                        .cloned()
535                        .unwrap_or_default();
536                    pipe.hset(
537                        key(generation, IDX_ATTAINMENTS_BY_PERSON_COURSE),
538                        field,
539                        serde_json::to_string(&ids)?,
540                    )
541                    .ignore();
542                }
543            }
544        }
545        pipe.lpush(key(generation, CALLS), serde_json::to_string(call)?)
546            .ignore();
547        pipe.ltrim(
548            key(generation, CALLS),
549            0,
550            call_log_capacity.saturating_sub(1) as isize,
551        )
552        .ignore();
553        pipe.query_async::<()>(&mut conn).await?;
554        Ok(())
555    }
556
557    pub async fn next_call_seq(&self, generation: &str) -> anyhow::Result<u64> {
558        let mut conn = self.conn().await?;
559        let seq: i64 = conn.incr(key(generation, SEQ_CALL), 1).await?;
560        Ok(seq.max(0) as u64)
561    }
562
563    pub async fn next_person_seq(&self, generation: &str) -> anyhow::Result<i64> {
564        let mut conn = self.conn().await?;
565        Ok(conn.incr(key(generation, SEQ_PERSON), 1).await?)
566    }
567
568    /// Re-arming an id takes a fresh one, so the fault moves to the back of arm order.
569    pub async fn next_fault_seq(&self, generation: &str) -> anyhow::Result<u64> {
570        let mut conn = self.conn().await?;
571        let seq: i64 = conn.incr(key(generation, SEQ_FAULT), 1).await?;
572        Ok(seq.max(0) as u64)
573    }
574
575    /// The caller acts on the returned value, never on a separate read.
576    pub async fn draw(&self, generation: &str, fault_id: &str, delta: i64) -> anyhow::Result<i64> {
577        let mut conn = self.conn().await?;
578        Ok(conn
579            .hincr(key(generation, FAULTS_REMAINING), fault_id, delta)
580            .await?)
581    }
582
583    pub async fn upsert_json<T: Serialize>(
584        &self,
585        generation: &str,
586        hash: EntityHash,
587        entries: &BTreeMap<String, T>,
588    ) -> anyhow::Result<()> {
589        if entries.is_empty() {
590            return Ok(());
591        }
592        let mut conn = self.conn().await?;
593        let mut pipe = redis::pipe();
594        pipe.atomic();
595        for (field, value) in entries {
596            pipe.hset(
597                key(generation, hash.name()),
598                field,
599                serde_json::to_string(value)?,
600            )
601            .ignore();
602        }
603        pipe.query_async::<()>(&mut conn).await?;
604        Ok(())
605    }
606
607    pub async fn get_json<T: DeserializeOwned>(
608        &self,
609        generation: &str,
610        hash: EntityHash,
611        field: &str,
612    ) -> anyhow::Result<Option<T>> {
613        let mut conn = self.conn().await?;
614        let raw: Option<String> = conn.hget(key(generation, hash.name()), field).await?;
615        Ok(match raw {
616            Some(raw) => Some(serde_json::from_str(&raw)?),
617            None => None,
618        })
619    }
620
621    pub async fn all_json<T: DeserializeOwned>(
622        &self,
623        generation: &str,
624        hash: EntityHash,
625    ) -> anyhow::Result<BTreeMap<String, T>> {
626        let mut conn = self.conn().await?;
627        let raw: HashMap<String, String> = conn.hgetall(key(generation, hash.name())).await?;
628        raw.into_iter()
629            .map(|(field, value)| Ok((field, serde_json::from_str(&value)?)))
630            .collect()
631    }
632
633    pub async fn delete_fields(
634        &self,
635        generation: &str,
636        hash: EntityHash,
637        fields: &[String],
638    ) -> anyhow::Result<()> {
639        if fields.is_empty() {
640            return Ok(());
641        }
642        let mut conn = self.conn().await?;
643        conn.hdel::<_, _, ()>(key(generation, hash.name()), fields)
644            .await?;
645        Ok(())
646    }
647
648    pub async fn owner_keys(
649        &self,
650        generation: &str,
651        field: &str,
652    ) -> anyhow::Result<Option<OwnerKeys>> {
653        self.get_json(generation, EntityHash::OwnerKeys, field)
654            .await
655    }
656
657    pub async fn known_owner_refs(&self, generation: &str) -> anyhow::Result<Vec<String>> {
658        let mut conn = self.conn().await?;
659        let fields: Vec<String> = conn.hkeys(key(generation, IDX_OWNER_KEYS)).await?;
660        Ok(fields)
661    }
662
663    pub async fn faults(&self, generation: &str) -> anyhow::Result<Vec<Fault>> {
664        let mut faults: Vec<Fault> = self
665            .all_json::<Fault>(generation, EntityHash::Faults)
666            .await?
667            .into_values()
668            .collect();
669        faults.sort_by_key(|fault| fault.seq);
670        Ok(faults)
671    }
672
673    pub async fn remaining_budgets(
674        &self,
675        generation: &str,
676    ) -> anyhow::Result<HashMap<String, i64>> {
677        let mut conn = self.conn().await?;
678        Ok(conn.hgetall(key(generation, FAULTS_REMAINING)).await?)
679    }
680
681    pub async fn arm_fault(&self, generation: &str, fault: &Fault) -> anyhow::Result<()> {
682        let mut conn = self.conn().await?;
683        let mut pipe = redis::pipe();
684        pipe.atomic();
685        pipe.hset(
686            key(generation, FAULTS),
687            &fault.id,
688            serde_json::to_string(fault)?,
689        )
690        .ignore();
691        pipe.hset(
692            key(generation, FAULTS_REMAINING),
693            &fault.id,
694            fault.lifetime.budget().unwrap_or(0) as i64,
695        )
696        .ignore();
697        pipe.query_async::<()>(&mut conn).await?;
698        Ok(())
699    }
700
701    pub async fn disarm_faults(&self, generation: &str, ids: &[String]) -> anyhow::Result<()> {
702        if ids.is_empty() {
703            return Ok(());
704        }
705        let mut conn = self.conn().await?;
706        let mut pipe = redis::pipe();
707        pipe.atomic();
708        pipe.hdel(key(generation, FAULTS), ids).ignore();
709        pipe.hdel(key(generation, FAULTS_REMAINING), ids).ignore();
710        pipe.query_async::<()>(&mut conn).await?;
711        Ok(())
712    }
713
714    /// Every read of the call log is bounded: a list has no index, so a filter is always a scan.
715    pub async fn recent_calls(
716        &self,
717        generation: &str,
718        limit: usize,
719    ) -> anyhow::Result<Vec<RecordedCall>> {
720        if limit == 0 {
721            return Ok(Vec::new());
722        }
723        let mut conn = self.conn().await?;
724        let raw: Vec<String> = conn
725            .lrange(key(generation, CALLS), 0, limit as isize - 1)
726            .await?;
727        raw.iter()
728            .map(|entry| {
729                serde_json::from_str(entry).context("a stored call-log entry is unreadable")
730            })
731            .collect()
732    }
733
734    pub async fn counts(&self, generation: &str) -> anyhow::Result<WorldCounts> {
735        let mut conn = self.conn().await?;
736        let (
737            persons,
738            course_units,
739            enrolments,
740            attainments,
741            submissions,
742            product_tokens,
743            call_log_len,
744        ): (usize, usize, usize, usize, usize, usize, usize) = redis::pipe()
745            .hlen(key(generation, PERSONS))
746            .hlen(key(generation, COURSE_UNITS))
747            .hlen(key(generation, ENROLMENTS))
748            .hlen(key(generation, ATTAINMENTS))
749            .hlen(key(generation, SUBMISSIONS))
750            .hlen(key(generation, PRODUCT_TOKENS))
751            .llen(key(generation, CALLS))
752            .query_async(&mut conn)
753            .await?;
754        let faults = self.faults(generation).await?;
755        let remaining = self.remaining_budgets(generation).await?;
756        let spent = faults
757            .iter()
758            .filter(|fault| {
759                fault.lifetime.budget().is_some()
760                    && remaining.get(&fault.id).copied().unwrap_or(0) <= 0
761            })
762            .count();
763        Ok(WorldCounts {
764            persons,
765            course_units,
766            enrolments,
767            attainments,
768            submissions,
769            product_tokens,
770            faults_armed: faults.len() - spent,
771            faults_spent: spent,
772            call_log_len,
773        })
774    }
775
776    pub async fn set_defaults(
777        &self,
778        generation: &str,
779        defaults: &WorldDefaults,
780    ) -> anyhow::Result<()> {
781        let mut conn = self.conn().await?;
782        conn.hset::<_, _, _, ()>(
783            key(generation, META),
784            "defaults",
785            serde_json::to_string(defaults)?,
786        )
787        .await?;
788        Ok(())
789    }
790
791    pub async fn clear_hash(&self, generation: &str, hash: EntityHash) -> anyhow::Result<()> {
792        let mut conn = self.conn().await?;
793        conn.del::<_, ()>(key(generation, hash.name())).await?;
794        Ok(())
795    }
796
797    pub async fn clear_faults(&self, generation: &str) -> anyhow::Result<()> {
798        let mut conn = self.conn().await?;
799        let mut pipe = redis::pipe();
800        pipe.atomic();
801        for name in [FAULTS, FAULTS_REMAINING] {
802            pipe.del(key(generation, name)).ignore();
803        }
804        pipe.query_async::<()>(&mut conn).await?;
805        Ok(())
806    }
807
808    /// Rebuilds every derived index from the stored entities, so an upsert cannot leave one behind.
809    pub async fn reindex(&self, generation: &str) -> anyhow::Result<()> {
810        let world = World {
811            defaults: WorldDefaults::default(),
812            persons: self.all_json(generation, EntityHash::Persons).await?,
813            course_units: self.all_json(generation, EntityHash::CourseUnits).await?,
814            enrolments: self.all_json(generation, EntityHash::Enrolments).await?,
815            attainments: self.all_json(generation, EntityHash::Attainments).await?,
816            submissions: self.all_json(generation, EntityHash::Submissions).await?,
817            product_tokens: self.all_json(generation, EntityHash::ProductTokens).await?,
818        };
819
820        let mut conn = self.conn().await?;
821        let mut pipe = redis::pipe();
822        pipe.atomic();
823        for name in [
824            IDX_ENROLMENTS_BY_PERSON,
825            IDX_ENROLMENTS_BY_REALISATION,
826            IDX_ATTAINMENTS_BY_PERSON_COURSE,
827            IDX_SUBMISSIONS_BY_PERSON_COURSE,
828            IDX_OWNER_KEYS,
829        ] {
830            pipe.del(key(generation, name)).ignore();
831        }
832        write_derived_indexes(&mut pipe, generation, &world)?;
833        pipe.query_async::<()>(&mut conn).await?;
834        Ok(())
835    }
836}
837
838/// Hashes the command surface reads and writes by name.
839#[derive(Debug, Clone, Copy, PartialEq, Eq)]
840pub enum EntityHash {
841    Persons,
842    CourseUnits,
843    Enrolments,
844    Attainments,
845    Submissions,
846    ProductTokens,
847    Faults,
848    OwnerKeys,
849    Calls,
850}
851
852impl EntityHash {
853    fn name(self) -> &'static str {
854        match self {
855            Self::Persons => PERSONS,
856            Self::CourseUnits => COURSE_UNITS,
857            Self::Enrolments => ENROLMENTS,
858            Self::Attainments => ATTAINMENTS,
859            Self::Submissions => SUBMISSIONS,
860            Self::ProductTokens => PRODUCT_TOKENS,
861            Self::Faults => FAULTS,
862            Self::OwnerKeys => IDX_OWNER_KEYS,
863            Self::Calls => CALLS,
864        }
865    }
866}
867
868fn database_url(redis_url: &str, db_index: i64) -> anyhow::Result<String> {
869    let mut url = Url::parse(redis_url).context("REDIS_URL is not a url")?;
870    url.set_path(&db_index.to_string());
871    Ok(url.to_string())
872}
873
874fn key(generation: &str, name: &str) -> String {
875    format!("ms:{generation}:{name}")
876}
877
878fn write_entity_hash<T: Serialize>(
879    pipe: &mut redis::Pipeline,
880    generation: &str,
881    name: &str,
882    entries: &BTreeMap<String, T>,
883) -> anyhow::Result<()> {
884    for (field, value) in entries {
885        pipe.hset(key(generation, name), field, serde_json::to_string(value)?)
886            .ignore();
887    }
888    Ok(())
889}
890
891fn write_derived_indexes(
892    pipe: &mut redis::Pipeline,
893    generation: &str,
894    world: &World,
895) -> anyhow::Result<()> {
896    let mut by_person: BTreeMap<String, Vec<String>> = BTreeMap::new();
897    let mut by_realisation: BTreeMap<String, Vec<String>> = BTreeMap::new();
898    for enrolment in world.enrolments.values() {
899        by_person
900            .entry(enrolment.student_number.clone())
901            .or_default()
902            .push(enrolment.id.clone());
903        by_realisation
904            .entry(enrolment.realisation_id.clone())
905            .or_default()
906            .push(enrolment.id.clone());
907    }
908    let mut attainments_by_person_course: BTreeMap<String, Vec<String>> = BTreeMap::new();
909    for attainment in world.attainments.values() {
910        attainments_by_person_course
911            .entry(person_course_key(
912                &attainment.student_number,
913                &attainment.course_code,
914            ))
915            .or_default()
916            .push(attainment.id.clone());
917    }
918    let mut submissions_by_person_course: BTreeMap<String, Vec<String>> = BTreeMap::new();
919    for submission in world.submissions.values() {
920        submissions_by_person_course
921            .entry(person_course_key(
922                &submission.student_number,
923                &submission.course_code,
924            ))
925            .or_default()
926            .push(submission.submitted_attainment_id.clone());
927    }
928
929    let mut owner_keys: BTreeMap<String, OwnerKeys> = BTreeMap::new();
930    for person in world.persons.values() {
931        if let Some(email) = &person.owner_user_email {
932            owner_keys
933                .entry(format!("user:{email}"))
934                .or_default()
935                .student_numbers
936                .push(person.student_number.clone());
937        }
938    }
939    for unit in world.course_units.values() {
940        if let Some(slug) = &unit.owner_course_slug {
941            let entry = owner_keys.entry(format!("course:{slug}")).or_default();
942            entry.course_codes.push(unit.course_code.clone());
943            for realisation in &unit.realisations {
944                if let Some(product_id) = &realisation.open_university_product_id {
945                    entry.product_ids.push(product_id.clone());
946                }
947            }
948        }
949    }
950
951    for (name, index) in [
952        (IDX_ENROLMENTS_BY_PERSON, by_person),
953        (IDX_ENROLMENTS_BY_REALISATION, by_realisation),
954        (
955            IDX_ATTAINMENTS_BY_PERSON_COURSE,
956            attainments_by_person_course,
957        ),
958        (
959            IDX_SUBMISSIONS_BY_PERSON_COURSE,
960            submissions_by_person_course,
961        ),
962    ] {
963        for (field, ids) in index {
964            pipe.hset(key(generation, name), field, serde_json::to_string(&ids)?)
965                .ignore();
966        }
967    }
968    for (field, keys) in owner_keys {
969        pipe.hset(
970            key(generation, IDX_OWNER_KEYS),
971            field,
972            serde_json::to_string(&keys)?,
973        )
974        .ignore();
975    }
976    Ok(())
977}
978
979/// Always `HMGET`, so the reply is an array whatever the field count — a single-field `HGET` answers
980/// with a bare string, and no fields at all is a protocol error.
981fn hmget_cmd(hash: &str, fields: &[String]) -> redis::Cmd {
982    let mut cmd = redis::cmd("HMGET");
983    cmd.arg(hash);
984    if fields.is_empty() {
985        cmd.arg("");
986    }
987    for field in fields {
988        cmd.arg(field);
989    }
990    cmd
991}
992
993fn push_hmget(pipe: &mut redis::Pipeline, hash: &str, fields: &[String]) {
994    pipe.add_command(hmget_cmd(hash, fields));
995}
996
997async fn hmget_json<T: DeserializeOwned>(
998    conn: &mut ConnectionManager,
999    hash: &str,
1000    fields: &[String],
1001) -> anyhow::Result<BTreeMap<String, T>> {
1002    if fields.is_empty() {
1003        return Ok(BTreeMap::new());
1004    }
1005    let values: Vec<Option<String>> = hmget_cmd(hash, fields).query_async(conn).await?;
1006    zip_json(fields, values)
1007}
1008
1009fn zip_json<T: DeserializeOwned>(
1010    fields: &[String],
1011    values: Vec<Option<String>>,
1012) -> anyhow::Result<BTreeMap<String, T>> {
1013    let mut out = BTreeMap::new();
1014    for (field, value) in fields.iter().zip(values) {
1015        if let Some(value) = value {
1016            out.insert(
1017                field.clone(),
1018                serde_json::from_str(&value)
1019                    .with_context(|| format!("stored value for `{field}` is unreadable"))?,
1020            );
1021        }
1022    }
1023    Ok(out)
1024}
1025
1026fn flatten<'a, I: Iterator<Item = &'a Vec<String>>>(lists: I) -> Vec<String> {
1027    lists
1028        .flat_map(|list| list.iter().cloned())
1029        .unique()
1030        .collect()
1031}
1032
1033#[cfg(test)]
1034mod tests {
1035    use super::*;
1036
1037    /// "`FLUSHDB` is safe" rests on the swap, and the deployed url carries the cache's index 1 in its
1038    /// path.
1039    #[test]
1040    fn the_configured_url_is_moved_off_the_caches_index() {
1041        assert_eq!(
1042            database_url("redis://redis.default.svc.cluster.local/1", 2)
1043                .expect("the deployed url parses"),
1044            "redis://redis.default.svc.cluster.local/2"
1045        );
1046        assert!(MockSuotarStore::new("redis://redis.default.svc.cluster.local/1", 2).is_ok());
1047    }
1048}
headless_lms_server/controllers/mock_suotar/store.rs

headless_lms_server/controllers/mock_suotar/
store.rs
