feat: tx db client

.github/workflows/lint-test.yml

@@ -18,7 +18,7 @@ jobs:
       - name: Set up Go
         uses: actions/setup-go@v2
         with:
-          go-version: 1.18
+          go-version: '1.20'
 
       - name: Check out code
         uses: actions/checkout@v2
@@ -33,7 +33,7 @@ jobs:
       - name: Set up Go
         uses: actions/setup-go@v2
         with:
-          go-version: 1.18
+          go-version: '1.20'
 
       - name: Check out code
         uses: actions/checkout@v2

docker-compose.test.yml

@@ -9,17 +9,33 @@ services:
     ports:
       - 8080:8080
     environment:
-      CONNECTION_STRING: postgres://postgres:pgpassword@transaction-db:5432/?sslmode=disable # pragma: allowlist secret
+      CONNECTION_STRING: postgres://postgres:pgpassword@transaction-db-http:5432/?sslmode=disable # pragma: allowlist secret
       API_KEYS: test_api_key
       MAX_RELAY_BATCH_SIZE: 1
       MAX_SERVICE_RECORD_BATCH_SIZE: 1
     depends_on:
       transaction-db:
         condition: service_healthy
 
+  # DB Used for http client tests
+  transaction-db-http:
+    image: pocketfoundation/test-transaction-postgres:latest
+    container_name: transaction-db-http
+    restart: always
+    ports:
+      - 5433:5432
+    environment:
+      POSTGRES_PASSWORD: pgpassword
+      POSTGRES_DB: postgres
+    healthcheck:
+      test: pg_isready -U postgres
+      interval: 5s
+      retries: 3
+
+  # DB Used for db client tests
   transaction-db:
     image: pocketfoundation/test-transaction-postgres:latest
-    container_name: transaction-http-db
+    container_name: transaction-db
     restart: always
     ports:
       - 5432:5432

go.mod

