package message

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"net/http/httptest"
	"sync"

	"github.com/gin-gonic/gin"
	. "github.com/onsi/ginkgo/v2"
	. "github.com/onsi/gomega"
	devopsapi "github.com/tidbcloud/devops/api"
	msgapi "github.com/tidbcloud/devops/api/message"
	msgent "github.com/tidbcloud/devops/entity/message"
	"github.com/tidbcloud/devops/pkg/infra/consts"
	msgsvcapi "github.com/tidbcloud/devops/pkg/service/message/api"
)

var _ = Describe("SendEmailBatch", func() {
	var (
		controller *MessageController
		messageSvc *fakeMessageService
	)

	BeforeEach(func() {
		gin.SetMode(gin.TestMode)
		messageSvc = &fakeMessageService{}
		controller = &MessageController{Message: messageSvc}
	})

	sendRequest := func(req *msgapi.SendEmailBatchRequest) *httptest.ResponseRecorder {
		body, err := json.Marshal(req)
		Expect(err).NotTo(HaveOccurred())

		recorder := httptest.NewRecorder()
		ctx, _ := gin.CreateTestContext(recorder)
		ctx.Request = httptest.NewRequest(http.MethodPost, "/message/email/send_batch", bytes.NewReader(body))
		ctx.Request.Header.Set("Content-Type", "application/json")
		ctx.Set(consts.CtxUserEmail, "applicant@pingcap.com")
		controller.SendEmailBatch(ctx)
		return recorder
	}

	It("expands tenant and user targets and sends them in one bulk call", func() {
		req := &msgapi.SendEmailBatchRequest{
			Name: "batch-name",
			TargetTenants: &msgapi.SendEmailBatchTenant{
				TenantIDs: []string{"tenant-a", "tenant-b"},
				Role:      "Owner",
			},
			TargetUsers: &msgapi.SendEmailBatchUsers{
				Emails: []string{"first@example.com", "second@example.com"},
			},
			Template: msgent.EmailTemplate{
				ID:        "template-id",
				Variables: map[string]string{"subject": "test"},
			},
			Reason:      "test reason",
			FromAddress: "no-reply@tidbcloud.com",
		}
		messageSvc.returnAllTargets = true

		recorder := sendRequest(req)

		Expect(recorder.Code).To(Equal(http.StatusOK))
		Expect(messageSvc.calls()).To(Equal(1))
		params := messageSvc.params()
		Expect(params[0]).To(Equal(msgsvcapi.SendBulkEmailsParam{
			Name: "batch-name",
			Targets: []msgent.EmailTarget{
				{Tenant: &msgent.EmailTargetTenant{TenantID: "tenant-a", Role: "Owner"}},
				{Tenant: &msgent.EmailTargetTenant{TenantID: "tenant-b", Role: "Owner"}},
				{Users: &msgent.EmailTargetUsers{Emails: []string{"first@example.com", "second@example.com"}}},
			},
			Template: msgent.EmailTemplate{
				ID:        "template-id",
				Variables: map[string]string{"subject": "test"},
			},
			Reason:      "test reason",
			Applicant:   "applicant@pingcap.com",
			FromAddress: "no-reply@tidbcloud.com",
		}))

		resp := &msgapi.SendEmailBatchResponse{}
		Expect(json.Unmarshal(recorder.Body.Bytes(), resp)).To(Succeed())
		Expect(resp.ErrCode).To(Equal(devopsapi.ErrCodeOK))
		Expect(resp.SuccessTargets).To(Equal(params[0].Targets))
		Expect(resp.FailedTargets).To(BeEmpty())
	})

	It("splits targets by the requested batch size", func() {
		tenantIDs := []string{"tenant-a", "tenant-b", "tenant-c", "tenant-d", "tenant-e"}
		messageSvc.returnAllTargets = true

		recorder := sendRequest(&msgapi.SendEmailBatchRequest{
			TargetTenants: &msgapi.SendEmailBatchTenant{TenantIDs: tenantIDs},
			Concurrency:   1,
			PerBatch:      2,
		})

		Expect(recorder.Code).To(Equal(http.StatusOK))
		params := messageSvc.params()
		Expect(params).To(HaveLen(3))
		Expect(params[0].Targets).To(HaveLen(2))
		Expect(params[1].Targets).To(HaveLen(2))
		Expect(params[2].Targets).To(HaveLen(1))
	})

	DescribeTable("uses 100 as the default batch size",
		func(perBatch int) {
			tenantIDs := make([]string, 101)
			for i := range tenantIDs {
				tenantIDs[i] = fmt.Sprintf("tenant-%d", i)
			}
			messageSvc.returnAllTargets = true

			recorder := sendRequest(&msgapi.SendEmailBatchRequest{
				TargetTenants: &msgapi.SendEmailBatchTenant{TenantIDs: tenantIDs},
				Concurrency:   1,
				PerBatch:      perBatch,
			})

			Expect(recorder.Code).To(Equal(http.StatusOK))
			params := messageSvc.params()
			Expect(params).To(HaveLen(2))
			Expect(params[0].Targets).To(HaveLen(defaultSendEmailBatchPerBatch))
			Expect(params[1].Targets).To(HaveLen(1))
		},
		Entry("when omitted", 0),
		Entry("when negative", -1),
	)

	It("runs batches concurrently up to the requested concurrency", func() {
		started := make(chan struct{}, 3)
		release := make(chan struct{})
		var stateMu sync.Mutex
		inFlight := 0
		maxInFlight := 0
		messageSvc.bulkFn = func(
			_ context.Context,
			param msgsvcapi.SendBulkEmailsParam,
		) ([]msgent.EmailTarget, []msgent.EmailTarget, error) {
			stateMu.Lock()
			inFlight++
			maxInFlight = max(maxInFlight, inFlight)
			stateMu.Unlock()
			started <- struct{}{}
			<-release
			stateMu.Lock()
			inFlight--
			stateMu.Unlock()
			return param.Targets, nil, nil
		}

		result := make(chan *httptest.ResponseRecorder, 1)
		go func() {
			defer GinkgoRecover()
			result <- sendRequest(&msgapi.SendEmailBatchRequest{
				TargetTenants: &msgapi.SendEmailBatchTenant{
					TenantIDs: []string{"tenant-a", "tenant-b", "tenant-c"},
				},
				Concurrency: 2,
				PerBatch:    1,
			})
		}()

		Eventually(started).Should(Receive())
		Eventually(started).Should(Receive())
		Consistently(started).ShouldNot(Receive())
		close(release)
		var recorder *httptest.ResponseRecorder
		Eventually(result).Should(Receive(&recorder))

		Expect(recorder.Code).To(Equal(http.StatusOK))
		stateMu.Lock()
		Expect(maxInFlight).To(Equal(2))
		stateMu.Unlock()
	})

	It("uses one worker when concurrency is not positive", func() {
		started := make(chan struct{}, 2)
		release := make(chan struct{}, 2)
		messageSvc.bulkFn = func(
			_ context.Context,
			param msgsvcapi.SendBulkEmailsParam,
		) ([]msgent.EmailTarget, []msgent.EmailTarget, error) {
			started <- struct{}{}
			<-release
			return param.Targets, nil, nil
		}

		result := make(chan *httptest.ResponseRecorder, 1)
		go func() {
			defer GinkgoRecover()
			result <- sendRequest(&msgapi.SendEmailBatchRequest{
				TargetTenants: &msgapi.SendEmailBatchTenant{
					TenantIDs: []string{"tenant-a", "tenant-b"},
				},
				Concurrency: 0,
				PerBatch:    1,
			})
		}()

		Eventually(started).Should(Receive())
		Consistently(started).ShouldNot(Receive())
		release <- struct{}{}
		Eventually(started).Should(Receive())
		release <- struct{}{}
		var recorder *httptest.ResponseRecorder
		Eventually(result).Should(Receive(&recorder))
		Expect(recorder.Code).To(Equal(http.StatusOK))
	})

	It("returns the success and failed targets from the service", func() {
		req := &msgapi.SendEmailBatchRequest{
			TargetTenants: &msgapi.SendEmailBatchTenant{
				TenantIDs: []string{"tenant-a", "tenant-a", "tenant-b"},
				Role:      "Owner",
			},
			TargetUsers: &msgapi.SendEmailBatchUsers{},
		}
		messageSvc.successTargets = []msgent.EmailTarget{
			{Tenant: &msgent.EmailTargetTenant{TenantID: "tenant-a", Role: "Owner"}},
			{Tenant: &msgent.EmailTargetTenant{TenantID: "tenant-b", Role: "Owner"}},
		}
		messageSvc.failedTargets = []msgent.EmailTarget{
			{Tenant: &msgent.EmailTargetTenant{TenantID: "tenant-a", Role: "Owner"}},
			{Users: &msgent.EmailTargetUsers{}},
		}

		recorder := sendRequest(req)

		Expect(recorder.Code).To(Equal(http.StatusOK))
		Expect(messageSvc.calls()).To(Equal(1))
		resp := &msgapi.SendEmailBatchResponse{}
		Expect(json.Unmarshal(recorder.Body.Bytes(), resp)).To(Succeed())
		Expect(resp.ErrCode).To(Equal(devopsapi.ErrCodeOK))
		Expect(resp.SuccessTargets).To(Equal(messageSvc.successTargets))
		Expect(resp.FailedTargets).To(Equal(messageSvc.failedTargets))
	})

	It("marks a failed batch as failed targets and continues other batches", func() {
		messageSvc.bulkFn = func(
			_ context.Context,
			param msgsvcapi.SendBulkEmailsParam,
		) ([]msgent.EmailTarget, []msgent.EmailTarget, error) {
			if param.Targets[0].Tenant.TenantID == "tenant-a" {
				return nil, nil, errors.New("resolve targets failed")
			}
			return param.Targets, nil, nil
		}
		req := &msgapi.SendEmailBatchRequest{
			TargetTenants: &msgapi.SendEmailBatchTenant{
				TenantIDs: []string{"tenant-a", "tenant-b"},
			},
			Concurrency: 1,
			PerBatch:    1,
		}

		recorder := sendRequest(req)

		Expect(recorder.Code).To(Equal(http.StatusOK))
		Expect(messageSvc.calls()).To(Equal(2))
		resp := &msgapi.SendEmailBatchResponse{}
		Expect(json.Unmarshal(recorder.Body.Bytes(), resp)).To(Succeed())
		Expect(resp.ErrCode).To(Equal(devopsapi.ErrCodeOK))
		Expect(resp.SuccessTargets).To(Equal([]msgent.EmailTarget{
			{Tenant: &msgent.EmailTargetTenant{TenantID: "tenant-b"}},
		}))
		Expect(resp.FailedTargets).To(Equal([]msgent.EmailTarget{
			{Tenant: &msgent.EmailTargetTenant{TenantID: "tenant-a"}},
		}))
	})
})

