package setupcmd

import (
	"context"
	"encoding/hex"
	"encoding/json"
	"flag"
	"fmt"
	"io"
	"os"
	"strings"
	"time"

	"github.com/nexuscrm/backend/internal/bootstrap"
	"github.com/nexuscrm/backend/internal/interfaces/rest/downloads"
	"github.com/nexuscrm/backend/pkg/serverruntime"
	database "github.com/nexuscrm/db"
	"github.com/nexuscrm/db/pkg/bootstate"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/storage"
	controlplane "github.com/pingcap-inc/nexuscrm-controlplane"
	"github.com/pingcap-inc/nexuscrm-controlplane/packs"
	"github.com/rs/zerolog"
	"github.com/rs/zerolog/log"
)

type productDistributionRegistrar func() (serverruntime.ProductHooks, error)

type offlineCandidateApplier func(
	context.Context,
	*database.TiDBConnection,
	controlplane.OfflineCandidateRequest,
	controlplane.InstallPackOptions,
	controlplane.RunnerOptions,
) (controlplane.OfflineCandidateReceipt, error)

type stoppedSourceSetApplier func(
	context.Context,
	*database.TiDBConnection,
	controlplane.StoppedSourceSetRequest,
	controlplane.RunnerOptions,
) (controlplane.StoppedSourceSetResult, error)

const applyAllSourceRevisionPlanCapability = "nexuscrm.setup.apply-all-source-revision-plan.v2"
const applySourceSetCommand = "apply-source-set"

