+8 -168

main.go

reviewed

··· 20 20 "tangled.org/whey.party/red-dwarf-server/microcosm/slingshot" 21 21 appbskyactordefs "tangled.org/whey.party/red-dwarf-server/shims/lex/app/bsky/actor/defs" 22 22 appbskyfeeddefs "tangled.org/whey.party/red-dwarf-server/shims/lex/app/bsky/feed/defs" 23 23 + appbskyunspeccedgetpostthreadv2 "tangled.org/whey.party/red-dwarf-server/shims/lex/app/bsky/unspecced/getpostthreadv2" 23 24 "tangled.org/whey.party/red-dwarf-server/shims/utils" 24 25 "tangled.org/whey.party/red-dwarf-server/sticket" 25 26 "tangled.org/whey.party/red-dwarf-server/store" ··· 79 80 cs := constellation.NewConstellation(CONSTELLATION_URL) 80 81 // spacedust is type definitions only 81 82 // jetstream types is probably available from jetstream/pkg/models 83 83 + 84 84 + //threadGraphCache := make(map[syntax.ATURI]appbskyunspeccedgetpostthreadv2.ThreadGraph) 82 85 83 86 router_raw := gin.New() 84 87 router_raw.Use(gin.Logger()) ··· 626 629 } 627 630 router.GET("/xrpc/app.bsky.unspecced.getPostThreadV2", 628 631 func(c *gin.Context) { 629 629 - ctx := c.Request.Context() 630 630 - 631 631 - rawdid := c.GetString("user_did") 632 632 - var viewer *utils.DID 633 633 - didval, errdid := utils.NewDID(rawdid) 634 634 - if errdid != nil { 635 635 - viewer = nil 636 636 - } else { 637 637 - viewer = &didval 638 638 - } 639 639 - 640 640 - threadAnchorURIraw := c.Query("anchor") 641 641 - if threadAnchorURIraw == "" { 642 642 - c.JSON(http.StatusBadRequest, gin.H{"error": "Missing feed param"}) 643 643 - return 644 644 - } 645 645 - 646 646 - threadAnchorURI, err := syntax.ParseATURI(threadAnchorURIraw) 647 647 - if err != nil { 648 648 - return 649 649 - } 650 650 - 651 651 - //var thread []*appbsky.UnspeccedGetPostThreadOtherV2_ThreadItem 652 652 - 653 653 - var skeletonposts []string 654 654 - 655 655 - emptystrarray := &[]string{} 656 656 - limit := 100 657 657 - 658 658 - // recurse to the top 659 659 - parentsMap := map[string]*appbsky.FeedDefs_PostView{} 660 660 - current := threadAnchorURI 661 661 - iteration := 0 662 662 - for true { 663 663 - iteration = iteration + 1 664 664 - if iteration > 10 { 665 665 - break 666 666 - } 667 667 - recordaturi, err := syntax.ParseATURI("at://" + current.Authority().String() + "/" + string(current.Collection()) + "/" + string(current.RecordKey())) 668 668 - if err != nil { 669 669 - break 670 670 - } 671 671 - // loop 672 672 - postView, postRecord, err := appbskyfeeddefs.PostView(ctx, recordaturi.String(), sl, cs, BSKYIMAGECDN_URL, viewer, 3) 673 673 - if err != nil { 674 674 - break 675 675 - } 676 676 - if postView == nil { 677 677 - break 678 678 - } 679 679 - //if iteration != 1 { 680 680 - skeletonposts = append([]string{recordaturi.String()}, skeletonposts...) 681 681 - //} 682 682 - parentsMap[recordaturi.String()] = postView 683 683 - if postRecord != nil && postRecord.Reply != nil && postRecord.Reply.Parent != nil && postRecord.Reply.Root != nil { 684 684 - if postRecord.Reply.Parent.Uri == postRecord.Reply.Root.Uri { 685 685 - //break 686 686 - iteration = 9 687 687 - } 688 688 - 689 689 - current, err = syntax.ParseATURI(postRecord.Reply.Parent.Uri) 690 690 - if err != nil { 691 691 - break 692 692 - } 693 693 - } else { 694 694 - log.Println("what huh what hwat") 695 695 - break //whatt 696 696 - } 697 697 - 698 698 - } 699 699 - 700 700 - //skeletonposts = append(skeletonposts, threadAnchorURI.String()) 701 701 - // todo: theres a cursor!!! pagination please! 702 702 - // todo: also i doubt im gonna do proper threadding right now, so make sure to remind me to do it properly some time later thanks 703 703 - //rootReplies, _ := constellation.GetBacklinks(ctx, cs, string(threadAnchorURI), "app.bsky.feed.post:reply.root.uri", *emptystrarray, &limit, nil) 704 704 - parentReplies, _ := constellation.GetBacklinks(ctx, cs, string(threadAnchorURI), "app.bsky.feed.post:reply.parent.uri", *emptystrarray, &limit, nil) 705 705 - 706 706 - for _, rec := range parentReplies.Records { 707 707 - recordaturi, err := syntax.ParseATURI("at://" + rec.Did + "/" + rec.Collection + "/" + rec.Rkey) 708 708 - if err != nil { 709 709 - continue 710 710 - } 711 711 - skeletonposts = append(skeletonposts, recordaturi.String()) 712 712 - } 713 713 - maplen := len(parentsMap) 714 714 - concurrentResults := MapConcurrent( 715 715 - ctx, 716 716 - skeletonposts, 717 717 - 20, 718 718 - func(ctx context.Context, raw string, idx int) (*appbsky.UnspeccedGetPostThreadOtherV2_ThreadItem, error) { 719 719 - var postView *appbsky.FeedDefs_PostView 720 720 - fromParentChain := false 721 721 - postView, ok := parentsMap[raw] 722 722 - if !ok { 723 723 - post, _, err := appbskyfeeddefs.PostView(ctx, raw, sl, cs, BSKYIMAGECDN_URL, viewer, 3) 724 724 - if err != nil { 725 725 - return nil, err 726 726 - } 727 727 - if post == nil { 728 728 - return nil, fmt.Errorf("post not found") 729 729 - } 730 730 - postView = post 731 731 - } else { 732 732 - fromParentChain = true 733 733 - } 734 734 - 735 735 - depth := int64(1) 736 736 - if raw == threadAnchorURI.String() { 737 737 - depth = 0 738 738 - } 739 739 - if fromParentChain { 740 740 - depth = int64(0 - maplen + idx + 1) 741 741 - } 742 742 - 743 743 - return &appbsky.UnspeccedGetPostThreadOtherV2_ThreadItem{ 744 744 - // Depth int64 `json:"depth" cborgen:"depth"` 745 745 - Depth: depth, // todo: placeholder 746 746 - // Uri string `json:"uri" cborgen:"uri"` 747 747 - Uri: raw, 748 748 - // Value *UnspeccedGetPostThreadOtherV2_ThreadItem_Value `json:"value" cborgen:"value"` 749 749 - Value: &appbsky.UnspeccedGetPostThreadOtherV2_ThreadItem_Value{ 750 750 - // UnspeccedDefs_ThreadItemPost *UnspeccedDefs_ThreadItemPost 751 751 - UnspeccedDefs_ThreadItemPost: &appbsky.UnspeccedDefs_ThreadItemPost{ 752 752 - // LexiconTypeID string `json:"$type" cborgen:"$type,const=app.bsky.unspecced.defs#threadItemPost"` 753 753 - LexiconTypeID: "app.bsky.unspecced.defs#threadItemPost", 754 754 - // // hiddenByThreadgate: The threadgate created by the author indicates this post as a reply to be hidden for everyone consuming the thread. 755 755 - // HiddenByThreadgate bool `json:"hiddenByThreadgate" cborgen:"hiddenByThreadgate"` 756 756 - HiddenByThreadgate: false, // todo: placeholder 757 757 - // // moreParents: This post has more parents that were not present in the response. This is just a boolean, without the number of parents. 758 758 - // MoreParents bool `json:"moreParents" cborgen:"moreParents"` 759 759 - MoreParents: false, // todo: placeholder 760 760 - // // moreReplies: This post has more replies that were not present in the response. This is a numeric value, which is best-effort and might not be accurate. 761 761 - // MoreReplies int64 `json:"moreReplies" cborgen:"moreReplies"` 762 762 - MoreReplies: 0, // todo: placeholder 763 763 - // // mutedByViewer: This is by an account muted by the viewer requesting it. 764 764 - // MutedByViewer bool `json:"mutedByViewer" cborgen:"mutedByViewer"` 765 765 - MutedByViewer: false, // todo: placeholder 766 766 - // // opThread: This post is part of a contiguous thread by the OP from the thread root. Many different OP threads can happen in the same thread. 767 767 - // OpThread bool `json:"opThread" cborgen:"opThread"` 768 768 - OpThread: false, // todo: placeholder 769 769 - // Post *FeedDefs_PostView `json:"post" cborgen:"post"` 770 770 - Post: postView, 771 771 - }, 772 772 - }, 773 773 - }, nil 774 774 - }, 775 775 - ) 776 776 - 777 777 - // build final slice 778 778 - out := make([]*appbsky.UnspeccedGetPostThreadOtherV2_ThreadItem, 0, len(concurrentResults)) 779 779 - for _, r := range concurrentResults { 780 780 - if r.Err == nil && r.Value != nil && r.Value.Value != nil && r.Value.Value.UnspeccedDefs_ThreadItemPost != nil && r.Value.Value.UnspeccedDefs_ThreadItemPost.Post != nil { 781 781 - out = append(out, r.Value) 782 782 - } 783 783 - } 784 784 - 785 785 - // c.JSON(http.StatusOK, &appbsky.UnspeccedGetPostThreadOtherV2_Output{ 786 786 - // // Thread []*UnspeccedGetPostThreadOtherV2_ThreadItem `json:"thread" cborgen:"thread"` 787 787 - // Thread: out, 788 788 - // HasOtherReplies: false, 789 789 - // }) 790 790 - resp := &GetPostThreadOtherV2_Output_WithOtherReplies{ 791 791 - UnspeccedGetPostThreadOtherV2_Output: appbsky.UnspeccedGetPostThreadOtherV2_Output{ 792 792 - Thread: out, 793 793 - }, 794 794 - HasOtherReplies: false, 795 795 - } 796 796 - c.JSON(http.StatusOK, resp) 632 632 + //appbskyunspeccedgetpostthreadv2.HandleGetPostThreadV2(c, sl, cs, BSKYIMAGECDN_URL) 633 633 + // V2V2 still doesnt work. should probably make the handler from scratch to fully use the thread grapher. 634 634 + // also the thread grapher is still sequental. pls fix that 635 635 + //appbskyunspeccedgetpostthreadv2.HandleGetPostThreadV2V2(c, sl, cs, BSKYIMAGECDN_URL) 636 636 + appbskyunspeccedgetpostthreadv2.HandleGetPostThreadV2V3(c, sl, cs, BSKYIMAGECDN_URL) 797 637 }) 798 638 799 639 // weird stuff

