package storage

import (
	"errors"
	"fmt"
	"log"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"sync/atomic"

	"github.com/SimonWaldherr/tinySQL/internal/storage/pager"
)

// PagedIndexBackend adapts the pager B+Tree format to StorageBackend. Unlike
// ModeIndex's legacy GOB table files, it persists table rows and materialized
// secondary indexes in distinct B+Trees. Exact seeks can therefore read an
// index key, its row locator and only the requested rows/BLOBs.
//
// It is deliberately a separate mode: existing ModeIndex directories retain
// their GOB compatibility, while published `paged_index` artifacts have an
// explicit format boundary and can be opened safely read-only.
type PagedIndexBackend struct {
	page             *pager.PageBackend
	path             string
	memoryLimitBytes int64
	readOnly         atomic.Bool

	// metadata contains schema and index roots only (never rows/BLOBs). It is
	// immutable after a published read-only open and avoids a catalog B+Tree
	// traversal plus JSON decode on every prepared point lookup.
	metadataMu sync.RWMutex
	metadata   map[string]*pagedIndexTableMetadata

	// pool caches whole *Table objects (rows included) loaded via LoadTable,
	// as a bounded, evictable lease — mirroring HybridBackend's BufferPool.
	// Without it, a mutation made in place on the *Table DB.Get returns for
	// this evictable storage mode (see DB.backendTablesEvictable) would be
	// invisible to both a later LoadTable in the same process (LoadTable
	// used to always decode a fresh copy from the pager) and to DB.Sync.
	pool *BufferPool

	// versions tracks the version at last load/save per tenant+table,
	// mirroring DiskBackend.versions, so IsDirty can tell whether a pooled
	// table needs to be re-saved.
	versionsMu sync.Mutex
	versions   map[string]int
}

// pagedIndexTableMetadata is the immutable serving descriptor cached per
// table. It intentionally contains no rows or BLOBs: only schema and B+Tree
// locators needed to avoid re-reading the catalog on every point lookup.
type pagedIndexTableMetadata struct {
	table     *Table
	tableRoot pager.PageID
	indexRoot map[string]pager.PageID
}

const pagedIndexFilename = "tinysql.pages"

func NewPagedIndexBackend(dir string, maxMemoryBytes int64, readOnly bool) (*PagedIndexBackend, error) {
	if readOnly {
		if info, err := os.Stat(dir); err != nil || !info.IsDir() {
			if err == nil {
				err = fmt.Errorf("not a directory")
			}
			return nil, fmt.Errorf("read-only paged index requires existing directory %q: %w", dir, err)
		}
	} else if err := os.MkdirAll(dir, 0o755); err != nil {
		return nil, err
	}
	path := filepath.Join(dir, pagedIndexFilename)
	if readOnly {
		if _, err := os.Stat(path); err != nil {
			return nil, fmt.Errorf("read-only paged index requires published artifact %q: %w", path, err)
		}
	}
	maxMemoryBytes = normalizePagedIndexMemoryLimit(maxMemoryBytes)
	// The page cache and table-object pool are separate residents. Splitting
	// the caller's budget here makes MaxMemoryBytes a backend-wide limit rather
	// than allowing each cache to consume the full amount independently.
	pageBudget, poolBudget := splitPagedIndexMemoryBudget(maxMemoryBytes)
	maxPages := int(pageBudget / pager.DefaultPageSize)
	if maxPages < 1 {
		maxPages = 1
	}
	pb, err := pager.NewPageBackend(pager.PageBackendConfig{
		Path:          path,
		MaxCachePages: maxPages,
		ReadOnly:      readOnly,
	})
	if err != nil {
		return nil, err
	}
	pool := NewBufferPool(&MemoryPolicy{
		MaxMemoryBytes:      poolBudget,
		Strategy:            StrategyLRU,
		EvictionThreshold:   0.85,
		EnableEviction:      true,
		EvictionBatchSize:   5,
		TrackAccessPatterns: true,
	})
	b := &PagedIndexBackend{
		page:             pb,
		path:             path,
		memoryLimitBytes: maxMemoryBytes,
		metadata:         make(map[string]*pagedIndexTableMetadata),
		pool:             pool,
		versions:         make(map[string]int),
	}
	// A pool eviction (triggered from within LoadTable on an unrelated
	// table) must not silently drop a dirty table that was mutated in place
	// after being loaded — see the DB.Get doc comment on the query-scoped
	// lease. Save it first, going straight to the pager (not through
	// PagedIndexBackend.SaveTable, which would call pool.Put and deadlock:
	// this callback runs with the pool's own lock already held).
	pool.SetEvictionSaver(func(tenant, name string, t *Table) {
		if b.IsDirty(tenant, name, t.Version) {
			if err := b.saveToPager(tenant, t); err != nil {
				log.Printf("warning: flush-before-evict failed for paged index %s/%s: %v", tenant, name, err)
			}
		}
	})
	b.readOnly.Store(readOnly)
	return b, nil
}

