package spindle

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"net"
	"net/http"
	"net/http/httptest"
	"os"
	"path/filepath"
	"slices"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

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

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

// a short dir, because unix socket paths are capped at ~104 bytes on darwin
// and t.TempDir() alone gets close to that
func shortTempDir(t *testing.T) string {
	t.Helper()
	dir, err := os.MkdirTemp("/tmp", "bf-")
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { os.RemoveAll(dir) })
	return dir
}

// fakeHydrant serves the parts of hydrant's api the backfill uses
type fakeHydrant struct {
	mu       sync.Mutex
	events   []tapc.Event
	puts     [][]syntax.DID
	erased   []syntax.DID
	tracked  map[syntax.DID]bool
	repos    map[syntax.DID]hydrantRepo
	resynced []syntax.DID
	onResync func(did syntax.DID)
	onTrack  func(did syntax.DID)
	// owners hydrant's retry worker requeues every other time spindle looks
	flapping map[syntax.DID]bool
	listings int
	cursors  []string
	starts   int
	failNext error
	// every start fails, not just the next
	failStarts error
	ops        []string

	srv       *httptest.Server
	done      chan struct{}
	shutdowns atomic.Int32
}

func (f *fakeHydrant) subscriptions() int {
	f.mu.Lock()
	defer f.mu.Unlock()
	return len(f.cursors)
}

func (f *fakeHydrant) Done() <-chan struct{}        { return f.done }
func (f *fakeHydrant) Err() error                   { return nil }
func (f *fakeHydrant) Shutdown(time.Duration) error { f.srv.Close(); f.shutdowns.Add(1); return nil }

func (f *fakeHydrant) addRecord(id int64, did, rkey string) {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.events = append(f.events, tapc.Event{
		ID:   id,
		Type: tapc.EvtRecord,
		Record: &tapc.RecordEventData{
			Did:        syntax.DID(did),
			Collection: syntax.NSID(org_tangled.RepoDeclarationNSID),
			Rkey:       syntax.RecordKey(rkey),
			Action:     tapc.RecordCreateAction,
			Record:     json.RawMessage(`{}`),
		},
	})
}

func (f *fakeHydrant) addEvent(rec tapc.RecordEventData) {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.events = append(f.events, tapc.Event{ID: int64(len(f.events)), Type: tapc.EvtRecord, Record: &rec})
}

func (f *fakeHydrant) erasedDids() []syntax.DID {
	f.mu.Lock()
	defer f.mu.Unlock()
	return slices.Clone(f.erased)
}

// empty for an owner hydrant doesn't know
func (f *fakeHydrant) repoStatus(did syntax.DID) string {
	f.mu.Lock()
	defer f.mu.Unlock()
	return f.repos[did].Status
}

func (f *fakeHydrant) failStartsWith(err error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.failStarts = err
}

// identity events aren't replayed, so only live connections get this one
func (f *fakeHydrant) addLiveOnly(id int64, did string) {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.events = append(f.events, tapc.Event{
		ID:       id,
		Type:     tapc.EvtIdentity,
		Identity: &tapc.IdentityEventData{DID: syntax.DID(did), IsActive: true},
	})
}

// setRepo is what hydrant reports for repo.Did, which it then already
// tracks. owners nobody set are synced
func (f *fakeHydrant) setRepo(repo hydrantRepo) {
	f.mu.Lock()
	defer f.mu.Unlock()
	if f.repos == nil {
		f.repos = make(map[syntax.DID]hydrantRepo)
		f.tracked = make(map[syntax.DID]bool)
	}
	f.repos[repo.Did] = repo
	f.tracked[repo.Did] = true
}

func synced(did syntax.DID) hydrantRepo { return hydrantRepo{Did: did, Status: "synced"} }

// retrying is a failed owner hydrant queued again
func retrying(did syntax.DID) hydrantRepo {
	return hydrantRepo{Did: did, Status: "error(transport error)", Pending: true}
}

func failedWith(did syntax.DID, kind string, hostFailures int) hydrantRepo {
	return hydrantRepo{
		Did:    did,
		Status: "error(" + kind + ")",
		Resync: &hydrantResync{State: "error", Kind: kind, HostFailures: hostFailures},
	}
}

