07 — Integrations and jobs

Everything TMS v2 sends to or receives from other systems, every background job and timer, and every log it keeps with how long it keeps it. Read this when you need to know "what talks to the outside, how does it retry, and what is left behind".

Built or designed. Every row is tagged:

• Built — in the code of bigc-tms-core-v2, branch main-v2-prototype, commit 3d70475a (worktree /Users/thanosorn/Works/Software/tms/wt-v2-prototype-core).
• Designed — only in the design documents (/Users/thanosorn/Works/Software/tms/agent-task/.work/tms-v2-discovery/modules/<module>/design.md, and the jobs and Kafka notes in /Users/thanosorn/Works/Software/tms/_docs/bigc-tms-core-v2/).

State machines for jobs, Kafka messages and attachments are in 04-states-and-flows.md.

---

1. Kafka

1.1 How the connection is configured (Built)

item	value	where
settings record	kafka in the central settings module: bootstrap_servers, authen_type (PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL), authen_setting, enabled, events (topic name and on/off per event); sasl_password and ssl_key_pem stored encrypted	src/kafka/kafka.settings.ts
change without deploy	each pod re-reads settings on a Redis settings:changed message and every 60 s, then reconnects producer and consumers if the config changed	kafka.service.ts _reconcile; settings.service.ts
consumer group	<KAFKA_GROUP_PREFIX or KAFKA_CLIENT_ID or bigc-tms-core-v2>.<event>, one group per consume event, starts at the latest offset	kafka.consumer.ts _start
producer	one long-lived producer per pod	kafka.producer.ts
pod status	each pod writes kafka:status:<pod> to Redis every 10 s (TTL 30 s); GET /v1/kafka/status lists them	kafka.service.ts
connection test	POST /v1/kafka/test-connection with an unsaved config, or the saved one	kafka.controller.ts
extra outbound topics	env KAFKA_TARGET_TOPICS="topic-a,topic-b:at_least_once" registers more send targets	kafka.target.ts parseTopics

1.2 Topics

dir	event	default topic	payload essentials	handler	idempotency key	status
in	do_consume	bigc.lastmiles.do-post	the OMS order: delivery_no, message_id, channel, source store, destination address and latlong, expected date and window, is_cod, cod_amount, items, optional status	KafkaDoConsume → DeliveryOrderService.upsertFromKafka (create if new delivery_no, else update)	consume claim on message_id (fallback topic:partition:offset), plus a unique activity event_key do:<delivery_no>:msg:<message_id> so a replay changes nothing	Built
out	do_update_oms	lastmiles.oms.do-update	v1 field set and order: delivery_no, external_delivery_no, order_type, dates, chanel, source and destination address, is_cod, cod_amount, items, Status	jobs fast lane → KafkaTarget.send; Kafka key = delivery_no; headers event_id, traceparent, trace_id	job dedupe_key do_update_oms:<delivery_no>:<version>; version from a per-DO counter	Built for channels DONJAI, CHOKCHAI, HLH
out	do_update_staffapp	lastmiles.staffapp.do-update	v1 field set: as above plus delivery_type, invoice, pickup_info, POD, payment proof and driver blocks when present, status, note	same	do_update_staffapp:<delivery_no>:<version>	Built for channels SPO, OD, HLH
out	load_update	lastmiles.staffapp.load-update	—	target registered, nothing enqueues to it	—	Declared in code, no producer (v2 has no load entity)
out	task_update	lastmiles.staffapp.task-update	task state for staffapp	target registered, nothing enqueues to it	—	Declared in code, no producer; Designed with execution
out	driver_update	lastmiles.staffapp.driver-update	driver state for staffapp	target registered, nothing enqueues to it	—	Declared in code, no producer; Designed with workforce

When a DO message goes out (Built). One row per routed topic on every DO create (admin or Kafka), and on every update whose status differs from before. A channel not in either list sends nothing and logs a warning. The full payload mapping against v1 samples is parked; field shapes today follow src/delivery_order/do-update.payload.ts.

When it should go out (Designed). Only where v1 emits today, measured per trigger from PROD kafka_logs and mapped onto each v2 DO transition; both DO topics carry the same events during the store-by-store cut-over.

1.3 Inbound mechanics (Built)

