package triggersub

import (
	"context"
	"crypto/sha1"
	"database/sql"
	"encoding/hex"
	"errors"
	"fmt"
	"log"
	"strings"

	"github.com/nexuscrm/automation/triggerprojection"
	"github.com/nexuscrm/shared/pkg/models"
)

const tableName = "_System_TriggerSubscription"

type Executor interface {
	ExecContext(ctx context.Context, query string, args ...interface{}) (sql.Result, error)
	QueryContext(ctx context.Context, query string, args ...interface{}) (*sql.Rows, error)
	QueryRowContext(ctx context.Context, query string, args ...interface{}) *sql.Row
}

type txStarter interface {
	BeginTx(ctx context.Context, opts *sql.TxOptions) (*sql.Tx, error)
}

type Row struct {
	Id                string
	KeyPrefix         string
	EventType         string
	RecordTriggerType string
	FlowDefinitionId  string
	FlowVersionId     string
	IsActive          bool
}

func ExpectedActiveRows(ctx context.Context, exec Executor) ([]Row, error) {
	if exec == nil {
		return nil, fmt.Errorf("trigger subscription expected-row resolution requires an executor")
	}
	definitions, err := loadActiveDefinitions(ctx, exec)
	if err != nil {
		return nil, err
	}
	expectedRows := make([]Row, 0, len(definitions)*2)
	for _, item := range definitions {
		rows, err := expectedRowsForDefinition(ctx, exec, item.definition, item.version)
		if err != nil {
			return nil, err
		}
		expectedRows = append(expectedRows, rows...)
	}
	return expectedRows, nil
}

func Backfill(ctx context.Context, exec Executor) error {
	if exec == nil {
		return fmt.Errorf("trigger subscription backfill requires an executor")
	}
	return withWriteTx(ctx, exec, "backfill", func(tx Executor) error {
		expectedRows, err := ExpectedActiveRows(ctx, tx)
		if err != nil {
			return err
		}
		if _, err := tx.ExecContext(ctx, `UPDATE `+tableName+` SET IsActive = 0, LastModifiedDate = NOW() WHERE IsActive = 1`); err != nil {
			return fmt.Errorf("disable active trigger subscriptions before backfill: %w", err)
		}
		return upsertRows(ctx, tx, expectedRows)
	})
}

func ReconcileDefinition(ctx context.Context, exec Executor, definitionID string) error {
	if exec == nil {
		return fmt.Errorf("trigger subscription reconcile requires an executor")
	}
	if strings.TrimSpace(definitionID) == "" {
		return nil
	}
	return withWriteTx(ctx, exec, "reconcile", func(tx Executor) error {
		definition, version, err := loadDefinitionAndVersion(ctx, tx, definitionID)
		if err != nil {
			return err
		}

		var rows []Row
		if definition != nil && version != nil {
			rows, err = expectedRowsForDefinition(ctx, tx, definition, version)
			if err != nil {
				return err
			}
		}
		if _, err := tx.ExecContext(ctx, `UPDATE `+tableName+` SET IsActive = 0, LastModifiedDate = NOW() WHERE FlowDefinitionId = ?`, definitionID); err != nil {
			return fmt.Errorf("disable trigger subscriptions for flow definition %s: %w", definitionID, err)
		}
		return upsertRows(ctx, tx, rows)
	})
}

func Disable(ctx context.Context, exec Executor, definitionID string) error {
	if exec == nil {
		return fmt.Errorf("trigger subscription disable requires an executor")
	}
	if strings.TrimSpace(definitionID) == "" {
		return nil
	}
	if _, err := exec.ExecContext(ctx, `UPDATE `+tableName+` SET IsActive = 0, LastModifiedDate = NOW() WHERE FlowDefinitionId = ?`, definitionID); err != nil {
		return fmt.Errorf("disable trigger subscriptions for flow definition %s: %w", definitionID, err)
	}
	return nil
}

func Delete(ctx context.Context, exec Executor, definitionID string) error {
	if exec == nil {
		return fmt.Errorf("trigger subscription delete requires an executor")
	}
	if strings.TrimSpace(definitionID) == "" {
		return nil
	}
	if _, err := exec.ExecContext(ctx, `DELETE FROM `+tableName+` WHERE FlowDefinitionId = ?`, definitionID); err != nil {
		return fmt.Errorf("delete trigger subscriptions for flow definition %s: %w", definitionID, err)
	}
	return nil
}

