package automation

import (
	"context"
	"encoding/json"
	"fmt"
	"os"
	"time"

	"github.com/google/uuid"

	"github.com/nexuscrm/backend/internal/domain/ports"
	apperrors "github.com/nexuscrm/backend/pkg/errors"
	"github.com/nexuscrm/shared/pkg/api"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
)

// FlowInterviewService manages flow interview (running/paused flow instances)
type FlowInterviewService struct {
	persistence PersistenceProvider
	query       ports.QueryService
}

// NewFlowInterviewService creates a new FlowInterviewService
func NewFlowInterviewService(p PersistenceProvider, q ports.QueryService) *FlowInterviewService {
	return &FlowInterviewService{
		persistence: p,
		query:       q,
	}
}

// CreateTransientInterview returns an in-memory FlowInterview without any
// database write, signature-compatible with CreateInterview so callers
// can swap modes without a branch. The returned struct is marked
// IsTransient so PauseInterview — the one persistence-touching method
// that could still see the Id via an external caller — rejects it at
// runtime instead of silently no-op-updating a non-existent row.
//
// Use this for synchronous record-triggered flows that run to completion
// in a single call. A record with many before-save flows (e.g. Lead has
// ~24) would otherwise do 2 * N DB writes against the FlowInterview table,
// which dominated save latency on remote TiDB (~160s measured on
// staging for a single Lead update). Salesforce itself defaults to not
// persisting finished record-triggered interviews — see Process
// Automation Settings → "Store Flow Interviews".
//
// The error return only fires if the crypto/rand backed UUID generator
// fails (`uuid.NewRandom`). On a working Linux kernel that implies much
// broader system failure, but we still surface the error so the caller
// can return a clean 5xx rather than relying on `uuid.New()`'s panic.
func (s *FlowInterviewService) CreateTransientInterview(flowVersionId string, label string, user *models.UserSession) (*models.FlowInterview, error) {
	id, err := uuid.NewRandom()
	if err != nil {
		return nil, apperrors.NewInternalError("failed to generate transient interview id", err)
	}
	ownerId := ""
	if user != nil {
		ownerId = user.ID
	}
	return &models.FlowInterview{
		Id:              id.String(),
		FlowVersionId:   flowVersionId,
		InterviewLabel:  label,
		InterviewStatus: models.FlowInterviewStatusRunning,
		OwnerId:         ownerId,
		ContextData:     make(map[string]interface{}),
		IsTransient:     true,
	}, nil
}

// CreateInterview creates a new flow interview instance
func (s *FlowInterviewService) CreateInterview(ctx context.Context, flowVersionId string, label string, user *models.UserSession) (*models.FlowInterview, error) {
	contextData := make(map[string]interface{})
	contextJSON, _ := json.Marshal(contextData)

	record := models.SObject{
		constants.FieldID: "", // Will be generated
		"FlowVersionId":   flowVersionId,
		"InterviewLabel":  label,
		"InterviewStatus": models.FlowInterviewStatusRunning,
		"OwnerId":         user.ID,
		"ContextData":     string(contextJSON),
	}

	result, err := s.persistence.Insert(ctx, "FlowInterview", record, user)
	if err != nil {
		return nil, err
	}

	return &models.FlowInterview{
		Id:              result[constants.FieldID].(string),
		FlowVersionId:   flowVersionId,
		InterviewLabel:  label,
		InterviewStatus: models.FlowInterviewStatusRunning,
		OwnerId:         user.ID,
		ContextData:     contextData,
	}, nil
}

// GetInterview retrieves a flow interview by ID
func (s *FlowInterviewService) GetInterview(ctx context.Context, interviewId string, user *models.UserSession) (*models.FlowInterview, error) {
	result, err := s.query.Query(ctx, api.QueryRequest{
		ObjectAPIName: "FlowInterview",
		Criteria:      []api.QueryCriterion{{Field: constants.FieldID, Op: "=", Val: interviewId}},
		Limit:         1,
	}, user)
	if err != nil {
		return nil, err
	}
	if result.TotalSize == 0 || len(result.Records) == 0 {
		return nil, nil
	}

	return recordToInterview(result.Records[0]), nil
}

// UpdateInterview updates a flow interview.
// Transient runtime fields are stripped before serialisation via
// marshalPersistableContext (shared with pauseForWait) to avoid persisting
// large record snapshots on every checkpoint. The original in-memory
// ContextData is never mutated.
func (s *FlowInterviewService) UpdateInterview(ctx context.Context, interview *models.FlowInterview, user *models.UserSession) error {
	contextJSON, err := marshalPersistableContext(interview.ContextData)
	if err != nil {
		return fmt.Errorf("failed to marshal context data: %w", err)
	}

	updates := models.SObject{
		"InterviewStatus": interview.InterviewStatus,
		"CurrentElement":  interview.CurrentElement,
		"ContextData":     string(contextJSON),
	}

	return s.persistence.Update(ctx, "FlowInterview", interview.Id, updates, user)
}