step	what happens
claim	per message, an upsert on kafka_consumed by consume_id; a done row skips it, a processing row younger than 5 min makes the batch stop and re-fetch from that offset after 5 s
handle	each message runs in its own request context carrying the producer's traceparent
permanent failure	the handler returns a reason (for example missing message_id): row upserted into kafka_consume_dlq (pending, attempts + 1), claim set failed, offset committed
crash or thrown error	claims of the batch are deleted and the error is rethrown, so Kafka redelivers the batch
success	claim set done; any kafka_consume_dlq row for the id set resolved
log	one kafka_logs row per handled message (direction in, body masked and capped)
replay	none; a DLQ row is applied again only if Kafka delivers the same id again. Designed: intake_message park-and-replay (see 04-states-and-flows.md §8a)

1.4 Outbound mechanics (Built)

Outbound Kafka uses an outbox: the business change and its jobs rows are written in one Mongo transaction, so a committed change always has its message and a rolled-back one has none.

rule	value
order per DO	rows partitioned by hash of delivery_no (64 partitions); the pod that leases a partition sends its rows sorted by key and version; a retrying row holds later versions of the same DO on the same topic
delivery mode	at-most-once for every declared topic (a send with no answer goes DEAD for a person to retry or skip, never resent automatically); at-least-once can be set per topic through KAFKA_TARGET_TOPICS
failure classes	broker refused the record → retry with backoff; broker unreachable → pause the topic 30 s, attempts unchanged; no answer or broker timeout → DEAD on at-most-once; invalid record → DEAD at once
headers	event_id (job id), traceparent, trace_id
log	one kafka_logs row per send (direction out)

---

2. Other external systems

system	operation	when	timeout	retry	logged	status
Google Route Optimization	POST https://routeoptimization.googleapis.com/v1/projects/<id>:optimizeTours (avoid tolls and highways, road traffic on, polylines on)	plan re_plan, move_stop, assign_stop, unassign_stop, inside the HTTP request	120 s	none; the caller gets the error and a failure row in plan_history	google_logs (auth header masked)	Built
Google OAuth token	service-account token, scope cloud-platform	before each optimise call	30 s	none	google_logs	Built
Google Maps key	public endpoint GET /v1/settings/public returns google_map_api_key to clients; the server makes no Maps call	client maps	—	—	no	Built
Google optimise as a background job	same API, called from a long-lane job	optimise and generate	—	jobs engine retry	google_logs	Designed
AWS S3	presigned PUT (5 min) to v2/incoming/<uuid>, HeadObject, ranged GetObject of the first 4 KB (magic bytes), CopyObject to v2/files/<uuid>, DeleteObject, presigned GET (15 min)	attachment upload, confirm, sweep, read	10 s per call (S3_TIMEOUT_MS)	none per call; a stuck scan is retried by the sweep	app log only (no DB log)	Built
Storage config	bucket, region, prefix, KMS, credential mode IAM role or static key; secret encrypted	admin Storage setting	—	—	—	Built
Malware scanner	scan(bucket, key)	attachment confirm	—	—	app log	Built as a stub that always answers clean
Redis	settings change pub/sub, Kafka pod status, Bull queues (side_effect_lossy, attachment_sweep), sessions	always	—	—	—	Built
CloudFront for public files	stable URL for files whose purpose is public	attachment read	—	—	—	Designed (only purpose admin_file, private, is built)
SMTP email	account activation, staff PIN, other templates	through a jobs-backed notification outbox	—	5 / 15 / 30 min, then FAILED	—	Designed
SMS gateway	OTP	Redis/Bull (may be lost, user asks again)	—	—	—	Designed
Driver push	task offers and changes	Redis/Bull lossy side effect	—	none	—	Designed

Every outbound call goes through one externalCall helper that sets the timeout (src/common/utils/external_call.ts). Writing a DB log is opt-in per module: today Google and Kafka have one, S3 does not.

---

3. Jobs engine (Built)

One engine runs both queued side effects and schedule slots. Source: src/jobs/, src/schedules/; walkthrough: tms-v2-jobs-engine-code-walkthrough.md.