+1 -1

readme.md

reviewed

··· 15 15 - `app.bsky.feed.getPosts` (post rendering is incomplete) 16 16 - `app.bsky.feed.getFeed` (post rendering is incomplete) 17 17 - `app.bsky.unspecced.getConfig` (placeholder) 18 18 - - `app.bsky.unspecced.getPostThreadV2` (thread rendering is incomplete) 18 18 + - `app.bsky.unspecced.getPostThreadV2` (mostly working! doesnt use prefered sort, not performant)

+1 -2

shims/lex/app/bsky/actor/defs/profileview.go

reviewed

··· 3 3 import ( 4 4 "context" 5 5 "encoding/json" 6 6 - "log" 7 6 8 7 "github.com/bluesky-social/indigo/api/agnostic" 9 8 appbsky "github.com/bluesky-social/indigo/api/bsky" ··· 91 90 //viewerProfileURI, err_viewerProfileURI := syntax.ParseATURI("at://" + viewerProfileDID + "/app.bsky.actor.profile/self") 92 91 //targetProfileURI, err_targetProfileURI := syntax.ParseATURI("at://" + targetProfileDID + "/app.bsky.actor.profile/self") 93 92 94 94 - log.Println("viewerProfileDID: " + viewerProfileDID + " and targetProfileDID: " + targetProfileDID) 93 93 + //log.Println("viewerProfileDID: " + viewerProfileDID + " and targetProfileDID: " + targetProfileDID) 95 94 96 95 if viewerProfileDID != "" && targetProfileDID != "" { 97 96 blockingBacklink, err_blockingBacklink := constellation.GetBacklinks(ctx, cs, targetProfileDID, "app.bsky.graph.block:subject", []string{viewerProfileDID}, nil, nil)

+525

shims/lex/app/bsky/unspecced/getpostthreadv2/query.go

reviewed

··· 1 1 + package appbskyunspeccedgetpostthreadv2 2 2 + 3 3 + import ( 4 4 + "net/http" 5 5 + "strconv" 6 6 + "sync" 7 7 + 8 8 + "context" 9 9 + "fmt" 10 10 + "log" 11 11 + 12 12 + "github.com/bluesky-social/indigo/atproto/syntax" 13 13 + "github.com/gin-gonic/gin" 14 14 + "tangled.org/whey.party/red-dwarf-server/microcosm" 15 15 + "tangled.org/whey.party/red-dwarf-server/shims/utils" 16 16 + 17 17 + "tangled.org/whey.party/red-dwarf-server/microcosm/constellation" 18 18 + appbskyfeeddefs "tangled.org/whey.party/red-dwarf-server/shims/lex/app/bsky/feed/defs" 19 19 + 20 20 + appbsky "github.com/bluesky-social/indigo/api/bsky" 21 21 + ) 22 22 + 23 23 + func HandleGetPostThreadV2(c *gin.Context, sl *microcosm.MicrocosmClient, cs *microcosm.MicrocosmClient, imgcdn string) { 24 24 + ctx := c.Request.Context() 25 25 + 26 26 + rawdid := c.GetString("user_did") 27 27 + var viewer *utils.DID 28 28 + didval, errdid := utils.NewDID(rawdid) 29 29 + if errdid != nil { 30 30 + viewer = nil 31 31 + } else { 32 32 + viewer = &didval 33 33 + } 34 34 + 35 35 + threadAnchorURIraw := c.Query("anchor") 36 36 + if threadAnchorURIraw == "" { 37 37 + c.JSON(http.StatusBadRequest, gin.H{"error": "Missing feed param"}) 38 38 + return 39 39 + } 40 40 + 41 41 + threadAnchorURI, err := syntax.ParseATURI(threadAnchorURIraw) 42 42 + if err != nil { 43 43 + return 44 44 + } 45 45 + 46 46 + //var thread []*appbsky.UnspeccedGetPostThreadOtherV2_ThreadItem 47 47 + 48 48 + var skeletonposts []string 49 49 + 50 50 + emptystrarray := &[]string{} 51 51 + limit := 100 52 52 + 53 53 + // recurse to the top 54 54 + parentsMap := map[string]*appbsky.FeedDefs_PostView{} 55 55 + current := threadAnchorURI 56 56 + iteration := 0 57 57 + for true { 58 58 + iteration = iteration + 1 59 59 + if iteration > 10 { 60 60 + break 61 61 + } 62 62 + recordaturi, err := syntax.ParseATURI("at://" + current.Authority().String() + "/" + string(current.Collection()) + "/" + string(current.RecordKey())) 63 63 + if err != nil { 64 64 + break 65 65 + } 66 66 + // loop 67 67 + postView, postRecord, err := appbskyfeeddefs.PostView(ctx, recordaturi.String(), sl, cs, imgcdn, viewer, 3) 68 68 + if err != nil { 69 69 + break 70 70 + } 71 71 + if postView == nil { 72 72 + break 73 73 + } 74 74 + //if iteration != 1 { 75 75 + skeletonposts = append([]string{recordaturi.String()}, skeletonposts...) 76 76 + //} 77 77 + parentsMap[recordaturi.String()] = postView 78 78 + if postRecord != nil && postRecord.Reply != nil && postRecord.Reply.Parent != nil && postRecord.Reply.Root != nil { 79 79 + if postRecord.Reply.Parent.Uri == postRecord.Reply.Root.Uri { 80 80 + //break 81 81 + iteration = 9 82 82 + } 83 83 + 84 84 + current, err = syntax.ParseATURI(postRecord.Reply.Parent.Uri) 85 85 + if err != nil { 86 86 + break 87 87 + } 88 88 + } else { 89 89 + log.Println("what huh what hwat") 90 90 + break //whatt 91 91 + } 92 92 + 93 93 + } 94 94 + 95 95 + //skeletonposts = append(skeletonposts, threadAnchorURI.String()) 96 96 + // todo: theres a cursor!!! pagination please! 97 97 + // todo: also i doubt im gonna do proper threadding right now, so make sure to remind me to do it properly some time later thanks 98 98 + //rootReplies, _ := constellation.GetBacklinks(ctx, cs, string(threadAnchorURI), "app.bsky.feed.post:reply.root.uri", *emptystrarray, &limit, nil) 99 99 + parentReplies, _ := constellation.GetBacklinks(ctx, cs, string(threadAnchorURI), "app.bsky.feed.post:reply.parent.uri", *emptystrarray, &limit, nil) 100 100 + 101 101 + for _, rec := range parentReplies.Records { 102 102 + recordaturi, err := syntax.ParseATURI("at://" + rec.Did + "/" + rec.Collection + "/" + rec.Rkey) 103 103 + if err != nil { 104 104 + continue 105 105 + } 106 106 + skeletonposts = append(skeletonposts, recordaturi.String()) 107 107 + } 108 108 + maplen := len(parentsMap) 109 109 + concurrentResults := MapConcurrent( 110 110 + ctx, 111 111 + skeletonposts, 112 112 + 20, 113 113 + func(ctx context.Context, raw string, idx int) (*appbsky.UnspeccedGetPostThreadOtherV2_ThreadItem, error) { 114 114 + var postView *appbsky.FeedDefs_PostView 115 115 + fromParentChain := false 116 116 + postView, ok := parentsMap[raw] 117 117 + if !ok { 118 118 + post, _, err := appbskyfeeddefs.PostView(ctx, raw, sl, cs, imgcdn, viewer, 3) 119 119 + if err != nil { 120 120 + return nil, err 121 121 + } 122 122 + if post == nil { 123 123 + return nil, fmt.Errorf("post not found") 124 124 + } 125 125 + postView = post 126 126 + } else { 127 127 + fromParentChain = true 128 128 + } 129 129 + 130 130 + depth := int64(1) 131 131 + if raw == threadAnchorURI.String() { 132 132 + depth = 0 133 133 + } 134 134 + if fromParentChain { 135 135 + depth = int64(0 - maplen + idx + 1) 136 136 + } 137 137 + 138 138 + return &appbsky.UnspeccedGetPostThreadOtherV2_ThreadItem{ 139 139 + // Depth int64 `json:"depth" cborgen:"depth"` 140 140 + Depth: depth, // todo: placeholder 141 141 + // Uri string `json:"uri" cborgen:"uri"` 142 142 + Uri: raw, 143 143 + // Value *UnspeccedGetPostThreadOtherV2_ThreadItem_Value `json:"value" cborgen:"value"` 144 144 + Value: &appbsky.UnspeccedGetPostThreadOtherV2_ThreadItem_Value{ 145 145 + // UnspeccedDefs_ThreadItemPost *UnspeccedDefs_ThreadItemPost 146 146 + UnspeccedDefs_ThreadItemPost: &appbsky.UnspeccedDefs_ThreadItemPost{ 147 147 + // LexiconTypeID string `json:"$type" cborgen:"$type,const=app.bsky.unspecced.defs#threadItemPost"` 148 148 + LexiconTypeID: "app.bsky.unspecced.defs#threadItemPost", 149 149 + // // hiddenByThreadgate: The threadgate created by the author indicates this post as a reply to be hidden for everyone consuming the thread. 150 150 + // HiddenByThreadgate bool `json:"hiddenByThreadgate" cborgen:"hiddenByThreadgate"` 151 151 + HiddenByThreadgate: false, // todo: placeholder 152 152 + // // moreParents: This post has more parents that were not present in the response. This is just a boolean, without the number of parents. 153 153 + // MoreParents bool `json:"moreParents" cborgen:"moreParents"` 154 154 + MoreParents: false, // todo: placeholder 155 155 + // // moreReplies: This post has more replies that were not present in the response. This is a numeric value, which is best-effort and might not be accurate. 156 156 + // MoreReplies int64 `json:"moreReplies" cborgen:"moreReplies"` 157 157 + MoreReplies: 0, // todo: placeholder 158 158 + // // mutedByViewer: This is by an account muted by the viewer requesting it. 159 159 + // MutedByViewer bool `json:"mutedByViewer" cborgen:"mutedByViewer"` 160 160 + MutedByViewer: false, // todo: placeholder 161 161 + // // opThread: This post is part of a contiguous thread by the OP from the thread root. Many different OP threads can happen in the same thread. 162 162 + // OpThread bool `json:"opThread" cborgen:"opThread"` 163 163 + OpThread: false, // todo: placeholder 164 164 + // Post *FeedDefs_PostView `json:"post" cborgen:"post"` 165 165 + Post: postView, 166 166 + }, 167 167 + }, 168 168 + }, nil 169 169 + }, 170 170 + ) 171 171 + 172 172 + // build final slice 173 173 + out := make([]*appbsky.UnspeccedGetPostThreadOtherV2_ThreadItem, 0, len(concurrentResults)) 174 174 + for _, r := range concurrentResults { 175 175 + if r.Err == nil && r.Value != nil && r.Value.Value != nil && r.Value.Value.UnspeccedDefs_ThreadItemPost != nil && r.Value.Value.UnspeccedDefs_ThreadItemPost.Post != nil { 176 176 + out = append(out, r.Value) 177 177 + } 178 178 + } 179 179 + 180 180 + // c.JSON(http.StatusOK, &appbsky.UnspeccedGetPostThreadOtherV2_Output{ 181 181 + // // Thread []*UnspeccedGetPostThreadOtherV2_ThreadItem `json:"thread" cborgen:"thread"` 182 182 + // Thread: out, 183 183 + // HasOtherReplies: false, 184 184 + // }) 185 185 + resp := &GetPostThreadOtherV2_Output_WithOtherReplies{ 186 186 + UnspeccedGetPostThreadOtherV2_Output: appbsky.UnspeccedGetPostThreadOtherV2_Output{ 187 187 + Thread: out, 188 188 + }, 189 189 + HasOtherReplies: false, 190 190 + } 191 191 + c.JSON(http.StatusOK, resp) 192 192 + } 193 193 + 194 194 + type GetPostThreadOtherV2_Output_WithOtherReplies struct { 195 195 + appbsky.UnspeccedGetPostThreadOtherV2_Output 196 196 + HasOtherReplies bool `json:"hasOtherReplies"` 197 197 + } 198 198 + 199 199 + type AsyncResult[T any] struct { 200 200 + Value T 201 201 + Err error 202 202 + } 203 203 + 204 204 + func MapConcurrent[T any, R any]( 205 205 + ctx context.Context, 206 206 + items []T, 207 207 + concurrencyLimit int, 208 208 + mapper func(context.Context, T, int) (R, error), 209 209 + ) []AsyncResult[R] { 210 210 + if len(items) == 0 { 211 211 + return nil 212 212 + } 213 213 + 214 214 + results := make([]AsyncResult[R], len(items)) 215 215 + var wg sync.WaitGroup 216 216 + 217 217 + sem := make(chan struct{}, concurrencyLimit) 218 218 + 219 219 + for i, item := range items { 220 220 + wg.Add(1) 221 221 + go func(idx int, input T) { 222 222 + defer wg.Done() 223 223 + 224 224 + sem <- struct{}{} 225 225 + defer func() { <-sem }() 226 226 + 227 227 + if ctx.Err() != nil { 228 228 + results[idx] = AsyncResult[R]{Err: ctx.Err()} 229 229 + return 230 230 + } 231 231 + 232 232 + val, err := mapper(ctx, input, idx) 233 233 + results[idx] = AsyncResult[R]{Value: val, Err: err} 234 234 + }(i, item) 235 235 + } 236 236 + 237 237 + wg.Wait() 238 238 + return results 239 239 + } 240 240 + 241 241 + type SkeletonPost struct { 242 242 + post syntax.ATURI 243 243 + depth int 244 244 + } 245 245 + 246 246 + func HandleGetPostThreadV2V3(c *gin.Context, sl *microcosm.MicrocosmClient, cs *microcosm.MicrocosmClient, imgcdn string) { 247 247 + ctx := c.Request.Context() 248 248 + 249 249 + rawdid := c.GetString("user_did") 250 250 + var viewer *utils.DID 251 251 + didval, errdid := utils.NewDID(rawdid) 252 252 + if errdid != nil { 253 253 + viewer = nil 254 254 + } else { 255 255 + viewer = &didval 256 256 + } 257 257 + 258 258 + threadAnchorURIraw := c.Query("anchor") 259 259 + if threadAnchorURIraw == "" { 260 260 + c.JSON(http.StatusBadRequest, gin.H{"error": "Missing feed param"}) 261 261 + return 262 262 + } 263 263 + 264 264 + // "Whether to include parents above the anchor. 265 265 + // bool as string 266 266 + // true as default 267 267 + aboveParam := c.Query("above") // why would you need above = false ? 268 268 + above := true 269 269 + if aboveParam == "false" { 270 270 + above = false 271 271 + } 272 272 + 273 273 + // "How many levels of replies to include below the anchor." 274 274 + // integer as string 275 275 + // default: 6, min: 0, max: 20 276 276 + belowParam := c.Query("below") // bskydefault: 10 277 277 + below, err_below := strconv.ParseInt(belowParam, 10, 64) 278 278 + if err_below != nil { 279 279 + below = 6 280 280 + } else { 281 281 + if below > 20 { 282 282 + below = 20 283 283 + } 284 284 + if below < 0 { 285 285 + below = 0 286 286 + } 287 287 + } 288 288 + 289 289 + // "Maximum of replies to include at each level of the thread, except for the direct replies to the anchor, 290 290 + // which are (NOTE: currently, during unspecced phase) all returned (NOTE: later they might be paginated)." 291 291 + // integer as string 292 292 + // default: 10, min: 0, max: 100 293 293 + branchingFactorParam := c.Query("branchingFactor") // bskydefault: 1 294 294 + branchingFactor, err_branchingFactor := strconv.ParseInt(branchingFactorParam, 10, 64) 295 295 + if err_branchingFactor != nil { 296 296 + branchingFactor = 10 297 297 + } else { 298 298 + if branchingFactor > 100 { 299 299 + branchingFactor = 100 300 300 + } 301 301 + if branchingFactor < 0 { 302 302 + branchingFactor = 0 303 303 + } 304 304 + } 305 305 + 306 306 + // "Sorting for the thread replies" 307 307 + // string (enum) ["newest", "oldest", "top"] 308 308 + // default: "oldest" 309 309 + sortParam := c.Query("sort") // bskydefault: top 310 310 + sort := sortParam 311 311 + if sort != "newest" && sort != "oldest" && sort != "top" { 312 312 + sort = "top" 313 313 + } 314 314 + 315 315 + threadAnchorURI, err := syntax.ParseATURI(threadAnchorURIraw) 316 316 + if err != nil { 317 317 + return 318 318 + } 319 319 + 320 320 + threadGraph, err := ThreadGrapher(ctx, cs, sl, threadAnchorURI) 321 321 + if err != nil { 322 322 + c.JSON(http.StatusBadRequest, gin.H{"error": ("failed to graph the thread: " + err.Error())}) 323 323 + return 324 324 + } 325 325 + var skeletonposts []SkeletonPost 326 326 + 327 327 + // Parent Chain 328 328 + parentChainHeight := 0 329 329 + 330 330 + // root := threadGraph.RootURI 331 331 + if above { 332 332 + current := threadAnchorURI 333 333 + 334 334 + for true { 335 335 + log.Println("[parent threader] current: " + current.Authority().String() + string(current.Collection()) + string(current.RecordKey())) 336 336 + if parentChainHeight > 20 { 337 337 + // root = current 338 338 + break 339 339 + } 340 340 + parent, ok := threadGraph.ParentsMap[current] 341 341 + if !ok { 342 342 + // root = current 343 343 + break 344 344 + } 345 345 + parentChainHeight = parentChainHeight + 1 346 346 + skeletonposts = append(skeletonposts, SkeletonPost{ 347 347 + post: parent, 348 348 + depth: -parentChainHeight, 349 349 + }) 350 350 + current = parent 351 351 + } 352 352 + flipArray(&skeletonposts) 353 353 + } 354 354 + 355 355 + // handled by the recurser 356 356 + // // Anchor Post 357 357 + // skeletonposts = append(skeletonposts, SkeletonPost{ 358 358 + // post: threadAnchorURI, 359 359 + // depth: 0, 360 360 + // }) 361 361 + 362 362 + // Tree Replies (with OP thread priority) 363 363 + // should probably be recursive 364 364 + recursiveHandleV2V3TreeReplies(threadGraph, &skeletonposts, threadAnchorURI, &below, &branchingFactor, &sort, 0, 0) 365 365 + 366 366 + //maplen := len(parentsMap) 367 367 + concurrentResults := MapConcurrent( 368 368 + ctx, 369 369 + skeletonposts, 370 370 + 20, 371 371 + func(ctx context.Context, raw SkeletonPost, idx int) (*appbsky.UnspeccedGetPostThreadOtherV2_ThreadItem, error) { 372 372 + var postView *appbsky.FeedDefs_PostView 373 373 + //fromParentChain := false 374 374 + //postView, ok := parentsMap[raw] 375 375 + //if !ok { 376 376 + post, _, err := appbskyfeeddefs.PostView(ctx, raw.post.String(), sl, cs, imgcdn, viewer, 3) 377 377 + if err != nil { 378 378 + return nil, err 379 379 + } 380 380 + if post == nil { 381 381 + return nil, fmt.Errorf("post not found") 382 382 + } 383 383 + postView = post 384 384 + //} else { 385 385 + // fromParentChain = true 386 386 + //} 387 387 + 388 388 + depth := int64(1) 389 389 + // if raw == threadAnchorURI.String() { 390 390 + // depth = 0 391 391 + // } 392 392 + // if fromParentChain { 393 393 + // depth = int64(0 - parentChainHeight + idx + 1) 394 394 + // } 395 395 + depth = int64(raw.depth) 396 396 + 397 397 + return &appbsky.UnspeccedGetPostThreadOtherV2_ThreadItem{ 398 398 + // Depth int64 `json:"depth" cborgen:"depth"` 399 399 + Depth: depth, // todo: placeholder 400 400 + // Uri string `json:"uri" cborgen:"uri"` 401 401 + Uri: raw.post.String(), 402 402 + // Value *UnspeccedGetPostThreadOtherV2_ThreadItem_Value `json:"value" cborgen:"value"` 403 403 + Value: &appbsky.UnspeccedGetPostThreadOtherV2_ThreadItem_Value{ 404 404 + // UnspeccedDefs_ThreadItemPost *UnspeccedDefs_ThreadItemPost 405 405 + UnspeccedDefs_ThreadItemPost: &appbsky.UnspeccedDefs_ThreadItemPost{ 406 406 + // LexiconTypeID string `json:"$type" cborgen:"$type,const=app.bsky.unspecced.defs#threadItemPost"` 407 407 + LexiconTypeID: "app.bsky.unspecced.defs#threadItemPost", 408 408 + // // hiddenByThreadgate: The threadgate created by the author indicates this post as a reply to be hidden for everyone consuming the thread. 409 409 + // HiddenByThreadgate bool `json:"hiddenByThreadgate" cborgen:"hiddenByThreadgate"` 410 410 + HiddenByThreadgate: false, // todo: placeholder 411 411 + // // moreParents: This post has more parents that were not present in the response. This is just a boolean, without the number of parents. 412 412 + // MoreParents bool `json:"moreParents" cborgen:"moreParents"` 413 413 + MoreParents: false, // todo: placeholder 414 414 + // // moreReplies: This post has more replies that were not present in the response. This is a numeric value, which is best-effort and might not be accurate. 415 415 + // MoreReplies int64 `json:"moreReplies" cborgen:"moreReplies"` 416 416 + MoreReplies: 0, // todo: placeholder 417 417 + // // mutedByViewer: This is by an account muted by the viewer requesting it. 418 418 + // MutedByViewer bool `json:"mutedByViewer" cborgen:"mutedByViewer"` 419 419 + MutedByViewer: false, // todo: placeholder 420 420 + // // opThread: This post is part of a contiguous thread by the OP from the thread root. Many different OP threads can happen in the same thread. 421 421 + // OpThread bool `json:"opThread" cborgen:"opThread"` 422 422 + OpThread: false, // todo: placeholder 423 423 + // Post *FeedDefs_PostView `json:"post" cborgen:"post"` 424 424 + Post: postView, 425 425 + }, 426 426 + }, 427 427 + }, nil 428 428 + }, 429 429 + ) 430 430 + 431 431 + // build final slice 432 432 + out := make([]*appbsky.UnspeccedGetPostThreadOtherV2_ThreadItem, 0, len(concurrentResults)) 433 433 + for _, r := range concurrentResults { 434 434 + if r.Err == nil && r.Value != nil && r.Value.Value != nil && r.Value.Value.UnspeccedDefs_ThreadItemPost != nil && r.Value.Value.UnspeccedDefs_ThreadItemPost.Post != nil { 435 435 + out = append(out, r.Value) 436 436 + } 437 437 + } 438 438 + 439 439 + // c.JSON(http.StatusOK, &appbsky.UnspeccedGetPostThreadOtherV2_Output{ 440 440 + // // Thread []*UnspeccedGetPostThreadOtherV2_ThreadItem `json:"thread" cborgen:"thread"` 441 441 + // Thread: out, 442 442 + // HasOtherReplies: false, 443 443 + // }) 444 444 + resp := &GetPostThreadOtherV2_Output_WithOtherReplies{ 445 445 + UnspeccedGetPostThreadOtherV2_Output: appbsky.UnspeccedGetPostThreadOtherV2_Output{ 446 446 + Thread: out, 447 447 + }, 448 448 + HasOtherReplies: false, 449 449 + } 450 450 + c.JSON(http.StatusOK, resp) 451 451 + } 452 452 + 453 453 + func flipArray[T any](s *[]T) { 454 454 + for i, j := 0, len(*s)-1; i < j; i, j = i+1, j-1 { 455 455 + (*s)[i], (*s)[j] = (*s)[j], (*s)[i] 456 456 + } 457 457 + } 458 458 + 459 459 + func recursiveHandleV2V3TreeReplies(threadGraph *ThreadGraph, skeletonposts *[]SkeletonPost, current syntax.ATURI, below *int64, branchingFactor *int64, sort *string, verticalPos int64, horizontalPos int64) { 460 460 + log.Println("[V3 Recurse] at y: " + fmt.Sprint(verticalPos) + ", x: " + fmt.Sprint(horizontalPos)) 461 461 + // breakings 462 462 + if below != nil && verticalPos > *below { 463 463 + log.Println("[V3 Recurse] exit by too low") 464 464 + return 465 465 + } 466 466 + if branchingFactor != nil && horizontalPos > *branchingFactor && verticalPos > 1 { 467 467 + log.Println("[V3 Recurse] exit by too wide; branchingFactor: " + fmt.Sprint(*branchingFactor) + "; horizontalPos: " + fmt.Sprint(horizontalPos) + "; verticalPos: " + fmt.Sprint(verticalPos)) 468 468 + return 469 469 + } 470 470 + 471 471 + // the things to do if not recurse 472 472 + if skeletonposts != nil { 473 473 + *skeletonposts = append(*skeletonposts, SkeletonPost{ 474 474 + post: current, 475 475 + depth: int(verticalPos), 476 476 + }) 477 477 + } else { 478 478 + log.Println("[V3 Recurse] exit by no skeleton posts") 479 479 + return 480 480 + } 481 481 + 482 482 + repliesAtThisPosition, ok := threadGraph.ChildrenMap[current] 483 483 + if !ok { 484 484 + log.Println("[V3 Recurse] exit by no replies") 485 485 + return 486 486 + } 487 487 + 488 488 + // recurse 489 489 + op := threadGraph.RootURI.Authority() 490 490 + 491 491 + // We need a modifyable copy of the slice because we are about to reorder it 492 492 + // and we don't want to mess up the original map in case it's used elsewhere. 493 493 + sortedReplies := make([]syntax.ATURI, len(repliesAtThisPosition)) 494 494 + copy(sortedReplies, repliesAtThisPosition) 495 495 + 496 496 + // 1. Find the "best" OP reply (lexicographically smallest rkey usually means oldest) 497 497 + bestOpIndex := -1 498 498 + var bestOpURI syntax.ATURI 499 499 + 500 500 + for i, replyURI := range sortedReplies { 501 501 + if replyURI.Authority() == op { 502 502 + // If this is the first OP reply we found, or if it's lexicographically smaller (older) than the previous best 503 503 + if bestOpIndex == -1 || replyURI.String() < bestOpURI.String() { 504 504 + bestOpIndex = i 505 505 + bestOpURI = replyURI 506 506 + } 507 507 + } 508 508 + } 509 509 + 510 510 + // 2. If we found an OP reply, move it to index 0 511 511 + if bestOpIndex > 0 { // If it's already 0, no need to move 512 512 + // Remove the best OP reply from its current spot 513 513 + // (Go slice trick: append everything before it + everything after it) 514 514 + withoutBest := append(sortedReplies[:bestOpIndex], sortedReplies[bestOpIndex+1:]...) 515 515 + 516 516 + // Prepend it to the front 517 517 + sortedReplies = append([]syntax.ATURI{bestOpURI}, withoutBest...) 518 518 + } 519 519 + 520 520 + for idx, reply := range sortedReplies { 521 521 + recursiveHandleV2V3TreeReplies(threadGraph, skeletonposts, reply, below, branchingFactor, sort, verticalPos+1, int64(idx+1)) 522 522 + } 523 523 + 524 524 + // going up 525 525 + }

+145

shims/lex/app/bsky/unspecced/getpostthreadv2/threadgrapher.go

reviewed

··· 1 1 + package appbskyunspeccedgetpostthreadv2 2 2 + 3 3 + import ( 4 4 + "context" 5 5 + "encoding/json" 6 6 + "fmt" 7 7 + "log" 8 8 + "sync" 9 9 + 10 10 + "github.com/bluesky-social/indigo/api/agnostic" 11 11 + //comatproto "github.com/bluesky-social/indigo/api/atproto" 12 12 + appbsky "github.com/bluesky-social/indigo/api/bsky" 13 13 + "github.com/bluesky-social/indigo/atproto/syntax" 14 14 + 15 15 + "tangled.org/whey.party/red-dwarf-server/microcosm" 16 16 + "tangled.org/whey.party/red-dwarf-server/microcosm/constellation" 17 17 + ) 18 18 + 19 19 + type ThreadGraph struct { 20 20 + mu sync.RWMutex // Mutex required for thread-safe updates (Firehose) 21 21 + 22 22 + RootURI syntax.ATURI 23 23 + AnchorURI syntax.ATURI 24 24 + 25 25 + ParentsMap map[syntax.ATURI]syntax.ATURI 26 26 + 27 27 + ChildrenMap map[syntax.ATURI][]syntax.ATURI 28 28 + } 29 29 + 30 30 + func NewThreadGraph() *ThreadGraph { 31 31 + return &ThreadGraph{ 32 32 + ParentsMap: make(map[syntax.ATURI]syntax.ATURI), 33 33 + ChildrenMap: make(map[syntax.ATURI][]syntax.ATURI), 34 34 + } 35 35 + } 36 36 + 37 37 + func ThreadGrapher(ctx context.Context, cs *microcosm.MicrocosmClient, sl *microcosm.MicrocosmClient, threadAnchorURI syntax.ATURI) (*ThreadGraph, error) { 38 38 + tg := NewThreadGraph() 39 39 + tg.AnchorURI = threadAnchorURI 40 40 + 41 41 + anchorPostRecordResponse, err := agnostic.RepoGetRecord(ctx, sl, "", "app.bsky.feed.post", threadAnchorURI.Authority().String(), string(threadAnchorURI.RecordKey())) 42 42 + if err != nil { 43 43 + log.Println("[ThreadGrapher] exit by anchor post resolve failure") 44 44 + return nil, fmt.Errorf("failed to fetch anchor: %w", err) 45 45 + } 46 46 + 47 47 + var anchorPost appbsky.FeedPost 48 48 + if err := json.Unmarshal(*anchorPostRecordResponse.Value, &anchorPost); err != nil { 49 49 + log.Println("[ThreadGrapher] exit by no json") 50 50 + return nil, fmt.Errorf("error: Failed to parse post record JSON") 51 51 + } 52 52 + 53 53 + var rootURI syntax.ATURI 54 54 + 55 55 + if anchorPost.Reply != nil && anchorPost.Reply.Root != nil { 56 56 + rURI, err := syntax.ParseATURI(anchorPost.Reply.Root.Uri) 57 57 + if err != nil { 58 58 + log.Println("[ThreadGrapher] exit by invalid root uri") 59 59 + return nil, fmt.Errorf("invalid root uri in record: %w", err) 60 60 + } 61 61 + rootURI = rURI 62 62 + 63 63 + // todo: fiine we wont fetch the root post, but we still need to mark it as deleted somehow 64 64 + // rootPostRecordResponse, err := agnostic.RepoGetRecord(ctx, sl, "", "app.bsky.feed.post", rootURI.Authority().String(), string(rootURI.RecordKey())) 65 65 + // if err != nil { 66 66 + // log.Println("[ThreadGrapher] exit by cant fetch root") 67 67 + // return nil, fmt.Errorf("failed to fetch root: %w", err) 68 68 + // } 69 69 + 70 70 + // var rootPost appbsky.FeedPost 71 71 + // if err := json.Unmarshal(*rootPostRecordResponse.Value, &rootPost); err != nil { 72 72 + // log.Println("[ThreadGrapher] exit by cant parse root json") 73 73 + // return nil, fmt.Errorf("error: Failed to parse post record JSON") 74 74 + // } 75 75 + // if err == nil { 76 76 + // tg.AddNode(rootURI, &rootPost) 77 77 + // } 78 78 + } else { 79 79 + rootURI = threadAnchorURI 80 80 + } 81 81 + tg.RootURI = rootURI 82 82 + 83 83 + emptystrarray := &[]string{} 84 84 + var allRepliesATURI []syntax.ATURI 85 85 + limit := 100 86 86 + var cursor *string 87 87 + shouldContinue := true 88 88 + for shouldContinue { 89 89 + results, err := constellation.GetBacklinks(ctx, cs, rootURI.String(), "app.bsky.feed.post:reply.root.uri", *emptystrarray, &limit, cursor) 90 90 + if err != nil { 91 91 + log.Println("[ThreadGrapher] [root graphing] exit by backlink failure") 92 92 + return nil, fmt.Errorf("failed to get backlinks: %w", err) 93 93 + } 94 94 + if results.Records != nil { 95 95 + for _, record := range results.Records { 96 96 + aturi, err := syntax.ParseATURI("at://" + record.Did + "/" + record.Collection + "/" + record.Rkey) 97 97 + if err == nil { 98 98 + allRepliesATURI = append(allRepliesATURI, aturi) 99 99 + } 100 100 + } 101 101 + } 102 102 + if results.Cursor != nil { 103 103 + cursor = results.Cursor 104 104 + } else { 105 105 + shouldContinue = false 106 106 + } 107 107 + } 108 108 + 109 109 + processingQueue := append([]syntax.ATURI{rootURI}, allRepliesATURI...) 110 110 + for _, aturi := range processingQueue { 111 111 + //tg.Nodes = append(tg.Nodes, aturi) 112 112 + // graphinger 113 113 + emptystrarray := &[]string{} 114 114 + var localRepliesATURI []syntax.ATURI 115 115 + limit := 100 116 116 + var cursor *string 117 117 + shouldContinue := true 118 118 + for shouldContinue { 119 119 + results, err := constellation.GetBacklinks(ctx, cs, aturi.String(), "app.bsky.feed.post:reply.parent.uri", *emptystrarray, &limit, cursor) 120 120 + if err != nil { 121 121 + log.Println("[ThreadGrapher] [parent graphing] exit by no replies") 122 122 + return nil, fmt.Errorf("failed to get backlinks: %w", err) 123 123 + } 124 124 + if results.Records != nil { 125 125 + for _, record := range results.Records { 126 126 + aturi, err := syntax.ParseATURI("at://" + record.Did + "/" + record.Collection + "/" + record.Rkey) 127 127 + if err == nil { 128 128 + localRepliesATURI = append(localRepliesATURI, aturi) 129 129 + } 130 130 + } 131 131 + } 132 132 + if results.Cursor != nil { 133 133 + cursor = results.Cursor 134 134 + } else { 135 135 + shouldContinue = false 136 136 + } 137 137 + } 138 138 + for _, reply := range localRepliesATURI { 139 139 + tg.ParentsMap[reply] = aturi 140 140 + tg.ChildrenMap[aturi] = append(tg.ChildrenMap[aturi], reply) 141 141 + } 142 142 + } 143 143 + 144 144 + return tg, nil 145 145 + }