package spindle

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

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

	"tangled.org/core/api/org_tangled"
	"tangled.org/core/jetstream"
	"tangled.org/core/log"
	"tangled.org/core/rbac"
	"tangled.org/core/spindle/config"
	"tangled.org/core/spindle/db"
	"tangled.org/core/spindle/knotclient"
	"tangled.org/core/spindle/models"
	"tangled.org/core/spindle/secrets"
	"tangled.org/core/tapc"
)

const faultOtherRepoDid = syntax.DID("did:plc:repotwo")

const testRetryTick = 50 * time.Millisecond

type jsEvent struct {
	timeUS int64
	wire   []byte
}

type fakeJetstream struct {
	srv *httptest.Server

	mu      sync.Mutex
	events  []jsEvent
	oldest  int
	last    int64
	cursors []int64
	conns   map[*websocket.Conn]struct{}
}

func newFakeJetstream(t *testing.T) *fakeJetstream {
	t.Helper()
	j := &fakeJetstream{conns: map[*websocket.Conn]struct{}{}, last: time.Now().Add(-time.Hour).UnixMicro()}
	j.srv = httptest.NewServer(http.HandlerFunc(j.serve))
	t.Cleanup(func() {
		j.dropConnections()
		j.srv.Close()
	})
	return j
}

func (j *fakeJetstream) url() string {
	return "ws" + strings.TrimPrefix(j.srv.URL, "http") + "/subscribe"
}

// 5s apart, so the 1s rewind on resume only replays the last event
func (j *fakeJetstream) add(e jetstream.Event) int64 {
	j.mu.Lock()
	defer j.mu.Unlock()
	j.last += 5_000_000
	e.TimeUS = j.last
	raw, err := json.Marshal(e)
	if err != nil {
		panic(err)
	}
	enc, _ := zstd.NewWriter(nil)
	j.events = append(j.events, jsEvent{timeUS: e.TimeUS, wire: enc.EncodeAll(raw, nil)})
	return e.TimeUS
}

func (j *fakeJetstream) commit(did syntax.DID, collection, rkey, op string, record any) int64 {
	commit := &jetstream.Commit{Rev: "rev", Operation: op, Collection: collection, RKey: rkey, CID: "bafy"}
	if record != nil {
		raw, err := json.Marshal(record)
		if err != nil {
			panic(err)
		}
		commit.Record = raw
	}
	return j.add(jetstream.Event{Did: did.String(), Kind: jetstream.EventKindCommit, Commit: commit})
}

// the ingester ignores identity events, but the cursor still moves past them
func (j *fakeJetstream) filler() int64 {
	return j.add(jetstream.Event{Did: testOwner.String(), Kind: jetstream.EventKindIdentity})
}

// like retention, a resume can't go back before this point
func (j *fakeJetstream) forget() {
	j.mu.Lock()
	defer j.mu.Unlock()
	j.oldest = len(j.events)
}

func (j *fakeJetstream) dropConnections() {
	j.mu.Lock()
	defer j.mu.Unlock()
	for c := range j.conns {
		c.Close()
	}
}

func (j *fakeJetstream) connects() int {
	j.mu.Lock()
	defer j.mu.Unlock()
	return len(j.cursors)
}

func (j *fakeJetstream) serve(w http.ResponseWriter, r *http.Request) {
	conn, err := (&websocket.Upgrader{}).Upgrade(w, r, nil)
	if err != nil {
		return
	}
	defer conn.Close()

	j.mu.Lock()
	j.conns[conn] = struct{}{}
	next, cursor := len(j.events), int64(-1)
	if raw := r.URL.Query().Get("cursor"); raw != "" {
		cursor, _ = strconv.ParseInt(raw, 10, 64)
		next = j.oldest
		for next < len(j.events) && j.events[next].timeUS < cursor {
			next++
		}
	}
	j.cursors = append(j.cursors, cursor)
	j.mu.Unlock()
	defer func() {
		j.mu.Lock()
		delete(j.conns, conn)
		j.mu.Unlock()
	}()

	gone := closedWhenDone(conn)
	for {
		j.mu.Lock()
		var batch [][]byte
		for ; next < len(j.events); next++ {
			batch = append(batch, j.events[next].wire)
		}
		j.mu.Unlock()
		for _, wire := range batch {
			if err := conn.WriteMessage(websocket.BinaryMessage, wire); err != nil {
				return
			}
		}
		select {
		case <-gone:
			return
		case <-time.After(5 * time.Millisecond):
		}
	}
}

