package host

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"fmt"
	"strconv"
	"strings"
	"time"

	"hyapp/pkg/tencentim"
	"hyapp/pkg/xerr"
	hostdomain "hyapp/services/user-service/internal/domain/host"
)

const (
	privateMessageHistoryHours        = 7 * 24
	privateMessageHistoryReadyDelay   = 2 * time.Hour
	privateMessageHistoryRetryDelay   = 10 * time.Minute
	privateMessageHistoryImportMax    = 500
	privateMessageHistoryEmptyMaxWait = 4 * time.Hour
)

// PrivateMessageHistoryClient 是腾讯 IM 离线归档边界；它只在 cron 批处理中调用，
// 不参与客户端发送、接收或 UserSig 签发链路。
type PrivateMessageHistoryClient interface {
	ListC2CHistoryFiles(ctx context.Context, hourStartUTC time.Time) ([]tencentim.C2CHistoryFile, error)
	StreamC2CHistory(ctx context.Context, file tencentim.C2CHistoryFile, consume func(tencentim.C2CHistoryMessage) error) (int, error)
}

// PrivateMessageHistoryRepository 保存小时同步租约和最小私聊统计事实。
type PrivateMessageHistoryRepository interface {
	PreparePrivateMessageHistoryHours(ctx context.Context, sdkAppID int64, fromHourMS int64, throughHourMS int64, nowMS int64) error
	ClaimPrivateMessageHistoryHour(ctx context.Context, sdkAppID int64, runID string, fromHourMS int64, throughHourMS int64, nowMS int64, lockTTL time.Duration) (hourStartMS int64, attempts int, claimed bool, err error)
	ImportPrivateMessageEvents(ctx context.Context, events []hostdomain.PrivateMessageEvent) (int, error)
	CompletePrivateMessageHistoryHour(ctx context.Context, sdkAppID int64, hourStartMS int64, runID string, sourceMessages int, importedEvents int, nowMS int64) error
	RetryPrivateMessageHistoryHour(ctx context.Context, sdkAppID int64, hourStartMS int64, runID string, retryAtMS int64, errorMessage string, nowMS int64) error
}

// PrivateMessageHistoryBatchResult 汇总一次小时归档处理，不把下载地址或消息正文带到 cron 日志。
type PrivateMessageHistoryBatchResult struct {
	Claimed       bool
	SourceCount   int
	ImportedCount int
	RetryPending  bool
}

// WithPrivateMessageHistorySource 挂载腾讯归档和 owner repository。SDKAppID 是共享归档主键，
// 即使 cron 按多个 app_code 并发触发，同一小时也只会被认领一次。
func WithPrivateMessageHistorySource(sdkAppID int64, client PrivateMessageHistoryClient) Option {
	return func(s *Service) {
		if sdkAppID <= 0 || client == nil {
			return
		}
		repository, ok := s.repository.(PrivateMessageHistoryRepository)
		if !ok {
			return
		}
		s.privateMessageHistorySDKAppID = sdkAppID
		s.privateMessageHistoryClient = client
		s.privateMessageHistoryRepository = repository
	}
}

