/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.celeborn.server.common.service.config;

import java.util.HashMap;
import java.util.Map;
import java.util.Set;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.celeborn.common.CelebornConf;
import org.apache.celeborn.common.internal.config.ConfigEntry;
import org.apache.celeborn.common.quota.StorageQuota;
import org.apache.celeborn.common.quota.UserTrafficQuota;
import org.apache.celeborn.common.quota.WorkerTrafficQuota;
import org.apache.celeborn.common.tags.WorkerTagsMeta;
import org.apache.celeborn.common.util.Utils;

/**
 * Dynamic configuration is a type of configuration that can be changed at runtime as needed. It can
 * be used at system level/tenant level/tenant user level. When applying dynamic configuration, the
 * priority order is as follows: tenant level overrides system level, which in turn overrides static
 * configuration(CelebornConf). This means that if a configuration is defined at the tenant level,
 * it will be used instead of the system level or static configuration(CelebornConf). If the
 * tenant-level configuration is missing, the system-level configuration will be used. If the
 * system-level configuration is also missing, CelebornConf will be used as the default value.
 */
public abstract class DynamicConfig {
  private static final Logger LOG = LoggerFactory.getLogger(DynamicConfig.class);
  protected volatile Map<String, String> configs = new HashMap<>();
  protected volatile Map<String, Set<String>> tags = null;

  public abstract DynamicConfig getParentLevelConfig();

  public <T> T getWithDefaultValue(
      String configKey, T defaultValue, Class<T> finalType, ConfigType configType) {
    String configValue = configs.get(configKey);
    T formatValue =
        configValue != null ? formatValue(configKey, configValue, finalType, configType) : null;
    if (formatValue == null) {
      return defaultValue;
    } else {
      return formatValue;
    }
  }

  public <T, V> T getValue(
      String configKey, ConfigEntry<V> configEntry, Class<T> finalType, ConfigType configType) {
    String configValue = configs.get(configKey);
    T formatValue =
        configValue != null ? formatValue(configKey, configValue, finalType, configType) : null;
    if (formatValue == null) {
      DynamicConfig parentLevelConfig = getParentLevelConfig();
      return parentLevelConfig != null
          ? parentLevelConfig.getValue(configKey, configEntry, finalType, configType)
          : null;
    } else {
      return formatValue;
    }
  }

  public <T> T formatValue(
      String configKey, String configValue, Class<T> finalType, ConfigType configType) {
    try {
      if (configValue != null) {
        if (ConfigType.BYTES == configType) {
          return convert(finalType, String.valueOf(Utils.byteStringAsBytes(configValue)));
        } else if (ConfigType.TIME_MS == configType) {
          return convert(finalType, String.valueOf(Utils.timeStringAsMs(configValue)));
        } else {
          return convert(finalType, configValue);
        }
      }
    } catch (Exception e) {
      LOG.warn("Config {} value format is not valid, refer to parent if exist", configKey, e);
    }
    return null;
  }

  public StorageQuota getTenantStorageQuota() {
    return new StorageQuota(
        getValue(
            CelebornConf.QUOTA_TENANT_DISK_BYTES_WRITTEN().key(),
            CelebornConf.QUOTA_TENANT_DISK_BYTES_WRITTEN(),
            Long.TYPE,
            ConfigType.BYTES),
        getValue(
            CelebornConf.QUOTA_TENANT_DISK_FILE_COUNT().key(),
            CelebornConf.QUOTA_TENANT_DISK_FILE_COUNT(),
            Long.TYPE,
            ConfigType.STRING),
        getValue(
            CelebornConf.QUOTA_TENANT_HDFS_BYTES_WRITTEN().key(),
            CelebornConf.QUOTA_TENANT_HDFS_BYTES_WRITTEN(),
            Long.TYPE,
            ConfigType.BYTES),
        getValue(
            CelebornConf.QUOTA_TENANT_HDFS_FILE_COUNT().key(),
            CelebornConf.QUOTA_TENANT_HDFS_FILE_COUNT(),
            Long.TYPE,
            ConfigType.STRING));
  }

