run

var APICmd = &cli.Command{
	Name:  "api",
	Usage: "Run the singularity API",
	Flags: []cli.Flag{
		NoAutoMigrateFlag,
		&cli.StringFlag{
			Name:  "bind",
			Usage: "Bind address for the API server. The API is unauthenticated and operator-only -- bind beyond loopback only behind an authenticating proxy.",
			Value: "127.0.0.1:9090",
		},
	},
	Action: func(c *cli.Context) error {

		_, closer, err := openAndMigrate(c)
		if err != nil {
			return err
		}
		closer.Close()

		return api.Run(c)
	},
}

var ContentProviderCmd = &cli.Command{
	Name:  "content-provider",
	Usage: "Start a content provider that serves retrieval requests",
	Flags: []cli.Flag{
		NoAutoMigrateFlag,
		&cli.StringFlag{
			Category: "HTTP Retrieval",
			Name:     "http-bind",
			Usage:    "Address to bind the HTTP server to",
			Value:    "127.0.0.1:7777",
		},
		&cli.BoolFlag{
			Category: "HTTP Piece Retrieval",
			Name:     "enable-http-piece",
			Usage:    "Enable HTTP Piece retrieval",
			Aliases:  []string{"enable-http"},
			Value:    true,
		},
		&cli.BoolFlag{
			Category: "HTTP Piece Metadata Retrieval",
			Name:     "enable-http-piece-metadata",
			Usage:    "Enable HTTP Piece Metadata, this is to be used with the download server",
			Value:    true,
		},
		&cli.BoolFlag{
			Category: "HTTP IPFS Gateway",
			Name:     "enable-http-ipfs",
			Usage:    "Enable trustless IPFS gateway on /ipfs/",
			Value:    true,
		},
		&cli.IntFlag{
			Category: "HTTP IPFS Gateway",
			Name:     "ipfs-span-blocks",
			Usage:    "Blocks (~1MiB each) fetched per backend read",
			Value:    8,
		},
		&cli.IntFlag{
			Category: "HTTP IPFS Gateway",
			Name:     "ipfs-prefetch-spans",
			Usage:    "Spans prefetched ahead on sequential access",
			Value:    2,
		},
		&cli.IntFlag{
			Category: "HTTP IPFS Gateway",
			Name:     "ipfs-max-backend-reads",
			Usage:    "Max concurrent source storage reads across all requests",
			Value:    64,
		},
		&cli.IntFlag{
			Category:    "HTTP IPFS Gateway",
			Name:        "ipfs-cache-blocks",
			Usage:       "Block cache capacity in blocks (~1MiB each); 0 sizes it to max-backend-reads * span-blocks * (1 + prefetch-spans)",
			Value:       0,
			DefaultText: "derived",
		},
		&cli.DurationFlag{
			Category: "HTTP IPFS Gateway",
			Name:     "ipfs-read-timeout",
			Usage:    "Max duration of a single backend span read once it holds a connection slot",
			Value:    2 * time.Minute,
		},
	},
	Action: func(c *cli.Context) error {
		db, closer, err := openAndMigrate(c)
		if err != nil {
			return errors.WithStack(err)
		}
		defer closer.Close()

		config := contentprovider.Config{
			HTTP: contentprovider.HTTPConfig{
				EnablePiece:         c.Bool("enable-http-piece"),
				EnablePieceMetadata: c.Bool("enable-http-piece-metadata"),
				EnableIPFS:          c.Bool("enable-http-ipfs"),
				IPFSSpan: store.SpanConfig{
					SpanBlocks:      c.Int("ipfs-span-blocks"),
					PrefetchSpans:   c.Int("ipfs-prefetch-spans"),
					MaxBackendReads: c.Int("ipfs-max-backend-reads"),
					CacheBlocks:     c.Int("ipfs-cache-blocks"),
					ReadTimeout:     c.Duration("ipfs-read-timeout"),
				},
				Bind: c.String("http-bind"),
			},
		}

		s, err := contentprovider.NewService(db, config)
		if err != nil {
			return errors.WithStack(err)
		}
		return s.Start(c.Context)
	},
}

