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.
type RowPage struct {
	Columns []string
	Rows    [][]string
	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{},
		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))
		for i, r := range refs {
			if r.fromKey {
				row[i] = renderCell(ctx, ns, keyDesc, r.idx, key)
			} else {
				row[i] = renderCell(ctx, ns, valDesc, r.idx, value)
			}
		}
		page.Rows = append(page.Rows, row)
	}

	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.
func renderCell(ctx context.Context, ns tree.NodeStore, td *val.TupleDesc, i int, tup val.Tuple) (out string) {
	defer func() {
		if r := recover(); r != nil {
			out = placeholderUnreadable
		}
	}()

	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
	}
	if td.IsNull(i, tup) {
		return placeholderNull
	}

	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
	}

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

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

// 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.
func resolveStringAdaptive(ctx context.Context, ns tree.NodeStore, td *val.TupleDesc, i int, tup val.Tuple) string {
	v, ok, err := td.GetStringAdaptiveValue(ctx, i, ns, tup)
	if err != nil || !ok {
		if err != nil {
			return placeholderUnreadable
		}
		return placeholderNull
	}
	switch s := v.(type) {
	case string:
		return s
	case *val.TextStorage:
		str, err := s.Unwrap(ctx)
		if err != nil {
			return placeholderUnreadable
		}
		return str
	default:
		return placeholderUnreadable
	}
}