package service

import (
	"dddexample/normal/common"
	model2 "dddexample/normal/model"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/IBM/sarama"
	"io"
	"net/http"
)

func TransferOld(sourceUserId int, targetAccountNumber string, targetAmount float64, targetCurrency string) error {
	// 1. 从数据库读取数据，忽略所有校验逻辑如账号是否存在等,错误忽略
	var targetAccountDO, sourceAccountDO model2.Account
	model2.DB.Model(&model2.Account{}).Find(&sourceAccountDO, sourceUserId)
	model2.DB.Model(&model2.Account{}).Where(&model2.Account{TelNum: targetAccountNumber}).Find(&targetAccountDO)

	// 2. 业务参数校验
	if targetAccountDO.Currency != targetCurrency {
		return errors.New("收款人货币种类与转账使用货币不同")
	}

	// 3. 获取外部数据，并且包含一定的业务逻辑 获取汇率
	exchangeRate := 1.0
	if sourceAccountDO.Currency != targetCurrency {
		var err error

		res, err := http.Get("http://www.google.com?now=" + sourceAccountDO.Currency + "&targetCurrency=" + targetCurrency)

		if err != nil {
			return err
		}

		body, err := io.ReadAll(res.Body)
		res.Body.Close()

		if err != nil {
			return err
		}

		if res.StatusCode != 200 {
			return errors.New(fmt.Sprintf("Response failed with status code: %d and\nbody: %s\n", res.StatusCode, body))
		}

		var yahuRes common.YaHuRes
		err = json.Unmarshal(body, &yahuRes)

		if err != nil {
			return err
		}
		exchangeRate = yahuRes.Rate
	}

	sourceAmount := targetAmount * exchangeRate

	// 4. 业务参数校验
	if sourceAccountDO.Amount < sourceAmount {
		return errors.New("余额不足")
	}

	if sourceAccountDO.DailyLimit < sourceAmount {
		return errors.New("超出限额")
	}

	// 5. 计算新值，并且更新字段
	sourceAccountDO.Amount -= sourceAmount
	targetAccountDO.Amount += sourceAmount

	// 6. 更新到数据库
	model2.DB.Save(sourceAccountDO)
	model2.DB.Save(targetAccountDO)

	// 7. 发送审计消息String message = sourceUserId + "," + targetAccountNumber + "," + targetAmount + "," + targetCurrency;
	//common.SendKafka("kafka", "")
	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForAll
	config.Producer.Retry.Max = 5
	config.Producer.Return.Successes = true

	producer, err := sarama.NewSyncProducer([]string{"localhost:9092"}, config)
	if err != nil {
		fmt.Println("Failed to create producer:", err)
		return err
	}

	defer producer.Close()

	message := &sarama.ProducerMessage{
		Topic: "my_topic",
		Value: sarama.StringEncoder("Hello, Kafka!"),
	}
	_, _, err = producer.SendMessage(message)

	return err
}

func Transfer(sourceUserId int, targetAccountNumber string, targetAmount float64, targetCurrency string) error {
	// 1. 从数据库读取数据，忽略所有校验逻辑如账号是否存在等
	sourceAccountDO := model2.SelectByUserId(sourceUserId)
	targetAccountDO := model2.SelectByAccountNumber(targetAccountNumber)

	// 2. 业务参数校验
	if targetAccountDO.Currency != targetCurrency {
		return errors.New("收款人货币种类与转账使用货币不同")
	}

	// 3. 获取外部数据，并且包含一定的业务逻辑 获取汇率
	exchangeRate := 1.0
	if sourceAccountDO.Currency != targetCurrency {
		var err error
		exchangeRate, err = common.GetExchangeRate(sourceAccountDO.Currency, targetCurrency)
		if err != nil {
			return err
		}
	}

	sourceAmount := targetAmount * exchangeRate

	// 4. 业务参数校验
	if sourceAccountDO.Amount < sourceAmount {
		return errors.New("余额不足")
	}

	if sourceAccountDO.DailyLimit < sourceAmount {
		return errors.New("超出限额")
	}

	// 5. 计算新值，并且更新字段
	sourceAccountDO.Amount -= sourceAmount
	targetAccountDO.Amount += sourceAmount

	// 6. 更新到数据库
	model2.DB.Save(sourceAccountDO)
	model2.DB.Save(targetAccountDO)

	// 7. 发送审计消息String message = sourceUserId + "," + targetAccountNumber + "," + targetAmount + "," + targetCurrency;
	common.SendKafka("kafka", "")
	return nil
}