func (f *fakeHydrant) start(settings map[string]string) (embeddedHydrant, error) {
	f.mu.Lock()
	f.starts++
	err := f.failNext
	f.failNext = nil
	if err == nil {
		err = f.failStarts
	}
	f.mu.Unlock()
	if err != nil {
		return nil, err
	}

	l, err := net.Listen("unix", strings.TrimPrefix(settings["HYDRANT_API_BIND"], "unix:"))
	if err != nil {
		return nil, err
	}
	f.srv = httptest.NewUnstartedServer(http.HandlerFunc(f.serve))
	f.srv.Listener = l
	f.srv.Start()
	f.done = make(chan struct{})
	return f, nil
}

func (f *fakeHydrant) serve(w http.ResponseWriter, r *http.Request) {
	switch {
	case r.URL.Path == "/stream/head":
		f.mu.Lock()
		var head *int64
		for _, evt := range f.events {
			if evt.Type == tapc.EvtRecord {
				head = &evt.ID
			}
		}
		f.mu.Unlock()
		json.NewEncoder(w).Encode(map[string]any{"id": head})
	case strings.HasPrefix(r.URL.Path, "/redactions/"):
		did := syntax.DID(strings.TrimPrefix(r.URL.Path, "/redactions/"))
		f.mu.Lock()
		f.erased = append(f.erased, did)
		f.ops = append(f.ops, "erase "+did.String())
		delete(f.tracked, did)
		delete(f.repos, did)
		// the event stays so its id still counts, only the body goes
		for i, evt := range f.events {
			if evt.Record != nil && evt.Record.Did == did {
				f.events[i].Record = nil
			}
		}
		f.mu.Unlock()
		w.Write([]byte("{}"))
	case r.URL.Path == "/repos" && r.Method == http.MethodGet:
		f.mu.Lock()
		var repos []hydrantRepo
		f.listings++
		for did := range f.tracked {
			repo, ok := f.repos[did]
			if !ok {
				repo = hydrantRepo{Did: did, Status: "synced"}
			}
			if f.flapping[did] && f.listings%2 == 0 {
				repo.Resync, repo.Pending = nil, true
			}
			if r.URL.Query().Get("queued") == "true" && !repo.Pending && repo.Resync == nil {
				continue
			}
			repo.Tracked = true
			repos = append(repos, repo)
		}
		f.mu.Unlock()
		slices.SortFunc(repos, func(a, b hydrantRepo) int { return strings.Compare(string(a.Did), string(b.Did)) })
		if cursor := syntax.DID(r.URL.Query().Get("cursor")); cursor != "" {
			repos = slices.DeleteFunc(repos, func(repo hydrantRepo) bool { return repo.Did <= cursor })
		}
		if limit, err := strconv.Atoi(r.URL.Query().Get("limit")); err == nil && len(repos) > limit {
			repos = repos[:limit]
		}
		json.NewEncoder(w).Encode(repos)
	case r.URL.Path == "/repos/resync":
		var body []struct {
			Did syntax.DID `json:"did"`
		}
		if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
			http.Error(w, err.Error(), http.StatusBadRequest)
			return
		}
		f.mu.Lock()
		onResync := f.onResync
		for _, b := range body {
			f.resynced = append(f.resynced, b.Did)
		}
		f.mu.Unlock()
		for _, b := range body {
			if onResync != nil {
				onResync(b.Did)
			}
		}
		w.Write([]byte("[]"))
	case r.URL.Path == "/repos":
		var body []struct {
			Did syntax.DID `json:"did"`
		}
		if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
			http.Error(w, err.Error(), http.StatusBadRequest)
			return
		}
		var dids []syntax.DID
		for _, b := range body {
			dids = append(dids, b.Did)
		}
		slices.Sort(dids)
		f.mu.Lock()
		if f.tracked == nil {
			f.tracked = make(map[syntax.DID]bool)
		}
		f.puts = append(f.puts, dids)
		f.ops = append(f.ops, fmt.Sprintf("put %d", len(dids)))
		for _, did := range dids {
			f.tracked[did] = true
		}
		onTrack := f.onTrack
		f.mu.Unlock()
		for _, did := range dids {
			if onTrack != nil {
				onTrack(did)
			}
		}
		w.Write([]byte("[]"))
	case r.URL.Path == "/stream":
		f.stream(w, r)
	default:
		http.NotFound(w, r)
	}
}

