package spindle

import (
	"context"
	"fmt"
	"net/http"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/bluesky-social/indigo/atproto/syntax"
	"tangled.org/core/api/org_tangled"
	"tangled.org/core/spindle/db"
	"tangled.org/core/spindle/knotclient"
	"tangled.org/core/tapc"
)

func didSharingStripe(with syntax.DID) syntax.DID {
	for i := 0; ; i++ {
		did := syntax.DID(fmt.Sprintf("did:plc:share%d", i))
		if did != with && stripeOf(did) == stripeOf(with) {
			return did
		}
	}
}

func goErr(fn func() error) <-chan error {
	done := make(chan error, 1)
	go func() { done <- fn() }()
	return done
}

func waitErr(t *testing.T, done <-chan error, within time.Duration, timedOut string) error {
	t.Helper()
	select {
	case err := <-done:
		return err
	case <-time.After(within):
		t.Fatal(timedOut)
		return nil
	}
}

func ensureAsync(env *ensureEnv) <-chan error {
	return goErr(func() error {
		_, err := env.s.ensureRepo(context.Background(), env.k, testRepoDid)
		return err
	})
}

func pullEventRows(t *testing.T, s *Spindle, repo syntax.DID) int {
	t.Helper()
	var n int
	if err := s.db.QueryRow("select count(*) from pull_events where repo_did = ?", repo.String()).Scan(&n); err != nil {
		t.Fatal(err)
	}
	return n
}

func hungPatchHandler(t *testing.T) (e *pullEnv, done <-chan error, release func()) {
	t.Helper()
	e = newPullEnv(t)
	e.putTicket(testRepoDid, pullRkey, "open", testBob, true)
	e.putPatch(testRepoDid, patchOne, pullRkey, patchSha1, testBob)
	if repo, err := e.n.s.servedRepo(testRepoDid); err != nil || repo == nil {
		t.Fatalf("setup: the repo isn't served: %v", err)
	}
	rec := &tapc.RecordEventData{
		Did:        testRepoDid,
		Collection: syntax.NSID(org_tangled.ReviewPatchNSID),
		Rkey:       syntax.RecordKey(patchOne),
		Action:     tapc.RecordCreateAction,
		Live:       true,
	}
	retry := db.RecordRetry{RepoDid: testRepoDid, Collection: org_tangled.ReviewPatchNSID, Rkey: patchOne}
	entered, gate := make(chan struct{}), make(chan struct{})
	done = goErr(func() error {
		return e.n.s.applyPullRecord(context.Background(), rec, retry, func(ctx context.Context, repo *db.Repo, rec *tapc.RecordEventData) error {
			close(entered)
			<-gate
			return e.n.s.applyPatch(ctx, repo, rec)
		})
	})
	var once sync.Once
	release = func() { once.Do(func() { close(gate) }) }
	t.Cleanup(release)
	select {
	case <-entered:
	case <-time.After(5 * time.Second):
		t.Fatal("the handler never started")
	}
	return e, done, release
}

func TestRepoLockExcludesTheSameDid(t *testing.T) {
	s := &Spindle{}
	a := syntax.DID("did:plc:locka")

	unlock := s.lockRepo(a).Unlock
	got := make(chan struct{})
	go func() {
		s.lockRepo(a).Unlock()
		close(got)
	}()
	select {
	case <-got:
		t.Fatal("a second lock on the same repo was granted while the first was held")
	case <-time.After(200 * time.Millisecond):
	}
	unlock()
	select {
	case <-got:
	case <-time.After(time.Second):
		t.Fatal("the second lock was never granted after the first was released")
	}
}

func hungPush(t *testing.T) (e *pushEnv, release func(fail bool), answers <-chan notifyAnswer) {
	t.Helper()
	e = newPushEnv(t)
	release = e.w.holdGit()
	answers = e.notifyAsync(context.Background(), e.n.s, e.firstPush(t))
	e.waitClone(t)
	return e, release, answers
}

