create table tenant (
    tenant_id        uuid primary key,
    tenant_code      text not null unique,
    name             text not null,
    status           text not null check (status in ('ACTIVE', 'SUSPENDED', 'ARCHIVED')),
    created_at       timestamptz not null default now()
);

create table app_user (
    user_id          uuid primary key,
    tenant_id        uuid not null references tenant(tenant_id),
    email            text not null,
    display_name     text not null,
    status           text not null check (status in ('ACTIVE', 'DISABLED')),
    created_at       timestamptz not null default now(),
    unique (tenant_id, email)
);

create table case_state_catalog (
    state_code       text primary key,
    state_order      integer not null unique,
    is_terminal      boolean not null,
    description      text not null
);

insert into case_state_catalog(state_code, state_order, is_terminal, description) values
('INTAKE', 10, false, 'Initial intake before triage'),
('TRIAGE', 20, false, 'Triage and jurisdiction check'),
('INVESTIGATION', 30, false, 'Evidence gathering and analysis'),
('REVIEW', 40, false, 'Formal review and decision'),
('REMEDIATION', 50, false, 'Remediation plan execution'),
('MONITORING', 60, false, 'Post-remediation monitoring'),
('ENFORCEMENT_ACTION', 70, false, 'Formal enforcement action'),
('CLOSED_NO_ACTION', 900, true, 'Closed without action'),
('CLOSED_RESOLVED', 910, true, 'Closed after remediation'),
('CLOSED_ENFORCED', 920, true, 'Closed after enforcement');

create table case_transition_catalog (
    transition_code       text primary key,
    from_state            text not null references case_state_catalog(state_code),
    to_state              text not null references case_state_catalog(state_code),
    requires_approval     boolean not null default false,
    requires_reason       boolean not null default true,
    is_active             boolean not null default true,
    unique (from_state, to_state)
);

insert into case_transition_catalog
(transition_code, from_state, to_state, requires_approval, requires_reason) values
('SUBMIT_INTAKE', 'INTAKE', 'TRIAGE', false, true),
('ACCEPT_TRIAGE', 'TRIAGE', 'INVESTIGATION', false, true),
('REJECT_TRIAGE', 'TRIAGE', 'CLOSED_NO_ACTION', true, true),
('COMPLETE_EVIDENCE', 'INVESTIGATION', 'REVIEW', false, true),
('REQUIRE_REMEDIATION', 'REVIEW', 'REMEDIATION', true, true),
('ESCALATE_ENFORCEMENT', 'REVIEW', 'ENFORCEMENT_ACTION', true, true),
('NO_BREACH', 'REVIEW', 'CLOSED_NO_ACTION', true, true),
('ACCEPT_REMEDIATION', 'REMEDIATION', 'MONITORING', false, true),
('COMPLETE_MONITORING', 'MONITORING', 'CLOSED_RESOLVED', true, true),
('FINALIZE_ACTION', 'ENFORCEMENT_ACTION', 'CLOSED_ENFORCED', true, true);

create table case_file (
    case_id              uuid primary key,
    tenant_id            uuid not null references tenant(tenant_id),
    case_number          text not null,
    title                text not null,
    case_type            text not null,
    severity             text not null check (severity in ('LOW', 'MEDIUM', 'HIGH', 'CRITICAL')),
    current_state        text not null references case_state_catalog(state_code),
    current_owner_id     uuid references app_user(user_id),
    opened_at            timestamptz not null default now(),
    closed_at            timestamptz,
    version              bigint not null default 0,
    created_by           uuid not null references app_user(user_id),
    created_at           timestamptz not null default now(),
    updated_at           timestamptz not null default now(),

    unique (tenant_id, case_number),
    check (
        (closed_at is null and current_state not like 'CLOSED_%')
        or
        (closed_at is not null and current_state like 'CLOSED_%')
    )
);

create table case_transition (
    transition_id       uuid primary key,
    tenant_id           uuid not null references tenant(tenant_id),
    case_id             uuid not null references case_file(case_id),
    transition_code     text not null references case_transition_catalog(transition_code),
    from_state          text not null references case_state_catalog(state_code),
    to_state            text not null references case_state_catalog(state_code),
    reason              text not null,
    actor_user_id       uuid not null references app_user(user_id),
    occurred_at         timestamptz not null default now(),
    command_id          uuid not null,
    metadata            jsonb not null default '{}'::jsonb,

    unique (tenant_id, command_id),
    foreign key (transition_code, from_state, to_state)
        references case_transition_catalog(transition_code, from_state, to_state)
);

