Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
3.2 kB · 80 lines
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081package media
import ( "context" "fmt"
"github.com/go-gst/go-gst/gst" "stream.place/streamplace/pkg/atproto" "stream.place/streamplace/pkg/comatproto" "stream.place/streamplace/pkg/model")
// PlaceStreamError is the $type of the websocket frame the dashboard renders as// a stream "problem" (see js/.../websocket-consumer.tsx).const PlaceStreamError = "place.stream.error"
// StreamKick is published to a streamer's bus to forcibly end their live ingest// for a server-side reason (today: exceeding --maximum-live-bitrate). It rides// the same per-streamer bus the ban/key-revocation watcher listens on, so// watchKeyRevocation tears the stream down on it across every ingest path —// in-process (errors the gst pipeline) and isolated (kills the worker// subprocess). It also marshals to a place.stream.error frame, so the websocket// fan-out delivers it to the streamer's dashboard as a problem explaining why// they were disconnected. One publish does both jobs.type StreamKick struct { LexiconTypeID string `json:"$type"` Code string `json:"code"` Message string `json:"message"`}
// NewStreamKick builds a StreamKick carrying the correct $type.func NewStreamKick(code, message string) *StreamKick { return &StreamKick{LexiconTypeID: PlaceStreamError, Code: code, Message: message}}
// watchKeyRevocation blocks until the streamer's signing key is revoked, the// streamer is banned, or a StreamKick is published for them (or ctx ends), then// calls onRevoked once with a reason and returns. It takes the streamer's bus// key + DID as plain strings (not a MediaSigner) so it can also be driven from// just the resume metadata of a worker that outlived a main restart. It is the// shared core behind both the in-process HandleKeyRevocation (which errors the// gst pipeline) and the isolated-worker supervisors (which kill the worker// subprocess). The latter matters because an isolated worker has no bus/model of// its own, so it cannot notice a ban — main has to watch on its behalf, or a// banned (or over-bitrate) user keeps streaming.func (mm *MediaManager) watchKeyRevocation(ctx context.Context, streamer, did string, onRevoked func(reason string)) { sub := mm.bus.Subscribe(streamer) defer mm.bus.Unsubscribe(streamer, sub) for { select { case <-ctx.Done(): return case msg := <-sub: switch v := msg.(type) { case *model.SigningKey: if v.RevokedAt != nil && v.DID == did { onRevoked(fmt.Sprintf("signing key revoked: %s", v.RKey)) return } case *comatproto.LabelDefs_Label: if atproto.IsBanned(v) { onRevoked(fmt.Sprintf("user banned: %s", v.Uri)) return } case *StreamKick: onRevoked(v.Message) return } } }}
// HandleKeyRevocation shuts down an in-process ingest pipeline when the// streamer's signing key is revoked or they get banned.func (mm *MediaManager) HandleKeyRevocation(ctx context.Context, ms MediaSigner, pipeline *gst.Pipeline) { mm.watchKeyRevocation(ctx, ms.Streamer(), ms.DID(), func(reason string) { err := fmt.Errorf("ending stream: %s", reason) pipeline.Error(err.Error(), err) })}