package task

import (
	"context"
	"crypto/sha256"
	"encoding/csv"
	"fmt"
	"io"
	"os"
	"strings"

	"github.com/bytedance/sonic"
	"github.com/hibiken/asynq"
	"go.uber.org/zap"
	"gorm.io/gorm"

	"github.com/break/junhong_cmp_fiber/internal/infrastructure/audit"
	"github.com/break/junhong_cmp_fiber/internal/model"
	"github.com/break/junhong_cmp_fiber/internal/store/postgres"
	"github.com/break/junhong_cmp_fiber/pkg/auditcontext"
	"github.com/break/junhong_cmp_fiber/pkg/auditfailure"
	"github.com/break/junhong_cmp_fiber/pkg/constants"
	pkgerrors "github.com/break/junhong_cmp_fiber/pkg/errors"
	"github.com/break/junhong_cmp_fiber/pkg/storage"
)

// OrderPackageInvalidatePayload 批量失效订单套餐任务载荷
type OrderPackageInvalidatePayload struct {
	TaskID uint `json:"task_id"`
}

// OrderPackageInvalidateHandler 批量失效订单套餐任务处理器
type OrderPackageInvalidateHandler struct {
	taskStore         *postgres.OrderPackageInvalidateTaskStore
	orderStore        *postgres.OrderStore
	packageUsageStore *postgres.PackageUsageStore
	storageService    *storage.Service
	logger            *zap.Logger
	auditWriter       *audit.Writer
}

// NewOrderPackageInvalidateHandler 创建处理器实例
func NewOrderPackageInvalidateHandler(
	taskStore *postgres.OrderPackageInvalidateTaskStore,
	orderStore *postgres.OrderStore,
	packageUsageStore *postgres.PackageUsageStore,
	storageSvc *storage.Service,
	logger *zap.Logger,
	auditWriters ...*audit.Writer,
) *OrderPackageInvalidateHandler {
	handler := &OrderPackageInvalidateHandler{
		taskStore:         taskStore,
		orderStore:        orderStore,
		packageUsageStore: packageUsageStore,
		storageService:    storageSvc,
		logger:            logger,
	}
	if len(auditWriters) > 0 {
		handler.auditWriter = auditWriters[0]
	}
	return handler
}

// Handle 处理批量失效订单套餐任务
// POST /api/admin/order-package-invalidate-tasks
func (h *OrderPackageInvalidateHandler) Handle(ctx context.Context, t *asynq.Task) error {
	var payload OrderPackageInvalidatePayload
	if err := sonic.Unmarshal(t.Payload(), &payload); err != nil {
		h.logger.Error("解析批量失效任务载荷失败",
			zap.Error(err),
			zap.String("task_id", t.ResultWriter().TaskID()),
		)
		return asynq.SkipRetry
	}

	importTask, err := h.taskStore.GetByID(ctx, payload.TaskID)
	if err != nil {
		h.logger.Error("获取批量失效任务失败",
			zap.Uint("task_id", payload.TaskID),
			zap.Error(err),
		)
		return asynq.SkipRetry
	}
	rootEventID := audit.TaskEventID(constants.AuditResourceOrderPackageInvalidateTask, importTask.ID, "completed")
	ctx = auditcontext.With(ctx, auditcontext.Context{
		ActorKind: constants.AuditActorSystemTask, ActorID: constants.TaskTypeOrderPackageInvalidate,
		ActorName: "订单套餐批量失效任务", Source: constants.AuditSourceWorker,
		CorrelationID: importTask.TaskNo, ParentEventID: rootEventID,
	})
	claimed, err := h.taskStore.Claim(ctx, importTask.ID)
	if err != nil {
		return err
	}
	if !claimed {
		h.logger.Info("批量失效任务已处理，跳过",
			zap.Uint("task_id", payload.TaskID),
			zap.Int("status", importTask.Status),
		)
		return nil
	}

	h.logger.Info("开始处理批量失效订单套餐任务",
		zap.Uint("task_id", importTask.ID),
		zap.String("task_no", importTask.TaskNo),
	)

	orderNos, err := h.downloadAndParseCSV(ctx, importTask)
	if err != nil {
		h.logger.Error("下载或解析 CSV 失败",
			zap.Uint("task_id", importTask.ID),
			zap.Error(err),
		)
		if finishErr := h.finishInvalidateTask(ctx, importTask, 0, 0, 1, model.ImportTaskStatusFailed, err.Error(), nil); finishErr != nil {
			h.resetInvalidateTaskForRetry(ctx, importTask.ID)
			return finishErr
		}
		return asynq.SkipRetry
	}

	successCount, failedItems := h.processRows(ctx, importTask.ID, orderNos)
	failCount := len(failedItems)
	totalCount := len(orderNos)

	status, errorMessage := model.ImportTaskStatusCompleted, ""
	if failCount > 0 && successCount == 0 {
		status, errorMessage = model.ImportTaskStatusFailed, "所有行均处理失败"
	}
	if err := h.finishInvalidateTask(ctx, importTask, totalCount, successCount, failCount, status, errorMessage, model.ImportResultItems(toImportResultItems(failedItems))); err != nil {
		h.resetInvalidateTaskForRetry(ctx, importTask.ID)
		return err
	}

	h.logger.Info("批量失效订单套餐任务完成",
		zap.Uint("task_id", importTask.ID),
		zap.Int("total", totalCount),
		zap.Int("success", successCount),
		zap.Int("fail", failCount),
	)

	return nil
}

