Record and inspect business events

When to use this guide

Use these tasks to record a trusted application action or inspect audit events for an authorized tenant. Install Echo Pro first and load projection keys from your runtime secret manager. Run examples with:

go test -run Example ./audit/...

The audit overview explains event preparation, capacity and database restores. The reference lists configuration fields and defaults.

Record an event from a verified principal

Register the public amount metadata field in the schema. Read the user ID from authentication and verify the tenant restriction before constructing the event. This complete example tests the authenticated actor, target and recorded amount:

func ExampleSchema_businessEvent() {
  ctx := context.Background()
  material := make([]byte, 32)
  if _, err := rand.Read(material); err != nil {
    panic(err)
  }
  keys, err := redact.NewKeyring([]redact.Key{{ID: "v1", Material: material}})
  if err != nil {
    panic(err)
  }
  store, err := audit.NewMemoryStore(audit.MemoryStoreConfig{
    Capacity: audit.Capacity{Events: 100, Bytes: 1 << 20},
  })
  if err != nil {
    panic(err)
  }
  pepper := make([]byte, 32)
  if _, err := rand.Read(pepper); err != nil {
    panic(err)
  }
  ring := auth.PepperRing{{Version: 1, Key: pepper}}
  token, record, err := auth.IssueKey("demo", ring, "alice", auth.KeyOptions{
    Kind: auth.User, TenantRestriction: "acme", Scopes: []string{"credit:record"},
  })
  if err != nil {
    panic(err)
  }
  keyStore := &auth.MemoryKeyStore{Capacity: 10}
  if err := keyStore.Put(ctx, record); err != nil {
    panic(err)
  }
  authMW, err := (auth.Config{
    Schemes: map[string]auth.Scheme{"key": auth.APIKey{
      Namespace: "operators", Prefix: "demo", Peppers: ring, Store: keyStore,
    }},
    Fallback: auth.Require(map[string][]string{"key": {"credit:record"}}),
  }).ToMiddleware()
  if err != nil {
    panic(err)
  }
  e := echo.New()
  // BEGIN business event
  schema, err := audit.NewSchema(audit.SchemaConfig{
    Keys: keys, Fields: map[string]redact.Class{"amount": redact.Public},
  })
  if err != nil {
    panic(err)
  }
  e.Group("/v1", authMW).POST("/credits/record", func(c *echo.Context) error {
    p, ok := auth.FromEcho(c)
    if !ok {
      return echo.ErrUnauthorized
    }
    if p.Kind() != auth.User || p.TenantRestriction() != "acme" {
      return echo.ErrForbidden
    }
    // Server-confirmed grant to this tenant's customer; no client actor input.
    prepared, err := schema.Prepare(audit.Event{
      ID: "grant_1234567890a", Tenant: p.TenantRestriction(),
      Actor: audit.Actor{Kind: audit.User, ID: redact.Value{
        Class: redact.Identifier,
        Text:  p.ID(),
      }},
      Target: audit.Target{Kind: "customer", ID: redact.Value{
        Class: redact.Identifier,
        Text:  "customer_42",
      }},
      Action: "credit.granted", Outcome: audit.Succeeded,
      Metadata: map[string]redact.Value{"amount": {
        Class: redact.Public,
        Text:  "100",
      }},
    })
    if err != nil {
      return err
    }
    if _, err := store.Record(c.Request().Context(), prepared); err != nil {
      return err
    }
    return c.NoContent(http.StatusCreated)
  })
  // END business event
  req := httptest.NewRequestWithContext(ctx, http.MethodPost, "/v1/credits/record", nil)
  req.Header.Set("Authorization", "ApiKey "+token)
  res := httptest.NewRecorder()
  e.ServeHTTP(res, req)
  fmt.Println(res.Code)
  reader, err := store.Reader(audit.ReaderConfig{Schema: schema})
  if err != nil {
    panic(err)
  }
  entry, err := reader.Get(ctx, "acme", "grant_1234567890a")
  if err != nil {
    panic(err)
  }
  fmt.Println(
    entry.Actor.Kind,
    entry.Actor.ID,
    entry.Target.Kind,
    entry.Target.ID,
    entry.Metadata["amount"],
  )
  // Output:
  // 201
  // user operators:alice customer customer_42 100
}

For unrestricted users, resolve membership and read the selected tenant before preparation. Use authbind’s actor mapping so user and service principals get the right actor type. Generate the event ID once and call Prepare once. Retry Record with the same prepared event. Do not take actor ID, tenant authority or outcome from unchecked body fields.

Record intent before an external call

Prepare separate intent and outcome events with one OperationID. Record intent before calling the effect. If that record fails, return without performing the effect. The example store holds one event, so recording the outcome fails after the effect runs:

