Live Queries & Real-Time Features¶

KILL "a1b2c3d4-e5f6-7890-abcd-ef1234567890";

DEFINE EVENT audit_employees ON TABLE employees
WHEN $event = "CREATE" OR $event = "UPDATE" OR $event = "DELETE"
THEN {
    CREATE audit_log SET
        table      = "employees",
        record_id  = $value.id,
        action     = $event,
        before     = $before,
        after      = $value,
        changed_by = $auth.id,
        changed_at = time::now()
    ;
};

DEFINE EVENT track_salary_change ON TABLE employees
WHEN $event = "UPDATE" AND $before.salary != $value.salary
THEN {
    CREATE salary_changes SET
        employee_id = $value.id,
        old_salary  = $before.salary,
        new_salary  = $value.salary,
        delta       = $value.salary - $before.salary,
        changed_by  = $auth.id,
        changed_at  = time::now()
    ;
};

DEFINE EVENT archive_deleted_customers ON TABLE customers
WHEN $event = "DELETE"
THEN {
    CREATE customer_archive CONTENT {
        original_id:  $before.id,
        data:         $before,
        deleted_by:   $auth.id,
        deleted_at:   time::now(),
        deletion_reason: $input.reason  -- passed as query variable
    };
};

DEFINE EVENT cascade_department_rename ON TABLE departments
WHEN $event = "UPDATE" AND $before.name != $value.name
THEN {
    UPDATE employees
        SET department_name = $value.name
        WHERE department_id = $value.id
    ;
};

DEFINE EVENT notify_low_stock ON TABLE inventory
WHEN $event = "UPDATE"
  AND $value.quantity < 10
  AND $before.quantity >= 10
THEN {
    CREATE notifications SET
        type       = "low_stock",
        product_id = $value.id,
        quantity   = $value.quantity,
        threshold  = 10,
        created_at = time::now(),
        resolved   = false
    ;
};

DEFINE EVENT orders_activity ON TABLE orders
WHEN $event IN ["CREATE", "UPDATE"]
THEN {
    CREATE activity_feed SET
        entity_type = "order",
        entity_id   = $value.id,
        action      = $event,
        actor       = $auth.id,
        summary     = string::concat("Order ", string::from($value.id),
                                     " was ", string::lowercase($event), "d"),
        payload     = $value,
        occurred_at = time::now()
    ;
};

DEFINE EVENT customers_activity ON TABLE customers
WHEN $event IN ["CREATE", "UPDATE", "DELETE"]
THEN {
    CREATE activity_feed SET
        entity_type = "customer",
        entity_id   = $value.id ?? $before.id,
        action      = $event,
        actor       = $auth.id,
        summary     = string::concat("Customer ", string::lowercase($event), "d"),
        payload     = $value ?? $before,
        occurred_at = time::now()
    ;
};

-- Client records a high-water mark
LET $last_seen = "2025-11-14T09:00:00Z";

SELECT * FROM events
WHERE occurred_at > $last_seen
ORDER BY occurred_at ASC
LIMIT 1000;

-- Read with version
SELECT id, name, version FROM documents WHERE id = "doc:42";
-- version = 7

-- Write only if version matches (optimistic lock)
UPDATE documents SET
    name    = "New Name",
    version = version + 1
WHERE id = "doc:42"
  AND version = 7;
-- Returns empty result if another writer incremented version first

DEFINE TABLE changelog SCHEMAFULL;
DEFINE FIELD table_name  ON changelog TYPE string;
DEFINE FIELD record_id   ON changelog TYPE record;
DEFINE FIELD operation   ON changelog TYPE string;  -- CREATE | UPDATE | DELETE
DEFINE FIELD delta       ON changelog TYPE object;  -- field-level diff
DEFINE FIELD lsn         ON changelog TYPE int;     -- log sequence number
DEFINE FIELD occurred_at ON changelog TYPE datetime;

