Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161package deliberi
import ( "context" "log/slog" "maps" "sync" "time"
"github.com/samber/lo" deldb "tangled.org/core/deliberi/db" "tangled.org/core/deliberi/models" "tangled.org/core/orm")
const ( inviteReconcileInterval = 10 * time.Second inviteFetchTimeout = 2 * time.Second)
var inviteNotificationTypes = []models.NotificationType{ models.NotificationTypeCollaboratorInvited,}
type InviteSync struct { db *deldb.DB invites inviteLister logger *slog.Logger
mu sync.Mutex lastRun map[string]time.Time swept time.Time}
func NewInviteSync(database *deldb.DB, invites inviteLister, logger *slog.Logger) *InviteSync { return &InviteSync{ db: database, invites: invites, logger: logger, lastRun: map[string]time.Time{}, }}
func (s *InviteSync) Reconcile(ctx context.Context, recipientDid string) { if !s.claim(recipientDid) { return }
answer, err := s.fetch(ctx, recipientDid) if err != nil { s.logger.Warn("listing invites, keeping stored rows", "err", err, "recipient", recipientDid) return }
stored, err := deldb.GetNotifications(s.db, recipientDid, 0, orm.FilterIn("type", inviteNotificationTypes)) if err != nil { s.logger.Warn("reading stored invite rows", "err", err, "recipient", recipientDid) return }
offered := keySet(answer.Offers, func(o InviteOffer) string { return o.Uri }) unread := keySet(answer.Pending, func(knot string) string { return knot }) held := keySet(stored, func(n *models.Notification) string { return n.AtUri })
withdrawn := lo.Filter(stored, func(n *models.Notification, _ int) bool { _, standing := offered[n.AtUri] _, cold := unread[n.KnotDid] return !standing && !cold && !answer.Truncated }) fresh := lo.Map( lo.Filter(answer.Offers, func(o InviteOffer, _ int) bool { _, cached := held[o.Uri] return !cached && o.AddedBy != recipientDid }), func(o InviteOffer, _ int) *models.Notification { return inviteNotification(recipientDid, o) }, )
if err := s.settle(recipientDid, withdrawn, fresh); err != nil { s.logger.Warn("writing reconciled invite rows", "err", err, "recipient", recipientDid) }}
func (s *InviteSync) settle(recipientDid string, withdrawn, fresh []*models.Notification) error { if len(withdrawn) == 0 && len(fresh) == 0 { return nil }
tx, err := s.db.Begin() if err != nil { return err } defer func() { _ = tx.Rollback() }()
if err := firstErr(withdrawn, func(n *models.Notification) error { return deldb.DeleteNotification(tx, recipientDid, n.AtUri) }); err != nil { return err } if err := firstErr(fresh, func(n *models.Notification) error { return deldb.CreateNotification(tx, n) }); err != nil { return err }
return tx.Commit()}
func (s *InviteSync) claim(recipientDid string) bool { now := time.Now() s.mu.Lock() defer s.mu.Unlock()
if now.Sub(s.swept) >= inviteReconcileInterval { maps.DeleteFunc(s.lastRun, func(_ string, at time.Time) bool { return now.Sub(at) >= inviteReconcileInterval }) s.swept = now } if at, seen := s.lastRun[recipientDid]; seen && now.Sub(at) < inviteReconcileInterval { return false } s.lastRun[recipientDid] = now return true}
func (s *InviteSync) fetch(ctx context.Context, recipientDid string) (InviteAnswer, error) { ctx, cancel := context.WithTimeout(ctx, inviteFetchTimeout) defer cancel()
return s.invites.ListCollaboratorInvitesBy(ctx, recipientDid)}
func keySet[T any, K comparable](items []T, key func(T) K) map[K]struct{} { return lo.SliceToMap(items, func(item T) (K, struct{}) { return key(item), struct{}{} })}
func firstErr[T any](items []T, act func(T) error) error { return lo.Reduce(items, func(err error, item T, _ int) error { if err != nil { return err } return act(item) }, nil)}
func inviteNotification(recipientDid string, o InviteOffer) *models.Notification { return &models.Notification{ RecipientDid: recipientDid, AtUri: o.Uri, Type: models.NotificationTypeCollaboratorInvited, ActorDid: o.AddedBy, KnotDid: o.KnotDid, RepoDid: o.RepoDid, Created: o.CreatedAt, }}