feat(connector): Discord, Elasticsearch, Datadog, Redis, and MongoDB connectors

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=

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

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=

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
-}

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
-}

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
 }
 

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
+}