package mysql

import (
	"context"
	"crypto/sha256"
	"database/sql"
	"errors"
	"os"
	"reflect"
	"sync"
	"testing"
	"time"

	core "git.yuntuo.hk/k2safe/orbit/model"
	platform "git.yuntuo.hk/k2safe/orbit/platform/model"
	pulse "git.yuntuo.hk/k2safe/orbit/pulse/model"
	weave "git.yuntuo.hk/k2safe/orbit/weave/model"
	mysqldriver "github.com/go-sql-driver/mysql"
)

func TestIntegrationSchemaAndConstraints(t *testing.T) {
	dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN")
	if dsn == "" {
		t.Skip("set ORBIT_TEST_MYSQL_DSN to a disposable database, or run make integration")
	}
	ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
	defer cancel()
	db, err := Open(ctx, dsn, 6, 2)
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	var version string
	if err := db.QueryRowContext(ctx, "SELECT VERSION()").Scan(&version); err != nil {
		t.Fatal(err)
	}
	t.Log("MySQL", version)
	var wg sync.WaitGroup
	results := make(chan error, 2)
	for i := 0; i < 2; i++ {
		wg.Add(1)
		go func() { defer wg.Done(); results <- Migrate(ctx, db) }()
	}
	wg.Wait()
	close(results)
	for err := range results {
		if err != nil {
			t.Fatal(err)
		}
	}
	if err := CheckSchema(ctx, db); err != nil {
		t.Fatal(err)
	}
	exec := func(q string, args ...any) {
		t.Helper()
		if _, err := db.ExecContext(ctx, q, args...); err != nil {
			t.Fatal(err)
		}
	}
	id := func() core.ID {
		t.Helper()
		v, err := core.NewID()
		if err != nil {
			t.Fatal(err)
		}
		return v
	}
	reject := func(code uint16, q string, args ...any) {
		t.Helper()
		_, err := db.ExecContext(ctx, q, args...)
		var me *mysqldriver.MySQLError
		if !errors.As(err, &me) || me.Number != code {
			t.Fatalf("wanted MySQL %d, got %v", code, err)
		}
	}
	app, nsA, nsB, queueA := id(), id(), id(), id()
	exec("INSERT INTO orbit_applications(id,name) VALUES (?,?)", app, string(app))
	for _, ns := range []core.ID{nsA, nsB} {
		exec("INSERT INTO orbit_namespaces(id,application_id,name,environment) VALUES (?,?,?,'test')", ns, app, string(ns))
	}
	exec("INSERT INTO pulse_queues(namespace_id,id,name,max_concurrency,rate_per_second,policy) VALUES (?,?,?,4,10,'{}')", nsA, queueA, string(queueA))
	taskSQL := `INSERT INTO pulse_tasks(namespace_id,id,queue_id,handler,handler_version,payload_version,payload,policy_snapshot,available_at,deadline_at,attempt_timeout_ms,max_attempts)
        VALUES (?,?,?,'object.prepare','1','1','{}','{}',UTC_TIMESTAMP(6),DATE_ADD(UTC_TIMESTAMP(6),INTERVAL 1 HOUR),5000,3)`
	taskA, taskB := id(), id()
	exec(taskSQL, nsA, taskA, queueA)
	exec(taskSQL, nsA, taskB, queueA)
	t.Run("namespace_isolation_and_safe_task_states", func(t *testing.T) {
		reject(1452, taskSQL, nsB, id(), queueA)
		reject(3819, "UPDATE pulse_tasks SET state='RUNNING' WHERE namespace_id=? AND id=?", nsA, taskA)
		reject(3819, "UPDATE pulse_tasks SET state='UNRECOGNIZED' WHERE namespace_id=? AND id=?", nsA, taskA)
		reject(3819, "UPDATE pulse_tasks SET state='SUCCEEDED' WHERE namespace_id=? AND id=?", nsA, taskA)
		reject(3819, "UPDATE pulse_tasks SET attempt_count=4 WHERE namespace_id=? AND id=?", nsA, taskA)
		var scanned core.ID
		var available time.Time
		if err := db.QueryRowContext(ctx, "SELECT id,available_at FROM pulse_tasks WHERE namespace_id=? AND id=?", nsA, taskA).Scan(&scanned, &available); err != nil || scanned != taskA || available.Location() != time.UTC {
			t.Fatalf("binary/time roundtrip: %s %v %v", scanned, available, err)
		}
	})
	t.Run("transaction_rollback", func(t *testing.T) {
		sentinel := errors.New("abort workflow update")
		err := WithTx(ctx, db, func(tx *sql.Tx) error {
			if _, err := tx.ExecContext(ctx, "UPDATE pulse_queues SET paused=TRUE WHERE namespace_id=? AND id=?", nsA, queueA); err != nil {
				return err
			}
			return sentinel
		})
		if !errors.Is(err, sentinel) {
			t.Fatal(err)
		}
		var paused bool
		if err := db.QueryRowContext(ctx, "SELECT paused FROM pulse_queues WHERE namespace_id=? AND id=?", nsA, queueA).Scan(&paused); err != nil || paused {
			t.Fatalf("rollback failed: %v %v", paused, err)
		}
	})
	hash := core.Digest(sha256.Sum256([]byte("test-operation")))
	t.Run("idempotency_scope", func(t *testing.T) {
		q := `INSERT INTO orbit_idempotency_keys(namespace_id,id,operation,scope_key,key_digest,request_digest,resource_kind,resource_id,response,expires_at) VALUES (?,? ,?,'',?,?,'task',?,'{}',DATE_ADD(UTC_TIMESTAMP(6),INTERVAL 1 DAY))`
		exec(q, nsA, id(), "task.submit", hash, hash, taskA)
		reject(1062, q, nsA, id(), "task.submit", hash, hash, taskB)
		exec(q, nsB, id(), "task.submit", hash, hash, taskB)
		exec(q, nsA, id(), "workflow.start", hash, hash, taskB)
	})
	definition, defVersion, gate, workflow, step, activation := id(), id(), id(), id(), id(), id()
	exec("INSERT INTO weave_definitions(namespace_id,id,name) VALUES (?,?,?)", nsA, definition, string(definition))
	versionSQL := "INSERT INTO weave_definition_versions(namespace_id,id,definition_id,version,schema_version,specification,content_digest,published_by) VALUES (?,?,?,'v1',1,'{}',?,'test')"
	exec(versionSQL, nsA, defVersion, definition, hash)
	exec("INSERT INTO pulse_execution_gates(namespace_id,id) VALUES (?,?)", nsA, gate)
	exec(`INSERT INTO weave_instances(namespace_id,id,definition_id,definition_version_id,gate_id,input,policy_snapshot,deadline_at) VALUES (?,?,?,?,?,'{}','{}',DATE_ADD(UTC_TIMESTAMP(6),INTERVAL 1 HOUR))`, nsA, workflow, definition, defVersion, gate)
	exec("INSERT INTO weave_steps(namespace_id,id,workflow_id,`key`,kind) VALUES (?,?,?,'poll','POLL')", nsA, step, workflow)
	exec("INSERT INTO weave_activations(namespace_id,id,workflow_id,step_id,sequence,input) VALUES (?,?,?,?,1,'{}')", nsA, activation, workflow, step)
	exec("UPDATE weave_steps SET current_activation_id=? WHERE namespace_id=? AND id=?", activation, nsA, step)
	t.Run("published_definitions_are_immutable", func(t *testing.T) {
		reject(1062, versionSQL, nsA, id(), definition, hash)
		reject(1644, "UPDATE weave_definition_versions SET specification='{}' WHERE namespace_id=? AND id=?", nsA, defVersion)
	})
	t.Run("poll_identity_and_budget", func(t *testing.T) {
		poll := id()
		exec("INSERT INTO weave_poll_states(namespace_id,id,workflow_id,activation_id,max_rounds,max_attempts_per_round,interval_ms,deadline_at) VALUES (?,?,?,?,2,3,1000,DATE_ADD(UTC_TIMESTAMP(6),INTERVAL 1 HOUR))", nsA, poll, workflow, activation)
		q := "INSERT INTO weave_poll_rounds(namespace_id,id,workflow_id,activation_id,round,task_id) VALUES (?,?,?,?,?,?)"
		exec(q, nsA, id(), workflow, activation, 1, taskA)
		reject(1062, q, nsA, id(), workflow, activation, 1, taskB)
		reject(1062, q, nsA, id(), workflow, activation, 2, taskA)
		reject(3819, "UPDATE weave_poll_states SET used_rounds=3 WHERE namespace_id=? AND id=?", nsA, poll)
	})
	t.Run("events_and_audit_are_append_only", func(t *testing.T) {
		event, audit := id(), id()
		q := "INSERT INTO orbit_events(namespace_id,id,aggregate_kind,aggregate_id,aggregate_revision,type,payload,occurred_at) VALUES (?,?,'task',?,1,'task.accepted','{}',UTC_TIMESTAMP(6))"
		exec(q, nsA, event, taskA)
		reject(1062, q, nsA, id(), taskA)
		reject(1644, "UPDATE orbit_events SET payload='{}' WHERE namespace_id=? AND id=?", nsA, event)
		exec("INSERT INTO orbit_audit_events(namespace_id,id,operation_id,actor_id,action,resource_kind,resource_id,reason) VALUES (?,?,?,'test','pause','workflow',?,'integration test')", nsA, audit, id(), workflow)
		reject(1644, "UPDATE orbit_audit_events SET reason='rewritten' WHERE namespace_id=? AND id=?", nsA, audit)
	})
	t.Run("model_columns_match_migrated_database", func(t *testing.T) { checkModelColumns(t, ctx, db) })
	t.Run("dirty_or_modified_migration_refuses_startup", func(t *testing.T) {
		exec("UPDATE orbit_schema_migrations SET dirty=TRUE WHERE version=1")
		if err := Migrate(ctx, db); err == nil {
			t.Fatal("continued dirty migration")
		}
		if err := CheckSchema(ctx, db); err == nil {
			t.Fatal("accepted dirty schema")
		}
		exec("UPDATE orbit_schema_migrations SET dirty=FALSE,checksum=REPEAT('0',64) WHERE version=1")
		if err := Migrate(ctx, db); err == nil {
			t.Fatal("accepted modified migration")
		}
		migrations, err := Migrations()
		if err != nil {
			t.Fatal(err)
		}
		exec("UPDATE orbit_schema_migrations SET checksum=? WHERE version=1", migrations[0].Checksum)
		if err := CheckSchema(ctx, db); err != nil {
			t.Fatal(err)
		}
	})
}

