package server

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"net/http"
	"strings"
	"time"

	"github.com/tidbcloud/AgentTeam/internal/api/identity/lookup"
	"github.com/tidbcloud/AgentTeam/internal/api/im/fanout"
	"github.com/tidbcloud/AgentTeam/internal/config"
	"github.com/tidbcloud/AgentTeam/internal/coredb"
	"github.com/tidbcloud/AgentTeam/internal/inprochttp"
	"github.com/tidbcloud/AgentTeam/internal/kbgovernance"
	"github.com/tidbcloud/AgentTeam/internal/meshxassets"
	"github.com/tidbcloud/AgentTeam/internal/meshxruntime"
	"github.com/tidbcloud/AgentTeam/internal/observability"
	"github.com/tidbcloud/AgentTeam/internal/observability/logging"
	"github.com/tidbcloud/AgentTeam/internal/router"
	"github.com/tidbcloud/AgentTeam/internal/schedule"
	"github.com/tidbcloud/AgentTeam/internal/service/approvalapp"
	"github.com/tidbcloud/AgentTeam/internal/service/appruntime"
	"github.com/tidbcloud/AgentTeam/internal/service/knowledgeqa"
	"github.com/tidbcloud/AgentTeam/internal/service/role/execution"
	"github.com/tidbcloud/AgentTeam/internal/service/serviceaccount"
	"github.com/tidbcloud/AgentTeam/internal/store"
	"github.com/tidbcloud/AgentTeam/internal/store/agentops"
	"github.com/tidbcloud/AgentTeam/internal/store/serviceactor"
	"github.com/tidbcloud/AgentTeam/internal/unread"
	"github.com/tidbcloud/AgentTeam/internal/workflow/invocation"
	"github.com/tidbcloud/AgentTeam/internal/workflow/orchestrator"
	"github.com/tidbcloud/AgentTeam/internal/workspace/owner"
)

// Server wires together infrastructure and the HTTP server.
type Server struct {
	// Configuration and logging.
	config  config.Config
	logger  *slog.Logger
	profile router.Profile

	// Core infrastructure.
	db                coredb.AdminDB // owned database connection, closed on Shutdown
	stores            *store.Graph
	managedWorkspaces owner.ManagedWorkspaceSet

	// Background workers.
	schedules                    *schedule.Service // schedule automation execution
	workflowInvocationSchedules  *invocation.ScheduleWorker
	scopeFanoutRunner            *fanout.ScopeFanoutRunner // message fan-out outbox replay
	unreadRuntime                *unread.Service           // owner-local unread memory lifecycle
	assetAccessWorker            *meshxassets.Worker       // MeshX Memory/Skill access event replay
	workflowRunWorker            *orchestrator.Worker      // first-class PlanVersion graph runtime
	internalAppDispatcher        *appruntime.InternalDispatcher
	approvalProjectionReconciler *approvalapp.ProjectionReconciler
	platformServiceAccounts      *serviceaccount.PlatformService
	knowledgeQAWorker            *knowledgeqa.Worker
	operationWorker              *agentops.OperationWorker
	roleManagementWorker         *execution.Worker
	embeddedMeshX                embeddedMeshXRuntime
	meshXWorkspaceScope          func(context.Context) ([]string, error)
	closers                      []func()

	// HTTP server.
	httpServer    *http.Server
	observability *observability.Provider
}

type embeddedMeshXRuntime interface {
	StartWorkers(meshxruntime.WorkerOwnership) error
	Ready(context.Context) error
	QuiesceWorkers(context.Context) error
	Shutdown(context.Context) error
}

