streamplace / pkg / config / config.go

package config

import (
	"context"
	"crypto/rsa"
	"crypto/x509"
	"encoding/json"
	"encoding/pem"
	"errors"
	"flag"
	"fmt"
	"io"
	"net"
	"os"
	"path/filepath"
	"runtime"
	"strconv"
	"strings"
	"time"

	"math/rand/v2"

	"github.com/lestrrat-go/jwx/v2/jwk"
	"github.com/livepeer/go-livepeer/cmd/livepeer/starter"
	"github.com/lmittmann/tint"
	slogGorm "github.com/orandin/slog-gorm"
	"github.com/peterbourgon/ff/v3"
	"stream.place/streamplace/pkg/aqtime"
	"stream.place/streamplace/pkg/constants"
	"stream.place/streamplace/pkg/crypto/aqpub"
	"stream.place/streamplace/pkg/integrations/discord/discordtypes"
	"stream.place/streamplace/pkg/log"
)

const SPDataDir = "$SP_DATA_DIR"
const SegmentsDir = "segments"

type BuildFlags struct {
	Version   string
	BuildTime int64
	UUID      string
}

func (b BuildFlags) BuildTimeStr() string {
	ts := time.Unix(b.BuildTime, 0)
	return ts.UTC().Format(time.RFC3339)
}

func (b BuildFlags) BuildTimeStrExpo() string {
	ts := time.Unix(b.BuildTime, 0)
	return ts.UTC().Format("2006-01-02T15:04:05.000Z")
}

type CLI struct {
	AdminAccount               string
	Build                      *BuildFlags
	DataDir                    string
	DBURL                      string
	EthAccountAddr             string
	EthKeystorePath            string
	EthPassword                string
	FirebaseServiceAccount     string
	FirebaseServiceAccountFile string
	GitLabURL                  string
	HTTPAddr                   string
	HTTPInternalAddr           string
	HTTPSAddr                  string
	RtmpsAddr                  string
	Secure                     bool
	NoMist                     bool
	MistAdminPort              int
	MistHTTPPort               int
	MistRTMPPort               int
	SigningKeyPath             string
	TAURL                      string
	TLSCertPath                string
	TLSKeyPath                 string
	PKCS11ModulePath           string
	PKCS11Pin                  string
	PKCS11TokenSlot            string
	PKCS11TokenLabel           string
	PKCS11TokenSerial          string
	PKCS11KeypairLabel         string
	PKCS11KeypairID            string
	StreamerName               string
	RelayHost                  string
	Debug                      map[string]map[string]int
	AllowedStreams             []string
	WideOpen                   bool
	Peers                      []string
	Redirects                  []string
	TestStream                 bool
	FrontendProxy              string
	PublicOAuth                bool
	AppBundleID                string
	NoFirehose                 bool
	PrintChat                  bool
	Color                      string
	LivepeerGatewayURL         string
	LivepeerGateway            bool
	WHIPTest                   string
	Thumbnail                  bool
	SmearAudio                 bool
	ExternalSigning            bool
	RTMPServerAddon            string
	TracingEndpoint            string
	BroadcasterHost            string
	XXDeprecatedPublicHost     string
	ServerHost                 string
	RateLimitPerSecond         int
	RateLimitBurst             int
	RateLimitWebsocket         int
	JWK                        jwk.Key
	AccessJWK                  jwk.Key
	dataDirFlags               []*string
	DiscordWebhooks            []*discordtypes.Webhook
	NewWebRTCPlayback          bool
	AppleTeamID                string
	AndroidCertFingerprint     string
	Labelers                   []string
	AtprotoDID                 string
	LivepeerHelp               bool
	PLCURL                     string
	ContentFilters             *ContentFilters
	SQLLogging                 bool
	SentryDSN                  string
	LivepeerDebug              bool
	Tickets                    []string
	IrohTopic                  string
	DID                        string
	DisableIrohRelay           bool
	DevAccountCreds            map[string]string
	StreamSessionTimeout       time.Duration
	Replicators                []string
	WebsocketURL               string
	BehindHTTPSProxy           bool
	SegmentDebugDir            string
	Syndicate                  []string
}

