package spindle

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

	"github.com/bluesky-social/indigo/atproto/syntax"
	"github.com/gorilla/websocket"
	hydrant "tangled.org/ptr.pet/hydrant/ffi/go"

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

const (
	backfillPollInterval   = 3 * time.Second
	backfillStableChecks   = 2
	backfillEventAttempts  = 3
	backfillRetryDelay     = time.Minute
	hydrantShutdownTimeout = 30 * time.Second

	// owners whose backfill failed get another round after this, doubling
	// each time it fails again. the cap is what dead accounts cost forever
	failedOwnerRetryMin = 2 * time.Minute
	failedOwnerRetryMax = time.Hour

	// six doublings from a minute add up to about an hour before giving up
	parkedRetryMin = time.Minute
	parkedRetries  = 6

	pendingCatchUpMaxAge = 7 * 24 * time.Hour

	hydrantRequestTimeout = 30 * time.Second
	hydrantEraseTimeout   = 10 * time.Minute
	// hydrant caps a request body at 8 MiB, which is about 195k dids
	hydrantDidBatch = 1000
)

// jetstream has the live changes, so these are only the gaps it missed
var backfillCollections = []string{
	org_tangled.RepoDeclarationNSID,
	org_tangled.RepoManifestNSID,
	org_tangled.GraphCollaboratorNSID,
}

type embeddedHydrant interface {
	Done() <-chan struct{}
	Err() error
	Shutdown(timeout time.Duration) error
}

func startHydrant(settings map[string]string) (embeddedHydrant, error) {
	return hydrant.Start(settings)
}

// backfill runs an embedded hydrant while owners have records to backfill and
// shuts it down once spindle has processed all of them, so its memory only
// gets used when there is work
type backfill struct {
	s      *Spindle
	l      *slog.Logger
	dir    string
	start  func(settings map[string]string) (embeddedHydrant, error)
	handle func(ctx context.Context, evt tapc.Event) error
	// a manifest deleted during a gap never streams, so only the knot shows it
	refresh func(ctx context.Context, repoDid syntax.DID) error
	wake    chan struct{}

	poll        time.Duration
	retryBase   time.Duration
	retryFailed time.Duration
	retryParked time.Duration

	mu      sync.Mutex
	tracked map[syntax.DID]struct{}
	add     map[syntax.DID]struct{}
	remove  map[syntax.DID]struct{}

	// owners hydrant couldn't backfill. hydrant only retries them while it
	// runs, so without this they'd wait for a new owner or a restart. only
	// rounds touch it, and those never overlap
	failing map[syntax.DID]failure

	// id of the last stream event spindle finished with, -1 before the first
	last atomic.Int64

	// a gap during catch-up only sets catchUpAgain, the running loop goes once more
	catchingUp   bool
	catchUpAgain bool
	catchUps     sync.WaitGroup
	closed       bool
}

func newBackfill(s *Spindle, l *slog.Logger, dir string) *backfill {
	return &backfill{
		s:           s,
		l:           l,
		dir:         dir,
		start:       startHydrant,
		handle:      s.markBackfilled,
		refresh:     func(_ context.Context, repoDid syntax.DID) error { return s.markRepo(repoDid) },
		wake:        make(chan struct{}, 1),
		poll:        backfillPollInterval,
		retryBase:   time.Second,
		retryFailed: failedOwnerRetryMin,
		retryParked: parkedRetryMin,
		tracked:     make(map[syntax.DID]struct{}),
		add:         make(map[syntax.DID]struct{}),
		remove:      make(map[syntax.DID]struct{}),
		failing:     make(map[syntax.DID]failure),
	}
}

type failure struct {
	rounds  int
	retryAt time.Time
}

func (b *backfill) failedOwnerRetry(rounds int) time.Duration {
	return min(b.retryFailed<<min(rounds-1, 10), failedOwnerRetryMax)
}

// nextRetry is when the earliest failed owner is due again, zero if none is
func (b *backfill) nextRetry() time.Time {
	var next time.Time
	for _, f := range b.failing {
		if next.IsZero() || f.retryAt.Before(next) {
			next = f.retryAt
		}
	}
	return next
}

