[SM-548] add instant offer request

DIMO-Network · Feb 26, 2024 · cb9fc93 · cb9fc93
1 parent e279783
commit cb9fc93
Show file tree

Hide file tree

Showing 12 changed files with 72 additions and 66 deletions.
diff --git a/charts/devices-api/values-prod.yaml b/charts/devices-api/values-prod.yaml
@@ -42,6 +42,11 @@ env:
   AFTERMARKET_DEVICE_CONTRACT_ADDRESS: '0x9c94c395cbcbde662235e0a9d3bb87ad708561ba'
   DIMO_CONTRACT_APIURL: https://contracts-api.dimo.zone/
   NATS_URL: nats-prod:4222
+  NATS_STREAM_NAME: DD_VALUATION_TASKS
+  NATS_VALUATION_SUBJECT: dd_valuation_tasks
+  NATS_OFFER_SUBJECT: dd_offer_tasks
+  NATS_DURABLE_CONSUMER: dd-valuation-task-consumer
+  NATS_ACK_TIMEOUT: 2m
   SYNTHETIC_DEVICES_ENABLED: true
   SYNTHETIC_WALLET_GRPC_ADDR: synthetic-wallet-instance-prod:8086
   DEVICE_DATA_GRPC_ADDR: device-data-api-prod:8086

diff --git a/charts/devices-api/values.yaml b/charts/devices-api/values.yaml
@@ -83,6 +83,7 @@ env:
   NATS_URL: nats-dev:4222
   NATS_STREAM_NAME: DD_VALUATION_TASKS
   NATS_VALUATION_SUBJECT: dd_valuation_tasks
+  NATS_OFFER_SUBJECT: dd_offer_tasks
   NATS_DURABLE_CONSUMER: dd-valuation-task-consumer
   NATS_ACK_TIMEOUT: 2m
   SYNTHETIC_DEVICES_ENABLED: true

