package spindle

import (
	"context"
	"errors"
	"slices"
	"strings"
	"testing"
	"time"

	"github.com/bluesky-social/indigo/atproto/syntax"

	"tangled.org/core/api/org_tangled"
	"tangled.org/core/spindle/db"
	"tangled.org/core/spindle/models"
	"tangled.org/core/spindle/xrpc"
	"tangled.org/core/workflow"
)

const (
	prWorkflow   = "engine: dummy\nwhen:\n  - event: pull_request\n    branch: main\nsteps:\n  - name: hi\n    command: echo hi\n"
	doneWorkflow = "engine: dummy\nwhen:\n  - event: pull_request\n    branch: main\n    types: [closed, merged]\nsteps:\n  - name: bye\n    command: echo bye\n"
	otherPull    = "3kpull0000002"
)

type prPipelineEnv struct {
	*pullEnv
	repo *fakeRepo
}

func newPRPipelineEnv(t *testing.T) *prPipelineEnv {
	t.Helper()
	w := newWorld(t)
	repo := w.serveGitRepo(testRepoDid)
	n := w.boot()
	w.seed(n, testBob)
	e := &pullEnv{world: w, n: n, sink: newSink(t)}
	w.subscribe(n, e.sink, pullEvents...)
	return &prPipelineEnv{pullEnv: e, repo: repo}
}

func (e *prPipelineEnv) head(t *testing.T) string {
	t.Helper()
	return e.repo.commitWorkflows(t, map[string]string{"pr.yaml": prWorkflow, "done.yaml": doneWorkflow})
}

func (e *prPipelineEnv) ticketAs(rkey, state string, editor syntax.DID, op string) {
	e.putTicket(testRepoDid, rkey, state, editor, true)
	e.n.applied(testRepoDid, e.emitTicket(testRepoDid, rkey, op))
}

func (e *prPipelineEnv) patchAs(rkey, ticket, head string, editor syntax.DID) {
	e.putPatch(testRepoDid, rkey, ticket, head, editor)
	e.n.applied(testRepoDid, e.emitPatch(testRepoDid, rkey, opCreate))
}

func prPipelines(t *testing.T, n *node) []*models.PipelineRecord {
	t.Helper()
	records, _, _, err := n.s.db.QueryPipelines(context.Background(), testRepoDid.String(), nil, "", []string{"pull_request"}, 100)
	if err != nil {
		t.Fatal(err)
	}
	slices.SortFunc(records, func(a, b *models.PipelineRecord) int { return strings.Compare(string(a.ID), string(b.ID)) })
	return records
}

// like the sink, waits for the count and then sees that nothing else follows
func expectPRPipelines(t *testing.T, n *node, want int) []*models.PipelineRecord {
	t.Helper()
	if want > 0 {
		eventually(t, 30*time.Second, "the pull request pipelines", func() bool { return len(prPipelines(t, n)) >= want })
	}
	time.Sleep(settleTime)
	got := prPipelines(t, n)
	if len(got) != want {
		t.Fatalf("got %d pull request pipelines, want %d: %s", len(got), want, describePipelines(got))
	}
	return got
}

func describePipelines(records []*models.PipelineRecord) string {
	var out []string
	for _, r := range records {
		pr := r.Trigger.PullRequest
		action := ""
		if pr.Action != nil {
			action = *pr.Action
		}
		out = append(out, action+"@"+r.Commit)
	}
	return "[" + strings.Join(out, " ") + "]"
}

