```meta
title: PII data flow
subtitle: Where personal fields travel from ingestion to the warehouse, and where they stop.
tag: DRAFT
```

Personal data enters through three doors and leaves through one. Every path crosses the PII gate before it reaches a Kafka topic. No downstream consumer ever sees a raw email, phone number, or national ID. The warehouse holds two copies of each customer. Analysts query a tokenized row; only the compliance role can join the restricted identity row.

```callout
tone: note
title: Assumptions
body: "The request names no system, so this doc assumes the current stack. Sources are the Signup web form, the iOS/Android app, and a nightly partner CSV. The pipeline is the Ingest API (Go), Kafka, a Tokenization service with a Postgres token vault, and a Flink Stream Loader. The warehouse is Snowflake with `raw`, `core`, and `pii` schemas. Field names and retention periods are invented to be realistic. Confirm them against the data inventory."
```

## Path of a personal field

Three sources produce PII. The Signup form and the mobile app send JSON to the Ingest API. The partner CSV lands in S3, and the same API reads it in batch mode. The Ingest API does not store anything. It calls the Tokenization service, which writes the plaintext to the token vault, returns a stable token per field, and only then publishes to `events.clean`.

The `events.raw` topic exists only for replay of the API's own audit log; it carries field hashes, not values. The Stream Loader copies `events.clean` into `raw.events`, and the nightly dbt run builds `core.dim_customer` and `core.fact_event` from it. The `pii.customer_identity` table is loaded on a separate path from the token vault, never from Kafka.

```dfd
id: pii-dfd
dir: LR
nodes:
  - { id: signup, col: 1, row: 1, kind: external, name: Signup form }
  - { id: app, col: 1, row: 2, kind: external, name: Mobile app }
  - { id: partner, col: 1, row: 3, kind: external, name: Partner CSV (S3) }
  - { id: ingest, col: 2, row: 2, kind: process, name: Ingest API, num: 1 }
  - { id: tokenize, col: 3, row: 1, kind: process, name: Tokenization service, num: 2 }
  - { id: vault, col: 4, row: 1, kind: store, name: Token vault }
  - { id: raw, col: 3, row: 3, kind: store, name: events.raw (hashes only) }
  - { id: clean, col: 4, row: 2, kind: store, name: events.clean }
  - { id: loader, col: 5, row: 2, kind: process, name: Stream Loader, num: 3 }
  - { id: rawzone, col: 6, row: 2, kind: store, name: raw.events }
  - { id: dbt, col: 7, row: 2, kind: process, name: dbt nightly, num: 4 }
  - { id: core, col: 8, row: 2, kind: store, name: core.dim_customer }
  - { id: vaultsync, col: 5, row: 1, kind: process, name: Vault sync, num: 5 }
  - { id: piischema, col: 8, row: 1, kind: store, name: pii.customer_identity }
  - { id: analyst, col: 9, row: 2, kind: external, name: Analysts }
  - { id: compliance, col: 9, row: 1, kind: external, name: Compliance role }
edges:
  - { from: signup, to: ingest, label: "email, name, phone" }
  - { from: app, to: ingest, label: "email, device_id, geo" }
  - { from: partner, to: ingest, label: "email, national_id, dob" }
  - { from: ingest, to: tokenize, label: plaintext fields }
  - { from: tokenize, to: vault, label: "plaintext + token" }
  - { from: tokenize, to: ingest, label: tokens }
  - { from: ingest, to: raw, label: audit record (sha256) }
  - { from: ingest, to: clean, label: tokenized event }
  - { from: clean, to: loader, label: consume }
  - { from: loader, to: rawzone, label: append }
  - { from: rawzone, to: dbt, label: read }
  - { from: dbt, to: core, label: build }
  - { from: vault, to: vaultsync, label: "CDC, hourly" }
  - { from: vaultsync, to: piischema, label: encrypted load }
  - { from: core, to: analyst, label: query }
  - { from: piischema, to: compliance, label: join on customer_token }
```

## Field catalog

Each field has one transform at the gate and one landing column. `Tokenize` means a random token that the vault maps back to the value. `Hash` means sha256 with a per-tenant salt and no reverse path. `Coarsen` means precision is reduced before publishing. A field that is dropped never reaches a topic.

```table
title: PII fields and their transform at the gate
columns: [Field, Source, Class, Gate transform, Warehouse column, Readable by]
rows:
  - [email, "Signup, app, partner", Direct identifier, Tokenize, core.dim_customer.email_token, Analysts]
  - [full_name, Signup, Direct identifier, Tokenize, core.dim_customer.name_token, Analysts]
  - [phone, Signup, Direct identifier, Tokenize, core.dim_customer.phone_token, Analysts]
  - [national_id, Partner, Special category, Tokenize, "pii.customer_identity.national_id (encrypted)", Compliance role]
  - [dob, Partner, Quasi identifier, Coarsen to birth_year, core.dim_customer.birth_year, Analysts]
  - [device_id, App, Quasi identifier, Hash, core.fact_event.device_hash, Analysts]
  - [geo (lat/lon), App, Quasi identifier, Coarsen to city, core.fact_event.city, Analysts]
  - [ip_address, "Signup, app", Quasi identifier, { v: Drop, tone: neg }, { v: none, tone: muted }, { v: nobody, tone: muted }]
  - [user_agent, "Signup, app", Non-PII, Pass through, raw.events.user_agent, Analysts]
```

## Shape at rest in the warehouse

Analysts join on `customer_token`. The `pii` schema shares that key, so a compliance query can reattach identity to any row. No `core` table carries a column that can be reversed without the vault.

```erd
id: pii-erd
dir: LR
entities:
  - name: dim_customer
    schema: core
    columns:
      - customer_token uuid pk
      - email_token text unique !null
      - name_token text
      - phone_token text
      - birth_year int
      - country text
      - created_at timestamp !null
  - name: fact_event
    schema: core
    columns:
      - event_id uuid pk
      - customer_token uuid fk -> dim_customer.customer_token
      - event_type text !null
      - device_hash text
      - city text
      - occurred_at timestamp !null
  - name: customer_identity
    schema: pii
    note: "Row-level encrypted; readable only by the compliance role. Loaded from the token vault, not from Kafka."
    columns:
      - customer_token uuid pk
      - email text !null
      - full_name text
      - phone text
      - national_id text
      - dob date
      - vault_version int !null
      - synced_at timestamp !null
relations:
  - dim_customer ||--o{ fact_event: has
  - dim_customer ||--|| customer_identity: resolves to
```

## Invariants

```spec
title: Rules the pipeline must keep
accent: red
rows:
  - { label: Gate, value: "Every producer calls the Ingest API. No service publishes to events.clean directly." }
  - { label: Plaintext boundary, value: "Plaintext PII exists in the Ingest API request memory, the Tokenization service, the token vault, and pii.customer_identity. Nowhere else." }
  - { label: Kafka, value: "events.raw carries sha256 hashes for audit replay. events.clean carries tokens. Neither topic carries a reversible value." }
  - { label: Deletion, steps: [Erasure request arrives, Vault deletes the plaintext row, Vault sync removes the pii row within one hour, "core tokens stay but resolve to nothing"] }
  - { label: Retention, value: "Token vault 7 years; pii.customer_identity mirrors the vault; raw.events 90 days; core tables indefinite (tokens only)." }
  - { label: Access, value: "Analysts read core and raw. The compliance role reads pii. No role has both write access to the vault and read access to pii." }
```