alter table case_transition_catalog
add constraint uq_transition_catalog_triplet
unique (transition_code, from_state, to_state);

create table case_assignment (
    assignment_id        uuid primary key,
    tenant_id            uuid not null references tenant(tenant_id),
    case_id              uuid not null references case_file(case_id),
    assignee_user_id     uuid not null references app_user(user_id),
    assigned_by          uuid not null references app_user(user_id),
    assigned_at          timestamptz not null default now(),
    unassigned_at        timestamptz,
    assignment_reason    text not null,
    check (unassigned_at is null or unassigned_at > assigned_at)
);

create unique index uq_active_assignment_per_case
on case_assignment(tenant_id, case_id)
where unassigned_at is null;

create table case_evidence (
    evidence_id          uuid primary key,
    tenant_id            uuid not null references tenant(tenant_id),
    case_id              uuid not null references case_file(case_id),
    evidence_type        text not null,
    storage_uri          text not null,
    content_hash         text not null,
    status               text not null check (status in ('UPLOADED', 'ACCEPTED', 'REJECTED', 'REDACTED')),
    submitted_by         uuid not null references app_user(user_id),
    submitted_at         timestamptz not null default now(),
    accepted_by          uuid references app_user(user_id),
    accepted_at          timestamptz,
    metadata             jsonb not null default '{}'::jsonb,

    unique (tenant_id, content_hash),
    check (
        (status = 'ACCEPTED' and accepted_by is not null and accepted_at is not null)
        or
        (status <> 'ACCEPTED')
    )
);

create table case_deadline (
    deadline_id          uuid primary key,
    tenant_id            uuid not null references tenant(tenant_id),
    case_id              uuid not null references case_file(case_id),
    deadline_type        text not null,
    due_at               timestamptz not null,
    status               text not null check (status in ('ACTIVE', 'MET', 'MISSED', 'CANCELLED')),
    created_at           timestamptz not null default now(),
    resolved_at          timestamptz,
    check (resolved_at is null or resolved_at >= created_at)
);

create unique index uq_active_deadline_per_case_type
on case_deadline(tenant_id, case_id, deadline_type)
where status = 'ACTIVE';

create table case_approval (
    approval_id          uuid primary key,
    tenant_id            uuid not null references tenant(tenant_id),
    case_id              uuid not null references case_file(case_id),
    transition_code      text not null references case_transition_catalog(transition_code),
    requested_by         uuid not null references app_user(user_id),
    requested_at         timestamptz not null default now(),
    decided_by           uuid references app_user(user_id),
    decided_at           timestamptz,
    decision             text not null check (decision in ('PENDING', 'APPROVED', 'REJECTED', 'CANCELLED')),
    reason               text,
    command_id           uuid not null,

    unique (tenant_id, command_id),
    check (
        (decision = 'PENDING' and decided_by is null and decided_at is null)
        or
        (decision <> 'PENDING' and decided_by is not null and decided_at is not null)
    )
);

create unique index uq_pending_approval_per_case_transition
on case_approval(tenant_id, case_id, transition_code)
where decision = 'PENDING';

create table outbox_event (
    outbox_id          uuid primary key,
    tenant_id          uuid not null references tenant(tenant_id),
    aggregate_type     text not null,
    aggregate_id       uuid not null,
    event_type         text not null,
    event_version      integer not null,
    payload            jsonb not null,
    created_at         timestamptz not null default now(),
    published_at       timestamptz,
    attempts           integer not null default 0,
    last_error         text,
    command_id         uuid not null,

    unique (tenant_id, command_id, event_type)
);

create index idx_outbox_unpublished
on outbox_event(created_at, outbox_id)
where published_at is null;

1. update case state in database
2. publish event to broker

1. update case state
2. insert outbox event
3. commit both together
4. relay publishes outbox later

update case_file
set current_state = 'INVESTIGATION'
where case_id = :case_id;

begin;

