package persistence

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"strings"
)

const (
	runtimeFlowInvocationTable     = "_System_RuntimeFlowInvocation"
	RuntimeFlowInvocationAccepted  = "ACCEPTED"
	RuntimeFlowInvocationCompleted = "COMPLETED"
)

var (
	ErrRuntimeFlowInvocationConflict              = errors.New("runtime Flow invocation idempotency identity is bound to different content")
	ErrRuntimeFlowInvocationServingBindingChanged = errors.New("runtime Flow invocation is bound to a different serving revision")
	ErrRuntimeFlowInvocationTransactionRequired   = errors.New("runtime Flow invocation requires an active caller-owned transaction")
)

var runtimeFlowInvocationRequiredColumns = map[string]string{
	"Id":                "varchar(255)",
	"ScopeDigest":       "char(64)",
	"IdempotencyKey":    "varchar(255)",
	"DefinitionRef":     "varchar(255)",
	"DefinitionVersion": "varchar(255)",
	"ClosureDigest":     "char(71)",
	"RequestDigest":     "char(71)",
	"Receipts":          "json",
	"State":             "varchar(16)",
	"ResultPayload":     "mediumblob",
	"CreatedDate":       "datetime(6)",
	"LastModifiedDate":  "datetime(6)",
}

type RuntimeFlowInvocationRepository struct {
	db           *sql.DB
	transactions *TransactionManager
}

func NewRuntimeFlowInvocationRepository(
	db *sql.DB,
	transactions *TransactionManager,
) *RuntimeFlowInvocationRepository {
	return &RuntimeFlowInvocationRepository{db: db, transactions: transactions}
}

func VerifyRuntimeFlowInvocationSchema(ctx context.Context, db *sql.DB) error {
	if ctx == nil || db == nil {
		return errors.New("runtime Flow invocation schema: context and database are required")
	}
	if err := VerifyRequiredTableColumns(ctx, db, runtimeFlowInvocationTable, runtimeFlowInvocationRequiredColumns); err != nil {
		return fmt.Errorf("runtime Flow invocation schema: %w", err)
	}
	if err := VerifyExactTableIndex(
		ctx, db, runtimeFlowInvocationTable, "PRIMARY", true, []string{"Id"},
	); err != nil {
		return fmt.Errorf("runtime Flow invocation schema: %w", err)
	}
	if err := VerifyExactTableIndex(
		ctx,
		db,
		runtimeFlowInvocationTable,
		"ux_runtime_flow_invocation_scope",
		true,
		[]string{"ScopeDigest", "IdempotencyKey"},
	); err != nil {
		return fmt.Errorf("runtime Flow invocation schema: %w", err)
	}
	return nil
}

type RuntimeFlowInvocationAcceptance struct {
	RowID             string
	ScopeDigest       string
	IdempotencyKey    string
	DefinitionRef     string
	DefinitionVersion string
	ClosureDigest     string
	RequestDigest     string
	Receipts          []byte
}

type RuntimeFlowInvocationRow struct {
	RowID             string
	ScopeDigest       string
	IdempotencyKey    string
	DefinitionRef     string
	DefinitionVersion string
	ClosureDigest     string
	RequestDigest     string
	Receipts          []byte
	State             string
	ResultPayload     []byte
}

func (r *RuntimeFlowInvocationRepository) Accept(
	ctx context.Context,
	input RuntimeFlowInvocationAcceptance,
) (RuntimeFlowInvocationRow, bool, error) {
	if r == nil || r.db == nil || r.transactions == nil {
		return RuntimeFlowInvocationRow{}, false, errors.New("runtime Flow invocation repository is unavailable")
	}
	if !hexDigestPattern.MatchString(input.ScopeDigest) ||
		input.RowID == "" || input.IdempotencyKey == "" ||
		input.DefinitionRef == "" || input.DefinitionVersion == "" ||
		!validPrefixedSHA256(input.ClosureDigest) || !validPrefixedSHA256(input.RequestDigest) ||
		len(input.Receipts) == 0 {
		return RuntimeFlowInvocationRow{}, false, errors.New("runtime Flow invocation acceptance is incomplete")
	}
	exec, err := r.activeTransaction(ctx)
	if err != nil {
		return RuntimeFlowInvocationRow{}, false, err
	}
	result, err := exec.ExecContext(ctx, `INSERT IGNORE INTO _System_RuntimeFlowInvocation
(Id, ScopeDigest, IdempotencyKey, DefinitionRef, DefinitionVersion, ClosureDigest, RequestDigest, Receipts, State)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'ACCEPTED')`,
		input.RowID,
		input.ScopeDigest,
		input.IdempotencyKey,
		input.DefinitionRef,
		input.DefinitionVersion,
		input.ClosureDigest,
		input.RequestDigest,
		input.Receipts,
	)
	if err != nil {
		return RuntimeFlowInvocationRow{}, false, fmt.Errorf("accept runtime Flow invocation: %w", err)
	}
	affected, err := result.RowsAffected()
	if err != nil {
		return RuntimeFlowInvocationRow{}, false, fmt.Errorf("accept runtime Flow invocation rows affected: %w", err)
	}
	row, err := r.load(ctx, exec, input.ScopeDigest, input.RowID, input.IdempotencyKey)
	if err != nil {
		return RuntimeFlowInvocationRow{}, false, err
	}
	if row.RowID != input.RowID ||
		row.ScopeDigest != input.ScopeDigest ||
		row.IdempotencyKey != input.IdempotencyKey {
		return row, true, ErrRuntimeFlowInvocationConflict
	}
	sameReceipts, err := equivalentJSONDocuments(row.Receipts, input.Receipts)
	if err != nil {
		return row, true, fmt.Errorf("compare runtime Flow invocation receipts: %w", err)
	}
	if !sameReceipts {
		return row, true, ErrRuntimeFlowInvocationServingBindingChanged
	}
	if row.DefinitionRef != input.DefinitionRef ||
		row.DefinitionVersion != input.DefinitionVersion ||
		row.ClosureDigest != input.ClosureDigest ||
		row.RequestDigest != input.RequestDigest {
		return row, true, ErrRuntimeFlowInvocationConflict
	}
	return row, affected == 0, nil
}

