Materialized Views for AI Agents

-- Product usage events from Kafka. The watermark lets windows close and state be dropped.
CREATE SOURCE usage_events (
  account_id  BIGINT,
  user_id     BIGINT,
  feature     VARCHAR,
  event_type  VARCHAR,          -- 'use' | 'error' | 'export'
  event_ts    TIMESTAMPTZ,
  WATERMARK FOR event_ts AS event_ts - INTERVAL '1 minute'
) WITH (
  connector = 'kafka', topic = 'usage',
  properties.bootstrap.server = 'broker:9092'
) FORMAT PLAIN ENCODE JSON;

-- Billing and support come from PostgreSQL as change streams.
CREATE SOURCE saas_pg WITH (
  connector = 'postgres-cdc', hostname = 'pg', port = '5432',
  username = 'rw', password = '...', database.name = 'saas', schema.name = 'public'
);

CREATE TABLE accounts (account_id BIGINT PRIMARY KEY, name VARCHAR, plan VARCHAR,
                       seats INT, renewal_date DATE, owner_email VARCHAR)
  FROM saas_pg TABLE 'public.accounts';

CREATE TABLE invoices (invoice_id BIGINT PRIMARY KEY, account_id BIGINT, amount NUMERIC,
                       status VARCHAR, due_date DATE, issued_at TIMESTAMPTZ)
  FROM saas_pg TABLE 'public.invoices';

CREATE TABLE tickets (ticket_id BIGINT PRIMARY KEY, account_id BIGINT, priority VARCHAR,
                      status VARCHAR, opened_at TIMESTAMPTZ)
  FROM saas_pg TABLE 'public.tickets';

-- Plan catalogue: small, slowly changing, used for enrichment.
CREATE TABLE plans (plan VARCHAR PRIMARY KEY, seat_limit INT, export_limit_per_day INT);

-- Pattern 1. Entity state: one row per account, joined from several sources.
-- RisingWave updates only the affected account's row when any input changes.
CREATE MATERIALIZED VIEW account_profile AS
SELECT a.account_id, a.name, a.plan, a.seats, a.renewal_date,
       COUNT(DISTINCT t.ticket_id) FILTER (WHERE t.status <> 'closed')             AS open_tickets,
       COUNT(DISTINCT t.ticket_id) FILTER (WHERE t.priority = 'urgent'
                                             AND t.status <> 'closed')             AS urgent_tickets,
       COALESCE(SUM(i.amount) FILTER (WHERE i.status = 'overdue'), 0)              AS overdue_amount,
       MAX(i.issued_at)                                                            AS last_invoice_at
FROM accounts a
LEFT JOIN tickets  t ON t.account_id = a.account_id
LEFT JOIN invoices i ON i.account_id = a.account_id
GROUP BY a.account_id, a.name, a.plan, a.seats, a.renewal_date;

-- Pattern 2. Latest per key: the most recent invoice per account.
-- ROW_NUMBER with a rank filter is recognised as group Top-N and keeps one row of state per key.
CREATE MATERIALIZED VIEW latest_invoice AS
SELECT account_id, invoice_id, amount, status, due_date, issued_at
FROM (
  SELECT *, ROW_NUMBER() OVER (PARTITION BY account_id ORDER BY issued_at DESC) AS rn
  FROM invoices
) t
WHERE rn = 1;

-- Pattern 3. Rolling window: usage over the last 7 days, advancing every hour.
-- The agent reads the row with the latest window_end; older windows are dropped once the watermark passes.
CREATE MATERIALIZED VIEW usage_7d AS
SELECT account_id, window_start, window_end,
       COUNT(DISTINCT user_id)                              AS active_users_7d,
       COUNT(*) FILTER (WHERE event_type = 'use')           AS actions_7d,
       COUNT(*) FILTER (WHERE event_type = 'error')         AS errors_7d,
       COUNT(DISTINCT feature)                              AS features_used_7d
FROM HOP(usage_events, event_ts, INTERVAL '1 hour', INTERVAL '7 days')
GROUP BY account_id, window_start, window_end;

-- Pattern 4. Finalised window: one immutable row per account per day, emitted when the day closes.
-- Use this for history the agent cites ("usage fell 40% week over week"), not for the live figure.
CREATE MATERIALIZED VIEW daily_usage AS
SELECT account_id, window_start AS day,
       COUNT(DISTINCT user_id)                      AS active_users,
       COUNT(*) FILTER (WHERE event_type = 'use')   AS actions,
       COUNT(*) FILTER (WHERE event_type = 'export') AS exports
FROM TUMBLE(usage_events, event_ts, INTERVAL '1 day')
GROUP BY account_id, window_start
EMIT ON WINDOW CLOSE;