// PauseInterview pauses a running flow interview.
//
// The interview argument is a pointer rather than an id so the method can
// reject transient (in-memory only) interviews at the type boundary:
// calling Update against a transient interview's UUID would silently
// match zero rows on TiDB and leave the caller believing the flow was
// paused when nothing was persisted. Callers that only have an id in
// hand must first fetch the interview via GetInterview.
func (s *FlowInterviewService) PauseInterview(ctx context.Context, interview *models.FlowInterview, currentElement string, user *models.UserSession) error {
	if interview == nil {
		return apperrors.NewValidationError("interview", "interview cannot be nil")
	}
	if interview.IsTransient {
		// A transient interview has no row to update. Pausing it would
		// corrupt the flow lifecycle — return a loud error instead of
		// silently dropping the state transition.
		return apperrors.NewValidationError("interview", "cannot pause a transient flow interview; caller must promote it via CreateInterview first")
	}
	if err := validateIdentifier(interview.Id); err != nil {
		return apperrors.NewValidationError("interviewId", err.Error())
	}

	updates := models.SObject{
		"InterviewStatus": models.FlowInterviewStatusPaused,
		"CurrentElement":  currentElement,
	}

	return s.persistence.Update(ctx, "FlowInterview", interview.Id, updates, user)
}

// ResumeRunning flips a Paused interview back to Running and clears
// PausedUntil, so the scheduler query
//
//	SELECT ... FROM FlowInterview WHERE InterviewStatus='Paused' AND PausedUntil <= NOW()
//
// will no longer match it. Also syncs the in-memory struct.
func (s *FlowInterviewService) ResumeRunning(ctx context.Context, interview *models.FlowInterview, user *models.UserSession) error {
	if interview == nil {
		return apperrors.NewValidationError("interview", "interview cannot be nil")
	}
	if err := validateIdentifier(interview.Id); err != nil {
		return apperrors.NewValidationError("interviewId", err.Error())
	}
	updates := models.SObject{
		constants.FieldFlowInterview_InterviewStatus: models.FlowInterviewStatusRunning,
		// nil clears the TIMESTAMP column to NULL in persistence layer.
		constants.FieldFlowInterview_PausedUntil: nil,
	}
	if err := s.persistence.Update(ctx, constants.TableFlowInterview, interview.Id, updates, user); err != nil {
		return err
	}
	interview.InterviewStatus = models.FlowInterviewStatusRunning
	interview.PausedUntil = nil
	return nil
}

// ClaimPausedInterview atomically claims a Paused interview for resume.
// Uses a transaction with GetRaw (which reads the current row) + status
// check + Update. Returns (true, nil) if this process won the claim,
// (false, nil) if the interview is no longer Paused (another scheduler
// already claimed it). This prevents double-resume in multi-instance
// deployments.
func (s *FlowInterviewService) ClaimPausedInterview(ctx context.Context, interviewId string, user *models.UserSession) (bool, error) {
	var claimed bool
	err := s.persistence.RunInTransaction(ctx, func(txCtx context.Context) error {
		// SELECT ... FOR UPDATE: acquires a row lock so concurrent schedulers
		// block until this tx commits. Without FOR UPDATE, TiDB's snapshot
		// isolation allows two txs to both read Paused and both claim.
		row, err := s.persistence.GetLock(txCtx, constants.TableFlowInterview, interviewId)
		if err != nil {
			return err
		}
		if row == nil {
			return nil // interview gone
		}
		status, _ := row[constants.FieldFlowInterview_InterviewStatus].(string)
		if status != models.FlowInterviewStatusPaused {
			claimed = false
			return nil // already claimed by another scheduler
		}
		// Flip to Running inside the same tx, recording which node claimed it.
		nodeId, _ := os.Hostname()
		updates := models.SObject{
			constants.FieldFlowInterview_InterviewStatus: models.FlowInterviewStatusRunning,
			constants.FieldFlowInterview_PausedUntil:     nil,
			constants.FieldFlowInterview_ClaimedBy:       nodeId,
			constants.FieldFlowInterview_ClaimedAt:       time.Now().UTC().Format("2006-01-02 15:04:05"),
		}
		if err := s.persistence.Update(txCtx, constants.TableFlowInterview, interviewId, updates, user); err != nil {
			return err
		}
		claimed = true
		return nil
	})
	return claimed, err
}

