package com.atlassian.bitbucket.mesh.repository;

import com.atlassian.bitbucket.mesh.Product;
import com.atlassian.bitbucket.mesh.app.DirectoryLock;
import com.atlassian.bitbucket.mesh.app.VersionUpdate;
import com.atlassian.bitbucket.mesh.git.hook.GitHookService;
import com.atlassian.bitbucket.mesh.io.MoreFiles;
import com.atlassian.bitbucket.mesh.partition.Partition;
import com.atlassian.bitbucket.mesh.util.Version;
import com.fasterxml.jackson.core.JsonGenerator;
import com.google.common.collect.ImmutableList;

import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.nio.channels.OverlappingFileLockException;
import java.nio.file.*;
import java.nio.file.attribute.BasicFileAttributes;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.regex.Pattern;
import java.util.stream.Stream;

import static com.atlassian.bitbucket.mesh.io.MoreFiles.pruneStaleFiles;
import static com.atlassian.bitbucket.mesh.repository.DefaultSharedObjectsRepository.SHARED_OBJECTS;
import static com.atlassian.bitbucket.mesh.repository.ManagedDataStore.LAYOUT_VERSION;
import static com.atlassian.bitbucket.mesh.repository.ManagedDataStore.TYPE;
import static com.atlassian.bitbucket.mesh.util.ResourceUtils.unpackTar;
import static com.google.common.base.Preconditions.checkArgument;
import static java.lang.String.format;
import static java.nio.file.StandardCopyOption.ATOMIC_MOVE;
import static java.util.Collections.binarySearch;
import static java.util.Collections.unmodifiableList;
import static java.util.Objects.requireNonNull;
import static java.util.UUID.randomUUID;

/**
 * A shard of a {@link ManagedDataStore}. Shards allow the managed store to span multiple mounts.
 */
class ManagedShard extends AbstractShard {

    private static final Pattern PATTERN_PARTITION_DIR = Pattern.compile("[0-9a-f]{4,15}");

    private final DirectoryLock directoryLock;
    private final Path hooksDir;
    private final List<Partition> partitions;
    private final Path templateDir;
    private final Path workTreesDir;
    private volatile ShardState state;

    ManagedShard(@Nonnull Path rootDir) {
        super(rootDir);

        directoryLock = new DirectoryLock(rootDir, false);
        partitions = new CopyOnWriteArrayList<>();
        state = ShardState.STARTING;

        Path defaultConfigDir = rootDir.resolve("config").resolve("default");
        hooksDir = defaultConfigDir.resolve("hooks");
        templateDir = defaultConfigDir.resolve("templates");
        workTreesDir = MoreFiles.mkdir(rootDir, "work-trees");
    }

    public void clearPartitionDeleted(Partition deletedPartition) {
        if (!state.isAccessible()) {
            log.error("Partition {}: Data store is not available at path: {} ({})", deletedPartition, rootDir, state);
            throw new DataStoreUnavailableException("Data store is not available");
        }
        try {
            Files.deleteIfExists(getPartitionDeletedMarker(deletedPartition.getHex()));
        } catch (IOException e) {
            log.warn("Failed to clear deletion marker for partition {}", deletedPartition);
        }
    }