func TestHungFetchDoesNotDelayAnUnrelatedRepo(t *testing.T) {
	e, release, answers := hungPush(t)
	defer release(false)

	other := didSharingStripe(testRepoDid)
	row := db.Repo{Knot: e.w.knotRef().Host, Owner: testOwner, Rkey: "other", RepoDid: other, Name: "other"}
	if err := e.n.s.db.AddRepo(row); err != nil {
		t.Fatal(err)
	}
	push := faultPush("refs/heads/main", zeroSha, strings.Repeat("b", 40))
	push.EventPush.PushOptions = []string{"skip-ci"}

	done := goErr(func() error {
		_, err := e.n.s.NotifyEvent(context.Background(), e.w.knotRef(), other, push)
		return err
	})
	if err := waitErr(t, done, 3*time.Second, "a notify for an unrelated repo took more than 3s while another repo's fetch hung"); err != nil {
		t.Fatalf("notify for the unrelated repo: %v", err)
	}
	release(false)
	awaitAnswer(t, answers)
}

func TestHungFetchDoesNotDelayARefreshOfTheSameRepo(t *testing.T) {
	e, release, answers := hungPush(t)
	defer release(false)

	done := goErr(func() error {
		_, err := e.n.s.ensureRepo(context.Background(), e.w.knotRef(), testRepoDid)
		return err
	})
	if err := waitErr(t, done, 3*time.Second, "a refresh of a repo took more than 3s while its own push's fetch hung"); err != nil {
		t.Fatalf("refresh of the repo: %v", err)
	}
	release(false)
	awaitAnswer(t, answers)
}

func TestPipelineInsertWaitsForTheRepoLock(t *testing.T) {
	e, release, answers := hungPush(t)
	defer release(false)

	locked := make(chan func(), 1)
	go func() { locked <- e.n.s.lockRepo(testRepoDid).Unlock }()
	var unlock func()
	select {
	case unlock = <-locked:
	case <-time.After(3 * time.Second):
		t.Fatal("the repo lock couldn't be taken while a fetch for the repo hung")
	}

	release(false)
	select {
	case a := <-answers:
		unlock()
		t.Fatalf("the push answered %q, %v while the repo lock was held", a.uri, a.err)
	case <-time.After(1500 * time.Millisecond):
	}
	e.wantPipelines(t, 0, "while the repo lock was held")

	unlock()
	if a := awaitAnswer(t, answers); a.err != nil || a.uri == "" {
		t.Fatalf("the push answered %q, %v after the lock was released, want a pipeline", a.uri, a.err)
	}
	e.wantPipelines(t, 1, "after the repo lock was released")
}

func TestConcurrentEnrolmentMakesOneRow(t *testing.T) {
	env := newEnsureEnv(t)
	env.serve("repo", testBob)

	var wg sync.WaitGroup
	errs := make(chan error, 8)
	for range 8 {
		wg.Go(func() {
			_, err := env.s.ensureRepo(context.Background(), env.k, testRepoDid)
			errs <- err
		})
	}
	wg.Wait()
	close(errs)
	for err := range errs {
		if err != nil {
			t.Fatalf("ensureRepo: %v", err)
		}
	}
	if rows := env.rows(t); len(rows) != 1 {
		t.Fatalf("%d repo rows after 8 concurrent enrolments, want 1: %+v", len(rows), rows)
	}
	collaborators, err := env.s.db.ListCollaboratorsByRepoDid(testRepoDid)
	if err != nil || len(collaborators) != 1 {
		t.Fatalf("collaborators = %+v, %v, want exactly one", collaborators, err)
	}
}

func TestPullHandlerHeldMidIoLeavesNoRowsAfterAWipe(t *testing.T) {
	e, done, release := hungPatchHandler(t)

	if err := e.n.s.WipeRepo(context.Background(), testRepoDid, "test"); err != nil {
		t.Fatalf("wipe: %v", err)
	}
	release()
	waitErr(t, done, 10*time.Second, "the handler never finished")

	for _, table := range []string{"pull_tickets", "pull_patches", "pull_events"} {
		var n int
		if err := e.n.s.db.QueryRow("select count(*) from "+table+" where repo_did = ?", testRepoDid.String()).Scan(&n); err != nil {
			t.Fatal(err)
		}
		if n != 0 {
			t.Errorf("%d rows in %s for a wiped repo, want 0", n, table)
		}
	}
}

type gatedTransport struct {
	match func(*http.Request) bool
	// answers before the gate, and the caller gets the old reply
	afterwards bool
	first      atomic.Bool
	entered    chan struct{}
	gate       chan struct{}
}

