Removed unbufferedMsg references from motorq example

go.mod

@@ -36,7 +36,6 @@ require (
 	github.com/volatiletech/null/v8 v8.1.2
 	github.com/volatiletech/sqlboiler/v4 v4.18.0
 	github.com/volatiletech/strmangle v0.0.8
-	go.uber.org/mock v0.5.0
 	golang.org/x/sync v0.13.0
 	golang.org/x/time v0.11.0
 	gotest.tools/v3 v3.5.2

go.sum

@@ -891,8 +891,6 @@ go.opentelemetry.io/proto/otlp v1.5.0/go.mod h1:keN8WnHxOy8PG0rQZjJJ5A2ebUoafqWp
 go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
 go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
 go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
-go.uber.org/mock v0.5.0 h1:KAMbZvZPyBPWgD14IrIQ38QCyjwpvVVV6K/bHl1IwQU=
-go.uber.org/mock v0.5.0/go.mod h1:ge71pBPLYDk7QIi1LupWxdAykm7KIEFchiOqd6z7qMM=
 go.uber.org/multierr v1.6.0/go.mod h1:cdWPpRnG4AhwMwsgIHip0KRBQjJy5kYEpYjJxpXp9iU=
 go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
 go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=

internal/config/settings.go

@@ -35,11 +35,10 @@ type Settings struct {
 	UnbufferedTelemetryConsumerGroup string `yaml:"UNBUFFERED_TELEMETRY_CONSUMER_GROUP"`
 
 	// DIS - DIMO Ingest Service
-	ConvertToCloudEvent bool   `yaml:"CONVERT_TO_CLOUD_EVENT"` // if true, telemetry is converted to cloud events before sending to DIS
-	DimoNodeEndpoint    string `yaml:"DIMO_NODE_ENDPOINT"`
-	Cert                string `yaml:"CERT"`     // should be secrets
-	CertKey             string `yaml:"CERT_KEY"` // should be secrets
-	CACert              string `yaml:"CA_CERT"`  // DIMO Root CA, same for everybody
+	DimoNodeEndpoint string `yaml:"DIMO_NODE_ENDPOINT"`
+	Cert             string `yaml:"CERT"`     // should be secrets
+	CertKey          string `yaml:"CERT_KEY"` // should be secrets
+	CACert           string `yaml:"CA_CERT"`  // DIMO Root CA, same for everybody
 
 	// Chain - These are standard Polygon values for DIMO
 	ChainID             int64          `yaml:"CHAIN_ID"`

internal/convert/convert.go

@@ -2,17 +2,18 @@ package convert
 
 import (
 	"encoding/json"
+	"fmt"
 	"github.com/DIMO-Network/cloudevent"
 	"github.com/DIMO-Network/model-garage/pkg/defaultmodule"
 	"github.com/DIMO-Network/oracle-example/internal/config"
 	dbmodels "github.com/DIMO-Network/oracle-example/internal/db/models"
 	"github.com/DIMO-Network/oracle-example/internal/models"
-	"github.com/segmentio/ksuid"
+	"github.com/rs/zerolog"
 	"time"
 )
 
-// ToCloudEvent Convert the external vendor msg payload to a CloudEvent
-func ToCloudEvent(veh dbmodels.Vin, msg models.UnbufferedMessageValue, settings config.Settings) (*cloudevent.CloudEvent[json.RawMessage], error) {
+// SetProducerAndSubject sets the producer and consumer fields in the CloudEvent
+func SetProducerAndSubject(veh dbmodels.Vin, ce *cloudevent.CloudEvent[json.RawMessage], settings config.Settings) error {
 	// Construct the producer DID
 	producer := cloudevent.NFTDID{
 		ChainID:         uint64(settings.ChainID),
@@ -31,44 +32,52 @@ func ToCloudEvent(veh dbmodels.Vin, msg models.UnbufferedMessageValue, settings
 		}.String()
 	}
 
-	ch, err := createCloudEventHeader(msg.Timestamp, producer, subject, cloudevent.TypeStatus)
-	if err != nil {
-		return nil, err
-	}
+	ce.Subject = subject
+	ce.Producer = producer
 
-	// transform the data to default DIS format
-	signals, err := mapDataToSignals(msg.Data, msg.Timestamp)
+	return nil
+}
 
-	if err != nil {
-		return nil, err
+// ValidateSignals validates the signals from the message
+func ValidateSignals(signals interface{}, logger zerolog.Logger) error {
+
+	signalsArr, ok := signals.([]interface{})
+	if !ok {
+		err := fmt.Errorf("signals is not of type []interface{}")
+		logger.Error().Err(err).Msg("Invalid type for signals")
+		return err
 	}
 
-	// Wrap the signals into a struct
-	wrappedData := struct {
-		Signals []*defaultmodule.Signal `json:"signals"`
-		Vin     string                  `json:"vin"`
-	}{
-		Signals: signals,
-		Vin:     veh.Vin,
+	signalArray, err := CastToSliceOfMaps(signalsArr)
+	if err != nil {
+		logger.Error().Err(err).Msg("Failed to cast signals to slice of maps")
+		return err
 	}
 
-	// Marshal the wrapped data to json.RawMessage
-	data, err := json.Marshal(wrappedData)
+	// Load the signal map
+	sigMap, err := defaultmodule.LoadSignalMap()
 	if err != nil {
-		return nil, err
+		return err
 	}
 
-	// Create the CloudEvent
-	ce := &cloudevent.CloudEvent[json.RawMessage]{
-		CloudEventHeader: ch,
-		Data:             data,
+	for _, signal := range signalArray {
+		name, ok := signal["name"].(string)
+		if !ok {
+			logger.Warn().Msgf("Signal name is missing or not a string: %v\n", signal)
+			continue
+		}
+
+		if _, exists := sigMap[name]; !exists {
+			logger.Warn().Msgf("Signal %s is not in the signal map\n", name)
+			continue
+		}
 	}
 
-	return ce, nil
+	return nil
 }
 
-// mapDataToSignals maps the data from the message to the default DIS signals
-func mapDataToSignals(data models.Data, ts time.Time) ([]*defaultmodule.Signal, error) {
+// MapDataToSignals maps the data from the message to the default DIS signals
+func MapDataToSignals(data models.Data, ts time.Time) ([]*defaultmodule.Signal, error) {
 	var signals []*defaultmodule.Signal
 
 	sigMap, err := defaultmodule.LoadSignalMap()
@@ -124,16 +133,16 @@ func mapDataToSignals(data models.Data, ts time.Time) ([]*defaultmodule.Signal,
 	return signals, nil
 }
 
-// createCloudEvent creates a cloud event from autopi event.
-func createCloudEventHeader(ts time.Time, producer, subject, eventType string) (cloudevent.CloudEventHeader, error) {
-	return cloudevent.CloudEventHeader{
-		DataContentType: "application/json",
-		ID:              ksuid.New().String(),
-		Subject:         subject,
-		SpecVersion:     "1.0",
-		Time:            ts,
-		Type:            eventType,
-		DataVersion:     "default/v1.0",
-		Producer:        producer,
-	}, nil
+func CastToSliceOfMaps(signals []interface{}) ([]map[string]interface{}, error) {
+	var result []map[string]interface{}
+
+	for _, item := range signals {
+		castItem, ok := item.(map[string]interface{})
+		if !ok {
+			return nil, fmt.Errorf("failed to cast item to map[string]interface{}: %v", item)
+		}
+		result = append(result, castItem)
+	}
+
+	return result, nil
 }

internal/models/models.go

@@ -1,9 +1,5 @@
 package models
 
-import (
-	"time"
-)
-
 type GraphQLRequest struct {
 	Query string `json:"query"`
 }
@@ -75,15 +71,6 @@ type GraphQlData[T any] struct {
 	Data T `json:"data"`
 }
 
-type UnbufferedMessageValue struct {
-	ID        string    `json:"id"`
-	DataType  string    `json:"dataType"`
-	VehicleID string    `json:"vehicleId"`
-	DeviceID  *string   `json:"deviceId"`
-	Timestamp time.Time `json:"timestamp"`
-	Data      Data      `json:"data"`
-}
-
 type Data struct {
 	Location  Location `json:"location"`
 	Speed     Signal   `json:"speed"`