package control

import (
	"context"
	"encoding/json"
	"fmt"
	"log/slog"
	"os"
	"path/filepath"
	"sort"
	"sync"
	"time"
)

type FivetranFact struct {
	ConnectorID string
	SyncState   string
	UpdateState string
	Paused      bool
	SucceededAt *time.Time
	FailedAt    *time.Time
}

type BillingInputFact struct {
	TotalRows     int64
	ActualRows    int64
	PredictedRows int64
	LastSyncedAt  *time.Time
}

type TableFact struct {
	Name         string
	Rows         int64
	LastModified *time.Time
}

type MonthlyOutputFact struct {
	Orders      TableFact
	Collections TableFact
}

type FinanceTableFact struct {
	Total                int64
	Synced               int64
	Unsynced             int64
	IdentityRows         int64
	RequiredIdentityRows int64
	EstimatedRows        int64
	ActualRows           int64
	LockedRows           int64
	LastUpdated          *time.Time
	LastSynced           *time.Time
}

type FinanceCloseFact struct {
	BreakdownSucceededAt *time.Time
	BreakdownReceiptID   int64
	Order                FinanceTableFact
	RevRec               FinanceTableFact
	Collection           FinanceTableFact
}

type CRMTableFact struct {
	RunID           string
	Status          string
	Rows            int64
	DuplicatePKRows int64
	ObservedAt      *time.Time
}

type CRMPipelineFact struct {
	Order     CRMTableFact
	OrderItem CRMTableFact
	Product2  CRMTableFact
}

type SalesforceOrderFact struct {
	RunID            string
	Status           string
	SourceRows       int64
	TargetRows       int64
	DuplicateKeyRows int64
	CriticalNullRows int64
	FinishedAt       *time.Time
}

type SchedulerFact struct {
	Name            string
	State           string
	Schedule        string
	TimeZone        string
	NextRunAt       *time.Time
	LastAttemptAt   *time.Time
	LastStatusCode  int64
	Target          string
	HasExplicitBody bool
}

type ServiceFact struct {
	URL        string
	Healthy    bool
	ObservedAt *time.Time
}

type LiveFacts struct {
	ObservedAt     time.Time
	Fivetran       FivetranFact
	BillingInput   BillingInputFact
	MonthlyOutput  MonthlyOutputFact
	Finance        FinanceCloseFact
	CRM            CRMPipelineFact
	Salesforce     SalesforceOrderFact
	Scheduler      SchedulerFact
	FinanceService ServiceFact
	Dependencies   []DependencyObservation
}

type LiveDataSource interface {
	Read(context.Context, string, string) LiveFacts
}

type ObserverHealth struct {
	Status           string     `json:"status"`
	LastObservedAt   *time.Time `json:"last_observed_at,omitempty"`
	DependencyCount  int        `json:"dependency_count"`
	UnavailableCount int        `json:"unavailable_count"`
}

type LiveObserver struct {
	source     LiveDataSource
	statusPath string
	logger     *slog.Logger
	mu         sync.RWMutex
	health     ObserverHealth
	cache      map[string]*CloseObservation
}

func NewLiveObserver(source LiveDataSource, statusPath string, logger *slog.Logger) *LiveObserver {
	if logger == nil {
		logger = slog.Default()
	}
	return &LiveObserver{source: source, statusPath: statusPath, logger: logger, health: ObserverHealth{Status: "not_observed"}, cache: map[string]*CloseObservation{}}
}

func (o *LiveObserver) Start(ctx context.Context) {
	go func() {
		tick := time.NewTicker(time.Minute)
		defer tick.Stop()
		for {
			for _, mode := range []string{"predicted", "actual"} {
				call, cancel := context.WithTimeout(ctx, 45*time.Second)
				_, err := o.Observe(call, previousBillMonth(), mode)
				cancel()
				if err != nil {
					o.logger.Warn("close_observation_failed", "mode", mode)
				}
				if ctx.Err() != nil {
					return
				}
			}
			select {
			case <-ctx.Done():
				return
			case <-tick.C:
			}
		}
	}()
}

