amazon: add a Ready() signal for ProcessEvents' LISTEN registration

Tests (and any other caller) previously had no way to know when the
Postgres LISTEN behind the reactive event-processing loop had actually
registered, forcing a guessed sleep before relying on it. Mocks now
exposes Ready() <-chan struct{}, closed once listenForNotifications
successfully issues LISTEN. Purely additive - ProcessEvents' signature
is unchanged.

Updates the integration tests to wait on Ready() instead of a flat
sleep, which also cut TestProcessEvents_ReactsToNotification's runtime
from ~0.25s to ~0.06s with no flakes across repeated runs.

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

domains/amazon/mock.go

@@ -7,6 +7,7 @@ import (
 	"ruben/inventory2/domains/accounts"
 	"ruben/inventory2/domains/raw_events"
 	"ruben/inventory2/logging"
+	"sync"
 	"time"
 
 	"github.com/jackc/pgx/v5"
@@ -18,6 +19,9 @@ type (
 		log      *logging.Logger
 		db       *pgxpool.Pool
 		listener MockEventListener
+
+		ready     chan struct{}
+		readyOnce sync.Once
 	}
 
 	MockEventListener interface {
@@ -31,8 +35,9 @@ const (
 
 func NewMocks(log *logging.Logger, db *pgxpool.Pool) *Mocks {
 	return &Mocks{
-		log: log,
-		db:  db,
+		log:   log,
+		db:    db,
+		ready: make(chan struct{}),
 	}
 }
 
@@ -41,6 +46,14 @@ func (m *Mocks) SetListener(l MockEventListener) *Mocks {
 	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 -
+// should wait on this instead of guessing with a sleep.
+func (m *Mocks) Ready() <-chan struct{} {
+	return m.ready
+}
+
 func (m *Mocks) ProcessEvents(ctx context.Context) error {
 	notifCh, errCh := m.listenForNotifications(ctx)
 
@@ -83,6 +96,8 @@ func (m *Mocks) listenForNotifications(ctx context.Context) (<-chan struct{}, <-
 		return nil, errCh
 	}
 
+	m.readyOnce.Do(func() { close(m.ready) })
+
 	ch := make(chan struct{})
 
 	go func() (err error) {

domains/amazon/mock_test.go

@@ -184,14 +184,11 @@ func TestProcessEvents_ReactsToNotification(t *testing.T) {
 		errCh <- m.ProcessEvents(ctx)
 	}()
 
-	// ProcessEvents registers its Postgres LISTEN synchronously before
-	// entering its main loop, but that registration still races against
-	// this goroutine actually getting scheduled - the production code
-	// exposes no "listening now" signal a caller (or a test) can wait
-	// on. A short sleep is the only lever available from outside; see
-	// the accompanying report for why that's a real testability/design
-	// gap, not just a test-flakiness workaround.
-	time.Sleep(200 * time.Millisecond)
+	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")
 
@@ -234,7 +231,11 @@ func TestProcessEvents_ShutsDownOnContextCancel(t *testing.T) {
 		errCh <- m.ProcessEvents(ctx)
 	}()
 
-	time.Sleep(50 * time.Millisecond) // let it reach its first select
+	select {
+	case <-m.Ready():
+	case <-time.After(5 * time.Second):
+		t.Fatal("ProcessEvents() did not become ready (LISTEN registered) within 5s")
+	}
 	cancel()
 
 	select {