Routine: enforcement.transition_case(...)
Kind: command function
Owner: Enforcement Platform Team
Caller: case-management-api, escalation-worker
Mutation scope: one case + one transition event + optional outbox row
Expected latency: p95 < 50 ms, p99 < 250 ms
Max affected domain rows: case=1, transition_event=1, outbox<=1
Concurrency model: row lock on enforcement_case(id)
Idempotency: request_id unique per command source
Retryable SQLSTATE: 40001, 40P01
Terminal SQLSTATE: P24xx domain validation failures
Alert signals: error rate, p99 latency, lock wait, outbox lag, transition rejection spike
Rollback path: deploy previous function body if signature unchanged; otherwise compatibility wrapper
Runbook: RUNBOOK-ENF-CASE-TRANSITION

select
  pid,
  usename,
  application_name,
  state,
  wait_event_type,
  wait_event,
  now() - xact_start as xact_age,
  now() - query_start as query_age,
  left(query, 160) as query_preview
from pg_stat_activity
where datname = current_database()
order by xact_start nulls last, query_start nulls last;

select
  a.pid,
  a.application_name,
  a.state,
  l.locktype,
  l.mode,
  l.granted,
  l.relation::regclass as relation_name,
  now() - a.query_start as query_age,
  left(a.query, 120) as query_preview
from pg_locks l
join pg_stat_activity a on a.pid = l.pid
where a.datname = current_database()
order by l.granted, query_age desc;

select
  schemaname,
  funcname,
  calls,
  round(total_time::numeric, 2) as total_ms,
  round(self_time::numeric, 2) as self_ms,
  round((total_time / nullif(calls, 0))::numeric, 3) as avg_ms
from pg_stat_user_functions
where calls > 0
order by total_time desc
limit 30;

select
  calls,
  round(total_exec_time::numeric, 2) as total_exec_ms,
  round(mean_exec_time::numeric, 3) as mean_exec_ms,
  rows,
  left(query, 180) as query_preview
from pg_stat_statements
where dbid = (select oid from pg_database where datname = current_database())
order by total_exec_time desc
limit 30;

-- Example for a controlled diagnostic session, not a blanket production recommendation.
load 'auto_explain';
set auto_explain.log_min_duration = '500ms';
set auto_explain.log_analyze = on;
set auto_explain.log_buffers = on;
set auto_explain.log_nested_statements = on;

raise notice 'failed here';

raise warning
  'op=case_transition status=rejected case_id=% from=% to=% reason=% request_id=%',
  p_case_id,
  v_current_status,
  p_target_status,
  v_reason_code,
  p_request_id;

create schema if not exists ops;

create table if not exists ops.plpgsql_operation_event (
  event_id       bigint generated always as identity primary key,
  occurred_at    timestamptz not null default clock_timestamp(),
  schema_name    text not null,
  routine_name   text not null,
  operation      text not null,
  severity       text not null check (severity in ('debug', 'info', 'warning', 'error')),
  request_id     text,
  actor_id       text,
  subject_type   text,
  subject_id     text,
  sqlstate       text,
  message        text not null,
  detail         jsonb not null default '{}'::jsonb
);

create index if not exists idx_plpgsql_operation_event_lookup
  on ops.plpgsql_operation_event (routine_name, occurred_at desc);

create index if not exists idx_plpgsql_operation_event_request
  on ops.plpgsql_operation_event (request_id)
  where request_id is not null;

create or replace function ops.record_plpgsql_event(
  p_schema_name  text,
  p_routine_name text,
  p_operation    text,
  p_severity     text,
  p_request_id   text,
  p_actor_id     text,
  p_subject_type text,
  p_subject_id   text,
  p_sqlstate     text,
  p_message      text,
  p_detail       jsonb default '{}'::jsonb
)
returns void
language plpgsql
security definer
set search_path = ops, pg_catalog
as $$
begin
  insert into ops.plpgsql_operation_event (
    schema_name,
    routine_name,
    operation,
    severity,
    request_id,
    actor_id,
    subject_type,
    subject_id,
    sqlstate,
    message,
    detail
  ) values (
    p_schema_name,
    p_routine_name,
    p_operation,
    p_severity,
    p_request_id,
    p_actor_id,
    p_subject_type,
    p_subject_id,
    p_sqlstate,
    p_message,
    coalesce(p_detail, '{}'::jsonb)
  );
