appview: replace `PullComment` to `Comment`

+268

appview/db/comments.go

··· 1 + package db 2 + 3 + import ( 4 + "database/sql" 5 + "encoding/json" 6 + "fmt" 7 + "sort" 8 + "strings" 9 + "time" 10 + 11 + "github.com/bluesky-social/indigo/api/atproto" 12 + "github.com/bluesky-social/indigo/atproto/syntax" 13 + "tangled.org/core/api/tangled" 14 + "tangled.org/core/appview/models" 15 + "tangled.org/core/orm" 16 + ) 17 + 18 + func PutComment(tx *sql.Tx, c *models.Comment, references []syntax.ATURI) error { 19 + if c.Collection == "" { 20 + c.Collection = tangled.FeedCommentNSID 21 + } 22 + 23 + var bodyBlobs, replyToUri, replyToCid *string 24 + if len(c.Body.Blobs) > 0 { 25 + encoded, err := json.Marshal(c.Body.Blobs) 26 + if err != nil { 27 + return fmt.Errorf("encoding blobs to json: %w", err) 28 + } 29 + encodedStr := string(encoded) 30 + bodyBlobs = &encodedStr 31 + } 32 + if c.ReplyTo != nil { 33 + replyToUri = &c.ReplyTo.Uri 34 + replyToCid = &c.ReplyTo.Cid 35 + } 36 + result, err := tx.Exec( 37 + // users can change the 'created' date. 38 + // skip update entirely if cid is unchanged. 39 + `insert into comments ( 40 + did, 41 + collection, 42 + rkey, 43 + cid, 44 + subject_uri, 45 + subject_cid, 46 + body_text, 47 + body_original, 48 + body_blobs, 49 + created, 50 + reply_to_uri, 51 + reply_to_cid, 52 + pull_round_idx 53 + ) 54 + values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) 55 + on conflict(did, collection, rkey) 56 + do update set 57 + cid = excluded.cid, 58 + subject_uri = excluded.subject_uri, 59 + subject_cid = excluded.subject_cid, 60 + body_text = excluded.body_text, 61 + body_original = excluded.body_original, 62 + body_blobs = excluded.body_blobs, 63 + created = excluded.created, 64 + reply_to_uri = excluded.reply_to_uri, 65 + reply_to_cid = excluded.reply_to_cid, 66 + pull_round_idx = excluded.pull_round_idx, 67 + edited = ? 68 + where comments.cid != excluded.cid`, 69 + c.Did, 70 + c.Collection, 71 + c.Rkey, 72 + c.Cid, 73 + c.Subject.Uri, 74 + c.Subject.Cid, 75 + c.Body.Text, 76 + c.Body.Original, 77 + bodyBlobs, 78 + c.Created.Format(time.RFC3339), 79 + replyToUri, 80 + replyToCid, 81 + c.PullRoundIdx, 82 + time.Now().Format(time.RFC3339), 83 + ) 84 + if err != nil { 85 + return err 86 + } 87 + 88 + c.Id, err = result.LastInsertId() 89 + if err != nil { 90 + return err 91 + } 92 + 93 + affected, err := result.RowsAffected() 94 + if err != nil { 95 + return err 96 + } 97 + 98 + if affected > 0 { 99 + // update references when comment is updated 100 + if err := putReferences(tx, c.AtUri(), references); err != nil { 101 + return fmt.Errorf("put reference_links: %w", err) 102 + } 103 + } 104 + 105 + return nil 106 + } 107 + 108 + // PurgeComments actually purges a comment row from db instead of marking it as "deleted" 109 + func PurgeComments(e Execer, filters ...orm.Filter) error { 110 + var conditions []string 111 + var args []any 112 + for _, filter := range filters { 113 + conditions = append(conditions, filter.Condition()) 114 + args = append(args, filter.Arg()...) 115 + } 116 + 117 + whereClause := "" 118 + if conditions != nil { 119 + whereClause = " where " + strings.Join(conditions, " and ") 120 + } 121 + 122 + _, err := e.Exec(fmt.Sprintf(`delete from comments %s`, whereClause), args...) 123 + return err 124 + } 125 + 126 + func DeleteComments(e Execer, filters ...orm.Filter) error { 127 + var conditions []string 128 + var args []any 129 + for _, filter := range filters { 130 + conditions = append(conditions, filter.Condition()) 131 + args = append(args, filter.Arg()...) 132 + } 133 + 134 + whereClause := "" 135 + if conditions != nil { 136 + whereClause = " where " + strings.Join(conditions, " and ") 137 + } 138 + 139 + query := fmt.Sprintf( 140 + `update comments 141 + set body_text = "", 142 + body_original = null, 143 + body_blobs = null, 144 + deleted = strftime('%%Y-%%m-%%dT%%H:%%M:%%SZ', 'now') 145 + %s`, 146 + whereClause, 147 + ) 148 + 149 + _, err := e.Exec(query, args...) 150 + return err 151 + } 152 + 153 + func GetComments(e Execer, filters ...orm.Filter) ([]models.Comment, error) { 154 + var comments []models.Comment 155 + 156 + var conditions []string 157 + var args []any 158 + for _, filter := range filters { 159 + conditions = append(conditions, filter.Condition()) 160 + args = append(args, filter.Arg()...) 161 + } 162 + 163 + whereClause := "" 164 + if conditions != nil { 165 + whereClause = " where " + strings.Join(conditions, " and ") 166 + } 167 + 168 + query := fmt.Sprintf(` 169 + select 170 + id, 171 + did, 172 + collection, 173 + rkey, 174 + cid, 175 + subject_uri, 176 + subject_cid, 177 + body_text, 178 + body_original, 179 + body_blobs, 180 + created, 181 + reply_to_uri, 182 + reply_to_cid, 183 + pull_round_idx, 184 + edited, 185 + deleted 186 + from 187 + comments 188 + %s 189 + `, whereClause) 190 + 191 + rows, err := e.Query(query, args...) 192 + if err != nil { 193 + return nil, err 194 + } 195 + defer rows.Close() 196 + 197 + for rows.Next() { 198 + var comment models.Comment 199 + var created string 200 + var cid, bodyBlobs, replyToUri, replyToCid, edited, deleted sql.Null[string] 201 + err := rows.Scan( 202 + &comment.Id, 203 + &comment.Did, 204 + &comment.Collection, 205 + &comment.Rkey, 206 + &cid, 207 + &comment.Subject.Uri, 208 + &comment.Subject.Cid, 209 + &comment.Body.Text, 210 + &comment.Body.Original, 211 + &bodyBlobs, 212 + &created, 213 + &replyToUri, 214 + &replyToCid, 215 + &comment.PullRoundIdx, 216 + &edited, 217 + &deleted, 218 + ) 219 + if err != nil { 220 + return nil, err 221 + } 222 + 223 + if cid.Valid && cid.V != "" { 224 + comment.Cid = syntax.CID(cid.V) 225 + } 226 + 227 + if bodyBlobs.Valid && bodyBlobs.V != "" { 228 + if err := json.Unmarshal([]byte(bodyBlobs.V), &comment.Body.Blobs); err != nil { 229 + return nil, fmt.Errorf("decoding blobs: %w", err) 230 + } 231 + } 232 + 233 + if t, err := time.Parse(time.RFC3339, created); err == nil { 234 + comment.Created = t 235 + } 236 + 237 + if replyToUri.Valid && replyToCid.Valid { 238 + comment.ReplyTo = &atproto.RepoStrongRef{ 239 + Uri: replyToUri.V, 240 + Cid: replyToCid.V, 241 + } 242 + } 243 + 244 + if edited.Valid { 245 + if t, err := time.Parse(time.RFC3339, edited.V); err == nil { 246 + comment.Edited = &t 247 + } 248 + } 249 + 250 + if deleted.Valid { 251 + if t, err := time.Parse(time.RFC3339, deleted.V); err == nil { 252 + comment.Deleted = &t 253 + } 254 + } 255 + 256 + comments = append(comments, comment) 257 + } 258 + 259 + if err := rows.Err(); err != nil { 260 + return nil, err 261 + } 262 + 263 + sort.Slice(comments, func(i, j int) bool { 264 + return comments[i].Created.Before(comments[j].Created) 265 + }) 266 + 267 + return comments, nil 268 + }

+96

appview/db/db.go

··· 1409 1409 return err 1410 1410 }) 1411 1411 1412 + orm.RunMigration(conn, logger, "add-comments-table", func(tx *sql.Tx) error { 1413 + _, err := tx.Exec(` 1414 + drop table if exists comments; 1415 + 1416 + create table comments ( 1417 + -- identifiers 1418 + id integer primary key autoincrement, 1419 + 1420 + did text not null, 1421 + collection text not null default 'sh.tangled.feed.comment', 1422 + rkey text not null, 1423 + at_uri text generated always as ('at://' || did || '/' || collection || '/' || rkey) stored, 1424 + cid text, 1425 + 1426 + -- content 1427 + subject_uri text not null, -- at_uri of subject (issue, pr, string) 1428 + subject_cid text not null, -- cid of subject 1429 + 1430 + body_text text not null, 1431 + body_original text, 1432 + body_blobs text, -- json 1433 + 1434 + created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 1435 + 1436 + reply_to_uri text, -- at_uri of parent comment 1437 + reply_to_cid text, -- cid of parent comment 1438 + 1439 + pull_round_idx integer, -- pull round index. required when subject is sh.tangled.repo.pull 1440 + 1441 + -- appview-local information 1442 + edited text, 1443 + deleted text, 1444 + 1445 + unique(did, collection, rkey) 1446 + ); 1447 + 1448 + insert into comments ( 1449 + did, 1450 + collection, 1451 + rkey, 1452 + subject_uri, 1453 + subject_cid, -- we need to know cid 1454 + body_text, 1455 + created, 1456 + reply_to_uri, 1457 + reply_to_cid, -- we need to know cid 1458 + edited, 1459 + deleted 1460 + ) 1461 + select 1462 + did, 1463 + 'sh.tangled.repo.issue.comment', 1464 + rkey, 1465 + issue_at, 1466 + '', 1467 + body, 1468 + created, 1469 + reply_to, 1470 + '', 1471 + edited, 1472 + deleted 1473 + from issue_comments 1474 + where rkey is not null; 1475 + 1476 + insert into comments ( 1477 + did, 1478 + collection, 1479 + rkey, 1480 + subject_uri, 1481 + subject_cid, -- we need to know cid 1482 + body_text, 1483 + created, 1484 + pull_round_idx 1485 + ) 1486 + select 1487 + c.owner_did, 1488 + 'sh.tangled.repo.pull.comment', 1489 + substr( 1490 + substr(c.comment_at, 6 + instr(substr(c.comment_at, 6), '/')), -- nsid/rkey 1491 + instr( 1492 + substr(c.comment_at, 6 + instr(substr(c.comment_at, 6), '/')), -- nsid/rkey 1493 + '/' 1494 + ) + 1 1495 + ), -- rkey 1496 + p.at_uri, 1497 + '', 1498 + c.body, 1499 + c.created, 1500 + s.round_number 1501 + from pull_comments c 1502 + join pulls p on c.repo_at = p.repo_at and c.pull_id = p.pull_id 1503 + join pull_submissions s on s.id = c.submission_id; 1504 + `) 1505 + return err 1506 + }) 1507 + 1412 1508 return &DB{ 1413 1509 db, 1414 1510 logger,

+15 -132

appview/db/pulls.go

··· 524 524 } 525 525 defer rows.Close() 526 526 527 - submissionMap := make(map[int]*models.PullSubmission) 527 + pullMap := make(map[syntax.ATURI][]*models.PullSubmission) 528 528 529 529 for rows.Next() { 530 530 var submission models.PullSubmission ··· 572 572 submission.Blob.Size = patchBlobSize.V 573 573 } 574 574 575 - submissionMap[submission.ID] = &submission 575 + pullMap[submission.PullAt] = append(pullMap[submission.PullAt], &submission) 576 576 } 577 577 578 578 if err := rows.Err(); err != nil { 579 579 return nil, err 580 580 } 581 581 582 - // Get comments for all submissions using GetPullComments 583 - submissionIds := slices.Collect(maps.Keys(submissionMap)) 584 - comments, err := GetPullComments(e, orm.FilterIn("submission_id", submissionIds)) 582 + // Get comments for all submissions using GetComments 583 + pullAts := slices.Collect(maps.Keys(pullMap)) 584 + comments, err := GetComments(e, orm.FilterIn("subject_uri", pullAts)) 585 585 if err != nil { 586 586 return nil, fmt.Errorf("failed to get pull comments: %w", err) 587 587 } 588 588 for _, comment := range comments { 589 - if submission, ok := submissionMap[comment.SubmissionId]; ok { 590 - submission.Comments = append(submission.Comments, comment) 589 + if comment.PullRoundIdx != nil { 590 + roundIdx := *comment.PullRoundIdx 591 + if submissions, ok := pullMap[syntax.ATURI(comment.Subject.Uri)]; ok { 592 + if roundIdx < len(submissions) { 593 + submission := submissions[roundIdx] 594 + submission.Comments = append(submission.Comments, comment) 595 + } 596 + } 591 597 } 592 598 } 593 599 594 - // group the submissions by pull_at 595 - m := make(map[syntax.ATURI][]*models.PullSubmission) 596 - for _, s := range submissionMap { 597 - m[s.PullAt] = append(m[s.PullAt], s) 598 - } 599 - 600 600 // sort each one by round number 601 - for _, s := range m { 601 + for _, s := range pullMap { 602 602 slices.SortFunc(s, func(a, b *models.PullSubmission) int { 603 603 return cmp.Compare(a.RoundNumber, b.RoundNumber) 604 604 }) 605 605 } 606 606 607 - return m, nil 608 - } 609 - 610 - func GetPullComments(e Execer, filters ...orm.Filter) ([]models.PullComment, error) { 611 - var conditions []string 612 - var args []any 613 - for _, filter := range filters { 614 - conditions = append(conditions, filter.Condition()) 615 - args = append(args, filter.Arg()...) 616 - } 617 - 618 - whereClause := "" 619 - if conditions != nil { 620 - whereClause = " where " + strings.Join(conditions, " and ") 621 - } 622 - 623 - query := fmt.Sprintf(` 624 - select 625 - id, 626 - pull_id, 627 - submission_id, 628 - repo_at, 629 - owner_did, 630 - comment_at, 631 - body, 632 - created 633 - from 634 - pull_comments 635 - %s 636 - order by 637 - created asc 638 - `, whereClause) 639 - 640 - rows, err := e.Query(query, args...) 641 - if err != nil { 642 - return nil, err 643 - } 644 - defer rows.Close() 645 - 646 - commentMap := make(map[string]*models.PullComment) 647 - for rows.Next() { 648 - var comment models.PullComment 649 - var createdAt string 650 - err := rows.Scan( 651 - &comment.ID, 652 - &comment.PullId, 653 - &comment.SubmissionId, 654 - &comment.RepoAt, 655 - &comment.OwnerDid, 656 - &comment.CommentAt, 657 - &comment.Body, 658 - &createdAt, 659 - ) 660 - if err != nil { 661 - return nil, err 662 - } 663 - 664 - if t, err := time.Parse(time.RFC3339, createdAt); err == nil { 665 - comment.Created = t 666 - } 667 - 668 - atUri := comment.AtUri().String() 669 - commentMap[atUri] = &comment 670 - } 671 - 672 - if err := rows.Err(); err != nil { 673 - return nil, err 674 - } 675 - 676 - // collect references for each comments 677 - commentAts := slices.Collect(maps.Keys(commentMap)) 678 - allReferences, err := GetReferencesAll(e, orm.FilterIn("from_at", commentAts)) 679 - if err != nil { 680 - return nil, fmt.Errorf("failed to query reference_links: %w", err) 681 - } 682 - for commentAt, references := range allReferences { 683 - if comment, ok := commentMap[commentAt.String()]; ok { 684 - comment.References = references 685 - } 686 - } 687 - 688 - var comments []models.PullComment 689 - for _, c := range commentMap { 690 - comments = append(comments, *c) 691 - } 692 - 693 - sort.Slice(comments, func(i, j int) bool { 694 - return comments[i].Created.Before(comments[j].Created) 695 - }) 696 - 697 - return comments, nil 607 + return pullMap, nil 698 608 } 699 609 700 610 // timeframe here is directly passed into the sql query filter, and any ··· 771 681 } 772 682 773 683 return pulls, nil 774 - } 775 - 776 - func NewPullComment(tx *sql.Tx, comment *models.PullComment) (int64, error) { 777 - query := `insert into pull_comments (owner_did, repo_at, submission_id, comment_at, pull_id, body) values (?, ?, ?, ?, ?, ?)` 778 - res, err := tx.Exec( 779 - query, 780 - comment.OwnerDid, 781 - comment.RepoAt, 782 - comment.SubmissionId, 783 - comment.CommentAt, 784 - comment.PullId, 785 - comment.Body, 786 - ) 787 - if err != nil { 788 - return 0, err 789 - } 790 - 791 - i, err := res.LastInsertId() 792 - if err != nil { 793 - return 0, err 794 - } 795 - 796 - if err := putReferences(tx, comment.AtUri(), comment.References); err != nil { 797 - return 0, fmt.Errorf("put reference_links: %w", err) 798 - } 799 - 800 - return i, nil 801 684 } 802 685 803 686 // use with transaction

