CREATE TABLE data_classification_catalog (
  catalog_id bigserial PRIMARY KEY,
  schema_name text NOT NULL,
  table_name text NOT NULL,
  column_name text NOT NULL,
  classification text NOT NULL CHECK (
    classification IN ('public', 'internal', 'confidential', 'pii', 'sensitive_pii', 'secret', 'legal_evidence')
  ),
  purpose_code text NOT NULL,
  retention_policy_code text NOT NULL,
  masking_policy_code text,
  owner_team text NOT NULL,
  reviewed_at timestamptz NOT NULL DEFAULT now(),
  UNIQUE (schema_name, table_name, column_name)
);

CREATE TABLE case_subject (
  subject_id uuid PRIMARY KEY,
  full_name text,
  date_of_birth date,
  national_id text,
  passport_number text,
  phone text,
  email text,
  home_address text,
  employer text,
  bank_account text,
  notes text
);

CREATE TABLE case_subject (
  subject_id uuid PRIMARY KEY,
  tenant_id uuid NOT NULL,
  display_name text NOT NULL,
  subject_type text NOT NULL,
  created_at timestamptz NOT NULL DEFAULT now()
);

CREATE TABLE subject_contact_point (
  contact_point_id uuid PRIMARY KEY,
  subject_id uuid NOT NULL REFERENCES case_subject(subject_id),
  contact_type text NOT NULL CHECK (contact_type IN ('email', 'phone', 'postal_address')),
  contact_value_encrypted bytea NOT NULL,
  purpose_code text NOT NULL,
  verified_at timestamptz,
  expires_at timestamptz,
  created_at timestamptz NOT NULL DEFAULT now()
);

CREATE TABLE subject_sensitive_identifier (
  identifier_id uuid PRIMARY KEY,
  subject_id uuid NOT NULL REFERENCES case_subject(subject_id),
  identifier_type text NOT NULL,
  identifier_hash bytea NOT NULL,
  identifier_encrypted bytea,
  purpose_code text NOT NULL,
  retention_policy_code text NOT NULL,
  created_at timestamptz NOT NULL DEFAULT now()
);

CREATE TABLE processing_purpose (
  purpose_code text PRIMARY KEY,
  description text NOT NULL,
  lawful_basis text,
  owner_team text NOT NULL,
  default_retention_policy_code text NOT NULL,
  status text NOT NULL CHECK (status IN ('active', 'retired')),
  created_at timestamptz NOT NULL DEFAULT now()
);

CREATE TABLE personal_data_processing_record (
  processing_id uuid PRIMARY KEY,
  subject_id uuid NOT NULL,
  purpose_code text NOT NULL REFERENCES processing_purpose(purpose_code),
  source_system text NOT NULL,
  collected_at timestamptz NOT NULL,
  valid_until timestamptz,
  consent_reference_id uuid,
  metadata jsonb NOT NULL DEFAULT '{}'::jsonb
);

SELECT purpose_code, count(*)
FROM personal_data_processing_record
GROUP BY purpose_code;

SELECT *
FROM personal_data_processing_record p
WHERE p.valid_until < now()
  AND NOT EXISTS (
    SELECT 1
    FROM legal_hold h
    WHERE h.subject_id = p.subject_id
      AND h.status = 'active'
  );

CREATE TABLE retention_policy (
  retention_policy_code text PRIMARY KEY,
  description text NOT NULL,
  retention_period interval NOT NULL,
  action text NOT NULL CHECK (action IN ('delete', 'anonymize', 'pseudonymize', 'archive', 'review')),
  legal_basis text,
  owner_team text NOT NULL,
  status text NOT NULL CHECK (status IN ('active', 'retired'))
);

ALTER TABLE subject_contact_point
ADD COLUMN retention_policy_code text NOT NULL REFERENCES retention_policy(retention_policy_code),
ADD COLUMN retain_until timestamptz;

UPDATE subject_contact_point cp
SET retain_until = cp.created_at + rp.retention_period
FROM retention_policy rp
WHERE rp.retention_policy_code = cp.retention_policy_code
  AND cp.retain_until IS NULL;

