package data

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/google/uuid"
	"sort"
	"strings"
	"sync"
	"time"

	"github.com/gogf/gf/v2/frame/g"
	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/consts"
	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/model"
)

// 首查 +3min，此后每 3min 重试 3 次（+3/+6/+9/+12min）。
// 2026-10-15 潍坊、濮阳实测：10/30/100 卡共12笔首次可见终态约 65~225s；
// 15s 采样只能给出完成区间，不能据此承诺所有高峰时段的完成上限。
// 偶发超长任务不增加自动重试；超窗记 review，保留受理号和原因供手动查询。
// 未确认不得记办理失败，也不得重新提交办理请求。
const (
	subBatchLease      = 5 * time.Minute
	firstResultDelay   = 3 * time.Minute
	furtherResultDelay = 3 * time.Minute
	maxResultQueries   = 4
)

func newSubBatch(task *Task, details []*Detail) SubBatch {
	keys := make([]*model.SimKey, 0, len(details))
	ids := make([]int64, 0, len(details))
	for _, detail := range details {
		ids = append(ids, detail.ID)
		keys = append(keys, &model.SimKey{ChanKey: detail.ChanKey, ICCID: detail.ICCID, MSISDN: detail.MSISDN, IMSI: detail.IMSI})
	}
	raw, _ := json.Marshal(struct {
		BatchNo   string
		BatchType int
		IDs       []int64
	}{task.Batch.BatchNo, task.Batch.BatchType, ids})
	digest := sha256.Sum256(raw)
	id := hex.EncodeToString(digest[:])
	input := model.SimCardApnChangeInput{SimKeys: keys, ReqData: model.SimCardApnChangeInputItem{OperType: task.Operation.ApnOperType, ApnName: task.ApnName, OperationID: id, SubmitOnly: true}}
	return SubBatch{BatchNo: task.Batch.BatchNo, BatchType: task.Batch.BatchType, ChanKey: details[0].ChanKey, OperationID: id, DetailIDs: ids, Request: input, Status: "prepared"}
}

func splitAsyncDetails(details []*Detail) [][]*Detail {
	withMSISDN := make([]*Detail, 0, len(details))
	batches := make([][]*Detail, 0)
	for _, detail := range details {
		if strings.TrimSpace(detail.MSISDN) == "" {
			batches = append(batches, []*Detail{detail})
		} else {
			withMSISDN = append(withMSISDN, detail)
		}
	}
	for _, chunk := range splitDetails(withMSISDN, consts.BatchSubBatchSize) {
		if len(chunk) < 10 {
			for _, detail := range chunk {
				batches = append(batches, []*Detail{detail})
			}
		} else {
			batches = append(batches, chunk)
		}
	}
	return batches
}

// prepareAndSubmit persists every intent before a single write leaves the process.
func (e *executor) prepareAndSubmit(ctx context.Context, task *Task, repo SubBatchRepository, upstream AsyncUpstream) error {
	groups := make(map[string][]*Detail)
	for _, detail := range task.Details {
		if detail != nil && detail.Reason == "" && (detail.DetailStatus == 0 || detail.DetailStatus == consts.BatchDetailStatusRunning) {
			groups[detail.ChanKey] = append(groups[detail.ChanKey], detail)
		}
	}
	keys := make([]string, 0, len(groups))
	for key := range groups {
		keys = append(keys, key)
	}
	sort.Strings(keys)
	for _, key := range keys {
		for _, slice := range splitAsyncDetails(groups[key]) {
			sub := newSubBatch(task, slice)
			if err := repo.PrepareSubBatch(ctx, sub); err != nil {
				return fmt.Errorf("保存APN提交意图失败（未提交运营商）: %w", err)
			}
		}
	}
	return nil
}

