diff --git a/deploy/deploy.yaml b/deploy/deploy.yaml index 4cd9a355..0cfc852c 100644 --- a/deploy/deploy.yaml +++ b/deploy/deploy.yaml @@ -1161,6 +1161,9 @@ lanes: WORKBENCH_ACTIVITY_RETRY_INITIAL_INTERVAL_MS: "1000" WORKBENCH_ACTIVITY_RETRY_MAXIMUM_INTERVAL_MS: "30000" WORKBENCH_ACTIVITY_RETRY_MAXIMUM_ATTEMPTS: "5" + HWLAB_KAFKA_BOOTSTRAP_SERVERS: platform-infra-kafka-kafka-bootstrap.platform-infra.svc.cluster.local:9092 + HWLAB_KAFKA_EVENT_TOPIC: hwlab.event.v1 + HWLAB_KAFKA_CLIENT_ID: hwlab-v03-workbench-worker OTEL_SERVICE_NAME: hwlab-workbench-worker production: name: Production diff --git a/internal/workbench/kafka-event-publisher.test.ts b/internal/workbench/kafka-event-publisher.test.ts index 8f3d8603..2e1a94bf 100644 --- a/internal/workbench/kafka-event-publisher.test.ts +++ b/internal/workbench/kafka-event-publisher.test.ts @@ -60,16 +60,8 @@ test("L0 publisher writes semantic activity to hwlab.event.v1", async () => { function kafkaEnvironment(): Record { return { - HWLAB_KAFKA_DIRECT_PUBLISH_ENABLED: "false", - HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED: "true", - HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED: "false", - HWLAB_KAFKA_TRANSACTIONAL_PROJECTOR_ENABLED: "false", - HWLAB_KAFKA_PROJECTION_OUTBOX_RELAY_ENABLED: "false", - HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED: "false", HWLAB_KAFKA_BOOTSTRAP_SERVERS: "127.0.0.1:9092", - HWLAB_KAFKA_AGENTRUN_EVENT_TOPIC: "agentrun.event.v1", HWLAB_KAFKA_EVENT_TOPIC: "hwlab.event.v1", HWLAB_KAFKA_CLIENT_ID: "hwlab-workbench-l0", - HWLAB_KAFKA_HWLAB_EVENT_GROUP_ID: "hwlab-workbench-l0-live", }; } diff --git a/internal/workbench/kafka-event-publisher.ts b/internal/workbench/kafka-event-publisher.ts index 09551ade..e9f1c96d 100644 --- a/internal/workbench/kafka-event-publisher.ts +++ b/internal/workbench/kafka-event-publisher.ts @@ -3,7 +3,6 @@ import { randomUUID } from "node:crypto"; import { buildHwlabKafkaProducerMessage, defaultKafkaFactory, - kafkaEventBridgeConfig, projectAgentRunKafkaEventToHwlabEvent, } from "../cloud/kafka-event-bridge.ts"; import type { WorkbenchEventPublisher } from "./contracts.ts"; @@ -13,8 +12,7 @@ export function createWorkbenchKafkaEventPublisher(options: { kafkaFactory?: typeof defaultKafkaFactory; now?: () => string; }): WorkbenchEventPublisher { - const config = kafkaEventBridgeConfig(options.env); - if (!config) throw codedError("workbench_kafka_publisher_unconfigured", "Workbench event publisher requires Kafka realtime configuration."); + const config = workbenchKafkaPublisherConfig(options.env); const kafka = (options.kafkaFactory ?? defaultKafkaFactory)(config); const producer = kafka.producer({ allowAutoTopicCreation: false }); let ready: Promise | null = null; @@ -63,6 +61,26 @@ export function createWorkbenchKafkaEventPublisher(options: { }; } +function workbenchKafkaPublisherConfig(env: Record) { + const brokers = csv(env.HWLAB_KAFKA_BOOTSTRAP_SERVERS); + const hwlabTopic = text(env.HWLAB_KAFKA_EVENT_TOPIC); + const clientId = text(env.HWLAB_KAFKA_CLIENT_ID); + const missing = [ + brokers.length === 0 ? "HWLAB_KAFKA_BOOTSTRAP_SERVERS" : null, + hwlabTopic ? null : "HWLAB_KAFKA_EVENT_TOPIC", + clientId ? null : "HWLAB_KAFKA_CLIENT_ID", + ].filter(Boolean); + if (missing.length > 0) throw codedError("workbench_kafka_publisher_unconfigured", `Workbench event publisher configuration is incomplete: ${missing.join(", ")}`); + return { + brokers, + hwlabTopic, + clientId, + dnsServers: csv(env.HWLAB_KAFKA_DNS_SERVERS), + dnsSearchDomains: csv(env.HWLAB_KAFKA_DNS_SEARCH_DOMAINS), + }; +} + +function csv(value: unknown): string[] { return text(value).split(",").map((item) => item.trim()).filter(Boolean); } function text(value: unknown): string { return String(value ?? "").trim(); } function timestamp(value: unknown): string | null { const result = text(value); return result && Number.isFinite(Date.parse(result)) ? new Date(Date.parse(result)).toISOString() : null; } function codedError(code: string, message: string) { return Object.assign(new Error(message), { code }); }