package mill

import (
	"context"
	"errors"
	"io"
	"log/slog"
	"net/http"
	"net/http/httptest"
	"path/filepath"
	"strings"
	"testing"
	"time"

	"tangled.org/core/notifier"
	"tangled.org/core/spindle/config"
	"tangled.org/core/spindle/db"
	"tangled.org/core/spindle/engine"
	"tangled.org/core/spindle/engines/dummy"
	"tangled.org/core/spindle/mill/executor"
	"tangled.org/core/spindle/models"
	"tangled.org/core/spindle/secrets"
	"tangled.org/core/workflow"
)

// the outcome is published before teardown. a held teardown keeps the lease open after it
type heldTeardownEngine struct {
	*dummy.DummyEngine
	stepErr  error
	teardown chan struct{}
}

func (e *heldTeardownEngine) RunStep(ctx context.Context, wid models.WorkflowId, w *models.Workflow, idx int, s []secrets.UnlockedSecret, wl models.WorkflowLogger) error {
	if e.stepErr != nil {
		return e.stepErr
	}
	return e.DummyEngine.RunStep(ctx, wid, w, idx, s, wl)
}

func (e *heldTeardownEngine) DestroyWorkflow(ctx context.Context, wid models.WorkflowId) error {
	<-e.teardown
	return nil
}

type lostExecutorWorld struct {
	t      *testing.T
	ctx    context.Context
	mill   *Mill
	bdb    *db.DB
	wsURL  string
	wid    models.WorkflowId
	result chan error
}

func newLostExecutorWorld(t *testing.T, grace time.Duration) *lostExecutorWorld {
	t.Helper()
	ctx, cancel := context.WithCancel(context.Background())
	t.Cleanup(cancel)
	l := slog.New(slog.NewTextHandler(io.Discard, nil))
	millDir := t.TempDir()
	bdb, err := db.Make(ctx, filepath.Join(millDir, "mill.db"))
	if err != nil {
		t.Fatalf("mill db: %v", err)
	}
	bn := notifier.New()
	m := New(l, Config{LogDir: millDir, ReconnectGrace: grace, BidTimeout: 2 * time.Second})
	m.Attach(bdb, &bn, testQuotaManager(t, bdb))
	registerTestExecutor(t, bdb, "exec-1", HashToken("test-token"), nil)
	srv := httptest.NewServer(http.HandlerFunc(m.HandleExecutorConn))
	t.Cleanup(srv.Close)
	return &lostExecutorWorld{
		t:     t,
		ctx:   ctx,
		mill:  m,
		bdb:   bdb,
		wsURL: "ws" + strings.TrimPrefix(srv.URL, "http"),
		wid:   models.WorkflowId{PipelineId: models.PipelineId("rkey1"), Name: "build"},
	}
}

func (w *lostExecutorWorld) startExecutor(dbPath string, eng models.Engine) context.CancelFunc {
	w.t.Helper()
	ctx, cancel := context.WithCancel(w.ctx)
	l := slog.New(slog.NewTextHandler(io.Discard, nil))
	edb, err := db.Make(ctx, dbPath)
	if err != nil {
		w.t.Fatalf("exec db: %v", err)
	}
	en := notifier.New()
	cfg := &config.Config{}
	cfg.Server.Dev = true
	cfg.Server.LogDir = filepath.Dir(dbPath)
	cfg.ArtifactStores.Disk.Dir = filepath.Join(filepath.Dir(dbPath), "artifacts")
	cfg.Server.Hostname = "exec-1"
	cfg.Mill.URL = w.wsURL
	cfg.Mill.Seats = 1
	cfg.Mill.SharedSecret = "test-token"
	exec, err := executor.New(cfg, map[string]models.Engine{"dummy": eng}, edb, &en, l, nil, nil)
	if err != nil {
		w.t.Fatalf("executor.New: %v", err)
	}
	go exec.Connect(ctx)
	w.waitFor("executor session", func() bool { return w.mill.sessionForNode("exec-1") != nil })
	return cancel
}

