package spindle

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

	"tangled.org/core/api/org_tangled"
	"tangled.org/core/gitutil"
	"tangled.org/core/knotfeed"
	"tangled.org/core/spindle/models"
	pipelinecodec "tangled.org/core/spindle/pipeline"
	"tangled.org/core/spindle/webhook"
	"tangled.org/core/spindle/xrpc"
	"tangled.org/core/workflow"
)

const (
	testHexOld = "1111111111111111111111111111111111111111"
	testHexNew = "2222222222222222222222222222222222222222"
)

func mustObjectID(hex string) knotfeed.ObjectID {
	sha, ok := knotfeed.ParseObjectID(hex)
	if !ok {
		panic("test fixture " + hex + " is not an object id")
	}
	return sha
}

func pushEvent(ref string, options ...string) *org_tangled.EventNotifyEvent_Input_Event {
	actor := testBob.String()
	return &org_tangled.EventNotifyEvent_Input_Event{EventPush: &org_tangled.EventPush{
		Repo:        testRepoDid.String(),
		Ref:         ref,
		OldSha:      testHexOld,
		NewSha:      testHexNew,
		Actor:       &actor,
		PushOptions: options,
	}}
}

type hookEnv struct {
	*ensureEnv
	pushes chan string
}

func newHookEnv(t *testing.T) *hookEnv {
	t.Helper()
	env := newEnsureEnv(t)
	env.s.wh = webhook.New(env.s.db, true, pauseGate(env.s.db, &env.s.cfg.Server))
	env.serve("repo")
	if env.ensure(t) == nil {
		t.Fatal("fixture repo not served")
	}

	pushes := make(chan string, 4)
	receiver := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		body, _ := io.ReadAll(r.Body)
		pushes <- string(body)
	}))
	t.Cleanup(receiver.Close)
	hook := &models.Webhook{RepoDid: testRepoDid, Url: receiver.URL, Active: true, Events: []string{"push"}}
	if created, err := env.s.db.AddWebhook(context.Background(), hook, 0, 0); err != nil || !created {
		t.Fatalf("add webhook: created=%v err=%v", created, err)
	}
	return &hookEnv{ensureEnv: env, pushes: pushes}
}

func (h *hookEnv) notifyErr(event *org_tangled.EventNotifyEvent_Input_Event) (models.PipelineId, error) {
	return h.s.NotifyEvent(context.Background(), h.k, testRepoDid, event)
}

func (h *hookEnv) notify(t *testing.T, event *org_tangled.EventNotifyEvent_Input_Event) string {
	t.Helper()
	pipeline, err := h.notifyErr(event)
	if err != nil {
		t.Fatalf("NotifyEvent: %v", err)
	}
	return string(pipeline)
}

func TestNotifyEventSkipCIFiresWebhookButSchedulesNothing(t *testing.T) {
	h := newHookEnv(t)
	if pipeline := h.notify(t, pushEvent("refs/heads/main", "skip-ci")); pipeline != "" {
		t.Fatalf("pipeline = %q, want none", pipeline)
	}
	select {
	case body := <-h.pushes:
		if !strings.Contains(body, testHexNew) {
			t.Errorf("webhook body %s lacks the new sha", body)
		}
	case <-time.After(5 * time.Second):
		t.Fatal("skip-ci push never reached the webhook")
	}
}

func TestNotifyEventIgnoresRefsThatRunNothing(t *testing.T) {
	h := newHookEnv(t)
	deleted := pushEvent("refs/heads/main")
	deleted.EventPush.NewSha = strings.Repeat("0", 40)

	for name, event := range map[string]*org_tangled.EventNotifyEvent_Input_Event{
		"ref outside heads and tags": pushEvent("refs/notes/wip"),
		"deleted ref":                deleted,
	} {
		if pipeline := h.notify(t, event); pipeline != "" {
			t.Errorf("%s: pipeline = %q, want none", name, pipeline)
		}
	}
	h.expectNoPush(t, "webhook fired for an ignored event")
}

func (h *hookEnv) expectPush(t *testing.T) {
	t.Helper()
	select {
	case <-h.pushes:
	case <-time.After(5 * time.Second):
		t.Fatal("push never reached the webhook")
	}
}

