Record Triggers & Event Lifecycle

Record triggers allow developers to inject custom business logic, calculations, data normalization, and integrations directly into Valstorm's Create, Update, and Delete (CUD) database request lifecycle.

Triggers execute within a secure, sandboxed environment and receive a RecordTriggerContext object containing the modified dataset, diff change detection, and unified access to the Platform Context.

1. Trigger Lifecycle & Execution Phases

Valstorm executes triggers across nine distinct lifecycle phases spanning the CUD pipeline:

Phase	Execution Type	Primary Use Cases
`Before Create`	Synchronous	Data enrichment, setting default values, auto-numbering, in-place validation (raising `HTTPException`).
`After Create`	Synchronous	Creating dependent records (e.g. initial task/wallet), sending urgent notifications, updating parent summaries.
`Async Create`	Asynchronous (Celery)	External system sync (Salesforce, Stripe), heavy calculations, large file processing.
`Before Update`	Synchronous	In-place field normalization, field immutability checks, state machine validation before save.
`After Update`	Synchronous	Parent rollup re-calculations, child record cascades, triggering secondary workflows on field change.
`Async Update`	Asynchronous (Celery)	Background syncing to external APIs when specific fields change.
`Before Delete`	Synchronous	Dependency checks, blocking deletion of records in use, archive cloning.
`After Delete`	Synchronous	Cleaning up orphaned lookup relations, recalculating parent totals.
`Async Delete`	Asynchronous (Celery)	External system de-provisioning and archiving.

  HTTP API Request (CUD)
           │
           ▼
    ┌──────────────┐
    │  Before Phase│ ─── In-place mutations on context.new_map (or raise HTTPException)
    └──────┬───────┘
           ▼
    ┌──────────────┐
    │ Database CUD │ ─── Commit to MongoDB Tenant DB
    └──────┬───────┘
           ▼
    ┌──────────────┐
    │  After Phase │ ─── Synchronous side-effects (context.records.create/update, parent rollups)
    └──────┬───────┘
           │
     HTTP Response 200 OK
           │
           ▼
    ┌──────────────┐
    │  Async Phase │ ─── Asynchronous background execution via Celery workers
    └──────────────┘

2. Trigger Structure & Entry Point

Every trigger script must define an async def execute(context: RecordTriggerContext) entry point:

PYTHON

from fastapi import HTTPException
from valstorm_platform.trigger_context import RecordTriggerContext

async def execute(context: RecordTriggerContext) -> None:
    """
    Standard Record Trigger entry point.
    """
    context.log(f"Executing trigger for schema '{context.schema}' in phase '{context.trigger_context}'", "info")
    
    # 1. Before Create: Normalize and enrich incoming data
    if 'Before' in context.trigger_context and 'Create' in context.trigger_context:
        for record in context.new_map.values():
            if not record.get("status"):
                record["status"] = "Draft"
                
    # 2. After Update: Detect status transition
    if 'After' in context.trigger_context and 'Update' in context.trigger_context:
        for record_id, record in context.new_map.items():
            if context.is_changed(record_id, "status") and record.get("status") == "Approved":
                await context.workflows.run_function(
                    function_name="generate_contract.py",
                    kwargs={"quote_id": record_id}
                )

3. The `RecordTriggerContext` Object

The context object provides batch records, differential tracking, and direct access to platform tools:

Core Properties

Property	Type	Description
`context.schema`	`str`	The API name of the target object (e.g. `'contact'`, `'deal'`).
`context.trigger_context`	`str`	The active lifecycle phase (e.g. `'Before Create'`, `'After Update'`).
`context.user`	`User`	Authenticated user executing the operation (`id`, `organization_id`, `email`, `role`).
`context.new_map`	`dict[str, dict]`	Map of `{record_id: record_dict}` representing the new state (populated on Create and Update).
`context.old_map`	`dict[str, dict]`	Map of `{record_id: record_dict}` representing the previous state (populated on Update and Delete).
`context.changes`	`dict[str, set[str]]`	Map of `{record_id: set_of_changed_field_names}` calculated automatically on updates.
`context.transaction`	`TriggerTransaction`	Request transaction scope enforcing recursion depth limits (max depth: 25) and loop guards.

Helper Methods

context.is_changed(record_id: str, field: str) -> bool: Preferred method for change detection. Returns True on Create, and accurately diffs nested dictionaries/models while treating None and omitted keys as identical.
context.get_list() -> list[dict]: Returns list(context.new_map.values()).
context.get_schema_from_id_async(record_id: str, db) -> str | None: Resolves an ID prefix (e.g. cont_123, cst1_456) into its schema API name.
context.log(message: str, level: str = "info"): Centralized execution logging ("info", "warning", "error", "debug").