type tableModel interface{ TableName() string }

func checkModelColumns(t *testing.T, ctx context.Context, db *sql.DB) {
	t.Helper()
	models := []tableModel{
		platform.Application{},
		platform.Namespace{},
		platform.Credential{},
		platform.IdempotencyKey{},
		platform.Event{},
		platform.Subscription{},
		platform.Delivery{},
		platform.AuditEvent{},
		pulse.Queue{},
		pulse.ExecutionGate{},
		pulse.WorkerSession{},
		pulse.Task{},
		pulse.Attempt{},
		pulse.AttemptReceipt{},
		pulse.ConcurrencyLimit{},
		pulse.Permit{},
		pulse.Schedule{},
		pulse.ScheduleVersion{},
		pulse.Occurrence{},
		weave.Definition{},
		weave.DefinitionVersion{},
		weave.Instance{},
		weave.Step{},
		weave.Activation{},
		weave.Wait{},
		weave.Signal{},
		weave.PollState{},
		weave.PollRound{},
		weave.InboxEvent{},
		weave.PendingAdvance{},
	}
	for _, m := range models {
		fields := map[string]bool{}
		var visit func(reflect.Type)
		visit = func(typ reflect.Type) {
			for i := 0; i < typ.NumField(); i++ {
				f := typ.Field(i)
				if f.Anonymous {
					visit(f.Type)
				} else if col := f.Tag.Get("db"); col != "" {
					fields[col] = true
				}
			}
		}
		visit(reflect.TypeOf(m))
		rows, err := db.QueryContext(ctx, "SELECT COLUMN_NAME FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=?", m.TableName())
		if err != nil {
			t.Fatal(err)
		}
		count := 0
		for rows.Next() {
			var column string
			if err := rows.Scan(&column); err != nil {
				t.Fatal(err)
			}
			count++
			if !fields[column] {
				t.Errorf("%s.%s has no Go field", m.TableName(), column)
			}
			delete(fields, column)
		}
		err = rows.Err()
		rows.Close()
		if err != nil {
			t.Fatal(err)
		}
		if count == 0 || len(fields) > 0 {
			t.Errorf("%s missing columns: %v", m.TableName(), fields)
		}
	}
}