func (h *OrderPackageInvalidateHandler) resetInvalidateTaskForRetry(ctx context.Context, taskID uint) {
	_ = h.taskStore.DB().WithContext(ctx).Model(&model.OrderPackageInvalidateTask{}).
		Where("id = ? AND status = ?", taskID, model.ImportTaskStatusProcessing).
		Updates(map[string]any{"status": model.ImportTaskStatusPending, "started_at": nil}).Error
}

// invalidateRow 单行处理结果
type invalidateRow struct {
	line    int
	orderNo string
	reason  string
}

// processRows 逐行处理订单号，返回成功数和失败列表
func (h *OrderPackageInvalidateHandler) processRows(ctx context.Context, taskID uint, rows []string) (int, []invalidateRow) {
	successCount := 0
	var failed []invalidateRow

	for i, orderNo := range rows {
		line := i + 2 // 第1行为表头，数据从第2行开始
		if err := h.processOneOrder(ctx, taskID, orderNo); err != nil {
			failed = append(failed, invalidateRow{line: line, orderNo: orderNo, reason: err.Error()})
		} else {
			successCount++
		}
	}

	return successCount, failed
}

// processOneOrder 处理单个订单号：查订单 → 查套餐 → 批量更新状态=4
func (h *OrderPackageInvalidateHandler) processOneOrder(ctx context.Context, taskID uint, orderNo string) error {
	order, err := h.orderStore.GetByOrderNo(ctx, orderNo)
	if err != nil {
		h.appendInvalidateFailure(ctx, taskID, &model.Order{OrderNo: orderNo}, "订单不存在")
		return errOrderNotFound(orderNo)
	}

	queryFailed := false
	err = h.taskStore.DB().WithContext(ctx).Transaction(func(tx *gorm.DB) error {
		usages, queryErr := postgres.NewPackageUsageStore(tx, nil).ListActiveByOrderID(ctx, order.ID)
		if queryErr != nil {
			queryFailed = true
			return queryErr
		}
		if len(usages) == 0 {
			return nil
		}
		ids := make([]uint, 0, len(usages))
		resources := []audit.ResourceInput{audit.OrderResource(order, constants.AuditResourceRelationPrimary, constants.AuditResourceRoleOrderTarget)}
		for _, usage := range usages {
			ids = append(ids, usage.ID)
			resources = append(resources, audit.PackageUsageResource(usage, constants.AuditResourceRelationAffected, constants.AuditResourceRolePackageUsageTarget,
				map[string]any{"status": usage.Status}, map[string]any{"status": constants.PackageUsageStatusInvalidated}))
		}
		if err := postgres.NewPackageUsageStore(tx, nil).BatchUpdateStatus(ctx, ids, constants.PackageUsageStatusInvalidated); err != nil {
			return err
		}
		return h.auditWriter.Append(ctx, tx, audit.AppendInput{
			EventID:    audit.TaskEventID(constants.AuditResourceOrderPackageInvalidateTask, taskID, fmt.Sprintf("item:%d", order.ID)),
			ActionCode: constants.AuditActionOrderPackageInvalidateItem, Summary: "失效订单套餐权益",
			ScopeType: constants.AuditScopePlatform, Result: constants.AuditResultSuccess, Resources: resources,
		})
	})
	if err != nil {
		summary := "更新套餐状态失败"
		if queryFailed {
			summary = "查询套餐失败"
		}
		h.appendInvalidateFailure(ctx, taskID, order, summary)
		if queryFailed {
			return errQueryFailed(orderNo)
		}
		return errUpdateFailed(orderNo)
	}

	return nil
}

