05 · Kafka (outbound outbox, inbound consume, settings, DO topics)

As-built reference for Kafka in TMS v2 phase 1. Repo for every path:line below is bigc-tms-core-v2 (worktree wt-v2-prototype-core) unless a line says store-v2 (worktree wt-v2-prototype-store).

Outbound sends run on the jobs engine. Its lanes, partitions, leases, retry delays, the jobs table and the shared outcome path (JobOutcome) are described in 04-jobs-engine.md and are not repeated here. This file covers what is specific to Kafka: the topics, the producer and its error classes, the DO payloads, the consumer, dedupe, the DLQ, settings reload and the web pages.

1. Summary

Outbound: a business module writes its change and one jobs row per routed topic in one Mongo transaction (the transactional outbox). The jobs fast lane is the relay: it reads due rows and hands each one to the kafka:<topic> target, which sends it through the pod's single producer (no kafkajs retries). Each Kafka error is sorted into "not sent", "may be written" or "refused", and the delivery mode of the topic (at-most-once by default) decides whether the row is resent, paused or closed DEAD. Inbound: one consumer group per declared consume event. Every message is claimed in kafka_consumed by message_id, handled, then marked done. A permanent failure is written to kafka_consume_dlq. A crash or thrown error releases the claims so Kafka redelivers the batch. Brokers, auth, the on/off switch and per-event topic names live in the kafka settings record. Each pod reconciles its producer and consumers when that record changes.

repo	branch	sha	folders / files
bigc-tms-core-v2	origin/main-v2-prototype	3d70475a	src/kafka/, src/common/utils/kafka.ts, src/delivery_order/{delivery_order.service.ts,delivery_order.kafka_consumer.ts,do-update.payload.ts,delivery_order_version.schema.ts}, src/settings/ (reload), src/main.ts (shutdown hooks)
bigc-tms-store-v2	main-v2-prototype	0851a3a	src/pages/setting/kafka/, src/services/kafka/, src/pages/api/kafka/test-connection/, src/pages/setting/outbound-events/, src/pages/api/outbound_events/, src/apis/{client,server}/outbound_event/

History, so the commit messages make sense (current code wins):

commit	what it introduced	status at 3d70475a
931237b3	Outbox rows in the DO transaction, jobs.relay.ts (claim + lease, at-most-once), /outbound_events admin API, v1-shaped DO payloads, channel routing, event_id/trace_id headers	Relay and /outbound_events API removed in 38dad930. Payload builder, routing and headers kept (moved into delivery_order)
b9513620	Relay per-row catch, drain on shutdown, "hold when Kafka off" (D-212), do-post consume end to end	Drain and do-post kept. Hold when Kafka off is lost (G3)
73ab7b66	Consume claim: retry after failed, stale processing reclaim after a 5 min lease, DLQ write failure redelivers	Kept, moved to src/kafka/kafka.consumer.ts
38dad930	Jobs engine; kafka:<topic> fast-lane target, error classes, delivery modes	Kept, target moved to src/kafka/kafka.target.ts in 3dd1c5a0
3dd1c5a0	Split KafkaService into producer / consumer runner / config reader; kafka settings record with onChange reconnect (replaces the setting poll); Redis pod status; kafka_logs; request context and trace headers	Current

2. File map

file	responsibility
src/kafka/kafka.constants.ts	KAFKA_EVENTS (the declared events: name, direction, default topic), GROUP_PREFIX, Redis status key prefix, heartbeat 10 s / TTL 30 s, kafka_logs name and 60-day retention
src/kafka/kafka.enum.ts	KafkaEventType CONSUME / PRODUCE
src/kafka/kafka.types.ts	ConsumeMessage, ConsumeHandler (resolves a failure reason string or undefined)
src/kafka/kafka.settings.ts	The kafka settings record and its keys with class-validator schemas; two encrypted keys
src/kafka/kafka.config.ts	KafkaConfigReader: reads the settings cache, decrypts secrets into the kafkajs auth shape, topicOf(event)
src/kafka/kafka.service.ts	Public API (send, topicOf, registerConsumer, testConnection, status); reconcile on settings change; 10 s heartbeat that writes Redis status and retries failed connections; shutdown order
src/kafka/kafka.producer.ts	KafkaProducerLifecycle: one lazy, long-lived producer per pod, no kafkajs retry, trace headers, kafka_logs out rows, KafkaSendError with phase connect / send
src/kafka/kafka.consumer.ts	KafkaConsumerRunner: one consumer per consume event, per-message claim, handler call in a request context, done / DLQ / release, busy hold and seek
src/kafka/kafka.target.ts	Registers kafka:<topic> jobs targets; kafkaFailure error classifier; maps classes to TargetUnreachable / TargetOutcomeUnknown / dead_now
src/kafka/kafka.controller.ts	GET /api/v1/kafka/status, POST /api/v1/kafka/test-connection
src/kafka/kafka_log.schema.ts	kafka_logs buffered log definition
src/kafka/consumers/kafka.consumes.schema.ts	kafka_consumed (claim / dedupe, 7-day TTL) and kafka_consume_dlq
src/kafka/kafka.module.ts	Wires settings refs, the log, JobsModule, the two consume models, controller, KafkaService, KafkaTarget
src/common/utils/kafka.ts	buildKafkaConfig (PLAINTEXT / SSL / SASL_PLAINTEXT / SASL_SSL → kafkajs ssl / sasl), createKafka, createProducer
src/delivery_order/delivery_order.service.ts	Outbound caller: _enqueueDoUpdate routes by channel and enqueues inside the DO transaction. Inbound: upsertFromKafka
src/delivery_order/do-update.payload.ts	buildDoUpdateOms, buildDoUpdateStaffapp: v1 field set and order per topic
src/delivery_order/delivery_order_version.schema.ts	JobCounter: per-DO version counter
src/delivery_order/delivery_order.kafka_consumer.ts	KafkaDoConsume: registers the do_consume handler
src/settings/settings.service.ts	Settings cache, 60 s reload, Redis settings:changed pub/sub, onChange handlers (Kafka reconcile hangs off this)
src/main.ts:52	enableShutdownHooks(): SIGTERM runs onModuleDestroy then onApplicationShutdown

