package outboxarchiverestore

import (
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"errors"
	"fmt"
	"reflect"
	"strings"
	"time"

	_ "github.com/go-sql-driver/mysql"
)

const (
	RehearsalTable = "wallet_outbox_restore_rehearsals_v1"
	RowsTable      = "wallet_outbox_restore_rows_v1"
	restoreState   = "RESTORING"
)

// RestoreResult is safe to print: it identifies the selected schema and fixed
// table contract but never includes the DSN or credentials used to reach it.
type RestoreResult struct {
	State              string `json:"state"`
	RehearsalID        string `json:"rehearsal_id"`
	BatchID            string `json:"batch_id"`
	AppCode            string `json:"app_code"`
	RowCount           int    `json:"row_count"`
	TargetDatabase     string `json:"target_database"`
	RehearsalTable     string `json:"rehearsal_table"`
	RowsTable          string `json:"rows_table"`
	DataObjectKey      string `json:"data_object_key"`
	FirstCursor        Cursor `json:"first_cursor"`
	LastCursor         Cursor `json:"last_cursor"`
	NDJSONSHA256       string `json:"ndjson_sha256"`
	GzipSHA256         string `json:"gzip_sha256"`
	DataCRC64          string `json:"data_crc64"`
	ManifestSHA256     string `json:"manifest_sha256"`
	ManifestCRC64      string `json:"manifest_crc64"`
	SuccessSHA256      string `json:"success_sha256"`
	SuccessCRC64       string `json:"success_crc64"`
	AvailableTotal     string `json:"available_delta_total"`
	FrozenTotal        string `json:"frozen_delta_total"`
	VerifiedAtMS       int64  `json:"verified_at_ms"`
	IdempotentReadback bool   `json:"idempotent_readback"`
}

// Restore persists a fully verified batch into versioned isolation tables. All
// DDL targets new/fixed table names; all DML is scoped by rehearsal PK. It never
// reads, locks, updates, or deletes wallet_outbox.
func Restore(ctx context.Context, dsn, expectedDatabase, rehearsalID string, verification Verification) (RestoreResult, error) {
	dsn = strings.TrimSpace(dsn)
	expectedDatabase = strings.TrimSpace(expectedDatabase)
	rehearsalID = strings.TrimSpace(rehearsalID)
	if dsn == "" || expectedDatabase == "" {
		return RestoreResult{}, errors.New("restore DSN and exact expected database are required")
	}
	if !validRehearsalID(rehearsalID) {
		return RestoreResult{}, errors.New("rehearsal_id must be 1..64 ASCII letters, digits, dot, underscore, or dash")
	}
	if verification.State != filesVerifiedState || len(verification.Records) != verification.Manifest.RowCount {
		return RestoreResult{}, errors.New("local archive verification is incomplete")
	}

	db, err := sql.Open("mysql", dsn)
	if err != nil {
		return RestoreResult{}, err
	}
	defer db.Close()
	// A rehearsal is intentionally single-connection and serial; this prevents a
	// restore tool from consuming the production-sized connection pool by mistake.
	db.SetMaxOpenConns(1)
	db.SetMaxIdleConns(1)
	db.SetConnMaxLifetime(10 * time.Minute)
	conn, err := db.Conn(ctx)
	if err != nil {
		return RestoreResult{}, err
	}
	defer conn.Close()

	var databaseName string
	if err := conn.QueryRowContext(ctx, `SELECT DATABASE()`).Scan(&databaseName); err != nil {
		return RestoreResult{}, err
	}
	if databaseName != expectedDatabase {
		return RestoreResult{}, fmt.Errorf("restore database mismatch: connected=%q expected=%q", databaseName, expectedDatabase)
	}
	if err := ensureIsolationSchema(ctx, conn); err != nil {
		return RestoreResult{}, err
	}
	release, acquired, err := acquireRestoreLock(ctx, conn, rehearsalID)
	if err != nil {
		return RestoreResult{}, err
	}
	if !acquired {
		return RestoreResult{}, errors.New("another process holds the same restore rehearsal lock")
	}
	defer release()

	if existing, found, err := readExistingRestore(ctx, conn, rehearsalID, verification); err != nil {
		return RestoreResult{}, err
	} else if found {
		return existing, nil
	}

	tx, err := conn.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
	if err != nil {
		return RestoreResult{}, err
	}
	defer tx.Rollback()
	nowMS := time.Now().UTC().UnixMilli()
	if err := insertRehearsal(ctx, tx, rehearsalID, databaseName, verification, nowMS); err != nil {
		return RestoreResult{}, err
	}
	for offset := 0; offset < len(verification.Records); offset += 100 {
		end := min(offset+100, len(verification.Records))
		if err := insertRestoreRows(ctx, tx, rehearsalID, verification.Manifest.BatchID, nowMS, verification.Records[offset:end]); err != nil {
			return RestoreResult{}, err
		}
	}
	// These aggregates only scan one bounded rehearsal partition in the isolation
	// table. They never aggregate the live outbox and are the database-level proof
	// that JSON conversion and INSERT semantics preserved the archive controls.
	if err := verifyRestoredControls(ctx, tx, rehearsalID, verification); err != nil {
		return RestoreResult{}, err
	}
	if err := verifyRestoredRowsExact(ctx, tx, rehearsalID, verification.Records); err != nil {
		return RestoreResult{}, err
	}
	result := newRestoreResult(rehearsalID, databaseName, verification, nowMS, false)
	update, err := tx.ExecContext(ctx, `
		UPDATE wallet_outbox_restore_rehearsals_v1
		SET state = ?, verified_at_ms = ?, updated_at_ms = ?
		WHERE rehearsal_id = ? AND state = ?`, restoreVerifiedState, nowMS, nowMS, rehearsalID, restoreState)
	if err != nil {
		return RestoreResult{}, err
	}
	if affected, err := update.RowsAffected(); err != nil || affected != 1 {
		return RestoreResult{}, errors.New("restore rehearsal state transition did not affect exactly one row")
	}
	if err := tx.Commit(); err != nil {
		return RestoreResult{}, err
	}
	return result, nil
}