part	how it works	numbers
jobs collection	must-not-lose work rows: PENDING, RUNNING, DEAD; a finished row is deleted, not marked done	unique dedupe_key; partial indexes on PENDING and RUNNING
fast lane	short handlers (Kafka sends). A pod leases whole partitions and drains them with no per-row claim write	64 partitions in job_partitions; lease 30 s, heartbeat 10 s; batch ≤ 100
long lane	long handlers (schedule slots). Batch reserve with a claim token; exclusive targets run one at a time per code across all pods	batch ≤ 10; lease 5 min, heartbeat 1 min
poll	full batch → poll again at once; partial → 1 s; empty → 1 s, 2 s, then 5 s	—
retry	answered failure → wait 1, 2, 4, 8, 16 min, then DEAD (6 runs); per-code override (schedules use their retry_count as the run limit)	—
unreachable target	paused 30 s per pod, one probe row let through, attempts not counted	—
takeover	a row in flight when its partition is taken over goes DEAD on an at-most-once target, runs again on an at-least-once one	—
DEAD	logged as job dead alert; holds nothing; admin POST /v1/jobs/:id/retry (back to PENDING, attempts 0) or POST /v1/jobs/:id/skip (row deleted); GET /v1/jobs lists rows	no TTL
lossy side effects	a target marked lossy goes to Bull side_effect_lossy after commit, one try, dropped on failure	no lossy target is registered today
pods that only enqueue	env JOBS_WORKERS=off turns off lanes and the schedule tick on that pod	—
trace of finished work	one structured app-log line per run: code, target, dedupe_key, attempts, duration, result	not stored in Mongo

3.1 Schedules (Built engine, no schedule declared)

item	value
declare	a module calls schedules.declare({ code, cron, handler, missed_slot_policy?, retry_count? }) in onModuleInit; code owns cron, policy and retry count
config row	schedules collection: code, cron, enabled, cron_override, missed_slot_policy, retry_count (default 3), last_run_at, last_skipped_at, next_run_at
admin	GET /v1/schedules, POST /v1/schedules/:code/enable and /disable, PUT and DELETE /v1/schedules/:code/cron_override
tick	every 15 s each pod enqueues every due slot as a long-lane job with dedupe_key <code>:<slot ISO time>; the unique key makes concurrent pods insert one row
time zone	crons fire in UTC+7 (Asia/Bangkok, no DST)
missed slots	a slot more than 1 min late is missed. Policy run_once (default) runs the newest missed slot once and skips older ones; skip (default for crons every 5 min or less) skips them; recorded in last_skipped_at
overlap	one run at a time per schedule code across pods (exclusive target)
declared schedules	none in production code; demo schedules live only in schedules.service.spec.ts

3.2 Every job and timer

code	trigger	what it does	retry / missed-slot policy	status
kafka:lastmiles.oms.do-update	DO create, DO status change	sends do_update_oms	fast lane, 1-2-4-8-16 min then DEAD; at-most-once	Built
kafka:lastmiles.staffapp.do-update	same	sends do_update_staffapp	same	Built
kafka:lastmiles.staffapp.load-update, …task-update, …driver-update	nothing enqueues yet	send target only	same	Built target, no producer
schedule:<code>	schedule tick	runs a declared schedule handler for one slot	long lane, exclusive; retry_count runs (default 3), backoff as above; missed slots per policy	Built engine, nothing declared
attachment_sweep	Bull repeat every 60 s (not the jobs engine)	fails pending attachments older than 6 min and deletes their incoming object; re-runs scans stuck in scanning 5 min; ≤ 100 rows each	next run in 60 s; skipped while storage is not configured	Built
Kafka reconcile and status heartbeat	every 10 s per pod	rewrites pod status in Redis; reconnects a failed producer or consumer	next beat	Built
settings reload	every 60 s per pod, plus Redis settings:changed	reloads setting values into the pod cache	next interval	Built
buffered log flush	every 1 s or 100 rows per log per pod	inserts buffered kafka_logs and google_logs rows; past 1,000 rows the oldest is dropped	dropped rows counted and printed	Built
fast lane poll / partition heartbeat	adaptive 0–5 s / 10 s	sends due rows; renews leases; closes taken-over rows	—	Built
long lane poll / lease heartbeat	adaptive 0–5 s / 1 min	runs schedule slots; renews leases	—	Built
plan optimise job	Optimise pressed	runs Google optimise outside the request, plan optimising → planned or failed	lease; retry from admin	Designed
plan generate job	Generate pressed	per open route one transaction: task, stops, DO claim (ASSIGNED), duty window	lease lapses → retry resumes at first open route	Designed
end-of-date claim release	end of the plan date, store local time (mechanism not specified)	clears the plan claim on orders still unassigned, so the next day's plan can pick them	—	Designed (assumption)
task overdue cancel	scheduled job	cancels tasks still PENDING or ACCEPTED 24 h after their date (task_start_grace_hours); DOs stay ASSIGNED on their route	run_once	Designed (needs customer confirm)
task overdue list	running task 24 h past its date (mechanism not specified)	puts the task on an ops overdue list; ops runs close-out for the driver	—	Designed
workforce day close	every 15 min	ends duty records 2 h after their last window: OFF_DUTY auto-closed or ABSENT; bookings CLOSED	re-run safe (each record closed once); engine default for a 15-min cron is run_once	Designed
on-demand acceptance timeout	per offered task, Redis/Bull timer	when the driver does not accept, hold the driver (duty HELD) and offer the task to another driver	lossy; the timer job retries the hold command	Designed
notification send	jobs-backed outbox row written with the business change	renders and sends email / SMS	5 / 15 / 30 min then FAILED	Designed
export run	jobs row, status kept in export_runs	builds an export file in batches, stores it as an attachment	jobs engine	Designed
intake replay	ops presses replay, or the missing store / channel appears	replays parked intake_message rows per order, oldest first	per row: back to PARKED or DEAD with replay_count + 1	Designed
DEAD job clean-up	TTL index on closed_at	removes DEAD rows after 30 days	—	Designed (assumed decision)

