package fanout

import (
	"context"
	"fmt"
	"log/slog"
	"os"
	"strings"
	"time"

	"github.com/tidbcloud/AgentTeam/internal/coredb"
	"github.com/tidbcloud/AgentTeam/internal/observability/logging"
	"github.com/tidbcloud/AgentTeam/internal/store/message"
)

const (
	defaultScopeFanoutInterval = time.Second
	defaultScopeFanoutLimit    = 50
)

// ScopeFanoutReplayAdapter replays persisted message fan-out payloads through
// the same delivery path used by the synchronous request handlers.
type ScopeFanoutReplayAdapter struct {
	fanout            DeliveryFanout
	wakeStatusEmitter WorkflowWakeStatusEmitter
}

func NewScopeFanoutReplayAdapter(fanout DeliveryFanout) ScopeFanoutReplayAdapter {
	return ScopeFanoutReplayAdapter{fanout: fanout}
}

// WorkflowWakeStatusEmitter projects a final durable Workflow delivery
// failure into the existing browser wake lifecycle.
type WorkflowWakeStatusEmitter interface {
	EmitAgentWakeStatusForScope(workspaceID string, scopeID string, agentID string, messageID string, wakeToken string, target string, status string, replyMessageID string, detail string, timestamp string)
}

func (adapter ScopeFanoutReplayAdapter) FanoutFailed(_ context.Context, outbox message.ScopeFanoutOutboxRecord, cause error) {
	if adapter.wakeStatusEmitter == nil ||
		strings.TrimSpace(outbox.EventRefType) != message.DeliverySourceRefMessage ||
		strings.TrimSpace(outbox.FanoutType) != message.ScopeOutboxFanoutScope {
		return
	}
	payload, err := message.UnmarshalScopeFanoutPayload(outbox.Payload)
	if err != nil || (strings.TrimSpace(payload.AppMessage.ChannelType) != message.WorkflowScopeType && !payload.Delivery.RequireLiveHandoff) {
		return
	}
	messageID := strings.TrimSpace(outbox.MessageID)
	if messageID == "" {
		messageID = strings.TrimSpace(payload.AppMessage.ID)
	}
	scopeID := strings.TrimSpace(outbox.ScopeID)
	if scopeID == "" {
		scopeID = strings.TrimSpace(payload.AppMessage.ChannelID)
	}
	if messageID == "" || scopeID == "" {
		return
	}
	detail := ""
	if cause != nil {
		detail = strings.TrimSpace(cause.Error())
	}
	target := ""
	if planID := strings.TrimSpace(payload.Delivery.Message.ChannelName); planID != "" {
		target = "workflow:" + planID
	}
	seen := make(map[string]struct{}, len(payload.Delivery.WakeRecipients))
	for _, recipient := range payload.Delivery.WakeRecipients {
		agentID := strings.TrimSpace(recipient.AgentID)
		if agentID == "" {
			continue
		}
		if _, exists := seen[agentID]; exists {
			continue
		}
		seen[agentID] = struct{}{}
		adapter.wakeStatusEmitter.EmitAgentWakeStatusForScope(
			strings.TrimSpace(outbox.WorkspaceID),
			scopeID,
			agentID,
			messageID,
			"workflow:"+messageID,
			target,
			"failed",
			"",
			detail,
			time.Now().UTC().Format(time.RFC3339Nano),
		)
	}
}

func (adapter ScopeFanoutReplayAdapter) Fanout(ctx context.Context, outbox message.ScopeFanoutOutboxRecord) error {
	switch strings.TrimSpace(outbox.FanoutType) {
	case message.ScopeOutboxFanoutMessageMutation:
		return adapter.fanoutMessageMutationPayload(ctx, outbox)
	case message.ScopeOutboxFanoutScope:
	default:
		return fmt.Errorf("unsupported scope fanout type %q", outbox.FanoutType)
	}
	switch strings.TrimSpace(outbox.EventRefType) {
	case message.DeliverySourceRefMessage:
		return adapter.fanoutMessagePayload(ctx, outbox)
	case message.DeliverySourceRefActivityEvent:
		return adapter.fanoutDeliveryIntentPayload(ctx, outbox)
	default:
		return fmt.Errorf("unsupported scope fanout event ref type %q", outbox.EventRefType)
	}
}

