/*
 * 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
 *
 *   https://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.accumulo.server.conf;

import static com.google.common.base.Suppliers.memoize;
import static java.util.concurrent.TimeUnit.MINUTES;
import static java.util.concurrent.TimeUnit.NANOSECONDS;

import java.util.Map;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;

import org.apache.accumulo.core.client.TableNotFoundException;
import org.apache.accumulo.core.conf.AccumuloConfiguration;
import org.apache.accumulo.core.conf.ConfigCheckUtil;
import org.apache.accumulo.core.conf.DefaultConfiguration;
import org.apache.accumulo.core.conf.SiteConfiguration;
import org.apache.accumulo.core.data.NamespaceId;
import org.apache.accumulo.core.data.ResourceGroupId;
import org.apache.accumulo.core.data.TableId;
import org.apache.accumulo.core.util.cache.Caches;
import org.apache.accumulo.core.util.cache.Caches.CacheName;
import org.apache.accumulo.core.util.threads.ThreadPools;
import org.apache.accumulo.core.util.threads.Threads;
import org.apache.accumulo.server.ServerContext;
import org.apache.accumulo.server.conf.store.NamespacePropKey;
import org.apache.accumulo.server.conf.store.PropChangeListener;
import org.apache.accumulo.server.conf.store.PropStore;
import org.apache.accumulo.server.conf.store.PropStoreKey;
import org.apache.accumulo.server.conf.store.ResourceGroupPropKey;
import org.apache.accumulo.server.conf.store.SystemPropKey;
import org.apache.accumulo.server.conf.store.TablePropKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.github.benmanes.caffeine.cache.Cache;

import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;

/**
 * A factory for configurations used by a server process. Instance of this class are thread-safe.
 */
public class ServerConfigurationFactory extends ServerConfiguration {
  private final static Logger log = LoggerFactory.getLogger(ServerConfigurationFactory.class);

  // cache expiration is used to remove configurations after deletion, not time sensitive
  private static final int CACHE_EXPIRATION_HRS = 1;
  private final Supplier<SystemConfiguration> systemConfig;
  private final Supplier<ResourceGroupConfiguration> resourceGroupConfig;
  private final Cache<TableId,NamespaceConfiguration> tableParentConfigs;
  private final Cache<TableId,TableConfiguration> tableConfigs;
  private final Cache<NamespaceId,NamespaceConfiguration> namespaceConfigs;

  private final ServerContext context;
  private final SiteConfiguration siteConfig;
  private final ChangeWatcher changeWatcher = new ChangeWatcher();

  private static final int REFRESH_PERIOD_MINUTES = 15;

  private final ConfigRefreshRunner refresher;

  public ServerConfigurationFactory(ServerContext context, SiteConfiguration siteConfig,
      ResourceGroupId rgid) {
    this.context = context;
    this.siteConfig = siteConfig;
    this.systemConfig = memoize(() -> {
      var sysConf = new SystemConfiguration(context, SystemPropKey.of(), siteConfig);
      ConfigCheckUtil.validate(sysConf, "system config");
      return sysConf;
    });
    this.resourceGroupConfig = memoize(() -> {
      return new ResourceGroupConfiguration(context, ResourceGroupPropKey.of(rgid),
          (SystemConfiguration) getSystemConfiguration());
    });
    tableParentConfigs =
        Caches.getInstance().createNewBuilder(CacheName.TABLE_PARENT_CONFIGS, false)
            .expireAfterAccess(CACHE_EXPIRATION_HRS, TimeUnit.HOURS).build();
    tableConfigs = Caches.getInstance().createNewBuilder(CacheName.TABLE_CONFIGS, false)
        .expireAfterAccess(CACHE_EXPIRATION_HRS, TimeUnit.HOURS).build();
    namespaceConfigs = Caches.getInstance().createNewBuilder(CacheName.NAMESPACE_CONFIGS, false)
        .expireAfterAccess(CACHE_EXPIRATION_HRS, TimeUnit.HOURS).build();

    refresher = new ConfigRefreshRunner();
    Runtime.getRuntime().addShutdownHook(
        Threads.createNonCriticalThread("config-refresh-shutdownHook", refresher::shutdown));
  }