package auditpg_test

import (
  "context"
  "errors"
  "fmt"
  "time"

  "github.com/jackc/pgx/v5/pgxpool"

  "echo.labstack.com/pro/redact"

  "echo.labstack.com/pro/audit"
  "echo.labstack.com/pro/audit/auditpg"
)

func ExampleNew() {
  // Construction refuses a missing caller-owned pool before any I/O.
  _, err := openAuditStore(context.Background(), nil, nil)
  e, ok := errors.AsType[*audit.ConfigError](err)
  fmt.Println(ok, e.Field)
  // Output: true pool
}

func ExampleFork() {
  // Recovery needs no constructed Store; invalid evidence is rejected before I/O.
  err := auditpg.Fork(
    context.Background(),
    nil,
    auditpg.SchemaConfig{},
    auditpg.ForkRequest{
      Reason: "physical_restore",
    },
  )
  e, ok := errors.AsType[*audit.ValidationError](err)
  fmt.Println(ok, e.Field)
  // Output: true deployment evidence
}

func ExampleStore_Record() {
  // The same intent/outcome recipe accepts auditpg.Store as its Recorder.
  // A bounded MemoryStore makes the failure path executable without a database.
  ctx := context.Background()
  schema, err := audit.NewSchema(audit.SchemaConfig{Policy: redact.StorePolicy{
    AllowMaskFallback: true,
  }})
  if err != nil {
    fmt.Println(err)
    return
  }
  event := audit.Event{
    ID:          "intent_123456789",
    Actor:       audit.Actor{Kind: audit.System},
    Action:      "job.completed",
    Outcome:     audit.Unknown,
    Phase:       audit.Intent,
    OperationID: "operation_123456",
  }
  intent, err := schema.Prepare(event)
  if err != nil {
    fmt.Println(err)
    return
  }
  event.ID = "outcome_123456789"
  event.Phase = audit.OutcomePhase
  event.Outcome = audit.Succeeded
  outcome, err := schema.Prepare(event)
  if err != nil {
    fmt.Println(err)
    return
  }
  store, err := audit.NewMemoryStore(audit.MemoryStoreConfig{Capacity: audit.Capacity{
    Events: 1,
    Bytes:  16384,
  }})
  if err != nil {
    fmt.Println(err)
    return
  }
  err = recordEffect(
    ctx,
    store,
    intent,
    outcome,
    func() error { fmt.Println("effect performed"); return nil },
  )
  e, ok := errors.AsType[*audit.OutcomeUnrecordedError](err)
  fmt.Println(ok, errors.Is(e, audit.ErrCapacity))
  // Output:
  // effect performed
  // true true
}

func recordEffect(
  ctx context.Context,
  recorder audit.Recorder,
  intent, outcome audit.PreparedEvent,
  effect func() error,
) error {
  if _, err := recorder.Record(ctx, intent); err != nil {
    return err
  }
  if err := effect(); err != nil {
    return err
  }
  completion, cancel := context.WithTimeout(context.WithoutCancel(ctx), 2*time.Second)
  defer cancel()
  if _, err := recorder.Record(completion, outcome); err != nil {
    return &audit.OutcomeUnrecordedError{OperationID: "operation_123456", Cause: err}
  }
  return nil
}

// openAuditStore uses pre-provisioned roles and caller-owned pools connected
// to the same database. The deployment pool must be authorized to SET ROLE to
// the trusted NOLOGIN owner. The recorder pool uses the recorder runtime role.
func openAuditStore(ctx context.Context, deploymentPool, recorderPool *pgxpool.Pool) (
  *auditpg.Store,
  error,
) {
  cfg := auditpg.SchemaConfig{
    Schema: "business_audit",
    Roles: auditpg.Roles{
      Deployment:  "audit_deployment",
      Recorder:    "audit_recorder",
      Relay:       "audit_relay",
      Maintenance: "audit_maintenance",
      Reader:      "audit_reader",
    },
    Timeout: 5 * time.Second, MigrateTimeout: time.Minute,
  }
  if err := auditpg.Migrate(ctx, deploymentPool, cfg); err != nil {
    return nil, err
  }
  return auditpg.New(ctx, recorderPool, cfg)
}

It prints effect performed followed by true true, confirming *audit.OutcomeUnrecordedError and ErrCapacity. Keep the operation ID and check whether the effect happened. Retrying the effect would risk duplication. For external sends that need journaled uncertainty and provider records, add execution.

Record authentication and authorization refusals

Install denial middleware outside the guard. By default it records responses with status 401 or 403. Configure FailClosed only when the refusal policy requires an audit record before sending the denial response. This fixture records one 403:

func ExampleDenialConfig_ToMiddleware() {
  schema, err := audit.NewSchema(audit.SchemaConfig{Policy: redact.StorePolicy{
    AllowMaskFallback: true,
  }})
  if err != nil {
    panic(err)
  }
  store, err := audit.NewMemoryStore(audit.MemoryStoreConfig{Capacity: audit.Capacity{
    Events: 10,
    Bytes:  1 << 20,
  }})
  if err != nil {
    panic(err)
  }
  denials, err := (audit.DenialConfig{Schema: schema, Recorder: store}).ToMiddleware()
  if err != nil {
    panic(err)
  }
  e := echo.New()
  // Denial recording is outside this group's authorization middleware.
  group := e.Group("/private", denials, func(next echo.HandlerFunc) echo.HandlerFunc {
    return func(*echo.Context) error { return echo.NewHTTPError(403, "Forbidden") }
  })
  group.GET("/jobs", func(c *echo.Context) error { return c.NoContent(200) })
  ctx := context.TODO()
  w := httptest.NewRecorder()
  e.ServeHTTP(w, httptest.NewRequestWithContext(ctx, "GET", "/private/jobs", nil))
  events, err := store.TestEvents(ctx)
  if err != nil {
    panic(err)
  }
  fmt.Println(w.Code, len(events))
  // Output: 403 1
}

Denial callbacks must be safe for concurrent use and must honour cancellation. Recording uses its own context with a timeout, so a cancelled request does not stop it. Size MaxInFlight for your sink. When it is full, the event is not recorded. The request stays refused; FailClosed selects the recording-error response.

Query and export authorized events

Authorize the exact tenant before every reader call, including continuation requests. The query example writes one event and prints its action, outcome and empty continuation:

func ExampleReader_Query() {
  // Synthetic example key. Load production keys through your secret manager.
  keys, err := redact.NewKeyring([]redact.Key{{ID: "k1", Material: []byte(
    "01234567890123456789012345678901",
  )}})
  if err != nil {
    panic(err)
  }
  schema, err := audit.NewSchema(audit.SchemaConfig{Keys: keys})
  if err != nil {
    panic(err)
  }
  store, err := audit.NewMemoryStore(audit.MemoryStoreConfig{Capacity: audit.Capacity{
    Events: 10,
    Bytes:  1 << 20,
  }})
  if err != nil {
    panic(err)
  }
  ctx := context.TODO()
  p, err := schema.Prepare(audit.Event{ID: "example_1234567890", OccurredAt: time.Date(
    2026,
    1,
    1,
    0,
    0,
    0,
    0,
    time.UTC,
  ), Tenant: "tenant1", Actor: audit.Actor{
    Kind: audit.System,
  }, Action: "job.completed", Outcome: audit.Succeeded})
  if err != nil {
    panic(err)
  }
  if _, err := store.Record(ctx, p); err != nil {
    panic(err)
  }
  r, err := store.Reader(audit.ReaderConfig{Schema: schema})
  if err != nil {
    panic(err)
  }
  // Authorize tenant1 before every call, including calls with Page.Next.
  page, err := r.Query(ctx, audit.QueryRequest{Filter: audit.Filter{Tenant: "tenant1"}})
  if err != nil {
    panic(err)
  }
  entry, err := r.Get(ctx, "tenant1", page.Entries[0].ID)
  if err != nil {
    panic(err)
  }
  fmt.Println(entry.Action, entry.Outcome, page.Next == "")
  // Output: job.completed succeeded true
}

Resend the same filter with Page.Next. Treat search probe text as confidential. For a bulk export, provide a context with the desired end-to-end deadline and a writer owned by the caller:

func ExampleReader_Export() {
  // Synthetic example key; production keys come from a secret manager.
  keys, err := redact.NewKeyring([]redact.Key{{ID: "k1", Material: []byte(
    "01234567890123456789012345678901",
  )}})
  if err != nil {
    panic(err)
  }
  schema, err := audit.NewSchema(audit.SchemaConfig{Keys: keys})
  if err != nil {
    panic(err)
  }
  store, err := audit.NewMemoryStore(audit.MemoryStoreConfig{Capacity: audit.Capacity{
    Events: 10,
    Bytes:  1 << 20,
  }})
  if err != nil {
    panic(err)
  }
  // PostgreSQL deployments use separate reader-role and recorder-role stores.
  r, err := store.Reader(audit.ReaderConfig{Schema: schema, Access: store})
  if err != nil {
    panic(err)
  }
  var output bytes.Buffer
  // Authorize this tenant and the representation before calling Export.
  result, err := r.Export(context.TODO(), audit.ExportRequest{Filter: audit.Filter{
    Tenant: "tenant1",
    Action: "job.",
  }, Actor: audit.Actor{Kind: audit.System}, Reason: "investigation"}, &output)
  if err != nil {
    panic(err)
  }
  lines := strings.Split(strings.TrimSpace(output.String()), "\n")
  var trailer struct {
    Schema    string `json:"schema"`
    Integrity string `json:"integrity"`
  }
  if err := json.Unmarshal([]byte(lines[len(lines)-1]), &trailer); err != nil {
    panic(err)
  }
  fmt.Printf(
    "%s: %d events, integrity=%s\n",
    trailer.Schema,
    result.Events,
    trailer.Integrity,
  )
  // Output:
  // pro.audit/v1/export-trailer: 0 events, integrity=unauthenticated
}

