package automation

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

	"github.com/google/uuid"
	eventfeature "github.com/nexuscrm/events"
	"github.com/nexuscrm/metadata/catalogsnapshot"
	"github.com/nexuscrm/persistence"
	flowsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowv1"
)

const maxTimedFlowOffset = 366 * 24 * time.Hour

// TimedFlowDueRequest is one record-triggered scheduled path selected from an
// already verified immutable Flow view. The producer derives event time from
// the durable record-event row; callers cannot supply a process clock. An
// AfterCommit request is due at that durable event time, while a timed request
// must retain a positive bounded offset.
type TimedFlowDueRequest struct {
	Binding          catalogsnapshot.RuntimeWorkBinding
	FlowDefinitionID string
	FlowVersionID    string
	PathName         string
	EntryElement     string
	RecordID         string
	Offset           time.Duration
	AfterCommit      bool
	CanonicalInputs  []byte
}

func (request TimedFlowDueRequest) validateLaunchBound() error {
	if request.PathName == "" || len(request.PathName) > 191 ||
		request.EntryElement == "" || len(request.EntryElement) > 191 ||
		request.RecordID == "" || len(request.RecordID) > 191 ||
		(request.AfterCommit && request.Offset != 0) ||
		(!request.AfterCommit && (request.Offset <= 0 || request.Offset > maxTimedFlowOffset)) {
		return errors.New("timed Flow due producer: path identity or offset is outside the launch bound")
	}
	return nil
}

// TimedFlowDueProducer writes the timed occurrence in the record-event
// effect/ack transaction. It neither starts nor selects a worker. Production
// readiness separately proves that the scheduled work family has an active
// owner; the consumer still transactionally fences its exact deployment tuple.
type TimedFlowDueProducer struct {
	db           *sql.DB
	transactions scheduledRunTransactionExtractor
	ownership    *persistence.RuntimeWorkOwnershipRepository
	idFn         func() string
}

func newTimedFlowDueProducer(
	db *sql.DB,
	transactions scheduledRunTransactionExtractor,
) (*TimedFlowDueProducer, error) {
	if db == nil || transactions == nil {
		return nil, errors.New("timed Flow due producer: database and transaction extractor are required")
	}
	return &TimedFlowDueProducer{
		db: db, transactions: transactions, idFn: func() string { return uuid.NewString() },
	}, nil
}

// NewFencedTimedFlowDueProducer binds acceptance to the exact scheduled
// consumer deployment selected by the stopped-upgrade composition. The
// ownership row is locked in the same record-event transaction as the new due
// occurrence, so transferred or mismatched authority cannot accept work.
func NewFencedTimedFlowDueProducer(
	db *sql.DB,
	transactions scheduledRunTransactionExtractor,
	expected persistence.RuntimeWorkOwnership,
) (*TimedFlowDueProducer, error) {
	producer, err := newTimedFlowDueProducer(db, transactions)
	if err != nil {
		return nil, err
	}
	if expected.WorkFamily != RuntimeScheduledFlowWorkFamily {
		return nil, errors.New("timed Flow due producer: exact scheduled Flow ownership is required")
	}
	ownership, err := persistence.NewRuntimeWorkOwnershipRepository(db, expected)
	if err != nil {
		return nil, err
	}
	producer.ownership = ownership
	return producer, nil
}

func (p *TimedFlowDueProducer) CheckReady(ctx context.Context) error {
	if p == nil || p.db == nil || p.transactions == nil {
		return errors.New("timed Flow due producer: producer is unavailable")
	}
	if err := verifyFencedScheduledRunSchema(ctx, p.db); err != nil {
		return err
	}
	var err error
	if p.ownership != nil {
		err = p.ownership.CheckReady(ctx)
	} else {
		err = persistence.VerifyActiveRuntimeWorkFamily(ctx, p.db, RuntimeScheduledFlowWorkFamily)
	}
	if err != nil {
		return fmt.Errorf(
			"timed Flow due producer: scheduled work family %s: %w",
			RuntimeScheduledFlowWorkFamily,
			err,
		)
	}
	return nil
}

