package service

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"time"

	"hyapp/pkg/appcode"
	"hyapp/pkg/giftlimits"
	"hyapp/pkg/xerr"
	"hyapp/services/room-service/internal/room/command"
	"hyapp/services/room-service/internal/room/outbox"
	"hyapp/services/room-service/internal/room/rank"
	"hyapp/services/room-service/internal/room/state"
	"hyapp/services/room-service/internal/router"
)

const (
	roomMutationCommitTimeout        = 5 * time.Second
	roomMutationConfirmCommitTimeout = 2 * time.Second
)

func (s *Service) mutateRoom(ctx context.Context, cmd command.Command, replayable bool, mutate func(now time.Time, current *state.RoomState, localRank *rank.LocalRank) (mutationResult, []outbox.Record, error)) (mutationResult, error) {
	startedAt := time.Now()
	var idempotencyMS int64
	var leaseMS int64
	var saveMutationMS int64
	var snapshotMS int64
	var projectionMS int64
	if cmd.RoomID() == "" || cmd.ID() == "" || cmd.ActorUserID() == 0 {
		// command meta 是幂等、路由、审计和权限判断的共同基础，缺一不可。
		return mutationResult{}, xerr.New(xerr.InvalidArgument, "command meta is incomplete")
	}
	if !giftlimits.ValidCommandID(cmd.ID()) {
		// room_command_log.command_id 是 VARCHAR(128)；入口显式拒绝比依赖 MySQL 截断/报错更安全，
		// 同时确保 wallet、lucky-gift 和 Room Cell 使用完全相同的稳定幂等键。
		return mutationResult{}, xerr.New(xerr.InvalidArgument, "command_id is too long")
	}
	if s.isDraining() {
		// 下线中的节点只让已经进入执行链路的命令收尾，不再接收新的 Room Cell 命令。
		return mutationResult{}, xerr.New(xerr.Unavailable, "room service is draining")
	}

	idempotencyStartedAt := time.Now()
	payload, existingCommand, seen, err := s.commandRecordForIdempotency(ctx, cmd)
	idempotencyMS = elapsedMS(idempotencyStartedAt)
	if err != nil {
		return mutationResult{}, err
	}
	if seen {
		// 幂等命令不重新执行 mutate，也不会再次调用 wallet 或写 outbox；SendGift V2 从首次 result_payload
		// 重放账务和幸运礼物结果，旧命令没有结果时才维持历史 snapshot-only 兼容语义。
		snapshot, err := s.currentSnapshot(ctx, cmd.RoomID())
		if err != nil {
			return mutationResult{}, err
		}

		return replayMutationResult(existingCommand, snapshot)
	}

	// 确保当前节点持有房间 lease，并在没有内存 Cell 时完成恢复。
	leaseStartedAt := time.Now()
	roomCell, lease, err := s.ensureCell(ctx, cmd.RoomID())
	leaseMS = elapsedMS(leaseStartedAt)
	if err != nil {
		return mutationResult{}, err
	}

	value, err := roomCell.Do(ctx, func(current *state.RoomState, currentRank *rank.LocalRank) (any, error) {
		// 外层预检与进入 Room Cell 之间可能已有相同 command_id 完成提交；必须在串行临界区内权威重查，
		// 否则第二个请求会复用 wallet 幂等回执却再次增加房间热度、榜单和 outbox。
		_, committedCommand, alreadyCommitted, err := s.commandRecordForIdempotency(ctx, cmd)
		if err != nil {
			return nil, err
		}
		if alreadyCommitted {
			return replayMutationResult(committedCommand, snapshotWithApp(ctx, current.ToProto()))
		}
		// 在副本上执行命令，只有 command log/outbox 保存成功后才整体替换当前状态。
		nextState := current.Clone()
		nextRank := currentRank.Clone()
		now := s.clock.Now()

		result, outboxRecords, err := mutate(now, nextState, nextRank)
		if err != nil {
			// 校验、钱包或业务失败时不提交状态，不写 command log。
			return nil, err
		}

		if result.snapshot == nil {
			// 无特殊返回需求的命令默认返回变更后的快照。
			result.snapshot = nextState.ToProto()
		}
		result.snapshot = snapshotWithApp(ctx, result.snapshot)

		stateChanged := nextState.Version != current.Version
		if stateChanged && nextState.Version != current.Version+1 {
			// Room Cell 命令日志按每次状态变更推进一个版本恢复；跳版本或复用旧版本都说明命令实现破坏了恢复语义。
			return nil, xerr.New(xerr.Internal, fmt.Sprintf("room version advanced from %d to %d", current.Version, nextState.Version))
		}

		if stateChanged {
			// 版本变化代表本命令产生实际状态变更，需要持久化命令和事件。
			if err := s.ensureLeaseStillOwned(ctx, cmd.RoomID(), lease); err != nil {
				return nil, err
			}
			// 机器人展示只是 App 视觉补偿，提交前按短窗口降采样；命令日志、房间状态和主 outbox 不受影响。
			robotDisplayRecords := s.sampleRobotDisplayRecords(result.robotDisplayRecords)
			durableOutboxRecords, directIMRecords := splitRoomDirectIMRecords(outboxRecords, result.suppressDirectIMEventTypes)
			// mutate 可额外返回只服务客户端展示的直发 IM，典型场景是重复 JoinRoom 的入房飘窗；
			// 这类事件不能进入 durable outbox，否则会污染活跃统计和补偿消费。
			directIMRecords = append(directIMRecords, result.directIMRecords...)
			saveStartedAt := time.Now()
			commitCtx, commitCancel := roomMutationCommitContext(ctx)
			resultPayload, err := marshalMutationResult(result)
			if err != nil {
				commitCancel()
				return nil, err
			}
			commandRecord := CommandRecord{
				AppCode:       appcode.FromContext(ctx),
				RoomID:        cmd.RoomID(),
				RoomVersion:   nextState.Version,
				CommandID:     cmd.ID(),
				CommandType:   cmd.Type(),
				Payload:       commandPayloadForRecord(payload, result),
				ResultPayload: resultPayload,
				Replayable:    replayable,
				OwnerNodeID:   s.nodeID,
				LeaseToken:    lease.LeaseToken,
				CreatedAtMS:   now.UnixMilli(),
			}
			err = s.repository.SaveMutation(commitCtx, MutationCommit{
				Command:                         commandRecord,
				OutboxRecords:                   durableOutboxRecords,
				DeleteRoom:                      result.deleteRoom,
				RoomStatus:                      result.roomStatus,
				RoomSeatCount:                   result.roomSeatCount,
				RoomPasswordHash:                result.roomPasswordHash,
				RoomMode:                        result.roomMode,
				VisibleRegionID:                 result.visibleRegionID,
				CloseInfo:                       result.closeInfo,
				RoomUserGiftStats:               result.roomUserGiftStats,
				RoomUserContributionPeriodStats: result.roomUserContributionPeriodStats,
				RoomWeeklyContribution:          result.roomWeeklyContribution,
				GiftOperation:                   giftOperationCommitForResult(cmd, resultPayload, now),
			})
			commitCancel()
			if errors.Is(err, ErrCommandAlreadyCommitted) {
				// MySQL 唯一键是跨进程最后防线。事务已经整体回滚，此分支只读取首次结果，绝不能替换当前 Cell 副本。
				committed, exists, loadErr := s.loadCommandAfterMutationError(ctx, cmd.RoomID(), cmd.ID())
				if loadErr != nil {
					return nil, loadErr
				}
				if !exists {
					return nil, err
				}
				return replayMutationResult(committed, snapshotWithApp(ctx, current.ToProto()))
			}
			if err != nil {
				// COMMIT can succeed in MySQL while its acknowledgement is lost to a connection reset. The gift
				// operation is then already committed and recovery workers correctly stop claiming it, so returning
				// the transport error without advancing this Cell would leave memory permanently behind its owner DB.
				// Only the exact record produced above, including owner lease, room version and typed result, is
				// sufficient proof; all earlier SQL failures still return the original error unchanged.
				if !s.mutationCommitConfirmed(ctx, commandRecord) {
					return nil, err
				}
			}
			saveMutationMS += elapsedMS(saveStartedAt)
			// 保存已提交 outbox 记录，出 Cell 锁后由 outbox worker 异步投递 activity/MQ 业务事实，避免外部慢调用占住房间串行执行队列。
			result.outboxRecords = durableOutboxRecords
			result.robotDisplayRecords = robotDisplayRecords
			result.directIMRecords = directIMRecords
			// MySQL 事务成功提交后再发本进程唤醒信号；信号丢失只影响延迟，不影响后续周期扫描补偿。
			s.notifyOutboxCommitted(durableOutboxRecords)
			// 直接 IM lane 只承载展示事件；RoomRocketIgnited 同时保留 durable outbox，让延迟发射调度继续可重试。
			s.publishDirectIMRecordsBestEffort(ctx, directIMRecords)
			// 机器人展示事件允许极小概率丢失，不写 MySQL outbox；提交成功后后台直发 IM，失败只记录日志。
			s.publishRobotDisplayRecordsBestEffort(ctx, robotDisplayRecords)

			// 持久化成功后再替换内存状态，避免内存提交领先于恢复来源。
			current.ReplaceWith(nextState)
			currentRank.ReplaceWith(nextRank)
			result.snapshot = snapshotWithApp(ctx, current.ToProto())
			result.applied = true
		} else {
			// no-op 命令仍写入 command log，保证相同 command_id 重试可命中幂等，
			// 不同 payload 复用 command_id 会被后续入口拒绝为 CONFLICT。
			if err := s.ensureLeaseStillOwned(ctx, cmd.RoomID(), lease); err != nil {
				return nil, err
			}
			saveStartedAt := time.Now()
			commitCtx, commitCancel := roomMutationCommitContext(ctx)
			commandRecord := CommandRecord{
				AppCode:     appcode.FromContext(ctx),
				RoomID:      cmd.RoomID(),
				RoomVersion: current.Version,
				CommandID:   cmd.ID(),
				CommandType: cmd.Type(),
				Payload:     payload,
				Replayable:  false,
				OwnerNodeID: s.nodeID,
				LeaseToken:  lease.LeaseToken,
				CreatedAtMS: now.UnixMilli(),
			}
			err := s.repository.SaveMutation(commitCtx, MutationCommit{Command: commandRecord})
			commitCancel()
			if err != nil && !s.mutationCommitConfirmed(ctx, commandRecord) {
				return nil, err
			}
			saveMutationMS += elapsedMS(saveStartedAt)
		}

		return result, nil
	})
	if err != nil {
		return mutationResult{}, err
	}

	result := value.(mutationResult)
	if result.applied && result.roomRocketProgress != nil && s.roomRocketProgressCoalescer != nil {
		// 火箭进度合并必须发生在命令日志、快照来源和非火箭 outbox 已提交之后；未提交命令不能泄露展示事件。
		s.roomRocketProgressCoalescer.enqueue(*result.roomRocketProgress)
	}

	if !result.deleteRoom && (result.applied || result.forceSnapshot) && result.snapshot != nil && (result.forceSnapshot || shouldSnapshot(result.snapshot, s.snapshotEveryN)) {
		// 快照失败时返回错误，因为没有快照会增加恢复成本并可能影响 guard 查询。
		snapshotStartedAt := time.Now()
		if err := s.persistSnapshot(ctx, result.snapshot); err != nil {
			return mutationResult{}, err
		}
		snapshotMS = elapsedMS(snapshotStartedAt)
	}
	if result.applied && !result.deleteRoom {
		projectionStartedAt := time.Now()
		// 列表读模型最终一致，失败只记录日志，不回滚已经提交的 Room Cell 命令。
		s.projectRoomListBestEffort(ctx, result.snapshot)
		// 当前房间恢复读模型同样最终一致；权威校验仍以 Room Cell 快照为准。
		s.projectRoomPresenceForCommandBestEffort(ctx, cmd, result.snapshot, result.user)
		// 跨房间贡献榜是 Redis 轻量读模型，不能影响已提交的房间状态和账务事实。
		s.recordRoomGiftLeaderboardBestEffort(ctx, result.roomGiftLeaderboard)
		projectionMS = elapsedMS(projectionStartedAt)
	}

	logRoomCommandTiming(ctx, cmd.RoomID(), cmd.ID(), cmd.Type(), result.applied, elapsedMS(startedAt), idempotencyMS, leaseMS, saveMutationMS, snapshotMS, projectionMS, result.walletDebitMS, 0, 0, len(result.outboxRecords) > 0, result.syncEvent != nil)

	return result, nil
}

