package spindle

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

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

	"tangled.org/core/api/org_tangled"
	"tangled.org/core/jetstream"
	"tangled.org/core/rbac"
	"tangled.org/core/spindle/knotclient"
	"tangled.org/core/spindle/models"
)

var (
	opCreate = jetstream.CommitOperationCreate
	opDelete = jetstream.CommitOperationDelete
)

// waits until spindle has finished the setup work, so a restart doesn't find some of it left over
func (w *world) seed(n *node, collaborators ...syntax.DID) {
	t := w.t
	t.Helper()
	w.serve(testRepoDid, "repo", collaborators...)
	w.emitDeclaration("repo", testRepoDid)
	last := w.emitManifest(testRepoDid)
	eventually(t, 15*time.Second, "the repo to be served", func() bool { return len(repoRows(t, n.s, testRepoDid)) == 1 })
	for _, c := range collaborators {
		eventually(t, 15*time.Second, "the collaborators to be synced", func() bool { return isCollaborator(t, n.s, testRepoDid, c) })
	}
	eventually(t, 15*time.Second, "hydrant to backfill the owner", func() bool { return w.hyd.repoStatus(testOwner) == synced(testOwner).Status })
	n.applied(testRepoDid, last)
}

// a restart rewinds the cursor a second, so push it past everything first
func (w *world) pushCursorPast(n *node) {
	t := w.t
	t.Helper()
	at := w.js.filler()
	n.consumed(at)
	n.stop()
	if n.cursor() <= at {
		t.Fatalf("setup: the saved cursor %d never reached the filler event at %d", n.cursor(), at)
	}
}

func TestShortOutageReplaysFromTheSavedCursor(t *testing.T) {
	w := newWorld(t)
	n := w.boot()
	w.seed(n, testBob, testCarol)
	n.stop()
	erased := len(w.hyd.erasedDids())

	w.addCollaborator(testRepoDid, testDave)
	w.emitCollaborator(testRepoDid, testDave, opCreate)
	w.removeCollaborator(testRepoDid, testCarol)
	w.emitCollaborator(testRepoDid, testCarol, opDelete)
	w.setManifest(testRepoDid, "Repo", w.spindleDid())
	w.emitManifest(testRepoDid)

	n2 := w.boot()
	eventually(t, 15*time.Second, "the replay to bring back the collaborators and the manifest", func() bool {
		rows := repoRows(t, n2.s, testRepoDid)
		return isCollaborator(t, n2.s, testRepoDid, testDave) && !isCollaborator(t, n2.s, testRepoDid, testCarol) && len(rows) == 1 && rows[0].Name == "Repo"
	})
	if !isCollaborator(t, n2.s, testRepoDid, testBob) {
		t.Error("bob lost access during the replay")
	}
	if got := w.hyd.erasedDids(); len(got) != erased {
		t.Errorf("a catch-up ran for a short outage, hydrant erased %v", got[erased:])
	}
}

