stream.place
Live video on the AT Protocol
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, repo, 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 *streamplace.ServerSettings:
167 _, err := atsync.SyncBlueskyRepoCached(ctx, userDID, atsync.Model)
168 if err != nil {
169 return fmt.Errorf("failed to sync bluesky repo: %w", err)
170 }
171 settings := &model.ServerSettings{
172 Server: rkey.String(),
173 RepoDID: userDID,
174 Record: recCBOR,
175 }
176 err = atsync.Model.UpdateServerSettings(ctx, settings)
177 if err != nil {
178 log.Error(ctx, "failed to create server settings", "err", err)
179 }
180
181 case *bsky.FeedPost:
182 // jsonData, err := json.Marshal(d)
183 // if err != nil {
184 // log.Error(ctx, "failed to marshal record data", "err", err)
185 // } else {
186 // log.Log(ctx, "record data", "json", string(jsonData))
187 // }
188
189 createdAt, err := time.Parse(time.RFC3339, rec.CreatedAt)
190 if err != nil {
191 return fmt.Errorf("failed to parse createdAt: %w", err)
192 }
193
194 if livestream, ok := d["place.stream.livestream"]; ok {
195 repo, err := atsync.SyncBlueskyRepoCached(ctx, userDID, atsync.Model)
196 if err != nil {
197 return fmt.Errorf("failed to sync bluesky repo: %w", err)
198 }
199 livestream, ok := livestream.(map[string]interface{})
200 if !ok {
201 return fmt.Errorf("livestream is not a map")
202 }
203 url, ok := livestream["url"].(string)
204 if !ok {
205 return fmt.Errorf("livestream url is not a string")
206 }
207 log.Debug(ctx, "livestream url", "url", url)
208 if err := atsync.Model.CreateFeedPost(ctx, &model.FeedPost{
209 CID: cid,
210 CreatedAt: createdAt,
211 FeedPost: recCBOR,
212 RepoDID: userDID,
213 Repo: repo,
214 Type: "livestream",
215 URI: aturi.String(),
216 IndexedAt: &now,
217 }); err != nil {
218 return fmt.Errorf("failed to create bluesky post: %w", err)
219 }
220 } else {
221 if rec.Reply == nil || rec.Reply.Root == nil {
222 return nil
223 }
224 livestream, err := atsync.Model.GetLivestreamByPostCID(rec.Reply.Root.Cid)
225 if err != nil {
226 return fmt.Errorf("failed to get livestream: %w", err)
227 }
228 if livestream == nil {
229 return nil
230 }
231 // log.Warn(ctx, "chat message detected", "uri", livestream.URI)
232 // if this post is a reply to someone's livestream post
233 // log.Warn(ctx, "chat message detected", "message", rec.Text)
234 repo, err := atsync.SyncBlueskyRepoCached(ctx, userDID, atsync.Model)
235 if err != nil {
236 return fmt.Errorf("failed to sync bluesky repo: %w", err)
237 }
238
239 // log.Warn(ctx, "chat message detected", "message", rec.Text, "repo", repo.Handle)
240 block, err := atsync.Model.GetUserBlock(ctx, livestream.RepoDID, userDID)
241 if err != nil {
242 return fmt.Errorf("failed to get user block: %w", err)
243 }
244 if block != nil {
245 log.Warn(ctx, "excluding message from blocked user", "userDID", userDID, "subjectDID", livestream.RepoDID)
246 return nil
247 }
248 // if fc.cli.PrintChat {
249 // fmt.Printf("@%s%s %s\n", blue.Sprintf(repo.Handle), green.Sprintf(":"), rec.Text)
250 // }
251 fp := &model.FeedPost{
252 CID: cid,
253 CreatedAt: createdAt,
254 FeedPost: recCBOR,
255 RepoDID: userDID,
256 Type: "reply",
257 Repo: repo,
258 ReplyRootCID: &livestream.PostCID,
259 ReplyRootRepoDID: &livestream.RepoDID,
260 URI: aturi.String(),
261 IndexedAt: &now,
262 }
263 err = atsync.Model.CreateFeedPost(ctx, fp)
264 if err != nil {
265 log.Error(ctx, "failed to create feed post", "err", err)
266 }
267 postView, err := fp.ToBskyPostView()
268 if err != nil {
269 log.Error(ctx, "failed to convert feed post to bsky post view", "err", err)
270 }
271 go atsync.Bus.Publish(livestream.RepoDID, postView)
272 }
273
274 case *streamplace.Livestream:
275 var u string
276 if rec.Url != nil {
277 u = *rec.Url
278 }
279 if r == nil {
280 // we don't know about this repo
281 return nil
282 }
283 createdAt, err := time.Parse(time.RFC3339, rec.CreatedAt)
284 if err != nil {
285 log.Error(ctx, "failed to parse createdAt", "err", err)
286 return nil
287 }
288 ls := &model.Livestream{
289 CID: cid,
290 URI: aturi.String(),
291 CreatedAt: createdAt,
292 Livestream: recCBOR,
293 RepoDID: userDID,
294 }
295 if rec.Post != nil {
296 ls.PostCID = rec.Post.Cid
297 ls.PostURI = rec.Post.Uri
298 }
299 err = atsync.Model.CreateLivestream(ctx, ls)
300 if err != nil {
301 return fmt.Errorf("failed to create livestream: %w", err)
302 }
303 lsHydrated, err := atsync.Model.GetLatestLivestreamForRepo(userDID)
304 if err != nil {
305 return fmt.Errorf("failed to get latest livestream for repo: %w", err)
306 }
307 lsv, err := lsHydrated.ToLivestreamView()
308 if err != nil {
309 return fmt.Errorf("failed to convert livestream to bsky post view: %w", err)
310 }
311 go atsync.Bus.Publish(userDID, lsv)
312
313 if !isUpdate && !isFirstSync {
314 log.Warn(ctx, "Livestream detected! Blasting followers!", "title", rec.Title, "url", u, "createdAt", rec.CreatedAt, "repo", userDID)
315 notifications, err := atsync.Model.GetFollowersNotificationTokens(userDID)
316 if err != nil {
317 return err
318 }
319
320 nb := ¬ificationpkg.NotificationBlast{
321 Title: fmt.Sprintf("🔴 @%s is LIVE!", r.Handle),
322 Body: rec.Title,
323 Data: map[string]string{
324 "path": fmt.Sprintf("/%s", r.Handle),
325 },
326 }
327 if atsync.Noter != nil {
328 err := atsync.Noter.Blast(ctx, notifications, nb)
329 if err != nil {
330 log.Error(ctx, "failed to blast notifications", "err", err)
331 } else {
332 log.Log(ctx, "sent notifications", "user", userDID, "count", len(notifications), "content", nb)
333 }
334 } else {
335 log.Log(ctx, "no notifier configured, skipping notifications", "user", userDID, "count", len(notifications), "content", nb)
336 }
337
338 var postView *bsky.FeedDefs_PostView
339 if lsHydrated.Post != nil {
340 postView, err = lsHydrated.Post.ToBskyPostView()
341 if err != nil {
342 log.Error(ctx, "failed to convert livestream post to bsky post view", "err", err)
343 }
344 } else {
345 log.Warn(ctx, "no post found for livestream", "livestream", lsHydrated)
346 }
347
348 var spcp *streamplace.ChatProfile
349 cp, err := atsync.Model.GetChatProfile(ctx, userDID)
350 if err != nil {
351 log.Error(ctx, "failed to get chat profile", "err", err)
352 }
353 if cp != nil {
354 spcp, err = cp.ToStreamplaceChatProfile()
355 if err != nil {
356 log.Error(ctx, "failed to convert chat profile to streamplace chat profile", "err", err)
357 }
358 }
359
360 for _, webhook := range atsync.CLI.DiscordWebhooks {
361 if webhook.DID == userDID && webhook.Type == "livestream" {
362 go func() {
363 err := discord.SendLivestream(ctx, webhook, r, lsv, postView, spcp)
364 if err != nil {
365 log.Error(ctx, "failed to send livestream to discord", "err", err)
366 } else {
367 log.Log(ctx, "sent livestream to discord", "user", userDID, "webhook", webhook.URL)
368 }
369 }()
370 }
371 }
372 }
373
374 case *streamplace.Key:
375 log.Debug(ctx, "creating key", "key", rec)
376 time, err := aqtime.FromString(rec.CreatedAt)
377 if err != nil {
378 return fmt.Errorf("failed to parse createdAt: %w", err)
379 }
380 key := model.SigningKey{
381 DID: rec.SigningKey,
382 RKey: rkey.String(),
383 CreatedAt: time.Time(),
384 RepoDID: userDID,
385 }
386 err = atsync.Model.UpdateSigningKey(&key)
387 if err != nil {
388 log.Error(ctx, "failed to create signing key", "err", err)
389 }
390
391 default:
392 log.Debug(ctx, "unhandled record type", "type", reflect.TypeOf(rec))
393 }
394 return nil
395}