package cloud.nstart.terravault.ordinis.restapi.service.schemaworkflow;

import cloud.nstart.terravault.ordinis.domain.definition.DictionaryDefinition;
import cloud.nstart.terravault.ordinis.domain.schemaworkflow.DictionarySchemaDraft;
import cloud.nstart.terravault.ordinis.domain.schemaworkflow.SchemaDraftRepository;
import cloud.nstart.terravault.ordinis.domain.schemaworkflow.SchemaDraftStatus;
import cloud.nstart.terravault.ordinis.events.EventEnvelope;
import cloud.nstart.terravault.ordinis.outbox.OutboxRecorder;
import cloud.nstart.terravault.ordinis.restapi.audit.AuditLogger;
import cloud.nstart.terravault.ordinis.restapi.auth.WorkflowRoles;
import cloud.nstart.terravault.ordinis.restapi.dto.schemaworkflow.CreateSchemaDraftRequest;
import cloud.nstart.terravault.ordinis.restapi.dto.schemaworkflow.DecisionRequest;
import cloud.nstart.terravault.ordinis.restapi.dto.schemaworkflow.PublishDraftRequest;
import cloud.nstart.terravault.ordinis.restapi.dto.schemaworkflow.SubmitForReviewRequest;
import cloud.nstart.terravault.ordinis.restapi.dto.schemaworkflow.UpdateSchemaDraftRequest;
import cloud.nstart.terravault.ordinis.restapi.error.OrdinisException;
import cloud.nstart.terravault.ordinis.restapi.service.DictionaryDefinitionService;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ArrayNode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.dao.DataIntegrityViolationException;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.Pageable;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Isolation;
import org.springframework.transaction.annotation.Transactional;

import java.time.OffsetDateTime;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;

/**
 * Dictionary schema workflow API — backend pending endpoint #3.
 *
 * <p>State machine: см. {@link SchemaDraftStatus}.
 *
 * <p>Endpoints (см. {@link cloud.nstart.terravault.ordinis.restapi.web.SchemaDraftController}):
 * <ul>
 *   <li>{@link #create} — maker создаёт draft от current HEAD</li>
 *   <li>{@link #submitForReview} — draft → review_pending</li>
 *   <li>{@link #decide} — reviewer approves / requests changes / rejects</li>
 *   <li>{@link #publish} — approved draft → live (bumps schema version)</li>
 *   <li>{@link #withdraw} — maker отменяет non-terminal draft</li>
 * </ul>
 *
 * <p>Все mutator-методы emit'ят outbox event ({@code SchemaDraft*}) и audit
 * (через {@link DictionaryDefinitionService} updateSchema path для publish).
 *
 * <p>Outbox event types следуют existing naming convention (PascalCase):
 * {@code SchemaDraftCreated}, {@code SchemaDraftReviewRequested},
 * {@code SchemaDraftDecided}, {@code SchemaDraftPublished},
 * {@code SchemaDraftWithdrawn}. Frontend webhook router маппит их на
 * {@code dict.draft.*} category (см. webhook subscriptions table).
 */
@Service
public class SchemaDraftService {

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

  private final SchemaDraftRepository repository;
  private final DictionaryDefinitionService definitionService;
  private final OutboxRecorder outboxRecorder;
  private final ObjectMapper objectMapper;
  /** Optional — NULL в test ctor'е (без audit_log writes). */
  private final AuditLogger auditLogger;
  /** Optional — NULL в test ctor'е (skip role enforcement в unit tests). */
  private final WorkflowRoles workflowRoles;

  @org.springframework.beans.factory.annotation.Autowired
  public SchemaDraftService(
      SchemaDraftRepository repository,
      DictionaryDefinitionService definitionService,
      OutboxRecorder outboxRecorder,
      ObjectMapper objectMapper,
      AuditLogger auditLogger,
      WorkflowRoles workflowRoles) {
    this.repository = repository;
    this.definitionService = definitionService;
    this.outboxRecorder = outboxRecorder;
    this.objectMapper = objectMapper;
    this.auditLogger = auditLogger;
    this.workflowRoles = workflowRoles;
  }

