Worker Structure & Runtime Profiles (SVC)

Overview

line-management-worker-go is the background worker service written in Go, ported one-to-one from the original NestJS worker.

The entire repository compiles into a single binary (cmd/worker) that can take on 7 different roles, or profiles, selected at runtime through the SVC environment variable. Each profile registers its own set of consumers and cron jobs and connects to its own set of dependencies, which allows each one to be deployed as a separate pod sized to its actual workload — trigger-worker, for example, can scale independently of main.

All profiles share a single bootstrap path, internal/app.New: load configuration from environment variables, connect the dependencies the profile needs (Postgres / Redis / RabbitMQ), start the admin server for metrics and health checks, start the outage supervisor, then wait for a shutdown signal.

Business Flow

1. Select the profile — entrypoint.sh or start.sh sets SVC (defaulting to main). An unrecognised value causes the process to exit immediately.
2. Assign the admin port — when ADMIN_ADDR is not set, the admin port is assigned automatically per profile (:9104 through :9110) so that all 7 profiles do not contend for port :9100 when run on the same machine.
3. Bootstrap — app.New loads the configuration, validates it fail-fast via Validate(), connects the dependencies listed in depsFor(profile) within a 30-second timeout, and creates a namespaced RedisService.
4. Dispatch to the runner — each profile's runner creates an mq.Client, asserts the topology, assembles the services for every domain, and registers queue-to-handler pairs into a single shared mq.Registrar.
5. Wait for a signal — the main goroutine blocks on ctx.Done() (SIGTERM or SIGINT), then calls a.Shutdown(), which fails the readiness probe first, stops accepting new work, drains work in flight, and closes dependencies in reverse order.
6. Watch for outages — the supervisor monitors each dependency independently. On failure it retries every 5 seconds during the first minute and every 10 seconds thereafter, fires an alert to the chat webhook at the 1-minute mark, and at 3 minutes re-alerts and exits so that Kubernetes restarts the pod.

Profile summary

SVC	Responsibility	Dependencies	Admin port (default)
main	23 core delivery consumers plus the DLQ monitor	PG + Redis + Rabbit	:9104
trigger-worker	8 trigger-engine consumers plus a per-minute due-actions cron	PG + Redis + Rabbit	:9105
web-request-worker	Consumer for the web_request_execute queue	PG + Redis + Rabbit	:9106
bigquery-sync	Dynamic cron driven by DB config, syncing BigQuery into PG	PG only	:9107
knowledge-worker	Consumer for the knowledge_index queue	PG + Redis + Rabbit	:9108
pg-listener	Bridges PG LISTEN/NOTIFY into RabbitMQ and sweeps the outbox	PG + Rabbit	:9109
cron-scheduler	10 cron scanners and 4 consumers (mbox, booking)	PG + Redis + Rabbit	:9110

Key Files & Functions

• cmd/worker/main.go — main(), validProfiles, and all 7 runners: runMain, runPgListener, runTriggerWorker, runCronScheduler, runBigquerySync, runWebRequestWorker, runKnowledgeWorker
• cmd/worker/integration.go — every cross-domain adapter. Each package declares a narrow interface for the sibling services it calls, and this binary constructs the real implementations and injects them.
• internal/app/app.go — New(), depsFor(profile), Shutdown()
• internal/config/config.go — Config, Queues, Load(), Validate(), HealthPort()
• internal/supervisor/supervisor.go — the per-dependency outage state machine; evaluate is written as a pure function with unit-test coverage
• internal/health/health.go — the /livez, /readyz, and /healthz endpoints
• internal/obs/admin.go and internal/obs/metrics.go — the admin port serving /metrics, /version, and pprof (gated by PPROF_ENABLED)
• internal/logx/logx.go — structured JSON logging with a level that can be changed via SIGHUP
• internal/signalx/signalx.go — multiplexes SIGTERM/SIGINT for draining and SIGHUP for reloading

Connections to Other Services

• PostgreSQL (TYPEORM_*) — a shared pgx pool via internal/platform/pg; every domain uses raw SQL copied verbatim from the original TypeORM queries
• Redis / Dragonfly (REDIS_URL, REDIS_NAMESPACE) — caching, distributed locking, deduplication, and session state
• RabbitMQ (RABBITMQ_URL) — the primary broker, receiving jobs from webhook-go, cms-api-go, client-api-go, and from the pg-listener profile via pg_notify
• S3-compatible storage (STORAGE_*) — audience CSV files, import archives, and error logs
• LINE Messaging API (LINE_ENDPOINT) — a per-OA client built from line_oa.channel_access_token
• CMS API (CMS_API_BASE_URL, INTERNAL_API_KEY) — called back for audience refresh
• Meilisearch and the Embedding API (MEILISEARCH_*, EMBEDDING_API_URL) — knowledge indexing and AI responses
• SMTP (MAIL_SMTP_*) — password reset email
• Alert webhook (ALERT_WEBHOOK_URL, DLQ_ALERT_WEBHOOK_URL) — notifications to Google Chat or Slack