func (g *gatedTransport) RoundTrip(r *http.Request) (*http.Response, error) {
	if !g.match(r) || !g.first.CompareAndSwap(false, true) {
		return http.DefaultTransport.RoundTrip(r)
	}
	var resp *http.Response
	if g.afterwards {
		var err error
		if resp, err = http.DefaultTransport.RoundTrip(r); err != nil {
			return nil, err
		}
	}
	close(g.entered)
	select {
	case <-g.gate:
	case <-r.Context().Done():
		return nil, r.Context().Err()
	}
	if resp != nil {
		return resp, nil
	}
	return http.DefaultTransport.RoundTrip(r)
}

func TestStaleRefreshDoesNotOverwriteANewerOne(t *testing.T) {
	env := newEnsureEnv(t)
	env.serve("repo")
	if env.ensure(t) == nil {
		t.Fatal("setup: the repo isn't served")
	}

	gate := &gatedTransport{
		match: func(r *http.Request) bool {
			return strings.Contains(r.URL.RawQuery, "collection="+org_tangled.RepoDeclarationNSID)
		},
		entered: make(chan struct{}),
		gate:    make(chan struct{}),
	}
	env.s.knots = knotclient.New(env.s.res.Directory(), &http.Client{Transport: gate}, true)

	first := ensureAsync(env)
	select {
	case <-gate.entered:
	case <-time.After(5 * time.Second):
		t.Fatal("the first refresh never reached its declaration read")
	}

	env.manifest("repo", "did:web:other.example.com")
	second := ensureAsync(env)
	// the second can finish alone or wait on the first, either is fine
	time.Sleep(2 * time.Second)

	close(gate.gate)
	if err := waitErr(t, first, 5*time.Second, "the first refresh never finished"); err != nil {
		t.Fatalf("first refresh: %v", err)
	}
	if err := waitErr(t, second, 5*time.Second, "the second refresh never finished"); err != nil {
		t.Fatalf("second refresh: %v", err)
	}
	if rows := env.rows(t); len(rows) != 0 {
		t.Fatalf("repo rows = %+v, want none: the refresh that read the old manifest was applied after the newer one", rows)
	}
}

func TestStaleForgetDoesNotUndoANewerRegister(t *testing.T) {
	env := newEnsureEnv(t)
	env.serve("repo")
	if env.ensure(t) == nil {
		t.Fatal("setup: the repo isn't served")
	}

	gate := &gatedTransport{
		match: func(r *http.Request) bool {
			return strings.Contains(r.URL.RawQuery, "collection="+org_tangled.RepoManifestNSID)
		},
		afterwards: true,
		entered:    make(chan struct{}),
		gate:       make(chan struct{}),
	}
	env.manifest("repo", "did:web:other.example.com")
	env.s.knots = knotclient.New(env.s.res.Directory(), &http.Client{Transport: gate}, true)

	first := ensureAsync(env)
	select {
	case <-gate.entered:
	case <-time.After(5 * time.Second):
		t.Fatal("the first refresh never read the manifest")
	}

	env.manifest("repo", env.spindleDid())
	if _, err := env.s.ensureRepo(context.Background(), env.k, testRepoDid); err != nil {
		t.Fatalf("second refresh: %v", err)
	}
	close(gate.gate)
	if err := waitErr(t, first, 5*time.Second, "the first refresh never finished"); err != nil {
		t.Fatalf("first refresh: %v", err)
	}
	if rows := env.rows(t); len(rows) != 1 {
		t.Fatalf("%d repo rows, want 1: the refresh that read the old manifest forgot a repo a newer one had kept", len(rows))
	}
}

// turns are ordered per did, sharing a stripe only shares the mutex
func TestTurnsOfDidsSharingAStripeBothApply(t *testing.T) {
	env := newEnsureEnv(t)
	a := syntax.DID("did:plc:stripea")
	b := didSharingStripe(a)

	older, newer := env.s.beginTurn(a), env.s.beginTurn(b)
	for name, turn := range map[string]*repoTurn{"newer": newer, "older": older} {
		applied, err := turn.apply(func() error { return nil })
		if err != nil || !applied {
			t.Errorf("the %s turn applied=%v err=%v, want it applied", name, applied, err)
		}
	}
}