with target_case as (
    select
        c.case_id,
        c.tenant_id,
        c.current_state,
        c.version,
        s.is_terminal
    from case_file c
    join case_state_catalog s
      on s.state_code = c.current_state
    where c.tenant_id = :tenant_id
      and c.case_id = :case_id
    for update
), valid_transition as (
    select
        tc.case_id,
        tc.tenant_id,
        tc.current_state as from_state,
        t.to_state,
        t.transition_code,
        t.requires_approval
    from target_case tc
    join case_transition_catalog t
      on t.from_state = tc.current_state
     and t.transition_code = :transition_code
     and t.is_active = true
    where tc.is_terminal = false
      and tc.version = :expected_version
), updated_case as (
    update case_file c
    set current_state = vt.to_state,
        closed_at = case when vt.to_state like 'CLOSED_%' then now() else c.closed_at end,
        version = c.version + 1,
        updated_at = now()
    from valid_transition vt
    where c.tenant_id = vt.tenant_id
      and c.case_id = vt.case_id
      and vt.requires_approval = false
    returning
        c.case_id,
        c.tenant_id,
        vt.transition_code,
        vt.from_state,
        c.current_state as to_state,
        c.version
), inserted_transition as (
    insert into case_transition(
        transition_id,
        tenant_id,
        case_id,
        transition_code,
        from_state,
        to_state,
        reason,
        actor_user_id,
        command_id,
        metadata
    )
    select
        gen_random_uuid(),
        uc.tenant_id,
        uc.case_id,
        uc.transition_code,
        uc.from_state,
        uc.to_state,
        :reason,
        :actor_user_id,
        :command_id,
        jsonb_build_object('new_version', uc.version)
    from updated_case uc
    on conflict (tenant_id, command_id) do nothing
    returning transition_id, tenant_id, case_id, transition_code, from_state, to_state
), inserted_outbox as (
    insert into outbox_event(
        outbox_id,
        tenant_id,
        aggregate_type,
        aggregate_id,
        event_type,
        event_version,
        payload,
        command_id
    )
    select
        gen_random_uuid(),
        it.tenant_id,
        'case_file',
        it.case_id,
        'CaseStateChanged',
        1,
        jsonb_build_object(
            'case_id', it.case_id,
            'transition_code', it.transition_code,
            'from_state', it.from_state,
            'to_state', it.to_state,
            'actor_user_id', :actor_user_id,
            'reason', :reason
        ),
        :command_id
    from inserted_transition it
    on conflict (tenant_id, command_id, event_type) do nothing
    returning outbox_id
)
select
    (select count(*) from updated_case) as updated_count,
    (select count(*) from inserted_transition) as transition_count,
    (select count(*) from inserted_outbox) as outbox_count;

commit;

update case_file
set current_state = 'ENFORCEMENT_ACTION'
where case_id = :case_id;

insert into case_approval(
    approval_id,
    tenant_id,
    case_id,
    transition_code,
    requested_by,
    decision,
    reason,
    command_id
)
select
    gen_random_uuid(),
    c.tenant_id,
    c.case_id,
    t.transition_code,
    :requested_by,
    'PENDING',
    :reason,
    :command_id
from case_file c
join case_transition_catalog t
  on t.from_state = c.current_state
 and t.transition_code = :transition_code
where c.tenant_id = :tenant_id
  and c.case_id = :case_id
  and t.requires_approval = true
  and t.is_active = true
on conflict (tenant_id, command_id) do nothing;

begin;

with decided as (
    update case_approval a
    set decision = :decision,
        decided_by = :decided_by,
        decided_at = now(),
        reason = coalesce(:decision_reason, a.reason)
    where a.tenant_id = :tenant_id
      and a.approval_id = :approval_id
      and a.decision = 'PENDING'
    returning *
), outbox as (
    insert into outbox_event(
        outbox_id,
        tenant_id,
        aggregate_type,
        aggregate_id,
        event_type,
        event_version,
        payload,
        command_id
    )
    select
        gen_random_uuid(),
        d.tenant_id,
        'case_approval',
        d.approval_id,
        'CaseApprovalDecided',
        1,
        jsonb_build_object(
            'approval_id', d.approval_id,
            'case_id', d.case_id,
            'transition_code', d.transition_code,
            'decision', d.decision,
            'decided_by', d.decided_by
        ),
        :command_id
    from decided d
    on conflict (tenant_id, command_id, event_type) do nothing
)
select count(*) from decided;

