package control

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"log/slog"
	"strconv"
	"time"

	"cloud.google.com/go/bigquery"
	"google.golang.org/api/iterator"
)

type HistorySource interface {
	Name() string
	Collect(context.Context, CollectionState) (RunBatch, error)
}
type historySource struct {
	name    string
	collect func(context.Context, CollectionState) (RunBatch, error)
}

func (s historySource) Name() string { return s.name }
func (s historySource) Collect(c context.Context, p CollectionState) (RunBatch, error) {
	return s.collect(c, p)
}

func (o *Operations) StartCollectors(ctx context.Context, sources []HistorySource, interval time.Duration) {
	if interval < time.Second {
		interval = time.Minute
	}
	for _, source := range sources {
		source := source
		go func() {
			tick := time.NewTicker(interval)
			defer tick.Stop()
			backfillPages := 0
			for {
				previous := o.Checkpoint(ctx, source.Name())
				call, cancel := context.WithTimeout(ctx, 40*time.Second)
				batch, err := source.Collect(call, previous)
				cancel()
				if ctx.Err() != nil {
					return
				}
				if err != nil {
					batch = RunBatch{State: previous}
					batch.State.Status = "unavailable"
					batch.State.Message = "采集失败；保留最后可信数据，检查 provider 权限与审计读路径"
				}
				batch.State.Provider = source.Name()
				batch.State.ObservedAt = time.Now().UTC()
				if err = o.CommitBatch(ctx, batch); err != nil {
					slog.Error("operations_commit_failed", "provider", source.Name(), "error_type", fmt.Sprintf("%T", err))
				}
				if err == nil && source.Name() == "crm_load" && batch.State.Status == "partial" && batch.State.Cursor != "" && batch.State.Cursor != previous.Cursor && backfillPages < 4 {
					backfillPages++
					continue
				}
				backfillPages = 0
				select {
				case <-ctx.Done():
					return
				case <-tick.C:
				}
			}
		}()
	}
}

func normalizedRunStatus(raw string) string {
	switch raw {
	case "success", "warning", "succeeded", "completed", "loaded":
		return "succeeded"
	case "failed", "blocked", "error":
		return "failed"
	case "running", "started":
		return "running"
	case "cancelled", "canceled":
		return "cancelled"
	default:
		return "unknown"
	}
}

func (s *ExternalLiveSource) HistorySources() []HistorySource {
	return []HistorySource{
		historySource{"finance", s.financeHistory}, historySource{"crm_load", s.crmHistory}, historySource{"finance_scheduler", s.schedulerHistory},
		historySource{"finance_scheduler_logs", s.schedulerLogsHistory},
		historySource{"crm_extract", func(ctx context.Context, p CollectionState) (RunBatch, error) {
			return s.crmJobHistory(ctx, p, "crm_extract")
		}},
		historySource{"crm_load_job", func(ctx context.Context, p CollectionState) (RunBatch, error) {
			return s.crmJobHistory(ctx, p, "crm_load_job")
		}},
		historySource{"fivetran", s.fivetranHistory},
		historySource{"dbt", s.dbtHistory},
		historySource{"finance_runtime", func(context.Context, CollectionState) (RunBatch, error) {
			return RunBatch{State: CollectionState{Status: "unconfigured", Message: "历史业务任务审计适配器尚未绑定"}}, nil
		}},
	}
}