CREATE TABLE legal_hold (
  legal_hold_id uuid PRIMARY KEY,
  hold_scope text NOT NULL CHECK (hold_scope IN ('subject', 'case', 'tenant', 'dataset')),
  subject_id uuid,
  case_id uuid,
  tenant_id uuid,
  dataset_code text,
  reason text NOT NULL,
  status text NOT NULL CHECK (status IN ('active', 'released')),
  created_by uuid NOT NULL,
  created_at timestamptz NOT NULL DEFAULT now(),
  released_by uuid,
  released_at timestamptz
);

SELECT cp.*
FROM subject_contact_point cp
WHERE cp.retain_until < now()
  AND NOT EXISTS (
    SELECT 1
    FROM legal_hold h
    WHERE h.status = 'active'
      AND (
        (h.hold_scope = 'subject' AND h.subject_id = cp.subject_id)
        OR (h.hold_scope = 'tenant' AND h.tenant_id = cp.tenant_id)
      )
  );

CREATE TABLE privacy_request (
  request_id uuid PRIMARY KEY,
  request_type text NOT NULL CHECK (request_type IN ('access', 'rectification', 'erasure', 'restriction', 'export')),
  subject_id uuid NOT NULL,
  status text NOT NULL CHECK (
    status IN ('requested', 'validating', 'approved', 'rejected', 'executing', 'completed', 'failed')
  ),
  reason text,
  requested_at timestamptz NOT NULL DEFAULT now(),
  completed_at timestamptz
);

CREATE TABLE privacy_request_action (
  action_id uuid PRIMARY KEY,
  request_id uuid NOT NULL REFERENCES privacy_request(request_id),
  target_system text NOT NULL,
  target_table text NOT NULL,
  target_key text NOT NULL,
  action text NOT NULL,
  status text NOT NULL CHECK (status IN ('pending', 'done', 'failed', 'skipped')),
  result_summary text,
  executed_at timestamptz
);

CREATE TABLE subject_identity_link (
  subject_id uuid NOT NULL,
  identifier_type text NOT NULL,
  identifier_hash bytea NOT NULL,
  source_system text NOT NULL,
  confidence text NOT NULL CHECK (confidence IN ('exact', 'strong', 'weak')),
  created_at timestamptz NOT NULL DEFAULT now(),
  PRIMARY KEY (subject_id, identifier_type, identifier_hash, source_system)
);

CREATE INDEX idx_subject_identity_link_identifier
ON subject_identity_link (identifier_type, identifier_hash);

normalized_email = lower(trim(email))
identifier_hash = HMAC-SHA256(secret_pepper, normalized_email)

-- Risky: raw national_id in index structure
CREATE INDEX idx_subject_national_id
ON subject_sensitive_identifier (national_id);

CREATE UNIQUE INDEX uq_subject_identifier_hash
ON subject_sensitive_identifier (identifier_type, identifier_hash);

CREATE TABLE subject_sensitive_identifier (
  identifier_id uuid PRIMARY KEY,
  subject_id uuid NOT NULL,
  identifier_type text NOT NULL,
  identifier_hash bytea NOT NULL,
  identifier_encrypted bytea,
  purpose_code text NOT NULL,
  retention_policy_code text NOT NULL,
  created_at timestamptz NOT NULL DEFAULT now()
);

CREATE VIEW case_subject_masked_view AS
SELECT
  s.subject_id,
  s.tenant_id,
  s.display_name,
  left(s.display_name, 1) || '***' AS masked_name,
  s.subject_type,
  s.created_at
FROM case_subject s;

CREATE FUNCTION mask_email(p_email text)
RETURNS text
LANGUAGE sql
IMMUTABLE
AS $$
  SELECT regexp_replace(p_email, '(^.).*(@.*$)', '\1***\2')
$$;

CREATE VIEW subject_contact_masked_view AS
SELECT
  contact_point_id,
  subject_id,
  contact_type,
  CASE
    WHEN contact_type = 'email' THEN mask_email(decrypt_contact(contact_value_encrypted))
    ELSE '***'
  END AS masked_contact_value,
  purpose_code,
  created_at
FROM subject_contact_point;

