diff --git a/backend/cmd/server/main.go b/backend/cmd/server/main.go index 640c3ae..f513fdc 100644 --- a/backend/cmd/server/main.go +++ b/backend/cmd/server/main.go @@ -20,6 +20,7 @@ import ( "margin.at/internal/db" "margin.at/internal/firehose" "margin.at/internal/oauth" + "margin.at/internal/sync" ) func main() { @@ -35,13 +36,15 @@ func main() { log.Fatalf("Failed to run migrations: %v", err) } - oauthHandler, err := oauth.NewHandler(database) + syncSvc := sync.NewService(database) + + oauthHandler, err := oauth.NewHandler(database, syncSvc) if err != nil { log.Fatalf("Failed to initialize OAuth: %v", err) } - ingester := firehose.NewIngester(database) - firehose.RelayURL = getEnv("BLOCK_RELAY_URL", "wss://bsky.network/xrpc/com.atproto.sync.subscribeRepos") + ingester := firehose.NewIngester(database, syncSvc) + firehose.RelayURL = getEnv("BLOCK_RELAY_URL", "wss://jetstream2.us-east.bsky.network/subscribe") log.Printf("Firehose URL: %s", firehose.RelayURL) go func() { @@ -71,7 +74,7 @@ func main() { tokenRefresher := api.NewTokenRefresher(database, oauthHandler.GetPrivateKey()) annotationSvc := api.NewAnnotationService(database, tokenRefresher) - handler := api.NewHandler(database, annotationSvc, tokenRefresher) + handler := api.NewHandler(database, annotationSvc, tokenRefresher, syncSvc) handler.RegisterRoutes(r) r.Post("/api/annotations", annotationSvc.CreateAnnotation) diff --git a/backend/go.mod b/backend/go.mod index 3c12a47..e81fec5 100644 --- a/backend/go.mod +++ b/backend/go.mod @@ -3,12 +3,10 @@ module margin.at go 1.24.0 require ( - github.com/fxamacker/cbor/v2 v2.9.0 github.com/go-chi/chi/v5 v5.1.0 github.com/go-chi/cors v1.2.1 github.com/go-jose/go-jose/v4 v4.0.4 github.com/gorilla/websocket v1.5.3 - github.com/ipfs/go-cid v0.6.0 github.com/joho/godotenv v1.5.1 github.com/lib/pq v1.10.9 github.com/mattn/go-sqlite3 v1.14.22 @@ -17,20 +15,8 @@ require ( require ( github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // 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 ff92a28..f6bbbc9 100644 --- a/backend/go.sum +++ b/backend/go.sum @@ -1,7 +1,5 @@ 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= @@ -12,48 +10,21 @@ 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/annotations.go b/backend/internal/api/annotations.go index cb36791..e6ad150 100644 --- a/backend/internal/api/annotations.go +++ b/backend/internal/api/annotations.go @@ -505,36 +505,6 @@ func (s *AnnotationService) DeleteReply(w http.ResponseWriter, r *http.Request) json.NewEncoder(w).Encode(map[string]bool{"success": true}) } -func resolveDIDToPDS(did string) (string, error) { - if strings.HasPrefix(did, "did:plc:") { - client := &http.Client{ - Timeout: 10 * time.Second, - } - resp, err := client.Get("https://plc.directory/" + did) - if err != nil { - return "", err - } - defer resp.Body.Close() - - var doc struct { - Service []struct { - Type string `json:"type"` - ServiceEndpoint string `json:"serviceEndpoint"` - } `json:"service"` - } - if err := json.NewDecoder(resp.Body).Decode(&doc); err != nil { - return "", err - } - - for _, svc := range doc.Service { - if svc.Type == "AtprotoPersonalDataServer" { - return svc.ServiceEndpoint, nil - } - } - } - return "", nil -} - type CreateHighlightRequest struct { URL string `json:"url"` Title string `json:"title,omitempty"` diff --git a/backend/internal/api/apikey.go b/backend/internal/api/apikey.go index f597e01..8ca8d2d 100644 --- a/backend/internal/api/apikey.go +++ b/backend/internal/api/apikey.go @@ -480,7 +480,7 @@ func (h *APIKeyHandler) getSessionByDID(did string) (*SessionData, error) { return nil, fmt.Errorf("invalid session DPoP key: %w", err) } - pds, err := resolveDIDToPDS(sessDID) + pds, err := xrpc.ResolveDIDToPDS(sessDID) if err != nil { return nil, fmt.Errorf("failed to resolve PDS: %w", err) } diff --git a/backend/internal/api/handler.go b/backend/internal/api/handler.go index 72c8306..36f18f8 100644 --- a/backend/internal/api/handler.go +++ b/backend/internal/api/handler.go @@ -14,6 +14,7 @@ import ( "github.com/go-chi/chi/v5" "margin.at/internal/db" + internal_sync "margin.at/internal/sync" "margin.at/internal/xrpc" ) @@ -22,14 +23,16 @@ type Handler struct { annotationService *AnnotationService refresher *TokenRefresher apiKeys *APIKeyHandler + syncService *internal_sync.Service } -func NewHandler(database *db.DB, annotationService *AnnotationService, refresher *TokenRefresher) *Handler { +func NewHandler(database *db.DB, annotationService *AnnotationService, refresher *TokenRefresher, syncService *internal_sync.Service) *Handler { return &Handler{ db: database, annotationService: annotationService, refresher: refresher, apiKeys: NewAPIKeyHandler(database, refresher), + syncService: syncService, } } diff --git a/backend/internal/api/sync.go b/backend/internal/api/sync.go index 53f23fa..c950214 100644 --- a/backend/internal/api/sync.go +++ b/backend/internal/api/sync.go @@ -1,13 +1,10 @@ package api import ( + "context" "encoding/json" - "fmt" - "io" "net/http" - "time" - "margin.at/internal/db" "margin.at/internal/xrpc" ) @@ -18,110 +15,13 @@ func (h *Handler) SyncAll(w http.ResponseWriter, r *http.Request) { return } - collections := []string{ - xrpc.CollectionAnnotation, - xrpc.CollectionHighlight, - xrpc.CollectionBookmark, - xrpc.CollectionReply, - xrpc.CollectionLike, - xrpc.CollectionCollection, - xrpc.CollectionCollectionItem, - } - - results := make(map[string]string) - - err = h.refresher.ExecuteWithAutoRefresh(r, session, func(client *xrpc.Client, did string) error { - for _, collectionNSID := range collections { - count := 0 - cursor := "" - fetchedURIs := make(map[string]bool) - - for { - url := fmt.Sprintf("%s/xrpc/com.atproto.repo.listRecords?repo=%s&collection=%s&limit=100", client.PDS, did, collectionNSID) - if cursor != "" { - url += "&cursor=" + cursor - } - - req, _ := http.NewRequestWithContext(r.Context(), "GET", url, nil) - req.Header.Set("Authorization", "Bearer "+client.AccessToken) - - resp, err := http.DefaultClient.Do(req) - if err != nil { - return fmt.Errorf("failed to fetch %s: %w", collectionNSID, err) - } - defer resp.Body.Close() - - if resp.StatusCode != 200 { - body, _ := io.ReadAll(resp.Body) - results[collectionNSID] = fmt.Sprintf("error: %s", string(body)) - break - } - - var output struct { - Records []struct { - URI string `json:"uri"` - CID string `json:"cid"` - Value json.RawMessage `json:"value"` - } `json:"records"` - Cursor string `json:"cursor"` - } - - if err := json.NewDecoder(resp.Body).Decode(&output); err != nil { - return err - } - - for _, rec := range output.Records { - err := h.upsertRecord(did, collectionNSID, rec.URI, rec.CID, rec.Value) - if err != nil { - fmt.Printf("Error upserting %s: %v\n", rec.URI, err) - } else { - count++ - fetchedURIs[rec.URI] = true - } - } - - if output.Cursor == "" { - break - } - cursor = output.Cursor - } - - deletedCount := 0 - if results[collectionNSID] == "" { - var localURIs []string - var err error - - switch collectionNSID { - case xrpc.CollectionAnnotation: - localURIs, err = h.db.GetAnnotationURIs(did) - case xrpc.CollectionHighlight: - localURIs, err = h.db.GetHighlightURIs(did) - case xrpc.CollectionBookmark: - localURIs, err = h.db.GetBookmarkURIs(did) - } - - if err == nil { - for _, uri := range localURIs { - if !fetchedURIs[uri] { - switch collectionNSID { - case xrpc.CollectionAnnotation: - _ = h.db.DeleteAnnotation(uri) - case xrpc.CollectionHighlight: - _ = h.db.DeleteHighlight(uri) - case xrpc.CollectionBookmark: - _ = h.db.DeleteBookmark(uri) - } - deletedCount++ - } - } - } - } - - if results[collectionNSID] == "" { - results[collectionNSID] = fmt.Sprintf("synced %d records, deleted %d stale", count, deletedCount) - } - } - return nil + results, err := h.syncService.PerformSync(r.Context(), session.DID, func(ctx context.Context, did string) (*xrpc.Client, error) { + var client *xrpc.Client + err := h.refresher.ExecuteWithAutoRefresh(r, session, func(c *xrpc.Client, d string) error { + client = c + return nil + }) + return client, err }) if err != nil { @@ -132,220 +32,3 @@ func (h *Handler) SyncAll(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) json.NewEncoder(w).Encode(results) } - -func strPtr(s string) *string { - if s == "" { - return nil - } - return &s -} - -func (h *Handler) upsertRecord(did, collection, uri, cid string, value json.RawMessage) error { - cidPtr := strPtr(cid) - switch collection { - case xrpc.CollectionAnnotation: - var record xrpc.AnnotationRecord - if err := json.Unmarshal(value, &record); err != nil { - return err - } - - createdAt, _ := time.Parse(time.RFC3339, record.CreatedAt) - - targetSource := record.Target.Source - if targetSource == "" { - - } - - targetHash := record.Target.SourceHash - if targetHash == "" && targetSource != "" { - targetHash = db.HashURL(targetSource) - } - - motivation := record.Motivation - if motivation == "" { - motivation = "commenting" - } - - var bodyValuePtr, bodyFormatPtr, bodyURIPtr, targetTitlePtr, selectorJSONPtr, tagsJSONPtr *string - if record.Body != nil { - if record.Body.Value != "" { - val := record.Body.Value - bodyValuePtr = &val - } - if record.Body.Format != "" { - fmt := record.Body.Format - bodyFormatPtr = &fmt - } - } - if record.Target.Title != "" { - t := record.Target.Title - targetTitlePtr = &t - } - if len(record.Target.Selector) > 0 { - selectorStr := string(record.Target.Selector) - selectorJSONPtr = &selectorStr - } - if len(record.Tags) > 0 { - tagsBytes, _ := json.Marshal(record.Tags) - tagsStr := string(tagsBytes) - tagsJSONPtr = &tagsStr - } - - return h.db.CreateAnnotation(&db.Annotation{ - URI: uri, - AuthorDID: did, - Motivation: motivation, - BodyValue: bodyValuePtr, - BodyFormat: bodyFormatPtr, - BodyURI: bodyURIPtr, - TargetSource: targetSource, - TargetHash: targetHash, - TargetTitle: targetTitlePtr, - SelectorJSON: selectorJSONPtr, - TagsJSON: tagsJSONPtr, - CreatedAt: createdAt, - IndexedAt: time.Now(), - CID: cidPtr, - }) - - case xrpc.CollectionHighlight: - var record xrpc.HighlightRecord - if err := json.Unmarshal(value, &record); err != nil { - return err - } - - createdAt, _ := time.Parse(time.RFC3339, record.CreatedAt) - if createdAt.IsZero() { - createdAt = time.Now() - } - - targetHash := record.Target.SourceHash - if targetHash == "" && record.Target.Source != "" { - targetHash = db.HashURL(record.Target.Source) - } - - var titlePtr, selectorJSONPtr, colorPtr, tagsJSONPtr *string - if record.Target.Title != "" { - t := record.Target.Title - titlePtr = &t - } - if len(record.Target.Selector) > 0 { - selectorStr := string(record.Target.Selector) - selectorJSONPtr = &selectorStr - } - if record.Color != "" { - c := record.Color - colorPtr = &c - } - if len(record.Tags) > 0 { - tagsBytes, _ := json.Marshal(record.Tags) - tagsStr := string(tagsBytes) - tagsJSONPtr = &tagsStr - } - - return h.db.CreateHighlight(&db.Highlight{ - URI: uri, - AuthorDID: did, - TargetSource: record.Target.Source, - TargetHash: targetHash, - TargetTitle: titlePtr, - SelectorJSON: selectorJSONPtr, - Color: colorPtr, - TagsJSON: tagsJSONPtr, - CreatedAt: createdAt, - IndexedAt: time.Now(), - CID: cidPtr, - }) - - case xrpc.CollectionBookmark: - var record xrpc.BookmarkRecord - if err := json.Unmarshal(value, &record); err != nil { - return err - } - - createdAt, _ := time.Parse(time.RFC3339, record.CreatedAt) - - sourceHash := record.SourceHash - if sourceHash == "" && record.Source != "" { - sourceHash = db.HashURL(record.Source) - } - - var titlePtr, descPtr, tagsJSONPtr *string - if record.Title != "" { - t := record.Title - titlePtr = &t - } - if record.Description != "" { - d := record.Description - descPtr = &d - } - if len(record.Tags) > 0 { - tagsBytes, _ := json.Marshal(record.Tags) - tagsStr := string(tagsBytes) - tagsJSONPtr = &tagsStr - } - - return h.db.CreateBookmark(&db.Bookmark{ - URI: uri, - AuthorDID: did, - Source: record.Source, - SourceHash: sourceHash, - Title: titlePtr, - Description: descPtr, - TagsJSON: tagsJSONPtr, - CreatedAt: createdAt, - IndexedAt: time.Now(), - CID: cidPtr, - }) - - case xrpc.CollectionCollection: - var record xrpc.CollectionRecord - if err := json.Unmarshal(value, &record); err != nil { - return err - } - createdAt, _ := time.Parse(time.RFC3339, record.CreatedAt) - - var descPtr, iconPtr *string - if record.Description != "" { - d := record.Description - descPtr = &d - } - if record.Icon != "" { - i := record.Icon - iconPtr = &i - } - - return h.db.CreateCollection(&db.Collection{ - URI: uri, - AuthorDID: did, - Name: record.Name, - Description: descPtr, - Icon: iconPtr, - CreatedAt: createdAt, - IndexedAt: time.Now(), - }) - - case xrpc.CollectionCollectionItem: - var record xrpc.CollectionItemRecord - if err := json.Unmarshal(value, &record); err != nil { - return err - } - createdAt, _ := time.Parse(time.RFC3339, record.CreatedAt) - - return h.db.AddToCollection(&db.CollectionItem{ - URI: uri, - AuthorDID: did, - CollectionURI: record.Collection, - AnnotationURI: record.Annotation, - Position: record.Position, - CreatedAt: createdAt, - IndexedAt: time.Now(), - }) - - case xrpc.CollectionReply: - return nil - case xrpc.CollectionLike: - return nil - } - return nil -} diff --git a/backend/internal/api/token_refresh.go b/backend/internal/api/token_refresh.go index f72645b..9c32b1b 100644 --- a/backend/internal/api/token_refresh.go +++ b/backend/internal/api/token_refresh.go @@ -89,7 +89,7 @@ func (tr *TokenRefresher) GetSessionWithAutoRefresh(r *http.Request) (*SessionDa return nil, fmt.Errorf("invalid session DPoP key") } - pds, err := resolveDIDToPDS(did) + pds, err := xrpc.ResolveDIDToPDS(did) if err != nil { return nil, fmt.Errorf("failed to resolve PDS") } diff --git a/backend/internal/db/tags.go b/backend/internal/db/tags.go index b14779c..bd86844 100644 --- a/backend/internal/db/tags.go +++ b/backend/internal/db/tags.go @@ -1,5 +1,7 @@ package db +import "fmt" + type TrendingTag struct { Tag string `json:"tag"` Count int `json:"count"` @@ -14,12 +16,19 @@ func (db *DB) GetTrendingTags(limit int) ([]TrendingTag, error) { WHERE tags_json IS NOT NULL AND tags_json != '' AND tags_json != '[]' + AND created_at > %s GROUP BY tag + HAVING count > 2 ORDER BY count DESC LIMIT ? ` - rows, err := db.Query(db.Rebind(query), limit) + dateFilter := "datetime('now', '-7 days')" + if db.driver == "postgres" { + dateFilter = "NOW() - INTERVAL '7 days'" + } + + rows, err := db.Query(db.Rebind(fmt.Sprintf(query, dateFilter)), limit) if err != nil { return nil, err } diff --git a/backend/internal/firehose/ingester.go b/backend/internal/firehose/ingester.go index 2af0844..1b6eca5 100644 --- a/backend/internal/firehose/ingester.go +++ b/backend/internal/firehose/ingester.go @@ -1,20 +1,18 @@ package firehose import ( - "bytes" "context" - "encoding/binary" "encoding/json" "fmt" - "io" "log" + "strings" + "sync" "time" - "github.com/fxamacker/cbor/v2" "github.com/gorilla/websocket" - "github.com/ipfs/go-cid" - "margin.at/internal/db" + internal_sync "margin.at/internal/sync" + "margin.at/internal/xrpc" ) const ( @@ -27,15 +25,37 @@ const ( CollectionCollectionItem = "at.margin.collectionItem" ) -var RelayURL = "wss://bsky.network/xrpc/com.atproto.sync.subscribeRepos" +var RelayURL = "wss://jetstream2.us-east.bsky.network/subscribe" type Ingester struct { - db *db.DB - cancel context.CancelFunc + db *db.DB + sync *internal_sync.Service + cancel context.CancelFunc + handlers map[string]RecordHandler +} + +type RecordHandler func(event *FirehoseEvent) + +func NewIngester(database *db.DB, syncService *internal_sync.Service) *Ingester { + i := &Ingester{ + db: database, + sync: syncService, + handlers: make(map[string]RecordHandler), + } + + i.RegisterHandler(CollectionAnnotation, i.handleAnnotation) + i.RegisterHandler(CollectionHighlight, i.handleHighlight) + i.RegisterHandler(CollectionBookmark, i.handleBookmark) + i.RegisterHandler(CollectionReply, i.handleReply) + i.RegisterHandler(CollectionLike, i.handleLike) + i.RegisterHandler(CollectionCollection, i.handleCollection) + i.RegisterHandler(CollectionCollectionItem, i.handleCollectionItem) + + return i } -func NewIngester(database *db.DB) *Ingester { - return &Ingester{db: database} +func (i *Ingester) RegisterHandler(collection string, handler RecordHandler) { + i.handlers[collection] = handler } func (i *Ingester) Start(ctx context.Context) error { @@ -59,7 +79,7 @@ func (i *Ingester) run(ctx context.Context) { return default: if err := i.subscribe(ctx); err != nil { - log.Printf("Firehose error: %v, reconnecting in 5s...", err) + log.Printf("Jetstream error: %v, reconnecting in 5s...", err) if ctx.Err() != nil { return } @@ -69,35 +89,36 @@ func (i *Ingester) run(ctx context.Context) { } } -type FrameHeader struct { - Op int `cbor:"op"` - T string `cbor:"t"` -} -type Commit struct { - Repo string `cbor:"repo"` - Rev string `cbor:"rev"` - Seq int64 `cbor:"seq"` - Prev *cid.Cid `cbor:"prev"` - Time string `cbor:"time"` - Blocks []byte `cbor:"blocks"` - Ops []RepoOp `cbor:"ops"` +type JetstreamEvent struct { + Did string `json:"did"` + Time int64 `json:"time_us"` + Kind string `json:"kind"` + Commit *JetstreamCommit `json:"commit,omitempty"` } -type RepoOp struct { - Action string `cbor:"action"` - Path string `cbor:"path"` - Cid *cid.Cid `cbor:"cid"` +type JetstreamCommit struct { + Rev string `json:"rev"` + Operation string `json:"operation"` + Collection string `json:"collection"` + Rkey string `json:"rkey"` + Record json.RawMessage `json:"record,omitempty"` + Cid string `json:"cid,omitempty"` } func (i *Ingester) subscribe(ctx context.Context) error { cursor := i.getLastCursor() - url := RelayURL + var collections []string + for collection := range i.handlers { + collections = append(collections, collection) + } + + url := fmt.Sprintf("%s?wantedCollections=%s", RelayURL, strings.Join(collections, "&wantedCollections=")) if cursor > 0 { - url = fmt.Sprintf("%s?cursor=%d", RelayURL, cursor) + url = fmt.Sprintf("%s&cursor=%d", url, cursor) } - log.Printf("Connecting to firehose: %s", url) + log.Printf("Connecting to Jetstream: %s", url) conn, _, err := websocket.DefaultDialer.DialContext(ctx, url, nil) if err != nil { @@ -105,7 +126,7 @@ func (i *Ingester) subscribe(ctx context.Context) error { } defer conn.Close() - log.Printf("Connected to firehose") + log.Printf("Connected to Jetstream") for { select { @@ -119,155 +140,79 @@ func (i *Ingester) subscribe(ctx context.Context) error { return fmt.Errorf("websocket read failed: %w", err) } - i.handleMessage(message) - } -} - -func (i *Ingester) handleMessage(data []byte) { - reader := bytes.NewReader(data) - - var header FrameHeader - decoder := cbor.NewDecoder(reader) - if err := decoder.Decode(&header); err != nil { - return - } - - if header.Op != 1 { - return - } - - if header.T != "#commit" { - return - } - - var commit Commit - if err := decoder.Decode(&commit); err != nil { - return - } - - for _, op := range commit.Ops { - collection, rkey := parseOpPath(op.Path) - if !isMarginCollection(collection) { + var event JetstreamEvent + if err := json.Unmarshal(message, &event); err != nil { continue } - uri := fmt.Sprintf("at://%s/%s/%s", commit.Repo, collection, rkey) + if event.Kind == "commit" && event.Commit != nil { + i.handleCommit(event) - switch op.Action { - case "create", "update": - if op.Cid != nil && len(commit.Blocks) > 0 { - record := extractRecord(commit.Blocks, *op.Cid) - if record != nil { - i.handleRecord(commit.Repo, collection, rkey, record, commit.Seq) + if event.Time > 0 { + if err := i.db.SetCursor("firehose_cursor", event.Time); err != nil { + log.Printf("Failed to save cursor: %v", err) } } - case "delete": - i.handleDelete(collection, uri) - } - } - - if commit.Seq > 0 { - if err := i.db.SetCursor("firehose_cursor", commit.Seq); err != nil { - log.Printf("Failed to save cursor: %v", err) } } } -func parseOpPath(path string) (collection, rkey string) { - for i := len(path) - 1; i >= 0; i-- { - if path[i] == '/' { - return path[:i], path[i+1:] +func (i *Ingester) handleCommit(event JetstreamEvent) { + commit := event.Commit + uri := fmt.Sprintf("at://%s/%s/%s", event.Did, commit.Collection, commit.Rkey) + + switch commit.Operation { + case "create", "update": + if len(commit.Record) > 0 { + firehoseEvent := &FirehoseEvent{ + Repo: event.Did, + Collection: commit.Collection, + Rkey: commit.Rkey, + Record: commit.Record, + Operation: commit.Operation, + Cursor: event.Time, + } + + i.dispatchToHandler(firehoseEvent) + + go i.triggerLazySync(event.Did) } + case "delete": + i.handleDelete(commit.Collection, uri) } - return path, "" } -func isMarginCollection(collection string) bool { - switch collection { - case CollectionAnnotation, CollectionHighlight, CollectionBookmark, - CollectionReply, CollectionLike, CollectionCollection, CollectionCollectionItem: - return true +func (i *Ingester) dispatchToHandler(event *FirehoseEvent) { + if handler, ok := i.handlers[event.Collection]; ok { + handler(event) } - return false } -func extractRecord(blocks []byte, targetCid cid.Cid) map[string]interface{} { - reader := bytes.NewReader(blocks) - - headerLen, err := binary.ReadUvarint(reader) - if err != nil { - return nil - } - reader.Seek(int64(headerLen), io.SeekCurrent) - - for reader.Len() > 0 { - blockLen, err := binary.ReadUvarint(reader) - if err != nil { - break - } - - blockData := make([]byte, blockLen) - if _, err := io.ReadFull(reader, blockData); err != nil { - break - } - - blockCid, cidLen, err := parseCidFromBlock(blockData) - if err != nil { - continue - } +var lastSyncAttempts sync.Map - if blockCid.Equals(targetCid) { - var record map[string]interface{} - if err := cbor.Unmarshal(blockData[cidLen:], &record); err != nil { - return nil - } - return record +func (i *Ingester) triggerLazySync(did string) { + lastSync, ok := lastSyncAttempts.Load(did) + if ok { + if time.Since(lastSync.(time.Time)) < 5*time.Minute { + return } } + lastSyncAttempts.Store(did, time.Now()) - return nil -} - -func parseCidFromBlock(data []byte) (cid.Cid, int, error) { - if len(data) < 2 { - return cid.Cid{}, 0, fmt.Errorf("data too short") - } - version, n1 := binary.Uvarint(data) - if n1 <= 0 { - return cid.Cid{}, 0, fmt.Errorf("invalid version varint") + pds, err := xrpc.ResolveDIDToPDS(did) + if err != nil || pds == "" { + return } - if version == 1 { - codec, n2 := binary.Uvarint(data[n1:]) - if n2 <= 0 { - return cid.Cid{}, 0, fmt.Errorf("invalid codec varint") - } - - mhStart := n1 + n2 - hashType, n3 := binary.Uvarint(data[mhStart:]) - if n3 <= 0 { - return cid.Cid{}, 0, fmt.Errorf("invalid hash type varint") - } - - hashLen, n4 := binary.Uvarint(data[mhStart+n3:]) - if n4 <= 0 { - return cid.Cid{}, 0, fmt.Errorf("invalid hash length varint") - } - - totalCidLen := mhStart + n3 + n4 + int(hashLen) - - c, err := cid.Cast(data[:totalCidLen]) - if err != nil { - return cid.Cid{}, 0, err - } - - _ = codec - _ = hashType + _, err = i.sync.PerformSync(context.Background(), did, func(ctx context.Context, _ string) (*xrpc.Client, error) { + return &xrpc.Client{ + PDS: pds, + }, nil + }) - return c, totalCidLen, nil + if err == nil { + log.Printf("Auto-synced repo for active user: %s", did) } - - return cid.Cid{}, 0, fmt.Errorf("unsupported CID version") } func (i *Ingester) handleDelete(collection, uri string) { @@ -289,39 +234,13 @@ func (i *Ingester) handleDelete(collection, uri string) { } } -func (i *Ingester) handleRecord(repo, collection, rkey string, record map[string]interface{}, seq int64) { - _ = fmt.Sprintf("at://%s/%s/%s", repo, collection, rkey) - - recordJSON, err := json.Marshal(record) +func (i *Ingester) getLastCursor() int64 { + cursor, err := i.db.GetCursor("firehose_cursor") if err != nil { - return - } - - event := &FirehoseEvent{ - Repo: repo, - Collection: collection, - Rkey: rkey, - Record: recordJSON, - Operation: "create", - Cursor: seq, - } - - switch collection { - case CollectionAnnotation: - i.handleAnnotation(event) - case CollectionHighlight: - i.handleHighlight(event) - case CollectionBookmark: - i.handleBookmark(event) - case CollectionReply: - i.handleReply(event) - case CollectionLike: - i.handleLike(event) - case CollectionCollection: - i.handleCollection(event) - case CollectionCollectionItem: - i.handleCollectionItem(event) + log.Printf("Failed to get last cursor from DB: %v", err) + return 0 } + return cursor } type FirehoseEvent struct { @@ -711,12 +630,3 @@ func (i *Ingester) handleCollectionItem(event *FirehoseEvent) { log.Printf("Indexed collection item from %s", event.Repo) } } - -func (i *Ingester) getLastCursor() int64 { - cursor, err := i.db.GetCursor("firehose_cursor") - if err != nil { - log.Printf("Failed to get last cursor from DB: %v", err) - return 0 - } - return cursor -} diff --git a/backend/internal/oauth/handler.go b/backend/internal/oauth/handler.go index f2aa410..f64e1c4 100644 --- a/backend/internal/oauth/handler.go +++ b/backend/internal/oauth/handler.go @@ -17,6 +17,7 @@ import ( "time" "margin.at/internal/db" + internal_sync "margin.at/internal/sync" "margin.at/internal/xrpc" ) @@ -26,9 +27,10 @@ type Handler struct { privateKey *ecdsa.PrivateKey pending map[string]*PendingAuth pendingMu sync.RWMutex + syncService *internal_sync.Service } -func NewHandler(database *db.DB) (*Handler, error) { +func NewHandler(database *db.DB, syncService *internal_sync.Service) (*Handler, error) { configuredBaseURL := os.Getenv("BASE_URL") @@ -42,6 +44,7 @@ func NewHandler(database *db.DB) (*Handler, error) { configuredBaseURL: configuredBaseURL, privateKey: privateKey, pending: make(map[string]*PendingAuth), + syncService: syncService, }, nil } @@ -364,6 +367,18 @@ func (h *Handler) HandleCallback(w http.ResponseWriter, r *http.Request) { }) go h.cleanupOrphanedReplies(tokenResp.Sub, tokenResp.AccessToken, string(dpopKeyPEM), pending.PDS) + go func() { + log.Printf("Starting background sync for %s...", tokenResp.Sub) + _, err := h.syncService.PerformSync(context.Background(), tokenResp.Sub, func(ctx context.Context, did string) (*xrpc.Client, error) { + return xrpc.NewClient(pending.PDS, tokenResp.AccessToken, pending.DPoPKey), nil + }) + + if err != nil { + log.Printf("Background sync failed for %s: %v", tokenResp.Sub, err) + } else { + log.Printf("Background sync completed for %s", tokenResp.Sub) + } + }() http.Redirect(w, r, "/?logged_in=true", http.StatusFound) } diff --git a/backend/internal/sync/service.go b/backend/internal/sync/service.go new file mode 100644 index 0000000..7a01967 --- /dev/null +++ b/backend/internal/sync/service.go @@ -0,0 +1,428 @@ +package sync + +import ( + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "time" + + "margin.at/internal/db" + "margin.at/internal/xrpc" +) + +type Service struct { + db *db.DB +} + +func NewService(database *db.DB) *Service { + return &Service{db: database} +} + +func (s *Service) PerformSync(ctx context.Context, did string, getClient func(context.Context, string) (*xrpc.Client, error)) (map[string]string, error) { + collections := []string{ + xrpc.CollectionAnnotation, + xrpc.CollectionHighlight, + xrpc.CollectionBookmark, + xrpc.CollectionReply, + xrpc.CollectionLike, + xrpc.CollectionCollection, + xrpc.CollectionCollectionItem, + } + + results := make(map[string]string) + + client, err := getClient(ctx, did) + if err != nil { + return nil, err + } + + for _, collectionNSID := range collections { + count := 0 + cursor := "" + fetchedURIs := make(map[string]bool) + + for { + url := fmt.Sprintf("%s/xrpc/com.atproto.repo.listRecords?repo=%s&collection=%s&limit=100", client.PDS, did, collectionNSID) + if cursor != "" { + url += "&cursor=" + cursor + } + + req, _ := http.NewRequestWithContext(ctx, "GET", url, nil) + req.Header.Set("Authorization", "Bearer "+client.AccessToken) + + resp, err := http.DefaultClient.Do(req) + if err != nil { + return nil, fmt.Errorf("failed to fetch %s: %w", collectionNSID, err) + } + defer resp.Body.Close() + + if resp.StatusCode != 200 { + body, _ := io.ReadAll(resp.Body) + results[collectionNSID] = fmt.Sprintf("error: %s", string(body)) + break + } + + var output struct { + Records []struct { + URI string `json:"uri"` + CID string `json:"cid"` + Value json.RawMessage `json:"value"` + } `json:"records"` + Cursor string `json:"cursor"` + } + + if err := json.NewDecoder(resp.Body).Decode(&output); err != nil { + return nil, err + } + + for _, rec := range output.Records { + err := s.upsertRecord(did, collectionNSID, rec.URI, rec.CID, rec.Value) + if err != nil { + fmt.Printf("Error upserting %s: %v\n", rec.URI, err) + } else { + count++ + fetchedURIs[rec.URI] = true + } + } + + if output.Cursor == "" { + break + } + cursor = output.Cursor + } + + deletedCount := 0 + if results[collectionNSID] == "" { + var localURIs []string + var err error + + switch collectionNSID { + case xrpc.CollectionAnnotation: + localURIs, err = s.db.GetAnnotationURIs(did) + case xrpc.CollectionHighlight: + localURIs, err = s.db.GetHighlightURIs(did) + case xrpc.CollectionBookmark: + localURIs, err = s.db.GetBookmarkURIs(did) + case xrpc.CollectionCollection: + cols, e := s.db.GetCollectionsByAuthor(did) + if e == nil { + for _, c := range cols { + localURIs = append(localURIs, c.URI) + } + } else { + err = e + } + case xrpc.CollectionCollectionItem: + items, e := s.db.GetCollectionItemsByAuthor(did) + if e == nil { + for _, item := range items { + localURIs = append(localURIs, item.URI) + } + } else { + err = e + } + case xrpc.CollectionReply: + replies, e := s.db.GetRepliesByAuthor(did) + if e == nil { + for _, r := range replies { + localURIs = append(localURIs, r.URI) + } + } else { + err = e + } + case xrpc.CollectionLike: + likes, e := s.db.GetLikesByAuthor(did) + if e == nil { + for _, l := range likes { + localURIs = append(localURIs, l.URI) + } + } else { + err = e + } + } + + if err == nil { + for _, uri := range localURIs { + if !fetchedURIs[uri] { + switch collectionNSID { + case xrpc.CollectionAnnotation: + _ = s.db.DeleteAnnotation(uri) + case xrpc.CollectionHighlight: + _ = s.db.DeleteHighlight(uri) + case xrpc.CollectionBookmark: + _ = s.db.DeleteBookmark(uri) + case xrpc.CollectionCollection: + _ = s.db.DeleteCollection(uri) + case xrpc.CollectionCollectionItem: + _ = s.db.RemoveFromCollection(uri) + case xrpc.CollectionReply: + _ = s.db.DeleteReply(uri) + case xrpc.CollectionLike: + _ = s.db.DeleteLike(uri) + } + deletedCount++ + } + } + } + } + + if results[collectionNSID] == "" { + results[collectionNSID] = fmt.Sprintf("synced %d records, deleted %d stale", count, deletedCount) + } + } + return results, nil +} + +func strPtr(s string) *string { + if s == "" { + return nil + } + return &s +} + +func (s *Service) upsertRecord(did, collection, uri, cid string, value json.RawMessage) error { + cidPtr := strPtr(cid) + switch collection { + case xrpc.CollectionAnnotation: + var record xrpc.AnnotationRecord + if err := json.Unmarshal(value, &record); err != nil { + return err + } + + createdAt, _ := time.Parse(time.RFC3339, record.CreatedAt) + + targetSource := record.Target.Source + if targetSource == "" { + + } + + targetHash := record.Target.SourceHash + if targetHash == "" && targetSource != "" { + targetHash = db.HashURL(targetSource) + } + + motivation := record.Motivation + if motivation == "" { + motivation = "commenting" + } + + var bodyValuePtr, bodyFormatPtr, bodyURIPtr, targetTitlePtr, selectorJSONPtr, tagsJSONPtr *string + if record.Body != nil { + if record.Body.Value != "" { + val := record.Body.Value + bodyValuePtr = &val + } + if record.Body.Format != "" { + fmt := record.Body.Format + bodyFormatPtr = &fmt + } + } + if record.Target.Title != "" { + t := record.Target.Title + targetTitlePtr = &t + } + if len(record.Target.Selector) > 0 { + selectorStr := string(record.Target.Selector) + selectorJSONPtr = &selectorStr + } + if len(record.Tags) > 0 { + tagsBytes, _ := json.Marshal(record.Tags) + tagsStr := string(tagsBytes) + tagsJSONPtr = &tagsStr + } + + return s.db.CreateAnnotation(&db.Annotation{ + URI: uri, + AuthorDID: did, + Motivation: motivation, + BodyValue: bodyValuePtr, + BodyFormat: bodyFormatPtr, + BodyURI: bodyURIPtr, + TargetSource: targetSource, + TargetHash: targetHash, + TargetTitle: targetTitlePtr, + SelectorJSON: selectorJSONPtr, + TagsJSON: tagsJSONPtr, + CreatedAt: createdAt, + IndexedAt: time.Now(), + CID: cidPtr, + }) + + case xrpc.CollectionHighlight: + var record xrpc.HighlightRecord + if err := json.Unmarshal(value, &record); err != nil { + return err + } + + createdAt, _ := time.Parse(time.RFC3339, record.CreatedAt) + if createdAt.IsZero() { + createdAt = time.Now() + } + + targetHash := record.Target.SourceHash + if targetHash == "" && record.Target.Source != "" { + targetHash = db.HashURL(record.Target.Source) + } + + var titlePtr, selectorJSONPtr, colorPtr, tagsJSONPtr *string + if record.Target.Title != "" { + t := record.Target.Title + titlePtr = &t + } + if len(record.Target.Selector) > 0 { + selectorStr := string(record.Target.Selector) + selectorJSONPtr = &selectorStr + } + if record.Color != "" { + c := record.Color + colorPtr = &c + } + if len(record.Tags) > 0 { + tagsBytes, _ := json.Marshal(record.Tags) + tagsStr := string(tagsBytes) + tagsJSONPtr = &tagsStr + } + + return s.db.CreateHighlight(&db.Highlight{ + URI: uri, + AuthorDID: did, + TargetSource: record.Target.Source, + TargetHash: targetHash, + TargetTitle: titlePtr, + SelectorJSON: selectorJSONPtr, + Color: colorPtr, + TagsJSON: tagsJSONPtr, + CreatedAt: createdAt, + IndexedAt: time.Now(), + CID: cidPtr, + }) + + case xrpc.CollectionBookmark: + var record xrpc.BookmarkRecord + if err := json.Unmarshal(value, &record); err != nil { + return err + } + + createdAt, _ := time.Parse(time.RFC3339, record.CreatedAt) + + sourceHash := record.SourceHash + if sourceHash == "" && record.Source != "" { + sourceHash = db.HashURL(record.Source) + } + + var titlePtr, descPtr, tagsJSONPtr *string + if record.Title != "" { + t := record.Title + titlePtr = &t + } + if record.Description != "" { + d := record.Description + descPtr = &d + } + if len(record.Tags) > 0 { + tagsBytes, _ := json.Marshal(record.Tags) + tagsStr := string(tagsBytes) + tagsJSONPtr = &tagsStr + } + + return s.db.CreateBookmark(&db.Bookmark{ + URI: uri, + AuthorDID: did, + Source: record.Source, + SourceHash: sourceHash, + Title: titlePtr, + Description: descPtr, + TagsJSON: tagsJSONPtr, + CreatedAt: createdAt, + IndexedAt: time.Now(), + CID: cidPtr, + }) + + case xrpc.CollectionCollection: + var record xrpc.CollectionRecord + if err := json.Unmarshal(value, &record); err != nil { + return err + } + createdAt, _ := time.Parse(time.RFC3339, record.CreatedAt) + + var descPtr, iconPtr *string + if record.Description != "" { + d := record.Description + descPtr = &d + } + if record.Icon != "" { + i := record.Icon + iconPtr = &i + } + + return s.db.CreateCollection(&db.Collection{ + URI: uri, + AuthorDID: did, + Name: record.Name, + Description: descPtr, + Icon: iconPtr, + CreatedAt: createdAt, + IndexedAt: time.Now(), + }) + + case xrpc.CollectionCollectionItem: + var record xrpc.CollectionItemRecord + if err := json.Unmarshal(value, &record); err != nil { + return err + } + createdAt, _ := time.Parse(time.RFC3339, record.CreatedAt) + + return s.db.AddToCollection(&db.CollectionItem{ + URI: uri, + AuthorDID: did, + CollectionURI: record.Collection, + AnnotationURI: record.Annotation, + Position: record.Position, + CreatedAt: createdAt, + IndexedAt: time.Now(), + }) + + case xrpc.CollectionReply: + var record xrpc.ReplyRecord + if err := json.Unmarshal(value, &record); err != nil { + return err + } + createdAt, _ := time.Parse(time.RFC3339, record.CreatedAt) + + var formatPtr *string + if record.Format != "" { + f := record.Format + formatPtr = &f + } + + return s.db.CreateReply(&db.Reply{ + URI: uri, + AuthorDID: did, + ParentURI: record.Parent.URI, + RootURI: record.Root.URI, + Text: record.Text, + Format: formatPtr, + CreatedAt: createdAt, + IndexedAt: time.Now(), + CID: cidPtr, + }) + + case xrpc.CollectionLike: + var record xrpc.LikeRecord + if err := json.Unmarshal(value, &record); err != nil { + return err + } + createdAt, _ := time.Parse(time.RFC3339, record.CreatedAt) + + return s.db.CreateLike(&db.Like{ + URI: uri, + AuthorDID: did, + SubjectURI: record.Subject.URI, + CreatedAt: createdAt, + IndexedAt: time.Now(), + }) + } + return nil +} diff --git a/backend/internal/xrpc/utils.go b/backend/internal/xrpc/utils.go new file mode 100644 index 0000000..d95bae6 --- /dev/null +++ b/backend/internal/xrpc/utils.go @@ -0,0 +1,51 @@ +package xrpc + +import ( + "encoding/json" + "fmt" + "net/http" + "strings" + "time" +) + +func ResolveDIDToPDS(did string) (string, error) { + var docURL string + if strings.HasPrefix(did, "did:plc:") { + docURL = fmt.Sprintf("https://plc.directory/%s", did) + } else if strings.HasPrefix(did, "did:web:") { + domain := strings.TrimPrefix(did, "did:web:") + docURL = fmt.Sprintf("https://%s/.well-known/did.json", domain) + } else { + return "", nil + } + + client := &http.Client{ + Timeout: 10 * time.Second, + } + resp, err := client.Get(docURL) + if err != nil { + return "", err + } + defer resp.Body.Close() + + if resp.StatusCode != 200 { + return "", fmt.Errorf("failed to fetch DID doc: %d", resp.StatusCode) + } + + var doc struct { + Service []struct { + Type string `json:"type"` + ServiceEndpoint string `json:"serviceEndpoint"` + } `json:"service"` + } + if err := json.NewDecoder(resp.Body).Decode(&doc); err != nil { + return "", err + } + + for _, svc := range doc.Service { + if svc.Type == "AtprotoPersonalDataServer" { + return svc.ServiceEndpoint, nil + } + } + return "", nil +} diff --git a/web/src/components/SignUpModal.jsx b/web/src/components/SignUpModal.jsx index 058bc4a..127c0a0 100644 --- a/web/src/components/SignUpModal.jsx +++ b/web/src/components/SignUpModal.jsx @@ -122,9 +122,10 @@ export default function SignUpModal({ onClose }) { domain = "." + domain; } - const fullHandle = formData.handle.endsWith(domain) - ? formData.handle - : `${formData.handle}${domain}`; + const cleanHandle = formData.handle.trim().replace(/^@/, ""); + const fullHandle = cleanHandle.endsWith(domain) + ? cleanHandle + : `${cleanHandle}${domain}`; try { await createAccount(serverInfo.service, {