package im

import (
	"context"
	"errors"
	"net/http"
	"strings"

	"github.com/tidbcloud/AgentTeam/internal/api"
	"github.com/tidbcloud/AgentTeam/internal/api/dto"
	"github.com/tidbcloud/AgentTeam/internal/api/im/fanout"
	"github.com/tidbcloud/AgentTeam/internal/runtimecontrol"
	messageservice "github.com/tidbcloud/AgentTeam/internal/service/message"
	"github.com/tidbcloud/AgentTeam/internal/store/identity"
	"github.com/tidbcloud/AgentTeam/internal/store/message"
	"github.com/tidbcloud/AgentTeam/internal/store/rewind"
	"github.com/tidbcloud/AgentTeam/internal/unread"
)

type imRewindRestoreError struct {
	err            error
	repairRequired bool
}

func (err *imRewindRestoreError) Error() string {
	if err == nil || err.err == nil {
		return ""
	}
	return err.err.Error()
}

func (err *imRewindRestoreError) Unwrap() error {
	if err == nil {
		return nil
	}
	return err.err
}

// RewindHandler owns conversation rewind orchestration for Workspace IM.
type RewindHandler struct {
	store         RewindStore
	hub           RewindHub
	streamWriter  StreamEventWriter
	noticeWriter  HumanMessageWriter
	unreadRuntime fanout.UnreadRuntime
	unreadSink    fanout.UnreadDeltaSink
	logger        interface{ Warn(msg string, args ...any) }
}

// Rewind handles POST /api/v1/workspaces/:workspaceId/messages/:messageId/rewind.
// The mutation lives on the Workspace IM surface; browser invalidation is driven
// by im_stream_events rather than legacy Socket.IO conversation:rewound events.
func (handler *RewindHandler) Rewind(r api.WorkspaceScopedRequest) (api.Response, error) {
	if handler.store == nil {
		return api.Response{}, api.NewHTTPError(http.StatusNotImplemented, "rewind is not supported")
	}
	messageID, err := routeStandaloneMessageID(r)
	if err != nil {
		return api.Response{}, err
	}
	if _, err := handler.rewindMessage(r.Context(), r.User.User, r.WorkspaceID(), messageID, ""); err != nil {
		return api.Response{}, err
	}
	return api.OK(dto.AppMessageRewindResponse{OK: true}), nil
}