The example’s trailer says integrity=unauthenticated. It does not claim a signature or tamper-proof history. Configure an access recorder when export policy needs an audit record of the read itself.

Deploy PostgreSQL audit storage

Provision the deployment NOLOGIN owner and the independent recorder, relay, maintenance and reader roles through your database administration process. Use auditpg.Migrate(ctx, deploymentPool, cfg) before constructing runtime stores. Supply the same schema and role configuration to New. It checks role grants, the cluster ID and stored transaction progress. Your application still owns and closes the pool.

This compiled helper takes deployment and recorder pools connected to the same database. The roles must already exist. It applies the migration through the deployment connection and returns a recorder store through the runtime connection. It returns the original error from either call:

// openAuditStore uses pre-provisioned roles and caller-owned pools connected
// to the same database. The deployment pool must be authorized to SET ROLE to
// the trusted NOLOGIN owner. The recorder pool uses the recorder runtime role.
func openAuditStore(ctx context.Context, deploymentPool, recorderPool *pgxpool.Pool) (
  *auditpg.Store,
  error,
) {
  cfg := auditpg.SchemaConfig{
    Schema: "business_audit",
    Roles: auditpg.Roles{
      Deployment:  "audit_deployment",
      Recorder:    "audit_recorder",
      Relay:       "audit_relay",
      Maintenance: "audit_maintenance",
      Reader:      "audit_reader",
    },
    Timeout: 5 * time.Second, MigrateTimeout: time.Minute,
  }
  if err := auditpg.Migrate(ctx, deploymentPool, cfg); err != nil {
    return nil, err
  }
  return auditpg.New(ctx, recorderPool, cfg)
}

Replace the schema and role names with your provisioned names. Keep separate pools for relay, maintenance and authorized reads. Close them only after their application-owned work stops. This helper compiles with the package examples; it does not run PostgreSQL without the role topology. The real backend suite needs those independently assigned roles.

Use READ COMMITTED for recorder transactions. Keep projection keys for the retained events you need to search. Run retention and relay work through bounded application-owned loops. After a restore, stop writes and follow the database restore procedure before Rebind or Fork. Reconnecting to the restored database does not make lost records safe to ignore.

The full console demo reports when these role requirements are unavailable and uses memory audit instead. Memory audit is fine for UI tests. It loses all records when the process exits. It does not test PostgreSQL recording or restore behavior.

Expose a reader in the console

Build a reader with its access recorder and pass it to auditadmin.NewPanel. Showing unredacted values needs a separate reveal grant. The console checks it on every request. This executable construction prints the panel ID and view/action counts:

func ExampleNewPanel() {
  // Synthetic example key; obtain production keys from a secret manager.
  keys, err := redact.NewKeyring([]redact.Key{{ID: "k1", Material: []byte(
    "01234567890123456789012345678901",
  )}})
  if err != nil {
    panic(err)
  }
  schema, err := audit.NewSchema(audit.SchemaConfig{Keys: keys})
  if err != nil {
    panic(err)
  }
  store, err := audit.NewMemoryStore(audit.MemoryStoreConfig{Capacity: audit.Capacity{
    Events: 100,
    Bytes:  1 << 20,
  }})
  if err != nil {
    panic(err)
  }
  reader, err := store.Reader(audit.ReaderConfig{Schema: schema, Access: store})
  if err != nil {
    panic(err)
  }
  panel, err := auditadmin.NewPanel(auditadmin.PanelConfig{
    Reader:       reader,
    EnableReveal: true,
  })
  if err != nil {
    panic(err)
  }
  // Mount panel in admin.Config.Panels with Keys and authenticated principals.
  // Reveal requires the principal's separate Reveal grant on every request.
  descriptor := panel.AdminPanel().Describe()
  fmt.Println(descriptor.ID, len(descriptor.Views), len(descriptor.Actions))
  // Output: audit 3 0
}

Mount it through the admin console. A missing audit write for a console mutation returns admin audit unavailable before the mutation. If the outcome record is missing, check both the business state and the audit sink.