func TestLongOutageCatchUpBringsBackWhatChangedWhileDown(t *testing.T) {
	scenarios := []struct {
		name   string
		change func(w *world)
		check  func(t *testing.T, n *node) bool
	}{
		{
			name:   "collaborator added",
			change: func(w *world) { w.addCollaborator(testRepoDid, testDave) },
			check: func(t *testing.T, n *node) bool {
				return isCollaborator(t, n.s, testRepoDid, testDave) && isCollaborator(t, n.s, testRepoDid, testBob)
			},
		},
		{
			name:   "collaborator removed",
			change: func(w *world) { w.removeCollaborator(testRepoDid, testBob) },
			check: func(t *testing.T, n *node) bool {
				if isCollaborator(t, n.s, testRepoDid, testBob) {
					return false
				}
				if ok, _ := n.s.e.IsSettingsAllowed(testBob.String(), rbac.ThisServer, testRepoDid.String()); ok {
					t.Error("a removed collaborator can still reach the repo's secrets")
				}
				if secretCount(t, n.s, testRepoDid) != 1 {
					t.Error("removing a collaborator dropped the secrets")
				}
				return isCollaborator(t, n.s, testRepoDid, testCarol)
			},
		},
		{
			name:   "manifest switched to another spindle",
			change: func(w *world) { w.setManifest(testRepoDid, "repo", "did:web:other.example.com") },
			check: func(t *testing.T, n *node) bool {
				if len(repoRows(t, n.s, testRepoDid)) != 0 {
					return false
				}
				if ok, _ := n.s.e.IsRepoOwner(testOwner.String(), rbac.ThisServer, testRepoDid.String()); ok {
					t.Error("the owner kept access to a repo this spindle forgot")
				}
				if isCollaborator(t, n.s, testRepoDid, testBob) {
					t.Error("a collaborator kept access to a repo this spindle forgot")
				}
				if secretCount(t, n.s, testRepoDid) != 1 {
					t.Error("forgetting a repo dropped its secrets, they have to stay for a move back")
				}
				return true
			},
		},
		{
			name: "manifest switched to this spindle",
			change: func(w *world) {
				w.setManifest(faultOtherRepoDid, "other", w.spindleDid())
			},
			check: func(t *testing.T, n *node) bool {
				return len(repoRows(t, n.s, faultOtherRepoDid)) == 1 && isCollaborator(t, n.s, faultOtherRepoDid, testBob)
			},
		},
		{
			name: "repo deleted",
			change: func(w *world) {
				w.knot.deleteRepo(testRepoDid)
				w.removeCollaborator(testRepoDid, testBob)
				w.removeCollaborator(testRepoDid, testCarol)
				w.undeclare("repo")
			},
			check: func(t *testing.T, n *node) bool {
				if len(repoRows(t, n.s, testRepoDid)) != 0 {
					return false
				}
				if secretCount(t, n.s, testRepoDid) != 0 {
					t.Error("a deleted repo kept its secrets")
				}
				return true
			},
		},
		{
			name: "repo renamed",
			change: func(w *world) {
				w.setManifest(testRepoDid, "New-Name", w.spindleDid())
				w.undeclare("repo")
				w.declare("New-Name", testRepoDid)
			},
			check: func(t *testing.T, n *node) bool {
				rows := repoRows(t, n.s, testRepoDid)
				if len(rows) != 1 || rows[0].Rkey != "new-name" {
					return false
				}
				if rows[0].Name != "New-Name" || !isCollaborator(t, n.s, testRepoDid, testBob) || secretCount(t, n.s, testRepoDid) != 1 {
					t.Errorf("rename lost something, rows = %s, bob = %v, secrets = %d", describeRows(rows), isCollaborator(t, n.s, testRepoDid, testBob), secretCount(t, n.s, testRepoDid))
				}
				return true
			},
		},
	}

	for _, sc := range scenarios {
		t.Run(sc.name, func(t *testing.T) {
			w := newWorld(t)
			w.setManifest(faultOtherRepoDid, "other", "did:web:other.example.com")
			w.declare("other", faultOtherRepoDid)
			w.addCollaborator(faultOtherRepoDid, testBob)

			n := w.boot()
			w.seed(n, testBob, testCarol)
			addTestSecret(t, n.s, testRepoDid)
			if len(repoRows(t, n.s, faultOtherRepoDid)) != 0 {
				t.Fatal("setup: the other repo is served already")
			}
			n.stop()

			// jetstream forgets, so only hydrant can bring these changes back
			sc.change(w)
			w.js.forget()
			w.js.filler()

			n2 := w.boot()
			eventually(t, 20*time.Second, "the catch-up to reach hydrant", func() bool {
				return slices.Contains(w.hyd.erasedDids(), testRepoDid)
			})
			eventually(t, 20*time.Second, "the catch-up to bring back what changed", func() bool { return sc.check(t, n2) })
		})
	}
}

func TestAnEventCutShortByShutdownIsNotLost(t *testing.T) {
	w := newWorld(t)
	n := w.boot()
	// a first boot with no saved cursor connects live, so save one first
	n.consumed(w.js.filler())

	w.serve(testRepoDid, "repo", testBob)
	release := w.holdManifest()
	defer release(false)
	w.emitManifest(testRepoDid)
	eventually(t, 10*time.Second, "the event to start its refresh", func() bool { return w.heldReads.Load() > 0 })

	n.stop()
	release(false)
	if len(repoRows(t, n.s, testRepoDid)) != 0 {
		t.Fatal("setup: the cut event was handled anyway")
	}

	n2 := w.boot()
	eventually(t, 15*time.Second, "the cut event to be replayed", func() bool {
		return len(repoRows(t, n2.s, testRepoDid)) == 1 && isCollaborator(t, n2.s, testRepoDid, testBob)
	})
}