func Main(registerProductDistribution productDistributionRegistrar) {
	if registerProductDistribution == nil {
		registerProductDistribution = registerNoopProductDistribution
	}

	// Initialize logging
	zerolog.TimeFieldFormat = zerolog.TimeFormatUnix
	if os.Getenv("LOG_FORMAT") != "json" {
		log.Logger = log.Output(zerolog.ConsoleWriter{Out: os.Stderr, TimeFormat: time.RFC3339})
	}
	productHooks, err := registerProductDistribution()
	if err != nil {
		log.Fatal().Err(err).Msg("Failed to compose product distribution")
	}
	registrationSet := productHooks.ComponentKindRegistrationSet()
	runnerOptions := controlplane.RunnerOptions{ComponentKindRegistrationSet: registrationSet}

	if len(os.Args) < 2 {
		usage()
		os.Exit(1)
	}

	if os.Args[1] == "validate-pack" {
		os.Exit(runValidatePackWithComponentKindRegistrationSet(os.Args[2:], os.Stdout, os.Stderr, registrationSet))
	}

	if os.Args[1] == "publish-artifact" {
		os.Exit(runPublishArtifact(os.Args[2:], os.Stdout, os.Stderr))
	}

	var parsedApplyArgs *applyArgs
	switch os.Args[1] {
	case "apply", "apply-if-needed":
		args, err := parseApplyArgs(os.Args[2:], os.Stderr)
		if err != nil {
			fmt.Fprintln(os.Stderr, err)
			os.Exit(2)
		}
		parsedApplyArgs = &args
	}

	// Connect to database
	db, err := database.GetInstance()
	if err != nil {
		log.Fatal().Err(err).Msg("Failed to connect to database")
	}
	defer func() {
		if err := db.Close(); err != nil {
			log.Warn().Err(err).Msg("Failed to close database connection")
		}
	}()

	// Wire the control-plane seam ports (W5/S7) before any apply/verify/reapply
	// subcommand runs. The control-plane packages no longer import
	// internal/bootstrap; their seam vars are nil until the composition root
	// injects the concrete bootstrap-backed implementations / reader functions.
	// cmd/setup is the operator apply/verify entry point, so it MUST wire here or
	// the apply path nil-derefs. Behavior is identical to the prior in-CP default.
	serverruntime.WireControlplaneSeams()

	// Install the L0-only field-provenance classifier before any operation that
	// might touch ComputeIntrinsic. App-pack declarations are registered by the
	// source-backed install/runtime metadata context, not by process bootstrap.
	if err := bootstrap.InitFieldProvenance(bootstrap.DefaultPlatformStandardFields()); err != nil {
		log.Fatal().Err(err).Msg("Failed to initialize field-provenance classifier")
	}

	ctx := context.Background()

	switch os.Args[1] {
	case connectorRemoteIdentityRecoveryCommand:
		if code := runConnectorRemoteIdentityRecoveryCommand(
			ctx,
			db,
			productHooks,
			os.Args[2:],
			os.Stdout,
			os.Stderr,
			verifyConnectorRemoteIdentityRecoveryRuntime,
			buildConnectorRemoteIdentityRecoveryRunner,
		); code != 0 {
			os.Exit(code)
		}

	case permissionStateActivationCommand:
		if code := runPermissionStateActivationCommand(
			ctx,
			os.Args[2:],
			os.Stdout,
			os.Stderr,
			func() (permissionStateActivator, error) {
				return persistence.NewPermissionStateRepository(db.DB())
			},
		); code != 0 {
			os.Exit(code)
		}

	case permissionStateInitializationCommand:
		if code := runPermissionStateInitializationCommand(
			ctx,
			os.Args[2:],
			os.Stdout,
			os.Stderr,
			func() (permissionStateInitializer, error) {
				return persistence.NewPermissionStateRepository(db.DB())
			},
		); code != 0 {
			os.Exit(code)
		}

	case "apply":
		args := *parsedApplyArgs
		options := applyInstallPackOptions(args)
		options.ApplyAllFinalVerify = setupApplyFinalVerify(productHooks, registrationSet)
		if err := preflightApplyAllSourceRevisionAdvances(ctx, db, options); err != nil {
			log.Fatal().Err(err).Msg("apply source-revision plan preflight failed")
		}
		if err := controlplane.ApplyAllWithOptionsAndRunnerOptions(ctx, db, options, runnerOptions); err != nil {
			log.Fatal().Err(err).Msg("apply failed")
		}
		if args.skipPolymorphicBackfill {
			log.Error().Msg("apply finished with polymorphic discriminator backfill skipped; this is a PARTIAL control plane and was NOT marked bootstrapped. Rerun a full `setup apply` (without --skip-polymorphic-backfill) before relying on auto start.")
			os.Exit(3)
		}
		if args.deferFormulaViewRebuild {
			log.Warn().Msg("apply finished with formula view rebuild DEFERRED: every formula view is still the previous release's compiler output and a durable rebuild debt is outstanding. Auto start refuses to serve until `setup rebuild` (inside the deployment's quiesced activation window) settles it.")
		}

	case "apply-if-needed":
		args := *parsedApplyArgs
		if args.deferFormulaViewRebuild {
			// Fail loudly rather than silently ignoring a safety flag: defer
			// is an explicit cutover-deploy contract, and the if-needed/auto
			// path is exactly what the control plane's rebuild-debt gate
			// refuses while that contract is in flight.
			fmt.Fprintln(os.Stderr, "apply-if-needed: --defer-formula-view-rebuild is not supported; deferred (cutover) applies must use the explicit `setup apply`")
			os.Exit(2)
		}
		options := applyInstallPackOptions(args)
		options.ApplyAllFinalVerify = setupApplyIfNeededFinalVerify(productHooks)
		if err := preflightApplyAllSourceRevisionAdvances(ctx, db, options); err != nil {
			log.Fatal().Err(err).Msg("apply-if-needed source-revision plan preflight failed")
		}
		applied, err := controlplane.ApplyAllIfNeededWithOptionsAndRunnerOptions(ctx, db, options, runnerOptions)
		if err != nil {
			log.Fatal().Err(err).Msg("apply-if-needed failed")
		}
		if args.skipPolymorphicBackfill {
			log.Error().Msg("apply-if-needed finished with polymorphic discriminator backfill skipped; this is a PARTIAL control plane and was NOT marked bootstrapped. Rerun a full apply (without --skip-polymorphic-backfill) before relying on auto start.")
			os.Exit(3)
		}
		if tenantOverlayApplyRequested() {
			// Baseline control-plane apply may skip on restart because the
			// server's readiness controlplane.Verify is a fail-closed backstop
			// for baseline convergence. The tenant overlay has NO such complete
			// backstop — startup verify is tenant-agnostic and the overlay's own
			// Verify only checks object/field shape, not the record types,
			// auto-number, action/related-list overlays, insert/child-clone
			// rules, object SharingModel, or tenant flows that apply-tenant also
			// writes. So the overlay cannot safely be skipped on the strength of
			// a marker; instead it is re-applied here (idempotent reconcile),
			// which converges every overlay surface and heals drift rather than
			// silently serving it. This runs only when the operator opted into
			// overlay mutation via NEXUSCRM_APPLY_TENANT_OVERLAY=1; a pure
			// no-mutation restart leaves that flag unset. Making the overlay
			// skip-on-restart safe would require a total tenant readiness verify
			// (tracked as follow-up); until then, converge by re-applying.
			if err := runTenantOverlayApply(ctx, runnerOptions); err != nil {
				log.Fatal().Err(err).Msg("apply-if-needed tenant overlay failed")
			}
			break
		}
		if !applied {
			log.Info().Msg("apply-if-needed: full control-plane apply already completed; no-op")
		}

	case "apply-offline-candidate":
		if code := runApplyOfflineCandidateCommand(
			ctx,
			db,
			productHooks,
			registrationSet,
			os.Args[2:],
			os.Stdout,
			os.Stderr,
			controlplane.ApplyOfflineCandidateWithRunnerOptions,
		); code != 0 {
			os.Exit(code)
		}

	case applySourceSetCommand:
		if code := runApplySourceSetCommand(
			ctx,
			db,
			productHooks,
			os.Args[2:],
			os.Stdout,
			os.Stderr,
			controlplane.ApplyStoppedSourceSetWithRunnerOptions,
		); code != 0 {
			os.Exit(code)
		}

	case routingIdentityMappingPreflightCommand:
		if code := runRoutingIdentityMappingPreflightCommand(
			ctx,
			db.DB(),
			os.Args[2:],
			os.Stdout,
			os.Stderr,
			inspectRoutingIdentityMapping,
		); code != 0 {
			os.Exit(code)
		}

	case "apply-source-convergence":
		args, err := parseApplySourceConvergenceArgs(os.Args[2:], os.Stderr)
		if err != nil {
			fmt.Fprintln(os.Stderr, err)
			os.Exit(2)
		}
		if err := controlplane.ApplySourceConvergenceWithRunnerOptions(ctx, db, controlplane.InstallPackOptions{
			SkipPolymorphicDiscriminatorBackfill: args.skipPolymorphicBackfill,
			RepairLegacyUniqueDuplicates:         args.repairLegacyUniqueDuplicates,
		}, runnerOptions); err != nil {
			log.Fatal().Err(err).Msg("apply-source-convergence failed")
		}
		if args.skipPolymorphicBackfill {
			log.Error().Msg("apply-source-convergence finished with polymorphic discriminator backfill skipped; this is a partial operator recovery result, not a fully repaired control plane. Rerun setup repair or setup apply without --skip-polymorphic-backfill before claiming readiness")
			os.Exit(3)
		}

	case "init-l0":
		if err := controlplane.InitL0(ctx, db); err != nil {
			log.Fatal().Err(err).Msg("init-l0 failed")
		}

	case "ensure-l0-schema-ddl":
		// Stopped L0 upgrades may add neutral system tables without changing any
		// L1/L2 metadata. Converge only the additive physical L0 schema here;
		// application installation remains owned by the pack lifecycle commands.
		if _, err := controlplane.EnsureL0SchemaDDL(ctx, db); err != nil {
			log.Fatal().Err(err).Msg("ensure-l0-schema-ddl failed")
		}

	case "apply-metadata":
		if _, err := controlplane.ApplyMetadataWithLock(ctx, db); err != nil {
			log.Fatal().Err(err).Msg("apply-metadata failed")
		}

	case "apply-packs":
		r, err := controlplane.NewRunnerWithOptions(db, runnerOptions)
		if err != nil {
			log.Fatal().Err(err).Msg("Failed to create control-plane runner")
		}
		if err := r.RefreshCache(ctx); err != nil {
			log.Fatal().Err(err).Msg("Failed to refresh cache")
		}
		if err := controlplane.ApplyPacksWithLock(ctx, r, !isProduction()); err != nil {
			log.Fatal().Err(err).Msg("apply-packs failed")
		}

	case "apply-tenant":
		// Layers tenant-specific metadata on top of an already-applied
		// platform baseline. Deployment tooling may invoke this from
		// infra/entrypoint.sh apply modes only when
		// NEXUSCRM_APPLY_TENANT_OVERLAY=1.
		// Greenfield orgs never run this.
		if err := runTenantOverlayApplyWithDB(ctx, db, runnerOptions); err != nil {
			log.Fatal().Err(err).Msg("apply-tenant failed")
		}

	case "install-pack":
		source, err := parseInstallPackArgs(os.Args[2:], os.Stderr)
		if err != nil {
			fmt.Fprintln(os.Stderr, err)
			os.Exit(2)
		}
		ownerPack, err := setupPackNameFromSource(source)
		if err != nil {
			log.Fatal().Err(err).Msg("resolve install-pack source failed")
		}
		r, err := controlplane.NewRunnerWithOptions(db, runnerOptions)
		if err != nil {
			log.Fatal().Err(err).Msg("Failed to create control-plane runner")
		}
		if err := r.RefreshCache(ctx); err != nil {
			log.Fatal().Err(err).Msg("Failed to refresh cache")
		}
		if err := controlplane.InstallPackFromSource(ctx, r, source); err != nil {
			log.Fatal().Err(err).Msg("install-pack failed")
		}
		if err := verifyRunnerMetadataProgramsWithComponentKindRegistrationSet(ctx, r, registrationSet, ownerPack); err != nil {
			log.Fatal().Err(err).Msg("metadata program check failed")
		}

	case "reapply-pack":
		args, err := parseReapplyPackArgs(os.Args[2:], os.Stderr)
		if err != nil {
			fmt.Fprintln(os.Stderr, err)
			os.Exit(2)
		}
		options := controlplane.InstallPackOptions{
			SkipAttributionPreflight:             args.skipAttributionPreflight,
			SkipPolymorphicDiscriminatorBackfill: args.skipPolymorphicBackfill,
			RepairLegacyUniqueDuplicates:         args.repairLegacyUniqueDuplicates,
			VerifyAfterApply:                     args.verifyPack,
		}
		if args.verifyPack {
			options.CandidateFinalVerify = func(ctx context.Context, runner *controlplane.Runner, candidates []packs.Pack) error {
				return verifyCandidateMetadataProgramsWithComponentKindRegistrationSet(ctx, runner, registrationSet, candidates)
			}
		}
		if args.allowSourceRevisionAdvance {
			options.SourceRevisionAdvance = &controlplane.SourceRevisionAdvanceAuthorization{
				FromRevision: args.fromRevision,
				ToRevision:   args.toRevision,
			}
		}
		db, err = controlplane.ReapplyPackFromSourceWithSchemaReadinessAndRunnerOptions(ctx, db, args.source, options, runnerOptions)
		if err != nil {
			log.Fatal().Err(err).Msg("reapply-pack failed")
		}

	case "remove-pack-owned-field":
		args, err := parseRemovePackOwnedFieldArgs(os.Args[2:], os.Stderr)
		if err != nil {
			fmt.Fprintln(os.Stderr, err)
			os.Exit(2)
		}
		r, err := controlplane.NewRunnerWithOptions(db, runnerOptions)
		if err != nil {
			log.Fatal().Err(err).Msg("Failed to create control-plane runner")
		}
		if err := r.RefreshCache(ctx); err != nil {
			log.Fatal().Err(err).Msg("Failed to refresh cache")
		}
		if err := controlplane.RemovePackOwnedField(ctx, r, args.pack, args.objectAPIName, args.fieldAPIName); err != nil {
			log.Fatal().Err(err).Msg("remove-pack-owned-field failed")
		}
		log.Info().Str("pack", args.pack).Str("object", args.objectAPIName).Str("field", args.fieldAPIName).Msg("✅ Pack-owned field removal completed")

	case "repair":
		if err := controlplane.RunRepairWithLockAndRunnerOptions(ctx, db, runnerOptions); err != nil {
			log.Fatal().Err(err).Msg("repair failed")
		}

	case "ensure-indexes":
		if err := controlplane.EnsureIndexesWithLock(ctx, db); err != nil {
			log.Fatal().Err(err).Msg("ensure-indexes failed")
		}

	case "rebuild":
		if err := controlplane.RebuildDerivedWithLockAndRunnerOptions(ctx, db, runnerOptions); err != nil {
			log.Fatal().Err(err).Msg("rebuild failed")
		}

	case "is-bootstrapped":
		// Read-only operator diagnostic: exit 0 only when the full ApplyAll
		// completion marker, release schema, and active app-pack composition pass
		// their checks. It is intentionally not the auto-start authority:
		// completed markers can predate later physical migrations, so
		// infra/entrypoint.sh always uses apply-if-needed's lock-held classifier.
		ok, err := isBootstrappedForEntrypoint(ctx, db, productHooks)
		if err != nil {
			log.Warn().Err(err).Msg("is-bootstrapped: probe failed; treating as not bootstrapped")
			os.Exit(1)
		}
		if !ok {
			log.Info().Msg("is-bootstrapped: full control-plane apply not completed")
			os.Exit(1)
		}
		log.Info().Msg("is-bootstrapped: full control-plane apply and product composition verified")
		os.Exit(0)

	case "is-legacy-unmigrated":
		// Read-only diagnostic probe: exit 0 when the database was prepared under
		// a PRIOR release that predates the
		// control_plane_apply_all marker and has not been migrated to this release
		// (no marker row, but a completed schema bootstrap). Such a database must
		// be migrated by a one-time explicit apply rather than auto-applied (would
		// recouple restart to a full apply, issue #1590) or auto-started start-only
		// (the current release's apply may include repairs it has never run, and
		// pod-readiness verify is not a complete substitute). The entrypoint no
		// longer relies on this probe for safety; apply-if-needed reclassifies the
		// state under the apply lock. Greenfield and any current-release apply
		// state (completed/running/failed) return non-zero.
		ok, err := bootstate.IsLegacyUnmigrated(db.DB())
		if err != nil {
			log.Warn().Err(err).Msg("is-legacy-unmigrated: probe failed; treating as not legacy")
			os.Exit(1)
		}
		if !ok {
			os.Exit(1)
		}
		log.Info().Msg("is-legacy-unmigrated: database prepared under a prior release; one-time migration apply required")
		os.Exit(0)

	case "verify":
		// Operator verify — runs the full gate set (includes
		// autonumber V5/V7 live-data scans). Assumes writers are
		// quiesced per the cmd/setup apply prerequisite; not safe on
		// a live cluster under rolling deploy load. Pod readiness
		// uses controlplane.Verify (shallow) instead.
		//
		// --reconcile-flow-orphans: when set, repairs broken
		// FlowDefinition.ActiveVersionId refs (sets to NULL) before
		// running the full verify gate.  Without this flag, verify is
		// strict and aborts on any orphan (current default behavior).
		verifyFS := flag.NewFlagSet("verify", flag.ContinueOnError)
		verifyFS.SetOutput(os.Stderr)
		reconcileOrphans := verifyFS.Bool("reconcile-flow-orphans", false,
			"repair broken FlowDefinition.ActiveVersionId refs before verify")
		if err := verifyFS.Parse(os.Args[2:]); err != nil {
			fmt.Fprintln(os.Stderr, err)
			os.Exit(2)
		}
		if *reconcileOrphans {
			touched, err := controlplane.ReconcileFlowOrphans(ctx, db)
			if err != nil {
				log.Fatal().Err(err).Msg("reconcile-flow-orphans failed")
			}
			log.Info().Int("touched", touched).Msg("reconciled orphan ActiveVersionId rows")
		}
		if err := persistence.VerifySessionBearerSchemaReadiness(ctx, db.DB()); err != nil {
			log.Fatal().Err(err).Msg("session bearer schema readiness verification failed")
		}
		if err := persistence.VerifyOAuthGrantSchemaReadiness(ctx, db.DB()); err != nil {
			log.Fatal().Err(err).Msg("OAuth authorization-grant schema readiness verification failed")
		}
		if err := controlplane.VerifyDeepWithRunnerOptions(ctx, db, runnerOptions); err != nil {
			log.Fatal().Err(err).Msg("verify failed")
		}
		if err := verifyProductDistributionComposition(ctx, db, productHooks); err != nil {
			log.Fatal().Err(err).Msg("product distribution composition verification failed")
		}
		if err := verifyInstalledPackMetadataProgramsWithComponentKindRegistrationSet(ctx, db, registrationSet); err != nil {
			log.Fatal().Err(err).Msg("metadata program verification failed")
		}
		log.Info().Msg("✅ All readiness checks passed")

	case "verify-l0":
		if err := controlplane.VerifyL0(ctx, db); err != nil {
			log.Fatal().Err(err).Msg("verify-l0 failed")
		}
		log.Info().Msg("✅ Clean L0 verification passed")

	case "verify-profile":
		// Narrow operator-facing verify for the profile convergence line.
		// This intentionally surfaces only schema readiness + profile
		// permission coverage, so unrelated global verify noise does not
		// obscure the profile-specific signal.
		if err := controlplane.VerifyProfile(ctx, db); err != nil {
			log.Fatal().Err(err).Msg("verify-profile failed")
		}
		log.Info().Msg("✅ Profile convergence checks passed")

	case "verify-pack":
		name, err := parseVerifyPackArgs(os.Args[2:], os.Stderr)
		if err != nil {
			fmt.Fprintln(os.Stderr, err)
			os.Exit(2)
		}
		if err := controlplane.VerifyInstalledPackByNameWithRunnerOptions(ctx, db.DB(), name, true, runnerOptions); err != nil {
			log.Fatal().Err(err).Msg("verify-pack failed")
		}
		if err := verifyInstalledPackMetadataProgramsWithComponentKindRegistrationSet(ctx, db, registrationSet, name); err != nil {
			log.Fatal().Err(err).Msg("metadata program verification failed")
		}
		log.Info().Str("pack", name).Msg("✅ Installed pack verification passed")

	default:
		fmt.Fprintf(os.Stderr, "unknown command: %s\n", os.Args[1])
		usage()
		os.Exit(1)
	}
}