func (h *hookEnv) seedPushPipeline(t *testing.T, id, ref, oldSha, newSha string) {
	t.Helper()
	repoDid := testRepoDid.String()
	actor := testBob.String()
	raw := workflow.PipelineSpec{
		TriggerMetadata: &workflow.TriggerMetadata{
			Kind:  string(workflow.TriggerKindPush),
			Repo:  &workflow.TriggerRepo{Knot: h.k.Host, RepoDid: &repoDid, Did: testOwner.String()},
			Push:  &workflow.PushTriggerData{Ref: ref, OldSha: oldSha, NewSha: newSha},
			Actor: &actor,
		},
		Workflows: []*workflow.WorkflowSpec{{Name: "ci.yml"}},
	}
	record, err := pipelinecodec.FromTangled(models.PipelineId(id), time.Now(), raw)
	if err != nil {
		t.Fatal(err)
	}
	if err := h.s.db.CreatePipeline(record); err != nil {
		t.Fatal(err)
	}
	if _, err := h.s.db.RecordPush(testRepoDid, ref, oldSha, newSha, time.Now()); err != nil {
		t.Fatal(err)
	}
}

func (h *hookEnv) expectNoPush(t *testing.T, why string) {
	t.Helper()
	select {
	case body := <-h.pushes:
		t.Fatalf("%s: %s", why, body)
	case <-time.After(300 * time.Millisecond):
	}
}

type notifyResult struct {
	pipeline models.PipelineId
	err      error
}

func (h *hookEnv) notifyAsync(event *org_tangled.EventNotifyEvent_Input_Event) <-chan notifyResult {
	out := make(chan notifyResult, 1)
	go func() {
		pipeline, err := h.notifyErr(event)
		out <- notifyResult{pipeline, err}
	}()
	return out
}

func TestNotifyEventAnswersOnlyOnceThePushIsHandled(t *testing.T) {
	h := newHookEnv(t)

	unlock := h.s.lockRepo(testRepoDid).Unlock
	answer := h.notifyAsync(pushEvent("refs/heads/main", "skip-ci"))
	select {
	case res := <-answer:
		t.Fatalf("answered %+v while the push was still waiting on its work", res)
	case <-time.After(2500 * time.Millisecond):
	}
	h.expectNoPush(t, "work ran while the repo was locked")
	unlock()
	if res := <-answer; res.err != nil {
		t.Fatal(res.err)
	}
	h.expectPush(t)
}

func TestNotifyEventReplayOfAPipelinePushAnswersWithItAndRunsNothing(t *testing.T) {
	h := newHookEnv(t)
	h.seedPushPipeline(t, "first", "refs/heads/main", testHexOld, testHexNew)

	// skip-ci makes no new pipeline, the id can only come from the replay lookup
	for range 2 {
		got := h.notify(t, pushEvent("refs/heads/main", "skip-ci"))
		if got != "first" {
			t.Fatalf("replay answered %q, want the first delivery's pipeline id", got)
		}
		if _, err := h.s.db.GetPipeline(context.Background(), models.PipelineId(got)); err != nil {
			t.Fatalf("answer %q is not a stored pipeline id: %v", got, err)
		}
	}
	h.expectNoPush(t, "a replayed push fired a webhook")
}

func TestNotifyEventDedupeKeyIsRepoRefOldAndNew(t *testing.T) {
	h := newHookEnv(t)
	h.seedPushPipeline(t, "first", "refs/heads/main", testHexOld, testHexNew)

	other := strings.Repeat("3", 40)
	for name, mutate := range map[string]func(*org_tangled.EventPush){
		"another ref":     func(p *org_tangled.EventPush) { p.Ref = "refs/heads/dev" },
		"another old sha": func(p *org_tangled.EventPush) { p.OldSha = other },
		"another new sha": func(p *org_tangled.EventPush) { p.NewSha = other },
		// old and new are ordered, so A to B then B to A is two pushes
		"old and new swapped": func(p *org_tangled.EventPush) { p.OldSha, p.NewSha = p.NewSha, p.OldSha },
	} {
		event := pushEvent("refs/heads/main", "skip-ci")
		mutate(event.EventPush)
		if got := h.notify(t, event); got != "" {
			t.Errorf("%s: answered %q, want the push treated as new", name, got)
		}
		h.expectPush(t)
	}
}

func TestNotifyEventReplayWithoutAPipelineFiresTheWebhookOnce(t *testing.T) {
	h := newHookEnv(t)
	event := pushEvent("refs/heads/main", "skip-ci")
	h.notify(t, event)
	h.expectPush(t)
	h.notify(t, event)
	h.expectNoPush(t, "a replay of a push that made no pipeline fired the webhook again")
}

