Number Intelligence Service — Data Model

Version: 1.0 Status: Draft Owner: Messaging Core Last Updated: 2026-04-21 Companion: DOMAIN_MODEL · SECURITY_MODEL · SYNC_CONTRACT

Schema: numint. Owned exclusively by number-intelligence-service. No other service reads or writes these tables directly — integration is over gRPC (ResolveMsisdn, ResolveBatch, ProbeHlr, LookupPorting, LookupEir, GetMnpHistory), HTTP REST (admin / tenant / regulator), or NATS events.

1. Tables

-- =====================================================================
--                            ENUM TYPES
-- =====================================================================

CREATE SCHEMA IF NOT EXISTS numint;

CREATE TYPE numint.line_type   AS ENUM ('MOBILE','FIXED','VOIP','UNKNOWN');
CREATE TYPE numint.mnp_status  AS ENUM ('NATIVE','PORTED_IN','PORTED_OUT','UNKNOWN');
CREATE TYPE numint.confidence  AS ENUM ('HIGH','MEDIUM','LOW','UNKNOWN');
CREATE TYPE numint.attribution_source AS ENUM (
  'LRU','REDIS','POSTGRES',
  'LIVE_HLR_MAP','LIVE_HLR_REST',
  'MNP_RECON','PREFIX_FALLBACK','STALE_THROTTLED','ADMIN_OVERRIDE','MNO_HLR_DUMP'
);
CREATE TYPE numint.eir_state   AS ENUM ('WHITELIST','GREYLIST','BLACKLIST','UNKNOWN');
CREATE TYPE numint.port_direction AS ENUM ('IN','OUT');
CREATE TYPE numint.probe_transport AS ENUM ('MAP_SRI_SM','REST_ADAPTER');
CREATE TYPE numint.probe_status AS ENUM ('OK','TIMEOUT','MAP_ABORT','REST_5XX','THROTTLED','ADAPTER_DOWN');
CREATE TYPE numint.recon_kind  AS ENUM ('MNP','EIR');
CREATE TYPE numint.recon_status AS ENUM ('PENDING','RUNNING','COMPLETED','FAILED');
CREATE TYPE numint.conflict_severity AS ENUM ('HIGH','MEDIUM','LOW');
CREATE TYPE numint.conflict_resolution AS ENUM ('A_WINS','B_WINS','KEEP_BOTH_PENDING_VENDOR_CONFIRM','DISCARDED');
CREATE TYPE numint.lookup_result_class AS ENUM ('SUCCESS','INVALID_MSISDN','QUOTA_EXCEEDED','RATE_LIMITED','ERROR');
CREATE TYPE numint.lookup_tier AS ENUM ('LRU','REDIS','PG','LIVE','FALLBACK');
CREATE TYPE numint.risk_flag AS ENUM ('STOLEN_DEVICE','MNP_DIVERGENCE','ABNORMAL_MNP_CHURN','PREFIX_MISMATCH','UNUSUAL_VLR');

-- =====================================================================
--                           NUMBER RECORDS (authoritative)
-- =====================================================================

CREATE TABLE numint.number_records (
  msisdn_hash        BYTEA PRIMARY KEY,               -- sha256(e164 || pepper)
  e164               TEXT NOT NULL,                   -- canonical E.164
  mno_id             TEXT NOT NULL,                   -- slug
  original_mno_id    TEXT,
  line_type          numint.line_type NOT NULL DEFAULT 'UNKNOWN',
  country            CHAR(2) NOT NULL,                -- ISO 3166-1 alpha-2
  mnp_status         numint.mnp_status NOT NULL DEFAULT 'UNKNOWN',
  vlr                TEXT,
  imsi_prefix        TEXT,                            -- MCC+MNC (6 digits)
  source             numint.attribution_source NOT NULL,
  confidence         numint.confidence NOT NULL,
  risk_flags         numint.risk_flag[] NOT NULL DEFAULT '{}',
  last_seen          TIMESTAMPTZ NOT NULL DEFAULT now(),
  cached_at          TIMESTAMPTZ NOT NULL DEFAULT now(),
  lookup_count       BIGINT NOT NULL DEFAULT 0,
  version            INT NOT NULL DEFAULT 1,
  pepper_version     TEXT NOT NULL DEFAULT 'v1',
  CONSTRAINT chk_native_no_original_mno CHECK (
    (mnp_status = 'NATIVE' AND original_mno_id IS NULL)
    OR mnp_status <> 'NATIVE'
  ),
  CONSTRAINT chk_e164 CHECK (e164 ~ '^\+[1-9][0-9]{6,14}$')
);