func (f *fakeHydrant) stream(w http.ResponseWriter, r *http.Request) {
	conn, err := (&websocket.Upgrader{}).Upgrade(w, r, nil)
	if err != nil {
		return
	}
	defer conn.Close()
	next, _ := strconv.ParseInt(r.URL.Query().Get("cursor"), 10, 64)
	f.mu.Lock()
	f.cursors = append(f.cursors, r.URL.Query().Get("cursor"))
	connectedAt := len(f.events)
	f.mu.Unlock()
	gone := closedWhenDone(conn)
	for {
		f.mu.Lock()
		var events []tapc.Event
		for i, evt := range f.events {
			if evt.ID >= next && (evt.Type == tapc.EvtRecord || i >= connectedAt) {
				events = append(events, evt)
			}
		}
		f.mu.Unlock()
		for _, evt := range events {
			if err := conn.WriteJSON(evt); err != nil {
				return
			}
			next = evt.ID + 1
		}
		select {
		case <-gone:
			return
		case <-time.After(5 * time.Millisecond):
		}
	}
}

// the server side of a websocket only learns the client left by reading
func closedWhenDone(conn *websocket.Conn) <-chan struct{} {
	gone := make(chan struct{})
	go func() {
		defer close(gone)
		for {
			if _, _, err := conn.ReadMessage(); err != nil {
				return
			}
		}
	}()
	return gone
}

func newTestBackfill(t *testing.T, f *fakeHydrant) (*backfill, *[]int64) {
	t.Helper()
	d, _ := newTestSpindleDB(t)
	b := newBackfill(&Spindle{cfg: &config.Config{}, db: d}, slog.Default(), shortTempDir(t))
	b.start = f.start
	b.poll = 10 * time.Millisecond
	b.retryBase = time.Millisecond
	var mu sync.Mutex
	handled := []int64{}
	b.handle = func(_ context.Context, evt tapc.Event) error {
		mu.Lock()
		defer mu.Unlock()
		handled = append(handled, evt.ID)
		return nil
	}
	return b, &handled
}

func runRound(t *testing.T, b *backfill) {
	t.Helper()
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	if err := b.round(ctx); err != nil {
		t.Fatalf("round: %v", err)
	}
}

func TestBackfillRoundFeedsTheStreamThenShutsDown(t *testing.T) {
	f := &fakeHydrant{}
	f.addRecord(0, "did:plc:a", "one")
	f.addRecord(1, "did:plc:a", "two")
	f.addRecord(2, "did:plc:b", "three")
	b, handled := newTestBackfill(t, f)

	b.Track("did:plc:b", "did:plc:a")
	runRound(t, b)

	if !slices.Equal(*handled, []int64{0, 1, 2}) {
		t.Fatalf("handled %v, want [0 1 2]", *handled)
	}
	if len(f.puts) != 1 || !slices.Equal(f.puts[0], []syntax.DID{"did:plc:a", "did:plc:b"}) {
		t.Fatalf("puts = %v, want one put of both owners", f.puts)
	}
	if f.shutdowns.Load() != 1 {
		t.Fatalf("shutdowns = %d, want 1", f.shutdowns.Load())
	}
	if last, err := b.loadCursor(); err != nil || last != 2 {
		t.Fatalf("saved cursor = %d, %v, want 2", last, err)
	}

	// a known owner doesn't start hydrant again, and a new one resumes after the cursor
	b.Track("did:plc:a")
	runRound(t, b)
	if f.starts != 1 {
		t.Fatalf("starts = %d after tracking a known owner, want 1", f.starts)
	}
	f.addRecord(3, "did:plc:c", "four")
	b.Track("did:plc:c")
	runRound(t, b)
	if !slices.Equal(*handled, []int64{0, 1, 2, 3}) {
		t.Fatalf("handled %v, want [0 1 2 3]", *handled)
	}
	if !slices.Equal(f.cursors, []string{"0", "3"}) {
		t.Fatalf("stream cursors = %v, want [0 3]", f.cursors)
	}
}