func TestNotifyEventDuplicateWhileRunningJoinsTheWork(t *testing.T) {
	h := newHookEnv(t)
	event := pushEvent("refs/heads/main", "skip-ci")

	unlock := h.s.lockRepo(testRepoDid).Unlock
	first := h.notifyAsync(event)
	time.Sleep(100 * time.Millisecond)
	second := h.notifyAsync(event)
	select {
	case res := <-second:
		t.Fatalf("duplicate answered %+v before the first finished", res)
	case <-time.After(300 * time.Millisecond):
	}
	unlock()
	a, b := <-first, <-second
	if a.err != nil || b.err != nil || a.pipeline != b.pipeline {
		t.Fatalf("answers = %+v and %+v, want the same one twice", a, b)
	}
	h.expectPush(t)
	h.expectNoPush(t, "the duplicate ran its own work")
}

func TestNotifyEventWorkSurvivesTheFirstCallerGivingUp(t *testing.T) {
	h := newHookEnv(t)
	event := pushEvent("refs/heads/main", "skip-ci")

	unlock := h.s.lockRepo(testRepoDid).Unlock
	ctx, hangUp := context.WithCancel(context.Background())
	first := make(chan error, 1)
	go func() {
		_, err := h.s.NotifyEvent(ctx, h.k, testRepoDid, event)
		first <- err
	}()
	time.Sleep(100 * time.Millisecond)
	second := h.notifyAsync(event)
	time.Sleep(100 * time.Millisecond)

	hangUp()
	if err := <-first; !errors.Is(err, context.Canceled) {
		t.Fatalf("first caller err = %v, want its own cancellation", err)
	}
	unlock()
	if res := <-second; res.err != nil {
		t.Fatalf("second caller inherited the first one's hangup: %v", res.err)
	}
	h.expectPush(t)
	h.expectNoPush(t, "the push ran twice")
}

func TestNotifyEventAnswersAnErrorWhenTheWorkFails(t *testing.T) {
	h := newHookEnv(t)
	h.knot.failing = true
	h.s.db.DeleteReposByDid(testRepoDid)
	event := pushEvent("refs/heads/main", "skip-ci")

	// an error here is what makes the knot retry
	if _, err := h.notifyErr(event); err == nil {
		t.Fatal("answered success for a push that was not handled")
	}
	h.expectNoPush(t, "a push that failed still fired")

	h.knot.failing = false
	h.notify(t, event)
	h.expectPush(t)
}

func TestNotifyEventRefusesPastThePool(t *testing.T) {
	h := newHookEnv(t)
	h.s.cfg.Server.NotifySlots = 3
	others := []string{"one.test", "two.test", "three.test"}
	for _, host := range others {
		if !h.s.notifySlots.acquire(host, h.s.notifyLimits()) {
			t.Fatal("slot refused below the pool")
		}
	}
	_, err := h.notifyErr(pushEvent("refs/heads/main", "skip-ci"))
	if !errors.Is(err, xrpc.ErrNotifyBusy) {
		t.Fatalf("err = %v, want ErrNotifyBusy", err)
	}
	for _, host := range others {
		h.s.notifySlots.release(host)
	}
	h.notify(t, pushEvent("refs/heads/main", "skip-ci"))
	h.expectPush(t)
}

func TestNotifyEventRejectsMalformedShas(t *testing.T) {
	h := newHookEnv(t)
	event := pushEvent("refs/heads/main")
	event.EventPush.NewSha = "not-a-sha"
	if _, err := h.notifyErr(event); err == nil {
		t.Fatal("malformed sha accepted")
	}
}

func TestNotifyEventIgnoresUnservedRepos(t *testing.T) {
	env := newEnsureEnv(t)
	env.s.wh = webhook.New(env.s.db, true)
	env.manifest("repo", "did:web:other.example.com")
	env.knot.declare(testOwner, "repo", testRepoDid)

	pipeline, err := env.s.NotifyEvent(context.Background(), env.k, testRepoDid, pushEvent("refs/heads/main"))
	if err != nil || pipeline != "" {
		t.Fatalf("got pipeline %q, err %v, want an empty answer", pipeline, err)
	}
	if rows := env.rows(t); len(rows) != 0 {
		t.Errorf("unserved repo was recorded: %+v", rows)
	}
}