// Track queues owners for backfill and wakes hydrant up if any are new to this
// process. hydrant skips owners it already backfilled, so repeats are cheap.
func (b *backfill) Track(dids ...syntax.DID) {
	b.mu.Lock()
	queued := false
	for _, did := range dids {
		if _, ok := b.tracked[did]; ok || did == "" {
			continue
		}
		// an owner untracked earlier stays queued for removal, since nothing
		// followed their repo meanwhile. erasing then adding them again is
		// what makes hydrant backfill them from scratch instead of skipping
		b.tracked[did] = struct{}{}
		b.add[did] = struct{}{}
		queued = true
	}
	b.mu.Unlock()
	if queued {
		nudge(b.wake)
	}
}

// hydrant skips dids it already has, so the catch-up erases and re-tracks them
// saved first so a restart can't lose the gap, and a did leaves the list once backfilled
func (b *backfill) CatchUp() error {
	owners, err := b.s.backfillOwners()
	if err != nil {
		return err
	}
	repos, err := b.s.db.AllRepos()
	if err != nil {
		return fmt.Errorf("loading repos to catch up: %w", err)
	}
	dids := owners
	for _, repo := range repos {
		dids = append(dids, repo.RepoDid)
	}
	dids = slices.DeleteFunc(dids, func(did syntax.DID) bool { return did == "" })
	saveErr := b.s.db.AddPendingCatchUp(dids, time.Now())
	b.queueCatchUp(dids)
	if saveErr != nil {
		return fmt.Errorf("saving the pending catch-up: %w", saveErr)
	}
	return nil
}

// called from the jetstream reader, which can't block on the catch-up
func (b *backfill) CatchUpLater() {
	b.mu.Lock()
	defer b.mu.Unlock()
	if b.catchingUp {
		b.catchUpAgain = true
		return
	}
	// close waits on catchUps, and nothing may start once closed is set
	if b.closed {
		return
	}
	b.catchingUp = true
	b.catchUps.Go(func() {
		for {
			if err := b.CatchUp(); err != nil {
				b.l.Error("jetstream missed events and the catch-up couldn't start", "err", err)
			}
			b.mu.Lock()
			again := b.catchUpAgain
			b.catchUpAgain = false
			b.catchingUp = again
			b.mu.Unlock()
			if !again {
				return
			}
		}
	})
}

func (s *Spindle) markBackfilled(ctx context.Context, evt tapc.Event) error {
	if evt.Type == tapc.EvtRecord && evt.Record != nil {
		switch evt.Record.Collection.String() {
		case org_tangled.RepoManifestNSID, org_tangled.GraphCollaboratorNSID:
			if served, err := s.knowsRepo(evt.Record.Did); err != nil || served {
				return err
			}
		case org_tangled.RepoDeclarationNSID:
			if served, err := s.declaresServedRepo(evt.Record); err != nil || served {
				return err
			}
		}
	}
	return s.processEvent(withMarkOnly(ctx), evt)
}

func (s *Spindle) declaresServedRepo(rec *tapc.RecordEventData) (bool, error) {
	if rec.Action == tapc.RecordDeleteAction {
		return false, nil
	}
	var declaration org_tangled.RepoDeclaration
	if err := json.Unmarshal(rec.Record, &declaration); err != nil {
		return false, nil
	}
	repoDid, err := syntax.ParseDID(declaration.Repo)
	if err != nil {
		return false, nil
	}
	rows, err := s.db.ReposByDid(repoDid)
	if err != nil {
		return false, err
	}
	return slices.ContainsFunc(rows, func(r db.Repo) bool { return r.Owner == rec.Did && r.Rkey == rec.Rkey }), nil
}

func (b *backfill) Resume() error {
	b.dropStalePending()
	pending, err := b.s.db.PendingCatchUp()
	if err != nil {
		return fmt.Errorf("loading the pending catch-up: %w", err)
	}
	if len(pending) > 0 {
		b.l.Info("resuming an unfinished catch-up", "dids", len(pending))
		b.queueCatchUp(pending)
	}
	return nil
}

func (b *backfill) dropStalePending() {
	dropped, err := b.s.db.DropStalePendingCatchUp(time.Now().Add(-pendingCatchUpMaxAge))
	if err != nil {
		b.l.Error("dropping stale pending catch-up", "err", err)
		return
	}
	for _, did := range dropped {
		b.l.Warn("giving up on catching up a did that never backfilled", "did", did)
	}
}

func (b *backfill) queueCatchUp(dids []syntax.DID) {
	b.mu.Lock()
	for _, did := range dids {
		b.tracked[did] = struct{}{}
		b.add[did] = struct{}{}
		b.remove[did] = struct{}{}
	}
	b.mu.Unlock()
	nudge(b.wake)
}