func splitPagedIndexMemoryBudget(maxMemoryBytes int64) (pageBudget, poolBudget int64) {
	maxMemoryBytes = normalizePagedIndexMemoryLimit(maxMemoryBytes)
	// Point and range access use the pager directly; table objects are only a
	// compatibility cache for callers that explicitly load whole tables.
	pageBudget = maxMemoryBytes * 3 / 4
	poolBudget = maxMemoryBytes - pageBudget
	if pageBudget < pager.DefaultPageSize {
		pageBudget = pager.DefaultPageSize
	}
	if poolBudget < pager.DefaultPageSize {
		poolBudget = pager.DefaultPageSize
	}
	return pageBudget, poolBudget
}

func normalizePagedIndexMemoryLimit(maxMemoryBytes int64) int64 {
	if maxMemoryBytes <= 0 {
		return 64 * 1024 * 1024
	}
	// A pager frame and a compatibility table-pool floor are both required.
	// Report this effective bound rather than claiming an impossible smaller
	// budget to callers and BackendStats observers.
	minimum := int64(2 * pager.DefaultPageSize)
	if maxMemoryBytes < minimum {
		return minimum
	}
	return maxMemoryBytes
}

func (b *PagedIndexBackend) LoadTable(tenant, name string) (*Table, error) {
	tn := strings.ToLower(tenant)
	lc := strings.ToLower(name)
	if t, ok := b.pool.Get(tn, lc); ok {
		return t, nil
	}

	td, err := b.page.LoadTable(tenant, name)
	if err != nil || td == nil {
		return nil, err
	}
	t := NewTable(td.Name, pagerColumnsToStorage(td.Columns), td.IsTemp)
	t.Rows = td.Rows
	t.Version = td.Version
	for _, index := range td.Indexes {
		if err := t.CreateSecondaryIndex(index.Name, index.Columns, index.Unique); err != nil {
			return nil, fmt.Errorf("rebuild paged index %s: %w", index.Name, err)
		}
	}

	b.versionsMu.Lock()
	b.versions[pagedMetadataKey(tenant, name)] = t.Version
	b.versionsMu.Unlock()

	if err := b.pool.Put(tn, lc, t); err != nil {
		// Can't fit in the bounded pool (e.g. memory limit without room to
		// evict). Still return the table; it just won't be cached, and a
		// mutation made on it is durable only via an explicit SaveTable/Put.
		_ = err
	}

	return t, nil
}

// saveToPager writes t to the underlying pager and records the saved
// version, without touching the table-object pool. Used by both SaveTable
// (which additionally refreshes the pool entry) and the pool's
// eviction-saver callback (which must not call back into the pool).
func (b *PagedIndexBackend) saveToPager(tenant string, t *Table) error {
	if b.IsReadOnly() {
		return ErrReadOnlyStorage
	}
	td := &pager.TableData{
		Name:    t.Name,
		Columns: storageColumnsToPager(t.Cols),
		Rows:    t.Rows,
		IsTemp:  t.IsTemp,
		Version: t.Version,
		Indexes: make([]pager.IndexInfo, 0, len(t.Indexes)),
	}
	for _, index := range t.Indexes {
		td.Indexes = append(td.Indexes, pager.IndexInfo{
			Name:    index.Name,
			Columns: append([]string(nil), index.Columns...),
			Unique:  index.Unique,
			Entries: indexEntriesToPager(index.Entries),
		})
	}
	if err := b.page.SaveTable(tenant, td); err != nil {
		return err
	}
	b.dropMetadata(tenant, t.Name)
	b.versionsMu.Lock()
	b.versions[pagedMetadataKey(tenant, t.Name)] = t.Version
	b.versionsMu.Unlock()
	return nil
}

func (b *PagedIndexBackend) SaveTable(tenant string, t *Table) error {
	if err := b.saveToPager(tenant, t); err != nil {
		return err
	}
	tn := strings.ToLower(tenant)
	lc := strings.ToLower(t.Name)
	if err := b.pool.Put(tn, lc, t); err != nil {
		// The pager write above is durable; cache admission is an optional
		// optimization for compatibility callers that explicitly SaveTable a
		// table larger than the serving budget. Bounded MBTiles builds use
		// AppendRows and never rely on this cache admission path.
		log.Printf("info: paged index cache admission skipped for %s/%s: %v", tn, lc, err)
	}
	return nil
}

