Building a Secure Entity Registry Database Schema #

CREATE EXTENSION IF NOT EXISTS "pgcrypto";
CREATE EXTENSION IF NOT EXISTS "uuid-ossp";

-- Core entity identity (immutable public metadata)
CREATE TABLE entities (
    entity_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    legal_name TEXT NOT NULL,
    jurisdiction_code CHAR(2) NOT NULL,
    formation_date DATE NOT NULL,
    status VARCHAR(20) NOT NULL CHECK (status IN ('ACTIVE', 'DISSOLVED', 'MERGED', 'SUSPENDED')),
    registered_agent_name TEXT,
    registered_agent_address TEXT,
    created_at TIMESTAMPTZ DEFAULT NOW(),
    updated_at TIMESTAMPTZ DEFAULT NOW()
);

-- Jurisdictional compliance mapping (filing deadlines, tax IDs, state portals)
CREATE TABLE compliance_profiles (
    profile_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    entity_id UUID NOT NULL REFERENCES entities(entity_id) ON DELETE CASCADE,
    state_tax_id_encrypted BYTEA NOT NULL,
    annual_filing_due_month INT CHECK (annual_filing_due_month BETWEEN 1 AND 12),
    franchise_tax_amount NUMERIC(10,2),
    next_filing_deadline DATE GENERATED ALWAYS AS (
        CASE WHEN EXTRACT(MONTH FROM CURRENT_DATE) > annual_filing_due_month
             THEN MAKE_DATE(EXTRACT(YEAR FROM CURRENT_DATE)::INT + 1, annual_filing_due_month, 1)
             ELSE MAKE_DATE(EXTRACT(YEAR FROM CURRENT_DATE)::INT, annual_filing_due_month, 1)
        END
    ) STORED,
    portal_endpoint_url TEXT,
    CONSTRAINT unique_entity_profile UNIQUE (entity_id)
);

-- Beneficial ownership (FinCEN CTA compliant, strictly isolated)
CREATE TABLE beneficial_owners (
    owner_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    entity_id UUID NOT NULL REFERENCES entities(entity_id) ON DELETE CASCADE,
    fincen_id_encrypted BYTEA NOT NULL,
    full_name_encrypted BYTEA NOT NULL,
    dob_encrypted BYTEA NOT NULL,
    ownership_pct DECIMAL(5,2) CHECK (ownership_pct > 0 AND ownership_pct <= 100),
    control_type VARCHAR(30) CHECK (control_type IN ('SUBSTANTIAL_CONTROL', 'EQUITY_OWNER', 'SENIOR_OFFICER')),
    reported_at TIMESTAMPTZ DEFAULT NOW()
);

