package taskcenter

import (
	"bytes"
	"compress/gzip"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"os"
	"path"
	"sort"
	"strconv"
	"strings"
	"time"

	"chatapp3-golang/internal/model"

	"github.com/tencentyun/cos-go-sdk-v5"
	"gorm.io/gorm"
	"gorm.io/gorm/clause"
)

const (
	archiveGCLeaseDuration          = 5 * time.Minute
	maxArchiveGCSourceObjectsPerRun = 10
)

var archiveGCWorkerIdentity = buildArchiveGCWorkerIdentity()

type archiveGCRecoveryRow struct {
	ID            int64      `json:"id,string"`
	SysOrigin     string     `json:"sysOrigin"`
	EventID       string     `json:"eventId"`
	EventType     string     `json:"eventType"`
	UserID        int64      `json:"userId,string"`
	OccurredAt    time.Time  `json:"occurredAt"`
	RawJSON       string     `json:"rawJson"`
	Status        string     `json:"status"`
	RetryCount    int        `json:"retryCount"`
	COSKey        string     `json:"cosKey"`
	FailureReason string     `json:"failureReason"`
	CreateTime    time.Time  `json:"createTime"`
	UpdateTime    time.Time  `json:"updateTime"`
	ArchivedAt    *time.Time `json:"archivedAt"`
}

// ArchiveGCTick 由 chatapp-cron 显式触发且每次只处理一个小批次。COS 回读和恢复包上传都
// 位于数据库删除之前；任一步失败都会暂停本次 run，绝不继续“尽力删除”。
func (s *Service) ArchiveGCTick(ctx context.Context) (*ArchiveGCTickResponse, error) {
	if err := s.requireArchiveGCReady(ctx); err != nil {
		return nil, err
	}
	leaseToken := fmt.Sprintf("%s:%d", archiveGCWorkerIdentity, time.Now().UnixNano())
	run, acquired, reason, err := s.acquireArchiveGCLease(ctx, leaseToken)
	if err != nil {
		return nil, err
	}
	if !acquired {
		return &ArchiveGCTickResponse{RunID: run.RunID, Status: run.Status, SkippedReason: reason, Finished: run.Status == ArchiveGCStatusCompleted}, nil
	}
	if run.CutoffTime == nil {
		err := NewAppError(http.StatusConflict, "archive_gc_cutoff_missing", "archive GC cutoff is missing")
		s.pauseArchiveGCTick(run.RunID, leaseToken, err)
		return nil, err
	}

	rows, err := s.findArchiveGCCandidates(ctx, *run.CutoffTime, run.BatchSize)
	if err != nil {
		s.pauseArchiveGCTick(run.RunID, leaseToken, err)
		return nil, err
	}
	if len(rows) == 0 {
		if err := s.completeArchiveGCTick(ctx, run.RunID, leaseToken); err != nil {
			return nil, err
		}
		return &ArchiveGCTickResponse{RunID: run.RunID, Status: ArchiveGCStatusCompleted, Finished: true}, nil
	}
	rows = limitArchiveGCSourceObjects(rows, maxArchiveGCSourceObjectsPerRun)
	client, err := s.newCOSClient()
	if err != nil {
		s.pauseArchiveGCTick(run.RunID, leaseToken, err)
		return nil, err
	}
	if err := s.verifyArchiveGCSources(ctx, client, rows); err != nil {
		s.pauseArchiveGCTick(run.RunID, leaseToken, err)
		return nil, err
	}
	recoveryKey, recoverySHA, err := s.createAndVerifyArchiveGCRecovery(ctx, client, run.RunID, rows)
	if err != nil {
		s.pauseArchiveGCTick(run.RunID, leaseToken, err)
		return nil, err
	}
	deleted, protected, err := s.deleteVerifiedArchiveGCRows(ctx, run, leaseToken, recoveryKey, recoverySHA, rows)
	if err != nil {
		s.pauseArchiveGCTick(run.RunID, leaseToken, err)
		return nil, err
	}
	return &ArchiveGCTickResponse{
		RunID: run.RunID, Status: ArchiveGCStatusRunning, ProcessedRows: len(rows), DeletedRows: deleted,
		ProtectedRows: protected, RecoveryCOSKey: recoveryKey,
	}, nil
}