func (o *LiveObserver) Cached(billMonth, mode string) (*CloseObservation, bool) {
	o.mu.RLock()
	defer o.mu.RUnlock()
	value, ok := o.cache[billMonth+":"+mode]
	if !ok {
		return nil, false
	}
	p, _ := json.Marshal(value)
	var copy CloseObservation
	_ = json.Unmarshal(p, &copy)
	return &copy, true
}

func (o *LiveObserver) Observe(ctx context.Context, billMonth, mode string) (*CloseObservation, error) {
	if !billMonthPattern.MatchString(billMonth) {
		return nil, fmt.Errorf("bill_month must be YYYY-MM")
	}
	if mode != "predicted" && mode != "actual" {
		return nil, fmt.Errorf("mode must be predicted or actual")
	}
	facts := o.source.Read(ctx, billMonth, mode)
	observation := SynthesizeCloseObservation(billMonth, mode, facts)
	health := summarizeObserverHealth(observation)
	o.mu.Lock()
	o.health = health
	o.cache[billMonth+":"+mode] = observation
	o.mu.Unlock()
	if err := o.persist(observation); err != nil {
		o.logger.Warn("live-observation-persist-failed", "error_type", fmt.Sprintf("%T", err))
	}
	o.logger.Info("live-observation-completed",
		"bill_month", billMonth,
		"mode", mode,
		"close_status", observation.Close.Status,
		"next_gate", observation.NextGate,
		"dependency_status", health.Status,
		"unavailable_dependencies", health.UnavailableCount,
	)
	return observation, nil
}

func (o *LiveObserver) Health() ObserverHealth {
	o.mu.RLock()
	defer o.mu.RUnlock()
	return o.health
}

func (o *LiveObserver) persist(observation *CloseObservation) error {
	if o.statusPath == "" {
		return nil
	}
	if err := os.MkdirAll(filepath.Dir(o.statusPath), 0o755); err != nil {
		return err
	}
	payload, err := json.MarshalIndent(observation, "", "  ")
	if err != nil {
		return err
	}
	temporary := o.statusPath + ".tmp"
	if err := os.WriteFile(temporary, payload, 0o600); err != nil {
		return err
	}
	return os.Rename(temporary, o.statusPath)
}

func summarizeObserverHealth(observation *CloseObservation) ObserverHealth {
	health := ObserverHealth{Status: "healthy", LastObservedAt: &observation.ObservedAt, DependencyCount: len(observation.Dependencies)}
	degraded := false
	for _, dependency := range observation.Dependencies {
		if dependency.Status == "unavailable" {
			health.UnavailableCount++
		}
		if dependency.Status != "healthy" {
			degraded = true
		}
	}
	if health.UnavailableCount == len(observation.Dependencies) && health.DependencyCount > 0 {
		health.Status = "down"
	} else if degraded {
		health.Status = "degraded"
	}
	return health
}

