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
2026-08-20 00:35:22 -06:00
parent 579eae5097
commit 984a4465f8
6 changed files with 346 additions and 48 deletions
@@ -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
 }