infra+dotnet: replace Kafka with in-process Channel<T> message bus (HOL-23)

infra/docker/.env.example

@@ -20,11 +20,9 @@ COMPOSE_PATH_SEPARATOR=:
 
 # Infra dependencies — pinned for reproducibility.
 CLICKHOUSE_IMAGE_NAME=clickhouse/clickhouse-server:24.3.15.72-alpine
-KAFKA_IMAGE_NAME=confluentinc/cp-kafka:7.7.0
 POSTGRES_IMAGE_NAME=ankane/pgvector:v0.5.1
 # OTEL collector image vars removed — HOL-21 dropped the collector container.
 # Backend hosts the OTLP receiver at /otel/v1/{logs,traces,metrics}.
-ZOOKEEPER_IMAGE_NAME=confluentinc/cp-zookeeper:7.7.0
 
 # ── ClickHouse (analytics) ──────────────────────────────────────────
 CLICKHOUSE_DATABASE=default
@@ -39,9 +37,9 @@ PSQL_PORT=5432
 PSQL_USER=postgres
 
 # ── Kafka ───────────────────────────────────────────────────────────
-KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092
-KAFKA_SERVERS=kafka:9092
-KAFKA_TOPIC=dev
+# Kafka removed in HOL-23 — backend uses an in-process Channel-backed
+# message bus. Re-introduce these vars only if swapping a Kafka-backed
+# IMessageBus implementation back in for multi-node deployments.
 
 # ── Redis ───────────────────────────────────────────────────────────
 # Removed in HOL-22 — backend uses IMemoryCache + Postgres instead.

infra/docker/compose.hobby-dotnet.yml

@@ -21,7 +21,6 @@ services:
             - ClickHouse__Password=${CLICKHOUSE_PASSWORD:-}
             - ClickHouse__Migrations__Path=/app/clickhouse-migrations
             - ClickHouse__Migrations__Disabled=${CLICKHOUSE_MIGRATIONS_DISABLED:-false}