// Scan a bounded audit window by a stable keyset; when reaching its end restart
// the overlapping window to reconcile running rows and late terminal updates.
func historyWindow(p CollectionState) (time.Time, time.Time, string) {
	from := p.CoveredFrom
	windowStart := time.Now().UTC().AddDate(0, 0, -30)
	if from.IsZero() || from.Before(windowStart) {
		from = windowStart
	}
	var cursor struct {
		At  time.Time
		Key string
	}
	if p.Cursor != "" {
		_ = decodeCursor(p.Cursor, &cursor)
	}
	return from, cursor.At, cursor.Key
}
func encodeCursor(at time.Time, key string) string {
	b, _ := json.Marshal(struct {
		At  time.Time
		Key string
	}{at, key})
	return string(b)
}
func decodeCursor(value string, destination any) error {
	return json.Unmarshal([]byte(value), destination)
}
func (s *ExternalLiveSource) financeHistory(ctx context.Context, p CollectionState) (RunBatch, error) {
	if s.config.FinanceDatabaseURL == "" {
		if s.config.FinanceHistoryAPIURL != "" {
			return s.financeAPIHistory(ctx)
		}
		return RunBatch{State: CollectionState{Status: "unconfigured", Message: "缺少 Finance 审计只读连接配置；未采集到运行，不代表任务执行失败"}}, nil
	}
	if s.finance == nil {
		return RunBatch{}, fmt.Errorf("finance read identity unavailable")
	}
	from, after, key := historyWindow(p)
	if after.IsZero() {
		after = from
	}
	rows, err := s.finance.QueryContext(ctx, `SELECT run_id,target_table,status,started_at,finished_at FROM data_platform_load_table_audit
WHERE target_table IN ('salesforce_org','salesforce_account','salesforce_order') AND started_at>=? AND (started_at>? OR (started_at=? AND CONCAT(run_id,':',target_table)>?)) ORDER BY started_at,run_id,target_table LIMIT 200`, from, after, after, key)
	if err != nil {
		return RunBatch{}, err
	}
	defer rows.Close()
	b := RunBatch{State: CollectionState{Status: "partial", CoveredFrom: from, Message: "近30天起始窗口的逐表发布审计；不含未生成审计的触发失败；未绑定 close"}}
	for rows.Next() {
		var run ProviderRun
		var end sql.NullTime
		if err = rows.Scan(&run.ExternalID, &run.Task, &run.RawStatus, &run.StartedAt, &end); err != nil {
			return RunBatch{}, err
		}
		run.Provider = "finance"
		run.Status = normalizedRunStatus(run.RawStatus)
		run.EventAt = run.StartedAt
		if end.Valid {
			run.FinishedAt = &end.Time
			run.EventAt = end.Time
		}
		run.Verification = "not_checked"
		run.TimeBasis = "audit_recorded_at"
		run.Message = "owner 审计状态；独立业务读回未执行"
		run.OutputRefs = []string{run.Task}
		b.Runs = append(b.Runs, run)
		b.State.Cursor = encodeCursor(run.StartedAt, run.ExternalID+":"+run.Task)
	}
	if err = rows.Err(); err != nil {
		return RunBatch{}, err
	}
	if len(b.Runs) < 200 {
		b.State.Cursor = ""
	}
	return b, nil
}