// untracking only remembers the wish, so nothing happens until a round has
// something to add. a restart forgets it, and hydrant's copy goes then
func TestBackfillUntrackIsLazy(t *testing.T) {
	f := &fakeHydrant{}
	b, _ := newTestBackfill(t, f)
	b.Track("did:plc:a", "did:plc:b")
	runRound(t, b)

	b.Untrack("did:plc:b")
	runRound(t, b)
	if f.starts != 1 {
		t.Fatalf("starts = %d, untracking alone must not start hydrant", f.starts)
	}

	b.Track("did:plc:c")
	runRound(t, b)
	if !slices.Equal(f.erased, []syntax.DID{"did:plc:b"}) {
		t.Fatalf("erased = %v, want [did:plc:b]", f.erased)
	}
	if !slices.Equal(f.puts[len(f.puts)-1], []syntax.DID{"did:plc:c"}) {
		t.Fatalf("last put = %v, want [did:plc:c]", f.puts[len(f.puts)-1])
	}

	// nothing followed a while it was untracked, so tracking it again has to
	// delete it first, or hydrant would call it backfilled and skip it
	b.Untrack("did:plc:a")
	b.Track("did:plc:a")
	add, remove := b.take()
	if !slices.Equal(add, []syntax.DID{"did:plc:a"}) || !slices.Equal(remove, []syntax.DID{"did:plc:a"}) {
		t.Fatalf("take = %v, %v, want did:plc:a in both", add, remove)
	}

	// a new process only knows who it wants, so c goes
	b, _ = newTestBackfill(t, f)
	b.Track("did:plc:a")
	runRound(t, b)
	if !slices.Equal(f.erased, []syntax.DID{"did:plc:b", "did:plc:c"}) {
		t.Fatalf("erased = %v, want b and c", f.erased)
	}
}

func TestBackfillWaitsForOwnersItCanStillGet(t *testing.T) {
	f := &fakeHydrant{}
	f.addRecord(0, "did:plc:a", "one")
	// a first backfill still running, and a live pds asking hydrant to slow down
	f.setRepo(hydrantRepo{Did: "did:plc:a", Status: "desynchronized", Pending: true})
	f.setRepo(failedWith("did:plc:b", "ratelimited", 0))
	b, _ := newTestBackfill(t, f)
	b.Track("did:plc:a", "did:plc:b")

	go func() {
		time.Sleep(200 * time.Millisecond)
		f.setRepo(synced("did:plc:a"))
		f.setRepo(synced("did:plc:b"))
	}()
	started := time.Now()
	runRound(t, b)
	if time.Since(started) < 200*time.Millisecond {
		t.Fatalf("round ended after %v while owners could still be backfilled", time.Since(started))
	}
	if len(b.failing) != 0 {
		t.Fatalf("failing = %v, want none", b.failing)
	}
}

// hydrant keeps moving owners it failed back into its queue on its own
// backoff, so pending hardly ever reads 0. none of that can gain anything this
// round, so it mustn't keep the round going
func TestBackfillEndsWhileFailedOwnersKeepRetrying(t *testing.T) {
	f := &fakeHydrant{flapping: map[syntax.DID]bool{"did:plc:b": true}}
	f.addRecord(0, "did:plc:a", "one")
	f.setRepo(failedWith("did:plc:b", "generic", 0))
	// ratelimited only because its pds is down and hydrant skips it
	f.setRepo(failedWith("did:plc:c", "ratelimited", 4))
	f.setRepo(hydrantRepo{Did: "did:plc:d", Status: "takendown", Resync: &hydrantResync{State: "gone"}})
	// hydrant dropped it from its queues without syncing it
	f.setRepo(hydrantRepo{Did: "did:plc:e", Status: "desynchronized"})
	b, handled := newTestBackfill(t, f)
	b.Track("did:plc:a", "did:plc:b", "did:plc:c", "did:plc:d", "did:plc:e")

	runRound(t, b)

	if !slices.Equal(*handled, []int64{0}) {
		t.Fatalf("handled %v, want [0]", *handled)
	}
	// spindle didn't know they failed before, so they got retried straight away
	if !slices.Equal(f.resynced, []syntax.DID{"did:plc:b", "did:plc:c", "did:plc:e"}) {
		t.Fatalf("resynced = %v, want b, c and e", f.resynced)
	}
	if len(b.failing) != 3 || b.failing["did:plc:b"].rounds != 1 || b.failing["did:plc:e"].rounds != 1 {
		t.Fatalf("failing = %v, want b, c and e after one round", b.failing)
	}
	if wait := time.Until(b.nextRetry()); wait < time.Minute || wait > failedOwnerRetryMin {
		t.Fatalf("next retry in %v, want about %v", wait, failedOwnerRetryMin)
	}
}