func registerNoopProductDistribution() (serverruntime.ProductHooks, error) {
	return serverruntime.RegisterProductDistributionCapabilities(serverruntime.NoopProductDistribution())
}

func setupApplyFinalVerify(productHooks serverruntime.ProductHooks, registrationSet packs.ComponentKindRegistrationSet) func(context.Context, *database.TiDBConnection) error {
	return setupApplyFinalVerifyWithMetadataProgramVerifier(productHooks, func(ctx context.Context, db *database.TiDBConnection, ownerPacks ...string) error {
		return verifyInstalledPackMetadataProgramsWithComponentKindRegistrationSet(ctx, db, registrationSet, ownerPacks...)
	})
}

func setupApplyIfNeededFinalVerify(productHooks serverruntime.ProductHooks) func(context.Context, *database.TiDBConnection) error {
	return setupApplyFinalVerifyWithMetadataProgramVerifier(productHooks, verifyInstalledPackPersistedMetadataPrograms)
}

// The offline candidate verifier already validates metadata programs against
// the exact immutable candidate pack tuple before activation. Its final gate
// keeps schema and product-composition checks, but must not reopen installed
// sources through a second metadata-program pass after activation.
func setupOfflineCandidateFinalVerify(productHooks serverruntime.ProductHooks) func(context.Context, *database.TiDBConnection) error {
	return setupApplyFinalVerifyWithMetadataProgramVerifier(productHooks, nil)
}

