# GB32960 Service Split Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** Split the current all-in-one GB32960 production path into separate ingest, history, and analytics runnable apps with Kafka as the durability boundary and ACK after Kafka success.

**Architecture:** Add three Spring Boot app modules while keeping `bootstrap-all` available for development and rollback. First isolate runtime composition, then introduce GB32960-specific Kafka topics, then change the GB32960 ACK path so realtime/report ACKs are written after required Kafka production succeeds. History and analytics become Kafka consumers in separate apps and no longer share a JVM with TCP ingest.

**Tech Stack:** Java 25, Spring Boot 3.5, Maven multi-module build, Netty, Kafka clients, Protobuf envelope, DuckDB/Parquet event store, local archive sink.

---

## Current-State Notes

- Existing all-in-one entrypoint: `modules/apps/bootstrap-all/src/main/java/com/lingniu/ingest/bootstrap/IngestApplication.java`.
- Existing all-in-one module at the time of this plan imported protocols,
  sinks, history, state, stat, command gateway, MQTT, and JT modules. Xinda was
  later removed from source and build configuration.
- Existing GB32960 handler: `modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandler.java`.
- Existing handler currently writes report ACK before `dispatcher.dispatch(rf)`.
- Existing `DisruptorEventBus` publishes to sinks asynchronously and does not wait for sink completion.
- Existing `KafkaEventSink.accepts()` rejects `VehicleEvent.RawArchive`.
- Existing versioned topic names are not yet present; current topic names live under `lingniu.ingest.sink.kafka.topics`.
- Existing history consumer logic: `modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestor.java`.
- Existing analytics consumers: `VehicleStateEnvelopeIngestor` and `VehicleStatEnvelopeIngestor`.
- There is an unrelated working-tree change in `modules/apps/bootstrap-all/src/main/resources/application.yml` for GB32960 diagnostics capacity. Do not revert it. Do not include it in service-split commits unless the user explicitly asks.

## File Structure

Create app modules:

- `modules/apps/gb32960-ingest-app/pom.xml`
- `modules/apps/gb32960-ingest-app/src/main/java/com/lingniu/ingest/gb32960app/Gb32960IngestApplication.java`
- `modules/apps/gb32960-ingest-app/src/main/resources/application.yml`
- `modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/Gb32960IngestAppCompositionTest.java`
- `modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/Gb32960IngestAppDefaultsTest.java`

- `modules/apps/vehicle-history-app/pom.xml`
- `modules/apps/vehicle-history-app/src/main/java/com/lingniu/ingest/historyapp/VehicleHistoryApplication.java`
- `modules/apps/vehicle-history-app/src/main/resources/application.yml`
- `modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppCompositionTest.java`
- `modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppDefaultsTest.java`

- `modules/apps/vehicle-analytics-app/pom.xml`
- `modules/apps/vehicle-analytics-app/src/main/java/com/lingniu/ingest/analyticsapp/VehicleAnalyticsApplication.java`
- `modules/apps/vehicle-analytics-app/src/main/resources/application.yml`
- `modules/apps/vehicle-analytics-app/src/test/java/com/lingniu/ingest/analyticsapp/VehicleAnalyticsAppCompositionTest.java`
- `modules/apps/vehicle-analytics-app/src/test/java/com/lingniu/ingest/analyticsapp/VehicleAnalyticsAppDefaultsTest.java`

Modify shared build and Kafka contract files:

- `pom.xml`
- `modules/sinks/sink-kafka/src/main/java/com/lingniu/ingest/sink/kafka/KafkaSinkProperties.java`
- `modules/sinks/sink-kafka/src/main/java/com/lingniu/ingest/sink/kafka/TopicRouter.java`
- `modules/sinks/sink-kafka/src/main/java/com/lingniu/ingest/sink/kafka/KafkaEventSink.java`
- `modules/sinks/sink-kafka/src/test/java/com/lingniu/ingest/sink/kafka/TopicRouterTest.java`
- `modules/sinks/sink-kafka/src/test/java/com/lingniu/ingest/sink/kafka/KafkaEventSinkTest.java`

Modify ACK/Kafka durability boundary files:

- `modules/core/ingest-api/src/main/java/com/lingniu/ingest/api/sink/EventSink.java`
- `modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/dispatcher/Dispatcher.java`
- `modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/concurrency/DisruptorEventBus.java`
- `modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandler.java`
- `modules/protocols/protocol-gb32960/src/test/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandlerAckBoundaryTest.java`

Optional later follow-up files after the first split is running:

- `docs/operations/gb32960-service-split-runbook.md`
- Deployment manifests or service manager files used by the user's environment.

---

### Task 1: Register the Three New App Modules

**Files:**
- Modify: `pom.xml`
- Create: `modules/apps/gb32960-ingest-app/pom.xml`
- Create: `modules/apps/vehicle-history-app/pom.xml`
- Create: `modules/apps/vehicle-analytics-app/pom.xml`

- [ ] **Step 1: Add failing module-presence test by running Maven before modules exist**

Run:

```bash
export JAVA_HOME=/opt/homebrew/opt/openjdk@25/libexec/openjdk.jdk/Contents/Home
export PATH="$JAVA_HOME/bin:/opt/homebrew/bin:$PATH"
mvn -pl :gb32960-ingest-app -am validate
```

Expected: FAIL with a message that Maven cannot find project `:gb32960-ingest-app`.

- [ ] **Step 2: Add modules to the root reactor**

In `pom.xml`, add these module entries after `modules/apps/command-gateway` and before `modules/apps/bootstrap-all`:

```xml
        <module>modules/apps/gb32960-ingest-app</module>
        <module>modules/apps/vehicle-history-app</module>
        <module>modules/apps/vehicle-analytics-app</module>
```

Also add dependency management entries near the existing `command-gateway` entry:

```xml
            <dependency>
                <groupId>com.lingniu.ingest</groupId>
                <artifactId>gb32960-ingest-app</artifactId>
                <version>${project.version}</version>
            </dependency>
            <dependency>
                <groupId>com.lingniu.ingest</groupId>
                <artifactId>vehicle-history-app</artifactId>
                <version>${project.version}</version>
            </dependency>
            <dependency>
                <groupId>com.lingniu.ingest</groupId>
                <artifactId>vehicle-analytics-app</artifactId>
                <version>${project.version}</version>
            </dependency>
```

- [ ] **Step 3: Create `gb32960-ingest-app` POM**

Create `modules/apps/gb32960-ingest-app/pom.xml`:

```xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>com.lingniu.ingest</groupId>
        <artifactId>lingniu-vehicle-ingest</artifactId>
        <version>0.1.0-SNAPSHOT</version>
        <relativePath>../../../pom.xml</relativePath>
    </parent>

    <artifactId>gb32960-ingest-app</artifactId>
    <name>gb32960-ingest-app</name>
    <description>GB32960 protocol ingress runtime: TCP decode, auth, Kafka production, and ACK.</description>

    <dependencies>
        <dependency>
            <groupId>com.lingniu.ingest</groupId>
            <artifactId>ingest-core</artifactId>
        </dependency>
        <dependency>
            <groupId>com.lingniu.ingest</groupId>
            <artifactId>session-core</artifactId>
        </dependency>
        <dependency>
            <groupId>com.lingniu.ingest</groupId>
            <artifactId>observability</artifactId>
        </dependency>
        <dependency>
            <groupId>com.lingniu.ingest</groupId>
            <artifactId>protocol-gb32960</artifactId>
        </dependency>
        <dependency>
            <groupId>com.lingniu.ingest</groupId>
            <artifactId>sink-kafka</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <build>
        <finalName>gb32960-ingest-app</finalName>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <configuration>
                    <jvmArguments>--sun-misc-unsafe-memory-access=allow</jvmArguments>
                </configuration>
                <executions>
                    <execution>
                        <goals>
                            <goal>repackage</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>
```

