package automation

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

	"github.com/go-sql-driver/mysql"
	"github.com/google/uuid"
)

const scheduledRunReservationGlobalInput = "__global__"

type ScheduledRunReserveStatus string

const (
	ScheduledRunReserveStatusReserved  ScheduledRunReserveStatus = "reserved"
	ScheduledRunReserveStatusDuplicate ScheduledRunReserveStatus = "duplicate"
)

type ScheduledRunExecutionStatus string

const (
	ScheduledRunExecutionStatusAccepted  ScheduledRunExecutionStatus = "accepted"
	ScheduledRunExecutionStatusReserved  ScheduledRunExecutionStatus = "reserved"
	ScheduledRunExecutionStatusCompleted ScheduledRunExecutionStatus = "completed"
	ScheduledRunExecutionStatusFailed    ScheduledRunExecutionStatus = "failed"
)

type ScheduledRunReservationKey struct {
	FlowDefinitionID string
	FlowVersionID    string
	WindowToken      string
	InputKey         string
}

func NewScheduledRunReservationKey(flowDefinitionID, flowVersionID, windowToken, inputKey string) (ScheduledRunReservationKey, error) {
	key := ScheduledRunReservationKey{
		FlowDefinitionID: flowDefinitionID,
		FlowVersionID:    flowVersionID,
		WindowToken:      windowToken,
		InputKey:         inputKey,
	}
	return normalizeScheduledRunReservationKey(key)
}

func normalizeScheduledRunReservationKey(key ScheduledRunReservationKey) (ScheduledRunReservationKey, error) {
	key.FlowDefinitionID = strings.TrimSpace(key.FlowDefinitionID)
	key.FlowVersionID = strings.TrimSpace(key.FlowVersionID)
	key.WindowToken = strings.TrimSpace(key.WindowToken)
	key.InputKey = strings.TrimSpace(key.InputKey)
	if key.FlowDefinitionID == "" {
		return ScheduledRunReservationKey{}, fmt.Errorf("scheduled run reservation key: flow definition id is required")
	}
	if key.FlowVersionID == "" {
		return ScheduledRunReservationKey{}, fmt.Errorf("scheduled run reservation key: flow version id is required")
	}
	if key.WindowToken == "" {
		return ScheduledRunReservationKey{}, fmt.Errorf("scheduled run reservation key: window token is required")
	}
	if key.InputKey == "" {
		key.InputKey = scheduledRunReservationGlobalInput
	}
	if len(key.FlowDefinitionID) > 191 || len(key.FlowVersionID) > 191 ||
		len(key.WindowToken) > 191 || len(key.InputKey) > 191 {
		return ScheduledRunReservationKey{}, fmt.Errorf("scheduled run reservation key: identity exceeds the durable column bound")
	}
	return key, nil
}

func (k ScheduledRunReservationKey) CanonicalID() string {
	return strings.Join([]string{k.FlowDefinitionID, k.FlowVersionID, k.WindowToken, k.InputKey}, "\x1f")
}

type ScheduledRunReserveResult struct {
	Status        ScheduledRunReserveStatus
	ReservationID string
}

type ScheduledRunReservationService struct {
	db    *sql.DB
	nowFn func() time.Time
	idFn  func() string
}

func NewScheduledRunReservationService(db *sql.DB) *ScheduledRunReservationService {
	return &ScheduledRunReservationService{
		db:    db,
		nowFn: time.Now,
		idFn:  func() string { return uuid.NewString() },
	}
}

