package beads

import (
	"errors"
	"fmt"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"

	"sourcecraft.dev/bigbes/sr-ht-dolt/browse"
)

// --- fixtures ----------------------------------------------------------------

// prefixTracker is a beads database whose config names an issue prefix, with the
// settings a real tracker carries beside it — the projection has to pick one row
// out of the table rather than read the first one.
func prefixTracker(head, prefix string) *fakeReadyDB {
	return &fakeReadyDB{
		branches: []browse.Branch{{Name: "main", Head: head}},
		tables:   beadsTables(),
		rows: map[string]*browse.RowPage{
			"config": {
				Columns: []string{"key", "value"},
				Rows: [][]string{
					{"compact_tier2_days", "30"},
					{"kv.memory.handoff", "not a prefix"},
					{"issue_prefix", prefix},
				},
				Total: 3,
			},
		},
	}
}

// prefixFixture is two trackers, "global" and "artifacts", as the instance has
// them: one cross-project tracker and one per-project.
func prefixFixture() (*readyInstance, []ReadyDatabase) {
	in := &readyInstance{dbs: map[int]*fakeReadyDB{
		1: prefixTracker("h-global", "global"),
		2: prefixTracker("h-artifacts", "artifacts"),
	}}
	return in, []ReadyDatabase{
		{ID: 1, OwnerName: "bigbes", Name: "beads-global"},
		{ID: 2, OwnerName: "bigbes", Name: "sourcehut-artifacts"},
	}
}

// --- tests -------------------------------------------------------------------

// The index: one prefix per database, read out of config, and nothing claimed
// for a prefix no database named.
func TestPrefixesAcrossIndexesTheDatabasesItIsHanded(t *testing.T) {
	in, dbs := prefixFixture()

	index := PrefixesAcross(t.Context(), dbs, in.open, &PrefixCache{}, readyNow)

	assert.Equal(t, []string{"artifacts", "global"}, index.Prefixes())
	d, ok := index.Lookup("global")
	require.True(t, ok)
	assert.Equal(t, "bigbes/beads-global", d.Slug())
	d, ok = index.Lookup("artifacts")
	require.True(t, ok)
	assert.Equal(t, "bigbes/sourcehut-artifacts", d.Slug())
	_, ok = index.Lookup("nosuch")
	assert.False(t, ok, "a prefix no database named is not in the index")
	assert.Empty(t, index.Failed)

	// Every session opened is closed again: this is the read pattern the
	// per-request browse discipline is bounded against.
	for id, f := range in.dbs {
		assert.Equal(t, f.opens, f.closes, "database %d: %d opens, %d closes", id, f.opens, f.closes)
	}
}

// The head-hash gate: a second build with unmoved heads reads no rows. Asserted
// by counting reads on the fake, never by timing.
func TestPrefixesAcrossHeadHashGateSkipsTheRead(t *testing.T) {
	in, dbs := prefixFixture()
	cache := &PrefixCache{}

	first := PrefixesAcross(t.Context(), dbs, in.open, cache, readyNow)
	require.Len(t, first.Prefixes(), 2)
	global := in.dbs[1]
	require.Greater(t, global.rowReads, 0, "the first build must read config")
	reads := global.rowReads

	second := PrefixesAcross(t.Context(), dbs, in.open, cache, readyNow.Add(30*time.Second))

	assert.Equal(t, reads, global.rowReads, "a second build with an unmoved head must read no rows")
	assert.Equal(t, first.Prefixes(), second.Prefixes(), "and it is the same answer")
	// The session is still opened and its branches listed — that is what the gate
	// is gated on, and it is the cheap half.
	assert.Equal(t, 2, global.opens)
	assert.Equal(t, 2, global.closes)
}

// A head that moved is a prefix that may have been renamed: read again, even
// well inside the TTL.
func TestPrefixesAcrossMovedHeadForcesAReread(t *testing.T) {
	in, dbs := prefixFixture()
	cache := &PrefixCache{}
	global := in.dbs[1]

	PrefixesAcross(t.Context(), dbs, in.open, cache, readyNow)
	reads := global.rowReads

	global.branches = []browse.Branch{{Name: "main", Head: "h-global-2"}}
	PrefixesAcross(t.Context(), dbs, in.open, cache, readyNow.Add(time.Second))

	assert.Greater(t, global.rowReads, reads, "a moved head must be re-read")
	assert.Equal(t, 1, in.dbs[2].rowReads, "the sibling's head did not move")
}

