sse

type ActivityPayload ¶

type ActivityPayload struct {
	Type     string          `json:"type"`
	Activity *activity.Event `json:"activity,omitempty"`
}

ActivityPayload is the message envelope for qui-owned server activity events. It is intentionally distinct from StreamPayload (whose Data is a torrent response) so the frontend's torrent-stream router never sees activity events: they are delivered as a separate named "activity" SSE event with their own handler that invalidates cached queries.

type StreamDelta ¶ added in v1.21.0

type StreamDelta struct {
	Order *[]string `json:"order,omitempty"`
}

StreamDelta describes how to reconcile a group's page-0 window against the previous frame on a "delta" event. The added or changed rows themselves ride in StreamPayload.Data.Torrents (single-instance) or Data.CrossInstanceTorrents (cross-instance); Order lists the full page key sequence and is present only when membership or ordering changed. When Order is omitted the client applies the changed rows in place at their existing positions. Keys are the torrent hash for single-instance streams and "<instanceID>:<hash>" for cross-instance streams.

Order is a pointer so a present-but-empty order (the page drained to zero rows, e.g. every match deleted) serializes as `[]` and stays distinct from an absent order. A plain `[]string` with omitempty would drop the empty slice on the wire, making a full clear indistinguishable from an aggregate-only tick and leaving the deleted rows on screen until the next keyframe.

type StreamManager ¶

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

StreamManager owns the SSE server and keeps subscriptions in sync with qBittorrent updates.

Lock hierarchy (acquire in this order to prevent deadlock):

1. m.mu (StreamManager.mu) - protects subscriptions, groups, loops
2. group.mu (subscriptionGroup.mu) - protects pending queue state
3. group.subsMu (subscriptionGroup.subsMu) - protects subscriber list

func NewStreamManager(clientPool *qbittorrent.ClientPool, syncManager syncProvider, instanceStore *models.InstanceStore) *StreamManager

func (m *StreamManager) HandleMainData(instanceID int, data *qbt.MainData)

func (m *StreamManager) HandleSyncError(instanceID int, err error)

func (m *StreamManager) HandleTrackerHealthUpdated(instanceID int)

func (m *StreamManager) PrepareBatch(ctx context.Context, requests []streamRequest) (context.Context, []string, error)

func (m *StreamManager) Serve(w http.ResponseWriter, r *http.Request)

func (m *StreamManager) SetActivityHub(hub *activity.Hub)

func (m *StreamManager) Shutdown(ctx context.Context) error

func (m *StreamManager) Stats() StreamStats

func (m *StreamManager) Unregister(id string)

type StreamMeta ¶

type StreamMeta struct {
	InstanceID int       `json:"instanceId"`
	RID        int64     `json:"rid,omitempty"`
	FullUpdate bool      `json:"fullUpdate,omitempty"`
	Timestamp  time.Time `json:"timestamp"`
	// IncludeCounts asks materialization to include cached aggregate counts even
	// when the tick skips tracker hydration. Used after tracker-health cache writes
	// so dashboards receive new health totals without inline qbit work.
	IncludeCounts bool `json:"includeCounts,omitempty"`
	// LastSuccessfulSync is when the instance's data last actually updated, sourced
	// from the success-only sync clock. On stream-error frames it lets the client
	// show how stale the retained torrents are ("data from N ago") without that age
	// resetting on every failed attempt. A pointer so the zero time is omitted on the
	// wire rather than serialized as 0001-01-01 (omitempty has no effect on a struct).
	LastSuccessfulSync *time.Time `json:"lastSuccessfulSync,omitempty"`
	RetryInSeconds     int        `json:"retryInSeconds,omitempty"`
	StreamKey          string     `json:"streamKey,omitempty"`
}

StreamMeta carries lightweight metadata about the sync update.

type StreamOptions ¶

type StreamOptions struct {
	InstanceID  int
	InstanceIDs []int
	Page        int
	Limit       int
	Sort        string
	Order       string
	Search      string
	Filters     qbittorrent.FilterOptions
}

StreamOptions captures the torrent view that the subscriber wants to keep in sync.

A subscription is single-instance when InstanceIDs is empty (keyed by InstanceID), or multi-instance (aggregated/cross-instance) when InstanceIDs holds one or more concrete instance ids. Multi-instance subscriptions are kept in sync by every one of their member instances.

type StreamPayload ¶

type StreamPayload struct {
	Type  string                       `json:"type"`
	Data  *qbittorrent.TorrentResponse `json:"data,omitempty"`
	Delta *StreamDelta                 `json:"delta,omitempty"`
	Meta  *StreamMeta                  `json:"meta,omitempty"`
	Err   string                       `json:"error,omitempty"`
}

StreamPayload is the message envelope sent to the frontend.

type StreamStats ¶

type StreamStats struct {
	ActiveSubscriptions int    // currently connected subscribers
	ActiveGroups        int    // distinct view groups being served
	ActiveSyncLoops     int    // per-instance sync loops running
	EventsPublished     uint64 // lifetime SSE messages successfully published
	EventsDropped       uint64 // lifetime messages dropped (marshal/publish failures)
	SyncErrors          uint64 // lifetime sync errors propagated to subscribers
}

StreamStats is a point-in-time snapshot of SSE subsystem activity. It is exported so the metrics layer can surface it (e.g. as Prometheus gauges/counters).