func SynthesizeCloseObservation(billMonth, mode string, facts LiveFacts) *CloseObservation {
	observedAt := facts.ObservedAt.UTC()
	if observedAt.IsZero() {
		observedAt = time.Now().UTC()
	}
	closeRun := newCloseRun(billMonth, mode, observedAt)
	closeRun.ID = fmt.Sprintf("live_%s_%s", billMonth[0:4]+billMonth[5:7], mode)
	closeRun.Profile = "monthly_cloud_close_live"

	inputReceipt := map[string]any{
		"total_rows":              facts.BillingInput.TotalRows,
		"predicted_rows":          facts.BillingInput.PredictedRows,
		"actual_rows":             facts.BillingInput.ActualRows,
		"bigquery_last_synced_at": facts.BillingInput.LastSyncedAt,
	}
	if facts.BillingInput.TotalRows > 0 {
		markPassed(closeRun, "ar_table_ready", facts.BillingInput.LastSyncedAt, facts.BillingInput.LastSyncedAt, inputReceipt)
	}

	fivetranReceipt := map[string]any{
		"connector_id":  facts.Fivetran.ConnectorID,
		"sync_state":    facts.Fivetran.SyncState,
		"update_state":  facts.Fivetran.UpdateState,
		"paused":        facts.Fivetran.Paused,
		"bigquery_rows": facts.BillingInput.TotalRows,
	}
	if facts.Fivetran.SucceededAt != nil && !facts.Fivetran.Paused && facts.BillingInput.TotalRows > 0 {
		markPassed(closeRun, "sync_ar_to_bigquery", facts.Fivetran.SucceededAt, facts.BillingInput.LastSyncedAt, fivetranReceipt)
	} else {
		markObserved(closeRun, "sync_ar_to_bigquery", facts.Fivetran.SucceededAt, fivetranReceipt)
	}

	selectedRows := facts.BillingInput.PredictedRows
	selectedLabel := "predicted_rows"
	if mode == "actual" {
		selectedRows = facts.BillingInput.ActualRows
		selectedLabel = "actual_rows"
	}
	verificationReceipt := cloneReceipt(inputReceipt)
	verificationReceipt["selected_mode"] = mode
	verificationReceipt["selected_rows"] = selectedRows
	if selectedRows > 0 {
		markPassed(closeRun, "verify_billing_inputs", facts.BillingInput.LastSyncedAt, &observedAt, verificationReceipt)
	} else if facts.BillingInput.TotalRows > 0 {
		markFailed(closeRun, "verify_billing_inputs", fmt.Sprintf("%s is 0; %s close input is not ready", selectedLabel, mode), verificationReceipt)
	}

	outputReceipt := map[string]any{
		"orders_table":      facts.MonthlyOutput.Orders.Name,
		"order_rows":        facts.MonthlyOutput.Orders.Rows,
		"collections_table": facts.MonthlyOutput.Collections.Name,
		"collection_rows":   facts.MonthlyOutput.Collections.Rows,
	}
	outputAt := latestTime(facts.MonthlyOutput.Orders.LastModified, facts.MonthlyOutput.Collections.LastModified)
	if selectedRows > 0 && facts.MonthlyOutput.Orders.Rows > 0 && facts.MonthlyOutput.Collections.Rows > 0 {
		markPassed(closeRun, "run_monthly_cloud_billing", outputAt, outputAt, outputReceipt)
	} else if facts.MonthlyOutput.Orders.Rows > 0 || facts.MonthlyOutput.Collections.Rows > 0 {
		markObserved(closeRun, "run_monthly_cloud_billing", outputAt, outputReceipt)
	}

	breakdownReceipt := map[string]any{
		"operation_receipt_id":  facts.Finance.BreakdownReceiptID,
		"order_total":           facts.Finance.Order.Total,
		"revrec_total":          facts.Finance.RevRec.Total,
		"collection_total":      facts.Finance.Collection.Total,
		"estimated_revrec_rows": facts.Finance.RevRec.EstimatedRows,
		"actual_revrec_rows":    facts.Finance.RevRec.ActualRows,
	}
	modeRows := facts.Finance.RevRec.EstimatedRows
	if mode == "actual" {
		modeRows = facts.Finance.RevRec.ActualRows
	}
	if facts.Finance.BreakdownSucceededAt != nil && facts.Finance.Order.Total > 0 && modeRows > 0 && facts.Finance.Collection.Total > 0 {
		markPassed(closeRun, "run_cloud_breakdown", facts.Finance.BreakdownSucceededAt, &observedAt, breakdownReceipt)
	} else if facts.Finance.Order.Total > 0 || facts.Finance.RevRec.Total > 0 || facts.Finance.Collection.Total > 0 {
		markObserved(closeRun, "run_cloud_breakdown", facts.Finance.BreakdownSucceededAt, breakdownReceipt)
	}

	orderReceipt := financeReceipt(facts.Finance.Order)
	if facts.Finance.Order.Total > 0 && facts.Finance.Order.Unsynced == 0 && facts.Finance.Order.IdentityRows >= facts.Finance.Order.RequiredIdentityRows {
		markPassed(closeRun, "sync_order_to_crm", facts.Finance.Order.LastSynced, facts.Finance.Order.LastSynced, orderReceipt)
	} else if facts.Finance.Order.Total > 0 {
		markObserved(closeRun, "sync_order_to_crm", facts.Finance.Order.LastSynced, orderReceipt)
	}

	crmReceipt := map[string]any{
		"order_run_id":     facts.CRM.Order.RunID,
		"order_rows":       facts.CRM.Order.Rows,
		"orderitem_run_id": facts.CRM.OrderItem.RunID,
		"orderitem_rows":   facts.CRM.OrderItem.Rows,
		"product2_run_id":  facts.CRM.Product2.RunID,
		"product2_rows":    facts.CRM.Product2.Rows,
		"scope":            []string{"nexuscrm_order", "nexuscrm_orderitem"},
	}
	crmAt := latestTime(facts.CRM.Order.ObservedAt, facts.CRM.OrderItem.ObservedAt)
	orderStep := closeRun.Steps["sync_order_to_crm"]
	if orderStep.Verification == VerificationPassed && facts.CRM.Order.RunID != "" && facts.CRM.Order.RunID == facts.CRM.OrderItem.RunID && afterOrEqual(crmAt, orderStep.LastVerifiedAt) {
		markPassed(closeRun, "refresh_crm_order_bundle", crmAt, crmAt, crmReceipt)
	} else if facts.CRM.Order.RunID != "" || facts.CRM.OrderItem.RunID != "" {
		markObserved(closeRun, "refresh_crm_order_bundle", crmAt, crmReceipt)
	}

	salesforceReceipt := map[string]any{
		"run_id":             facts.Salesforce.RunID,
		"status":             facts.Salesforce.Status,
		"source_rows":        facts.Salesforce.SourceRows,
		"target_rows":        facts.Salesforce.TargetRows,
		"duplicate_key_rows": facts.Salesforce.DuplicateKeyRows,
		"critical_null_rows": facts.Salesforce.CriticalNullRows,
	}
	crmStep := closeRun.Steps["refresh_crm_order_bundle"]
	if crmStep.Verification == VerificationPassed && (facts.Salesforce.Status == "success" || facts.Salesforce.Status == "warning") && afterOrEqual(facts.Salesforce.FinishedAt, crmStep.LastVerifiedAt) {
		markPassed(closeRun, "publish_salesforce_order", facts.Salesforce.FinishedAt, facts.Salesforce.FinishedAt, salesforceReceipt)
	} else if facts.Salesforce.RunID != "" {
		markObserved(closeRun, "publish_salesforce_order", facts.Salesforce.FinishedAt, salesforceReceipt)
	}

	revrecReceipt := financeReceipt(facts.Finance.RevRec)
	if facts.Finance.RevRec.Total > 0 && facts.Finance.RevRec.Unsynced == 0 {
		markPassed(closeRun, "finalize_revrec", facts.Finance.RevRec.LastSynced, facts.Finance.RevRec.LastSynced, revrecReceipt)
	} else if facts.Finance.RevRec.Total > 0 {
		markObserved(closeRun, "finalize_revrec", facts.Finance.RevRec.LastSynced, revrecReceipt)
	}
	collectionReceipt := financeReceipt(facts.Finance.Collection)
	if facts.Finance.Collection.Total > 0 && facts.Finance.Collection.Unsynced == 0 {
		markPassed(closeRun, "finalize_collection", facts.Finance.Collection.LastSynced, facts.Finance.Collection.LastSynced, collectionReceipt)
	} else if facts.Finance.Collection.Total > 0 {
		markObserved(closeRun, "finalize_collection", facts.Finance.Collection.LastSynced, collectionReceipt)
	}

	if closeRun.Steps["finalize_revrec"].Verification == VerificationPassed && closeRun.Steps["finalize_collection"].Verification == VerificationPassed {
		finalVerification := closeRun.Steps["verify_close_complete"]
		finalVerification.ObservedAt = &observedAt
		finalVerification.Receipt = map[string]any{
			"finance_tables_complete": true,
			"netsuite_readback":       "not_configured",
		}
	}

	closeRun.UpdatedAt = observedAt
	// Counts and timestamps are unbound observations, not accepted close receipts.
	for _, step := range closeRun.Steps {
		if step.Verification == VerificationPassed {
			step.Verification = VerificationNotChecked
			step.LastVerifiedAt = nil
			step.Receipt["binding_status"] = "unbound"
			step.Receipt["verification_gap"] = "exact input/output lineage and independent verifier are not configured"
		}
	}
	deriveClose(closeRun)
	closeRun.Status = "blocked"
	for _, step := range closeRun.Steps {
		if step.Action.Type == "inspect" {
			continue
		}
		step.Action.Enabled = false
		step.Action.Reason = "observe-only：尚未启用写凭证与人工授权"
	}

	assets := assetsForClose(closeRun)
	nextGate := "verify_close_complete"
	for _, definition := range MonthlyCloudClose {
		if closeRun.Steps[definition.Key].Verification != VerificationPassed {
			nextGate = definition.Key
			break
		}
	}
	dependencies := append([]DependencyObservation(nil), facts.Dependencies...)
	sort.Slice(dependencies, func(i, j int) bool { return dependencies[i].Key < dependencies[j].Key })
	gaps := make([]string, 0)
	for _, dependency := range dependencies {
		if !dependency.CredentialConfigured {
			gaps = append(gaps, dependency.Key)
		}
	}
	return &CloseObservation{
		ObservedAt:          observedAt,
		Close:               closeRun,
		Assets:              assets,
		Dependencies:        dependencies,
		NextGate:            nextGate,
		WriteActionsEnabled: false,
		WriteGate:           "Production 写操作需单独授权、writer identity、幂等 receipt 与 exact-scope 验证",
		CredentialGaps:      gaps,
	}
}

