Phase/9c migration reconciliation operations

.env.example

@@ -151,6 +151,23 @@ OTEL_SERVICE_NAME=mosaic-api
 # Empty means record in-process without exporting. Export failure never stops
 # Mosaic.
 OTEL_EXPORTER_OTLP_ENDPOINT=
+# Transport for the endpoint above: http/protobuf (default, usually port 4318)
+# or grpc (usually port 4317). Ignored when no endpoint is set.
+OTEL_EXPORTER_OTLP_PROTOCOL=
+# Export headers as key1=value1,key2=value2 with percent-encoded values. This is
+# where a hosted collector's token goes, so it is a secret:
+# OTEL_EXPORTER_OTLP_HEADERS=authorization=Bearer%20token
+OTEL_EXPORTER_OTLP_HEADERS=
+# Skip collector certificate verification. https:// endpoints only, for a
+# private CA or self-signed certificate.
+MOSAIC_OTEL_EXPORTER_TLS_SKIP_VERIFY=false
+# Acknowledge a plaintext or unverified collector connection outside development
+# and test. Without it, those are rejected at startup.
+MOSAIC_OTEL_EXPORTER_ALLOW_INSECURE=false
+# Ship log records to the collector alongside traces and metrics. Logs always
+# keep going to stdout as well; set false when log volume is the cost that
+# matters. Ignored when no endpoint is set.
+MOSAIC_OTEL_LOGS_ENABLED=true
 
 # =============================================================================
 # Object storage (S3-compatible)
@@ -177,6 +194,18 @@ MOSAIC_ASSET_MAX_UPLOAD_BYTES=10485760
 # Must be an absolute HTTPS URL without credentials.
 MOSAIC_PUBLIC_ASSET_BASE_URL=https://localhost:8443/v1/sdk/assets
 
+# Phase 9C migration source evidence uses a separate private bucket and a
+# separate versioned AES-256 keyring. Never reuse the public Asset bucket or
+# MOSAIC_PROVIDER_CREDENTIAL_KEYRING. The keyring has the same JSON envelope
+# shape, but must contain independently generated key material:
+# {"version":1,"activeKeyId":"migration-2026-01","keys":{"migration-2026-01":"<base64url-32-byte-key>"}}
+MOSAIC_BILLING_MIGRATION_ENABLED=false
+MOSAIC_BILLING_MIGRATION_SOURCE_KEYRING=
+MOSAIC_BILLING_MIGRATION_SOURCE_BUCKET=mosaic-migration-private
+MOSAIC_BILLING_MIGRATION_SOURCE_CHUNK_BYTES=262144
+MOSAIC_BILLING_MIGRATION_SOURCE_OPERATION_TIMEOUT=5m
+MOSAIC_BILLING_MIGRATION_WORKER_POLL_INTERVAL=1s
+
 # =============================================================================
 # Browser sessions
 # =============================================================================

apps/api/cmd/api/main.go

