amazon: replace fire-and-forget listener notifications with ack-based retry

Dispatching an event to the listener previously happened from an
un-awaited goroutine, with only a log line on failure - once
processed_at was set, a dropped or failed notification was permanently
and silently lost, with no way to tell it had happened.

Replaces the processed/processed_successfully booleans with a
notified_at/processed_at pair (migration 000031): the dispatcher sets
notified_at and hands the event to MockEventListener.Notify, which now
also receives an ack callback the listener calls whenever it's truly
done, synchronously or arbitrarily later. Anything still "notified" but
unacked past notifyRetryAfter (a tunable field, not a stored
per-row timestamp) gets notified again on the dispatcher's normal
poll/reactive loop - no new retry mechanism needed. ack is idempotent,
since a late ack from an earlier attempt and one from a retry can both
eventually fire for the same event.

Dispatch is deferred until after the transaction that recorded
notified_at has actually committed, so ack's independent write can't
race a still-open transaction it implicitly depends on being visible.

The real SSE listener (server/sse/db_event_publisher.go) acks inline,
since its work is synchronous.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XEDaCB7C2NEBgyvqEtZuxY

database_migrations/000031_amazon_event_notify_ack.down.sql

@@ -0,0 +1,18 @@
+BEGIN;
+
+ALTER TABLE mock.shop_amazon_events
+	ADD COLUMN processed BOOLEAN DEFAULT FALSE,
+	ADD COLUMN processed_successfully BOOLEAN DEFAULT FALSE;
+
+UPDATE mock.shop_amazon_events
+	SET processed = (processed_at IS NOT NULL),
+		processed_successfully = (processed_at IS NOT NULL);
+
+DROP INDEX IF EXISTS mock.shop_amazon_events_by_timestamp_unprocessed;
+CREATE INDEX shop_amazon_events_by_timestamp_unprocessed ON mock.shop_amazon_events (shop_id, event_timestamp) WHERE NOT processed;
+
+ALTER TABLE mock.shop_amazon_events
+	DROP COLUMN notified_at,
+	DROP COLUMN processed_at;
+
+COMMIT;

database_migrations/000031_amazon_event_notify_ack.up.sql

@@ -0,0 +1,29 @@
+BEGIN;
+
+-- Replaces the boolean processed/processed_successfully pair with a
+-- three-state model driven by two nullable timestamps:
+--   unprocessed: notified_at IS NULL
+--   notified:    notified_at IS NOT NULL AND processed_at IS NULL
+--   processed:   processed_at IS NOT NULL
+-- The dispatcher sets notified_at each time it hands an event to the
+-- listener (initially, and again on retry if the listener never acks).
+-- Only the listener's ack, via its own timestamped write, sets
+-- processed_at - it's a separate write specifically so it never races the
+-- transaction that recorded notified_at.
+
+ALTER TABLE mock.shop_amazon_events
+	ADD COLUMN notified_at TIMESTAMPTZ,
+	ADD COLUMN processed_at TIMESTAMPTZ;
+
+UPDATE mock.shop_amazon_events
+	SET processed_at = NOW()
+	WHERE processed;
+
+ALTER TABLE mock.shop_amazon_events
+	DROP COLUMN processed,
+	DROP COLUMN processed_successfully;
+
+DROP INDEX IF EXISTS mock.shop_amazon_events_by_timestamp_unprocessed;
+CREATE INDEX shop_amazon_events_by_timestamp_unprocessed ON mock.shop_amazon_events (shop_id, event_timestamp) WHERE processed_at IS NULL;
+
+COMMIT;

domains/amazon/mock.go

@@ -20,24 +20,42 @@ type (
 		db       *pgxpool.Pool
 		listener MockEventListener
 
+		// notifyRetryAfter is how long an event can sit "notified" (handed
+		// to the listener) without an ack before the dispatcher notifies
+		// it again. Deliberately a field, not a stored notify_again_at
+		// column, so the cadence can vary (or be tuned in tests) without
+		// touching any row.
+		notifyRetryAfter time.Duration
+
 		ready     chan struct{}
 		readyOnce sync.Once
 	}
 