-            - Kafka__BootstrapServers=${KAFKA_SERVERS:-kafka:9092}
             - Storage__Type=filesystem
             - Storage__FilesystemRoot=/highlight-data
             - Frontend__Uri=${REACT_APP_FRONTEND_URI:-http://localhost:3000}

infra/docker/compose.yml

@@ -3,43 +3,10 @@ x-local-logging: &local-logging
 
 # HoldFast services for the development deployment.
 services:
-    zookeeper:
-        logging: *local-logging
-        image: ${ZOOKEEPER_IMAGE_NAME}
-        container_name: zookeeper
-        restart: on-failure
-        volumes:
-            - zoo-data:/var/lib/zookeeper/data
-            - zoo-log:/var/lib/zookeeper/log
-        environment:
-            ZOOKEEPER_CLIENT_PORT: 2181
-            ZOOKEEPER_TICK_TIME: 2000
-
-    kafka:
-        logging: *local-logging
-        image: ${KAFKA_IMAGE_NAME}
-        container_name: kafka
-        volumes:
-            - kafka-data:/var/lib/kafka/data
-        ports:
-            - '0.0.0.0:9092:9092'
-        restart: on-failure
-        depends_on:
-            - zookeeper
-        environment:
-            KAFKA_ADVERTISED_LISTENERS: ${KAFKA_ADVERTISED_LISTENERS}
-            KAFKA_BROKER_ID: 1
-            KAFKA_CONSUMER_MAX_PARTITION_FETCH_BYTES: 268435456
-            KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT
-            KAFKA_LOG_RETENTION_HOURS: 1
-            KAFKA_LOG_SEGMENT_BYTES: 268435456
-            KAFKA_MESSAGE_MAX_BYTES: 268435456
-            KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
-            KAFKA_PRODUCER_MAX_REQUEST_SIZE: 268435456
-            KAFKA_REPLICA_FETCH_MAX_BYTES: 268435456
-            KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
-            KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
-            KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
+    # Kafka + Zookeeper removed in HOL-23 — backend uses an in-process
+    # Channel-backed message bus (HoldFast.Shared.Messaging.InProcessMessageBus).
+    # Producers and consumers live in the same .NET host (all-in-one runtime
+    # mode), so no broker is required at hobby/lean self-hosted scale.
 
     # Redis removed in HOL-22 — RedisService was registered in DI but no
     # code path actually injected it. Self-hosted scale doesn't need a
@@ -88,6 +55,3 @@ volumes:
     postgres-data:
     clickhouse-data:
     clickhouse-logs:
-    kafka-data:
-    zoo-log:
-    zoo-data:

infra/docker/env.sh

@@ -26,8 +26,6 @@ fi
 if [[ "$*" == *"--go-docker"* ]]; then
     export CLICKHOUSE_ADDRESS=clickhouse:9000
     export IN_DOCKER_GO=true
-    export KAFKA_ADVERTISED_LISTENERS="PLAINTEXT://kafka:9092"
-    export KAFKA_SERVERS=kafka:9092
     export ON_PREM=true
     # HOL-21: collector container removed; backend hosts the OTLP receiver.
     export OTLP_DOGFOOD_ENDPOINT=http://backend:8082/otel
@@ -36,8 +34,6 @@ if [[ "$*" == *"--go-docker"* ]]; then
     echo "Using docker-internal infra."
 else
     export CLICKHOUSE_ADDRESS=localhost:9000
-    export KAFKA_ADVERTISED_LISTENERS="PLAINTEXT://localhost:9092"
-    export KAFKA_SERVERS=localhost:9092
     export OTLP_DOGFOOD_ENDPOINT=http://localhost:8082/otel
     export OTLP_ENDPOINT=http://localhost:8082/otel
     export PSQL_HOST=localhost

infra/docker/start-infra.sh

@@ -4,12 +4,9 @@ source env.sh
 
 # startup the infra
 
-SERVICES="clickhouse kafka postgres redis zookeeper"
-BUILD_ARGS="--build-arg OTEL_COLLECTOR_ALPINE_IMAGE_NAME=${OTEL_COLLECTOR_ALPINE_IMAGE_NAME} \
-            --build-arg OTEL_COLLECTOR_IMAGE_NAME=${OTEL_COLLECTOR_IMAGE_NAME}"
+SERVICES="clickhouse postgres"
 
 docker compose pull $SERVICES
-docker compose build --pull $BUILD_ARGS $SERVICES
 docker compose up --detach --wait --remove-orphans $SERVICES
 
 if [[ "$*" != *"--go-docker"* ]]; then

src/dotnet/src/HoldFast.Api/Program.cs

@@ -10,6 +10,7 @@
 using HoldFast.Shared.Notifications;
 using HoldFast.Shared.ErrorGrouping;
 using HoldFast.Shared.Kafka;
+using HoldFast.Shared.Messaging;
 using HoldFast.Shared.Runtime;
 using HoldFast.Shared.SessionProcessing;
 using HoldFast.Storage;
@@ -137,20 +138,15 @@ req.RequestUri is null ||
 });
 builder.Services.AddScoped<IAuthorizationService, AuthorizationService>();
 
-// ── Kafka ─────────────────────────────────────────────────────────────
-builder.Services.Configure<KafkaOptions>(
-    builder.Configuration.GetSection("Kafka"));
-builder.Services.AddSingleton<KafkaProducerService>();
+// ── Message bus ──────────────────────────────────────────────────────
+// HOL-23: in-process Channel<T>-backed bus replaces Kafka for hobby/lean
+// self-hosted deployments. Producer and consumers both run in the same
+// .NET host (all-in-one runtime mode); see HoldFast.Shared.Messaging.
+// To run the worker in a separate process from the API, swap in a
+// Kafka-backed IMessageBus implementation that honors the same shape.
+builder.Services.AddSingleton<IMessageBus, InProcessMessageBus>();
 builder.Services.AddSingleton<IKafkaProducer, KafkaProducerAdapter>();
 
-// Topic bootstrap — pre-create required topics so consumers don't crash on
-// subscribe (Confluent.Kafka auto-create only triggers on producer writes).
-// Disable via Kafka__TopicBootstrap__Disabled=true when topics are managed
-// externally (Strimzi KafkaTopic resources, Helm pre-jobs, etc).
-builder.Services.Configure<KafkaTopicBootstrapOptions>(
-    builder.Configuration.GetSection("Kafka:TopicBootstrap"));
-builder.Services.AddHostedService<KafkaTopicBootstrapService>();
-
 // ── ClickHouse ────────────────────────────────────────────────────────
 builder.Services.Configure<ClickHouseOptions>(
     builder.Configuration.GetSection("ClickHouse"));

src/dotnet/src/HoldFast.GraphQL.Public/KafkaProducerAdapter.cs

@@ -1,25 +1,29 @@
 using System.Text.Json;
 using HoldFast.GraphQL.Public.InputTypes;
 using HoldFast.Shared.Kafka;
+using HoldFast.Shared.Messaging;
 
 namespace HoldFast.GraphQL.Public;
 
 /// <summary>
