Handle messages with no entityID

internal/reconcilers/repository.go

@@ -71,24 +71,35 @@ func (r *Reconciler) handleArtifactsReconcilerEvent(ctx context.Context, evt *me
 
 	m := message.NewMessage(uuid.New().String(), nil)
 	if err := entRefresh.ToMessage(m); err != nil {
-		return fmt.Errorf("error creating message: %w", err)
+		zerolog.Ctx(ctx).Error().Err(err).Msg("error marshalling message")
+		// no point in retrying, so we return nil
+		return nil
+	}
+
+	if evt.EntityID == uuid.Nil {
+		// this might happen if we process old messages during an upgrade, but there's no point in retrying
+		zerolog.Ctx(ctx).Error().Msg("entityID is nil")
+		return nil
 	}
 
 	m.SetContext(ctx)
 	if err := r.evt.Publish(events.TopicQueueRefreshEntityByIDAndEvaluate, m); err != nil {
+		// we retry in case watermill is having a bad day
 		return fmt.Errorf("error publishing message: %w", err)
 	}
 
 	// the code below this line will be refactored in a follow-up PR
 	ewp, err := r.propService.EntityWithPropertiesByID(ctx, evt.EntityID, nil)
 	if err != nil {
+		// database might be down, so we retry
 		log.Printf("error retrieving entity with properties: %v", err)
 		return nil
 	}
 
 	repoID, err := ewp.Properties.GetProperty(properties.PropertyUpstreamID).AsInt64()
 	if err != nil {
 		log.Printf("error getting property upstreamID: %v", err)
+		// no point in retrying, so we just return nil
 		return nil
 	}
 
@@ -101,6 +112,7 @@ func (r *Reconciler) handleArtifactsReconcilerEvent(ctx context.Context, evt *me
 		return nil
 	}
 	if err != nil {
+		// database might be down, so we retry
 		return fmt.Errorf("error retrieving repository %d in project %s: %w", repoID, evt.Project, err)
 	}
 

internal/reconcilers/repository_test.go

@@ -42,6 +42,7 @@ func Test_handleRepoReconcilerEvent(t *testing.T) {
 		setupPropSvcMocks func() fixtures.MockPropertyServiceBuilder
 		expectedPublish   bool
 		expectedErr       bool
+		entityID          uuid.UUID
 		topic             string
 	}{
 		{
@@ -54,9 +55,19 @@ func Test_handleRepoReconcilerEvent(t *testing.T) {
 				)
 			},
 			topic:           events.TopicQueueRefreshEntityByIDAndEvaluate,
+			entityID:        testRepoID,
 			expectedPublish: true,
 			expectedErr:     false,
 		},
+		{
+			name: "event with no upstream ID",
+			setupPropSvcMocks: func() fixtures.MockPropertyServiceBuilder {
+				return fixtures.NewMockPropertiesService()
+			},
+			entityID:        uuid.Nil,
+			expectedPublish: false,
+			expectedErr:     false,
+		},
 	}
 
 	for _, scenario := range scenarios {
@@ -66,7 +77,7 @@ func Test_handleRepoReconcilerEvent(t *testing.T) {
 			ctrl := gomock.NewController(t)
 			defer ctrl.Finish()
 
-			msg, err := messages.NewRepoReconcilerMessage(testProviderID, testRepoID, testProjectID)
+			msg, err := messages.NewRepoReconcilerMessage(testProviderID, scenario.entityID, testProjectID)
 			require.NoError(t, err)
 			require.NotNil(t, msg)