RHINENG-26348: add skip notifications for platform events

base/mqueue/platform_event.go

@@ -22,6 +22,10 @@ type PlatformEvent struct {
 	URL         *string                 `json:"url"`
 	SystemIDs   []uuid.UUID             `json:"system_ids,omitempty"`
 	RequestIDs  []string                `json:"request_ids,omitempty"`
+	// SkipNotifications suppresses instant advisory notification publish for this event.
+	// Evaluator still marks matching advisory_account_data.notified so later evals do not flood.
+	// Used by recovery recalc; omit/false for normal upload/recalc traffic.
+	SkipNotifications bool `json:"skip_notifications,omitempty"`
 }
 
 type EvalData struct {

base/mqueue/platform_event_test.go

@@ -9,6 +9,31 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
+func TestPlatformEventSkipNotificationsJSON(t *testing.T) {
+	orgID := "org_1"
+	event := PlatformEvent{
+		AccountID:         1,
+		OrgID:             &orgID,
+		SkipNotifications: true,
+	}
+	data, err := sonic.Marshal(event)
+	assert.NoError(t, err)
+
+	var parsed PlatformEvent
+	assert.NoError(t, sonic.Unmarshal(data, &parsed))
+	assert.True(t, parsed.SkipNotifications)
+
+	// omitempty: false must not appear in JSON; fresh unmarshal defaults to false
+	event.SkipNotifications = false
+	data, err = sonic.Marshal(event)
+	assert.NoError(t, err)
+	assert.NotContains(t, string(data), "skip_notifications")
+
+	var parsedFalse PlatformEvent
+	assert.NoError(t, sonic.Unmarshal(data, &parsedFalse))
+	assert.False(t, parsedFalse.SkipNotifications)
+}
+
 func TestWriteEventsOfInventoryAccounts(t *testing.T) {
 	var (
 		acc  = 1

evaluator/evaluate.go

@@ -520,9 +520,11 @@ func evaluateAndStore(system *models.SystemPlatformV2,
 		}
 	}
 
-	// Send instant notification with new advisories
-	if enableInstantNotifications {
-		err = publishNewAdvisoriesNotification(tx, system, event.GetOrgID(), systemAdvisoriesNew)
+	// Instant notifications, or mark-notified only when the event opts out of publishing
+	// (e.g. recovery recalc with skip_notifications).
+	if event.SkipNotifications || enableInstantNotifications {
+		err = publishNewAdvisoriesNotification(tx, system, event.GetOrgID(), systemAdvisoriesNew,
+			event.SkipNotifications)
 		if err != nil {
 			evaluationCnt.WithLabelValues("error-advisory-notification").Inc()
 			utils.LogError("orgID", event.GetOrgID(), "inventoryID", system.GetInventoryID(), "err", err,

evaluator/notifications.go

@@ -66,12 +66,24 @@ func getSystemTags(tx *gorm.DB, system *models.SystemPlatformV2) ([]ntf.SystemTa
 	return tags, nil
 }
 
-func publishNewAdvisoriesNotification(tx *gorm.DB, system *models.SystemPlatformV2, orgID string,
-	newAdvisories SystemAdvisoryMap) error {
-	if notificationsPublisher == nil {
+func markAdvisoriesNotified(tx *gorm.DB, accountID int, advisoryIDs []int64) error {
+	if len(advisoryIDs) == 0 {
 		return nil
 	}
+	err := tx.Table("advisory_account_data").
+		Where("rh_account_id = ? AND advisory_id IN (?)", accountID, advisoryIDs).
+		Update("notified", time.Now()).Error
+	if err != nil {
+		return errors.Wrap(err, "updating notified column failed")
+	}
+	return nil
+}
 
+// publishNewAdvisoriesNotification publishes instant new-advisory notifications unless
+// skipPublish is true. In both cases, matching advisory_account_data rows are marked notified
+// when there is something to notify about (so skipPublish still prevents later flood).
+func publishNewAdvisoriesNotification(tx *gorm.DB, system *models.SystemPlatformV2, orgID string,
+	newAdvisories SystemAdvisoryMap, skipPublish bool) error {
 	defer utils.ObserveSecondsSince(time.Now(), evaluationPartDuration.WithLabelValues("advisory-notification-publish"))
 
 	advisories, err := getUnnotifiedAdvisories(tx, system.Inventory.RhAccountID, newAdvisories)
@@ -82,6 +94,21 @@ func publishNewAdvisoriesNotification(tx *gorm.DB, system *models.SystemPlatform
 		return nil
 	}
 
+	advisoryIDs := make([]int64, 0, len(advisories))
+	for _, a := range advisories {
+		advisoryIDs = append(advisoryIDs, a.AdvisoryID)
+	}
+
+	if skipPublish {
+		utils.LogInfo("inventoryID", system.GetInventoryID(), "advisoryIDs", advisoryIDs, "orgID", orgID,
+			"skipping advisory notification publish")
+		return markAdvisoriesNotified(tx, system.Inventory.RhAccountID, advisoryIDs)
+	}
+
+	if notificationsPublisher == nil {
+		return nil
+	}
+
 	events := make([]ntf.Event, 0, len(advisories))
 	for _, advisory := range advisories {
 		// At least empty metadata required to avoid NPE further on at the time of writing.
@@ -108,20 +135,8 @@ func publishNewAdvisoriesNotification(tx *gorm.DB, system *models.SystemPlatform
 		return errors.Wrap(err, "writing message to notifications publisher failed")
 	}
 
-	advisoryIDs := make([]int64, 0, len(advisories))
-	for _, a := range advisories {
-		advisoryIDs = append(advisoryIDs, a.AdvisoryID)
-	}
-
 	utils.LogInfo("inventoryID", system.GetInventoryID(), "advisoryIDs", advisoryIDs, "orgID", orgID,
 		"notification sent successfully")
 
-	err = tx.Table("advisory_account_data").
-		Where("rh_account_id = ? AND advisory_id IN (?)", system.Inventory.RhAccountID, advisoryIDs).
-		Update("notified", time.Now()).Error
-	if err != nil {
-		return errors.Wrap(err, "updating notified column failed")
-	}
-
-	return nil
+	return markAdvisoriesNotified(tx, system.Inventory.RhAccountID, advisoryIDs)
 }

evaluator/notifications_test.go

@@ -86,6 +86,50 @@ func TestAdvisoriesNotificationPublish(t *testing.T) {
 	database.DeleteAdvisoryAccountData(t, rhAccountID, oldSystemAdvisoryIDs)
 }
 
+// TestAdvisoriesNotificationSkipPublishViaEvaluate runs a full evaluateHandler path with
+// skip_notifications set: advisories are stored and marked notified, but no Kafka notify is sent.
+func TestAdvisoriesNotificationSkipPublishViaEvaluate(t *testing.T) {
+	utils.SkipWithoutDB(t)
+	utils.SkipWithoutPlatform(t)
+	core.SetupTestEnvironment()
+
+	configure()
+	loadCache()
+	mockWriter := mqueue.MockKafkaWriter{}
+	notificationsPublisher = &mockWriter
+
+	expectedAddedAdvisories := []string{"RH-1", "RH-2", "RH-100"}
+	expectedAdvisoryIDs := []int64{1, 2}
+	oldSystemAdvisoryIDs := []int64{1, 3, 4}
+
+	database.DeleteSystemAdvisories(t, testDBID, expectedAdvisoryIDs)
+	database.DeleteAdvisoryAccountData(t, rhAccountID, expectedAdvisoryIDs)
+	database.CreateSystemAdvisories(t, rhAccountID, testDBID, oldSystemAdvisoryIDs)
+	database.CreateAdvisoryAccountData(t, rhAccountID, oldSystemAdvisoryIDs, 1)
+	database.CheckCachesValid(t)
+	database.CheckAdvisoriesAccountDataNotified(t, rhAccountID, oldSystemAdvisoryIDs, false)
+
+	orgID := "1234567"
+	data, err := sonic.Marshal(mqueue.PlatformEvent{
+		SystemIDs:         []uuid.UUID{testInventoryID},
+		RequestIDs:        []string{"request-skip-notif"},
+		AccountID:         rhAccountID,
+		OrgID:             &orgID,
+		SkipNotifications: true,
+	})
+	assert.NoError(t, err)
+	err = evaluateHandler(mqueue.KafkaMessage{Value: data})
+	assert.NoError(t, err)
+
+	advisoryIDs := database.CheckAdvisoriesInDB(t, expectedAddedAdvisories)
+	database.CheckAdvisoriesAccountDataNotified(t, rhAccountID, expectedAdvisoryIDs, true)
+	assert.Empty(t, mockWriter.Messages, "no notification should be sent when skip_notifications is set")
+
+	database.DeleteSystemAdvisories(t, testDBID, advisoryIDs)
+	database.DeleteAdvisoryAccountData(t, rhAccountID, advisoryIDs)
+	database.DeleteAdvisoryAccountData(t, rhAccountID, oldSystemAdvisoryIDs)
+}
+
 func TestAdvisoriesNotificationMessage(t *testing.T) {
 	events := make([]ntf.Event, 1)
 	events[0] = ntf.Event{
@@ -188,7 +232,7 @@ func TestAdvisoriesNotificationAlreadyNotified(t *testing.T) {
 		"RH-2": {AdvisoryID: 2},
 	}
 
-	err := publishNewAdvisoriesNotification(database.DB, system, orgID, newAdvs)
+	err := publishNewAdvisoriesNotification(database.DB, system, orgID, newAdvs, false)
 	assert.NoError(t, err)
 	assert.Empty(t, mockWriter.Messages, "no notification should be sent when all advisories are already notified")
 }
@@ -215,10 +259,46 @@ func TestAdvisoriesNotificationEmptyAdvisoryMap(t *testing.T) {
 	}
 
 	// An empty map means there is nothing to query — no messages should be produced regardless.
-	publishNewAdvisoriesNotification(database.DB, system, orgID, SystemAdvisoryMap{}) //nolint:errcheck
+	publishNewAdvisoriesNotification(database.DB, system, orgID, SystemAdvisoryMap{}, false) //nolint:errcheck
 	assert.Empty(t, mockWriter.Messages, "no notification should be sent when the advisory map is empty")
 }
 
+// TestAdvisoriesNotificationSkipPublish verifies recovery-style skip_notifications: no Kafka
+// message is sent, but advisory_account_data.notified is still set.
+func TestAdvisoriesNotificationSkipPublish(t *testing.T) {
+	utils.SkipWithoutDB(t)
+	core.SetupTestEnvironment()
+	configure()
+
+	mockWriter := mqueue.MockKafkaWriter{}
+	notificationsPublisher = &mockWriter
+
+	advisoryIDs := []int64{1, 2}
+	database.DeleteAdvisoryAccountData(t, rhAccountID, advisoryIDs)
+	database.CreateAdvisoryAccountData(t, rhAccountID, advisoryIDs, 1)
+	database.CheckAdvisoriesAccountDataNotified(t, rhAccountID, advisoryIDs, false)
+	defer database.DeleteAdvisoryAccountData(t, rhAccountID, advisoryIDs)
+
+	system := &models.SystemPlatformV2{
+		Inventory: models.SystemInventory{
+			ID:          1,
+			RhAccountID: rhAccountID,
+			InventoryID: uuid.MustParse("00000000-0000-0000-0000-000000000001"),
+			DisplayName: "display name",
+		},
+		Patch: models.SystemPatch{},
+	}
+	newAdvs := SystemAdvisoryMap{
+		"RH-1": {AdvisoryID: 1},
+		"RH-2": {AdvisoryID: 2},
+	}
+
+	err := publishNewAdvisoriesNotification(database.DB, system, orgID, newAdvs, true)
+	assert.NoError(t, err)
+	assert.Empty(t, mockWriter.Messages, "no notification should be sent when skipPublish is true")
+	database.CheckAdvisoriesAccountDataNotified(t, rhAccountID, advisoryIDs, true)
+}
+
 // TestGetUnnotifiedAdvisoriesReturnsEmpty documents the return-type contract of
 // getUnnotifiedAdvisories: when all candidate advisories are already notified the function
 // must return a non-nil empty slice (not nil). This prevents a future nil-vs-empty regression