package pgsql

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"testing"
	"time"

	"github.com/jackc/pgx/v5"

	"git.warky.dev/wdevs/relspecgo/pkg/models"
	"git.warky.dev/wdevs/relspecgo/pkg/writers"
)

func liveWriterConn(t *testing.T) string {
	t.Helper()
	conn := os.Getenv("RELSPEC_TEST_PG_CONN")
	if conn == "" {
		t.Skip("RELSPEC_TEST_PG_CONN not set")
	}
	return conn
}

// liveWriterSchema returns a unique schema name and drops it on cleanup.
func liveWriterSchema(t *testing.T, connString string) (string, *pgx.Conn) {
	t.Helper()
	ctx := context.Background()
	conn, err := pgx.Connect(ctx, connString)
	if err != nil {
		t.Fatalf("connect: %v", err)
	}
	name := fmt.Sprintf("pgw_test_%d", time.Now().UnixNano())
	t.Cleanup(func() {
		_, _ = conn.Exec(ctx, "DROP SCHEMA IF EXISTS "+name+" CASCADE")
		_ = conn.Close(ctx)
	})
	return name, conn
}

func liveModel(schemaName string, columns map[string]string) *models.Database {
	db := models.InitDatabase("live")
	s := models.InitSchema(schemaName)
	tbl := models.InitTable("accounts", schemaName)
	id := models.InitColumn("id", "accounts", schemaName)
	id.Type = "integer"
	id.NotNull = true
	id.IsPrimaryKey = true
	tbl.Columns["id"] = id
	for name, typ := range columns {
		c := models.InitColumn(name, "accounts", schemaName)
		c.Type = typ
		tbl.Columns[name] = c
	}
	s.Tables = append(s.Tables, tbl)
	db.Schemas = append(db.Schemas, s)
	return db
}

func runLiveWrite(t *testing.T, connString string, db *models.Database, meta map[string]interface{}) (*ExecutionReport, error) {
	t.Helper()
	m := map[string]interface{}{"connection_string": connString}
	for k, v := range meta {
		m[k] = v
	}
	w := NewWriter(&writers.WriterOptions{Metadata: m})
	err := w.WriteDatabase(db)
	return w.executionReport, err
}

func columnExists(t *testing.T, conn *pgx.Conn, schema, table, column string) bool {
	t.Helper()
	var ok bool
	err := conn.QueryRow(context.Background(),
		`SELECT EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema=$1 AND table_name=$2 AND column_name=$3)`,
		schema, table, column).Scan(&ok)
	if err != nil {
		t.Fatal(err)
	}
	return ok
}

func TestLive_WriteDatabaseEmptyThenIdenticalThenDrifted(t *testing.T) {
	connString := liveWriterConn(t)
	schema, conn := liveWriterSchema(t, connString)
	reportPath := filepath.Join(t.TempDir(), "report.json")
	meta := map[string]interface{}{"report_path": reportPath}

	// Empty database: schema and table are created.
	rep, err := runLiveWrite(t, connString, liveModel(schema, map[string]string{"name": "text"}), meta)
	if err != nil {
		t.Fatal(err)
	}
	if rep.FailedStatements != 0 || rep.ExecutedStatements == 0 {
		t.Fatalf("first run report: %+v", rep)
	}
	if !columnExists(t, conn, schema, "accounts", "name") {
		t.Fatal("column name not created")
	}
	data, err := os.ReadFile(reportPath)
	if err != nil {
		t.Fatalf("report not written: %v", err)
	}
	var onDisk ExecutionReport
	if err := json.Unmarshal(data, &onDisk); err != nil || onDisk.TotalStatements != rep.TotalStatements {
		t.Errorf("report on disk mismatch: %v %+v", err, onDisk)
	}
	created := false
	for _, s := range rep.Schemas {
		for _, tb := range s.Tables {
			if tb.Name == "accounts" && tb.Created {
				created = true
			}
		}
	}
	if !created {
		t.Errorf("table creation not tracked: %+v", rep.Schemas)
	}

	// Identical database: nothing to execute.
	rep, err = runLiveWrite(t, connString, liveModel(schema, map[string]string{"name": "text"}), nil)
	if err != nil {
		t.Fatal(err)
	}
	if rep.TotalStatements != 0 {
		t.Errorf("identical DB must produce no statements, got %d", rep.TotalStatements)
	}

	// Drifted database: only the new column is added.
	rep, err = runLiveWrite(t, connString, liveModel(schema, map[string]string{"name": "text", "email": "text"}), nil)
	if err != nil {
		t.Fatal(err)
	}
	if rep.FailedStatements != 0 || rep.TotalStatements == 0 {
		t.Errorf("drift report: %+v", rep)
	}
	if !columnExists(t, conn, schema, "accounts", "email") {
		t.Error("drifted column email not added")
	}
}

