appview/db: add subscription tables and email digest schema · willow.sh/tangled-core@85bb677

+24

appview/db/db.go

··· 701 701 updated text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')) 702 702 ); 703 703 704 + create table if not exists issue_subscriptions ( 705 + user_did text not null, 706 + issue_id integer not null references issues(id) on delete cascade, 707 + subscribed integer not null default 1, 708 + primary key(user_did, issue_id) 709 + ); 710 + 711 + create table if not exists pull_subscriptions ( 712 + user_did text not null, 713 + pull_id integer not null references pulls(id) on delete cascade, 714 + subscribed integer not null default 1, 715 + primary key(user_did, pull_id) 716 + ); 717 + 704 718 create table if not exists migrations ( 705 719 id integer primary key autoincrement, 706 720 name text unique ··· 713 727 create index if not exists idx_references_to_at on reference_links(to_at); 714 728 create index if not exists idx_webhook_deliveries_webhook_id on webhook_deliveries(webhook_id); 715 729 create index if not exists idx_newsletter_prefs_user_did on newsletter_preferences(user_did); 730 + create index if not exists idx_issue_subscriptions_issue on issue_subscriptions(issue_id, subscribed); 731 + create index if not exists idx_pull_subscriptions_pull on pull_subscriptions(pull_id, subscribed); 716 732 `) 717 733 if err != nil { 718 734 return nil, err ··· 2441 2457 return err 2442 2458 }) 2443 2459 2460 + orm.RunMigration(conn, logger, "add-emailed-to-notifications", func(tx *sql.Tx) error { 2461 + _, err := tx.Exec(` 2462 + ALTER TABLE notifications ADD COLUMN emailed INTEGER NOT NULL DEFAULT 0; 2463 + CREATE INDEX IF NOT EXISTS idx_notifications_emailed 2464 + ON notifications(recipient_did, emailed, created); 2465 + `) 2466 + return err 2467 + }) 2444 2468 return &DB{ 2445 2469 db, 2446 2470 logger,

+196

appview/db/notifications.go

View file