// serves did docs so spindle's real resolver runs unchanged
type fakePLC struct {
	srv *httptest.Server

	mu    sync.Mutex
	knots map[syntax.DID]string
	pdses map[syntax.DID]string
}

func newFakePLC(t *testing.T) *fakePLC {
	t.Helper()
	p := &fakePLC{knots: map[syntax.DID]string{}, pdses: map[syntax.DID]string{}}
	p.srv = httptest.NewServer(http.HandlerFunc(p.serve))
	t.Cleanup(p.srv.Close)
	return p
}

func (p *fakePLC) setKnot(repo syntax.DID, url string) {
	p.mu.Lock()
	defer p.mu.Unlock()
	p.knots[repo] = url
}

func (p *fakePLC) setPDS(owner syntax.DID, url string) {
	p.mu.Lock()
	defer p.mu.Unlock()
	p.pdses[owner] = url
}

func (p *fakePLC) serve(w http.ResponseWriter, r *http.Request) {
	did := syntax.DID(strings.TrimPrefix(r.URL.Path, "/"))
	p.mu.Lock()
	knot, isRepo := p.knots[did]
	pds, isOwner := p.pdses[did]
	p.mu.Unlock()
	if !isRepo && !isOwner {
		http.NotFound(w, r)
		return
	}
	type service struct {
		ID       string `json:"id"`
		Type     string `json:"type"`
		Endpoint string `json:"serviceEndpoint"`
	}
	var services []service
	if isRepo {
		services = append(services, service{"#tangled_knot", "TangledKnot", knot})
	}
	if isOwner {
		services = append(services, service{"#atproto_pds", "AtprotoPersonalDataServer", pds})
	}
	json.NewEncoder(w).Encode(map[string]any{
		"@context":           []string{"https://www.w3.org/ns/did/v1"},
		"id":                 did,
		"alsoKnownAs":        []string{},
		"verificationMethod": []any{},
		"service":            services,
	})
}

// requests wait on it until released
type hold struct {
	release chan struct{}
	fail    bool
}

func (h *hold) wait(r *http.Request) bool {
	select {
	case <-h.release:
		return true
	case <-r.Context().Done():
		return false
	}
}

type fakeRepo struct {
	work string
	bare string
	n    int
}

func gitExecPath(t *testing.T) string {
	t.Helper()
	out, err := exec.Command("git", "--exec-path").Output()
	if err != nil {
		t.Skipf("git not available: %v", err)
	}
	backend := filepath.Join(strings.TrimSpace(string(out)), "git-http-backend")
	if _, err := os.Stat(backend); err != nil {
		t.Skipf("git http-backend not available: %v", err)
	}
	return backend
}

func (r *fakeRepo) git(t *testing.T, dir string, args ...string) string {
	t.Helper()
	cmd := exec.Command("git", append([]string{"-C", dir}, args...)...)
	out, err := cmd.CombinedOutput()
	if err != nil {
		t.Fatalf("git %v: %v\n%s", args, err, out)
	}
	return strings.TrimSpace(string(out))
}

func (r *fakeRepo) commit(t *testing.T) string {
	t.Helper()
	return r.commitWorkflows(t, map[string]string{
		"ci.yaml": "engine: dummy\nwhen:\n  - event: push\n    branch: main\nsteps:\n  - name: hi\n    command: echo hi\n",
	})
}

// pushed to its own branch so the sha stays reachable
func (r *fakeRepo) commitWorkflows(t *testing.T, files map[string]string) string {
	t.Helper()
	r.n++
	workflows := filepath.Join(r.work, ".tangled", "workflows")
	if err := os.MkdirAll(workflows, 0o755); err != nil {
		t.Fatal(err)
	}
	for name, contents := range files {
		if err := os.WriteFile(filepath.Join(workflows, name), []byte(contents), 0o644); err != nil {
			t.Fatal(err)
		}
	}
	if err := os.WriteFile(filepath.Join(r.work, "n"), []byte(strconv.Itoa(r.n)), 0o644); err != nil {
		t.Fatal(err)
	}
	r.git(t, r.work, "add", "-A")
	r.git(t, r.work, "commit", "-q", "-m", "c"+strconv.Itoa(r.n))
	sha := r.git(t, r.work, "rev-parse", "HEAD")
	r.git(t, r.work, "push", "-q", r.bare, sha+":refs/heads/c"+strconv.Itoa(r.n))
	return sha
}

