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

package io.airbyte.integrations.destination.mqtt;

import com.fasterxml.jackson.databind.JsonNode;
import java.util.UUID;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;

public class MqttDestinationConfig {

  private final String clientId;
  private final String serverUri;
  private final String topicPattern;
  private final String testTopic;
  private final MqttConnectOptions options;
  private final boolean retained;
  private final boolean sync;
  private final int qOs;

  private MqttDestinationConfig(final JsonNode config) {
    this.clientId = buildClientId(config);
    this.serverUri = buildServerUri(config);
    this.testTopic = buildTestTopic(config);
    this.topicPattern = buildTopicPattern(config);
    this.options = buildMqttConnectOptions(config);
    this.retained = isRetained(config);
    this.sync = isSyncProducer(config);
    this.qOs = buildQos(config);
  }

  public static MqttDestinationConfig getMqttDestinationConfig(final JsonNode config) {
    return new MqttDestinationConfig(config);
  }

  public String getClientId() {
    return clientId;
  }

  public int getQos() {
    return qOs;
  }

  public MqttConnectOptions getMqttConnectOptions() {
    return options;
  }

  public String getServerUri() {
    return serverUri;
  }

  public String getTestTopic() {
    return testTopic;
  }

  public String getTopicPattern() {
    return topicPattern;
  }

  public boolean isSync() {
    return sync;
  }

  public boolean isRetainedMessage() {
    return retained;
  }

  private String buildClientId(final JsonNode config) {
    if (config.has("client_id")) {
      return config.get("client_id").asText();
    }
    return "airbyte-" + UUID.randomUUID();
  }

  private MqttConnectOptions buildMqttConnectOptions(final JsonNode config) {
    final MqttConnectOptions options = new MqttConnectOptions();
    options.setConnectionTimeout(config.get("connect_timeout").intValue());
    options.setAutomaticReconnect(config.get("automatic_reconnect").booleanValue());
    options.setCleanSession(config.get("clean_session").booleanValue());
    if (config.has("username") && !config.get("username").asText().isBlank()) {
      options.setUserName(config.get("username").asText());
    }
    if (config.has("password") && !config.get("password").asText().isBlank()) {
      options.setPassword(config.get("password").asText().toCharArray());
    }

    return options;
  }

  private String buildServerUri(final JsonNode config) {
    return String.format("%s://%s:%s",
        config.get("use_tls").asBoolean() ? "ssl" : "tcp",
        config.get("broker_host").asText(),
        config.get("broker_port").intValue());
  }

  private String buildTestTopic(final JsonNode config) {
    return config.has("test_topic") ? config.get("test_topic").asText() : "";
  }

  private String buildTopicPattern(final JsonNode config) {
    return config.get("topic_pattern").asText();
  }

  private boolean isRetained(final JsonNode config) {
    return config.get("message_retained").asBoolean();
  }

  private boolean isSyncProducer(final JsonNode config) {
    return config.get("publisher_sync").asBoolean();
  }

  private int buildQos(final JsonNode config) {
    return MessageQoS.valueOf(config.get("message_qos").asText()).getQos();
  }

  private enum MessageQoS {

    AT_MOST_ONCE(0),
    AT_LEAST_ONCE(1),
    EXACTLY_ONCE(2);

    private final int qos;

    MessageQoS(final int qos) {
      this.qos = qos;
    }

    public int getQos() {
      return qos;
    }

  }

}