func setupApplyFinalVerifyWithMetadataProgramVerifier(
	productHooks serverruntime.ProductHooks,
	verifyMetadataPrograms func(context.Context, *database.TiDBConnection, ...string) error,
) func(context.Context, *database.TiDBConnection) error {
	return func(ctx context.Context, db *database.TiDBConnection) error {
		if err := persistence.VerifySessionBearerSchemaReadiness(ctx, db.DB()); err != nil {
			return err
		}
		if err := persistence.VerifyOAuthGrantSchemaReadiness(ctx, db.DB()); err != nil {
			return err
		}
		if err := serverruntime.VerifySessionControlSchemaReadiness(db.DB()); err != nil {
			return err
		}
		if err := verifyProductDistributionComposition(ctx, db, productHooks); err != nil {
			return err
		}
		if verifyMetadataPrograms == nil {
			return nil
		}
		activeAppPacks, err := packs.ActiveInstalledAppPacks(ctx, db.DB())
		if err != nil {
			return fmt.Errorf("resolve active app packs for metadata program check: %w", err)
		}
		activePackNames := activeInstalledAppPackNames(activeAppPacks)
		if len(activePackNames) == 0 {
			return nil
		}
		if err := verifyMetadataPrograms(ctx, db, activePackNames...); err != nil {
			return fmt.Errorf("metadata program check: %w", err)
		}
		return nil
	}
}

func activeInstalledAppPackNames(rows []packs.ActiveInstalledAppPack) []string {
	names := make([]string, 0, len(rows))
	for _, row := range rows {
		name := strings.TrimSpace(row.Name)
		if name != "" {
			names = append(names, name)
		}
	}
	return names
}

func setupPackNameFromSource(rawSource string) (string, error) {
	source, diagnostics := packs.ResolvePackSourceWithOptions(rawSource, packs.DefaultPackSourceOptions())
	if len(diagnostics) > 0 {
		return "", fmt.Errorf("resolve pack source: %s", setupPackDiagnosticsString(diagnostics))
	}
	if source == nil || source.Manifest == nil {
		return "", fmt.Errorf("resolve pack source: manifest is unavailable")
	}
	name := strings.TrimSpace(source.Manifest.Name)
	if name == "" {
		return "", fmt.Errorf("resolve pack source: manifest name is empty")
	}
	return name, nil
}

func setupPackDiagnosticsString(diagnostics []packs.PackDiagnostic) string {
	lines := make([]string, 0, len(diagnostics))
	for _, diagnostic := range diagnostics {
		lines = append(lines, diagnostic.String())
	}
	return strings.Join(lines, "; ")
}

func verifyProductDistributionComposition(ctx context.Context, db *database.TiDBConnection, productHooks serverruntime.ProductHooks) error {
	activeAppPacks, err := packs.ActiveInstalledAppPacks(ctx, db.DB())
	if err != nil {
		return fmt.Errorf("resolve active app packs for product distribution composition: %w", err)
	}
	if err := serverruntime.ValidateProductComposition(ctx, productHooks, db.DB(), activeAppPacks); err != nil {
		return fmt.Errorf("product distribution composition: %w", err)
	}
	return nil
}

func isBootstrappedForEntrypoint(ctx context.Context, db *database.TiDBConnection, productHooks serverruntime.ProductHooks) (bool, error) {
	ok, err := bootstate.IsControlPlaneApplyAllCompleted(db.DB())
	if err != nil || !ok {
		return ok, err
	}
	// The completed marker alone is debt-blind: a deferred (cutover) apply
	// stamps it while every formula view is still the previous release's
	// compiler output, with a durable rebuild-debt receipt outstanding.
	// Serving on the marker alone would fail open on exactly the state the
	// receipt exists to prevent. Refuse the fast start path instead: the
	// The diagnostic therefore reports false; auto-start independently routes
	// through apply-if-needed, whose control-plane entry gate fails closed with
	// the settle instruction (`setup rebuild` inside the deployment's quiesced
	// activation window).
	owed, err := controlplane.FormulaViewRebuildOwed(ctx, db.DB())
	if err != nil {
		return false, err
	}
	if owed {
		log.Warn().Msg("formula view rebuild debt outstanding from a deferred apply; refusing fast start (run `setup rebuild` inside the deployment's quiesced activation window to settle it)")
		return false, nil
	}
	if err := verifyProductDistributionComposition(ctx, db, productHooks); err != nil {
		return false, err
	}
	return true, nil
}

func usage() {
	fmt.Fprintf(os.Stderr, `Usage: setup <command>

Commands:
  recover-connector-remote-identity-batch  Manifest-bound authored-evidence recovery; dry-run by default, explicit guarded --apply
  activate-permission-state  One-way dark-to-enforced CAS for a fully stopped maintenance window
                   Flags: --expected-lineage <32-hex>, --expected-revision <positive-decimal>, --confirm-maintenance-window
  ensure-permission-state  Idempotently create the permission-state singleton during a control-plane operation
  apply            Full control plane: metadata + repair + packs + rebuild + indexes + verify
                   Flags: --skip-polymorphic-backfill, --repair-legacy-unique-duplicates, --defer-formula-view-rebuild, repeatable --source-revision-advance <exact JSON tuple>
  apply-if-needed  Entrypoint auto path: run apply unless the ApplyAll marker, release repairs, and required live schema are complete under the apply lock
                   Flags: --skip-polymorphic-backfill, --repair-legacy-unique-duplicates, repeatable --source-revision-advance <exact JSON tuple> (--defer-formula-view-rebuild is rejected here)
  apply-offline-candidate  Apply one complete Catalog-owned source set while Runtime is stopped and emit its JSON receipt
                   Flags: --operation-id, repeatable --source, repeatable --release-source-receipt <exact JSON>
	apply-source-set  Apply a pack set while Runtime is stopped and emit its JSON result
	                 Flags: --operation-id, repeatable --source, repeatable --release-source <exact JSON>, repeatable --official-source-revision <exact JSON name/revision>
  inspect-routing-identity-mapping  Read-only stopped-phase check of a canonical Profile/Role mapping proposal against the active tenant target and live identities
                   Flags: --proposal-file <canonical binary>
  apply-source-convergence  Partial operator path: metadata + repair + packs + rebuild + indexes (no final verify)
                   Flags: --skip-polymorphic-backfill, --repair-legacy-unique-duplicates
  init-l0          Initialize only the application-neutral L0 substrate
  ensure-l0-schema-ddl  Additively converge physical L0/system tables for a stopped L0 release upgrade
  validate-pack    Read-only source/manifest validation for an L1 pack
  install-pack     Install a source-backed L1 pack
  reapply-pack     Reapply an installed source-backed L1 pack with explicit operator overrides
                   Flags: --source, --skip-attribution-preflight, --skip-polymorphic-backfill, --repair-legacy-unique-duplicates, --allow-source-revision-advance, --from-revision, --to-revision, --verify-pack
  remove-pack-owned-field  Explicitly remove one stale pack-owned field after operator confirmation
  verify-pack      Verify one installed source-backed L1 pack by stable name
  apply-metadata   Substrate DDL + managed metadata reconciliation only
  repair           Discovery + backfills + compatibility fixes
  apply-packs      Apply runtime platform packs plus optional dev sample
  apply-tenant     Layer tenant-specific metadata on top of the platform baseline (opt-in; skip on greenfield)
  ensure-indexes   Create standard indexes (~900, can be slow)
  rebuild          Rebuild derived state (formula views, caches)
  is-bootstrapped  Read-only diagnostic: exit 0 if the full ApplyAll marker, release schema, and product composition are verified
  is-legacy-unmigrated  Read-only diagnostic: exit 0 if prepared under a prior release with no ApplyAll marker (needs one-time migration apply), else non-zero
  verify           Read-only readiness check
  verify-l0        Read-only clean-L0 verification
  verify-profile   Narrow profile convergence verify
  publish-artifact Publish a local file into the object store as a release artifact
                   Flags: --kind <cli|plugin>, --file <localpath>, --filename <name>
`)
}

