angel
9c9b33e34f
- dev auth flow (side-step OAuth) - db event processing integration tests - dev scripts (eg Makefile) - db / test db migration setup scripts.
249 lines
8.2 KiB
Go
249 lines
8.2 KiB
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 ¬ifySpy{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)
|
|
}()
|
|
|
|
// 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)
|
|
|
|
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)
|
|
}()
|
|
|
|
time.Sleep(50 * time.Millisecond) // let it reach its first select
|
|
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")
|
|
}
|
|
}
|