func roomMutationCommitContext(ctx context.Context) (context.Context, context.CancelFunc) {
	// 钱包扣费等外部副作用完成后，command log/outbox 是恢复和补偿的权威来源；
	// 这里保留 app_code 等 context values，但不继承 HTTP 断连取消信号，避免已扣费命令在提交阶段被调用方取消打断。
	return context.WithTimeout(context.WithoutCancel(ctx), roomMutationCommitTimeout)
}

func (s *Service) loadCommandAfterMutationError(ctx context.Context, roomID string, commandID string) (CommandRecord, bool, error) {
	// A caller disconnect or the expired recovery-attempt budget must not cancel the one authoritative
	// read that resolves an unknown MySQL COMMIT result. The short independent deadline prevents this
	// safety check from becoming another way to hold the Room Cell indefinitely.
	confirmCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), roomMutationConfirmCommitTimeout)
	defer cancel()
	return s.repository.GetCommand(confirmCtx, roomID, commandID)
}

func (s *Service) mutationCommitConfirmed(ctx context.Context, expected CommandRecord) bool {
	committed, exists, err := s.loadCommandAfterMutationError(ctx, expected.RoomID, expected.CommandID)
	if err != nil || !exists {
		return false
	}
	// Exact byte equality is intentional: SendGift replaces its command payload with the settled wallet
	// facts and stores a protobuf typed result. Idempotency-equivalent input alone cannot prove that this
	// particular nextState/outbox transaction committed under the lease currently owned by the Cell.
	return committed.AppCode == expected.AppCode &&
		committed.RoomID == expected.RoomID &&
		committed.RoomVersion == expected.RoomVersion &&
		committed.CommandID == expected.CommandID &&
		committed.CommandType == expected.CommandType &&
		committed.OwnerNodeID == expected.OwnerNodeID &&
		committed.LeaseToken == expected.LeaseToken &&
		committed.Replayable == expected.Replayable &&
		committed.CreatedAtMS == expected.CreatedAtMS &&
		bytes.Equal(committed.Payload, expected.Payload) &&
		bytes.Equal(committed.ResultPayload, expected.ResultPayload)
}

