package dbmigrate

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"strings"
	"testing"
	"testing/fstest"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgconn"
)

// --- fake executor ---------------------------------------------------------

// fakeRow answers a single-column QueryRow result.
type fakeRow struct {
	val any
	err error
}

func (r fakeRow) Scan(dest ...any) error {
	if r.err != nil {
		return r.err
	}
	if len(dest) != 1 {
		return fmt.Errorf("fakeRow: want 1 dest, got %d", len(dest))
	}
	switch d := dest[0].(type) {
	case *bool:
		v, ok := r.val.(bool)
		if !ok {
			return fmt.Errorf("fakeRow: bool scan got %T", r.val)
		}
		*d = v
	case *[]int64:
		v, ok := r.val.([]int64)
		if !ok {
			return fmt.Errorf("fakeRow: []int64 scan got %T", r.val)
		}
		*d = v
	default:
		return fmt.Errorf("fakeRow: unsupported dest %T", dest[0])
	}
	return nil
}

type execCall struct {
	sql  string
	args []any
}

// fakeTx implements pgx.Tx. Version inserts are buffered and only become
// visible to subsequent reads on Commit, mirroring real transaction semantics.
type fakeTx struct {
	db      *fakePool
	pending []int64
	done    bool
}

func (t *fakeTx) Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error) {
	t.db.execLog = append(t.db.execLog, execCall{sql: sql, args: args})
	if err := t.db.forcedError(sql); err != nil {
		return pgconn.CommandTag{}, err
	}
	if strings.Contains(sql, "INSERT INTO schema_migrations") {
		if len(args) != 1 {
			return pgconn.CommandTag{}, fmt.Errorf("fakeTx: insert wants 1 arg, got %d", len(args))
		}
		v, ok := args[0].(int64)
		if !ok {
			return pgconn.CommandTag{}, fmt.Errorf("fakeTx: version arg is %T, want int64", args[0])
		}
		t.pending = append(t.pending, v)
	}
	return pgconn.CommandTag{}, nil
}

func (t *fakeTx) Commit(ctx context.Context) error {
	t.done = true
	t.db.commits++
	t.db.recorded = append(t.db.recorded, t.pending...)
	t.pending = nil
	return nil
}

func (t *fakeTx) Rollback(ctx context.Context) error {
	t.done = true
	t.db.rollbacks++
	t.pending = nil
	return nil
}

func (t *fakeTx) Begin(ctx context.Context) (pgx.Tx, error) {
	return t.db.Begin(ctx)
}
func (t *fakeTx) CopyFrom(ctx context.Context, tableName pgx.Identifier, columnNames []string, rowSrc pgx.CopyFromSource) (int64, error) {
	return 0, errors.New("fakeTx: CopyFrom not implemented")
}
func (t *fakeTx) SendBatch(ctx context.Context, b *pgx.Batch) pgx.BatchResults {
	panic("fakeTx: SendBatch not implemented")
}
func (t *fakeTx) LargeObjects() pgx.LargeObjects { panic("fakeTx: LargeObjects not implemented") }
func (t *fakeTx) Prepare(ctx context.Context, name, sql string) (*pgconn.StatementDescription, error) {
	return nil, errors.New("fakeTx: Prepare not implemented")
}
func (t *fakeTx) Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error) {
	return nil, errors.New("fakeTx: Query not implemented")
}
func (t *fakeTx) QueryRow(ctx context.Context, sql string, args ...any) pgx.Row {
	return fakeRow{err: errors.New("fakeTx: QueryRow not implemented")}
}
func (t *fakeTx) Conn() *pgx.Conn { panic("fakeTx: Conn not implemented") }

// fakePool implements Pool. It answers the two runner probes (users-table
// existence, recorded versions), records every statement, and simulates
// transactional visibility of the tracking rows.
type fakePool struct {
	usersTable bool
	recorded   []int64 // committed schema_migrations rows
	failOn     string  // substring of SQL that makes Exec fail

	execLog   []execCall
	begins    int
	commits   int
	rollbacks int
	openTxs   []*fakeTx // begun but neither committed nor rolled back
}