- [ ] **Step 4: Create `vehicle-history-app` POM**

Create `modules/apps/vehicle-history-app/pom.xml`:

```xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>com.lingniu.ingest</groupId>
        <artifactId>lingniu-vehicle-ingest</artifactId>
        <version>0.1.0-SNAPSHOT</version>
        <relativePath>../../../pom.xml</relativePath>
    </parent>

    <artifactId>vehicle-history-app</artifactId>
    <name>vehicle-history-app</name>
    <description>Vehicle raw archive, event history, and GB32960 frame query runtime.</description>

    <dependencies>
        <dependency>
            <groupId>com.lingniu.ingest</groupId>
            <artifactId>ingest-api</artifactId>
        </dependency>
        <dependency>
            <groupId>com.lingniu.ingest</groupId>
            <artifactId>observability</artifactId>
        </dependency>
        <dependency>
            <groupId>com.lingniu.ingest</groupId>
            <artifactId>sink-kafka</artifactId>
        </dependency>
        <dependency>
            <groupId>com.lingniu.ingest</groupId>
            <artifactId>sink-archive</artifactId>
        </dependency>
        <dependency>
            <groupId>com.lingniu.ingest</groupId>
            <artifactId>event-file-store</artifactId>
        </dependency>
        <dependency>
            <groupId>com.lingniu.ingest</groupId>
            <artifactId>event-history-service</artifactId>
        </dependency>
        <dependency>
            <groupId>com.lingniu.ingest</groupId>
            <artifactId>protocol-gb32960</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <build>
        <finalName>vehicle-history-app</finalName>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <configuration>
                    <jvmArguments>--sun-misc-unsafe-memory-access=allow</jvmArguments>
                </configuration>
                <executions>
                    <execution>
                        <goals>
                            <goal>repackage</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>
```

- [ ] **Step 5: Create `vehicle-analytics-app` POM**

Create `modules/apps/vehicle-analytics-app/pom.xml`:

```xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>com.lingniu.ingest</groupId>
        <artifactId>lingniu-vehicle-ingest</artifactId>
        <version>0.1.0-SNAPSHOT</version>
        <relativePath>../../../pom.xml</relativePath>
    </parent>

    <artifactId>vehicle-analytics-app</artifactId>
    <name>vehicle-analytics-app</name>
    <description>Vehicle state, daily statistics, alarms, and analytics runtime.</description>

    <dependencies>
        <dependency>
            <groupId>com.lingniu.ingest</groupId>
            <artifactId>ingest-api</artifactId>
        </dependency>
        <dependency>
            <groupId>com.lingniu.ingest</groupId>
            <artifactId>observability</artifactId>
        </dependency>
        <dependency>
            <groupId>com.lingniu.ingest</groupId>
            <artifactId>sink-kafka</artifactId>
        </dependency>
        <dependency>
            <groupId>com.lingniu.ingest</groupId>
            <artifactId>vehicle-state-service</artifactId>
        </dependency>
        <dependency>
            <groupId>com.lingniu.ingest</groupId>
            <artifactId>vehicle-stat-service</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <build>
        <finalName>vehicle-analytics-app</finalName>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <configuration>
                    <jvmArguments>--sun-misc-unsafe-memory-access=allow</jvmArguments>
                </configuration>
                <executions>
                    <execution>
                        <goals>
                            <goal>repackage</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>
```

- [ ] **Step 6: Verify the modules are visible to Maven**

Run:

```bash
export JAVA_HOME=/opt/homebrew/opt/openjdk@25/libexec/openjdk.jdk/Contents/Home
export PATH="$JAVA_HOME/bin:/opt/homebrew/bin:$PATH"
mvn -pl :gb32960-ingest-app,:vehicle-history-app,:vehicle-analytics-app -am validate
```

Expected: BUILD SUCCESS.

- [ ] **Step 7: Commit**

```bash
git add pom.xml modules/apps/gb32960-ingest-app/pom.xml modules/apps/vehicle-history-app/pom.xml modules/apps/vehicle-analytics-app/pom.xml
git commit -m "build: add split service app modules"
```

---

### Task 2: Add App Entrypoints and Runtime Defaults

**Files:**
- Create: `modules/apps/gb32960-ingest-app/src/main/java/com/lingniu/ingest/gb32960app/Gb32960IngestApplication.java`
- Create: `modules/apps/gb32960-ingest-app/src/main/resources/application.yml`
- Create: `modules/apps/vehicle-history-app/src/main/java/com/lingniu/ingest/historyapp/VehicleHistoryApplication.java`
- Create: `modules/apps/vehicle-history-app/src/main/resources/application.yml`
- Create: `modules/apps/vehicle-analytics-app/src/main/java/com/lingniu/ingest/analyticsapp/VehicleAnalyticsApplication.java`
- Create: `modules/apps/vehicle-analytics-app/src/main/resources/application.yml`

- [ ] **Step 1: Create the GB32960 ingest main class**

Create `modules/apps/gb32960-ingest-app/src/main/java/com/lingniu/ingest/gb32960app/Gb32960IngestApplication.java`:

```java
package com.lingniu.ingest.gb32960app;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication(scanBasePackages = "com.lingniu.ingest")
public class Gb32960IngestApplication {

    private static final Logger log = LoggerFactory.getLogger(Gb32960IngestApplication.class);

    public static void main(String[] args) {
        try {
            SpringApplication.run(Gb32960IngestApplication.class, args);
        } catch (Throwable t) {
            log.error("GB32960 ingest application failed to start, forcing JVM exit", t);
            System.exit(1);
        }
    }
}
```

- [ ] **Step 2: Create the GB32960 ingest config**

Create `modules/apps/gb32960-ingest-app/src/main/resources/application.yml`:

```yaml
spring:
  application:
    name: gb32960-ingest-app
  threads:
    virtual:
      enabled: true

server:
  port: ${HTTP_PORT:20100}

lingniu:
  ingest:
    gb32960:
      enabled: true
      port: ${GB32960_PORT:32960}
      boss-threads: ${GB32960_BOSS_THREADS:1}
      worker-threads: ${GB32960_WORKER_THREADS:0}
      auth:
        enabled: ${GB32960_AUTH_ENABLED:false}
        case-sensitive: false
        whitelist: []
        platforms:
          - username: ${GB32960_PLATFORM_USER_HYUNDAI:Hyundai}
            password: ${GB32960_PLATFORM_PWD_HYUNDAI:f2e3445d7cda409fb4f278f6fb890734}
            allowed-ips:
              - ${GB32960_PLATFORM_IP_HYUNDAI:8.134.95.166}
            description: 外部下级平台 - Hyundai
      tls:
        enabled: ${GB32960_TLS_ENABLED:false}
        cert-path: ${GB32960_TLS_CERT:}
        key-path: ${GB32960_TLS_KEY:}
        trust-cert-path: ${GB32960_TLS_CA:}
        require-client-auth: true
      vendor-extensions:
        - name: guangdong-fc
          match:
            platform-accounts:
              - Hyundai
            vin-prefixes: []
            vins: []
      parse:
        lenient-block-failure: true
      diagnostics:
        max-logged-frame-keys-per-channel: ${GB32960_DIAGNOSTICS_MAX_LOGGED_FRAME_KEYS_PER_CHANNEL:4096}
    pipeline:
      disruptor:
        ring-buffer-size: ${PIPELINE_RING_BUFFER_SIZE:131072}
        wait-strategy: ${PIPELINE_WAIT_STRATEGY:yielding}
        producer-type: multi
      dedup:
        enabled: true
        cache-size: 200000
        ttl-seconds: 600
      rate-limit:
        per-vin-qps: 50
    session:
      store: ${SESSION_STORE:redis}
      ttl: ${SESSION_TTL:30m}
    identity:
      store: ${VEHICLE_IDENTITY_STORE:mysql}
      mysql:
        jdbc-url: ${VEHICLE_IDENTITY_MYSQL_JDBC_URL:jdbc:mysql://127.0.0.1:3306/lingniu_vehicle?useUnicode=true&characterEncoding=utf8&useSSL=false&serverTimezone=Asia/Shanghai}
        username: ${VEHICLE_IDENTITY_MYSQL_USERNAME:root}
        password: ${VEHICLE_IDENTITY_MYSQL_PASSWORD:}
        table-name: ${VEHICLE_IDENTITY_MYSQL_TABLE:vehicle_identity_bindings}
        initialize-schema: ${VEHICLE_IDENTITY_MYSQL_INITIALIZE_SCHEMA:true}
    sink:
      kafka:
        enabled: ${KAFKA_ENABLED:true}
        bootstrap-servers: ${KAFKA_BROKERS:114.55.58.251:9092}
        compression-type: zstd
        linger-ms: 20
        batch-size: 65536
        acks: all
        enable-idempotence: true
        node-id: ${KAFKA_NODE_ID:gb32960-ingest-local}
        topics:
          realtime: ${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1}
          location: ${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1}
          alarm: ${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1}
          session: ${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1}
          media-meta: ${KAFKA_TOPIC_MEDIA_META:vehicle.media.meta.v1}
          raw-archive: ${KAFKA_TOPIC_GB32960_RAW:vehicle.raw.gb32960.v1}
          dlq: ${KAFKA_TOPIC_GB32960_DLQ:vehicle.dlq.gb32960.v1}
        consumer:
          enabled: false
    event-file-store:
      enabled: false
    event-history:
      enabled: false
    vehicle-state:
      enabled: false
    vehicle-stat:
      enabled: false

management:
  endpoints:
    web:
      exposure:
        include: health,info,metrics,prometheus,env
  health:
    redis:
      enabled: ${MANAGEMENT_HEALTH_REDIS_ENABLED:false}
  metrics:
    tags:
      application: gb32960-ingest-app

logging:
  level:
    root: INFO
    com.lingniu.ingest: INFO
```

- [ ] **Step 3: Create the history app main class**

Create `modules/apps/vehicle-history-app/src/main/java/com/lingniu/ingest/historyapp/VehicleHistoryApplication.java`:

```java
package com.lingniu.ingest.historyapp;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication(scanBasePackages = "com.lingniu.ingest")
public class VehicleHistoryApplication {

    private static final Logger log = LoggerFactory.getLogger(VehicleHistoryApplication.class);

    public static void main(String[] args) {
        try {
            SpringApplication.run(VehicleHistoryApplication.class, args);
        } catch (Throwable t) {
            log.error("Vehicle history application failed to start, forcing JVM exit", t);
            System.exit(1);
        }
    }
}
```

- [ ] **Step 4: Create the history app config**

Create `modules/apps/vehicle-history-app/src/main/resources/application.yml`:

```yaml
spring:
  application:
    name: vehicle-history-app
  threads:
    virtual:
      enabled: true

server:
  port: ${HTTP_PORT:20200}

lingniu:
  ingest:
    gb32960:
      enabled: false
    sink:
      kafka:
        enabled: ${KAFKA_ENABLED:true}
        bootstrap-servers: ${KAFKA_BROKERS:114.55.58.251:9092}
        topics:
          realtime: ${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1}
          raw-archive: ${KAFKA_TOPIC_GB32960_RAW:vehicle.raw.gb32960.v1}
          dlq: ${KAFKA_TOPIC_GB32960_DLQ:vehicle.dlq.gb32960.v1}
        consumer:
          enabled: ${KAFKA_CONSUMER_ENABLED:true}
          client-id-prefix: ${KAFKA_CONSUMER_CLIENT_ID_PREFIX:vehicle-history}
          auto-offset-reset: ${KAFKA_CONSUMER_AUTO_OFFSET_RESET:earliest}
          max-poll-records: ${KAFKA_CONSUMER_MAX_POLL_RECORDS:500}
          bindings:
            eventHistoryEnvelopeConsumerProcessor:
              enabled: true
              group-id: ${KAFKA_GROUP_HISTORY:vehicle-history}
              topics:
                - ${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1}
                - ${KAFKA_TOPIC_GB32960_RAW:vehicle.raw.gb32960.v1}
      archive:
        enabled: ${SINK_ARCHIVE_ENABLED:true}
        type: local
        path: ${SINK_ARCHIVE_PATH:./archive/}
    event-file-store:
      enabled: ${EVENT_FILE_STORE_ENABLED:true}
      path: ${EVENT_FILE_STORE_PATH:./target/event-store/}
      zone-id: ${EVENT_FILE_STORE_ZONE_ID:Asia/Shanghai}
      batch-size: ${EVENT_FILE_STORE_BATCH_SIZE:500}
      flush-interval-millis: ${EVENT_FILE_STORE_FLUSH_INTERVAL_MILLIS:1000}
    event-history:
      enabled: true
    vehicle-state:
      enabled: false
    vehicle-stat:
      enabled: false

management:
  endpoints:
    web:
      exposure:
        include: health,info,metrics,prometheus,env
  health:
    redis:
      enabled: ${MANAGEMENT_HEALTH_REDIS_ENABLED:false}
  metrics:
    tags:
      application: vehicle-history-app
```

- [ ] **Step 5: Create the analytics app main class**

Create `modules/apps/vehicle-analytics-app/src/main/java/com/lingniu/ingest/analyticsapp/VehicleAnalyticsApplication.java`:

```java
package com.lingniu.ingest.analyticsapp;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication(scanBasePackages = "com.lingniu.ingest")
public class VehicleAnalyticsApplication {

    private static final Logger log = LoggerFactory.getLogger(VehicleAnalyticsApplication.class);

    public static void main(String[] args) {
        try {
            SpringApplication.run(VehicleAnalyticsApplication.class, args);
        } catch (Throwable t) {
            log.error("Vehicle analytics application failed to start, forcing JVM exit", t);
            System.exit(1);
        }
    }
}
```

- [ ] **Step 6: Create the analytics app config**

Create `modules/apps/vehicle-analytics-app/src/main/resources/application.yml`:

```yaml
spring:
  application:
    name: vehicle-analytics-app
  threads:
    virtual:
      enabled: true

server:
  port: ${HTTP_PORT:20300}

lingniu:
  ingest:
    gb32960:
      enabled: false
    sink:
      kafka:
        enabled: ${KAFKA_ENABLED:true}
        bootstrap-servers: ${KAFKA_BROKERS:114.55.58.251:9092}
        topics:
          realtime: ${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1}
          dlq: ${KAFKA_TOPIC_GB32960_DLQ:vehicle.dlq.gb32960.v1}
        consumer:
          enabled: ${KAFKA_CONSUMER_ENABLED:true}
          client-id-prefix: ${KAFKA_CONSUMER_CLIENT_ID_PREFIX:vehicle-analytics}
          auto-offset-reset: ${KAFKA_CONSUMER_AUTO_OFFSET_RESET:earliest}
          max-poll-records: ${KAFKA_CONSUMER_MAX_POLL_RECORDS:500}
          bindings:
            vehicleStateEnvelopeConsumerProcessor:
              enabled: ${VEHICLE_STATE_ENABLED:false}
              group-id: ${KAFKA_GROUP_STATE:vehicle-state}
              topics:
                - ${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1}
            vehicleStatEnvelopeConsumerProcessor:
              enabled: ${VEHICLE_STAT_ENABLED:true}
              group-id: ${KAFKA_GROUP_STAT:vehicle-stat}
              topics:
                - ${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1}
      archive:
        enabled: false
    event-file-store:
      enabled: false
    event-history:
      enabled: false
    vehicle-state:
      enabled: ${VEHICLE_STATE_ENABLED:false}
    vehicle-stat:
      enabled: ${VEHICLE_STAT_ENABLED:true}
      file-path: ${VEHICLE_STAT_FILE_PATH:./target/vehicle-stat/}
      zone-id: ${VEHICLE_STAT_ZONE_ID:Asia/Shanghai}

management:
  endpoints:
    web:
      exposure:
        include: health,info,metrics,prometheus,env
  health:
    redis:
      enabled: ${MANAGEMENT_HEALTH_REDIS_ENABLED:false}
  metrics:
    tags:
      application: vehicle-analytics-app
```

- [ ] **Step 7: Build all three app jars**

Run:

```bash
export JAVA_HOME=/opt/homebrew/opt/openjdk@25/libexec/openjdk.jdk/Contents/Home
export PATH="$JAVA_HOME/bin:/opt/homebrew/bin:$PATH"
mvn -pl :gb32960-ingest-app,:vehicle-history-app,:vehicle-analytics-app -am package -Dmaven.test.skip=true
```

Expected: BUILD SUCCESS and jars under each app module's `target/`.

- [ ] **Step 8: Commit**

```bash
git add modules/apps/gb32960-ingest-app modules/apps/vehicle-history-app modules/apps/vehicle-analytics-app
git commit -m "feat: add split service entrypoints"
```

---

### Task 3: Add Composition Tests for Service Boundaries

**Files:**
- Create: `modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/Gb32960IngestAppCompositionTest.java`
- Create: `modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppCompositionTest.java`
- Create: `modules/apps/vehicle-analytics-app/src/test/java/com/lingniu/ingest/analyticsapp/VehicleAnalyticsAppCompositionTest.java`

- [ ] **Step 1: Write GB32960 ingest composition test**

Create `modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/Gb32960IngestAppCompositionTest.java`:

```java
package com.lingniu.ingest.gb32960app;

import com.lingniu.ingest.eventfilestore.EventFileStore;
import com.lingniu.ingest.protocol.gb32960.config.Gb32960AutoConfiguration;
import com.lingniu.ingest.protocol.gb32960.inbound.Gb32960NettyServer;
import com.lingniu.ingest.sink.archive.ArchiveStore;
import com.lingniu.ingest.sink.kafka.KafkaEventSink;
import org.junit.jupiter.api.Test;
import org.springframework.boot.autoconfigure.AutoConfigurations;
import org.springframework.boot.test.context.runner.ApplicationContextRunner;

import static org.assertj.core.api.Assertions.assertThat;

class Gb32960IngestAppCompositionTest {

    @Test
    void gb32960IngestRuntimeStartsProtocolAndKafkaWithoutHistoryStorage() {
        new ApplicationContextRunner()
                .withConfiguration(AutoConfigurations.of(Gb32960AutoConfiguration.class))
                .withPropertyValues(
                        "lingniu.ingest.gb32960.enabled=true",
                        "lingniu.ingest.gb32960.port=0",
                        "lingniu.ingest.sink.kafka.enabled=false",
                        "lingniu.ingest.sink.archive.enabled=false",
                        "lingniu.ingest.event-file-store.enabled=false",
                        "lingniu.ingest.event-history.enabled=false",
                        "lingniu.ingest.vehicle-state.enabled=false",
                        "lingniu.ingest.vehicle-stat.enabled=false")
                .run(context -> {
                    assertThat(context).hasSingleBean(Gb32960NettyServer.class);
                    assertThat(context).doesNotHaveBean(ArchiveStore.class);
                    assertThat(context).doesNotHaveBean(EventFileStore.class);
                    assertThat(context).doesNotHaveBean(KafkaEventSink.class);
                });
    }
}
```

If this test does not create required supporting beans, add the same auto-config classes used by existing bootstrap tests: `IngestCoreAutoConfiguration`, `SessionCoreAutoConfiguration`, and `VehicleIdentityAutoConfiguration`.

- [ ] **Step 2: Run GB32960 ingest composition test**

Run:

```bash
mvn -pl :gb32960-ingest-app -Dtest=Gb32960IngestAppCompositionTest test
```

Expected after adding required auto-config classes: PASS.

- [ ] **Step 3: Write history composition test**

Create `modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppCompositionTest.java`:

```java
package com.lingniu.ingest.historyapp;

import com.lingniu.ingest.eventfilestore.EventFileStore;
import com.lingniu.ingest.eventhistory.EventHistoryController;
import com.lingniu.ingest.protocol.gb32960.inbound.Gb32960NettyServer;
import com.lingniu.ingest.sink.archive.ArchiveStore;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.springframework.boot.autoconfigure.AutoConfigurations;
import org.springframework.boot.test.context.runner.ApplicationContextRunner;

import java.nio.file.Path;

import static org.assertj.core.api.Assertions.assertThat;

class VehicleHistoryAppCompositionTest {

    @TempDir
    Path tempDir;

    @Test
    void historyRuntimeCreatesStorageAndQueryBeansWithoutProtocolListener() {
        new ApplicationContextRunner()
                .withPropertyValues(
                        "lingniu.ingest.gb32960.enabled=false",
                        "lingniu.ingest.sink.archive.enabled=true",
                        "lingniu.ingest.sink.archive.path=" + tempDir.resolve("archive"),
                        "lingniu.ingest.event-file-store.enabled=true",
                        "lingniu.ingest.event-file-store.path=" + tempDir.resolve("event-store"),
                        "lingniu.ingest.event-history.enabled=true",
                        "lingniu.ingest.vehicle-state.enabled=false",
                        "lingniu.ingest.vehicle-stat.enabled=false",
                        "lingniu.ingest.sink.kafka.enabled=false")
                .withConfiguration(AutoConfigurations.of(
                        com.lingniu.ingest.sink.archive.config.SinkArchiveAutoConfiguration.class,
                        com.lingniu.ingest.eventfilestore.config.EventFileStoreAutoConfiguration.class,
                        com.lingniu.ingest.eventhistory.config.EventHistoryAutoConfiguration.class))
                .run(context -> {
                    assertThat(context).hasSingleBean(ArchiveStore.class);
                    assertThat(context).hasSingleBean(EventFileStore.class);
                    assertThat(context).hasSingleBean(EventHistoryController.class);
                    assertThat(context).doesNotHaveBean(Gb32960NettyServer.class);
                });
    }
}
```

- [ ] **Step 4: Run history composition test**

