package spindle

import (
	"context"
	"testing"
	"time"

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

func ticketEvent(rkey string) tapc.Event {
	return recordEvent(testRepoDid, org_tangled.TrackTicketNSID, rkey, tapc.RecordCreateAction, map[string]string{})
}

func dueRecords(t *testing.T, env *ensureEnv) []db.RecordRetry {
	t.Helper()
	due, err := env.s.db.DueRecordRetries(time.Now())
	if err != nil {
		t.Fatal(err)
	}
	return due
}

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

	before := env.knot.count()
	if err := env.s.processEvent(ctx, ticketEvent("3kticket00001")); err != nil {
		t.Fatal(err)
	}
	if got := env.knot.count(); got != before {
		t.Fatalf("marking a pull record cost %d knot requests", got-before)
	}
	if got := dueRecords(t, env); len(got) != 1 || got[0].Rkey != "3kticket00001" {
		t.Fatalf("record marks = %+v, want the ticket", got)
	}

	env.s.runRepoJob(context.Background(), testRepoDid)
	if env.knot.count() == before {
		t.Fatal("the job never read the ticket from the knot")
	}
	if got := dueRecords(t, env); len(got) != 0 {
		t.Fatalf("a finished job left record marks %+v", got)
	}
}

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

	if err := env.s.processEvent(ctx, markEvent(testRepoDid, env.spindleDid(), testOwner)); err != nil {
		t.Fatal(err)
	}
	if err := env.s.processEvent(ctx, ticketEvent("3kticket00001")); err != nil {
		t.Fatal(err)
	}
	if got := dueRecords(t, env); len(got) != 1 {
		t.Fatalf("a record for a repo that is about to be served was dropped, marks = %+v", got)
	}

	env.s.runRepoJob(context.Background(), testRepoDid)
	if rows := env.rows(t); len(rows) != 1 {
		t.Fatalf("the job didn't serve the repo, rows = %+v", rows)
	}
	if got := dueRecords(t, env); len(got) != 0 {
		t.Fatalf("the record mark outlived the repo's job: %+v", got)
	}
}

func TestAPullRecordForAnUnknownRepoLeavesNoMark(t *testing.T) {
	env := newEnsureEnv(t)
	if err := env.s.processEvent(withMarkOnly(context.Background()), ticketEvent("3kticket00001")); err != nil {
		t.Fatal(err)
	}
	if got := dueRecords(t, env); len(got) != 0 {
		t.Fatalf("an unknown repo's record left marks %+v", got)
	}
}

func TestAFailingRecordIsBackedOffInsteadOfSpinningTheJob(t *testing.T) {
	env := newEnsureEnv(t)
	env.serve("repo")
	env.ensure(t)
	rec := ticketEvent("3kticket00001").Record
	if err := env.s.markRecord(rec); err != nil {
		t.Fatal(err)
	}

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

	r := db.RecordRetry{RepoDid: testRepoDid, Collection: org_tangled.TrackTicketNSID, Rkey: "3kticket00001"}
	if attempts, err := env.s.db.RecordRetryAttempts(r); err != nil || attempts != 1 {
		t.Fatalf("attempts = %d, %v, want one failure on the mark", attempts, err)
	}
	if got := dueRecords(t, env); len(got) != 0 {
		t.Fatalf("a failed record is due again at once: %+v", got)
	}
}

func TestAFloodOfRecordsForAnotherRepoLeavesAServedReposMarkAlone(t *testing.T) {
	env := newEnsureEnv(t)
	env.serve("repo")
	env.ensure(t)
	if _, err := env.s.db.Exec(`insert into record_retries (repo_did, collection, rkey, attempts, due, epoch)
		with recursive n(i) as (select 0 union all select i + 1 from n where i < 30000)
		select 'did:plc:flood' || (i % 40), 'c', 'k' || i, 0, 0, 1 from n`); err != nil {
		t.Fatal(err)
	}

	if err := env.s.processEvent(withMarkOnly(context.Background()), ticketEvent("3kticket00001")); err != nil {
		t.Fatal(err)
	}
	if marks, err := env.s.db.DueRecordMarksOf(testRepoDid, time.Now()); err != nil || len(marks) != 1 {
		t.Fatalf("marks = %+v, %v, want the served repo's event kept while other repos fill the table", marks, err)
	}
}