+7 -8

appview/db/reference.go

··· 124 124 values %s 125 125 ) 126 126 select 127 - p.owner_did, p.rkey, 128 - c.comment_at 127 + p.owner_did, p.rkey, c.at_uri 129 128 from input inp 130 129 join repos r 131 130 on r.did = inp.owner_did ··· 133 132 join pulls p 134 133 on p.repo_at = r.at_uri 135 134 and p.pull_id = inp.pull_id 136 - left join pull_comments c 135 + left join comments c 137 136 on inp.comment_id is not null 138 - and c.repo_at = r.at_uri and c.pull_id = p.pull_id 137 + and c.subject_uri = ('at://' || p.owner_did || '/' || 'sh.tangled.repo.pull' || '/' || p.rkey) 139 138 and c.id = inp.comment_id 140 139 `, 141 140 strings.Join(vals, ","), ··· 293 292 return nil, fmt.Errorf("get pull backlinks: %w", err) 294 293 } 295 294 backlinks = append(backlinks, ls...) 296 - ls, err = getPullCommentBacklinks(e, target, backlinksMap[tangled.RepoPullCommentNSID]) 295 + ls, err = getPullCommentBacklinks(e, target, backlinksMap[tangled.FeedCommentNSID]) 297 296 if err != nil { 298 297 return nil, fmt.Errorf("get pull_comment backlinks: %w", err) 299 298 } ··· 430 429 if len(aturis) == 0 { 431 430 return nil, nil 432 431 } 433 - filter := orm.FilterIn("c.comment_at", aturis) 432 + filter := orm.FilterIn("c.at_uri", aturis) 434 433 exclude := orm.FilterNotEq("p.at_uri", target) 435 434 rows, err := e.Query( 436 435 fmt.Sprintf( ··· 438 437 from repos r 439 438 join pulls p 440 439 on r.at_uri = p.repo_at 441 - join pull_comments c 442 - on r.at_uri = c.repo_at and p.pull_id = c.pull_id 440 + join comments c 441 + on ('at://' || p.owner_did || '/' || 'sh.tangled.repo.pull' || '/' || p.rkey) = c.subject_uri 443 442 where %s and %s`, 444 443 filter.Condition(), 445 444 exclude.Condition(),