type world struct {
	t      *testing.T
	dir    string
	hydDir string
	js     *fakeJetstream
	plc    *fakePLC
	knot   *fakeKnot
	other  *fakeKnot
	hyd    *fakeHydrant
	skew   atomic.Int64
	cgi    *cgi.Handler
	// empty means open admission and the usual spindle owner
	admission config.Admission
	owner     syntax.DID

	mu           sync.Mutex
	hydFailing   map[syntax.DID]bool
	gitHold      *hold
	manifestHold *hold
	gitFetches   atomic.Int32
	gitFailing   atomic.Bool
	heldReads    atomic.Int32
}

func newWorld(t *testing.T) *world {
	t.Helper()
	t.Parallel()

	w := &world{t: t, dir: t.TempDir(), hydDir: shortTempDir(t), hydFailing: map[syntax.DID]bool{}}
	w.js = newFakeJetstream(t)
	w.plc = newFakePLC(t)
	w.other = newFakeKnot(t)

	w.knot = &fakeKnot{records: map[string]any{}}
	w.knot.srv = httptest.NewServer(http.HandlerFunc(w.serveKnot))
	t.Cleanup(w.knot.srv.Close)

	w.plc.setPDS(testOwner, w.knot.srv.URL)
	w.plc.setKnot(testRepoDid, w.knot.srv.URL)
	w.plc.setKnot(faultOtherRepoDid, w.knot.srv.URL)
	w.hyd = &fakeHydrant{}
	w.hyd.onTrack = func(did syntax.DID) {
		if w.hyd.repoStatus(did) != synced(did).Status {
			w.backfillOwner(did)
		}
	}
	w.hyd.onResync = func(did syntax.DID) {
		if w.hyd.repoStatus(did) != "" {
			w.backfillOwner(did)
		}
	}
	return w
}

func (w *world) dbPath() string { return filepath.Join(w.dir, "spindle.db") }

func (w *world) spindleDid() string { return config.Server{Hostname: faultHostname}.Did().String() }

const faultHostname = "spindle.example.com"

func (w *world) knotRef() knotclient.Knot {
	return knotclient.Knot{Did: syntax.DID("did:web:" + w.knot.host()), Host: w.knot.host(), Scheme: "http"}
}

func (w *world) serveKnot(rw http.ResponseWriter, r *http.Request) {
	if strings.HasPrefix(r.URL.Path, "/did:") {
		w.serveGit(rw, r)
		return
	}
	w.mu.Lock()
	held := w.manifestHold
	w.mu.Unlock()
	if held != nil && r.URL.Query().Get("collection") == org_tangled.RepoManifestNSID {
		w.heldReads.Add(1)
		if !held.wait(r) {
			return
		}
	}
	w.knot.serve(rw, r)
}

func (w *world) serveGit(rw http.ResponseWriter, r *http.Request) {
	if strings.HasSuffix(r.URL.Path, "/info/refs") {
		w.gitFetches.Add(1)
	}
	w.mu.Lock()
	held, handler := w.gitHold, w.cgi
	w.mu.Unlock()
	if held != nil {
		if !held.wait(r) {
			return
		}
		if held.fail {
			http.Error(rw, "boom", http.StatusInternalServerError)
			return
		}
	}
	if w.gitFailing.Load() {
		http.Error(rw, "boom", http.StatusInternalServerError)
		return
	}
	if handler == nil {
		http.Error(rw, "no git", http.StatusNotFound)
		return
	}
	handler.ServeHTTP(rw, r)
}