// Untrack waits for the next time hydrant runs anyway, since forgetting an
// owner isn't worth starting it for. a restart loses the queue, but each round
// also drops whatever hydrant tracks that spindle no longer wants
func (b *backfill) Untrack(did syntax.DID) {
	b.mu.Lock()
	delete(b.tracked, did)
	delete(b.add, did)
	b.remove[did] = struct{}{}
	b.mu.Unlock()
	if err := b.s.db.DeletePendingCatchUp(did); err != nil {
		b.l.Error("forgetting a pending catch-up", "did", did, "err", err)
	}
}

func (b *backfill) take() (add, remove []syntax.DID) {
	b.mu.Lock()
	defer b.mu.Unlock()
	for did := range b.add {
		add = append(add, did)
	}
	for did := range b.remove {
		remove = append(remove, did)
	}
	clear(b.add)
	clear(b.remove)
	return add, remove
}

// requeue puts back work a failed round didn't get to, unless newer calls
// already changed their mind about those owners
func (b *backfill) requeue(add, remove []syntax.DID) {
	b.mu.Lock()
	defer b.mu.Unlock()
	for _, did := range add {
		if _, ok := b.tracked[did]; ok {
			b.add[did] = struct{}{}
		}
	}
	for _, did := range remove {
		b.remove[did] = struct{}{}
	}
}

// unwanted is every did in dids that spindle isn't tracking
func (b *backfill) unwanted(dids []syntax.DID) []syntax.DID {
	b.mu.Lock()
	defer b.mu.Unlock()
	var out []syntax.DID
	for _, did := range dids {
		if _, ok := b.tracked[did]; !ok {
			out = append(out, did)
		}
	}
	return out
}

func (b *backfill) run(ctx context.Context) {
	defer func() {
		b.mu.Lock()
		b.closed = true
		b.mu.Unlock()
		b.catchUps.Wait()
	}()
	// due far enough ahead that every parked record counts as due on startup
	due := time.Now().Add(b.retryParked << parkedRetries)
	for {
		b.retryParkedRecords(ctx, due)
		err := b.round(ctx)
		if ctx.Err() != nil {
			return
		}
		if err != nil {
			b.l.Error("backfill failed, retrying", "err", err, "in", backfillRetryDelay)
			select {
			case <-ctx.Done():
				return
			case <-time.After(backfillRetryDelay):
			}
			due = time.Now()
			continue
		}
		var retry <-chan time.Time
		if next := b.nextWake(); !next.IsZero() {
			retry = time.After(time.Until(next))
		}
		select {
		case <-ctx.Done():
			return
		case <-b.wake:
		case <-retry:
		}
		due = time.Now()
	}
}

func (b *backfill) nextWake() time.Time {
	next := b.nextRetry()
	parked, err := b.s.db.NextParkedRetry()
	if err != nil {
		b.l.Error("reading when parked records are due", "err", err)
		return next
	}
	if next.IsZero() || (!parked.IsZero() && parked.Before(next)) {
		next = parked
	}
	return next
}

func (b *backfill) socketPath() string {
	return filepath.Join(b.dir, "api.sock")
}

func (b *backfill) cursorPath() string {
	return filepath.Join(b.dir, "cursor")
}

func (b *backfill) settings() map[string]string {
	settings := map[string]string{
		"HYDRANT_DATABASE_PATH":      filepath.Join(b.dir, "db"),
		"HYDRANT_API_BIND":           "unix:" + b.socketPath(),
		"HYDRANT_PLC_URL":            b.s.cfg.Server.PlcUrl,
		"HYDRANT_FILTER_COLLECTIONS": strings.Join(backfillCollections, ","),
		// backfill only, owners are added explicitly and live events come from jetstream
		"HYDRANT_ENABLE_FIREHOSE":     "false",
		"HYDRANT_RELAY_HOSTS":         "",
		"HYDRANT_SEED_HOSTS":          "",
		"HYDRANT_ENABLE_CRAWLER":      "false",
		"HYDRANT_CRAWLER_URLS":        "",
		"HYDRANT_IDENTITY_CACHE_SIZE": "50000",
		"RUST_LOG":                    "warn",
	}
	// dev hosts resolve to loopback, so hydrant needs the private network flag
	if b.s.cfg.Server.Dev {
		settings["HYDRANT_ALLOW_PRIVATE_NETWORKS"] = "true"
	}
	return settings
}