func TestPushTriggerNamesTheActor(t *testing.T) {
	event := pushEvent("refs/heads/main").EventPush
	trigger := pushTrigger(nil, event, mustObjectID(testHexOld), mustObjectID(testHexNew))
	if trigger.Actor == nil || *trigger.Actor != testBob.String() {
		t.Errorf("actor = %v, want %s", trigger.Actor, testBob)
	}
	if trigger.Kind != string(workflow.TriggerKindPush) || trigger.Push.Ref != "refs/heads/main" || trigger.Push.OldSha != testHexOld || trigger.Push.NewSha != testHexNew {
		t.Errorf("trigger = %+v", trigger)
	}

	event.Actor = nil
	if trigger := pushTrigger(nil, event, mustObjectID(testHexOld), mustObjectID(testHexNew)); trigger.Actor != nil {
		t.Errorf("actor = %v, want nil when the knot sent none", *trigger.Actor)
	}
}

func repeatedPaths(n int, path string) []string {
	paths := make([]string, n)
	for i := range paths {
		paths[i] = path
	}
	return paths
}

func TestAdmitChangedFilesUnderBudget(t *testing.T) {
	if got, ok := admitChangedFiles([]string{"a.go", "b/c.go"}); !ok || !slices.Equal(got, []string{"a.go", "b/c.go"}) {
		t.Errorf("small listing = %v, %v, want both paths", got, ok)
	}
	if got, ok := admitChangedFiles(repeatedPaths(11000, "src/f.go")); ok || got != nil {
		t.Errorf("listing over the entry cap = %v, %v, want nothing admitted", got, ok)
	}
	wide := repeatedPaths(8192, "src/f.go")
	if got, ok := admitChangedFiles(wide); !ok || !slices.Equal(got, wide) {
		t.Errorf("listing at the entry cap = %d paths, %v, want all %d", len(got), ok, len(wide))
	}
	if got, ok := admitChangedFiles(nil); !ok || got == nil || len(got) != 0 {
		t.Errorf("an empty listing = %v, %v, want an admitted list that is empty and not nil", got, ok)
	}
	big := repeatedPaths(11000, strings.Repeat("x", 64))
	if got, ok := admitChangedFiles(big); ok || got != nil {
		t.Errorf("listing over the byte budget = %d paths, %v, want nothing admitted", len(got), ok)
	}
}

func TestChangedFilesUnderBudgetGuardsFirstSight(t *testing.T) {
	quiet := slog.New(slog.NewTextHandler(io.Discard, nil))
	if got, known := changedFilesUnderBudget(context.Background(), quiet, "/nonexistent/repo", knotfeed.ObjectID{}, mustObjectID(testHexNew)); known || got != nil {
		t.Errorf("first sight of a ref = %v, %v, want unknown without touching the repo", got, known)
	}
}

func TestChangedFilesUnderBudgetWithRealRepo(t *testing.T) {
	for _, format := range []string{"sha1", "sha256"} {
		t.Run(format, func(t *testing.T) { changedFilesInRealRepo(t, format) })
	}
}

func commitHead(t *testing.T, repoPath string) knotfeed.ObjectID {
	t.Helper()
	commitEverything(t, repoPath)
	return mustObjectID(gitInDir(t, repoPath, "rev-parse", "HEAD"))
}