{
  "event": "subject.updated",
  "before": {
    "name": "Alice Example",
    "email": "alice@example.com",
    "national_id": "123456789"
  },
  "after": {
    "name": "Alice Example",
    "email": "alice.new@example.com",
    "national_id": "123456789"
  }
}

{
  "event": "subject.updated",
  "subject_id": "...",
  "changed_fields": ["email"],
  "actor_user_id": "...",
  "purpose": "case_contact_update",
  "pii_payload_stored": false
}

CREATE TABLE sensitive_audit_payload (
  audit_payload_id uuid PRIMARY KEY,
  audit_event_id uuid NOT NULL,
  payload_encrypted bytea NOT NULL,
  retention_policy_code text NOT NULL,
  retain_until timestamptz NOT NULL,
  access_reason_required boolean NOT NULL DEFAULT true
);

CREATE TABLE erasure_ledger (
  erasure_id uuid PRIMARY KEY,
  subject_id uuid NOT NULL,
  request_id uuid NOT NULL,
  erasure_scope text NOT NULL,
  executed_at timestamptz NOT NULL DEFAULT now(),
  proof_hash bytea NOT NULL,
  notes text
);

restore backup
apply migrations
replay erasure ledger after backup timestamp
validate erased subjects absent/masked
open system for traffic

CREATE TABLE privacy_action_outbox (
  outbox_id uuid PRIMARY KEY,
  subject_id uuid NOT NULL,
  action text NOT NULL CHECK (action IN ('erase', 'anonymize', 'restrict', 'unrestrict')),
  scope jsonb NOT NULL,
  created_at timestamptz NOT NULL DEFAULT now(),
  published_at timestamptz
);

CREATE TABLE privacy_action_inbox (
  consumer_name text NOT NULL,
  outbox_id uuid NOT NULL,
  processed_at timestamptz NOT NULL DEFAULT now(),
  PRIMARY KEY (consumer_name, outbox_id)
);

CREATE TABLE case_daily_metrics (
  metric_date date NOT NULL,
  tenant_id uuid NOT NULL,
  case_type text NOT NULL,
  region_code text NOT NULL,
  opened_count bigint NOT NULL,
  closed_count bigint NOT NULL,
  avg_resolution_hours numeric,
  PRIMARY KEY (metric_date, tenant_id, case_type, region_code)
);

CREATE TABLE retention_job_run (
  job_run_id uuid PRIMARY KEY,
  policy_code text NOT NULL,
  status text NOT NULL CHECK (status IN ('planned', 'running', 'completed', 'failed', 'cancelled')),
  dry_run boolean NOT NULL DEFAULT true,
  candidate_count bigint,
  affected_count bigint,
  started_at timestamptz,
  completed_at timestamptz,
  error_message text
);

CREATE TABLE retention_job_item (
  job_run_id uuid NOT NULL REFERENCES retention_job_run(job_run_id),
  target_table text NOT NULL,
  target_pk text NOT NULL,
  action text NOT NULL,
  status text NOT NULL CHECK (status IN ('candidate', 'done', 'skipped', 'failed')),
  reason text,
  PRIMARY KEY (job_run_id, target_table, target_pk)
);

WITH candidate AS (
  SELECT contact_point_id
  FROM subject_contact_point
  WHERE retain_until < now()
  ORDER BY retain_until
  LIMIT 1000
)
DELETE FROM subject_contact_point cp
USING candidate c
WHERE cp.contact_point_id = c.contact_point_id
RETURNING cp.contact_point_id;

CREATE TABLE compliance_control_evidence (
  evidence_id uuid PRIMARY KEY,
  control_code text NOT NULL,
  evidence_type text NOT NULL CHECK (evidence_type IN ('query_result', 'job_run', 'review', 'approval', 'test_result')),
  scope text NOT NULL,
  evidence_hash bytea,
  summary text NOT NULL,
  produced_by text NOT NULL,
  produced_at timestamptz NOT NULL DEFAULT now()
);

For subject X, show:
- what personal data we hold,
- why we hold it,
- where it exists,
- who can access it,
- when it expires,
- whether legal hold applies,
- how it will be erased/anonymized,
- and proof of past privacy actions.