package beads

import (
	"context"
	"errors"
	"fmt"
	"net/url"
	"testing"
	"time"

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

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

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

// readyNow is the instant the aggregation's clock is pinned to. Nothing about it
// is special beyond being fixed: the TTL is a function of the fixture rather
// than of when the suite ran.
var readyNow = time.Date(2026, 8, 12, 12, 0, 0, 0, time.UTC)

// fakeReadyDB is one database behind the ReadyOpener seam, counting what each
// call actually read. The counters are the point: the head-hash gate is a claim
// about reads not happening, and only a counter can check it — a timing
// measurement would pass on a fast machine whatever the code did.
type fakeReadyDB struct {
	branches []browse.Branch
	tables   []browse.TableInfo
	rows     map[string]*browse.RowPage
	commits  []browse.CommitInfo

	openErr     error
	branchesErr error
	logErr      error

	opens      int
	closes     int
	rowReads   int
	tableReads int
	logReads   int
}

func (f *fakeReadyDB) Branches(context.Context) ([]browse.Branch, error) {
	if f.branchesErr != nil {
		return nil, f.branchesErr
	}
	return f.branches, nil
}

func (f *fakeReadyDB) Tables(context.Context, string) ([]browse.TableInfo, error) {
	f.tableReads++
	return f.tables, nil
}

func (f *fakeReadyDB) Log(_ context.Context, _, _ string, _ int) ([]browse.CommitInfo, string, error) {
	f.logReads++
	if f.logErr != nil {
		return nil, "", f.logErr
	}
	return f.commits, "", nil
}

func (f *fakeReadyDB) Rows(_ context.Context, _, table string, _, _ int) (*browse.RowPage, error) {
	f.rowReads++
	if p, ok := f.rows[table]; ok {
		return p, nil
	}
	return nil, fmt.Errorf("%w: %s", browse.ErrTableNotFound, table)
}

func (f *fakeReadyDB) Close() error { f.closes++; return nil }

// readyInstance is a set of databases addressed by repository id, plus the
// opener over them.
type readyInstance struct {
	dbs map[int]*fakeReadyDB
}

func (in *readyInstance) open(_ context.Context, d ReadyDatabase) (ReadySession, error) {
	f, ok := in.dbs[d.ID]
	if !ok {
		return nil, fmt.Errorf("no such database: %s", d.Slug())
	}
	if f.openErr != nil {
		return nil, f.openErr
	}
	f.opens++
	return f, nil
}

// readyTracker builds a beads database whose issues page carries the given rows.
// The column order puts is_blocked, is_template and ephemeral last so the
// projection's name-based (not positional) cell mapping is exercised.
func readyTracker(head string, issues [][]string, deps [][]string) *fakeReadyDB {
	return &fakeReadyDB{
		branches: []browse.Branch{{Name: "main", Head: head}},
		tables:   beadsTables(),
		commits: []browse.CommitInfo{{
			Hash: head, Author: "Eugene Blikh", Date: readyNow.Add(-4 * time.Minute),
			Message: "bd: create (auto-commit)",
		}},
		rows: map[string]*browse.RowPage{
			"issues": {
				Columns: []string{"id", "title", "status", "priority", "issue_type", "assignee", "is_blocked", "is_template", "ephemeral"},
				Rows:    issues,
				Total:   len(issues),
			},
			"dependencies": {
				Columns: []string{"id", "issue_id", "depends_on_issue_id", "type"},
				Rows:    deps,
				Total:   len(deps),
			},
		},
	}
}

// readyFixture is two trackers: one with three ready issues among six, one with
// a single ready issue.
func readyFixture() (*readyInstance, []ReadyDatabase) {
	alpha := readyTracker("h-alpha", [][]string{
		{"a-1", "Ready, top priority", "open", "0", "task", "alice", "0", "0", "0"},
		{"a-2", "Ready, lower", "open", "2", "bug", "bob", "0", "0", "0"},
		{"a-3", "Ready, same priority as a-2", "open", "2", "task", "alice", "0", "0", "0"},
		{"a-4", "In progress, not ready", "in_progress", "0", "task", "alice", "0", "0", "0"},
		{"a-5", "Blocked, not ready", "open", "0", "task", "bob", "1", "0", "0"},
		{"a-6", "Closed, not ready", "closed", "0", "task", "bob", "0", "0", "0"},
	}, nil)
	beta := readyTracker("h-beta", [][]string{
		{"b-1", "The one ready thing", "open", "1", "task", "carol", "0", "0", "0"},
		{"b-2", "A template scaffold", "open", "0", "task", "carol", "0", "1", "0"},
		{"b-3", "An ephemeral scratch", "open", "0", "task", "carol", "0", "0", "1"},
	}, nil)
	in := &readyInstance{dbs: map[int]*fakeReadyDB{1: alpha, 2: beta}}
	return in, []ReadyDatabase{
		{ID: 1, OwnerName: "alice", Name: "alpha"},
		{ID: 2, OwnerName: "bob", Name: "beta"},
	}
}

func across(in *readyInstance, dbs []ReadyDatabase, cache *ReadyCache, f ReadyFilter, now time.Time) *ReadyView {
	return ReadyAcross(context.Background(), dbs, in.open, cache, f, now)
}

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

// The projection: which issues are ready, how the groups are ordered, how the
// cards inside one are, and that each group carries its own database's head.
func TestReadyAcrossGroupsAndOrders(t *testing.T) {
	in, dbs := readyFixture()

	view := across(in, dbs, NewReadyCache(), ReadyFilter{}, readyNow)

	require.Len(t, view.Groups, 2)
	assert.Equal(t, 4, view.Total)
	assert.Equal(t, 2, view.Considered)
	assert.False(t, view.Capped)
	assert.Empty(t, view.Failed)

	// Groups: ready count desc, then name.
	assert.Equal(t, "alice/alpha", view.Groups[0].Database.Slug())
	assert.Equal(t, "bob/beta", view.Groups[1].Database.Slug())

	// Cards: priority, then id. In-progress, blocked, closed, template and
	// ephemeral issues are not ready.
	var ids []string
	for _, c := range view.Groups[0].Cards {
		ids = append(ids, c.ID)
		assert.True(t, c.Ready, "%s must be marked ready", c.ID)
	}
	assert.Equal(t, []string{"a-1", "a-2", "a-3"}, ids)
	require.Len(t, view.Groups[1].Cards, 1)
	assert.Equal(t, "b-1", view.Groups[1].Cards[0].ID)

	// Each group carries its own database's head, so "ready" from a store that
	// stopped receiving pushes is not claimed silently.
	require.NotNil(t, view.Groups[0].Head)
	assert.Equal(t, "h-alpha", view.Groups[0].Head.Hash)
	assert.Equal(t, "main", view.Groups[0].Ref)
	require.NotNil(t, view.Groups[1].Head)
	assert.Equal(t, "h-beta", view.Groups[1].Head.Hash)

	// The filter options come from the whole ready set.
	assert.Equal(t, []string{"alice", "bob", "carol"}, view.Options.Assignees)
	assert.Equal(t, []string{"0", "1", "2"}, view.Options.Priorities)

	// Every session opened is closed again: this page opens N stores, and one
	// leaked handle per request per database is the failure mode it 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)
	}
}

