# Repo map — NATS JetStream (ground truth)

The survey behind this course. Every lesson's `file:line` anchors resolve here. When code moves, fix
this file first, then the lessons. Grouped by the flow: connect → streams → publish → consume.

**Orientation:** the repo runs **both** NATS JetStream and Kafka. The shared JetStream client lives in
`internal/golibs/nats/`; streams are provisioned by the **fink** job; all subject/stream/consumer
names are centralized in `internal/golibs/constants/common.go`. The two learner-owned services are
`cmd/server/{conversationmgmt,notificationmgmt}/gserver.go`.
**Correction up front:** `chunk_handler.go` is NOT payload chunking — it's a generic concurrent
batch-worker utility that happens to live in the `nats` package (§8).

---

## 1. The golib — `internal/golibs/nats/jetstream.go` (840 lines)

- **`JetStreamManagement` interface** (`:67-83`): `ConnectToJS`, `GetJS`, `UpsertStream`,
  `UpsertConsumer`, `PublishAsyncContext`, `PublishContext`, `TracedPublishAsync`, `TracedPublish`,
  `Subscribe`, `QueueSubscribe`, `PullSubscribe`, `Close`, `RegisterReconnectHandler`,
  `RegisterDisconnectErrHandler`. Impl `jetStreamManagementImpl` (`:85-102`). Constructor
  `NewJetStreamManagement(url,user,pw,maxReconnect,reconnectWait,isLocal,logger)` (`:273-301`) — does
  NOT connect yet.
- **Connect/reconnect:** `ConnectToJS()` (`:325-350`) — `nats.Connect` with UserInfo, ReconnectWait,
  MaxReconnects, disconnect/reconnect handlers, 5s timeout; then `conn.JetStream(PublishAsyncMaxPending(200000))`.
  `Close()` (`:819-834`) drains each tracked sub, then `conn.Close()`.
- **`UpsertStream`** (`:360-403`): idempotent hand-rolled upsert. `isLocal`→Replicas=1; retention
  differs → **delete+recreate**; subjects/replicas/maxage/maxbytes differ → `UpdateStream`; not found
  → `AddStream`. **`UpsertConsumer`** (`:405-442`): not found → `AddConsumer`; drift on any of
  MaxDeliver/AckWait/DeliverGroup/OptStartTime/AckPolicy/DeliverSubject/FilterSubject/DeliverPolicy/
  OptStartSeq/MaxAckPending → **delete+recreate** (logs `cmp.Diff`).
- **Publish** (all wrap `npb.DataInMessage{Payload, ResourcePath, UserId}` + `proto.Marshal`, and set
  `nats.MsgId(idutil.ULIDNow())` for dedup): `PublishAsyncContext` (`:444-465`), `TracedPublishAsync`
  (`:467-505`, + OTel producer span + B3 `TraceInfo` injected into the envelope), `PublishContext`
  (`:507-528`, sync), `TracedPublish` (`:530-554`, sync + span).
- **Subscribe** (all prepend `AckExplicit()`; local adds short `AckWait`): `Subscribe` (`:556-581`,
  push, no queue, binds), `QueueSubscribe` (`:583-623`, **push queue-group — the common path**: sets
  `FilterSubject`+`DeliverGroup`, calls `UpsertConsumer` with `try.Do` 5×, then `js.QueueSubscribe`),
  `PullSubscribe` (`:754-817`, the only pull path — used by zeus activity-log; loops `sub.Fetch`).
- **Subscribe options** (`JSSubOption` functional options, `:142-268`): `Bind`, `ManualAck`,
  `MaxDeliver`, `MaxAckPending`, `FitlerSubject`(sic), `AckWait`, `AckAll`, `AckExplicit`, `StartTime`,
  `DeliverNew`, `DeliverLast`, `StartSequence`, `Durable` (rejects names with `.`), `DeliverSubject`.