type applySourceConvergenceArgs struct {
	skipPolymorphicBackfill      bool
	repairLegacyUniqueDuplicates bool
}

type offlineCandidateArgs struct {
	operationID        string
	packSources        []string
	releaseRepoSources []controlplane.OfflineCandidateReleaseRepoSource
}

func runApplyOfflineCandidateCommand(
	ctx context.Context,
	db *database.TiDBConnection,
	productHooks serverruntime.ProductHooks,
	registrationSet packs.ComponentKindRegistrationSet,
	args []string,
	stdout io.Writer,
	stderr io.Writer,
	apply offlineCandidateApplier,
) int {
	parsed, err := parseOfflineCandidateArgs(args, stderr)
	if err != nil {
		fmt.Fprintln(stderr, err)
		return 2
	}
	options := controlplane.InstallPackOptions{
		CandidateFinalVerify: func(ctx context.Context, runner *controlplane.Runner, candidates []packs.Pack) error {
			return verifyCandidateMetadataProgramsWithComponentKindRegistrationSet(ctx, runner, registrationSet, candidates)
		},
		ApplyAllFinalVerify: setupOfflineCandidateFinalVerify(productHooks),
	}
	receipt, err := apply(
		ctx,
		db,
		controlplane.OfflineCandidateRequest{
			OperationID:        parsed.operationID,
			UpgradeMode:        controlplane.UpgradeModeOfflineRequired,
			PackSources:        parsed.packSources,
			ReleaseRepoSources: parsed.releaseRepoSources,
		},
		options,
		controlplane.RunnerOptions{ComponentKindRegistrationSet: registrationSet},
	)
	if err != nil {
		fmt.Fprintf(stderr, "apply-offline-candidate: %v\n", err)
		return 1
	}
	if err := json.NewEncoder(stdout).Encode(receipt); err != nil {
		fmt.Fprintf(stderr, "apply-offline-candidate: encode receipt: %v\n", err)
		return 1
	}
	return 0
}

func parseOfflineCandidateArgs(args []string, stderr io.Writer) (offlineCandidateArgs, error) {
	fs := flag.NewFlagSet("apply-offline-candidate", flag.ContinueOnError)
	fs.SetOutput(stderr)
	operationID := fs.String("operation-id", "", "Deployment operation identifier recorded in the candidate receipt")
	var packSources []string
	var releaseRepoSources []controlplane.OfflineCandidateReleaseRepoSource
	fs.Func("source", "Complete desired pack source; repeat once per pack", func(source string) error {
		source = strings.TrimSpace(source)
		if source == "" {
			return fmt.Errorf("source must not be empty")
		}
		packSources = append(packSources, source)
		return nil
	})
	fs.Func("release-source-receipt", "Exact release_repo source authority JSON; repeat once per release source", func(raw string) error {
		decoder := json.NewDecoder(strings.NewReader(raw))
		decoder.DisallowUnknownFields()
		var source controlplane.OfflineCandidateReleaseRepoSource
		if err := decoder.Decode(&source); err != nil {
			return fmt.Errorf("decode release source receipt: %w", err)
		}
		if decoder.More() {
			return fmt.Errorf("release source receipt must contain exactly one JSON object")
		}
		var trailing any
		if err := decoder.Decode(&trailing); err != io.EOF {
			return fmt.Errorf("release source receipt must contain exactly one JSON object")
		}
		releaseRepoSources = append(releaseRepoSources, source)
		return nil
	})
	if err := fs.Parse(args); err != nil {
		return offlineCandidateArgs{}, err
	}
	operation := strings.TrimSpace(*operationID)
	if operation == "" {
		return offlineCandidateArgs{}, fmt.Errorf("apply-offline-candidate: --operation-id is required")
	}
	if len(packSources) == 0 {
		return offlineCandidateArgs{}, fmt.Errorf("apply-offline-candidate: at least one --source is required")
	}
	if fs.NArg() != 0 {
		return offlineCandidateArgs{}, fmt.Errorf("apply-offline-candidate: unexpected arguments: %s", strings.Join(fs.Args(), " "))
	}
	return offlineCandidateArgs{
		operationID: operation, packSources: packSources, releaseRepoSources: releaseRepoSources,
	}, nil
}

type sourceSetArgs struct {
	operationID                 string
	packSources                 []string
	releaseRepoSources          []controlplane.StoppedSourceSetReleaseRepoSource
	officialRepoSourceRevisions map[string]string
}

type officialSourceRevision struct {
	Name     string `json:"name"`
	Revision string `json:"revision"`
}

func runApplySourceSetCommand(
	ctx context.Context,
	db *database.TiDBConnection,
	productHooks serverruntime.ProductHooks,
	args []string,
	stdout io.Writer,
	stderr io.Writer,
	applySourceSet stoppedSourceSetApplier,
) int {
	parsed, err := parseSourceSetArgs(args, stderr)
	if err != nil {
		fmt.Fprintln(stderr, err)
		return 2
	}
	// The request is the operation-scoped source authority. Controlplane resolves
	// each official source from the configured L1 repository at this exact
	// revision and validates its named manifest and component contract. Compiled
	// product locks describe packs supplied by that distribution; they are not a
	// whitelist for independently selected official L1 packs.
	result, err := applySourceSet(
		ctx,
		db,
		controlplane.StoppedSourceSetRequest{
			OperationID:                 parsed.operationID,
			PackSources:                 parsed.packSources,
			ReleaseRepoSources:          parsed.releaseRepoSources,
			OfficialRepoSourceRevisions: parsed.officialRepoSourceRevisions,
		},
		controlplane.RunnerOptions{ComponentKindRegistrationSet: productHooks.ComponentKindRegistrationSet()},
	)
	if err != nil {
		fmt.Fprintf(stderr, "%s: %v\n", applySourceSetCommand, err)
		return 1
	}
	if err := json.NewEncoder(stdout).Encode(result); err != nil {
		fmt.Fprintf(stderr, "%s: encode result: %v\n", applySourceSetCommand, err)
		return 1
	}
	return 0
}

