// Package sandbox — db_module.go
//
//
// Permissions:
//   db.read  → db.query(), db.count(), db.aggregate(), db.query_batch(), db.view(), db.list_tables()
//   db.write → all of the above + db.insert(), db.update(), db.delete()
//
// Starlark API:
//
//	rows  = db.query("logs", filters={"user_id": "abc"}, order="created_at", limit=50, before={"created_at": "2026-01-01"}, after={"count": 5}, search_like={"title": "%hello%"})
//	n     = db.count("logs", filters={"user_id": "abc"})
//	val   = db.aggregate("orders", "amount", "sum", filters={"status": "paid"})
//	batch = db.query_batch([{"table": "logs", "filters": {"user_id": "abc"}}, {"table": "tasks"}])
//	row   = db.insert("logs", {"message": "hello"})
//	ok    = db.update("logs", row_id, {"message": "updated"})
//	ok    = db.delete("logs", row_id)
//	names = db.list_tables()
//	rows  = db.view("users", filters={"id": "abc"})
//
// All table access is scoped to ext_{package_id}_{table_name}.
// Platform views (ext_view_users) are accessible via db.view()
// and are read-only regardless of permission level.
//
// Queries are parameterized — no raw SQL is ever accepted from scripts.
// Dialect differences (placeholder style) are handled internally.
package sandbox

import (
	"context"
	"crypto/rand"
	"database/sql"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"math"
	"sort"
	"strings"

	"go.starlark.net/starlark"
	"go.starlark.net/starlarkstruct"
)

// DBModuleConfig holds configuration for a db module instance.
type DBModuleConfig struct {
	PackageID   string
	CanWrite    bool // true when db.write permission is granted
	DB          *sql.DB
	IsPostgres  bool
	HasPgvector bool // true when pgvector extension is available on Postgres
}

// allowedViews is the set of platform views extensions may query via db.view().
var allowedViews = map[string]bool{
	"users": true,
}

// BuildDBModule creates the "db" starlark module for an extension.
// The module enforces namespace isolation — all physical table names are
// prefixed with ext_{packageID}_. Write operations are guarded by CanWrite.
func BuildDBModule(ctx context.Context, cfg DBModuleConfig) *starlarkstruct.Module {
	fns := starlark.StringDict{
		"query":         starlark.NewBuiltin("db.query", dbQuery(ctx, cfg)),
		"count":         starlark.NewBuiltin("db.count", dbCount(ctx, cfg)),
		"aggregate":     starlark.NewBuiltin("db.aggregate", dbAggregate(ctx, cfg)),
		"query_batch":   starlark.NewBuiltin("db.query_batch", dbQueryBatch(ctx, cfg)),
		"query_similar": starlark.NewBuiltin("db.query_similar", dbQuerySimilar(ctx, cfg)),
		"view":          starlark.NewBuiltin("db.view", dbView(ctx, cfg)),
		"list_tables":   starlark.NewBuiltin("db.list_tables", dbListTables(ctx, cfg)),
	}

	if cfg.CanWrite {
		fns["insert"] = starlark.NewBuiltin("db.insert", dbInsert(ctx, cfg))
		fns["update"] = starlark.NewBuiltin("db.update", dbUpdate(ctx, cfg))
		fns["delete"] = starlark.NewBuiltin("db.delete", dbDelete(ctx, cfg))
	}

	return MakeModule("db", fns)
}

// ── Helpers ──────────────────────────────────

// physicalTable returns the fully-qualified table name for an extension table.
func (cfg DBModuleConfig) physicalTable(logicalName string) (string, error) {
	if logicalName == "" {
		return "", fmt.Errorf("db: table name must not be empty")
	}
	if strings.ContainsAny(logicalName, " \t\n\"';-") {
		return "", fmt.Errorf("db: invalid table name %q", logicalName)
	}
	// Sanitize package ID for use in identifiers (replace - with _)
	pkgSlug := strings.ReplaceAll(cfg.PackageID, "-", "_")
	return fmt.Sprintf("ext_%s_%s", pkgSlug, logicalName), nil
}

// ph returns the parameter placeholder for the nth argument (1-based).
func (cfg DBModuleConfig) ph(n int) string {
	if cfg.IsPostgres {
		return fmt.Sprintf("$%d", n)
	}
	return "?"
}

// generateID produces a random hex ID for new rows.
func generateID() string {
	b := make([]byte, 16)
	_, _ = rand.Read(b)
	return hex.EncodeToString(b)
}

// starlarkFiltersToSQL converts a Starlark dict of {col: val} into a
// WHERE clause and argument slice. Returns empty string if filters is None.
func (cfg DBModuleConfig) starlarkFiltersToSQL(filters starlark.Value, startIdx int) (string, []any, error) {
	if filters == starlark.None || filters == nil {
		return "", nil, nil
	}
	d, ok := filters.(*starlark.Dict)
	if !ok {
		return "", nil, fmt.Errorf("db: filters must be a dict, got %s", filters.Type())
	}
	if d.Len() == 0 {
		return "", nil, nil
	}

	var parts []string
	var args []any
	idx := startIdx

	for _, item := range d.Items() {
		col, ok := item[0].(starlark.String)
		if !ok {
			return "", nil, fmt.Errorf("db: filter key must be a string, got %s", item[0].Type())
		}
		colStr := string(col)
		if strings.ContainsAny(colStr, " \t\n\"';-") {
			return "", nil, fmt.Errorf("db: invalid column name %q", colStr)
		}

		val, err := starlarkToGoValue(item[1])
		if err != nil {
			return "", nil, fmt.Errorf("db: filter value for %q: %w", colStr, err)
		}

		parts = append(parts, fmt.Sprintf("%s = %s", colStr, cfg.ph(idx)))
		args = append(args, val)
		idx++
	}

	return "WHERE " + strings.Join(parts, " AND "), args, nil
}