// ContentFilters represents the content filtering configuration
type ContentFilters struct {
	ContentWarnings struct {
		Enabled         bool     `json:"enabled"`
		BlockedWarnings []string `json:"blocked_warnings"`
	} `json:"content_warnings"`
	DistributionPolicy struct {
		Enabled bool `json:"enabled"`
	} `json:"distribution_policy"`
}

const (
	ReplicatorWebsocket string = "websocket"
	ReplicatorIroh      string = "iroh"
)

func (cli *CLI) NewFlagSet(name string) *flag.FlagSet {
	fs := flag.NewFlagSet("streamplace", flag.ExitOnError)
	fs.StringVar(&cli.DataDir, "data-dir", DefaultDataDir(), "directory for keeping all streamplace data")
	fs.StringVar(&cli.HTTPAddr, "http-addr", ":38080", "Public HTTP address")
	fs.StringVar(&cli.HTTPInternalAddr, "http-internal-addr", "127.0.0.1:39090", "Private, admin-only HTTP address")
	fs.StringVar(&cli.HTTPSAddr, "https-addr", ":38443", "Public HTTPS address")
	fs.BoolVar(&cli.Secure, "secure", false, "Run with HTTPS. Required for WebRTC output")
	cli.DataDirFlag(fs, &cli.TLSCertPath, "tls-cert", filepath.Join("tls", "tls.crt"), "Path to TLS certificate")
	cli.DataDirFlag(fs, &cli.TLSKeyPath, "tls-key", filepath.Join("tls", "tls.key"), "Path to TLS key")
	fs.StringVar(&cli.SigningKeyPath, "signing-key", "", "Path to signing key for pushing OTA updates to the app")
	fs.StringVar(&cli.DBURL, "db-url", "sqlite://$SP_DATA_DIR/state.sqlite", "URL of the database to use for storing private streamplace state")
	cli.dataDirFlags = append(cli.dataDirFlags, &cli.DBURL)
	fs.StringVar(&cli.AdminAccount, "admin-account", "", "ethereum account that administrates this streamplace node")
	fs.StringVar(&cli.FirebaseServiceAccount, "firebase-service-account", "", "Base64-encoded JSON string of a firebase service account key")
	fs.StringVar(&cli.FirebaseServiceAccountFile, "firebase-service-account-file", "", "Path to a JSON file containing a firebase service account key")
	fs.StringVar(&cli.GitLabURL, "gitlab-url", "https://git.stream.place/api/v4/projects/1", "gitlab url for generating download links")
	cli.DataDirFlag(fs, &cli.EthKeystorePath, "eth-keystore-path", "keystore", "path to ethereum keystore")
	fs.StringVar(&cli.EthAccountAddr, "eth-account-addr", "", "ethereum account address to use (if keystore contains more than one)")
	fs.StringVar(&cli.EthPassword, "eth-password", "", "password for encrypting keystore")
	fs.StringVar(&cli.TAURL, "ta-url", "http://timestamp.digicert.com", "timestamp authority server for signing")
	fs.StringVar(&cli.PKCS11ModulePath, "pkcs11-module-path", "", "path to a PKCS11 module for HSM signing, for example /usr/lib/x86_64-linux-gnu/opensc-pkcs11.so")
	fs.StringVar(&cli.PKCS11Pin, "pkcs11-pin", "", "PIN for logging into PKCS11 token. if not provided, will be prompted interactively")
	fs.StringVar(&cli.PKCS11TokenSlot, "pkcs11-token-slot", "", "slot number of PKCS11 token (only use one of slot, label, or serial)")
	fs.StringVar(&cli.PKCS11TokenLabel, "pkcs11-token-label", "", "label of PKCS11 token (only use one of slot, label, or serial)")
	fs.StringVar(&cli.PKCS11TokenSerial, "pkcs11-token-serial", "", "serial number of PKCS11 token (only use one of slot, label, or serial)")
	fs.StringVar(&cli.PKCS11KeypairLabel, "pkcs11-keypair-label", "", "label of signing keypair on PKCS11 token")
	fs.StringVar(&cli.PKCS11KeypairID, "pkcs11-keypair-id", "", "id of signing keypair on PKCS11 token")
	fs.StringVar(&cli.AppBundleID, "app-bundle-id", "", "bundle id of an app that we facilitate oauth login for")
	fs.StringVar(&cli.StreamerName, "streamer-name", "", "name of the person streaming from this streamplace node")
	fs.StringVar(&cli.FrontendProxy, "dev-frontend-proxy", "", "(FOR DEVELOPMENT ONLY) proxy frontend requests to this address instead of using the bundled frontend")
	fs.BoolVar(&cli.PublicOAuth, "dev-public-oauth", false, "(FOR DEVELOPMENT ONLY) enable public oauth login for http://127.0.0.1 development")
	fs.StringVar(&cli.LivepeerGatewayURL, "livepeer-gateway-url", "", "URL of the Livepeer Gateway to use for transcoding")
	fs.BoolVar(&cli.LivepeerGateway, "livepeer-gateway", false, "enable embedded Livepeer Gateway")
	fs.BoolVar(&cli.WideOpen, "wide-open", false, "allow ALL streams to be uploaded to this node (not recommended for production)")
	cli.StringSliceFlag(fs, &cli.AllowedStreams, "allowed-streams", []string{}, "if set, only allow these addresses or atproto DIDs to upload to this node")
	cli.StringSliceFlag(fs, &cli.Peers, "peers", []string{}, "other streamplace nodes to replicate to")
	cli.StringSliceFlag(fs, &cli.Redirects, "redirects", []string{}, "http 302s /path/one:/path/two,/path/three:/path/four")
	cli.DebugFlag(fs, &cli.Debug, "debug", "", "modified log verbosity for specific functions or files in form func=ToHLS:3,file=gstreamer.go:4")
	fs.BoolVar(&cli.TestStream, "test-stream", false, "run a built-in test stream on boot")
	fs.BoolVar(&cli.NoFirehose, "no-firehose", false, "disable the bluesky firehose")
	fs.BoolVar(&cli.PrintChat, "print-chat", false, "print chat messages to stdout")
	fs.StringVar(&cli.WHIPTest, "whip-test", "", "run a WHIP self-test with the given parameters")
	fs.StringVar(&cli.RelayHost, "relay-host", "wss://bsky.network", "websocket url for relay firehose")
	fs.StringVar(&cli.Color, "color", "", "'true' to enable colorized logging, 'false' to disable")
	fs.StringVar(&cli.BroadcasterHost, "broadcaster-host", "", "public host for the broadcaster group that this node is a part of (excluding https:// e.g. stream.place)")
	fs.StringVar(&cli.XXDeprecatedPublicHost, "public-host", "", "deprecated, use broadcaster-host or server-host instead as appropriate")
	fs.StringVar(&cli.ServerHost, "server-host", "", "public host for this particular physical streamplace node. defaults to broadcaster-host and only must be set for multi-node broadcasters")
	fs.BoolVar(&cli.Thumbnail, "thumbnail", true, "enable thumbnail generation")
	fs.BoolVar(&cli.SmearAudio, "smear-audio", false, "enable audio smearing to create 'perfect' segment timestamps")
	fs.StringVar(&cli.TracingEndpoint, "tracing-endpoint", "", "gRPC endpoint to send traces to")
	fs.IntVar(&cli.RateLimitPerSecond, "rate-limit-per-second", 0, "rate limit for requests per second per ip")
	fs.IntVar(&cli.RateLimitBurst, "rate-limit-burst", 0, "rate limit burst for requests per ip")
	fs.IntVar(&cli.RateLimitWebsocket, "rate-limit-websocket", 10, "number of concurrent websocket connections allowed per ip")
	fs.StringVar(&cli.RTMPServerAddon, "rtmp-server-addon", "", "address of external RTMP server to forward streams to")
	fs.StringVar(&cli.RtmpsAddr, "rtmps-addr", ":1935", "address to listen for RTMPS connections")
	cli.JSONFlag(fs, &cli.DiscordWebhooks, "discord-webhooks", "[]", "JSON array of Discord webhooks to send notifications to")
	fs.BoolVar(&cli.NewWebRTCPlayback, "new-webrtc-playback", true, "enable new webrtc playback")
	fs.StringVar(&cli.AppleTeamID, "apple-team-id", "", "apple team id for deep linking")
	fs.StringVar(&cli.AndroidCertFingerprint, "android-cert-fingerprint", "", "android cert fingerprint for deep linking")
	cli.StringSliceFlag(fs, &cli.Labelers, "labelers", []string{}, "did of labelers that this instance should subscribe to")
	fs.StringVar(&cli.AtprotoDID, "atproto-did", "", "atproto did to respond to on /.well-known/atproto-did (default did:web:PUBLIC_HOST)")
	cli.JSONFlag(fs, &cli.ContentFilters, "content-filters", "{}", "JSON content filtering rules")
	fs.BoolVar(&cli.LivepeerHelp, "livepeer-help", false, "print help for livepeer flags and exit")
	fs.StringVar(&cli.PLCURL, "plc-url", "https://plc.directory", "url of the plc directory")
	fs.BoolVar(&cli.SQLLogging, "sql-logging", false, "enable sql logging")
	fs.StringVar(&cli.SentryDSN, "sentry-dsn", "", "sentry dsn for error reporting")
	fs.BoolVar(&cli.LivepeerDebug, "livepeer-debug", false, "log livepeer segments to $SP_DATA_DIR/livepeer-debug")
	fs.StringVar(&cli.SegmentDebugDir, "segment-debug-dir", "", "directory to log segment validation to")
	cli.StringSliceFlag(fs, &cli.Tickets, "tickets", []string{}, "tickets to join the swarm with")
	fs.StringVar(&cli.IrohTopic, "iroh-topic", "", "topic to use for the iroh swarm (must be 32 bytes in hex)")
	fs.BoolVar(&cli.DisableIrohRelay, "disable-iroh-relay", false, "disable the iroh relay")
	cli.KVSliceFlag(fs, &cli.DevAccountCreds, "dev-account-creds", "", "(FOR DEVELOPMENT ONLY) did=password pairs for logging into test accounts without oauth")
	fs.DurationVar(&cli.StreamSessionTimeout, "stream-session-timeout", 60*time.Second, "how long to wait before considering a stream inactive on this node?")
	cli.StringSliceFlag(fs, &cli.Replicators, "replicators", []string{ReplicatorWebsocket}, "list of replication protocols to use (http, iroh)")
	fs.StringVar(&cli.WebsocketURL, "websocket-url", "", "override the websocket (ws:// or wss://) url to use for replication (normally not necessary, used for testing)")
	fs.BoolVar(&cli.BehindHTTPSProxy, "behind-https-proxy", false, "set to true if this node is behind an https proxy and we should report https URLs even though the node isn't serving HTTPS")
	cli.StringSliceFlag(fs, &cli.Syndicate, "syndicate", []string{}, "list of DIDs that we should rebroadcast ('*' for everybody)")

	fs.Bool("external-signing", true, "DEPRECATED, does nothing.")
	fs.Bool("insecure", false, "DEPRECATED, does nothing.")

	lpFlags := flag.NewFlagSet("livepeer", flag.ContinueOnError)
	_ = starter.NewLivepeerConfig(lpFlags)
	lpFlags.VisitAll(func(f *flag.Flag) {
		adapted := LivepeerFlags.CamelToSnake[f.Name]
		fs.Var(f.Value, fmt.Sprintf("livepeer.%s", adapted), f.Usage)
	})

	if runtime.GOOS == "linux" {
		fs.BoolVar(&cli.NoMist, "no-mist", true, "Disable MistServer")
		fs.IntVar(&cli.MistAdminPort, "mist-admin-port", 14242, "MistServer admin port (internal use only)")
		fs.IntVar(&cli.MistRTMPPort, "mist-rtmp-port", 11935, "MistServer RTMP port (internal use only)")
		fs.IntVar(&cli.MistHTTPPort, "mist-http-port", 18080, "MistServer HTTP port (internal use only)")
	}
	return fs
}