Run:

```bash
mvn -pl :vehicle-history-app -Dtest=VehicleHistoryAppCompositionTest test
```

Expected: PASS.

- [ ] **Step 5: Write analytics composition test**

Create `modules/apps/vehicle-analytics-app/src/test/java/com/lingniu/ingest/analyticsapp/VehicleAnalyticsAppCompositionTest.java`:

```java
package com.lingniu.ingest.analyticsapp;

import com.lingniu.ingest.eventfilestore.EventFileStore;
import com.lingniu.ingest.protocol.gb32960.inbound.Gb32960NettyServer;
import com.lingniu.ingest.vehiclestat.DailyVehicleStatService;
import com.lingniu.ingest.vehiclestate.VehicleStateRepository;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.springframework.boot.autoconfigure.AutoConfigurations;
import org.springframework.boot.test.context.runner.ApplicationContextRunner;

import java.nio.file.Path;

import static org.assertj.core.api.Assertions.assertThat;

class VehicleAnalyticsAppCompositionTest {

    @TempDir
    Path tempDir;

    @Test
    void analyticsRuntimeCreatesStatisticsWithoutProtocolOrHistoryStorage() {
        new ApplicationContextRunner()
                .withPropertyValues(
                        "lingniu.ingest.gb32960.enabled=false",
                        "lingniu.ingest.sink.archive.enabled=false",
                        "lingniu.ingest.event-file-store.enabled=false",
                        "lingniu.ingest.event-history.enabled=false",
                        "lingniu.ingest.vehicle-state.enabled=false",
                        "lingniu.ingest.vehicle-stat.enabled=true",
                        "lingniu.ingest.vehicle-stat.file-path=" + tempDir.resolve("vehicle-stat"),
                        "lingniu.ingest.sink.kafka.enabled=false")
                .withConfiguration(AutoConfigurations.of(
                        com.lingniu.ingest.vehiclestat.config.VehicleStatAutoConfiguration.class,
                        com.lingniu.ingest.vehiclestate.config.VehicleStateAutoConfiguration.class))
                .run(context -> {
                    assertThat(context).hasSingleBean(DailyVehicleStatService.class);
                    assertThat(context).doesNotHaveBean(VehicleStateRepository.class);
                    assertThat(context).doesNotHaveBean(Gb32960NettyServer.class);
                    assertThat(context).doesNotHaveBean(EventFileStore.class);
                });
    }
}
```

- [ ] **Step 6: Run analytics composition test**

Run:

```bash
mvn -pl :vehicle-analytics-app -Dtest=VehicleAnalyticsAppCompositionTest test
```

Expected: PASS.

- [ ] **Step 7: Commit**

```bash
git add modules/apps/gb32960-ingest-app/src/test modules/apps/vehicle-history-app/src/test modules/apps/vehicle-analytics-app/src/test
git commit -m "test: cover split service composition"
```

---

### Task 4: Version Kafka Topics for GB32960 Raw/Event/DLQ

**Files:**
- Modify: `modules/sinks/sink-kafka/src/main/java/com/lingniu/ingest/sink/kafka/KafkaSinkProperties.java`
- Modify: `modules/sinks/sink-kafka/src/main/java/com/lingniu/ingest/sink/kafka/TopicRouter.java`
- Modify: `modules/sinks/sink-kafka/src/main/java/com/lingniu/ingest/sink/kafka/KafkaEventSink.java`
- Create or modify: `modules/sinks/sink-kafka/src/test/java/com/lingniu/ingest/sink/kafka/TopicRouterTest.java`
- Create or modify: `modules/sinks/sink-kafka/src/test/java/com/lingniu/ingest/sink/kafka/KafkaEventSinkTest.java`

- [ ] **Step 1: Write topic routing tests**

Create `modules/sinks/sink-kafka/src/test/java/com/lingniu/ingest/sink/kafka/TopicRouterTest.java` if it does not exist. Add tests proving GB32960 raw and normalized events can route to versioned topics. Use the existing `VehicleEvent` constructors from `EventFileStoreSinkTest` and `EnvelopeMapperTelemetrySnapshotTest` as examples.

Expected test names:

```java
@Test
void rawArchiveRoutesToVersionedGb32960RawTopic() {
    KafkaSinkProperties.Topics topics = new KafkaSinkProperties.Topics();
    topics.setRawArchive("vehicle.raw.gb32960.v1");
    TopicRouter router = new TopicRouter(topics);

    VehicleEvent.RawArchive raw = new VehicleEvent.RawArchive(
            "event-1",
            "trace-1",
            ProtocolId.GB32960,
            "VIN001",
            Instant.parse("2026-06-23T07:00:00Z"),
            0x02,
            new byte[] {0x23, 0x23},
            Map.of("platformAccount", "Hyundai"));

    assertThat(router.route(raw)).isEqualTo("vehicle.raw.gb32960.v1");
}
```

Also add a normalized realtime event test:

```java
@Test
void realtimeRoutesToVersionedGb32960EventTopic() {
    KafkaSinkProperties.Topics topics = new KafkaSinkProperties.Topics();
    topics.setRealtime("vehicle.event.gb32960.v1");
    TopicRouter router = new TopicRouter(topics);

    VehicleEvent.Realtime realtime = new VehicleEvent.Realtime(
            "event-2",
            "trace-2",
            ProtocolId.GB32960,
            "VIN001",
            Instant.parse("2026-06-23T07:00:01Z"),
            0x02,
            Map.of("soc", "80"),
            Map.of("platformAccount", "Hyundai"));

    assertThat(router.route(realtime)).isEqualTo("vehicle.event.gb32960.v1");
}
```

Adjust constructor arguments to match the current sealed `VehicleEvent` definitions in `modules/core/ingest-api/src/main/java/com/lingniu/ingest/api/event/VehicleEvent.java`.

- [ ] **Step 2: Run topic router test and verify failure if topic defaults are still old**

Run:

```bash
mvn -pl :sink-kafka -Dtest=TopicRouterTest test
```

Expected before implementation: compile failure if constructors are wrong or assertion failure if defaults are old. Fix constructors first; keep routing assertions.

- [ ] **Step 3: Update `KafkaSinkProperties.Topics` defaults**

In `KafkaSinkProperties.Topics`, change defaults:

```java
private String realtime = "vehicle.event.gb32960.v1";
private String location = "vehicle.event.gb32960.v1";
private String alarm = "vehicle.event.gb32960.v1";
private String session = "vehicle.event.gb32960.v1";
private String mediaMeta = "vehicle.media.meta.v1";
private String rawArchive = "vehicle.raw.gb32960.v1";
private String dlq = "vehicle.dlq.gb32960.v1";
```

- [ ] **Step 4: Allow raw archive records in KafkaEventSink**

Change `KafkaEventSink.accepts()` to:

```java
@Override
public boolean accepts(VehicleEvent event) {
    return true;
}
```

Update the JavaDoc above it to say Kafka carries both normalized events and raw archive records in split-service mode.

- [ ] **Step 5: Run sink-kafka tests**

Run:

```bash
mvn -pl :sink-kafka test
```

Expected: PASS.

- [ ] **Step 6: Commit**