  /** Test ctor — без audit_log + без role enforcement. */
  public SchemaDraftService(
      SchemaDraftRepository repository,
      DictionaryDefinitionService definitionService,
      OutboxRecorder outboxRecorder,
      ObjectMapper objectMapper) {
    this(repository, definitionService, outboxRecorder, objectMapper, null, null);
  }

  /** Phase 1d guard: enforce realm role если WorkflowRoles bean wired. */
  private void requireRoleIfEnforced(String role) {
    if (workflowRoles != null) {
      workflowRoles.requireRole(role);
    }
  }

  /**
   * Phase 2: dict-level минимальная роль ревьюера (поле
   * {@code DictionaryDefinition.approvalMinRole}). Если задано —
   * требуем чтобы reviewer имел эту роль вдобавок к стандартной
   * {@code SCHEMA_REVIEWER}. Admin super-role проходит автоматически.
   * Зеркало {@code DraftService.requireDictMinRoleIfSet} для записей.
   */
  private void requireDictMinRoleIfSet(DictionaryDefinition def) {
    if (workflowRoles == null) return;
    String minRole = def.getApprovalMinRole();
    if (minRole == null || minRole.isBlank()) return;
    workflowRoles.requireRole(minRole);
  }

  // ─────────────────────────────────────────────────────────────────────
  //  Reads
  // ─────────────────────────────────────────────────────────────────────

  @Transactional(readOnly = true)
  public DictionarySchemaDraft findById(UUID id) {
    return repository.findById(id).orElseThrow(() -> OrdinisException.notFound(
        "schema_draft_not_found",
        "Schema draft " + id + " не найден."));
  }

  @Transactional(readOnly = true)
  public List<DictionarySchemaDraft> listByDictionary(String dictionaryName) {
    DictionaryDefinition def = definitionService.findByName(dictionaryName);
    return repository.findByDictionary(def.getId());
  }

  @Transactional(readOnly = true)
  public java.util.Optional<DictionarySchemaDraft> findActive(String dictionaryName) {
    DictionaryDefinition def = definitionService.findByName(dictionaryName);
    return repository.findActiveByDictionary(def.getId());
  }

  @Transactional(readOnly = true)
  public Page<DictionarySchemaDraft> listPendingReview(Pageable pageable) {
    return repository.findPendingReview(pageable);
  }

  /**
   * Maker self-tracking — все schema drafts которые создал actor, любой
   * статус (DRAFT / REVIEW_PENDING / APPROVED / REJECTED / CHANGES_REQUESTED
   * / WITHDRAWN / PUBLISHED), отсортированы createdAt DESC.
   *
   * <p>Зеркало record-side {@code DraftService.listByMaker} — frontend
   * "Мои черновики" (route {@code /my-drafts}) показывает оба типа,
   * чтобы maker не запоминал где какие drafts лежат.
   */
  @Transactional(readOnly = true)
  public Page<DictionarySchemaDraft> listByMaker(String makerId, Pageable pageable) {
    return repository.findByMaker(makerId, pageable);
  }

  // ─────────────────────────────────────────────────────────────────────
  //  Mutators (state machine transitions)
  // ─────────────────────────────────────────────────────────────────────

