/*
 * 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.master.file;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;

import alluxio.AlluxioURI;
import alluxio.conf.Configuration;
import alluxio.conf.PropertyKey;
import alluxio.exception.AccessControlException;
import alluxio.exception.FileAlreadyExistsException;
import alluxio.exception.FileDoesNotExistException;
import alluxio.exception.InvalidPathException;
import alluxio.file.options.DescendantType;
import alluxio.grpc.CompleteFilePOptions;
import alluxio.grpc.CreateFilePOptions;
import alluxio.grpc.DeletePOptions;
import alluxio.grpc.FileSystemMasterCommonPOptions;
import alluxio.grpc.GetStatusPOptions;
import alluxio.grpc.ListStatusPOptions;
import alluxio.grpc.WritePType;
import alluxio.heartbeat.HeartbeatContext;
import alluxio.heartbeat.ManuallyScheduleHeartbeat;
import alluxio.master.CoreMasterContext;
import alluxio.master.MasterFactory;
import alluxio.master.MasterRegistry;
import alluxio.master.MasterTestUtils;
import alluxio.master.block.BlockMaster;
import alluxio.master.block.BlockMasterFactory;
import alluxio.master.file.contexts.CompleteFileContext;
import alluxio.master.file.contexts.CreateDirectoryContext;
import alluxio.master.file.contexts.CreateFileContext;
import alluxio.master.file.contexts.DeleteContext;
import alluxio.master.file.contexts.GetStatusContext;
import alluxio.master.file.contexts.ListStatusContext;
import alluxio.master.file.contexts.MountContext;
import alluxio.master.file.meta.Inode;
import alluxio.master.file.meta.LockedInodePath;
import alluxio.master.journal.JournalSystem;
import alluxio.master.journal.JournalTestUtils;
import alluxio.master.journal.JournalType;
import alluxio.master.metrics.MetricsMasterFactory;
import alluxio.security.authentication.AuthenticatedClientUser;
import alluxio.security.user.UserState;
import alluxio.underfs.Fingerprint;
import alluxio.underfs.UfsDirectoryStatus;
import alluxio.underfs.UfsFileStatus;
import alluxio.underfs.UfsMode;
import alluxio.underfs.UfsStatus;
import alluxio.underfs.UnderFileSystem;
import alluxio.underfs.options.GetStatusOptions;
import alluxio.util.IdUtils;
import alluxio.util.ModeUtils;
import alluxio.util.ThreadFactoryUtils;
import alluxio.util.executor.ExecutorServiceFactories;
import alluxio.util.executor.ExecutorServiceFactory;
import alluxio.util.io.PathUtils;
import alluxio.wire.FileInfo;

import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import org.junit.runner.RunWith;
import org.mockito.Mockito;
import org.powermock.api.mockito.PowerMockito;
import org.powermock.core.classloader.annotations.PrepareForTest;
import org.powermock.modules.junit4.PowerMockRunner;

import java.io.File;
import java.io.IOException;
import java.time.Clock;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Function;
import javax.annotation.Nullable;

/**
 * Unit tests for {@link FileSystemMaster}.
 */
@RunWith(PowerMockRunner.class)
@PrepareForTest({UnderFileSystem.Factory.class})
public final class FileSystemMasterSyncMetadataTest {
  private File mJournalFolder;
  private MasterRegistry mRegistry;
  private FileSystemMaster mFileSystemMaster;
  private UnderFileSystem mUfs;
  private ExecutorService mExecutorService;

  @Rule
  public ManuallyScheduleHeartbeat mManualScheduler =
      new ManuallyScheduleHeartbeat(HeartbeatContext.MASTER_PERSISTENCE_CHECKER,
          HeartbeatContext.MASTER_PERSISTENCE_SCHEDULER);

