Lesson 7 · Publishing, consuming & delivery · the flagship

Subscribing & the handler

Between the wire and your code sits one function — and it does the cleverest thing in the whole layer: it makes an async event feel exactly like a synchronous, tenant-scoped request.

Your win: trace a message through handleMsg step by step, and explain how tenancy and tracing are rebuilt on the consumer so RLS survives the async hop.

Your handler, and the wrapper around it

Your subscriber gives the golib a callback with this signature:

internal/golibs/nats/jetstream.go:271
type MsgHandler func(ctx context.Context, data []byte) (retry bool, err error)

You get a ctx and the payload bytes; you return an error and a retry bool that decides what happens on failure (Lesson 8). But you never see the raw NATS message — the golib's handleMsg wraps your callback and does the plumbing first.

handleMsg — the seven steps

This is the heart of every subscription. Read it top to bottom; each numbered step is a real thing the golib does before (and after) your handler runs:

internal/golibs/nats/jetstream.go:663-740 (condensed)
ctx, cancel := context.WithTimeout(context.Background(), 360*time.Second)   // ① a fresh ctx (no request ctx exists)
msgID := msg.Header.Get("Nats-Msg-Id")                                // ② for logging + metrics (deferred)

if skipMsgOlderThan != nil && msg too old { return }                     // ③ stale-message guard (DENY_CONSUME_OLD_MESSAGE)

var dataInMsg npb.DataInMessage
if err := proto.Unmarshal(msg.Data, &dataInMsg); err != nil {
    msg.Ack(); return                                                   // ④ can't parse → ACK (drop the poison)
}

if dataInMsg.TraceInfo != nil {
    ctx = ContextWithTraceInfo(ctx, dataInMsg.TraceInfo)                    // ⑤a rehydrate the trace…
    ctx, span = interceptors.StartSpan(ctx, spanName, SpanKindConsumer)     //    …and start a consumer span
}

ctx = interceptors.ContextWithJWTClaims(ctx, &CustomClaims{                 // ⑤b ← THE FLAGSHIP
    Manabie: &ManabieClaims{ ResourcePath: dataInMsg.ResourcePath, UserID: dataInMsg.UserId }})

retry, err := cb(ctx, dataInMsg.Payload)                                    // ⑥ YOUR handler runs, with a full ctx

if err != nil { if !retry { msg.Ack() }; return }                         // ⑦ ack/retry/drop (Lesson 8)
msg.Ack()

⑤b is the whole point: tenancy survives the hop

Close the loop with the Auth course In Lesson 6, publishing packed the tenant (resource_path) into the envelope. Here, handleMsg takes it back out and calls ContextWithJWTClaims — the exact same function the auth interceptor uses on a synchronous request (Auth course, Lesson 4). So by the time your handler runs, ResourcePathFromContext(ctx) returns the right tenant, and any DB query the handler makes gets the correct set_config('permission.resource_path') — Row-Level Security works exactly as it would on a live request.

This is the single cleverest thing in the repo's NATS layer. An event is processed seconds or minutes after the request that caused it, on a different pod, with no request context of its own — yet tenant isolation and distributed tracing both hold, because they were smuggled through the message and re-injected here. The publish/consume symmetry (envelope in → context out) is what makes async eventing safe in a multi-tenant system.

The three subscribe entry points Your subscribers reach handleMsg through one of three: Subscribe (push, no queue group), QueueSubscribe (push queue-group — the common path, Lesson 4), and PullSubscribe (pull batches — only the activity-log consumer, Lesson 11). All force AckExplicit; the queue/pull variants provision the consumer (UpsertConsumer) first. Different subscription shapes, same per-message plumbing.
Read this next

Consuming & acknowledging

How a subscriber receives and acks messages — the mechanics handleMsg wraps.

→ docs.nats.io — consuming messages
→ in-repo internal/golibs/nats/jetstream.go:663-740 · the Auth course's ContextWithJWTClaims (RLS)

Check yourself (from memory)

Q1. How does a NATS subscriber's DB query stay tenant-scoped (RLS)?

The tenant rode in the envelope; handleMsg rebuilds the JWT claims on ctx, so RLS works just like a synchronous request.

Q2. What does the MsgHandler's retry return value control?

On error: retry=true → don't ack → redeliver; retry=false → ack → drop. (Full logic in Lesson 8.)

Q3. What happens if handleMsg can't unmarshal the envelope?

A message it can never parse is poison — it acks to drop it (no infinite redelivery) and tags a PROTO_UNMARSHAL_ERROR metric.
Recall: handleMsg — the seven steps + the flagship.
the steps + tenancy-survives, then reveal
Handler: func(ctx, []byte) (retry bool, err error). handleMsg wraps it: ① fresh ctx (360s timeout) → ② read Nats-Msg-Id + defer metric → ③ SkipMsgOlderThan? skip → ④ proto.Unmarshal DataInMessage (fail → Ack, drop poison) → ⑤a rehydrate trace from TraceInfo → consumer span → ⑤b re-inject tenancy: ctx = ContextWithJWTClaims(ResourcePath, UserID) (the same fn the auth interceptor uses → RLS survives the async hop) → ⑥ cb(ctx, payload) → ⑦ success Ack; err+retry no-ack (redeliver); err+!retry Ack (drop). Three entries: Subscribe/QueueSubscribe (push)/PullSubscribe (pull, activity-log only) — all force AckExplicit. Publish-in-envelope + consume-out-to-ctx = the symmetry that makes async multi-tenant eventing safe.
🎯 Interview one-liner "How do tenant isolation and tracing survive an asynchronous event?" → "We pack the resource_path and the trace context into the message envelope at publish time. On the consumer, before the handler runs, we unpack them and re-inject them into the context using the same claims function the synchronous auth interceptor uses — so Row-Level Security and distributed tracing behave exactly as they would on a live request, even though the event is processed later, on another pod."
Last lesson of Part 2: delivery semantics — the ack model, redelivery, dedup, and the honest gaps (no Nak/Term, no DLQ). Ask me to walk the seven steps again — this is the lesson worth over-learning.

1. In-repo: internal/golibs/nats/jetstream.go:271,663-740. docs.nats.io — consuming.