package server

import (
	"encoding/json"
	"errors"
	"io"
	"net/http"
	"os"

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

// migrationRunRequest is the body for POST /migrations/{name}/run.
//
// All fields are optional; the body may be empty. A nil Retries means the
// field was omitted and DefaultRunRetries applies.
type migrationRunRequest struct {
	Retries *int `json:"retries"`
}

// maxPrepareBodyBytes caps the compiled-plan body accepted by POST
// /migrations.
//
// A var so tests can shrink it.
var maxPrepareBodyBytes = 16 << 20

// handleMigrationsPrepare stages + registers a migration from a compiled
// plan submitted as the raw request body.
//
// The migration name is the {name} path parameter. On success the migration
// is live immediately.
func handleMigrationsPrepare(r *Registry) http.HandlerFunc {
	return func(w http.ResponseWriter, req *http.Request) {
		name := req.PathValue("name")
		compiledPlan, err := io.ReadAll(io.LimitReader(req.Body, int64(maxPrepareBodyBytes)+1))
		if err != nil {
			http.Error(w, "read body: "+err.Error(), http.StatusBadRequest)
			return
		}
		if len(compiledPlan) > maxPrepareBodyBytes {
			http.Error(w, "compiled plan exceeds body limit", http.StatusRequestEntityTooLarge)
			return
		}
		m, err := r.Prepare(name, compiledPlan)
		if err != nil {
			respondPrepareErr(w, err)
			return
		}
		w.WriteHeader(http.StatusCreated)
		writeJSON(w, m.Status())
	}
}

// respondPrepareErr maps a Prepare error to its HTTP status and writes the
// response.
func respondPrepareErr(w http.ResponseWriter, err error) {
	switch {
	case errors.Is(err, errInvalidName):
		http.Error(w, err.Error(), http.StatusBadRequest)
	case errors.Is(err, errMigrationExists), errors.Is(err, errStagingLeftover):
		http.Error(w, err.Error(), http.StatusConflict)
	case errors.Is(err, errInvalidPlan):
		http.Error(w, err.Error(), http.StatusUnprocessableEntity)
	default:
		http.Error(w, err.Error(), http.StatusInternalServerError)
	}
}

// handleMigrationsList returns a snapshot of every migration's status, in
// registration order, wrapped with the total count.
func handleMigrationsList(r *Registry) http.HandlerFunc {
	return func(w http.ResponseWriter, _ *http.Request) {
		ms := r.List()
		out := make([]migrationStatus, 0, len(ms))
		for _, m := range ms {
			out = append(out, m.Status())
		}
		writeJSON(w, migrationsListResponse{Total: len(out), Migrations: out})
	}
}

// migrationsListResponse is the GET /migrations body.
type migrationsListResponse struct {
	Total      int               `json:"total"`
	Migrations []migrationStatus `json:"migrations"`
}

// handleMigrationGet returns one migration's status snapshot.
func handleMigrationGet(w http.ResponseWriter, _ *http.Request, m *Migration) {
	writeJSON(w, m.Status())
}

// handleMigrationRunJob starts or resumes a migrate-run job.
func handleMigrationRunJob(w http.ResponseWriter, r *http.Request, m *Migration) {
	var req migrationRunRequest
	// ContentLength != 0, not > 0: a chunked body (ContentLength -1) still
	// carries retries. Unknown fields are ignored, like everywhere else.
	if r.ContentLength != 0 {
		if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
			http.Error(w, "invalid JSON: "+err.Error(), http.StatusBadRequest)
			return
		}
	}
	retries := DefaultRunRetries
	if req.Retries != nil {
		if *req.Retries < 0 {
			http.Error(w, "retries must be >= 0", http.StatusBadRequest)
			return
		}
		retries = *req.Retries
	}

	id, err := m.StartLoopRun(retries)
	writeStartResponse(w, id, err)
}

// handleMigrationVmselectDeploy starts a vmselect-deploy job.
//
// The job applies the Deployment + Service for both targets and waits for
// each rollout; progress streams on /logs.
func handleMigrationVmselectDeploy(w http.ResponseWriter, _ *http.Request, m *Migration) {
	id, err := m.StartVmselectDeploy()
	writeStartResponse(w, id, err)
}

// handleMigrationVmselectUndeploy starts a vmselect-undeploy job.
//
// Absent objects count as success; the job waits until both are gone so a
// re-deploy with new storage ids never hits a Terminating name.
func handleMigrationVmselectUndeploy(w http.ResponseWriter, _ *http.Request, m *Migration) {
	id, err := m.StartVmselectUndeploy()
	writeStartResponse(w, id, err)
}

