package beads

import (
	"context"
	"fmt"
	"testing"
	"time"

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

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

// --- the three cross-cutting projections, read past the cap -------------------
//
// The seam is truncation_test.go's clippingSession: it honours offset/limit and
// reports the table's true row count, which is what a real read at limit=Max
// does. Everything here builds on it rather than lowering Max or hand-writing a
// page whose Total merely disagrees with its own rows — the row that was left
// behind has to be genuinely absent from what the projection is handed, or the
// test asserts a flag without ever producing the situation the flag is about.

// clippingReadyDB is a ReadySession over that seam: the clipping row reads, plus
// the branch list, table list and log the cross-database readings need. It
// counts row reads, because the head-hash gate is a claim about reads not
// happening and only a counter can check it.
type clippingReadyDB struct {
	*clippingSession
	branches []browse.Branch
	tables   []browse.TableInfo
	commits  []browse.CommitInfo

	opens    int
	closes   int
	rowReads int
}

func (f *clippingReadyDB) Rows(ctx context.Context, ref, table string, offset, limit int) (*browse.RowPage, error) {
	f.rowReads++
	return f.clippingSession.Rows(ctx, ref, table, offset, limit)
}

func (f *clippingReadyDB) Branches(context.Context) ([]browse.Branch, error) {
	return f.branches, nil
}

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

func (f *clippingReadyDB) Log(_ context.Context, _, _ string, _ int) ([]browse.CommitInfo, string, error) {
	return f.commits, "", nil
}

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

// clippingTracker is one beads database at head, over the given tables.
func clippingTracker(head string, tables map[string]*browse.RowPage) *clippingReadyDB {
	return &clippingReadyDB{
		clippingSession: &clippingSession{rowsByTable: tables},
		branches:        []browse.Branch{{Name: "main", Head: head}},
		tables:          beadsTables(),
		commits: []browse.CommitInfo{{
			Hash: head, Author: "bigbes", Date: readyNow.Add(-4 * time.Minute),
			Message: "bd: create (auto-commit)",
		}},
	}
}

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

func (in *clippingInstance) 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())
	}
	f.opens++
	return f, nil
}

// issueColumns is manyIssues' column order, needed by the fixtures below that
// build their own issue rows.
var issueColumns = []string{
	"id", "title", "status", "priority", "issue_type", "assignee", "created_at", "is_blocked",
}

// emptyDeps is a present but empty dependencies table. The table is required —
// a ready projection that cannot read it fails rather than answering — so every
// tracker here carries one.
func emptyDeps() *browse.RowPage {
	return &browse.RowPage{Columns: []string{"id", "issue_id", "depends_on_issue_id", "type"}}
}

// readyTables is the two tables a ready projection requires, plus the issues
// page under test.
func readyTables(issues *browse.RowPage) map[string]*browse.RowPage {
	return map[string]*browse.RowPage{"issues": issues, "dependencies": emptyDeps()}
}

// tailReadyIssues builds n issue rows in which only the rows from index `from`
// on are open; everything before it is closed. With from == Max the tracker's
// entire ready set sits past the cap, so a clipped read finds nothing ready —
// and a database with nothing ready is absent from the groups, which is exactly
// the case a per-group flag cannot report.
func tailReadyIssues(n, from int) *browse.RowPage {
	rows := make([][]string, 0, n)
	for i := 0; i < n; i++ {
		status := "closed"
		if i >= from {
			status = "open"
		}
		rows = append(rows, []string{
			issueID(i), fmt.Sprintf("Issue %d", i), status,
			"1", "task", "alice", "2024-01-01 00:00:00", "0",
		})
	}
	return &browse.RowPage{Columns: issueColumns, Rows: rows, Total: n}
}

// groupFor returns the group for a database slug, failing the test when the
// answer carries none.
func groupFor(t *testing.T, view *ReadyView, slug string) ReadyGroup {
	t.Helper()
	for _, g := range view.Groups {
		if g.Database.Slug() == slug {
			return g
		}
	}
	require.FailNowf(t, "no group", "the answer carries no group for %s", slug)
	return ReadyGroup{}
}

