Runlog refactor Tier 3

blockchain/agoric/agoric.go

@@ -0,0 +1,30 @@
+package agoric
+
+import (
+	"github.com/smartcontractkit/external-initiator/store"
+	"github.com/smartcontractkit/external-initiator/subscriber"
+)
+
+const Name = "agoric"
+
+type manager struct {
+	jobid string
+
+	conn *subscriber.WebsocketConnection
+}
+
+func createManager(sub store.Subscription) (*manager, error) {
+	conn, err := subscriber.NewCoreWebsocketConnection(sub.Endpoint)
+	if err != nil {
+		return nil, err
+	}
+
+	return &manager{
+		jobid: sub.Job,
+		conn:  conn,
+	}, nil
+}
+
+func (m manager) Stop() {
+	m.conn.Stop()
+}

blockchain/agoric/runlog.go

@@ -0,0 +1,130 @@
+package agoric
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+
+	"github.com/smartcontractkit/external-initiator/blockchain/common"
+	"github.com/smartcontractkit/external-initiator/store"
+
+	"github.com/smartcontractkit/chainlink/core/logger"
+)
+
+// linkDecimals is the number of decimal places in $LINK
+// This value must be greater than linkAgoricDecimals
+const linkDecimals = 18
+
+// linkAgoricDecimals is the number of decimal places in a uaglink token
+// FIXME: Ideally the same as linkDecimals.
+// This value must be lower than linkDecimals
+const linkAgoricDecimals = 6
+
+var (
+	errNoJobMatch = errors.New("event did not match a job")
+)
+
+type runlogManager struct {
+	*manager
+}
+
+func CreateRunlogManager(sub store.Subscription) (*runlogManager, error) {
+	manager, err := createManager(sub)
+	if err != nil {
+		return nil, err
+	}
+
+	return &runlogManager{
+		manager: manager,
+	}, nil
+}
+
+func (r runlogManager) SubscribeEvents(ctx context.Context, ch chan<- common.RunlogRequest) error {
+	msgs := make(chan []byte)
+	go r.conn.Read(msgs)
+
+	go func() {
+		for {
+			select {
+			case msg := <-msgs:
+				req, err := r.parseRequests(msg)
+				if err == errNoJobMatch {
+					continue
+				} else if err != nil {
+					logger.Error(err)
+					continue
+				}
+				ch <- req
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+
+	return nil
+}
+
+func (r runlogManager) CreateJobRun(request common.RunlogRequest) map[string]interface{} {
+	// This implementation does not need to make any changes
+	// to the request payload.
+	return request
+}
+
+type agoricEvent struct {
+	Type string          `json:"type"`
+	Data json.RawMessage `json:"data"`
+}
+
+type agoricOnQueryData struct {
+	QueryID string          `json:"queryId"`
+	Query   json.RawMessage `json:"query"`
+	Fee     int64           `json:"fee"`
+}
+
+type chainlinkQuery struct {
+	JobID  string                 `json:"jobid"`
+	Params map[string]interface{} `json:"params"`
+}
+
+func (r runlogManager) parseRequests(data []byte) (common.RunlogRequest, error) {
+	var agEvent agoricEvent
+	err := json.Unmarshal(data, &agEvent)
+	if err != nil {
+		return nil, err
+	}
+
+	if agEvent.Type != "oracleServer/onQuery" {
+		return nil, errNoJobMatch
+	}
+
+	var onQueryData agoricOnQueryData
+	err = json.Unmarshal(agEvent.Data, &onQueryData)
+	if err != nil {
+		return nil, err
+	}
+
+	var query chainlinkQuery
+	err = json.Unmarshal(onQueryData.Query, &query)
+	if err != nil {
+		return nil, err
+	}
+
+	// Check that the job ID matches.
+	if query.JobID != r.jobid {
+		return nil, errNoJobMatch
+	}
+
+	var requestParams map[string]interface{}
+	if query.Params == nil {
+		requestParams = make(map[string]interface{})
+	} else {
+		requestParams = query.Params
+	}
+	requestParams["request_id"] = onQueryData.QueryID
+	requestParams["payment"] = fmt.Sprint(onQueryData.Fee) +
+		strings.Repeat("0", linkDecimals-linkAgoricDecimals)
+
+	return requestParams, nil
+}

blockchain/agoric/runlog_test.go

@@ -0,0 +1,81 @@
+package agoric
+
+import (
+	"reflect"
+	"testing"
+
+	"github.com/smartcontractkit/external-initiator/blockchain/common"
+)
+
+func TestRunlogManager_ParseRequests(t *testing.T) {
+	type fields struct {
+		jobid string
+	}
+	type args struct {
+		data []byte
+	}
+	tests := []struct {
+		name    string
+		fields  fields
+		args    args
+		want    common.RunlogRequest
+		wantErr bool
+	}{
+		{
+			"fails parsing invalid payload",
+			fields{},
+			args{data: []byte(`invalid`)},
+			nil,
+			true,
+		},
+		{
+			"fails parsing invalid WS body",
+			fields{},
+			args{data: []byte(`{}`)},
+			nil,
+			true,
+		},
+		{
+			"fails parsing invalid WS type",
+			fields{},
+			args{data: []byte(`{"type":"oracleServer/wrongType"}`)},
+			nil,
+			true,
+		},
+		{
+			"successfully parses WS Oracle request",
+			fields{jobid: "9999"},
+			args{data: []byte(`{"type":"oracleServer/onQuery","data":{"query":{"jobID":"9999","params":{"path":"foo"}},"queryId":"123","fee":191919}}`)},
+			common.RunlogRequest{
+				"path":       "foo",
+				"payment":    "191919000000000000",
+				"request_id": "123",
+			},
+			false,
+		},
+		{
+			"skips unfiltered WS Oracle request",
+			fields{jobid: "Z9999"},
+			args{data: []byte(`{"type":"oracleServer/onQuery","data":{"query":{"jobID":"9999","params":{"path":"foo"}},"queryId":"123","fee":191919}}`)},
+			nil,
+			true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			e := runlogManager{
+				manager: &manager{
+					jobid: tt.fields.jobid,
+				},
+			}
+			got, err := e.parseRequests(tt.args.data)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("parseRequests() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("parseRequests() got = %s, want %s", got, tt.want)
+			}
+		})
+	}
+}