4. Platform API Namespaces

RecordTriggerContext exposes platform delegates directly on context:

1. Database Operations (`context.records` & `context.query`)

await context.records.create(api_name, input_data, async_run=False, skip_triggers=False): Creates single or bulk records.
await context.records.update(api_name, input_data, async_run=False, skip_triggers=False): Updates records by ID.
await context.records.delete(api_name, input_data, async_run=False, skip_triggers=False): Deletes records by ID.
await context.records.hydrate_batch(ids: list[str]): Resolves IDs into {id: {"name": ..., "schema": ...}}.
await context.records.calculate_rollup(record_id, api_name, rollup_field_api_name): Calculates rollup values.
await context.query.sql(query, bypass_cache=True): Runs SQL queries with dynamic filters (today, last_n_days:30, ME).
await context.query.mongo(collection, pipeline): Runs MongoDB aggregation pipelines.

2. Workflows & Functions (`context.workflows`)

await context.workflows.run_function(function_name, kwargs): Executes a custom Python function.
await context.workflows.run_workflow(workflow_id, data): Triggers a Flow Builder automation.

3. Metadata & App Configuration (`context.metadata`)

await context.metadata.get_config(api_name): Retrieves organization metadata settings (e.g. "B2B Sales Config").

4. Communications & Notifications (`context.communications`)

await context.communications.send_sms(to_phone, message): Dispatches outbound SMS messages.
await context.communications.send_email(request): Sends transactional or templated emails.

5. External Integrations (`context.integrations`)

context.integrations.salesforce: query(), create(), and update() connected Salesforce objects.
context.integrations.google: Google Drive (create_drive_folder, upload_drive_file, export_drive_file) and Gmail label operations.
context.integrations.twilio: Phone number lookups and conversation management.

6. Utilities & Formatting (`context.utils`)

context.utils.phone_formatter(phone): Cleans and formats phone numbers to standard and E.164 formats.
context.utils.aware_datetime(dt): Returns UTC-aware datetime objects.
context.utils.render_template(template_string, data): Renders text/HTML templates.
await context.utils.html_to_md(html) / await context.utils.md_to_html(md): Document format conversions.

5. Security Sandbox & Gatekeeper AST Validation

All trigger code is automatically scanned and transformed by the Gatekeeper AST Validator before compilation:

Allowed Modules: Standard utilities (typing, datetime, collections, re, uuid, time, base64, io, json, math, asyncio, decimal), data science tools (pandas, matplotlib, seaborn, numpy), and platform APIs (valstorm_platform, ai, fastapi, twilio, pydantic).
Forbidden Operations: Dynamic code execution (eval, exec, compile, globals, locals), direct filesystem access (open, disk-based read_csv/to_csv), and internal dunder inspection (__class__, __dict__).
Loop Safety Injections: The compiler automatically injects await asyncio.sleep(0) into loops within asynchronous routines to prevent thread starvation and guarantee cooperative multitasking.
Recursion & Cycle Prevention: The transaction engine tracks execution depth (max_depth=25) and records execution history (has_ran(schema, trigger, context)), preventing circular trigger cascades.

6. Common Implementation Patterns

Pattern 1: In-Place Data Normalization (`Before Create` / `Before Update`)

Modify records directly in context.new_map. No database update call is needed:

PYTHON

from valstorm_platform.trigger_context import RecordTriggerContext

async def execute(context: RecordTriggerContext) -> None:
    if 'Before' in context.trigger_context:
        for record in context.new_map.values():
            first = (record.get('first_name') or '').strip()
            last = (record.get('last_name') or '').strip()
            if first or last:
                record['name'] = f"{first} {last}".strip()
                
            # Normalize phone formatting
            if record.get('phone'):
                formatted = context.utils.phone_formatter(record['phone'])
                if formatted.get('e164'):
                    record['phone'] = formatted['e164']

Pattern 2: Blocking Validation with HTTP Exceptions (`Before`)

To reject invalid mutations and rollback the transaction, raise an HTTPException:

PYTHON

from fastapi import HTTPException
from valstorm_platform.trigger_context import RecordTriggerContext

async def execute(context: RecordTriggerContext) -> None:
    if 'Before' in context.trigger_context:
        for record in context.new_map.values():
            if record.get('discount_percentage', 0) > 40:
                raise HTTPException(
                    status_code=400,
                    detail="Discounts exceeding 40% require Manager Approval."
                )

