package records

import (
	"context"
	"encoding/json"
	"errors"
	"time"

	"github.com/nexuscrm/persistence"
	semantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordmutationv1"
)

// RuntimeRecordCommandLifecycleFence is the database-owned worker generation
// authority. Closure must compare it in the same transaction that writes the
// terminal command outcome.
type RuntimeRecordCommandLifecycleFence interface {
	LockActiveContext(context.Context) error
}

type runtimeRecordCommandLifecycleStore interface {
	ListExpiredAccepted(context.Context, int) ([]string, error)
	LockCommand(context.Context, string) (persistence.RuntimeRecordCommandRow, error)
	LockOperations(context.Context, string) ([]persistence.RuntimeRecordCommandOperationRow, error)
	CompleteOperation(context.Context, string, int, []byte) error
	CompleteCommand(context.Context, string, []byte, time.Time) error
}

// RuntimeRecordCommandLifecycle closes accepted commands that exhausted their
// immutable execution window. It never replays caller data and never opens a
// Catalog or Security revision: committed independent positions are preserved,
// pending positions become caller-safe terminal failures, and atomic commands
// retain zero effects.
type RuntimeRecordCommandLifecycle struct {
	commands runtimeRecordCommandLifecycleStore
	tx       RuntimeRecordMutationTransactions
	fence    RuntimeRecordCommandLifecycleFence
}

func NewRuntimeRecordCommandLifecycle(commands runtimeRecordCommandLifecycleStore, tx RuntimeRecordMutationTransactions, fence RuntimeRecordCommandLifecycleFence) (*RuntimeRecordCommandLifecycle, error) {
	if commands == nil || tx == nil || fence == nil {
		return nil, errors.New("runtime record command lifecycle: command store, transactions, and ownership fence are required")
	}
	return &RuntimeRecordCommandLifecycle{commands: commands, tx: tx, fence: fence}, nil
}

// ResolveExpired closes at most batchSize commands expired according to the
// database clock. The original acceptance-to-expiry interval becomes the fresh
// terminal outcome retention, so an outage cannot make the proof expire before
// it is first observable.
func (l *RuntimeRecordCommandLifecycle) ResolveExpired(ctx context.Context, batchSize int) (int64, error) {
	if l == nil || l.commands == nil || l.tx == nil || l.fence == nil || batchSize <= 0 || batchSize > 10_000 {
		return 0, errors.New("runtime record command lifecycle: invalid resolution request")
	}
	ids, err := l.commands.ListExpiredAccepted(ctx, batchSize)
	if err != nil {
		return 0, err
	}
	var resolved int64
	for _, id := range ids {
		closed := false
		err := l.tx.WithPessimisticRetry(ctx, func(txCtx context.Context) error {
			// A retry reuses this callback closure. Do not let an aborted attempt
			// contribute a committed-resolution metric.
			closed = false
			if err := l.fence.LockActiveContext(txCtx); err != nil {
				return err
			}
			command, err := l.commands.LockCommand(txCtx, id)
			if err != nil {
				return err
			}
			databaseNow, err := l.tx.ReadTransactionTime(txCtx)
			if err != nil {
				return err
			}
			if command.State == persistence.RuntimeRecordCommandCompleted || command.OutcomeExpiresAt.After(databaseNow) {
				return nil
			}
			retention, err := runtimeRecordCommandOutcomeRetention(command)
			if err != nil {
				return errors.New("runtime record command lifecycle: accepted command has invalid retention lineage")
			}
			slots, err := l.commands.LockOperations(txCtx, command.RowID)
			if err != nil {
				return err
			}
			results, pending, err := interruptedRuntimeRecordCommandResults(command, slots)
			if err != nil {
				return err
			}
			for _, index := range pending {
				payload, err := json.Marshal(results[index])
				if err != nil {
					return err
				}
				if err := l.commands.CompleteOperation(txCtx, command.RowID, index, payload); err != nil {
					return err
				}
			}
			var receipts semantic.Receipts
			if err := json.Unmarshal(command.ReceiptsPayload, &receipts); err != nil {
				return err
			}
			// The expiry is both a DATETIME(6) column and part of the durable JSON
			// response. Keep the two representations byte-stable on reconciliation.
			terminalExpiry := databaseNow.UTC().Add(retention).Truncate(time.Microsecond)
			command.OutcomeExpiresAt = terminalExpiry
			response := semantic.Response{
				CommandID: command.CommandID, CommandDigest: command.RequestDigest,
				Results: results, Receipts: receipts, OutcomeExpiresAtUTC: terminalExpiry.Format(time.RFC3339Nano),
			}
			shape, err := runtimeRecordCommandShape(command, slots)
			if err != nil {
				return err
			}
			statusRequest := semantic.StatusRequest{CommandID: command.CommandID, CommandDigest: command.RequestDigest}
			status := semantic.StatusResponse{
				CommandID: response.CommandID, CommandDigest: response.CommandDigest, State: semantic.CommandCompleted,
				Results: response.Results, Receipts: &response.Receipts, OutcomeExpiresAtUTC: response.OutcomeExpiresAtUTC,
			}
			if err := semantic.ValidateStatusResponse(shape, statusRequest, status); err != nil {
				return errors.New("runtime record command lifecycle: terminal outcome is invalid")
			}
			payload, err := json.Marshal(response)
			if err != nil {
				return err
			}
			if err := l.commands.CompleteCommand(txCtx, command.RowID, payload, terminalExpiry); err != nil {
				return err
			}
			closed = true
			return nil
		}, runtimeRecordCommandTransactionRetries)
		if err != nil {
			return resolved, err
		}
		if closed {
			resolved++
		}
	}
	return resolved, nil
}