package engine

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

	"tangled.org/core/spindle/config"
	"tangled.org/core/spindle/db"
	"tangled.org/core/spindle/models"
	"tangled.org/core/spindle/observability"
	"tangled.org/core/spindle/quota"
)

const (
	minutesOwner = "did:web:alice"
	minutesRepo  = "did:web:alice/repo"
)

func minutesQuotaManager(t *testing.T, database *db.DB, userLimit int64) (*quota.Manager, *db.QuotaStore) {
	t.Helper()
	qs := db.NewQuotaStore(database, quota.Defaults{
		quota.ScopeUser: quota.Resources{quota.ResourceCIMinutesMonthly: userLimit},
	})
	qm := quota.NewManager(qs, 50*time.Millisecond, nil)
	t.Cleanup(qm.Close)
	return qm, qs
}

func runMinutesWorkflow(t *testing.T, qm *quota.Manager, database *db.DB, ctx context.Context, eng models.Engine, name string) models.WorkflowId {
	t.Helper()
	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
	cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}}
	pipelineID := models.PipelineId("rkey")
	pipeline := &models.Pipeline{
		RepoDid: minutesRepo,
		Workflows: map[models.Engine][]models.Workflow{
			eng: {{
				Name:     name,
				Steps:    []models.Step{mockStep{name: "step1"}},
				OwnerDID: minutesOwner,
				RepoDID:  minutesRepo,
			}},
		},
	}
	StartWorkflows(logger, nil, cfg, qm, nil, database, nil, nil, nil, ctx, pipeline, pipelineID)
	return models.WorkflowId{PipelineId: pipelineID, Name: name}
}

func workflowError(t *testing.T, database *db.DB, wid models.WorkflowId) string {
	t.Helper()
	var msg string
	err := database.QueryRow(`
		select coalesce(error, '') from workflow_statuses
		where pipeline_id = ? and workflow = ? and status = 'failed'
	`, string(wid.PipelineId), wid.Name).Scan(&msg)
	if err != nil {
		t.Fatalf("no failed status for %s: %v", wid, err)
	}
	return msg
}

func TestStartWorkflows_DeniedWhenMonthlyMinutesAreUsedUp(t *testing.T) {
	database := newTestDB(t)
	qm, qs := minutesQuotaManager(t, database, 3)
	if err := qs.RecordMinutes(context.Background(), "earlier", quota.Identity{OwnerDID: minutesOwner, RepoDID: minutesRepo}, 3); err != nil {
		t.Fatal(err)
	}

	metrics := observability.NewMetrics()
	ctx := observability.WithMetrics(context.Background(), metrics)
	eng := &mockEngine{}
	wid := runMinutesWorkflow(t, qm, database, ctx, eng, "job1")

	eng.mu.Lock()
	ran := len(eng.setupCalls) + len(eng.runStepCalls)
	eng.mu.Unlock()
	if ran != 0 {
		t.Fatalf("denied workflow still reached the engine %d times", ran)
	}
	if st, err := database.GetStatus(wid); err != nil || st != models.StatusKindFailed {
		t.Fatalf("status = %q, %v, want failed", st, err)
	}
	msg := workflowError(t, database, wid)
	if !strings.HasPrefix(msg, "monthly CI minutes used up (owner): 3 of 3, resets ") || !strings.HasSuffix(msg, " 00:00 UTC") {
		t.Fatalf("status message = %q", msg)
	}

	if !quotaDeniedRecorded(t, metrics) {
		t.Fatal("denial was not attributed to quota_denied")
	}
}

func quotaDeniedRecorded(t *testing.T, metrics *observability.Metrics) bool {
	t.Helper()
	families, err := metrics.Registry().Gather()
	if err != nil {
		t.Fatal(err)
	}
	for _, family := range families {
		if family.GetName() != "spindle_workflow_terminations_total" {
			continue
		}
		for _, metric := range family.GetMetric() {
			labels := map[string]string{}
			for _, label := range metric.GetLabel() {
				labels[label.GetName()] = label.GetValue()
			}
			if labels["failure_class"] == string(FailureClassPolicy) && labels["reason"] == string(FailureReasonQuotaDenied) && metric.GetCounter().GetValue() == 1 {
				return true
			}
		}
	}
	return false
}