// A "blocks" edge to a still-open target blocks its source; the same edge to a
// closed one does not. This is the ready rule the board shares, exercised
// through the aggregation rather than restated in it.
func TestReadyAcrossHonoursOpenBlockers(t *testing.T) {
	tracker := readyTracker("h1", [][]string{
		{"i-1", "Blocked by an open issue", "open", "1", "task", "alice", "0", "0", "0"},
		{"i-2", "The open blocker", "open", "1", "task", "alice", "0", "0", "0"},
		{"i-3", "Blocked by a closed issue", "open", "1", "task", "alice", "0", "0", "0"},
		{"i-4", "The closed blocker", "closed", "1", "task", "alice", "0", "0", "0"},
		{"i-5", "A subtask of an open epic", "open", "1", "task", "alice", "0", "0", "0"},
		{"i-6", "The epic", "open", "1", "epic", "alice", "0", "0", "0"},
	}, [][]string{
		{"d1", "i-1", "i-2", "blocks"},
		{"d2", "i-3", "i-4", "blocks"},
		{"d3", "i-5", "i-6", "parent-child"},
	})
	in := &readyInstance{dbs: map[int]*fakeReadyDB{1: tracker}}
	dbs := []ReadyDatabase{{ID: 1, OwnerName: "alice", Name: "alpha"}}

	view := across(in, dbs, NewReadyCache(), ReadyFilter{}, readyNow)

	require.Len(t, view.Groups, 1)
	var ids []string
	for _, c := range view.Groups[0].Cards {
		ids = append(ids, c.ID)
	}
	// i-1 waits on an open issue; everything else is actionable — including the
	// subtask, because parent-child is hierarchy and not a blocker.
	assert.Equal(t, []string{"i-2", "i-3", "i-5", "i-6"}, ids)
}