3. Data model

Outbox: jobs

Outbound rows are ordinary jobs rows, full schema in 04 §3. Kafka uses these fields as follows:

field	Kafka meaning	set at
target	kafka:<default_topic>	delivery_order.service.ts:186
code	the producer event (do_update_oms, do_update_staffapp). Selects the topic_name override	:187, read at kafka.target.ts:81
key	delivery_no; becomes the Kafka message key, so one DO stays on one partition	:188, sent at kafka.target.ts:83
version	per-DO counter value; fast lane sends in key + version order	:189
dedupe_key	<event>:<delivery_no>:<version> (unique)	:190
headers	{ event_id }, a new ObjectId hex per event	:183,191
payload	built at commit time by the topic's builder; sent as JSON.stringify(payload)	:192, kafka.target.ts:83

job_counters (model job_counter, delivery_order_version.schema.ts:6-15)

field	type	req	default	note
_id	string	yes	—	delivery_no
seq	number	no	0	$inc 1 per DO create that is routed, and per DO update (delivery_order.service.ts:144,159-166,180)

A separate collection, so deleting sent jobs never resets a version (delivery_order_version.schema.ts:5).

kafka_consumed (model kafka_consumed, consumers/kafka.consumes.schema.ts:5-27)

Collection name follows Mongoose default pluralisation of the model name (no explicit collection option).

field	type	req	default	enum / validation	index	note
consume_id	string	yes	—	—	unique	payload.message_id, else <topic>:<partition>:<offset> (kafka.consumer.ts:203)
event	string	yes	—	—	—	declared event, e.g. do_consume
topic	string	no	—	—	—	—
status	string	no	processing	processing / done / failed (not enforced)	—	see §6
consumed_at	Date	no	now	—	TTL 7 days	claim time; also the lease start (kafka.consumer.ts:215-217)
created_at, updated_at	Date	—	timestamps	—	—	—

kafka_consume_dlq (model kafka_consume_dlq, consumers/kafka.consumes.schema.ts:30-69)

field	type	req	default	enum / validation	index	note
consume_id	string	yes	—	—	unique	same id as the claim
event	string	yes	—	—	—	—
topic, partition, offset, key	string / number / string / string	no	—	—	—	where the message was read
raw	Mixed	no	—	—	—	parsed payload, or the raw string when it was not JSON (kafka.consumer.ts:140)
error	string	no	—	—	—	handler's failure reason
attempts	number	no	1	$inc 1 per failed delivery (:141)	—	—
status	string	no	pending	pending / resolved (not enforced)	—	resolved when the same id later succeeds (:131)
failed_at	Date	no	now	—	—	last failure

No TTL: DLQ rows stay until deleted by hand.

kafka_logs (kafka_log.schema.ts:7-32, buffered log)

field	type	req	enum	note
at, expire_at, trace_id	Date, Date, string	at yes	—	from BaseLog (log/base_log.schema.ts:4-8), written by prepareLogRow (log/log.registry.ts:44)
direction	string	yes	in / out	—
topic, key, message_id, partition, offset	—	no	—	message_id is read from the JSON body (kafka.producer.ts:135, kafka.consumer.ts:233)
req_body, res_body	Mixed	no	—	masked (maskBody / maskDeep) and cut by cutLogBodies
req_body_truncated, req_body_bytes, res_body_truncated, res_body_bytes	—	no	—	set by cutLogBodies (64 KB cap per kafka_log.spec.ts:95)
status	string	yes	sent / consumed / failed	—
error.message	string	no	—	—

Indexes: {at: -1} (kafka_log.schema.ts:31), TTL on expire_at (log/log.registry.ts:23), default retention 60 days (kafka.constants.ts:27).

Settings record kafka

Stored in the central settings collection as one record, code kafka, owner kafka (kafka.settings.ts:67-71). Keys are in §8.

Redis

key / channel	value	TTL	writer → reader
kafka:status:<hostname>:<pid>	{pod_id, status: connected | disconnected | disabled, last_connected_at}	30 s	each pod every 10 s and after each reconcile (kafka.service.ts:178-189, kafka.constants.ts:5-8) → GET /kafka/status via KEYS kafka:status:* (:128-132)
pub/sub settings:changed	{code, rev}	—	settings save (settings/settings.service.ts:247) → every pod reloads the record (:323-327)

Relations

