Live video on the AT Protocol
stream.place
1package atproto
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "reflect"
8 "time"
9
10 "github.com/bluesky-social/indigo/api/bsky"
11 "github.com/bluesky-social/indigo/atproto/data"
12 "github.com/bluesky-social/indigo/atproto/syntax"
13 "stream.place/streamplace/pkg/aqtime"
14 "stream.place/streamplace/pkg/integrations/discord"
15 "stream.place/streamplace/pkg/log"
16 "stream.place/streamplace/pkg/model"
17 notificationpkg "stream.place/streamplace/pkg/notifications"
18 "stream.place/streamplace/pkg/streamplace"
19
20 lexutil "github.com/bluesky-social/indigo/lex/util"
21)
22
23func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userDID string, rkey syntax.RecordKey, recCBOR *[]byte, cid string, collection syntax.NSID, isUpdate bool, isFirstSync bool) error {
24 ctx = log.WithLogValues(ctx, "func", "handleCreateUpdate", "userDID", userDID, "rkey", rkey.String(), "cid", cid, "collection", collection.String())
25 now := time.Now()
26 r, err := atsync.Model.GetRepo(userDID)
27 if err != nil {
28 return fmt.Errorf("failed to get repo: %w", err)
29 }
30 maybeATURI := fmt.Sprintf("at://%s/%s/%s", userDID, collection.String(), rkey.String())
31 aturi, err := syntax.ParseATURI(maybeATURI)
32 if err != nil {
33 return fmt.Errorf("failed to parse ATURI: %w", err)
34 }
35 d, err := data.UnmarshalCBOR(*recCBOR)
36 if err != nil {
37 return fmt.Errorf("failed to unmarhsal record CBOR: %w", err)
38 }
39 cb, err := lexutil.CborDecodeValue(*recCBOR)
40 if errors.Is(err, lexutil.ErrUnrecognizedType) {
41 log.Debug(ctx, "unrecognized record type", "key", rkey.String(), "type", err)
42 return nil
43 } else if err != nil {
44 return fmt.Errorf("failed to decode record CBOR: %w", err)
45 }
46 switch rec := cb.(type) {
47 case *bsky.GraphFollow:
48 if r == nil {
49 // someone we don't know about
50 return nil
51 }
52 log.Debug(ctx, "creating follow", "userDID", userDID, "subjectDID", rec.Subject)
53 err := atsync.Model.CreateFollow(ctx, userDID, rkey.String(), rec)
54 if err != nil {
55 log.Debug(ctx, "failed to create follow", "err", err)
56 }
57
58 case *bsky.GraphBlock:
59 if r == nil {
60 // someone we don't know about
61 return nil
62 }
63 log.Debug(ctx, "creating block", "userDID", userDID, "subjectDID", rec.Subject)
64 block := &model.Block{
65 RKey: rkey.String(),
66 RepoDID: userDID,
67 SubjectDID: rec.Subject,
68 Record: *recCBOR,
69 CID: cid,
70 }
71 err := atsync.Model.CreateBlock(ctx, block)
72 if err != nil {
73 return fmt.Errorf("failed to create block: %w", err)
74 }
75 block, err = atsync.Model.GetBlock(ctx, rkey.String())
76 if err != nil {
77 return fmt.Errorf("failed to get block after we just saved it?!: %w", err)
78 }
79 streamplaceBlock, err := block.ToStreamplaceBlock()
80 if err != nil {
81 return fmt.Errorf("failed to convert block to streamplace block: %w", err)
82 }
83 go atsync.Bus.Publish(userDID, streamplaceBlock)
84
85 case *streamplace.ChatMessage:
86 repo, err := atsync.SyncBlueskyRepoCached(ctx, userDID, atsync.Model)
87 if err != nil {
88 return fmt.Errorf("failed to sync bluesky repo: %w", err)
89 }
90
91 _, err = atsync.SyncBlueskyRepoCached(ctx, rec.Streamer, atsync.Model)
92 if err != nil {
93 log.Error(ctx, "failed to sync bluesky repo", "err", err)
94 }
95
96 log.Debug(ctx, "streamplace.ChatMessage detected", "message", rec.Text, "repo", repo.Handle)
97 block, err := atsync.Model.GetUserBlock(ctx, rec.Streamer, userDID)
98 if err != nil {
99 return fmt.Errorf("failed to get user block: %w", err)
100 }
101 if block != nil {
102 log.Debug(ctx, "excluding message from blocked user", "userDID", userDID, "subjectDID", rec.Streamer)
103 return nil
104 }
105 mcm := &model.ChatMessage{
106 CID: cid,
107 URI: aturi.String(),
108 CreatedAt: now,
109 ChatMessage: recCBOR,
110 RepoDID: userDID,
111 Repo: repo,
112 StreamerRepoDID: rec.Streamer,
113 IndexedAt: &now,
114 }
115 if rec.Reply != nil && rec.Reply.Parent != nil && rec.Reply.Root != nil {
116 mcm.ReplyToCID = &rec.Reply.Parent.Cid
117 }
118 err = atsync.Model.CreateChatMessage(ctx, mcm)
119 if err != nil {
120 log.Error(ctx, "failed to create chat message", "err", err)
121 }
122 mcm, err = atsync.Model.GetChatMessage(cid)
123 if err != nil {
124 log.Error(ctx, "failed to get just-saved chat message", "err", err)
125 }
126 if mcm == nil {
127 log.Error(ctx, "failed to retrieve just-saved chat message", "err", err)
128 return nil
129 }
130 scm, err := mcm.ToStreamplaceMessageView()
131 if err != nil {
132 log.Error(ctx, "failed to convert chat message to streamplace message view", "err", err)
133 }
134 go atsync.Bus.Publish(rec.Streamer, scm)
135
136 if !isUpdate && !isFirstSync {
137 for _, webhook := range atsync.CLI.DiscordWebhooks {
138 if webhook.DID == rec.Streamer && webhook.Type == "chat" {
139 go func() {
140 err := discord.SendChat(ctx, webhook, r, scm)
141 if err != nil {
142 log.Error(ctx, "failed to send livestream to discord", "err", err)
143 } else {
144 log.Log(ctx, "sent livestream to discord", "user", userDID, "webhook", webhook.URL)
145 }
146 }()
147 }
148 }
149 }
150
151 case *streamplace.ChatProfile:
152 repo, err := atsync.SyncBlueskyRepoCached(ctx, userDID, atsync.Model)
153 if err != nil {
154 return fmt.Errorf("failed to sync bluesky repo: %w", err)
155 }
156 mcm := &model.ChatProfile{
157 RepoDID: userDID,
158 Repo: repo,
159 Record: recCBOR,
160 }
161 err = atsync.Model.CreateChatProfile(ctx, mcm)
162 if err != nil {
163 log.Error(ctx, "failed to create chat profile", "err", err)
164 }
165
166 case *bsky.FeedPost:
167 // jsonData, err := json.Marshal(d)
168 // if err != nil {
169 // log.Error(ctx, "failed to marshal record data", "err", err)
170 // } else {
171 // log.Log(ctx, "record data", "json", string(jsonData))
172 // }
173
174 createdAt, err := time.Parse(time.RFC3339, rec.CreatedAt)
175 if err != nil {
176 return fmt.Errorf("failed to parse createdAt: %w", err)
177 }
178
179 if livestream, ok := d["place.stream.livestream"]; ok {
180 repo, err := atsync.SyncBlueskyRepoCached(ctx, userDID, atsync.Model)
181 if err != nil {
182 return fmt.Errorf("failed to sync bluesky repo: %w", err)
183 }
184 livestream, ok := livestream.(map[string]interface{})
185 if !ok {
186 return fmt.Errorf("livestream is not a map")
187 }
188 url, ok := livestream["url"].(string)
189 if !ok {
190 return fmt.Errorf("livestream url is not a string")
191 }
192 log.Debug(ctx, "livestream url", "url", url)
193 if err := atsync.Model.CreateFeedPost(ctx, &model.FeedPost{
194 CID: cid,
195 CreatedAt: createdAt,
196 FeedPost: recCBOR,
197 RepoDID: userDID,
198 Repo: repo,
199 Type: "livestream",
200 URI: aturi.String(),
201 IndexedAt: &now,
202 }); err != nil {
203 return fmt.Errorf("failed to create bluesky post: %w", err)
204 }
205 } else {
206 if rec.Reply == nil || rec.Reply.Root == nil {
207 return nil
208 }
209 livestream, err := atsync.Model.GetLivestreamByPostCID(rec.Reply.Root.Cid)
210 if err != nil {
211 return fmt.Errorf("failed to get livestream: %w", err)
212 }
213 if livestream == nil {
214 return nil
215 }
216 // log.Warn(ctx, "chat message detected", "uri", livestream.URI)
217 // if this post is a reply to someone's livestream post
218 // log.Warn(ctx, "chat message detected", "message", rec.Text)
219 repo, err := atsync.SyncBlueskyRepoCached(ctx, userDID, atsync.Model)
220 if err != nil {
221 return fmt.Errorf("failed to sync bluesky repo: %w", err)
222 }
223
224 // log.Warn(ctx, "chat message detected", "message", rec.Text, "repo", repo.Handle)
225 block, err := atsync.Model.GetUserBlock(ctx, livestream.RepoDID, userDID)
226 if err != nil {
227 return fmt.Errorf("failed to get user block: %w", err)
228 }
229 if block != nil {
230 log.Warn(ctx, "excluding message from blocked user", "userDID", userDID, "subjectDID", livestream.RepoDID)
231 return nil
232 }
233 // if fc.cli.PrintChat {
234 // fmt.Printf("@%s%s %s\n", blue.Sprintf(repo.Handle), green.Sprintf(":"), rec.Text)
235 // }
236 fp := &model.FeedPost{
237 CID: cid,
238 CreatedAt: createdAt,
239 FeedPost: recCBOR,
240 RepoDID: userDID,
241 Type: "reply",
242 Repo: repo,
243 ReplyRootCID: &livestream.PostCID,
244 ReplyRootRepoDID: &livestream.RepoDID,
245 URI: aturi.String(),
246 IndexedAt: &now,
247 }
248 err = atsync.Model.CreateFeedPost(ctx, fp)
249 if err != nil {
250 log.Error(ctx, "failed to create feed post", "err", err)
251 }
252 postView, err := fp.ToBskyPostView()
253 if err != nil {
254 log.Error(ctx, "failed to convert feed post to bsky post view", "err", err)
255 }
256 go atsync.Bus.Publish(livestream.RepoDID, postView)
257 }
258
259 case *streamplace.Livestream:
260 var u string
261 if rec.Url != nil {
262 u = *rec.Url
263 }
264 if r == nil {
265 // we don't know about this repo
266 return nil
267 }
268 createdAt, err := time.Parse(time.RFC3339, rec.CreatedAt)
269 if err != nil {
270 log.Error(ctx, "failed to parse createdAt", "err", err)
271 return nil
272 }
273 ls := &model.Livestream{
274 CID: cid,
275 URI: aturi.String(),
276 CreatedAt: createdAt,
277 Livestream: recCBOR,
278 RepoDID: userDID,
279 }
280 if rec.Post != nil {
281 ls.PostCID = rec.Post.Cid
282 ls.PostURI = rec.Post.Uri
283 }
284 err = atsync.Model.CreateLivestream(ctx, ls)
285 if err != nil {
286 return fmt.Errorf("failed to create livestream: %w", err)
287 }
288 lsHydrated, err := atsync.Model.GetLatestLivestreamForRepo(userDID)
289 if err != nil {
290 return fmt.Errorf("failed to get latest livestream for repo: %w", err)
291 }
292 lsv, err := lsHydrated.ToLivestreamView()
293 if err != nil {
294 return fmt.Errorf("failed to convert livestream to bsky post view: %w", err)
295 }
296 go atsync.Bus.Publish(userDID, lsv)
297
298 if !isUpdate && !isFirstSync {
299 log.Warn(ctx, "Livestream detected! Blasting followers!", "title", rec.Title, "url", u, "createdAt", rec.CreatedAt, "repo", userDID)
300 notifications, err := atsync.Model.GetFollowersNotificationTokens(userDID)
301 if err != nil {
302 return err
303 }
304
305 nb := ¬ificationpkg.NotificationBlast{
306 Title: fmt.Sprintf("🔴 @%s is LIVE!", r.Handle),
307 Body: rec.Title,
308 Data: map[string]string{
309 "path": fmt.Sprintf("/%s", r.Handle),
310 },
311 }
312 if atsync.Noter != nil {
313 err := atsync.Noter.Blast(ctx, notifications, nb)
314 if err != nil {
315 log.Error(ctx, "failed to blast notifications", "err", err)
316 } else {
317 log.Log(ctx, "sent notifications", "user", userDID, "count", len(notifications), "content", nb)
318 }
319 } else {
320 log.Log(ctx, "no notifier configured, skipping notifications", "user", userDID, "count", len(notifications), "content", nb)
321 }
322
323 var postView *bsky.FeedDefs_PostView
324 if lsHydrated.Post != nil {
325 postView, err = lsHydrated.Post.ToBskyPostView()
326 if err != nil {
327 log.Error(ctx, "failed to convert livestream post to bsky post view", "err", err)
328 }
329 } else {
330 log.Warn(ctx, "no post found for livestream", "livestream", lsHydrated)
331 }
332
333 var spcp *streamplace.ChatProfile
334 cp, err := atsync.Model.GetChatProfile(ctx, userDID)
335 if err != nil {
336 log.Error(ctx, "failed to get chat profile", "err", err)
337 }
338 if cp != nil {
339 spcp, err = cp.ToStreamplaceChatProfile()
340 if err != nil {
341 log.Error(ctx, "failed to convert chat profile to streamplace chat profile", "err", err)
342 }
343 }
344
345 for _, webhook := range atsync.CLI.DiscordWebhooks {
346 if webhook.DID == userDID && webhook.Type == "livestream" {
347 go func() {
348 err := discord.SendLivestream(ctx, webhook, r, lsv, postView, spcp)
349 if err != nil {
350 log.Error(ctx, "failed to send livestream to discord", "err", err)
351 } else {
352 log.Log(ctx, "sent livestream to discord", "user", userDID, "webhook", webhook.URL)
353 }
354 }()
355 }
356 }
357 }
358
359 case *streamplace.Key:
360 log.Debug(ctx, "creating key", "key", rec)
361 time, err := aqtime.FromString(rec.CreatedAt)
362 if err != nil {
363 return fmt.Errorf("failed to parse createdAt: %w", err)
364 }
365 key := model.SigningKey{
366 DID: rec.SigningKey,
367 RKey: rkey.String(),
368 CreatedAt: time.Time(),
369 RepoDID: userDID,
370 }
371 err = atsync.Model.UpdateSigningKey(&key)
372 if err != nil {
373 log.Error(ctx, "failed to create signing key", "err", err)
374 }
375
376 default:
377 log.Debug(ctx, "unhandled record type", "type", reflect.TypeOf(rec))
378 }
379 return nil
380}