Play Data in Consumer Graph (Neo4j)

Play Data in Consumer Graph (Neo4j)

1. Problem Statement

The consumer graph in Neo4j currently models purchase history and repurchase candidates but has no representation of Play activity. This means cross-domain intelligence — “users who play Coin Master also buy X”, Play engagement as a signal for repurchase nudges, or identifying high-value users across both shopping and gaming — is not possible.

Play events (POINTS_AWARDED, APP_INSTALL) are already published to SNS by gameplay-service, and multiple consumers (recommendations, battlepass, play-support-service) already subscribe. Adding the consumer graph as another subscriber follows an established pattern.

2. Background & Context

2.1 Current State

Consumer Graph (Neo4j):

• Nodes: User, Product (FIDO), Retailer, Category, Brand
• Edges: PURCHASED, PURCHASED_AT, IN_CATEGORY, BY_BRAND
• consumer-graph-worker writes purchase data from Snowflake/Purchase History Service
• CCS reads repurchase candidates via internal/clients/neo4j.go

Play Event Surface (from RFD):

• SNS topic {env}-gameplay-service-events emits POINTS_AWARDED and APP_INSTALL
• 5 existing subscribers (push notifications, battlepass, recommendations, high-value tracking, referrals)
• play-support-service already consumes these events into DynamoDB

2.2 Key Decisions Already Made

1. REST-first for Play context — 001-play-context uses REST polling for initial Play integration. Graph integration adds the event-driven path for cross-domain queries.
2. consumer-graph-worker is the graph writer — All Neo4j writes go through consumer-graph-worker, not CCS directly. CCS is read-only.
3. SNS subscription pattern — consumer-graph-worker will subscribe to the same SNS topic play-support-service uses, with the same filter policy.

2.3 Glossary

Term	Definition
consumer-graph-worker	Service that writes to Neo4j — purchase history, repurchase candidates, and (proposed) Play data
POINTS_AWARDED	SNS event emitted when a user completes a game milestone and earns points
APP_INSTALL	SNS event emitted when a user installs a game via Adjoe
Cross-domain query	Graph query that spans both purchase and Play data (e.g. “Play users who also buy at Target”)

3. Requirements

3.1 Functional Requirements

1. The system MUST consume POINTS_AWARDED events from {env}-gameplay-service-events SNS topic and write game milestone completions to Neo4j.
2. The system MUST consume APP_INSTALL events and write game install relationships to Neo4j.
3. The system MUST create Game nodes with properties: appId, appName, platform, appCategory.
4. The system MUST create PLAYS edges from User to Game with temporal properties following the existing PURCHASED edge pattern: timestamps (datetime[]), transaction_ids (string[]), points_earned (int[]), times (int), last (datetime), avg_interval_days (float), and play_engagement (float, computed).
5. The system MUST append milestone completion timestamps to the timestamps array on each POINTS_AWARDED event, using transaction_ids for deduplication (same pattern as receipt_ids on PURCHASED edges).
6. The system MUST use a two-phase write strategy: Phase 1 appends raw event data in batches, Phase 2 recomputes aggregates (avg_interval_days, play_engagement) from the final timestamp arrays using reduce() min/max (not positional access).
7. The system MUST cap the timestamps array at 50 most recent entries to prevent unbounded growth (Play generates more events per user than purchases).
8. The system MUST support idempotent writes — replaying the same event (same transaction_id) MUST NOT modify the graph.
9. The system SHOULD support a 90-day lookback filter for Play activity queries (same pattern as repurchase candidates).
10. CCS MUST expose graph-based Play queries via the existing Neo4j client pattern.
11. The system MAY track mini-game streak data as a property on the User node (currentStreakDays, maxStreakDays) if mini-game events become available on SNS.

3.2 Non-Functional Requirements

1. Event processing MUST handle at least 100 events/second sustained (peak Play activity).
2. Neo4j writes MUST be idempotent — replaying events MUST NOT create duplicate nodes or edges.
3. Graph query latency for Play-related queries MUST be < 500ms at p95.
4. Event processing lag MUST be < 30 seconds from SNS publish to Neo4j write at p95.

3.3 Acceptance Criteria

• AC-1: Given a POINTS_AWARDED event for a new game, when processed, then a Game node and User -[:PLAYS]-> Game edge are created in Neo4j.
• AC-2: Given a POINTS_AWARDED event for an existing game, when processed, then the timestamp is appended to PLAYS.timestamps, the transaction_id is appended to PLAYS.transaction_ids, times is incremented, and last is updated.
• AC-3: Given an APP_INSTALL event, when processed, then a Game node and PLAYS edge with installDate and initial timestamps array are created.
• AC-3.1: Given multiple POINTS_AWARDED events processed, when aggregates are recomputed (Phase 2), then avg_interval_days is computed as (latest - earliest) / (count - 1) and play_engagement is computed from the temporal distribution.
• AC-4: Given a user with both purchase history and Play activity in the graph, when a cross-domain query is executed, then results include both domains.
• AC-5: Given the same event replayed twice, when processed, then the graph state is identical to processing it once (idempotent).
• AC-6: Given a query for Play activity within 90 days, when executed, then only recent activity is returned.