var DatasetWorkerCmd = &cli.Command{
	Name:  "dataset-worker",
	Usage: "Start a dataset preparation worker to process dataset scanning and preparation tasks",
	Flags: []cli.Flag{
		NoAutoMigrateFlag,
		&cli.IntFlag{
			Name:  "concurrency",
			Usage: "Number of concurrent workers to run",
			Value: 1,
		},
		&cli.BoolFlag{
			Name:  "enable-scan",
			Usage: "Enable scanning of datasets",
			Value: true,
		},
		&cli.BoolFlag{
			Name:  "enable-pack",
			Usage: "Enable packing of datasets that calculates CIDs and packs them into CAR files",
			Value: true,
		},
		&cli.BoolFlag{
			Name:  "enable-dag",
			Usage: "Enable dag generation of datasets that maintains the directory structure of datasets",
			Value: true,
		},
		&cli.BoolFlag{
			Name:  "enable-reaper",
			Usage: "Enable the orphan-record reaper. Exactly one dataset-worker process per deployment should enable this; running multiple reapers contends on the same rows and can livelock.",
			Value: true,
		},
		&cli.BoolFlag{
			Name:  "exit-on-complete",
			Usage: "Exit the worker when there is no more work to do",
			Value: false,
		},
		&cli.BoolFlag{
			Name:  "exit-on-error",
			Usage: "Exit the worker when there is any error",
			Value: false,
		},
		&cli.DurationFlag{
			Name:  "min-interval",
			Usage: "How often to check for new jobs (minimum)",
			Value: 5 * time.Second,
		},
		&cli.DurationFlag{
			Name:  "max-interval",
			Usage: "How often to check for new jobs (maximum)",
			Value: 160 * time.Second,
		},
	},
	Action: func(c *cli.Context) error {
		db, closer, err := openAndMigrate(c)
		if err != nil {
			return errors.WithStack(err)
		}
		defer closer.Close()
		worker := datasetworker.NewWorker(
			db,
			datasetworker.Config{
				Concurrency:    c.Int("concurrency"),
				EnableScan:     c.Bool("enable-scan"),
				EnablePack:     c.Bool("enable-pack"),
				EnableDag:      c.Bool("enable-dag"),
				EnableReaper:   c.Bool("enable-reaper"),
				ExitOnComplete: c.Bool("exit-on-complete"),
				ExitOnError:    c.Bool("exit-on-error"),
				MinInterval:    c.Duration("min-interval"),
				MaxInterval:    c.Duration("max-interval"),
			})
		err = worker.Run(c.Context)
		if err != nil {
			return errors.WithStack(err)
		}
		return nil
	},
}

