package spindle

import (
	"context"
	"sync"
	"testing"
	"time"

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

	"tangled.org/core/api/org_tangled"
	"tangled.org/core/spindle/db"
	"tangled.org/core/tapc"
)

type jobLog struct {
	mu      sync.Mutex
	runs    map[syntax.DID]int
	current map[syntax.DID]int
	overlap bool
}

func (l *jobLog) started(did syntax.DID) int {
	l.mu.Lock()
	defer l.mu.Unlock()
	if l.runs == nil {
		l.runs, l.current = map[syntax.DID]int{}, map[syntax.DID]int{}
	}
	l.runs[did]++
	l.current[did]++
	if l.current[did] > 1 {
		l.overlap = true
	}
	return l.runs[did]
}

func (l *jobLog) ended(did syntax.DID) {
	l.mu.Lock()
	defer l.mu.Unlock()
	l.current[did]--
}

func (l *jobLog) count(did syntax.DID) int {
	l.mu.Lock()
	defer l.mu.Unlock()
	return l.runs[did]
}

func runWorkers(t *testing.T, jobs *repoJobs, n int, run func(context.Context, syntax.DID)) {
	ctx, cancel := context.WithCancel(context.Background())
	var workers sync.WaitGroup
	for range n {
		workers.Go(func() { jobs.work(ctx, run) })
	}
	t.Cleanup(func() { cancel(); workers.Wait() })
}

func TestRepoJobsRunOneJobPerRepoAndOneRerunForKicksDuringIt(t *testing.T) {
	var jobs repoJobs
	var log jobLog
	gate := make(chan struct{})
	stuck, other := syntax.DID("did:plc:stuck"), syntax.DID("did:plc:other")

	runWorkers(t, &jobs, 4, func(_ context.Context, did syntax.DID) {
		defer log.ended(did)
		if log.started(did) == 1 && did == stuck {
			<-gate
		}
	})

	jobs.kick(stuck)
	eventually(t, 2*time.Second, "the stuck repo's job to start", func() bool { return log.count(stuck) == 1 })

	jobs.kick(other)
	eventually(t, 2*time.Second, "another repo to run while the first is stuck", func() bool { return log.count(other) == 1 })

	for range 5 {
		jobs.kick(stuck)
	}
	time.Sleep(50 * time.Millisecond)
	if got := log.count(stuck); got != 1 {
		t.Fatalf("the stuck repo ran %d times while its job was still running", got)
	}

	close(gate)
	eventually(t, 2*time.Second, "the kicks during the job to cause a rerun", func() bool { return log.count(stuck) == 2 })
	time.Sleep(100 * time.Millisecond)
	if got := log.count(stuck); got != 2 {
		t.Fatalf("five kicks during one job ran it %d times, want one rerun", got)
	}
	if log.overlap {
		t.Fatal("two jobs for one repo ran at once")
	}
}

func markEvent(did syntax.DID, spindle string, owner syntax.DID) tapc.Event {
	manifest := org_tangled.RepoManifest{
		Spindle:     ptr(spindle),
		Declaration: &org_tangled.RepoManifest_Declaration{Owner: owner.String(), Name: "repo"},
	}
	return recordEvent(did, org_tangled.RepoManifestNSID, "self", tapc.RecordUpdateAction, manifest)
}

func TestAnEventUnderMarkOnlyCostsNoReadsAndLeavesAMarkForTheJob(t *testing.T) {
	env := newEnsureEnv(t)
	env.serve("repo", testBob)
	ctx := withMarkOnly(context.Background())

	if err := env.s.processEvent(ctx, markEvent(testRepoDid, env.spindleDid(), testOwner)); err != nil {
		t.Fatal(err)
	}
	if got := env.knot.count(); got != 0 {
		t.Fatalf("marking a repo cost %d knot requests", got)
	}
	if attempts, marked, err := env.s.db.RepoRetryRow(testRepoDid); err != nil || !marked || attempts != 0 {
		t.Fatalf("mark = attempts %d, exists %v, %v, want a fresh row", attempts, marked, err)
	}
	if rows := env.rows(t); len(rows) != 0 {
		t.Fatalf("the repo was served before its job ran, rows = %+v", rows)
	}

	env.s.runRepoJob(context.Background(), testRepoDid)
	if rows := env.rows(t); len(rows) != 1 || !env.collaborator(t, testBob) {
		t.Fatalf("the job didn't serve the repo, rows = %+v", rows)
	}
	requireMarked(t, env.s.db, testRepoDid, false, "a finished job left its mark")

	before := env.knot.count()
	env.knot.erase(testRepoDid, org_tangled.GraphCollaboratorNSID, testBob.String())
	collaborator := recordEvent(testRepoDid, org_tangled.GraphCollaboratorNSID, testBob.String(), tapc.RecordDeleteAction, nil)
	if err := env.s.processEvent(ctx, collaborator); err != nil {
		t.Fatal(err)
	}
	if got := env.knot.count(); got != before {
		t.Fatalf("a collaborator event cost %d knot requests under mark-only", got-before)
	}
	requireMarked(t, env.s.db, testRepoDid, true, "the collaborator event left no mark")
}

