inventory-plus-plus/domains/amazon/mock_test.go

package amazon

import (
	"context"
	"sync"
	"testing"
	"time"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5/pgxpool"

	"ruben/inventory2/domains/raw_events"
	"ruben/inventory2/internal/testdb"
)

// notifySpy is a MockEventListener that records every event it's notified
// 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
	acks           []func(context.Context) error
	completedCount int // Notify calls that have returned, including any auto-ack
	autoAck        bool
}

func newNotifySpy() *notifySpy {
	return &notifySpy{autoAck: true}
}

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()

	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 {
	s.mu.Lock()
	defer s.mu.Unlock()
	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.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)
	}
}

// insertRawAmazonEvent inserts directly into mock.raw_shop_events, the same
// entry point real mock sale/refund/inventory simulations use. A DB trigger
// (mock.process_raw_amazon_event, see migrations 000026/000028) copies the
// row into mock.shop_amazon_events and fires pg_notify on
// mock_shop_amazon_event_inserted - so this one insert exercises the exact
// same path production traffic does, instead of faking the downstream
// table directly.
func insertRawAmazonEvent(t *testing.T, pool *pgxpool.Pool, shopID, eventID string) {
	t.Helper()
	_, err := pool.Exec(context.Background(), `
		INSERT INTO mock.raw_shop_events (platform, shop_id, event_timestamp, event_id, raw_payload)
		VALUES ('amazon', $1, NOW(), $2, '{}'::jsonb)
	`, shopID, eventID)
	if err != nil {
		t.Fatalf("failed to insert raw amazon event: %v", err)
	}
}

// insertRawAmazonEvents bulk-inserts n events for shopID in one statement,
// each with a distinct event_id/event_timestamp.
func insertRawAmazonEvents(t *testing.T, pool *pgxpool.Pool, shopID string, n int) {
	t.Helper()
	_, err := pool.Exec(context.Background(), `
		INSERT INTO mock.raw_shop_events (platform, shop_id, event_timestamp, event_id, raw_payload)
		SELECT 'amazon', $1, NOW() + (s || ' milliseconds')::interval, 'evt-' || s, '{}'::jsonb
		FROM generate_series(1, $2) AS s
	`, shopID, n)
	if err != nil {
		t.Fatalf("failed to insert %d raw amazon events: %v", n, err)
	}
}