// round starts hydrant, hands it the queued owners and due retries, and
// feeds its stream to spindle until every owner is settled and processed,
// then shuts it down
func (b *backfill) round(ctx context.Context) (err error) {
	add, remove := b.take()
	if len(add) == 0 && !b.retryDue(time.Now()) {
		b.requeue(nil, remove)
		return nil
	}
	defer func() {
		if err != nil {
			b.requeue(add, remove)
			b.s.metrics.RecordBackfillRound("failed")
		} else {
			b.s.metrics.RecordBackfillRound("done")
		}
	}()

	if err := os.MkdirAll(b.dir, 0o700); err != nil {
		return fmt.Errorf("creating hydrant dir: %w", err)
	}
	last, err := b.loadCursor()
	if err != nil {
		return err
	}

	started := time.Now()
	h, err := b.start(b.settings())
	if err != nil {
		return fmt.Errorf("starting hydrant: %w", err)
	}
	b.s.metrics.SetBackfillRunning(true)
	defer func() {
		if err := h.Shutdown(hydrantShutdownTimeout); err != nil {
			b.l.Error("hydrant did not shut down", "err", err)
		}
		b.s.metrics.SetBackfillRunning(false)
	}()

	api := newHydrantAPI(b.socketPath())
	if err := api.waitReady(ctx, h); err != nil {
		return err
	}
	head, err := api.head(ctx)
	if err != nil {
		return err
	}
	// past the head means the db was wiped or swapped under the cursor
	if last >= 0 && (head == nil || last > *head) {
		b.l.Warn("hydrant cursor is past the end of its stream, replaying the stream from the start", "cursor", last)
		last = -1
	}
	b.last.Store(last)

	known, err := api.trackedRepos(ctx)
	if err != nil {
		return err
	}
	var dids []syntax.DID
	for _, repo := range known {
		dids = append(dids, repo.Did)
	}
	remove = append(remove, b.unwanted(dids)...)
	slices.Sort(remove)
	remove = slices.Compact(remove)
	if err := api.erase(ctx, remove); err != nil {
		return err
	}
	if err := api.track(ctx, add); err != nil {
		return err
	}
	known = slices.DeleteFunc(known, func(repo hydrantRepo) bool {
		_, gone := slices.BinarySearch(remove, repo.Did)
		return gone
	})
	retried := b.dueRetries(known, started)
	if err := api.resync(ctx, retried); err != nil {
		return err
	}
	b.l.Info("backfilling owners", "owners", len(add), "retrying", len(retried), "untracked", len(remove), "cursor", last)

	streamCtx, stopStream := context.WithCancel(ctx)
	streamDone := make(chan struct{})
	go func() {
		defer close(streamDone)
		b.follow(streamCtx, api)
	}()
	defer func() {
		stopStream()
		<-streamDone
		if err := b.saveCursor(b.last.Load()); err != nil {
			b.l.Warn("saving hydrant cursor", "err", err)
		}
	}()

	// an outcome sticks for the round, because hydrant keeps moving failed
	// owners back into its queue on its own backoff, so a round that waited for
	// that to calm down would never end
	settled := make(map[syntax.DID]outcome)
	ticker := time.NewTicker(b.poll)
	defer ticker.Stop()
	ready := 0
	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-h.Done():
			return fmt.Errorf("hydrant stopped: %w", h.Err())
		case <-b.wake:
			more, less := b.take()
			add, remove = append(add, more...), append(remove, less...)
			if err := api.erase(ctx, less); err != nil {
				return err
			}
			if err := api.track(ctx, more); err != nil {
				return err
			}
			// a catch-up can re-track a did this round already settled, so forget its outcome
			for _, did := range more {
				delete(settled, did)
			}
			ready = 0
		case <-ticker.C:
			repos, err := api.queuedRepos(ctx)
			if err != nil {
				return err
			}
			head, err := api.head(ctx)
			if err != nil {
				return err
			}
			left := 0
			for _, repo := range repos {
				if settled[repo.Did] == waiting {
					settled[repo.Did] = repo.outcome()
				}
				if settled[repo.Did] == waiting {
					left++
				}
			}
			if left == 0 && (head == nil || b.last.Load() >= *head) {
				ready++
			} else {
				ready = 0
			}
			if ready >= backfillStableChecks {
				// the queue only shows what hydrant is still working on, so an
				// owner it let go of without syncing only turns up in the full list
				final, err := api.trackedRepos(ctx)
				if err != nil {
					return err
				}
				for _, repo := range final {
					if settled[repo.Did] == waiting {
						settled[repo.Did] = repo.outcome()
					}
				}
				failing := b.recordFailures(final, settled, retried, time.Now())
				b.markFailedServed(settled)
				b.settlePending(ctx, add, settled)
				b.dropStalePending()
				b.l.Info("backfill done, shutting hydrant down",
					"owners", len(add), "failed", failing, "retry_at", b.nextRetry(),
					"took", time.Since(started).Round(time.Second).String())
				return nil
			}
		}
	}
}

