PLT-410: Store UUIDs in consumer-context-graph instead of base64-encoded IDs

Ticket: https://fetchrewards.atlassian.net/browse/PLT-410 Repo: consumer-graph-worker Date: 2026-03-12

Summary

The consumer-context-graph (Neo4j) previously stored product IDs (FIDOs) in their base64-encoded form as received from upstream sources (Kafka factoids, receipt events, Purchase History API). The graph worker then decoded base64 to UUID at read time before sending data to the consumer-context-service (CCS) and consumer-agent.

This change:

Decodes base64 to UUID at write time so Neo4j stores UUID strings directly.
Makes the read path tolerant of both formats using a new fido.ToUUID() helper, so there is no service disruption during the transition.

Problem

Neo4j stored FIDOs as base64 strings (e.g., dGVzdC1wcm9kLWlk...)
The graph worker decoded them to UUIDs (e.g., 550e8400-e29b-41d4-a716-446655440000) before sending to CCS
This meant the graph data was not human-readable and required an unnecessary decode step on every read
Other consumers of the graph (MCP, analytics) also had to decode

Solution

Write path (decode at ingestion)

Three transformers updated to decode base64 FIDOs to UUID before persisting to Neo4j:

File	Source
`internal/kafka/transformer.go`	Kafka factoid events (`fido.GetId()` is base64)
`internal/kafka/receipt_transformer.go`	Kafka receipt events (`item.GetFido()` is base64)
`internal/api/purchasehistory/transformer.go`	Purchase History API backfill (`purchases.Fido[i]` is base64)

In each transformer, fido.ToUUID(rawID) is called during the enrichment/collection step. ToUUID tries uuid.Parse first (to handle already-migrated UUIDs), then falls back to DecodeFIDO (base64 decode). If decoding fails for a particular FIDO, a warning is logged and that item is skipped (does not fail the entire batch). Decoded UUIDs are deduplicated before calling the FIDO enrichment service.

Additional safety guards on the write path:

Purchase History transformer: If all purchases have empty or undecodable FIDOs, returns an error instead of producing an empty graph — prevents the full-load cleanup from wiping existing data with nothing to replace it.
Factoid transformer: If all FIDOs fail to decode, returns nil, nil (non-fatal skip) instead of an error, since Kafka events are not full-load and no cleanup will run.
Factoid processor: Guards against nil graph data after transform to handle the skip.

Read path (backward-compatible)

A new fido.ToUUID(id) function was added to internal/api/fido/decoder.go. It:

Tries uuid.Parse(id) first — if the ID is already a UUID, returns it as-is
Falls back to fido.DecodeFIDO(id) — decodes base64 to UUID

Two read-path callers updated to use fido.ToUUID() instead of fido.DecodeFIDO():

File	Method
`internal/notification/repurchase/adapter.go`	`sendWithEnrichment()`
`internal/notification/repurchase/evaluate.go`	`enrichOnly()`, `enrichAndCheck()`

This means the read path works with both old base64 product IDs and new UUID product IDs in Neo4j, so there is no service disruption during the transition.

Rollout Plan

Phase 1: Deploy code (safe, no-downtime)

Deploy the updated consumer-graph-worker to stage, then prod. Immediately after deploy:

New data (from Kafka events) will be stored as UUIDs in Neo4j
Old data in Neo4j still has base64 product IDs
Read path handles both formats via fido.ToUUID() — no breakage

Risk — Duplicate Product nodes (Kafka only): Between deploy and backfill completion, new Kafka events write Product nodes keyed by UUID while old base64-keyed nodes remain. If the same product appears in both old and new events, two Product nodes exist temporarily. This only affects the small number of products that receive new Kafka events before that user’s backfill runs.

In practice, the impact is nearly transparent:

The old base64 Product node retains the full purchase history (times, avg_interval_days, repurchase_likelihood). The scheduler query (r.times >= 2) continues to find it as a repurchase candidate, and the read path normalizes its product_id via fido.ToUUID().
The new UUID Product node starts with times=1 and is filtered out by the scheduler’s r.times >= 2 threshold, so it has no effect on notifications.