from	to	how
jobs (target kafka:*)	delivery_order	key = delivery_no; the job is written in the DO transaction
jobs.version	job_counters	seq for _id = delivery_no
kafka_consumed.consume_id	inbound message_id	1 : 1
kafka_consume_dlq.consume_id	kafka_consumed.consume_id	same id
inbound message	activity_logs.event_key	do:<delivery_no>:msg:<message_id> (unique; a replay aborts the DO transaction) (delivery_order.service.ts:204,118,150)

4. API

Global prefix api, URI versioning (main.ts:32-37), so routes are /api/v1/.... A guard returns false → Nest 403.

method + route	permission	request	response	errors	code
GET /api/v1/kafka/status	any valid Bearer session (no @Roles)	—	KafkaPodStatus[] sorted by pod_id	403 without a session	kafka.controller.ts:8-11, kafka.service.ts:128-132
POST /api/v1/kafka/test-connection	any valid Bearer session (no @Roles)	body {bootstrap_servers?, authen_type?, authen_setting?}. Empty bootstrap_servers → the saved config is tested. Blank, absent or SETTING_SECRET_MASK password / ssl_key_pem take the saved secret; nothing is saved	always 200 {success, message}; message is Connection successful, No bootstrap_servers configured, or the kafkajs error text	403 without a session	kafka.controller.ts:14-19, kafka.service.ts:90-125
GET /api/v1/settings/kafka	admin	—	the record with values; secrets masked	404 not_found before the cache loads	settings/settings.controller.ts:21-24
PUT /api/v1/settings/kafka	admin	{rev, values}	{rev}	409 rev_conflict; 400 on a schema failure	settings/settings.controller.ts:26-33, settings.service.ts:359
POST /api/v1/settings/kafka/reset	admin	ResetSettingsInput	{rev}	as above	settings/settings.controller.ts:35-42
GET /api/v1/jobs?target=kafka:<topic> · POST /api/v1/jobs/:id/retry · POST /api/v1/jobs/:id/skip	admin	see 04 §4	see 04 §4	see 04 §4	jobs/jobs.controller.ts:22-62

Not served: /api/v1/outbound_events (removed in 38dad930; still called by store-v2, §10). No DLQ list, replay or resolve endpoint exists.

5. Conditions

when …	the system must …	enforced at	test
a DO is created	enqueue one job per topic routed by chanel[0].chanel_code in the same transaction	delivery_order.service.ts:112-122,168-195	delivery_order.service.spec.ts:104,235
a DO is updated and status differs from before	enqueue the routed jobs with a new version	:152-154	:104
a DO update leaves status unchanged	enqueue nothing (version still increments for the activity key)	:144,152	:124 (duplicate update → one job)
chanel_code ∈ DONJAI, CHOKCHAI, HLH	route to do_update_oms	:27,173	:104
chanel_code ∈ SPO, OD, HLH	route to do_update_staffapp (HLH goes to both)	:28,174	untested for SPO/OD
channel in neither list	log do update not enqueued (channel not routed), no job, no version	:176-179	untested
the transaction aborts	leave no DO and no job	common/utils/transaction.ts, :112,137	delivery_order.service.spec.ts:235
the per-event topic_name is set for the job's code	send to that topic instead of the default	kafka.target.ts:81, kafka.config.ts:35-39	delivery_order.service.spec.ts:199,208
a send gets no broker connection	throw TargetUnreachable: target paused, attempts kept	kafka.target.ts:86	kafka.target.spec.ts:130,471
a send may have been written (no answer)	at-most-once → DEAD and pause; at-least-once → keep PENDING, pause	kafka.target.ts:87, jobs/fast_lane.worker.ts:74-77	kafka.target.spec.ts:235,253
the broker answered REQUEST_TIMED_OUT / NETWORK_EXCEPTION	at-most-once → DEAD, target not paused; at-least-once → PENDING, paused	kafka.target.ts:89	kafka.target.spec.ts:415,434
the client rejected the record (KafkaJSNonRetriableError, e.g. empty value)	DEAD at once on either mode, target not paused	kafka.target.ts:91	kafka.target.spec.ts:444,451
the broker refused the record (non-retriable protocol error, e.g. too large)	answered failure: retry 1/2/4/8/16 min, DEAD at the 6th run	kafka.target.ts:42,92	kafka.target.spec.ts:142
a produce fails	drop the producer; the next send reconnects	kafka.producer.ts:65-68,99-104	kafka.service.spec.ts:78
kafkajs would retry a produce	never: retries: 0 on client and producer	kafka.producer.ts:14-16,124-126	kafka.service.spec.ts:70
an inbound message has no delivery_no	fail with delivery_no is required → DLQ	delivery_order.service.ts:198-200	untested
an inbound message has no message_id (absent or "")	fail with missing message_id → DLQ, no DO write	:201-203	delivery_order.service.spec.ts:310
an inbound delivery_no exists	update it; else create; on a create race (E11000) update instead	:205-216	delivery_order.service.spec.ts:286
the same message_id is applied again	activity event_key duplicate → ActivityAlreadyApplied → no-op, return current DO	:204,217-222, activity_log/activity_log.service.ts:52	delivery_order.service.spec.ts:286
a claim row is done	skip the message and commit its offset	kafka.consumer.ts:221-224,104,164	kafka.consumer.it.spec.ts:122
a claim row is processing and younger than 5 min	stop the batch there, commit only what came before, wait 5 s, seek back	:99-102,167-171	untested
a claim row is failed, or processing older than 5 min	claim again and process	:212-219	untested
the handler resolves a reason	DLQ upsert (pending, attempts +1), claim failed, offset committed	:123,133-155	kafka_log.spec.ts:127 (log row only)
the handler resolves undefined	claim done, DLQ row for the id resolved	:129-132	kafka.consumer.it.spec.ts:110
the handler throws, or a done / DLQ write fails	delete this batch's processing claims and rethrow so Kafka redelivers	:156-162	untested
enabled is false, or a consume event row has enabled: false, or no handler is registered	that consumer is not run (stopped if running)	kafka.consumer.ts:66,73-77	kafka.consumer.it.spec.ts:90
enabled is false	disconnect the producer at reconcile	kafka.producer.ts:82-91	kafka.service.spec.ts:153
brokers, auth or topic change	rebuild the producer / that consumer (signature compare)	kafka.producer.ts:85-88, kafka.consumer.ts:68-78	kafka.service.spec.ts:168
a consumer or the producer failed to start	the 10 s heartbeat reconciles again (only while enabled)	kafka.service.ts:164-172	kafka.service.spec.ts:236,304,331
a message is sent or handled	write one kafka_logs row, secrets masked, bodies cut at 64 KB; a log-store outage never blocks Kafka	kafka.producer.ts:128-143, kafka.consumer.ts:228-241	kafka_log.spec.ts:60-141