  /**
   * Maker создаёт draft от текущего HEAD schema.
   *
   * <ul>
   *   <li>Проверяет {@code fromVersion} == {@code definition.schemaVersion}
   *       (optimistic lock против concurrent publishes).</li>
   *   <li>Если уже есть active draft для этого dict'а → 409
   *       ({@code active_draft_exists}) — EXCLUDE constraint в БД дополнительно
   *       enforce'ит, но мы fail-early до DB hit'а.</li>
   * </ul>
   */
  @Transactional
  public DictionarySchemaDraft create(String dictionaryName, CreateSchemaDraftRequest req) {
    DictionaryDefinition def = definitionService.findByName(dictionaryName);

    // Optimistic version check: maker должен fetch'нуть HEAD прежде чем
    // создавать draft. Если кто-то опубликовал свой draft пока этот maker
    // готовился — он refetch'нёт через 409.
    if (!def.getSchemaVersion().equals(req.fromVersion())) {
      throw OrdinisException.conflict(
          "version_mismatch",
          "fromVersion=" + req.fromVersion() + " не совпадает с current HEAD "
              + def.getSchemaVersion() + ". Refetch и retry.");
    }

    // Early check для better error message (БД constraint всё равно отловит race).
    repository.findActiveByDictionary(def.getId()).ifPresent(existing -> {
      throw OrdinisException.conflict(
          "active_draft_exists",
          "Активный draft " + existing.getId() + " (status=" + existing.getStatus().dbValue()
              + ") уже существует для " + dictionaryName + ". Закройте его перед созданием нового.");
    });

    DictionarySchemaDraft draft = new DictionarySchemaDraft(
        UUID.randomUUID(),
        def.getId(),
        req.proposedSchema(),
        req.fromVersion(),
        currentUserId(),
        req.reason());

    try {
      DictionarySchemaDraft saved = repository.save(draft);
      emitEvent("SchemaDraftCreated", saved, def, Map.of(
          "fromVersion", req.fromVersion()));
      if (auditLogger != null) {
        auditLogger.schemaDraftEvent(
            "SCHEMA_DRAFT_CREATED", "DRAFT_CREATE", def,
            def.getSchemaJson(), saved.getProposedSchema());
      }
      log.info("schema_draft_created id={} dict={} maker={}",
          saved.getId(), dictionaryName, saved.getMakerId());
      return saved;
    } catch (DataIntegrityViolationException e) {
      // Race с EXCLUDE constraint — другой thread параллельно создал draft.
      throw OrdinisException.conflict(
          "active_draft_exists",
          "Активный draft уже существует для " + dictionaryName
              + " (concurrent create).");
    }
  }

  /**
   * Maker submit'ит draft на review. Transitions: DRAFT → REVIEW_PENDING,
   * CHANGES_REQUESTED → REVIEW_PENDING (re-submit после доработки).
   *
   * <p>Только maker может submit'нуть свой draft. Reviewer'ы — opt-in список
   * (NULL = pool, любой approver видит в queue).
   */
  @Transactional
  public DictionarySchemaDraft submitForReview(UUID draftId, SubmitForReviewRequest req) {
    DictionarySchemaDraft draft = findById(draftId);
    assertMakerActor(draft, "submit_review");

    SchemaDraftStatus current = draft.getStatus();
    if (current != SchemaDraftStatus.DRAFT && current != SchemaDraftStatus.CHANGES_REQUESTED) {
      throw OrdinisException.conflict(
          "invalid_transition",
          "Draft " + draftId + " status=" + current.dbValue()
              + " — submit_review возможен только из draft или changes_requested.");
    }

    draft.setStatus(SchemaDraftStatus.REVIEW_PENDING);
    draft.setSubmittedAt(OffsetDateTime.now());
    draft.setReviewNote(req.note());
    draft.setReviewerAssignments(toJsonArray(req.reviewers()));

    DictionarySchemaDraft saved = repository.save(draft);
    DictionaryDefinition def = definitionService.findById(saved.getDictionaryId());
    emitEvent("SchemaDraftReviewRequested", saved, def, Map.of(
        "reviewers", req.reviewers() == null ? List.of() : req.reviewers(),
        "note", req.note() == null ? "" : req.note()));
    if (auditLogger != null) {
      auditLogger.schemaDraftEvent(
          "SCHEMA_DRAFT_REVIEW_REQUESTED", "DRAFT_REVIEW", def,
          def.getSchemaJson(), saved.getProposedSchema());
    }
    log.info("schema_draft_review_pending id={} dict={} reviewers={}",
        saved.getId(), def.getName(), req.reviewers());
    return saved;
  }