commit;

select
    c.case_id,
    c.case_number,
    c.title,
    c.severity,
    c.current_state,
    d.due_at,
    c.updated_at
from case_file c
left join case_deadline d
  on d.tenant_id = c.tenant_id
 and d.case_id = c.case_id
 and d.deadline_type = 'INVESTIGATION_DUE'
 and d.status = 'ACTIVE'
where c.tenant_id = :tenant_id
  and c.current_owner_id = :user_id
  and c.current_state in ('INVESTIGATION', 'REVIEW')
order by
    d.due_at nulls last,
    case c.severity
        when 'CRITICAL' then 1
        when 'HIGH' then 2
        when 'MEDIUM' then 3
        else 4
    end,
    c.updated_at desc,
    c.case_id
fetch first 50 rows only;

create index idx_case_worklist_owner_state
on case_file(
    tenant_id,
    current_owner_id,
    current_state,
    updated_at desc,
    case_id
)
where current_state in ('INVESTIGATION', 'REVIEW');

create index idx_active_deadline_case_type
on case_deadline(tenant_id, case_id, deadline_type, due_at)
where status = 'ACTIVE';

select
    event_type,
    occurred_at,
    actor_user_id,
    summary,
    payload
from (
    select
        'STATE_TRANSITION' as event_type,
        t.occurred_at,
        t.actor_user_id,
        t.from_state || ' -> ' || t.to_state as summary,
        jsonb_build_object(
            'transition_code', t.transition_code,
            'reason', t.reason
        ) as payload
    from case_transition t
    where t.tenant_id = :tenant_id
      and t.case_id = :case_id

    union all

    select
        'EVIDENCE_SUBMITTED' as event_type,
        e.submitted_at as occurred_at,
        e.submitted_by as actor_user_id,
        e.evidence_type || ' evidence submitted' as summary,
        jsonb_build_object(
            'evidence_id', e.evidence_id,
            'status', e.status,
            'content_hash', e.content_hash
        ) as payload
    from case_evidence e
    where e.tenant_id = :tenant_id
      and e.case_id = :case_id

    union all

    select
        'ASSIGNMENT' as event_type,
        a.assigned_at as occurred_at,
        a.assigned_by as actor_user_id,
        'Assigned to user' as summary,
        jsonb_build_object(
            'assignee_user_id', a.assignee_user_id,
            'unassigned_at', a.unassigned_at,
            'reason', a.assignment_reason
        ) as payload
    from case_assignment a
    where a.tenant_id = :tenant_id
      and a.case_id = :case_id
) timeline
order by occurred_at, event_type;

create index idx_transition_timeline
on case_transition(tenant_id, case_id, occurred_at, transition_id);

create index idx_evidence_timeline
on case_evidence(tenant_id, case_id, submitted_at, evidence_id);

create index idx_assignment_timeline
on case_assignment(tenant_id, case_id, assigned_at, assignment_id);

select
    c.tenant_id,
    c.case_id,
    c.case_number,
    c.current_state,
    c.current_owner_id,
    d.deadline_type,
    d.due_at
from case_deadline d
join case_file c
  on c.tenant_id = d.tenant_id
 and c.case_id = d.case_id
where d.status = 'ACTIVE'
  and d.due_at < now()
  and c.current_state not like 'CLOSED_%'
order by d.due_at, c.case_id;

create index idx_deadline_active_due
on case_deadline(status, due_at, tenant_id, case_id)
where status = 'ACTIVE';

create index idx_deadline_active_due_tenant
on case_deadline(due_at, tenant_id, case_id)
where status = 'ACTIVE';

update case_deadline
set status = 'MISSED',
    resolved_at = now()
where status = 'ACTIVE'
  and due_at < now()
returning tenant_id, case_id, deadline_type, due_at;

with overdue as (
    select deadline_id
    from case_deadline
    where status = 'ACTIVE'
      and due_at < now()
    order by due_at, deadline_id
    for update skip locked
    limit 500
), updated as (
    update case_deadline d
    set status = 'MISSED',
        resolved_at = now()
    from overdue o
    where d.deadline_id = o.deadline_id
    returning d.*
)
select count(*) from updated;