-- Attach to any table via DEFINE EVENT
DEFINE EVENT changelog_products ON TABLE products
WHEN $event IN ["CREATE", "UPDATE", "DELETE"]
THEN {
    CREATE changelog SET
        table_name  = "products",
        record_id   = $value.id ?? $before.id,
        operation   = $event,
        delta       = object::diff($before, $value),
        lsn         = sequence::next("global_lsn"),
        occurred_at = time::now()
    ;
};

-- Consumer keeps track of: last_lsn = 1042
SELECT * FROM changelog
WHERE lsn > $last_lsn
ORDER BY lsn ASC
LIMIT 100;
-- Next cursor = max(lsn) from results

-- Snapshot A (stored by bitemporal history or explicit snapshot table)
-- Snapshot B (current state)

SELECT b.id, b.value, a.value AS previous_value
FROM current_state AS b
LEFT JOIN snapshot_20251101 AS a ON a.id = b.id
WHERE b.value != a.value OR a.id IS NULL;

-- Keep only the latest state per record, discard intermediate changes older than 30 days
DELETE changelog
WHERE occurred_at < time::now() - 30d
  AND operation != "DELETE";

-- For each remaining DELETE, ensure the record is gone from the live table
-- (handled by the DEFINE EVENT archive pattern above)

DEFINE TABLE inbox SCHEMAFULL;
DEFINE FIELD agent_id    ON inbox TYPE string;
DEFINE FIELD task        ON inbox TYPE object;
DEFINE FIELD status      ON inbox TYPE string DEFAULT "pending";
DEFINE FIELD created_at  ON inbox TYPE datetime DEFAULT time::now();

-- Agent polls its inbox
SELECT * FROM inbox
WHERE agent_id = $agent_id
  AND status   = "pending"
ORDER BY created_at ASC
LIMIT 1;

-- Mark in-progress
UPDATE inbox SET status = "in_progress", started_at = time::now()
WHERE id = $task_id AND status = "pending";

-- Mark done
UPDATE inbox SET status = "done", completed_at = time::now(), result = $result
WHERE id = $task_id;

DEFINE TABLE task_queue SCHEMAFULL;
DEFINE FIELD queue_name  ON task_queue TYPE string;
DEFINE FIELD payload     ON task_queue TYPE object;
DEFINE FIELD status      ON task_queue TYPE string DEFAULT "queued";
DEFINE FIELD priority    ON task_queue TYPE int    DEFAULT 0;
DEFINE FIELD enqueued_at ON task_queue TYPE datetime DEFAULT time::now();
DEFINE INDEX idx_queue_priority ON task_queue FIELDS queue_name, priority, enqueued_at;

-- Dequeue (atomic claim)
BEGIN TRANSACTION;
    LET $task = (
        SELECT * FROM task_queue
        WHERE queue_name = "embedding_jobs"
          AND status     = "queued"
        ORDER BY priority DESC, enqueued_at ASC
        LIMIT 1
    )[0];

    IF $task != NONE {
        UPDATE task_queue SET status = "running", claimed_at = time::now()
        WHERE id = $task.id AND status = "queued";
    };

    RETURN $task;
COMMIT TRANSACTION;

DEFINE TABLE pubsub_channels SCHEMAFULL;
DEFINE FIELD channel   ON pubsub_channels TYPE string;
DEFINE FIELD message   ON pubsub_channels TYPE object;
DEFINE FIELD published_at ON pubsub_channels TYPE datetime DEFAULT time::now();
DEFINE FIELD ttl       ON pubsub_channels TYPE duration DEFAULT 1h;

-- Publish
CREATE pubsub_channels SET
    channel = "model.inference.complete",
    message = {
        model_id:   "gpt-4o",
        request_id: "req-abc123",
        tokens:     1024,
        latency_ms: 387
    };

-- Subscribe (via LIVE SELECT)
LIVE SELECT * FROM pubsub_channels
WHERE channel = "model.inference.complete";

