package task

import (
	"bytes"
	"context"
	"encoding/csv"
	"encoding/json"
	stderrors "errors"
	"github.com/break/junhong_cmp_fiber/pkg/auditcontext"
	"io"
	"os"
	"strings"
	"time"

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

	"github.com/break/junhong_cmp_fiber/internal/infrastructure/pollingpause"
	"github.com/break/junhong_cmp_fiber/internal/model"
	"github.com/break/junhong_cmp_fiber/internal/model/dto"
	"github.com/break/junhong_cmp_fiber/pkg/constants"
	"github.com/break/junhong_cmp_fiber/pkg/storage"
	"github.com/break/junhong_cmp_fiber/pkg/utils"
)

// PollingToggler 改一个资产的轮询开关并返回逐项结果（由资产轮询服务实现）。
type PollingToggler interface {
	ToggleAsset(ctx context.Context, assetType string, assetID uint, enable, scoped bool) dto.PollingToggleItem
}

// pollingStatusImportPayload 与 API 侧 PollingStatusImportPayload 同一 JSON 契约。
type pollingStatusImportPayload struct {
	TaskID uint `json:"task_id"`
}

// PollingStatusImportHandler 轮询开关文件导入：逐行独立处理（规则同勾选批量），部分成功不回滚，逐行记录结果。
type PollingStatusImportHandler struct {
	db      *gorm.DB
	storage *storage.Service
	toggler PollingToggler
	logger  *zap.Logger
}

// NewPollingStatusImportHandler 创建轮询开关导入任务处理器。
func NewPollingStatusImportHandler(db *gorm.DB, storageService *storage.Service, toggler PollingToggler, logger *zap.Logger) *PollingStatusImportHandler {
	if logger == nil {
		logger = zap.NewNop()
	}
	return &PollingStatusImportHandler{db: db, storage: storageService, toggler: toggler, logger: logger}
}

// Handle 处理一个导入任务；任务只会从待处理被领取一次。
func (h *PollingStatusImportHandler) Handle(ctx context.Context, t *asynq.Task) error {
	var payload pollingStatusImportPayload
	if err := json.Unmarshal(t.Payload(), &payload); err != nil || payload.TaskID == 0 {
		h.logger.Warn("轮询开关导入任务载荷无效", zap.ByteString("payload", t.Payload()))
		return nil
	}
	now := time.Now()
	claimed := h.db.WithContext(ctx).Model(&model.PollingStatusImportTask{}).
		Where("id = ? AND status = ?", payload.TaskID, model.ImportTaskStatusPending).
		Updates(map[string]any{"status": model.ImportTaskStatusProcessing, "started_at": now, "updated_at": now})
	if claimed.Error != nil {
		return claimed.Error
	}
	if claimed.RowsAffected == 0 {
		return nil
	}
	var task model.PollingStatusImportTask
	if err := h.db.WithContext(ctx).First(&task, payload.TaskID).Error; err != nil {
		return err
	}
	identifiers, err := h.downloadAndParse(ctx, task.StorageKey)
	if err != nil {
		h.fail(ctx, task.ID, err.Error())
		return nil
	}
	rowCtx := pollingpause.WithOrigin(ctx, pollingpause.Origin{Source: constants.PollingToggleSourceImport, Reason: task.Reason,
		ImportTaskID: task.ID, OperatorID: task.Creator})
	rowCtx = auditcontext.With(rowCtx, auditcontext.Context{ActorKind: constants.AuditActorSystemTask, ActorID: constants.TaskTypePollingStatusImport,
		ActorName: "轮询开关导入任务", Source: constants.AuditSourceWorker, CorrelationID: task.TaskNo})
	results := make(model.PollingStatusImportResults, 0, len(identifiers))
	success, skipped, failed := 0, 0, 0
	for index, identifier := range identifiers {
		item := h.processRow(rowCtx, task, index+1, identifier)
		switch item.Status {
		case constants.PollingToggleItemSuccess:
			success++
		case constants.PollingToggleItemSkipped:
			skipped++
		default:
			failed++
		}
		results = append(results, item)
	}
	completed := time.Now()
	if err := h.db.WithContext(context.WithoutCancel(ctx)).Model(&model.PollingStatusImportTask{}).Where("id = ?", task.ID).Updates(map[string]any{
		"status": model.ImportTaskStatusCompleted, "total_count": len(identifiers), "success_count": success, "skip_count": skipped,
		"fail_count": failed, "result_items": results, "completed_at": completed, "updated_at": completed,
	}).Error; err != nil {
		h.logger.Error("保存轮询开关导入结果失败", zap.Uint("task_id", task.ID), zap.Error(err))
		return err
	}
	h.logger.Info("轮询开关导入完成", zap.Uint("task_id", task.ID), zap.Int("total", len(identifiers)), zap.Int("success", success),
		zap.Int("skipped", skipped), zap.Int("failed", failed))
	return nil
}