func ensureIsolationSchema(ctx context.Context, conn *sql.Conn) error {
	// CREATE IF NOT EXISTS takes metadata locks only on the dedicated table names;
	// unlike ALTER wallet_outbox it cannot rebuild or block the hot owner table.
	statements := []string{`
		CREATE TABLE IF NOT EXISTS wallet_outbox_restore_rehearsals_v1 (
		  rehearsal_id VARCHAR(64) NOT NULL,
		  source_app_code VARCHAR(32) NOT NULL,
		  source_batch_id CHAR(64) NOT NULL,
		  state VARCHAR(32) NOT NULL,
		  schema_version INT NOT NULL,
		  data_object_key VARCHAR(512) NOT NULL,
		  row_count INT NOT NULL,
		  first_updated_at_ms BIGINT NOT NULL,
		  first_event_id VARCHAR(128) NOT NULL,
		  last_updated_at_ms BIGINT NOT NULL,
		  last_event_id VARCHAR(128) NOT NULL,
		  available_delta_total VARCHAR(80) NOT NULL,
		  frozen_delta_total VARCHAR(80) NOT NULL,
		  ndjson_sha256 CHAR(64) NOT NULL,
		  gzip_sha256 CHAR(64) NOT NULL,
		  data_crc64 VARCHAR(32) NOT NULL,
		  manifest_sha256 CHAR(64) NOT NULL,
		  manifest_crc64 VARCHAR(32) NOT NULL,
		  success_sha256 CHAR(64) NOT NULL,
		  success_crc64 VARCHAR(32) NOT NULL,
		  target_database VARCHAR(128) NOT NULL,
		  verified_at_ms BIGINT NOT NULL,
		  created_at_ms BIGINT NOT NULL,
		  updated_at_ms BIGINT NOT NULL,
		  PRIMARY KEY (rehearsal_id),
		  KEY idx_wallet_outbox_restore_batch (source_app_code, source_batch_id, state)
		) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='wallet outbox archive restore rehearsal evidence v1'`, `
		CREATE TABLE IF NOT EXISTS wallet_outbox_restore_rows_v1 (
		  rehearsal_id VARCHAR(64) NOT NULL,
		  source_batch_id CHAR(64) NOT NULL,
		  app_code VARCHAR(32) NOT NULL,
		  event_id VARCHAR(128) NOT NULL,
		  event_type VARCHAR(64) NOT NULL,
		  transaction_id VARCHAR(96) NOT NULL,
		  command_id VARCHAR(128) NOT NULL,
		  user_id BIGINT NOT NULL,
		  asset_type VARCHAR(32) NOT NULL,
		  available_delta BIGINT NOT NULL,
		  frozen_delta BIGINT NOT NULL,
		  payload JSON NOT NULL,
		  status VARCHAR(32) NOT NULL,
		  worker_id VARCHAR(128) NOT NULL,
		  lock_until_ms BIGINT NULL,
		  retry_count INT NOT NULL,
		  next_retry_at_ms BIGINT NULL,
		  last_error TEXT NULL,
		  created_at_ms BIGINT NOT NULL,
		  updated_at_ms BIGINT NOT NULL,
		  restored_at_ms BIGINT NOT NULL,
		  PRIMARY KEY (rehearsal_id, app_code, event_id),
		  KEY idx_wallet_outbox_restore_cursor (rehearsal_id, app_code, updated_at_ms, event_id),
		  KEY idx_wallet_outbox_restore_batch (source_batch_id, app_code, event_id)
		) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='isolated wallet_outbox-compatible restored rows v1'`}
	for _, statement := range statements {
		if _, err := conn.ExecContext(ctx, statement); err != nil {
			return err
		}
	}
	return nil
}

