headless_lms_models/
credit_registration_events.rs

1//! Append-only audit trail for the credit registration ledger.
2//!
3//! No retention sweep touches this table, so every Suotar payload must go through
4//! [`scrub_suotar_body`](crate::library::credit_registration::scrub::scrub_suotar_body) at the
5//! write site — redacting on read would leave the raw values on disk.
6use std::collections::HashMap;
7
8use serde_json::Value;
9use utoipa::ToSchema;
10
11use crate::credit_registrations::{CreditRegistrationErrorCode, CreditRegistrationState};
12use crate::prelude::*;
13use crate::suotar_api_calls::SuotarEndpoint;
14
15#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, Clone, Copy, Hash, Type, ToSchema)]
16#[sqlx(
17    type_name = "credit_registration_event_kind",
18    rename_all = "snake_case"
19)]
20#[serde(rename_all = "snake_case")]
21pub enum CreditRegistrationEventKind {
22    Created,
23    StateChanged,
24    SuotarResponse,
25    RetryScheduled,
26    AdminAction,
27    StudentAction,
28    Cancelled,
29}
30
31/// What Suotar did with one row of a request.
32#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, Clone, Copy, Hash, Type, ToSchema)]
33#[sqlx(type_name = "suotar_answer", rename_all = "snake_case")]
34#[serde(rename_all = "snake_case")]
35pub enum SuotarAnswer {
36    Answered,
37    /// Suotar answered the request but left this row out.
38    Unanswered,
39    /// Suotar refused the whole request, or it never got there.
40    Refused,
41}
42
43#[derive(Debug, Serialize, Deserialize, PartialEq, Clone, ToSchema)]
44pub struct CreditRegistrationEvent {
45    pub id: Uuid,
46    pub created_at: DateTime<Utc>,
47    pub updated_at: DateTime<Utc>,
48    pub deleted_at: Option<DateTime<Utc>>,
49    pub credit_registration_id: Uuid,
50    pub kind: CreditRegistrationEventKind,
51    pub from_state: Option<CreditRegistrationState>,
52    pub to_state: Option<CreditRegistrationState>,
53    pub error_code: Option<CreditRegistrationErrorCode>,
54    pub message: Option<String>,
55    pub suotar_api_call_id: Option<Uuid>,
56    pub actor_user_id: Option<Uuid>,
57    pub details: Option<Value>,
58    /// The requestItemId the row went out under in the call behind this event.
59    pub request_item_id: Option<String>,
60    /// Outlives `suotar_api_call_id`, whose call row is swept after 90 days.
61    pub suotar_endpoint: Option<SuotarEndpoint>,
62    pub suotar_requested_at: Option<DateTime<Utc>>,
63    /// Orders the timeline where set: events written in one transaction share `created_at`.
64    pub suotar_answered_at: Option<DateTime<Utc>>,
65    pub suotar_answer: Option<SuotarAnswer>,
66}
67
68#[derive(Debug, Clone, PartialEq)]
69pub struct NewCreditRegistrationEvent {
70    pub credit_registration_id: Uuid,
71    pub kind: CreditRegistrationEventKind,
72    pub from_state: Option<CreditRegistrationState>,
73    pub to_state: Option<CreditRegistrationState>,
74    pub error_code: Option<CreditRegistrationErrorCode>,
75    pub message: Option<String>,
76    pub suotar_api_call_id: Option<Uuid>,
77    pub actor_user_id: Option<Uuid>,
78    /// Build with
79    /// [`suotar_exchange_details`](crate::library::credit_registration::scrub::suotar_exchange_details)
80    /// so it is scrubbed.
81    pub details: Option<Value>,
82    pub request_item_id: Option<String>,
83    pub suotar_endpoint: Option<SuotarEndpoint>,
84    pub suotar_requested_at: Option<DateTime<Utc>>,
85    pub suotar_answered_at: Option<DateTime<Utc>>,
86    pub suotar_answer: Option<SuotarAnswer>,
87}
88
89impl NewCreditRegistrationEvent {
90    pub fn new(credit_registration_id: Uuid, kind: CreditRegistrationEventKind) -> Self {
91        Self {
92            credit_registration_id,
93            kind,
94            from_state: None,
95            to_state: None,
96            error_code: None,
97            message: None,
98            suotar_api_call_id: None,
99            actor_user_id: None,
100            details: None,
101            request_item_id: None,
102            suotar_endpoint: None,
103            suotar_requested_at: None,
104            suotar_answered_at: None,
105            suotar_answer: None,
106        }
107    }
108}
109
110/// Callers that also change `state` must go through `credit_registrations::transition` instead,
111/// which writes both in one transaction.
112pub async fn insert(
113    conn: &mut PgConnection,
114    new: &NewCreditRegistrationEvent,
115) -> ModelResult<Uuid> {
116    let res = sqlx::query!(
117        r#"
118INSERT INTO credit_registration_events (
119    credit_registration_id,
120    kind,
121    from_state,
122    to_state,
123    error_code,
124    message,
125    suotar_api_call_id,
126    actor_user_id,
127    details,
128    request_item_id,
129    suotar_endpoint,
130    suotar_requested_at,
131    suotar_answered_at,
132    suotar_answer
133  )
134VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14)
135RETURNING id
136        "#,
137        new.credit_registration_id,
138        new.kind as CreditRegistrationEventKind,
139        new.from_state as Option<CreditRegistrationState>,
140        new.to_state as Option<CreditRegistrationState>,
141        new.error_code as Option<CreditRegistrationErrorCode>,
142        new.message,
143        new.suotar_api_call_id,
144        new.actor_user_id,
145        new.details,
146        new.request_item_id,
147        new.suotar_endpoint as Option<SuotarEndpoint>,
148        new.suotar_requested_at,
149        new.suotar_answered_at,
150        new.suotar_answer as Option<SuotarAnswer>,
151    )
152    .fetch_one(conn)
153    .await?;
154    Ok(res.id)
155}
156
157/// Appends one event per element, in one statement. The column list is [`insert`]'s, so a batch
158/// writes the same rows a loop would.
159pub async fn insert_batch(
160    conn: &mut PgConnection,
161    events: &[NewCreditRegistrationEvent],
162) -> ModelResult<()> {
163    if events.is_empty() {
164        return Ok(());
165    }
166    let ids: Vec<Uuid> = events.iter().map(|e| e.credit_registration_id).collect();
167    let kinds: Vec<CreditRegistrationEventKind> = events.iter().map(|e| e.kind).collect();
168    let from_states: Vec<Option<CreditRegistrationState>> =
169        events.iter().map(|e| e.from_state).collect();
170    let to_states: Vec<Option<CreditRegistrationState>> =
171        events.iter().map(|e| e.to_state).collect();
172    let error_codes: Vec<Option<CreditRegistrationErrorCode>> =
173        events.iter().map(|e| e.error_code).collect();
174    let messages: Vec<Option<String>> = events.iter().map(|e| e.message.clone()).collect();
175    let call_ids: Vec<Option<Uuid>> = events.iter().map(|e| e.suotar_api_call_id).collect();
176    let actors: Vec<Option<Uuid>> = events.iter().map(|e| e.actor_user_id).collect();
177    let details: Vec<Option<Value>> = events.iter().map(|e| e.details.clone()).collect();
178    let request_item_ids: Vec<Option<String>> =
179        events.iter().map(|e| e.request_item_id.clone()).collect();
180    let endpoints: Vec<Option<SuotarEndpoint>> = events.iter().map(|e| e.suotar_endpoint).collect();
181    let requested_ats: Vec<Option<DateTime<Utc>>> =
182        events.iter().map(|e| e.suotar_requested_at).collect();
183    let answered_ats: Vec<Option<DateTime<Utc>>> =
184        events.iter().map(|e| e.suotar_answered_at).collect();
185    let answers: Vec<Option<SuotarAnswer>> = events.iter().map(|e| e.suotar_answer).collect();
186    sqlx::query!(
187        r#"
188INSERT INTO credit_registration_events (
189    credit_registration_id,
190    kind,
191    from_state,
192    to_state,
193    error_code,
194    message,
195    suotar_api_call_id,
196    actor_user_id,
197    details,
198    request_item_id,
199    suotar_endpoint,
200    suotar_requested_at,
201    suotar_answered_at,
202    suotar_answer
203  )
204SELECT *
205FROM UNNEST(
206    $1::uuid [],
207    $2::credit_registration_event_kind [],
208    $3::credit_registration_state [],
209    $4::credit_registration_state [],
210    $5::credit_registration_error_code [],
211    $6::text [],
212    $7::uuid [],
213    $8::uuid [],
214    $9::jsonb [],
215    $10::text [],
216    $11::suotar_endpoint [],
217    $12::timestamptz [],
218    $13::timestamptz [],
219    $14::suotar_answer []
220  )
221        "#,
222        &ids,
223        &kinds as &[CreditRegistrationEventKind],
224        &from_states as &[Option<CreditRegistrationState>],
225        &to_states as &[Option<CreditRegistrationState>],
226        &error_codes as &[Option<CreditRegistrationErrorCode>],
227        &messages as &[Option<String>],
228        &call_ids as &[Option<Uuid>],
229        &actors as &[Option<Uuid>],
230        &details as &[Option<Value>],
231        &request_item_ids as &[Option<String>],
232        &endpoints as &[Option<SuotarEndpoint>],
233        &requested_ats as &[Option<DateTime<Utc>>],
234        &answered_ats as &[Option<DateTime<Utc>>],
235        &answers as &[Option<SuotarAnswer>],
236    )
237    .execute(conn)
238    .await?;
239    Ok(())
240}
241
242/// Appends the same event to many rows in one round trip; [`insert_batch`] for events that differ.
243pub async fn insert_many(
244    conn: &mut PgConnection,
245    credit_registration_ids: &[Uuid],
246    kind: CreditRegistrationEventKind,
247    actor_user_id: Option<Uuid>,
248    message: Option<&str>,
249) -> ModelResult<()> {
250    if credit_registration_ids.is_empty() {
251        return Ok(());
252    }
253    sqlx::query!(
254        r#"
255INSERT INTO credit_registration_events (credit_registration_id, kind, actor_user_id, message)
256SELECT id, $2, $3, $4
257FROM UNNEST($1::uuid []) AS id
258        "#,
259        credit_registration_ids,
260        kind as CreditRegistrationEventKind,
261        actor_user_id,
262        message,
263    )
264    .execute(conn)
265    .await?;
266    Ok(())
267}
268
269/// The per-item timeline, newest first.
270pub async fn get_by_registration_id(
271    conn: &mut PgConnection,
272    credit_registration_id: Uuid,
273) -> ModelResult<Vec<CreditRegistrationEvent>> {
274    let res = sqlx::query_as!(
275        CreditRegistrationEvent,
276        r#"
277SELECT *
278FROM credit_registration_events
279WHERE credit_registration_id = $1
280  AND deleted_at IS NULL
281ORDER BY COALESCE(suotar_answered_at, created_at) DESC,
282  id DESC
283        "#,
284        credit_registration_id
285    )
286    .fetch_all(conn)
287    .await?;
288    Ok(res)
289}
290
291/// The timelines of several rows merged into one list, oldest first.
292pub async fn get_by_registration_ids(
293    conn: &mut PgConnection,
294    credit_registration_ids: &[Uuid],
295) -> ModelResult<Vec<CreditRegistrationEvent>> {
296    let res = sqlx::query_as!(
297        CreditRegistrationEvent,
298        r#"
299SELECT *
300FROM credit_registration_events
301WHERE credit_registration_id = ANY($1)
302  AND deleted_at IS NULL
303ORDER BY created_at,
304  COALESCE(suotar_answered_at, created_at) DESC,
305  id DESC
306        "#,
307        credit_registration_ids
308    )
309    .fetch_all(conn)
310    .await?;
311    Ok(res)
312}
313
314/// When each of the rows last entered each of `states`, keyed by (row, state). A pair missing
315/// from the map never entered that state; staying in a state is not entering it.
316pub async fn get_latest_state_entries(
317    conn: &mut PgConnection,
318    credit_registration_ids: &[Uuid],
319    states: &[CreditRegistrationState],
320) -> ModelResult<HashMap<(Uuid, CreditRegistrationState), DateTime<Utc>>> {
321    let rows = sqlx::query!(
322        r#"
323SELECT DISTINCT ON (credit_registration_id, to_state) credit_registration_id,
324  to_state AS "to_state!: CreditRegistrationState",
325  created_at
326FROM credit_registration_events
327WHERE credit_registration_id = ANY($1)
328  AND to_state = ANY($2::credit_registration_state [])
329  AND from_state IS DISTINCT FROM to_state
330  AND deleted_at IS NULL
331ORDER BY credit_registration_id,
332  to_state,
333  created_at DESC,
334  id DESC
335        "#,
336        credit_registration_ids,
337        states as &[CreditRegistrationState],
338    )
339    .fetch_all(conn)
340    .await?;
341    Ok(rows
342        .into_iter()
343        .map(|row| ((row.credit_registration_id, row.to_state), row.created_at))
344        .collect())
345}
346
347/// The per-item timeline entries one study registry call produced, oldest first: what the answer
348/// did to each row it covered.
349pub async fn get_by_suotar_api_call_id(
350    conn: &mut PgConnection,
351    suotar_api_call_id: Uuid,
352) -> ModelResult<Vec<CreditRegistrationEvent>> {
353    let res = sqlx::query_as!(
354        CreditRegistrationEvent,
355        r#"
356SELECT *
357FROM credit_registration_events
358WHERE suotar_api_call_id = $1
359  AND deleted_at IS NULL
360ORDER BY COALESCE(suotar_answered_at, created_at),
361  id
362        "#,
363        suotar_api_call_id
364    )
365    .fetch_all(conn)
366    .await?;
367    Ok(res)
368}
369
370/// The requestItemId each of `credit_registration_ids` went out under among `request_item_ids`,
371/// which are one call's. A row the call carried but no event recorded is missing from the map.
372pub async fn get_request_item_ids_in_call(
373    conn: &mut PgConnection,
374    credit_registration_ids: &[Uuid],
375    request_item_ids: &[String],
376) -> ModelResult<HashMap<Uuid, String>> {
377    let rows = sqlx::query!(
378        r#"
379SELECT DISTINCT ON (credit_registration_id) credit_registration_id,
380  request_item_id AS "request_item_id!"
381FROM credit_registration_events
382WHERE credit_registration_id = ANY($1)
383  AND request_item_id = ANY($2)
384  AND deleted_at IS NULL
385ORDER BY credit_registration_id,
386  COALESCE(suotar_answered_at, created_at),
387  id
388        "#,
389        credit_registration_ids,
390        request_item_ids,
391    )
392    .fetch_all(conn)
393    .await?;
394    Ok(rows
395        .into_iter()
396        .map(|row| (row.credit_registration_id, row.request_item_id))
397        .collect())
398}
399
400/// The attainment the study registry pointed at when it turned a submission down as no improvement.
401///
402/// Read back off the event the answer was recorded on rather than stored on the ledger row: the row
403/// holds what we sent, and this is what the registry already had.
404#[derive(Debug, Serialize, Deserialize, PartialEq, Clone, ToSchema)]
405pub struct NotImprovedAttainment {
406    pub grade_id: Option<String>,
407    /// Names the scale `grade_id` is on, without which "1" reads as a one out of five when it means
408    /// a pass.
409    pub grade_scale_id: Option<String>,
410}
411
412/// The registry's verdict for a row it declined as no improvement. `None` for every other row, and
413/// for one whose answer named no attainment.
414pub async fn get_not_improved_attainment(
415    conn: &mut PgConnection,
416    credit_registration_id: Uuid,
417) -> ModelResult<Option<NotImprovedAttainment>> {
418    let found = sqlx::query_scalar!(
419        r#"
420SELECT details #> '{response,result,previousAttainment}' AS "attainment!"
421FROM credit_registration_events
422WHERE credit_registration_id = $1
423  AND to_state = 'not_improved'
424  AND details #> '{response,result,previousAttainment}' IS NOT NULL
425  AND deleted_at IS NULL
426ORDER BY COALESCE(suotar_answered_at, created_at) DESC,
427  id DESC
428LIMIT 1
429        "#,
430        credit_registration_id
431    )
432    .fetch_optional(conn)
433    .await?;
434    Ok(found.map(|attainment| NotImprovedAttainment {
435        grade_id: string_field(&attainment, "gradeId"),
436        grade_scale_id: string_field(&attainment, "gradeScaleId"),
437    }))
438}
439
440fn string_field(value: &Value, key: &str) -> Option<String> {
441    Some(value.get(key)?.as_str()?.to_string())
442}
443
444/// How the study registry answered the items it was asked about in a window.
445#[derive(Debug, Clone, PartialEq, Default)]
446pub struct SuotarItemOutcomeTotals {
447    pub item_count: i64,
448    /// Items that failed on Suotar or Sisu being down or slow, the only proxy we have for Sisu's
449    /// uptime.
450    pub service_unavailable_count: i64,
451    pub last_service_unavailable_at: Option<DateTime<Utc>>,
452}
453
454/// Per-item outcomes in `[since, now)`, counted over events rather than rows: an item that failed
455/// really failed, whatever its row has since become.
456pub async fn count_item_outcomes_since(
457    conn: &mut PgConnection,
458    since: DateTime<Utc>,
459) -> ModelResult<SuotarItemOutcomeTotals> {
460    let res = sqlx::query_as!(
461        SuotarItemOutcomeTotals,
462        r#"
463SELECT COUNT(*) AS "item_count!",
464  COUNT(*) FILTER (
465    WHERE error_code IN ('service_temporarily_unavailable', 'sisu_timeout')
466  ) AS "service_unavailable_count!",
467  MAX(created_at) FILTER (
468    WHERE error_code IN ('service_temporarily_unavailable', 'sisu_timeout')
469  ) AS "last_service_unavailable_at"
470FROM credit_registration_events
471WHERE kind = 'suotar_response'
472  AND created_at >= $1
473  AND deleted_at IS NULL
474        "#,
475        since,
476    )
477    .fetch_one(conn)
478    .await?;
479    Ok(res)
480}
481
482/// One error code's standing over a window and the window before it.
483#[derive(Debug, Clone, PartialEq)]
484pub struct ErrorCodeWindowCounts {
485    pub error_code: CreditRegistrationErrorCode,
486    pub current_count: i64,
487    /// The equally long window immediately before, which is what a spike is measured against.
488    pub previous_count: i64,
489    pub user_count: i64,
490    pub course_count: i64,
491    pub first_seen_at: Option<DateTime<Utc>>,
492    pub last_seen_at: Option<DateTime<Utc>>,
493    /// The endpoints the code arrived on, empty for one recorded without a call of ours.
494    pub endpoints: Vec<SuotarEndpoint>,
495}
496
497/// Error events per code over `[now - 2 * window, now)`, split at `now - window`.
498///
499/// Counts events, so errors on attempts since superseded are included: an `invalid_credits` on
500/// attempt 1 is the configuration bug, whether or not attempt 2 succeeded.
501pub async fn get_error_code_counts_for_window(
502    conn: &mut PgConnection,
503    window_secs: i64,
504) -> ModelResult<Vec<ErrorCodeWindowCounts>> {
505    let res = sqlx::query_as!(
506        ErrorCodeWindowCounts,
507        r#"
508SELECT e.error_code AS "error_code!: CreditRegistrationErrorCode",
509  COUNT(*) FILTER (
510    WHERE e.created_at > now() - MAKE_INTERVAL(secs => $1::double precision)
511  ) AS "current_count!",
512  COUNT(*) FILTER (
513    WHERE e.created_at <= now() - MAKE_INTERVAL(secs => $1::double precision)
514  ) AS "previous_count!",
515  COUNT(DISTINCT cr.user_id) AS "user_count!",
516  COUNT(DISTINCT cr.course_id) AS "course_count!",
517  MIN(e.created_at) AS "first_seen_at",
518  MAX(e.created_at) AS "last_seen_at",
519  COALESCE(
520    ARRAY_AGG(DISTINCT call.endpoint) FILTER (
521      WHERE call.endpoint IS NOT NULL
522    ),
523    '{}'
524  ) AS "endpoints!: Vec<SuotarEndpoint>"
525FROM credit_registration_events e
526  JOIN credit_registrations cr ON cr.id = e.credit_registration_id
527  LEFT JOIN suotar_api_calls call ON call.id = e.suotar_api_call_id
528  AND call.deleted_at IS NULL
529WHERE e.error_code IS NOT NULL
530  AND e.created_at > now() - MAKE_INTERVAL(secs => $1::double precision * 2)
531  AND e.deleted_at IS NULL
532  AND cr.deleted_at IS NULL
533GROUP BY e.error_code
534ORDER BY 2 DESC
535        "#,
536        window_secs as f64,
537    )
538    .fetch_all(conn)
539    .await?;
540    Ok(res)
541}
542
543/// Registrations whose answers named more than one submitted attainment id, which is the shape a
544/// double submission would leave behind.
545///
546/// Per registration, not per completion: a grade improvement is a second registration row and a
547/// second attainment on purpose.
548pub async fn get_ids_with_several_submitted_attainments(
549    conn: &mut PgConnection,
550    limit: i64,
551) -> ModelResult<Vec<Uuid>> {
552    let res = sqlx::query_scalar!(
553        r#"
554SELECT credit_registration_id
555FROM credit_registration_events
556WHERE details #>> '{response,submittedAttainmentId}' IS NOT NULL
557  AND deleted_at IS NULL
558GROUP BY credit_registration_id
559HAVING COUNT(
560    DISTINCT details #>> '{response,submittedAttainmentId}'
561  ) > 1
562LIMIT $1
563        "#,
564        limit,
565    )
566    .fetch_all(conn)
567    .await?;
568    Ok(res)
569}
headless_lms_models/credit_registration_events.rs

headless_lms_models/
credit_registration_events.rs