CREATE INDEX ix_number_records_mno      ON numint.number_records (mno_id);
CREATE INDEX ix_number_records_country  ON numint.number_records (country);
CREATE INDEX ix_number_records_mnp      ON numint.number_records (mnp_status) WHERE mnp_status <> 'NATIVE';
CREATE INDEX ix_number_records_updated  ON numint.number_records USING brin (cached_at);  -- hot-scan of recently-updated rows
CREATE INDEX ix_number_records_popular  ON numint.number_records (lookup_count DESC) INCLUDE (msisdn_hash) WHERE lookup_count > 100;

-- =====================================================================
--                          PORTABILITY HISTORY (hash-chained, append-only)
-- =====================================================================

CREATE TABLE numint.portability_history (
  port_id            TEXT NOT NULL,                   -- 'ni_<ULID>'
  msisdn_hash        BYTEA NOT NULL,
  donor_mno_id       TEXT NOT NULL,
  recipient_mno_id   TEXT NOT NULL,
  port_date          DATE NOT NULL,
  direction          numint.port_direction NOT NULL DEFAULT 'IN',
  source_feed        TEXT NOT NULL,
  recon_run_id       TEXT NOT NULL,
  seq                BIGINT NOT NULL,                 -- monotonic per msisdn_hash
  prev_chain_hash    BYTEA NOT NULL,
  record_hash        BYTEA NOT NULL,
  signing_key_id     TEXT NOT NULL,
  observed_at        TIMESTAMPTZ NOT NULL DEFAULT now(),
  PRIMARY KEY (observed_at, msisdn_hash, port_date, recipient_mno_id)
) PARTITION BY RANGE (observed_at);

-- Partition example (monthly, provisioned by PartitionMaintainer)
-- CREATE TABLE numint.portability_history_2026_04 PARTITION OF numint.portability_history
--   FOR VALUES FROM ('2026-04-01') TO ('2026-05-01');

CREATE INDEX ix_port_hist_msisdn        ON numint.portability_history (msisdn_hash, port_date DESC);
CREATE INDEX ix_port_hist_mno           ON numint.portability_history (recipient_mno_id, observed_at);
CREATE INDEX ix_port_hist_recon         ON numint.portability_history (recon_run_id);
CREATE UNIQUE INDEX ux_port_hist_dedup
  ON numint.portability_history (msisdn_hash, port_date, recipient_mno_id, source_feed);

-- Append-only protection
CREATE RULE port_hist_no_update AS ON UPDATE TO numint.portability_history DO INSTEAD NOTHING;
CREATE RULE port_hist_no_delete AS ON DELETE TO numint.portability_history DO INSTEAD NOTHING;

-- =====================================================================
--                          RECONCILIATION RUNS
-- =====================================================================

CREATE TABLE numint.reconciliation_runs (
  run_id             TEXT PRIMARY KEY,                -- 'rcn_<ULID>'
  kind               numint.recon_kind NOT NULL,
  mno_id             TEXT,
  file_sha256        TEXT,
  total_records      INT NOT NULL DEFAULT 0,
  accepted           INT NOT NULL DEFAULT 0,
  rejected           INT NOT NULL DEFAULT 0,
  conflicts_count    INT NOT NULL DEFAULT 0,
  duration_ms        INT,
  status             numint.recon_status NOT NULL DEFAULT 'PENDING',
  prev_chain_hash    BYTEA,                           -- per-MNO chain
  record_hash        BYTEA,
  signing_key_id     TEXT,
  error_message      TEXT,
  started_at         TIMESTAMPTZ NOT NULL DEFAULT now(),
  completed_at       TIMESTAMPTZ
);