---

4. Logs and retention

Each log declares its default retention in code; the value lives in the central setting log.<key> (1–365 days) and can be changed in admin without a deploy. Every row stores expire_at = write time + retention, with a TTL index on expire_at, so a change applies to new rows only. Source: src/log/.

log	collection	one row per	write mode	default retention	status
activity log	activity_logs	business action on a record (do-create, do-update, RE_PLAN, …): actor, channel, reason, changed fields, unique event_key	in the business transaction; a duplicate event_key aborts it (this is how a replayed Kafka message becomes a no-op)	60 days, one value, no per-entity override	Built
Kafka log	kafka_logs	message sent (out) or handled (in): topic, key, message_id, partition, offset, masked bodies, status sent / consumed / failed	buffered (1 s / 100 rows)	60 days	Built
Google log	google_logs	Google call: operation, url (masked), status, outcome OK / ERROR / TIMEOUT, duration, masked headers and bodies	buffered	30 days	Built
plan history	plan_history	plan save or failed optimise: action, revision, diff, duration	in the plan transaction	no TTL	Built
Kafka consume claim	kafka_consumed	consumed message id	per message	7 days (fixed TTL on consumed_at)	Built
Kafka dead letters	kafka_consume_dlq	failed inbound message with raw payload and error	per message	no TTL	Built
DEAD jobs	jobs (status DEAD)	job that used up its tries, with full payload	—	no TTL; 30 days designed	Built (TTL Designed)
finished jobs	app log line only	job run	stdout	log platform's retention	Built
app log	stdout (winston)	everything else, with trace_id	—	outside TMS	Built
S3 calls	app log only	—	—	—	Built (no DB log by choice)
exception log	—	—	—	parked until ops confirm New Relic on PROD	Designed (on hold)
GPS points	—	driver position	—	90 days	Designed (needs customer confirm)
delivery_order_revision	collection declared	one placement of a DO on a task	—	—	Built schema, no writer

Masking: Authorization, cookies, password, secret, token, api_key, key, pin, otp, client_secret in headers, bodies at any depth and URL query; bodies are size-capped with a truncated flag and the original size.

---

5. Where code and design disagree

#	area	code (Built)	design	effect
1	DO emit points	every create, and every update that changes status, for routed channels	only where v1 emits, measured from PROD kafka_logs per trigger	may send more or fewer messages than v1 until the emit table is applied
2	Inbound replay	dead-letter row, no replay	intake_message park, ordered replay, WAITING_ON_ORDER	an order message failing on a missing store stays failed until Kafka redelivers it
3	Plan optimise	synchronous Google call in the HTTP request, 120 s	background long-lane job	the request blocks for the whole optimise
4	Schedule retry	retry_count 3 = 3 runs in total	"retries up to 3 times" = 4 runs	one fewer retry
5	DEAD jobs	kept forever	TTL 30 days after closed_at (assumed)	unbounded growth
6	Attachment sweep	Bull repeat every 60 s, outside the jobs engine	"a sweep" (engine not named); every other periodic job is designed on the schedules engine	runs on every pod's Bull worker, not visible in /v1/schedules
7	Plan history	own plan_history collection with no TTL, plus an activity row	planning keeps plan_history; platform moves plan rows into activity_logs	two stores; plan_history never expires
8	Declared topics	load_update, task_update, driver_update registered as send targets	v2 has no load entity; task and driver messages come with execution and workforce	dead targets until those modules exist
9	Malware scan	stub, always clean	real scan before a file is ready	every file is marked clean