Skip to main content

Customer 360 as a Connected Graph

Customer context typically lives in five or more separated systems: CRM, billing, support, product analytics, and marketing automation. Getting a complete picture of one customer requires manual cross-referencing across all of them.

A connected graph collapses that. Put every customer-related record into RushDB — accounts, subscriptions, invoices, touchpoints, support tickets, feature usage — and you can answer complex, cross-domain questions with one query.

Graph shape​

LabelWhat it represents
USERAn individual user with login/profile data
ACCOUNTAn organization or company
SUBSCRIPTIONAn active or cancelled plan
INVOICEA billing invoice
TOUCHPOINTA marketing or sales interaction (email, call, demo)
SUPPORT_TICKETAn inbound support request
FEATURE_USAGEA log of which product features are used and how often

Step 1: Ingest account and user data​

from rushdb import RushDB
import os

db = RushDB(os.environ["RUSHDB_API_KEY"], base_url="https://api.rushdb.com/api/v1")

account = db.records.create("ACCOUNT", {
    "name": "Acme Corp",
    "plan": "enterprise",
    "region": "EU",
    "mrr": 1200,
    "createdAt": "2024-01-15"
})

db.records.import_json({
    "label": "USER",
    "data": [
        {"email": "alice@acme.com", "name": "Alice", "role": "admin", "lastActiveAt": "2025-03-01"},
        {"email": "bob@acme.com",   "name": "Bob",   "role": "member", "lastActiveAt": "2025-02-20"}
    ]
})

users = db.records.find({
    "labels": ["USER"],
    "where": {"email": {"$in": ["alice@acme.com", "bob@acme.com"]}}
})

for user in users.data:
    db.records.attach(user.id, account.id, {"type": "BELONGS_TO", "direction": "out"})

Step 2: Add subscription and invoice history​

subscription = db.records.create("SUBSCRIPTION", {
    "plan": "enterprise",
    "status": "active",
    "startedAt": "2024-01-15",
    "renewsAt": "2026-01-15",
    "seats": 25
})

db.records.attach(account.id, subscription.id, {"type": "HAS_SUBSCRIPTION", "direction": "out"})

db.records.import_json({
    "label": "INVOICE",
    "data": [
        {"month": "2025-01", "amount": 1200, "status": "paid", "paidAt": "2025-01-05"},
        {"month": "2025-02", "amount": 1200, "status": "paid", "paidAt": "2025-02-05"},
        {"month": "2025-03", "amount": 1200, "status": "overdue", "paidAt": None}
    ]
})

invoices = db.records.find({"labels": ["INVOICE"]})
for invoice in invoices.data:
    db.records.attach(subscription.id, invoice.id, {"type": "HAS_INVOICE", "direction": "out"})

Step 3: Query accounts with overdue invoices​

Find enterprise accounts that have at least one overdue invoice — the churn-risk signal.

overdue_accounts = db.records.find({
    "labels": ["ACCOUNT"],
    "where": {
        "plan": "enterprise",
        "SUBSCRIPTION": {
            "$relation": {"type": "HAS_SUBSCRIPTION", "direction": "out"},
            "INVOICE": {
                "$relation": {"type": "HAS_INVOICE", "direction": "out"},
                "status": "overdue"
            }
        }
    }
})

print(f"Accounts with overdue invoices: {overdue_accounts.total}")
for acct in overdue_accounts.data:
    print(f"  {acct.get('name')} — MRR: {acct.get('mrr')}")

Step 4: MRR by region (select expressions)​

mrr_by_region = db.records.find({
    "labels": ["ACCOUNT"],
    "where": {
        "SUBSCRIPTION": {
            "$relation": {"type": "HAS_SUBSCRIPTION", "direction": "out"},
            "status": "active"
        }
    },
    "select": {
        "totalMrr": {"$sum": "$record.mrr"},
        "region": "$record.region"
    },
    "groupBy": ["region", "totalMrr"],
    "orderBy": {"totalMrr": "desc"}
})

for row in mrr_by_region.data:
    print(f"{row.data.get('region')}: ${row.data.get('totalMrr')}")

Step 5: Full context for a support ticket​

When a support ticket comes in, retrieve the full account context automatically — plan, MRR, open invoices, and recent touchpoints — so the agent or support rep starts with the complete picture.

from concurrent.futures import ThreadPoolExecutor


def get_account_context(account_id: str) -> dict:
    with ThreadPoolExecutor(max_workers=4) as executor:
        fut_acct  = executor.submit(db.records.find, {"labels": ["ACCOUNT"], "where": {"__id": account_id}})
        fut_sub   = executor.submit(db.records.find, {"labels": ["SUBSCRIPTION"], "where": {"ACCOUNT": {"$relation": {"type": "HAS_SUBSCRIPTION", "direction": "in"}, "__id": account_id}, "status": "active"}})
        fut_inv   = executor.submit(db.records.find, {"labels": ["INVOICE"], "where": {"SUBSCRIPTION": {"$relation": {"type": "HAS_INVOICE", "direction": "in"}, "ACCOUNT": {"$relation": {"type": "HAS_SUBSCRIPTION", "direction": "in"}, "__id": account_id}}, "status": "overdue"}})
        fut_tick  = executor.submit(db.records.find, {"labels": ["SUPPORT_TICKET"], "where": {"USER": {"$relation": {"type": "CREATED", "direction": "in"}, "ACCOUNT": {"$relation": {"type": "BELONGS_TO", "direction": "in"}, "__id": account_id}}}, "orderBy": {"createdAt": "desc"}, "limit": 5})

    return {
        "account": fut_acct.result().data[0].data if fut_acct.result().data else None,
        "activeSubscription": fut_sub.result().data[0].data if fut_sub.result().data else None,
        "overdueCount": fut_inv.result().total,
        "recentTickets": [t.data for t in fut_tick.result().data]
    }

Production caveat​

Customer graphs grow with every activity event. Feature usage logs and touchpoints can reach millions of records for enterprise accounts. Use limit and orderBy on time-sorted fields (occurredAt, createdAt) to bound retrieval, and compute feature usage counts with select/groupBy rather than surfacing every raw log entry.

Next steps​