package outbox

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

	orbit "git.yuntuo.hk/k2safe/orbit/sdk/go"
	"github.com/go-sql-driver/mysql"
)

func integrationDB(t *testing.T) (*sql.DB, string) {
	t.Helper()
	dsn := os.Getenv("ORBIT_TEST_SENDER_DSN")
	if dsn == "" {
		t.Skip("requires isolated sender database")
	}
	cfg, err := mysql.ParseDSN(dsn)
	if err != nil || cfg.DBName != "orbit_sender_test" || !strings.HasPrefix(cfg.Addr, "127.0.0.1:") {
		t.Fatal("sender tests require 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, testID(t), testID(t))
	if err != nil {
		t.Fatal(err)
	}
	return s
}
func enqueue(t *testing.T, s *Store) string {
	t.Helper()
	id := testID(t)
	if _, err := s.Enqueue(context.Background(), id, "objects/test", testRequest(s.queue)); err != nil {
		t.Fatal(err)
	}
	return id
}
func status(t *testing.T, s *Store, id string) Record {
	t.Helper()
	r, err := s.Status(context.Background(), id)
	if err != nil {
		t.Fatal(err)
	}
	return r
}
func execSQL(t *testing.T, db *sql.DB, q string, args ...any) {
	t.Helper()
	ctx, stop := context.WithTimeout(context.Background(), 5*time.Second)
	defer stop()
	if _, err := db.ExecContext(ctx, q, args...); err != nil {
		t.Fatal(err)
	}
}
func countRows(t *testing.T, s *Store, table string) int {
	t.Helper()
	if table != "sender_jobs" && table != "sender_outbox" {
		t.Fatal("invalid test table")
	}
	var n int
	query := `SELECT COUNT(*) FROM sender_outbox WHERE namespace_id=? AND queue_id=?`
	if table == "sender_jobs" {
		query = `SELECT COUNT(*) FROM sender_jobs j JOIN sender_outbox o ON o.namespace_id=j.namespace_id AND o.operation_id=j.operation_id WHERE o.namespace_id=? AND o.queue_id=?`
	}
	if err := s.db.QueryRow(query, s.namespace, s.queue).Scan(&n); err != nil {
		t.Fatal(err)
	}
	return n
}
func fakeReceipt(t *testing.T, s *Store) orbit.SubmitTaskResult {
	return orbit.SubmitTaskResult{Task: orbit.Task{ID: testID(t), NamespaceID: s.namespace, QueueID: s.queue, Revision: 1, State: orbit.TaskWaiting}}
}

func TestIntegrationSenderSchema(t *testing.T) {
	db, _ := integrationDB(t)
	var fk int
	if err := db.QueryRow(`SELECT COUNT(*) FROM information_schema.REFERENTIAL_CONSTRAINTS WHERE CONSTRAINT_SCHEMA=DATABASE()`).Scan(&fk); err != nil || fk != 0 {
		t.Fatal(fk, err)
	}
	rows, err := db.Query(`SELECT COLUMN_COMMENT FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME IN ('sender_outbox','sender_jobs')`)
	if err != nil {
		t.Fatal(err)
	}
	defer rows.Close()
	n := 0
	for rows.Next() {
		var c string
		if rows.Scan(&c) != nil || !strings.ContainsFunc(c, func(r rune) bool { return unicode.Is(unicode.Han, r) }) {
			t.Fatal("missing Chinese column comment")
		}
		n++
	}
	if rows.Err() != nil || n != 22 {
		t.Fatal(n, rows.Err())
	}
	rows.Close()
	s := testStore(t, db)
	id := enqueue(t, s)
	for _, assignment := range []string{"state='ACCEPTED'", "send_count=101,lease_epoch=101", "request_json='not json'", "send_deadline_at=TIMESTAMPADD(HOUR,48,created_at)"} {
		_, err = db.Exec(`UPDATE sender_outbox SET `+assignment+` WHERE namespace_id=? AND operation_id=?`, s.namespace, id)
		var rejected *mysql.MySQLError
		if !errors.As(err, &rejected) || rejected.Number != 3819 {
			t.Fatal("invalid persisted state did not violate CHECK", assignment, err)
		}
	}
	var plan string
	if err = db.QueryRow(`EXPLAIN FORMAT=JSON SELECT operation_id FROM sender_outbox FORCE INDEX(ix_sender_due) WHERE namespace_id=? AND queue_id=? AND wake_at<=UTC_TIMESTAMP(6) ORDER BY wake_at,operation_id LIMIT 1 FOR UPDATE SKIP LOCKED`, s.namespace, s.queue).Scan(&plan); err != nil {
		t.Fatal(err)
	}
	if !strings.Contains(plan, `"key": "ix_sender_due"`) || strings.Contains(plan, `"using_filesort": true`) {
		t.Fatal("sender due query lost ordered index", plan)
	}
	t.Log("two business tables, 22 Chinese-commented columns, zero foreign keys")
}
func TestIntegrationSenderConcurrentEnqueueAndConflict(t *testing.T) {
	db, _ := integrationDB(t)
	s := testStore(t, db)
	id := testID(t)
	in := testRequest(s.queue)
	invalid := in
	invalid.BusinessKey = "operation-\xff"
	if _, err := s.Enqueue(context.Background(), id, "objects/test", invalid); !errors.Is(err, ErrInvalid) || countRows(t, s, "sender_jobs") != 0 || countRows(t, s, "sender_outbox") != 0 {
		t.Fatal("invalid UTF-8 changed business state", err)
	}
	type outcome struct {
		duplicate bool
		err       error
	}
	results := make(chan outcome, 8)
	var wg sync.WaitGroup
	for range 8 {
		wg.Go(func() {
			dup, err := s.Enqueue(context.Background(), id, "objects/test", in)
			results <- outcome{dup, err}
		})
	}
	wg.Wait()
	close(results)
	duplicates := 0
	for r := range results {
		if r.err != nil {
			t.Fatal(r.err)
		}
		if r.duplicate {
			duplicates++
		}
	}
	if duplicates != 7 || countRows(t, s, "sender_jobs") != 1 || countRows(t, s, "sender_outbox") != 1 {
		t.Fatal("duplicate business transaction", duplicates)
	}
	if _, err := s.Enqueue(context.Background(), id, "objects/other", in); !errors.Is(err, ErrConflict) {
		t.Fatal("business identity conflict lost", err)
	}
	in.Payload = json.RawMessage(`{"sequence":9007199254740992}`)
	if _, err := s.Enqueue(context.Background(), id, "objects/test", in); !errors.Is(err, ErrConflict) {
		t.Fatal("request content conflict lost", err)
	}
	other, _ := New(db, s.namespace, testID(t))
	if _, err := other.Status(context.Background(), id); !errors.Is(err, ErrNotFound) {
		t.Fatal("cross queue record returned", err)
	}
	if _, err := other.Enqueue(context.Background(), id, "objects/test", testRequest(other.queue)); !errors.Is(err, ErrConflict) {
		t.Fatal("operation was moved to another queue", err)
	}
	if !strings.Contains(string(status(t, s, id).RequestJSON), "9007199254740993") {
		t.Fatal("large integer rounded")
	}
}
func TestIntegrationSenderBusinessRollbackAndConfirmationAtomicity(t *testing.T) {
	db, _ := integrationDB(t)
	s := testStore(t, db)
	id := testID(t)
	execSQL(t, db, `CREATE TRIGGER sender_insert_fail BEFORE INSERT ON sender_jobs FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='injected business insert failure'`)
	t.Cleanup(func() {
		db.Exec(`DROP TRIGGER IF EXISTS sender_insert_fail`)
		db.Exec(`DROP TRIGGER IF EXISTS sender_update_fail`)
	})
	if _, err := s.Enqueue(context.Background(), id, "objects/test", testRequest(s.queue)); !errors.Is(err, ErrStorage) {
		t.Fatal(err)
	}
	if countRows(t, s, "sender_outbox") != 0 || countRows(t, s, "sender_jobs") != 0 {
		t.Fatal("business rollback left a request")
	}
	execSQL(t, db, `DROP TRIGGER sender_insert_fail`)
	if _, err := s.Enqueue(context.Background(), id, "objects/test", testRequest(s.queue)); err != nil {
		t.Fatal(err)
	}
	claim, err := s.claim(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	receipt := fakeReceipt(t, s)
	execSQL(t, db, `CREATE TRIGGER sender_update_fail BEFORE UPDATE ON sender_jobs FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='injected business confirmation failure'`)
	if err = s.finish(context.Background(), claim, "ACCEPTED", "ACCEPTED", 0, &receipt); !errors.Is(err, ErrStorage) {
		t.Fatal(err)
	}
	if r := status(t, s, id); r.State != "SENDING" || r.TaskID != nil {
		t.Fatal("partial confirmation committed")
	}
	execSQL(t, db, `DROP TRIGGER sender_update_fail`)
	if err = s.finish(context.Background(), claim, "ACCEPTED", "ACCEPTED", 0, &receipt); err != nil {
		t.Fatal(err)
	}
	var task string
	if err = db.QueryRow(`SELECT task_id FROM sender_jobs WHERE namespace_id=? AND operation_id=?`, s.namespace, id).Scan(&task); err != nil || task != receipt.Task.ID {
		t.Fatal("business confirmation missing", err)
	}
}
func TestIntegrationSenderConcurrentClaimsAndStaleGeneration(t *testing.T) {
	db, _ := integrationDB(t)
	s := testStore(t, db)
	for range 20 {
		enqueue(t, s)
	}
	claims := make(chan Record, 20)
	errs := make(chan error, 8)
	var wg sync.WaitGroup
	for range 8 {
		wg.Go(func() {
			for {
				r, err := s.claim(context.Background())
				if errors.Is(err, ErrIdle) {
					return
				}
				if err != nil {
					errs <- err
					return
				}
				claims <- r
			}
		})
	}
	wg.Wait()
	close(claims)
	close(errs)
	for err := range errs {
		t.Fatal(err)
	}
	seen := map[string]bool{}
	var first Record
	for r := range claims {
		if seen[r.OperationID] || r.SendCount != 1 {
			t.Fatal("duplicate claim")
		}
		seen[r.OperationID] = true
		first = r
	}
	if len(seen) != 20 {
		t.Fatal("unclaimed records", len(seen))
	}
	execSQL(t, db, `UPDATE sender_outbox SET wake_at=TIMESTAMPADD(SECOND,-1,UTC_TIMESTAMP(6)) WHERE namespace_id=? AND operation_id=?`, s.namespace, first.OperationID)
	replacement, err := s.claim(context.Background())
	if err != nil || replacement.LeaseEpoch != 2 || replacement.IdempotencyKey != first.IdempotencyKey {
		t.Fatal(replacement, err)
	}
	receipt := fakeReceipt(t, s)
	if err = s.finish(context.Background(), first, "ACCEPTED", "ACCEPTED", 0, &receipt); !errors.Is(err, ErrLeaseLost) {
		t.Fatal("stale sender overwrote result", err)
	}
	if err = s.finish(context.Background(), replacement, "ACCEPTED", "ACCEPTED", 0, &receipt); err != nil {
		t.Fatal(err)
	}
}
func TestIntegrationSenderHTTPClassificationAndBudgets(t *testing.T) {
	db, _ := integrationDB(t)
	for _, mode := range []string{"success", "backpressure", "forbidden", "conflict", "unknown", "wrong_target", "invalid_receipt", "corrupt_request", "exhausted", "last_attempt", "deadline"} {
		t.Run(mode, func(t *testing.T) {
			s := testStore(t, db)
			id := enqueue(t, s)
			var posts atomic.Int32
			if mode == "corrupt_request" {
				execSQL(t, db, `UPDATE sender_outbox SET request_json='{}' WHERE namespace_id=?`, s.namespace)
			}
			if mode == "exhausted" {
				execSQL(t, db, `UPDATE sender_outbox SET send_count=100,lease_epoch=100 WHERE namespace_id=?`, s.namespace)
			}
			if mode == "last_attempt" {
				execSQL(t, db, `UPDATE sender_outbox SET send_count=99,lease_epoch=99 WHERE namespace_id=?`, s.namespace)
			}
			if mode == "deadline" {
				execSQL(t, db, `UPDATE sender_outbox SET send_deadline_at=TIMESTAMPADD(MICROSECOND,1,created_at) WHERE namespace_id=?`, s.namespace)
			}
			receipt := fakeReceipt(t, s)
			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
				if r.Method == "GET" {
					ns := s.namespace
					if mode == "wrong_target" {
						ns = testID(t)
					}
					json.NewEncoder(w).Encode(map[string]any{"request_id": id, "data": orbit.Queue{ID: s.queue, NamespaceID: ns}})
					return
				}
				posts.Add(1)
				if r.Header.Get("Idempotency-Key") != "outbox-task-"+id {
					t.Error("wrong submission identity")
				}
				switch mode {
				case "backpressure":
					w.Header().Set("Retry-After", "30")
					w.WriteHeader(429)
					json.NewEncoder(w).Encode(map[string]any{"request_id": id, "error": map[string]string{"code": "QUEUE_FULL", "message": "full"}})
				case "forbidden", "conflict":
					code, status := "PERMISSION_DENIED", 403
					if mode == "conflict" {
						code, status = "IDEMPOTENCY_CONFLICT", 409
					}
					w.WriteHeader(status)
					json.NewEncoder(w).Encode(map[string]any{"request_id": id, "error": map[string]string{"code": code, "message": "rejected"}})
				case "unknown", "last_attempt":
					w.WriteHeader(502)
				default:
					reply := receipt
					if mode == "invalid_receipt" {
						reply.Task.NamespaceID = testID(t)
					}
					json.NewEncoder(w).Encode(map[string]any{"request_id": id, "data": reply})
				}
			}))
			defer server.Close()
			api, _ := orbit.NewClient(server.URL, orbit.Options{Token: "test-token"})
			r, err := s.SendOnce(context.Background(), api)
			if err != nil {
				t.Fatal(err)
			}
			want := "REVIEW"
			if mode == "success" {
				want = "ACCEPTED"
			}
			if mode == "backpressure" || mode == "unknown" {
				want = "PENDING"
			}
			if r.State != want {
				t.Fatal(mode, r.State, r.LastCode)
			}
			if (mode == "wrong_target" || mode == "corrupt_request" || mode == "exhausted" || mode == "deadline") && posts.Load() != 0 {
				t.Fatal("unsafe HTTP sent")
			}
			if mode == "backpressure" && (r.WakeAt == nil || time.Until(*r.WakeAt) < 29*time.Second) {
				t.Fatal("Retry-After was ignored")
			}
		})
	}
}