package im

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

	"github.com/tidbcloud/AgentTeam/internal/store/taskboard"
	"github.com/tidbcloud/AgentTeam/internal/unread"
)

const taskChangePostCommitTimeout = 10 * time.Second

type taskChangeUnreadRuntime interface {
	ApplyScopeInvalidatedWithDeltas(event unread.ScopeInvalidated) ([]unread.UnreadDelta, error)
}

type taskChangeUnreadDeltaSink interface {
	EmitUnreadUpdated(workspaceID string, delta unread.UnreadDelta)
}

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

// TaskChangeBridge translates transport-neutral taskboard change callbacks into
// streaming-native runtime.task_* live-only events.
type TaskChangeBridge struct {
	emitter       *LiveEventEmitter
	unread        taskChangeUnreadRuntime
	unreadSink    taskChangeUnreadDeltaSink
	taskLifecycle taskChangeLifecycleGuard
}

// NewTaskChangeBridge returns a taskboard.TaskChangeSink that fans task changes
// out as runtime.task_created / runtime.task_updated / runtime.task_deleted
// live-only events.
func NewTaskChangeBridge(emitter *LiveEventEmitter) *TaskChangeBridge {
	if emitter == nil {
		return nil
	}
	return &TaskChangeBridge{emitter: emitter}
}

// WithUnreadRuntime binds Task deletion to the owner-local unread runtime.
// Permanent invalidation and its zero deltas are published before
// runtime.task_deleted, so late outbox callbacks cannot resurrect attention.
func (bridge *TaskChangeBridge) WithUnreadRuntime(runtime taskChangeUnreadRuntime, sink taskChangeUnreadDeltaSink) *TaskChangeBridge {
	if bridge == nil {
		return nil
	}
	bridge.unread = runtime
	bridge.unreadSink = sink
	return bridge
}

// WithTaskScopeLifecycleGuard serializes Task create/update publication against
// Task deletion. A deletion callback is already post-commit and deliberately
// bypasses this guard; the delete transaction itself waits for any guarded
// publication that acquired the same Task lifecycle rows.
func (bridge *TaskChangeBridge) WithTaskScopeLifecycleGuard(guard taskChangeLifecycleGuard) *TaskChangeBridge {
	if bridge == nil {
		return nil
	}
	bridge.taskLifecycle = guard
	return bridge
}

// HandleTaskChange implements taskboard.TaskChangeSink.
func (bridge *TaskChangeBridge) HandleTaskChange(ctx context.Context, event taskboard.TaskChangeEvent) {
	if bridge == nil || bridge.emitter == nil {
		return
	}
	workspaceID := strings.TrimSpace(event.Context.WorkspaceID)
	if workspaceID == "" {
		return
	}
	switch event.EventType {
	case taskboard.TaskChangeCreated:
		publishCtx, cancel := taskChangePostCommitContext(ctx)
		if cancel == nil {
			return
		}
		defer cancel()
		for _, task := range event.Tasks {
			task := task
			bridge.publishForActiveTasks(publishCtx, workspaceID, taskChangeScopeIDs([]taskboard.Task{task}), func() {
				bridge.emitter.EmitTaskCreated(workspaceID, event.ChannelID, []taskboard.Task{task})
			})
		}
	case taskboard.TaskChangeUpdated:
		publishCtx, cancel := taskChangePostCommitContext(ctx)
		if cancel == nil {
			return
		}
		defer cancel()
		bridge.publishForActiveTasks(publishCtx, workspaceID, taskChangeScopeIDs([]taskboard.Task{event.Task}), func() {
			bridge.emitter.EmitTaskUpdated(workspaceID, event.ChannelID, event.Task)
		})
	case taskboard.TaskChangeDeleted:
		scopeID := strings.TrimSpace(event.TaskID)
		if bridge.unread != nil && scopeID != "" {
			deltas, err := bridge.unread.ApplyScopeInvalidatedWithDeltas(unread.ScopeInvalidated{
				WorkspaceID: workspaceID,
				ScopeID:     scopeID,
				Reason:      "task_deleted",
				Permanent:   true,
			})
			if err != nil {
				if bridge.emitter.logger != nil {
					bridge.emitter.logger.Warn(
						"permanently invalidate deleted Task unread",
						slog.String("workspace_id", workspaceID),
						slog.String("scope_id", scopeID),
						slog.Any("error", err),
					)
				}
			} else if bridge.unreadSink != nil {
				for _, delta := range deltas {
					bridge.unreadSink.EmitUnreadUpdated(workspaceID, delta)
				}
			}
		}
		bridge.emitter.EmitTaskDeleted(workspaceID, event.ChannelID, event.TaskID)
	}
}

func taskChangePostCommitContext(parent context.Context) (context.Context, context.CancelFunc) {
	if parent == nil {
		return nil, nil
	}
	// TaskChangeSink runs after the mutation transaction commits. Preserve
	// request-scoped values, but do not let a disconnected caller cancel the
	// final lifecycle check and leave the committed change unpublished.
	return context.WithTimeout(context.WithoutCancel(parent), taskChangePostCommitTimeout)
}

func (bridge *TaskChangeBridge) publishForActiveTasks(
	ctx context.Context,
	workspaceID string,
	scopeIDs []string,
	publish func(),
) {
	if publish == nil {
		return
	}
	if bridge.taskLifecycle == nil {
		publish()
		return
	}
	if ctx == nil || len(scopeIDs) == 0 {
		return
	}
	_, err := bridge.taskLifecycle.WithActiveTaskScopes(ctx, workspaceID, scopeIDs, func() error {
		publish()
		return nil
	})
	if err != nil && bridge.emitter.logger != nil {
		bridge.emitter.logger.Warn(
			"guard Task change publication",
			slog.String("workspace_id", workspaceID),
			slog.Any("scope_ids", scopeIDs),
			slog.Any("error", err),
		)
	}
}

func taskChangeScopeIDs(tasks []taskboard.Task) []string {
	scopeIDs := make([]string, 0, len(tasks))
	seen := make(map[string]struct{}, len(tasks))
	for _, task := range tasks {
		scopeID := strings.TrimSpace(task.TaskScopeID)
		if scopeID == "" {
			// Canonical Task scopes intentionally share the Task ID. Keep this
			// fallback for rolling callbacks created before TaskScopeID was
			// populated on the public Task payload.
			scopeID = strings.TrimSpace(task.ID)
		}
		if scopeID == "" {
			return nil
		}
		if _, exists := seen[scopeID]; exists {
			continue
		}
		seen[scopeID] = struct{}{}
		scopeIDs = append(scopeIDs, scopeID)
	}
	return scopeIDs
}