Something went wrong. Try again.
Something went wrong. Try again.
33 kB · 767 lines
Go
at next
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768package statedb
import ( "context" "encoding/json" "errors" "fmt" "time"
"github.com/bluesky-social/indigo/atproto/syntax" "github.com/bluesky-social/indigo/util" "github.com/bluesky-social/indigo/xrpc" glex "github.com/streamplace/glex/runtime" "golang.org/x/sync/errgroup" "gorm.io/gorm" "stream.place/streamplace/pkg/appbsky" "stream.place/streamplace/pkg/integrations/webhook" "stream.place/streamplace/pkg/log" "stream.place/streamplace/pkg/model" notificationpkg "stream.place/streamplace/pkg/notifications" "stream.place/streamplace/pkg/placestream"
"stream.place/streamplace/pkg/comatproto")
var TaskNotification = "notification"var TaskChat = "chat"var TaskStreamReceived = "stream_received"var TaskFinalizeLivestream = "finalize_livestream"var TaskFinalizeLivestreamVOD = "finalize_livestream_vod"var TaskVODProcess = "vod_process"var TaskViewCountAggregate = "view_count_aggregate"var TaskCDNLogIngest = "cdn_log_ingest"var TaskAutoPublishVOD = "auto_publish_vod"
// nonVODTaskTypes is every task type handled by the general queue worker.// VOD processing runs on its own dedicated pool (see ProcessQueue) so a// slow remux can't block these lighter tasks, so VOD is deliberately// excluded here. Keep this list in sync when adding a new task type that// is NOT VOD — anything missing from both this list and the VOD pool will// never be dequeued.var nonVODTaskTypes = []string{ TaskNotification, TaskChat, TaskStreamReceived, TaskFinalizeLivestream, TaskViewCountAggregate, TaskCDNLogIngest, TaskAutoPublishVOD,}
type NotificationTask struct { Livestream placestream.Livestream_LivestreamView FeedPost *appbsky.FeedDefs_PostView ChatProfile placestream.ChatProfile PDSURL string}
type ChatTask struct { MessageView placestream.ChatDefs_MessageView}
type StreamReceivedTask struct { StreamerDID string `json:"streamerDID"`}
type FinalizeLivestreamTask struct { LivestreamURI string `json:"livestreamURI"` // StaleSince is the record's lastSeenAt as it stood when this task was // rescheduled to give the heartbeat one more idle window. If it is still // that when the task runs again, the heartbeat is frozen and the record // is ended; without it the task rescheduled itself forever for a repo's // latest un-ended record. StaleSince string `json:"staleSince,omitempty"`}
// heartbeatFrozen reports whether a rescheduled finalize found the record's// heartbeat exactly where it left it a full idle window ago.func heartbeatFrozen(task FinalizeLivestreamTask, lastSeenAt string) bool { return task.StaleSince != "" && task.StaleSince == lastSeenAt}
// VODProcessTask is enqueued by the upload manager when a resumable user// upload completes. The processor probes the file, generates MUXL tracks,// and creates the place.stream.video record set. The actual processing is// not yet implemented — for now this just acknowledges receipt.type VODProcessTask struct { UploadID string `json:"uploadId"` RepoDID string `json:"repoDID"` MimeType string `json:"mimeType"` Filename string `json:"filename,omitempty"` Size int64 `json:"size"` Backend string `json:"backend"` Location string `json:"location"`}
// FinalizeLivestreamVODTask is enqueued by the place.stream.media.finalizeLivestream// procedure to turn a finished livestream's recorded MUXL objects into a VOD.// UploadID is the synthetic Upload row the client polls (getUploadStatus) and// then publishes (publishVideo), exactly as for a resumable upload.type FinalizeLivestreamVODTask struct { UploadID string `json:"uploadId"` RepoDID string `json:"repoDID"` LivestreamURI string `json:"livestreamURI"` // LivestreamURIs, when set, are all the livestream records whose // recordings make up the VOD, in order (LivestreamURI is the first); // empty means just LivestreamURI. LivestreamURIs []string `json:"livestreamURIs,omitempty"` // Publish, when set, describes the place.stream.video record to // publish in the streamer's repo (with their stored session) as soon // as the VOD is finalized, instead of leaving a draft for them to // publish from the app (unless publishing fails). Set by the // operator's finalize route and by automatic VOD publishing // (AutoPublishVODTask). Publish *VideoDraft `json:"publish,omitempty"`}
// VideoDraft is the part of a place.stream.video record known before the// VOD exists: what the operator (or the livestream record) says about it.// It is not a placestream.Video because that record's source union has no// value until the finalizer has the MUXL CID, and an empty union will not// marshal, so a Video cannot sit in a task payload; the publisher builds// the record from this with Record() and fills in the source, duration and// thumbnail from the finished upload.type VideoDraft struct { Title string `json:"title"` Description *string `json:"description,omitempty"` Tags []string `json:"tags,omitempty"` Activity *placestream.Video_Activity `json:"activity,omitempty"` Connections []placestream.Video_Connections_Elem `json:"connections,omitempty"`}
// Record is the place.stream.video record for the draft, minus the fields// that come from the finalized upload.func (d *VideoDraft) Record() *placestream.Video { return &placestream.Video{ LexiconTypeID: "place.stream.video", Title: d.Title, Description: d.Description, Tags: d.Tags, Activity: d.Activity, Connections: d.Connections, }}
// VideoPublisher publishes a finalized livestream VOD's place.stream.video// record; installed at bootstrap like LivestreamVODFinalizer.type VideoPublisher func(ctx context.Context, t FinalizeLivestreamVODTask) (uri, cid string, err error)
func (state *StatefulDB) SetVideoPublisher(f VideoPublisher) { state.videoPublisher = f}
// ViewCountAggregateTask is the payload for one aggregation window.// Enqueued by every streamplace node at the configured interval; the// unique task key (built from WindowStart/End) ensures only one node's// enqueue + dequeue actually runs each window.type ViewCountAggregateTask struct { WindowStart time.Time `json:"windowStart"` WindowEnd time.Time `json:"windowEnd"`}
// CDNLogIngestTask is the payload for one scheduled pass over the// CDN's archived access logs. Tick is the UTC-aligned schedule slot;// it's in the dedup key, not used by the ingester itself (which// always looks for whatever is new).type CDNLogIngestTask struct { Tick time.Time `json:"tick"`}
// ProcessQueue runs the task queue until ctx is cancelled. VOD tasks are// handled by a dedicated pool of vodConcurrency workers, so a slow remux// can't block quick uploads behind it (or starve the lighter tasks);// everything else runs on a single general worker. DequeueTask uses// FOR UPDATE SKIP LOCKED on Postgres, so the workers never claim the same// row. vodConcurrency is clamped to at least 1.func (state *StatefulDB) ProcessQueue(ctx context.Context, vodConcurrency int) error { if vodConcurrency < 1 { vodConcurrency = 1 } log.Log(ctx, "starting task queue", "vod_concurrency", vodConcurrency)
group, ctx := errgroup.WithContext(ctx)
// General worker: everything except VOD. group.Go(func() error { return state.runQueueWorker(ctx, "queue_processor", nonVODTaskTypes) })
// Dedicated VOD pool. Live-to-VOD finalize also runs here: it's heavy // I/O (a full read of the recorded stream to hash + index it) that would // otherwise hog the single general worker and stall light tasks. for i := 0; i < vodConcurrency; i++ { workerID := fmt.Sprintf("vod_worker_%d", i) group.Go(func() error { return state.runQueueWorker(ctx, workerID, []string{TaskVODProcess, TaskFinalizeLivestreamVOD}) }) }
return group.Wait()}
// runQueueWorker pulls and processes tasks of the given types until ctx is// cancelled. A failed task is only logged here; it stays locked until its// lease expires and then becomes eligible for retry (capped by max_tries),// matching the queue's existing retry semantics. Empty dequeues back off// on a 1s timer or a queue poke.func (state *StatefulDB) runQueueWorker(ctx context.Context, workerID string, taskTypes []string) error { for { task, err := state.DequeueTask(ctx, workerID, taskTypes...) if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) { return err } if task != nil { // Defense-in-depth: a panic inside a task handler (e.g. a // nil-deref in a codec, or a malformed record) must not crash // this worker goroutine — errgroup would cancel the whole queue. // Recover it into a logged error so the task fails and its retry // semantics (lease expiry + max_tries) still apply. if err := log.Recover(ctx, func() error { return state.processTask(ctx, task) }); err != nil { log.Error(ctx, "failed to process task", "err", err, "worker", workerID, "taskId", fmt.Sprintf("%d", task.ID)) } continue } select { case <-ctx.Done(): return ctx.Err() case <-time.After(1 * time.Second): case <-state.pokeQueue: } }}
func (state *StatefulDB) processTask(ctx context.Context, task *AppTask) error { ctx = log.WithLogValues(ctx, "taskType", task.Type, "taskId", fmt.Sprintf("%d", task.ID)) switch task.Type { case TaskNotification: return state.processNotificationTask(ctx, task) case TaskChat: return state.processChatMessageTask(ctx, task) case TaskStreamReceived: return state.processStreamReceivedTask(ctx, task) case TaskFinalizeLivestream: return state.processFinalizeLivestreamTask(ctx, task) case TaskVODProcess: return state.processVODProcessTask(ctx, task) case TaskFinalizeLivestreamVOD: return state.processFinalizeLivestreamVODTask(ctx, task) case TaskViewCountAggregate: return state.processViewCountAggregateTask(ctx, task) case TaskCDNLogIngest: return state.processCDNLogIngestTask(ctx, task) case TaskAutoPublishVOD: return state.processAutoPublishVODTask(ctx, task) default: return fmt.Errorf("unknown task type: %s", task.Type) }}
// VODProcessor runs the gstreamer + muxl + S3 pipeline for one upload// and returns the resulting BDASL CID. The function-pointer indirection// keeps pkg/statedb from importing pkg/vod (which transitively pulls in// gstreamer); the bootstrap (pkg/cmd) installs the concrete// implementation at startup.type VODProcessor func(ctx context.Context, t VODProcessTask) (cid string, err error)
func (state *StatefulDB) SetVODProcessor(f VODProcessor) { state.vodProcessor = f }
// SetNotifier installs the notification notifier after construction. This is// needed because building the Web Push notifier requires VAPID keys, which// are stored in the DB — so the DB must exist before the notifier can be// fully assembled. The queue processor checks for nil, so a brief window// with no notifier is safe.func (state *StatefulDB) SetNotifier(n notificationpkg.Notifier) { state.noter = n }
func (state *StatefulDB) processVODProcessTask(ctx context.Context, task *AppTask) error { ctx = log.WithLogValues(ctx, "func", "processVODProcessTask") var t VODProcessTask if err := json.Unmarshal(task.Payload, &t); err != nil { return err } // Thread the upload + owner into the log context as early as possible so // every downstream log line (and the error returned below, which is // re-logged by runQueueWorker with the loop's context) carries the user. ctx = log.WithLogValues(ctx, "uploadId", t.UploadID, "did", t.RepoDID) log.Log(ctx, "dequeued vod-process task") if state.vodProcessor == nil { log.Warn(ctx, "no VOD processor configured; dropping task", "uploadId", t.UploadID, "did", t.RepoDID) return state.CompleteTask(ctx, task.ID) } if err := state.SetUploadProcessing(ctx, t.UploadID); err != nil { log.Warn(ctx, "failed to mark upload as processing", "uploadId", t.UploadID, "error", err) } cid, err := state.vodProcessor(ctx, t) if err != nil { if ferr := state.SetUploadFailed(ctx, t.UploadID, err.Error()); ferr != nil { log.Warn(ctx, "failed to mark upload as failed", "uploadId", t.UploadID, "error", ferr) } // Flip any tied draft to 'error' too, so the user sees the failure in // the Drafts tab instead of an indefinite 'processing' state. if derr := state.SetDraftError(ctx, t.UploadID, err.Error()); derr != nil { log.Warn(ctx, "failed to mark draft as failed", "uploadId", t.UploadID, "error", derr) } // Complete the task so it doesn't retry — most VOD failures are // permanent (unsupported codec, corrupted file, etc.). _ = state.CompleteTask(ctx, task.ID) // The upload ID + DID are now in the log context (set above), so the // error string no longer needs to embed them for traceability — the // runQueueWorker re-log picks them up from context. return fmt.Errorf("vod processing upload %s: %w", t.UploadID, err) } // The processor (vod.ProcessVOD) calls SetUploadProcessed deep inside its // own publish path, so by the time it returns the Upload row carries the // finished TrackURIs / DurationMS / ContentCID. Re-read them and flip the // tied draft to 'ready'. A missing draft (pre-drafts-era upload, or one // whose create failed) is a no-op, not an error. if err := state.markDraftReadyFromUpload(ctx, t.UploadID); err != nil { log.Warn(ctx, "failed to mark draft ready", "uploadId", t.UploadID, "error", err) } log.Log(ctx, "vod processed", "uploadId", t.UploadID, "cid", cid) return state.CompleteTask(ctx, task.ID)}
// LivestreamVODFinalizer concatenates a finished livestream's recorded MUXL// objects into a content-addressed VOD blob, derives its sidecars, and// publishes the origin + track records. Returns the resulting BDASL CID. Same// function-pointer indirection as VODProcessor so pkg/statedb doesn't import// the blob.Store/muxl-heavy pkg/vod.type LivestreamVODFinalizer func(ctx context.Context, t FinalizeLivestreamVODTask) (cid string, err error)
func (state *StatefulDB) SetLivestreamVODFinalizer(f LivestreamVODFinalizer) { state.livestreamVODFinalizer = f}
func (state *StatefulDB) processFinalizeLivestreamVODTask(ctx context.Context, task *AppTask) error { ctx = log.WithLogValues(ctx, "func", "processFinalizeLivestreamVODTask") var t FinalizeLivestreamVODTask if err := json.Unmarshal(task.Payload, &t); err != nil { return err } // Thread the upload + owner into the log context so every downstream log // line (and the error returned below, re-logged by runQueueWorker) carries // the user. ctx = log.WithLogValues(ctx, "uploadId", t.UploadID, "did", t.RepoDID, "livestream", t.LivestreamURI) log.Log(ctx, "dequeued finalize-livestream-vod task") if state.livestreamVODFinalizer == nil { log.Warn(ctx, "no livestream VOD finalizer configured; dropping task", "uploadId", t.UploadID, "did", t.RepoDID) return state.CompleteTask(ctx, task.ID) } if err := state.SetUploadProcessing(ctx, t.UploadID); err != nil { log.Warn(ctx, "failed to mark upload as processing", "uploadId", t.UploadID, "error", err) } cid, err := state.livestreamVODFinalizer(ctx, t) if err != nil { if ferr := state.SetUploadFailed(ctx, t.UploadID, err.Error()); ferr != nil { log.Warn(ctx, "failed to mark upload as failed", "uploadId", t.UploadID, "error", ferr) } // Flip the tied draft to 'error' so the user sees the failure. if derr := state.SetDraftError(ctx, t.UploadID, err.Error()); derr != nil { log.Warn(ctx, "failed to mark draft as failed", "uploadId", t.UploadID, "error", derr) } // Complete so it doesn't retry: most finalize failures are permanent // (missing objects, unreadable bytes, no OAuth session). _ = state.CompleteTask(ctx, task.ID) return fmt.Errorf("finalize livestream VOD upload %s: %w", t.UploadID, err) } // As with VODProcess, the finalizer calls SetUploadProcessed internally, so // re-read the finished Upload row and flip the tied draft to 'ready'. if err := state.markDraftReadyFromUpload(ctx, t.UploadID); err != nil { log.Warn(ctx, "failed to mark draft ready", "uploadId", t.UploadID, "error", err) } log.Log(ctx, "livestream VOD finalized", "uploadId", t.UploadID, "cid", cid) if t.Publish != nil { // The VOD is finalized either way, so a failed publish is not // retried (a retry would finalize the recording again). It leaves // the streamer a ready draft instead, to publish from the Drafts tab; // the publish remembers any track records it got as far as minting on // the upload, and the draft's publish reuses them. var perr error if state.videoPublisher == nil { perr = errors.New("no video publisher configured") } else if uri, vcid, err := state.videoPublisher(ctx, t); err != nil { perr = err } else { log.Log(ctx, "livestream VOD published", "uploadId", t.UploadID, "uri", uri, "cid", vcid) } if perr != nil { log.Error(ctx, "finalize-livestream-vod: VOD finalized but publishing the video record failed; leaving a draft", "uploadId", t.UploadID, "error", perr) if _, err := state.CreateLivestreamDraft(ctx, t.RepoDID, t.UploadID, t.Publish); err != nil { log.Error(ctx, "finalize-livestream-vod: could not create a draft for the unpublished VOD", "uploadId", t.UploadID, "error", err) } else if err := state.markDraftReadyFromUpload(ctx, t.UploadID); err != nil { log.Warn(ctx, "failed to mark draft ready", "uploadId", t.UploadID, "error", err) } } } return state.CompleteTask(ctx, task.ID)}
// ViewCountAggregator runs the view-log → place.stream.media.viewCount// aggregation for one window. Same function-pointer indirection trick// as VODProcessor: pkg/statedb stays ignorant of pkg/viewlog (which// pulls in blob storage + atproto publishing).type ViewCountAggregator func(ctx context.Context, t ViewCountAggregateTask) error
func (state *StatefulDB) SetViewCountAggregator(f ViewCountAggregator) { state.viewCountAggregator = f}
func (state *StatefulDB) processViewCountAggregateTask(ctx context.Context, task *AppTask) error { ctx = log.WithLogValues(ctx, "func", "processViewCountAggregateTask") var t ViewCountAggregateTask if err := json.Unmarshal(task.Payload, &t); err != nil { return err } if state.viewCountAggregator == nil { log.Warn(ctx, "no view-count aggregator configured; dropping task", "windowStart", t.WindowStart, "windowEnd", t.WindowEnd) return state.CompleteTask(ctx, task.ID) } if err := state.viewCountAggregator(ctx, t); err != nil { return fmt.Errorf("view-count aggregation: %w", err) } return state.CompleteTask(ctx, task.ID)}
// CDNLogIngester pulls newly archived CDN access logs into the view-log// store. Same indirection as ViewCountAggregator.type CDNLogIngester func(ctx context.Context) error
func (state *StatefulDB) SetCDNLogIngester(f CDNLogIngester) { state.cdnLogIngester = f}
func (state *StatefulDB) processCDNLogIngestTask(ctx context.Context, task *AppTask) error { ctx = log.WithLogValues(ctx, "func", "processCDNLogIngestTask") if state.cdnLogIngester == nil { log.Warn(ctx, "no cdn log ingester configured; dropping task") return state.CompleteTask(ctx, task.ID) } if err := state.cdnLogIngester(ctx); err != nil { return fmt.Errorf("cdn log ingest: %w", err) } return state.CompleteTask(ctx, task.ID)}
func (state *StatefulDB) processFinalizeLivestreamTask(ctx context.Context, task *AppTask) error { ctx = log.WithLogValues(ctx, "func", "processFinalizeLivestreamTask") log.Debug(ctx, "processing finalize livestream task") var finalizeLivestreamTask FinalizeLivestreamTask if err := json.Unmarshal(task.Payload, &finalizeLivestreamTask); err != nil { return err } livestream, err := state.model.GetLivestream(finalizeLivestreamTask.LivestreamURI) if err != nil { return fmt.Errorf("failed to get latest livestream for userDID: %w", err) } if livestream == nil { return fmt.Errorf("no livestream found for URI: %s", finalizeLivestreamTask.LivestreamURI) } lastLivestreamView, err := livestream.ToLivestreamView() if err != nil { return fmt.Errorf("failed to convert livestream to streamplace livestream: %w", err) } rec, ok := lastLivestreamView.Record.Val.(*placestream.Livestream) if !ok { return fmt.Errorf("livestream is not a streamplace livestream") } if rec.LastSeenAt == nil { return fmt.Errorf("livestream has no last seen at") } lastSeenTime, err := time.Parse(time.RFC3339, *rec.LastSeenAt) if err != nil { return fmt.Errorf("could not parse last seen at: %w", err) } if rec.IdleTimeoutSeconds == nil || *rec.IdleTimeoutSeconds == 0 { log.Debug(ctx, "livestream has no idle timeout, skipping finalization", "uri", livestream.URI) return nil } if time.Since(lastSeenTime) < (time.Duration(*rec.IdleTimeoutSeconds) * time.Second) { log.Debug(ctx, "livestream is active, skipping finalization", "lastSeenAt", lastSeenTime) return nil } // If this record is still the streamer's latest livestream, do NOT end it // on a stale lastSeenAt alone. lastSeenAt only advances via the per-segment // heartbeat (StreamSession.doUpdateLivestream), which is coupled to segment // arrival and can lag behind actual ingestion — e.g. after an ingest gap // long enough to tear down and restart the StreamSession, the new session's // heartbeat may not land on this record before the idle timer fires. Ending // here would set endedAt on the record the active stream is publishing // under, taking the stream pre-live underneath a still-flowing ingest. // // Instead, reschedule the check for one more idle window: if the stream is // truly abandoned the heartbeat stays frozen and we end it on the next // pass; if it's a heartbeat-lag artifact, the heartbeat catches up and the // rescheduled task hits the "active" early-return above. // // BUT the heartbeat-lag guard only applies to repos this node is actively // ingesting — the heartbeat runs inside StreamSession.doUpdateLivestream, // which requires the streamer to have a local OAuth session on this node. // If no session exists, there is no heartbeat to wait for and no write // access to end the record anyway: the record arrived via firehose sync // from an account that never connected here. Rescheduling in that case just // respawns the task every idle window forever (the rescheduled key embeds a // fresh timestamp, so dedup never fires), flooding the logs and growing the // task table without bound. So drop the task instead. latest, err := state.model.GetLatestLivestreamForRepo(livestream.RepoDID) if err != nil { return fmt.Errorf("failed to get latest livestream for repo: %w", err) } if latest != nil && latest.URI == livestream.URI { // Check for a local session before rescheduling. GetSessionByDID // returns gorm.ErrRecordNotFound (or nil session via callers that // swallow it) when the repo has never logged in here. session, err := state.GetSessionByDID(livestream.RepoDID) if errors.Is(err, gorm.ErrRecordNotFound) || (err == nil && session == nil) { log.Debug(ctx, "stale latest livestream has no local session; dropping finalize task (firehose-observed, no heartbeat to wait for)", "uri", livestream.URI, "lastSeenAt", lastSeenTime) return state.CompleteTask(ctx, task.ID) } if err != nil { return fmt.Errorf("failed to get session for finalize-livestream guard: %w", err) } if heartbeatFrozen(finalizeLivestreamTask, *rec.LastSeenAt) { log.Log(ctx, "livestream is latest for repo and its heartbeat has not moved in a full idle window; ending it", "uri", livestream.URI, "lastSeenAt", lastSeenTime) } else { rescheduledAt := time.Now().Add(time.Duration(*rec.IdleTimeoutSeconds) * time.Second).UTC() rescheduledKey := fmt.Sprintf("finalize-livestream::%s::%s", livestream.URI, rescheduledAt.Format(util.ISO8601)) next := finalizeLivestreamTask next.StaleSince = *rec.LastSeenAt _, err = state.EnqueueTask(ctx, TaskFinalizeLivestream, next, WithTaskKey(rescheduledKey), WithScheduledAt(rescheduledAt)) if err != nil { return fmt.Errorf("failed to reschedule finalize livestream task: %w", err) } log.Log(ctx, "livestream is latest for repo but lastSeenAt is stale; rescheduling finalize once to let heartbeat catch up", "uri", livestream.URI, "lastSeenAt", lastSeenTime, "rescheduledAt", rescheduledAt) return nil } } return state.EndLivestreamRecord(ctx, livestream, rec)}
// EndLivestreamRecord sets endedAt (= lastSeenAt) on a livestream record in// the streamer's repo, with the streamer's stored OAuth session. Used by the// idle finalize task and by the operator's finalize route for a record the// streamer never stopped.func (state *StatefulDB) EndLivestreamRecord(ctx context.Context, livestream *model.Livestream, rec *placestream.Livestream) error { session, err := state.GetSessionByDID(livestream.RepoDID) if err != nil { return fmt.Errorf("failed to get session: %w", err) } session, err = state.OATProxy.RefreshIfNeeded(session) if err != nil { return fmt.Errorf("failed to refresh session: %w", err) } client, err := state.OATProxy.GetXrpcClient(session) if err != nil { return fmt.Errorf("failed to get xrpc client: %w", err) } if rec.EndedAt != nil { log.Debug(ctx, "livestream has already ended, skipping", "uri", livestream.URI, "endedAt", *rec.EndedAt) return nil }
uri, err := syntax.ParseATURI(livestream.URI) if err != nil { return fmt.Errorf("failed to parse ATURI: %w", err) }
rec.EndedAt = rec.LastSeenAt
inp := comatproto.RepoPutRecord_Input{ Collection: "place.stream.livestream", Record: &glex.LexiconTypeDecoder{Val: rec}, Rkey: uri.RecordKey().String(), Repo: livestream.RepoDID, SwapRecord: &livestream.CID, } out := comatproto.RepoPutRecord_Output{}
err = client.Do(ctx, xrpc.Procedure, "application/json", "com.atproto.repo.putRecord", map[string]any{}, inp, &out) if err != nil { return fmt.Errorf("failed to update livestream record: %w", err) }
log.Log(ctx, "livestream finalized", "uri", livestream.URI, "endedAt", *rec.EndedAt)
return nil}
func (state *StatefulDB) processNotificationTask(ctx context.Context, task *AppTask) error { var notificationTask NotificationTask if err := json.Unmarshal(task.Payload, ¬ificationTask); err != nil { return err } lsv := notificationTask.Livestream rec, err := glex.RecordAs[placestream.Livestream](lsv.Record.Val) if err != nil { return fmt.Errorf("invalid livestream record: %w", err) } userDID := lsv.Author.Did
log.Warn(ctx, "Livestream detected! Blasting followers!", "title", rec.Title, "url", rec.Url, "createdAt", rec.CreatedAt, "repo", userDID) followers, err := state.model.GetUserFollowers(ctx, userDID) if err != nil { return err }
followersDIDs := make([]string, 0, len(followers)) for _, follower := range followers { followersDIDs = append(followersDIDs, follower.UserDID) }
log.Log(ctx, "found followers", "count", len(followersDIDs))
notifications, err := state.GetManyNotifications(followersDIDs) if err != nil { return err }
if state.noter != nil { user := model.HandleOrDID(lsv.Author.Handle, userDID) name := user if model.IsUsableHandle(lsv.Author.Handle) { name = "@" + name } nb := ¬ificationpkg.NotificationBlast{ Title: fmt.Sprintf("🔴 %s is LIVE!", name), Body: rec.Title, Data: map[string]string{ "path": fmt.Sprintf("/%s", user), }, } targets := make([]notificationpkg.NotificationTarget, len(notifications)) for i, n := range notifications { targets[i] = notificationpkg.NotificationTarget{Token: n.Token, Type: n.Type} } err = state.noter.Blast(ctx, targets, nb) if err != nil { log.Error(ctx, "failed to blast notifications", "err", err) } else { log.Log(ctx, "sent notifications", "user", userDID, "count", len(notifications), "content", nb) } // Prune web push subscriptions whose endpoints returned 410 Gone / // 404 — they're dead and would just fail again on every future blast. for _, token := range notificationpkg.ExpiredTokens(err) { if delErr := state.DeleteNotification(token); delErr != nil { log.Error(ctx, "failed to prune expired notification", "token", token, "err", delErr) } else { log.Log(ctx, "pruned expired notification", "token", token) } } } else { log.Log(ctx, "no notifier configured, skipping notifications", "user", userDID, "count", len(notifications)) }
// Send to webhooks using webhook manager webhooks, err := state.GetActiveWebhooksForUser(userDID, "livestream") if err != nil { log.Error(ctx, "failed to get livestream webhooks", "err", err) } else { for _, w := range webhooks { lexiconWebhook, err := w.ToLexicon() if err != nil { log.Error(ctx, "failed to convert webhook to lexicon", "err", err, "webhook_id", w.ID) continue } go func(lexiconWebhook placestream.ServerDefs_Webhook, wid string) { err := webhook.SendLivestreamWebhook(ctx, &lexiconWebhook, notificationTask.PDSURL, &lsv, notificationTask.FeedPost, ¬ificationTask.ChatProfile) if err != nil { log.Error(ctx, "failed to send livestream to webhook", "err", err, "webhook_id", wid) err := state.IncrementWebhookError(wid) if err != nil { log.Error(ctx, "failed to increment webhook error count", "err", err, "webhook_id", wid) } } else { log.Log(ctx, "sent livestream to webhook", "webhook_id", wid) err := state.ResetWebhookError(wid) if err != nil { log.Error(ctx, "failed to reset webhook error count", "err", err, "webhook_id", wid) } } }(lexiconWebhook, w.ID) } } return nil}
func (state *StatefulDB) processStreamReceivedTask(ctx context.Context, task *AppTask) error { var streamReceivedTask StreamReceivedTask if err := json.Unmarshal(task.Payload, &streamReceivedTask); err != nil { return err }
webhooks, err := state.GetActiveWebhooksForUser(streamReceivedTask.StreamerDID, "stream.received") if err != nil { return fmt.Errorf("failed to get stream.received webhooks: %w", err) } for _, w := range webhooks { lexiconWebhook, err := w.ToLexicon() if err != nil { log.Error(ctx, "failed to convert webhook to lexicon", "err", err, "webhook_id", w.ID) continue } go func(lexiconWebhook placestream.ServerDefs_Webhook, wid string) { err := webhook.SendStreamReceivedWebhook(ctx, &lexiconWebhook, streamReceivedTask.StreamerDID) if err != nil { log.Error(ctx, "failed to send stream.received webhook", "err", err, "webhook_id", wid) err = state.IncrementWebhookError(wid) if err != nil { log.Error(ctx, "failed to increment webhook error count", "err", err, "webhook_id", wid) } } else { log.Log(ctx, "sent stream.received webhook", "webhook_id", wid) err = state.ResetWebhookError(wid) if err != nil { log.Error(ctx, "failed to reset webhook error count", "err", err, "webhook_id", wid) } } }(lexiconWebhook, w.ID) } return nil}
func (state *StatefulDB) processChatMessageTask(ctx context.Context, task *AppTask) error { var chatTask ChatTask if err := json.Unmarshal(task.Payload, &chatTask); err != nil { return err } scm := chatTask.MessageView rec, err := glex.RecordAs[placestream.ChatMessage](scm.Record.Val) if err != nil { return fmt.Errorf("invalid chat message record: %w", err) }
// Send to webhooks using webhook manager webhooks, err := state.GetActiveWebhooksForUser(rec.Streamer, "chat") if err != nil { log.Error(ctx, "failed to get chat webhooks", "err", err) } else { for _, w := range webhooks { lexiconWebhook, err := w.ToLexicon() if err != nil { log.Error(ctx, "failed to convert webhook to lexicon", "err", err, "webhook_id", w.ID) continue } go func(lexiconWebhook placestream.ServerDefs_Webhook, wid string) { err := webhook.SendChatWebhook(ctx, &lexiconWebhook, scm.Author.Did, &scm) if err != nil { log.Error(ctx, "failed to send chat to webhook", "err", err, "webhook_id", wid) err = state.IncrementWebhookError(wid) if err != nil { log.Error(ctx, "failed to increment webhook error count", "err", err, "webhook_id", wid) } } else { log.Log(ctx, "sent chat to webhook", "webhook_id", wid) err = state.ResetWebhookError(wid) if err != nil { log.Error(ctx, "failed to reset webhook error count", "err", err, "webhook_id", wid) } } }(lexiconWebhook, w.ID) } } return nil}