begin;
select version, current_state
from case_file
where tenant_id = :tenant_id and case_id = :case_id;

begin;
select version, current_state
from case_file
where tenant_id = :tenant_id and case_id = :case_id;

update case_file
set current_state = 'INVESTIGATION', version = version + 1
where tenant_id = :tenant_id
  and case_id = :case_id;

commit;

update case_file
set current_state = 'CLOSED_NO_ACTION', version = version + 1
where tenant_id = :tenant_id
  and case_id = :case_id;

commit;

update case_file
set current_state = :to_state,
    version = version + 1
where tenant_id = :tenant_id
  and case_id = :case_id
  and version = :expected_version
  and current_state = :expected_from_state;

select count(*)
from case_assignment
where case_id = :case_id
  and role = 'REVIEWER'
  and unassigned_at is null;

-- if count > 1, unassign self

begin;
update case_file set updated_at = now() where case_id = :case_1;
update case_file set updated_at = now() where case_id = :case_2;
commit;

begin;
update case_file set updated_at = now() where case_id = :case_2;
update case_file set updated_at = now() where case_id = :case_1;
commit;

select case_id
from case_file
where case_id in (:case_1, :case_2)
order by case_id
for update;

select *
from case_file
where tenant_id = :tenant_id
  and current_owner_id = :user_id
  and lower(current_state) = 'investigation'
order by updated_at desc
limit 50;

select
    case_id,
    case_number,
    title,
    severity,
    current_state,
    updated_at
from case_file
where tenant_id = :tenant_id
  and current_owner_id = :user_id
  and current_state = 'INVESTIGATION'
order by updated_at desc, case_id
limit 50;

create index idx_case_worklist_investigation
on case_file(tenant_id, current_owner_id, updated_at desc, case_id)
where current_state = 'INVESTIGATION';

select c.case_id
from case_file c
left join case_transition t
  on t.tenant_id = c.tenant_id
 and t.case_id = c.case_id
 and t.from_state = 'INTAKE'
where c.created_at < now() - interval '5 minutes'
  and t.transition_id is null;

with latest_transition as (
    select distinct on (tenant_id, case_id)
        tenant_id,
        case_id,
        to_state,
        occurred_at
    from case_transition
    order by tenant_id, case_id, occurred_at desc, transition_id desc
)
select
    c.tenant_id,
    c.case_id,
    c.current_state,
    lt.to_state as latest_transition_state
from case_file c
join latest_transition lt
  on lt.tenant_id = c.tenant_id
 and lt.case_id = c.case_id
where c.current_state <> lt.to_state;

select
    c.tenant_id,
    c.case_id,
    c.current_owner_id,
    a.assignee_user_id as active_assignment_user_id
from case_file c
left join case_assignment a
  on a.tenant_id = c.tenant_id
 and a.case_id = c.case_id
 and a.unassigned_at is null
where c.current_owner_id is distinct from a.assignee_user_id;

select tenant_id, case_id, deadline_type, count(*) as active_count
from case_deadline
where status = 'ACTIVE'
group by tenant_id, case_id, deadline_type
having count(*) > 1;

LEGAL_REVIEW

alter table case_file
add constraint chk_state
check (current_state in (..., 'LEGAL_REVIEW'));

update application immediately;

insert into case_state_catalog(state_code, state_order, is_terminal, description)
values ('LEGAL_REVIEW', 65, false, 'Legal review before enforcement action');

insert into case_transition_catalog
(transition_code, from_state, to_state, requires_approval, requires_reason, is_active)
values
('SEND_TO_LEGAL_REVIEW', 'REVIEW', 'LEGAL_REVIEW', true, true, false),
('LEGAL_APPROVE_ENFORCEMENT', 'LEGAL_REVIEW', 'ENFORCEMENT_ACTION', true, true, false),
('LEGAL_NO_ACTION', 'LEGAL_REVIEW', 'CLOSED_NO_ACTION', true, true, false);

-- Example report grouping should not silently bucket unknown states.
select current_state, count(*)
from case_file
group by current_state
order by current_state;