@@ -6,15 +6,18 @@ require (
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/gojek/valkyrie v0.0.0-20190210220504-8f62c1e7ba45 // indirect
 	github.com/kr/text v0.2.0 // indirect
+	github.com/lib/pq v1.10.9 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/stretchr/objx v0.5.0 // indirect
+	golang.org/x/crypto v0.9.0 // indirect
+	golang.org/x/sys v0.8.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
 
 require (
 	github.com/gojek/heimdall v5.0.2+incompatible
 	github.com/gojek/heimdall/v7 v7.0.2
-	github.com/pokt-foundation/transaction-db v1.15.2
+	github.com/pokt-foundation/transaction-db v1.17.0
 	github.com/stretchr/testify v1.8.4
 )

go.sum

@@ -17,12 +17,14 @@ github.com/jtolds/gls v4.20.0+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfV
 github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
+github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw=
+github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/pokt-foundation/transaction-db v1.15.2 h1:J/L7Km7BJgk4idtLSSvgK7eQkVYv98gGxTyG+T4U3SM=
-github.com/pokt-foundation/transaction-db v1.15.2/go.mod h1:bAoDJ1/VVT2XBX7uHCvtOfkP1W4EcLHbitlzGB0c4cU=
+github.com/pokt-foundation/transaction-db v1.17.0 h1:tlBKyIlVUfnMmh2xGG0tdoiyMwSgH4zy1vrMDGD1kHY=
+github.com/pokt-foundation/transaction-db v1.17.0/go.mod h1:8fFJ/zVEqtlQFKE5nzoDBCNg5T/4Ek+4gWnkPLNU2J8=
 github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4=
 github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d/go.mod h1:OnSkiWE9lh6wB0YB77sQom3nweQdgAjqCqsofrRNTgc=
 github.com/smartystreets/goconvey v1.6.4/go.mod h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9QV7WQ/tjFTllLA=
@@ -37,8 +39,13 @@ github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO
 github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk=
 github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/crypto v0.9.0 h1:LF6fAI+IutBocDJ2OT0Q1g8plpYljMZ4+lty+dsqw3g=
+golang.org/x/crypto v0.9.0/go.mod h1:yrmDGqONDYtNj3tH8X9dzUun2m2lzPa9ngI6/RUPGR0=
 golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.8.0 h1:EBmGv8NaZBZTWvrbjNoL6HVt+IVy3QDQpJs7VRIw3tU=
+golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/term v0.8.0 h1:n5xxQn2i3PC0yLAbjTpNT85q/Kgzcr2gIoX9OrJUols=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/tools v0.0.0-20190328211700-ab21143f2384/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

txclient/db_client.go

@@ -0,0 +1,129 @@
+package txclient
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	postgresdriver "github.com/pokt-foundation/transaction-db/postgres-driver"
+	"github.com/pokt-foundation/transaction-db/types"
+)
+
+type DBClient struct {
+	driver  *postgresdriver.PostgresDriver
+	Timeout time.Duration
+}
+
+func NewDBClient(connectionString string, timeout time.Duration) (*DBClient, error) {
+	driver, err := postgresdriver.NewPostgresDriver(connectionString)
+	if err != nil {
+		return nil, err
+	}
+
+	return &DBClient{
+		driver:  driver,
+		Timeout: timeout,
+	}, nil
+}
+
+func (db DBClient) CreateSession(session types.PocketSession) error {
+	if err := session.Validate(); err != nil {
+		return err
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), db.Timeout)
+	defer cancel()
+
+	return db.driver.WriteSession(ctx, session)
+}
+
+func (db DBClient) CreateRegion(region types.PortalRegion) error {
+	ctx, cancel := context.WithTimeout(context.Background(), db.Timeout)
+	defer cancel()
+
+	return db.driver.WriteRegion(ctx, region)
+}
+
+func (db DBClient) CreateRelay(relay types.Relay) error {
+	if err := relay.Validate(); err != nil {
+		return err
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), db.Timeout)
+	defer cancel()
+
+	return db.driver.WriteRelay(ctx, relay)
+}
+
+func (db DBClient) CreateRelays(relays []types.Relay) error {
+	var relayErrors []error
+	var validRelays []*types.Relay
+
+	for _, relay := range relays {
+		if err := relay.Validate(); err != nil {
+			relayErrors = append(relayErrors, err)
+		}
+		validRelays = append(validRelays, &relay)
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), db.Timeout)
+	defer cancel()
+
+	err := db.driver.WriteRelays(ctx, validRelays)
+
+	if relayErrors != nil || err != nil {
+		relayErrors = append(relayErrors, err)
+		return errors.Join(relayErrors...)
+	}
+
+	return nil
+}
+
+func (db DBClient) GetRelay(id int) (types.Relay, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), db.Timeout)
+	defer cancel()
+
+	return db.driver.ReadRelay(ctx, id)
+}
+
+func (db DBClient) CreateServiceRecord(record types.ServiceRecord) error {
+	if err := record.Validate(); err != nil {
+		return err
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), db.Timeout)
+	defer cancel()
+
+	return db.driver.WriteServiceRecord(ctx, record)
+}
+
+func (db DBClient) CreateServiceRecords(records []types.ServiceRecord) error {
+	var recordrrors []error
+	var validRecords []*types.ServiceRecord
+
+	for _, relay := range records {
+		if err := relay.Validate(); err != nil {
+			recordrrors = append(recordrrors, err)
+		}
+		validRecords = append(validRecords, &relay)
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), db.Timeout)
+	defer cancel()
+
+	err := db.driver.WriteServiceRecords(ctx, validRecords)
+
+	if recordrrors != nil || err != nil {
+		recordrrors = append(recordrrors, err)
+		return errors.Join(recordrrors...)
+	}
+
+	return nil
+}
+
+func (db DBClient) GetServiceRecord(id int) (types.ServiceRecord, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), db.Timeout)
+	defer cancel()
+
+	return db.driver.ReadServiceRecord(ctx, id)
+}

txclient/client.go → txclient/http_client.go

@@ -51,32 +51,32 @@ type TxClientRW interface {
 	TxClientWrite
 }
 
-var _ TxClientRW = TXClient{}
+var _ TxClientRW = HttpClient{}
 