func List(ctx context.Context, exec Executor, keyPrefix string, eventType string, recordTriggerType string) ([]Row, error) {
	if exec == nil {
		return nil, fmt.Errorf("trigger subscription list requires an executor")
	}
	rows, err := exec.QueryContext(ctx, `
		SELECT Id, KeyPrefix, EventType, RecordTriggerType, FlowDefinitionId, FlowVersionId, IsActive
		FROM `+tableName+`
		WHERE KeyPrefix COLLATE utf8mb4_bin = ? AND EventType = ? AND RecordTriggerType = ? AND IsActive = 1
		ORDER BY FlowDefinitionId, FlowVersionId`,
		strings.TrimSpace(keyPrefix),
		strings.TrimSpace(eventType),
		strings.TrimSpace(recordTriggerType),
	)
	if err != nil {
		return nil, fmt.Errorf("list trigger subscriptions: %w", err)
	}
	defer rows.Close()

	var out []Row
	for rows.Next() {
		var row Row
		if err := rows.Scan(&row.Id, &row.KeyPrefix, &row.EventType, &row.RecordTriggerType, &row.FlowDefinitionId, &row.FlowVersionId, &row.IsActive); err != nil {
			return nil, fmt.Errorf("scan trigger subscription row: %w", err)
		}
		out = append(out, row)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate trigger subscriptions: %w", err)
	}
	return out, nil
}

type activeDefinition struct {
	definition *models.FlowDefinition
	version    *models.FlowVersion
}

func loadActiveDefinitions(ctx context.Context, exec Executor) ([]activeDefinition, error) {
	rows, err := exec.QueryContext(ctx, `
		SELECT
			fd.Id,
			fd.DeveloperName,
			fd.ProcessType,
			fd.ActiveVersionId,
			fv.Id,
			fv.DefinitionId,
			fv.VersionNumber,
			fv.Status,
			fv.Description,
			fv.ProcessType,
			fv.Metadata,
			fv.Source,
			COALESCE(fv.IsDeleted, 0)
		FROM FlowDefinition fd
		LEFT JOIN Flow fv ON fv.Id = fd.ActiveVersionId
		WHERE COALESCE(fd.IsDeleted, 0) = 0
		  AND fd.ActiveVersionId IS NOT NULL
		ORDER BY fd.Id`)
	if err != nil {
		return nil, fmt.Errorf("query active flow definitions for trigger subscriptions: %w", err)
	}
	defer rows.Close()

	var out []activeDefinition
	for rows.Next() {
		item, err := scanActiveDefinition(rows)
		if err != nil {
			return nil, err
		}
		out = append(out, item)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate active flow definitions for trigger subscriptions: %w", err)
	}
	return out, nil
}

func loadDefinitionAndVersion(ctx context.Context, exec Executor, definitionID string) (*models.FlowDefinition, *models.FlowVersion, error) {
	rows, err := exec.QueryContext(ctx, `
		SELECT
			fd.Id,
			fd.DeveloperName,
			fd.ProcessType,
			fd.ActiveVersionId,
			fv.Id,
			fv.DefinitionId,
			fv.VersionNumber,
			fv.Status,
			fv.Description,
			fv.ProcessType,
			fv.Metadata,
			fv.Source,
			COALESCE(fv.IsDeleted, 0)
		FROM FlowDefinition fd
		LEFT JOIN Flow fv ON fv.Id = fd.ActiveVersionId
		WHERE fd.Id = ?
		  AND COALESCE(fd.IsDeleted, 0) = 0
		LIMIT 1
		FOR UPDATE`, definitionID)
	if err != nil {
		return nil, nil, fmt.Errorf("query flow definition %s for trigger subscriptions: %w", definitionID, err)
	}
	defer rows.Close()

	if !rows.Next() {
		return nil, nil, nil
	}
	item, err := scanActiveDefinition(rows)
	if err != nil {
		return nil, nil, err
	}
	return item.definition, item.version, nil
}