func parseSourceSetArgs(args []string, stderr io.Writer) (sourceSetArgs, error) {
	fs := flag.NewFlagSet(applySourceSetCommand, flag.ContinueOnError)
	fs.SetOutput(stderr)
	operationID := fs.String("operation-id", "", "Deployment operation identifier recorded in the stopped source-set result")
	var packSources []string
	var releaseRepoSources []controlplane.StoppedSourceSetReleaseRepoSource
	officialRepoSourceRevisions := make(map[string]string)
	fs.Func("source", "Pack source; repeat once per pack", func(source string) error {
		source = strings.TrimSpace(source)
		if source == "" {
			return fmt.Errorf("source must not be empty")
		}
		packSources = append(packSources, source)
		return nil
	})
	fs.Func("release-source", "release_repo source root and revision JSON; repeat once per release source", func(raw string) error {
		var source controlplane.StoppedSourceSetReleaseRepoSource
		if err := decodeExactJSONObject(raw, "release source", &source); err != nil {
			return err
		}
		releaseRepoSources = append(releaseRepoSources, source)
		return nil
	})
	fs.Func("official-source-revision", "Exact official_repo source revision JSON with canonical lowercase name and revision; repeat once per official source", func(raw string) error {
		source, err := parseOfficialSourceRevision(raw)
		if err != nil {
			return err
		}
		key := source.Name
		if _, duplicate := officialRepoSourceRevisions[key]; duplicate {
			return fmt.Errorf("official source revision for %q is duplicated", source.Name)
		}
		officialRepoSourceRevisions[key] = source.Revision
		return nil
	})
	if err := fs.Parse(args); err != nil {
		return sourceSetArgs{}, err
	}
	operation := strings.TrimSpace(*operationID)
	if operation == "" {
		return sourceSetArgs{}, fmt.Errorf("%s: --operation-id is required", applySourceSetCommand)
	}
	if len(packSources) == 0 {
		return sourceSetArgs{}, fmt.Errorf("%s: at least one --source is required", applySourceSetCommand)
	}
	if fs.NArg() != 0 {
		return sourceSetArgs{}, fmt.Errorf("%s: unexpected arguments: %s", applySourceSetCommand, strings.Join(fs.Args(), " "))
	}
	return sourceSetArgs{
		operationID:                 operation,
		packSources:                 packSources,
		releaseRepoSources:          releaseRepoSources,
		officialRepoSourceRevisions: officialRepoSourceRevisions,
	}, nil
}

func decodeExactJSONObject(raw, label string, destination any) error {
	if err := rejectDuplicateJSONKeys([]byte(raw)); err != nil {
		return fmt.Errorf("decode %s: %w", label, err)
	}
	decoder := json.NewDecoder(strings.NewReader(raw))
	decoder.DisallowUnknownFields()
	if err := decoder.Decode(destination); err != nil {
		return fmt.Errorf("decode %s: %w", label, err)
	}
	var trailing any
	if err := decoder.Decode(&trailing); err != io.EOF {
		return fmt.Errorf("%s must contain exactly one JSON object", label)
	}
	return nil
}

func parseOfficialSourceRevision(raw string) (officialSourceRevision, error) {
	var source officialSourceRevision
	if err := decodeExactJSONObject(raw, "official source revision", &source); err != nil {
		return officialSourceRevision{}, err
	}
	source.Name = strings.TrimSpace(source.Name)
	source.Revision = strings.ToLower(strings.TrimSpace(source.Revision))
	if source.Name == "" || source.Revision == "" {
		return officialSourceRevision{}, fmt.Errorf("official source revision requires non-empty name and revision")
	}
	if source.Name != strings.ToLower(source.Name) {
		return officialSourceRevision{}, fmt.Errorf("official source revision name %q must use canonical lowercase", source.Name)
	}
	if len(source.Revision) != 40 {
		return officialSourceRevision{}, fmt.Errorf("official source revision for %q must be an immutable 40-hex revision", source.Name)
	}
	if _, err := hex.DecodeString(source.Revision); err != nil {
		return officialSourceRevision{}, fmt.Errorf("official source revision for %q must be an immutable 40-hex revision", source.Name)
	}
	return source, nil
}

type applyArgs struct {
	skipPolymorphicBackfill      bool
	repairLegacyUniqueDuplicates bool
	deferFormulaViewRebuild      bool
	sourceRevisionAdvancePlan    []controlplane.SourceRevisionAdvanceAuthorization
}

type sourceRevisionAdvanceFlag struct {
	Name         string `json:"name"`
	Kind         string `json:"kind"`
	URI          string `json:"uri"`
	FromRevision string `json:"from_revision"`
	ToRevision   string `json:"to_revision"`
}

func parseApplyArgs(args []string, stderr io.Writer) (applyArgs, error) {
	fs := flag.NewFlagSet("apply", flag.ContinueOnError)
	fs.SetOutput(stderr)
	skipPolymorphicBackfill := fs.Bool("skip-polymorphic-backfill", false, "Skip polymorphic discriminator backfill during full control-plane apply for this operator run only")
	repairLegacyUniqueDuplicates := fs.Bool("repair-legacy-unique-duplicates", false, "Clear non-canonical duplicate values from nullable unique fields before unique-index enforcement")
	deferFormulaViewRebuild := fs.Bool("defer-formula-view-rebuild", false, "Cutover deploys only: neither create nor replace any formula view during this apply; the durable rebuild debt must be settled by `setup rebuild` inside the deployment's quiesced activation window before any serving starts (apply only; rejected by apply-if-needed)")
	var sourceRevisionAdvancePlan []controlplane.SourceRevisionAdvanceAuthorization
	fs.Func("source-revision-advance", "Exact JSON tuple with name, kind, uri, from_revision, and to_revision; repeat once per distribution-authorized edge; capability "+applyAllSourceRevisionPlanCapability, func(raw string) error {
		advance, err := parseSourceRevisionAdvanceFlag(raw)
		if err != nil {
			return err
		}
		sourceRevisionAdvancePlan = append(sourceRevisionAdvancePlan, advance)
		return nil
	})
	if err := fs.Parse(args); err != nil {
		return applyArgs{}, err
	}
	if fs.NArg() != 0 {
		return applyArgs{}, fmt.Errorf("apply: unexpected arguments: %s", strings.Join(fs.Args(), " "))
	}
	return applyArgs{
		skipPolymorphicBackfill:      *skipPolymorphicBackfill,
		repairLegacyUniqueDuplicates: *repairLegacyUniqueDuplicates,
		deferFormulaViewRebuild:      *deferFormulaViewRebuild,
		sourceRevisionAdvancePlan:    sourceRevisionAdvancePlan,
	}, nil
}

func parseSourceRevisionAdvanceFlag(raw string) (controlplane.SourceRevisionAdvanceAuthorization, error) {
	decoder := json.NewDecoder(strings.NewReader(raw))
	open, err := decoder.Token()
	if err != nil {
		return controlplane.SourceRevisionAdvanceAuthorization{}, fmt.Errorf("source-revision advance must be one exact JSON tuple: %w", err)
	}
	if delimiter, ok := open.(json.Delim); !ok || delimiter != '{' {
		return controlplane.SourceRevisionAdvanceAuthorization{}, fmt.Errorf("source-revision advance must be one exact JSON object")
	}
	values := make(map[string]string, 5)
	for decoder.More() {
		rawKey, err := decoder.Token()
		if err != nil {
			return controlplane.SourceRevisionAdvanceAuthorization{}, fmt.Errorf("source-revision advance key: %w", err)
		}
		key, ok := rawKey.(string)
		if !ok {
			return controlplane.SourceRevisionAdvanceAuthorization{}, fmt.Errorf("source-revision advance contains a non-string key")
		}
		switch key {
		case "name", "kind", "uri", "from_revision", "to_revision":
		default:
			return controlplane.SourceRevisionAdvanceAuthorization{}, fmt.Errorf("source-revision advance contains unknown field %q", key)
		}
		if _, duplicate := values[key]; duplicate {
			return controlplane.SourceRevisionAdvanceAuthorization{}, fmt.Errorf("source-revision advance contains duplicate field %q", key)
		}
		var value string
		if err := decoder.Decode(&value); err != nil {
			return controlplane.SourceRevisionAdvanceAuthorization{}, fmt.Errorf("source-revision advance field %q must be a string: %w", key, err)
		}
		values[key] = strings.TrimSpace(value)
	}
	closeToken, err := decoder.Token()
	if err != nil {
		return controlplane.SourceRevisionAdvanceAuthorization{}, fmt.Errorf("source-revision advance closing object: %w", err)
	}
	if delimiter, ok := closeToken.(json.Delim); !ok || delimiter != '}' {
		return controlplane.SourceRevisionAdvanceAuthorization{}, fmt.Errorf("source-revision advance must end after one exact JSON object")
	}
	if err := decoder.Decode(&struct{}{}); err != io.EOF {
		if err == nil {
			err = fmt.Errorf("multiple JSON values")
		}
		return controlplane.SourceRevisionAdvanceAuthorization{}, fmt.Errorf("source-revision advance must be one exact JSON tuple: %w", err)
	}
	value := sourceRevisionAdvanceFlag{
		Name:         values["name"],
		Kind:         values["kind"],
		URI:          values["uri"],
		FromRevision: values["from_revision"],
		ToRevision:   values["to_revision"],
	}
	if value.Name == "" || value.Kind == "" || value.URI == "" ||
		value.FromRevision == "" || value.ToRevision == "" {
		return controlplane.SourceRevisionAdvanceAuthorization{}, fmt.Errorf("source-revision advance requires non-empty name, kind, uri, from_revision, and to_revision")
	}
	if value.FromRevision == value.ToRevision {
		return controlplane.SourceRevisionAdvanceAuthorization{}, fmt.Errorf("source-revision advance requires distinct from_revision and to_revision")
	}
	return controlplane.SourceRevisionAdvanceAuthorization{
		Name:         value.Name,
		Kind:         value.Kind,
		URI:          value.URI,
		FromRevision: value.FromRevision,
		ToRevision:   value.ToRevision,
	}, nil
}

