feat(projection-writer): Kafka consumer + Redis materialization per-locale

ordinis-projection-writer полностью реализован, last module: - OrdinisProjectionWriterApplication: Spring Boot main + JpaRepositories - RedisKeys: convention 'ord:<bundle>:<dictionary>:<bk>:<locale>' для flat плюс ':_raw' для multi-locale, плюс 'ord:idx:<bundle>:<dictionary>' set index - ProjectionWriter: для каждой supported locale справочника flatten data и записать в Redis. Также пишет _raw multi-locale ключ. На delete стирает все варианты + удаляет из index. Метрики ordinis_projection_records_* - RecordEventListener: @KafkaListener на 3 scope-топика, group ordinis-projection-cuod. Парсит EventEnvelope, разбирает payload.type (RecordCreated/Updated/Deleted), вызывает upsert или delete. - application.yml profile-based: dev (port-forward), k8s (Vault role ordinis-projection-writer) - logback-spring.xml как у других сервисов ordinis-events-api: добавлен явный 'type' accessor в каждом record (@JsonProperty type). Без него Jackson не сериализовал discriminator из @JsonTypeInfo (при generic T payload Jackson не знает declared type). ordinis-infra: Redis StatefulSet (redis:7.4-alpine, AOF, maxmemory 256mb LRU, secret password) в namespace ordinis-dev на :6379. E2E (3 сервиса параллельно): POST /spacecraft/records {AQUA-1, name:{ru-RU:'Аква',en-US:'Aqua'}, ...} → outbox row id=7 published_at=true → ordinis_outbox_published_total +1 → projection log: 'Projection upserted: spacecraft:AQUA-1 (2 locales)' → Redis ord:cuod:spacecraft:AQUA-1:ru-RU = {name:'Аква',...} → Redis ord:cuod:spacecraft:AQUA-1:en-US = {name:'Aqua',...} → Redis ord:cuod:spacecraft:AQUA-1:_raw = {name:{ru-RU,en-US},...} → Redis ord:idx:cuod:spacecraft set member 'AQUA-1' 10 of 10 модулей реализованы. Last module done.
2026-05-03 15:59:40 +03:00
parent ae69a7642b
commit 6a6067e70b
12 changed files with 469 additions and 5 deletions
@@ -0,0 +1,16 @@
+package cloud.nstart.terravault.ordinis.projection;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.autoconfigure.domain.EntityScan;
+import org.springframework.data.jpa.repository.config.EnableJpaRepositories;
+
+@SpringBootApplication(scanBasePackages = "cloud.nstart.terravault.ordinis")
+@EnableJpaRepositories(basePackages = "cloud.nstart.terravault.ordinis")
+@EntityScan(basePackages = "cloud.nstart.terravault.ordinis")
+public class OrdinisProjectionWriterApplication {
+
+  public static void main(String[] args) {
+    SpringApplication.run(OrdinisProjectionWriterApplication.class, args);
+  }
+}
@@ -0,0 +1,122 @@
+package cloud.nstart.terravault.ordinis.projection;
+
+import cloud.nstart.terravault.ordinis.domain.definition.DictionaryDefinition;
+import cloud.nstart.terravault.ordinis.domain.definition.DictionaryDefinitionRepository;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.MeterRegistry;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+/**
+ * Записывает projection в Redis: per-locale сплющенный JSON + raw multi-locale.
+ * При delete — удаляет все варианты ключей и снимает с индекса.
+ */
+@Component
+public class ProjectionWriter {
+
+  private static final Logger log = LoggerFactory.getLogger(ProjectionWriter.class);
+
+  private final StringRedisTemplate redis;
+  private final DictionaryDefinitionRepository definitionRepository;
+  private final ObjectMapper objectMapper;
+  private final Counter recordsWrittenCounter;
+  private final Counter recordsDeletedCounter;
+
+  public ProjectionWriter(
+      StringRedisTemplate redis,
+      DictionaryDefinitionRepository definitionRepository,
+      ObjectMapper objectMapper,
+      MeterRegistry meterRegistry) {
+    this.redis = redis;
+    this.definitionRepository = definitionRepository;
+    this.objectMapper = objectMapper;
+    this.recordsWrittenCounter = Counter.builder("ordinis_projection_records_written_total")
+        .description("Записано в Redis projection (per locale)")
+        .register(meterRegistry);
+    this.recordsDeletedCounter = Counter.builder("ordinis_projection_records_deleted_total")
+        .description("Удалено из Redis projection")
+        .register(meterRegistry);
+  }
+
+  @Transactional(readOnly = true)
+  public void upsert(String bundle, String dictionary, String businessKey, JsonNode data) throws Exception {
+    Optional<DictionaryDefinition> defOpt = definitionRepository.findByName(dictionary);
+    if (defOpt.isEmpty()) {
+      log.warn("Projection upsert: definition '{}' не найдена, skip", dictionary);
+      return;
+    }
+    DictionaryDefinition def = defOpt.get();
+    JsonNode schema = def.getSchemaJson();
+    String[] locales = def.getSupportedLocales();
+
+    String rawKey = RedisKeys.recordRaw(bundle, dictionary, businessKey);
+    redis.opsForValue().set(rawKey, objectMapper.writeValueAsString(data));
+
+    for (String locale : locales) {
+      JsonNode flat = flatten(schema, data, locale, def.getDefaultLocale());
+      String key = RedisKeys.record(bundle, dictionary, businessKey, locale);
+      redis.opsForValue().set(key, objectMapper.writeValueAsString(flat));
+      recordsWrittenCounter.increment();
+    }
+
+    redis.opsForSet().add(RedisKeys.dictionaryIndex(bundle, dictionary), businessKey);
+    log.debug("Projection upserted: {}:{} ({} locales)", dictionary, businessKey, locales.length);
+  }
+
+  public void delete(String bundle, String dictionary, String businessKey) {
+    Set<String> matching = redis.keys(RedisKeys.recordKeyPattern(bundle, dictionary, businessKey));
+    if (matching != null && !matching.isEmpty()) {
+      redis.delete(matching);
+      recordsDeletedCounter.increment(matching.size());
+    }
+    redis.opsForSet().remove(RedisKeys.dictionaryIndex(bundle, dictionary), businessKey);
+    log.debug("Projection deleted: {}:{}", dictionary, businessKey);
+  }
+
+  // --- mirror flattener (упрощённая версия из read-api, чтобы не тянуть зависимость) ---
+  private JsonNode flatten(JsonNode schema, JsonNode data, String locale, String defaultLocale) {
+    if (data == null || !data.isObject()) return data;
+    ObjectNode out = data.deepCopy();
+    walk(schema, out, locale, defaultLocale);
+    return out;
+  }
+
+  private void walk(JsonNode schema, ObjectNode data, String locale, String defaultLocale) {
+    if (schema == null || !schema.isObject()) return;
+    JsonNode properties = schema.get("properties");
+    if (properties == null || !properties.isObject()) return;
+    Iterator<Map.Entry<String, JsonNode>> it = properties.fields();
+    while (it.hasNext()) {
+      var e = it.next();
+      String name = e.getKey();
+      JsonNode fieldSchema = e.getValue();
+      JsonNode xLoc = fieldSchema.get("x-localized");
+      if (xLoc != null && xLoc.isBoolean() && xLoc.asBoolean()) {
+        flattenField(data, name, locale, defaultLocale);
+      } else if (fieldSchema.has("properties") && data.has(name) && data.get(name).isObject()) {
+        walk(fieldSchema, (ObjectNode) data.get(name), locale, defaultLocale);
+      }
+    }
+  }
+
+  private void flattenField(ObjectNode data, String field, String locale, String defaultLocale) {
+    JsonNode v = data.get(field);
+    if (v == null || !v.isObject()) return;
+    JsonNode picked = v.get(locale);
+    if (picked == null || picked.isNull()) picked = v.get(defaultLocale);
+    if (picked == null || picked.isNull()) data.set(field, objectMapper.nullNode());
+    else data.set(field, picked);
+  }
+}
@@ -0,0 +1,89 @@
+package cloud.nstart.terravault.ordinis.projection;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.MeterRegistry;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.support.KafkaHeaders;
+import org.springframework.messaging.handler.annotation.Header;
+import org.springframework.messaging.handler.annotation.Payload;
+import org.springframework.stereotype.Component;
+
+/**
+ * Подписан на 3 scope-топика. На RecordCreated/Updated делает upsert
+ * projection, на Deleted — delete. На Definition events ничего не делает
+ * (projection writer не материализует definitions).
+ *
+ * <p>Группа consumer'а: {@code ordinis-projection-cuod}. ACL Strimzi даёт
+ * этому пользователю Read доступ ко всем 3 scope-топикам и Read к группе
+ * {@code ordinis-projection-*} (см. ordinis-infra/k8s/kafka/topology.yaml).
+ */
+@Component
+public class RecordEventListener {
+
+  private static final Logger log = LoggerFactory.getLogger(RecordEventListener.class);
+
+  private final ProjectionWriter writer;
+  private final ObjectMapper objectMapper;
+  private final Counter eventsCounter;
+  private final Counter errorsCounter;
+
+  public RecordEventListener(
+      ProjectionWriter writer, ObjectMapper objectMapper, MeterRegistry meterRegistry) {
+    this.writer = writer;
+    this.objectMapper = objectMapper;
+    this.eventsCounter = Counter.builder("ordinis_projection_events_processed_total")
+        .description("Kafka events обработанные projection writer'ом")
+        .register(meterRegistry);
+    this.errorsCounter = Counter.builder("ordinis_projection_events_errors_total")
+        .description("Ошибки обработки event'a в projection writer'е")
+        .register(meterRegistry);
+  }
+
+  @KafkaListener(
+      topics = {
+          "ordinis.cuod.events.public",
+          "ordinis.cuod.events.internal",
+          "ordinis.cuod.events.restricted"
+      },
+      groupId = "ordinis-projection-cuod",
+      containerFactory = "kafkaListenerContainerFactory"
+  )
+  public void onMessage(
+      @Payload String value,
+      @Header(name = KafkaHeaders.RECEIVED_KEY, required = false) String key) {
+    try {
+      JsonNode envelope = objectMapper.readTree(value);
+      String dictionary = envelope.path("dictionaryName").asText(null);
+      String bundle = envelope.path("bundle").asText("cuod");
+      JsonNode payload = envelope.path("payload");
+      String type = payload.path("type").asText(null);
+
+      if (type == null || dictionary == null) {
+        log.debug("Skip event (no type or dictionary): key={}", key);
+        return;
+      }
+
+      String businessKey = payload.path("businessKey").asText(null);
+      switch (type) {
+        case "RecordCreated", "RecordUpdated" -> {
+          if (businessKey == null) return;
+          JsonNode data = payload.path("data");
+          writer.upsert(bundle, dictionary, businessKey, data);
+        }
+        case "RecordDeleted" -> {
+          if (businessKey == null) return;
+          writer.delete(bundle, dictionary, businessKey);
+        }
+        default -> log.debug("Unhandled event type {} for {}", type, dictionary);
+      }
+      eventsCounter.increment();
+    } catch (Exception e) {
+      errorsCounter.increment();
+      log.error("Failed to process projection event key={}: {}", key, e.getMessage(), e);
+    }
+  }
+}
@@ -0,0 +1,40 @@
+package cloud.nstart.terravault.ordinis.projection;
+
+/**
+ * Конвенция ключей в Redis для projection.
+ *
+ * <p><b>Per-locale flat record:</b><br>
+ * {@code ord:<bundle>:<dictionary>:<businessKey>:<locale>}<br>
+ * Value = JSON со сплющенным data на эту локаль (готов к отдаче без processing).
+ *
+ * <p><b>Multi-locale raw record:</b><br>
+ * {@code ord:<bundle>:<dictionary>:<businessKey>:_raw}<br>
+ * Value = полный multi-locale JSON. Используется когда client запросил {@code Accept-Language: *}.
+ *
+ * <p><b>Index по dictionary (set):</b><br>
+ * {@code ord:idx:<bundle>:<dictionary>}<br>
+ * Members = businessKey'и активных записей. Для list endpoint.
+ *
+ * <p><b>Last seen kafka offset (для idempotency consumer'a):</b><br>
+ * {@code ord:offset:<topic>:<partition>}
+ */
+public final class RedisKeys {
+
+  private RedisKeys() {}
+
+  public static String record(String bundle, String dictionary, String businessKey, String locale) {
+    return "ord:" + bundle + ":" + dictionary + ":" + businessKey + ":" + locale;
+  }
+
+  public static String recordRaw(String bundle, String dictionary, String businessKey) {
+    return "ord:" + bundle + ":" + dictionary + ":" + businessKey + ":_raw";
+  }
+
+  public static String dictionaryIndex(String bundle, String dictionary) {
+    return "ord:idx:" + bundle + ":" + dictionary;
+  }
+
+  public static String recordKeyPattern(String bundle, String dictionary, String businessKey) {
+    return "ord:" + bundle + ":" + dictionary + ":" + businessKey + ":*";
+  }
+}
@@ -0,0 +1,121 @@
+spring:
+  application:
+    name: ordinis-projection-writer
+  profiles:
+    active: ${SPRING_PROFILES_ACTIVE:dev}
+
+server:
+  port: 8082
+  shutdown: graceful
+
+management:
+  endpoints:
+    web:
+      exposure:
+        include: health,info,prometheus,refresh,metrics
+  endpoint:
+    health:
+      probes:
+        enabled: true
+      show-details: when-authorized
+      group:
+        liveness:
+          include: livenessState
+        readiness:
+          include: readinessState,redis
+  prometheus:
+    metrics:
+      export:
+        enabled: true
+  metrics:
+    tags:
+      application: ${spring.application.name}
+      environment: ${ENVIRONMENT:dev}
+
+logging:
+  level:
+    root: INFO
+    cloud.nstart.terravault.ordinis: DEBUG
+
+---
+spring:
+  config:
+    activate:
+      on-profile: dev
+  cloud:
+    config:
+      enabled: false
+    vault:
+      enabled: false
+  autoconfigure:
+    exclude:
+      - org.springframework.cloud.vault.config.VaultAutoConfiguration
+      - org.springframework.cloud.vault.config.VaultObservationAutoConfiguration
+      - org.springframework.cloud.vault.config.VaultReactiveAutoConfiguration
+  datasource:
+    url: jdbc:postgresql://${ORDINIS_PG_HOST:localhost}:${ORDINIS_PG_PORT:5433}/${ORDINIS_PG_DB:ordinis}
+    username: ${ORDINIS_PG_USER:ordinis_app}
+    password: ${ORDINIS_PG_PASSWORD:}
+    driver-class-name: org.postgresql.Driver
+    hikari:
+      read-only: true
+  jpa:
+    hibernate:
+      ddl-auto: none
+    open-in-view: false
+  data:
+    redis:
+      host: ${ORDINIS_REDIS_HOST:localhost}
+      port: ${ORDINIS_REDIS_PORT:6379}
+      password: ${ORDINIS_REDIS_PASSWORD:}
+      timeout: 2s
+  kafka:
+    bootstrap-servers: ${ORDINIS_KAFKA_BOOTSTRAP:localhost:32094}
+    properties:
+      security.protocol: ${ORDINIS_KAFKA_SECURITY:SASL_PLAINTEXT}
+      sasl.mechanism: ${ORDINIS_KAFKA_SASL_MECHANISM:SCRAM-SHA-512}
+      sasl.jaas.config: >-
+        org.apache.kafka.common.security.scram.ScramLoginModule required
+        username="${ORDINIS_KAFKA_USER:ordinis-projection-writer}"
+        password="${ORDINIS_KAFKA_PASSWORD:}";
+    consumer:
+      group-id: ordinis-projection-cuod
+      auto-offset-reset: earliest
+      enable-auto-commit: false
+      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+    listener:
+      ack-mode: record
+
+otel:
+  sdk:
+    disabled: true
+
+---
+spring:
+  config:
+    activate:
+      on-profile: k8s
+    import:
+      - "configserver:${SPRING_CLOUD_CONFIG_URI:http://config-server.ordinis-dev:8888}"
+      - "vault://"
+  cloud:
+    config:
+      enabled: true
+    vault:
+      enabled: true
+      uri: ${VAULT_URI:http://vault.config:8200}
+      authentication: KUBERNETES
+      kubernetes:
+        role: ordinis-projection-writer
+        service-account-token-file: /var/run/secrets/kubernetes.io/serviceaccount/token
+      kv:
+        enabled: true
+        backend: secret
+        application-name: ordinis/cuod
+  datasource:
+    hikari:
+      read-only: true
+  jpa:
+    hibernate:
+      ddl-auto: none
@@ -0,0 +1,35 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<configuration>
+  <springProperty scope="context" name="appName" source="spring.application.name" defaultValue="ordinis-projection-writer"/>
+  <springProperty scope="context" name="environment" source="ENVIRONMENT" defaultValue="dev"/>
+
+  <appender name="STDOUT_JSON" class="ch.qos.logback.core.ConsoleAppender">
+    <encoder class="net.logstash.logback.encoder.LogstashEncoder">
+      <includeMdcKeyName>trace_id</includeMdcKeyName>
+      <includeMdcKeyName>span_id</includeMdcKeyName>
+      <includeMdcKeyName>scope</includeMdcKeyName>
+      <includeMdcKeyName>dictionary</includeMdcKeyName>
+      <customFields>{"service":"${appName}","environment":"${environment}"}</customFields>
+    </encoder>
+  </appender>
+
+  <appender name="STDOUT_HUMAN" class="ch.qos.logback.core.ConsoleAppender">
+    <encoder>
+      <pattern>%d{HH:mm:ss.SSS} %-5level [%thread] %logger{36} - %msg%n</pattern>
+    </encoder>
+  </appender>
+
+  <springProfile name="dev">
+    <root level="INFO">
+      <appender-ref ref="STDOUT_HUMAN"/>
+    </root>
+    <logger name="cloud.nstart.terravault.ordinis" level="DEBUG"/>
+  </springProfile>
+
+  <springProfile name="!dev">
+    <root level="INFO">
+      <appender-ref ref="STDOUT_JSON"/>
+    </root>
+    <logger name="cloud.nstart.terravault.ordinis" level="DEBUG"/>
+  </springProfile>
+</configuration>