// the first round leaves did:plc:a failed, with its retry a backoff away
func failedOnce(t *testing.T) (*fakeHydrant, *backfill) {
	t.Helper()
	f := &fakeHydrant{}
	f.setRepo(failedWith("did:plc:a", "transport", 0))
	b, _ := newTestBackfill(t, f)
	b.Track("did:plc:a", "did:plc:b")
	runRound(t, b)
	return f, b
}

// stands in for waiting out the backoff
func retryNow(b *backfill) {
	for did, f := range b.failing {
		f.retryAt = time.Now()
		b.failing[did] = f
	}
}

func TestBackfillRetriesAFailedOwnerAfterABackoff(t *testing.T) {
	f, b := failedOnce(t)
	if wait := time.Until(b.nextRetry()); wait < time.Minute || wait > failedOwnerRetryMin {
		t.Fatalf("next retry in %v, want about %v", wait, failedOwnerRetryMin)
	}

	runRound(t, b)
	if f.starts != 1 {
		t.Fatalf("starts = %d before the retry was due, want 1", f.starts)
	}

	retryNow(b)
	before := len(f.resynced)
	f.onResync = func(did syntax.DID) { f.setRepo(synced(did)) }
	runRound(t, b)
	if got := f.resynced[before:]; !slices.Equal(got, []syntax.DID{"did:plc:a"}) {
		t.Fatalf("resynced %v in the retry round, want did:plc:a", got)
	}
}

func TestBackfillBacksOffFurtherWhenTheRetryFails(t *testing.T) {
	f, b := failedOnce(t)
	retryNow(b)
	// hydrant shows the retry queued first, so the old failure can't count
	// as the new outcome
	f.onResync = func(did syntax.DID) {
		f.setRepo(retrying(did))
		go func() {
			time.Sleep(50 * time.Millisecond)
			f.setRepo(failedWith(did, "transport", 0))
		}()
	}
	started := time.Now()
	runRound(t, b)
	if time.Since(started) < 50*time.Millisecond {
		t.Fatalf("round ended after %v before the retry finished", time.Since(started))
	}
	if wait := time.Until(b.nextRetry()); wait <= failedOwnerRetryMin {
		t.Fatalf("next retry in %v, want it backed off past %v", wait, failedOwnerRetryMin)
	}
}

func TestBackfillForgetsAFailedOwnerOnceItBackfills(t *testing.T) {
	f, b := failedOnce(t)
	retryNow(b)
	f.onResync = func(did syntax.DID) { f.setRepo(synced(did)) }
	runRound(t, b)
	if !b.nextRetry().IsZero() {
		t.Fatalf("next retry at %v after the owner backfilled, want none", b.nextRetry())
	}
}

// with no new owners coming, run itself has to come back for a failed one
func TestBackfillRunComesBackForFailedOwners(t *testing.T) {
	f := &fakeHydrant{}
	f.setRepo(failedWith("did:plc:a", "transport", 0))
	f.onResync = func(did syntax.DID) {
		f.mu.Lock()
		retries := len(f.resynced)
		f.mu.Unlock()
		// the first round's retry fails again, and the next one gets through
		if retries > 1 {
			f.setRepo(synced(did))
		}
	}
	b, _ := newTestBackfill(t, f)
	b.retryFailed = 50 * time.Millisecond
	b.Track("did:plc:a")

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	done := make(chan struct{})
	go func() {
		defer close(done)
		b.run(ctx)
	}()
	for {
		f.mu.Lock()
		retries := len(f.resynced)
		f.mu.Unlock()
		if retries >= 2 {
			break
		}
		if ctx.Err() != nil {
			t.Fatal("the failed owner was never retried")
		}
		time.Sleep(10 * time.Millisecond)
	}
	cancel()
	<-done
	if f.starts < 2 {
		t.Fatalf("starts = %d, want a retry round", f.starts)
	}
}

func TestBackfillParksARecordThatKeepsFailing(t *testing.T) {
	f := &fakeHydrant{}
	f.addRecord(0, "did:plc:a", "one")
	f.addRecord(1, "did:plc:a", "bad")
	f.addRecord(2, "did:plc:a", "three")
	b, _ := newTestBackfill(t, f)
	attempts := map[int64]int{}
	b.handle = func(_ context.Context, evt tapc.Event) error {
		attempts[evt.ID]++
		if evt.ID == 1 {
			return errors.New("database is locked")
		}
		return nil
	}

	b.Track("did:plc:a")
	runRound(t, b)

	want := map[int64]int{0: 1, 1: backfillEventAttempts, 2: 1}
	if fmt.Sprint(attempts) != fmt.Sprint(want) {
		t.Fatalf("attempts = %v, want %v", attempts, want)
	}
	if last, err := b.loadCursor(); err != nil || last != 2 {
		t.Fatalf("saved cursor = %d, %v, want 2", last, err)
	}
	parked := dueParked(t, b)
	if len(parked) != 1 || parked[0].URI != "at://did:plc:a/"+org_tangled.RepoDeclarationNSID+"/bad" || parked[0].Err != "database is locked" {
		t.Fatalf("parked = %+v, want the bad record with its error", parked)
	}
}

