records_reporter_test.go

package activationrecords

import (
	"fmt"
	"testing"

	"github.com/google/uuid"
	"github.com/segmentio/go-hll"
	"github.com/stretchr/testify/require"

	"github.com/rudderlabs/rudder-go-kit/config"
	"github.com/rudderlabs/rudder-go-kit/logger"
	"github.com/rudderlabs/rudder-go-kit/stats"
	"github.com/rudderlabs/rudder-go-kit/stats/memstats"

	"github.com/rudderlabs/rudder-server/jobsdb"
)

func TestUniqueActivationRecordsReporter(t *testing.T) {
	// prepareJob builds a job with the standard activation payload shape. The Parameters
	// intentionally omit source_category: on the internal-batch ingestion path it is empty,
	// and the reporter classifies reverse-ETL sources from the config map (metadataByID
	// below), not the param.
	prepareJob := func(sourceID, destinationID, fingerprint, origin, workspaceID string) *jobsdb.JobT {
		return &jobsdb.JobT{
			Parameters:   fmt.Appendf(nil, `{"source_id":%q,"destination_id":%q}`, sourceID, destinationID),
			EventPayload: fmt.Appendf(nil, `{"batch":[{"context":{"activation":{"fingerprint":%q,"origin":%q}}}]}`, fingerprint, origin),
			UserID:       uuid.NewString(),
			UUID:         uuid.New(),
			CustomVal:    "GW",
			WorkspaceId:  workspaceID,
		}
	}

	// prepareJobNoDstID builds a job whose Parameters JSON omits the destination_id key entirely.
	prepareJobNoDstID := func(sourceID, fingerprint, origin, workspaceID string) *jobsdb.JobT {
		return &jobsdb.JobT{
			Parameters:   fmt.Appendf(nil, `{"source_id":%q}`, sourceID),
			EventPayload: fmt.Appendf(nil, `{"batch":[{"context":{"activation":{"fingerprint":%q,"origin":%q}}}]}`, fingerprint, origin),
			UserID:       uuid.NewString(),
			UUID:         uuid.New(),
			CustomVal:    "GW",
			WorkspaceId:  workspaceID,
		}
	}

	// metadataByID maps source_id -> SourceDefinition metadata, mirroring the map the
	// processor passes from the backend config. "src1" is a reverse-ETL SQL warehouse
	// source; "src-eventstream" is not. Unknown source_ids resolve to zero-value metadata
	// and are treated as non-rETL.
	metadataByID := map[string]SourceMetadata{
		"src1":            {Category: "warehouse", Name: "Snowflake"},
		"src-eventstream": {Category: "eventStream", Name: "javascript"},
		"src-s3":          {Category: "warehouse", Name: "s3"},
	}

	t.Run("constructor validates HLL settings", func(t *testing.T) {
		// Default settings are valid.
		_, err := NewUniqueActivationRecordsReporter(logger.NOP, config.Default, stats.NOP)
		require.NoError(t, err)

		// An out-of-range precision (Log2m valid range is [4, 31]) must fail fast at
		// construction rather than panicking later in the processor hot path.
		badConf := config.New()
		badConf.Set("ActivationRecords.precision", 99)
		_, err = NewUniqueActivationRecordsReporter(logger.NOP, badConf, stats.NOP)
		require.Error(t, err)
	})

	t.Run("GenerateReportsFromJobs", func(t *testing.T) {
		reporter, err := NewUniqueActivationRecordsReporter(logger.NOP, config.Default, stats.NOP)
		require.NoError(t, err)

		testCases := []struct {
			name string
			jobs []*jobsdb.JobT
			// verify is called with the resulting reports slice.
			verify func(t *testing.T, reports []*ActivationRecord)
		}{
			{
				name: "both fields present - one report on right grain, origin carried",
				jobs: []*jobsdb.JobT{
					prepareJob("src1", "dst1", "fp1", "org1", "ws1"),
				},
				verify: func(t *testing.T, reports []*ActivationRecord) {
					require.Len(t, reports, 1)
					r := reports[0]
					require.Equal(t, "ws1", r.WorkspaceID)
					require.Equal(t, "src1", r.SourceID)
					require.Equal(t, "dst1", r.DestinationID)
					require.Equal(t, "org1", r.Origin)
					require.NotNil(t, r.FingerprintHll)
					require.Equal(t, uint64(1), r.FingerprintHll.Cardinality())
				},
			},
			{
				name: "allow-listed warehouse missing fingerprint - skip",
				jobs: []*jobsdb.JobT{
					prepareJob("src1", "dst1", "", "org1", "ws1"),
				},
				verify: func(t *testing.T, reports []*ActivationRecord) {
					require.Empty(t, reports)
				},
			},
			{
				name: "missing origin - skip",
				jobs: []*jobsdb.JobT{
					prepareJob("src1", "dst1", "fp1", "", "ws1"),
				},
				verify: func(t *testing.T, reports []*ActivationRecord) {
					require.Empty(t, reports)
				},
			},
			{
				name: "missing destination_id in params - skip",
				jobs: []*jobsdb.JobT{
					prepareJobNoDstID("src1", "fp1", "org1", "ws1"),
				},
				verify: func(t *testing.T, reports []*ActivationRecord) {
					require.Empty(t, reports)
				},
			},
			{
				name: "non-rETL source - skip even with fingerprint",
				jobs: []*jobsdb.JobT{
					{
						// An event-stream source (per the config map) whose payload carries a
						// client-stamped fingerprint must NOT be metered: MAR meters warehouse
						// sources only, classified from config.
						Parameters:   fmt.Appendf(nil, `{"source_id":%q,"destination_id":%q}`, "src-eventstream", "dst1"),
						EventPayload: []byte(`{"batch":[{"context":{"activation":{"fingerprint":"fp-injected","origin":"org1"}}}]}`),
						UserID:       uuid.NewString(),
						UUID:         uuid.New(),
						CustomVal:    "GW",
						WorkspaceId:  "ws1",
					},
				},
				verify: func(t *testing.T, reports []*ActivationRecord) {
					require.Empty(t, reports)
				},
			},
			{
				name: "warehouse source not in allow-list - skip silently even with fingerprint",
				jobs: []*jobsdb.JobT{
					prepareJob("src-s3", "dst1", "fp-cloud-storage", "org1", "ws1"),
				},
				verify: func(t *testing.T, reports []*ActivationRecord) {
					require.Empty(t, reports)
				},
			},
			{
				name: "config category is authoritative over a spoofed source_category param - skip",
				jobs: []*jobsdb.JobT{
					{
						// The job param claims warehouse, but the source is event-stream in the
						// config map; the param is not trusted, so the record is not metered.
						Parameters:   fmt.Appendf(nil, `{"source_id":%q,"destination_id":%q,"source_category":"warehouse"}`, "src-eventstream", "dst1"),
						EventPayload: []byte(`{"batch":[{"context":{"activation":{"fingerprint":"fp-spoof","origin":"org1"}}}]}`),
						UserID:       uuid.NewString(),
						UUID:         uuid.New(),
						CustomVal:    "GW",
						WorkspaceId:  "ws1",
					},
				},
				verify: func(t *testing.T, reports []*ActivationRecord) {
					require.Empty(t, reports)
				},
			},
			{
				name: "unknown source not in config - skip fail-closed",
				jobs: []*jobsdb.JobT{
					{
						// A source_id absent from the config lookup resolves to "" and is
						// skipped, even with a valid fingerprint.
						Parameters:   fmt.Appendf(nil, `{"source_id":%q,"destination_id":%q}`, "src-unknown", "dst1"),
						EventPayload: []byte(`{"batch":[{"context":{"activation":{"fingerprint":"fp-unknown","origin":"org1"}}}]}`),
						UserID:       uuid.NewString(),
						UUID:         uuid.New(),
						CustomVal:    "GW",
						WorkspaceId:  "ws1",
					},
				},
				verify: func(t *testing.T, reports []*ActivationRecord) {
					require.Empty(t, reports)
				},
			},
			{
				name: "duplicate fingerprint same connection - cardinality 1",
				jobs: []*jobsdb.JobT{
					prepareJob("src1", "dst1", "fp1", "org1", "ws1"),
					prepareJob("src1", "dst1", "fp1", "org1", "ws1"),
				},
				verify: func(t *testing.T, reports []*ActivationRecord) {
					require.Len(t, reports, 1)
					require.Equal(t, uint64(1), reports[0].FingerprintHll.Cardinality())
				},
			},
			{
				name: "two destinations same fingerprint - two reports",
				jobs: []*jobsdb.JobT{
					prepareJob("src1", "dst1", "fp1", "org1", "ws1"),
					prepareJob("src1", "dst2", "fp1", "org1", "ws1"),
				},
				verify: func(t *testing.T, reports []*ActivationRecord) {
					require.Len(t, reports, 2)
					dstIDs := make([]string, 0, len(reports))
					for _, r := range reports {
						dstIDs = append(dstIDs, r.DestinationID)
						require.Equal(t, uint64(1), r.FingerprintHll.Cardinality())
					}
					require.ElementsMatch(t, []string{"dst1", "dst2"}, dstIDs)
				},
			},
		}

		for _, tc := range testCases {
			t.Run(tc.name, func(t *testing.T) {
				reports := reporter.GenerateReportsFromJobs(tc.jobs, metadataByID)
				tc.verify(t, reports)
			})
		}
	})

	t.Run("GenerateReportsFromJobs_SourceDefinitionGateStats", func(t *testing.T) {
		t.Run("warehouse source not in allow-list skips silently without missing fingerprint stat", func(t *testing.T) {
			statsStore, err := memstats.New()
			require.NoError(t, err)
			reporter, err := NewUniqueActivationRecordsReporter(logger.NOP, config.New(), statsStore)
			require.NoError(t, err)

			reports := reporter.GenerateReportsFromJobs([]*jobsdb.JobT{
				prepareJob("src-s3", "dst1", "", "org1", "ws1"),
			}, metadataByID)

			require.Empty(t, reports)
			require.Empty(t, statsStore.GetByName("activation_records_skipped"))
		})

		t.Run("allow-listed warehouse missing fingerprint increments missing_fingerprint", func(t *testing.T) {
			statsStore, err := memstats.New()
			require.NoError(t, err)
			reporter, err := NewUniqueActivationRecordsReporter(logger.NOP, config.New(), statsStore)
			require.NoError(t, err)

			reports := reporter.GenerateReportsFromJobs([]*jobsdb.JobT{
				prepareJob("src1", "dst1", "", "org1", "ws1"),
			}, metadataByID)

			require.Empty(t, reports)
			metric := statsStore.Get("activation_records_skipped", stats.Tags{"reason": "missing_fingerprint"})
			require.NotNil(t, metric)
			require.Equal(t, float64(1), metric.LastValue())
		})

		t.Run("allow-list config changes are observed without reconstructing reporter", func(t *testing.T) {
			conf := config.New()
			conf.Set("ActivationRecords.allowedSourceDefinitionNames", []string{"snowflake"})
			reporter, err := NewUniqueActivationRecordsReporter(logger.NOP, conf, stats.NOP)
			require.NoError(t, err)

			customMetadataByID := map[string]SourceMetadata{
				"src-custom": {Category: "warehouse", Name: "custom_sql"},
			}
			job := prepareJob("src-custom", "dst1", "fp1", "org1", "ws1")

			require.Empty(t, reporter.GenerateReportsFromJobs([]*jobsdb.JobT{job}, customMetadataByID))

			conf.Set("ActivationRecords.allowedSourceDefinitionNames", []string{"custom_sql"})

			reports := reporter.GenerateReportsFromJobs([]*jobsdb.JobT{job}, customMetadataByID)
			require.Len(t, reports, 1)
			require.Equal(t, uint64(1), reports[0].FingerprintHll.Cardinality())
		})
	})

	t.Run("GenerateReportsFromJobs_MultiEventBatch", func(t *testing.T) {
		reporter, err := NewUniqueActivationRecordsReporter(logger.NOP, config.Default, stats.NOP)
		require.NoError(t, err)

		// prepareTwoEventJob builds a job whose batch has TWO events with distinct fingerprints.
		prepareTwoEventJob := func(sourceID, destinationID, fp1, fp2, origin, workspaceID string) *jobsdb.JobT {
			return &jobsdb.JobT{
				Parameters: fmt.Appendf(nil, `{"source_id":%q,"destination_id":%q}`, sourceID, destinationID),
				EventPayload: fmt.Appendf(nil,
					`{"batch":[{"context":{"activation":{"fingerprint":%q,"origin":%q}}},{"context":{"activation":{"fingerprint":%q,"origin":%q}}}]}`,
					fp1, origin, fp2, origin,
				),
				UserID:      uuid.NewString(),
				UUID:        uuid.New(),
				CustomVal:   "GW",
				WorkspaceId: workspaceID,
			}
		}

		t.Run("two distinct fingerprints in same batch => cardinality 2", func(t *testing.T) {
			job := prepareTwoEventJob("src1", "dst1", "fp-1", "fp-2", "org1", "ws1")
			reports := reporter.GenerateReportsFromJobs([]*jobsdb.JobT{job}, metadataByID)
			require.Len(t, reports, 1)
			require.Equal(t, "org1", reports[0].Origin)
			require.NotNil(t, reports[0].FingerprintHll)
			require.Equal(t, uint64(2), reports[0].FingerprintHll.Cardinality())
		})

		t.Run("batch with one valid and one invalid element - valid element still metered", func(t *testing.T) {
			// batch[0] has no fingerprint; batch[1] is valid.
			job := &jobsdb.JobT{
				Parameters:   fmt.Appendf(nil, `{"source_id":%q,"destination_id":%q}`, "src1", "dst1"),
				EventPayload: []byte(`{"batch":[{"context":{"activation":{"origin":"org1"}}},{"context":{"activation":{"fingerprint":"fp-valid","origin":"org1"}}}]}`),
				UserID:       uuid.NewString(),
				UUID:         uuid.New(),
				CustomVal:    "GW",
				WorkspaceId:  "ws1",
			}
			reports := reporter.GenerateReportsFromJobs([]*jobsdb.JobT{job}, metadataByID)
			require.Len(t, reports, 1)
			require.Equal(t, uint64(1), reports[0].FingerprintHll.Cardinality())
		})
	})

	t.Run("HLLSettings", func(t *testing.T) {
		reporter, err := NewUniqueActivationRecordsReporter(logger.NOP, config.Default, stats.NOP)
		require.NoError(t, err)
		require.Equal(t, 16, reporter.hllSettings.Log2m)
		require.Equal(t, 5, reporter.hllSettings.Regwidth)
	})

	t.Run("ForceFullSyncReplay", func(t *testing.T) {
		// A force-full-sync replays the SAME jobs in a SEPARATE reporting cycle.
		// The backend merges sketches across cycles (postgres hll union), so a
		// replay of the same fingerprint MUST NOT inflate the cardinality —
		// otherwise every retried sync would double-count monthly active records.
		reporter, err := NewUniqueActivationRecordsReporter(logger.NOP, config.Default, stats.NOP)
		require.NoError(t, err)

		job := prepareJob("src1", "dst1", "fp1", "org1", "ws1")

		first := reporter.GenerateReportsFromJobs([]*jobsdb.JobT{job}, metadataByID)
		require.Len(t, first, 1)
		require.Equal(t, uint64(1), first[0].FingerprintHll.Cardinality())

		second := reporter.GenerateReportsFromJobs([]*jobsdb.JobT{job}, metadataByID)
		require.Len(t, second, 1)
		require.Equal(t, uint64(1), second[0].FingerprintHll.Cardinality())

		// Merge the two cycles the way the backend would: union both sketches into
		// a fresh HLL built with the reporter's settings. Same fingerprint twice
		// across cycles still collapses to cardinality 1.
		union, err := hll.NewHll(*reporter.hllSettings)
		require.NoError(t, err)
		union.Union(*first[0].FingerprintHll)
		union.Union(*second[0].FingerprintHll)
		require.Equal(t, uint64(1), union.Cardinality())
	})
}