// Package dbmigrate applies the embedded SQL migration directory to
// PostgreSQL on backend startup. It is deliberately tiny — no external
// tooling (golang-migrate/atlasexec would need binaries or revision tables
// this project's psql-applied history disagrees with), one transaction per
// migration file, and a baseline rule that guarantees zero schema changes on
// databases that predate this runner (they were built by db/schema.sql +
// seed.sql or the compose db-seed service).
//
// Tracking lives in schema_migrations(version bigint PRIMARY KEY,
// applied_at timestamptz NOT NULL DEFAULT now()). It intentionally shares
// nothing with the legacy `migrations` table created by 000001 nor with
// Atlas's atlas_schema_revisions. A database is baselined — every known
// revision recorded without executing anything — when it already has a users
// table but an empty tracking table, mirroring the docker-compose db-seed
// sentinel. Otherwise pending files run in ascending version order; each file
// commits together with its tracking row, so a failure leaves no partial
// state and startup aborts with an error naming the file.
package dbmigrate

import (
	"context"
	"fmt"
	"io/fs"
	"log/slog"
	"regexp"
	"sort"
	"strconv"

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

	"github.com/TaruLab/keluargamart-backend/migrations"
)

// Pool is the narrow surface of *pgxpool.Pool required by Run.
type Pool interface {
	Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error)
	QueryRow(ctx context.Context, sql string, args ...any) pgx.Row
	Begin(ctx context.Context) (pgx.Tx, error)
}

const createTrackingTable = `CREATE TABLE IF NOT EXISTS schema_migrations (
	version    BIGINT PRIMARY KEY,
	applied_at TIMESTAMPTZ NOT NULL DEFAULT now()
)`

// usersTableExists mirrors the docker-compose db-seed sentinel (users table
// in the session's default schema). current_schema() keeps non-public
// search_path deployments safe: they baseline instead of failing fast.
const usersTableExists = `SELECT EXISTS (
	SELECT 1 FROM information_schema.tables
	WHERE table_schema = current_schema() AND table_name = 'users'
)`

// appliedVersions returns all recorded revisions as one row so the runner
// only needs QueryRow. Empty table -> empty array via coalesce.
const appliedVersionsSQL = `SELECT coalesce(array_agg(version ORDER BY version), ARRAY[]::BIGINT[])
	FROM schema_migrations`

const insertVersionSQL = `INSERT INTO schema_migrations (version) VALUES ($1)`

// nameRe matches "NNNNNN_description.sql". Anything else in the directory
// (READMEs, atlas.sum, editor droppings) is ignored.
var nameRe = regexp.MustCompile(`^(\d{6})_.+\.sql$`)

type revision struct {
	Version int64
	Name    string
}

// Run applies every pending migration embedded from the migrations directory.
// It must be called after the pool connects and before the server starts
// serving traffic; a returned error means the schema could not be brought to
// the expected state and startup should abort.
func Run(ctx context.Context, pool Pool, log *slog.Logger) error {
	return runFS(ctx, pool, migrations.FS, log)
}

func runFS(ctx context.Context, pool Pool, fsys fs.FS, log *slog.Logger) error {
	revs, err := parseRevisions(fsys)
	if err != nil {
		return err
	}
	if len(revs) == 0 {
		log.Info("dbmigrate: no migration files found")
		return nil
	}

	if _, err := pool.Exec(ctx, createTrackingTable); err != nil {
		return fmt.Errorf("dbmigrate: ensure schema_migrations: %w", err)
	}

	var usersExists bool
	if err := pool.QueryRow(ctx, usersTableExists).Scan(&usersExists); err != nil {
		return fmt.Errorf("dbmigrate: detect existing schema: %w", err)
	}
	applied, err := queryAppliedVersions(ctx, pool)
	if err != nil {
		return fmt.Errorf("dbmigrate: read applied revisions: %w", err)
	}

	// Baseline rule: a database predating this runner was seeded/applied
	// manually (compose db-seed, psql). Record every revision WITHOUT
	// executing anything so adoption never mutates an existing deployment.
	if usersExists && len(applied) == 0 {
		if err := recordRevisions(ctx, pool, revs); err != nil {
			return fmt.Errorf("dbmigrate: baseline existing schema: %w", err)
		}
		log.Info(fmt.Sprintf("dbmigrate: baselined %d revisions on existing schema", len(revs)))
		return nil
	}

	pending := make([]revision, 0, len(revs))
	for _, r := range revs {
		if !applied[r.Version] {
			pending = append(pending, r)
		}
	}
	for _, r := range pending {
		if err := applyRevision(ctx, pool, fsys, r); err != nil {
			return err // already wraps the filename; fail fast, never serve
		}
		log.Info(fmt.Sprintf("dbmigrate: applied %s", r.Name))
	}
	switch len(pending) {
	case 0:
		log.Info("dbmigrate: schema up to date")
	default:
		log.Info(fmt.Sprintf("dbmigrate: %d migration(s) applied", len(pending)))
	}
	return nil
}

