// Command drive9-server starts the drive9 HTTP server and exposes operational
// schema tooling.
//
// Usage:
//
//	drive9-server [listen-addr]
//	drive9-server schema dump-init-sql [flags]
package main

import (
	"context"
	"crypto/sha256"
	"encoding/binary"
	"encoding/hex"
	"errors"
	"fmt"
	"io"
	"math"
	"net"
	"os"
	"os/signal"
	"strconv"
	"strings"
	"syscall"
	"time"

	"go.uber.org/zap"

	"github.com/mem9-ai/drive9/pkg/backend"
	"github.com/mem9-ai/drive9/pkg/buildinfo"
	"github.com/mem9-ai/drive9/pkg/embedding"
	"github.com/mem9-ai/drive9/pkg/encrypt"
	"github.com/mem9-ai/drive9/pkg/innerauth"
	"github.com/mem9-ai/drive9/pkg/leader"
	"github.com/mem9-ai/drive9/pkg/logger"
	"github.com/mem9-ai/drive9/pkg/meta"
	"github.com/mem9-ai/drive9/pkg/metrics"
	"github.com/mem9-ai/drive9/pkg/s3client"
	"github.com/mem9-ai/drive9/pkg/server"
	"github.com/mem9-ai/drive9/pkg/slockoauth"
	"github.com/mem9-ai/drive9/pkg/tenant"
	"github.com/mem9-ai/drive9/pkg/tenant/db9"
	"github.com/mem9-ai/drive9/pkg/tenant/local"
	"github.com/mem9-ai/drive9/pkg/tenant/schema"
	"github.com/mem9-ai/drive9/pkg/tenant/starter"
	"github.com/mem9-ai/drive9/pkg/tenant/tidbcloudnative"
	"github.com/mem9-ai/drive9/pkg/tenant/tidbzero"
)

const (
	defaultListenAddr                   = ":9009"
	defaultS3Dir                        = "s3"
	defaultTenantBackendCacheMaxTenants = 1024
	defaultSharedBackendCacheMaxTenants = 20480
)

type s3Config struct {
	Dir              string
	Bucket           string
	Region           string
	Prefix           string
	RoleARN          string
	Endpoint         string
	ForcePathStyle   bool
	AccessKeyID      string
	SecretAccessKey  string
	SessionToken     string
	EncryptionPolicy meta.S3EncryptionPolicy
	ExpressBucket    string
	ExpressPrefix    string
}

