package control

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"strings"
	"time"
)

// Contract: https://fivetran.com/docs/developer-resources/rest-api/api-reference/connections/retrieve-connection-sync-history
// This GET is sync history, never the similarly named historical re-sync command.
func (s *ExternalLiveSource) fivetranHistory(ctx context.Context, p CollectionState) (RunBatch, error) {
	if s.config.FivetranAPIKey == "" || s.config.FivetranAPISecret == "" {
		return RunBatch{State: CollectionState{Status: "unconfigured", Message: "缺少 Fivetran 只读 API 身份"}}, nil
	}
	now := time.Now().UTC().Truncate(time.Second)
	var cursor struct {
		Start time.Time
		End   time.Time
		Next  string
	}
	if p.Cursor != "" {
		if err := json.Unmarshal([]byte(p.Cursor), &cursor); err != nil {
			return RunBatch{}, fmt.Errorf("invalid history checkpoint")
		}
	}
	// A six-day window leaves room for pagination without crossing the seven-day
	// API retention limit while a cursor is in flight.
	if cursor.Next == "" || cursor.Start.Before(now.Add(-7*24*time.Hour+time.Hour)) {
		cursor.Start = now.Add(-6 * 24 * time.Hour)
		cursor.End = now
		cursor.Next = ""
	}
	query := url.Values{"start_time": {cursor.Start.Format(time.RFC3339)}, "end_time": {cursor.End.Format(time.RFC3339)}, "limit": {"200"}}
	if cursor.Next != "" {
		query.Set("cursor", cursor.Next)
	}
	req, err := http.NewRequestWithContext(ctx, "GET", "https://api.fivetran.com/v1/connections/"+url.PathEscape(s.config.FivetranConnectorID)+"/sync-history?"+query.Encode(), nil)
	if err != nil {
		return RunBatch{}, err
	}
	req.SetBasicAuth(s.config.FivetranAPIKey, s.config.FivetranAPISecret)
	req.Header.Set("Accept", "application/json;version=2")
	response, err := s.httpClient.Do(req)
	if err != nil {
		return RunBatch{}, fmt.Errorf("fivetran history transport failed")
	}
	defer response.Body.Close()
	if response.StatusCode != http.StatusOK {
		return RunBatch{}, fmt.Errorf("fivetran history status %d", response.StatusCode)
	}
	var payload struct {
		Code string `json:"code"`
		Data struct {
			Items []struct {
				ID     string     `json:"sync_id"`
				Start  *time.Time `json:"start"`
				End    *time.Time `json:"end"`
				Status string     `json:"status"`
			} `json:"items"`
			Next string `json:"next_cursor"`
		} `json:"data"`
	}
	if err = json.NewDecoder(io.LimitReader(response.Body, 4<<20)).Decode(&payload); err != nil {
		return RunBatch{}, fmt.Errorf("fivetran history response invalid")
	}
	if payload.Code != "Success" {
		return RunBatch{}, fmt.Errorf("fivetran history not successful")
	}
	b := RunBatch{State: CollectionState{Status: "partial", CoveredFrom: cursor.Start, Message: "六天分页扫描窗口；历史更早或未开始同步不保证覆盖；产出未经独立验证"}}
	for _, item := range payload.Data.Items {
		if item.ID == "" {
			return RunBatch{}, fmt.Errorf("sync history missing stable identity")
		}
		if item.Start == nil {
			b.State.Message = "包含未开始/无开始时间的同步；运行投影暂不收录，历史覆盖不完整"
			continue
		}
		status := "unknown"
		switch strings.ToUpper(item.Status) {
		case "COMPLETED":
			status = "succeeded"
		case "FAILURE":
			status = "failed"
		case "CANCELED":
			status = "cancelled"
		case "INCOMPLETE":
			if item.End == nil {
				status = "running"
			}
		}
		event := *item.Start
		if item.End != nil {
			event = *item.End
		}
		b.Runs = append(b.Runs, ProviderRun{Provider: "fivetran", Task: "ar_sync", ExternalID: item.ID, Status: status, RawStatus: item.Status, Verification: "not_checked", StartedAt: *item.Start, FinishedAt: item.End, EventAt: event, Message: "Fivetran 同步回执；目标月份数据落地需独立验证"})
	}
	cursor.Next = payload.Data.Next
	if cursor.Next != "" {
		encoded, _ := json.Marshal(cursor)
		b.State.Cursor = string(encoded)
	}
	return b, nil
}