package server

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

	"github.com/tidbcloud/AgentTeam/internal/observability/logging"
	"github.com/tidbcloud/AgentTeam/internal/service/automation"
	contentstoreservice "github.com/tidbcloud/AgentTeam/internal/service/content"
	contentstorestore "github.com/tidbcloud/AgentTeam/internal/store/content"
)

const contentOutboxPollInterval = time.Second

type contentOutboxRunner interface {
	RunOnce(context.Context, int) (contentstoreservice.OutboxWorkerResult, error)
}

type versionCommittedConsumers []contentstoreservice.VersionCommittedConsumer

func (consumers versionCommittedConsumers) ConsumeVersionCommitted(ctx context.Context, event contentstoreservice.VersionCommittedEvent) error {
	for _, consumer := range consumers {
		if consumer == nil {
			continue
		}
		if err := consumer.ConsumeVersionCommitted(ctx, event); err != nil {
			return err
		}
	}
	return nil
}

type authorityGatedVersionCommittedConsumers struct {
	content interface {
		GetContent(context.Context, string, string) (contentstoreservice.Content, error)
		GetVersion(context.Context, string, string, string) (contentstoreservice.Version, error)
	}
	authority interface {
		GetAuthority(context.Context, string, string, string) (contentstorestore.Authority, bool, error)
	}
	automationConsumers versionCommittedConsumers
	otherConsumers      versionCommittedConsumers
}

// ConsumeVersionCommitted opens the Automation/Activity group only after the
// exact product binding is native-active. Missing/legacy authority is retried,
// not acknowledged, so dormant migration publications cannot execute or emit
// Activity before the one-way activation gate.
func (group authorityGatedVersionCommittedConsumers) ConsumeVersionCommitted(ctx context.Context, event contentstoreservice.VersionCommittedEvent) error {
	if group.content == nil || group.authority == nil {
		return fmt.Errorf("content version consumer authority is not configured")
	}
	content, err := group.content.GetContent(ctx, event.WorkspaceID, event.ContentID)
	if err != nil {
		return fmt.Errorf("load content version consumer binding: %w", err)
	}
	authority, found, err := group.authority.GetAuthority(ctx, content.WorkspaceID, content.Domain, content.BindingKey)
	if err != nil {
		return fmt.Errorf("resolve content version consumer authority: %w", err)
	}
	if !found || authority.State != contentstorestore.AuthorityStateNativeActive || authority.Generation != content.Generation {
		return fmt.Errorf("content version consumer binding is not native-active")
	}
	if content.Domain != "automation" {
		return group.otherConsumers.ConsumeVersionCommitted(ctx, event)
	}
	version, err := group.content.GetVersion(ctx, event.WorkspaceID, event.ContentID, event.VersionID)
	if err != nil {
		return fmt.Errorf("load content version consumer principal: %w", err)
	}
	ctx = automation.WithAutomationContentAccess(ctx, content.WorkspaceID, content.BindingKey, content.ID, content.Generation, version.CreatedBy,
		contentstoreservice.AccessReadMetadata, contentstoreservice.AccessReadPayload)
	return group.automationConsumers.ConsumeVersionCommitted(ctx, event)
}

func startContentOutboxWorker(worker contentOutboxRunner) (context.CancelFunc, <-chan struct{}) {
	if worker == nil {
		return nil, nil
	}
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan struct{})
	logger := slog.Default().With(slog.String("domain", "content_store"), slog.String("component", "outbox"))
	logging.Go(logger, func() {
		defer close(done)
		run := func() {
			if _, err := worker.RunOnce(ctx, contentstorestore.MaxOutboxLeaseBatch); err != nil && ctx.Err() == nil {
				logger.ErrorContext(ctx, "content outbox delivery failed", slog.String("error_class", "worker_failed"))
			}
		}
		run()
		ticker := time.NewTicker(contentOutboxPollInterval)
		defer ticker.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
				run()
			}
		}
	})
	return cancel, done
}