-/// Implements IKafkaProducer by forwarding to the shared KafkaProducerService.
+/// Implements IKafkaProducer by forwarding to the shared <see cref="IMessageBus"/>.
+/// (HOL-23: the bus is now an in-process Channel-backed implementation by default,
+/// not Confluent.Kafka. The class name is kept for now to keep the diff narrow —
+/// rename pending in a follow-up.)
 /// </summary>
 public class KafkaProducerAdapter : IKafkaProducer
 {
-    private readonly KafkaProducerService _producer;
+    private readonly IMessageBus _producer;
 
-    public KafkaProducerAdapter(KafkaProducerService producer)
+    public KafkaProducerAdapter(IMessageBus producer)
     {
         _producer = producer;
     }
 
     public Task ProduceSessionEventsAsync(string sessionSecureId, long payloadId, string data, CancellationToken ct)
     {
         var message = new { SessionSecureId = sessionSecureId, PayloadId = payloadId, Data = data };
-        return _producer.ProduceAsync(KafkaTopics.SessionEvents, sessionSecureId, message, ct);
+        return _producer.PublishAsync(KafkaTopics.SessionEvents, sessionSecureId, message, ct);
     }
 
     public async Task ProducePushPayloadAsync(string sessionSecureId, long payloadId, string events,
@@ -37,7 +41,7 @@ public async Task ProducePushPayloadAsync(string sessionSecureId, long payloadId
             PayloadId = payloadId,
             Data = events,
         };
-        await _producer.ProduceAsync(KafkaTopics.SessionEvents, sessionSecureId, sessionMsg, ct);
+        await _producer.PublishAsync(KafkaTopics.SessionEvents, sessionSecureId, sessionMsg, ct);
 
         // Each frontend error → FrontendErrors topic so FrontendErrorsConsumer can
         // group them into ClickHouse error_objects. Errors are dropped on the
@@ -58,7 +62,7 @@ public async Task ProducePushPayloadAsync(string sessionSecureId, long payloadId
                 Timestamp = err.Timestamp,
                 Payload = err.Payload,
             };
-            await _producer.ProduceAsync(KafkaTopics.FrontendErrors, sessionSecureId, frontendMsg, ct);
+            await _producer.PublishAsync(KafkaTopics.FrontendErrors, sessionSecureId, frontendMsg, ct);
         }
 
         // _ = messages; _ = resources; _ = webSocketEvents; _ = isBeacon;
@@ -71,21 +75,21 @@ public async Task ProducePushPayloadAsync(string sessionSecureId, long payloadId
 
     public Task ProduceBackendErrorAsync(string? projectId, BackendErrorObjectInput error, CancellationToken ct)
     {
-        return _producer.ProduceAsync(KafkaTopics.BackendErrors, projectId ?? "unknown", error, ct);
+        return _producer.PublishAsync(KafkaTopics.BackendErrors, projectId ?? "unknown", error, ct);
     }
 
     public Task ProduceMetricAsync(MetricInput metric, CancellationToken ct)
     {
-        return _producer.ProduceAsync(KafkaTopics.Metrics, metric.SessionSecureId, metric, ct);
+        return _producer.PublishAsync(KafkaTopics.Metrics, metric.SessionSecureId, metric, ct);
     }
 
     public Task ProduceLogAsync(LogInput log, CancellationToken ct)
     {
-        return _producer.ProduceAsync(KafkaTopics.Logs, log.TraceId, log, ct);
+        return _producer.PublishAsync(KafkaTopics.Logs, log.TraceId, log, ct);
     }
 
     public Task ProduceTraceAsync(TraceInput trace, CancellationToken ct)
     {
-        return _producer.ProduceAsync(KafkaTopics.Traces, trace.TraceId, trace, ct);
+        return _producer.PublishAsync(KafkaTopics.Traces, trace.TraceId, trace, ct);
     }
 }

src/dotnet/src/HoldFast.Shared/HoldFast.Shared.csproj

@@ -6,7 +6,6 @@
   </ItemGroup>
 
   <ItemGroup>
-    <PackageReference Include="Confluent.Kafka" Version="2.13.2" />
     <PackageReference Include="Microsoft.AspNetCore.Authentication.JwtBearer" Version="10.0.5" />
     <PackageReference Include="OpenTelemetry.Exporter.OpenTelemetryProtocol" Version="1.15.3" />
     <PackageReference Include="OpenTelemetry.Instrumentation.AspNetCore" Version="1.15.2" />