    /**
     * Performs an atomic move of the partition directory to a temporary directory to immediately hide the directory.
     * Then return a task that can then be used to clean up the temporary directory.
     *
     * @param partitionToDelete the partition to delete
     * @return a task that can be run to clean up the temporary files
     */
    public DeleteTask deletePartitionFiles(Partition partitionToDelete) {
        Path partitionDir = getPartitionDir(partitionToDelete.getHex());
        //To handle the case when we add a partition again after deleting it, then delete it again while we're still
        //deleting the first one, we give the temporary directory a unique name
        Path temporaryDeletingDirectory =
                partitionDir.resolveSibling(partitionDir.getFileName() + ".deleting-" + randomUUID());

        //First remove the partition from the list, then rename the directory while holding the lock. This ensures
        //that if a partition of the same name is added again immediately after, it can be added while we delete
        //it.
        synchronized (partitions) {
            boolean partitionRemoved = partitions.remove(partitionToDelete);
            if (!partitionRemoved) {
                return () -> false;
            }
            try {
                try {
                    Files.move(partitionDir, temporaryDeletingDirectory, ATOMIC_MOVE);
                } catch (AtomicMoveNotSupportedException e) {
                    //noinspection PlaceholderCountMatchesArgumentCount
                    log.warn("Atomic move not supported when deleting {}, falling back to regular move", partitionToDelete,
                            log.isDebugEnabled() ? e : null);
                    Files.move(partitionDir, temporaryDeletingDirectory);
                }
            } catch (IOException e) {
                addPartition(partitionToDelete);
                throw new UncheckedIOException(
                        format("Unable to delete partition %s, aborting", partitionToDelete), e);
            }
        }

        //Return a task that can be run later to actually delete the files
        return () -> {
            //Now actually delete the files
            try {
                MoreFiles.deleteRecursively(temporaryDeletingDirectory);
            } catch (IOException e) {
                log.error("Unable to cleanup files for partition {}. The directory could not be deleted: {}",
                        partitionToDelete, temporaryDeletingDirectory);
            }

            //The last file remaining is the deletion marker, delete that too if it exists
            try {
                Files.deleteIfExists(getPartitionDeletedMarker(partitionToDelete.getHex()));
            } catch (IOException e) {
                log.warn("Unable to cleanup marker file for partition {}", partitionToDelete);
            }

            return true;
        };
    }

    @Nullable
    public Instant getPartitionDeletedTime(Partition partition) {
        if (!state.isAccessible()) {
            log.error("Partition {}: Data store is not available at path: {} ({})", partition, rootDir, state);
            throw new DataStoreUnavailableException("Data store is not available");
        }
        try {
            Path marker = getPartitionDeletedMarker(partition.getHex());
            if (Files.exists(marker)) {
                try (Stream<String> lines = Files.lines(marker)) {
                    return lines
                            .limit(1)
                            .findAny()
                            .map(Long::parseLong)
                            .map(Instant::ofEpochSecond)
                            .orElse(null);
                }
            }
        } catch (NoSuchFileException | FileNotFoundException e) {
            //Intentional fall through
        } catch (NumberFormatException e) {
            log.warn("Failed to parse partition marker file {}. Clearing marker", partition,
                    log.isDebugEnabled() ? e : null);
            clearPartitionDeleted(partition);
            //Intentional fall through
        } catch (IOException e) {
            log.warn("Failed to read partition marker file {}", partition, e);
        }
        return null;
    }

    @Nonnull
    public Path getWorkTreesDir() {
        return workTreesDir;
    }

    public boolean markPartitionDeleted(Partition deletedPartition) {
        if (!state.isAccessible()) {
            log.error("{}: Data store is not available at path: {} ({})", deletedPartition, rootDir, state);
            throw new DataStoreUnavailableException("Data store is not available");
        }
        try {
            Files.write(getPartitionDeletedMarker(deletedPartition.getHex()),
                    ImmutableList.of(String.valueOf(Instant.now().getEpochSecond())),
                    StandardOpenOption.CREATE_NEW);
            return true;
        } catch (FileAlreadyExistsException e) {
            log.debug("{} was already marked as deleted", deletedPartition);
            return true;
        } catch (IOException e) {
            log.warn("Failed to mark {} as deleted", deletedPartition);
        }
        return false;
    }

    public void streamDeletedPartitions(BiConsumer<Partition, Instant> callback) {
        partitions.forEach(partition -> {
            Instant time = getPartitionDeletedTime(partition);
            if (time != null) {
                callback.accept(partition, time);
            }
        });
    }

    @Override
    public String toString() {
        return "ManagedShard{root=" + rootDir + '}';
    }

