AI Agent Data Infrastructure

-- Behavioural and transactional events arrive on Kafka as append-only streams.
CREATE SOURCE payments (
  payment_id   BIGINT,
  account_id   BIGINT,
  merchant_id  BIGINT,
  amount       NUMERIC,
  currency     VARCHAR,
  status       VARCHAR,          -- authorized | captured | declined | refunded
  paid_at      TIMESTAMPTZ,
  WATERMARK FOR paid_at AS paid_at - INTERVAL '30 seconds'
) WITH (
  connector = 'kafka',
  topic = 'payments',
  properties.bootstrap.server = 'broker:9092'
) FORMAT PLAIN ENCODE JSON;

CREATE SOURCE inventory_events (
  sku         VARCHAR,
  warehouse   VARCHAR,
  delta       INT,               -- positive for receipts, negative for picks
  on_hand     INT,               -- absolute count after the event
  event_ts    TIMESTAMPTZ,
  WATERMARK FOR event_ts AS event_ts - INTERVAL '10 seconds'
) WITH (
  connector = 'kafka',
  topic = 'inventory',
  properties.bootstrap.server = 'broker:9092'
) FORMAT PLAIN ENCODE JSON;

-- Operational tables come in as change streams through one shared CDC source.
-- Every agent that needs account or merchant state reads these, never the OLTP database.
CREATE SOURCE core_pg WITH (
  connector = 'postgres-cdc',
  hostname = 'pg', port = '5432',
  username = 'rw', password = '...',
  database.name = 'core', schema.name = 'public'
);

CREATE TABLE accounts (
  account_id BIGINT PRIMARY KEY, email VARCHAR, country VARCHAR,
  kyc_level VARCHAR, created_at TIMESTAMPTZ
) FROM core_pg TABLE 'public.accounts';

CREATE TABLE merchants (
  merchant_id BIGINT PRIMARY KEY, name VARCHAR, category VARCHAR, risk_tier VARCHAR
) FROM core_pg TABLE 'public.merchants';

-- Entity state: one row per account, every agent's starting point.
CREATE MATERIALIZED VIEW account_state AS
SELECT a.account_id, a.country, a.kyc_level,
       COUNT(p.payment_id) FILTER (WHERE p.status = 'captured')             AS payments_total,
       COALESCE(SUM(p.amount) FILTER (WHERE p.status = 'captured'), 0)      AS volume_total,
       COUNT(p.payment_id) FILTER (WHERE p.status = 'declined')             AS declines_total,
       MAX(p.paid_at)                                                       AS last_payment_at
FROM accounts a
LEFT JOIN payments p ON p.account_id = a.account_id
GROUP BY a.account_id, a.country, a.kyc_level;

-- Rolling risk signal: declines and volume per account over the last hour, advancing every minute.
CREATE MATERIALIZED VIEW account_risk_1h AS
SELECT account_id, window_start, window_end,
       COUNT(*) FILTER (WHERE status = 'declined')            AS declines_1h,
       COUNT(DISTINCT merchant_id)                            AS merchants_1h,
       COALESCE(SUM(amount) FILTER (WHERE status = 'captured'), 0) AS volume_1h
FROM HOP(payments, paid_at, INTERVAL '1 minute', INTERVAL '1 hour')
GROUP BY account_id, window_start, window_end;

-- Live position: the latest inventory count per SKU and warehouse (group Top-N keeps one row per key).
CREATE MATERIALIZED VIEW inventory_position AS
SELECT sku, warehouse, on_hand, event_ts AS as_of
FROM (
  SELECT *, ROW_NUMBER() OVER (PARTITION BY sku, warehouse ORDER BY event_ts DESC) AS rn
  FROM inventory_events
) t
WHERE rn = 1;

-- Derived candidates: a view other agents and a push subscription can consume.
CREATE MATERIALIZED VIEW fraud_candidates AS
SELECT r.account_id, r.window_end AS flagged_at, r.declines_1h, r.merchants_1h, r.volume_1h,
       s.kyc_level, s.country
FROM account_risk_1h r
JOIN account_state s ON s.account_id = r.account_id
WHERE r.declines_1h >= 3
   OR (r.merchants_1h >= 5 AND s.kyc_level = 'basic');

-- Path 1: pull. An index turns the account lookup into a millisecond point read.
CREATE INDEX account_state_by_id ON account_state (account_id);

-- The support agent's per-turn read (bind account_id as $1).
SELECT to_jsonb(s) AS account,
       (SELECT to_jsonb(r) FROM account_risk_1h r
         WHERE r.account_id = s.account_id
         ORDER BY r.window_end DESC LIMIT 1)        AS risk_last_hour
FROM account_state s
WHERE s.account_id = $1;