func (b *backfill) retryDue(now time.Time) bool {
	next := b.nextRetry()
	return !next.IsZero() && !next.After(now)
}

// dueRetries picks the failed owners whose retry is due. one spindle has no
// record of is due right away, since it either just failed for the first time
// or this process restarted and forgot
func (b *backfill) dueRetries(repos []hydrantRepo, now time.Time) []syntax.DID {
	var due []syntax.DID
	for _, repo := range repos {
		if repo.outcome() != failed {
			continue
		}
		if f, ok := b.failing[repo.Did]; !ok || !f.retryAt.After(now) {
			due = append(due, repo.Did)
		}
	}
	return due
}

// recordFailures pushes back the next retry of every owner that failed again
// after being retried, and forgets the ones that made it. an owner that wasn't
// due keeps its time even if hydrant retried and failed it on its own
func (b *backfill) recordFailures(repos []hydrantRepo, settled map[syntax.DID]outcome, retried []syntax.DID, now time.Time) int {
	wasRetried := make(map[syntax.DID]struct{}, len(retried))
	for _, did := range retried {
		wasRetried[did] = struct{}{}
	}
	failing := make(map[syntax.DID]failure)
	for _, repo := range repos {
		if settled[repo.Did] != failed {
			continue
		}
		f, known := b.failing[repo.Did]
		if _, ok := wasRetried[repo.Did]; !known || ok {
			f.rounds++
			f.retryAt = now.Add(b.failedOwnerRetry(f.rounds))
		}
		failing[repo.Did] = f
	}
	b.failing = failing
	b.s.metrics.SetBackfillFailingOwners(len(failing))
	return len(failing)
}

// follow feeds hydrant's stream to spindle from the saved cursor, reconnecting
// at the last processed event if the connection drops
func (b *backfill) follow(ctx context.Context, api hydrantAPI) {
	for ctx.Err() == nil {
		err := b.followOnce(ctx, api)
		if ctx.Err() != nil {
			return
		}
		b.l.Warn("hydrant stream dropped, reconnecting", "err", err, "cursor", b.last.Load())
		select {
		case <-ctx.Done():
		case <-time.After(time.Second):
		}
	}
}

func (b *backfill) followOnce(ctx context.Context, api hydrantAPI) error {
	conn, err := api.subscribe(ctx, b.last.Load()+1)
	if err != nil {
		return err
	}
	defer conn.Close()
	// ReadMessage doesn't watch ctx, so closing the conn is what unblocks it
	stop := context.AfterFunc(ctx, func() { conn.Close() })
	defer stop()

	lastSaved := time.Now()
	for {
		_, msg, err := conn.ReadMessage()
		if err != nil {
			return err
		}
		var evt tapc.Event
		if err := json.Unmarshal(msg, &evt); err != nil {
			return fmt.Errorf("parsing hydrant event: %w", err)
		}
		if evt.Type == "error" {
			return fmt.Errorf("hydrant closed the stream: %s", msg)
		}
		if err := b.process(ctx, evt); err != nil {
			return err
		}
		// only records move the cursor, other event ids run past the stream's head
		if evt.Type != tapc.EvtRecord {
			continue
		}
		b.last.Store(evt.ID)
		// a crash replays at most a second of events, which is fine since
		// processing a record twice changes nothing
		if time.Since(lastSaved) >= time.Second {
			if err := b.saveCursor(evt.ID); err != nil {
				b.l.Warn("saving hydrant cursor", "err", err)
			}
			lastSaved = time.Now()
		}
	}
}

