" + s.sanitizer.Sanitize(article.TextContent) + "
" + } + + s.log.Info("Readability scrape successful", + zap.String("entry_id", entry.ID), + zap.String("title", article.Title), + zap.Int("content_length", len(cleanHTML)), + ) + + now := time.Now().UTC() + variant := models.EntryVariant{ + EntryID: entry.ID, + VariantKey: "readability_full_text", + ContentType: "html", + TextValue: cleanHTML, + Metadata: map[string]interface{}{ + "title": article.Title, + "byline": article.Byline, + "excerpt": article.Excerpt, + "length": len(article.TextContent), + "scraped_at": now.Format(time.RFC3339), + }, + CreatedAt: now, + UpdatedAt: now, + } + + return &worker.ActionResult{ + ResultVariant: &variant, + }, nil + } +} diff --git a/internal/api/handlers.go b/internal/api/handlers.go index a2d69f0..e6aa3df 100644 --- a/internal/api/handlers.go +++ b/internal/api/handlers.go @@ -17,6 +17,7 @@ import ( "git.pengzhan.dev/haopengzhan/rsspipe/internal/models" "git.pengzhan.dev/haopengzhan/rsspipe/internal/opml" "git.pengzhan.dev/haopengzhan/rsspipe/internal/parser" + "git.pengzhan.dev/haopengzhan/rsspipe/internal/worker" "github.com/go-chi/chi/v5" ) @@ -25,14 +26,16 @@ type Handler struct { auth *auth.AuthService crawler *crawler.Crawler parser *parser.Parser + pool *worker.Pool } -func NewHandler(database *db.DB, authService *auth.AuthService, cr *crawler.Crawler) *Handler { +func NewHandler(database *db.DB, authService *auth.AuthService, cr *crawler.Crawler, pool *worker.Pool) *Handler { return &Handler{ db: database, auth: authService, crawler: cr, parser: parser.NewParser(), + pool: pool, } } @@ -547,3 +550,160 @@ func (h *Handler) ViewSharedEntry(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "text/html; charset=utf-8") _, _ = w.Write([]byte(html)) } + +// GetEntry returns a single entry by ID with its loaded variants +func (h *Handler) GetEntry(w http.ResponseWriter, r *http.Request) { + id := chi.URLParam(r, "id") + entry, err := h.db.GetEntry(id) + if err != nil { + if err == sql.ErrNoRows { + h.writeError(w, http.StatusNotFound, "entry not found") + return + } + h.writeError(w, http.StatusInternalServerError, err.Error()) + return + } + h.writeJSON(w, http.StatusOK, entry) +} + +// ListEntryVariants returns all variants for a given entry ID +func (h *Handler) ListEntryVariants(w http.ResponseWriter, r *http.Request) { + id := chi.URLParam(r, "id") + variants, err := h.db.ListVariantsForEntry(id) + if err != nil { + h.writeError(w, http.StatusInternalServerError, err.Error()) + return + } + h.writeJSON(w, http.StatusOK, variants) +} + +// GetVariantBlobStream streams the audio/binary file for a variant with HTTP 206 Partial Content support +func (h *Handler) GetVariantBlobStream(w http.ResponseWriter, r *http.Request) { + id := chi.URLParam(r, "id") + key := chi.URLParam(r, "key") + variant, err := h.db.GetVariant(id, key) + if err != nil { + h.writeError(w, http.StatusNotFound, "variant not found") + return + } + if variant.BlobPath == "" { + h.writeError(w, http.StatusNotFound, "variant has no binary media blob") + return + } + + http.ServeFile(w, r, variant.BlobPath) +} + +// ScrapeEntry manually triggers readability full text scraping and saves to entry_variants +func (h *Handler) ScrapeEntry(w http.ResponseWriter, r *http.Request) { + id := chi.URLParam(r, "id") + entry, err := h.db.GetEntry(id) + if err != nil { + h.writeError(w, http.StatusNotFound, "entry not found") + return + } + + if h.pool == nil { + h.writeError(w, http.StatusInternalServerError, "worker pool not configured") + return + } + + res, err := h.pool.ExecuteDirect(r.Context(), "readability_scrape", *entry, nil) + if err != nil { + h.writeError(w, http.StatusInternalServerError, fmt.Sprintf("scraping failed: %v", err)) + return + } + + if res != nil && res.ResultVariant != nil { + _ = h.db.SetVariant(*res.ResultVariant) + } + + // Reload entry with newly saved variant + updatedEntry, err := h.db.GetEntry(id) + if err != nil { + h.writeError(w, http.StatusInternalServerError, err.Error()) + return + } + + h.writeJSON(w, http.StatusOK, updatedEntry) +} + +// Rules API Handlers +func (h *Handler) ListRules(w http.ResponseWriter, r *http.Request) { + rules, err := h.db.ListRules() + if err != nil { + h.writeError(w, http.StatusInternalServerError, err.Error()) + return + } + if rules == nil { + rules = []models.Rule{} + } + h.writeJSON(w, http.StatusOK, rules) +} + +type CreateRuleRequest struct { + Name string `json:"name"` + Trigger string `json:"trigger"` + MatchMode string `json:"match_mode"` + Conditions []models.RuleCondition `json:"conditions"` + Actions []models.RuleAction `json:"actions"` +} + +func (h *Handler) CreateRule(w http.ResponseWriter, r *http.Request) { + var body CreateRuleRequest + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + h.writeError(w, http.StatusBadRequest, "invalid request body") + return + } + if strings.TrimSpace(body.Name) == "" { + h.writeError(w, http.StatusBadRequest, "rule name is required") + return + } + if strings.TrimSpace(body.Trigger) == "" { + body.Trigger = "on_post_created" + } + if strings.TrimSpace(body.MatchMode) == "" { + body.MatchMode = "all" + } + + rule, err := h.db.CreateRule(body.Name, body.Trigger, body.MatchMode, body.Conditions, body.Actions) + if err != nil { + h.writeError(w, http.StatusInternalServerError, err.Error()) + return + } + h.writeJSON(w, http.StatusCreated, rule) +} + +type UpdateRuleRequest struct { + Name string `json:"name"` + Trigger string `json:"trigger"` + MatchMode string `json:"match_mode"` + Conditions []models.RuleCondition `json:"conditions"` + Actions []models.RuleAction `json:"actions"` + IsActive bool `json:"is_active"` +} + +func (h *Handler) UpdateRule(w http.ResponseWriter, r *http.Request) { + id := chi.URLParam(r, "id") + var body UpdateRuleRequest + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + h.writeError(w, http.StatusBadRequest, "invalid request body") + return + } + + rule, err := h.db.UpdateRule(id, body.Name, body.Trigger, body.MatchMode, body.Conditions, body.Actions, body.IsActive) + if err != nil { + h.writeError(w, http.StatusInternalServerError, err.Error()) + return + } + h.writeJSON(w, http.StatusOK, rule) +} + +func (h *Handler) DeleteRule(w http.ResponseWriter, r *http.Request) { + id := chi.URLParam(r, "id") + if err := h.db.DeleteRule(id); err != nil { + h.writeError(w, http.StatusInternalServerError, err.Error()) + return + } + h.writeJSON(w, http.StatusOK, map[string]string{"message": "rule deleted"}) +} diff --git a/internal/api/server.go b/internal/api/server.go index 5a75487..aa1aa56 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -62,9 +62,19 @@ func NewRouter(h *Handler, authService *auth.AuthService, staticDir string) http // Entries r.Get("/entries", h.ListEntries) + r.Get("/entries/{id}", h.GetEntry) r.Put("/entries/{id}/read", h.UpdateEntryRead) r.Put("/entries/{id}/star", h.UpdateEntryStar) r.Post("/entries/mark-all-read", h.MarkAllRead) + r.Post("/entries/{id}/scrape", h.ScrapeEntry) + r.Get("/entries/{id}/variants", h.ListEntryVariants) + r.Get("/entries/{id}/variants/{key}/stream", h.GetVariantBlobStream) + + // Rules + r.Get("/rules", h.ListRules) + r.Post("/rules", h.CreateRule) + r.Put("/rules/{id}", h.UpdateRule) + r.Delete("/rules/{id}", h.DeleteRule) // OPML r.Post("/opml/import", h.ImportOPML) diff --git a/internal/crawler/crawler.go b/internal/crawler/crawler.go index f1ebd64..b750672 100644 --- a/internal/crawler/crawler.go +++ b/internal/crawler/crawler.go @@ -3,35 +3,41 @@ package crawler import ( "context" "fmt" - "log" "sync" "time" "git.pengzhan.dev/haopengzhan/rsspipe/internal/db" + "git.pengzhan.dev/haopengzhan/rsspipe/internal/events" + "git.pengzhan.dev/haopengzhan/rsspipe/internal/logger" "git.pengzhan.dev/haopengzhan/rsspipe/internal/models" "git.pengzhan.dev/haopengzhan/rsspipe/internal/parser" + "go.uber.org/zap" ) type Crawler struct { db *db.DB parser *parser.Parser + bus *events.EventBus intervalMinutes int workerCount int isRefreshing bool mu sync.Mutex + log *zap.Logger } -func NewCrawler(database *db.DB, intervalMinutes, workers int) *Crawler { +func NewCrawler(database *db.DB, bus *events.EventBus, intervalMinutes, workers int) *Crawler { return &Crawler{ db: database, parser: parser.NewParser(), + bus: bus, intervalMinutes: intervalMinutes, workerCount: workers, + log: logger.Named("crawler"), } } func (c *Crawler) Start(ctx context.Context) { - log.Printf("==> Starting RSS background crawler scheduler (Workers: %d)\n", c.workerCount) + c.log.Info("Starting RSS background crawler scheduler", zap.Int("worker_count", c.workerCount)) // Run initial fetch asynchronously after 2 seconds startup go func() { @@ -72,7 +78,7 @@ func (c *Crawler) RefreshDue(ctx context.Context) { feeds, err := c.db.ListFeeds() if err != nil { - log.Printf("Crawler error listing feeds: %v\n", err) + c.log.Error("Crawler error listing feeds", zap.Error(err)) return } @@ -93,7 +99,7 @@ func (c *Crawler) RefreshDue(ctx context.Context) { return } - log.Printf("Crawler: Refreshing %d due feeds (out of %d total)...\n", len(dueFeeds), len(feeds)) + c.log.Info("Crawler refreshing due feeds", zap.Int("due_count", len(dueFeeds)), zap.Int("total_count", len(feeds))) c.runWorkerPool(ctx, dueFeeds) } @@ -102,7 +108,7 @@ func (c *Crawler) RefreshAll(ctx context.Context) { c.mu.Lock() if c.isRefreshing { c.mu.Unlock() - log.Println("Crawler: RefreshAll already in progress, skipping") + c.log.Debug("Crawler RefreshAll already in progress, skipping") return } c.isRefreshing = true @@ -116,7 +122,7 @@ func (c *Crawler) RefreshAll(ctx context.Context) { feeds, err := c.db.ListFeeds() if err != nil { - log.Printf("Crawler error listing feeds: %v\n", err) + c.log.Error("Crawler error listing feeds", zap.Error(err)) return } @@ -124,7 +130,7 @@ func (c *Crawler) RefreshAll(ctx context.Context) { return } - log.Printf("Crawler: Refreshing all %d feeds...\n", len(feeds)) + c.log.Info("Crawler refreshing all feeds", zap.Int("count", len(feeds))) c.runWorkerPool(ctx, feeds) } @@ -148,7 +154,7 @@ func (c *Crawler) runWorkerPool(ctx context.Context, feeds []models.Feed) { close(jobs) wg.Wait() - log.Printf("Crawler: Completed refresh of %d feeds.\n", len(feeds)) + c.log.Info("Crawler completed refresh batch", zap.Int("feed_count", len(feeds))) } func (c *Crawler) RefreshSingle(ctx context.Context, feedID string) error { @@ -169,7 +175,11 @@ func (c *Crawler) fetchSingleFeed(ctx context.Context, f models.Feed) error { res, err := c.parser.FetchFeed(feedCtx, f.FeedURL, f.ETag, f.LastModified) if err != nil { - log.Printf("Crawler error fetching [%s]: %v\n", f.Title, err) + c.log.Warn("Crawler error fetching feed", + zap.String("feed_title", f.Title), + zap.String("feed_url", f.FeedURL), + zap.Error(err), + ) _ = c.db.UpdateFeedFetchStatus(f.ID, f.ETag, f.LastModified, err.Error()) return err } @@ -180,11 +190,30 @@ func (c *Crawler) fetchSingleFeed(ctx context.Context, f models.Feed) error { } // Upsert entries - inserted, err := c.db.UpsertFeedEntries(f.ID, res.Entries) + newEntries, err := c.db.UpsertFeedEntries(f.ID, res.Entries) if err != nil { - log.Printf("Crawler error upserting entries for [%s]: %v\n", f.Title, err) - } else if inserted > 0 { - log.Printf("Crawler [%s]: inserted %d new entries\n", f.Title, inserted) + c.log.Error("Crawler error upserting entries", + zap.String("feed_title", f.Title), + zap.Error(err), + ) + } else if len(newEntries) > 0 { + c.log.Info("Crawler inserted new entries", + zap.String("feed_title", f.Title), + zap.Int("new_entries", len(newEntries)), + ) + + // Publish on_post_created events for newly inserted entries + if c.bus != nil { + for _, entry := range newEntries { + entry.FeedTitle = f.Title + entry.FeedSiteURL = f.SiteURL + c.bus.Publish(ctx, events.Event{ + Type: events.EventPostCreated, + Entry: entry, + Depth: 0, + }) + } + } } _ = c.db.UpdateFeedFetchStatus(f.ID, res.ETag, res.LastModified, "") diff --git a/internal/db/db.go b/internal/db/db.go index 933bb0a..c5acb2d 100644 --- a/internal/db/db.go +++ b/internal/db/db.go @@ -114,15 +114,28 @@ func (d *DB) migrate() error { CREATE INDEX IF NOT EXISTS idx_entries_is_starred ON feed_entries(is_starred); CREATE INDEX IF NOT EXISTS idx_entries_share_uuid ON feed_entries(share_uuid); + CREATE TABLE IF NOT EXISTS entry_variants ( + entry_id TEXT NOT NULL, + variant_key TEXT NOT NULL, + content_type TEXT NOT NULL, + text_value TEXT, + blob_path TEXT, + metadata TEXT DEFAULT '{}', + created_at DATETIME NOT NULL, + updated_at DATETIME NOT NULL, + PRIMARY KEY (entry_id, variant_key), + FOREIGN KEY(entry_id) REFERENCES feed_entries(id) ON DELETE CASCADE + ); + CREATE INDEX IF NOT EXISTS idx_entry_variants_key ON entry_variants(variant_key); + CREATE TABLE IF NOT EXISTS rules ( id TEXT PRIMARY KEY, name TEXT NOT NULL, - rule_type TEXT NOT NULL, - is_active BOOLEAN DEFAULT 1, - trigger_type TEXT NOT NULL, - trigger_config TEXT DEFAULT '{}', + trigger TEXT NOT NULL, conditions TEXT DEFAULT '[]', + match_mode TEXT DEFAULT 'all', actions TEXT DEFAULT '[]', + is_active BOOLEAN DEFAULT 1, created_at DATETIME NOT NULL, updated_at DATETIME NOT NULL ); @@ -132,6 +145,26 @@ func (d *DB) migrate() error { } // Migration for timeout_seconds column if older DB exists _, _ = d.Exec("ALTER TABLE feeds ADD COLUMN timeout_seconds INTEGER DEFAULT 10") + + // Migration for rules table schema if older table exists + var testCol string + err := d.QueryRow("SELECT trigger FROM rules LIMIT 1").Scan(&testCol) + if err != nil && err != sql.ErrNoRows { + _, _ = d.Exec("DROP TABLE IF EXISTS rules") + _, _ = d.Exec(` + CREATE TABLE IF NOT EXISTS rules ( + id TEXT PRIMARY KEY, + name TEXT NOT NULL, + trigger TEXT NOT NULL, + conditions TEXT DEFAULT '[]', + match_mode TEXT DEFAULT 'all', + actions TEXT DEFAULT '[]', + is_active BOOLEAN DEFAULT 1, + created_at DATETIME NOT NULL, + updated_at DATETIME NOT NULL + ); + `) + } return nil } @@ -474,51 +507,67 @@ func (d *DB) DeleteFeed(id string) error { } // Feed Entries operations -func (d *DB) UpsertFeedEntries(feedID string, entries []models.FeedEntry) (int, error) { +func (d *DB) UpsertFeedEntries(feedID string, entries []models.FeedEntry) ([]models.FeedEntry, error) { if len(entries) == 0 { - return 0, nil + return nil, nil } - stmt, err := d.Prepare(` + var newEntries []models.FeedEntry + now := time.Now().UTC() + + insertStmt, err := d.Prepare(` INSERT INTO feed_entries ( id, feed_id, guid, url, title, author, content, raw_content, published_at, is_read, is_starred, metadata, created_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 0, 0, ?, ?) - ON CONFLICT(feed_id, guid) DO UPDATE SET - title = excluded.title, - content = CASE WHEN feed_entries.content = '' THEN excluded.content ELSE feed_entries.content END, - url = excluded.url, - author = excluded.author `) if err != nil { - return 0, err + return nil, err } defer func() { - _ = stmt.Close() + _ = insertStmt.Close() }() - insertedCount := 0 - for _, e := range entries { - id := uuid.New().String() - now := time.Now().UTC() - pubAt := e.PublishedAt - if pubAt.IsZero() { - pubAt = now - } - metaBytes, _ := json.Marshal(e.Metadata) - if len(metaBytes) == 0 { - metaBytes = []byte("{}") - } + updateStmt, err := d.Prepare(` + UPDATE feed_entries + SET title = ?, url = ?, author = ?, content = ?, raw_content = ? + WHERE feed_id = ? AND guid = ? + `) + if err != nil { + return nil, err + } + defer func() { + _ = updateStmt.Close() + }() - res, err := stmt.Exec( - id, feedID, e.GUID, e.URL, e.Title, e.Author, e.Content, e.RawContent, pubAt, string(metaBytes), now, - ) - if err == nil { - if n, _ := res.RowsAffected(); n > 0 { - insertedCount++ + for _, e := range entries { + var existingID string + err := d.QueryRow("SELECT id FROM feed_entries WHERE feed_id = ? AND guid = ?", feedID, e.GUID).Scan(&existingID) + switch err { + case sql.ErrNoRows: + e.ID = uuid.New().String() + e.FeedID = feedID + pubAt := e.PublishedAt + if pubAt.IsZero() { + pubAt = now } + metaBytes, _ := json.Marshal(e.Metadata) + if len(metaBytes) == 0 { + metaBytes = []byte("{}") + } + + _, err := insertStmt.Exec( + e.ID, feedID, e.GUID, e.URL, e.Title, e.Author, e.Content, e.RawContent, pubAt, string(metaBytes), now, + ) + if err == nil { + e.CreatedAt = now + newEntries = append(newEntries, e) + } + case nil: + _, _ = updateStmt.Exec(e.Title, e.URL, e.Author, e.Content, e.RawContent, feedID, e.GUID) } } - return insertedCount, nil + + return newEntries, nil } type EntryFilter struct { @@ -530,67 +579,98 @@ type EntryFilter struct { Offset int } -func (d *DB) ListEntries(f EntryFilter) ([]models.FeedEntry, int, error) { - where := []string{"1=1"} - args := []interface{}{} +func (d *DB) GetEntry(id string) (*models.FeedEntry, error) { + query := ` + SELECT + e.id, e.feed_id, f.title as feed_title, COALESCE(f.site_url, '') as feed_site_url, + e.guid, e.url, e.title, e.author, e.content, e.raw_content, + e.published_at, e.is_read, e.is_starred, e.share_uuid, e.metadata, e.created_at + FROM feed_entries e + JOIN feeds f ON f.id = e.feed_id + WHERE e.id = ? + ` + var e models.FeedEntry + var author, rawContent, shareUUID, metaStr sql.NullString + err := d.QueryRow(query, id).Scan( + &e.ID, &e.FeedID, &e.FeedTitle, &e.FeedSiteURL, + &e.GUID, &e.URL, &e.Title, &author, &e.Content, &rawContent, + &e.PublishedAt, &e.IsRead, &e.IsStarred, &shareUUID, &metaStr, &e.CreatedAt, + ) + if err != nil { + return nil, err + } + if author.Valid { + e.Author = author.String + } + if rawContent.Valid { + e.RawContent = rawContent.String + } + if shareUUID.Valid { + s := shareUUID.String + e.ShareUUID = &s + } + e.Metadata = make(map[string]interface{}) + if metaStr.Valid && metaStr.String != "" { + _ = json.Unmarshal([]byte(metaStr.String), &e.Metadata) + } - switch f.View { + // Load variants + variants, _ := d.ListVariantsForEntry(e.ID) + e.Variants = variants + + return &e, nil +} + +func (d *DB) ListEntries(filter EntryFilter) ([]models.FeedEntry, int, error) { + var whereClauses []string + var args []interface{} + + switch filter.View { case "inbox": - where = append(where, "e.is_read = 0") + whereClauses = append(whereClauses, "e.is_read = 0") case "starred": - where = append(where, "e.is_starred = 1") + whereClauses = append(whereClauses, "e.is_starred = 1") } - if f.FeedID != "" { - where = append(where, "e.feed_id = ?") - args = append(args, f.FeedID) + if filter.FeedID != "" { + whereClauses = append(whereClauses, "e.feed_id = ?") + args = append(args, filter.FeedID) } - if f.CategoryID != "" { - where = append(where, "f.category_id = ?") - args = append(args, f.CategoryID) + if filter.CategoryID != "" { + whereClauses = append(whereClauses, "f.category_id = ?") + args = append(args, filter.CategoryID) } - if f.Search != "" { - searchTerms := strings.Fields(f.Search) - for _, term := range searchTerms { - lower := strings.ToLower(term) - if strings.HasPrefix(lower, "is:unread") { - where = append(where, "e.is_read = 0") - } else if strings.HasPrefix(lower, "is:read") { - where = append(where, "e.is_read = 1") - } else if strings.HasPrefix(lower, "is:starred") { - where = append(where, "e.is_starred = 1") - } else if strings.HasPrefix(lower, "feed:") { - feedVal := strings.TrimPrefix(term, "feed:") - where = append(where, "(f.title LIKE ? OR f.feed_url LIKE ?)") - args = append(args, "%"+feedVal+"%", "%"+feedVal+"%") - } else { - where = append(where, "(e.title LIKE ? OR e.content LIKE ? OR e.author LIKE ?)") - args = append(args, "%"+term+"%", "%"+term+"%", "%"+term+"%") - } - } + if filter.Search != "" { + whereClauses = append(whereClauses, "(e.title LIKE ? OR e.content LIKE ? OR e.author LIKE ?)") + pattern := "%" + filter.Search + "%" + args = append(args, pattern, pattern, pattern) } - whereClause := strings.Join(where, " AND ") + whereSQL := "" + if len(whereClauses) > 0 { + whereSQL = "WHERE " + strings.Join(whereClauses, " AND ") + } - // Count total countQuery := fmt.Sprintf(` - SELECT COUNT(*) - FROM feed_entries e - JOIN feeds f ON f.id = e.feed_id - WHERE %s - `, whereClause) + SELECT COUNT(e.id) + FROM feed_entries e + JOIN feeds f ON f.id = e.feed_id + %s + `, whereSQL) + var total int - if err := d.QueryRow(countQuery, args...).Scan(&total); err != nil { + err := d.QueryRow(countQuery, args...).Scan(&total) + if err != nil { return nil, 0, err } - limit := f.Limit - if limit <= 0 || limit > 100 { - limit = 50 + limit := filter.Limit + if limit <= 0 || limit > 500 { + limit = 100 } - offset := f.Offset + offset := filter.Offset if offset < 0 { offset = 0 } @@ -602,10 +682,10 @@ func (d *DB) ListEntries(f EntryFilter) ([]models.FeedEntry, int, error) { e.published_at, e.is_read, e.is_starred, e.share_uuid, e.metadata, e.created_at FROM feed_entries e JOIN feeds f ON f.id = e.feed_id - WHERE %s - ORDER BY e.published_at DESC + %s + ORDER BY e.published_at DESC, e.created_at DESC LIMIT ? OFFSET ? - `, whereClause) + `, whereSQL) queryArgs := append(args, limit, offset) rows, err := d.Query(query, queryArgs...) @@ -619,13 +699,13 @@ func (d *DB) ListEntries(f EntryFilter) ([]models.FeedEntry, int, error) { var entries []models.FeedEntry for rows.Next() { var e models.FeedEntry - var author, rawContent, metaStr sql.NullString - var shareUUID sql.NullString - if err := rows.Scan( + var author, rawContent, shareUUID, metaStr sql.NullString + err := rows.Scan( &e.ID, &e.FeedID, &e.FeedTitle, &e.FeedSiteURL, &e.GUID, &e.URL, &e.Title, &author, &e.Content, &rawContent, &e.PublishedAt, &e.IsRead, &e.IsStarred, &shareUUID, &metaStr, &e.CreatedAt, - ); err != nil { + ) + if err != nil { return nil, 0, err } if author.Valid { @@ -675,6 +755,232 @@ func (d *DB) MarkAllRead(feedID, categoryID *string) error { return err } +// Entry Variants CRUD +func (d *DB) GetVariant(entryID, key string) (*models.EntryVariant, error) { + query := ` + SELECT entry_id, variant_key, content_type, COALESCE(text_value, ''), COALESCE(blob_path, ''), metadata, created_at, updated_at + FROM entry_variants + WHERE entry_id = ? AND variant_key = ? + ` + var v models.EntryVariant + var metaStr sql.NullString + err := d.QueryRow(query, entryID, key).Scan( + &v.EntryID, &v.VariantKey, &v.ContentType, &v.TextValue, &v.BlobPath, &metaStr, &v.CreatedAt, &v.UpdatedAt, + ) + if err != nil { + return nil, err + } + v.Metadata = make(map[string]interface{}) + if metaStr.Valid && metaStr.String != "" { + _ = json.Unmarshal([]byte(metaStr.String), &v.Metadata) + } + return &v, nil +} + +func (d *DB) ListVariantsForEntry(entryID string) (map[string]models.EntryVariant, error) { + query := ` + SELECT entry_id, variant_key, content_type, COALESCE(text_value, ''), COALESCE(blob_path, ''), metadata, created_at, updated_at + FROM entry_variants + WHERE entry_id = ? + ` + rows, err := d.Query(query, entryID) + if err != nil { + return nil, err + } + defer func() { + _ = rows.Close() + }() + + variants := make(map[string]models.EntryVariant) + for rows.Next() { + var v models.EntryVariant + var metaStr sql.NullString + err := rows.Scan( + &v.EntryID, &v.VariantKey, &v.ContentType, &v.TextValue, &v.BlobPath, &metaStr, &v.CreatedAt, &v.UpdatedAt, + ) + if err != nil { + return nil, err + } + v.Metadata = make(map[string]interface{}) + if metaStr.Valid && metaStr.String != "" { + _ = json.Unmarshal([]byte(metaStr.String), &v.Metadata) + } + variants[v.VariantKey] = v + } + return variants, nil +} + +func (d *DB) SetVariant(v models.EntryVariant) error { + now := time.Now().UTC() + metaBytes, _ := json.Marshal(v.Metadata) + if len(metaBytes) == 0 { + metaBytes = []byte("{}") + } + + query := ` + INSERT INTO entry_variants ( + entry_id, variant_key, content_type, text_value, blob_path, metadata, created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(entry_id, variant_key) DO UPDATE SET + content_type = excluded.content_type, + text_value = excluded.text_value, + blob_path = excluded.blob_path, + metadata = excluded.metadata, + updated_at = excluded.updated_at + ` + _, err := d.Exec(query, v.EntryID, v.VariantKey, v.ContentType, v.TextValue, v.BlobPath, string(metaBytes), now, now) + return err +} + +func (d *DB) DeleteVariant(entryID, key string) error { + _, err := d.Exec("DELETE FROM entry_variants WHERE entry_id = ? AND variant_key = ?", entryID, key) + return err +} + +// Rules CRUD +func (d *DB) ListRules() ([]models.Rule, error) { + query := ` + SELECT id, name, trigger, conditions, match_mode, actions, is_active, created_at, updated_at + FROM rules + ORDER BY created_at ASC + ` + rows, err := d.Query(query) + if err != nil { + return nil, err + } + defer func() { + _ = rows.Close() + }() + + var rules []models.Rule + for rows.Next() { + var r models.Rule + var condsStr, actionsStr string + err := rows.Scan( + &r.ID, &r.Name, &r.Trigger, &condsStr, &r.MatchMode, &actionsStr, &r.IsActive, &r.CreatedAt, &r.UpdatedAt, + ) + if err != nil { + return nil, err + } + _ = json.Unmarshal([]byte(condsStr), &r.Conditions) + _ = json.Unmarshal([]byte(actionsStr), &r.Actions) + rules = append(rules, r) + } + return rules, nil +} + +func (d *DB) GetActiveRules(trigger string) ([]models.Rule, error) { + query := ` + SELECT id, name, trigger, conditions, match_mode, actions, is_active, created_at, updated_at + FROM rules + WHERE is_active = 1 AND (trigger = ? OR trigger = '*') + ORDER BY created_at ASC + ` + rows, err := d.Query(query, trigger) + if err != nil { + return nil, err + } + defer func() { + _ = rows.Close() + }() + + var rules []models.Rule + for rows.Next() { + var r models.Rule + var condsStr, actionsStr string + err := rows.Scan( + &r.ID, &r.Name, &r.Trigger, &condsStr, &r.MatchMode, &actionsStr, &r.IsActive, &r.CreatedAt, &r.UpdatedAt, + ) + if err != nil { + return nil, err + } + _ = json.Unmarshal([]byte(condsStr), &r.Conditions) + _ = json.Unmarshal([]byte(actionsStr), &r.Actions) + rules = append(rules, r) + } + return rules, nil +} + +func (d *DB) GetRule(id string) (*models.Rule, error) { + query := ` + SELECT id, name, trigger, conditions, match_mode, actions, is_active, created_at, updated_at + FROM rules + WHERE id = ? + ` + var r models.Rule + var condsStr, actionsStr string + err := d.QueryRow(query, id).Scan( + &r.ID, &r.Name, &r.Trigger, &condsStr, &r.MatchMode, &actionsStr, &r.IsActive, &r.CreatedAt, &r.UpdatedAt, + ) + if err != nil { + return nil, err + } + _ = json.Unmarshal([]byte(condsStr), &r.Conditions) + _ = json.Unmarshal([]byte(actionsStr), &r.Actions) + return &r, nil +} + +func (d *DB) CreateRule(name, trigger, matchMode string, conditions []models.RuleCondition, actions []models.RuleAction) (*models.Rule, error) { + id := uuid.New().String() + now := time.Now().UTC() + if matchMode == "" { + matchMode = "all" + } + condsBytes, _ := json.Marshal(conditions) + actionsBytes, _ := json.Marshal(actions) + + query := ` + INSERT INTO rules ( + id, name, trigger, conditions, match_mode, actions, is_active, created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, 1, ?, ?) + ` + _, err := d.Exec(query, id, name, trigger, string(condsBytes), matchMode, string(actionsBytes), now, now) + if err != nil { + return nil, err + } + + return &models.Rule{ + ID: id, + Name: name, + Trigger: trigger, + Conditions: conditions, + MatchMode: matchMode, + Actions: actions, + IsActive: true, + CreatedAt: now, + UpdatedAt: now, + }, nil +} + +func (d *DB) UpdateRule(id string, name, trigger, matchMode string, conditions []models.RuleCondition, actions []models.RuleAction, isActive bool) (*models.Rule, error) { + now := time.Now().UTC() + if matchMode == "" { + matchMode = "all" + } + condsBytes, _ := json.Marshal(conditions) + actionsBytes, _ := json.Marshal(actions) + + query := ` + UPDATE rules + SET name = ?, trigger = ?, conditions = ?, match_mode = ?, actions = ?, is_active = ?, updated_at = ? + WHERE id = ? + ` + res, err := d.Exec(query, name, trigger, string(condsBytes), matchMode, string(actionsBytes), isActive, now, id) + if err != nil { + return nil, err + } + if n, _ := res.RowsAffected(); n == 0 { + return nil, fmt.Errorf("rule not found") + } + + return d.GetRule(id) +} + +func (d *DB) DeleteRule(id string) error { + _, err := d.Exec("DELETE FROM rules WHERE id = ?", id) + return err +} + func (d *DB) GetStats() (*models.Stats, error) { var s models.Stats err := d.QueryRow("SELECT COUNT(*) FROM feeds").Scan(&s.TotalFeeds) @@ -693,5 +999,6 @@ func (d *DB) GetStats() (*models.Stats, error) { if err != nil { return nil, err } + _ = d.QueryRow("SELECT COUNT(*) FROM rules").Scan(&s.TotalRules) return &s, nil } diff --git a/internal/db/db_test.go b/internal/db/db_test.go index 42a1d32..5a80541 100644 --- a/internal/db/db_test.go +++ b/internal/db/db_test.go @@ -72,8 +72,8 @@ func TestDatabaseOperations(t *testing.T) { if err != nil { t.Fatalf("UpsertFeedEntries failed: %v", err) } - if inserted != 2 { - t.Fatalf("expected 2 inserted entries, got %d", inserted) + if len(inserted) != 2 { + t.Fatalf("expected 2 inserted entries, got %d", len(inserted)) } // 5. List entries @@ -94,12 +94,69 @@ func TestDatabaseOperations(t *testing.T) { t.Fatalf("UpdateEntryStarred failed: %v", err) } - // 7. Verify Stats + // 7. Test Variant CRUD + variant := models.EntryVariant{ + EntryID: firstID, + VariantKey: "readability_full_text", + ContentType: "html", + TextValue: "") + sb.WriteString(html.EscapeString(l)) + sb.WriteString("
\n") + } + } + } + + for _, block := range payload.Blocks { + txt := strings.TrimSpace(block.Text) + if txt != "" { + sb.WriteString("") + sb.WriteString(html.EscapeString(txt)) + sb.WriteString("
\n") + } + } + + for _, ent := range payload.EntityMap { + if strings.EqualFold(ent.Type, "IMAGE") && ent.Data.Src != "" { + fmt.Fprintf(&sb, `
就这样吧评论: 奥德赛终于看完了奥德赛，非常精彩！
") { + t.Fatalf("expected formatted paragraph in output, got: %s", combined) + } + if !strings.Contains(combined, "img9.doubanio.com/poster.jpg") { + t.Fatalf("expected poster in output, got: %s", combined) + } +} diff --git a/internal/rules/engine.go b/internal/rules/engine.go new file mode 100644 index 0000000..cb973e4 --- /dev/null +++ b/internal/rules/engine.go @@ -0,0 +1,82 @@ +package rules + +import ( + "context" + + "git.pengzhan.dev/haopengzhan/rsspipe/internal/db" + "git.pengzhan.dev/haopengzhan/rsspipe/internal/events" + "git.pengzhan.dev/haopengzhan/rsspipe/internal/logger" + "git.pengzhan.dev/haopengzhan/rsspipe/internal/models" + "go.uber.org/zap" +) + +type WorkerDispatcher interface { + Dispatch(ctx context.Context, action models.RuleAction, entry models.FeedEntry, depth int) +} + +type Engine struct { + db *db.DB + bus *events.EventBus + evaluator *Evaluator + dispatcher WorkerDispatcher + log *zap.Logger +} + +func NewEngine(database *db.DB, bus *events.EventBus, dispatcher WorkerDispatcher) *Engine { + return &Engine{ + db: database, + bus: bus, + evaluator: NewEvaluator(), + dispatcher: dispatcher, + log: logger.Named("rules-engine"), + } +} + +// Start registers event bus subscriptions to evaluate rules reactively. +func (e *Engine) Start(ctx context.Context) { + e.bus.Subscribe(events.EventPostCreated, e.handleEvent) + e.bus.Subscribe(events.EventPostUpdated, e.handleEvent) + e.bus.Subscribe(events.EventVariantUpdated, e.handleEvent) + e.bus.Subscribe(events.EventManualTrigger, e.handleEvent) + e.log.Info("Rules engine started, listening for events") +} + +func (e *Engine) handleEvent(ctx context.Context, evt events.Event) error { + trigger := string(evt.Type) + rules, err := e.db.GetActiveRules(trigger) + if err != nil { + e.log.Error("Failed to fetch active rules", zap.String("trigger", trigger), zap.Error(err)) + return err + } + + if len(rules) == 0 { + return nil + } + + // Fetch full entry with latest variants if entry ID is present + entry := evt.Entry + if entry.ID != "" { + if full, err := e.db.GetEntry(entry.ID); err == nil && full != nil { + entry = *full + } + } + + for _, r := range rules { + if e.evaluator.Evaluate(r, entry) { + e.log.Info("Rule matched", + zap.String("rule_id", r.ID), + zap.String("rule_name", r.Name), + zap.String("entry_id", entry.ID), + zap.Int("actions_count", len(r.Actions)), + ) + + for _, action := range r.Actions { + if e.dispatcher != nil { + e.dispatcher.Dispatch(ctx, action, entry, evt.Depth+1) + } + } + } + } + + return nil +} diff --git a/internal/rules/evaluator.go b/internal/rules/evaluator.go new file mode 100644 index 0000000..8fbc78e --- /dev/null +++ b/internal/rules/evaluator.go @@ -0,0 +1,133 @@ +package rules + +import ( + "fmt" + "regexp" + "strings" + + "git.pengzhan.dev/haopengzhan/rsspipe/internal/models" +) + +type Evaluator struct{} + +func NewEvaluator() *Evaluator { + return &Evaluator{} +} + +// Evaluate checks whether a FeedEntry satisfies a Rule's conditions. +func (e *Evaluator) Evaluate(rule models.Rule, entry models.FeedEntry) bool { + if len(rule.Conditions) == 0 { + return true + } + + matchMode := strings.ToLower(rule.MatchMode) + if matchMode == "" || matchMode == "all" { + for _, cond := range rule.Conditions { + if !e.evaluateCondition(cond, entry) { + return false + } + } + return true + } + + // matchMode == "any" + for _, cond := range rule.Conditions { + if e.evaluateCondition(cond, entry) { + return true + } + } + return false +} + +func (e *Evaluator) evaluateCondition(cond models.RuleCondition, entry models.FeedEntry) bool { + actualValue := e.resolveField(cond.Field, entry) + return e.compare(actualValue, cond.Operator, cond.Value) +} + +func (e *Evaluator) resolveField(field string, entry models.FeedEntry) interface{} { + switch strings.ToLower(field) { + case "feed_id": + return entry.FeedID + case "feed_title": + return entry.FeedTitle + case "title": + return entry.Title + case "author": + return entry.Author + case "url": + return entry.URL + case "content", "rss_content": + return entry.Content + case "is_read": + return entry.IsRead + case "is_starred": + return entry.IsStarred + } + + // Variant field resolution: "variants.keywords", "variants.readability_full_text", "variants.tags" + if strings.HasPrefix(field, "variants.") { + variantKey := strings.TrimPrefix(field, "variants.") + if entry.Variants != nil { + if v, ok := entry.Variants[variantKey]; ok { + return v.TextValue + } + } + } + + return nil +} + +func (e *Evaluator) compare(actual interface{}, op string, expected interface{}) bool { + if actual == nil { + return op == "neq" + } + + actualStr := fmt.Sprintf("%v", actual) + expectedStr := fmt.Sprintf("%v", expected) + + switch strings.ToLower(op) { + case "eq", "==": + return strings.EqualFold(actualStr, expectedStr) + case "neq", "!=": + return !strings.EqualFold(actualStr, expectedStr) + case "contains": + return strings.Contains(strings.ToLower(actualStr), strings.ToLower(expectedStr)) + case "not_contains": + return !strings.Contains(strings.ToLower(actualStr), strings.ToLower(expectedStr)) + case "in": + // expected can be []interface{}, []string, or comma-separated string + switch v := expected.(type) { + case []interface{}: + for _, item := range v { + if strings.EqualFold(actualStr, fmt.Sprintf("%v", item)) { + return true + } + } + return false + case []string: + for _, item := range v { + if strings.EqualFold(actualStr, item) { + return true + } + } + return false + case string: + items := strings.Split(v, ",") + for _, item := range items { + if strings.EqualFold(strings.TrimSpace(actualStr), strings.TrimSpace(item)) { + return true + } + } + return false + } + return false + case "matches", "regex": + re, err := regexp.Compile(expectedStr) + if err != nil { + return false + } + return re.MatchString(actualStr) + default: + return false + } +} diff --git a/internal/rules/rules_test.go b/internal/rules/rules_test.go new file mode 100644 index 0000000..ff4ee4b --- /dev/null +++ b/internal/rules/rules_test.go @@ -0,0 +1,70 @@ +package rules + +import ( + "testing" + "time" + + "git.pengzhan.dev/haopengzhan/rsspipe/internal/models" +) + +func TestRuleEvaluator(t *testing.T) { + eval := NewEvaluator() + + entry := models.FeedEntry{ + ID: "entry-1", + FeedID: "feed-hn", + FeedTitle: "Hacker News", + Title: "Announcing Go 1.24 Release", + Author: "Go Team", + Content: "Go 1.24 brings new features...
", + Variants: map[string]models.EntryVariant{ + "keywords": { + EntryID: "entry-1", + VariantKey: "keywords", + ContentType: "json", + TextValue: `["go", "compiler", "release"]`, + }, + }, + PublishedAt: time.Now(), + } + + // Test 1: Title contains Go AND FeedID == feed-hn + rule1 := models.Rule{ + ID: "r1", + Name: "Tech Feeds", + MatchMode: "all", + Conditions: []models.RuleCondition{ + {Field: "feed_id", Operator: "eq", Value: "feed-hn"}, + {Field: "title", Operator: "contains", Value: "Go 1.24"}, + }, + } + if !eval.Evaluate(rule1, entry) { + t.Error("expected rule1 to match entry") + } + + // Test 2: Negative condition (author != Python Team) + rule2 := models.Rule{ + ID: "r2", + Name: "Author check", + MatchMode: "all", + Conditions: []models.RuleCondition{ + {Field: "author", Operator: "neq", Value: "Python Team"}, + }, + } + if !eval.Evaluate(rule2, entry) { + t.Error("expected rule2 to match entry") + } + + // Test 3: Variant matching (variants.keywords contains compiler) + rule3 := models.Rule{ + ID: "r3", + Name: "Keywords check", + MatchMode: "all", + Conditions: []models.RuleCondition{ + {Field: "variants.keywords", Operator: "contains", Value: "compiler"}, + }, + } + if !eval.Evaluate(rule3, entry) { + t.Error("expected rule3 to match entry variant") + } +} diff --git a/internal/worker/pool.go b/internal/worker/pool.go new file mode 100644 index 0000000..64b7502 --- /dev/null +++ b/internal/worker/pool.go @@ -0,0 +1,179 @@ +package worker + +import ( + "context" + "fmt" + "sync" + "time" + + "git.pengzhan.dev/haopengzhan/rsspipe/internal/db" + "git.pengzhan.dev/haopengzhan/rsspipe/internal/events" + "git.pengzhan.dev/haopengzhan/rsspipe/internal/logger" + "git.pengzhan.dev/haopengzhan/rsspipe/internal/models" + "go.uber.org/zap" +) + +type ActionResult struct { + ResultVariant *models.EntryVariant + MarkRead *bool + MarkStarred *bool +} + +type ActionHandler func(ctx context.Context, entry models.FeedEntry, params map[string]interface{}) (*ActionResult, error) + +type Task struct { + Action models.RuleAction + Entry models.FeedEntry + Depth int +} + +type Pool struct { + db *db.DB + bus *events.EventBus + handlers map[string]ActionHandler + queue chan Task + workers int + wg sync.WaitGroup + log *zap.Logger + mu sync.RWMutex +} + +func NewPool(database *db.DB, bus *events.EventBus, workers, queueSize int) *Pool { + if workers <= 0 { + workers = 5 + } + if queueSize <= 0 { + queueSize = 100 + } + return &Pool{ + db: database, + bus: bus, + handlers: make(map[string]ActionHandler), + queue: make(chan Task, queueSize), + workers: workers, + log: logger.Named("worker-pool"), + } +} + +func (p *Pool) RegisterHandler(actionType string, handler ActionHandler) { + p.mu.Lock() + defer p.mu.Unlock() + p.handlers[actionType] = handler + p.log.Info("Registered action handler", zap.String("action", actionType)) +} + +func (p *Pool) Start(ctx context.Context) { + p.log.Info("Starting worker pool", zap.Int("worker_count", p.workers)) + for i := 0; i < p.workers; i++ { + p.wg.Add(1) + go p.worker(ctx, i) + } +} + +func (p *Pool) Stop() { + close(p.queue) + p.wg.Wait() + p.log.Info("Worker pool stopped") +} + +func (p *Pool) Dispatch(ctx context.Context, action models.RuleAction, entry models.FeedEntry, depth int) { + select { + case p.queue <- Task{Action: action, Entry: entry, Depth: depth}: + default: + p.log.Warn("Worker pool queue full, task dropped", + zap.String("action", action.Type), + zap.String("entry_id", entry.ID), + ) + } +} + +func (p *Pool) ExecuteDirect(ctx context.Context, actionType string, entry models.FeedEntry, params map[string]interface{}) (*ActionResult, error) { + p.mu.RLock() + handler, ok := p.handlers[actionType] + p.mu.RUnlock() + + if !ok { + return nil, fmt.Errorf("unknown action handler: %s", actionType) + } + + return handler(ctx, entry, params) +} + +func (p *Pool) worker(ctx context.Context, id int) { + defer p.wg.Done() + for { + select { + case <-ctx.Done(): + return + case task, ok := <-p.queue: + if !ok { + return + } + p.processTask(ctx, task) + } + } +} + +func (p *Pool) processTask(ctx context.Context, task Task) { + p.mu.RLock() + handler, ok := p.handlers[task.Action.Type] + p.mu.RUnlock() + + if !ok { + p.log.Warn("No handler registered for action", zap.String("type", task.Action.Type)) + return + } + + start := time.Now() + res, err := handler(ctx, task.Entry, task.Action.Params) + duration := time.Since(start) + + if err != nil { + p.log.Error("Action execution failed", + zap.String("action", task.Action.Type), + zap.String("entry_id", task.Entry.ID), + zap.Duration("duration", duration), + zap.Error(err), + ) + return + } + + if res == nil { + return + } + + // 1. Save Variant if returned + if res.ResultVariant != nil { + if err := p.db.SetVariant(*res.ResultVariant); err != nil { + p.log.Error("Failed to save entry variant", + zap.String("entry_id", task.Entry.ID), + zap.String("variant_key", res.ResultVariant.VariantKey), + zap.Error(err), + ) + } else { + p.log.Info("Entry variant saved", + zap.String("entry_id", task.Entry.ID), + zap.String("variant_key", res.ResultVariant.VariantKey), + zap.Duration("duration", duration), + ) + + // Publish EventVariantUpdated to enable downstream chained triggers + if p.bus != nil { + p.bus.Publish(ctx, events.Event{ + Type: events.EventVariantUpdated, + Entry: task.Entry, + VariantKey: res.ResultVariant.VariantKey, + Depth: task.Depth, + }) + } + } + } + + // 2. Apply Read/Star updates + if res.MarkRead != nil { + _ = p.db.UpdateEntryRead(task.Entry.ID, *res.MarkRead) + } + if res.MarkStarred != nil { + _ = p.db.UpdateEntryStarred(task.Entry.ID, *res.MarkStarred) + } +} diff --git a/web/src/App.tsx b/web/src/App.tsx index 5f75239..bfc4c52 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -159,6 +159,19 @@ export const App: React.FC = () => { setFocusedIndex(entries.findIndex((e) => e.id === selectedEntryId)); return; } + // If not found in current list, load directly so reading pane stays populated + api.getEntry(selectedEntryId).then((entry) => { + setActiveArticle(entry); + }).catch(() => { + // fallback to top + if (entries[0]) { + setFocusedIndex(0); + setActiveArticle(entries[0]); + setSelectedEntryId(entries[0].id); + localStorage.setItem('rsspipe_selected_entry_id', entries[0].id); + } + }); + return; } setFocusedIndex(0); @@ -177,36 +190,7 @@ export const App: React.FC = () => { } }; - // Auto-Refresh & Background Polling (Every 30s + on focus / visibility change) - useEffect(() => { - if (!currentUser) return; - - const interval = setInterval(() => { - loadSidebarData(); - loadEntries(); - }, 30000); - - const handleWindowFocus = () => { - loadSidebarData(); - loadEntries(); - }; - - const handleVisibilityChange = () => { - if (document.visibilityState === 'visible') { - loadSidebarData(); - loadEntries(); - } - }; - - window.addEventListener('focus', handleWindowFocus); - document.addEventListener('visibilitychange', handleVisibilityChange); - - return () => { - clearInterval(interval); - window.removeEventListener('focus', handleWindowFocus); - document.removeEventListener('visibilitychange', handleVisibilityChange); - }; - }, [currentUser, loadSidebarData, loadEntries]); + // Refresh data only on initial mount or when user clicks refresh / changes filter // Toggle Read Action with optimistic update const handleToggleRead = async (entry: FeedEntry, e?: React.MouseEvent) => { @@ -516,6 +500,10 @@ export const App: React.FC = () => { onToggleRead={handleToggleRead} onToggleStar={handleToggleStar} onMarkAllRead={handleMarkAllRead} + onUpdateEntry={(updated) => { + setActiveArticle(updated); + setEntries((prev) => prev.map((e) => (e.id === updated.id ? updated : e))); + }} sidebarCollapsed={sidebarCollapsed} /> ) : viewMode === 'waterfall' ? ( @@ -526,6 +514,10 @@ export const App: React.FC = () => { entries={entries} onToggleRead={handleToggleRead} onToggleStar={handleToggleStar} + onUpdateEntry={(updated) => { + setActiveArticle(updated); + setEntries((prev) => prev.map((e) => (e.id === updated.id ? updated : e))); + }} /> ) : (