end;
$$;

select set_config('app.request_id', :request_id, true);
select set_config('app.actor_id', :actor_id, true);
select set_config('app.source', 'case-management-api', true);

create or replace function app_context.request_id()
returns text
language sql
stable
as $$
  select nullif(current_setting('app.request_id', true), '')
$$;

create or replace function app_context.actor_id()
returns text
language sql
stable
as $$
  select nullif(current_setting('app.actor_id', true), '')
$$;

perform ops.record_plpgsql_event(
  'enforcement',
  'transition_case',
  'transition_rejected',
  'warning',
  app_context.request_id(),
  app_context.actor_id(),
  'case',
  p_case_id::text,
  'P2401',
  'Invalid transition rejected',
  jsonb_build_object(
    'from_status', v_current_status,
    'to_status', p_target_status,
    'reason_code', v_reason_code
  )
);

select
  count(*) filter (where status = 'pending') as pending_count,
  min(created_at) filter (where status = 'pending') as oldest_pending_at,
  now() - min(created_at) filter (where status = 'pending') as oldest_pending_age
from integration.outbox_event;

select
  run_type,
  count(*) filter (where status = 'failed') as failed_runs,
  max(finished_at) filter (where status = 'failed') as latest_failure_at
from ops.batch_run
where started_at >= now() - interval '1 hour'
group by run_type
having count(*) filter (where status = 'failed') > 0;

with transitions as (
  select date_trunc('minute', occurred_at) as bucket, count(*) as transition_count
  from enforcement.case_transition_event
  where occurred_at >= now() - interval '1 hour'
  group by 1
), audits as (
  select date_trunc('minute', occurred_at) as bucket, count(*) as audit_count
  from audit.row_change_event
  where table_name = 'enforcement_case'
    and occurred_at >= now() - interval '1 hour'
  group by 1
)
select
  coalesce(t.bucket, a.bucket) as bucket,
  coalesce(t.transition_count, 0) as transition_count,
  coalesce(a.audit_count, 0) as audit_count
from transitions t
full join audits a using (bucket)
where coalesce(t.transition_count, 0) <> coalesce(a.audit_count, 0)
order by bucket;

select
  pid,
  application_name,
  usename,
  state,
  now() - xact_start as xact_age,
  wait_event_type,
  wait_event,
  left(query, 120) as query_preview
from pg_stat_activity
where xact_start is not null
  and now() - xact_start > interval '5 minutes'
order by xact_age desc;

select
  datname,
  deadlocks,
  temp_files,
  temp_bytes,
  conflicts
from pg_stat_database
where datname = current_database();

# RUNBOOK: <Routine / Workflow Name>

## Owner
- Team:
- Slack / Pager:
- Escalation:

## Purpose
What business workflow does this protect?

## Normal Behavior
- Expected call rate:
- Expected latency:
- Expected queue age:
- Expected row effect:
- Expected SQLSTATEs:

## Alert Signals
- Alert name:
- Query / dashboard:
- Threshold:
- Severity:

## First Checks
1. Check active sessions.
2. Check lock waits.
3. Check latest failures.
4. Check run ledger.
5. Check recent deploys.

## Diagnosis Branches
- If queue is stale:
- If lock wait is high:
- If domain rejection spike:
- If deadlock spike:
- If audit gap:

## Safe Actions
- Retry command:
- Resume batch:
- Disable worker:
- Pause caller:
- Rebuild derived table:
- Roll forward migration:

## Dangerous Actions
- Do not truncate:
- Do not manually update status:
- Do not disable trigger except under approved incident procedure:

## Rollback / Forward Fix
- Compatible rollback:
- Function body rollback:
- Data repair:
- Validation query:

## Post-Incident Checks
- Invariant checks:
- Audit checks:
- Replay checks:
- Customer/case impact query:

select
  status,
  count(*) as task_count,
  min(due_at) as oldest_due_at,
  max(updated_at) as latest_update_at
from enforcement.escalation_task
group by status
order by status;

select
  pid,
  application_name,
  state,
  wait_event_type,
  wait_event,
  now() - query_start as query_age,
  left(query, 160) as query_preview
from pg_stat_activity
where application_name like 'escalation-worker%'
order by query_age desc;

select
  a.pid,
  a.application_name,
  l.mode,
  l.granted,
  l.relation::regclass as relation_name,
  now() - a.query_start as query_age,
  left(a.query, 120) as query_preview
from pg_locks l
join pg_stat_activity a on a.pid = l.pid
where a.datname = current_database()
  and (
    l.relation = 'enforcement.escalation_task'::regclass
    or a.query ilike '%escalation_task%'
  )
order by granted, query_age desc;

select
  occurred_at,
  severity,
  request_id,
  subject_id,
  sqlstate,
  message,
  detail
from ops.plpgsql_operation_event
where routine_name in ('claim_due_escalations', 'process_escalation_task')
  and occurred_at >= now() - interval '2 hours'
order by occurred_at desc
limit 100;

update enforcement.escalation_task
set
  status = 'pending',
  locked_by = null,
  locked_at = null,
  updated_at = clock_timestamp(),
  retry_count = retry_count + 1
where status = 'processing'
  and locked_at < clock_timestamp() - interval '15 minutes'
  and retry_count < 5;

select
  count(*) filter (where status = 'pending') as pending,
  count(*) filter (where status = 'processing') as processing,
  count(*) filter (where status = 'failed') as failed,
  min(due_at) filter (where status = 'pending') as oldest_pending_due
from enforcement.escalation_task;

create table if not exists ops.procedure_run (
  run_id          bigint generated always as identity primary key,
  procedure_name  text not null,
  requested_by    text,
  request_id      text,
  status          text not null check (status in ('running', 'succeeded', 'failed', 'cancelled')),
  started_at      timestamptz not null default clock_timestamp(),
  finished_at     timestamptz,
  input           jsonb not null default '{}'::jsonb,
  result          jsonb not null default '{}'::jsonb,
  error_sqlstate  text,
  error_message   text
);

create index if not exists idx_procedure_run_name_started
  on ops.procedure_run (procedure_name, started_at desc);

create unique index if not exists uq_procedure_run_request
  on ops.procedure_run (procedure_name, request_id)
  where request_id is not null;

create or replace procedure ops.run_partition_maintenance(
  p_request_id text,
  p_until_date date
)
language plpgsql
as $$
declare
  v_run_id bigint;
  v_rows_processed bigint := 0;
  v_sqlstate text;
  v_message text;
begin
  insert into ops.procedure_run (
    procedure_name,
    request_id,
    requested_by,
    status,
    input
  ) values (
    'ops.run_partition_maintenance',
    p_request_id,
    current_user,
    'running',
    jsonb_build_object('until_date', p_until_date)
  )
  returning run_id into v_run_id;

  -- Work happens here.
  -- Keep phases visible and bounded.

  update ops.procedure_run
  set
    status = 'succeeded',
    finished_at = clock_timestamp(),
    result = jsonb_build_object('rows_processed', v_rows_processed)
  where run_id = v_run_id;

exception
  when others then
    get stacked diagnostics
      v_sqlstate = returned_sqlstate,
      v_message = message_text;

    update ops.procedure_run
    set
      status = 'failed',
      finished_at = clock_timestamp(),
      error_sqlstate = v_sqlstate,
      error_message = v_message
    where run_id = v_run_id;

    raise;
end;
$$;

create table if not exists ops.runtime_switch (
  switch_name text primary key,
  enabled boolean not null,
  updated_at timestamptz not null default clock_timestamp(),
  updated_by text not null default current_user,
  reason text not null
);

if exists (
  select 1
  from ops.runtime_switch s
  where s.switch_name = 'enforcement.escalation.enabled'
    and s.enabled = false
) then
  raise exception using
    errcode = 'P3701',
    message = 'Escalation processing is temporarily disabled',
    detail = 'Runtime switch enforcement.escalation.enabled is false.';