func (adapter ScopeFanoutReplayAdapter) fanoutMessageMutationPayload(ctx context.Context, outbox message.ScopeFanoutOutboxRecord) error {
	if strings.TrimSpace(outbox.EventRefType) != message.ScopeMessageMutationEventRefType || strings.TrimSpace(outbox.EventRefID) == "" {
		return fmt.Errorf("message mutation outbox source ref is invalid")
	}
	payload, err := message.UnmarshalScopeMessageMutationFanoutPayload(outbox.Payload)
	if err != nil {
		return err
	}
	if strings.TrimSpace(payload.Message.ID) != strings.TrimSpace(outbox.MessageID) || strings.TrimSpace(payload.Message.ChannelID) != strings.TrimSpace(outbox.ScopeID) {
		return fmt.Errorf("message mutation outbox payload target mismatch")
	}
	_, err = adapter.fanout.withActiveTaskScopes(
		ctx,
		outbox.WorkspaceID,
		taskDeliveryTreeScopeIDs(payload.Message, message.MessageDelivery{}),
		func(deliveryCtx context.Context) error {
			switch payload.Kind {
			case message.ScopeMessageMutationUpdate:
				return adapter.fanout.emitMessageEditedStrict(deliveryCtx, outbox.WorkspaceID, payload.Message, outbox.EventRefID)
			case message.ScopeMessageMutationRecall:
				return adapter.fanout.emitMessageRecalledStrict(deliveryCtx, outbox.WorkspaceID, payload.Message, payload.UpdateConversation, outbox.EventRefID)
			default:
				return fmt.Errorf("unsupported message mutation kind %q", payload.Kind)
			}
		},
	)
	return err
}

func (adapter ScopeFanoutReplayAdapter) fanoutMessagePayload(ctx context.Context, outbox message.ScopeFanoutOutboxRecord) error {
	payload, err := message.UnmarshalScopeFanoutPayload(outbox.Payload)
	if err != nil {
		return err
	}
	delivery := payload.Delivery
	// The adapter is, by definition, replaying a durable outbox row. Ordinary
	// messages suppress browser/unread re-projection; targeted Task messages
	// deliberately override that suppression with stable event ids and frozen
	// recipients to repair a post-commit crash window.
	delivery.IdempotentReplay = true
	appMessage := payload.AppMessage
	delivery.AppMessage = &appMessage
	taskScopeIDs := taskDeliveryTreeScopeIDs(appMessage, delivery)
	workflowDelivery := strings.TrimSpace(appMessage.ChannelType) == message.WorkflowScopeType || delivery.RequireLiveHandoff
	if (len(taskScopeIDs) > 0 || workflowDelivery) && adapter.fanout.scopeFanoutLeases != nil {
		if len(taskScopeIDs) > 0 {
			if err := validateClaimedTaskFanout(outbox, outbox.WorkspaceID, appMessage, delivery); err != nil {
				return err
			}
		} else if err := validateClaimedWorkflowFanout(outbox, appMessage); err != nil {
			return err
		}
		ctx = withScopeFanoutLeaseAttempt(ctx, outbox)
	}
	return adapter.fanout.deliverCommittedMessage(ctx, outbox.WorkspaceID, appMessage, delivery)
}

func validateClaimedWorkflowFanout(outbox message.ScopeFanoutOutboxRecord, appMessage message.AppMessage) error {
	if strings.TrimSpace(outbox.MessageID) == "" ||
		strings.TrimSpace(outbox.MessageID) != strings.TrimSpace(appMessage.ID) ||
		strings.TrimSpace(outbox.ScopeID) == "" ||
		strings.TrimSpace(outbox.ScopeID) != strings.TrimSpace(appMessage.ChannelID) {
		return fmt.Errorf("claimed Workflow fanout payload target mismatch")
	}
	return nil
}