// The head-hash gate: a second call with unmoved heads reads no rows, no tables
// and no log. Asserted by counting reads on the fake, never by timing.
func TestReadyAcrossHeadHashGateSkipsEveryRead(t *testing.T) {
	in, dbs := readyFixture()
	cache := NewReadyCache()

	first := across(in, dbs, cache, ReadyFilter{}, readyNow)
	require.Len(t, first.Groups, 2)
	alpha := in.dbs[1]
	firstReads := alpha.rowReads
	require.Greater(t, firstReads, 0, "the first call must read rows")

	second := across(in, dbs, cache, ReadyFilter{}, readyNow.Add(30*time.Second))

	assert.Equal(t, firstReads, alpha.rowReads, "a second call with an unmoved head must read no rows")
	assert.Equal(t, 1, alpha.tableReads, "nor list tables again")
	assert.Equal(t, 1, alpha.logReads, "nor read the log again")
	// 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, alpha.opens)
	assert.Equal(t, 2, alpha.closes)
	// And the answer is the same one.
	assert.Equal(t, first.Total, second.Total)
	require.Len(t, second.Groups, 2)
	assert.Equal(t, first.Groups[0].Cards, second.Groups[0].Cards)
	require.NotNil(t, second.Groups[0].Head)
	assert.Equal(t, "h-alpha", second.Groups[0].Head.Hash)
}

// A head that moved is a projection that no longer stands: the rows are read
// again even well inside the TTL.
func TestReadyAcrossMovedHeadForcesAReread(t *testing.T) {
	in, dbs := readyFixture()
	cache := NewReadyCache()
	alpha := in.dbs[1]

	across(in, dbs, cache, ReadyFilter{}, readyNow)
	firstReads := alpha.rowReads

	alpha.branches = []browse.Branch{{Name: "main", Head: "h-alpha-2"}}
	across(in, dbs, cache, ReadyFilter{}, readyNow.Add(time.Second))

	assert.Greater(t, alpha.rowReads, firstReads, "a moved head must be re-read")
	// The other database's head did not move, so it was not re-read.
	assert.Equal(t, 1, in.dbs[2].tableReads)
}

// The TTL: past it the projection is re-read even though the head has not
// moved, so a cache can never be the reason a reader sees yesterday's answer.
func TestReadyAcrossTTLExpiryForcesAReread(t *testing.T) {
	in, dbs := readyFixture()
	cache := NewReadyCache()
	alpha := in.dbs[1]

	across(in, dbs, cache, ReadyFilter{}, readyNow)
	firstReads := alpha.rowReads

	// One tick short of the TTL: still cached.
	across(in, dbs, cache, ReadyFilter{}, readyNow.Add(ReadyCacheTTL-time.Nanosecond))
	assert.Equal(t, firstReads, alpha.rowReads, "inside the TTL the projection stands")

	// At the TTL: read again.
	across(in, dbs, cache, ReadyFilter{}, readyNow.Add(ReadyCacheTTL))
	assert.Greater(t, alpha.rowReads, firstReads, "at the TTL the projection is re-read")
}

// The ceiling: more candidates than ReadyMaxDatabases and the view says it was
// capped, having opened exactly the ceiling's worth of stores.
func TestReadyAcrossCeiling(t *testing.T) {
	in := &readyInstance{dbs: map[int]*fakeReadyDB{}}
	var dbs []ReadyDatabase
	for i := 1; i <= ReadyMaxDatabases+3; i++ {
		in.dbs[i] = readyTracker(fmt.Sprintf("h%d", i), [][]string{
			{fmt.Sprintf("t%d-1", i), "Ready", "open", "1", "task", "alice", "0", "0", "0"},
		}, nil)
		dbs = append(dbs, ReadyDatabase{ID: i, OwnerName: "alice", Name: fmt.Sprintf("db%02d", i)})
	}

	view := across(in, dbs, NewReadyCache(), ReadyFilter{}, readyNow)

	assert.True(t, view.Capped, "a page that hit the ceiling must say so")
	assert.Equal(t, ReadyMaxDatabases, view.Considered)
	assert.Equal(t, ReadyMaxDatabases, view.Max)
	assert.Len(t, view.Groups, ReadyMaxDatabases)
	// The three past the ceiling were never opened.
	for i := ReadyMaxDatabases + 1; i <= ReadyMaxDatabases+3; i++ {
		assert.Zero(t, in.dbs[i].opens, "database %d is past the ceiling", i)
	}
}