// ListDuePausedInterviews returns the IDs of Paused interviews whose
// PausedUntil has elapsed, plus any stale-claimed interviews (Running but
// ClaimedAt older than 10 minutes, indicating a crashed scheduler node).
func (s *FlowInterviewService) ListDuePausedInterviews(ctx context.Context, now time.Time, limit int, user *models.UserSession) ([]string, error) {
	if limit <= 0 {
		limit = 100
	}
	nowSQL := now.UTC().Format("2006-01-02 15:04:05")

	// 1. Due paused interviews (normal path)
	result, err := s.query.Query(ctx, api.QueryRequest{
		ObjectAPIName: constants.TableFlowInterview,
		Criteria: []api.QueryCriterion{
			{Field: constants.FieldFlowInterview_InterviewStatus, Op: "=", Val: models.FlowInterviewStatusPaused},
			{Field: constants.FieldFlowInterview_PausedUntil, Op: "<=", Val: nowSQL},
		},
		SortField:     constants.FieldFlowInterview_PausedUntil,
		SortDirection: "ASC",
		Limit:         limit,
	}, user)
	if err != nil {
		return nil, err
	}

	ids := make([]string, 0, len(result.Records))
	for _, rec := range result.Records {
		if id, ok := rec[constants.FieldID].(string); ok && id != "" {
			ids = append(ids, id)
		}
	}

	// 2. Stale-claimed recovery: interviews that were claimed (Running) but
	// the claiming node crashed before completing. ClaimedAt > 10min ago
	// means the claim is abandoned. Reset them to Paused so they can be
	// re-claimed. This is safe because ClaimPausedInterview uses FOR UPDATE.
	staleThreshold := now.UTC().Add(-10 * time.Minute).Format("2006-01-02 15:04:05")
	staleResult, err := s.query.Query(ctx, api.QueryRequest{
		ObjectAPIName: constants.TableFlowInterview,
		Criteria: []api.QueryCriterion{
			{Field: constants.FieldFlowInterview_InterviewStatus, Op: "=", Val: models.FlowInterviewStatusRunning},
			{Field: constants.FieldFlowInterview_ClaimedAt, Op: "<=", Val: staleThreshold},
			{Field: constants.FieldFlowInterview_ClaimedAt, Op: "!=", Val: ""},
		},
		Limit: limit / 4, // cap stale recovery to avoid dominating the batch
	}, user)
	if err == nil {
		for _, rec := range staleResult.Records {
			if id, ok := rec[constants.FieldID].(string); ok && id != "" {
				// Reset to Paused so ClaimPausedInterview can re-claim it.
				_ = s.persistence.Update(ctx, constants.TableFlowInterview, id, models.SObject{
					constants.FieldFlowInterview_InterviewStatus: models.FlowInterviewStatusPaused,
					constants.FieldFlowInterview_ClaimedBy:       nil,
					constants.FieldFlowInterview_ClaimedAt:       nil,
				}, user)
				ids = append(ids, id)
			}
		}
	}

	return ids, nil
}

// CompleteInterview marks a flow interview as complete
func (s *FlowInterviewService) CompleteInterview(ctx context.Context, interviewId string, status string, user *models.UserSession) error {
	if err := validateIdentifier(interviewId); err != nil {
		return apperrors.NewValidationError("interviewId", err.Error())
	}

	if status == "" {
		status = models.FlowInterviewStatusFinished
	}

	updates := models.SObject{
		"InterviewStatus": status,
	}

	return s.persistence.Update(ctx, "FlowInterview", interviewId, updates, user)
}

// ListInterviews returns all flow interviews for the current user
func (s *FlowInterviewService) ListInterviews(ctx context.Context, user *models.UserSession) ([]models.FlowInterview, error) {
	result, err := s.query.Query(ctx, api.QueryRequest{
		ObjectAPIName: "FlowInterview",
		Criteria: []api.QueryCriterion{
			{Field: "InterviewStatus", Op: "IN", Val: []string{"Running", "Paused"}},
		},
		SortField:     "CreatedDate",
		SortDirection: "DESC",
		Limit:         100,
	}, user)
	if err != nil {
		return nil, err
	}

	interviews := make([]models.FlowInterview, 0, len(result.Records))
	for _, record := range result.Records {
		interview := recordToInterview(record)
		if interview != nil {
			interviews = append(interviews, *interview)
		}
	}

	return interviews, nil
}