+	// MockEventListener is handed events as they arrive. Notify should
+	// return promptly - any real work (including calling out to another
+	// system) can happen after it returns - but it must eventually call
+	// ack exactly once, when the event has been fully handled. Until ack
+	// is called, the dispatcher considers the event merely "notified"
+	// and will call Notify again after notifyRetryAfter, so ack may be
+	// called more than once total across retries; only the first call
+	// has any effect; further calls are safe (see (*Mocks).ack).
 	MockEventListener interface {
-		Notify(context.Context, raw_events.Event) error
+		Notify(ctx context.Context, e raw_events.Event, ack func(context.Context) error) error
 	}
 )
 
 const (
 	eventChannelName = "mock_shop_amazon_event_inserted"
+
+	defaultNotifyRetryAfter = 30 * time.Second
 )
 
 func NewMocks(log *logging.Logger, db *pgxpool.Pool) *Mocks {
 	return &Mocks{
-		log:   log,
-		db:    db,
-		ready: make(chan struct{}),
+		log:              log,
+		db:               db,
+		notifyRetryAfter: defaultNotifyRetryAfter,
+		ready:            make(chan struct{}),
 	}
 }
 
@@ -46,6 +64,14 @@ func (m *Mocks) SetListener(l MockEventListener) *Mocks {
 	return m
 }
 
+// WithNotifyRetryAfter overrides how long an event can sit "notified"
+// without an ack before being notified again. Mainly useful for tests that
+// don't want to wait out the default.
+func (m *Mocks) WithNotifyRetryAfter(d time.Duration) *Mocks {
+	m.notifyRetryAfter = d
+	return m
+}
+
 // Ready returns a channel that's closed once ProcessEvents has registered
 // its Postgres LISTEN and is actively watching for notifications. Callers
 // that need to know the reactive path is live - tests in particular -
@@ -125,10 +151,18 @@ func (m *Mocks) listenForNotifications(ctx context.Context) (<-chan struct{}, <-
 	return ch, errCh
 }
 
