/*
 * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
 */

package io.airbyte.integrations.debezium.internals;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import io.airbyte.commons.json.Jsons;
import io.airbyte.integrations.debezium.CdcMetadataInjector;
import io.airbyte.protocol.models.AirbyteMessage;
import io.airbyte.protocol.models.AirbyteRecordMessage;
import io.debezium.engine.ChangeEvent;
import java.sql.Timestamp;
import java.time.Instant;

public class DebeziumEventUtils {

  public static final String CDC_UPDATED_AT = "_ab_cdc_updated_at";
  public static final String CDC_DELETED_AT = "_ab_cdc_deleted_at";

  public static AirbyteMessage toAirbyteMessage(final ChangeEvent<String, String> event,
                                                final CdcMetadataInjector cdcMetadataInjector,
                                                final Instant emittedAt) {
    final JsonNode debeziumRecord = Jsons.deserialize(event.value());
    final JsonNode before = debeziumRecord.get("before");
    final JsonNode after = debeziumRecord.get("after");
    final JsonNode source = debeziumRecord.get("source");

    final JsonNode data = formatDebeziumData(before, after, source, cdcMetadataInjector);
    final String schemaName = cdcMetadataInjector.namespace(source);
    final String streamName = source.get("table").asText();

    final AirbyteRecordMessage airbyteRecordMessage = new AirbyteRecordMessage()
        .withStream(streamName)
        .withNamespace(schemaName)
        .withEmittedAt(emittedAt.toEpochMilli())
        .withData(data);

    return new AirbyteMessage()
        .withType(AirbyteMessage.Type.RECORD)
        .withRecord(airbyteRecordMessage);
  }

  // warning mutates input args.
  private static JsonNode formatDebeziumData(final JsonNode before,
                                             final JsonNode after,
                                             final JsonNode source,
                                             final CdcMetadataInjector cdcMetadataInjector) {
    final ObjectNode base = (ObjectNode) (after.isNull() ? before : after);

    final long transactionMillis = source.get("ts_ms").asLong();
    final String transactionTimestamp = new Timestamp(transactionMillis).toInstant().toString();

    base.put(CDC_UPDATED_AT, transactionTimestamp);
    cdcMetadataInjector.addMetaData(base, source);

    if (after.isNull()) {
      base.put(CDC_DELETED_AT, transactionTimestamp);
    } else {
      base.put(CDC_DELETED_AT, (String) null);
    }

    return base;
  }

}