func (w *world) serveGitRepo(repoDid syntax.DID) *fakeRepo {
	t := w.t
	t.Helper()
	backend := gitExecPath(t)
	repo := &fakeRepo{work: filepath.Join(w.dir, "work"), bare: filepath.Join(w.dir, "git", repoDid.String())}
	for _, dir := range []string{repo.work, repo.bare} {
		if err := os.MkdirAll(dir, 0o755); err != nil {
			t.Fatal(err)
		}
	}
	repo.git(t, repo.work, "init", "-q", "-b", "main")
	repo.git(t, repo.bare, "init", "-q", "--bare")
	repo.git(t, repo.bare, "config", "uploadpack.allowFilter", "true")
	repo.git(t, repo.bare, "config", "uploadpack.allowAnySHA1InWant", "true")
	w.mu.Lock()
	defer w.mu.Unlock()
	w.cgi = &cgi.Handler{
		Path:       backend,
		Root:       "/",
		Env:        []string{"GIT_PROJECT_ROOT=" + filepath.Join(w.dir, "git"), "GIT_HTTP_EXPORT_ALL=1"},
		InheritEnv: []string{"PATH", "HOME", "GIT_CONFIG_GLOBAL", "GIT_CONFIG_SYSTEM"},
	}
	return repo
}

func (w *world) hold(slot **hold) (release func(fail bool)) {
	h := &hold{release: make(chan struct{})}
	w.mu.Lock()
	*slot = h
	w.mu.Unlock()
	var once sync.Once
	return func(fail bool) {
		once.Do(func() {
			h.fail = fail
			w.mu.Lock()
			*slot = nil
			w.mu.Unlock()
			close(h.release)
		})
	}
}

func (w *world) holdGit() func(fail bool)      { return w.hold(&w.gitHold) }
func (w *world) holdManifest() func(fail bool) { return w.hold(&w.manifestHold) }

func (w *world) setManifest(repo syntax.DID, name, spindle string) {
	w.knot.put(repo, org_tangled.RepoManifestNSID, knotclient.ManifestRkey, manifestRecord(testOwner, name, spindle))
}

func manifestRecord(owner syntax.DID, name, spindle string) org_tangled.RepoManifest {
	m := org_tangled.RepoManifest{
		LexiconTypeID: org_tangled.RepoManifestNSID,
		CreatedAt:     "2026-01-01T00:00:00Z",
		Declaration:   &org_tangled.RepoManifest_Declaration{Owner: owner.String(), Name: name},
	}
	if spindle != "" {
		m.Spindle = &spindle
	}
	return m
}

func (w *world) declare(name string, repo syntax.DID) {
	w.knot.declare(testOwner, strings.ToLower(name), repo)
}

func (w *world) undeclare(name string) { w.knot.undeclare(testOwner, strings.ToLower(name)) }

func (w *world) addCollaborator(repo, who syntax.DID) {
	w.knot.put(repo, org_tangled.GraphCollaboratorNSID, who.String(), map[string]string{"createdAt": "2026-01-01T00:00:00Z"})
}

func (w *world) removeCollaborator(repo, who syntax.DID) {
	w.knot.erase(repo, org_tangled.GraphCollaboratorNSID, who.String())
}

func (w *world) serve(repo syntax.DID, name string, collaborators ...syntax.DID) {
	w.setManifest(repo, name, w.spindleDid())
	w.declare(name, repo)
	for _, c := range collaborators {
		w.addCollaborator(repo, c)
	}
}

func (w *world) emitManifest(repo syntax.DID) int64 {
	return w.js.manifest(repo, w.knot.record(repo, org_tangled.RepoManifestNSID, knotclient.ManifestRkey))
}

func (w *world) emitCollaborator(repo, who syntax.DID, op string) int64 {
	return w.js.collaborator(repo, who, op)
}

func (w *world) emitDeclaration(name string, repo syntax.DID) int64 {
	return w.js.declaration(testOwner, name, repo)
}

func (j *fakeJetstream) manifest(repo syntax.DID, value any) int64 {
	return j.commit(repo, org_tangled.RepoManifestNSID, knotclient.ManifestRkey, jetstream.CommitOperationUpdate, value)
}

func (j *fakeJetstream) collaborator(repo, who syntax.DID, op string) int64 {
	var record any
	if op != jetstream.CommitOperationDelete {
		record = map[string]string{"createdAt": "2026-01-01T00:00:00Z"}
	}
	return j.commit(repo, org_tangled.GraphCollaboratorNSID, who.String(), op, record)
}

