aboutsummaryrefslogtreecommitdiff
path: root/backend/fetchers
diff options
context:
space:
mode:
authorGuillermo Serrahima <dev@vylion.com>2026-04-11 07:40:30 +0200
committerGuillermo Serrahima <dev@vylion.com>2026-04-11 23:39:19 +0200
commit9f0a9a6779f00786e4728ad5c3fad36814a293b0 (patch)
tree4b16d535c5d41ef46980608a88b7581be3c2bfd0 /backend/fetchers
parentf6d60077b7cfc0d589fa7961e2bb59aea9f993a3 (diff)
downloadllostre-9f0a9a6779f00786e4728ad5c3fad36814a293b0.tar.xz
llostre-9f0a9a6779f00786e4728ad5c3fad36814a293b0.zip
project restructure
Diffstat (limited to 'backend/fetchers')
-rw-r--r--backend/fetchers/app.bsky.actor.go184
-rw-r--r--backend/fetchers/app.bsky.feed.go217
-rw-r--r--backend/fetchers/cache.go166
-rw-r--r--backend/fetchers/config.go10
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, &note); err != nil {
- return nil, false
- }
-
- return &note, 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, &note); 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
-}