// a failed park returns an error, which stops the stream so the event replays
func (b *backfill) process(ctx context.Context, evt tapc.Event) error {
	if evt.Type != tapc.EvtRecord || evt.Record == nil {
		return nil
	}
	var err error
	for attempt := 1; ; attempt++ {
		if err = b.handle(ctx, evt); err == nil {
			return nil
		}
		if ctx.Err() != nil {
			return ctx.Err()
		}
		if attempt == backfillEventAttempts {
			break
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(time.Duration(attempt) * b.retryBase):
		}
	}
	return b.park(evt, err)
}

func (b *backfill) park(evt tapc.Event, cause error) error {
	raw, err := json.Marshal(evt)
	if err != nil {
		return err
	}
	uri := evt.Record.AtUri().String()
	if err := b.s.db.ParkRecord(db.ParkedRecord{
		URI:     uri,
		Rev:     evt.Record.Rev,
		Event:   raw,
		RetryAt: time.Now().Add(b.retryParked),
		Err:     cause.Error(),
	}); err != nil {
		return fmt.Errorf("parking %s: %w", uri, err)
	}
	b.l.Warn("parked a backfilled record that kept failing", "uri", uri, "err", cause, "retry_in", b.retryParked.String())
	b.s.metrics.RecordBackfillParked("parked")
	b.countParked()
	return nil
}

// retryParkedRecords runs off the stored event, so hydrant doesn't have to be up
func (b *backfill) retryParkedRecords(ctx context.Context, due time.Time) {
	parked, err := b.s.db.DueParkedRecords(due)
	if err != nil {
		b.l.Error("listing parked records", "err", err)
		return
	}
	defer b.countParked()
	for _, p := range parked {
		var evt tapc.Event
		err := json.Unmarshal(p.Event, &evt)
		if err == nil {
			err = b.handle(ctx, evt)
		}
		if ctx.Err() != nil {
			return
		}
		attempts := p.Attempts + 1
		var saveErr error
		switch {
		case err == nil:
			b.l.Info("parked record went through", "uri", p.URI, "attempts", attempts)
			b.s.metrics.RecordBackfillParked("recovered")
			saveErr = b.s.db.UnparkRecord(p.URI, p.Rev)
		case attempts >= parkedRetries:
			b.l.Error("gave up on a parked record", "uri", p.URI, "attempts", attempts, "err", err)
			b.s.metrics.RecordBackfillParked("gave_up")
			saveErr = b.s.db.UnparkRecord(p.URI, p.Rev)
		default:
			wait := b.retryParked << attempts
			b.l.Warn("parked record failed again", "uri", p.URI, "attempts", attempts, "retry_in", wait.String(), "err", err)
			saveErr = b.s.db.RescheduleParked(p.URI, p.Rev, attempts, time.Now().Add(wait), err.Error())
		}
		if saveErr != nil {
			b.l.Error("updating parked record", "uri", p.URI, "err", saveErr)
		}
	}
}

func (b *backfill) countParked() {
	n, err := b.s.db.CountParked()
	if err != nil {
		b.l.Error("counting parked records", "err", err)
		return
	}
	b.s.metrics.SetBackfillParkedRecords(n)
}

// the cursor lives next to hydrant's db rather than in spindle's, so wiping
// that dir also starts the stream over instead of skipping a new db's events
func (b *backfill) loadCursor() (int64, error) {
	raw, err := os.ReadFile(b.cursorPath())
	if errors.Is(err, os.ErrNotExist) {
		return -1, nil
	}
	if err != nil {
		return 0, fmt.Errorf("reading hydrant cursor: %w", err)
	}
	last, err := strconv.ParseInt(strings.TrimSpace(string(raw)), 10, 64)
	if err != nil {
		return 0, fmt.Errorf("parsing hydrant cursor: %w", err)
	}
	return last, nil
}

func (b *backfill) saveCursor(last int64) error {
	if last < 0 {
		return nil
	}
	tmp := b.cursorPath() + ".tmp"
	if err := os.WriteFile(tmp, []byte(strconv.FormatInt(last, 10)), 0o600); err != nil {
		return err
	}
	return os.Rename(tmp, b.cursorPath())
}

// hydrantAPI talks to hydrant's http api over its unix socket
type hydrantAPI struct {
	socket string
	http   *http.Client
}

