diff --git a/appview/db/comments.go b/appview/db/comments.go new file mode 100644 --- /dev/null +++ b/appview/db/comments.go @@ -0,0 +1,268 @@ +package db + +import ( + "database/sql" + "encoding/json" + "fmt" + "sort" + "strings" + "time" + + "github.com/bluesky-social/indigo/api/atproto" + "github.com/bluesky-social/indigo/atproto/syntax" + "tangled.org/core/api/tangled" + "tangled.org/core/appview/models" + "tangled.org/core/orm" +) + +func PutComment(tx *sql.Tx, c *models.Comment, references []syntax.ATURI) error { + if c.Collection == "" { + c.Collection = tangled.FeedCommentNSID + } + + var bodyBlobs, replyToUri, replyToCid *string + if len(c.Body.Blobs) > 0 { + encoded, err := json.Marshal(c.Body.Blobs) + if err != nil { + return fmt.Errorf("encoding blobs to json: %w", err) + } + encodedStr := string(encoded) + bodyBlobs = &encodedStr + } + if c.ReplyTo != nil { + replyToUri = &c.ReplyTo.Uri + replyToCid = &c.ReplyTo.Cid + } + result, err := tx.Exec( + // users can change the 'created' date. + // skip update entirely if cid is unchanged. + `insert into comments ( + did, + collection, + rkey, + cid, + subject_uri, + subject_cid, + body_text, + body_original, + body_blobs, + created, + reply_to_uri, + reply_to_cid, + pull_round_idx + ) + values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + on conflict(did, collection, rkey) + do update set + cid = excluded.cid, + subject_uri = excluded.subject_uri, + subject_cid = excluded.subject_cid, + body_text = excluded.body_text, + body_original = excluded.body_original, + body_blobs = excluded.body_blobs, + created = excluded.created, + reply_to_uri = excluded.reply_to_uri, + reply_to_cid = excluded.reply_to_cid, + pull_round_idx = excluded.pull_round_idx, + edited = ? + where comments.cid != excluded.cid`, + c.Did, + c.Collection, + c.Rkey, + c.Cid, + c.Subject.Uri, + c.Subject.Cid, + c.Body.Text, + c.Body.Original, + bodyBlobs, + c.Created.Format(time.RFC3339), + replyToUri, + replyToCid, + c.PullRoundIdx, + time.Now().Format(time.RFC3339), + ) + if err != nil { + return err + } + + c.Id, err = result.LastInsertId() + if err != nil { + return err + } + + affected, err := result.RowsAffected() + if err != nil { + return err + } + + if affected > 0 { + // update references when comment is updated + if err := putReferences(tx, c.AtUri(), references); err != nil { + return fmt.Errorf("put reference_links: %w", err) + } + } + + return nil +} + +// PurgeComments actually purges a comment row from db instead of marking it as "deleted" +func PurgeComments(e Execer, filters ...orm.Filter) 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 ") + } + + _, err := e.Exec(fmt.Sprintf(`delete from comments %s`, whereClause), args...) + return err +} + +func DeleteComments(e Execer, filters ...orm.Filter) 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( + `update comments + set body_text = "", + body_original = null, + body_blobs = null, + deleted = strftime('%%Y-%%m-%%dT%%H:%%M:%%SZ', 'now') + %s`, + whereClause, + ) + + _, err := e.Exec(query, args...) + return err +} + +func GetComments(e Execer, filters ...orm.Filter) ([]models.Comment, error) { + var comments []models.Comment + + 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, + did, + collection, + rkey, + cid, + subject_uri, + subject_cid, + body_text, + body_original, + body_blobs, + created, + reply_to_uri, + reply_to_cid, + pull_round_idx, + edited, + deleted + from + comments + %s + `, whereClause) + + rows, err := e.Query(query, args...) + if err != nil { + return nil, err + } + defer rows.Close() + + for rows.Next() { + var comment models.Comment + var created string + var cid, bodyBlobs, replyToUri, replyToCid, edited, deleted sql.Null[string] + err := rows.Scan( + &comment.Id, + &comment.Did, + &comment.Collection, + &comment.Rkey, + &cid, + &comment.Subject.Uri, + &comment.Subject.Cid, + &comment.Body.Text, + &comment.Body.Original, + &bodyBlobs, + &created, + &replyToUri, + &replyToCid, + &comment.PullRoundIdx, + &edited, + &deleted, + ) + if err != nil { + return nil, err + } + + if cid.Valid && cid.V != "" { + comment.Cid = syntax.CID(cid.V) + } + + if bodyBlobs.Valid && bodyBlobs.V != "" { + if err := json.Unmarshal([]byte(bodyBlobs.V), &comment.Body.Blobs); err != nil { + return nil, fmt.Errorf("decoding blobs: %w", err) + } + } + + if t, err := time.Parse(time.RFC3339, created); err == nil { + comment.Created = t + } + + if replyToUri.Valid && replyToCid.Valid { + comment.ReplyTo = &atproto.RepoStrongRef{ + Uri: replyToUri.V, + Cid: replyToCid.V, + } + } + + if edited.Valid { + if t, err := time.Parse(time.RFC3339, edited.V); err == nil { + comment.Edited = &t + } + } + + if deleted.Valid { + if t, err := time.Parse(time.RFC3339, deleted.V); err == nil { + comment.Deleted = &t + } + } + + comments = append(comments, comment) + } + + if err := rows.Err(); err != nil { + return nil, err + } + + sort.Slice(comments, func(i, j int) bool { + return comments[i].Created.Before(comments[j].Created) + }) + + return comments, nil +} diff --git a/appview/db/db.go b/appview/db/db.go --- a/appview/db/db.go +++ b/appview/db/db.go @@ -1409,6 +1409,102 @@ return err }) + orm.RunMigration(conn, logger, "add-comments-table", func(tx *sql.Tx) error { + _, err := tx.Exec(` + drop table if exists comments; + + create table comments ( + -- identifiers + id integer primary key autoincrement, + + did text not null, + collection text not null default 'sh.tangled.feed.comment', + rkey text not null, + at_uri text generated always as ('at://' || did || '/' || collection || '/' || rkey) stored, + cid text, + + -- content + subject_uri text not null, -- at_uri of subject (issue, pr, string) + subject_cid text not null, -- cid of subject + + body_text text not null, + body_original text, + body_blobs text, -- json + + created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), + + reply_to_uri text, -- at_uri of parent comment + reply_to_cid text, -- cid of parent comment + + pull_round_idx integer, -- pull round index. required when subject is sh.tangled.repo.pull + + -- appview-local information + edited text, + deleted text, + + unique(did, collection, rkey) + ); + + insert into comments ( + did, + collection, + rkey, + subject_uri, + subject_cid, -- we need to know cid + body_text, + created, + reply_to_uri, + reply_to_cid, -- we need to know cid + edited, + deleted + ) + select + did, + 'sh.tangled.repo.issue.comment', + rkey, + issue_at, + '', + body, + created, + reply_to, + '', + edited, + deleted + from issue_comments + where rkey is not null; + + insert into comments ( + did, + collection, + rkey, + subject_uri, + subject_cid, -- we need to know cid + body_text, + created, + pull_round_idx + ) + select + c.owner_did, + 'sh.tangled.repo.pull.comment', + substr( + substr(c.comment_at, 6 + instr(substr(c.comment_at, 6), '/')), -- nsid/rkey + instr( + substr(c.comment_at, 6 + instr(substr(c.comment_at, 6), '/')), -- nsid/rkey + '/' + ) + 1 + ), -- rkey + p.at_uri, + '', + c.body, + c.created, + s.round_number + from pull_comments c + join pulls p on c.repo_at = p.repo_at and c.pull_id = p.pull_id + join pull_submissions s on s.id = c.submission_id; + `) + return err + }) + return &DB{ db, logger, diff --git a/appview/db/pulls.go b/appview/db/pulls.go --- a/appview/db/pulls.go +++ b/appview/db/pulls.go @@ -524,7 +524,7 @@ return nil, err } defer rows.Close() - submissionMap := make(map[int]*models.PullSubmission) + pullMap := make(map[syntax.ATURI][]*models.PullSubmission) for rows.Next() { var submission models.PullSubmission @@ -572,129 +572,39 @@ if patchBlobSize.Valid { submission.Blob.Size = patchBlobSize.V } - submissionMap[submission.ID] = &submission + pullMap[submission.PullAt] = append(pullMap[submission.PullAt], &submission) } if err := rows.Err(); err != nil { return nil, err } - // Get comments for all submissions using GetPullComments - submissionIds := slices.Collect(maps.Keys(submissionMap)) - comments, err := GetPullComments(e, orm.FilterIn("submission_id", submissionIds)) + // Get comments for all submissions using GetComments + pullAts := slices.Collect(maps.Keys(pullMap)) + comments, err := GetComments(e, orm.FilterIn("subject_uri", pullAts)) if err != nil { return nil, fmt.Errorf("failed to get pull comments: %w", err) } for _, comment := range comments { - if submission, ok := submissionMap[comment.SubmissionId]; ok { - submission.Comments = append(submission.Comments, comment) + if comment.PullRoundIdx != nil { + roundIdx := *comment.PullRoundIdx + if submissions, ok := pullMap[syntax.ATURI(comment.Subject.Uri)]; ok { + if roundIdx < len(submissions) { + submission := submissions[roundIdx] + submission.Comments = append(submission.Comments, comment) + } + } } } - // group the submissions by pull_at - m := make(map[syntax.ATURI][]*models.PullSubmission) - for _, s := range submissionMap { - m[s.PullAt] = append(m[s.PullAt], s) - } - // sort each one by round number - for _, s := range m { + for _, s := range pullMap { slices.SortFunc(s, func(a, b *models.PullSubmission) int { return cmp.Compare(a.RoundNumber, b.RoundNumber) }) } - return m, nil -} - -func GetPullComments(e Execer, filters ...orm.Filter) ([]models.PullComment, 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 - id, - pull_id, - submission_id, - repo_at, - owner_did, - comment_at, - body, - created - from - pull_comments - %s - order by - created asc - `, whereClause) - - rows, err := e.Query(query, args...) - if err != nil { - return nil, err - } - defer rows.Close() - - commentMap := make(map[string]*models.PullComment) - for rows.Next() { - var comment models.PullComment - var createdAt string - err := rows.Scan( - &comment.ID, - &comment.PullId, - &comment.SubmissionId, - &comment.RepoAt, - &comment.OwnerDid, - &comment.CommentAt, - &comment.Body, - &createdAt, - ) - if err != nil { - return nil, err - } - - if t, err := time.Parse(time.RFC3339, createdAt); err == nil { - comment.Created = t - } - - atUri := comment.AtUri().String() - commentMap[atUri] = &comment - } - - if err := rows.Err(); err != nil { - return nil, err - } - - // collect references for each comments - commentAts := slices.Collect(maps.Keys(commentMap)) - allReferences, err := GetReferencesAll(e, orm.FilterIn("from_at", commentAts)) - if err != nil { - return nil, fmt.Errorf("failed to query reference_links: %w", err) - } - for commentAt, references := range allReferences { - if comment, ok := commentMap[commentAt.String()]; ok { - comment.References = references - } - } - - var comments []models.PullComment - for _, c := range commentMap { - comments = append(comments, *c) - } - - sort.Slice(comments, func(i, j int) bool { - return comments[i].Created.Before(comments[j].Created) - }) - - return comments, nil + return pullMap, nil } // timeframe here is directly passed into the sql query filter, and any @@ -771,33 +681,6 @@ return nil, err } return pulls, nil -} - -func NewPullComment(tx *sql.Tx, comment *models.PullComment) (int64, error) { - query := `insert into pull_comments (owner_did, repo_at, submission_id, comment_at, pull_id, body) values (?, ?, ?, ?, ?, ?)` - res, err := tx.Exec( - query, - comment.OwnerDid, - comment.RepoAt, - comment.SubmissionId, - comment.CommentAt, - comment.PullId, - comment.Body, - ) - if err != nil { - return 0, err - } - - i, err := res.LastInsertId() - if err != nil { - return 0, err - } - - if err := putReferences(tx, comment.AtUri(), comment.References); err != nil { - return 0, fmt.Errorf("put reference_links: %w", err) - } - - return i, nil } // use with transaction diff --git a/appview/db/reference.go b/appview/db/reference.go --- a/appview/db/reference.go +++ b/appview/db/reference.go @@ -124,8 +124,7 @@ `with input(owner_did, name, pull_id, comment_id) as ( values %s ) select - p.owner_did, p.rkey, - c.comment_at + p.owner_did, p.rkey, c.at_uri from input inp join repos r on r.did = inp.owner_did @@ -133,9 +132,9 @@ and r.name = inp.name join pulls p on p.repo_at = r.at_uri and p.pull_id = inp.pull_id - left join pull_comments c + left join comments c on inp.comment_id is not null - and c.repo_at = r.at_uri and c.pull_id = p.pull_id + and c.subject_uri = ('at://' || p.owner_did || '/' || 'sh.tangled.repo.pull' || '/' || p.rkey) and c.id = inp.comment_id `, strings.Join(vals, ","), @@ -293,7 +292,7 @@ if err != nil { return nil, fmt.Errorf("get pull backlinks: %w", err) } backlinks = append(backlinks, ls...) - ls, err = getPullCommentBacklinks(e, target, backlinksMap[tangled.RepoPullCommentNSID]) + ls, err = getPullCommentBacklinks(e, target, backlinksMap[tangled.FeedCommentNSID]) if err != nil { return nil, fmt.Errorf("get pull_comment backlinks: %w", err) } @@ -430,7 +429,7 @@ func getPullCommentBacklinks(e Execer, target syntax.ATURI, aturis []syntax.ATURI) ([]models.RichReferenceLink, error) { if len(aturis) == 0 { return nil, nil } - filter := orm.FilterIn("c.comment_at", aturis) + filter := orm.FilterIn("c.at_uri", aturis) exclude := orm.FilterNotEq("p.at_uri", target) rows, err := e.Query( fmt.Sprintf( @@ -438,8 +437,8 @@ `select r.did, r.name, p.pull_id, c.id, p.title, p.state from repos r join pulls p on r.at_uri = p.repo_at - join pull_comments c - on r.at_uri = c.repo_at and p.pull_id = c.pull_id + join comments c + on ('at://' || p.owner_did || '/' || 'sh.tangled.repo.pull' || '/' || p.rkey) = c.subject_uri where %s and %s`, filter.Condition(), exclude.Condition(), diff --git a/appview/ingester.go b/appview/ingester.go --- a/appview/ingester.go +++ b/appview/ingester.go @@ -25,7 +25,9 @@ "golang.org/x/sync/errgroup" "tangled.org/core/api/tangled" "tangled.org/core/appview/config" "tangled.org/core/appview/db" + "tangled.org/core/appview/mentions" "tangled.org/core/appview/models" + "tangled.org/core/appview/notify" "tangled.org/core/appview/serververify" "tangled.org/core/appview/validator" "tangled.org/core/idresolver" @@ -34,12 +36,14 @@ "tangled.org/core/rbac" ) type Ingester struct { - Db db.DbWrapper - Enforcer *rbac.Enforcer - IdResolver *idresolver.Resolver - Config *config.Config - Logger *slog.Logger - Validator *validator.Validator + Db db.DbWrapper + Enforcer *rbac.Enforcer + IdResolver *idresolver.Resolver + Config *config.Config + Logger *slog.Logger + Validator *validator.Validator + MentionsResolver *mentions.Resolver + Notifier notify.Notifier } type processFunc func(ctx context.Context, e *jmodels.Event) error @@ -87,8 +91,12 @@ case tangled.RepoIssueNSID: err = i.ingestIssue(ctx, e) case tangled.RepoPullNSID: err = i.ingestPull(ctx, e) + case tangled.FeedCommentNSID: + err = i.ingestComment(e) case tangled.RepoIssueCommentNSID: err = i.ingestIssueComment(e) + case tangled.RepoPullCommentNSID: + err = i.ingestPullComment(e) case tangled.LabelDefinitionNSID: err = i.ingestLabelDefinition(e) case tangled.LabelOpNSID: @@ -1159,6 +1167,100 @@ orm.FilterEq("did", did), orm.FilterEq("rkey", rkey), ); err != nil { return fmt.Errorf("failed to delete issue comment record: %w", err) + } + + return nil + } + + return nil +} + +// ingestPullComment ingests legacy sh.tangled.repo.pull.comment deletions +func (i *Ingester) ingestPullComment(e *jmodels.Event) error { + l := i.Logger.With("handler", "ingestPullComment", "nsid", e.Commit.Collection, "did", e.Did, "rkey", e.Commit.RKey) + l.Info("ingesting record") + + switch e.Commit.Operation { + case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: + // no-op. sh.tangled.repo.pull.comment is deprecated + + case jmodels.CommitOperationDelete: + if err := db.PurgeComments( + i.Db, + orm.FilterEq("did", e.Did), + orm.FilterEq("collection", e.Commit.Collection), + orm.FilterEq("rkey", e.Commit.RKey), + ); err != nil { + return fmt.Errorf("failed to delete comment record: %w", err) + } + } + + return nil +} + +func (i *Ingester) ingestComment(e *jmodels.Event) error { + did := e.Did + rkey := e.Commit.RKey + cid := e.Commit.CID + + var err error + + l := i.Logger.With("handler", "ingestComment", "nsid", e.Commit.Collection, "did", did, "rkey", rkey) + l.Info("ingesting record") + + ddb, ok := i.Db.Execer.(*db.DB) + if !ok { + return fmt.Errorf("failed to index issue comment record, invalid db cast") + } + + ctx := context.Background() + + switch e.Commit.Operation { + case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: + raw := json.RawMessage(e.Commit.Record) + record := tangled.FeedComment{} + err = json.Unmarshal(raw, &record) + if err != nil { + return fmt.Errorf("invalid record: %w", err) + } + + comment, err := models.CommentFromRecord(syntax.DID(did), syntax.RecordKey(rkey), syntax.CID(cid), record) + if err != nil { + return fmt.Errorf("failed to parse comment from record: %w", err) + } + + if err := comment.Validate(); err != nil { + return fmt.Errorf("failed to validate comment: %w", err) + } + + var references []syntax.ATURI + if comment.Body.Original != nil { + _, references = i.MentionsResolver.Resolve(ctx, *comment.Body.Original) + } + + tx, err := ddb.Begin() + if err != nil { + return fmt.Errorf("failed to start transaction: %w", err) + } + defer tx.Rollback() + + err = db.PutComment(tx, comment, references) + if err != nil { + return fmt.Errorf("failed to create comment: %w", err) + } + + if err := tx.Commit(); err != nil { + return err + } + + case jmodels.CommitOperationDelete: + if err := db.DeleteComments( + ddb, + orm.FilterEq("did", did), + orm.FilterEq("collection", e.Commit.Collection), + orm.FilterEq("rkey", rkey), + ); err != nil { + return fmt.Errorf("failed to delete comment record: %w", err) } return nil diff --git a/appview/models/comment.go b/appview/models/comment.go new file mode 100644 --- /dev/null +++ b/appview/models/comment.go @@ -0,0 +1,147 @@ +package models + +import ( + "fmt" + "strings" + "time" + + comatproto "github.com/bluesky-social/indigo/api/atproto" + "github.com/bluesky-social/indigo/atproto/syntax" + typegen "github.com/whyrusleeping/cbor-gen" + "tangled.org/core/api/tangled" +) + +type Comment struct { + Id int64 + + Did syntax.DID + Collection syntax.NSID + Rkey syntax.RecordKey + Cid syntax.CID + + // record content + Subject comatproto.RepoStrongRef + Body tangled.MarkupMarkdown // markup body type. only markdown is supported right now + Created time.Time + ReplyTo *comatproto.RepoStrongRef // (optional) parent comment + PullRoundIdx *int // (optional) pull round number used when subject is sh.tangled.repo.pull + + // store on db, but not on PDS + Edited *time.Time + Deleted *time.Time +} + +func (c *Comment) AtUri() syntax.ATURI { + return syntax.ATURI(fmt.Sprintf("at://%s/%s/%s", c.Did, c.Collection, c.Rkey)) +} + +func (c *Comment) StrongRef() comatproto.RepoStrongRef { + return comatproto.RepoStrongRef{ + Uri: c.AtUri().String(), + Cid: c.Cid.String(), + } +} + +func (c *Comment) AsRecord() typegen.CBORMarshaler { + // can't convert to record for legacy types + if c.Collection != tangled.FeedCommentNSID { + return nil + } + var pullRoundIdx int64 + if c.PullRoundIdx != nil { + pullRoundIdx = int64(*c.PullRoundIdx) + } + return &tangled.FeedComment{ + Subject: &c.Subject, + Body: &tangled.FeedComment_Body{MarkupMarkdown: &c.Body}, + CreatedAt: c.Created.Format(time.RFC3339), + ReplyTo: c.ReplyTo, + PullRoundIdx: &pullRoundIdx, + } +} + +func (c *Comment) IsTopLevel() bool { + return c.ReplyTo == nil +} + +func (c *Comment) IsReply() bool { + return c.ReplyTo != nil +} + +func (c *Comment) Validate() error { + // TODO: sanitize the body and then trim space + if sb := strings.TrimSpace(c.Body.Text); sb == "" { + return fmt.Errorf("body is empty after HTML sanitization") + } + + // if it's for PR, PullSubmissionId should not be nil + subjectAt, err := syntax.ParseATURI(c.Subject.Uri) + if err != nil { + return fmt.Errorf("subject.uri is not valid at-uri: %w", err) + } + if subjectAt.Collection().String() == tangled.RepoPullNSID { + if c.PullRoundIdx == nil { + return fmt.Errorf("pullSubmissionId should not be nil when subject is sh.tangled.repo.pull") + } + } + return nil +} + +func CommentFromRecord(did syntax.DID, rkey syntax.RecordKey, cid syntax.CID, record tangled.FeedComment) (*Comment, error) { + created, err := time.Parse(time.RFC3339, record.CreatedAt) + if err != nil { + created = time.Now() + } + + if record.Subject == nil { + return nil, fmt.Errorf("subject can't be nil") + } + subjectAt, err := syntax.ParseATURI(record.Subject.Uri) + if err != nil { + return nil, fmt.Errorf("invalid subject uri: %w", err) + } + if _, err = syntax.ParseCID(record.Subject.Cid); err != nil { + return nil, fmt.Errorf("invalid subject cid: %w", err) + } + + if subjectAt.Collection() == tangled.RepoPullNSID { + if record.PullRoundIdx == nil { + return nil, fmt.Errorf("pullRoundIdx can't be nil when subject is sh.tangled.repo.pull") + } + } + + if record.Body == nil { + return nil, fmt.Errorf("body can't be nil") + } + if record.Body.MarkupMarkdown == nil { + return nil, fmt.Errorf("body should be markdown type") + } + + if record.ReplyTo != nil { + if _, err = syntax.ParseATURI(record.ReplyTo.Uri); err != nil { + return nil, fmt.Errorf("invalid replyTo uri: %w", err) + } + if _, err = syntax.ParseCID(record.ReplyTo.Cid); err != nil { + return nil, fmt.Errorf("invalid replyTo cid: %w", err) + } + } + + var pullRoundIdx *int + if record.PullRoundIdx != nil { + pullRoundIdx = new(int) + *pullRoundIdx = int(*record.PullRoundIdx) + } + + return &Comment{ + Did: did, + Collection: tangled.FeedCommentNSID, + Rkey: rkey, + Cid: cid, + + Subject: *record.Subject, + Body: *record.Body.MarkupMarkdown, + Created: created, + ReplyTo: record.ReplyTo, + PullRoundIdx: pullRoundIdx, + }, nil +} diff --git a/appview/models/pull.go b/appview/models/pull.go --- a/appview/models/pull.go +++ b/appview/models/pull.go @@ -299,37 +299,11 @@ RoundNumber int Blob lexutil.LexBlob Patch string Combined string - Comments []PullComment + Comments []Comment SourceRev string // include the rev that was used to create this submission: only for branch/fork PRs // meta Created time.Time -} - -type PullComment struct { - // ids - ID int - PullId int - SubmissionId int - - // at ids - RepoAt string - OwnerDid string - CommentAt string - - // content - Body string - - // meta - Mentions []syntax.DID - References []syntax.ATURI - - // meta - Created time.Time -} - -func (p *PullComment) AtUri() syntax.ATURI { - return syntax.ATURI(p.CommentAt) } func (p *Pull) TotalComments() int { @@ -451,7 +425,7 @@ addParticipant(s.PullAt.Authority().String()) for _, c := range s.Comments { - addParticipant(c.OwnerDid) + addParticipant(c.Did.String()) } return participants diff --git a/appview/notify/db/db.go b/appview/notify/db/db.go --- a/appview/notify/db/db.go +++ b/appview/notify/db/db.go @@ -281,19 +281,25 @@ pullId, ) } -func (n *databaseNotifier) NewPullComment(ctx context.Context, comment *models.PullComment, mentions []syntax.DID) { +func (n *databaseNotifier) NewPullComment(ctx context.Context, comment *models.Comment, mentions []syntax.DID) { l := log.FromContext(ctx) - pull, err := db.GetPull(n.db, - orm.FilterEq("repo_at", syntax.ATURI(comment.RepoAt)), - orm.FilterEq("pull_id", comment.PullId), + subjectAt := syntax.ATURI(comment.Subject.Uri) + pulls, err := db.GetPulls(n.db, + orm.FilterEq("owner_did", subjectAt.Authority()), + orm.FilterEq("rkey", subjectAt.RecordKey()), ) if err != nil { - l.Error("failed to get pulls", "err", err) + l.Error("failed to get pull", "err", err) return } + if len(pulls) == 0 { + l.Error("NewPullComment: no pull found", "aturi", comment.Subject) + return + } + pull := pulls[0] - repo, err := db.GetRepo(n.db, orm.FilterEq("at_uri", comment.RepoAt)) + repo, err := db.GetRepo(n.db, orm.FilterEq("at_uri", pull.RepoAt)) if err != nil { l.Error("failed to get repos", "err", err) return @@ -311,7 +317,7 @@ for _, m := range mentions { recipients.Remove(m) } - actorDid := syntax.DID(comment.OwnerDid) + actorDid := comment.Did eventType := models.NotificationTypePullCommented entityType := "pull" entityId := pull.AtUri().String() diff --git a/appview/notify/logging/notifier.go b/appview/notify/logging/notifier.go --- a/appview/notify/logging/notifier.go +++ b/appview/notify/logging/notifier.go @@ -86,7 +86,7 @@ ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "NewPull")) l.inner.NewPull(ctx, pull) } -func (l *loggingNotifier) NewPullComment(ctx context.Context, comment *models.PullComment, mentions []syntax.DID) { +func (l *loggingNotifier) NewPullComment(ctx context.Context, comment *models.Comment, mentions []syntax.DID) { ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "NewPullComment")) l.inner.NewPullComment(ctx, comment, mentions) } diff --git a/appview/notify/merged_notifier.go b/appview/notify/merged_notifier.go --- a/appview/notify/merged_notifier.go +++ b/appview/notify/merged_notifier.go @@ -82,7 +82,7 @@ func (m *mergedNotifier) NewPull(ctx context.Context, pull *models.Pull) { m.fanout(func(n Notifier) { n.NewPull(ctx, pull) }) } -func (m *mergedNotifier) NewPullComment(ctx context.Context, comment *models.PullComment, mentions []syntax.DID) { +func (m *mergedNotifier) NewPullComment(ctx context.Context, comment *models.Comment, mentions []syntax.DID) { m.fanout(func(n Notifier) { n.NewPullComment(ctx, comment, mentions) }) } diff --git a/appview/notify/notifier.go b/appview/notify/notifier.go --- a/appview/notify/notifier.go +++ b/appview/notify/notifier.go @@ -23,7 +23,7 @@ NewFollow(ctx context.Context, follow *models.Follow) DeleteFollow(ctx context.Context, follow *models.Follow) NewPull(ctx context.Context, pull *models.Pull) - NewPullComment(ctx context.Context, comment *models.PullComment, mentions []syntax.DID) + NewPullComment(ctx context.Context, comment *models.Comment, mentions []syntax.DID) NewPullState(ctx context.Context, actor syntax.DID, pull *models.Pull) NewIssueLabelOp(ctx context.Context, issue *models.Issue) @@ -64,7 +64,7 @@ func (m *BaseNotifier) NewFollow(ctx context.Context, follow *models.Follow) {} func (m *BaseNotifier) DeleteFollow(ctx context.Context, follow *models.Follow) {} func (m *BaseNotifier) NewPull(ctx context.Context, pull *models.Pull) {} -func (m *BaseNotifier) NewPullComment(ctx context.Context, models *models.PullComment, mentions []syntax.DID) { +func (m *BaseNotifier) NewPullComment(ctx context.Context, models *models.Comment, mentions []syntax.DID) { } func (m *BaseNotifier) NewPullState(ctx context.Context, actor syntax.DID, pull *models.Pull) {} diff --git a/appview/notify/posthog/notifier.go b/appview/notify/posthog/notifier.go --- a/appview/notify/posthog/notifier.go +++ b/appview/notify/posthog/notifier.go @@ -86,13 +86,12 @@ log.Println("failed to enqueue posthog event:", err) } } -func (n *posthogNotifier) NewPullComment(ctx context.Context, comment *models.PullComment, mentions []syntax.DID) { +func (n *posthogNotifier) NewPullComment(ctx context.Context, comment *models.Comment, mentions []syntax.DID) { err := n.client.Enqueue(posthog.Capture{ - DistinctId: comment.OwnerDid, + DistinctId: comment.Did.String(), Event: "new_pull_comment", Properties: posthog.Properties{ - "repo_at": comment.RepoAt, - "pull_id": comment.PullId, + "pull_at": comment.Subject, "mentions": mentions, }, }) diff --git a/appview/oauth/scopes.go b/appview/oauth/scopes.go --- a/appview/oauth/scopes.go +++ b/appview/oauth/scopes.go @@ -16,6 +16,7 @@ "repo:sh.tangled.knot.member", "repo:sh.tangled.spindle", "repo:sh.tangled.spindle.member", "repo:sh.tangled.graph.follow", + "repo:sh.tangled.feed.comment", "repo:sh.tangled.feed.star", "repo:sh.tangled.feed.reaction", "repo:sh.tangled.label.definition", diff --git a/appview/pages/templates/repo/pulls/pull.html b/appview/pages/templates/repo/pulls/pull.html --- a/appview/pages/templates/repo/pulls/pull.html +++ b/appview/pages/templates/repo/pulls/pull.html @@ -625,25 +625,25 @@ {{ end }} {{ define "submissionComment" }} -
+
- {{ template "user/fragments/picLink" (list .OwnerDid "size-8") }} + {{ template "user/fragments/picLink" (list .Did.String "size-8") }}
- {{ $handle := resolve .OwnerDid }} + {{ $handle := resolve .Did.String }} {{ $handle }} - + {{ template "repo/fragments/shortTime" .Created }}
- {{ .Body | markdown }} + {{ .Body.Text | markdown }}
diff --git a/appview/pulls/pulls.go b/appview/pulls/pulls.go --- a/appview/pulls/pulls.go +++ b/appview/pulls/pulls.go @@ -838,60 +838,103 @@ return case http.MethodPost: body := r.FormValue("body") if body == "" { - s.pages.Notice(w, "pull", "Comment body is required") + s.pages.Notice(w, "pull-comment", "Comment body is required") return } + // TODO(boltless): normalize markdown body + normalizedBody := body mentions, references := s.mentionsResolver.Resolve(r.Context(), body) - // Start a transaction - tx, err := s.db.BeginTx(r.Context(), nil) + markdownBody := tangled.MarkupMarkdown{ + Text: normalizedBody, + Original: &body, + Blobs: nil, + } + + // ingest CID of PR record on-demand. + // TODO(boltless): appview should ingest CID of atproto records + cid, err := func() (syntax.CID, error) { + ident, err := s.idResolver.ResolveIdent(r.Context(), pull.OwnerDid) + if err != nil { + return "", err + } + + xrpcc := indigoxrpc.Client{Host: ident.PDSEndpoint()} + out, err := comatproto.RepoGetRecord(r.Context(), &xrpcc, "", tangled.RepoPullNSID, pull.OwnerDid, pull.Rkey) + if err != nil { + return "", err + } + if out.Cid == nil { + return "", fmt.Errorf("record CID is empty") + } + + cid, err := syntax.ParseCID(*out.Cid) + if err != nil { + return "", err + } + + return cid, nil + }() if err != nil { - l.Error("failed to start transaction", "err", err) + s.logger.Error("failed to backfill subject PR record", "err", err) + s.pages.Notice(w, "pull-comment", "failed to backfill subject record") + return + } + pullStrongRef := comatproto.RepoStrongRef{ + Uri: pull.AtUri().String(), + Cid: cid.String(), + } + + comment := models.Comment{ + Did: syntax.DID(user.Did), + Collection: tangled.FeedCommentNSID, + Rkey: syntax.RecordKey(tid.TID()), + + Subject: pullStrongRef, + Body: markdownBody, + Created: time.Now(), + ReplyTo: nil, + PullRoundIdx: &roundNumber, + } + if err = comment.Validate(); err != nil { + s.logger.Error("failed to validate comment", "err", err) s.pages.Notice(w, "pull-comment", "Failed to create comment.") return } - defer tx.Rollback() - - createdAt := time.Now().Format(time.RFC3339) client, err := s.oauth.AuthorizedClient(r) if err != nil { - l.Error("failed to get authorized client", "err", err) + s.logger.Error("failed to get authorized client", "err", err) s.pages.Notice(w, "pull-comment", "Failed to create comment.") return } - atResp, err := comatproto.RepoPutRecord(r.Context(), client, &comatproto.RepoPutRecord_Input{ - Collection: tangled.RepoPullCommentNSID, - Repo: user.Did, - Rkey: tid.TID(), - Record: &lexutil.LexiconTypeDecoder{ - Val: &tangled.RepoPullComment{ - Pull: pull.AtUri().String(), - Body: body, - CreatedAt: createdAt, - }, - }, + + out, err := comatproto.RepoPutRecord(r.Context(), client, &comatproto.RepoPutRecord_Input{ + Collection: comment.Collection.String(), + Repo: comment.Did.String(), + Rkey: comment.Rkey.String(), + Record: &lexutil.LexiconTypeDecoder{Val: comment.AsRecord()}, }) if err != nil { - l.Error("failed to create pull comment", "err", err) + s.logger.Error("failed to create pull comment", "err", err) s.pages.Notice(w, "pull-comment", "Failed to create comment.") return } - comment := &models.PullComment{ - OwnerDid: user.Did, - RepoAt: f.RepoAt().String(), - PullId: pull.PullId, - Body: body, - CommentAt: atResp.Uri, - SubmissionId: pull.Submissions[roundNumber].ID, - Mentions: mentions, - References: references, + comment.Cid = syntax.CID(out.Cid) + + // Start a transaction + tx, err := s.db.BeginTx(r.Context(), nil) + if err != nil { + l.Error("failed to start transaction", "err", err) + s.pages.Notice(w, "pull-comment", "Failed to create comment.") + return } + defer tx.Rollback() - // Create the pull comment in the database with the commentAt field - commentId, err := db.NewPullComment(tx, comment) + // Create the pull comment in the database + err = db.PutComment(tx, &comment, references) if err != nil { l.Error("failed to create pull comment in database", "err", err) s.pages.Notice(w, "pull-comment", "Failed to create comment.") @@ -905,10 +948,10 @@ s.pages.Notice(w, "pull-comment", "Failed to create comment.") return } - s.notifier.NewPullComment(r.Context(), comment, mentions) + s.notifier.NewPullComment(r.Context(), &comment, mentions) ownerSlashRepo := reporesolver.GetBaseRepoPath(r, f) - s.pages.HxLocation(w, fmt.Sprintf("/%s/pulls/%d#comment-%d", ownerSlashRepo, pull.PullId, commentId)) + s.pages.HxLocation(w, fmt.Sprintf("/%s/pulls/%d#comment-%d", ownerSlashRepo, pull.PullId, comment.Id)) return } } diff --git a/appview/state/state.go b/appview/state/state.go --- a/appview/state/state.go +++ b/appview/state/state.go @@ -126,8 +126,10 @@ tangled.SpindleNSID, tangled.KnotNSID, tangled.StringNSID, tangled.RepoPullNSID, + tangled.RepoPullCommentNSID, tangled.RepoIssueNSID, tangled.RepoIssueCommentNSID, + tangled.FeedCommentNSID, tangled.LabelDefinitionNSID, tangled.LabelOpNSID, }, @@ -146,19 +148,6 @@ } if err := BackfillDefaultDefs(d, res, config.Label.DefaultLabelDefs); err != nil { return nil, fmt.Errorf("failed to backfill default label defs: %w", err) - } - - ingester := appview.Ingester{ - Db: wrapper, - Enforcer: enforcer, - IdResolver: res, - Config: config, - Logger: log.SubLogger(logger, "ingester"), - Validator: validator, - } - err = jc.StartJetstream(ctx, ingester.Ingest()) - if err != nil { - return nil, fmt.Errorf("failed to start jetstream watcher: %w", err) } var notifiers []notify.Notifier @@ -176,6 +165,21 @@ notifiers = append(notifiers, whnotify.NewNotifier(d)) notifier := notify.NewMergedNotifier(notifiers) notifier = lognotify.NewLoggingNotifier(notifier, tlog.SubLogger(logger, "notify")) + + ingester := appview.Ingester{ + Db: wrapper, + Enforcer: enforcer, + IdResolver: res, + Config: config, + Logger: log.SubLogger(logger, "ingester"), + Validator: validator, + MentionsResolver: mentionsResolver, + Notifier: notifier, + } + err = jc.StartJetstream(ctx, ingester.Ingest()) + if err != nil { + return nil, fmt.Errorf("failed to start jetstream watcher: %w", err) + } var cfClient *cloudflare.Client if config.Cloudflare.ApiToken != "" {