// processSubBatch only submits while prepared. Any interrupted submission is recovered by operationId, never repeated.
func (e *executor) processSubBatch(ctx context.Context, repo SubBatchRepository, upstream AsyncUpstream, operationID string) error {
	channel, err := repo.SubBatchChannel(ctx, operationID)
	if err != nil {
		return err
	}
	value, _ := e.channelSlots.LoadOrStore(channel, make(chan struct{}, consts.BatchSubBatchConcurrency))
	slots := value.(chan struct{})
	select {
	case slots <- struct{}{}:
		defer func() { <-slots }()
	case <-ctx.Done():
		return ctx.Err()
	}
	token := uuid.NewString()
	callCtx, cancel := context.WithTimeout(ctx, 4*time.Minute)
	defer cancel()
	if checker, ok := upstream.(interface {
		CheckSimApnAsyncCapability(context.Context, *model.SimCardApnChangeInput) error
	}); ok {
		snapshot, err := repo.GetSubBatch(callCtx, operationID)
		if err != nil || snapshot == nil {
			return err
		}
		if snapshot.Status == "done" || snapshot.Status == "review" {
			return nil
		}
		if snapshot.Status == "prepared" {
			if err := checker.CheckSimApnAsyncCapability(callCtx, &snapshot.Request); err != nil {
				persistCtx, stop := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
				defer stop()
				if delayErr := repo.DelaySubBatchPrecheck(persistCtx, *snapshot, time.Now().Add(30*time.Second), err.Error()); delayErr != nil {
					return fmt.Errorf("保存APN预检调度退避失败: %w", delayErr)
				}
				return fmt.Errorf("APN提交预检未就绪，未领取提交意图: %w", err)
			}
		}
	}
	sub, err := repo.ClaimSubBatch(ctx, operationID, token, time.Now())
	if err != nil || sub == nil {
		return err
	}
	if sub.Status == "prepared" {
		response, submitErr := upstream.SubmitSimApnFunction(callCtx, &sub.Request)
		if submitErr != nil {
			if isAPNRequestNotSent(submitErr) {
				return releaseUnsent(ctx, repo, *sub, submitErr)
			}
			return e.rememberUncertain(ctx, repo, *sub, "提交结果未确认（网络超时或连接中断）: "+submitErr.Error())
		}
		return e.handleResponse(ctx, repo, *sub, response, false)
	}
	if sub.Status == "submitting" || sub.Status == "uncertain" || (sub.Status == "accepted" && sub.JobID == "") {
		sub.Request.ReqData.RecoverOnly = true
		sub.Request.ReqData.QueryJobID = ""
		response, queryErr := upstream.QuerySimApnFunctionResult(callCtx, &sub.Request)
		if queryErr != nil {
			if isAPNRequestNotSent(queryErr) {
				return releaseUnsent(ctx, repo, *sub, queryErr)
			}
			return e.rememberUncertain(ctx, repo, *sub, "受理映射恢复失败（不重发）: "+queryErr.Error())
		}
		return e.handleResponse(ctx, repo, *sub, response, false)
	}
	sub.Request.ReqData.QueryJobID = sub.JobID
	sub.Request.ReqData.RecoverOnly = false
	response, queryErr := upstream.QuerySimApnFunctionResult(callCtx, &sub.Request)
	if queryErr != nil {
		if isAPNRequestNotSent(queryErr) {
			return releaseUnsent(ctx, repo, *sub, queryErr)
		}
		return e.queryUnavailable(ctx, repo, *sub, "查询请求超时或不可用: "+queryErr.Error())
	}
	return e.handleResponse(ctx, repo, *sub, response, true)
}

// QueryReview仅执行一次只读补查，不重新开启自动轮询或提交办理。
func (e *executor) QueryReview(ctx context.Context, operationID string) error {
	operationID = strings.TrimSpace(operationID)
	if operationID == "" {
		return fmt.Errorf("operationId不能为空")
	}
	repo, ok := e.repository.(SubBatchRepository)
	if !ok {
		return fmt.Errorf("未配置APN持久化仓储")
	}
	reviewRepo, ok := e.repository.(ReviewRepository)
	if !ok {
		return fmt.Errorf("未配置APN待核查查询仓储")
	}
	upstream, ok := e.upstream.(AsyncUpstream)
	if !ok {
		return fmt.Errorf("未配置APN只读补查客户端")
	}
	sub, err := reviewRepo.ClaimReviewSubBatch(ctx, operationID, uuid.NewString(), time.Now())
	if err != nil {
		return err
	}
	if sub == nil {
		snapshot, err := repo.GetSubBatch(ctx, operationID)
		if err != nil {
			return err
		}
		if snapshot != nil && snapshot.Status == "done" {
			return repo.FinalizeReadyBatch(ctx, snapshot.BatchNo)
		}
		return fmt.Errorf("子批不处于可查询的待核查状态，或已有查询/汇总正在运行")
	}
	callCtx, cancel := context.WithTimeout(ctx, 4*time.Minute)
	defer cancel()
	sub.Request.ReqData.SubmitOnly = false
	sub.Request.ReqData.QueryJobID = sub.JobID
	sub.Request.ReqData.RecoverOnly = sub.JobID == ""
	items, queryErr := upstream.QuerySimApnFunctionResult(callCtx, &sub.Request)
	if queryErr != nil {
		if err := e.deferResult(ctx, repo, *sub, "review", "手动补查未确认："+queryErr.Error()); err != nil {
			return err
		}
		return queryErr
	}
	return e.handleResponse(ctx, repo, *sub, items, true)
}

