/*
 * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
 * (the "License"). You may not use this work except in compliance with the License, which is
 * available at www.apache.org/licenses/LICENSE-2.0
 *
 * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
 * either express or implied, as more fully set forth in the License.
 *
 * See the NOTICE file distributed with this work for information regarding copyright ownership.
 */

package alluxio.underfs;

import alluxio.AlluxioURI;
import alluxio.conf.Configuration;
import alluxio.exception.InvalidPathException;
import alluxio.master.journal.DelegatingJournaled;
import alluxio.master.journal.JournalContext;
import alluxio.master.journal.Journaled;
import alluxio.master.journal.checkpoint.CheckpointName;
import alluxio.proto.journal.File;
import alluxio.proto.journal.File.UpdateUfsModeEntry;
import alluxio.proto.journal.Journal.JournalEntry;
import alluxio.recorder.Recorder;
import alluxio.resource.CloseableIterator;
import alluxio.util.network.NetworkAddressUtils;

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

import java.io.IOException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.function.Supplier;
import javax.annotation.concurrent.ThreadSafe;

/**
 * A class that manages the UFS for master servers.
 */
@ThreadSafe
public final class MasterUfsManager extends AbstractUfsManager implements DelegatingJournaled {
  private static final Logger LOG = LoggerFactory.getLogger(MasterUfsManager.class);

  private final State mState = new State();

  /** A set of all managed ufs roots. */
  private final Set<String> mUfsRoots = new HashSet<>();

  /** Mapping from mount ID to ufs root. */
  private final Map<Long, String> mIdToRoot = new HashMap<>();

  @Override
  protected void connectUfs(UnderFileSystem fs) throws IOException {
    fs.connectFromMaster(
        NetworkAddressUtils.getConnectHost(NetworkAddressUtils.ServiceType.MASTER_RPC,
            Configuration.global()));
  }

  @Override
  public synchronized void addMount(long mountId, final AlluxioURI ufsUri,
      final UnderFileSystemConfiguration ufsConf) {
    addMountWithRecorder(mountId, ufsUri, ufsConf, Recorder.noopRecorder());
  }

  @Override
  public synchronized void addMountWithRecorder(long mountId, final AlluxioURI ufsUri,
      final UnderFileSystemConfiguration ufsConf, Recorder recorder) {
    super.addMountWithRecorder(mountId, ufsUri, ufsConf, recorder);
    String root = ufsUri.getRootPath();
    mUfsRoots.add(root);
    mIdToRoot.put(mountId, root);
  }

  @Override
  public synchronized void removeMount(long mountId) {
    mIdToRoot.remove(mountId);
    super.removeMount(mountId);
  }

  @Override
  public boolean hasMount(long mountId) {
    return mIdToRoot.containsKey(mountId);
  }

  /**
   * Get the physical ufs operation modes for the {@link UnderFileSystem} under the given Mount
   * table resolution.
   *
   * @param physicalStores the physical stores for the mount resolution
   * @return the state of physical UFS for given mount resolution
   */
  public synchronized Map<String, UfsMode> getPhysicalUfsState(
      List<String> physicalStores) {
    Map<String, UfsMode> ufsModeState = new HashMap<>();
    for (String physicalUfs : physicalStores) {
      ufsModeState.put(physicalUfs, mState.getUfsMode(new AlluxioURI(physicalUfs).getRootPath()));
    }
    return ufsModeState;
  }

  /**
   * Set the operation mode the given physical ufs.
   *
   * @param journalContext the journal context
   * @param ufsPath the physical ufs path (scheme and authority only)
   * @param ufsMode the ufs operation mode
   * @throws InvalidPathException if no managed ufs covers the given path
   */
  public synchronized void setUfsMode(Supplier<JournalContext> journalContext, AlluxioURI ufsPath,
      UfsMode ufsMode) throws InvalidPathException {
    LOG.info("Set ufs mode for {} to {}", ufsPath, ufsMode);

    String root = ufsPath.getRootPath();
    if (!mUfsRoots.contains(root)) {
      LOG.warn("No managed ufs for physical ufs path {}", root);
      throw new InvalidPathException(String.format("Unknown Ufs path %s", root));
    }

    mState.applyAndJournal(journalContext, UpdateUfsModeEntry.newBuilder()
        .setUfsPath(ufsPath.getRootPath())
        .setUfsMode(File.UfsMode.valueOf(ufsMode.name()))
        .build());
  }

  @Override
  public Journaled getDelegate() {
    return mState;
  }

  @Override
  public CloseableIterator<JournalEntry> getJournalEntryIterator() {
    return mState.getJournalEntryIterator();
  }

  private static class State implements Journaled {
    // The physical ufs state for all managed mounts. The keys are URIs normalized to set the path
    // to "/", e.g. "hdfs://namenode/" or just "/" for local filesystem.
    private final Map<String, UfsMode> mUfsModes = new HashMap<>();

    /**
     * @param key a root ufs path
     * @return the mode for that ufs
     */
    public UfsMode getUfsMode(String key) {
      return mUfsModes.getOrDefault(key, UfsMode.READ_WRITE);
    }

    @Override
    public void resetState() {
      mUfsModes.clear();
    }

    @Override
    public boolean processJournalEntry(JournalEntry entry) {
      if (entry.hasUpdateUfsMode()) {
        apply(entry.getUpdateUfsMode());
      } else {
        return false;
      }
      return true;
    }

    /**
     * @param context journal context
     * @param entry update ufs mode entry
     */
    public void applyAndJournal(Supplier<JournalContext> context, UpdateUfsModeEntry entry) {
      applyAndJournal(context, JournalEntry.newBuilder().setUpdateUfsMode(entry).build());
    }

    private void apply(UpdateUfsModeEntry entry) {
      mUfsModes.put(entry.getUfsPath(), UfsMode.valueOf(entry.getUfsMode().name()));
    }

    @Override
    public CloseableIterator<JournalEntry> getJournalEntryIterator() {
      return CloseableIterator.noopCloseable(mUfsModes.entrySet().stream()
          .map(e -> JournalEntry.newBuilder().setUpdateUfsMode(UpdateUfsModeEntry.newBuilder()
              .setUfsPath(e.getKey())
              .setUfsMode(File.UfsMode.valueOf(e.getValue().name())))
              .build())
          .iterator());
    }

    @Override
    public CheckpointName getCheckpointName() {
      return CheckpointName.MASTER_UFS_MANAGER;
    }
  }
}
