fix: 切换 HWLAB 投影实时权威

2026-07-13 20:28:47 +02:00
parent d420251f2e
commit c1af68475e
6 changed files with 382 additions and 34 deletions
@@ -545,6 +545,7 @@ export interface HwlabRuntimeKafkaShadowProducerSpec {

 export interface HwlabRuntimeKafkaEventBridgeSpec {
  readonly enabled: boolean;
+  readonly authority: "transactional-projection" | "direct-live";
  readonly features: {
    readonly directPublish: boolean;
    readonly liveKafkaSse: boolean;
@@ -560,6 +561,28 @@ export interface HwlabRuntimeKafkaEventBridgeSpec {
    readonly matchedEventLimit: number;
    readonly liveBufferLimit: number;
  };
+  readonly transactionalProjector?: {
+    readonly heartbeatIntervalMs: number;
+  };
+  readonly projectionOutboxRelay?: {
+    readonly intervalMs: number;
+    readonly batchSize: number;
+    readonly leaseMs: number;
+    readonly sendTimeoutMs: number;
+    readonly retryBackoffMs: number;
+  };
+  readonly projectionRealtime?: {
+    readonly outboxTailBatchSize: number;
+    readonly sseHeartbeatMs: number;
+    readonly sseDrainTimeoutMs: number;
+  };
+  readonly healthThresholds: {
+    readonly consumerLagWarning: number;
+    readonly outboxBacklogWarning: number;
+    readonly retryCountWarning: number;
+    readonly failedInboxWarning: number;
+    readonly dlqWarning: number;
+  };
  readonly configRef: string;
  readonly bootstrapServers: string;
  readonly stdioTopic: string;
@@ -1221,8 +1244,14 @@ function codeAgentKafkaShadowProducerConfig(value: unknown, path: string): Hwlab
 function codeAgentKafkaEventBridgeConfig(value: unknown, path: string): HwlabRuntimeKafkaEventBridgeSpec {
  const raw = asRecord(value, path);
  const features = asRecord(raw.features, `${path}.features`);
+  const enabled = booleanField(raw, "enabled", path);
+  const authority = enumStringField(raw, "authority", path, ["transactional-projection", "direct-live"] as const);
+  const directPublish = booleanField(features, "directPublish", `${path}.features`);
  const liveKafkaSse = booleanField(features, "liveKafkaSse", `${path}.features`);
  const kafkaRefreshReplay = booleanField(features, "kafkaRefreshReplay", `${path}.features`);
+  const transactionalProjector = booleanField(features, "transactionalProjector", `${path}.features`);
+  const projectionOutboxRelay = booleanField(features, "projectionOutboxRelay", `${path}.features`);
+  const projectionRealtime = booleanField(features, "projectionRealtime", `${path}.features`);
  const refreshReplay = raw.refreshReplay === undefined
    ? undefined
    : codeAgentKafkaRefreshReplayConfig(raw.refreshReplay, `${path}.refreshReplay`);
@@ -1232,6 +1261,24 @@ function codeAgentKafkaEventBridgeConfig(value: unknown, path: string): HwlabRun
  if (kafkaRefreshReplay && !liveKafkaSse) {
    throw new Error(`${path}.features.liveKafkaSse must be true when ${path}.features.kafkaRefreshReplay is true`);
  }
+  if (authority === "transactional-projection" && (!enabled || directPublish || liveKafkaSse || kafkaRefreshReplay || !transactionalProjector || !projectionOutboxRelay || !projectionRealtime)) {
+    throw new Error(`${path}.authority transactional-projection requires enabled transactionalProjector/projectionOutboxRelay/projectionRealtime and disables directPublish/liveKafkaSse/kafkaRefreshReplay`);
+  }
+  if (authority === "direct-live" && (!enabled || !directPublish || !liveKafkaSse || transactionalProjector || projectionOutboxRelay || projectionRealtime)) {
+    throw new Error(`${path}.authority direct-live requires enabled directPublish/liveKafkaSse and disables transactional projection capabilities`);
+  }
+  const transactionalProjectorConfig = raw.transactionalProjector === undefined
+    ? undefined
+    : codeAgentKafkaTransactionalProjectorConfig(raw.transactionalProjector, `${path}.transactionalProjector`);
+  const projectionOutboxRelayConfig = raw.projectionOutboxRelay === undefined
+    ? undefined
+    : codeAgentKafkaProjectionOutboxRelayConfig(raw.projectionOutboxRelay, `${path}.projectionOutboxRelay`);
+  const projectionRealtimeConfig = raw.projectionRealtime === undefined
+    ? undefined
+    : codeAgentKafkaProjectionRealtimeConfig(raw.projectionRealtime, `${path}.projectionRealtime`);
+  if (transactionalProjector && transactionalProjectorConfig === undefined) throw new Error(`${path}.transactionalProjector is required when ${path}.features.transactionalProjector is true`);
+  if (projectionOutboxRelay && projectionOutboxRelayConfig === undefined) throw new Error(`${path}.projectionOutboxRelay is required when ${path}.features.projectionOutboxRelay is true`);
+  if (projectionRealtime && projectionRealtimeConfig === undefined) throw new Error(`${path}.projectionRealtime is required when ${path}.features.projectionRealtime is true`);
  const directPublishConsumerGroupId = stringField(raw, "directPublishConsumerGroupId", path);
  const transactionalProjectorConsumerGroupId = stringField(raw, "transactionalProjectorConsumerGroupId", path);
  const hwlabEventConsumerGroupId = stringField(raw, "hwlabEventConsumerGroupId", path);
@@ -1239,16 +1286,21 @@ function codeAgentKafkaEventBridgeConfig(value: unknown, path: string): HwlabRun
    throw new Error(`${path} consumer group ids must be distinct so enabled capabilities receive independent event streams`);
  }
  return {
-    enabled: booleanField(raw, "enabled", path),
+    enabled,
+    authority,
    features: {
-      directPublish: booleanField(features, "directPublish", `${path}.features`),
+      directPublish,
      liveKafkaSse,
      kafkaRefreshReplay,
-      transactionalProjector: booleanField(features, "transactionalProjector", `${path}.features`),
-      projectionOutboxRelay: booleanField(features, "projectionOutboxRelay", `${path}.features`),
-      projectionRealtime: booleanField(features, "projectionRealtime", `${path}.features`),
+      transactionalProjector,
+      projectionOutboxRelay,
+      projectionRealtime,
    },
    ...(refreshReplay === undefined ? {} : { refreshReplay }),
+    ...(transactionalProjectorConfig === undefined ? {} : { transactionalProjector: transactionalProjectorConfig }),
+    ...(projectionOutboxRelayConfig === undefined ? {} : { projectionOutboxRelay: projectionOutboxRelayConfig }),
+    ...(projectionRealtimeConfig === undefined ? {} : { projectionRealtime: projectionRealtimeConfig }),
+    healthThresholds: codeAgentKafkaHealthThresholdsConfig(raw.healthThresholds, `${path}.healthThresholds`),
    configRef: stringField(raw, "configRef", path),
    bootstrapServers: stringField(raw, "bootstrapServers", path),
    stdioTopic: stringField(raw, "stdioTopic", path),
@@ -1261,6 +1313,42 @@ function codeAgentKafkaEventBridgeConfig(value: unknown, path: string): HwlabRun
  };
 }

+function codeAgentKafkaTransactionalProjectorConfig(value: unknown, path: string): NonNullable<HwlabRuntimeKafkaEventBridgeSpec["transactionalProjector"]> {
+  const raw = asRecord(value, path);
+  return { heartbeatIntervalMs: numberField(raw, "heartbeatIntervalMs", path) };
+}
+
+function codeAgentKafkaProjectionOutboxRelayConfig(value: unknown, path: string): NonNullable<HwlabRuntimeKafkaEventBridgeSpec["projectionOutboxRelay"]> {
+  const raw = asRecord(value, path);
+  return {
+    intervalMs: numberField(raw, "intervalMs", path),
+    batchSize: numberField(raw, "batchSize", path),
+    leaseMs: numberField(raw, "leaseMs", path),
+    sendTimeoutMs: numberField(raw, "sendTimeoutMs", path),
+    retryBackoffMs: numberField(raw, "retryBackoffMs", path),
+  };
+}
+
+function codeAgentKafkaProjectionRealtimeConfig(value: unknown, path: string): NonNullable<HwlabRuntimeKafkaEventBridgeSpec["projectionRealtime"]> {
+  const raw = asRecord(value, path);
+  return {
+    outboxTailBatchSize: numberField(raw, "outboxTailBatchSize", path),
+    sseHeartbeatMs: numberField(raw, "sseHeartbeatMs", path),
+    sseDrainTimeoutMs: numberField(raw, "sseDrainTimeoutMs", path),
+  };
+}
+
+function codeAgentKafkaHealthThresholdsConfig(value: unknown, path: string): HwlabRuntimeKafkaEventBridgeSpec["healthThresholds"] {
+  const raw = asRecord(value, path);
+  return {
+    consumerLagWarning: nonNegativeIntegerField(raw, "consumerLagWarning", path),
+    outboxBacklogWarning: nonNegativeIntegerField(raw, "outboxBacklogWarning", path),
+    retryCountWarning: nonNegativeIntegerField(raw, "retryCountWarning", path),
+    failedInboxWarning: nonNegativeIntegerField(raw, "failedInboxWarning", path),
+    dlqWarning: nonNegativeIntegerField(raw, "dlqWarning", path),
+  };
+}
+
 function codeAgentKafkaRefreshReplayConfig(value: unknown, path: string): NonNullable<HwlabRuntimeKafkaEventBridgeSpec["refreshReplay"]> {
  const raw = asRecord(value, path);
  return {