// New creates the server and validates its core infrastructure.
func New(ctx context.Context, cfg config.Config, logger *slog.Logger) (*Server, error) {
	observabilityProvider, err := observability.NewBase(cfg.Observability, logger, observability.WithEnvironment(cfg.Environment))
	if err != nil {
		return nil, fmt.Errorf("initialize observability: %w", err)
	}
	cleanupObservability := true
	defer func() {
		if cleanupObservability {
			shutdownObservabilityProvider(logger, observabilityProvider, cfg.HTTP.ShutdownTimeout)
		}
	}()

	// Build foundational infrastructure: database, external clients, store graph.
	infra, err := buildInfra(ctx, cfg, logger, coredb.SQLOpener(observabilityProvider.SQLOpener()), observabilityProvider)
	if err != nil {
		return nil, err
	}
	defer infra.closeDBOnError()

	if err := observabilityProvider.RegisterDBPool(infra.coreDBPool); err != nil {
		return nil, fmt.Errorf("register database pool metrics: %w", err)
	}
	profile, err := routerProfileForConfig(cfg.Server.ProfileOrDefault())
	if err != nil {
		return nil, err
	}
	managedWorkspaces, err := managedWorkspaceSetForProfile(ctx, profile, cfg.Server.WorkspaceGroupID, infra.stores.Workspace)
	if err != nil {
		return nil, err
	}
	var meshXWorkspaceScope func(context.Context) ([]string, error)
	if profile == router.ProfileWorkspace {
		workspaceGroupID := strings.TrimSpace(cfg.Server.WorkspaceGroupID)
		meshXWorkspaceScope = func(ctx context.Context) ([]string, error) {
			return infra.stores.Workspace.ListRoutableWorkspaceIDsForGroup(ctx, workspaceGroupID)
		}
	}
	// Build the profile-owned MeshX runtime before AgentTeam forwarders so their existing
	// HTTP contract can use an in-process RoundTripper rather than loopback TCP.
	var embeddedMeshX *meshxruntime.Runtime
	var embeddedMeshXTransport http.RoundTripper
	if profileComposesMeshX(profile, cfg.MeshX.Enabled) {
		embeddedMeshX, err = meshxruntime.NewFromEnv()
		if err != nil {
			return nil, fmt.Errorf("initialize embedded MeshX runtime: %w", err)
		}
		embeddedMeshXTransport = inprochttp.RoundTripper{Handler: embeddedMeshX.Handler()}
	}
	// Wire HTTP-layer services and assemble router dependencies.
	routerDeps, side, err := wireRouterDeps(ctx, cfg, logger, infra, observabilityProvider, profile, managedWorkspaces, embeddedMeshXTransport, embeddedKBGovernanceAuthority(embeddedMeshX))
	if err != nil {
		if embeddedMeshX != nil {
			_ = embeddedMeshX.Shutdown(context.Background())
		}
		return nil, err
	}
	if err := configureEmbeddedWikiGenerationApprovals(embeddedMeshX, side); err != nil {
		return nil, err
	}
	routerDeps.HTTPMetrics = observabilityProvider.HTTPMetrics()
	routerDeps.HTTPTracing = observabilityProvider.HTTPTracing()
	if err := routerDeps.ValidateForProfile(profile); err != nil {
		runClosers(side.closers)
		if embeddedMeshX != nil {
			_ = embeddedMeshX.Shutdown(context.Background())
		}
		return nil, fmt.Errorf("validate router dependencies: %w", err)
	}
	if err := configureEmbeddedMeshXRoutes(embeddedMeshX, profile, cfg, &routerDeps, infra); err != nil {
		return nil, err
	}
	// Create HTTP server.
	httpRouter := router.NewRouterForProfile(profile, routerDeps)
	server := &http.Server{
		Addr:              cfg.HTTP.Address,
		Handler:           httpRouter,
		ErrorLog:          logging.HTTPErrorLog(logger),
		ReadHeaderTimeout: 5 * time.Second,
	}

	// Transfer DB ownership to Server (prevents deferred close).
	serverDB := infra.coreDBConn
	infra.coreDBConn = nil
	cleanupObservability = false

	return &Server{
		config:  cfg,
		logger:  logger,
		profile: profile,

		db:                serverDB,
		stores:            infra.stores,
		managedWorkspaces: managedWorkspaces,

		schedules:                    side.scheduleService,
		workflowInvocationSchedules:  side.workflowInvocationSchedules,
		scopeFanoutRunner:            side.scopeFanoutRunner,
		unreadRuntime:                side.unreadRuntime,
		assetAccessWorker:            side.assetAccessWorker,
		workflowRunWorker:            side.workflowRunWorker,
		internalAppDispatcher:        side.internalAppDispatcher,
		approvalProjectionReconciler: side.approvalProjectionReconciler,
		platformServiceAccounts:      side.platformAccounts,
		knowledgeQAWorker:            side.knowledgeQAWorker,
		operationWorker:              side.operationWorker,
		roleManagementWorker:         side.roleManagementWorker,
		embeddedMeshX:                optionalEmbeddedMeshXRuntime(embeddedMeshX),
		meshXWorkspaceScope:          meshXWorkspaceScope,
		closers:                      side.closers,

		httpServer:    server,
		observability: observabilityProvider,
	}, nil
}