func scanActiveDefinition(rows *sql.Rows) (activeDefinition, error) {
	var (
		definition models.FlowDefinition
		version    models.FlowVersion
		activeID   sql.NullString
		versionID  sql.NullString
		versionDef sql.NullString
		versionNum sql.NullInt64
		status     sql.NullString
		desc       sql.NullString
		process    sql.NullString
		metadata   sql.NullString
		source     sql.NullString
		isDeleted  bool
	)
	if err := rows.Scan(
		&definition.Id,
		&definition.DeveloperName,
		&definition.ProcessType,
		&activeID,
		&versionID,
		&versionDef,
		&versionNum,
		&status,
		&desc,
		&process,
		&metadata,
		&source,
		&isDeleted,
	); err != nil {
		return activeDefinition{}, fmt.Errorf("scan active flow definition for trigger subscriptions: %w", err)
	}
	if activeID.Valid {
		definition.ActiveVersionId = activeID.String
	}
	if strings.TrimSpace(definition.ActiveVersionId) == "" {
		return activeDefinition{
			definition: &definition,
			version:    nil,
		}, nil
	}
	if !versionID.Valid {
		return activeDefinition{}, fmt.Errorf("flow definition %s has ActiveVersionId=%q but no matching Flow row", definition.Id, definition.ActiveVersionId)
	}

	version.Id = versionID.String
	version.DefinitionId = versionDef.String
	if versionNum.Valid {
		version.VersionNumber = int(versionNum.Int64)
	}
	version.Status = status.String
	version.Description = desc.String
	version.ProcessType = process.String
	version.MetadataJSON = metadata.String
	version.Source = source.String
	version.IsDeleted = isDeleted
	version.HydrateMetadata()

	return activeDefinition{
		definition: &definition,
		version:    &version,
	}, nil
}

func expectedRowsForDefinition(ctx context.Context, exec Executor, definition *models.FlowDefinition, version *models.FlowVersion) ([]Row, error) {
	if definition == nil || version == nil {
		return nil, nil
	}
	if version.Status != models.FlowStatusActive || version.IsDeleted {
		return nil, nil
	}
	if err := triggerprojection.ValidateAtomicTriggerSupport(definition, version); err != nil {
		if errors.Is(err, triggerprojection.ErrUnsupportedAtomicTriggerShape) {
			log.Printf(
				"TriggerSubscription: skipping unsupported active atomic trigger flow definition=%s developer=%s version=%s: %v",
				strings.TrimSpace(definition.Id),
				strings.TrimSpace(definition.DeveloperName),
				strings.TrimSpace(version.Id),
				err,
			)
			return nil, nil
		}
		return nil, fmt.Errorf("validate atomic trigger support for flow definition %s: %w", definition.Id, err)
	}

	projection, err := triggerprojection.Resolve(definition, version)
	if err != nil {
		return nil, fmt.Errorf("resolve trigger projection for flow definition %s: %w", definition.Id, err)
	}
	if projection.Kind != triggerprojection.KindRecord {
		return nil, nil
	}

	keyPrefix, err := keyPrefixForObject(ctx, exec, projection.ObjectOrEvent)
	if err != nil {
		return nil, fmt.Errorf("resolve key prefix for flow definition %s object %s: %w", definition.Id, projection.ObjectOrEvent, err)
	}

	recordTriggerTypes, err := expandRecordTriggerTypes(definition, version, projection.TriggerType)
	if err != nil {
		return nil, fmt.Errorf("resolve record trigger types for flow definition %s: %w", definition.Id, err)
	}

	rows := make([]Row, 0, len(recordTriggerTypes))
	for _, recordTriggerType := range recordTriggerTypes {
		rows = append(rows, Row{
			Id:                rowID(definition.Id, keyPrefix, projection.TriggerType, recordTriggerType),
			KeyPrefix:         keyPrefix,
			EventType:         projection.TriggerType,
			RecordTriggerType: recordTriggerType,
			FlowDefinitionId:  definition.Id,
			FlowVersionId:     version.Id,
			IsActive:          true,
		})
	}
	return rows, nil
}

