Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810package db
import ( "context" "database/sql" "errors" "fmt" "strings" "time"
"github.com/bluesky-social/indigo/atproto/syntax" "tangled.org/core/appview/models" "tangled.org/core/appview/pagination" "tangled.org/core/orm")
func CreateNotification(e Execer, notification *models.Notification) error { query := ` INSERT INTO notifications (recipient_did, actor_did, type, entity_type, entity_id, read, repo_id, issue_id, pull_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) `
result, err := e.Exec(query, notification.RecipientDid, notification.ActorDid, string(notification.Type), notification.EntityType, notification.EntityId, notification.Read, notification.RepoId, notification.IssueId, notification.PullId, ) if err != nil { return fmt.Errorf("failed to create notification: %w", err) }
id, err := result.LastInsertId() if err != nil { return fmt.Errorf("failed to get notification ID: %w", err) }
notification.ID = id return nil}
// GetNotificationsPaginated retrieves notifications with filters and paginationfunc GetNotificationsPaginated(e Execer, page pagination.Page, filters ...orm.Filter) ([]*models.Notification, error) { var conditions []string var args []any
for _, filter := range filters { conditions = append(conditions, filter.Condition()) args = append(args, filter.Arg()...) }
whereClause := "" if len(conditions) > 0 { whereClause = "WHERE " + conditions[0] for _, condition := range conditions[1:] { whereClause += " AND " + condition } } pageClause := "" if page.Limit > 0 { pageClause = " limit ? offset ? " args = append(args, page.Limit, page.Offset) }
query := fmt.Sprintf(` select id, recipient_did, actor_did, type, entity_type, entity_id, read, created, repo_id, issue_id, pull_id from notifications %s order by created desc %s `, whereClause, pageClause)
rows, err := e.QueryContext(context.Background(), query, args...) if err != nil { return nil, fmt.Errorf("failed to query notifications: %w", err) } defer rows.Close()
var notifications []*models.Notification for rows.Next() { var n models.Notification var typeStr string var createdStr string err := rows.Scan( &n.ID, &n.RecipientDid, &n.ActorDid, &typeStr, &n.EntityType, &n.EntityId, &n.Read, &createdStr, &n.RepoId, &n.IssueId, &n.PullId, ) if err != nil { return nil, fmt.Errorf("failed to scan notification: %w", err) } n.Type = models.NotificationType(typeStr) n.Created, err = time.Parse(time.RFC3339, createdStr) if err != nil { return nil, fmt.Errorf("failed to parse created timestamp: %w", err) } notifications = append(notifications, &n) }
return notifications, nil}
func GetNotificationWithEntity(e Execer, notificationID int64, userDID string) (*models.NotificationWithEntity, error) { results, err := GetNotificationsWithEntities(e, pagination.Page{Limit: 1, Offset: 0}, orm.FilterEq("n.id", notificationID), orm.FilterEq("n.recipient_did", userDID), ) if err != nil { return nil, err } if len(results) == 0 { return nil, fmt.Errorf("notification not found") } return results[0], nil}
// GetNotificationsWithEntities retrieves notifications with their related entitiesfunc GetNotificationsWithEntities(e Execer, page pagination.Page, filters ...orm.Filter) ([]*models.NotificationWithEntity, error) { var conditions []string var args []any
for _, filter := range filters { conditions = append(conditions, filter.Condition()) args = append(args, filter.Arg()...) }
whereClause := "" if len(conditions) > 0 { whereClause = "WHERE " + conditions[0] for _, condition := range conditions[1:] { whereClause += " AND " + condition } }
query := fmt.Sprintf(` select n.id, n.recipient_did, n.actor_did, n.type, n.entity_type, n.entity_id, n.read, n.created, n.repo_id, n.issue_id, n.pull_id, 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, 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, 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 from notifications n left join repos r on n.repo_id = r.id left join issues i on n.issue_id = i.id left join pulls p on n.pull_id = p.id %s order by n.created desc limit ? offset ? `, whereClause)
args = append(args, page.Limit, page.Offset)
rows, err := e.QueryContext(context.Background(), query, args...) if err != nil { return nil, fmt.Errorf("failed to query notifications with entities: %w", err) } defer rows.Close()
var notifications []*models.NotificationWithEntity for rows.Next() { var n models.Notification var typeStr string var createdStr string var repo models.Repo var issue models.Issue var pull models.Pull var rId, iId, pId sql.NullInt64 var rDid, rRkey, rName, rDescription, rWebsite, rTopicStr sql.NullString var iDid sql.NullString var iIssueId sql.NullInt64 var iTitle sql.NullString var iOpen sql.NullBool var pOwnerDid sql.NullString var pPullId sql.NullInt64 var pTitle sql.NullString var pState sql.NullInt64
err := rows.Scan( &n.ID, &n.RecipientDid, &n.ActorDid, &typeStr, &n.EntityType, &n.EntityId, &n.Read, &createdStr, &n.RepoId, &n.IssueId, &n.PullId, &rId, &rDid, &rRkey, &rName, &rDescription, &rWebsite, &rTopicStr, &iId, &iDid, &iIssueId, &iTitle, &iOpen, &pId, &pOwnerDid, &pPullId, &pTitle, &pState, ) if err != nil { return nil, fmt.Errorf("failed to scan notification with entities: %w", err) }
n.Type = models.NotificationType(typeStr) n.Created, err = time.Parse(time.RFC3339, createdStr) if err != nil { return nil, fmt.Errorf("failed to parse created timestamp: %w", err) }
entry := &models.NotificationWithEntity{Notification: &n}
// populate repo if present if rId.Valid { repo.Id = rId.Int64 if rDid.Valid { repo.Did = rDid.String } if rRkey.Valid { repo.Rkey = rRkey.String } if rName.Valid { repo.Name = rName.String } if rDescription.Valid { repo.Description = rDescription.String } if rWebsite.Valid { repo.Website = rWebsite.String } if rTopicStr.Valid { repo.Topics = strings.Fields(rTopicStr.String) } entry.Repo = &repo }
// populate issue if present if iId.Valid { issue.Id = iId.Int64 if iDid.Valid { issue.Did = iDid.String } if iIssueId.Valid { issue.IssueId = int(iIssueId.Int64) } if iTitle.Valid { issue.Title = iTitle.String } if iOpen.Valid { issue.Open = iOpen.Bool } entry.Issue = &issue }
// populate pull if present if pId.Valid { pull.ID = int(pId.Int64) if pOwnerDid.Valid { pull.OwnerDid = pOwnerDid.String } if pPullId.Valid { pull.PullId = int(pPullId.Int64) } if pTitle.Valid { pull.Title = pTitle.String } if pState.Valid { pull.State = models.PullState(pState.Int64) } entry.Pull = &pull }
notifications = append(notifications, entry) }
return notifications, nil}
// GetNotifications retrieves notifications with filtersfunc GetNotifications(e Execer, filters ...orm.Filter) ([]*models.Notification, error) { return GetNotificationsPaginated(e, pagination.FirstPage(), filters...)}
func CountNotifications(e Execer, filters ...orm.Filter) (int64, error) { var conditions []string var args []any for _, filter := range filters { conditions = append(conditions, filter.Condition()) args = append(args, filter.Arg()...) }
whereClause := "" if conditions != nil { whereClause = " where " + strings.Join(conditions, " and ") }
query := fmt.Sprintf(`select count(1) from notifications %s`, whereClause) var count int64 err := e.QueryRow(query, args...).Scan(&count)
if !errors.Is(err, sql.ErrNoRows) && err != nil { return 0, err }
return count, nil}
func MarkNotificationRead(e Execer, notificationID int64, userDID string) error { idFilter := orm.FilterEq("id", notificationID) recipientFilter := orm.FilterEq("recipient_did", userDID)
query := fmt.Sprintf(` UPDATE notifications SET read = 1 WHERE %s AND %s `, idFilter.Condition(), recipientFilter.Condition())
args := append(idFilter.Arg(), recipientFilter.Arg()...)
result, err := e.Exec(query, args...) if err != nil { return fmt.Errorf("failed to mark notification as read: %w", err) }
rowsAffected, err := result.RowsAffected() if err != nil { return fmt.Errorf("failed to get rows affected: %w", err) }
if rowsAffected == 0 { return fmt.Errorf("notification not found or access denied") }
return nil}
func MarkNotificationsReadForIssue(e Execer, userDID, repoDid string, issueNum int) error { query := ` update notifications set read = 1 where recipient_did = ? and read = 0 and issue_id = (select id from issues where repo_did = ? and issue_id = ?) ` _, err := e.Exec(query, userDID, repoDid, issueNum) return err}
func MarkNotificationsReadForPull(e Execer, userDID, repoDid string, pullNum int) error { query := ` update notifications set read = 1 where recipient_did = ? and read = 0 and pull_id = (select p.id from pulls p where p.pull_id = ? and p.repo_did = ?) ` _, err := e.Exec(query, userDID, pullNum, repoDid) return err}
func MarkNotificationUnread(e Execer, notificationID int64, userDID string) error { idFilter := orm.FilterEq("id", notificationID) recipientFilter := orm.FilterEq("recipient_did", userDID)
query := fmt.Sprintf(` UPDATE notifications SET read = 0 WHERE %s AND %s `, idFilter.Condition(), recipientFilter.Condition())
args := append(idFilter.Arg(), recipientFilter.Arg()...)
result, err := e.Exec(query, args...) if err != nil { return fmt.Errorf("failed to mark notification as unread: %w", err) }
rowsAffected, err := result.RowsAffected() if err != nil { return fmt.Errorf("failed to get rows affected: %w", err) }
if rowsAffected == 0 { return fmt.Errorf("notification not found or access denied") }
return nil}
func MarkAllNotificationsRead(e Execer, userDID string) error { recipientFilter := orm.FilterEq("recipient_did", userDID) readFilter := orm.FilterEq("read", 0)
query := fmt.Sprintf(` UPDATE notifications SET read = 1 WHERE %s AND %s `, recipientFilter.Condition(), readFilter.Condition())
args := append(recipientFilter.Arg(), readFilter.Arg()...)
_, err := e.Exec(query, args...) if err != nil { return fmt.Errorf("failed to mark all notifications as read: %w", err) }
return nil}
func DeleteNotification(e Execer, notificationID int64, userDID string) error { idFilter := orm.FilterEq("id", notificationID) recipientFilter := orm.FilterEq("recipient_did", userDID)
query := fmt.Sprintf(` DELETE FROM notifications WHERE %s AND %s `, idFilter.Condition(), recipientFilter.Condition())
args := append(idFilter.Arg(), recipientFilter.Arg()...)
result, err := e.Exec(query, args...) if err != nil { return fmt.Errorf("failed to delete notification: %w", err) }
rowsAffected, err := result.RowsAffected() if err != nil { return fmt.Errorf("failed to get rows affected: %w", err) }
if rowsAffected == 0 { return fmt.Errorf("notification not found or access denied") }
return nil}
func GetNotificationPreference(e Execer, userDid string) (*models.NotificationPreferences, error) { prefs, err := GetNotificationPreferences(e, orm.FilterEq("user_did", userDid)) if err != nil { return nil, err }
p, ok := prefs[syntax.DID(userDid)] if !ok { return models.DefaultNotificationPreferences(syntax.DID(userDid)), nil }
return p, nil}
func GetNotificationPreferences(e Execer, filters ...orm.Filter) (map[syntax.DID]*models.NotificationPreferences, error) { prefsMap := make(map[syntax.DID]*models.NotificationPreferences)
var conditions []string var args []any for _, filter := range filters { conditions = append(conditions, filter.Condition()) args = append(args, filter.Arg()...) }
whereClause := "" if conditions != nil { whereClause = " where " + strings.Join(conditions, " and ") }
query := fmt.Sprintf(` select id, user_did, repo_starred, issue_created, issue_commented, pull_created, pull_commented, followed, user_mentioned, pull_merged, issue_closed, email_notifications from notification_preferences %s `, whereClause)
rows, err := e.Query(query, args...) if err != nil { return nil, err } defer rows.Close()
for rows.Next() { var prefs models.NotificationPreferences if err := rows.Scan( &prefs.ID, &prefs.UserDid, &prefs.RepoStarred, &prefs.IssueCreated, &prefs.IssueCommented, &prefs.PullCreated, &prefs.PullCommented, &prefs.Followed, &prefs.UserMentioned, &prefs.PullMerged, &prefs.IssueClosed, &prefs.EmailNotifications, ); err != nil { return nil, err }
prefsMap[prefs.UserDid] = &prefs }
if err := rows.Err(); err != nil { return nil, err }
return prefsMap, nil}
func (d *DB) UpdateNotificationPreferences(ctx context.Context, prefs *models.NotificationPreferences) error { tx, err := d.DB.BeginTx(ctx, nil) if err != nil { return fmt.Errorf("failed to begin transaction: %w", err) } defer tx.Rollback()
var prevEmailEnabled bool var hadPrefs bool row := tx.QueryRowContext(ctx, `SELECT email_notifications FROM notification_preferences WHERE user_did = ?`, prefs.UserDid, ) switch err := row.Scan(&prevEmailEnabled); err { case nil: hadPrefs = true case sql.ErrNoRows: hadPrefs = false default: return fmt.Errorf("failed to read existing preferences: %w", err) }
query := ` INSERT OR REPLACE INTO notification_preferences (user_did, repo_starred, issue_created, issue_commented, pull_created, pull_commented, followed, user_mentioned, pull_merged, issue_closed, email_notifications) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) `
result, err := tx.ExecContext(ctx, query, prefs.UserDid, prefs.RepoStarred, prefs.IssueCreated, prefs.IssueCommented, prefs.PullCreated, prefs.PullCommented, prefs.Followed, prefs.UserMentioned, prefs.PullMerged, prefs.IssueClosed, prefs.EmailNotifications, ) if err != nil { return fmt.Errorf("failed to update notification preferences: %w", err) }
// on enabling email notifications (from disabled or from no prior setting), // mark existing unemailed notifications as emailed so they aren't dispatched // as a backlog on the next digest tick. if prefs.EmailNotifications && (!hadPrefs || !prevEmailEnabled) { if err := MarkAllNotificationsEmailed(tx, string(prefs.UserDid)); err != nil { return fmt.Errorf("failed to suppress notification backlog: %w", err) } }
if prefs.ID == 0 { id, err := result.LastInsertId() if err != nil { return fmt.Errorf("failed to get preferences ID: %w", err) } prefs.ID = id }
return tx.Commit()}
// GetPendingEmailDigestRecipients returns DIDs of users who have email// notifications enabled, have a verified primary email, and have unread,// unemailed notifications older than olderThan for email-eligible notification// types.func GetPendingEmailDigestRecipients(e Execer, olderThan time.Time) ([]string, error) { placeholders := make([]string, len(models.EmailNotificationTypes)) args := []any{olderThan.UTC().Format(time.RFC3339)} for i, t := range models.EmailNotificationTypes { placeholders[i] = "?" args = append(args, string(t)) } inClause := strings.Join(placeholders, ", ")
query := fmt.Sprintf(` SELECT DISTINCT n.recipient_did FROM notifications n JOIN notification_preferences np ON np.user_did = n.recipient_did JOIN emails e ON e.did = n.recipient_did AND e.is_primary = 1 AND e.verified = 1 WHERE n.emailed = 0 AND n.read = 0 AND n.created < ? AND np.email_notifications = 1 AND n.type IN (%s) `, inClause)
rows, err := e.Query(query, args...) if err != nil { return nil, fmt.Errorf("failed to query email digest recipients: %w", err) } defer rows.Close()
var dids []string for rows.Next() { var did string if err := rows.Scan(&did); err != nil { return nil, err } dids = append(dids, did) } return dids, rows.Err()}
// GetPendingNotificationsForEmailDigest returns all unread, unemailed// notifications older than olderThan for email-eligible types for a specific// recipient.func GetPendingNotificationsForEmailDigest(e Execer, recipientDid string, olderThan time.Time) ([]*models.NotificationWithEntity, error) { placeholders := make([]string, len(models.EmailNotificationTypes)) args := []any{recipientDid, olderThan.UTC().Format(time.RFC3339)} for i, t := range models.EmailNotificationTypes { placeholders[i] = "?" args = append(args, string(t)) } inClause := strings.Join(placeholders, ", ")
query := fmt.Sprintf(` SELECT n.id, n.recipient_did, n.actor_did, n.type, n.entity_type, n.entity_id, n.read, n.created, n.repo_id, n.issue_id, n.pull_id, 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, 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, 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 FROM notifications n LEFT JOIN repos r ON n.repo_id = r.id LEFT JOIN issues i ON n.issue_id = i.id LEFT JOIN pulls p ON n.pull_id = p.id WHERE n.recipient_did = ? AND n.emailed = 0 AND n.read = 0 AND n.created < ? AND n.type IN (%s) ORDER BY n.created DESC `, inClause)
rows, err := e.QueryContext(context.Background(), query, args...) if err != nil { return nil, fmt.Errorf("failed to query pending email notifications: %w", err) } defer rows.Close()
var notifications []*models.NotificationWithEntity for rows.Next() { var n models.Notification var typeStr string var createdStr string var repo models.Repo var issue models.Issue var pull models.Pull var rId, iId, pId sql.NullInt64 var rDid, rRkey, rName, rDescription, rWebsite, rTopicStr sql.NullString var iDid sql.NullString var iIssueId sql.NullInt64 var iTitle sql.NullString var iOpen sql.NullBool var pOwnerDid sql.NullString var pPullId sql.NullInt64 var pTitle sql.NullString var pState sql.NullInt64
err := rows.Scan( &n.ID, &n.RecipientDid, &n.ActorDid, &typeStr, &n.EntityType, &n.EntityId, &n.Read, &createdStr, &n.RepoId, &n.IssueId, &n.PullId, &rId, &rDid, &rRkey, &rName, &rDescription, &rWebsite, &rTopicStr, &iId, &iDid, &iIssueId, &iTitle, &iOpen, &pId, &pOwnerDid, &pPullId, &pTitle, &pState, ) if err != nil { return nil, fmt.Errorf("failed to scan email digest notification: %w", err) }
n.Type = models.NotificationType(typeStr) n.Created, err = time.Parse(time.RFC3339, createdStr) if err != nil { return nil, fmt.Errorf("failed to parse created timestamp: %w", err) }
entry := &models.NotificationWithEntity{Notification: &n}
if rId.Valid { repo.Id = rId.Int64 if rDid.Valid { repo.Did = rDid.String } if rRkey.Valid { repo.Rkey = rRkey.String } if rName.Valid { repo.Name = rName.String } if rDescription.Valid { repo.Description = rDescription.String } if rWebsite.Valid { repo.Website = rWebsite.String } if rTopicStr.Valid { repo.Topics = strings.Fields(rTopicStr.String) } entry.Repo = &repo }
if iId.Valid { issue.Id = iId.Int64 if iDid.Valid { issue.Did = iDid.String } if iIssueId.Valid { issue.IssueId = int(iIssueId.Int64) } if iTitle.Valid { issue.Title = iTitle.String } if iOpen.Valid { issue.Open = iOpen.Bool } entry.Issue = &issue }
if pId.Valid { pull.ID = int(pId.Int64) if pOwnerDid.Valid { pull.OwnerDid = pOwnerDid.String } if pPullId.Valid { pull.PullId = int(pPullId.Int64) } if pTitle.Valid { pull.Title = pTitle.String } if pState.Valid { pull.State = models.PullState(pState.Int64) } entry.Pull = &pull }
notifications = append(notifications, entry) }
return notifications, rows.Err()}
// MarkNotificationsEmailed marks the given notification IDs as emailed=1.// Uses explicit IDs (not recipient_did) to avoid racing with new notifications.func MarkNotificationsEmailed(e Execer, ids []int64) error { if len(ids) == 0 { return nil } placeholders := make([]string, len(ids)) args := make([]any, len(ids)) for i, id := range ids { placeholders[i] = "?" args[i] = id } query := fmt.Sprintf( `UPDATE notifications SET emailed = 1 WHERE id IN (%s)`, strings.Join(placeholders, ", "), ) _, err := e.Exec(query, args...) return err}
// MarkAllNotificationsEmailed marks every currently-unemailed notification for// the recipient as emailed=1. Used when a user re-enables email notifications so// the pre-existing backlog isn't dispatched in one digest.func MarkAllNotificationsEmailed(e Execer, recipientDid string) error { _, err := e.Exec( `UPDATE notifications SET emailed = 1 WHERE recipient_did = ? AND emailed = 0`, recipientDid, ) return err}
func (d *DB) ClearOldNotifications(ctx context.Context, olderThan time.Duration) error { cutoff := time.Now().Add(-olderThan) createdFilter := orm.FilterLte("created", cutoff)
query := fmt.Sprintf(` DELETE FROM notifications WHERE %s `, createdFilter.Condition())
_, err := d.DB.ExecContext(ctx, query, createdFilter.Arg()...) if err != nil { return fmt.Errorf("failed to cleanup old notifications: %w", err) }
return nil}