func TestAFailedJobKeepsItsMarkAndBacksOff(t *testing.T) {
	env := newEnsureEnv(t)
	env.serve("repo")
	env.ensure(t)
	if err := env.s.markRepo(testRepoDid); err != nil {
		t.Fatal(err)
	}

	env.knot.failing = true
	env.s.runRepoJob(context.Background(), testRepoDid)

	if attempts, marked, err := env.s.db.RepoRetryRow(testRepoDid); err != nil || !marked || attempts != 1 {
		t.Fatalf("after a failed job: attempts %d, exists %v, %v, want one failure on the row", attempts, marked, err)
	}
	if due := pendingRetriesAt(t, env, time.Now()); len(due) != 0 {
		t.Fatalf("a failed job is due again at once: %v", due)
	}
}

func pendingRetriesAt(t *testing.T, env *ensureEnv, at time.Time) []syntax.DID {
	t.Helper()
	due, err := env.s.db.DueRepoRetries(at)
	if err != nil {
		t.Fatal(err)
	}
	return due
}

func TestAJobForADidWithNoKnotDropsItsMark(t *testing.T) {
	env, did := noKnotEnv(t)

	ctx := withMarkOnly(context.Background())
	if err := env.s.processEvent(ctx, markEvent(did, env.spindleDid(), testOwner)); err != nil {
		t.Fatal(err)
	}
	requireMarked(t, env.s.db, did, true, "setup: no mark")

	env.s.runRepoJob(context.Background(), did)
	requireMarked(t, env.s.db, did, false, "a did with no knot kept its mark")
}

func TestRetriesAfterOneOutageDontAllFallDueInTheSameSecond(t *testing.T) {
	d, _ := newTestSpindleDB(t)
	now := time.Unix(100000, 0)
	r := &repoRetries{now: func() time.Time { return now }}
	for i := range 60 {
		r.fail(d, syntax.DID("did:plc:out"+string(rune('a'+i%26))+string(rune('a'+i/26))))
	}

	rows, err := d.Query(`select due from repo_retries`)
	if err != nil {
		t.Fatal(err)
	}
	defer rows.Close()
	seconds := map[int64]bool{}
	for rows.Next() {
		var due int64
		if err := rows.Scan(&due); err != nil {
			t.Fatal(err)
		}
		if lo, hi := now.Add(repoRetryBase*3/4).Unix(), now.Add(repoRetryBase).Unix(); due < lo || due > hi {
			t.Fatalf("due %d is outside the jitter window %d to %d", due, lo, hi)
		}
		seconds[due] = true
	}
	if len(seconds) < 4 {
		t.Fatalf("60 retries share %d due seconds, want them spread out", len(seconds))
	}
}

func TestAPanickingRepoJobLeavesItsWorkerRunning(t *testing.T) {
	var jobs repoJobs
	var log jobLog
	poison, fine := syntax.DID("did:plc:poison"), syntax.DID("did:plc:fine")

	runWorkers(t, &jobs, 1, func(_ context.Context, did syntax.DID) {
		log.started(did)
		if did == poison {
			panic("boom")
		}
	})

	jobs.kick(poison)
	jobs.kick(fine)
	eventually(t, 5*time.Second, "the only worker to run the next repo after the panic", func() bool { return log.count(fine) == 1 })

	jobs.kick(poison)
	eventually(t, 5*time.Second, "the poisoned repo to be kickable again", func() bool { return log.count(poison) == 2 })
}

func TestTheTickKicksEveryRepoWithWorkDueOnceAndSkipsOnesBackingOff(t *testing.T) {
	env := newEnsureEnv(t)
	now := time.Now()
	withMarks, backingOff := syntax.DID("did:plc:withmarks"), syntax.DID("did:plc:backingoff")
	for _, did := range []syntax.DID{withMarks, backingOff} {
		for _, rkey := range []string{"k1", "k2"} {
			r := db.RecordRetry{RepoDid: did, Collection: "org.tangled.track.ticket", Rkey: rkey}
			if err := env.s.db.MarkRecordDirty(r, now, 100); err != nil {
				t.Fatal(err)
			}
		}
	}
	if err := env.s.db.PutRepoRetry(backingOff, 1, now.Add(time.Hour)); err != nil {
		t.Fatal(err)
	}

	env.s.kickDueRepos()

	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	did, ok := env.s.retries.jobs.next(ctx)
	if !ok || did != withMarks {
		t.Fatalf("first kicked repo = %q, %v, want %s", did, ok, withMarks)
	}
	short, stop := context.WithTimeout(context.Background(), 200*time.Millisecond)
	defer stop()
	if did, ok := env.s.retries.jobs.next(short); ok {
		t.Fatalf("a second repo %q was kicked, want only the one with marks that isn't backing off", did)
	}
}