diff options
Diffstat (limited to 'backend/fetchers')
| -rw-r--r-- | backend/fetchers/app.bsky.actor.go | 184 | ||||
| -rw-r--r-- | backend/fetchers/app.bsky.feed.go | 217 | ||||
| -rw-r--r-- | backend/fetchers/cache.go | 166 | ||||
| -rw-r--r-- | backend/fetchers/config.go | 10 |
4 files changed, 0 insertions, 577 deletions
diff --git a/backend/fetchers/app.bsky.actor.go b/backend/fetchers/app.bsky.actor.go deleted file mode 100644 index 41a9c17..0000000 --- a/backend/fetchers/app.bsky.actor.go +++ /dev/null @@ -1,184 +0,0 @@ -package fetchers - -import ( - "encoding/json" - "fmt" - "io" - "llostre/backend/auth" - "llostre/backend/general" - "llostre/backend/notes" - "net" - "net/http" - "strings" - "time" -) - -type DIDDocument struct { - AlsoKnownAs []string `json:"alsoKnownAs"` - Service []struct { - ID string `json:"id"` - Type string `json:"type"` - ServiceEndpoint string `json:"serviceEndpoint"` - } `json:"service"` -} - -func FetchDID(handle string) (string, error) { - // return if it's a DID already - if strings.HasPrefix(handle, "did:") { - return handle, nil - } - - // resolve DNS record first - txtRecords, err := net.LookupTXT("_atproto." + handle) - if err == nil { - for _, txt := range txtRecords { - if after, ok := strings.CutPrefix(txt, "did="); ok { - return after, nil - } - } - } - - // fallback to HTTPS well-known - resp, err := http.Get(fmt.Sprintf("https://%s/.well-known/atproto-did", handle)) - if err != nil { - return "", err - } - defer resp.Body.Close() - - body, err := io.ReadAll(resp.Body) - if err != nil { - return "", err - } - - did := strings.TrimSpace(string(body)) - if !strings.HasPrefix(did, "did:") { - return "", fmt.Errorf("invalid DID returned: %s", did) - } - - return did, nil -} - -func FetchDidDoc(did string) (*DIDDocument, error) { - didDocURL := fmt.Sprintf("https://plc.directory/%s", did) - - resp, err := http.Get(didDocURL) - if err != nil { - return nil, err - } - defer resp.Body.Close() - - if resp.StatusCode != 200 { - return nil, fmt.Errorf("DID resolution failed: %d", resp.StatusCode) - } - - var didDoc DIDDocument - if err := json.NewDecoder(resp.Body).Decode(&didDoc); err != nil { - return nil, err - } - - return &didDoc, nil -} - -func FetchHandle(did string) (string, error) { - var didDoc, err = FetchDidDoc(did) - if err != nil { - return "", err - } - - if len(didDoc.AlsoKnownAs) == 0 { - return "", fmt.Errorf("no handle found") - } - - handle := strings.TrimPrefix(didDoc.AlsoKnownAs[0], "at://") - return handle, nil -} - -func FetchPDS(did string) (string, error) { - var didDoc, err = FetchDidDoc(did) - if err != nil { - return "", err - } - - for _, svc := range didDoc.Service { - if svc.Type == "AtprotoPersonalDataServer" { - return svc.ServiceEndpoint, nil - } - } - - return "", fmt.Errorf("no PDS found for DID %s", did) -} - -func FetchHandleAndPDS(did string) (string, string, error) { - var didDoc, err = FetchDidDoc(did) - if err != nil { - return "", "", err - } - - if len(didDoc.AlsoKnownAs) == 0 { - return "", "", fmt.Errorf("no handle found") - } - - var pds string - for _, svc := range didDoc.Service { - if svc.Type == "AtprotoPersonalDataServer" { - pds = svc.ServiceEndpoint - break - } - } - - handle := strings.TrimPrefix(didDoc.AlsoKnownAs[0], "at://") - - return handle, pds, nil -} - -func FetchAppBskyActorProfile(conf *Config, did, pdsURL string, session *auth.OAuthState) (*notes.Author, error) { - // construct profile record URI - uri := fmt.Sprintf("at://%s/app.bsky.actor.profile/self", did) - - // try slingshot first - if conf.SlingshotURL != "" && general.SlingshotAvailable() { - rawRecord, err := general.FetchRawRecordByURI(conf.Logger, uri, conf.SlingshotURL, session) - if err != nil { - conf.Logger.Warn("slingshot profile fetch failed", "did", did, "error", err) - general.SetSlingshotBackoff(5 * time.Minute) - } else { - return parseAppBskyActorProfile(did, pdsURL, rawRecord, conf.Cache) - } - } - - // fallback to PDS - rawRecord, err := general.FetchRawRecordByURI(conf.Logger, uri, pdsURL, session) - if err != nil { - return nil, err - } - - return parseAppBskyActorProfile(did, pdsURL, rawRecord, conf.Cache) -} - -func FetchAppBskyActorProfiles(conf *Config, dids []string, pdsURL string, session *auth.OAuthState) map[string]notes.Author { - authors := make(map[string]notes.Author) - for _, did := range dids { - author, err := FetchAppBskyActorProfile(conf, did, pdsURL, session) - if err == nil { - authors[did] = *author - } - } - return authors -} - -func parseAppBskyActorProfile(did, pds string, rawRecord *general.RawRecord, cache *Cache) (*notes.Author, error) { - var profile notes.AppBskyActorProfileRecord - if err := json.Unmarshal(rawRecord.Value, &profile); err != nil { - return nil, err - } - - handle, err := FetchHandle(did) - if err != nil { - handle = "handle.invalid" - } - - author, err := notes.AuthorFromAppBskyActorProfileRecord(profile, did, handle, pds) - - cache.SetAuthor(did, author) - return author, nil -} diff --git a/backend/fetchers/app.bsky.feed.go b/backend/fetchers/app.bsky.feed.go deleted file mode 100644 index 8562a9c..0000000 --- a/backend/fetchers/app.bsky.feed.go +++ /dev/null @@ -1,217 +0,0 @@ -package fetchers - -import ( - "encoding/json" - "fmt" - "llostre/backend/auth" - "llostre/backend/general" - "llostre/backend/notes" - "time" -) - -func FetchAppBskyFeedPosts(conf *Config, did, pdsURL string, session *auth.OAuthState) ([]notes.Note, error) { - // get author of the note - author, found := conf.Cache.GetAuthor(did) - var err error - if !found { - author, err = FetchAppBskyActorProfile(conf, did, pdsURL, session) - } - if err != nil { - return nil, err - } - - // try slingshot first if given - if conf.SlingshotURL != "" && general.SlingshotAvailable() { - records, err := general.FetchRawRecordCollection(conf.Logger, did, "app.bsky.feed.post", conf.SlingshotURL, session) - if err != nil { - conf.Logger.Error("fetching from slingshot failed, falling back to PDS") - general.SetSlingshotBackoff(5 * time.Minute) - } else { - return parseAppBskyFeedPostCollection(conf, session, records, author) - } - } - - records, err := general.FetchRawRecordCollection(conf.Logger, did, "app.bsky.feed.post", pdsURL, session) - if err != nil { - return nil, err - } - - return parseAppBskyFeedPostCollection(conf, session, records, author) -} - -func FetchAppBskyFeedPostsWithCursor(conf *Config, did, pdsURL string, session *auth.OAuthState, cursor string) ([]notes.Note, error) { - author, found := conf.Cache.GetAuthor(did) - var err error - if !found { - author, err = FetchAppBskyActorProfile(conf, did, pdsURL, session) - } - if err != nil { - return nil, err - } - - if conf.SlingshotURL != "" && general.SlingshotAvailable() { - records, nextCursor, err := general.FetchRawRecordCollectionWithCursor(conf.Logger, did, "app.bsky.feed.post", conf.SlingshotURL, session, cursor, conf.PageSize) - if err != nil { - conf.Logger.Error("fetching from slingshot failed, falling back to PDS") - general.SetSlingshotBackoff(5 * time.Minute) - } else { - notes, err := parseAppBskyFeedPostCollection(conf, session, records, author) - conf.Cache.SetCursor(did, nextCursor) - return notes, err - } - } - - records, nextCursor, err := general.FetchRawRecordCollectionWithCursor(conf.Logger, did, "app.bsky.feed.post", pdsURL, session, cursor, conf.PageSize) - if err != nil { - return nil, err - } - - notes, err := parseAppBskyFeedPostCollection(conf, session, records, author) - - conf.Cache.SetCursor(did, nextCursor) - return notes, err -} - -func GetAppBskyFeedPostByURI(conf *Config, uri string, session *auth.OAuthState) (*notes.Note, error) { - // check cache, return if it was cached without auth or if it was cached with auth and we have auth - conf.Logger.Debug("Getting app.bsky.feed.post", "uri", uri) - if cached, found := conf.Cache.Get(uri); found { - conf.Logger.Debug("Found in the cache", "public", !cached.AuthorNoUnauth) - if !cached.AuthorNoUnauth { - return cached, nil - } - - // author requires auth - if we have a session, serve it - // if the token was invalid, trust that it would have been caught by the "refresh token" call in main - if session != nil { - return cached, nil - } - } - - return FetchAppBskyFeedPostByURI(conf, uri, session) -} - -func FetchAppBskyFeedPostByURI(conf *Config, uri string, session *auth.OAuthState) (*notes.Note, error) { - repo, _, _, err := general.ParseURI(uri) - - // need to resolve DID to PDS - pdsURL, err := FetchPDS(repo) - if err != nil { - return nil, err - } - conf.Logger.Debug("fetched origin PDS", "did", repo, "pdsURL", pdsURL) - - // get author of the note - author, found := conf.Cache.GetAuthor(repo) - if !found { - author, err = FetchAppBskyActorProfile(conf, repo, pdsURL, session) - } - if err != nil { - return nil, err - } - - // try slingshot first if given - if conf.SlingshotURL != "" && general.SlingshotAvailable() { - rawRecord, err := general.FetchRawRecordByURI(conf.Logger, uri, conf.SlingshotURL, session) - if err != nil { - conf.Logger.Error("fetching from slingshot failed", "error", err) - general.SetSlingshotBackoff(5 * time.Minute) - } else { - return parseAppBskyFeedPostRecord(rawRecord, author, conf) - } - } - - rawRecord, err := general.FetchRawRecordByURI(conf.Logger, uri, pdsURL, session) - if err != nil { - conf.Logger.Error("Fetching from PDS failed") - return nil, err - } - - return parseAppBskyFeedPostRecord(rawRecord, author, conf) -} - -func parseAppBskyFeedPostRecord(rawRecord *general.RawRecord, author *notes.Author, conf *Config) (*notes.Note, error) { - if rawRecord == nil { - return nil, fmt.Errorf("unable to parse nil record") - } - - record := notes.AppBskyFeedPostRecord{ - URI: rawRecord.URI, - } - err := json.Unmarshal(rawRecord.Value, &record.Value) - if err != nil { - return nil, err - } - - note, err := notes.NoteFromAppBskyFeedPostRecord(record, author) - if err != nil { - return nil, err - } - - // cache it - conf.Cache.Set(note.URI, note) - - conf.Logger.Debug("parsed note", "uri", note.URI, "media_count", len(note.Media), "embed", record.Value.Embed) - - return note, nil -} - -func parseAppBskyFeedPostCollection(conf *Config, session *auth.OAuthState, records []general.RawRecord, author *notes.Author) ([]notes.Note, error) { - posts := make([]notes.Note, 0, len(records)) - for _, record := range records { - note, err := parseAppBskyFeedPostRecord(&record, author, conf) - if err != nil { - conf.Logger.Error("failed to parse post record", "error", err) - continue - } - if note.ReplyTo != nil { - parents, err := FetchParents(conf, *note.ReplyTo, session, 10) - if err != nil { - conf.Logger.Warn("failed to fetch parent chain", "uri", note.URI, "error", err) - } - note.Parents = parents - } - conf.Cache.Set(note.URI, note) - posts = append(posts, *note) - } - - return notes.DeduplicateThreads(posts), nil -} - -func FetchParents(conf *Config, replyToURI string, session *auth.OAuthState, maxDepth int) ([]notes.Note, error) { - parents := []notes.Note{} - currentURI := replyToURI - - for currentURI != "" { - note, err := GetAppBskyFeedPostByURI(conf, currentURI, session) - if err != nil { - conf.Logger.Warn("failed to fetch parent", "uri", currentURI, "error", err) - note = notes.MissingNote(currentURI) - - conf.Cache.Set(currentURI, note) - parents = append(parents, *note) - break - } - - // preemptively fetch posts from author to avoid hammering servers if there's a back and forth - if _, found := conf.Cache.GetAuthor(note.AuthorDID); !found { - conf.Logger.Debug("caching posts from thread participant", "did", note.AuthorDID) - authorPDS, err := FetchPDS(note.AuthorDID) - if err != nil { - conf.Logger.Warn("failed to fetch parent", "uri", currentURI, "error", err) - break - } - go FetchAppBskyFeedPostsWithCursor(conf, note.AuthorDID, authorPDS, session, "") - } - - parents = append(parents, *note) - - if note.ReplyTo == nil { - break - } - - currentURI = *note.ReplyTo - } - - return parents, nil -} diff --git a/backend/fetchers/cache.go b/backend/fetchers/cache.go deleted file mode 100644 index ac91259..0000000 --- a/backend/fetchers/cache.go +++ /dev/null @@ -1,166 +0,0 @@ -package fetchers - -import ( - "context" - "encoding/json" - "fmt" - "llostre/backend/notes" - "sort" - "time" - - "github.com/redis/go-redis/v9" -) - -type Cache struct { - client *redis.Client - ctx context.Context - expiryTime time.Duration -} - -func NewCache(addr, password string, expiryMinutes int) (*Cache, error) { - client := redis.NewClient(&redis.Options{ - Addr: addr, - Password: password, - DB: 0, - }) - - ctx := context.Background() - if err := client.Ping(ctx).Err(); err != nil { - return nil, fmt.Errorf("redis connection failed: %w", err) - } - - return &Cache{ - client: client, - ctx: ctx, - expiryTime: time.Duration(expiryMinutes) * time.Minute, - }, nil -} - -func (c *Cache) Get(uri string) (*notes.Note, bool) { - key := fmt.Sprintf("post:%s", uri) - data, err := c.client.Get(c.ctx, key).Bytes() - if err != nil { - return nil, false - } - - var note notes.Note - if err := json.Unmarshal(data, ¬e); err != nil { - return nil, false - } - - return ¬e, true -} - -func (c *Cache) Set(uri string, note *notes.Note) { - key := fmt.Sprintf("post:%s", uri) - data, err := json.Marshal(note) - if err != nil { - return - } - - c.client.Set(c.ctx, key, data, c.expiryTime) -} - -func (c *Cache) GetAuthor(did string) (*notes.Author, bool) { - key := fmt.Sprintf("profile:%s", did) - data, err := c.client.Get(c.ctx, key).Bytes() - if err != nil { - return nil, false - } - - var author notes.Author - if err := json.Unmarshal(data, &author); err != nil { - return nil, false - } - - return &author, true -} - -func (c *Cache) SetAuthor(did string, author *notes.Author) { - key := fmt.Sprintf("profile:%s", did) - data, err := json.Marshal(author) - if err != nil { - return - } - - c.client.Set(c.ctx, key, data, c.expiryTime) -} - -func (c *Cache) GetCursor(did string) (string, bool) { - key := fmt.Sprintf("cursor:%s", did) - cursor, err := c.client.Get(c.ctx, key).Result() - if err != nil { - return "", false - } - return cursor, true -} - -func (c *Cache) SetCursor(did, cursor string) { - key := fmt.Sprintf("cursor:%s", did) - if cursor == "" { - c.client.Del(c.ctx, key) - return - } - c.client.Set(c.ctx, key, cursor, 24*time.Hour) -} - -func (c *Cache) GetPostsByDID(did string) ([]notes.Note, error) { - pattern := fmt.Sprintf("post:at://%s/*", did) - var posts []notes.Note - - iter := c.client.Scan(c.ctx, 0, pattern, 0).Iterator() - for iter.Next(c.ctx) { - data, err := c.client.Get(c.ctx, iter.Val()).Bytes() - if err != nil { - continue - } - var note notes.Note - if err := json.Unmarshal(data, ¬e); err != nil { - continue - } - posts = append(posts, note) - } - - if err := iter.Err(); err != nil { - return nil, err - } - - sort.Slice(posts, func(i, j int) bool { - return posts[i].CreatedAt.After(posts[j].CreatedAt) - }) - - return posts, nil -} - -// track oldest sent post per profile -func (c *Cache) GetOldestSent(did string) (time.Time, bool) { - key := fmt.Sprintf("oldest_sent:%s", did) - ts, err := c.client.Get(c.ctx, key).Int64() - if err != nil { - return time.Time{}, false - } - return time.Unix(ts, 0), true -} - -func (c *Cache) SetOldestSent(did string, timestamp time.Time) { - key := fmt.Sprintf("oldest_sent:%s", did) - c.client.Set(c.ctx, key, timestamp.Unix(), 24*time.Hour) -} - -// count posts older than timestamp for a profile -func (c *Cache) CountPostsOlderThan(did string, timestamp time.Time) (int, error) { - posts, err := c.GetPostsByDID(did) - if err != nil { - return 0, err - } - for i, p := range posts { - if p.CreatedAt.Before(timestamp) { - return len(posts) - i, nil - } - } - return 0, nil -} - -func (c *Cache) Close() error { - return c.client.Close() -} diff --git a/backend/fetchers/config.go b/backend/fetchers/config.go deleted file mode 100644 index 090eb09..0000000 --- a/backend/fetchers/config.go +++ /dev/null @@ -1,10 +0,0 @@ -package fetchers - -import "log/slog" - -type Config struct { - SlingshotURL string - PageSize int - Logger *slog.Logger - Cache *Cache -} |
