lewis.moe
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) NewIssueState(ctx context.Context, actor syntax.DID, issue *models.Issue) {
365 l := log.FromContext(ctx)
366
367 collaborators, err := db.GetCollaborators(n.db, orm.FilterEq("repo_at", issue.Repo.RepoAt()))
368 if err != nil {
369 l.Error("failed to fetch collaborators", "err", err)
370 return
371 }
372
373 // build up the recipients list:
374 // - repo owner
375 // - repo collaborators
376 // - all issue participants
377 recipients := sets.Singleton(syntax.DID(issue.Repo.Did))
378 for _, c := range collaborators {
379 recipients.Insert(c.SubjectDid)
380 }
381 for _, p := range issue.Participants() {
382 recipients.Insert(syntax.DID(p))
383 }
384
385 entityType := "issue"
386 entityId := issue.AtUri().String()
387 repoId := &issue.Repo.Id
388 issueId := &issue.Id
389 var pullId *int64
390 var eventType models.NotificationType
391
392 if issue.Open {
393 eventType = models.NotificationTypeIssueReopen
394 } else {
395 eventType = models.NotificationTypeIssueClosed
396 }
397
398 n.notifyEvent(
399 ctx,
400 actor,
401 recipients,
402 eventType,
403 entityType,
404 entityId,
405 repoId,
406 issueId,
407 pullId,
408 )
409}
410
411func (n *databaseNotifier) NewPullState(ctx context.Context, actor syntax.DID, pull *models.Pull) {
412 l := log.FromContext(ctx)
413
414 // Get repo details
415 repo, err := db.GetRepo(n.db, orm.FilterEq("at_uri", string(pull.RepoAt)))
416 if err != nil {
417 l.Error("failed to get repos", "err", err)
418 return
419 }
420
421 collaborators, err := db.GetCollaborators(n.db, orm.FilterEq("repo_at", repo.RepoAt()))
422 if err != nil {
423 l.Error("failed to fetch collaborators", "err", err)
424 return
425 }
426
427 // build up the recipients list:
428 // - repo owner
429 // - all pull participants
430 recipients := sets.Singleton(syntax.DID(repo.Did))
431 for _, c := range collaborators {
432 recipients.Insert(c.SubjectDid)
433 }
434 for _, p := range pull.Participants() {
435 recipients.Insert(syntax.DID(p))
436 }
437
438 entityType := "pull"
439 entityId := pull.AtUri().String()
440 repoId := &repo.Id
441 var issueId *int64
442 var eventType models.NotificationType
443 switch pull.State {
444 case models.PullClosed:
445 eventType = models.NotificationTypePullClosed
446 case models.PullOpen:
447 eventType = models.NotificationTypePullReopen
448 case models.PullMerged:
449 eventType = models.NotificationTypePullMerged
450 default:
451 l.Error("unexpected new PR state", "state", pull.State)
452 return
453 }
454 p := int64(pull.ID)
455 pullId := &p
456
457 n.notifyEvent(
458 ctx,
459 actor,
460 recipients,
461 eventType,
462 entityType,
463 entityId,
464 repoId,
465 issueId,
466 pullId,
467 )
468}
469
470func (n *databaseNotifier) notifyEvent(
471 ctx context.Context,
472 actorDid syntax.DID,
473 recipients sets.Set[syntax.DID],
474 eventType models.NotificationType,
475 entityType string,
476 entityId string,
477 repoId *int64,
478 issueId *int64,
479 pullId *int64,
480) {
481 l := log.FromContext(ctx)
482
483 // if the user is attempting to mention >maxMentions users, this is probably spam, do not mention anybody
484 if eventType == models.NotificationTypeUserMentioned && recipients.Len() > maxMentions {
485 return
486 }
487
488 recipients.Remove(actorDid)
489
490 prefMap, err := db.GetNotificationPreferences(
491 n.db,
492 orm.FilterIn("user_did", slices.Collect(recipients.All())),
493 )
494 if err != nil {
495 // failed to get prefs for users
496 return
497 }
498
499 // create a transaction for bulk notification storage
500 tx, err := n.db.Begin()
501 if err != nil {
502 // failed to start tx
503 return
504 }
505 defer tx.Rollback()
506
507 // filter based on preferences
508 for recipientDid := range recipients.All() {
509 prefs, ok := prefMap[recipientDid]
510 if !ok {
511 prefs = models.DefaultNotificationPreferences(recipientDid)
512 }
513
514 // skip users who don’t want this type
515 if !prefs.ShouldNotify(eventType) {
516 continue
517 }
518
519 // create notification
520 notif := &models.Notification{
521 RecipientDid: recipientDid.String(),
522 ActorDid: actorDid.String(),
523 Type: eventType,
524 EntityType: entityType,
525 EntityId: entityId,
526 RepoId: repoId,
527 IssueId: issueId,
528 PullId: pullId,
529 }
530
531 if err := db.CreateNotification(tx, notif); err != nil {
532 l.Error("failed to create notification", "recipientDid", recipientDid, "err", err)
533 }
534 }
535
536 if err := tx.Commit(); err != nil {
537 // failed to commit
538 return
539 }
540}