cluster

type Caps struct {
	// Transcode is true when ffmpeg is available on the node.
	Transcode bool `json:"transcode"`
	// HWAccel names the hardware transcoder ("videotoolbox", "nvenc", …);
	// empty or "none" for software only.
	HWAccel string `json:"hwAccel,omitempty"`
	// Weight is the node's share weight for account-wide budgets such as
	// Usenet connections (config server.clusterWeight; 0 means 1).
	Weight int `json:"weight,omitempty"`
	// EgressIP is the public address the node reaches the internet from,
	// when known (providers may flag one account used from several).
	EgressIP string `json:"egressIp,omitempty"`
}

type Cluster interface {
	// Mode is ModeSolo or ModePostgres.
	Mode() string
	// Self describes this node.
	Self() Node
	// IsLeader reports whether this node currently holds leadership.
	IsLeader() bool
	// Epoch is the leadership epoch this node acquired (0 while not
	// leading; always 1 for Solo). Every acquisition anywhere in the
	// cluster increments it, so a singleton writer can fence a write with
	// "... WHERE (SELECT epoch FROM cluster_leader WHERE id=1) = ?".
	Epoch() int64
	// OnLeaderChange calls fn (synchronously, from the election goroutine;
	// keep it quick) whenever leadership is gained or lost.
	OnLeaderChange(fn func(leader bool)) (unsubscribe func())
	// RunAsLeader runs fn while this node leads, cancelling its context as
	// soon as leadership is lost and starting it again when regained. It
	// blocks until ctx ends and the last fn has returned. A fn that returns
	// on its own is not restarted until leadership is lost and regained.
	RunAsLeader(ctx context.Context, name string, fn func(ctx context.Context))
	// Publish sends payload (≤ MaxPayload) to every subscriber of topic on
	// every node, this one included (local subscribers synchronously).
	Publish(ctx context.Context, topic string, payload []byte) error
	// Subscribe registers fn for topic. fn runs on the bus goroutine (or
	// the publisher's, for local events); it must not block and must treat
	// payload as read-only.
	Subscribe(topic string, fn func(payload []byte)) (unsubscribe func())
	// Nodes lists live members, ordered by name.
	Nodes(ctx context.Context) ([]Node, error)
	// AllNodes lists every recorded member, dead or alive, ordered by name.
	AllNodes(ctx context.Context) ([]Node, error)
	// Start begins heartbeats, election and the bus (no-op for Solo).
	Start(ctx context.Context) error
	// Close leaves the cluster: stops background work, releases
	// leadership and removes this node's heartbeat row.
	Close() error
}

type Health struct {
	Channel       string    `json:"channel"`
	BusConnected  bool      `json:"busConnected"`
	LastHeartbeat time.Time `json:"lastHeartbeat,omitempty"`
	// ClockSkew is this node's clock minus the database server's, measured
	// at Start.
	ClockSkew time.Duration `json:"clockSkewNs"`
}

type Identity struct {
	// NodeID is generated on first run and persisted in NodeFile.
	NodeID string `json:"nodeId"`
	// Name is the display name: NodeFile's "name" when the owner set one,
	// else the hostname.
	Name string `json:"name,omitempty"`
}

type LeaseHolder struct {
	NodeID     string    `json:"nodeId"`
	Version    string    `json:"version"`
	AcquiredAt time.Time `json:"acquiredAt"`
	ExpiresAt  time.Time `json:"expiresAt"`
}

type Local struct {
	Identity Identity
	Version  string
	Addrs    func(ctx context.Context) []string
	Caps     func() Caps
	Load     func() int
	// Draining reports whether the node is draining for an update; a
	// draining node also hands leadership to another live node.
	Draining func() bool
	// Migrations are the schema migrations this binary knows
	// (store.KnownMigrations), heartbeated for the schema compatibility
	// check.
	Migrations map[string]int
}

type Node struct {
	ID      string `json:"id"`
	Name    string `json:"name"`
	Version string `json:"version"`
	// Addrs are the connection URIs this node would publish to plex.tv
	// (LAN, WAN and custom URLs, as the registrar builds them).
	Addrs     []string  `json:"addrs"`
	Caps      Caps      `json:"caps"`
	Load      int       `json:"load"` // active playback sessions
	StartedAt time.Time `json:"startedAt"`
	LastSeen  time.Time `json:"lastSeen"`
	Leader    bool      `json:"leader"`
	// Live is true when the node heartbeated recently (always true from
	// Nodes, may be false from AllNodes).
	Live bool `json:"live"`
	// Draining is true while the node prepares to restart for an update:
	// it serves its existing sessions but admits no new ones, and its
	// addresses are left out of the published connections.
	Draining bool `json:"draining,omitempty"`
	// Migrations are the schema migrations the node's binary knows
	// (component -> highest version); nil when the node predates this
	// report.
	Migrations map[string]int `json:"migrations,omitempty"`
}

type Options struct {
	// DB is the node's pool on the shared database (store "ether-pgx").
	DB    *sql.DB
	Local Local
	// ConnConfig is used for the dedicated leader-lock and LISTEN
	// connections; nil copies the configuration of a pool connection, so
	// they reach the same server, database and search_path.
	ConnConfig *pgx.ConnConfig
	Logger     *slog.Logger
	Now        func() time.Time

	HeartbeatInterval time.Duration // heartbeat upsert (5s)
	LiveAfter         time.Duration // a node is live if seen within this (20s)
	PruneAfter        time.Duration // rows dead this long are deleted (24h)
	ElectionInterval  time.Duration // lock attempt / leader probe (2s)
	ProbeTimeout      time.Duration // bound on each probe query (3s)
	// StepDownHold keeps a node that just lost leadership out of the
	// election for a while so a healthy node takes over (5s).
	StepDownHold time.Duration
	// IdleSessionTimeout is set as idle_session_timeout on the lock
	// connection: if this node freezes or is partitioned from the database,
	// the server ends the session and releases the lock after this long
	// rather than waiting for TCP keepalives (15s; must exceed
	// ElectionInterval).
	IdleSessionTimeout time.Duration
	ListenPing         time.Duration // ping an idle LISTEN connection (15s)
	MinBackoff         time.Duration // bus reconnect backoff (500ms)
	MaxBackoff         time.Duration // (30s)
	AddrTimeout        time.Duration // bound on Local.Addrs (30s)
	// IdentityPath names this node's node.json in duplicate-node errors.
	IdentityPath string

	LockKey int32  // DefaultLockKey
	Channel string // derived from the schema when empty

	// Availability, when set, receives every heartbeat's outcome: the
	// node's shared view of database reachability, which lets deferrable
	// writes queue at once during an outage and replay as soon as a
	// heartbeat succeeds again.
	Availability *store.Availability

	// OnFatal is called once, from a runtime goroutine, when this node
	// must stop serving as a cluster member: another live process
	// heartbeats under the same node ID (a cloned machine or copied data
	// directory). The runtime has already stopped heartbeating, released
	// leadership and left the bus when it is called. The error wraps
	// ErrDuplicateNode.
	OnFatal func(error)
}

type Postgres struct {
	// contains filtered or unexported fields
}

type Solo struct {
	// contains filtered or unexported fields
}

type UpdateLease struct {
	DB     *sql.DB
	NodeID string
	// TTL is the lease lifetime from each acquisition or renewal
	// (DefaultUpdateLeaseTTL).
	TTL time.Duration
	Now func() time.Time
}