func wantPRPipeline(t *testing.T, r *models.PipelineRecord, action, sha, ticket string, workflows ...string) {
	t.Helper()
	pr := r.Trigger.PullRequest
	if pr == nil || pr.Action == nil || *pr.Action != action {
		t.Fatalf("pipeline %s: trigger %+v, want action %s", r.ID, pr, action)
	}
	if r.Commit != sha || pr.SourceCommit != sha {
		t.Errorf("%s: commit %s source %s, want %s", action, r.Commit, pr.SourceCommit, sha)
	}
	if pr.TargetBranch != "main" {
		t.Errorf("%s: target branch %q, want main", action, pr.TargetBranch)
	}
	if want := "at://" + testRepoDid.String() + "/" + org_tangled.TrackTicketNSID + "/" + ticket; pr.Pull == nil || *pr.Pull != want {
		t.Errorf("%s: pull %v, want %s", action, pr.Pull, want)
	}
	var names []string
	for _, w := range r.Workflows {
		names = append(names, w.Name)
	}
	if !slices.Equal(names, workflows) {
		t.Errorf("%s: workflows %v, want %v", action, names, workflows)
	}
}

func TestPullRequestActionsTriggerPipelines(t *testing.T) {
	e := newPRPipelineEnv(t)
	sha1, sha2 := e.head(t), e.head(t)

	e.ticketAs(pullRkey, "open", testBob, opCreate)
	e.patchAs(patchOne, pullRkey, sha1, testBob)
	got := expectPRPipelines(t, e.n, 1)
	wantPRPipeline(t, got[0], "opened", sha1, pullRkey, "pr.yaml")

	e.patchAs(patchTwo, pullRkey, sha2, testBob)
	got = expectPRPipelines(t, e.n, 2)
	wantPRPipeline(t, got[1], "synchronize", sha2, pullRkey, "pr.yaml")

	// only the types: [closed, merged] workflow runs on a close or a merge
	e.ticketAs(pullRkey, "closed", testOwner, "update")
	got = expectPRPipelines(t, e.n, 3)
	wantPRPipeline(t, got[2], "closed", sha2, pullRkey, "done.yaml")

	e.ticketAs(pullRkey, "open", testBob, "update")
	got = expectPRPipelines(t, e.n, 4)
	wantPRPipeline(t, got[3], "reopened", sha2, pullRkey, "pr.yaml")

	e.ticketAs(pullRkey, "completed", testOwner, "update")
	got = expectPRPipelines(t, e.n, 5)
	wantPRPipeline(t, got[4], "merged", sha2, pullRkey, "done.yaml")
	e.sink.expect(t, 5)
}

// a blocked pipeline has blocked workflows and no job, a running one has neither
func wantBlocked(t *testing.T, n *node, r *models.PipelineRecord, blocked bool) {
	t.Helper()
	for _, w := range r.Workflows {
		if (w.Status == string(models.StatusKindBlocked)) != blocked {
			t.Errorf("pipeline %s workflow %s is %s, want blocked %v", r.ID, w.Name, w.Status, blocked)
		}
	}
	if !blocked {
		return
	}
	var jobs int
	if err := n.s.db.QueryRow(`select count(*) from jobs where pipeline_id = ?`, r.ID).Scan(&jobs); err != nil {
		t.Fatal(err)
	}
	if jobs != 0 {
		t.Errorf("blocked pipeline %s has %d jobs, want none", r.ID, jobs)
	}
}

func TestPullRequestPipelineNeedsPushAccess(t *testing.T) {
	e := newPRPipelineEnv(t)
	sha := e.head(t)

	// dave isn't a collaborator, so his pull's pipelines wait for approval
	e.ticketAs(otherPull, "open", testDave, opCreate)
	e.patchAs("bafydavepatch", otherPull, sha, testDave)
	e.ticketAs(otherPull, "closed", testDave, "update")
	e.ticketAs(otherPull, "open", testDave, "update")
	e.sink.expect(t, 3)
	for _, r := range expectPRPipelines(t, e.n, 3) {
		wantBlocked(t, e.n, r, true)
	}

	e.ticketAs(pullRkey, "open", testBob, opCreate)
	e.patchAs(patchOne, pullRkey, sha, testBob)
	got := expectPRPipelines(t, e.n, 4)
	wantBlocked(t, e.n, got[3], false)

	// bob may run pipelines, but carol acting on his pull may not
	e.ticketAs(pullRkey, "closed", testCarol, "update")
	e.patchAs(patchTwo, pullRkey, e.head(t), testCarol)
	e.sink.expect(t, 6)
	got = expectPRPipelines(t, e.n, 6)
	wantBlocked(t, e.n, got[4], true)
	wantBlocked(t, e.n, got[5], true)

	e.ticketAs(pullRkey, "open", testOwner, "update")
	got = expectPRPipelines(t, e.n, 7)
	if a := got[6].Trigger.PullRequest.Action; a == nil || *a != "reopened" {
		t.Errorf("the owner's reopen ran %v", a)
	}
	wantBlocked(t, e.n, got[6], false)
}