func main() {
	if len(os.Args) > 1 {
		switch os.Args[1] {
		case "-h", "--help", "help":
			usage(os.Stdout, 0)
			return
		case "version":
			if len(os.Args) != 2 {
				usage(os.Stderr, 2)
			}
			_, _ = fmt.Fprint(os.Stdout, versionText())
			return
		case "schema":
			die(schema.ConfigureTiDBAutoEmbeddingFromEnv())
			die(runSchemaCommand(os.Args[2:]))
			return
		}
	}
	if len(os.Args) > 2 {
		usage(os.Stderr, 2)
	}
	autoEmbeddingConfig, err := schema.TiDBAutoEmbeddingConfigFromEnv()
	die(err)
	die(schema.ConfigureTiDBAutoEmbedding(autoEmbeddingConfig))

	cliAddr := ""
	if len(os.Args) == 2 {
		cliAddr = os.Args[1]
	}
	localProvider := localProviderFromEnv()
	addr := resolveServerListenAddr(cliAddr, os.Getenv("DRIVE9_LISTEN_ADDR"), localProvider)
	if localProvider {
		applyLocalProviderEnvDefaults(addr)
		die(rejectLocalProviderNonLoopbackDefaults(addr))
	}

	srvLogger, err := logger.NewServerLogger()
	if err != nil {
		die(fmt.Errorf("create logger: %w", err))
	}
	defer func() { _ = srvLogger.Sync() }()
	logger.Set(srvLogger)
	logger.Info(context.Background(), "build_info", buildinfo.Fields("drive9-server")...)

	metaDSN := os.Getenv("DRIVE9_META_DSN")
	if metaDSN == "" {
		die(fmt.Errorf("DRIVE9_META_DSN is required"))
	}

	s3cfg := s3ConfigFromEnv()
	if err := s3cfg.validate(); err != nil {
		die(err)
	}
	backendOptions, imageExtractDefaultProvider, err := buildBackendOptionsFromEnv()
	if err != nil {
		die(err)
	}
	embeddingProcess, err := buildEmbeddingProcessConfigFromEnv(context.Background())
	if err != nil {
		die(err)
	}
	autoEmbeddingAPIKey := strings.TrimSpace(os.Getenv(schema.EnvTiDBAutoEmbeddingAPIKey))
	autoEmbeddingAPIBase := strings.TrimSpace(os.Getenv(schema.EnvTiDBAutoEmbeddingAPIBase))
	tenantWorkerOpts := tenantWorkerOptionsFromEnv()
	var semanticEmbedder embedding.Client
	if embeddingProcess.Runtime != nil {
		semanticEmbedder = embeddingProcess.Runtime.Client
		backendOptions.QueryEmbedding = backend.QueryEmbeddingOptions{Client: embeddingProcess.Runtime.Client}
		backendOptions.AppSemanticTasksEnabled = true
		backendOptions.AppEmbeddingGeneration = embeddingProcess.Runtime.Generation
		backendOptions.AppEmbeddingConfigRevision = embeddingProcess.Runtime.ConfigRevision
	}

	// P1-3: Configurable quota cache refresh interval (default 30s).
	// In multi-pod deployments, increasing this reduces per-tenant-per-pod DB reads.
	backend.InitQuotaConfigCacheRefreshInterval(envInt("DRIVE9_QUOTA_CACHE_REFRESH_SECONDS", 0))
	backend.InitQuotaAdmissionCacheTTLs(
		envDuration("DRIVE9_QUOTA_USAGE_CACHE_TTL", 0),
		envDuration("DRIVE9_QUOTA_PENDING_DELTAS_CACHE_TTL", 0),
	)

	store, err := openControlPlaneStoreWithRetry(context.Background(), metaDSN, defaultStartupRetryOptions())
	if err != nil {
		die(fmt.Errorf("open control-plane store: %w", err))
	}
	if raw := os.Getenv("DRIVE9_DEFAULT_STORAGE_QUOTA_BYTES"); raw != "" {
		if v, err := strconv.ParseInt(raw, 10, 64); err == nil && v > 0 {
			meta.SetDefaultMaxStorageBytes(v)
		}
	}
	applyQuotaDefaultsFromEnv()
	defer func() { _ = store.Close() }()

	// Continuously probe the long-lived metadata ("meta") store so a control-plane
	// DB outage after startup is visible via drive9_db_up and logs. Tenant user
	// and user_schema pools are workload pools; probing them would only keep
	// short-lived connections warm and compete with business traffic.
	probeInterval := time.Duration(envInt("DRIVE9_DB_HEALTH_PROBE_INTERVAL_SECONDS", 15)) * time.Second
	probeTimeout := time.Duration(envInt("DRIVE9_DB_HEALTH_PROBE_TIMEOUT_SECONDS", 3)) * time.Second
	metrics.StartDBHealthProbeWithOptions(context.Background(), probeInterval, probeTimeout, dbHealthProbeOptionsFromEnv(), func(role string, up bool, err error) {
		fields := []zap.Field{zap.String("role", role)}
		if up {
			logger.Info(context.Background(), "db_recovered", fields...)
			return
		}
		fields = append(fields, zap.Error(err))
		logger.Error(context.Background(), "db_unavailable", fields...)
	})

	if s3cfg.Bucket == "" {
		if err := os.MkdirAll(s3cfg.Dir, 0o755); err != nil {
			die(fmt.Errorf("create s3 dir: %w", err))
		}
		logger.Info(context.Background(), "s3_mode_local", zap.String("dir", s3cfg.Dir))
	} else {
		logger.Info(context.Background(), "s3_mode_aws",
			zap.String("bucket", s3cfg.Bucket),
			zap.String("region", s3cfg.Region),
			zap.String("endpoint", s3cfg.Endpoint),
			zap.Bool("path_style", s3cfg.ForcePathStyle),
			zap.String("credentials", s3client.CredentialLogValue(s3cfg.AccessKeyID)),
			zap.String("role", s3client.RoleLogValue(s3cfg.RoleARN)))
	}

	encryptType := envOr("DRIVE9_ENCRYPT_TYPE", "local_aes")
	masterHex := os.Getenv("DRIVE9_MASTER_KEY")
	kmsKey := os.Getenv("DRIVE9_ENCRYPT_KEY")
	aliyunKMSEndpoint := strings.TrimSpace(os.Getenv("DRIVE9_ALIYUN_KMS_ENDPOINT"))
	tokenHex := os.Getenv("DRIVE9_TOKEN_SIGNING_KEY")
	vaultMKHex := os.Getenv("DRIVE9_VAULT_MASTER_KEY")
	providerType := envOr("DRIVE9_TENANT_PROVIDER", tenant.ProviderTiDBZero)
	tenantPoolMaxSize, err := tenantPoolMaxSizeFromEnv()
	if err != nil {
		die(err)
	}
	tenantPoolRefillFreeRatio, err := tenantPoolRefillFreeRatioFromEnv()
	if err != nil {
		die(err)
	}
	sharedDBHardCapRatio, err := sharedDBHardCapRatioFromEnv()
	if err != nil {
		die(err)
	}
	sharedDBReopenRatio, err := sharedDBReopenRatioFromEnv()
	if err != nil {
		die(err)
	}
	sharedDBMaxTenants, err := sharedDBMaxTenantsFromEnv()
	if err != nil {
		die(err)
	}
	sharedDBDefaultSpendingLimit, err := sharedDBDefaultSpendingLimitFromEnv()
	if err != nil {
		die(err)
	}
	tidbCloudNonFreePlanCacheTTL, err := tiDBCloudNonFreePlanCacheTTLFromEnv()
	if err != nil {
		die(err)
	}
	tidbCloudIAMCacheTTL, err := tiDBCloudIAMCacheTTLFromEnv()
	if err != nil {
		die(err)
	}
	maxUploadBytes := server.DefaultMaxUploadBytes
	if raw := os.Getenv("DRIVE9_MAX_UPLOAD_BYTES"); raw != "" {
		maxUploadBytes, err = strconv.ParseInt(raw, 10, 64)
		if err != nil || maxUploadBytes <= 0 {
			die(fmt.Errorf("invalid DRIVE9_MAX_UPLOAD_BYTES: must be a positive integer"))
		}
		if maxUploadBytes < 1<<20 {
			die(fmt.Errorf("DRIVE9_MAX_UPLOAD_BYTES too small: minimum 1048576 (1MiB)"))
		}
	}
	backendOptions.MaxUploadBytes = maxUploadBytes
	meta.SetDefaultMaxFileSizeBytes(maxUploadBytes)
	maxTenantFileCount, err := maxTenantFileCountFromEnv()
	if err != nil {
		die(err)
	}
	meta.SetDefaultMaxFileCount(maxTenantFileCount)
	tidbCloudFreePlanLimits, err := tiDBCloudFreePlanLimitsFromEnv(maxUploadBytes)
	if err != nil {
		die(err)
	}
	providerType, err = tenant.NormalizeProvider(providerType)
	if err != nil {
		die(err)
	}
	apiModes := tiDBCloudAPIModes{iam: apiModeOpenAPI, billing: apiModeOpenAPI}
	var innerIAMClient, innerBillingClient *innerauth.Client
	if tenant.UsesTiDBCloudNativeCredentials(providerType) {
		apiModes, err = tiDBCloudAPIModesFromEnv(providerType)
		if err != nil {
			die(err)
		}
		innerIAMClient, err = innerClientFromEnv(apiModes.iam, tidbcloudnative.EnvTiDBCloudIAMAPIURL)
		if err != nil {
			die(err)
		}
		innerBillingClient, err = innerClientFromEnv(apiModes.billing, tidbcloudnative.EnvTiDBCloudBillingAPIURL)
		if err != nil {
			die(err)
		}
	}
	innerDataClient, err := innerDataClientFromEnv(providerType)
	if err != nil {
		die(err)
	}
	if innerDataClient != nil {
		logger.Info(context.Background(), "tidbcloud_data_plane_validation_enabled",
			zap.String("data_api_url", innerDataClient.BaseURL()))
	}
	disableDatabaseAutoEmbedding := envBool("DRIVE9_DISABLE_AUTO_EMBEDDING", false)
	if tenant.UsesTiDBAutoEmbedding(providerType) && !disableDatabaseAutoEmbedding {
		die(schema.ValidateTiDBAutoEmbeddingProviderConfig(schema.TiDBAutoEmbeddingProviderConfig{
			Model:   autoEmbeddingConfig.Model,
			APIKey:  autoEmbeddingAPIKey,
			APIBase: autoEmbeddingAPIBase,
		}))
	}
	logger.Info(context.Background(), "tenant_provider_selected", zap.String("provider", providerType))

	var provisioner tenant.Provisioner
	var provisionerErr error
	switch providerType {
	case tenant.ProviderTiDBZero:
		provisioner, provisionerErr = tidbzero.NewProvisionerFromEnv()
	case tenant.ProviderTiDBCloudNative, tenant.ProviderTiDBCloudNativeShared:
		provisioner, provisionerErr = tidbcloudnative.NewProvisionerFromEnvWithOptions(providerType, tidbcloudnative.ProvisionerOptions{
			InnerBilling: apiModes.billing == apiModeInner,
		})
	case tenant.ProviderDB9:
		provisioner, provisionerErr = db9.NewProvisionerFromEnv()
	case tenant.ProviderLocal:
		provisioner, provisionerErr = local.NewProvisionerFromEnv()
	}
	var innerDispatcher *innerauth.Dispatcher
	if provisionerErr == nil && innerIAMClient != nil {
		defaults := tenant.ProviderDefaultCredentials{Provisioner: provisioner, Provider: providerType}
		credentials, ok := defaults.DefaultCredentials()
		if !ok || strings.TrimSpace(credentials.PublicKey) == "" || strings.TrimSpace(credentials.PrivateKey) == "" {
			provisionerErr = innerauth.ErrDefaultCredentialsUnavailable
		} else {
			innerDispatcher = innerauth.NewDispatcher(innerIAMClient, defaults, tidbCloudIAMCacheTTL)
		}
	}
	if provisionerErr == nil && providerType == tenant.ProviderTiDBCloudNativeShared {
		validator, ok := provisioner.(interface {
			ValidateSharedAccessWithInnerAPIs(context.Context, innerauth.AccountProvider, innerauth.BillingPlanProvider) error
		})
		if !ok {
			provisionerErr = fmt.Errorf("shared TiDB Cloud access validation is not supported")
		} else {
			sharedIAM, sharedBilling := sharedAccessProviders(innerIAMClient, innerBillingClient, innerDataClient)
			if err := validator.ValidateSharedAccessWithInnerAPIs(
				context.Background(),
				sharedIAM,
				sharedBilling,
			); err != nil {
				provisionerErr = err
			}
		}
	}
	if provisionerErr != nil {
		if tenant.UsesTiDBCloudNativeCredentials(providerType) || providerType == tenant.ProviderLocal {
			logger.Error(context.Background(), "provisioner_failed", zap.String("provider", providerType), zap.Error(provisionerErr))
			os.Exit(1)
		}
		logger.Warn(context.Background(), "provisioner_not_configured", zap.String("provider", providerType), zap.Error(provisionerErr))
	} else {
		logger.Info(context.Background(), "provisioner_configured", zap.String("provider", providerType))
	}

	var legacyStarterProvisioner tenant.Provisioner
	if legacy, err := starter.NewLegacyProvisionerFromEnv(); err != nil {
		if starter.LegacyEnvPresent() {
			logger.Warn(context.Background(), "legacy_starter_provisioner_not_configured", zap.Error(err))
		}
	} else {
		legacyStarterProvisioner = legacy
		logger.Info(context.Background(), "legacy_starter_provisioner_configured")
	}

	var vaultMasterKey []byte
	if vaultMKHex != "" {
		vaultMasterKey, err = hex.DecodeString(vaultMKHex)
		if err != nil {
			die(fmt.Errorf("invalid DRIVE9_VAULT_MASTER_KEY: %w", err))
		}
	}

	var pool *tenant.Pool
	var leaderManager *leader.Manager
	var tokenSecret []byte
	if tokenHex != "" {
		tokenSecret, err = hex.DecodeString(tokenHex)
		if err != nil {
			die(fmt.Errorf("invalid DRIVE9_TOKEN_SIGNING_KEY: %w", err))
		}
		eKey := masterHex
		eType := encrypt.Type(encryptType)
		if strings.EqualFold(string(eType), string(encrypt.TypeKMS)) || strings.EqualFold(string(eType), string(encrypt.TypeAliyunKMS)) || strings.EqualFold(string(eType), string(encrypt.TypeTencentKMS)) {
			eKey = kmsKey
		}
		enc, err := encrypt.New(context.Background(), encrypt.Config{
			Type:              eType,
			Key:               eKey,
			Region:            s3cfg.Region,
			AliyunKMSEndpoint: aliyunKMSEndpoint,
		})
		if err != nil {
			die(fmt.Errorf("create encryptor: %w", err))
		}

		if err := pingControlPlaneDBWithRetry(context.Background(), store, defaultStartupRetryOptions()); err != nil {
			die(fmt.Errorf("control-plane db unavailable: %w", err))
		}

		// Create a leader election manager for gating background schedulers.
		// When disabled (single-pod), IsLeader always returns true and workers
		// start immediately. In multi-pod mode, workers start only on the leader.
		if os.Getenv("DRIVE9_LEADER_DISABLED") == "1" || os.Getenv("DRIVE9_LEADER_DISABLED") == "true" {
			leaderManager = leader.NewManager(store.DB(), leader.WithDisabled())
		} else {
			leaderOpts := []leader.Option{
				leader.WithLockName(envOr("DRIVE9_LEADER_LOCK_NAME", "drive9:leader")),
			}
			if v := envOr("DRIVE9_LEADER_HEARTBEAT_INTERVAL", ""); v != "" {
				if d, err := time.ParseDuration(v); err == nil {
					leaderOpts = append(leaderOpts, leader.WithHeartbeatInterval(d))
				} else {
					logger.Warn(context.Background(), "leader_heartbeat_interval_invalid_falling_back_to_default",
						zap.String("env", "DRIVE9_LEADER_HEARTBEAT_INTERVAL"),
						zap.String("value", v),
						zap.Error(err))
				}
			}
			leaderManager = leader.NewManager(store.DB(), leaderOpts...)
		}
		defer leaderManager.Stop()

		// The extract config resolver is wired below whenever the
		// control-plane store and tenant pool both exist, so every pool
		// backend must advertise the image-extract capability — even when
		// the process-default extractor is disabled — because the resolver
		// supplies the runtime at task execution time.
		backendOptions = withExtractResolverCapabilities(backendOptions)

		pool = tenant.NewPool(tenant.PoolConfig{
			S3Dir:                        s3cfg.Dir,
			PublicURL:                    publicBaseURL(addr),
			S3Bucket:                     s3cfg.Bucket,
			S3Region:                     s3cfg.Region,
			S3Prefix:                     s3cfg.Prefix,
			S3RoleARN:                    s3cfg.RoleARN,
			S3Endpoint:                   s3cfg.Endpoint,
			S3ForcePathStyle:             s3cfg.ForcePathStyle,
			S3AccessKeyID:                s3cfg.AccessKeyID,
			S3SecretAccessKey:            s3cfg.SecretAccessKey,
			S3SessionToken:               s3cfg.SessionToken,
			S3EncryptionPolicy:           s3cfg.EncryptionPolicy,
			S3ExpressBucket:              s3cfg.ExpressBucket,
			S3ExpressPrefix:              s3cfg.ExpressPrefix,
			BackendOptions:               backendOptions,
			MaxTenants:                   tenantBackendCacheMaxTenants(providerType),
			IdleTimeout:                  envDuration("DRIVE9_POOL_IDLE_TTL", 5*time.Minute),
			IdleReapInterval:             envDuration("DRIVE9_POOL_IDLE_REAP_INTERVAL", 2*time.Minute),
			DisableDatabaseAutoEmbedding: disableDatabaseAutoEmbedding,
			LeaderChecker:                leaderManager,
			// Capture once at startup: LocalClustersAPI TiDB is plaintext.
			// Do not re-read DRIVE9_TIDBCLOUD_CLUSTERS_BACKEND on each open.
			SharedDBForcePlaintext: tidbcloudnative.IsLocalClustersBackend(),
		}, enc)
		// The cross-tenant quota mutation dispatcher (started in
		// server.startNotifyInfrastructure) must outlive every backend:
		// backend.Close waits for its queued items to be processed through
		// the dispatcher. Defers run LIFO, so registering this before
		// pool.Close makes it run after the pool (and all backends) closed.
		// On process kill neither defer runs; MutationReplayWorker recovers
		// any items still pending in the durable quota_mutation_log.
		defer backend.StopMutationDispatcher()
		defer pool.Close()

		pool.SetMetaStore(store)
		pool.Start(context.Background())

		// The mutation replay and expiry sweep workers are owned by the server
		// (started/stopped in its leader-gated startLeaderWorkers/stopLeaderWorkers),
		// so they follow leadership transitions alongside the other leader-gated
		// schedulers. main.go no longer registers a competing SetCallbacks pair
		// (that earlier pair was clobbered by the server's own SetCallbacks and
		// never fired).

		// TODO: Run ValidateDurableAsyncExtractRequiresTenantWorker only when this process
		// can serve tenants that enqueue durable audio_extract_text / img_extract_text
		// (database auto-embedding: tidb_zero, tidb_cloud_native). pool != nil is too broad
		// for db9-only pools, which never hit that path but still get forced to configure
		// DRIVE9_EMBED_* when async extract is wired on the template (PR #159 review).
		if err := server.ValidateDurableAsyncExtractRequiresTenantWorker(server.Config{
			Meta:                         store,
			Pool:                         pool,
			Provisioner:                  provisioner,
			LegacyStarterProvisioner:     legacyStarterProvisioner,
			TokenSecret:                  tokenSecret,
			VaultMasterKey:               vaultMasterKey,
			S3Dir:                        s3cfg.Dir,
			MaxUploadBytes:               maxUploadBytes,
			InlineThreshold:              backendOptions.InlineThreshold,
			Logger:                       srvLogger,
			SemanticEmbedder:             semanticEmbedder,
			TenantWorkers:                tenantWorkerOpts,
			TiDBAutoEmbeddingConfig:      autoEmbeddingConfig,
			TiDBAutoEmbeddingAPIKey:      autoEmbeddingAPIKey,
			TiDBAutoEmbeddingAPIBase:     autoEmbeddingAPIBase,
			DisableDatabaseAutoEmbedding: disableDatabaseAutoEmbedding,
		}, backendOptions, false); err != nil {
			die(err)
		}
	}

	slockOAuth, err := slockOAuthFromEnv()
	if err != nil {
		die(err)
	}
	if slockOAuth != nil {
		logger.Info(context.Background(), "slock_oauth_enabled")
	} else {
		logger.Info(context.Background(), "slock_oauth_disabled")
	}

	// Unified tenant outbox notification configuration. The outbox poller reads
	// the central tenant_notify_outbox (always-provisioned meta DB) at 200ms and
	// dispatches by work_mask. See pkg/server/tenant_outbox_poller.go.
	sseNotifyRetention := envDuration("DRIVE9_SSE_NOTIFY_RETENTION", time.Hour)
	// fs_events is the durable SSE replay log. Its retention is deliberately
	// independent of the outbox retention above: production sets 168h so
	// consumer outages within a week remain replayable. The same value feeds
	// both sweep paths (lazy write-path sweep + tenant worker backstop).
	fsEventsRetention := envDuration("DRIVE9_FS_EVENTS_RETENTION", time.Hour)
	// The write-path lazy sweep is throttled per tenant (default 1h); local
	// e2e may shorten it to observe pruning without waiting an hour.
	fsEventsLazySweepInterval := envDuration("DRIVE9_FS_EVENTS_LAZY_SWEEP_INTERVAL", time.Hour)
	tenantWorkerOpts.FSEventsRetention = fsEventsRetention
	podID := strings.TrimSpace(os.Getenv("DRIVE9_POD_ID"))
	podAddr := strings.TrimSpace(os.Getenv("DRIVE9_POD_ADDR"))
	// Default podAddr to the listen addr only if it's a real, non-wildcard
	// address. The pod registry requires a full base URL for peer routing.
	// Wildcard binds like ":9009", "0.0.0.0:9009", or "[::]:9009" are not
	// dialable, so we leave podAddr empty in that case — the podRegistry still
	// reports subscriptions and the leader sweeps stale pods.
	if podAddr == "" && podID != "" {
		if host, port, err := net.SplitHostPort(addr); err == nil && host != "" && host != "0.0.0.0" && host != "::" {
			podAddr = "http://" + net.JoinHostPort(host, port)
		}
	}
	if podID != "" {
		logger.Info(context.Background(), "tenant_outbox_config",
			zap.String("pod_id", podID),
			zap.String("pod_addr", podAddr),
			zap.Duration("retention", sseNotifyRetention),
			zap.Duration("fs_events_retention", fsEventsRetention))
	}

	// Per-tenant delete-orphan reconciliation throttle (piggyback backstop
	// cadence and kick-driven throttle). The package default (10m) applies
	// when the env is unset; only call the setter when it is set so an unset
	// env never logs a misleading fallback.
	if interval, configured := deleteReconcileIntervalFromEnv(); configured {
		server.SetDeleteReconcileInterval(interval)
	}

	srv := server.NewWithConfig(server.Config{
		Meta:                         store,
		Pool:                         pool,
		Provisioner:                  provisioner,
		DefaultTenantProvider:        providerType,
		SharedDBMaxTenants:           sharedDBMaxTenants,
		SharedDBHardCapRatio:         sharedDBHardCapRatio,
		SharedDBReopenRatio:          sharedDBReopenRatio,
		SharedDBSpendingLimit:        sharedDBDefaultSpendingLimit,
		TiDBCloudIAMCacheTTL:         tidbCloudIAMCacheTTL,
		TiDBCloudNonFreePlanCacheTTL: tidbCloudNonFreePlanCacheTTL,
		TiDBCloudFreePlanLimits:      tidbCloudFreePlanLimits,
		InnerAuth:                    innerDispatcher,
		InnerBilling:                 innerBillingPlanProvider(innerBillingClient),
		ManagedSharedDBCloudBatchSize: envInt("DRIVE9_TIDBCLOUD_NATIVE_SHARED_CLOUD_BATCH_SIZE",
			server.DefaultManagedSharedDBCloudBatchSize),
		ManagedSharedDBMetadataWorkers: envInt("DRIVE9_TIDBCLOUD_NATIVE_SHARED_METADATA_WORKERS",
			server.DefaultManagedSharedDBMetadataWorkers),
		ManagedSharedDBMetadataBatchSize: envInt("DRIVE9_TIDBCLOUD_NATIVE_SHARED_METADATA_BATCH_SIZE",
			server.DefaultManagedSharedDBMetadataBatchSize),
		ManagedSharedDBMetadataPollInterval: envDuration("DRIVE9_TIDBCLOUD_NATIVE_SHARED_METADATA_POLL_INTERVAL",
			server.DefaultManagedSharedDBMetadataPollInterval),
		ManagedSharedDBProvisioningWorkers: envInt("DRIVE9_TIDBCLOUD_NATIVE_SHARED_PROVISIONING_WORKERS",
			server.DefaultManagedSharedDBProvisioningWorkers),
		TenantPoolReconcileInterval: envDuration("DRIVE9_TENANT_POOL_RECONCILE_INTERVAL",
			server.DefaultTenantPoolReconcileInterval),
		TenantPoolReconcileWorkers: envInt("DRIVE9_TENANT_POOL_RECONCILE_WORKERS",
			server.DefaultTenantPoolReconcileWorkers),
		TenantPoolReconcileWorkerRest: envDuration("DRIVE9_TENANT_POOL_RECONCILE_WORKER_REST",
			server.DefaultTenantPoolReconcileWorkerRest),
		ManagedSharedDBStuckTimeout: envDuration("DRIVE9_TIDBCLOUD_NATIVE_SHARED_STUCK_TIMEOUT",
			server.DefaultManagedSharedDBStuckTimeout),
		ManagedSharedDBFailedCleanupInterval: envDuration("DRIVE9_TIDBCLOUD_NATIVE_SHARED_FAILED_CLEANUP_INTERVAL",
			server.DefaultManagedSharedDBFailedCleanupInterval),
		ManagedSharedDBFailedCleanupBatchSize: envInt("DRIVE9_TIDBCLOUD_NATIVE_SHARED_FAILED_CLEANUP_BATCH_SIZE",
			server.DefaultManagedSharedDBFailedCleanupBatchSize),
		LegacyStarterProvisioner:        legacyStarterProvisioner,
		TokenSecret:                     tokenSecret,
		VaultMasterKey:                  vaultMasterKey,
		VaultIssuerURL:                  vaultIssuerURL(addr),
		PublicURL:                       publicBaseURL(addr),
		S3Dir:                           s3cfg.Dir,
		MaxUploadBytes:                  maxUploadBytes,
		ObjectGCBatchSize:               objectGCBatchSizeFromEnv(),
		TenantPoolMaxSize:               tenantPoolMaxSize,
		TenantPoolRefillFreeRatio:       tenantPoolRefillFreeRatio,
		InlineThreshold:                 backendOptions.InlineThreshold,
		Logger:                          srvLogger,
		SemanticEmbedder:                semanticEmbedder,
		EmbeddingDefaultRuntime:         embeddingProcess.Runtime,
		EmbeddingDefaultProvider:        embeddingProcess.Provider,
		TenantWorkers:                   tenantWorkerOpts,
		SlockOAuth:                      slockOAuth,
		TiDBAutoEmbeddingConfig:         autoEmbeddingConfig,
		TiDBAutoEmbeddingAPIKey:         autoEmbeddingAPIKey,
		TiDBAutoEmbeddingAPIBase:        autoEmbeddingAPIBase,
		DisableDatabaseAutoEmbedding:    disableDatabaseAutoEmbedding,
		Leader:                          leaderManager,
		TenantOutboxPollInterval:        envDurationCompat("DRIVE9_TENANT_OUTBOX_POLL_INTERVAL", "DRIVE9_TENANT_OUTBOX_POLL_INTERVAL_MS", 200*time.Millisecond),
		TenantOutboxCursorFlushInterval: envDurationCompat("DRIVE9_TENANT_OUTBOX_CURSOR_FLUSH_INTERVAL", "DRIVE9_TENANT_OUTBOX_CURSOR_FLUSH_MS", 5*time.Second),
		TenantShardRefreshInterval:      envDurationCompat("DRIVE9_TENANT_SHARD_REFRESH_INTERVAL", "DRIVE9_TENANT_SHARD_REFRESH_MS", 5*time.Second),
		TenantMaintenanceInterval:       envDurationCompat("DRIVE9_TENANT_MAINTENANCE_INTERVAL", "DRIVE9_TENANT_MAINTENANCE_INTERVAL_MS", 5*time.Minute),
		SafetyNetScanInterval:           envDurationCompat("DRIVE9_SAFETY_NET_SCAN_INTERVAL", "DRIVE9_SAFETY_NET_SCAN_INTERVAL_MS", 5*time.Minute),
		SSENotifyRetention:              sseNotifyRetention,
		FSEventsRetention:               fsEventsRetention,
		FSEventsLazySweepInterval:       fsEventsLazySweepInterval,
		SSELivenessPollInterval:         envDuration("DRIVE9_SSE_LIVENESS_POLL_INTERVAL", 0),
		PodID:                           podID,
		PodAddr:                         podAddr,
	})

	// Per-tenant extract config resolution: when the control-plane store and
	// tenant pool both exist, wire the resolver so img_extract_text tasks
	// execute with the tenant's effective runtime (custom config, off-marker,
	// or process default). provider=local uses this path too.
	wiredExtractResolver, err := configureExtractResolver(srv, store, pool, backendOptions, imageExtractDefaultProvider)
	if err != nil {
		die(err)
	}
	if wiredExtractResolver {
		logger.Info(context.Background(), "extract_config_resolver_wired",
			zap.Bool("process_default_configured", backendOptions.AsyncImageExtract.Enabled))
	}
	if store != nil && pool != nil {
		logger.Info(context.Background(), "embedding_config_resolver_wired",
			zap.Bool("process_default_configured", embeddingProcess.Runtime != nil))
	}
	// Cluster image-extract kill switch (default off). While on, workers stop
	// claiming new img_extract_text tasks; file writes keep enqueueing.
	imageExtractGlobalDisabled := imageExtractGlobalDisabledFromEnv()
	srv.SetImageExtractGlobalDisabled(imageExtractGlobalDisabled)
	if imageExtractGlobalDisabled {
		logger.Info(context.Background(), "image_extract_global_disabled")
	}

	// Graceful shutdown: on SIGINT/SIGTERM drain in-flight requests so
	// ListenAndServe returns and srv.Close() below can run the final notify
	// coalescer flush (and worker teardown) BEFORE the deferred pool.Close,
	// StopMutationDispatcher, and store close registered above tear down the
	// stores the flush still needs.
	stopCtx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()

	// Billing ops flush: every pod periodically persists its in-process file
	// ops counter deltas to the central meta DB for the billing worker. The
	// stopCtx cancels the loop during graceful shutdown; a final flush runs
	// after srv.Close() below, before the deferred store teardown. The
	// tracker gate skips hot-path bookkeeping entirely when flush is off.
	billingFlushInterval := envDuration("DRIVE9_BILLING_OPS_FLUSH_INTERVAL", 0)
	metrics.SetFileOpsFlushEnabled(billingFlushInterval > 0)
	srv.StartBillingOpsFlush(stopCtx, billingFlushInterval)
	shutdownDone := make(chan struct{})
	go func() {
		<-stopCtx.Done()
		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
		defer cancel()
		if err := srv.Shutdown(ctx); err != nil {
			logger.Warn(context.Background(), "server_shutdown_failed", zap.Error(err))
		}
		close(shutdownDone)
	}()

	die(srv.ListenAndServe(addr))
	// ListenAndServe returns as soon as Shutdown closes the listener, which can
	// be BEFORE in-flight requests have drained. When the return was
	// signal-driven, join the shutdown goroutine first so srv.Close() (final
	// notify flush, worker teardown) and the deferred pool/dispatcher/store
	// teardown never run while handlers are still writing.
	if stopCtx.Err() != nil {
		<-shutdownDone
	}
	// Direct call, not a defer: defers run only after main returns, and this
	// must precede the deferred teardown registered above.
	srv.Close()
	// Final billing ops flush: stop the periodic loop, wait for any in-flight
	// flush, then persist remaining increments with a fresh timeout budget.
	// Must run before the deferred store teardown.
	srv.StopBillingOpsFlush()
}