The only edge case that loses signal: a user had exactly 1 prior purchase (base64 node, times=1), then repurchases via Kafka after deploy (new UUID node, times=1). Neither node hits times >= 2, so a real repurchase is missed until backfill merges them. This is a very narrow window affecting very few users.

Backfill timing: Run backfill promptly after deploy to close the times=1 gap. The system is fully functional in the meantime — there is no breakage or data loss, just a narrow window where some repurchase signals are missed.

Phase 2: Backfill existing users

Run the existing backfill tool to re-process all users:

# From consumer-graph-worker repo
go run cmd/backfill/main.go \
  --user-file prod-neo4j-user-ids.txt \
  --queue-url $SQS_QUEUE_URL \
  --rate-limit 10 \
  --lookback-days 365

This sends user_load_request messages to SQS. The worker re-fetches purchase history from the API and re-writes the graph with UUID product IDs.

Atomic cleanup: The backfill uses FullLoad=true, which triggers a new cleanup step in PersistUserGraph (added in writer.go). Before writing UUID-keyed products, the transaction deletes the user’s PURCHASED relationships to any base64-keyed Product nodes and removes orphaned Product nodes. Because this happens inside the same Neo4j transaction as the new writes, there is no window where data is duplicated or missing.

After backfill, all product IDs for that user are UUIDs. No manual cleanup needed.

Phase 3: Verify cleanup (optional, for confirmation)

The backfill handles cleanup atomically per user. After backfill is complete, verify that no base64-keyed Product nodes remain:

// Find Product nodes that look like base64 (not UUID format)
MATCH (p:Product)
WHERE NOT p.product_id =~ '(?i)^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$'
DETACH DELETE p

Verify first with a count query:

MATCH (p:Product)
WHERE NOT p.product_id =~ '(?i)^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$'
RETURN count(p)

Phase 4: Remove backward compatibility (future)

Once all data is confirmed migrated, the fido.ToUUID() fallback and fido.DecodeFIDO() can be simplified. The read path can just use uuid.Parse() directly. This is low priority since ToUUID() adds negligible overhead.

Files Changed

File	Change
`internal/api/fido/decoder.go`	Added `ToUUID()` helper
`internal/kafka/transformer.go`	Decode to UUID at ingestion, deduplicate UUIDs, non-fatal skip on all-fail
`internal/kafka/receipt_transformer.go`	Decode to UUID at ingestion, deduplicate UUIDs
`internal/api/purchasehistory/transformer.go`	Decode to UUID at ingestion, deduplicate UUIDs, empty graph guard
`internal/graph/writer.go`	Atomic base64 cleanup during full-load backfill
`internal/kafka/processor.go`	Nil graph data guard for non-fatal factoid skip
`internal/notification/repurchase/adapter.go`	Use `ToUUID()` instead of `DecodeFIDO()`
`internal/notification/repurchase/evaluate.go`	Use `ToUUID()` instead of `DecodeFIDO()`
`internal/kafka/transformer_test.go`	Updated test data to use base64 FIDOs, assert UUID outputs
`internal/kafka/receipt_transformer_test.go`	Updated test data to use base64 FIDOs, assert UUID outputs
`internal/kafka/processor_test.go`	Mock responses keyed by UUID, updated assertions
`internal/kafka/receipt_processor_test.go`	Mock responses keyed by UUID, updated assertions
`internal/api/purchasehistory/transformer_test.go`	Updated test data, added empty-FIDO guard test
`internal/notification/repurchase/adapter_integration_test.go`	Use `ToUUID()` in tests

Testing

purchasehistory/transformer_test.go — all tests pass (verified locally)
kafka/transformer_test.go — syntax-verified (gofmt), cannot run locally due to Buf CSR auth
kafka/receipt_transformer_test.go — syntax-verified, same auth blocker
notification/repurchase/ — syntax-verified, same auth blocker
Full test suite should be validated in CI (GitHub Actions has the required credentials)