Emit records from Go

Record an application's actions from its own process: register the catalogue at start-up, create the emitter, record.

Before you start

• You need a catalogue, a receiver in the application's namespace and the application's projected ServiceAccount token.

• The emitter is a Go module of its own: go get github.com/truvity/sluis/audit/sdk. It does not bring in the writer's database driver or object-store client (layout).

• A block action fails while the receiver is down. An async record queues and drops only when the queue overflows (recover from an outage).

The working code is examples/emit.

sequenceDiagram
  participant App as application
  participant R as receiver
  participant S3 as archive
  App->>R: RegisterCatalogue (at start-up)
  R-->>App: accepted, or refused with reasons
  Note over App: a request arrives
  App->>App: emitter.Record: fill, validate
  App->>R: Record (block: wait)
  R->>S3: one locked copy per profile
  R-->>App: durable
  Note over App: only now does the request complete

Steps

1. Register the catalogue

err := emit.Register(ctx, emit.Registration{
    URL:    receiverURL,
    Source: shop.Source, Version: shop.Version,
    Document: document,
    Schemas:  map[string][]byte{"https://schemas.example.com/shop/order-placed.json": schema},
    HTTP:     auth.TokenFile(os.Getenv("AUDIT_TOKEN_FILE")),
})

Do not start if Register returns emit.ErrCatalogueRefused. Retrying cannot fix a malformed document. You may retry an unreachable receiver. Registering the same version again is not an error.

Mount a projected token with the installation's audience, default audit, and point AUDIT_TOKEN_FILE at it. auth.TokenFile re-reads it on every request.

volumes:
  - name: audit-token
    projected:
      sources:
        - serviceAccountToken: { path: token, audience: audit, expirationSeconds: 3600 }

2. Create the emitter

emitter, err := emit.New(emit.Options{
    Source:    shop.Source,
    Catalogue: shop,
    Sink:      sink.NewClient(auth.TokenFile(tokenPath), receiverURL),
    Version:   "1.0.0",
    Instance:  os.Getenv("HOSTNAME"),
    Queue:     1024,              // how many async records may wait; this is the default
    Hooks: emit.Hooks{
        OnDropped: func(r *record.Record, reason string) { /* alert, and log it */ },
    },
})
defer emitter.Close()

Queue, Batch and Flush size the async path: 1024 records, 100 per send and one second by default. Wrap the hooks with emit.Instrument(hooks, otel.GetMeterProvider()) to count written, dropped and refused records.

Watch audit.emit.queue.pending: a climbing number means the receiver is slow or gone. Alert on audit.emit.records.dropped. The emitter also logs each drop through Options.Logger.

3. Record

err := emitter.Record(r.Context(), &record.Record{
    Action:    "shop.order.placed",
    Operation: auditv1.Operation_OPERATION_CREATE,
    TenantId:  "acme",
    Actor:     &record.Actor{Kind: "customer", Id: customerID},
    Targets:   []*record.Target{{Type: "order", Id: orderID}},
    Outcome:   &record.Outcome{Result: auditv1.Outcome_RESULT_SUCCESS},
    Data:      data,
})
if err != nil {
    // block: the action did not happen as far as anyone can prove.
}

The emitter fills the identifier, times, versions and sequence, and validates the record against the catalogue. Set TenantId to the customer organisation, or @platform for the application's own operations. Put identifiers in the record, never names or e-mail addresses. Record failures too: RESULT_FAILURE or RESULT_DENIED with a reason.

Wrap the HTTP handler in emit.Middleware(trustedHops) to add client address, user agent, request id and trace id. A wrong trustedHops records your load balancer as every actor's address.

4. Record from TypeScript

There is no TypeScript emitter. Call the receiver with the generated contract in ts/src/gen.

import { readFileSync } from "node:fs";
import { createClient } from "@connectrpc/connect";
import { createConnectTransport } from "@connectrpc/connect-node";
import { Delivery, SinkService } from "./gen/audit/v1/sink_pb";

const receiver = createClient(SinkService, createConnectTransport({
  baseUrl: process.env.AUDIT_RECEIVER!,
  httpVersion: "1.1",
  interceptors: [(next) => async (req) => {
    req.header.set("Authorization", `Bearer ${readFileSync(tokenFile, "utf8").trim()}`);
    return next(req);
  }],
}));
await receiver.write({ records: [record], delivery: Delivery.BLOCK });

Fill id (a UUIDv7), occurred_at, schema_version, catalogue_version and source yourself.

Verify

Perform an action, then find it with read the trail.

Decided in

0053, 0054.