func envOr(key, fallback string) string {
	if value := os.Getenv(key); value != "" {
		return value
	}
	return fallback
}

func localProviderFromEnv() bool {
	return strings.TrimSpace(os.Getenv("DRIVE9_TENANT_PROVIDER")) == tenant.ProviderLocal
}

func slockOAuthFromEnv() (*slockoauth.Client, error) {
	origin := strings.TrimSpace(os.Getenv("DRIVE9_SLOCK_ORIGIN"))
	if origin == "" {
		return nil, nil
	}
	cfg := slockoauth.Config{
		Origin:       origin,
		APIOrigin:    strings.TrimSpace(os.Getenv("DRIVE9_SLOCK_API_ORIGIN")),
		ClientID:     strings.TrimSpace(os.Getenv("DRIVE9_SLOCK_CLIENT_ID")),
		ClientSecret: strings.TrimSpace(os.Getenv("DRIVE9_SLOCK_CLIENT_SECRET")),
		PublicURL:    strings.TrimSpace(os.Getenv("DRIVE9_PUBLIC_URL")),
	}
	if cfg.APIOrigin == "" || cfg.ClientID == "" || cfg.ClientSecret == "" || cfg.PublicURL == "" {
		return nil, fmt.Errorf("DRIVE9_SLOCK_ORIGIN enables Slock OAuth; DRIVE9_SLOCK_API_ORIGIN, DRIVE9_SLOCK_CLIENT_ID, DRIVE9_SLOCK_CLIENT_SECRET, and DRIVE9_PUBLIC_URL must also be set")
	}
	return slockoauth.New(cfg)
}

func s3ConfigFromEnv() s3Config {
	dir := strings.TrimSpace(os.Getenv("DRIVE9_S3_DIR"))
	if dir == "" {
		dir = defaultS3Dir
	}
	region := strings.TrimSpace(os.Getenv("DRIVE9_S3_REGION"))
	if region == "" {
		region = "us-east-1"
	}
	encryptionPolicy := meta.DefaultS3EncryptionPolicy()
	if mode := strings.TrimSpace(os.Getenv("DRIVE9_S3_ENCRYPTION_MODE")); mode != "" {
		encryptionPolicy.Mode = meta.S3EncryptionMode(mode)
	}
	encryptionPolicy.KMSKeyID = strings.TrimSpace(os.Getenv("DRIVE9_S3_KMS_KEY_ID"))
	encryptionPolicy.BucketKeyEnabled = envBool("DRIVE9_S3_BUCKET_KEY_ENABLED", true)
	return s3Config{
		Dir:              dir,
		Bucket:           strings.TrimSpace(os.Getenv("DRIVE9_S3_BUCKET")),
		Region:           region,
		Prefix:           strings.TrimSpace(os.Getenv("DRIVE9_S3_PREFIX")),
		RoleARN:          strings.TrimSpace(os.Getenv("DRIVE9_S3_ROLE_ARN")),
		Endpoint:         strings.TrimSpace(os.Getenv("DRIVE9_S3_ENDPOINT")),
		ForcePathStyle:   envBool("DRIVE9_S3_FORCE_PATH_STYLE", false),
		AccessKeyID:      strings.TrimSpace(os.Getenv("DRIVE9_S3_ACCESS_KEY_ID")),
		SecretAccessKey:  strings.TrimSpace(os.Getenv("DRIVE9_S3_SECRET_ACCESS_KEY")),
		SessionToken:     strings.TrimSpace(os.Getenv("DRIVE9_S3_SESSION_TOKEN")),
		EncryptionPolicy: encryptionPolicy,
		ExpressBucket:    strings.TrimSpace(os.Getenv("DRIVE9_S3_EXPRESS_BUCKET")),
		ExpressPrefix:    strings.TrimSpace(os.Getenv("DRIVE9_S3_EXPRESS_PREFIX")),
	}
}

func (cfg s3Config) validate() error {
	if err := meta.ValidateGlobalS3EncryptionPolicy(cfg.EncryptionPolicy); err != nil {
		return err
	}
	if cfg.Bucket == "" {
		return nil
	}
	return cfg.awsConfig().Validate()
}

func (cfg s3Config) awsConfig() s3client.AWSConfig {
	return s3client.AWSConfig{
		Region:          cfg.Region,
		Bucket:          cfg.Bucket,
		Prefix:          cfg.Prefix,
		RoleARN:         cfg.RoleARN,
		Endpoint:        cfg.Endpoint,
		ForcePathStyle:  cfg.ForcePathStyle,
		AccessKeyID:     cfg.AccessKeyID,
		SecretAccessKey: cfg.SecretAccessKey,
		SessionToken:    cfg.SessionToken,
	}
}