func (s *ExternalLiveSource) crmHistory(ctx context.Context, p CollectionState) (RunBatch, error) {
	p = crmScopeCheckpoint(p)
	if s.bq == nil {
		return RunBatch{}, fmt.Errorf("bigquery identity unavailable")
	}
	from, after, key := historyWindow(p)
	if after.IsZero() {
		after = from
	}
	if !qualifiedTablePattern.MatchString(s.config.CRMControlDataset + ".table_loads") {
		return RunBatch{}, fmt.Errorf("invalid CRM audit binding")
	}
	q := s.bq.Query(fmt.Sprintf(`SELECT t.run_id,t.source_table,t.final_table,t.status,t.created_at,p.manifest_uri,p.started_at AS load_started_at
FROM %s t LEFT JOIN (
 SELECT source_name,run_id,ANY_VALUE(manifest_uri) AS manifest_uri,ANY_VALUE(started_at) AS started_at
 FROM %s WHERE source_name='nexuscrm' GROUP BY source_name,run_id HAVING COUNT(*)=1
) p ON t.source_name=p.source_name AND t.run_id=p.run_id
WHERE t.source_name='nexuscrm' AND t.final_table IN UNNEST(@finance_tables)
AND t.created_at>=@from AND (t.created_at>@after OR (t.created_at=@after AND CONCAT(t.run_id,':',t.final_table)>@key))
ORDER BY t.created_at,t.run_id,t.final_table LIMIT 20`, quoteBQTable(s.config.CRMControlDataset+".table_loads"), quoteBQTable(s.config.CRMControlDataset+".pipeline_runs")))
	q.MaxBytesBilled = 1 << 30
	q.DisableQueryCache = true
	q.Parameters = []bigquery.QueryParameter{{Name: "from", Value: from}, {Name: "after", Value: after}, {Name: "key", Value: key}, {Name: "finance_tables", Value: financeCRMTables(false)}}
	it, err := q.Read(ctx)
	if err != nil {
		return RunBatch{}, err
	}
	b := RunBatch{State: CollectionState{ScopeVersion: financeCRMScopeVersion, Status: "partial", CoveredFrom: from, Message: "财务前置任务所需9张CRM表的加载审计；不采集无关表，不等于dbt全部依赖或财务验收"}}
	type auditRow struct {
		ID          string                 `bigquery:"run_id"`
		Table       string                 `bigquery:"final_table"`
		Status      string                 `bigquery:"status"`
		At          time.Time              `bigquery:"created_at"`
		SourceTable string                 `bigquery:"source_table"`
		ManifestURI bigquery.NullString    `bigquery:"manifest_uri"`
		LoadStarted bigquery.NullTimestamp `bigquery:"load_started_at"`
	}
	rows := []auditRow{}
	requests := map[string]string{}
	for {
		var row auditRow
		err = it.Next(&row)
		if err == iterator.Done {
			break
		}
		if err != nil {
			return RunBatch{}, err
		}
		rows = append(rows, row)
		requests[row.ManifestURI.StringVal] = row.ID
	}
	artifacts, err := fetchManifestArtifacts(ctx, requests, s.readCRMManifest)
	if err != nil {
		return RunBatch{}, err
	}
	for _, row := range rows {
		artifact := artifacts[row.ManifestURI.StringVal]
		evidence := evaluateManifest(row.ManifestURI.StringVal, row.ID, row.SourceTable, row.LoadStarted.Timestamp, row.At, time.Now().UTC(), artifact)
		b.Runs = append(b.Runs, ProviderRun{Provider: "crm_load", Task: "crm_load", ExternalID: row.ID + ":" + row.Table, RawStatus: row.Status, Status: normalizedRunStatus(row.Status), StartedAt: row.At, EventAt: row.At, Verification: "not_checked", OutputRefs: []string{row.Table}, Manifest: &evidence, TimeBasis: "audit_recorded_at", Message: "加载审计时间；真实逐表执行起止时间未知；manifest 关联不等于目标读回"})
		b.State.Cursor = encodeCursor(row.At, row.ID+":"+row.Table)
	}
	if len(b.Runs) < 20 {
		b.State.Cursor = ""
	}
	return b, nil
}

func (s *ExternalLiveSource) schedulerHistory(ctx context.Context, p CollectionState) (RunBatch, error) {
	fact, dep := s.readScheduler(ctx)
	if dep.Status == "unavailable" {
		return RunBatch{}, fmt.Errorf("scheduler unavailable")
	}
	now := time.Now().UTC()
	b := RunBatch{State: CollectionState{Status: "partial", CoveredFrom: p.CoveredFrom, Message: "仅最近调度尝试；轮询间隔内可能遗漏，不能证明完整触发历史"}}
	if b.State.CoveredFrom.IsZero() {
		b.State.CoveredFrom = now
	}
	b.Schedules = []ScheduleObservation{{ID: s.config.SchedulerJobName, Provider: "finance_scheduler", State: fact.State, Expression: fact.Schedule, Timezone: fact.TimeZone, NextAt: fact.NextRunAt, ObservedAt: now, Coverage: b.State.Message}}
	if fact.LastAttemptAt != nil {
		status := "accepted"
		if fact.LastStatusCode != 0 {
			status = "failed"
		}
		b.Triggers = []TriggerObservation{{ID: stableID(s.config.SchedulerJobName, stamp(*fact.LastAttemptAt)), ScheduleID: s.config.SchedulerJobName, At: *fact.LastAttemptAt, Status: status, ObservedAt: now}}
		b.State.Cursor = strconv.FormatInt(fact.LastAttemptAt.Unix(), 10)
	}
	return b, nil
}