fix(db): atomic record_event, safe orphan purge, RFC3339 retention cutoff

crates/trapfall-core/src/store.rs

@@ -95,6 +95,19 @@ impl Store {
         self.db.upsert_issue(project_id, fingerprint, title, culprit, level).await
     }
 
+    /// Atomically upsert an issue and insert its event.
+    pub async fn record_event(
+        &self,
+        project_id: &str,
+        fingerprint: &str,
+        title: &str,
+        culprit: Option<&str>,
+        level: Level,
+        event_data: &str,
+    ) -> Result<Issue> {
+        self.db.record_event(project_id, fingerprint, title, culprit, level, event_data).await
+    }
+
     pub async fn get_issue(&self, issue_id: &str) -> Result<Option<Issue>> {
         self.db.get_issue(issue_id).await
     }

crates/trapfall-db/src/common.rs

@@ -246,6 +246,12 @@ pub fn extract_dsn_host(dsn: &str) -> String {
         .to_string()
 }
 
+/// RFC3339 cutoff `days` days before now. Same format as stored timestamps
+/// (`now_rfc3339`), so text comparison is consistent on both backends.
+pub fn cutoff_rfc3339_days(days: i64) -> String {
+    (chrono::Utc::now() - chrono::Duration::days(days)).to_rfc3339()
+}
+
 /// Current time as RFC3339 string.
 pub fn now_rfc3339() -> String {
     chrono::Utc::now().to_rfc3339()

crates/trapfall-db/src/lib.rs

@@ -244,6 +244,17 @@ pub trait Database: Send + Sync {
     // ── Events ──────────────────────────────────────────────────────────
 
     async fn insert_event(&self, issue_id: &str, project_id: &str, event_data: &str) -> Result<String>;
+    /// Upsert the issue and insert its event in ONE transaction, so an event
+    /// failure never leaves a bumped issue count without an event row.
+    async fn record_event(
+        &self,
+        project_id: &str,
+        fingerprint: &str,
+        title: &str,
+        culprit: Option<&str>,
+        level: Level,
+        event_data: &str,
+    ) -> Result<Issue>;
     async fn list_events(&self, issue_id: &str, limit: i64, offset: i64) -> Result<Vec<StoredEvent>>;
     async fn count_events(&self, issue_id: &str) -> Result<i64>;
 
@@ -337,8 +348,10 @@ pub trait Database: Send + Sync {
 
     /// Delete events older than `days` days. Returns count of deleted rows.
     async fn purge_old_events(&self, days: i64) -> Result<u64>;
-    /// Delete orphaned issues (no remaining events). Best-effort.
-    async fn purge_orphan_issues(&self) -> Result<()>;
+    /// Delete orphaned issues: no remaining events, not resolved/ignored, and
+    /// `last_seen` older than `days` days (so a freshly upserted issue whose
+    /// event is not yet inserted is never removed). Best-effort.
+    async fn purge_orphan_issues(&self, days: i64) -> Result<()>;
     /// Delete auth attempts older than 30 days. Best-effort.
     async fn purge_stale_auth_attempts(&self) -> Result<()>;
 

crates/trapfall-db/src/postgres.rs

@@ -201,29 +201,25 @@ impl Database for PostgresBackend {
         culprit: Option<&str>,
         level: Level,
     ) -> Result<Issue> {
-        let level_str = level_to_str(level);
-        let id = new_id();
-        let now = now_rfc3339();
-
-        // INSERT ... ON CONFLICT DO UPDATE — returns the row in both cases.
-        let row = sqlx::query_as::<_, IssueRow>(
-            "INSERT INTO issues (id, project_id, fingerprint, title, culprit, status, level, count, user_count, first_seen, last_seen) \
-             VALUES ($1, $2, $3, $4, $5, 'unresolved', $6, 1, 0, $7, $7) \
-             ON CONFLICT (project_id, fingerprint) DO UPDATE SET \
-                 count = issues.count + 1, last_seen = $7, level = $6 \
-             RETURNING id, project_id, fingerprint, title, culprit, status, level, count, user_count, first_seen, last_seen",
-        )
-        .bind(&id)
-        .bind(project_id)
-        .bind(fingerprint)
-        .bind(title)
-        .bind(culprit)
-        .bind(&level_str)
-        .bind(&now)
-        .fetch_one(&self.pool)
-        .await?;
+        let mut conn = self.pool.acquire().await?;
+        upsert_issue_conn(&mut conn, project_id, fingerprint, title, culprit, level).await
+    }
 
-        Ok(row.into())
+    async fn record_event(
+        &self,
+        project_id: &str,
+        fingerprint: &str,
+        title: &str,
+        culprit: Option<&str>,
+        level: Level,
+        event_data: &str,
+    ) -> Result<Issue> {
+        // One transaction: if the event insert fails the issue bump rolls back.
+        let mut tx = self.pool.begin().await?;
+        let issue = upsert_issue_conn(&mut tx, project_id, fingerprint, title, culprit, level).await?;
+        insert_event_conn(&mut tx, &issue.id, project_id, event_data).await?;
+        tx.commit().await?;
+        Ok(issue)
     }
 
     async fn get_issue(&self, issue_id: &str) -> Result<Option<Issue>> {
@@ -321,17 +317,8 @@ impl Database for PostgresBackend {
     // ── Events ─────────────────────────────────────────────────────────
 
     async fn insert_event(&self, issue_id: &str, project_id: &str, event_data: &str) -> Result<String> {
-        let id = new_id();
-        let now = now_rfc3339();
-        sqlx::query("INSERT INTO events (id, issue_id, project_id, data, received_at) VALUES ($1, $2, $3, $4, $5)")
-            .bind(&id)
-            .bind(issue_id)
-            .bind(project_id)
-            .bind(event_data)
-            .bind(&now)
-            .execute(&self.pool)
-            .await?;
-        Ok(id)
+        let mut conn = self.pool.acquire().await?;
+        insert_event_conn(&mut conn, issue_id, project_id, event_data).await
     }
 
     async fn list_events(&self, issue_id: &str, limit: i64, offset: i64) -> Result<Vec<StoredEvent>> {
@@ -843,20 +830,27 @@ impl Database for PostgresBackend {
     // ── Retention ──────────────────────────────────────────────────────
 
     async fn purge_old_events(&self, days: i64) -> Result<u64> {
-        let cutoff = (chrono::Utc::now() - chrono::Duration::days(days)).to_rfc3339();
+        let cutoff = cutoff_rfc3339_days(days);
         let result = sqlx::query("DELETE FROM events WHERE received_at < $1").bind(&cutoff).execute(&self.pool).await?;
         Ok(result.rows_affected())
     }
 
-    async fn purge_orphan_issues(&self) -> Result<()> {
-        sqlx::query("DELETE FROM issues WHERE id NOT IN (SELECT DISTINCT issue_id FROM events)")
-            .execute(&self.pool)
-            .await?;
+    async fn purge_orphan_issues(&self, days: i64) -> Result<()> {
+        // Single atomic statement. `last_seen < cutoff` protects a freshly
+        // upserted issue whose event is not yet inserted; resolved and ignored
+        // issues are kept as history.
+        sqlx::query(
+            "DELETE FROM issues WHERE last_seen < $1 AND status NOT IN ('resolved', 'ignored') \
+             AND NOT EXISTS (SELECT 1 FROM events WHERE events.issue_id = issues.id)",
+        )
+        .bind(cutoff_rfc3339_days(days))
+        .execute(&self.pool)
+        .await?;
         Ok(())
     }
 
     async fn purge_stale_auth_attempts(&self) -> Result<()> {
-        let cutoff = (chrono::Utc::now() - chrono::Duration::days(30)).to_rfc3339();
+        let cutoff = cutoff_rfc3339_days(30);
         sqlx::query("DELETE FROM auth_attempts WHERE created_at < $1").bind(&cutoff).execute(&self.pool).await?;
         Ok(())
     }
@@ -1072,3 +1066,57 @@ impl Database for PostgresBackend {
         Ok(row.map(Into::into))
     }
 }
+
+/// Atomic issue upsert on an existing connection/transaction.
+async fn upsert_issue_conn(
+    conn: &mut sqlx::PgConnection,
+    project_id: &str,
+    fingerprint: &str,
+    title: &str,
+    culprit: Option<&str>,
+    level: Level,
+) -> Result<Issue> {
+    let level_str = level_to_str(level);
+    let id = new_id();
+    let now = now_rfc3339();
+
+    // INSERT ... ON CONFLICT DO UPDATE returns the row in both cases.
+    let row = sqlx::query_as::<_, IssueRow>(
+        "INSERT INTO issues (id, project_id, fingerprint, title, culprit, status, level, count, user_count, first_seen, last_seen) \
+         VALUES ($1, $2, $3, $4, $5, 'unresolved', $6, 1, 0, $7, $7) \
+         ON CONFLICT (project_id, fingerprint) DO UPDATE SET \
+             count = issues.count + 1, last_seen = $7, level = $6 \
+         RETURNING id, project_id, fingerprint, title, culprit, status, level, count, user_count, first_seen, last_seen",
+    )
+    .bind(&id)
+    .bind(project_id)
+    .bind(fingerprint)
+    .bind(title)
+    .bind(culprit)
+    .bind(&level_str)
+    .bind(&now)
+    .fetch_one(&mut *conn)
+    .await?;
+
+    Ok(row.into())
+}
+
+/// Insert an event row on an existing connection/transaction.
+async fn insert_event_conn(
+    conn: &mut sqlx::PgConnection,
+    issue_id: &str,
+    project_id: &str,
+    event_data: &str,
+) -> Result<String> {
+    let id = new_id();
+    let now = now_rfc3339();
+    sqlx::query("INSERT INTO events (id, issue_id, project_id, data, received_at) VALUES ($1, $2, $3, $4, $5)")
+        .bind(&id)
+        .bind(issue_id)
+        .bind(project_id)
+        .bind(event_data)
+        .bind(&now)
+        .execute(&mut *conn)
+        .await?;
+    Ok(id)
+}