package control

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

// Account-specific Admin API URLs use the account's full cell hostname.
// https://docs.getdbt.com/docs/platform/about-platform/access-regions-ip-addresses
var dbtCellHost = regexp.MustCompile(`^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.(?:us[1235]|eu[1234]|au[12]|jp1)\.dbt\.com$`)

func validatedDBTAPIBase(base string) (string, error) {
	u, err := url.Parse(base)
	if err != nil || u.Scheme != "https" || u.User != nil || u.Port() != "" || u.RawQuery != "" || u.ForceQuery || u.Fragment != "" || u.RawPath != "" || (u.Path != "" && u.Path != "/") {
		return "", fmt.Errorf("invalid dbt API host")
	}
	host := strings.ToLower(u.Hostname())
	if host != "cloud.getdbt.com" && host != "emea.dbt.com" && host != "au.dbt.com" && !dbtCellHost.MatchString(host) {
		return "", fmt.Errorf("invalid dbt API host")
	}
	return "https://" + host, nil
}

// dbt's live Admin API also returns "2026-09-08 00:00:01+00:00".
// Require an explicit timezone; never interpret provider timestamps in local time.
type dbtTimestamp struct{ time.Time }

func (t *dbtTimestamp) UnmarshalJSON(data []byte) error {
	if string(data) == "null" {
		t.Time = time.Time{}
		return nil
	}
	var value string
	if err := json.Unmarshal(data, &value); err != nil {
		return fmt.Errorf("invalid dbt timestamp")
	}
	if len(value) > 10 && value[10] == ' ' {
		value = value[:10] + "T" + value[11:]
	}
	parsed, err := time.Parse(time.RFC3339Nano, value)
	if err != nil || parsed.IsZero() {
		return fmt.Errorf("invalid dbt timestamp")
	}
	t.Time = parsed.UTC()
	return nil
}

// List-run contract follows the dbt Cloud v2 API and its maintained integration:
// https://github.com/dagster-io/dagster/blob/master/python_modules/libraries/dagster-dbt/dagster_dbt/cloud/resources.py
// Only the explicitly configured job is observed; no account-wide dispatch.
func (s *ExternalLiveSource) dbtHistory(ctx context.Context, p CollectionState) (RunBatch, error) {
	account := os.Getenv("FINANCE_PORTAL_DBT_ACCOUNT_ID")
	job := os.Getenv("FINANCE_PORTAL_DBT_JOB_ID")
	token := os.Getenv("DBT_CLOUD_API_TOKEN")
	base := envOrValue("FINANCE_PORTAL_DBT_API_BASE", "https://cloud.getdbt.com")
	if account == "" || job == "" || token == "" {
		return RunBatch{State: CollectionState{Status: "unconfigured", Message: "需配置 dbt account、批准的财务 Job 与只读 API 身份"}}, nil
	}
	if n, err := strconv.ParseUint(account, 10, 64); err != nil || n == 0 {
		return RunBatch{}, fmt.Errorf("invalid dbt account")
	}
	if n, err := strconv.ParseUint(job, 10, 64); err != nil || n == 0 {
		return RunBatch{}, fmt.Errorf("invalid dbt job")
	}
	base, err := validatedDBTAPIBase(base)
	if err != nil {
		return RunBatch{}, err
	}
	offset := 0
	if p.Cursor != "" {
		offset, err = strconv.Atoi(p.Cursor)
		if err != nil || offset < 0 || offset > 5000 {
			return RunBatch{}, fmt.Errorf("invalid dbt checkpoint")
		}
	}
	q := url.Values{"job_definition_id": {job}, "order_by": {"-id"}, "limit": {"100"}, "offset": {strconv.Itoa(offset)}, "include_related": {"[]"}}
	req, err := http.NewRequestWithContext(ctx, "GET", strings.TrimRight(base, "/")+"/api/v2/accounts/"+account+"/runs/?"+q.Encode(), nil)
	if err != nil {
		return RunBatch{}, err
	}
	req.Header.Set("Authorization", "Token "+token)
	req.Header.Set("Accept", "application/json")
	response, err := s.httpClient.Do(req)
	if err != nil {
		return RunBatch{}, fmt.Errorf("dbt history transport failed")
	}
	defer response.Body.Close()
	if response.StatusCode != 200 {
		return RunBatch{}, fmt.Errorf("dbt history status %d", response.StatusCode)
	}
	var payload struct {
		Status struct {
			Success bool `json:"is_success"`
		} `json:"status"`
		Data []struct {
			ID       int64         `json:"id"`
			JobID    int64         `json:"job_definition_id"`
			Created  dbtTimestamp  `json:"created_at"`
			Started  *dbtTimestamp `json:"started_at"`
			Finished *dbtTimestamp `json:"finished_at"`
			Updated  dbtTimestamp  `json:"updated_at"`
			Status   int           `json:"status"`
			Human    string        `json:"status_humanized"`
		} `json:"data"`
	}
	if err = json.NewDecoder(io.LimitReader(response.Body, 4<<20)).Decode(&payload); err != nil {
		return RunBatch{}, fmt.Errorf("dbt history response invalid")
	}
	if !payload.Status.Success {
		return RunBatch{}, fmt.Errorf("dbt history unsuccessful")
	}
	from := time.Now().UTC().AddDate(0, 0, -30)
	b := RunBatch{State: CollectionState{Status: "partial", CoveredFrom: from, Message: "明确财务 Job 的近30天历史；每轮最多5100项，offset 分页可能有变化，产出与实际 lineage 未验收"}}
	more := len(payload.Data) == 100
	for _, item := range payload.Data {
		if item.ID == 0 || strconv.FormatInt(item.JobID, 10) != job || item.Created.IsZero() {
			return RunBatch{}, fmt.Errorf("dbt run identity or scope mismatch")
		}
		if item.Created.Before(from) {
			more = false
			continue
		}
		start := item.Created.Time
		message := "dbt 执行回执；tests/产出独立验收未绑定"
		if item.Started != nil {
			start = item.Started.Time
		} else {
			message = "使用创建时间排序；实际开始时间未知，产出验收未绑定"
		}
		event := item.Updated.Time
		if event.IsZero() {
			event = start
		}
		var finished *time.Time
		if item.Finished != nil {
			if item.Finished.Before(start) {
				return RunBatch{}, fmt.Errorf("invalid dbt run time order")
			}
			finished = &item.Finished.Time
			if item.Finished.After(event) {
				event = item.Finished.Time
			}
		}
		status := "unknown"
		switch item.Status {
		case 1, 2:
			status = "pending"
		case 3:
			status = "running"
		case 10:
			status = "succeeded"
		case 20:
			status = "failed"
		case 30:
			status = "cancelled"
		}
		b.Runs = append(b.Runs, ProviderRun{Provider: "dbt", Task: "dbt_billing", ExternalID: account + ":" + strconv.FormatInt(item.ID, 10), Status: status, RawStatus: strconv.Itoa(item.Status), Verification: "not_checked", StartedAt: start, FinishedAt: finished, EventAt: event, Message: message})
	}
	if more && offset < 5000 {
		b.State.Cursor = strconv.Itoa(offset + 100)
	}
	return b, nil
}