type fakeMessageService struct {
	mu               sync.Mutex
	bulkParams       []msgsvcapi.SendBulkEmailsParam
	successTargets   []msgent.EmailTarget
	failedTargets    []msgent.EmailTarget
	bulkErr          error
	bulkCalls        int
	returnAllTargets bool
	bulkFn           func(context.Context, msgsvcapi.SendBulkEmailsParam) ([]msgent.EmailTarget, []msgent.EmailTarget, error)
}

var _ msgsvcapi.MessageServiceAPI = (*fakeMessageService)(nil)

func (f *fakeMessageService) SendBulkEmails(
	ctx context.Context,
	param msgsvcapi.SendBulkEmailsParam,
) ([]msgent.EmailTarget, []msgent.EmailTarget, error) {
	f.mu.Lock()
	f.bulkCalls++
	f.bulkParams = append(f.bulkParams, param)
	bulkFn := f.bulkFn
	returnAllTargets := f.returnAllTargets
	successTargets := f.successTargets
	failedTargets := f.failedTargets
	bulkErr := f.bulkErr
	f.mu.Unlock()

	if bulkFn != nil {
		return bulkFn(ctx, param)
	}
	if returnAllTargets {
		return param.Targets, nil, bulkErr
	}
	return successTargets, failedTargets, bulkErr
}

func (f *fakeMessageService) calls() int {
	f.mu.Lock()
	defer f.mu.Unlock()
	return f.bulkCalls
}

func (f *fakeMessageService) params() []msgsvcapi.SendBulkEmailsParam {
	f.mu.Lock()
	defer f.mu.Unlock()
	params := make([]msgsvcapi.SendBulkEmailsParam, len(f.bulkParams))
	copy(params, f.bulkParams)
	return params
}

func (f *fakeMessageService) SendEmail(context.Context, msgsvcapi.SendEmailParam) error {
	return nil
}

func (f *fakeMessageService) ListEmailRecords(
	context.Context,
	msgsvcapi.ListEmailRecordsParam,
) ([]*msgent.EmailRecord, int64, error) {
	return nil, 0, nil
}

func (f *fakeMessageService) GetEmailRecord(context.Context, uint64) (*msgent.EmailRecord, error) {
	return nil, nil
}