> ## Documentation Index
> Fetch the complete documentation index at: https://docs.kernite.io/llms.txt
> Use this file to discover all available pages before exploring further.

# Postgres Evidence Sink

> Persist governance decisions in Postgres alongside write transactions.

Kernite is intentionally stateless. Persist decision evidence in your app.

## Minimal Schema

```sql theme={null}
create table if not exists governance_events (
  id bigserial primary key,
  created_at timestamptz not null default now(),
  workspace_id text not null,
  principal_id text not null,
  object_type text not null,
  operation text not null,
  decision text not null,
  reason_codes jsonb not null,
  ctx_id text not null,
  trace_hash text not null,
  idempotency_key text,
  response_json jsonb not null
);

create unique index if not exists governance_events_workspace_idempotency_uq
  on governance_events (workspace_id, idempotency_key)
  where idempotency_key is not null;
```

## Transaction Placement

```python theme={null}
from __future__ import annotations

import json

from kernite import evaluate_execute


def create_invoice_with_governance(conn, execute_request: dict, invoice_payload: dict):
    governance = evaluate_execute(
        execute_request,
        idempotency_key=execute_request.get("idempotency_key"),
    )
    data = governance["data"]

    with conn:
        with conn.cursor() as cur:
            cur.execute(
                """
                insert into governance_events (
                    workspace_id,
                    principal_id,
                    object_type,
                    operation,
                    decision,
                    reason_codes,
                    ctx_id,
                    trace_hash,
                    idempotency_key,
                    response_json
                )
                values (%s, %s, %s, %s, %s, %s::jsonb, %s, %s, %s, %s::jsonb)
                """,
                (
                    execute_request["workspace_id"],
                    execute_request["principal"]["id"],
                    execute_request["object_type"],
                    execute_request["operation"],
                    data["decision"],
                    json.dumps(data.get("reason_codes", [])),
                    governance["ctx_id"],
                    data["trace_hash"],
                    data.get("idempotency_key"),
                    json.dumps(governance),
                ),
            )

            if data["decision"] != "approved":
                return {"ok": False, "governance": governance}

            cur.execute(
                "insert into invoices (workspace_id, amount, currency) values (%s, %s, %s) returning id",
                (
                    execute_request["workspace_id"],
                    invoice_payload["amount"],
                    invoice_payload["currency"],
                ),
            )
            invoice_id = cur.fetchone()[0]

    return {"ok": True, "id": invoice_id, "governance": governance}
```

## Integration Rules

* Store denied and approved events.
* Prefer same-transaction evidence + mutation for strict auditability.
* Redact sensitive payloads before persisting full response JSON.