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

import cloud.nstart.terravault.ordinis.auth.ScopeContext;
import cloud.nstart.terravault.ordinis.domain.DataScope;
import cloud.nstart.terravault.ordinis.domain.definition.DictionaryDefinition;
import cloud.nstart.terravault.ordinis.domain.record.DictionaryRecord;
import cloud.nstart.terravault.ordinis.domain.record.DictionaryRecordRepository;
import cloud.nstart.terravault.ordinis.events.EventEnvelope;
import cloud.nstart.terravault.ordinis.events.record.DictionaryRecordDeleted;
import cloud.nstart.terravault.ordinis.outbox.OutboxRecorder;
import cloud.nstart.terravault.ordinis.restapi.audit.AuditLogger;
import cloud.nstart.terravault.ordinis.restapi.error.OrdinisException;
import cloud.nstart.terravault.ordinis.restapi.service.DictionaryDefinitionService;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Tags;
import io.micrometer.core.instrument.Timer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.QueryTimeoutException;
import org.springframework.stereotype.Service;
import org.springframework.transaction.TransactionTimedOutException;
import org.springframework.transaction.annotation.Isolation;
import org.springframework.transaction.annotation.Transactional;

import java.time.OffsetDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.Set;

/**
 * Cascade close engine — Phase 3 dict-relationships-v2 epic.
 *
 * <p>Запускается когда close target record имеет dependents (записи в других
 * словарях через {@code x-references}). Branching по {@link OnCloseAction}
 * объявленному в schema referencing dict'а:
 *
 * <ul>
 *   <li>{@link OnCloseAction#BLOCK} — close отклоняется с 409
 *       {@code x_references_blocked_by_dependents}, dependents list в response.</li>
 *   <li>{@link OnCloseAction#WARN} — close проходит, warnings возвращаются
 *       в response (orphan scanner подхватит на metric).</li>
 *   <li>{@link OnCloseAction#CASCADE} — требует explicit confirm через
 *       cascade-close endpoint. Direct dependents закрываются в той же
 *       transaction (one hop, F.OUT по design).</li>
 * </ul>
 *
 * <p>Concurrency: SERIALIZABLE isolation. Postgres exclusion constraint +
 * SERIALIZABLE retry — second concurrent close видит conflict и retry'ится
 * (или возвращает 409 если retry exhausted).
 *
 * <p>Tx timeout: {@code @Transactional(timeout = 30)} на {@link #executeCascade}.
 * При timeout'е → 503 {@code x_references_cascade_timeout}, rollback clean.
 */
@Service
public class CascadeCloseService {

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

  /** Hard upper bound на cascade size (one hop). Свыше — bulk-close ручкой. */
  private static final int CASCADE_MAX_PER_REQUEST = 500;

  private final LineageIndexService lineage;
  private final RecordCloseSink closeSink;
  private final ScopeContext scopeContext;
  /** Used to enforce the approvalRequired gate on the target + every cascaded
   *  dependent dict. Optional: null in legacy test ctors that don't exercise
   *  the gate. Production wiring always provides it. */
  private final DictionaryDefinitionService defService;
  /** Phase 4 SLO. Optional — null в test ctor'е. */
  private final Optional<MeterRegistry> meterRegistry;

  @Autowired
  public CascadeCloseService(
      LineageIndexService lineage,
      DictionaryRecordRepository recordRepo,
      DictionaryDefinitionService defService,
      AuditLogger auditLogger,
      OutboxRecorder outbox,
      ScopeContext scopeContext,
      Optional<MeterRegistry> meterRegistry) {
    this(lineage,
        new DefaultRecordCloseSink(recordRepo, defService, auditLogger, outbox),
        scopeContext,
        defService,
        meterRegistry);
  }

  /** Test-only ctor — sink инжектится напрямую (Mockito не мокает concrete). */
  CascadeCloseService(
      LineageIndexService lineage,
      RecordCloseSink closeSink,
      ScopeContext scopeContext) {
    this(lineage, closeSink, scopeContext, null, Optional.empty());
  }

  CascadeCloseService(
      LineageIndexService lineage,
      RecordCloseSink closeSink,
      ScopeContext scopeContext,
      DictionaryDefinitionService defService,
      Optional<MeterRegistry> meterRegistry) {
    this.lineage = lineage;
    this.closeSink = closeSink;
    this.scopeContext = scopeContext;
    this.defService = defService;
    this.meterRegistry = meterRegistry;
  }

