package tenantoverlay

import (
	"bytes"
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"time"

	"github.com/pingcap-inc/nexuscrm-controlplane/packs"
	"github.com/nexuscrm/automation/flowauthoring"
	"github.com/nexuscrm/automation/triggerprojection"
	"github.com/nexuscrm/kernel/pkg/formula"
	"github.com/nexuscrm/shared/pkg/models"
)

// flowFilenameSuffix bounds which files are treated as flows.
// Anything else in the flows/ directory (e.g. README, .keep) is ignored.
const flowFilenameSuffix = ".flow.json"

// retiredTenantFlowDeveloperNames enumerates tenant flow DeveloperNames that
// used to ship from this quarantine directory but no longer do. When a flow is
// removed from the pack, add its DeveloperName here in the same commit so apply
// can soft-delete the previously managed definition without reintroducing a
// package-ownership registry.
var retiredTenantFlowDeveloperNames = []string{}

// seedTenantFlows reconciles tenants/flows/*.flow.json into the
// FlowDefinition / Flow tables. FlowDefinition.IsCustomized is the ownership
// guard: tenant apply may reconcile only definitions that remain
// IsCustomized=0.
func (p *Pack) seedTenantFlows(ctx context.Context, deps packs.ApplyDeps) error {
	flowDir := p.path(FlowsDir)
	entries, err := os.ReadDir(flowDir)
	if err != nil {
		if errors.Is(err, os.ErrNotExist) {
			return nil
		}
		return fmt.Errorf("read flows directory: %w", err)
	}

	shipped := make(map[string]struct{}, len(entries))
	names := make([]string, 0, len(entries))
	for _, e := range entries {
		if e.IsDir() || !strings.HasSuffix(e.Name(), flowFilenameSuffix) {
			continue
		}
		names = append(names, e.Name())
	}
	sort.Strings(names)

	for _, fileName := range names {
		developerName := strings.TrimSuffix(fileName, flowFilenameSuffix)
		shipped[developerName] = struct{}{}

		raw, err := os.ReadFile(filepath.Join(flowDir, fileName))
		if err != nil {
			return fmt.Errorf("read flow %q: %w", fileName, err)
		}
		if err := withTenantFlowWriteTx(ctx, deps, func(txCtx context.Context) error {
			return reconcileTenantFlow(txCtx, deps, developerName, raw)
		}); err != nil {
			return fmt.Errorf("reconcile flow %q: %w", developerName, err)
		}
	}

	if err := withTenantFlowWriteTx(ctx, deps, func(txCtx context.Context) error {
		return softDeleteMissingTenantFlows(txCtx, deps, shipped)
	}); err != nil {
		return err
	}

	// Tenant flow writes go through MetadataRepository directly, which bypasses
	// MetadataService's per-flow trigger shadow-state reconcile hooks. Rebuild
	// the authoritative runtime projection here so API-triggered workflow/flow
	// execution sees the newly seeded or retired tenant definitions immediately.
	if deps.Metadata != nil {
		if err := packs.RunFlowRuntimeMaintenanceAfterOrphanRepair(ctx, deps.DBConn, deps.Metadata, "tenant-overlay"); err != nil {
			return fmt.Errorf("tenant flow runtime maintenance: %w", err)
		}
	}
	return nil
}

func withTenantFlowWriteTx(ctx context.Context, deps packs.ApplyDeps, fn func(context.Context) error) error {
	if deps.TxManager == nil {
		return fn(ctx)
	}
	return deps.TxManager.WithPessimisticTransaction(ctx, fn)
}