// QueryReviewBatch 是"手动同步上游结果"入口：只对父批下已冻结的待核查子批
// 做一次只读补查，然后尝试汇总父批。
// 不重发办理请求、不清零自动补查计数、不重新开启轮询调度。
func (e *executor) QueryReviewBatch(ctx context.Context, batchNo string) (int, error) {
	batchNo = strings.TrimSpace(batchNo)
	if batchNo == "" {
		return 0, fmt.Errorf("批次编号不能为空")
	}
	repo, ok := e.repository.(ReviewBatchRepository)
	if !ok {
		return 0, fmt.Errorf("未配置APN待核查批量同步仓储")
	}
	if _, ok := e.upstream.(AsyncUpstream); !ok {
		return 0, fmt.Errorf("未配置APN只读补查客户端")
	}
	ids, err := repo.ListReviewSubBatches(ctx, batchNo)
	if err != nil {
		return 0, err
	}
	queried := 0
	var firstErr error
	for _, id := range ids {
		if err := e.QueryReview(ctx, id); err != nil {
			g.Log().Warningf(ctx, "手动同步待核查子批未完成 batchNo=%s operationId=%s error=%v", batchNo, id, err)
			if firstErr == nil {
				firstErr = fmt.Errorf("operationId=%s: %w", id, err)
			}
			continue
		}
		queried++
	}
	// 已确认终态的子批会在这里收敛父批；仍为 review 的子批继续留在待核查。
	if err := repo.FinalizeReadyBatch(ctx, batchNo); err != nil && firstErr == nil {
		firstErr = err
	}
	g.Log().Infof(ctx, "手动同步APN上游结果 batchNo=%s 待核查子批=%d 本次已同步=%d", batchNo, len(ids), queried)
	if firstErr != nil {
		return queried, firstErr
	}
	return queried, nil
}

func isAPNRequestNotSent(err error) bool {
	var notSent interface{ APNRequestNotSent() bool }
	return errors.As(err, &notSent) && notSent.APNRequestNotSent()
}
func releaseUnsent(ctx context.Context, repo SubBatchRepository, sub SubBatch, cause error) error {
	persistCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
	defer cancel()
	return repo.ReleaseUnsentSubBatch(persistCtx, sub, cause.Error())
}

func nextResultCheck(sub SubBatch) time.Time {
	base := sub.SubmittedAt
	if base.IsZero() {
		base = time.Now()
	}
	return base.Add(firstResultDelay + time.Duration(sub.PollCount)*furtherResultDelay)
}

func (e *executor) rememberUncertain(ctx context.Context, repo SubBatchRepository, sub SubBatch, reason string) error {
	return e.deferResult(ctx, repo, sub, "uncertain", reason)
}

func (e *executor) queryUnavailable(ctx context.Context, repo SubBatchRepository, sub SubBatch, reason string) error {
	return e.deferResult(ctx, repo, sub, "accepted", reason)
}

func (e *executor) deferResult(ctx context.Context, repo SubBatchRepository, sub SubBatch, state, reason string) error {
	persistCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
	defer cancel()
	g.Log().Warningf(persistCtx, "APN补查 operationId=%s jobId=%s pollCount=%d reason=%s", sub.OperationID, sub.JobID, sub.PollCount, reason)
	if sub.Status == "review_query" || sub.PollCount >= maxResultQueries {
		if err := repo.UpdateSubBatch(persistCtx, sub, "review", sub.JobID, "补查窗口结束，办理结果未知："+reason, time.Time{}, sub.PollCount); err != nil {
			return err
		}
		return repo.FinalizeReadyBatch(persistCtx, sub.BatchNo)
	}
	return repo.UpdateSubBatch(persistCtx, sub, state, sub.JobID, reason, nextResultCheck(sub), sub.PollCount)
}