func configureEmbeddedMeshXRoutes(runtime *meshxruntime.Runtime, profile router.Profile, cfg config.Config, deps *router.Dependencies, infra *infra) error {
	if runtime == nil {
		return nil
	}
	switch profile {
	case router.ProfileAll:
		if err := runtime.ConfigureStandaloneWorkspaceAGSSeam(infra.stores.Workspace); err != nil {
			return err
		}
		deps.API.MeshXWorkspaceAGS = runtime.Handler()
	case router.ProfileWorkspace:
		if err := runtime.ConfigureWorkspaceAGSSeam(cfg.Server.WorkspaceGroupID, infra.stores.Workspace); err != nil {
			return err
		}
		deps.API.MeshXWorkspaceAGS = runtime.Handler()
	}
	deps.Health.AdditionalReady = runtime.Ready
	if profile != router.ProfileWorkspace || strings.TrimSpace(cfg.Server.WorkspaceGroupID) == cfg.Server.DefaultWorkspaceGroupIDOrDefault() {
		deps.API.MeshXInternalAdmin = runtime.Handler()
	}
	deps.API.RuntimeIdentity = lookup.Dependencies{
		Resolver: lookup.NewAuthenticatedResolver(
			lookup.NewLegacyRuntimeResolver(infra.stores.Agent, infra.stores.Agent, infra.stores.App),
			deps.API.Workspace.AuthStore, deps.API.Workspace.WorkspaceService,
		),
		InternalAdminToken: cfg.Server.InternalAdminToken,
	}
	return nil
}

func embeddedKBGovernanceAuthority(runtime *meshxruntime.Runtime) kbgovernance.Authority {
	if runtime == nil {
		return nil
	}
	return runtime.KBGovernanceAuthority()
}

func configureEmbeddedWikiGenerationApprovals(runtime *meshxruntime.Runtime, side wireSide) error {
	if runtime == nil {
		return nil
	}
	if err := runtime.ConfigureWikiGenerationApprovalGateway(side.wikiGenerationApprovals); err != nil {
		runClosers(side.closers)
		_ = runtime.Shutdown(context.Background())
		return fmt.Errorf("configure embedded MeshX Wiki generation approvals: %w", err)
	}
	if err := runtime.ConfigureWikiGenerationTerminalGateway(side.wikiGenerationTerminals); err != nil {
		runClosers(side.closers)
		_ = runtime.Shutdown(context.Background())
		return fmt.Errorf("configure embedded MeshX Wiki generation terminal notifications: %w", err)
	}
	if err := runtime.ConfigurePublicationApprovalGateway(side.publicationApprovals); err != nil {
		runClosers(side.closers)
		_ = runtime.Shutdown(context.Background())
		return fmt.Errorf("configure embedded MeshX publication approvals: %w", err)
	}
	return nil
}

// profileComposesMeshX reports whether the profile embeds the MeshX runtime
// in-process. The central and extension profiles do not compose MeshX: central
// delegates it, and extension is a narrow runtime that must not start the
// workspace/MeshX data-plane (spec §6.1).
func profileComposesMeshX(profile router.Profile, enabled bool) bool {
	if !enabled {
		return false
	}
	switch profile {
	case router.ProfileAll, router.ProfileWorkspace:
		return true
	default:
		return false
	}
}

func optionalEmbeddedMeshXRuntime(runtime *meshxruntime.Runtime) embeddedMeshXRuntime {
	if runtime == nil {
		return nil
	}
	return runtime
}

