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;