var StreamplaceSchemePrefix = "streamplace://"

func (cli *CLI) OwnPublicURL() string {
	//  No errors because we know it's valid from AddrFlag
	host, port, _ := net.SplitHostPort(cli.HTTPAddr)

	ip := net.ParseIP(host)
	if host == "" || ip.IsUnspecified() {
		host = "127.0.0.1"
	}
	addr := net.JoinHostPort(host, port)
	return fmt.Sprintf("http://%s", addr)
}

func (cli *CLI) OwnInternalURL() string {
	//  No errors because we know it's valid from AddrFlag
	host, port, _ := net.SplitHostPort(cli.HTTPInternalAddr)

	ip := net.ParseIP(host)
	if ip.IsUnspecified() {
		host = "127.0.0.1"
	}
	addr := net.JoinHostPort(host, port)
	return fmt.Sprintf("http://%s", addr)
}

func (cli *CLI) ParseSigningKey() (*rsa.PrivateKey, error) {
	bs, err := os.ReadFile(cli.SigningKeyPath)
	if err != nil {
		return nil, err
	}
	block, _ := pem.Decode(bs)
	if block == nil {
		return nil, fmt.Errorf("no RSA key found in signing key")
	}
	key, err := x509.ParsePKCS1PrivateKey(block.Bytes)
	if err != nil {
		return nil, err
	}
	return key, nil
}