func newHydrantAPI(socket string) hydrantAPI {
	dial := func(ctx context.Context, _, _ string) (net.Conn, error) {
		var d net.Dialer
		return d.DialContext(ctx, "unix", socket)
	}
	return hydrantAPI{
		socket: socket,
		http:   &http.Client{Transport: &http.Transport{DialContext: dial}},
	}
}

func (a hydrantAPI) do(ctx context.Context, method, path string, body any, out any) error {
	return a.doWithin(ctx, hydrantRequestTimeout, method, path, body, out)
}

func (a hydrantAPI) doWithin(ctx context.Context, timeout time.Duration, method, path string, body any, out any) error {
	ctx, cancel := context.WithTimeout(ctx, timeout)
	defer cancel()
	var reqBody io.Reader
	if body != nil {
		raw, err := json.Marshal(body)
		if err != nil {
			return err
		}
		reqBody = bytes.NewReader(raw)
	}
	req, err := http.NewRequestWithContext(ctx, method, "http://hydrant"+path, reqBody)
	if err != nil {
		return err
	}
	req.Header.Set("Accept", "application/json")
	if body != nil {
		req.Header.Set("Content-Type", "application/json")
	}
	resp, err := a.http.Do(req)
	if err != nil {
		return fmt.Errorf("hydrant %s %s: %w", method, path, err)
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		msg, _ := io.ReadAll(io.LimitReader(resp.Body, 512))
		return fmt.Errorf("hydrant %s %s: %s: %s", method, path, resp.Status, bytes.TrimSpace(msg))
	}
	if out == nil {
		return nil
	}
	return json.NewDecoder(resp.Body).Decode(out)
}

// the api binds after start, so the first probes are expected to fail
func (a hydrantAPI) waitReady(ctx context.Context, h embeddedHydrant) error {
	ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
	defer cancel()
	for {
		err := a.doWithin(ctx, time.Second, http.MethodGet, "/stream/head", nil, nil)
		if err == nil {
			return nil
		}
		select {
		case <-ctx.Done():
			return fmt.Errorf("hydrant api never came up: %w", err)
		case <-h.Done():
			return fmt.Errorf("hydrant stopped before its api came up: %w", h.Err())
		case <-time.After(100 * time.Millisecond):
		}
	}
}

func (a hydrantAPI) track(ctx context.Context, dids []syntax.DID) error {
	return a.sendDids(ctx, http.MethodPut, "/repos", dids)
}

// erase makes hydrant forget these owners entirely. a plain untrack keeps
// their records, and a later backfill would skip every one it still has
func (a hydrantAPI) erase(ctx context.Context, dids []syntax.DID) error {
	for _, did := range dids {
		if err := a.doWithin(ctx, hydrantEraseTimeout, http.MethodPost, "/redactions/"+did.String(), nil, nil); err != nil {
			return err
		}
	}
	return nil
}

// resync makes hydrant retry these owners right away instead of on its own backoff
func (a hydrantAPI) resync(ctx context.Context, dids []syntax.DID) error {
	return a.sendDids(ctx, http.MethodPost, "/repos/resync", dids)
}

// both endpoints are idempotent, so a failed round can resend the whole list
func (a hydrantAPI) sendDids(ctx context.Context, method, path string, dids []syntax.DID) error {
	type didRef struct {
		Did syntax.DID `json:"did"`
	}
	for batch := range slices.Chunk(dids, hydrantDidBatch) {
		body := make([]didRef, len(batch))
		for i, did := range batch {
			body[i] = didRef{did}
		}
		if err := a.do(ctx, method, path, body, nil); err != nil {
			return err
		}
	}
	return nil
}

// hydrantRepo is what GET /repos says about one owner
type hydrantRepo struct {
	Did     syntax.DID `json:"did"`
	Tracked bool       `json:"tracked"`
	Status  string     `json:"status"`
	Pending bool       `json:"pending"`
	// set while the owner sits in hydrant's resync queue
	Resync *hydrantResync `json:"resync,omitempty"`
}

type hydrantResync struct {
	State        string `json:"state"`
	Kind         string `json:"kind,omitempty"`
	HostFailures int    `json:"host_failures,omitempty"`
}

type outcome int

const (
	waiting outcome = iota
	done
	failed
)

