package server

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"os"
	"path/filepath"
	"sync"
	"time"

	"k8s.io/client-go/rest"

	"github.com/tidbcloud/observability/vm-migration/internal/jsonio"
	"github.com/tidbcloud/observability/vm-migration/internal/migration"
	"github.com/tidbcloud/observability/vm-migration/internal/schema"
	"github.com/tidbcloud/observability/vm-migration/internal/status"
)

// Registry owns the set of Migrations under a single migrations directory
// (typically <data-root>/migrations/).
//
// Registry is the top-level serve entry: HTTP handlers route by migration name
// to a Migration via Get and List. Migrations enter the Registry via
// LoadFromDisk at startup or Prepare (POST /migrations) at runtime. The
// Registry is a best-effort in-memory mirror of the
// migrations directory, not a CRUD store.
//
// Concurrency contract:
//   - LoadFromDisk, Get, and List are safe for concurrent HTTP use.
//   - Each Migration owns its own mutex; the Registry only guards the map.
type Registry struct {
	migrationsDir string
	root          context.Context

	mu            sync.Mutex
	migrations    map[string]*Migration
	order         []string // migration names in registration order
	loopOptions   []migration.Option
	retryInterval time.Duration
	provisioner   migration.VmselectProvisioner
}

// NewRegistry constructs an empty Registry.
//
// migrationsDir is the directory containing prepared migration subdirectories
// (one per migration). root is the serve process's context; per-migration jobs
// derive child contexts so Cancel (and root cancellation) propagate to
// in-flight actions. The Kubernetes provisioner is built from the in-cluster
// service account; outside a pod it is nil and deploy/undeploy return 503.
func NewRegistry(root context.Context, migrationsDir string) *Registry {
	return &Registry{
		migrationsDir: migrationsDir,
		root:          root,
		migrations:    map[string]*Migration{},
		provisioner:   newRegistryProvisioner(),
	}
}

// newRegistryProvisioner builds the in-cluster provisioner, nil when the
// Kubernetes credentials are unavailable.
func newRegistryProvisioner() migration.VmselectProvisioner {
	cfg, err := rest.InClusterConfig()
	if err != nil {
		slog.Info("kubernetes in-cluster config unavailable; deploy/undeploy will 503", "err", err)
		return nil
	}
	provisioner, err := migration.NewKubeProvisioner(cfg)
	if err != nil {
		slog.Warn("kubernetes provisioner unavailable; deploy/undeploy will 503", "err", err)
		return nil
	}
	return provisioner
}

// LoadFromDisk scans <cwd>/migrations/*/ and loads every directory that
// contains migration-loop.json as a Migration into the Registry.
//
// Directories without migration-loop.json are skipped with a warning (treated
// as "not prepared yet"). LoadFromDisk is best-effort: read/parse errors are
// logged and the offending directory is skipped so one bad migration doesn't
// block the rest. LoadFromDisk always returns nil; the caller (serve) doesn't
// have a useful response to a startup-scan error beyond surfacing it in logs.
// LoadFromDisk is idempotent: calling twice re-loads every dir, overwriting
// in-memory entries with freshly-parsed copies (disk is the source of truth).
func (r *Registry) LoadFromDisk() {
	entries, err := os.ReadDir(r.migrationsDir)
	if err != nil {
		if !errors.Is(err, os.ErrNotExist) {
			slog.Warn("registry load: read migrations dir", "err", err)
		}
		return
	}
	for _, e := range entries {
		if !e.IsDir() {
			continue
		}
		dir := filepath.Join(r.migrationsDir, e.Name())
		if _, err := os.Stat(filepath.Join(dir, loopFile)); err != nil {
			slog.Warn("registry load: missing migration-loop.json, skipping", "dir", dir)
			continue
		}
		m, err := newMigrationFromDir(r.root, dir)
		if err != nil {
			slog.Warn("registry load: migration construction failed", "dir", dir, "err", err)
			continue
		}
		r.register(m)
	}
}

// newMigrationFromDir constructs a *Migration from an existing on-disk directory.
//
// The directory must contain migration-loop.json. The migration name is taken
// from the directory base name.
func newMigrationFromDir(root context.Context, dir string) (*Migration, error) {
	loopPath := filepath.Join(dir, loopFile)
	var p schema.MigrationLoopPlan
	if err := jsonio.ReadFile(loopPath, &p); err != nil {
		return nil, fmt.Errorf("%s: %w", loopPath, err)
	}
	return &Migration{
		name:    filepath.Base(dir),
		dir:     dir,
		plan:    &p,
		rootCtx: root,
		logs:    newLogHub(),
	}, nil
}

// Get returns the named Migration, or nil/false if unknown.
func (r *Registry) Get(name string) (*Migration, bool) {
	r.mu.Lock()
	defer r.mu.Unlock()
	m, ok := r.migrations[name]
	return m, ok
}

// List returns a snapshot of all Migrations in registration order:
// startup disk-scan order first, then migrations added at runtime in
// arrival order.
func (r *Registry) List() []*Migration {
	r.mu.Lock()
	defer r.mu.Unlock()
	out := make([]*Migration, 0, len(r.order))
	for _, name := range r.order {
		out = append(out, r.migrations[name])
	}
	return out
}