func (s *Service) findArchiveGCCandidates(ctx context.Context, cutoff time.Time, limit int) ([]model.TaskCenterEventArchiveOutbox, error) {
	if limit <= 0 || limit > maxArchiveGCBatch+1 {
		limit = defaultArchiveGCBatch
	}
	var rows []model.TaskCenterEventArchiveOutbox
	// FORCE INDEX 将扫描固定在 (status,create_time,id)；NOT EXISTS 只针对命中的小范围行
	// 查询活动小表。MAX_EXECUTION_TIME 防止长期活动保护了大量最旧行时，为凑满一批而
	// 在生产大表持续扫描；超时会让 tick 暂停且不产生 DELETE。
	err := s.db.WithContext(ctx).Raw(`
SELECT /*+ MAX_EXECUTION_TIME(5000) */ archive_row.*
FROM task_center_event_archive_outbox AS archive_row
FORCE INDEX (idx_task_center_archive_claim_create)
WHERE archive_row.status = ?
  AND archive_row.create_time < ?
  AND archive_row.archived_at IS NOT NULL
  AND archive_row.archived_at < ?
  AND archive_row.occurred_at IS NOT NULL
  AND archive_row.occurred_at < ?
  AND archive_row.cos_key <> ''
  AND (
    archive_row.event_type <> ?
    OR NOT EXISTS (
      SELECT 1
      FROM yumi_game_king_activity AS activity
      WHERE activity.enabled = 1
	        AND activity.settlement_status <> 'COMPLETED'
        AND activity.sys_origin = archive_row.sys_origin
        AND archive_row.occurred_at >= activity.start_time
        AND archive_row.occurred_at < activity.end_time
    )
  )
ORDER BY archive_row.create_time ASC, archive_row.id ASC
LIMIT ?`, ArchiveStatusUploaded, cutoff, cutoff, cutoff, EventTypeGameConsumeGold, limit).Scan(&rows).Error
	return rows, err
}

func (s *Service) acquireArchiveGCLease(ctx context.Context, leaseToken string) (model.TaskCenterArchiveGCRun, bool, string, error) {
	var run model.TaskCenterArchiveGCRun
	claimed := false
	err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("id = ?", archiveGCControlID).First(&run).Error; err != nil {
			return archiveGCMigrationError(err)
		}
		if run.Status != ArchiveGCStatusRunning {
			return nil
		}
		var databaseClock struct {
			Now time.Time `gorm:"column:now_time"`
		}
		if err := tx.Raw("SELECT NOW(3) AS now_time").Scan(&databaseClock).Error; err != nil {
			return err
		}
		now := databaseClock.Now
		if now.IsZero() {
			return fmt.Errorf("database clock returned zero time")
		}
		if run.LeaseUntil != nil && run.LeaseUntil.After(now) {
			return nil
		}
		leaseUntil := now.Add(archiveGCLeaseDuration)
		if err := tx.Model(&model.TaskCenterArchiveGCRun{}).Where("id = ?", archiveGCControlID).Updates(map[string]any{
			"lease_owner": leaseToken, "lease_until": leaseUntil, "update_time": now,
		}).Error; err != nil {
			return err
		}
		run.LeaseOwner = leaseToken
		run.LeaseUntil = &leaseUntil
		claimed = true
		return nil
	})
	if err != nil {
		return run, false, "", err
	}
	if run.Status != ArchiveGCStatusRunning {
		return run, false, "not_running", nil
	}
	if !claimed {
		return run, false, "lease_busy", nil
	}
	return run, true, "", nil
}