- **`MsgHandler`** (`:271`): `func(ctx, data []byte) (retry bool, err error)`.

### `handleMsg` — the consumer heart (`:663-740`) — the flagship
- 360s ctx timeout; reads stream sequence + `Nats-Msg-Id`; defers `processMsgCounter` metric.
- `SkipMsgOlderThan` → skip+return if the message is older than the threshold (`DENY_CONSUME_OLD_MESSAGE`).
- `proto.Unmarshal` into `DataInMessage`; **on failure → `msg.Ack()` (drop) + `PROTO_UNMARSHAL_ERROR`**.
- If `TraceInfo` → rehydrate trace ctx + start a consumer span.
- **Re-inject tenancy** (`:711-717`): `ContextWithJWTClaims(ctx, CustomClaims{Manabie:{ResourcePath,
  UserID}})` → **RLS + identity survive the async hop** (the repo's cleverest bit).
- `cb(ctx, payload)` → on `err && !retry` → `msg.Ack()` (drop poison); on `err && retry` → **do
  nothing** → redelivered after `AckWait`; on success → `msg.Ack()`. **No `Nak()`/`Term()`.**

### Other golib files
- `metrics.go` — Prometheus `manabie_app_nats_disconnect` gauge (via reconnect/disconnect handlers).
  The **OpenCensus views** (`jetstream.go:30-56`) `JetstreamProcessedMessagesView` + latency; registered
  in `bootstrap/monitor.go:116-121`.
- `marshal.go` — an alternative `npb.WrapperMsg` (anypb) with B3 trace; distinct from the live
  `DataInMessage` path. `trace.go` — `B3Carrier` (proto ↔ b3 headers). `config.go` — `Confer`.

---

## 2. Streams & consumers — configuration

- **Streams provisioned by `fink`:** `cmd/server/fink/upsert_streams_nats.go` — `init()` registers
  job `"upsert_streams"` (`:19`); `upsertStreams` (`:31-67`) concatenates per-service configs and calls
  `jsm.UpsertStream(sc)`. Per-service defs `cmd/server/fink/streams/*.go` (one per owning service).
- **Representative:** `streams/notificationmgmt.go:9-21` — `StreamNotification` ("notification"),
  `InterestPolicy`, subjects `["Notification.*"]`, `Replicas: 3`, `MaxBytes: 4 GB`. `streams/tom.go`
  (chat streams, `chat.chat.>`), `streams/usermgmt.go` (`User.*`/`Staff.*`/`UserGroup.*`), etc.
- **Almost all streams:** retention `InterestPolicy`; storage default (file, per `nats.conf`);
  `Replicas: 3` (prod), forced to 1 locally.
- **conversationmgmt owns NO NATS streams** — it's subscriber-only (consumes usermgmt's `StreamStaff`/
  `StreamUserGroup`/`StreamUser`; publishes to `Notification.Created`).
- **Consumers** are created at subscribe time by `QueueSubscribe`→`UpsertConsumer` (durable push
  queue-group). Naming triad per subscription: `Durable*` + `Queue*` + `Deliver*`.

## 3. Bootstrap integration

- `internal/golibs/bootstrap/nats.go`: `NatsServicer[T]` (`:15-19`) — services opt in via
  `RegisterNatsSubscribers(ctx, cfg, *Resources) error`. `initnats` (`:21-35`) reads config → `rsc.WithNATSC`.
- `internal/golibs/bootstrap/resource.go`: `NATS()` (`:455`, panics if unavailable) / `GetNATS()`
  (`:464-481`, **lazy** — builds the client under `natsjsOnce`, then `ConnectToJS()`). So the connection
  is opened on first `rsc.NATS()`.
- `monitor.go:46-53` — the bootstrapper registers the NATS OpenCensus views iff the service is a `NatsServicer`.

## 4. Learner-owned subscribers (the anchor)

