Skip to main content

Audit Trails with Immutable Events and Derived State

Current-state records are useful for answering "what is true now?". They are not useful for answering "what happened, and when, and who did it?"

An audit trail separates the two concerns. Events are immutable — append-only records that capture intent, actor, and timestamp. The current state record is derived from events but is mutable. Both live in the graph; only events form the log.

Graph shape​

LabelWhat it represents
ENTITYThe mutable record reflecting current state
EVENTAn immutable fact: what changed, when, and who triggered it
ACTORThe user or service that performed the action

Step 1: Create an entity with its first event atomically​

Use a transaction to guarantee entity creation and the corresponding created event land together or not at all.

import json, os
from datetime import datetime, timezone
from rushdb import RushDB

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


def create_order_with_audit(order_data: dict, actor_id: str):
    tx = db.transactions.begin()
    try:
        order = db.records.create("ORDER", {**order_data, "status": "pending"}, transaction=tx)

        event = db.records.create("EVENT", {
            "type": "created",
            "actorId": actor_id,
            "entityId": order.id,
            "occurredAt": datetime.now(timezone.utc).isoformat(),
            "snapshot": json.dumps(order_data)
        }, transaction=tx)

        db.records.attach(order.id, event.id, {"type": "HAS_EVENT", "direction": "out"}, transaction=tx)

        db.transactions.commit(tx)
        return order
    except Exception as e:
        db.transactions.rollback(tx)
        raise


order = create_order_with_audit(
    {"customerId": "cust-42", "amount": 299.99, "currency": "USD"},
    "user-101"
)

Step 2: Record a state change with its audit event​

When the order status changes, update the entity and append a new EVENT — all in one transaction.

def change_status(order_id: str, new_status: str, actor_id: str, reason: str | None = None):
    current = db.records.find({"labels": ["ORDER"], "where": {"__id": order_id}})
    prev_status = current.data[0].get("status") if current.data else None

    tx = db.transactions.begin()
    try:
        db.records.update(order_id, {"status": new_status}, transaction=tx)

        event = db.records.create("EVENT", {
            "type": "status_changed",
            "actorId": actor_id,
            "entityId": order_id,
            "from": prev_status,
            "to": new_status,
            "reason": reason,
            "occurredAt": datetime.now(timezone.utc).isoformat()
        }, transaction=tx)

        db.records.attach(
            current.data[0].id,
            event.id,
            {"type": "HAS_EVENT", "direction": "out"},
            transaction=tx
        )
        db.transactions.commit(tx)
    except Exception as e:
        db.transactions.rollback(tx)
        raise


change_status(order.id, "shipped", "service-fulfillment", "Dispatched from warehouse")

Step 3: Query the full event history​

Retrieve all events for an entity ordered by occurrence time.

history = db.records.find({
    "labels": ["EVENT"],
    "where": {
        "ORDER": {
            "$relation": {"type": "HAS_EVENT", "direction": "in"},
            "__id": order.id
        }
    },
    "orderBy": {"occurredAt": "asc"}
})

for event in history.data:
    print(f"[{event.data.get('occurredAt')}] {event.data.get('type')} by {event.data.get('actorId')}")
    if event.data.get("from"):
        print(f"  {event.data['from']} → {event.data['to']}")

Step 4: Select expressions for a compliance report​

Count events by type and actor across a time window.

report = db.records.find({
    "labels": ["EVENT"],
    "where": {
        "occurredAt": {"$gte": "2025-01-01", "$lte": "2025-03-31"},
        "type": "status_changed"
    },
    "select": {
        "count":   {"$count": "*"},
        "actorId": "$record.actorId"
    },
    "groupBy": ["actorId", "count"],
    "orderBy": {"count": "desc"}
})

print("State changes by actor (Q1):")
for row in report.data:
    print(f"  {row.data.get('actorId')}: {row.data.get('count')}")

Design rules for immutable audit trails​

  1. Never update or delete EVENT records — they are the immutable log; treat them as write-once
  2. Always write entity update + event in a single transaction — no partial audit trails
  3. Store from and to on every state-change event — makes reconstruction possible without replaying all prior events
  4. Store actorId on every event — automated services have service IDs, not just human users
  5. Store occurredAt as an ISO 8601 string — enables $gte/$lte filtering on dates

Production caveat​

Audit trails grow with every write. For high-write systems (thousands of events per hour), plan for periodic archival of old events. A safe pattern: copy events older than 90 days into a separate RushDB project (archive-{year}) and mark them as archived in the source project. This preserves queryability while keeping the primary project lean.

Next steps​