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

headless_lms_server/mock_suotar/
store.rs