func RandomTrailer(length int) string {
	const charset = "abcdefghijklmnopqrstuvwxyz0123456789"

	res := make([]byte, length)
	for i := 0; i < length; i++ {
		res[i] = charset[rand.IntN(len(charset))]
	}
	return string(res)
}

func DefaultDataDir() string {
	home, err := os.UserHomeDir()
	if err != nil {
		// not fatal unless the user doesn't set one later
		return ""
	}
	return filepath.Join(home, ".streamplace")
}

var GormLogger = slogGorm.New(
	slogGorm.WithHandler(tint.NewHandler(os.Stderr, &tint.Options{
		TimeFormat: time.RFC3339,
	})),
	slogGorm.WithTraceAll(),
)

func DisableSQLLogging() {
	GormLogger = slogGorm.New(
		slogGorm.WithHandler(tint.NewHandler(os.Stderr, &tint.Options{
			TimeFormat: time.RFC3339,
		})),
	)
}

func EnableSQLLogging() {
	GormLogger = slogGorm.New(
		slogGorm.WithHandler(tint.NewHandler(os.Stderr, &tint.Options{
			TimeFormat: time.RFC3339,
		})),
		slogGorm.WithTraceAll(),
	)
}

func (cli *CLI) Parse(fs *flag.FlagSet, args []string) error {
	err := ff.Parse(
		fs, args,
		ff.WithEnvVarPrefix("SP"),
	)
	if err != nil {
		return err
	}
	if cli.DataDir == "" {
		return fmt.Errorf("could not determine default data dir (no $HOME) and none provided, please set --data-dir")
	}
	if cli.LivepeerGateway && cli.LivepeerGatewayURL != "" {
		return fmt.Errorf("defining both livepeer-gateway and livepeer-gateway-url doesn't make sense. do you want an embedded gateway or an external one?")
	}
	if cli.LivepeerGateway {
		log.MonkeypatchStderr()
		gatewayPath := cli.DataFilePath([]string{"livepeer", "gateway"})
		err = fs.Set("livepeer.rtmp-addr", "127.0.0.1:0")
		if err != nil {
			return err
		}
		err = fs.Set("livepeer.data-dir", gatewayPath)
		if err != nil {
			return err
		}
		err = fs.Set("livepeer.gateway", "true")
		if err != nil {
			return err
		}
		httpAddrFlag := fs.Lookup("livepeer.http-addr")
		if httpAddrFlag == nil {
			return fmt.Errorf("livepeer.http-addr not found")
		}
		httpAddr := httpAddrFlag.Value.String()
		if httpAddr == "" {
			httpAddr = "127.0.0.1:8935"
			err = fs.Set("livepeer.http-addr", httpAddr)
			if err != nil {
				return err
			}
		}
		cli.LivepeerGatewayURL = fmt.Sprintf("http://%s", httpAddr)
	}
	for _, dest := range cli.dataDirFlags {
		*dest = strings.Replace(*dest, SPDataDir, cli.DataDir, 1)
	}
	if !cli.SQLLogging {
		DisableSQLLogging()
	} else {
		EnableSQLLogging()
	}
	if cli.XXDeprecatedPublicHost != "" && cli.BroadcasterHost == "" {
		log.Warn(context.Background(), "public-host is deprecated, use broadcaster-host or server-host instead as appropriate")
		cli.BroadcasterHost = cli.XXDeprecatedPublicHost
	}
	if cli.ServerHost == "" && cli.BroadcasterHost != "" {
		cli.ServerHost = cli.BroadcasterHost
	}
	if cli.PublicOAuth {
		log.Warn(context.Background(), "--dev-public-oauth is set, this is not recommended for production")
	}
	if cli.FirebaseServiceAccount != "" && cli.FirebaseServiceAccountFile != "" {
		return fmt.Errorf("defining both firebase-service-account and firebase-service-account-file doesn't make sense. do you want a base64-encoded string or a file?")
	}
	if cli.FirebaseServiceAccountFile != "" {
		bs, err := os.ReadFile(cli.FirebaseServiceAccountFile)
		if err != nil {
			return err
		}
		cli.FirebaseServiceAccount = string(bs)
	}
	return nil
}