  /**
   * Reviewer decision: approve / request_changes / reject.
   *
   * <p>Self-approve запрещён (БД constraint + early check в коде для better
   * error message). Reviewer должен быть != maker.
   */
  @Transactional
  public DictionarySchemaDraft decide(UUID draftId, DecisionRequest req) {
    // Phase 1d: realm role gate. Без 'ordinis:schema:reviewer' actor
    // получит 403 forbidden_role до даже status checks — fail-fast.
    requireRoleIfEnforced(WorkflowRoles.SCHEMA_REVIEWER);
    DictionarySchemaDraft draft = findById(draftId);
    String reviewerId = currentUserId();

    if (draft.getStatus() != SchemaDraftStatus.REVIEW_PENDING) {
      throw OrdinisException.conflict(
          "invalid_transition",
          "Draft " + draftId + " status=" + draft.getStatus().dbValue()
              + " — decision требует review_pending.");
    }
    if (draft.getMakerId().equals(reviewerId)) {
      throw OrdinisException.conflict(
          "self_approve_forbidden",
          "Maker '" + reviewerId + "' не может decide свой draft (maker-checker).");
    }
    // Phase 2: dict-level минимальная роль ревьюера. Resolveим def early,
    // чтобы вернуть 403 forbidden_role до того как меняем status.
    DictionaryDefinition decideDef = definitionService.findById(draft.getDictionaryId());
    requireDictMinRoleIfSet(decideDef);

    SchemaDraftStatus nextStatus = switch (req.decision()) {
      case APPROVE -> SchemaDraftStatus.APPROVED;
      case REQUEST_CHANGES -> SchemaDraftStatus.CHANGES_REQUESTED;
      case REJECT -> SchemaDraftStatus.REJECTED;
    };

    draft.setStatus(nextStatus);
    draft.setDecisionUserId(reviewerId);
    draft.setDecisionAt(OffsetDateTime.now());
    draft.setDecisionComment(req.comment());

    DictionarySchemaDraft saved = repository.save(draft);
    // decideDef уже resolved выше для min-role check — reuse.
    DictionaryDefinition def = decideDef;
    emitEvent("SchemaDraftDecided", saved, def, Map.of(
        "decision", req.decision().name().toLowerCase(),
        "reviewerId", reviewerId,
        "comment", req.comment() == null ? "" : req.comment()));
    if (auditLogger != null) {
      // Action разделён по verdict чтобы timeline UI мог красить approve
      // зелёным, reject красным, changes_requested желтым.
      String action = switch (req.decision()) {
        case APPROVE -> "DRAFT_APPROVE";
        case REQUEST_CHANGES -> "DRAFT_CHANGES_REQUESTED";
        case REJECT -> "DRAFT_REJECT";
      };
      auditLogger.schemaDraftEvent(
          "SCHEMA_DRAFT_DECIDED", action, def,
          def.getSchemaJson(), saved.getProposedSchema());
    }
    log.info("schema_draft_decided id={} dict={} decision={} reviewer={}",
        saved.getId(), def.getName(), req.decision(), reviewerId);
    return saved;
  }

