CDR Mediation Service — Data Model

Version: 1.0 Status: Draft Owner: Commerce + Regulator Liaison Last Updated: 2026-04-21 Companion: DOMAIN_MODEL · SECURITY_MODEL · SYNC_CONTRACT

Schema: cdr. Owned exclusively by cdr-mediation-service. No other service reads or writes these tables directly — integration is via REST (/v1/cdr/*) or NATS events.

Two storage tiers:

• Hot (PostgreSQL) — last 13 months, partitioned monthly, per-operator indexed. Target size: ~2–4 TB per region at steady state (estimated from ADR-0004 §9 NFRs: 500 msg/s peak × 86,400 s × 365 d).
• Cold (S3-compatible object store) — 7-year retention, Object-Lock Compliance mode, zstd JSONL with _MANIFEST.json per hour.

---

1. Postgres tables

-- =====================================================================
--                            ENUMS
-- =====================================================================

CREATE SCHEMA IF NOT EXISTS cdr;

CREATE TYPE cdr.direction         AS ENUM ('MO','MT');
CREATE TYPE cdr.charge_type       AS ENUM ('MO','MT','BROADCAST','INTERNATIONAL_MT','P2P','A2P');
CREATE TYPE cdr.billing_indicator AS ENUM ('CHARGEABLE','FREE','REVERSE_CHARGED','CPP','MPP','UNKNOWN');
CREATE TYPE cdr.encoding          AS ENUM ('GSM7','UCS2','GSM8_LATIN');
CREATE TYPE cdr.state             AS ENUM ('RAW','AGGREGATED','EXPORTED','ACKED','ADJUSTED','ARCHIVED');
CREATE TYPE cdr.adjustment_type   AS ENUM ('CORRECTION','VOID','RE_RATE');
CREATE TYPE cdr.export_type       AS ENUM ('TAP_3_12','RAP_1_5','CSV_ANCILLARY');
CREATE TYPE cdr.delivery_state    AS ENUM (
  'PENDING','UPLOADING','DELIVERED_SFTP','DELIVERED_API',
  'FAILED','REJECTED','ACKED','INTEGRITY_MISMATCH'
);
CREATE TYPE cdr.verification_status AS ENUM (
  'VERIFIED','INTEGRITY_MISMATCH','SCHEMA_REJECTED','UNKNOWN'
);
CREATE TYPE cdr.audit_entry_type AS ENUM (
  'CHAIN_VERIFY_OK','CHAIN_BREAK_DETECTED',
  'TRANSPARENCY_ANCHORED','SCHEMA_CHANGED',
  'KEY_ROTATED','EXPORT_REPLAYED','ADJUSTMENT_ISSUED'
);
CREATE TYPE cdr.operator_id AS ENUM (
  'AWCC','MTN_AF','ETS','ROSHAN','SALAAM','INTL_IN','INTL_OUT'
);

-- =====================================================================
--                   CDR RECORDS (partitioned monthly on bucket_hour)
-- =====================================================================

CREATE TABLE cdr.records (
  cdr_id                 UUID NOT NULL,
  cdr_sequence           BIGINT NOT NULL,
  bucket_hour            TIMESTAMPTZ NOT NULL,
  operator_id            cdr.operator_id NOT NULL,
  tenant_id              UUID,
  message_id             UUID NOT NULL,
  source_event_id        UUID NOT NULL,         -- sms.dlr.inbound event id (idempotency)
  msisdn_hash_to         BYTEA NOT NULL,
  msisdn_hash_from       BYTEA,
  sender_id_raw          TEXT NOT NULL,
  recording_entity       TEXT NOT NULL,
  service_center_address TEXT NOT NULL,
  message_reference      TEXT NOT NULL,
  event_timestamp        TIMESTAMPTZ NOT NULL,
  local_timestamp        TIMESTAMPTZ NOT NULL,
  charge_amount          NUMERIC(18,6),
  charge_currency        CHAR(3) NOT NULL DEFAULT 'AFN',
  charge_type            cdr.charge_type NOT NULL,
  billing_indicator      cdr.billing_indicator NOT NULL DEFAULT 'UNKNOWN',
  tap_tariff_class       CHAR(4),
  segment_count          SMALLINT NOT NULL,
  encoding               cdr.encoding NOT NULL,
  direction              cdr.direction NOT NULL,
  state                  cdr.state NOT NULL DEFAULT 'RAW',
  chain_hash_prev        BYTEA NOT NULL,
  row_hash               BYTEA NOT NULL,
  adjustment_of          UUID,
  adjustment_type        cdr.adjustment_type,
  void_reason            TEXT,
  ticket_id              TEXT,
  corrected_fields       JSONB,
  created_at             TIMESTAMPTZ NOT NULL DEFAULT now(),
  PRIMARY KEY (bucket_hour, cdr_id)
) PARTITION BY RANGE (bucket_hour);

-- Uniqueness to guarantee idempotent projection
CREATE UNIQUE INDEX ux_records_source_event
  ON cdr.records (source_event_id) WHERE adjustment_of IS NULL;

CREATE INDEX ix_records_operator_hour
  ON cdr.records (operator_id, bucket_hour, cdr_sequence);

CREATE INDEX ix_records_message_id
  ON cdr.records (message_id, bucket_hour);

CREATE INDEX ix_records_tenant_hour
  ON cdr.records (tenant_id, bucket_hour DESC) WHERE tenant_id IS NOT NULL;

CREATE INDEX ix_records_adjustment_of
  ON cdr.records (adjustment_of) WHERE adjustment_of IS NOT NULL;

-- BRIN index for bulk TAP encoder range scans
CREATE INDEX ix_records_hour_brin ON cdr.records USING BRIN (bucket_hour);

-- =====================================================================
--                   HOURLY ROLLUPS (chain heads)
-- =====================================================================

CREATE TABLE cdr.rollups (
  rollup_id                UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  bucket_hour              TIMESTAMPTZ NOT NULL,
  operator_id              cdr.operator_id NOT NULL,
  record_count             BIGINT NOT NULL,
  mo_count                 BIGINT NOT NULL DEFAULT 0,
  mt_count                 BIGINT NOT NULL DEFAULT 0,
  broadcast_count          BIGINT NOT NULL DEFAULT 0,
  international_mt_count   BIGINT NOT NULL DEFAULT 0,
  chargeable_sum           NUMERIC(22,6) NOT NULL DEFAULT 0,
  free_sum                 NUMERIC(22,6) NOT NULL DEFAULT 0,
  bucket_root              BYTEA NOT NULL,
  chain_hash               BYTEA NOT NULL,
  prev_bucket_chain_hash   BYTEA NOT NULL,
  sealed_at                TIMESTAMPTZ NOT NULL DEFAULT now(),
  signer_key_id            TEXT NOT NULL,
  empty_bucket             BOOLEAN NOT NULL DEFAULT FALSE,
  object_store_root        TEXT,                       -- S3 manifest URI once mirrored
  mirrored_at              TIMESTAMPTZ,
  UNIQUE (bucket_hour, operator_id)
);

CREATE INDEX ix_rollups_operator_hour
  ON cdr.rollups (operator_id, bucket_hour DESC);

CREATE INDEX ix_rollups_chain_hash
  ON cdr.rollups (chain_hash);

-- =====================================================================
--                   EXPORTS + DELIVERY LOG
-- =====================================================================

CREATE TABLE cdr.exports (
  export_id                UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  export_type              cdr.export_type NOT NULL,
  schema_variant           TEXT NOT NULL,
  recording_entity         TEXT NOT NULL,
  sender_roaming_partner   TEXT,
  settlement_day           DATE NOT NULL,
  file_sequence_number     BIGINT NOT NULL,
  file_name                TEXT NOT NULL,
  file_sha256              BYTEA NOT NULL,
  byte_size                BIGINT NOT NULL,
  records_included         BIGINT NOT NULL,
  quarantined_count        BIGINT NOT NULL DEFAULT 0,
  signer_key_id            TEXT NOT NULL,
  signature_base64         TEXT NOT NULL,
  object_store_uri         TEXT NOT NULL,
  delivery_state           cdr.delivery_state NOT NULL DEFAULT 'PENDING',
  atra_receipt_id          TEXT,
  atra_verification_status cdr.verification_status,
  delivered_at             TIMESTAMPTZ,
  acked_at                 TIMESTAMPTZ,
  retry_count              INT NOT NULL DEFAULT 0,
  last_error_reason        TEXT,
  created_at               TIMESTAMPTZ NOT NULL DEFAULT now(),
  UNIQUE (recording_entity, export_type, file_sequence_number)
);
CREATE INDEX ix_exports_state ON cdr.exports (delivery_state, settlement_day DESC);
CREATE INDEX ix_exports_day   ON cdr.exports (settlement_day DESC);

CREATE TABLE cdr.export_delivery_log (
  log_id           UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  export_id        UUID NOT NULL REFERENCES cdr.exports(export_id),
  channel          TEXT NOT NULL,            -- 'SFTP' | 'API'
  attempt_number   INT NOT NULL,
  started_at       TIMESTAMPTZ NOT NULL,
  finished_at      TIMESTAMPTZ,
  outcome          TEXT NOT NULL,            -- 'SUCCESS' | 'FAIL' | 'TIMEOUT'
  latency_ms       INT,
  remote_host      TEXT,
  error_code       TEXT,
  error_message    TEXT,
  atra_receipt_id  TEXT
);
CREATE INDEX ix_export_delivery_log_export
  ON cdr.export_delivery_log (export_id, started_at DESC);

-- Atomic file sequence counter per (recording_entity, export_type)
CREATE TABLE cdr.tap_sequence (
  recording_entity TEXT NOT NULL,
  export_type      cdr.export_type NOT NULL,
  last_sequence    BIGINT NOT NULL DEFAULT 0,
  updated_at       TIMESTAMPTZ NOT NULL DEFAULT now(),
  PRIMARY KEY (recording_entity, export_type)
);

-- =====================================================================
--                   ADJUSTMENTS (metadata beyond cdr.records row)
-- =====================================================================

CREATE TABLE cdr.adjustments (
  adjustment_id          UUID PRIMARY KEY,      -- matches cdr_id of adjustment row in cdr.records
  original_cdr_id        UUID NOT NULL,
  original_bucket_hour   TIMESTAMPTZ NOT NULL,
  adjustment_type        cdr.adjustment_type NOT NULL,
  reason                 TEXT NOT NULL,
  ticket_id              TEXT NOT NULL,
  issued_by_user_id      UUID NOT NULL,
  approver_user_id       UUID,
  corrected_fields       JSONB,
  void_reason            TEXT,
  new_pricing_table_id   UUID,
  rap_batched_at         TIMESTAMPTZ,
  rap_file_id            UUID REFERENCES cdr.exports(export_id),
  job_id                 UUID,
  issued_at              TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE INDEX ix_adjustments_original
  ON cdr.adjustments (original_cdr_id);
CREATE INDEX ix_adjustments_rap
  ON cdr.adjustments (rap_batched_at) WHERE rap_batched_at IS NULL;
CREATE INDEX ix_adjustments_job
  ON cdr.adjustments (job_id) WHERE job_id IS NOT NULL;

CREATE TABLE cdr.adjustment_jobs (
  job_id            UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  adjustment_type   cdr.adjustment_type NOT NULL,
  filter            JSONB NOT NULL,
  new_pricing_table_id UUID,
  estimated_impact  BIGINT NOT NULL,
  processed         BIGINT NOT NULL DEFAULT 0,
  errors            JSONB NOT NULL DEFAULT '[]',
  status            TEXT NOT NULL DEFAULT 'PENDING',   -- PENDING|RUNNING|COMPLETED|FAILED
  requested_by      UUID NOT NULL,
  approver_id       UUID,
  ticket_id         TEXT NOT NULL,
  started_at        TIMESTAMPTZ,
  finished_at       TIMESTAMPTZ,
  created_at        TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- =====================================================================
--                   AUDIT (append-only, partitioned monthly)
-- =====================================================================

CREATE TABLE cdr.audit (
  audit_id       UUID NOT NULL DEFAULT gen_random_uuid(),
  entry_type     cdr.audit_entry_type NOT NULL,
  bucket_hour    TIMESTAMPTZ,
  operator_id    cdr.operator_id,
  actor_user_id  UUID,
  details        JSONB NOT NULL DEFAULT '{}',
  trace_id       TEXT,
  occurred_at    TIMESTAMPTZ NOT NULL DEFAULT now(),
  PRIMARY KEY (occurred_at, audit_id)
) PARTITION BY RANGE (occurred_at);

CREATE INDEX ix_audit_type_time
  ON cdr.audit (entry_type, occurred_at DESC);
CREATE INDEX ix_audit_bucket
  ON cdr.audit (bucket_hour, operator_id) WHERE bucket_hour IS NOT NULL;

-- =====================================================================
--                   REGULATOR SCHEMA ADAPTERS
-- =====================================================================

CREATE TABLE cdr.regulator_schemas (
  variant_id       TEXT PRIMARY KEY,             -- e.g. 'atra-tap-v2'
  export_type      cdr.export_type NOT NULL,
  asn1_module_path TEXT NOT NULL,                -- filesystem path to ASN.1 module
  adapter_class    TEXT NOT NULL,                -- e.g. 'AtraTapV2Adapter'
  active           BOOLEAN NOT NULL DEFAULT FALSE,
  activated_at     TIMESTAMPTZ,
  activated_by     UUID,
  description      TEXT,
  created_at       TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE UNIQUE INDEX ux_regulator_schemas_active_by_type
  ON cdr.regulator_schemas (export_type) WHERE active = TRUE;

-- =====================================================================
--                   QUARANTINE (per-record encoder failure)
-- =====================================================================

CREATE TABLE cdr.quarantine (
  quarantine_id     UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  cdr_id            UUID NOT NULL,
  cdr_bucket_hour   TIMESTAMPTZ NOT NULL,
  encoder_error     TEXT NOT NULL,
  encoder_variant   TEXT NOT NULL,
  encoder_version   TEXT NOT NULL,
  row_snapshot      JSONB NOT NULL,
  quarantined_at    TIMESTAMPTZ NOT NULL DEFAULT now(),
  resolution        TEXT,                       -- 'RESUBMITTED' | 'DISCARDED'
  resolved_by       UUID,
  resolved_at       TIMESTAMPTZ
);
CREATE INDEX ix_quarantine_unresolved
  ON cdr.quarantine (quarantined_at) WHERE resolution IS NULL;

-- =====================================================================
--                   TRANSPARENCY ANCHORS
-- =====================================================================

CREATE TABLE cdr.transparency_anchors (
  anchor_id       UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  day             DATE NOT NULL,
  operator_id     cdr.operator_id NOT NULL,
  chain_hash      BYTEA NOT NULL,
  leaf_index      BIGINT NOT NULL,
  log_root_hash   BYTEA NOT NULL,
  sle_signature   BYTEA NOT NULL,       -- signed log entry from Trillian
  inclusion_proof JSONB,
  submitted_at    TIMESTAMPTZ NOT NULL DEFAULT now(),
  UNIQUE (day, operator_id)
);

-- =====================================================================
--                   MSISDN VAULT (per-operator KEK-encrypted MSISDNs)
-- =====================================================================

CREATE TABLE cdr.msisdn_vault (
  cdr_id          UUID PRIMARY KEY,
  bucket_hour     TIMESTAMPTZ NOT NULL,
  operator_id     cdr.operator_id NOT NULL,
  msisdn_to_ct    BYTEA NOT NULL,        -- AES-256-GCM ciphertext
  msisdn_to_iv    BYTEA NOT NULL,
  msisdn_to_tag   BYTEA NOT NULL,
  msisdn_from_ct  BYTEA,
  msisdn_from_iv  BYTEA,
  msisdn_from_tag BYTEA,
  kek_id          TEXT NOT NULL,         -- Vault Transit key ref (per-operator)
  created_at      TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE INDEX ix_msisdn_vault_hour
  ON cdr.msisdn_vault (bucket_hour);

-- =====================================================================
--                   ARCHIVE LOG (cold-tier migrations)
-- =====================================================================

CREATE TABLE cdr.archive_log (
  archive_id     UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  partition_name TEXT NOT NULL,
  row_count      BIGINT NOT NULL,
  byte_size      BIGINT NOT NULL,
  s3_manifest    TEXT NOT NULL,
  dropped_at     TIMESTAMPTZ NOT NULL DEFAULT now(),
  dropped_by     TEXT NOT NULL DEFAULT 'SYSTEM'
);

-- =====================================================================
--                   OUTBOX
-- =====================================================================

CREATE TABLE cdr.outbox (
  event_id      UUID PRIMARY KEY,
  subject       TEXT NOT NULL,
  partition_key TEXT,
  payload       JSONB NOT NULL,
  created_at    TIMESTAMPTZ NOT NULL DEFAULT now(),
  published_at  TIMESTAMPTZ
);
CREATE INDEX ix_cdr_outbox_unpublished
  ON cdr.outbox (created_at) WHERE published_at IS NULL;

Monthly partition provisioning (example — PartitionMaintenanceWorker pre-creates 3 months ahead):

CREATE TABLE cdr.records_2026_04
  PARTITION OF cdr.records
  FOR VALUES FROM ('2026-04-01') TO ('2026-05-01');

CREATE TABLE cdr.audit_2026_04
  PARTITION OF cdr.audit
  FOR VALUES FROM ('2026-04-01') TO ('2026-05-01');

---

2. S3 cold-tier layout

s3://ghasi-cdr-archive/
    v1/
        region={kabul|mazar|dxb}/
            operator={AWCC|MTN_AF|...}/
                year=YYYY/
                    month=MM/
                        day=DD/
                            hour=HH/
                                0001.cdr.jsonl.zst           (1000 rows)
                                0002.cdr.jsonl.zst
                                ...
                                _MANIFEST.json

Per-file attributes:

• Content-Type: application/x-ndjson+zstd
• x-amz-server-side-encryption: aws:kms (KMS key ghasi-cdr-archive-kek)
• x-amz-object-lock-mode: COMPLIANCE
• x-amz-object-lock-retain-until-date: <sealedAt + 7 years>
• x-amz-storage-class: STANDARD_IA (transitions to GLACIER_IR after 12 months via bucket lifecycle rule)

_MANIFEST.json schema:

{
  "bucketHour": "2026-04-20T13:00:00Z",
  "operatorId": "AWCC",
  "recordCount": 8421,
  "bucketRoot": "9f3c...",
  "chainHash":  "73a8...",
  "prevChainHash": "1200...",
  "files": [
    { "name": "0001.cdr.jsonl.zst", "sha256": "...", "recordCount": 1000 }
  ],
  "signerKeyId": "cdr-export-signer-v1",
  "schemaVersion": "1",
  "writtenAt": "2026-04-20T13:05:49Z"
}

Separate bucket ghasi-cdr-exports holds signed TAP/RAP files with Object-Lock Compliance for 7 years (required for regulator evidence).

---

3. Row-Level Security & append-only protection

3.1 Append-only

-- cdr.records: rejects UPDATE on any column EXCEPT state transitions
CREATE OR REPLACE FUNCTION cdr.records_guard_update()
  RETURNS trigger LANGUAGE plpgsql AS $$
BEGIN
  IF NEW.* IS DISTINCT FROM OLD.* AND NOT (
        OLD.state IS DISTINCT FROM NEW.state
    AND NEW.cdr_id = OLD.cdr_id
    AND NEW.row_hash = OLD.row_hash
    AND NEW.chain_hash_prev = OLD.chain_hash_prev
  ) THEN
    RAISE EXCEPTION 'cdr.records is append-only except state transitions';
  END IF;
  RETURN NEW;
END $$;

CREATE TRIGGER trg_records_guard_update
  BEFORE UPDATE ON cdr.records
  FOR EACH ROW EXECUTE FUNCTION cdr.records_guard_update();

CREATE RULE records_no_delete AS
  ON DELETE TO cdr.records DO INSTEAD NOTHING;

-- cdr.audit, cdr.adjustments, cdr.export_delivery_log: reject UPDATE + DELETE
CREATE RULE audit_no_update ON UPDATE TO cdr.audit DO INSTEAD NOTHING;
CREATE RULE audit_no_delete ON DELETE TO cdr.audit DO INSTEAD NOTHING;
CREATE RULE adjustments_no_delete ON DELETE TO cdr.adjustments DO INSTEAD NOTHING;
CREATE RULE delivery_log_no_update ON UPDATE TO cdr.export_delivery_log DO INSTEAD NOTHING;
CREATE RULE delivery_log_no_delete ON DELETE TO cdr.export_delivery_log DO INSTEAD NOTHING;

Retention is managed by dropping old partitions via PartitionMaintenanceWorker, not via DELETE.

3.2 Row-Level Security

Tenant-scoped read via RLS (for future tenant self-service and analytics-service multi-tenant views); admin sees all.

ALTER TABLE cdr.records ENABLE ROW LEVEL SECURITY;

CREATE POLICY records_tenant_read ON cdr.records
  FOR SELECT
  USING (
    tenant_id = current_setting('app.current_tenant_id', true)::uuid
    OR current_setting('app.caller_role', true) IN
         ('cdr-admin','regulator-auditor','platform.auditor')
  );

CREATE POLICY records_admin_all ON cdr.records
  USING (current_setting('app.caller_role', true) IN
         ('cdr-admin','regulator-auditor'));

The REST layer sets SET LOCAL app.current_tenant_id = '<uuid>' per request from the Kong-injected X-Tenant-Id header and SET LOCAL app.caller_role = '<role>' from X-Roles.

---

4. Redis keys

All keys namespaced cdr:*.

Key	TTL	Purpose
cdr:operator:reg-entity:{operatorId}	3600 s	Recording-entity cache from operator-management-service
cdr:billing:snapshot:{messageId}	300 s	Pricing snapshot cache to amortise GetPricingSnapshot
cdr:chain:last:{operatorId}:{bucketHour}	120 s	Last rowHash per partition — accelerates UC-01 step 8
cdr:seal:lock	600 s	Distributed cron lock for UC-03
cdr:tap:lock	3600 s	Cron lock for UC-07
cdr:rap:lock	3600 s	Cron lock for UC-08
cdr:verify:lock	7200 s	Cron lock for UC-12
cdr:archive:lock	10800 s	Cron lock for UC-15
cdr:schema:variant:{exportType}	60 s	Active schema variant (hot-reload marker)
cdr:export:seq:{recordingEntity}:{exportType}	no TTL	Cache of last sequence (write-through against cdr.tap_sequence)
cdr:outbox:relay:lock	60 s	Outbox publisher single-leader lock

Redis cluster DB index 5 reserved for cdr-mediation (per platform convention; compliance uses 3, firewall uses 4).

---

5. ID prefixes (external)

Prefix	Entity
cdr_	CdrRecord
roll_	CdrRollup
exp_	CdrExport
adj_	CdrAdjustment
adjjob_	Bulk adjustment job
qur_	Quarantine entry
anchr_	Transparency anchor
aud_	CdrAuditEntry

Internally all entities use UUIDv7 (time-sortable) where order matters, UUIDv4 otherwise. Prefixes apply only to externally-exposed identifiers on REST responses.

---

6. PII, encryption & retention

Field	Class	Protection
cdr.records.msisdn_hash_*	INTERNAL	One-way SHA-256 — no PII recoverable
cdr.msisdn_vault.msisdn_*_ct	RESTRICTED	AES-256-GCM; per-operator KEK in Vault Transit (transit/ghasi-cdr-msisdn-{operatorId}); decrypted only inside TAP encoder or by regulator-auditor role with four-eyes approval
cdr.records.sender_id_raw	CONFIDENTIAL	Stored in plain; masked in tenant-portal responses
cdr.records.charge_amount	CONFIDENTIAL	Plain; regulator exports it in cleartext; tenant portal sees only their own via RLS
cdr.exports.signature_base64	INTERNAL	Signature is not secret; private key never stored
cdr.audit.details	CONFIDENTIAL	May include row hashes, operator codes; no PII

Retention:

• cdr.records, cdr.adjustments, cdr.msisdn_vault: 13 months hot + 7 years cold (S3 Object-Lock Compliance). Partition drop after 13 months.
• cdr.rollups: 13 months hot (kept alongside records for chain walks) + permanent in manifest.
• cdr.exports, cdr.export_delivery_log: permanent in hot (small footprint; regulator reference).
• cdr.audit: 13 months hot + permanent in cold. Partition drop after 13 months, never DELETE.
• cdr.transparency_anchors: permanent (small footprint; external cryptographic reference).
• cdr.quarantine: 180 days post-resolution; then cold archive for 2 years.
• cdr.outbox: rows with published_at IS NOT NULL purged after 7 days; unpublished > 7 days → SEV1.

---

7. Migration strategy

• Partitions are provisioned 3 months ahead by PartitionMaintenanceWorker. Missing future partitions trigger CdrPartitionMissing HIGH alert.
• Retention is enforced by dropping partitions older than the retention window, never by DELETE.
• Schema changes follow expand/contract; all migrations forward-only. Every new column that holds any MSISDN-derived data requires Security review.
• Object-store bucket policy changes (e.g. lifecycle, retention) require cdr:admin + Security dual-approval; change logged to cdr.audit with entryType=SCHEMA_CHANGED.

---

8. Cross-References

• DOMAIN_MODEL.md — aggregates + invariants
• APPLICATION_LOGIC.md — use cases touch each table
• SECURITY_MODEL.md §3 — encryption keys and masking
• sms-firewall-service/DATA_MODEL.md — sibling hash-chained append-only model
• compliance-engine/DATA_MODEL.md — outbox precedent

End of DATA_MODEL.md