-- Clean up expired messages
DELETE pubsub_channels WHERE published_at < time::now() - ttl;

DEFINE TABLE agent_state SCHEMAFULL;
DEFINE FIELD agent_id       ON agent_state TYPE string;
DEFINE FIELD current_state  ON agent_state TYPE string;
DEFINE FIELD context        ON agent_state TYPE object;
DEFINE FIELD transitions    ON agent_state TYPE array;
DEFINE FIELD updated_at     ON agent_state TYPE datetime;

-- Transition with history
UPDATE agent_state SET
    current_state = $new_state,
    transitions   = array::push(transitions, {
        from:  current_state,
        to:    $new_state,
        at:    time::now(),
        reason: $reason
    }),
    updated_at = time::now()
WHERE agent_id = $agent_id
  AND current_state IN $allowed_from_states;

DEFINE TABLE documents SCHEMAFULL;
DEFINE FIELD content    ON documents TYPE string;
DEFINE FIELD version    ON documents TYPE int DEFAULT 0;
DEFINE FIELD editors    ON documents TYPE array;
DEFINE FIELD updated_at ON documents TYPE datetime;

-- Conflict-free update with version check
BEGIN TRANSACTION;
    LET $doc = SELECT * FROM documents WHERE id = $doc_id;

    IF $doc.version != $expected_version {
        THROW "Conflict: document was modified by another editor (version mismatch)";
    };

    UPDATE documents SET
        content    = $new_content,
        version    = version + 1,
        editors    = array::push(editors, { user: $auth.id, at: time::now() }),
        updated_at = time::now()
    WHERE id = $doc_id;
COMMIT TRANSACTION;

DEFINE TABLE agent_heartbeats SCHEMAFULL;
DEFINE FIELD agent_id    ON agent_heartbeats TYPE string;
DEFINE FIELD last_seen   ON agent_heartbeats TYPE datetime;
DEFINE FIELD metadata    ON agent_heartbeats TYPE object;

-- Agent sends heartbeat every 30s
UPSERT agent_heartbeats SET
    last_seen = time::now(),
    metadata  = { load: $load, tasks_completed: $tasks_completed }
WHERE agent_id = $agent_id;

-- Monitor detects stale agents (not seen in 2 minutes)
SELECT agent_id, last_seen,
       duration::from::secs(time::unix() - time::unix(last_seen)) AS silent_for
FROM agent_heartbeats
WHERE last_seen < time::now() - 2m
ORDER BY last_seen ASC;

DEFINE TABLE event_store SCHEMAFULL;
DEFINE FIELD aggregate_id   ON event_store TYPE string;
DEFINE FIELD aggregate_type ON event_store TYPE string;
DEFINE FIELD event_type     ON event_store TYPE string;
DEFINE FIELD payload        ON event_store TYPE object;
DEFINE FIELD sequence_nr    ON event_store TYPE int;
DEFINE FIELD occurred_at    ON event_store TYPE datetime DEFAULT time::now();
DEFINE INDEX idx_aggregate ON event_store FIELDS aggregate_id, sequence_nr UNIQUE;

-- Append event
CREATE event_store SET
    aggregate_id   = "order:abc123",
    aggregate_type = "Order",
    event_type     = "OrderShipped",
    payload        = { tracking_number: "UPS-98765", carrier: "UPS" },
    sequence_nr    = (SELECT VALUE count() FROM event_store WHERE aggregate_id = "order:abc123") + 1;

-- Replay aggregate (get all events in order)
SELECT event_type, payload, occurred_at
FROM event_store
WHERE aggregate_id = "order:abc123"
ORDER BY sequence_nr ASC;

-- Snapshot for performance (store materialized state)
CREATE order_snapshots SET
    aggregate_id = "order:abc123",
    state        = $current_state,
    at_sequence  = $last_sequence_nr,
    snapshot_at  = time::now();