// outcome says whether waiting longer this round can still get this owner
// backfilled
func (r hydrantRepo) outcome() outcome {
	switch {
	case r.Resync != nil && r.Resync.State == "gone":
		return done
	case r.Resync != nil:
		// a live pds pacing hydrant gets through soon. a pds that keeps
		// failing makes hydrant file its other owners as ratelimited too
		if r.Resync.Kind == "ratelimited" && r.Resync.HostFailures == 0 {
			return waiting
		}
		return failed
	case r.Pending:
		return waiting
	}
	switch r.Status {
	case "synced", "deactivated", "takendown", "suspended", "deleted", "throttled":
		return done
	}
	// hydrant let go of it without calling it synced, so retrying later is
	// all that's left
	return failed
}

// trackedRepos lists every owner hydrant keeps up to date
func (a hydrantAPI) trackedRepos(ctx context.Context) ([]hydrantRepo, error) {
	return a.listRepos(ctx, false)
}

// queuedRepos lists the owners hydrant still has a backfill or a retry queued
// for. an owner missing from it is done with for now, so a round can watch
// this instead of every owner it knows
func (a hydrantAPI) queuedRepos(ctx context.Context) ([]hydrantRepo, error) {
	return a.listRepos(ctx, true)
}

func (a hydrantAPI) listRepos(ctx context.Context, queued bool) ([]hydrantRepo, error) {
	const limit = 1000
	var repos []hydrantRepo
	cursor := ""
	for {
		var page []hydrantRepo
		query := url.Values{"limit": {strconv.Itoa(limit)}}
		if queued {
			query.Set("queued", "true")
		}
		if cursor != "" {
			query.Set("cursor", cursor)
		}
		if err := a.do(ctx, http.MethodGet, "/repos?"+query.Encode(), nil, &page); err != nil {
			return nil, err
		}
		for _, repo := range page {
			if repo.Tracked {
				repos = append(repos, repo)
			}
		}
		if len(page) < limit {
			return repos, nil
		}
		cursor = page[len(page)-1].Did.String()
	}
}

// head is the id of the newest event a replay can deliver, nil while there are none
func (a hydrantAPI) head(ctx context.Context) (*int64, error) {
	var out struct {
		ID *int64 `json:"id"`
	}
	if err := a.do(ctx, http.MethodGet, "/stream/head", nil, &out); err != nil {
		return nil, err
	}
	return out.ID, nil
}

func (a hydrantAPI) subscribe(ctx context.Context, cursor int64) (*websocket.Conn, error) {
	dialer := websocket.Dialer{
		NetDialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
			var d net.Dialer
			return d.DialContext(ctx, "unix", a.socket)
		},
		HandshakeTimeout: 10 * time.Second,
	}
	url := "ws://hydrant/stream?cursor=" + strconv.FormatInt(max(cursor, 0), 10)
	conn, _, err := dialer.DialContext(ctx, url, nil)
	if err != nil {
		return nil, fmt.Errorf("subscribing to hydrant stream: %w", err)
	}
	return conn, nil
}

func (b *backfill) markFailedServed(settled map[syntax.DID]outcome) {
	for did, o := range settled {
		if o != failed {
			continue
		}
		if served, err := b.s.knowsRepo(did); err != nil || !served {
			continue
		}
		if err := b.s.markRepo(did); err != nil {
			b.l.Error("marking a repo whose backfill failed", "repo", did, "err", err)
		}
	}
}

// only this round's outcomes count, an older one can predate a later catch-up
func (b *backfill) settlePending(ctx context.Context, roundDids []syntax.DID, settled map[syntax.DID]outcome) {
	var backfilled []syntax.DID
	for _, did := range roundDids {
		if settled[did] != done {
			continue
		}
		if ctx.Err() != nil {
			break
		}
		if b.refreshServed(ctx, did) {
			backfilled = append(backfilled, did)
		}
	}
	if err := b.s.db.DeletePendingCatchUp(backfilled...); err != nil {
		b.l.Error("clearing the pending catch-up", "err", err)
	}
}

// false keeps the did pending, a failed refresh doesn't because the retry set owns it
func (b *backfill) refreshServed(ctx context.Context, did syntax.DID) bool {
	served, err := b.s.knowsRepo(did)
	if err != nil {
		b.l.Error("checking whether a backfilled did is a served repo", "did", did, "err", err)
		return false
	}
	if !served {
		return true
	}
	if err := b.refresh(ctx, did); err != nil {
		if ctx.Err() != nil {
			return false
		}
		b.l.Warn("refreshing a repo after the catch-up failed", "repo", did, "err", err)
		b.s.failRepoRefresh(did)
	}
	return true
}