package persistence

import (
	"context"
	"encoding/json"
	"fmt"
	"strconv"
	"strings"
	"time"

	apperrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/query"
	"github.com/nexuscrm/shared/pkg/models"
)

type bulkInsertOptions struct {
	batchSize        int
	ignoreDuplicates bool
}

type bulkInsertChunk struct {
	offset  int
	records []models.SObject
	columns []string
	raw     bool
}

func parseBulkInsertOptions(options interface{}) bulkInsertOptions {
	opts := bulkInsertOptions{batchSize: 100}
	if v, ok := options.(int); ok {
		opts.batchSize = v
		return opts
	}
	if raw, ok := options.(map[string]interface{}); ok {
		if v, ok := raw["batchSize"].(int); ok {
			opts.batchSize = v
		}
		if v, ok := raw["ignoreDuplicates"].(bool); ok {
			opts.ignoreDuplicates = v
		}
	}
	return opts
}

func buildBulkInsertChunks(records []models.SObject, allColumns []string, opts bulkInsertOptions) []bulkInsertChunk {
	if len(records) > 100 {
		return buildRawBulkInsertChunks(records, allColumns)
	}
	return buildParameterizedBulkInsertChunks(records, allColumns, opts)
}

func buildParameterizedBulkInsertChunks(records []models.SObject, allColumns []string, opts bulkInsertOptions) []bulkInsertChunk {
	chunks := make([]bulkInsertChunk, 0)
	totalRecords := len(records)
	cursor := 0

	for cursor < totalRecords {
		// Chunk-by-identical-active-column-set (PR #2057 round-6 F3).
		// Pre-fix this used a 1000-row lookAhead UNION, which meant a
		// chunk's column list could include columns that were populated
		// in some rows but absent in others — the SQL builder then bound
		// nil for the absent rows, silently writing NULL into columns
		// the row's caller intended to omit. Frozen-formula fetch errors
		// (round-6 F3 deletes them from the row's SObject) MUST result
		// in those rows landing in their own chunk WITHOUT the column,
		// matching single-insert "absent → no SQL write" semantics.
		// Lockstep walk: the first row's active columns define the chunk;
		// rows are admitted while their active-set matches; the first
		// mismatching row closes the chunk and starts the next one.
		firstActive := activeColumnsForRecords(records[cursor:cursor+1], allColumns)
		firstActiveSet := make(map[string]struct{}, len(firstActive))
		for _, c := range firstActive {
			firstActiveSet[c] = struct{}{}
		}

		numParams := len(firstActive)
		if numParams == 0 {
			numParams = 1
		}
		safeSize := 65000 / numParams
		if safeSize > 2000 {
			safeSize = 2000
		}
		if opts.batchSize > 0 && safeSize > opts.batchSize {
			safeSize = opts.batchSize
		}

		// Walk forward: extend chunk while the next row's active set is
		// identical AND we have not exceeded safeSize. Identical column-set
		// means SQL builder's `record[col]` binds a non-nil value for every
		// row in the chunk (no silent NULL fill).
		chunkEnd := cursor + 1
		for chunkEnd < totalRecords && (chunkEnd-cursor) < safeSize {
			rowActive := activeColumnsForRecords(records[chunkEnd:chunkEnd+1], allColumns)
			if !sameActiveColumnSet(rowActive, firstActiveSet) {
				break
			}
			chunkEnd++
		}

		chunks = append(chunks, bulkInsertChunk{
			offset:  cursor,
			records: records[cursor:chunkEnd],
			columns: firstActive,
		})
		cursor = chunkEnd
	}

	return chunks
}

// sameActiveColumnSet returns true when rowActive (a slice) is element-wise
// equal to expected (a set built from a reference row's columns). Used by
// the chunker to enforce uniform column sets within a chunk so the SQL
// builder cannot silently bind NULL for an absent column on one row when
// a sibling row populated it (PR #2057 round-6 F3).
func sameActiveColumnSet(rowActive []string, expected map[string]struct{}) bool {
	if len(rowActive) != len(expected) {
		return false
	}
	for _, c := range rowActive {
		if _, ok := expected[c]; !ok {
			return false
		}
	}
	return true
}