4. Solution Design

4.1 Architecture

gameplay-service → SNS ({env}-gameplay-service-events)
                      │
                      ├── SQS (play-support-in-app-event-queue) → play-support-service
                      ├── SQS (battlepass-service-play-events-queue) → battlepass-service
                      ├── SQS (recommendations-play-engagement) → recs-orchestrator
                      └── SQS (consumer-graph-play-events-queue) → consumer-graph-worker [NEW]
                                                                        │
                                                                        ▼
                                                                     Neo4j
                                                                        │
                                                                        ▼
                                                                CCS (read queries)

4.2 Data Model

Following the existing temporal pattern from PURCHASED edges in consumer-graph-worker:

// Nodes
(:User {user_id: string})
(:Game {
  app_id: string,
  app_name: string,
  platform: string,
  app_category: string
})

// Edges — temporal-aware, following PURCHASED edge pattern + validity windows
(:User)-[:PLAYS {
  // === Temporal arrays (append-only, capped at 50) ===
  timestamps: datetime[],       // Milestone completion times (parallel to transaction_ids)
  transaction_ids: string[],    // Adjoe trans_uuid for deduplication (parallel to timestamps)
  points_earned: int[],         // Points per event (parallel to timestamps)

  // === Running counters (survive array truncation) ===
  times: int,                   // Total milestone completions
  total_points: int,            // Sum of all points earned

  // === Temporal aggregates (recomputed in Phase 2) ===
  last: datetime,               // Most recent activity
  first: datetime,              // First activity (install or first milestone)
  avg_interval_days: float,     // (latest - earliest) / (count - 1)
  play_engagement: float,       // 1.0 / (1.0 + days_since_last / avg_interval) — mirrors repurchase_likelihood

  // === Validity window (game offer lifecycle) ===
  valid_at: datetime,           // When game offer became available to user (from APP_INSTALL)
  invalid_at: datetime,         // When game offer actually expired/completed (null if still active)
  expires_at: datetime,         // Scheduled expiration from Adjoe (OfferExpiresAt field)
  status: string,               // "active", "completed", "expired"
  completed_at: datetime,       // When all milestones completed (null if not yet)

  // === Static metadata ===
  install_date: datetime,       // From APP_INSTALL event
  is_active: boolean            // Quick filter for current games (denormalized from status)
}]->(:Game)

Design rationale:

Timestamp arrays (event history):

• timestamps[] / transaction_ids[] / points_earned[] are parallel arrays (same pattern as timestamps[] / receipt_ids[] on PURCHASED edges)
• transaction_ids enables idempotent deduplication (same role as receipt_ids)
• play_engagement uses the same formula as repurchase_likelihood on PURCHASED edges — enables cross-domain scoring (“users who both buy frequently AND play frequently”)
• Arrays capped at 50 (vs unbounded on PURCHASED) because Play generates more events per user
• avg_interval_days enables “how often does this user play?” queries for engagement scoring

Validity windows (offer lifecycle):

• valid_at / invalid_at / expires_at follow the rewards layer pattern from consumer-graph-mcp (REWARDS_SCHEMA_DESIGN.md)
• expires_at is the scheduled expiration from Adjoe (OfferExpiresAt on the rich install payload); invalid_at is when it actually became invalid (may differ if completed early or expired naturally)
• is_active is a denormalized shortcut for fast filtering — must be kept in sync on status transitions
• Status transitions are non-lossy: completing or expiring a game sets invalid_at/completed_at but never deletes the edge

Temporal classification (per OpenAI Cookbook pattern):

PLAYS edge state	Classification	valid_at	invalid_at
User actively playing game	Dynamic	Install time	null
User completed all milestones	Static	Install time	Completion time
Game offer expired	Static	Install time	Expiration time

4.3 API Contracts

CCS would expose graph-based Play queries through the existing Neo4j client:

// Get user's Play activity from graph (cross-domain capable)
GetPlayGraph(ctx, userId) (*PlayGraphData, error)

// Cross-domain: users who play game X and also purchase at retailer Y
GetPlayAndPurchaseOverlap(ctx, appId, retailerId) ([]UserOverlap, error)

4.4 Key Implementation Details

SQS Subscription Filter Policy:

{
  "EventType": ["POINTS_AWARDED", "APP_INSTALL"]
}

Phase 1: Append Raw Data (Batched)