func (h *OrderPackageInvalidateHandler) appendInvalidateFailure(ctx context.Context, taskID uint, order *model.Order, summary string) {
	if h.auditWriter == nil || order == nil || order.OrderNo == "" {
		return
	}
	err := h.taskStore.DB().WithContext(ctx).Transaction(func(tx *gorm.DB) error {
		keyHash := sha256.Sum256([]byte(order.OrderNo))
		return h.auditWriter.Append(ctx, tx, audit.AppendInput{
			EventID:    fmt.Sprintf("task:order_invalidate:%d:failed:%x", taskID, keyHash[:6]),
			ActionCode: constants.AuditActionOrderPackageInvalidateItem, Summary: summary,
			ScopeType: constants.AuditScopePlatform, Result: constants.AuditResultFailed,
			ErrorCode: fmt.Sprintf("%d", pkgerrors.CodeDatabaseError), ErrorSummary: summary,
			Resources: []audit.ResourceInput{audit.OrderResource(order, constants.AuditResourceRelationPrimary, constants.AuditResourceRoleOrderTarget)},
		})
	})
	if err != nil {
		auditfailure.RecordSecondaryWriteFailure(constants.AuditActionOrderPackageInvalidateItem, order.OrderNo, "", auditcontext.From(ctx).CorrelationID, fmt.Sprintf("%d", pkgerrors.CodeDatabaseError), err)
	}
}

