Consumer Graph — Neo4j Schema Reference

Consumer Graph — Neo4j Schema Reference

Complete schema of the Neo4j knowledge graph, with every property mapped to its upstream data source.

---

Nodes

User

Property	Type	Set On	Data Source
user_id	string	CREATE	Factoid: factoid.GetUserId() / Receipt: receipt.GetUserId()
timezone	string	CREATE; COALESCE on MATCH	Hardcoded "UTC" from both transformers. Falls back to "America/Chicago" at write time if empty.
created_at	datetime	CREATE only	Neo4j datetime() — server clock at first write
last_updated_at	datetime	MATCH only	Neo4j datetime() — server clock on subsequent writes

Product

Property	Type	Set On	Data Source
product_id	string	CREATE	Kafka message: FIDO ID from factoid data.fidos[].id or receipt item.GetFido()
name	string	CREATE; COALESCE on MATCH	FIDO Product Service name field, fallback description field, fallback Receipt item.GetDescription(), final fallback "Product {fidoID}"
brand	string	CREATE; COALESCE on MATCH	FIDO Product Service brand field, fallback "ENRICHMENT_UNAVAILABLE"
category	string	CREATE; COALESCE on MATCH	FIDO Product Service category field, fallback "UNCATEGORIZED"
created_at	datetime	CREATE only	Neo4j datetime()

Category

Property	Type	Set On	Data Source
category_id	string	CREATE	Derived: "CAT_" + normalized(product.category) — lowercase, non-alphanumeric replaced with _ (see known issue)
name	string	CREATE only	FIDO Product Service category field (via product enrichment), fallback "UNCATEGORIZED"
product_count	int	CREATE only	Hardcoded 0 — set on CREATE, never incremented

Retailer

Property	Type	Set On	Data Source
retailer_id	string	CREATE	Derived: "RET_" + storeName
name	string	CREATE only	Factoid: factoid.data.retailer_info.name / Receipt: receipt.GetStoreName()
venue_type	string	CREATE only	Factoid: hardcoded "Grocery". Receipt: Retailer Service GetRetailerByStoreName() normalized to one of: Grocery, Drugstore, Warehouse, Convenience, Big Box; defaults to "Grocery"

Community

Property	Type	Set On	Data Source
community_id	string	CREATE	Derived: "COMM_{categoryID}_{zipCode}" — built from user’s top categories + zip
name	string	CREATE only	Derived: "{categoryName} Shoppers - {zipCode}"
member_count	int	CREATE=1; +1 on MATCH	Computed: incremented each time any user is assigned
primary_category	string	CREATE only	FIDO Product Service category name (via the category the community is built around)
zip_code	string	CREATE only	Receipt: receipt.GetZip(). Factoid path provides empty string (factoids don’t carry zip codes).

---

Relationships

PURCHASED (User -> Product)

Property	Type	Set On	Data Source
timestamps	datetime[]	CREATE=[ts]; append on MATCH	Factoid: factoid.GetOccurredAt() / Receipt: receipt.GetPurchaseDate(), fallback receipt.GetScanDate()
receipt_ids	string[]	CREATE=[id]; append on MATCH	Factoid: factoid.GetSummaryId() / Receipt: receipt.GetId()
times	int	CREATE=1; +1 on MATCH	Computed: running count of writes
last	datetime	CREATE=ts; max(current, new) on MATCH	Factoid: occurred_at / Receipt: purchase_date or scan_date
avg_interval_days	float	CREATE=0; recalculated on MATCH	Computed in Cypher: (last_timestamp - first_timestamp) / (count - 1) using the timestamps array
repurchase_likelihood	float	CREATE=0.0; recalculated on MATCH	Computed in Cypher: 1.0 / (1.0 + days_since_last / (avg_interval + 1)) when times >= 2, else 0.5

IN_CATEGORY (Product -> Category)

Property	Type	Data Source
(none)	—	Derived: links product to its category. Category ID derived from Product.category at write time.

SHOPS_IN (User -> Category)

Property	Type	Set On	Data Source
purchase_count	int	CREATE=n; +=n on MATCH	Computed in transformer: count of items in this category within a single Kafka message

SHOPS_AT (User -> Retailer)

