package collector

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

	"github.com/nexuscrm/sentry/internal/types"
)

// TiDBCollector queries CRM tables to produce data-volume snapshots.
//
// Default path per table: COUNT(*) + MAX(Id) + MAX(LastModifiedDate) + MAX(CreatedDate),
// run in parallel across a worker pool. For any table whose query exceeds
// queryTimeout (typically 5s), degrade to information_schema.TABLES.TABLE_ROWS
// (a TiDB statistics-based estimate). Callers may also pre-mark tables as slow
// via the slowTables argument.
type TiDBCollector struct {
	db           *sql.DB
	workerPool   int
	queryTimeout time.Duration
}

// NewTiDBCollector constructs a collector.
// workerPool is the number of concurrent table queries (typical: 8).
// queryTimeout bounds each per-table query (typical: 5s).
func NewTiDBCollector(db *sql.DB, workerPool int, queryTimeout time.Duration) *TiDBCollector {
	if workerPool <= 0 {
		workerPool = 1
	}
	return &TiDBCollector{db: db, workerPool: workerPool, queryTimeout: queryTimeout}
}

// DiscoverTables returns all base table names in the current database,
// excluding sentry's own artifacts (any name starting with "sentry_").
// New tables added to the schema are picked up automatically on the
// next cycle — no sentry redeploy required.
//
// The "sentry_" filter is applied in Go rather than via SQL LIKE … ESCAPE
// because TiDB's parser rejects the `ESCAPE '\'` form (backslash is a
// string-literal escape in default sql_mode).
func (c *TiDBCollector) DiscoverTables(ctx context.Context) ([]string, error) {
	const q = `
SELECT TABLE_NAME
FROM information_schema.TABLES
WHERE TABLE_SCHEMA = DATABASE()
  AND TABLE_TYPE = 'BASE TABLE'
ORDER BY TABLE_NAME`
	rows, err := c.db.QueryContext(ctx, q)
	if err != nil {
		return nil, fmt.Errorf("discover tables: %w", err)
	}
	defer rows.Close()
	var out []string
	for rows.Next() {
		var t string
		if err := rows.Scan(&t); err != nil {
			return nil, fmt.Errorf("discover tables: scan: %w", err)
		}
		if strings.HasPrefix(t, "sentry_") {
			continue
		}
		out = append(out, t)
	}
	return out, rows.Err()
}

// Snapshot runs per-table queries in parallel, degrading slow tables to
// information_schema row estimates. `slowTables` is a set of table names
// the caller wants pre-classified as slow (loaded from S3 across restarts).
// A `nil` or empty `slowTables` set causes all tables to attempt the exact path.
func (c *TiDBCollector) Snapshot(
	ctx context.Context,
	tables []string,
	slowTables map[string]bool,
) ([]types.TableRow, error) {
	// Pre-fetch row estimates for every table with a single bulk query.
	// We always run this (not just when slowTables is pre-populated)
	// because any per-table query may time out mid-snapshot and need a
	// fallback number — without this, a first-run timeout degrades to 0.
	estimates := map[string]int64{}
	{
		rows, err := c.db.QueryContext(ctx,
			`SELECT TABLE_NAME, TABLE_ROWS FROM information_schema.TABLES
			 WHERE TABLE_SCHEMA = DATABASE()`)
		if err == nil {
			for rows.Next() {
				var name string
				var n sql.NullInt64
				if err := rows.Scan(&name, &n); err != nil {
					continue
				}
				if n.Valid {
					estimates[name] = n.Int64
				}
			}
			rows.Close()
		}
		// Silently tolerate discovery errors — the per-table path still works.
	}

	var (
		wg  sync.WaitGroup
		mu  sync.Mutex
		out = make([]types.TableRow, 0, len(tables))
		sem = make(chan struct{}, c.workerPool)
	)
	for _, table := range tables {
		table := table
		sem <- struct{}{}
		wg.Add(1)
		go func() {
			defer wg.Done()
			defer func() { <-sem }()
			row := c.snapshotOne(ctx, table, slowTables, estimates)
			mu.Lock()
			out = append(out, row)
			mu.Unlock()
		}()
	}
	wg.Wait()
	return out, nil
}

func (c *TiDBCollector) snapshotOne(
	ctx context.Context,
	table string,
	slowTables map[string]bool,
	estimates map[string]int64,
) types.TableRow {
	row := types.TableRow{Name: table, Source: types.SourceCountStar}

	if slowTables[table] {
		row.RowCount = estimates[table]
		row.Source = types.SourceInfoSchemaEst
		return row
	}

	qCtx, cancel := context.WithTimeout(ctx, c.queryTimeout)
	defer cancel()

	// Table name came from information_schema, NOT user input.
	// Defense-in-depth: double any literal backticks so a pathological
	// identifier like `foo`;DROP` can't break out of the quoted identifier.
	// MySQL/TiDB identifier quoting rule: `` → doubled backtick inside ``.
	safe := "`" + strings.ReplaceAll(table, "`", "``") + "`"
	q := fmt.Sprintf(
		"SELECT COUNT(*), MAX(`Id`), MAX(`LastModifiedDate`), MAX(`CreatedDate`) FROM %s",
		safe,
	)
	var (
		count      int64
		maxID      sql.NullString
		maxUpdated sql.NullTime
		maxCreated sql.NullTime
	)
	err := c.db.QueryRowContext(qCtx, q).Scan(&count, &maxID, &maxUpdated, &maxCreated)
	if err != nil {
		// Degrade: any error — missing Id column, missing CreatedDate, or timeout —
		// falls back to the information_schema estimate. The caller is expected to
		// add this table to the persisted slow_tables set so the next cycle skips
		// straight to the estimate.
		row.RowCount = estimates[table]
		row.Source = types.SourceInfoSchemaEst
		return row
	}
	row.RowCount = count
	if maxID.Valid {
		v := maxID.String
		row.MaxID = &v
	}
	if maxUpdated.Valid {
		v := maxUpdated.Time
		row.MaxUpdatedAt = &v
	}
	if maxCreated.Valid {
		v := maxCreated.Time
		row.MaxCreatedAt = &v
	}
	return row
}