func TestLive_WriteDatabaseFailedStatementReturnsError(t *testing.T) {
	connString := liveWriterConn(t)
	schema, _ := liveWriterSchema(t, connString)
	reportPath := filepath.Join(t.TempDir(), "report.json")

	db := liveModel(schema, map[string]string{"bad": "no_such_type_xyz"})
	rep, err := runLiveWrite(t, connString, db, map[string]interface{}{"full_ddl": true, "report_path": reportPath})
	var failure *ExecutionFailure
	if !errors.As(err, &failure) {
		t.Fatalf("expected structured failure, got: %v", err)
	}
	if rep.FailedStatements == 0 || len(rep.Errors) != rep.FailedStatements {
		t.Fatalf("expected recorded failures: %+v", rep)
	}
	e := rep.Errors[0]
	if e.StatementNumber == 0 || e.Statement == "" || e.Error == "" {
		t.Errorf("incomplete error entry: %+v", e)
	}
	if _, err := os.Stat(reportPath); err != nil {
		t.Errorf("report must be written even on failures: %v", err)
	}
	failedTable := false
	for _, s := range rep.Schemas {
		for _, tb := range s.Tables {
			if tb.Name == "accounts" && !tb.Created && tb.Error != "" {
				failedTable = true
			}
		}
	}
	if !failedTable {
		t.Errorf("failed table creation not tracked: %+v", rep.Schemas)
	}
}

func TestLive_WriteDatabaseFlattenFallsBackToFullDDL(t *testing.T) {
	connString := liveWriterConn(t)
	schema, conn := liveWriterSchema(t, connString)

	w := NewWriter(&writers.WriterOptions{
		FlattenSchema: true,
		Metadata:      map[string]interface{}{"connection_string": connString},
	})
	if err := w.WriteDatabase(liveModel(schema, nil)); err != nil {
		t.Fatal(err)
	}
	// Flattened output lands in public as <schema>_<table>.
	flat := "public." + schema + "_accounts"
	t.Cleanup(func() { _, _ = conn.Exec(context.Background(), "DROP TABLE IF EXISTS "+flat+" CASCADE") })
	var ok bool
	if err := conn.QueryRow(context.Background(), "SELECT to_regclass($1) IS NOT NULL", flat).Scan(&ok); err != nil || !ok {
		t.Errorf("flattened table %s not created (ok=%v err=%v)", flat, ok, err)
	}
}

func TestGenerateLiveDiffStatements_FlattenRejected(t *testing.T) {
	w := NewWriter(&writers.WriterOptions{FlattenSchema: true})
	if _, err := w.generateLiveDiffStatements(models.InitDatabase("x"), "postgres://unused"); err == nil {
		t.Error("flatten must be rejected before connecting")
	}
}

func TestExecuteStatements_ConnectFailure(t *testing.T) {
	w := NewWriter(&writers.WriterOptions{})
	w.executionReport = &ExecutionReport{}
	err := w.executeStatements([]string{"SELECT 1"}, "postgres://nobody:x@127.0.0.1:1/none?connect_timeout=1")
	if err == nil {
		t.Error("expected connect failure")
	}
	if w.executionReport.TotalStatements != 1 {
		t.Errorf("total not recorded: %+v", w.executionReport)
	}
}

func TestLive_ExecuteStatementsSkipsCommentsAndBlank(t *testing.T) {
	connString := liveWriterConn(t)
	schema, conn := liveWriterSchema(t, connString)

	w := NewWriter(&writers.WriterOptions{})
	w.executionReport = &ExecutionReport{}
	stmts := []string{
		"-- Schema: " + schema,
		"   ",
		"CREATE SCHEMA " + schema,
		"CREATE TABLE " + schema + ".t (id int)",
		"-- plain comment",
	}
	if err := w.executeStatements(stmts, connString); err != nil {
		t.Fatal(err)
	}
	r := w.executionReport
	if r.ExecutedStatements != 2 || r.FailedStatements != 0 || r.TotalStatements != 5 || r.SkippedStatements != 3 {
		t.Errorf("counts: %+v", r)
	}
	if len(r.Schemas) != 1 || r.Schemas[0].Name != schema || len(r.Schemas[0].Tables) != 1 || !r.Schemas[0].Tables[0].Created {
		t.Errorf("schema tracking: %+v", r.Schemas)
	}
	var ok bool
	if err := conn.QueryRow(context.Background(), "SELECT to_regclass($1) IS NOT NULL", schema+".t").Scan(&ok); err != nil || !ok {
		t.Errorf("table not created: %v", err)
	}
}