// Run starts the HTTP server and shuts it down when the context is cancelled.
func (srv *Server) Run(ctx context.Context) error {
	if srv.embeddedMeshX != nil {
		requireGroup := srv.profileOrDefault() == router.ProfileWorkspace
		meshXGroupID := strings.TrimSpace(srv.config.Server.WorkspaceGroupID)
		if !requireGroup {
			meshXGroupID = srv.config.Server.DefaultWorkspaceGroupIDOrDefault()
		}
		// The all profile and the designated workspace owner run the one global
		// maintenance set. Other workspace groups run only group-scoped workers.
		startGlobalWorkers := !requireGroup ||
			strings.TrimSpace(srv.config.Server.WorkspaceGroupID) == srv.config.Server.DefaultWorkspaceGroupIDOrDefault()
		if err := srv.embeddedMeshX.StartWorkers(meshxruntime.WorkerOwnership{
			WorkspaceGroupID:   meshXGroupID,
			RequireGroup:       requireGroup,
			StartGlobalWorkers: startGlobalWorkers,
			WorkspaceScope:     srv.meshXWorkspaceScope,
		}); err != nil {
			return fmt.Errorf("start embedded MeshX runtime: %w", err)
		}
		readyCtx, cancelReady := context.WithTimeout(ctx, srv.config.HTTP.EffectiveStartupSchemaTimeout())
		err := srv.embeddedMeshX.Ready(readyCtx)
		cancelReady()
		if err != nil {
			shutdownCtx, cancel := context.WithTimeout(context.Background(), srv.config.HTTP.ShutdownTimeout)
			defer cancel()
			_ = srv.embeddedMeshX.Shutdown(shutdownCtx)
			return fmt.Errorf("ready embedded MeshX runtime: %w", err)
		}
	}
	if srv.platformServiceAccounts != nil && profileRunsPlatformServiceAccounts(srv.profileOrDefault()) {
		logging.Go(srv.logger, func() { srv.platformServiceAccounts.Run(ctx, time.Second) })
	}
	if srv.runsWorkspaceBackgroundLoops() {
		logging.Go(srv.logger, func() { srv.runUnclaimedAttachmentGC(ctx) })
		logging.Go(srv.logger, func() { srv.runServiceEventRetentionGC(ctx) })
		logging.Go(srv.logger, func() { srv.runAppCallbackDispatcher(ctx) })
		logging.Go(srv.logger, func() { srv.runAppNotificationExpiryWorker(ctx) })
		if srv.internalAppDispatcher != nil {
			logging.Go(srv.logger, func() { srv.internalAppDispatcher.Run(ctx, 0) })
		}
		if srv.approvalProjectionReconciler != nil {
			logging.Go(srv.logger, func() { srv.approvalProjectionReconciler.Run(ctx, 0) })
		}
		if srv.operationWorker != nil {
			logging.Go(srv.logger, func() { srv.operationWorker.Run(ctx) })
		}
		if srv.roleManagementWorker != nil {
			logging.Go(srv.logger, func() { srv.roleManagementWorker.Run(ctx) })
		}
		if srv.schedules != nil {
			logging.Go(srv.logger, func() { srv.schedules.Run(ctx) })
		}
		if srv.workflowInvocationSchedules != nil {
			srv.workflowInvocationSchedules.Start(ctx)
		}
		if srv.scopeFanoutRunner != nil {
			srv.scopeFanoutRunner.Start(ctx)
		}
		if srv.unreadRuntime != nil {
			srv.unreadRuntime.Start(ctx)
		}
		if srv.assetAccessWorker != nil {
			logging.Go(srv.logger, func() { srv.assetAccessWorker.Run(ctx) })
		}
		if srv.workflowRunWorker != nil {
			srv.workflowRunWorker.Start(ctx)
		}
		if srv.knowledgeQAWorker != nil {
			srv.knowledgeQAWorker.Start(ctx)
		}
	}

	errCh := make(chan error, 1)
	logging.Go(srv.logger, func() {
		srv.logger.Info("http server listening", slog.String("address", srv.config.HTTP.Address), slog.String("profile", string(srv.profileOrDefault())))
		errCh <- srv.httpServer.ListenAndServe()
	})

	var metricsErrCh chan error
	if srv.observability != nil {
		observabilityCtx, cancelObservability := context.WithCancel(context.Background())
		defer cancelObservability()
		metricsErrCh = make(chan error, 1)
		logging.Go(srv.logger, func() {
			metricsErrCh <- srv.observability.Run(observabilityCtx)
		})
	}

	for {
		select {
		case err := <-errCh:
			srv.shutdownObservability()
			return err
		case err := <-metricsErrCh:
			if err != nil && !errors.Is(err, http.ErrServerClosed) {
				srv.logger.Error("metrics server failed", slog.Any("error", err))
			}
			metricsErrCh = nil
		case <-ctx.Done():
			return srv.gracefulShutdown()
		}
	}
}