func versionText() string {
	return buildinfo.String("drive9-server")
}

func usage(out io.Writer, exitCode int) {
	_, _ = fmt.Fprintf(out, `usage:
  drive9-server [listen-addr]
  drive9-server version
  drive9-server schema dump-init-sql --provider <db9|tidb_zero|tidb_cloud_native|tidb_cloud_native_shared>

environment:
  DRIVE9_LISTEN_ADDR serve listen address (default: :9009)
  DRIVE9_PUBLIC_URL  externally reachable base URL for presigned URLs (required for remote clients)
  DRIVE9_META_DSN    control-plane MySQL DSN (required)
  DRIVE9_POOL_MAX_TENANTS max cached tenant backends per pod (default: 20480 for tidb_cloud_native_shared, 1024 otherwise)
  DRIVE9_POOL_IDLE_TTL  idle duration before a standalone tenant backend is evicted (default: 5m, 0=disabled; shared tenants use capacity LRU only)
  DRIVE9_POOL_IDLE_REAP_INTERVAL  how often the idle reaper scans (default: 2m)
  DRIVE9_META_DB_MAX_OPEN_CONNS max open connections for the per-pod meta DB pool (default: 100)
  DRIVE9_META_DB_MAX_IDLE_CONNS max idle connections for the per-pod meta DB pool (default: 20)
  DRIVE9_META_DB_CONN_MAX_LIFETIME maximum meta connection lifetime (default: 10m; non-positive/invalid values use default)
  DRIVE9_META_DB_CONN_MAX_IDLE_TIME maximum meta connection idle time (default: 1m, 0s=disabled)
  DRIVE9_USER_DB_MAX_OPEN_CONNS max open connections for each cached tenant user DB pool (default: 6)
  DRIVE9_USER_DB_MAX_IDLE_CONNS max idle connections for each cached tenant user DB pool (default: 2)
  DRIVE9_USER_SCHEMA_DB_MAX_OPEN_CONNS max open connections for tenant schema-init DB pools (default: 8)
  DRIVE9_USER_SCHEMA_DB_MAX_IDLE_CONNS max idle connections for tenant schema-init DB pools (default: 2)
  DRIVE9_TIDBCLOUD_CLUSTERS_BACKEND http|local TiDB Cloud control-plane backend (default: http; local = Docker TiDB per cluster)
  DRIVE9_SHARED_DB_MAX_OPEN_CONNS max open connections for each shared-schema DB handle (default: 300)
  DRIVE9_SHARED_DB_MAX_IDLE_CONNS max idle connections for each shared-schema DB handle (default: 50)
  DRIVE9_SHARED_DB_CONN_MAX_LIFETIME maximum shared connection lifetime (default: 30m)
  DRIVE9_SHARED_DB_CONN_MAX_IDLE_TIME maximum shared connection idle time (default: 5m)
  DRIVE9_DB_HEALTH_PROBE_INTERVAL_SECONDS DB health probe interval seconds (default: 15)
  DRIVE9_DB_HEALTH_PROBE_TIMEOUT_SECONDS DB health probe timeout seconds (default: 3)
  DRIVE9_DB_HEALTH_PROBE_META_ENABLED true|false to probe the control-plane DB (default: true)
  DRIVE9_ENCRYPT_TYPE local_aes|kms|aliyun_kms|tencent_kms
  DRIVE9_MASTER_KEY  32-byte hex key for local_aes encryptor
  DRIVE9_ENCRYPT_KEY KMS key id or alias (required for kms/aliyun_kms/tencent_kms)
  DRIVE9_ALIYUN_KMS_ENDPOINT custom Aliyun KMS endpoint, e.g. a VPC endpoint (no https:// prefix)
                             note: aliyun_kms/tencent_kms reads the region from DRIVE9_S3_REGION
                             note: TLS verification is skipped automatically when this is set
  DRIVE9_TOKEN_SIGNING_KEY  32-byte hex key for JWT API key signing
  DRIVE9_VAULT_MASTER_KEY   32-byte hex key for vault DEK wrapping (omit to disable vault)
  DRIVE9_MAX_UPLOAD_BYTES maximum allowed upload size in bytes (default: %d, minimum: 1048576)
  DRIVE9_DEFAULT_STORAGE_QUOTA_BYTES fallback per-tenant total storage limit when no explicit quota is configured (default: %d)
  DRIVE9_LOG_LEVEL debug|info|warn|error (default: info)
  DRIVE9_BENCH_TIMING_LOG_ENABLED true|false to emit benchmark timing logs on successful server hot paths (default: false)
  DRIVE9_OPEN_POOL_TIMING_LOG_ENABLED true|false to emit slow tenant pool and authentication timing logs (default: true)
  DRIVE9_OPEN_POOL_TIMING_SLOW_MS minimum duration in ms for open-pool timing logs (default: 500, 0=all)
  DRIVE9_DB_TRACE_LOG_ENABLED true|false to emit DB operation trace logs with redacted SQL (default: true)
  DRIVE9_DB_SLOW_TRACE_MS minimum DB operation duration in ms for DB trace logs (default: 300, 0=all)
  DRIVE9_QUOTA_USAGE_CACHE_TTL soft small-write central usage cache TTL, e.g. 250ms or 1s
  DRIVE9_QUOTA_PENDING_DELTAS_CACHE_TTL soft small-write in-process pending mutation cache TTL, e.g. 250ms or 1s
  DRIVE9_QUOTA_REPLAY_POLL_MS central quota mutation replay poll interval in milliseconds (default: 1000)
  DRIVE9_QUOTA_REPLAY_MIN_AGE_MS minimum pending mutation age before replay in milliseconds (default: 1000)
  DRIVE9_QUOTA_REPLAY_OBSERVE_MS replay backlog observation interval in milliseconds (default: 10000)
  DRIVE9_BILLING_OPS_FLUSH_INTERVAL per-pod flush interval for billing ops counter deltas (default: 0 = disabled; set e.g. 30s to enable metering in a region)
  DRIVE9_OBJECT_GC_GRACE_PERIOD delay before a new ordinary object GC candidate is eligible (default: 24h; failed_write uses a fixed 1h safety grace)
  DRIVE9_OBJECT_GC_BATCH_SIZE due object GC candidates processed per minute poll (default: 500, max: 5000)
  DRIVE9_DISABLE_AUTO_EMBEDDING true|false disable TiDB database-managed auto-embedding (default: false)
                                set to true when the TiDB Cloud cluster has no supported embedding provider
  DRIVE9_TIDB_AUTO_EMBEDDING_MODEL TiDB EMBED_TEXT model for auto-embedding generated columns
                                   supported provider prefixes include tidbcloud_free/amazon, tidbcloud_free/cohere,
                                   openai, cohere, jina_ai, gemini, huggingface, nvidia_nim, nvidia
                                   (default: tidbcloud_free/amazon/titan-embed-text-v2)
  DRIVE9_TIDB_AUTO_EMBEDDING_DIMENSIONS TiDB EMBED_TEXT vector dimensions
                                        known models use documented defaults; unknown BYOK models require this value
  DRIVE9_TIDB_AUTO_EMBEDDING_API_KEY provider API key for BYOK auto-embedding models
                                    required by openai, cohere, jina_ai, gemini, huggingface, nvidia_nim, nvidia
  DRIVE9_TIDB_AUTO_EMBEDDING_API_BASE provider base endpoint for models that require it
                                     optional for openai models; set it for Azure OpenAI endpoints
  DRIVE9_TENANT_PROVIDER db9|tidb_zero|tidb_cloud_native|tidb_cloud_native_shared|local (default for provisioning)
  DRIVE9_LOCAL_MYSQL_DSN admin DSN for provider=local (falls back to DRIVE9_LOCAL_DSN or DRIVE9_META_DSN)
  DRIVE9_LOCAL_EMBEDDING_MODE app for provider=local tenant schema (default app)
  DRIVE9_LOCAL_ALLOW_NON_LOOPBACK allow provider=local to bind a non-loopback address (insecure: no tenant SQL isolation)
  DRIVE9_TIDBCLOUD_DEFAULT_SPENDING_LIMIT default TiDB Cloud Cluster spendingLimit.monthly; native defaults to 1000 when unset
  DRIVE9_TIDBCLOUD_NATIVE_API_URL TiDB Cloud Cluster API base URL for tidb_cloud_native
  DRIVE9_TIDBCLOUD_IAM_API_MODE openapi|inner for shared-mode control-plane IAM authentication (default: openapi)
  DRIVE9_TIDBCLOUD_BILLING_API_MODE openapi|inner for shared-mode Billing plan lookups (default: openapi)
  DRIVE9_TIDBCLOUD_INNER_API_TOKEN shared Basic Auth token for inner IAM and Billing APIs
  DRIVE9_TIDBCLOUD_DATA_API_URL data-plane inner API base URL for server-owned shared credential validation
                                 (tidb_cloud_native_shared only; when set, shared validation uses the inner
                                 API regardless of the IAM/Billing API modes)
  DRIVE9_TIDBCLOUD_DATA_API_TOKEN Basic Auth token for DRIVE9_TIDBCLOUD_DATA_API_URL (required when it is set)
  DRIVE9_TIDBCLOUD_IAM_API_URL TiDB Cloud IAM API base URL for cluster API identity lookups
  DRIVE9_TIDBCLOUD_BILLING_API_URL TiDB Cloud Billing API base URL when Billing mode is openapi
  DRIVE9_TIDBCLOUD_IAM_CACHE_TTL TiDB Cloud API-key IAM cache TTL (default: 30m)
  DRIVE9_TIDBCLOUD_NON_FREE_PLAN_CACHE_TTL TiDB Cloud Billing non-free plan cache TTL (default: 15m)
  DRIVE9_TIDBCLOUD_NATIVE_CLOUD_PROVIDER cloud provider for tidb_cloud_native cluster creation, e.g. aws
  DRIVE9_TIDBCLOUD_NATIVE_REGION region for tidb_cloud_native cluster creation, e.g. us-east-1
  DRIVE9_TIDBCLOUD_NATIVE_DEFAULT_DATABASE_NAME default tidb_cloud_native database name (default: tidbcloud_fs)
  DRIVE9_TIDBCLOUD_NATIVE_PUBLIC_KEY optional default TiDB Cloud API public key for tidb_cloud_native create/delete when caller omits it
  DRIVE9_TIDBCLOUD_NATIVE_PRIVATE_KEY optional default TiDB Cloud API private key for tidb_cloud_native create/delete when caller omits it
  DRIVE9_TIDBCLOUD_NATIVE_SHARED_PUBLIC_KEY server-managed TiDB Cloud API public key for shared DB pool operations; required for tidb_cloud_native_shared
  DRIVE9_TIDBCLOUD_NATIVE_SHARED_PRIVATE_KEY server-managed TiDB Cloud API private key for shared DB pool operations; required for tidb_cloud_native_shared
  DRIVE9_TIDBCLOUD_NATIVE_USE_PRIVATE_ENDPOINT true|false to use TiDB Cloud private endpoint hosts for tidb_cloud_native
  DRIVE9_TIDBCLOUD_NATIVE_SHARED_MAX_TENANTS soft capacity for new managed shared DB pools (default: 100)
  DRIVE9_TIDBCLOUD_NATIVE_SHARED_HARD_CAP_RATIO emergency hard-cap ratio > 1 after physical create failure (default: 1.2)
  DRIVE9_TIDBCLOUD_NATIVE_SHARED_REOPEN_RATIO reopen ratio for a latched shared pool (default: 0.8)
  DRIVE9_TIDBCLOUD_NATIVE_DB_POOL_DEFAULT_SPENDING_LIMIT physical spending-limit target for new managed shared DB pools (default: 1000000)
  DRIVE9_TIDBCLOUD_NATIVE_SHARED_CLOUD_BATCH_SIZE maximum physical pools in one refill Cloud request (default: 50)
  DRIVE9_TIDBCLOUD_NATIVE_SHARED_METADATA_WORKERS concurrent pending metadata workers (default and maximum: 15)
  DRIVE9_TIDBCLOUD_NATIVE_SHARED_METADATA_BATCH_SIZE physical pools per metadata list request (default and maximum: 30)
  DRIVE9_TIDBCLOUD_NATIVE_SHARED_METADATA_POLL_INTERVAL delay between metadata list rounds (default: 15s)
  DRIVE9_TIDBCLOUD_NATIVE_SHARED_PROVISIONING_WORKERS concurrent shared schema-init workers (default: 100)
  DRIVE9_TENANT_POOL_RECONCILE_INTERVAL leader logical-pool refill scan interval (default: 5s)
  DRIVE9_TENANT_POOL_RECONCILE_WORKERS concurrent queued leader logical-pool refill workers (default: 15)
  DRIVE9_TENANT_POOL_RECONCILE_WORKER_REST delay before a completed refill worker takes another task (default: 5s)
  DRIVE9_TIDBCLOUD_NATIVE_SHARED_STUCK_TIMEOUT pending/provisioning no-progress timeout (default: 30m)
  DRIVE9_TIDBCLOUD_NATIVE_SHARED_FAILED_CLEANUP_INTERVAL failed physical-pool cleanup interval (default: 1m)
  DRIVE9_TIDBCLOUD_NATIVE_SHARED_FAILED_CLEANUP_BATCH_SIZE failed physical pools processed per cleanup pass (default: 5)
  DRIVE9_TIDBCLOUD_PRIVATE_ENDPOINT_HOST_MAP comma-separated public_host=private_host mappings (also accepts public_host:private_host);
                                             when set, disables legacy single-host private endpoint overrides
  DRIVE9_TIDBCLOUD_TENCENT_PRIVATE_ENDPOINT_HOST legacy tencentcloud private endpoint fallback host, used only when host map is unset
  DRIVE9_TIDBCLOUD_ALICLOUD_PRIVATE_ENDPOINT_DOMAIN legacy alicloud private endpoint fallback domain, used only when host map is unset
  DRIVE9_TENANT_POOL_MAX_SIZE maximum admin tenant pool size (default: %d)
  DRIVE9_TENANT_POOL_REFILL_FREE_RATIO claim-triggered refill runs when active free tenants fall below this ratio (default: %.2f)
  DRIVE9_SLOCK_ORIGIN Slock browser origin; when set, enables /v1/auth/slock/*
  DRIVE9_SLOCK_API_ORIGIN Slock API origin (required when DRIVE9_SLOCK_ORIGIN is set)
  DRIVE9_SLOCK_CLIENT_ID Slock connected-app client id (required when DRIVE9_SLOCK_ORIGIN is set)
  DRIVE9_SLOCK_CLIENT_SECRET Slock connected-app client secret (required when DRIVE9_SLOCK_ORIGIN is set)

  SSE cross-pod notification (set DRIVE9_POD_ID to enable; required for large multi-tenant deployments):
  DRIVE9_POD_ID     unique pod identifier; when set, this pod registers in the central
                    pod_registry and reports its SSE subscriber tenant set. The outbox
                    poller runs on all multi-tenant pods regardless.
  DRIVE9_POD_ADDR   full base URL (e.g. http://10.0.0.5:9009) advertised in the pod
                    registry. Defaults to http://<listen-addr> when the listen address
                    is a non-wildcard host; must be set explicitly in multi-host
                    deployments.
  DRIVE9_SSE_NOTIFY_RETENTION      how long tenant_notify_outbox rows are kept before
                    leader pruning (default: 1h)
  DRIVE9_FS_EVENTS_RETENTION       how long fs_events rows are kept before lazy
                    write-path and piggyback-maintenance pruning (default: 1h;
                    production: 168h). Independent of DRIVE9_SSE_NOTIFY_RETENTION:
                    fs_events is the durable replay log, the outbox is a lossy hint.
  DRIVE9_FS_EVENTS_LAZY_SWEEP_INTERVAL
                    per-tenant throttle for the write-path fs_events retention
                    sweep (default: 1h). Keep 1h in production; local e2e may
                    shorten it to observe pruning quickly.
  DRIVE9_SSE_LIVENESS_POLL_INTERVAL
                    optional per-connection SSE liveness poll interval covering
                    signal loss for connected clients (default: 0 = off)
  DRIVE9_TENANT_OUTBOX_POLL_INTERVAL     unified outbox poll interval (default: 200ms)
  DRIVE9_TENANT_OUTBOX_CURSOR_FLUSH_INTERVAL
                    how often the poller persists its cursor (default: 5s)
  DRIVE9_TENANT_SHARD_REFRESH_INTERVAL   shard resolver pod ring refresh interval (default: 5s)
  DRIVE9_TENANT_MAINTENANCE_INTERVAL     piggyback maintenance throttle per tenant (default: 5m)
  DRIVE9_SAFETY_NET_SCAN_INTERVAL        per-pod safety-net scan interval (default: 5m);
                    0 disables the scan
  Interval vars accept Go durations (e.g. 500ms, 5m). The former *_MS names
  (DRIVE9_TENANT_OUTBOX_POLL_INTERVAL_MS, DRIVE9_TENANT_OUTBOX_CURSOR_FLUSH_MS,
  DRIVE9_TENANT_SHARD_REFRESH_MS, DRIVE9_TENANT_MAINTENANCE_INTERVAL_MS,
  DRIVE9_SAFETY_NET_SCAN_INTERVAL_MS) are deprecated but still honored.

  S3 storage (set DRIVE9_S3_BUCKET to enable AWS S3, otherwise local mock):
  DRIVE9_S3_BUCKET   S3 bucket name (enables AWS S3 mode)
  DRIVE9_S3_REGION   AWS region (default: us-east-1)
  DRIVE9_S3_PREFIX   S3 key prefix (e.g. "tenants")
  DRIVE9_S3_ENDPOINT custom S3 endpoint URL for S3-compatible stores such as MinIO (optional)
  DRIVE9_S3_FORCE_PATH_STYLE true|false to force path-style S3 URLs (default: false)
  DRIVE9_S3_ACCESS_KEY_ID static S3 access key id (optional; requires DRIVE9_S3_SECRET_ACCESS_KEY)
  DRIVE9_S3_SECRET_ACCESS_KEY static S3 secret access key (optional; requires DRIVE9_S3_ACCESS_KEY_ID)
  DRIVE9_S3_SESSION_TOKEN static S3 session token (optional; requires DRIVE9_S3_ACCESS_KEY_ID and DRIVE9_S3_SECRET_ACCESS_KEY)
  DRIVE9_S3_ROLE_ARN IAM role ARN to assume via STS (optional)
  DRIVE9_S3_DIR      local s3 mock root directory (default: ./s3, only used without DRIVE9_S3_BUCKET)
  DRIVE9_S3_ENCRYPTION_MODE none|sse-s3|sse-kms|dsse-kms (default: none; recommend sse-kms for production)
  DRIVE9_S3_KMS_KEY_ID KMS key ARN/id required for sse-kms/dsse-kms
  DRIVE9_S3_BUCKET_KEY_ENABLED true|false for SSE-KMS bucket keys (default: true)

  App-managed embedding (shared and native fts_only; used for writes and queries):
  DRIVE9_EMBED_ENABLED  true|false (default: false)
  DRIVE9_EMBED_API_BASE OpenAI-compatible base URL (required when enabled)
  DRIVE9_EMBED_API_KEY  API key for DRIVE9_EMBED_API_BASE (required when enabled)
  DRIVE9_EMBED_MODEL    model name (required when enabled)
  DRIVE9_EMBED_TIMEOUT_SECONDS embed request timeout seconds (default: 20)

  DRIVE9_SEMANTIC_WORKERS number of background workers (default: 1)
  DRIVE9_SEMANTIC_POLL_INTERVAL_MS worker poll interval in milliseconds (default: 200)
  DRIVE9_SEMANTIC_LEASE_SECONDS task lease duration in seconds (default: 30)
  DRIVE9_SEMANTIC_RETRY_BASE_MS base retry backoff in milliseconds (default: 200)
  DRIVE9_SEMANTIC_RETRY_MAX_MS max retry backoff in milliseconds (default: 30000)
  DRIVE9_SEMANTIC_PER_TENANT_CONCURRENCY max concurrent tasks per tenant (default: 1)

  Image extraction (async image -> text for search):
  DRIVE9_IMAGE_EXTRACT_ENABLED true|false (default: false)
  DRIVE9_IMAGE_EXTRACT_QUEUE_SIZE buffered task queue size (default: 128)
  DRIVE9_IMAGE_EXTRACT_WORKERS number of workers (default: 1)
  DRIVE9_IMAGE_EXTRACT_MAX_BYTES max image bytes processed per task (default: 8388608)
  DRIVE9_IMAGE_EXTRACT_TIMEOUT_SECONDS extractor timeout seconds (default: 120)
  DRIVE9_IMAGE_EXTRACT_MAX_TEXT_BYTES max extracted text stored in files.content_text (default: 32768)
  DRIVE9_IMAGE_EXTRACT_API_BASE OpenAI-compatible base URL (optional)
  DRIVE9_IMAGE_EXTRACT_API_KEY  API key for DRIVE9_IMAGE_EXTRACT_API_BASE (optional)
  DRIVE9_IMAGE_EXTRACT_MODEL    model name for vision extraction (optional)
  DRIVE9_IMAGE_EXTRACT_PROMPT   custom extraction prompt (optional)
  DRIVE9_IMAGE_EXTRACT_MAX_TOKENS max model output tokens (default: 4096)
  DRIVE9_EXTRACT_TMP_DIR media extraction task workspace (default: system temp directory)
  DRIVE9_EXTRACT_MAX_CONCURRENT_TASKS max concurrent media staging/preprocessing tasks (default: 4)

  Audio extraction (async audio -> text for search; MVP durable path is TiDB auto-embedding only):
  Durable audio_extract_text tasks enqueue only for tenants with database auto-embedding
  (tidb_zero / tidb_cloud_native). For db9-only or other app-managed tenants these vars do
  not enable that semantic_tasks pipeline. When enabled, explicit provider wiring is required:
  DRIVE9_AUDIO_EXTRACT_ENABLED true|false (default: false)
  DRIVE9_AUDIO_EXTRACT_MODE     openai|qwen-asr (default: openai); stub only with DRIVE9_TENANT_PROVIDER=local
  DRIVE9_AUDIO_EXTRACT_MAX_BYTES max audio bytes processed per task (default: 33554432 for openai, 7340032 for qwen-asr)
  DRIVE9_AUDIO_EXTRACT_TIMEOUT_SECONDS extractor timeout seconds (default: 120)
  DRIVE9_AUDIO_EXTRACT_MAX_TEXT_BYTES max transcript bytes stored in files.content_text (default: 8192)
  DRIVE9_AUDIO_EXTRACT_API_BASE OpenAI-compatible base URL (required when enabled)
  DRIVE9_AUDIO_EXTRACT_API_KEY  API key for DRIVE9_AUDIO_EXTRACT_API_BASE (required when enabled)
  DRIVE9_AUDIO_EXTRACT_MODEL    model name for audio transcription (required when enabled)
  DRIVE9_AUDIO_EXTRACT_PROMPT   optional provider prompt for transcription

schema tooling:
  dump-init-sql writes the exact init schema SQL to stdout so external systems
  can stay in sync with drive9's schema source of truth.
`, server.DefaultMaxUploadBytes, meta.DefaultMaxStorageBytes(), server.DefaultTenantPoolMaxSize, server.DefaultTenantPoolRefillFreeRatio)
	os.Exit(exitCode)
}