func TestLive_ExecuteStatementsFailureModes(t *testing.T) {
	connString := liveWriterConn(t)
	for _, continueOnError := range []bool{false, true} {
		t.Run(fmt.Sprintf("continue=%v", continueOnError), func(t *testing.T) {
			schema, conn := liveWriterSchema(t, connString)
			reportPath := filepath.Join(t.TempDir(), "report.json")
			w := NewWriter(&writers.WriterOptions{
				ContinueOnError: continueOnError,
				Metadata:        map[string]interface{}{"report_path": reportPath},
			})
			w.executionReport = &ExecutionReport{}
			err := w.executeStatements([]string{
				"CREATE SCHEMA " + schema,
				"SELECT 1 / 0",
				"CREATE TABLE " + schema + ".after_failure (id int)",
			}, connString)
			var failure *ExecutionFailure
			if !errors.As(err, &failure) || failure.Report != w.executionReport {
				t.Fatalf("expected structured execution failure, got %v", err)
			}
			var report ExecutionReport
			data, err := os.ReadFile(reportPath)
			if err != nil {
				t.Fatal(err)
			}
			if err := json.Unmarshal(data, &report); err != nil {
				t.Fatal(err)
			}
			executed, skipped := 1, 1
			if continueOnError {
				executed, skipped = 2, 0
			}
			if report.TotalStatements != 3 || report.ExecutedStatements != executed || report.FailedStatements != 1 || report.SkippedStatements != skipped || report.EndTime == "" {
				t.Fatalf("incorrect partial application report: %+v", report)
			}
			if len(report.Errors) != 1 || report.Errors[0].StatementNumber != 2 {
				t.Fatalf("incorrect failure details: %+v", report.Errors)
			}
			if exists := columnExists(t, conn, schema, "after_failure", "id"); exists != continueOnError {
				t.Fatalf("later statement executed=%v, continue=%v", exists, continueOnError)
			}
		})
	}
}

func TestLive_IncompatibleTypeChangePreservesColumnAndData(t *testing.T) {
	connString := liveWriterConn(t)
	schema, conn := liveWriterSchema(t, connString)
	ctx := context.Background()
	_, err := conn.Exec(ctx, "CREATE SCHEMA "+schema+"; CREATE TABLE "+schema+".accounts (id integer PRIMARY KEY, name text UNIQUE); INSERT INTO "+schema+".accounts VALUES (1, 'not-an-integer')")
	if err != nil {
		t.Fatal(err)
	}
	var originalIndex uint32
	if err := conn.QueryRow(ctx, "SELECT to_regclass($1)::oid", schema+".accounts_name_key").Scan(&originalIndex); err != nil {
		t.Fatal(err)
	}

	// Exercise both full-DDL and diff generation against populated data.
	for _, fullDDL := range []bool{false, true} {
		t.Run(fmt.Sprintf("full_ddl=%v", fullDDL), func(t *testing.T) {
			model := liveModel(schema, map[string]string{"name": "integer"})
			unique := models.InitConstraint("accounts_name_key", models.UniqueConstraint)
			unique.Columns = []string{"name"}
			model.Schemas[0].Tables[0].Constraints[unique.Name] = unique
			report, err := runLiveWrite(t, connString, model, map[string]interface{}{"full_ddl": fullDDL})
			var failure *ExecutionFailure
			if !errors.As(err, &failure) || report.FailedStatements != 1 {
				t.Fatalf("expected failed conversion, got report=%+v err=%v", report, err)
			}
			var value, typ string
			if err := conn.QueryRow(ctx, "SELECT name, pg_typeof(name)::text FROM "+schema+".accounts WHERE id=1").Scan(&value, &typ); err != nil {
				t.Fatal(err)
			}
			if value != "not-an-integer" || typ != "text" {
				t.Fatalf("original column changed: value=%q type=%q", value, typ)
			}
			var columns int
			if err := conn.QueryRow(ctx, "SELECT count(*) FROM information_schema.columns WHERE table_schema=$1 AND table_name='accounts'", schema).Scan(&columns); err != nil || columns != 2 {
				t.Fatalf("unexpected replacement column: columns=%d err=%v", columns, err)
			}
			var index uint32
			if err := conn.QueryRow(ctx, "SELECT to_regclass($1)::oid", schema+".accounts_name_key").Scan(&index); err != nil || index != originalIndex {
				t.Fatalf("unique index changed: index=%d original=%d err=%v", index, originalIndex, err)
			}
			if _, err := conn.Exec(ctx, "INSERT INTO "+schema+".accounts VALUES (2, 'not-an-integer')"); err == nil {
				t.Fatal("unique constraint no longer enforced")
			}
		})
	}
}