func (s *Service) commandPayloadForIdempotency(ctx context.Context, cmd command.Command) ([]byte, bool, error) {
	payload, _, seen, err := s.commandRecordForIdempotency(ctx, cmd)
	return payload, seen, err
}

func (s *Service) commandRecordForIdempotency(ctx context.Context, cmd command.Command) ([]byte, CommandRecord, bool, error) {
	// 先序列化当前命令，后续无论首次提交还是冲突校验都使用同一份 payload。
	payload, err := command.Serialize(cmd)
	if err != nil {
		return nil, CommandRecord{}, false, err
	}

	// command_id 在同一 room_id 内唯一；已有记录表示客户端重试或 command_id 被错误复用。
	record, exists, err := s.repository.GetCommand(ctx, cmd.RoomID(), cmd.ID())
	if err != nil {
		return nil, CommandRecord{}, false, err
	}
	if !exists {
		return payload, CommandRecord{}, false, nil
	}
	if cmd.Type() == (command.SetRoomPassword{}).Type() && record.CommandType == (command.SetRoomPassword{}).Type() {
		// 密码命令的持久 payload 只有哈希，必须走专门逻辑比较 transient 明文和已存哈希。
		matched, err := setRoomPasswordIdempotencyMatches(cmd, payload, record.Payload)
		if err != nil {
			return nil, record, true, err
		}
		if !matched {
			return nil, record, true, xerr.New(xerr.Conflict, "command_id payload conflict")
		}
		return payload, record, true, nil
	}
	currentIDPayload, err := command.IdempotencyPayload(payload)
	if err != nil {
		return nil, record, true, err
	}
	storedIDPayload, err := command.IdempotencyPayload(record.Payload)
	if err != nil {
		return nil, record, true, err
	}
	if record.CommandType != cmd.Type() || !bytes.Equal(storedIDPayload, currentIDPayload) {
		// 同一个 command_id 携带不同业务载荷必须显式冲突，不能返回旧命令结果掩盖客户端错误。
		return nil, record, true, xerr.New(xerr.Conflict, "command_id payload conflict")
	}

	return payload, record, true, nil
}