+108 -6

appview/ingester.go

··· 25 25 "tangled.org/core/api/tangled" 26 26 "tangled.org/core/appview/config" 27 27 "tangled.org/core/appview/db" 28 + "tangled.org/core/appview/mentions" 28 29 "tangled.org/core/appview/models" 30 + "tangled.org/core/appview/notify" 29 31 "tangled.org/core/appview/serververify" 30 32 "tangled.org/core/appview/validator" 31 33 "tangled.org/core/idresolver" ··· 34 36 ) 35 37 36 38 type Ingester struct { 37 - Db db.DbWrapper 38 - Enforcer *rbac.Enforcer 39 - IdResolver *idresolver.Resolver 40 - Config *config.Config 41 - Logger *slog.Logger 42 - Validator *validator.Validator 39 + Db db.DbWrapper 40 + Enforcer *rbac.Enforcer 41 + IdResolver *idresolver.Resolver 42 + Config *config.Config 43 + Logger *slog.Logger 44 + Validator *validator.Validator 45 + MentionsResolver *mentions.Resolver 46 + Notifier notify.Notifier 43 47 } 44 48 45 49 type processFunc func(ctx context.Context, e *jmodels.Event) error ··· 87 91 err = i.ingestIssue(ctx, e) 88 92 case tangled.RepoPullNSID: 89 93 err = i.ingestPull(ctx, e) 94 + case tangled.FeedCommentNSID: 95 + err = i.ingestComment(e) 90 96 case tangled.RepoIssueCommentNSID: 91 97 err = i.ingestIssueComment(e) 98 + case tangled.RepoPullCommentNSID: 99 + err = i.ingestPullComment(e) 92 100 case tangled.LabelDefinitionNSID: 93 101 err = i.ingestLabelDefinition(e) 94 102 case tangled.LabelOpNSID: ··· 1159 1167 orm.FilterEq("rkey", rkey), 1160 1168 ); err != nil { 1161 1169 return fmt.Errorf("failed to delete issue comment record: %w", err) 1170 + } 1171 + 1172 + return nil 1173 + } 1174 + 1175 + return nil 1176 + } 1177 + 1178 + // ingestPullComment ingests legacy sh.tangled.repo.pull.comment deletions 1179 + func (i *Ingester) ingestPullComment(e *jmodels.Event) error { 1180 + l := i.Logger.With("handler", "ingestPullComment", "nsid", e.Commit.Collection, "did", e.Did, "rkey", e.Commit.RKey) 1181 + l.Info("ingesting record") 1182 + 1183 + switch e.Commit.Operation { 1184 + case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: 1185 + // no-op. sh.tangled.repo.pull.comment is deprecated 1186 + 1187 + case jmodels.CommitOperationDelete: 1188 + if err := db.PurgeComments( 1189 + i.Db, 1190 + orm.FilterEq("did", e.Did), 1191 + orm.FilterEq("collection", e.Commit.Collection), 1192 + orm.FilterEq("rkey", e.Commit.RKey), 1193 + ); err != nil { 1194 + return fmt.Errorf("failed to delete comment record: %w", err) 1195 + } 1196 + } 1197 + 1198 + return nil 1199 + } 1200 + 1201 + func (i *Ingester) ingestComment(e *jmodels.Event) error { 1202 + did := e.Did 1203 + rkey := e.Commit.RKey 1204 + cid := e.Commit.CID 1205 + 1206 + var err error 1207 + 1208 + l := i.Logger.With("handler", "ingestComment", "nsid", e.Commit.Collection, "did", did, "rkey", rkey) 1209 + l.Info("ingesting record") 1210 + 1211 + ddb, ok := i.Db.Execer.(*db.DB) 1212 + if !ok { 1213 + return fmt.Errorf("failed to index issue comment record, invalid db cast") 1214 + } 1215 + 1216 + ctx := context.Background() 1217 + 1218 + switch e.Commit.Operation { 1219 + case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: 1220 + raw := json.RawMessage(e.Commit.Record) 1221 + record := tangled.FeedComment{} 1222 + err = json.Unmarshal(raw, &record) 1223 + if err != nil { 1224 + return fmt.Errorf("invalid record: %w", err) 1225 + } 1226 + 1227 + comment, err := models.CommentFromRecord(syntax.DID(did), syntax.RecordKey(rkey), syntax.CID(cid), record) 1228 + if err != nil { 1229 + return fmt.Errorf("failed to parse comment from record: %w", err) 1230 + } 1231 + 1232 + if err := comment.Validate(); err != nil { 1233 + return fmt.Errorf("failed to validate comment: %w", err) 1234 + } 1235 + 1236 + var references []syntax.ATURI 1237 + if comment.Body.Original != nil { 1238 + _, references = i.MentionsResolver.Resolve(ctx, *comment.Body.Original) 1239 + } 1240 + 1241 + tx, err := ddb.Begin() 1242 + if err != nil { 1243 + return fmt.Errorf("failed to start transaction: %w", err) 1244 + } 1245 + defer tx.Rollback() 1246 + 1247 + err = db.PutComment(tx, comment, references) 1248 + if err != nil { 1249 + return fmt.Errorf("failed to create comment: %w", err) 1250 + } 1251 + 1252 + if err := tx.Commit(); err != nil { 1253 + return err 1254 + } 1255 + 1256 + case jmodels.CommitOperationDelete: 1257 + if err := db.DeleteComments( 1258 + ddb, 1259 + orm.FilterEq("did", did), 1260 + orm.FilterEq("collection", e.Commit.Collection), 1261 + orm.FilterEq("rkey", rkey), 1262 + ); err != nil { 1263 + return fmt.Errorf("failed to delete comment record: %w", err) 1162 1264 } 1163 1265 1164 1266 return nil