UNWIND $events AS evt
MATCH (u:User {user_id: evt.userId})
MERGE (g:Game {app_id: evt.appId})
ON CREATE SET g.app_name = evt.appName, g.platform = evt.platform, g.app_category = evt.appCategory
ON MATCH SET g.app_name = evt.appName

MERGE (u)-[r:PLAYS]->(g)
ON CREATE SET
    r.timestamps = [datetime(evt.timestamp)],
    r.transaction_ids = [evt.transactionId],
    r.points_earned = [evt.points],
    r.times = 1,
    r.total_points = evt.points,
    r.last = datetime(evt.timestamp),
    r.first = datetime(evt.timestamp),
    r.install_date = CASE WHEN evt.eventType = 'APP_INSTALL' THEN datetime(evt.timestamp) ELSE null END,
    r.avg_interval_days = 0,
    r.play_engagement = 0.0,
    // Validity window
    r.valid_at = datetime(evt.timestamp),
    r.invalid_at = null,
    r.expires_at = CASE WHEN evt.expiresAt IS NOT NULL THEN datetime(evt.expiresAt) ELSE null END,
    r.status = 'active',
    r.completed_at = null,
    r.is_active = true
ON MATCH SET
    r.transaction_ids = CASE
        WHEN evt.transactionId IN r.transaction_ids THEN r.transaction_ids
        ELSE (r.transaction_ids + evt.transactionId)[-50..]
    END,
    r.timestamps = CASE
        WHEN evt.transactionId IN r.transaction_ids THEN r.timestamps
        ELSE (r.timestamps + datetime(evt.timestamp))[-50..]
    END,
    r.points_earned = CASE
        WHEN evt.transactionId IN r.transaction_ids THEN r.points_earned
        ELSE (r.points_earned + evt.points)[-50..]
    END,
    r.times = CASE
        WHEN evt.transactionId IN r.transaction_ids THEN r.times
        ELSE r.times + 1
    END,
    r.total_points = CASE
        WHEN evt.transactionId IN r.transaction_ids THEN r.total_points
        ELSE r.total_points + evt.points
    END,
    r.last = CASE
        WHEN NOT evt.transactionId IN r.transaction_ids AND datetime(evt.timestamp) > r.last
        THEN datetime(evt.timestamp)
        ELSE r.last
    END,
    r.install_date = CASE
        WHEN evt.eventType = 'APP_INSTALL' AND r.install_date IS NULL
        THEN datetime(evt.timestamp)
        ELSE r.install_date
    END

Phase 2: Recompute Aggregates

UNWIND $appIDs AS aid
MATCH (u:User {user_id: $userID})-[r:PLAYS]->(g:Game {app_id: aid})
WHERE size(r.timestamps) >= 2
WITH r,
     reduce(mn = r.timestamps[0], t IN r.timestamps | CASE WHEN t < mn THEN t ELSE mn END) AS earliest,
     reduce(mx = r.timestamps[0], t IN r.timestamps | CASE WHEN t > mx THEN t ELSE mx END) AS latest
SET r.first = earliest,
    r.last = latest,
    r.avg_interval_days = toFloat(duration.inDays(earliest, latest).days) / (size(r.timestamps) - 1),
    r.play_engagement = CASE
        WHEN duration.inDays(earliest, latest).days > 0
        THEN 1.0 / (1.0 + toFloat(duration.inDays(latest, datetime()).days) /
             (toFloat(duration.inDays(earliest, latest).days) / (size(r.timestamps) - 1) + 1))
        ELSE 0.0
    END

Lifecycle Transitions (non-lossy):

// Game completed: user finished all milestones
MATCH (u:User {user_id: $userId})-[r:PLAYS]->(g:Game {app_id: $appId})
SET r.status = 'completed',
    r.completed_at = datetime(),
    r.invalid_at = datetime(),
    r.is_active = false

// Game expired: offer time ran out
MATCH (u:User)-[r:PLAYS]->(g:Game)
WHERE r.status = 'active'
  AND r.expires_at IS NOT NULL
  AND r.expires_at <= datetime()
SET r.status = 'expired',
    r.invalid_at = datetime(),
    r.is_active = false

Point-in-Time Query (as_of):

// What games was this user actively playing on a specific date?
MATCH (u:User {user_id: $userId})-[r:PLAYS]->(g:Game)
WHERE r.valid_at <= datetime($asOf)
  AND (r.invalid_at IS NULL OR r.invalid_at > datetime($asOf))
RETURN g.app_name, r.total_points, r.times, r.status

// What's currently active? (fast path)
MATCH (u:User {user_id: $userId})-[r:PLAYS {is_active: true}]->(g:Game)
RETURN g.app_name, r.total_points, r.play_engagement, r.expires_at,
       duration.inDays(datetime(), r.expires_at).days AS days_remaining
ORDER BY r.play_engagement DESC

