Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
16 kB · 491 lines
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492package api
import ( "bytes" "context" "encoding/json" "fmt" "net" "net/http" "strings" "sync/atomic" "time"
"github.com/google/uuid" "github.com/gorilla/websocket" "github.com/julienschmidt/httprouter" "stream.place/streamplace/pkg/appbsky"
"stream.place/streamplace/pkg/atproto" apierrors "stream.place/streamplace/pkg/errors" "stream.place/streamplace/pkg/log" "stream.place/streamplace/pkg/placestream" "stream.place/streamplace/pkg/renditions" "stream.place/streamplace/pkg/spmetrics")
// todo: does this mean a whole message has to fit within the buffer?var upgrader = websocket.Upgrader{ ReadBufferSize: 1024, WriteBufferSize: 1024, CheckOrigin: func(r *http.Request) bool { return true },}
var pingPeriod = 5 * time.Second
func (a *StreamplaceAPI) HandleWebsocket(ctx context.Context) httprouter.Handle { ctx = log.WithLogValues(ctx, "func", "HandleWebsocket") return func(w http.ResponseWriter, req *http.Request, params httprouter.Params) { ip, _, err := net.SplitHostPort(req.RemoteAddr) if err != nil { ip = req.RemoteAddr }
if a.CLI.RateLimitWebsocket > 0 { if !a.connTracker.AddConnection(ip) { log.Warn(ctx, "rate limit exceeded", "ip", ip, "path", req.URL.Path) apierrors.WriteHTTPTooManyRequests(w, "rate limit exceeded") return }
defer a.connTracker.RemoveConnection(ip) }
uu, _ := uuid.NewV7() connID := uu.String()
ctx = log.WithLogValues(ctx, "uuid", connID, "remoteAddr", req.RemoteAddr, "url", req.URL.String()) log.Log(ctx, "websocket opened") spmetrics.WebsocketsOpen.Inc() defer spmetrics.WebsocketsOpen.Dec() user := params.ByName("repoDID") if user == "" { apierrors.WriteHTTPBadRequest(w, "user required", nil) return } repoDID, err := a.NormalizeUser(ctx, user) if err != nil { apierrors.WriteHTTPNotFound(w, "user not found", err) return } conn, err := upgrader.Upgrade(w, req, nil) if err != nil { apierrors.WriteHTTPInternalServerError(w, "could not upgrade to websocket", err) return } ctx, cancel := context.WithCancel(ctx) defer cancel() defer conn.Close()
initialBurst := make(chan any, 200) err = conn.SetReadDeadline(time.Now().Add(30 * time.Second)) if err != nil { log.Error(ctx, "could not set read deadline", "error", err) return }
pongCh := make(chan struct{})
go func() { for { select { case <-ctx.Done(): return case <-pongCh: err := conn.SetReadDeadline(time.Now().Add(30 * time.Second)) if err != nil { log.Error(ctx, "could not set read deadline", "error", err) return } case <-time.After(30 * time.Second): log.Log(ctx, "websocket timeout, closing connection") // timeout! conn.Close() cancel() return } } }()
conn.SetPongHandler(func(appData string) error { log.Debug(ctx, "received pong", "appData", appData) pongCh <- struct{}{} return nil }) // The rendition list last sent to this viewer, as its joined names. var sentRenditions atomic.Value sentRenditions.Store("") go func() {
ch := a.Bus.SubscribeWithBacklogLimit(repoDID, func() { conn.Close() cancel() }) defer a.Bus.Unsubscribe(repoDID, ch) // The ingest problems last sent to this viewer. Nothing matches // this to begin with, so the first tick always sends the current // list, even an empty one: a reconnecting client may still hold // problems that have since cleared. sentProblems := "\x00unsent" // Create a ticker that fires every 3 seconds ticker := time.NewTicker(3 * time.Second) pingTicker := time.NewTicker(pingPeriod) defer ticker.Stop() defer pingTicker.Stop()
send := func(msg any) { bs, err := json.Marshal(msg) if err != nil { log.Error(ctx, "could not marshal message", "error", err) return } log.Debug(ctx, "sending message", "message", string(bs)) err = conn.WriteMessage(websocket.TextMessage, bs) if err != nil { log.Error(ctx, "could not write message", "error", err) return } }
// Subscribe before reading the snapshot so changes during the lookup // are buffered. Send the snapshot first, then consume those changes so // a cancellation cannot be followed by the stale record. teleport, err := a.Model.GetTeleportForViewer(repoDID, req.URL.Query().Get("teleportUri"), req.URL.Query().Get("teleportCid")) if err != nil { log.Error(ctx, "could not get teleport", "error", err) } else if teleport != nil && teleport.Teleport != nil { var record placestream.LiveTeleport if err := record.UnmarshalCBOR(bytes.NewReader(*teleport.Teleport)); err != nil { log.Error(ctx, "could not decode teleport", "error", err, "uri", teleport.URI) } else { teleportMessage := map[string]any{ "$type": record.RecordTypeID(), "uri": teleport.URI, "cid": teleport.CID, "streamer": record.Streamer, "startsAt": record.StartsAt, } if record.DurationSeconds != nil { teleportMessage["durationSeconds"] = *record.DurationSeconds } if record.Livestream != nil { teleportMessage["livestream"] = record.Livestream } send(teleportMessage) } }
for { select { case msg := <-ch: send(msg) case msg := <-initialBurst: send(msg) case <-ticker.C: // The renditions a viewer can pick are whatever this // node's live window holds now: they appear a little // after the stream starts (the transcoder's round trip), // and on a syndicating node they arrive from the origin. // Including the change to none: renditions age out of // the window on their own when the transcoder stops, // and the viewer's menu must lose them too. names := a.MediaManager.LiveRenditionNames(repoDID) if key := strings.Join(names, ","); sentRenditions.Swap(key) != key { send(renditionsMessage(names)) } // Problems with how the streamer is connected to this // node, e.g. through a deprecated ingest hostname: they // live as long as the connection, not in any segment, // so they're sent when they change. if probs := a.ingestProblems(ctx, repoDID); problemsKey(probs) != sentProblems { sentProblems = problemsKey(probs) send(probs) } bs, err := json.Marshal(a.viewerCountMessage(ctx, repoDID)) if err != nil { log.Error(ctx, "could not marshal view count", "error", err) continue } err = conn.WriteMessage(websocket.TextMessage, bs) if err != nil { log.Error(ctx, "could not write ping message", "error", err) return } case <-pingTicker.C: err := conn.WriteMessage(websocket.PingMessage, []byte{}) if err != nil { log.Error(ctx, "could not write ping message", "error", err) return } case <-ctx.Done(): log.Debug(ctx, "context done, stopping websocket sender") return } } }()
go func() { profile, err := a.Model.GetRepo(repoDID) if err != nil { log.Error(ctx, "could not get profile", "error", err) return } if profile != nil { p := map[string]any{ "$type": "app.bsky.actor.defs#profileViewBasic", "did": repoDID, "handle": profile.Handle, } // Display name and avatar from the indexed profile record, so // the stream page names the streamer the way chat rows do // without a trip to an app view. if bp, err := a.Model.GetBskyProfile(ctx, repoDID, false); err == nil && bp != nil { if bp.DisplayName != nil && *bp.DisplayName != "" { p["displayName"] = *bp.DisplayName } if bp.Avatar != nil { p["avatar"] = fmt.Sprintf("https://cdn.bsky.app/img/avatar/plain/%s/%s@jpeg", repoDID, bp.Avatar.Ref.String()) } } // The streamer's verified badge comes from the same place as // chat authors'. if state := a.ATSync.VerificationState(ctx, repoDID, repoDID); state != nil { p["verification"] = state } initialBurst <- p } }()
go func() { seg, err := a.LocalDB.LatestSegmentForUser(repoDID) if err != nil { log.Error(ctx, "could not get replies", "error", err) return } spSeg, err := seg.ToStreamplaceSegment() if err != nil { log.Error(ctx, "could not convert segment to streamplace segment", "error", err) return } initialBurst <- spSeg names := a.MediaManager.LiveRenditionNames(repoDID) if len(names) == 0 && a.CLI.LivepeerGatewayURL != "" { // This node transcodes but hasn't a rendition in its window // yet: the profiles it is about to produce. videoRenditions, err := renditions.GenerateRenditionsFrom(spSeg, a.CLI.RenditionLadder()) if err != nil { log.Error(ctx, "could not generate renditions", "error", err) return } for _, r := range videoRenditions { names = append(names, r.Name) } } sentRenditions.Store(strings.Join(names, ",")) initialBurst <- renditionsMessage(names) }()
go func() { ls, err := a.Model.GetLatestLivestreamForRepo(repoDID) if err != nil { log.Error(ctx, "could not get latest livestream", "error", err) return } if ls == nil { log.Error(ctx, "no livestream found", "repoDID", repoDID) return } lsv, err := ls.ToLivestreamView() if err != nil { log.Error(ctx, "could not marshal livestream", "error", err) return } initialBurst <- lsv }()
go func() { initialBurst <- a.viewerCountMessage(ctx, repoDID) }()
go func() { // Withhold chat older than the node's retention window: a viewer // arriving the next day should land on an empty chat, not // yesterday's conversation. A zero retention serves everything. var since time.Time if a.CLI.ChatMessageRetention > 0 { since = time.Now().Add(-a.CLI.ChatMessageRetention) } messages, err := a.Model.MostRecentChatMessages(repoDID, since) if err != nil { log.Error(ctx, "could not get chat messages", "error", err) return }
// Add mod badges to messages issuerDID := fmt.Sprintf("did:web:%s", a.CLI.BroadcasterHost) for _, message := range messages { if !a.ATSync.ChatAllowed(ctx, repoDID, message.Author.Did) { continue } err := atproto.AddModBadgeIfApplicable(ctx, &message, repoDID, issuerDID, a.Model) if err != nil { log.Error(ctx, "failed to add mod badge to message", "error", err) } a.ATSync.DecorateVerification(ctx, repoDID, &message) if message.Author.Handle == "" || message.Author.Handle == "handle.invalid" { message.Author.Handle = a.ATSync.ResolveAuthorHandle(ctx, message.Author.Did) } initialBurst <- message } }()
// get the latest active pinned message for the repo go func() { pin, err := a.Model.GetActivePinnedRecord(ctx, repoDID) if err != nil { log.Error(ctx, "could not get pinned record", "error", err) return } if pin != nil { prv, err := pin.ToStreamplacePinnedRecordView() if err != nil { log.Error(ctx, "could not convert pinned record to streamplace view", "error", err) return } // look up the original message, pinner msg, err := a.Model.GetChatMessage(prv.Record.PinnedMessage) if err != nil { log.Error(ctx, "failed to get pinned message", err) return } // if the message was deleted, treat as no pinned message if msg != nil && msg.DeletedAt != nil { log.Log(ctx, "pinned message was deleted, skipping", "uri", msg.URI) return } // if no pinned by, use the repo owner as the pinner if prv.Record.PinnedBy == nil { prv.Record.PinnedBy = &repoDID } profile, err := a.Model.GetChatProfile(ctx, *prv.Record.PinnedBy) if err != nil { log.Error(ctx, "failed to get chat profile", err) return } if msg != nil { msgView, err := msg.ToStreamplaceMessageView() if err != nil { log.Error(ctx, "failed to convert chat message: %w", err) return } if msgView.Author.Handle == "" || msgView.Author.Handle == "handle.invalid" { msgView.Author.Handle = a.ATSync.ResolveAuthorHandle(ctx, msgView.Author.Did) } prv.Message = msgView } if profile != nil { profileView, err := profile.ToStreamplaceChatProfile() if err != nil { log.Error(ctx, "failed to convert chat profile: %w", err) return } prv.PinnedBy = &profileView } initialBurst <- prv } }()
go func() { teleports, err := a.Model.GetActiveTeleportsToRepo(repoDID) if err != nil { log.Error(ctx, "could not get active teleports", "error", err) return } // just send the latest one if it started <3m ago if len(teleports) > 0 && teleports[0].StartsAt.After(time.Now().Add(-3*time.Minute)) { tp := teleports[0] if tp.Repo == nil { log.Error(ctx, "teleportee repo is nil", "uri", tp.URI) return } viewerCount := a.Bus.GetViewerCount(tp.RepoDID) arrivalMsg := placestream.Livestream_TeleportArrival{ LexiconTypeID: "place.stream.livestream#teleportArrival", TeleportUri: tp.URI, Source: appbsky.ActorDefs_ProfileViewBasic{ Did: tp.RepoDID, Handle: tp.Repo.Handle, }, ViewerCount: int64(viewerCount), StartsAt: tp.StartsAt.Format(time.RFC3339), }
// get the source chat profile chatProfile, err := a.Model.GetChatProfile(ctx, tp.RepoDID) if err == nil && chatProfile != nil { spcp, err := chatProfile.ToStreamplaceChatProfile() if err == nil { arrivalMsg.ChatProfile = &spcp } }
initialBurst <- arrivalMsg } }()
for { messageType, message, err := conn.ReadMessage() if err != nil { log.Error(ctx, "error reading message", "error", err) break } log.Log(ctx, "received message", "messageType", messageType, "message", string(message)) } }}
// renditionsMessage is the place.stream.defs#renditions message listing the// video renditions a viewer can pick, plus the audio-only rendition every// stream has.func renditionsMessage(names []string) placestream.Defs_Renditions { out := placestream.Defs_Renditions{ LexiconTypeID: "place.stream.defs#renditions", Renditions: make([]placestream.Defs_Rendition, 0, len(names)+1), } for _, n := range names { out.Renditions = append(out.Renditions, placestream.Defs_Rendition{LexiconTypeID: "place.stream.defs#rendition", Name: n}) } out.Renditions = append(out.Renditions, placestream.Defs_Rendition{LexiconTypeID: "place.stream.defs#rendition", Name: renditions.AudioRendition.Name}) return out}
// viewerCountMessage is the viewerCount event: who is watching now (the// bus) and how many sessions this livestream has had (statedb's running// total, when the node keeps one).func (a *StreamplaceAPI) viewerCountMessage(ctx context.Context, repoDID string) placestream.Livestream_ViewerCount { msg := placestream.Livestream_ViewerCount{ Count: int64(a.Bus.GetViewerCount(repoDID)), LexiconTypeID: "place.stream.livestream#viewerCount", } if a.StatefulDB == nil { return msg } // The total of the streamer's current livestream record, looked up by // its URI: the most recently counted-into row can still be the previous // record's for a moment after a new record is indexed. if ls, err := a.Model.GetLatestLivestreamForRepo(repoDID); err == nil && ls != nil { if totals, err := a.StatefulDB.LivestreamViewTotals(ctx, []string{ls.URI}); err == nil { if total := totals[ls.URI]; total > 0 { msg.Total = &total } return msg } } if total, _, err := a.StatefulDB.GetStreamViewTotal(ctx, repoDID); err == nil && total > 0 { msg.Total = &total } return msg}