feat: introduce Temporal as workflow orchestration engine#1
Conversation
- Temporal server (auto-setup) + UI in docker-compose, reusing Postgres, namespace company-brain with 7d retention - packages/workflows: deterministic workflow definitions (hello, health-check, storage) with signals, queries, retry policies and task-queue/type constants - packages/activities: side-effecting activity implementations (printMessage, checkServices, uploadFile) over long-lived clients - services/temporal-worker: worker host with typed config, connection retry, health endpoint on :4100 and graceful shutdown - API: TemporalService client wrapper, temporal in /health report, demo workflow routes under /api/v1/workflows (Bearer auth) - docs/temporal.md developer guide; README updates
📝 WalkthroughWalkthroughTemporal workflow infrastructure is added across the monorepo, including workflow and activity packages, a Temporal worker, API endpoints and health checks, Docker services, environment configuration, dashboard support, and documentation. ChangesTemporal integration
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant Client
participant API
participant TemporalService
participant TemporalServer
participant TemporalWorker
participant Activities
Client->>API: submit workflow request
API->>TemporalService: start or execute workflow
TemporalService->>TemporalServer: send workflow command
TemporalServer->>TemporalWorker: dispatch workflow task
TemporalWorker->>Activities: run side-effect activity
Activities-->>TemporalWorker: return activity result
TemporalWorker-->>TemporalServer: complete workflow task
TemporalServer-->>API: return workflow result or status
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Warning There were issues while running some tools. Please review the errors and either fix the tool's configuration or disable the tool if it's a critical failure. 🔧 ESLint
ESLint install failed: private package registry requires authentication. Disable ESLint in CodeRabbit settings or use public packages. Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 9
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@apps/api/src/services/temporal.service.ts`:
- Around line 125-129: Update TemporalService.close() to await any in-flight
this.connecting promise before closing the client, and ensure the pending
connection cannot be installed after shutdown. Preserve cleanup of this.client,
this.connecting, and the existing connection-close behavior.
In `@docs/temporal.md`:
- Around line 145-149: Update the curl examples for helloWorkflow and
storageWorkflow in the temporal documentation to include the required
Authorization Bearer header, matching the documented authentication format so
both copied requests are authenticated.
- Around line 151-154: Update the curl examples in the workflow API section to
define a WORKFLOW_ID placeholder and use "${WORKFLOW_ID}" within quoted URLs for
the status, skip, and workflow-detail requests, avoiding shell redirection
parsing.
- Line 30: Update the fenced code block in docs/temporal.md to include an
explicit language identifier, such as text or ascii, so it satisfies
markdownlint MD040.
In `@packages/activities/src/activities.ts`:
- Around line 77-80: Update the fetch call inside the Qdrant readiness probe to
pass an AbortSignal with an appropriate timeout, ensuring hung requests abort
and the probe returns a down status instead of waiting indefinitely.
- Around line 43-46: Update the connection callback in socket.connect to call
socket.destroy() instead of socket.end() after a successful connection, while
preserving the existing resolve() behavior.
In `@packages/activities/src/context.ts`:
- Around line 46-50: Update the Redis client initialization in the context setup
to connect eagerly during worker boot, ensuring lazyConnect does not buffer the
first Redis operation in ioredis’s offline queue. Preserve maxRetriesPerRequest:
2 and verify startup surfaces connection failures before activities begin; only
disable the offline queue if this eager startup connection is implemented.
In `@services/temporal-worker/src/config.ts`:
- Around line 13-16: Remove the duplicated Temporal worker health endpoint
defaults by establishing one shared source of truth for the port and deriving
the API health URL from it. Update services/temporal-worker/src/config.ts lines
13-16 and apps/api/src/config/env.ts lines 49-53 so both services use the same
TEMPORAL_WORKER_HEALTH_PORT value, with the API constructing the health URL
rather than maintaining an independent full-URL default.
In `@services/temporal-worker/src/health-server.ts`:
- Around line 16-45: Register an error listener on the HTTP server created in
startHealthServer before calling server.listen. Handle listen-time errors
through the provided logger, including the error details, so health-server
failures do not become unhandled EventEmitter errors that crash the worker.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 4caf2ae3-ceae-4cf0-af80-6bcc4e393fcf
⛔ Files ignored due to path filters (1)
pnpm-lock.yamlis excluded by!**/pnpm-lock.yaml
📒 Files selected for processing (41)
.env.exampleREADME.mdapps/api/package.jsonapps/api/src/app.tsapps/api/src/config/env.tsapps/api/src/config/index.tsapps/api/src/modules/health/health.routes.tsapps/api/src/modules/workflows/workflow.routes.tsapps/api/src/modules/workflows/workflow.schemas.tsapps/api/src/plugins/services.tsapps/api/src/services/temporal.service.tsapps/web/src/app/(app)/dashboard/page.tsxdocs/temporal.mdinfrastructure/docker/docker-compose.ymlpackage.jsonpackages/activities/eslint.config.mjspackages/activities/package.jsonpackages/activities/src/activities.tspackages/activities/src/context.tspackages/activities/src/index.tspackages/activities/tsconfig.jsonpackages/types/src/api.tspackages/workflows/eslint.config.mjspackages/workflows/package.jsonpackages/workflows/src/constants.tspackages/workflows/src/definitions.tspackages/workflows/src/index.tspackages/workflows/src/retry-policies.tspackages/workflows/src/workflows/health-check.workflow.tspackages/workflows/src/workflows/hello.workflow.tspackages/workflows/src/workflows/storage.workflow.tspackages/workflows/tsconfig.jsonpnpm-workspace.yamlservices/temporal-worker/eslint.config.mjsservices/temporal-worker/package.jsonservices/temporal-worker/src/config.tsservices/temporal-worker/src/connection.tsservices/temporal-worker/src/health-server.tsservices/temporal-worker/src/index.tsservices/temporal-worker/tsconfig.build.jsonservices/temporal-worker/tsconfig.json
| async close(): Promise<void> { | ||
| if (this.client) await this.client.connection.close(); | ||
| this.client = null; | ||
| this.connecting = null; | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
close() doesn't wait for an in-flight connection attempt.
If close() runs while getClient()'s this.connecting promise is still pending, this.client is still null so nothing is closed here — but the pending promise's .then() will later set this.client to a fresh connection that's never closed, leaking a gRPC connection past shutdown.
🩹 Proposed fix
async close(): Promise<void> {
- if (this.client) await this.client.connection.close();
+ if (this.connecting) {
+ await this.connecting.catch(() => undefined);
+ }
+ if (this.client) await this.client.connection.close();
this.client = null;
this.connecting = null;
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| async close(): Promise<void> { | |
| if (this.client) await this.client.connection.close(); | |
| this.client = null; | |
| this.connecting = null; | |
| } | |
| async close(): Promise<void> { | |
| if (this.connecting) { | |
| await this.connecting.catch(() => undefined); | |
| } | |
| if (this.client) await this.client.connection.close(); | |
| this.client = null; | |
| this.connecting = null; | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@apps/api/src/services/temporal.service.ts` around lines 125 - 129, Update
TemporalService.close() to await any in-flight this.connecting promise before
closing the client, and ensure the pending connection cannot be installed after
shutdown. Preserve cleanup of this.client, this.connecting, and the existing
connection-close behavior.
|
|
||
| ## Architecture | ||
|
|
||
| ``` |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Specify the fenced-block language.
markdownlint reports MD040 for this fence. Use a language such as text or ascii.
Proposed fix
-```
+```text📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| ``` |
🧰 Tools
🪛 markdownlint-cli2 (0.23.0)
[warning] 30-30: Fenced code blocks should have a language specified
(MD040, fenced-code-language)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@docs/temporal.md` at line 30, Update the fenced code block in
docs/temporal.md to include an explicit language identifier, such as text or
ascii, so it satisfies markdownlint MD040.
Source: Linters/SAST tools
| ```bash | ||
| # start helloWorkflow (returns workflowId) | ||
| curl -X POST localhost:4000/api/v1/workflows/hello \ | ||
| -H 'content-type: application/json' \ | ||
| -d '{"name":"Ada"}' |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Include the required Bearer header in every authenticated example.
The section states that Bearer authentication is required, but the helloWorkflow and storageWorkflow requests omit the authorization header; copied commands will return 401.
Also applies to: 157-160
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@docs/temporal.md` around lines 145 - 149, Update the curl examples for
helloWorkflow and storageWorkflow in the temporal documentation to include the
required Authorization Bearer header, matching the documented authentication
format so both copied requests are authenticated.
| # query its phase / signal it to finish early / describe it | ||
| curl localhost:4000/api/v1/workflows/<workflowId>/status -H "authorization: Bearer $TOKEN" | ||
| curl -X POST localhost:4000/api/v1/workflows/<workflowId>/skip -H "authorization: Bearer $TOKEN" | ||
| curl localhost:4000/api/v1/workflows/<workflowId> -H "authorization: Bearer $TOKEN" |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Use a shell-safe workflow ID placeholder.
Unquoted <workflowId> is parsed by the shell as redirection syntax. Define WORKFLOW_ID and use ${WORKFLOW_ID} in quoted URLs instead.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@docs/temporal.md` around lines 151 - 154, Update the curl examples in the
workflow API section to define a WORKFLOW_ID placeholder and use
"${WORKFLOW_ID}" within quoted URLs for the status, skip, and workflow-detail
requests, avoiding shell redirection parsing.
| socket.connect(port, host, () => { | ||
| socket.end(); | ||
| resolve(); | ||
| }); |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Use socket.destroy() to immediately release resources.
Using socket.end() performs a graceful half-close (sending a FIN) but leaves the socket and its associated timeout active until the remote peer also closes the connection. For a simple TCP health probe, immediately destroying the socket upon successful connection is safer as it guarantees all timers are cleared and resources are freed without waiting.
🔧 Proposed fix
socket.connect(port, host, () => {
- socket.end();
+ socket.destroy();
resolve();
});📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| socket.connect(port, host, () => { | |
| socket.end(); | |
| resolve(); | |
| }); | |
| socket.connect(port, host, () => { | |
| socket.destroy(); | |
| resolve(); | |
| }); |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@packages/activities/src/activities.ts` around lines 43 - 46, Update the
connection callback in socket.connect to call socket.destroy() instead of
socket.end() after a successful connection, while preserving the existing
resolve() behavior.
| probe(async () => { | ||
| const response = await fetch(`${ctx.config.qdrantUrl}/readyz`); | ||
| if (!response.ok) throw new Error(`qdrant readyz returned ${response.status}`); | ||
| }), |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Add a timeout to the fetch request.
Node's fetch does not have a default timeout. If the Qdrant server accepts the connection but hangs without sending a response, this probe will wait indefinitely, leaking a promise and a connection in the background. Supplying an AbortSignal ensures the probe fails fast and accurately returns a 'down' status.
🔧 Proposed fix
- probe(async () => {
- const response = await fetch(`${ctx.config.qdrantUrl}/readyz`);
- if (!response.ok) throw new Error(`qdrant readyz returned ${response.status}`);
- }),
+ probe(async () => {
+ const response = await fetch(`${ctx.config.qdrantUrl}/readyz`, {
+ signal: AbortSignal.timeout(5000),
+ });
+ if (!response.ok) throw new Error(`qdrant readyz returned ${response.status}`);
+ }),📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| probe(async () => { | |
| const response = await fetch(`${ctx.config.qdrantUrl}/readyz`); | |
| if (!response.ok) throw new Error(`qdrant readyz returned ${response.status}`); | |
| }), | |
| probe(async () => { | |
| const response = await fetch(`${ctx.config.qdrantUrl}/readyz`, { | |
| signal: AbortSignal.timeout(5000), | |
| }); | |
| if (!response.ok) throw new Error(`qdrant readyz returned ${response.status}`); | |
| }), |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@packages/activities/src/activities.ts` around lines 77 - 80, Update the fetch
call inside the Qdrant readiness probe to pass an AbortSignal with an
appropriate timeout, ensuring hung requests abort and the probe returns a down
status instead of waiting indefinitely.
| // Do not queue commands forever while Redis is down — activities should | ||
| // fail fast and let Temporal's retry policy handle it. | ||
| maxRetriesPerRequest: 2, | ||
| lazyConnect: true, | ||
| }); |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Description: Test ioredis lazyConnect and enableOfflineQueue interaction.
npm install ioredis --no-save > /dev/null 2>&1
node -e "
const { Redis } = require('ioredis');
const redis = new Redis({
host: '127.0.0.1',
port: 9999, // Unreachable port to simulate offline
maxRetriesPerRequest: 2,
enableOfflineQueue: false,
lazyConnect: true,
});
redis.ping().then(() => console.log('Ping success')).catch(err => {
console.error('Failure:', err.message);
process.exit(0);
});
"Repository: shibadityadeb/Context-Brain
Length of output: 157
🏁 Script executed:
set -euo pipefail
printf '\n## context.ts\n'
sed -n '1,140p' packages/activities/src/context.ts
printf '\n## dependency metadata\n'
git ls-files | rg '(^|/)(package.json|package-lock.json|pnpm-lock.yaml|yarn.lock)$'
if [ -f package.json ]; then
printf '\n### root package.json (ioredis refs)\n'
rg -n '"ioredis"|enableOfflineQueue|lazyConnect|maxRetriesPerRequest' package.json packages -g 'package.json' -g '*.ts' -g '*.js'
fi
printf '\n### files mentioning ioredis\n'
rg -n 'ioredis|enableOfflineQueue|lazyConnect|maxRetriesPerRequest' . -g '!**/node_modules/**'Repository: shibadityadeb/Context-Brain
Length of output: 4234
🌐 Web query:
ioredis enableOfflineQueue lazyConnect maxRetriesPerRequest offline queue documentation
💡 Result:
In ioredis, these options control how the client handles connection states and command execution [1][2]. enableOfflineQueue (default: true) By default, if you send a command before the connection to the Redis server is established or "ready," ioredis adds that command to an internal "offline queue" and executes it automatically once the connection becomes ready [1][3]. If you set enableOfflineQueue to false, any command sent while the connection is not ready will immediately return an error [1][4]. lazyConnect (default: false) By default, an ioredis instance attempts to establish a connection to the Redis server immediately upon initialization [5][6]. When lazyConnect is set to true, this automatic connection attempt is disabled [5]. The client will instead wait until the first command is sent or until you manually call redis.connect [5][6]. Calling redis.connect returns a promise that resolves when the connection is established or rejects if it fails [5][6]. maxRetriesPerRequest (default: 20) This option manages how commands are handled when a connection is lost during execution [2]. If a command does not receive a reply because the connection was lost, ioredis queues it to be resent upon reconnection (provided the retryStrategy allows it) [2]. The maxRetriesPerRequest option limits the number of reconnection attempts allowed for these queued commands before they are flushed and rejected with a MaxRetriesPerRequestError [2]. You can set this to null to allow commands to wait indefinitely for the connection to be restored [2].
Citations:
- 1: https://redis.github.io/ioredis/interfaces/CommonRedisOptions.html
- 2: https://github.com/redis/ioredis/blob/623cee51/lib/redis/RedisOptions.ts
- 3: https://ioredis.readthedocs.io/en/stable/README/
- 4: https://ioredis.readthedocs.io/en/latest/API/
- 5: https://github.com/redis/ioredis/blob/4f167bb/lib/redis/RedisOptions.ts
- 6: https://github.com/redis/ioredis/blob/623cee51/docs/interfaces/CommonRedisOptions.html
🌐 Web query:
site:github.com/redis/ioredis "enableOfflineQueue" "lazyConnect" first command
💡 Result:
In the ioredis library, lazyConnect and enableOfflineQueue both influence how the client handles commands when it is not yet connected to a Redis server [1][2]. lazyConnect: When set to true, the client delays establishing a connection to the Redis server until either the first command is sent or the redis.connect method is called explicitly [1][2]. By default (false), the client attempts to connect immediately upon initialization [1][2]. enableOfflineQueue: When set to true (the default), commands sent while the client is not connected are added to an internal queue and executed automatically once the connection is established and the client is ready [1][2]. If this is set to false, commands sent while the connection is not ready will immediately return an error [1][2]. Interaction: When both are used, the "first command" mentioned in the lazyConnect description acts as the trigger to initiate the connection [2]. If enableOfflineQueue is also true, that first command (and any subsequent commands sent before the connection is established) will be queued until the connection is successfully opened [2]. If enableOfflineQueue is false and you send a command before calling redis.connect, the command will fail because the client is not connected and is not permitted to queue the request [2].
Citations:
- 1: https://github.com/redis/ioredis/blob/623cee51/docs/interfaces/CommonRedisOptions.html
- 2: https://github.com/redis/ioredis/blob/623cee51/lib/redis/RedisOptions.ts
🏁 Script executed:
set -euo pipefail
rg -n 'lazyConnect:\s*true|enableOfflineQueue:\s*false|maxRetriesPerRequest:\s*2' packages . -g '!**/node_modules/**'Repository: shibadityadeb/Context-Brain
Length of output: 425
🏁 Script executed:
set -euo pipefail
rg -n "createActivityContext\(" packages services -g '!**/node_modules/**'Repository: shibadityadeb/Context-Brain
Length of output: 393
Fail fast on Redis startup maxRetriesPerRequest: 2 still leaves ioredis’ offline queue in place. With lazyConnect: true, the first Redis call will be buffered unless you connect during worker boot; if the goal is immediate activity failure, connect eagerly (or disable the offline queue only with that startup path in place).
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@packages/activities/src/context.ts` around lines 46 - 50, Update the Redis
client initialization in the context setup to connect eagerly during worker
boot, ensuring lazyConnect does not buffer the first Redis operation in
ioredis’s offline queue. Preserve maxRetriesPerRequest: 2 and verify startup
surfaces connection failures before activities begin; only disable the offline
queue if this eager startup connection is implemented.
| TEMPORAL_ADDRESS: z.string().default('localhost:7233'), | ||
| TEMPORAL_NAMESPACE: z.string().default('company-brain'), | ||
| TEMPORAL_TASK_QUEUE: z.string().default('brain-core'), | ||
| TEMPORAL_WORKER_HEALTH_PORT: z.coerce.number().int().positive().default(4100), |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Worker health endpoint value duplicated across two services with no shared source of truth. The health port/URL that gates the API's Temporal status reporting is independently defaulted in both services; nothing enforces they stay consistent.
services/temporal-worker/src/config.ts#L13-L16: this is whereTEMPORAL_WORKER_HEALTH_PORT(default4100) is actually bound byhealth-server.ts.apps/api/src/config/env.ts#L49-L53:TEMPORAL_WORKER_HEALTH_URL(defaulthttp://localhost:4100/health) must be kept manually in sync with the port above; consider deriving one from the other (e.g. document the coupling in.env.example, or have the API construct the URL from a sharedTEMPORAL_WORKER_HEALTH_PORTenv var instead of a separate full URL) to remove the drift risk.
📍 Affects 2 files
services/temporal-worker/src/config.ts#L13-L16(this comment)apps/api/src/config/env.ts#L49-L53
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@services/temporal-worker/src/config.ts` around lines 13 - 16, Remove the
duplicated Temporal worker health endpoint defaults by establishing one shared
source of truth for the port and deriving the API health URL from it. Update
services/temporal-worker/src/config.ts lines 13-16 and
apps/api/src/config/env.ts lines 49-53 so both services use the same
TEMPORAL_WORKER_HEALTH_PORT value, with the API constructing the health URL
rather than maintaining an independent full-URL default.
| export function startHealthServer(source: WorkerStatusSource, logger: Logger): Server { | ||
| const startedAt = Date.now(); | ||
|
|
||
| const server = createServer((request, response) => { | ||
| if (request.url !== '/health') { | ||
| response.writeHead(404, { 'content-type': 'application/json' }); | ||
| response.end(JSON.stringify({ error: 'not found' })); | ||
| return; | ||
| } | ||
| const workerState = source.getWorkerState(); | ||
| const connection = source.getConnectionStatus(); | ||
| const healthy = workerState === 'RUNNING' && connection === 'connected'; | ||
| response.writeHead(healthy ? 200 : 503, { 'content-type': 'application/json' }); | ||
| response.end( | ||
| JSON.stringify({ | ||
| status: healthy ? 'healthy' : 'degraded', | ||
| worker: workerState, | ||
| connection, | ||
| namespace: config.temporal.namespace, | ||
| taskQueue: config.temporal.taskQueue, | ||
| uptimeSeconds: Math.round((Date.now() - startedAt) / 1000), | ||
| }), | ||
| ); | ||
| }); | ||
|
|
||
| server.listen(config.temporal.healthPort, () => | ||
| logger.info({ port: config.temporal.healthPort }, 'worker health endpoint listening'), | ||
| ); | ||
| return server; | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Missing error handler on the health HTTP server.
server.listen() is called with no error listener registered on the server. If the health port is already in use (or any other listen-time error occurs), Node will throw the unhandled 'error' event and crash the entire worker process — not just the health check — since http.Server is an EventEmitter and unhandled 'error' events throw.
🩹 Proposed fix
+ server.on('error', (error) => {
+ logger.error({ err: error, port: config.temporal.healthPort }, 'health server error');
+ });
+
server.listen(config.temporal.healthPort, () =>
logger.info({ port: config.temporal.healthPort }, 'worker health endpoint listening'),
);📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| export function startHealthServer(source: WorkerStatusSource, logger: Logger): Server { | |
| const startedAt = Date.now(); | |
| const server = createServer((request, response) => { | |
| if (request.url !== '/health') { | |
| response.writeHead(404, { 'content-type': 'application/json' }); | |
| response.end(JSON.stringify({ error: 'not found' })); | |
| return; | |
| } | |
| const workerState = source.getWorkerState(); | |
| const connection = source.getConnectionStatus(); | |
| const healthy = workerState === 'RUNNING' && connection === 'connected'; | |
| response.writeHead(healthy ? 200 : 503, { 'content-type': 'application/json' }); | |
| response.end( | |
| JSON.stringify({ | |
| status: healthy ? 'healthy' : 'degraded', | |
| worker: workerState, | |
| connection, | |
| namespace: config.temporal.namespace, | |
| taskQueue: config.temporal.taskQueue, | |
| uptimeSeconds: Math.round((Date.now() - startedAt) / 1000), | |
| }), | |
| ); | |
| }); | |
| server.listen(config.temporal.healthPort, () => | |
| logger.info({ port: config.temporal.healthPort }, 'worker health endpoint listening'), | |
| ); | |
| return server; | |
| } | |
| export function startHealthServer(source: WorkerStatusSource, logger: Logger): Server { | |
| const startedAt = Date.now(); | |
| const server = createServer((request, response) => { | |
| if (request.url !== '/health') { | |
| response.writeHead(404, { 'content-type': 'application/json' }); | |
| response.end(JSON.stringify({ error: 'not found' })); | |
| return; | |
| } | |
| const workerState = source.getWorkerState(); | |
| const connection = source.getConnectionStatus(); | |
| const healthy = workerState === 'RUNNING' && connection === 'connected'; | |
| response.writeHead(healthy ? 200 : 503, { 'content-type': 'application/json' }); | |
| response.end( | |
| JSON.stringify({ | |
| status: healthy ? 'healthy' : 'degraded', | |
| worker: workerState, | |
| connection, | |
| namespace: config.temporal.namespace, | |
| taskQueue: config.temporal.taskQueue, | |
| uptimeSeconds: Math.round((Date.now() - startedAt) / 1000), | |
| }), | |
| ); | |
| }); | |
| server.on('error', (error) => { | |
| logger.error({ err: error, port: config.temporal.healthPort }, 'health server error'); | |
| }); | |
| server.listen(config.temporal.healthPort, () => | |
| logger.info({ port: config.temporal.healthPort }, 'worker health endpoint listening'), | |
| ); | |
| return server; | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@services/temporal-worker/src/health-server.ts` around lines 16 - 45, Register
an error listener on the HTTP server created in startHealthServer before calling
server.listen. Handle listen-time errors through the provided logger, including
the error details, so health-server failures do not become unhandled
EventEmitter errors that crash the worker.
1 participant
Summary
Adds Temporal as the workflow orchestration engine for all future long-running processes (document ingestion, meeting processing, email/calendar/GitHub sync, memory updates, nightly jobs, approval steps). BullMQ stays for simple fire-and-forget jobs. Infrastructure only — no AI functionality.
Changes
Infrastructure
temporalio/auto-setup:1.27) on :7233 in docker-compose, reusing the existing Postgres container; auto-registers thecompany-brainnamespace (7d retention)TEMPORAL_ADDRESS,TEMPORAL_NAMESPACE,TEMPORAL_TASK_QUEUE,TEMPORAL_WORKER_HEALTH_PORTNew packages
packages/workflows— deterministic workflow definitions (helloWorkflow,healthCheckWorkflow,storageWorkflow), shared signal/query definitions, retry policies, task-queue and workflow-type constants. Compiled with no Node types (sandbox-safe).packages/activities— all side effects:printMessage,checkServices(probes Postgres/Redis/MinIO/Qdrant),uploadFile(MinIO), built as a factory over long-lived clients.New service
services/temporal-worker— zod-typed config, connection manager with exponential-backoff retry, worker on task queuebrain-core, health/status endpoint on :4100, graceful shutdown (drains in-flight tasks).API
TemporalService(app.temporal): lazy client, workflow-ID generation, start/execute/signal/query/describe, worker-status proxy/healthnow reportstemporal/api/v1/workflows(Bearer auth): start hello, run health-check/storage, describe, query status, skip signal, server+worker statusDocs
docs/temporal.md: why Temporal, how workflows/activities work, config reference, execution examples, checklist for adding future workflowsTesting
pnpm typecheck,pnpm lint,pnpm testgreen across the workspaceskipDelaysignal andgetStatus/getReportqueries verified; storage workflow uploaded to MinIO; graceful shutdown drained cleanly;/healthreports all services upSummary by CodeRabbit