amazon: make the poll fallback interval configurable and observable

ProcessEvents' safety-net poll was a bare time.Minute literal inline in
its select statement - no way to verify the fallback path works without
waiting 60+ seconds in a test, no way to tune the cadence without a code
change, and no way to tell afterward whether a given loop iteration was
triggered by a real NOTIFY or by the poll timer.

Adds Mocks.pollInterval (default time.Minute) and a WithPollInterval(d)
builder, mirroring WithNotifyRetryAfter's existing pattern - deploy-time
configurable via construction, not a live/API-adjustable knob, and not
wired through .env, matching how notifyRetryAfter already works.

Adds a Debug log line on each of the two meaningful wake-up branches so
which path fired is now observable.

New test proves the poll branch actually works without waiting or
touching the shared DB trigger (which would be unsafe against the real
dev DB): it reuses the retry mechanism from the previous commit so a
second dispatch can only come from the poll timer, since nothing else
ever notifies again for the rest of the test.

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

domains/amazon/mock.go

@@ -27,6 +27,13 @@ type (
 		// touching any row.
 		notifyRetryAfter time.Duration
 
+		// pollInterval is the fallback cadence ProcessEvents' main loop
+		// checks for unprocessed/retry-due events on its own, independent
+		// of Postgres NOTIFY - a safety net for events that end up in
+		// mock.shop_amazon_events without ever going through the
+		// mock.raw_shop_events insert+trigger path that fires NOTIFY.
+		pollInterval time.Duration
+
 		ready     chan struct{}
 		readyOnce sync.Once
 	}
@@ -48,6 +55,7 @@ const (
 	eventChannelName = "mock_shop_amazon_event_inserted"
 
 	defaultNotifyRetryAfter = 30 * time.Second
+	defaultPollInterval     = time.Minute
 )
 
 func NewMocks(log *logging.Logger, db *pgxpool.Pool) *Mocks {
@@ -55,6 +63,7 @@ func NewMocks(log *logging.Logger, db *pgxpool.Pool) *Mocks {
 		log:              log,
 		db:               db,
 		notifyRetryAfter: defaultNotifyRetryAfter,
+		pollInterval:     defaultPollInterval,
 		ready:            make(chan struct{}),
 	}
 }
@@ -72,6 +81,14 @@ func (m *Mocks) WithNotifyRetryAfter(d time.Duration) *Mocks {
 	return m
 }
 
+// WithPollInterval overrides how often ProcessEvents checks for
+// unprocessed/retry-due events on its own, independent of NOTIFY. Mainly
+// useful for tests that don't want to wait out the default.
+func (m *Mocks) WithPollInterval(d time.Duration) *Mocks {
+	m.pollInterval = 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 -
@@ -101,7 +118,9 @@ func (m *Mocks) ProcessEvents(ctx context.Context) error {
 			if !ok {
 				return <-errCh
 			}
-		case <-time.After(time.Minute):
+			m.log.Debug("woke up: notification received")
+		case <-time.After(m.pollInterval):
+			m.log.Debug("woke up: poll interval elapsed")
 		}
 	}
 }

domains/amazon/mock_test.go

@@ -288,6 +288,60 @@ func TestProcessUnprocessedEvents_RetriesUnackedNotification(t *testing.T) {
 	}
 }
 
+// TestProcessEvents_PollFallbackPicksUpRetryDueEvents proves the poll
+// branch of ProcessEvents' select actually causes reprocessing, decoupled
+// from NOTIFY entirely: after the one real insert (which does fire NOTIFY,
+// same as any other test here), nothing ever triggers another notification
+// for the rest of the test. The event becomes retry-due almost immediately
+// (notifyRetryAfter is tiny), so the *only* way it can be dispatched a
+// second time is the poll timer in the select firing on its own.
+func TestProcessEvents_PollFallbackPicksUpRetryDueEvents(t *testing.T) {
+	pool := testdb.Pool(t)
+	spy := newNotifySpy()
+	spy.setAutoAck(false)
+	m := NewMocks(testdb.Logger(), pool).
+		SetListener(spy).
+		WithNotifyRetryAfter(30 * time.Millisecond).
+		WithPollInterval(60 * time.Millisecond)
+
+	shopID := "test-shop-" + uuid.NewString()
+	cleanupShop(t, pool, shopID)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	errCh := make(chan error, 1)
+	go func() {
+		errCh <- m.ProcessEvents(ctx)
+	}()
+
+	select {
+	case <-m.Ready():
+	case <-time.After(5 * time.Second):
+		t.Fatal("ProcessEvents() did not become ready (LISTEN registered) within 5s")
+	}
+
+	insertRawAmazonEvent(t, pool, shopID, "evt-1")
+
+	// first dispatch, via the real NOTIFY.
+	spy.waitForCount(t, 1, 2*time.Second)
+
+	// second dispatch: nothing will notify again from here on, so this can
+	// only come from the poll branch of the select waking the loop up on
+	// its own and finding the event retry-due.
+	spy.waitForCount(t, 2, 3*time.Second)
+
+	cancel()
+	select {
+	case err := <-errCh:
+		if err != nil {
+			t.Fatalf("ProcessEvents() returned error = %v after context cancellation, want nil", err)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("ProcessEvents() did not return within 5s of context cancellation")
+	}
+}
+
 // TestProcessEvents_ReactsToNotification drives the actual long-running
 // loop: LISTEN registration, a real Postgres NOTIFY fired by the DB trigger
 // on insert, WaitForNotification waking the loop, and the listener callback