// parseRevisions reads fsys and returns the valid migration files sorted
// ascending by numeric version. Files not matching NNNNNN_*.sql are skipped;
// duplicate versions are ambiguous and rejected.
func parseRevisions(fsys fs.FS) ([]revision, error) {
	entries, err := fs.ReadDir(fsys, ".")
	if err != nil {
		return nil, fmt.Errorf("dbmigrate: read migration dir: %w", err)
	}
	revs := make([]revision, 0, len(entries))
	seen := map[int64]string{}
	for _, e := range entries {
		if e.IsDir() || !nameRe.MatchString(e.Name()) {
			continue
		}
		v, _ := strconv.ParseInt(e.Name()[:6], 10, 64)
		if prev, dup := seen[v]; dup {
			return nil, fmt.Errorf("dbmigrate: duplicate migration version %06d in %q and %q", v, prev, e.Name())
		}
		seen[v] = e.Name()
		revs = append(revs, revision{Version: v, Name: e.Name()})
	}
	sort.Slice(revs, func(i, j int) bool { return revs[i].Version < revs[j].Version })
	return revs, nil
}

func queryAppliedVersions(ctx context.Context, pool Pool) (map[int64]bool, error) {
	var versions []int64
	if err := pool.QueryRow(ctx, appliedVersionsSQL).Scan(&versions); err != nil {
		return nil, err
	}
	applied := make(map[int64]bool, len(versions))
	for _, v := range versions {
		applied[v] = true
	}
	return applied, nil
}

// recordRevisions marks every revision applied without executing any SQL file
// (baseline path). One transaction so tracking is all-or-nothing.
func recordRevisions(ctx context.Context, pool Pool, revs []revision) error {
	tx, err := pool.Begin(ctx)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback(ctx) }() // no-op after commit
	for _, r := range revs {
		if _, err := tx.Exec(ctx, insertVersionSQL, r.Version); err != nil {
			return fmt.Errorf("record %s: %w", r.Name, err)
		}
	}
	return tx.Commit(ctx)
}

// applyRevision executes one migration file inside its own transaction,
// recording the version in the same transaction. Any failure rolls back both
// the schema change and the tracking row and reports the offending file.
func applyRevision(ctx context.Context, pool Pool, fsys fs.FS, r revision) error {
	sqlBytes, err := fs.ReadFile(fsys, r.Name)
	if err != nil {
		return fmt.Errorf("dbmigrate: read %s: %w", r.Name, err)
	}
	tx, err := pool.Begin(ctx)
	if err != nil {
		return fmt.Errorf("dbmigrate: apply %s: %w", r.Name, err)
	}
	defer func() { _ = tx.Rollback(ctx) }() // no-op after commit
	if _, err := tx.Exec(ctx, string(sqlBytes)); err != nil {
		return fmt.Errorf("dbmigrate: apply %s: %w", r.Name, err)
	}
	if _, err := tx.Exec(ctx, insertVersionSQL, r.Version); err != nil {
		return fmt.Errorf("dbmigrate: apply %s: record version: %w", r.Name, err)
	}
	if err := tx.Commit(ctx); err != nil {
		return fmt.Errorf("dbmigrate: apply %s: commit: %w", r.Name, err)
	}
	return nil
}