update case_transition_catalog
set is_active = true
where transition_code in (
    'SEND_TO_LEGAL_REVIEW',
    'LEGAL_APPROVE_ENFORCEMENT',
    'LEGAL_NO_ACTION'
);

select case_id
from case_file
where current_state = 'LEGAL_REVIEW'
  and severity not in ('HIGH', 'CRITICAL');

where tenant_id = :tenant_id

alter table case_file enable row level security;

create policy tenant_case_isolation
on case_file
using (tenant_id = current_setting('app.tenant_id')::uuid);

begin;
select set_config('app.tenant_id', :tenant_id::text, true);
-- tenant-scoped queries
commit;

select *
from case_file
where tenant_id = $1
  and case_number = $2;

"select * from case_file where case_number = '" + input + "'"

allowed_sort = {
  "updatedAt": "updated_at",
  "severity": "severity",
  "dueAt": "due_at"
}

with next_events as (
    select outbox_id
    from outbox_event
    where published_at is null
    order by created_at, outbox_id
    for update skip locked
    limit 100
)
select o.*
from outbox_event o
join next_events n
  on n.outbox_id = o.outbox_id;

update outbox_event
set published_at = now(),
    attempts = attempts + 1,
    last_error = null
where outbox_id = any(:published_outbox_ids);

update outbox_event
set attempts = attempts + 1,
    last_error = :error
where outbox_id = :outbox_id;

create table consumed_event (
    consumer_name       text not null,
    event_id            uuid not null,
    consumed_at         timestamptz not null default now(),
    primary key (consumer_name, event_id)
);

begin;

insert into consumed_event(consumer_name, event_id)
values (:consumer_name, :event_id)
on conflict do nothing;

-- If row inserted, apply side effect.
-- If conflict, skip.

commit;

select
    date_trunc('week', opened_at) as week_start,
    count(*) as opened_cases,
    count(*) filter (where current_state like 'CLOSED_%') as currently_closed_cases
from case_file
where tenant_id = :tenant_id
  and opened_at >= :from_date
  and opened_at < :to_date
group by 1
order by 1;

select
    date_trunc('week', closed_at) as week_start,
    count(*) as closed_cases
from case_file
where tenant_id = :tenant_id
  and closed_at >= :from_date
  and closed_at < :to_date
group by 1
order by 1;

with ordered as (
    select
        tenant_id,
        case_id,
        from_state,
        to_state,
        occurred_at,
        lead(occurred_at) over (
            partition by tenant_id, case_id
            order by occurred_at, transition_id
        ) as next_occurred_at
    from case_transition
    where tenant_id = :tenant_id
), state_spans as (
    select
        case_id,
        to_state as state_code,
        occurred_at as entered_at,
        coalesce(next_occurred_at, now()) as exited_at
    from ordered
)
select
    state_code,
    percentile_cont(0.5) within group (order by exited_at - entered_at) as p50_duration,
    percentile_cont(0.95) within group (order by exited_at - entered_at) as p95_duration
from state_spans
group by state_code
order by state_code;

with case_flags as (
    select
        tenant_id,
        case_id,
        bool_or(to_state = 'TRIAGE') as reached_triage,
        bool_or(to_state = 'INVESTIGATION') as reached_investigation,
        bool_or(to_state = 'REVIEW') as reached_review,
        bool_or(to_state in ('CLOSED_RESOLVED', 'CLOSED_ENFORCED')) as reached_positive_closure
    from case_transition
    where tenant_id = :tenant_id
      and occurred_at >= :from_date
      and occurred_at < :to_date
    group by tenant_id, case_id
)
select
    count(*) as total_cases,
    count(*) filter (where reached_triage) as triage,
    count(*) filter (where reached_investigation) as investigation,
    count(*) filter (where reached_review) as review,
    count(*) filter (where reached_positive_closure) as positive_closure
from case_flags;

select
    count(*) as unpublished_count,
    min(created_at) as oldest_unpublished_at,
    now() - min(created_at) as oldest_age
from outbox_event
where published_at is null;

learn-sql-in-action-part-001-kaufman-skill-map-and-sql-operating-model.mdx

learn-sql-in-action-part-035-capstone-sql-in-action-production-casebook.mdx