func TestSecretsSurviveEveryWayTheWorldCanLookBroken(t *testing.T) {
	w := newWorld(t)
	n := w.boot()
	w.seed(n, testBob)
	addTestSecret(t, n.s, testRepoDid)

	faults := []struct {
		name string
		set  func(on bool)
	}{
		{"bare 404 from a proxy", w.setBare404},
		{"did document names another knot", func(on bool) {
			knot := w.knot
			if on {
				knot = w.other
			}
			w.plc.setKnot(testRepoDid, knot.srv.URL)
			n.s.res.InvalidateIdent(context.Background(), testRepoDid.String())
		}},
		{"declaration missing", func(on bool) {
			if on {
				w.undeclare("repo")
			} else {
				w.declare("repo", testRepoDid)
			}
		}},
		{"pds error", w.setPDSFailing},
		{"knot outage", w.setKnotDown},
	}
	for i, f := range faults {
		f.set(true)
		before := w.traffic()
		w.emitManifest(testRepoDid)
		eventually(t, 10*time.Second, "spindle to look at the repo", func() bool { return w.traffic() > before })
		w.quiet()
		f.set(false)
		// skip the backoff the failed refresh left behind
		w.skew.Add(int64(2 * time.Hour))

		// events run in order, so the marker landing means the broken one is done
		marker := syntax.DID("did:plc:marker" + strings.Repeat("x", i+1))
		w.addCollaborator(testRepoDid, marker)
		w.emitCollaborator(testRepoDid, marker, opCreate)
		eventually(t, 20*time.Second, "the next change to be applied", func() bool { return isCollaborator(t, n.s, testRepoDid, marker) })

		if rows := repoRows(t, n.s, testRepoDid); len(rows) != 1 {
			t.Errorf("%s: rows = %s", f.name, describeRows(rows))
		}
		if got := secretCount(t, n.s, testRepoDid); got != 1 {
			t.Errorf("%s: %d secrets left, want 1", f.name, got)
		}
		if !isCollaborator(t, n.s, testRepoDid, testBob) {
			t.Errorf("%s: bob lost access", f.name)
		}
	}
}

func TestRefreshThatFailedAfterALiveEventIsRetriedAfterARestart(t *testing.T) {
	w := newWorld(t)
	n := w.boot()
	w.seed(n, testBob)

	w.setKnotDown(true)
	before := w.traffic()
	w.addCollaborator(testRepoDid, testCarol)
	w.emitCollaborator(testRepoDid, testCarol, opCreate)
	eventually(t, 10*time.Second, "the refresh to be attempted", func() bool { return w.traffic() > before })
	w.quiet()
	w.pushCursorPast(n)

	// retry backoff is too long to wait out, so skew the clock forward
	w.setKnotDown(false)
	w.skew.Store(int64(time.Hour))
	n2 := w.boot()
	eventually(t, 40*time.Second, "carol to get access after the restart", func() bool { return isCollaborator(t, n2.s, testRepoDid, testCarol) })
}

func TestCatchUpInterruptedByARestartIsNotForgotten(t *testing.T) {
	w := newWorld(t)
	n := w.boot()
	w.seed(n, testBob)
	n.stop()

	w.addCollaborator(testRepoDid, testCarol)
	w.emitCollaborator(testRepoDid, testCarol, opCreate)
	w.js.forget()
	w.js.filler()
	w.hyd.failStartsWith(errors.New("hydrant cannot start"))

	n2 := w.boot()
	w.pushCursorPast(n2)

	w.hyd.failStartsWith(nil)
	n3 := w.boot()
	eventually(t, 40*time.Second, "the catch-up to run on the next boot", func() bool { return isCollaborator(t, n3.s, testRepoDid, testCarol) })
}

func TestPartlyFailedCatchUpIsRetriedOnTheNextBoot(t *testing.T) {
	w := newWorld(t)
	n := w.boot()
	w.seed(n, testBob)
	n.stop()

	// only the repo backfill can see this, the old owner's declaration is gone
	w.knot.put(testRepoDid, org_tangled.RepoManifestNSID, knotclient.ManifestRkey, manifestRecord(testNewOwner, "repo", w.spindleDid()))
	w.undeclare("repo")
	w.knot.declare(testNewOwner, "repo", testRepoDid)
	w.plc.setPDS(testNewOwner, w.knot.srv.URL)
	w.js.forget()
	w.js.filler()
	w.setHydrantFailing(testRepoDid, true)

	n2 := w.boot()
	eventually(t, 20*time.Second, "hydrant to give up on the repo", func() bool { return w.hyd.repoStatus(testRepoDid) == failedWith(testRepoDid, "transport", 0).Status })
	// wait until the catch-up is done with everything but the repo, which is left for the next boot
	eventually(t, 20*time.Second, "the catch-up to keep only the repo pending", func() bool {
		pending, err := n2.s.db.PendingCatchUp()
		return err == nil && slices.Equal(pending, []syntax.DID{testRepoDid})
	})
	w.pushCursorPast(n2)

	w.setHydrantFailing(testRepoDid, false)
	erased := len(w.hyd.erasedDids())
	n3 := w.boot()
	eventually(t, 40*time.Second, "the repo's backfill to be retried on the next boot", func() bool { return w.hyd.repoStatus(testRepoDid) == synced(testRepoDid).Status })
	eventually(t, 40*time.Second, "the repo to have moved", func() bool {
		rows := repoRows(t, n3.s, testRepoDid)
		return len(rows) == 1 && rows[0].Owner == testNewOwner
	})
	if got := w.hyd.erasedDids()[erased:]; slices.Contains(got, testOwner) {
		t.Errorf("the owner backfilled in the first catch-up but was redone, erased %v", got)
	}
}