··· 550 550 return nil 551 551 } 552 552 553 + // GetPendingEmailDigestRecipients returns DIDs of users who have email 554 + // notifications enabled, have a verified primary email, and have unemaileD 555 + // notifications older than olderThan for email-eligible notification types. 556 + func GetPendingEmailDigestRecipients(e Execer, olderThan time.Time) ([]string, error) { 557 + placeholders := make([]string, len(models.EmailNotificationTypes)) 558 + args := []any{olderThan.UTC().Format(time.RFC3339)} 559 + for i, t := range models.EmailNotificationTypes { 560 + placeholders[i] = "?" 561 + args = append(args, string(t)) 562 + } 563 + inClause := strings.Join(placeholders, ", ") 564 + 565 + query := fmt.Sprintf(` 566 + SELECT DISTINCT n.recipient_did 567 + FROM notifications n 568 + JOIN notification_preferences np ON np.user_did = n.recipient_did 569 + JOIN emails e ON e.did = n.recipient_did AND e.is_primary = 1 AND e.verified = 1 570 + WHERE n.emailed = 0 571 + AND n.created < ? 572 + AND np.email_notifications = 1 573 + AND n.type IN (%s) 574 + `, inClause) 575 + 576 + rows, err := e.Query(query, args...) 577 + if err != nil { 578 + return nil, fmt.Errorf("failed to query email digest recipients: %w", err) 579 + } 580 + defer rows.Close() 581 + 582 + var dids []string 583 + for rows.Next() { 584 + var did string 585 + if err := rows.Scan(&did); err != nil { 586 + return nil, err 587 + } 588 + dids = append(dids, did) 589 + } 590 + return dids, rows.Err() 591 + } 592 + 593 + // GetPendingNotificationsForEmailDigest returns all unemailed notifications 594 + // older than olderThan for email-eligible types for a specific recipient. 595 + func GetPendingNotificationsForEmailDigest(e Execer, recipientDid string, olderThan time.Time) ([]*models.NotificationWithEntity, error) { 596 + placeholders := make([]string, len(models.EmailNotificationTypes)) 597 + args := []any{recipientDid, olderThan.UTC().Format(time.RFC3339)} 598 + for i, t := range models.EmailNotificationTypes { 599 + placeholders[i] = "?" 600 + args = append(args, string(t)) 601 + } 602 + inClause := strings.Join(placeholders, ", ") 603 + 604 + query := fmt.Sprintf(` 605 + SELECT 606 + n.id, n.recipient_did, n.actor_did, n.type, n.entity_type, n.entity_id, 607 + n.read, n.created, n.repo_id, n.issue_id, n.pull_id, 608 + 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, 609 + 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, 610 + 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 611 + FROM notifications n 612 + LEFT JOIN repos r ON n.repo_id = r.id 613 + LEFT JOIN issues i ON n.issue_id = i.id 614 + LEFT JOIN pulls p ON n.pull_id = p.id 615 + WHERE n.recipient_did = ? 616 + AND n.emailed = 0 617 + AND n.created < ? 618 + AND n.type IN (%s) 619 + ORDER BY n.created DESC 620 + `, inClause) 621 + 622 + rows, err := e.QueryContext(context.Background(), query, args...) 623 + if err != nil { 624 + return nil, fmt.Errorf("failed to query pending email notifications: %w", err) 625 + } 626 + defer rows.Close() 627 + 628 + var notifications []*models.NotificationWithEntity 629 + for rows.Next() { 630 + var n models.Notification 631 + var typeStr string 632 + var createdStr string 633 + var repo models.Repo 634 + var issue models.Issue 635 + var pull models.Pull 636 + var rId, iId, pId sql.NullInt64 637 + var rDid, rRkey, rName, rDescription, rWebsite, rTopicStr sql.NullString 638 + var iDid sql.NullString 639 + var iIssueId sql.NullInt64 640 + var iTitle sql.NullString 641 + var iOpen sql.NullBool 642 + var pOwnerDid sql.NullString 643 + var pPullId sql.NullInt64 644 + var pTitle sql.NullString 645 + var pState sql.NullInt64 646 + 647 + err := rows.Scan( 648 + &n.ID, &n.RecipientDid, &n.ActorDid, &typeStr, &n.EntityType, &n.EntityId, 649 + &n.Read, &createdStr, &n.RepoId, &n.IssueId, &n.PullId, 650 + &rId, &rDid, &rRkey, &rName, &rDescription, &rWebsite, &rTopicStr, 651 + &iId, &iDid, &iIssueId, &iTitle, &iOpen, 652 + &pId, &pOwnerDid, &pPullId, &pTitle, &pState, 653 + ) 654 + if err != nil { 655 + return nil, fmt.Errorf("failed to scan email digest notification: %w", err) 656 + } 657 + 658 + n.Type = models.NotificationType(typeStr) 659 + n.Created, err = time.Parse(time.RFC3339, createdStr) 660 + if err != nil { 661 + return nil, fmt.Errorf("failed to parse created timestamp: %w", err) 662 + } 663 + 664 + entry := &models.NotificationWithEntity{Notification: &n} 665 + 666 + if rId.Valid { 667 + repo.Id = rId.Int64 668 + if rDid.Valid { 669 + repo.Did = rDid.String 670 + } 671 + if rRkey.Valid { 672 + repo.Rkey = rRkey.String 673 + } 674 + if rName.Valid { 675 + repo.Name = rName.String 676 + } 677 + if rDescription.Valid { 678 + repo.Description = rDescription.String 679 + } 680 + if rWebsite.Valid { 681 + repo.Website = rWebsite.String 682 + } 683 + if rTopicStr.Valid { 684 + repo.Topics = strings.Fields(rTopicStr.String) 685 + } 686 + entry.Repo = &repo 687 + } 688 + 689 + if iId.Valid { 690 + issue.Id = iId.Int64 691 + if iDid.Valid { 692 + issue.Did = iDid.String 693 + } 694 + if iIssueId.Valid { 695 + issue.IssueId = int(iIssueId.Int64) 696 + } 697 + if iTitle.Valid { 698 + issue.Title = iTitle.String 699 + } 700 + if iOpen.Valid { 701 + issue.Open = iOpen.Bool 702 + } 703 + entry.Issue = &issue 704 + } 705 + 706 + if pId.Valid { 707 + pull.ID = int(pId.Int64) 708 + if pOwnerDid.Valid { 709 + pull.OwnerDid = pOwnerDid.String 710 + } 711 + if pPullId.Valid { 712 + pull.PullId = int(pPullId.Int64) 713 + } 714 + if pTitle.Valid { 715 + pull.Title = pTitle.String 716 + } 717 + if pState.Valid { 718 + pull.State = models.PullState(pState.Int64) 719 + } 720 + entry.Pull = &pull 721 + } 722 + 723 + notifications = append(notifications, entry) 724 + } 725 + 726 + return notifications, rows.Err() 727 + } 728 + 729 + // MarkNotificationsEmailed marks the given notification IDs as emailed=1. 730 + // Uses explicit IDs (not recipient_did) to avoid racing with new notifications. 731 + func MarkNotificationsEmailed(e Execer, ids []int64) error { 732 + if len(ids) == 0 { 733 + return nil 734 + } 735 + placeholders := make([]string, len(ids)) 736 + args := make([]any, len(ids)) 737 + for i, id := range ids { 738 + placeholders[i] = "?" 739 + args[i] = id 740 + } 741 + query := fmt.Sprintf( 742 + `UPDATE notifications SET emailed = 1 WHERE id IN (%s)`, 743 + strings.Join(placeholders, ", "), 744 + ) 745 + _, err := e.Exec(query, args...) 746 + return err 747 + } 748 + 553 749 func (d *DB) ClearOldNotifications(ctx context.Context, olderThan time.Duration) error { 554 750 cutoff := time.Now().Add(-olderThan) 555 751 createdFilter := orm.FilterLte("created", cutoff)

