//go:build integration

// Package autonumber integration tests — require a real TiDB/MySQL-
// compatible DB via TIDB_TEST_DSN (project memory: 4Qbn3F3nyaU9DLk.root).
package autonumber

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"os"
	"sync"
	"testing"
	"time"

	_ "github.com/go-sql-driver/mysql"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

// openIntegrationDB opens the test DB. Skips the test if
// TIDB_TEST_DSN is unset or the DB is unreachable.
func openIntegrationDB(t *testing.T) *sql.DB {
	t.Helper()
	dsn := os.Getenv("TIDB_TEST_DSN")
	if dsn == "" {
		t.Skip("TIDB_TEST_DSN not set; skipping autonumber integration test")
	}
	db, err := sql.Open("mysql", dsn)
	if err != nil {
		t.Fatalf("sql.Open: %v", err)
	}
	if err := db.Ping(); err != nil {
		_ = db.Close()
		t.Skipf("TIDB_TEST_DSN set but DB unreachable (%v); skipping", err)
	}
	db.SetMaxOpenConns(10)
	t.Cleanup(func() { _ = db.Close() })
	return db
}

// ensureCounterTable creates the counter table if it's not already
// present. The Wave 1 schema test guarantees the bootstrap pipeline
// lands this table; the integration tests below may run on a branch
// that hasn't been bootstrapped yet, so we DDL defensively.
func ensureCounterTable(t *testing.T, db *sql.DB) {
	t.Helper()
	const ddl = `CREATE TABLE IF NOT EXISTS _System_AutoNumberCounter (
	  Id               VARCHAR(255) NOT NULL PRIMARY KEY,
	  FieldId          VARCHAR(255) NOT NULL,
	  ScopeKey         VARCHAR(50)  NOT NULL DEFAULT '',
	  NextValue        BIGINT       NOT NULL DEFAULT 1,
	  CreatedDate      DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP,
	  LastModifiedDate DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP,
	  UNIQUE KEY uk_field_scope (FieldId, ScopeKey),
	  KEY idx_field (FieldId)
	)`
	if _, err := db.Exec(ddl); err != nil {
		t.Fatalf("ensureCounterTable: %v", err)
	}
}

// resetField deletes all counter rows for a given fieldID so a test
// starts from a known state.
func resetField(t *testing.T, db *sql.DB, fieldID string) {
	t.Helper()
	_, err := db.Exec("DELETE FROM _System_AutoNumberCounter WHERE FieldId = ?", fieldID)
	require.NoError(t, err)
}

// seedGlobalCounter seeds one scope=global row at the given
// NextValue. Mirrors SeedFieldCounter but test-local to avoid
// coupling to the engine init path.
func seedGlobalCounter(t *testing.T, db *sql.DB, fieldID string, nextValue int64) {
	t.Helper()
	rowID := generateCounterRowID(fieldID, "")
	_, err := db.Exec(
		`INSERT INTO _System_AutoNumberCounter
           (Id, FieldId, ScopeKey, NextValue, CreatedDate, LastModifiedDate)
         VALUES (?, ?, '', ?, NOW(), NOW())
         ON DUPLICATE KEY UPDATE NextValue = VALUES(NextValue)`,
		rowID, fieldID, nextValue,
	)
	require.NoError(t, err)
}

// TestReserve_SimpleGlobal — test 17 smoke: two sequential reserveOne
// returns distinct contiguous values.
func TestReserve_SimpleGlobal(t *testing.T) {
	db := openIntegrationDB(t)
	ensureCounterTable(t, db)
	fieldID := "fld_testReserveSimpleGlobal_" + randomSuffix()
	defer resetField(t, db, fieldID)

	seedGlobalCounter(t, db, fieldID, 1)

	r := newReserver(db, db, 4, newReservationMetrics())

	in := reserveInput{fieldID: fieldID, scope: ScopeGlobal, startValue: 1}
	_, v1, err := r.reserveOne(context.Background(), in)
	require.NoError(t, err)
	_, v2, err := r.reserveOne(context.Background(), in)
	require.NoError(t, err)
	assert.Equal(t, int64(1), v1)
	assert.Equal(t, int64(2), v2)
}

// TestReserve_FirstUseRace_Global — test 17: 50 concurrent
// goroutines reserve against a pre-seeded global counter; all values
// distinct + contiguous.
func TestReserve_FirstUseRace_Global(t *testing.T) {
	db := openIntegrationDB(t)
	ensureCounterTable(t, db)
	fieldID := "fld_testReserveRace_" + randomSuffix()
	defer resetField(t, db, fieldID)
	seedGlobalCounter(t, db, fieldID, 1)

	r := newReserver(db, db, 4, newReservationMetrics())
	const N = 50

	var wg sync.WaitGroup
	values := make([]int64, N)
	errs := make([]error, N)
	for i := 0; i < N; i++ {
		wg.Add(1)
		go func(idx int) {
			defer wg.Done()
			in := reserveInput{fieldID: fieldID, scope: ScopeGlobal, startValue: 1}
			_, v, err := r.reserveOne(context.Background(), in)
			values[idx] = v
			errs[idx] = err
		}(i)
	}
	wg.Wait()

	seen := map[int64]bool{}
	for i, v := range values {
		require.NoErrorf(t, errs[i], "goroutine %d errored", i)
		assert.Falsef(t, seen[v], "duplicate value %d", v)
		seen[v] = true
		assert.GreaterOrEqual(t, v, int64(1))
		assert.LessOrEqual(t, v, int64(N))
	}
	assert.Len(t, seen, N, "all N values must be unique")
}