func buildRawBulkInsertChunks(records []models.SObject, allColumns []string) []bulkInsertChunk {
	chunks := make([]bulkInsertChunk, 0)
	totalRecords := len(records)
	cursor := 0

	const rawChunkLimit = 10000

	for cursor < totalRecords {
		// Chunk-by-identical-active-column-set, same lockstep-walk shape as
		// buildParameterizedBulkInsertChunks (PR #2057 round-7 F4 — Codex
		// Review HIGH). Pre-fix the raw path took a single 10k-row chunk and
		// UNIONed active columns across the whole batch; rows that did not
		// populate a column got NULL written via buildRawBulkInsertSQL,
		// silently overriding column defaults — and re-introduced the same
		// frozen-formula fetch-error corruption F3 closed for the
		// parameterized path. Round-7 ports the lockstep-walk so raw and
		// parameterized paths agree byte-for-byte on chunk boundaries.
		firstActive := activeColumnsForRecords(records[cursor:cursor+1], allColumns)
		firstActiveSet := make(map[string]struct{}, len(firstActive))
		for _, c := range firstActive {
			firstActiveSet[c] = struct{}{}
		}

		chunkEnd := cursor + 1
		for chunkEnd < totalRecords && (chunkEnd-cursor) < rawChunkLimit {
			rowActive := activeColumnsForRecords(records[chunkEnd:chunkEnd+1], allColumns)
			if !sameActiveColumnSet(rowActive, firstActiveSet) {
				break
			}
			chunkEnd++
		}

		chunks = append(chunks, bulkInsertChunk{
			offset:  cursor,
			records: records[cursor:chunkEnd],
			columns: firstActive,
			raw:     true,
		})
		cursor = chunkEnd
	}

	return chunks
}

// BulkInsert executes a multi-row INSERT statement for batch insertions.
// Foreign-key enforcement remains enabled on every generic write path. Callers
// must submit parent rows before dependent rows; dependency-order recovery
// belongs to an owning lifecycle/import coordinator, not a session-wide
// FOREIGN_KEY_CHECKS bypass inside the generic record repository.
func (r *RecordRepository) BulkInsert(ctx context.Context, tableName string, records []models.SObject, allColumns []string, options interface{}) error {
	opts := parseBulkInsertOptions(options)

	if len(records) == 0 {
		return nil
	}

	if !isValidFieldName(tableName) {
		return apperrors.NewValidationError("tableName", fmt.Sprintf("invalid table name: %s", tableName))
	}

	// Schema Resolution — reuse shared helper
	schema, _ := r.getSchema(ctx, tableName)
	if err := RejectAuthorityOnlyFieldWrites(tableName, schema, records...); err != nil {
		return err
	}
	exec := r.GetExecutor(ctx)
	if _, err := managedRowMutationProtectionFor(ctx, exec, tableName, schema); err != nil {
		return apperrors.NewInternalError("inspect managed row mutation storage", err)
	}
	resolver := func(field string) string {
		return resolvePhysicalName(schema, field)
	}

	// HEURISTIC: Use Raw SQL for batches > 100 to avoid driver overhead on params
	if len(records) > 100 {
		return r.bulkInsertRaw(ctx, tableName, records, allColumns, opts.ignoreDuplicates, resolver)
	}

	for _, chunk := range buildParameterizedBulkInsertChunks(records, allColumns, opts) {
		sqlStr, params := query.BulkInsertOrdered(tableName, chunk.columns, sObjectsToMaps(chunk.records), resolver)
		if opts.ignoreDuplicates {
			sqlStr = toInsertIgnore(sqlStr)
		}
		if sqlStr != "" {
			if _, err := exec.ExecContext(ctx, sqlStr, params...); err != nil {
				return apperrors.NewInternalError(fmt.Sprintf("bulk insert adaptable batch failed at offset %d", chunk.offset), err)
			}
		}
	}
	return nil
}