// every round's pipelines link back to the patch they ran, even a close that came later
func TestPullRequestPipelineKeepsItsPatch(t *testing.T) {
	e := newPRPipelineEnv(t)
	e.ticketAs(pullRkey, "open", testBob, opCreate)
	e.patchAs(patchOne, pullRkey, e.head(t), testBob)
	expectPRPipelines(t, e.n, 1)
	e.patchAs(patchTwo, pullRkey, e.head(t), testBob)
	expectPRPipelines(t, e.n, 2)
	e.ticketAs(pullRkey, "closed", testOwner, "update")
	got := expectPRPipelines(t, e.n, 3)
	for i, rkey := range []string{patchOne, patchTwo, patchTwo} {
		var patch *string
		if err := e.n.s.db.QueryRow(`select patch_uri from pipelines where pipeline_id = ?`, got[i].ID).Scan(&patch); err != nil {
			t.Fatal(err)
		}
		if want := "at://" + testRepoDid.String() + "/" + org_tangled.ReviewPatchNSID + "/" + rkey; patch == nil || *patch != want {
			t.Errorf("pipeline %d patch_uri %v, want %s", i, patch, want)
		}
	}
}

func TestApprovedPullRequestPipelineIsQueued(t *testing.T) {
	e := newPRPipelineEnv(t)
	e.ticketAs(otherPull, "open", testDave, opCreate)
	e.patchAs("bafydavepatch", otherPull, e.head(t), testDave)
	got := expectPRPipelines(t, e.n, 1)
	wantBlocked(t, e.n, got[0], true)

	if err := e.n.s.ApprovePipeline(context.Background(), testRepoDid, got[0].ID); err != nil {
		t.Fatal(err)
	}
	// the test world runs no job workers, so the job waits in the queue
	wid := models.WorkflowId{PipelineId: got[0].ID, Name: "pr.yaml"}
	if st, err := e.n.s.db.GetStatus(wid); err != nil || st != models.StatusKindPending {
		t.Fatalf("approved workflow is %s (%v), want pending", st, err)
	}
	var jobs int
	if err := e.n.s.db.QueryRow(`select count(*) from jobs where pipeline_id = ?`, got[0].ID).Scan(&jobs); err != nil || jobs != 1 {
		t.Fatalf("approved pipeline has %d jobs (%v), want 1", jobs, err)
	}
	if err := e.n.s.ApprovePipeline(context.Background(), testRepoDid, got[0].ID); !errors.Is(err, db.ErrNotBlocked) {
		t.Fatalf("second approve = %v, want ErrNotBlocked", err)
	}
}