func (f *fakePool) forcedError(sql string) error {
	if f.failOn != "" && strings.Contains(sql, f.failOn) {
		return fmt.Errorf("forced failure matching %q", f.failOn)
	}
	return nil
}

func (f *fakePool) Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error) {
	f.execLog = append(f.execLog, execCall{sql: sql, args: args})
	if err := f.forcedError(sql); err != nil {
		return pgconn.CommandTag{}, err
	}
	return pgconn.CommandTag{}, nil
}

func (f *fakePool) QueryRow(ctx context.Context, sql string, args ...any) pgx.Row {
	f.execLog = append(f.execLog, execCall{sql: sql})
	switch {
	case strings.Contains(sql, "information_schema.tables"):
		return fakeRow{val: f.usersTable}
	case strings.Contains(sql, "schema_migrations"):
		return fakeRow{val: append([]int64(nil), f.recorded...)}
	}
	return fakeRow{err: fmt.Errorf("fakePool: unexpected query %q", sql)}
}

func (f *fakePool) Begin(ctx context.Context) (pgx.Tx, error) {
	f.begins++
	tx := &fakeTx{db: f}
	f.openTxs = append(f.openTxs, tx)
	return tx, nil
}

// --- helpers ---------------------------------------------------------------

var testLogger = slog.New(slog.NewTextHandler(&strings.Builder{}, &slog.HandlerOptions{Level: slog.LevelDebug}))

func mapFS(files map[string]string) fstest.MapFS {
	m := fstest.MapFS{}
	for name, body := range files {
		m[name] = &fstest.MapFile{Data: []byte(body)}
	}
	return m
}

// fileExecs returns the SQL bodies executed that look like migration content
// (anything that is not runner bookkeeping).
func fileExecs(f *fakePool) []string {
	var out []string
	for _, c := range f.execLog {
		if strings.HasPrefix(c.sql, "CREATE TABLE IF NOT EXISTS schema_migrations") ||
			strings.Contains(c.sql, "INSERT INTO schema_migrations") ||
			strings.Contains(c.sql, "information_schema.tables") ||
			strings.Contains(c.sql, "array_agg") {
			continue
		}
		out = append(out, c.sql)
	}
	return out
}

func versionsOf(xs []int64) []int64 { return xs }

// --- tests -----------------------------------------------------------------