-- Path 2: cache. Mirror a hot view into Redis for an agent that reads thousands of keys per second.
CREATE SINK inventory_cache FROM inventory_position
WITH (
  connector = 'redis',
  primary_key = 'sku,warehouse',
  redis.url = 'redis://cache:6379/'
) FORMAT PLAIN ENCODE JSON (force_append_only = 'true');

-- Path 3: push. A subscription records every change to fraud_candidates for a day.
CREATE SUBSCRIPTION fraud_feed FROM fraud_candidates WITH (retention = '1D');

import json, psycopg

# The fraud-review agent does not poll. It holds a cursor on the subscription
# and wakes when a row is added to or removed from fraud_candidates.
rw = psycopg.connect("postgresql://fraud_agent@rw:4566/dev", autocommit=True)
cur = rw.cursor()
cur.execute("DECLARE fraud_cur SUBSCRIPTION CURSOR FOR fraud_feed")   # changes from now on

while True:
    cur.execute("FETCH NEXT FROM fraud_cur WITH (timeout = '5s')")   # blocks up to 5 s
    row = cur.fetchone()
    if row is None:
        continue                                                     # nothing changed
    *fields, op, rw_ts = row
    if op == "Insert" or op == "UpdateInsert":
        case = dict(zip(["account_id", "flagged_at", "declines_1h", "merchants_1h",
                         "volume_1h", "kyc_level", "country"], fields))
        decision = review_with_llm(case)                            # your agent loop
        cur.execute(
            "INSERT INTO agent_actions VALUES (%s, 'fraud_review', %s, %s, %s, now())",
            (f"{case['account_id']}-{rw_ts}", case["account_id"],
             decision["action"], json.dumps(decision)))
    persist_checkpoint(rw_ts)   # store rw_timestamp; resume later with 'since <unix_ms>'

-- Memory: every agent writes what it did to one table. The key includes the agent's name.
CREATE TABLE agent_actions (
  action_id   VARCHAR PRIMARY KEY,
  agent       VARCHAR,
  account_id  BIGINT,
  action      VARCHAR,           -- e.g. 'hold', 'clear', 'refund', 'reorder'
  detail      JSONB,
  acted_at    TIMESTAMPTZ
);

-- Feed memory back into state: the last five actions per account, visible to every agent.
CREATE MATERIALIZED VIEW account_recent_actions AS
SELECT account_id,
       ARRAY_AGG(agent || ':' || action ORDER BY acted_at DESC) AS recent_actions
FROM (
  SELECT *, ROW_NUMBER() OVER (PARTITION BY account_id ORDER BY acted_at DESC) AS rn
  FROM agent_actions
) t
WHERE rn <= 5
GROUP BY account_id;

-- Check memory against reality: holds placed on accounts that have since cleared the risk view.
CREATE MATERIALIZED VIEW stale_holds AS
SELECT a.account_id, a.acted_at AS held_at
FROM agent_actions a
LEFT JOIN fraud_candidates f ON f.account_id = a.account_id
WHERE a.action = 'hold' AND f.account_id IS NULL;

-- Archive: the full action history lands in Iceberg for audit and offline evaluation.
CREATE SINK agent_actions_archive FROM agent_actions
WITH (
  connector = 'iceberg',
  type = 'upsert',
  primary_key = 'action_id',
  warehouse.path = 's3://agent-lake/warehouse',
  database.name = 'agents',
  table.name = 'actions',
  create_table_if_not_exists = 'true',
  commit_checkpoint_interval = '60',
  catalog.type = 'glue',
  catalog.name = 'agent_catalog',
  s3.region = 'us-east-1'
);

-- One user per agent. No agent touches a source, a CDC table or another agent's memory.
CREATE USER support_agent WITH PASSWORD '...';
CREATE USER fraud_agent   WITH PASSWORD '...';
CREATE USER restock_agent WITH PASSWORD '...';

GRANT SELECT ON MATERIALIZED VIEW account_state, account_risk_1h, account_recent_actions
  TO support_agent;

GRANT SELECT ON MATERIALIZED VIEW fraud_candidates, account_state, account_recent_actions
  TO fraud_agent;
GRANT INSERT ON TABLE agent_actions TO fraud_agent;

GRANT SELECT ON MATERIALIZED VIEW inventory_position TO restock_agent;
GRANT INSERT ON TABLE agent_actions TO restock_agent;

-- Observe the layer itself.
SHOW JOBS;                                    -- backfill progress of views being created
SELECT * FROM rw_catalog.rw_table_stats;      -- key count and bytes of state per table and view
SELECT * FROM rw_catalog.rw_streaming_parallelism;

{
  "mcpServers": {
    "risingwave-support": {
      "command": "python",
      "args": ["/opt/risingwave-mcp/src/main.py"],
      "env": {
        "MCP_TRANSPORT": "stdio",
        "RISINGWAVE_CONNECTION_STR": "postgresql://support_agent:...@rw:4566/dev"
      }
    }
  }
}