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"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"

	"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()
	require.NoError(t, ack(context.Background()), "ack()")
}

// completedCountSnapshot is safe to call from another goroutine (e.g. from
// inside require.Eventually's condition), unlike helpers that call t.Fatal.
func (s *notifySpy) completedCountSnapshot() int {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.completedCount
}

// 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()
	require.Eventually(t, func() bool {
		return s.completedCountSnapshot() >= n
	}, timeout, 5*time.Millisecond, "timed out waiting for %d total completed notifications (got %d)", n, s.completedCountSnapshot())
}

// 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)
	require.NoError(t, err, "insert raw amazon event")
}

// 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)
	require.NoError(t, err, "insert %d raw amazon events", n)
}

// queryIsProcessed has no *testing.T dependency, so it's safe to call from
// require.Eventually's condition function (which testify runs on a
// separate goroutine - t.Fatal/require must only be called from the main
// test goroutine). isProcessed wraps it for direct, main-goroutine use.
func queryIsProcessed(ctx context.Context, pool *pgxpool.Pool, shopID, eventID string) (bool, error) {
	var processed bool
	err := pool.QueryRow(ctx, `
		SELECT processed_at IS NOT NULL
		FROM mock.shop_amazon_events
		WHERE shop_id = $1 AND event_id = $2
	`, shopID, eventID).Scan(&processed)
	return processed, err
}

func isProcessed(t *testing.T, pool *pgxpool.Pool, shopID, eventID string) bool {
	t.Helper()
	processed, err := queryIsProcessed(context.Background(), pool, shopID, eventID)
	require.NoError(t, err, "check processed state")
	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)
	require.NoError(t, err, "check notified state")
	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)
	require.NoError(t, err, "count unprocessed events")
	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)
	require.NoError(t, err, "find the LISTEN connection's backend pid")

	_, err = pool.Exec(ctx, `SELECT pg_terminate_backend($1)`, pid)
	require.NoError(t, err, "terminate backend %d", pid)
}

// waitReady blocks until m signals it's actively listening, failing the
// test if that doesn't happen within timeout.
func waitReady(t *testing.T, m *Mocks, timeout time.Duration) {
	t.Helper()
	select {
	case <-m.Ready():
	case <-time.After(timeout):
		require.Fail(t, "ProcessEvents() did not become ready (LISTEN registered)", "within %v", timeout)
	}
}

// waitStopped blocks until ProcessEvents returns on errCh, asserting it
// returns a nil error, failing the test if that doesn't happen within
// timeout.
func waitStopped(t *testing.T, errCh <-chan error, timeout time.Duration) {
	t.Helper()
	select {
	case err := <-errCh:
		require.NoError(t, err, "ProcessEvents() after context cancellation")
	case <-time.After(timeout):
		require.Fail(t, "ProcessEvents() did not return", "within %v of context cancellation", timeout)
	}
}

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)

	require.NoError(t, m.processUnprocessedEvents(ctx), "processUnprocessedEvents()")

	spy.waitForCount(t, n, 5*time.Second)

	assert.Zero(t, countUnprocessed(t, pool, shopID),
		"countUnprocessed() should be 0 (all %d events should be processed across multiple 100-row batches)", 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")

	require.NoError(t, m.processUnprocessedEvents(ctx), "processUnprocessedEvents()")

	// with no listener, nothing ever acks - the event is expected to stay
	// "notified" forever, not silently marked processed.
	assert.False(t, isProcessed(t, pool, shopID, "evt-1"),
		"event was marked processed despite no listener being configured to ack it")
	assert.True(t, isNotified(t, pool, shopID, "evt-1"),
		"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")

	require.NoError(t, m.processUnprocessedEvents(ctx), "processUnprocessedEvents() (1st pass)")
	spy.waitForCount(t, 1, 2*time.Second)

	require.False(t, isProcessed(t, pool, shopID, "evt-1"), "event was marked processed despite the listener never acking")
	require.True(t, isNotified(t, pool, shopID, "evt-1"), "event should be in the notified state after the first dispatch")

	// still within notifyRetryAfter: shouldn't be re-notified yet.
	require.NoError(t, m.processUnprocessedEvents(ctx), "processUnprocessedEvents() (immediate 2nd pass)")
	require.Len(t, spy.eventsSnapshot(), 1, "listener should not have been notified again before notifyRetryAfter elapsed")

	time.Sleep(60 * time.Millisecond) // past notifyRetryAfter

	require.NoError(t, m.processUnprocessedEvents(ctx), "processUnprocessedEvents() (3rd pass, after retry window)")
	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)

	require.True(t, isProcessed(t, pool, shopID, "evt-1"), "event should be processed once any recorded ack for it is called")

	require.NoError(t, m.processUnprocessedEvents(ctx), "processUnprocessedEvents() (4th pass, after ack)")
	assert.Len(t, spy.eventsSnapshot(), 2, "listener should not be notified again after being acked")
}

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

	waitReady(t, m, 5*time.Second)

	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()
	waitStopped(t, errCh, 5*time.Second)
}

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

	waitReady(t, m, 5*time.Second)

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

	require.Eventually(t, func() bool {
		processed, _ := queryIsProcessed(context.Background(), pool, shopID, "evt-1")
		return processed
	}, 5*time.Second, 20*time.Millisecond,
		"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)")

	spy.waitForCount(t, 1, 2*time.Second)
	got := spy.eventsSnapshot()[0]
	assert.Equal(t, shopID, got.StoreID, "listener notified with unexpected StoreID")
	assert.Equal(t, "evt-1", got.EventID, "listener notified with unexpected EventID")

	cancel()
	waitStopped(t, errCh, 5*time.Second)
}

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

	waitReady(t, m, 5*time.Second)
	cancel()

	waitStopped(t, errCh, 5*time.Second)
}

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

	waitReady(t, m, 5*time.Second)

	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:
		require.Fail(t, "ProcessEvents() returned after its LISTEN connection was killed, want it to reconnect and keep running",
			"err = %v", err)
	case <-time.After(2 * time.Second):
	}

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

	require.Eventually(t, func() bool {
		processed, _ := queryIsProcessed(context.Background(), pool, shopID, "evt-1")
		return processed
	}, 5*time.Second, 20*time.Millisecond,
		"event was not processed within 5s of insertion after the LISTEN connection was forcibly dropped - "+
			"reconnection did not restore the reactive path")

	cancel()
	waitStopped(t, errCh, 5*time.Second)
}