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.

Backend use case Backend use case: this is the cross-cutting example that proves not every NATS path in the repo looks like the learner-owned push subscribers.

Common mistake Common mistake: assuming the activity-log path is just another queue-group subscriber when it is the one pull batch path.

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

In plain English Plain English: the activity log path is the repo’s “everything that happened” firehose, and it uses the one pull consumer because batch control matters there.

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.