diff --git a/deliberi/db/ingest.go b/deliberi/db/ingest.go index 41387080d..8d20f8a4b 100644 --- a/deliberi/db/ingest.go +++ b/deliberi/db/ingest.go @@ -89,9 +89,10 @@ func ThreadParticipants(e Execer, parentUri string) ([]string, error) { func TicketParticipants(e Execer, ticketUri string) ([]string, error) { return queryStrings(e, - `select distinct author_did from comments where ticket_uri = ? + `select distinct author_did from comments + where ticket_uri = ? or ticket_uri in (select uri from patches where ticket_uri = ?) union select author_did from entity_titles where at_uri = ? and author_did != ''`, - ticketUri, ticketUri, + ticketUri, ticketUri, ticketUri, ) } diff --git a/deliberi/ingest.go b/deliberi/ingest.go index 47b02c811..8567bb0d8 100644 --- a/deliberi/ingest.go +++ b/deliberi/ingest.go @@ -5,12 +5,10 @@ import ( "encoding/json" "fmt" "log/slog" - "net/http" "slices" "time" "github.com/bluesky-social/indigo/atproto/syntax" - indigoxrpc "github.com/bluesky-social/indigo/xrpc" deldb "tangled.org/core/deliberi/db" models "tangled.org/core/deliberi/models" "tangled.org/core/idresolver" @@ -321,19 +319,16 @@ func (i *Ingester) planComment(ctx context.Context, e *jetstream.Event, uri stri if err != nil || subject.Authority().String() != e.Did { return nil } - var ticketUri string + ticketUri := subject.String() switch subject.Collection().String() { case ticketNSID: - ticketUri = subject.String() case patchNSID: - ticketUri = deldb.GetPatchTicket(i.db, subject.String()) - } - ticket, err := syntax.ParseATURI(ticketUri) - if err != nil { - return nil - } - ent, learned := i.resolveTicket(ctx, ticket, notify) - if ent.Kind == "" { + // 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 } @@ -342,16 +337,21 @@ func (i *Ingester) planComment(ctx context.Context, e *jetstream.Event, uri stri if rec.ReplyTo != nil { comment.ReplyTo = rec.ReplyTo.Uri } - - var owner string - var mentions []string - if notify { - owner = i.repoOwner(ctx, e.Did) - mentions = i.mentionDids(ctx, actor, bodyText(rec.Body)) + 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 := deldb.PutComment(x, comment); err != nil { + if err := store(x); err != nil { return err } if learned { @@ -359,9 +359,6 @@ func (i *Ingester) planComment(ctx context.Context, e *jetstream.Event, uri stri return err } } - if !notify { - return nil - } var participants []string var err error @@ -551,24 +548,6 @@ func (i *Ingester) repoOwner(ctx context.Context, repoDid string) string { return owner } -func (i *Ingester) resolveTicket(ctx context.Context, uri syntax.ATURI, notify bool) (ent deldb.Entity, learned bool) { - ent = deldb.GetEntity(i.db, uri.String()) - if ent.Kind == "" && i.idResolver != nil { - rec, err := i.fetchTicket(ctx, uri) - if err != nil { - i.logger.Warn("hydrating ticket", "err", err, "uri", uri) - return ent, false - } - // no state, the pds is past this event and would fake changes on old updates - ent = deldb.Entity{Title: rec.title(), RepoDid: uri.Authority().String(), Kind: rec.kind()} - learned = true - } - if notify && ent.Kind != "" && i.fillOpener(ctx, uri.String(), &ent) { - learned = true - } - return ent, learned -} - // 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 { @@ -583,40 +562,6 @@ func (i *Ingester) fillOpener(ctx context.Context, ticketUri string, ent *deldb. return author != "" } -// indigo's typed getRecord only decodes registered $types, and org.tangled -// records have no go bindings yet -func (i *Ingester) fetchTicket(ctx context.Context, uri syntax.ATURI) (*ticketRecord, error) { - ident, err := i.idResolver.ResolveIdent(ctx, uri.Authority().String()) - if err != nil { - return nil, fmt.Errorf("resolving %s: %w", uri.Authority(), err) - } - - xc := &indigoxrpc.Client{ - Host: ident.PDSEndpoint(), - Client: &http.Client{Timeout: 10 * time.Second}, - } - var out struct { - Value json.RawMessage `json:"value"` - } - params := map[string]any{ - "repo": ident.DID.String(), - "collection": ticketNSID, - "rkey": uri.RecordKey().String(), - } - if err := xc.Do(ctx, indigoxrpc.Query, "", "com.atproto.repo.getRecord", params, nil, &out); err != nil { - return nil, fmt.Errorf("getting record: %w", err) - } - if len(out.Value) == 0 { - return nil, fmt.Errorf("record has no value") - } - - var rec ticketRecord - if err := json.Unmarshal(out.Value, &rec); err != nil { - return nil, fmt.Errorf("decoding ticket: %w", err) - } - return &rec, nil -} - func (i *Ingester) deliver(x deldb.Execer, recipientDid string, n models.Notification) error { if recipientDid == "" || recipientDid == n.ActorDid { return nil diff --git a/deliberi/ingest_test.go b/deliberi/ingest_test.go index bcda0d051..d79793d1c 100644 --- a/deliberi/ingest_test.go +++ b/deliberi/ingest_test.go @@ -509,6 +509,39 @@ func TestCommentOnPatchResolvesToTheTicket(t *testing.T) { } } +// hydrant backfills a repo in key order, so feed.comment, then review.patch, then track.ticket +func TestBackfillKeepsCommentsSentBeforeTheirTicket(t *testing.T) { + i := newTestIngester(t, fakeBobbin{}) + patchUri := "at://" + repoDid + "/org.tangled.review.patch/bafypatch" + mustProcess(t, i, + backfill(commentFrame("c1", "r1", ticketUri, "", otherDid)), + backfill(commentFrame("c2", "r2", patchUri, "", "did:plc:sub")), + backfill(frame(repoDid, jetstream.CommitOperationCreate, patchNSID, "bafypatch", "r3", `{"ticket":"1","base":"a","head":"b","createdAt":"2026-01-01T00:00:00Z"}`)), + backfill(ticketFrame(jetstream.CommitOperationCreate, "r4", "open", editorDid, pullExtra)), + ticketFrame(jetstream.CommitOperationUpdate, "r5", "closed", ownerDid, pullExtra), + ) + + for _, did := range []string{otherDid, "did:plc:sub"} { + if n := onlyFor(t, i, did); n.Type != models.NotificationTypePullClosed { + t.Errorf("%s got %q, want the close", did, n.Type) + } + } +} + +// like a comment on a deleted ticket. nothing to name in a notice, but the commenter still counts +func TestLiveCommentOnAnUnseenTicketOnlyKeepsTheCommenter(t *testing.T) { + i := newTestIngester(t, fakeBobbin{}) + mustProcess(t, i, + commentFrame("c1", "r1", ticketUri, "", otherDid), + // the first version deliberi sees is only a baseline + ticketFrame(jetstream.CommitOperationUpdate, "r2", "closed", ownerDid, ""), + ticketFrame(jetstream.CommitOperationUpdate, "r3", "open", ownerDid, ""), + ) + if n := onlyFor(t, i, otherDid); n.Type != models.NotificationTypeIssueReopen { + t.Errorf("commenter got %q, want only the reopen", n.Type) + } +} + func TestCommentOnAnotherReposTicketIgnored(t *testing.T) { i := newTestIngester(t, fakeBobbin{}) const foreign = "at://did:plc:elsewhere/org.tangled.track.ticket/1"