func die(err error) {
	if err == nil {
		return
	}
	fmt.Fprintf(os.Stderr, "drive9-server: %v\n", err)
	os.Exit(1)
}

// vaultIssuerURL returns the canonical issuer URL used as the `iss` claim on
// vault grants. DRIVE9_VAULT_ISSUER_URL is the explicit override for sites that
// want the grant issuer to be a different URL than the public object URL (e.g.
// when signed URLs go to a CDN and grant validation happens at a control-plane
// host). When unset, we fall back to the same canonical URL the object plane
// uses, which is what the end-state spec §16 expects (single server identity).
func vaultIssuerURL(listenAddr string) string {
	if v := strings.TrimRight(strings.TrimSpace(os.Getenv("DRIVE9_VAULT_ISSUER_URL")), "/"); v != "" {
		return v
	}
	return publicBaseURL(listenAddr)
}

func publicBaseURL(listenAddr string) string {
	if v := strings.TrimRight(os.Getenv("DRIVE9_PUBLIC_URL"), "/"); v != "" {
		return v
	}

	// Without DRIVE9_PUBLIC_URL, only allow explicit loopback addresses.
	// Wildcard or non-loopback addresses would produce unreachable presigned URLs.
	switch {
	case strings.HasPrefix(listenAddr, "127.0.0.1:"),
		strings.HasPrefix(listenAddr, "localhost:"),
		strings.HasPrefix(listenAddr, "[::1]:"):
		return "http://" + listenAddr
	case strings.HasPrefix(listenAddr, "http://"), strings.HasPrefix(listenAddr, "https://"):
		return strings.TrimRight(listenAddr, "/")
	default:
		fmt.Fprintf(os.Stderr, "drive9-server: DRIVE9_PUBLIC_URL is required when listen address is %q (wildcard or non-loopback). Set DRIVE9_PUBLIC_URL to the externally reachable base URL.\n", listenAddr)
		os.Exit(1)
		return "" // unreachable
	}
}

func applyQuotaDefaultsFromEnv() {
	if raw := os.Getenv("DRIVE9_MEDIA_EXTRACT_MAX_FILES"); raw != "" {
		if v, err := strconv.ParseInt(raw, 10, 64); err == nil && v > 0 {
			meta.SetDefaultMaxMediaLLMFiles(v)
		}
	}
	if raw := os.Getenv("DRIVE9_VIDEO_EXTRACT_MAX_FILES"); raw != "" {
		if v, err := strconv.ParseInt(raw, 10, 64); err == nil && v > 0 {
			meta.SetDefaultMaxVideoLLMFiles(v)
		}
	}
}