// handleMigrationSteps returns what the next run would do: per-tenant
// pending steps computed from the current on-disk status, in plan tenant
// order. Read-only — no job is dispatched, nothing is persisted. While a
// job runs the steps view reflects its live progress.
func handleMigrationSteps(w http.ResponseWriter, _ *http.Request, m *Migration) {
	loop, err := m.buildLoop()
	if respondStartErr(w, err) {
		return
	}
	pending := loop.PendingSteps()
	writeJSON(w, pendingStepsResponse{Tenants: pending})
}

// handleMigrationPlan serves the migration's runtime plan verbatim — the
// immutable input the loop executes (window, vmselect endpoints, tenants,
// per-step actions). A Migration exists only when its plan file does, so
// read failures are 500s, not 404s.
func handleMigrationPlan(w http.ResponseWriter, _ *http.Request, m *Migration) {
	data, err := os.ReadFile(m.loopPath())
	if err != nil {
		http.Error(w, "read plan: "+err.Error(), http.StatusInternalServerError)
		return
	}
	w.Header().Set("Content-Type", "application/json")
	_, _ = w.Write(data)
}

// handleMigrationProbe dispatches an export probe job.
//
// The request body is empty; defaults (match-all, 60s timeout, no rate-limit)
// are baked in. The job is single-flight per migration like the other Start*
// endpoints.
func handleMigrationProbe(w http.ResponseWriter, _ *http.Request, m *Migration) {
	id, err := m.StartExportProbe()
	writeStartResponse(w, id, err)
}

// startResponse is the 202 body for the Start* endpoints: the accepted job's
// ID, usable immediately for cancel and job polling.
type startResponse struct {
	Started bool   `json:"started"`
	JobId   string `json:"jobId"`
}

// pendingStepsResponse is the GET /migrations/{name}/steps body:
// per-tenant pending steps in plan tenant order.
type pendingStepsResponse struct {
	Tenants []schema.MigrationLoopPending `json:"tenants"`
}

// respondStartErr maps a StartXxx error to its HTTP status and writes the
// response.
//
// respondStartErr returns true if the caller should return (err was non-nil).
func respondStartErr(w http.ResponseWriter, err error) bool {
	if err == nil {
		return false
	}
	switch {
	case errors.Is(err, errAlreadyRunning):
		http.Error(w, err.Error(), http.StatusConflict)
	case errors.Is(err, errPlanMissingVmselect):
		http.Error(w, err.Error(), http.StatusServiceUnavailable)
	case errors.Is(err, errPlanMissingRuntime):
		http.Error(w, err.Error(), http.StatusServiceUnavailable)
	case errors.Is(err, errNoProvisioner):
		http.Error(w, err.Error(), http.StatusServiceUnavailable)
	default:
		http.Error(w, err.Error(), http.StatusInternalServerError)
	}
	return true
}

// writeStartResponse writes the Start* endpoints' 202 body, or the error
// mapped by respondStartErr.
func writeStartResponse(w http.ResponseWriter, id string, err error) {
	if respondStartErr(w, err) {
		return
	}
	w.WriteHeader(http.StatusAccepted)
	writeJSON(w, startResponse{Started: true, JobId: id})
}

// migrationCancelRequest is the body for POST /migrations/{name}/cancel.
type migrationCancelRequest struct {
	JobId string `json:"jobId"`
}

// handleMigrationCancelJob cancels the in-flight job.
//
// Returns 409 if no job is running or the body's jobId does not match the
// in-flight job, guarding a stale cancel against a new job.
func handleMigrationCancelJob(w http.ResponseWriter, r *http.Request, m *Migration) {
	var req migrationCancelRequest
	if r.ContentLength != 0 {
		if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
			http.Error(w, "invalid JSON: "+err.Error(), http.StatusBadRequest)
			return
		}
	}
	if err := m.Cancel(req.JobId); err != nil {
		http.Error(w, err.Error(), http.StatusConflict)
		return
	}
	w.WriteHeader(http.StatusAccepted)
	writeJSON(w, map[string]bool{"cancelled": true})
}