func TestRetryRerunsTheStoredSpec(t *testing.T) {
	e := newPRPipelineEnv(t)
	ctx := context.Background()
	sha := e.head(t)
	e.ticketAs(pullRkey, "open", testBob, opCreate)
	e.patchAs(patchOne, pullRkey, sha, testBob)
	got := expectPRPipelines(t, e.n, 1)

	retryId, err := e.n.s.RetryPipeline(ctx, testOwner, testRepoDid, got[0].ID, nil)
	if err != nil {
		t.Fatal(err)
	}
	got = expectPRPipelines(t, e.n, 2)
	retried := got[1]
	if retried.ID != retryId || retried.ID == got[0].ID {
		t.Fatalf("retry made %s, want a new pipeline %s", retried.ID, retryId)
	}
	wantPRPipeline(t, retried, "opened", sha, pullRkey, "pr.yaml")
	wantBlocked(t, e.n, retried, false)
	stored, err := e.n.s.db.GetStoredPipeline(retryId)
	if err != nil {
		t.Fatal(err)
	}
	if a := stored.Spec.TriggerMetadata.Actor; a == nil || *a != testOwner.String() {
		t.Errorf("retry ran as %v, want the retrier %s", a, testOwner)
	}
	var patches int
	if err := e.n.s.db.QueryRow(`select count(distinct patch_uri) from pipelines where pipeline_id in (?, ?)`, got[0].ID, retryId).Scan(&patches); err != nil || patches != 1 {
		t.Errorf("the retry links %d patches (%v), want the original's", patches, err)
	}

	if _, err := e.n.s.RetryPipeline(ctx, testOwner, testRepoDid, got[0].ID, []string{"nope.yaml"}); !errors.Is(err, xrpc.ErrNoMatchingWorkflows) {
		t.Errorf("retrying only an unknown workflow = %v, want ErrNoMatchingWorkflows", err)
	}

	// dave's pipeline has to be approved, a retry would skip that
	e.ticketAs(otherPull, "open", testDave, opCreate)
	e.patchAs("bafydavepatch", otherPull, sha, testDave)
	got = expectPRPipelines(t, e.n, 3)
	if _, err := e.n.s.RetryPipeline(ctx, testOwner, testRepoDid, got[2].ID, nil); !errors.Is(err, xrpc.ErrNotRetryable) {
		t.Fatalf("retrying a blocked pipeline = %v, want ErrNotRetryable", err)
	}
	if err := e.n.s.ApprovePipeline(ctx, testRepoDid, got[2].ID); err != nil {
		t.Fatal(err)
	}
	if _, err := e.n.s.RetryPipeline(ctx, testOwner, testRepoDid, got[2].ID, nil); err != nil {
		t.Fatalf("retrying an approved pipeline: %v", err)
	}

	// pipelines from before specs were stored can't be rerun
	if err := e.n.s.db.CreatePipeline(&models.PipelineRecord{ID: "legacy", RepoDID: testRepoDid.String()}); err != nil {
		t.Fatal(err)
	}
	if _, err := e.n.s.RetryPipeline(ctx, testOwner, testRepoDid, "legacy", nil); !errors.Is(err, xrpc.ErrNotRetryable) {
		t.Fatalf("retrying a pipeline without a spec = %v, want ErrNotRetryable", err)
	}
}

func TestPullRequestPipelineSkipsBannedRepo(t *testing.T) {
	e := newPRPipelineEnv(t)
	sha := e.head(t)
	if err := e.n.s.db.PutBan(db.BanEntry{SubjectDid: testRepoDid}); err != nil {
		t.Fatal(err)
	}
	e.ticketAs(pullRkey, "open", testBob, opCreate)
	e.patchAs(patchOne, pullRkey, sha, testBob)
	e.sink.expect(t, 1)
	expectPRPipelines(t, e.n, 0)
}

// the retry loop only looks every tick, so move the clock past the backoff and wait for it
func (e *prPipelineEnv) skipBackoff() {
	e.skew.Add(int64(2 * time.Hour))
}

func TestPullRequestEventSurvivesAKnotOutage(t *testing.T) {
	e := newPRPipelineEnv(t)
	sha := e.head(t)
	e.ticketAs(pullRkey, "open", testBob, opCreate)

	e.setKnotDown(true)
	e.patchAs(patchOne, pullRkey, sha, testBob)
	e.sink.expect(t, 0)
	expectPRPipelines(t, e.n, 0)

	e.setKnotDown(false)
	e.skipBackoff()
	got := expectPRPipelines(t, e.n, 1)
	wantPRPipeline(t, got[0], "opened", sha, pullRkey, "pr.yaml")
	e.sink.expect(t, 1)

	// another round of the retry loop finds nothing left to do
	e.skipBackoff()
	time.Sleep(5 * testRetryTick)
	expectPRPipelines(t, e.n, 1)
	e.sink.expect(t, 1)
}