// ResumeInterrupted re-dispatches jobs whose in-flight marker shows they were
// interrupted before finishing (serve crash, SIGKILL, process shutdown).
//
// A marker whose ID matches a finished record in jobs.jsonl is a stale
// leftover from a crash between the append and the marker removal and is only
// cleaned. Manually cancelled and normally finished jobs carry no marker and
// are left alone; failed jobs are not retried. An interrupted migrate-run is
// re-dispatched (resuming from on-disk progress) under a new job ID; an
// interrupted export-probe is not re-run (read-only) but finalized into
// jobs.jsonl as not-resumed. The interrupted attempt lands in history when
// its append succeeds: the probe path keeps the marker for retry on
// failure; the migrate-run path appends after dispatch (marker already
// superseded), so a failure there is only logged. Corrupt markers are kept
// on disk for inspection and logged. ResumeInterrupted is best-effort:
// dispatch errors are logged and the migration is skipped. Call after
// LoadFromDisk and before serving traffic; concurrent invocation with
// StartXxx methods is not expected.
func (r *Registry) ResumeInterrupted() []string {
	var resumed []string
	for _, m := range r.List() {
		if r.resumeInterrupted(m) {
			resumed = append(resumed, m.Name())
		}
	}
	return resumed
}

// resumeInterrupted applies the marker decision table to one migration.
func (r *Registry) resumeInterrupted(m *Migration) bool {
	resumeFail := func(err error) bool {
		slog.Warn("resume: failed; keeping marker", "migration", m.Name(), "err", err)
		m.NoteStartFailure(err)
		return false
	}
	marker, err := m.readInflightMarker()
	if marker == nil && err == nil {
		return false
	}
	if err != nil {
		return resumeFail(err)
	}
	finished, err := finishedJobIDs(m)
	if err != nil {
		// An unreadable jobs.jsonl must not read as "no history": that
		// would re-dispatch a finished job. Keep the marker and skip.
		return resumeFail(fmt.Errorf("read job history: %w", err))
	}
	if finished[marker.ID] {
		removeMarker(m, "stale (job already recorded)")
		return false
	}
	switch marker.Kind {
	case jobMigrateRun:
		// The retry budget is not persisted in the marker; a resumed job
		// runs with the default budget. Dispatch happens before the
		// interrupted record is appended: appending first would let a crash
		// in between mark the marker stale on the next start and silently
		// skip the resume. The trade is that an append failure after
		// dispatch is only logged — the marker is already superseded by the
		// new job.
		newID, err := m.StartLoopRun(DefaultRunRetries)
		if err != nil {
			return resumeFail(fmt.Errorf("dispatch migrate-run: %w", err))
		}
		if err := finalizeInterrupted(m, marker, "interrupted; resumed as "+newID); err != nil {
			slog.Error("jobs.jsonl append (interrupted record)", "migration", m.Name(), "job", marker.ID, "err", err)
		}
		slog.Info("resumed interrupted migrate-run", "migration", m.Name(), "job", marker.ID, "resumedAs", newID)
		return true
	case jobExportProbe:
		// Read-only probe: rerunning adds nothing. Finalize the attempt so
		// it doesn't vanish from history, then drop the marker.
		if err := finalizeInterrupted(m, marker, "interrupted; not resumed (read-only job)"); err != nil {
			return resumeFail(err)
		}
		removeMarker(m, "export-probe (read-only)")
		return false
	case jobVmselectDeploy, jobVmselectUndeploy:
		// Cluster mutation is operator-triggered only: never auto-resume.
		// Both jobs are idempotent, so re-POSTing retries safely.
		removeMarker(m, string(marker.Kind)+" (idempotent; re-POST to retry)")
		return false
	default:
		return resumeFail(fmt.Errorf("unknown job kind %q in in-flight marker", marker.Kind))
	}
}

// finalizeInterrupted appends marker's record to jobs.jsonl as a finished
// attempt: FinishedAt is the resume time, Result carries why the attempt
// ended. Keeps history complete when a marker is consumed without a re-run
// or is superseded by a resumed job's new record.
func finalizeInterrupted(m *Migration, marker *jobRecord, result string) error {
	r := *marker
	r.FinishedAt = status.FormatTime(status.UTCNow())
	r.Result = result
	return appendJobRecord(m.jobsJSONLPath(), &r)
}

// removeMarker deletes the in-flight marker after a no-resume decision.
func removeMarker(m *Migration, reason string) {
	if err := os.Remove(m.inflightMarkerPath()); err != nil && !os.IsNotExist(err) {
		slog.Warn("resume: remove in-flight marker", "migration", m.Name(), "err", err)
		return
	}
	slog.Info("removed in-flight marker", "migration", m.Name(), "reason", reason)
}

// finishedJobIDs returns the IDs of recorded jobs with a finish timestamp.
//
// Read errors are returned (not swallowed into an empty set) so callers can
// distinguish "no history" from "history unreadable".
func finishedJobIDs(m *Migration) (map[string]bool, error) {
	records, err := readJobRecords(m.jobsJSONLPath())
	if err != nil {
		return nil, err
	}
	out := make(map[string]bool, len(records))
	for _, r := range records {
		if r.FinishedAt != "" {
			out[r.ID] = true
		}
	}
	return out, nil
}