func TestRunFS(t *testing.T) {
	ctx := context.Background()
	tests := []struct {
		name         string
		files        map[string]string
		usersTable   bool
		recorded     []int64
		failOn       string
		wantErrSub   string
		wantRecorded []int64
		wantBodies   []string // expected migration-file executions, in order
	}{
		{
			name: "fresh empty database applies all in order",
			files: map[string]string{
				"000001_create_users.sql": "CREATE TABLE users_1();",
				"000002_add_cart.sql":     "CREATE TABLE carts_2();",
				"000010_later.sql":        "CREATE TABLE later_10();",
			},
			usersTable:   false,
			wantRecorded: []int64{1, 2, 10},
			wantBodies:   []string{"CREATE TABLE users_1();", "CREATE TABLE carts_2();", "CREATE TABLE later_10();"},
		},
		{
			name: "legacy database with users table baselines without executing",
			files: map[string]string{
				"000001_create_users.sql": "CREATE TABLE must_not_run_1();",
				"000002_add_cart.sql":     "CREATE TABLE must_not_run_2();",
			},
			usersTable:   true,
			wantRecorded: []int64{1, 2},
			wantBodies:   nil,
		},
		{
			name: "partially recorded database applies only pending",
			files: map[string]string{
				"000001_done.sql":    "CREATE TABLE already_done();",
				"000002_pending.sql": "CREATE TABLE pending_now();",
				"000003_pending.sql": "CREATE TABLE pending_too();",
			},
			usersTable:   true,
			recorded:     []int64{1},
			wantRecorded: []int64{1, 2, 3},
			wantBodies:   []string{"CREATE TABLE pending_now();", "CREATE TABLE pending_too();"},
		},
		{
			name: "fully migrated database is a no-op",
			files: map[string]string{
				"000001_a.sql": "CREATE TABLE never_1();",
				"000002_b.sql": "CREATE TABLE never_2();",
			},
			usersTable:   true,
			recorded:     []int64{1, 2},
			wantRecorded: []int64{1, 2},
			wantBodies:   nil,
		},
		{
			name: "malformed filenames are ignored",
			files: map[string]string{
				"README.md":          "not sql",
				"12_short.sql":       "missing digits;",
				"0000012_long.sql":   "seven digits;",
				"abc123_nope.sql":    "letters prefix;",
				"000003_nodot_sql":   "bad extension;",
				"000002_ok_name.sql": "CREATE TABLE ok_two();",
			},
			usersTable:   false,
			wantRecorded: []int64{2},
			wantBodies:   []string{"CREATE TABLE ok_two();"},
		},
		{
			name: "sql failure aborts startup and rolls back",
			files: map[string]string{
				"000001_ok.sql":    "CREATE TABLE fine();",
				"000002_boom.sql":  "ALTER TABLE boom_broken;",
				"000003_after.sql": "CREATE TABLE never_reached();",
			},
			// Revision 1 already recorded, so the runner takes the normal
			// apply path (not the baseline path) and hits the failure.
			usersTable:   true,
			recorded:     []int64{1},
			failOn:       "boom_broken",
			wantErrSub:   "000002_boom.sql",
			wantRecorded: []int64{1},                           // failed revision not recorded
			wantBodies:   []string{"ALTER TABLE boom_broken;"}, // only pending file ran
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			pool := &fakePool{usersTable: tt.usersTable, recorded: append([]int64(nil), tt.recorded...), failOn: tt.failOn}
			err := runFS(ctx, pool, mapFS(tt.files), testLogger)

			if tt.wantErrSub != "" {
				if err == nil || !strings.Contains(err.Error(), tt.wantErrSub) {
					t.Fatalf("runFS() error = %v, want containing %q", err, tt.wantErrSub)
				}
			} else if err != nil {
				t.Fatalf("runFS() unexpected error: %v", err)
			}

			if got := versionsOf(pool.recorded); !equalVersions(got, tt.wantRecorded) {
				t.Errorf("recorded versions = %v, want %v", got, tt.wantRecorded)
			}
			gotBodies := fileExecs(pool)
			if len(gotBodies) != len(tt.wantBodies) {
				t.Fatalf("migration-file executions = %q, want %q", gotBodies, tt.wantBodies)
			}
			for i, want := range tt.wantBodies {
				if gotBodies[i] != want {
					t.Errorf("execution[%d] = %q, want %q (order matters)", i, gotBodies[i], want)
				}
			}

			// No transaction may be left dangling: every Begin is paired with
			// a Commit or Rollback.
			open := 0
			for _, tx := range pool.openTxs {
				if !tx.done {
					open++
				}
			}
			if open != 0 {
				t.Errorf("%d transaction(s) left open (begins=%d commits=%d rollbacks=%d)", open, pool.begins, pool.commits, pool.rollbacks)
			}
		})
	}
}

func TestRunFS_SQLFailureRollsBackTrackingRow(t *testing.T) {
	ctx := context.Background()
	pool := &fakePool{failOn: "boom"}
	files := mapFS(map[string]string{
		"000001_ok.sql":   "CREATE TABLE fine();",
		"000002_boom.sql": "DROP TABLE boom;",
	})
	err := runFS(ctx, pool, files, testLogger)
	if err == nil || !strings.Contains(err.Error(), "000002_boom.sql") {
		t.Fatalf("error = %v, want failure naming 000002_boom.sql", err)
	}
	if pool.rollbacks == 0 {
		t.Error("expected at least one rollback after the failed migration")
	}
	for _, v := range pool.recorded {
		if v == 2 {
			t.Error("failed revision was recorded in schema_migrations")
		}
	}
}

func TestParseRevisions_DuplicateVersionRejected(t *testing.T) {
	_, err := parseRevisions(mapFS(map[string]string{
		"000001_a.sql": "x;",
		"000001_b.sql": "y;",
	}))
	if err == nil || !strings.Contains(err.Error(), "duplicate") {
		t.Fatalf("error = %v, want duplicate-version rejection", err)
	}
}

func equalVersions(a, b []int64) bool {
	if len(a) != len(b) {
		return false
	}
	for i := range a {
		if a[i] != b[i] {
			return false
		}
	}
	return true
}