  public ServerContext getServerContext() {
    return context;
  }

  public SiteConfiguration getSiteConfiguration() {
    return siteConfig;
  }

  public DefaultConfiguration getDefaultConfiguration() {
    return DefaultConfiguration.getInstance();
  }

  public ResourceGroupConfiguration getResourceGroupConfiguration() {
    return resourceGroupConfig.get();
  }

  @Override
  public AccumuloConfiguration getSystemConfiguration() {
    return systemConfig.get();
  }

  @Override
  public TableConfiguration getTableConfiguration(TableId tableId) {
    return tableConfigs.get(tableId, key -> {
      if (context.tableNodeExists(tableId)) {
        context.getPropStore().registerAsListener(TablePropKey.of(tableId), changeWatcher);
        var conf =
            new TableConfiguration(context, tableId, getNamespaceConfigurationForTable(tableId));
        ConfigCheckUtil.validate(conf, "table id: " + tableId.toString());

        return conf;
      }
      return null;
    });
  }

  public NamespaceConfiguration getNamespaceConfigurationForTable(TableId tableId) {
    NamespaceId namespaceId;
    try {
      namespaceId = context.getNamespaceId(tableId);
    } catch (TableNotFoundException e) {
      throw new IllegalStateException(e);
    }
    return tableParentConfigs.get(tableId, key -> getNamespaceConfiguration(namespaceId));
  }

  @Override
  public NamespaceConfiguration getNamespaceConfiguration(NamespaceId namespaceId) {
    return namespaceConfigs.get(namespaceId, key -> {
      context.getPropStore().registerAsListener(NamespacePropKey.of(namespaceId), changeWatcher);
      var conf = new NamespaceConfiguration(context, namespaceId, getSystemConfiguration());
      ConfigCheckUtil.validate(conf, "namespace id: " + namespaceId.toString());
      return conf;
    });
  }

  private class ChangeWatcher implements PropChangeListener {

    @Override
    public void zkChangeEvent(PropStoreKey propStoreKey) {
      clearLocalOnEvent(propStoreKey);
    }

    @Override
    public void cacheChangeEvent(PropStoreKey propStoreKey) {
      clearLocalOnEvent(propStoreKey);
    }

    @Override
    public void deleteEvent(PropStoreKey propStoreKey) {
      clearLocalOnEvent(propStoreKey);
    }

    private void clearLocalOnEvent(PropStoreKey propStoreKey) {
      // clearing the local secondary cache stored in this class forces a re-read from the prop
      // store
      // to guarantee that the updated vales(s) are re-read on a ZooKeeper change.
      if (propStoreKey instanceof NamespacePropKey) {
        log.trace("configuration snapshot refresh: Handle namespace change for {}", propStoreKey);
        namespaceConfigs.invalidate(((NamespacePropKey) propStoreKey).getId());
        return;
      }
      if (propStoreKey instanceof TablePropKey) {
        log.trace("configuration snapshot refresh: Handle table change for {}", propStoreKey);
        tableConfigs.invalidate(((TablePropKey) propStoreKey).getId());
        tableParentConfigs.invalidate(((TablePropKey) propStoreKey).getId());
      }
    }

    @Override
    public void connectionEvent() {
      // no-op. changes handled by prop store impl
    }
  }

  private class ConfigRefreshRunner {
    private static final long MIN_JITTER_DELAY = 1;
    private static final long MAX_JITTER_DELAY = 23;
    private final ScheduledFuture<?> refreshTaskFuture;