```bash
git add modules/sinks/sink-kafka/src/main/java/com/lingniu/ingest/sink/kafka/KafkaSinkProperties.java \
        modules/sinks/sink-kafka/src/main/java/com/lingniu/ingest/sink/kafka/TopicRouter.java \
        modules/sinks/sink-kafka/src/main/java/com/lingniu/ingest/sink/kafka/KafkaEventSink.java \
        modules/sinks/sink-kafka/src/test/java/com/lingniu/ingest/sink/kafka/TopicRouterTest.java \
        modules/sinks/sink-kafka/src/test/java/com/lingniu/ingest/sink/kafka/KafkaEventSinkTest.java
git commit -m "feat: route gb32960 kafka topics"
```

---

### Task 5: Introduce an ACK-Aware GB32960 Dispatch Boundary

**Files:**
- Modify: `modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/dispatcher/Dispatcher.java`
- Modify: `modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/concurrency/DisruptorEventBus.java`
- Modify: `modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandler.java`
- Create: `modules/protocols/protocol-gb32960/src/test/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandlerAckBoundaryTest.java`

Design choice for first implementation:

- Keep `Dispatcher.dispatch(RawFrame)` for existing async protocols and `bootstrap-all`.
- Add `Dispatcher.dispatchAndAwait(RawFrame, Duration)` returning `CompletableFuture<Void>` or throwing a checked runtime exception on timeout/failure.
- Implement `DisruptorEventBus.publishAndAwait(VehicleEvent, Duration)` by synchronously calling accepted sinks' `publish()` futures and waiting for all to complete.
- Use the synchronous path only from GB32960 report/heartbeat/session commands that require ACK after Kafka success.

- [ ] **Step 1: Write failing ACK boundary test**

Create `Gb32960ChannelHandlerAckBoundaryTest` using `EmbeddedChannel` and a fake dispatcher that records whether ACK happens before the future completes.

Core assertion:

```java
@Test
void realtimeReportAckWaitsForKafkaDurabilityBoundary() {
    CompletableFuture<Void> durability = new CompletableFuture<>();
    RecordingAckService ackService = new RecordingAckService();
    AwaitingDispatcher dispatcher = new AwaitingDispatcher(durability);
    Gb32960ChannelHandler handler = handlerWith(dispatcher, ackService);

    EmbeddedChannel channel = new EmbeddedChannel(handler);
    channel.writeInbound(validRealtimeFrameBytes());

    assertThat(ackService.writtenTags()).isEmpty();

    durability.complete(null);
    channel.runPendingTasks();

    assertThat(ackService.writtenTags()).contains("report-ack-0x2");
}
```

Implementation notes:

- Reuse valid frame fixture builders from existing GB32960 codec tests.
- If `Gb32960AckService` is hard to spy because methods are concrete, subclass it in test and override `writeResponse`/`writeAndClose` to record tags instead of writing bytes.
- If `Dispatcher` is not subclassable, extract a small interface such as `FrameDispatcher` in `ingest-core` and adapt `Dispatcher` to implement it.

- [ ] **Step 2: Run the failing test**

Run:

```bash
mvn -pl :protocol-gb32960 -Dtest=Gb32960ChannelHandlerAckBoundaryTest test
```

Expected before implementation: FAIL because current handler writes ACK before dispatch completion.

- [ ] **Step 3: Add await API to event bus**

In `DisruptorEventBus`, add:

```java
public CompletableFuture<Void> publishAndAwait(VehicleEvent event) {
    List<CompletableFuture<Void>> futures = sinks.stream()
            .filter(sink -> sink.accepts(event))
            .map(sink -> sink.publish(event))
            .toList();
    if (futures.isEmpty()) {
        return CompletableFuture.completedFuture(null);
    }
    return CompletableFuture.allOf(futures.toArray(CompletableFuture[]::new));
}
```

This requires promoting `sinks` from constructor-local variable to a field:

```java
private final List<EventSink> sinks;
```

Set it in the constructor:

```java
this.sinks = List.copyOf(sinks);
```

- [ ] **Step 4: Add await API to Dispatcher**

In `Dispatcher`, add:

```java
public CompletableFuture<Void> dispatchAndAwait(RawFrame frame) {
    List<VehicleEvent> events = mapFrameToEvents(frame);
    CompletableFuture<?>[] futures = events.stream()
            .map(eventBus::publishAndAwait)
            .toArray(CompletableFuture[]::new);
    return CompletableFuture.allOf(futures);
}
```

Refactor existing `dispatch(RawFrame)` to reuse `mapFrameToEvents(frame)`:

```java
public void dispatch(RawFrame frame) {
    for (VehicleEvent e : mapFrameToEvents(frame)) {
        eventBus.publish(e);
    }
}
```

If `Dispatcher` currently interleaves raw archive event publication and handler mapping, move that logic into a private method:

```java
private List<VehicleEvent> mapFrameToEvents(RawFrame frame) {
    List<VehicleEvent> events = new ArrayList<>();
    events.add(toRawArchive(frame));
    events.addAll(handlerEvents(frame));
    return events;
}
```

Use exact existing helper names from `Dispatcher.java`; do not duplicate mapping logic.

- [ ] **Step 5: Move GB32960 ACK after dispatch completion**

In `Gb32960ChannelHandler.channelRead0`, for commands that currently ACK before `dispatcher.dispatch(rf)`, build `RawFrame` first, call `dispatcher.dispatchAndAwait(rf)`, and write ACK in the completion callback:

```java
dispatcher.dispatchAndAwait(rf).whenComplete((ignored, ex) -> {
    if (ex != null) {
        log.warn("[gb32960] dispatch failed before ack peer={} vin={} cmd=0x{}",
                addr(ctx), vin, Integer.toHexString(cmd.code()), ex);
        ctx.close();
        return;
    }
    writeAckForCommand(ctx, msg, rawVin);
});
```

Extract current ACK logic into command-specific method:

```java
private void writeAckForCommand(ChannelHandlerContext ctx, Gb32960Message msg, byte[] rawVin) {
    switch (msg.header().command()) {
        case VEHICLE_LOGIN -> writeVehicleLoginAck(ctx, msg, rawVin);
        case VEHICLE_LOGOUT -> writeVehicleLogoutAck(ctx, msg, rawVin);
        case PLATFORM_LOGIN -> writePlatformLoginAck(ctx, msg, rawVin);
        case PLATFORM_LOGOUT -> writePlatformLogoutAck(ctx, msg, rawVin);
        case REALTIME_REPORT, RESEND_REPORT, HEARTBEAT -> writeReportOrHeartbeatAck(ctx, msg, rawVin);
        case TIME_CALIBRATION -> writeTimeCalibrationAck(ctx, msg, rawVin);
        default -> { }
    }
}
```

Keep authorization failure ACKs immediate because they are not accepted frames and do not cross the Kafka durability boundary.

- [ ] **Step 6: Run ACK boundary test**

Run:

```bash
mvn -pl :protocol-gb32960 -Dtest=Gb32960ChannelHandlerAckBoundaryTest test
```

Expected: PASS.

- [ ] **Step 7: Run core dispatcher tests**

Run:

```bash
mvn -pl :ingest-core test
```

Expected: PASS.

- [ ] **Step 8: Commit**

```bash
git add modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/dispatcher/Dispatcher.java \
        modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/concurrency/DisruptorEventBus.java \
        modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandler.java \
        modules/protocols/protocol-gb32960/src/test/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandlerAckBoundaryTest.java
git commit -m "feat: ack gb32960 after kafka boundary"
```

---

### Task 6: Wire History and Analytics Consumer Bindings