func TestPullRequestEventRunsOnceAfterARestart(t *testing.T) {
	e := newPRPipelineEnv(t)
	sha := e.head(t)
	e.ticketAs(pullRkey, "open", testBob, opCreate)

	// the webhook goes out, then spindle dies while it fetches the head
	release := e.holdGit()
	e.putPatch(testRepoDid, patchOne, pullRkey, sha, testBob)
	at := e.emitPatch(testRepoDid, patchOne, opCreate)
	eventually(t, 15*time.Second, "the head fetch to start", func() bool { return e.gitFetches.Load() > 0 })
	e.sink.expect(t, 1)
	e.n.stop()
	release(false)

	n2 := e.boot()
	got := expectPRPipelines(t, n2, 1)
	wantPRPipeline(t, got[0], "opened", sha, pullRkey, "pr.yaml")
	n2.consumed(at)
	e.sink.expect(t, 1)
}

func TestPullRequestEventHangingKnotDoesNotStallIngest(t *testing.T) {
	e := newPRPipelineEnv(t)
	sha := e.head(t)
	e.ticketAs(pullRkey, "open", testBob, opCreate)

	e.serve(faultOtherRepoDid, "other", testBob)
	e.emitDeclaration("other", faultOtherRepoDid)
	e.n.consumed(e.emitManifest(faultOtherRepoDid))
	eventually(t, 15*time.Second, "the other repo to be served", func() bool { return len(repoRows(t, e.n.s, faultOtherRepoDid)) == 1 })

	release := e.holdGit()
	defer release(false)
	e.putPatch(testRepoDid, patchOne, pullRkey, sha, testBob)
	e.emitPatch(testRepoDid, patchOne, opCreate)
	eventually(t, 15*time.Second, "the head fetch to start", func() bool { return e.gitFetches.Load() > 0 })

	// the next event is for another repo, so the stuck fetch can't be what it waits on
	e.putTicket(faultOtherRepoDid, pullRkey, "open", testBob, true)
	e.n.applied(faultOtherRepoDid, e.emitTicket(faultOtherRepoDid, pullRkey, opCreate))
	if _, err := e.n.s.db.GetPullTicket(faultOtherRepoDid, pullRkey); err != nil {
		t.Fatalf("the other repo's ticket wasn't stored: %v", err)
	}

	release(false)
	got := expectPRPipelines(t, e.n, 1)
	wantPRPipeline(t, got[0], "opened", sha, pullRkey, "pr.yaml")
}

func TestPullRequestPipelineRetriesAFailedFetch(t *testing.T) {
	e := newPRPipelineEnv(t)
	sha := e.head(t)
	e.ticketAs(pullRkey, "open", testBob, opCreate)

	e.gitFailing.Store(true)
	e.patchAs(patchOne, pullRkey, sha, testBob)
	e.sink.expect(t, 1)
	expectPRPipelines(t, e.n, 0)

	e.gitFailing.Store(false)
	e.skipBackoff()
	got := expectPRPipelines(t, e.n, 1)
	wantPRPipeline(t, got[0], "opened", sha, pullRkey, "pr.yaml")
	e.sink.expect(t, 1)
}

func TestPullRequestReplaysRunNothingTwice(t *testing.T) {
	e := newPRPipelineEnv(t)
	sha := e.head(t)
	e.ticketAs(pullRkey, "open", testBob, opCreate)
	e.patchAs(patchOne, pullRkey, sha, testBob)
	e.ticketAs(pullRkey, "closed", testOwner, "update")
	expectPRPipelines(t, e.n, 2)
	e.sink.expect(t, 2)

	e.emitTicket(testRepoDid, pullRkey, opCreate)
	e.emitPatch(testRepoDid, patchOne, opCreate)
	e.n.consumed(e.emitTicket(testRepoDid, pullRkey, "update"))
	expectPRPipelines(t, e.n, 2)
	e.sink.expect(t, 2)
}

