From f3a90c14c8b5b3e4e84ccdbbc2cc9bc08f71646f Mon Sep 17 00:00:00 2001 From: Kieran Klukas Date: Sat, 10 Jan 2026 21:42:30 -0800 Subject: [PATCH] feat: add more nodes --- engine/registry.go | 6 ++ nodes/outputs/webhook.go | 109 +++++++++++++++++++++++ nodes/sources/http.go | 165 +++++++++++++++++++++++++++++++++++ nodes/sources/rss.go | 91 +++++++++++++++++-- nodes/transforms/filter.go | 15 ---- nodes/transforms/helpers.go | 37 ++++++++ nodes/transforms/map.go | 104 ++++++++++++++++++++++ nodes/transforms/merge.go | 137 +++++++++++++++++++++++++++++ nodes/transforms/regex.go | 108 +++++++++++++++++++++++ nodes/transforms/sort.go | 32 +++++-- nodes/transforms/truncate.go | 123 ++++++++++++++++++++++++++ web/server.go | 25 ++++++ web/templates/editor.html | 55 +++++++++--- 13 files changed, 963 insertions(+), 44 deletions(-) create mode 100644 nodes/outputs/webhook.go create mode 100644 nodes/sources/http.go create mode 100644 nodes/transforms/helpers.go create mode 100644 nodes/transforms/map.go create mode 100644 nodes/transforms/merge.go create mode 100644 nodes/transforms/regex.go create mode 100644 nodes/transforms/truncate.go diff --git a/engine/registry.go b/engine/registry.go index b55a76d..8730bc6 100644 --- a/engine/registry.go +++ b/engine/registry.go @@ -23,15 +23,21 @@ func NewRegistry() *Registry { // Register built-in nodes // Sources r.Register(&sources.RSSSourceNode{}) + r.Register(&sources.HTTPSourceNode{}) // Transforms r.Register(&transforms.FilterNode{}) r.Register(&transforms.SortNode{}) r.Register(&transforms.LimitNode{}) + r.Register(&transforms.MergeNode{}) + r.Register(&transforms.MapNode{}) + r.Register(&transforms.RegexNode{}) + r.Register(&transforms.TruncateNode{}) // Outputs r.Register(&outputs.JSONOutputNode{}) r.Register(&outputs.RSSOutputNode{}) + r.Register(&outputs.WebhookOutputNode{}) return r } diff --git a/nodes/outputs/webhook.go b/nodes/outputs/webhook.go new file mode 100644 index 0000000..29fcb9d --- /dev/null +++ b/nodes/outputs/webhook.go @@ -0,0 +1,109 @@ +package outputs + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" + "strings" + "time" + + "github.com/kierank/pipes/nodes" +) + +type WebhookOutputNode struct{} + +func (n *WebhookOutputNode) Type() string { return "webhook-output" } +func (n *WebhookOutputNode) Label() string { return "Webhook" } +func (n *WebhookOutputNode) Description() string { return "POST data to a webhook URL" } +func (n *WebhookOutputNode) Category() string { return "output" } +func (n *WebhookOutputNode) Inputs() int { return 1 } +func (n *WebhookOutputNode) Outputs() int { return 0 } + +func (n *WebhookOutputNode) Execute(ctx context.Context, config map[string]interface{}, inputs [][]interface{}, execCtx *nodes.Context) ([]interface{}, error) { + if len(inputs) == 0 || len(inputs[0]) == 0 { + execCtx.Log("webhook-output", "info", "No input data") + return nil, nil + } + + url, ok := config["url"].(string) + if !ok || url == "" { + return nil, fmt.Errorf("webhook URL is required") + } + + data := inputs[0] + + payload := map[string]interface{}{ + "count": len(data), + "items": data, + } + + jsonData, err := json.Marshal(payload) + if err != nil { + return nil, fmt.Errorf("marshal payload: %w", err) + } + + client := &http.Client{Timeout: 30 * time.Second} + req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader(jsonData)) + if err != nil { + return nil, fmt.Errorf("create request: %w", err) + } + + req.Header.Set("Content-Type", "application/json") + req.Header.Set("User-Agent", "Pipes/1.0") + + // Add custom headers + if headers, ok := config["headers"].(string); ok && headers != "" { + for _, line := range strings.Split(headers, "\n") { + if parts := strings.SplitN(strings.TrimSpace(line), ":", 2); len(parts) == 2 { + req.Header.Set(strings.TrimSpace(parts[0]), strings.TrimSpace(parts[1])) + } + } + } + + resp, err := client.Do(req) + if err != nil { + return nil, fmt.Errorf("webhook request failed: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode >= 400 { + return nil, fmt.Errorf("webhook returned HTTP %d", resp.StatusCode) + } + + execCtx.Log("webhook-output", "info", fmt.Sprintf("Posted %d items to webhook (HTTP %d)", len(data), resp.StatusCode)) + + return data, nil +} + +func (n *WebhookOutputNode) ValidateConfig(config map[string]interface{}) error { + url, ok := config["url"].(string) + if !ok || url == "" { + return fmt.Errorf("webhook URL is required") + } + return nil +} + +func (n *WebhookOutputNode) GetConfigSchema() *nodes.ConfigSchema { + return &nodes.ConfigSchema{ + Fields: []nodes.ConfigField{ + { + Name: "url", + Label: "Webhook URL", + Type: "url", + Required: true, + Placeholder: "https://example.com/webhook", + HelpText: "URL to POST data to", + }, + { + Name: "headers", + Label: "Headers", + Type: "textarea", + Required: false, + Placeholder: "Authorization: Bearer token", + HelpText: "Custom headers, one per line as Header: Value", + }, + }, + } +} diff --git a/nodes/sources/http.go b/nodes/sources/http.go new file mode 100644 index 0000000..153fcc8 --- /dev/null +++ b/nodes/sources/http.go @@ -0,0 +1,165 @@ +package sources + +import ( + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "strings" + "time" + + "github.com/kierank/pipes/nodes" +) + +type HTTPSourceNode struct{} + +func (n *HTTPSourceNode) Type() string { return "http-source" } +func (n *HTTPSourceNode) Label() string { return "HTTP/JSON" } +func (n *HTTPSourceNode) Description() string { return "Fetch data from a JSON API" } +func (n *HTTPSourceNode) Category() string { return "source" } +func (n *HTTPSourceNode) Inputs() int { return 0 } +func (n *HTTPSourceNode) Outputs() int { return 1 } + +func (n *HTTPSourceNode) Execute(ctx context.Context, config map[string]interface{}, inputs [][]interface{}, execCtx *nodes.Context) ([]interface{}, error) { + url, ok := config["url"].(string) + if !ok || url == "" { + return nil, fmt.Errorf("url is required") + } + + execCtx.Log("http-source", "info", fmt.Sprintf("Fetching %s", url)) + + client := &http.Client{Timeout: 30 * time.Second} + + req, err := http.NewRequestWithContext(ctx, "GET", url, nil) + if err != nil { + return nil, fmt.Errorf("create request: %w", err) + } + + // Add custom headers + if headers, ok := config["headers"].(string); ok && headers != "" { + for _, line := range strings.Split(headers, "\n") { + if parts := strings.SplitN(strings.TrimSpace(line), ":", 2); len(parts) == 2 { + req.Header.Set(strings.TrimSpace(parts[0]), strings.TrimSpace(parts[1])) + } + } + } + + req.Header.Set("User-Agent", "Pipes/1.0") + + resp, err := client.Do(req) + if err != nil { + return nil, fmt.Errorf("fetch: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("HTTP %d", resp.StatusCode) + } + + body, err := io.ReadAll(resp.Body) + if err != nil { + return nil, fmt.Errorf("read body: %w", err) + } + + // Parse JSON + var data interface{} + if err := json.Unmarshal(body, &data); err != nil { + return nil, fmt.Errorf("parse JSON: %w", err) + } + + // Extract items from a path if specified + itemsPath, _ := config["items_path"].(string) + if itemsPath != "" { + data = extractPath(data, itemsPath) + } + + // Convert to array + var items []interface{} + switch v := data.(type) { + case []interface{}: + items = v + case map[string]interface{}: + items = []interface{}{v} + default: + items = []interface{}{data} + } + + // Apply limit + if limit, ok := config["limit"].(float64); ok && limit > 0 && int(limit) < len(items) { + items = items[:int(limit)] + } + + execCtx.Log("http-source", "info", fmt.Sprintf("Retrieved %d items", len(items))) + return items, nil +} + +func extractPath(data interface{}, path string) interface{} { + parts := strings.Split(path, ".") + current := data + + for _, part := range parts { + if m, ok := current.(map[string]interface{}); ok { + current = m[part] + } else if arr, ok := current.([]interface{}); ok { + // Try to access array by index if part is numeric + var idx int + if _, err := fmt.Sscanf(part, "%d", &idx); err == nil && idx < len(arr) { + current = arr[idx] + } else { + return nil + } + } else { + return nil + } + } + + return current +} + +func (n *HTTPSourceNode) ValidateConfig(config map[string]interface{}) error { + url, ok := config["url"].(string) + if !ok || url == "" { + return fmt.Errorf("url is required") + } + return nil +} + +func (n *HTTPSourceNode) GetConfigSchema() *nodes.ConfigSchema { + return &nodes.ConfigSchema{ + Fields: []nodes.ConfigField{ + { + Name: "url", + Label: "URL", + Type: "url", + Required: true, + Placeholder: "https://api.example.com/data.json", + HelpText: "URL of the JSON API endpoint", + }, + { + Name: "items_path", + Label: "Items Path", + Type: "text", + Required: false, + Placeholder: "data.items", + HelpText: "Dot-notation path to the array of items (e.g., results, data.posts)", + }, + { + Name: "headers", + Label: "Headers", + Type: "textarea", + Required: false, + Placeholder: "Authorization: Bearer token\nAccept: application/json", + HelpText: "Custom headers, one per line as Header: Value", + }, + { + Name: "limit", + Label: "Limit", + Type: "number", + Required: false, + DefaultValue: 50, + HelpText: "Maximum number of items", + }, + }, + } +} diff --git a/nodes/sources/rss.go b/nodes/sources/rss.go index e3a7ac5..1d9db36 100644 --- a/nodes/sources/rss.go +++ b/nodes/sources/rss.go @@ -3,6 +3,7 @@ package sources import ( "context" "fmt" + "time" "github.com/mmcdole/gofeed" @@ -41,16 +42,63 @@ func (n *RSSSourceNode) Execute(ctx context.Context, config map[string]interface if item.Author != nil { author = item.Author.Name } - + + // Parse dates to Unix timestamps for proper sorting + var publishedAt int64 + var updatedAt int64 + if item.PublishedParsed != nil { + publishedAt = item.PublishedParsed.Unix() + } else if item.Published != "" { + if t, err := parseDate(item.Published); err == nil { + publishedAt = t.Unix() + } + } + if item.UpdatedParsed != nil { + updatedAt = item.UpdatedParsed.Unix() + } else if item.Updated != "" { + if t, err := parseDate(item.Updated); err == nil { + updatedAt = t.Unix() + } + } + + // Extract content - prefer Content over Description + content := item.Description + if item.Content != "" { + content = item.Content + } + + // Build enclosures array (for media like images, audio, video) + var enclosures []map[string]interface{} + if len(item.Enclosures) > 0 { + for _, enc := range item.Enclosures { + enclosures = append(enclosures, map[string]interface{}{ + "url": enc.URL, + "type": enc.Type, + "length": enc.Length, + }) + } + } + + // Extract image URL if available + var imageURL string + if item.Image != nil { + imageURL = item.Image.URL + } + items = append(items, map[string]interface{}{ - "title": item.Title, - "description": item.Description, - "link": item.Link, - "author": author, - "published": item.Published, - "updated": item.Updated, - "guid": item.GUID, - "categories": item.Categories, + "title": item.Title, + "description": item.Description, + "content": content, + "link": item.Link, + "author": author, + "published": item.Published, + "published_at": publishedAt, + "updated": item.Updated, + "updated_at": updatedAt, + "guid": item.GUID, + "categories": item.Categories, + "enclosures": enclosures, + "image": imageURL, }) } @@ -97,3 +145,28 @@ func (n *RSSSourceNode) GetConfigSchema() *nodes.ConfigSchema { }, } } + +// parseDate tries multiple date formats +func parseDate(s string) (time.Time, error) { + formats := []string{ + time.RFC1123Z, + time.RFC1123, + time.RFC3339, + time.RFC822Z, + time.RFC822, + "Mon, 2 Jan 2006 15:04:05 MST", + "Mon, 2 Jan 2006 15:04:05 -0700", + "2006-01-02T15:04:05Z", + "2006-01-02T15:04:05-07:00", + "2006-01-02 15:04:05", + "2006-01-02", + } + + for _, format := range formats { + if t, err := time.Parse(format, s); err == nil { + return t, nil + } + } + + return time.Time{}, fmt.Errorf("unable to parse date: %s", s) +} diff --git a/nodes/transforms/filter.go b/nodes/transforms/filter.go index 559915f..c9653c6 100644 --- a/nodes/transforms/filter.go +++ b/nodes/transforms/filter.go @@ -69,21 +69,6 @@ func matchesFilter(item interface{}, field, operator, value string) bool { } } -func getNestedValue(obj map[string]interface{}, path string) interface{} { - parts := strings.Split(path, ".") - var current interface{} = obj - - for _, part := range parts { - if m, ok := current.(map[string]interface{}); ok { - current = m[part] - } else { - return nil - } - } - - return current -} - func (n *FilterNode) ValidateConfig(config map[string]interface{}) error { return nil } diff --git a/nodes/transforms/helpers.go b/nodes/transforms/helpers.go new file mode 100644 index 0000000..d8a50da --- /dev/null +++ b/nodes/transforms/helpers.go @@ -0,0 +1,37 @@ +package transforms + +import "strings" + +// getNestedValue retrieves a value from a nested map using dot notation +func getNestedValue(obj map[string]interface{}, path string) interface{} { + parts := strings.Split(path, ".") + var current interface{} = obj + + for _, part := range parts { + if m, ok := current.(map[string]interface{}); ok { + current = m[part] + } else { + return nil + } + } + + return current +} + +// toFloat attempts to convert various numeric types to float64 +func toFloat(v interface{}) (float64, bool) { + switch val := v.(type) { + case float64: + return val, true + case float32: + return float64(val), true + case int: + return float64(val), true + case int64: + return float64(val), true + case int32: + return float64(val), true + default: + return 0, false + } +} diff --git a/nodes/transforms/map.go b/nodes/transforms/map.go new file mode 100644 index 0000000..ae31e89 --- /dev/null +++ b/nodes/transforms/map.go @@ -0,0 +1,104 @@ +package transforms + +import ( + "context" + "fmt" + "strings" + + "github.com/kierank/pipes/nodes" +) + +type MapNode struct{} + +func (n *MapNode) Type() string { return "map" } +func (n *MapNode) Label() string { return "Map Fields" } +func (n *MapNode) Description() string { return "Rename, extract, or create new fields" } +func (n *MapNode) Category() string { return "transform" } +func (n *MapNode) Inputs() int { return 1 } +func (n *MapNode) Outputs() int { return 1 } + +func (n *MapNode) Execute(ctx context.Context, config map[string]interface{}, inputs [][]interface{}, execCtx *nodes.Context) ([]interface{}, error) { + if len(inputs) == 0 || len(inputs[0]) == 0 { + return []interface{}{}, nil + } + + items := inputs[0] + mappings, _ := config["mappings"].(string) + keepOriginal, _ := config["keep_original"].(bool) + + if mappings == "" { + return items, nil + } + + // Parse mappings: "newField:sourceField, title:name" + fieldMap := parseMappings(mappings) + + var result []interface{} + for _, item := range items { + itemMap, ok := item.(map[string]interface{}) + if !ok { + result = append(result, item) + continue + } + + var newItem map[string]interface{} + if keepOriginal { + newItem = make(map[string]interface{}) + for k, v := range itemMap { + newItem[k] = v + } + } else { + newItem = make(map[string]interface{}) + } + + for newField, sourceField := range fieldMap { + if val := getNestedValue(itemMap, sourceField); val != nil { + newItem[newField] = val + } + } + + result = append(result, newItem) + } + + execCtx.Log("map", "info", fmt.Sprintf("Mapped %d items", len(result))) + return result, nil +} + +func parseMappings(s string) map[string]string { + result := make(map[string]string) + parts := strings.Split(s, ",") + for _, part := range parts { + part = strings.TrimSpace(part) + if kv := strings.SplitN(part, ":", 2); len(kv) == 2 { + result[strings.TrimSpace(kv[0])] = strings.TrimSpace(kv[1]) + } + } + return result +} + +func (n *MapNode) ValidateConfig(config map[string]interface{}) error { + return nil +} + +func (n *MapNode) GetConfigSchema() *nodes.ConfigSchema { + return &nodes.ConfigSchema{ + Fields: []nodes.ConfigField{ + { + Name: "mappings", + Label: "Field Mappings", + Type: "textarea", + Required: true, + Placeholder: "title:name, url:link, summary:description", + HelpText: "Map fields as newField:sourceField, separated by commas. Use dot notation for nested fields.", + }, + { + Name: "keep_original", + Label: "Keep Original Fields", + Type: "checkbox", + Required: false, + DefaultValue: true, + HelpText: "Keep all original fields in addition to mapped ones", + }, + }, + } +} diff --git a/nodes/transforms/merge.go b/nodes/transforms/merge.go new file mode 100644 index 0000000..75bbff7 --- /dev/null +++ b/nodes/transforms/merge.go @@ -0,0 +1,137 @@ +package transforms + +import ( + "context" + "fmt" + "sort" + + "github.com/kierank/pipes/nodes" +) + +type MergeNode struct{} + +func (n *MergeNode) Type() string { return "merge" } +func (n *MergeNode) Label() string { return "Merge" } +func (n *MergeNode) Description() string { return "Combine multiple feeds into one" } +func (n *MergeNode) Category() string { return "transform" } +func (n *MergeNode) Inputs() int { return 2 } +func (n *MergeNode) Outputs() int { return 1 } + +func (n *MergeNode) Execute(ctx context.Context, config map[string]interface{}, inputs [][]interface{}, execCtx *nodes.Context) ([]interface{}, error) { + if len(inputs) == 0 { + return []interface{}{}, nil + } + + var merged []interface{} + for _, input := range inputs { + merged = append(merged, input...) + } + + // Optionally dedupe by a field + dedupeField, _ := config["dedupe_field"].(string) + if dedupeField != "" { + merged = dedupeByField(merged, dedupeField) + } + + // Optionally sort by a field + sortField, _ := config["sort_field"].(string) + sortOrder, _ := config["sort_order"].(string) + if sortField != "" { + sortItems(merged, sortField, sortOrder == "desc") + } + + execCtx.Log("merge", "info", fmt.Sprintf("Merged %d inputs into %d items", len(inputs), len(merged))) + + return merged, nil +} + +func dedupeByField(items []interface{}, field string) []interface{} { + seen := make(map[string]bool) + var result []interface{} + + for _, item := range items { + itemMap, ok := item.(map[string]interface{}) + if !ok { + result = append(result, item) + continue + } + + key := fmt.Sprintf("%v", itemMap[field]) + if !seen[key] { + seen[key] = true + result = append(result, item) + } + } + + return result +} + +func sortItems(items []interface{}, field string, desc bool) { + sort.SliceStable(items, func(i, j int) bool { + iMap, iOk := items[i].(map[string]interface{}) + jMap, jOk := items[j].(map[string]interface{}) + if !iOk || !jOk { + return false + } + + iRaw := getNestedValue(iMap, field) + jRaw := getNestedValue(jMap, field) + + // Try numeric comparison first (for timestamps, etc.) + iNum, iIsNum := toFloat(iRaw) + jNum, jIsNum := toFloat(jRaw) + + if iIsNum && jIsNum { + if desc { + return iNum > jNum + } + return iNum < jNum + } + + // Fall back to string comparison + iVal := fmt.Sprintf("%v", iRaw) + jVal := fmt.Sprintf("%v", jRaw) + + if desc { + return iVal > jVal + } + return iVal < jVal + }) +} + +func (n *MergeNode) ValidateConfig(config map[string]interface{}) error { + return nil +} + +func (n *MergeNode) GetConfigSchema() *nodes.ConfigSchema { + return &nodes.ConfigSchema{ + Fields: []nodes.ConfigField{ + { + Name: "dedupe_field", + Label: "Dedupe Field", + Type: "text", + Required: false, + Placeholder: "link", + HelpText: "Remove duplicates based on this field (e.g., link, guid)", + }, + { + Name: "sort_field", + Label: "Sort By", + Type: "text", + Required: false, + Placeholder: "published_at", + HelpText: "Field to sort merged results by (use published_at for date sorting)", + }, + { + Name: "sort_order", + Label: "Sort Order", + Type: "select", + Required: false, + Options: []nodes.FieldOption{ + {Value: "desc", Label: "Newest First"}, + {Value: "asc", Label: "Oldest First"}, + }, + }, + }, + } +} diff --git a/nodes/transforms/regex.go b/nodes/transforms/regex.go new file mode 100644 index 0000000..44f677b --- /dev/null +++ b/nodes/transforms/regex.go @@ -0,0 +1,108 @@ +package transforms + +import ( + "context" + "fmt" + "regexp" + + "github.com/kierank/pipes/nodes" +) + +type RegexNode struct{} + +func (n *RegexNode) Type() string { return "regex" } +func (n *RegexNode) Label() string { return "Regex Replace" } +func (n *RegexNode) Description() string { return "Search and replace text using regex" } +func (n *RegexNode) Category() string { return "transform" } +func (n *RegexNode) Inputs() int { return 1 } +func (n *RegexNode) Outputs() int { return 1 } + +func (n *RegexNode) Execute(ctx context.Context, config map[string]interface{}, inputs [][]interface{}, execCtx *nodes.Context) ([]interface{}, error) { + if len(inputs) == 0 || len(inputs[0]) == 0 { + return []interface{}{}, nil + } + + items := inputs[0] + field, _ := config["field"].(string) + pattern, _ := config["pattern"].(string) + replacement, _ := config["replacement"].(string) + + if field == "" || pattern == "" { + return items, nil + } + + re, err := regexp.Compile(pattern) + if err != nil { + return nil, fmt.Errorf("invalid regex: %w", err) + } + + var result []interface{} + modified := 0 + + for _, item := range items { + itemMap, ok := item.(map[string]interface{}) + if !ok { + result = append(result, item) + continue + } + + newItem := make(map[string]interface{}) + for k, v := range itemMap { + newItem[k] = v + } + + if val, ok := newItem[field].(string); ok { + newVal := re.ReplaceAllString(val, replacement) + if newVal != val { + modified++ + } + newItem[field] = newVal + } + + result = append(result, newItem) + } + + execCtx.Log("regex", "info", fmt.Sprintf("Modified %d of %d items", modified, len(result))) + return result, nil +} + +func (n *RegexNode) ValidateConfig(config map[string]interface{}) error { + pattern, _ := config["pattern"].(string) + if pattern != "" { + if _, err := regexp.Compile(pattern); err != nil { + return fmt.Errorf("invalid regex pattern: %w", err) + } + } + return nil +} + +func (n *RegexNode) GetConfigSchema() *nodes.ConfigSchema { + return &nodes.ConfigSchema{ + Fields: []nodes.ConfigField{ + { + Name: "field", + Label: "Field", + Type: "text", + Required: true, + Placeholder: "title", + HelpText: "Field to apply regex to", + }, + { + Name: "pattern", + Label: "Pattern", + Type: "text", + Required: true, + Placeholder: "\\[.*?\\]", + HelpText: "Regex pattern to match", + }, + { + Name: "replacement", + Label: "Replacement", + Type: "text", + Required: false, + Placeholder: "", + HelpText: "Text to replace matches with (use $1, $2 for groups)", + }, + }, + } +} diff --git a/nodes/transforms/sort.go b/nodes/transforms/sort.go index 7960c15..24d2386 100644 --- a/nodes/transforms/sort.go +++ b/nodes/transforms/sort.go @@ -31,7 +31,7 @@ func (n *SortNode) Execute(ctx context.Context, config map[string]interface{}, i } if order == "" { - order = "asc" + order = "desc" } // Create a sortable slice @@ -46,8 +46,23 @@ func (n *SortNode) Execute(ctx context.Context, config map[string]interface{}, i return false } - iVal := fmt.Sprintf("%v", getNestedValue(iMap, field)) - jVal := fmt.Sprintf("%v", getNestedValue(jMap, field)) + iRaw := getNestedValue(iMap, field) + jRaw := getNestedValue(jMap, field) + + // Try numeric comparison first (for timestamps, etc.) + iNum, iIsNum := toFloat(iRaw) + jNum, jIsNum := toFloat(jRaw) + + if iIsNum && jIsNum { + if order == "desc" { + return iNum > jNum + } + return iNum < jNum + } + + // Fall back to string comparison + iVal := fmt.Sprintf("%v", iRaw) + jVal := fmt.Sprintf("%v", jRaw) if order == "desc" { return iVal > jVal @@ -72,18 +87,19 @@ func (n *SortNode) GetConfigSchema() *nodes.ConfigSchema { Label: "Field Path", Type: "text", Required: true, - Placeholder: "published", - HelpText: "Field to sort by", + Placeholder: "published_at", + HelpText: "Field to sort by (use published_at or updated_at for date sorting)", }, { Name: "order", Label: "Order", Type: "select", Required: false, - DefaultValue: "asc", + DefaultValue: "desc", + HelpText: "Descending = newest first (for dates), Ascending = oldest first", Options: []nodes.FieldOption{ - {Value: "asc", Label: "Ascending"}, - {Value: "desc", Label: "Descending"}, + {Value: "desc", Label: "Descending (newest first)"}, + {Value: "asc", Label: "Ascending (oldest first)"}, }, }, }, diff --git a/nodes/transforms/truncate.go b/nodes/transforms/truncate.go new file mode 100644 index 0000000..1dd4e1e --- /dev/null +++ b/nodes/transforms/truncate.go @@ -0,0 +1,123 @@ +package transforms + +import ( + "context" + "fmt" + "strings" + + "github.com/kierank/pipes/nodes" +) + +type TruncateNode struct{} + +func (n *TruncateNode) Type() string { return "truncate" } +func (n *TruncateNode) Label() string { return "Truncate" } +func (n *TruncateNode) Description() string { return "Limit text length in a field" } +func (n *TruncateNode) Category() string { return "transform" } +func (n *TruncateNode) Inputs() int { return 1 } +func (n *TruncateNode) Outputs() int { return 1 } + +func (n *TruncateNode) Execute(ctx context.Context, config map[string]interface{}, inputs [][]interface{}, execCtx *nodes.Context) ([]interface{}, error) { + if len(inputs) == 0 || len(inputs[0]) == 0 { + return []interface{}{}, nil + } + + items := inputs[0] + field, _ := config["field"].(string) + maxLength := 200 + if ml, ok := config["max_length"].(float64); ok { + maxLength = int(ml) + } + suffix, _ := config["suffix"].(string) + if suffix == "" { + suffix = "..." + } + + if field == "" { + return items, nil + } + + var result []interface{} + for _, item := range items { + itemMap, ok := item.(map[string]interface{}) + if !ok { + result = append(result, item) + continue + } + + newItem := make(map[string]interface{}) + for k, v := range itemMap { + newItem[k] = v + } + + if val, ok := newItem[field].(string); ok { + // Strip HTML tags first + val = stripHTML(val) + if len(val) > maxLength { + // Find last space before maxLength to avoid cutting words + cutoff := maxLength + if idx := strings.LastIndex(val[:maxLength], " "); idx > maxLength/2 { + cutoff = idx + } + newItem[field] = strings.TrimSpace(val[:cutoff]) + suffix + } else { + newItem[field] = val + } + } + + result = append(result, newItem) + } + + execCtx.Log("truncate", "info", fmt.Sprintf("Truncated %d items", len(result))) + return result, nil +} + +func stripHTML(s string) string { + var result strings.Builder + inTag := false + for _, r := range s { + if r == '<' { + inTag = true + } else if r == '>' { + inTag = false + } else if !inTag { + result.WriteRune(r) + } + } + return strings.TrimSpace(result.String()) +} + +func (n *TruncateNode) ValidateConfig(config map[string]interface{}) error { + return nil +} + +func (n *TruncateNode) GetConfigSchema() *nodes.ConfigSchema { + return &nodes.ConfigSchema{ + Fields: []nodes.ConfigField{ + { + Name: "field", + Label: "Field", + Type: "text", + Required: true, + Placeholder: "description", + HelpText: "Field to truncate", + }, + { + Name: "max_length", + Label: "Max Length", + Type: "number", + Required: false, + DefaultValue: 200, + HelpText: "Maximum character length", + }, + { + Name: "suffix", + Label: "Suffix", + Type: "text", + Required: false, + DefaultValue: "...", + HelpText: "Text to append when truncated", + }, + }, + } +} diff --git a/web/server.go b/web/server.go index f6a034a..db8aa3f 100644 --- a/web/server.go +++ b/web/server.go @@ -14,6 +14,7 @@ import ( "github.com/kierank/pipes/config" "github.com/kierank/pipes/engine" "github.com/kierank/pipes/store" + "github.com/mmcdole/gofeed" ) type Server struct { @@ -68,6 +69,7 @@ func (s *Server) Start() error { mux.HandleFunc("/api/pipes/", s.sessionManager.RequireAuth(s.handleAPIPipe)) mux.HandleFunc("/api/node-types", s.handleAPINodeTypes) mux.HandleFunc("/api/executions/", s.sessionManager.RequireAuth(s.handleAPIExecution)) + mux.HandleFunc("/api/feed-info", s.sessionManager.RequireAuth(s.handleAPIFeedInfo)) // Public feed routes mux.HandleFunc("/feeds/", s.handlePublicFeed) @@ -400,6 +402,29 @@ func (s *Server) handleAPINodeTypes(w http.ResponseWriter, r *http.Request) { json.NewEncoder(w).Encode(nodeTypes) } +func (s *Server) handleAPIFeedInfo(w http.ResponseWriter, r *http.Request) { + url := r.URL.Query().Get("url") + if url == "" { + http.Error(w, "url parameter required", http.StatusBadRequest) + return + } + + fp := gofeed.NewParser() + feed, err := fp.ParseURLWithContext(url, r.Context()) + if err != nil { + http.Error(w, fmt.Sprintf("Failed to parse feed: %v", err), http.StatusBadRequest) + return + } + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]interface{}{ + "title": feed.Title, + "description": feed.Description, + "link": feed.Link, + "item_count": len(feed.Items), + }) +} + func (s *Server) handlePipeExecute(w http.ResponseWriter, r *http.Request, pipeID string, user *store.User) { if r.Method != "POST" { http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) diff --git a/web/templates/editor.html b/web/templates/editor.html index 8d74ba8..cc9629d 100644 --- a/web/templates/editor.html +++ b/web/templates/editor.html @@ -818,8 +818,8 @@ if (!draggedNode) return; const containerRect = document.getElementById('canvas-container').getBoundingClientRect(); - draggedNode.position.x = e.clientX - containerRect.left - dragOffset.x; - draggedNode.position.y = e.clientY - containerRect.top - dragOffset.y; + draggedNode.position.x = e.clientX - containerRect.left - dragOffset.x - panOffset.x; + draggedNode.position.y = e.clientY - containerRect.top - dragOffset.y - panOffset.y; render(); } @@ -867,8 +867,8 @@ function onConnectionMouseMove(e) { const containerRect = document.getElementById('canvas-container').getBoundingClientRect(); - const mouseX = e.clientX - containerRect.left; - const mouseY = e.clientY - containerRect.top; + const mouseX = e.clientX - containerRect.left - panOffset.x; + const mouseY = e.clientY - containerRect.top - panOffset.y; connectionDrag = { x: mouseX, y: mouseY }; @@ -992,8 +992,25 @@ } input.value = node.config[field.name] || field.defaultValue || ''; - input.addEventListener('change', (e) => { + input.addEventListener('change', async (e) => { node.config[field.name] = e.target.value; + + // Auto-fetch feed title for RSS source URL field + if (node.type === 'rss-source' && field.name === 'url' && e.target.value) { + try { + const res = await fetch(`/api/feed-info?url=${encodeURIComponent(e.target.value)}`); + if (res.ok) { + const info = await res.json(); + if (info.title && node.label === 'RSS Feed') { + node.label = info.title; + render(); + showToast(`Feed: ${info.title}`, 'success'); + } + } + } catch (err) { + // Ignore fetch errors + } + } }); group.appendChild(input); @@ -1232,6 +1249,9 @@ } async function executePipe() { + // Save first + await savePipe(); + const res = await fetch(`/api/pipes/${pipeID}/execute`, { method: 'POST' }); @@ -1249,7 +1269,8 @@ // Poll for completion pollExecutionStatus(data.executionId); } else { - showToast('Failed to execute pipe', 'error'); + const errorText = await res.text(); + showToast(errorText || 'Failed to execute pipe', 'error'); } } @@ -1315,16 +1336,26 @@ const toast = document.createElement('div'); toast.className = `toast ${type}`; toast.textContent = message; + + // Click to dismiss + toast.style.cursor = 'pointer'; + toast.onclick = () => { + toast.classList.add('exit'); + setTimeout(() => container.removeChild(toast), 200); + }; container.appendChild(toast); - // Auto-dismiss after 3 seconds + // Auto-dismiss: errors stay longer + const duration = type === 'error' ? 8000 : 3000; setTimeout(() => { - toast.classList.add('exit'); - setTimeout(() => { - container.removeChild(toast); - }, 200); // Match animation duration - }, 3000); + if (toast.parentNode) { + toast.classList.add('exit'); + setTimeout(() => { + if (toast.parentNode) container.removeChild(toast); + }, 200); + } + }, duration); } -- 2.51.2