// Package autonumber — live-max query for per_prefix lazy-init
// (Codex round-3 CRITICAL follow-up, "Option B" runtime race-closure).
//
// When the reserver lazy-creates a `_System_AutoNumberCounter` row
// for a per_prefix bucket, it calls queryLiveMaxForPrefix to absorb
// any existing records in that bucket. The initial NextValue is then
// max(startValue, live_max+1), so a first post-cutover insert into a
// bucket that already contains records (from legacy Apex triggers or
// a concurrent writer slipping in mid-seed) continues the sequence
// rather than colliding at 1.
//
// This file deliberately duplicates a small amount of SQL-identifier-
// whitelist logic already in `bootstrap/autonumber_seed.go` rather
// than import-cycling `bootstrap ↔ autonumber`. The whitelist regex
// is identical to validSQLIdentifier there.
package autonumber

import (
	"context"
	"database/sql"
	"fmt"
	"regexp"
	"strconv"
	"strings"
)

// engineIdentifierRE mirrors bootstrap.validSQLIdentifier. Kept here
// so the engine does not import bootstrap.
var engineIdentifierRE = regexp.MustCompile(`^[A-Za-z][A-Za-z0-9_]{0,62}$`)

// queryLiveMaxForPrefix returns the maximum numeric counter observed
// on rows of `<objectName>.<fieldName>` whose value matches the
// (prefixBefore, prefixAfter) pair — i.e. rows where
// `value == prefixBefore + <digits> + prefixAfter`. Rows whose
// middle segment is not a valid integer are ignored (defensive:
// legacy random-8-digit data coexisting with the new format should
// not block the engine).
//
// Split-prefix semantics (Codex round-6 HIGH fix): for formats where
// the counter is NOT the final token (e.g. `{00}-END`,
// `{Country}-{00}-X`), the rendered pre-counter text and post-counter
// text must be matched separately — otherwise values like
// `01-END` would be missed by a single `LIKE 'before%'` anchor. For
// the common counter-last case prefixAfter is "" and the behavior
// reduces to `LIKE 'before%'`.
//
// Returns 0 when:
//   - objectName / fieldName fail identifier whitelist (defensive
//     guard; engine wires from Rule metadata which is already
//     admin-controlled, but the whitelist keeps us one layer of
//     defense against a corrupt Rule).
//   - prefixBefore AND prefixAfter are both empty (degenerate
//     counter-only format — no partitioning dimension).
//   - The target column has no rows matching.
//
// SQL shape:
//
//	SELECT `<field>` FROM `<object>`
//	 WHERE `<field>` LIKE CONCAT(?, '%', ?)
//
// Counter extraction happens in Go (HasPrefix / HasSuffix / slice /
// ParseInt) so the SQL stays portable and does not rely on REGEXP /
// SUBSTRING semantics that differ across MySQL/TiDB versions.
//
// Executes against the supplied *sql.DB (engine's reservationDB) via
// a single round-trip — no transaction because the query is a plain
// read and does not need to serialize against concurrent writers
// (the caller's lazy-init UPSERT is itself idempotent via
// ON DUPLICATE KEY UPDATE).
func queryLiveMaxForPrefix(
	ctx context.Context,
	db *sql.DB,
	objectName, fieldName, prefixBefore, prefixAfter string,
) (int64, error) {
	if db == nil {
		return 0, fmt.Errorf("queryLiveMaxForPrefix: db is nil")
	}
	if prefixBefore == "" && prefixAfter == "" {
		// No partitioning dimension — counter-only format. Caller
		// should not invoke us in this case, but return 0 so
		// fallback initialNext=startValue.
		return 0, nil
	}
	if !engineIdentifierRE.MatchString(objectName) {
		return 0, fmt.Errorf("queryLiveMaxForPrefix: invalid objectName %q", objectName)
	}
	if !engineIdentifierRE.MatchString(fieldName) {
		return 0, fmt.Errorf("queryLiveMaxForPrefix: invalid fieldName %q", fieldName)
	}

	// Probe physical column existence — fresh DB may not have the
	// object materialized yet (analogous to
	// scanLivePerPrefixCounters' physicalColumnExists check).
	var nCols int
	if err := db.QueryRowContext(ctx,
		"SELECT COUNT(*) FROM INFORMATION_SCHEMA.COLUMNS "+
			"WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND COLUMN_NAME = ?",
		objectName, fieldName,
	).Scan(&nCols); err != nil {
		return 0, fmt.Errorf("queryLiveMaxForPrefix: probe column: %w", err)
	}
	if nCols == 0 {
		return 0, nil
	}

	query := fmt.Sprintf(
		"SELECT `%s` FROM `%s` WHERE `%s` LIKE CONCAT(?, '%%', ?)",
		fieldName, objectName, fieldName,
	)
	rows, err := db.QueryContext(ctx, query, prefixBefore, prefixAfter)
	if err != nil {
		return 0, fmt.Errorf("queryLiveMaxForPrefix: query: %w", err)
	}
	defer rows.Close()

	var maxCounter int64
	for rows.Next() {
		var v sql.NullString
		if err := rows.Scan(&v); err != nil {
			return 0, fmt.Errorf("queryLiveMaxForPrefix: scan: %w", err)
		}
		if !v.Valid || v.String == "" {
			continue
		}
		// Defense: verify the Go-side boundary, since LIKE can be
		// case-insensitive on some TiDB collations.
		if !strings.HasPrefix(v.String, prefixBefore) || !strings.HasSuffix(v.String, prefixAfter) {
			continue
		}
		// Extract the counter-substring between before and after.
		middle := v.String[len(prefixBefore) : len(v.String)-len(prefixAfter)]
		if middle == "" {
			continue
		}
		// Defensive: skip non-all-digit middles (legacy data may
		// contain historical non-numeric values).
		allDigits := true
		for i := 0; i < len(middle); i++ {
			if middle[i] < '0' || middle[i] > '9' {
				allDigits = false
				break
			}
		}
		if !allDigits {
			continue
		}
		n, err := strconv.ParseInt(middle, 10, 64)
		if err != nil {
			continue
		}
		if n > maxCounter {
			maxCounter = n
		}
	}
	if err := rows.Err(); err != nil {
		return 0, fmt.Errorf("queryLiveMaxForPrefix: iterate: %w", err)
	}
	return maxCounter, nil
}