# Infrastructure Plane

Cross-cutting services used by all planes. No business logic. No protocol-specific code.

---

## infra_repo — Database Access

Single `Ecto.Repo` for the entire umbrella. All apps reference `InfraRepo.Repo`.

### Tables by Owning App

| Table | Owning App | Purpose |
|-------|-----------|---------|
| `audit_events` | mw_audit | Compliance audit trail |
| `route_rules` | mw_router | Dynamic routing configuration |
| `api_keys` | mw_auth | Hashed M2M API credentials |
| `token_revocations` | mw_auth | JWT JTI blocklist |
| `dead_letter_queue` | infra_queue | Failed Broadway messages |
| `async_jobs` | gateway_api | Background job status |
| `transform_rules` | mw_transform | Field mapping rules |
| `schema_registry` | mw_transform | JSON Schema per message type |
| `device_registrations` | gateway_mobile | FCM/APNS device tokens |
| `file_processing_log` | adapter_file | SFTP file cursor tracking |

### Migration Naming Convention
```
YYYYMMDDHHMMSS_<owning_app>_<description>.exs

Examples:
  20260426100000_mw_audit_create_audit_events.exs
  20260426100001_mw_router_create_route_rules.exs
  20260426100002_mw_auth_create_api_keys.exs
```

### Connection Pool Sizing
```elixir
# config/runtime.exs
config :infra_repo, InfraRepo.Repo,
  pool_size: String.to_integer(System.get_env("DB_POOL_SIZE", "10")),
  queue_target: 50,       # ms — start queuing after 50ms
  queue_interval: 1_000   # ms — measure every 1s
```

Total connections = `pool_size` × number of nodes.
MySQL `max_connections` must be > `pool_size × N` + buffer.

---

## infra_cache — Caching Layer

### ETS Tables

| Table Name | Contents | TTL | Owner |
|------------|---------|-----|-------|
| `:mw_route_table` | Routing rules | ∞ (invalidated by PubSub) | mw_router |
| `:mw_auth_api_key_index` | Key prefix → hashed record | 60s (soft) | mw_auth |
| `:mw_auth_revocations` | JTI → revoked_at | Until JWT exp | mw_auth |
| `:mw_transform_rules` | {message_type, direction} → rules | ∞ (invalidated) | mw_transform |
| `:mw_mobile_idempotency` | idempotency_key → response | 24h | gateway_mobile |

### ETS Configuration

All tables use:
- `:named_table` — access by atom name
- `:set` — unique keys
- `:public` — readable by all processes
- `read_concurrency: true` — optimised for high read, low write

### Optional Redis L2

When running multi-node and shared state is needed beyond PubSub (e.g., distributed
rate limit counters, shared idempotency keys):

```elixir
# config/runtime.exs
config :infra_cache, :redis,
  enabled: System.get_env("REDIS_URL") != nil,
  url: System.get_env("REDIS_URL", "redis://localhost:6379"),
  pool_size: 5
```

Rate limit counters via Redis (replaces `ex_rated` ETS when multi-node):
```elixir
defmodule InfraCache.RedisCache do
  def increment_rate_limit(key, window_ms) do
    {:ok, count} = Redix.command(:redix, ["INCR", key])
    if count == 1, do: Redix.command(:redix, ["PEXPIRE", key, window_ms])
    count
  end
end
```

---

## infra_queue — Broadway Pipelines

### Pipeline Registry

| Pipeline | Producer | Processor | Batcher | DLQ |
|----------|---------|-----------|---------|-----|
| `FilePipeline` | `AdapterFile.FileWatcher` | Row → canonical | `AdapterDw.BatchLoader` | `DeadLetterStore` |
| `CallbackPipeline` | `AdapterBanking.CallbackHandler` | Enrich + DB update | PubSub broadcast | `DeadLetterStore` |

### Back-pressure Tuning

```elixir
# File pipeline — optimised for throughput
producers: [concurrency: 1],
processors: [default: [concurrency: 10, min_demand: 5, max_demand: 50]],
batchers: [dw: [batch_size: 500, batch_timeout: 5_000, concurrency: 2]]

# Callback pipeline — optimised for latency
producers: [concurrency: 1],
processors: [default: [concurrency: 5, min_demand: 1, max_demand: 10]],
batchers: [pubsub: [batch_size: 10, batch_timeout: 100, concurrency: 1]]
```

### DLQ Operations

```elixir
# Re-queue: send failed message back through Broadway
InfraQueue.DeadLetterStore.requeue(entry_id)

# Bulk discard: remove entries older than N days
InfraQueue.DeadLetterStore.discard_before(~D[2026-04-01])
```

DLQ entries are never automatically deleted — only by operator action.

---

## infra_telemetry — Observability

### Telemetry Event Catalogue

All events follow the `:mw_core` prefix namespace:

| Event | Measurements | Metadata |
|-------|-------------|----------|
| `[:mw_router, :request, :stop]` | `duration` | `adapter, status, tenant_id` |
| `[:mw_auth, :verify, :stop]` | `duration` | `method, result` |
| `[:mw_auth, :failure]` | `count: 1` | `reason, tenant_id` |
| `[:adapter_banking, :send, :stop]` | `duration` | `status` |
| `[:adapter_banking, :circuit, :state_change]` | — | `from, to` |
| `[:infra_queue, :broadway, :message, :stop]` | `duration` | `pipeline, status` |
| `[:infra_queue, :dlq, :depth]` | `count` | `pipeline` |

### OTel Span Names

| Span | Set by |
|------|--------|
| `gateway_api.request` | `opentelemetry_phoenix` (auto) |
| `mw_auth.verify` | `infra_telemetry` custom span |
| `mw_router.pipeline` | `infra_telemetry` custom span |
| `adapter_banking.send` | `infra_telemetry` custom span |
| `ecto.query` | `opentelemetry_ecto` (auto) |
| `finch.request` | `opentelemetry_finch` (auto) |

### LiveDashboard

Available at `/admin/dashboard` (via `gateway_web`):
```elixir
live_dashboard "/dashboard",
  metrics: InfraTelemetry.Metrics,
  ecto_repos: [InfraRepo.Repo],
  additional_pages: [
    broadway: {BroadwayDashboard, pipelines: [InfraQueue.FilePipeline]}
  ]
```