forked from
tangled.org/core
Monorepo for Tangled
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)
21
22type databaseNotifier struct {
23 db *db.DB
24 res *idresolver.Resolver
25}
26
27func NewDatabaseNotifier(database *db.DB, resolver *idresolver.Resolver) notify.Notifier {
28 return &databaseNotifier{
29 db: database,
30 res: resolver,
31 }
32}
33
34var _ notify.Notifier = &databaseNotifier{}
35
36func (n *databaseNotifier) NewRepo(ctx context.Context, repo *models.Repo) {
37 // no-op for now
38}
39
40func (n *databaseNotifier) NewStar(ctx context.Context, star *models.Star) {
41 l := log.FromContext(ctx)
42
43 if star.RepoAt.Collection().String() != tangled.RepoNSID {
44 // skip string stars for now
45 return
46 }
47 var err error
48 repo, err := db.GetRepo(n.db, orm.FilterEq("at_uri", string(star.RepoAt)))
49 if err != nil {
50 l.Error("failed to get repos", "err", err)
51 return
52 }
53
54 actorDid := syntax.DID(star.Did)
55 recipients := sets.Singleton(syntax.DID(repo.Did))
56 eventType := models.NotificationTypeRepoStarred
57 entityType := "repo"
58 entityId := star.RepoAt.String()
59 repoId := &repo.Id
60 var issueId *int64
61 var pullId *int64
62
63 n.notifyEvent(
64 ctx,
65 actorDid,
66 recipients,
67 eventType,
68 entityType,
69 entityId,
70 repoId,
71 issueId,
72 pullId,
73 )
74}
75
76func (n *databaseNotifier) DeleteStar(ctx context.Context, star *models.Star) {
77 // no-op
78}
79
80func (n *databaseNotifier) NewIssue(ctx context.Context, issue *models.Issue, mentions []syntax.DID) {
81 l := log.FromContext(ctx)
82
83 collaborators, err := db.GetCollaborators(n.db, orm.FilterEq("repo_at", issue.Repo.RepoAt()))
84 if err != nil {
85 l.Error("failed to fetch collaborators", "err", err)
86 return
87 }
88
89 // build the recipients list
90 // - owner of the repo
91 // - collaborators in the repo
92 // - remove users already mentioned
93 recipients := sets.Singleton(syntax.DID(issue.Repo.Did))
94 for _, c := range collaborators {
95 recipients.Insert(c.SubjectDid)
96 }
97 for _, m := range mentions {
98 recipients.Remove(m)
99 }
100
101 actorDid := syntax.DID(issue.Did)
102 entityType := "issue"
103 entityId := issue.AtUri().String()
104 repoId := &issue.Repo.Id
105 issueId := &issue.Id
106 var pullId *int64
107
108 n.notifyEvent(
109 ctx,
110 actorDid,
111 recipients,
112 models.NotificationTypeIssueCreated,
113 entityType,
114 entityId,
115 repoId,
116 issueId,
117 pullId,
118 )
119 n.notifyEvent(
120 ctx,
121 actorDid,
122 sets.Collect(slices.Values(mentions)),
123 models.NotificationTypeUserMentioned,
124 entityType,
125 entityId,
126 repoId,
127 issueId,
128 pullId,
129 )
130}
131
132func (n *databaseNotifier) NewIssueComment(ctx context.Context, comment *models.IssueComment, mentions []syntax.DID) {
133 l := log.FromContext(ctx)
134
135 issues, err := db.GetIssues(n.db, orm.FilterEq("at_uri", comment.IssueAt))
136 if err != nil {
137 l.Error("failed to get issues", "err", err)
138 return
139 }
140 if len(issues) == 0 {
141 l.Error("no issue found for", "err", comment.IssueAt)
142 return
143 }
144 issue := issues[0]
145
146 // built the recipients list:
147 // - the owner of the repo
148 // - | if the comment is a reply -> everybody on that thread
149 // | if the comment is a top level -> just the issue owner
150 // - remove mentioned users from the recipients list
151 recipients := sets.Singleton(syntax.DID(issue.Repo.Did))
152
153 if comment.IsReply() {
154 // if this comment is a reply, then notify everybody in that thread
155 parentAtUri := *comment.ReplyTo
156
157 // find the parent thread, and add all DIDs from here to the recipient list
158 for _, t := range issue.CommentList() {
159 if t.Self.AtUri().String() == parentAtUri {
160 for _, p := range t.Participants() {
161 recipients.Insert(p)
162 }
163 }
164 }
165 } else {
166 // not a reply, notify just the issue author
167 recipients.Insert(syntax.DID(issue.Did))
168 }
169
170 for _, m := range mentions {
171 recipients.Remove(m)
172 }
173
174 actorDid := syntax.DID(comment.Did)
175 entityType := "issue"
176 entityId := issue.AtUri().String()
177 repoId := &issue.Repo.Id
178 issueId := &issue.Id
179 var pullId *int64
180
181 n.notifyEvent(
182 ctx,
183 actorDid,
184 recipients,
185 models.NotificationTypeIssueCommented,
186 entityType,
187 entityId,
188 repoId,
189 issueId,
190 pullId,
191 )
192 n.notifyEvent(
193 ctx,
194 actorDid,
195 sets.Collect(slices.Values(mentions)),
196 models.NotificationTypeUserMentioned,
197 entityType,
198 entityId,
199 repoId,
200 issueId,
201 pullId,
202 )
203}
204
205func (n *databaseNotifier) DeleteIssue(ctx context.Context, issue *models.Issue) {
206 // no-op for now
207}
208
209func (n *databaseNotifier) NewIssueLabelOp(ctx context.Context, issue *models.Issue) {}
210func (n *databaseNotifier) NewPullLabelOp(ctx context.Context, pull *models.Pull) {}
211
212func (n *databaseNotifier) NewFollow(ctx context.Context, follow *models.Follow) {
213 actorDid := syntax.DID(follow.UserDid)
214 recipients := sets.Singleton(syntax.DID(follow.SubjectDid))
215 eventType := models.NotificationTypeFollowed
216 entityType := "follow"
217 entityId := follow.UserDid
218 var repoId, issueId, pullId *int64
219
220 n.notifyEvent(
221 ctx,
222 actorDid,
223 recipients,
224 eventType,
225 entityType,
226 entityId,
227 repoId,
228 issueId,
229 pullId,
230 )
231}
232
233func (n *databaseNotifier) DeleteFollow(ctx context.Context, follow *models.Follow) {
234 // no-op
235}
236
237func (n *databaseNotifier) NewPull(ctx context.Context, pull *models.Pull) {
238 l := log.FromContext(ctx)
239
240 repo, err := db.GetRepo(n.db, orm.FilterEq("at_uri", string(pull.RepoAt)))
241 if err != nil {
242 l.Error("failed to get repos", "err", err)
243 return
244 }
245 collaborators, err := db.GetCollaborators(n.db, orm.FilterEq("repo_at", repo.RepoAt()))
246 if err != nil {
247 l.Error("failed to fetch collaborators", "err", err)
248 return
249 }
250
251 // build the recipients list
252 // - owner of the repo
253 // - collaborators in the repo
254 recipients := sets.Singleton(syntax.DID(repo.Did))
255 for _, c := range collaborators {
256 recipients.Insert(c.SubjectDid)
257 }
258
259 actorDid := syntax.DID(pull.OwnerDid)
260 eventType := models.NotificationTypePullCreated
261 entityType := "pull"
262 entityId := pull.AtUri().String()
263 repoId := &repo.Id
264 var issueId *int64
265 p := int64(pull.ID)
266 pullId := &p
267
268 n.notifyEvent(
269 ctx,
270 actorDid,
271 recipients,
272 eventType,
273 entityType,
274 entityId,
275 repoId,
276 issueId,
277 pullId,
278 )
279}
280
281func (n *databaseNotifier) NewPullComment(ctx context.Context, comment *models.PullComment, mentions []syntax.DID) {
282 l := log.FromContext(ctx)
283
284 pull, err := db.GetPull(n.db,
285 syntax.ATURI(comment.RepoAt),
286 comment.PullId,
287 )
288 if err != nil {
289 l.Error("failed to get pulls", "err", err)
290 return
291 }
292
293 repo, err := db.GetRepo(n.db, orm.FilterEq("at_uri", comment.RepoAt))
294 if err != nil {
295 l.Error("failed to get repos", "err", err)
296 return
297 }
298
299 // build up the recipients list:
300 // - repo owner
301 // - all pull participants
302 // - remove those already mentioned
303 recipients := sets.Singleton(syntax.DID(repo.Did))
304 for _, p := range pull.Participants() {
305 recipients.Insert(syntax.DID(p))
306 }
307 for _, m := range mentions {
308 recipients.Remove(m)
309 }
310
311 actorDid := syntax.DID(comment.OwnerDid)
312 eventType := models.NotificationTypePullCommented
313 entityType := "pull"
314 entityId := pull.AtUri().String()
315 repoId := &repo.Id
316 var issueId *int64
317 p := int64(pull.ID)
318 pullId := &p
319
320 n.notifyEvent(
321 ctx,
322 actorDid,
323 recipients,
324 eventType,
325 entityType,
326 entityId,
327 repoId,
328 issueId,
329 pullId,
330 )
331 n.notifyEvent(
332 ctx,
333 actorDid,
334 sets.Collect(slices.Values(mentions)),
335 models.NotificationTypeUserMentioned,
336 entityType,
337 entityId,
338 repoId,
339 issueId,
340 pullId,
341 )
342}
343
344func (n *databaseNotifier) UpdateProfile(ctx context.Context, profile *models.Profile) {
345 // no-op
346}
347
348func (n *databaseNotifier) DeleteString(ctx context.Context, did, rkey string) {
349 // no-op
350}
351
352func (n *databaseNotifier) EditString(ctx context.Context, string *models.String) {
353 // no-op
354}
355
356func (n *databaseNotifier) NewString(ctx context.Context, string *models.String) {
357 // no-op
358}
359
360func (n *databaseNotifier) Push(ctx context.Context, repo *models.Repo, ref, oldSha, newSha, committerDid string) {
361 // no-op for now; webhooks are handled by the webhook notifier
362}
363
364func (n *databaseNotifier) Clone(ctx context.Context, repo *models.Repo) {
365 // no-op
366}
367
368func (n *databaseNotifier) NewIssueState(ctx context.Context, actor syntax.DID, issue *models.Issue) {
369 l := log.FromContext(ctx)
370
371 collaborators, err := db.GetCollaborators(n.db, orm.FilterEq("repo_at", issue.Repo.RepoAt()))
372 if err != nil {
373 l.Error("failed to fetch collaborators", "err", err)
374 return
375 }
376
377 // build up the recipients list:
378 // - repo owner
379 // - repo collaborators
380 // - all issue participants
381 recipients := sets.Singleton(syntax.DID(issue.Repo.Did))
382 for _, c := range collaborators {
383 recipients.Insert(c.SubjectDid)
384 }
385 for _, p := range issue.Participants() {
386 recipients.Insert(syntax.DID(p))
387 }
388
389 entityType := "issue"
390 entityId := issue.AtUri().String()
391 repoId := &issue.Repo.Id
392 issueId := &issue.Id
393 var pullId *int64
394 var eventType models.NotificationType
395
396 if issue.Open {
397 eventType = models.NotificationTypeIssueReopen
398 } else {
399 eventType = models.NotificationTypeIssueClosed
400 }
401
402 n.notifyEvent(
403 ctx,
404 actor,
405 recipients,
406 eventType,
407 entityType,
408 entityId,
409 repoId,
410 issueId,
411 pullId,
412 )
413}
414
415func (n *databaseNotifier) NewPullState(ctx context.Context, actor syntax.DID, pull *models.Pull) {
416 l := log.FromContext(ctx)
417
418 // Get repo details
419 repo, err := db.GetRepo(n.db, orm.FilterEq("at_uri", string(pull.RepoAt)))
420 if err != nil {
421 l.Error("failed to get repos", "err", err)
422 return
423 }
424
425 collaborators, err := db.GetCollaborators(n.db, orm.FilterEq("repo_at", repo.RepoAt()))
426 if err != nil {
427 l.Error("failed to fetch collaborators", "err", err)
428 return
429 }
430
431 // build up the recipients list:
432 // - repo owner
433 // - all pull participants
434 recipients := sets.Singleton(syntax.DID(repo.Did))
435 for _, c := range collaborators {
436 recipients.Insert(c.SubjectDid)
437 }
438 for _, p := range pull.Participants() {
439 recipients.Insert(syntax.DID(p))
440 }
441
442 entityType := "pull"
443 entityId := pull.AtUri().String()
444 repoId := &repo.Id
445 var issueId *int64
446 var eventType models.NotificationType
447 switch pull.State {
448 case models.PullClosed:
449 eventType = models.NotificationTypePullClosed
450 case models.PullOpen:
451 eventType = models.NotificationTypePullReopen
452 case models.PullMerged:
453 eventType = models.NotificationTypePullMerged
454 default:
455 l.Error("unexpected new PR state", "state", pull.State)
456 return
457 }
458 p := int64(pull.ID)
459 pullId := &p
460
461 n.notifyEvent(
462 ctx,
463 actor,
464 recipients,
465 eventType,
466 entityType,
467 entityId,
468 repoId,
469 issueId,
470 pullId,
471 )
472}
473
474func (n *databaseNotifier) notifyEvent(
475 ctx context.Context,
476 actorDid syntax.DID,
477 recipients sets.Set[syntax.DID],
478 eventType models.NotificationType,
479 entityType string,
480 entityId string,
481 repoId *int64,
482 issueId *int64,
483 pullId *int64,
484) {
485 l := log.FromContext(ctx)
486
487 // if the user is attempting to mention >maxMentions users, this is probably spam, do not mention anybody
488 if eventType == models.NotificationTypeUserMentioned && recipients.Len() > maxMentions {
489 return
490 }
491
492 recipients.Remove(actorDid)
493
494 prefMap, err := db.GetNotificationPreferences(
495 n.db,
496 orm.FilterIn("user_did", slices.Collect(recipients.All())),
497 )
498 if err != nil {
499 // failed to get prefs for users
500 return
501 }
502
503 // create a transaction for bulk notification storage
504 tx, err := n.db.Begin()
505 if err != nil {
506 // failed to start tx
507 return
508 }
509 defer tx.Rollback()
510
511 // filter based on preferences
512 for recipientDid := range recipients.All() {
513 prefs, ok := prefMap[recipientDid]
514 if !ok {
515 prefs = models.DefaultNotificationPreferences(recipientDid)
516 }
517
518 // skip users who don’t want this type
519 if !prefs.ShouldNotify(eventType) {
520 continue
521 }
522
523 // create notification
524 notif := &models.Notification{
525 RecipientDid: recipientDid.String(),
526 ActorDid: actorDid.String(),
527 Type: eventType,
528 EntityType: entityType,
529 EntityId: entityId,
530 RepoId: repoId,
531 IssueId: issueId,
532 PullId: pullId,
533 }
534
535 if err := db.CreateNotification(tx, notif); err != nil {
536 l.Error("failed to create notification", "recipientDid", recipientDid, "err", err)
537 }
538 }
539
540 if err := tx.Commit(); err != nil {
541 // failed to commit
542 return
543 }
544}