type pushEnv struct {
	w    *world
	n    *node
	repo *fakeRepo
}

func newPushEnv(t *testing.T) *pushEnv {
	t.Helper()
	w := newWorld(t)
	repo := w.serveGitRepo(testRepoDid)
	n := w.boot()
	w.seed(n, testBob)
	return &pushEnv{w: w, n: n, repo: repo}
}

func faultPush(ref, oldSha, newSha 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: oldSha, NewSha: newSha, Actor: &actor,
	}}
}

var zeroSha = strings.Repeat("0", 40)

func (e *pushEnv) notify(s *Spindle, event *org_tangled.EventNotifyEvent_Input_Event) (models.PipelineId, error) {
	return s.NotifyEvent(context.Background(), e.w.knotRef(), testRepoDid, event)
}

func (e *pushEnv) firstPush(t *testing.T) *org_tangled.EventNotifyEvent_Input_Event {
	t.Helper()
	return faultPush("refs/heads/main", zeroSha, e.repo.commit(t))
}

func (e *pushEnv) waitClone(t *testing.T) {
	t.Helper()
	eventually(t, 10*time.Second, "the clone to start", func() bool { return e.w.gitFetches.Load() > 0 })
}

func (e *pushEnv) wantPipelines(t *testing.T, want int, when string) {
	t.Helper()
	if got := e.n.pipelines(testRepoDid); got != want {
		t.Fatalf("%d pipelines %s, want %d", got, when, want)
	}
}

func (e *pushEnv) mustNotify(t *testing.T, event *org_tangled.EventNotifyEvent_Input_Event) models.PipelineId {
	t.Helper()
	uri, err := e.notify(e.n.s, event)
	if err != nil {
		t.Fatalf("NotifyEvent: %v", err)
	}
	return uri
}

type notifyAnswer struct {
	uri models.PipelineId
	err error
}

func (e *pushEnv) notifyAsync(ctx context.Context, s *Spindle, event *org_tangled.EventNotifyEvent_Input_Event) <-chan notifyAnswer {
	answers := make(chan notifyAnswer, 1)
	go func() {
		uri, err := s.NotifyEvent(ctx, e.w.knotRef(), testRepoDid, event)
		answers <- notifyAnswer{uri, err}
	}()
	return answers
}

func awaitAnswer(t *testing.T, answers <-chan notifyAnswer) notifyAnswer {
	t.Helper()
	select {
	case a := <-answers:
		return a
	case <-time.After(30 * time.Second):
		t.Fatal("NotifyEvent never answered")
		return notifyAnswer{}
	}
}

func TestNotifyDoesNotAnswerBeforeThePushIsHandled(t *testing.T) {
	e := newPushEnv(t)
	push := e.firstPush(t)

	release := e.w.holdGit()
	defer release(false)
	answers := e.notifyAsync(context.Background(), e.n.s, push)
	e.waitClone(t)

	// the knot waits on this call, an early answer drops a push that then fails
	select {
	case a := <-answers:
		t.Fatalf("answered %q, %v while the push was still being cloned", a.uri, a.err)
	case <-time.After(3 * time.Second):
	}

	release(true)
	if a := awaitAnswer(t, answers); a.err == nil {
		t.Fatalf("answered %q without an error for a push that failed", a.uri)
	}
	e.wantPipelines(t, 0, "after a failed push")

	if uri := e.mustNotify(t, push); uri == "" {
		t.Fatal("the retry made no pipeline")
	}
	e.wantPipelines(t, 1, "after the retry")
}