diff --git a/cmd/devices-api/api.go b/cmd/devices-api/api.go
@@ -3,7 +3,6 @@ package main
 import (
 	"context"
 	"encoding/base64"
-	"github.com/DIMO-Network/devices-api/internal/services/fingerprint"
 	"math/big"
 	"net"
 	"os"
@@ -12,6 +11,8 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/DIMO-Network/devices-api/internal/services/fingerprint"
+
 	"github.com/DIMO-Network/devices-api/internal/middleware"
 
 	"github.com/DIMO-Network/devices-api/internal/rpc"

diff --git a/internal/config/settings.go b/internal/config/settings.go
@@ -71,6 +71,7 @@ type Settings struct {
 	NATSURL                           string `yaml:"NATS_URL"`
 	NATSStreamName                    string `yaml:"NATS_STREAM_NAME"`
 	NATSValuationSubject              string `yaml:"NATS_VALUATION_SUBJECT"`
+	NATSOfferSubject                  string `yaml:"NATS_OFFER_SUBJECT"`
 	NATSAckTimeout                    string `yaml:"NATS_ACK_TIMEOUT"`
 	NATSDurableConsumer               string `yaml:"NATS_DURABLE_CONSUMER"`
 	ValuationsAPIGRPCAddr             string `yaml:"VALUATIONS_GRPC_ADDR"`

diff --git a/internal/controllers/user_devices_controller.go b/internal/controllers/user_devices_controller.go
@@ -616,29 +616,56 @@ func (udc *UserDevicesController) RegisterDeviceForUserFromVIN(c *fiber.Ctx) err
 
 	// request valuation
 	if udc.Settings.IsProduction() {
-		message := services.ValuationDecodeCommand{
-			VIN:          vin,
-			UserDeviceID: udFull.ID,
-		}
-		messageBytes, err := json.Marshal(message)
-
-		if err != nil {
-			localLog.Err(err).Msg("Failed to marshal message.")
-		} else {
-			pubAck, err := udc.NATSSvc.JetStream.Publish(udc.NATSSvc.JetStreamSubject, messageBytes)
-			if err != nil {
-				localLog.Err(err).Msg("failed to publish to NATS")
-			} else {
-				localLog.Info().Str("user_device_id", udFull.ID).Msgf("published valuation request to NATS with Ack: %+v", pubAck)
-			}
-		}
+		udc.requestValuation(vin, udFull.ID)
+		udc.requestInstantOffer(udFull.ID)
 	}
 
 	return c.Status(fiber.StatusCreated).JSON(fiber.Map{
 		"userDevice": udFull,
 	})
 }
 
+func (udc *UserDevicesController) requestValuation(vin string, userDeviceID string) {
+	message := services.ValuationDecodeCommand{
+		VIN:          vin,
+		UserDeviceID: userDeviceID,
+	}
+	messageBytes, err := json.Marshal(message)
+
+	if err != nil {
+		udc.log.Err(err).Msg("Failed to marshal message.")
+		return
+	}
+
+	pubAck, err := udc.NATSSvc.JetStream.Publish(udc.NATSSvc.ValuationSubject, messageBytes)
+	if err != nil {
+		udc.log.Err(err).Msg("failed to publish to NATS")
+		return
+	}
+
+	udc.log.Info().Str("user_device_id", userDeviceID).Msgf("published valuation request to NATS with Ack: %+v", pubAck)
+}
+
+func (udc *UserDevicesController) requestInstantOffer(userDeviceID string) {
+	message := services.OfferRequest{
+		UserDeviceID: userDeviceID,
+	}
+	messageBytes, err := json.Marshal(message)
+
+	if err != nil {
+		udc.log.Err(err).Msg("Failed to marshal message.")
+		return
+	}
+
+	pubAck, err := udc.NATSSvc.JetStream.Publish(udc.NATSSvc.OfferSubject, messageBytes)
+	if err != nil {
+		udc.log.Err(err).Msg("failed to publish to NATS")
+		return
+	}
+
+	udc.log.Info().Str("user_device_id", userDeviceID).Msgf("published instant offer request to NATS with Ack: %+v", pubAck)
+}
+
 // RegisterDeviceForUserFromSmartcar godoc
 // @Description adds a device to a user by decoding VIN from Smartcar. If cannot decode returns 424 or 500 if error.
 // @Description If the user device already exists from a different integration, for the same user, this will return a 200 with the full user device object
@@ -786,26 +813,7 @@ func (udc *UserDevicesController) RegisterDeviceForUserFromSmartcar(c *fiber.Ctx
 	}
 
 	if udc.Settings.IsProduction() {
-
-		message := services.ValuationDecodeCommand{
-			VIN:          vin,
-			UserDeviceID: udFull.ID,
-		}
-
-		messageBytes, err := json.Marshal(message)
-
-		if err != nil {
-			localLog.Err(err).Msg("Failed to marshal message.")
-		} else {
-			pubAck, err := udc.NATSSvc.JetStream.Publish(udc.NATSSvc.JetStreamSubject, messageBytes)
-
-			if err != nil {
-				localLog.Err(err).Msg("Failed to publish to NATS.")
-			} else {
-				localLog.Info().Str("vin", vin).Str("user_id", userID).Str("user_device_id", udFull.ID).Msgf("Published valuation request to NATS with Ack: %+v", pubAck)
-			}
-		}
-
+		udc.requestValuation(vin, udFull.ID)
 	}
 
 	return c.Status(fiber.StatusCreated).JSON(fiber.Map{

diff --git a/internal/controllers/user_devices_controller_test.go b/internal/controllers/user_devices_controller_test.go
@@ -373,7 +373,7 @@ func (s *UserDevicesControllerTestSuite) TestPostUserDeviceFromVIN() {
 	assert.Equal(s.T(), "6", *regUserResp.Metadata.CANProtocol)
 	assert.EqualValues(s.T(), "ICE", *regUserResp.Metadata.PowertrainType)
 
-	msg, responseError := s.natsService.JetStream.GetLastMsg(natsStreamName, s.natsService.JetStreamSubject)
+	msg, responseError := s.natsService.JetStream.GetLastMsg(natsStreamName, s.natsService.ValuationSubject)
 	assert.NoError(s.T(), responseError, "expected no error from nats")
 	vinResult := gjson.GetBytes(msg.Data, "vin")
 	assert.Equal(s.T(), vinny, vinResult.Str)
@@ -447,7 +447,7 @@ func (s *UserDevicesControllerTestSuite) TestPostUserDeviceFromVIN_SameUser_Dupl
 	assert.Equal(s.T(), integration.Type, regUserResp.DeviceDefinition.CompatibleIntegrations[0].Type)
 	assert.Equal(s.T(), integration.Id, regUserResp.DeviceDefinition.CompatibleIntegrations[0].ID)
 
-	msg, responseError := s.natsService.JetStream.GetLastMsg(natsStreamName, s.natsService.JetStreamSubject)
+	msg, responseError := s.natsService.JetStream.GetLastMsg(natsStreamName, s.natsService.ValuationSubject)
 	assert.NoError(s.T(), responseError, "expected no error from nats")
 	vinResult := gjson.GetBytes(msg.Data, "vin")
 	assert.Equal(s.T(), vinny, vinResult.Str)

diff --git a/internal/controllers/user_integrations_auth_controller.go b/internal/controllers/user_integrations_auth_controller.go
@@ -4,6 +4,9 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"strconv"
+	"time"
+
 	"github.com/DIMO-Network/devices-api/internal/config"
 	"github.com/DIMO-Network/devices-api/internal/constants"
 	"github.com/DIMO-Network/devices-api/internal/controllers/helpers"
@@ -14,8 +17,6 @@ import (
 	"github.com/DIMO-Network/shared/redis"
 	"github.com/gofiber/fiber/v2"
 	"github.com/rs/zerolog"
-	"strconv"
-	"time"
 )
 
 type UserIntegrationAuthController struct {

diff --git a/internal/controllers/user_integrations_auth_controller_test.go b/internal/controllers/user_integrations_auth_controller_test.go
@@ -4,6 +4,10 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"io"
+	"testing"
+	"time"
+
 	ddgrpc "github.com/DIMO-Network/device-definitions-api/pkg/grpc"
 	"github.com/DIMO-Network/devices-api/internal/config"
 	"github.com/DIMO-Network/devices-api/internal/constants"
@@ -20,9 +24,6 @@ import (
 	"github.com/stretchr/testify/suite"
 	"github.com/testcontainers/testcontainers-go"
 	"go.uber.org/mock/gomock"
-	"io"
-	"testing"
-	"time"
 )
 
 type UserIntegrationAuthControllerTestSuite struct {

diff --git a/internal/controllers/user_integrations_controller.go b/internal/controllers/user_integrations_controller.go
@@ -1899,25 +1899,7 @@ func (udc *UserDevicesController) registerDeviceTesla(c *fiber.Ctx, logger *zero
 	}
 
 	if udc.Settings.IsProduction() {
-		message := services.ValuationDecodeCommand{
-			VIN:          v.VIN,
-			UserDeviceID: userDeviceID,
-		}
-
-		messageBytes, err := json.Marshal(message)
-
-		if err != nil {
-			udc.log.Err(err).Msg("Failed to marshal valuation decode command.")
-		} else {
-			pubAck, err := udc.NATSSvc.JetStream.Publish(udc.NATSSvc.JetStreamSubject, messageBytes)
-
-			if err != nil {
-				udc.log.Err(err).Msg("Failed to publish valuation decode command for Tesla Device.")
-			} else {
-				udc.log.Info().Str("vin", v.VIN).Msgf("Published valuation decode command with sequence %d.", pubAck.Sequence)
-			}
-		}
-
+		udc.requestValuation(v.VIN, userDeviceID)
 	}
 
 	if err := udc.teslaTaskService.StartPoll(v, &integration); err != nil {

diff --git a/internal/services/mocks/nats_service_mock.go b/internal/services/mocks/nats_service_mock.go
@@ -76,7 +76,7 @@ func NewMockNATSService(streamName string) (*services.NATSService, *server.Serve
 	natsSvc := &services.NATSService{
 		JetStream:        js,
 		JetStreamName:    streamName,
-		JetStreamSubject: "test-subject",
+		ValuationSubject: "test-subject",
 		AckTimeout:       to,
 		DurableConsumer:  "test-durable-consumer",
 	}

diff --git a/internal/services/models.go b/internal/services/models.go
@@ -186,3 +186,7 @@ type ValuationDecodeCommand struct {
 	VIN          string `json:"vin"`
 	UserDeviceID string `json:"userDeviceId"`
 }
+
+type OfferRequest struct {
+	UserDeviceID string `json:"user_device_id"`
+}
diff --git a/internal/services/nats_service.go b/internal/services/nats_service.go
@@ -12,7 +12,8 @@ type NATSService struct {
 	log              *zerolog.Logger
 	JetStream        nats.JetStreamContext
 	JetStreamName    string
-	JetStreamSubject string
+	ValuationSubject string
+	OfferSubject     string
 	AckTimeout       time.Duration
 	DurableConsumer  string
 }
@@ -37,7 +38,8 @@ func NewNATSService(settings *config.Settings, log *zerolog.Logger) (*NATSServic
 		log:              log,
 		JetStream:        js,
 		JetStreamName:    settings.NATSStreamName,
-		JetStreamSubject: settings.NATSValuationSubject,
+		ValuationSubject: settings.NATSValuationSubject,
+		OfferSubject:     settings.NATSOfferSubject,
 		AckTimeout:       to,
 		DurableConsumer:  settings.NATSDurableConsumer}