### conversationmgmt (`gserver.go:168-179`)
- Two registrars (`agora_usermgmt` + `conversation`), both with `rsc.DBWith("tom")` + `rsc.NATS()`.
- `UserCreatedSubscriber` (`agora_usermgmt/controller/nats/user_created_subscriber.go:28-45`):
  `QueueSubscribe(SubjectUserCreated="User.Created", QueueConversationMgmtUserCreated)`, `StreamUser`/
  `DurableConversationMgmtUserCreated`, `DeliverSubject`, `MaxDeliver(3)`, `AckWait(30s)`, `MaxAckPending(30)`,
  `ManualAck`. Handler creates the user in the Agora chat vendor.
- `UserGroupUpdateSubscriber` (`conversation/controller/nats/user_group_update_subscriber.go:20-43`):
  `UserGroup.Upserted` on `StreamUserGroup`, `MaxDeliver(10)`, `AckWait(300s)`, `DeliverNew()`,
  `SkipMsgOlderThan: aWeekAgo`.
- `StaffUpsertedSubscriber` (`conversation/controller/nats/staff_upserted_subscriber.go:22-45`):
  `Staff.Upserted` on `StreamStaff`, `MaxDeliver(10)`, `AckWait(30s)`, `DeliverNew()`, `aWeekAgo`.
- Two `StudentCourseClassSynced` subscribers are **commented out** (migrated to Kafka) —
  `subscribers_registered.go:41-74`.
- **Publisher:** `internal/conversationmgmt/modules/conversation/infrastructure/nats/notification.go:32-89`
  — `PushNotificationForConversation` builds `ypb.NatsCreateNotificationRequest`, `TracedPublish(ctx,
  "Conversation.PushNotificaton", SubjectNotificationCreated, msg)` + `try.Do` 5×. The convo→notification link.

### notificationmgmt (`gserver.go:89`)
- `initEventPushNotification` + `initEventSyncJprepStudentClass`; student-package v2 subs commented out
  (Kafka migration, `:109-111`).
- Core subscriber `internal/notification/transports/nats/push_notification.go`: `StartSubscribe` (`:30-47`)
  `QueueSubscribe(SubjectNotificationCreated="Notification.Created", QueueNotification)`, `StreamNotification`/
  `DurableNotification`, `DeliverSubject(DeliverNotification)`, `MaxDeliver(3)`, `AckWait(90s)`,
  `MaxAckPending(30)`, `ManualAck`. `HandleMessage` (`:50-82`) unmarshals `NatsCreateNotificationRequest`,
  dispatches on `SendingMethods`, returns `(retryable, err)`.
- Transport is **protocol-only**; business logic is the transport-agnostic `internal/notification/subscribers/`
  (`NotificationSubscriber`) reused by BOTH NATS and Kafka.

### Names (`internal/golibs/constants/common.go`, ~320 consts)
- Family per event: `Subject<Name>` (dotted PascalCase, `"Staff.Upserted"`), `Stream<Name>` (lowercase,
  `"staff"`), `Queue<Name>`, `Durable<Name>`, `Deliver<Name>`. Streams use wildcards (`User.*`, `chat.chat.>`);
  consumers filter to concrete subjects. `NatsJSMaxRedeliveryTimes=10` (`:12-15`). Kafka topics live in
  the same file (`:555-650`, `{team}.{topic}`) — a teaching contrast.

## 5. Activity-log-over-NATS (the cross-cutting example)

Two producers publish `ActivityLog.Created` / `StreamActivityLog`:
- **HTTP interceptor** `internal/platform/activitylog/http.go` — `Middleware` tees bodies →
  `ProcessMessage` builds `npb.ActivityLogEvtCreated` → `jsm.PublishAsyncContext(SubjectActivityLogCreated)`.
- **gRPC interceptor** `internal/golibs/tracer/logger.go` — `UnaryActivityLogRequestInterceptor(jsm, logger)`
  in a deferred goroutine (fires even on panic) → `PublishAsyncContext`. Wired into both learner services.