func insertRehearsal(ctx context.Context, tx *sql.Tx, rehearsalID, databaseName string, verification Verification, nowMS int64) error {
	manifest := verification.Manifest
	_, err := tx.ExecContext(ctx, `
		INSERT INTO wallet_outbox_restore_rehearsals_v1 (
		  rehearsal_id, source_app_code, source_batch_id, state, schema_version, data_object_key, row_count,
		  first_updated_at_ms, first_event_id, last_updated_at_ms, last_event_id,
		  available_delta_total, frozen_delta_total, ndjson_sha256, gzip_sha256, data_crc64,
		  manifest_sha256, manifest_crc64, success_sha256, success_crc64, target_database,
		  verified_at_ms, created_at_ms, updated_at_ms
		) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?)`,
		rehearsalID, manifest.AppCode, manifest.BatchID, restoreState, manifest.SchemaVersion, manifest.DataObjectKey, manifest.RowCount,
		manifest.FirstCursor.UpdatedAtMS, manifest.FirstCursor.EventID, manifest.LastCursor.UpdatedAtMS, manifest.LastCursor.EventID,
		manifest.AvailableDeltaTotal, manifest.FrozenDeltaTotal, manifest.NDJSONSHA256, manifest.GzipSHA256, manifest.DataCRC64,
		verification.ManifestSHA256, verification.ManifestCRC64, verification.SuccessSHA256, verification.SuccessCRC64, databaseName,
		nowMS, nowMS)
	return err
}

func insertRestoreRows(ctx context.Context, tx *sql.Tx, rehearsalID, batchID string, restoredAtMS int64, records []Record) error {
	if len(records) == 0 || len(records) > 100 {
		return errors.New("restore insert batch must contain 1..100 rows")
	}
	const columns = `(rehearsal_id, source_batch_id, app_code, event_id, event_type, transaction_id, command_id,
		user_id, asset_type, available_delta, frozen_delta, payload, status, worker_id, lock_until_ms,
		retry_count, next_retry_at_ms, last_error, created_at_ms, updated_at_ms, restored_at_ms)`
	placeholders := make([]string, 0, len(records))
	args := make([]any, 0, len(records)*21)
	for _, record := range records {
		placeholders = append(placeholders, `(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`)
		args = append(args,
			rehearsalID, batchID, record.AppCode, record.EventID, record.EventType, record.TransactionID, record.CommandID,
			record.UserID, record.AssetType, record.AvailableDelta, record.FrozenDelta, record.PayloadJSON, record.Status, record.WorkerID,
			nullInt64(record.LockUntilMS), record.RetryCount, nullInt64(record.NextRetryAtMS), nullString(record.LastError),
			record.CreatedAtMS, record.UpdatedAtMS, restoredAtMS,
		)
	}
	_, err := tx.ExecContext(ctx, `INSERT INTO wallet_outbox_restore_rows_v1 `+columns+` VALUES `+strings.Join(placeholders, ","), args...)
	return err
}