// looks a day ahead, so records not due yet come back too
func dueParked(t *testing.T, b *backfill) []db.ParkedRecord {
	t.Helper()
	parked, err := b.s.db.DueParkedRecords(time.Now().Add(24 * time.Hour))
	if err != nil {
		t.Fatal(err)
	}
	return parked
}

func parkFailing(t *testing.T, b *backfill, f *fakeHydrant) {
	t.Helper()
	f.addRecord(0, "did:plc:a", "bad")
	handle := b.handle
	b.handle = func(context.Context, tapc.Event) error { return errors.New("database is locked") }
	b.Track("did:plc:a")
	runRound(t, b)
	b.handle = handle
	if len(dueParked(t, b)) != 1 {
		t.Fatal("nothing got parked")
	}
}

func TestBackfillRetriesAParkedRecordUntilItGoesThrough(t *testing.T) {
	f := &fakeHydrant{}
	b, handled := newTestBackfill(t, f)
	parkFailing(t, b, f)

	b.retryParkedRecords(context.Background(), time.Now())
	if len(*handled) != 0 {
		t.Fatalf("handled %v before the retry was due", *handled)
	}

	fails := 2
	handle := b.handle
	b.handle = func(ctx context.Context, evt tapc.Event) error {
		if fails > 0 {
			fails--
			return errors.New("still locked")
		}
		return handle(ctx, evt)
	}
	for range 3 {
		b.retryParkedRecords(context.Background(), time.Now().Add(24*time.Hour))
	}
	if !slices.Equal(*handled, []int64{0}) {
		t.Fatalf("handled %v, want the parked record once it worked", *handled)
	}
	if parked := dueParked(t, b); len(parked) != 0 {
		t.Fatalf("parked = %+v after it went through, want none", parked)
	}
}

func TestBackfillGivesUpOnAParkedRecordAfterTheLastRetry(t *testing.T) {
	f := &fakeHydrant{}
	b, _ := newTestBackfill(t, f)
	b.retryParked = time.Minute
	parkFailing(t, b, f)
	b.handle = func(context.Context, tapc.Event) error { return errors.New("broken for good") }

	var waits []time.Duration
	for range parkedRetries - 1 {
		b.retryParkedRecords(context.Background(), time.Now().Add(24*time.Hour))
		parked := dueParked(t, b)
		if len(parked) != 1 {
			t.Fatalf("parked = %+v before the last retry", parked)
		}
		waits = append(waits, time.Until(parked[0].RetryAt).Round(time.Minute))
	}
	want := []time.Duration{2 * time.Minute, 4 * time.Minute, 8 * time.Minute, 16 * time.Minute, 32 * time.Minute}
	if !slices.Equal(waits, want) {
		t.Fatalf("waits = %v, want %v", waits, want)
	}
	b.retryParkedRecords(context.Background(), time.Now().Add(24*time.Hour))
	if parked := dueParked(t, b); len(parked) != 0 {
		t.Fatalf("parked = %+v after the last retry, want it given up", parked)
	}
}

func TestBackfillRetriesEveryParkedRecordAtStartup(t *testing.T) {
	f := &fakeHydrant{}
	b, handled := newTestBackfill(t, f)
	b.retryParked = time.Hour
	parkFailing(t, b, f)

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	done := make(chan struct{})
	go func() {
		defer close(done)
		b.run(ctx)
	}()
	for len(dueParked(t, b)) != 0 {
		if ctx.Err() != nil {
			t.Fatal("the parked record wasn't retried at startup")
		}
		time.Sleep(10 * time.Millisecond)
	}
	cancel()
	<-done
	if !slices.Equal(*handled, []int64{0}) {
		t.Fatalf("handled %v, want the parked record", *handled)
	}
}