  /**
   * Publish approved draft → live schema. Bumps version (minor by default).
   *
   * <p>Optimistic lock: {@code expectedHeadVersion} проверяется против
   * {@code definition.schemaVersion}. Если другой draft опубликовался
   * первым → 409 ({@code version_mismatch}).
   *
   * <p>SERIALIZABLE isolation — нужна consistency между update'ом
   * definition.schemaJson + draft status'а в одной транзакции.
   */
  @Transactional(isolation = Isolation.SERIALIZABLE)
  public DictionarySchemaDraft publish(UUID draftId, PublishDraftRequest req) {
    // Phase 1d: publish — отдельная role от reviewer'а. Publishers
    // обычно более узкая группа (production gatekeepers) чем reviewers.
    requireRoleIfEnforced(WorkflowRoles.SCHEMA_PUBLISHER);
    DictionarySchemaDraft draft = findById(draftId);
    DictionaryDefinition def = definitionService.findById(draft.getDictionaryId());
    // Phase 2: dict-level минимальная роль (e.g. "ordinis:restricted"). Publish
    // — самое чувствительное действие (меняет live schema), требуем тот же
    // gate что у decide. Если поле пусто — no-op.
    requireDictMinRoleIfSet(def);

    if (draft.getStatus() != SchemaDraftStatus.APPROVED) {
      throw OrdinisException.conflict(
          "invalid_transition",
          "Draft " + draftId + " status=" + draft.getStatus().dbValue()
              + " — publish требует approved.");
    }
    if (!def.getSchemaVersion().equals(req.expectedHeadVersion())) {
      throw OrdinisException.conflict(
          "version_mismatch",
          "expectedHeadVersion=" + req.expectedHeadVersion()
              + " не совпадает с current HEAD " + def.getSchemaVersion()
              + ". Кто-то другой опубликовал свой draft — refetch и retry.");
    }

    String newVersion = computeNextVersion(def.getSchemaVersion(), def.getSchemaJson(),
        draft.getProposedSchema());

    // Apply schema через repository напрямую — мы избегаем
    // DictionaryDefinitionService.updateSchema потому что он принимает
    // CreateDictionaryRequest и заново валидирует supportedLocales etc;
    // здесь нам нужно только set'нуть schemaJson + version + emit
    // definition-level outbox через тот же recorder для согласованности.
    def.setSchemaJson(draft.getProposedSchema());
    def.setSchemaVersion(newVersion);
    // updated_at установится AuditingEntityListener'ом.

    OffsetDateTime publishedAt = OffsetDateTime.now();
    draft.setStatus(SchemaDraftStatus.PUBLISHED);
    draft.setPublishedAt(publishedAt);
    draft.setPublishedVersion(newVersion);
    draft.setPublishNote(req.publishNote());

    DictionarySchemaDraft saved = repository.save(draft);
    emitEvent("SchemaDraftPublished", saved, def, Map.of(
        "newVersion", newVersion,
        "publishedAt", publishedAt.toString(),
        "publishNote", req.publishNote() == null ? "" : req.publishNote()));
    if (auditLogger != null) {
      // Schema-level write идёт через draft path — DEFINITION_UPDATED тоже
      // имеет смысл (мы только что обновили def.schemaJson + version), но
      // чтобы не дублировать запись пишем единственное SCHEMA_DRAFT_PUBLISHED
      // — оно self-describes как version-bump (before/after schemas есть).
      // ChangelogService рендерит как "v1.5.0 (опубликован из draft'а)".
      auditLogger.schemaDraftEvent(
          "SCHEMA_DRAFT_PUBLISHED", "DRAFT_PUBLISH", def,
          draft.getBranchedFromVersion() == null ? null : def.getSchemaJson(),
          draft.getProposedSchema());
    }
    log.info("schema_draft_published id={} dict={} newVersion={}",
        saved.getId(), def.getName(), newVersion);
    return saved;
  }