func (s *Service) verifyArchiveGCSources(ctx context.Context, client *cos.Client, rows []model.TaskCenterEventArchiveOutbox) error {
	groups := make(map[string][]model.TaskCenterEventArchiveOutbox)
	for _, row := range rows {
		groups[row.COSKey] = append(groups[row.COSKey], row)
	}
	keys := make([]string, 0, len(groups))
	for key := range groups {
		keys = append(keys, key)
	}
	sort.Strings(keys)
	for _, cosKey := range keys {
		// HEAD 确认对象当前版本，GET+gunzip 再确认候选 raw_json 确实包含在对象内；
		// 不能因为数据库已有 cos_key 就假设历史 PUT 完整成功。
		head, err := headAndVerifyArchiveObject(ctx, client, cosKey, "", -1, false)
		if err != nil {
			return fmt.Errorf("verify source archive %s HEAD: %w", cosKey, err)
		}
		if head.CompressedSize > maxArchiveCOSObjectBytes {
			return fmt.Errorf("source archive %s exceeds %d bytes", cosKey, maxArchiveCOSObjectBytes)
		}
		body, _, err := downloadArchiveObject(ctx, client, cosKey)
		if err != nil {
			return fmt.Errorf("verify source archive %s GET: %w", cosKey, err)
		}
		if int64(len(body)) != head.CompressedSize {
			return fmt.Errorf("source archive %s changed between HEAD and GET", cosKey)
		}
		sha := archiveSHA256(body)
		verifiedHead, err := headAndVerifyArchiveObject(ctx, client, cosKey, sha, int64(len(body)), false)
		if err != nil {
			return fmt.Errorf("verify source archive %s metadata: %w", cosKey, err)
		}
		lines, err := archiveObjectLines(body)
		if err != nil {
			return fmt.Errorf("verify source archive %s gzip: %w", cosKey, err)
		}
		available := make(map[string]int, len(lines))
		for _, line := range lines {
			available[archiveLineSHA256(line)]++
		}
		for _, row := range groups[cosKey] {
			hash := archiveLineSHA256(row.RawJSON)
			if available[hash] <= 0 {
				return fmt.Errorf("source archive %s does not contain outbox row %d", cosKey, row.ID)
			}
			available[hash]--
		}
		var existing model.TaskCenterArchiveManifest
		existingErr := s.db.WithContext(ctx).Where("cos_key = ?", cosKey).First(&existing).Error
		if existingErr == nil && existing.SHA256 != "" && !strings.EqualFold(existing.SHA256, sha) {
			return fmt.Errorf("source archive %s no longer matches its manifest", cosKey)
		}
		if existingErr != nil && !errors.Is(existingErr, gorm.ErrRecordNotFound) {
			return existingErr
		}
		verifiedAt := time.Now()
		if err := s.upsertArchiveManifest(ctx, model.TaskCenterArchiveManifest{
			ObjectType: archiveManifestObjectArchive, COSKey: cosKey, SHA256: sha, ETag: verifiedHead.ETag,
			CompressedSize: int64(len(body)), RowCount: len(lines), MinOutboxID: existing.MinOutboxID,
			MaxOutboxID: existing.MaxOutboxID, MinOccurredAt: existing.MinOccurredAt,
			MaxOccurredAt: existing.MaxOccurredAt, VerificationStatus: archiveManifestVerified,
			VerifiedAt: &verifiedAt,
		}); err != nil {
			return err
		}
	}
	return nil
}

