headless_lms_models/
errors.rs

1use crate::prelude::*;
2use headless_lms_utils::error_identifier::{
3    calculate_error_grouping_identifier, calculate_exact_error_identifier, normalize_message,
4    normalize_stack_trace,
5};
6use rand::RngExt;
7use utoipa::ToSchema;
8
9#[derive(Debug, Serialize, Deserialize, PartialEq, Clone, Copy, Type, ToSchema)]
10#[serde(rename_all = "snake_case")]
11#[sqlx(type_name = "error_source", rename_all = "snake_case")]
12pub enum ErrorSource {
13    Backend,
14    Frontend,
15}
16
17#[derive(Debug, Serialize, Deserialize, Clone, ToSchema)]
18pub struct ErrorVariant {
19    pub id: Uuid,
20    pub service: String,
21    pub exact_error_identifier: String,
22    pub error_grouping_identifier: String,
23    pub error_source: ErrorSource,
24    pub example_message: String,
25    pub example_stack_trace: Option<String>,
26    pub normalized_message: String,
27    pub normalized_stack_trace: Option<String>,
28    pub occurrence_count: i32,
29    pub last_seen_at: DateTime<Utc>,
30    pub resolved_at: Option<DateTime<Utc>>,
31    pub created_at: DateTime<Utc>,
32    pub updated_at: DateTime<Utc>,
33    pub deleted_at: Option<DateTime<Utc>>,
34}
35
36#[derive(Debug, Serialize, Deserialize, Clone, ToSchema)]
37pub struct NewErrorReport {
38    pub service: String,
39    pub error_source: Option<ErrorSource>,
40    pub message: String,
41    pub stack_trace: Option<String>,
42    pub path: Option<String>,
43    pub app_version: Option<String>,
44    pub details: Option<serde_json::Value>,
45}
46
47/// Inserts one error occurrence and upserts its variant in a single transaction.
48/// The variant service is treated as immutable after creation; any future service
49/// rename must update both error_variants.service and related error_occurrences.service
50/// in the same transaction.
51pub async fn insert(
52    conn: &mut PgConnection,
53    user_id: Option<Uuid>,
54    report: &NewErrorReport,
55) -> ModelResult<Uuid> {
56    let service = report.service.trim();
57    if service.is_empty() {
58        return Err(model_err!(
59            InvalidRequest,
60            "service must not be empty".to_string()
61        ));
62    }
63
64    let error_source = report.error_source.unwrap_or(ErrorSource::Frontend);
65    let error_source_str = match error_source {
66        ErrorSource::Frontend => "frontend",
67        ErrorSource::Backend => "backend",
68    };
69
70    let normalized_message = normalize_message(&report.message);
71    let normalized_stack_trace = report.stack_trace.as_deref().map(normalize_stack_trace);
72
73    let exact_error_identifier = calculate_exact_error_identifier(
74        service,
75        error_source_str,
76        &normalized_message,
77        normalized_stack_trace.as_deref(),
78    );
79    let error_grouping_identifier =
80        calculate_error_grouping_identifier(service, error_source_str, &normalized_message);
81
82    let mut tx = conn.begin().await?;
83    let variant_id = sqlx::query!(
84        r#"
85INSERT INTO error_variants (
86    id,
87    service,
88    exact_error_identifier,
89    error_grouping_identifier,
90    error_source,
91    example_message,
92    example_stack_trace,
93    normalized_message,
94    normalized_stack_trace
95)
96VALUES (uuid_generate_v4(), $1, $2, $3, $4, $5, $6, $7, $8)
97ON CONFLICT (service, exact_error_identifier, deleted_at) DO UPDATE SET
98    deleted_at = NULL,
99    resolved_at = NULL,
100    occurrence_count = error_variants.occurrence_count + 1,
101    last_seen_at = now(),
102    updated_at = now()
103RETURNING *
104        "#,
105        service,
106        exact_error_identifier,
107        error_grouping_identifier,
108        error_source as ErrorSource,
109        report.message,
110        report.stack_trace,
111        normalized_message,
112        normalized_stack_trace
113    )
114    .fetch_one(&mut *tx)
115    .await?
116    .id;
117
118    let occurrence_id = Uuid::new_v4();
119    sqlx::query!(
120        "
121INSERT INTO error_occurrences (id, error_variant_id, service, user_id, path, app_version, details)
122VALUES ($1, $2, $3, $4, $5, $6, $7)
123        ",
124        occurrence_id,
125        variant_id,
126        service,
127        user_id,
128        report.path,
129        report.app_version,
130        report.details
131    )
132    .execute(&mut *tx)
133    .await?;
134
135    let service_consistent = sqlx::query_scalar!(
136        r#"
137SELECT EXISTS(
138    SELECT 1
139    FROM error_occurrences o
140    JOIN error_variants v ON v.id = o.error_variant_id
141    WHERE o.id = $1
142      AND o.service = v.service
143) AS "exists!"
144        "#,
145        occurrence_id
146    )
147    .fetch_one(&mut *tx)
148    .await?;
149    if !service_consistent {
150        return Err(model_err!(
151            Generic,
152            "error occurrence service must match variant service".to_string()
153        ));
154    }
155
156    tx.commit().await?;
157
158    Ok(variant_id)
159}
160
161pub async fn get_all_variants(
162    conn: &mut PgConnection,
163    pagination: Pagination,
164) -> ModelResult<Vec<ErrorVariant>> {
165    let res = sqlx::query!(
166        r#"
167SELECT
168    *
169FROM error_variants
170WHERE deleted_at IS NULL
171ORDER BY last_seen_at DESC
172LIMIT $1 OFFSET $2
173        "#,
174        pagination.limit(),
175        pagination.offset()
176    )
177    .map(|r| ErrorVariant {
178        id: r.id,
179        service: r.service,
180        exact_error_identifier: r.exact_error_identifier,
181        error_grouping_identifier: r.error_grouping_identifier,
182        error_source: r.error_source,
183        example_message: r.example_message,
184        example_stack_trace: r.example_stack_trace,
185        normalized_message: r.normalized_message,
186        normalized_stack_trace: r.normalized_stack_trace,
187        occurrence_count: r.occurrence_count,
188        last_seen_at: r.last_seen_at,
189        resolved_at: r.resolved_at,
190        created_at: r.created_at,
191        updated_at: r.updated_at,
192        deleted_at: r.deleted_at,
193    })
194    .fetch_all(conn)
195    .await?;
196    Ok(res)
197}
198
199pub async fn delete_expired(conn: &mut PgConnection) -> ModelResult<()> {
200    use std::collections::HashSet;
201
202    let mut tx = conn.begin().await?;
203    let deleted_variant_ids = sqlx::query!(
204        r#"
205UPDATE error_occurrences
206SET deleted_at = now()
207WHERE created_at < now() - interval '2 months'
208  AND deleted_at IS NULL
209RETURNING *
210        "#
211    )
212    .fetch_all(&mut *tx)
213    .await?
214    .into_iter()
215    .map(|r| r.error_variant_id)
216    .collect::<HashSet<_>>()
217    .into_iter()
218    .collect::<Vec<_>>();
219
220    // Benign race: a concurrent insert between DELETE and this UPDATE can briefly skew aggregates.
221    sqlx::query!(
222        r#"
223UPDATE error_variants v
224SET
225    occurrence_count = (
226        SELECT COUNT(*)::int
227        FROM error_occurrences
228        WHERE error_variant_id = v.id
229          AND deleted_at IS NULL
230    ),
231    last_seen_at = COALESCE(
232        (
233            SELECT MAX(created_at)
234            FROM error_occurrences
235            WHERE error_variant_id = v.id
236              AND deleted_at IS NULL
237        ),
238        v.created_at
239    ),
240    updated_at = now()
241WHERE v.id = ANY($1)
242        "#,
243        &deleted_variant_ids
244    )
245    .execute(&mut *tx)
246    .await?;
247
248    tx.commit().await?;
249    Ok(())
250}
251
252pub async fn maybe_delete_expired(conn: &mut PgConnection) -> ModelResult<()> {
253    if rand::rng().random_range(1..=1000) == 1 {
254        info!("Cleaning up expired errors");
255        delete_expired(conn).await?;
256    }
257    Ok(())
258}
headless_lms_models/errors.rs

headless_lms_models/
errors.rs
