alice.pds.demo.boltless.dev
This repository has no description
1package db
2
3import (
4 "context"
5 "slices"
6
7 "github.com/bluesky-social/indigo/atproto/syntax"
8 "tangled.org/core/api/tangled"
9 "tangled.org/core/appview/db"
10 "tangled.org/core/appview/models"
11 "tangled.org/core/appview/notify"
12 "tangled.org/core/idresolver"
13 "tangled.org/core/log"
14 "tangled.org/core/orm"
15 "tangled.org/core/sets"
16)
17
18const (
19 maxMentions = 8
20 assigneeLabelAt = "at://did:plc:wshs7t2adsemcrrd4snkeqli/sh.tangled.label.definition/assignee"
21)
22
23type databaseNotifier struct {
24 db *db.DB
25 res *idresolver.Resolver
26}
27
28func NewDatabaseNotifier(database *db.DB, resolver *idresolver.Resolver) notify.Notifier {
29 return &databaseNotifier{
30 db: database,
31 res: resolver,
32 }
33}
34
35var _ notify.Notifier = &databaseNotifier{}
36
37func (n *databaseNotifier) NewRepo(ctx context.Context, repo *models.Repo) {
38 // no-op for now
39}
40func (n *databaseNotifier) DeleteRepo(ctx context.Context, repo *models.Repo) {
41 // no-op for now
42}
43
44func (n *databaseNotifier) RenameRepo(ctx context.Context, actor syntax.DID, oldRepo, newRepo *models.Repo) {
45}
46
47func (n *databaseNotifier) NewStar(ctx context.Context, star *models.Star) {
48 l := log.FromContext(ctx)
49
50 if star.SubjectType != models.StarSubjectRepo {
51 return
52 }
53
54 repo, err := db.GetRepo(n.db, orm.FilterEq("repo_did", star.Subject))
55 if err != nil {
56 l.Error("failed to get repos", "err", err)
57 return
58 }
59
60 actorDid := syntax.DID(star.Did)
61 recipients := sets.Singleton(syntax.DID(repo.Did))
62 eventType := models.NotificationTypeRepoStarred
63 entityType := "repo"
64 entityId := star.Subject
65 repoId := &repo.Id
66 var issueId *int64
67 var pullId *int64
68
69 n.notifyEvent(
70 ctx,
71 actorDid,
72 recipients,
73 eventType,
74 entityType,
75 entityId,
76 repoId,
77 issueId,
78 pullId,
79 )
80}
81
82func (n *databaseNotifier) DeleteStar(ctx context.Context, star *models.Star) {
83 // no-op
84}
85
86func (n *databaseNotifier) NewComment(ctx context.Context, comment *models.Comment, mentions []syntax.DID) {
87 l := log.FromContext(ctx)
88
89 var (
90 // built the recipients list:
91 // - the owner of the repo
92 // - | if the comment is a reply -> everybody on that thread
93 // | if the comment is a top level -> just the issue owner
94 // - remove mentioned users from the recipients list
95 recipients = sets.New[syntax.DID]()
96 entityType string
97 entityId string
98 repoId *int64
99 issueId *int64
100 pullId *int64
101 )
102
103 subjectAt := syntax.ATURI(comment.Subject.Uri)
104
105 switch subjectAt.Collection() {
106 case tangled.RepoIssueNSID:
107 issues, err := db.GetIssues(
108 n.db,
109 orm.FilterEq("at_uri", subjectAt),
110 )
111 if err != nil {
112 l.Error("failed to get issues", "err", err)
113 return
114 }
115 if len(issues) == 0 {
116 l.Error("no issue found", "subject", comment.Subject)
117 return
118 }
119 issue := issues[0]
120
121 recipients.Insert(syntax.DID(issue.Repo.Did))
122 if comment.IsReply() {
123 // if this comment is a reply, then notify everybody in that thread
124 parent := *comment.ReplyTo
125
126 // find the parent thread, and add all DIDs from here to the recipient list
127 for _, t := range models.NewCommentList(issue.Comments) {
128 if t.Self.AtUri() == syntax.ATURI(parent.Uri) {
129 for _, p := range t.Participants() {
130 recipients.Insert(p)
131 }
132 }
133 }
134 } else {
135 // not a reply, notify just the issue author
136 recipients.Insert(syntax.DID(issue.Did))
137 }
138
139 entityType = "issue"
140 entityId = issue.AtUri().String()
141 repoId = &issue.Repo.Id
142 issueId = &issue.Id
143
144 for _, m := range mentions {
145 recipients.Remove(m)
146 }
147
148 n.notifyEvent(
149 ctx,
150 comment.Did,
151 recipients,
152 models.NotificationTypeIssueCommented,
153 entityType,
154 entityId,
155 repoId,
156 issueId,
157 pullId,
158 )
159
160 case tangled.RepoPullNSID:
161 pull, err := db.GetPull(ctx, n.db,
162 orm.FilterEq("owner_did", subjectAt.Authority()),
163 orm.FilterEq("rkey", subjectAt.RecordKey()),
164 )
165 if err != nil {
166 l.Error("NewComment: failed to get pull", "err", err)
167 return
168 }
169
170 pull.Repo, err = db.GetRepo(n.db, orm.FilterEq("repo_did", pull.RepoDid))
171 if err != nil {
172 l.Error("NewComment: failed to get repo", "err", err)
173 return
174 }
175
176 recipients.Insert(syntax.DID(pull.Repo.Did))
177 for _, p := range pull.Participants() {
178 recipients.Insert(syntax.DID(p))
179 }
180
181 entityType = "pull"
182 entityId = pull.AtUri().String()
183 repoId = &pull.Repo.Id
184 p := int64(pull.ID)
185 pullId = &p
186
187 for _, m := range mentions {
188 recipients.Remove(m)
189 }
190
191 n.notifyEvent(
192 ctx,
193 comment.Did,
194 recipients,
195 models.NotificationTypePullCommented,
196 entityType,
197 entityId,
198 repoId,
199 issueId,
200 pullId,
201 )
202 default:
203 return // no-op
204 }
205
206 n.notifyEvent(
207 ctx,
208 comment.Did,
209 sets.Collect(slices.Values(mentions)),
210 models.NotificationTypeUserMentioned,
211 entityType,
212 entityId,
213 repoId,
214 issueId,
215 pullId,
216 )
217}
218
219func (n *databaseNotifier) DeleteComment(ctx context.Context, comment *models.Comment) {
220 // no-op
221}
222
223func (n *databaseNotifier) NewIssue(ctx context.Context, issue *models.Issue, mentions []syntax.DID) {
224 l := log.FromContext(ctx)
225
226 collaborators, err := db.GetCollaborators(n.db, orm.FilterEq("repo_did", string(issue.RepoDid)))
227 if err != nil {
228 l.Error("failed to fetch collaborators", "err", err)
229 return
230 }
231
232 // build the recipients list
233 // - owner of the repo
234 // - collaborators in the repo
235 // - remove users already mentioned
236 recipients := sets.Singleton(syntax.DID(issue.Repo.Did))
237 for _, c := range collaborators {
238 recipients.Insert(c.SubjectDid)
239 }
240 for _, m := range mentions {
241 recipients.Remove(m)
242 }
243
244 actorDid := syntax.DID(issue.Did)
245 entityType := "issue"
246 entityId := issue.AtUri().String()
247 repoId := &issue.Repo.Id
248 issueId := &issue.Id
249 var pullId *int64
250
251 n.notifyEvent(
252 ctx,
253 actorDid,
254 recipients,
255 models.NotificationTypeIssueCreated,
256 entityType,
257 entityId,
258 repoId,
259 issueId,
260 pullId,
261 )
262 n.notifyEvent(
263 ctx,
264 actorDid,
265 sets.Collect(slices.Values(mentions)),
266 models.NotificationTypeUserMentioned,
267 entityType,
268 entityId,
269 repoId,
270 issueId,
271 pullId,
272 )
273}
274
275func (n *databaseNotifier) DeleteIssue(ctx context.Context, issue *models.Issue) {
276 // no-op for now
277}
278
279func (n *databaseNotifier) NewIssueLabelOp(ctx context.Context, actor syntax.DID, issue *models.Issue, ops []models.LabelOp) {
280 entityType := "issue"
281 entityId := issue.AtUri().String()
282 repoId := &issue.Repo.Id
283 issueId := &issue.Id
284 var pullId *int64
285
286 assigned := sets.New[syntax.DID]()
287 unassigned := sets.New[syntax.DID]()
288 for _, op := range ops {
289 if op.OperandKey != assigneeLabelAt {
290 continue
291 }
292 assignee := syntax.DID(op.OperandValue)
293 switch op.Operation {
294 case models.LabelOperationAdd:
295 assigned.Insert(assignee)
296 case models.LabelOperationDel:
297 unassigned.Insert(assignee)
298 default:
299 continue
300 }
301 }
302
303 n.notifyEvent(ctx, actor, assigned, models.NotificationTypeIssueAssigned, entityType, entityId, repoId, issueId, pullId)
304 n.notifyEvent(ctx, actor, unassigned, models.NotificationTypeIssueUnassigned, entityType, entityId, repoId, issueId, pullId)
305}
306
307func (n *databaseNotifier) NewPullLabelOp(ctx context.Context, actor syntax.DID, pull *models.Pull, ops []models.LabelOp) {
308 entityType := "pull"
309 entityId := pull.AtUri().String()
310 repoId := &pull.Repo.Id
311 var issueId *int64
312 p := int64(pull.ID)
313 pullId := &p
314
315 assigned := sets.New[syntax.DID]()
316 unassigned := sets.New[syntax.DID]()
317 for _, op := range ops {
318 if op.OperandKey != assigneeLabelAt {
319 continue
320 }
321 assignee := syntax.DID(op.OperandValue)
322 switch op.Operation {
323 case models.LabelOperationAdd:
324 assigned.Insert(assignee)
325 case models.LabelOperationDel:
326 unassigned.Insert(assignee)
327 default:
328 continue
329 }
330 }
331
332 n.notifyEvent(ctx, actor, assigned, models.NotificationTypePullAssigned, entityType, entityId, repoId, issueId, pullId)
333 n.notifyEvent(ctx, actor, unassigned, models.NotificationTypePullUnassigned, entityType, entityId, repoId, issueId, pullId)
334}
335
336func (n *databaseNotifier) NewFollow(ctx context.Context, follow *models.Follow) {
337 actorDid := syntax.DID(follow.UserDid)
338 recipients := sets.Singleton(syntax.DID(follow.SubjectDid))
339 eventType := models.NotificationTypeFollowed
340 entityType := "follow"
341 entityId := follow.UserDid
342 var repoId, issueId, pullId *int64
343
344 n.notifyEvent(
345 ctx,
346 actorDid,
347 recipients,
348 eventType,
349 entityType,
350 entityId,
351 repoId,
352 issueId,
353 pullId,
354 )
355}
356
357func (n *databaseNotifier) DeleteFollow(ctx context.Context, follow *models.Follow) {
358 // no-op
359}
360
361func (n *databaseNotifier) NewPull(ctx context.Context, pull *models.Pull) {
362 l := log.FromContext(ctx)
363
364 repo, err := db.GetRepo(n.db, orm.FilterEq("repo_did", string(pull.RepoDid)))
365 if err != nil {
366 l.Error("failed to get repos", "err", err)
367 return
368 }
369 collaborators, err := db.GetCollaborators(n.db, orm.FilterEq("repo_did", string(pull.RepoDid)))
370 if err != nil {
371 l.Error("failed to fetch collaborators", "err", err)
372 return
373 }
374
375 // build the recipients list
376 // - owner of the repo
377 // - collaborators in the repo
378 recipients := sets.Singleton(syntax.DID(repo.Did))
379 for _, c := range collaborators {
380 recipients.Insert(c.SubjectDid)
381 }
382
383 actorDid := pull.OwnerDid
384 eventType := models.NotificationTypePullCreated
385 entityType := "pull"
386 entityId := pull.AtUri().String()
387 repoId := &repo.Id
388 var issueId *int64
389 p := int64(pull.ID)
390 pullId := &p
391
392 n.notifyEvent(
393 ctx,
394 actorDid,
395 recipients,
396 eventType,
397 entityType,
398 entityId,
399 repoId,
400 issueId,
401 pullId,
402 )
403}
404
405func (n *databaseNotifier) UpdateProfile(ctx context.Context, profile *models.Profile) {
406 // no-op
407}
408
409func (n *databaseNotifier) DeleteString(ctx context.Context, did, rkey string) {
410 // no-op
411}
412
413func (n *databaseNotifier) EditString(ctx context.Context, string *models.String) {
414 // no-op
415}
416
417func (n *databaseNotifier) NewString(ctx context.Context, string *models.String) {
418 // no-op
419}
420
421func (n *databaseNotifier) Push(ctx context.Context, repo *models.Repo, ref, oldSha, newSha, committerDid string) {
422 // no-op for now; webhooks are handled by the webhook notifier
423}
424
425func (n *databaseNotifier) Clone(ctx context.Context, repo *models.Repo) {
426 // no-op
427}
428
429func (n *databaseNotifier) NewIssueState(ctx context.Context, actor syntax.DID, issue *models.Issue) {
430 l := log.FromContext(ctx)
431
432 collaborators, err := db.GetCollaborators(n.db, orm.FilterEq("repo_did", string(issue.RepoDid)))
433 if err != nil {
434 l.Error("failed to fetch collaborators", "err", err)
435 return
436 }
437
438 // build up the recipients list:
439 // - repo owner
440 // - repo collaborators
441 // - all issue participants
442 recipients := sets.Singleton(syntax.DID(issue.Repo.Did))
443 for _, c := range collaborators {
444 recipients.Insert(c.SubjectDid)
445 }
446 for _, p := range issue.Participants() {
447 recipients.Insert(syntax.DID(p))
448 }
449
450 entityType := "issue"
451 entityId := issue.AtUri().String()
452 repoId := &issue.Repo.Id
453 issueId := &issue.Id
454 var pullId *int64
455 var eventType models.NotificationType
456
457 if issue.Open {
458 eventType = models.NotificationTypeIssueReopen
459 } else {
460 eventType = models.NotificationTypeIssueClosed
461 }
462
463 n.notifyEvent(
464 ctx,
465 actor,
466 recipients,
467 eventType,
468 entityType,
469 entityId,
470 repoId,
471 issueId,
472 pullId,
473 )
474}
475
476func (n *databaseNotifier) ResubmitPull(ctx context.Context, pull *models.Pull) {
477 // no-op for now
478}
479
480func (n *databaseNotifier) NewPullState(ctx context.Context, actor syntax.DID, pull *models.Pull) {
481 l := log.FromContext(ctx)
482
483 // Get repo details
484 repo, err := db.GetRepo(n.db, orm.FilterEq("repo_did", string(pull.RepoDid)))
485 if err != nil {
486 l.Error("failed to get repos", "err", err)
487 return
488 }
489
490 collaborators, err := db.GetCollaborators(n.db, orm.FilterEq("repo_did", string(pull.RepoDid)))
491 if err != nil {
492 l.Error("failed to fetch collaborators", "err", err)
493 return
494 }
495
496 // build up the recipients list:
497 // - repo owner
498 // - all pull participants
499 recipients := sets.Singleton(syntax.DID(repo.Did))
500 for _, c := range collaborators {
501 recipients.Insert(c.SubjectDid)
502 }
503 for _, p := range pull.Participants() {
504 recipients.Insert(p)
505 }
506
507 entityType := "pull"
508 entityId := pull.AtUri().String()
509 repoId := &repo.Id
510 var issueId *int64
511 var eventType models.NotificationType
512 switch pull.State {
513 case models.PullClosed:
514 eventType = models.NotificationTypePullClosed
515 case models.PullOpen:
516 eventType = models.NotificationTypePullReopen
517 case models.PullMerged:
518 eventType = models.NotificationTypePullMerged
519 default:
520 l.Error("unexpected new PR state", "state", pull.State)
521 return
522 }
523 p := int64(pull.ID)
524 pullId := &p
525
526 n.notifyEvent(
527 ctx,
528 actor,
529 recipients,
530 eventType,
531 entityType,
532 entityId,
533 repoId,
534 issueId,
535 pullId,
536 )
537}
538
539func (n *databaseNotifier) notifyEvent(
540 ctx context.Context,
541 actorDid syntax.DID,
542 recipients sets.Set[syntax.DID],
543 eventType models.NotificationType,
544 entityType string,
545 entityId string,
546 repoId *int64,
547 issueId *int64,
548 pullId *int64,
549) {
550 l := log.FromContext(ctx)
551
552 // if the user is attempting to mention >maxMentions users, this is probably spam, do not mention anybody
553 if eventType == models.NotificationTypeUserMentioned && recipients.Len() > maxMentions {
554 return
555 }
556
557 // Apply subscription overrides for thread-activity events only.
558 // Mention and assignment events are targeted at specific users and should
559 // not be broadcast to all thread subscribers.
560 isThreadEvent := eventType != models.NotificationTypeUserMentioned &&
561 eventType != models.NotificationTypeIssueAssigned &&
562 eventType != models.NotificationTypeIssueUnassigned &&
563 eventType != models.NotificationTypePullAssigned &&
564 eventType != models.NotificationTypePullUnassigned
565
566 switch {
567 case issueId != nil && isThreadEvent:
568 if subs, err := db.GetIssueSubscribers(n.db, *issueId); err == nil {
569 for _, did := range subs {
570 recipients.Insert(syntax.DID(did))
571 }
572 }
573 if unsubs, err := db.GetIssueUnsubscribers(n.db, *issueId); err == nil {
574 for _, did := range unsubs {
575 recipients.Remove(syntax.DID(did))
576 }
577 }
578 case pullId != nil && isThreadEvent:
579 if subs, err := db.GetPullSubscribers(n.db, *pullId); err == nil {
580 for _, did := range subs {
581 recipients.Insert(syntax.DID(did))
582 }
583 }
584 if unsubs, err := db.GetPullUnsubscribers(n.db, *pullId); err == nil {
585 for _, did := range unsubs {
586 recipients.Remove(syntax.DID(did))
587 }
588 }
589 }
590
591 recipients.Remove(actorDid)
592
593 // Auto-subscribe the actor to this issue/pull when they interact with it.
594 // This happens outside the transaction since it’s best-effort.
595 switch {
596 case issueId != nil:
597 if err := db.UpsertIssueSubscription(n.db, actorDid.String(), *issueId, true); err != nil {
598 l.Warn("failed to auto-subscribe actor to issue", "actor", actorDid, "issueId", *issueId, "err", err)
599 }
600 case pullId != nil:
601 if err := db.UpsertPullSubscription(n.db, actorDid.String(), *pullId, true); err != nil {
602 l.Warn("failed to auto-subscribe actor to pull", "actor", actorDid, "pullId", *pullId, "err", err)
603 }
604 }
605
606 prefMap, err := db.GetNotificationPreferences(
607 n.db,
608 orm.FilterIn("user_did", slices.Collect(recipients.All())),
609 )
610 if err != nil {
611 // failed to get prefs for users
612 return
613 }
614
615 // create a transaction for bulk notification storage
616 tx, err := n.db.Begin()
617 if err != nil {
618 // failed to start tx
619 return
620 }
621 defer tx.Rollback()
622
623 // filter based on preferences
624 for recipientDid := range recipients.All() {
625 prefs, ok := prefMap[recipientDid]
626 if !ok {
627 prefs = models.DefaultNotificationPreferences(recipientDid)
628 }
629
630 // skip users who don’t want this type
631 if !prefs.ShouldNotify(eventType) {
632 continue
633 }
634
635 // create notification
636 notif := &models.Notification{
637 RecipientDid: recipientDid.String(),
638 ActorDid: actorDid.String(),
639 Type: eventType,
640 EntityType: entityType,
641 EntityId: entityId,
642 RepoId: repoId,
643 IssueId: issueId,
644 PullId: pullId,
645 }
646
647 if err := db.CreateNotification(tx, notif); err != nil {
648 l.Error("failed to create notification", "recipientDid", recipientDid, "err", err)
649 }
650 }
651
652 if err := tx.Commit(); err != nil {
653 // failed to commit
654 return
655 }
656}