func (cli *CLI) DataFilePath(fpath []string) string {
	if cli.DataDir == "" {
		panic("no data dir configured")
	}
	// windows does not like colons
	safe := []string{}
	for _, p := range fpath {
		safe = append(safe, strings.ReplaceAll(p, ":", "-"))
	}
	fpath = append([]string{cli.DataDir}, safe...)
	fdpath := filepath.Join(fpath...)
	return fdpath
}

// does a file exist in our data dir?
func (cli *CLI) DataFileExists(fpath []string) (bool, error) {
	ddpath := cli.DataFilePath(fpath)
	_, err := os.Stat(ddpath)
	if err == nil {
		return true, nil
	}
	if errors.Is(err, os.ErrNotExist) {
		return false, nil
	}
	return false, err
}

// write a file to our data dir
func (cli *CLI) DataFileWrite(fpath []string, r io.Reader, overwrite bool) error {
	fd, err := cli.DataFileCreate(fpath, overwrite)
	if err != nil {
		return err
	}
	defer fd.Close()
	_, err = io.Copy(fd, r)
	if err != nil {
		return err
	}

	return nil
}

// create a file in our data dir. don't forget to close it!
func (cli *CLI) DataFileCreate(fpath []string, overwrite bool) (*os.File, error) {
	ddpath := cli.DataFilePath(fpath)
	if !overwrite {
		exists, err := cli.DataFileExists(fpath)
		if err != nil {
			return nil, err
		}
		if exists {
			return nil, fmt.Errorf("refusing to overwrite file that exists: %s", ddpath)
		}
	}
	if len(fpath) > 1 {
		dirs, _ := filepath.Split(ddpath)
		err := os.MkdirAll(dirs, os.ModePerm)
		if err != nil {
			return nil, fmt.Errorf("error creating subdirectories for %s: %w", ddpath, err)
		}
	}
	return os.Create(ddpath)
}

