# Cheat sheet — NATS JetStream (this repo)

Dense revision sheet. One glance per lesson. Interview one-liners at the bottom.

---

## Core NATS vs JetStream

```
core NATS   pub/sub, at-MOST-once   no persistence   "fire and forget" fan-out
JetStream   streams + consumers, at-LEAST-once   persistence, replay, dedup, acks   ← this course
```

## The flow (say it out loud)

```
publisher ──Subject.Name──▶ STREAM (captures subjects, retention=interest, file, replicas=3)
   DataInMessage{Payload, ResourcePath, UserId, TraceInfo} ; Nats-Msg-Id = ULID (dedup)
                                     │
                          CONSUMER (durable, push queue-group, AckExplicit, DeliverNew, FilterSubject)
                                     │  server pushes to one member of the queue group
                          handleMsg: unmarshal → re-inject tenancy (ctx) → span → cb(ctx, payload) → Ack
                                     │
                          subscriber (conversationmgmt / notification)
```
Streams provisioned by **fink**; consumers provisioned at subscribe time by the golib.

## Subjects & wildcards

- Subject = dotted PascalCase: `Staff.Upserted`, `User.Created`, `Notification.Created`.
- `*` = one token (`User.*`); `>` = the rest (`chat.chat.>`). Streams capture wildcards; consumers
  `FilterSubject` to a concrete one.
- **Queue group** = each message → exactly one member (load balancing).

## Streams (fink)

```go
// cmd/server/fink/streams/notificationmgmt.go
&nats.StreamConfig{
  Name:      constants.StreamNotification,   // "notification"
  Retention: nats.InterestPolicy,            // keep only while a consumer still needs it
  Subjects:  []string{"Notification.*"},
  Replicas:  3,                              // (isLocal → 1)
  MaxBytes:  4 GB,
}
```
Retention: **Limits** (age/bytes/msgs) · **WorkQueue** (delete on ack) · **Interest** (delete when no
consumer needs it) — repo uses **Interest**. Storage = file (durable). `UpsertStream` = idempotent
(delete+recreate on retention drift, else UpdateStream, else AddStream).

## Consumers (`QueueSubscribe`)

```go
jsm.QueueSubscribe(SubjectUserCreated, QueueConversationMgmtUserCreated, nats.Option{
  JetStreamOptions: []JSSubOption{
    nats.Bind(StreamUser, DurableConversationMgmtUserCreated),
    nats.DeliverNew(), nats.ManualAck(),
    nats.MaxDeliver(3), nats.AckWait(30*time.Second), nats.MaxAckPending(30),
  }}, handler)
```
- **push** (server delivers) vs **pull** (`Fetch` batches — only zeus activity-log).
- **durable** (named, survives restart, resumes) vs ephemeral.
- Deliver policy: **DeliverNew** (repo default) / All / Last / by-time / by-seq.
- Ack policy: **AckExplicit** (forced). `AckWait` → redeliver on no-ack. `MaxDeliver` → cap (then drop).
- `UpsertConsumer` (idempotent, delete+recreate on config drift) runs at subscribe time.

## The consumer heart — `handleMsg` (jetstream.go:663-740)

```
1. read Nats-Msg-Id + sequence ; defer metric
2. SkipMsgOlderThan? → skip + return (DENY_CONSUME_OLD_MESSAGE)
3. proto.Unmarshal DataInMessage ; unmarshal fail → Ack (DROP)
4. rehydrate trace from TraceInfo → consumer span
5. re-inject tenancy: ctx = ContextWithJWTClaims(ResourcePath, UserID)   ← RLS SURVIVES THE HOP
6. retry, err := cb(ctx, payload)
      success        → msg.Ack()
      err, retry=true  → (no ack) → redeliver after AckWait
      err, retry=false → msg.Ack()  (drop poison)
   // NO Nak() / Term() ; MsgHandler = func(ctx, []byte) (retry bool, err error)
```

## Publishing

```go
jsm.TracedPublish(ctx, spanName, SubjectNotificationCreated, msgBytes)   // sync + OTel span
jsm.PublishAsyncContext(ctx, subject, data)                             // async (buffer 200k)
// wraps DataInMessage{Payload, ResourcePath, UserId (from ctx), TraceInfo (B3)} ; Nats-Msg-Id = ULID
// callers wrap in try.Do (5 attempts) for publish retry
```

## Delivery semantics

- **At-least-once** (explicit ack). **Dedup** via `Nats-Msg-Id` (ULID) in a 2-min window (+ app-level
  idempotency). **Redelivery** = no-ack → `AckWait` → up to `MaxDeliver`, then **dropped (no DLQ)**.
- Repo uses only **`Ack`** (no `Nak`/`Term`); retryable failure = don't-ack-and-wait.
- `MaxAckPending` = backpressure. `isLocal` → short `AckWait`, 1 replica.

## The learner services

| Subscriber | Subject | Stream | Notes |
|---|---|---|---|
| conversationmgmt `UserCreated` | `User.Created` | user | MaxDeliver 3, AckWait 30s |
| conversationmgmt `StaffUpserted` | `Staff.Upserted` | staff | DeliverNew, SkipMsgOlderThan aWeekAgo |
| conversationmgmt `UserGroupUpdate` | `UserGroup.Upserted` | usergroup | MaxDeliver 10, AckWait 300s |
| notification `push_notification` | `Notification.Created` | notification | MaxDeliver 3, AckWait 90s |

conversationmgmt **publishes** `Notification.Created` (→ notification) via `TracedPublish`. Business
logic is transport-agnostic (`internal/notification/subscribers/`), reused by NATS **and** Kafka.

## NATS vs Kafka here

- **NATS** = internal service events / fan-out + the activity-log firehose (dotted subjects, interest).
- **Kafka** = durable streams / CDC / high-throughput pipelines (`{team}.{topic}`).
- Active **NATS→Kafka migration** (Unleash-flagged; commented-out subscribers; shared domain handler).

## Ops

`nats.conf`: jetstream (file store, 4GiB mem / 10GiB file), client `:4223`, monitoring `:8223`,
**per-service publish/subscribe ACLs**. Helm statefulset, 3 replicas. Mock: `mock/golibs/nats`
(testify). Streams by fink; `isLocal` = 1 replica + short AckWait.

---

## Interview one-liners

- **"Core NATS vs JetStream?"** — core NATS is at-most-once pub/sub with no persistence; JetStream
  adds streams and durable consumers for at-least-once delivery, replay, and dedup.
- **"Stream vs consumer?"** — a stream stores messages for a set of subjects; a consumer is a stateful
  cursor a subscriber reads it through, with its own ack/redelivery config.
- **"Push vs pull?"** — push: the server delivers as messages arrive; pull: the client fetches batches
  and controls flow. We use push queue-groups everywhere except the activity-log pull consumer.
- **"How is delivery at-least-once?"** — explicit ack. A processed message is acked; a retryable
  failure isn't acked, so it redelivers after `AckWait`, up to `MaxDeliver`. Dedup by `Nats-Msg-Id`.
- **"How does RLS survive an async event?"** — we wrap the payload in a proto envelope carrying
  `resource_path` and the trace context, and re-inject them into `context` on the consumer, so
  tenant-scoped queries and distributed tracing work exactly as on a synchronous request.
- **"NATS or Kafka?"** — NATS for low-latency internal service events and fan-out; Kafka for durable,
  high-throughput streams and CDC. We run both and are migrating some flows from NATS to Kafka.
- **"A gap in your setup?"** — no dead-letter queue and no `Nak`/`Term`: a message that exhausts
  `MaxDeliver` is just dropped (surfaced only via metrics/logs).