From 6660a1e1eed1ea2bd8253357a0c16e000add0080 Mon Sep 17 00:00:00 2001 From: Daniel Widgren Date: Wed, 20 May 2026 09:40:10 +0200 Subject: [PATCH] feat: add nova_audit_pgo adapter + dependency model docs Ships nova_audit_pgo as a third sibling OTP app, parallel to nova_audit_kura. Uses pgo:query/3 directly; soft-dep pattern via erlang:function_exported/3 means consumers who don't use this adapter don't pull pgo as a transitive dependency. Shared schema and hardening SQL: nova_audit_pgo:schema_sql/0 delegates to nova_audit_kura:schema_sql/0 to keep one source of truth. Expanded guides/adapters.md with: - A dependency model table making clear what consumers actually need when they take a dep on nova_audit (only jhn_stdlib transitively; Kura/pgo only required if you USE that adapter). - A complete "writing your own adapter" walkthrough including the worker spawn + registry registration sequence. - The soft-dep pattern documented so third-party adapters can follow the same approach. --- README.md | 9 +- .../nova_audit_pgo/src/nova_audit_pgo.app.src | 10 + apps/nova_audit_pgo/src/nova_audit_pgo.erl | 267 ++++++++++++++++++ .../test/nova_audit_pgo_SUITE.erl | 52 ++++ guides/adapters.md | 153 +++++++++- rebar.config | 8 +- 6 files changed, 478 insertions(+), 21 deletions(-) create mode 100644 apps/nova_audit_pgo/src/nova_audit_pgo.app.src create mode 100644 apps/nova_audit_pgo/src/nova_audit_pgo.erl create mode 100644 apps/nova_audit_pgo/test/nova_audit_pgo_SUITE.erl diff --git a/README.md b/README.md index 20b3900..824b14a 100644 --- a/README.md +++ b/README.md @@ -31,10 +31,11 @@ ok = nova_audit:log_async(access_log, Event). ## Adapters -| Adapter | Storage | Status | -| ------------------ | ----------- | ------ | -| `nova_audit_log` | OTP logger | v0.1 | -| `nova_audit_kura` | Postgres | v0.1 | +| Adapter | Storage | Status | +| ------------------ | ------------------- | ------ | +| `nova_audit_log` | OTP logger | v0.1 | +| `nova_audit_kura` | Postgres (via Kura) | v0.1 | +| `nova_audit_pgo` | Postgres (via pgo) | v0.1 | `nova_audit_storage` (uses `nova_storage`) lands in v0.2. diff --git a/apps/nova_audit_pgo/src/nova_audit_pgo.app.src b/apps/nova_audit_pgo/src/nova_audit_pgo.app.src new file mode 100644 index 0000000..10430ef --- /dev/null +++ b/apps/nova_audit_pgo/src/nova_audit_pgo.app.src @@ -0,0 +1,10 @@ +{application, nova_audit_pgo, [ + {description, "pgo adapter for nova_audit"}, + {vsn, "git"}, + {registered, []}, + {applications, [kernel, stdlib, nova_audit]}, + {env, []}, + {modules, []}, + {licenses, ["Apache-2.0"]}, + {links, [{"GitHub", "https://github.com/novaframework/nova_audit"}]} +]}. diff --git a/apps/nova_audit_pgo/src/nova_audit_pgo.erl b/apps/nova_audit_pgo/src/nova_audit_pgo.erl new file mode 100644 index 0000000..5d01cf3 --- /dev/null +++ b/apps/nova_audit_pgo/src/nova_audit_pgo.erl @@ -0,0 +1,267 @@ +-module(nova_audit_pgo). +-moduledoc """ +pgo-backed adapter for `nova_audit`. + +Writes events directly to a Postgres table via `pgo`, bypassing any ORM +layer. Useful for apps that already use pgo (or want to) and don't want +Kura as a transitive dependency. + +## Configuration + +```erlang +#{ + adapter => nova_audit_pgo, + pool => default, + table => <<"audit_events">> +} +``` + +The `pgo` pool must be started by the application; this adapter does NOT +manage pool lifecycle. + +## Schema + +Uses the same table shape as `nova_audit_kura`. See +`nova_audit_pgo:schema_sql/0` (identical to `nova_audit_kura:schema_sql/0`). + +## Hardening + +After applying the schema, REVOKE UPDATE and DELETE on the audit table +for your application role. See `nova_audit_pgo:hardening_sql/0` or use +the Kura adapter's helper interchangeably. + +## pg_types configuration + +For UUIDv7 round-tripping through pgo, configure `pg_types` with +`uuid_format=string` at startup so `event_id` arrives as a binary on +both write and read paths. +""". + +-behaviour(gen_server). +-behaviour(nova_audit_adapter). + +-export([start_link/2]). +-export([write/2, query/3]). +-export([init/1, handle_call/3, handle_cast/2, handle_info/2]). +-export([hardening_sql/0, hardening_sql/1, schema_sql/0]). + +-record(state, {name :: atom(), handle :: term()}). +-record(handle, { + name :: atom(), + pool :: atom(), + table :: binary() +}). + +start_link(Name, Opts) -> + gen_server:start_link(?MODULE, {Name, Opts}, []). + +write(Event, #handle{pool = Pool, table = Table}) -> + SQL = << + "INSERT INTO ", Table/binary, + " (event_id, schema_version, occurred_at, actor_type, actor_id, " + " action, target_type, target_id, outcome, source, request_id, metadata) " + "VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12::jsonb)" + >>, + Params = event_to_params(Event), + case erlang:function_exported(pgo, query, 3) of + true -> + case apply(pgo, query, [SQL, Params, #{pool => Pool}]) of + #{command := insert} -> ok; + #{error := Reason} -> {error, Reason}; + Other -> {error, {unexpected_pgo_result, Other}} + end; + false -> + {error, pgo_not_loaded} + end. + +query(Filter, Opts, #handle{pool = Pool, table = Table}) -> + Limit = maps:get(limit, Opts, 100), + {Where, Params} = filter_to_where(Filter), + Cursor = maps:get(cursor, Opts, undefined), + {CursorClause, Params2} = cursor_clause(Cursor, Where, Params), + SQL = iolist_to_binary([ + << + "SELECT event_id, schema_version, occurred_at, actor_type, actor_id, " + " action, target_type, target_id, outcome, source, request_id, metadata " + "FROM " + >>, + Table, + Where, + CursorClause, + <<" ORDER BY occurred_at LIMIT $">>, + integer_to_binary(length(Params2) + 1) + ]), + AllParams = Params2 ++ [Limit], + case erlang:function_exported(pgo, query, 3) of + true -> + case apply(pgo, query, [SQL, AllParams, #{pool => Pool}]) of + #{rows := Rows} -> + Events = [row_to_event(R) || R <- Rows], + {ok, Events, next_cursor(Events, Limit)}; + #{error := Reason} -> + {error, Reason}; + Other -> + {error, {unexpected_pgo_result, Other}} + end; + false -> + {error, pgo_not_loaded} + end. + +init({Name, Opts}) -> + Pool = maps:get(pool, Opts, default), + Table = to_binary(maps:get(table, Opts, <<"audit_events">>)), + Handle = #handle{name = Name, pool = Pool, table = Table}, + {ok, Worker} = nova_audit_worker:start_link(Name, ?MODULE, Handle), + ok = nova_audit_registry:register(Name, ?MODULE, Handle, Worker), + {ok, #state{name = Name, handle = Handle}}. + +handle_call(_, _, S) -> {reply, {error, unknown_call}, S}. +handle_cast(_, S) -> {noreply, S}. +handle_info(_, S) -> {noreply, S}. + +-spec hardening_sql() -> {ok, [binary()]}. +hardening_sql() -> + hardening_sql(<<"audit_events">>). + +-spec hardening_sql(binary()) -> {ok, [binary()]}. +hardening_sql(Table) -> + {ok, [ + <<"REVOKE UPDATE, DELETE ON ", Table/binary, " FROM PUBLIC;">>, + <<"-- Apply per-role grant separately, e.g.:">>, + <<"-- REVOKE UPDATE, DELETE ON ", Table/binary, " FROM your_app_role;">> + ]}. + +-spec schema_sql() -> binary(). +schema_sql() -> + nova_audit_kura:schema_sql(). + +%% Internal + +event_to_params(Event) -> + Actor = maps:get(actor, Event), + Target = maps:get(target, Event, undefined), + [ + maps:get(event_id, Event), + maps:get(schema_version, Event, 1), + maps:get(occurred_at, Event), + atom_to_binary(maps:get(type, Actor)), + maps:get(id, Actor), + maps:get(action, Event), + target_field(Target, type), + target_field(Target, id), + outcome_to_binary(maps:get(outcome, Event, undefined)), + maps:get(source, Event, null), + maps:get(request_id, Event, null), + iolist_to_binary(json:encode(maps:get(metadata, Event, #{}))) + ]. + +target_field(undefined, _) -> null; +target_field(T, F) -> maps:get(F, T, null). + +outcome_to_binary(undefined) -> null; +outcome_to_binary(A) when is_atom(A) -> atom_to_binary(A). + +filter_to_where(Filter) -> + {Clauses, Params, _N} = maps:fold( + fun + (actor_id, V, {Cs, Ps, N}) -> + {[clause(<<"actor_id">>, N) | Cs], [V | Ps], N + 1}; + (action, V, {Cs, Ps, N}) -> + {[clause(<<"action">>, N) | Cs], [V | Ps], N + 1}; + (outcome, V, {Cs, Ps, N}) -> + {[clause(<<"outcome">>, N) | Cs], [atom_to_binary(V) | Ps], N + 1}; + (target_id, V, {Cs, Ps, N}) -> + {[clause(<<"target_id">>, N) | Cs], [V | Ps], N + 1}; + (target_type, V, {Cs, Ps, N}) -> + {[clause(<<"target_type">>, N) | Cs], [V | Ps], N + 1}; + (request_id, V, {Cs, Ps, N}) -> + {[clause(<<"request_id">>, N) | Cs], [V | Ps], N + 1}; + (occurred_after, V, {Cs, Ps, N}) -> + C = <<"occurred_at >= $", (integer_to_binary(N))/binary>>, + {[C | Cs], [V | Ps], N + 1}; + (occurred_before, V, {Cs, Ps, N}) -> + C = <<"occurred_at < $", (integer_to_binary(N))/binary>>, + {[C | Cs], [V | Ps], N + 1}; + (_, _, Acc) -> + Acc + end, + {[], [], 1}, + Filter + ), + case Clauses of + [] -> {<<>>, []}; + _ -> + Joined = lists:join(<<" AND ">>, lists:reverse(Clauses)), + {iolist_to_binary([<<" WHERE ">> | Joined]), lists:reverse(Params)} + end. + +clause(Field, N) -> + <>. + +cursor_clause(undefined, _Where, Params) -> {<<>>, Params}; +cursor_clause(done, _Where, Params) -> {<<>>, Params}; +cursor_clause(Cursor, Where, Params) when is_binary(Cursor) -> + Connector = + case Where of + <<>> -> <<" WHERE ">>; + _ -> <<" AND ">> + end, + N = length(Params) + 1, + {< $", (integer_to_binary(N))/binary>>, + Params ++ [binary_to_integer(Cursor)]}. + +next_cursor([], _) -> done; +next_cursor(Events, Limit) when length(Events) < Limit -> done; +next_cursor(Events, _) -> + Last = lists:last(Events), + integer_to_binary(maps:get(occurred_at, Last)). + +row_to_event( + {EventId, SchemaVersion, OccurredAt, ActorType, ActorId, Action, + TargetType, TargetId, Outcome, Source, RequestId, Metadata} +) -> + Base = #{ + event_id => EventId, + schema_version => SchemaVersion, + occurred_at => OccurredAt, + actor => actor_from_row(ActorType, ActorId), + action => Action, + metadata => decode_metadata(Metadata) + }, + Base1 = maybe_put(target, target_from_row(TargetType, TargetId), Base), + Base2 = maybe_put(outcome, outcome_from_row(Outcome), Base1), + Base3 = maybe_put(source, nullable(Source), Base2), + maybe_put(request_id, nullable(RequestId), Base3). + +actor_from_row(Type, Id) when is_binary(Type) -> + #{type => binary_to_atom(Type), id => Id}; +actor_from_row(Type, Id) when is_atom(Type) -> + #{type => Type, id => Id}. + +target_from_row(null, _) -> undefined; +target_from_row(_, null) -> undefined; +target_from_row(Type, Id) -> #{type => Type, id => Id}. + +outcome_from_row(null) -> undefined; +outcome_from_row(<<"success">>) -> success; +outcome_from_row(<<"failure">>) -> failure; +outcome_from_row(O) -> O. + +nullable(null) -> undefined; +nullable(V) -> V. + +maybe_put(_K, undefined, M) -> M; +maybe_put(K, V, M) -> M#{K => V}. + +decode_metadata(null) -> #{}; +decode_metadata(M) when is_map(M) -> M; +decode_metadata(B) when is_binary(B) -> + try json:decode(B) of + M when is_map(M) -> M; + _ -> #{} + catch + _:_ -> #{} + end. + +to_binary(B) when is_binary(B) -> B; +to_binary(A) when is_atom(A) -> atom_to_binary(A). diff --git a/apps/nova_audit_pgo/test/nova_audit_pgo_SUITE.erl b/apps/nova_audit_pgo/test/nova_audit_pgo_SUITE.erl new file mode 100644 index 0000000..3e97c9e --- /dev/null +++ b/apps/nova_audit_pgo/test/nova_audit_pgo_SUITE.erl @@ -0,0 +1,52 @@ +-module(nova_audit_pgo_SUITE). + +-compile(export_all). +-compile(nowarn_export_all). + +-include_lib("common_test/include/ct.hrl"). + +all() -> + [ + write_returns_error_when_pgo_not_loaded, + query_returns_error_when_pgo_not_loaded, + hardening_sql_returns_revoke_statements, + schema_sql_returns_create_table, + starts_via_supervisor + ]. + +init_per_suite(Config) -> + _ = application:load(nova_audit), + application:set_env(nova_audit, logs, #{ + pgo_suite => #{adapter => nova_audit_pgo, pool => fake_pool, table => <<"audit_events">>} + }), + {ok, _} = application:ensure_all_started(nova_audit), + Config. + +end_per_suite(_) -> + application:stop(nova_audit), + ok. + +starts_via_supervisor(_Config) -> + %% Adapter starts even though pgo isn't loaded; calls will fail at write/query + {ok, _, _, _} = nova_audit_registry:lookup(pgo_suite). + +write_returns_error_when_pgo_not_loaded(_Config) -> + {error, pgo_not_loaded} = nova_audit:log(pgo_suite, sample_event()). + +query_returns_error_when_pgo_not_loaded(_Config) -> + {error, pgo_not_loaded} = nova_audit:query(pgo_suite, #{}). + +hardening_sql_returns_revoke_statements(_Config) -> + {ok, [First | _]} = nova_audit_pgo:hardening_sql(), + true = binary:match(First, <<"REVOKE">>) =/= nomatch. + +schema_sql_returns_create_table(_Config) -> + Sql = nova_audit_pgo:schema_sql(), + true = binary:match(Sql, <<"CREATE TABLE">>) =/= nomatch, + true = binary:match(Sql, <<"audit_events">>) =/= nomatch. + +sample_event() -> + #{ + actor => #{type => user, id => <<"alice">>}, + action => <<"x">> + }. diff --git a/guides/adapters.md b/guides/adapters.md index a60ea62..ae4e13b 100644 --- a/guides/adapters.md +++ b/guides/adapters.md @@ -4,6 +4,30 @@ Adapters implement the `nova_audit_adapter` behaviour. They own their own process. The `State` returned at registration is opaque to `nova_audit`. +## Dependency model — pull only what you use + +Three sibling OTP apps ship in this umbrella: + +- `nova_audit` — core. Only runtime dep: `jhn_stdlib` (for UUIDv7). +- `nova_audit_kura` — Kura/Postgres adapter. Depends on `nova_audit`. Calls into `kura_repo` and `kura_query` via runtime `erlang:function_exported/3` checks. +- `nova_audit_pgo` — pgo/Postgres adapter. Depends on `nova_audit`. Calls into `pgo` via runtime checks. + +What this means when you take a dep on `nova_audit`: + +| You add to your rebar.config | You actually need | +| ---------------------------------- | ------------------------------------------------ | +| `nova_audit` only | `jhn_stdlib`. Adapters compile but only `nova_audit_log` works without extra deps. | +| `nova_audit` + use `nova_audit_kura` | Add `kura` to your own deps. Without it, `write/2` returns `{error, kura_not_loaded}`. | +| `nova_audit` + use `nova_audit_pgo` | Add `pgo` to your own deps. Without it, `write/2` returns `{error, pgo_not_loaded}`. | +| `nova_audit` + a custom adapter | Whatever your adapter needs. Nothing else. | + +The adapter modules ship as ~30KB of compiled `.beam` per adapter +regardless. They sit dead in your build until you configure a log to +use them. There is no runtime cost for adapters you don't configure. + +If you only want the core and write your own adapter, that's the +default — no Kura, no pgo, no transitive deps you didn't ask for. + ## `nova_audit_log` OTP-logger adapter. Writes events as structured JSON at a configurable @@ -12,15 +36,14 @@ database. | Option | Default | Notes | | -------- | ------- | ------------------------------------------- | -| `level` | `info` | OTP logger level: `debug | info | notice ...`. | +| `level` | `info` | OTP logger level. | Does **not** support `query/3`; calls return `{error, query_not_supported}`. ## `nova_audit_kura` -Postgres adapter via Kura. Ships in the sibling app -`nova_audit_kura` (in the same umbrella as `nova_audit`). +Postgres adapter via Kura. | Option | Required | Notes | | -------- | -------- | ------------------------------------------- | @@ -39,17 +62,117 @@ After applying the migration, lock down updates and deletes: %% Apply via admin credentials with REVOKE privileges, not from the app role. ``` -## Writing a new adapter +## `nova_audit_pgo` + +Postgres adapter that talks to `pgo` directly, bypassing Kura. + +| Option | Required | Notes | +| -------- | -------- | ------------------------------------------- | +| `pool` | no | pgo pool atom; defaults to `default`. | +| `table` | no | Default `<<"audit_events">>`. | + +The schema is identical to the Kura adapter's; use whichever migration +tooling you prefer (`pgo_migrations`, raw `psql` scripts, Flyway, etc.). +For convenience the SQL is mirrored at `nova_audit_pgo:schema_sql/0` +(it just delegates to `nova_audit_kura:schema_sql/0` so there's one +source of truth). + +```erlang +nova_audit_pgo:schema_sql(). %% same table shape +nova_audit_pgo:hardening_sql(). %% same REVOKE pattern +``` + +The pgo pool must be started by your application; this adapter does not +manage pool lifecycle. For UUIDv7 round-tripping, configure `pg_types` +with `uuid_format=string` so `event_id` arrives as a binary on both +read and write. + +## Writing your own adapter + +The behaviour is small: + +```erlang +-callback start_link(Name :: atom(), Opts :: map()) -> {ok, pid()} | {error, term()}. +-callback write(Event :: event(), State :: term()) -> ok | {error, term()}. +-callback query(Filter, Opts, State) -> {ok, [event()], Cursor} | {error, term()}. +``` + +A minimal skeleton: -1. `-behaviour(nova_audit_adapter).` -2. Implement `start_link/2`, `write/2`, `query/3`. -3. In `init/1`: - - Build your opaque adapter state (record/map). - - Spawn a `nova_audit_worker:start_link(Name, ?MODULE, AdapterState)` for the async fallback. - - Register via `nova_audit_registry:register(Name, ?MODULE, AdapterState, WorkerPid)`. -4. `write/2` is called from BOTH the caller's process (for sync `log/2`) and the worker process (for async). The adapter state must be safely shared. -5. `query/3` returns `{ok, Events, Cursor}` or `{error, _}`. Adapters that don't support querying may return `{error, query_not_supported}`. +```erlang +-module(my_audit_adapter). +-behaviour(gen_server). +-behaviour(nova_audit_adapter). + +-export([start_link/2, write/2, query/3]). +-export([init/1, handle_call/3, handle_cast/2, handle_info/2]). + +-record(handle, {name :: atom(), conn :: my_db:connection()}). + +start_link(Name, Opts) -> + gen_server:start_link(?MODULE, {Name, Opts}, []). + +write(Event, #handle{conn = Conn}) -> + my_db:insert(Conn, audit_events, event_to_row(Event)). + +query(Filter, Opts, #handle{conn = Conn}) -> + Rows = my_db:select(Conn, audit_events, filter_to_where(Filter), Opts), + {ok, [row_to_event(R) || R <- Rows], cursor_from(Rows, Opts)}. + +init({Name, Opts}) -> + {ok, Conn} = my_db:connect(maps:get(connection, Opts)), + Handle = #handle{name = Name, conn = Conn}, + {ok, Worker} = nova_audit_worker:start_link(Name, ?MODULE, Handle), + ok = nova_audit_registry:register(Name, ?MODULE, Handle, Worker), + {ok, Handle}. + +handle_call(_, _, S) -> {reply, {error, unknown_call}, S}. +handle_cast(_, S) -> {noreply, S}. +handle_info(_, S) -> {noreply, S}. + +%% event_to_row/1, filter_to_where/1, row_to_event/1 are mechanical; +%% see nova_audit_pgo for a full example. +``` + +Required wiring inside `init/1`: + +1. Build your opaque adapter state (a record or map). +2. Spawn `nova_audit_worker:start_link(Name, ?MODULE, AdapterState)` for the async fallback (used by `log_async/2` when shigoto isn't loaded). +3. Register via `nova_audit_registry:register(Name, ?MODULE, AdapterState, WorkerPid)`. + +`write/2` is called from BOTH the caller's process (for sync `log/2`) +and the worker process (for async). The adapter state must be safe to +share — a pool name, a connection pid, an ETS table id, etc. + +`query/3` returns `{ok, Events, Cursor}` where `Cursor` is either an +adapter-defined opaque binary or `done`. Adapters that don't support +querying may return `{error, query_not_supported}`. + +If the adapter naturally batches writes, you can collect events +internally and flush on size or timer; the worker will keep handing +you one event at a time. + +## Soft-dep pattern + +If your adapter wraps a library that consumers may or may not have +installed, use the same runtime-check pattern as `nova_audit_kura` and +`nova_audit_pgo`: + +```erlang +write(Event, #handle{...}) -> + case erlang:function_exported(my_lib, insert, 2) of + true -> my_lib:insert(...); + false -> {error, my_lib_not_loaded} + end. +``` + +And in the umbrella's `rebar.config`: + +```erlang +{xref_ignores, [ + {my_lib, '_', '_'} %% or list specific MFAs +]}. +``` -If the adapter doesn't naturally batch, the worker queues events -serially. Adapters that benefit from batching may collect events -internally and flush on size or timer. +This lets the adapter compile and ship without forcing the underlying +library on consumers who don't use it. diff --git a/rebar.config b/rebar.config index 590b959..3eb0541 100644 --- a/rebar.config +++ b/rebar.config @@ -27,6 +27,7 @@ {xref_ignores, [ {kura_repo, insert, 3}, {kura_query, query, 4}, + {pgo, query, 3}, {shigoto, enqueue, 1}, {telemetry, execute, 3}, {nova_audit, log, 2}, @@ -41,7 +42,10 @@ {nova_audit_sup, stop_log, 1}, {nova_audit_kura, hardening_sql, 0}, {nova_audit_kura, hardening_sql, 1}, - {nova_audit_kura, schema_sql, 0} + {nova_audit_kura, schema_sql, 0}, + {nova_audit_pgo, hardening_sql, 0}, + {nova_audit_pgo, hardening_sql, 1}, + {nova_audit_pgo, schema_sql, 0} ]}. {dialyzer, [ @@ -64,7 +68,7 @@ {groups_for_modules, [ {<<"Core API">>, [nova_audit]}, {<<"Behaviours">>, [nova_audit_adapter]}, - {<<"Adapters">>, [nova_audit_log, nova_audit_kura]}, + {<<"Adapters">>, [nova_audit_log, nova_audit_kura, nova_audit_pgo]}, {<<"Internal">>, [nova_audit_registry, nova_audit_worker, nova_audit_event]} ]} ]}.