package im

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"time"

	"github.com/tidbcloud/AgentTeam/internal/api/task"
	storeim "github.com/tidbcloud/AgentTeam/internal/store/im"
	"github.com/tidbcloud/AgentTeam/internal/store/message"
)

// TaskConversationProjector adapts Task detail self mutations into the same
// member-local Inbox stream projection used by generic conversation actions.
type TaskConversationProjector struct {
	conversations      ConversationStore
	streamWriter       StreamEventWriter
	taskScopeLifecycle taskConversationLifecycleGuard
	unreadRuntime      UnreadRuntime
	unreadSink         UnreadDeltaSink
}

type taskConversationLifecycleGuard interface {
	WithActiveTaskScopes(ctx context.Context, workspaceID string, scopeIDs []string, fanout func() error) (bool, error)
}

// taskConversationMaterializer is the optional canonical Task Watch Inbox
// materializer the projector uses when a Watch lands on a Task that has no
// owner-local conversation row yet. *message.Store implements it.
type taskConversationMaterializer interface {
	MaterializeTaskConversationForMember(ctx context.Context, workspaceID string, memberID string, taskScopeID string) error
}

func NewTaskConversationProjector(
	conversations ConversationStore,
	streamWriter StreamEventWriter,
	unreadRuntime UnreadRuntime,
	unreadSink UnreadDeltaSink,
) *TaskConversationProjector {
	projector := &TaskConversationProjector{
		conversations: conversations,
		streamWriter:  streamWriter,
		unreadRuntime: unreadRuntime,
		unreadSink:    unreadSink,
	}
	if guard, ok := conversations.(taskConversationLifecycleGuard); ok {
		projector.taskScopeLifecycle = guard
	}
	return projector
}

func (projector *TaskConversationProjector) ProjectTaskConversationRead(ctx context.Context, projection task.TaskConversationReadProjection) error {
	if projector == nil {
		return nil
	}
	workspaceID := strings.TrimSpace(projection.WorkspaceID)
	userID := strings.TrimSpace(projection.UserID)
	memberID := strings.TrimSpace(projection.MemberID)
	taskID := strings.TrimSpace(projection.TaskID)
	scopeID := strings.TrimSpace(projection.ScopeID)
	if workspaceID == "" || userID == "" || memberID == "" || taskID == "" || scopeID == "" {
		return fmt.Errorf("Task conversation read projection requires workspace, user, member, Task, and scope")
	}
	return projector.withActiveTaskScope(ctx, workspaceID, scopeID, func(projectionCtx context.Context) error {
		return projector.projectTaskConversationRead(projectionCtx, workspaceID, userID, memberID, taskID, scopeID, projection)
	})
}

func (projector *TaskConversationProjector) projectTaskConversationRead(
	ctx context.Context,
	workspaceID string,
	userID string,
	memberID string,
	taskID string,
	scopeID string,
	projection task.TaskConversationReadProjection,
) error {
	// The Task handler has already advanced both in-memory cursors. Emit the
	// resulting owner-local unread state through the same live delta path as a
	// generic Inbox mark-read. Do this after the durable self event is appended:
	// a causally newer message or Activity may be accounted while this method is
	// loading the Inbox projection, and the final live value must repair that
	// interleaving rather than letting the older read action clear it.
	helper := ConversationsHandler{
		unreadRuntime: projector.unreadRuntime,
		unreadSink:    projector.unreadSink,
	}
	defer helper.emitCurrentUnreadState(workspaceID, userID, memberID, scopeID, message.ConversationKindTask)

	if projector.streamWriter == nil {
		return nil
	}
	if projector.conversations == nil {
		return projector.publishTaskConversationReadWithoutIndex(ctx, workspaceID, memberID, taskID, scopeID, projection)
	}
	item, err := projector.conversations.LoadConversationForUser(ctx, userID, workspaceID, scopeID, true)
	if err != nil {
		if errors.Is(err, message.ErrNotFound) {
			return projector.publishTaskConversationReadWithoutIndex(ctx, workspaceID, memberID, taskID, scopeID, projection)
		}
		return fmt.Errorf("load Task conversation after mark read: %w", err)
	}
	item.Kind = message.ConversationKindTask
	item.ScopeID = scopeID
	item.UnreadCount, item.ManualUnread = projector.currentTaskUnreadState(workspaceID, memberID, scopeID)
	item.LastReadSeq = projection.LastMessageReadSeq
	item.LastActivityReadSeq = projection.LastActivityReadSeq
	if err := publishConversationActionEvent(ctx, projector.streamWriter, workspaceID, memberID, message.ConversationActionMarkRead, item); err != nil {
		return fmt.Errorf("publish Task conversation read projection: %w", err)
	}
	return nil
}

func (projector *TaskConversationProjector) publishTaskConversationReadWithoutIndex(
	ctx context.Context,
	workspaceID string,
	memberID string,
	taskID string,
	scopeID string,
	projection task.TaskConversationReadProjection,
) error {
	// Mark-read may win a race with the asynchronous attention projector. Keep
	// the read action durable for other tabs without manufacturing an Inbox row;
	// a later targeted event remains the only operation that materializes one.
	unreadCount, manualUnread := projector.currentTaskUnreadState(workspaceID, memberID, scopeID)
	_, err := projector.streamWriter.AppendMemberEvent(ctx, workspaceID, memberID, scopeID, storeim.StreamEvent{
		Type: imEventTypeConversationRead,
		Payload: map[string]any{
			"action":              message.ConversationActionMarkRead,
			"kind":                message.ConversationKindTask,
			"taskId":              taskID,
			"scopeId":             scopeID,
			"unreadCount":         unreadCount,
			"manualUnread":        manualUnread,
			"lastMessageReadSeq":  projection.LastMessageReadSeq,
			"lastActivityReadSeq": projection.LastActivityReadSeq,
		},
		CreatedAt: time.Now().UTC(),
	}, nil)
	if err != nil {
		return fmt.Errorf("append Task read self stream event: %w", err)
	}
	return nil
}

