Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
some PR comments, updated register unregister trigger
- Loading branch information
1 parent
de300e0
commit 62d4dbc
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -12,6 +12,7 @@ import ( | |
| "github.com/smartcontractkit/chainlink-common/pkg/capabilities" | ||
| "github.com/smartcontractkit/chainlink-common/pkg/services" | ||
| "github.com/smartcontractkit/chainlink-common/pkg/types/core" | ||
| "github.com/smartcontractkit/chainlink-common/pkg/values" | ||
| "github.com/smartcontractkit/chainlink/v2/core/logger" | ||
| "github.com/smartcontractkit/chainlink/v2/core/services/gateway/api" | ||
| "github.com/smartcontractkit/chainlink/v2/core/services/gateway/connector" | ||
|
|
@@ -21,24 +22,29 @@ import ( | |
|
|
||
| const defaultSendChannelBufferSize = 1000 | ||
|
|
||
| const triggerType = "[email protected]" | ||
|
|
||
| type Response struct { | ||
| // TODO: what is the format for ACCEPTED, PENDING, COMPLETED status? | ||
| // Status string `json:"status"`? | ||
| Success bool `json:"success"` | ||
| ErrorMessage string `json:"error_message,omitempty"` | ||
| } | ||
|
|
||
| // Handles connections to the webapi trigger | ||
| type triggerConnectorHandler struct { | ||
| services.StateMachine | ||
|
|
||
| capabilities.CapabilityInfo | ||
| connector connector.GatewayConnector | ||
| lggr logger.Logger | ||
| mu sync.Mutex | ||
| registeredWorkflows map[string]chan capabilities.TriggerResponse | ||
| signerKey *ecdsa.PrivateKey | ||
| } | ||
|
|
||
| var _ capabilities.TriggerCapability = (*triggerConnectorHandler)(nil) | ||
| var _ services.Service = &triggerConnectorHandler{} | ||
|
|
||
| func NewTrigger(config string, registry core.CapabilitiesRegistry, connector connector.GatewayConnector, signerKey *ecdsa.PrivateKey, lggr logger.Logger) (job.ServiceCtx, error) { | ||
| // TODO (CAPPL-22, CAPPL-24): | ||
|
|
@@ -48,18 +54,48 @@ func NewTrigger(config string, registry core.CapabilitiesRegistry, connector con | |
| // - manage trigger subscriptions | ||
| // - process incoming trigger events and related metadata | ||
|
|
||
| handler := &triggerConnectorHandler{ | ||
| connector: connector, | ||
| signerKey: signerKey, | ||
| lggr: lggr.Named("WorkflowConnectorHandler"), | ||
| } | ||
|
|
||
| return handler, nil | ||
| } | ||
|
|
||
| // https://gateway-us-1.chain.link/web-trigger | ||
| // { | ||
| // jsonrpc: "2.0", | ||
| // id: "...", | ||
| // method: "web-trigger", | ||
| // params: { | ||
| // signature: "...", | ||
| // body: { | ||
| // don_id: "workflow_123", | ||
| // payload: { | ||
| // trigger_id: "[email protected]", | ||
| // trigger_event_id: "action_1234567890", | ||
| // timestamp: 1234567890, | ||
| // sub-events: [ | ||
| // { | ||
| // topics: ["daily_price_update"], | ||
| // params: { | ||
| // bid: "101", | ||
| // ask: "102" | ||
| // } | ||
| // }, | ||
| // { | ||
| // topics: ["daily_message", "summary"], | ||
| // params: { | ||
| // message: "all good!", | ||
| // } | ||
| // }, | ||
| // ] | ||
| // } | ||
| // } | ||
| // } | ||
| // } | ||
|
|
||
| // from Web API Trigger Doc | ||
| // trigger_id - ID of the trigger corresponding to the capability ID | ||
| // trigger_event_id - uniquely identifies generated event (scoped to trigger_id and sender)xx | ||
|
|
@@ -69,95 +105,112 @@ func NewTrigger(config string, registry core.CapabilitiesRegistry, connector con | |
| // workflow_owners - [OPTIONAL] list of workflow owners allowed to receive this event (affects all workflows if empty) | ||
| // params - key-value pairs that will be used as trigger output in the workflow Engine (translated to values.Map) | ||
|
|
||
| type TriggerRequestPayload struct { | ||
| TriggerId string `json:"trigger_id"` | ||
| TriggerEventId string `json:"trigger_event_id"` | ||
| // how are timestamps defined? ISO-8601 or UTC seconds or UTC ms? | ||
| Timestamp uint `json:"timestamp"` | ||
| SubEvents []SubEvents `json:"sub_events"` | ||
| } | ||
|
|
||
| type SubEvents struct { | ||
| Topics []string `json:"topics"` | ||
| Params values.Map `json:"params"` | ||
| } | ||
|
|
||
| func (h *triggerConnectorHandler) HandleGatewayMessage(ctx context.Context, gatewayID string, msg *api.Message) { | ||
| body := &msg.Body | ||
| fromAddr := ethCommon.HexToAddress(body.Sender) | ||
| // TODO: apply allowlist and rate-limiting | ||
| h.lggr.Debugw("handling gateway request", "id", gatewayID, "method", body.Method, "sender", fromAddr) | ||
| var payload TriggerRequestPayload | ||
| err := json.Unmarshal(body.Payload, &payload) | ||
| if err != nil { | ||
| return | ||
| } | ||
|
|
||
| // TODO: how to convert payload to *values.Map. Parse directly to that instead of the structs? | ||
|
|
||
| // TODO: How/where to check timestamp for freshness | ||
|
|
||
| switch body.Method { | ||
| case workflow.MethodWebAPITrigger: | ||
| h.lggr.Debugw("added MethodWebAPITrigger message", "payload", string(body.Payload)) | ||
|
|
||
| for triggerID, trigger := range h.registeredWorkflows { | ||
|
|
||
| // TODO: CAPPL-24 extract the topic and then match the subscriber to this messages triggers. | ||
| // TODO: CAPPL-24 check the topic to see if the method is a duplicate and the trigger has been sent, ie PENDING | ||
| // TODO: Question asked in Web API trigger about checking for completed Triggers to return COMPLETED | ||
| tr := capabilities.TriggerResponse{ | ||
| Event: capabilities.TriggerEvent{ | ||
| TriggerType: triggerType, | ||
| ID: triggerID, | ||
| Outputs: payload, // must be *values.Map | ||
|
Check failure on line 149 in core/capabilities/webapi/trigger.go
|
||
| }, | ||
| } | ||
| trigger <- tr | ||
| } | ||
|
|
||
| // TODO: ACCEPTED, PENDING, COMPLETED | ||
| response := Response{Success: true} | ||
| h.sendResponse(ctx, gatewayID, body, response) | ||
| default: | ||
| h.lggr.Errorw("unsupported method", "id", gatewayID, "method", body.Method) | ||
| } | ||
| } | ||
|
|
||
| func (h *triggerConnectorHandler) RegisterTrigger(ctx context.Context, req capabilities.TriggerRegistrationRequest) (<-chan capabilities.TriggerResponse, error) { | ||
| h.mu.Lock() | ||
| defer h.mu.Unlock() | ||
| _, ok := h.registeredWorkflows[req.TriggerID] | ||
| if ok { | ||
| return nil, fmt.Errorf("triggerId %s already registered", req.TriggerID) | ||
| } | ||
|
|
||
| callbackCh := make(chan capabilities.TriggerResponse, defaultSendChannelBufferSize) | ||
| // TODO: CAPPL-24 how do we extract the topic and then define the trigger by that? | ||
| // It's not TriggerID because TriggerID is concat of workflow ID and the trigger's index in the spec (what does that mean?) | ||
| h.registeredWorkflows[req.TriggerID] = callbackCh | ||
|
|
||
| return callbackCh, nil | ||
| } | ||
|
|
||
| func (h *triggerConnectorHandler) UnregisterTrigger(ctx context.Context, req capabilities.TriggerRegistrationRequest) error { | ||
| h.mu.Lock() | ||
| defer h.mu.Unlock() | ||
| trigger, ok := h.registeredWorkflows[req.TriggerID] | ||
| if ok { | ||
| return fmt.Errorf("triggerId %s not registered", req.TriggerID) | ||
| } | ||
|
|
||
| // Close callback channel | ||
| close(trigger) | ||
| // Remove from triggers context | ||
| delete(h.registeredWorkflows, req.TriggerID) | ||
| return nil | ||
| } | ||
|
|
||
| func (h *triggerConnectorHandler) Start(ctx context.Context) error { | ||
| return h.StartOnce("GatewayConnectorServiceWrapper", func() error { | ||
| return h.connector.AddHandler([]string{"web_trigger"}, h) | ||
| }) | ||
| } | ||
| func (h *triggerConnectorHandler) Close() error { | ||
| return h.StopOnce("GatewayConnectorServiceWrapper", func() error { | ||
| return nil | ||
| }) | ||
| } | ||
|
|
||
| func (h *triggerConnectorHandler) HealthReport() map[string]error { | ||
| return map[string]error{h.Name(): h.Healthy()} | ||
| } | ||
|
|
||
| func (h *triggerConnectorHandler) Name() string { | ||
| return "WebAPITrigger" | ||
| } | ||
|
|
||
| func (h *triggerConnectorHandler) sendResponse(ctx context.Context, gatewayID string, requestBody *api.MessageBody, payload any) error { | ||
| payloadJson, err := json.Marshal(payload) | ||
| if err != nil { | ||
| return err | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -16,13 +16,48 @@ import ( | |
| "github.com/smartcontractkit/chainlink/v2/core/services/gateway/api" | ||
| ) | ||
|
|
||
| // https://gateway-us-1.chain.link/web-trigger | ||
| // { | ||
| // jsonrpc: "2.0", | ||
| // id: "...", | ||
| // method: "web-trigger", | ||
| // params: { | ||
| // signature: "...", | ||
| // body: { | ||
| // don_id: "workflow_123", | ||
| // payload: { | ||
| // trigger_id: "[email protected]", | ||
| // trigger_event_id: "action_1234567890", | ||
| // timestamp: 1234567890, | ||
| // sub-events: [ | ||
| // { | ||
| // topics: ["daily_price_update"], | ||
| // params: { | ||
| // bid: "101", | ||
| // ask: "102" | ||
| // } | ||
| // }, | ||
| // { | ||
| // topics: ["daily_message", "summary"], | ||
| // params: { | ||
| // message: "all good!", | ||
| // } | ||
| // }, | ||
| // ] | ||
| // } | ||
| // } | ||
| // } | ||
| // } | ||
|
|
||
| func main() { | ||
| gatewayURL := flag.String("gateway_url", "http://localhost:5002", "Gateway URL") | ||
| privateKey := flag.String("private_key", "65456ffb8af4a2b93959256a8e04f6f2fe0943579fb3c9c3350593aabb89023f", "Private key to sign the message with") | ||
| messageID := flag.String("id", "12345", "Request ID") | ||
| methodName := flag.String("method", "web_trigger", "Method name") | ||
| donID := flag.String("don_id", "workflow_don_1", "DON ID") | ||
| // workflowSpec := flag.String("workflow_spec", "[my spec abcd]", "Workflow Spec") | ||
| // payloadJSON := []byte("{\"spec\": \"" + *workflowSpec + "\"}") | ||
|
|
||
| flag.Parse() | ||
|
|
||
| if privateKey == nil || *privateKey == "" { | ||
|
|
@@ -41,10 +76,29 @@ func main() { | |
| fmt.Println("error parsing private key", err) | ||
| return | ||
| } | ||
|
|
||
| payload := `{ | ||
| trigger_id: "[email protected]", | ||
| trigger_event_id: "action_1234567890", | ||
| timestamp: 1234567890, | ||
| sub-events: [ | ||
| { | ||
| topics: ["daily_price_update"], | ||
| params: { | ||
| bid: "101", | ||
| ask: "102" | ||
| } | ||
| }, | ||
| { | ||
| topics: ["daily_message", "summary"], | ||
| params: { | ||
| message: "all good!", | ||
| } | ||
| }, | ||
| ] | ||
| } | ||
| ` | ||
| payloadJSON := []byte(payload) | ||
| msg := &api.Message{ | ||
| Body: api.MessageBody{ | ||
| MessageId: *messageID, | ||
|
|
@@ -53,7 +107,6 @@ func main() { | |
| Payload: json.RawMessage(payloadJSON), | ||
| }, | ||
| } | ||
| if err = msg.Sign(key); err != nil { | ||
| fmt.Println("error signing message", err) | ||
| return | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters