package browse

import (
	"context"
	"fmt"
	"io"

	"github.com/dolthub/dolt/go/libraries/doltcore/doltdb"
	"github.com/dolthub/dolt/go/libraries/doltcore/doltdb/durable"
	"github.com/dolthub/dolt/go/libraries/doltcore/schema"
	"github.com/dolthub/dolt/go/store/prolly/tree"
	"github.com/dolthub/dolt/go/store/val"
)

// Cell placeholders. Rendering must never panic and never emit non-printable
// bytes; exotic or out-of-band values degrade to one of these.
const (
	placeholderNull       = "NULL"
	placeholderBinary     = "<binary>"
	placeholderUnreadable = "<unreadable>"
)

// ColumnInfo describes one column of a table schema.
type ColumnInfo struct {
	Name       string
	Type       string
	PrimaryKey bool
	Nullable   bool
}

// TableInfo is a table name, its schema, and its row count at a ref.
type TableInfo struct {
	Name     string
	Columns  []ColumnInfo
	RowCount uint64
}

// RowPage is a paginated slice of a table's rows rendered to strings. Columns
// lists the column names in the same order as each row's cells. For keyed
// tables columns are primary-key columns first, then the rest.
//
// # Why a NULL renders as "NULL" and there is a mask beside it
//
// A cell that holds no value renders as the string "NULL" in Rows, which is
// exactly what a row that genuinely stores the four characters N,U,L,L renders
// as. That flattening stays: Rows is what pages and projections put on screen,
// where "NULL" is the conventional and readable answer, and every consumer of
// this package reads Rows as display strings — encoding nullness into the
// string instead (a sentinel, an empty string, a marker) would ripple through
// every template and every projection for no gain to a reader.
//
// Nulls is for the callers where the two are not interchangeable: an API that
// hands rows to a machine (the MCP row reader) gives out strings with no schema
// beside them, so without the mask an agent cannot tell an absent value from
// the text "NULL" — and answering "is there a value here?" is not something it
// can recover from the rendered string afterwards.
//
// Nulls is parallel to Rows: Nulls[i][j] reports whether Rows[i][j] was a real
// SQL NULL, so any index valid for Rows is valid for Nulls. It costs one bool
// per cell, which is what makes it affordable on a full page.
//
// Only NULL gets this treatment. The other two placeholders, "<binary>" and
// "<unreadable>", answer what a value *is*; NULL answers whether there is one
// at all, and only that question is unanswerable from the rendered string.
type RowPage struct {
	Columns []string
	Rows    [][]string
	// Nulls[i][j] is true when Rows[i][j] is a real NULL rather than a value
	// that happens to render like one. Same shape as Rows.
	Nulls  [][]bool
	Offset int
	Total  int
}

// Tables lists the tables in the committed root at ref (a branch name or
// commit hash), each with its schema and row count.
func (db *DB) Tables(ctx context.Context, refStr string) ([]TableInfo, error) {
	root, err := db.resolveRoot(ctx, refStr)
	if err != nil {
		return nil, err
	}

	names, err := root.GetTableNames(ctx, doltdb.DefaultSchemaName, false)
	if err != nil {
		return nil, fmt.Errorf("browse: list tables at %q: %w", refStr, err)
	}

	infos := make([]TableInfo, 0, len(names))
	for _, name := range names {
		tbl, ok, err := root.GetTable(ctx, doltdb.TableName{Name: name})
		if err != nil {
			return nil, fmt.Errorf("browse: get table %q: %w", name, err)
		}
		if !ok {
			// Listed by GetTableNames but not resolvable: inconsistent root.
			return nil, fmt.Errorf("browse: table %q listed but missing", name)
		}

		sch, err := tbl.GetSchema(ctx)
		if err != nil {
			return nil, fmt.Errorf("browse: schema of %q: %w", name, err)
		}

		idx, err := tbl.GetRowData(ctx)
		if err != nil {
			return nil, fmt.Errorf("browse: row data of %q: %w", name, err)
		}
		count, err := idx.Count()
		if err != nil {
			return nil, fmt.Errorf("browse: row count of %q: %w", name, err)
		}

		infos = append(infos, TableInfo{
			Name:     name,
			Columns:  columnInfos(sch),
			RowCount: count,
		})
	}

	return infos, nil
}

