Lesson 9 · The repo in practice · your service

conversationmgmt subscribers

Your service, reacting to the rest of the platform — three subscribers that keep the chat vendor in sync with users, staff, and groups it never talks to directly.

Your win: read conversationmgmt's real subscribers end to end — how they register, what they consume, and why their config differs — using everything from Parts 1–2.

Registration: where the subscribers wire up

Recall RegisterNatsSubscribers (Lesson 5) — conversationmgmt's is two registrars, each starting a family of subscribers:

cmd/server/conversationmgmt/gserver.go:168-179 (condensed)
agora := agora_usermgmt_nats.NewSubscribersRegistered(rsc.DBWith("tom"), rsc.NATS(), …)
agora.StartSubscribeForAllSubscribers()          // User.Created, Staff.Upserted
convo := convo_nats.NewSubscribersRegistered(rsc.DBWith("tom"), rsc.NATS(), …)
convo.StartSubscribeForAllSubscribers()          // UserGroup.Upserted, Staff.Upserted
conversationmgmt is subscriber-only Notice it takes rsc.NATS() and rsc.DBWith("tom") but declares no streams — because it doesn't publish domain events into JetStream (it only publishes Notification.Created to another service's stream, Lesson 10). It consumes streams that usermgmt owns (user, staff, usergroup). This is the fan-out model from Lesson 1: usermgmt publishes a domain event once; conversationmgmt (and others) react. Your service is a reactor.

A subscriber is a struct + StartSubscribe + a handler

Every subscriber follows the same tiny shape. Here's the one that reacts to a new user:

internal/conversationmgmt/modules/agora_usermgmt/controller/nats/user_created_subscriber.go:28-44
func (s *UserCreatedSubscriber) StartSubscribe() error {
    opts := nats.Option{ JetStreamOptions: []nats.JSSubOption{
        nats.ManualAck(),
        nats.Bind(constants.StreamUser, constants.DurableConversationMgmtUserCreated),  // usermgmt's stream
        nats.DeliverSubject(constants.DeliverConversationMgmtUserCreated),
        nats.MaxDeliver(3), nats.AckWait(30*time.Second), nats.MaxAckPending(30),
    }}
    return s.nats.QueueSubscribe(constants.SubjectUserCreated,        // "User.Created"
        constants.QueueConversationMgmtUserCreated, opts, s.Handle)  // s.Handle = create the user in Agora chat
}

The Handle callback is your MsgHandler (Lesson 7): when usermgmt creates a user, conversationmgmt gets the event and creates the matching user in the Agora chat vendor — so the chat system stays in sync with a service it has no direct call to.

Three subscribers, three tunings

The three conversationmgmt subscribers are the same shape with config tuned to the work:

SubscriberSubject / StreamTuning
UserCreatedUser.Created / userMaxDeliver 3, AckWait 30s
StaffUpsertedStaff.Upserted / staffMaxDeliver 10, DeliverNew, SkipMsgOlderThan aWeekAgo
UserGroupUpdateUserGroup.Upserted / usergroupMaxDeliver 10, AckWait 300s, DeliverNew, aWeekAgo

The UserGroupUpdate handler does more work (reconciling group membership), so it gets a longer AckWait (300s) and more retries (10). Both group/staff subscribers use DeliverNew (don't replay history) and SkipMsgOlderThan aWeekAgo (Lesson 8 — ignore events older than a week, so a redeploy after downtime doesn't reprocess a flood). Same machinery, tuned per workload.

The migration, in your own code Open conversation/controller/nats/subscribers_registered.go and you'll find two StudentCourseClassSynced subscribers commented out, with a note that they moved to Kafka (Lesson 12). So your own service is a live example of the NATS→Kafka migration: some events it used to consume over NATS now arrive over Kafka, with the domain handler reused. When you see a commented-out subscriber, that's what happened.
Read this next

Your own subscribers

Open the real files — they're short, and they're the template for anything you'll add.

→ in-repo internal/conversationmgmt/modules/agora_usermgmt/controller/nats/ + …/conversation/controller/nats/
→ docs.nats.io — Consumers (the config you're reading)

Check yourself (from memory)

Q1. Why does conversationmgmt declare no NATS streams?

Fan-out model: usermgmt publishes domain events into its streams; conversationmgmt reacts by consuming them. It owns no streams.

Q2. Why does UserGroupUpdate use a 300s AckWait vs UserCreated's 30s?

AckWait is tuned to the work: a slower handler needs a longer window before the server assumes failure and redelivers.

Q3. A subscriber in this repo is made of…

Struct + StartSubscribe (the QueueSubscribe with tuned options) + a MsgHandler. Every subscriber is a variation on this.
Recall: conversationmgmt's subscribers.
registration + shape + the three, then reveal
Registration: RegisterNatsSubscribers (gserver.go:168-179) → two registrars (agora + conversation), each StartSubscribeForAllSubscribers. Subscriber-only — declares no streams; consumes usermgmt's user/staff/usergroup streams (fan-out: usermgmt publishes, convo reacts). Shape: struct + StartSubscribe (QueueSubscribe(subject, queue, opts, Handle)) + a handler. Three: UserCreated (User.Created; MaxDeliver 3, AckWait 30s; creates the Agora chat user) · StaffUpserted (Staff.Upserted; MaxDeliver 10, DeliverNew, SkipMsgOlderThan aWeekAgo) · UserGroupUpdate (UserGroup.Upserted; MaxDeliver 10, AckWait 300s, DeliverNew, aWeekAgo — heavier handler → longer window). Some subs (StudentCourseClassSynced) are commented out — migrated to Kafka.
🎯 Interview one-liner "How does your service react to events from other services?" → "NATS JetStream subscribers. conversationmgmt is subscriber-only — it consumes usermgmt's user/staff/usergroup streams and keeps the chat vendor in sync. Each subscriber is a durable push queue-group consumer with config tuned to the work: heavier handlers get longer ack-waits and more retries, and use DeliverNew plus a stale-message skip so a redeploy doesn't reprocess history."
Next: your other service — notification — and the NATS link that connects the two of them. Ask me to walk any of the three subscribers in full.

1. In-repo: …/agora_usermgmt/controller/nats/user_created_subscriber.go:28-44, …/conversation/controller/nats/subscribers_registered.go, cmd/server/conversationmgmt/gserver.go:168-179.