  public UserTrafficQuota getUserTrafficQuota() {
    return new UserTrafficQuota(
        getValue(
            CelebornConf.WORKER_CONGESTION_CONTROL_USER_PRODUCE_SPEED_HIGH_WATERMARK().key(),
            CelebornConf.WORKER_CONGESTION_CONTROL_USER_PRODUCE_SPEED_HIGH_WATERMARK(),
            Long.TYPE,
            ConfigType.BYTES),
        getValue(
            CelebornConf.WORKER_CONGESTION_CONTROL_USER_PRODUCE_SPEED_LOW_WATERMARK().key(),
            CelebornConf.WORKER_CONGESTION_CONTROL_USER_PRODUCE_SPEED_LOW_WATERMARK(),
            Long.TYPE,
            ConfigType.BYTES));
  }

  public WorkerTrafficQuota getWorkerTrafficQuota() {
    return new WorkerTrafficQuota(
        getValue(
            CelebornConf.WORKER_CONGESTION_CONTROL_DISK_BUFFER_HIGH_WATERMARK().key(),
            CelebornConf.WORKER_CONGESTION_CONTROL_DISK_BUFFER_HIGH_WATERMARK(),
            Long.TYPE,
            ConfigType.BYTES),
        getValue(
            CelebornConf.WORKER_CONGESTION_CONTROL_DISK_BUFFER_LOW_WATERMARK().key(),
            CelebornConf.WORKER_CONGESTION_CONTROL_DISK_BUFFER_LOW_WATERMARK(),
            Long.TYPE,
            ConfigType.BYTES),
        getValue(
            CelebornConf.WORKER_CONGESTION_CONTROL_WORKER_PRODUCE_SPEED_HIGH_WATERMARK().key(),
            CelebornConf.WORKER_CONGESTION_CONTROL_WORKER_PRODUCE_SPEED_HIGH_WATERMARK(),
            Long.TYPE,
            ConfigType.BYTES),
        getValue(
            CelebornConf.WORKER_CONGESTION_CONTROL_WORKER_PRODUCE_SPEED_LOW_WATERMARK().key(),
            CelebornConf.WORKER_CONGESTION_CONTROL_WORKER_PRODUCE_SPEED_LOW_WATERMARK(),
            Long.TYPE,
            ConfigType.BYTES));
  }

  public WorkerTagsMeta getWorkerTagsMeta() {
    return new WorkerTagsMeta(
        getValue(
            CelebornConf.TAGS_EXPR().key(),
            CelebornConf.TAGS_EXPR(),
            String.class,
            ConfigType.STRING),
        getValue(
            CelebornConf.PREFER_CLIENT_TAGS_EXPR().key(),
            CelebornConf.PREFER_CLIENT_TAGS_EXPR(),
            Boolean.TYPE,
            ConfigType.STRING));
  }

  public StorageQuota getClusterStorageQuota() {
    return new StorageQuota(
        getValue(
            CelebornConf.QUOTA_CLUSTER_DISK_BYTES_WRITTEN().key(),
            CelebornConf.QUOTA_CLUSTER_DISK_BYTES_WRITTEN(),
            Long.TYPE,
            ConfigType.BYTES),
        getValue(
            CelebornConf.QUOTA_CLUSTER_DISK_FILE_COUNT().key(),
            CelebornConf.QUOTA_CLUSTER_DISK_FILE_COUNT(),
            Long.TYPE,
            ConfigType.STRING),
        getValue(
            CelebornConf.QUOTA_CLUSTER_HDFS_BYTES_WRITTEN().key(),
            CelebornConf.QUOTA_CLUSTER_HDFS_BYTES_WRITTEN(),
            Long.TYPE,
            ConfigType.BYTES),
        getValue(
            CelebornConf.QUOTA_CLUSTER_HDFS_FILE_COUNT().key(),
            CelebornConf.QUOTA_CLUSTER_HDFS_FILE_COUNT(),
            Long.TYPE,
            ConfigType.STRING));
  }