func TestNotifyRetryAfterACrashBeforeThePipelineRowRunsThePush(t *testing.T) {
	e := newPushEnv(t)
	push := e.firstPush(t)

	release := e.w.holdGit()
	defer release(true)
	ctx, cancel := context.WithCancel(context.Background())
	answers := e.notifyAsync(ctx, e.n.s, push)
	e.waitClone(t)

	e.n.stop()
	cancel()
	if a := awaitAnswer(t, answers); a.err == nil {
		t.Fatalf("answered %q without an error although spindle went down first", a.uri)
	}
	release(true)

	n2 := e.w.boot()
	if got := n2.pipelines(testRepoDid); got != 0 {
		t.Fatalf("%d pipelines before the knot's retry", got)
	}
	uri, err := e.notify(n2.s, push)
	if err != nil || uri == "" {
		t.Fatalf("the knot's retry got %q, %v", uri, err)
	}
	if got := n2.pipelines(testRepoDid); got != 1 {
		t.Fatalf("%d pipelines after the retry, want 1", got)
	}
}

func TestNotifyReplayWhileTheFirstRunIsInFlightJoinsIt(t *testing.T) {
	e := newPushEnv(t)
	push := e.firstPush(t)

	release := e.w.holdGit()
	defer release(false)
	first := e.notifyAsync(context.Background(), e.n.s, push)
	e.waitClone(t)
	second := e.notifyAsync(context.Background(), e.n.s, push)
	time.Sleep(300 * time.Millisecond)
	release(false)

	a, b := awaitAnswer(t, first), awaitAnswer(t, second)
	if a.err != nil || b.err != nil || a.uri == "" || a.uri != b.uri {
		t.Fatalf("answers %+v and %+v, want one pipeline for both", a, b)
	}
	e.wantPipelines(t, 1, "after a replay joined the run in flight")
}

func TestNotifyReplayAfterItFinishedIsSkipped(t *testing.T) {
	// a restart forgets everything in memory, so that answer has to come from the db
	for _, restart := range []bool{false, true} {
		t.Run(fmt.Sprintf("restart=%v", restart), func(t *testing.T) {
			e := newPushEnv(t)
			push := e.firstPush(t)
			first := e.mustNotify(t, push)
			if first == "" {
				t.Fatal("the push made no pipeline")
			}
			fetches := e.w.gitFetches.Load()
			if restart {
				e.n.stop()
				e.n = e.w.boot()
			}

			if again := e.mustNotify(t, push); again != first {
				t.Fatalf("the replay answered %q, want %q", again, first)
			}
			e.wantPipelines(t, 1, "after a replay")
			if got := e.w.gitFetches.Load(); got != fetches {
				t.Fatalf("the replay cloned again, %d fetches after %d", got, fetches)
			}
		})
	}
}

func TestNotifyDedupeIsPerRepoRefOldAndNew(t *testing.T) {
	e := newPushEnv(t)
	c1, c2 := e.repo.commit(t), e.repo.commit(t)
	main := "refs/heads/main"

	created := e.mustNotify(t, faultPush(main, zeroSha, c1))
	moved := e.mustNotify(t, faultPush(main, c1, c2))
	// a revert to an old commit is a new push and still runs
	back := e.mustNotify(t, faultPush(main, c2, c1))
	if created == "" || moved == "" || back == "" || created == moved || moved == back || created == back {
		t.Fatalf("pipelines %q, %q, %q, want three different ones", created, moved, back)
	}
	e.wantPipelines(t, 3, "after three pushes")

	// the same old and new twice is skipped on purpose
	if again := e.mustNotify(t, faultPush(main, c1, c2)); again != moved {
		t.Fatalf("the repeated push answered %q, want %q", again, moved)
	}
	e.wantPipelines(t, 3, "after the repeat")
}

func TestNotifyDeleteAndRecreateRunsAgain(t *testing.T) {
	e := newPushEnv(t)
	c0, c1 := e.repo.commit(t), e.repo.commit(t)
	main := "refs/heads/main"

	e.mustNotify(t, faultPush(main, c0, c1))
	if gone := e.mustNotify(t, faultPush(main, c1, zeroSha)); gone != "" {
		t.Fatalf("a deleted branch answered %q", gone)
	}
	if again := e.mustNotify(t, faultPush(main, zeroSha, c1)); again == "" {
		t.Fatal("the branch came back at a commit that already ran and nothing ran")
	}
	e.wantPipelines(t, 2, "after the recreate")
}