func routerProfileForConfig(profile string) (router.Profile, error) {
	switch profile {
	case config.ServerProfileAll:
		return router.ProfileAll, nil
	case config.ServerProfileCentral:
		return router.ProfileCentral, nil
	case config.ServerProfileWorkspace:
		return router.ProfileWorkspace, nil
	case config.ServerProfileExtension:
		return router.ProfileExtension, nil
	default:
		return "", fmt.Errorf("unsupported server profile %q", profile)
	}
}

func (srv *Server) profileOrDefault() router.Profile {
	if srv.profile == "" {
		return router.ProfileAll
	}
	return srv.profile
}

// runsWorkspaceBackgroundLoops reports whether this profile owns the workspace
// data-plane background loops. Only the all and workspace profiles run them. The
// extension profile is a narrow Extension runtime and MUST NOT inherit Workspace
// workers merely because it is not central (spec §6.1) — this is an explicit
// allow-list, not a "not central" fallback.
func (srv *Server) runsWorkspaceBackgroundLoops() bool {
	switch srv.profileOrDefault() {
	case router.ProfileAll, router.ProfileWorkspace:
		return true
	default:
		return false
	}
}

// profileRunsPlatformServiceAccounts reports whether a profile runs the
// platform service-account reconciler. The extension profile is worker-free
// (spec §6.1: only Extension runtime routes + common health/observability), so
// it does not run platform workers either.
func profileRunsPlatformServiceAccounts(profile router.Profile) bool {
	return profile != router.ProfileExtension
}

func (srv *Server) gracefulShutdown() error {
	httpCtx, cancelHTTP := context.WithTimeout(context.Background(), srv.config.HTTP.ShutdownTimeout)
	httpErr := srv.httpServer.Shutdown(httpCtx)
	cancelHTTP()
	if srv.embeddedMeshX != nil {
		workerCtx, cancelWorkers := context.WithTimeout(context.Background(), srv.config.HTTP.ShutdownTimeout)
		if err := srv.embeddedMeshX.QuiesceWorkers(workerCtx); err != nil {
			srv.logger.Warn("embedded MeshX worker quiesce", slog.Any("error", err))
		}
		cancelWorkers()
		closeCtx, cancelClose := context.WithTimeout(context.Background(), srv.config.HTTP.ShutdownTimeout)
		if err := srv.embeddedMeshX.Shutdown(closeCtx); err != nil {
			srv.logger.Warn("embedded MeshX shutdown", slog.Any("error", err))
		}
		cancelClose()
	}
	if observabilityErr := srv.shutdownObservabilityWithTimeout(srv.config.HTTP.ShutdownTimeout); observabilityErr != nil && !errors.Is(observabilityErr, http.ErrServerClosed) {
		srv.logger.Warn("observability shutdown", slog.Any("error", observabilityErr))
	}
	return httpErr
}

func (srv *Server) shutdownObservability() {
	if err := srv.shutdownObservabilityWithTimeout(srv.config.HTTP.ShutdownTimeout); err != nil && !errors.Is(err, http.ErrServerClosed) {
		srv.logger.Warn("observability shutdown", slog.Any("error", err))
	}
}

func (srv *Server) shutdownObservabilityWithTimeout(timeout time.Duration) error {
	shutdownCtx, cancel := context.WithTimeout(context.Background(), observabilityShutdownTimeout(timeout))
	defer cancel()
	return srv.shutdownObservabilityWithContext(shutdownCtx)
}