func (r *RuntimeFlowInvocationRepository) load(
	ctx context.Context,
	exec Executor,
	scopeDigest string,
	rowID string,
	idempotencyKey string,
) (RuntimeFlowInvocationRow, error) {
	rows, err := exec.QueryContext(ctx, `SELECT Id, ScopeDigest, IdempotencyKey, DefinitionRef,
       DefinitionVersion, ClosureDigest, RequestDigest, Receipts, State, COALESCE(ResultPayload, '')
  FROM _System_RuntimeFlowInvocation
 WHERE Id = ? OR (ScopeDigest = ? AND IdempotencyKey = ?)
 ORDER BY Id
 LIMIT 2`, rowID, scopeDigest, idempotencyKey)
	if err != nil {
		return RuntimeFlowInvocationRow{}, fmt.Errorf("load runtime Flow invocation: %w", err)
	}
	defer rows.Close()
	var result RuntimeFlowInvocationRow
	count := 0
	for rows.Next() {
		var row RuntimeFlowInvocationRow
		if err := rows.Scan(
			&row.RowID,
			&row.ScopeDigest,
			&row.IdempotencyKey,
			&row.DefinitionRef,
			&row.DefinitionVersion,
			&row.ClosureDigest,
			&row.RequestDigest,
			&row.Receipts,
			&row.State,
			&row.ResultPayload,
		); err != nil {
			return RuntimeFlowInvocationRow{}, fmt.Errorf("load runtime Flow invocation row: %w", err)
		}
		count++
		result = row
	}
	if err := rows.Err(); err != nil {
		return RuntimeFlowInvocationRow{}, fmt.Errorf("load runtime Flow invocation rows: %w", err)
	}
	if count == 0 {
		return RuntimeFlowInvocationRow{}, errors.New("runtime Flow invocation insert was not observable")
	}
	if count != 1 {
		return RuntimeFlowInvocationRow{}, ErrRuntimeFlowInvocationConflict
	}
	return result, nil
}

func (r *RuntimeFlowInvocationRepository) Complete(ctx context.Context, rowID string, resultPayload []byte) error {
	if r == nil || r.db == nil || r.transactions == nil || rowID == "" || len(resultPayload) == 0 {
		return errors.New("runtime Flow invocation completion is incomplete")
	}
	exec, err := r.activeTransaction(ctx)
	if err != nil {
		return err
	}
	result, err := exec.ExecContext(ctx, `UPDATE _System_RuntimeFlowInvocation
SET State = 'COMPLETED', ResultPayload = ?, LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE Id = ? AND State = 'ACCEPTED'`, resultPayload, rowID)
	if err != nil {
		return fmt.Errorf("complete runtime Flow invocation: %w", err)
	}
	affected, err := result.RowsAffected()
	if err != nil {
		return fmt.Errorf("complete runtime Flow invocation rows affected: %w", err)
	}
	if affected != 1 {
		return errors.New("runtime Flow invocation completion lost ownership")
	}
	return nil
}

func (r *RuntimeFlowInvocationRepository) activeTransaction(ctx context.Context) (*sql.Tx, error) {
	if r == nil || r.transactions == nil {
		return nil, ErrRuntimeFlowInvocationTransactionRequired
	}
	tx, ok := r.transactions.ExtractTx(ctx).(*sql.Tx)
	if !ok || tx == nil {
		return nil, ErrRuntimeFlowInvocationTransactionRequired
	}
	return tx, nil
}

func validPrefixedSHA256(value string) bool {
	return strings.HasPrefix(value, "sha256:") &&
		len(value) == len("sha256:")+64 &&
		hexDigestPattern.MatchString(strings.TrimPrefix(value, "sha256:"))
}