var DealPusherCmd = &cli.Command{
	Name:  "deal-pusher",
	Usage: "Start a deal pusher that monitors deal schedules and pushes deals to storage providers",
	Flags: []cli.Flag{
		NoAutoMigrateFlag,
		&cli.UintFlag{
			Name:    "deal-attempts",
			Usage:   "Number of times to attempt a deal before giving up",
			Aliases: []string{"d"},
			Value:   3,
		},
		&cli.UintFlag{
			Name:        "max-replication-factor",
			Usage:       "Max number of replicas for each individual PieceCID across all clients and providers",
			Aliases:     []string{"M"},
			DefaultText: "Unlimited",
		},
		&cli.IntFlag{
			Name:  "pdp-batch-size",
			Usage: "Number of pieces to include in each /pdp/piece/pull request",
			Value: 128,
		},
		&cli.IntFlag{
			Name:  "pdp-max-pieces-per-proofset",
			Usage: "Maximum pieces per proof set before starting a new one",
			Value: 1024,
		},
		&cli.DurationFlag{
			Name:  "pdp-pull-timeout",
			Usage: "How long to wait for Curio to finish pulling a batch (per request)",
			Value: 5 * time.Minute,
		},
		&cli.StringFlag{
			Name:    "pdp-source-url-base",
			Usage:   "HTTPS base URL where Curio fetches pieces from; sourceUrl is built as <base>/piece/<pieceCid>",
			EnvVars: []string{"PDP_SOURCE_URL_BASE"},
		},
		&cli.StringFlag{
			Name:    "pdp-record-keeper",
			Usage:   "FWSS contract address (recordKeeper). Defaults to the network default from go-synapse.",
			EnvVars: []string{"PDP_RECORD_KEEPER"},
		},
		&cli.StringFlag{
			Name:    "eth-rpc",
			Usage:   "Ethereum RPC endpoint for FEVM (required to execute PDP and DDO schedules on-chain)",
			EnvVars: []string{"ETH_RPC_URL"},
		},
		&cli.StringFlag{
			Name:    "ddo-contract",
			Usage:   "DDO Diamond proxy contract address",
			EnvVars: []string{"DDO_CONTRACT_ADDRESS"},
		},
		&cli.StringFlag{
			Name:    "ddo-payments-contract",
			Usage:   "DDO Payments proxy contract address",
			EnvVars: []string{"DDO_PAYMENTS_CONTRACT_ADDRESS"},
		},
		&cli.StringFlag{
			Name:    "ddo-payment-token",
			Usage:   "ERC20 payment token address (e.g. USDFC)",
			EnvVars: []string{"DDO_PAYMENT_TOKEN"},
		},
		&cli.IntFlag{
			Name:  "ddo-batch-size",
			Usage: "Number of pieces per DDO allocation transaction",
			Value: 10,
		},
		&cli.Uint64Flag{
			Name:  "ddo-confirmation-depth",
			Usage: "Number of block confirmations required for DDO transactions",
			Value: 5,
		},
		&cli.DurationFlag{
			Name:  "ddo-poll-interval",
			Usage: "Polling interval for DDO transaction confirmation checks",
			Value: 30 * time.Second,
		},
		&cli.Int64Flag{
			Name:  "ddo-term-min",
			Usage: "Minimum term in epochs (~6 months default)",
			Value: 518400,
		},
		&cli.Int64Flag{
			Name:  "ddo-term-max",
			Usage: "Maximum term in epochs (~5 years default)",
			Value: 5256000,
		},
		&cli.Int64Flag{
			Name:  "ddo-expiration-offset",
			Usage: "Expiration offset in epochs",
			Value: 172800,
		},
	},
	Action: func(c *cli.Context) error {
		db, closer, err := openAndMigrate(c)
		if err != nil {
			return errors.WithStack(err)
		}
		defer closer.Close()
		lotusAPI := c.String("lotus-api")
		lotusToken := c.String("lotus-token")
		err = epochutil.Initialize(c.Context, lotusAPI, lotusToken)
		if err != nil {
			return errors.WithStack(err)
		}

		pdpCfg := dealpusher.PDPSchedulingConfig{
			BatchSize:            c.Int("pdp-batch-size"),
			MaxPiecesPerProofSet: c.Int("pdp-max-pieces-per-proofset"),
			PullTimeout:          c.Duration("pdp-pull-timeout"),
		}
		if err := pdpCfg.Validate(); err != nil {
			return errors.WithStack(err)
		}

		opts := []dealpusher.Option{
			dealpusher.WithPDPSchedulingConfig(pdpCfg),
		}
		if rpcURL := c.String("eth-rpc"); rpcURL != "" {
			adapterCfg := dealpusher.OnChainPDPConfig{
				DB:            db,
				RPCURL:        rpcURL,
				SourceURLBase: c.String("pdp-source-url-base"),
				RecordKeeper:  c.String("pdp-record-keeper"),
			}
			pdpAdapter, err := dealpusher.NewOnChainPDP(c.Context, adapterCfg)
			if err != nil {
				return errors.Wrap(err, "failed to initialize PDP on-chain adapter")
			}
			defer pdpAdapter.Close()

			opts = append(opts, dealpusher.WithPDPProofSetManager(pdpAdapter))
		}

		if ddoContract := c.String("ddo-contract"); ddoContract != "" {
			ddoCfg := dealpusher.DDOSchedulingConfig{
				BatchSize:         c.Int("ddo-batch-size"),
				ConfirmationDepth: c.Uint64("ddo-confirmation-depth"),
				PollingInterval:   c.Duration("ddo-poll-interval"),
				TermMin:           c.Int64("ddo-term-min"),
				TermMax:           c.Int64("ddo-term-max"),
				ExpirationOffset:  c.Int64("ddo-expiration-offset"),
			}
			if err := ddoCfg.Validate(); err != nil {
				return errors.WithStack(err)
			}
			opts = append(opts, dealpusher.WithDDOSchedulingConfig(ddoCfg))

			rpcURL := c.String("eth-rpc")
			if rpcURL == "" {
				return errors.New("--eth-rpc is required when --ddo-contract is set")
			}
			ddoAdapter, err := dealpusher.NewOnChainDDO(c.Context,
				rpcURL,
				ddoContract,
				c.String("ddo-payments-contract"),
				c.String("ddo-payment-token"),
			)
			if err != nil {
				return errors.Wrap(err, "failed to initialize DDO on-chain adapter")
			}
			defer ddoAdapter.Close()

			opts = append(opts, dealpusher.WithDDODealManager(ddoAdapter))
		}

		dm, err := dealpusher.NewDealPusher(
			db,
			c.String("lotus-api"),
			c.String("lotus-token"),
			c.Uint("deal-attempts"),
			c.Uint("max-replication-factor"),
			opts...,
		)
		if err != nil {
			return errors.WithStack(err)
		}
		return service.StartServers(c.Context, dealpusher.Logger, dm)
	},
}