  /**
   * Read-only план: какие dependents есть, сгруппированные по mode.
   * UI вызывает перед close чтобы показать preview dialog.
   */
  @Transactional(readOnly = true)
  public CascadePlan evaluatePlan(
      String targetDict, String targetKey, OffsetDateTime at) {
    Set<DataScope> scopes = scopeContext.currentScopes();
    OffsetDateTime when = at == null ? OffsetDateTime.now() : at;
    var page = lineage.findRecordDependents(
        targetDict, targetKey, scopes, 0, CASCADE_MAX_PER_REQUEST);

    List<CascadeEntry> blockers = new ArrayList<>();
    List<CascadeEntry> warnings = new ArrayList<>();
    List<CascadeEntry> cascade = new ArrayList<>();
    for (var d : page.items()) {
      var entry = new CascadeEntry(
          d.sourceDict(), d.sourceDisplayName(), d.businessKey(),
          d.sourceField(), d.onClose(), d.id());
      switch (d.onClose()) {
        case BLOCK -> blockers.add(entry);
        case WARN -> warnings.add(entry);
        case CASCADE -> cascade.add(entry);
      }
    }
    return new CascadePlan(
        targetDict, targetKey, blockers, warnings, cascade, page.total());
  }

  /**
   * Atomic execute: проверяет план, кидает 409 если blockers есть,
   * иначе закрывает target + all CASCADE dependents в одной transaction.
   * WARN dependents остаются orphan (warning'и возвращаются в response).
   *
   * @param requireConfirm true → если в плане CASCADE есть, требует explicit confirm.
   *                       false → запускает только если blockers + cascade пусты
   *                       (used by existing DELETE endpoint default path).
   */
  @Transactional(isolation = Isolation.SERIALIZABLE, timeout = 30)
  public CascadeCloseResult executeCascade(
      String targetDict,
      String targetKey,
      OffsetDateTime closeAt,
      String reason,
      boolean confirmed) {
    Timer.Sample sample = startTimer();
    OffsetDateTime when = closeAt == null ? OffsetDateTime.now() : closeAt;
    CascadePlan plan = evaluatePlan(targetDict, targetKey, when);

    // Approval gate: cascade-close goes through closeSink which writes
    // straight to the repo + outbox + audit, completely skipping
    // DictionaryRecordService.close → requireApprovalGate. So we have to
    // enforce here, against BOTH the target dict and every cascaded
    // dependent. If any of them is approval-required, refuse the whole
    // cascade — it cannot atomically transition to drafts mid-cascade,
    // and partial cascade would leave dangling references.
    // (Future work: auto-create one parent draft + N dependent drafts in
    // a single submit-for-review bundle. Out of scope for the gap fix.)
    if (defService != null) {
      assertNotApprovalRequired(targetDict);
      for (var dep : plan.cascade()) {
        assertNotApprovalRequired(dep.sourceDict());
      }
    }

    if (!plan.blockers().isEmpty()) {
      stopTimerAndCount(sample, targetDict, "blocked");
      throw OrdinisException.conflict(
          "x_references_blocked_by_dependents",
          "Cannot close " + targetDict + "/" + targetKey + " — "
              + plan.blockers().size() + " active dependents block close. "
              + "Resolve them first, or use cascade-close endpoint with confirmation.");
    }

    if (!plan.cascade().isEmpty() && !confirmed) {
      stopTimerAndCount(sample, targetDict, "cascade_required");
      throw OrdinisException.conflict(
          "x_references_cascade_required",
          plan.cascade().size() + " dependents will be cascaded. "
              + "Re-call with confirmed=true to proceed.");
    }

    if (plan.cascade().size() > CASCADE_MAX_PER_REQUEST) {
      stopTimerAndCount(sample, targetDict, "too_large");
      throw OrdinisException.badRequest(
          "x_references_cascade_too_large",
          "Cascade size " + plan.cascade().size() + " exceeds limit "
              + CASCADE_MAX_PER_REQUEST + ". Split via bulk-close.");
    }

    try {
      // Close target first (FK semantics не enforce'ятся в Postgres для
      // JSON refs, но логика "сначала parent, потом children" чище).
      closeSink.closeRecord(targetDict, targetKey, when, reason);
      String cascadeReason = "cascaded from " + targetDict + "/" + targetKey;
      for (var dep : plan.cascade()) {
        closeSink.closeRecord(dep.sourceDict(), dep.businessKey(), when, cascadeReason);
      }
      log.info("Cascade close OK: target={}/{} cascaded={} warnings={}",
          targetDict, targetKey, plan.cascade().size(), plan.warnings().size());
      stopTimerAndCount(sample, targetDict, "success");
      return new CascadeCloseResult(
          targetDict, targetKey, plan.cascade(), plan.warnings(), 1 + plan.cascade().size());
    } catch (TransactionTimedOutException | QueryTimeoutException e) {
      log.error("Cascade close timeout for {}/{} after 30s: {}", targetDict, targetKey, e.getMessage());
      stopTimerAndCount(sample, targetDict, "timeout");
      throw new OrdinisException(
          org.springframework.http.HttpStatus.SERVICE_UNAVAILABLE,
          "x_references_cascade_timeout",
          "Cascade close transaction exceeded 30s timeout. Rollback complete. "
              + "Consider splitting via bulk-close or reducing cascade fan-out.");
    } catch (RuntimeException e) {
      stopTimerAndCount(sample, targetDict, "error");
      throw e;
    }
  }

