story/internal/app/app.go

package app

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

	"data-terminal-cloud/internal/alertnotification"
	"data-terminal-cloud/internal/alerts"
	"data-terminal-cloud/internal/audit"
	"data-terminal-cloud/internal/billing"
	"data-terminal-cloud/internal/cache"
	"data-terminal-cloud/internal/callmemory"
	"data-terminal-cloud/internal/config"
	"data-terminal-cloud/internal/contentmodel"
	"data-terminal-cloud/internal/contentpolicy"
	"data-terminal-cloud/internal/database"
	"data-terminal-cloud/internal/dataexport"
	"data-terminal-cloud/internal/device"
	"data-terminal-cloud/internal/deviceauth"
	"data-terminal-cloud/internal/devicefleet"
	"data-terminal-cloud/internal/devicereset"
	"data-terminal-cloud/internal/diagnostics"
	"data-terminal-cloud/internal/elderagent"
	"data-terminal-cloud/internal/embeddingprovider"
	"data-terminal-cloud/internal/externalprobe"
	"data-terminal-cloud/internal/extraction"
	"data-terminal-cloud/internal/hardwareprovision"
	"data-terminal-cloud/internal/health"
	"data-terminal-cloud/internal/httpapi"
	"data-terminal-cloud/internal/idempotency"
	"data-terminal-cloud/internal/identity"
	"data-terminal-cloud/internal/media"
	"data-terminal-cloud/internal/memoir"
	"data-terminal-cloud/internal/memoryindex"
	"data-terminal-cloud/internal/memorylifecycle"
	"data-terminal-cloud/internal/memorytool"
	"data-terminal-cloud/internal/objectstore"
	"data-terminal-cloud/internal/observability"
	"data-terminal-cloud/internal/onboarding"
	"data-terminal-cloud/internal/opsaccess"
	"data-terminal-cloud/internal/opsconfig"
	"data-terminal-cloud/internal/opsdevice"
	"data-terminal-cloud/internal/orchestration"
	"data-terminal-cloud/internal/outbox"
	"data-terminal-cloud/internal/pagination"
	"data-terminal-cloud/internal/privacy"
	"data-terminal-cloud/internal/providermemory"
	"data-terminal-cloud/internal/ratelimit"
	"data-terminal-cloud/internal/rtc"
	"data-terminal-cloud/internal/saas"
	"data-terminal-cloud/internal/sharing"
	"data-terminal-cloud/internal/vikingmemory"
	"data-terminal-cloud/internal/voiceprovider"
	"data-terminal-cloud/internal/webevents"

	"github.com/jackc/pgx/v5/pgxpool"
	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
)

