Add kv store implementation

core/services/job/kv_orm.go

@@ -0,0 +1,68 @@
+package job
+
+import (
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"github.com/jmoiron/sqlx"
+	"github.com/jmoiron/sqlx/types"
+
+	"github.com/smartcontractkit/chainlink/v2/core/logger"
+	"github.com/smartcontractkit/chainlink/v2/core/services/pg"
+)
+
+//go:generate mockery --quiet --name KVStore --output ./mocks/ --case=underscore
+
+type KVStore interface {
+	Store(key string, val interface{}) error
+	Get(key string, dest interface{}) error
+}
+
+type jobKVStore struct {
+	jobID int32
+	q     pg.Q
+	lggr  logger.SugaredLogger
+	cfg   pg.QConfig
+}
+
+var _ KVStore = (*jobKVStore)(nil)
+
+func NewJobKVStore(jobID int32, db *sqlx.DB, cfg pg.QConfig, lggr logger.Logger) KVStore {
+	namedLogger := logger.Sugared(lggr.Named("JobORM"))
+	return &jobKVStore{
+		jobID: jobID,
+		q:     pg.NewQ(db, namedLogger, cfg),
+		lggr:  namedLogger,
+	}
+}
+
+func (kv jobKVStore) Store(key string, val interface{}) error {
+	jsonVal, err := json.Marshal(val)
+	if err != nil {
+		return err
+	}
+
+	sql := `INSERT INTO job_kv_store (id, key, val)
+       	 	VALUES ($1, $2, $3)
+        	ON CONFLICT (id, key) DO UPDATE SET
+				val = EXCLUDED.val,
+				updated_at = $4
+        	RETURNING id;`
+
+	if err = kv.q.ExecQ(sql, kv.jobID, key, types.JSONText(jsonVal), time.Now()); err != nil {
+		return fmt.Errorf("failed to store value: %s for key: %s for jobID: %d : %w", string(jsonVal), key, kv.jobID, err)
+	}
+	return nil
+}
+
+func (kv jobKVStore) Get(key string, dest interface{}) error {
+	var ret types.JSONText
+
+	sql := "SELECT val FROM job_kv_store WHERE id = $1 AND key = $2"
+	if err := kv.q.Get(&ret, sql, kv.jobID, key); err != nil {
+		return fmt.Errorf("failed to get value by key: %s for jobID: %d : %w", key, kv.jobID, err)
+	}
+
+	return ret.Unmarshal(dest)
+}

core/services/job/kv_orm_test.go

@@ -0,0 +1,56 @@
+package job_test
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/smartcontractkit/chainlink/v2/core/bridges"
+	"github.com/smartcontractkit/chainlink/v2/core/internal/cltest"
+	"github.com/smartcontractkit/chainlink/v2/core/internal/testutils/configtest"
+	"github.com/smartcontractkit/chainlink/v2/core/internal/testutils/pgtest"
+	"github.com/smartcontractkit/chainlink/v2/core/logger"
+	"github.com/smartcontractkit/chainlink/v2/core/services/directrequest"
+	"github.com/smartcontractkit/chainlink/v2/core/services/job"
+	"github.com/smartcontractkit/chainlink/v2/core/services/pipeline"
+	"github.com/smartcontractkit/chainlink/v2/core/testdata/testspecs"
+)
+
+func TestJobKVStore_Store(t *testing.T) {
+	config := configtest.NewTestGeneralConfig(t)
+	db := pgtest.NewSqlxDB(t)
+
+	lggr := logger.TestLogger(t)
+
+	pipelineORM := pipeline.NewORM(db, logger.TestLogger(t), config.Database(), config.JobPipeline().MaxSuccessfulRuns())
+	bridgesORM := bridges.NewORM(db, logger.TestLogger(t), config.Database())
+
+	jobID := int32(1337)
+	kvStore := job.NewJobKVStore(jobID, db, config.Database(), lggr)
+	jobORM := NewTestORM(t, db, pipelineORM, bridgesORM, cltest.NewKeyStore(t, db, config.Database()), config.Database())
+
+	jb, err := directrequest.ValidatedDirectRequestSpec(testspecs.GetDirectRequestSpec())
+	require.NoError(t, err)
+	jb.ID = jobID
+	require.NoError(t, jobORM.CreateJob(&jb))
+
+	key := "test_key"
+
+	type testData struct {
+		Test string `json:"test"`
+	}
+
+	td1 := testData{Test: "value1"}
+	td2 := testData{Test: "value2"}
+
+	var retData testData
+	require.NoError(t, kvStore.Store(key, td1))
+	require.NoError(t, kvStore.Get(key, &retData))
+	require.Equal(t, td1, retData)
+
+	require.NoError(t, kvStore.Store(key, td2))
+	require.NoError(t, kvStore.Get(key, &retData))
+	require.Equal(t, td2, retData)
+
+	t.Fail()
+}