func (e *executor) handleResponse(ctx context.Context, repo SubBatchRepository, sub SubBatch, items []model.SimBaseRes[model.SimCardApnChangeOutputItem], queried bool) error {
	persistCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 15*time.Second)
	defer cancel()
	outcomes := make([]Outcome, len(sub.Request.SimKeys))
	used := make([]bool, len(items))
	pending := false
	reason := "未返回该卡结果，办理未确认"
	for i, key := range sub.Request.SimKeys {
		matched := -1
		for j, item := range items {
			if !used[j] && sameSim(&Detail{ICCID: key.ICCID, MSISDN: key.MSISDN, IMSI: key.IMSI}, item.SimKey) {
				matched = j
				break
			}
		}
		if matched < 0 {
			outcomes[i] = Outcome{Pending: true, Message: reason}
			pending = true
			continue
		}
		used[matched] = true
		item := items[matched]
		if item.ResData.Pending || item.ResData.Uncertain {
			pending = true
			outcomes[i] = Outcome{Pending: true, Code: item.Code, Message: item.Message}
			if strings.TrimSpace(item.Message) != "" {
				reason = item.Message
			}
			// Only Pending identifies a batch job: a synchronous per-card order is never queried as a job.
			if item.ResData.Pending && strings.TrimSpace(item.ResData.OrderNum) != "" {
				sub.JobID = strings.TrimSpace(item.ResData.OrderNum)
			}
			continue
		}
		outcomes[i] = Outcome{Success: item.Code == 0, Code: item.Code, Message: item.Message, OrderNum: item.ResData.OrderNum}
	}
	if pending {
		// Confirmed cards are committed now. Later missing/pending responses cannot overwrite their terminal rows.
		if err := repo.SaveSubBatchProgress(persistCtx, sub, outcomes); err != nil {
			return err
		}
		state := "accepted"
		if sub.JobID == "" {
			state = "uncertain"
			reason += "（缺少jobId，仅按operationId恢复，不重发）"
		}
		return e.deferResult(persistCtx, repo, sub, state, reason)
	}
	if err := repo.CompleteSubBatch(persistCtx, sub, outcomes); err != nil {
		return err
	}
	return repo.FinalizeReadyBatch(persistCtx, sub.BatchNo)
}

func mapOutcomesFromKeys(keys []*model.SimKey, response []model.SimBaseRes[model.SimCardApnChangeOutputItem]) []Outcome {
	details := make([]*Detail, 0, len(keys))
	for _, key := range keys {
		details = append(details, &Detail{ICCID: key.ICCID, MSISDN: key.MSISDN, IMSI: key.IMSI})
	}
	return mapOutcomes(details, response)
}

// RunReconciler uses a fixed global worker bound; waiting for a job never retains a worker slot.
func (e *executor) RunReconciler(ctx context.Context) error {
	repo, ok := e.repository.(SubBatchRepository)
	if !ok {
		return nil
	}
	upstream, ok := e.upstream.(AsyncUpstream)
	if !ok {
		return nil
	}
	jobs := make(chan string, 64)
	var inFlight sync.Map
	var workers sync.WaitGroup
	for i := 0; i < 20; i++ {
		workers.Add(1)
		go func() {
			defer workers.Done()
			for {
				select {
				case <-ctx.Done():
					return
				case id := <-jobs:
					if err := e.processSubBatch(ctx, repo, upstream, id); err != nil {
						g.Log().Errorf(ctx, "APN恢复失败 operationId=%s error=%v", id, err)
					}
					inFlight.Delete(id)
				}
			}
		}()
	}
	defer workers.Wait()
	ticker := time.NewTicker(5 * time.Second)
	defer ticker.Stop()
	for {
		batches, err := repo.ListRecoverableBatches(ctx, 32)
		if err != nil {
			g.Log().Errorf(ctx, "读取APN恢复父批失败: %v", err)
		} else {
			for _, batchNo := range batches {
				if err := e.Execute(ctx, batchNo); err != nil {
					g.Log().Errorf(ctx, "恢复APN父批失败 batchNo=%s error=%v", batchNo, err)
				}
			}
		}
		ids, err := repo.ListDueSubBatches(ctx, time.Now(), 128)
		if err != nil {
			g.Log().Errorf(ctx, "读取APN补查子批失败: %v", err)
		} else {
			for _, id := range ids {
				if _, loaded := inFlight.LoadOrStore(id, true); loaded {
					continue
				}
				select {
				case jobs <- id:
				case <-ctx.Done():
					inFlight.Delete(id)
					return nil
				default:
					inFlight.Delete(id)
				}
			}
		}
		select {
		case <-ctx.Done():
			return nil
		case <-ticker.C:
		}
	}
}