  @Before
  public void before() throws Exception {
    UserState s = UserState.Factory.create(Configuration.global());
    AuthenticatedClientUser.set(s.getUser().getName());
    TemporaryFolder tmpFolder = new TemporaryFolder();
    tmpFolder.create();
    File ufsRoot = tmpFolder.newFolder();
    Configuration.set(PropertyKey.MASTER_JOURNAL_TYPE, JournalType.UFS);
    Configuration.set(PropertyKey.MASTER_MOUNT_TABLE_ROOT_UFS, ufsRoot.getAbsolutePath());
    Configuration.set(PropertyKey.MASTER_PERSISTENCE_INITIAL_INTERVAL_MS, 0);
    Configuration.set(PropertyKey.MASTER_PERSISTENCE_MAX_INTERVAL_MS, 1000);
    Configuration.set(PropertyKey.MASTER_PERSISTENCE_MAX_TOTAL_WAIT_TIME_MS, 1000);
    mJournalFolder = tmpFolder.newFolder();
    startServices();
  }

  /**
   * Resets global state after each test run.
   */
  @After
  public void after() throws Exception {
    stopServices();
  }

  @Test
  public void completeFileWithOutOfDateHash() throws Exception {
    // In this test we want to simulate a concurrent write to the UFS
    // while the file is being created in Alluxio.
    // When creating the file in Alluxio, we will use the fingerprint
    // of the created file, and not the one on the UFS.
    // Thus, when performing a metadata sync there should be a fingerprint
    // mismatch.
    AlluxioURI ufsMount = setupMockUfsS3Mount();
    String fname = "file";
    AlluxioURI uri = new AlluxioURI("/mnt/local/" + fname);

    // The fingerprint of the file created in Alluxio
    String alluxioContentHash = "hashOnComplete";
    // The fingerprint of the file in the UFS
    String ufsContentHash = "ufsHash";

    AlluxioURI filePath = ufsMount.join("file");
    UfsFileStatus fileStatus =
        new UfsFileStatus("file", ufsContentHash, 0L, System.currentTimeMillis(), "owner1",
            "owner1", (short) 777, null, 100L);
    Mockito.doAnswer(invocation -> Fingerprint.create("s3", fileStatus, invocation.getArgument(1)))
           .when(mUfs).getParsedFingerprint(eq(filePath.toString()), anyString());
    Mockito.doAnswer(invocation -> Fingerprint.create("s3", fileStatus)).when(mUfs)
           .getParsedFingerprint(eq(filePath.toString()));
    Mockito.when(mUfs.exists(filePath.toString())).thenReturn(true);
    Mockito.when(mUfs.isDirectory(filePath.toString())).thenReturn(false);
    Mockito.when(mUfs.isFile(filePath.toString())).thenReturn(true);
    Mockito.when(mUfs.getStatus(eq(filePath.toString()), any(GetStatusOptions.class)))
           .thenReturn(fileStatus);
    Mockito.when(mUfs.getOperationMode(any())).thenReturn(UfsMode.READ_WRITE);

    mFileSystemMaster.createFile(uri, CreateFileContext.mergeFrom(
        CreateFilePOptions.newBuilder().setWriteType(WritePType.THROUGH)));
    mFileSystemMaster.completeFile(uri, CompleteFileContext.mergeFrom(
        CompleteFilePOptions.newBuilder().setContentHash(alluxioContentHash)));

    FileInfo info = mFileSystemMaster.getFileInfo(uri, GetStatusContext.defaults());
    assertEquals(alluxioContentHash,
        Fingerprint.parse(info.getUfsFingerprint()).getTag(Fingerprint.Tag.CONTENT_HASH));

    // After syncing we should have the new version of the file with the new fingerprint
    info = mFileSystemMaster.getFileInfo(uri, GetStatusContext.mergeFrom(
        GetStatusPOptions.newBuilder().setCommonOptions(
            FileSystemMasterCommonPOptions.newBuilder().setSyncIntervalMs(0).build())));
    assertEquals(ufsContentHash,
        Fingerprint.parse(info.getUfsFingerprint()).getTag(Fingerprint.Tag.CONTENT_HASH));
  }

