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;
  private final Counter recordsSkippedCounter;

  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);
    this.recordsSkippedCounter = Counter.builder("ordinis_projection_records_skipped_total")
        .description("Skipped (dict has redis_projection_enabled=false). CEO E2 tiered perf.")
        .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();

    // CEO plan E2 (Tiered perf): per-dict opt-in. Default false → projection
    // не материализуется в Redis (read-api идёт в PG read replica).
    // Admin переключает true когда dict упирается в RPS limit на PG.
    if (!def.isRedisProjectionEnabled()) {
      recordsSkippedCounter.increment();
      log.trace("Projection upsert SKIPPED ({} has redis_projection_enabled=false): {}",
          dictionary, businessKey);
      return;
    }

    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);
  }

  @Transactional(readOnly = true)
  public void delete(String bundle, String dictionary, String businessKey) {
    Optional<DictionaryDefinition> defOpt = definitionRepository.findByName(dictionary);
    if (defOpt.isPresent() && !defOpt.get().isRedisProjectionEnabled()) {
      // Symmetric с upsert: nothing to delete если проекция выключена.
      // Защита от случая когда раньше flag был true → есть keys → потом стал
      // false → keys остаются stale. Этот edge case покрывается через
      // отдельную one-shot cleanup job (TODO Phase 2 read-api routing).
      recordsSkippedCounter.increment();
      return;
    }

    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);
  }
}