// The TTL is the same one /ready is bounded by, because it is the same cache.
func TestPrefixesAcrossTTLExpiryForcesAReread(t *testing.T) {
	in, dbs := prefixFixture()
	cache := &PrefixCache{}
	global := in.dbs[1]

	PrefixesAcross(t.Context(), dbs, in.open, cache, readyNow)
	reads := global.rowReads

	PrefixesAcross(t.Context(), dbs, in.open, cache, readyNow.Add(ReadyCacheTTL-time.Nanosecond))
	assert.Equal(t, reads, global.rowReads, "inside the TTL the projection stands")

	PrefixesAcross(t.Context(), dbs, in.open, cache, readyNow.Add(ReadyCacheTTL))
	assert.Greater(t, global.rowReads, reads, "at the TTL the projection is re-read")
}

// The ceiling: at most ReadyMaxDatabases stores opened per build, and the first
// database listed — the caller puts the page's own database there — is inside
// it.
func TestPrefixesAcrossCeiling(t *testing.T) {
	in := &readyInstance{dbs: map[int]*fakeReadyDB{}}
	var dbs []ReadyDatabase
	for i := 1; i <= ReadyMaxDatabases+3; i++ {
		in.dbs[i] = prefixTracker(fmt.Sprintf("h%d", i), fmt.Sprintf("p%02d", i))
		dbs = append(dbs, ReadyDatabase{ID: i, OwnerName: "alice", Name: fmt.Sprintf("db%02d", i)})
	}

	index := PrefixesAcross(t.Context(), dbs, in.open, &PrefixCache{}, readyNow)

	assert.Len(t, index.Prefixes(), ReadyMaxDatabases)
	_, ok := index.Lookup("p01")
	assert.True(t, ok, "the first database listed is never dropped by the ceiling")
	for i := ReadyMaxDatabases + 1; i <= ReadyMaxDatabases+3; i++ {
		assert.Zero(t, in.dbs[i].opens, "database %d is past the ceiling", i)
	}
}

// Two databases claiming one prefix: neither is linked. Linking to one of two
// candidates would be a guess rendered as a fact, and the id reads as text
// either way.
func TestPrefixesAcrossDropsAnAmbiguousPrefix(t *testing.T) {
	in := &readyInstance{dbs: map[int]*fakeReadyDB{
		1: prefixTracker("h1", "shared"),
		2: prefixTracker("h2", "shared"),
		3: prefixTracker("h3", "mine"),
	}}
	dbs := []ReadyDatabase{
		{ID: 1, OwnerName: "alice", Name: "one"},
		{ID: 2, OwnerName: "bob", Name: "two"},
		{ID: 3, OwnerName: "carol", Name: "three"},
	}

	index := PrefixesAcross(t.Context(), dbs, in.open, &PrefixCache{}, readyNow)

	_, ok := index.Lookup("shared")
	assert.False(t, ok, "a prefix two databases claim belongs to neither")
	assert.Equal(t, []string{"mine"}, index.Prefixes(), "the unambiguous ones are unaffected")
}

// A database that names no prefix — no config table at all, or a config without
// the row — is simply not in the index.
func TestPrefixesAcrossSkipsADatabaseWithNoPrefix(t *testing.T) {
	noConfig := &fakeReadyDB{branches: []browse.Branch{{Name: "main", Head: "h1"}}, tables: beadsTables()}
	noRow := &fakeReadyDB{
		branches: []browse.Branch{{Name: "main", Head: "h2"}},
		tables:   beadsTables(),
		rows: map[string]*browse.RowPage{
			"config": {Columns: []string{"key", "value"}, Rows: [][]string{{"compact_tier2_days", "30"}}, Total: 1},
		},
	}
	empty := &fakeReadyDB{} // a store with no branches: never pushed to
	in := &readyInstance{dbs: map[int]*fakeReadyDB{1: noConfig, 2: noRow, 3: empty}}
	dbs := []ReadyDatabase{
		{ID: 1, OwnerName: "alice", Name: "plain"},
		{ID: 2, OwnerName: "alice", Name: "settings-only"},
		{ID: 3, OwnerName: "alice", Name: "fresh"},
	}

	index := PrefixesAcross(t.Context(), dbs, in.open, &PrefixCache{}, readyNow)

	assert.Empty(t, index.Prefixes())
	assert.Empty(t, index.Failed, "naming no prefix is not a failure")
	assert.Zero(t, empty.rowReads, "a store with no branches is not read")
}