// get a path to a segment file in our database
func (cli *CLI) SegmentFilePath(user string, file string) (string, error) {
	ext := filepath.Ext(file)
	base := strings.TrimSuffix(file, ext)
	aqt, err := aqtime.FromString(base)
	if err != nil {
		return "", err
	}
	fname := fmt.Sprintf("%s%s", aqt.FileSafeString(), ext)
	yr, mon, day, hr, min, _, _ := aqt.Parts()
	return cli.DataFilePath([]string{SegmentsDir, user, yr, mon, day, hr, min, fname}), nil
}

// get a path to a segment file in our database
func (cli *CLI) HLSDir(user string) (string, error) {
	return cli.DataFilePath([]string{SegmentsDir, "hls", user}), nil
}

// create a segment file in our database
func (cli *CLI) SegmentFileCreate(user string, aqt aqtime.AQTime, ext string) (*os.File, error) {
	fname := fmt.Sprintf("%s.%s", aqt.FileSafeString(), ext)
	yr, mon, day, hr, min, _, _ := aqt.Parts()
	return cli.DataFileCreate([]string{SegmentsDir, user, yr, mon, day, hr, min, fname}, false)
}

// read a file from our data dir
func (cli *CLI) DataFileRead(fpath []string, w io.Writer) error {
	ddpath := cli.DataFilePath(fpath)

	fd, err := os.Open(ddpath)
	if err != nil {
		return err
	}
	_, err = io.Copy(w, fd)
	if err != nil {
		return err
	}

	return nil
}

