orbit/examples/result-inbox/internal/inbox/store_integration_test.go

package inbox

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"os"
	"strings"
	"sync"
	"testing"
	"time"
	"unicode"

	"github.com/go-sql-driver/mysql"
)

func integrationDB(t *testing.T) (*sql.DB, string) {
	t.Helper()
	dsn := os.Getenv("ORBIT_TEST_RECEIVER_DSN")
	if dsn == "" {
		t.Skip("requires dedicated disposable receiver database")
	}
	cfg, err := mysql.ParseDSN(dsn)
	if err != nil || cfg.DBName != "orbit_receiver_test" || !strings.HasPrefix(cfg.Addr, "127.0.0.1:") {
		t.Fatal("receiver integration must use dedicated disposable loopback database")
	}
	ctx, stop := context.WithTimeout(context.Background(), 5*time.Second)
	defer stop()
	db, err := Open(ctx, dsn)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { db.Close() })
	return db, dsn
}
func testStore(t *testing.T, db *sql.DB) *Store {
	t.Helper()
	s, err := New(db, "test-"+newID(t), newID(t))
	if err != nil {
		t.Fatal(err)
	}
	return s
}
func inboxCount(t *testing.T, ctx context.Context, s *Store) int {
	t.Helper()
	var count int
	if err := s.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM receiver_inbox WHERE consumer_name=? AND queue_id=?`, s.consumer, s.queue).Scan(&count); err != nil {
		t.Fatal(err)
	}
	return count
}
func TestIntegrationReceiverSchema(t *testing.T) {
	db, _ := integrationDB(t)
	ctx, stop := context.WithTimeout(context.Background(), 5*time.Second)
	defer stop()
	var fk int
	if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM information_schema.REFERENTIAL_CONSTRAINTS WHERE CONSTRAINT_SCHEMA=DATABASE()`).Scan(&fk); err != nil || fk != 0 {
		t.Fatal("receiver foreign keys", fk, err)
	}
	rows, err := db.QueryContext(ctx, `SELECT COLUMN_COMMENT FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME IN ('receiver_inbox','receiver_task_views')`)
	if err != nil {
		t.Fatal(err)
	}
	defer rows.Close()
	count := 0
	for rows.Next() {
		var comment string
		if err = rows.Scan(&comment); err != nil {
			t.Fatal(err)
		}
		if !strings.ContainsFunc(comment, func(r rune) bool { return unicode.Is(unicode.Han, r) }) {
			t.Fatal("receiver column missing Chinese comment")
		}
		count++
	}
	if rows.Err() != nil || count != 17 {
		t.Fatal("receiver model column count", count, rows.Err())
	}
	t.Log("receiver schema has two tables, 17 Chinese column comments and zero foreign keys")
}
func TestIntegrationReceiverConcurrentDuplicateAndConflict(t *testing.T) {
	db, _ := integrationDB(t)
	s := testStore(t, db)
	ctx, stop := context.WithTimeout(context.Background(), 15*time.Second)
	defer stop()
	event := testEvent(t, s.queue)
	type outcome struct {
		result Result
		err    error
	}
	results := make(chan outcome, 8)
	start := make(chan struct{})
	var wg sync.WaitGroup
	for range 8 {
		wg.Add(1)
		go func() { defer wg.Done(); <-start; result, err := s.Apply(ctx, event); results <- outcome{result, err} }()
	}
	close(start)
	wg.Wait()
	close(results)
	duplicates := 0
	for result := range results {
		if result.err != nil {
			t.Fatal(result.err)
		}
		if result.result.Duplicate {
			duplicates++
		}
	}
	view, err := s.View(ctx, event.AggregateID)
	if err != nil || duplicates != 7 || view.AppliedEvents != 1 || view.LastRevision != event.AggregateRevision || inboxCount(t, ctx, s) != 1 {
		t.Fatal("concurrent duplicate changed business effect", duplicates, err)
	}
	event.Snapshot.Result = json.RawMessage(`{"tampered":true}`)
	if _, err = s.Apply(ctx, event); !errors.Is(err, ErrEventConflict) {
		t.Fatal("same identity different content acknowledged", err)
	}
	unchanged, err := s.View(ctx, event.AggregateID)
	if err != nil || string(unchanged.ResultJSON) != string(view.ResultJSON) || unchanged.AppliedEvents != 1 {
		t.Fatal("conflicting event changed view", err)
	}
}
func TestIntegrationReceiverLateVersionsAndIsolation(t *testing.T) {
	db, _ := integrationDB(t)
	s := testStore(t, db)
	ctx, stop := context.WithTimeout(context.Background(), 10*time.Second)
	defer stop()
	event := testEvent(t, s.queue)
	if _, err := s.Apply(ctx, event); err != nil {
		t.Fatal(err)
	}
	old := event
	old.ID = newID(t)
	old.AggregateRevision--
	old.Snapshot.Revision--
	old.Snapshot.Result = json.RawMessage(`{"older":true}`)
	result, err := s.Apply(ctx, old)
	if err != nil || result.Applied || result.Duplicate {
		t.Fatal("old version incorrectly applied", err)
	}
	view, err := s.View(ctx, event.AggregateID)
	if err != nil || view.LastRevision != event.AggregateRevision || view.AppliedEvents != 1 || inboxCount(t, ctx, s) != 2 {
		t.Fatal("late event reverted business state", err)
	}
	conflict := event
	conflict.ID = newID(t)
	if _, err = s.Apply(ctx, conflict); !errors.Is(err, ErrRevisionConflict) || inboxCount(t, ctx, s) != 2 {
		t.Fatal("same task revision conflict persisted", err)
	}
	other, err := New(db, s.consumer+"-other", s.queue)
	if err != nil {
		t.Fatal(err)
	}
	if result, err = other.Apply(ctx, event); err != nil || result.Duplicate {
		t.Fatal("different consumer purpose shared dedup scope", err)
	}
	wrongQueue := testStore(t, db)
	if _, err = wrongQueue.Apply(ctx, event); !errors.Is(err, ErrInvalidEvent) {
		t.Fatal("cross-queue event accepted", err)
	}
}
func TestIntegrationReceiverTransactionRollback(t *testing.T) {
	db, _ := integrationDB(t)
	s := testStore(t, db)
	ctx, stop := context.WithTimeout(context.Background(), 10*time.Second)
	defer stop()
	event := testEvent(t, s.queue)
	_, err := db.ExecContext(ctx, `CREATE TRIGGER receiver_test_rollback BEFORE UPDATE ON receiver_task_views FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='injected receiver projection rollback'`)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _, _ = db.Exec(`DROP TRIGGER IF EXISTS receiver_test_rollback`) })
	if _, err = s.Apply(ctx, event); !errors.Is(err, ErrStorage) {
		t.Fatal("injected update did not fail", err)
	}
	var views int
	if err = db.QueryRowContext(ctx, `SELECT COUNT(*) FROM receiver_task_views WHERE consumer_name=? AND queue_id=?`, s.consumer, s.queue).Scan(&views); err != nil || views != 0 || inboxCount(t, ctx, s) != 0 {
		t.Fatal("partial transaction persisted", err)
	}
	if _, err = db.ExecContext(ctx, `DROP TRIGGER receiver_test_rollback`); err != nil {
		t.Fatal(err)
	}
	if result, err := s.Apply(ctx, event); err != nil || !result.Applied || result.Duplicate {
		t.Fatal("rollback poisoned event identity", err)
	}
}