  @Test
  public void setAttributeOwnerGroupOnMetadataUpdate() throws Exception {
    AlluxioURI ufsMount = setupMockUfsS3Mount();
    String fname = "file";
    AlluxioURI uri = new AlluxioURI("/mnt/local/" + fname);
    short mode = ModeUtils.getUMask("0700").toShort();

    // Mock dir1 ufs path
    AlluxioURI filePath = ufsMount.join("file");
    UfsFileStatus fileStatus =
        new UfsFileStatus("file", "", 0L, System.currentTimeMillis(), "owner1", "owner1",
            (short) 777, null, 100L);
    Mockito.when(mUfs.getParsedFingerprint(filePath.toString()))
           .thenReturn(Fingerprint.create("s3", fileStatus));
    Mockito.when(mUfs.exists(filePath.toString())).thenReturn(true);
    Mockito.when(mUfs.isDirectory(filePath.toString())).thenReturn(false);
    Mockito.when(mUfs.isFile(filePath.toString())).thenReturn(true);
    Mockito.when(mUfs.getStatus(eq(filePath.toString()), any(GetStatusOptions.class)))
           .thenReturn(fileStatus);

    List<FileInfo> f1 = mFileSystemMaster.listStatus(uri, ListStatusContext.mergeFrom(
        ListStatusPOptions.newBuilder().setCommonOptions(
            FileSystemMasterCommonPOptions.newBuilder().setSyncIntervalMs(0).build())));
    UfsFileStatus updatedStatus =
        new UfsFileStatus("file", "", 0, System.currentTimeMillis(), "owner2", "owner2",
            (short) 777, null, 100);
    Mockito.when(mUfs.getStatus(eq(filePath.toString()), any(GetStatusOptions.class)))
           .thenReturn(updatedStatus);
    Mockito.when(mUfs.getParsedFingerprint(filePath.toString()))
           .thenReturn(Fingerprint.create("s3", updatedStatus));

    FileInfo res = mFileSystemMaster.getFileInfo(uri, GetStatusContext.mergeFrom(
        GetStatusPOptions.newBuilder().setCommonOptions(
            FileSystemMasterCommonPOptions.newBuilder().setSyncIntervalMs(0).build())));
    assertEquals("owner2", res.getOwner());
    assertEquals("owner2", res.getGroup());
  }

  @Deprecated
  public void listStatusWithSyncMetadataAndEmptyS3Owner() throws Exception {

    AlluxioURI ufsMount = setupMockUfsS3Mount();
    short mode = ModeUtils.getUMask("0700").toShort();

    // Mock dir1 ufs path
    AlluxioURI dir1Path = ufsMount.join("dir1");
    UfsDirectoryStatus dir1Status = new UfsDirectoryStatus(dir1Path.getPath(), "", "", mode);
    Mockito.when(mUfs.getParsedFingerprint(dir1Path.toString()))
           .thenReturn(Fingerprint.create("s3", dir1Status));
    Mockito.when(mUfs.exists(dir1Path.toString())).thenReturn(true);
    Mockito.when(mUfs.isDirectory(dir1Path.toString())).thenReturn(true);
    Mockito.when(mUfs.isFile(dir1Path.toString())).thenReturn(false);
    Mockito.when(mUfs.getStatus(eq(dir1Path.toString()), any(GetStatusOptions.class)))
           .thenReturn(dir1Status);
    Mockito.when(mUfs.getDirectoryStatus(dir1Path.toString())).thenReturn(dir1Status);

    // Mock nested ufs path
    AlluxioURI nestedFilePath = ufsMount.join("dir1").join("file1");
    UfsFileStatus nestedFileStatus =
        new UfsFileStatus(nestedFilePath.getPath(), "dummy", 0, null, "", "", mode, 1024);
    Mockito.when(mUfs.getParsedFingerprint(nestedFilePath.toString()))
           .thenReturn(Fingerprint.create("s3", nestedFileStatus));
    Mockito.when(mUfs.getStatus(eq(nestedFilePath.toString()), any(GetStatusOptions.class)))
           .thenReturn(nestedFileStatus);
    Mockito.when(mUfs.isDirectory(nestedFilePath.toString())).thenReturn(false);
    Mockito.when(mUfs.isFile(nestedFilePath.toString())).thenReturn(true);
    Mockito.when(mUfs.getFileStatus(nestedFilePath.toString())).thenReturn(nestedFileStatus);
    Mockito.when(mUfs.exists(nestedFilePath.toString())).thenReturn(true);

    // Create directory in Alluxio only
    AlluxioURI dir1 = new AlluxioURI("/mnt/local/dir1");
    mFileSystemMaster.createDirectory(dir1, CreateDirectoryContext.defaults());

    // Mock creating the same directory and nested file in UFS out of band
    Mockito.when(mUfs.listStatus(eq(dir1Path.toString()))).thenReturn(
        new UfsStatus[] {new UfsFileStatus("file1", "dummy", 0, null, "", "", mode, 1024)});

    // List with sync.interval=0
    List<FileInfo> fileInfoList = mFileSystemMaster.listStatus(dir1, ListStatusContext.mergeFrom(
        ListStatusPOptions.newBuilder().setCommonOptions(
            FileSystemMasterCommonPOptions.newBuilder().setSyncIntervalMs(0).build())));
    assertEquals(1, fileInfoList.size());

    // Verify owner/group is not empty
    FileInfo mountLocalInfo =
        mFileSystemMaster.getFileInfo(new AlluxioURI("/mnt/local"), GetStatusContext.defaults());
    assertEquals(mountLocalInfo.getOwner(),
        mFileSystemMaster.getFileInfo(dir1, GetStatusContext.defaults()).getOwner());
    assertEquals(mountLocalInfo.getGroup(),
        mFileSystemMaster.getFileInfo(dir1, GetStatusContext.defaults()).getGroup());
    AlluxioURI file1 = new AlluxioURI("/mnt/local/dir1/file1");
    assertEquals(mountLocalInfo.getOwner(),
        mFileSystemMaster.getFileInfo(file1, GetStatusContext.defaults()).getOwner());
    assertEquals(mountLocalInfo.getGroup(),
        mFileSystemMaster.getFileInfo(file1, GetStatusContext.defaults()).getGroup());
  }