func TestLive_AtomicExecutionRollback(t *testing.T) {
	connString := liveWriterConn(t)
	schema, conn := liveWriterSchema(t, connString)
	w := NewWriter(&writers.WriterOptions{AtomicExecution: true})
	w.executionReport = &ExecutionReport{}
	err := w.executeStatements([]string{
		"-- Schema: " + schema,
		"CREATE SCHEMA " + schema,
		"CREATE TABLE " + schema + ".created (id int)",
		"SELECT 1 / 0",
		"SELECT 1",
	}, connString)
	var failure *ExecutionFailure
	if !errors.As(err, &failure) {
		t.Fatalf("expected execution failure: %v", err)
	}
	report := failure.Report
	if report.Outcome != "rolled_back" || report.RolledBackStatements != 2 || report.ExecutedStatements != 0 || report.FailedStatements != 1 || report.SkippedStatements != 2 {
		t.Fatalf("rollback report: %+v", report)
	}
	if report.Schemas[0].Tables[0].Created {
		t.Fatal("rolled-back table must not be reported created")
	}
	if columnExists(t, conn, schema, "created", "id") {
		t.Fatal("table survived transaction rollback")
	}
}

func TestLive_TypePreflightPrecedesDropsEvenWithContinuation(t *testing.T) {
	connString := liveWriterConn(t)
	schema, conn := liveWriterSchema(t, connString)
	_, err := conn.Exec(context.Background(), "CREATE SCHEMA "+schema+"; CREATE TABLE "+schema+".accounts (id integer PRIMARY KEY, name text UNIQUE); INSERT INTO "+schema+".accounts VALUES (1, '123'), (2, 'bad')")
	if err != nil {
		t.Fatal(err)
	}
	// The desired model would drop the unique constraint. Incompatible data must
	// be caught first, including bad data after the first row.
	model := liveModel(schema, map[string]string{"name": "integer"})
	w := NewWriter(&writers.WriterOptions{ContinueOnError: true, Metadata: map[string]interface{}{"connection_string": connString}})
	err = w.WriteDatabase(model)
	var failure *ExecutionFailure
	if !errors.As(err, &failure) || failure.Report.FailedStatements != 1 || failure.Report.ExecutedStatements != 0 {
		t.Fatalf("preflight must stop mutations: %+v %v", w.executionReport, err)
	}
	var exists bool
	if err := conn.QueryRow(context.Background(), "SELECT to_regclass($1) IS NOT NULL", schema+".accounts_name_key").Scan(&exists); err != nil || !exists {
		t.Fatalf("constraint was dropped before preflight: %v", err)
	}
}

func TestLive_CustomTypeConversionAndPredicate(t *testing.T) {
	connString := liveWriterConn(t)
	for _, fullDDL := range []bool{false, true} {
		t.Run(fmt.Sprintf("full_ddl=%v", fullDDL), func(t *testing.T) {
			schema, conn := liveWriterSchema(t, connString)
			_, err := conn.Exec(context.Background(), "CREATE SCHEMA "+schema+"; CREATE TABLE "+schema+".accounts (id integer PRIMARY KEY, name text); INSERT INTO "+schema+".accounts VALUES (1,'42 units'),(2,'7 units')")
			if err != nil {
				t.Fatal(err)
			}
			model := liveModel(schema, map[string]string{"name": "integer"})
			col := model.Schemas[0].Tables[0].Columns["name"]
			col.Metadata["pgsql_using"] = "split_part(name, ' ', 1)::integer"
			col.Metadata["pgsql_preflight"] = "name ~ '^[0-9]+ units$' AND name NOT LIKE '7 %'"
			meta := map[string]interface{}{"full_ddl": fullDDL}
			if _, err := runLiveWrite(t, connString, model, meta); err == nil || !strings.Contains(err.Error(), "preflight rejected rows") {
				t.Fatalf("predicate must reject: %v", err)
			}
			col.Metadata["pgsql_preflight"] = "name ~ '^[0-9]+ units$'"
			if _, err := runLiveWrite(t, connString, model, meta); err != nil {
				t.Fatal(err)
			}
			var value int
			if err := conn.QueryRow(context.Background(), "SELECT sum(name) FROM "+schema+".accounts").Scan(&value); err != nil || value != 49 {
				t.Fatalf("custom conversion result %d: %v", value, err)
			}
		})
	}
}