6. States and flows

Outbound job row (Kafka view of the 04 state machine)

from	event	to	code
—	DO transaction commits	PENDING (attempts 0)	delivery_order.service.ts:182-194
PENDING	send acked	row deleted, kafka_logs out sent	jobs/fast_lane.worker.ts:91-96, kafka.producer.ts:64
PENDING	not_sent	PENDING, attempts kept, target paused 30 s	kafka.target.ts:86
PENDING	unknown / timed_out, at-most-once	DEAD (closed_at)	kafka.target.ts:87-89, fast_lane.worker.ts:74-76
PENDING	unknown / timed_out, at-least-once	PENDING, attempts kept, paused	fast_lane.worker.ts:77
PENDING	answered	PENDING + backoff, or DEAD at the 6th run	fast_lane.worker.ts:104-121
PENDING	invalid	DEAD	kafka.target.ts:91
PENDING (in flight)	owner pod dies, partition taken over	at-most-once DEAD; at-least-once stays PENDING and is resent	fast_lane.worker.ts:249-264
DEAD	admin retry / skip	PENDING from scratch / deleted	jobs/jobs.controller.ts:46-62

Kafka error classes (kafkaFailure, kafka.target.ts:40-51)

Checked in this order, first match wins:

#	condition on the error chain (cause / originalError, up to 10)	class	thrown to the lane
1	a KafkaJSProtocolError with retriable === false	answered	the original error (retry path)
2	no KafkaJS* error in the chain	answered	original
3	KafkaSendError with phase: connect	not_sent	TargetUnreachable
4	KafkaJSNonRetriableError and none of KafkaJSProtocolError, KafkaJSConnectionError, KafkaJSRequestTimeoutError, KafkaJSNumberOfRetriesExceeded	invalid	Error + dead_now
5	protocol error type REQUEST_TIMED_OUT or NETWORK_EXCEPTION	timed_out	TargetOutcomeUnknown + target_up
6	protocol error whose type is not UNKNOWN_TOPIC_OR_PARTITION, LEADER_NOT_AVAILABLE, NOT_LEADER_FOR_PARTITION, NOT_ENOUGH_REPLICAS	unknown	TargetOutcomeUnknown
7	KafkaJSBrokerNotFound, KafkaJSMetadataNotLoaded, KafkaJSTopicMetadataNotLoaded, a remaining protocol error, or a KafkaJSConnectionError matching ECONNREFUSED|ENOTFOUND|EHOSTUNREACH|ENETUNREACH|EAI_AGAIN|^Connection timeout	not_sent	TargetUnreachable
8	any other KafkaSendError (send phase)	unknown	TargetOutcomeUnknown
9	otherwise	not_sent	TargetUnreachable

At-most-once vs at-least-once

	at_most_once (default for every declared topic)	at_least_once (only via KAFKA_TARGET_TOPICS=topic:at_least_once)
send may have landed, no answer	row DEAD, a person retries or skips	row kept PENDING, resent after the 30 s probe
pod died with the row in flight	DEAD on takeover	resent by the new owner
duplicates on the topic	never from TMS	possible; receivers dedupe on header event_id
why	v1 receivers do not dedupe (D-151, D-165)	opt-in per topic for receivers that dedupe
code	kafka.target.ts:61-63, jobs/job_targets.registry.ts:46	kafka.target.ts:12-16,64-66

Inbound claim (one row in kafka_consumed)

Inbound batch (consumeBatch, kafka.consumer.ts:89-172)

do-post end to end

Settings change and reconcile

Signatures: producer GROUP_PREFIX|brokers|auth (kafka.producer.ts:34), consumer GROUP_PREFIX|brokers|topic|auth (kafka.consumer.ts:68). Auth includes the decrypted secrets, so changing a password reconnects.

Pod status

status	when	code
connected	producer present and running consumers ≥ expected	kafka.service.ts:170-172,184
disconnected	enabled true but not healthy	:184
disabled	enabled false	:184

