feat(validation,rest-api,outbox): write-side end-to-end working

ordinis-validation: - ValidationError, ValidationResult, ValidationException - RecordValidator: networknt JSON Schema (Draft-7) + custom rule для x-localized полей. Walk schema, для каждого x-localized:true property проверяет: значение это object с locale keys (BCP 47 pattern xx-XX), все ключи в supported_locales, default_locale обязательно присутствует, значения non-empty strings. ordinis-outbox: - KafkaTopicResolver: ordinis.<bundle>.events.<scope> - OutboxRecorder: запись события в outbox в той же транзакции (MANDATORY propagation), с tracer для trace_id/span_id из MDC если OTel доступен - OutboxPoller: @Scheduled, batch_size + poll_interval из config, метрики ordinis_outbox_published_total / publish_errors_total / kafka_publish_duration_seconds - OutboxLagGauge: ordinis_outbox_pending_events (gauge для алертов) - ConditionalOnProperty ordinis.outbox.enabled — отключаем в read-api/projection-writer ordinis-rest-api: - DTOs: CreateDictionaryRequest, DictionaryResponse, CreateRecordRequest, RecordResponse (с WKT serialization geometry) - DictionaryDefinitionService: CRUD definitions, outbox events DefinitionCreated/Updated - DictionaryRecordService: bitemporal write-side, SERIALIZABLE isolation, bound check FAR_FUTURE valid_to. Update = close current + create new (никогда in-place). Outbox event RecordCreated/Updated/Deleted. - DictionaryDefinitionController: GET list/by-name, POST create, PUT update - DictionaryRecordController: GET active/history, POST create, PUT update, DELETE (close) - OrdinisException + ApiErrorResponse + GlobalExceptionHandler (422 validation, 404 not found, 409 conflict, 500 internal) ordinis-app: - @EnableJpaRepositories + @EntityScan для multi-package сканирования - application.yml profile-based: dev (PG localhost через port-forward, ddl-auto=update, OTel disabled), k8s (Cloud Config + Vault Kubernetes auth) - spring.kafka producer config с SASL SCRAM-SHA-512 ordinis-domain: - JpaAuditingConfig: добавлен DateTimeProvider возвращающий OffsetDateTime (Spring Data Auditing по дефолту его не поддерживает, без него падает IllegalArgumentException 'Cannot convert LocalDateTime to OffsetDateTime') - @ConditionalOnBean(DataSource) убран — race с lifecycle ломал auditing E2E test (port-forward к cluster PG+Kafka): - POST dictionary 'ground_station' → 201, JSONB schema сохранена, supported_locales = {ru-RU, en-US}, default_locale = ru-RU - POST record MOSCOW-1 multi-locale name + geometry POINT(37.618 55.751) → 201 - POST record с en-US без ru-RU → 422 с двумя ошибками: * networknt 'required property ru-RU not found' * custom 'x-localized.missing_default Default locale ru-RU is required' - outbox_events: 2 row (DefinitionCreated + RecordCreated), topic ordinis.cuod.events.public, ключи правильные - revinfo + dictionary_records_aud: 1 row (Envers tx-time history активен)
2026-05-03 13:04:36 +03:00
parent b778832f49
commit be77c30687
24 changed files with 1246 additions and 19 deletions
@@ -23,6 +23,18 @@
      <groupId>org.springframework.kafka</groupId>
      <artifactId>spring-kafka</artifactId>
    </dependency>
+    <dependency>
+      <groupId>com.fasterxml.jackson.core</groupId>
+      <artifactId>jackson-databind</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>io.micrometer</groupId>
+      <artifactId>micrometer-core</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>io.micrometer</groupId>
+      <artifactId>micrometer-tracing</artifactId>
+    </dependency>
    <dependency>
      <groupId>cloud.nstart.terravault.ordinis</groupId>
      <artifactId>ordinis-domain</artifactId>