// BulkInsertWithLanding executes an insert and reports the landing status of
// each input record. The normal non-IGNORE path remains batched: if the batch
// succeeds, every input row landed. Under INSERT IGNORE, exact per-row landing
// status is only available from each statement's RowsAffected, so this method
// uses one statement per input record instead of guessing from post-insert
// existence. Like BulkInsert, every landing path retains foreign-key
// enforcement and requires dependency-ordered input.
func (r *RecordRepository) BulkInsertWithLanding(ctx context.Context, tableName string, records []models.SObject, allColumns []string, options interface{}) ([]bool, error) {
	landed := make([]bool, len(records))
	if len(records) == 0 {
		return landed, nil
	}

	opts := parseBulkInsertOptions(options)
	if !opts.ignoreDuplicates {
		if err := r.BulkInsert(ctx, tableName, records, allColumns, options); err != nil {
			return nil, err
		}
		for i := range landed {
			landed[i] = true
		}
		return landed, nil
	}

	if !isValidFieldName(tableName) {
		return nil, apperrors.NewValidationError("tableName", fmt.Sprintf("invalid table name: %s", tableName))
	}

	schema, _ := r.getSchema(ctx, tableName)
	if err := RejectAuthorityOnlyFieldWrites(tableName, schema, records...); err != nil {
		return nil, err
	}
	exec := r.GetExecutor(ctx)
	if _, err := managedRowMutationProtectionFor(ctx, exec, tableName, schema); err != nil {
		return nil, apperrors.NewInternalError("inspect managed row mutation storage", err)
	}
	resolver := func(field string) string {
		return resolvePhysicalName(schema, field)
	}

	for _, chunk := range buildBulkInsertChunks(records, allColumns, opts) {
		for i, record := range chunk.records {
			absoluteIndex := chunk.offset + i
			var (
				affected int64
				err      error
			)
			if chunk.raw {
				affected, err = r.execRawInsertOne(ctx, exec, tableName, record, chunk.columns, true, resolver)
			} else {
				affected, err = r.execParameterizedInsertOne(ctx, exec, tableName, record, chunk.columns, resolver)
			}
			if err != nil {
				return nil, apperrors.NewInternalError(fmt.Sprintf("bulk insert landing batch failed at offset %d", absoluteIndex), err)
			}
			landed[absoluteIndex] = affected > 0
		}
	}
	return landed, nil
}

func (r *RecordRepository) execParameterizedInsertOne(ctx context.Context, exec Executor, tableName string, record models.SObject, activeColumns []string, resolver func(string) string) (int64, error) {
	sqlStr, params := query.BulkInsertOrdered(tableName, activeColumns, []map[string]interface{}{record}, resolver)
	sqlStr = toInsertIgnore(sqlStr)
	if sqlStr == "" {
		return 0, nil
	}
	res, err := exec.ExecContext(ctx, sqlStr, params...)
	if err != nil {
		return 0, err
	}
	return res.RowsAffected()
}

func (r *RecordRepository) execRawInsertOne(ctx context.Context, exec Executor, tableName string, record models.SObject, activeColumns []string, ignoreDuplicates bool, resolver func(string) string) (int64, error) {
	sqlStr := buildRawBulkInsertSQL(tableName, []models.SObject{record}, activeColumns, ignoreDuplicates, resolver)
	if sqlStr == "" {
		return 0, nil
	}
	res, err := exec.ExecContext(ctx, sqlStr)
	if err != nil {
		return 0, err
	}
	return res.RowsAffected()
}

func sObjectsToMaps(records []models.SObject) []map[string]interface{} {
	out := make([]map[string]interface{}, len(records))
	for i, rec := range records {
		out[i] = rec
	}
	return out
}

func activeColumnsForRecords(records []models.SObject, allColumns []string) []string {
	activeColsMap := make(map[string]bool)
	for _, rec := range records {
		for k, v := range rec {
			if v != nil {
				activeColsMap[k] = true
			}
		}
	}
	return activeColumnsFromMap(activeColsMap, allColumns)
}