  /**
   * Maker правит proposedSchema или reason в существующем draft'е без
   * recreate. Полезно после CHANGES_REQUESTED: вместо withdraw + create
   * (с потерей draftId, history события, reviewer assignment) — просто
   * обновляем content.
   *
   * <p>Status-gate: только {@code DRAFT} и {@code CHANGES_REQUESTED}.
   * Из {@code REVIEW_PENDING} update запрещён — pending review должен
   * сначала разрешиться decision'ом (reviewer не должен ловить moving
   * target). Из {@code APPROVED} тоже запрещён — после approve content
   * заморожен до publish (иначе одобрили одну схему, опубликовали другую).
   *
   * <p>Side effect: если status был {@code CHANGES_REQUESTED}, переводим
   * обратно в {@code DRAFT} чтобы maker должен был осознанно submit'нуть
   * на ревью заново. Это match'ит логичную ментальную модель «правки →
   * новый draft state → re-submit».
   *
   * <p>Audit + outbox event {@code SchemaDraftUpdated} даёт changelog
   * timeline видимость "maker правил черновик после review feedback".
   */
  @Transactional
  public DictionarySchemaDraft updateContent(UUID draftId, UpdateSchemaDraftRequest req) {
    DictionarySchemaDraft draft = findById(draftId);
    assertMakerActor(draft, "update_content");

    SchemaDraftStatus current = draft.getStatus();
    if (current != SchemaDraftStatus.DRAFT && current != SchemaDraftStatus.CHANGES_REQUESTED) {
      throw OrdinisException.conflict(
          "invalid_transition",
          "Draft " + draftId + " status=" + current.dbValue()
              + " — update возможен только из draft или changes_requested.");
    }

    JsonNode beforeSchema = draft.getProposedSchema();
    if (req.proposedSchema() != null) {
      draft.setProposedSchema(req.proposedSchema());
    }
    if (req.reason() != null) {
      draft.setReason(req.reason());
    }
    // CHANGES_REQUESTED → DRAFT: maker должен заново submit'нуть после правок.
    if (current == SchemaDraftStatus.CHANGES_REQUESTED) {
      draft.setStatus(SchemaDraftStatus.DRAFT);
    }

    DictionarySchemaDraft saved = repository.save(draft);
    DictionaryDefinition def = definitionService.findById(saved.getDictionaryId());
    emitEvent("SchemaDraftUpdated", saved, def, Map.of(
        "previousStatus", current.dbValue()));
    if (auditLogger != null) {
      auditLogger.schemaDraftEvent(
          "SCHEMA_DRAFT_UPDATED", "DRAFT_UPDATE", def,
          beforeSchema, saved.getProposedSchema());
    }
    log.info("schema_draft_updated id={} dict={} maker={} prevStatus={}",
        saved.getId(), def.getName(), saved.getMakerId(), current.dbValue());
    return saved;
  }

  /** Maker отзывает свой draft (только non-terminal статусы). */
  @Transactional
  public DictionarySchemaDraft withdraw(UUID draftId) {
    DictionarySchemaDraft draft = findById(draftId);
    assertMakerActor(draft, "withdraw");
    if (draft.getStatus().isTerminal()) {
      throw OrdinisException.conflict(
          "invalid_transition",
          "Draft " + draftId + " уже в terminal status'е " + draft.getStatus().dbValue()
              + " — withdraw невозможен.");
    }

    draft.setStatus(SchemaDraftStatus.WITHDRAWN);
    DictionarySchemaDraft saved = repository.save(draft);
    DictionaryDefinition def = definitionService.findById(saved.getDictionaryId());
    emitEvent("SchemaDraftWithdrawn", saved, def, Map.of());
    if (auditLogger != null) {
      auditLogger.schemaDraftEvent(
          "SCHEMA_DRAFT_WITHDRAWN", "DRAFT_WITHDRAW", def,
          def.getSchemaJson(), saved.getProposedSchema());
    }
    log.info("schema_draft_withdrawn id={} dict={} maker={}",
        saved.getId(), def.getName(), saved.getMakerId());
    return saved;
  }

  // ─────────────────────────────────────────────────────────────────────
  //  Helpers
  // ─────────────────────────────────────────────────────────────────────

  private void assertMakerActor(DictionarySchemaDraft draft, String op) {
    String actor = currentUserId();
    if (!draft.getMakerId().equals(actor)) {
      throw OrdinisException.forbidden(
          "maker_only",
          "Operation '" + op + "' возможна только для maker'а draft'а ("
              + draft.getMakerId() + "), actor=" + actor + ".");
    }
  }