func setRoomPasswordIdempotencyMatches(cmd command.Command, currentPayload []byte, storedPayload []byte) (bool, error) {
	// 先比较去除哈希后的业务载荷，locked/room/actor 等字段不同仍然是冲突。
	currentIDPayload, err := command.IdempotencyPayload(currentPayload)
	if err != nil {
		return false, err
	}
	storedIDPayload, err := command.IdempotencyPayload(storedPayload)
	if err != nil {
		return false, err
	}
	if !bytes.Equal(storedIDPayload, currentIDPayload) {
		return false, nil
	}

	current, ok := setRoomPasswordCommand(cmd)
	if !ok {
		return false, nil
	}
	if !current.Locked {
		// 解锁命令没有密码语义；通用 payload 相同即可视作同一动作重试。
		return true, nil
	}
	stored, err := command.Deserialize((command.SetRoomPassword{}).Type(), storedPayload)
	if err != nil {
		return false, err
	}
	storedPassword, ok := stored.(*command.SetRoomPassword)
	if !ok {
		return false, nil
	}
	// bcrypt 哈希带随机盐，不能直接比较两次请求生成的哈希；只在当前请求内用 transient 明文校验已持久化哈希。
	return roomPasswordMatches(storedPassword.PasswordHash, current.Password), nil
}