func markPassed(closeRun *CloseRun, key string, executedAt, verifiedAt *time.Time, receipt map[string]any) {
	step := closeRun.Steps[key]
	step.Execution = ExecutionSucceeded
	step.Verification = VerificationPassed
	step.Receipt = receipt
	step.LastSucceededAt = executedAt
	step.LastVerifiedAt = verifiedAt
	step.ObservedAt = verifiedAt
}

func markObserved(closeRun *CloseRun, key string, executedAt *time.Time, receipt map[string]any) {
	step := closeRun.Steps[key]
	step.Receipt = receipt
	step.ObservedAt = executedAt
	if executedAt != nil {
		step.Execution = ExecutionSucceeded
		step.LastSucceededAt = executedAt
	}
}

func markFailed(closeRun *CloseRun, key, message string, receipt map[string]any) {
	step := closeRun.Steps[key]
	step.Execution = ExecutionSucceeded
	step.Verification = VerificationFailed
	step.Error = message
	step.Receipt = receipt
	now := time.Now().UTC()
	step.ObservedAt = &now
}

func financeReceipt(fact FinanceTableFact) map[string]any {
	return map[string]any{
		"total_rows":             fact.Total,
		"synced_rows":            fact.Synced,
		"unsynced_rows":          fact.Unsynced,
		"identity_rows":          fact.IdentityRows,
		"required_identity_rows": fact.RequiredIdentityRows,
		"estimated_rows":         fact.EstimatedRows,
		"actual_rows":            fact.ActualRows,
		"locked_rows":            fact.LockedRows,
		"last_updated_at":        fact.LastUpdated,
		"last_synced_at":         fact.LastSynced,
	}
}