func isProcessed(t *testing.T, pool *pgxpool.Pool, shopID, eventID string) bool {
	t.Helper()
	var processed bool
	err := pool.QueryRow(context.Background(), `
		SELECT processed_at IS NOT NULL
		FROM mock.shop_amazon_events
		WHERE shop_id = $1 AND event_id = $2
	`, shopID, eventID).Scan(&processed)
	if err != nil {
		t.Fatalf("failed to check processed state: %v", err)
	}
	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 processed_at IS NULL
	`, shopID).Scan(&n)
	if err != nil {
		t.Fatalf("failed to count unprocessed events: %v", err)
	}
	return n
}

// cleanupShop registers deletion of every row this test's shopID may have
// produced, in FK-safe order (shop_amazon_events references raw_shop_events).
func cleanupShop(t *testing.T, pool *pgxpool.Pool, shopID string) {
	t.Cleanup(func() {
		ctx := context.Background()
		pool.Exec(ctx, `DELETE FROM mock.shop_amazon_events WHERE shop_id = $1`, shopID)
		pool.Exec(ctx, `DELETE FROM mock.raw_shop_events WHERE platform = 'amazon' AND shop_id = $1`, shopID)
	})
}

// terminateListenConnection finds the backend holding this package's LISTEN
// registration (identified by its last query text, which Postgres keeps
// showing while the connection sits idle waiting for notifications) and
// forcibly kills it - the same failure mode a dropped connection or a
// Postgres restart produces, so tests can exercise real reconnect behavior
// instead of a simulated one.
func terminateListenConnection(t *testing.T, pool *pgxpool.Pool) {
	t.Helper()
	ctx := context.Background()

	var pid int
	err := pool.QueryRow(ctx, `
		SELECT pid FROM pg_stat_activity
		WHERE query = 'LISTEN ' || $1
		ORDER BY backend_start DESC
		LIMIT 1
	`, eventChannelName).Scan(&pid)
	if err != nil {
		t.Fatalf("failed to find the LISTEN connection's backend pid: %v", err)
	}

	if _, err := pool.Exec(ctx, `SELECT pg_terminate_backend($1)`, pid); err != nil {
		t.Fatalf("failed to terminate backend %d: %v", pid, err)
	}
}

func TestProcessUnprocessedEvents_ProcessesAllEventsAcrossBatches(t *testing.T) {
	pool := testdb.Pool(t)
	spy := newNotifySpy()
	m := NewMocks(testdb.Logger(), pool).SetListener(spy)
	ctx := context.Background()

	shopID := "test-shop-" + uuid.NewString()
	cleanupShop(t, pool, shopID)

	const n = 150 // exceeds the 100-row LIMIT per batch inside processUnprocessedEvents
	insertRawAmazonEvents(t, pool, shopID, n)

	if err := m.processUnprocessedEvents(ctx); err != nil {
		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)
	}
}

func TestProcessUnprocessedEvents_NoListenerConfigured(t *testing.T) {
	pool := testdb.Pool(t)
	m := NewMocks(testdb.Logger(), pool) // no SetListener call
	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() 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.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)
	}
}

// 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
// - the full stateful path, not just the deterministic batch-processing
// core covered above.
func TestProcessEvents_ReactsToNotification(t *testing.T) {
	pool := testdb.Pool(t)
	spy := newNotifySpy()
	m := NewMocks(testdb.Logger(), pool).SetListener(spy)

	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")

	deadline := time.Now().Add(5 * time.Second)
	for !isProcessed(t, pool, shopID, "evt-1") {
		if time.Now().After(deadline) {
			t.Fatal("event was not processed within 5s of insertion - the reactive LISTEN/NOTIFY wake-up did not fire (the 1-minute poll fallback would eventually catch it, but this test intentionally doesn't wait that long)")
		}
		time.Sleep(20 * time.Millisecond)
	}

	spy.waitForCount(t, 1, 2*time.Second)
	if got := spy.eventsSnapshot()[0]; got.StoreID != shopID || got.EventID != "evt-1" {
		t.Errorf("listener notified with %+v, want StoreID=%q EventID=%q", got, shopID, "evt-1")
	}

	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_ShutsDownOnContextCancel checks the lifecycle in
// isolation, without depending on NOTIFY timing at all - a fast, low-flake
// guard against shutdown regressions (hangs, goroutine leaks) independent
// of whether the reactive path above is working.
func TestProcessEvents_ShutsDownOnContextCancel(t *testing.T) {
	pool := testdb.Pool(t)
	m := NewMocks(testdb.Logger(), pool)

	ctx, cancel := context.WithCancel(context.Background())

	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")
	}
	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_ReconnectsAfterListenConnectionDrops is insight #4's
// fix: forcibly kills the real backend connection ProcessEvents is
// LISTEN-ing on (the same failure mode a dropped connection or a Postgres
// restart produces) and confirms it reconnects and keeps working on its
// own, rather than the error propagating out of ProcessEvents entirely.
func TestProcessEvents_ReconnectsAfterListenConnectionDrops(t *testing.T) {
	pool := testdb.Pool(t)
	spy := newNotifySpy()
	m := NewMocks(testdb.Logger(), pool).SetListener(spy)

	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")
	}

	terminateListenConnection(t, pool)

	// ProcessEvents should still be running, just reconnecting (initial
	// backoff is 1s) - it must not have returned because of this.
	select {
	case err := <-errCh:
		t.Fatalf("ProcessEvents() returned (err = %v) after its LISTEN connection was killed, want it to reconnect and keep running", err)
	case <-time.After(2 * time.Second):
	}

	insertRawAmazonEvent(t, pool, shopID, "evt-1")

	deadline := time.Now().Add(5 * time.Second)
	for !isProcessed(t, pool, shopID, "evt-1") {
		if time.Now().After(deadline) {
			t.Fatal("event was not processed within 5s of insertion after the LISTEN connection was forcibly dropped - reconnection did not restore the reactive path")
		}
		time.Sleep(20 * time.Millisecond)
	}

	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")
	}
}