last_connected_at is set on each transition into healthy (:180).

7. Jobs, schedules, queues, events

Every topic

direction	topic (default)	setting key (kafka.events[] row)	payload schema	producer / consumer file	delivery guarantee
in	bigc.lastmiles.do-post	event: do_consume → topic_name, enabled	JSON DeliveryOrderInput fields (delivery_order/delivery_order.dto.ts) plus required message_id, delivery_no; not validated by class-validator on this path; Mongoose strict drops unknown fields	consumer src/kafka/kafka.consumer.ts, handler src/delivery_order/delivery_order.kafka_consumer.ts:20-33 → delivery_order.service.ts:197-223	at-least-once delivery from Kafka; applied at most once per message_id (claim + activity event_key); permanent failure → DLQ, offset committed
out	lastmiles.oms.do-update	event: do_update_oms → topic_name (enabled ignored, G3)	OMS_SHAPE (do-update.payload.ts:15-69)	delivery_order.service.ts:168-195 → src/kafka/kafka.target.ts	at-most-once
out	lastmiles.staffapp.do-update	event: do_update_staffapp → topic_name	STAFFAPP_SHAPE (do-update.payload.ts:71-141)	same	at-most-once
out	lastmiles.staffapp.load-update	event: load_update	none (no producer)	target registered only (kafka.constants.ts:20)	at-most-once
out	lastmiles.staffapp.task-update	event: task_update	none	target only (:21)	at-most-once
out	lastmiles.staffapp.driver-update	event: driver_update	none	target only (:22)	at-most-once
out	any topic in KAFKA_TARGET_TOPICS	none (no topic_name override: topicOf only knows declared events)	caller-defined	target only, no caller in code	at-most-once, or at-least-once with :at_least_once

Every outbound message: key = job key; value = JSON.stringify(payload); headers event_id (= job.headers.event_id, else the job _id), traceparent (W3C), trace_id (kafka.target.ts:83, kafka.producer.ts:28-32,62). Consumer group per consume event: <GROUP_PREFIX>.<event>, fromBeginning: false, fetch max wait 1 s (kafka.consumer.ts:175-178,20).

DO-update payload (do-update.payload.ts)

Both builders project the DO through a fixed shape. Rules (:143-169):

leaf type	output
any empty value (null, undefined, "", [])	key left out; an object whose keys are all left out is left out too
string	String(v); an object becomes JSON.stringify
number	Number(v); boolean → 1/0; not a number → the string
date	ISO string for a Date, else String(v)
raw	value as is; an object becomes JSON.stringify
array_string	array of non-empty strings
object	first element when the source is an array
array_object	each element projected

Source fixes before projecting (:171-175): expected_delivery_date_from falls back to expect_delivery_date; pickup_info falls back to source. OMS adds Status (capital S) = DO status (:177-178); staffapp uses status.

OMS top-level fields (order)

staffapp top-level fields (order)

delivery_no, external_delivery_no, order_type, order_date, chanel{chanel_code, chanel_name}, expected_delivery_date_from, expect_delivery_times_from, expect_delivery_times_to, source{source_store_code, source_store_name, source_address{phone, street1, postal_code, latlong, district, district_code, subdistrict, subdistrict_code}}, destination{destination_address{first_name, phone, street1, postal_code, district, subdistrict, province, latlong, country, country_code, remark}}, is_cod, cod_amount, cod_currency, cod_payment_method, items[]{item_no, product_sku, status[], volume_cbm, weight_kg, quantity}, Status

delivery_no, external_delivery_no, chanel{…}, order_type, order_date, delivery_type, external_inv_no, total_amount, comit_delivery_times, estimate_delivering_ontime, delivering_ontime, actual_delivered, expected_delivery_date_from, expect_delivery_times_from, expect_delivery_times_to, pickup_info{source_store_code, source_store_name}, is_cod, cod_amount, cod_currency, cod_payment_method, items[]{item_no, product_sku, ean, product_name, quantity, load_method, uom, weight_kg, amount, currency}, payment_proof_info{…}, POD_info{…}, driver_info{driver, driver_no, phone, partner, vehicle_type, plate_number, additional_info{note}}, status, note

The meta argument (event_id, version, status_at) is accepted but not written into either body (:177-181); the event id travels only as the event_id header (G5).

Timers

timer	period	what	code
Kafka heartbeat	10 s per pod	reconcile if enabled and unhealthy, else rewrite status	kafka.service.ts:66,164-167
settings reload	60 s (SETTINGS_RELOAD_INTERVAL_MS)	reload every settings record, fire onChange on newer rev	settings/settings.service.ts:63-65, settings.types.ts:4
busy re-fetch	5 s	wait before seeking back to a held message	kafka.consumer.ts:16,169
consume lease	5 min	age after which a processing claim is retaken	kafka.consumer.ts:18,215
buffered log flush	see 07 doc (1 s / 100 rows)	inserts kafka_logs	src/log/buffered.writer.ts

8. Settings and env

Settings record kafka (kafka.settings.ts:73-94)