- **Consumer** `internal/zeus/subscriptions/activity_log_subscription.go` — the **only pull consumer**:
  `PullSubscribe(SubjectActivityLogCreated, DurableActivityLogCreatedPull, BatchCreateActivityLog)` with
  `MaxDeliver(10)`, `AckWait(30s)`, `DeliverNew()`, `MaxAckPending(100000)`, `PullOpt{FetchSize:500, BatchSize:10}`.

## 6. Delivery semantics & reliability

- **At-least-once** via forced `AckExplicit()` + caller `ManualAck()`.
- **Ack/retry/drop** (`handleMsg`): success→`Ack`; `err+retry=true`→no-ack→redeliver after `AckWait`;
  `err+retry=false`→`Ack` (drop poison); unmarshal fail→`Ack` (drop). **No `Nak()`/`Term()`.**
- **Redelivery cap** `MaxDeliver` (3 or 10); after it, the server drops — **no DLQ** (only metrics/logs).
- **Dedup:** every publish stamps `Nats-Msg-Id = ULID` → JetStream dedup window (default 2 min). Plus
  app-level idempotency (notification `IdempotentService`).
- **Stale-message guard** `SkipMsgOlderThan` (`aWeekAgo`) — skip+ack messages older than a threshold.
- **`MaxAckPending`** throttles in-flight unacked (30 push-notification, 5000 student-package, 100000 activity-log).

## 7. NATS vs Kafka in this repo

- **NATS JetStream** = internal service-to-service domain events / fan-out: user/staff/usergroup upserts,
  chat/lesson events, push-notification requests, the activity-log firehose. Dotted-PascalCase subjects,
  interest retention, 3 replicas.
- **Kafka** = durable streams / CDC / high-throughput pipelines: Debezium CDC (`docs/cdc/`), the outbox
  pattern, email, bulk jobs, DWH sync. `{team}.{topic}` topics.
- **Active NATS→Kafka migration** — several NATS subscribers commented out; gated by Unleash
  `Communication_Notification_Use_Kafka_Instead_Of_NATs`; the same domain handler reused by both transports.

## 8. Ops & tests

- **Server:** local `nats:2.8.4-alpine3.15` (`local/docker-compose.infra.yaml:126-140`); `nats.conf`
  `jetstream{store_dir:/data/jetstream; max_mem:4GiB; max_file:10GiB}`, client `:4223`, monitoring `:8223`,
  **per-service publish/subscribe ACLs** (each service may only publish its own subjects; `Fink` unrestricted).
  Helm `deployments/helm/platforms/nats-jetstream/` (statefulset, fileStorage 10Gi, 3 replicas, cluster routes).
- **Tests:** mock `mock/golibs/nats/jet_stream_management.go` (mockery); publishers/subscribers unit-tested
  against the mock. Golib's own `jetstream_test.go` = option-mapping tests. BDD exercises real NATS.

## Notable / bespoke (the "not vanilla" points)
- **Tenancy + tracing smuggled through the proto envelope** (`DataInMessage`) → re-injected via
  `ContextWithJWTClaims` on the consumer → **RLS survives the async hop**.
- **Hand-rolled idempotent `UpsertStream`/`UpsertConsumer`** (delete+recreate on drift) — vanilla NATS
  has no such helper.
- **`chunk_handler.go` = a batch splitter, not payload chunking.** No large-payload framing exists.
- **No `Nak`/`Term`, no DLQ** — redelivery is `AckWait`-timeout + `MaxDeliver`-cap; poison messages
  ack-dropped. A gap vs vanilla best practice.
- **Push queue-group everywhere except one pull consumer** (zeus activity-log).
- **Strict per-service NATS account ACLs** in `nats.conf`.
- **`isLocal` overrides** — Replicas=1, short `AckWait` (2s/4s) so local dev doesn't hang on redelivery.
- **Mid-migration NATS→Kafka** — several subscribers are commented-out husks.