diff --git a/app/src/components/notifications/NotificationCenter.tsx b/app/src/components/notifications/NotificationCenter.tsx index d5674998c..5bb7d6135 100644 --- a/app/src/components/notifications/NotificationCenter.tsx +++ b/app/src/components/notifications/NotificationCenter.tsx @@ -30,6 +30,9 @@ const NotificationCenter = () => { // All providers seen across unfiltered loads — kept separate so the filter // pill row doesn't collapse when a provider filter is active. const [allProviders, setAllProviders] = useState([]); + const visibleItems = items.filter( + n => n.status !== 'dismissed' && (!selectedProvider || n.provider === selectedProvider) + ); // Fetch on mount and when provider filter changes. useEffect(() => { @@ -80,10 +83,10 @@ const NotificationCenter = () => { }; // Unread count scoped to the currently displayed (filtered) items. - const filteredUnreadCount = items.filter(n => n.status === 'unread').length; + const filteredUnreadCount = visibleItems.filter(n => n.status === 'unread').length; const handleMarkAllRead = async () => { - const unreadIds = items.filter(n => n.status === 'unread').map(n => n.id); + const unreadIds = visibleItems.filter(n => n.status === 'unread').map(n => n.id); for (const id of unreadIds) { dispatch(markIntegrationRead(id)); try { @@ -158,7 +161,7 @@ const NotificationCenter = () => { )} - {!loading && !error && items.length === 0 && ( + {!loading && !error && visibleItems.length === 0 && (
{
)} - {!loading && !error && items.length > 0 && ( + {!loading && !error && visibleItems.length > 0 && (
- {items.map(n => ( + {visibleItems.map(n => ( = {} +): IntegrationNotification => ({ + id: 'n-1', + provider: 'slack', + title: 'Test notification', + body: 'Hello', + raw_payload: {}, + status: 'unread', + received_at: '2026-04-23T00:00:00Z', + ...overrides, +}); + +const baseState = notificationReducer(undefined, { type: '@@init' }); +const initialState = { + ...baseState, + integrationItems: [], + integrationUnreadCount: 0, + integrationLoading: false, + integrationError: null, +}; + +describe('notificationSlice — dismissIntegrationNotification', () => { + it('marks unread notifications as dismissed and decrements unread count', () => { + const unread = makeNotification({ id: 'n-unread', status: 'unread' }); + const loaded = notificationReducer( + initialState, + setIntegrationNotifications({ items: [unread], unread_count: 1 }) + ); + + const state = notificationReducer(loaded, dismissIntegrationNotification('n-unread')); + + expect(state.integrationItems[0].status).toBe('dismissed'); + expect(state.integrationUnreadCount).toBe(0); + }); + + it('marks read notifications as dismissed without changing unread count', () => { + const read = makeNotification({ id: 'n-read', status: 'read' }); + const loaded = notificationReducer( + initialState, + setIntegrationNotifications({ items: [read], unread_count: 0 }) + ); + + const state = notificationReducer(loaded, dismissIntegrationNotification('n-read')); + + expect(state.integrationItems[0].status).toBe('dismissed'); + expect(state.integrationUnreadCount).toBe(0); + }); + + it('is a no-op when the id does not exist', () => { + const item = makeNotification({ id: 'n-1', status: 'unread' }); + const loaded = notificationReducer( + initialState, + setIntegrationNotifications({ items: [item], unread_count: 1 }) + ); + + const state = notificationReducer(loaded, dismissIntegrationNotification('does-not-exist')); + + expect(state.integrationItems[0].status).toBe('unread'); + expect(state.integrationUnreadCount).toBe(1); + }); + + it('does not decrement unread count when dismissing an already dismissed item', () => { + const dismissed = makeNotification({ id: 'n-dismissed', status: 'dismissed' }); + const loaded = notificationReducer( + initialState, + setIntegrationNotifications({ items: [dismissed], unread_count: 0 }) + ); + + const state = notificationReducer(loaded, dismissIntegrationNotification('n-dismissed')); + + expect(state.integrationItems[0].status).toBe('dismissed'); + expect(state.integrationUnreadCount).toBe(0); + }); +}); diff --git a/src/openhuman/notifications/store.rs b/src/openhuman/notifications/store.rs index 987f272ba..42b09ee70 100644 --- a/src/openhuman/notifications/store.rs +++ b/src/openhuman/notifications/store.rs @@ -118,62 +118,71 @@ pub fn insert(config: &Config, n: &IntegrationNotification) -> Result<()> { /// Returns `true` when inserted, `false` when skipped as duplicate. pub fn insert_if_not_recent(config: &Config, n: &IntegrationNotification) -> Result { with_connection(config, |conn| { - let tx = conn - .unchecked_transaction() + conn.execute_batch("BEGIN IMMEDIATE") .context("[notifications::store] begin insert_if_not_recent tx failed")?; - let count: i64 = match n.account_id.as_deref() { - Some(aid) => tx.query_row( - "SELECT COUNT(*) FROM integration_notifications - WHERE provider = ?1 AND account_id = ?2 - AND title = ?3 AND body = ?4 - AND unixepoch(received_at) >= unixepoch('now', '-60 seconds')", - params![&n.provider, aid, &n.title, &n.body], - |row| row.get(0), - ), - None => tx.query_row( - "SELECT COUNT(*) FROM integration_notifications - WHERE provider = ?1 AND account_id IS NULL - AND title = ?2 AND body = ?3 - AND unixepoch(received_at) >= unixepoch('now', '-60 seconds')", - params![&n.provider, &n.title, &n.body], - |row| row.get(0), - ), - } - .context("[notifications::store] insert_if_not_recent dedup query failed")?; + let result: Result = (|| { + let count: i64 = match n.account_id.as_deref() { + Some(aid) => conn.query_row( + "SELECT COUNT(*) FROM integration_notifications + WHERE provider = ?1 AND account_id = ?2 + AND title = ?3 AND body = ?4 + AND unixepoch(received_at) >= unixepoch('now', '-60 seconds')", + params![&n.provider, aid, &n.title, &n.body], + |row| row.get(0), + ), + None => conn.query_row( + "SELECT COUNT(*) FROM integration_notifications + WHERE provider = ?1 AND account_id IS NULL + AND title = ?2 AND body = ?3 + AND unixepoch(received_at) >= unixepoch('now', '-60 seconds')", + params![&n.provider, &n.title, &n.body], + |row| row.get(0), + ), + } + .context("[notifications::store] insert_if_not_recent dedup query failed")?; - if count > 0 { - tx.commit() - .context("[notifications::store] commit duplicate tx failed")?; - return Ok(false); + if count > 0 { + return Ok(false); + } + + conn.execute( + "INSERT INTO integration_notifications + (id, provider, account_id, title, body, raw_payload, + importance_score, triage_action, triage_reason, status, + received_at, scored_at) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)", + params![ + n.id, + n.provider, + n.account_id, + n.title, + n.body, + n.raw_payload.to_string(), + n.importance_score, + n.triage_action, + n.triage_reason, + n.status.as_str(), + n.received_at.to_rfc3339(), + n.scored_at.map(|t| t.to_rfc3339()), + ], + ) + .context("[notifications::store] insert_if_not_recent insert failed")?; + + Ok(true) + })(); + + if result.is_ok() { + conn.execute_batch("COMMIT") + .context("[notifications::store] commit insert_if_not_recent tx failed")?; + } else if let Err(rollback_err) = conn.execute_batch("ROLLBACK") { + tracing::warn!( + error = %rollback_err, + "[notifications::store] rollback insert_if_not_recent tx failed" + ); } - tx.execute( - "INSERT INTO integration_notifications - (id, provider, account_id, title, body, raw_payload, - importance_score, triage_action, triage_reason, status, - received_at, scored_at) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)", - params![ - n.id, - n.provider, - n.account_id, - n.title, - n.body, - n.raw_payload.to_string(), - n.importance_score, - n.triage_action, - n.triage_reason, - n.status.as_str(), - n.received_at.to_rfc3339(), - n.scored_at.map(|t| t.to_rfc3339()), - ], - ) - .context("[notifications::store] insert_if_not_recent insert failed")?; - - tx.commit() - .context("[notifications::store] commit insert_if_not_recent tx failed")?; - Ok(true) + result }) } @@ -581,6 +590,7 @@ fn row_to_notification(row: &rusqlite::Row<'_>) -> Result Config { @@ -683,6 +693,34 @@ mod tests { assert!(insert_if_not_recent(&config, &fresh_same_content).unwrap()); } + #[test] + fn insert_if_not_recent_is_atomic_under_concurrent_calls() { + let dir = TempDir::new().unwrap(); + let config = Arc::new(test_config(&dir)); + let gate = Arc::new(Barrier::new(3)); + + let run = |id: &'static str, gate: Arc, config: Arc| { + std::thread::spawn(move || { + let n = sample_notification(id, "slack"); + gate.wait(); + insert_if_not_recent(&config, &n) + }) + }; + + let t1 = run("race-a", Arc::clone(&gate), Arc::clone(&config)); + let t2 = run("race-b", Arc::clone(&gate), Arc::clone(&config)); + + gate.wait(); + let inserted_1 = t1.join().unwrap().unwrap(); + let inserted_2 = t2.join().unwrap().unwrap(); + + let inserted_total = usize::from(inserted_1) + usize::from(inserted_2); + assert_eq!(inserted_total, 1); + + let items = list(&config, 10, 0, Some("slack"), None).unwrap(); + assert_eq!(items.len(), 1); + } + #[test] fn exists_recent_rejects_expired_notification() { let dir = TempDir::new().unwrap(); diff --git a/tests/json_rpc_e2e.rs b/tests/json_rpc_e2e.rs index 375b475ee..127312457 100644 --- a/tests/json_rpc_e2e.rs +++ b/tests/json_rpc_e2e.rs @@ -175,11 +175,31 @@ fn mock_upstream_router() -> Router { if let Some(model) = body.get("model").and_then(Value::as_str) { with_chat_completion_models(|models| models.push(model.to_string())); } + let is_triage_turn = body + .get("messages") + .and_then(Value::as_array) + .map(|messages| { + messages.iter().any(|m| { + m.get("content") + .and_then(Value::as_str) + .is_some_and(|content| { + content.contains("SOURCE: ") + && content.contains("DISPLAY_LABEL: ") + && content.contains("PAYLOAD:") + }) + }) + }) + .unwrap_or(false); + let content = if is_triage_turn { + "{\"action\":\"react\",\"reason\":\"e2e triage mock\"}" + } else { + "Hello from e2e mock agent" + }; Json(json!({ "choices": [{ "message": { "role": "assistant", - "content": "Hello from e2e mock agent" + "content": content } }] }))