key	default	validation	effect	read at
bootstrap_servers	KAFKA_BROKERS split on ,, else ['localhost:9092']	non-empty string array	broker list for producer, consumers and test	kafka.config.ts:22
authen_type	PLAINTEXT	PLAINTEXT / SSL / SASL_PLAINTEXT / SASL_SSL	turns on ssl and/or sasl	common/utils/kafka.ts:24-46
authen_setting	{}	sasl_mechanism, username, ssl_ca_pem, ssl_cert_pem strings; ssl_reject_unauthorized bool	SASL mechanism defaults to plain; SSL without PEMs = system CA	kafka.config.ts:24-28
sasl_password	null	string, encrypted	SASL password	kafka.config.ts:26
ssl_key_pem	null	string, encrypted	client key	kafka.config.ts:27
enabled	true	boolean	false: consumers stopped, producer disconnected at reconcile, heartbeat stops retrying. Does not stop sends (G3)	kafka.service.ts:150, kafka.consumer.ts:66, kafka.producer.ts:84
events	one row per KAFKA_EVENTS entry with topic_name = default_topic, enabled: true	array of {event, event_type?, topic_name?, enabled?}	per-event topic and (consume only) on/off; matched by event alone, code owns direction	kafka.config.ts:13,35-39, kafka.consumer.ts:64-66

An invalid stored value falls back to the code default with a warning (settings/settings.service.ts:279-285).

Env

key	default	effect	read at
KAFKA_GROUP_PREFIX	KAFKA_CLIENT_ID, else bigc-tms-core-v2	kafkajs clientId and consumer group prefix	kafka.constants.ts:4
KAFKA_CLIENT_ID	—	fallback for the prefix	same
KAFKA_BROKERS	localhost:9092	default of bootstrap_servers only (a saved value wins)	kafka.settings.ts:75
KAFKA_TARGET_TOPICS	empty	extra kafka:<topic> targets; :at_least_once switches a topic (declared ones too)	kafka.target.ts:64
SETTINGS_RELOAD_INTERVAL_MS	60000	settings poll period	settings/settings.types.ts:4
JOBS_WORKERS	on	off: this pod enqueues but never sends (consumers still run)	jobs/jobs.constants.ts:40
KAFKA_IT_BROKERS, KAFKA_IT_MONGO (tests)	unset → spec skipped	real-broker integration spec	kafka.consumer.it.spec.ts:11-14

Code constants

constant	value	code
producer client	retries: 0, connect 3 s, request 10 s	kafka.producer.ts:15-16
base client (consumers)	retries: 8, initial 300 ms, log level ERROR	common/utils/kafka.ts:17-22
test-connection client	retries: 1, connect 3 s, request 5 s	kafka.service.ts:100-104
channel routing	OMS: DONJAI, CHOKCHAI, HLH; staffapp: SPO, OD, HLH	delivery_order.service.ts:27-28
claim TTL	7 days	consumers/kafka.consumes.schema.ts:27

9. Concurrency and recovery

Outbound recovery (partition takeover, stall past lease, broker down for days, shutdown drain of the lanes) is in 04 §9. Kafka-specific rows:

situation	what the code does	code
many concurrent first sends on one pod	one shared producerConnecting promise, one producer	kafka.producer.ts:72-80, kafka.service.spec.ts:50
produce request lost mid-flight	never retried inside kafkajs; the row's delivery mode decides	kafka.producer.ts:14-16, kafka.target.ts:87
Kafka off in settings while rows are due	the fast lane still sends: ensure() connects lazily with the stored brokers, no enabled check. If brokers are unreachable, rows pile up PENDING paused	kafka.producer.ts:55-80,106-121 (G3)
SIGTERM during a send	onModuleDestroy: lanes stop and await the current tick; consumers disconnect (kafkajs waits for the running batch). onApplicationShutdown disconnects the producer after that, so the in-flight send completes	kafka.service.ts:70-80, jobs/fast_lane.worker.ts:151-162, kafka.service.spec.ts:111, kafka.consumer.it.spec.ts:122
same inbound message on two pods (rebalance)	unique consume_id: one claims, the other gets busy and re-fetches; if both reach the service, the activity event_key aborts the loser's transaction → no-op	kafka.consumer.ts:209-226, delivery_order.service.ts:217-222, spec :286
two messages create the same new DO	second create hits unique delivery_no → retried as an update	delivery_order.service.ts:210-216, delivery_order.schema.ts:14
pod dies after claim, before done	claim stays processing; redelivered message is busy for up to 5 min (partition re-checks every 5 s), then retaken and processed. Order kept	kafka.consumer.ts:18,212-219,167-171
handler runs longer than 5 min	no lease renewal: another member may retake the claim and run it again	kafka.consumer.ts:18 (G6)
handler throws	claims released, error rethrown, kafkajs restarts the consumer and redelivers. Not reachable for do_consume: KafkaDoConsume catches every error and returns it as a reason	kafka.consumer.ts:156-162, delivery_order.kafka_consumer.ts:26-32 (G1)
DLQ write fails	rethrow before commit, claims released, batch redelivered; a lasting DLQ fault loops the consumer	kafka.consumer.ts:149-152
a sender resends a message_id that failed	the failed claim is retaken and processed; on success the DLQ row turns resolved	kafka.consumer.ts:215,131
config change while a reconcile runs	pending flag → one more run with the latest cache	kafka.service.ts:139-161, spec :183
stale settings snapshot after a newer write	ignored by rev compare	settings/settings.service.ts:308-311
Redis down	status write warns; settings change reaches other pods on the next 60 s poll	kafka.service.ts:187-188, settings/settings.service.ts:247-250
Redis restarted	status key rewritten on the next 10 s beat	kafka.service.ts:164-167, spec :213
pod killed	its status key expires after 30 s	kafka.constants.ts:8, spec :197
kafka_logs store down	rows dropped and counted; send and consume continue	kafka_log.spec.ts:72

