Lesson 11 · The repo in practice

The activity-log firehose

Every request in the platform becomes an audit event — the one flow high-volume enough to justify the codebase's only pull consumer.

Your win: explain how every request is published as an audit event, and why this one flow uses a pull consumer that batches — the exception to "push everywhere."

The producers: every request, an event

The audit trail is generated automatically — two interceptors publish an ActivityLog.Created event for every request that flows through your services:

Producer	Where	How
gRPC interceptor	`internal/golibs/tracer/logger.go`	a deferred goroutine — fires even if the handler panics — `PublishAsyncContext(ActivityLog.Created)`
HTTP middleware	`internal/platform/activitylog/http.go`	tees the request/response, then `PublishAsyncContext(ActivityLog.Created)`

Both are wired into your services (conversationmgmt and notification each add the gRPC one). They use PublishAsyncContext — the fire-and-buffer variant (Lesson 6) — because there are thousands of these per second and blocking on each ack would cripple request latency. That's the firehose: a huge, one-way stream of audit events.

The consumer: the one pull consumer in the repo

Every other subscription in the codebase is a push queue-group consumer. The activity-log consumer (in zeus) is the single exception — a pull consumer:

internal/zeus/subscriptions/activity_log_subscription.go:50-68 (condensed)

opts := nats.Option{ JetStreamOptions: []nats.JSSubOption{
    nats.ManualAck(),
    nats.Bind(constants.StreamActivityLog, constants.DurableActivityLogCreatedPull),
    nats.MaxDeliver(10), nats.AckWait(30*time.Second), nats.DeliverNew(),
    nats.MaxAckPending(100000),                          // ← huge in-flight cap for a firehose
  }, PullOpt: nats.PullSubscribeOption{ FetchSize: 500, BatchSize: 10 }}   // ← fetch in batches
jsm.PullSubscribe(constants.SubjectActivityLogCreated, constants.DurableActivityLogCreatedPull,
                  s.BatchCreateActivityLog, opts)

Why pull, and only here A push consumer gets messages one at a time as they arrive — fine for the odd user-created event. But at firehose volume you want to batch: fetch 500 at once and do one bulk database insert instead of 500 tiny ones. A pull consumer gives you that control — the client fetches batches on its own schedule (Lesson 4). Hence the huge MaxAckPending(100000) and the FetchSize: 500. It's the textbook case for pull: high throughput, batch-friendly work, client-driven flow control.

Batch process, then batch ack

activity_log_subscription.go:70-88 (BatchCreateActivityLog, condensed)

_, err := s.partedBatchCreate(ctx, msgs)     // ONE bulk insert into the partitioned activity-log table
for _, msg := range msgs {
    msg.Ack()                                // then ack every message in the batch
}

It bulk-inserts the whole batch into a partitioned activity_log_parted table (using each message's Nats-Msg-Id as the row id), then acks them all. One DB round-trip for hundreds of events — the efficiency that pull + batch buys.

Even the firehose carries tenancy These audit events go through the same DataInMessage envelope (Lesson 6), so each carries its resource_path — the audit record is stamped with the tenant it belongs to, and the batch insert lands it in the right partition. The publish-envelope / consume-context machinery you learned in Part 2 applies uniformly, from a single user-created event to the whole audit firehose.

Pull consumers

Why and how a client fetches batches — the model behind the one pull consumer here.

→ docs.nats.io — Consumers (pull)
→ in-repo internal/zeus/subscriptions/activity_log_subscription.go, internal/golibs/tracer/logger.go

Check yourself (from memory)

Q1. How is the activity-log event produced?

gRPC + HTTP interceptors PublishAsyncContext an event per request — the gRPC one in a deferred goroutine so it fires even on panic.

Q2. Why is the activity-log consumer a pull consumer?

Pull lets the client fetch 500 at a time and batch-insert — far more efficient than push one-at-a-time for a high-volume stream.

Q3. After processing a batch, when does it ack?

Process first (one bulk insert), then ack all — so a failed insert leaves the whole batch unacked for redelivery.

Recall: the activity-log firehose + the pull consumer.

producers + why pull + batch, then reveal

Producers: two interceptors publish ActivityLog.Created for every request — gRPC (tracer/logger.go, a deferred goroutine so it fires even on panic) + HTTP (activitylog/http.go) — via PublishAsyncContext (fire-and-buffer, for throughput). Wired into your services. Consumer: zeus is the only pull consumer in the repo (activity_log_subscription.go): PullSubscribe(ActivityLog.Created, DurablePull),

PullOpt{FetchSize 500, BatchSize
    10}

, MaxAckPending 100000. Why pull: firehose volume → fetch batches → one bulk insert (into a partitioned table, keyed by Nats-Msg-Id) → then ack the whole batch. Pull = client-driven flow + batching. Even audit events carry the tenant via the DataInMessage envelope.

🎯 Interview one-liner "When would you use a pull consumer instead of push?" → "For high-volume, batch-friendly work. Our audit-log firehose publishes an event per request asynchronously; the one pull consumer fetches 500 at a time and does a single bulk insert, rather than a push consumer handling them one by one. Pull gives the client flow control and lets you amortize the write — everywhere else we use push queue-groups."

Last lesson: the big-picture comparison — NATS vs Kafka, the live migration, ops, and the whole-course recap. Ask me about push vs pull if the trade-off needs another pass.

1. In-repo: internal/zeus/subscriptions/activity_log_subscription.go:50-88, internal/golibs/tracer/logger.go, internal/platform/activitylog/http.go.