// AppendRows adds newRows (and their unique-index keys) to an existing
// table's on-disk B+Trees without reading or rewriting rows already there --
// see pager.PageBackend.AppendRows. Memory and I/O cost are proportional to
// len(newRows), not to how much of the table already exists, which is what
// makes building a tileset larger than memory possible: SaveTable cannot do
// this, since it always frees and rebuilds the whole tree from a complete row
// set. ImportMBTiles (internal/importer) uses this for a PagedIndexBackend
// destination.
//
// The table must already exist (created once via SaveTable/CreateTable) and
// declare its unique index; AppendRows only appends.
func (b *PagedIndexBackend) AppendRows(tenant, name string, newRows [][]any, newIndexKeys map[string][][]byte) (int64, error) {
	if b.IsReadOnly() {
		return 0, ErrReadOnlyStorage
	}
	firstRowID, err := b.page.AppendRows(tenant, name, newRows, newIndexKeys)
	if err != nil {
		return 0, err
	}
	// The table root and index roots just changed on disk. Drop the cheap
	// locator cache (LookupIndexRows would otherwise keep resolving against a
	// stale root) and any pooled *Table, whose Rows/Version would otherwise
	// be stale for a caller that later does a plain Get. The next Get/Put
	// reloads fresh from disk and repopulates both.
	b.dropMetadata(tenant, name)
	tn := strings.ToLower(tenant)
	lc := strings.ToLower(name)
	b.pool.Remove(tn, lc)
	b.versionsMu.Lock()
	delete(b.versions, pagedMetadataKey(tenant, name))
	b.versionsMu.Unlock()
	return firstRowID, nil
}

// ScanRows streams a table's rows to fn without materializing all of them at
// once -- see pager.PageBackend.ScanTableRows. This bypasses both LoadTable
// and the table-object pool entirely: a caller using it wants to read a
// table that may be far larger than the pool's memory budget, so pooling it
// (or any part of it) would defeat the point.
func (b *PagedIndexBackend) ScanRows(tenant, name string, fn func(row []any) bool) error {
	return b.page.ScanTableRows(tenant, name, fn)
}

// IsDirty reports whether name's in-memory version has diverged from the
// version last saved to the pager. Satisfies the dirtyTracker interface.
func (b *PagedIndexBackend) IsDirty(tenant, name string, currentVersion int) bool {
	b.versionsMu.Lock()
	defer b.versionsMu.Unlock()
	if v, ok := b.versions[pagedMetadataKey(tenant, name)]; ok {
		return currentVersion != v
	}
	return true // unknown -> assume dirty
}

// PooledTables returns every table currently resident in the table-object
// pool, satisfying the pooledTableSource interface. DB.Sync/DB.Close use
// this to find and flush a table that DB.Get loaded lazily for
// ModePagedIndex without registering it in DB.tenants.
func (b *PagedIndexBackend) PooledTables() []PooledTableRef {
	entries := b.pool.Snapshot()
	out := make([]PooledTableRef, 0, len(entries))
	for _, e := range entries {
		out = append(out, PooledTableRef{Tenant: e.Tenant, Table: e.Table})
	}
	return out
}

func (b *PagedIndexBackend) DeleteTable(tenant, name string) error {
	if b.IsReadOnly() {
		return ErrReadOnlyStorage
	}
	if err := b.page.DeleteTable(tenant, name); err != nil {
		return err
	}
	b.dropMetadata(tenant, name)
	b.pool.Remove(strings.ToLower(tenant), strings.ToLower(name))
	b.versionsMu.Lock()
	delete(b.versions, pagedMetadataKey(tenant, name))
	b.versionsMu.Unlock()
	return nil
}

func (b *PagedIndexBackend) ListTableNames(tenant string) ([]string, error) {
	return b.page.ListTableNames(tenant)
}

func (b *PagedIndexBackend) TableExists(tenant, name string) bool {
	return b.page.TableExists(tenant, name)
}

func (b *PagedIndexBackend) Sync() error {
	if b.IsReadOnly() {
		return nil
	}
	return b.page.Sync()
}

func (b *PagedIndexBackend) Close() error              { return b.page.Close() }
func (b *PagedIndexBackend) Mode() StorageMode         { return ModePagedIndex }
func (b *PagedIndexBackend) SetReadOnly(readOnly bool) { b.readOnly.Store(readOnly) }
func (b *PagedIndexBackend) IsReadOnly() bool          { return b != nil && b.readOnly.Load() }