// handleMigrationSummary returns the loop-summary snapshot.
//
// The summary is computed by internal/migration.ReadLoopSummary over the migration
// directory. The default response is JSON; pass ?format=text for a
// human-readable progress table.
func handleMigrationSummary(w http.ResponseWriter, r *http.Request, m *Migration) {
	s, err := migration.ReadLoopSummary(m.loopPath())
	if err != nil {
		if errors.Is(err, os.ErrNotExist) {
			http.Error(w, "loop has not been run yet", http.StatusServiceUnavailable)
			return
		}
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	if r.URL.Query().Get("format") == "text" {
		w.Header().Set("Content-Type", "text/plain; charset=utf-8")
		migration.RenderText(w, s)
		return
	}
	writeJSON(w, s)
}

// handleMigrationTuneGet returns the effective knobs per section: plan
// values with tune.json overrides applied.
func handleMigrationTuneGet(w http.ResponseWriter, _ *http.Request, m *Migration) {
	view, err := m.Tune()
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	writeJSON(w, view)
}

// handleMigrationTuneSet applies a sectioned replacement.
//
// The body is a TuneSetRequest (strict decode): a present section replaces
// its override wholesale, an empty section removes it, an absent section is
// untouched. The response is the full effective view — vmctl for the next
// shard (the in-flight shard keeps its process-start flags), vmselect for
// the next deploy.
func handleMigrationTuneSet(w http.ResponseWriter, r *http.Request, m *Migration) {
	var req schema.TuneSetRequest
	if err := jsonio.DecodeStrict(r.Body, &req); err != nil {
		http.Error(w, err.Error(), http.StatusUnprocessableEntity)
		return
	}
	view, err := m.SetTune(req)
	if err != nil {
		if errors.Is(err, errInvalidTune) {
			http.Error(w, err.Error(), http.StatusUnprocessableEntity)
			return
		}
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	writeJSON(w, view)
}

// handleMigrationLogs tails the aggregated action-output stream for the
// migration's current (or most recent) job as chunked text/plain.
//
// The stream follows the whole job across tenants and stays open, emitting
// live bytes as the action produces them.
func handleMigrationLogs(w http.ResponseWriter, r *http.Request, m *Migration) {
	stream := m.logs.Get()
	if stream == nil {
		// No job has produced output yet. Short-circuit so the client
		// doesn't block on a stream that will never produce anything.
		w.Header().Set("Content-Type", "text/plain; charset=utf-8")
		w.WriteHeader(http.StatusOK)
		io.WriteString(w, "=== no job running ===\n")
		return
	}
	serveLogStream(w, r, stream)
}

// handleMigrationJobsList returns the full job history plus the in-flight
// record, in submission order, wrapped with the total count.
func handleMigrationJobsList(w http.ResponseWriter, _ *http.Request, m *Migration) {
	jobs := m.Jobs()
	writeJSON(w, jobsListResponse{Total: len(jobs), Jobs: jobs})
}

// jobsListResponse is the GET /migrations/{name}/jobs body.
type jobsListResponse struct {
	Total int         `json:"total"`
	Jobs  []jobRecord `json:"jobs"`
}

// handleMigrationJobGet returns one job's record by id.
func handleMigrationJobGet(w http.ResponseWriter, r *http.Request, m *Migration) {
	id := r.PathValue("id")
	r2, ok := m.GetJob(id)
	if !ok {
		http.Error(w, "job not found", http.StatusNotFound)
		return
	}
	writeJSON(w, r2)
}

// handleMigrationJobArtifact serves artifacts/<id>.json.
//
// Returns 404 when the job has no artifact (migrate-run) or the id is unknown.
// With ?format=text the probe artifact renders as a fixed-width table
// (the probe is the only artifact producer today).
func handleMigrationJobArtifact(w http.ResponseWriter, r *http.Request, m *Migration) {
	id := r.PathValue("id")
	if _, ok := m.GetJob(id); !ok {
		http.Error(w, "job not found", http.StatusNotFound)
		return
	}
	data, err := m.JobArtifact(id)
	if err != nil {
		if os.IsNotExist(err) {
			http.Error(w, "job has no artifact", http.StatusNotFound)
			return
		}
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	if r.URL.Query().Get("format") == "text" {
		var a probeArtifact
		if err := json.Unmarshal(data, &a); err != nil {
			http.Error(w, "decode artifact: "+err.Error(), http.StatusInternalServerError)
			return
		}
		if a.Kind != string(jobExportProbe) {
			http.Error(w, "text format supports export-probe artifacts only", http.StatusBadRequest)
			return
		}
		w.Header().Set("Content-Type", "text/plain; charset=utf-8")
		renderProbeText(w, &a)
		return
	}
	w.Header().Set("Content-Type", "application/json")
	_, _ = w.Write(data)
}