@@ -20,6 +20,7 @@ import (
 	"github.com/Mujhtech/mosaic/apps/api/internal/billingcustomer"
 	"github.com/Mujhtech/mosaic/apps/api/internal/billingdiagnostics"
 	"github.com/Mujhtech/mosaic/apps/api/internal/billinggrant"
+	"github.com/Mujhtech/mosaic/apps/api/internal/billingmigration"
 	"github.com/Mujhtech/mosaic/apps/api/internal/billingoperator"
 	"github.com/Mujhtech/mosaic/apps/api/internal/billingprojection"
 	"github.com/Mujhtech/mosaic/apps/api/internal/billingrestore"
@@ -38,6 +39,9 @@ import (
 	"github.com/Mujhtech/mosaic/apps/api/internal/platform/billingdiagnosticspostgres"
 	"github.com/Mujhtech/mosaic/apps/api/internal/platform/billinggrantpostgres"
 	"github.com/Mujhtech/mosaic/apps/api/internal/platform/billingkeys"
+	"github.com/Mujhtech/mosaic/apps/api/internal/platform/billingmigrationpostgres"
+	"github.com/Mujhtech/mosaic/apps/api/internal/platform/billingmigrationrepair"
+	"github.com/Mujhtech/mosaic/apps/api/internal/platform/billingmigrationvalidation"
 	"github.com/Mujhtech/mosaic/apps/api/internal/platform/billingoperatorpostgres"
 	"github.com/Mujhtech/mosaic/apps/api/internal/platform/billingpostgres"
 	"github.com/Mujhtech/mosaic/apps/api/internal/platform/billingprojectionpostgres"
@@ -108,6 +112,10 @@ func closeSchemas(readers map[protocolschema.Schema]io.ReadCloser) {
 	}
 }
 
+func repairExecutionEnabled(cfg config.Config) bool {
+	return cfg.Billing.Enabled && cfg.Migration.Enabled
+}
+
 func run() (runErr error) {
 	cfg, err := config.Load()
 	if err != nil {
@@ -119,11 +127,16 @@ func run() (runErr error) {
 		return fmt.Errorf("configure logging: %w", err)
 	}
 	build := buildinfo.Current()
-	logger = logger.With().
-		Str("service", cfg.Telemetry.ServiceName).
-		Str("environment", cfg.Environment).
-		Str("version", build.Version).
-		Logger()
+	// Applied to whichever logger ends up in use, so the local stream and the
+	// exported records carry the same service identity.
+	withServiceContext := func(base zerolog.Logger) zerolog.Logger {
+		return base.With().
+			Str("service", cfg.Telemetry.ServiceName).
+			Str("environment", cfg.Environment).
+			Str("version", build.Version).
+			Logger()
+	}
+	logger = withServiceContext(logger)
 
 	runContext, stop := signal.NotifyContext(
 		context.Background(),
@@ -136,11 +149,30 @@ func run() (runErr error) {
 		ServiceName:  cfg.Telemetry.ServiceName,
 		Environment:  cfg.Environment,
 		OTLPEndpoint: cfg.Telemetry.OTLPEndpoint,
-		Logger:       logger,
+		OTLPProtocol: cfg.Telemetry.OTLPProtocol,
+		OTLPHeaders:  cfg.Telemetry.OTLPHeaders,
+		// Startup validation has already refused an unverified collector in a
+		// production-like environment unless it was explicitly acknowledged.
+		OTLPTLSSkipVerify: cfg.Telemetry.TLSSkipVerify,
+		DisableLogExport:  !cfg.Telemetry.ExportLogs(),
+		// Telemetry keeps the local-only logger: routing its own export failures
+		// through the exporting logger would feed the failing exporter.
+		Logger: logger,
 	})
 	if err != nil {
 		return fmt.Errorf("configure telemetry: %w", err)
 	}
+	if cfg.Telemetry.ExportLogs() {
+		// Swapped in only once the logger provider exists, so every record this
+		// logger writes locally also reaches the collector.
+		exportingLogger, err := logging.NewExporting(
+			cfg.Log.Level, cfg.Log.Format, os.Stdout, cfg.Telemetry.ServiceName,
+		)
+		if err != nil {
+			return fmt.Errorf("configure log export: %w", err)
+		}
+		logger = withServiceContext(exportingLogger)
+	}
 	defer func() {
 		// Telemetry flush has its own budget so a slow collector cannot consume
 		// the HTTP drain budget or delay closing the database pool.
@@ -276,6 +308,14 @@ func run() (runErr error) {
 	var billingAccessService *billingaccess.Service
 	var billingDiagnosticsService *billingdiagnostics.Service
 	var billingGrantService *billinggrant.Service
+	var billingMigrationService *billingmigration.Service
+	var billingMigrationSourcePull *billingmigration.SourcePullService
+	var billingMigrationOperations *billingmigration.OperationsService
+	var billingMigrationRedelivery *billingmigration.RedeliveryService
+	var billingMigrationReads *billingmigration.OperationalReadService
+	var billingMigrationStabilization *billingmigration.StabilizationService
+	var billingMigrationRollbackReadiness *billingmigration.RollbackReadinessService
+	var billingMigrationRepairOnline bool
 	var billingRestoreService *billingrestore.Service
 	var billingCustomerService *billingcustomer.Service
 	var billingOperatorService *billingoperator.Service
@@ -287,6 +327,29 @@ func run() (runErr error) {
 		if err != nil {
 			return fmt.Errorf("configure billing credential encryption: %w", err)
 		}
+		migrationRevenueCatClient, err := revenuecat.New(revenuecat.Config{
+			BaseURL: cfg.Providers.RevenueCatBaseURL, RequestTimeout: cfg.Providers.RequestTimeout,
+			OperationTimeout: cfg.Providers.OperationTimeout, ConnectTimeout: cfg.Providers.ConnectTimeout,
+			MaxResponseBytes: cfg.Providers.MaxResponseBytes, MaxAttempts: cfg.Providers.MaxAttempts,
+		})
+		if err != nil {
+			return fmt.Errorf("configure RevenueCat migration adapter: %w", err)
+		}
+		migrationRepository := billingmigrationpostgres.New(databasePool)
+		billingMigrationService = billingmigration.NewService(
+			migrationRepository, billingCipher, migrationRevenueCatClient)
+		// Non-repair operator controls remain readable/usable when the optional
+		// execution plane is disabled; repair itself fails closed through a nil
+		// executor and the explicit transport gate below.
+		billingMigrationOperations = billingmigration.NewOperationsService(
+			migrationRepository, migrationRepository, nil)
+		billingMigrationRedelivery = billingmigration.NewRedeliveryService(migrationRepository, nil)
+		billingMigrationReads = billingmigration.NewOperationalReadService(migrationRepository, migrationRepository)
+		billingMigrationStabilization = billingmigration.NewStabilizationService(migrationRepository, migrationRepository)
+		billingMigrationRollbackReadiness = billingmigration.NewRollbackReadinessService(migrationRepository, migrationRepository)
+		if cfg.Migration.Enabled {
+			billingMigrationSourcePull = billingmigration.NewSourcePullService(migrationRepository, nil)
+		}
 		// The Apple root is compiled in, so a broken embed fails startup rather
 		// than the first notification.
 		verifier, err := appstorejws.NewVerifier()
@@ -352,6 +415,17 @@ func run() (runErr error) {
 			billing.WithNotificationBaseURL(cfg.Billing.NotificationBaseURL),
 			billing.WithSeam(billingseam.New(billingCustomerService, billingAccessService),
 				billingseam.New(billingCustomerService, billingAccessService)))
+		if repairExecutionEnabled(cfg) {
+			migrationRepairExecutor := billingmigrationrepair.NewProductionExecutor(
+				billingmigrationrepair.NewPostgresStore(databasePool),
+				billingmigrationvalidation.New(billingService),
+				billingProjectionService,
+				projectionRepository,
+			)
+			billingMigrationOperations = billingmigration.NewOperationsService(
+				migrationRepository, migrationRepository, migrationRepairExecutor)
+			billingMigrationRepairOnline = true
+		}
 		billingGrantService = billinggrant.NewService(billinggrantpostgres.New(databasePool))
 		// The operator surface reads through the same repositories the trusted
 		// APIs read through, so the dashboard and an application backend see one
@@ -392,35 +466,43 @@ func run() (runErr error) {
 		TrustedProxyCIDRs: cfg.HTTP.TrustedProxyCIDRs,
 		EnableHSTS:        cfg.ProductionLike(),
 	}, logger, httpserver.Dependencies{
-		BrowserAuth:            browserAuthService,
-		BrowserAuthConfig:      browserauthhttp.Config{CookieSecure: cfg.BrowserAuth.CookieSecure, CookieDomain: cfg.BrowserAuth.CookieDomain, AllowedOrigins: cfg.HTTP.CORSAllowedOrigins, RateLimiter: authenticationLimiter},
-		CloudWorkspace:         workspaceService,
-		HostedPublishing:       publishingService,
-		PlacementDecision:      placementDecisionService,
-		PrincipalResolver:      authn.NewBrowserSessionResolver(browserAuthService),
-		DeliveryLimiter:        deliveryLimiter,
-		Analytics:              analyticsService,
-		AnalyticsIPLimiter:     analyticsIPLimiter,
-		AnalyticsKeyLimiter:    analyticsKeyLimiter,
-		AnalyticsEventLimiter:  analyticsEventLimiter,
-		Experiment:             experimentService,
-		Billing:                billingService,
-		BillingAccess:          billingAccessService,
-		BillingDiagnostics:     billingDiagnosticsService,
-		BillingGrant:           billingGrantService,
-		BillingRestore:         billingRestoreService,
-		BillingCustomer:        billingCustomerService,
-		BillingOperator:        billingOperatorService,
-		BillingWebhook:         billingWebhookService,
-		BillingIPLimiter:       billingIPLimiter,
-		BillingKeyLimiter:      billingKeyLimiter,
-		EntitlementSyncLimiter: entitlementSyncLimiter,
-		APILimiter:             apiLimiter,
-		DecisionLimiter:        decisionLimiter,
-		UploadLimiter:          uploadLimiter,
-		ExportLimiter:          exportLimiter,
-		Readiness:              readiness,
-		ReadinessChecker:       database.HealthChecker{Pinger: databasePool},
+		BrowserAuth:                       browserAuthService,
+		BrowserAuthConfig:                 browserauthhttp.Config{CookieSecure: cfg.BrowserAuth.CookieSecure, CookieDomain: cfg.BrowserAuth.CookieDomain, AllowedOrigins: cfg.HTTP.CORSAllowedOrigins, RateLimiter: authenticationLimiter},
+		CloudWorkspace:                    workspaceService,
+		HostedPublishing:                  publishingService,
+		PlacementDecision:                 placementDecisionService,
+		PrincipalResolver:                 authn.NewBrowserSessionResolver(browserAuthService),
+		DeliveryLimiter:                   deliveryLimiter,
+		Analytics:                         analyticsService,
+		AnalyticsIPLimiter:                analyticsIPLimiter,
+		AnalyticsKeyLimiter:               analyticsKeyLimiter,
+		AnalyticsEventLimiter:             analyticsEventLimiter,
+		Experiment:                        experimentService,
+		Billing:                           billingService,
+		BillingAccess:                     billingAccessService,
+		BillingDiagnostics:                billingDiagnosticsService,
+		BillingGrant:                      billingGrantService,
+		BillingMigration:                  billingMigrationService,
+		BillingMigrationSourcePull:        billingMigrationSourcePull,
+		BillingMigrationOperations:        billingMigrationOperations,
+		BillingMigrationRedelivery:        billingMigrationRedelivery,
+		BillingMigrationReads:             billingMigrationReads,
+		BillingMigrationStabilization:     billingMigrationStabilization,
+		BillingMigrationRollbackReadiness: billingMigrationRollbackReadiness,
+		BillingMigrationRepairOnline:      billingMigrationRepairOnline,
+		BillingRestore:                    billingRestoreService,
+		BillingCustomer:                   billingCustomerService,
+		BillingOperator:                   billingOperatorService,
+		BillingWebhook:                    billingWebhookService,
+		BillingIPLimiter:                  billingIPLimiter,
+		BillingKeyLimiter:                 billingKeyLimiter,
+		EntitlementSyncLimiter:            entitlementSyncLimiter,
+		APILimiter:                        apiLimiter,
+		DecisionLimiter:                   decisionLimiter,
+		UploadLimiter:                     uploadLimiter,
+		ExportLimiter:                     exportLimiter,
+		Readiness:                         readiness,
+		ReadinessChecker:                  database.HealthChecker{Pinger: databasePool},
 	})
 
 	server := &http.Server{

apps/api/cmd/api/main_test.go

@@ -4,8 +4,28 @@ import (
 	"os"
 	"strings"
 	"testing"
+
+	"github.com/Mujhtech/mosaic/apps/api/internal/platform/config"
 )
 
+func TestRepairExecutionRequiresBillingAndMigrationOptIn(t *testing.T) {
+	for _, test := range []struct {
+		billing, migration, want bool
+	}{
+		{billing: false, migration: false, want: false},
+		{billing: true, migration: false, want: false},
+		{billing: false, migration: true, want: false},
+		{billing: true, migration: true, want: true},
+	} {
+		cfg := config.Config{}
+		cfg.Billing.Enabled = test.billing
+		cfg.Migration.Enabled = test.migration
+		if got := repairExecutionEnabled(cfg); got != test.want {
+			t.Fatalf("billing=%v migration=%v enabled=%v want=%v", test.billing, test.migration, got, test.want)
+		}
+	}
+}
+
 func TestProductionWiringUsesPostgreSQLOnly(t *testing.T) {
 	source, err := os.ReadFile("main.go")
 	if err != nil {