ocsf_ingest

# ocsf_ingest

[![License](https://img.shields.io/badge/license-Apache%202.0-blue.svg)](LICENSE)

Backend-agnostic ingestion pipeline for the
[`ocsf`](https://hex.pm/packages/ocsf) Elixir library. Accepts OCSF events on a
non-blocking hot path, buffers them durably, batches them, and writes them
through any `OCSF.Sink` implementation — without depending on the underlying
database.

- **Non-blocking dispatch** — `OCSF.Ingest.dispatch/2` enqueues and returns
  immediately; writes happen asynchronously in batches.
- **Durable buffer** — an ETS staging area with a two-phase
  (`:pending`/in-flight) record model survives process crashes and graceful
  restarts.
- **Back-pressure, not memory blowup** — GenStage demand absorbs slow sinks; a
  bounded buffer drops explicitly (with telemetry) instead of growing without
  limit.
- **Idempotent end-to-end** — at-least-once delivery combined with sinks that
  write idempotently on `metadata.uid` means retries never duplicate rows.
- **Per-tenant isolation that scales** — events are partitioned per tenant
  (by key and counters, not one table per tenant), so one noisy tenant can only
  fill and overflow its own quota, never another's — even with thousands of
  tenants.
- **Adapts to the sink** — batch size, timeout, queue size, and concurrency are
  resolved per sink from configuration.

## Why

Emission code on an authentication or security hot path must never block on a
slow or degraded database. `ocsf_ingest` sits between the core library and the
sink, turning a synchronous `OCSF.Sink.write/1` into a buffered, batched,
crash-safe pipeline. It is the layer that **relieves the database**: instead of
one synchronous `INSERT` per event, the database receives a steady stream of
batched bulk writes, rate-limited by back-pressure (see
[Database load](#database-load--tuning)).

## When to use it

Use `ocsf_ingest` when emission is on a hot path, volume is non-trivial, or you
want batching, back-pressure, crash-safety, and per-tenant isolation — i.e. most
production emission.

You can also **skip it** and call a sink directly (e.g.
`OCSF.Ecto.Sink.write([event])`) when:

- volume is low and the calling context can afford a synchronous DB round-trip;
- you need a synchronous success/failure result at the call site;
- it's a one-off backfill or a script.

Direct writes are synchronous and block until the database returns;
`dispatch/2` never blocks and never surfaces sink errors at the call site.

## Architecture

```mermaid
flowchart LR
  E[OCSF.Event] -->|dispatch/2| B[("ETS buffer\nper tenant")]
  B --> P[GenStage producer pool]
  P --> BW[Broadway pipeline]
  BW -->|handle_batch| S[[OCSF.Sink.write/1]]
  S --> J[Janitor]
```

`ocsf_ingest` depends only on `ocsf`, `broadway`, and `telemetry`. The concrete
sink (e.g. `ocsf_ecto` for Postgres) is injected by configuration and resolved
at runtime — this library never depends on a database driver.

## Adoption

### 1. Add the dependencies

You need the core, this pipeline, and at least one sink:

```elixir
# mix.exs
def deps do
  [
    {:ocsf, "~> 0.2"},
    {:ocsf_ingest, "~> 0.1"},
    {:ocsf_ecto, "~> 0.1"}
  ]
end
```

### 2. Configure the sink(s)

Each configured sink gets its own pipeline tree. Declare them under
`config :ocsf_ingest, OCSF.Ingest`:

```elixir
# config/config.exs
config :ocsf_ingest, OCSF.Ingest,
  # how to attribute an event to a tenant (its isolation boundary)
  tenant_path: [:user, :org, :uid],
  default_tenant: :system,
  sinks: [
    {OCSF.Ecto.Sink,
     batch_size: 500,
     batch_timeout_ms: 100,
     max_queue_size: 30_000,            # per-tenant pending cap
     max_global_queue_size: 1_000_000}  # node-level pending cap for this sink
  ]
```

The tenant is resolved per event from `tenant_path` (falling back to
`default_tenant`), or overridden per call with `dispatch(event, tenant: ...)`.

### 3. Start it

The library **starts itself** through its OTP application
(`OCSF.Ingest.Application`) as soon as it is a dependency — there is nothing to
add to your supervision tree. Booting reads the `sinks` config and starts one
supervised tree per sink. (Configure the sink's own runtime too — e.g. point
`ocsf_ecto` at a Repo and run its migration; see step 4.)

### 4. Set up the sink's storage

Each sink owns its storage setup. For `ocsf_ecto`: configure the Repo + Cloak
vault and run the migration (`OCSF.Ecto.Migration`) — see the
[`ocsf_ecto` README](https://hex.pm/packages/ocsf_ecto).

### 5. Dispatch events

```elixir
# Build an event via the core library
{:ok, event} =
  OCSF.Events.Authentication.logon(
    user: %{uid: "u1", org: %{uid: "acme"}},
    service: %{name: "My Auth"},
    status: :Success,
    severity: :Informational
  )

# Hand it to the pipeline — returns immediately
:ok = OCSF.Ingest.dispatch(event)

# Batch dispatch
:ok = OCSF.Ingest.dispatch(many_events)

# Target a specific sink and/or override the tenant
:ok = OCSF.Ingest.dispatch(event, sink: OCSF.Ecto.Sink, tenant: "acme")
```

`dispatch/2` returns `{:error, :overflow}` when the target tenant's buffer is
full, or `{:error, :no_sink}` when no sink can be resolved. All other conditions
(sink errors, transient outages) are handled asynchronously downstream and
retried. All emission is async — there is no blocking variant.

## Per-sink knobs

Knobs are resolved per sink, later sources winning:

1. the options in `sinks:` config (deployment);
2. the sink module's `ingest_defaults/0`, if it exports one;
3. the library defaults below.

| Knob                    | Default                      | Effect |
|-------------------------|------------------------------|--------|
| `batch_size`            | 500                          | rows per `insert_all` — bigger ⇒ fewer DB round-trips |
| `batch_timeout_ms`      | 100                          | max wait before flushing a partial batch |
| `concurrency`           | `System.schedulers_online()` | parallel batchers ⇒ bounds concurrent DB writes |
| `max_queue_size`        | 30 000                       | per-tenant pending cap (overflow isolation) |
| `max_global_queue_size` | 1 000 000                    | node-level pending cap (memory bound) |

## Database load & tuning

The whole point of the pipeline is to **relieve the database**:

- **Batched bulk writes** — events are grouped (`batch_size` / `batch_timeout_ms`)
  and written with a single `OCSF.Sink.write/1` per batch (e.g. one multi-row
  `INSERT` via `Repo.insert_all/3`), not one `INSERT` per event.
- **Back-pressure** — the pipeline only pulls what it can write; if the database
  slows, demand drops and events accumulate in the ETS buffer instead of
  saturating the connection pool.
- **Burst smoothing** — spikes are absorbed in memory and drained at a steady,
  database-friendly rate.
- **Bounded concurrency** — `concurrency` caps simultaneous writes/connections.
- **Overflow, not collapse** — past `max_queue_size` / `max_global_queue_size`,
  the newest events are dropped (with telemetry) rather than overwhelming the DB.

Tune `batch_size` up (and `batch_timeout_ms` to taste) to trade latency for
fewer, larger writes when the database is the bottleneck. For very high volume,
add a columnar sink (e.g. ClickHouse) alongside Postgres — the pipeline
fans out to multiple sinks unchanged.

## Telemetry

| Event                          | Measurements              | Metadata                              |
|--------------------------------|---------------------------|---------------------------------------|
| `[:ocsf, :ingest, :dispatch]`  | `%{count}`                | `%{sink, tenant}`                     |
| `[:ocsf, :ingest, :batch]`     | `%{count, duration}`      | `%{sink, result}`                     |
| `[:ocsf, :ingest, :overflow]`  | `%{count}`                | `%{sink, tenant, reason}`             |
| `[:ocsf, :ingest, :buffer]`    | `%{tenants, pending, inflight}` | `%{sink}`                       |

`tenant` is in event metadata for routing/sampling but **must not** be exported
as a metric label by default — thousands of tenants would explode metric series.
The `[:ocsf, :ingest, :buffer]` gauge is aggregated per sink. Attach a
module-based handler (never an anonymous function — `:telemetry` warns about it).

## Durability

Events are staged in ETS as `:pending`, marked in-flight when taken by the
pipeline, and deleted only after the sink confirms the write. A crash mid-flight
leaves events recoverable: the in-flight records of a dead producer are
reclaimed to `:pending` and re-delivered. Because sinks write idempotently on
`metadata.uid`, retries are no-ops at the row level.

The buffer is in-memory (ETS): it survives process crashes, but not a full node
crash or power loss, and it drops the newest events of a tenant whose queue
overflows. This at-least-once, in-memory guarantee is the v0 durability level.

## Links

- [`ocsf`](https://hex.pm/packages/ocsf) — core library
- [`ocsf_ecto`](https://hex.pm/packages/ocsf_ecto) — Postgres sink
- [OCSF 1.9 Schema](https://schema.ocsf.io/1.9.0/)
- [Broadway](https://hex.pm/packages/broadway) — batching pipeline

## License

Apache-2.0