mirror of
https://github.com/R0m1k3/FlowReader.git
synced 2026-10-22 22:58:06 +02:00
fix(backend): harden security and speed up feeds and article API
Security: - WebSocket events are routed to their owner only (no cross-user leak); hub close is idempotent (fixes double-close panic), adds ping/pong and write deadlines. - Session tokens stored as SHA-256 (migration 008 keeps sessions valid); single-query auth middleware puts the user in the request context. - Client IP only trusts X-Forwarded-For from TRUSTED_PROXIES; rate limiter map is bounded; per-user limit on AI summaries. - Argon2id at OWASP minimum with a concurrency cap; constant-time login for unknown emails; atomic first-admin bootstrap; REGISTRATION_ENABLED. - CSP/HSTS/COOP headers, same-origin guard on mutations, body size limits, wider SSRF denylist, bounded feed/page/AI response reads, generic errors. - Upgrade chi, pgx, x/net, x/text, x/crypto (known CVEs); commit go.sum. Performance: - List endpoints return a plain-text excerpt and reading time instead of full HTML; content is sanitized once at ingest (legacy rows backfilled). - Keyset pagination on (sort_at, id) with matching partial indexes; redundant indexes dropped (migration 007). - Fetcher: bounded worker pool, conditional GET (ETag/Last-Modified), exponential backoff, dedupe before insert, column-safe truncation, retention-aware ingest, per-user refresh coalescing. - Read/favorite/read-all are single ownership-scoped statements. - gzip compression, immutable caching for hashed assets, path-safe SPA handler, server timeouts; expired sessions purged. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
d037e2be34
commit
03e57e4308
40 files changed
+2231
-1898
No files matched your search
+236
-54
@@ -2,8 +2,13 @@ package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log"
|
||||
"math/rand/v2"
|
||||
"net"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
@@ -12,25 +17,52 @@ import (
|
||||
"github.com/michael/flowreader/internal/ws"
|
||||
)
|
||||
|
||||
const (
|
||||
// FetchInterval is the normal delay between two fetches of a feed.
|
||||
FetchInterval = 15 * time.Minute
|
||||
// maxBackoff caps the retry delay of a failing feed.
|
||||
maxBackoff = 24 * time.Hour
|
||||
// ArticleRetention is how long non-favorite articles are kept. Items
|
||||
// older than this are not (re-)ingested, so the cleaner can't resurrect
|
||||
// old posts as unread.
|
||||
ArticleRetention = 30 * 24 * time.Hour
|
||||
// perFeedTimeout bounds a single feed download + parse + insert.
|
||||
perFeedTimeout = 45 * time.Second
|
||||
// refreshCooldown skips feeds fetched very recently on manual refresh.
|
||||
refreshCooldown = 2 * time.Minute
|
||||
)
|
||||
|
||||
// FetchService handles fetching and parsing feeds.
|
||||
type FetchService struct {
|
||||
feedRepo domain.FeedRepository
|
||||
articleRepo domain.ArticleRepository
|
||||
parser *parser.FeedParser
|
||||
hub *ws.Hub
|
||||
|
||||
// sem bounds concurrent outbound fetches across the scheduler and
|
||||
// manual refreshes.
|
||||
sem chan struct{}
|
||||
|
||||
mu sync.Mutex
|
||||
refreshing map[uuid.UUID]bool
|
||||
}
|
||||
|
||||
// NewFetchService creates a new fetch service.
|
||||
func NewFetchService(feedRepo domain.FeedRepository, articleRepo domain.ArticleRepository, hub *ws.Hub) *FetchService {
|
||||
func NewFetchService(feedRepo domain.FeedRepository, articleRepo domain.ArticleRepository, hub *ws.Hub, concurrency int) *FetchService {
|
||||
if concurrency <= 0 {
|
||||
concurrency = 4
|
||||
}
|
||||
return &FetchService{
|
||||
feedRepo: feedRepo,
|
||||
articleRepo: articleRepo,
|
||||
parser: parser.NewFeedParser(),
|
||||
hub: hub,
|
||||
sem: make(chan struct{}, concurrency),
|
||||
refreshing: make(map[uuid.UUID]bool),
|
||||
}
|
||||
}
|
||||
|
||||
// FetchFeed fetches a single feed and updates articles.
|
||||
// FetchFeed fetches a single feed by ID (used right after a feed is added).
|
||||
func (s *FetchService) FetchFeed(ctx context.Context, feedID uuid.UUID) error {
|
||||
feed, err := s.feedRepo.GetByID(feedID)
|
||||
if err != nil {
|
||||
@@ -39,76 +71,226 @@ func (s *FetchService) FetchFeed(ctx context.Context, feedID uuid.UUID) error {
|
||||
if feed == nil {
|
||||
return fmt.Errorf("feed not found: %s", feedID)
|
||||
}
|
||||
s.sem <- struct{}{}
|
||||
defer func() { <-s.sem }()
|
||||
return s.fetchOne(ctx, feed)
|
||||
}
|
||||
|
||||
// Parse the feed
|
||||
parsedFeed, err := s.parser.Parse(ctx, feed.URL, feed.ID)
|
||||
// fetchOne downloads, parses and ingests one feed, then records the outcome.
|
||||
func (s *FetchService) fetchOne(ctx context.Context, feed *domain.Feed) error {
|
||||
ctx, cancel := context.WithTimeout(ctx, perFeedTimeout)
|
||||
defer cancel()
|
||||
|
||||
now := time.Now()
|
||||
parsed, err := s.parser.Parse(ctx, feed)
|
||||
switch {
|
||||
case errors.Is(err, parser.ErrNotModified):
|
||||
return s.saveSuccess(feed, now, nil)
|
||||
case err != nil:
|
||||
s.saveFailure(feed, now, err)
|
||||
return err
|
||||
}
|
||||
|
||||
inserted, err := s.ingest(ctx, feed.ID, parsed, now)
|
||||
if err != nil {
|
||||
// Update feed with error
|
||||
s.feedRepo.UpdateFetchStatus(feed.ID, time.Now(), err.Error())
|
||||
return fmt.Errorf("parsing feed: %w", err)
|
||||
s.saveFailure(feed, now, err)
|
||||
return err
|
||||
}
|
||||
|
||||
// Update feed metadata
|
||||
// Only update title if it's empty or looks like a URL (initial state)
|
||||
if feed.Title == "" || feed.Title == feed.URL {
|
||||
feed.Title = parsedFeed.Title
|
||||
}
|
||||
|
||||
feed.Description = parsedFeed.Description
|
||||
feed.SiteURL = parsedFeed.SiteURL
|
||||
feed.ImageURL = parsedFeed.ImageURL
|
||||
if err := s.feedRepo.Update(feed); err != nil {
|
||||
log.Printf("Warning: failed to update feed metadata: %v", err)
|
||||
if err := s.saveSuccess(feed, now, parsed); err != nil {
|
||||
log.Printf("Warning: saving fetch result for %s: %v", feed.URL, err)
|
||||
}
|
||||
|
||||
// Ingest articles
|
||||
if len(parsedFeed.Articles) > 0 {
|
||||
if err := s.articleRepo.CreateBatch(parsedFeed.Articles); err != nil {
|
||||
return fmt.Errorf("ingesting articles: %w", err)
|
||||
}
|
||||
|
||||
// Broadcast update
|
||||
if s.hub != nil {
|
||||
s.hub.Broadcast("new_articles", map[string]interface{}{
|
||||
"feed_id": feed.ID,
|
||||
"feed_title": feed.Title,
|
||||
"count": len(parsedFeed.Articles),
|
||||
})
|
||||
}
|
||||
if inserted > 0 && s.hub != nil {
|
||||
s.hub.SendToUser(feed.UserID, "new_articles", map[string]any{
|
||||
"feed_id": feed.ID,
|
||||
"count": inserted,
|
||||
})
|
||||
}
|
||||
|
||||
// Mark fetch as successful
|
||||
s.feedRepo.UpdateFetchStatus(feed.ID, time.Now(), "")
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// FetchAllPending fetches all feeds that need updating.
|
||||
func (s *FetchService) FetchAllPending(ctx context.Context, concurrency int) (int, error) {
|
||||
feeds, err := s.feedRepo.GetFeedsToFetch(100)
|
||||
// ingest inserts only the items not already stored, converting (sanitizing,
|
||||
// image lookup) just those.
|
||||
func (s *FetchService) ingest(ctx context.Context, feedID uuid.UUID, parsed *parser.ParsedFeed, now time.Time) (int, error) {
|
||||
if len(parsed.Items) == 0 {
|
||||
return 0, nil
|
||||
}
|
||||
guids := make([]string, len(parsed.Items))
|
||||
for i, it := range parsed.Items {
|
||||
guids[i] = it.GUID
|
||||
}
|
||||
existing, err := s.articleRepo.ExistingGUIDs(ctx, feedID, guids)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
cutoff := now.Add(-ArticleRetention)
|
||||
var fresh []*domain.Article
|
||||
for _, it := range parsed.Items {
|
||||
if _, ok := existing[it.GUID]; ok {
|
||||
continue
|
||||
}
|
||||
if p := it.PublishedAt(); p != nil && p.Before(cutoff) {
|
||||
continue
|
||||
}
|
||||
fresh = append(fresh, s.parser.ToArticle(it, feedID, now))
|
||||
}
|
||||
if len(fresh) == 0 {
|
||||
return 0, nil
|
||||
}
|
||||
n, err := s.articleRepo.InsertNew(ctx, feedID, fresh)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("ingesting articles: %w", err)
|
||||
}
|
||||
return n, nil
|
||||
}
|
||||
|
||||
func (s *FetchService) saveSuccess(feed *domain.Feed, now time.Time, parsed *parser.ParsedFeed) error {
|
||||
res := domain.FetchResult{
|
||||
FetchedAt: now,
|
||||
NextFetchAt: now.Add(jitter(FetchInterval)),
|
||||
}
|
||||
if parsed != nil {
|
||||
res.ETag = parsed.ETag
|
||||
res.LastModified = parsed.LastModified
|
||||
res.Title = nonEmpty(parsed.Title)
|
||||
res.Description = nonEmpty(parsed.Description)
|
||||
res.SiteURL = nonEmpty(parsed.SiteURL)
|
||||
res.ImageURL = nonEmpty(parsed.ImageURL)
|
||||
}
|
||||
return s.feedRepo.SaveFetchResult(feed.ID, res)
|
||||
}
|
||||
|
||||
func (s *FetchService) saveFailure(feed *domain.Feed, now time.Time, cause error) {
|
||||
count := feed.ErrorCount + 1
|
||||
backoff := FetchInterval << min(count-1, 7) // 15m, 30m, 1h … capped below
|
||||
if backoff > maxBackoff || backoff <= 0 {
|
||||
backoff = maxBackoff
|
||||
}
|
||||
log.Printf("Fetch failed for %s (attempt %d): %v", feed.URL, count, cause)
|
||||
err := s.feedRepo.SaveFetchResult(feed.ID, domain.FetchResult{
|
||||
FetchedAt: now,
|
||||
NextFetchAt: now.Add(jitter(backoff)),
|
||||
Error: publicFetchError(cause),
|
||||
ErrorCount: count,
|
||||
})
|
||||
if err != nil {
|
||||
log.Printf("Warning: saving fetch failure for %s: %v", feed.URL, err)
|
||||
}
|
||||
}
|
||||
|
||||
// publicFetchError maps an internal error to a short user-facing message,
|
||||
// without leaking resolver or dial details (port/DNS scanning oracle).
|
||||
func publicFetchError(err error) string {
|
||||
var netErr net.Error
|
||||
msg := err.Error()
|
||||
switch {
|
||||
case errors.Is(err, context.DeadlineExceeded) || (errors.As(err, &netErr) && netErr.Timeout()):
|
||||
return "Délai dépassé"
|
||||
case strings.Contains(msg, "unexpected status code: "):
|
||||
return "Réponse HTTP " + msg[strings.LastIndex(msg, " ")+1:]
|
||||
case strings.Contains(msg, "parsing feed"):
|
||||
return "Flux illisible (RSS/Atom invalide)"
|
||||
case strings.Contains(msg, "too large"):
|
||||
return "Flux trop volumineux"
|
||||
case strings.Contains(msg, "ingesting"):
|
||||
return "Erreur d'enregistrement"
|
||||
case strings.Contains(msg, "not allowed") || strings.Contains(msg, "private") || strings.Contains(msg, "blocked"):
|
||||
return "Adresse refusée"
|
||||
default:
|
||||
return "Serveur injoignable"
|
||||
}
|
||||
}
|
||||
|
||||
// FetchAllPending fetches every due feed with bounded concurrency.
|
||||
func (s *FetchService) FetchAllPending(ctx context.Context) (int, error) {
|
||||
feeds, err := s.feedRepo.GetFeedsToFetch(200)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("getting feeds to fetch: %w", err)
|
||||
}
|
||||
return s.fetchMany(ctx, feeds), nil
|
||||
}
|
||||
|
||||
if len(feeds) == 0 {
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
// Simple sequential fetch for now (Story 2.5 will add worker pool)
|
||||
fetchedCount := 0
|
||||
func (s *FetchService) fetchMany(ctx context.Context, feeds []*domain.Feed) int {
|
||||
var (
|
||||
wg sync.WaitGroup
|
||||
mu sync.Mutex
|
||||
ok int
|
||||
)
|
||||
for _, feed := range feeds {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return fetchedCount, ctx.Err()
|
||||
default:
|
||||
}
|
||||
|
||||
if err := s.FetchFeed(ctx, feed.ID); err != nil {
|
||||
log.Printf("Error fetching feed %s: %v", feed.URL, err)
|
||||
} else {
|
||||
fetchedCount++
|
||||
wg.Wait()
|
||||
return ok
|
||||
case s.sem <- struct{}{}:
|
||||
}
|
||||
wg.Add(1)
|
||||
go func(f *domain.Feed) {
|
||||
defer wg.Done()
|
||||
defer func() { <-s.sem }()
|
||||
if err := s.fetchOne(ctx, f); err == nil {
|
||||
mu.Lock()
|
||||
ok++
|
||||
mu.Unlock()
|
||||
}
|
||||
}(feed)
|
||||
}
|
||||
wg.Wait()
|
||||
return ok
|
||||
}
|
||||
|
||||
return fetchedCount, nil
|
||||
// RefreshUser fetches the user's feeds in the background. It returns false
|
||||
// when a refresh for this user is already running (clicks are coalesced).
|
||||
func (s *FetchService) RefreshUser(userID uuid.UUID) bool {
|
||||
s.mu.Lock()
|
||||
if s.refreshing[userID] {
|
||||
s.mu.Unlock()
|
||||
return false
|
||||
}
|
||||
s.refreshing[userID] = true
|
||||
s.mu.Unlock()
|
||||
|
||||
go func() {
|
||||
defer func() {
|
||||
s.mu.Lock()
|
||||
delete(s.refreshing, userID)
|
||||
s.mu.Unlock()
|
||||
}()
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
|
||||
defer cancel()
|
||||
|
||||
feeds, err := s.feedRepo.GetByUserID(userID)
|
||||
if err != nil {
|
||||
log.Printf("Refresh: listing feeds: %v", err)
|
||||
return
|
||||
}
|
||||
due := feeds[:0]
|
||||
for _, f := range feeds {
|
||||
if f.LastFetchedAt == nil || time.Since(*f.LastFetchedAt) > refreshCooldown {
|
||||
due = append(due, f)
|
||||
}
|
||||
}
|
||||
s.fetchMany(ctx, due)
|
||||
if s.hub != nil {
|
||||
s.hub.SendToUser(userID, "refresh_done", map[string]any{"count": len(due)})
|
||||
}
|
||||
}()
|
||||
return true
|
||||
}
|
||||
|
||||
// jitter spreads fetches by ±10% so feeds don't all fire on the same tick.
|
||||
func jitter(d time.Duration) time.Duration {
|
||||
spread := int64(d) / 10
|
||||
if spread <= 0 {
|
||||
return d
|
||||
}
|
||||
return d + time.Duration(rand.Int64N(2*spread)-spread)
|
||||
}
|
||||
|
||||
func nonEmpty(s string) *string {
|
||||
if s == "" {
|
||||
return nil
|
||||
}
|
||||
return &s
|
||||
}
|
||||
Reference in new issue
Block a user