func buildBackendOptionsFromEnv() (backend.Options, *server.ExtractDefaultProvider, error) {
	var opts backend.Options
	opts.ExtractTmpDir = strings.TrimSpace(os.Getenv("DRIVE9_EXTRACT_TMP_DIR"))
	opts.ExtractMaxConcurrentTasks = backend.DefaultMediaExtractMaxConcurrentTasks
	if raw := strings.TrimSpace(os.Getenv("DRIVE9_EXTRACT_MAX_CONCURRENT_TASKS")); raw != "" {
		maxConcurrentTasks, err := strconv.Atoi(raw)
		if err != nil || maxConcurrentTasks <= 0 {
			return backend.Options{}, nil, fmt.Errorf("DRIVE9_EXTRACT_MAX_CONCURRENT_TASKS must be a positive integer")
		}
		opts.ExtractMaxConcurrentTasks = maxConcurrentTasks
	}
	if strings.TrimSpace(os.Getenv("DRIVE9_QUOTA_SOURCE")) != "" {
		return backend.Options{}, nil, fmt.Errorf("DRIVE9_QUOTA_SOURCE has been removed; central quota is now driven by meta-store wiring")
	}
	opts.MaxTenantStorageBytes = envInt64("DRIVE9_MAX_TENANT_STORAGE_BYTES", 50*(1<<30))
	if opts.MaxTenantStorageBytes <= 0 {
		return backend.Options{}, nil, fmt.Errorf("DRIVE9_MAX_TENANT_STORAGE_BYTES must be a positive integer")
	}
	objectGCGracePeriod, err := objectGCGracePeriodFromEnv()
	if err != nil {
		return backend.Options{}, nil, err
	}
	opts.ObjectGCGracePeriod = objectGCGracePeriod

	opts.InlineThreshold = envInt64("DRIVE9_INLINE_THRESHOLD", backend.DefaultInlineThreshold)
	if opts.InlineThreshold <= 0 {
		return backend.Options{}, nil, fmt.Errorf("DRIVE9_INLINE_THRESHOLD must be a positive integer")
	}
	opts.TextExtractMaxBytes = envInt64("DRIVE9_TEXT_EXTRACT_MAX_BYTES", backend.DefaultTextExtractMaxBytes)
	if opts.TextExtractMaxBytes <= 0 {
		return backend.Options{}, nil, fmt.Errorf("DRIVE9_TEXT_EXTRACT_MAX_BYTES must be a positive integer")
	}

	imageExtract, imageExtractDefaultProvider, err := buildImageExtractOptionsFromEnv()
	if err != nil {
		return backend.Options{}, nil, err
	}
	opts.AsyncImageExtract = imageExtract
	audioExtract, err := buildAudioExtractOptionsFromEnv()
	if err != nil {
		return backend.Options{}, nil, err
	}
	opts.AsyncAudioExtract = audioExtract
	videoExtract, err := buildVideoExtractOptionsFromEnv()
	if err != nil {
		return backend.Options{}, nil, err
	}
	opts.AsyncVideoExtract = videoExtract
	return opts, imageExtractDefaultProvider, nil
}

func objectGCGracePeriodFromEnv() (time.Duration, error) {
	const key = "DRIVE9_OBJECT_GC_GRACE_PERIOD"
	raw := strings.TrimSpace(os.Getenv(key))
	if raw == "" {
		return backend.DefaultObjectGCGracePeriod, nil
	}
	value, err := time.ParseDuration(raw)
	if err != nil {
		return 0, fmt.Errorf("%s must be a valid duration: %w", key, err)
	}
	if value <= 0 {
		return 0, fmt.Errorf("%s must be positive", key)
	}
	return value, nil
}

func objectGCBatchSizeFromEnv() int {
	const key = "DRIVE9_OBJECT_GC_BATCH_SIZE"
	raw := strings.TrimSpace(os.Getenv(key))
	if raw == "" {
		return server.DefaultObjectGCBatchSize
	}
	value, err := strconv.Atoi(raw)
	if err != nil || value <= 0 || value > server.MaxObjectGCBatchSize {
		fmt.Fprintf(os.Stderr, "invalid %s=%q: must be in [1,%d]; using %d\n",
			key, raw, server.MaxObjectGCBatchSize, server.DefaultObjectGCBatchSize)
		return server.DefaultObjectGCBatchSize
	}
	return value
}

// imageExtractTimeoutFromEnv parses DRIVE9_IMAGE_EXTRACT_TIMEOUT_SECONDS
// (seconds, default 120). It feeds both the process-default extractor and the
// tenant resolver fallback so the two can never drift apart.
func imageExtractTimeoutFromEnv() time.Duration {
	return time.Duration(envInt("DRIVE9_IMAGE_EXTRACT_TIMEOUT_SECONDS", 120)) * time.Second
}

func buildImageExtractOptionsFromEnv() (backend.AsyncImageExtractOptions, *server.ExtractDefaultProvider, error) {
	async := backend.AsyncImageExtractOptions{
		MaxSourceBytes: envInt64("DRIVE9_IMAGE_EXTRACT_MAX_SOURCE_BYTES", backend.DefaultImageExtractMaxSourceBytes),
	}
	if async.MaxSourceBytes <= 0 {
		return backend.AsyncImageExtractOptions{}, nil, fmt.Errorf("DRIVE9_IMAGE_EXTRACT_MAX_SOURCE_BYTES must be a positive integer")
	}
	if !envBool("DRIVE9_IMAGE_EXTRACT_ENABLED", false) {
		return async, nil, nil
	}
	async.Enabled = true
	async.QueueSize = envInt("DRIVE9_IMAGE_EXTRACT_QUEUE_SIZE", 128)
	async.Workers = envInt("DRIVE9_IMAGE_EXTRACT_WORKERS", 1)
	async.MaxImageBytes = envInt64("DRIVE9_IMAGE_EXTRACT_MAX_BYTES", 8<<20)
	async.TaskTimeout = imageExtractTimeoutFromEnv()
	async.MaxExtractTextBytes = envInt("DRIVE9_IMAGE_EXTRACT_MAX_TEXT_BYTES", backend.DefaultImageExtractMaxTextBytes)

	baseURL := strings.TrimSpace(os.Getenv("DRIVE9_IMAGE_EXTRACT_API_BASE"))
	apiKey := strings.TrimSpace(os.Getenv("DRIVE9_IMAGE_EXTRACT_API_KEY"))
	model := strings.TrimSpace(os.Getenv("DRIVE9_IMAGE_EXTRACT_MODEL"))
	prompt := strings.TrimSpace(os.Getenv("DRIVE9_IMAGE_EXTRACT_PROMPT"))
	maxTokens := envInt("DRIVE9_IMAGE_EXTRACT_MAX_TOKENS", backend.DefaultOpenAIImageExtractMaxTokens)

	configured := baseURL != "" || apiKey != "" || model != ""
	if configured {
		if baseURL == "" || apiKey == "" || model == "" {
			logger.Error(context.Background(), "image_extract_mode_invalid_config",
				zap.Bool("base_url_present", baseURL != ""),
				zap.Bool("api_key_present", apiKey != ""),
				zap.Bool("model_present", model != ""))
			return backend.AsyncImageExtractOptions{}, nil, fmt.Errorf("DRIVE9_IMAGE_EXTRACT_API_BASE, DRIVE9_IMAGE_EXTRACT_API_KEY and DRIVE9_IMAGE_EXTRACT_MODEL must be set together")
		}
		extractor, err := backend.NewOpenAIImageTextExtractor(backend.OpenAIImageTextExtractorConfig{
			BaseURL:   baseURL,
			APIKey:    apiKey,
			Model:     model,
			Prompt:    backend.ComposeImageExtractPrompt(prompt),
			MaxTokens: maxTokens,
			Timeout:   async.TaskTimeout,
		})
		if err != nil {
			return backend.AsyncImageExtractOptions{}, nil, fmt.Errorf("init image extractor: %w", err)
		}
		async.Extractor = backend.NewFallbackImageTextExtractor(extractor, backend.NewBasicImageTextExtractor())
		logger.Info(context.Background(), "image_extract_mode_openai_compatible",
			zap.String("model", model), zap.String("base_url", baseURL))
		return async, &server.ExtractDefaultProvider{
			APIBase: baseURL,
			APIKey:  apiKey,
			Model:   model,
			Prompt:  prompt,
		}, nil
	} else {
		async.Extractor = backend.NewBasicImageTextExtractor()
		logger.Info(context.Background(), "image_extract_mode_basic_fallback")
	}

	return async, nil, nil
}

// withExtractResolverCapabilities marks backend options with the media task
// capabilities supplied by the per-tenant resolver. They are independent of
// process-default providers and do not perform MetaDB reads during upload.
func withExtractResolverCapabilities(opts backend.Options) backend.Options {
	opts.AsyncImageExtract.Capability = true
	opts.AsyncAudioExtract.Capability = true
	opts.AsyncVideoExtract.Capability = true
	return opts
}

// imageExtractGlobalDisabledFromEnv reads the cluster image-extract kill
// switch (default false).
func imageExtractGlobalDisabledFromEnv() bool {
	return envBool("DRIVE9_IMAGE_EXTRACT_GLOBAL_DISABLED", false)
}

// defaultImageExtractRuntime builds the process-default image extract runtime
// from the async options; nil when the process-default extractor is disabled.
func defaultImageExtractRuntime(async backend.AsyncImageExtractOptions) *backend.ImageExtractRuntime {
	if !async.Enabled || async.Extractor == nil {
		return nil
	}
	_, basicOnly := async.Extractor.(*backend.BasicImageTextExtractor)
	return &backend.ImageExtractRuntime{
		Extractor:               async.Extractor,
		Timeout:                 async.TaskTimeout,
		MaxSourceBytes:          async.MaxSourceBytes,
		MaxImageBytes:           async.MaxImageBytes,
		MaxExtractTextBytes:     async.MaxExtractTextBytes,
		RequireRasterImageBytes: !basicOnly,
	}
}

func defaultAudioExtractRuntime(async backend.AsyncAudioExtractOptions, workspace *backend.MediaExtractWorkspace) *backend.AudioExtractRuntime {
	if !async.Enabled || async.Extractor == nil {
		return nil
	}
	return &backend.AudioExtractRuntime{
		Extractor: async.Extractor, Timeout: async.TaskTimeout, MaxSourceBytes: async.MaxSourceBytes, MaxAudioBytes: async.MaxAudioBytes,
		MaxExtractTextBytes: async.MaxExtractTextBytes, Workspace: workspace,
	}
}

func defaultVideoExtractRuntime(async backend.AsyncVideoExtractOptions, workspace *backend.MediaExtractWorkspace) *backend.VideoExtractRuntime {
	if !async.Enabled || async.Extractor == nil {
		return nil
	}
	return &backend.VideoExtractRuntime{
		Extractor: async.Extractor, Timeout: async.TaskTimeout, MaxSourceBytes: async.MaxSourceBytes,
		MaxExtractTextBytes: async.MaxExtractTextBytes, Workspace: workspace,
	}
}

// configureExtractResolver wires the per-tenant extract config resolver into
// srv when both the control-plane store and the tenant pool exist, and
// reports whether it was wired.
func configureExtractResolver(srv *server.Server, store *meta.Store, pool *tenant.Pool, opts backend.Options, provider *server.ExtractDefaultProvider) (bool, error) {
	if store == nil || pool == nil {
		return false, nil
	}
	workspace, err := backend.NewMediaExtractWorkspace(backend.MediaExtractWorkspaceOptions{
		RootDir:            opts.ExtractTmpDir,
		MaxConcurrentTasks: opts.ExtractMaxConcurrentTasks,
	})
	if err != nil {
		return false, fmt.Errorf("configure extract workspace: %w", err)
	}
	defaults := server.ExtractResolverDefaults{
		Image:                        defaultImageExtractRuntime(opts.AsyncImageExtract),
		Audio:                        defaultAudioExtractRuntime(opts.AsyncAudioExtract, workspace),
		AudioProtocol:                opts.AsyncAudioExtract.Protocol,
		Video:                        defaultVideoExtractRuntime(opts.AsyncVideoExtract, workspace),
		ImageMaxSourceBytes:          opts.AsyncImageExtract.MaxSourceBytes,
		AudioMaxSourceBytes:          opts.AsyncAudioExtract.MaxSourceBytes,
		VideoMaxSourceBytes:          opts.AsyncVideoExtract.MaxSourceBytes,
		VideoMaxProviderPayloadBytes: opts.AsyncVideoExtract.MaxProviderPayloadBytes,
		AudioTenantFallbackTimeout:   opts.AsyncAudioExtract.TaskTimeout,
		VideoTenantFallbackTimeout:   opts.AsyncVideoExtract.TaskTimeout,
		Workspace:                    workspace,
		ImageTenantFallbackTimeout:   imageExtractTimeoutFromEnv(),
	}
	if defaults.Video != nil {
		defaults.Video.AudioRuntime = defaults.Audio
	}
	srv.ConfigureExtractResolverDefaults(store, pool, defaults, provider)
	return true, nil
}