  /**
   * Block cascade-close from bypassing the maker-checker draft workflow on
   * approval-required dictionaries. Throws 422 {@code approval_required}
   * pointing the caller at the drafts API.
   */
  private void assertNotApprovalRequired(String dict) {
    DictionaryDefinition d;
    try {
      d = defService.findByName(dict);
    } catch (RuntimeException e) {
      // If we can't resolve the dict we definitely shouldn't proceed.
      throw OrdinisException.badRequest(
          "dictionary_not_found",
          "Cascade close: cannot resolve dictionary " + dict);
    }
    if (d.isApprovalRequired()) {
      throw OrdinisException.unprocessableEntity(
          "approval_required",
          "Cascade close cannot proceed: dictionary " + dict
              + " requires the draft workflow. Open a record draft via "
              + "POST /api/v1/dictionaries/" + dict + "/records/{businessKey}/drafts "
              + "with operation=CLOSE for each affected record.");
    }
  }

  // === Metrics helpers ===

  private Timer.Sample startTimer() {
    if (meterRegistry.isEmpty()) return null;
    return Timer.start(meterRegistry.get());
  }

  /** Records timer + increments outcome counter (success/blocked/timeout/...). */
  private void stopTimerAndCount(Timer.Sample sample, String targetDict, String outcome) {
    if (meterRegistry.isEmpty()) return;
    var registry = meterRegistry.get();
    Tags tags = Tags.of("target_dict", targetDict, "outcome", outcome);
    if (sample != null) {
      Timer t = Timer.builder("nsi_cascade_close_duration_seconds")
          .description("Cascade close transaction latency")
          .tags(tags)
          .publishPercentiles(0.5, 0.95, 0.99)
          .register(registry);
      sample.stop(t);
    }
    registry.counter("nsi_cascade_close_total", tags).increment();
  }

  /**
   * Узкий test-friendly seam над close mechanics (repo + audit + outbox).
   * Concrete prod impl ниже; тесты подменяют через subclass / lambda.
   */
  interface RecordCloseSink {
    void closeRecord(String dictionaryName, String businessKey,
        OffsetDateTime when, String reason);
  }

  /** Production impl: repo save + audit + outbox в одной operation. */
  static final class DefaultRecordCloseSink implements RecordCloseSink {
    private final DictionaryRecordRepository recordRepo;
    private final DictionaryDefinitionService defService;
    private final AuditLogger auditLogger;
    private final OutboxRecorder outbox;

    DefaultRecordCloseSink(
        DictionaryRecordRepository recordRepo,
        DictionaryDefinitionService defService,
        AuditLogger auditLogger,
        OutboxRecorder outbox) {
      this.recordRepo = recordRepo;
      this.defService = defService;
      this.auditLogger = auditLogger;
      this.outbox = outbox;
    }

    @Override
    public void closeRecord(
        String dictionaryName, String businessKey, OffsetDateTime when, String reason) {
      DictionaryDefinition d = defService.findByName(dictionaryName);
      DictionaryRecord current = recordRepo.findActiveAt(d.getId(), businessKey, when)
          .orElseThrow(() -> OrdinisException.notFound(
              "record_not_active",
              "Нет активной записи business_key=" + businessKey + " в " + dictionaryName));

      current.setValidTo(when);
      recordRepo.save(current);
      auditLogger.recordClose(d, current, reason);
      publishDeleted(d, current, when, reason);
    }

    private void publishDeleted(
        DictionaryDefinition d, DictionaryRecord r, OffsetDateTime closedAt, String reason) {
      var payload = new DictionaryRecordDeleted(
          r.getId(), r.getBusinessKey(), closedAt, OffsetDateTime.now(),
          r.getUpdatedBy(), reason);
      var envelope = EventEnvelope.of(
          d.getName(),
          cloud.nstart.terravault.ordinis.events.DataScope.valueOf(d.getScope().name()),
          d.getBundle(), payload);
      outbox.record(
          "RecordDeleted", "DictionaryRecord", r.getId().toString(),
          d.getName(), d.getScope(), d.getBundle(),
          d.getName() + ":" + r.getBusinessKey(), envelope);
    }
  }

  // === DTOs ===

  /** Single cascade entry: одна dependent record + её mode. */
  public record CascadeEntry(
      String sourceDict,
      String sourceDisplayName,
      String businessKey,
      String sourceField,
      OnCloseAction onClose,
      java.util.UUID recordId) {}

  /** Read-only план: что произойдёт при close. */
  public record CascadePlan(
      String targetDict,
      String targetKey,
      List<CascadeEntry> blockers,
      List<CascadeEntry> warnings,
      List<CascadeEntry> cascade,
      long totalDependents) {}

  /** Результат execute: сколько записей закрылось + warnings. */
  public record CascadeCloseResult(
      String targetDict,
      String targetKey,
      List<CascadeEntry> cascadedRecords,
      List<CascadeEntry> warnings,
      int totalClosed) {}
}
