WIP fix handling of dosing decisions

Author: ewollesen

data/events/alerts.go

@@ -112,36 +112,46 @@ func isActivityAndActivityOnly(updatedFields []string) bool {
 	return hasActivity
 }
 
+// deviceDataIdentifiers are common to DosingDecision and Glucose.
+//
+// They facilitate queries of the latest data for alerts evaluation.
+type deviceDataIdentifiers struct {
+	UploadID string `json:"uploadId,omitempty" bson:"uploadId,omitempty"`
+	UserID   string `json:"-" bson:"_userId,omitempty"`
+}
+
 func (c *Consumer) consumeDeviceData(ctx context.Context,
 	session sarama.ConsumerGroupSession, msg *sarama.ConsumerMessage) error {
 
-	datum := &Glucose{}
-	if _, err := unmarshalMessageValue(msg.Value, datum); err != nil {
-		return err
-	}
 	lgr := c.logger(ctx)
-	lgr.WithField("data", datum).Info("consuming a device data message")
 
-	if datum.UserID == nil {
-		return errors.New("Unable to retrieve alerts configs: userID is nil")
+	id := &deviceDataIdentifiers{}
+	if _, err := unmarshalMessageValue(msg.Value, id); err != nil {
+		return errors.Wrap(err, "Unable to unmarshal device data message")
+	}
+
+	lgr.WithField("identifiers", id).Info("consuming a device data message")
+
+	if id.UserID == "" {
+		return errors.New("Unable to retrieve alerts configs: userID is empty")
 	}
-	if datum.UploadID == nil {
-		return errors.New("Unable to retrieve alerts configs: DataSetID is nil")
+	if id.UploadID == "" {
+		return errors.New("Unable to retrieve alerts configs: DataSetID is empty")
 	}
-	ctx = log.NewContextWithLogger(ctx, lgr.WithField("followedUserID", *datum.UserID))
+	ctx = log.NewContextWithLogger(ctx, lgr.WithField("followedUserID", id.UserID))
 	lastComm := alerts.LastCommunication{
-		UserID:                 *datum.UserID,
+		UserID:                 id.UserID,
 		LastReceivedDeviceData: time.Now(),
-		DataSetID:              *datum.UploadID,
+		DataSetID:              id.UploadID,
 	}
 	err := c.LastCommunications.RecordReceivedDeviceData(ctx, lastComm)
 	if err != nil {
 		lgr.WithError(err).Info("Unable to record device data received")
 	}
-	notes, err := c.Evaluator.EvaluateData(ctx, *datum.UserID, *datum.UploadID)
+	notes, err := c.Evaluator.EvaluateData(ctx, id.UserID, id.UploadID)
 	if err != nil {
 		format := "Unable to evalaute device data triggered event for user %s"
-		return errors.Wrapf(err, format, *datum.UserID)
+		return errors.Wrapf(err, format, id.UserID)
 	}
 	for idx, note := range notes {
 		lgr.WithField("idx", idx).WithField("note", note).Debug("notes")

data/events/alerts_test.go

@@ -84,27 +84,27 @@ var _ = Describe("Consumer", func() {
 			Expect(deps.Session.MarkCalls).To(Equal(1))
 		})
 
-		It("errors out when the datum's UserID is nil", func() {
+		It("errors out when the datum's UserID is empty", func() {
 			blood := newTestStaticDatumMmolL(7.2)
 			blood.UserID = nil
 			kafkaMsg := newAlertsMockConsumerMessage(".data.deviceData.alerts", blood)
 			docs := []interface{}{bson.M{}}
 			c, deps := newConsumerTestDeps(docs)
 
 			Expect(c.Consume(deps.Context, deps.Session, kafkaMsg)).
-				To(MatchError(ContainSubstring("userID is nil")))
+				To(MatchError(ContainSubstring("userID is empty")))
 			Expect(deps.Session.MarkCalls).To(Equal(0))
 		})
 
-		It("errors out when the datum's UploadID is nil", func() {
+		It("errors out when the datum's UploadID is empty", func() {
 			blood := newTestStaticDatumMmolL(7.2)
 			blood.UploadID = nil
 			kafkaMsg := newAlertsMockConsumerMessage(".data.deviceData.alerts", blood)
 			docs := []interface{}{bson.M{}}
 			c, deps := newConsumerTestDeps(docs)
 
 			Expect(c.Consume(deps.Context, deps.Session, kafkaMsg)).
-				To(MatchError(ContainSubstring("DataSetID is nil")))
+				To(MatchError(ContainSubstring("DataSetID is empty")))
 			Expect(deps.Session.MarkCalls).To(Equal(0))
 		})
 

data/events/events.go

@@ -171,12 +171,14 @@ func CappedExponentialBinaryDelay(cap time.Duration) func(int) time.Duration {
 
 type AlertsEventsConsumer struct {
 	Consumer asyncevents.SaramaMessageConsumer
+	Logger   log.Logger
 }
 
 func (c *AlertsEventsConsumer) Consume(ctx context.Context,
 	session sarama.ConsumerGroupSession, message *sarama.ConsumerMessage) error {
 	err := c.Consumer.Consume(ctx, session, message)
 	if err != nil {
+		c.Logger.WithError(err).Info("Unable to consumer alerts event")
 		session.MarkMessage(message, fmt.Sprintf("I have given up after error: %s", err))
 		return err
 	}

data/service/service/standard.go

@@ -507,6 +507,7 @@ func (s *Standard) initializeAlertsEventsHandler() error {
 		Topics:  prefixedTopics,
 		Sarama:  commonConfig.SaramaConfig,
 		MessageConsumer: &dataEvents.AlertsEventsConsumer{
+			Logger:   s.Logger(),
 			Consumer: ec,
 		},
 	}