// A database named by ?db= is never dropped by the ceiling: the filter narrows
// the candidates first.
func TestReadyAcrossCeilingAppliesAfterTheDatabaseFilter(t *testing.T) {
	in := &readyInstance{dbs: map[int]*fakeReadyDB{}}
	var dbs []ReadyDatabase
	for i := 1; i <= ReadyMaxDatabases+3; i++ {
		in.dbs[i] = readyTracker(fmt.Sprintf("h%d", i), [][]string{
			{fmt.Sprintf("t%d-1", i), "Ready", "open", "1", "task", "alice", "0", "0", "0"},
		}, nil)
		dbs = append(dbs, ReadyDatabase{ID: i, OwnerName: "alice", Name: fmt.Sprintf("db%02d", i)})
	}
	last := dbs[len(dbs)-1]

	view := across(in, dbs, NewReadyCache(), ReadyFilter{Databases: []string{last.Slug()}}, readyNow)

	assert.False(t, view.Capped)
	assert.Equal(t, 1, view.Considered)
	require.Len(t, view.Groups, 1)
	assert.Equal(t, last.Slug(), view.Groups[0].Database.Slug())
}

// A database that cannot be opened, or whose rows cannot be read, costs itself
// only. The error is carried out for the caller's log and the rest of the answer
// stands.
func TestReadyAcrossFailingDatabaseCostsItselfOnly(t *testing.T) {
	in, dbs := readyFixture()
	in.dbs[1].openErr = errors.New("browse: open store: no such file or directory")

	view := across(in, dbs, NewReadyCache(), ReadyFilter{}, readyNow)

	require.Len(t, view.Failed, 1)
	assert.Equal(t, "alice/alpha", view.Failed[0].Database.Slug())
	assert.ErrorContains(t, view.Failed[0].Err, "no such file")
	require.Len(t, view.Groups, 1)
	assert.Equal(t, "bob/beta", view.Groups[0].Database.Slug())

	// A store that opens but cannot list its branches fails the same way.
	in2, dbs2 := readyFixture()
	in2.dbs[1].branchesErr = errors.New("browse: list branches: corrupt chunk")
	view2 := across(in2, dbs2, NewReadyCache(), ReadyFilter{}, readyNow)
	require.Len(t, view2.Failed, 1)
	require.Len(t, view2.Groups, 1)
	assert.Equal(t, 1, in2.dbs[1].closes, "a failing database still closes its session")
}

// A database whose tables are not a beads schema is skipped silently: no group,
// no failure, and no second table listing on the next request.
func TestReadyAcrossSkipsNonBeadsDatabases(t *testing.T) {
	in, dbs := readyFixture()
	plain := &fakeReadyDB{
		branches: []browse.Branch{{Name: "main", Head: "h-plain"}},
		tables:   []browse.TableInfo{{Name: "measurements", Columns: []browse.ColumnInfo{{Name: "id"}}}},
	}
	in.dbs[3] = plain
	dbs = append(dbs, ReadyDatabase{ID: 3, OwnerName: "carol", Name: "sensors"})
	cache := NewReadyCache()

	view := across(in, dbs, cache, ReadyFilter{}, readyNow)
	assert.Len(t, view.Groups, 2)
	assert.Empty(t, view.Failed)
	assert.Zero(t, plain.rowReads, "a non-beads database is never read for rows")

	across(in, dbs, cache, ReadyFilter{}, readyNow.Add(time.Second))
	assert.Equal(t, 1, plain.tableReads, "the fingerprint is worth one table listing")
}