// reconcileTenantFlow converges one FlowDefinition and its versions onto the
// shipped JSON. Customized rows are preserved with an explicit warning rather
// than being silently overwritten.
func reconcileTenantFlow(ctx context.Context, deps packs.ApplyDeps, developerName string, rawJSON []byte) error {
	canonical, err := canonicalizeFlowMetadata(rawJSON)
	if err != nil {
		return fmt.Errorf("canonicalize metadata: %w", err)
	}

	var meta models.FlowMetadata
	if err := json.Unmarshal(rawJSON, &meta); err != nil {
		return fmt.Errorf("parse metadata: %w", err)
	}
	processType := extractFlowProcessType(rawJSON)
	if err := triggerprojection.ValidateAtomicTriggerSupport(
		&models.FlowDefinition{ProcessType: processType},
		&models.FlowVersion{ProcessType: processType, Metadata: meta},
	); err != nil {
		return fmt.Errorf("validate atomic trigger support: %w", err)
	}

	// Validate filterFormula on the tenant-overlay apply path using the
	// cross-object-aware validator (ValidateFilterFormulaExpressionWithSchemas):
	//   Tier 1: syntax + registered-function check (does NOT reject $-globals).
	//   Tier 2: global-allowlist check (rejects $Flow, $Setup, $Label; permits
	//           $Permission, $User, $Profile, $UserRole, $Record, $Record__Old).
	//   Tier 3: schema-aware boolean + field-ref check against live schema.
	//           Skipped when schema is nil (object not found) or deps.Metadata is nil.
	//           __c custom fields and __r relationship paths are tolerated (#1018, #1724).
	//   extraSchemas: full object index so cross-object paths ($Record.Owner.Name)
	//           are validated at apply time (HIGH-2 fix, #2047 round-4).
	if meta.Start != nil {
		if expr := strings.TrimSpace(meta.Start.FilterFormula); expr != "" {
			var objSchema *models.ObjectMetadata
			var extraSchemas map[string]*models.ObjectMetadata
			if deps.Metadata != nil {
				objectName := strings.TrimSpace(meta.Start.Object)
				if objectName != "" {
					objSchema = deps.Metadata.GetSchema(ctx, objectName)
				}
				// Build a cross-object index for __r path validation.
				// Use GetSchemas (decorated, served from coreCache) NOT GetAllObjects
				// (raw repo): the trigger schema above comes from GetSchema (decorated),
				// so the cross-object index must be decorated too — otherwise it lacks
				// decorator-injected virtual fields such as User.Name (compound), and
				// $Record.Owner.Name would false-reject here while the trigger path
				// accepts it. GetSchemas is also cache-served (no per-apply full scan).
				allObjs := deps.Metadata.GetSchemas(ctx)
				extraSchemas = make(map[string]*models.ObjectMetadata, len(allObjs))
				for _, obj := range allObjs {
					key := strings.ToLower(strings.TrimSpace(obj.APIName))
					if key != "" {
						extraSchemas[key] = obj
					}
				}
			}
			eng := formula.NewEngine()
			if err := eng.ValidateFilterFormulaExpressionWithSchemas(expr, objSchema, extraSchemas); err != nil {
				return fmt.Errorf("flow %q: start.filterFormula: %w", developerName, err)
			}
		}
	}
	if meta.Start != nil && meta.Start.ConnectorSource != nil {
		return fmt.Errorf("tenant overlay flow %q cannot declare connectorSource; only pack-owned standard flows may use a connector directory source", developerName)
	}

	existing, err := deps.MetadataRepo.GetFlowDefinitionByName(ctx, developerName)
	if err != nil {
		return fmt.Errorf("lookup existing definition: %w", err)
	}

	if existing == nil {
		defID, err := newTenantFlowID(ctx, deps.DB, "FlowDefinition")
		if err != nil {
			return fmt.Errorf("generate definition id: %w", err)
		}
		stake := &models.FlowDefinition{
			Id:            defID,
			DeveloperName: developerName,
		}
		existing, err = deps.MetadataRepo.EnsureFlowDefinitionByDeveloperName(ctx, stake)
		if err != nil {
			return fmt.Errorf("stake-out definition: %w", err)
		}
	} else {
		existing, err = deps.MetadataRepo.LockFlowDefinitionByDeveloperNameForUpdate(ctx, developerName)
		if err != nil {
			return fmt.Errorf("lock definition by name: %w", err)
		}
	}
	if existing == nil {
		return fmt.Errorf("definition disappeared after lock: %s", developerName)
	}
	if existing.IsCustomized {
		log.Printf("   ⚠️  tenant flow %s: preserved customized FlowDefinition; skipping tenant reconcile", developerName)
		return nil
	}

	targetVersionID, err := ensureActiveFlowVersionMatches(ctx, deps, existing, &meta, processType, canonical)
	if err != nil {
		return err
	}

	projection, err := triggerprojection.Resolve(
		&models.FlowDefinition{ProcessType: processType},
		&models.FlowVersion{ProcessType: processType, Metadata: meta},
	)
	if err != nil {
		return fmt.Errorf("resolve trigger projection: %w", err)
	}
	target := &models.FlowDefinition{
		DeveloperName:        existing.DeveloperName,
		MasterLabel:          firstNonEmpty(meta.Label, developerName),
		Description:          meta.Description,
		ProcessType:          processType,
		TriggerType:          projection.TriggerType,
		TriggerObjectOrEvent: projection.ObjectOrEvent,
		ActiveVersionId:      targetVersionID,
		LatestVersionId:      targetVersionID,
		IsDeleted:            false,
	}
	if err := deps.MetadataRepo.UpdateFlowDefinition(ctx, existing.Id, target); err != nil {
		return fmt.Errorf("sync definition projection: %w", err)
	}
	return nil
}

