Claude-assisted improvements (untested)

- dev auth flow (side-step OAuth)
- db event processing integration tests
- dev scripts (eg Makefile)
- db / test db migration setup scripts.

domains/accounts/accounts.go

@@ -1468,8 +1468,6 @@ func (db *Store) SetListingInListingInMockSyncGroupBeingEdited(ctx context.Conte
 			default:
 				return fmt.Errorf("unexpected number of rows affected: %d", n)
 			}
-
-			return nil
 		}
 
 		return consts.ErrNotFound

domains/accounts/accounts_test.go

@@ -0,0 +1,133 @@
+package accounts_test
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"ruben/inventory2/consts"
+	"ruben/inventory2/domains/accounts"
+	"ruben/inventory2/internal/testdb"
+)
+
+func TestCreateAccount(t *testing.T) {
+	pool := testdb.Pool(t)
+	store := accounts.NewStore(testdb.Logger(), pool)
+	ctx := context.Background()
+
+	userID := testdb.NewUserID(t)
+	testdb.SeedOAuthUser(t, pool, userID)
+
+	email := userID + "@example.com"
+
+	acct, err := store.CreateAccount(ctx, userID, email)
+	if err != nil {
+		t.Fatalf("CreateAccount() error = %v", err)
+	}
+	t.Cleanup(func() {
+		pool.Exec(context.Background(), "DELETE FROM accounts WHERE account_id = $1", acct.AccountID)
+	})
+
+	if acct.AccountID == 0 {
+		t.Error("CreateAccount() returned a zero AccountID")
+	}
+	if acct.UserID != userID {
+		t.Errorf("CreateAccount() UserID = %q, want %q", acct.UserID, userID)
+	}
+	if acct.Email != email {
+		t.Errorf("CreateAccount() Email = %q, want %q", acct.Email, email)
+	}
+
+	got, err := store.GetAccount(ctx, acct.AccountID)
+	if err != nil {
+		t.Fatalf("GetAccount() error = %v", err)
+	}
+	if got != acct {
+		t.Errorf("GetAccount() = %+v, want %+v", got, acct)
+	}
+}
+
+func TestCreateAccount_DuplicateUserIsConflict(t *testing.T) {
+	pool := testdb.Pool(t)
+	store := accounts.NewStore(testdb.Logger(), pool)
+	ctx := context.Background()
+
+	userID := testdb.NewUserID(t)
+	testdb.SeedOAuthUser(t, pool, userID)
+
+	acct, err := store.CreateAccount(ctx, userID, userID+"@example.com")
+	if err != nil {
+		t.Fatalf("first CreateAccount() error = %v", err)
+	}
+	t.Cleanup(func() {
+		pool.Exec(context.Background(), "DELETE FROM accounts WHERE account_id = $1", acct.AccountID)
+	})
+
+	_, err = store.CreateAccount(ctx, userID, userID+"-other@example.com")
+	if !errors.Is(err, consts.ErrConflict) {
+		t.Fatalf("second CreateAccount() error = %v, want %v", err, consts.ErrConflict)
+	}
+}
+
+func TestGetAccount_NotFound(t *testing.T) {
+	pool := testdb.Pool(t)
+	store := accounts.NewStore(testdb.Logger(), pool)
+	ctx := context.Background()
+
+	_, err := store.GetAccount(ctx, -1)
+	if !errors.Is(err, consts.ErrNotFound) {
+		t.Fatalf("GetAccount() error = %v, want %v", err, consts.ErrNotFound)
+	}
+}
+
+func TestGetUserAndAccountByAccessToken(t *testing.T) {
+	pool := testdb.Pool(t)
+	store := accounts.NewStore(testdb.Logger(), pool)
+	ctx := context.Background()
+
+	userID := testdb.NewUserID(t)
+	accessToken := testdb.SeedOAuthSession(t, pool, userID)
+
+	// before an account exists: user resolves, account does not.
+	user, acct, err := store.GetUserAndAccountByAccessToken(ctx, accessToken)
+	if err != nil {
+		t.Fatalf("GetUserAndAccountByAccessToken() before account creation: error = %v", err)
+	}
+	if user.UserID != userID {
+		t.Errorf("GetUserAndAccountByAccessToken() UserID = %q, want %q", user.UserID, userID)
+	}
+	if acct != nil {
+		t.Errorf("GetUserAndAccountByAccessToken() Account = %+v, want nil before an account is created", acct)
+	}
+
+	created, err := store.CreateAccount(ctx, userID, userID+"@example.com")
+	if err != nil {
+		t.Fatalf("CreateAccount() error = %v", err)
+	}
+	t.Cleanup(func() {
+		pool.Exec(context.Background(), "DELETE FROM accounts WHERE account_id = $1", created.AccountID)
+	})
+
+	// after an account exists: both resolve.
+	user, acct, err = store.GetUserAndAccountByAccessToken(ctx, accessToken)
+	if err != nil {
+		t.Fatalf("GetUserAndAccountByAccessToken() after account creation: error = %v", err)
+	}
+	if user.UserID != userID {
+		t.Errorf("GetUserAndAccountByAccessToken() UserID = %q, want %q", user.UserID, userID)
+	}
+	if acct == nil || acct.AccountID != created.AccountID {
+		t.Errorf("GetUserAndAccountByAccessToken() Account = %+v, want AccountID %d", acct, created.AccountID)
+	}
+}
+
+func TestGetUserAndAccountByAccessToken_UnknownToken(t *testing.T) {
+	pool := testdb.Pool(t)
+	store := accounts.NewStore(testdb.Logger(), pool)
+	ctx := context.Background()
+
+	_, _, err := store.GetUserAndAccountByAccessToken(ctx, "no-such-token-"+testdb.NewUserID(t))
+	if !errors.Is(err, consts.ErrNotFound) {
+		t.Fatalf("GetUserAndAccountByAccessToken() error = %v, want %v", err, consts.ErrNotFound)
+	}
+}