// A database that cannot be opened, or whose config cannot be read, costs the
// ids it owns their link and nothing else.
func TestPrefixesAcrossFailingDatabaseCostsItselfOnly(t *testing.T) {
	in, dbs := prefixFixture()
	in.dbs[1].openErr = errors.New("browse: open store: no such file or directory")

	index := PrefixesAcross(t.Context(), dbs, in.open, &PrefixCache{}, readyNow)

	require.Len(t, index.Failed, 1)
	assert.Equal(t, "bigbes/beads-global", index.Failed[0].Database.Slug())
	assert.ErrorContains(t, index.Failed[0].Err, "no such file")
	assert.Equal(t, []string{"artifacts"}, index.Prefixes())

	// A store that opens but cannot list its branches fails the same way, and
	// still closes its session.
	in2, dbs2 := prefixFixture()
	in2.dbs[1].branchesErr = errors.New("browse: list branches: corrupt chunk")
	index2 := PrefixesAcross(t.Context(), dbs2, in2.open, &PrefixCache{}, readyNow)
	require.Len(t, index2.Failed, 1)
	assert.Equal(t, []string{"artifacts"}, index2.Prefixes())
	assert.Equal(t, 1, in2.dbs[1].closes)
}

// The pattern: what counts as an id and what does not. The suffix carries no
// hyphen, which is what makes the split unambiguous for a hyphenated prefix.
func TestPrefixIndexScan(t *testing.T) {
	index := &PrefixIndex{byPrefix: map[string]ReadyDatabase{
		"global":     {ID: 1, OwnerName: "bigbes", Name: "beads-global"},
		"artifacts":  {ID: 2, OwnerName: "bigbes", Name: "sourcehut-artifacts"},
		"sr-ht-dolt": {ID: 3, OwnerName: "bigbes", Name: "sourcehut-dolt"},
	}}

	for _, tc := range []struct {
		name string
		text string
		want []string
	}{
		{"a plain id", "blocked by artifacts-nex", []string{"artifacts-nex"}},
		{"the subtask form", "see artifacts-46c.2 for the bucket", []string{"artifacts-46c.2"}},
		{"a hyphenated prefix", "sr-ht-dolt-44n.6 lands first", []string{"sr-ht-dolt-44n.6"}},
		{"several in one line", "global-b08 needs artifacts-46c.2", []string{"global-b08", "artifacts-46c.2"}},
		{"an unknown prefix", "nosuch-46c and other-1 are not ids", nil},
		{"a hyphenated word", "the read-only surface", nil},
		{"an id at the end of a sentence", "closed by artifacts-46c.", []string{"artifacts-46c"}},
		{"an id in brackets", "(artifacts-46c) and [global-b08]", []string{"artifacts-46c", "global-b08"}},
		{"an id inside a longer word", "xxartifacts-46c and artifactsx-1", nil},
		{"uppercase is not an id", "ARTIFACTS-46C", nil},
		{"a prefix on its own", "the artifacts tracker", nil},
		{"a shorter known prefix inside a longer token", "some-global-b08", nil},
		{"empty text", "", nil},
	} {
		t.Run(tc.name, func(t *testing.T) {
			var got []string
			for _, ref := range index.Scan(tc.text) {
				got = append(got, ref.ID)
				assert.Equal(t, ref.ID, tc.text[ref.Start:ref.End],
					"the offsets must name the id they carry")
			}
			assert.Equal(t, tc.want, got)
		})
	}

	// Each reference names the database that owns it.
	refs := index.Scan("global-b08 blocks artifacts-46c.2")
	require.Len(t, refs, 2)
	assert.Equal(t, "bigbes/beads-global", refs[0].Database.Slug())
	assert.Equal(t, "bigbes/sourcehut-artifacts", refs[1].Database.Slug())
}

// An index that was never built knows no prefix — the same answer an unknown one
// gets, which is what keeps "the caller may not see that database" and "there is
// no such database" indistinguishable.
func TestNilPrefixIndexKnowsNothing(t *testing.T) {
	var index *PrefixIndex

	assert.Nil(t, index.Scan("artifacts-46c.2"))
	assert.Nil(t, index.Prefixes())
	_, ok := index.Lookup("artifacts")
	assert.False(t, ok)

	// So does an empty one.
	empty := PrefixesAcross(t.Context(), nil, nil, &PrefixCache{}, readyNow)
	assert.Nil(t, empty.Scan("artifacts-46c.2"))
}