func (j *fakeJetstream) declaration(owner syntax.DID, name string, repo syntax.DID) int64 {
	record := org_tangled.RepoDeclaration{LexiconTypeID: org_tangled.RepoDeclarationNSID, Repo: repo.String()}
	return j.commit(owner, org_tangled.RepoDeclarationNSID, strings.ToLower(name), opCreate, record)
}

func (k *fakeKnot) record(repo syntax.DID, collection, rkey string) any {
	k.mu.Lock()
	defer k.mu.Unlock()
	return k.records[repo.String()+"|"+collection+"|"+rkey]
}

func (w *world) backfillOwner(did syntax.DID) {
	w.mu.Lock()
	failing := w.hydFailing[did]
	w.mu.Unlock()
	if failing {
		w.hyd.setRepo(failedWith(did, "transport", 0))
		return
	}
	w.hyd.backfill(did, w.knot.recordsOf(did))
}

// set the repo first, or a listing can call it backfilled before the events land
func (f *fakeHydrant) backfill(did syntax.DID, events []tapc.RecordEventData) {
	f.setRepo(retrying(did))
	for _, rec := range events {
		f.addEvent(rec)
	}
	f.setRepo(synced(did))
}

func (w *world) setHydrantFailing(did syntax.DID, on bool) {
	w.mu.Lock()
	defer w.mu.Unlock()
	w.hydFailing[did] = on
}

// backfill only sees current state, so every record is a create
func (k *fakeKnot) recordsOf(did syntax.DID) []tapc.RecordEventData {
	var out []tapc.RecordEventData
	k.mu.Lock()
	defer k.mu.Unlock()
	prefix := did.String() + "|"
	for key, value := range k.records {
		rest, ok := strings.CutPrefix(key, prefix)
		if !ok {
			continue
		}
		collection, rkey, _ := strings.Cut(rest, "|")
		raw, err := json.Marshal(value)
		if err != nil {
			panic(err)
		}
		out = append(out, tapc.RecordEventData{Did: did, Collection: syntax.NSID(collection), Rkey: syntax.RecordKey(rkey), Rev: "rev", Action: tapc.RecordCreateAction, Record: raw})
	}
	return out
}

type node struct {
	t      *testing.T
	s      *Spindle
	cancel context.CancelFunc
	done   chan error
}

func (w *world) boot() *node {
	t := w.t
	t.Helper()
	connects := w.js.connects()
	ctx := log.IntoContext(context.Background(), slog.Default().With("test", t.Name()))
	d, err := db.Make(ctx, w.dbPath())
	if err != nil {
		t.Fatal(err)
	}
	admission, owner := w.admission, w.owner
	if admission == "" {
		admission = config.AdmissionOpen
	}
	if owner == "" {
		owner = testOwner
	}
	cfg := &config.Config{Role: config.RoleStandalone}
	cfg.Server = config.Server{
		ListenAddr:        "127.0.0.1:0",
		DBPath:            w.dbPath(),
		RepoDir:           filepath.Join(w.dir, "repos"),
		LogDir:            filepath.Join(w.dir, "logs"),
		Hostname:          faultHostname,
		JetstreamEndpoint: w.js.url(),
		PlcUrl:            w.plc.srv.URL,
		Hydrant:           config.Hydrant{Dir: w.hydDir},
		Dev:               true,
		Owner:             owner.String(),
		Admission:         admission,
	}
	cfg.ArtifactStores.Disk.Dir = filepath.Join(w.dir, "artifacts")

	s, err := New(ctx, cfg, d, map[string]models.Engine{}, nil)
	if err != nil {
		t.Fatalf("New: %v", err)
	}
	s.backfill.start = w.hyd.start
	s.backfill.poll = 10 * time.Millisecond
	s.backfill.retryBase = time.Millisecond
	s.retries.now = func() time.Time { return time.Now().Add(time.Duration(w.skew.Load())) }
	s.retries.tick = testRetryTick

	runCtx, cancel := context.WithCancel(ctx)
	n := &node{t: t, s: s, cancel: cancel, done: make(chan error, 1)}
	go func() { n.done <- s.Start(runCtx) }()
	t.Cleanup(n.stop)
	// a live connection only gets events sent after it connects, so wait for it
	eventually(t, 10*time.Second, "jetstream to connect", func() bool { return w.js.connects() > connects })
	return n
}

