-- Daily rollup table for the workspace `/{slug}/dashboard` page. Mirrors -- the per-runtime rollup in migration 073 (`task_usage_daily`) but indexes -- on a different set of dimensions: -- -- (bucket_date, workspace_id, agent_id, project_id, model) -- -- vs. the existing 073: -- -- (bucket_date, workspace_id, runtime_id, provider, model) -- -- The dashboard queries don't filter by `runtime_id` and the per-runtime -- table doesn't carry `agent_id` / `project_id`, so we materialise a -- dedicated rollup instead of extending the existing one and balloonng -- its cardinality (`runtime × agent × project × model` per day vs. -- `runtime × model`). The cron entry point + dirty-queue invalidation -- pattern is otherwise identical. -- -- `project_id` is the project at rollup time, snapshotted via -- `agent_task_queue.issue_id → issue.project_id`. Re-attribution after a -- user reassigns an issue's project lands via the `issue` trigger below; -- historical buckets that aren't touched again stay attributed where -- they were when the rollup ran. (Operator follow-up: re-run the backfill -- command on the affected window if a bulk project move needs to -- propagate to old data.) -- -- `project_id` is nullable — issues without a project still produce -- usage rows. We use `UNIQUE NULLS NOT DISTINCT` (PG 15+) so NULL is -- treated as a single distinct value in the unique key, which lets -- `INSERT ... ON CONFLICT` upsert "no-project" buckets the same way it -- handles a specific project. -- -- Bucket date is computed in **UTC**. The per-runtime rollup uses the -- runtime's tz (migration 082) because each row has a single runtime; -- this rollup aggregates *across* runtimes that may sit in different -- tzs, and there is no single correct local boundary. The dashboard -- frontend just renders the raw ISO date. CREATE TABLE task_usage_dashboard_daily ( bucket_date DATE NOT NULL, workspace_id UUID NOT NULL, agent_id UUID NOT NULL, project_id UUID, model TEXT NOT NULL, input_tokens BIGINT NOT NULL DEFAULT 0, output_tokens BIGINT NOT NULL DEFAULT 0, cache_read_tokens BIGINT NOT NULL DEFAULT 0, cache_write_tokens BIGINT NOT NULL DEFAULT 0, task_count BIGINT NOT NULL DEFAULT 0, event_count BIGINT NOT NULL DEFAULT 0, updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), CONSTRAINT uq_task_usage_dashboard_daily_key UNIQUE NULLS NOT DISTINCT (bucket_date, workspace_id, agent_id, project_id, model) ); -- Workspace-wide reads (no project filter) hit this index. CREATE INDEX idx_task_usage_dashboard_daily_workspace_date ON task_usage_dashboard_daily (workspace_id, bucket_date DESC); -- Per-project reads. Partial index because most queries either filter to -- one project or "all" — both extremes benefit from this layout. CREATE INDEX idx_task_usage_dashboard_daily_project_date ON task_usage_dashboard_daily (workspace_id, project_id, bucket_date DESC); -- Per-agent reads (the "by agent" panel). CREATE INDEX idx_task_usage_dashboard_daily_agent_date ON task_usage_dashboard_daily (workspace_id, agent_id, bucket_date DESC); CREATE TABLE task_usage_dashboard_rollup_state ( id SMALLINT PRIMARY KEY DEFAULT 1 CHECK (id = 1), watermark_at TIMESTAMPTZ NOT NULL DEFAULT '1970-01-01 00:00:00+00', last_run_started_at TIMESTAMPTZ, last_run_finished_at TIMESTAMPTZ, last_run_rows BIGINT NOT NULL DEFAULT 0, last_error TEXT ); INSERT INTO task_usage_dashboard_rollup_state (id) VALUES (1) ON CONFLICT DO NOTHING; -- Dirty queue for invalidations the `updated_at` watermark can't see: -- * DELETE on `task_usage` (no row left for the watermark). -- * DELETE cascade through `agent_task_queue` (task_usage rows gone). -- * UPDATE of `issue.project_id` — moves the bucket to a new key. -- -- bucket key matches the rollup table so the queue can be UNIONed into -- `dirty_keys` in the window function with no translation. project_id is -- nullable here for the same reason as in the rollup table. CREATE TABLE task_usage_dashboard_dirty ( bucket_date DATE NOT NULL, workspace_id UUID NOT NULL, agent_id UUID NOT NULL, project_id UUID, model TEXT NOT NULL, enqueued_at TIMESTAMPTZ NOT NULL DEFAULT now(), CONSTRAINT uq_task_usage_dashboard_dirty_key UNIQUE NULLS NOT DISTINCT (bucket_date, workspace_id, agent_id, project_id, model) ); CREATE INDEX idx_task_usage_dashboard_dirty_enqueued_at ON task_usage_dashboard_dirty (enqueued_at); -- Trigger 1: agent_task_queue BEFORE UPDATE OF issue_id OR DELETE. -- -- Two cases: -- -- * UPDATE OF issue_id — currently only `LinkTaskToIssue` (quick-create -- tasks attaching to the issue the agent just produced) writes here, -- moving the task from `issue_id IS NULL` to a real issue. If usage -- already rolled up under the no-project bucket, we have to enqueue -- both OLD (NULL project) AND NEW (the new issue's project) so the -- next tick re-attributes the tokens. -- -- * DELETE — direct atq deletions land here with the issue row still -- alive, so `LEFT JOIN issue` resolves the project correctly. -- Cascade DELETE driven from `DELETE FROM issue` is handled by -- `enqueue_task_usage_dashboard_dirty_for_issue_delete` below (which -- fires *before* the cascade, while `issue.project_id` is still -- readable); this trigger may also fire during that cascade, but the -- join returns no row → workspace_id is missing, so the JOIN on -- `agent` keeps the enqueue safe and the resulting NULL-project key -- no-ops on recompute (deleted_empty path). CREATE OR REPLACE FUNCTION enqueue_task_usage_dashboard_dirty_for_atq() RETURNS TRIGGER LANGUAGE plpgsql AS $$ BEGIN IF TG_OP = 'UPDATE' THEN IF OLD.issue_id IS DISTINCT FROM NEW.issue_id THEN -- OLD side: project_id of OLD.issue_id (NULL when OLD.issue_id IS NULL, -- e.g. the quick-create starting state). INSERT INTO task_usage_dashboard_dirty ( bucket_date, workspace_id, agent_id, project_id, model ) SELECT DISTINCT DATE(tu.created_at), a.workspace_id, OLD.agent_id, i_old.project_id, tu.model FROM task_usage tu JOIN agent a ON a.id = OLD.agent_id LEFT JOIN issue i_old ON i_old.id = OLD.issue_id WHERE tu.task_id = OLD.id ON CONFLICT ON CONSTRAINT uq_task_usage_dashboard_dirty_key DO UPDATE SET enqueued_at = GREATEST(task_usage_dashboard_dirty.enqueued_at, EXCLUDED.enqueued_at); -- NEW side: project_id of NEW.issue_id. INSERT INTO task_usage_dashboard_dirty ( bucket_date, workspace_id, agent_id, project_id, model ) SELECT DISTINCT DATE(tu.created_at), a.workspace_id, NEW.agent_id, i_new.project_id, tu.model FROM task_usage tu JOIN agent a ON a.id = NEW.agent_id LEFT JOIN issue i_new ON i_new.id = NEW.issue_id WHERE tu.task_id = NEW.id ON CONFLICT ON CONSTRAINT uq_task_usage_dashboard_dirty_key DO UPDATE SET enqueued_at = GREATEST(task_usage_dashboard_dirty.enqueued_at, EXCLUDED.enqueued_at); END IF; RETURN NEW; ELSIF TG_OP = 'DELETE' THEN INSERT INTO task_usage_dashboard_dirty ( bucket_date, workspace_id, agent_id, project_id, model ) SELECT DISTINCT DATE(tu.created_at), a.workspace_id, OLD.agent_id, i.project_id, tu.model FROM task_usage tu JOIN agent a ON a.id = OLD.agent_id LEFT JOIN issue i ON i.id = OLD.issue_id WHERE tu.task_id = OLD.id ON CONFLICT ON CONSTRAINT uq_task_usage_dashboard_dirty_key DO UPDATE SET enqueued_at = GREATEST(task_usage_dashboard_dirty.enqueued_at, EXCLUDED.enqueued_at); RETURN OLD; END IF; RETURN NULL; END; $$; CREATE TRIGGER trg_atq_dirty_dashboard BEFORE UPDATE OF issue_id OR DELETE ON agent_task_queue FOR EACH ROW EXECUTE FUNCTION enqueue_task_usage_dashboard_dirty_for_atq(); -- Trigger 1b: issue BEFORE DELETE. -- -- `DELETE FROM issue` cascades to `agent_task_queue` and onward to -- `task_usage`. By the time the atq BEFORE DELETE trigger runs, the -- issue row is gone and `LEFT JOIN issue` returns NULL for project_id, -- so the atq trigger would enqueue a NULL-project key — the rollup row -- under the original project would never get cleared and would keep -- billing the workspace for tokens that no longer have a source. -- -- This trigger fires BEFORE the cascade, while `OLD.project_id` is still -- readable, and enqueues one dirty row per (date, agent, model) the -- issue's tasks contributed to. The next rollup tick recomputes the -- bucket, finds no source rows under the original project, and drops it -- (deleted_empty path). CREATE OR REPLACE FUNCTION enqueue_task_usage_dashboard_dirty_for_issue_delete() RETURNS TRIGGER LANGUAGE plpgsql AS $$ BEGIN INSERT INTO task_usage_dashboard_dirty ( bucket_date, workspace_id, agent_id, project_id, model ) SELECT DISTINCT DATE(tu.created_at), OLD.workspace_id, atq.agent_id, OLD.project_id, tu.model FROM agent_task_queue atq JOIN task_usage tu ON tu.task_id = atq.id WHERE atq.issue_id = OLD.id ON CONFLICT ON CONSTRAINT uq_task_usage_dashboard_dirty_key DO UPDATE SET enqueued_at = GREATEST(task_usage_dashboard_dirty.enqueued_at, EXCLUDED.enqueued_at); RETURN OLD; END; $$; CREATE TRIGGER trg_issue_delete_dirty_dashboard BEFORE DELETE ON issue FOR EACH ROW EXECUTE FUNCTION enqueue_task_usage_dashboard_dirty_for_issue_delete(); -- Trigger 2: task_usage BEFORE DELETE. -- Rare in practice (no direct DELETE call sites today) but ensures the -- rollup converges if one is added. UPDATE / INSERT are caught by the -- updated_at watermark in the window function. CREATE OR REPLACE FUNCTION enqueue_task_usage_dashboard_dirty_for_tu() RETURNS TRIGGER LANGUAGE plpgsql AS $$ BEGIN INSERT INTO task_usage_dashboard_dirty ( bucket_date, workspace_id, agent_id, project_id, model ) SELECT DATE(OLD.created_at), a.workspace_id, atq.agent_id, i.project_id, OLD.model FROM agent_task_queue atq JOIN agent a ON a.id = atq.agent_id LEFT JOIN issue i ON i.id = atq.issue_id WHERE atq.id = OLD.task_id ON CONFLICT ON CONSTRAINT uq_task_usage_dashboard_dirty_key DO UPDATE SET enqueued_at = GREATEST(task_usage_dashboard_dirty.enqueued_at, EXCLUDED.enqueued_at); RETURN OLD; END; $$; CREATE TRIGGER trg_tu_dirty_dashboard BEFORE DELETE ON task_usage FOR EACH ROW EXECUTE FUNCTION enqueue_task_usage_dashboard_dirty_for_tu(); -- Trigger 3: issue BEFORE UPDATE OF project_id. -- Re-attribute every (date, agent, model) bucket touched by tasks under -- this issue: enqueue OLD project_id (so it stops claiming the tokens) -- AND NEW project_id (so it picks them up). Both go through the same -- dirty queue; the window function recomputes from the live join in -- recomputed CTE, which now sees NEW.project_id, so the OLD bucket -- either drops to 0 (deleted_empty path) or shrinks. CREATE OR REPLACE FUNCTION enqueue_task_usage_dashboard_dirty_for_issue_project() RETURNS TRIGGER LANGUAGE plpgsql AS $$ BEGIN IF OLD.project_id IS DISTINCT FROM NEW.project_id THEN -- OLD project buckets. INSERT INTO task_usage_dashboard_dirty ( bucket_date, workspace_id, agent_id, project_id, model ) SELECT DISTINCT DATE(tu.created_at), NEW.workspace_id, atq.agent_id, OLD.project_id, tu.model FROM agent_task_queue atq JOIN task_usage tu ON tu.task_id = atq.id WHERE atq.issue_id = NEW.id ON CONFLICT ON CONSTRAINT uq_task_usage_dashboard_dirty_key DO UPDATE SET enqueued_at = GREATEST(task_usage_dashboard_dirty.enqueued_at, EXCLUDED.enqueued_at); -- NEW project buckets. INSERT INTO task_usage_dashboard_dirty ( bucket_date, workspace_id, agent_id, project_id, model ) SELECT DISTINCT DATE(tu.created_at), NEW.workspace_id, atq.agent_id, NEW.project_id, tu.model FROM agent_task_queue atq JOIN task_usage tu ON tu.task_id = atq.id WHERE atq.issue_id = NEW.id ON CONFLICT ON CONSTRAINT uq_task_usage_dashboard_dirty_key DO UPDATE SET enqueued_at = GREATEST(task_usage_dashboard_dirty.enqueued_at, EXCLUDED.enqueued_at); END IF; RETURN NEW; END; $$; CREATE TRIGGER trg_issue_project_dirty_dashboard BEFORE UPDATE OF project_id ON issue FOR EACH ROW EXECUTE FUNCTION enqueue_task_usage_dashboard_dirty_for_issue_project(); -- Window function. Same shape as 077's rollup_task_usage_daily_window: -- 1. Discover dirty keys from updated_at watermark + the explicit queue. -- 2. Recompute each from raw via the agent_task_queue + issue join. -- 3. Upsert present buckets, delete buckets that recomputed to nothing. -- 4. Drain the queue rows whose enqueued_at < p_to. -- -- IDEMPOTENCY: re-running the same window yields the same final state, -- because each touched key is rebuilt from raw, not deltaed. CREATE OR REPLACE FUNCTION rollup_task_usage_dashboard_daily_window( p_from TIMESTAMPTZ, p_to TIMESTAMPTZ ) RETURNS BIGINT LANGUAGE plpgsql AS $$ DECLARE v_rows BIGINT; BEGIN IF p_from >= p_to THEN RETURN 0; END IF; WITH dirty_from_updates AS ( SELECT DISTINCT DATE(tu.created_at) AS bucket_date, a.workspace_id AS workspace_id, atq.agent_id AS agent_id, i.project_id AS project_id, tu.model AS model FROM task_usage tu JOIN agent_task_queue atq ON atq.id = tu.task_id JOIN agent a ON a.id = atq.agent_id LEFT JOIN issue i ON i.id = atq.issue_id WHERE ( (tu.updated_at >= p_from AND tu.updated_at < p_to) -- Legacy `updated_at IS NULL` rows: 077 handles this via a -- partial index. We rely on the same `idx_task_usage_created_at_legacy` -- (migration 078) for this branch. OR (tu.updated_at IS NULL AND tu.created_at >= p_from AND tu.created_at < p_to) ) ), dirty_from_queue AS ( SELECT bucket_date, workspace_id, agent_id, project_id, model FROM task_usage_dashboard_dirty WHERE enqueued_at < p_to ), dirty_keys AS ( SELECT * FROM dirty_from_updates UNION SELECT * FROM dirty_from_queue ), recomputed AS ( SELECT dk.bucket_date, dk.workspace_id, dk.agent_id, dk.project_id, dk.model, SUM(tu.input_tokens)::bigint AS input_tokens, SUM(tu.output_tokens)::bigint AS output_tokens, SUM(tu.cache_read_tokens)::bigint AS cache_read_tokens, SUM(tu.cache_write_tokens)::bigint AS cache_write_tokens, COUNT(DISTINCT tu.task_id)::bigint AS task_count, COUNT(*)::bigint AS event_count FROM dirty_keys dk JOIN agent_task_queue atq ON atq.agent_id = dk.agent_id JOIN agent a ON a.id = atq.agent_id AND a.workspace_id = dk.workspace_id LEFT JOIN issue i ON i.id = atq.issue_id JOIN task_usage tu ON tu.task_id = atq.id AND tu.model = dk.model AND DATE(tu.created_at) = dk.bucket_date WHERE (i.project_id IS NOT DISTINCT FROM dk.project_id) GROUP BY 1, 2, 3, 4, 5 ), upserted AS ( INSERT INTO task_usage_dashboard_daily AS d ( bucket_date, workspace_id, agent_id, project_id, model, input_tokens, output_tokens, cache_read_tokens, cache_write_tokens, task_count, event_count ) SELECT bucket_date, workspace_id, agent_id, project_id, model, input_tokens, output_tokens, cache_read_tokens, cache_write_tokens, task_count, event_count FROM recomputed ON CONFLICT ON CONSTRAINT uq_task_usage_dashboard_daily_key DO UPDATE SET input_tokens = EXCLUDED.input_tokens, output_tokens = EXCLUDED.output_tokens, cache_read_tokens = EXCLUDED.cache_read_tokens, cache_write_tokens = EXCLUDED.cache_write_tokens, task_count = EXCLUDED.task_count, event_count = EXCLUDED.event_count, updated_at = now() RETURNING 1 ), deleted_empty AS ( DELETE FROM task_usage_dashboard_daily d USING dirty_keys dk WHERE d.bucket_date = dk.bucket_date AND d.workspace_id = dk.workspace_id AND d.agent_id = dk.agent_id AND d.project_id IS NOT DISTINCT FROM dk.project_id AND d.model = dk.model AND NOT EXISTS ( SELECT 1 FROM recomputed r WHERE r.bucket_date = dk.bucket_date AND r.workspace_id = dk.workspace_id AND r.agent_id = dk.agent_id AND r.project_id IS NOT DISTINCT FROM dk.project_id AND r.model = dk.model ) RETURNING 1 ) SELECT (SELECT COUNT(*) FROM upserted) + (SELECT COUNT(*) FROM deleted_empty) INTO v_rows; DELETE FROM task_usage_dashboard_dirty WHERE enqueued_at < p_to; RETURN v_rows; END; $$; -- Cron entry. Mirrors `rollup_task_usage_daily` (migration 073) — same -- advisory-lock + watermark + error-recovery shape. Uses lock id 4244 -- so it serialises independently of the per-runtime rollup (4242). CREATE OR REPLACE FUNCTION rollup_task_usage_dashboard_daily() RETURNS BIGINT LANGUAGE plpgsql AS $$ DECLARE v_lock_ok BOOLEAN; v_from TIMESTAMPTZ; v_to TIMESTAMPTZ; v_rows BIGINT := 0; BEGIN SELECT pg_try_advisory_lock(4244) INTO v_lock_ok; IF NOT v_lock_ok THEN RETURN 0; END IF; BEGIN UPDATE task_usage_dashboard_rollup_state SET last_run_started_at = now(), last_error = NULL WHERE id = 1 RETURNING watermark_at INTO v_from; v_to := now() - INTERVAL '5 minutes'; IF v_from < v_to THEN v_rows := rollup_task_usage_dashboard_daily_window(v_from, v_to); UPDATE task_usage_dashboard_rollup_state SET watermark_at = v_to, last_run_finished_at = now(), last_run_rows = v_rows WHERE id = 1; ELSE UPDATE task_usage_dashboard_rollup_state SET last_run_finished_at = now(), last_run_rows = 0 WHERE id = 1; END IF; PERFORM pg_advisory_unlock(4244); RETURN v_rows; EXCEPTION WHEN OTHERS THEN UPDATE task_usage_dashboard_rollup_state SET last_error = SQLERRM, last_run_finished_at = now() WHERE id = 1; PERFORM pg_advisory_unlock(4244); RAISE; END; END; $$; -- Health-check helper mirroring task_usage_rollup_lag_seconds() (076). -- Alert via monitoring on NULL > 15 min after deploy, or value > 900s. CREATE OR REPLACE FUNCTION task_usage_dashboard_rollup_lag_seconds() RETURNS DOUBLE PRECISION LANGUAGE sql STABLE AS $$ SELECT EXTRACT(EPOCH FROM (now() - last_run_finished_at)) FROM task_usage_dashboard_rollup_state WHERE id = 1; $$; -- NOTE: cron job is NOT scheduled by this migration — same convention as -- 076 for the per-runtime rollup. Operator playbook: -- 1) Apply migrations through 084. -- 2) Run `go run ./cmd/backfill_task_usage_dashboard_daily`. -- 3) Set USAGE_DASHBOARD_ROLLUP_ENABLED=true on the API and roll out. -- 4) As superuser: -- SELECT cron.schedule( -- 'rollup_task_usage_dashboard_daily', -- '*/5 * * * *', -- $$SELECT rollup_task_usage_dashboard_daily()$$ -- );