func (h *PollingStatusImportHandler) processRow(ctx context.Context, task model.PollingStatusImportTask, line int, identifier string) model.PollingStatusImportResultItem {
	item := model.PollingStatusImportResultItem{Line: line, Identifier: identifier}
	if identifier == "" {
		item.Status, item.Reason = constants.PollingToggleItemFailed, "标识为空"
		return item
	}
	assetID, err := h.resolve(ctx, task.AssetType, identifier)
	if err != nil {
		item.Status, item.Reason = constants.PollingToggleItemForbidden, "资产不存在"
		if !stderrors.Is(err, gorm.ErrRecordNotFound) {
			item.Status, item.Reason = constants.PollingToggleItemFailed, "查询资产失败"
		}
		return item
	}
	item.AssetID = assetID
	result := h.toggler.ToggleAsset(ctx, task.AssetType, assetID, task.EnablePolling, false)
	item.Status, item.Reason = result.Status, result.Reason
	return item
}

// resolve 按 ICCID（卡）或 IMEI / 虚拟号（设备）查找资产。
func (h *PollingStatusImportHandler) resolve(ctx context.Context, assetType, identifier string) (uint, error) {
	if assetType == constants.AssetTypeDevice {
		var device model.Device
		err := h.db.WithContext(ctx).Select("id").Where("imei = ? OR virtual_no = ?", identifier, identifier).Order("id").First(&device).Error
		return device.ID, err
	}
	// 与 IotCardStore.GetByICCID 同口径：19 位按 iccid_19、20 位按 iccid_20 匹配（联通卡两种位数都可识别）。
	column := "iccid"
	switch len(identifier) {
	case 19:
		column = "iccid_19"
	case 20:
		column = "iccid_20"
	}
	var card model.IotCard
	err := h.db.WithContext(ctx).Select("id").Where(column+" = ?", identifier).First(&card).Error
	return card.ID, err
}

// downloadAndParse 下载 CSV 并取每行首列作为资产标识；首行为表头，空行忽略。
func (h *PollingStatusImportHandler) downloadAndParse(ctx context.Context, key string) ([]string, error) {
	if h.storage == nil {
		return nil, stderrors.New("对象存储服务未配置")
	}
	localPath, cleanup, err := h.storage.DownloadToTemp(ctx, key)
	if err != nil {
		return nil, stderrors.New("下载导入文件失败")
	}
	defer cleanup()
	data, err := os.ReadFile(localPath)
	if err != nil {
		return nil, stderrors.New("读取导入文件失败")
	}
	decoded, err := utils.DecodeTextToUTF8(data)
	if err != nil {
		return nil, stderrors.New("导入文件编码无法识别，请使用 UTF-8 或 GBK")
	}
	reader := csv.NewReader(bytes.NewReader(decoded))
	reader.FieldsPerRecord = -1
	reader.TrimLeadingSpace = true
	identifiers := make([]string, 0)
	header := true
	for {
		record, readErr := reader.Read()
		if readErr == io.EOF {
			break
		}
		if readErr != nil {
			return nil, stderrors.New("导入文件不是有效的 CSV")
		}
		if header {
			header = false
			continue
		}
		if len(record) == 0 || strings.TrimSpace(strings.Join(record, "")) == "" {
			continue
		}
		identifiers = append(identifiers, strings.TrimSpace(strings.TrimPrefix(record[0], "\uFEFF")))
		if len(identifiers) > constants.PollingStatusImportMaxRows {
			return nil, stderrors.New("导入文件超过 50000 行")
		}
	}
	if len(identifiers) == 0 {
		return nil, stderrors.New("导入文件没有数据行")
	}
	return identifiers, nil
}

func (h *PollingStatusImportHandler) fail(ctx context.Context, taskID uint, message string) {
	now := time.Now()
	if err := h.db.WithContext(context.WithoutCancel(ctx)).Model(&model.PollingStatusImportTask{}).Where("id = ?", taskID).Updates(map[string]any{
		"status": model.ImportTaskStatusFailed, "error_message": message, "completed_at": now, "updated_at": now,
	}).Error; err != nil {
		h.logger.Error("标记轮询开关导入任务失败时出错", zap.Uint("task_id", taskID), zap.Error(err))
	}
}