func activeColumnsFromMap(activeColsMap map[string]bool, allColumns []string) []string {
	var activeColumns []string
	for _, col := range allColumns {
		if activeColsMap[col] {
			activeColumns = append(activeColumns, col)
		}
	}
	if len(activeColumns) == 0 {
		activeColumns = []string{"id"}
	}
	return activeColumns
}

func toInsertIgnore(sqlStr string) string {
	if strings.HasPrefix(sqlStr, "INSERT INTO") {
		return strings.Replace(sqlStr, "INSERT INTO", "INSERT IGNORE INTO", 1)
	}
	if strings.HasPrefix(sqlStr, "INSERT ") {
		return strings.Replace(sqlStr, "INSERT ", "INSERT IGNORE ", 1)
	}
	return sqlStr
}

// bulkInsertRaw constructs a raw SQL statement for high-volume inserts (>100 records)
func (r *RecordRepository) bulkInsertRaw(ctx context.Context, tableName string, records []models.SObject, allColumns []string, ignoreDuplicates bool, resolver func(string) string) error {
	exec := r.GetExecutor(ctx)

	for _, chunk := range buildRawBulkInsertChunks(records, allColumns) {
		queryStr := buildRawBulkInsertSQL(tableName, chunk.records, chunk.columns, ignoreDuplicates, resolver)
		if _, err := exec.ExecContext(ctx, queryStr); err != nil {
			return apperrors.NewInternalError(fmt.Sprintf("raw bulk insert failed at offset %d", chunk.offset), err)
		}
	}
	return nil
}

func buildRawBulkInsertSQL(tableName string, records []models.SObject, activeColumns []string, ignoreDuplicates bool, resolver func(string) string) string {
	if len(records) == 0 || len(activeColumns) == 0 {
		return ""
	}

	var sb strings.Builder
	sb.Grow(10 * 1024 * 1024)
	if ignoreDuplicates {
		sb.WriteString("INSERT IGNORE INTO `")
	} else {
		sb.WriteString("INSERT INTO `")
	}
	sb.WriteString(tableName)
	sb.WriteString("` (")
	for i, col := range activeColumns {
		if i > 0 {
			sb.WriteString(",")
		}
		sb.WriteString("`")
		physicalName := col
		if resolver != nil {
			physicalName = resolver(col)
		}
		sb.WriteString(physicalName)
		sb.WriteString("`")
	}
	sb.WriteString(") VALUES ")

	for i, rec := range records {
		if i > 0 {
			sb.WriteString(",")
		}
		sb.WriteString("(")
		for j, col := range activeColumns {
			if j > 0 {
				sb.WriteString(",")
			}
			val := rec[col]
			if val == nil {
				sb.WriteString("NULL")
			} else {
				writeEscapedValue(&sb, val)
			}
		}
		sb.WriteString(")")
	}

	return sb.String()
}

// writeEscapedValue writes the escaped string representation of a value for raw SQL
func writeEscapedValue(sb *strings.Builder, v interface{}) {
	switch val := v.(type) {
	case string:
		sb.WriteString("'")
		repl := strings.NewReplacer("'", "''", "\\", "\\\\", "\x00", "\\0")
		repl.WriteString(sb, val)
		sb.WriteString("'")
	case int:
		sb.WriteString(strconv.Itoa(val))
	case int64:
		sb.WriteString(strconv.FormatInt(val, 10))
	case float64:
		sb.WriteString(strconv.FormatFloat(val, 'f', -1, 64))
	case bool:
		if val {
			sb.WriteString("1")
		} else {
			sb.WriteString("0")
		}
	case time.Time:
		sb.WriteString("'")
		sb.WriteString(val.Format("2006-01-02 15:04:05"))
		sb.WriteString("'")
	case map[string]interface{}:
		b, err := json.Marshal(val)
		if err != nil {
			sb.WriteString("'{}'")
		} else {
			writeEscapedValue(sb, string(b))
		}
	default:
		sb.WriteString("'")
		str := fmt.Sprintf("%v", val)
		repl := strings.NewReplacer("'", "''", "\\", "\\\\")
		repl.WriteString(sb, str)
		sb.WriteString("'")
	}
}