func (b *PagedIndexBackend) Stats() BackendStats {
	stats := b.page.Stats()
	info, _ := os.Stat(b.path)
	var size int64
	if info != nil {
		size = info.Size()
	}
	var hitRate float64
	if total := stats.CacheHits + stats.CacheMisses; total > 0 {
		hitRate = float64(stats.CacheHits) / float64(total)
	}
	return BackendStats{
		Mode:            ModePagedIndex,
		DiskUsedBytes:   size,
		MemoryUsedBytes: int64(stats.CachedPages+stats.TransientFrames)*int64(stats.PageSize) + b.pool.GetMemoryUsage(),
		// MaxMemoryBytes is a backend-wide budget shared by the pager and the
		// compatibility table pool. The page cache is intentionally only one
		// portion of it, so reporting its rounded page capacity here would make
		// operational metrics contradict the caller's configured bound.
		MemoryLimitBytes: b.memoryLimitBytes,
		CacheHitRate:     hitRate,
		SyncCount:        stats.SyncCount,
		LoadCount:        stats.LoadCount,
		PageReads:        stats.PageReads,
		CacheHits:        stats.CacheHits,
		CacheMisses:      stats.CacheMisses,
		CachedPages:      stats.CachedPages,
		PinnedPages:      stats.PinnedPages,
		TransientPages:   stats.TransientPages,
		TransientFrames:  stats.TransientFrames,
		MaxCachePages:    stats.MaxCachePages,
	}
}

// IndexMetadata creates a schema-only Table with physical index metadata. It
// is a cheap operation (catalog B+Tree only) and lets the SQL planner choose a
// page index before calling LoadTable.
func (b *PagedIndexBackend) IndexMetadata(tenant, name string) (*Table, error) {
	key := pagedMetadataKey(tenant, name)
	b.metadataMu.RLock()
	cached := b.metadata[key]
	b.metadataMu.RUnlock()
	if cached != nil {
		return cached.table, nil
	}
	entry, err := b.page.TableEntry(tenant, name)
	if err != nil || entry == nil {
		return nil, err
	}
	t := NewTable(entry.Table, pagerCatalogColumnsToStorage(entry.Columns), false)
	t.Version = entry.Version
	for _, index := range entry.Indexes {
		t.Indexes[strings.ToLower(index.Name)] = &SecondaryIndex{
			Name:    index.Name,
			Columns: append([]string(nil), index.Columns...),
			Unique:  index.Unique,
		}
	}
	metadata := &pagedIndexTableMetadata{
		table:     t,
		tableRoot: entry.RootPageID,
		indexRoot: make(map[string]pager.PageID, len(entry.Indexes)),
	}
	for _, index := range entry.Indexes {
		metadata.indexRoot[strings.ToLower(index.Name)] = index.RootPageID
	}
	b.metadataMu.Lock()
	if existing := b.metadata[key]; existing != nil {
		b.metadataMu.Unlock()
		return existing.table, nil
	}
	b.metadata[key] = metadata
	b.metadataMu.Unlock()
	return t, nil
}

// LookupIndexRows resolves a complete canonical composite key through the
// on-disk index and returns only matching rows. The caller owns the returned
// slices and can safely project/copy BLOBs without exposing page-cache memory.
func (b *PagedIndexBackend) LookupIndexRows(tenant, table, indexName string, values []any) ([][]any, bool, error) {
	key := pagedMetadataKey(tenant, table)
	b.metadataMu.RLock()
	metadata := b.metadata[key]
	b.metadataMu.RUnlock()
	if metadata == nil {
		// Populate the immutable locator cache once.
		if _, err := b.IndexMetadata(tenant, table); err != nil {
			return nil, false, err
		}
		b.metadataMu.RLock()
		metadata = b.metadata[key]
		b.metadataMu.RUnlock()
	}
	if metadata == nil {
		return nil, false, nil
	}
	root, ok := metadata.indexRoot[strings.ToLower(indexName)]
	if !ok || root == pager.InvalidPageID {
		return nil, false, nil
	}
	return b.page.LookupIndexRowsByRoot(metadata.tableRoot, root, indexName, CanonicalIndexKey(values))
}

// PagedIndexLocator is an immutable resolved table/index pair. It is valid for
// the lifetime of its read-only PagedIndexBackend and lets typed serving paths
// bypass repeated catalog and metadata-map work.
type PagedIndexLocator struct {
	page      *pager.PageBackend
	tableRoot pager.PageID
	indexRoot pager.PageID
	indexName string
}