func applyInstallPackOptions(args applyArgs) controlplane.InstallPackOptions {
	return controlplane.InstallPackOptions{
		SkipPolymorphicDiscriminatorBackfill: args.skipPolymorphicBackfill,
		RepairLegacyUniqueDuplicates:         args.repairLegacyUniqueDuplicates,
		DeferFormulaViewRebuild:              args.deferFormulaViewRebuild,
		ApplyAllSourceRevisionAdvances:       append([]controlplane.SourceRevisionAdvanceAuthorization(nil), args.sourceRevisionAdvancePlan...),
	}
}

func preflightApplyAllSourceRevisionAdvances(
	ctx context.Context,
	db *database.TiDBConnection,
	options controlplane.InstallPackOptions,
) error {
	if len(options.ApplyAllSourceRevisionAdvances) == 0 {
		return nil
	}
	_, err := controlplane.PendingApplyAllSourceRevisionAdvances(ctx, db.DB(), options)
	return err
}

func parseApplySourceConvergenceArgs(args []string, stderr io.Writer) (applySourceConvergenceArgs, error) {
	fs := flag.NewFlagSet("apply-source-convergence", flag.ContinueOnError)
	fs.SetOutput(stderr)
	skipPolymorphicBackfill := fs.Bool("skip-polymorphic-backfill", false, "Skip polymorphic discriminator backfill during this partial source-convergence run only; command exits non-zero afterward because full repair coverage was intentionally skipped")
	repairLegacyUniqueDuplicates := fs.Bool("repair-legacy-unique-duplicates", false, "Clear non-canonical duplicate values from nullable unique fields before unique-index enforcement")
	if err := fs.Parse(args); err != nil {
		return applySourceConvergenceArgs{}, err
	}
	if fs.NArg() != 0 {
		return applySourceConvergenceArgs{}, fmt.Errorf("apply-source-convergence: unexpected arguments: %s", strings.Join(fs.Args(), " "))
	}
	return applySourceConvergenceArgs{
		skipPolymorphicBackfill:      *skipPolymorphicBackfill,
		repairLegacyUniqueDuplicates: *repairLegacyUniqueDuplicates,
	}, nil
}

func parseInstallPackArgs(args []string, stderr io.Writer) (string, error) {
	fs := flag.NewFlagSet("install-pack", flag.ContinueOnError)
	fs.SetOutput(stderr)
	source := fs.String("source", "", "Pack source, e.g. official_repo:example or local_path:/abs/path/to/example")
	if err := fs.Parse(args); err != nil {
		return "", err
	}
	if *source == "" {
		return "", fmt.Errorf("install-pack: --source is required")
	}
	if fs.NArg() != 0 {
		return "", fmt.Errorf("install-pack: unexpected arguments: %s", strings.Join(fs.Args(), " "))
	}
	return *source, nil
}

type reapplyPackArgs struct {
	source                       string
	skipAttributionPreflight     bool
	skipPolymorphicBackfill      bool
	repairLegacyUniqueDuplicates bool
	allowSourceRevisionAdvance   bool
	fromRevision                 string
	toRevision                   string
	verifyPack                   bool
}

func parseReapplyPackArgs(args []string, stderr io.Writer) (reapplyPackArgs, error) {
	fs := flag.NewFlagSet("reapply-pack", flag.ContinueOnError)
	fs.SetOutput(stderr)
	source := fs.String("source", "", "Pack source, e.g. official_repo:example or local_path:/abs/path/to/example")
	skipAttributionPreflight := fs.Bool("skip-attribution-preflight", false, "Skip install-pack attribution preflight for this operator reapply only")
	skipPolymorphicBackfill := fs.Bool("skip-polymorphic-backfill", false, "Skip polymorphic discriminator backfill during object-spec apply for this operator reapply only")
	repairLegacyUniqueDuplicates := fs.Bool("repair-legacy-unique-duplicates", false, "Clear non-canonical duplicate values from nullable unique fields before unique-index enforcement")
	allowSourceRevisionAdvance := fs.Bool("allow-source-revision-advance", false, "Allow this reapply to advance the installed source revision when --from-revision and --to-revision match exactly")
	fromRevision := fs.String("from-revision", "", "Expected installed SourceRevision when --allow-source-revision-advance is set")
	toRevision := fs.String("to-revision", "", "Expected resolved SourceRevision when --allow-source-revision-advance is set")
	verifyPack := fs.Bool("verify-pack", false, "Run verify-pack automatically after reapply succeeds")
	if err := fs.Parse(args); err != nil {
		return reapplyPackArgs{}, err
	}
	if *source == "" {
		return reapplyPackArgs{}, fmt.Errorf("reapply-pack: --source is required")
	}
	from := strings.TrimSpace(*fromRevision)
	to := strings.TrimSpace(*toRevision)
	if *allowSourceRevisionAdvance {
		if from == "" {
			return reapplyPackArgs{}, fmt.Errorf("reapply-pack: --from-revision is required with --allow-source-revision-advance")
		}
		if to == "" {
			return reapplyPackArgs{}, fmt.Errorf("reapply-pack: --to-revision is required with --allow-source-revision-advance")
		}
		if from == to {
			return reapplyPackArgs{}, fmt.Errorf("reapply-pack: --from-revision and --to-revision must differ")
		}
		if !*verifyPack {
			return reapplyPackArgs{}, fmt.Errorf("reapply-pack: --verify-pack is required with --allow-source-revision-advance")
		}
	} else if from != "" || to != "" {
		return reapplyPackArgs{}, fmt.Errorf("reapply-pack: --from-revision/--to-revision require --allow-source-revision-advance")
	}
	if fs.NArg() != 0 {
		return reapplyPackArgs{}, fmt.Errorf("reapply-pack: unexpected arguments: %s", strings.Join(fs.Args(), " "))
	}
	return reapplyPackArgs{
		source:                       *source,
		skipAttributionPreflight:     *skipAttributionPreflight,
		skipPolymorphicBackfill:      *skipPolymorphicBackfill,
		repairLegacyUniqueDuplicates: *repairLegacyUniqueDuplicates,
		allowSourceRevisionAdvance:   *allowSourceRevisionAdvance,
		fromRevision:                 from,
		toRevision:                   to,
		verifyPack:                   *verifyPack,
	}, nil
}

