package outbound

import (
	"context"
	"errors"
	"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/library/batchmetadata"
	"strings"

	"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/consts"
	"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/dao"
	"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/library/batchimport"
	"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/library/contexts"
	"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/library/rabbitmq"
	"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/logic/packagecustomer"
	"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/model"
	"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/model/do"

	"github.com/gogf/gf/v2/database/gdb"
	"github.com/gogf/gf/v2/encoding/gjson"
	"github.com/gogf/gf/v2/errors/gerror"
	"github.com/gogf/gf/v2/frame/g"
	"github.com/gogf/gf/v2/net/ghttp"
	"github.com/gogf/gf/v2/os/gtime"
)

type AsyncMessage struct {
	BatchNo string `json:"batchNo"`
}

var errOutboundValidationRejected = errors.New("存在不满足出库条件的卡，本次未执行")

const outboundNotExecutedReason = "校验通过，但本批次存在失败明细，整批未执行出库"

func SubmitExcelAsync(ctx context.Context, file *ghttp.UploadFile, in *model.TSimCardOutboundExcelSubmitParam) (*batchimport.Result, error) {
	operator := batchimport.OperatorContext{TenantId: contexts.GetTenantId(ctx), DeptId: contexts.GetDeptId(ctx), UserId: contexts.GetUserId(ctx)}
	processor := NewProcessor(PreviewOptions{DeptId: operator.DeptId})
	result, err := batchimport.CreatePendingUpload(ctx, batchimport.SubmitRequest{
		SourceChannel: batchmetadata.Console,
		ExecutionMode: batchmetadata.Bulk,
		File:          file, BizParams: in, SourceType: consts.BatchSourceTypeFile, Operator: operator,
	}, processor, batchimport.PendingOptions{})
	if err != nil {
		return nil, err
	}
	return result, publishAsync(ctx, result.BatchNo)
}

func SubmitRowsAsync(ctx context.Context, in any, rows []batchimport.Row, sourceType int, sourceBatchNo string, strict bool) (*batchimport.Result, error) {
	operator := batchimport.OperatorContext{TenantId: contexts.GetTenantId(ctx), DeptId: contexts.GetDeptId(ctx), UserId: contexts.GetUserId(ctx)}
	processor := NewProcessor(PreviewOptions{DeptId: operator.DeptId, StrictStockIdentity: strict})
	result, err := batchimport.CreatePendingRows(ctx, batchimport.SubmitRequest{
		SourceChannel: batchmetadata.Console,
		ExecutionMode: batchmetadata.Bulk,
		BizParams:     in, SourceType: sourceType, SourceBatchNo: sourceBatchNo, Operator: operator,
	}, processor, rows, batchimport.PendingOptions{})
	if err != nil {
		return nil, err
	}
	return result, publishAsync(ctx, result.BatchNo)
}

func publishAsync(ctx context.Context, batchNo string) error {
	if strings.TrimSpace(batchNo) == "" {
		return gerror.New("出库批次号不能为空")
	}
	if !rabbitmq.Enabled(ctx) {
		err := gerror.New("RabbitMQ未启用，无法提交异步出库任务")
		if markErr := markAsyncFailed(ctx, batchNo, err); markErr != nil {
			return markErr
		}
		return err
	}
	if err := rabbitmq.Publish(ctx, consts.QueueBatchOutboundExecute, AsyncMessage{BatchNo: batchNo}); err != nil {
		publishErr := gerror.Wrap(err, "发布异步出库任务失败")
		if markErr := markAsyncFailed(ctx, batchNo, publishErr); markErr != nil {
			return markErr
		}
		return publishErr
	}
	return nil
}

func ConsumeAsync(ctx context.Context) error {
	return rabbitmq.ConsumeJSON[AsyncMessage](ctx, consts.QueueBatchOutboundExecute, "iotlcm-admin-outbound", handleAsyncMessage)
}