func (srv *Server) shutdownObservabilityWithContext(ctx context.Context) error {
	if srv.observability == nil {
		return nil
	}
	return srv.observability.Shutdown(ctx)
}

const (
	unclaimedAttachmentGCInterval  = time.Hour
	unclaimedAttachmentGCBatchSize = 100
	unclaimedAttachmentGCTTL       = 24 * time.Hour

	serviceEventRetentionGCInterval                = time.Hour
	serviceEventRetentionGCBatchSize               = 500
	pullRequestSubscriptionRateLimitEventRetention = 7 * 24 * time.Hour
	serviceEventWebhookDeliveryRetention           = 30 * 24 * time.Hour
	serviceEventDeliveryRetention                  = 30 * 24 * time.Hour
	serviceEventDeliveryRetentionBatchSize         = 500

	appCallbackDispatchInterval      = 2 * time.Second
	appCallbackDispatchTargetLimit   = 100
	appCallbackDispatchDeliveryLimit = 10
)

func (srv *Server) runAppCallbackDispatcher(ctx context.Context) {
	if srv.stores == nil || srv.stores.App == nil || srv.stores.ServiceActor == nil {
		return
	}
	callbackService := appruntime.NewCallbackService(appruntime.CallbackServiceOptions{
		Store:                    srv.stores.App,
		Logger:                   srv.logger,
		AllowInsecureCallbackURL: !strings.EqualFold(srv.config.Environment, "production"),
	})

	runPass := func() {
		now := time.Now().UTC()
		for _, subscriptionKind := range serviceactor.AppRuntimeSubscriptionKinds() {
			targets, err := srv.stores.ServiceActor.ListRuntimeCallbackDispatchTargets(ctx, serviceactor.RuntimeCallbackDispatchTargetList{
				WorkspaceGroupID: schedulerWorkspaceGroupIDForConfig(srv.config.Server),
				ServiceKind:      serviceactor.KindApp,
				SubscriptionKind: subscriptionKind,
				Now:              now,
				Limit:            appCallbackDispatchTargetLimit,
			})
			if err != nil {
				if !errors.Is(err, context.Canceled) {
					srv.logger.Warn("app callback dispatch target scan failed", slog.String("subscription_kind", subscriptionKind), slog.Any("error", err))
				}
				continue
			}
			for _, target := range targets {
				if !owner.IsManaged(srv.managedWorkspaces, target.WorkspaceID) {
					continue
				}
				result, err := callbackService.DispatchCallbackDeliveries(ctx, appruntime.CallbackDispatchParams{
					WorkspaceID:           target.WorkspaceID,
					ServiceInstallationID: target.ServiceInstallationID,
					SubscriptionKind:      subscriptionKind,
					Limit:                 appCallbackDispatchDeliveryLimit,
					Now:                   now,
				})
				if err != nil {
					if !errors.Is(err, context.Canceled) {
						srv.logger.Warn("app callback dispatch failed", slog.String("workspace_id", target.WorkspaceID), slog.String("service_installation_id", target.ServiceInstallationID), slog.String("subscription_kind", subscriptionKind), slog.Any("error", err))
					}
					continue
				}
				if result.Attempted > 0 {
					srv.logger.Info(
						"app callback dispatch completed",
						slog.String("workspace_id", target.WorkspaceID),
						slog.String("service_installation_id", target.ServiceInstallationID),
						slog.String("subscription_kind", subscriptionKind),
						slog.Int("attempted", result.Attempted),
						slog.Int("delivered", result.Delivered),
						slog.Int("retried", result.Retried),
					)
				}
			}
		}
	}

	runPass()
	ticker := time.NewTicker(appCallbackDispatchInterval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			runPass()
		}
	}
}