Property	Type	Set On	Data Source
frequency	int	CREATE=visits; +=visits on MATCH	Hardcoded 1 per event (both transformers set Visits: 1)
last_visit	datetime	CREATE=ts; max(current, new) on MATCH	Factoid: factoid.GetOccurredAt() / Receipt: purchase_date or scan_date

MEMBER_OF (User -> Community)

Property	Type	Data Source
(none)	—	Derived: assigned by CommunityAssigner from user’s top 1-3 categories by purchase_count + zip code

---

Data Sources

Source	Transport	What It Provides
Kafka factoid-stream	Protobuf via Buf CSR, {env}-factoids cluster	user_id, summary_id, occurred_at, FIDO IDs (with CENTS/COUNT), retailer_info.name
Kafka receipt-events	Plain protobuf, {env}-receipt-infra cluster	user_id, receipt_id, purchase_date/scan_date, store_name, zip, item FIDOs + descriptions
FIDO Product Service	HTTP batch API (up to 1000 FIDOs, 30s timeout, 3 retries)	name, description, brand, category, barcode per FIDO ID. Note: barcode is fetched but never persisted to the graph.
Retailer Service	HTTP with 30-min cache	RewardsStoreType per store name, normalized to venue_type. Only used by receipt path; factoid path hardcodes "Grocery".
Neo4j server clock	Internal	created_at, last_updated_at on nodes. avg_interval_days and repurchase_likelihood computed in Cypher at write time.
Derived / hardcoded	Internal	Category IDs (CAT_*), Retailer IDs (RET_*), Community IDs (COMM_*), community names, product_count=0, default timezone "America/Chicago".

---

Read Queries

Repurchase Candidates (Scheduler)

The notification scheduler reads from the graph to find users likely to repurchase:

MATCH (u:User)-[r:PURCHASED]->(p:Product)
WHERE r.times >= 2
  AND p.category <> 'Alcohol'
  AND r.last >= datetime() - duration({days: $lookbackDays})
  AND r.avg_interval_days > 0
RETURN
  u.user_id, u.timezone,
  p.product_id, p.name, p.brand, p.category,
  r.times, r.avg_interval_days, r.last,
  r.repurchase_likelihood, r.timestamps
ORDER BY r.repurchase_likelihood DESC
LIMIT 10000

Filters: 2+ purchases, not Alcohol, last purchase within lookback window (default 90 days), has a computed avg interval.

Returned fields consumed by scheduler: timestamps array is used for time-of-day analysis to determine optimal notification delivery time. timezone is used to convert to local time.

---

Known Issues

Known issue: categoryIDFromName mismatch

There are two different implementations of categoryIDFromName:

• internal/kafka/transformer.go:238 — normalizes: lowercase, replaces [^a-z0-9]+ with _, trims underscores. Example: "Meat & Seafood" becomes CAT_meat_seafood.
• internal/graph/writer.go:383 — no normalization: fmt.Sprintf("CAT_%s", name). Example: "Meat & Seafood" becomes CAT_Meat & Seafood.

The transformer creates CategoryData with the normalized ID, but linkProductsToCategories in the writer re-derives the category ID from the raw Product.Category name using the writer’s non-normalizing version. This means the IN_CATEGORY MATCH may fail to find the category node that was created with the normalized ID.

Unbounded array growth on PURCHASED

The timestamps and receipt_ids arrays on PURCHASED relationships grow without bound — every new purchase appends. For frequently-purchased products this can become large over time.

Community member_count only increments

Community.member_count increments on every user assignment but never decrements if a user’s category preferences shift. The count drifts upward over time.

Indexes

Indexes are created at worker boot via Repository.EnsureIndexes in internal/graph/writer.go (idempotent via IF NOT EXISTS):

Index	Type	On
user_id_idx	RANGE	(:User).user_id
product_id_idx	RANGE	(:Product).product_id
category_id_idx	RANGE	(:Category).category_id
retailer_id_idx	RANGE	(:Retailer).retailer_id
community_id_idx	RANGE	(:Community).community_id
purchased_last_idx	RANGE	()-[:PURCHASED]-().last — added so the scheduler’s repurchase-candidate query can range-seek the time window instead of post-filter scanning. See internal/notification/repurchase/neo4j.go GetRepurchaseCandidatesBatch.

No CREATE CONSTRAINT statements exist in the codebase.