# Repo Kafka Map — ground truth

How Kafka actually works in **our** codebase, focused on conversationmgmt,
notification (`internal/notification`), and spike (`internal/spike`). This is the
factual backbone every lesson is anchored to. All paths are repo-relative;
`file:line` refs point at the code as it exists on this branch.

> Verified by a code scan on 2026-07-11. If a lesson and this file disagree, trust
> the code — re-verify and fix here.

---

## The one-paragraph model

Our services talk to Kafka through **one shared wrapper**,
`internal/golibs/kafka` (interface `KafkaManagement`, `kafka.go:109`), which sits on
top of the Go client **`github.com/segmentio/kafka-go` v0.4.39**. A producer calls
`PublishContext(ctx, topic, key, value)`; a consumer declares a *triple* —
`(topic, consumerGroupID, handler)` — via `Consume(...)`. Payloads are **JSON**
(no Avro/Schema Registry/protobuf-on-the-wire). Topics are **created ahead of time**
by the `fink` job `upsert_kafka_topics`, not auto-created by producers.

---

## Client & wrapper

| Thing | Where | Note |
|---|---|---|
| Client library | imports in `internal/golibs/kafka/*.go` | `segmentio/kafka-go v0.4.39` (NOT sarama, which is in go.mod but unused here) |
| Central interface | `internal/golibs/kafka/kafka.go:109` | `KafkaManagement` |
| Impl | `internal/golibs/kafka/kafka.go:151` | `kafkaManagementImpl` |
| Constructor | `internal/golibs/kafka/kafka.go:183` | `NewKafkaManagement(...)` |
| Bootstrap access | `internal/golibs/bootstrap/resource.go:487` | `rsc.Kafka()` |

Key methods:

```go
PublishContext(ctx, topic string, key []byte, value []byte) error          // kafka.go:453
TracedPublishContext(ctx, spanName, topic string, key, value []byte) error // kafka.go:487  (adds a trace span)
PublishBatchContext(ctx, topic string, messages []Message) error           // kafka.go:367
Consume(topic, consumerGroupID string, opt Option, h MsgHandler) error     // kafka.go:525
GenNewConsumerGroupID(serviceName, topicName string) string                // kafka.go:255
UpsertTopic(topicConfig *kafka.TopicConfig) error                          // kafka.go:773
```

- **`MsgHandler`** = `func(ctx, data []byte) (bool, error)` (`kafka.go:93`). The
  **bool is "retryable"**: `(false, err)` → give up (log & drop / stall by mode);
  `(true, err)` → retry after a wait.
- Producer defaults: `Balancer: &kafka.Hash{}` (key → partition), `RequiredAcks:
  RequireOne` (`kafka.go:465-473`), one lazily-created writer per topic.
- **Topic name prefixing:** `GetTopicNameWithPrefix` (`utils.go:13`) prepends
  `object_name_prefix` (e.g. `prod.tokyo.`). Base constant
  `communication.email-sending` → runtime `prod.tokyo.communication.email-sending`.
- **Consumer group id:** `GenNewConsumerGroupID` →
  `"{prefix}{serviceName}.consumer-group.{topic}"`, e.g.
  `prod.tokyo.spike.consumer-group.communication.email-sending`.

## Topic creation (why publish fails if a topic is "missing")

Producers do **not** create topics. The `fink` job `upsert_kafka_topics`
(`cmd/server/fink/upsert_kafka_topics.go:33`) collects `[]*kafka.TopicConfig` from
each service's `GetTopicConfig*()` in `cmd/server/fink/topics/` and calls
`UpsertTopic` for each. A topic must be declared there (partitions, replication,
`retention.ms`) and created by that job before produce/consume works; otherwise the
segmentio writer errors on an unknown topic.

---

## Topics catalog (base names — constants in `internal/golibs/constants/common.go`)

| Constant | Value | Consumer | Producer |
|---|---|---|---|
| `EmailSendingTopic` | `communication.email-sending` | **spike** | spike gRPC `SendEmail` |
| `EmailSendingHighPriorityTopic` | `communication.email-sending.high-priority` | **spike** (own group) | spike gRPC (high priority) |
| `AsyncNotificationUpsertingTopic` | `notification.async-notification-upserting` | **notification** (normal lane) | notification async svc; conversationmgmt |
| `AsyncNotificationUpsertingHighPriorityTopic` | `notification.async-notification-upserting.high-priority` | **notification** (high lane) | notification; **conversationmgmt** (`infrastructure/kafka/notification.go:74`) |
| `SystemNotificationUpsertingTopic` | `notification.system-notification-upserting` | **notification** | payment, discount, notif cronjob |
| `InvalidDeviceTokenTopic` | `notification.invalid-device-token` | **notification** | notification (self) |
| `NotificationBulkJobTopic` | `notification.bulk-job` | **notification** (bulk_job) | notification `upsert_bulk_job` |
| `NotificationExecScheduledBulkJobTopic` | `notification.exec-scheduled-bulk-job` | **notification** | notification `exec_scheduled_bulk_job` |
| `AgoraUserConnection` | `communication.agora-user-connection` | **conversationmgmt** | conversationmgmt Agora webhook |
| `ChatV2SystemMessage` | `communication.chatv2-system-message` | **conversationmgmt** | conversationmgmt `MessageRepo.SendSystemMessage` |
| `ChatMessageExportTransform` | `communication.chatv2-message-export-transform` | **conversationmgmt** | conversationmgmt backup + job |
| `StudentCourseClassAcademicYear` | `arch.student-course-class-with-academic-year` | **conversationmgmt** & **notification** | external (arch/mastermgmt) |