-type TXClient struct {
+type HttpClient struct {
 	httpClient *httpclient.Client
 	config     Config
 	headers    http.Header
 }
 
-func NewTXClient(config Config) (*TXClient, error) {
+func NewHttpClient(config Config) (*HttpClient, error) {
 	if err := config.validate(); err != nil {
-		return &TXClient{}, err
+		return &HttpClient{}, err
 	}
 
-	return &TXClient{
+	return &HttpClient{
 		httpClient: newHTTPClient(config),
 		config:     config,
 		headers:    http.Header{"Authorization": {config.APIKey}, "Content-Type": {"application/json"}},
 	}, nil
 }
 
 // versionedBasePath returns the base path for a given data type eg. `https://pocket.http-db-url.com/v1/application`
-func (db TXClient) versionedBasePath(dataTypePath typePath) string {
+func (db HttpClient) versionedBasePath(dataTypePath typePath) string {
 	return fmt.Sprintf("%s/%s/%s", db.config.BaseURL, db.config.Version, dataTypePath)
 }
 
-func (db TXClient) CreateSession(session types.PocketSession) error {
+func (db HttpClient) CreateSession(session types.PocketSession) error {
 	body, err := json.Marshal(session)
 	if err != nil {
 		return err
@@ -86,7 +86,7 @@ func (db TXClient) CreateSession(session types.PocketSession) error {
 	return err
 }
 
-func (db TXClient) CreateRegion(region types.PortalRegion) error {
+func (db HttpClient) CreateRegion(region types.PortalRegion) error {
 	body, err := json.Marshal(region)
 	if err != nil {
 		return err
@@ -96,7 +96,7 @@ func (db TXClient) CreateRegion(region types.PortalRegion) error {
 	return err
 }
 
-func (db TXClient) CreateRelay(relay types.Relay) error {
+func (db HttpClient) CreateRelay(relay types.Relay) error {
 	if err := relay.Validate(); err != nil {
 		return err
 	}
@@ -110,7 +110,7 @@ func (db TXClient) CreateRelay(relay types.Relay) error {
 	return err
 }
 
-func (db TXClient) CreateRelays(relays []types.Relay) error {
+func (db HttpClient) CreateRelays(relays []types.Relay) error {
 	for _, relay := range relays {
 		if err := relay.Validate(); err != nil {
 			return err
@@ -126,7 +126,7 @@ func (db TXClient) CreateRelays(relays []types.Relay) error {
 	return err
 }
 
-func (db TXClient) CreateServiceRecord(sr types.ServiceRecord) error {
+func (db HttpClient) CreateServiceRecord(sr types.ServiceRecord) error {
 	if err := sr.Validate(); err != nil {
 		return err
 	}
@@ -140,7 +140,7 @@ func (db TXClient) CreateServiceRecord(sr types.ServiceRecord) error {
 	return err
 }
 
-func (db TXClient) CreateServiceRecords(srs []types.ServiceRecord) error {
+func (db HttpClient) CreateServiceRecords(srs []types.ServiceRecord) error {
 	for _, record := range srs {
 		if err := record.Validate(); err != nil {
 			return err
@@ -156,12 +156,12 @@ func (db TXClient) CreateServiceRecords(srs []types.ServiceRecord) error {
 	return err
 }
 
-func (db TXClient) GetRelay(id int) (types.Relay, error) {
+func (db HttpClient) GetRelay(id int) (types.Relay, error) {
 	path := fmt.Sprintf("%s/%d", db.versionedBasePath(relayPath), id)
 	return performHttpReq[types.Relay](http.MethodGet, path, db.headers, nil, db.httpClient)
 }
 
-func (db TXClient) GetServiceRecord(id int) (types.ServiceRecord, error) {
+func (db HttpClient) GetServiceRecord(id int) (types.ServiceRecord, error) {
 	path := fmt.Sprintf("%s/%d", db.versionedBasePath(serviceRecordPath), id)
 	fmt.Println("path", path)
 	return performHttpReq[types.ServiceRecord](http.MethodGet, path, db.headers, nil, db.httpClient)

txclient/suite_test.go

@@ -94,7 +94,7 @@ func (ts *txClientTestSuite) SetupSuite() {
 }
 
 func (ts *txClientTestSuite) initClient() error {
-	client, err := NewTXClient(Config{
+	client, err := NewHttpClient(Config{
 		BaseURL: "http://localhost:8080",
 		APIKey:  "test_api_key",
 		Version: V0,