// A store with no branches at all — freshly initialised, never pushed to — is
// skipped rather than failing the page.
func TestReadyAcrossSkipsAnEmptyStore(t *testing.T) {
	in := &readyInstance{dbs: map[int]*fakeReadyDB{1: {}}}
	dbs := []ReadyDatabase{{ID: 1, OwnerName: "alice", Name: "fresh"}}

	view := across(in, dbs, NewReadyCache(), ReadyFilter{}, readyNow)

	assert.Empty(t, view.Groups)
	assert.Empty(t, view.Failed)
	assert.Equal(t, 1, view.Considered)
}

// A log that cannot be read costs the group its freshness line and nothing else.
func TestReadyAcrossSurvivesAnUnreadableLog(t *testing.T) {
	in, dbs := readyFixture()
	in.dbs[1].logErr = errors.New("browse: walk commits: corrupt chunk")

	view := across(in, dbs, NewReadyCache(), ReadyFilter{}, readyNow)

	require.Len(t, view.Groups, 2)
	assert.Nil(t, view.Groups[0].Head, "no head is renderable")
	assert.Len(t, view.Groups[0].Cards, 3, "the ready set is still answered")
}

func TestReadyFilters(t *testing.T) {
	in, dbs := readyFixture()

	t.Run("q narrows over id and title", func(t *testing.T) {
		view := across(in, dbs, NewReadyCache(), ReadyFilter{Query: "TOP priority"}, readyNow)
		require.Len(t, view.Groups, 1)
		require.Len(t, view.Groups[0].Cards, 1)
		assert.Equal(t, "a-1", view.Groups[0].Cards[0].ID)
		assert.Equal(t, 1, view.Total)
	})

	t.Run("assignee is exact", func(t *testing.T) {
		view := across(in, dbs, NewReadyCache(), ReadyFilter{Assignee: "alice"}, readyNow)
		require.Len(t, view.Groups, 1, "only alpha has alice's ready work")
		assert.Equal(t, "alice/alpha", view.Groups[0].Database.Slug())
		assert.Equal(t, 2, view.Total)
	})

	t.Run("priority is exact", func(t *testing.T) {
		view := across(in, dbs, NewReadyCache(), ReadyFilter{Priority: "2"}, readyNow)
		require.Len(t, view.Groups, 1)
		assert.Equal(t, 2, view.Total)
	})

	t.Run("db selects databases", func(t *testing.T) {
		// Its own fixture: this one asserts that a database was never opened, and
		// the shared one has been opened by every subtest above.
		in, dbs := readyFixture()
		view := across(in, dbs, NewReadyCache(), ReadyFilter{Databases: []string{"bob/beta"}}, readyNow)
		require.Len(t, view.Groups, 1)
		assert.Equal(t, "bob/beta", view.Groups[0].Database.Slug())
		assert.Equal(t, 1, view.Considered, "the other database is never opened")
		assert.Zero(t, in.dbs[1].opens)
	})

	t.Run("a filter that matches nothing empties the page", func(t *testing.T) {
		view := across(in, dbs, NewReadyCache(), ReadyFilter{Assignee: "nobody"}, readyNow)
		assert.Empty(t, view.Groups)
		assert.Zero(t, view.Total)
		// The options still list the real values, so the reader can pick another.
		assert.Equal(t, []string{"alice", "bob", "carol"}, view.Options.Assignees)
	})
}

func TestParseReadyFilter(t *testing.T) {
	q, err := url.ParseQuery("q=+cache+&assignee=alice&priority=1&db=alice%2Falpha&db=+&db=bob%2Fbeta")
	require.NoError(t, err)

	f := ParseReadyFilter(q)

	assert.Equal(t, "cache", f.Query)
	assert.Equal(t, "alice", f.Assignee)
	assert.Equal(t, "1", f.Priority)
	assert.Equal(t, []string{"alice/alpha", "bob/beta"}, f.Databases, "a blank ?db= is not a database")
	assert.True(t, f.Active())
	assert.False(t, ParseReadyFilter(url.Values{}).Active())
}

// The cache is bounded by entry count: it is a cache and not a store, and an
// instance with more databases than the ceiling must not grow it without end.
func TestReadyCacheIsBoundedByEntryCount(t *testing.T) {
	cache := NewReadyCache()
	for i := 1; i <= readyCacheMaxEntries+10; i++ {
		cache.store(i, readyEntry{head: fmt.Sprintf("h%d", i), at: readyNow, beads: true})
	}
	cache.mu.Lock()
	n := len(cache.entries)
	cache.mu.Unlock()
	assert.LessOrEqual(t, n, readyCacheMaxEntries)
}