func (p *TimedFlowDueProducer) Accept(
	ctx context.Context,
	request TimedFlowDueRequest,
) error {
	if p == nil || p.db == nil || p.transactions == nil {
		return errors.New("timed Flow due producer: producer is unavailable")
	}
	if ctx == nil || ctx.Err() != nil {
		return errors.New("timed Flow due producer: live effect transaction context is required")
	}
	eventID, ok := eventfeature.CurrentOutboxEventID(ctx)
	if !ok {
		return errors.New("timed Flow due producer: durable record-event identity is required")
	}
	if err := request.Binding.Validate(); err != nil {
		return fmt.Errorf("timed Flow due producer: immutable Flow binding: %w", err)
	}
	if request.Binding.RuntimeCapabilityID != flowsemantic.CapabilityName {
		return errors.New("timed Flow due producer: binding does not own Flow execution")
	}
	request.PathName = strings.TrimSpace(request.PathName)
	request.EntryElement = strings.TrimSpace(request.EntryElement)
	request.RecordID = strings.TrimSpace(request.RecordID)
	if err := request.validateLaunchBound(); err != nil {
		return err
	}
	if err := validateScheduledRunCanonicalInputs(request.CanonicalInputs); err != nil {
		return err
	}
	windowToken := timedFlowWindowToken(eventID, request.PathName)
	key, err := NewScheduledRunReservationKey(
		request.FlowDefinitionID,
		request.FlowVersionID,
		windowToken,
		request.RecordID,
	)
	if err != nil {
		return err
	}
	binding, err := catalogsnapshot.EncodeRuntimeWorkBinding(request.Binding)
	if err != nil {
		return fmt.Errorf("timed Flow due producer: encode immutable Flow binding: %w", err)
	}
	flowScope := persistence.RuntimeWorkScope{
		OrgID: request.Binding.OrgID, AppInstallationID: request.Binding.AppInstallationID,
		RuntimeCapabilityID: request.Binding.RuntimeCapabilityID,
	}
	if err := flowScope.Validate(); err != nil {
		return fmt.Errorf("timed Flow due producer: Flow work scope: %w", err)
	}
	eventScope, ok := persistence.RuntimeWorkScopeFromContext(ctx)
	if !ok {
		return errors.New("timed Flow due producer: durable record-event scope is required")
	}
	if eventScope.OrgID != flowScope.OrgID ||
		eventScope.AppInstallationID != flowScope.AppInstallationID {
		return errors.New("timed Flow due producer: durable record event is outside the Flow installation")
	}
	eventScopeDigest, err := eventScope.Digest()
	if err != nil {
		return fmt.Errorf("timed Flow due producer: derive record-event work scope digest: %w", err)
	}
	tx, ok := p.transactions.ExtractTx(ctx).(*sql.Tx)
	if !ok || tx == nil {
		return errors.New("timed Flow due producer: record-event effect transaction is required")
	}
	if p.ownership != nil {
		if err := p.ownership.LockActive(ctx, tx); err != nil {
			return fmt.Errorf("timed Flow due producer: scheduled consumer authority: %w", err)
		}
	}

	var eventCreatedAt time.Time
	if err := tx.QueryRowContext(ctx, `SELECT CreatedDate
FROM _System_RecordEventOutbox
WHERE Id = ? AND ExecutionScopeDigest = ? AND OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
FOR UPDATE`,
		eventID, eventScopeDigest, eventScope.OrgID, eventScope.AppInstallationID, eventScope.RuntimeCapabilityID,
	).Scan(&eventCreatedAt); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return errors.New("timed Flow due producer: durable record event is outside the Flow scope")
		}
		return fmt.Errorf("timed Flow due producer: read durable event time: %w", err)
	}
	databaseDueAt := eventCreatedAt.Add(request.Offset)
	dueAt := scheduledRunDatabaseUTC(eventCreatedAt).Add(request.Offset)
	identityDigest := scheduledRunOccurrenceIdentityDigest(flowScope, key)
	batchDigest := scheduledRunBatchIdentityDigest(flowScope, key.FlowDefinitionID, key.FlowVersionID, key.WindowToken)

	var existing struct {
		flowDefinitionID string
		flowVersionID    string
		windowToken      string
		inputKey         string
		state            string
		status           string
		entryElement     string
		dueAt            time.Time
		metadataBinding  string
		payloadFormat    sql.NullInt64
		payload          []byte
	}
	err = tx.QueryRowContext(ctx, `SELECT FlowDefinitionId, FlowVersionId, COALESCE(WindowToken, ''), InputKey,
       State, COALESCE(Status, ''), COALESCE(EntryElement, ''), DueAt,
       COALESCE(MetadataBinding, ''), InputPayloadFormatVersion, InputPayload
FROM _System_ScheduledRunReservation
WHERE OccurrenceIdentityDigest = ?
FOR UPDATE`,
		identityDigest,
	).Scan(
		&existing.flowDefinitionID, &existing.flowVersionID, &existing.windowToken, &existing.inputKey,
		&existing.state, &existing.status, &existing.entryElement, &existing.dueAt,
		&existing.metadataBinding, &existing.payloadFormat, &existing.payload,
	)
	if err == nil {
		existing.dueAt = scheduledRunDatabaseUTC(existing.dueAt)
		if existing.flowDefinitionID != key.FlowDefinitionID ||
			existing.flowVersionID != key.FlowVersionID ||
			existing.windowToken != key.WindowToken ||
			existing.inputKey != key.InputKey ||
			existing.entryElement != request.EntryElement ||
			!existing.dueAt.Equal(dueAt) ||
			existing.metadataBinding != binding ||
			!existing.payloadFormat.Valid ||
			existing.payloadFormat.Int64 != scheduledRunInputPayloadFormatV1 ||
			!bytes.Equal(existing.payload, request.CanonicalInputs) ||
			existing.state != legacyScheduledRunExecutionState(ScheduledRunExecutionStatus(existing.status)) {
			return errors.New("timed Flow due producer: existing occurrence disagrees with the durable event")
		}
		switch ScheduledRunExecutionStatus(existing.status) {
		case ScheduledRunExecutionStatusAccepted,
			ScheduledRunExecutionStatusReserved,
			ScheduledRunExecutionStatusCompleted,
			ScheduledRunExecutionStatusFailed,
			ScheduledRunExecutionStatusPoisoned:
			return nil
		default:
			return errors.New("timed Flow due producer: existing occurrence has invalid state")
		}
	}
	if !errors.Is(err, sql.ErrNoRows) {
		return fmt.Errorf("timed Flow due producer: inspect occurrence: %w", err)
	}

	_, err = tx.ExecContext(ctx, `INSERT INTO _System_ScheduledRunReservation
(Id, OrgId, AppInstallationId, RuntimeCapabilityId, OccurrenceIdentityDigest, BatchIdentityDigest,
 FlowDefinitionId, FlowVersionId, ScheduleWindowKey, WindowToken, InputKey,
 State, Status, Attempt, EntryElement, DueAt,
 MetadataBinding, InputPayloadFormatVersion, InputPayload)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
        'ACCEPTED', 'accepted', 0, ?, ?, ?, ?, ?)`,
		p.idFn(),
		flowScope.OrgID, flowScope.AppInstallationID, flowScope.RuntimeCapabilityID, identityDigest, batchDigest,
		key.FlowDefinitionID, key.FlowVersionID, key.WindowToken, key.WindowToken, key.InputKey,
		request.EntryElement, databaseDueAt, binding, scheduledRunInputPayloadFormatV1,
		request.CanonicalInputs,
	)
	if err != nil {
		return fmt.Errorf("timed Flow due producer: persist accepted occurrence: %w", err)
	}
	return nil
}

func timedFlowWindowToken(eventID, pathName string) string {
	sum := sha256.Sum256([]byte(strings.TrimSpace(eventID) + "\x00" + strings.TrimSpace(pathName)))
	return "timed:" + hex.EncodeToString(sum[:])
}