// ensureActiveFlowVersionMatches returns the Id of a FlowVersion whose
// (canonical metadata, processType) tuple equals the shipped tuple
// AND whose Status is Active. Caller must hold the FlowDefinition
// FOR UPDATE lock for existing.Id. It locks/reads the version set with
// SELECT FOR UPDATE so TiDB observes the latest committed versions after
// waiting for the parent lock.
func ensureActiveFlowVersionMatches(
	ctx context.Context,
	deps packs.ApplyDeps,
	existing *models.FlowDefinition,
	meta *models.FlowMetadata,
	processType string,
	canonicalMetadata []byte,
) (string, error) {
	versions, err := deps.MetadataRepo.GetFlowVersionsForUpdate(ctx, existing.Id)
	if err != nil {
		return "", fmt.Errorf("lock versions: %w", err)
	}

	if existing.ActiveVersionId != "" {
		var active *models.FlowVersion
		for _, v := range versions {
			if v.Id == existing.ActiveVersionId {
				active = v
				break
			}
		}
		if active != nil {
			stored, err := canonicalizeFlowMetadata([]byte(active.MetadataJSON))
			if err != nil {
				return "", fmt.Errorf("canonicalize stored metadata: %w", err)
			}
			if bytesEqual(stored, canonicalMetadata) &&
				active.ProcessType == processType &&
				strings.EqualFold(active.Source, "custom") {
				return active.Id, nil
			}
		}
	}

	nextVer := 1
	for _, v := range versions {
		if v.VersionNumber >= nextVer {
			nextVer = v.VersionNumber + 1
		}
	}

	newVersionID, err := newTenantFlowID(ctx, deps.DB, "Flow")
	if err != nil {
		return "", fmt.Errorf("generate flow version id: %w", err)
	}
	if err := deps.MetadataRepo.CreateTenantFlowVersion(ctx, &models.FlowVersion{
		Id:            newVersionID,
		DefinitionId:  existing.Id,
		VersionNumber: nextVer,
		Status:        models.FlowStatusDraft,
		Description:   meta.Description,
		ProcessType:   processType,
		MetadataRaw:   canonicalMetadata,
	}); err != nil {
		return "", fmt.Errorf("create version v%d: %w", nextVer, err)
	}
	if err := deps.MetadataRepo.ActivateFlowVersion(ctx, newVersionID); err != nil {
		return "", fmt.Errorf("activate version v%d: %w", nextVer, err)
	}
	return newVersionID, nil
}