10. Web (store-v2)

page / route	file (store-v2)	calls (via Next proxy → API_GATEWAY)	gating	status
/setting/kafka	src/pages/setting/kafka/index.tsx:16-185, hook src/services/kafka/index.ts:11-54	GET/PUT /settings/kafka (src/apis/client/setting/index.ts:10,14); POST /kafka/test-connection (:21-22, proxy src/pages/api/kafka/test-connection/index.ts:4-17)	menu entry src/constants/menu.ts:88-90; core enforces admin on settings, only a session on test-connection	Works. Form: enabled, authen_type, bootstrap_servers list, SASL / SSL fields, encrypted sasl_password / ssl_key_pem, events table with topic_name and enabled per event (index.tsx:125-165). The enabled toggle has no effect on producer events (G3). No pod-status view: GET /kafka/status has no caller
/setting/outbound-events	src/pages/setting/outbound-events/index.tsx:20-181	GET /outbound_events?page&limit&status&topic&key, POST /outbound_events/:id/retry (src/apis/client/outbound_event/index.ts:20-26; proxies src/pages/api/outbound_events/index.ts, [id]/retry.ts; server src/apis/server/outbound_event/index.ts:17-23)	menu entry "Outbound Events" src/constants/menu.ts:102-106, no roles	Broken — confirmed. Core at 3d70475a serves no outbound_events route (no match in src; the only jobs routes are jobs/jobs.controller.ts:23-62). Even pointed at /jobs, the shapes differ: page reads items (core returns data, jobs.controller.ts:42), row.id (lean rows have _id), topic / event (core has target / code), filter topic (core filters target). No Skip action

No DLQ page exists.

11. Tests

Run Jest with --maxWorkers=3. Specs that touch Mongo need a replica set (see 04 §11).

spec	covers
src/kafka/kafka.service.spec.ts	One producer over 1,000 sends; failed connect → next send reconnects; retries 0; failed send drops producer; connect-phase error; shutdown order (send in flight completes, disconnect at application shutdown); settings read + decrypted secrets; enabled false/true reconcile; broker change reconnect; change during reconcile; two-pod status + TTL; Redis restart; disabled at boot; heartbeat retries producer and consumers; test-connection secret merge
src/kafka/kafka.consumer.it.spec.ts	Real broker + Mongo (skipped without KAFKA_IT_*): disable stops an idle consumer within 2 s; re-enable consumes once; in-flight handler finishes on disable, commits once, no duplicate after re-enable
src/kafka/kafka_log.spec.ts	out + in rows share trace_id; log store down; 1.2 KB whole, 90 KB cut to 64 KB; thrown and returned failures write failed in rows; 60-day expiry; secrets masked
src/kafka/kafka.target.spec.ts	Key/header/topic; delivery-mode config; every error class; 10 min outage drain in order; takeover at-most-once / at-least-once; hung broker; stall past lease; invalid records; target owned by the Kafka module (details in 04 §11)
src/common/utils/kafka.spec.ts	authen_type → kafkajs ssl / sasl mapping, defaults
src/common/context/context.spec.ts:73	outbound send carries traceparent and trace_id
src/delivery_order/delivery_order.service.spec.ts:103-342	DO change → one job → one message; duplicate update; broker down during emit; old relay-shape rows ignored; wire contract (event_id header, trace_id, bytes); topic override; rollback; activity rows; same Kafka message on 2 pods; missing message_id

Untested: the claim paths busy / stale-lease retake / failed retake, DLQ upsert and resolved, claim release on a thrown handler or failed write, non-JSON payloads, delivery_no is required, channel not routed, SPO/OD routing, the store-v2 pages, the 403 on the Kafka controller.

12. Gaps and known issues