  private AlluxioURI setupMockUfsS3Mount()
      throws IOException, FileDoesNotExistException, FileAlreadyExistsException,
      AccessControlException, InvalidPathException {
    mFileSystemMaster.createDirectory(new AlluxioURI("/mnt/"), CreateDirectoryContext.defaults());
    // Mock ufs mount
    AlluxioURI ufsMount = new AlluxioURI("s3a://bucket/");
    Mockito.when(mUfs.getUnderFSType()).thenReturn("s3");
    Mockito.when(mUfs.isObjectStorage()).thenReturn(true);
    Mockito.when(mUfs.isDirectory(ufsMount.toString())).thenReturn(true);
    short mode = ModeUtils.getUMask("0700").toShort();
    Mockito.when(mUfs.getExistingDirectoryStatus(ufsMount.toString()))
           .thenReturn(new UfsDirectoryStatus(ufsMount.toString(), "", "", mode));
    Mockito.when(mUfs.resolveUri(Mockito.eq(ufsMount), anyString())).thenAnswer(
        invocation -> new AlluxioURI(ufsMount,
            PathUtils.concatPath(ufsMount.getPath(), invocation.getArgument(1, String.class)),
            false));

    // Mount
    AlluxioURI mountLocal = new AlluxioURI("/mnt/local");
    mFileSystemMaster.mount(mountLocal, ufsMount, MountContext.defaults());

    return ufsMount;
  }

