Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
43 kB · 1225 lines
Go
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226package atproto
import ( "context" "encoding/json" "errors" "fmt" "reflect" "strings" "time"
"github.com/bluesky-social/indigo/atproto/atdata" "github.com/bluesky-social/indigo/atproto/syntax" "github.com/bluesky-social/indigo/util" "stream.place/streamplace/pkg/appbsky" "stream.place/streamplace/pkg/aqtime" "stream.place/streamplace/pkg/constants" "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/statedb"
glex "github.com/streamplace/glex/runtime")
// chatLiveWindow is how recently a chat message must have been written to be// broadcast to live consumers (the chat websocket, the notification task).// Anything older is history -- a deepen window, a backfill, a firehose replay// of a span this node missed -- that belongs in the index but not on screen as// if it were arriving right now. Generous enough that ordinary client clock// skew does not eat a genuinely live message.const chatLiveWindow = 2 * time.Minute
// The commit revision orders permission updates and deletions on the client.type moderationPermissionViewEvent struct { placestream.ModerationDefs_PermissionView RepoRev string `json:"repoRev,omitempty"`}
// handleCreateUpdate indexes one record. It is called at least once per record// -- firehose cursor replay, a backfill walk restarting against a new head, and// the same commit arriving from several relays all deliver records we already// have -- so every write it makes has to be idempotent, and every side effect// (bus fanout, notification tasks) has to be skipped when nothing changed. The// model layer signals that with [model.ErrAlreadyIndexed]; see pkg/model/indexed.go.func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userDID string, rkey syntax.RecordKey, recCBOR *[]byte, cid string, collection syntax.NSID, isUpdate bool, isFirstSync bool, repoRev string) error { if _, err := atsync.indexingDID(ctx, userDID); err != nil { return err } ctx = log.WithLogValues(ctx, "func", "handleCreateUpdate", "userDID", userDID, "rkey", rkey.String(), "cid", cid, "collection", collection.String()) now := time.Now() r, err := atsync.Model.GetRepo(userDID) if err != nil { return fmt.Errorf("failed to get repo: %w", err) } maybeATURI := fmt.Sprintf("at://%s/%s/%s", userDID, collection.String(), rkey.String()) aturi, err := syntax.ParseATURI(maybeATURI) if err != nil { return fmt.Errorf("failed to parse ATURI: %w", err) } d, err := atdata.UnmarshalCBOR(*recCBOR) if err != nil { return fmt.Errorf("failed to unmarhsal record CBOR: %w", err) } cb, err := glex.CborDecodeValue(*recCBOR) if errors.Is(err, glex.ErrUnrecognizedType) { log.Debug(ctx, "unrecognized record type", "key", rkey.String(), "type", err) return nil } else if err != nil { return fmt.Errorf("failed to decode record CBOR: %w", err) } switch rec := cb.(type) { case *appbsky.GraphFollow: if r == nil { // someone we don't know about return nil } log.Debug(ctx, "creating follow", "userDID", userDID, "subjectDID", rec.Subject) err := atsync.Model.CreateFollow(ctx, userDID, rkey.String(), *rec) if err != nil { log.Debug(ctx, "failed to create follow", "err", err) }
case *appbsky.GraphBlock: if r == nil { // someone we don't know about return nil } log.Debug(ctx, "creating block", "userDID", userDID, "subjectDID", rec.Subject) block := &model.Block{ RKey: rkey.String(), RepoDID: userDID, SubjectDID: rec.Subject, Record: *recCBOR, CID: cid, } err := atsync.Model.CreateBlock(ctx, block) if errors.Is(err, model.ErrAlreadyIndexed) { return nil } if err != nil { return fmt.Errorf("failed to create block: %w", err) } block, err = atsync.Model.GetBlock(ctx, rkey.String()) if err != nil || block == nil { return fmt.Errorf("failed to get block after we just saved it?!: %w", err) } streamplaceBlock, err := block.ToStreamplaceBlock() if err != nil { return fmt.Errorf("failed to convert block to streamplace block: %w", err) } go atsync.Bus.Publish(userDID, streamplaceBlock)
case *appbsky.GraphVerification: // Indexed from any repo, known or not: verifiers are few, and which // ones this node trusts is decided by branding after the fact. v := &model.Verification{ URI: aturi.String(), CID: cid, IssuerDID: userDID, SubjectDID: rec.Subject, Handle: rec.Handle, DisplayName: rec.DisplayName, } if created, err := aqtime.FromString(rec.CreatedAt); err == nil { v.CreatedAt = created.Time() } err := atsync.Model.CreateVerification(ctx, v) if errors.Is(err, model.ErrAlreadyIndexed) { return nil } if err != nil { return fmt.Errorf("failed to create verification: %w", err) } return nil
case *appbsky.ActorProfile: if r == nil { // someone we don't know about return nil } wasStreamplace, _ := d[constants.BlueskyProfileGoliveKey].(bool) err := atsync.Model.UpsertBskyProfile(ctx, aturi, *recCBOR, wasStreamplace) if err != nil { return fmt.Errorf("failed to upsert bsky profile: %w", err) }
case *placestream.ChatMessage: repo, err := atsync.SyncBlueskyRepoCached(ctx, userDID) if err != nil { return fmt.Errorf("failed to sync bluesky repo: %w", err) }
go func() { // Its own err on purpose: assigning the enclosing function's err // from this goroutine races every later use of it. if _, err := atsync.SyncBlueskyRepoCached(ctx, rec.Streamer); err != nil { log.Error(ctx, "failed to sync bluesky repo", "err", err) } }()
log.Debug(ctx, "placestream.ChatMessage detected", "message", rec.Text, "repo", repo.Handle) block, err := atsync.Model.GetUserBlock(ctx, rec.Streamer, userDID) if err != nil { return fmt.Errorf("failed to get user block: %w", err) } if block != nil { log.Debug(ctx, "excluding message from blocked user", "userDID", userDID, "subjectDID", rec.Streamer) return nil } // created is this message's position in chat order, and the client's // own createdAt is trusted only backwards. An honest live message // keeps its stamp; a backfilled message from June lands in June, // instead of at the top of the hydration window just because a walk // indexed it today; and a stamp from the future is clamped to now, so // nobody pins a message to the bottom of a channel by post-dating it. created := now if aqt, err := aqtime.FromString(rec.CreatedAt); err == nil && aqt.Time().Before(now) { created = aqt.Time() } mcm := &model.ChatMessage{ CID: cid, URI: aturi.String(), CreatedAt: created, ChatMessage: recCBOR, RepoDID: userDID, Repo: repo, StreamerRepoDID: rec.Streamer, IndexedAt: &now, } if rec.Reply != nil && rec.Reply.Parent.Uri != "" && rec.Reply.Root.Uri != "" { mcm.ReplyToCID = &rec.Reply.Parent.Cid }
// check if we have any link facets with 'javascript:' links for _, facet := range rec.Facets { for _, feature := range facet.Features { if link := feature.RichtextFacet_Link; link != nil { if link.Uri != "" && strings.HasPrefix(strings.ToLower(link.Uri), "javascript:") { log.Warn(ctx, "excluding message with javascript: link", "uri", aturi.String(), "link", link.Uri) return nil } } } }
err = atsync.Model.CreateChatMessage(ctx, mcm) if errors.Is(err, model.ErrAlreadyIndexed) { // Already in the index: this is a cursor replay or a re-walk, not a // new message. Publishing it again would show it in chat twice. log.Debug(ctx, "skipping redelivered chat message", "uri", aturi.String()) return nil } if err != nil { log.Error(ctx, "failed to create chat message", "err", err) return nil } // Everything below builds the message view for live consumers -- the // chat websocket and the notification task -- and only messages that // are actually live belong there. Both indexing paths deliver // history: walks by construction (backfill, deepen, repair), and the // firehose whenever it replays a span this node missed. Spraying that // at an open chat renders hours of scroll as if it were arriving // right now. The message's own timestamp is the test, rather than // which path carried it, because a brand-new user's first message is // indexed by the very walk that message triggers -- the live event // then finds it already indexed and stays quiet, so a walked-but- // fresh message must still broadcast. if aqt, err := aqtime.FromString(rec.CreatedAt); err != nil || time.Since(aqt.Time()) > chatLiveWindow { return nil } mcm, err = atsync.Model.GetChatMessage(aturi.String()) if err != nil { log.Error(ctx, "failed to get just-saved chat message", "err", err) return nil } if mcm == nil { log.Error(ctx, "failed to retrieve just-saved chat message", "err", err) return nil } scm, err := mcm.ToStreamplaceMessageView() if err != nil { log.Error(ctx, "failed to convert chat message to streamplace message view", "err", err) return nil }
// Add mod badge if the author is a moderator issuerDID := fmt.Sprintf("did:web:%s", atsync.CLI.BroadcasterHost) err = AddModBadgeIfApplicable(ctx, scm, rec.Streamer, issuerDID, atsync.Model) if err != nil { log.Error(ctx, "failed to add mod badge", "err", err) }
if scm.Author.Handle == "" || scm.Author.Handle == "handle.invalid" { scm.Author.Handle = atsync.ResolveAuthorHandle(ctx, scm.Author.Did) } atsync.DecorateVerification(ctx, rec.Streamer, scm) if !atsync.ChatAllowed(ctx, rec.Streamer, scm.Author.Did) { // The streamer's chat access rules refuse this author: the message // is indexed (the rules may change) but not shown. return nil }
go atsync.Bus.Publish(rec.Streamer, scm)
if !isUpdate {
task := &statedb.ChatTask{ MessageView: *scm, }
_, err = atsync.StatefulDB.EnqueueTask(ctx, statedb.TaskChat, task, statedb.WithTaskKey(fmt.Sprintf("chat-message::%s", aturi.String()))) if err != nil { log.Error(ctx, "failed to enqueue notification task", "err", err) } }
case *placestream.ChatGate: repo, err := atsync.SyncBlueskyRepoCached(ctx, userDID) if err != nil { return fmt.Errorf("failed to sync bluesky repo: %w", err) } if r == nil { // someone we don't know about return nil } log.Debug(ctx, "creating gate", "userDID", userDID, "hiddenMessage", rec.HiddenMessage) gate := &model.Gate{ RKey: rkey.String(), RepoDID: userDID, HiddenMessage: rec.HiddenMessage, CID: cid, CreatedAt: now, Repo: repo, } err = atsync.Model.CreateGate(ctx, gate) if errors.Is(err, model.ErrAlreadyIndexed) { return nil } if err != nil { return fmt.Errorf("failed to create gate: %w", err) } gate, err = atsync.Model.GetGate(ctx, rkey.String()) if err != nil { return fmt.Errorf("failed to get gate after we just saved it?!: %w", err) } streamplaceGate, err := gate.ToStreamplaceGate() if err != nil { return fmt.Errorf("failed to convert gate to streamplace gate: %w", err) } go atsync.Bus.Publish(userDID, streamplaceGate)
case *placestream.ChatPinnedRecord: repo, err := atsync.SyncBlueskyRepoCached(ctx, userDID) if err != nil { return fmt.Errorf("failed to sync bluesky repo: %w", err) } if r == nil { return nil } log.Debug(ctx, "creating pinned record", "userDID", userDID, "pinnedMessage", rec.PinnedMessage) // err = atsync.Model.DeleteAllPinnedRecords(ctx, userDID) // if err != nil { // log.Error(ctx, "failed to delete existing pinned records", "err", err) // } // Parse optional expiresAt var expiresAt *time.Time if rec.ExpiresAt != nil { t, err := time.Parse(time.RFC3339, *rec.ExpiresAt) if err == nil { expiresAt = &t } } // serialise createdAt createdAt, err := time.Parse(time.RFC3339, rec.CreatedAt) if err != nil { return fmt.Errorf("failed to parse createdAt: %w", err) }
var pinnedBy string if rec.PinnedBy == nil { pinnedBy = userDID } else { pinnedBy = *rec.PinnedBy }
pin := &model.PinnedRecord{ Uri: aturi.String(), RepoDID: userDID, PinnedMessage: rec.PinnedMessage, PinnedBy: pinnedBy, IndexedAt: &now, CID: cid, CreatedAt: createdAt, Repo: repo, ExpiresAt: expiresAt, } err = atsync.Model.CreatePinnedRecord(ctx, pin) if errors.Is(err, model.ErrAlreadyIndexed) { return nil } if err != nil { return fmt.Errorf("failed to create pinned record: %w", err) } pin, err = atsync.Model.GetPinnedRecord(ctx, pin.Uri) if err != nil { return fmt.Errorf("failed to get pinned record after we just saved it: %w", err) } pinnedView, err := pin.ToStreamplacePinnedRecordView() if err != nil { return fmt.Errorf("failed to convert pinned record: %w", err) } // look up the original message, pinner msg, err := atsync.Model.GetChatMessage(pinnedView.Record.PinnedMessage) if err != nil { return fmt.Errorf("failed to get chat message: %w", err) } profile, err := atsync.Model.GetChatProfile(ctx, pinnedBy) if err != nil { return fmt.Errorf("failed to get chat profile: %w", err) } if msg != nil { msgView, err := msg.ToStreamplaceMessageView() if err != nil { return fmt.Errorf("failed to convert chat message: %w", err) } pinnedView.Message = msgView } if profile != nil { profileView, err := profile.ToStreamplaceChatProfile() if err != nil { return fmt.Errorf("failed to convert chat profile: %w", err) } pinnedView.PinnedBy = &profileView } go atsync.Bus.Publish(userDID, pinnedView)
case *placestream.ChatProfile: repo, err := atsync.SyncBlueskyRepoCached(ctx, userDID) if err != nil { return fmt.Errorf("failed to sync bluesky repo: %w", err) } mcm := &model.ChatProfile{ RepoDID: userDID, Repo: repo, Record: recCBOR, } err = atsync.Model.CreateChatProfile(ctx, mcm) if err != nil { log.Error(ctx, "failed to create chat profile", "err", err) }
case *placestream.ServerSettings: _, err := atsync.SyncBlueskyRepoCached(ctx, userDID) if err != nil { return fmt.Errorf("failed to sync bluesky repo: %w", err) } settings := &model.ServerSettings{ Server: rkey.String(), RepoDID: userDID, Record: recCBOR, } err = atsync.Model.UpdateServerSettings(ctx, settings) if err != nil { log.Error(ctx, "failed to create server settings", "err", err) }
case *appbsky.FeedPost: // jsonData, err := json.Marshal(d) // if err != nil { // log.Error(ctx, "failed to marshal record data", "err", err) // } else { // log.Log(ctx, "record data", "json", string(jsonData)) // }
createdAt, err := time.Parse(time.RFC3339, rec.CreatedAt) if err != nil { return fmt.Errorf("failed to parse createdAt: %w", err) }
if livestream, ok := d["place.stream.livestream"]; ok { repo, err := atsync.SyncBlueskyRepoCached(ctx, userDID) if err != nil { return fmt.Errorf("failed to sync bluesky repo: %w", err) } livestream, ok := livestream.(map[string]interface{}) if !ok { return fmt.Errorf("livestream is not a map") } url, ok := livestream["url"].(string) if !ok { return fmt.Errorf("livestream url is not a string") } log.Debug(ctx, "livestream url", "url", url) err = atsync.Model.CreateFeedPost(ctx, &model.FeedPost{ CID: cid, CreatedAt: createdAt, FeedPost: recCBOR, RepoDID: userDID, Repo: repo, Type: "livestream", URI: aturi.String(), IndexedAt: &now, }) if err != nil && !errors.Is(err, model.ErrAlreadyIndexed) { return fmt.Errorf("failed to create bluesky post: %w", err) } } else { if rec.Reply == nil || rec.Reply.Root.Uri == "" { return nil } livestream, err := atsync.Model.GetLivestreamByPostURI(rec.Reply.Root.Uri) if err != nil { return fmt.Errorf("failed to get livestream: %w", err) } if livestream == nil { return nil } // log.Warn(ctx, "chat message detected", "uri", livestream.URI) // if this post is a reply to someone's livestream post // log.Warn(ctx, "chat message detected", "message", rec.Text) repo, err := atsync.SyncBlueskyRepoCached(ctx, userDID) if err != nil { return fmt.Errorf("failed to sync bluesky repo: %w", err) }
// log.Warn(ctx, "chat message detected", "message", rec.Text, "repo", repo.Handle) block, err := atsync.Model.GetUserBlock(ctx, livestream.RepoDID, userDID) if err != nil { return fmt.Errorf("failed to get user block: %w", err) } if block != nil { log.Warn(ctx, "excluding message from blocked user", "userDID", userDID, "subjectDID", livestream.RepoDID) return nil } // if fc.cli.PrintChat { // fmt.Printf("@%s%s %s\n", blue.Sprintf(repo.Handle), green.Sprintf(":"), rec.Text) // } fp := &model.FeedPost{ CID: cid, CreatedAt: createdAt, FeedPost: recCBOR, RepoDID: userDID, Type: "reply", Repo: repo, ReplyRootURI: &livestream.PostURI, ReplyRootRepoDID: &livestream.RepoDID, URI: aturi.String(), IndexedAt: &now, } err = atsync.Model.CreateFeedPost(ctx, fp) if errors.Is(err, model.ErrAlreadyIndexed) { // A reply we already have: the bus already saw it. return nil } if err != nil { log.Error(ctx, "failed to create feed post", "err", err) } postView, err := fp.ToBskyPostView() if err != nil { log.Error(ctx, "failed to convert feed post to bsky post view", "err", err) } go atsync.Bus.Publish(livestream.RepoDID, postView) }
case *placestream.Livestream: if r == nil { // we don't know about this repo return nil } createdAt, err := time.Parse(time.RFC3339, rec.CreatedAt) if err != nil { log.Error(ctx, "failed to parse createdAt", "err", err) return nil } ls := &model.Livestream{ CID: cid, URI: aturi.String(), CreatedAt: createdAt, Livestream: recCBOR, RepoDID: userDID, } if rec.Post != nil { ls.PostCID = rec.Post.Cid ls.PostURI = rec.Post.Uri } // Whether this version is the node seeing the livestream end: the // version it had indexed, if any, had not ended. Read before indexing // replaces it. seenEnding := false if rec.EndedAt != nil && !isFirstSync { prev, err := atsync.Model.GetLivestream(aturi.String()) if err != nil { return fmt.Errorf("failed to get livestream: %w", err) } seenEnding = prev == nil if prev != nil { prevView, err := prev.ToLivestreamView() if err != nil { return fmt.Errorf("failed to decode indexed livestream: %w", err) } prevRec, ok := prevView.Record.Val.(*placestream.Livestream) seenEnding = ok && prevRec.EndedAt == nil } } err = atsync.Model.CreateLivestream(ctx, ls) if errors.Is(err, model.ErrAlreadyIndexed) { // Re-announcing an unchanged livestream would light the red circle // up again and re-queue its finalize task. return nil } if err != nil { return fmt.Errorf("failed to create livestream: %w", err) } lsHydrated, err := atsync.Model.GetLatestLivestreamForRepo(userDID) if err != nil { return fmt.Errorf("failed to get latest livestream for repo: %w", err) } lsv, err := lsHydrated.ToLivestreamView() if err != nil { return fmt.Errorf("failed to convert livestream to bsky post view: %w", err) } go atsync.Bus.Publish(userDID, lsv)
if !isFirstSync { if atsync.CLI.StreamIsAllowed(userDID) != nil { // they're live somewhere but they don't have nothin' to do with us return nil } if seenEnding { // Publish its VOD if the streamer has that on now. Only on // seeing it end, never for an ended record seen again (a // redelivery, an edit, a backfilled one), so turning it on // later doesn't publish an old recording. if err := atsync.StatefulDB.ScheduleAutoPublishVOD(ctx, userDID, aturi.String()); err != nil { return fmt.Errorf("failed to schedule automatic VOD publishing: %w", err) } return nil } log.Debug(ctx, "stream is allowed, queuing finalize task") // queue a task to clean up the livestream if it's been inactive for too long task := &statedb.FinalizeLivestreamTask{ LivestreamURI: aturi.String(), } if rec.LastSeenAt == nil || rec.IdleTimeoutSeconds == nil || *rec.IdleTimeoutSeconds == 0 || rec.EndedAt != nil { return nil } scheduledAt, err := time.Parse(time.RFC3339, *rec.LastSeenAt) if err != nil { log.Error(ctx, "failed to parse last seen at", "err", err) return nil }
// if we check after exactly rec.IdleTimeoutSeconds we might miss the finalization by a few seconds scheduledAt = scheduledAt.Add((time.Duration(*rec.IdleTimeoutSeconds) * time.Second) + (10 * time.Second)).UTC() taskKey := fmt.Sprintf("finalize-livestream::%s::%s", aturi.String(), scheduledAt.Format(util.ISO8601)) _, err = atsync.StatefulDB.EnqueueTask(ctx, statedb.TaskFinalizeLivestream, task, statedb.WithTaskKey(taskKey), statedb.WithScheduledAt(scheduledAt)) if err != nil { return fmt.Errorf("failed to enqueue remove red circle task: %w", err) }
}
case *placestream.LiveTeleport: if r == nil { return nil } startsAt, err := time.Parse(time.RFC3339, rec.StartsAt) if err != nil { log.Error(ctx, "failed to parse startsAt", "err", err) return nil } viewerCount := atsync.Bus.GetViewerCount(userDID) tp := &model.Teleport{ CID: cid, URI: aturi.String(), StartsAt: startsAt, DurationSeconds: rec.DurationSeconds, ViewerCount: int64(viewerCount), Teleport: recCBOR, RepoDID: userDID, TargetDID: rec.Streamer, } err = atsync.Model.CreateTeleport(ctx, tp) if errors.Is(err, model.ErrAlreadyIndexed) { // Otherwise every redelivery schedules another arrival notification. return nil } if err != nil { return fmt.Errorf("failed to create teleport: %w", err) } // Relay reconnects can replay history; only tolerate brief delivery delays. if startsAt.Before(now.Add(-30 * time.Second)) { return nil } teleportMessage := map[string]any{ "$type": rec.RecordTypeID(), "uri": aturi.String(), "cid": cid, "streamer": rec.Streamer, "startsAt": rec.StartsAt, } if rec.DurationSeconds != nil { teleportMessage["durationSeconds"] = *rec.DurationSeconds } if rec.Livestream != nil { teleportMessage["livestream"] = rec.Livestream } // Backfills restore pending countdowns; live events may be slightly late. if !isFirstSync || startsAt.After(now) { atsync.Bus.Publish(userDID, teleportMessage) }
if isFirstSync { // A backfill reads history, and a teleport out of history has // already happened: announcing it would tell a streamer somebody is // arriving who arrived last year. The record is indexed either way; // only the announcement is a live-only thing. return nil }
// schedule arrival notification 10 seconds after startsAt arrivalTime := startsAt.Add(10 * time.Second) waitDuration := time.Until(arrivalTime) if waitDuration < 0 { waitDuration = 0 }
time.AfterFunc(waitDuration, func() { // verify teleport still exists existingTp, err := atsync.Model.GetTeleportByURI(aturi.String()) if err != nil { log.Error(ctx, "failed to get teleport by uri", "err", err) return } if existingTp == nil || existingTp.Denied { log.Debug(ctx, "teleport no longer active, skipping arrival notification", "uri", aturi.String()) return }
// get the source profile sourceRepo, err := atsync.Model.GetRepo(userDID) if err != nil { log.Error(ctx, "failed to get source repo", "err", err) return }
viewerCount := existingTp.ViewerCount
arrivalMsg := placestream.Livestream_TeleportArrival{ LexiconTypeID: "place.stream.livestream#teleportArrival", TeleportUri: aturi.String(), Source: appbsky.ActorDefs_ProfileViewBasic{ Did: userDID, Handle: sourceRepo.Handle, }, ViewerCount: int64(viewerCount), StartsAt: rec.StartsAt, }
// get the source chat profile chatProfile, err := atsync.Model.GetChatProfile(ctx, userDID) if err == nil && chatProfile != nil { spcp, err := chatProfile.ToStreamplaceChatProfile() if err == nil { arrivalMsg.ChatProfile = &spcp } }
atsync.Bus.Publish(rec.Streamer, arrivalMsg)
// A teleport is our version of a "raid": it sends the source // streamer's viewers to the target. Unlike a raid, though, it // previously left the source stream live — so viewers could just // navigate back. End the source streamer's livestream here (the // same record update place.stream.live.stopLivestream performs, // setting endedAt so the streamer returns to "pre-live"), now // that viewers have been sent over. The exact stream to end is // pinned by the teleport record's `livestream` strongRef, so a // newer stream the streamer may have started in the meantime is // never terminated by mistake. Best-effort: a failure only logs // and never blocks the arrival notification. atsync.endLivestreamForTeleport(ctx, userDID, rec.Livestream) })
case *placestream.Key: log.Debug(ctx, "creating key", "key", rec) time, err := aqtime.FromString(rec.CreatedAt) if err != nil { return fmt.Errorf("failed to parse createdAt: %w", err) } key := model.SigningKey{ DID: rec.SigningKey, RKey: rkey.String(), CreatedAt: time.Time(), RepoDID: userDID, } err = atsync.Model.UpdateSigningKey(&key) if err != nil { log.Error(ctx, "failed to create signing key", "err", err) }
case *placestream.BroadcastOrigin: repo, err := atsync.SyncBlueskyRepoCached(ctx, userDID) if err != nil { return fmt.Errorf("failed to sync broadcast origin creator bluesky repo: %w", err) } _, err = atsync.SyncBlueskyRepoCached(ctx, rec.Streamer) if err != nil { return fmt.Errorf("failed to sync broadcast origin streamer bluesky repo: %w", err) } err = atsync.Model.UpdateBroadcastOrigin(ctx, *rec, aturi) if err != nil { log.Error(ctx, "failed to update broadcast origin", "err", err) } log.Log(ctx, "indexed broadcast origin", "streamer", rec.Streamer, "server", rec.Server) view := placestream.BroadcastDefs_BroadcastOriginView{ Uri: aturi.String(), Cid: cid, Author: appbsky.ActorDefs_ProfileViewBasic{ Did: userDID, Handle: repo.Handle, }, Record: &glex.LexiconTypeDecoder{Val: rec}, } // publishes with an empty string because we're discovering the stream; // as a pointer, which is what the replicator's subscription asserts go atsync.Bus.Publish("", &view)
case *placestream.MetadataConfiguration: repo, err := atsync.SyncBlueskyRepoCached(ctx, userDID) if err != nil { return fmt.Errorf("failed to sync bluesky repo: %w", err) } log.Debug(ctx, "creating metadata configuration", "metadata", rec) metadata := &model.MetadataConfiguration{ RepoDID: userDID, Record: recCBOR, Repo: repo, } err = atsync.Model.CreateMetadataConfiguration(ctx, metadata) if err != nil { log.Error(ctx, "failed to create metadata configuration", "err", err) }
case *placestream.ChatAccess: if _, err := atsync.SyncBlueskyRepoCached(ctx, userDID); err != nil { return fmt.Errorf("failed to sync bluesky repo: %w", err) } row, err := model.ChatAccessRuleFromRecord(rec, aturi) if errors.Is(err, model.ErrChatAccessSubjectUnknown) { log.Warn(ctx, "chat access rule with an unknown subject type, skipping", "uri", aturi.String()) return nil } if err != nil { return fmt.Errorf("invalid chat access rule: %w", err) } err = atsync.Model.CreateChatAccessRule(ctx, row) if errors.Is(err, model.ErrAlreadyIndexed) { return nil } if err != nil { return fmt.Errorf("failed to index chat access rule: %w", err) } atsync.NoteChatAccessRule(ctx, row) log.Log(ctx, "indexed chat access rule", "streamer", userDID, "action", row.Action, "subject", row.SubjectType, "did", row.SubjectDID)
case *placestream.ModerationPermission: repo, err := atsync.SyncBlueskyRepoCached(ctx, userDID) if err != nil { return fmt.Errorf("failed to sync bluesky repo: %w", err) } log.Debug(ctx, "creating moderation delegation", "streamerDID", userDID, "moderatorDID", rec.Moderator)
err = atsync.Model.CreateModerationDelegation(ctx, *rec, aturi) if errors.Is(err, model.ErrAlreadyIndexed) { return nil } if err != nil { return fmt.Errorf("failed to create moderation delegation: %w", err) }
view := placestream.ModerationDefs_PermissionView{ LexiconTypeID: "place.stream.moderation.defs#permissionView", Uri: aturi.String(), Cid: cid, Author: appbsky.ActorDefs_ProfileViewBasic{ Did: userDID, Handle: repo.Handle, }, Record: &glex.LexiconTypeDecoder{Val: rec}, } // Publish moderation permission view to WebSocket bus for real-time updates // This allows moderators to see their permissions instantly without page refresh go atsync.Bus.Publish(userDID, moderationPermissionViewEvent{ ModerationDefs_PermissionView: view, RepoRev: repoRev, })
case *placestream.LiveViewerCount: log.Debug(ctx, "indexing view count", "streamer", rec.Streamer, "server", rec.Server, "count", rec.Count) // Our own record loops back through our own firehose; indexing it // would stack the federated copy of our local count on top of the // live one, double-counting every local viewer. if rec.Server == atsync.CLI.ServerDID() { break } // Check if the reporting server's DID is labeled as banned or !no-viewers serverLabels, err := atsync.Model.GetActiveLabels(rec.Server) if err != nil { log.Error(ctx, "failed to get labels for server", "server", rec.Server, "error", err) } else if IsViewerBanned(serverLabels...) { log.Warn(ctx, "discarding view count from labeled server", "server", rec.Server) break } atsync.Bus.SetFederatedViewCount(rec.Streamer, rec.Server, int(rec.Count))
case *placestream.LiveRecommendations: log.Debug(ctx, "creating recommendations", "userDID", userDID, "count", len(rec.Streamers))
// Validate max 8 streamers if len(rec.Streamers) > 8 { log.Warn(ctx, "recommendations exceed maximum of 8", "count", len(rec.Streamers)) return fmt.Errorf("maximum 8 recommendations allowed, got %d", len(rec.Streamers)) }
// Marshal streamers to JSON streamersJSON, err := json.Marshal(rec.Streamers) if err != nil { return fmt.Errorf("failed to marshal streamers: %w", err) }
// Parse createdAt timestamp createdAt, err := time.Parse(time.RFC3339, rec.CreatedAt) if err != nil { return fmt.Errorf("failed to parse createdAt: %w", err) }
recommendation := &model.Recommendation{ UserDID: userDID, Streamers: json.RawMessage(streamersJSON), CreatedAt: createdAt, }
err = atsync.Model.UpsertRecommendation(recommendation) if err != nil { return fmt.Errorf("failed to upsert recommendation: %w", err) }
case *placestream.BadgeDef: def := &model.BadgeDef{ URI: aturi.String(), CID: cid, RepoDID: userDID, RKey: rkey.String(), Name: rec.Name, BadgeType: rec.BadgeType, Record: *recCBOR, IndexedAt: now, } if rec.Description != nil { def.Description = *rec.Description } if rec.Image != nil { def.ImageCID = rec.Image.Ref.String() def.ImageMimeType = rec.Image.MimeType } if err := atsync.Model.UpsertBadgeDef(ctx, def); err != nil { return fmt.Errorf("failed to upsert badge def: %w", err) } log.Debug(ctx, "indexed badge def", "uri", aturi.String(), "name", rec.Name)
case *placestream.BadgeIssuance: issuance := &model.BadgeIssuance{ URI: aturi.String(), CID: cid, RepoDID: userDID, RKey: rkey.String(), RecipientDID: rec.Did, BadgeURI: rec.Badge.Uri, Record: *recCBOR, IndexedAt: now, } if err := atsync.Model.UpsertBadgeIssuance(ctx, issuance); err != nil { return fmt.Errorf("failed to upsert badge issuance: %w", err) } log.Debug(ctx, "indexed badge issuance", "uri", aturi.String(), "recipient", rec.Did)
case *placestream.Video: _, err := atsync.SyncBlueskyRepoCached(ctx, userDID) if err != nil { return fmt.Errorf("failed to sync bluesky repo: %w", err) } if err := atsync.Model.UpsertVideo(ctx, *rec, aturi); err != nil { return fmt.Errorf("failed to upsert video: %w", err) } log.Debug(ctx, "indexed video", "uri", aturi.String(), "title", rec.Title)
case *placestream.MediaTrack: // Tracks not backed by a muxlTrack (we don't define any other // shape yet) are skipped with a warning — there'd be no blob // to key the row off of. if rec.Track.MediaDefs_MuxlTrack == nil { log.Warn(ctx, "track record missing muxlTrack; skipping", "uri", aturi.String()) return nil } if err := atsync.Model.UpsertMediaTrack(ctx, *rec, aturi); err != nil { return fmt.Errorf("failed to upsert media track: %w", err) } mt := rec.Track.MediaDefs_MuxlTrack log.Debug(ctx, "indexed media track", "uri", aturi.String(), "blob", mt.Blob, "mediaType", mt.MediaType)
case *placestream.MediaOrigin: // Origin records are published by streamplace nodes (not users) // against their own server-repo DID. The aturi's authority is // the publishing server. if err := atsync.Model.UpsertMediaOrigin(ctx, *rec, aturi); err != nil { return fmt.Errorf("failed to upsert media origin: %w", err) } log.Debug(ctx, "indexed media origin", "uri", aturi.String(), "blob", rec.Blob, "server", userDID)
case *placestream.BetaInvite: // Invite records grant a specific account access to a named // beta feature. We index all of them as they fly past; gate // callers filter by RepoDID to a single operator-configured // issuer (the `--beta-invite-did` flag), so anyone else // minting these records is harmless noise. if err := atsync.Model.UpsertBetaInvite(ctx, *rec, aturi); err != nil { return fmt.Errorf("failed to upsert beta invite: %w", err) } log.Debug(ctx, "indexed beta invite", "uri", aturi.String(), "did", rec.Did, "feature", rec.Feature)
// Notify the invited account that they're off the waitlist — but // only for a genuinely new invite arriving live from the trusted // issuer. Backfill/first-sync and record updates re-index existing // invites on every restart and must not re-notify. if !isFirstSync && !isUpdate && atsync.CLI.BetaInviteDID != "" && userDID == atsync.CLI.BetaInviteDID { atsync.notifyBetaInvite(ctx, rec) }
case *placestream.BetaRequest: // Access requests are published by users in their own repos. We // index them so operators can see who's waiting and so // place.stream.beta.getStatus can report "requested". if err := atsync.Model.UpsertBetaRequest(ctx, *rec, aturi); err != nil { return fmt.Errorf("failed to upsert beta request: %w", err) } log.Debug(ctx, "indexed beta request", "uri", aturi.String(), "did", userDID, "feature", rec.Feature)
case *placestream.MediaViewCount: // View-count records are published by streamplace nodes (in // their server repos) reporting on traffic they observed. // Multiple reporters publish records for the same video; the // query layer (place.stream.media.getVideo) sums across them. if err := atsync.Model.UpsertMediaViewCount(ctx, *rec, aturi); err != nil { return fmt.Errorf("failed to upsert media view count: %w", err) } log.Debug(ctx, "indexed media view count", "uri", aturi.String(), "video", rec.Video, "count", rec.Count, "reporter", userDID)
case *placestream.VodComment: repo, err := atsync.SyncBlueskyRepoCached(ctx, userDID) if err != nil { return fmt.Errorf("failed to sync bluesky repo: %w", err) }
log.Debug(ctx, "place.stream.vod.comment detected", "video", rec.Video, "repo", repo.Handle)
// Check if the video author has blocked the commenter videoATURI, parseErr := syntax.ParseATURI(rec.Video) var videoAuthor string if parseErr == nil { videoAuthor = videoATURI.Authority().String() block, err := atsync.Model.GetUserBlock(ctx, videoAuthor, userDID) if err != nil { log.Warn(ctx, "failed to check user block for VOD comment", "err", err) } else if block != nil { log.Debug(ctx, "excluding VOD comment from blocked user", "userDID", userDID, "videoAuthor", videoAuthor) return nil } } else { log.Warn(ctx, "failed to parse video URI for block check", "video", rec.Video, "err", err) }
vc := &model.VodComment{ CID: cid, URI: aturi.String(), CreatedAt: now, Comment: recCBOR, RepoDID: userDID, Repo: repo, VideoURI: rec.Video, VideoAuthorDID: videoAuthor, IndexedAt: &now, } if rec.Reply != nil && rec.Reply.Parent.Uri != "" && rec.Reply.Root.Uri != "" { vc.ReplyToCID = &rec.Reply.Parent.Cid }
// check for javascript: links in facets for _, facet := range rec.Facets { for _, feature := range facet.Features { if link := feature.RichtextFacet_Link; link != nil { if link.Uri != "" && strings.HasPrefix(strings.ToLower(link.Uri), "javascript:") { log.Warn(ctx, "excluding comment with javascript: link", "uri", aturi.String(), "link", link.Uri) return nil } } } }
err = atsync.Model.CreateVodComment(ctx, vc) if errors.Is(err, model.ErrAlreadyIndexed) { log.Debug(ctx, "skipping redelivered VOD comment", "uri", aturi.String()) return nil } if err != nil { log.Error(ctx, "failed to create VOD comment", "err", err) return nil } vc, err = atsync.Model.GetVodComment(aturi.String()) if err != nil { log.Error(ctx, "failed to get just-saved VOD comment", "err", err) return nil } if vc == nil { log.Error(ctx, "failed to retrieve just-saved VOD comment") return nil } sc, err := vc.ToStreamplaceCommentView() if err != nil { log.Error(ctx, "failed to convert VOD comment to view", "err", err) return nil }
if sc.Author.Handle == "" || sc.Author.Handle == "handle.invalid" { sc.Author.Handle = atsync.ResolveAuthorHandle(ctx, sc.Author.Did) }
if videoAuthor != "" { go atsync.Bus.Publish(videoAuthor, sc) } else { go atsync.Bus.Publish(userDID, sc) }
case *placestream.Like: repo, err := atsync.SyncBlueskyRepoCached(ctx, userDID) if err != nil { return fmt.Errorf("failed to sync bluesky repo: %w", err) }
log.Debug(ctx, "place.stream.like detected", "subject", rec.Subject, "repo", repo.Handle)
// A user can only like a subject once — refuse to index a duplicate // rather than inflating the count with a second row. existing, err := atsync.Model.GetLikeBySubjectAndUser(ctx, rec.Subject, userDID) if err != nil { return fmt.Errorf("check existing like: %w", err) } if existing != nil { log.Debug(ctx, "ignoring duplicate like", "subject", rec.Subject, "repo", userDID) return nil }
like := &model.Like{ CID: cid, URI: aturi.String(), Subject: rec.Subject, RepoDID: userDID, Repo: repo, IndexedAt: &now, CreatedAt: now, } err = atsync.Model.CreateLike(ctx, like) if err != nil { log.Error(ctx, "failed to create VOD like", "err", err) return nil }
case *placestream.VodGate: repo, err := atsync.SyncBlueskyRepoCached(ctx, userDID) if err != nil { return fmt.Errorf("failed to sync bluesky repo: %w", err) } if r == nil { // someone we don't know about return nil } log.Debug(ctx, "creating VOD gate", "userDID", userDID, "hiddenComment", rec.HiddenComment) gate := &model.VodGate{ RKey: rkey.String(), RepoDID: userDID, HiddenComment: rec.HiddenComment, CID: cid, CreatedAt: now, Repo: repo, } err = atsync.Model.CreateVodGate(ctx, gate) if err != nil && !errors.Is(err, model.ErrAlreadyIndexed) { return fmt.Errorf("failed to create VOD gate: %w", err) }
default: log.Debug(ctx, "unhandled record type", "type", reflect.TypeOf(rec)) } return nil}
// notifyBetaInvite pushes a "you're off the waitlist" notification to the// account named by a freshly-issued, trusted beta invite. Best-effort: any// failure is logged, never returned, since the invite is already indexed and// the upload gate works regardless of whether the push lands.func (atsync *ATProtoSynchronizer) notifyBetaInvite(ctx context.Context, rec *placestream.BetaInvite) { if atsync.Noter == nil || atsync.StatefulDB == nil { return } notifications, err := atsync.StatefulDB.GetManyNotifications([]string{rec.Did}) if err != nil { log.Error(ctx, "beta invite notification: failed to load tokens", "did", rec.Did, "err", err) return } if len(notifications) == 0 { log.Debug(ctx, "beta invite notification: no device tokens for invitee", "did", rec.Did, "feature", rec.Feature) return } blast := betaInviteBlast(rec.Feature) targets := make([]notificationpkg.NotificationTarget, len(notifications)) for i, n := range notifications { targets[i] = notificationpkg.NotificationTarget{Token: n.Token, Type: n.Type} } if err := atsync.Noter.Blast(ctx, targets, blast); err != nil { log.Error(ctx, "beta invite notification: blast failed", "did", rec.Did, "feature", rec.Feature, "err", err) } else { log.Log(ctx, "sent beta invite notification", "did", rec.Did, "feature", rec.Feature, "tokens", len(notifications)) } // Prune dead web push subscriptions so they don't accumulate. for _, token := range notificationpkg.ExpiredTokens(err) { if delErr := atsync.StatefulDB.DeleteNotification(token); delErr != nil { log.Error(ctx, "beta invite notification: failed to prune expired", "token", token, "err", delErr) } }}
// betaInviteBlast builds the push payload for a newly-granted beta feature.// Copy is feature-aware where we have something specific to say.func betaInviteBlast(feature string) *notificationpkg.NotificationBlast { switch feature { case "vod": // Uploads are a web flow today, and pushes land on the native app, so // we route to home rather than a route the app doesn't register. return ¬ificationpkg.NotificationBlast{ Title: "🎉 You're off the waitlist!", Body: "You can now upload videos to Streamplace.", Data: map[string]string{"path": "/"}, } default: return ¬ificationpkg.NotificationBlast{ Title: "🎉 You're off the waitlist!", Body: fmt.Sprintf("You've been granted access to the %s beta.", feature), Data: map[string]string{"path": "/"}, } }}