func (n *node) stop() {
	if n.cancel == nil {
		return
	}
	n.cancel()
	n.cancel = nil
	select {
	case err := <-n.done:
		if err != nil {
			n.t.Errorf("Start: %v", err)
		}
	case <-time.After(60 * time.Second):
		n.t.Fatal("spindle did not stop")
	}
}

func repoRows(t *testing.T, s *Spindle, repo syntax.DID) []db.Repo {
	t.Helper()
	rows, err := s.db.ReposByDid(repo)
	if err != nil {
		t.Fatal(err)
	}
	return rows
}

func isCollaborator(t *testing.T, s *Spindle, repo, who syntax.DID) bool {
	t.Helper()
	ok, err := s.e.IsRepoCollaborator(who.String(), rbac.ThisServer, repo.String())
	if err != nil {
		t.Fatal(err)
	}
	return ok
}

func secretCount(t *testing.T, s *Spindle, repo syntax.DID) int {
	t.Helper()
	got, err := s.vault.GetSecretsUnlocked(context.Background(), secrets.RepoIdentifier(repo.String()))
	if err != nil {
		t.Fatal(err)
	}
	return len(got)
}

func addTestSecret(t *testing.T, s *Spindle, repo syntax.DID) {
	t.Helper()
	mustAddSecret(t, s.vault, repo.String(), "api_key", "v", time.Now(), testOwner.String())
}

func (n *node) pipelines(repo syntax.DID) int {
	n.t.Helper()
	records, _, _, err := n.s.db.QueryPipelines(context.Background(), repo.String(), nil, "", nil, 100)
	if err != nil {
		n.t.Fatal(err)
	}
	return len(records)
}

// a restarted spindle works its backlog in the background, so poll
func eventually(t *testing.T, within time.Duration, what string, cond func() bool) {
	t.Helper()
	deadline := time.Now().Add(within)
	for !cond() {
		if time.Now().After(deadline) {
			t.Fatalf("timed out after %v waiting for %s", within, what)
		}
		time.Sleep(20 * time.Millisecond)
	}
}

func (w *world) quiet() {
	w.t.Helper()
	last := -1
	stable := 0
	for stable < 5 {
		time.Sleep(40 * time.Millisecond)
		count := w.traffic()
		if count == last {
			stable++
		} else {
			stable, last = 0, count
		}
	}
}

func (w *world) traffic() int { return w.knot.count() + w.other.count() }

func (w *world) setKnotDown(on bool) {
	w.knot.mu.Lock()
	defer w.knot.mu.Unlock()
	w.knot.failing = on
}

func (w *world) setBare404(on bool) {
	w.knot.mu.Lock()
	defer w.knot.mu.Unlock()
	w.knot.bare404 = on
}

func (w *world) setPDSFailing(on bool) {
	w.knot.mu.Lock()
	defer w.knot.mu.Unlock()
	w.knot.failCollection = ""
	if on {
		w.knot.failCollection = org_tangled.RepoDeclarationNSID
	}
}

func describeRows(rows []db.Repo) string {
	var out []string
	for _, r := range rows {
		out = append(out, fmt.Sprintf("%s/%s@%s", r.Owner, r.Rkey, r.Knot))
	}
	return "[" + strings.Join(out, " ") + "]"
}

// wait on this before a stop, or the cursor stays behind
func (n *node) consumed(us int64) {
	n.t.Helper()
	eventually(n.t, 15*time.Second, "spindle to finish the jetstream events", func() bool { return n.s.jc.LastSeenUs() > us })
}

// spindle reads an event before it acts on it. this waits until it has acted, or failed and scheduled a retry
func (n *node) applied(repo syntax.DID, us int64) {
	n.t.Helper()
	n.consumed(us)
	eventually(n.t, 15*time.Second, "the repo job to try its marks", func() bool {
		now := n.s.retries.clock()
		_, due, err := n.s.db.RepoRetryDue(repo, now)
		if err != nil || due {
			return false
		}
		marks, err := n.s.db.DueRecordMarksOf(repo, now)
		return err == nil && len(marks) == 0
	})
}

func (n *node) cursor() int64 {
	n.t.Helper()
	us, err := n.s.db.GetLastTimeUs()
	if errors.Is(err, sql.ErrNoRows) {
		return 0
	}
	if err != nil {
		n.t.Fatal(err)
	}
	return us
}