CREATE INDEX ix_recon_runs_mno_time ON numint.reconciliation_runs (mno_id, started_at DESC);
CREATE INDEX ix_recon_runs_status   ON numint.reconciliation_runs (status) WHERE status IN ('PENDING','RUNNING','FAILED');

-- =====================================================================
--                          RECONCILIATION CONFLICTS
-- =====================================================================

CREATE TABLE numint.reconciliation_conflicts (
  conflict_id        TEXT PRIMARY KEY,                -- 'cfl_<ULID>'
  msisdn_hash        BYTEA NOT NULL,
  candidate_a        JSONB NOT NULL,                  -- { mno_id, port_date, source_feed }
  candidate_b        JSONB NOT NULL,
  severity           numint.conflict_severity NOT NULL,
  resolution         numint.conflict_resolution,
  resolved_by        UUID,
  resolved_at        TIMESTAMPTZ,
  note               TEXT,
  created_at         TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE INDEX ix_conflicts_open ON numint.reconciliation_conflicts (created_at) WHERE resolution IS NULL;
CREATE INDEX ix_conflicts_msisdn ON numint.reconciliation_conflicts (msisdn_hash);

-- =====================================================================
--                          HLR PROBES (audit)
-- =====================================================================

CREATE TABLE numint.hlr_probes (
  probe_id           TEXT NOT NULL,                   -- 'prb_<ULID>'
  msisdn_hash        BYTEA NOT NULL,
  mno_hint           TEXT,
  transport          numint.probe_transport NOT NULL,
  gcc_invoke_id      INT,
  status             numint.probe_status NOT NULL,
  duration_ms        INT NOT NULL,
  result_snapshot    JSONB,
  started_at         TIMESTAMPTZ NOT NULL DEFAULT now(),
  ended_at           TIMESTAMPTZ,
  PRIMARY KEY (started_at, probe_id)
) PARTITION BY RANGE (started_at);

CREATE INDEX ix_hlr_probes_msisdn ON numint.hlr_probes (msisdn_hash, started_at DESC);
CREATE INDEX ix_hlr_probes_mno    ON numint.hlr_probes (mno_hint, started_at DESC);

-- =====================================================================
--                          LOOKUP AUDIT (hash-chained, partitioned monthly)
-- =====================================================================

CREATE TABLE numint.lookup_audit (
  audit_id           TEXT NOT NULL,                   -- 'nia_<ULID>'
  partition_name     TEXT NOT NULL,
  seq                BIGINT NOT NULL,
  tenant_id          UUID NOT NULL,
  actor_sub          TEXT,
  msisdn_hash        BYTEA NOT NULL,                  -- tenant-salted hash
  result_class       numint.lookup_result_class NOT NULL,
  result_mno         TEXT,
  staleness_seconds  INT,
  source             numint.attribution_source,
  tier               numint.lookup_tier,
  ip_address         INET,
  user_agent         TEXT,
  request_id         TEXT,
  prev_hash          BYTEA NOT NULL,
  record_hash        BYTEA NOT NULL,
  signing_key_id     TEXT NOT NULL,
  occurred_at        TIMESTAMPTZ NOT NULL DEFAULT now(),
  PRIMARY KEY (occurred_at, partition_name, seq)
) PARTITION BY RANGE (occurred_at);

CREATE INDEX ix_lookup_audit_tenant_time   ON numint.lookup_audit (tenant_id, occurred_at DESC);
CREATE INDEX ix_lookup_audit_msisdn_time   ON numint.lookup_audit (msisdn_hash, occurred_at DESC);
CREATE UNIQUE INDEX ux_lookup_audit_partition_seq
  ON numint.lookup_audit (partition_name, seq);

CREATE RULE lookup_audit_no_update AS ON UPDATE TO numint.lookup_audit DO INSTEAD NOTHING;
CREATE RULE lookup_audit_no_delete AS ON DELETE TO numint.lookup_audit DO INSTEAD NOTHING;

-- =====================================================================
--                          EIR RECORDS
-- =====================================================================

CREATE TABLE numint.eir_records (
  imei_hash          BYTEA PRIMARY KEY,
  imei               TEXT,                            -- retained for admin only; can be null if sourced as hash
  status             numint.eir_state NOT NULL,
  effective_status   numint.eir_state NOT NULL,
  reason_code        TEXT,
  reported_by        TEXT[] NOT NULL DEFAULT '{}',
  first_seen_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
  last_updated_at    TIMESTAMPTZ NOT NULL DEFAULT now(),
  CONSTRAINT chk_imei_luhn CHECK (imei IS NULL OR imei ~ '^[0-9]{15}$')
);

CREATE INDEX ix_eir_effective_status ON numint.eir_records (effective_status) WHERE effective_status <> 'WHITELIST';
CREATE INDEX ix_eir_last_updated     ON numint.eir_records USING brin (last_updated_at);

-- =====================================================================
--                          MSISDN ↔ IMEI OBSERVATION
-- =====================================================================

CREATE TABLE numint.msisdn_imei_observation (
  msisdn_hash        BYTEA NOT NULL,
  imei_hash          BYTEA NOT NULL,
  first_observed_at  TIMESTAMPTZ NOT NULL DEFAULT now(),
  last_observed_at   TIMESTAMPTZ NOT NULL DEFAULT now(),
  observation_count  INT NOT NULL DEFAULT 1,
  source             numint.attribution_source NOT NULL,
  PRIMARY KEY (msisdn_hash, imei_hash)
);

CREATE INDEX ix_mi_obs_msisdn ON numint.msisdn_imei_observation (msisdn_hash, last_observed_at DESC);

-- =====================================================================
--                          MNO SNAPSHOT (mirror from operator-management)
-- =====================================================================

CREATE TABLE numint.mno_snapshots (
  mno_id             TEXT PRIMARY KEY,
  name               TEXT NOT NULL,
  country            CHAR(2) NOT NULL,
  prefixes           TEXT[] NOT NULL DEFAULT '{}',
  hlr_endpoint       JSONB NOT NULL,                  -- { kind: 'MAP'|'REST', ... }
  mnp_sftp_endpoint  TEXT,
  tps_limit          INT NOT NULL DEFAULT 50,
  map_timeout_ms     INT NOT NULL DEFAULT 1500,
  rest_timeout_ms    INT NOT NULL DEFAULT 800,
  active             BOOLEAN NOT NULL DEFAULT TRUE,
  config_version     INT NOT NULL DEFAULT 1,
  updated_at         TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- =====================================================================
--                          TENANT LOOKUP QUOTAS
-- =====================================================================

CREATE TABLE numint.tenant_lookup_quotas (
  tenant_id              UUID PRIMARY KEY,
  rps_limit              INT NOT NULL DEFAULT 10,
  fresh_lookup_rps_limit INT NOT NULL DEFAULT 2,
  monthly_quota          INT NOT NULL DEFAULT 100000,
  current_month_used     INT NOT NULL DEFAULT 0,
  overage_mode           TEXT NOT NULL DEFAULT 'REJECT',
  plan_version           INT NOT NULL DEFAULT 1,
  updated_at             TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- =====================================================================
--                          AUDIT LOG (administrative)
-- =====================================================================

CREATE TABLE numint.audit_log (
  audit_id           TEXT PRIMARY KEY,                -- 'aud_<ULID>'
  entity_type        TEXT NOT NULL,                   -- MNP_CONFLICT, OVERRIDE, ADAPTER, QUOTA, ...
  entity_id          TEXT NOT NULL,
  action             TEXT NOT NULL,
  actor_user_id      UUID,
  before_state       JSONB,
  after_state        JSONB,
  ip                 INET,
  user_agent         TEXT,
  trace_id           TEXT,
  occurred_at        TIMESTAMPTZ NOT NULL DEFAULT now()
) PARTITION BY RANGE (occurred_at);

CREATE INDEX ix_audit_log_entity ON numint.audit_log (entity_type, entity_id, occurred_at DESC);

CREATE RULE audit_log_no_update AS ON UPDATE TO numint.audit_log DO INSTEAD NOTHING;
CREATE RULE audit_log_no_delete AS ON DELETE TO numint.audit_log DO INSTEAD NOTHING;

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

CREATE TABLE numint.outbox (
  event_id        UUID PRIMARY KEY,
  subject         TEXT NOT NULL,                      -- e.g., 'numint.mnp.changed.v1'
  payload         JSONB NOT NULL,
  published_at    TIMESTAMPTZ,
  attempts        INT NOT NULL DEFAULT 0,
  last_error      TEXT,
  created_at      TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE INDEX ix_outbox_unpublished ON numint.outbox (created_at) WHERE published_at IS NULL;

-- =====================================================================
--                          IDEMPOTENCY KEYS
-- =====================================================================

CREATE TABLE numint.idempotency_keys (
  tenant_id        UUID NOT NULL,
  idempotency_key  TEXT NOT NULL,
  request_hash     BYTEA NOT NULL,
  response         JSONB NOT NULL,
  created_at       TIMESTAMPTZ NOT NULL DEFAULT now(),
  expires_at       TIMESTAMPTZ NOT NULL DEFAULT (now() + INTERVAL '24 hours'),
  PRIMARY KEY (tenant_id, idempotency_key)
);
CREATE INDEX ix_numint_idempotency_expires ON numint.idempotency_keys (expires_at);

2. Redis keys

Namespace numint:. Redis cluster mode; {hash} slot-tag used where per-MSISDN atomicity matters.

Key	TTL	Purpose
`numint:lookup:{msisdnHash}`	per-class (30 d line_type / 24 h mno / 5 min vlr)	Hot cache for `ResolveMsisdn`
`numint:mnp:{msisdnHash}`	24 h	Fast `LookupPorting` response
`numint:eir:{imeiHash}`	24 h	`LookupEir` response
`numint:mno_snapshot:{mnoId}`	300 s	Operator config snapshot
`numint:quota:{tenantId}`	60 s	`TenantLookupQuota` snapshot
`numint:tps:hlr:{mno}`	3600 s	Per-MNO SS7 TPS token bucket
`numint:tps:lookup:{tenantId}`	3600 s	Per-tenant RPS bucket
`numint:quota:lookup:{tenantId}:{yyyymm}`	until 1st next month	Monthly counter
`numint:lock:mnp_recon:{mnoId}`	1800 s	Distributed cron lock
`numint:lock:eir_recon`	1800 s
`numint:lock:audit_verifier`	3600 s
`numint:lock:cache_warm`	600 s
`numint:warm:mnos`	21 600 s	List of MSISDNs to pre-warm

3. Append-only & access rules

3.1 Append-only tables

numint.portability_history — rules reject UPDATE / DELETE.
numint.lookup_audit — same.
numint.audit_log — same.

Retention is enforced by detaching and archiving partitions, never by DELETE.

3.2 RLS intentionally absent on attribution tables

Per DOMAIN_MODEL §6, number attribution is platform-public authoritative fact (like an E.164 prefix-to-MNO table would be). No Row-Level Security is enabled on number_records, portability_history, eir_records, or msisdn_imei_observation. Access control is enforced at the API layer (SPIFFE SAN allowlist on gRPC, Kong+JWT on REST).

3.3 RLS on `lookup_audit` (tenant scope)

ALTER TABLE numint.lookup_audit ENABLE ROW LEVEL SECURITY;

CREATE POLICY lookup_audit_tenant_read ON numint.lookup_audit
  FOR SELECT
  USING (
    tenant_id = current_setting('app.current_tenant_id', true)::uuid
    OR current_setting('app.caller_role', true) IN ('platform.numint.admin','platform.regulator')
  );

Tenants read their own audit; admins/regulators read all.

4. ID prefixes

Prefix	Entity
`ni_`	PortabilityRecord (port event)
`rcn_`	ReconciliationRun
`cfl_`	ReconciliationConflict
`prb_`	HlrProbe
`nia_`	LookupAuditEntry
`aud_`	AdminAuditLog
`cvr_`	ChainVerifierRun

ULIDs for all *_id values; prefixes only on externally-exposed identifiers.

5. PII, encryption & retention

Field	Class	Protection
`number_records.e164`	INTERNAL-PLATFORM (carrier attribution is public telecom fact, not subscriber PII per ATRA guidance; retained for joinability)	No RLS; TLS in transit; encryption-at-rest via PG transparent disk encryption per platform 13-sct §3
`number_records.msisdn_hash`	INTERNAL	sha256 with platform pepper (Vault KV `secret/ghasi/numint/msisdn_pepper`); rotated quarterly with envelope re-keying; `pepper_version` column tracks
`portability_history.msisdn_hash`	INTERNAL	Same
`lookup_audit.msisdn_hash`	INTERNAL	sha256 with per-tenant salt (`secret/ghasi/numint/tenant-salts/{tenantId}`); cross-tenant re-derivation requires salt
`eir_records.imei`	RESTRICTED	Optional plain retention for admin view; hash is the key
`eir_records.imei_hash`	INTERNAL	sha256 with platform pepper
`hlr_probes.result_snapshot`	INTERNAL	JSONB; MSISDN only as hash; VLR retained
`mno_snapshots.hlr_endpoint` auth material	RESTRICTED	Credentials stored in Vault KV; snapshot holds only endpoint URL + auth-profile reference

5.1 Retention

Table	Hot (Postgres)	Cold (S3)	Rule
`number_records`	Indefinite	—	Active records; decommissioned MSISDNs hard-delete after 24 months of `last_seen` inactivity
`portability_history`	24 months	7 y Object Lock	Regulator audit
`reconciliation_runs`	24 months	7 y Object Lock
`reconciliation_conflicts`	Indefinite (while unresolved) + 12 months after resolve	—
`hlr_probes`	90 days	—	SS7 PCAP correlation window
`lookup_audit`	13 months	7 y Object Lock	Regulator + billing dispute
`audit_log`	13 months	7 y Object Lock
`eir_records`	Indefinite	—	Authoritative mirror
`msisdn_imei_observation`	180 days	—	Observational; not audit evidence
`outbox`	7 days post-publish	—
`idempotency_keys`	24 h	—

5.2 Cold archive flow

Per partitioned table:

Daily cron numint-partition-archiver identifies partitions older than hot window.
pg_dump --schema-only --table=… + COPY … TO PROGRAM 'zstd -19 > /tmp/….zst' to S3 with SSE-KMS.
Object Lock governance mode retention per rule above (7 y where applicable).
Verification: re-read first/last/middle rows; for hash-chained partitions, recompute chain across boundary.
Drop partition from Postgres.

6. Migration strategy

Partitions provisioned 3 months ahead by PartitionMaintainer. Missing future partition fires NumIntPartitionMissing (HIGH).
Schema changes use expand/contract; forward-only migrations; Security review for any column holding hashable material.
ENUM evolution: ALTER TYPE numint.<enum> ADD VALUE 'NEW' BEFORE 'UNKNOWN' only; never rename or remove within v1.
Index changes: CREATE INDEX CONCURRENTLY; pg_repack for rewrites > 100 GB.
Redis schema versioning: key prefix change on shape change (numint:lookup:v2:…); old keys expire naturally.
Pepper rotation is envelope-style: each row carries pepper_version; a background job re-derives msisdn_hash with the new pepper and double-writes until cutover, then drops the old-version rows. Tenant salts rotate independently; cross-tenant correlation attacks via salt replay are thereby bounded in time.

1. Tables​

2. Redis keys​

3. Append-only & access rules​

3.1 Append-only tables​

3.2 RLS intentionally absent on attribution tables​

3.3 RLS on lookup_audit (tenant scope)​

4. ID prefixes​

5. PII, encryption & retention​

5.1 Retention​

5.2 Cold archive flow​

6. Migration strategy​