---

## The recurring worked example — spike email flow

The clearest end-to-end path; used across lessons.

```
Client → gRPC SendEmail                        internal/spike/.../grpc/email_modifier_send_email.go:47
   │  TopicByPriority(priority) picks topic     internal/spike/.../grpc/priority.go:11
   │  TracedPublishContext(topic, key=EmailID, json(SendEmailEvent))
   ▼
[ Kafka topic  communication.email-sending  (+ .high-priority lane) ]
   │  key = EmailID  → Hash balancer → same email always same partition
   ▼
spike consumer  SendEmailConsumer.Consume()    internal/spike/.../kafka/send_email_consumer.go:21
   │  group = GenNewConsumerGroupID("spike", topic),  AlwaysCommit()
   ▼
SendEmailHandler.Handle(ctx, data)             internal/spike/.../consumers/send_email_handler.go:46
   │  json.Unmarshal → SendEmailEvent → send via SendGrid → update DB status
   │  return (retryable, err):  SendGrid fail → (true,…);  ErrPayloadTooLarge → (false,…)
   ▼
email status PROCESSED / PROCESSED_FAILED
```

Two **lanes** (normal + high priority) = two topics + two consumer groups in the
*same* pod, so a normal-topic backlog never delays urgent email
(`send_email_consumer.go:39-41`).

---

## Delivery semantics in our code (interview-critical)

- **No exactly-once, no Kafka transactions.** Idempotency is done in the app layer —
  notification's `idempotentService` keyed by `IdempotentID`
  (`consumer_upsert_async_notification.go:127-148`).
- **Two commit modes** (`internal/golibs/kafka/consumer.go`):
  - **Strict (default)** — `strictConsumer` (`consumer.go:203`): `FetchMessage`, then
    `CommitMessages` **only after** the handler succeeds → **at-least-once**; a poison
    message can **block the partition**. Used by notification's async & system consumers.
  - **Loose / `AlwaysCommit()`** — `looseConsumer` (`consumer.go:165`): `ReadMessage`
    auto-advances the offset regardless of handler outcome → failed message is
    effectively dropped, but the partition never stalls. Used by spike,
    conversationmgmt, and notification's device-token/sync consumers.
- **No dead-letter topic.** "DLQ-like" = strict consumer stalls on the bad message, or
  the loose/`handleMessageStrictly` path logs and drops it
  (`consumer_upsert_async_notification.go:107-111`).
- **acks:** `RequireOne` on the producer (`kafka.go:465`).
- **Retention:** set per topic in `cmd/server/fink/topics/*.go` (7–30 days for
  notification/conversationmgmt topics; email topics have no override).

## Consumer options (`internal/golibs/kafka/consumer_option.go`)

`MaxBytes`, `MaxAttempts`, `RebalanceTimeout`, `HeartbeatIntervalTime`,
`ReconnectAttempts`, `RetryLogicAttempts` (default 10), `WaitingTimeToRetryLogic`,
`AlwaysCommit()`, `StartFromLastOffset()`, `Instances(n)` (default 1 goroutine per
group), `RetryTimeout` (default 360s). Retry loop: `consumer.go:110-163`.

## Where consumers are started

- Implement `KafkaServicer.InitKafkaConsumers(ctx, cfg, *Resources)`
  (`bootstrap/kafka.go:16`), registered with `WithKafkaServicer(s)`.
  - conversationmgmt: `cmd/server/conversationmgmt/gserver.go:239`
  - notification (gRPC pod): `cmd/server/notificationmgmt/gserver.go:307`
  - notification (dedicated gjob `notificationmgmt_consumers`): `cmd/server/notificationmgmt/consumers.go:32`
  - spike: `cmd/server/spike/gserver.go:116`
- Liveness: `AllConsumersRunning()` (`kafka.go:263`) → `/healthz` 503 restarts wedged
  pods (`consumers.go:110`).