func verifyRestoredControls(ctx context.Context, query interface {
	QueryRowContext(context.Context, string, ...any) *sql.Row
}, rehearsalID string, verification Verification) error {
	var count int
	var availableTotal, frozenTotal string
	var minCreated, maxCreated, minUpdated, maxUpdated int64
	if err := query.QueryRowContext(ctx, `
		SELECT COUNT(*),
		       CAST(COALESCE(SUM(available_delta), 0) AS CHAR),
		       CAST(COALESCE(SUM(frozen_delta), 0) AS CHAR),
		       COALESCE(MIN(created_at_ms), 0), COALESCE(MAX(created_at_ms), 0),
		       COALESCE(MIN(updated_at_ms), 0), COALESCE(MAX(updated_at_ms), 0)
		FROM wallet_outbox_restore_rows_v1 FORCE INDEX (PRIMARY)
		WHERE rehearsal_id = ?`, rehearsalID).Scan(&count, &availableTotal, &frozenTotal, &minCreated, &maxCreated, &minUpdated, &maxUpdated); err != nil {
		return err
	}
	manifest := verification.Manifest
	if count != manifest.RowCount || availableTotal != manifest.AvailableDeltaTotal || frozenTotal != manifest.FrozenDeltaTotal || minCreated != manifest.MinCreatedAtMS || maxCreated != manifest.MaxCreatedAtMS || minUpdated != manifest.MinUpdatedAtMS || maxUpdated != manifest.MaxUpdatedAtMS {
		return errors.New("isolation-table row count, time bounds, or control totals do not match manifest")
	}
	var firstUpdated, lastUpdated int64
	var firstEvent, lastEvent string
	if err := query.QueryRowContext(ctx, `
		SELECT updated_at_ms, event_id FROM wallet_outbox_restore_rows_v1 FORCE INDEX (idx_wallet_outbox_restore_cursor)
		WHERE rehearsal_id = ? AND app_code = ? ORDER BY updated_at_ms ASC, event_id ASC LIMIT 1`, rehearsalID, manifest.AppCode).Scan(&firstUpdated, &firstEvent); err != nil {
		return err
	}
	if err := query.QueryRowContext(ctx, `
		SELECT updated_at_ms, event_id FROM wallet_outbox_restore_rows_v1 FORCE INDEX (idx_wallet_outbox_restore_cursor)
		WHERE rehearsal_id = ? AND app_code = ? ORDER BY updated_at_ms DESC, event_id DESC LIMIT 1`, rehearsalID, manifest.AppCode).Scan(&lastUpdated, &lastEvent); err != nil {
		return err
	}
	if (Cursor{UpdatedAtMS: firstUpdated, EventID: firstEvent}) != manifest.FirstCursor || (Cursor{UpdatedAtMS: lastUpdated, EventID: lastEvent}) != manifest.LastCursor {
		return errors.New("isolation-table cursor bounds do not match manifest")
	}
	return nil
}

func readExistingRestore(ctx context.Context, conn *sql.Conn, rehearsalID string, verification Verification) (RestoreResult, bool, error) {
	var state, appCode, batchID, manifestSHA, manifestCRC, successCRC, availableTotal, frozenTotal string
	var rowCount int
	var verifiedAtMS int64
	err := conn.QueryRowContext(ctx, `
		SELECT state, source_app_code, source_batch_id, row_count, manifest_sha256, manifest_crc64, success_crc64,
		       available_delta_total, frozen_delta_total, verified_at_ms
		FROM wallet_outbox_restore_rehearsals_v1 WHERE rehearsal_id = ?`, rehearsalID).
		Scan(&state, &appCode, &batchID, &rowCount, &manifestSHA, &manifestCRC, &successCRC, &availableTotal, &frozenTotal, &verifiedAtMS)
	if err == sql.ErrNoRows {
		return RestoreResult{}, false, nil
	}
	if err != nil {
		return RestoreResult{}, false, err
	}
	manifest := verification.Manifest
	if state != restoreVerifiedState || appCode != manifest.AppCode || batchID != manifest.BatchID || rowCount != manifest.RowCount || manifestSHA != verification.ManifestSHA256 || manifestCRC != verification.ManifestCRC64 || successCRC != verification.SuccessCRC64 || availableTotal != manifest.AvailableDeltaTotal || frozenTotal != manifest.FrozenDeltaTotal {
		return RestoreResult{}, false, errors.New("rehearsal_id already exists with non-matching or incomplete evidence")
	}
	if err := verifyRestoredControls(ctx, conn, rehearsalID, verification); err != nil {
		return RestoreResult{}, false, fmt.Errorf("existing restore readback failed: %w", err)
	}
	if err := verifyRestoredRowsExact(ctx, conn, rehearsalID, verification.Records); err != nil {
		return RestoreResult{}, false, fmt.Errorf("existing restore exact-row readback failed: %w", err)
	}
	result := newRestoreResult(rehearsalID, "", verification, verifiedAtMS, true)
	if err := conn.QueryRowContext(ctx, `SELECT DATABASE()`).Scan(&result.TargetDatabase); err != nil {
		return RestoreResult{}, false, err
	}
	return result, true, nil
}