func (handler *RewindHandler) rewindMessage(ctx context.Context, user identity.User, workspaceID string, messageID string, expectedScopeID string) (rewind.Plan, error) {
	if handler.store == nil {
		return rewind.Plan{}, api.NewHTTPError(http.StatusNotImplemented, "rewind is not supported")
	}
	plan, err := handler.store.PrepareRewindForUser(ctx, user.ID, workspaceID, messageID)
	if err != nil {
		return rewind.Plan{}, err
	}
	if expected := strings.TrimSpace(expectedScopeID); expected != "" && strings.TrimSpace(plan.RewindScopeID()) != expected {
		return rewind.Plan{}, api.NewHTTPError(http.StatusNotFound, "message not found")
	}
	operationID, err := handler.store.BeginRewindOperation(ctx, plan)
	if err != nil {
		return rewind.Plan{}, err
	}
	markFailed := func(reason string) {
		if markErr := handler.store.MarkRewindOperationFailed(context.Background(), operationID, reason); markErr != nil && handler.logger != nil {
			handler.logger.Warn("mark rewind failed", "operation_id", operationID, "error", markErr)
		}
	}
	markRepairRequired := func(reason string) {
		if markErr := handler.store.MarkRewindOperationRepairRequired(context.Background(), operationID, reason); markErr != nil && handler.logger != nil {
			handler.logger.Warn("mark rewind repair required", "operation_id", operationID, "error", markErr)
		}
	}
	if err := handler.store.ValidateRewindPlan(ctx, plan); err != nil {
		markFailed(err.Error())
		return rewind.Plan{}, err
	}
	if (len(plan.RelatedAgents) > 0 || len(plan.PendingInterruptAgents) > 0) && handler.hub == nil {
		markFailed("daemon hub is unavailable")
		return rewind.Plan{}, api.NewHTTPError(http.StatusServiceUnavailable, "rewind is temporarily unavailable")
	}
	adjustedRuntime := len(plan.RelatedAgents) > 0
	if err := handler.restoreRewindRelatedAgents(ctx, plan.RelatedAgents); err != nil {
		var restoreErr *imRewindRestoreError
		if errors.As(err, &restoreErr) && restoreErr.repairRequired {
			markRepairRequired(err.Error())
		} else {
			markFailed(err.Error())
		}
		return rewind.Plan{}, api.NewHTTPError(http.StatusServiceUnavailable, "rewind restore failed: "+err.Error())
	}
	if err := handler.interruptRewindPendingAgents(ctx, plan.PendingInterruptAgents); err != nil {
		var interruptErr *imRewindRestoreError
		if adjustedRuntime || (errors.As(err, &interruptErr) && interruptErr.repairRequired) {
			markRepairRequired(err.Error())
		} else {
			markFailed(err.Error())
		}
		return rewind.Plan{}, api.NewHTTPError(http.StatusServiceUnavailable, "rewind interrupt failed: "+err.Error())
	}
	adjustedRuntime = adjustedRuntime || len(plan.PendingInterruptAgents) > 0
	if err := handler.store.ApplyRewind(ctx, operationID, plan); err != nil {
		if !adjustedRuntime {
			markFailed(err.Error())
			return rewind.Plan{}, err
		}
		markRepairRequired(err.Error())
		return rewind.Plan{}, api.NewHTTPError(http.StatusServiceUnavailable, "rewind apply failed after restore")
	}
	handler.applyMessageRewound(workspaceID, plan)
	if err := publishConversationRewoundEvent(ctx, handler.streamWriter, workspaceID, plan); err != nil {
		return rewind.Plan{}, err
	}
	if err := publishMessagePinInvalidationsForRewind(ctx, handler.streamWriter, workspaceID, plan); err != nil {
		return rewind.Plan{}, err
	}
	if err := handler.publishRewindNotice(ctx, user, workspaceID, operationID, plan); err != nil && handler.logger != nil {
		handler.logger.Warn("publish rewind notice", "operation_id", operationID, "error", err)
	}
	return plan, nil
}

func (handler *RewindHandler) publishRewindNotice(ctx context.Context, user identity.User, workspaceID string, operationID string, plan rewind.Plan) error {
	if handler == nil || handler.noticeWriter == nil {
		return nil
	}
	scopeID := strings.TrimSpace(plan.RewindScopeID())
	if scopeID == "" {
		return nil
	}
	actorDisplayName := rewindNoticeActorDisplayName(user)
	appMessage, delivery, err := handler.noticeWriter.SendHumanMessage(ctx, user, workspaceID, scopeID, rewindNoticeContent, nil, messageservice.SendOptions{
		MessageType:     message.MessageTypeSystem,
		Metadata:        message.BuildRewindSystemNoticeMetadata(actorDisplayName),
		ClientMessageID: rewindNoticeClientMessageID(operationID),
	})
	if err != nil {
		return err
	}
	if delivery.IdempotentReplay {
		return nil
	}
	delivery.SuppressBrowserNotification = true
	return publishMessageCreatedEvent(ctx, handler.streamWriter, workspaceID, scopeID, appMessage, delivery)
}

const rewindNoticeContent = "system.rewind.conversation"

func rewindNoticeActorDisplayName(user identity.User) string {
	name := firstNonEmptyString(user.DisplayName, user.Name, user.Email, user.ID)
	if name == "" {
		name = "user"
	}
	return name
}

func rewindNoticeClientMessageID(operationID string) string {
	trimmedOperationID := strings.TrimSpace(operationID)
	if trimmedOperationID == "" {
		return ""
	}
	return "rewind-notice:" + trimmedOperationID
}