// --- the cross-database ready set ---------------------------------------------

func TestReadyAcrossReportsAClippedRead(t *testing.T) {
	in := &clippingInstance{dbs: map[int]*clippingReadyDB{
		1: clippingTracker("h-alpha", readyTables(manyIssues(Max+5))),
	}}
	dbs := []ReadyDatabase{{ID: 1, OwnerName: "alice", Name: "alpha"}}

	view := ReadyAcross(t.Context(), dbs, in.open, NewReadyCache(), ReadyFilter{}, readyNow)

	require.Len(t, view.Truncated, 1, "the database's read was partial and the answer must say so")
	assert.Equal(t, "alice/alpha", view.Truncated[0].Database.Slug())
	assert.Equal(t, Max+5, view.Truncated[0].ShownOf, "the tracker's true issue count, not the number read")
	assert.True(t, view.Truncated[0].IssuesClipped())

	// The group carries the same fact, so a rendered group needs no lookup.
	g := groupFor(t, view, "alice/alpha")
	assert.True(t, g.Truncated)
	assert.True(t, g.IssuesClipped())
	assert.Equal(t, Max+5, g.ShownOf)

	// And the other bound is untouched: one database is not a ceiling on
	// databases.
	assert.False(t, view.Capped)
	assert.Equal(t, 1, view.Considered)
}

func TestReadyAcrossOnACompleteReadReportsNoClip(t *testing.T) {
	in := &clippingInstance{dbs: map[int]*clippingReadyDB{
		1: clippingTracker("h-alpha", readyTables(manyIssues(9))),
	}}
	dbs := []ReadyDatabase{{ID: 1, OwnerName: "alice", Name: "alpha"}}

	view := ReadyAcross(t.Context(), dbs, in.open, NewReadyCache(), ReadyFilter{}, readyNow)

	assert.Empty(t, view.Truncated)
	g := groupFor(t, view, "alice/alpha")
	assert.False(t, g.Truncated)
	assert.False(t, g.IssuesClipped())
	assert.Equal(t, 9, g.ShownOf, "nothing was left behind, so this is what was read")
	assert.Len(t, g.Cards, 3, "one issue in three is open in this fixture")
}

// The attribution: two databases, one clipped. The answer names that one and
// says nothing about the other — "some read somewhere was partial" is not
// actionable on an instance with sixty trackers.
func TestReadyAcrossAttributesTheClipToItsDatabase(t *testing.T) {
	in := &clippingInstance{dbs: map[int]*clippingReadyDB{
		1: clippingTracker("h-alpha", readyTables(manyIssues(Max+5))),
		2: clippingTracker("h-beta", readyTables(manyIssues(6))),
	}}
	dbs := []ReadyDatabase{
		{ID: 1, OwnerName: "alice", Name: "alpha"},
		{ID: 2, OwnerName: "bob", Name: "beta"},
	}

	view := ReadyAcross(t.Context(), dbs, in.open, NewReadyCache(), ReadyFilter{}, readyNow)

	require.Len(t, view.Groups, 2, "both databases have ready work")
	require.Len(t, view.Truncated, 1, "only one of the two was read in part")
	assert.Equal(t, "alice/alpha", view.Truncated[0].Database.Slug())
	assert.Equal(t, Max+5, view.Truncated[0].ShownOf)

	assert.True(t, groupFor(t, view, "alice/alpha").Truncated)
	beta := groupFor(t, view, "bob/beta")
	assert.False(t, beta.Truncated, "the sibling's read was complete")
	assert.Equal(t, 6, beta.ShownOf)
}