func newRestoreResult(rehearsalID, databaseName string, verification Verification, verifiedAtMS int64, idempotent bool) RestoreResult {
	manifest := verification.Manifest
	return RestoreResult{
		State: restoreVerifiedState, RehearsalID: rehearsalID, BatchID: manifest.BatchID, AppCode: manifest.AppCode,
		RowCount: manifest.RowCount, TargetDatabase: databaseName, RehearsalTable: RehearsalTable, RowsTable: RowsTable,
		DataObjectKey: manifest.DataObjectKey, FirstCursor: manifest.FirstCursor, LastCursor: manifest.LastCursor,
		NDJSONSHA256: manifest.NDJSONSHA256, GzipSHA256: manifest.GzipSHA256, DataCRC64: manifest.DataCRC64,
		ManifestSHA256: verification.ManifestSHA256, ManifestCRC64: verification.ManifestCRC64,
		SuccessSHA256: verification.SuccessSHA256, SuccessCRC64: verification.SuccessCRC64,
		AvailableTotal: manifest.AvailableDeltaTotal, FrozenTotal: manifest.FrozenDeltaTotal,
		VerifiedAtMS: verifiedAtMS, IdempotentReadback: idempotent,
	}
}

func verifyRestoredRowsExact(ctx context.Context, query interface {
	QueryContext(context.Context, string, ...any) (*sql.Rows, error)
}, rehearsalID string, expected []Record) error {
	rows, err := query.QueryContext(ctx, `
		SELECT app_code, event_id, event_type, transaction_id, command_id, user_id, asset_type,
		       available_delta, frozen_delta, CAST(payload AS CHAR), status, worker_id,
		       lock_until_ms, retry_count, next_retry_at_ms, last_error, created_at_ms, updated_at_ms
		FROM wallet_outbox_restore_rows_v1 FORCE INDEX (idx_wallet_outbox_restore_cursor)
		WHERE rehearsal_id = ?
		ORDER BY app_code ASC, updated_at_ms ASC, event_id ASC`, rehearsalID)
	if err != nil {
		return err
	}
	defer rows.Close()
	index := 0
	for rows.Next() {
		if index >= len(expected) {
			return errors.New("isolation table contains more rows than verified archive")
		}
		var actual Record
		var lockUntilMS, nextRetryAtMS sql.NullInt64
		var lastError sql.NullString
		if err := rows.Scan(
			&actual.AppCode, &actual.EventID, &actual.EventType, &actual.TransactionID, &actual.CommandID,
			&actual.UserID, &actual.AssetType, &actual.AvailableDelta, &actual.FrozenDelta, &actual.PayloadJSON,
			&actual.Status, &actual.WorkerID, &lockUntilMS, &actual.RetryCount, &nextRetryAtMS, &lastError,
			&actual.CreatedAtMS, &actual.UpdatedAtMS,
		); err != nil {
			return err
		}
		if lockUntilMS.Valid {
			value := lockUntilMS.Int64
			actual.LockUntilMS = &value
		}
		if nextRetryAtMS.Valid {
			value := nextRetryAtMS.Int64
			actual.NextRetryAtMS = &value
		}
		if lastError.Valid {
			value := lastError.String
			actual.LastError = &value
		}
		// Source payload was archived from CAST(payload AS CHAR); casting the restored
		// JSON back through MySQL proves JSON parsing did not alter the owner snapshot.
		if !reflect.DeepEqual(actual, expected[index]) {
			return fmt.Errorf("isolation table row %d does not exactly match archive", index+1)
		}
		index++
	}
	if err := rows.Err(); err != nil {
		return err
	}
	if index != len(expected) {
		return errors.New("isolation table contains fewer rows than verified archive")
	}
	return nil
}

func acquireRestoreLock(ctx context.Context, conn *sql.Conn, rehearsalID string) (func(), bool, error) {
	hash := sha256.Sum256([]byte(rehearsalID))
	lockName := "wallet_outbox_restore:" + hex.EncodeToString(hash[:16])
	var acquired sql.NullInt64
	if err := conn.QueryRowContext(ctx, `SELECT GET_LOCK(?, 0)`, lockName).Scan(&acquired); err != nil {
		return nil, false, err
	}
	if !acquired.Valid || acquired.Int64 != 1 {
		return func() {}, false, nil
	}
	return func() {
		releaseCtx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
		defer cancel()
		var released sql.NullInt64
		_ = conn.QueryRowContext(releaseCtx, `SELECT RELEASE_LOCK(?)`, lockName).Scan(&released)
	}, true, nil
}

func validRehearsalID(value string) bool {
	if len(value) == 0 || len(value) > 64 {
		return false
	}
	for _, char := range value {
		if (char >= 'a' && char <= 'z') || (char >= 'A' && char <= 'Z') || (char >= '0' && char <= '9') || char == '.' || char == '_' || char == '-' {
			continue
		}
		return false
	}
	return true
}

func nullInt64(value *int64) any {
	if value == nil {
		return nil
	}
	return *value
}

func nullString(value *string) any {
	if value == nil {
		return nil
	}
	return *value
}