func (w *lostExecutorWorld) runUntilOutcomeThenKill(dbPath string, stepErr error) {
	w.t.Helper()
	eng := &heldTeardownEngine{DummyEngine: dummy.New(slog.New(slog.NewTextHandler(io.Discard, nil))), stepErr: stepErr, teardown: make(chan struct{})}
	kill := w.startExecutor(dbPath, eng)

	wf, err := NewEngine("dummy", w.mill).InitWorkflow(workflow.WorkflowSpec{
		Name: "build",
		Raw:  "steps:\n  - name: hello\n    command: echo hi\n",
	}, testPipeline())
	if err != nil {
		w.t.Fatalf("InitWorkflow: %v", err)
	}
	placeCtx, placeCancel := context.WithTimeout(w.ctx, 10*time.Second)
	defer placeCancel()
	slot, err := w.mill.place(placeCtx, "dummy", w.wid, wf)
	if err != nil {
		w.t.Fatalf("place: %v", err)
	}
	w.t.Cleanup(slot.Release)
	w.result = make(chan error, 1)
	go func() {
		ctx, cancel := context.WithTimeout(w.ctx, time.Minute)
		defer cancel()
		w.result <- w.mill.commitAndWait(ctx, wf, nil)
	}()

	outcome := models.StatusKindSuccess
	if stepErr != nil {
		outcome = models.StatusKindFailed
	}
	w.waitFor("recorded outcome", func() bool {
		st, err := w.bdb.GetStatus(w.wid)
		return err == nil && st == outcome
	})
	kill()
	w.waitFor("session loss", func() bool { return w.mill.sessionForNode("exec-1") == nil })
}

func (w *lostExecutorWorld) waitFor(what string, cond func() bool) {
	w.t.Helper()
	deadline := time.Now().Add(5 * time.Second)
	for time.Now().Before(deadline) {
		if cond() {
			return
		}
		time.Sleep(10 * time.Millisecond)
	}
	w.t.Fatalf("timed out waiting for %s", what)
}

func (w *lostExecutorWorld) runStepResult() error {
	w.t.Helper()
	select {
	case err := <-w.result:
		return err
	case <-time.After(5 * time.Second):
		w.t.Fatal("RunStep did not return after the lease was finished")
		return nil
	}
}

type statusRow struct {
	status string
	err    string
}

func (w *lostExecutorWorld) statusHistory() []statusRow {
	w.t.Helper()
	rows, err := w.bdb.Query(
		`select status, coalesce(error, '') from workflow_statuses where pipeline_id = ? and workflow = ? order by id`,
		string(w.wid.PipelineId), w.wid.Name,
	)
	if err != nil {
		w.t.Fatal(err)
	}
	defer rows.Close()
	var out []statusRow
	for rows.Next() {
		var r statusRow
		if err := rows.Scan(&r.status, &r.err); err != nil {
			w.t.Fatal(err)
		}
		out = append(out, r)
	}
	return out
}

func (w *lostExecutorWorld) assertSuccessKept() {
	w.t.Helper()
	if err := w.runStepResult(); err != nil {
		w.t.Fatalf("RunStep = %v, want success", err)
	}
	h := w.statusHistory()
	if last := h[len(h)-1]; last.status != string(models.StatusKindSuccess) {
		w.t.Fatalf("status history %+v, want it to end in success", h)
	}
}

func TestExecutorRestartAfterOutcomeKeepsSuccess(t *testing.T) {
	w := newLostExecutorWorld(t, time.Minute)
	dbPath := filepath.Join(t.TempDir(), "exec.db")
	w.runUntilOutcomeThenKill(dbPath, nil)
	w.startExecutor(dbPath, dummy.New(slog.New(slog.NewTextHandler(io.Discard, nil))))
	w.assertSuccessKept()
}

func TestWipedExecutorAfterOutcomeKeepsSuccess(t *testing.T) {
	w := newLostExecutorWorld(t, time.Minute)
	w.runUntilOutcomeThenKill(filepath.Join(t.TempDir(), "exec.db"), nil)
	w.startExecutor(filepath.Join(t.TempDir(), "exec.db"), dummy.New(slog.New(slog.NewTextHandler(io.Discard, nil))))
	w.assertSuccessKept()
}

func TestExecutorLostAfterOutcomeKeepsSuccess(t *testing.T) {
	w := newLostExecutorWorld(t, time.Second)
	w.runUntilOutcomeThenKill(filepath.Join(t.TempDir(), "exec.db"), nil)
	w.assertSuccessKept()
}

func TestExecutorRestartAfterFailedOutcomeKeepsItsError(t *testing.T) {
	w := newLostExecutorWorld(t, time.Minute)
	dbPath := filepath.Join(t.TempDir(), "exec.db")
	w.runUntilOutcomeThenKill(dbPath, errors.New("make: *** [test] Error 2"))
	recorded := w.statusHistory()
	w.startExecutor(dbPath, dummy.New(slog.New(slog.NewTextHandler(io.Discard, nil))))

	if err := w.runStepResult(); !errors.Is(err, engine.ErrWorkflowFailed) {
		t.Fatalf("RunStep = %v, want %v", err, engine.ErrWorkflowFailed)
	}
	h := w.statusHistory()
	last := h[len(h)-1]
	if len(h) != len(recorded) || last != recorded[len(recorded)-1] || !strings.Contains(last.err, "Error 2") {
		t.Fatalf("status history %+v, want it unchanged from %+v", h, recorded)
	}
}