feat: implement entry variant tagging, backend action handlers, batch variant loading, and rule application API.

This commit is contained in:
2026-08-25 22:57:42 -07:00
parent d440728ca5
commit 235966e933
6 changed files with 1382 additions and 240 deletions
+43 -1
View File
@@ -16,6 +16,7 @@ import (
"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/rules"
"git.pengzhan.dev/haopengzhan/rsspipe/internal/worker"
"go.uber.org/zap"
@@ -50,9 +51,50 @@ func main() {
bus := events.NewEventBus()
pool := worker.NewPool(database, bus, 5, 100)
// Register Scraper Action
// Register Core Action Handlers
scraper := readability.NewScraper(15 * time.Second)
pool.RegisterHandler("readability_scrape", scraper.Handler())
pool.RegisterHandler("set_variant", func(ctx context.Context, entry models.FeedEntry, params map[string]interface{}) (*worker.ActionResult, error) {
key, _ := params["variant_key"].(string)
if key == "" {
key, _ = params["key"].(string)
}
if key == "" {
return nil, nil
}
val, _ := params["value"].(string)
contentType, _ := params["content_type"].(string)
if contentType == "" {
contentType = "text"
}
variant := models.EntryVariant{
EntryID: entry.ID,
VariantKey: key,
ContentType: contentType,
TextValue: val,
CreatedAt: time.Now(),
UpdatedAt: time.Now(),
}
return &worker.ActionResult{
ResultVariant: &variant,
}, nil
})
pool.RegisterHandler("mark_read", func(ctx context.Context, entry models.FeedEntry, params map[string]interface{}) (*worker.ActionResult, error) {
trueVal := true
return &worker.ActionResult{
MarkRead: &trueVal,
}, nil
})
pool.RegisterHandler("mark_starred", func(ctx context.Context, entry models.FeedEntry, params map[string]interface{}) (*worker.ActionResult, error) {
trueVal := true
return &worker.ActionResult{
MarkStarred: &trueVal,
}, nil
})
pool.Start(ctx)
// 4. Initialize Rules Engine
+150
View File
@@ -0,0 +1,150 @@
# RSSPipe Rules & Automation Hooks Specification (`005_rules_and_hooks_specification.md`)
## 1. Overview & Vision
The **Rules & Hooks Engine** is the reactive event-processing backbone of RSSPipe. It connects the ingestion lifecycle with the **Multi-Variant Data Store** (`entry_variants`), allowing users to automate content enrichment (scraping, tagging, AI, audio), triage (read, starred, archive), and notifications without writing code.
---
## 2. Core Mental Model: 3-Level Execution Flow
Every Rule is evaluated deterministically using a three-level hierarchy:
```mermaid
graph TD
A[Level 1: Event Trigger] --> B{Level 2: Conditions Evaluator}
B -- All / Any Match --> C[Level 3: Action Execution Pipeline]
B -- No Match --> D[No Action]
C --> E[Store Variant / Mutate Status / Dispatch Event]
E -. New Variant Event .-> A
```
---
## 3. Level 1: Event Triggers
The trigger specifies **when** a rule is evaluated.
| Trigger Type | Identifier | Description | Sub-Options / Parameters |
| :--- | :--- | :--- | :--- |
| **New Entry Ingested** | `new_entry` (`on_post_created`) | Fires immediately after an article is discovered and parsed by crawler/OPML. | None |
| **Variant Changed** | `variant_change` (`on_variant_updated`) | Fires when a specific variant key is added, updated, or deleted. | **Sub-Choice (Event Type)**:<br>• `key_added` (Key didn't exist, now created)<br>• `value_changed` (Key existed, value modified)<br>• `key_deleted` (Variant removed)<br>**Target Key Filter**:<br>• Specific `variant_key` (e.g. `keywords`, `readability_full_text`, `ai_summary`) or `*` (any variant). |
| **Cron Schedule (Future)** | `cron` | Recurring background schedule (e.g., daily digests, archival runs). | Cron expression (e.g. `0 8 * * *`) |
| **Manual Execution** | `manual` | Triggered on-demand via UI or API. | None |
---
## 4. Level 2: Conditions & Allowlisted Fields
Rules evaluate conditions against an **allowlist of strongly typed fields**. The UI dynamically presents appropriate operators and input controls based on the field type.
### 4.1 Allowlisted Fields & Operators
| Field Name | Type | Allowed Operators | UI Input Type & Behavior | Example Evaluation |
| :--- | :--- | :--- | :--- | :--- |
| `feed_id` | `string` / `set` | `==`, `!=`, `in`, `not_in` | **Multi-Select Feed Picker** with search, "Select All", and "Invert" buttons. | `feed_id in ["feed-1", "feed-2", ...]` (Applies to 24/30 feeds in 2 clicks) |
| `category_id` | `string` / `set` | `==`, `!=`, `in`, `not_in` | **Category Dropdown / Multi-Select Chips**. | `category_id in ["cat-tech", "cat-ai"]` |
| `title` | `string` | `contains`, `not_contains`, `==`, `!=`, `matches` (regex), `starts_with`, `ends_with` | Text input or comma-separated token badges. | `title contains "DeepSeek"` |
| `raw_content` / `content` | `string` | `contains`, `not_contains`, `matches` (regex) | Text input. | `content contains "breaking news"` |
| `author` | `string` | `contains`, `not_contains`, `==`, `!=` | Text input. | `author == "Paul Graham"` |
| `created_at` / `published_at` | `timestamp` | `>=`, `<=`, `within_last_hours`, `older_than_days` | Number + unit dropdown (e.g. "Within last 24 hours"). | `published_at within_last_hours 48` |
| `is_read` | `boolean` | `==`, `!=` | Toggle (Read / Unread). | `is_read == false` |
| `is_starred` | `boolean` | `==`, `!=` | Toggle (Starred / Not Starred). | `is_starred == true` |
| `variants.<key>` | `variant_value` | `exists`, `not_exists`, `contains`, `not_contains`, `==`, `!=` | Key name + value input. | `variants.keywords contains "LLM"` |
### 4.2 Match Modes
- **`all` (AND)**: Every condition in the rule must evaluate to true.
- **`any` (OR)**: At least one condition must evaluate to true.
---
## 5. Level 3: Action Execution Pipeline
When a rule's conditions are satisfied, one or more actions execute sequentially.
### 5.1 Supported Core Actions
| Action Identifier | Parameters | Description |
| :--- | :--- | :--- |
| **`set_variant`** | `variant_key` (string), `value` (string/json/yaml), `content_type` (string) | Adds or updates a key-value variant (e.g., adding tags, keywords, custom annotations). |
| **`readability_scrape`** | None | Fetches webpage full-text HTML and stores in `entry_variants(key='readability_full_text')`. |
| **`mark_read`** | None | Sets `is_read = true` (auto-triage/archive). |
| **`mark_starred`** | None | Sets `is_starred = true`. |
---
## 6. Safety Architecture: Endless Loop Prevention
Because actions like `set_variant` emit `on_variant_updated` events, rules could theoretically trigger cyclic loops (e.g., Rule A triggers on `variant_change` for `key="tags"` and adds a new tag to `key="tags"`).
### Loop Guard Mechanisms
1. **Static Validation (Rule Creation Time)**:
- If Trigger is `variant_change` on `key="X"` and an Action is `set_variant` on `key="X"`, the UI / API warns and rejects identical recursive rules unless an explicit guard predicate (e.g., `value not contains "Y"`) is specified.
2. **Runtime Execution Context & Depth Guard**:
- Every event emitted by the Event Bus carries a `RecursionDepth: int` (max depth = 5) and `ExecutionPath: []string` tracking rule IDs.
- If an event exceeds depth 5 or encounters the same Rule ID in its call stack, execution is aborted with an error log: `WARN: Cyclic rule execution aborted for rule: {rule_id}`.
---
## 7. UI/UX Design Specification for Settings
### 7.1 Rule List View
- Clean card layout grouped by Trigger type (`New Entry`, `Variant Change`, `Cron`).
- Card shows a natural language sentence:
> *"When **New Post Arrives** in **24 Feeds** and **Title contains 'AI'** $\rightarrow$ **Scrape Full Text**, **Mark Starred**"*
- Toggle switch (Active/Inactive), Edit button, Test button, Delete button.
### 7.2 Interactive Rule Builder Modal Layout
```
┌────────────────────────────────────────────────────────────────────────┐
│ Create Automation Rule [ ✕ ] │
├────────────────────────────────────────────────────────────────────────┤
│ Rule Name: [ Auto-Scrape Tech Feeds ] │
│ │
│ 1. WHEN (Trigger) │
│ (●) New Entry Arrives ( ) Variant Changed ( ) Schedule │
│ │
│ [If Variant Changed Selected] │
│ Event: [ Key Added ▼ ] Target Key: [ keywords ] │
│ │
│ 2. IF (Conditions) Match: [ All Conditions (AND) ]│
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ Field: [ Feed ID ▼ ] Op: [ In List (24 Feeds) ▼ ] │ │
│ │ [ Quick Select: All | None | Invert ] │ │
│ │ [✓] Hacker News [✓] 36Kr [✓] FT Chinese [✓] InfoQ [ ] Douban│ │
│ │ [ ✕ Remove]│ │
│ ├─────────────────────────────────────────────────────────────────┤ │
│ │ Field: [ Title ▼ ] Op: [ Contains Any ▼ ] │ │
│ │ Value: [ AI, LLM, Machine Learning ] │ │
│ │ [ ✕ Remove]│ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ [ + Add Condition ] │
│ │
│ 3. THEN DO (Actions) │
│ [✓] ⚡ Scrape Full Text (Readability) │
│ [✓] ⭐ Mark as Starred │
│ [ ] 🏷️ Set Variant / Tag -> Key: [tags ] Val: [ai-digest ] │
│ [ ] ✓ Mark as Read │
│ │
│ [ Cancel ] [ Create Rule ] │
└────────────────────────────────────────────────────────────────────────┘
```
---
## 8. Backlog Roadmap for Rules & Hooks
The following capabilities are reserved for upcoming iterations:
1. **Auto-Delete / Retention Rule Action** (`delete_entry`):
- Scope: Automatically delete matching entries older than $N$ days (requires retention policy discussion).
2. **Aggregated Shares & Auto-Share Action** (`share_entry`):
- Scope: Action to automatically generate public share links for high-priority articles, plus a dedicated `/shares` dashboard aggregating all shared items.
3. **Advanced AI & Audio Pipeline Actions**:
- `generate_ai_summary`: Calls LLM to summarize content $\rightarrow$ stores in `entry_variants.ai_summary`.
- `generate_tts`: Synthesizes article voice audio $\rightarrow$ stores `.mp3` blob in `./data/blobs/{id}/tts.mp3`.
- `generate_podcast`: Generates dual-speaker conversational audio discussions.
+17 -2
View File
@@ -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/rules"
"git.pengzhan.dev/haopengzhan/rsspipe/internal/worker"
"github.com/go-chi/chi/v5"
)
@@ -662,8 +663,13 @@ func (h *Handler) CreateRule(w http.ResponseWriter, r *http.Request) {
if strings.TrimSpace(body.Trigger) == "" {
body.Trigger = "on_post_created"
}
if strings.TrimSpace(body.MatchMode) == "" {
body.MatchMode = "all"
if err := rules.ValidateRuleSafety(models.Rule{
Trigger: body.Trigger,
Conditions: body.Conditions,
Actions: body.Actions,
}); err != nil {
h.writeError(w, http.StatusBadRequest, err.Error())
return
}
rule, err := h.db.CreateRule(body.Name, body.Trigger, body.MatchMode, body.Conditions, body.Actions)
@@ -691,6 +697,15 @@ func (h *Handler) UpdateRule(w http.ResponseWriter, r *http.Request) {
return
}
if err := rules.ValidateRuleSafety(models.Rule{
Trigger: body.Trigger,
Conditions: body.Conditions,
Actions: body.Actions,
}); err != nil {
h.writeError(w, http.StatusBadRequest, err.Error())
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())
+161 -20
View File
@@ -3,7 +3,9 @@ package rules
import (
"fmt"
"regexp"
"strconv"
"strings"
"time"
"git.pengzhan.dev/haopengzhan/rsspipe/internal/models"
)
@@ -56,12 +58,19 @@ func (e *Evaluator) resolveField(field string, entry models.FeedEntry) interface
return entry.Author
case "url":
return entry.URL
case "content", "rss_content":
case "content", "raw_content", "rss_content":
if entry.Content != "" {
return entry.Content
}
return entry.RawContent
case "is_read":
return entry.IsRead
case "is_starred":
return entry.IsStarred
case "published_at":
return entry.PublishedAt
case "created_at":
return entry.CreatedAt
}
// Variant field resolution: "variants.keywords", "variants.readability_full_text", "variants.tags"
@@ -78,14 +87,41 @@ func (e *Evaluator) resolveField(field string, entry models.FeedEntry) interface
}
func (e *Evaluator) compare(actual interface{}, op string, expected interface{}) bool {
opLower := strings.ToLower(strings.TrimSpace(op))
// Handle exists / not_exists operators
if opLower == "exists" {
if actual == nil {
return op == "neq"
return false
}
if s, ok := actual.(string); ok && s == "" {
return false
}
return true
}
if opLower == "not_exists" {
if actual == nil {
return true
}
if s, ok := actual.(string); ok && s == "" {
return true
}
return false
}
if actual == nil {
return opLower == "neq" || opLower == "!=" || opLower == "not_in" || opLower == "not_contains"
}
// Handle Time/Timestamp comparisons
if actualTime, ok := actual.(time.Time); ok {
return e.compareTime(actualTime, opLower, expected)
}
actualStr := fmt.Sprintf("%v", actual)
expectedStr := fmt.Sprintf("%v", expected)
switch strings.ToLower(op) {
switch opLower {
case "eq", "==":
return strings.EqualFold(actualStr, expectedStr)
case "neq", "!=":
@@ -94,33 +130,34 @@ func (e *Evaluator) compare(actual interface{}, op string, expected interface{})
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)) {
case "starts_with":
return strings.HasPrefix(strings.ToLower(actualStr), strings.ToLower(expectedStr))
case "ends_with":
return strings.HasSuffix(strings.ToLower(actualStr), strings.ToLower(expectedStr))
case "contains_any":
tokens := parseStringList(expected)
for _, token := range tokens {
if strings.Contains(strings.ToLower(actualStr), strings.ToLower(token)) {
return true
}
}
return false
case []string:
for _, item := range v {
case "in":
tokens := parseStringList(expected)
for _, item := range tokens {
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)) {
case "not_in":
tokens := parseStringList(expected)
for _, item := range tokens {
if strings.EqualFold(actualStr, item) {
return false
}
}
return true
}
}
return false
}
return false
case "matches", "regex":
re, err := regexp.Compile(expectedStr)
if err != nil {
@@ -131,3 +168,107 @@ func (e *Evaluator) compare(actual interface{}, op string, expected interface{})
return false
}
}
func (e *Evaluator) compareTime(t time.Time, op string, expected interface{}) bool {
now := time.Now()
switch op {
case "within_last_hours":
hours, err := toFloat64(expected)
if err != nil {
return false
}
cutoff := now.Add(-time.Duration(hours * float64(time.Hour)))
return t.After(cutoff)
case "older_than_days":
days, err := toFloat64(expected)
if err != nil {
return false
}
cutoff := now.Add(-time.Duration(days * 24 * float64(time.Hour)))
return t.Before(cutoff)
case ">=", "after":
targetTime, err := parseTime(expected)
if err != nil {
return false
}
return t.After(targetTime) || t.Equal(targetTime)
case "<=", "before":
targetTime, err := parseTime(expected)
if err != nil {
return false
}
return t.Before(targetTime) || t.Equal(targetTime)
}
return false
}
func parseStringList(val interface{}) []string {
var result []string
switch v := val.(type) {
case []interface{}:
for _, item := range v {
result = append(result, strings.TrimSpace(fmt.Sprintf("%v", item)))
}
case []string:
for _, item := range v {
result = append(result, strings.TrimSpace(item))
}
case string:
parts := strings.Split(v, ",")
for _, part := range parts {
trimmed := strings.TrimSpace(part)
if trimmed != "" {
result = append(result, trimmed)
}
}
}
return result
}
func toFloat64(val interface{}) (float64, error) {
switch v := val.(type) {
case float64:
return v, nil
case int:
return float64(v), nil
case int64:
return float64(v), nil
case string:
return strconv.ParseFloat(strings.TrimSpace(v), 64)
}
return 0, fmt.Errorf("cannot convert %v to float64", val)
}
func parseTime(val interface{}) (time.Time, error) {
if t, ok := val.(time.Time); ok {
return t, nil
}
s := fmt.Sprintf("%v", val)
return time.Parse(time.RFC3339, s)
}
// ValidateRuleSafety checks for potential cyclic/endless loops at rule creation time.
func ValidateRuleSafety(rule models.Rule) error {
trigger := strings.ToLower(rule.Trigger)
// If trigger is on_variant_updated / variant_change, check for recursive set_variant on same key
if trigger == "on_variant_updated" || trigger == "variant_change" {
for _, action := range rule.Actions {
if action.Type == "set_variant" {
targetKey, _ := action.Params["variant_key"].(string)
if targetKey == "" {
targetKey, _ = action.Params["key"].(string)
}
// Check if trigger targets this exact key without safe conditions
for _, cond := range rule.Conditions {
if cond.Field == "variants."+targetKey && (cond.Operator == "exists" || cond.Operator == "eq") {
return fmt.Errorf("cyclic rule detected: rule triggers on variant '%s' and unconditionally writes to '%s'", targetKey, targetKey)
}
}
}
}
}
return nil
}
+59 -11
View File
@@ -25,16 +25,17 @@ func TestRuleEvaluator(t *testing.T) {
TextValue: `["go", "compiler", "release"]`,
},
},
PublishedAt: time.Now(),
PublishedAt: time.Now().Add(-2 * time.Hour),
CreatedAt: time.Now().Add(-2 * time.Hour),
}
// Test 1: Title contains Go AND FeedID == feed-hn
// Test 1: Title contains Go AND FeedID in list
rule1 := models.Rule{
ID: "r1",
Name: "Tech Feeds",
Name: "Tech Feeds in list",
MatchMode: "all",
Conditions: []models.RuleCondition{
{Field: "feed_id", Operator: "eq", Value: "feed-hn"},
{Field: "feed_id", Operator: "in", Value: []string{"feed-hn", "feed-36kr", "feed-ft"}},
{Field: "title", Operator: "contains", Value: "Go 1.24"},
},
}
@@ -42,29 +43,76 @@ func TestRuleEvaluator(t *testing.T) {
t.Error("expected rule1 to match entry")
}
// Test 2: Negative condition (author != Python Team)
// Test 2: not_in operator
rule2 := models.Rule{
ID: "r2",
Name: "Author check",
Name: "Feed not in noise",
MatchMode: "all",
Conditions: []models.RuleCondition{
{Field: "author", Operator: "neq", Value: "Python Team"},
{Field: "feed_id", Operator: "not_in", Value: "feed-spam,feed-ads"},
},
}
if !eval.Evaluate(rule2, entry) {
t.Error("expected rule2 to match entry")
}
// Test 3: Variant matching (variants.keywords contains compiler)
// Test 3: contains_any operator
rule3 := models.Rule{
ID: "r3",
Name: "Keywords check",
Name: "Title match any keyword",
MatchMode: "all",
Conditions: []models.RuleCondition{
{Field: "variants.keywords", Operator: "contains", Value: "compiler"},
{Field: "title", Operator: "contains_any", Value: "Rust, Python, Go 1.24"},
},
}
if !eval.Evaluate(rule3, entry) {
t.Error("expected rule3 to match entry variant")
t.Error("expected rule3 to match entry with contains_any")
}
// Test 4: exists operator on variants
rule4 := models.Rule{
ID: "r4",
Name: "Keywords exist",
MatchMode: "all",
Conditions: []models.RuleCondition{
{Field: "variants.keywords", Operator: "exists"},
{Field: "variants.nonexistent", Operator: "not_exists"},
},
}
if !eval.Evaluate(rule4, entry) {
t.Error("expected rule4 to evaluate variant existence correctly")
}
// Test 5: Timestamp within_last_hours
rule5 := models.Rule{
ID: "r5",
Name: "Recent article",
MatchMode: "all",
Conditions: []models.RuleCondition{
{Field: "published_at", Operator: "within_last_hours", Value: 24},
},
}
if !eval.Evaluate(rule5, entry) {
t.Error("expected rule5 to match article within last 24h")
}
// Test 6: Safety validator
unsafeRule := models.Rule{
Trigger: "variant_change",
Conditions: []models.RuleCondition{
{Field: "variants.tags", Operator: "exists"},
},
Actions: []models.RuleAction{
{
Type: "set_variant",
Params: map[string]interface{}{
"variant_key": "tags",
"value": "new-tag",
},
},
},
}
if err := ValidateRuleSafety(unsafeRule); err == nil {
t.Error("expected ValidateRuleSafety to detect cyclic loop")
}
}
File diff suppressed because it is too large Load Diff