alice.pds.demo.boltless.dev
This repository has no description
1package db
2
3import (
4 "context"
5 "database/sql"
6 "errors"
7 "fmt"
8 "strings"
9 "time"
10
11 "github.com/bluesky-social/indigo/atproto/syntax"
12 "tangled.org/core/appview/models"
13 "tangled.org/core/appview/pagination"
14 "tangled.org/core/orm"
15)
16
17func CreateNotification(e Execer, notification *models.Notification) error {
18 query := `
19 INSERT INTO notifications (recipient_did, actor_did, type, entity_type, entity_id, read, repo_id, issue_id, pull_id)
20 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
21 `
22
23 result, err := e.Exec(query,
24 notification.RecipientDid,
25 notification.ActorDid,
26 string(notification.Type),
27 notification.EntityType,
28 notification.EntityId,
29 notification.Read,
30 notification.RepoId,
31 notification.IssueId,
32 notification.PullId,
33 )
34 if err != nil {
35 return fmt.Errorf("failed to create notification: %w", err)
36 }
37
38 id, err := result.LastInsertId()
39 if err != nil {
40 return fmt.Errorf("failed to get notification ID: %w", err)
41 }
42
43 notification.ID = id
44 return nil
45}
46
47// GetNotificationsPaginated retrieves notifications with filters and pagination
48func GetNotificationsPaginated(e Execer, page pagination.Page, filters ...orm.Filter) ([]*models.Notification, error) {
49 var conditions []string
50 var args []any
51
52 for _, filter := range filters {
53 conditions = append(conditions, filter.Condition())
54 args = append(args, filter.Arg()...)
55 }
56
57 whereClause := ""
58 if len(conditions) > 0 {
59 whereClause = "WHERE " + conditions[0]
60 for _, condition := range conditions[1:] {
61 whereClause += " AND " + condition
62 }
63 }
64 pageClause := ""
65 if page.Limit > 0 {
66 pageClause = " limit ? offset ? "
67 args = append(args, page.Limit, page.Offset)
68 }
69
70 query := fmt.Sprintf(`
71 select id, recipient_did, actor_did, type, entity_type, entity_id, read, created, repo_id, issue_id, pull_id
72 from notifications
73 %s
74 order by created desc
75 %s
76 `, whereClause, pageClause)
77
78 rows, err := e.QueryContext(context.Background(), query, args...)
79 if err != nil {
80 return nil, fmt.Errorf("failed to query notifications: %w", err)
81 }
82 defer rows.Close()
83
84 var notifications []*models.Notification
85 for rows.Next() {
86 var n models.Notification
87 var typeStr string
88 var createdStr string
89 err := rows.Scan(
90 &n.ID,
91 &n.RecipientDid,
92 &n.ActorDid,
93 &typeStr,
94 &n.EntityType,
95 &n.EntityId,
96 &n.Read,
97 &createdStr,
98 &n.RepoId,
99 &n.IssueId,
100 &n.PullId,
101 )
102 if err != nil {
103 return nil, fmt.Errorf("failed to scan notification: %w", err)
104 }
105 n.Type = models.NotificationType(typeStr)
106 n.Created, err = time.Parse(time.RFC3339, createdStr)
107 if err != nil {
108 return nil, fmt.Errorf("failed to parse created timestamp: %w", err)
109 }
110 notifications = append(notifications, &n)
111 }
112
113 return notifications, nil
114}
115
116func GetNotificationWithEntity(e Execer, notificationID int64, userDID string) (*models.NotificationWithEntity, error) {
117 results, err := GetNotificationsWithEntities(e, pagination.Page{Limit: 1, Offset: 0},
118 orm.FilterEq("n.id", notificationID),
119 orm.FilterEq("n.recipient_did", userDID),
120 )
121 if err != nil {
122 return nil, err
123 }
124 if len(results) == 0 {
125 return nil, fmt.Errorf("notification not found")
126 }
127 return results[0], nil
128}
129
130// GetNotificationsWithEntities retrieves notifications with their related entities
131func GetNotificationsWithEntities(e Execer, page pagination.Page, filters ...orm.Filter) ([]*models.NotificationWithEntity, error) {
132 var conditions []string
133 var args []any
134
135 for _, filter := range filters {
136 conditions = append(conditions, filter.Condition())
137 args = append(args, filter.Arg()...)
138 }
139
140 whereClause := ""
141 if len(conditions) > 0 {
142 whereClause = "WHERE " + conditions[0]
143 for _, condition := range conditions[1:] {
144 whereClause += " AND " + condition
145 }
146 }
147
148 query := fmt.Sprintf(`
149 select
150 n.id, n.recipient_did, n.actor_did, n.type, n.entity_type, n.entity_id,
151 n.read, n.created, n.repo_id, n.issue_id, n.pull_id,
152 r.id as r_id, r.did as r_did, r.rkey as r_rkey, r.name as r_name, r.description as r_description, r.website as r_website, r.topics as r_topics,
153 i.id as i_id, i.did as i_did, i.issue_id as i_issue_id, i.title as i_title, i.open as i_open,
154 p.id as p_id, p.owner_did as p_owner_did, p.pull_id as p_pull_id, p.title as p_title, p.state as p_state
155 from notifications n
156 left join repos r on n.repo_id = r.id
157 left join issues i on n.issue_id = i.id
158 left join pulls p on n.pull_id = p.id
159 %s
160 order by n.created desc
161 limit ? offset ?
162 `, whereClause)
163
164 args = append(args, page.Limit, page.Offset)
165
166 rows, err := e.QueryContext(context.Background(), query, args...)
167 if err != nil {
168 return nil, fmt.Errorf("failed to query notifications with entities: %w", err)
169 }
170 defer rows.Close()
171
172 var notifications []*models.NotificationWithEntity
173 for rows.Next() {
174 var n models.Notification
175 var typeStr string
176 var createdStr string
177 var repo models.Repo
178 var issue models.Issue
179 var pull models.Pull
180 var rId, iId, pId sql.NullInt64
181 var rDid, rRkey, rName, rDescription, rWebsite, rTopicStr sql.NullString
182 var iDid sql.NullString
183 var iIssueId sql.NullInt64
184 var iTitle sql.NullString
185 var iOpen sql.NullBool
186 var pOwnerDid sql.NullString
187 var pPullId sql.NullInt64
188 var pTitle sql.NullString
189 var pState sql.NullInt64
190
191 err := rows.Scan(
192 &n.ID, &n.RecipientDid, &n.ActorDid, &typeStr, &n.EntityType, &n.EntityId,
193 &n.Read, &createdStr, &n.RepoId, &n.IssueId, &n.PullId,
194 &rId, &rDid, &rRkey, &rName, &rDescription, &rWebsite, &rTopicStr,
195 &iId, &iDid, &iIssueId, &iTitle, &iOpen,
196 &pId, &pOwnerDid, &pPullId, &pTitle, &pState,
197 )
198 if err != nil {
199 return nil, fmt.Errorf("failed to scan notification with entities: %w", err)
200 }
201
202 n.Type = models.NotificationType(typeStr)
203 n.Created, err = time.Parse(time.RFC3339, createdStr)
204 if err != nil {
205 return nil, fmt.Errorf("failed to parse created timestamp: %w", err)
206 }
207
208 entry := &models.NotificationWithEntity{Notification: &n}
209
210 // populate repo if present
211 if rId.Valid {
212 repo.Id = rId.Int64
213 if rDid.Valid {
214 repo.Did = rDid.String
215 }
216 if rRkey.Valid {
217 repo.Rkey = rRkey.String
218 }
219 if rName.Valid {
220 repo.Name = rName.String
221 }
222 if rDescription.Valid {
223 repo.Description = rDescription.String
224 }
225 if rWebsite.Valid {
226 repo.Website = rWebsite.String
227 }
228 if rTopicStr.Valid {
229 repo.Topics = strings.Fields(rTopicStr.String)
230 }
231 entry.Repo = &repo
232 }
233
234 // populate issue if present
235 if iId.Valid {
236 issue.Id = iId.Int64
237 if iDid.Valid {
238 issue.Did = iDid.String
239 }
240 if iIssueId.Valid {
241 issue.IssueId = int(iIssueId.Int64)
242 }
243 if iTitle.Valid {
244 issue.Title = iTitle.String
245 }
246 if iOpen.Valid {
247 issue.Open = iOpen.Bool
248 }
249 entry.Issue = &issue
250 }
251
252 // populate pull if present
253 if pId.Valid {
254 pull.ID = int(pId.Int64)
255 if pOwnerDid.Valid {
256 pull.OwnerDid = pOwnerDid.String
257 }
258 if pPullId.Valid {
259 pull.PullId = int(pPullId.Int64)
260 }
261 if pTitle.Valid {
262 pull.Title = pTitle.String
263 }
264 if pState.Valid {
265 pull.State = models.PullState(pState.Int64)
266 }
267 entry.Pull = &pull
268 }
269
270 notifications = append(notifications, entry)
271 }
272
273 return notifications, nil
274}
275
276// GetNotifications retrieves notifications with filters
277func GetNotifications(e Execer, filters ...orm.Filter) ([]*models.Notification, error) {
278 return GetNotificationsPaginated(e, pagination.FirstPage(), filters...)
279}
280
281func CountNotifications(e Execer, filters ...orm.Filter) (int64, error) {
282 var conditions []string
283 var args []any
284 for _, filter := range filters {
285 conditions = append(conditions, filter.Condition())
286 args = append(args, filter.Arg()...)
287 }
288
289 whereClause := ""
290 if conditions != nil {
291 whereClause = " where " + strings.Join(conditions, " and ")
292 }
293
294 query := fmt.Sprintf(`select count(1) from notifications %s`, whereClause)
295 var count int64
296 err := e.QueryRow(query, args...).Scan(&count)
297
298 if !errors.Is(err, sql.ErrNoRows) && err != nil {
299 return 0, err
300 }
301
302 return count, nil
303}
304
305func MarkNotificationRead(e Execer, notificationID int64, userDID string) error {
306 idFilter := orm.FilterEq("id", notificationID)
307 recipientFilter := orm.FilterEq("recipient_did", userDID)
308
309 query := fmt.Sprintf(`
310 UPDATE notifications
311 SET read = 1
312 WHERE %s AND %s
313 `, idFilter.Condition(), recipientFilter.Condition())
314
315 args := append(idFilter.Arg(), recipientFilter.Arg()...)
316
317 result, err := e.Exec(query, args...)
318 if err != nil {
319 return fmt.Errorf("failed to mark notification as read: %w", err)
320 }
321
322 rowsAffected, err := result.RowsAffected()
323 if err != nil {
324 return fmt.Errorf("failed to get rows affected: %w", err)
325 }
326
327 if rowsAffected == 0 {
328 return fmt.Errorf("notification not found or access denied")
329 }
330
331 return nil
332}
333
334func MarkNotificationsReadForIssue(e Execer, userDID, repoDid string, issueNum int) error {
335 query := `
336 update notifications set read = 1
337 where recipient_did = ?
338 and read = 0
339 and issue_id = (select id from issues where repo_did = ? and issue_id = ?)
340 `
341 _, err := e.Exec(query, userDID, repoDid, issueNum)
342 return err
343}
344
345func MarkNotificationsReadForPull(e Execer, userDID, repoDid string, pullNum int) error {
346 query := `
347 update notifications set read = 1
348 where recipient_did = ?
349 and read = 0
350 and pull_id = (select p.id from pulls p where p.pull_id = ? and p.repo_did = ?)
351 `
352 _, err := e.Exec(query, userDID, pullNum, repoDid)
353 return err
354}
355
356func MarkNotificationUnread(e Execer, notificationID int64, userDID string) error {
357 idFilter := orm.FilterEq("id", notificationID)
358 recipientFilter := orm.FilterEq("recipient_did", userDID)
359
360 query := fmt.Sprintf(`
361 UPDATE notifications
362 SET read = 0
363 WHERE %s AND %s
364 `, idFilter.Condition(), recipientFilter.Condition())
365
366 args := append(idFilter.Arg(), recipientFilter.Arg()...)
367
368 result, err := e.Exec(query, args...)
369 if err != nil {
370 return fmt.Errorf("failed to mark notification as unread: %w", err)
371 }
372
373 rowsAffected, err := result.RowsAffected()
374 if err != nil {
375 return fmt.Errorf("failed to get rows affected: %w", err)
376 }
377
378 if rowsAffected == 0 {
379 return fmt.Errorf("notification not found or access denied")
380 }
381
382 return nil
383}
384
385func MarkAllNotificationsRead(e Execer, userDID string) error {
386 recipientFilter := orm.FilterEq("recipient_did", userDID)
387 readFilter := orm.FilterEq("read", 0)
388
389 query := fmt.Sprintf(`
390 UPDATE notifications
391 SET read = 1
392 WHERE %s AND %s
393 `, recipientFilter.Condition(), readFilter.Condition())
394
395 args := append(recipientFilter.Arg(), readFilter.Arg()...)
396
397 _, err := e.Exec(query, args...)
398 if err != nil {
399 return fmt.Errorf("failed to mark all notifications as read: %w", err)
400 }
401
402 return nil
403}
404
405func DeleteNotification(e Execer, notificationID int64, userDID string) error {
406 idFilter := orm.FilterEq("id", notificationID)
407 recipientFilter := orm.FilterEq("recipient_did", userDID)
408
409 query := fmt.Sprintf(`
410 DELETE FROM notifications
411 WHERE %s AND %s
412 `, idFilter.Condition(), recipientFilter.Condition())
413
414 args := append(idFilter.Arg(), recipientFilter.Arg()...)
415
416 result, err := e.Exec(query, args...)
417 if err != nil {
418 return fmt.Errorf("failed to delete notification: %w", err)
419 }
420
421 rowsAffected, err := result.RowsAffected()
422 if err != nil {
423 return fmt.Errorf("failed to get rows affected: %w", err)
424 }
425
426 if rowsAffected == 0 {
427 return fmt.Errorf("notification not found or access denied")
428 }
429
430 return nil
431}
432
433func GetNotificationPreference(e Execer, userDid string) (*models.NotificationPreferences, error) {
434 prefs, err := GetNotificationPreferences(e, orm.FilterEq("user_did", userDid))
435 if err != nil {
436 return nil, err
437 }
438
439 p, ok := prefs[syntax.DID(userDid)]
440 if !ok {
441 return models.DefaultNotificationPreferences(syntax.DID(userDid)), nil
442 }
443
444 return p, nil
445}
446
447func GetNotificationPreferences(e Execer, filters ...orm.Filter) (map[syntax.DID]*models.NotificationPreferences, error) {
448 prefsMap := make(map[syntax.DID]*models.NotificationPreferences)
449
450 var conditions []string
451 var args []any
452 for _, filter := range filters {
453 conditions = append(conditions, filter.Condition())
454 args = append(args, filter.Arg()...)
455 }
456
457 whereClause := ""
458 if conditions != nil {
459 whereClause = " where " + strings.Join(conditions, " and ")
460 }
461
462 query := fmt.Sprintf(`
463 select
464 id,
465 user_did,
466 repo_starred,
467 issue_created,
468 issue_commented,
469 pull_created,
470 pull_commented,
471 followed,
472 user_mentioned,
473 pull_merged,
474 issue_closed,
475 email_notifications
476 from
477 notification_preferences
478 %s
479 `, whereClause)
480
481 rows, err := e.Query(query, args...)
482 if err != nil {
483 return nil, err
484 }
485 defer rows.Close()
486
487 for rows.Next() {
488 var prefs models.NotificationPreferences
489 if err := rows.Scan(
490 &prefs.ID,
491 &prefs.UserDid,
492 &prefs.RepoStarred,
493 &prefs.IssueCreated,
494 &prefs.IssueCommented,
495 &prefs.PullCreated,
496 &prefs.PullCommented,
497 &prefs.Followed,
498 &prefs.UserMentioned,
499 &prefs.PullMerged,
500 &prefs.IssueClosed,
501 &prefs.EmailNotifications,
502 ); err != nil {
503 return nil, err
504 }
505
506 prefsMap[prefs.UserDid] = &prefs
507 }
508
509 if err := rows.Err(); err != nil {
510 return nil, err
511 }
512
513 return prefsMap, nil
514}
515
516func (d *DB) UpdateNotificationPreferences(ctx context.Context, prefs *models.NotificationPreferences) error {
517 tx, err := d.DB.BeginTx(ctx, nil)
518 if err != nil {
519 return fmt.Errorf("failed to begin transaction: %w", err)
520 }
521 defer tx.Rollback()
522
523 var prevEmailEnabled bool
524 var hadPrefs bool
525 row := tx.QueryRowContext(ctx,
526 `SELECT email_notifications FROM notification_preferences WHERE user_did = ?`,
527 prefs.UserDid,
528 )
529 switch err := row.Scan(&prevEmailEnabled); err {
530 case nil:
531 hadPrefs = true
532 case sql.ErrNoRows:
533 hadPrefs = false
534 default:
535 return fmt.Errorf("failed to read existing preferences: %w", err)
536 }
537
538 query := `
539 INSERT OR REPLACE INTO notification_preferences
540 (user_did, repo_starred, issue_created, issue_commented, pull_created,
541 pull_commented, followed, user_mentioned, pull_merged, issue_closed,
542 email_notifications)
543 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
544 `
545
546 result, err := tx.ExecContext(ctx, query,
547 prefs.UserDid,
548 prefs.RepoStarred,
549 prefs.IssueCreated,
550 prefs.IssueCommented,
551 prefs.PullCreated,
552 prefs.PullCommented,
553 prefs.Followed,
554 prefs.UserMentioned,
555 prefs.PullMerged,
556 prefs.IssueClosed,
557 prefs.EmailNotifications,
558 )
559 if err != nil {
560 return fmt.Errorf("failed to update notification preferences: %w", err)
561 }
562
563 // on enabling email notifications (from disabled or from no prior setting),
564 // mark existing unemailed notifications as emailed so they aren't dispatched
565 // as a backlog on the next digest tick.
566 if prefs.EmailNotifications && (!hadPrefs || !prevEmailEnabled) {
567 if err := MarkAllNotificationsEmailed(tx, string(prefs.UserDid)); err != nil {
568 return fmt.Errorf("failed to suppress notification backlog: %w", err)
569 }
570 }
571
572 if prefs.ID == 0 {
573 id, err := result.LastInsertId()
574 if err != nil {
575 return fmt.Errorf("failed to get preferences ID: %w", err)
576 }
577 prefs.ID = id
578 }
579
580 return tx.Commit()
581}
582
583// GetPendingEmailDigestRecipients returns DIDs of users who have email
584// notifications enabled, have a verified primary email, and have unread,
585// unemailed notifications older than olderThan for email-eligible notification
586// types.
587func GetPendingEmailDigestRecipients(e Execer, olderThan time.Time) ([]string, error) {
588 placeholders := make([]string, len(models.EmailNotificationTypes))
589 args := []any{olderThan.UTC().Format(time.RFC3339)}
590 for i, t := range models.EmailNotificationTypes {
591 placeholders[i] = "?"
592 args = append(args, string(t))
593 }
594 inClause := strings.Join(placeholders, ", ")
595
596 query := fmt.Sprintf(`
597 SELECT DISTINCT n.recipient_did
598 FROM notifications n
599 JOIN notification_preferences np ON np.user_did = n.recipient_did
600 JOIN emails e ON e.did = n.recipient_did AND e.is_primary = 1 AND e.verified = 1
601 WHERE n.emailed = 0
602 AND n.read = 0
603 AND n.created < ?
604 AND np.email_notifications = 1
605 AND n.type IN (%s)
606 `, inClause)
607
608 rows, err := e.Query(query, args...)
609 if err != nil {
610 return nil, fmt.Errorf("failed to query email digest recipients: %w", err)
611 }
612 defer rows.Close()
613
614 var dids []string
615 for rows.Next() {
616 var did string
617 if err := rows.Scan(&did); err != nil {
618 return nil, err
619 }
620 dids = append(dids, did)
621 }
622 return dids, rows.Err()
623}
624
625// GetPendingNotificationsForEmailDigest returns all unread, unemailed
626// notifications older than olderThan for email-eligible types for a specific
627// recipient.
628func GetPendingNotificationsForEmailDigest(e Execer, recipientDid string, olderThan time.Time) ([]*models.NotificationWithEntity, error) {
629 placeholders := make([]string, len(models.EmailNotificationTypes))
630 args := []any{recipientDid, olderThan.UTC().Format(time.RFC3339)}
631 for i, t := range models.EmailNotificationTypes {
632 placeholders[i] = "?"
633 args = append(args, string(t))
634 }
635 inClause := strings.Join(placeholders, ", ")
636
637 query := fmt.Sprintf(`
638 SELECT
639 n.id, n.recipient_did, n.actor_did, n.type, n.entity_type, n.entity_id,
640 n.read, n.created, n.repo_id, n.issue_id, n.pull_id,
641 r.id as r_id, r.did as r_did, r.rkey as r_rkey, r.name as r_name, r.description as r_description, r.website as r_website, r.topics as r_topics,
642 i.id as i_id, i.did as i_did, i.issue_id as i_issue_id, i.title as i_title, i.open as i_open,
643 p.id as p_id, p.owner_did as p_owner_did, p.pull_id as p_pull_id, p.title as p_title, p.state as p_state
644 FROM notifications n
645 LEFT JOIN repos r ON n.repo_id = r.id
646 LEFT JOIN issues i ON n.issue_id = i.id
647 LEFT JOIN pulls p ON n.pull_id = p.id
648 WHERE n.recipient_did = ?
649 AND n.emailed = 0
650 AND n.read = 0
651 AND n.created < ?
652 AND n.type IN (%s)
653 ORDER BY n.created DESC
654 `, inClause)
655
656 rows, err := e.QueryContext(context.Background(), query, args...)
657 if err != nil {
658 return nil, fmt.Errorf("failed to query pending email notifications: %w", err)
659 }
660 defer rows.Close()
661
662 var notifications []*models.NotificationWithEntity
663 for rows.Next() {
664 var n models.Notification
665 var typeStr string
666 var createdStr string
667 var repo models.Repo
668 var issue models.Issue
669 var pull models.Pull
670 var rId, iId, pId sql.NullInt64
671 var rDid, rRkey, rName, rDescription, rWebsite, rTopicStr sql.NullString
672 var iDid sql.NullString
673 var iIssueId sql.NullInt64
674 var iTitle sql.NullString
675 var iOpen sql.NullBool
676 var pOwnerDid sql.NullString
677 var pPullId sql.NullInt64
678 var pTitle sql.NullString
679 var pState sql.NullInt64
680
681 err := rows.Scan(
682 &n.ID, &n.RecipientDid, &n.ActorDid, &typeStr, &n.EntityType, &n.EntityId,
683 &n.Read, &createdStr, &n.RepoId, &n.IssueId, &n.PullId,
684 &rId, &rDid, &rRkey, &rName, &rDescription, &rWebsite, &rTopicStr,
685 &iId, &iDid, &iIssueId, &iTitle, &iOpen,
686 &pId, &pOwnerDid, &pPullId, &pTitle, &pState,
687 )
688 if err != nil {
689 return nil, fmt.Errorf("failed to scan email digest notification: %w", err)
690 }
691
692 n.Type = models.NotificationType(typeStr)
693 n.Created, err = time.Parse(time.RFC3339, createdStr)
694 if err != nil {
695 return nil, fmt.Errorf("failed to parse created timestamp: %w", err)
696 }
697
698 entry := &models.NotificationWithEntity{Notification: &n}
699
700 if rId.Valid {
701 repo.Id = rId.Int64
702 if rDid.Valid {
703 repo.Did = rDid.String
704 }
705 if rRkey.Valid {
706 repo.Rkey = rRkey.String
707 }
708 if rName.Valid {
709 repo.Name = rName.String
710 }
711 if rDescription.Valid {
712 repo.Description = rDescription.String
713 }
714 if rWebsite.Valid {
715 repo.Website = rWebsite.String
716 }
717 if rTopicStr.Valid {
718 repo.Topics = strings.Fields(rTopicStr.String)
719 }
720 entry.Repo = &repo
721 }
722
723 if iId.Valid {
724 issue.Id = iId.Int64
725 if iDid.Valid {
726 issue.Did = iDid.String
727 }
728 if iIssueId.Valid {
729 issue.IssueId = int(iIssueId.Int64)
730 }
731 if iTitle.Valid {
732 issue.Title = iTitle.String
733 }
734 if iOpen.Valid {
735 issue.Open = iOpen.Bool
736 }
737 entry.Issue = &issue
738 }
739
740 if pId.Valid {
741 pull.ID = int(pId.Int64)
742 if pOwnerDid.Valid {
743 pull.OwnerDid = pOwnerDid.String
744 }
745 if pPullId.Valid {
746 pull.PullId = int(pPullId.Int64)
747 }
748 if pTitle.Valid {
749 pull.Title = pTitle.String
750 }
751 if pState.Valid {
752 pull.State = models.PullState(pState.Int64)
753 }
754 entry.Pull = &pull
755 }
756
757 notifications = append(notifications, entry)
758 }
759
760 return notifications, rows.Err()
761}
762
763// MarkNotificationsEmailed marks the given notification IDs as emailed=1.
764// Uses explicit IDs (not recipient_did) to avoid racing with new notifications.
765func MarkNotificationsEmailed(e Execer, ids []int64) error {
766 if len(ids) == 0 {
767 return nil
768 }
769 placeholders := make([]string, len(ids))
770 args := make([]any, len(ids))
771 for i, id := range ids {
772 placeholders[i] = "?"
773 args[i] = id
774 }
775 query := fmt.Sprintf(
776 `UPDATE notifications SET emailed = 1 WHERE id IN (%s)`,
777 strings.Join(placeholders, ", "),
778 )
779 _, err := e.Exec(query, args...)
780 return err
781}
782
783// MarkAllNotificationsEmailed marks every currently-unemailed notification for
784// the recipient as emailed=1. Used when a user re-enables email notifications so
785// the pre-existing backlog isn't dispatched in one digest.
786func MarkAllNotificationsEmailed(e Execer, recipientDid string) error {
787 _, err := e.Exec(
788 `UPDATE notifications SET emailed = 1 WHERE recipient_did = ? AND emailed = 0`,
789 recipientDid,
790 )
791 return err
792}
793
794func (d *DB) ClearOldNotifications(ctx context.Context, olderThan time.Duration) error {
795 cutoff := time.Now().Add(-olderThan)
796 createdFilter := orm.FilterLte("created", cutoff)
797
798 query := fmt.Sprintf(`
799 DELETE FROM notifications
800 WHERE %s
801 `, createdFilter.Condition())
802
803 _, err := d.DB.ExecContext(ctx, query, createdFilter.Arg()...)
804 if err != nil {
805 return fmt.Errorf("failed to cleanup old notifications: %w", err)
806 }
807
808 return nil
809}