func handleAsyncMessage(ctx context.Context, message AsyncMessage) error {
	if strings.TrimSpace(message.BatchNo) == "" {
		return nil
	}
	main := &struct {
		BatchNo          string
		BatchType        int
		BatchAsyncStatus int
		BatchInfo        *gjson.Json
		SourceBatchNo    string
		TenantId         string
		CreatedDept      int64
		CreatedBy        int64
		TotalCount       int
	}{}
	if err := dao.BatchMain.Ctx(ctx).Where(dao.BatchMain.Columns().BatchNo, message.BatchNo).Scan(main); err != nil {
		return err
	}
	if main.BatchNo == "" || main.BatchType != consts.BatchTypeOutbound {
		return nil
	}
	strictStockIdentity := false
	if main.BatchInfo != nil {
		strictStockIdentity = main.BatchInfo.Get("strictStockIdentity").Bool()
	}
	processor := NewProcessor(PreviewOptions{DeptId: main.CreatedDept, StrictStockIdentity: strictStockIdentity})
	param, err := submitParamFromInfo(main.BatchInfo)
	if err != nil {
		return markAsyncFailed(ctx, message.BatchNo, err)
	}
	rows, err := loadAsyncRows(ctx, message.BatchNo)
	if err != nil {
		return markAsyncFailed(ctx, message.BatchNo, err)
	}
	operator := batchimport.OperatorContext{TenantId: main.TenantId, DeptId: main.CreatedDept, UserId: main.CreatedBy}
	requiredSource := ""
	if main.BatchInfo != nil {
		requiredSource = strings.TrimSpace(main.BatchInfo.Get("inboundBatchNo").String())
	}
	var (
		claimed            bool
		output             *CompleteOutput
		validationRejected bool
	)
	err = g.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
		claimResult, claimErr := tx.Ctx(ctx).Model(dao.BatchMain.Table()).
			Where(dao.BatchMain.Columns().BatchNo, message.BatchNo).
			Where(dao.BatchMain.Columns().BatchAsyncStatus, consts.BatchAsyncStatusPending).
			Data(do.BatchMain{
				BatchAsyncStatus: consts.BatchAsyncStatusProcessing,
				RunResult:        consts.BatchRunResultRunning,
				StartTime:        gtime.Now(),
				UpdatedBy:        main.CreatedBy,
			}).
			Update()
		if claimErr != nil {
			return claimErr
		}
		claimCount, claimErr := claimResult.RowsAffected()
		if claimErr != nil {
			return gerror.Wrap(claimErr, "读取异步出库任务领取结果失败")
		}
		if claimCount != 1 {
			return nil
		}
		claimed = true
		if _, updateErr := tx.Ctx(ctx).Model(dao.BatchDetail.Table()).
			Where(dao.BatchDetail.Columns().BatchNo, message.BatchNo).
			Data(do.BatchDetail{DetailStatus: consts.BatchDetailStatusRunning, UpdatedBy: main.CreatedBy}).
			Update(); updateErr != nil {
			return updateErr
		}
		var executeErr error
		output, executeErr = processor.ExecutePending(ctx, tx, message.BatchNo, param, rows, operator, requiredSource)
		if executeErr != nil {
			validationRejected = errors.Is(executeErr, errOutboundValidationRejected)
			return executeErr
		}
		if _, updateErr := tx.Ctx(ctx).Model(dao.BatchMain.Table()).Where(dao.BatchMain.Columns().BatchNo, message.BatchNo).Data(do.BatchMain{
			OrderSn: output.OrderSn, BatchAsyncStatus: consts.BatchAsyncStatusSuccess, RunResult: consts.BatchRunResultSuccess,
			SuccessCount: output.Quantity, ValidCount: output.Quantity, FailCount: 0, EndTime: gtime.Now(), UpdatedBy: main.CreatedBy,
		}).Update(); updateErr != nil {
			return updateErr
		}
		if _, updateErr := tx.Ctx(ctx).Model(dao.BatchDetail.Table()).Where(dao.BatchDetail.Columns().BatchNo, message.BatchNo).Data(do.BatchDetail{
			DetailStatus: consts.BatchDetailStatusSuccess, ErrorCode: "", ErrorReason: "", UpdatedBy: main.CreatedBy,
		}).Update(); updateErr != nil {
			return updateErr
		}
		return nil
	})
	if err != nil {
		if validationRejected {
			return markAsyncValidationFailed(ctx, message.BatchNo, main.TenantId, rows, err)
		}
		return markAsyncFailed(ctx, message.BatchNo, err)
	}
	if !claimed {
		return nil
	}
	if output != nil && output.Customer != nil && output.Customer.DeptId > 0 {
		packagecustomer.PublishSyncSafe(ctx, model.PackageCustomerSyncMessage{
			TenantId:      operator.TenantId,
			DeptId:        output.Customer.DeptId,
			UserId:        operator.UserId,
			Trigger:       "outbound",
			SourceBatchNo: output.BatchNo,
		})
	}
	return nil
}