// The case the group cannot carry: every ready issue sits past the cap, so the
// database has no group at all. Without the view's list, a tracker missing its
// ready work is indistinguishable from a tracker with none.
func TestReadyAcrossReportsAClippedDatabaseThatProducedNoGroup(t *testing.T) {
	in := &clippingInstance{dbs: map[int]*clippingReadyDB{
		1: clippingTracker("h-alpha", readyTables(tailReadyIssues(Max+5, Max))),
	}}
	dbs := []ReadyDatabase{{ID: 1, OwnerName: "alice", Name: "alpha"}}

	view := ReadyAcross(t.Context(), dbs, in.open, NewReadyCache(), ReadyFilter{}, readyNow)

	assert.Empty(t, view.Groups, "nothing in the rows that were read is ready")
	assert.Zero(t, view.Total)
	assert.Empty(t, view.Failed)
	require.Len(t, view.Truncated, 1, "and the reason the database is absent is on the record")
	assert.Equal(t, "alice/alpha", view.Truncated[0].Database.Slug())
	assert.Equal(t, Max+5, view.Truncated[0].ShownOf)
}

// A clip in one of the other tables the ready rule is computed from: the cards
// are all there, but the verdict on them was reached over a partial read — a
// blocking edge past the cap is a card called ready that is not.
func TestReadyAcrossReportsAClippedDependenciesRead(t *testing.T) {
	deps := emptyDeps()
	for i := 0; i < Max+1; i++ {
		deps.Rows = append(deps.Rows, []string{fmt.Sprintf("d%d", i), "i-0002", "i-0000", "blocks"})
	}
	in := &clippingInstance{dbs: map[int]*clippingReadyDB{
		1: clippingTracker("h-alpha", map[string]*browse.RowPage{
			"issues": manyIssues(9), "dependencies": deps,
		}),
	}}
	dbs := []ReadyDatabase{{ID: 1, OwnerName: "alice", Name: "alpha"}}

	view := ReadyAcross(t.Context(), dbs, in.open, NewReadyCache(), ReadyFilter{}, readyNow)

	require.Len(t, view.Truncated, 1)
	assert.Equal(t, 9, view.Truncated[0].ShownOf)
	assert.False(t, view.Truncated[0].IssuesClipped(), "the issue set itself is whole")
	g := groupFor(t, view, "alice/alpha")
	assert.True(t, g.Truncated, "dependencies decide which of those issues are ready")
	assert.False(t, g.IssuesClipped())
}

// The clip is cached with the cards. A second call under an unmoved head reads
// no row it could learn this from, so an uncached flag would quietly become
// false on the second request.
func TestReadyAcrossClipSurvivesTheCache(t *testing.T) {
	in := &clippingInstance{dbs: map[int]*clippingReadyDB{
		1: clippingTracker("h-alpha", readyTables(manyIssues(Max+5))),
	}}
	dbs := []ReadyDatabase{{ID: 1, OwnerName: "alice", Name: "alpha"}}
	cache := NewReadyCache()

	first := ReadyAcross(t.Context(), dbs, in.open, cache, ReadyFilter{}, readyNow)
	require.Len(t, first.Truncated, 1)
	reads := in.dbs[1].rowReads
	require.Greater(t, reads, 0, "the first call must read rows")

	second := ReadyAcross(t.Context(), dbs, in.open, cache, ReadyFilter{}, readyNow.Add(30*time.Second))

	assert.Equal(t, reads, in.dbs[1].rowReads, "a second call with an unmoved head must read no rows")
	require.Len(t, second.Truncated, 1, "and must still report the clipped read")
	assert.Equal(t, Max+5, second.Truncated[0].ShownOf)
	assert.True(t, groupFor(t, second, "alice/alpha").Truncated)
}

// --- the memory list ----------------------------------------------------------

// clippingHistory is a MemorySession whose config reads clip through the same
// seam: one clippingSession per ref, plus the table hashes and the log the
// revision walk runs on.
type clippingHistory struct {
	refs    map[string]*clippingSession // ref → the tables there
	hashes  map[string]string           // ref → config's content hash; "" = table absent
	commits []browse.CommitInfo         // newest first, as Log returns them
}

func (f *clippingHistory) Rows(ctx context.Context, ref, table string, offset, limit int) (*browse.RowPage, error) {
	s, ok := f.refs[ref]
	if !ok {
		return nil, fmt.Errorf("%w: %s", browse.ErrTableNotFound, table)
	}
	return s.Rows(ctx, ref, table, offset, limit)
}

func (f *clippingHistory) Log(_ context.Context, _, _ string, limit int) ([]browse.CommitInfo, string, error) {
	if limit < len(f.commits) {
		return f.commits[:limit], f.commits[limit].Hash, nil
	}
	return f.commits, "", nil
}