func (s *ScheduledRunReservationService) Reserve(ctx context.Context, key ScheduledRunReservationKey) (ScheduledRunReserveResult, error) {
	if s == nil || s.db == nil {
		return ScheduledRunReserveResult{}, fmt.Errorf("scheduled run reservation service: db is required")
	}
	key, err := normalizeScheduledRunReservationKey(key)
	if err != nil {
		return ScheduledRunReserveResult{}, err
	}
	now := s.nowFn().UTC()
	id := s.idFn()
	identityDigest := legacyScheduledRunOccurrenceIdentityDigest(key)
	_, err = s.db.ExecContext(ctx,
		`INSERT INTO _System_ScheduledRunReservation
			(Id, OccurrenceIdentityDigest, FlowDefinitionId, FlowVersionId, ScheduleWindowKey, WindowToken, InputKey, State, Status, ReservedAt, CreatedDate, LastModifiedDate)
		  VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
		id, identityDigest, key.FlowDefinitionID, key.FlowVersionID, key.WindowToken, key.WindowToken, key.InputKey,
		legacyScheduledRunExecutionState(ScheduledRunExecutionStatusReserved), string(ScheduledRunExecutionStatusReserved), now, now, now,
	)
	if err == nil {
		return ScheduledRunReserveResult{
			Status:        ScheduledRunReserveStatusReserved,
			ReservationID: id,
		}, nil
	}
	if isScheduledRunReservationDuplicateErr(err) {
		return ScheduledRunReserveResult{
			Status: ScheduledRunReserveStatusDuplicate,
		}, nil
	}
	return ScheduledRunReserveResult{}, err
}

func (s *ScheduledRunReservationService) Complete(ctx context.Context, reservationID string, key ScheduledRunReservationKey, status ScheduledRunExecutionStatus, lastError string) error {
	if s == nil || s.db == nil {
		return fmt.Errorf("scheduled run reservation service: db is required")
	}
	reservationID = strings.TrimSpace(reservationID)
	if reservationID == "" {
		return fmt.Errorf("scheduled run reservation complete: reservation id is required")
	}
	key, err := normalizeScheduledRunReservationKey(key)
	if err != nil {
		return err
	}
	if status != ScheduledRunExecutionStatusCompleted && status != ScheduledRunExecutionStatusFailed {
		return fmt.Errorf("scheduled run reservation complete: invalid terminal status %q", status)
	}
	now := s.nowFn().UTC()
	res, err := s.db.ExecContext(ctx,
		`UPDATE _System_ScheduledRunReservation
		    SET State = ?, Status = ?, CompletedAt = ?, LastError = ?, LastModifiedDate = ?
		  WHERE Id = ? AND FlowDefinitionId = ? AND FlowVersionId = ? AND WindowToken = ? AND InputKey = ? AND Status = ?`,
		legacyScheduledRunExecutionState(status), string(status), now, nullableScheduledRunLastError(lastError), now,
		reservationID, key.FlowDefinitionID, key.FlowVersionID, key.WindowToken, key.InputKey, string(ScheduledRunExecutionStatusReserved),
	)
	if err != nil {
		return err
	}
	rows, err := res.RowsAffected()
	if err != nil {
		return err
	}
	if rows == 0 {
		return fmt.Errorf("scheduled run reservation complete: reservation not active for %s (%s)", reservationID, key.CanonicalID())
	}
	return nil
}

func legacyScheduledRunOccurrenceIdentityDigest(key ScheduledRunReservationKey) string {
	return scheduledRunIdentityDigest(
		"legacy",
		key.FlowDefinitionID,
		key.FlowVersionID,
		key.WindowToken,
		key.InputKey,
	)
}

// scheduledRunIdentityDigest uses length-delimited components. Delimiter-based
// concatenation would let a control character embedded in one bounded VARCHAR
// make two different authority tuples hash the same byte sequence.
func scheduledRunIdentityDigest(parts ...string) string {
	digest := sha256.New()
	var size [8]byte
	for _, part := range parts {
		binary.BigEndian.PutUint64(size[:], uint64(len(part)))
		_, _ = digest.Write(size[:])
		_, _ = digest.Write([]byte(part))
	}
	return hex.EncodeToString(digest.Sum(nil))
}

func nullableScheduledRunLastError(value string) any {
	if strings.TrimSpace(value) == "" {
		return nil
	}
	return value
}

func legacyScheduledRunExecutionState(status ScheduledRunExecutionStatus) string {
	return strings.ToUpper(string(status))
}

func isScheduledRunReservationDuplicateErr(err error) bool {
	if err == nil {
		return false
	}

	var mysqlErr *mysql.MySQLError
	if errors.As(err, &mysqlErr) {
		return mysqlErr.Number == 1062
	}

	msg := err.Error()
	return strings.Contains(msg, "Duplicate entry") ||
		strings.Contains(msg, "Error 1062")
}