// ProcessPrivateMessageHistoryBatch 处理最近七天中一个已完成的 UTC 小时。
// get_history 文件由腾讯异步生成，因此统计最多延迟约两小时，但不会给 IM 发消息增加等待。
func (s *Service) ProcessPrivateMessageHistoryBatch(ctx context.Context, runID string, lockTTL time.Duration, importBatchSize int) (PrivateMessageHistoryBatchResult, error) {
	if s == nil || s.privateMessageHistoryClient == nil || s.privateMessageHistoryRepository == nil || s.privateMessageHistorySDKAppID <= 0 {
		return PrivateMessageHistoryBatchResult{}, xerr.New(xerr.Unavailable, "private message history sync is not configured")
	}
	runID = strings.TrimSpace(runID)
	if runID == "" {
		return PrivateMessageHistoryBatchResult{}, xerr.New(xerr.InvalidArgument, "run_id is required")
	}
	if importBatchSize <= 0 || importBatchSize > privateMessageHistoryImportMax {
		importBatchSize = privateMessageHistoryImportMax
	}
	if lockTTL <= 0 {
		lockTTL = 5 * time.Minute
	}

	now := s.now().UTC()
	throughHour := now.Truncate(time.Hour).Add(-privateMessageHistoryReadyDelay)
	fromHour := throughHour.Add(-time.Duration(privateMessageHistoryHours-1) * time.Hour)
	if err := s.privateMessageHistoryRepository.PreparePrivateMessageHistoryHours(
		ctx,
		s.privateMessageHistorySDKAppID,
		fromHour.UnixMilli(),
		throughHour.UnixMilli(),
		now.UnixMilli(),
	); err != nil {
		return PrivateMessageHistoryBatchResult{}, err
	}

	hourStartMS, attempts, claimed, err := s.privateMessageHistoryRepository.ClaimPrivateMessageHistoryHour(
		ctx,
		s.privateMessageHistorySDKAppID,
		runID,
		fromHour.UnixMilli(),
		throughHour.UnixMilli(),
		now.UnixMilli(),
		lockTTL,
	)
	if err != nil || !claimed {
		return PrivateMessageHistoryBatchResult{}, err
	}
	result := PrivateMessageHistoryBatchResult{Claimed: true}
	hourStart := time.UnixMilli(hourStartMS).UTC()

	files, err := s.privateMessageHistoryClient.ListC2CHistoryFiles(ctx, hourStart)
	if err != nil {
		if tencentim.IsRESTErrorCode(err, 1004) {
			// 1004 同时表示“无消息”和“文件尚未生成”。最近小时先重试；超过生成窗口后
			// 视为空小时完成，避免一个永久空小时阻塞七天回补。
			if attempts < 3 && now.Sub(hourStart) < privateMessageHistoryEmptyMaxWait {
				if retryErr := s.markPrivateMessageHistoryRetry(ctx, hourStartMS, runID, err, now); retryErr != nil {
					return result, retryErr
				}
				result.RetryPending = true
				return result, nil
			}
			if completeErr := s.privateMessageHistoryRepository.CompletePrivateMessageHistoryHour(
				ctx, s.privateMessageHistorySDKAppID, hourStartMS, runID, 0, 0, now.UnixMilli(),
			); completeErr != nil {
				return result, completeErr
			}
			return result, nil
		}
		if retryErr := s.markPrivateMessageHistoryRetry(ctx, hourStartMS, runID, err, now); retryErr != nil {
			return result, retryErr
		}
		return result, err
	}

	pending := make([]hostdomain.PrivateMessageEvent, 0, importBatchSize)
	flush := func() error {
		if len(pending) == 0 {
			return nil
		}
		imported, importErr := s.privateMessageHistoryRepository.ImportPrivateMessageEvents(ctx, pending)
		if importErr != nil {
			return importErr
		}
		result.ImportedCount += imported
		pending = pending[:0]
		return nil
	}
	consume := func(message tencentim.C2CHistoryMessage) error {
		event, ok := privateMessageEventFromHistory(message)
		if !ok {
			return nil
		}
		pending = append(pending, event)
		if len(pending) >= importBatchSize {
			return flush()
		}
		return nil
	}

	for _, file := range files {
		processed, streamErr := s.privateMessageHistoryClient.StreamC2CHistory(ctx, file, consume)
		result.SourceCount += processed
		if streamErr != nil {
			if retryErr := s.markPrivateMessageHistoryRetry(ctx, hourStartMS, runID, streamErr, now); retryErr != nil {
				return result, retryErr
			}
			return result, streamErr
		}
	}
	if err := flush(); err != nil {
		if retryErr := s.markPrivateMessageHistoryRetry(ctx, hourStartMS, runID, err, now); retryErr != nil {
			return result, retryErr
		}
		return result, err
	}
	if err := s.privateMessageHistoryRepository.CompletePrivateMessageHistoryHour(
		ctx,
		s.privateMessageHistorySDKAppID,
		hourStartMS,
		runID,
		result.SourceCount,
		result.ImportedCount,
		s.now().UTC().UnixMilli(),
	); err != nil {
		return result, err
	}
	return result, nil
}

func (s *Service) markPrivateMessageHistoryRetry(ctx context.Context, hourStartMS int64, runID string, cause error, now time.Time) error {
	retryErr := s.privateMessageHistoryRepository.RetryPrivateMessageHistoryHour(
		ctx,
		s.privateMessageHistorySDKAppID,
		hourStartMS,
		runID,
		now.Add(privateMessageHistoryRetryDelay).UnixMilli(),
		cause.Error(),
		now.UnixMilli(),
	)
	if retryErr == nil {
		return nil
	}
	return errors.Join(cause, retryErr)
}

func privateMessageEventFromHistory(message tencentim.C2CHistoryMessage) (hostdomain.PrivateMessageEvent, bool) {
	senderText := strings.TrimSpace(message.FromAccount)
	targetText := strings.TrimSpace(message.ToAccount)
	senderUserID, senderErr := strconv.ParseInt(senderText, 10, 64)
	targetUserID, targetErr := strconv.ParseInt(targetText, 10, 64)
	if senderErr != nil || targetErr != nil || senderUserID <= 0 || targetUserID <= 0 || message.Timestamp <= 0 {
		// 共享 SDKAppID 可能包含管理员/第三方非数字 identifier；只有项目内部不可变 user_id
		// 才能安全映射 users 主键，其余消息不属于 Host Center 统计事实。
		return hostdomain.PrivateMessageEvent{}, false
	}
	identity := fmt.Sprintf(
		"%s\x00%s\x00%d\x00%d\x00%d",
		senderText,
		targetText,
		message.Timestamp,
		message.Sequence,
		message.Random,
	)
	hash := sha256.Sum256([]byte(identity))
	return hostdomain.PrivateMessageEvent{
		EventID:      "im-history:" + hex.EncodeToString(hash[:]),
		SenderUserID: senderUserID,
		TargetUserID: targetUserID,
		OccurredAtMS: message.Timestamp * int64(time.Second/time.Millisecond),
	}, true
}