package com.atlassian.bitbucket.mesh.repository;

import com.atlassian.bitbucket.mesh.git.hook.GitHookService;
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 com.google.common.collect.ImmutableSet;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nonnull;
import java.io.IOException;
import java.nio.file.DirectoryStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.time.Instant;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.function.BiConsumer;
import java.util.function.Consumer;

import static com.atlassian.bitbucket.mesh.util.MoreCollectors.toImmutableSet;
import static java.util.Objects.requireNonNull;
import static java.util.stream.Collectors.toList;

public class ManagedDataStore implements DataStore {

    /**
     * The version of the data store layout in use. This should be updated when <i>material changes</i> are made to how
     * or where data is stored in the data store directory. Database version changes are also considered material
     * changes so a backward-incompatible database upgrade warrants a version bump as well.
     * <p>
     * The version used here follows semver semantics:
     * <ul>
     *     <li>Major version increments indicate <i>backward-incompatible</i> changes have happened, which means
     *     attempts to downgrade directly to older versions should be blocked</li>
     *     <li>Minor version increments indicate <i>backward-compatible but significant</i> changes have happened</li>
     * </ul>
     * Introducing <i>new data</i>, for example, would be a minor version change--downgrading to a previous build
     * would work without issue because the older build would simply not "see" the new data.
     * <p>
     * Moving existing data to a <i>new location</i>, on the other hand, would be a major version change--downgrading
     * to a previous build would result in "losing" data, because the existing data would no longer be stored where
     * the older build expects to find it.
     * <p>
     * <i>The data store layout version is not coupled to the build version or the home layout version.</i> It is
     * tracked independently to allow more fine-grained control over compatibility. Many minor releases, for example,
     * do not <i>materially change</i> the data store layout, meaning downgrading should work without issue, and in
     * (rare) circumstances it may be necessary for a patch release to make a backward-incompatible change to the
     * layout. Decoupling the versions allows the data store layout to change as necessary.
     * <p>
     * As a result, <i>data store layout major version changes may ship in any build version.</i> It is not necessary
     * to wait for a new major build version to ship a new major layout.
     * <table>
     *   <caption> Data store layout versions </caption>
     *   <tr>
     *     <td>Layout version</td> <td>App version</td> <td>Details</td>
     *   </tr>
     *   <tr>
     *     <td>1</td> <td>1.0.0</td> <td>Initial layout version</td>
     *   </tr>
     * </table>
     */
    public static final Version LAYOUT_VERSION = new Version(1);
    public static final String TYPE = "managed";
    private static final Set<String> IGNORED_ENTRIES = ImmutableSet.of(".DS_Store", "lost+found");

    private static final Logger log = LoggerFactory.getLogger(ManagedDataStore.class);

    private final Version buildVersion;
    private final GitHookService hookService;
    private final Object lock;
    private final List<ManagedShard> shards;
    private final ConcurrentMap<String, ManagedShard> shardsByPartition;

    public ManagedDataStore(Version buildVersion, GitHookService hookService, List<Path> shardDirs) {
        this.buildVersion = buildVersion;
        this.hookService = hookService;

        lock = new Object();
        shards = shardDirs.stream()
                .map(ManagedDataStore::createShard)
                .filter(Objects::nonNull)
                .collect(toList());
        shardsByPartition = new ConcurrentHashMap<>();
    }

    public void clearPartitionDeleted(Partition deletedPartition) {
        ManagedShard shard = shardsByPartition.get(deletedPartition.getHex());
        if (shard != null) {
            shard.clearPartitionDeleted(deletedPartition);
        }
    }

    public boolean deletePartitionFiles(Partition partitionToDelete) {
        ManagedShard shard = shardsByPartition.get(partitionToDelete.getHex());
        if (shard == null) {
            return false;
        }
        ManagedShard.DeleteTask task = null;
        synchronized (lock) {
            shard = shardsByPartition.get(partitionToDelete.getHex());
            if (shard != null) {
                task = shard.deletePartitionFiles(partitionToDelete);
            }
            shardsByPartition.remove(partitionToDelete.getHex());
        }
        return task != null && task.execute();
    }

    @Nonnull
    @Override
    public DataStoreInfo getInfo() {
        ImmutableList.Builder<ShardInfo> builder = ImmutableList.builder();
        if (shards.size() == 1) {
            //if there is only one shard configured all partitions will be on that shard, so we don't
            //need to list all partitions and thus save a reasonable amount of bytes on the wire.
            ManagedShard shard = shards.get(0);
            builder.add(new ShardInfo.Builder(shard.rootDir.toString(), getUsableSpace(shard))
                    .build());
        } else {
            shards.forEach(shard -> {
                builder.add(new ShardInfo.Builder(shard.rootDir.toString(), getUsableSpace(shard))
                        .partitions(shard.getPartitions())
                        .build());
            });
        }

        return new DataStoreInfo(DataStoreType.MANAGED, builder.build());
    }

    @Nonnull
    @Override
    public DataStoreType getType() {
        return DataStoreType.MANAGED;
    }

