Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658package db
import ( "context" "slices"
"github.com/bluesky-social/indigo/atproto/syntax" "tangled.org/core/api/tangled" "tangled.org/core/appview/db" "tangled.org/core/appview/models" "tangled.org/core/appview/notify" "tangled.org/core/idresolver" "tangled.org/core/log" "tangled.org/core/orm" "tangled.org/core/sets")
const ( maxMentions = 8 assigneeLabelAt = "at://did:plc:wshs7t2adsemcrrd4snkeqli/sh.tangled.label.definition/assignee")
type databaseNotifier struct { db *db.DB res *idresolver.Resolver}
func NewDatabaseNotifier(database *db.DB, resolver *idresolver.Resolver) notify.Notifier { return &databaseNotifier{ db: database, res: resolver, }}
var _ notify.Notifier = &databaseNotifier{}
func (n *databaseNotifier) NewRepo(ctx context.Context, repo *models.Repo) { // no-op for now}func (n *databaseNotifier) DeleteRepo(ctx context.Context, repo *models.Repo) { // no-op for now}
func (n *databaseNotifier) RenameRepo(ctx context.Context, actor syntax.DID, oldRepo, newRepo *models.Repo) {}
func (n *databaseNotifier) NewStar(ctx context.Context, star *models.Star) { l := log.FromContext(ctx)
if star.SubjectType != models.StarSubjectRepo { return }
repo, err := db.GetRepo(n.db, orm.FilterEq("repo_did", star.Subject)) if err != nil { l.Error("failed to get repos", "err", err) return }
actorDid := syntax.DID(star.Did) recipients := sets.Singleton(syntax.DID(repo.Did)) eventType := models.NotificationTypeRepoStarred entityType := "repo" entityId := star.Subject repoId := &repo.Id var issueId *int64 var pullId *int64
n.notifyEvent( ctx, actorDid, recipients, eventType, entityType, entityId, repoId, issueId, pullId, )}
func (n *databaseNotifier) DeleteStar(ctx context.Context, star *models.Star) { // no-op}
func (n *databaseNotifier) NewComment(ctx context.Context, comment *models.Comment, mentions []syntax.DID) { l := log.FromContext(ctx)
var ( // built the recipients list: // - the owner of the repo // - | if the comment is a reply -> everybody on that thread // | if the comment is a top level -> just the issue owner // - remove mentioned users from the recipients list recipients = sets.New[syntax.DID]() entityType string entityId string repoId *int64 issueId *int64 pullId *int64 )
subjectAt := syntax.ATURI(comment.Subject.Uri)
switch subjectAt.Collection() { case tangled.RepoIssueNSID: issues, err := db.GetIssues( n.db, orm.FilterEq("at_uri", subjectAt), ) if err != nil { l.Error("failed to get issues", "err", err) return } if len(issues) == 0 { l.Error("no issue found", "subject", comment.Subject) return } issue := issues[0]
recipients.Insert(syntax.DID(issue.Repo.Did)) if comment.IsReply() { // if this comment is a reply, then notify everybody in that thread parent := *comment.ReplyTo
// find the parent thread, and add all DIDs from here to the recipient list for _, t := range models.NewCommentList(issue.Comments) { if t.Self.AtUri() == syntax.ATURI(parent.Uri) { for _, p := range t.Participants() { recipients.Insert(p) } } } } else { // not a reply, notify just the issue author recipients.Insert(syntax.DID(issue.Did)) }
entityType = "issue" entityId = issue.AtUri().String() repoId = &issue.Repo.Id issueId = &issue.Id
for _, m := range mentions { recipients.Remove(m) }
n.notifyEvent( ctx, comment.Did, recipients, models.NotificationTypeIssueCommented, entityType, entityId, repoId, issueId, pullId, )
case tangled.RepoPullNSID: pull, err := db.GetPull( n.db, orm.FilterEq("owner_did", subjectAt.Authority()), orm.FilterEq("rkey", subjectAt.RecordKey()), ) if err != nil { l.Error("NewComment: failed to get pull", "err", err) return }
pull.Repo, err = db.GetRepo(n.db, orm.FilterEq("repo_did", pull.RepoDid)) if err != nil { l.Error("NewComment: failed to get repo", "err", err) return }
recipients.Insert(syntax.DID(pull.Repo.Did)) for _, p := range pull.Participants() { recipients.Insert(syntax.DID(p)) }
entityType = "pull" entityId = pull.AtUri().String() repoId = &pull.Repo.Id p := int64(pull.ID) pullId = &p
for _, m := range mentions { recipients.Remove(m) }
n.notifyEvent( ctx, comment.Did, recipients, models.NotificationTypePullCommented, entityType, entityId, repoId, issueId, pullId, ) default: return // no-op }
n.notifyEvent( ctx, comment.Did, sets.Collect(slices.Values(mentions)), models.NotificationTypeUserMentioned, entityType, entityId, repoId, issueId, pullId, )}
func (n *databaseNotifier) DeleteComment(ctx context.Context, comment *models.Comment) { // no-op}
func (n *databaseNotifier) NewIssue(ctx context.Context, issue *models.Issue, mentions []syntax.DID) { l := log.FromContext(ctx)
collaborators, err := db.GetCollaborators(n.db, orm.FilterEq("repo_did", string(issue.RepoDid))) if err != nil { l.Error("failed to fetch collaborators", "err", err) return }
// build the recipients list // - owner of the repo // - collaborators in the repo // - remove users already mentioned recipients := sets.Singleton(syntax.DID(issue.Repo.Did)) for _, c := range collaborators { recipients.Insert(c.SubjectDid) } for _, m := range mentions { recipients.Remove(m) }
actorDid := syntax.DID(issue.Did) entityType := "issue" entityId := issue.AtUri().String() repoId := &issue.Repo.Id issueId := &issue.Id var pullId *int64
n.notifyEvent( ctx, actorDid, recipients, models.NotificationTypeIssueCreated, entityType, entityId, repoId, issueId, pullId, ) n.notifyEvent( ctx, actorDid, sets.Collect(slices.Values(mentions)), models.NotificationTypeUserMentioned, entityType, entityId, repoId, issueId, pullId, )}
func (n *databaseNotifier) DeleteIssue(ctx context.Context, issue *models.Issue) { // no-op for now}
func (n *databaseNotifier) NewIssueLabelOp(ctx context.Context, actor syntax.DID, issue *models.Issue, ops []models.LabelOp) { entityType := "issue" entityId := issue.AtUri().String() repoId := &issue.Repo.Id issueId := &issue.Id var pullId *int64
assigned := sets.New[syntax.DID]() unassigned := sets.New[syntax.DID]() for _, op := range ops { if op.OperandKey != assigneeLabelAt { continue } assignee := syntax.DID(op.OperandValue) switch op.Operation { case models.LabelOperationAdd: assigned.Insert(assignee) case models.LabelOperationDel: unassigned.Insert(assignee) default: continue } }
n.notifyEvent(ctx, actor, assigned, models.NotificationTypeIssueAssigned, entityType, entityId, repoId, issueId, pullId) n.notifyEvent(ctx, actor, unassigned, models.NotificationTypeIssueUnassigned, entityType, entityId, repoId, issueId, pullId)}
func (n *databaseNotifier) NewPullLabelOp(ctx context.Context, actor syntax.DID, pull *models.Pull, ops []models.LabelOp) { entityType := "pull" entityId := pull.AtUri().String() repoId := &pull.Repo.Id var issueId *int64 p := int64(pull.ID) pullId := &p
assigned := sets.New[syntax.DID]() unassigned := sets.New[syntax.DID]() for _, op := range ops { if op.OperandKey != assigneeLabelAt { continue } assignee := syntax.DID(op.OperandValue) switch op.Operation { case models.LabelOperationAdd: assigned.Insert(assignee) case models.LabelOperationDel: unassigned.Insert(assignee) default: continue } }
n.notifyEvent(ctx, actor, assigned, models.NotificationTypePullAssigned, entityType, entityId, repoId, issueId, pullId) n.notifyEvent(ctx, actor, unassigned, models.NotificationTypePullUnassigned, entityType, entityId, repoId, issueId, pullId)}
func (n *databaseNotifier) NewFollow(ctx context.Context, follow *models.Follow) { actorDid := syntax.DID(follow.UserDid) recipients := sets.Singleton(syntax.DID(follow.SubjectDid)) eventType := models.NotificationTypeFollowed entityType := "follow" entityId := follow.UserDid var repoId, issueId, pullId *int64
n.notifyEvent( ctx, actorDid, recipients, eventType, entityType, entityId, repoId, issueId, pullId, )}
func (n *databaseNotifier) DeleteFollow(ctx context.Context, follow *models.Follow) { // no-op}
func (n *databaseNotifier) NewPull(ctx context.Context, pull *models.Pull) { l := log.FromContext(ctx)
repo, err := db.GetRepo(n.db, orm.FilterEq("repo_did", string(pull.RepoDid))) if err != nil { l.Error("failed to get repos", "err", err) return } collaborators, err := db.GetCollaborators(n.db, orm.FilterEq("repo_did", string(pull.RepoDid))) if err != nil { l.Error("failed to fetch collaborators", "err", err) return }
// build the recipients list // - owner of the repo // - collaborators in the repo recipients := sets.Singleton(syntax.DID(repo.Did)) for _, c := range collaborators { recipients.Insert(c.SubjectDid) }
actorDid := syntax.DID(pull.OwnerDid) eventType := models.NotificationTypePullCreated entityType := "pull" entityId := pull.AtUri().String() repoId := &repo.Id var issueId *int64 p := int64(pull.ID) pullId := &p
n.notifyEvent( ctx, actorDid, recipients, eventType, entityType, entityId, repoId, issueId, pullId, )}
func (n *databaseNotifier) UpdateProfile(ctx context.Context, profile *models.Profile) { // no-op}
func (n *databaseNotifier) DeleteString(ctx context.Context, did, rkey string) { // no-op}
func (n *databaseNotifier) EditString(ctx context.Context, string *models.String) { // no-op}
func (n *databaseNotifier) NewString(ctx context.Context, string *models.String) { // no-op}
func (n *databaseNotifier) Push(ctx context.Context, repo *models.Repo, ref, oldSha, newSha, committerDid string) { // no-op for now; webhooks are handled by the webhook notifier}
func (n *databaseNotifier) Clone(ctx context.Context, repo *models.Repo) { // no-op}
func (n *databaseNotifier) NewIssueState(ctx context.Context, actor syntax.DID, issue *models.Issue) { l := log.FromContext(ctx)
collaborators, err := db.GetCollaborators(n.db, orm.FilterEq("repo_did", string(issue.RepoDid))) if err != nil { l.Error("failed to fetch collaborators", "err", err) return }
// build up the recipients list: // - repo owner // - repo collaborators // - all issue participants recipients := sets.Singleton(syntax.DID(issue.Repo.Did)) for _, c := range collaborators { recipients.Insert(c.SubjectDid) } for _, p := range issue.Participants() { recipients.Insert(syntax.DID(p)) }
entityType := "issue" entityId := issue.AtUri().String() repoId := &issue.Repo.Id issueId := &issue.Id var pullId *int64 var eventType models.NotificationType
if issue.Open { eventType = models.NotificationTypeIssueReopen } else { eventType = models.NotificationTypeIssueClosed }
n.notifyEvent( ctx, actor, recipients, eventType, entityType, entityId, repoId, issueId, pullId, )}
func (n *databaseNotifier) ResubmitPull(ctx context.Context, pull *models.Pull) { // no-op for now}
func (n *databaseNotifier) NewPullState(ctx context.Context, actor syntax.DID, pull *models.Pull) { l := log.FromContext(ctx)
// Get repo details repo, err := db.GetRepo(n.db, orm.FilterEq("repo_did", string(pull.RepoDid))) if err != nil { l.Error("failed to get repos", "err", err) return }
collaborators, err := db.GetCollaborators(n.db, orm.FilterEq("repo_did", string(pull.RepoDid))) if err != nil { l.Error("failed to fetch collaborators", "err", err) return }
// build up the recipients list: // - repo owner // - all pull participants recipients := sets.Singleton(syntax.DID(repo.Did)) for _, c := range collaborators { recipients.Insert(c.SubjectDid) } for _, p := range pull.Participants() { recipients.Insert(p) }
entityType := "pull" entityId := pull.AtUri().String() repoId := &repo.Id var issueId *int64 var eventType models.NotificationType switch pull.State { case models.PullClosed: eventType = models.NotificationTypePullClosed case models.PullOpen: eventType = models.NotificationTypePullReopen case models.PullMerged: eventType = models.NotificationTypePullMerged default: l.Error("unexpected new PR state", "state", pull.State) return } p := int64(pull.ID) pullId := &p
n.notifyEvent( ctx, actor, recipients, eventType, entityType, entityId, repoId, issueId, pullId, )}
func (n *databaseNotifier) notifyEvent( ctx context.Context, actorDid syntax.DID, recipients sets.Set[syntax.DID], eventType models.NotificationType, entityType string, entityId string, repoId *int64, issueId *int64, pullId *int64,) { l := log.FromContext(ctx)
// if the user is attempting to mention >maxMentions users, this is probably spam, do not mention anybody if eventType == models.NotificationTypeUserMentioned && recipients.Len() > maxMentions { return }
// Apply subscription overrides for thread-activity events only. // Mention and assignment events are targeted at specific users and should // not be broadcast to all thread subscribers. isThreadEvent := eventType != models.NotificationTypeUserMentioned && eventType != models.NotificationTypeIssueAssigned && eventType != models.NotificationTypeIssueUnassigned && eventType != models.NotificationTypePullAssigned && eventType != models.NotificationTypePullUnassigned
switch { case issueId != nil && isThreadEvent: if subs, err := db.GetIssueSubscribers(n.db, *issueId); err == nil { for _, did := range subs { recipients.Insert(syntax.DID(did)) } } if unsubs, err := db.GetIssueUnsubscribers(n.db, *issueId); err == nil { for _, did := range unsubs { recipients.Remove(syntax.DID(did)) } } case pullId != nil && isThreadEvent: if subs, err := db.GetPullSubscribers(n.db, *pullId); err == nil { for _, did := range subs { recipients.Insert(syntax.DID(did)) } } if unsubs, err := db.GetPullUnsubscribers(n.db, *pullId); err == nil { for _, did := range unsubs { recipients.Remove(syntax.DID(did)) } } }
recipients.Remove(actorDid)
// Auto-subscribe the actor to this issue/pull when they interact with it. // This happens outside the transaction since it’s best-effort. switch { case issueId != nil: if err := db.UpsertIssueSubscription(n.db, actorDid.String(), *issueId, true); err != nil { l.Warn("failed to auto-subscribe actor to issue", "actor", actorDid, "issueId", *issueId, "err", err) } case pullId != nil: if err := db.UpsertPullSubscription(n.db, actorDid.String(), *pullId, true); err != nil { l.Warn("failed to auto-subscribe actor to pull", "actor", actorDid, "pullId", *pullId, "err", err) } }
prefMap, err := db.GetNotificationPreferences( n.db, orm.FilterIn("user_did", slices.Collect(recipients.All())), ) if err != nil { // failed to get prefs for users return }
// create a transaction for bulk notification storage tx, err := n.db.Begin() if err != nil { // failed to start tx return } defer tx.Rollback()
// filter based on preferences for recipientDid := range recipients.All() { prefs, ok := prefMap[recipientDid] if !ok { prefs = models.DefaultNotificationPreferences(recipientDid) }
// skip users who don’t want this type if !prefs.ShouldNotify(eventType) { continue }
// create notification notif := &models.Notification{ RecipientDid: recipientDid.String(), ActorDid: actorDid.String(), Type: eventType, EntityType: entityType, EntityId: entityId, RepoId: repoId, IssueId: issueId, PullId: pullId, }
if err := db.CreateNotification(tx, notif); err != nil { l.Error("failed to create notification", "recipientDid", recipientDid, "err", err) } }
if err := tx.Commit(); err != nil { // failed to commit return }}