func (s *Service) createAndVerifyArchiveGCRecovery(
	ctx context.Context,
	client *cos.Client,
	runID string,
	rows []model.TaskCenterEventArchiveOutbox,
) (string, string, error) {
	body, err := gzipArchiveGCRecoveryRows(rows)
	if err != nil {
		return "", "", err
	}
	cosKey := s.buildArchiveGCRecoveryKey(runID, rows)
	verified, err := putAndHeadArchiveObject(ctx, client, cosKey, body, map[string]string{
		"object-type": "gc-recovery", "run-id": runID, "row-count": strconv.Itoa(len(rows)),
	})
	if err != nil {
		return "", "", err
	}
	// 恢复包本身也必须 GET+gunzip 并逐行比对 ID/raw_json，HEAD 成功不等于内容可恢复。
	downloaded, _, err := downloadArchiveObject(ctx, client, cosKey)
	if err != nil {
		return "", "", err
	}
	if archiveSHA256(downloaded) != verified.SHA256 {
		return "", "", fmt.Errorf("recovery object %s sha256 mismatch after upload", cosKey)
	}
	lines, err := archiveObjectLines(downloaded)
	if err != nil {
		return "", "", err
	}
	if len(lines) != len(rows) {
		return "", "", fmt.Errorf("recovery object %s row count %d does not match %d", cosKey, len(lines), len(rows))
	}
	expected := make(map[int64]string, len(rows))
	for _, row := range rows {
		expected[row.ID] = archiveLineSHA256(row.RawJSON)
	}
	for _, line := range lines {
		var recovered archiveGCRecoveryRow
		if err := json.Unmarshal([]byte(line), &recovered); err != nil {
			return "", "", fmt.Errorf("decode recovery object %s: %w", cosKey, err)
		}
		expectedHash, ok := expected[recovered.ID]
		if !ok || expectedHash != archiveLineSHA256(recovered.RawJSON) {
			return "", "", fmt.Errorf("recovery object %s has unexpected row %d", cosKey, recovered.ID)
		}
		delete(expected, recovered.ID)
	}
	if len(expected) != 0 {
		return "", "", fmt.Errorf("recovery object %s is missing rows", cosKey)
	}
	verifiedAt := time.Now()
	minID, maxID, minOccurredAt, maxOccurredAt := archiveGroupBounds(rows)
	parentCOSKey := ""
	if rowsShareArchiveCOSKey(rows) {
		parentCOSKey = rows[0].COSKey
	}
	if err := s.upsertArchiveManifest(ctx, model.TaskCenterArchiveManifest{
		ObjectType: archiveManifestObjectRecovery, RunID: runID, COSKey: cosKey, ParentCOSKey: parentCOSKey,
		SHA256: verified.SHA256, ETag: verified.ETag, CompressedSize: verified.CompressedSize,
		RowCount: len(rows), MinOutboxID: minID, MaxOutboxID: maxID, MinOccurredAt: minOccurredAt,
		MaxOccurredAt: maxOccurredAt, VerificationStatus: archiveManifestVerified, VerifiedAt: &verifiedAt,
	}); err != nil {
		return "", "", err
	}
	return cosKey, verified.SHA256, nil
}