#	gap	code vs design	risk
G1	do_consume dead-letters transient errors. KafkaDoConsume._onDoConsume catches every error and returns it as a permanent reason (delivery_order.kafka_consumer.ts:26-32), so a Mongo blip, a transaction write conflict or a not-yet-existing store (store/store.service.ts:21-29 → not_found) goes to the DLQ and the offset is committed. The runner's throw → release → redeliver path (kafka.consumer.ts:156-162) is never used by DO	design r3: "retry after failure" via redelivery; reason strings were meant for permanent failures (kafka.types.ts:13)	An order is acknowledged and not applied until OMS resends the same message_id
G2	No DLQ replay, list or resolve, and kafka_consume_dlq has no TTL	07 doc: replay is designed (intake_message), not built	DLQ rows are visible only in Mongo; the only replay is a resend from the sender
G3	Hold when Kafka off is not built any more. KafkaTarget.send never checks enabled or the per-event enabled; the producer reconnects lazily on the next send (kafka.producer.ts:55-80,106-121). Per-event enabled only affects consumers (kafka.consumer.ts:66)	D-212 (built in b9513620 relay): Kafka off → hold rows; one topic disabled → drop. Lost when the relay was replaced (38dad930). Same as 04 G5	Turning Kafka off in the UI does not stop DO-update messages; the per-event toggle on the Kafka page is a no-op for producers
G4	store-v2 Outbound Events page is broken (route /outbound_events gone, shape mismatch, §10). No DLQ or status page either	admin UI parked (D-214); stale menu entry left	Admins hit an error page; DEAD rows only via /api/v1/jobs
G5	The DO-update bodies carry no event_id, version or message_id: meta is ignored by both builders (do-update.payload.ts:177-181). The id is only in the event_id header, and kafka_logs out rows have an empty message_id (kafka.producer.ts:135)	v1 field-set parity was the goal; r2 noted event_id header = message_id	Receivers that dedupe must read the header; log search by message id does not work for outbound
G6	Consume claim has no lease renewal: a handler > 5 min can run twice; after a crash a partition stalls up to 5 min	r3 L1/L2, accepted	Handlers must be idempotent (DO is, via event_key)
G7	3 producer events (load_update, task_update, driver_update) have targets but no caller	—	Topics exist only on paper
G8	DO emit points are "every create + every status change" for routed channels, not v1's measured emit table	07 doc open item 1	More or fewer messages than v1 during cut-over
G9	Kafka controller has no @Roles: any valid session can read pod status and make the server connect to arbitrary brokers through test-connection (kafka.controller.ts:4-19)	settings routes are admin-only	Low; a non-admin user can probe network hosts
G10	New consumer groups start at fromBeginning: false (kafka.consumer.ts:178); changing a consume topic_name keeps the same group on the new topic	—	Messages produced before the first start on a new topic/group are skipped
G11	No nack/retry budget inbound: a message that keeps throwing (only possible for non-DO handlers today) is redelivered forever; a lasting DLQ write fault also loops	r3 accepted low	Partition blocked until fixed
G12	GET /kafka/status uses Redis KEYS kafka:status:* (kafka.service.ts:129)	—	O(keyspace) on a shared Redis
G13	CENTRAL_MASK_KEYS contains key (common/utils/mask.ts:79), so any body field literally named key is masked in kafka_logs	—	Low; log readability

13. How to add the next one

A. A new outbound topic (business event → Kafka)

1. Declare it: add { event: '<event>', type: KafkaEventType.produce, default_topic: '<topic>' } to KAFKA_EVENTS (src/kafka/kafka.constants.ts:16-23). KafkaTarget.onModuleInit registers kafka:<topic> at-most-once (kafka.target.ts:60-68); the events setting default gains a row, so ops can override topic_name.
2. Write a payload builder next to the business module, pure and deterministic, copying src/delivery_order/do-update.payload.ts (shape table + _project). Put any id the receiver dedupes on in the body if it needs it (G5).
3. Enqueue inside the business transaction, copying delivery_order.service.ts:168-195: sideEffect.enqueueMany([{ target: kafkaTargetName(default_topic), code: '<event>', key: <entity no>, version, dedupe_key: '<event>:<key>:<version>', headers: { event_id }, payload }], tx). Use a per-entity counter like job_counters (delivery_order_version.schema.ts) for version. Import JobsModule and KafkaModule in your module (delivery_order.module.ts:29-31).
4. Pick the delivery mode: leave at-most-once unless the receiver dedupes on event_id; then set KAFKA_TARGET_TOPICS=<topic>:at_least_once.
5. Tests: copy delivery_order.service.spec.ts:103-235 (one change → one job → one message; duplicate; broker down; wire contract; topic override; rollback). Generic send behaviour is already covered by kafka.target.spec.ts.

B. A new inbound consumer

1. Declare it: { event: '<event>', type: KafkaEventType.consume, default_topic: '<topic>' } in KAFKA_EVENTS. The consumer starts only once a handler is registered and the event is enabled (kafka.consumer.ts:66); its group is <GROUP_PREFIX>.<event>.
2. Register the handler in a provider of the owning module, copying src/delivery_order/delivery_order.kafka_consumer.ts: onModuleInit() { this.kafka.registerConsumer('<event>', (m) => this._handle(m)) }.
3. Handler contract (kafka.types.ts:13-14):
   • resolve undefined when applied (or already applied);
   • resolve a reason string only for a permanent failure (bad payload, missing id) → DLQ, offset committed;
   • throw for a transient failure (DB, timeout, missing reference that will appear) → claims released, batch redelivered. Do not copy the catch-all in KafkaDoConsume (G1).
4. Make the apply idempotent per message_id: require it, and write an activity_logs row with event_key = <area>:<entity>:msg:<message_id> in the same transaction, then treat isAlreadyApplied as success (delivery_order.service.ts:197-223).
5. Keep the handler well under 5 min (claim lease, G6).
6. Tests: a service spec like delivery_order.service.spec.ts:286-327 (same message twice on 2 pods, missing id), plus a claim/DLQ spec for the runner paths that are untested today (§11), modelled on kafka.consumer.it.spec.ts.

C. A new Kafka setting

Add a defineSetting(KAFKA_SETTINGS, { key, schema, default, encrypt? }) in src/kafka/kafka.settings.ts, include it in KAFKA_SETTING_REFS, read it in KafkaConfigReader.get() (kafka.config.ts:19-32). If it changes connections, put it into the producer/consumer signature (kafka.producer.ts:34, kafka.consumer.ts:68) so reconcile rebuilds the client. Add the field to src/pages/setting/kafka/index.tsx (store-v2).