  @Deprecated
  public void deleteAlluxioOnlyNoSync() throws Exception {
    // Prepare files
    mFileSystemMaster.createDirectory(new AlluxioURI("/a/"), CreateDirectoryContext.defaults());
    mFileSystemMaster.createDirectory(new AlluxioURI("/a/b/"), CreateDirectoryContext.defaults());
    mFileSystemMaster.createDirectory(new AlluxioURI("/b/"), CreateDirectoryContext.defaults());
    // If the sync operation happens, the flag will be marked
    SyncAwareFileSystemMaster delegateMaster = (SyncAwareFileSystemMaster) mFileSystemMaster;
    delegateMaster.setSynced(false);

    delegateMaster.delete(new AlluxioURI("/a/"), DeleteContext.mergeFrom(
        DeletePOptions.newBuilder().setRecursive(true).setAlluxioOnly(true)));
    // The files have been deleted
    assertEquals(IdUtils.INVALID_FILE_ID, mFileSystemMaster.getFileId(new AlluxioURI("/a/")));
    assertEquals(IdUtils.INVALID_FILE_ID, mFileSystemMaster.getFileId(new AlluxioURI("/a/b/")));
    // Irrelevant files are not affected
    assertNotEquals(IdUtils.INVALID_FILE_ID, mFileSystemMaster.getFileId(new AlluxioURI("/b/")));
    // Sync has not happened
    assertFalse(delegateMaster.mSynced.get());
  }

  /**
   * Tests the getStatus operation does not trigger a metadata sync that loads its children.
   */
  @Test
  public void getStatusOnDirectory() throws Exception {
    AlluxioURI ufsMount = setupMockUfsS3Mount();
    short mode = ModeUtils.getUMask("0700").toShort();

    // Mock dir1 ufs path
    AlluxioURI dir1Path = ufsMount.join("dir1");
    UfsDirectoryStatus dir1Status = new UfsDirectoryStatus(dir1Path.getPath(), "", "", mode);
    Mockito.when(mUfs.getParsedFingerprint(dir1Path.toString()))
        .thenReturn(Fingerprint.create("s3", dir1Status));
    Mockito.when(mUfs.exists(dir1Path.toString())).thenReturn(true);
    Mockito.when(mUfs.isDirectory(dir1Path.toString())).thenReturn(true);
    Mockito.when(mUfs.isFile(dir1Path.toString())).thenReturn(false);
    Mockito.when(mUfs.getStatus(eq(dir1Path.toString()), any(GetStatusOptions.class)))
           .thenReturn(dir1Status);
    Mockito.when(mUfs.getDirectoryStatus(dir1Path.toString())).thenReturn(dir1Status);

    // Mock nested ufs path /dir1/dir2
    AlluxioURI nestedDirectoryPath = ufsMount.join("dir1").join("dir2");
    UfsDirectoryStatus nestedDirStatus =
        new UfsDirectoryStatus(dir1Path.getPath(), "", "", mode);

    Mockito.when(mUfs.getParsedFingerprint(nestedDirectoryPath.toString()))
        .thenReturn(Fingerprint.create("s3", nestedDirStatus));
    Mockito.when(mUfs.exists(nestedDirectoryPath.toString())).thenReturn(true);
    Mockito.when(mUfs.isDirectory(nestedDirectoryPath.toString())).thenReturn(true);
    Mockito.when(mUfs.isFile(nestedDirectoryPath.toString())).thenReturn(false);
    Mockito.when(mUfs.getStatus(eq(nestedDirectoryPath.toString()), any(GetStatusOptions.class)))
           .thenReturn(nestedDirStatus);
    Mockito.when(mUfs.getDirectoryStatus(nestedDirectoryPath.toString()))
        .thenReturn(nestedDirStatus);

    // Mock creating the same directory and nested file in UFS out of band
    AlluxioURI dir1 = new AlluxioURI("/mnt/local/dir1");
    AlluxioURI dir2 = new AlluxioURI("/mnt/local/dir1/dir2");
    Mockito.when(mUfs.listStatus(eq(dir1Path.toString())))
        .thenReturn(new UfsStatus[]{new UfsDirectoryStatus("dir2", "", "", mode)});
    Mockito.when(mUfs.listStatus(eq(nestedDirectoryPath.toString())))
        .thenReturn(new UfsStatus[]{});

    // List the nested directory
    // listStatus is called on UFS /dir1/dir2
    mFileSystemMaster.listStatus(dir2, ListStatusContext.mergeFrom(
            ListStatusPOptions.newBuilder().setCommonOptions(
                FileSystemMasterCommonPOptions.newBuilder().setSyncIntervalMs(0).build())));
    Mockito.verify(mUfs, Mockito.times(0))
        .listStatus(eq(dir1Path.toString()));
    Mockito.verify(mUfs, Mockito.times(1))
        .listStatus(eq(nestedDirectoryPath.toString()));
    Mockito.verify(mUfs, Mockito.times(1))
        .getStatus(eq(nestedDirectoryPath.toString()), any(GetStatusOptions.class));

    // Get the file info of the directory /dir1
    // listStatus is called on UFS /dir1/dir2
    // Make sure there is neither list nor get on UFS /dir1/dir2
    mFileSystemMaster.getFileInfo(dir1, GetStatusContext.mergeFrom(
        GetStatusPOptions.newBuilder().setCommonOptions(
            FileSystemMasterCommonPOptions.newBuilder().setSyncIntervalMs(0).build())));
    Mockito.verify(mUfs, Mockito.times(0))
        .listStatus(eq(dir1Path.toString()));
    Mockito.verify(mUfs, Mockito.times(1))
        .listStatus(eq(nestedDirectoryPath.toString()));
    Mockito.verify(mUfs, Mockito.times(1))
        .getStatus(eq(nestedDirectoryPath.toString()), any(GetStatusOptions.class));
  }