func (srv *Server) runServiceEventRetentionGC(ctx context.Context) {
	if srv.stores == nil || srv.stores.App == nil || srv.stores.ServiceActor == nil {
		return
	}

	runPass := func() {
		now := time.Now().UTC()
		deletedRateLimitEvents, rateLimitErr := srv.stores.App.DeletePullRequestSubscriptionRateLimitEventsBefore(ctx, now.Add(-pullRequestSubscriptionRateLimitEventRetention), serviceEventRetentionGCBatchSize)
		deletedWebhookDeliveries, webhookErr := srv.stores.App.DeleteServiceEventWebhookDeliveriesBefore(ctx, now.Add(-serviceEventWebhookDeliveryRetention), serviceEventRetentionGCBatchSize)
		deletedEventDeliveries, eventDeliveryErr := srv.stores.ServiceActor.DeleteEventDeliveriesBefore(ctx, serviceactor.EventDeliveryRetentionFilter{
			ServiceKind:      serviceactor.KindApp,
			SubscriptionKind: "pull_request",
			Cutoff:           now.Add(-serviceEventDeliveryRetention),
			Limit:            serviceEventDeliveryRetentionBatchSize,
		})
		for label, err := range map[string]error{
			"rateLimitEvents":   rateLimitErr,
			"webhookDeliveries": webhookErr,
			"serviceDeliveries": eventDeliveryErr,
		} {
			if err != nil && !errors.Is(err, context.Canceled) {
				srv.logger.Warn("service event retention gc pass failed", slog.String("scope", label), slog.Any("error", err))
			}
		}
		if deletedRateLimitEvents > 0 || deletedWebhookDeliveries > 0 || deletedEventDeliveries > 0 {
			srv.logger.Info(
				"service event retention gc pass completed",
				slog.Int64("deletedRateLimitEvents", deletedRateLimitEvents),
				slog.Int64("deletedWebhookDeliveries", deletedWebhookDeliveries),
				slog.Int64("deletedServiceDeliveries", deletedEventDeliveries),
			)
		}
	}

	runPass()
	ticker := time.NewTicker(serviceEventRetentionGCInterval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			runPass()
		}
	}
}

func (srv *Server) runUnclaimedAttachmentGC(ctx context.Context) {
	if srv.stores == nil || srv.stores.Attachment == nil {
		return
	}

	runPass := func() {
		deletedCount, err := srv.stores.Attachment.DeleteUnclaimedAttachmentsBefore(
			ctx,
			time.Now().UTC().Add(-unclaimedAttachmentGCTTL),
			unclaimedAttachmentGCBatchSize,
		)
		if err != nil && !errors.Is(err, context.Canceled) {
			srv.logger.Warn("unclaimed attachment gc pass failed", slog.Any("error", err))
		}
		if deletedCount > 0 {
			srv.logger.Info("unclaimed attachment gc pass completed", slog.Int64("deletedAttachments", deletedCount))
		}
	}

	runPass()
	ticker := time.NewTicker(unclaimedAttachmentGCInterval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			runPass()
		}
	}
}

// Close releases infrastructure resources.
func (srv *Server) Close() {
	if srv.embeddedMeshX != nil {
		shutdownCtx, cancel := context.WithTimeout(context.Background(), observabilityShutdownTimeout(srv.config.HTTP.ShutdownTimeout))
		if err := srv.embeddedMeshX.Shutdown(shutdownCtx); err != nil && srv.logger != nil {
			srv.logger.Warn("embedded MeshX close", slog.Any("error", err))
		}
		cancel()
		srv.embeddedMeshX = nil
	}
	srv.shutdownObservability()
	runClosers(srv.closers)
	srv.closers = nil
	if srv.db != nil {
		srv.db.Close()
	}
}

func shutdownObservabilityProvider(logger *slog.Logger, provider *observability.Provider, timeout time.Duration) {
	if provider == nil {
		return
	}
	shutdownCtx, cancel := context.WithTimeout(context.Background(), observabilityShutdownTimeout(timeout))
	defer cancel()
	if err := provider.Shutdown(shutdownCtx); err != nil && !errors.Is(err, http.ErrServerClosed) {
		if logger != nil {
			logger.Warn("observability shutdown after initialization failure", slog.Any("error", err))
		}
	}
}

func observabilityShutdownTimeout(timeout time.Duration) time.Duration {
	if timeout > 0 {
		return timeout
	}
	return 5 * time.Second
}

func runClosers(closers []func()) {
	for i := len(closers) - 1; i >= 0; i-- {
		if closers[i] != nil {
			closers[i]()
		}
	}
}