func softDeleteMissingTenantFlows(ctx context.Context, deps packs.ApplyDeps, shipped map[string]struct{}) error {
	for _, developerName := range retiredTenantFlowDeveloperNames {
		if _, keep := shipped[developerName]; keep {
			continue
		}
		def, err := deps.MetadataRepo.GetFlowDefinitionByName(ctx, developerName)
		if err != nil {
			return fmt.Errorf("lookup retired flow %q: %w", developerName, err)
		}
		if def == nil || def.IsDeleted || def.IsCustomized {
			continue
		}
		target := &models.FlowDefinition{
			DeveloperName:        def.DeveloperName,
			MasterLabel:          def.MasterLabel,
			Description:          def.Description,
			ProcessType:          def.ProcessType,
			TriggerType:          def.TriggerType,
			TriggerObjectOrEvent: def.TriggerObjectOrEvent,
			ActiveVersionId:      "",
			LatestVersionId:      def.LatestVersionId,
			IsDeleted:            true,
		}
		if err := deps.MetadataRepo.UpdateFlowDefinition(ctx, def.Id, target); err != nil {
			return fmt.Errorf("soft-delete retired tenant flow %q: %w", developerName, err)
		}
	}
	return nil
}

// canonicalizeFlowMetadata validates and normalises flow metadata JSON.
func canonicalizeFlowMetadata(raw []byte) ([]byte, error) {
	if len(raw) == 0 {
		return []byte("{}"), nil
	}
	// Validate via typed struct — catches malformed metadata early.
	var meta models.FlowMetadata
	if err := json.Unmarshal(raw, &meta); err != nil {
		return nil, err
	}
	if err := flowauthoring.ValidateNoLegacyEmailAlertActionCalls(meta); err != nil {
		return nil, fmt.Errorf("tenant flow metadata: %w", err)
	}
	processType := extractFlowProcessType(raw)
	if err := triggerprojection.ValidateAtomicTriggerSupport(
		&models.FlowDefinition{ProcessType: processType},
		&models.FlowVersion{ProcessType: processType, Metadata: meta},
	); err != nil {
		return nil, err
	}
	// Canonical output via json.Compact — preserves ALL fields (including
	// ones not modeled by FlowMetadata/FlowStart) while producing
	// deterministic byte-equal output for the same input.
	var buf bytes.Buffer
	if err := json.Compact(&buf, raw); err != nil {
		return nil, err
	}
	return buf.Bytes(), nil
}

// extractFlowProcessType pulls "processType" out of the raw JSON because
// FlowMetadata does not model that field.
func extractFlowProcessType(raw []byte) string {
	var wire struct {
		ProcessType string `json:"processType"`
	}
	_ = json.Unmarshal(raw, &wire)
	if wire.ProcessType == "" {
		return "AutoLaunchedFlow"
	}
	return wire.ProcessType
}

func firstNonEmpty(candidates ...string) string {
	for _, c := range candidates {
		if c != "" {
			return c
		}
	}
	return ""
}

// newTenantFlowID generates a Salesforce-compatible 18-char record ID for a
// Flow SObject row from _System_Object.KeyPrefix metadata.
func newTenantFlowID(ctx context.Context, db queryableDB, objectAPIName string) (string, error) {
	prefix, err := keyPrefixForObject(ctx, db, objectAPIName)
	if err != nil {
		return "", fmt.Errorf("key prefix metadata for %q: %w", objectAPIName, err)
	}
	if prefix == "" {
		return "", fmt.Errorf("no key prefix metadata for %q", objectAPIName)
	}
	return fmt.Sprintf("%s%015d", prefix, time.Now().UnixNano()%1000000000000000), nil
}

type queryableDB interface {
	QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row
}

func keyPrefixForObject(ctx context.Context, db queryableDB, objectAPIName string) (string, error) {
	if db == nil {
		return "", fmt.Errorf("nil db")
	}
	var prefix string
	if err := db.QueryRowContext(ctx,
		"SELECT KeyPrefix FROM _System_Object WHERE ApiName = ? AND IsDeleted = 0 LIMIT 1",
		objectAPIName,
	).Scan(&prefix); err != nil {
		return "", err
	}
	return prefix, nil
}

func bytesEqual(a, b []byte) bool {
	if len(a) != len(b) {
		return false
	}
	for i := range a {
		if a[i] != b[i] {
			return false
		}
	}
	return true
}