package packs

import (
	"context"
	"database/sql"
	"fmt"
	"sort"
	"strings"
	"sync"

	"github.com/nexuscrm/kernel/pkg/utils"
	cp "github.com/pingcap-inc/nexuscrm-contract/pkg/controlplane"
)

// ComponentColumnCatalog returns the live columns for a logical table. Row
// authorization is whole-table scoped; the host must reject shared-table
// registrations until the contract exposes a machine-readable subfact scope.
type ComponentColumnCatalog interface {
	Columns(context.Context, string) ([]string, error)
}

type sqlComponentColumnCatalog struct{ db *sql.DB }

func (c sqlComponentColumnCatalog) Columns(ctx context.Context, table string) ([]string, error) {
	if err := validateComponentSQLIdentifier(table); err != nil {
		return nil, err
	}
	if c.db == nil {
		return nil, fmt.Errorf("managed component column catalog database is required")
	}
	rows, err := c.db.QueryContext(ctx, `
SELECT COLUMN_NAME
FROM INFORMATION_SCHEMA.COLUMNS
WHERE TABLE_SCHEMA = DATABASE()
  AND TABLE_NAME = ?
ORDER BY ORDINAL_POSITION`, table)
	if err != nil {
		return nil, fmt.Errorf("query managed component columns for %s: %w", table, err)
	}
	defer rows.Close()
	columns := make([]string, 0)
	for rows.Next() {
		var column string
		if err := rows.Scan(&column); err != nil {
			return nil, fmt.Errorf("scan managed component column for %s: %w", table, err)
		}
		columns = append(columns, column)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate managed component columns for %s: %w", table, err)
	}
	return columns, nil
}

type componentSQLExecutor interface {
	QueryContext(context.Context, string, ...any) (*sql.Rows, error)
	ExecContext(context.Context, string, ...any) (sql.Result, error)
}

type managedComponentRowStore struct {
	exec     componentSQLExecutor
	catalog  ComponentColumnCatalog
	tables   map[string]struct{}
	columns  map[string]map[string]struct{}
	columnMu sync.Mutex
}

var _ cp.ComponentRowReader = (*managedComponentRowStore)(nil)

type managedComponentMutationStore struct{ *managedComponentRowStore }

var _ cp.ComponentMutationStore = (*managedComponentMutationStore)(nil)

func newManagedComponentRowReader(exec componentSQLExecutor, catalog ComponentColumnCatalog, writeSet []cp.ComponentWriteSurface) (cp.ComponentRowReader, error) {
	store, err := newManagedComponentRowStore(exec, catalog, writeSet)
	if err != nil {
		return nil, err
	}
	return store, nil
}

// NewTransactionComponentMutationStore binds the canonical SQL-backed
// component store to a caller-owned transaction. The caller remains
// responsible for committing or rolling back tx.
func NewTransactionComponentMutationStore(tx *sql.Tx, db *sql.DB, writeSet []cp.ComponentWriteSurface) (cp.ComponentMutationStore, error) {
	if db == nil {
		return nil, fmt.Errorf("managed component column catalog database is required")
	}
	return newManagedComponentMutationStore(tx, sqlComponentColumnCatalog{db: db}, writeSet)
}

func newManagedComponentMutationStore(tx *sql.Tx, catalog ComponentColumnCatalog, writeSet []cp.ComponentWriteSurface) (cp.ComponentMutationStore, error) {
	if tx == nil {
		return nil, fmt.Errorf("managed component transaction is required")
	}
	store, err := newManagedComponentRowStore(tx, catalog, writeSet)
	if err != nil {
		return nil, err
	}
	return &managedComponentMutationStore{managedComponentRowStore: store}, nil
}

func newManagedComponentRowStore(exec componentSQLExecutor, catalog ComponentColumnCatalog, writeSet []cp.ComponentWriteSurface) (*managedComponentRowStore, error) {
	if exec == nil {
		return nil, fmt.Errorf("managed component SQL executor is required")
	}
	if catalog == nil {
		return nil, fmt.Errorf("managed component column catalog is required")
	}
	tables := make(map[string]struct{}, len(writeSet))
	for _, surface := range writeSet {
		if err := validateComponentSQLIdentifier(surface.Table); err != nil {
			return nil, fmt.Errorf("managed component write set table: %w", err)
		}
		tables[surface.Table] = struct{}{}
	}
	if len(tables) == 0 {
		return nil, fmt.Errorf("managed component write set is required")
	}
	return &managedComponentRowStore{
		exec: exec, catalog: catalog, tables: tables,
		columns: make(map[string]map[string]struct{}),
	}, nil
}

