package executor

import (
	"context"
	"io"
	"log/slog"
	"testing"
	"time"

	"tangled.org/core/api/tangled"
	"tangled.org/core/spindle/engine"
	"tangled.org/core/spindle/mill/proto"
	millv1 "tangled.org/core/spindle/mill/proto/gen"
	"tangled.org/core/spindle/models"
	"tangled.org/core/spindle/secrets"
)

type captureEncoder struct {
	messages chan *millproto.Message
}

func newCaptureEncoder() *captureEncoder {
	return &captureEncoder{messages: make(chan *millproto.Message, 4)}
}

func (e *captureEncoder) Encode(msg *millproto.Message) error {
	e.messages <- msg
	return nil
}

type fakeSlot struct{ released int }

func (s *fakeSlot) Release() { s.released++ }

// fakeEngine records which methods got called and never acquires a slot of its
// own; the wrapper must supply the slot.
type fakeEngine struct {
	setupCalled   bool
	runCalled     bool
	destroyCalled bool
	acquireCalled bool
}

func (e *fakeEngine) InitWorkflow(twf tangled.Pipeline_Workflow, tpl tangled.Pipeline) (*models.Workflow, error) {
	return &models.Workflow{Name: twf.Name}, nil
}
func (e *fakeEngine) SetupWorkflow(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, l models.WorkflowLogger) error {
	e.setupCalled = true
	return nil
}
func (e *fakeEngine) WorkflowTimeout() time.Duration { return 7 * time.Minute }
func (e *fakeEngine) DestroyWorkflow(ctx context.Context, wid models.WorkflowId) error {
	e.destroyCalled = true
	return nil
}
func (e *fakeEngine) RunStep(ctx context.Context, wid models.WorkflowId, w *models.Workflow, idx int, s []secrets.UnlockedSecret, l models.WorkflowLogger) error {
	e.runCalled = true
	return nil
}

func (e *fakeEngine) AcquireWorkflowSlot(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, mode engine.AcquireMode) (engine.WorkflowSlot, error) {
	e.acquireCalled = true
	return engine.NoopSlot{}, nil
}

func TestReservedEngineHandsBackHeldSlotOnce(t *testing.T) {
	inner := &fakeEngine{}
	slot := &fakeSlot{}
	re := newReservedEngine(inner, slot)

	got, err := re.(engine.WorkflowSlotter).AcquireWorkflowSlot(context.Background(), models.WorkflowId{}, nil, engine.Wait)
	if err != nil {
		t.Fatalf("AcquireWorkflowSlot() error = %v", err)
	}
	if got != engine.WorkflowSlot(slot) {
		t.Fatal("AcquireWorkflowSlot() did not return the held slot")
	}
	if inner.acquireCalled {
		t.Fatal("wrapper must not call the inner engine's AcquireWorkflowSlot")
	}

	if _, err := re.(engine.WorkflowSlotter).AcquireWorkflowSlot(context.Background(), models.WorkflowId{}, nil, engine.Wait); err == nil {
		t.Fatal("second AcquireWorkflowSlot() should error")
	}
}

func TestReservedEngineForwardsLifecycle(t *testing.T) {
	inner := &fakeEngine{}
	re := newReservedEngine(inner, &fakeSlot{})

	if d := re.WorkflowTimeout(); d != 7*time.Minute {
		t.Fatalf("WorkflowTimeout() = %v, want 7m", d)
	}
	if err := re.SetupWorkflow(context.Background(), models.WorkflowId{}, &models.Workflow{}, models.NullLogger{}); err != nil {
		t.Fatal(err)
	}
	if err := re.RunStep(context.Background(), models.WorkflowId{}, &models.Workflow{Steps: []models.Step{}}, 0, nil, models.NullLogger{}); err != nil {
		t.Fatal(err)
	}
	if err := re.DestroyWorkflow(context.Background(), models.WorkflowId{}); err != nil {
		t.Fatal(err)
	}
	if !inner.setupCalled || !inner.runCalled || !inner.destroyCalled {
		t.Fatalf("lifecycle not forwarded: %+v", inner)
	}
}

func TestHandleCommitIsIdempotent(t *testing.T) {
	enc := newCaptureEncoder()

	e := &Executor{
		l:      slog.New(slog.NewTextHandler(io.Discard, nil)),
		enc:    enc,
		active: map[string]*reservation{"lease-1": {leaseID: "lease-1", committed: true}},
	}

	e.handleCommit(context.Background(), &millv1.CommitLease{LeaseId: "lease-1"})
	msg := <-enc.messages
	if got := msg.GetCommitted().GetLeaseId(); got != "lease-1" {
		t.Fatalf("Committed lease = %q, want lease-1", got)
	}
}

func TestFinishJobReportsCancelledReservationAsCancelled(t *testing.T) {
	res := &reservation{
		leaseID:   "lease-1",
		wid:       models.WorkflowId{PipelineId: models.PipelineId{Knot: "k", Rkey: "r"}, Name: "build"},
		cancelled: true,
	}
	e := &Executor{
		relay:  newRelayLog(),
		active: map[string]*reservation{res.leaseID: res},
	}

	e.finishJob(res, &tangled.PipelineStatus{
		Pipeline: string(res.wid.PipelineId.AtUri()),
		Workflow: res.wid.Name,
		Status:   string(models.StatusKindFailed),
	})

	if len(e.relay.entries) != 1 {
		t.Fatalf("relay entries = %d, want 1", len(e.relay.entries))
	}
	got := e.relay.entries[0].GetAttemptResult().GetTerminalStatus()
	if got != string(models.StatusKindCancelled) {
		t.Fatalf("terminal status = %q, want cancelled", got)
	}
}