**Files:**
- Modify: `modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestor.java`
- Modify: `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/VehicleStatEnvelopeIngestor.java`
- Modify: `modules/services/vehicle-state-service/src/main/java/com/lingniu/ingest/vehiclestate/VehicleStateEnvelopeIngestor.java`
- Modify tests under the same service modules.

- [ ] **Step 1: Verify event-history accepts raw and normalized envelopes**

Run:

```bash
rg -n "RawArchive|raw_archive|TelemetrySnapshot|Envelope" modules/services/event-history-service/src/main/java modules/services/event-history-service/src/test/java
```

Expected: identify whether `EventHistoryEnvelopeIngestor` already persists raw archive envelopes. If raw archive envelopes are ignored, continue with Step 2.

- [ ] **Step 2: Add event-history raw archive ingestion test**

Add a test to the existing event-history test suite that creates an envelope with `raw_archive` populated and asserts an `EventFileRecord` with `eventType = RAW_ARCHIVE` is written with `rawArchiveUri`.

Expected assertion shape:

```java
assertThat(record.eventType()).isEqualTo("RAW_ARCHIVE");
assertThat(record.rawArchiveUri()).isEqualTo("archive://2026/06/23/GB32960/VIN001/event-1.bin");
```

- [ ] **Step 3: Implement raw archive ingestion if missing**

In `EventHistoryEnvelopeIngestor`, map raw archive envelopes to `EventFileRecord` using the same conventions as `EventFileStoreSink`.

Implementation rule:

- Do not embed raw bytes into Parquet history records.
- Store `rawArchiveUri`, event id, protocol, VIN, command, event time, and metadata.

- [ ] **Step 4: Verify analytics consumers ignore raw topic records**

Add or update tests for `VehicleStatEnvelopeIngestor` and `VehicleStateEnvelopeIngestor`:

```java
@Test
void ignoresRawArchiveEnvelopeForAnalytics() {
    ConsumerProcessResult result = ingestor.ingest(rawArchiveEnvelopeRecord());
    assertThat(result.success()).isTrue();
    assertThat(repositoryWrites()).isEmpty();
}
```

Use the actual result type currently returned by the ingestor.

- [ ] **Step 5: Run service consumer tests**

Run:

```bash
mvn -pl :event-history-service,:vehicle-state-service,:vehicle-stat-service test
```

Expected: PASS.

- [ ] **Step 6: Commit**

```bash
git add modules/services/event-history-service modules/services/vehicle-state-service modules/services/vehicle-stat-service
git commit -m "feat: split history and analytics consumers"
```

---

### Task 7: Add App Default Tests

**Files:**
- Create: `modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/Gb32960IngestAppDefaultsTest.java`
- Create: `modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppDefaultsTest.java`
- Create: `modules/apps/vehicle-analytics-app/src/test/java/com/lingniu/ingest/analyticsapp/VehicleAnalyticsAppDefaultsTest.java`

- [ ] **Step 1: Add YAML property loader helper to each test**

Use this helper in each defaults test:

```java
private static Properties loadApplicationProperties() {
    YamlPropertiesFactoryBean yaml = new YamlPropertiesFactoryBean();
    yaml.setResources(new ClassPathResource("application.yml"));
    Properties props = yaml.getObject();
    assertThat(props).isNotNull();
    return props;
}
```

Imports:

```java
import org.springframework.beans.factory.config.YamlPropertiesFactoryBean;
import org.springframework.core.io.ClassPathResource;
import java.util.Properties;
import static org.assertj.core.api.Assertions.assertThat;
```

- [ ] **Step 2: Assert GB32960 ingest defaults**

Create `Gb32960IngestAppDefaultsTest`:

```java
@Test
void gb32960IngestDefaultsOnlyEnableProtocolAndKafkaProducer() {
    Properties props = loadApplicationProperties();

    assertThat(props.getProperty("spring.application.name")).isEqualTo("gb32960-ingest-app");
    assertThat(props.getProperty("lingniu.ingest.gb32960.enabled")).isEqualTo("true");
    assertThat(props.getProperty("lingniu.ingest.gb32960.port")).isEqualTo("${GB32960_PORT:32960}");
    assertThat(props.getProperty("lingniu.ingest.sink.kafka.enabled")).isEqualTo("${KAFKA_ENABLED:true}");
    assertThat(props.getProperty("lingniu.ingest.sink.kafka.consumer.enabled")).isEqualTo("false");
    assertThat(props.getProperty("lingniu.ingest.event-file-store.enabled")).isEqualTo("false");
    assertThat(props.getProperty("lingniu.ingest.event-history.enabled")).isEqualTo("false");
    assertThat(props.getProperty("lingniu.ingest.vehicle-state.enabled")).isEqualTo("false");
    assertThat(props.getProperty("lingniu.ingest.vehicle-stat.enabled")).isEqualTo("false");
}
```

- [ ] **Step 3: Assert history defaults**

Create `VehicleHistoryAppDefaultsTest`:

```java
@Test
void historyDefaultsEnableStorageAndHistoryConsumerOnly() {
    Properties props = loadApplicationProperties();

    assertThat(props.getProperty("spring.application.name")).isEqualTo("vehicle-history-app");
    assertThat(props.getProperty("lingniu.ingest.gb32960.enabled")).isEqualTo("false");
    assertThat(props.getProperty("lingniu.ingest.sink.archive.enabled")).isEqualTo("${SINK_ARCHIVE_ENABLED:true}");
    assertThat(props.getProperty("lingniu.ingest.event-file-store.enabled")).isEqualTo("${EVENT_FILE_STORE_ENABLED:true}");
    assertThat(props.getProperty("lingniu.ingest.event-history.enabled")).isEqualTo("true");
    assertThat(props.getProperty("lingniu.ingest.vehicle-state.enabled")).isEqualTo("false");
    assertThat(props.getProperty("lingniu.ingest.vehicle-stat.enabled")).isEqualTo("false");
    assertThat(props.getProperty("lingniu.ingest.sink.kafka.consumer.bindings.eventHistoryEnvelopeConsumerProcessor.group-id"))
            .isEqualTo("${KAFKA_GROUP_HISTORY:vehicle-history}");
}
```

- [ ] **Step 4: Assert analytics defaults**

Create `VehicleAnalyticsAppDefaultsTest`:

```java
@Test
void analyticsDefaultsEnableStatConsumerAndDisableProtocolAndHistoryStorage() {
    Properties props = loadApplicationProperties();

    assertThat(props.getProperty("spring.application.name")).isEqualTo("vehicle-analytics-app");
    assertThat(props.getProperty("lingniu.ingest.gb32960.enabled")).isEqualTo("false");
    assertThat(props.getProperty("lingniu.ingest.sink.archive.enabled")).isEqualTo("false");
    assertThat(props.getProperty("lingniu.ingest.event-file-store.enabled")).isEqualTo("false");
    assertThat(props.getProperty("lingniu.ingest.event-history.enabled")).isEqualTo("false");
    assertThat(props.getProperty("lingniu.ingest.vehicle-stat.enabled")).isEqualTo("${VEHICLE_STAT_ENABLED:true}");
    assertThat(props.getProperty("lingniu.ingest.sink.kafka.consumer.bindings.vehicleStatEnvelopeConsumerProcessor.group-id"))
            .isEqualTo("${KAFKA_GROUP_STAT:vehicle-stat}");
}
```

- [ ] **Step 5: Run app default tests**

Run:

```bash
mvn -pl :gb32960-ingest-app,:vehicle-history-app,:vehicle-analytics-app test
```

Expected: PASS.

- [ ] **Step 6: Commit**