func (s *managedComponentRowStore) ReadRows(ctx context.Context, query cp.ComponentRowQuery) ([]cp.ComponentRow, error) {
	return s.readRows(ctx, query, false)
}

func (s *managedComponentMutationStore) LockRows(ctx context.Context, query cp.ComponentRowQuery) ([]cp.ComponentRow, error) {
	return s.readRows(ctx, query, true)
}

func (s *managedComponentRowStore) readRows(ctx context.Context, query cp.ComponentRowQuery, lock bool) ([]cp.ComponentRow, error) {
	if err := query.Validate(); err != nil {
		return nil, err
	}
	matchColumns := sortedComponentRowKeys(query.Equal)
	columns := append(append([]string{}, query.Columns...), matchColumns...)
	if err := s.allow(ctx, query.Table, columns...); err != nil {
		return nil, err
	}
	where, args := componentPredicates(query.Equal, matchColumns)
	statement := "SELECT " + quoteComponentIdentifiers(query.Columns) + " FROM " + quoteComponentIdentifier(query.Table) + " WHERE " + where
	if lock {
		statement += " FOR UPDATE"
	}
	rows, err := s.exec.QueryContext(ctx, statement, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	out := make([]cp.ComponentRow, 0)
	for rows.Next() {
		values := make([]any, len(query.Columns))
		destinations := make([]any, len(values))
		for i := range values {
			destinations[i] = &values[i]
		}
		if err := rows.Scan(destinations...); err != nil {
			return nil, err
		}
		row := make(cp.ComponentRow, len(values))
		for i, column := range query.Columns {
			value := values[i]
			// database/sql preserves driver-specific text as []byte; providers
			// consume the neutral metadata contract as strings.
			if raw, ok := value.([]byte); ok {
				value = string(raw)
			}
			row[column] = value
		}
		out = append(out, row)
	}
	return out, rows.Err()
}

func (s *managedComponentMutationStore) NewID() string { return utils.GenerateID() }

func (s *managedComponentMutationStore) InsertRow(ctx context.Context, insert cp.ComponentInsert) error {
	if err := insert.Validate(); err != nil {
		return err
	}
	valueColumns := sortedComponentRowKeys(insert.Values)
	columns := append(append([]string{}, valueColumns...), insert.CurrentTimestampColumns...)
	if err := s.allow(ctx, insert.Table, columns...); err != nil {
		return err
	}
	expressions, args := componentMutationExpressions(insert.Values, valueColumns, insert.CurrentTimestampColumns, false)
	_, err := s.exec.ExecContext(ctx, "INSERT INTO "+quoteComponentIdentifier(insert.Table)+" ("+quoteComponentIdentifiers(columns)+") VALUES ("+strings.Join(expressions, ", ")+")", args...)
	return err
}

func (s *managedComponentMutationStore) UpdateRows(ctx context.Context, update cp.ComponentUpdate) (int64, error) {
	if err := update.Validate(); err != nil {
		return 0, err
	}
	valueColumns, matchColumns := sortedComponentRowKeys(update.Values), sortedComponentRowKeys(update.Match)
	columns := append(append(append([]string{}, valueColumns...), update.CurrentTimestampColumns...), matchColumns...)
	if err := s.allow(ctx, update.Table, columns...); err != nil {
		return 0, err
	}
	expressions, args := componentMutationExpressions(update.Values, valueColumns, update.CurrentTimestampColumns, true)
	where, matchArgs := componentPredicates(update.Match, matchColumns)
	result, err := s.exec.ExecContext(ctx, "UPDATE "+quoteComponentIdentifier(update.Table)+" SET "+strings.Join(expressions, ", ")+" WHERE "+where, append(args, matchArgs...)...)
	return componentRowsAffected(result, err)
}

func (s *managedComponentMutationStore) DeleteRows(ctx context.Context, delete cp.ComponentDelete) (int64, error) {
	if err := delete.Validate(); err != nil {
		return 0, err
	}
	matchColumns := sortedComponentRowKeys(delete.Match)
	if err := s.allow(ctx, delete.Table, matchColumns...); err != nil {
		return 0, err
	}
	where, args := componentPredicates(delete.Match, matchColumns)
	result, err := s.exec.ExecContext(ctx, "DELETE FROM "+quoteComponentIdentifier(delete.Table)+" WHERE "+where, args...)
	return componentRowsAffected(result, err)
}

func (s *managedComponentRowStore) allow(ctx context.Context, table string, columns ...string) error {
	if err := validateComponentSQLIdentifier(table); err != nil {
		return err
	}
	if _, ok := s.tables[table]; !ok {
		return fmt.Errorf("table %q is not in the component write set", table)
	}
	s.columnMu.Lock()
	allowed, ok := s.columns[table]
	if !ok {
		listed, err := s.catalog.Columns(ctx, table)
		if err != nil {
			s.columnMu.Unlock()
			return fmt.Errorf("resolve columns for %s: %w", table, err)
		}
		allowed = make(map[string]struct{}, len(listed))
		for _, column := range listed {
			if err := validateComponentSQLIdentifier(column); err != nil {
				s.columnMu.Unlock()
				return fmt.Errorf("catalog column for %s: %w", table, err)
			}
			allowed[column] = struct{}{}
		}
		s.columns[table] = allowed
	}
	s.columnMu.Unlock()
	for _, column := range columns {
		if err := validateComponentSQLIdentifier(column); err != nil {
			return err
		}
		if _, ok := allowed[column]; !ok {
			return fmt.Errorf("column %q is not present in the live schema for table %q", column, table)
		}
	}
	return nil
}

func validateComponentSQLIdentifier(identifier string) error {
	if identifier == "" || strings.TrimSpace(identifier) != identifier {
		return fmt.Errorf("invalid SQL identifier %q", identifier)
	}
	for i := 0; i < len(identifier); i++ {
		character := identifier[i]
		if (character >= 'a' && character <= 'z') || (character >= 'A' && character <= 'Z') || character == '_' || (i > 0 && character >= '0' && character <= '9') {
			continue
		}
		return fmt.Errorf("invalid SQL identifier %q", identifier)
	}
	return nil
}

func sortedComponentRowKeys(row cp.ComponentRow) []string {
	keys := make([]string, 0, len(row))
	for key := range row {
		keys = append(keys, key)
	}
	sort.Strings(keys)
	return keys
}

func componentPredicates(row cp.ComponentRow, columns []string) (string, []any) {
	parts, args := make([]string, 0, len(columns)), make([]any, 0, len(columns))
	for _, column := range columns {
		if row[column] == nil {
			parts = append(parts, quoteComponentIdentifier(column)+" IS NULL")
			continue
		}
		parts = append(parts, quoteComponentIdentifier(column)+" = ?")
		args = append(args, row[column])
	}
	return strings.Join(parts, " AND "), args
}

func componentMutationExpressions(values cp.ComponentRow, valueColumns, timestampColumns []string, assignments bool) ([]string, []any) {
	parts, args := make([]string, 0, len(valueColumns)+len(timestampColumns)), make([]any, 0, len(valueColumns))
	for _, column := range valueColumns {
		expression := "?"
		if assignments {
			expression = quoteComponentIdentifier(column) + " = ?"
		}
		parts = append(parts, expression)
		args = append(args, values[column])
	}
	for _, column := range timestampColumns {
		expression := "CURRENT_TIMESTAMP"
		if assignments {
			expression = quoteComponentIdentifier(column) + " = CURRENT_TIMESTAMP"
		}
		parts = append(parts, expression)
	}
	return parts, args
}

func quoteComponentIdentifier(identifier string) string { return "`" + identifier + "`" }

func quoteComponentIdentifiers(identifiers []string) string {
	quoted := make([]string, len(identifiers))
	for i, identifier := range identifiers {
		quoted[i] = quoteComponentIdentifier(identifier)
	}
	return strings.Join(quoted, ", ")
}

func componentRowsAffected(result sql.Result, err error) (int64, error) {
	if err != nil {
		return 0, err
	}
	return result.RowsAffected()
}