    void addPartition(@Nonnull Partition partition) {
        requireNonNull(partition, "partition");

        synchronized (partitions) {
            int index = binarySearch(partitions, partition);
            if (index < 0) {
                //index < 0 means that the partition isn't in the list, and binarySearch returns (-(insertion point) - 1)
                index = -index - 1;
                partitions.add(index, partition);
            }

            //Make sure the partition directory is created on disk to ensure that next time we start up, we detect that
            //the partition has been mapped to this shard
            MoreFiles.mkdir(getPartitionDir(partition.getHex()));
        }
    }

    @Nonnull
    StorageStatus attachStorage(@Nonnull DefaultManagedRepository repository) {
        if (state != ShardState.AVAILABLE) {
            log.error("{}: Data store is not available at path: {} ({})", repository.getId(), rootDir, state);
            throw new DataStoreUnavailableException("Data store is not available");
        }

        repository.attachTo(this);

        try {
            BasicFileAttributes attributes = Files.readAttributes(repository.getPath(), BasicFileAttributes.class);
            if (attributes.isDirectory()) {
                //If the directory exists, check whether the repository has been marked as deleted and
                //return the status accordingly
                return Files.exists(repository.getDeletedMarker()) ? StorageStatus.DELETED : StorageStatus.ATTACHED;
            }

            //If there's something other than a directory at the target path, the repository is occluded
            return StorageStatus.OCCLUDED;
        } catch (FileNotFoundException | NoSuchFileException e) {
            return StorageStatus.NONEXISTENT;
        } catch (IOException e) {
            //If we can't check the path, return UNKNOWN status to allow the caller to decide whether to
            //proceed or preemptively fail
            log.warn("{}: Failed to check repository path: {}", repository.getId(), repository.getPath());
            return StorageStatus.UNKNOWN;
        }
    }

    @Nonnull
    @Override
    Path getHooksDir() {
        return hooksDir;
    }

    @Nonnull
    Path getLfsDir(@Nonnull String partition, @Nonnull String hierarchyId) {
        return getHierarchyDir(partition, hierarchyId).resolve("lfs");
    }

    /**
     * @return the list of partitions currently located on this shard
     */
    @Nonnull
    List<Partition> getPartitions() {
        return unmodifiableList(partitions);
    }

    @Nonnull
    Path getRepositoriesDir(String partition, String hierarchy) {
        return getHierarchyDir(partition, hierarchy).resolve("repos");
    }

    @Nullable
    DefaultManagedRepository getRepository(@Nonnull String partition, @Nonnull String hierarchyId,
                                           @Nonnull String localId) {
        DefaultManagedRepository repository = new DefaultManagedRepository(partition, hierarchyId, localId);
        if (attachStorage(repository) == StorageStatus.ATTACHED) {
            return repository;
        }

        return null;
    }

    @Nonnull
    DefaultSharedObjectsRepository getSharedObjects(@Nonnull String partition, @Nonnull String hierarchyId) {
        return new DefaultSharedObjectsRepository(this, partition, hierarchyId);
    }

    /**
     * @return the directory that contains the templates for the {@code shared-objects} repository, which is used to
     *         share objects between all repositories in the hierarchy.
     */
    @Nonnull
    Path getSharedObjectsTemplateDir() {
        return rootDir.resolve("config/shared-objects/templates");
    }

    @Nonnull
    @Override
    ShardState getState() {
        return state;
    }

    /**
     * @return the directory that contains the templates for a default repository, which shares objects with other
     *         repositories in the hierarchy through an alternates to the {@code shared-objects} repository.
     */
    @Nonnull
    @Override
    Path getTemplateDir() {
        return templateDir;
    }

