package deliberi
import (
"context"
"encoding/json"
"fmt"
"log/slog"
"slices"
"time"
"github.com/bluesky-social/indigo/atproto/syntax"
"github.com/posthog/posthog-go"
"tangled.org/core/deliberi/analytics"
deldb "tangled.org/core/deliberi/db"
"tangled.org/core/deliberi/kv"
models "tangled.org/core/deliberi/models"
"tangled.org/core/idresolver"
"tangled.org/core/jetstream"
"tangled.org/core/sets"
)
const (
maxMentions = 8
idleCursorEvery = time.Minute
// rewinding replays events, the rev check in plan drops the repeats
cursorRewind = 5 * time.Minute
)
type Ingester struct {
db *deldb.DB
bobbin bobbinLookup
idResolver *idresolver.Resolver
endpoint string
logger *slog.Logger
// nil when analytics is off
analytics *analytics.Client
// nil when the KV cache is off
emailKV kv.EmailKV
idleCursorUs int64
}
var ingestCollections = []string{
repoManifestNSID,
ticketNSID,
commentNSID,
patchNSID,
collaboratorNSID,
subscriptionNSID,
starNSID,
followNSID,
}
// star and follow stay out, their handlers only notify on live creates
var revGuarded = []string{
repoManifestNSID,
ticketNSID,
commentNSID,
patchNSID,
collaboratorNSID,
subscriptionNSID,
}
func NewIngester(database *deldb.DB, bobbin bobbinLookup, idRes *idresolver.Resolver, endpoint string, logger *slog.Logger) *Ingester {
return &Ingester{
db: database,
bobbin: bobbin,
idResolver: idRes,
endpoint: endpoint,
logger: logger,
}
}
func (i *Ingester) Run(ctx context.Context) error {
stream, err := jetstream.NewStream(jetstream.StreamConfig{
Endpoint: i.endpoint,
Collections: ingestCollections,
// hydrant only sends backfilled commits when historical is asked for
EventTypes: []string{"live", "historical"},
Cursor: func() (int64, bool, error) { return deldb.GetCursor(i.db) },
Rewind: cursorRewind,
UserAgent: "deliberi",
}, i.logger)
if err != nil {
return err
}
return stream.Run(ctx, i.process)
}
// runs in the event tx, so network calls belong in plan
type writer func(x deldb.Execer) error
func (i *Ingester) process(ctx context.Context, e *jetstream.Event) error {
if e.Kind == jetstream.EventKindAccount && e.Account != nil {
return i.processAccount(e)
}
var write writer
if e.Kind == jetstream.EventKindCommit && e.Commit != nil {
write = i.plan(ctx, e)
}
if write == nil {
return i.advance(e.TimeUS)
}
tx, err := i.db.Begin()
if err != nil {
return fmt.Errorf("starting event tx: %w", err)
}
defer tx.Rollback()
if err := write(tx); err != nil {
return err
}
if err := deldb.SaveCursor(tx, e.TimeUS); err != nil {
return fmt.Errorf("saving cursor: %w", err)
}
return tx.Commit()
}
// only "deleted" is final, deactivated and takendown accounts can come back
func (i *Ingester) processAccount(e *jetstream.Event) error {
if e.Account.Active || e.Account.Status == nil || *e.Account.Status != "deleted" {
return i.advance(e.TimeUS)
}
tx, err := i.db.Begin()
if err != nil {
return fmt.Errorf("starting account tx: %w", err)
}
defer tx.Rollback()
emails, err := deldb.DeleteAccountEmails(tx, e.Did)
if err != nil {
return fmt.Errorf("deleting emails for %s: %w", e.Did, err)
}
if err := deldb.SaveCursor(tx, e.TimeUS); err != nil {
return fmt.Errorf("saving cursor: %w", err)
}
if err := tx.Commit(); err != nil {
return err
}
if len(emails) > 0 {
i.logger.Info("removed emails of deleted account", "did", e.Did, "count", len(emails))
}
if i.emailKV != nil {
for _, email := range emails {
i.emailKV.DeleteEmailKey(email)
}
i.emailKV.DeletePrimaryEmail(e.Did)
}
return nil
}
func (i *Ingester) advance(timeUs int64) error {
if timeUs-i.idleCursorUs < idleCursorEvery.Microseconds() {
return nil
}
i.idleCursorUs = timeUs
return deldb.SaveCursor(i.db, timeUs)
}
func (i *Ingester) plan(ctx context.Context, e *jetstream.Event) writer {
c := e.Commit
uri := fmt.Sprintf("at://%s/%s/%s", e.Did, c.Collection, c.RKey)
guarded := c.Rev != "" && slices.Contains(revGuarded, c.Collection)
if guarded && !deldb.RevIsNew(i.db, uri, c.Rev) {
return nil
}
notify := i.notifying(e)
var write writer
switch c.Collection {
case repoManifestNSID:
write = i.planManifest(e, uri)
case ticketNSID:
write = i.planTicket(ctx, e, uri, notify)
case commentNSID:
write = i.planComment(ctx, e, uri, notify)
case patchNSID:
write = i.planPatch(e, uri)
case collaboratorNSID:
write = i.planCollaborator(e)
case subscriptionNSID:
write = i.planSubscription(e, uri)
case starNSID:
write = i.planStar(ctx, e, uri, notify)
case followNSID:
write = i.planFollow(e, uri, notify)
}
if write == nil || !guarded {
return write
}
return func(x deldb.Execer) error {
if err := deldb.PutRev(x, uri, c.Rev); err != nil {
return err
}
return write(x)
}
}
// bootstrap time is the cutoff, anything older only builds state
func (i *Ingester) notifying(e *jetstream.Event) bool {
return e.Commit.IsLive() && e.TimeUS >= deldb.GetBootstrapUs(i.db)
}
func (i *Ingester) planManifest(e *jetstream.Event, uri string) writer {
if e.Commit.Operation == jetstream.CommitOperationDelete || len(e.Commit.Record) == 0 {
return nil
}
var rec repoManifest
if err := json.Unmarshal(e.Commit.Record, &rec); err != nil {
i.logger.Warn("decoding manifest record", "err", err, "uri", uri)
return nil
}
if rec.Declaration == nil || rec.Declaration.Owner == "" {
return nil
}
return func(x deldb.Execer) error {
return deldb.PutRepoName(x, e.Did, rec.Declaration.Owner, rec.Declaration.Name)
}
}
func (i *Ingester) planTicket(ctx context.Context, e *jetstream.Event, uri string, notify bool) writer {
c := e.Commit
if c.Operation == jetstream.CommitOperationDelete {
return func(x deldb.Execer) error {
if err := deldb.DeleteEntity(x, uri); err != nil {
return err
}
return deldb.DeleteNotificationsAbout(x, uri)
}
}
if len(c.Record) == 0 {
return nil
}
var rec ticketRecord
if err := json.Unmarshal(c.Record, &rec); err != nil {
i.logger.Warn("decoding ticket record", "err", err, "uri", uri)
return nil
}
actor := editorOr(rec.Editor, e.Did)
known := deldb.GetEntity(i.db, uri)
next := deldb.Entity{
Title: rec.title(),
RepoDid: e.Did,
Kind: rec.kind(),
State: rec.state(),
Cid: c.CID,
Author: known.Author,
}
// a backfill sends the latest version as a create, and a maintainer may have
// written that one, so only a live create names who opened the ticket
opened := c.IsLive() && c.Operation == jetstream.CommitOperationCreate && known.Kind == ""
if opened {
next.Author = actor
}
store := func(x deldb.Execer) error { return deldb.PutEntity(x, uri, next) }
if !notify {
return store
}
i.fillOpener(ctx, uri, &next)
var kind models.NotificationType
switch {
case opened:
kind = models.NotificationTypeIssueCreated
if next.Kind == models.EntityKindPull {
kind = models.NotificationTypePullCreated
}
case known.State != "":
kind = stateNotification(next.Kind, known.State, next.State)
}
// todo: assigned and unassigned, once the knot takes assignees
if kind == "" {
return store
}
owner := i.repoOwner(ctx, e.Did)
var mentions []string
if opened {
mentions = i.mentionDids(ctx, actor, bodyText(rec.Body))
}
return func(x deldb.Execer) error {
if err := store(x); err != nil {
return err
}
collaborators, err := deldb.ListCollaborators(x, e.Did)
if err != nil {
return err
}
recipients := append([]string{owner}, collaborators...)
n := models.Notification{
AtUri: uri,
Type: kind,
ActorDid: actor,
RepoDid: e.Did,
EntityAt: uri,
EntityKind: next.Kind,
EntityTitle: next.Title,
Created: time.UnixMicro(e.TimeUS).UTC(),
}
if !opened {
// a ticket can close, reopen and close again, so each change gets its own uri
n.AtUri = uri + "#" + c.Rev
participants, err := deldb.TicketParticipants(x, uri)
if err != nil {
return err
}
recipients = append(recipients, participants...)
}
if err := i.fanOut(x, n, recipients, mentions); err != nil {
return err
}
if event := stateEvent(kind); event != "" {
// the ticket's author is the subject, as the appview's events had it
i.analytics.Capture(event, editorOr(next.Author, actor), uri+"#"+c.Rev, posthog.NewProperties().
Set("repo_did", e.Did).
Set("uri", uri).
Set("actor", actor))
}
return nil
}
}
// the appview's names, so existing dashboards keep counting
func stateEvent(kind models.NotificationType) string {
switch kind {
case models.NotificationTypeIssueClosed:
return "issue_closed"
case models.NotificationTypeIssueReopen:
return "issue_reopen"
case models.NotificationTypePullClosed:
return "pull_closed"
case models.NotificationTypePullReopen:
return "pull_reopen"
case models.NotificationTypePullMerged:
return "pull_merged"
}
return ""
}
func stateNotification(kind models.EntityKind, from, to string) models.NotificationType {
if from == to {
return ""
}
pull := kind == models.EntityKindPull
pick := func(issueType, pullType models.NotificationType) models.NotificationType {
if pull {
return pullType
}
return issueType
}
switch to {
case ticketStateClosed:
return pick(models.NotificationTypeIssueClosed, models.NotificationTypePullClosed)
case ticketStateCompleted:
// completed on a pull means it merged
return pick(models.NotificationTypeIssueClosed, models.NotificationTypePullMerged)
case ticketStateOpen:
return pick(models.NotificationTypeIssueReopen, models.NotificationTypePullReopen)
}
return ""
}
func (i *Ingester) planComment(ctx context.Context, e *jetstream.Event, uri string, notify bool) writer {
c := e.Commit
if c.Operation == jetstream.CommitOperationDelete {
return func(x deldb.Execer) error {
if err := deldb.DeleteComment(x, uri); err != nil {
return err
}
return deldb.DeleteNotificationsFrom(x, uri)
}
}
// an edit can't change the author or reply target, so only creates count
if c.Operation != jetstream.CommitOperationCreate || len(c.Record) == 0 {
return nil
}
var rec commentRecord
if err := json.Unmarshal(c.Record, &rec); err != nil {
i.logger.Warn("decoding comment record", "err", err, "uri", uri)
return nil
}
if rec.Subject == nil {
return nil
}
subject, err := syntax.ParseATURI(rec.Subject.Uri)
if err != nil || subject.Authority().String() != e.Did {
return nil
}
ticketUri := subject.String()
switch subject.Collection().String() {
case ticketNSID:
case patchNSID:
// a backfill sends a repo's comments before its patches, so an unknown patch
// stays the key and TicketParticipants finds it once the patch arrives
if t := deldb.GetPatchTicket(i.db, ticketUri); t != "" {
ticketUri = t
}
default:
return nil
}
actor := editorOr(rec.Editor, e.Did)
comment := deldb.Comment{Uri: uri, TicketUri: ticketUri, Author: actor}
if rec.ReplyTo != nil {
comment.ReplyTo = rec.ReplyTo.Uri
}
store := func(x deldb.Execer) error { return deldb.PutComment(x, comment) }
if !notify {
return store
}
// no ticket yet, or a deleted one, so there's nothing to name in a notice
ent := deldb.GetEntity(i.db, ticketUri)
if ent.Kind == "" {
return store
}
learned := i.fillOpener(ctx, ticketUri, &ent)
owner := i.repoOwner(ctx, e.Did)
mentions := i.mentionDids(ctx, actor, bodyText(rec.Body))
return func(x deldb.Execer) error {
if err := store(x); err != nil {
return err
}
if learned {
if err := deldb.PutEntity(x, ticketUri, ent); err != nil {
return err
}
}
var participants []string
var err error
switch {
case ent.Kind == models.EntityKindPull:
participants, err = deldb.TicketParticipants(x, ticketUri)
case comment.ReplyTo != "":
participants, err = deldb.ThreadParticipants(x, comment.ReplyTo)
default:
participants = []string{ent.Author}
}
if err != nil {
return err
}
n := models.Notification{
AtUri: uri,
Type: models.NotificationTypeIssueCommented,
ActorDid: actor,
RepoDid: e.Did,
EntityAt: ticketUri,
EntityKind: ent.Kind,
EntityTitle: ent.Title,
Created: time.UnixMicro(e.TimeUS).UTC(),
}
if ent.Kind == models.EntityKindPull {
n.Type = models.NotificationTypePullCommented
}
return i.fanOut(x, n, append([]string{owner}, participants...), mentions)
}
}
func (i *Ingester) planPatch(e *jetstream.Event, uri string) writer {
if e.Commit.Operation == jetstream.CommitOperationDelete {
return func(x deldb.Execer) error { return deldb.DeletePatch(x, uri) }
}
var rec patchRecord
if err := json.Unmarshal(e.Commit.Record, &rec); err != nil || rec.Ticket == "" {
i.logger.Warn("decoding patch record", "err", err, "uri", uri)
return nil
}
ticketUri := fmt.Sprintf("at://%s/%s/%s", e.Did, ticketNSID, rec.Ticket)
return func(x deldb.Execer) error { return deldb.PutPatch(x, uri, ticketUri) }
}
func (i *Ingester) planCollaborator(e *jetstream.Event) writer {
did, err := syntax.ParseDID(e.Commit.RKey)
if err != nil {
return nil
}
if e.Commit.Operation == jetstream.CommitOperationDelete {
return func(x deldb.Execer) error { return deldb.DeleteCollaborator(x, e.Did, did.String()) }
}
return func(x deldb.Execer) error { return deldb.PutCollaborator(x, e.Did, did.String()) }
}
func (i *Ingester) planSubscription(e *jetstream.Event, uri string) writer {
c := e.Commit
if c.Operation == jetstream.CommitOperationDelete {
return func(x deldb.Execer) error { return deldb.DeleteSubscription(x, e.Did, c.RKey) }
}
var rec subscriptionRecord
if err := json.Unmarshal(c.Record, &rec); err != nil || rec.subject() == "" {
i.logger.Warn("decoding subscription record", "err", err, "uri", uri)
return nil
}
return func(x deldb.Execer) error {
return deldb.PutSubscription(x, e.Did, c.RKey, rec.subject(), rec.Ignore)
}
}
// undoing a star or a follow takes its notice back, the same as erasing a comment
func takeBack(uri string) writer {
return func(x deldb.Execer) error { return deldb.DeleteNotificationsFrom(x, uri) }
}
func (i *Ingester) planStar(ctx context.Context, e *jetstream.Event, uri string, notify bool) writer {
if e.Commit.Operation == jetstream.CommitOperationDelete {
return takeBack(uri)
}
if e.Commit.Operation != jetstream.CommitOperationCreate || !notify {
return nil
}
var rec starRecord
if err := json.Unmarshal(e.Commit.Record, &rec); err != nil {
i.logger.Warn("decoding star record", "err", err, "uri", uri)
return nil
}
if rec.Subject == nil || rec.Subject.Type != starRepoType || rec.Subject.Identity == "" {
return nil
}
repoDid := rec.Subject.Identity
ownerDid := i.repoOwner(ctx, repoDid)
if ownerDid == "" {
i.logger.Warn("star: could not resolve repo owner, skipping notification", "repoDid", repoDid)
return nil
}
return func(x deldb.Execer) error {
return i.deliver(x, ownerDid, models.Notification{
AtUri: uri,
Type: models.NotificationTypeRepoStarred,
ActorDid: e.Did,
RepoDid: repoDid,
Created: time.UnixMicro(e.TimeUS).UTC(),
})
}
}
func (i *Ingester) planFollow(e *jetstream.Event, uri string, notify bool) writer {
if e.Commit.Operation == jetstream.CommitOperationDelete {
return takeBack(uri)
}
if e.Commit.Operation != jetstream.CommitOperationCreate || !notify {
return nil
}
subject, err := syntax.ParseDID(e.Commit.RKey)
if err != nil {
return nil
}
return func(x deldb.Execer) error {
return i.deliver(x, subject.String(), models.Notification{
AtUri: uri,
Type: models.NotificationTypeFollowed,
ActorDid: e.Did,
Created: time.UnixMicro(e.TimeUS).UTC(),
})
}
}
func (i *Ingester) fanOut(x deldb.Execer, n models.Notification, implicit, mentions []string) error {
// over the cap is spam, so drop them all instead of trimming to the cap
if len(mentions) > maxMentions {
mentions = nil
}
recipients := sets.New[string]()
for _, did := range implicit {
if did != "" {
recipients.Insert(did)
}
}
subscribers, ignorers, err := deldb.ListSubscribers(x, n.EntityAt, n.RepoDid)
if err != nil {
return fmt.Errorf("listing subscribers: %w", err)
}
for _, did := range subscribers {
recipients.Insert(did)
}
for _, did := range mentions {
recipients.Remove(did)
}
// a mute beats everything, the owner too
for _, did := range ignorers {
recipients.Remove(did)
}
for did := range recipients.All() {
if err := i.deliver(x, did, n); err != nil {
return err
}
}
// mentions ignore mutes, so a muted thread still pings you by name
mention := n
mention.Type = models.NotificationTypeUserMentioned
for _, did := range mentions {
if err := i.deliver(x, did, mention); err != nil {
return err
}
}
return nil
}
func (i *Ingester) repoOwner(ctx context.Context, repoDid string) string {
if owner := deldb.GetRepoOwner(i.db, repoDid); owner != "" {
return owner
}
if i.bobbin == nil {
return ""
}
owner, name, err := i.bobbin.RepoOwner(ctx, repoDid)
if err != nil {
i.logger.Warn("resolving repo owner", "err", err, "repoDid", repoDid)
return ""
}
if owner == "" {
return ""
}
// a nameless record must not blank out a name we already cached
if name == "" {
name = deldb.GetRepoName(i.db, repoDid)
}
if err := deldb.PutRepoName(i.db, repoDid, owner, name); err != nil {
i.logger.Warn("caching repo owner", "err", err, "repoDid", repoDid)
}
return owner
}
// a backfilled ticket only names its last editor, so bobbin says who opened it. true when it found out
func (i *Ingester) fillOpener(ctx context.Context, ticketUri string, ent *deldb.Entity) bool {
if ent.Author != "" || i.bobbin == nil {
return false
}
author, err := i.bobbin.TicketAuthor(ctx, ticketUri)
if err != nil {
i.logger.Warn("looking up who opened a ticket", "err", err, "uri", ticketUri)
return false
}
ent.Author = author
return author != ""
}
func (i *Ingester) deliver(x deldb.Execer, recipientDid string, n models.Notification) error {
if recipientDid == "" || recipientDid == n.ActorDid {
return nil
}
prefs, err := deldb.GetNotificationPreference(x, recipientDid)
if err != nil {
return fmt.Errorf("loading prefs for %s: %w", recipientDid, err)
}
if !prefs.ShouldNotify(n.Type) {
return nil
}
n.RecipientDid = recipientDid
return deldb.CreateNotification(x, &n)
}