func Run(ctx context.Context, serviceName string) error {
	cfg, err := config.Load(serviceName)
	if err != nil {
		return fmt.Errorf("load configuration: %w", err)
	}
	logger := newLogger(cfg)
	traceProvider, err := observability.SetupTracing(ctx, serviceName, cfg.Environment, cfg.OTLPEndpoint)
	if err != nil {
		return err
	}
	defer func() {
		shutdownContext, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		if err := traceProvider.Shutdown(shutdownContext); err != nil {
			logger.Error("trace provider shutdown failed", "error", err)
		}
	}()
	databasePool, err := database.Open(ctx, cfg.DatabaseURL, serviceName)
	if err != nil {
		return err
	}
	defer databasePool.Close()
	redisClient := cache.Open(cfg.RedisAddr, cfg.RedisPassword, cfg.RedisDB)
	defer func() { _ = redisClient.Close() }()
	var exportStore *objectstore.Store
	var memoryEmbedding embeddingprovider.Provider
	var providerMemoryClient providermemory.Client
	var vikingMemory *vikingmemory.Service
	callMemoryService := callmemory.NewService(databasePool, logger)
	if bucket := cfg.TOSRecordingBucket; bucket != "" && serviceName == "api" {
		recordings, recordingErr := callmemory.NewTOSRecordings(bucket, cfg.TOSRecordingRegion, cfg.VolcAccessKeyID, cfg.VolcSecretAccessKey, cfg.VolcSessionToken)
		if recordingErr != nil {
			return fmt.Errorf("configure cloud recordings: %w", recordingErr)
		}
		if err := recordings.ConfigurePlayback(cfg.TOSRecordingPlaybackEndpoint); err != nil {
			return err
		}
		callMemoryService.ConfigureRecordings(recordings)
		if account := cfg.TOSRecordingAccountID; account != "" {
			callMemoryService.ConfigureMixedRecorder(callmemory.NewMixedRecorder(cfg.VolcAccessKeyID, cfg.VolcSecretAccessKey, cfg.VolcSessionToken, account, bucket))
		}
	}
	var directTemplate []byte
	directTemplatePath := os.Getenv("DIRECT_CHAT_TEMPLATE_FILE")
	if directTemplatePath == "" {
		directTemplatePath = cfg.VolcVoiceChatTemplate
	}
	if directTemplatePath != "" {
		directTemplate, err = os.ReadFile(directTemplatePath)
		if err != nil {
			return fmt.Errorf("read direct chat template: %w", err)
		}
	}
	if len(directTemplate) > 0 {
		directTemplate, err = cfg.ApplyVolcVoiceTemplate(directTemplate)
		if err != nil {
			return err
		}
	}
	callMemoryService.ConfigureDirect(callmemory.DirectOptions{Template: directTemplate, BotID: cfg.VolcHardwareAgentBotID, CallbackURL: cfg.SubtitleCallbackURL, CallbackSecret: []byte(cfg.SubtitleCallbackSecret)})
	var elderAgents *elderagent.Service
	if len(directTemplate) > 0 && (serviceName == "api" || serviceName == "worker") {
		elderAgents = &elderagent.Service{Pool: databasePool, Provider: elderagent.NewClient(cfg.VolcAccessKeyID, cfg.VolcSecretAccessKey, cfg.VolcSessionToken, cfg.VolcRTCBaseURL, cfg.VolcRTCRegion).ForProduct(cfg.VolcHardwareInstanceID, cfg.VolcHardwareProductKey), Template: directTemplate, CallbackURL: cfg.SubtitleCallbackURL, CallbackSecret: []byte(cfg.SubtitleCallbackSecret), VikingCollectionName: cfg.VikingCollectionName, VikingEventType: cfg.VikingEventType}
	}
	if serviceName == "api" || serviceName == "worker" {
		if cfg.VikingResourceID != "" {
			client, vikingErr := vikingmemory.NewClient(vikingmemory.Options{ResourceID: cfg.VikingResourceID, CollectionName: cfg.VikingCollectionName, APIKey: cfg.VikingAPIKey, EventType: cfg.VikingEventType, ProfileType: cfg.VikingProfileType, Timeout: 8 * time.Second})
			if vikingErr != nil {
				return fmt.Errorf("configure Viking memory: %w", vikingErr)
			}
			vikingMemory = vikingmemory.NewService(databasePool, client)
			callMemoryService.ConfigureVikingConsole(client)
			managedMemory := vikingmemory.NewManaged(databasePool, client)
			callMemoryService.ConfigureProjection(managedMemory)
			callMemoryService.ConfigureManagedSync(managedMemory.SyncNext)
		}
		memoryEmbedding, err = newMemoryEmbeddingProvider(cfg)
		if err != nil {
			return fmt.Errorf("create memory embedding provider: %w", err)
		}
		if cfg.ProviderMemoryMode == "managed_http" {
			providerMemoryClient, err = providermemory.NewHTTPClient(cfg.ProviderMemoryBaseURL, cfg.ProviderMemoryAPIKey, cfg.ProviderMemoryTimeout)
			if err != nil {
				return fmt.Errorf("create provider memory client: %w", err)
			}
		}
		exportStore, err = objectstore.New(cfg.ObjectStoreEndpoint, cfg.ObjectStorePublic, cfg.ObjectStoreAccessKey, cfg.ObjectStoreSecretKey, cfg.ObjectStoreBucket, cfg.ObjectStoreRegion, cfg.ObjectStoreSecure, cfg.ObjectStorePublicSecure)
		if err != nil {
			return fmt.Errorf("create private export object store: %w", err)
		}
		ensureContext, cancel := context.WithTimeout(ctx, cfg.DependencyTimeout)
		err = exportStore.EnsureBucket(ensureContext)
		cancel()
		if err != nil {
			return fmt.Errorf("ensure private export bucket: %w", err)
		}
	}

	readinessChecks := map[string]health.CheckFunc{
		"postgres": observability.InstrumentDependencyCheck(serviceName, "postgres", databasePool.Ping),
		"redis": observability.InstrumentDependencyCheck(serviceName, "redis", func(checkContext context.Context) error {
			return redisClient.Ping(checkContext).Err()
		}),
	}
	if exportStore != nil {
		readinessChecks["object_store"] = observability.InstrumentDependencyCheck(serviceName, "object_store", exportStore.Ready)
	}
	readiness := health.NewChecker(cfg.DependencyTimeout, readinessChecks)
	externalProbeRunner, err := newExternalProbeRunner(cfg, logger, databasePool)
	if err != nil {
		return fmt.Errorf("create external dependency probes: %w", err)
	}
	tokenVerifier := identity.NewOIDCVerifier(ctx, cfg.AuthIssuerURL, cfg.AuthJWKSURL, cfg.AuthAudience)
	identityRepository := identity.NewRepository(databasePool)
	var deviceVerifier httpapi.DeviceRequestVerifier
	var onboardingService *onboarding.Service
	var deviceRepository *device.Repository
	var deviceTelemetryService *device.TelemetryService
	var conversationCreator *orchestration.Service
	var conversationHeartbeater *orchestration.HeartbeatService
	var conversationTranscriptSubmitter *orchestration.TranscriptBatchService
	var providerEventProcessor *orchestration.ProviderEventProcessor
	var saasService *saas.Service
	var mediaService *media.Service
	var webEventService *webevents.Service
	var sharingService *sharing.Service
	var opsConfigService *opsconfig.Service
	var billingService *billing.Service
	var alertService *alerts.Service
	var diagnosticService *diagnostics.Service
	var opsAccessService *opsaccess.Service
	var opsDeviceService *opsdevice.Service
	var memoryToolService *memorytool.Service
	var memoryToolSigner *memorytool.TokenSigner
	var contentModelService *contentmodel.Service
	var contentModelRegistry *contentmodel.Registry
	var memoirService *memoir.Service
	var hardwareProvisioningService *hardwareprovision.Service
	var deviceFactoryResetService *devicereset.Service
	if serviceName == "api" || serviceName == "worker" {
		var fallback extraction.Generator = extraction.DeterministicGenerator{}
		if cfg.ContentGenerator == "ark" {
			fallback, err = extraction.NewArkGenerator(cfg.ArkBaseURL, cfg.ArkAPIKey, cfg.ArkContentModel, cfg.ArkRequestTimeout)
			if err != nil {
				return fmt.Errorf("create fallback Ark content generator: %w", err)
			}
		}
		contentModelRegistry, err = contentmodel.NewRegistry(databasePool, cfg.ContentModelMasterKey, fallback)
		if err != nil {
			return fmt.Errorf("create content model registry: %w", err)
		}
		memoirService = &memoir.Service{Pool: databasePool, Model: cfg.ArkMemoirModel, Safety: contentpolicy.LocalClassifier{}}
		if cfg.ArkAPIKey != "" {
			memoirService.Generator, err = extraction.NewArkJSONGenerator(cfg.ArkBaseURL, cfg.ArkAPIKey, cfg.ArkMemoirModel, 180*time.Second, 65536)
			if err != nil {
				return err
			}
			memoirService.Illustrator = memoir.NewIllustrator(cfg.ArkBaseURL, cfg.ArkAPIKey, cfg.ArkMemoirImageModel)
		}
		if cfg.DifyEnabled {
			memoirService.Workflow, err = memoir.NewDifyWorkflow(cfg.DifyBaseURL, cfg.DifyMemoirAPIKey, cfg.DifyBridgeURL, cfg.DifyBridgeSecret)
			if err != nil {
				return err
			}
		}
	}
	if serviceName == "api" || serviceName == "callbacks" {
		diagnosticService, err = diagnostics.NewService(databasePool, idempotency.NewExecutor(databasePool), audit.NewRepository())
		if err != nil {
			return fmt.Errorf("create diagnostic service: %w", err)
		}
	}
	if serviceName == "api" {
		contentModelService, err = contentmodel.NewService(contentModelRegistry, audit.NewRepository())
		if err != nil {
			return fmt.Errorf("create content model service: %w", err)
		}
		memoryToolSigner, err = memorytool.NewTokenSigner([]byte(cfg.MemoryToolSigningSecret))
		if err != nil {
			return fmt.Errorf("create memory tool signer: %w", err)
		}
		memoryToolService, err = memorytool.NewService(databasePool, memoryEmbedding, audit.NewRepository())
		if err != nil {
			return fmt.Errorf("create memory tool service: %w", err)
		}
		webEventService = webevents.NewService(databasePool)
		cursorCodec, cursorErr := pagination.NewCodec([]byte(cfg.CursorSigningSecret))
		if cursorErr != nil {
			return fmt.Errorf("create pagination cursor codec: %w", cursorErr)
		}
		saasService = saas.NewService(databasePool, audit.NewRepository(), redisClient, cursorCodec)
		saasService.ConfigureProviderMemory(cfg.ProviderMemoryMode == "managed_http")
		if elderAgents != nil {
			saasService.ConfigureVoiceAgents(elderAgents)
			callMemoryService.ConfigureDirect(callmemory.DirectOptions{Prepare: elderAgents.Prepare, PrepareWithWelcome: elderAgents.PrepareWithWelcome, CallbackURL: cfg.SubtitleCallbackURL, CallbackSecret: []byte(cfg.SubtitleCallbackSecret)})
		}
		opsConfigService, err = opsconfig.NewService(databasePool, audit.NewRepository())
		if err != nil {
			return fmt.Errorf("create operations configuration service: %w", err)
		}
		billingService, err = billing.NewService(databasePool, audit.NewRepository())
		if err != nil {
			return fmt.Errorf("create usage billing service: %w", err)
		}
		alertService, err = alerts.NewService(databasePool, audit.NewRepository())
		if err != nil {
			return fmt.Errorf("create alert operations service: %w", err)
		}
		opsAccessService, err = opsaccess.NewService(databasePool, idempotency.NewExecutor(databasePool), audit.NewRepository(), exportStore, 5*time.Minute)
		if err != nil {
			return fmt.Errorf("create temporary sensitive access service: %w", err)
		}
		saasService.ConfigureExportSigning(exportStore, cfg.ExportDownloadTTL)
		saasService.ConfigureFirmwareStore(exportStore, 5*time.Minute)
		if exportStore != nil {
			mediaService, err = media.NewService(databasePool, outbox.NewStore(databasePool), audit.NewRepository(), exportStore, 15*time.Minute)
			if err != nil {
				return fmt.Errorf("create media service: %w", err)
			}
			sharingService, err = sharing.NewService(databasePool, audit.NewRepository(), exportStore, []byte(cfg.ShareTokenSecret), 5*time.Minute)
			if err != nil {
				return fmt.Errorf("create sharing service: %w", err)
			}
		}
		deviceIdentityRepository := deviceauth.NewRepository(databasePool)
		deviceSecretResolver, err := deviceauth.NewDatabaseSecretResolver(databasePool, cfg.DeviceCredentialMasterKey)
		if err != nil {
			return fmt.Errorf("create database device secret resolver: %w", err)
		}
		if len(cfg.DeviceLocalSecret) >= 32 {
			if err := deviceSecretResolver.ImportMissingLegacyCredentials(ctx, map[string][]byte{
				"local-dev://simulated-terminal-1": []byte(cfg.DeviceLocalSecret),
				"local-dev://unbound-terminal-2":   []byte(cfg.DeviceLocalSecret),
			}); err != nil {
				return fmt.Errorf("import legacy device credentials: %w", err)
			}
		}
		opsDeviceService, err = opsdevice.NewService(databasePool, deviceSecretResolver, audit.NewRepository(), opsdevice.Options{
			OnboardingMasterKey: cfg.Onboarding.MasterKey,
			DeviceOrigin:        cfg.Onboarding.EntryURL,
		})
		if err != nil {
			return fmt.Errorf("create operations device service: %w", err)
		}
		deviceVerifier, err = deviceauth.NewVerifier(
			deviceIdentityRepository,
			deviceSecretResolver,
			deviceauth.NewRedisReplayGuard(redisClient, "voice-platform"),
			cfg.DeviceClockSkew,
		)
		if err != nil {
			return fmt.Errorf("create device verifier: %w", err)
		}
		hardwareProvisioningService, err = hardwareprovision.NewService(databasePool, audit.NewRepository(), hardwareprovision.Options{
			Enabled: cfg.VolcHardwareProvisioningEnabled, InstanceID: cfg.VolcHardwareInstanceID,
			ProductKey: cfg.VolcHardwareProductKey, BotID: cfg.VolcHardwareAgentBotID,
			Transport: cfg.VolcHardwareTransport, MinimumSDKVersion: cfg.VolcHardwareMinimumSDKVersion,
			ProductSecret:       cfg.VolcHardwareProductSecret,
			CredentialMasterKey: cfg.VolcHardwareDeviceCredentialMasterKey, RegisterURL: cfg.VolcHardwareRegisterURL,
			RegisterTimeout: cfg.VolcHardwareRegisterTimeout, DeviceNamePrefix: cfg.VolcHardwareDeviceNamePrefix,
		})
		if err != nil {
			return fmt.Errorf("create hardware provider provisioning service: %w", err)
		}
		deviceFactoryResetService, err = devicereset.NewService(databasePool, idempotency.NewExecutor(databasePool), audit.NewRepository())
		if err != nil {
			return fmt.Errorf("create device factory reset service: %w", err)
		}
		deviceRepository = device.NewRepository(databasePool)
		deviceTelemetryService, err = device.NewTelemetryService(databasePool, idempotency.NewExecutor(databasePool))
		if err != nil {
			return fmt.Errorf("create device telemetry service: %w", err)
		}
		var tokenIssuer rtc.TokenIssuer
		if cfg.RTCTokenIssuer == "volcengine" {
			tokenIssuer, err = rtc.NewVolcengineTokenIssuer(cfg.RTCAppID, cfg.RTCAppKey, cfg.RTCTokenTTL)
		} else {
			tokenIssuer, err = rtc.NewLocalTokenIssuer([]byte(cfg.RTCTokenSecret))
		}
		if err != nil {
			return fmt.Errorf("create RTC token issuer: %w", err)
		}
		leaseCache, leaseErr := orchestration.NewRedisLeaseCache(redisClient, "voice-platform")
		if leaseErr != nil {
			return fmt.Errorf("create conversation lease cache: %w", leaseErr)
		}
		conversationCreator, err = orchestration.NewService(
			idempotency.NewExecutor(databasePool), orchestration.NewRepository(), audit.NewRepository(),
			outbox.NewStore(databasePool), tokenIssuer, leaseCache,
			orchestration.ServiceOptions{
				RTCAppID: cfg.RTCAppID, VoiceProvider: cfg.VoiceProvider, EngineMode: "s2s_hybrid", AllocationTTL: cfg.AllocationTTL,
				TokenTTL: cfg.RTCTokenTTL, HeartbeatInterval: cfg.HeartbeatInterval, RequirePublishedRouting: cfg.Environment == config.EnvironmentProd,
				HardwareBotID: cfg.VolcHardwareAgentBotID, HardwareTransport: cfg.VolcHardwareTransport,
				SubtitleCallbackURL: cfg.SubtitleCallbackURL, SubtitleCallbackSecret: []byte(cfg.SubtitleCallbackSecret), SubtitlePrimary: cfg.SubtitlePrimary,
			},
		)
		if err != nil {
			return fmt.Errorf("create conversation orchestration service: %w", err)
		}
		if err = conversationCreator.ConfigureMemoryEmbedding(databasePool, memoryEmbedding); err != nil {
			return fmt.Errorf("configure conversation memory embedding: %w", err)
		}
		conversationCreator.ConfigureVikingMemory(vikingMemory)
		memoryToolService.ConfigureVikingMemory(vikingMemory)
		if cfg.MemoryToolBaseURL != "" {
			if err = conversationCreator.ConfigureHardwareMemoryTool(memoryToolSigner, cfg.MemoryToolBaseURL, cfg.MemoryToolTokenTTL); err != nil {
				return fmt.Errorf("configure hardware memory tool: %w", err)
			}
		}
		credentialRenewalWindow := cfg.RTCTokenTTL / 3
		minimumRenewalWindow := time.Duration(cfg.HeartbeatInterval*2) * time.Second
		if credentialRenewalWindow < minimumRenewalWindow {
			credentialRenewalWindow = minimumRenewalWindow
		}
		if credentialRenewalWindow >= cfg.RTCTokenTTL {
			credentialRenewalWindow = cfg.RTCTokenTTL / 2
		}
		conversationHeartbeater, err = orchestration.NewHeartbeatService(
			databasePool, orchestration.NewRepository(), leaseCache, tokenIssuer,
			orchestration.HeartbeatServiceOptions{
				RTCAppID: cfg.RTCAppID, LeaseTTL: time.Duration(cfg.HeartbeatInterval*5) * time.Second,
				TokenTTL: cfg.RTCTokenTTL, CredentialRenewalWindow: credentialRenewalWindow,
			},
		)
		if err != nil {
			return fmt.Errorf("create conversation heartbeat service: %w", err)
		}
		conversationTranscriptSubmitter, err = orchestration.NewTranscriptBatchService(
			databasePool, idempotency.NewExecutor(databasePool), audit.NewRepository(), outbox.NewStore(databasePool),
		)
		if err != nil {
			return fmt.Errorf("create conversation transcript batch service: %w", err)
		}
	}
	if serviceName == "callbacks" {
		providerEventProcessor, err = orchestration.NewProviderEventProcessor(databasePool, time.Duration(cfg.HeartbeatInterval*5)*time.Second)
		if err != nil {
			return fmt.Errorf("create provider event processor: %w", err)
		}
	}
	limiter, err := ratelimit.NewPolicies(redisClient, "voice-platform:"+serviceName, map[string]ratelimit.Policy{
		"network":        {RatePerSecond: cfg.RateLimitRPS, Burst: cfg.RateLimitBurst},
		"user":           {RatePerSecond: cfg.RateLimitUserRPS, Burst: cfg.RateLimitUserBurst},
		"tenant":         {RatePerSecond: cfg.RateLimitTenantRPS, Burst: cfg.RateLimitTenantBurst},
		"device":         {RatePerSecond: cfg.RateLimitDeviceRPS, Burst: cfg.RateLimitDeviceBurst},
		"provider_start": {RatePerSecond: cfg.ProviderStartRPS, Burst: cfg.ProviderStartBurst},
		"memory_tool":    {RatePerSecond: cfg.MemoryToolRateRPS, Burst: cfg.MemoryToolRateBurst},
	})
	if err != nil {
		return fmt.Errorf("create rate limiter: %w", err)
	}
	if (serviceName == "api" || serviceName == "worker") && cfg.Onboarding.MasterKey != "" {
		var sms onboarding.SMSProvider
		if cfg.Onboarding.SMSLoginEnabled {
			if cfg.Onboarding.SMSProvider == "tencent" {
				if cfg.Onboarding.TencentSMSTemplates != nil {
					sms, err = onboarding.NewTencentSMSProviderWithTemplates(cfg.Onboarding.TencentSMSSecretID, cfg.Onboarding.TencentSMSSecretKey, cfg.Onboarding.TencentSMSAppID, cfg.Onboarding.TencentSMSSignName, cfg.Onboarding.TencentSMSTemplates, cfg.Onboarding.TencentSMSTemplateParams)
				} else {
					sms, err = onboarding.NewTencentSMSProvider(cfg.Onboarding.TencentSMSSecretID, cfg.Onboarding.TencentSMSSecretKey, cfg.Onboarding.TencentSMSAppID, cfg.Onboarding.TencentSMSSignName, cfg.Onboarding.TencentSMSTemplateID)
				}
			} else {
				sms, err = onboarding.NewHTTPSMSProvider(cfg.Onboarding.SMSGatewayURL, cfg.Onboarding.SMSGatewayToken)
			}
			if err != nil {
				return err
			}
		}
		captchaProvider, captchaErr := onboarding.NewTencentCaptchaProvider(cfg.Onboarding.TencentCaptcha)
		if captchaErr != nil {
			return captchaErr
		}
		onboardingService, err = onboarding.NewService(databasePool, redisClient, onboarding.Options{SaaSCaptchaDisabled: cfg.Onboarding.SaaSCaptchaDisabled, Captcha: captchaProvider, Enabled: cfg.Onboarding.Enabled, PauseClaims: cfg.Onboarding.PauseClaims, MasterKey: cfg.Onboarding.MasterKey, EntryURL: cfg.Onboarding.EntryURL, AllowLocalHTTP: cfg.Onboarding.AllowLocalHTTP, TrustedProxyCIDRs: cfg.Onboarding.TrustedProxyCIDRs, TokenVerifier: tokenVerifier, LegacyVerifier: deviceVerifier, ActivationEnabled: cfg.Onboarding.ActivationEnabled, ProviderMasterKey: cfg.VolcHardwareDeviceCredentialMasterKey, InstanceID: cfg.VolcHardwareInstanceID, ProductKey: cfg.VolcHardwareProductKey, ProductSecret: cfg.VolcHardwareProductSecret, BotID: cfg.VolcHardwareAgentBotID, Transport: cfg.VolcHardwareTransport, MinimumSDKVersion: cfg.VolcHardwareMinimumSDKVersion, SMS: sms, WeChatAppID: cfg.Onboarding.WeChatAppID, WeChatAppSecret: cfg.Onboarding.WeChatAppSecret, SMSLoginEnabled: cfg.Onboarding.SMSLoginEnabled, SelfSignupTenantIDs: cfg.Onboarding.SelfSignupTenantIDs, Logger: logger})
		if err != nil {
			return fmt.Errorf("create onboarding service: %w", err)
		}
		if serviceName == "api" {
			deviceVerifier = onboardingService
		}
	}
	if serviceName == "api" && saasService != nil {
		downloadBaseURL := strings.TrimRight(strings.TrimSpace(os.Getenv("FIRMWARE_DOWNLOAD_BASE_URL")), "/")
		if downloadBaseURL == "" {
			scheme := "http"
			if cfg.ObjectStorePublicSecure {
				scheme = "https"
			}
			downloadBaseURL = scheme + "://" + cfg.ObjectStorePublic
		}
		fleet := &devicefleet.Service{Pool: databasePool, Store: exportStore, BrokerURL: os.Getenv("DEVICE_MQTT_PUBLIC_URL"), DownloadBaseURL: downloadBaseURL}
		address := os.Getenv("DEVICE_MQTT_ADDRESS")
		if address == "" {
			address = ":1883"
		}
		closeBroker, fleetErr := fleet.StartBroker(ctx, address, logger)
		if fleetErr != nil {
			return fmt.Errorf("start device MQTT: %w", fleetErr)
		}
		defer closeBroker()
		saasService.Fleet = fleet
		if onboardingService != nil {
			onboardingService.Fleet = fleet
		}
	}
	if serviceName == "worker" {
		if elderAgents != nil {
			go elderAgents.RunPersonaSync(ctx, logger)
		}
		webEventService = webevents.NewService(databasePool)
		outbox.RegisterMetrics(databasePool)
		observability.RegisterBusinessMetrics(databasePool)
		outboxWorker := outbox.NewWorker(
			outbox.NewStore(databasePool),
			logger,
			outbox.WorkerOptions{PollInterval: cfg.WorkerPollInterval},
		)
		outboxWorker.Register("platform.noop", func(context.Context, outbox.Event) error { return nil })
		if onboardingService != nil {
			outboxWorker.Register(onboarding.ActivationEvent, onboardingService.ProcessActivation)
			go onboardingService.RunMaintenance(ctx)
		}
		alertWebhookProcessor, processorErr := alertnotification.NewProcessor(cfg.AlertWebhookURL, cfg.AlertWebhookSecret, cfg.AlertWebhookTimeout)
		if processorErr != nil {
			return fmt.Errorf("create alert webhook processor: %w", processorErr)
		}
		outboxWorker.Register("alert.webhook_requested", alertWebhookProcessor.Handle)
		outboxWorker.Register("alert.webhook_test_requested", alertWebhookProcessor.Handle)
		outboxWorker.Register("conversation.room_issued", func(_ context.Context, event outbox.Event) error {
			logger.Info("conversation room issued event observed", "event_id", event.ID, "conversation_id", event.AggregateID)
			return nil
		})
		var providerClient voiceprovider.Client
		if cfg.VoiceProvider == "volcengine" {
			template, readErr := os.ReadFile(cfg.VolcVoiceChatTemplate)
			if readErr != nil {
				return fmt.Errorf("read Volcengine voice-chat template: %w", readErr)
			}
			template, err = cfg.ApplyVolcVoiceTemplate(template)
			if err != nil {
				return err
			}
			providerClient, err = voiceprovider.NewVolcengineClient(voiceprovider.VolcengineOptions{
				BaseURL: cfg.VolcRTCBaseURL, AccessKeyID: cfg.VolcAccessKeyID, SecretAccessKey: cfg.VolcSecretAccessKey,
				SessionToken: cfg.VolcSessionToken, Region: cfg.VolcRTCRegion, AppID: cfg.RTCAppID,
				APIVersion: cfg.VolcRTCAPIVersion, Template: template, Timeout: cfg.ProviderRequestTimeout,
				RequireAllEngineModes: cfg.Environment == config.EnvironmentProd,
			})
		} else {
			providerClient, err = voiceprovider.NewHTTPClient(cfg.MockProviderBaseURL, cfg.MockProviderAPIKey, cfg.ProviderRequestTimeout)
		}
		if err != nil {
			return fmt.Errorf("create voice provider client: %w", err)
		}
		providerHealth, providerErr := orchestration.NewProviderHealth(databasePool, cfg.ProviderFailureThreshold, cfg.ProviderCircuitCooldown)
		if providerErr != nil {
			return fmt.Errorf("create voice provider health: %w", providerErr)
		}
		leaseTTL := time.Duration(cfg.HeartbeatInterval*5) * time.Second
		startProcessor, processorErr := orchestration.NewStartProcessor(databasePool, providerClient, providerHealth, leaseTTL)
		if processorErr != nil {
			return fmt.Errorf("create provider start processor: %w", processorErr)
		}
		if processorErr = startProcessor.ConfigureStartProtection(limiter, cfg.ProviderMaxActiveTasks); processorErr != nil {
			return fmt.Errorf("configure provider start protection: %w", processorErr)
		}
		if cfg.MemoryToolBaseURL != "" {
			workerMemoryToolSigner, signerErr := memorytool.NewTokenSigner([]byte(cfg.MemoryToolSigningSecret))
			if signerErr != nil {
				return fmt.Errorf("create worker memory tool signer: %w", signerErr)
			}
			if processorErr = startProcessor.ConfigureMemoryTool(workerMemoryToolSigner, cfg.MemoryToolBaseURL, cfg.MemoryToolTokenTTL); processorErr != nil {
				return fmt.Errorf("configure provider memory tool: %w", processorErr)
			}
		}
		outboxWorker.Register("conversation.agent_start_requested", startProcessor.Handle)
		outboxWorker.Register("conversation.agent_start_reconcile_requested", startProcessor.Handle)
		stopProcessor, processorErr := orchestration.NewStopProcessor(databasePool, providerClient, providerHealth)
		if processorErr != nil {
			return fmt.Errorf("create provider stop processor: %w", processorErr)
		}
		outboxWorker.Register("conversation.agent_stop_requested", stopProcessor.Handle)
		outboxWorker.Register("conversation.agent_stop_reconcile_requested", stopProcessor.Handle)
		var contentGenerator extraction.Generator = contentModelRegistry
		if cfg.DifyEnabled {
			contentGenerator, err = extraction.NewDifyGenerator(cfg.DifyBaseURL, cfg.DifyExtractionAPIKey, cfg.DifyBridgeURL, cfg.DifyBridgeSecret)
			if err != nil {
				return err
			}
		}
		var contentSafety contentpolicy.Classifier = contentpolicy.LocalClassifier{}
		if cfg.ContentSafetyProvider == "managed_http" {
			managedSafety, safetyErr := contentpolicy.NewManagedClassifier(cfg.ContentSafetyEndpoint, cfg.ContentSafetyAPIKey, cfg.ContentSafetyPolicyRef, cfg.ContentSafetyTimeout)
			if safetyErr != nil {
				return fmt.Errorf("create managed content safety classifier: %w", safetyErr)
			}
			contentSafety = managedSafety
		}
		extractionProcessor, processorErr := extraction.NewProcessorWithSafety(databasePool, contentGenerator, contentSafety)
		if processorErr != nil {
			return fmt.Errorf("create content extraction processor: %w", processorErr)
		}
		// CALL-MEMORY-1.1 has one authoritative extractor. Historical AddSession
		// queue entries must not repopulate independently extracted provider memory.
		outboxWorker.Register(vikingmemory.SyncEvent, func(context.Context, outbox.Event) error { return nil })
		outboxWorker.Register("conversation.content_extraction_requested", extractionProcessor.Handle)
		outboxWorker.Register(extraction.DirectStoryEvent, extractionProcessor.HandleDirectStory)
		outboxWorker.Register(extraction.ImportStoryEvent, extractionProcessor.HandleImport)
		memoirService.Safety = contentSafety
		go memoirService.Run(ctx)
		callMemoryService.ConfigureExtractor(contentModelRegistry)
		go callMemoryService.Run(ctx)
		memoryIndexProcessor, processorErr := memoryindex.NewProcessor(databasePool, memoryEmbedding)
		if processorErr != nil {
			return fmt.Errorf("create memory index processor: %w", processorErr)
		}
		outboxWorker.Register("memory.index_requested", memoryIndexProcessor.Handle)
		if providerMemoryClient != nil {
			providerMemoryProcessor, providerMemoryErr := providermemory.NewProcessor(databasePool, providerMemoryClient)
			if providerMemoryErr != nil {
				return fmt.Errorf("create provider memory processor: %w", providerMemoryErr)
			}
			outboxWorker.Register("memory.provider_sync_requested", providerMemoryProcessor.HandleSync)
			outboxWorker.Register("memory.provider_delete_requested", providerMemoryProcessor.HandleDelete)
		}
		memoryDeleteProcessor, processorErr := memorylifecycle.NewDeleteProcessor(databasePool)
		if processorErr != nil {
			return fmt.Errorf("create memory delete processor: %w", processorErr)
		}
		outboxWorker.Register("memory.delete_requested", memoryDeleteProcessor.Handle)
		privacyAccessProcessor, processorErr := privacy.NewAccessProcessor(databasePool)
		if processorErr != nil {
			return fmt.Errorf("create privacy access processor: %w", processorErr)
		}
		outboxWorker.Register("privacy.access_requested", privacyAccessProcessor.Handle)
		privacyDeleteProcessor, processorErr := privacy.NewDeleteProcessor(databasePool, redisClient, exportStore, providerMemoryClient)
		if processorErr != nil {
			return fmt.Errorf("create privacy delete processor: %w", processorErr)
		}
		if vikingMemory != nil {
			privacyDeleteProcessor.ConfigureVikingMemory(vikingMemory)
		}
		outboxWorker.Register("privacy.delete_requested", privacyDeleteProcessor.Handle)
		exportProcessor, processorErr := dataexport.NewProcessor(databasePool, exportStore, cfg.ExportObjectTTL)
		if processorErr != nil {
			return fmt.Errorf("create data export processor: %w", processorErr)
		}
		outboxWorker.Register("export.generate_requested", exportProcessor.Handle)
		if exportStore != nil {
			mediaProcessor, mediaErr := media.NewProcessor(databasePool, exportStore, audit.NewRepository(), cfg.MediaLocalDeleteGrace)
			if mediaErr != nil {
				return fmt.Errorf("create media verification processor: %w", mediaErr)
			}
			outboxWorker.Register("media.verify_requested", mediaProcessor.Handle)
		}
		expiryScanner, processorErr := dataexport.NewExpiryScanner(databasePool, exportStore, logger, cfg.WorkerPollInterval)
		if processorErr != nil {
			return fmt.Errorf("create export expiry scanner: %w", processorErr)
		}
		go expiryScanner.Run(ctx)
		go outboxWorker.Run(ctx)
		recoveryScanner, recoveryErr := orchestration.NewRecoveryScanner(
			databasePool, outbox.NewStore(databasePool), logger,
			orchestration.RecoveryOptions{
				ScanInterval: cfg.RecoveryScanInterval, StartingAfter: cfg.RecoveryStartingAfter,
				StoppingAfter: cfg.RecoveryStoppingAfter, BatchSize: 50,
			},
		)
		if recoveryErr != nil {
			return fmt.Errorf("create conversation recovery scanner: %w", recoveryErr)
		}
		go recoveryScanner.Run(ctx)
		alertScanner, alertErr := alerts.NewScanner(databasePool, logger, cfg.WorkerPollInterval)
		if alertErr != nil {
			return fmt.Errorf("create alert scanner: %w", alertErr)
		}
		go alertScanner.Run(ctx)
		go func() {
			ticker := time.NewTicker(time.Duration(cfg.HeartbeatInterval) * time.Second)
			defer ticker.Stop()
			for {
				select {
				case <-ctx.Done():
					return
				case <-ticker.C:
					staleBefore := time.Now().UTC().Add(-2 * leaseTTL)
					if _, err := webEventService.ExpireDevicePresence(ctx, staleBefore, 1000); err != nil {
						logger.Warn("device presence expiry scan failed", "error", err)
					}
					if _, err := webEventService.DeleteExpired(ctx, 5000); err != nil {
						logger.Warn("expired web event cleanup failed", "error", err)
					}
				}
			}
		}()
	}
	router := httpapi.NewRouter(httpapi.RouterOptions{
		OnboardingService:                onboardingService,
		ServiceName:                      serviceName,
		Logger:                           logger,
		RateLimiter:                      rateLimiterAdapter{limiter},
		Readiness:                        readiness,
		TokenVerifier:                    tokenVerifier,
		IdentityRepository:               identityRepository,
		CallMemoryService:                callMemoryService,
		MemoirService:                    memoirService,
		SaaSService:                      saasService,
		DeviceVerifier:                   deviceVerifier,
		DeviceService:                    deviceRepository,
		DeviceTelemetrySubmitter:         deviceTelemetryService,
		MediaService:                     mediaService,
		WebEventService:                  webEventService,
		SharingService:                   sharingService,
		OpsConfigService:                 opsConfigService,
		BillingService:                   billingService,
		AlertService:                     alertService,
		DiagnosticService:                diagnosticService,
		OpsAccessService:                 opsAccessService,
		OpsDeviceService:                 opsDeviceService,
		ContentModelService:              contentModelService,
		ConversationCreator:              conversationCreator,
		ConversationActivator:            conversationCreator,
		DeviceConversationStateReader:    conversationCreator,
		ConversationHeartbeater:          conversationHeartbeater,
		ConversationStopper:              conversationCreator,
		ConversationTranscriptSubmitter:  conversationTranscriptSubmitter,
		HardwareAgentSessionService:      conversationCreator,
		HardwareAgentProvisioningService: hardwareProvisioningService,
		DeviceFactoryResetService:        deviceFactoryResetService,
		ProviderCallbackSecret:           []byte(cfg.ProviderCallbackSecret),
		ProviderCallbackSkew:             cfg.ProviderCallbackSkew,
		ProviderCallbackProcessor:        providerEventProcessor,
		NativeSubtitleSecret:             []byte(cfg.SubtitleCallbackSecret),
		NativeSubtitleProcessor:          orchestration.NewNativeSubtitleService(databasePool),
		PlatformHealthCallbackSecret:     []byte(cfg.PlatformHealthSecret),
		PlatformHealthCallbackSkew:       cfg.PlatformHealthSkew,
		PlatformHealthProcessor:          diagnosticService,
		MemoryToolService:                memoryToolService,
		MemoryToolSigner:                 memoryToolSigner,
	})
	server := &http.Server{
		Addr:              cfg.HTTPAddr,
		Handler:           otelhttp.NewHandler(httpapi.Middleware(logger, serviceName, 1<<20, rateLimiterAdapter{limiter}, router), serviceName),
		ReadHeaderTimeout: 5 * time.Second,
		ReadTimeout:       15 * time.Second,
		WriteTimeout:      60 * time.Second,
		IdleTimeout:       60 * time.Second,
	}

	serverError := make(chan error, 1)
	go externalProbeRunner.Run(ctx)
	go func() {
		logger.Info("service listening", "address", cfg.HTTPAddr, "environment", cfg.Environment)
		serverError <- server.ListenAndServe()
	}()

	select {
	case <-ctx.Done():
		shutdownContext, cancel := context.WithTimeout(context.Background(), cfg.ShutdownTimeout)
		defer cancel()
		logger.Info("service shutting down")
		return server.Shutdown(shutdownContext)
	case err := <-serverError:
		if err == http.ErrServerClosed {
			return nil
		}
		return err
	}
}