func keyPrefixForObject(ctx context.Context, exec Executor, objectAPIName string) (string, error) {
	var keyPrefix sql.NullString
	if err := exec.QueryRowContext(ctx, `
		SELECT KeyPrefix
		FROM _System_Object
		WHERE LOWER(ApiName) = LOWER(?)
		  AND COALESCE(IsDeleted, 0) = 0
		LIMIT 1`, strings.TrimSpace(objectAPIName)).Scan(&keyPrefix); err != nil {
		if err == sql.ErrNoRows {
			return "", fmt.Errorf("object metadata missing")
		}
		return "", err
	}
	if !keyPrefix.Valid || strings.TrimSpace(keyPrefix.String) == "" {
		return "", fmt.Errorf("object metadata missing key prefix")
	}
	return strings.TrimSpace(keyPrefix.String), nil
}

func expandRecordTriggerTypes(definition *models.FlowDefinition, version *models.FlowVersion, eventType string) ([]string, error) {
	if version != nil && version.Metadata.Start != nil {
		if expanded, handled, err := expandExplicitRecordTriggerType(version.Metadata.Start.RecordTriggerType); handled {
			return expanded, err
		}
	}

	processType := strings.TrimSpace(version.ProcessType)
	if processType == "" && definition != nil {
		processType = strings.TrimSpace(definition.ProcessType)
	}
	if strings.EqualFold(processType, "Workflow") {
		return []string{"Create", "Update"}, nil
	}

	switch strings.TrimSpace(eventType) {
	case "RecordBeforeSave", "RecordAfterSave":
		return []string{"Create", "Update"}, nil
	case "RecordBeforeDelete", "RecordAfterDelete":
		return []string{"Delete"}, nil
	default:
		return nil, fmt.Errorf("unsupported record trigger event type %q", eventType)
	}
}

func expandExplicitRecordTriggerType(value string) ([]string, bool, error) {
	switch {
	case strings.TrimSpace(value) == "":
		return nil, false, nil
	case strings.EqualFold(value, "Create"):
		return []string{"Create"}, true, nil
	case strings.EqualFold(value, "Update"):
		return []string{"Update"}, true, nil
	case strings.EqualFold(value, "CreateAndUpdate"):
		return []string{"Create", "Update"}, true, nil
	case strings.EqualFold(value, "Delete"):
		return []string{"Delete"}, true, nil
	default:
		return nil, true, fmt.Errorf("unsupported RecordTriggerType %q", value)
	}
}

func upsertRows(ctx context.Context, exec Executor, rows []Row) error {
	for _, row := range rows {
		if _, err := exec.ExecContext(ctx, `
			INSERT INTO `+tableName+`
				(Id, KeyPrefix, EventType, RecordTriggerType, FlowDefinitionId, FlowVersionId, IsActive, ConditionShapeHash, CreatedDate, LastModifiedDate)
			VALUES (?, ?, ?, ?, ?, ?, ?, NULL, NOW(), NOW())
			ON DUPLICATE KEY UPDATE
				FlowVersionId = VALUES(FlowVersionId),
				IsActive = VALUES(IsActive),
				LastModifiedDate = NOW()`,
			row.Id,
			row.KeyPrefix,
			row.EventType,
			row.RecordTriggerType,
			row.FlowDefinitionId,
			row.FlowVersionId,
			row.IsActive,
		); err != nil {
			return fmt.Errorf("upsert trigger subscription row %s: %w", row.Id, err)
		}
	}
	return nil
}

func rowID(flowDefinitionID string, keyPrefix string, eventType string, recordTriggerType string) string {
	sum := sha1.Sum([]byte(strings.Join([]string{
		strings.TrimSpace(flowDefinitionID),
		strings.TrimSpace(keyPrefix),
		strings.TrimSpace(eventType),
		strings.TrimSpace(recordTriggerType),
	}, "\x00")))
	return "trigsub_" + hex.EncodeToString(sum[:])
}

func withWriteTx(ctx context.Context, exec Executor, operation string, fn func(tx Executor) error) error {
	if tx, ok := exec.(*sql.Tx); ok {
		return fn(tx)
	}
	starter, ok := exec.(txStarter)
	if !ok {
		return fmt.Errorf("trigger subscription %s requires a transactional executor", operation)
	}
	tx, err := starter.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("begin trigger subscription %s transaction: %w", operation, err)
	}
	defer func() {
		_ = tx.Rollback()
	}()
	if err := fn(tx); err != nil {
		return err
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("commit trigger subscription %s transaction: %w", operation, err)
	}
	return nil
}