func (cli *CLI) DataDirFlag(fs *flag.FlagSet, dest *string, name, defaultValue, usage string) {
	cli.dataDirFlags = append(cli.dataDirFlags, dest)
	*dest = filepath.Join(SPDataDir, defaultValue)
	usage = fmt.Sprintf(`%s (default: "%s")`, usage, *dest)
	fs.Func(name, usage, func(s string) error {
		*dest = s
		return nil
	})
}

func (cli *CLI) HasMist() bool {
	return runtime.GOOS == "linux"
}

// type for comma-separated ethereum addresses
func (cli *CLI) AddressSliceFlag(fs *flag.FlagSet, dest *[]aqpub.Pub, name, defaultValue, usage string) {
	*dest = []aqpub.Pub{}
	usage = fmt.Sprintf(`%s (default: "%s")`, usage, *dest)
	fs.Func(name, usage, func(s string) error {
		if s == "" {
			return nil
		}
		strs := strings.Split(s, ",")
		for _, str := range strs {
			pub, err := aqpub.FromHexString(str)
			if err != nil {
				return err
			}
			*dest = append(*dest, pub)
		}
		return nil
	})
}

func (cli *CLI) StringSliceFlag(fs *flag.FlagSet, dest *[]string, name string, defaultValue []string, usage string) {
	*dest = defaultValue
	usage = fmt.Sprintf(`%s (default: "%s")`, usage, *dest)
	fs.Func(name, usage, func(s string) error {
		if s == "" {
			return nil
		}
		strs := strings.Split(s, ",")
		*dest = append([]string{}, strs...)
		return nil
	})
}

func (cli *CLI) KVSliceFlag(fs *flag.FlagSet, dest *map[string]string, name, defaultValue, usage string) {
	*dest = map[string]string{}
	usage = fmt.Sprintf(`%s (default: "%s")`, usage, *dest)
	fs.Func(name, usage, func(s string) error {
		if s == "" {
			return nil
		}
		pairs := strings.Split(s, ",")
		for _, pair := range pairs {
			parts := strings.Split(pair, "=")
			if len(parts) != 2 {
				return fmt.Errorf("invalid kv flag: %s", pair)
			}
			(*dest)[parts[0]] = parts[1]
		}
		return nil
	})
}