  public StorageQuota getUserStorageQuota() {
    return new StorageQuota(
        getValue(
            CelebornConf.QUOTA_USER_DISK_BYTES_WRITTEN().key(),
            CelebornConf.QUOTA_USER_DISK_BYTES_WRITTEN(),
            Long.TYPE,
            ConfigType.BYTES),
        getValue(
            CelebornConf.QUOTA_USER_DISK_FILE_COUNT().key(),
            CelebornConf.QUOTA_USER_DISK_FILE_COUNT(),
            Long.TYPE,
            ConfigType.STRING),
        getValue(
            CelebornConf.QUOTA_USER_HDFS_BYTES_WRITTEN().key(),
            CelebornConf.QUOTA_USER_HDFS_BYTES_WRITTEN(),
            Long.TYPE,
            ConfigType.BYTES),
        getValue(
            CelebornConf.QUOTA_USER_HDFS_FILE_COUNT().key(),
            CelebornConf.QUOTA_USER_HDFS_FILE_COUNT(),
            Long.TYPE,
            ConfigType.STRING));
  }

  public boolean interruptShuffleEnabled() {
    return getValue(
        CelebornConf.QUOTA_INTERRUPT_SHUFFLE_ENABLED().key(),
        CelebornConf.QUOTA_INTERRUPT_SHUFFLE_ENABLED(),
        Boolean.TYPE,
        ConfigType.BOOLEAN);
  }

  public boolean clusterQuotaEnabled() {
    return getValue(
        CelebornConf.CLUSTER_QUOTA_ENABLED().key(),
        CelebornConf.CLUSTER_QUOTA_ENABLED(),
        Boolean.TYPE,
        ConfigType.BOOLEAN);
  }

  public boolean tenantQuotaEnabled() {
    return getValue(
        CelebornConf.TENANT_QUOTA_ENABLED().key(),
        CelebornConf.TENANT_QUOTA_ENABLED(),
        Boolean.TYPE,
        ConfigType.BOOLEAN);
  }

  public boolean userQuotaEnabled() {
    return getValue(
        CelebornConf.USER_QUOTA_ENABLED().key(),
        CelebornConf.USER_QUOTA_ENABLED(),
        Boolean.TYPE,
        ConfigType.BOOLEAN);
  }

  public Map<String, String> getConfigs() {
    return configs;
  }

  public Map<String, Set<String>> getTags() {
    if (tags == null) {
      synchronized (DynamicConfig.class) {
        if (tags == null) {
          tags = currentTags();
        }
      }
    }
    return tags;
  }

  protected Map<String, Set<String>> currentTags() {
    return null;
  }

  @Override
  public String toString() {
    final StringBuilder sb = new StringBuilder("DynamicConfig{");
    sb.append("configs=").append(configs);
    sb.append('}');
    return sb.toString();
  }

  public enum ConfigType {
    BYTES,
    STRING,
    TIME_MS,
    BOOLEAN
  }

  public static <T> T convert(Class<T> clazz, String value) {
    if (Boolean.TYPE == clazz) {
      return (T) Boolean.valueOf(value);
    } else if (Byte.TYPE == clazz) {
      return (T) Byte.valueOf(value);
    } else if (Short.TYPE == clazz) {
      return (T) Short.valueOf(value);
    } else if (Integer.TYPE == clazz) {
      return (T) Integer.valueOf(value);
    } else if (Long.TYPE == clazz) {
      return (T) Long.valueOf(value);
    } else if (Float.TYPE == clazz) {
      return (T) Float.valueOf(value);
    } else if (Double.TYPE == clazz) {
      return (T) Double.valueOf(value);
    }
    return (T) value;
  }
}