    void start(@Nonnull Version buildVersion, @Nonnull GitHookService hookService) {
        MoreFiles.mkdir(rootDir);

        try {
            if (directoryLock.lock()) {
                log.info("Successfully acquired lock on shard at {}", rootDir);

                Instant localTime = Instant.now();
                Instant start;
                try {
                    start = MoreFiles.getFileSystemTime(rootDir);
                    if (Math.abs(start.toEpochMilli() - localTime.toEpochMilli()) > 10_000L) {
                        log.warn("The file server clock for {} diverges from the local clock by more than " +
                                "10s ({} vs {})!", rootDir, start, localTime);
                    }
                } catch (IOException e) {
                    log.warn("Could not determine file system time for {}. Using local clock instead", rootDir, e);
                    start = localTime;
                }
                unpackTar("/templates/managed-datastore.tar", rootDir);
                hookService.installHooks(hooksDir);

                //Prune hook scripts and template files left behind by previous versions
                pruneStaleConfig(start);
            }
        } catch (OverlappingFileLockException | IOException e) {
            log.warn("Failed to lock or initialize shard at {}. Some repositories will not be available!", rootDir, e);
        }

        state = validateLayout(buildVersion);
        initPartitions();
    }

    void stop() {
        state = ShardState.STOPPED;

        log.info("Releasing lock on data store {}", rootDir);
        directoryLock.close();
    }