func (s *Service) deleteVerifiedArchiveGCRows(
	ctx context.Context,
	run model.TaskCenterArchiveGCRun,
	leaseToken, recoveryKey, recoverySHA string,
	candidates []model.TaskCenterEventArchiveOutbox,
) (int, int, error) {
	if run.CutoffTime == nil {
		return 0, 0, NewAppError(http.StatusConflict, "archive_gc_cutoff_missing", "archive GC cutoff is missing")
	}
	candidateByID := make(map[int64]model.TaskCenterEventArchiveOutbox, len(candidates))
	ids := make([]int64, 0, len(candidates))
	for _, row := range candidates {
		candidateByID[row.ID] = row
		ids = append(ids, row.ID)
	}
	deleted, protected := 0, 0
	err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
		var lockedRun model.TaskCenterArchiveGCRun
		// 与 Game King 启用门禁锁同一行。GC 在这把锁下复核活动窗口并删行，启用请求
		// 要么先完成并被本事务看到，要么等待删除提交后发现 GC 仍为 RUNNING 而失败。
		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("id = ?", archiveGCControlID).First(&lockedRun).Error; err != nil {
			return archiveGCMigrationError(err)
		}
		if lockedRun.RunID != run.RunID || lockedRun.Status != ArchiveGCStatusRunning || lockedRun.LeaseOwner != leaseToken {
			return NewAppError(http.StatusConflict, "archive_gc_lease_lost", "archive GC lease or state changed before delete")
		}
		var recoveryManifest model.TaskCenterArchiveManifest
		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where(
			"cos_key = ? AND run_id = ? AND object_type = ? AND verification_status = ? AND sha256 = ?",
			recoveryKey, run.RunID, archiveManifestObjectRecovery, archiveManifestVerified, recoverySHA,
		).First(&recoveryManifest).Error; err != nil {
			if errors.Is(err, gorm.ErrRecordNotFound) {
				return NewAppError(http.StatusConflict, "archive_gc_recovery_manifest_missing", "verified recovery manifest changed or is missing before delete")
			}
			return err
		}
		if recoveryManifest.RowCount != len(candidates) || recoveryManifest.VerifiedAt == nil {
			return NewAppError(http.StatusConflict, "archive_gc_recovery_manifest_invalid", "recovery manifest row count or verification time is invalid")
		}

		var lockedRows []model.TaskCenterEventArchiveOutbox
		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("id IN ?", ids).Order("create_time ASC, id ASC").Find(&lockedRows).Error; err != nil {
			return err
		}
		var protectedActivities []model.YumiGameKingActivity
		if err := tx.Where("enabled = ? AND settlement_status <> ?", true, "COMPLETED").Find(&protectedActivities).Error; err != nil {
			return err
		}
		deleteRows := make([]model.TaskCenterEventArchiveOutbox, 0, len(lockedRows))
		for _, row := range lockedRows {
			original, exists := candidateByID[row.ID]
			if !exists || row.Status != ArchiveStatusUploaded || row.COSKey == "" || row.COSKey != original.COSKey ||
				archiveLineSHA256(row.RawJSON) != archiveLineSHA256(original.RawJSON) || row.ArchivedAt == nil ||
				!row.CreateTime.Before(*run.CutoffTime) || !row.ArchivedAt.Before(*run.CutoffTime) ||
				row.OccurredAt.IsZero() || !row.OccurredAt.Before(*run.CutoffTime) || archiveGCActivityProtects(row, protectedActivities) {
				protected++
				continue
			}
			deleteRows = append(deleteRows, row)
		}
		if missing := len(candidates) - len(lockedRows); missing > 0 {
			// 另一个写路径删除/改动候选属于异常状态；恢复包仍完整，但本次只统计为保护跳过。
			protected += missing
		}
		deleteIDs := archiveRowIDs(deleteRows)
		if len(deleteIDs) > 0 {
			result := tx.Where("id IN ? AND status = ?", deleteIDs, ArchiveStatusUploaded).Delete(&model.TaskCenterEventArchiveOutbox{})
			if result.Error != nil {
				return result.Error
			}
			deleted = int(result.RowsAffected)
			if deleted != len(deleteIDs) {
				return fmt.Errorf("archive GC deleted %d rows, expected %d", deleted, len(deleteIDs))
			}
		}
		now := time.Now()
		updates := map[string]any{
			"scanned_rows":          gorm.Expr("scanned_rows + ?", len(candidates)),
			"eligible_rows":         gorm.Expr("eligible_rows + ?", len(deleteRows)),
			"protected_rows":        gorm.Expr("protected_rows + ?", protected),
			"deleted_rows":          gorm.Expr("deleted_rows + ?", deleted),
			"recovery_object_count": gorm.Expr("recovery_object_count + 1"),
			"lease_owner":           "", "lease_until": nil, "last_error": "", "update_time": now,
		}
		if deleted > 0 {
			last := deleteRows[len(deleteRows)-1]
			updates["cursor_create_time"] = last.CreateTime
			updates["cursor_id"] = last.ID
		}
		if err := tx.Model(&model.TaskCenterArchiveGCRun{}).Where("id = ?", archiveGCControlID).Updates(updates).Error; err != nil {
			return err
		}
		return nil
	})
	return deleted, protected, err
}

func (s *Service) completeArchiveGCTick(ctx context.Context, runID, leaseToken string) error {
	return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
		var locked model.TaskCenterArchiveGCRun
		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("id = ?", archiveGCControlID).First(&locked).Error; err != nil {
			return archiveGCMigrationError(err)
		}
		if locked.RunID != runID || locked.Status != ArchiveGCStatusRunning || locked.LeaseOwner != leaseToken {
			return NewAppError(http.StatusConflict, "archive_gc_lease_lost", "archive GC lease or state changed before completion")
		}
		return tx.Model(&model.TaskCenterArchiveGCRun{}).Where("id = ?", archiveGCControlID).Updates(map[string]any{
			"status": ArchiveGCStatusCompleted, "lease_owner": "", "lease_until": nil,
			"last_error": "", "update_time": time.Now(),
		}).Error
	})
}