func (projector *TaskConversationProjector) currentTaskUnreadState(workspaceID string, memberID string, scopeID string) (int64, bool) {
	if projector == nil || projector.unreadRuntime == nil {
		return 0, false
	}
	state, ok := projector.unreadRuntime.ScopeStates(workspaceID, memberID, []string{scopeID})[scopeID]
	if !ok {
		return 0, false
	}
	return state.Count(), state.ManualUnread
}

func (projector *TaskConversationProjector) ProjectTaskConversationPolicy(ctx context.Context, projection task.TaskConversationPolicyProjection) error {
	if projector == nil {
		return nil
	}
	workspaceID := strings.TrimSpace(projection.WorkspaceID)
	userID := strings.TrimSpace(projection.UserID)
	memberID := strings.TrimSpace(projection.MemberID)
	taskID := strings.TrimSpace(projection.TaskID)
	scopeID := strings.TrimSpace(projection.ScopeID)
	state := strings.TrimSpace(projection.State)
	if workspaceID == "" || userID == "" || memberID == "" || taskID == "" || scopeID == "" || state == "" {
		return fmt.Errorf("Task conversation policy projection requires workspace, user, member, Task, scope, and state")
	}
	return projector.withActiveTaskScope(ctx, workspaceID, scopeID, func(projectionCtx context.Context) error {
		return projector.projectTaskConversationPolicy(projectionCtx, workspaceID, userID, memberID, taskID, scopeID, state)
	})
}

func (projector *TaskConversationProjector) projectTaskConversationPolicy(
	ctx context.Context,
	workspaceID string,
	userID string,
	memberID string,
	taskID string,
	scopeID string,
	state string,
) error {
	action := message.ConversationActionMute
	if state == message.HumanScopeSubscriptionStateActive {
		action = message.ConversationActionUnmute
	}

	// Watch changes do not rewrite already-frozen unread recipients. Emit the
	// current value only so owner-local live consumers remain coherent. The
	// repair is intentionally deferred until after the durable policy event:
	// that event is member-local but must never finish a concurrent attention
	// race on a stale unread projection.
	helper := ConversationsHandler{
		unreadRuntime: projector.unreadRuntime,
		unreadSink:    projector.unreadSink,
	}
	defer helper.emitCurrentUnreadState(workspaceID, userID, memberID, scopeID, message.ConversationKindTask)

	if projector.streamWriter == nil {
		return nil
	}
	if projector.conversations != nil {
		item, err := projector.conversations.LoadConversationForUser(ctx, userID, workspaceID, scopeID, true)
		if err == nil {
			item.Kind = message.ConversationKindTask
			item.ScopeID = scopeID
			item.SubscriptionState = state
			if err := publishConversationActionEvent(ctx, projector.streamWriter, workspaceID, memberID, action, item); err != nil {
				return fmt.Errorf("publish Task conversation policy projection: %w", err)
			}
			return nil
		}
		if !errors.Is(err, message.ErrNotFound) {
			return fmt.Errorf("load Task conversation after policy update: %w", err)
		}
	}

	// A user may Watch a visible Task before it has ever become an Inbox
	// candidate. Materialize the canonical owner-local conversation row
	// idempotently (the store path also does this in the same transaction as the
	// watch-state change; this is the post-commit fallback) so the Inbox and
	// realtime consumers immediately see the Task. This never fabricates a
	// discussion message, activity event, InboxMessage, or Agent wake.
	if state == message.HumanScopeSubscriptionStateActive {
		if materializer, ok := projector.conversations.(taskConversationMaterializer); ok && materializer != nil {
			if materializeErr := materializer.MaterializeTaskConversationForMember(ctx, workspaceID, memberID, scopeID); materializeErr != nil {
				return fmt.Errorf("materialize Task conversation after policy update: %w", materializeErr)
			}
		}
	}
	_, err := projector.streamWriter.AppendMemberEvent(ctx, workspaceID, memberID, scopeID, storeim.StreamEvent{
		Type: imEventTypeConversationUpdated,
		Payload: map[string]any{
			"action":            action,
			"kind":              message.ConversationKindTask,
			"taskId":            taskID,
			"scopeId":           scopeID,
			"subscriptionState": state,
		},
		CreatedAt: time.Now().UTC(),
	}, nil)
	if err != nil {
		return fmt.Errorf("append Task policy self stream event: %w", err)
	}
	return nil
}

func (projector *TaskConversationProjector) withActiveTaskScope(
	ctx context.Context,
	workspaceID string,
	scopeID string,
	project func(context.Context) error,
) error {
	if projector.taskScopeLifecycle == nil {
		return project(ctx)
	}
	protectedCtx := storeim.WithTaskLifecycleProtectedScopes(ctx, workspaceID, []string{scopeID})
	_, err := projector.taskScopeLifecycle.WithActiveTaskScopes(ctx, workspaceID, []string{scopeID}, func() error {
		return project(protectedCtx)
	})
	if err != nil {
		return fmt.Errorf("guard Task conversation projection: %w", err)
	}
	return nil
}