    void streamRepositories(@Nonnull String partition, @Nonnull String hierarchyId,
                            @Nonnull Consumer<Repository> callback) {
        try {
            Path reposDir = getRepositoriesDir(partition, hierarchyId);
            Files.walkFileTree(reposDir, Collections.emptySet(), 1, new SimpleFileVisitor<Path>() {

                @Override
                public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) {
                    if (!attrs.isDirectory()) {
                        return FileVisitResult.CONTINUE;
                    }

                    String id = file.getFileName().toString();
                    if (!SHARED_OBJECTS.equals(id)) {
                        DefaultManagedRepository repository = new DefaultManagedRepository(partition, hierarchyId, id);
                        if (attachStorage(repository) == StorageStatus.ATTACHED) {
                            callback.accept(repository);
                        }
                    }
                    return FileVisitResult.CONTINUE;
                }
            });
        } catch (FileNotFoundException | NoSuchFileException e) {
            //Hierarchy directory does not exist, nothing to do
        } catch (IOException e) {
            throw new UncheckedIOException(
                    format("Could not stream repositories for hierarchy %s/%s", partition, hierarchyId), e);
        }
    }

    @Override
    void writeExtraSupportInfo(JsonGenerator json) throws IOException {
        json.writeArrayFieldStart("partitions");
        try {
            for (Partition partition : partitions) {
                json.writeString(partition.getHex());
            }
        } finally {
            json.writeEndArray();
        }
        json.writeStringField("state", state.name());
        try {
            DataStoreProperties properties = DataStoreProperties.load(rootDir);
            if (properties != null) {
                json.writeStringField("build-version", properties.getBuild().toString());
                json.writeStringField("layout-version", properties.getLayout().toString());
            }
        } catch (IOException e) {
            log.warn("Could not load data store properties", e);
        }
    }

    private Path getHierarchyDir(String partition, String hierarchyId) {
        requireNonNull(partition, "partition");
        requireNonNull(hierarchyId, "hierarchyId");
        checkArgument(hierarchyId.length() >= 20, "hierarchy ID must be at least 20 characters");

        return getPartitionDir(partition).resolve(hierarchyId);
    }

    private Path getPartitionDeletedMarker(String partition) {
        Path partitionDir = getPartitionDir(partition);
        return partitionDir.resolveSibling(partitionDir.getFileName() + ".deleted");
    }

    private Path getPartitionDir(String partition) {
        return rootDir.resolve("data").resolve(partition);
    }

    /**
     * @return the directory that contains the hooks for the {@code shared-objects} repository, which is used to
     *         share objects between all repositories in the hierarchy.
     */
    @Nonnull
    private Path getSharedObjectsHooksDir() {
        return rootDir.resolve("config/shared-objects/hooks");
    }

    private void initPartitions() {
        try (Stream<Path> stream = Files.list(rootDir.resolve("data"))) {
            stream.forEach(path -> {
                String filename = path.getFileName().toString();
                if (PATTERN_PARTITION_DIR.matcher(filename).matches() && Files.isDirectory(path)) {
                    addPartition(new Partition(filename));
                }
            });
        } catch (FileNotFoundException | NoSuchFileException e) {
            log.debug("{} contains no partitions", rootDir);
        } catch (IOException e) {
            log.warn("Error while scanning current partitions from {}", rootDir, e);
        }
    }

    private void pruneStaleConfig(Instant staleTimestamp) {
        List<Path> dirsToPrune = ImmutableList.of(
                getHooksDir(), getSharedObjectsHooksDir(),
                getTemplateDir(), getSharedObjectsTemplateDir());
        for (Path path : dirsToPrune) {
            try {
                List<Path> deleted = new ArrayList<>();
                pruneStaleFiles(path, staleTimestamp, false,
                        (deletePath, attributes) -> deleted.add(deletePath));
                if (!deleted.isEmpty()) {
                    log.info("Deleted stale configuration files: {}", deleted);
                }
            } catch (IOException e) {
                log.warn("Failed to prune stale files from {}", path, e);
            }
        }
    }

    private ShardState validateLayout(Version buildVersion) {
        try {
            DataStoreProperties properties = DataStoreProperties.load(rootDir);
            if (properties == null) {
                // Write the properties to disk
                new DataStoreProperties(buildVersion, LAYOUT_VERSION, TYPE)
                        .store(rootDir);
                return ShardState.AVAILABLE;
            }

            // Existing data store, check compatibility
            if (!TYPE.equals(properties.getType())) {
                throw new IllegalStateException(
                        rootDir + " is a " + properties.getType() + " shard, not a " + TYPE + " shard");
            }

            switch (VersionUpdate.of(LAYOUT_VERSION, properties.getLayout())) {
                case MAJOR_DOWNGRADE:
                    log.warn("Downgrading {} from {} to {} requires restoring from a backup taken prior to " +
                                    "the upgrade. The data store directory layout has been updated from {} to {} " +
                                    "and is no longer compatible",
                            Product.NAME, properties.getBuild(), buildVersion, LAYOUT_VERSION, properties.getLayout());
                    return ShardState.INCOMPATIBLE;
                case MAJOR_UPGRADE:
                    log.info("Data store {} needs to be upgraded from {} to {} and will not be available " +
                            "until the upgrade has completed. It will not be possible to downgrade to an older " +
                            "version without restoring from a backup", rootDir, properties.getLayout(), LAYOUT_VERSION);
                    return ShardState.AVAILABLE;
                case MINOR_UPGRADE:
                    log.info("Data store {} needs to be upgraded  from {} to {} but will be available " +
                            "while the upgrade is performed.", rootDir, properties.getLayout(), LAYOUT_VERSION);
                    return ShardState.AVAILABLE;
                case MINOR_DOWNGRADE:
                    //The new layout is different from _but compatible with_ the layout used by the
                    //last build to use the home directory
                    log.warn("Downgrading {} from {} to {} will not roll back the home directory layout from {} to {}",
                            Product.NAME, properties.getBuild(), buildVersion, properties.getLayout(), LAYOUT_VERSION);
                    return ShardState.AVAILABLE;
                default:
                    if (!buildVersion.equals(properties.getBuild())) {
                        new DataStoreProperties(buildVersion, properties.getLayout(), properties.getType())
                                .store(rootDir);
                    }
                    return ShardState.AVAILABLE;
            }
        } catch (IOException e) {
            //FileNotFoundException and NoSuchFileException are handled elsewhere, so this IOException indicates
            //some other failure, like a permissions issue, has happened
            log.warn("The data store directory cannot be validated for use by {} {}", Product.NAME, buildVersion, e);
            return ShardState.INCOMPATIBLE;
        }
    }

    @FunctionalInterface
    public interface DeleteTask {

        boolean execute();
    }
}