func TestOlderTurnOfTheSameDidIsSkipped(t *testing.T) {
	env := newEnsureEnv(t)
	a := syntax.DID("did:plc:turns")

	older, newer := env.s.beginTurn(a), env.s.beginTurn(a)
	if applied, err := newer.apply(func() error { return nil }); err != nil || !applied {
		t.Fatalf("the newer turn applied=%v err=%v", applied, err)
	}
	if applied, err := newer.apply(func() error { return nil }); err != nil || !applied {
		t.Fatalf("a second write of the same turn applied=%v err=%v, want it applied", applied, err)
	}
	ran := false
	if applied, err := older.apply(func() error { ran = true; return nil }); err != nil || applied || ran {
		t.Fatalf("the older turn applied=%v ran=%v err=%v, want it skipped", applied, ran, err)
	}
}

func TestHungPullHandlerDoesNotDelayAPushOrAPullEventOfAStripeNeighbour(t *testing.T) {
	e, _, _ := hungPatchHandler(t)
	neighbour := didSharingStripe(testRepoDid)
	row := db.Repo{Knot: e.knotRef().Host, Owner: testOwner, Rkey: "neighbour", RepoDid: neighbour, Name: "neighbour"}
	if err := e.n.s.db.AddRepo(row); err != nil {
		t.Fatal(err)
	}

	push := faultPush("refs/heads/main", zeroSha, strings.Repeat("b", 40))
	push.EventPush.PushOptions = []string{"skip-ci"}
	pushed := goErr(func() error {
		_, err := e.n.s.NotifyEvent(context.Background(), e.knotRef(), neighbour, push)
		return err
	})
	if err := waitErr(t, pushed, 3*time.Second, "a push for a repo on the same stripe took more than 3s while a pull handler hung on the knot"); err != nil {
		t.Fatalf("push for the neighbour: %v", err)
	}

	patch := db.PullPatch{Rkey: "p1", Ticket: "t1", Head: patchSha1, CreatedAt: "2026-02-01T00:00:00Z"}
	ev := db.PullEvent{RepoDid: neighbour, Ticket: "t1", Author: testBob, Sender: testBob, Pull: []byte("{}"), Pipeline: "p1"}
	if err := e.n.s.db.AddPullPatch(neighbour, patch, ev); err != nil {
		t.Fatal(err)
	}
	e.n.s.retries.jobs.kick(neighbour)
	eventually(t, 5*time.Second, "the neighbour's pull event to run while a pull handler hung", func() bool {
		return pullEventRows(t, e.n.s, neighbour) == 0
	})
}

func TestKicksDuringAPullEventDoNotRunItTwice(t *testing.T) {
	e := newPushEnv(t)
	sink := newSink(t)
	e.w.subscribe(e.n, sink, pullEvents...)
	sha := e.repo.commit(t)
	e.w.putTicket(testRepoDid, pullRkey, "open", testBob, true)
	e.w.putPatch(testRepoDid, patchOne, pullRkey, sha, testBob)

	release := e.w.holdGit()
	defer release(false)
	e.w.emitTicket(testRepoDid, pullRkey, opCreate)
	e.w.emitPatch(testRepoDid, patchOne, opCreate)
	e.waitClone(t)

	e.n.s.retries.jobs.kick(testRepoDid)
	e.n.s.retries.jobs.kick(testRepoDid)
	e.n.s.kickDueRepos()
	time.Sleep(500 * time.Millisecond)
	release(false)

	eventually(t, 10*time.Second, "the pull event to finish", func() bool { return pullEventRows(t, e.n.s, testRepoDid) == 0 })
	sink.expect(t, 1)
}

// the stored turn sits an hour ahead, like after a clock stepped back
func TestATurnAfterARestartIsNotBehindTheStoredOnes(t *testing.T) {
	s := newWipeTestSpindle(t)
	a := syntax.DID("did:plc:turns")
	ahead := time.Now().Add(time.Hour).UnixNano()
	if _, err := s.db.AdvanceRepoTurn(a, ahead, time.Now()); err != nil {
		t.Fatal(err)
	}

	applied, err := s.beginTurn(a).apply(func() error { return nil })
	if err != nil || !applied {
		t.Fatalf("the first turn after a restart applied=%v err=%v, want it applied over a stored one from the future", applied, err)
	}
}