From 1257864da73c213cad996d40848960766cd1dab8 Mon Sep 17 00:00:00 2001 From: scanash00 Date: Tue, 27 Jan 2026 20:05:30 -0900 Subject: [PATCH] refactor some things in backend and implement microcosm --- backend/go.mod | 14 + backend/go.sum | 29 ++ backend/internal/api/handler.go | 106 +++++++ backend/internal/api/hydration.go | 140 +++++----- backend/internal/constellation/client.go | 341 +++++++++++++++++++++++ backend/internal/crypto/cid.go | 228 +++++++++++++++ backend/internal/firehose/ingester.go | 47 +++- backend/internal/oauth/handler.go | 6 + backend/internal/slingshot/client.go | 179 ++++++++++++ backend/internal/sync/service.go | 11 + backend/internal/xrpc/client.go | 14 +- backend/internal/xrpc/utils.go | 107 +++++++ 12 files changed, 1144 insertions(+), 78 deletions(-) create mode 100644 backend/internal/constellation/client.go create mode 100644 backend/internal/crypto/cid.go create mode 100644 backend/internal/slingshot/client.go diff --git a/backend/go.mod b/backend/go.mod index e81fec5..a12fd81 100644 --- a/backend/go.mod +++ b/backend/go.mod @@ -15,8 +15,22 @@ require ( require ( github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect + github.com/fxamacker/cbor/v2 v2.9.0 // indirect + github.com/ipfs/go-cid v0.6.0 // indirect + github.com/klauspost/cpuid/v2 v2.0.9 // indirect + github.com/minio/sha256-simd v1.0.0 // indirect + github.com/mr-tron/base58 v1.2.0 // indirect + github.com/multiformats/go-base32 v0.0.3 // indirect + github.com/multiformats/go-base36 v0.1.0 // indirect + github.com/multiformats/go-multibase v0.2.0 // indirect + github.com/multiformats/go-multihash v0.2.3 // indirect + github.com/multiformats/go-varint v0.1.0 // indirect github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect + github.com/spaolacci/murmur3 v1.1.0 // indirect github.com/stretchr/testify v1.10.0 // indirect + github.com/x448/float16 v0.8.4 // indirect golang.org/x/crypto v0.35.0 // indirect + golang.org/x/sys v0.30.0 // indirect golang.org/x/text v0.32.0 // indirect + lukechampine.com/blake3 v1.1.6 // indirect ) diff --git a/backend/go.sum b/backend/go.sum index f6bbbc9..ff92a28 100644 --- a/backend/go.sum +++ b/backend/go.sum @@ -1,5 +1,7 @@ github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM= github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/fxamacker/cbor/v2 v2.9.0 h1:NpKPmjDBgUfBms6tr6JZkTHtfFGcMKsw3eGcmD/sapM= +github.com/fxamacker/cbor/v2 v2.9.0/go.mod h1:vM4b+DJCtHn+zz7h3FFp/hDAI9WNWCsZj23V5ytsSxQ= github.com/go-chi/chi/v5 v5.1.0 h1:acVI1TYaD+hhedDJ3r54HyA6sExp3HfXq7QWEEY/xMw= github.com/go-chi/chi/v5 v5.1.0/go.mod h1:DslCQbL2OYiznFReuXYUmQ2hGd1aDpCnlMNITLSKoi8= github.com/go-chi/cors v1.2.1 h1:xEC8UT3Rlp2QuWNEr4Fs/c2EAGVKBwy/1vHx3bppil4= @@ -10,21 +12,48 @@ github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI= github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg= github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= +github.com/ipfs/go-cid v0.6.0 h1:DlOReBV1xhHBhhfy/gBNNTSyfOM6rLiIx9J7A4DGf30= +github.com/ipfs/go-cid v0.6.0/go.mod h1:NC4kS1LZjzfhK40UGmpXv5/qD2kcMzACYJNntCUiDhQ= github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0= github.com/joho/godotenv v1.5.1/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwAbqwq4= +github.com/klauspost/cpuid/v2 v2.0.4/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg= +github.com/klauspost/cpuid/v2 v2.0.9 h1:lgaqFMSdTdQYdZ04uHyN2d/eKdOMyi2YLSvlQIBFYa4= +github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg= github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw= github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o= github.com/mattn/go-sqlite3 v1.14.22 h1:2gZY6PC6kBnID23Tichd1K+Z0oS6nE/XwU+Vz/5o4kU= github.com/mattn/go-sqlite3 v1.14.22/go.mod h1:Uh1q+B4BYcTPb+yiD3kU8Ct7aC0hY9fxUwlHK0RXw+Y= +github.com/minio/sha256-simd v1.0.0 h1:v1ta+49hkWZyvaKwrQB8elexRqm6Y0aMLjCNsrYxo6g= +github.com/minio/sha256-simd v1.0.0/go.mod h1:OuYzVNI5vcoYIAmbIvHPl3N3jUzVedXbKy5RFepssQM= +github.com/mr-tron/base58 v1.2.0 h1:T/HDJBh4ZCPbU39/+c3rRvE0uKBQlU27+QI8LJ4t64o= +github.com/mr-tron/base58 v1.2.0/go.mod h1:BinMc/sQntlIE1frQmRFPUoPA1Zkr8VRgBdjWI2mNwc= +github.com/multiformats/go-base32 v0.0.3 h1:tw5+NhuwaOjJCC5Pp82QuXbrmLzWg7uxlMFp8Nq/kkI= +github.com/multiformats/go-base32 v0.0.3/go.mod h1:pLiuGC8y0QR3Ue4Zug5UzK9LjgbkL8NSQj0zQ5Nz/AA= +github.com/multiformats/go-base36 v0.1.0 h1:JR6TyF7JjGd3m6FbLU2cOxhC0Li8z8dLNGQ89tUg4F4= +github.com/multiformats/go-base36 v0.1.0/go.mod h1:kFGE83c6s80PklsHO9sRn2NCoffoRdUUOENyW/Vv6sM= +github.com/multiformats/go-multibase v0.2.0 h1:isdYCVLvksgWlMW9OZRYJEa9pZETFivncJHmHnnd87g= +github.com/multiformats/go-multibase v0.2.0/go.mod h1:bFBZX4lKCA/2lyOFSAoKH5SS6oPyjtnzK/XTFDPkNuk= +github.com/multiformats/go-multihash v0.2.3 h1:7Lyc8XfX/IY2jWb/gI7JP+o7JEq9hOa7BFvVU9RSh+U= +github.com/multiformats/go-multihash v0.2.3/go.mod h1:dXgKXCXjBzdscBLk9JkjINiEsCKRVch90MdaGiKsvSM= +github.com/multiformats/go-varint v0.1.0 h1:i2wqFp4sdl3IcIxfAonHQV9qU5OsZ4Ts9IOoETFs5dI= +github.com/multiformats/go-varint v0.1.0/go.mod h1:5KVAVXegtfmNQQm/lCY+ATvDzvJJhSkUlGQV9wgObdI= github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U= github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/spaolacci/murmur3 v1.1.0 h1:7c1g84S4BPRrfL5Xrdp6fOJ206sU9y293DDHaoy0bLI= +github.com/spaolacci/murmur3 v1.1.0/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA= github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA= github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= +github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM= +github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg= golang.org/x/crypto v0.35.0 h1:b15kiHdrGCHrP6LvwaQ3c03kgNhhiMgvlhxHQhmg2Xs= golang.org/x/crypto v0.35.0/go.mod h1:dy7dXNW32cAb/6/PRuTNsix8T+vJAqvuIy5Bli/x0YQ= golang.org/x/image v0.34.0 h1:33gCkyw9hmwbZJeZkct8XyR11yH889EQt/QH4VmXMn8= golang.org/x/image v0.34.0/go.mod h1:2RNFBZRB+vnwwFil8GkMdRvrJOFd1AzdZI6vOY+eJVU= +golang.org/x/sys v0.30.0 h1:QjkSwP/36a20jFYWkSue1YwXzLmsV5Gfq7Eiy72C1uc= +golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= golang.org/x/text v0.32.0 h1:ZD01bjUt1FQ9WJ0ClOL5vxgxOI/sVCNgX1YtKwcY0mU= golang.org/x/text v0.32.0/go.mod h1:o/rUWzghvpD5TXrTIBuJU77MTaN0ljMWE47kxGJQ7jY= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +lukechampine.com/blake3 v1.1.6 h1:H3cROdztr7RCfoaTpGZFQsrqvweFLrqS73j7L7cmR5c= +lukechampine.com/blake3 v1.1.6/go.mod h1:tkKEOtDkNtklkXtLNEOGNq5tcV90tJiA1vAA12R78LA= diff --git a/backend/internal/api/handler.go b/backend/internal/api/handler.go index 728c301..535f10d 100644 --- a/backend/internal/api/handler.go +++ b/backend/internal/api/handler.go @@ -68,6 +68,7 @@ func (h *Handler) RegisterRoutes(r chi.Router) { r.Post("/sync", h.SyncAll) r.Get("/targets", h.GetByTarget) + r.Get("/discover", h.DiscoverForURL) r.Get("/users/{did}/annotations", h.GetUserAnnotations) r.Get("/users/{did}/highlights", h.GetUserHighlights) @@ -631,6 +632,111 @@ func (h *Handler) GetByTarget(w http.ResponseWriter, r *http.Request) { }) } +func (h *Handler) DiscoverForURL(w http.ResponseWriter, r *http.Request) { + source := r.URL.Query().Get("source") + if source == "" { + source = r.URL.Query().Get("url") + } + if source == "" { + http.Error(w, "source or url parameter required", http.StatusBadRequest) + return + } + + ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second) + defer cancel() + + annotations, highlights, bookmarks, err := ConstellationClient.GetAllItemsForURL(ctx, source) + if err != nil { + log.Printf("Constellation discover error, falling back to local: %v", err) + h.GetByTarget(w, r) + return + } + + var annotationURIs, highlightURIs, bookmarkURIs []string + seenURIs := make(map[string]bool) + + for _, link := range annotations { + if !seenURIs[link.URI] { + annotationURIs = append(annotationURIs, link.URI) + seenURIs[link.URI] = true + } + } + for _, link := range highlights { + if !seenURIs[link.URI] { + highlightURIs = append(highlightURIs, link.URI) + seenURIs[link.URI] = true + } + } + for _, link := range bookmarks { + if !seenURIs[link.URI] { + bookmarkURIs = append(bookmarkURIs, link.URI) + seenURIs[link.URI] = true + } + } + + localAnnotations, _ := h.db.GetAnnotationsByURIs(annotationURIs) + localHighlights, _ := h.db.GetHighlightsByURIs(highlightURIs) + localBookmarks, _ := h.db.GetBookmarksByURIs(bookmarkURIs) + + urlHash := db.HashURL(source) + dbAnnotations, _ := h.db.GetAnnotationsByTargetHash(urlHash, 100, 0) + dbHighlights, _ := h.db.GetHighlightsByTargetHash(urlHash, 100, 0) + dbBookmarks, _ := h.db.GetBookmarksByTargetHash(urlHash, 100, 0) + + annoMap := make(map[string]db.Annotation) + for _, a := range localAnnotations { + annoMap[a.URI] = a + } + for _, a := range dbAnnotations { + annoMap[a.URI] = a + } + + highMap := make(map[string]db.Highlight) + for _, h := range localHighlights { + highMap[h.URI] = h + } + for _, h := range dbHighlights { + highMap[h.URI] = h + } + + bookMap := make(map[string]db.Bookmark) + for _, b := range localBookmarks { + bookMap[b.URI] = b + } + for _, b := range dbBookmarks { + bookMap[b.URI] = b + } + + var mergedAnnotations []db.Annotation + for _, a := range annoMap { + mergedAnnotations = append(mergedAnnotations, a) + } + var mergedHighlights []db.Highlight + for _, h := range highMap { + mergedHighlights = append(mergedHighlights, h) + } + var mergedBookmarks []db.Bookmark + for _, b := range bookMap { + mergedBookmarks = append(mergedBookmarks, b) + } + + viewerDID := h.getViewerDID(r) + enrichedAnnotations, _ := hydrateAnnotations(h.db, mergedAnnotations, viewerDID) + enrichedHighlights, _ := hydrateHighlights(h.db, mergedHighlights, viewerDID) + enrichedBookmarks, _ := hydrateBookmarks(h.db, mergedBookmarks, viewerDID) + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]interface{}{ + "@context": "http://www.w3.org/ns/anno.jsonld", + "source": source, + "sourceHash": urlHash, + "annotations": enrichedAnnotations, + "highlights": enrichedHighlights, + "bookmarks": enrichedBookmarks, + "networkDiscovered": len(annotations) + len(highlights) + len(bookmarks), + }) +} + func (h *Handler) GetHighlights(w http.ResponseWriter, r *http.Request) { did := r.URL.Query().Get("creator") tag := r.URL.Query().Get("tag") diff --git a/backend/internal/api/hydration.go b/backend/internal/api/hydration.go index ae9aa63..f5e9b73 100644 --- a/backend/internal/api/hydration.go +++ b/backend/internal/api/hydration.go @@ -1,6 +1,7 @@ package api import ( + "context" "encoding/json" "fmt" "log" @@ -10,13 +11,19 @@ import ( "sync" "time" + "margin.at/internal/constellation" "margin.at/internal/db" ) var ( - Cache ProfileCache = NewInMemoryCache(5 * time.Minute) + Cache ProfileCache = NewInMemoryCache(5 * time.Minute) + ConstellationClient *constellation.Client = constellation.NewClient() // Enabled by default ) +func init() { + log.Printf("Constellation client initialized: %s", constellation.DefaultBaseURL) +} + type Author struct { DID string `json:"did"` Handle string `json:"handle"` @@ -145,23 +152,16 @@ type APINotification struct { ReadAt *time.Time `json:"readAt,omitempty"` } -func hydrateAnnotations(database *db.DB, annotations []db.Annotation, viewerDID string) ([]APIAnnotation, error) { - if len(annotations) == 0 { - return []APIAnnotation{}, nil - } - - profiles := fetchProfilesForDIDs(collectDIDs(annotations, func(a db.Annotation) string { return a.AuthorDID })) +func fetchCounts(ctx context.Context, database *db.DB, uris []string, viewerDID string) (likeCounts, replyCounts map[string]int, viewerLikes map[string]bool) { + likeCounts = make(map[string]int) + replyCounts = make(map[string]int) + viewerLikes = make(map[string]bool) - var likeCounts map[string]int - var replyCounts map[string]int - var viewerLikes map[string]bool + if len(uris) == 0 { + return + } if database != nil { - uris := make([]string, len(annotations)) - for i, a := range annotations { - uris[i] = a.URI - } - likeCounts, _ = database.GetLikeCounts(uris) replyCounts, _ = database.GetReplyCounts(uris) if viewerDID != "" { @@ -169,6 +169,42 @@ func hydrateAnnotations(database *db.DB, annotations []db.Annotation, viewerDID } } + if ConstellationClient != nil && len(uris) <= 5 { + constellationCounts, err := ConstellationClient.GetCountsBatch(ctx, uris) + if err != nil { + log.Printf("Constellation fetch error (non-fatal): %v", err) + return + } + + for uri, counts := range constellationCounts { + if counts.LikeCount > likeCounts[uri] { + likeCounts[uri] = counts.LikeCount + } + if counts.ReplyCount > replyCounts[uri] { + replyCounts[uri] = counts.ReplyCount + } + } + } + + return +} + +func hydrateAnnotations(database *db.DB, annotations []db.Annotation, viewerDID string) ([]APIAnnotation, error) { + if len(annotations) == 0 { + return []APIAnnotation{}, nil + } + + profiles := fetchProfilesForDIDs(collectDIDs(annotations, func(a db.Annotation) string { return a.AuthorDID })) + + uris := make([]string, len(annotations)) + for i, a := range annotations { + uris[i] = a.URI + } + + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + likeCounts, replyCounts, viewerLikes := fetchCounts(ctx, database, uris, viewerDID) + result := make([]APIAnnotation, len(annotations)) for i, a := range annotations { var body *APIBody @@ -228,12 +264,10 @@ func hydrateAnnotations(database *db.DB, annotations []db.Annotation, viewerDID IndexedAt: a.IndexedAt, } - if database != nil { - result[i].LikeCount = likeCounts[a.URI] - result[i].ReplyCount = replyCounts[a.URI] - if viewerLikes != nil && viewerLikes[a.URI] { - result[i].ViewerHasLiked = true - } + result[i].LikeCount = likeCounts[a.URI] + result[i].ReplyCount = replyCounts[a.URI] + if viewerLikes != nil && viewerLikes[a.URI] { + result[i].ViewerHasLiked = true } } @@ -247,23 +281,15 @@ func hydrateHighlights(database *db.DB, highlights []db.Highlight, viewerDID str profiles := fetchProfilesForDIDs(collectDIDs(highlights, func(h db.Highlight) string { return h.AuthorDID })) - var likeCounts map[string]int - var replyCounts map[string]int - var viewerLikes map[string]bool - - if database != nil { - uris := make([]string, len(highlights)) - for i, h := range highlights { - uris[i] = h.URI - } - - likeCounts, _ = database.GetLikeCounts(uris) - replyCounts, _ = database.GetReplyCounts(uris) - if viewerDID != "" { - viewerLikes, _ = database.GetViewerLikes(viewerDID, uris) - } + uris := make([]string, len(highlights)) + for i, h := range highlights { + uris[i] = h.URI } + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + likeCounts, replyCounts, viewerLikes := fetchCounts(ctx, database, uris, viewerDID) + result := make([]APIHighlight, len(highlights)) for i, h := range highlights { var selector *APISelector @@ -307,12 +333,10 @@ func hydrateHighlights(database *db.DB, highlights []db.Highlight, viewerDID str CID: cid, } - if database != nil { - result[i].LikeCount = likeCounts[h.URI] - result[i].ReplyCount = replyCounts[h.URI] - if viewerLikes != nil && viewerLikes[h.URI] { - result[i].ViewerHasLiked = true - } + result[i].LikeCount = likeCounts[h.URI] + result[i].ReplyCount = replyCounts[h.URI] + if viewerLikes != nil && viewerLikes[h.URI] { + result[i].ViewerHasLiked = true } } @@ -326,23 +350,15 @@ func hydrateBookmarks(database *db.DB, bookmarks []db.Bookmark, viewerDID string profiles := fetchProfilesForDIDs(collectDIDs(bookmarks, func(b db.Bookmark) string { return b.AuthorDID })) - var likeCounts map[string]int - var replyCounts map[string]int - var viewerLikes map[string]bool - - if database != nil { - uris := make([]string, len(bookmarks)) - for i, b := range bookmarks { - uris[i] = b.URI - } - - likeCounts, _ = database.GetLikeCounts(uris) - replyCounts, _ = database.GetReplyCounts(uris) - if viewerDID != "" { - viewerLikes, _ = database.GetViewerLikes(viewerDID, uris) - } + uris := make([]string, len(bookmarks)) + for i, b := range bookmarks { + uris[i] = b.URI } + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + likeCounts, replyCounts, viewerLikes := fetchCounts(ctx, database, uris, viewerDID) + result := make([]APIBookmark, len(bookmarks)) for i, b := range bookmarks { var tags []string @@ -376,12 +392,10 @@ func hydrateBookmarks(database *db.DB, bookmarks []db.Bookmark, viewerDID string CreatedAt: b.CreatedAt, CID: cid, } - if database != nil { - result[i].LikeCount = likeCounts[b.URI] - result[i].ReplyCount = replyCounts[b.URI] - if viewerLikes != nil && viewerLikes[b.URI] { - result[i].ViewerHasLiked = true - } + result[i].LikeCount = likeCounts[b.URI] + result[i].ReplyCount = replyCounts[b.URI] + if viewerLikes != nil && viewerLikes[b.URI] { + result[i].ViewerHasLiked = true } } diff --git a/backend/internal/constellation/client.go b/backend/internal/constellation/client.go new file mode 100644 index 0000000..85b3a34 --- /dev/null +++ b/backend/internal/constellation/client.go @@ -0,0 +1,341 @@ +package constellation + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/url" + "sync" + "time" +) + +const ( + DefaultBaseURL = "https://constellation.microcosm.blue" + DefaultTimeout = 5 * time.Second + UserAgent = "Margin (margin.at)" +) + +type Client struct { + baseURL string + httpClient *http.Client +} + +func NewClient() *Client { + return &Client{ + baseURL: DefaultBaseURL, + httpClient: &http.Client{ + Timeout: DefaultTimeout, + }, + } +} + +func NewClientWithURL(baseURL string) *Client { + return &Client{ + baseURL: baseURL, + httpClient: &http.Client{ + Timeout: DefaultTimeout, + }, + } +} + +type CountResponse struct { + Total int `json:"total"` +} + +type Link struct { + URI string `json:"uri"` + Collection string `json:"collection"` + DID string `json:"did"` + Path string `json:"path"` +} + +type LinksResponse struct { + Links []Link `json:"links"` + Cursor string `json:"cursor,omitempty"` +} + +func (c *Client) GetLikeCount(ctx context.Context, subjectURI string) (int, error) { + params := url.Values{} + params.Set("target", subjectURI) + params.Set("collection", "at.margin.like") + params.Set("path", ".subject.uri") + + endpoint := fmt.Sprintf("%s/links/count/distinct-dids?%s", c.baseURL, params.Encode()) + + req, err := http.NewRequestWithContext(ctx, "GET", endpoint, nil) + if err != nil { + return 0, fmt.Errorf("failed to create request: %w", err) + } + req.Header.Set("User-Agent", UserAgent) + + resp, err := c.httpClient.Do(req) + if err != nil { + return 0, fmt.Errorf("request failed: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return 0, fmt.Errorf("unexpected status code: %d", resp.StatusCode) + } + + var countResp CountResponse + if err := json.NewDecoder(resp.Body).Decode(&countResp); err != nil { + return 0, fmt.Errorf("failed to decode response: %w", err) + } + + return countResp.Total, nil +} + +func (c *Client) GetReplyCount(ctx context.Context, rootURI string) (int, error) { + params := url.Values{} + params.Set("target", rootURI) + params.Set("collection", "at.margin.reply") + params.Set("path", ".root.uri") + + endpoint := fmt.Sprintf("%s/links/count?%s", c.baseURL, params.Encode()) + + req, err := http.NewRequestWithContext(ctx, "GET", endpoint, nil) + if err != nil { + return 0, fmt.Errorf("failed to create request: %w", err) + } + req.Header.Set("User-Agent", UserAgent) + + resp, err := c.httpClient.Do(req) + if err != nil { + return 0, fmt.Errorf("request failed: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return 0, fmt.Errorf("unexpected status code: %d", resp.StatusCode) + } + + var countResp CountResponse + if err := json.NewDecoder(resp.Body).Decode(&countResp); err != nil { + return 0, fmt.Errorf("failed to decode response: %w", err) + } + + return countResp.Total, nil +} + +type CountsResult struct { + LikeCount int + ReplyCount int +} + +func (c *Client) GetCountsBatch(ctx context.Context, uris []string) (map[string]CountsResult, error) { + if len(uris) == 0 { + return map[string]CountsResult{}, nil + } + + results := make(map[string]CountsResult) + var mu sync.Mutex + var wg sync.WaitGroup + + semaphore := make(chan struct{}, 10) + + for _, uri := range uris { + wg.Add(1) + go func(u string) { + defer wg.Done() + semaphore <- struct{}{} + defer func() { <-semaphore }() + + likeCount, _ := c.GetLikeCount(ctx, u) + replyCount, _ := c.GetReplyCount(ctx, u) + + mu.Lock() + results[u] = CountsResult{ + LikeCount: likeCount, + ReplyCount: replyCount, + } + mu.Unlock() + }(uri) + } + + wg.Wait() + return results, nil +} + +func (c *Client) GetAnnotationsForURL(ctx context.Context, targetURL string) ([]Link, error) { + params := url.Values{} + params.Set("target", targetURL) + params.Set("collection", "at.margin.annotation") + params.Set("path", ".target.source") + + endpoint := fmt.Sprintf("%s/links?%s", c.baseURL, params.Encode()) + + req, err := http.NewRequestWithContext(ctx, "GET", endpoint, nil) + if err != nil { + return nil, fmt.Errorf("failed to create request: %w", err) + } + req.Header.Set("User-Agent", UserAgent) + + resp, err := c.httpClient.Do(req) + if err != nil { + return nil, fmt.Errorf("request failed: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("unexpected status code: %d", resp.StatusCode) + } + + var linksResp LinksResponse + if err := json.NewDecoder(resp.Body).Decode(&linksResp); err != nil { + return nil, fmt.Errorf("failed to decode response: %w", err) + } + + return linksResp.Links, nil +} + +func (c *Client) GetHighlightsForURL(ctx context.Context, targetURL string) ([]Link, error) { + params := url.Values{} + params.Set("target", targetURL) + params.Set("collection", "at.margin.highlight") + params.Set("path", ".target.source") + + endpoint := fmt.Sprintf("%s/links?%s", c.baseURL, params.Encode()) + + req, err := http.NewRequestWithContext(ctx, "GET", endpoint, nil) + if err != nil { + return nil, fmt.Errorf("failed to create request: %w", err) + } + req.Header.Set("User-Agent", UserAgent) + + resp, err := c.httpClient.Do(req) + if err != nil { + return nil, fmt.Errorf("request failed: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("unexpected status code: %d", resp.StatusCode) + } + + var linksResp LinksResponse + if err := json.NewDecoder(resp.Body).Decode(&linksResp); err != nil { + return nil, fmt.Errorf("failed to decode response: %w", err) + } + + return linksResp.Links, nil +} + +func (c *Client) GetBookmarksForURL(ctx context.Context, targetURL string) ([]Link, error) { + params := url.Values{} + params.Set("target", targetURL) + params.Set("collection", "at.margin.bookmark") + params.Set("path", ".source") + + endpoint := fmt.Sprintf("%s/links?%s", c.baseURL, params.Encode()) + + req, err := http.NewRequestWithContext(ctx, "GET", endpoint, nil) + if err != nil { + return nil, fmt.Errorf("failed to create request: %w", err) + } + req.Header.Set("User-Agent", UserAgent) + + resp, err := c.httpClient.Do(req) + if err != nil { + return nil, fmt.Errorf("request failed: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("unexpected status code: %d", resp.StatusCode) + } + + var linksResp LinksResponse + if err := json.NewDecoder(resp.Body).Decode(&linksResp); err != nil { + return nil, fmt.Errorf("failed to decode response: %w", err) + } + + return linksResp.Links, nil +} + +func (c *Client) GetAllItemsForURL(ctx context.Context, targetURL string) (annotations, highlights, bookmarks []Link, err error) { + var wg sync.WaitGroup + var mu sync.Mutex + var errs []error + + wg.Add(3) + + go func() { + defer wg.Done() + links, e := c.GetAnnotationsForURL(ctx, targetURL) + mu.Lock() + defer mu.Unlock() + if e != nil { + errs = append(errs, e) + } else { + annotations = links + } + }() + + go func() { + defer wg.Done() + links, e := c.GetHighlightsForURL(ctx, targetURL) + mu.Lock() + defer mu.Unlock() + if e != nil { + errs = append(errs, e) + } else { + highlights = links + } + }() + + go func() { + defer wg.Done() + links, e := c.GetBookmarksForURL(ctx, targetURL) + mu.Lock() + defer mu.Unlock() + if e != nil { + errs = append(errs, e) + } else { + bookmarks = links + } + }() + + wg.Wait() + + if len(errs) > 0 { + return annotations, highlights, bookmarks, errs[0] + } + + return annotations, highlights, bookmarks, nil +} + +func (c *Client) GetLikers(ctx context.Context, subjectURI string) ([]string, error) { + params := url.Values{} + params.Set("target", subjectURI) + params.Set("collection", "at.margin.like") + params.Set("path", ".subject.uri") + + endpoint := fmt.Sprintf("%s/links/distinct-dids?%s", c.baseURL, params.Encode()) + + req, err := http.NewRequestWithContext(ctx, "GET", endpoint, nil) + if err != nil { + return nil, fmt.Errorf("failed to create request: %w", err) + } + req.Header.Set("User-Agent", UserAgent) + + resp, err := c.httpClient.Do(req) + if err != nil { + return nil, fmt.Errorf("request failed: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("unexpected status code: %d", resp.StatusCode) + } + + var result struct { + DIDs []string `json:"dids"` + } + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return nil, fmt.Errorf("failed to decode response: %w", err) + } + + return result.DIDs, nil +} diff --git a/backend/internal/crypto/cid.go b/backend/internal/crypto/cid.go new file mode 100644 index 0000000..530c18e --- /dev/null +++ b/backend/internal/crypto/cid.go @@ -0,0 +1,228 @@ +package crypto + +import ( + "bytes" + "encoding/json" + "fmt" + "sort" + "strings" + + "github.com/fxamacker/cbor/v2" + "github.com/ipfs/go-cid" + "github.com/multiformats/go-multihash" +) + +const ( + DagCBORCodec = 0x71 + SHA256Code = multihash.SHA2_256 +) + +type CIDVerificationError struct { + ExpectedCID string + ComputedCID string + RecordURI string +} + +func (e *CIDVerificationError) Error() string { + return fmt.Sprintf("CID verification failed for %s: expected %s, computed %s", + e.RecordURI, e.ExpectedCID, e.ComputedCID) +} + +func VerifyRecordCID(recordJSON json.RawMessage, expectedCID string, recordURI string) error { + if expectedCID == "" { + return nil + } + + expectedC, err := cid.Decode(expectedCID) + if err != nil { + return fmt.Errorf("invalid CID format: %w", err) + } + + cborBytes, err := jsonToDAGCBOR(recordJSON) + if err != nil { + return fmt.Errorf("failed to encode as DAG-CBOR: %w", err) + } + + mh, err := multihash.Sum(cborBytes, SHA256Code, -1) + if err != nil { + return fmt.Errorf("failed to compute hash: %w", err) + } + + computedC := cid.NewCidV1(DagCBORCodec, mh) + + if !expectedC.Equals(computedC) { + return &CIDVerificationError{ + ExpectedCID: expectedCID, + ComputedCID: computedC.String(), + RecordURI: recordURI, + } + } + + return nil +} + +func jsonToDAGCBOR(jsonData json.RawMessage) ([]byte, error) { + var data interface{} + if err := json.Unmarshal(jsonData, &data); err != nil { + return nil, err + } + + processed := processValue(data) + + encMode, err := cbor.CanonicalEncOptions().EncMode() + if err != nil { + return nil, err + } + + return encMode.Marshal(processed) +} + +func processValue(v interface{}) interface{} { + switch val := v.(type) { + case map[string]interface{}: + return processMap(val) + case []interface{}: + result := make([]interface{}, len(val)) + for i, item := range val { + result[i] = processValue(item) + } + return result + case float64: + if val == float64(int64(val)) { + return int64(val) + } + return val + case string: + return val + default: + return val + } +} + +func processMap(m map[string]interface{}) interface{} { + if link, ok := m["$link"].(string); ok && len(m) == 1 { + c, err := cid.Decode(link) + if err == nil { + return cbor.Tag{ + Number: 42, + Content: append([]byte{0x00}, c.Bytes()...), + } + } + } + + if bytesStr, ok := m["$bytes"].(string); ok && len(m) == 1 { + bytesStr = strings.TrimRight(bytesStr, "=") + decoded := decodeBase64(bytesStr) + if decoded != nil { + return decoded + } + } + + keys := make([]string, 0, len(m)) + for k := range m { + keys = append(keys, k) + } + sort.Strings(keys) + + result := make(map[string]interface{}, len(m)) + for _, k := range keys { + result[k] = processValue(m[k]) + } + + return result +} + +func decodeBase64(s string) []byte { + switch len(s) % 4 { + case 2: + s += "==" + case 3: + s += "=" + } + + decoded := make([]byte, len(s)) + n := 0 + for i := 0; i < len(s); i += 4 { + if i+4 > len(s) { + break + } + chunk := s[i : i+4] + val := uint32(0) + for _, c := range chunk { + var v byte + switch { + case c >= 'A' && c <= 'Z': + v = byte(c - 'A') + case c >= 'a' && c <= 'z': + v = byte(c - 'a' + 26) + case c >= '0' && c <= '9': + v = byte(c - '0' + 52) + case c == '+' || c == '-': + v = 62 + case c == '/' || c == '_': + v = 63 + case c == '=': + v = 0 + default: + return nil + } + val = val<<6 | uint32(v) + } + decoded[n] = byte(val >> 16) + n++ + if chunk[2] != '=' { + decoded[n] = byte(val >> 8) + n++ + } + if chunk[3] != '=' { + decoded[n] = byte(val) + n++ + } + } + return decoded[:n] +} + +func VerifyRecordCIDBatch(records []struct { + JSON json.RawMessage + CID string + URI string +}) []error { + var errors []error + for _, r := range records { + if err := VerifyRecordCID(r.JSON, r.CID, r.URI); err != nil { + errors = append(errors, err) + } + } + return errors +} + +func MustVerifyRecordCID(recordJSON json.RawMessage, expectedCID string, recordURI string) bool { + return VerifyRecordCID(recordJSON, expectedCID, recordURI) == nil +} + +func ComputeRecordCID(recordJSON json.RawMessage) (string, error) { + cborBytes, err := jsonToDAGCBOR(recordJSON) + if err != nil { + return "", fmt.Errorf("failed to encode as DAG-CBOR: %w", err) + } + + mh, err := multihash.Sum(cborBytes, SHA256Code, -1) + if err != nil { + return "", fmt.Errorf("failed to compute hash: %w", err) + } + + c := cid.NewCidV1(DagCBORCodec, mh) + return c.String(), nil +} + +func CompareRecordBytes(a, b json.RawMessage) (bool, error) { + cborA, err := jsonToDAGCBOR(a) + if err != nil { + return false, err + } + cborB, err := jsonToDAGCBOR(b) + if err != nil { + return false, err + } + return bytes.Equal(cborA, cborB), nil +} diff --git a/backend/internal/firehose/ingester.go b/backend/internal/firehose/ingester.go index e1f2ff1..44dd336 100644 --- a/backend/internal/firehose/ingester.go +++ b/backend/internal/firehose/ingester.go @@ -10,11 +10,14 @@ import ( "time" "github.com/gorilla/websocket" + "margin.at/internal/crypto" "margin.at/internal/db" internal_sync "margin.at/internal/sync" "margin.at/internal/xrpc" ) +var CIDVerificationEnabled = true + const ( CollectionAnnotation = "at.margin.annotation" CollectionHighlight = "at.margin.highlight" @@ -28,13 +31,20 @@ const ( CollectionSembleCollection = "network.cosmik.collection" ) -var RelayURL = "wss://jetstream2.us-east.bsky.network/subscribe" +var RelayURLs = []string{ + "wss://jetstream2.us-east.bsky.network/subscribe", + "wss://jetstream2.fr.hose.cam/subscribe", + "wss://jetstream.fire.hose.cam/subscribe", +} + +var RelayURL = RelayURLs[0] type Ingester struct { - db *db.DB - sync *internal_sync.Service - cancel context.CancelFunc - handlers map[string]RecordHandler + db *db.DB + sync *internal_sync.Service + cancel context.CancelFunc + handlers map[string]RecordHandler + currentRelayIdx int } type RecordHandler func(event *FirehoseEvent) @@ -80,17 +90,30 @@ func (i *Ingester) Stop() { } func (i *Ingester) run(ctx context.Context) { + consecutiveFailures := 0 + maxFailuresBeforeSwitch := 3 + for { select { case <-ctx.Done(): return default: if err := i.subscribe(ctx); err != nil { - log.Printf("Jetstream error: %v, reconnecting in 5s...", err) + consecutiveFailures++ + log.Printf("Jetstream error (relay %d): %v, reconnecting in 5s...", i.currentRelayIdx, err) + + if consecutiveFailures >= maxFailuresBeforeSwitch { + i.currentRelayIdx = (i.currentRelayIdx + 1) % len(RelayURLs) + log.Printf("Switching to relay %d: %s", i.currentRelayIdx, RelayURLs[i.currentRelayIdx]) + consecutiveFailures = 0 + } + if ctx.Err() != nil { return } time.Sleep(5 * time.Second) + } else { + consecutiveFailures = 0 } } } @@ -120,7 +143,8 @@ func (i *Ingester) subscribe(ctx context.Context) error { collections = append(collections, collection) } - url := fmt.Sprintf("%s?wantedCollections=%s", RelayURL, strings.Join(collections, "&wantedCollections=")) + relayURL := RelayURLs[i.currentRelayIdx] + url := fmt.Sprintf("%s?wantedCollections=%s", relayURL, strings.Join(collections, "&wantedCollections=")) if cursor > 0 { url = fmt.Sprintf("%s&cursor=%d", url, cursor) } @@ -171,6 +195,13 @@ func (i *Ingester) handleCommit(event JetstreamEvent) { switch commit.Operation { case "create", "update": if len(commit.Record) > 0 { + if CIDVerificationEnabled && commit.Cid != "" { + if err := crypto.VerifyRecordCID(commit.Record, commit.Cid, uri); err != nil { + log.Printf("CID verification failed for %s: %v (skipping)", uri, err) + return + } + } + firehoseEvent := &FirehoseEvent{ Repo: event.Did, Collection: commit.Collection, @@ -178,6 +209,7 @@ func (i *Ingester) handleCommit(event JetstreamEvent) { Record: commit.Record, Operation: commit.Operation, Cursor: event.Time, + CID: commit.Cid, } i.dispatchToHandler(firehoseEvent) @@ -267,6 +299,7 @@ type FirehoseEvent struct { Record json.RawMessage `json:"record"` Operation string `json:"operation"` Cursor int64 `json:"cursor"` + CID string `json:"cid"` } func (i *Ingester) handleAnnotation(event *FirehoseEvent) { diff --git a/backend/internal/oauth/handler.go b/backend/internal/oauth/handler.go index f64e1c4..a64d0bc 100644 --- a/backend/internal/oauth/handler.go +++ b/backend/internal/oauth/handler.go @@ -330,6 +330,12 @@ func (h *Handler) HandleCallback(w http.ResponseWriter, r *http.Request) { return } + if tokenResp.Sub != pending.DID { + log.Printf("Security: OAuth sub mismatch, expected %s, got %s", pending.DID, tokenResp.Sub) + http.Error(w, "Account identity mismatch, authorization returned different account", http.StatusBadRequest) + return + } + _ = newNonce sessionID := generateSessionID() diff --git a/backend/internal/slingshot/client.go b/backend/internal/slingshot/client.go new file mode 100644 index 0000000..f7a12c6 --- /dev/null +++ b/backend/internal/slingshot/client.go @@ -0,0 +1,179 @@ +package slingshot + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/url" + "time" +) + +const ( + DefaultBaseURL = "https://slingshot.microcosm.blue" + DefaultTimeout = 5 * time.Second + UserAgent = "Margin (margin.at)" +) + +type Client struct { + baseURL string + httpClient *http.Client +} + +func NewClient() *Client { + return &Client{ + baseURL: DefaultBaseURL, + httpClient: &http.Client{ + Timeout: DefaultTimeout, + }, + } +} + +func NewClientWithURL(baseURL string) *Client { + return &Client{ + baseURL: baseURL, + httpClient: &http.Client{ + Timeout: DefaultTimeout, + }, + } +} + +type Identity struct { + DID string `json:"did"` + Handle string `json:"handle"` + PDS string `json:"pds"` +} + +type Record struct { + URI string `json:"uri"` + CID string `json:"cid"` + Value json.RawMessage `json:"value"` +} + +func (c *Client) ResolveIdentity(ctx context.Context, identifier string) (*Identity, error) { + endpoint := fmt.Sprintf("%s/identity/%s", c.baseURL, url.PathEscape(identifier)) + + req, err := http.NewRequestWithContext(ctx, "GET", endpoint, nil) + if err != nil { + return nil, fmt.Errorf("failed to create request: %w", err) + } + req.Header.Set("User-Agent", UserAgent) + + resp, err := c.httpClient.Do(req) + if err != nil { + return nil, fmt.Errorf("request failed: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode == http.StatusNotFound { + return nil, fmt.Errorf("identity not found: %s", identifier) + } + + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("unexpected status code: %d", resp.StatusCode) + } + + var identity Identity + if err := json.NewDecoder(resp.Body).Decode(&identity); err != nil { + return nil, fmt.Errorf("failed to decode response: %w", err) + } + + return &identity, nil +} + +func (c *Client) GetRecord(ctx context.Context, uri string) (*Record, error) { + params := url.Values{} + params.Set("uri", uri) + + endpoint := fmt.Sprintf("%s/record?%s", c.baseURL, params.Encode()) + + req, err := http.NewRequestWithContext(ctx, "GET", endpoint, nil) + if err != nil { + return nil, fmt.Errorf("failed to create request: %w", err) + } + req.Header.Set("User-Agent", UserAgent) + + resp, err := c.httpClient.Do(req) + if err != nil { + return nil, fmt.Errorf("request failed: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode == http.StatusNotFound { + return nil, fmt.Errorf("record not found: %s", uri) + } + + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("unexpected status code: %d", resp.StatusCode) + } + + var record Record + if err := json.NewDecoder(resp.Body).Decode(&record); err != nil { + return nil, fmt.Errorf("failed to decode response: %w", err) + } + + return &record, nil +} + +func (c *Client) GetRecordByParts(ctx context.Context, repo, collection, rkey string) (*Record, error) { + uri := fmt.Sprintf("at://%s/%s/%s", repo, collection, rkey) + return c.GetRecord(ctx, uri) +} + +type ListRecordsResponse struct { + Records []Record `json:"records"` + Cursor string `json:"cursor,omitempty"` +} + +func (c *Client) ListRecords(ctx context.Context, repo, collection string, limit int, cursor string) (*ListRecordsResponse, error) { + params := url.Values{} + params.Set("repo", repo) + params.Set("collection", collection) + if limit > 0 { + params.Set("limit", fmt.Sprintf("%d", limit)) + } + if cursor != "" { + params.Set("cursor", cursor) + } + + endpoint := fmt.Sprintf("%s/records?%s", c.baseURL, params.Encode()) + + req, err := http.NewRequestWithContext(ctx, "GET", endpoint, nil) + if err != nil { + return nil, fmt.Errorf("failed to create request: %w", err) + } + req.Header.Set("User-Agent", UserAgent) + + resp, err := c.httpClient.Do(req) + if err != nil { + return nil, fmt.Errorf("request failed: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("unexpected status code: %d", resp.StatusCode) + } + + var listResp ListRecordsResponse + if err := json.NewDecoder(resp.Body).Decode(&listResp); err != nil { + return nil, fmt.Errorf("failed to decode response: %w", err) + } + + return &listResp, nil +} + +func (c *Client) ResolveDID(ctx context.Context, did string) (string, error) { + identity, err := c.ResolveIdentity(ctx, did) + if err != nil { + return "", err + } + return identity.PDS, nil +} + +func (c *Client) ResolveHandle(ctx context.Context, handle string) (string, error) { + identity, err := c.ResolveIdentity(ctx, handle) + if err != nil { + return "", err + } + return identity.DID, nil +} diff --git a/backend/internal/sync/service.go b/backend/internal/sync/service.go index c356818..bdd7c8b 100644 --- a/backend/internal/sync/service.go +++ b/backend/internal/sync/service.go @@ -5,14 +5,18 @@ import ( "encoding/json" "fmt" "io" + "log" "net/http" "strings" "time" + "margin.at/internal/crypto" "margin.at/internal/db" "margin.at/internal/xrpc" ) +var CIDVerificationEnabled = true + type Service struct { db *db.DB } @@ -82,6 +86,13 @@ func (s *Service) PerformSync(ctx context.Context, did string, getClient func(co } for _, rec := range output.Records { + if CIDVerificationEnabled && rec.CID != "" { + if err := crypto.VerifyRecordCID(rec.Value, rec.CID, rec.URI); err != nil { + log.Printf("CID verification failed for %s: %v (skipping)", rec.URI, err) + continue + } + } + err := s.upsertRecord(did, collectionNSID, rec.URI, rec.CID, rec.Value) if err != nil { fmt.Printf("Error upserting %s: %v\n", rec.URI, err) diff --git a/backend/internal/xrpc/client.go b/backend/internal/xrpc/client.go index 82428a8..ad44ca4 100644 --- a/backend/internal/xrpc/client.go +++ b/backend/internal/xrpc/client.go @@ -11,7 +11,6 @@ import ( "fmt" "io" "net/http" - "strings" "time" "github.com/go-jose/go-jose/v4" @@ -193,17 +192,16 @@ func (c *Client) DeleteRecord(ctx context.Context, repo, collection, rkey string } func (c *Client) DeleteRecordByURI(ctx context.Context, uri string) error { - - if !strings.HasPrefix(uri, "at://") { - return fmt.Errorf("invalid AT URI format") + parsed, err := ParseATURI(uri) + if err != nil { + return err } - parts := strings.Split(strings.TrimPrefix(uri, "at://"), "/") - if len(parts) != 3 { - return fmt.Errorf("invalid AT URI format") + if parsed.Collection == "" || parsed.RKey == "" { + return fmt.Errorf("invalid AT-URI: must include collection and rkey") } - return c.DeleteRecord(ctx, parts[0], parts[1], parts[2]) + return c.DeleteRecord(ctx, parsed.DID, parsed.Collection, parsed.RKey) } type PutRecordInput struct { diff --git a/backend/internal/xrpc/utils.go b/backend/internal/xrpc/utils.go index c413869..6eebe69 100644 --- a/backend/internal/xrpc/utils.go +++ b/backend/internal/xrpc/utils.go @@ -1,14 +1,103 @@ package xrpc import ( + "context" "encoding/json" "fmt" + "log" "net/http" + "regexp" "strings" "time" + + "margin.at/internal/slingshot" +) + +var SlingshotClient = slingshot.NewClient() + +var ( + didPattern = regexp.MustCompile(`^did:[a-z]+:[a-zA-Z0-9._:%-]+$`) + nsidPattern = regexp.MustCompile(`^[a-z][a-z0-9]*(\.[a-z][a-z0-9]*)+$`) + rkeyPattern = regexp.MustCompile(`^[a-zA-Z0-9._-]+$`) ) +type ATURI struct { + DID string + Collection string + RKey string +} + +func ParseATURI(uri string) (*ATURI, error) { + if !strings.HasPrefix(uri, "at://") { + return nil, fmt.Errorf("invalid AT-URI: must start with at://") + } + + path := strings.TrimPrefix(uri, "at://") + parts := strings.Split(path, "/") + + if len(parts) < 1 || parts[0] == "" { + return nil, fmt.Errorf("invalid AT-URI: missing DID authority") + } + + did := parts[0] + if !didPattern.MatchString(did) { + return nil, fmt.Errorf("invalid AT-URI: malformed DID %q", did) + } + + result := &ATURI{DID: did} + + if len(parts) >= 2 && parts[1] != "" { + collection := parts[1] + if !nsidPattern.MatchString(collection) { + return nil, fmt.Errorf("invalid AT-URI: malformed collection NSID %q", collection) + } + result.Collection = collection + } + + if len(parts) >= 3 && parts[2] != "" { + rkey := parts[2] + if !rkeyPattern.MatchString(rkey) || strings.HasPrefix(rkey, ".") || strings.HasSuffix(rkey, ".") { + return nil, fmt.Errorf("invalid AT-URI: malformed record key %q", rkey) + } + if len(rkey) > 512 { + return nil, fmt.Errorf("invalid AT-URI: record key too long (max 512)") + } + result.RKey = rkey + } + + if len(parts) > 3 { + return nil, fmt.Errorf("invalid AT-URI: too many path segments") + } + + return result, nil +} + +func (a *ATURI) String() string { + if a.Collection == "" { + return fmt.Sprintf("at://%s", a.DID) + } + if a.RKey == "" { + return fmt.Sprintf("at://%s/%s", a.DID, a.Collection) + } + return fmt.Sprintf("at://%s/%s/%s", a.DID, a.Collection, a.RKey) +} + +func init() { + log.Printf("Slingshot client initialized: %s", slingshot.DefaultBaseURL) +} + func ResolveDIDToPDS(did string) (string, error) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + if pds, err := SlingshotClient.ResolveDID(ctx, did); err == nil && pds != "" { + return pds, nil + } + + return resolveDIDToPDSDirect(did) +} + +func resolveDIDToPDSDirect(did string) (string, error) { var docURL string if strings.HasPrefix(did, "did:plc:") { docURL = fmt.Sprintf("https://plc.directory/%s", did) @@ -34,6 +123,7 @@ func ResolveDIDToPDS(did string) (string, error) { var doc struct { Service []struct { + ID string `json:"id"` Type string `json:"type"` ServiceEndpoint string `json:"serviceEndpoint"` } `json:"service"` @@ -42,6 +132,11 @@ func ResolveDIDToPDS(did string) (string, error) { return "", err } + for _, svc := range doc.Service { + if svc.ID == "#atproto_pds" && svc.Type == "AtprotoPersonalDataServer" { + return svc.ServiceEndpoint, nil + } + } for _, svc := range doc.Service { if svc.Type == "AtprotoPersonalDataServer" { return svc.ServiceEndpoint, nil @@ -49,11 +144,23 @@ func ResolveDIDToPDS(did string) (string, error) { } return "", nil } + func ResolveHandle(handle string) (string, error) { if strings.HasPrefix(handle, "did:") { return handle, nil } + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + if did, err := SlingshotClient.ResolveHandle(ctx, handle); err == nil && did != "" { + return did, nil + } + + return resolveHandleDirect(handle) +} + +func resolveHandleDirect(handle string) (string, error) { url := fmt.Sprintf("https://public.api.bsky.app/xrpc/com.atproto.identity.resolveHandle?handle=%s", handle) client := &http.Client{ Timeout: 5 * time.Second, -- 2.51.2