+// processUnprocessedEvents finds every event that's either brand new or
+// has been "notified" for longer than notifyRetryAfter without an ack, and
+// (re)dispatches each to the listener. It never blocks on the listener:
+// notified_at is committed first, and only then - after the transaction
+// that recorded it has actually committed - is the listener told, from a
+// goroutine that isn't tied to this function's lifetime.
 func (m *Mocks) processUnprocessedEvents(ctx context.Context) error {
 	for done := false; !done; {
+		var toDispatch []raw_events.Event
+
 		err := pgx.BeginFunc(ctx, m.db, func(tx pgx.Tx) error {
-			rows, err := m.db.Query(
+			rows, err := tx.Query(
 				ctx,
 				`
 					SELECT
@@ -138,12 +172,16 @@ func (m *Mocks) processUnprocessedEvents(ctx context.Context) error {
 					FROM
 						mock.shop_amazon_events
 					WHERE
-						NOT processed
+						processed_at IS NULL
+						AND (notified_at IS NULL OR notified_at < @retry_after)
 					ORDER BY
 						shop_id, event_timestamp
 					LIMIT
 						100
 				`,
+				pgx.NamedArgs{
+					"retry_after": time.Now().Add(-m.notifyRetryAfter),
+				},
 			)
 			if err != nil {
 				return fmt.Errorf("failed to to perform query: %w", err)
@@ -159,18 +197,22 @@ func (m *Mocks) processUnprocessedEvents(ctx context.Context) error {
 			}
 
 			for _, e := range evts {
-				if err := m.processEvent(ctx, tx, raw_events.Event{
+				ev := raw_events.Event{
 					Platform:       string(accounts.Amazon),
 					StoreID:        e.Shop_id,
 					EventID:        e.Event_id,
 					EventTimestamp: e.Event_timestamp,
-				}); err != nil {
-					return fmt.Errorf("failed to process event: %w", err)
 				}
+
+				if err := markNotified(ctx, tx, ev); err != nil {
+					return fmt.Errorf("failed to mark event notified: %w", err)
+				}
+
+				toDispatch = append(toDispatch, ev)
 			}
 
 			if done = len(evts) == 0; !done {
-				m.log.Infof("processed %d events", len(evts))
+				m.log.Infof("notified listener for %d events", len(evts))
 			}
 
 			return nil
@@ -178,22 +220,42 @@ func (m *Mocks) processUnprocessedEvents(ctx context.Context) error {
 		if err != nil {
 			return err
 		}
+
+		for _, e := range toDispatch {
+			m.dispatch(ctx, e)
+		}
 	}
 
 	return nil
 }
 
-func (m *Mocks) processEvent(ctx context.Context, tx pgx.Tx, e raw_events.Event) error {
-	m.log.Debugf("processing (mock) event: %#v", e)
+// dispatch hands e to the configured listener without blocking. The
+// listener may call ack synchronously (e.g. inline once its own work is
+// done) or from elsewhere entirely, arbitrarily later - ack does its own
+// independent, idempotent write, so it's never at risk of racing (or being
+// silently lost to) this call's context being cancelled.
+func (m *Mocks) dispatch(ctx context.Context, e raw_events.Event) {
+	if m.listener == nil {
+		return
+	}
 
+	go func() {
+		if err := m.listener.Notify(ctx, e, func(ackCtx context.Context) error {
+			return ack(ackCtx, m.db, e)
+		}); err != nil {
+			m.log.Errorf("error incurred by event listener: %v", err)
+		}
+	}()
+}
+
+func markNotified(ctx context.Context, tx pgx.Tx, e raw_events.Event) error {
 	_, err := tx.Exec(
 		ctx,
 		`
 			UPDATE
 				mock.shop_amazon_events
 			SET
-				processed = true,
-				processed_successfully = true
+				notified_at = NOW()
 			WHERE
 				shop_id = @shop_id
 				AND event_id = @event_id
@@ -205,17 +267,34 @@ func (m *Mocks) processEvent(ctx context.Context, tx pgx.Tx, e raw_events.Event)
 			"event_timestamp": e.EventTimestamp,
 		},
 	)
+	return err
+}
+
+// ack records that e has been fully handled. It's idempotent - calling it
+// more than once (e.g. because a retried notification also eventually acks)
+// just leaves processed_at at whenever it was first set.
+func ack(ctx context.Context, db *pgxpool.Pool, e raw_events.Event) error {
+	_, err := db.Exec(
+		ctx,
+		`
+			UPDATE
+				mock.shop_amazon_events
+			SET
+				processed_at = NOW()
+			WHERE
+				shop_id = @shop_id
+				AND event_id = @event_id
+				AND event_timestamp = @event_timestamp
+				AND processed_at IS NULL
+		`,
+		pgx.NamedArgs{
+			"shop_id":         e.StoreID,
+			"event_id":        e.EventID,
+			"event_timestamp": e.EventTimestamp,
+		},
+	)
 	if err != nil {
 		return fmt.Errorf("failed to perform query: %w", err)
 	}
-
-	if m.listener != nil {
-		go func() {
-			if err := m.listener.Notify(ctx, e); err != nil {
-				m.log.Errorf("error incurred by event listener: %v", err)
-			}
-		}()
-	}
-
 	return nil
 }

domains/amazon/mock_test.go

@@ -14,25 +14,40 @@ import (
 )
 
 // notifySpy is a MockEventListener that records every event it's notified
-// about. (*Mocks).processEvent fires notifications from a goroutine without
-// waiting for them, so tests need to synchronize on notified rather than
-// asserting on the event list immediately.
+// about, along with the ack callback it was given. By default it acks
+// immediately (autoAck=true), matching a well-behaved listener; tests
+// exercising the retry path set autoAck=false to simulate a listener that
+// received the notification but hasn't finished yet, and call ackAt
+// explicitly once it "finishes".
 type notifySpy struct {
-	mu       sync.Mutex
-	events   []raw_events.Event
-	notified chan struct{}
+	mu             sync.Mutex
+	events         []raw_events.Event
+	acks           []func(context.Context) error
+	completedCount int // Notify calls that have returned, including any auto-ack
+	autoAck        bool
 }
 
 func newNotifySpy() *notifySpy {
-	return &notifySpy{notified: make(chan struct{}, 1000)}
+	return &notifySpy{autoAck: true}
 }
 
-func (s *notifySpy) Notify(_ context.Context, e raw_events.Event) error {
+func (s *notifySpy) Notify(ctx context.Context, e raw_events.Event, ack func(context.Context) error) error {
 	s.mu.Lock()
 	s.events = append(s.events, e)
+	s.acks = append(s.acks, ack)
+	autoAck := s.autoAck
 	s.mu.Unlock()
-	s.notified <- struct{}{}
-	return nil
+
+	var ackErr error
+	if autoAck {
+		ackErr = ack(ctx)
+	}
+
+	s.mu.Lock()
+	s.completedCount++
+	s.mu.Unlock()
+
+	return ackErr
 }
 
 func (s *notifySpy) eventsSnapshot() []raw_events.Event {
@@ -41,15 +56,43 @@ func (s *notifySpy) eventsSnapshot() []raw_events.Event {
 	return append([]raw_events.Event(nil), s.events...)
 }
 
+func (s *notifySpy) setAutoAck(v bool) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.autoAck = v
+}
+
+// ackAt manually invokes the i-th recorded ack callback (0-indexed, in
+// Notify call order), simulating the listener finally finishing work it
+// had earlier only been notified about.
+func (s *notifySpy) ackAt(t *testing.T, i int) {
+	t.Helper()
+	s.mu.Lock()
+	ack := s.acks[i]
+	s.mu.Unlock()
+	if err := ack(context.Background()); err != nil {
+		t.Fatalf("ack() error = %v", err)
+	}
+}
+
+// waitForCount blocks until at least n Notify calls have fully completed
+// (including any auto-ack), cumulatively across the test - safe to call
+// more than once with increasing n.
 func (s *notifySpy) waitForCount(t *testing.T, n int, timeout time.Duration) {
 	t.Helper()
-	deadline := time.After(timeout)
-	for i := 0; i < n; i++ {
-		select {
-		case <-s.notified:
-		case <-deadline:
-			t.Fatalf("timed out after %v waiting for notification %d/%d (got %d so far)", timeout, i+1, n, len(s.eventsSnapshot()))
+	deadline := time.Now().Add(timeout)
+	for {
+		s.mu.Lock()
+		got := s.completedCount
+		s.mu.Unlock()
+
+		if got >= n {
+			return
 		}
+		if time.Now().After(deadline) {
+			t.Fatalf("timed out after %v waiting for %d total completed notifications (got %d so far)", timeout, n, got)
+		}
+		time.Sleep(5 * time.Millisecond)
 	}
 }
 
@@ -87,23 +130,39 @@ func insertRawAmazonEvents(t *testing.T, pool *pgxpool.Pool, shopID string, n in
 
 func isProcessed(t *testing.T, pool *pgxpool.Pool, shopID, eventID string) bool {
 	t.Helper()
-	var processed, successful bool
+	var processed bool
 	err := pool.QueryRow(context.Background(), `
-		SELECT processed, processed_successfully
+		SELECT processed_at IS NOT NULL
 		FROM mock.shop_amazon_events
 		WHERE shop_id = $1 AND event_id = $2
-	`, shopID, eventID).Scan(&processed, &successful)
+	`, shopID, eventID).Scan(&processed)
 	if err != nil {
 		t.Fatalf("failed to check processed state: %v", err)
 	}
-	return processed && successful
+	return processed
+}
+
+// isNotified reports whether the event has been notified (at least once)
+// but not yet acked/processed.
+func isNotified(t *testing.T, pool *pgxpool.Pool, shopID, eventID string) bool {
+	t.Helper()
+	var notified bool
+	err := pool.QueryRow(context.Background(), `
+		SELECT notified_at IS NOT NULL AND processed_at IS NULL
+		FROM mock.shop_amazon_events
+		WHERE shop_id = $1 AND event_id = $2
+	`, shopID, eventID).Scan(&notified)
+	if err != nil {
+		t.Fatalf("failed to check notified state: %v", err)
+	}
+	return notified
 }
 
 func countUnprocessed(t *testing.T, pool *pgxpool.Pool, shopID string) int {
 	t.Helper()
 	var n int
 	err := pool.QueryRow(context.Background(), `
-		SELECT count(*) FROM mock.shop_amazon_events WHERE shop_id = $1 AND NOT processed
+		SELECT count(*) FROM mock.shop_amazon_events WHERE shop_id = $1 AND processed_at IS NULL
 	`, shopID).Scan(&n)
 	if err != nil {
 		t.Fatalf("failed to count unprocessed events: %v", err)
@@ -137,11 +196,11 @@ func TestProcessUnprocessedEvents_ProcessesAllEventsAcrossBatches(t *testing.T)
 		t.Fatalf("processUnprocessedEvents() error = %v", err)
 	}
 
+	spy.waitForCount(t, n, 5*time.Second)
+
 	if got := countUnprocessed(t, pool, shopID); got != 0 {
 		t.Errorf("countUnprocessed() = %d, want 0 (all %d events should be processed across multiple 100-row batches)", got, n)
 	}
-
-	spy.waitForCount(t, n, 5*time.Second)
 }
 
 func TestProcessUnprocessedEvents_NoListenerConfigured(t *testing.T) {
@@ -158,8 +217,74 @@ func TestProcessUnprocessedEvents_NoListenerConfigured(t *testing.T) {
 		t.Fatalf("processUnprocessedEvents() error = %v", err)
 	}
 
+	// with no listener, nothing ever acks - the event is expected to stay
+	// "notified" forever, not silently marked processed.
+	if isProcessed(t, pool, shopID, "evt-1") {
+		t.Error("event was marked processed despite no listener being configured to ack it")
+	}
+	if !isNotified(t, pool, shopID, "evt-1") {
+		t.Error("event should be in the notified state after being handed off with no listener to ack it")
+	}
+}
+
+// TestProcessUnprocessedEvents_RetriesUnackedNotification is the core of
+// insight #2's fix: a listener that receives a notification but never acks
+// gets notified again after notifyRetryAfter, and once it does ack (however
+// late), the event settles into processed and stops being retried.
+func TestProcessUnprocessedEvents_RetriesUnackedNotification(t *testing.T) {
+	pool := testdb.Pool(t)
+	spy := newNotifySpy()
+	spy.setAutoAck(false)
+	m := NewMocks(testdb.Logger(), pool).
+		SetListener(spy).
+		WithNotifyRetryAfter(50 * time.Millisecond)
+	ctx := context.Background()
+
+	shopID := "test-shop-" + uuid.NewString()
+	cleanupShop(t, pool, shopID)
+
+	insertRawAmazonEvent(t, pool, shopID, "evt-1")
+
+	if err := m.processUnprocessedEvents(ctx); err != nil {
+		t.Fatalf("processUnprocessedEvents() (1st pass) error = %v", err)
+	}
+	spy.waitForCount(t, 1, 2*time.Second)
+
+	if isProcessed(t, pool, shopID, "evt-1") {
+		t.Fatal("event was marked processed despite the listener never acking")
+	}
+	if !isNotified(t, pool, shopID, "evt-1") {
+		t.Fatal("event should be in the notified state after the first dispatch")
+	}
+
+	// still within notifyRetryAfter: shouldn't be re-notified yet.
+	if err := m.processUnprocessedEvents(ctx); err != nil {
+		t.Fatalf("processUnprocessedEvents() (immediate 2nd pass) error = %v", err)
+	}
+	if got := len(spy.eventsSnapshot()); got != 1 {
+		t.Fatalf("listener was notified %d times before notifyRetryAfter elapsed, want 1", got)
+	}
+
+	time.Sleep(60 * time.Millisecond) // past notifyRetryAfter
+
+	if err := m.processUnprocessedEvents(ctx); err != nil {
+		t.Fatalf("processUnprocessedEvents() (3rd pass, after retry window) error = %v", err)
+	}
+	spy.waitForCount(t, 2, 2*time.Second)
+
+	// the listener "finishes" the first notification late, via the ack it
+	// was originally handed - not a fresh one from the retry.
+	spy.ackAt(t, 0)
+
 	if !isProcessed(t, pool, shopID, "evt-1") {
-		t.Error("event was not marked processed despite no listener being configured")
+		t.Fatal("event should be processed once any recorded ack for it is called")
+	}
+
+	if err := m.processUnprocessedEvents(ctx); err != nil {
+		t.Fatalf("processUnprocessedEvents() (4th pass, after ack) error = %v", err)
+	}
+	if got := len(spy.eventsSnapshot()); got != 2 {
+		t.Fatalf("listener was notified again after being acked: got %d calls, want 2", got)
 	}
 }
 

server/sse/db_event_publisher.go

@@ -25,7 +25,10 @@ func (q *Queue) NewDBEventPublisher(
 	}
 }
 
-func (p *DBEventPublisher) Notify(ctx context.Context, e raw_events.Event) error {
+// Notify publishes e to the SSE queue. The publish itself is quick and
+// synchronous, so it acks inline once it succeeds - there's no separate
+// async completion to wait for here.
+func (p *DBEventPublisher) Notify(ctx context.Context, e raw_events.Event, ack func(context.Context) error) error {
 	acctID, err := p.getAccountID(ctx, e)
 	if err != nil {
 		return fmt.Errorf("failed to get account id: %w", err)
@@ -40,5 +43,9 @@ func (p *DBEventPublisher) Notify(ctx context.Context, e raw_events.Event) error
 		return fmt.Errorf("failed to send sse event to listener: %w", err)
 	}
 
+	if err := ack(ctx); err != nil {
+		return fmt.Errorf("failed to ack event: %w", err)
+	}
+
 	return nil
 }

work-summaries/work-summary-Claude-2026-08-05-2017.md

@@ -0,0 +1,40 @@
+# Work Summary — 2026-08-05 20:17
+
+## Task
+Insight #2 from the `domains/amazon` design review: fire-and-forget listener notifications meant a dropped/failed SSE push was silently permanent, with the underlying event already marked `processed`. Fixed per a design worked out collaboratively with the user (not unilaterally chosen - this one had real tradeoffs).
+
+## Design (agreed with user before implementing)
+Non-blocking dispatch is a hard constraint - events come from external systems whose APIs may change unexpectedly, so the dispatcher must never wait on the listener's real work. The agreed shape:
+
+- Three states, driven by two nullable timestamps rather than a stored `notify_again_at`: `unprocessed` (`notified_at IS NULL`) → `notified` (`notified_at` set, `processed_at` still null) → `processed` (`processed_at` set).
+- The dispatcher sets `notified_at` and calls the listener; it never sets `processed_at` itself.
+- The listener signals completion via a callback passed into `Notify` itself: `Notify(ctx, e, ack func(context.Context) error) error`. `ack` can be called synchronously (fast listeners, like the real SSE one) or arbitrarily later from elsewhere (slow/async listeners) - exactly once is the contract, and extra calls are safe (see below).
+- If an event sits in `notified` past a retry threshold without being acked, the dispatcher's normal loop re-notifies it - reusing the existing poll/reactive machinery, no new loop. The threshold (`notifyRetryAfter`) is a Go-side field with a sensible default, not a stored per-row timestamp, so the cadence can change without touching any row (explicit user preference over storing `notify_again_at`).
+
+## Changes
+- `database_migrations/000031_amazon_event_notify_ack.{up,down}.sql`: drops `processed`/`processed_successfully` booleans on `mock.shop_amazon_events`, adds nullable `notified_at`/`processed_at` timestamps, updates the partial index accordingly. Applied to both the test DB and the real dev DB.
+- `domains/amazon/mock.go`:
+  - `MockEventListener.Notify` gained the `ack` parameter.
+  - `Mocks` gained a `notifyRetryAfter` field (default 30s) and `WithNotifyRetryAfter(d)` builder for tuning/tests.
+  - `processUnprocessedEvents`'s query now selects events that are either brand new or notified-but-stale (`processed_at IS NULL AND (notified_at IS NULL OR notified_at < retry_after)`).
+  - Restructured so the async dispatch to the listener happens **after** the transaction that recorded `notified_at` has committed, not from inside it - `ack`'s own independent write can never race a still-open transaction it implicitly depends on being visible.
+  - `ack` is idempotent (`... AND processed_at IS NULL` in its UPDATE), since a late ack from an earlier notification and a fresh one from a retry can both eventually fire for the same event.
+- `server/sse/db_event_publisher.go`: the real listener now acks inline right after a successful SSE publish - its work is synchronous, so there's no reason to defer completion.
+- `domains/amazon/mock_test.go`: updated `notifySpy` for the new signature (records ack callbacks, supports disabling auto-ack to simulate a listener that doesn't finish), updated `isProcessed`/added `isNotified` to check the new columns, and added `TestProcessUnprocessedEvents_RetriesUnackedNotification` - the core new-behavior test: a non-acking listener gets re-notified after the retry window (not before), and once *any* recorded ack for that event fires (including a stale one from an earlier attempt, not just the latest), the event settles into `processed` and stops being retried.
+
+## A test bug caught and fixed along the way (not production code)
+First run: `TestProcessUnprocessedEvents_ProcessesAllEventsAcrossBatches` failed (8/150 events still unprocessed) and the new retry test hung/timed out. Both were bugs in my own `notifySpy`, not the production code:
+- It signaled "notified" *before* calling the auto-ack, so a test could observe "all N notified" before all N acks had actually landed in the DB.
+- `waitForCount` drained a fixed number of channel signals per call instead of tracking a cumulative total, so calling it twice (once for 1, once for 2) double-counted and hung waiting for a signal that would never come.
+
+Fixed by replacing the channel with a cumulative `completedCount` incremented only after any auto-ack attempt returns, and made `waitForCount` poll that count (safe to call repeatedly with increasing thresholds).
+
+## Verification
+- `go build ./...` / `go vet ./...` clean.
+- `go test ./domains/amazon/... -v -race`: all 5 tests pass (4 existing + 1 new).
+- 10x repeated runs (`-count=1 -race`) with no flakes, ~1.3-1.5s each.
+- `make test` and `make test-against-dev-db`: full suite green both ways; confirmed zero leftover rows in the dev DB afterward.
+- Migration applied cleanly to both the test DB and the real dev DB; verified `\d mock.shop_amazon_events` matches the intended shape on both.
+
+## Follow-ups / not done here
+- Insights #3 (hardcoded 1-minute poll interval) and #4 (a LISTEN-connection error is fatal to the *entire application*, most consequential) remain open - continuing one at a time per the user's request.