func (f *clippingHistory) TableHash(_ context.Context, ref, table string) (string, bool, error) {
	if table != memoryTable {
		return "", false, nil
	}
	h, ok := f.hashes[ref]
	if !ok || h == "" {
		return "", false, nil
	}
	return h, true, nil
}

// configPage builds a config table out of key/value pairs, in the key order it
// is given — a store returns them ordered by key, and which rows fall past the
// cap is a function of that order.
func configPage(rows [][]string) *browse.RowPage {
	return &browse.RowPage{Columns: []string{"key", "value"}, Rows: rows, Total: len(rows)}
}

// manyMemoryRows builds n memory rows, keys kv.memory.m0000… in order.
func manyMemoryRows(n int) [][]string {
	rows := make([][]string, 0, n)
	for i := 0; i < n; i++ {
		rows = append(rows, []string{fmt.Sprintf("%sm%04d", memoryPrefix, i), fmt.Sprintf("memory %d", i)})
	}
	return rows
}

func TestMemoriesReportAClippedConfigRead(t *testing.T) {
	rows := manyMemoryRows(Max + 3)
	config := &clippingSession{rowsByTable: map[string]*browse.RowPage{memoryTable: configPage(rows)}}
	sess := &clippingHistory{
		refs:    map[string]*clippingSession{"main": config, "c0": config},
		hashes:  map[string]string{"main": "h-a", "c0": "h-a"},
		commits: []browse.CommitInfo{{Hash: "c0", Author: "bigbes", Date: memoryNow.Add(-time.Hour)}},
	}

	v := buildMemories(t, sess, "")

	assert.True(t, v.ConfigTruncated, "the config table exceeded Max")
	assert.True(t, v.ConfigClipped(), "and it is the read the memories themselves come from")
	assert.Equal(t, Max+3, v.ConfigShownOf, "the table's true row count, not the number read")
	assert.Equal(t, Max, v.Total, "three memories exist that this list does not carry")
	assert.False(t, v.WalkTruncated, "which is a bound on commits and was not hit")
}

func TestMemoriesOnACompleteReadReportNoClip(t *testing.T) {
	v := buildMemories(t, memoryHistory(), "")

	assert.False(t, v.ConfigTruncated)
	assert.False(t, v.ConfigClipped())
	assert.Equal(t, 4, v.ConfigShownOf, "three memories and the tracker's issue_prefix")
	assert.Equal(t, 3, v.Total, "the settings row is not a memory")
}

// The other config read: the walk's own, at each commit it has to compare. A
// clip there leaves the list whole and the dates unsafe, and the two facts are
// told apart rather than merged.
func TestMemoriesReportAClippedConfigReadInsideTheWalk(t *testing.T) {
	head := &clippingSession{rowsByTable: map[string]*browse.RowPage{
		memoryTable: configPage([][]string{{memoryPrefix + "alpha", "second version"}}),
	}}
	// The root's config is the oversized one: the walk reads it to find out what
	// alpha said there, and gets its first Max rows.
	rootRows := append(manyMemoryRows(Max), []string{memoryPrefix + "alpha", "first version"})
	root := &clippingSession{rowsByTable: map[string]*browse.RowPage{memoryTable: configPage(rootRows)}}
	sess := &clippingHistory{
		refs:   map[string]*clippingSession{"main": head, "c1": head, "c0": root},
		hashes: map[string]string{"main": "h-b", "c1": "h-b", "c0": "h-a"},
		commits: []browse.CommitInfo{
			{Hash: "c1", Author: "bigbes", Date: memoryNow.Add(-time.Hour)},
			{Hash: "c0", Author: "alice", Date: memoryNow.Add(-48 * time.Hour)},
		},
	}

	v := buildMemories(t, sess, "")

	assert.True(t, v.ConfigTruncated, "a config read inside the walk was clipped")
	assert.False(t, v.ConfigClipped(), "though the list of memories itself is whole")
	assert.Equal(t, 1, v.ConfigShownOf)
	assert.Equal(t, 1, v.Total)
	assert.False(t, v.WalkTruncated, "the history is two commits and the walk saw both")
}

