Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
41 kB · 1172 lines
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173package 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 { 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 } 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 } 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) } go atsync.Bus.Publish(userDID, rec)
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 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": "/"}, } }}