+149

appview/db/subscriptions.go

View file

··· 1 + package db 2 + 3 + import "database/sql" 4 + 5 + // UpsertIssueSubscription inserts or updates a user's subscription for an issue. 6 + // subscribed=true means they want notifications; subscribed=false means explicitly unsubscribed. 7 + func UpsertIssueSubscription(e Execer, userDid string, issueId int64, subscribed bool) error { 8 + sub := 0 9 + if subscribed { 10 + sub = 1 11 + } 12 + _, err := e.Exec(` 13 + INSERT INTO issue_subscriptions (user_did, issue_id, subscribed) 14 + VALUES (?, ?, ?) 15 + ON CONFLICT(user_did, issue_id) DO UPDATE SET subscribed = excluded.subscribed 16 + `, userDid, issueId, sub) 17 + return err 18 + } 19 + 20 + // UpsertPullSubscription inserts or updates a user's subscription for a pull. 21 + func UpsertPullSubscription(e Execer, userDid string, pullId int64, subscribed bool) error { 22 + sub := 0 23 + if subscribed { 24 + sub = 1 25 + } 26 + _, err := e.Exec(` 27 + INSERT INTO pull_subscriptions (user_did, pull_id, subscribed) 28 + VALUES (?, ?, ?) 29 + ON CONFLICT(user_did, pull_id) DO UPDATE SET subscribed = excluded.subscribed 30 + `, userDid, pullId, sub) 31 + return err 32 + } 33 + 34 + // GetIssueSubscription returns (subscribed, found, err). 35 + // If no row exists, found=false (meaning no explicit subscription). 36 + func GetIssueSubscription(e Execer, userDid string, issueId int64) (subscribed bool, found bool, err error) { 37 + var sub int 38 + err = e.QueryRow(` 39 + SELECT subscribed FROM issue_subscriptions 40 + WHERE user_did = ? AND issue_id = ? 41 + `, userDid, issueId).Scan(&sub) 42 + if err == sql.ErrNoRows { 43 + return false, false, nil 44 + } 45 + if err != nil { 46 + return false, false, err 47 + } 48 + return sub == 1, true, nil 49 + } 50 + 51 + // GetPullSubscription returns (subscribed, found, err). 52 + func GetPullSubscription(e Execer, userDid string, pullId int64) (subscribed bool, found bool, err error) { 53 + var sub int 54 + err = e.QueryRow(` 55 + SELECT subscribed FROM pull_subscriptions 56 + WHERE user_did = ? AND pull_id = ? 57 + `, userDid, pullId).Scan(&sub) 58 + if err == sql.ErrNoRows { 59 + return false, false, nil 60 + } 61 + if err != nil { 62 + return false, false, err 63 + } 64 + return sub == 1, true, nil 65 + } 66 + 67 + // GetIssueSubscribers returns DIDs with subscribed=1 for an issue. 68 + func GetIssueSubscribers(e Execer, issueId int64) ([]string, error) { 69 + rows, err := e.Query(` 70 + SELECT user_did FROM issue_subscriptions 71 + WHERE issue_id = ? AND subscribed = 1 72 + `, issueId) 73 + if err != nil { 74 + return nil, err 75 + } 76 + defer rows.Close() 77 + var dids []string 78 + for rows.Next() { 79 + var did string 80 + if err := rows.Scan(&did); err != nil { 81 + return nil, err 82 + } 83 + dids = append(dids, did) 84 + } 85 + return dids, rows.Err() 86 + } 87 + 88 + // GetIssueUnsubscribers returns DIDs with subscribed=0 for an issue (opted out). 89 + func GetIssueUnsubscribers(e Execer, issueId int64) ([]string, error) { 90 + rows, err := e.Query(` 91 + SELECT user_did FROM issue_subscriptions 92 + WHERE issue_id = ? AND subscribed = 0 93 + `, issueId) 94 + if err != nil { 95 + return nil, err 96 + } 97 + defer rows.Close() 98 + var dids []string 99 + for rows.Next() { 100 + var did string 101 + if err := rows.Scan(&did); err != nil { 102 + return nil, err 103 + } 104 + dids = append(dids, did) 105 + } 106 + return dids, rows.Err() 107 + } 108 + 109 + // GetPullSubscribers returns DIDs with subscribed=1 for a pull. 110 + func GetPullSubscribers(e Execer, pullId int64) ([]string, error) { 111 + rows, err := e.Query(` 112 + SELECT user_did FROM pull_subscriptions 113 + WHERE pull_id = ? AND subscribed = 1 114 + `, pullId) 115 + if err != nil { 116 + return nil, err 117 + } 118 + defer rows.Close() 119 + var dids []string 120 + for rows.Next() { 121 + var did string 122 + if err := rows.Scan(&did); err != nil { 123 + return nil, err 124 + } 125 + dids = append(dids, did) 126 + } 127 + return dids, rows.Err() 128 + } 129 + 130 + // GetPullUnsubscribers returns DIDs with subscribed=0 for a pull (opted out). 131 + func GetPullUnsubscribers(e Execer, pullId int64) ([]string, error) { 132 + rows, err := e.Query(` 133 + SELECT user_did FROM pull_subscriptions 134 + WHERE pull_id = ? AND subscribed = 0 135 + `, pullId) 136 + if err != nil { 137 + return nil, err 138 + } 139 + defer rows.Close() 140 + var dids []string 141 + for rows.Next() { 142 + var did string 143 + if err := rows.Scan(&did); err != nil { 144 + return nil, err 145 + } 146 + dids = append(dids, did) 147 + } 148 + return dids, rows.Err() 149 + }

+20

appview/models/notifications.go

View file

··· 60 60 EntityType string 61 61 EntityId string 62 62 Read bool 63 + Emailed bool 63 64 Created time.Time 64 65 65 66 // foreign key references 66 67 RepoId *int64 67 68 IssueId *int64 68 69 PullId *int64 70 + } 71 + 72 + // EmailNotificationTypes are the notification types that are included in email digests. 73 + // Social types (repo_starred, followed) are intentionally excluded. 74 + var EmailNotificationTypes = []NotificationType{ 75 + NotificationTypeIssueCreated, 76 + NotificationTypeIssueCommented, 77 + NotificationTypeIssueClosed, 78 + NotificationTypeIssueReopen, 79 + NotificationTypePullCreated, 80 + NotificationTypePullCommented, 81 + NotificationTypePullMerged, 82 + NotificationTypePullClosed, 83 + NotificationTypePullReopen, 84 + NotificationTypeUserMentioned, 85 + NotificationTypeIssueAssigned, 86 + NotificationTypeIssueUnassigned, 87 + NotificationTypePullAssigned, 88 + NotificationTypePullUnassigned, 69 89 } 70 90 71 91 // lucide icon that represents this notification

Configure Feed

Configure Feed