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. (*Mocks).processEvent fires notifications from a goroutine without
// waiting for them, so tests need to synchronize on notified rather than
// asserting on the event list immediately.
type notifySpy struct {
	mu       sync.Mutex
	events   []raw_events.Event
	notified chan struct{}
}

func newNotifySpy() *notifySpy {
	return &notifySpy{notified: make(chan struct{}, 1000)}
}

func (s *notifySpy) Notify(_ context.Context, e raw_events.Event) error {
	s.mu.Lock()
	s.events = append(s.events, e)
	s.mu.Unlock()
	s.notified <- struct{}{}
	return nil
}

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) waitForCount(t *testing.T, n int, timeout time.Duration) {
	t.Helper()
	deadline := time.After(timeout)
	for i := 0; i < n; i++ {
		select {
		case <-s.notified:
		case <-deadline:
			t.Fatalf("timed out after %v waiting for notification %d/%d (got %d so far)", timeout, i+1, n, len(s.eventsSnapshot()))
		}
	}
}

// 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, successful bool
	err := pool.QueryRow(context.Background(), `
		SELECT processed, processed_successfully
		FROM mock.shop_amazon_events
		WHERE shop_id = $1 AND event_id = $2
	`, shopID, eventID).Scan(&processed, &successful)
	if err != nil {
		t.Fatalf("failed to check processed state: %v", err)
	}
	return processed && successful
}

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 NOT processed
	`, 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)
	})
}

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

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

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

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

	if !isProcessed(t, pool, shopID, "evt-1") {
		t.Error("event was not marked processed despite no listener being configured")
	}
}

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