-- Immutable audit trail (append-only, cryptographically chained)
-- prev_log_hash references payload_hash, not log_id, to support chain verification.
-- BYTEA foreign key requires a unique index on audit_log(payload_hash).
CREATE TABLE audit_log (
    log_id BIGSERIAL PRIMARY KEY,
    entity_id UUID REFERENCES entities(entity_id),
    action VARCHAR(30) NOT NULL,
    actor_role VARCHAR(30) NOT NULL,
    payload_hash BYTEA NOT NULL,
    prev_log_hash BYTEA,
    recorded_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE UNIQUE INDEX ON audit_log(payload_hash);

-- Row-Level Security Policies
ALTER TABLE entities ENABLE ROW LEVEL SECURITY;
ALTER TABLE compliance_profiles ENABLE ROW LEVEL SECURITY;
ALTER TABLE beneficial_owners ENABLE ROW LEVEL SECURITY;
ALTER TABLE audit_log ENABLE ROW LEVEL SECURITY;

-- Compliance automation role (read-only public metadata + deadlines)
CREATE POLICY compliance_read ON entities FOR SELECT USING (true);
CREATE POLICY compliance_read_profiles ON compliance_profiles FOR SELECT USING (true);

-- BOI role (strictly isolated, requires explicit session context)
CREATE POLICY boi_access ON beneficial_owners
    FOR ALL USING (current_setting('app.boi_authorized', true)::boolean = true);
CREATE POLICY boi_insert ON beneficial_owners
    FOR INSERT WITH CHECK (current_setting('app.boi_authorized', true)::boolean = true);

-- Audit append-only enforcement
CREATE POLICY audit_append ON audit_log FOR INSERT WITH CHECK (true);
CREATE POLICY audit_read ON audit_log FOR SELECT USING (true);

CREATE OR REPLACE FUNCTION enforce_audit_chain() RETURNS TRIGGER AS $$
DECLARE
    prev_hash BYTEA;
BEGIN
    SELECT payload_hash INTO prev_hash FROM audit_log ORDER BY log_id DESC LIMIT 1;
    NEW.prev_log_hash := prev_hash;
    NEW.payload_hash  := digest(
        NEW.entity_id::TEXT || NEW.action || NEW.actor_role || NEW.recorded_at::TEXT,
        'sha256'
    );
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER trg_audit_chain
    BEFORE INSERT ON audit_log
    FOR EACH ROW EXECUTE FUNCTION enforce_audit_chain();

import hashlib
import json
import os
from typing import Dict, Any
from datetime import date
from enum import Enum

import asyncpg
import structlog
from pydantic import BaseModel, Field
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

logger = structlog.get_logger()


class FilingStatus(str, Enum):
    PENDING = "PENDING"
    SUBMITTED = "SUBMITTED"
    REJECTED = "REJECTED"
    COMPLETED = "COMPLETED"


class FilingPayload(BaseModel):
    entity_id: str
    jurisdiction: str
    filing_type: str
    due_date: date
    portal_endpoint: str
    idempotency_key: str = Field(
        default_factory=lambda: hashlib.sha256(os.urandom(32)).hexdigest()
    )


class ComplianceService:
    def __init__(self, db_pool: asyncpg.Pool, redis_client: Any):
        self.db = db_pool
        self.redis = redis_client
        self.logger = structlog.get_logger().bind(service="compliance_automation")

    async def _invalidate_cache(self, entity_id: str) -> None:
        """Explicit cache invalidation via Redis pub/sub and key deletion."""
        await self.redis.delete(f"entity:calendar:{entity_id}")
        await self.redis.publish(
            "entity:updates",
            json.dumps({"entity_id": entity_id, "action": "invalidate"}),
        )

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10),
        retry=retry_if_exception_type((ConnectionError, TimeoutError)),
        reraise=True,
    )
    async def submit_to_state_portal(self, payload: FilingPayload) -> Dict[str, Any]:
        """Portal submission: idempotency header, ACTIVE-status gate, JSON/XML fallback."""
        self.logger.info(
            "submitting_to_portal",
            entity_id=payload.entity_id,
            endpoint=payload.portal_endpoint,
        )
        try:
            async with self.db.acquire() as conn:
                # Reads only public metadata — never touches beneficial_owners.
                status = await conn.fetchval(
                    "SELECT status FROM entities WHERE entity_id = $1", payload.entity_id
                )
                if status != "ACTIVE":
                    raise ValueError(
                        f"Entity {payload.entity_id} is {status}. Submission blocked."
                    )

            response = await self._http_post(
                url=payload.portal_endpoint,
                headers={
                    "Idempotency-Key": payload.idempotency_key,
                    "Content-Type": "application/json",
                },
                json=payload.model_dump(mode="json"),
            )
            return response
        except Exception as e:
            self.logger.warning("primary_portal_failed", error=str(e))
            return await self._fallback_submission(payload)

    async def _fallback_submission(self, payload: FilingPayload) -> Dict[str, Any]:
        """Fallback chain for portal rate limits or deprecated endpoints."""
        self.logger.info("invoking_fallback_chain", entity_id=payload.entity_id)
        try:
            return await self._http_post_xml(
                url=f"{payload.portal_endpoint}/legacy",
                payload=payload,
            )
        except Exception as xml_err:
            self.logger.error("xml_fallback_failed", error=str(xml_err))
            await self.redis.lpush(
                "manual_filing_queue",
                json.dumps(payload.model_dump(mode="json")),
            )
            return {"status": "QUEUED_MANUAL", "reason": "PORTAL_UNAVAILABLE"}

    async def process_annual_calendar(self) -> None:
        """High-throughput deadline scanner with cache-aware querying."""
        async with self.db.acquire() as conn:
            rows = await conn.fetch("""
                SELECT e.entity_id, e.jurisdiction_code,
                       cp.next_filing_deadline, cp.portal_endpoint_url
                FROM compliance_profiles cp
                JOIN entities e ON cp.entity_id = e.entity_id
                WHERE e.status = 'ACTIVE'
                  AND cp.next_filing_deadline <= CURRENT_DATE + INTERVAL '30 days'
            """)
            for row in rows:
                cache_key = f"entity:calendar:{row['entity_id']}"
                if await self.redis.exists(cache_key):
                    continue  # Already processed or cached this cycle.
                filing = FilingPayload(
                    entity_id=str(row["entity_id"]),
                    jurisdiction=row["jurisdiction_code"],
                    filing_type="ANNUAL_REPORT",
                    due_date=row["next_filing_deadline"],
                    portal_endpoint=row["portal_endpoint_url"],
                )
                await self.submit_to_state_portal(filing)
                await self._invalidate_cache(str(row["entity_id"]))

    async def _http_post(self, url: str, headers: dict, json: dict) -> dict:
        # Production: use aiohttp or httpx with connection pooling.
        raise NotImplementedError

    async def _http_post_xml(self, url: str, payload: FilingPayload) -> dict:
        # Production: serialize to jurisdiction-specific XML schema.
        raise NotImplementedError

SELECT log_id, payload_hash, prev_log_hash,
       LAG(payload_hash) OVER (ORDER BY log_id) AS expected_prev
FROM audit_log
WHERE prev_log_hash IS DISTINCT FROM LAG(payload_hash) OVER (ORDER BY log_id);