Pattern 3: Synchronous Parent Rollup Recalculation (`After Upsert / Delete`)

Recalculate parent records when child line items change:

PYTHON

from valstorm_platform.trigger_context import RecordTriggerContext

async def execute(context: RecordTriggerContext) -> None:
    if 'After' in context.trigger_context:
        # Collect affected parent invoice IDs
        source_records = context.new_map.values() if 'Delete' not in context.trigger_context else context.old_map.values()
        invoice_ids = {r.get('invoice_id') for r in source_records if r.get('invoice_id')}
        
        for invoice_id in invoice_ids:
            # Query all active line items for this invoice
            items = await context.query.sql(f"SELECT total FROM invoice_line_item WHERE invoice_id = '{invoice_id}'")
            total_sum = sum(item.get('total', 0) for item in items)
            
            await context.records.update('invoice', {
                'id': invoice_id,
                'total_amount': total_sum
            })

Pattern 4: Asynchronous External Sync (`Async Create` / `Async Update`)

Offload heavy third-party synchronizations to background Celery workers:

PYTHON

from valstorm_platform.trigger_context import RecordTriggerContext

async def execute(context: RecordTriggerContext) -> None:
    if 'Async' in context.trigger_context:
        for record_id, record in context.new_map.items():
            # Check if synced field was modified
            if context.is_changed(record_id, 'email') or 'Create' in context.trigger_context:
                sf_payload = {
                    "Email": record.get('email'),
                    "LastName": record.get('last_name', 'Unknown'),
                    "Company": record.get('company_name', 'Self')
                }
                
                if record.get('salesforce_id'):
                    sf_payload['Id'] = record['salesforce_id']
                    await context.integrations.salesforce.update([sf_payload])
                else:
                    res = await context.integrations.salesforce.create('Contact', sf_payload)
                    if res and res.get('id'):
                        await context.records.update(context.schema, {
                            'id': record_id,
                            'salesforce_id': res['id']
                        }, skip_triggers=True)

7. Best Practices

Practice	Guideline
Always use `context.is_changed()`	Never manually diff dictionaries (`old != new`). `context.is_changed()` correctly normalizes nested structures and handles omitted keys.
Never call `update()` in `Before` triggers	Directly mutate dictionaries in `context.new_map` to avoid infinite loops and redundant database writes.
Use `skip_triggers=True` when syncing back	When updating a record in an `After` or `Async` trigger from an external system response, pass `skip_triggers=True` on `context.records.update()` to prevent recursion.
Batch-Safe Iteration	Always iterate across `context.new_map.values()` or `context.old_map.values()`. Never assume payloads contain only a single record.
Guard Phase Execution	Explicitly check `if 'Before' in context.trigger_context:` or `if 'Update' in context.trigger_context:` to ensure logic only runs in intended lifecycle stages.
Use `context.log()`	Output logs via `context.log(msg, level)` rather than `print()` so logs appear in the platform's execution history.

Record Triggers

1. Trigger Lifecycle & Execution Phases

2. Trigger Structure & Entry Point

3. The RecordTriggerContext Object

Core Properties

Helper Methods

4. Platform API Namespaces

1. Database Operations (context.records & context.query)

2. Workflows & Functions (context.workflows)

3. Metadata & App Configuration (context.metadata)

4. Communications & Notifications (context.communications)

5. External Integrations (context.integrations)

6. Utilities & Formatting (context.utils)

5. Security Sandbox & Gatekeeper AST Validation

6. Common Implementation Patterns

Pattern 1: In-Place Data Normalization (Before Create / Before Update)

Pattern 2: Blocking Validation with HTTP Exceptions (Before)

Pattern 3: Synchronous Parent Rollup Recalculation (After Upsert / Delete)

Pattern 4: Asynchronous External Sync (Async Create / Async Update)

7. Best Practices

3. The `RecordTriggerContext` Object

1. Database Operations (`context.records` & `context.query`)

2. Workflows & Functions (`context.workflows`)

3. Metadata & App Configuration (`context.metadata`)

4. Communications & Notifications (`context.communications`)

5. External Integrations (`context.integrations`)

6. Utilities & Formatting (`context.utils`)

Pattern 1: In-Place Data Normalization (`Before Create` / `Before Update`)

Pattern 2: Blocking Validation with HTTP Exceptions (`Before`)

Pattern 3: Synchronous Parent Rollup Recalculation (`After Upsert / Delete`)

Pattern 4: Asynchronous External Sync (`Async Create` / `Async Update`)