func TestBackfillReplaysTheStreamOfANewDb(t *testing.T) {
	f := &fakeHydrant{}
	f.addRecord(0, "did:plc:a", "one")
	b, handled := newTestBackfill(t, f)
	if err := b.saveCursor(100); err != nil {
		t.Fatal(err)
	}
	b.Track("did:plc:a")
	runRound(t, b)
	if !slices.Equal(*handled, []int64{0}) {
		t.Fatalf("handled %v, want the new db's event", *handled)
	}
}

// head only counts records, so identity ids can't move the cursor
func TestBackfillCursorSkipsLiveOnlyEvents(t *testing.T) {
	f := &fakeHydrant{}
	f.addRecord(0, "did:plc:a", "one")
	f.setRepo(hydrantRepo{Did: "did:plc:a", Status: "desynchronized", Pending: true})
	b, handled := newTestBackfill(t, f)
	b.Track("did:plc:a")
	go func() {
		for f.subscriptions() == 0 {
			time.Sleep(5 * time.Millisecond)
		}
		f.addLiveOnly(1, "did:plc:a")
		time.Sleep(50 * time.Millisecond)
		f.setRepo(synced("did:plc:a"))
	}()
	runRound(t, b)
	if last, err := b.loadCursor(); err != nil || last != 0 {
		t.Fatalf("saved cursor = %d, %v, want the last record's id", last, err)
	}

	f.addRecord(2, "did:plc:b", "two")
	b.Track("did:plc:b")
	runRound(t, b)
	if !slices.Equal(*handled, []int64{0, 2}) {
		t.Fatalf("handled %v, want the new record without replaying the old one", *handled)
	}
}

func TestBackfillTracksOwnersInBatches(t *testing.T) {
	f := &fakeHydrant{}
	b, _ := newTestBackfill(t, f)
	var dids []syntax.DID
	for i := range 2*hydrantDidBatch + 1 {
		dids = append(dids, syntax.DID(fmt.Sprintf("did:plc:%024d", i)))
	}
	b.Track(dids...)
	runRound(t, b)

	var sizes []int
	for _, put := range f.puts {
		sizes = append(sizes, len(put))
	}
	if !slices.Equal(sizes, []int{hydrantDidBatch, hydrantDidBatch, 1}) {
		t.Fatalf("put sizes = %v, want two full batches and one more", sizes)
	}
	h, err := f.start(b.settings())
	if err != nil {
		t.Fatal(err)
	}
	defer h.Shutdown(0)
	repos, err := newHydrantAPI(b.socketPath()).trackedRepos(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	if len(repos) != len(dids) {
		t.Fatalf("listed %d owners over every page, want %d", len(repos), len(dids))
	}
}

func TestBackfillTakesOwnersAddedDuringARound(t *testing.T) {
	f := &fakeHydrant{}
	f.addRecord(0, "did:plc:a", "one")
	f.setRepo(hydrantRepo{Did: "did:plc:a", Status: "desynchronized", Pending: true})
	b, handled := newTestBackfill(t, f)
	b.Track("did:plc:a")

	go func() {
		for f.subscriptions() == 0 {
			time.Sleep(5 * time.Millisecond)
		}
		b.Track("did:plc:b")
		time.Sleep(50 * time.Millisecond)
		f.addRecord(1, "did:plc:b", "two")
		f.setRepo(synced("did:plc:a"))
	}()
	runRound(t, b)
	if !slices.Equal(*handled, []int64{0, 1}) {
		t.Fatalf("handled %v, want the record of the owner that joined", *handled)
	}
	if f.starts != 1 {
		t.Fatalf("starts = %d, want the owner taken by the running round", f.starts)
	}
}

func TestBackfillErasesAnOwnerBeforeTrackingItAgain(t *testing.T) {
	f := &fakeHydrant{}
	f.addRecord(0, "did:plc:a", "one")
	b, handled := newTestBackfill(t, f)
	b.Track("did:plc:a")
	runRound(t, b)

	b.Untrack("did:plc:a")
	b.Track("did:plc:a")
	f.onTrack = func(did syntax.DID) { f.addRecord(1, did.String(), "one") }
	// a fresh cursor replays the erased event, which has no body left
	os.Remove(b.cursorPath())
	runRound(t, b)
	if !slices.Equal(f.ops, []string{"put 1", "erase did:plc:a", "put 1"}) {
		t.Fatalf("ops = %v, want the erase before the second put", f.ops)
	}
	if !slices.Equal(*handled, []int64{0, 1}) {
		t.Fatalf("handled %v, want the first round's record and the fresh one", *handled)
	}
}

func TestBackfillCatchUpErasesAndTracksOwnersAndRepoDids(t *testing.T) {
	f := &fakeHydrant{}
	b, _ := newTestBackfill(t, f)
	b.s.l = slog.Default()
	b.refresh = func(context.Context, syntax.DID) error { return nil }
	if err := b.s.db.AddRepo(db.Repo{Knot: "knot.test", Owner: "did:plc:owner", Rkey: "repo", RepoDid: "did:plc:repo"}); err != nil {
		t.Fatal(err)
	}
	b.Track("did:plc:owner")
	runRound(t, b)

	if err := b.CatchUp(); err != nil {
		t.Fatal(err)
	}
	runRound(t, b)
	want := []string{"put 1", "erase did:plc:owner", "erase did:plc:repo", "put 2"}
	if !slices.Equal(f.ops, want) {
		t.Fatalf("ops = %v, want %v", f.ops, want)
	}
}

func TestHydrantReadinessNoticesHydrantExit(t *testing.T) {
	socket := shortTempDir(t) + "/api.sock"
	l, err := net.Listen("unix", socket)
	if err != nil {
		t.Fatal(err)
	}
	srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { <-r.Context().Done() }))
	srv.Listener = l
	srv.Start()
	defer srv.Close()
	h := &fakeHydrant{done: make(chan struct{})}
	close(h.done)

	started := time.Now()
	err = newHydrantAPI(socket).waitReady(context.Background(), h)
	if err == nil || time.Since(started) > 10*time.Second {
		t.Fatalf("waitReady = %v after %v, want it to notice hydrant exited", err, time.Since(started))
	}
}

