From 9d2f418c1cdfa7ed8eb8a80f9887f0587ee0acac Mon Sep 17 00:00:00 2001 From: Michael McNees Date: Sat, 30 May 2026 19:54:19 -0400 Subject: [PATCH 1/2] feat(connector): Discord, Elasticsearch, Datadog, Redis, and MongoDB connectors (#95, #98, #100, #104, #110) Adds 10 new connector actions across 5 services: - discord/send, discord/embed - elasticsearch/search, elasticsearch/index - datadog/submit_event, datadog/query_metrics - redis/get, redis/set, redis/publish - mongodb/find, mongodb/aggregate Also consolidates duplicate bearer-token extraction into a shared extractBearerToken helper in tools.go (removing local copies from airtable.go, asana.go, notion.go). Co-Authored-By: Claude Sonnet 4.6 --- go.work.sum | 6 + packages/engine/go.mod | 3 + packages/engine/go.sum | 6 + .../engine/internal/connector/airtable.go | 25 +- packages/engine/internal/connector/asana.go | 25 +- .../engine/internal/connector/connector.go | 11 + packages/engine/internal/connector/datadog.go | 203 ++++++++++++++++ .../engine/internal/connector/datadog_test.go | 230 ++++++++++++++++++ packages/engine/internal/connector/discord.go | 148 +++++++++++ .../engine/internal/connector/discord_test.go | 205 ++++++++++++++++ .../internal/connector/elasticsearch.go | 215 ++++++++++++++++ .../internal/connector/elasticsearch_test.go | 220 +++++++++++++++++ packages/engine/internal/connector/mongodb.go | 165 +++++++++++++ .../engine/internal/connector/mongodb_test.go | 122 ++++++++++ packages/engine/internal/connector/notion.go | 29 +-- packages/engine/internal/connector/redis.go | 119 +++++++++ .../engine/internal/connector/redis_test.go | 126 ++++++++++ packages/engine/internal/connector/tools.go | 25 ++ 18 files changed, 1810 insertions(+), 73 deletions(-) create mode 100644 packages/engine/internal/connector/datadog.go create mode 100644 packages/engine/internal/connector/datadog_test.go create mode 100644 packages/engine/internal/connector/discord.go create mode 100644 packages/engine/internal/connector/discord_test.go create mode 100644 packages/engine/internal/connector/elasticsearch.go create mode 100644 packages/engine/internal/connector/elasticsearch_test.go create mode 100644 packages/engine/internal/connector/mongodb.go create mode 100644 packages/engine/internal/connector/mongodb_test.go create mode 100644 packages/engine/internal/connector/redis.go create mode 100644 packages/engine/internal/connector/redis_test.go diff --git a/go.work.sum b/go.work.sum index 3739cc08..8bfbfec6 100644 --- a/go.work.sum +++ b/go.work.sum @@ -329,12 +329,18 @@ github.com/tursodatabase/libsql-client-go v0.0.0-20251219100830-236aa1ff8acc/go. github.com/urfave/cli/v2 v2.3.0/go.mod h1:LJmUH05zAU44vOAcrfzZQKsZbVcdbOG8rtL3/XcUArI= github.com/vertica/vertica-sql-go v1.3.5 h1:IrfH2WIgzZ45yDHyjVFrXU2LuKNIjF5Nwi90a6cfgUI= github.com/vertica/vertica-sql-go v1.3.5/go.mod h1:jnn2GFuv+O2Jcjktb7zyc4Utlbu9YVqpHH/lx63+1M4= +github.com/xdg-go/scram v1.2.0 h1:bYKF2AEwG5rqd1BumT4gAnvwU/M9nBp2pTSxeZw7Wvs= +github.com/xdg-go/scram v1.2.0/go.mod h1:3dlrS0iBaWKYVt2ZfA4cj48umJZ+cAEbR6/SjLA88I8= +github.com/xdg-go/stringprep v1.0.4 h1:XLI/Ng3O1Atzq0oBs3TWm+5ZVgkq2aqdlvP9JtoZ6c8= +github.com/xdg-go/stringprep v1.0.4/go.mod h1:mPGuuIYwz7CmR2bT9j4GbQqutWS1zV24gijq1dTyGkM= github.com/xhit/go-str2duration/v2 v2.1.0 h1:lxklc02Drh6ynqX+DdPyp5pCKLUQpRT8bp8Ydu2Bstc= github.com/xhit/go-str2duration/v2 v2.1.0/go.mod h1:ohY8p+0f07DiV6Em5LKB0s2YpLtXVyJfNt1+BlmyAsU= github.com/ydb-platform/ydb-go-genproto v0.0.0-20260128080146-c4ed16b24b37 h1:kUXMT/fM/DpDT66WQgRUf3I8VOAWjypkMf52W5PChwA= github.com/ydb-platform/ydb-go-genproto v0.0.0-20260128080146-c4ed16b24b37/go.mod h1:Er+FePu1dNUieD+XTMDduGpQuCPssK5Q4BjF+IIXJ3I= github.com/ydb-platform/ydb-go-sdk/v3 v3.127.0 h1:OfHS9ZkZgCy6y/CJ9N8123DXrgaY2BPxWsQiQ8e3wC8= github.com/ydb-platform/ydb-go-sdk/v3 v3.127.0/go.mod h1:stS1mQYjbJvwwYaYzKyFY9eMiuVXWWXQA6T+SpOLg9c= +github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78 h1:ilQV1hzziu+LLM3zUTJ0trRztfwgjqKnBWNtSRkbmwM= +github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78/go.mod h1:aL8wCCfTfSfmXjznFBSZNN13rSJjlIOI1fUNAtF7rmI= github.com/yuin/goldmark v1.4.13 h1:fVcFKWvrslecOb/tg+Cc05dkeYx540o0FuFt3nUVDoE= github.com/ziutek/mymysql v1.5.4 h1:GB0qdRGsTwQSBVYuVShFBKaXSnSnYYC2d9knnE1LHFs= github.com/ziutek/mymysql v1.5.4/go.mod h1:LMSpPZ6DbqWFxNCHW77HeMg9I646SAhApZ/wKdgO/C0= diff --git a/packages/engine/go.mod b/packages/engine/go.mod index 02a13c9d..2daed272 100644 --- a/packages/engine/go.mod +++ b/packages/engine/go.mod @@ -126,6 +126,7 @@ require ( github.com/prometheus/client_model v0.6.2 // indirect github.com/prometheus/common v0.66.1 // indirect github.com/prometheus/procfs v0.19.2 // indirect + github.com/redis/go-redis/v9 v9.20.0 // indirect github.com/sagikazarmark/locafero v0.11.0 // indirect github.com/sergi/go-diff v1.3.2-0.20230802210424-5b0b94c5c0d3 // indirect github.com/sethvargo/go-retry v0.3.0 // indirect @@ -141,12 +142,14 @@ require ( github.com/tklauser/numcpus v0.11.0 // indirect github.com/xanzy/ssh-agent v0.3.3 // indirect github.com/yusufpapurcu/wmi v1.2.4 // indirect + go.mongodb.org/mongo-driver/v2 v2.6.0 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.61.0 // indirect go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.65.0 // indirect go.opentelemetry.io/otel v1.41.0 // indirect go.opentelemetry.io/otel/metric v1.41.0 // indirect go.opentelemetry.io/otel/trace v1.41.0 // indirect + go.uber.org/atomic v1.11.0 // indirect go.uber.org/multierr v1.11.0 // indirect go.yaml.in/yaml/v2 v2.4.2 // indirect go.yaml.in/yaml/v3 v3.0.4 // indirect diff --git a/packages/engine/go.sum b/packages/engine/go.sum index 32d6d6e3..ce669e96 100644 --- a/packages/engine/go.sum +++ b/packages/engine/go.sum @@ -285,6 +285,8 @@ github.com/prometheus/common v0.66.1 h1:h5E0h5/Y8niHc5DlaLlWLArTQI7tMrsfQjHV+d9Z github.com/prometheus/common v0.66.1/go.mod h1:gcaUsgf3KfRSwHY4dIMXLPV0K/Wg1oZ8+SbZk/HH/dA= github.com/prometheus/procfs v0.19.2 h1:zUMhqEW66Ex7OXIiDkll3tl9a1ZdilUOd/F6ZXw4Vws= github.com/prometheus/procfs v0.19.2/go.mod h1:M0aotyiemPhBCM0z5w87kL22CxfcH05ZpYlu+b4J7mw= +github.com/redis/go-redis/v9 v9.20.0 h1:WnQYxLkgO2xiXTCJY0ldIiI8dNqCDlQAG+AtaH7a2a0= +github.com/redis/go-redis/v9 v9.20.0/go.mod h1:v/M13XI1PVCDcm01VtPFOADfZtHf8YW3baQf57KlIkA= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= @@ -340,6 +342,8 @@ github.com/xanzy/ssh-agent v0.3.3/go.mod h1:6dzNDKs0J9rVPHPhaGCukekBHKqfl+L3KghI github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= github.com/yusufpapurcu/wmi v1.2.4 h1:zFUKzehAFReQwLys1b/iSMl+JQGSCSjtVqQn9bBrPo0= github.com/yusufpapurcu/wmi v1.2.4/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0= +go.mongodb.org/mongo-driver/v2 v2.6.0 h1:b9sJOYrkmt4l8bY43ZenFBcPlhYIjaOfYHLtbB/5qi8= +go.mongodb.org/mongo-driver/v2 v2.6.0/go.mod h1:yOI9kBsufol30iFsl1slpdq1I0eHPzybRWdyYUs8K/0= go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.61.0 h1:q4XOmH/0opmeuJtPsbFNivyl7bCt7yRBbeEm2sC/XtQ= @@ -362,6 +366,8 @@ go.opentelemetry.io/otel/trace v1.41.0 h1:Vbk2co6bhj8L59ZJ6/xFTskY+tGAbOnCtQGVVa go.opentelemetry.io/otel/trace v1.41.0/go.mod h1:U1NU4ULCoxeDKc09yCWdWe+3QoyweJcISEVa1RBzOis= go.opentelemetry.io/proto/otlp v1.0.0 h1:T0TX0tmXU8a3CbNXzEKGeU5mIVOdf0oykP+u2lIVU/I= go.opentelemetry.io/proto/otlp v1.0.0/go.mod h1:Sy6pihPLfYHkr3NkUbEhGHFhINUSI/v80hjKIs5JXpM= +go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= +go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0= diff --git a/packages/engine/internal/connector/airtable.go b/packages/engine/internal/connector/airtable.go index d79c83ba..58011698 100644 --- a/packages/engine/internal/connector/airtable.go +++ b/packages/engine/internal/connector/airtable.go @@ -27,7 +27,7 @@ func (c *AirtableListConnector) apiURL(path string) string { } func (c *AirtableListConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { - token, err := extractAirtableToken(params) + token, err := extractBearerToken(params) if err != nil { return nil, fmt.Errorf("airtable/list: %w", err) } @@ -115,7 +115,7 @@ func (c *AirtableCreateRecordConnector) apiURL(path string) string { } func (c *AirtableCreateRecordConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { - token, err := extractAirtableToken(params) + token, err := extractBearerToken(params) if err != nil { return nil, fmt.Errorf("airtable/create_record: %w", err) } @@ -179,24 +179,3 @@ func (c *AirtableCreateRecordConnector) Execute(ctx context.Context, params map[ }, nil } -func extractAirtableToken(params map[string]any) (string, error) { - raw, ok := params["_credential"] - if !ok || raw == nil { - return "", fmt.Errorf("credential is required") - } - delete(params, "_credential") - - var token string - switch cred := raw.(type) { - case map[string]string: - token = cred["token"] - case map[string]any: - token, _ = cred["token"].(string) - default: - return "", fmt.Errorf("credential is required") - } - if token == "" { - return "", fmt.Errorf("credential must contain a 'token' field") - } - return token, nil -} diff --git a/packages/engine/internal/connector/asana.go b/packages/engine/internal/connector/asana.go index 2c69b7e3..634f7f9f 100644 --- a/packages/engine/internal/connector/asana.go +++ b/packages/engine/internal/connector/asana.go @@ -27,7 +27,7 @@ func (c *AsanaCreateTaskConnector) apiURL(path string) string { } func (c *AsanaCreateTaskConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { - token, err := extractAsanaToken(params) + token, err := extractBearerToken(params) if err != nil { return nil, fmt.Errorf("asana/create_task: %w", err) } @@ -120,7 +120,7 @@ func (c *AsanaSearchConnector) apiURL(path string) string { } func (c *AsanaSearchConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { - token, err := extractAsanaToken(params) + token, err := extractBearerToken(params) if err != nil { return nil, fmt.Errorf("asana/search: %w", err) } @@ -188,24 +188,3 @@ func (c *AsanaSearchConnector) Execute(ctx context.Context, params map[string]an }, nil } -func extractAsanaToken(params map[string]any) (string, error) { - raw, ok := params["_credential"] - if !ok || raw == nil { - return "", fmt.Errorf("credential is required") - } - delete(params, "_credential") - - var token string - switch cred := raw.(type) { - case map[string]string: - token = cred["token"] - case map[string]any: - token, _ = cred["token"].(string) - default: - return "", fmt.Errorf("credential is required") - } - if token == "" { - return "", fmt.Errorf("credential must contain a 'token' field") - } - return token, nil -} diff --git a/packages/engine/internal/connector/connector.go b/packages/engine/internal/connector/connector.go index 196632a7..c7199f0a 100644 --- a/packages/engine/internal/connector/connector.go +++ b/packages/engine/internal/connector/connector.go @@ -54,6 +54,17 @@ func NewRegistry() *Registry { r.Register("twilio/call", &TwilioCallConnector{}) r.Register("asana/create_task", &AsanaCreateTaskConnector{}) r.Register("asana/search", &AsanaSearchConnector{}) + r.Register("discord/send", &DiscordSendConnector{}) + r.Register("discord/embed", &DiscordEmbedConnector{}) + r.Register("elasticsearch/search", &ElasticsearchSearchConnector{}) + r.Register("elasticsearch/index", &ElasticsearchIndexConnector{}) + r.Register("datadog/submit_event", &DatadogSubmitEventConnector{}) + r.Register("datadog/query_metrics", &DatadogQueryMetricsConnector{}) + r.Register("redis/get", &RedisGetConnector{}) + r.Register("redis/set", &RedisSetConnector{}) + r.Register("redis/publish", &RedisPublishConnector{}) + r.Register("mongodb/find", &MongoFindConnector{}) + r.Register("mongodb/aggregate", &MongoAggregateConnector{}) return r } diff --git a/packages/engine/internal/connector/datadog.go b/packages/engine/internal/connector/datadog.go new file mode 100644 index 00000000..3fef8657 --- /dev/null +++ b/packages/engine/internal/connector/datadog.go @@ -0,0 +1,203 @@ +package connector + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" +) + +const datadogBaseURL = "https://api.datadoghq.com" + +// DatadogSubmitEventConnector submits an event to the Datadog event stream. +type DatadogSubmitEventConnector struct { + Client *http.Client + baseURL string // override for testing +} + +func (c *DatadogSubmitEventConnector) apiURL(path string) string { + base := c.baseURL + if base == "" { + base = datadogBaseURL + } + return base + path +} + +func (c *DatadogSubmitEventConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { + apiKey, appKey, err := extractDatadogCredential(params) + if err != nil { + return nil, fmt.Errorf("datadog/submit_event: %w", err) + } + + title, _ := params["title"].(string) + if title == "" { + return nil, fmt.Errorf("datadog/submit_event: title is required") + } + text, _ := params["text"].(string) + if text == "" { + return nil, fmt.Errorf("datadog/submit_event: text is required") + } + + event := map[string]any{ + "title": title, + "text": text, + } + if alertType, ok := params["alert_type"].(string); ok && alertType != "" { + event["alert_type"] = alertType + } + if priority, ok := params["priority"].(string); ok && priority != "" { + event["priority"] = priority + } + if tags, ok := params["tags"].([]any); ok && len(tags) > 0 { + event["tags"] = tags + } + if host, ok := params["host"].(string); ok && host != "" { + event["host"] = host + } + + reqJSON, err := json.Marshal(event) + if err != nil { + return nil, fmt.Errorf("datadog/submit_event: marshaling request: %w", err) + } + + req, err := http.NewRequestWithContext(ctx, "POST", c.apiURL("/api/v1/events"), bytes.NewReader(reqJSON)) + if err != nil { + return nil, fmt.Errorf("datadog/submit_event: creating request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("DD-API-KEY", apiKey) + if appKey != "" { + req.Header.Set("DD-APPLICATION-KEY", appKey) + } + + resp, err := httpClient(c.Client).Do(req) + if err != nil { + return nil, fmt.Errorf("datadog/submit_event: %w", err) + } + defer resp.Body.Close() + + respBody, err := io.ReadAll(io.LimitReader(resp.Body, DefaultMaxResponseBytes)) + if err != nil { + return nil, fmt.Errorf("datadog/submit_event: reading response: %w", err) + } + + if resp.StatusCode != http.StatusAccepted { + return nil, fmt.Errorf("datadog/submit_event: Datadog API returned %d: %s", resp.StatusCode, truncate(string(respBody), 500)) + } + + var result struct { + Status string `json:"status"` + Event struct { + ID int64 `json:"id"` + } `json:"event"` + } + if err := json.Unmarshal(respBody, &result); err != nil { + return nil, fmt.Errorf("datadog/submit_event: parsing response: %w", err) + } + + return map[string]any{ + "status": result.Status, + "event_id": result.Event.ID, + }, nil +} + +// DatadogQueryMetricsConnector queries time-series metrics from Datadog. +type DatadogQueryMetricsConnector struct { + Client *http.Client + baseURL string // override for testing +} + +func (c *DatadogQueryMetricsConnector) apiURL(path string) string { + base := c.baseURL + if base == "" { + base = datadogBaseURL + } + return base + path +} + +func (c *DatadogQueryMetricsConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { + apiKey, appKey, err := extractDatadogCredential(params) + if err != nil { + return nil, fmt.Errorf("datadog/query_metrics: %w", err) + } + + query, _ := params["query"].(string) + if query == "" { + return nil, fmt.Errorf("datadog/query_metrics: query is required") + } + from, fromOK := extractInt(params["from"]) + to, toOK := extractInt(params["to"]) + if !fromOK || !toOK { + return nil, fmt.Errorf("datadog/query_metrics: from and to (Unix timestamps) are required") + } + + q := url.Values{} + q.Set("query", query) + q.Set("from", fmt.Sprintf("%d", from)) + q.Set("to", fmt.Sprintf("%d", to)) + + req, err := http.NewRequestWithContext(ctx, "GET", c.apiURL("/api/v1/query?"+q.Encode()), nil) + if err != nil { + return nil, fmt.Errorf("datadog/query_metrics: creating request: %w", err) + } + req.Header.Set("DD-API-KEY", apiKey) + if appKey != "" { + req.Header.Set("DD-APPLICATION-KEY", appKey) + } + + resp, err := httpClient(c.Client).Do(req) + if err != nil { + return nil, fmt.Errorf("datadog/query_metrics: %w", err) + } + defer resp.Body.Close() + + respBody, err := io.ReadAll(io.LimitReader(resp.Body, DefaultMaxResponseBytes)) + if err != nil { + return nil, fmt.Errorf("datadog/query_metrics: reading response: %w", err) + } + + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("datadog/query_metrics: Datadog API returned %d: %s", resp.StatusCode, truncate(string(respBody), 500)) + } + + var result struct { + Status string `json:"status"` + Series []any `json:"series"` + } + if err := json.Unmarshal(respBody, &result); err != nil { + return nil, fmt.Errorf("datadog/query_metrics: parsing response: %w", err) + } + + return map[string]any{ + "status": result.Status, + "series": result.Series, + "count": len(result.Series), + }, nil +} + +// extractDatadogCredential pulls api_key and optional app_key from _credential. +func extractDatadogCredential(params map[string]any) (apiKey, appKey string, err error) { + raw, ok := params["_credential"] + if !ok || raw == nil { + return "", "", fmt.Errorf("credential is required") + } + delete(params, "_credential") + + switch cred := raw.(type) { + case map[string]string: + apiKey = cred["api_key"] + appKey = cred["app_key"] + case map[string]any: + apiKey, _ = cred["api_key"].(string) + appKey, _ = cred["app_key"].(string) + default: + return "", "", fmt.Errorf("credential is required") + } + if apiKey == "" { + return "", "", fmt.Errorf("credential must contain an 'api_key' field") + } + return apiKey, appKey, nil +} diff --git a/packages/engine/internal/connector/datadog_test.go b/packages/engine/internal/connector/datadog_test.go new file mode 100644 index 00000000..42defd38 --- /dev/null +++ b/packages/engine/internal/connector/datadog_test.go @@ -0,0 +1,230 @@ +package connector + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" +) + +func TestDatadogSubmitEventConnector_SubmitsEvent(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != "POST" || r.URL.Path != "/api/v1/events" { + t.Errorf("unexpected %s %s", r.Method, r.URL.Path) + } + if r.Header.Get("DD-API-KEY") != "dd-api-key" { + t.Errorf("unexpected DD-API-KEY: %s", r.Header.Get("DD-API-KEY")) + } + if r.Header.Get("DD-APPLICATION-KEY") != "dd-app-key" { + t.Errorf("unexpected DD-APPLICATION-KEY: %s", r.Header.Get("DD-APPLICATION-KEY")) + } + var body map[string]any + json.NewDecoder(r.Body).Decode(&body) + if body["title"] != "Deploy completed" { + t.Errorf("unexpected title: %v", body["title"]) + } + if body["text"] != "v1.2.3 deployed" { + t.Errorf("unexpected text: %v", body["text"]) + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusAccepted) + json.NewEncoder(w).Encode(map[string]any{ + "status": "ok", + "event": map[string]any{"id": int64(12345)}, + }) + })) + defer srv.Close() + + c := &DatadogSubmitEventConnector{baseURL: srv.URL} + out, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"api_key": "dd-api-key", "app_key": "dd-app-key"}, + "title": "Deploy completed", + "text": "v1.2.3 deployed", + }) + if err != nil { + t.Fatal(err) + } + if out["status"] != "ok" { + t.Errorf("expected status=ok, got %v", out["status"]) + } +} + +func TestDatadogSubmitEventConnector_WithOptionalFields(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var body map[string]any + json.NewDecoder(r.Body).Decode(&body) + if body["alert_type"] != "error" { + t.Errorf("unexpected alert_type: %v", body["alert_type"]) + } + if body["priority"] != "high" { + t.Errorf("unexpected priority: %v", body["priority"]) + } + tags, _ := body["tags"].([]any) + if len(tags) != 2 { + t.Errorf("expected 2 tags, got %d", len(tags)) + } + w.WriteHeader(http.StatusAccepted) + json.NewEncoder(w).Encode(map[string]any{"status": "ok", "event": map[string]any{"id": int64(1)}}) + })) + defer srv.Close() + + c := &DatadogSubmitEventConnector{baseURL: srv.URL} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"api_key": "key"}, + "title": "t", + "text": "txt", + "alert_type": "error", + "priority": "high", + "tags": []any{"env:prod", "service:api"}, + }) + if err != nil { + t.Fatal(err) + } +} + +func TestDatadogSubmitEventConnector_MissingTitle(t *testing.T) { + c := &DatadogSubmitEventConnector{baseURL: "http://unused"} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"api_key": "key"}, + "text": "txt", + }) + if err == nil { + t.Fatal("expected error for missing title") + } +} + +func TestDatadogSubmitEventConnector_MissingText(t *testing.T) { + c := &DatadogSubmitEventConnector{baseURL: "http://unused"} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"api_key": "key"}, + "title": "t", + }) + if err == nil { + t.Fatal("expected error for missing text") + } +} + +func TestDatadogSubmitEventConnector_MissingCredential(t *testing.T) { + c := &DatadogSubmitEventConnector{baseURL: "http://unused"} + _, err := c.Execute(t.Context(), map[string]any{ + "title": "t", + "text": "txt", + }) + if err == nil { + t.Fatal("expected error for missing credential") + } +} + +func TestDatadogSubmitEventConnector_APIError(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, `{"errors":["API key is invalid"]}`, http.StatusForbidden) + })) + defer srv.Close() + + c := &DatadogSubmitEventConnector{baseURL: srv.URL} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"api_key": "bad"}, + "title": "t", + "text": "txt", + }) + if err == nil { + t.Fatal("expected error for 403") + } +} + +func TestDatadogQueryMetricsConnector_QueriesMetrics(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != "GET" { + t.Errorf("expected GET, got %s", r.Method) + } + q := r.URL.Query() + if q.Get("query") != "avg:system.cpu.user{*}" { + t.Errorf("unexpected query: %s", q.Get("query")) + } + if q.Get("from") != "1700000000" { + t.Errorf("unexpected from: %s", q.Get("from")) + } + if q.Get("to") != "1700003600" { + t.Errorf("unexpected to: %s", q.Get("to")) + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{ + "status": "ok", + "series": []any{ + map[string]any{"metric": "system.cpu.user", "pointlist": []any{}}, + }, + }) + })) + defer srv.Close() + + c := &DatadogQueryMetricsConnector{baseURL: srv.URL} + out, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"api_key": "key"}, + "query": "avg:system.cpu.user{*}", + "from": 1700000000, + "to": 1700003600, + }) + if err != nil { + t.Fatal(err) + } + if out["status"] != "ok" { + t.Errorf("expected status=ok, got %v", out["status"]) + } + if out["count"].(int) != 1 { + t.Errorf("expected count=1, got %v", out["count"]) + } +} + +func TestDatadogQueryMetricsConnector_MissingQuery(t *testing.T) { + c := &DatadogQueryMetricsConnector{baseURL: "http://unused"} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"api_key": "key"}, + "from": 1700000000, + "to": 1700003600, + }) + if err == nil { + t.Fatal("expected error for missing query") + } +} + +func TestDatadogQueryMetricsConnector_MissingFromTo(t *testing.T) { + c := &DatadogQueryMetricsConnector{baseURL: "http://unused"} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"api_key": "key"}, + "query": "avg:system.cpu.user{*}", + }) + if err == nil { + t.Fatal("expected error for missing from/to") + } +} + +func TestDatadogSubmitEventConnector_MapAnyCredential(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("DD-API-KEY") != "mapkey" { + t.Errorf("unexpected DD-API-KEY: %s", r.Header.Get("DD-API-KEY")) + } + w.WriteHeader(http.StatusAccepted) + json.NewEncoder(w).Encode(map[string]any{"status": "ok", "event": map[string]any{"id": int64(1)}}) + })) + defer srv.Close() + + c := &DatadogSubmitEventConnector{baseURL: srv.URL} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]any{"api_key": "mapkey"}, + "title": "t", + "text": "txt", + }) + if err != nil { + t.Fatal(err) + } +} + +func TestRegistry_DatadogConnectors(t *testing.T) { + r := NewRegistry() + if _, err := r.Get("datadog/submit_event"); err != nil { + t.Errorf("datadog/submit_event not registered: %v", err) + } + if _, err := r.Get("datadog/query_metrics"); err != nil { + t.Errorf("datadog/query_metrics not registered: %v", err) + } +} diff --git a/packages/engine/internal/connector/discord.go b/packages/engine/internal/connector/discord.go new file mode 100644 index 00000000..ab47fab6 --- /dev/null +++ b/packages/engine/internal/connector/discord.go @@ -0,0 +1,148 @@ +package connector + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" +) + +const discordBaseURL = "https://discord.com/api/v10" + +// DiscordSendConnector sends a text message to a Discord channel. +type DiscordSendConnector struct { + Client *http.Client + baseURL string // override for testing +} + +func (c *DiscordSendConnector) apiURL(path string) string { + base := c.baseURL + if base == "" { + base = discordBaseURL + } + return base + path +} + +func (c *DiscordSendConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { + token, err := extractBearerToken(params) + if err != nil { + return nil, fmt.Errorf("discord/send: %w", err) + } + + channelID, _ := params["channel_id"].(string) + if channelID == "" { + return nil, fmt.Errorf("discord/send: channel_id is required") + } + content, _ := params["content"].(string) + if content == "" { + return nil, fmt.Errorf("discord/send: content is required") + } + + body := map[string]any{"content": content} + if tts, ok := params["tts"].(bool); ok { + body["tts"] = tts + } + + return discordPostMessage(ctx, c.apiURL(fmt.Sprintf("/channels/%s/messages", url.PathEscape(channelID))), token, body, "discord/send") +} + +// DiscordEmbedConnector sends an embed message to a Discord channel. +type DiscordEmbedConnector struct { + Client *http.Client + baseURL string // override for testing +} + +func (c *DiscordEmbedConnector) apiURL(path string) string { + base := c.baseURL + if base == "" { + base = discordBaseURL + } + return base + path +} + +func (c *DiscordEmbedConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { + token, err := extractBearerToken(params) + if err != nil { + return nil, fmt.Errorf("discord/embed: %w", err) + } + + channelID, _ := params["channel_id"].(string) + if channelID == "" { + return nil, fmt.Errorf("discord/embed: channel_id is required") + } + + embed := map[string]any{} + if title, ok := params["title"].(string); ok && title != "" { + embed["title"] = title + } + if description, ok := params["description"].(string); ok && description != "" { + embed["description"] = description + } + if color, ok := extractInt(params["color"]); ok { + embed["color"] = color + } + if url_, ok := params["url"].(string); ok && url_ != "" { + embed["url"] = url_ + } + if fields, ok := params["fields"].([]any); ok && len(fields) > 0 { + embed["fields"] = fields + } + if footer, ok := params["footer"].(map[string]any); ok { + embed["footer"] = footer + } + + body := map[string]any{"embeds": []any{embed}} + if content, ok := params["content"].(string); ok && content != "" { + body["content"] = content + } + + return discordPostMessage(ctx, c.apiURL(fmt.Sprintf("/channels/%s/messages", url.PathEscape(channelID))), token, body, "discord/embed") +} + +func discordPostMessage(ctx context.Context, endpoint, token string, body map[string]any, action string) (map[string]any, error) { + reqJSON, err := json.Marshal(body) + if err != nil { + return nil, fmt.Errorf("%s: marshaling request: %w", action, err) + } + + req, err := http.NewRequestWithContext(ctx, "POST", endpoint, bytes.NewReader(reqJSON)) + if err != nil { + return nil, fmt.Errorf("%s: creating request: %w", action, err) + } + req.Header.Set("Authorization", "Bot "+token) + req.Header.Set("Content-Type", "application/json") + + resp, err := httpClient(nil).Do(req) + if err != nil { + return nil, fmt.Errorf("%s: %w", action, err) + } + defer resp.Body.Close() + + respBody, err := io.ReadAll(io.LimitReader(resp.Body, DefaultMaxResponseBytes)) + if err != nil { + return nil, fmt.Errorf("%s: reading response: %w", action, err) + } + + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("%s: Discord API returned %d: %s", action, resp.StatusCode, truncate(string(respBody), 500)) + } + + var msg struct { + ID string `json:"id"` + ChannelID string `json:"channel_id"` + Timestamp string `json:"timestamp"` + } + if err := json.Unmarshal(respBody, &msg); err != nil { + return nil, fmt.Errorf("%s: parsing response: %w", action, err) + } + + return map[string]any{ + "id": msg.ID, + "channel_id": msg.ChannelID, + "timestamp": msg.Timestamp, + }, nil +} + diff --git a/packages/engine/internal/connector/discord_test.go b/packages/engine/internal/connector/discord_test.go new file mode 100644 index 00000000..ec369981 --- /dev/null +++ b/packages/engine/internal/connector/discord_test.go @@ -0,0 +1,205 @@ +package connector + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" +) + +func TestDiscordSendConnector_SendsMessage(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != "POST" { + t.Errorf("expected POST, got %s", r.Method) + } + if r.Header.Get("Authorization") != "Bot bot-token" { + t.Errorf("unexpected Authorization: %s", r.Header.Get("Authorization")) + } + var body map[string]any + json.NewDecoder(r.Body).Decode(&body) + if body["content"] != "Hello, world!" { + t.Errorf("unexpected content: %v", body["content"]) + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{ + "id": "1234567890", + "channel_id": "9876543210", + "timestamp": "2024-01-01T00:00:00.000Z", + }) + })) + defer srv.Close() + + c := &DiscordSendConnector{baseURL: srv.URL} + out, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"token": "bot-token"}, + "channel_id": "9876543210", + "content": "Hello, world!", + }) + if err != nil { + t.Fatal(err) + } + if out["id"] != "1234567890" { + t.Errorf("expected id=1234567890, got %v", out["id"]) + } +} + +func TestDiscordSendConnector_MissingContent(t *testing.T) { + c := &DiscordSendConnector{baseURL: "http://unused"} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"token": "bot-token"}, + "channel_id": "9876543210", + }) + if err == nil { + t.Fatal("expected error for missing content") + } +} + +func TestDiscordSendConnector_MissingChannelID(t *testing.T) { + c := &DiscordSendConnector{baseURL: "http://unused"} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"token": "bot-token"}, + "content": "Hello", + }) + if err == nil { + t.Fatal("expected error for missing channel_id") + } +} + +func TestDiscordSendConnector_MissingCredential(t *testing.T) { + c := &DiscordSendConnector{baseURL: "http://unused"} + _, err := c.Execute(t.Context(), map[string]any{ + "channel_id": "9876543210", + "content": "Hello", + }) + if err == nil { + t.Fatal("expected error for missing credential") + } +} + +func TestDiscordSendConnector_APIError(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, `{"code":50013,"message":"Missing Permissions"}`, http.StatusForbidden) + })) + defer srv.Close() + + c := &DiscordSendConnector{baseURL: srv.URL} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"token": "bot-token"}, + "channel_id": "9876543210", + "content": "Hello", + }) + if err == nil { + t.Fatal("expected error for 403") + } +} + +func TestDiscordEmbedConnector_SendsEmbed(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var body map[string]any + json.NewDecoder(r.Body).Decode(&body) + embeds, _ := body["embeds"].([]any) + if len(embeds) != 1 { + t.Errorf("expected 1 embed, got %d", len(embeds)) + } + embed, _ := embeds[0].(map[string]any) + if embed["title"] != "Alert" { + t.Errorf("unexpected title: %v", embed["title"]) + } + if embed["description"] != "Something happened" { + t.Errorf("unexpected description: %v", embed["description"]) + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{ + "id": "111", + "channel_id": "222", + "timestamp": "2024-01-01T00:00:00.000Z", + }) + })) + defer srv.Close() + + c := &DiscordEmbedConnector{baseURL: srv.URL} + out, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"token": "bot-token"}, + "channel_id": "222", + "title": "Alert", + "description": "Something happened", + "color": 16711680, + }) + if err != nil { + t.Fatal(err) + } + if out["id"] != "111" { + t.Errorf("expected id=111, got %v", out["id"]) + } +} + +func TestDiscordEmbedConnector_WithFields(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var body map[string]any + json.NewDecoder(r.Body).Decode(&body) + embeds := body["embeds"].([]any) + embed := embeds[0].(map[string]any) + fields, _ := embed["fields"].([]any) + if len(fields) != 2 { + t.Errorf("expected 2 fields, got %d", len(fields)) + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{"id": "1", "channel_id": "2", "timestamp": ""}) + })) + defer srv.Close() + + c := &DiscordEmbedConnector{baseURL: srv.URL} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"token": "bot-token"}, + "channel_id": "2", + "fields": []any{ + map[string]any{"name": "Status", "value": "OK"}, + map[string]any{"name": "Region", "value": "us-east-1"}, + }, + }) + if err != nil { + t.Fatal(err) + } +} + +func TestDiscordEmbedConnector_MissingChannelID(t *testing.T) { + c := &DiscordEmbedConnector{baseURL: "http://unused"} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"token": "bot-token"}, + "title": "Test", + }) + if err == nil { + t.Fatal("expected error for missing channel_id") + } +} + +func TestDiscordSendConnector_MapAnyCredential(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Authorization") != "Bot mapany" { + t.Errorf("unexpected auth: %s", r.Header.Get("Authorization")) + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{"id": "1", "channel_id": "2", "timestamp": ""}) + })) + defer srv.Close() + + c := &DiscordSendConnector{baseURL: srv.URL} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]any{"token": "mapany"}, + "channel_id": "2", + "content": "hi", + }) + if err != nil { + t.Fatal(err) + } +} + +func TestRegistry_DiscordConnectors(t *testing.T) { + r := NewRegistry() + if _, err := r.Get("discord/send"); err != nil { + t.Errorf("discord/send not registered: %v", err) + } + if _, err := r.Get("discord/embed"); err != nil { + t.Errorf("discord/embed not registered: %v", err) + } +} diff --git a/packages/engine/internal/connector/elasticsearch.go b/packages/engine/internal/connector/elasticsearch.go new file mode 100644 index 00000000..cdad994a --- /dev/null +++ b/packages/engine/internal/connector/elasticsearch.go @@ -0,0 +1,215 @@ +package connector + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" +) + +// ElasticsearchSearchConnector executes a search query against an Elasticsearch index. +type ElasticsearchSearchConnector struct { + Client *http.Client + baseURL string // override for testing +} + +func (c *ElasticsearchSearchConnector) apiURL(base, path string) string { + if base == "" { + base = c.baseURL + } + return base + path +} + +func (c *ElasticsearchSearchConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { + baseURL, auth, err := extractElasticsearchCredential(params) + if err != nil { + return nil, fmt.Errorf("elasticsearch/search: %w", err) + } + if c.baseURL != "" { + baseURL = c.baseURL + } + + index, _ := params["index"].(string) + if index == "" { + return nil, fmt.Errorf("elasticsearch/search: index is required") + } + + body := map[string]any{} + if query, ok := params["query"].(map[string]any); ok { + body["query"] = query + } + if size, ok := extractInt(params["size"]); ok && size > 0 { + body["size"] = size + } + if from, ok := extractInt(params["from"]); ok && from >= 0 { + body["from"] = from + } + if source, ok := params["_source"].([]any); ok { + body["_source"] = source + } + + reqJSON, err := json.Marshal(body) + if err != nil { + return nil, fmt.Errorf("elasticsearch/search: marshaling request: %w", err) + } + + path := fmt.Sprintf("/%s/_search", url.PathEscape(index)) + req, err := http.NewRequestWithContext(ctx, "POST", baseURL+path, bytes.NewReader(reqJSON)) + if err != nil { + return nil, fmt.Errorf("elasticsearch/search: creating request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + auth(req) + + resp, err := httpClient(c.Client).Do(req) + if err != nil { + return nil, fmt.Errorf("elasticsearch/search: %w", err) + } + defer resp.Body.Close() + + respBody, err := io.ReadAll(io.LimitReader(resp.Body, DefaultMaxResponseBytes)) + if err != nil { + return nil, fmt.Errorf("elasticsearch/search: reading response: %w", err) + } + + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("elasticsearch/search: Elasticsearch returned %d: %s", resp.StatusCode, truncate(string(respBody), 500)) + } + + var result struct { + Took int `json:"took"` + Hits struct { + Total struct { + Value int `json:"value"` + } `json:"total"` + Hits []any `json:"hits"` + } `json:"hits"` + } + if err := json.Unmarshal(respBody, &result); err != nil { + return nil, fmt.Errorf("elasticsearch/search: parsing response: %w", err) + } + + return map[string]any{ + "hits": result.Hits.Hits, + "total": result.Hits.Total.Value, + "took": result.Took, + "count": len(result.Hits.Hits), + }, nil +} + +// ElasticsearchIndexConnector indexes a document into Elasticsearch. +type ElasticsearchIndexConnector struct { + Client *http.Client + baseURL string // override for testing +} + +func (c *ElasticsearchIndexConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { + baseURL, auth, err := extractElasticsearchCredential(params) + if err != nil { + return nil, fmt.Errorf("elasticsearch/index: %w", err) + } + if c.baseURL != "" { + baseURL = c.baseURL + } + + index, _ := params["index"].(string) + if index == "" { + return nil, fmt.Errorf("elasticsearch/index: index is required") + } + document, _ := params["document"].(map[string]any) + if document == nil { + return nil, fmt.Errorf("elasticsearch/index: document is required") + } + + reqJSON, err := json.Marshal(document) + if err != nil { + return nil, fmt.Errorf("elasticsearch/index: marshaling document: %w", err) + } + + method := "POST" + path := fmt.Sprintf("/%s/_doc", url.PathEscape(index)) + if docID, ok := params["id"].(string); ok && docID != "" { + method = "PUT" + path = fmt.Sprintf("/%s/_doc/%s", url.PathEscape(index), url.PathEscape(docID)) + } + + req, err := http.NewRequestWithContext(ctx, method, baseURL+path, bytes.NewReader(reqJSON)) + if err != nil { + return nil, fmt.Errorf("elasticsearch/index: creating request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + auth(req) + + resp, err := httpClient(c.Client).Do(req) + if err != nil { + return nil, fmt.Errorf("elasticsearch/index: %w", err) + } + defer resp.Body.Close() + + respBody, err := io.ReadAll(io.LimitReader(resp.Body, DefaultMaxResponseBytes)) + if err != nil { + return nil, fmt.Errorf("elasticsearch/index: reading response: %w", err) + } + + if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated { + return nil, fmt.Errorf("elasticsearch/index: Elasticsearch returned %d: %s", resp.StatusCode, truncate(string(respBody), 500)) + } + + var result struct { + ID string `json:"_id"` + Index string `json:"_index"` + Result string `json:"result"` + } + if err := json.Unmarshal(respBody, &result); err != nil { + return nil, fmt.Errorf("elasticsearch/index: parsing response: %w", err) + } + + return map[string]any{ + "id": result.ID, + "index": result.Index, + "result": result.Result, + }, nil +} + +// extractElasticsearchCredential extracts the base URL and auth setter from _credential. +// Credential fields: url (required), username+password (basic auth) or api_key (ApiKey auth). +func extractElasticsearchCredential(params map[string]any) (baseURL string, setAuth func(*http.Request), err error) { + raw, ok := params["_credential"] + if !ok || raw == nil { + return "", nil, fmt.Errorf("credential is required") + } + delete(params, "_credential") + + var cred map[string]string + switch v := raw.(type) { + case map[string]string: + cred = v + case map[string]any: + cred = make(map[string]string, len(v)) + for k, val := range v { + if s, ok := val.(string); ok { + cred[k] = s + } + } + default: + return "", nil, fmt.Errorf("credential is required") + } + + baseURL = cred["url"] + if baseURL == "" { + return "", nil, fmt.Errorf("credential must contain a 'url' field") + } + + if apiKey := cred["api_key"]; apiKey != "" { + setAuth = func(r *http.Request) { r.Header.Set("Authorization", "ApiKey "+apiKey) } + } else if username := cred["username"]; username != "" { + setAuth = func(r *http.Request) { r.SetBasicAuth(username, cred["password"]) } + } else { + setAuth = func(r *http.Request) {} + } + + return baseURL, setAuth, nil +} diff --git a/packages/engine/internal/connector/elasticsearch_test.go b/packages/engine/internal/connector/elasticsearch_test.go new file mode 100644 index 00000000..89014384 --- /dev/null +++ b/packages/engine/internal/connector/elasticsearch_test.go @@ -0,0 +1,220 @@ +package connector + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" +) + +func TestElasticsearchSearchConnector_SearchesDocuments(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != "POST" || r.URL.Path != "/my-index/_search" { + t.Errorf("unexpected %s %s", r.Method, r.URL.Path) + } + user, pass, ok := r.BasicAuth() + if !ok || user != "elastic" || pass != "secret" { + t.Errorf("unexpected basic auth: %s/%s", user, pass) + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{ + "took": 5, + "hits": map[string]any{ + "total": map[string]any{"value": 2}, + "hits": []any{map[string]any{"_id": "1"}, map[string]any{"_id": "2"}}, + }, + }) + })) + defer srv.Close() + + c := &ElasticsearchSearchConnector{baseURL: srv.URL} + out, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"url": srv.URL, "username": "elastic", "password": "secret"}, + "index": "my-index", + "query": map[string]any{"match_all": map[string]any{}}, + }) + if err != nil { + t.Fatal(err) + } + if out["total"].(int) != 2 { + t.Errorf("expected total=2, got %v", out["total"]) + } + if out["count"].(int) != 2 { + t.Errorf("expected count=2, got %v", out["count"]) + } +} + +func TestElasticsearchSearchConnector_WithAPIKey(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Authorization") != "ApiKey myapikey" { + t.Errorf("unexpected Authorization: %s", r.Header.Get("Authorization")) + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{ + "took": 1, + "hits": map[string]any{ + "total": map[string]any{"value": 0}, + "hits": []any{}, + }, + }) + })) + defer srv.Close() + + c := &ElasticsearchSearchConnector{baseURL: srv.URL} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"url": srv.URL, "api_key": "myapikey"}, + "index": "my-index", + }) + if err != nil { + t.Fatal(err) + } +} + +func TestElasticsearchSearchConnector_MissingIndex(t *testing.T) { + c := &ElasticsearchSearchConnector{baseURL: "http://unused"} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"url": "http://unused"}, + }) + if err == nil { + t.Fatal("expected error for missing index") + } +} + +func TestElasticsearchSearchConnector_MissingCredential(t *testing.T) { + c := &ElasticsearchSearchConnector{baseURL: "http://unused"} + _, err := c.Execute(t.Context(), map[string]any{ + "index": "my-index", + }) + if err == nil { + t.Fatal("expected error for missing credential") + } +} + +func TestElasticsearchSearchConnector_MissingURL(t *testing.T) { + c := &ElasticsearchSearchConnector{baseURL: "http://unused"} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"username": "elastic"}, + "index": "my-index", + }) + if err == nil { + t.Fatal("expected error for missing url in credential") + } +} + +func TestElasticsearchSearchConnector_APIError(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, `{"error":"index_not_found"}`, http.StatusNotFound) + })) + defer srv.Close() + + c := &ElasticsearchSearchConnector{baseURL: srv.URL} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"url": srv.URL}, + "index": "missing-index", + }) + if err == nil { + t.Fatal("expected error for 404") + } +} + +func TestElasticsearchIndexConnector_IndexesDocument(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != "POST" || r.URL.Path != "/my-index/_doc" { + t.Errorf("unexpected %s %s", r.Method, r.URL.Path) + } + var body map[string]any + json.NewDecoder(r.Body).Decode(&body) + if body["name"] != "Alice" { + t.Errorf("unexpected document: %v", body) + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusCreated) + json.NewEncoder(w).Encode(map[string]any{ + "_id": "abc123", + "_index": "my-index", + "result": "created", + }) + })) + defer srv.Close() + + c := &ElasticsearchIndexConnector{baseURL: srv.URL} + out, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"url": srv.URL}, + "index": "my-index", + "document": map[string]any{"name": "Alice"}, + }) + if err != nil { + t.Fatal(err) + } + if out["id"] != "abc123" { + t.Errorf("expected id=abc123, got %v", out["id"]) + } + if out["result"] != "created" { + t.Errorf("expected result=created, got %v", out["result"]) + } +} + +func TestElasticsearchIndexConnector_WithExplicitID(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != "PUT" || r.URL.Path != "/my-index/_doc/myid" { + t.Errorf("unexpected %s %s", r.Method, r.URL.Path) + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{"_id": "myid", "_index": "my-index", "result": "updated"}) + })) + defer srv.Close() + + c := &ElasticsearchIndexConnector{baseURL: srv.URL} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"url": srv.URL}, + "index": "my-index", + "id": "myid", + "document": map[string]any{"name": "Bob"}, + }) + if err != nil { + t.Fatal(err) + } +} + +func TestElasticsearchIndexConnector_MissingDocument(t *testing.T) { + c := &ElasticsearchIndexConnector{baseURL: "http://unused"} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"url": "http://unused"}, + "index": "my-index", + }) + if err == nil { + t.Fatal("expected error for missing document") + } +} + +func TestElasticsearchSearchConnector_MapAnyCredential(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Authorization") != "ApiKey mapkey" { + t.Errorf("unexpected auth: %s", r.Header.Get("Authorization")) + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{ + "took": 1, "hits": map[string]any{"total": map[string]any{"value": 0}, "hits": []any{}}, + }) + })) + defer srv.Close() + + c := &ElasticsearchSearchConnector{baseURL: srv.URL} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]any{"url": srv.URL, "api_key": "mapkey"}, + "index": "my-index", + }) + if err != nil { + t.Fatal(err) + } +} + +func TestRegistry_ElasticsearchConnectors(t *testing.T) { + r := NewRegistry() + if _, err := r.Get("elasticsearch/search"); err != nil { + t.Errorf("elasticsearch/search not registered: %v", err) + } + if _, err := r.Get("elasticsearch/index"); err != nil { + t.Errorf("elasticsearch/index not registered: %v", err) + } +} diff --git a/packages/engine/internal/connector/mongodb.go b/packages/engine/internal/connector/mongodb.go new file mode 100644 index 00000000..58bbaa3c --- /dev/null +++ b/packages/engine/internal/connector/mongodb.go @@ -0,0 +1,165 @@ +package connector + +import ( + "context" + "fmt" + "time" + + "go.mongodb.org/mongo-driver/v2/bson" + "go.mongodb.org/mongo-driver/v2/mongo" + "go.mongodb.org/mongo-driver/v2/mongo/options" +) + +const mongoConnectTimeout = 10 * time.Second + +// MongoFindConnector queries documents from a MongoDB collection. +type MongoFindConnector struct{} + +func (c *MongoFindConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { + client, db, coll, err := newMongoClient(ctx, params) + if err != nil { + return nil, fmt.Errorf("mongodb/find: %w", err) + } + defer client.Disconnect(ctx) + + filter := bsonFromMap(params["filter"]) + opts := options.Find() + if limit, ok := extractInt(params["limit"]); ok && limit > 0 { + opts.SetLimit(int64(limit)) + } + if skip, ok := extractInt(params["skip"]); ok && skip > 0 { + opts.SetSkip(int64(skip)) + } + if projection, ok := params["projection"].(map[string]any); ok { + opts.SetProjection(bson.M(projection)) + } + + cursor, err := client.Database(db).Collection(coll).Find(ctx, filter, opts) + if err != nil { + return nil, fmt.Errorf("mongodb/find: %w", err) + } + defer cursor.Close(ctx) + + var documents []any + if err := cursor.All(ctx, &documents); err != nil { + return nil, fmt.Errorf("mongodb/find: decoding results: %w", err) + } + if documents == nil { + documents = []any{} + } + + return map[string]any{ + "documents": documents, + "count": len(documents), + }, nil +} + +// MongoAggregateConnector runs an aggregation pipeline on a MongoDB collection. +type MongoAggregateConnector struct{} + +func (c *MongoAggregateConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { + client, db, coll, err := newMongoClient(ctx, params) + if err != nil { + return nil, fmt.Errorf("mongodb/aggregate: %w", err) + } + defer client.Disconnect(ctx) + + pipelineRaw, _ := params["pipeline"].([]any) + if len(pipelineRaw) == 0 { + return nil, fmt.Errorf("mongodb/aggregate: pipeline is required") + } + + pipeline := make(bson.A, len(pipelineRaw)) + for i, stage := range pipelineRaw { + if m, ok := stage.(map[string]any); ok { + pipeline[i] = bson.M(m) + } else { + pipeline[i] = stage + } + } + + cursor, err := client.Database(db).Collection(coll).Aggregate(ctx, pipeline) + if err != nil { + return nil, fmt.Errorf("mongodb/aggregate: %w", err) + } + defer cursor.Close(ctx) + + var documents []any + if err := cursor.All(ctx, &documents); err != nil { + return nil, fmt.Errorf("mongodb/aggregate: decoding results: %w", err) + } + if documents == nil { + documents = []any{} + } + + return map[string]any{ + "documents": documents, + "count": len(documents), + }, nil +} + +// newMongoClient connects to MongoDB and returns the client plus target db/collection names. +// Credential: {uri: "mongodb://...", database: "mydb", collection: "mycoll"} +func newMongoClient(ctx context.Context, params map[string]any) (*mongo.Client, string, string, error) { + raw, ok := params["_credential"] + if !ok || raw == nil { + return nil, "", "", fmt.Errorf("credential is required") + } + delete(params, "_credential") + + var uri, credDB string + switch cred := raw.(type) { + case map[string]string: + uri = cred["uri"] + credDB = cred["database"] + case map[string]any: + uri, _ = cred["uri"].(string) + credDB, _ = cred["database"].(string) + default: + return nil, "", "", fmt.Errorf("credential is required") + } + if uri == "" { + return nil, "", "", fmt.Errorf("credential must contain a 'uri' field") + } + + db, _ := params["database"].(string) + if db == "" { + db = credDB + } + if db == "" { + return nil, "", "", fmt.Errorf("database is required (set in params or credential)") + } + + coll, _ := params["collection"].(string) + if coll == "" { + return nil, "", "", fmt.Errorf("collection is required") + } + + connectCtx, cancel := context.WithTimeout(ctx, mongoConnectTimeout) + defer cancel() + + client, err := mongo.Connect(options.Client().ApplyURI(uri)) + if err != nil { + return nil, "", "", fmt.Errorf("connecting to MongoDB: %w", err) + } + if err := client.Ping(connectCtx, nil); err != nil { + client.Disconnect(ctx) + return nil, "", "", fmt.Errorf("pinging MongoDB: %w", err) + } + return client, db, coll, nil +} + +// bsonFromMap converts a map[string]any filter param into a bson.D, or returns +// an empty bson.D (match-all) if the value is nil or not a map. +func bsonFromMap(v any) bson.D { + if m, ok := v.(map[string]any); ok && len(m) > 0 { + return bson.D(func() []bson.E { + out := make([]bson.E, 0, len(m)) + for k, val := range m { + out = append(out, bson.E{Key: k, Value: val}) + } + return out + }()) + } + return bson.D{} +} diff --git a/packages/engine/internal/connector/mongodb_test.go b/packages/engine/internal/connector/mongodb_test.go new file mode 100644 index 00000000..93332b65 --- /dev/null +++ b/packages/engine/internal/connector/mongodb_test.go @@ -0,0 +1,122 @@ +package connector + +import ( + "testing" +) + +func TestMongoFindConnector_MissingCredential(t *testing.T) { + c := &MongoFindConnector{} + _, err := c.Execute(t.Context(), map[string]any{ + "database": "mydb", + "collection": "users", + }) + if err == nil { + t.Fatal("expected error for missing credential") + } +} + +func TestMongoFindConnector_MissingURI(t *testing.T) { + c := &MongoFindConnector{} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"database": "mydb"}, + "database": "mydb", + "collection": "users", + }) + if err == nil { + t.Fatal("expected error for missing uri in credential") + } +} + +func TestMongoFindConnector_MissingDatabase(t *testing.T) { + c := &MongoFindConnector{} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"uri": "mongodb://localhost:27017"}, + "collection": "users", + }) + if err == nil { + t.Fatal("expected error for missing database") + } +} + +func TestMongoFindConnector_MissingCollection(t *testing.T) { + c := &MongoFindConnector{} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"uri": "mongodb://localhost:27017", "database": "mydb"}, + }) + if err == nil { + t.Fatal("expected error for missing collection") + } +} + +func TestMongoAggregateConnector_MissingCredential(t *testing.T) { + c := &MongoAggregateConnector{} + _, err := c.Execute(t.Context(), map[string]any{ + "database": "mydb", + "collection": "orders", + "pipeline": []any{map[string]any{"$match": map[string]any{}}}, + }) + if err == nil { + t.Fatal("expected error for missing credential") + } +} + +func TestMongoAggregateConnector_MissingPipeline(t *testing.T) { + c := &MongoAggregateConnector{} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"uri": "mongodb://localhost:27017", "database": "mydb"}, + "collection": "orders", + }) + if err == nil { + t.Fatal("expected error for missing pipeline") + } +} + +func TestMongoAggregateConnector_EmptyPipeline(t *testing.T) { + c := &MongoAggregateConnector{} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"uri": "mongodb://localhost:27017", "database": "mydb"}, + "collection": "orders", + "pipeline": []any{}, + }) + if err == nil { + t.Fatal("expected error for empty pipeline") + } +} + +func TestMongoFindConnector_DatabaseFromCredential(t *testing.T) { + // Verify database can come from credential — error should be about connection, not about missing database. + c := &MongoFindConnector{} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"uri": "mongodb://localhost:27017", "database": "mydb"}, + "collection": "users", + }) + // We expect a connection error (no real MongoDB), not a "database is required" error. + if err == nil { + t.Fatal("expected connection error") + } + if err.Error() == "mongodb/find: database is required (set in params or credential)" { + t.Errorf("database should have been resolved from credential, got: %v", err) + } +} + +func TestMongoFindConnector_MapAnyCredential(t *testing.T) { + c := &MongoFindConnector{} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]any{"uri": "mongodb://localhost:27017", "database": "mydb"}, + "collection": "users", + }) + // Expect a connection error (no real MongoDB running), not a credential parse error. + if err == nil { + t.Fatal("expected connection error") + } +} + +func TestRegistry_MongoDBConnectors(t *testing.T) { + r := NewRegistry() + if _, err := r.Get("mongodb/find"); err != nil { + t.Errorf("mongodb/find not registered: %v", err) + } + if _, err := r.Get("mongodb/aggregate"); err != nil { + t.Errorf("mongodb/aggregate not registered: %v", err) + } +} diff --git a/packages/engine/internal/connector/notion.go b/packages/engine/internal/connector/notion.go index 1189c4a2..524f61bb 100644 --- a/packages/engine/internal/connector/notion.go +++ b/packages/engine/internal/connector/notion.go @@ -30,7 +30,7 @@ func (c *NotionCreatePageConnector) apiURL(path string) string { } func (c *NotionCreatePageConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { - token, err := extractNotionToken(params) + token, err := extractBearerToken(params) if err != nil { return nil, fmt.Errorf("notion/create_page: %w", err) } @@ -141,7 +141,7 @@ func (c *NotionQueryDatabaseConnector) apiURL(path string) string { } func (c *NotionQueryDatabaseConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { - token, err := extractNotionToken(params) + token, err := extractBearerToken(params) if err != nil { return nil, fmt.Errorf("notion/query_database: %w", err) } @@ -213,28 +213,3 @@ func (c *NotionQueryDatabaseConnector) Execute(ctx context.Context, params map[s } return out, nil } - -// extractNotionToken pulls the Notion integration token from _credential. -// Accepts both map[string]string (engine-injected) and map[string]any -// (JSON/CEL-deserialised) credential shapes. -func extractNotionToken(params map[string]any) (string, error) { - raw, ok := params["_credential"] - if !ok || raw == nil { - return "", fmt.Errorf("credential is required") - } - delete(params, "_credential") - - var token string - switch cred := raw.(type) { - case map[string]string: - token = cred["token"] - case map[string]any: - token, _ = cred["token"].(string) - default: - return "", fmt.Errorf("credential is required") - } - if token == "" { - return "", fmt.Errorf("credential must contain a 'token' field") - } - return token, nil -} diff --git a/packages/engine/internal/connector/redis.go b/packages/engine/internal/connector/redis.go new file mode 100644 index 00000000..bfecef43 --- /dev/null +++ b/packages/engine/internal/connector/redis.go @@ -0,0 +1,119 @@ +package connector + +import ( + "context" + "fmt" + "time" + + "github.com/redis/go-redis/v9" +) + +// RedisGetConnector retrieves the value of a key from Redis. +type RedisGetConnector struct{} + +func (c *RedisGetConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { + client, err := newRedisClient(params) + if err != nil { + return nil, fmt.Errorf("redis/get: %w", err) + } + defer client.Close() + + key, _ := params["key"].(string) + if key == "" { + return nil, fmt.Errorf("redis/get: key is required") + } + + val, err := client.Get(ctx, key).Result() + if err == redis.Nil { + return map[string]any{"value": nil, "exists": false}, nil + } + if err != nil { + return nil, fmt.Errorf("redis/get: %w", err) + } + return map[string]any{"value": val, "exists": true}, nil +} + +// RedisSetConnector sets the value of a key in Redis. +type RedisSetConnector struct{} + +func (c *RedisSetConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { + client, err := newRedisClient(params) + if err != nil { + return nil, fmt.Errorf("redis/set: %w", err) + } + defer client.Close() + + key, _ := params["key"].(string) + if key == "" { + return nil, fmt.Errorf("redis/set: key is required") + } + value, ok := params["value"] + if !ok { + return nil, fmt.Errorf("redis/set: value is required") + } + + var ttl time.Duration + if ttlSec, ok := extractInt(params["ttl_seconds"]); ok && ttlSec > 0 { + ttl = time.Duration(ttlSec) * time.Second + } + + if err := client.Set(ctx, key, value, ttl).Err(); err != nil { + return nil, fmt.Errorf("redis/set: %w", err) + } + return map[string]any{"ok": true}, nil +} + +// RedisPublishConnector publishes a message to a Redis pub/sub channel. +type RedisPublishConnector struct{} + +func (c *RedisPublishConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { + client, err := newRedisClient(params) + if err != nil { + return nil, fmt.Errorf("redis/publish: %w", err) + } + defer client.Close() + + channel, _ := params["channel"].(string) + if channel == "" { + return nil, fmt.Errorf("redis/publish: channel is required") + } + message, _ := params["message"].(string) + if message == "" { + return nil, fmt.Errorf("redis/publish: message is required") + } + + receivers, err := client.Publish(ctx, channel, message).Result() + if err != nil { + return nil, fmt.Errorf("redis/publish: %w", err) + } + return map[string]any{"receivers": receivers}, nil +} + +// newRedisClient builds a redis.Client from the _credential param. +// Credential: {url: "redis://[:password@]host:port/db"} +func newRedisClient(params map[string]any) (*redis.Client, error) { + raw, ok := params["_credential"] + if !ok || raw == nil { + return nil, fmt.Errorf("credential is required") + } + delete(params, "_credential") + + var redisURL string + switch cred := raw.(type) { + case map[string]string: + redisURL = cred["url"] + case map[string]any: + redisURL, _ = cred["url"].(string) + default: + return nil, fmt.Errorf("credential is required") + } + if redisURL == "" { + return nil, fmt.Errorf("credential must contain a 'url' field") + } + + opts, err := redis.ParseURL(redisURL) + if err != nil { + return nil, fmt.Errorf("credential url is invalid: %w", err) + } + return redis.NewClient(opts), nil +} diff --git a/packages/engine/internal/connector/redis_test.go b/packages/engine/internal/connector/redis_test.go new file mode 100644 index 00000000..f152b640 --- /dev/null +++ b/packages/engine/internal/connector/redis_test.go @@ -0,0 +1,126 @@ +package connector + +import ( + "testing" +) + +func TestRedisGetConnector_MissingCredential(t *testing.T) { + c := &RedisGetConnector{} + _, err := c.Execute(t.Context(), map[string]any{ + "key": "mykey", + }) + if err == nil { + t.Fatal("expected error for missing credential") + } +} + +func TestRedisGetConnector_MissingKey(t *testing.T) { + c := &RedisGetConnector{} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"url": "redis://localhost:6379"}, + }) + if err == nil { + t.Fatal("expected error for missing key") + } +} + +func TestRedisGetConnector_InvalidURL(t *testing.T) { + c := &RedisGetConnector{} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"url": "not-a-valid-redis-url://???"}, + "key": "mykey", + }) + if err == nil { + t.Fatal("expected error for invalid redis url") + } +} + +func TestRedisGetConnector_MissingURLInCredential(t *testing.T) { + c := &RedisGetConnector{} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"host": "localhost"}, + "key": "mykey", + }) + if err == nil { + t.Fatal("expected error for missing url in credential") + } +} + +func TestRedisSetConnector_MissingKey(t *testing.T) { + c := &RedisSetConnector{} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"url": "redis://localhost:6379"}, + "value": "hello", + }) + if err == nil { + t.Fatal("expected error for missing key") + } +} + +func TestRedisSetConnector_MissingValue(t *testing.T) { + c := &RedisSetConnector{} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"url": "redis://localhost:6379"}, + "key": "mykey", + }) + if err == nil { + t.Fatal("expected error for missing value") + } +} + +func TestRedisSetConnector_MissingCredential(t *testing.T) { + c := &RedisSetConnector{} + _, err := c.Execute(t.Context(), map[string]any{ + "key": "mykey", + "value": "hello", + }) + if err == nil { + t.Fatal("expected error for missing credential") + } +} + +func TestRedisPublishConnector_MissingChannel(t *testing.T) { + c := &RedisPublishConnector{} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"url": "redis://localhost:6379"}, + "message": "hello", + }) + if err == nil { + t.Fatal("expected error for missing channel") + } +} + +func TestRedisPublishConnector_MissingMessage(t *testing.T) { + c := &RedisPublishConnector{} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"url": "redis://localhost:6379"}, + "channel": "events", + }) + if err == nil { + t.Fatal("expected error for missing message") + } +} + +func TestRedisGetConnector_MapAnyCredential(t *testing.T) { + c := &RedisGetConnector{} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]any{"url": "not-a-valid-redis-url://???"}, + "key": "mykey", + }) + if err == nil { + t.Fatal("expected error for invalid url from map[string]any credential") + } +} + +func TestRegistry_RedisConnectors(t *testing.T) { + r := NewRegistry() + if _, err := r.Get("redis/get"); err != nil { + t.Errorf("redis/get not registered: %v", err) + } + if _, err := r.Get("redis/set"); err != nil { + t.Errorf("redis/set not registered: %v", err) + } + if _, err := r.Get("redis/publish"); err != nil { + t.Errorf("redis/publish not registered: %v", err) + } +} diff --git a/packages/engine/internal/connector/tools.go b/packages/engine/internal/connector/tools.go index 92b07970..40d3836d 100644 --- a/packages/engine/internal/connector/tools.go +++ b/packages/engine/internal/connector/tools.go @@ -348,6 +348,31 @@ func truncateOlderToolResults(messages []map[string]any, targetBytes int) int { return truncated } +// extractBearerToken pulls a "token" field from the _credential param. +// Accepts both map[string]string (engine-injected) and map[string]any +// (JSON/CEL-deserialised) credential shapes, and deletes _credential from params. +func extractBearerToken(params map[string]any) (string, error) { + raw, ok := params["_credential"] + if !ok || raw == nil { + return "", fmt.Errorf("credential is required") + } + delete(params, "_credential") + + var token string + switch cred := raw.(type) { + case map[string]string: + token = cred["token"] + case map[string]any: + token, _ = cred["token"].(string) + default: + return "", fmt.Errorf("credential is required") + } + if token == "" { + return "", fmt.Errorf("credential must contain a 'token' field") + } + return token, nil +} + // extractInt attempts to extract an int from a value that may be int or float64 // (JSON unmarshalling often produces float64). func extractInt(v any) (int, bool) { From d05fba674d7eae5dcdb127c2d4817598ab8fe869 Mon Sep 17 00:00:00 2001 From: Michael McNees Date: Sat, 30 May 2026 19:59:31 -0400 Subject: [PATCH 2/2] fix(connector): remove unused ElasticsearchSearch.apiURL, pass Client through discordPostMessage Fixes lint: func (*ElasticsearchSearchConnector).apiURL is unused. Addresses Codex feedback: discordPostMessage now receives the connector's http.Client instead of always using the default client. Co-Authored-By: Claude Sonnet 4.6 --- packages/engine/internal/connector/discord.go | 8 ++++---- packages/engine/internal/connector/elasticsearch.go | 7 ------- 2 files changed, 4 insertions(+), 11 deletions(-) diff --git a/packages/engine/internal/connector/discord.go b/packages/engine/internal/connector/discord.go index ab47fab6..fbe53c96 100644 --- a/packages/engine/internal/connector/discord.go +++ b/packages/engine/internal/connector/discord.go @@ -46,7 +46,7 @@ func (c *DiscordSendConnector) Execute(ctx context.Context, params map[string]an body["tts"] = tts } - return discordPostMessage(ctx, c.apiURL(fmt.Sprintf("/channels/%s/messages", url.PathEscape(channelID))), token, body, "discord/send") + return discordPostMessage(ctx, c.Client, c.apiURL(fmt.Sprintf("/channels/%s/messages", url.PathEscape(channelID))), token, body, "discord/send") } // DiscordEmbedConnector sends an embed message to a Discord channel. @@ -99,10 +99,10 @@ func (c *DiscordEmbedConnector) Execute(ctx context.Context, params map[string]a body["content"] = content } - return discordPostMessage(ctx, c.apiURL(fmt.Sprintf("/channels/%s/messages", url.PathEscape(channelID))), token, body, "discord/embed") + return discordPostMessage(ctx, c.Client, c.apiURL(fmt.Sprintf("/channels/%s/messages", url.PathEscape(channelID))), token, body, "discord/embed") } -func discordPostMessage(ctx context.Context, endpoint, token string, body map[string]any, action string) (map[string]any, error) { +func discordPostMessage(ctx context.Context, client *http.Client, endpoint, token string, body map[string]any, action string) (map[string]any, error) { reqJSON, err := json.Marshal(body) if err != nil { return nil, fmt.Errorf("%s: marshaling request: %w", action, err) @@ -115,7 +115,7 @@ func discordPostMessage(ctx context.Context, endpoint, token string, body map[st req.Header.Set("Authorization", "Bot "+token) req.Header.Set("Content-Type", "application/json") - resp, err := httpClient(nil).Do(req) + resp, err := httpClient(client).Do(req) if err != nil { return nil, fmt.Errorf("%s: %w", action, err) } diff --git a/packages/engine/internal/connector/elasticsearch.go b/packages/engine/internal/connector/elasticsearch.go index cdad994a..2656c86f 100644 --- a/packages/engine/internal/connector/elasticsearch.go +++ b/packages/engine/internal/connector/elasticsearch.go @@ -16,13 +16,6 @@ type ElasticsearchSearchConnector struct { baseURL string // override for testing } -func (c *ElasticsearchSearchConnector) apiURL(base, path string) string { - if base == "" { - base = c.baseURL - } - return base + path -} - func (c *ElasticsearchSearchConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { baseURL, auth, err := extractElasticsearchCredential(params) if err != nil {