func markAsyncValidationFailed(ctx context.Context, batchNo, tenantID string, rows []*Row, cause error) error {
	validCount, failCount := 0, 0
	for _, row := range rows {
		if row == nil {
			continue
		}
		if row.Reason == outboundNotExecutedReason {
			validCount++
		} else {
			failCount++
		}
	}
	return g.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
		for _, row := range rows {
			if row == nil {
				continue
			}
			status := consts.BatchDetailStatusFailed
			if row.Reason == outboundNotExecutedReason {
				status = consts.BatchDetailStatusNotExecuted
			}
			if _, err := tx.Ctx(ctx).Model(dao.BatchDetail.Table()).
				Where(dao.BatchDetail.Columns().BatchNo, batchNo).
				Where(dao.BatchDetail.Columns().TenantId, tenantID).
				Where(dao.BatchDetail.Columns().RowNo, row.RowNo).
				Data(do.BatchDetail{DetailStatus: status, ErrorReason: row.Reason}).Update(); err != nil {
				return gerror.Wrap(err, "保存出库校验明细失败")
			}
		}
		_, err := tx.Ctx(ctx).Model(dao.BatchMain.Table()).
			Where(dao.BatchMain.Columns().BatchNo, batchNo).
			Where(dao.BatchMain.Columns().TenantId, tenantID).
			Data(do.BatchMain{BatchAsyncStatus: consts.BatchAsyncStatusFailed, RunResult: consts.BatchRunResultFailed, ValidCount: validCount, SuccessCount: 0, FailCount: failCount, FailReason: asyncFailureReason(cause), EndTime: gtime.Now()}).Update()
		return gerror.Wrap(err, "保存出库校验结果失败")
	})
}

func markAsyncFailed(ctx context.Context, batchNo string, err error) error {
	reason := asyncFailureReason(err)
	if _, updateErr := g.DB().Model(dao.BatchMain.Table()).Ctx(ctx).Where(dao.BatchMain.Columns().BatchNo, batchNo).Data(do.BatchMain{
		BatchAsyncStatus: consts.BatchAsyncStatusFailed,
		RunResult:        consts.BatchRunResultFailed,
		ValidCount:       0,
		SuccessCount:     0,
		FailCount:        gdb.Raw(dao.BatchMain.Columns().TotalCount),
		FailReason:       reason,
		EndTime:          gtime.Now(),
	}).Update(); updateErr != nil {
		return gerror.Wrap(updateErr, reason)
	}
	if _, updateErr := g.DB().Model(dao.BatchDetail.Table()).Ctx(ctx).Where(dao.BatchDetail.Columns().BatchNo, batchNo).Data(do.BatchDetail{
		DetailStatus: consts.BatchDetailStatusFailed,
		ErrorReason:  reason,
	}).Update(); updateErr != nil {
		return gerror.Wrap(updateErr, reason)
	}
	return nil
}

func loadAsyncRows(ctx context.Context, batchNo string) ([]*Row, error) {
	var details []*struct {
		RowNo    int
		ItemId   string
		Snapshot *gjson.Json
	}
	if err := dao.BatchDetail.Ctx(ctx).Fields(dao.BatchDetail.Columns().RowNo, dao.BatchDetail.Columns().ItemId, dao.BatchDetail.Columns().Snapshot).Where(dao.BatchDetail.Columns().BatchNo, batchNo).OrderAsc(dao.BatchDetail.Columns().RowNo).Scan(&details); err != nil {
		return nil, err
	}
	rows := make([]*Row, 0, len(details))
	for _, detail := range details {
		if detail == nil {
			continue
		}
		row := &Row{RowNo: detail.RowNo, Iccid: strings.TrimSpace(detail.ItemId)}
		if detail.Snapshot != nil {
			row.Msisdn = detail.Snapshot.Get("MSISDN").String()
			row.Imsi = detail.Snapshot.Get("IMSI").String()
			row.SourceBatchNo = detail.Snapshot.Get("batchNo").String()
		}
		rows = append(rows, row)
	}
	if len(rows) == 0 {
		return nil, gerror.New("出库批次没有明细")
	}
	return rows, nil
}

func submitParamFromInfo(info *gjson.Json) (*model.TSimCardOutboundSubmitParam, error) {
	if info == nil {
		return nil, gerror.New("出库批次参数不存在")
	}
	param := &model.TSimCardOutboundSubmitParam{}
	if err := info.Scan(param); err != nil {
		return nil, err
	}
	if param.CustomerAgentId <= 0 || strings.TrimSpace(param.PackageCode) == "" || param.PackageBeginTime == nil {
		return nil, gerror.New("出库批次参数不完整")
	}
	return param, nil
}

const maxAsyncFailureReasonLength = 512

func asyncFailureReason(err error) string {
	if err == nil {
		return "异步出库失败"
	}
	reason := strings.TrimSpace(err.Error())
	if reason == "" {
		return "异步出库失败"
	}
	if strings.Contains(reason, "Prepared statement contains too many placeholders") {
		return "新增正式卡失败：单次批量写入参数数量超过数据库限制"
	}
	if strings.Contains(reason, "Data too long for column 'error_reason'") {
		return "异步出库失败：失败原因超过数据库字段长度限制"
	}
	runes := []rune(reason)
	if len(runes) > maxAsyncFailureReasonLength {
		return string(runes[:maxAsyncFailureReasonLength])
	}
	return reason
}