```bash
git add modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/Gb32960IngestAppDefaultsTest.java \
        modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppDefaultsTest.java \
        modules/apps/vehicle-analytics-app/src/test/java/com/lingniu/ingest/analyticsapp/VehicleAnalyticsAppDefaultsTest.java
git commit -m "test: assert split app defaults"
```

---

### Task 8: End-to-End Verification With Local Kafka

**Files:**
- Create: `docs/operations/gb32960-service-split-runbook.md`

- [ ] **Step 1: Build all modules**

Run:

```bash
export JAVA_HOME=/opt/homebrew/opt/openjdk@25/libexec/openjdk.jdk/Contents/Home
export PATH="$JAVA_HOME/bin:/opt/homebrew/bin:$PATH"
mvn -pl :gb32960-ingest-app,:vehicle-history-app,:vehicle-analytics-app -am package -Dmaven.test.skip=true
```

Expected: BUILD SUCCESS.

- [ ] **Step 2: Start Kafka test environment**

Use the repository's existing Kafka setup if present. If there is no repository-local Kafka script, use a local Kafka compatible with `KAFKA_BROKERS=127.0.0.1:9092`.

Create topics:

```bash
kafka-topics --bootstrap-server 127.0.0.1:9092 --create --if-not-exists --topic vehicle.raw.gb32960.v1 --partitions 12 --replication-factor 1
kafka-topics --bootstrap-server 127.0.0.1:9092 --create --if-not-exists --topic vehicle.event.gb32960.v1 --partitions 12 --replication-factor 1
kafka-topics --bootstrap-server 127.0.0.1:9092 --create --if-not-exists --topic vehicle.dlq.gb32960.v1 --partitions 3 --replication-factor 1
```

- [ ] **Step 3: Start split services on local ports**

Run each command in a separate terminal:

```bash
KAFKA_BROKERS=127.0.0.1:9092 \
GB32960_PORT=32960 \
HTTP_PORT=20100 \
java --sun-misc-unsafe-memory-access=allow \
  -jar modules/apps/gb32960-ingest-app/target/gb32960-ingest-app.jar
```

```bash
KAFKA_BROKERS=127.0.0.1:9092 \
HTTP_PORT=20200 \
EVENT_FILE_STORE_PATH=./target/split-event-store \
SINK_ARCHIVE_PATH=./target/split-archive \
java --sun-misc-unsafe-memory-access=allow \
  -jar modules/apps/vehicle-history-app/target/vehicle-history-app.jar
```

```bash
KAFKA_BROKERS=127.0.0.1:9092 \
HTTP_PORT=20300 \
VEHICLE_STAT_FILE_PATH=./target/split-vehicle-stat \
java --sun-misc-unsafe-memory-access=allow \
  -jar modules/apps/vehicle-analytics-app/target/vehicle-analytics-app.jar
```

- [ ] **Step 4: Verify health endpoints**

Run:

```bash
curl -sS http://127.0.0.1:20100/actuator/health
curl -sS http://127.0.0.1:20200/actuator/health
curl -sS http://127.0.0.1:20300/actuator/health
```

Expected: each returns `{"status":"UP"}`.

- [ ] **Step 5: Send a known-valid GB32960 frame**

Use an existing GB32960 test fixture from `modules/protocols/protocol-gb32960/src/test/java`. If no command-line sender exists, add a small test-only utility or use `nc` with a hex-decoded fixture:

```bash
xxd -r -p /tmp/gb32960-valid-realtime.hex | nc 127.0.0.1 32960
```

Expected:

- Kafka topic `vehicle.raw.gb32960.v1` receives one raw record.
- Kafka topic `vehicle.event.gb32960.v1` receives one or more normalized records.
- Client receives a GB32960 success ACK.
- `target/split-archive` contains a `.bin` file.
- `target/split-event-store` contains `events.duckdb` or Parquet files after flush.
- `target/split-vehicle-stat` contains updated stat files after analytics consumes the event.

- [ ] **Step 6: Write runbook**

Create `docs/operations/gb32960-service-split-runbook.md` with:

```markdown
# GB32960 Split Service Runbook

## Services

- `gb32960-ingest-app`: TCP 32960, HTTP 20100
- `vehicle-history-app`: HTTP 20200
- `vehicle-analytics-app`: HTTP 20300

## Kafka Topics

- `vehicle.raw.gb32960.v1`
- `vehicle.event.gb32960.v1`
- `vehicle.dlq.gb32960.v1`

## ACK Semantics

GB32960 success ACK is sent after required Kafka production succeeds. History and analytics failures do not block ACK.

## Local Start

Use the commands from Task 8 Step 3.

## Verification

Use health endpoints, Kafka topic consumption, archive file count, event-store query APIs, analytics output files, and ACK latency metrics.

## Rollback

Stop split services and start `bootstrap-all` with the previous environment. Keep `bootstrap-all` available until split service counts and sample queries match production expectations.
```

- [ ] **Step 7: Commit**

```bash
git add docs/operations/gb32960-service-split-runbook.md
git commit -m "docs: add gb32960 split service runbook"
```

---

### Task 9: Final Verification and Cutover Checklist

**Files:**
- Modify: `docs/operations/gb32960-service-split-runbook.md`

- [ ] **Step 1: Run targeted tests**

Run:

```bash
mvn -pl :sink-kafka,:ingest-core,:protocol-gb32960,:event-history-service,:vehicle-state-service,:vehicle-stat-service test
```

Expected: BUILD SUCCESS.

- [ ] **Step 2: Run split app tests**

Run:

```bash
mvn -pl :gb32960-ingest-app,:vehicle-history-app,:vehicle-analytics-app test
```

Expected: BUILD SUCCESS.

- [ ] **Step 3: Run package verification**

Run:

```bash
mvn -pl :gb32960-ingest-app,:vehicle-history-app,:vehicle-analytics-app -am package -Dmaven.test.skip=true
```

Expected: BUILD SUCCESS.

- [ ] **Step 4: Update runbook with measured results**

Append a section:

```markdown
## Latest Local Verification

- Build:
- Unit tests:
- Split app startup:
- Kafka raw records:
- Kafka event records:
- Archive files:
- Event-store files:
- Analytics output:
- ACK behavior:
```

Fill the fields with the actual command outputs and counts from the run.

- [ ] **Step 5: Commit final verification docs**

```bash
git add docs/operations/gb32960-service-split-runbook.md
git commit -m "docs: record gb32960 split verification"
```

---

## Self-Review

Spec coverage:

- Three services are covered by Tasks 1, 2, 3, and 7.
- Kafka topic and routing contract is covered by Task 4.
- ACK after Kafka success is covered by Task 5.
- History consumption and storage is covered by Task 6 and Task 8.
- Analytics consumption and processing is covered by Task 6 and Task 8.
- `bootstrap-all` remains untouched except for pre-existing unrelated worktree changes.
- Verification and runbook are covered by Tasks 8 and 9.

Implementation risk notes:

- Task 5 is the most invasive task because it changes protocol ACK ordering. Implement it behind the smallest possible new API and keep `Dispatcher.dispatch(RawFrame)` behavior for other protocols.
- If Kafka transactions are needed to atomically produce raw and normalized records, add that as a follow-up after the split app boundary is working. The first implementation can use idempotent producer plus shared `eventId`/`sourceRawEventId`.
- If composition tests need additional auto-config classes, copy the exact pattern from `ProtocolAutoConfigurationCompositionTest` rather than widening app dependencies.