-- Pattern 5. Event-time enrichment: attach the plan limits in force when each export happened.
-- The temporal join stores only the plans side, not the event stream, so state stays small.
CREATE MATERIALIZED VIEW exports_vs_limit AS
SELECT e.account_id, e.event_ts, e.user_id, p.export_limit_per_day
FROM usage_events e
JOIN accounts a FOR SYSTEM_TIME AS OF PROCTIME() ON a.account_id = e.account_id
JOIN plans    p FOR SYSTEM_TIME AS OF PROCTIME() ON p.plan = a.plan
WHERE e.event_type = 'export';

-- Pattern 6. Ranked list: the five most used features per account over the rolling window.
-- First aggregate per feature, then Top-N. The agent gets a short list instead of a feature histogram.
CREATE MATERIALIZED VIEW feature_usage_7d AS
SELECT account_id, feature, window_end, COUNT(*) AS uses
FROM HOP(usage_events, event_ts, INTERVAL '1 hour', INTERVAL '7 days')
WHERE event_type = 'use'
GROUP BY account_id, feature, window_end;

CREATE MATERIALIZED VIEW top_features_7d AS
SELECT account_id, window_end, feature, uses
FROM (
  SELECT *, ROW_NUMBER() OVER (PARTITION BY account_id, window_end ORDER BY uses DESC) AS rn
  FROM feature_usage_7d
) t
WHERE rn <= 5;

-- Pattern 7. Cascade: a health score defined over the views above.
-- First pin the current rolling window per account (pattern 2 applied to pattern 3).
CREATE MATERIALIZED VIEW usage_7d_latest AS
SELECT account_id, window_end, active_users_7d, actions_7d, errors_7d, features_used_7d
FROM (
  SELECT *, ROW_NUMBER() OVER (PARTITION BY account_id ORDER BY window_end DESC) AS rn
  FROM usage_7d
) t
WHERE rn = 1;

-- Then derive the score. Every level advances at the same barrier, so the score
-- never mixes a new ticket count with an old usage figure.
CREATE MATERIALIZED VIEW account_health AS
SELECT p.account_id, p.name, p.plan, p.seats, p.renewal_date,
       u.active_users_7d, u.errors_7d, p.open_tickets, p.urgent_tickets, p.overdue_amount,
       CASE
         WHEN p.urgent_tickets > 0 OR p.overdue_amount > 0            THEN 'red'
         WHEN u.active_users_7d < p.seats * 0.3 OR u.errors_7d > 50   THEN 'amber'
         ELSE 'green'
       END AS health
FROM account_profile p
LEFT JOIN usage_7d_latest u ON u.account_id = p.account_id;

-- Pattern 8. Memory: the agent writes, a view summarises, the context query joins.
CREATE TABLE agent_notes (
  note_id    VARCHAR PRIMARY KEY,
  account_id BIGINT,
  kind       VARCHAR,            -- 'promise' | 'risk' | 'preference' | 'outreach'
  content    VARCHAR,
  created_at TIMESTAMPTZ
);

CREATE MATERIALIZED VIEW account_notes_recent AS
SELECT account_id,
       ARRAY_AGG(kind || ': ' || content ORDER BY created_at DESC) AS notes
FROM (
  SELECT *, ROW_NUMBER() OVER (PARTITION BY account_id ORDER BY created_at DESC) AS rn
  FROM agent_notes
) t
WHERE rn <= 5
GROUP BY account_id;

-- Serve: a point lookup that returns the whole context as one JSON object.
CREATE INDEX account_health_by_id ON account_health (account_id);

SELECT jsonb_build_object(
         'account',   to_jsonb(h),
         'last_invoice', (SELECT to_jsonb(l) FROM latest_invoice l WHERE l.account_id = h.account_id),
         'top_features', (SELECT jsonb_agg(jsonb_build_object('feature', f.feature, 'uses', f.uses))
                            FROM top_features_7d f
                           WHERE f.account_id = h.account_id
                             AND f.window_end = (SELECT MAX(window_end) FROM top_features_7d
                                                  WHERE account_id = h.account_id)),
         'last_30_days', (SELECT jsonb_agg(jsonb_build_object('day', d.day, 'active_users', d.active_users)
                                           ORDER BY d.day)
                            FROM daily_usage d
                           WHERE d.account_id = h.account_id
                             AND d.day >= CURRENT_DATE - INTERVAL '30 days'),
         'notes',     COALESCE(n.notes, ARRAY[]::VARCHAR[])
       ) AS context
FROM account_health h
LEFT JOIN account_notes_recent n ON n.account_id = h.account_id
WHERE h.account_id = $1;

-- See the streaming plan before you create the view. Look for StreamGroupTopN and temporal join operators.
EXPLAIN CREATE MATERIALIZED VIEW latest_invoice AS
SELECT ... ;

-- Create large views in the background and watch backfill progress.
SET BACKGROUND_DDL = true;
CREATE MATERIALIZED VIEW account_profile AS ... ;
SHOW JOBS;

-- How much state does each view hold?
SELECT * FROM rw_catalog.rw_table_stats ORDER BY total_key_count DESC;

-- Which jobs run at what parallelism?
SELECT * FROM rw_catalog.rw_streaming_parallelism;