func changedFilesInRealRepo(t *testing.T, format string) {
	repoPath := newSourceRepo(t, format)
	commit := func() knotfeed.ObjectID { return commitHead(t, repoPath) }

	writeTestFile(t, filepath.Join(repoPath, "README.md"), "one\n")
	shaFirst := commit()
	writeTestFile(t, filepath.Join(repoPath, ".tangled/workflows/ci.yaml"), "when:\n  event: push\n")
	writeTestFile(t, filepath.Join(repoPath, "src/app.go"), "package main\n")
	shaSecond := commit()
	gitInDir(t, repoPath, "mv", "src/app.go", "src/main.go")
	shaThird := commit()

	quiet := slog.New(slog.NewTextHandler(io.Discard, nil))

	synced := filepath.Join(t.TempDir(), "synced")
	if err := gitutil.SparseSync(context.Background(), repoPath, synced, shaSecond.String(), "/.tangled/workflows"); err != nil {
		t.Fatalf("sync: %v", err)
	}
	got, known := changedFilesUnderBudget(context.Background(), quiet, synced, shaFirst, shaSecond)
	if !known {
		t.Error("shallow checkout changed files are unknown")
	}
	slices.Sort(got)
	if want := []string{".tangled/workflows/ci.yaml", "src/app.go"}; !slices.Equal(got, want) {
		t.Errorf("shallow checkout changed files = %v, want %v", got, want)
	}

	for _, tc := range []struct {
		name   string
		oldSha knotfeed.ObjectID
		newSha knotfeed.ObjectID
		want   []string
	}{
		{"a push commit lists its added and modified files", shaFirst, shaSecond, []string{".tangled/workflows/ci.yaml", "src/app.go"}},
		{"a rename lists both the old and new path", shaSecond, shaThird, []string{"src/app.go", "src/main.go"}},
		{"a no-op revision range lists nothing", shaThird, shaThird, nil},
	} {
		t.Run(tc.name, func(t *testing.T) {
			got, known := changedFilesUnderBudget(context.Background(), quiet, repoPath, tc.oldSha, tc.newSha)
			slices.Sort(got)
			if !slices.Equal(got, tc.want) {
				t.Errorf("changed files = %v, want %v", got, tc.want)
			}
			// a nil list means unknown, and an empty change set still has to come back non-nil
			if !known || got == nil {
				t.Errorf("known = %v, list nil = %v, want a known list that is not nil", known, got == nil)
			}
		})
	}
}

func TestNotifyEventDoesNotLetGitNearAnInternalKnot(t *testing.T) {
	h := newHookEnv(t)
	h.s.cfg.Server.Dev = false
	before := h.knot.count()

	_, err := h.notifyErr(pushEvent("refs/heads/main"))
	if err == nil || err.Error() != "couldn't fetch the push from the knot" {
		t.Fatalf("err = %v, want the generic fetch error", err)
	}
	if got := h.knot.count(); got != before {
		t.Errorf("the knot saw %d requests, git should not have dialed a loopback host", got-before)
	}
}

func TestPathsGatedWorkflowsRunWhenTheChangedFilesCantBeTold(t *testing.T) {
	gated := workflow.Workflow{
		Name:   ".tangled/workflows/ci.yml",
		Engine: "nixery",
		When:   []workflow.Constraint{{Event: []string{"push"}, Branch: []string{"main"}, Paths: []string{"src/**"}}},
	}
	compile := func(changed []string) int {
		c := workflow.Compiler{
			Trigger: workflow.TriggerMetadata{
				Kind: string(workflow.TriggerKindPush),
				Push: &workflow.PushTriggerData{Ref: "refs/heads/main"},
			},
			ChangedFiles: changed,
		}
		return len(c.Compile([]workflow.Workflow{gated}).Workflows)
	}

	small, ok := admitChangedFiles([]string{"docs/guide.md"})
	if !ok || compile(small) != 0 {
		t.Errorf("a push that only touched docs ran the src workflow (known %v)", ok)
	}

	over := repeatedPaths(changedFilesMaxCount+1, "docs/f.md")
	tooMany, ok := admitChangedFiles(over)
	if ok || compile(tooMany) != 1 {
		t.Errorf("a push with %d changed files didn't run the paths-gated workflow (known %v)", len(over), ok)
	}
}

func TestOverBudgetPushIsLoggedWithItsFileCount(t *testing.T) {
	repoPath := newSourceRepo(t, "sha1")
	writeTestFile(t, filepath.Join(repoPath, "README.md"), "one\n")
	before := commitHead(t, repoPath)
	for i := range changedFilesMaxCount + 1 {
		writeTestFile(t, filepath.Join(repoPath, "gen", fmt.Sprintf("f%d.txt", i)), "x\n")
	}
	after := commitHead(t, repoPath)

	var logged strings.Builder
	l := slog.New(slog.NewTextHandler(&logged, nil))
	got, known := changedFilesUnderBudget(context.Background(), l, repoPath, before, after)
	if known || got != nil {
		t.Fatalf("changed files = %d, known %v, want unknown", len(got), known)
	}
	for _, want := range []string{before.String(), after.String(), fmt.Sprintf("files=%d", changedFilesMaxCount+1)} {
		if !strings.Contains(logged.String(), want) {
			t.Errorf("log %q lacks %q", logged.String(), want)
		}
	}
}