feat: add aggregator consumer

aggregator/aggregator.go

@@ -6,6 +6,7 @@ import (
 	"app/base/mqueue"
 	"app/base/utils"
 	"sync"
+	"time"
 
 	"github.com/gin-gonic/gin"
 )
@@ -18,11 +19,14 @@ var (
 
 func readPodConfig() {
 	consumerCount = utils.PodConfig.GetInt("consumer_count", 1)
-	enableNotifications = utils.PodConfig.GetBool("instant_notifications", false)
+	enableNotifications = utils.PodConfig.GetBool("enable_notifications", false)
+	batchSize = utils.PodConfig.GetInt("advisory_batch_size", 4000)
+	flushTimeout = time.Duration(utils.PodConfig.GetInt("advisory_flush_timeout_ms", 500)) * time.Millisecond
 }
 
 func configure() {
 	advisoryUpdateTopic = utils.FailIfEmpty(utils.CoreCfg.AdvisoryUpdateTopic, "ADVISORY_UPDATE_TOPIC")
+	initBuffer()
 	configureNotifications()
 }
 

aggregator/aggregator_test.go

@@ -0,0 +1,15 @@
+package aggregator
+
+import (
+	"app/base/database"
+	"os"
+	"testing"
+)
+
+func TestMain(m *testing.M) {
+	exitCode := m.Run()
+	if database.DB != nil {
+		database.DB.Exec("SELECT refresh_account_advisory_caches_multi(NULL, 1)")
+	}
+	os.Exit(exitCode)
+}

aggregator/drift_check.go

@@ -0,0 +1,67 @@
+package aggregator
+
+import (
+	"app/base/database"
+	"app/base/utils"
+)
+
+type advisoryCounts struct {
+	AdvisoryID         int64
+	SystemsApplicable  int
+	SystemsInstallable int
+}
+
+func checkAdvisoryDrift(rhAccountID int, advisoryIDs []int64) {
+	if len(advisoryIDs) == 0 {
+		return
+	}
+
+	var newCounts []advisoryCounts
+	err := database.DB.Table("account_advisory aa").
+		Select(`aa.advisory_id,
+			SUM(aa.systems_installable) as systems_installable,
+			SUM(aa.systems_applicable) as systems_applicable`).
+		Where("aa.rh_account_id = ? AND aa.advisory_id IN (?)", rhAccountID, advisoryIDs).
+		Group("aa.advisory_id").
+		Find(&newCounts).Error
+	if err != nil {
+		utils.LogError("err", err, "rh_account_id", rhAccountID, "drift check: failed to query account_advisory")
+		return
+	}
+
+	newCountsMap := make(map[int64]advisoryCounts, len(newCounts))
+	for _, c := range newCounts {
+		newCountsMap[c.AdvisoryID] = c
+	}
+
+	var legacyCounts []advisoryCounts
+	err = database.DB.Table("advisory_account_data").
+		Select("advisory_id, systems_applicable, systems_installable").
+		Where("rh_account_id = ? AND advisory_id IN (?)", rhAccountID, advisoryIDs).
+		Find(&legacyCounts).Error
+	if err != nil {
+		utils.LogError("err", err, "rh_account_id", rhAccountID, "drift check: failed to query advisory_account_data")
+		return
+	}
+
+	for _, legacy := range legacyCounts {
+		newVal, ok := newCountsMap[legacy.AdvisoryID]
+		if !ok {
+			utils.LogWarn("rh_account_id", rhAccountID, "advisory_id", legacy.AdvisoryID,
+				"drift check: advisory present in legacy table but missing from account_advisory")
+			continue
+		}
+		if legacy.SystemsApplicable != newVal.SystemsApplicable || legacy.SystemsInstallable != newVal.SystemsInstallable {
+			utils.LogWarn("rh_account_id", rhAccountID, "advisory_id", legacy.AdvisoryID,
+				"legacy_applicable", legacy.SystemsApplicable, "new_applicable", newVal.SystemsApplicable,
+				"legacy_installable", legacy.SystemsInstallable, "new_installable", newVal.SystemsInstallable,
+				"drift check: count mismatch between legacy and new table")
+		}
+		delete(newCountsMap, legacy.AdvisoryID)
+	}
+
+	for _, newVal := range newCountsMap {
+		utils.LogWarn("rh_account_id", rhAccountID, "advisory_id", newVal.AdvisoryID,
+			"drift check: advisory present in account_advisory but missing from legacy table")
+	}
+}

aggregator/drift_check_test.go

@@ -0,0 +1,61 @@
+package aggregator
+
+import (
+	"app/base/core"
+	"app/base/database"
+	"app/base/models"
+	"app/base/utils"
+	"testing"
+
+	log "github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestCheckAdvisoryDriftCountMismatch(t *testing.T) {
+	utils.SkipWithoutDB(t)
+	core.SetupTestEnvironment()
+
+	assert.Nil(t, database.DB.Exec("SELECT refresh_advisory_caches(NULL, 1)").Error)
+	assert.Nil(t, database.DB.Exec("SELECT backfill_account_advisory(1)").Error)
+	defer database.DeleteAccountAdvisoryByAccount(t, 1)
+
+	assert.Nil(t, database.DB.Model(&models.AccountAdvisory{}).
+		Where("advisory_id = 1 AND rh_account_id = 1").
+		Update("systems_installable", 999).Error)
+
+	hook := utils.NewTestLogHook(log.WarnLevel)
+	log.AddHook(hook)
+
+	checkAdvisoryDrift(1, []int64{1})
+
+	found := false
+	for _, entry := range hook.LogEntries {
+		if entry.Message == "drift check: count mismatch between legacy and new table" {
+			found = true
+			break
+		}
+	}
+	assert.True(t, found, "expected count mismatch warning")
+}
+
+func TestCheckAdvisoryDriftMissingFromNew(t *testing.T) {
+	utils.SkipWithoutDB(t)
+	core.SetupTestEnvironment()
+
+	assert.Nil(t, database.DB.Exec("SELECT refresh_advisory_caches(NULL, 1)").Error)
+	defer database.DeleteAccountAdvisoryByAccount(t, 1)
+
+	hook := utils.NewTestLogHook(log.WarnLevel)
+	log.AddHook(hook)
+
+	checkAdvisoryDrift(1, []int64{1, 2})
+
+	found := false
+	for _, entry := range hook.LogEntries {
+		if entry.Message == "drift check: advisory present in legacy table but missing from account_advisory" {
+			found = true
+			break
+		}
+	}
+	assert.True(t, found, "expected missing from new table warning")
+}

aggregator/events.go

@@ -1,20 +1,103 @@
 package aggregator
 
 import (
+	"app/base/database"
 	"app/base/mqueue"
 	"app/base/utils"
+	"sync"
+	"time"
 
 	"github.com/bytedance/sonic"
+	"github.com/lib/pq"
 )
 
-// TODO: stub - will process advisory update events in batches and update account_advisory table
+var (
+	batchSize      int
+	flushTimeout   time.Duration
+	advisoryBuffer []mqueue.AdvisoryUpdateEvent
+	bufferLock     sync.Mutex
+	flushTimer     *time.Timer
+)
+
+func initBuffer() {
+	advisoryBuffer = make([]mqueue.AdvisoryUpdateEvent, 0, batchSize+1)
+	flushTimer = time.AfterFunc(87600*time.Hour, func() {
+		utils.LogInfo("flushing advisory buffer after timeout")
+		flushAdvisoryBuffer()
+	})
+}
+
+func flushAdvisoryBuffer() {
+	bufferLock.Lock()
+	if len(advisoryBuffer) == 0 {
+		bufferLock.Unlock()
+		return
+	}
+	// Copy and unlock before processing so new events can accumulate while DB work runs
+	batch := make([]mqueue.AdvisoryUpdateEvent, len(advisoryBuffer))
+	copy(batch, advisoryBuffer)
+	advisoryBuffer = advisoryBuffer[:0]
+	bufferLock.Unlock()
+
+	grouped := groupAdvisoryUpdates(batch)
+	processAdvisoryBatch(grouped)
+}
+
 func advisoryUpdateHandler(m mqueue.KafkaMessage) error {
 	var event mqueue.AdvisoryUpdateEvent
 	if err := sonic.Unmarshal(m.Value, &event); err != nil {
-		utils.LogError("err", err, "Could not deserialize advisory update event")
+		utils.LogError("err", err, "could not deserialize advisory update event")
 		return nil
 	}
-	// TODO: advisory update code goes here
-	_ = event
+
+	bufferLock.Lock()
+	advisoryBuffer = append(advisoryBuffer, event)
+	flushTimer.Reset(flushTimeout)
+	shouldFlush := len(advisoryBuffer) >= batchSize
+	bufferLock.Unlock()
+
+	if shouldFlush {
+		utils.LogInfo("flushing full advisory buffer")
+		flushAdvisoryBuffer()
+	}
 	return nil
 }
+
+func groupAdvisoryUpdates(events []mqueue.AdvisoryUpdateEvent) map[int][]int64 {
+	sets := make(map[int]map[int64]struct{})
+	for _, e := range events {
+		if _, ok := sets[e.RhAccountID]; !ok {
+			sets[e.RhAccountID] = make(map[int64]struct{})
+		}
+		for _, id := range e.AdvisoryIDs {
+			sets[e.RhAccountID][id] = struct{}{}
+		}
+	}
+
+	grouped := make(map[int][]int64, len(sets))
+	for accID, idSet := range sets {
+		ids := make([]int64, 0, len(idSet))
+		for id := range idSet {
+			ids = append(ids, id)
+		}
+		grouped[accID] = ids
+	}
+	return grouped
+}
+
+func processAdvisoryBatch(grouped map[int][]int64) {
+	for rhAccountID, advisoryIDs := range grouped {
+		utils.LogInfo("rh_account_id", rhAccountID, "advisory_count", len(advisoryIDs), "refreshing account advisory caches")
+		err := database.DB.Exec("SELECT refresh_account_advisory_caches_multi(?, ?)", pq.Array(advisoryIDs), rhAccountID).Error //nolint:lll
+		if err != nil {
+			utils.LogError("err", err, "rh_account_id", rhAccountID, "failed to refresh account advisory caches")
+			continue
+		}
+
+		checkAdvisoryDrift(rhAccountID, advisoryIDs)
+
+		if err := publishNewAdvisoryNotification(rhAccountID, advisoryIDs); err != nil {
+			utils.LogError("err", err, "rh_account_id", rhAccountID, "failed to publish new advisory notification")
+		}
+	}
+}

aggregator/events_test.go

@@ -0,0 +1,93 @@
+package aggregator
+
+import (
+	"app/base/core"
+	"app/base/database"
+	"app/base/mqueue"
+	"app/base/utils"
+	"sort"
+	"testing"
+	"time"
+
+	"github.com/bytedance/sonic"
+	"github.com/stretchr/testify/assert"
+)
+
+func toKafkaMessage(t *testing.T, event mqueue.AdvisoryUpdateEvent) mqueue.KafkaMessage {
+	t.Helper()
+	data, err := sonic.Marshal(event)
+	assert.Nil(t, err)
+	return mqueue.KafkaMessage{Value: data}
+}
+
+func TestGroupAdvisoryUpdatesSingleAccount(t *testing.T) {
+	events := []mqueue.AdvisoryUpdateEvent{
+		{RhAccountID: 1, AdvisoryIDs: []int64{1, 2}},
+		{RhAccountID: 1, AdvisoryIDs: []int64{2, 3}},
+	}
+	grouped := groupAdvisoryUpdates(events)
+
+	assert.Len(t, grouped, 1)
+	ids := grouped[1]
+	sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] })
+	assert.Equal(t, []int64{1, 2, 3}, ids)
+}
+
+func TestGroupAdvisoryUpdatesMultipleAccounts(t *testing.T) {
+	events := []mqueue.AdvisoryUpdateEvent{
+		{RhAccountID: 1, AdvisoryIDs: []int64{1, 2}},
+		{RhAccountID: 2, AdvisoryIDs: []int64{5}},
+		{RhAccountID: 1, AdvisoryIDs: []int64{3}},
+	}
+	grouped := groupAdvisoryUpdates(events)
+
+	assert.Len(t, grouped, 2)
+
+	ids1 := grouped[1]
+	sort.Slice(ids1, func(i, j int) bool { return ids1[i] < ids1[j] })
+	assert.Equal(t, []int64{1, 2, 3}, ids1)
+	assert.Equal(t, []int64{int64(5)}, grouped[2])
+}
+
+func TestGroupAdvisoryUpdatesDeduplicates(t *testing.T) {
+	events := []mqueue.AdvisoryUpdateEvent{
+		{RhAccountID: 1, AdvisoryIDs: []int64{1, 1, 2}},
+		{RhAccountID: 1, AdvisoryIDs: []int64{2, 2, 1}},
+	}
+	grouped := groupAdvisoryUpdates(events)
+
+	ids := grouped[1]
+	sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] })
+	assert.Equal(t, []int64{1, 2}, ids)
+}
+
+func TestBufferedEventsProcessedOnBatchThreshold(t *testing.T) {
+	utils.SkipWithoutDB(t)
+	core.SetupTestEnvironment()
+
+	assert.Nil(t, database.DB.Exec("SELECT refresh_advisory_caches(NULL, 1)").Error)
+	defer database.DeleteAccountAdvisoryByAccount(t, 1)
+
+	batchSize = 3
+	flushTimeout = time.Hour
+	initBuffer()
+
+	msg := toKafkaMessage(t, mqueue.AdvisoryUpdateEvent{RhAccountID: 1, AdvisoryIDs: []int64{1, 2}})
+
+	// First two events accumulate in the buffer
+	assert.Nil(t, advisoryUpdateHandler(msg))
+	assert.Equal(t, 1, len(advisoryBuffer))
+	assert.Nil(t, advisoryUpdateHandler(msg))
+	assert.Equal(t, 2, len(advisoryBuffer))
+
+	// Third event triggers flush and processAdvisoryBatch runs
+	assert.Nil(t, advisoryUpdateHandler(msg))
+	assert.Equal(t, 0, len(advisoryBuffer))
+
+	// Verify account_advisory was populated
+	var count int64
+	assert.Nil(t, database.DB.Table("account_advisory").
+		Where("rh_account_id = 1 AND advisory_id IN (?)", []int64{1, 2}).
+		Count(&count).Error)
+	assert.True(t, count > 0, "account_advisory rows should be created after flush")
+}