func setRoomPasswordCommand(cmd command.Command) (command.SetRoomPassword, bool) {
	// mutateRoom 入口有时传值、有时传指针，幂等判断统一收敛成值对象。
	switch typed := cmd.(type) {
	case command.SetRoomPassword:
		return typed, true
	case *command.SetRoomPassword:
		if typed == nil {
			return command.SetRoomPassword{}, false
		}
		return *typed, true
	default:
		return command.SetRoomPassword{}, false
	}
}

func commandPayloadForRecord(defaultPayload []byte, result mutationResult) []byte {
	if len(result.commandPayload) > 0 {
		// SendGift 等命令需要把外部结算后的确定性结果写进 command log，恢复时不能重算。
		return result.commandPayload
	}
	return defaultPayload
}

func (s *Service) ensureLeaseStillOwned(ctx context.Context, roomID string, lease router.Lease) error {
	ok, err := s.directory.VerifyOwner(ctx, runtimeRoomKeyFromContext(ctx, roomID), s.nodeID, lease.LeaseToken, s.clock.Now())
	if err != nil {
		return err
	}
	if !ok {
		return xerr.New(xerr.Unavailable, "room lease is no longer owned by this node")
	}
	return nil
}

func elapsedMS(start time.Time) int64 {
	return time.Since(start).Milliseconds()
}