end if;

get diagnostics v_row_count = row_count;

if v_row_count > p_max_rows then
  raise exception using
    errcode = 'P3702',
    message = 'Mutation exceeded max row guard',
    detail = format('affected=%s max=%s', v_row_count, p_max_rows);
end if;

update account
set status = 'inactive'
where last_seen_at < now() - interval '2 years';

with candidate as (
  select account_id
  from account
  where last_seen_at < clock_timestamp() - interval '2 years'
    and status = 'active'
  order by account_id
  limit p_limit
), updated as (
  update account a
  set status = 'inactive'
  from candidate c
  where a.account_id = c.account_id
  returning a.account_id
)
select count(*) into v_row_count
from updated;

create table if not exists ops.approved_routine_hash (
  routine_identity text primary key,
  approved_hash text not null,
  approved_at timestamptz not null default clock_timestamp(),
  approved_by text not null,
  release_id text not null
);

select
  n.nspname as schema_name,
  p.proname as function_name,
  pg_get_function_identity_arguments(p.oid) as identity_arguments,
  md5(pg_get_functiondef(p.oid)) as current_hash,
  h.approved_hash,
  h.release_id
from pg_proc p
join pg_namespace n on n.oid = p.pronamespace
left join ops.approved_routine_hash h
  on h.routine_identity = n.nspname || '.' || p.proname || '(' || pg_get_function_identity_arguments(p.oid) || ')'
where n.nspname not in ('pg_catalog', 'information_schema')
  and p.prolang = (select oid from pg_language where lanname = 'plpgsql')
  and h.approved_hash is distinct from md5(pg_get_functiondef(p.oid))
order by schema_name, function_name;

begin;

create temporary table repair_candidate as
select case_id, status, escalated_at
from enforcement.enforcement_case
where status = 'open'
  and escalated_at is not null
  and escalation_state = 'not_escalated';

-- Review candidate count before mutation.
select count(*) from repair_candidate;

update enforcement.enforcement_case c
set escalation_state = 'escalated'
from repair_candidate r
where c.case_id = r.case_id;

-- Guard.
do $$
declare
  v_count bigint;
begin
  get diagnostics v_count = row_count;
  if v_count > 500 then
    raise exception 'repair affected too many rows: %', v_count;
  end if;
end;
$$;

insert into audit.manual_repair_event (
  incident_id,
  repaired_by,
  reason,
  candidate_count,
  script_hash
)
select
  'INC-2026-00123',
  current_user,
  'Repair escalation_state derived field after deployment bug',
  count(*),
  'sha256:...'
from repair_candidate;

commit;

## PL/pgSQL Operational Readiness Review

### Identity
- [ ] Routine name is stable and intention-revealing.
- [ ] Owner team is recorded.
- [ ] Callers are known.
- [ ] Deployment migration is reviewed.

### Runtime
- [ ] Runtime is bounded.
- [ ] Batch size is bounded.
- [ ] Row effect is checked.
- [ ] Locking behavior is known.
- [ ] Concurrent call behavior is tested.

### Failure
- [ ] SQLSTATE contract exists.
- [ ] Retryable and terminal errors are separated.
- [ ] Exceptions do not swallow unknown failures.
- [ ] Failure preserves useful diagnostics.

### Observability
- [ ] Function/statement stats can be inspected.
- [ ] Logs contain correlation context.
- [ ] Critical business events are durable.
- [ ] Dashboard panels exist.
- [ ] Alerts point to runbooks.

### Recovery
- [ ] Failed work can resume or be safely abandoned.
- [ ] Manual repair path is documented.
- [ ] Rollback or forward-fix path exists.
- [ ] Data validation queries exist.

### Security
- [ ] Privileges are least-privilege.
- [ ] `SECURITY DEFINER` is justified and hardened.
- [ ] `search_path` is explicitly set where needed.
- [ ] Sensitive data is not logged.

### Lifecycle
- [ ] Function hash/version is tracked.
- [ ] Change is compatible with rolling application deploy.
- [ ] Retirement path exists for obsolete routines.