// TestReserve_MissingGlobalRow_IsCorruption — test 17b: pre-delete a
// scope=global counter row; reserveOne returns ErrCounterMissing
// (does NOT silently reseed). Negative assertion on UPSERT path.
func TestReserve_MissingGlobalRow_IsCorruption(t *testing.T) {
	db := openIntegrationDB(t)
	ensureCounterTable(t, db)
	fieldID := "fld_testMissingGlobal_" + randomSuffix()
	defer resetField(t, db, fieldID)

	// Explicitly do NOT seed; simulate corruption.
	r := newReserver(db, db, 4, newReservationMetrics())
	in := reserveInput{fieldID: fieldID, scope: ScopeGlobal, startValue: 1}
	_, _, err := r.reserveOne(context.Background(), in)
	require.Error(t, err)
	assert.True(t, errors.Is(err, ErrCounterMissing),
		"missing scope=global row must surface ErrCounterMissing")

	// NEGATIVE ASSERT: no row was silently created.
	var cnt int
	err = db.QueryRow(
		"SELECT COUNT(*) FROM _System_AutoNumberCounter WHERE FieldId = ?",
		fieldID,
	).Scan(&cnt)
	require.NoError(t, err)
	assert.Equal(t, 0, cnt, "global corruption path MUST NOT silently UPSERT")
}

// TestReserve_MissingPerDay_OwnTxnUpsert — test 17a: new day for
// scope=per_day creates a row via own-txn UPSERT and reserveOne
// returns the initialNext value (no outer-txn rollback racing).
func TestReserve_MissingPerDay_OwnTxnUpsert(t *testing.T) {
	db := openIntegrationDB(t)
	ensureCounterTable(t, db)
	fieldID := "fld_testMissingPerDay_" + randomSuffix()
	defer resetField(t, db, fieldID)

	r := newReserver(db, db, 4, newReservationMetrics())
	in := reserveInput{fieldID: fieldID, scope: ScopePerDay, startValue: 1}
	_, v, err := r.reserveOne(context.Background(), in)
	require.NoError(t, err)
	assert.Equal(t, int64(1), v)

	// Row now exists for today's scope key; second reserve returns 2.
	_, v2, err := r.reserveOne(context.Background(), in)
	require.NoError(t, err)
	assert.Equal(t, int64(2), v2)

	// Exactly one row exists.
	var cnt int
	err = db.QueryRow(
		"SELECT COUNT(*) FROM _System_AutoNumberCounter WHERE FieldId = ?",
		fieldID,
	).Scan(&cnt)
	require.NoError(t, err)
	assert.Equal(t, 1, cnt)
}

// TestReserve_CounterOverflow_RendersCleanly — ensures reserveOne by
// itself does NOT return overflow; overflow is the renderer's
// responsibility (PRD §5.1 step 5). Smoke-check at the maxWidth=4
// boundary to guard against regressions.
func TestReserve_NoOverflowGuardOnReserve(t *testing.T) {
	db := openIntegrationDB(t)
	ensureCounterTable(t, db)
	fieldID := "fld_testNoOverflow_" + randomSuffix()
	defer resetField(t, db, fieldID)

	// Seed near the 4-digit limit.
	seedGlobalCounter(t, db, fieldID, 9998)

	r := newReserver(db, db, 4, newReservationMetrics())
	in := reserveInput{fieldID: fieldID, scope: ScopeGlobal, startValue: 1}

	_, v, err := r.reserveOne(context.Background(), in)
	require.NoError(t, err)
	assert.Equal(t, int64(9998), v)
	_, v, err = r.reserveOne(context.Background(), in)
	require.NoError(t, err)
	assert.Equal(t, int64(9999), v)
	_, v, err = r.reserveOne(context.Background(), in)
	require.NoError(t, err)
	assert.Equal(t, int64(10000), v, "reserver itself does not enforce width; overflow is renderer job")
}