// TableHash returns the content hash of one table in the committed root at ref
// (a branch name or a commit hash). It reads no rows: the hash is the address
// of the table struct itself, so answering "did this table change between two
// commits?" costs a root lookup and nothing more. That is what makes a history
// walk affordable — a walk that has to attribute a change to a commit can skip
// every commit whose table hash equals its neighbour's, and read rows only at
// the few commits that actually touched the table.
//
// The hash covers the whole table (schema, row data, secondary indexes), not
// just the rows. For change detection that is the wanted answer: a column added
// without touching a row is still a change to the table.
//
// A table that does not exist at ref is ok=false with a nil error, deliberately
// *not* ErrTableNotFound (which Rows returns). The caller of this primitive is
// a loop walking backwards through history asking "did it change?", and a table
// that had not been created yet at an old commit is an ordinary answer there,
// not a failure. Callers that need "absent" to be an error can test ok
// themselves; a caller that walked into an error at every pre-creation commit
// could not tell that case apart from a real one.
func (db *DB) TableHash(ctx context.Context, refStr, table string) (string, bool, error) {
	root, err := db.resolveRoot(ctx, refStr)
	if err != nil {
		return "", false, err
	}

	tbl, ok, err := root.GetTable(ctx, doltdb.TableName{Name: table})
	if err != nil {
		return "", false, fmt.Errorf("browse: get table %q at %q: %w", table, refStr, err)
	}
	if !ok {
		return "", false, nil
	}

	h, err := tbl.HashOf()
	if err != nil {
		return "", false, fmt.Errorf("browse: hash of table %q at %q: %w", table, refStr, err)
	}
	return h.String(), true, nil
}

// columnInfos renders a schema's columns in natural table order.
func columnInfos(sch schema.Schema) []ColumnInfo {
	cols := sch.GetAllCols().GetColumns()
	out := make([]ColumnInfo, len(cols))
	for i, c := range cols {
		typeStr := ""
		if c.TypeInfo != nil {
			if sqlType := c.TypeInfo.ToSqlType(); sqlType != nil {
				typeStr = sqlType.String()
			} else {
				typeStr = c.TypeInfo.String()
			}
		}
		out[i] = ColumnInfo{
			Name:       c.Name,
			Type:       typeStr,
			PrimaryKey: c.IsPartOfPK,
			Nullable:   c.IsNullable(),
		}
	}
	return out
}

// cellRef locates a column's value within a prolly row: either in the key
// tuple or the value tuple, at the given field index.
type cellRef struct {
	fromKey bool
	idx     int
}

// Rows returns a page of rows from table at ref, starting at offset (0-based)
// and returning at most limit rows. The page is read directly from the prolly
// map via an ordinal range, so it is O(limit) regardless of offset.
func (db *DB) Rows(ctx context.Context, refStr, table string, offset, limit int) (*RowPage, error) {
	if offset < 0 {
		return nil, fmt.Errorf("browse: offset must be non-negative, got %d", offset)
	}
	if limit <= 0 {
		return nil, fmt.Errorf("browse: limit must be positive, got %d", limit)
	}

	root, err := db.resolveRoot(ctx, refStr)
	if err != nil {
		return nil, err
	}

	tbl, ok, err := root.GetTable(ctx, doltdb.TableName{Name: table})
	if err != nil {
		return nil, fmt.Errorf("browse: get table %q: %w", table, err)
	}
	if !ok {
		return nil, fmt.Errorf("%w: %s", ErrTableNotFound, table)
	}

	sch, err := tbl.GetSchema(ctx)
	if err != nil {
		return nil, fmt.Errorf("browse: schema of %q: %w", table, err)
	}

	idx, err := tbl.GetRowData(ctx)
	if err != nil {
		return nil, fmt.Errorf("browse: row data of %q: %w", table, err)
	}
	total, err := idx.Count()
	if err != nil {
		return nil, fmt.Errorf("browse: row count of %q: %w", table, err)
	}

	colNames, refs := rowLayout(sch)

	page := &RowPage{
		Columns: colNames,
		Rows:    [][]string{},
		Nulls:   [][]bool{},
		Offset:  offset,
		Total:   int(total),
	}

	start := uint64(offset)
	if start >= total {
		// Past the end (also covers the empty-table case): no rows.
		return page, nil
	}
	stop := start + uint64(limit)
	if stop > total {
		stop = total
	}

	m, err := durable.ProllyMapFromIndex(idx)
	if err != nil {
		return nil, fmt.Errorf("browse: prolly map of %q: %w", table, err)
	}
	keyDesc, valDesc := m.Descriptors()

	iter, err := m.IterOrdinalRange(ctx, start, stop)
	if err != nil {
		return nil, fmt.Errorf("browse: iterate rows of %q: %w", table, err)
	}

	// ns dereferences out-of-line (address-encoded) values — text/longtext that
	// Dolt stores in a separate chunk rather than inline in the tuple.
	ns := m.NodeStore()
	for {
		key, value, err := iter.Next(ctx)
		if err == io.EOF {
			break
		}
		if err != nil {
			return nil, fmt.Errorf("browse: read row of %q: %w", table, err)
		}

		row := make([]string, len(refs))
		// One bool per cell, appended in lockstep with the row so the two can
		// never drift apart — including on a page that starts mid-table.
		nulls := make([]bool, len(refs))
		for i, r := range refs {
			if r.fromKey {
				row[i], nulls[i] = renderCell(ctx, ns, keyDesc, r.idx, key)
			} else {
				row[i], nulls[i] = renderCell(ctx, ns, valDesc, r.idx, value)
			}
		}
		page.Rows = append(page.Rows, row)
		page.Nulls = append(page.Nulls, nulls)
	}

	return page, nil
}

