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

import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Tags;
import jakarta.annotation.PostConstruct;
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.dao.DataAccessException;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.time.Instant;
import java.time.OffsetDateTime;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;

/* JDK 25 + Mockito не мочат concrete JdbcTemplate (subclass mock-maker
 * disabled). Обходим через MvGateway — узкий интерфейс над JdbcTemplate,
 * который тесты mock'ают. Производственная impl (см. nested class) делает
 * прямые JdbcTemplate calls. */

/**
 * Throttled refresh для {@code record_dependents_index} materialized view
 * (Phase 2 dict-relationships-v2).
 *
 * <p>Активируется когда {@code ordinis.lineage.mv.enabled=true}. Без MV
 * сервис не нужен — {@link DependentsQueryJdbc} даёт прямой JSONB query
 * без stale window'а.
 *
 * <p>Пути refresh'а:
 * <ul>
 *   <li><b>Scheduled</b> — каждые
 *       {@code ordinis.lineage.mv.refresh-interval-sec} (default 60).
 *       Простейший throttle: если refresh уже запущен — пропускаем тик.</li>
 *   <li><b>Force</b> — {@link #refreshNow()} вызывается из bundle import
 *       или admin endpoint. Также throttled: refused если последний refresh
 *       был меньше {@code min-interval-sec} назад.</li>
 * </ul>
 *
 * <p>REFRESH CONCURRENTLY — read concurrent с queries, требует UNIQUE INDEX
 * на MV (создан в migration 0016). Если MV пуста (initial state до первого
 * REFRESH) — CONCURRENTLY падает; fallback на non-CONCURRENTLY один раз.
 *
 * <p>Storm prevention (R2 eng review): even при 100 closes/sec, refresh
 * может запускаться не чаще раза в {@code refresh-interval-sec}. Closing
 * record делает MV stale — UI badge показывает "обновлено N min назад";
 * actual valid_to update в dictionary_records появится синхронно (через
 * record close transaction), MV догонит на next tick.
 */
@Component
@ConditionalOnProperty(
    name = "ordinis.lineage.mv.enabled",
    havingValue = "true",
    matchIfMissing = false)
public class MaterializedViewRefreshService {

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

  private final MvGateway gateway;
  /** Volatile guard — не позволяет двум refresh'ам параллельно. */
  private final AtomicReference<Instant> currentRefreshStart = new AtomicReference<>(null);
  /** Phase 4: gauges feed Prometheus + alerts. */
  private final AtomicLong lastRefreshEpochMs = new AtomicLong(0);
  private final AtomicLong lastRowCount = new AtomicLong(0);
  private final AtomicLong lastDurationMs = new AtomicLong(0);
  private final Optional<MeterRegistry> meterRegistry;

  @Value("${ordinis.lineage.mv.min-interval-sec:30}")
  private long minIntervalSec;

  public MaterializedViewRefreshService(JdbcTemplate jdbc, Optional<MeterRegistry> meterRegistry) {
    this(new JdbcMvGateway(jdbc), meterRegistry);
  }

  public MaterializedViewRefreshService(JdbcTemplate jdbc) {
    this(new JdbcMvGateway(jdbc), Optional.empty());
  }

  /** Test-only ctor — gateway инжектится напрямую. */
  MaterializedViewRefreshService(MvGateway gateway) {
    this(gateway, Optional.empty());
  }

  MaterializedViewRefreshService(MvGateway gateway, Optional<MeterRegistry> meterRegistry) {
    this.gateway = gateway;
    this.meterRegistry = meterRegistry;
  }

  @PostConstruct
  void registerGauges() {
    if (meterRegistry.isEmpty()) return;
    var reg = meterRegistry.get();
    // Last refresh age в секундах — используется в alert OrdinisLineageMvStale.
    reg.gauge("nsi_lineage_mv_last_refresh_seconds_ago", Tags.empty(), this,
        s -> s.lastRefreshEpochMs.get() == 0
            ? Double.NaN
            : (System.currentTimeMillis() - s.lastRefreshEpochMs.get()) / 1000.0);
    reg.gauge("nsi_lineage_mv_row_count", Tags.empty(), lastRowCount, AtomicLong::doubleValue);
    reg.gauge("nsi_lineage_mv_last_duration_seconds", Tags.empty(),
        lastDurationMs, l -> l.get() / 1000.0);
  }

  /**
   * Periodic tick. Проверяет cooldown + наличие work, делает refresh если
   * можно. fixedDelayString — следующий запуск считается ОТ конца предыдущего
   * (а не от старта), что предотвращает overlap при медленном refresh'е.
   */
  @Scheduled(
      initialDelayString = "${ordinis.lineage.mv.initial-delay-sec:30}000",
      fixedDelayString = "${ordinis.lineage.mv.refresh-interval-sec:60}000")
  public void scheduledRefresh() {
    refreshIfDue();
  }

  /**
   * Внешний trigger (bundle import, admin endpoint). Сам решает — можно
   * ли refresh, выкидывает {@link RefreshThrottledException} если cooldown
   * не прошёл. Caller отображает: "Refresh throttled, retry в N sec".
   */
  public void refreshNow() {
    if (!refreshIfDue()) {
      throw new RefreshThrottledException(
          "MV refresh throttled: last refresh < " + minIntervalSec + "s ago");
    }
  }

