watchlist add: event only if inserted; adr 0015 jetstream/relay accuracy; drop vacuous assert

crates/adapters/sqlite/src/repositories/watchlist/mod.rs

@@ -39,7 +39,7 @@ impl WatchlistRepository for SqliteWatchlistRepository {
             .begin()
             .await
             .map_err(crate::shared::map_sqlx_error)?;
-        sqlx::query(
+        let inserted = sqlx::query(
             "INSERT OR IGNORE INTO watchlist_entries (id, user_id, movie_id, added_at) \
              VALUES (?, ?, ?, ?)",
         )
@@ -49,8 +49,13 @@ impl WatchlistRepository for SqliteWatchlistRepository {
         .bind(&added_at)
         .execute(&mut *tx)
         .await
-        .map_err(crate::shared::map_sqlx_error)?;
+        .map_err(crate::shared::map_sqlx_error)?
+        .rows_affected()
+            > 0;
+        // A repeat add changed nothing, so it announces nothing.
+        if inserted {
             enqueue_events(&mut tx, events).await?;
+        }
         tx.commit().await.map_err(crate::shared::map_sqlx_error)
     }
 

crates/adapters/sqlite/tests/repositories/outbox_writes_test.rs

@@ -200,6 +200,24 @@ async fn adding_a_watchlist_entry_writes_the_entry_and_its_event_in_one_commit()
     assert_eq!(count(&pool, "event_queue").await, 1);
 }
 
+#[tokio::test]
+async fn re_adding_a_watchlist_entry_enqueues_no_second_event() {
+    let (pool, user_id, movie) = watchlist_fixture().await;
+    let repo = SqliteWatchlistRepository::new(pool.clone());
+    let first = WatchlistEntry::new(user_id.clone(), movie.id().clone());
+    repo.add(&first, &[watchlist_added(&first, &movie)])
+        .await
+        .unwrap();
+    let again = WatchlistEntry::new(user_id, movie.id().clone());
+
+    repo.add(&again, &[watchlist_added(&again, &movie)])
+        .await
+        .unwrap();
+
+    assert_eq!(count(&pool, "watchlist_entries").await, 1);
+    assert_eq!(count(&pool, "event_queue").await, 1, "no phantom event");
+}
+
 #[tokio::test]
 async fn a_failing_enqueue_rolls_back_the_watchlist_add() {
     let (pool, user_id, movie) = watchlist_fixture().await;

crates/application/tests/diary/review_logger_test.rs

@@ -270,7 +270,6 @@ async fn a_refused_save_propagates_and_records_nothing() {
         .await;
 
     assert!(result.is_err());
-    assert_eq!(movies.review_count(), 0);
     assert!(movies.recorded_events().is_empty());
 }
 

crates/application/tests/watchlist/add_test.rs

@@ -54,6 +54,11 @@ async fn adding_a_movie_already_on_the_watchlist_is_idempotent() {
     add::execute(&deps, make_cmd()).await.unwrap();
     add::execute(&deps, make_cmd()).await.unwrap();
 
+    assert_eq!(
+        movies.recorded_events().len(),
+        1,
+        "the repeat inserted nothing, so it announces nothing"
+    );
     assert_eq!(
         watchlist.watchlist_count(),
         1,

crates/bootstrap/tests/outbox_relay_test.rs

@@ -38,7 +38,7 @@ impl EventRelaySink for FakeSink {
         }
         let outage = self
             .fail_first
-            .fetch_update(
+            .try_update(
                 std::sync::atomic::Ordering::SeqCst,
                 std::sync::atomic::Ordering::SeqCst,
                 |n| n.checked_sub(1),

crates/domain/src/testing/fakes/store_watchlist.rs

@@ -16,15 +16,23 @@ use super::InMemoryStore;
 
 #[async_trait]
 impl WatchlistRepository for InMemoryStore {
-    /// `INSERT OR IGNORE` on UNIQUE(user_id, movie_id): adding twice is a no-op.
+    /// `INSERT OR IGNORE` on UNIQUE(user_id, movie_id): adding twice is a no-op
+    /// and, like the adapter, records events only for the row it inserted.
     async fn add(&self, entry: &WatchlistEntry, events: &[DomainEvent]) -> Result<(), DomainError> {
         let key = (entry.user_id.value(), entry.movie_id.value());
-        self.watchlist
+        let inserted = {
-            .write()
+            let mut entries = self.watchlist.write().unwrap();
-            .unwrap()
+            match entries.entry(key) {
-            .entry(key)
+                std::collections::hash_map::Entry::Vacant(slot) => {
-            .or_insert_with(|| entry.clone());
+                    slot.insert(entry.clone());
+                    true
+                }
+                std::collections::hash_map::Entry::Occupied(_) => false,
+            }
+        };
+        if inserted {
             self.record_events(events);
+        }
         Ok(())
     }
 

docs/adr/0015-transactional-outbox.md

@@ -25,8 +25,17 @@ Use cases never open transactions; the adapter does.
 **The server never talks to NATS.** With `EVENT_BUS_BACKEND=nats` the server still only inserts
 outbox rows. The worker runs a relay that claims pending rows and publishes them to JetStream with
 `Nats-Msg-Id` set to the event id, then marks the row done. With `db` the worker consumes the rows
-directly. `max_deliver` on the JetStream consumer equals `EVENT_QUEUE_MAX_ATTEMPTS`, so both
+directly. `max_deliver` on the JetStream consumer equals `EVENT_QUEUE_MAX_ATTEMPTS`, and a JetStream
-backends dead-letter after the same number of attempts.
+nak uses the same backoff as the SQLite queue (`domain::events::retry_backoff`: 5 s doubling per
+attempt, capped at 300 s), so both backends give up after the same number of deliveries on the same
+schedule. They differ in what giving up means. SQLite marks the row `dead_lettered`. JetStream has no
+DLQ: after `max_deliver` the message is no longer delivered but stays in the stream (until stream
+limits evict it). The consumer logs `event exhausted max deliveries` at error level with the event id
+on the final nak. To find such messages, watch the JetStream max-deliveries advisory
+(`$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.<stream>.<consumer>`) or inspect the stream
+(`nats stream view <stream>`) for sequences the consumer has not acked. The relay never spends
+attempts on transport failures: when the bus is unreachable it defers the row with growing backoff and
+pauses, so an outage cannot dead-letter rows.
 
 **Three dedup layers.**
 1. *Broker.* JetStream drops a repeated `Nats-Msg-Id` inside `duplicate_window`, set explicitly to 1