func buildAudioExtractOptionsFromEnv() (backend.AsyncAudioExtractOptions, error) {
	async := backend.AsyncAudioExtractOptions{
		MaxSourceBytes: envInt64("DRIVE9_AUDIO_EXTRACT_MAX_SOURCE_BYTES", backend.DefaultAudioExtractMaxSourceBytes),
	}
	if async.MaxSourceBytes <= 0 {
		return backend.AsyncAudioExtractOptions{}, fmt.Errorf("DRIVE9_AUDIO_EXTRACT_MAX_SOURCE_BYTES must be a positive integer")
	}
	if !envBool("DRIVE9_AUDIO_EXTRACT_ENABLED", false) {
		return async, nil
	}
	audioMode := strings.ToLower(strings.TrimSpace(os.Getenv("DRIVE9_AUDIO_EXTRACT_MODE")))
	if audioMode == "" {
		audioMode = "openai"
	}
	if audioMode == "stub" {
		if !localProviderFromEnv() {
			return backend.AsyncAudioExtractOptions{}, fmt.Errorf("DRIVE9_AUDIO_EXTRACT_MODE=stub is only allowed when DRIVE9_TENANT_PROVIDER=local")
		}
		return backend.AsyncAudioExtractOptions{
			Enabled:             true,
			Protocol:            "stub",
			MaxSourceBytes:      async.MaxSourceBytes,
			MaxAudioBytes:       envInt64("DRIVE9_AUDIO_EXTRACT_MAX_BYTES", 0),
			TaskTimeout:         time.Duration(envInt("DRIVE9_AUDIO_EXTRACT_TIMEOUT_SECONDS", 0)) * time.Second,
			MaxExtractTextBytes: envInt("DRIVE9_AUDIO_EXTRACT_MAX_TEXT_BYTES", 0),
			Extractor:           stubAudioTextExtractor{},
		}, nil
	}
	audioBaseURL := strings.TrimSpace(os.Getenv("DRIVE9_AUDIO_EXTRACT_API_BASE"))
	audioAPIKey := strings.TrimSpace(os.Getenv("DRIVE9_AUDIO_EXTRACT_API_KEY"))
	audioModel := strings.TrimSpace(os.Getenv("DRIVE9_AUDIO_EXTRACT_MODEL"))
	audioPrompt := strings.TrimSpace(os.Getenv("DRIVE9_AUDIO_EXTRACT_PROMPT"))
	if audioBaseURL == "" || audioAPIKey == "" || audioModel == "" {
		return backend.AsyncAudioExtractOptions{}, fmt.Errorf("DRIVE9_AUDIO_EXTRACT_API_BASE, DRIVE9_AUDIO_EXTRACT_API_KEY and DRIVE9_AUDIO_EXTRACT_MODEL must be set together when DRIVE9_AUDIO_EXTRACT_ENABLED=true")
	}
	audioTimeout := time.Duration(envInt("DRIVE9_AUDIO_EXTRACT_TIMEOUT_SECONDS", 120)) * time.Second
	maxAudioBytesDefault := int64(32 << 20)
	if audioMode == "qwen-asr" {
		// Qwen-ASR enforces a 10 MB cap on the base64-encoded data URL.
		// 7 MiB raw audio encodes to about 9.8 MB, leaving room for the prefix.
		maxAudioBytesDefault = 7 << 20
	}
	audioExtractor, err := backend.NewAudioTextExtractorForProtocol(audioMode, backend.AudioTextExtractorConfig{
		BaseURL: audioBaseURL, APIKey: audioAPIKey, Model: audioModel, Prompt: audioPrompt,
		ResponseFormat: strings.TrimSpace(os.Getenv("DRIVE9_AUDIO_EXTRACT_RESPONSE_FORMAT")), Timeout: audioTimeout,
	})
	if err != nil {
		return backend.AsyncAudioExtractOptions{}, fmt.Errorf("init audio extractor: %w", err)
	}

	async.Enabled = true
	async.Protocol = audioMode
	async.MaxAudioBytes = envInt64("DRIVE9_AUDIO_EXTRACT_MAX_BYTES", maxAudioBytesDefault)
	async.TaskTimeout = audioTimeout
	async.MaxExtractTextBytes = envInt("DRIVE9_AUDIO_EXTRACT_MAX_TEXT_BYTES", 8<<10)
	async.Extractor = audioExtractor
	logger.Info(context.Background(), "audio_extract_mode_configured",
		zap.String("mode", audioMode), zap.String("model", audioModel), zap.String("base_url", audioBaseURL))
	return async, nil
}

func buildVideoExtractOptionsFromEnv() (backend.AsyncVideoExtractOptions, error) {
	maxSourceDefault := backend.DefaultVideoExtractMaxSourceBytes
	if _, configured := os.LookupEnv("DRIVE9_VIDEO_EXTRACT_MAX_SOURCE_BYTES"); !configured {
		maxSourceDefault = envInt64("DRIVE9_VIDEO_EXTRACT_MAX_BYTES", maxSourceDefault)
	}
	maxSourceBytes := envInt64("DRIVE9_VIDEO_EXTRACT_MAX_SOURCE_BYTES", maxSourceDefault)
	maxProviderPayloadBytes := envInt64("DRIVE9_VIDEO_EXTRACT_MAX_PAYLOAD_BYTES", backend.DefaultVideoExtractMaxProviderPayloadBytes)
	videoTimeout := time.Duration(envInt("DRIVE9_VIDEO_EXTRACT_TIMEOUT_SECONDS", 300)) * time.Second
	if maxSourceBytes <= 0 {
		return backend.AsyncVideoExtractOptions{}, fmt.Errorf("DRIVE9_VIDEO_EXTRACT_MAX_SOURCE_BYTES must be a positive integer")
	}
	if maxProviderPayloadBytes <= 0 {
		return backend.AsyncVideoExtractOptions{}, fmt.Errorf("DRIVE9_VIDEO_EXTRACT_MAX_PAYLOAD_BYTES must be a positive integer")
	}
	if videoTimeout <= 0 {
		return backend.AsyncVideoExtractOptions{}, fmt.Errorf("DRIVE9_VIDEO_EXTRACT_TIMEOUT_SECONDS must be a positive integer")
	}
	if raw := strings.TrimSpace(os.Getenv("DRIVE9_VIDEO_EXTRACT_MAX_FRAMES")); raw != "" {
		logger.Warn(context.Background(), "video_extract_legacy_max_frames_ignored",
			zap.String("env", "DRIVE9_VIDEO_EXTRACT_MAX_FRAMES"))
	}
	// Single-config design: DRIVE9_VIDEO_EXTRACT_TENANT_ALLOWLIST controls
	// everything. Empty/unset = off. "*" = all tenants. Otherwise comma-
	// separated tenant IDs.
	raw := strings.TrimSpace(os.Getenv("DRIVE9_VIDEO_EXTRACT_TENANT_ALLOWLIST"))
	allTenants, tenantAllowlist, err := backend.ParseVideoExtractTenantAllowlist(raw)
	if err != nil {
		return backend.AsyncVideoExtractOptions{}, fmt.Errorf("DRIVE9_VIDEO_EXTRACT_TENANT_ALLOWLIST: %w", err)
	}
	// Parser returns nil allowlist + allTenants=false when input is empty
	// or contains only commas/whitespace → off, no runtime init needed.
	if !allTenants && tenantAllowlist == nil {
		return backend.AsyncVideoExtractOptions{MaxSourceBytes: maxSourceBytes, MaxProviderPayloadBytes: maxProviderPayloadBytes, TaskTimeout: videoTimeout}, nil
	}

	videoBaseURL := strings.TrimSpace(os.Getenv("DRIVE9_VIDEO_EXTRACT_API_BASE"))
	videoAPIKey := strings.TrimSpace(os.Getenv("DRIVE9_VIDEO_EXTRACT_API_KEY"))
	videoModel := strings.TrimSpace(os.Getenv("DRIVE9_VIDEO_EXTRACT_MODEL"))
	if videoBaseURL == "" || videoAPIKey == "" || videoModel == "" {
		return backend.AsyncVideoExtractOptions{}, fmt.Errorf("DRIVE9_VIDEO_EXTRACT_API_BASE, DRIVE9_VIDEO_EXTRACT_API_KEY and DRIVE9_VIDEO_EXTRACT_MODEL must be set when DRIVE9_VIDEO_EXTRACT_TENANT_ALLOWLIST is non-empty")
	}
	videoPrompt := strings.TrimSpace(os.Getenv("DRIVE9_VIDEO_EXTRACT_PROMPT"))
	ffmpegPath := strings.TrimSpace(os.Getenv("DRIVE9_FFMPEG_PATH"))
	extractor, err := backend.NewOpenAIVideoTextExtractor(backend.OpenAIVideoTextExtractorConfig{
		BaseURL:                 videoBaseURL,
		APIKey:                  videoAPIKey,
		Model:                   videoModel,
		Prompt:                  videoPrompt,
		Timeout:                 videoTimeout,
		FrameInterval:           envInt("DRIVE9_VIDEO_EXTRACT_FRAME_INTERVAL", 5),
		MaxProviderPayloadBytes: maxProviderPayloadBytes,
		FFmpegPath:              ffmpegPath,
	})
	if err != nil {
		return backend.AsyncVideoExtractOptions{}, fmt.Errorf("init video extractor: %w", err)
	}
	async := backend.AsyncVideoExtractOptions{
		Enabled:                 true,
		AllTenants:              allTenants,
		MaxSourceBytes:          maxSourceBytes,
		MaxProviderPayloadBytes: maxProviderPayloadBytes,
		MaxVideoLLMFiles:        envInt64("DRIVE9_VIDEO_EXTRACT_MAX_FILES", 30),
		TaskTimeout:             videoTimeout,
		MaxExtractTextBytes:     envInt("DRIVE9_VIDEO_EXTRACT_MAX_TEXT_BYTES", 32<<10),
		Extractor:               extractor,
		TenantAllowlist:         tenantAllowlist,
	}
	logger.Info(context.Background(), "video_extract_mode_configured",
		zap.String("model", videoModel), zap.String("base_url", videoBaseURL),
		zap.Bool("all_tenants", allTenants), zap.Int("tenant_allowlist_size", len(tenantAllowlist)))
	return async, nil
}

func tenantWorkerOptionsFromEnv() server.TenantWorkerOptions {
	// Warn about deprecated env vars that are no longer parsed.
	for _, deprecated := range []string{"DRIVE9_SEMANTIC_RECOVER_INTERVAL_MS", "DRIVE9_SEMANTIC_TENANT_LIMIT"} {
		if os.Getenv(deprecated) != "" {
			fmt.Fprintf(os.Stderr, "WARNING: %s is deprecated and no longer used; recovery is now kick-driven and tenant scan limit is removed\n", deprecated)
		}
	}
	return server.TenantWorkerOptions{
		Workers:              envInt("DRIVE9_SEMANTIC_WORKERS", 1),
		PollInterval:         time.Duration(envInt("DRIVE9_SEMANTIC_POLL_INTERVAL_MS", 200)) * time.Millisecond,
		LeaseDuration:        time.Duration(envInt("DRIVE9_SEMANTIC_LEASE_SECONDS", 30)) * time.Second,
		RetryBaseDelay:       time.Duration(envInt("DRIVE9_SEMANTIC_RETRY_BASE_MS", 200)) * time.Millisecond,
		RetryMaxDelay:        time.Duration(envInt("DRIVE9_SEMANTIC_RETRY_MAX_MS", 30000)) * time.Millisecond,
		PerTenantConcurrency: envInt("DRIVE9_SEMANTIC_PER_TENANT_CONCURRENCY", 1),
	}
}

type embeddingProcessConfig struct {
	Runtime  *backend.AppEmbeddingRuntime
	Provider *server.EmbeddingDefaultProvider
}

func buildEmbeddingProcessConfigFromEnv(ctx context.Context) (embeddingProcessConfig, error) {
	enabledRaw := strings.TrimSpace(os.Getenv("DRIVE9_EMBED_ENABLED"))
	enabled := false
	if enabledRaw != "" {
		var err error
		enabled, err = strconv.ParseBool(enabledRaw)
		if err != nil {
			return embeddingProcessConfig{}, fmt.Errorf("DRIVE9_EMBED_ENABLED must be true or false")
		}
	}
	if !enabled {
		return embeddingProcessConfig{}, nil
	}

	baseURL := strings.TrimSpace(os.Getenv("DRIVE9_EMBED_API_BASE"))
	apiKey := strings.TrimSpace(os.Getenv("DRIVE9_EMBED_API_KEY"))
	model := strings.TrimSpace(os.Getenv("DRIVE9_EMBED_MODEL"))
	if baseURL == "" || apiKey == "" || model == "" {
		return embeddingProcessConfig{}, fmt.Errorf("DRIVE9_EMBED_API_BASE, DRIVE9_EMBED_API_KEY and DRIVE9_EMBED_MODEL are required when DRIVE9_EMBED_ENABLED=true")
	}
	timeoutSeconds := 20
	if raw := strings.TrimSpace(os.Getenv("DRIVE9_EMBED_TIMEOUT_SECONDS")); raw != "" {
		parsed, err := strconv.Atoi(raw)
		if err != nil || parsed <= 0 {
			return embeddingProcessConfig{}, fmt.Errorf("DRIVE9_EMBED_TIMEOUT_SECONDS must be a positive integer")
		}
		timeoutSeconds = parsed
	}
	timeout := time.Duration(timeoutSeconds) * time.Second
	client, err := embedding.NewOpenAIClient(embedding.OpenAIClientConfig{
		BaseURL:    baseURL,
		APIKey:     apiKey,
		Model:      model,
		Dimensions: embedding.Dimensions,
		Timeout:    timeout,
	})
	if err != nil {
		return embeddingProcessConfig{}, fmt.Errorf("init embedding provider: %w", err)
	}
	probeCtx, cancel := context.WithTimeout(ctx, timeout)
	defer cancel()
	if _, err := client.EmbedText(probeCtx, "Drive9 embedding provider validation"); err != nil {
		return embeddingProcessConfig{}, embeddingStartupValidationError(err)
	}
	logger.Info(context.Background(), "semantic_embedding_mode_openai_compatible")
	keyHash := sha256.Sum256([]byte(apiKey))
	generation := embeddingFingerprint("generation-v1", baseURL, model, strconv.Itoa(embedding.Dimensions))
	configRevision := embeddingFingerprint("config-v1", baseURL, model, strconv.Itoa(embedding.Dimensions), hex.EncodeToString(keyHash[:]))
	return embeddingProcessConfig{
		Runtime: &backend.AppEmbeddingRuntime{
			Client: client, Generation: generation, ConfigRevision: configRevision,
		},
		Provider: &server.EmbeddingDefaultProvider{APIBase: baseURL, APIKey: apiKey, Model: model},
	}, nil
}