// LocateIndex resolves one physical index without reading table rows.
func (b *PagedIndexBackend) LocateIndex(tenant, table, indexName string) (*PagedIndexLocator, bool, error) {
	key := pagedMetadataKey(tenant, table)
	b.metadataMu.RLock()
	metadata := b.metadata[key]
	b.metadataMu.RUnlock()
	if metadata == nil {
		if _, err := b.IndexMetadata(tenant, table); err != nil {
			return nil, false, err
		}
		b.metadataMu.RLock()
		metadata = b.metadata[key]
		b.metadataMu.RUnlock()
	}
	if metadata == nil {
		return nil, false, nil
	}
	root, ok := metadata.indexRoot[strings.ToLower(indexName)]
	if !ok || root == pager.InvalidPageID {
		return nil, false, nil
	}
	return &PagedIndexLocator{page: b.page, tableRoot: metadata.tableRoot, indexRoot: root, indexName: indexName}, true, nil
}

// LookupUniqueColumn performs an exact seek through a unique index and
// invokes visit with one decoded column from the matching table row.
func (locator *PagedIndexLocator) LookupUniqueColumn(key []byte, column int, visit func(any) error) (bool, error) {
	if locator == nil || locator.page == nil {
		return false, errors.New("paged index locator is unavailable")
	}
	if visit == nil {
		return false, errors.New("paged index locator callback is nil")
	}
	return locator.page.LookupUniqueIndexRowColumnByRoot(locator.tableRoot, locator.indexRoot, locator.indexName, key, column, visit)
}

// ContainsUnique proves that a unique index key resolves to an existing table
// row without loading that row's value or any overflow BLOB pages.
func (locator *PagedIndexLocator) ContainsUnique(key []byte) (bool, error) {
	if locator == nil || locator.page == nil {
		return false, errors.New("paged index locator is unavailable")
	}
	return locator.page.LookupUniqueIndexRowExistsByRoot(locator.tableRoot, locator.indexRoot, locator.indexName, key)
}

// ScanIndexRowsRange streams rows in an ordered composite-index interval.
func (b *PagedIndexBackend) ScanIndexRowsRange(tenant, table, indexName string, startValues, endValues []any, fn func([]any) bool) error {
	return b.page.ScanIndexRowsRange(tenant, table, indexName, CanonicalIndexKey(startValues), CanonicalIndexKey(endValues), fn)
}

func storageColumnsToPager(cols []Column) []pager.ColumnInfo {
	out := make([]pager.ColumnInfo, len(cols))
	for i, col := range cols {
		out[i] = pager.ColumnInfo{
			Name:         col.Name,
			Type:         int(col.Type),
			Constraint:   int(col.Constraint),
			PointerTable: col.PointerTable,
		}
		if col.ForeignKey != nil {
			out[i].FKTable = col.ForeignKey.Table
			out[i].FKColumn = col.ForeignKey.Column
		}
	}
	return out
}

func pagerColumnsToStorage(cols []pager.ColumnInfo) []Column {
	out := make([]Column, len(cols))
	for i, col := range cols {
		out[i] = Column{
			Name:         col.Name,
			Type:         ColType(col.Type),
			Constraint:   ConstraintType(col.Constraint),
			PointerTable: col.PointerTable,
		}
		if col.FKTable != "" {
			out[i].ForeignKey = &ForeignKeyRef{Table: col.FKTable, Column: col.FKColumn}
		}
	}
	return out
}

func pagerCatalogColumnsToStorage(cols []pager.CatalogColumn) []Column {
	out := make([]Column, len(cols))
	for i, col := range cols {
		out[i] = Column{
			Name:         col.Name,
			Type:         ColType(col.Type),
			Constraint:   ConstraintType(col.Constraint),
			PointerTable: col.PtrTable,
		}
		if col.FKTable != "" {
			out[i].ForeignKey = &ForeignKeyRef{Table: col.FKTable, Column: col.FKColumn}
		}
	}
	return out
}

func indexEntriesToPager(entries []IndexEntry) []pager.IndexEntry {
	out := make([]pager.IndexEntry, len(entries))
	for i, entry := range entries {
		out[i].Key = append([]byte(nil), entry.Key...)
		out[i].RowIDs = append([]int(nil), entry.RowIDs...)
	}
	return out
}

func (b *PagedIndexBackend) dropMetadata(tenant, name string) {
	b.metadataMu.Lock()
	delete(b.metadata, pagedMetadataKey(tenant, name))
	b.metadataMu.Unlock()
}

func pagedMetadataKey(tenant, name string) string {
	return strings.ToLower(tenant) + "\x00" + strings.ToLower(name)
}