Why two phases? Same reason as PURCHASED edges — Phase 1 appends raw data in batches (handles memory limits), Phase 2 recomputes aggregates after all data is written. Prevents incorrect interval calculations from out-of-order batch arrivals. Uses reduce() min/max (not positional access) for robustness.

Array Capping: [-50..] slice keeps only the 50 most recent entries. Play generates more events per user than purchases (10-20 milestones per game × multiple games), so we cap more aggressively than PURCHASED (which is currently unbounded).

Cross-Domain Query Example:

// Users who actively play AND frequently repurchase — high engagement candidates
MATCH (u:User)-[plays:PLAYS {is_active: true}]->(g:Game)
MATCH (u)-[buys:PURCHASED]->(p:Product)
WHERE plays.play_engagement > 0.5
  AND buys.repurchase_likelihood > 0.5
  AND plays.last >= datetime() - duration({days: 90})
RETURN u.user_id,
       plays.play_engagement, plays.total_points, g.app_name,
       plays.expires_at,
       buys.repurchase_likelihood, p.name
ORDER BY plays.play_engagement + buys.repurchase_likelihood DESC

5. Dependencies

5.1 Spec Dependencies

Spec ID	What We Need From It	Why
001-play-context	Play domain types, upstream API understanding, event schema documentation	Graph integration builds on the same Play data model

5.2 External Dependencies

Dependency	Owner	Status	Blocker?
gameplay-service SNS topic	RAPPS collective	Live	No
consumer-graph-worker repo	Pilot team	Live	No
Neo4j cluster (stage/prod)	Pilot team / Platform	Live	No
SQS queue creation	Platform / SRE	Needs provisioning	Yes — need new queue

6. Risks & Open Questions

#	Risk / Question	Impact	Mitigation / Answer
1	Event volume — Play may generate more events than purchase history	Neo4j write pressure, SQS lag	Monitor event rates in stage; implement batch writes if needed
2	Timestamp array growth — Play generates 10-20 milestones per game × multiple games per user	Query performance, storage pressure	Cap arrays at 50 entries using [-50..] slice; total_points and times remain accurate as running counters outside the array
3	No mini-game events on SNS currently	Cannot track streaks via events	Use REST polling from 001-play-context for streak data; add event-driven when available
4	Historical backfill	Graph starts empty for existing Play users	Out of scope for v1; can backfill from Snowflake tables (GAMEPLAY_TRANSACTION_STAGE, GAMEPLAY_INSTALL_STAGE) later
5	Cross-repo changes	consumer-graph-worker and CCS changes must coordinate	Sequence: graph-worker first (writes), then CCS (reads)
6	What cross-domain queries does the Rewards Assistant actually need?	May build wrong graph schema	Validate query patterns with Rewards Assistant team before implementing

7. Testing Strategy

• Unit tests: Event parsing, Cypher query generation, idempotent write logic, milestone threshold filtering.
• Integration tests: Neo4j testcontainer with real event payloads → verify graph state. Replay events to verify idempotency.
• Contract tests: Validate SNS event schema against gameplay-service published schema.
• Manual validation: Subscribe to staging SNS topic, process real events, query Neo4j to verify graph state.

8. Rollout & Observability

8.1 Rollout Plan

1. Provision SQS queue with SNS subscription and filter policy in staging
2. Deploy consumer-graph-worker changes to staging
3. Monitor event processing lag and Neo4j write performance
4. Add CCS read queries once graph is populated
5. Deploy to production

8.2 Metrics & Alerts

• Event processing: Messages received, processed, failed, DLQ’d per minute
• Processing lag: Time from SNS publish to Neo4j write (p50/p95/p99)
• Neo4j writes: Write latency, error rate
• Graph size: Node count (Game), edge count (PLAYS, COMPLETED_MILESTONE) — track growth rate
• Query performance: CCS Play graph query latency

8.3 Rollback Plan

1. Disable SQS subscription (stop consuming events)
2. Remove CCS Play graph query endpoints
3. Graph data is inert — no need to delete nodes/edges unless storage is a concern

9. Appendix

SNS Event Schemas

See RFD Play Event Surface for full POINTS_AWARDED and APP_INSTALL event schemas.

Existing Play Event Consumers

Consumer	Queue	Purpose
play-support-service	play-support-in-app-event-queue	Materialized view for support dashboards
battlepass-service	battlepass-service-play-events-queue	Battle pass progression
recommendations-orchestrator	recommendations-play-engagement	Personalization input
push notifications	gameplay-push-notification-queue	User notifications

Snowflake Tables for Future Backfill

Table	Use
GAMEPLAY_TRANSACTION_STAGE	Historical milestone completions
GAMEPLAY_INSTALL_STAGE	Historical game installs
GAMEPLAY_USER_STAGE	Lifetime point totals per user