diff options
Diffstat (limited to 'backend/main.go')
| -rw-r--r-- | backend/main.go | 117 |
1 files changed, 82 insertions, 35 deletions
diff --git a/backend/main.go b/backend/main.go index eb6d70b..27767b0 100644 --- a/backend/main.go +++ b/backend/main.go @@ -5,11 +5,12 @@ import ( "fmt" "llostre/backend/auth" "llostre/backend/fetchers" + "llostre/backend/notes" "log/slog" "net/http" "net/url" "os" - "sort" + "strconv" "strings" "sync" "time" @@ -25,6 +26,7 @@ var ( sessionsMutex sync.RWMutex sessions = make(map[string]*auth.OAuthState) cache *fetchers.Cache + pageSize int ) func init() { @@ -57,8 +59,28 @@ func init() { })) slingshotURL = os.Getenv("SLINGSHOT_URL") + ttlMin := os.Getenv("TTL_MIN") - cache = fetchers.NewCache() + var expiryMinutes int + if ttlMin != "" { + var err error + expiryMinutes, err = strconv.Atoi(ttlMin) + if err != nil { + panic(err) + } + } else { + expiryMinutes = 15 + } + redisAddr := os.Getenv("REDIS_ADDR") + redisPassword := os.Getenv("REDIS_PASSWORD") + postCache, err := fetchers.NewCache(redisAddr, redisPassword, expiryMinutes) + if err != nil { + logger.Error("failed to connect to redis", "error", err) + panic(err) + } + + pageSize = 50 + cache = postCache } func loginHandler(w http.ResponseWriter, r *http.Request) { @@ -249,32 +271,42 @@ func sessionHandler(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) } -func refreshTokens(session *auth.OAuthState, cookieValue string) error { +func getSession(cookieValue string) (*auth.OAuthState, error) { // Check for access token refresh - needsRefresh := time.Now().After(session.ExpiresAt) - pds := session.PDS - refreshToken := session.RefreshToken - keypair := session.KeyPair + sessionsMutex.RLock() + session, exists := sessions[cookieValue] + if !exists { + return nil, fmt.Errorf("no session found") + } + sessionRead := session.Copy() sessionsMutex.RUnlock() + logger.Debug("session found", "did", session.DID, "pds", session.PDS, "access token length", len(session.AccessToken), "session keypair", session.KeyPair) + if session.KeyPair != nil { + logger.Debug("session public key", "key", session.KeyPair.PublicKey) + } + + needsRefresh := time.Now().After(sessionRead.ExpiresAt) + if needsRefresh { - newTokens, err := auth.RefreshAccessToken(pds, refreshToken, backendURL+"/callback", keypair) + newTokens, err := auth.RefreshAccessToken(sessionRead.PDS, sessionRead.RefreshToken, backendURL+"/callback", sessionRead.KeyPair) if err != nil { sessionsMutex.Lock() delete(sessions, cookieValue) sessionsMutex.Unlock() logger.Error("no authorization servers found for token refresh") - return fmt.Errorf("no authorization servers found") + return nil, fmt.Errorf("no authorization servers found") } sessionsMutex.Lock() + session, _ = sessions[cookieValue] session.AccessToken = newTokens.AccessToken session.RefreshToken = newTokens.RefreshToken session.ExpiresAt = time.Now().Add(time.Duration(newTokens.ExpiresIn) * time.Second) sessionsMutex.Unlock() } - return nil + return session, nil } func postsHandler(w http.ResponseWriter, r *http.Request) { @@ -289,33 +321,50 @@ func postsHandler(w http.ResponseWriter, r *http.Request) { logger.Debug("cookie received", "value", cookie.Value) - sessionsMutex.RLock() - session, exists := sessions[cookie.Value] - if !exists { - sessionsMutex.RUnlock() + session, err := getSession(cookie.Value) + if err != nil { http.Error(w, "invalid session", http.StatusUnauthorized) return } - logger.Debug("session found", "did", session.DID, "pds", session.PDS, "access token length", len(session.AccessToken), "session keypair", session.KeyPair) - if session.KeyPair != nil { - logger.Debug("session public key", "key", session.KeyPair.PublicKey) - } - - refreshTokens(session, cookie.Value) - // Fetch app.bsky.feed.post records - posts, err := fetchers.FetchAppBskyFeedPosts(logger, session.DID, session.PDS, slingshotURL, session, cache) + var posts []notes.Note + + cached, err := cache.GetPostsByDID(session.DID) + logger.Debug("cache read", "count", len(cached), "err", err) if err != nil { - logger.Error("fetching app.bsky.feed.posts failed", "error", err) - http.Error(w, err.Error(), http.StatusInternalServerError) - return + logger.Warn("cache read failed, falling back to fetch", "error", err) } - sort.Slice(posts, func(i, j int) bool { - return posts[i].CreatedAt.After(posts[j].CreatedAt) - }) + if len(cached) > 0 { + posts = notes.DeduplicateThreads(cached) + } else { + posts, err = fetchers.FetchAppBskyFeedPostsWithCursor(logger, session.DID, session.PDS, slingshotURL, session, cache, "", pageSize) + if err != nil { + logger.Error("fetching app.bsky.feed.posts failed", "error", err) + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + } + logger.Debug("posts ready", "count", len(posts)) + + // background refetch if cache is running low + oldestSent, hasOldest := cache.GetOldestSent(session.DID) + if !hasOldest || len(posts) > 0 { + oldest := posts[len(posts)-1].CreatedAt + count, _ := cache.CountPostsOlderThan(session.DID, oldest) + cursor, hasCursor := cache.GetCursor(session.DID) + logger.Debug("refetch check", "count", count, "pageSize", pageSize, "hasCursor", hasCursor, "oldest", oldest) + if count < pageSize && hasCursor { + sessionCopy := session.Copy() + go fetchers.FetchAppBskyFeedPostsWithCursor(logger, sessionCopy.DID, sessionCopy.PDS, slingshotURL, sessionCopy, cache, cursor, pageSize) + } + if !hasOldest || oldest.Before(oldestSent) { + cache.SetOldestSent(session.DID, oldest) + } + } + // return posts logger.Debug("successfully fetched posts", "number", len(posts)) w.Header().Set("Access-Control-Allow-Origin", "*") @@ -342,15 +391,11 @@ func postHandler(w http.ResponseWriter, r *http.Request) { if err != nil { logger.Warn("unauthorized") } else { - sessionsMutex.RLock() - session, exists := sessions[cookie.Value] - sessionsMutex.RUnlock() + session, err := getSession(cookie.Value) - if !exists { - logger.Warn("invalid session") + if err != nil { + logger.Warn("invalid session", "error", err) } else { - refreshTokens(session, cookie.Value) - currentSession = session } } @@ -366,6 +411,8 @@ func postHandler(w http.ResponseWriter, r *http.Request) { } func main() { + defer cache.Close() + // serve static files fs := http.FileServer(http.Dir("../frontend/dist")) http.Handle("/", fs) |