func (adapter ScopeFanoutReplayAdapter) fanoutDeliveryIntentPayload(ctx context.Context, outbox message.ScopeFanoutOutboxRecord) error {
	if strings.TrimSpace(outbox.MessageID) != "" {
		return fmt.Errorf("non-message scope delivery intent outbox must not include message_id")
	}
	payload, err := message.UnmarshalScopeDeliveryIntentPayload(outbox.Payload)
	if err != nil {
		return err
	}
	if strings.TrimSpace(payload.SourceRef.Type) != strings.TrimSpace(outbox.EventRefType) ||
		strings.TrimSpace(payload.SourceRef.ID) != strings.TrimSpace(outbox.EventRefID) {
		return fmt.Errorf("scope delivery intent payload source ref mismatch")
	}
	if isTaskDeliveryIntentPayload(outbox, payload) && adapter.fanout.scopeFanoutLeases != nil {
		ctx = withScopeFanoutLeaseAttempt(ctx, outbox)
	}
	return adapter.fanout.deliverDeliveryIntent(ctx, outbox.WorkspaceID, payload)
}

func isTaskDeliveryIntentPayload(outbox message.ScopeFanoutOutboxRecord, payload message.ScopeDeliveryIntentPayload) bool {
	if strings.TrimSpace(payload.Message.ChannelType) == "task" {
		return true
	}
	return strings.HasPrefix(strings.TrimSpace(outbox.ScopeID), "tsk_")
}

type ScopeFanoutRunner struct {
	worker   message.ScopeFanoutWorker
	logger   *slog.Logger
	interval time.Duration
	limit    int
}

// ScopeFanoutRunnerOptions bounds replay ownership for workspace-profile
// servers. Empty WorkspaceGroupID preserves global all-profile behavior.
type ScopeFanoutRunnerOptions struct {
	WorkspaceGroupID  string
	WakeStatusEmitter WorkflowWakeStatusEmitter
}

func NewScopeFanoutRunner(db coredb.DB, fanout DeliveryFanout, logger *slog.Logger) *ScopeFanoutRunner {
	return NewScopeFanoutRunnerWithOptions(db, fanout, logger, ScopeFanoutRunnerOptions{})
}

func NewScopeFanoutRunnerWithOptions(db coredb.DB, fanout DeliveryFanout, logger *slog.Logger, options ScopeFanoutRunnerOptions) *ScopeFanoutRunner {
	if db == nil {
		return nil
	}
	adapter := NewScopeFanoutReplayAdapter(fanout)
	adapter.wakeStatusEmitter = options.WakeStatusEmitter
	worker := message.NewScopeFanoutWorkerWithOptions(db, adapter, scopeFanoutLeaseOwner(), message.ScopeFanoutWorkerOptions{
		WorkspaceGroupID: options.WorkspaceGroupID,
	})
	return &ScopeFanoutRunner{
		worker:   worker,
		logger:   logger,
		interval: defaultScopeFanoutInterval,
		limit:    defaultScopeFanoutLimit,
	}
}

func (runner *ScopeFanoutRunner) Start(ctx context.Context) {
	if runner == nil {
		return
	}
	logging.Go(runner.logger, func() { runner.run(ctx) })
}

func (runner *ScopeFanoutRunner) run(ctx context.Context) {
	runner.runOnce(ctx)
	interval := runner.interval
	if interval <= 0 {
		interval = defaultScopeFanoutInterval
	}
	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			runner.runOnce(ctx)
		}
	}
}

func (runner *ScopeFanoutRunner) runOnce(ctx context.Context) {
	limit := runner.limit
	if limit <= 0 {
		limit = defaultScopeFanoutLimit
	}
	result, err := runner.worker.RunOnce(ctx, limit)
	if err != nil {
		if runner.logger != nil {
			runner.logger.Warn("run scope fanout outbox", slog.Any("error", err))
		}
		return
	}
	if runner.logger != nil && (result.Delivered > 0 || result.Retried > 0 || result.Failed > 0 || result.Suppressed > 0) {
		runner.logger.Debug("ran scope fanout outbox",
			slog.Int("leased", result.Leased),
			slog.Int("delivered", result.Delivered),
			slog.Int("retried", result.Retried),
			slog.Int("failed", result.Failed),
			slog.Int("suppressed", result.Suppressed),
		)
	}
}

func scopeFanoutLeaseOwner() string {
	hostname, err := os.Hostname()
	if err != nil || strings.TrimSpace(hostname) == "" {
		hostname = "agentteam-server"
	}
	return fmt.Sprintf("%s:%d", strings.TrimSpace(hostname), os.Getpid())
}