func parseVerifyPackArgs(args []string, stderr io.Writer) (string, error) {
	fs := flag.NewFlagSet("verify-pack", flag.ContinueOnError)
	fs.SetOutput(stderr)
	if err := fs.Parse(args); err != nil {
		return "", err
	}
	if fs.NArg() != 1 {
		return "", fmt.Errorf("verify-pack: expected exactly one pack name")
	}
	name := strings.TrimSpace(fs.Arg(0))
	if name == "" {
		return "", fmt.Errorf("verify-pack: pack name is required")
	}
	return name, nil
}

type removePackOwnedFieldArgs struct {
	pack          string
	objectAPIName string
	fieldAPIName  string
}

func parseRemovePackOwnedFieldArgs(args []string, stderr io.Writer) (removePackOwnedFieldArgs, error) {
	fs := flag.NewFlagSet("remove-pack-owned-field", flag.ContinueOnError)
	fs.SetOutput(stderr)
	pack := fs.String("pack", "", "Installed pack name")
	objectAPIName := fs.String("object", "", "Object API name")
	fieldAPIName := fs.String("field", "", "Field API name")
	confirm := fs.Bool("confirm-destructive-field-removal", false, "Required confirmation for physical field/column removal")
	if err := fs.Parse(args); err != nil {
		return removePackOwnedFieldArgs{}, err
	}
	if strings.TrimSpace(*pack) == "" {
		return removePackOwnedFieldArgs{}, fmt.Errorf("remove-pack-owned-field: --pack is required")
	}
	if strings.TrimSpace(*objectAPIName) == "" {
		return removePackOwnedFieldArgs{}, fmt.Errorf("remove-pack-owned-field: --object is required")
	}
	if strings.TrimSpace(*fieldAPIName) == "" {
		return removePackOwnedFieldArgs{}, fmt.Errorf("remove-pack-owned-field: --field is required")
	}
	if !*confirm {
		return removePackOwnedFieldArgs{}, fmt.Errorf("remove-pack-owned-field: --confirm-destructive-field-removal is required")
	}
	if fs.NArg() != 0 {
		return removePackOwnedFieldArgs{}, fmt.Errorf("remove-pack-owned-field: unexpected arguments: %s", strings.Join(fs.Args(), " "))
	}
	return removePackOwnedFieldArgs{
		pack:          strings.TrimSpace(*pack),
		objectAPIName: strings.TrimSpace(*objectAPIName),
		fieldAPIName:  strings.TrimSpace(*fieldAPIName),
	}, nil
}

func runValidatePackWithComponentKindRegistrationSet(args []string, stdout, stderr io.Writer, registrationSet packs.ComponentKindRegistrationSet) int {
	fs := flag.NewFlagSet("validate-pack", flag.ContinueOnError)
	fs.SetOutput(stderr)
	source := fs.String("source", "", "Pack source, e.g. official_repo:example or local_path:/abs/path/to/example")
	if err := fs.Parse(args); err != nil {
		return 2
	}
	if *source == "" {
		fmt.Fprintln(stderr, "validate-pack: --source is required")
		return 2
	}
	if fs.NArg() != 0 {
		fmt.Fprintf(stderr, "validate-pack: unexpected arguments: %s\n", strings.Join(fs.Args(), " "))
		return 2
	}

	report := packs.ValidateInstallablePackSourceWithComponentKindRegistrationSet(*source, packs.DefaultPackSourceOptions(), registrationSet)
	if report.HasErrors() {
		for _, diagnostic := range report.Diagnostics {
			fmt.Fprintf(stderr, "ERROR %s\n", diagnostic.String())
		}
		return 1
	}
	if report.Source == nil || report.Source.Manifest == nil {
		fmt.Fprintln(stderr, "validate-pack: source did not produce a manifest")
		return 1
	}
	fmt.Fprintf(stdout, "pack source valid: %s %s (%d components)\n",
		report.Source.Manifest.Name,
		report.Source.Manifest.Version,
		len(report.Source.Manifest.Components),
	)
	return 0
}

// runPublishArtifact publishes a local file into the object store at the fixed
// release-artifact key for the given kind, using the same storage abstraction
// (LoadConfigFromEnv + NewStore) as the runtime download surface. It is a
// deploy-pipeline step and needs no database connection.
func runPublishArtifact(args []string, stdout, stderr io.Writer) int {
	fs := flag.NewFlagSet("publish-artifact", flag.ContinueOnError)
	fs.SetOutput(stderr)
	kind := fs.String("kind", "", "Artifact kind: cli or plugin")
	file := fs.String("file", "", "Path to the local artifact file to publish")
	filename := fs.String("filename", "", "Optional friendly filename recorded on the object (informational)")
	if err := fs.Parse(args); err != nil {
		return 2
	}
	if *kind == "" {
		fmt.Fprintln(stderr, "publish-artifact: --kind is required")
		return 2
	}
	if *file == "" {
		fmt.Fprintln(stderr, "publish-artifact: --file is required")
		return 2
	}
	if fs.NArg() != 0 {
		fmt.Fprintf(stderr, "publish-artifact: unexpected arguments: %s\n", strings.Join(fs.Args(), " "))
		return 2
	}

	catalog, err := downloads.LoadCatalogFromEnv()
	if err != nil {
		fmt.Fprintf(stderr, "publish-artifact: invalid %s: %v\n", downloads.DownloadArtifactsEnv, err)
		return 2
	}
	art, ok := catalog[*kind]
	if !ok {
		fmt.Fprintf(stderr, "publish-artifact: unknown kind %q; configure it in %s\n", *kind, downloads.DownloadArtifactsEnv)
		return 2
	}
	key := art.Key
	friendly := strings.TrimSpace(*filename)
	if friendly == "" {
		friendly = art.Filename
	}

	f, err := os.Open(*file)
	if err != nil {
		fmt.Fprintf(stderr, "publish-artifact: open file: %v\n", err)
		return 1
	}
	defer func() { _ = f.Close() }()

	info, err := f.Stat()
	if err != nil {
		fmt.Fprintf(stderr, "publish-artifact: stat file: %v\n", err)
		return 1
	}
	size := info.Size()

	store, err := downloads.NewArtifactStoreFromEnv()
	if err != nil {
		fmt.Fprintf(stderr, "publish-artifact: init artifact store: %v\n", err)
		return 1
	}

	meta, err := store.PutObject(context.Background(), key, f, storage.PutObjectOptions{
		ContentType:   "application/octet-stream",
		ContentLength: &size,
	})
	if err != nil {
		fmt.Fprintf(stderr, "publish-artifact: put object: %v\n", err)
		return 1
	}

	fmt.Fprintf(stdout, "published %s artifact (%s) to %s (%d bytes)\n", *kind, friendly, meta.Key, meta.Size)
	return 0
}

func tenantOverlayApplyRequested() bool {
	value := strings.TrimSpace(os.Getenv("NEXUSCRM_APPLY_TENANT_OVERLAY"))
	return value == "1" || strings.EqualFold(value, "true")
}

func runTenantOverlayApply(ctx context.Context, runnerOptions controlplane.RunnerOptions) error {
	tenantDB, err := database.NewConnectionContext(ctx, database.DefaultPoolOptions())
	if err != nil {
		return fmt.Errorf("create tenant overlay apply database connection: %w", err)
	}
	defer tenantDB.Close()
	return runTenantOverlayApplyWithDB(ctx, tenantDB, runnerOptions)
}

func runTenantOverlayApplyWithDB(ctx context.Context, db *database.TiDBConnection, runnerOptions controlplane.RunnerOptions) error {
	r, err := controlplane.NewRunnerWithOptions(db, runnerOptions)
	if err != nil {
		return fmt.Errorf("create control-plane runner: %w", err)
	}
	if err := r.RefreshCache(ctx); err != nil {
		return fmt.Errorf("refresh cache: %w", err)
	}
	if err := controlplane.ApplyTenant(ctx, r); err != nil {
		return err
	}
	return nil
}

func isProduction() bool {
	env := os.Getenv("NEXUS_ENV")
	return env == "production" || env == "prod" || env == "staging"
}