func (cli *CLI) JSONFlag(fs *flag.FlagSet, dest any, name, defaultValue, usage string) {
	usage = fmt.Sprintf(`%s (default: "%s")`, usage, defaultValue)
	fs.Func(name, usage, func(s string) error {
		if s == "" {
			return nil
		}
		return json.Unmarshal([]byte(s), dest)
	})
}

// debug flag for turning func=ToHLS:3,file=gstreamer.go:4 into {"func": {"ToHLS": 3}, "file": {"gstreamer.go": 4}}
func (cli *CLI) DebugFlag(fs *flag.FlagSet, dest *map[string]map[string]int, name, defaultValue, usage string) {
	*dest = map[string]map[string]int{}
	fs.Func(name, usage, func(s string) error {
		if s == "" {
			return nil
		}
		pairs := strings.Split(s, ",")
		for _, pair := range pairs {
			scoreSplit := strings.Split(pair, ":")
			if len(scoreSplit) != 2 {
				return fmt.Errorf("invalid debug flag: %s", pair)
			}
			score, err := strconv.Atoi(scoreSplit[1])
			if err != nil {
				return fmt.Errorf("invalid debug flag: %s", pair)
			}
			selectorSplit := strings.Split(scoreSplit[0], "=")
			if len(selectorSplit) != 2 {
				return fmt.Errorf("invalid debug flag: %s", pair)
			}
			_, ok := (*dest)[selectorSplit[0]]
			if !ok {
				(*dest)[selectorSplit[0]] = map[string]int{}
			}
			(*dest)[selectorSplit[0]][selectorSplit[1]] = score
		}

		return nil
	})
}

func (cli *CLI) StreamIsAllowed(did string) error {
	if cli.WideOpen {
		return nil
	}
	// if the user set no test streams, anyone can stream
	openServer := len(cli.AllowedStreams) == 0 || (cli.TestStream && len(cli.AllowedStreams) == 1)
	// but only valid atproto accounts! did:key is only allowed for our local test stream
	isDIDKey := strings.HasPrefix(did, constants.DID_KEY_PREFIX)
	if openServer && !isDIDKey {
		return nil
	}
	for _, a := range cli.AllowedStreams {
		if a == did {
			return nil
		}
	}
	return fmt.Errorf("user is not allowed to stream")
}

func (cli *CLI) MyDID() string {
	return fmt.Sprintf("did:web:%s", cli.BroadcasterHost)
}

func (cli *CLI) HasHTTPS() bool {
	return cli.Secure || cli.BehindHTTPSProxy
}

func (cli *CLI) DumpDebugSegment(ctx context.Context, name string, r io.Reader) {
	if cli.SegmentDebugDir == "" {
		return
	}
	go func() {
		err := os.MkdirAll(cli.SegmentDebugDir, 0755)
		if err != nil {
			log.Error(ctx, "failed to create debug directory", "error", err)
			return
		}
		now := aqtime.FromTime(time.Now())
		outFile := filepath.Join(cli.SegmentDebugDir, fmt.Sprintf("%s-%s", now.FileSafeString(), strings.ReplaceAll(name, ":", "-")))
		fd, err := os.Create(outFile)
		if err != nil {
			log.Error(ctx, "failed to create debug file", "error", err)
			return
		}
		defer fd.Close()
		_, err = io.Copy(fd, r)
		if err != nil {
			log.Error(ctx, "failed to copy debug file", "error", err)
			return
		}
		log.Log(ctx, "wrote debug file", "path", outFile)
	}()
}

func (cli *CLI) ShouldSyndicate(did string) bool {
	for _, d := range cli.Syndicate {
		if d == "*" {
			return true
		}
		if d == did {
			return true
		}
	}
	return false
}