  /**
   * @return last refresh metadata из {@code lineage_mv_meta} table.
   *         Empty если migration ещё не накатилась или таблица недоступна.
   */
  public Optional<RefreshMeta> lastRefresh() {
    try {
      return Optional.ofNullable(gateway.loadMeta());
    } catch (DataAccessException e) {
      log.warn("Failed to read lineage_mv_meta: {}", e.getMessage());
      return Optional.empty();
    }
  }

  // === Internals ===

  private boolean refreshIfDue() {
    Instant now = Instant.now();
    Optional<RefreshMeta> last = lastRefresh();
    if (last.isPresent() && last.get().lastRefreshedAt() != null) {
      Duration since = Duration.between(last.get().lastRefreshedAt().toInstant(), now);
      if (since.toSeconds() < minIntervalSec) {
        log.debug("Skip MV refresh: last was {} sec ago (min interval {})",
            since.toSeconds(), minIntervalSec);
        return false;
      }
    }
    if (!currentRefreshStart.compareAndSet(null, now)) {
      log.debug("Skip MV refresh: previous tick still running since {}",
          currentRefreshStart.get());
      return false;
    }
    try {
      doRefresh();
      return true;
    } finally {
      currentRefreshStart.set(null);
    }
  }

  private void doRefresh() {
    long start = System.currentTimeMillis();
    String status = "ok";
    String error = null;
    try {
      // Try CONCURRENTLY first; fallback на non-CONCURRENTLY если MV пустой
      // (initial state — REFRESH CONCURRENTLY требует prior populate).
      try {
        gateway.refreshConcurrently();
      } catch (DataAccessException concurrentFail) {
        log.warn("REFRESH CONCURRENTLY failed, trying plain REFRESH: {}",
            concurrentFail.getMessage());
        gateway.refresh();
      }
    } catch (DataAccessException e) {
      status = "error";
      error = truncate(e.getMessage(), 1000);
      log.error("MV refresh failed: {}", e.getMessage(), e);
    }
    long duration = System.currentTimeMillis() - start;
    Long rowCount = null;
    try {
      rowCount = gateway.countRows();
    } catch (DataAccessException e) {
      log.warn("Failed to count rows after refresh: {}", e.getMessage());
    }
    try {
      gateway.updateMeta(duration, rowCount, status, error);
    } catch (DataAccessException e) {
      log.warn("Failed to update lineage_mv_meta: {}", e.getMessage());
    }
    // Phase 4: feed gauges + counter (separate from gateway updateMeta —
    // нужно даже при ошибке записи в lineage_mv_meta).
    lastDurationMs.set(duration);
    if (rowCount != null) lastRowCount.set(rowCount);
    if ("ok".equals(status)) {
      lastRefreshEpochMs.set(System.currentTimeMillis());
      log.info("MV record_dependents_index refreshed: {}ms, {} rows", duration, rowCount);
    }
    final String finalStatus = status;
    meterRegistry.ifPresent(reg -> reg.counter(
        "nsi_lineage_mv_refresh_total",
        Tags.of("outcome", finalStatus)).increment());
  }

  private static String truncate(String s, int max) {
    if (s == null) return null;
    return s.length() <= max ? s : s.substring(0, max);
  }

  /** Last refresh snapshot. */
  public record RefreshMeta(
      OffsetDateTime lastRefreshedAt,
      Long lastDurationMs,
      Long lastRowCount,
      String lastStatus,
      String lastError) {}

  /** Throw'ится из {@link #refreshNow()} если cooldown не прошёл. */
  public static class RefreshThrottledException extends RuntimeException {
    public RefreshThrottledException(String message) {
      super(message);
    }
  }

  /**
   * Узкий gateway над JdbcTemplate — позволяет тестам mock'нуть вместо
   * concrete JdbcTemplate (Mockito blocked на JDK 25).
   */
  interface MvGateway {
    void refreshConcurrently();

    void refresh();

    Long countRows();

    RefreshMeta loadMeta();

    void updateMeta(long durationMs, Long rowCount, String status, String error);
  }

  /** Production impl над JdbcTemplate. */
  static final class JdbcMvGateway implements MvGateway {
    private final JdbcTemplate jdbc;

    JdbcMvGateway(JdbcTemplate jdbc) {
      this.jdbc = jdbc;
    }

    @Override
    public void refreshConcurrently() {
      jdbc.execute("REFRESH MATERIALIZED VIEW CONCURRENTLY record_dependents_index");
    }

    @Override
    public void refresh() {
      jdbc.execute("REFRESH MATERIALIZED VIEW record_dependents_index");
    }

    @Override
    public Long countRows() {
      return jdbc.queryForObject(
          "SELECT COUNT(*) FROM record_dependents_index", Long.class);
    }

    @Override
    public RefreshMeta loadMeta() {
      return jdbc.query(
          "SELECT last_refreshed_at, last_duration_ms, last_row_count, last_status, last_error "
              + "FROM lineage_mv_meta WHERE id = 1",
          rs -> {
            if (!rs.next()) return null;
            OffsetDateTime ts = rs.getObject("last_refreshed_at", OffsetDateTime.class);
            return new RefreshMeta(
                ts,
                rs.getObject("last_duration_ms", Long.class),
                rs.getObject("last_row_count", Long.class),
                rs.getString("last_status"),
                rs.getString("last_error"));
          });
    }

    @Override
    public void updateMeta(long durationMs, Long rowCount, String status, String error) {
      jdbc.update(
          "UPDATE lineage_mv_meta SET last_refreshed_at = NOW(), last_duration_ms = ?, "
              + "last_row_count = ?, last_status = ?, last_error = ? WHERE id = 1",
          durationMs, rowCount, status, error);
    }
  }
}