func (handler *RewindHandler) applyMessageRewound(workspaceID string, plan rewind.Plan) {
	if handler == nil || handler.unreadRuntime == nil {
		return
	}
	event := unread.MessageRewound{
		WorkspaceID:      strings.TrimSpace(workspaceID),
		RewindGeneration: rewindGenerationForPlan(plan),
	}
	for _, scope := range plan.RewoundScopes {
		runtimeScope := unread.RewoundScope{
			ScopeID:   strings.TrimSpace(scope.ScopeID),
			LatestSeq: rewindLatestSeq(scope.LatestSeq),
		}
		for _, removed := range scope.RemovedMessages {
			if !message.IsHumanVisibleMessageType(removed.MessageType) {
				continue
			}
			runtimeScope.RemovedMessages = append(runtimeScope.RemovedMessages, unread.RewoundMessage{
				ScopeID:                 strings.TrimSpace(firstNonEmpty(removed.ScopeID, scope.ScopeID)),
				Seq:                     removed.Seq,
				AuthorMemberID:          strings.TrimSpace(removed.AuthorMemberID),
				NotifyPrincipalMemberID: strings.TrimSpace(removed.NotifyPrincipalMemberID),
				AllowAuthorUnread:       removed.AllowAuthorUnread,
				MessageType:             strings.TrimSpace(removed.MessageType),
				CreatedAt:               removed.CreatedAt,
			})
		}
		if runtimeScope.ScopeID != "" {
			event.Scopes = append(event.Scopes, runtimeScope)
		}
	}
	if len(event.Scopes) == 0 {
		handler.applyScopeInvalidated(workspaceID, plan.RewindScopeID(), "message_rewind_missing_facts")
		return
	}
	result, _ := handler.unreadRuntime.ApplyMessageRewound(event)
	handler.emitUnreadDeltas(workspaceID, result.Deltas)
}

func recallRewindStateFromPlan(plan rewind.Plan) *dto.AppMessageRecallRewindState {
	state := &dto.AppMessageRecallRewindState{
		ScopeID:           strings.TrimSpace(plan.RewindScopeID()),
		MessageID:         strings.TrimSpace(plan.TargetMessageID),
		FromSeq:           plan.TargetSeq,
		RewoundMessageIDs: append([]string(nil), plan.SuffixMessageIDs...),
	}
	rewoundIDsByScope := map[string][]string{
		strings.TrimSpace(plan.RewindScopeID()): append([]string(nil), plan.SuffixMessageIDs...),
	}
	for _, thread := range plan.CascadeReplyThreads {
		scopeID := strings.TrimSpace(thread.CascadeScopeID())
		if scopeID == "" {
			continue
		}
		rewoundIDsByScope[scopeID] = append([]string(nil), thread.MessageIDs...)
	}
	emittedScopes := make(map[string]struct{}, len(plan.RewoundScopes))
	for _, rewoundScope := range plan.RewoundScopes {
		scope := dto.AppMessageRecallRewoundScope{
			ScopeID:           strings.TrimSpace(rewoundScope.ScopeID),
			LatestSeq:         rewoundScope.LatestSeq,
			RewoundMessageIDs: append([]string(nil), rewoundIDsByScope[strings.TrimSpace(rewoundScope.ScopeID)]...),
		}
		for _, removed := range rewoundScope.RemovedMessages {
			if strings.TrimSpace(removed.ScopeID) != "" && strings.TrimSpace(scope.ScopeID) == "" {
				scope.ScopeID = strings.TrimSpace(removed.ScopeID)
			}
		}
		if scope.ScopeID != "" {
			emittedScopes[scope.ScopeID] = struct{}{}
			state.RewoundScopes = append(state.RewoundScopes, scope)
		}
	}
	for scopeID, messageIDs := range rewoundIDsByScope {
		if _, ok := emittedScopes[scopeID]; ok {
			continue
		}
		scope := dto.AppMessageRecallRewoundScope{
			ScopeID:           scopeID,
			LatestSeq:         0,
			RewoundMessageIDs: append([]string(nil), messageIDs...),
		}
		state.RewoundScopes = append(state.RewoundScopes, scope)
	}
	return state
}

func (handler *RewindHandler) applyScopeInvalidated(workspaceID string, scopeID string, reason string) {
	if handler == nil || handler.unreadRuntime == nil {
		return
	}
	_ = handler.unreadRuntime.ApplyScopeInvalidated(unread.ScopeInvalidated{
		WorkspaceID: strings.TrimSpace(workspaceID),
		ScopeID:     strings.TrimSpace(scopeID),
		Reason:      reason,
	})
}