+147

appview/models/comment.go

··· 1 + package models 2 + 3 + import ( 4 + "fmt" 5 + "strings" 6 + "time" 7 + 8 + comatproto "github.com/bluesky-social/indigo/api/atproto" 9 + "github.com/bluesky-social/indigo/atproto/syntax" 10 + typegen "github.com/whyrusleeping/cbor-gen" 11 + "tangled.org/core/api/tangled" 12 + ) 13 + 14 + type Comment struct { 15 + Id int64 16 + 17 + Did syntax.DID 18 + Collection syntax.NSID 19 + Rkey syntax.RecordKey 20 + Cid syntax.CID 21 + 22 + // record content 23 + Subject comatproto.RepoStrongRef 24 + Body tangled.MarkupMarkdown // markup body type. only markdown is supported right now 25 + Created time.Time 26 + ReplyTo *comatproto.RepoStrongRef // (optional) parent comment 27 + PullRoundIdx *int // (optional) pull round number used when subject is sh.tangled.repo.pull 28 + 29 + // store on db, but not on PDS 30 + Edited *time.Time 31 + Deleted *time.Time 32 + } 33 + 34 + func (c *Comment) AtUri() syntax.ATURI { 35 + return syntax.ATURI(fmt.Sprintf("at://%s/%s/%s", c.Did, c.Collection, c.Rkey)) 36 + } 37 + 38 + func (c *Comment) StrongRef() comatproto.RepoStrongRef { 39 + return comatproto.RepoStrongRef{ 40 + Uri: c.AtUri().String(), 41 + Cid: c.Cid.String(), 42 + } 43 + } 44 + 45 + func (c *Comment) AsRecord() typegen.CBORMarshaler { 46 + // can't convert to record for legacy types 47 + if c.Collection != tangled.FeedCommentNSID { 48 + return nil 49 + } 50 + var pullRoundIdx int64 51 + if c.PullRoundIdx != nil { 52 + pullRoundIdx = int64(*c.PullRoundIdx) 53 + } 54 + return &tangled.FeedComment{ 55 + Subject: &c.Subject, 56 + Body: &tangled.FeedComment_Body{MarkupMarkdown: &c.Body}, 57 + CreatedAt: c.Created.Format(time.RFC3339), 58 + ReplyTo: c.ReplyTo, 59 + PullRoundIdx: &pullRoundIdx, 60 + } 61 + } 62 + 63 + func (c *Comment) IsTopLevel() bool { 64 + return c.ReplyTo == nil 65 + } 66 + 67 + func (c *Comment) IsReply() bool { 68 + return c.ReplyTo != nil 69 + } 70 + 71 + func (c *Comment) Validate() error { 72 + // TODO: sanitize the body and then trim space 73 + if sb := strings.TrimSpace(c.Body.Text); sb == "" { 74 + return fmt.Errorf("body is empty after HTML sanitization") 75 + } 76 + 77 + // if it's for PR, PullSubmissionId should not be nil 78 + subjectAt, err := syntax.ParseATURI(c.Subject.Uri) 79 + if err != nil { 80 + return fmt.Errorf("subject.uri is not valid at-uri: %w", err) 81 + } 82 + if subjectAt.Collection().String() == tangled.RepoPullNSID { 83 + if c.PullRoundIdx == nil { 84 + return fmt.Errorf("pullSubmissionId should not be nil when subject is sh.tangled.repo.pull") 85 + } 86 + } 87 + return nil 88 + } 89 + 90 + func CommentFromRecord(did syntax.DID, rkey syntax.RecordKey, cid syntax.CID, record tangled.FeedComment) (*Comment, error) { 91 + created, err := time.Parse(time.RFC3339, record.CreatedAt) 92 + if err != nil { 93 + created = time.Now() 94 + } 95 + 96 + if record.Subject == nil { 97 + return nil, fmt.Errorf("subject can't be nil") 98 + } 99 + subjectAt, err := syntax.ParseATURI(record.Subject.Uri) 100 + if err != nil { 101 + return nil, fmt.Errorf("invalid subject uri: %w", err) 102 + } 103 + if _, err = syntax.ParseCID(record.Subject.Cid); err != nil { 104 + return nil, fmt.Errorf("invalid subject cid: %w", err) 105 + } 106 + 107 + if subjectAt.Collection() == tangled.RepoPullNSID { 108 + if record.PullRoundIdx == nil { 109 + return nil, fmt.Errorf("pullRoundIdx can't be nil when subject is sh.tangled.repo.pull") 110 + } 111 + } 112 + 113 + if record.Body == nil { 114 + return nil, fmt.Errorf("body can't be nil") 115 + } 116 + if record.Body.MarkupMarkdown == nil { 117 + return nil, fmt.Errorf("body should be markdown type") 118 + } 119 + 120 + if record.ReplyTo != nil { 121 + if _, err = syntax.ParseATURI(record.ReplyTo.Uri); err != nil { 122 + return nil, fmt.Errorf("invalid replyTo uri: %w", err) 123 + } 124 + if _, err = syntax.ParseCID(record.ReplyTo.Cid); err != nil { 125 + return nil, fmt.Errorf("invalid replyTo cid: %w", err) 126 + } 127 + } 128 + 129 + var pullRoundIdx *int 130 + if record.PullRoundIdx != nil { 131 + pullRoundIdx = new(int) 132 + *pullRoundIdx = int(*record.PullRoundIdx) 133 + } 134 + 135 + return &Comment{ 136 + Did: did, 137 + Collection: tangled.FeedCommentNSID, 138 + Rkey: rkey, 139 + Cid: cid, 140 + 141 + Subject: *record.Subject, 142 + Body: *record.Body.MarkupMarkdown, 143 + Created: created, 144 + ReplyTo: record.ReplyTo, 145 + PullRoundIdx: pullRoundIdx, 146 + }, nil 147 + }

+2 -28

appview/models/pull.go

··· 299 299 Blob lexutil.LexBlob 300 300 Patch string 301 301 Combined string 302 - Comments []PullComment 302 + Comments []Comment 303 303 SourceRev string // include the rev that was used to create this submission: only for branch/fork PRs 304 304 305 305 // meta 306 306 Created time.Time 307 - } 308 - 309 - type PullComment struct { 310 - // ids 311 - ID int 312 - PullId int 313 - SubmissionId int 314 - 315 - // at ids 316 - RepoAt string 317 - OwnerDid string 318 - CommentAt string 319 - 320 - // content 321 - Body string 322 - 323 - // meta 324 - Mentions []syntax.DID 325 - References []syntax.ATURI 326 - 327 - // meta 328 - Created time.Time 329 - } 330 - 331 - func (p *PullComment) AtUri() syntax.ATURI { 332 - return syntax.ATURI(p.CommentAt) 333 307 } 334 308 335 309 func (p *Pull) TotalComments() int { ··· 451 425 addParticipant(s.PullAt.Authority().String()) 452 426 453 427 for _, c := range s.Comments { 454 - addParticipant(c.OwnerDid) 428 + addParticipant(c.Did.String()) 455 429 } 456 430 457 431 return participants

+13 -7

appview/notify/db/db.go

··· 281 281 ) 282 282 } 283 283 284 - func (n *databaseNotifier) NewPullComment(ctx context.Context, comment *models.PullComment, mentions []syntax.DID) { 284 + func (n *databaseNotifier) NewPullComment(ctx context.Context, comment *models.Comment, mentions []syntax.DID) { 285 285 l := log.FromContext(ctx) 286 286 287 - pull, err := db.GetPull(n.db, 288 - orm.FilterEq("repo_at", syntax.ATURI(comment.RepoAt)), 289 - orm.FilterEq("pull_id", comment.PullId), 287 + subjectAt := syntax.ATURI(comment.Subject.Uri) 288 + pulls, err := db.GetPulls(n.db, 289 + orm.FilterEq("owner_did", subjectAt.Authority()), 290 + orm.FilterEq("rkey", subjectAt.RecordKey()), 290 291 ) 291 292 if err != nil { 292 - l.Error("failed to get pulls", "err", err) 293 + l.Error("failed to get pull", "err", err) 293 294 return 294 295 } 296 + if len(pulls) == 0 { 297 + l.Error("NewPullComment: no pull found", "aturi", comment.Subject) 298 + return 299 + } 300 + pull := pulls[0] 295 301 296 - repo, err := db.GetRepo(n.db, orm.FilterEq("at_uri", comment.RepoAt)) 302 + repo, err := db.GetRepo(n.db, orm.FilterEq("at_uri", pull.RepoAt)) 297 303 if err != nil { 298 304 l.Error("failed to get repos", "err", err) 299 305 return ··· 311 317 recipients.Remove(m) 312 318 } 313 319 314 - actorDid := syntax.DID(comment.OwnerDid) 320 + actorDid := comment.Did 315 321 eventType := models.NotificationTypePullCommented 316 322 entityType := "pull" 317 323 entityId := pull.AtUri().String()

+1 -1

appview/notify/logging/notifier.go

··· 86 86 l.inner.NewPull(ctx, pull) 87 87 } 88 88 89 - func (l *loggingNotifier) NewPullComment(ctx context.Context, comment *models.PullComment, mentions []syntax.DID) { 89 + func (l *loggingNotifier) NewPullComment(ctx context.Context, comment *models.Comment, mentions []syntax.DID) { 90 90 ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "NewPullComment")) 91 91 l.inner.NewPullComment(ctx, comment, mentions) 92 92 }

+1 -1

appview/notify/merged_notifier.go

··· 82 82 m.fanout(func(n Notifier) { n.NewPull(ctx, pull) }) 83 83 } 84 84 85 - func (m *mergedNotifier) NewPullComment(ctx context.Context, comment *models.PullComment, mentions []syntax.DID) { 85 + func (m *mergedNotifier) NewPullComment(ctx context.Context, comment *models.Comment, mentions []syntax.DID) { 86 86 m.fanout(func(n Notifier) { n.NewPullComment(ctx, comment, mentions) }) 87 87 } 88 88

+2 -2

appview/notify/notifier.go

··· 23 23 DeleteFollow(ctx context.Context, follow *models.Follow) 24 24 25 25 NewPull(ctx context.Context, pull *models.Pull) 26 - NewPullComment(ctx context.Context, comment *models.PullComment, mentions []syntax.DID) 26 + NewPullComment(ctx context.Context, comment *models.Comment, mentions []syntax.DID) 27 27 NewPullState(ctx context.Context, actor syntax.DID, pull *models.Pull) 28 28 29 29 NewIssueLabelOp(ctx context.Context, issue *models.Issue) ··· 64 64 func (m *BaseNotifier) DeleteFollow(ctx context.Context, follow *models.Follow) {} 65 65 66 66 func (m *BaseNotifier) NewPull(ctx context.Context, pull *models.Pull) {} 67 - func (m *BaseNotifier) NewPullComment(ctx context.Context, models *models.PullComment, mentions []syntax.DID) { 67 + func (m *BaseNotifier) NewPullComment(ctx context.Context, models *models.Comment, mentions []syntax.DID) { 68 68 } 69 69 func (m *BaseNotifier) NewPullState(ctx context.Context, actor syntax.DID, pull *models.Pull) {} 70 70

+3 -4

appview/notify/posthog/notifier.go

··· 86 86 } 87 87 } 88 88 89 - func (n *posthogNotifier) NewPullComment(ctx context.Context, comment *models.PullComment, mentions []syntax.DID) { 89 + func (n *posthogNotifier) NewPullComment(ctx context.Context, comment *models.Comment, mentions []syntax.DID) { 90 90 err := n.client.Enqueue(posthog.Capture{ 91 - DistinctId: comment.OwnerDid, 91 + DistinctId: comment.Did.String(), 92 92 Event: "new_pull_comment", 93 93 Properties: posthog.Properties{ 94 - "repo_at": comment.RepoAt, 95 - "pull_id": comment.PullId, 94 + "pull_at": comment.Subject, 96 95 "mentions": mentions, 97 96 }, 98 97 })

+1

appview/oauth/scopes.go

··· 16 16 "repo:sh.tangled.spindle", 17 17 "repo:sh.tangled.spindle.member", 18 18 "repo:sh.tangled.graph.follow", 19 + "repo:sh.tangled.feed.comment", 19 20 "repo:sh.tangled.feed.star", 20 21 "repo:sh.tangled.feed.reaction", 21 22 "repo:sh.tangled.label.definition",

+5 -5

appview/pages/templates/repo/pulls/pull.html

··· 625 625 {{ end }} 626 626 627 627 {{ define "submissionComment" }} 628 - <div id="comment-{{.ID}}" class="flex gap-2 -ml-4 py-4 w-full mx-auto group/comment"> 628 + <div id="comment-{{.Id}}" class="flex gap-2 -ml-4 py-4 w-full mx-auto group/comment"> 629 629  630 630 <div class="flex-shrink-0 h-fit relative"> 631 - {{ template "user/fragments/picLink" (list .OwnerDid "size-8") }} 631 + {{ template "user/fragments/picLink" (list .Did.String "size-8") }} 632 632 </div> 633 633  634 634 <div class="flex-1 min-w-0"> 635 635  636 636 <div class="text-sm text-gray-500 dark:text-gray-400 flex items-center gap-1 group-target/comment:bg-yellow-200/30 group-target/comment:dark:bg-yellow-600/30"> 637 - {{ $handle := resolve .OwnerDid }} 637 + {{ $handle := resolve .Did.String }} 638 638 <a class="text-gray-500 dark:text-gray-400 hover:text-gray-500 dark:hover:text-gray-300" href="/{{ $handle }}">{{ $handle }}</a> 639 639 <span class="before:content-['·']"></span> 640 - <a class="text-gray-500 dark:text-gray-400 hover:text-gray-500 dark:hover:text-gray-300" href="#comment-{{.ID}}"> 640 + <a class="text-gray-500 dark:text-gray-400 hover:text-gray-500 dark:hover:text-gray-300" href="#comment-{{.Id}}"> 641 641 {{ template "repo/fragments/shortTime" .Created }} 642 642 </a> 643 643 </div> 644 644  645 645 <div class="prose dark:prose-invert mt-1"> 646 - {{ .Body | markdown }} 646 + {{ .Body.Text | markdown }} 647 647 </div> 648 648 </div> 649 649 </div>

+76 -33

appview/pulls/pulls.go

··· 838 838 case http.MethodPost: 839 839 body := r.FormValue("body") 840 840 if body == "" { 841 - s.pages.Notice(w, "pull", "Comment body is required") 841 + s.pages.Notice(w, "pull-comment", "Comment body is required") 842 842 return 843 843 } 844 844 845 + // TODO(boltless): normalize markdown body 846 + normalizedBody := body 845 847 mentions, references := s.mentionsResolver.Resolve(r.Context(), body) 846 848 847 - // Start a transaction 848 - tx, err := s.db.BeginTx(r.Context(), nil) 849 + markdownBody := tangled.MarkupMarkdown{ 850 + Text: normalizedBody, 851 + Original: &body, 852 + Blobs: nil, 853 + } 854 + 855 + // ingest CID of PR record on-demand. 856 + // TODO(boltless): appview should ingest CID of atproto records 857 + cid, err := func() (syntax.CID, error) { 858 + ident, err := s.idResolver.ResolveIdent(r.Context(), pull.OwnerDid) 859 + if err != nil { 860 + return "", err 861 + } 862 + 863 + xrpcc := indigoxrpc.Client{Host: ident.PDSEndpoint()} 864 + out, err := comatproto.RepoGetRecord(r.Context(), &xrpcc, "", tangled.RepoPullNSID, pull.OwnerDid, pull.Rkey) 865 + if err != nil { 866 + return "", err 867 + } 868 + if out.Cid == nil { 869 + return "", fmt.Errorf("record CID is empty") 870 + } 871 + 872 + cid, err := syntax.ParseCID(*out.Cid) 873 + if err != nil { 874 + return "", err 875 + } 876 + 877 + return cid, nil 878 + }() 849 879 if err != nil { 850 - l.Error("failed to start transaction", "err", err) 880 + s.logger.Error("failed to backfill subject PR record", "err", err) 881 + s.pages.Notice(w, "pull-comment", "failed to backfill subject record") 882 + return 883 + } 884 + pullStrongRef := comatproto.RepoStrongRef{ 885 + Uri: pull.AtUri().String(), 886 + Cid: cid.String(), 887 + } 888 + 889 + comment := models.Comment{ 890 + Did: syntax.DID(user.Did), 891 + Collection: tangled.FeedCommentNSID, 892 + Rkey: syntax.RecordKey(tid.TID()), 893 + 894 + Subject: pullStrongRef, 895 + Body: markdownBody, 896 + Created: time.Now(), 897 + ReplyTo: nil, 898 + PullRoundIdx: &roundNumber, 899 + } 900 + if err = comment.Validate(); err != nil { 901 + s.logger.Error("failed to validate comment", "err", err) 851 902 s.pages.Notice(w, "pull-comment", "Failed to create comment.") 852 903 return 853 904 } 854 - defer tx.Rollback() 855 - 856 - createdAt := time.Now().Format(time.RFC3339) 857 905 858 906 client, err := s.oauth.AuthorizedClient(r) 859 907 if err != nil { 860 - l.Error("failed to get authorized client", "err", err) 908 + s.logger.Error("failed to get authorized client", "err", err) 861 909 s.pages.Notice(w, "pull-comment", "Failed to create comment.") 862 910 return 863 911 } 864 - atResp, err := comatproto.RepoPutRecord(r.Context(), client, &comatproto.RepoPutRecord_Input{ 865 - Collection: tangled.RepoPullCommentNSID, 866 - Repo: user.Did, 867 - Rkey: tid.TID(), 868 - Record: &lexutil.LexiconTypeDecoder{ 869 - Val: &tangled.RepoPullComment{ 870 - Pull: pull.AtUri().String(), 871 - Body: body, 872 - CreatedAt: createdAt, 873 - }, 874 - }, 912 + 913 + out, err := comatproto.RepoPutRecord(r.Context(), client, &comatproto.RepoPutRecord_Input{ 914 + Collection: comment.Collection.String(), 915 + Repo: comment.Did.String(), 916 + Rkey: comment.Rkey.String(), 917 + Record: &lexutil.LexiconTypeDecoder{Val: comment.AsRecord()}, 875 918 }) 876 919 if err != nil { 877 - l.Error("failed to create pull comment", "err", err) 920 + s.logger.Error("failed to create pull comment", "err", err) 878 921 s.pages.Notice(w, "pull-comment", "Failed to create comment.") 879 922 return 880 923 } 881 924 882 - comment := &models.PullComment{ 883 - OwnerDid: user.Did, 884 - RepoAt: f.RepoAt().String(), 885 - PullId: pull.PullId, 886 - Body: body, 887 - CommentAt: atResp.Uri, 888 - SubmissionId: pull.Submissions[roundNumber].ID, 889 - Mentions: mentions, 890 - References: references, 925 + comment.Cid = syntax.CID(out.Cid) 926 + 927 + // Start a transaction 928 + tx, err := s.db.BeginTx(r.Context(), nil) 929 + if err != nil { 930 + l.Error("failed to start transaction", "err", err) 931 + s.pages.Notice(w, "pull-comment", "Failed to create comment.") 932 + return 891 933 } 934 + defer tx.Rollback() 892 935 893 - // Create the pull comment in the database with the commentAt field 894 - commentId, err := db.NewPullComment(tx, comment) 936 + // Create the pull comment in the database 937 + err = db.PutComment(tx, &comment, references) 895 938 if err != nil { 896 939 l.Error("failed to create pull comment in database", "err", err) 897 940 s.pages.Notice(w, "pull-comment", "Failed to create comment.") ··· 905 948 return 906 949 } 907 950 908 - s.notifier.NewPullComment(r.Context(), comment, mentions) 951 + s.notifier.NewPullComment(r.Context(), &comment, mentions) 909 952 910 953 ownerSlashRepo := reporesolver.GetBaseRepoPath(r, f) 911 - s.pages.HxLocation(w, fmt.Sprintf("/%s/pulls/%d#comment-%d", ownerSlashRepo, pull.PullId, commentId)) 954 + s.pages.HxLocation(w, fmt.Sprintf("/%s/pulls/%d#comment-%d", ownerSlashRepo, pull.PullId, comment.Id)) 912 955 return 913 956 } 914 957 }

+17 -13

appview/state/state.go

··· 126 126 tangled.KnotNSID, 127 127 tangled.StringNSID, 128 128 tangled.RepoPullNSID, 129 + tangled.RepoPullCommentNSID, 129 130 tangled.RepoIssueNSID, 130 131 tangled.RepoIssueCommentNSID, 132 + tangled.FeedCommentNSID, 131 133 tangled.LabelDefinitionNSID, 132 134 tangled.LabelOpNSID, 133 135 }, ··· 146 148 147 149 if err := BackfillDefaultDefs(d, res, config.Label.DefaultLabelDefs); err != nil { 148 150 return nil, fmt.Errorf("failed to backfill default label defs: %w", err) 149 - } 150 - 151 - ingester := appview.Ingester{ 152 - Db: wrapper, 153 - Enforcer: enforcer, 154 - IdResolver: res, 155 - Config: config, 156 - Logger: log.SubLogger(logger, "ingester"), 157 - Validator: validator, 158 - } 159 - err = jc.StartJetstream(ctx, ingester.Ingest()) 160 - if err != nil { 161 - return nil, fmt.Errorf("failed to start jetstream watcher: %w", err) 162 151 } 163 152 164 153 var notifiers []notify.Notifier ··· 176 165 177 166 notifier := notify.NewMergedNotifier(notifiers) 178 167 notifier = lognotify.NewLoggingNotifier(notifier, tlog.SubLogger(logger, "notify")) 168 + 169 + ingester := appview.Ingester{ 170 + Db: wrapper, 171 + Enforcer: enforcer, 172 + IdResolver: res, 173 + Config: config, 174 + Logger: log.SubLogger(logger, "ingester"), 175 + Validator: validator, 176 + MentionsResolver: mentionsResolver, 177 + Notifier: notifier, 178 + } 179 + err = jc.StartJetstream(ctx, ingester.Ingest()) 180 + if err != nil { 181 + return nil, fmt.Errorf("failed to start jetstream watcher: %w", err) 182 + } 179 183 180 184 var cfClient *cloudflare.Client 181 185 if config.Cloudflare.ApiToken != "" {

Configure Feed

Configure Feed