    @Override
    @Nonnull
    public Set<Path> getWorkTreesDirs() {
        return shards.stream().map(ManagedShard::getWorkTreesDir).collect(toImmutableSet());
    }

    public boolean markPartitionDeleted(Partition deletedPartition) {
        ManagedShard shard = shardsByPartition.get(deletedPartition.getHex());
        if (shard != null) {
            return shard.markPartitionDeleted(deletedPartition);
        }
        return false;
    }

    @Override
    public void start() {
        for (ManagedShard shard : shards) {
            shard.start(buildVersion, hookService);

            for (Partition partition : shard.getPartitions()) {
                shardsByPartition.compute(partition.getHex(), (number, existing) -> {
                    if (existing == null) {
                        return shard;
                    }
                    log.warn("Partition {} was found on multiple data stores ({} and {})",
                            number, existing, shard);
                    return existing;
                });
            }
        }
    }

    @Override
    public void stop() {
        shards.forEach(ManagedShard::stop);
    }

    public void streamDeletedPartitions(BiConsumer<Partition, Instant> callback) {
        shards.forEach(shard -> shard.streamDeletedPartitions(callback));
    }

    @Override
    public String toString() {
        return "ManagedDataStore{" + shards.size() + " shards}";
    }

    @Override
    public void writeSupportInfo(@Nonnull JsonGenerator json) throws IOException {
        json.writeStartObject();
        try {
            json.writeStringField("type", "managed");
            json.writeArrayFieldStart("shards");
            try {
                for (ManagedShard shard : shards) {
                    shard.writeSupportInfo(json);
                }
            } finally {
                json.writeEndArray();
            }
        } finally {
            json.writeEndObject();
        }
    }

    @Nonnull
    StorageStatus attachStorage(@Nonnull DefaultManagedRepository repository) {
        String partition = requireNonNull(repository, "repository").getPartition();

        ManagedShard shard = shardsByPartition.get(partition);
        if (shard == null) {
            //The partition hasn't been mapped to a data store yet, let's do that now
            synchronized (lock) {
                //Double checked locking
                shard = shardsByPartition.get(partition);
                if (shard == null) {
                    //Pick the data store with the most free space
                    shard = findLargestShard();
                    if (shard == null) {
                        //Should never happen, but let's be paranoid
                        return StorageStatus.NO_STORE;
                    }

                    shard.addPartition(new Partition(partition));
                    shardsByPartition.put(partition, shard);
                }
            }
        }

        return shard.attachStorage(repository);
    }

    @Nonnull
    Iterable<String> getAllPartitions() {
        return () -> shardsByPartition.keySet().iterator();
    }

    void streamRepositories(@Nonnull String partition, @Nonnull String hierarchyId,
                            @Nonnull Consumer<Repository> callback) {
        ManagedShard shard = shardsByPartition.get(requireNonNull(partition, "partition"));
        if (shard != null) {
            shard.streamRepositories(partition, hierarchyId, callback);
        }
    }

    private static ManagedShard createShard(Path path) {
        if (isNewShard(requireNonNull(path, "path"))) {
            return new ManagedShard(path);
        }

        //It's an existing data store directory. Read store.properties to determine the type and version of the
        //data store.
        try {
            DataStoreProperties properties = DataStoreProperties.load(path);
            if (properties == null) {
                log.info("Not configuring data store {} because its type cannot be determined", path);
            } else if (TYPE.equals(properties.getType())) {
                return new ManagedShard(path);
            } else {
                log.warn("Data store of type '{}' is not supported", properties.getType());
            }
        } catch (IOException e) {
            log.warn("Not configuring data store {} because its properties could not be read", path, e);
        }

        return null;
    }

    private static boolean isNewShard(Path path) {
        if (Files.isDirectory(path)) {
            //Directory exists, verify that it's empty (ignoring some entries like lost+found)
            try (DirectoryStream<Path> directory = Files.newDirectoryStream(path)) {
                for (Path entry : directory) {
                    if (!IGNORED_ENTRIES.contains(entry.getFileName().toString())) {
                        return false;
                    }
                }
                return true;
            } catch (IOException e) {
                log.warn("Cannot read data store directory {}. ", path, e);
            }
        }

        //It's only a new shard if the path does not exist
        return !Files.exists(path);
    }

    private ManagedShard findLargestShard() {
        ManagedShard largest = null;

        long largestSpace = 0L;
        for (ManagedShard shard : shards) {
            try {
                long usableSpace = shard.getUsableSpace();
                if (usableSpace > largestSpace) {
                    largest = shard;
                    largestSpace = usableSpace;
                }
            } catch (IOException e) {
                log.warn("Could not determine the usable space for shard at {}; no partitions will be assigned",
                        shard.getRootDir(), e);
            }
        }

        return largest;
    }

    private long getUsableSpace(ManagedShard shard) {
        try {
            return shard.getUsableSpace();
        } catch (IOException e) {
            log.warn("Error while finding information about shard {}", shard, e);
            return -1L;
        }
    }
}