// rowLayout maps schema columns to their position in the prolly key/value
// tuples and produces the display column order.
//
//   - Keyed tables: primary-key columns (in key order) map to the key tuple,
//     the remaining columns (in stored order) to the value tuple.
//   - Keyless tables: every column is in the value tuple; field 0 is the
//     hidden cardinality, so column i lives at value index i+1, and there is
//     no meaningful key.
func rowLayout(sch schema.Schema) ([]string, []cellRef) {
	if schema.IsKeyless(sch) {
		cols := sch.GetNonPKCols().GetColumns()
		names := make([]string, len(cols))
		refs := make([]cellRef, len(cols))
		for i, c := range cols {
			names[i] = c.Name
			refs[i] = cellRef{fromKey: false, idx: i + 1}
		}
		return names, refs
	}

	pkCols := sch.GetPKCols().GetColumns()
	nonPKCols := sch.GetNonPKCols().GetColumns()
	names := make([]string, 0, len(pkCols)+len(nonPKCols))
	refs := make([]cellRef, 0, len(pkCols)+len(nonPKCols))
	for i, c := range pkCols {
		names = append(names, c.Name)
		refs = append(refs, cellRef{fromKey: true, idx: i})
	}
	for i, c := range nonPKCols {
		names = append(names, c.Name)
		refs = append(refs, cellRef{fromKey: false, idx: i})
	}
	return names, refs
}

// renderCell renders one tuple field to a display string. It never panics
// (recovering into a placeholder) and maps binary / out-of-band encodings to
// "<binary>" so a row preview never dumps opaque bytes.
//
// text/longtext columns are the exception: Dolt stores anything past a small
// inline threshold out-of-line, addressed by a content hash (StringAddrEnc, or
// StringAdaptiveEnc which is inline-or-address in the same field). Those are
// resolved through ns to their real content — without this they would render as
// "<binary>" (addr) or a raw hash string (adaptive), losing every long
// description, close reason, comment body, and audit payload.
//
// isNull is true only when the field holds no value. It is the answer the
// rendered string cannot carry, since a stored "NULL" renders the same way; see
// the RowPage doc. It comes from the nullness test the renderer already had to
// do, so reporting it costs no extra read of the tuple. A field that is
// unreadable (out of range, or a panic recovered here) is not null: nothing is
// known about it, which is a different answer from "there is no value".
func renderCell(ctx context.Context, ns tree.NodeStore, td *val.TupleDesc, i int, tup val.Tuple) (out string, isNull bool) {
	defer func() {
		if r := recover(); r != nil {
			out, isNull = placeholderUnreadable, false
		}
	}()

	if i < 0 || i >= td.Count() {
		// Column not present in this tuple (e.g. a virtual/dropped column that
		// isn't materialized): degrade rather than index out of range.
		return placeholderUnreadable, false
	}
	if td.IsNull(i, tup) {
		return placeholderNull, true
	}

	switch td.Types[i].Enc {
	case val.StringAddrEnc:
		return resolveStringAddr(ctx, ns, td, i, tup)
	case val.StringAdaptiveEnc:
		return resolveStringAdaptive(ctx, ns, td, i, tup)
	case val.ByteStringEnc, val.Hash128Enc, val.CellEnc,
		val.BytesAddrEnc, val.JSONAddrEnc,
		val.GeomAddrEnc, val.CommitAddrEnc,
		val.BytesAdaptiveEnc, val.GeomAdaptiveEnc:
		// Genuine binary / opaque out-of-band values: never dump raw bytes.
		return placeholderBinary, false
	}

	return td.FormatValue(ctx, i, td.GetField(i, tup)), false
}

// resolveStringAddr dereferences a StringAddrEnc field (text/longtext always
// stored out-of-line) to its full content. It returns the same (string,
// isNull) pair as renderCell.
func resolveStringAddr(ctx context.Context, ns tree.NodeStore, td *val.TupleDesc, i int, tup val.Tuple) (string, bool) {
	h, ok := td.GetStringAddr(i, tup)
	if !ok {
		return placeholderNull, true
	}
	s, err := val.NewTextStorage(h, ns).Unwrap(ctx)
	if err != nil {
		return placeholderUnreadable, false
	}
	return s, false
}

// resolveStringAdaptive reads a StringAdaptiveEnc field, which stores its value
// either inline (returned as a string) or out-of-line (returned as a
// *val.TextStorage to Unwrap) in the same field. It returns the same (string,
// isNull) pair as renderCell.
func resolveStringAdaptive(ctx context.Context, ns tree.NodeStore, td *val.TupleDesc, i int, tup val.Tuple) (string, bool) {
	v, ok, err := td.GetStringAdaptiveValue(ctx, i, ns, tup)
	if err != nil || !ok {
		if err != nil {
			return placeholderUnreadable, false
		}
		return placeholderNull, true
	}
	switch s := v.(type) {
	case string:
		return s, false
	case *val.TextStorage:
		str, err := s.Unwrap(ctx)
		if err != nil {
			return placeholderUnreadable, false
		}
		return str, false
	default:
		return placeholderUnreadable, false
	}
}