// DeleteInterview deletes a flow interview
func (s *FlowInterviewService) DeleteInterview(ctx context.Context, interviewId string, user *models.UserSession) error {
	// Validate interviewId to prevent SQL injection
	if err := validateIdentifier(interviewId); err != nil {
		return apperrors.NewValidationError("interviewId", err.Error())
	}

	return s.persistence.Delete(ctx, "FlowInterview", interviewId, user)
}

// GetProgress returns progress information for a flow interview (for approval workflows)
func (s *FlowInterviewService) GetProgress(ctx context.Context, interviewId string, user *models.UserSession) (*models.FlowInstanceProgress, error) {
	interview, err := s.GetInterview(ctx, interviewId, user)
	if err != nil {
		return nil, err
	}
	if interview == nil {
		return nil, nil
	}

	// Get flow version to determine total steps
	flowResult, err := s.query.Query(ctx, api.QueryRequest{
		ObjectAPIName: constants.TableFlow,
		Criteria:      []api.QueryCriterion{{Field: constants.FieldID, Op: "=", Val: interview.FlowVersionId}},
		Limit:         1,
	}, user)
	if err != nil {
		return nil, err
	}

	flowName := "Unknown Flow"
	totalSteps := 0
	completedSteps := 0

	if flowResult.TotalSize > 0 && len(flowResult.Records) > 0 {
		if label, ok := flowResult.Records[0]["MasterLabel"].(string); ok {
			flowName = label
		}
		// Calculate steps from metadata if available
		if metaJSON, ok := flowResult.Records[0]["Metadata"].(string); ok && metaJSON != "" {
			var meta models.FlowMetadata
			if json.Unmarshal([]byte(metaJSON), &meta) == nil {
				totalSteps = countFlowElements(&meta)
			}
		}
	}

	// Estimate completed steps based on context data
	if interview.ContextData != nil {
		if completed, ok := interview.ContextData["_completedSteps"].(float64); ok {
			completedSteps = int(completed)
		}
	}

	return &models.FlowInstanceProgress{
		InstanceId:     interview.Id,
		FlowName:       flowName,
		CurrentElement: interview.CurrentElement,
		Status:         interview.InterviewStatus,
		CompletedSteps: completedSteps,
		TotalSteps:     totalSteps,
	}, nil
}

// Helper: Convert SObject record to FlowInterview
func recordToInterview(record models.SObject) *models.FlowInterview {
	if record == nil {
		return nil
	}

	interview := &models.FlowInterview{}

	if id, ok := record[constants.FieldID].(string); ok {
		interview.Id = id
	}
	if flowVersionId, ok := record["FlowVersionId"].(string); ok {
		interview.FlowVersionId = flowVersionId
	}
	if label, ok := record["InterviewLabel"].(string); ok {
		interview.InterviewLabel = label
	}
	if status, ok := record["InterviewStatus"].(string); ok {
		interview.InterviewStatus = status
	}
	if currentElement, ok := record["CurrentElement"].(string); ok {
		interview.CurrentElement = currentElement
	}
	if ownerId, ok := record["OwnerId"].(string); ok {
		interview.OwnerId = ownerId
	}
	if contextJSON, ok := record["ContextData"].(string); ok && contextJSON != "" {
		var contextData map[string]interface{}
		if json.Unmarshal([]byte(contextJSON), &contextData) == nil {
			interview.ContextData = contextData
		}
	}
	// PausedUntil can come back as time.Time (native driver) or string
	// (JSON / SObject round-trip). Accept both so the scheduler's query
	// result can feed straight into the executor.
	if raw, ok := record[constants.FieldFlowInterview_PausedUntil]; ok && raw != nil {
		switch v := raw.(type) {
		case time.Time:
			if !v.IsZero() {
				t := v
				interview.PausedUntil = &t
			}
		case *time.Time:
			if v != nil && !v.IsZero() {
				interview.PausedUntil = v
			}
		case string:
			if v != "" {
				for _, f := range []string{time.RFC3339Nano, time.RFC3339, "2006-01-02 15:04:05", "2006-01-02T15:04:05Z"} {
					if t, err := time.Parse(f, v); err == nil {
						interview.PausedUntil = &t
						break
					}
				}
			}
		}
	}

	return interview
}

// Helper: Count total flow elements
func countFlowElements(meta *models.FlowMetadata) int {
	count := 0
	if meta == nil {
		return count
	}

	count += len(meta.Decisions)
	count += len(meta.Assignments)
	count += len(meta.RecordCreates)
	count += len(meta.RecordUpdates)
	count += len(meta.RecordDeletes)
	count += len(meta.RecordLookups)
	count += len(meta.ActionCalls)
	count += len(meta.Subflows)
	count += len(meta.Screens)

	return count
}