// TestReserveBatch_SingleBucket — test 19: batch of 100 returns 100
// contiguous values from one reservation.
func TestReserveBatch_SingleBucket(t *testing.T) {
	db := openIntegrationDB(t)
	ensureCounterTable(t, db)
	fieldID := "fld_testReserveBatchSingle_" + randomSuffix()
	defer resetField(t, db, fieldID)
	seedGlobalCounter(t, db, fieldID, 1)

	r := newReserver(db, db, 4, newReservationMetrics())
	in := reserveInput{fieldID: fieldID, scope: ScopeGlobal, startValue: 1}
	const N = 100
	_, start, err := r.reserveBatch(context.Background(), in, N)
	require.NoError(t, err)
	assert.Equal(t, int64(1), start)

	// Next reserveOne should yield N+1.
	_, v, err := r.reserveOne(context.Background(), in)
	require.NoError(t, err)
	assert.Equal(t, int64(N+1), v)
}

// TestAssignBatch_MixedBuckets_Throughput — test 19b (F13): 10 scope
// keys × 100 records each must complete via 10 reserveBatch calls
// (NOT 1000× reserveOne) and hit ≥ 200 records/sec throughput on a
// real TiDB.
func TestAssignBatch_MixedBuckets_Throughput(t *testing.T) {
	db := openIntegrationDB(t)
	ensureCounterTable(t, db)

	fieldID := "fld_testMixedBuckets_" + randomSuffix()
	defer resetField(t, db, fieldID)
	seedGlobalCounter(t, db, fieldID, 1)

	// Metadata provider with the field configured. Uses a cross-record
	// token so bucket partitioning is meaningful.
	rule := Rule{
		FieldID: fieldID,
		Format:  "{Account.BillingCountryCode}-A-{00000}",
		Scope:   ScopeGlobal,
		Start:   1,
		TokenRefs: []CrossRecordRef{{
			TokenLiteral: "{Account.BillingCountryCode}",
			TargetObject: "Account",
			TargetField:  "BillingCountryCode",
			ReadStrategy: "caller_txn",
			DirectField:  "Account_Country_Code",
		}},
	}
	provider := &fakeMetadataProvider{rules: []Rule{rule}}

	e, err := NewEngine(WithReservationDB(db), WithMetadataProvider(provider))
	require.NoError(t, err)

	// Build 10 buckets × 100 records each.
	countries := []string{"SG", "US", "JP", "CN", "DE", "FR", "UK", "IN", "AU", "BR"}
	records := make([]models.SObject, 0, 1000)
	for _, code := range countries {
		for i := 0; i < 100; i++ {
			records = append(records, models.SObject{
				"Account_Country_Code": code,
			})
		}
	}

	startTime := time.Now()
	out, err := e.AssignBatch(context.Background(), fieldID, records)
	elapsed := time.Since(startTime)
	require.NoError(t, err)
	require.Len(t, out, 1000)

	// Throughput: 200 records/sec threshold.
	throughput := float64(len(out)) / elapsed.Seconds()
	assert.GreaterOrEqualf(t, throughput, 200.0,
		"throughput %.2f rec/s < 200; mixed buckets must NOT degrade to reserveOne", throughput)

	// Distinctness: 1000 unique values.
	seen := map[string]bool{}
	for _, s := range out {
		require.NotEmpty(t, s)
		assert.Falsef(t, seen[s], "duplicate rendered value %q", s)
		seen[s] = true
	}
	assert.Len(t, seen, 1000)
}

// TestAssignBatch_ThunderingHerd — test 22a: concurrent exhaust of
// optimistic retries on the same counter must serialize via the
// per-counter semaphore without duplicating counter values.
func TestAssignBatch_ThunderingHerd(t *testing.T) {
	db := openIntegrationDB(t)
	ensureCounterTable(t, db)
	fieldID := "fld_testThunder_" + randomSuffix()
	defer resetField(t, db, fieldID)
	seedGlobalCounter(t, db, fieldID, 1)

	r := newReserver(db, db, 4, newReservationMetrics())
	const N = 30

	var wg sync.WaitGroup
	values := make([]int64, N)
	errs := make([]error, N)
	for i := 0; i < N; i++ {
		wg.Add(1)
		go func(idx int) {
			defer wg.Done()
			in := reserveInput{fieldID: fieldID, scope: ScopeGlobal, startValue: 1}
			_, v, err := r.reserveOne(context.Background(), in)
			values[idx] = v
			errs[idx] = err
		}(i)
	}
	wg.Wait()

	seen := map[int64]bool{}
	for i, v := range values {
		require.NoErrorf(t, errs[i], "goroutine %d errored: %v", i, errs[i])
		assert.Falsef(t, seen[v], "duplicate reserved value %d", v)
		seen[v] = true
	}
	assert.Len(t, seen, N)
}

// --- small helpers ---

// randomSuffix yields a test-fixture-unique field id suffix so
// parallel test instances don't collide on the same row.
func randomSuffix() string {
	return fmt.Sprintf("%d_%d", time.Now().UnixNano(), 1+int(time.Now().UnixNano()%10000))
}

// silences unused-import if go-sql-driver/mysql is not exercised
// elsewhere in this file (it is via sql.Open).
var _ = sql.Open