domains/amazon/mock_test.go

@@ -0,0 +1,248 @@
+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)
+	}()
+
+	// 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")
+	}
+}

domains/authentication/auth.go

@@ -17,26 +17,11 @@ import (
 	"ruben/inventory2/logging"
 )
 
-// TODO: move these to a config?
-const (
-	// The URL of our Auth0 Tenant Domain.
-	// If you're using a Custom Domain, be sure to set this to that value instead.
-	AUTH0_DOMAIN = "dev-uq3gqy5bdnwxmr6d.us.auth0.com"
-
-	// Our Auth0 application"s Client ID.
-	AUTH0_CLIENT_ID = "JEjrXTQ9fxlTLgp9RgTIACpUk8a2lqNT"
-
-	// Our Auth0 application"s Client Secret.
-	AUTH0_CLIENT_SECRET = "83U-iWdVaNnwk9XDzteo_2VMyOq_l1siKYqg1_2E7jCzgL8MnkaxlysPMcPMGlxA"
-
-	// The Callback URL of our application.
-	AUTH0_CALLBACK_URL = "https://inventory-plus-plus.com/api/auth/login/callback"
-)
-
 type (
 	// Authenticator is used to authenticate our users.
 	Authenticator struct {
-		log *logging.Logger
+		log    *logging.Logger
+		domain string
 		*oidc.Provider
 		oauth2.Config
 		db *pgxpool.Pool
@@ -65,10 +50,11 @@ func New(
 	ctx context.Context,
 	db *pgxpool.Pool,
 	logger *logging.Logger,
+	domain, clientID, clientSecret, callbackURL string,
 ) (*Authenticator, error) {
 	provider, err := oidc.NewProvider(
 		ctx,
-		"https://"+AUTH0_DOMAIN+"/",
+		"https://"+domain+"/",
 	)
 	if err != nil {
 		return nil, err
@@ -76,11 +62,12 @@ func New(
 
 	return &Authenticator{
 		log:      logger,
+		domain:   domain,
 		Provider: provider,
 		Config: oauth2.Config{
-			ClientID:     AUTH0_CLIENT_ID,
-			ClientSecret: AUTH0_CLIENT_SECRET,
-			RedirectURL:  AUTH0_CALLBACK_URL,
+			ClientID:     clientID,
+			ClientSecret: clientSecret,
+			RedirectURL:  callbackURL,
 			Endpoint:     provider.Endpoint(),
 			Scopes:       []string{oidc.ScopeOpenID, "profile"},
 		},
@@ -244,7 +231,7 @@ func (a *Authenticator) VerifyIDToken(ctx context.Context, token *oauth2.Token)
 func (a *Authenticator) GetLogoutURL(requestHost string) *url.URL {
 	return &url.URL{
 		Scheme: "https",
-		Host:   AUTH0_DOMAIN,
+		Host:   a.domain,
 		Path:   "/v2/logout",
 		RawQuery: url.Values{
 			"returnTo": {
@@ -253,7 +240,7 @@ func (a *Authenticator) GetLogoutURL(requestHost string) *url.URL {
 					Host:   requestHost,
 				}).String(),
 			},
-			"client_id": {AUTH0_CLIENT_ID},
+			"client_id": {a.Config.ClientID},
 		}.Encode(),
 	}
 }
@@ -267,7 +254,7 @@ func (a *Authenticator) RefreshAccessToken(
 	err error,
 ) {
 
-	refreshToken, tokenType, err := a.getRefreshTokenForAccessToken(ctx, accessToken)
+	refreshToken, tokenType, err := a.getRefreshTokenForAccessToken(ctx, oldAccessToken)
 	if err != nil {
 		return "", time.Time{}, fmt.Errorf("failed to load refresh token: %w", err)
 	}

domains/authentication/dev.go

@@ -0,0 +1,97 @@
+package authentication
+
+import (
+	"context"
+	"crypto/rand"
+	"encoding/hex"
+	"fmt"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgtype"
+)
+
+// DevLogin mints a local session for userID without going through the real
+// Auth0 OAuth flow. It writes a real oauth_users/oauth_tokens row, so every
+// other code path (identity lookup, account linking/creation, cookie
+// handling) treats it exactly like a normal login.
+//
+// Only ever call this from a route gated on an explicit dev-mode flag -
+// never wire it up unconditionally, since it lets the caller authenticate
+// as any user_id with no credentials.
+func (a *Authenticator) DevLogin(ctx context.Context, userID, name string) (accessToken string, expiration time.Time, err error) {
+	tokenBytes := make([]byte, 32)
+	if _, err := rand.Read(tokenBytes); err != nil {
+		return "", time.Time{}, fmt.Errorf("failed to generate access token: %w", err)
+	}
+	accessToken = "dev_" + hex.EncodeToString(tokenBytes)
+
+	// Set far in the future so the near-expiry refresh path in
+	// server/auth never tries to refresh a token that has no real
+	// Auth0-side refresh token behind it.
+	expiration = time.Now().Add(365 * 24 * time.Hour)
+
+	if _, err = a.db.Exec(
+		ctx,
+		`
+			WITH ensured_user AS (
+				INSERT INTO oauth_users (user_id)
+				VALUES (@user_id)
+				ON CONFLICT DO NOTHING
+			)
+			INSERT INTO oauth_tokens (
+				access_token,
+				token_type,
+				refresh_token,
+				expiry,
+
+				id_token_issuer,
+				id_token_audience,
+				id_token_subject,
+				id_token_expiry,
+				id_token_issued_at,
+				id_token_nonce,
+				id_token_access_token_hash,
+
+				id_token_custom_claims_family_name,
+				id_token_custom_claims_given_name,
+				id_token_custom_claims_name,
+				id_token_custom_claims_nickname,
+				id_token_custom_claims_picture,
+				id_token_custom_claims_updated_at
+			)
+			VALUES (
+				@access_token,
+				'dev',
+				'',
+				@expiry,
+
+				'dev-auth-bypass',
+				@audience,
+				@user_id,
+				@expiry,
+				NOW(),
+				'',
+				'',
+
+				@name,
+				@name,
+				@name,
+				@name,
+				'',
+				NOW()
+			)
+		`,
+		pgx.NamedArgs{
+			"access_token": accessToken,
+			"expiry":       expiration,
+			"user_id":      userID,
+			"name":         name,
+			"audience":     pgtype.FlatArray[string]{"dev"},
+		},
+	); err != nil {
+		return "", time.Time{}, fmt.Errorf("failed to save dev session: %w", err)
+	}
+
+	return accessToken, expiration, nil
+}

domains/authentication/dev_test.go

@@ -0,0 +1,95 @@
+package authentication
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"ruben/inventory2/consts"
+	"ruben/inventory2/internal/testdb"
+)
+
+func TestDevLogin(t *testing.T) {
+	pool := testdb.Pool(t)
+	auth := &Authenticator{db: pool}
+	ctx := context.Background()
+
+	userID := testdb.NewUserID(t)
+	t.Cleanup(func() {
+		pool.Exec(context.Background(), "DELETE FROM oauth_tokens WHERE id_token_subject = $1", userID)
+		pool.Exec(context.Background(), "DELETE FROM oauth_users WHERE user_id = $1", userID)
+	})
+
+	accessToken, expiration, err := auth.DevLogin(ctx, userID, "Test User")
+	if err != nil {
+		t.Fatalf("DevLogin() error = %v", err)
+	}
+	if accessToken == "" {
+		t.Fatal("DevLogin() returned an empty access token")
+	}
+	if !expiration.After(time.Now().Add(30 * 24 * time.Hour)) {
+		t.Errorf("DevLogin() expiration = %v, want something far enough out to avoid the near-expiry refresh path", expiration)
+	}
+
+	claims, err := auth.GetAccessTokenClaimsAndExpiration(ctx, accessToken)
+	if err != nil {
+		t.Fatalf("GetAccessTokenClaimsAndExpiration() error = %v", err)
+	}
+	if claims.Name != "Test User" {
+		t.Errorf("claims.Name = %q, want %q", claims.Name, "Test User")
+	}
+	// Postgres timestamptz has microsecond precision, so the round-tripped
+	// value loses the sub-microsecond portion of Go's nanosecond clock.
+	if diff := claims.Expiration.Sub(expiration); diff > time.Millisecond || diff < -time.Millisecond {
+		t.Errorf("claims.Expiration = %v, want ~%v (diff %v)", claims.Expiration, expiration, diff)
+	}
+
+	_, tokenType, err := auth.getRefreshTokenForAccessToken(ctx, accessToken)
+	if err != nil {
+		t.Fatalf("getRefreshTokenForAccessToken() error = %v", err)
+	}
+	if tokenType != "dev" {
+		t.Errorf("tokenType = %q, want %q", tokenType, "dev")
+	}
+}
+
+// DevLogin should be safe to call more than once for the same user_id -
+// e.g. testing multiple times as the same dev identity - since oauth_users
+// is keyed on user_id but each call mints its own oauth_tokens row.
+func TestDevLogin_SameUserIDTwice(t *testing.T) {
+	pool := testdb.Pool(t)
+	auth := &Authenticator{db: pool}
+	ctx := context.Background()
+
+	userID := testdb.NewUserID(t)
+	t.Cleanup(func() {
+		pool.Exec(context.Background(), "DELETE FROM oauth_tokens WHERE id_token_subject = $1", userID)
+		pool.Exec(context.Background(), "DELETE FROM oauth_users WHERE user_id = $1", userID)
+	})
+
+	token1, _, err := auth.DevLogin(ctx, userID, "Test User")
+	if err != nil {
+		t.Fatalf("first DevLogin() error = %v", err)
+	}
+
+	token2, _, err := auth.DevLogin(ctx, userID, "Test User")
+	if err != nil {
+		t.Fatalf("second DevLogin() error = %v", err)
+	}
+
+	if token1 == token2 {
+		t.Fatalf("DevLogin() returned the same access token twice: %q", token1)
+	}
+}
+
+func TestGetAccessTokenClaimsAndExpiration_UnknownToken(t *testing.T) {
+	pool := testdb.Pool(t)
+	auth := &Authenticator{db: pool}
+	ctx := context.Background()
+
+	_, err := auth.GetAccessTokenClaimsAndExpiration(ctx, "no-such-token-"+testdb.NewUserID(t))
+	if !errors.Is(err, consts.ErrNotFound) {
+		t.Fatalf("GetAccessTokenClaimsAndExpiration() error = %v, want %v", err, consts.ErrNotFound)
+	}
+}

domains/raw_events/events_test.go

@@ -0,0 +1,83 @@
+package raw_events_test
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+
+	"ruben/inventory2/domains/raw_events"
+	"ruben/inventory2/internal/testdb"
+)
+
+func TestSaveAndLoadEventsForStore(t *testing.T) {
+	pool := testdb.Pool(t)
+	store := raw_events.NewStore(testdb.Logger(), pool)
+	ctx := context.Background()
+
+	platform := "test-platform"
+	storeID := "test-store-" + uuid.NewString()
+
+	t.Cleanup(func() {
+		pool.Exec(context.Background(), "DELETE FROM raw_store_events WHERE platform = $1 AND store_id = $2", platform, storeID)
+	})
+
+	older := raw_events.Event{
+		Platform:       platform,
+		StoreID:        storeID,
+		EventID:        "evt-1",
+		EventTimestamp: time.Now().Add(-time.Hour).UTC(),
+		Payload:        json.RawMessage(`{"n":1}`),
+	}
+	newer := raw_events.Event{
+		Platform:       platform,
+		StoreID:        storeID,
+		EventID:        "evt-2",
+		EventTimestamp: time.Now().UTC(),
+		Payload:        json.RawMessage(`{"n":2}`),
+	}
+
+	if err := store.Save(ctx, &older); err != nil {
+		t.Fatalf("Save() older event error = %v", err)
+	}
+	if err := store.Save(ctx, &newer); err != nil {
+		t.Fatalf("Save() newer event error = %v", err)
+	}
+
+	got, err := store.LoadEventsForStore(ctx, platform, storeID)
+	if err != nil {
+		t.Fatalf("LoadEventsForStore() error = %v", err)
+	}
+
+	if len(got) != 2 {
+		t.Fatalf("LoadEventsForStore() returned %d events, want 2: %+v", len(got), got)
+	}
+
+	// ordered event_timestamp DESC - newest first.
+	if got[0].EventID != "evt-2" || got[1].EventID != "evt-1" {
+		t.Errorf("LoadEventsForStore() order = [%s, %s], want [evt-2, evt-1]", got[0].EventID, got[1].EventID)
+	}
+	var payload struct{ N int }
+	if err := json.Unmarshal(got[0].Payload, &payload); err != nil {
+		t.Fatalf("failed to unmarshal LoadEventsForStore()[0].Payload = %s: %v", got[0].Payload, err)
+	}
+	if payload.N != 2 {
+		t.Errorf("LoadEventsForStore()[0].Payload n = %d, want 2", payload.N)
+	}
+}
+
+func TestLoadEventsForStore_NoEvents(t *testing.T) {
+	pool := testdb.Pool(t)
+	store := raw_events.NewStore(testdb.Logger(), pool)
+	ctx := context.Background()
+
+	got, err := store.LoadEventsForStore(ctx, "test-platform", "no-such-store-"+uuid.NewString())
+	if err != nil {
+		t.Fatalf("LoadEventsForStore() error = %v", err)
+	}
+	if len(got) != 0 {
+		t.Fatalf("LoadEventsForStore() = %+v, want empty", got)
+	}
+}