func TestPullTriggerCarriesTheActorAndOnlyATargetRepoBranch(t *testing.T) {
	repo := &workflow.TriggerRepo{Did: testOwner.String()}
	for _, c := range []struct {
		name   string
		head   db.PullHead
		branch string
	}{
		{"an old patch", db.PullHead{Sha: patchSha2}, ""},
		{"a branch of the repo", db.PullHead{Sha: patchSha2, SourceRepo: testRepoDid.String(), SourceBranch: "feature"}, "feature"},
		{"a fork's branch", db.PullHead{Sha: patchSha2, SourceRepo: faultOtherRepoDid.String(), SourceBranch: "feature"}, ""},
	} {
		ev := db.PullEvent{RepoDid: testRepoDid, Ticket: pullRkey, Event: models.WebhookEventPullRequestMerged, Author: testBob, Sender: testOwner, Head: c.head}
		got := pullTrigger(repo, ev, "main")
		pr := got.PullRequest
		if got.Kind != "pull_request" || got.Repo != repo || got.Actor == nil || *got.Actor != testOwner.String() {
			t.Errorf("%s: trigger kind %q repo %v actor %v, want the merger %s as actor", c.name, got.Kind, got.Repo, got.Actor, testOwner)
		}
		if pr == nil || pr.Action == nil || *pr.Action != "merged" || pr.SourceSha != patchSha2 || pr.TargetBranch != "main" || pr.SourceBranch != c.branch {
			t.Errorf("%s: pull request trigger %+v, want source branch %q", c.name, pr, c.branch)
		}
	}
}

func TestPullRequestSourceBranchComesFromThePatch(t *testing.T) {
	e := newPRPipelineEnv(t)
	sha := e.head(t)
	e.ticketAs(pullRkey, "open", testBob, opCreate)
	record := e.putPatch(testRepoDid, patchOne, pullRkey, sha, testBob)
	record["source"] = map[string]any{"repo": testRepoDid.String(), "branch": "feature"}
	e.knot.putVersion(testRepoDid, org_tangled.ReviewPatchNSID, patchOne, testBob, record)
	e.n.consumed(e.emitPatch(testRepoDid, patchOne, opCreate))

	got := expectPRPipelines(t, e.n, 1)
	if b := got[0].Trigger.PullRequest.SourceBranch; b == nil || *b != "feature" {
		t.Errorf("pipeline source branch %v, want feature", b)
	}
	p := pullPayload(t, e.sink.expect(t, 1)[0])
	if p.PullRequest.Source == nil || p.PullRequest.Source.Branch != "feature" || p.PullRequest.Source.Repo != testRepoDid.String() || p.PullRequest.Source.Sha != sha {
		t.Errorf("webhook source %+v, want feature of %s at %s", p.PullRequest.Source, testRepoDid, sha)
	}
}

func TestRefusedByKnotOnlyCountsAnswersARetryWouldGetAgain(t *testing.T) {
	for stderr, want := range map[string]bool{
		"error: RPC failed; HTTP 400 curl 22 The requested URL returned error: 400\nfatal: expected 'packfile'": true,
		"fatal: unable to access 'https://knot.test/did:plc:repo/': The requested URL returned error: 404":      true,
		"fatal: unable to access 'https://knot.test/did:plc:repo/': The requested URL returned error: 429":      false,
		"error: RPC failed; HTTP 502 curl 22 The requested URL returned error: 502":                             false,
		"fatal: unable to access 'https://knot.test/did:plc:repo/': Could not resolve host: knot.test":          false,
		"fatal: remote error: upload-pack: not our ref 0123456789012345678901234567890123456789":                true,
		"error: Server does not allow request for unadvertised object 0123456789012345678901234567890123456789": true,
		"remote: no git\nfatal: repository 'http://127.0.0.1:42617/did:plc:repo/' not found":                    true,
		"fatal: unable to access 'https://knot.test/did:plc:repo/': Could not resolve host: not found":          false,
	} {
		if got := refusedByKnot(errors.New("git fetch: exit status 128: " + stderr)); got != want {
			t.Errorf("%q: refused = %v, want %v", stderr, got, want)
		}
	}
}