func (handler *RewindHandler) emitUnreadDeltas(workspaceID string, deltas []unread.UnreadDelta) {
	if handler == nil || handler.unreadSink == nil {
		return
	}
	for _, delta := range deltas {
		handler.unreadSink.EmitUnreadUpdated(workspaceID, delta)
	}
}

func publishMessagePinInvalidationsForRewind(ctx context.Context, writer StreamEventWriter, workspaceID string, plan rewind.Plan) error {
	if writer == nil {
		return nil
	}
	messageIDsByScope := make(map[string][]string)
	scopeOrder := []string{}
	addScopeMessages := func(scopeID string, messageIDs []string) {
		scopeID = strings.TrimSpace(scopeID)
		if scopeID == "" {
			return
		}
		if _, ok := messageIDsByScope[scopeID]; !ok {
			scopeOrder = append(scopeOrder, scopeID)
		}
		messageIDsByScope[scopeID] = append(messageIDsByScope[scopeID], messageIDs...)
	}
	addScopeMessages(plan.RewindScopeID(), plan.SuffixMessageIDs)
	for _, thread := range plan.CascadeReplyThreads {
		cascadeScopeID := thread.CascadeScopeID()
		addScopeMessages(cascadeScopeID, thread.MessageIDs)
	}
	for _, scopeID := range scopeOrder {
		messageIDs := messageIDsByScope[scopeID]
		if err := publishMessagePinsInvalidatedEvent(ctx, writer, workspaceID, scopeID, messageIDs); err != nil {
			return err
		}
	}
	return nil
}

func rewindLatestSeq(seq int64) int64 {
	if seq < 0 {
		return 0
	}
	return seq
}

func rewindGenerationForPlan(plan rewind.Plan) int64 {
	var generation int64
	for _, scope := range plan.RewoundScopes {
		for _, message := range scope.RemovedMessages {
			if message.CreatedAt.IsZero() {
				continue
			}
			generation = maxInt64(generation, message.CreatedAt.UnixNano())
		}
	}
	if generation > 0 {
		return generation
	}
	if !plan.TargetCreatedAt.IsZero() {
		return maxInt64(plan.TargetCreatedAt.UnixNano(), 0)
	}
	return maxInt64(plan.TargetSeq, 0)
}

func maxInt64(left, right int64) int64 {
	if left > right {
		return left
	}
	return right
}

func (handler *RewindHandler) restoreRewindRelatedAgents(ctx context.Context, relatedAgents []rewind.AgentPlan) error {
	return restoreIMRewindAgents(ctx, relatedAgents, func(ctx context.Context, relatedAgent rewind.AgentPlan) error {
		return handler.hub.RestoreAgentRecoveryPoint(ctx, relatedAgent.MachineID, runtimecontrol.RestoreAgentRecoveryPointRequest{
			AgentID:     relatedAgent.AgentID,
			Runtime:     relatedAgent.Runtime,
			Model:       relatedAgent.Model,
			SessionID:   relatedAgent.SessionID,
			Strategy:    relatedAgent.RecoveryStrategy,
			RecoveryRef: relatedAgent.RecoveryRef,
		})
	})
}

func (handler *RewindHandler) interruptRewindPendingAgents(_ context.Context, pendingAgents []rewind.AgentPlan) error {
	interruptedAny := false
	for _, pendingAgent := range pendingAgents {
		if err := handler.hub.InterruptAgentRun(pendingAgent.MachineID, pendingAgent.AgentID); err != nil {
			return &imRewindRestoreError{err: err, repairRequired: interruptedAny}
		}
		interruptedAny = true
	}
	return nil
}

func restoreIMRewindAgents(ctx context.Context, relatedAgents []rewind.AgentPlan, restore func(context.Context, rewind.AgentPlan) error) error {
	restoredAny := false
	for _, relatedAgent := range relatedAgents {
		if err := restore(ctx, relatedAgent); err != nil {
			return &imRewindRestoreError{err: err, repairRequired: restoredAny}
		}
		restoredAny = true
	}
	return nil
}