func assetsForClose(closeRun *CloseRun) []Asset {
	assets := make([]Asset, 0, len(MonthlyCloudClose))
	for _, definition := range MonthlyCloudClose {
		step := closeRun.Steps[definition.Key]
		status := "waiting"
		if step.Verification == VerificationPassed {
			status = "fresh"
		} else if step.Verification == VerificationFailed {
			status = "failed"
		}
		assets = append(assets, Asset{Key: definition.Key, Label: definition.Title, BusinessPeriod: closeRun.BillMonth, LastExecutionSucceeded: step.LastSucceededAt, LastOutputVerified: step.LastVerifiedAt, Status: status, Owner: definition.Owner, PhysicalScope: definition.Scope})
	}
	return assets
}

func cloneReceipt(source map[string]any) map[string]any {
	result := make(map[string]any, len(source)+2)
	for key, value := range source {
		result[key] = value
	}
	return result
}

func latestTime(values ...*time.Time) *time.Time {
	var latest *time.Time
	for _, value := range values {
		if value != nil && (latest == nil || value.After(*latest)) {
			copyValue := value.UTC()
			latest = &copyValue
		}
	}
	return latest
}

func afterOrEqual(value, boundary *time.Time) bool {
	return value != nil && boundary != nil && !value.Before(*boundary)
}