Federal Regulatory Data Hub schema design: per-vertical table layouts, entity_master bridge, and D1 indexing strategy for 35M records across 8 shards

-- financial-crimes: OFAC SDN (primary enforcement list)
CREATE TABLE ofac_sdn (
    id              INTEGER PRIMARY KEY,
    sdn_uid         INTEGER NOT NULL UNIQUE,
    sdn_name        TEXT    NOT NULL,
    sdn_type        TEXT    NOT NULL, -- 'Individual'|'Entity'|'Vessel'|'Aircraft'
    program_codes   TEXT    NOT NULL, -- pipe-separated, e.g. 'IRAN|SDGT'
    aliases         TEXT,             -- JSON array of alternate names
    addresses       TEXT,             -- JSON array of {address, city, country}
    nationalities   TEXT,             -- JSON array of ISO country codes
    id_documents    TEXT,             -- JSON array of {type, number, country}
    canonical_entity_id TEXT,         -- FK into entity_master (denormalized)
    last_updated    TEXT    NOT NULL
);

CREATE INDEX idx_ofac_program ON ofac_sdn (program_codes);
CREATE INDEX idx_ofac_canonical ON ofac_sdn (canonical_entity_id);

CREATE VIRTUAL TABLE ofac_sdn_fts USING fts5(
    sdn_name, aliases,
    content='ofac_sdn', content_rowid='id',
    tokenize='unicode61 remove_diacritics 2'
);

-- securities: EPA enforcement (covering index for entity queries)
CREATE TABLE epa_enforcement (
    id              INTEGER PRIMARY KEY,
    case_number     TEXT    NOT NULL UNIQUE,
    case_name       TEXT    NOT NULL,
    respondent_name TEXT    NOT NULL,
    statute         TEXT    NOT NULL,  -- 'CAA'|'CWA'|'RCRA'|'CERCLA'|'TSCA'|...
    activity_type   TEXT    NOT NULL,  -- 'Administrative'|'Civil Judicial'|'Criminal'
    initiated_date  TEXT,              -- ISO 8601 date
    concluded_date  TEXT,
    penalty_amount  REAL,              -- USD; NULL if no monetary penalty
    canonical_entity_id TEXT
);

-- Covering index: entity lookup returning summary columns without table read
CREATE INDEX idx_epa_entity_date ON epa_enforcement (
    canonical_entity_id,
    initiated_date DESC,
    case_number,
    penalty_amount
);

CREATE TABLE entity_master (
    canonical_id    TEXT    PRIMARY KEY,
    -- Format: '{entity_type}:{sha256_8hex}' e.g. 'CORP:a3f1c82d'
    entity_type     TEXT    NOT NULL, -- 'CORP'|'PERSON'|'VESSEL'|'AIRCRAFT'|'GOVT'
    canonical_name  TEXT    NOT NULL,
    ein             TEXT,
    cik             TEXT,
    lei             TEXT,
    uei             TEXT,   -- SAM.gov Unique Entity Identifier
    npi             TEXT,
    -- Bitmask of shards with records (bits 0-7 = 8 verticals above)
    shard_presence  INTEGER NOT NULL DEFAULT 0,
    last_updated    TEXT    NOT NULL
);

-- Covering index: canonical_id lookup returns name+bitmask without table scan
CREATE INDEX idx_entity_master_id ON entity_master (canonical_id)
    INCLUDE (canonical_name, shard_presence);

-- Identifier lookup indexes (partial: skip NULL values)
CREATE INDEX idx_entity_ein ON entity_master (ein) WHERE ein IS NOT NULL;
CREATE INDEX idx_entity_cik ON entity_master (cik) WHERE cik IS NOT NULL;
CREATE INDEX idx_entity_uei ON entity_master (uei) WHERE uei IS NOT NULL;

// TypeScript; Cloudflare Workers runtime
// Bit index -> (D1 binding, table name, date column)
const SHARDS: Record<number, [string, string, string]> = {
  0: ['SECURITIES',    'edgar_companies',    'last_updated'],
  1: ['FINANCIAL_CRIMES', 'ofac_sdn',         'last_updated'],
  2: ['HEALTHCARE',    'fda_warning_letters', 'issued_date'],
  3: ['LABOR_SAFETY',  'osha_citations',      'citation_date'],
  4: ['ENVIRONMENT',   'epa_enforcement',     'initiated_date'],
  5: ['TRANSPORTATION','nhtsa_recalls',       'recall_initiation_date'],
  6: ['ENFORCEMENT',   'doj_enforcement',     'action_date'],
  7: ['INFRASTRUCTURE','sam_entities',        'registration_date'],
};

export async function queryEntityAllShards(
  canonicalId: string, env: Env,
): Promise<Record<string, unknown>> {
  // Step 1: resolve entity + shard_presence bitmask (3-6ms)
  const row = await (env.ENTITY_MASTER as D1Database)
    .prepare('SELECT canonical_name, shard_presence FROM entity_master WHERE canonical_id = ?')
    .bind(canonicalId).first<{ canonical_name: string; shard_presence: number }>();
  if (!row) throw new Error('Entity not found');

  // Step 2: collect bit indexes for shards with records
  const bits = Object.keys(SHARDS).map(Number)
    .filter(b => row.shard_presence & (1 << b));

  // Step 3: parallel fan-out (8-18ms per shard; wall-clock = slowest shard)
  const results = await Promise.all(bits.map(async (bit) => {
    const [dbKey, table, dateCol] = SHARDS[bit];
    const db = (env as Record<string, D1Database>)[dbKey];
    const { results: rows } = await db
      .prepare('SELECT * FROM ' + table +
               ' WHERE canonical_entity_id = ?' +
               ' ORDER BY ' + dateCol + ' DESC LIMIT 50')
      .bind(canonicalId).all();
    return { table, rows };
  }));

  return {
    canonical_id: canonicalId,
    canonical_name: row.canonical_name,
    records: Object.fromEntries(results.map(r => [r.table, r.rows])),
  };
}