var DealTrackerCmd = &cli.Command{
	Name:  "deal-tracker",
	Usage: "Start a deal tracker that tracks the deal for all relevant wallets",
	Flags: []cli.Flag{
		NoAutoMigrateFlag,
		&cli.StringFlag{
			Name:    "market-deal-url",
			Usage:   "The URL for ZST compressed state market deals json. Set to empty to use Lotus API.",
			Aliases: []string{"m"},
			EnvVars: []string{"MARKET_DEAL_URL"},
			Value:   "https://marketdeals.s3.amazonaws.com/StateMarketDeals.json.zst",
		},
		&cli.DurationFlag{
			Name:    "interval",
			Usage:   "How often to check for new deals",
			Aliases: []string{"i"},
			Value:   1 * time.Hour,
		},
		&cli.BoolFlag{
			Name:  "once",
			Usage: "Run once and exit",
			Value: false,
		},
		&cli.StringFlag{
			Name:    "eth-rpc",
			Usage:   "Ethereum RPC endpoint for FEVM (required for DDO allocation tracking)",
			EnvVars: []string{"ETH_RPC_URL"},
		},
		&cli.StringFlag{
			Name:    "ddo-contract",
			Usage:   "DDO Diamond proxy contract address",
			EnvVars: []string{"DDO_CONTRACT_ADDRESS"},
		},
	},
	Action: func(c *cli.Context) error {
		db, closer, err := openAndMigrate(c)
		if err != nil {
			return errors.WithStack(err)
		}
		defer closer.Close()

		lotusAPI := c.String("lotus-api")
		lotusToken := c.String("lotus-token")
		err = epochutil.Initialize(c.Context, lotusAPI, lotusToken)
		if err != nil {
			return errors.WithStack(err)
		}

		var opts []dealtracker.DealTrackerOption
		if ddoContract := c.String("ddo-contract"); ddoContract != "" {
			rpcURL := c.String("eth-rpc")
			if rpcURL == "" {
				return errors.New("--eth-rpc is required when --ddo-contract is set")
			}
			ddoClient, err := dealtracker.NewDDOTrackingClient(rpcURL, ddoContract)
			if err != nil {
				return errors.Wrap(err, "failed to initialize DDO tracking client")
			}
			defer ddoClient.Close()
			opts = append(opts, dealtracker.WithDDOAllocationTracker(ddoClient))
		}

		tracker := dealtracker.NewDealTracker(db,
			c.Duration("interval"),
			c.String("market-deal-url"),
			c.String("lotus-api"),
			c.String("lotus-token"),
			c.Bool("once"),
			opts...,
		)

		return service.StartServers(c.Context, dealtracker.Logger, &tracker)
	},
}