// starlarkToGoValue converts a Starlark value to a Go value suitable for SQL.
func starlarkToGoValue(v starlark.Value) (any, error) {
	switch val := v.(type) {
	case starlark.String:
		return string(val), nil
	case starlark.Int:
		i, ok := val.Int64()
		if !ok {
			return nil, fmt.Errorf("integer overflow")
		}
		return i, nil
	case starlark.Float:
		return float64(val), nil
	case starlark.Bool:
		return bool(val), nil
	case starlark.NoneType:
		return nil, nil
	case *starlark.List:
		goSlice := make([]any, val.Len())
		for i := 0; i < val.Len(); i++ {
			elem, err := starlarkToGoValue(val.Index(i))
			if err != nil {
				return nil, fmt.Errorf("list[%d]: %w", i, err)
			}
			goSlice[i] = elem
		}
		b, err := json.Marshal(goSlice)
		if err != nil {
			return nil, fmt.Errorf("list marshal: %w", err)
		}
		return string(b), nil
	default:
		return nil, fmt.Errorf("unsupported type %s", v.Type())
	}
}

// rowsToStarlark converts sql.Rows to a Starlark list of dicts.
func rowsToStarlark(rows *sql.Rows) (*starlark.List, error) {
	cols, err := rows.Columns()
	if err != nil {
		return nil, err
	}

	var result []starlark.Value
	for rows.Next() {
		vals := make([]any, len(cols))
		ptrs := make([]any, len(cols))
		for i := range vals {
			ptrs[i] = &vals[i]
		}
		if err := rows.Scan(ptrs...); err != nil {
			return nil, err
		}

		d := starlark.NewDict(len(cols))
		for i, col := range cols {
			sv, err := goToStarlark(vals[i])
			if err != nil {
				return nil, fmt.Errorf("column %q: %w", col, err)
			}
			if err := d.SetKey(starlark.String(col), sv); err != nil {
				return nil, err
			}
		}
		result = append(result, d)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return starlark.NewList(result), nil
}

// goToStarlark converts a Go value (from sql.Scan) to a Starlark value.
func goToStarlark(v any) (starlark.Value, error) {
	if v == nil {
		return starlark.None, nil
	}
	switch val := v.(type) {
	case string:
		return starlark.String(val), nil
	case []byte:
		return starlark.String(string(val)), nil
	case int64:
		return starlark.MakeInt64(val), nil
	case float64:
		return starlark.Float(val), nil
	case bool:
		return starlark.Bool(val), nil
	default:
		// Fallback: stringify
		return starlark.String(fmt.Sprintf("%v", val)), nil
	}
}

// ── Builtins ─────────────────────────────────

// starlarkRangeToSQL converts a Starlark dict of {col: val} into range
// comparison clauses (e.g. col < $N or col > $N). op must be "<" or ">".
func (cfg DBModuleConfig) starlarkRangeToSQL(rangeVal starlark.Value, op string, startIdx int) ([]string, []any, error) {
	if rangeVal == starlark.None || rangeVal == nil {
		return nil, nil, nil
	}
	d, ok := rangeVal.(*starlark.Dict)
	if !ok {
		return nil, nil, fmt.Errorf("db: range param must be a dict, got %s", rangeVal.Type())
	}
	if d.Len() == 0 {
		return nil, nil, nil
	}

	var parts []string
	var args []any
	idx := startIdx

	for _, item := range d.Items() {
		col, ok := item[0].(starlark.String)
		if !ok {
			return nil, nil, fmt.Errorf("db: range key must be a string, got %s", item[0].Type())
		}
		colStr := string(col)
		if strings.ContainsAny(colStr, " \t\n\"';-") {
			return nil, nil, fmt.Errorf("db: invalid column name %q", colStr)
		}

		val, err := starlarkToGoValue(item[1])
		if err != nil {
			return nil, nil, fmt.Errorf("db: range value for %q: %w", colStr, err)
		}

		parts = append(parts, fmt.Sprintf("%s %s %s", colStr, op, cfg.ph(idx)))
		args = append(args, val)
		idx++
	}

	return parts, args, nil
}

// starlarkSearchLikeToSQL converts a Starlark dict of {col: pattern} into
// a parenthesized OR group of LIKE (SQLite) / ILIKE (Postgres) clauses.
// Example: {"title": "%hello%", "content": "%hello%"} → (title ILIKE $1 OR content ILIKE $2)
func (cfg DBModuleConfig) starlarkSearchLikeToSQL(searchVal starlark.Value, startIdx int) (string, []any, error) {
	if searchVal == starlark.None || searchVal == nil {
		return "", nil, nil
	}
	d, ok := searchVal.(*starlark.Dict)
	if !ok {
		return "", nil, fmt.Errorf("db: search_like must be a dict, got %s", searchVal.Type())
	}
	if d.Len() == 0 {
		return "", nil, nil
	}

	var parts []string
	var args []any
	idx := startIdx

	op := "LIKE" // SQLite — case-insensitive for ASCII by default
	if cfg.IsPostgres {
		op = "ILIKE" // Postgres — explicit case-insensitive
	}

	for _, item := range d.Items() {
		col, ok := item[0].(starlark.String)
		if !ok {
			return "", nil, fmt.Errorf("db: search_like key must be a string, got %s", item[0].Type())
		}
		colStr := string(col)
		if strings.ContainsAny(colStr, " \t\n\"';-") {
			return "", nil, fmt.Errorf("db: invalid column name %q", colStr)
		}

		val, err := starlarkToGoValue(item[1])
		if err != nil {
			return "", nil, fmt.Errorf("db: search_like value for %q: %w", colStr, err)
		}

		parts = append(parts, fmt.Sprintf("%s %s %s", colStr, op, cfg.ph(idx)))
		args = append(args, val)
		idx++
	}

	return "(" + strings.Join(parts, " OR ") + ")", args, nil
}

// querySpec holds parsed parameters for a single SELECT query.
type querySpec struct {
	table      string
	filters    starlark.Value // None or *Dict
	order      starlark.Value // None or String
	limit      int64
	before     starlark.Value
	after      starlark.Value
	searchLike starlark.Value
}

// buildSelectQuery constructs a "SELECT * FROM ... WHERE ... ORDER BY ... LIMIT ..."
// SQL statement from a querySpec. Returns the SQL string and args slice.
// Reused by dbQuery and dbQueryBatch.
func (cfg DBModuleConfig) buildSelectQuery(spec querySpec) (string, []any, error) {
	physTable, err := cfg.physicalTable(spec.table)
	if err != nil {
		return "", nil, err
	}

	whereClause, whereArgs, err := cfg.starlarkFiltersToSQL(spec.filters, 1)
	if err != nil {
		return "", nil, err
	}

	beforeParts, beforeArgs, err := cfg.starlarkRangeToSQL(spec.before, "<", len(whereArgs)+1)
	if err != nil {
		return "", nil, err
	}
	afterParts, afterArgs, err := cfg.starlarkRangeToSQL(spec.after, ">", len(whereArgs)+len(beforeArgs)+1)
	if err != nil {
		return "", nil, err
	}

	searchClause, searchArgs, err := cfg.starlarkSearchLikeToSQL(spec.searchLike, len(whereArgs)+len(beforeArgs)+len(afterArgs)+1)
	if err != nil {
		return "", nil, err
	}

	var allParts []string
	var allArgs []any

	if whereClause != "" {
		allParts = append(allParts, strings.TrimPrefix(whereClause, "WHERE "))
		allArgs = append(allArgs, whereArgs...)
	} else {
		allArgs = append(allArgs, whereArgs...)
	}
	allParts = append(allParts, beforeParts...)
	allArgs = append(allArgs, beforeArgs...)
	allParts = append(allParts, afterParts...)
	allArgs = append(allArgs, afterArgs...)
	if searchClause != "" {
		allParts = append(allParts, searchClause)
		allArgs = append(allArgs, searchArgs...)
	}

	lim := spec.limit
	if lim < 1 || lim > 1000 {
		lim = 100
	}

	query := fmt.Sprintf("SELECT * FROM %s", physTable)
	if len(allParts) > 0 {
		query += " WHERE " + strings.Join(allParts, " AND ")
	}

	if spec.order != starlark.None && spec.order != nil {
		col, ok := spec.order.(starlark.String)
		if !ok {
			return "", nil, fmt.Errorf("db.query: order must be a string column name")
		}
		colStr := string(col)
		dir := "ASC"
		if strings.HasPrefix(colStr, "-") {
			colStr = colStr[1:]
			dir = "DESC"
		}
		if strings.ContainsAny(colStr, " \t\n\"';") {
			return "", nil, fmt.Errorf("db.query: invalid order column %q", colStr)
		}
		query += fmt.Sprintf(" ORDER BY %s %s", colStr, dir)
	}

	limitPH := cfg.ph(len(allArgs) + 1)
	query += " LIMIT " + limitPH
	allArgs = append(allArgs, lim)

	return query, allArgs, nil
}

// dbQuery implements db.query(table, filters=None, order=None, limit=100, before=None, after=None, search_like=None).
func dbQuery(ctx context.Context, cfg DBModuleConfig) func(*starlark.Thread, *starlark.Builtin, starlark.Tuple, []starlark.Tuple) (starlark.Value, error) {
	return func(thread *starlark.Thread, b *starlark.Builtin, args starlark.Tuple, kwargs []starlark.Tuple) (starlark.Value, error) {
		var table string
		var filters starlark.Value = starlark.None
		var order starlark.Value = starlark.None
		var limit starlark.Int = starlark.MakeInt(100)
		var before starlark.Value = starlark.None
		var after starlark.Value = starlark.None
		var searchLike starlark.Value = starlark.None

		if err := starlark.UnpackArgs(b.Name(), args, kwargs,
			"table", &table,
			"filters?", &filters,
			"order?", &order,
			"limit?", &limit,
			"before?", &before,
			"after?", &after,
			"search_like?", &searchLike,
		); err != nil {
			return nil, err
		}

		lim, ok := limit.Int64()
		if !ok || lim < 1 || lim > 1000 {
			lim = 100
		}

		query, allArgs, err := cfg.buildSelectQuery(querySpec{
			table:      table,
			filters:    filters,
			order:      order,
			limit:      lim,
			before:     before,
			after:      after,
			searchLike: searchLike,
		})
		if err != nil {
			return nil, err
		}

		rows, err := cfg.DB.QueryContext(ctx, query, allArgs...)
		if err != nil {
			return nil, fmt.Errorf("db.query: %w", err)
		}
		defer rows.Close()

		return rowsToStarlark(rows)
	}
}

// dbCount implements db.count(table, filters={}).
// Returns the integer count of rows matching the given filters.
func dbCount(ctx context.Context, cfg DBModuleConfig) func(*starlark.Thread, *starlark.Builtin, starlark.Tuple, []starlark.Tuple) (starlark.Value, error) {
	return func(thread *starlark.Thread, b *starlark.Builtin, args starlark.Tuple, kwargs []starlark.Tuple) (starlark.Value, error) {
		var table string
		var filters starlark.Value = starlark.None

		if err := starlark.UnpackArgs(b.Name(), args, kwargs,
			"table", &table,
			"filters?", &filters,
		); err != nil {
			return nil, err
		}

		physTable, err := cfg.physicalTable(table)
		if err != nil {
			return nil, err
		}

		whereClause, whereArgs, err := cfg.starlarkFiltersToSQL(filters, 1)
		if err != nil {
			return nil, err
		}

		query := fmt.Sprintf("SELECT COUNT(*) FROM %s", physTable)
		if whereClause != "" {
			query += " " + whereClause
		}

		var count int64
		if err := cfg.DB.QueryRowContext(ctx, query, whereArgs...).Scan(&count); err != nil {
			return nil, fmt.Errorf("db.count: %w", err)
		}

		return starlark.MakeInt64(count), nil
	}
}

// allowedAggOps is the set of valid aggregation operations.
var allowedAggOps = map[string]bool{
	"count": true,
	"sum":   true,
	"avg":   true,
	"min":   true,
	"max":   true,
}

// dbAggregate implements db.aggregate(table, column, op, filters={}).
// op must be one of: count, sum, avg, min, max.
// Returns int, float, or None (if no matching rows).
func dbAggregate(ctx context.Context, cfg DBModuleConfig) func(*starlark.Thread, *starlark.Builtin, starlark.Tuple, []starlark.Tuple) (starlark.Value, error) {
	return func(thread *starlark.Thread, b *starlark.Builtin, args starlark.Tuple, kwargs []starlark.Tuple) (starlark.Value, error) {
		var table, column, op string
		var filters starlark.Value = starlark.None

		if err := starlark.UnpackArgs(b.Name(), args, kwargs,
			"table", &table,
			"column", &column,
			"op", &op,
			"filters?", &filters,
		); err != nil {
			return nil, err
		}

		if !allowedAggOps[op] {
			return nil, fmt.Errorf("db.aggregate: invalid op %q (allowed: count, sum, avg, min, max)", op)
		}

		if strings.ContainsAny(column, " \t\n\"';-") {
			return nil, fmt.Errorf("db.aggregate: invalid column name %q", column)
		}

		physTable, err := cfg.physicalTable(table)
		if err != nil {
			return nil, err
		}

		whereClause, whereArgs, err := cfg.starlarkFiltersToSQL(filters, 1)
		if err != nil {
			return nil, err
		}

		query := fmt.Sprintf("SELECT %s(%s) FROM %s", strings.ToUpper(op), column, physTable)
		if whereClause != "" {
			query += " " + whereClause
		}

		var result sql.NullFloat64
		if err := cfg.DB.QueryRowContext(ctx, query, whereArgs...).Scan(&result); err != nil {
			return nil, fmt.Errorf("db.aggregate: %w", err)
		}

		if !result.Valid {
			return starlark.None, nil
		}

		// Return int if the value is a whole number.
		if result.Float64 == float64(int64(result.Float64)) {
			return starlark.MakeInt64(int64(result.Float64)), nil
		}
		return starlark.Float(result.Float64), nil
	}
}

// dbQueryBatch implements db.query_batch(queries).
// Each element in the queries list is a dict with keys:
//
//	table (required), filters, order, limit, before, after, search_like (all optional).
//
// Returns a list of result lists — one per query spec.
func dbQueryBatch(ctx context.Context, cfg DBModuleConfig) func(*starlark.Thread, *starlark.Builtin, starlark.Tuple, []starlark.Tuple) (starlark.Value, error) {
	return func(thread *starlark.Thread, b *starlark.Builtin, args starlark.Tuple, kwargs []starlark.Tuple) (starlark.Value, error) {
		var queries *starlark.List

		if err := starlark.UnpackArgs(b.Name(), args, kwargs,
			"queries", &queries,
		); err != nil {
			return nil, err
		}

		n := queries.Len()
		if n == 0 {
			return nil, fmt.Errorf("db.query_batch: queries list is empty")
		}
		if n > 10 {
			return nil, fmt.Errorf("db.query_batch: max 10 queries, got %d", n)
		}

		var results []starlark.Value
		for i := 0; i < n; i++ {
			spec, err := dictToQuerySpec(queries.Index(i))
			if err != nil {
				return nil, fmt.Errorf("db.query_batch[%d]: %w", i, err)
			}

			query, queryArgs, err := cfg.buildSelectQuery(spec)
			if err != nil {
				return nil, fmt.Errorf("db.query_batch[%d]: %w", i, err)
			}

			rows, err := cfg.DB.QueryContext(ctx, query, queryArgs...)
			if err != nil {
				return nil, fmt.Errorf("db.query_batch[%d]: %w", i, err)
			}

			resultList, err := rowsToStarlark(rows)
			rows.Close()
			if err != nil {
				return nil, fmt.Errorf("db.query_batch[%d]: %w", i, err)
			}

			results = append(results, resultList)
		}

		return starlark.NewList(results), nil
	}
}

// dictToQuerySpec extracts query parameters from a Starlark dict.
func dictToQuerySpec(v starlark.Value) (querySpec, error) {
	d, ok := v.(*starlark.Dict)
	if !ok {
		return querySpec{}, fmt.Errorf("each query must be a dict, got %s", v.Type())
	}

	spec := querySpec{
		filters:    starlark.None,
		order:      starlark.None,
		limit:      100,
		before:     starlark.None,
		after:      starlark.None,
		searchLike: starlark.None,
	}

	tableVal, found, err := d.Get(starlark.String("table"))
	if err != nil {
		return querySpec{}, err
	}
	if !found || tableVal == starlark.None {
		return querySpec{}, fmt.Errorf("missing required key \"table\"")
	}
	tableStr, ok := starlark.AsString(tableVal)
	if !ok {
		return querySpec{}, fmt.Errorf("\"table\" must be a string, got %s", tableVal.Type())
	}
	spec.table = tableStr

	if v, found, _ := d.Get(starlark.String("filters")); found && v != starlark.None {
		spec.filters = v
	}
	if v, found, _ := d.Get(starlark.String("order")); found && v != starlark.None {
		spec.order = v
	}
	if v, found, _ := d.Get(starlark.String("limit")); found && v != starlark.None {
		if limInt, ok := v.(starlark.Int); ok {
			lim, _ := limInt.Int64()
			spec.limit = lim
		}
	}
	if v, found, _ := d.Get(starlark.String("before")); found && v != starlark.None {
		spec.before = v
	}
	if v, found, _ := d.Get(starlark.String("after")); found && v != starlark.None {
		spec.after = v
	}
	if v, found, _ := d.Get(starlark.String("search_like")); found && v != starlark.None {
		spec.searchLike = v
	}

	return spec, nil
}

// dbView implements db.view(view_name, filters=None, limit=100).
// Queries ext_view_{view_name} — only allowedViews are permitted.
func dbView(ctx context.Context, cfg DBModuleConfig) func(*starlark.Thread, *starlark.Builtin, starlark.Tuple, []starlark.Tuple) (starlark.Value, error) {
	return func(thread *starlark.Thread, b *starlark.Builtin, args starlark.Tuple, kwargs []starlark.Tuple) (starlark.Value, error) {
		var viewName string
		var filters starlark.Value = starlark.None
		var limit starlark.Int = starlark.MakeInt(100)

		if err := starlark.UnpackArgs(b.Name(), args, kwargs,
			"view_name", &viewName,
			"filters?", &filters,
			"limit?", &limit,
		); err != nil {
			return nil, err
		}

		if !allowedViews[viewName] {
			return nil, fmt.Errorf("db.view: unknown view %q (allowed: users)", viewName)
		}

		physView := "ext_view_" + viewName

		whereClause, whereArgs, err := cfg.starlarkFiltersToSQL(filters, 1)
		if err != nil {
			return nil, err
		}

		lim, ok := limit.Int64()
		if !ok || lim < 1 || lim > 1000 {
			lim = 100
		}

		query := fmt.Sprintf("SELECT * FROM %s", physView)
		if whereClause != "" {
			query += " " + whereClause
		}
		limitPH := cfg.ph(len(whereArgs) + 1)
		query += " LIMIT " + limitPH
		whereArgs = append(whereArgs, lim)

		rows, err := cfg.DB.QueryContext(ctx, query, whereArgs...)
		if err != nil {
			return nil, fmt.Errorf("db.view: %w", err)
		}
		defer rows.Close()

		return rowsToStarlark(rows)
	}
}

// dbListTables implements db.list_tables() → list of logical table names.
func dbListTables(ctx context.Context, cfg DBModuleConfig) func(*starlark.Thread, *starlark.Builtin, starlark.Tuple, []starlark.Tuple) (starlark.Value, error) {
	return func(thread *starlark.Thread, b *starlark.Builtin, args starlark.Tuple, kwargs []starlark.Tuple) (starlark.Value, error) {
		if err := starlark.UnpackPositionalArgs(b.Name(), args, kwargs, 0); err != nil {
			return nil, err
		}

		prefix := "ext_" + strings.ReplaceAll(cfg.PackageID, "-", "_") + "_"
		var query string
		var queryArgs []any

		if cfg.IsPostgres {
			query = `SELECT table_name FROM information_schema.tables
			          WHERE table_schema = 'public' AND table_name LIKE $1
			          ORDER BY table_name`
			queryArgs = []any{prefix + "%"}
		} else {
			query = `SELECT name FROM sqlite_master
			          WHERE type = 'table' AND name LIKE ?
			          ORDER BY name`
			queryArgs = []any{prefix + "%"}
		}

		rows, err := cfg.DB.QueryContext(ctx, query, queryArgs...)
		if err != nil {
			return nil, fmt.Errorf("db.list_tables: %w", err)
		}
		defer rows.Close()

		var names []starlark.Value
		for rows.Next() {
			var physName string
			if err := rows.Scan(&physName); err != nil {
				return nil, err
			}
			// Strip prefix to return logical name
			logical := strings.TrimPrefix(physName, prefix)
			names = append(names, starlark.String(logical))
		}
		if names == nil {
			names = []starlark.Value{}
		}
		return starlark.NewList(names), nil
	}
}

// dbInsert implements db.insert(table, row_dict) → inserted row dict with id.
func dbInsert(ctx context.Context, cfg DBModuleConfig) func(*starlark.Thread, *starlark.Builtin, starlark.Tuple, []starlark.Tuple) (starlark.Value, error) {
	return func(thread *starlark.Thread, b *starlark.Builtin, args starlark.Tuple, kwargs []starlark.Tuple) (starlark.Value, error) {
		var table string
		var rowVal starlark.Value

		if err := starlark.UnpackPositionalArgs(b.Name(), args, kwargs, 2, &table, &rowVal); err != nil {
			return nil, err
		}

		row, ok := rowVal.(*starlark.Dict)
		if !ok {
			return nil, fmt.Errorf("db.insert: row must be a dict, got %s", rowVal.Type())
		}

		physTable, err := cfg.physicalTable(table)
		if err != nil {
			return nil, err
		}

		id := generateID()
		cols := []string{"id"}
		placeholders := []string{cfg.ph(1)}
		vals := []any{id}
		idx := 2

		for _, item := range row.Items() {
			col, ok := item[0].(starlark.String)
			if !ok {
				return nil, fmt.Errorf("db.insert: column key must be a string")
			}
			colStr := string(col)
			if colStr == "id" || colStr == "created_at" {
				continue // auto-managed
			}
			if strings.ContainsAny(colStr, " \t\n\"';-") {
				return nil, fmt.Errorf("db.insert: invalid column name %q", colStr)
			}
			val, err := starlarkToGoValue(item[1])
			if err != nil {
				return nil, fmt.Errorf("db.insert: column %q: %w", colStr, err)
			}
			cols = append(cols, colStr)
			placeholders = append(placeholders, cfg.ph(idx))
			vals = append(vals, val)
			idx++
		}

		query := fmt.Sprintf("INSERT INTO %s (%s) VALUES (%s)",
			physTable,
			strings.Join(cols, ", "),
			strings.Join(placeholders, ", "),
		)

		if _, err := cfg.DB.ExecContext(ctx, query, vals...); err != nil {
			return nil, fmt.Errorf("db.insert: %w", err)
		}

		// Return the inserted row (re-query by id)
		selectQ := fmt.Sprintf("SELECT * FROM %s WHERE id = %s", physTable, cfg.ph(1))
		rows, err := cfg.DB.QueryContext(ctx, selectQ, id)
		if err != nil {
			return nil, fmt.Errorf("db.insert: re-query failed: %w", err)
		}
		defer rows.Close()

		list, err := rowsToStarlark(rows)
		if err != nil {
			return nil, err
		}
		if list.Len() == 0 {
			return starlark.None, nil
		}
		return list.Index(0), nil
	}
}

// dbUpdate implements db.update(table, id, partial_dict) → True.
func dbUpdate(ctx context.Context, cfg DBModuleConfig) func(*starlark.Thread, *starlark.Builtin, starlark.Tuple, []starlark.Tuple) (starlark.Value, error) {
	return func(thread *starlark.Thread, b *starlark.Builtin, args starlark.Tuple, kwargs []starlark.Tuple) (starlark.Value, error) {
		var table, rowID string
		var patchVal starlark.Value

		if err := starlark.UnpackPositionalArgs(b.Name(), args, kwargs, 3, &table, &rowID, &patchVal); err != nil {
			return nil, err
		}

		patch, ok := patchVal.(*starlark.Dict)
		if !ok {
			return nil, fmt.Errorf("db.update: patch must be a dict, got %s", patchVal.Type())
		}
		if patch.Len() == 0 {
			return starlark.True, nil // nothing to do
		}

		physTable, err := cfg.physicalTable(table)
		if err != nil {
			return nil, err
		}

		var setParts []string
		var vals []any
		idx := 1

		for _, item := range patch.Items() {
			col, ok := item[0].(starlark.String)
			if !ok {
				return nil, fmt.Errorf("db.update: column key must be a string")
			}
			colStr := string(col)
			if colStr == "id" || colStr == "created_at" {
				continue
			}
			if strings.ContainsAny(colStr, " \t\n\"';-") {
				return nil, fmt.Errorf("db.update: invalid column name %q", colStr)
			}
			val, err := starlarkToGoValue(item[1])
			if err != nil {
				return nil, fmt.Errorf("db.update: column %q: %w", colStr, err)
			}
			setParts = append(setParts, fmt.Sprintf("%s = %s", colStr, cfg.ph(idx)))
			vals = append(vals, val)
			idx++
		}

		if len(setParts) == 0 {
			return starlark.True, nil
		}

		vals = append(vals, rowID)
		query := fmt.Sprintf("UPDATE %s SET %s WHERE id = %s",
			physTable,
			strings.Join(setParts, ", "),
			cfg.ph(idx),
		)

		if _, err := cfg.DB.ExecContext(ctx, query, vals...); err != nil {
			return nil, fmt.Errorf("db.update: %w", err)
		}
		return starlark.True, nil
	}
}

// dbDelete implements db.delete(table, id) → True.
func dbDelete(ctx context.Context, cfg DBModuleConfig) func(*starlark.Thread, *starlark.Builtin, starlark.Tuple, []starlark.Tuple) (starlark.Value, error) {
	return func(thread *starlark.Thread, b *starlark.Builtin, args starlark.Tuple, kwargs []starlark.Tuple) (starlark.Value, error) {
		var table, rowID string
		if err := starlark.UnpackPositionalArgs(b.Name(), args, kwargs, 2, &table, &rowID); err != nil {
			return nil, err
		}

		physTable, err := cfg.physicalTable(table)
		if err != nil {
			return nil, err
		}

		query := fmt.Sprintf("DELETE FROM %s WHERE id = %s", physTable, cfg.ph(1))
		if _, err := cfg.DB.ExecContext(ctx, query, rowID); err != nil {
			return nil, fmt.Errorf("db.delete: %w", err)
		}
		return starlark.True, nil
	}
}

// ── Vector Similarity ─────────────────────────

// starlarkListToFloats converts a Starlark list to a Go float64 slice.
func starlarkListToFloats(list *starlark.List) ([]float64, error) {
	n := list.Len()
	if n == 0 {
		return nil, fmt.Errorf("vector must not be empty")
	}
	out := make([]float64, n)
	for i := 0; i < n; i++ {
		switch v := list.Index(i).(type) {
		case starlark.Float:
			out[i] = float64(v)
		case starlark.Int:
			i64, _ := v.Int64()
			out[i] = float64(i64)
		default:
			return nil, fmt.Errorf("vector[%d]: expected number, got %s", i, list.Index(i).Type())
		}
	}
	return out, nil
}

// floatsToVectorString serializes a float64 slice as a JSON array string.
// e.g. [0.1, 0.2, 0.3] → "[0.1,0.2,0.3]"
func floatsToVectorString(v []float64) string {
	b, _ := json.Marshal(v)
	return string(b)
}

// parseVectorJSON parses a JSON array string (or pgvector text) into float64 slice.
func parseVectorJSON(s string) ([]float64, error) {
	var out []float64
	if err := json.Unmarshal([]byte(s), &out); err != nil {
		return nil, fmt.Errorf("parse vector: %w", err)
	}
	return out, nil
}

// cosineDistance computes cosine distance between two vectors.
// Returns 0.0 for identical vectors, 1.0 for orthogonal, 2.0 for opposite.
func cosineDistance(a, b []float64) float64 {
	if len(a) != len(b) || len(a) == 0 {
		return 1.0
	}
	var dot, normA, normB float64
	for i := range a {
		dot += a[i] * b[i]
		normA += a[i] * a[i]
		normB += b[i] * b[i]
	}
	if normA == 0 || normB == 0 {
		return 1.0
	}
	return 1.0 - (dot / (math.Sqrt(normA) * math.Sqrt(normB)))
}

// dbQuerySimilar implements db.query_similar(table, column, vector=[], limit=10, filters=None, metric="cosine").
// Returns rows ordered by distance with an injected _distance float.
//
// Three dispatch paths:
//   - pgvector: native <=> operator with HNSW index
//   - fallback: fetch rows, compute cosine distance in Go, sort, return top N
func dbQuerySimilar(ctx context.Context, cfg DBModuleConfig) func(*starlark.Thread, *starlark.Builtin, starlark.Tuple, []starlark.Tuple) (starlark.Value, error) {
	return func(thread *starlark.Thread, b *starlark.Builtin, args starlark.Tuple, kwargs []starlark.Tuple) (starlark.Value, error) {
		var table, column string
		var vectorVal *starlark.List
		var limit starlark.Int = starlark.MakeInt(10)
		var filters starlark.Value = starlark.None
		var metric starlark.String = "cosine"

		if err := starlark.UnpackArgs(b.Name(), args, kwargs,
			"table", &table,
			"column", &column,
			"vector", &vectorVal,
			"limit?", &limit,
			"filters?", &filters,
			"metric?", &metric,
		); err != nil {
			return nil, err
		}

		if string(metric) != "cosine" {
			return nil, fmt.Errorf("db.query_similar: unsupported metric %q (only \"cosine\" is supported)", string(metric))
		}

		queryVec, err := starlarkListToFloats(vectorVal)
		if err != nil {
			return nil, fmt.Errorf("db.query_similar: %w", err)
		}

		lim, ok := limit.Int64()
		if !ok || lim < 1 {
			lim = 10
		}
		if lim > 100 {
			lim = 100
		}

		if strings.ContainsAny(column, " \t\n\"';-") {
			return nil, fmt.Errorf("db.query_similar: invalid column name %q", column)
		}

		physTable, err := cfg.physicalTable(table)
		if err != nil {
			return nil, err
		}

		if cfg.HasPgvector {
			return querySimilarPgvector(ctx, cfg, physTable, column, queryVec, int(lim), filters)
		}
		return querySimilarFallback(ctx, cfg, physTable, column, queryVec, int(lim), filters)
	}
}

// querySimilarPgvector uses the native pgvector <=> operator.
func querySimilarPgvector(ctx context.Context, cfg DBModuleConfig, physTable, column string, queryVec []float64, limit int, filters starlark.Value) (starlark.Value, error) {
	vecStr := floatsToVectorString(queryVec)

	whereClause, whereArgs, err := cfg.starlarkFiltersToSQL(filters, 1)
	if err != nil {
		return nil, err
	}

	nextIdx := len(whereArgs) + 1
	vecPH := cfg.ph(nextIdx)
	limitPH := cfg.ph(nextIdx + 1)

	query := fmt.Sprintf("SELECT *, (%s <=> %s::vector) AS _distance FROM %s",
		column, vecPH, physTable)
	if whereClause != "" {
		query += " " + whereClause
	}
	query += fmt.Sprintf(" ORDER BY %s <=> %s::vector LIMIT %s", column, vecPH, limitPH)

	allArgs := append(whereArgs, vecStr, limit)

	rows, err := cfg.DB.QueryContext(ctx, query, allArgs...)
	if err != nil {
		return nil, fmt.Errorf("db.query_similar: %w", err)
	}
	defer rows.Close()

	return rowsToStarlark(rows)
}

// querySimilarFallback fetches rows and computes cosine distance in Go.
// Used for SQLite (TEXT) and Postgres without pgvector (JSONB).
func querySimilarFallback(ctx context.Context, cfg DBModuleConfig, physTable, column string, queryVec []float64, limit int, filters starlark.Value) (starlark.Value, error) {
	whereClause, whereArgs, err := cfg.starlarkFiltersToSQL(filters, 1)
	if err != nil {
		return nil, err
	}

	// Fetch up to 1000 rows for distance computation.
	fetchLimit := limit * 10
	if fetchLimit > 1000 {
		fetchLimit = 1000
	}
	if fetchLimit < limit {
		fetchLimit = limit
	}

	limitPH := cfg.ph(len(whereArgs) + 1)
	query := fmt.Sprintf("SELECT * FROM %s", physTable)
	if whereClause != "" {
		query += " " + whereClause
	}
	query += " LIMIT " + limitPH
	allArgs := append(whereArgs, fetchLimit)

	rows, err := cfg.DB.QueryContext(ctx, query, allArgs...)
	if err != nil {
		return nil, fmt.Errorf("db.query_similar: %w", err)
	}
	defer rows.Close()

	cols, err := rows.Columns()
	if err != nil {
		return nil, err
	}

	// Find the vector column index.
	vecColIdx := -1
	for i, c := range cols {
		if c == column {
			vecColIdx = i
			break
		}
	}
	if vecColIdx == -1 {
		return nil, fmt.Errorf("db.query_similar: column %q not found in table", column)
	}

	type scoredRow struct {
		vals     []any
		distance float64
	}
	var scored []scoredRow

	for rows.Next() {
		vals := make([]any, len(cols))
		ptrs := make([]any, len(cols))
		for i := range vals {
			ptrs[i] = &vals[i]
		}
		if err := rows.Scan(ptrs...); err != nil {
			return nil, err
		}

		// Parse the vector column value.
		var vecStr string
		switch v := vals[vecColIdx].(type) {
		case string:
			vecStr = v
		case []byte:
			vecStr = string(v)
		default:
			continue // skip rows with unparseable vectors
		}

		rowVec, err := parseVectorJSON(vecStr)
		if err != nil {
			continue // skip malformed vectors
		}

		dist := cosineDistance(queryVec, rowVec)
		scored = append(scored, scoredRow{vals: vals, distance: dist})
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}

	// Sort by distance ascending.
	sort.Slice(scored, func(i, j int) bool {
		return scored[i].distance < scored[j].distance
	})

	// Take top N and build Starlark dicts.
	if len(scored) > limit {
		scored = scored[:limit]
	}

	var result []starlark.Value
	for _, sr := range scored {
		d := starlark.NewDict(len(cols) + 1)
		for i, col := range cols {
			sv, err := goToStarlark(sr.vals[i])
			if err != nil {
				return nil, fmt.Errorf("column %q: %w", col, err)
			}
			if err := d.SetKey(starlark.String(col), sv); err != nil {
				return nil, err
			}
		}
		// Inject _distance.
		if err := d.SetKey(starlark.String("_distance"), starlark.Float(sr.distance)); err != nil {
			return nil, err
		}
		result = append(result, d)
	}

	if result == nil {
		result = []starlark.Value{}
	}
	return starlark.NewList(result), nil
}