package service

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"strconv"
	"time"

	"github.com/spf13/viper"

	"finance-portal/pkg/errors"
	"finance-portal/pkg/log"
)

func dbtRun(ctx context.Context) error {
	log.L(ctx).Info("start dbtRun")

	baseUrl := viper.GetString("dbt.base_url")
	dbtToken := viper.GetString("dbt.token")
	accountID := viper.GetString("dbt.account_id")
	jobID := viper.GetString("dbt.job_id")
	jobTriggerURL := fmt.Sprintf("%s/%s/jobs/%s/run", baseUrl, accountID, jobID)

	body := map[string]interface{}{
		"cause": "Triggered via API",
	}

	// 将 body 转换为 JSON 字节
	b, _ := json.Marshal(body)
	// 创建请求
	req, err := http.NewRequest(http.MethodPost, jobTriggerURL, bytes.NewReader(b))
	if err != nil {
		log.L(ctx).Errorf("error creating request: %v", err)

		return err
	}

	// 设置请求头
	req.Header.Set("Authorization", "Token "+dbtToken)
	req.Header.Set("Content-Type", "application/json")

	// 发送请求

	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		log.L(ctx).Errorf("error sending request: %v", err)

		return err
	}

	// 检查响应状态码
	if resp.StatusCode != http.StatusOK {
		log.L(ctx).Errorf("unexpected status code: %v", resp.StatusCode)

		return errors.New("unexpected status code: " + strconv.Itoa(resp.StatusCode))
	}

	resMap := make(map[string]interface{})
	err = json.NewDecoder(resp.Body).Decode(&resMap)
	if err != nil {
		log.L(ctx).Errorf("error decoding response: %v", err)

		return err
	}

	runID := resMap["data"].(map[string]interface{})["id"].(float64)
	runIDStr := strconv.FormatFloat(runID, 'f', -1, 64)

	// 轮询查询状态
	runURL := fmt.Sprintf("%s/%s/runs/%s", baseUrl, accountID, runIDStr)
	req, err = http.NewRequest(http.MethodGet, runURL, nil)
	if err != nil {
		log.L(ctx).Errorf("error creating request: %v", err)

		return err
	}
	req.Header.Set("Authorization", "Token "+dbtToken)

	for {
		resp, err := http.DefaultClient.Do(req)
		if err != nil {
			log.L(ctx).Errorf("error sending request: %v", err)

			return err
		}

		// 检查响应状态码
		if resp.StatusCode != http.StatusOK {
			log.L(ctx).Errorf("unexpected status code: %v", resp.StatusCode)

			return errors.New("unexpected status code: " + strconv.Itoa(resp.StatusCode))
		}

		resMap = make(map[string]interface{})
		err = json.NewDecoder(resp.Body).Decode(&resMap)
		if err != nil {
			log.L(ctx).Errorf("error decoding response: %v", err)

			return err
		}

		state := resMap["data"].(map[string]interface{})["status_humanized"].(string)
		if state == "Success" {
			break
		}
		if state == "Error" {
			log.L(ctx).Errorf("dbt job failed:%v", runID)

			return errors.New("dbt job failed:" + runIDStr)
		}

		time.Sleep(40 * time.Second)
	}

	log.L(ctx).Info("end dbtRun")

	return nil
}