var DownloadServerCmd = &cli.Command{
	Name:        "download-server",
	Usage:       "An HTTP server connecting to remote metadata API to offer CAR file downloads",
	Description: "Example Usage:\n  singularity run download-server --metadata-api \"http://remote-metadata-api:7777\" --bind \"127.0.0.1:8888\"",
	Flags: func() []cli.Flag {
		flags := []cli.Flag{
			&cli.StringFlag{
				Name:     "metadata-api",
				Aliases:  []string{"api"},
				Usage:    "URL of the metadata API",
				Value:    "http://127.0.0.1:7777",
				Category: "General Config",
			},
			&cli.StringFlag{
				Name:     "bind",
				Usage:    "Address to bind the HTTP server to",
				Value:    "127.0.0.1:8888",
				Category: "General Config",
			},
		}

		flags = append(flags, storage.HTTPClientConfigFlagsForUpdate...)
		flags = append(flags, storage.CommonConfigFlags...)

		keys := make(map[string]struct{})
		for _, backend := range storagesystem.Backends {
			for _, providerOptions := range backend.ProviderOptions {
				for _, option := range providerOptions.Options {
					if !model.IsSecretConfigName(option.Name) {
						continue
					}
					flag := option.ToCLIFlag(backend.Prefix+"-", false, backend.Description)
					if _, ok := keys[flag.Names()[0]]; ok {
						continue
					}
					keys[flag.Names()[0]] = struct{}{}
					flags = append(flags, flag)
				}
			}
		}
		return flags
	}(),
	Action: func(c *cli.Context) error {
		api := c.String("metadata-api")
		bind := c.String("bind")
		config := map[string]string{}
		for _, key := range c.LocalFlagNames() {
			if c.IsSet(key) {
				if slices.Contains([]string{"api", "metadata-api", "bind"}, key) {
					continue
				}
				value := c.String(key)
				config[key] = value
			}
		}
		clientConfig, err := storage.GetClientConfigForUpdate(c)
		if err != nil {
			return errors.WithStack(err)
		}

		server := downloadserver.NewDownloadServer(bind, api, config, *clientConfig)
		return service.StartServers(c.Context, downloadserver.Logger, server)
	},
}

var NoAutoMigrateFlag = &cli.BoolFlag{
	Name:  "no-automigrate",
	Usage: "skip automatic database migration and correctness checks on startup; only use if you run 'admin init' on every upgrade or manually before starting daemons",
}

var PDPTrackerCmd = &cli.Command{
	Name:  "pdp-tracker",
	Usage: "Track PDP deals via Shovel event indexing (requires PostgreSQL)",
	Flags: []cli.Flag{
		NoAutoMigrateFlag,
		&cli.StringFlag{
			Name:    "eth-rpc",
			Usage:   "Ethereum RPC endpoint for FEVM",
			Value:   "https://api.node.glif.io/rpc/v1",
			EnvVars: []string{"ETH_RPC_URL"},
		},
		&cli.DurationFlag{
			Name:  "pdp-poll-interval",
			Usage: "How often to check for new events in Shovel tables",
			Value: 30 * time.Second,
		},
		&cli.BoolFlag{
			Name:  "full-sync",
			Usage: "Re-index events from contract deployment by resetting the Shovel cursor. Derived PDP state (proof sets, deals) is preserved and updated via upserts. Requires an archival RPC node. Involves one RPC call per historical proof set.",
		},
	},
	Action: func(c *cli.Context) error {
		rpcURL := c.String("eth-rpc")
		connStr := c.String("database-connection-string")
		if !strings.HasPrefix(connStr, "postgres:") && !strings.HasPrefix(connStr, "postgresql:") {
			return errors.New("PDP tracking requires PostgreSQL (Shovel is Postgres-only)")
		}

		db, closer, err := openAndMigrate(c)
		if err != nil {
			return errors.WithStack(err)
		}
		defer closer.Close()

		ethClient, err := ethclient.DialContext(c.Context, rpcURL)
		if err != nil {
			return errors.Wrap(err, "failed to connect to RPC")
		}
		network, chainID, err := synapse.DetectNetwork(c.Context, ethClient)
		ethClient.Close()
		if err != nil {
			return errors.Wrap(err, "failed to detect network")
		}

		contractAddr := constants.GetPDPVerifierAddress(network)
		if contractAddr == (common.Address{}) {
			return fmt.Errorf("no PDPVerifier contract for network %s", network)
		}

		pdptracker.Logger.Infow("detected PDP network",
			"network", network,
			"chainId", chainID,
			"contract", contractAddr.Hex(),
		)

		indexer, err := pdptracker.NewPDPIndexer(c.Context, connStr, rpcURL, uint64(chainID), contractAddr, c.Bool("full-sync"))
		if err != nil {
			return errors.Wrap(err, "failed to create PDP indexer")
		}

		rpcClient, err := pdptracker.NewPDPClient(c.Context, rpcURL, contractAddr)
		if err != nil {
			return errors.Wrap(err, "failed to create PDP RPC client")
		}
		defer rpcClient.Close()

		cfg := pdptracker.PDPConfig{
			PollingInterval: c.Duration("pdp-poll-interval"),
		}
		if err := cfg.Validate(); err != nil {
			return err
		}

		tracker := pdptracker.NewPDPTracker(db, cfg, rpcClient, false)

		return service.StartServers(c.Context, pdptracker.Logger, indexer, &tracker)
	},
}