func embeddingStartupValidationError(err error) error {
	if category, status, safeReason, ok := embedding.ProviderFailure(err); ok {
		return fmt.Errorf("validate embedding provider: %s (category=%s, status=%d)", safeReason, category, status)
	}
	if errors.Is(err, context.DeadlineExceeded) {
		return fmt.Errorf("validate embedding provider: provider request timed out")
	}
	return fmt.Errorf("validate embedding provider: provider request failed")
}

func embeddingFingerprint(parts ...string) uint64 {
	h := sha256.New()
	for _, part := range parts {
		_, _ = h.Write([]byte(part))
		_, _ = h.Write([]byte{0})
	}
	sum := h.Sum(nil)
	value := binary.BigEndian.Uint64(sum[:8]) & uint64(math.MaxInt64)
	if value == 0 {
		return 1
	}
	return value
}

func envBool(key string, fallback bool) bool {
	raw := strings.TrimSpace(os.Getenv(key))
	if raw == "" {
		return fallback
	}
	switch strings.ToLower(raw) {
	case "1", "true", "yes", "on":
		return true
	case "0", "false", "no", "off":
		return false
	default:
		return fallback
	}
}

func dbHealthProbeOptionsFromEnv() metrics.DBHealthProbeOptions {
	return metrics.DBHealthProbeOptions{
		ProbeMeta: envBool("DRIVE9_DB_HEALTH_PROBE_META_ENABLED", true),
	}
}

func tenantBackendCacheMaxTenants(provider string) int {
	fallback := defaultTenantBackendCacheMaxTenants
	if provider == tenant.ProviderTiDBCloudNativeShared {
		fallback = defaultSharedBackendCacheMaxTenants
	}
	maxTenants := envInt("DRIVE9_POOL_MAX_TENANTS", fallback)
	if maxTenants <= 0 {
		return fallback
	}
	return maxTenants
}

func envInt(key string, fallback int) int {
	raw := strings.TrimSpace(os.Getenv(key))
	if raw == "" {
		return fallback
	}
	v, err := strconv.Atoi(raw)
	if err != nil {
		return fallback
	}
	return v
}

func envInt64(key string, fallback int64) int64 {
	raw := strings.TrimSpace(os.Getenv(key))
	if raw == "" {
		return fallback
	}
	v, err := strconv.ParseInt(raw, 10, 64)
	if err != nil {
		return fallback
	}
	return v
}

func tenantPoolMaxSizeFromEnv() (int, error) {
	raw := strings.TrimSpace(os.Getenv("DRIVE9_TENANT_POOL_MAX_SIZE"))
	if raw == "" {
		return server.DefaultTenantPoolMaxSize, nil
	}
	v, err := strconv.Atoi(raw)
	if err != nil || v <= 0 {
		return 0, fmt.Errorf("invalid DRIVE9_TENANT_POOL_MAX_SIZE=%q: must be a positive integer", raw)
	}
	return v, nil
}

func tenantPoolRefillFreeRatioFromEnv() (float64, error) {
	raw := strings.TrimSpace(os.Getenv("DRIVE9_TENANT_POOL_REFILL_FREE_RATIO"))
	if raw == "" {
		return server.DefaultTenantPoolRefillFreeRatio, nil
	}
	v, err := strconv.ParseFloat(raw, 64)
	if err != nil || v <= 0 || v > 1 || v != v {
		return 0, fmt.Errorf("invalid DRIVE9_TENANT_POOL_REFILL_FREE_RATIO=%q: must be a number in (0,1]", raw)
	}
	return v, nil
}

func sharedDBHardCapRatioFromEnv() (float64, error) {
	raw := strings.TrimSpace(os.Getenv("DRIVE9_TIDBCLOUD_NATIVE_SHARED_HARD_CAP_RATIO"))
	if raw == "" {
		return server.DefaultSharedDBHardCapRatio, nil
	}
	v, err := strconv.ParseFloat(raw, 64)
	if err != nil || v <= 1 || math.IsNaN(v) || math.IsInf(v, 0) {
		return 0, fmt.Errorf("invalid DRIVE9_TIDBCLOUD_NATIVE_SHARED_HARD_CAP_RATIO=%q: must be a finite number > 1", raw)
	}
	return v, nil
}

func sharedDBReopenRatioFromEnv() (float64, error) {
	raw := strings.TrimSpace(os.Getenv("DRIVE9_TIDBCLOUD_NATIVE_SHARED_REOPEN_RATIO"))
	if raw == "" {
		return server.DefaultSharedDBReopenRatio, nil
	}
	v, err := strconv.ParseFloat(raw, 64)
	if err != nil || v <= 0 || v >= 1 || math.IsNaN(v) || math.IsInf(v, 0) {
		return 0, fmt.Errorf("invalid DRIVE9_TIDBCLOUD_NATIVE_SHARED_REOPEN_RATIO=%q: must be a finite number in (0,1)", raw)
	}
	return v, nil
}

func sharedDBMaxTenantsFromEnv() (int, error) {
	raw := strings.TrimSpace(os.Getenv("DRIVE9_TIDBCLOUD_NATIVE_SHARED_MAX_TENANTS"))
	if raw == "" {
		return 100, nil
	}
	v, err := strconv.Atoi(raw)
	if err != nil || v <= 0 {
		return 0, fmt.Errorf("invalid DRIVE9_TIDBCLOUD_NATIVE_SHARED_MAX_TENANTS=%q: must be a positive integer", raw)
	}
	return v, nil
}

func sharedDBDefaultSpendingLimitFromEnv() (int64, error) {
	raw := strings.TrimSpace(os.Getenv("DRIVE9_TIDBCLOUD_NATIVE_DB_POOL_DEFAULT_SPENDING_LIMIT"))
	if raw == "" {
		return server.DefaultManagedSharedDBSpendingLimit, nil
	}
	v, err := strconv.ParseInt(raw, 10, 64)
	if err != nil || v <= 0 || v > int64(math.MaxInt32) {
		return 0, fmt.Errorf("invalid DRIVE9_TIDBCLOUD_NATIVE_DB_POOL_DEFAULT_SPENDING_LIMIT=%q: must be in (0,%d]", raw, int64(math.MaxInt32))
	}
	return v, nil
}

func tiDBCloudNonFreePlanCacheTTLFromEnv() (time.Duration, error) {
	return positiveDurationEnv("DRIVE9_TIDBCLOUD_NON_FREE_PLAN_CACHE_TTL", 15*time.Minute)
}

func tiDBCloudIAMCacheTTLFromEnv() (time.Duration, error) {
	return positiveDurationEnv("DRIVE9_TIDBCLOUD_IAM_CACHE_TTL", 30*time.Minute)
}

func tiDBCloudFreePlanLimitsFromEnv(maxUploadBytes int64) (server.TiDBCloudFreePlanLimits, error) {
	limits := server.DefaultTiDBCloudFreePlanLimits()
	var err error
	if limits.TenantCount, err = positiveIntEnv("DRIVE9_TIDBCLOUD_FREE_TENANT_COUNT", limits.TenantCount); err != nil {
		return server.TiDBCloudFreePlanLimits{}, err
	}
	if limits.MaxStorageBytes, err = positiveInt64Env("DRIVE9_TIDBCLOUD_FREE_MAX_STORAGE_BYTES", limits.MaxStorageBytes); err != nil {
		return server.TiDBCloudFreePlanLimits{}, err
	}
	const freeMaxFileSizeKey = "DRIVE9_TIDBCLOUD_FREE_MAX_FILE_SIZE_BYTES"
	if strings.TrimSpace(os.Getenv(freeMaxFileSizeKey)) == "" {
		if limits.MaxFileSizeBytes > maxUploadBytes {
			limits.MaxFileSizeBytes = maxUploadBytes
		}
	} else if limits.MaxFileSizeBytes, err = positiveInt64Env(freeMaxFileSizeKey, limits.MaxFileSizeBytes); err != nil {
		return server.TiDBCloudFreePlanLimits{}, err
	}
	if limits.MaxFileCount, err = positiveInt64Env("DRIVE9_TIDBCLOUD_FREE_MAX_FILE_COUNT", limits.MaxFileCount); err != nil {
		return server.TiDBCloudFreePlanLimits{}, err
	}
	if limits.MaxFileSizeBytes > maxUploadBytes {
		return server.TiDBCloudFreePlanLimits{}, fmt.Errorf("DRIVE9_TIDBCLOUD_FREE_MAX_FILE_SIZE_BYTES must not exceed DRIVE9_MAX_UPLOAD_BYTES")
	}
	return limits, nil
}

func maxTenantFileCountFromEnv() (int64, error) {
	const key = "DRIVE9_MAX_TENANT_FILE_COUNT"
	raw := strings.TrimSpace(os.Getenv(key))
	if raw == "" {
		return 0, nil
	}
	value, err := strconv.ParseInt(raw, 10, 64)
	if err != nil || value < 0 {
		return 0, fmt.Errorf("invalid %s=%q: must be a non-negative integer", key, raw)
	}
	return value, nil
}

func positiveIntEnv(key string, fallback int) (int, error) {
	raw := strings.TrimSpace(os.Getenv(key))
	if raw == "" {
		return fallback, nil
	}
	value, err := strconv.Atoi(raw)
	if err != nil || value <= 0 {
		return 0, fmt.Errorf("invalid %s=%q: must be a positive integer", key, raw)
	}
	return value, nil
}

func positiveInt64Env(key string, fallback int64) (int64, error) {
	raw := strings.TrimSpace(os.Getenv(key))
	if raw == "" {
		return fallback, nil
	}
	value, err := strconv.ParseInt(raw, 10, 64)
	if err != nil || value <= 0 {
		return 0, fmt.Errorf("invalid %s=%q: must be a positive integer", key, raw)
	}
	return value, nil
}

func parseDurationEnv(key string, fallback time.Duration) (time.Duration, string, error) {
	raw := strings.TrimSpace(os.Getenv(key))
	if raw == "" {
		return fallback, raw, nil
	}
	value, err := time.ParseDuration(raw)
	return value, raw, err
}

func positiveDurationEnv(key string, fallback time.Duration) (time.Duration, error) {
	value, raw, err := parseDurationEnv(key, fallback)
	if err != nil || value <= 0 {
		return 0, fmt.Errorf("invalid %s=%q: must be a positive duration", key, raw)
	}
	return value, nil
}

func envDuration(key string, fallback time.Duration) time.Duration {
	v, raw, err := parseDurationEnv(key, fallback)
	if err != nil {
		fmt.Fprintf(os.Stderr, "invalid %s=%q: %v; using %s\n", key, raw, err, fallback)
		return fallback
	}
	if v < 0 {
		fmt.Fprintf(os.Stderr, "invalid %s=%q: duration must be non-negative; using %s\n", key, raw, fallback)
		return fallback
	}
	return v
}

func deleteReconcileIntervalFromEnv() (time.Duration, bool) {
	if strings.TrimSpace(os.Getenv("DRIVE9_DELETE_RECONCILE_INTERVAL")) == "" {
		return server.DefaultDeleteReconcileInterval, false
	}
	return envDuration("DRIVE9_DELETE_RECONCILE_INTERVAL", server.DefaultDeleteReconcileInterval), true
}

// envDurationCompat reads a Go-duration env var under its canonical name. When
// the canonical name is unset but the deprecated legacy name is set, the legacy
// value is honored with a deprecation warning. The canonical name wins when
// both are set.
func envDurationCompat(canonical, deprecated string, fallback time.Duration) time.Duration {
	if strings.TrimSpace(os.Getenv(canonical)) != "" {
		return envDuration(canonical, fallback)
	}
	if strings.TrimSpace(os.Getenv(deprecated)) != "" {
		fmt.Fprintf(os.Stderr, "warning: %s is deprecated, use %s instead\n", deprecated, canonical)
		return envDuration(deprecated, fallback)
	}
	return fallback
}