func TestBackfillRequeuesOwnersWhenHydrantFailsToStart(t *testing.T) {
	f := &fakeHydrant{failNext: errors.New("db is locked")}
	b, _ := newTestBackfill(t, f)
	b.Track("did:plc:a")

	if err := b.round(context.Background()); err == nil {
		t.Fatal("round succeeded although hydrant didn't start")
	}
	runRound(t, b)
	if len(f.puts) != 1 || !slices.Equal(f.puts[0], []syntax.DID{"did:plc:a"}) {
		t.Fatalf("puts = %v, the owner should have been retried", f.puts)
	}
}

func TestBackfillSettingsAllowPrivateNetworksOnlyInDev(t *testing.T) {
	for _, dev := range []bool{false, true} {
		cfg := &config.Config{}
		cfg.Server.Dev = dev
		b := newBackfill(&Spindle{cfg: cfg}, slog.Default(), shortTempDir(t))
		got, ok := b.settings()["HYDRANT_ALLOW_PRIVATE_NETWORKS"]
		if ok != dev || (dev && got != "true") {
			t.Errorf("dev=%v: HYDRANT_ALLOW_PRIVATE_NETWORKS = %q (set: %v)", dev, got, ok)
		}
	}
}

// real hydrant, offline. nothing resolves so nothing backfills, but it has to
// start, answer, shut down and let go of its db for the next round
func TestBackfillRunsEmbeddedHydrant(t *testing.T) {
	cfg := &config.Config{}
	cfg.Server.PlcUrl = "http://127.0.0.1:1"
	d, _ := newTestSpindleDB(t)
	b := newBackfill(&Spindle{cfg: cfg, db: d}, slog.Default(), shortTempDir(t))
	b.poll = 100 * time.Millisecond
	b.handle = func(context.Context, tapc.Event) error { return nil }

	round := func() {
		t.Helper()
		ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
		defer cancel()
		if err := b.round(ctx); err != nil {
			t.Fatalf("round: %v", err)
		}
	}
	b.Track("did:plc:aaaaaaaaaaaaaaaaaaaaaaaa")
	round()
	if _, err := os.Stat(filepath.Join(b.dir, "db")); err != nil {
		t.Fatalf("hydrant didn't put its db under the backfill dir: %v", err)
	}
	// the second start has to reopen the db, and erase the first owner through hydrant's api
	b.Untrack("did:plc:aaaaaaaaaaaaaaaaaaaaaaaa")
	b.Track("did:plc:bbbbbbbbbbbbbbbbbbbbbbbb")
	round()
}