  /**
   * Compute next schema version. Default: minor bump (1.4.0 → 1.5.0).
   * Future improvement (Phase 1): detect breaking keys в diff (type changes,
   * required removals, x-references changes) → major bump.
   *
   * <p>Сейчас простая эвристика — minor unless schema diff содержит явные
   * breaking keys в top-level properties. Полная семантика — будущее.
   */
  String computeNextVersion(String currentVersion, JsonNode oldSchema, JsonNode newSchema) {
    String[] parts = currentVersion.split("\\.");
    if (parts.length != 3) {
      // Fallback на patch bump если version не semver — лучше чем падать.
      return currentVersion + ".1";
    }
    boolean breaking = detectBreakingChange(oldSchema, newSchema);
    int major = Integer.parseInt(parts[0]);
    int minor = Integer.parseInt(parts[1]);
    if (breaking) {
      return (major + 1) + ".0.0";
    }
    return major + "." + (minor + 1) + ".0";
  }

  /**
   * Heuristic breaking detection: required property removed OR
   * existing property type changed. Conservative — false negatives OK
   * (publisher может всегда override через explicit version в future).
   */
  private boolean detectBreakingChange(JsonNode oldSchema, JsonNode newSchema) {
    if (oldSchema == null || newSchema == null) return false;

    // Removed required property?
    JsonNode oldRequired = oldSchema.path("required");
    JsonNode newRequired = newSchema.path("required");
    if (oldRequired.isArray()) {
      for (JsonNode item : oldRequired) {
        String k = item.asText();
        boolean stillRequired = false;
        if (newRequired.isArray()) {
          for (JsonNode n : newRequired) {
            if (n.asText().equals(k)) {
              stillRequired = true;
              break;
            }
          }
        }
        if (!stillRequired) return true;
      }
    }

    // Changed property type?
    JsonNode oldProps = oldSchema.path("properties");
    JsonNode newProps = newSchema.path("properties");
    if (oldProps.isObject() && newProps.isObject()) {
      var fields = oldProps.fieldNames();
      while (fields.hasNext()) {
        String fname = fields.next();
        JsonNode oldType = oldProps.path(fname).path("type");
        JsonNode newType = newProps.path(fname).path("type");
        if (!newType.isMissingNode() && !oldType.isMissingNode()
            && !oldType.asText().equals(newType.asText())) {
          return true;
        }
      }
    }
    return false;
  }

  private JsonNode toJsonArray(List<String> items) {
    if (items == null || items.isEmpty()) return null;
    ArrayNode arr = objectMapper.createArrayNode();
    items.forEach(arr::add);
    return arr;
  }

  private void emitEvent(
      String eventType,
      DictionarySchemaDraft draft,
      DictionaryDefinition def,
      Map<String, Object> extras) {
    Map<String, Object> payload = new LinkedHashMap<>();
    payload.put("draftId", draft.getId().toString());
    payload.put("dictionaryId", def.getId().toString());
    payload.put("dictionaryName", def.getName());
    payload.put("status", draft.getStatus().dbValue());
    payload.put("branchedFromVersion", draft.getBranchedFromVersion());
    payload.put("makerId", draft.getMakerId());
    payload.putAll(extras);

    var envelope = EventEnvelope.of(
        def.getName(),
        toEventScope(def.getScope()),
        def.getBundle(),
        payload);

    outboxRecorder.record(
        eventType,
        "DictionarySchemaDraft",
        draft.getId().toString(),
        def.getName(),
        def.getScope(),
        def.getBundle(),
        "schema-draft:" + draft.getId(),
        envelope);
  }

  private static cloud.nstart.terravault.ordinis.events.DataScope toEventScope(
      cloud.nstart.terravault.ordinis.domain.DataScope domain) {
    return cloud.nstart.terravault.ordinis.events.DataScope.valueOf(domain.name());
  }

  /** Resolve current user from Spring Security context (Keycloak subject). */
  static String currentUserId() {
    var auth = org.springframework.security.core.context.SecurityContextHolder
        .getContext().getAuthentication();
    if (auth == null || !auth.isAuthenticated()) return "anonymous";
    return auth.getName();
  }
}