type remoteSlotEngine struct {
	*mockEngine
	released chan struct{}
}

func (*remoteSlotEngine) AuthorsRemoteStatus() {}

func (e *remoteSlotEngine) AcquireWorkflowSlot(context.Context, models.WorkflowId, *models.Workflow, AcquireMode) (WorkflowSlot, error) {
	return releaseFunc(func() { close(e.released) }), nil
}

func TestStartWorkflows_RemoteEngineDenialStillRecordsTheTerminalMetric(t *testing.T) {
	database := newTestDB(t)
	qm, qs := minutesQuotaManager(t, database, 3)
	if err := qs.RecordMinutes(context.Background(), "earlier", quota.Identity{OwnerDID: minutesOwner, RepoDID: minutesRepo}, 3); err != nil {
		t.Fatal(err)
	}

	metrics := observability.NewMetrics()
	ctx := observability.WithMetrics(context.Background(), metrics)
	eng := &remoteSlotEngine{mockEngine: &mockEngine{}, released: make(chan struct{})}
	runMinutesWorkflow(t, qm, database, ctx, eng, "job1")

	select {
	case <-eng.released:
	default:
		t.Fatal("the remote slot was not released after the denial")
	}
	if !quotaDeniedRecorded(t, metrics) {
		t.Fatal("a denial after the remote slot was taken recorded no quota_denied termination")
	}
}

func TestStartWorkflows_RecordsMinutesWhenAWorkflowFinishes(t *testing.T) {
	database := newTestDB(t)
	qm, qs := minutesQuotaManager(t, database, 10)

	eng := &mockEngine{}
	wid := runMinutesWorkflow(t, qm, database, context.Background(), eng, "job1")
	if st, err := database.GetStatus(wid); err != nil || st != models.StatusKindSuccess {
		t.Fatalf("status = %q, %v, want success", st, err)
	}

	failing := &mockEngine{runStepFunc: func(context.Context, models.WorkflowId, int, models.WorkflowLogger) error {
		return io.ErrUnexpectedEOF
	}}
	runMinutesWorkflow(t, qm, database, context.Background(), failing, "job2")

	usages, err := qs.ListUsage(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	got := map[quota.Scope]int64{}
	for _, u := range usages {
		if u.Resource == quota.ResourceCIMinutesMonthly {
			got[u.Scope] = u.Used
		}
	}
	// instant workflows still bill the one-minute floor
	if got[quota.ScopeUser] != 2 || got[quota.ScopeRepo] != 2 {
		t.Fatalf("minutes used = %v, want 2 on both the owner and the repo", got)
	}
}

type flakyMinutesStore struct {
	spyStore
	failures int
	recorded []string
}

func (s *flakyMinutesStore) CheckMinutes(context.Context, quota.Identity) error { return nil }

func (s *flakyMinutesStore) RecordMinutes(_ context.Context, usageID string, _ quota.Identity, _ int64) error {
	if s.failures > 0 {
		s.failures--
		return errors.New("database is locked")
	}
	s.recorded = append(s.recorded, usageID)
	return nil
}

func TestRecordMinutesRetriesAFailedRecord(t *testing.T) {
	id := quota.Identity{OwnerDID: minutesOwner, RepoDID: minutesRepo}

	store := &flakyMinutesStore{failures: recordMinutesAttempts - 1}
	qm := quota.NewManager(store, 0, nil)
	t.Cleanup(qm.Close)
	if err := recordMinutes(context.Background(), qm, "wf", id, 1); err != nil || len(store.recorded) != 1 {
		t.Fatalf("recordMinutes = %v, recorded %v, want it to succeed on the last attempt", err, store.recorded)
	}

	store = &flakyMinutesStore{failures: recordMinutesAttempts}
	qm = quota.NewManager(store, 0, nil)
	t.Cleanup(qm.Close)
	if err := recordMinutes(context.Background(), qm, "wf", id, 1); err == nil || len(store.recorded) != 0 {
		t.Fatalf("recordMinutes = %v, recorded %v, want the error once the attempts run out", err, store.recorded)
	}
}