    ConfigRefreshRunner() {

      Runnable refreshTask = this::verifySnapshotVersions;

      ScheduledThreadPoolExecutor executor =
          ThreadPools.getServerThreadPools().createScheduledExecutorService(1, "config-refresh");

      // scheduleWithFixedDelay - used so only one task will run concurrently.
      // staggering the initial delay prevents synchronization of Accumulo servers communicating
      // with ZooKeeper for the sync process. (Value is 25% -> 100% of the refresh period.)
      long randDelay = jitter(REFRESH_PERIOD_MINUTES / 4, REFRESH_PERIOD_MINUTES);
      refreshTaskFuture =
          executor.scheduleWithFixedDelay(refreshTask, randDelay, REFRESH_PERIOD_MINUTES, MINUTES);
    }

    /**
     * Check that the stored version in ZooKeeper matches the version held in the local snapshot.
     * When a mismatch is detected, a change event is sent to the prop store which will cause a
     * re-load. If the Zookeeper node has been deleted, the local cache entries are removed.
     * <p>
     * This method is designed to be called as a scheduled task, so it does not propagate exceptions
     * other than interrupted Exceptions so the scheduled tasks will continue to run.
     */
    private void verifySnapshotVersions() {

      long refreshStart = System.nanoTime();
      int keyCount = 0;
      int keyChangedCount = 0;

      PropStore propStore = context.getPropStore();
      keyCount++;

      // rely on store to propagate change event if different
      propStore.validateDataVersion(SystemPropKey.of(),
          ((ZooBasedConfiguration) getSystemConfiguration()).getDataVersion());
      // small yield - spread out ZooKeeper calls
      jitterDelay();

      for (Map.Entry<NamespaceId,NamespaceConfiguration> entry : namespaceConfigs.asMap()
          .entrySet()) {
        keyCount++;
        PropStoreKey propKey = NamespacePropKey.of(entry.getKey());
        if (!propStore.validateDataVersion(propKey, entry.getValue().getDataVersion())) {
          keyChangedCount++;
          namespaceConfigs.invalidate(entry.getKey());
        }
        // small yield - spread out ZooKeeper calls between namespace config checks
        jitterDelay();
      }

      for (Map.Entry<TableId,TableConfiguration> entry : tableConfigs.asMap().entrySet()) {
        keyCount++;
        TableId tid = entry.getKey();
        PropStoreKey propKey = TablePropKey.of(tid);
        if (!propStore.validateDataVersion(propKey, entry.getValue().getDataVersion())) {
          keyChangedCount++;
          tableConfigs.invalidate(tid);
          tableParentConfigs.invalidate(tid);
          log.debug("data version sync: difference found. forcing configuration update for {}}",
              propKey);
        }
        // small yield - spread out ZooKeeper calls between table config checks
        jitterDelay();
      }

      log.debug("data version sync: Total runtime {} ms for {} entries, changes detected: {}",
          NANOSECONDS.toMillis(System.nanoTime() - refreshStart), keyCount, keyChangedCount);
    }

    /**
     * Generate a small random integer for jitter between [min,max).
     */
    @SuppressFBWarnings(value = "PREDICTABLE_RANDOM",
        justification = "random number not used in secure context")
    private long jitter(final long min, final long max) {
      return ThreadLocalRandom.current().nextLong(min, max);
    }

    /**
     * Sleep for a random jitter interval defined by MIN_JITTER_DELAY and MAX_JITTER_DELAY
     * </p>
     * Used to spread out operations so that Server config sync communications don't overwhelm
     * ZooKeeper and are not synchronized across the cluster.
     *
     */
    private void jitterDelay() {
      try {
        Thread.sleep(jitter(MIN_JITTER_DELAY, MAX_JITTER_DELAY));
      } catch (InterruptedException ex) {
        Thread.currentThread().interrupt();
        throw new IllegalStateException(ex);
      }
    }

    public void shutdown() {
      refreshTaskFuture.cancel(true);
    }
  }
}