  private static class SyncAwareFileSystemMaster extends DefaultFileSystemMaster {
    AtomicBoolean mSynced = new AtomicBoolean(false);

    public SyncAwareFileSystemMaster(BlockMaster blockMaster, CoreMasterContext masterContext,
        ExecutorServiceFactory executorServiceFactory) {
      super(blockMaster, masterContext, executorServiceFactory, Clock.systemUTC());
    }

    @Override
    InodeSyncStream.SyncStatus syncMetadata(RpcContext rpcContext, AlluxioURI path,
        FileSystemMasterCommonPOptions options, DescendantType syncDescendantType,
        @Nullable FileSystemMasterAuditContext auditContext,
        @Nullable Function<LockedInodePath, Inode> auditContextSrcInodeFunc)
        throws AccessControlException, InvalidPathException {
      mSynced.set(true);
      return super.syncMetadata(rpcContext, path, options, syncDescendantType, auditContext,
          auditContextSrcInodeFunc);
    }

    void setSynced(boolean synced) {
      mSynced.set(synced);
    }
  }

  private class SyncAwareFileSystemMasterFactory implements MasterFactory<CoreMasterContext> {
    @Override
    public boolean isEnabled() {
      return true;
    }

    @Override
    public String getName() {
      return "SyncAwareFileSystemMasterFactory";
    }

    @Override
    public FileSystemMaster create(MasterRegistry registry, CoreMasterContext context) {
      BlockMaster blockMaster = registry.get(BlockMaster.class);
      FileSystemMaster fileSystemMaster = new SyncAwareFileSystemMaster(blockMaster, context,
          ExecutorServiceFactories.constantExecutorServiceFactory(mExecutorService));
      registry.add(FileSystemMaster.class, fileSystemMaster);
      return fileSystemMaster;
    }
  }

  private void startServices() throws Exception {
    mExecutorService = Executors.newFixedThreadPool(4,
        ThreadFactoryUtils.build("DefaultFileSystemMasterTest-%d", true));
    mRegistry = new MasterRegistry();
    JournalSystem journalSystem =
        JournalTestUtils.createJournalSystem(mJournalFolder.getAbsolutePath());
    CoreMasterContext context = MasterTestUtils.testMasterContext(journalSystem);
    new MetricsMasterFactory().create(mRegistry, context);
    new BlockMasterFactory().create(mRegistry, context);
    mFileSystemMaster = new SyncAwareFileSystemMasterFactory().create(mRegistry, context);
    journalSystem.start();
    journalSystem.gainPrimacy();
    mRegistry.start(true);

    mUfs = Mockito.mock(UnderFileSystem.class);
    PowerMockito.mockStatic(UnderFileSystem.Factory.class);
    Mockito.when(UnderFileSystem.Factory.createWithRecorder(anyString(), any(), any()))
           .thenReturn(mUfs);
  }

  private void stopServices() throws Exception {
    mRegistry.stop();
  }
}