func (s *Service) pauseArchiveGCTick(runID, leaseToken string, workErr error) {
	// HTTP/cron 超时往往已取消传入 ctx；失败状态必须用独立短上下文落库，否则控制行会
	// 继续显示 RUNNING，直到租约过期后再次触发同一批 COS 工作。
	failureCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	message := "archive GC tick failed"
	if workErr != nil {
		message = truncateTaskCenterFailure(workErr.Error())
	}
	_ = s.db.WithContext(failureCtx).Model(&model.TaskCenterArchiveGCRun{}).
		Where("id = ? AND run_id = ? AND status = ? AND lease_owner = ?", archiveGCControlID, runID, ArchiveGCStatusRunning, leaseToken).
		Updates(map[string]any{
			"status": ArchiveGCStatusPaused, "lease_owner": "", "lease_until": nil,
			"last_error": message, "update_time": time.Now(),
		}).Error
}

func gzipArchiveGCRecoveryRows(rows []model.TaskCenterEventArchiveOutbox) ([]byte, error) {
	var buffer bytes.Buffer
	writer := gzip.NewWriter(&buffer)
	encoder := json.NewEncoder(writer)
	for _, row := range rows {
		if err := encoder.Encode(archiveGCRecoveryRow{
			ID: row.ID, SysOrigin: row.SysOrigin, EventID: row.EventID, EventType: row.EventType,
			UserID: row.UserID, OccurredAt: row.OccurredAt, RawJSON: row.RawJSON, Status: row.Status,
			RetryCount: row.RetryCount, COSKey: row.COSKey, FailureReason: row.FailureReason,
			CreateTime: row.CreateTime, UpdateTime: row.UpdateTime, ArchivedAt: row.ArchivedAt,
		}); err != nil {
			_ = writer.Close()
			return nil, err
		}
	}
	if err := writer.Close(); err != nil {
		return nil, err
	}
	return buffer.Bytes(), nil
}

func (s *Service) buildArchiveGCRecoveryKey(runID string, rows []model.TaskCenterEventArchiveOutbox) string {
	prefix := strings.Trim(strings.TrimSpace(s.cfg.TaskCenter.Archive.Prefix), "/")
	if prefix == "" {
		prefix = "task-center-events/v1"
	}
	return path.Join(prefix, "gc-recovery", "run="+sanitizeArchivePartition(runID), fmt.Sprintf(
		"part-%s-%d-%d.jsonl.gz", time.Now().UTC().Format("20060102T150405.000000000Z"), rows[0].ID, len(rows),
	))
}

func archiveGCActivityProtects(row model.TaskCenterEventArchiveOutbox, activities []model.YumiGameKingActivity) bool {
	if row.EventType != EventTypeGameConsumeGold {
		return false
	}
	for _, activity := range activities {
		if activity.SysOrigin == row.SysOrigin && !row.OccurredAt.Before(activity.StartTime) && row.OccurredAt.Before(activity.EndTime) {
			return true
		}
	}
	return false
}

func limitArchiveGCSourceObjects(rows []model.TaskCenterEventArchiveOutbox, maximum int) []model.TaskCenterEventArchiveOutbox {
	if maximum <= 0 {
		return nil
	}
	seen := make(map[string]struct{}, maximum)
	for index, row := range rows {
		if _, exists := seen[row.COSKey]; exists {
			continue
		}
		if len(seen) == maximum {
			return rows[:index]
		}
		seen[row.COSKey] = struct{}{}
	}
	return rows
}

func rowsShareArchiveCOSKey(rows []model.TaskCenterEventArchiveOutbox) bool {
	if len(rows) == 0 {
		return false
	}
	key := rows[0].COSKey
	for _, row := range rows[1:] {
		if row.COSKey != key {
			return false
		}
	}
	return true
}

func buildArchiveGCWorkerIdentity() string {
	hostname, _ := os.Hostname()
	if strings.TrimSpace(hostname) == "" {
		hostname = "unknown-host"
	}
	return fmt.Sprintf("%s:%d", hostname, os.Getpid())
}