// --- the prefix index ---------------------------------------------------------

// prefixConfigRows is a config table of n rows whose issue_prefix sits last, so
// a read that stops at Max never reaches it: the tracker keeps its prefix and
// the index loses it.
func prefixConfigRows(n int, prefix string) [][]string {
	rows := make([][]string, 0, n)
	for i := 0; i < n-1; i++ {
		rows = append(rows, []string{fmt.Sprintf("%sm%04d", memoryPrefix, i), "some memory"})
	}
	return append(rows, []string{prefixKey, prefix})
}

func clippingPrefixTracker(head string, rows [][]string) *clippingReadyDB {
	return clippingTracker(head, map[string]*browse.RowPage{memoryTable: configPage(rows)})
}

func TestPrefixesAcrossReportsAClippedConfigRead(t *testing.T) {
	in := &clippingInstance{dbs: map[int]*clippingReadyDB{
		1: clippingPrefixTracker("h-global", prefixConfigRows(Max+2, "global")),
	}}
	dbs := []ReadyDatabase{{ID: 1, OwnerName: "bigbes", Name: "beads-global"}}

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

	// The prefix is in the table and not in the index: this is the silent loss.
	assert.Empty(t, index.Prefixes())
	_, ok := index.Lookup("global")
	assert.False(t, ok)
	assert.Empty(t, index.Failed, "a clipped read is not a failed one")

	require.Len(t, index.Truncated, 1, "and the index must not pass that off as an answer")
	assert.Equal(t, "bigbes/beads-global", index.Truncated[0].Database.Slug())
	assert.Equal(t, Max+2, index.Truncated[0].ShownOf, "the config table's true row count")
}

func TestPrefixesAcrossOnACompleteReadReportsNoClip(t *testing.T) {
	in := &clippingInstance{dbs: map[int]*clippingReadyDB{
		1: clippingPrefixTracker("h-global", prefixConfigRows(3, "global")),
	}}
	dbs := []ReadyDatabase{{ID: 1, OwnerName: "bigbes", Name: "beads-global"}}

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

	assert.Equal(t, []string{"global"}, index.Prefixes())
	assert.Empty(t, index.Truncated)
}

func TestPrefixesAcrossAttributesTheClipToItsDatabase(t *testing.T) {
	in := &clippingInstance{dbs: map[int]*clippingReadyDB{
		1: clippingPrefixTracker("h-global", prefixConfigRows(Max+2, "global")),
		2: clippingPrefixTracker("h-artifacts", prefixConfigRows(3, "artifacts")),
	}}
	dbs := []ReadyDatabase{
		{ID: 1, OwnerName: "bigbes", Name: "beads-global"},
		{ID: 2, OwnerName: "bigbes", Name: "sourcehut-artifacts"},
	}

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

	assert.Equal(t, []string{"artifacts"}, index.Prefixes(), "the readable sibling is unaffected")
	require.Len(t, index.Truncated, 1, "only one of the two was read in part")
	assert.Equal(t, "bigbes/beads-global", index.Truncated[0].Database.Slug())
	assert.Equal(t, Max+2, index.Truncated[0].ShownOf)
}

// The clip is cached with the prefix, for the reason the ready set's is: a
// second build under an unmoved head reads no row, and a projection served from
// the cache has to say what the read that produced it said.
func TestPrefixesAcrossClipSurvivesTheCache(t *testing.T) {
	in := &clippingInstance{dbs: map[int]*clippingReadyDB{
		1: clippingPrefixTracker("h-global", prefixConfigRows(Max+2, "global")),
	}}
	dbs := []ReadyDatabase{{ID: 1, OwnerName: "bigbes", Name: "beads-global"}}
	cache := &PrefixCache{}

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

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

	assert.Equal(t, reads, in.dbs[1].rowReads, "a second build with an unmoved head must read no rows")
	require.Len(t, second.Truncated, 1, "and must still report the clipped read")
	assert.Equal(t, Max+2, second.Truncated[0].ShownOf)
}