func (h *OrderPackageInvalidateHandler) finishInvalidateTask(ctx context.Context, task *model.OrderPackageInvalidateTask, totalCount, successCount, failCount, status int, errorMessage string, failedItems model.ImportResultItems) error {
	if h.auditWriter == nil {
		return pkgerrors.New(pkgerrors.CodeInvalidStatus, "订单套餐失效任务统一审计接缝未配置")
	}
	return h.taskStore.DB().WithContext(ctx).Transaction(func(tx *gorm.DB) error {
		txStore := h.taskStore.WithTx(tx)
		if err := txStore.UpdateResult(ctx, task.ID, totalCount, successCount, failCount, failedItems); err != nil {
			return err
		}
		if err := txStore.UpdateStatus(ctx, task.ID, status, errorMessage); err != nil {
			return err
		}
		rootID := audit.TaskEventID(constants.AuditResourceOrderPackageInvalidateTask, task.ID, "completed")
		var actualSuccess, actualFail int64
		if err := tx.WithContext(ctx).Model(&model.AuditEvent{}).
			Where("parent_event_id = ? AND action_code = ? AND result = ?", rootID, constants.AuditActionOrderPackageInvalidateItem, constants.AuditResultSuccess).
			Count(&actualSuccess).Error; err != nil {
			return err
		}
		if err := tx.WithContext(ctx).Model(&model.AuditEvent{}).
			Where("parent_event_id = ? AND action_code = ? AND result = ?", rootID, constants.AuditActionOrderPackageInvalidateItem, constants.AuditResultFailed).
			Count(&actualFail).Error; err != nil {
			return err
		}
		auditFailCount := int(actualFail)
		if auditFailCount == 0 && failCount > 0 {
			auditFailCount = failCount
		}
		return h.auditWriter.WriteTask(ctx, tx, audit.TaskInput{
			EventID: rootID, ActionCode: constants.AuditActionOrderPackageInvalidateTaskCompleted,
			Summary: "完成订单套餐批量失效任务", TaskID: task.ID, TaskNo: task.TaskNo,
			Result: batchAuditResult(int(actualSuccess), auditFailCount), CorrelationID: task.TaskNo,
			ParentEventID: audit.TaskEventID(constants.AuditResourceOrderPackageInvalidateTask, task.ID, "created"),
			BatchTotal:    int(actualSuccess) + auditFailCount, SuccessCount: int(actualSuccess), FailCount: auditFailCount,
			IdentitySnapshot: map[string]any{"id": task.ID, "task_no": task.TaskNo, "file_name": task.FileName},
			BeforeData:       map[string]any{"status": model.ImportTaskStatusProcessing},
			AfterData: map[string]any{
				"status": status, "total_count": totalCount, "success_count": successCount, "fail_count": failCount,
			},
			Metadata: map[string]any{"task_success_count": successCount, "task_fail_count": failCount},
		})
	})
}

// downloadAndParseCSV 从对象存储下载 CSV 并解析 order_no 列
func (h *OrderPackageInvalidateHandler) downloadAndParseCSV(ctx context.Context, task *model.OrderPackageInvalidateTask) ([]string, error) {
	if h.storageService == nil {
		return nil, ErrStorageNotConfigured
	}
	if task.StorageKey == "" {
		return nil, ErrStorageKeyEmpty
	}

	localPath, cleanup, err := h.storageService.DownloadToTemp(ctx, task.StorageKey)
	if err != nil {
		return nil, err
	}
	defer cleanup()

	return parseOrderNoCSV(localPath)
}

// parseOrderNoCSV 解析单列 CSV（第一列为 order_no，首行为表头）
func parseOrderNoCSV(filePath string) ([]string, error) {
	f, err := os.Open(filePath)
	if err != nil {
		return nil, err
	}
	defer f.Close()

	reader := csv.NewReader(f)
	reader.TrimLeadingSpace = true

	var orderNos []string
	firstRow := true
	for {
		record, err := reader.Read()
		if err == io.EOF {
			break
		}
		if err != nil {
			return nil, err
		}
		if firstRow {
			firstRow = false
			continue // 跳过表头
		}
		if len(record) == 0 {
			continue
		}
		orderNo := strings.TrimSpace(record[0])
		if orderNo == "" {
			continue
		}
		orderNos = append(orderNos, orderNo)
	}

	return orderNos, nil
}

// toImportResultItems 将内部失败记录转换为通用失败明细格式
// 复用 ImportResultItem，ICCID 字段存储 order_no
func toImportResultItems(rows []invalidateRow) []model.ImportResultItem {
	items := make([]model.ImportResultItem, 0, len(rows))
	for _, r := range rows {
		items = append(items, model.ImportResultItem{
			Line:   r.line,
			ICCID:  r.orderNo,
			Reason: r.reason,
		})
	}
	return items
}

func errOrderNotFound(orderNo string) error {
	return orderInvalidateError("订单不存在: " + orderNo)
}

func errQueryFailed(orderNo string) error {
	return orderInvalidateError("查询套餐失败: " + orderNo)
}

func errUpdateFailed(orderNo string) error {
	return orderInvalidateError("更新套餐状态失败: " + orderNo)
}

type orderInvalidateError string

func (e orderInvalidateError) Error() string { return string(e) }