@@ -0,0 +1,20 @@
+package cloud.nstart.terravault.ordinis.outbox;
+
+import cloud.nstart.terravault.ordinis.domain.DataScope;
+
+/**
+ * Маршрутизация события в правильный scope-топик. Pattern:
+ * {@code ordinis.<bundle>.events.<scope>}.
+ *
+ * <p>Bundle сейчас всегда "cuod" (anchor customer); для второго клиента —
+ * другой bundle, своя тройка топиков. ACL/provisioning через Strimzi (см.
+ * ordinis-infra/k8s/kafka/topology.yaml).
+ */
+public final class KafkaTopicResolver {
+
+  private KafkaTopicResolver() {}
+
+  public static String topic(String bundle, DataScope scope) {
+    return "ordinis." + bundle + ".events." + scope.topicSuffix();
+  }
+}
@@ -0,0 +1,33 @@
+package cloud.nstart.terravault.ordinis.outbox;
+
+import cloud.nstart.terravault.ordinis.domain.outbox.OutboxEventRepository;
+import io.micrometer.core.instrument.Gauge;
+import io.micrometer.core.instrument.MeterRegistry;
+import jakarta.annotation.PostConstruct;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.stereotype.Component;
+
+/**
+ * Gauge {@code ordinis_outbox_pending_events} — сколько unpublished. Алертится
+ * Prometheus'ом если растёт (poller не успевает или Kafka недоступна).
+ */
+@Component
+@ConditionalOnProperty(name = "ordinis.outbox.enabled", havingValue = "true", matchIfMissing = false)
+public class OutboxLagGauge {
+
+  private final OutboxEventRepository repository;
+  private final MeterRegistry meterRegistry;
+
+  public OutboxLagGauge(OutboxEventRepository repository, MeterRegistry meterRegistry) {
+    this.repository = repository;
+    this.meterRegistry = meterRegistry;
+  }
+
+  @PostConstruct
+  void register() {
+    Gauge.builder("ordinis_outbox_pending_events", repository,
+            r -> (double) r.countByPublishedAtIsNull())
+        .description("Количество outbox events ожидающих публикацию в Kafka")
+        .register(meterRegistry);
+  }
+}
@@ -0,0 +1,112 @@
+package cloud.nstart.terravault.ordinis.outbox;
+
+import cloud.nstart.terravault.ordinis.domain.outbox.OutboxEvent;
+import cloud.nstart.terravault.ordinis.domain.outbox.OutboxEventRepository;
+import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Timer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.data.domain.PageRequest;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.support.SendResult;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+/**
+ * @Scheduled poller для outbox. Sweep каждые {@code ordinis.outbox.poll-interval-ms},
+ * batch size {@code ordinis.outbox.batch-size}. Активируется только если
+ * {@code ordinis.outbox.enabled=true} (отключаем в read-api / projection-writer).
+ */
+@Component
+@ConditionalOnProperty(name = "ordinis.outbox.enabled", havingValue = "true", matchIfMissing = false)
+public class OutboxPoller {
+
+  private static final Logger log = LoggerFactory.getLogger(OutboxPoller.class);
+
+  private final OutboxEventRepository repository;
+  private final KafkaTemplate<String, String> kafkaTemplate;
+  private final com.fasterxml.jackson.databind.ObjectMapper objectMapper;
+  private final Counter publishedCounter;
+  private final Counter failedCounter;
+  private final Timer publishTimer;
+
+  @Value("${ordinis.outbox.batch-size:100}")
+  private int batchSize;
+
+  @Value("${ordinis.outbox.send-timeout-ms:10000}")
+  private long sendTimeoutMs;
+
+  public OutboxPoller(
+      OutboxEventRepository repository,
+      KafkaTemplate<String, String> kafkaTemplate,
+      com.fasterxml.jackson.databind.ObjectMapper objectMapper,
+      MeterRegistry meterRegistry) {
+    this.repository = repository;
+    this.kafkaTemplate = kafkaTemplate;
+    this.objectMapper = objectMapper;
+    this.publishedCounter = Counter.builder("ordinis_outbox_published_total")
+        .description("Количество успешно опубликованных outbox events")
+        .register(meterRegistry);
+    this.failedCounter = Counter.builder("ordinis_outbox_publish_errors_total")
+        .description("Количество ошибок публикации в Kafka")
+        .register(meterRegistry);
+    this.publishTimer = Timer.builder("ordinis_kafka_publish_duration_seconds")
+        .description("Время публикации одного события в Kafka")
+        .register(meterRegistry);
+  }
+
+  @Scheduled(fixedDelayString = "${ordinis.outbox.poll-interval-ms:1000}")
+  @Transactional
+  public void poll() {
+    List<OutboxEvent> batch = repository.findUnpublished(PageRequest.of(0, batchSize));
+    if (batch.isEmpty()) return;
+
+    log.debug("Outbox poll: {} unpublished events", batch.size());
+
+    for (OutboxEvent event : batch) {
+      Timer.Sample sample = Timer.start();
+      try {
+        String value = objectMapper.writeValueAsString(event.getPayload());
+        SendResult<String, String> result = kafkaTemplate
+            .send(event.getKafkaTopic(), event.getKafkaKey(), value)
+            .get(sendTimeoutMs, TimeUnit.MILLISECONDS);
+
+        event.markPublished();
+        repository.save(event);
+        publishedCounter.increment();
+        sample.stop(publishTimer);
+
+        log.debug("Published outbox event id={} topic={} partition={} offset={}",
+            event.getId(),
+            event.getKafkaTopic(),
+            result.getRecordMetadata().partition(),
+            result.getRecordMetadata().offset());
+      } catch (InterruptedException e) {
+        Thread.currentThread().interrupt();
+        markFailed(event, "Interrupted: " + e.getMessage());
+        return;
+      } catch (ExecutionException | TimeoutException e) {
+        markFailed(event, e.getClass().getSimpleName() + ": " + e.getMessage());
+      } catch (Exception e) {
+        markFailed(event, e.getClass().getSimpleName() + ": " + e.getMessage());
+      }
+    }
+  }
+
+  private void markFailed(OutboxEvent event, String error) {
+    log.warn("Failed to publish outbox event id={} topic={}: {}",
+        event.getId(), event.getKafkaTopic(), error);
+    event.markFailed(error);
+    repository.save(event);
+    failedCounter.increment();
+  }
+}
@@ -0,0 +1,75 @@
+package cloud.nstart.terravault.ordinis.outbox;
+
+import cloud.nstart.terravault.ordinis.domain.DataScope;
+import cloud.nstart.terravault.ordinis.domain.outbox.OutboxEvent;
+import cloud.nstart.terravault.ordinis.domain.outbox.OutboxEventRepository;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.micrometer.tracing.Tracer;
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Transactional;
+
+/**
+ * Запись события в outbox в той же транзакции что и изменение справочной
+ * записи. Гарантирует at-least-once публикацию (poller потом отправит в Kafka).
+ *
+ * <p>Tracer optional — если в кластере включён OTel, prefix trace_id/span_id
+ * прокинется в outbox row для корреляции.
+ */
+@Component
+public class OutboxRecorder {
+
+  private final OutboxEventRepository repository;
+  private final ObjectMapper objectMapper;
+  private final ObjectProvider<Tracer> tracerProvider;
+
+  public OutboxRecorder(
+      OutboxEventRepository repository,
+      ObjectMapper objectMapper,
+      ObjectProvider<Tracer> tracerProvider) {
+    this.repository = repository;
+    this.objectMapper = objectMapper;
+    this.tracerProvider = tracerProvider;
+  }
+
+  /**
+   * @param eventType строковый тип (например "RecordCreated")
+   * @param aggregateType "DictionaryRecord" или "DictionaryDefinition"
+   * @param aggregateId стабильный id аггрегата (UUID или dictionary name)
+   * @param dictionaryName имя словаря (для рутинга в Kafka топик и для логов)
+   * @param scope scope записи → определяет в какой топик
+   * @param bundle bundle (cuod, energy, ...)
+   * @param kafkaKey ключ публикации (обычно {@code <dictionaryName>:<businessKey>})
+   * @param payload объект-payload (будет сериализован в JsonNode)
+   */
+  @Transactional(propagation = org.springframework.transaction.annotation.Propagation.MANDATORY)
+  public OutboxEvent record(
+      String eventType,
+      String aggregateType,
+      String aggregateId,
+      String dictionaryName,
+      DataScope scope,
+      String bundle,
+      String kafkaKey,
+      Object payload) {
+
+    JsonNode payloadJson = objectMapper.valueToTree(payload);
+    String topic = KafkaTopicResolver.topic(bundle, scope);
+
+    OutboxEvent event = new OutboxEvent(
+        eventType, aggregateType, aggregateId, scope, payloadJson, topic, kafkaKey);
+    event.setDictionaryName(dictionaryName);
+
+    Tracer tracer = tracerProvider.getIfAvailable();
+    if (tracer != null) {
+      var span = tracer.currentSpan();
+      if (span != null) {
+        event.setTraceId(span.context().traceId());
+        event.setSpanId(span.context().spanId());
+      }
+    }
+
+    return repository.save(event);
+  }
+}