func newExternalProbeRunner(cfg config.Config, logger *slog.Logger, pool *pgxpool.Pool) (*externalprobe.Runner, error) {
	checks, err := externalProbeChecks(cfg)
	if err != nil {
		return nil, err
	}
	dependencies := make([]string, 0, len(checks))
	for dependency := range checks {
		dependencies = append(dependencies, dependency)
	}
	reconcileContext, cancel := context.WithTimeout(context.Background(), cfg.DependencyTimeout)
	err = externalprobe.ReconcilePostgresState(reconcileContext, pool, cfg.ServiceName, dependencies)
	cancel()
	if err != nil {
		return nil, err
	}
	observer, err := externalprobe.PostgresObserver(pool, cfg.ServiceName)
	if err != nil {
		return nil, err
	}
	return externalprobe.NewRunner(cfg.ExternalProbeInterval, cfg.DependencyTimeout, checks, logger, observer)
}

func externalProbeChecks(cfg config.Config) (map[string]externalprobe.CheckFunc, error) {
	checks := map[string]externalprobe.CheckFunc{}
	if cfg.ServiceName == "api" {
		jwksCheck, err := externalprobe.JWKS(cfg.AuthJWKSURL, nil)
		if err != nil {
			return nil, err
		}
		checks["oidc_jwks"] = observability.InstrumentDependencyCheck(cfg.ServiceName, "oidc_jwks", jwksCheck)
	}
	if (cfg.ServiceName == "api" || cfg.ServiceName == "worker") && cfg.EmbeddingProvider == "ark" {
		arkCheck, err := externalprobe.TLSReachability(cfg.ArkBaseURL, nil)
		if err != nil {
			return nil, err
		}
		checks["ark_tls"] = observability.InstrumentDependencyCheck(cfg.ServiceName, "ark_tls", arkCheck)
	}
	if cfg.ServiceName == "worker" && cfg.VoiceProvider == "volcengine" {
		rtcCheck, err := externalprobe.TLSReachability(cfg.VolcRTCBaseURL, nil)
		if err != nil {
			return nil, err
		}
		checks["volc_rtc_tls"] = observability.InstrumentDependencyCheck(cfg.ServiceName, "volc_rtc_tls", rtcCheck)
	}
	if cfg.ServiceName == "worker" && cfg.ContentSafetyProvider == "managed_http" {
		contentSafetyCheck, err := externalprobe.TLSReachability(cfg.ContentSafetyEndpoint, nil)
		if err != nil {
			return nil, err
		}
		checks["content_safety_tls"] = observability.InstrumentDependencyCheck(cfg.ServiceName, "content_safety_tls", contentSafetyCheck)
	}
	return checks, nil
}

func newMemoryEmbeddingProvider(cfg config.Config) (embeddingprovider.Provider, error) {
	if cfg.EmbeddingProvider == "ark" {
		return embeddingprovider.NewArk(cfg.ArkBaseURL, cfg.ArkAPIKey, cfg.ArkEmbeddingModel, cfg.ArkRequestTimeout)
	}
	return embeddingprovider.Local{}, nil
}

type rateLimiterAdapter struct{ limiter *ratelimit.Limiter }

func (a rateLimiterAdapter) Allow(ctx context.Context, scope, identifier string) (httpapi.RateLimitResult, error) {
	result, err := a.limiter.Allow(ctx, scope, identifier)
	return httpapi.RateLimitResult{Allowed: result.Allowed, Remaining: result.Remaining, RetryAfter: result.RetryAfter}, err
}

func newLogger(cfg config.Config) *slog.Logger {
	level := slog.LevelInfo
	switch cfg.LogLevel {
	case "debug":
		level = slog.LevelDebug
	case "warn":
		level = slog.LevelWarn
	case "error":
		level = slog.LevelError
	}
	return slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: level})).With("service", cfg.ServiceName)
}