package com.atlassian.bitbucket.mesh.repository;

import com.atlassian.bitbucket.mesh.Product;
import com.atlassian.bitbucket.mesh.annotation.NotProfiled;
import com.atlassian.bitbucket.mesh.annotation.Profiled;
import com.atlassian.bitbucket.mesh.event.EventPublisher;
import com.atlassian.bitbucket.mesh.git.GitAgent;
import com.atlassian.bitbucket.mesh.git.GitRepositoryLayout;
import com.atlassian.bitbucket.mesh.git.GitTimers;
import com.atlassian.bitbucket.mesh.git.builder.GitProcessBuilderFactory;
import com.atlassian.bitbucket.mesh.git.exception.NoDefaultBranchException;
import com.atlassian.bitbucket.mesh.git.exception.SetDefaultBranchFailedException;
import com.atlassian.bitbucket.mesh.git.exception.SetMetadataFailedException;
import com.atlassian.bitbucket.mesh.git.ref.Branch;
import com.atlassian.bitbucket.mesh.io.MoreFiles;
import com.atlassian.bitbucket.mesh.io.UnderLock;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcSetDefaultBranchRequest;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcSetMetadataRequest;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcTransaction;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcTransactionCommitFailed;
import com.atlassian.bitbucket.mesh.transaction.AbstractPreparedAction;
import com.atlassian.bitbucket.mesh.transaction.TransactionCommitFailedException;
import com.atlassian.bitbucket.mesh.transaction.TransactionManager;
import com.atlassian.util.profiling.Ticker;
import com.google.common.base.Splitter;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.protobuf.ByteString;
import io.grpc.Context;
import io.grpc.Deadline;
import org.apache.commons.lang3.mutable.MutableInt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nonnull;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.Writer;
import java.nio.file.Files;
import java.nio.file.NoSuchFileException;
import java.nio.file.Path;
import java.time.Duration;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.stream.Stream;

import static com.atlassian.bitbucket.mesh.git.GitRefPattern.HEADS;
import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.Objects.requireNonNull;

@Profiled
public class DefaultRepositoryManager implements RepositoryManager {

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

    private static final String BANNER =
            "#>***********************************************\n" +
            "# THIS FILE IS MAINTAINED BY " + Product.FULL_NAME.toUpperCase(Locale.US) + "\n" +
            "# IT CONTAINS NO USER-SERVICEABLE PARTS.\n" +
            "#>***********************************************\n";
    private static final String PATH_REPOSITORY_CONFIG = "repository-config";

    private final GitAgent agent;
    private final RepositoryAttributeManager attributeManager;
    private final GitProcessBuilderFactory builderFactory;
    private final DataStoreManager dataStoreManager;
    private final EventPublisher eventPublisher;
    private final ScheduledExecutorService executorService;
    private final List<Function<String, Repository>> idParsers;
    private final ReplicaStateRegistry replicaStateRegistry;
    private final RepositoryAccessListener repositoryAccessListener;
    private final RepositoryHashService repositoryHashService;
    private final GitRepositoryLayout repositoryLayout;
    private final TransactionManager transactionManager;
    private final WorkTreeManager workTreeManager;

    public DefaultRepositoryManager(GitAgent agent, RepositoryAttributeManager attributeManager,
                                    GitProcessBuilderFactory builderFactory, DataStoreManager dataStoreManager,
                                    EventPublisher eventPublisher, ScheduledExecutorService executorService,
                                    ReplicaStateRegistry replicaStateRegistry,
                                    RepositoryAccessListener repositoryAccessListener,
                                    RepositoryHashService repositoryHashService, GitRepositoryLayout repositoryLayout,
                                    TransactionManager transactionManager, WorkTreeManager workTreeManager) {
        this.agent = agent;
        this.attributeManager = attributeManager;
        this.builderFactory = builderFactory;
        this.dataStoreManager = dataStoreManager;
        this.eventPublisher = eventPublisher;
        this.executorService = executorService;
        this.replicaStateRegistry = replicaStateRegistry;
        this.repositoryAccessListener = repositoryAccessListener;
        this.repositoryHashService = repositoryHashService;
        this.repositoryLayout = repositoryLayout;
        this.transactionManager = transactionManager;
        this.workTreeManager = workTreeManager;

        idParsers = ImmutableList.of(
                DefaultBitbucketServerRepository::parseId,
                DefaultManagedRepository::parseId,
                DefaultPathRepository::parseId);
    }

    @Nonnull
    @Override
    public Repository create(@Nonnull CreateRepositoryParameters parameters) {
        requireNonNull(parameters, "parameters");

        String repositoryId = parameters.getId();
        Repository repository = parseId(repositoryId);
        if (repository == null) {
            log.error("Cannot create repository {}; the ID is invalid", repositoryId);
            throw new CreateRepositoryFailedException(repositoryId, "Invalid repository ID");
        }

        StorageStatus storageStatus = dataStoreManager.attachStorage(repository);
        if (storageStatus == StorageStatus.NO_STORE) {
            log.error("Cannot create repository {}; no data store is available", repositoryId);
            throw new CreateRepositoryFailedException(repositoryId, "No DataStore is available");
        }

        Path repositoryDir = repository.getPath();
        if (storageStatus == StorageStatus.ATTACHED || storageStatus == StorageStatus.DELETED) {
            //Deleted is treated the same, here, because either way the repository exists
            log.error("Cannot create repository {}; the repository already exists (status = {})",
                    repositoryId, storageStatus);
            throw new RepositoryAlreadyExistsException(repositoryId, repositoryDir,
                    "Cannot create repository " + repositoryId + " at " + repositoryDir +
                            "; a directory exists at the same location");
        }
        if (storageStatus == StorageStatus.OCCLUDED) {
            log.error("Cannot create repository {}; a file exists at the same location", repositoryId);
            throw new RepositoryAlreadyExistsException(repositoryId, repositoryDir,
                    "Cannot create repository " + repositoryId + " at " + repositoryDir +
                            "; a file exists at the same location");
        }

        //If we make it here, the StorageStatus is either NONEXISTENT or UNKNOWN. For UNKNOWN we
        //allow creation to proceed because it may still succeed
        repositoryAccessListener.onRepositoryAccessed(repository);
        return transactionManager.execute(tx -> {
            boolean delete = false;
            try (Ticker ignored = GitTimers.CREATE_REPO.start(repositoryId)) {
                //Create the bare repository. The create logic is different for 'classic' Bitbucket Server repositories and
                //Mesh managed repositories, so use the visitor pattern to make repository creation dependent on the
                //data store in which the repository is created.
                log.debug("Creating repository for {}", repositoryId);
                repository.accept(new CreateRepositoryOperation(
                        builderFactory, eventPublisher, parameters, repositoryHashService, tx));
                delete = true;

                //A future version of Git will support specifying the default branch via configuration, but
                //with all currently released/supported versions we need to update manually
                if (parameters.getDefaultBranch() == null) {
                    log.debug("{}: No default branch was specified; will accept Git's default", repository);
                } else {
                    String qualified = HEADS.qualify(parameters.getDefaultBranch());
                    log.debug("{}: Configuring {} as the default branch", repository, qualified);

                    agent.setHead(repository.getPath(), qualified);
                }

                //Write repository-config, which contains any client-provided metadata. If the client didn't provide
                //any metadata, the file will be created and left empty
                log.debug("Writing metadata for {}", repository);
                writeRepositoryConfig(repository, parameters.getMetadata());

                attachAttributes(repository);
                delete = false; //Creation is successful, so don't delete

                log.info("Repository {} has been created and configured successfully", repository);
                return repository;
            } catch (RuntimeException e) {
                log.error("Failed to create repository {}", repositoryId, e);
                throw e;
            } finally {
                if (delete) {
                    try {
                        log.debug("Cleaning up partially-created repository for {}", repositoryId);

                        MoreFiles.deleteRecursively(repository.getPath());
                    } catch (Exception e) {
                        log.error("Could not delete partially-created repository {}", repositoryLayout, e);
                    }
                }
            }
        });
    }

    @Override
    public void delete(@Nonnull DeleteRepositoryParameters parameters) {
        Repository repository = parameters.getRepository();
        try (Ticker ignored = GitTimers.DELETE_REPO.start(repository.getId())) {
            transactionManager.executeWithoutResult(tx -> repository.accept(
                    new DeleteRepositoryOperation(builderFactory, eventPublisher, executorService, parameters, tx)));
        } catch (RuntimeException e) {
            log.error("Failed to delete repository {}", repository.getId(), e);
            throw e;
        }
    }

    @NotProfiled
    @Override
    public Repository findById(@Nonnull String id) {
        return findById(id, false);
    }

    @NotProfiled
    @Override
    public Repository findById(@Nonnull String id, boolean returnIfNotExists) {
        requireNonNull(id, "id");

        Repository repository = parseId(id);
        if (repository == null) {
            //If the repository ID is not understood, check to see if it's a work tree. If it's not,
            //this will return null and the caller can handle it
            return workTreeManager.findById(id);
        }

        StorageStatus storageStatus = dataStoreManager.attachStorage(repository);
        if (storageStatus == StorageStatus.ATTACHED) {
            attachAttributes(repository);

            repositoryAccessListener.onRepositoryAccessed(repository);
            return repository;
        }

        //If storage cannot be attached, meaning the repository does not exist on disk or is pending
        //deletion, treat it as not found
        if (returnIfNotExists) {
            repositoryAccessListener.onRepositoryAccessed(repository);
            return repository;
        }
        return null;
    }

    @Nonnull
    @NotProfiled
    @Override
    public Repository getById(@Nonnull String id) {
        Repository repository = findById(id);
        if (repository == null) {
            throw new RepositoryNotFoundException(id, "Repository with ID '" + id + "' not found");
        }

        return repository;
    }

    @Override
    public String getDefaultBranch(@Nonnull Repository repository) {
        try {
            return agent.getHead(repository);
        } catch (NoDefaultBranchException e) {
            return e.getBranchId();
        } catch (Exception e) {
            log.warn("{}: The default branch could not be determined", repository, e);
            return null;
        }
    }

    @Nonnull
    @Override
    public Map<String, String> getMetadata(@Nonnull Repository repository) {
        ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
        try (Stream<String> lines = Files.lines(repository.getPath().resolve(PATH_REPOSITORY_CONFIG), UTF_8)) {
            Splitter splitter = Splitter.on('=').trimResults();
            lines.forEach(line -> {
                List<String> parts = splitter.splitToList(line);
                if (parts.size() == 2) {
                    builder.put(parts.get(0), parts.get(1));
                }
            });
        } catch (FileNotFoundException | NoSuchFileException e) {
            //This may happen for PathRepository or a repository that was created without any metadata
            log.trace("[{}] Repository metadata not found", repository);
        } catch (IOException e) {
            log.warn("[{}] Failed to read repository metadata", repository, e);
        }
        return builder.build();
    }

    @Nonnull
    @Override
    public RepositoryReplicaDetails getReplicaDetails(@Nonnull String id) {
        Repository repository = findById(requireNonNull(id, "id"));
        if (repository == null) {
            return new RepositoryReplicaDetails.Builder(ReplicaState.MISSING)
                    .build();
        }

        return new RepositoryReplicaDetails.Builder(replicaStateRegistry.getDetails(repository))
                .contentHash(repositoryHashService.getContentHash(repository))
                .metadataHash(repositoryHashService.getMetadataHash(repository))
                .build();
    }

    @Nonnull
    @Override
    public Map<String, Long> getVersions(@Nonnull GetVersionsParameters parameters) {
        return attributeManager.getVersionsByPartitions(requireNonNull(parameters, "parameters").getPartitions());
    }

    @Override
    public int resetHierarchy(@Nonnull String partition, @Nonnull String hierarchyId) {
        requireNonNull(partition, "partition");
        requireNonNull(hierarchyId, "hierarchyId");

        MutableInt count = new MutableInt();
        ResetRepositoryOperation resetOp = new ResetRepositoryOperation(replicaStateRegistry, attributeManager);
        try {
            dataStoreManager.streamRepositories(partition, hierarchyId, repository -> {
                attachAttributes(repository);
                repository.accept(resetOp);
                count.increment();
            });
        } catch (RuntimeException e) {
            log.error("Failed to reset hierarchy {} on partition {}", hierarchyId, partition, e);
            throw new ResetHierarchyFailedException(hierarchyId, e);
        }
        return count.intValue();
    }

    @Override
    public void setDefaultBranch(@Nonnull Repository repository, @Nonnull String refName) {
        requireNonNull(repository, "repository");
        requireNonNull(refName, "refName");

        transactionManager.executeWithoutResult(tx -> {
            PreparedSetDefaultBranch action = new PreparedSetDefaultBranch(repository, refName);
            action.prepare();
            if (!tx.prepared(action).isSuccess()) {
                throw new SetDefaultBranchFailedException(refName, null);
            }
        });
    }

    @Override
    public void setMetadata(@Nonnull Repository repository, @Nonnull Map<String, String> metadata) {
        requireNonNull(repository, "repository");
        requireNonNull(metadata, "metadata");

        PreparedSetMetadata setMetadata = new PreparedSetMetadata(repository, metadata);
        setMetadata.prepare();

        transactionManager.executeWithoutResult(transaction -> {
            if (!transaction.prepared(setMetadata).isSuccess()) {
                throw new SetMetadataFailedException(null);
            }
        });
    }

    private static void writeRepositoryConfig(@Nonnull Writer writer,
                                              @Nonnull Map<String, String> metadata) throws IOException {
        StringBuilder builder = new StringBuilder()
                .append(BANNER)
                .append("[bitbucket]\n");
        metadata.forEach((key, value) ->
                builder.append('\t').append(key).append(" = ").append(value).append('\n'));

        writer.write(builder.toString());
    }

    private void attachAttributes(Repository repository) {
        if (repository instanceof AbstractRepository) {
            ((AbstractRepository) repository).attachAttributes(attributeManager);
        }
    }

    private Repository parseId(String id) {
        for (Function<String, Repository> parser : idParsers) {
            Repository repository = parser.apply(id);
            if (repository != null) {
                return repository;
            }
        }

        return null;
    }

    /**
     * For performance reasons, {@link Repository} details are stored outside the canonical {@code config} file.
     * This way, any time that data is changed the entire file can be overwritten, rather than forking multiple
     * {@code git config} processes to mutate the canonical {@code config} safely, the entire file can just be
     * overwritten. This is particularly important for storing the project key, where changing the key for a
     * single project could require updating the property in dozens of repositories.
     * <p>
     * The {@code templates/config} used when creating a new repository or fork includes {@code repository-config},
     * so these properties are all available to {@code git config}.
     *
     * @param repository the repository to write the {@code repository-config} for
     */
    private void writeRepositoryConfig(@Nonnull Repository repository, @Nonnull Map<String, String> metadata) {
        try {
            repositoryLayout.editFile(repository, PATH_REPOSITORY_CONFIG,
                    (ignored, writer) -> writeRepositoryConfig(writer, metadata));
        } catch (IOException e) {
            throw new CreateRepositoryFailedException(repository.getId(), "Failed to write repository metadata", e);
        }
    }

    private class PreparedSetDefaultBranch extends AbstractPreparedAction {

        private String refName;

        PreparedSetDefaultBranch(Repository repository, String refName) {
            super(repository);
            this.refName = refName;
        }

        @Override
        public void commit() {
            builderFactory.builder(repository)
                    .symbolicRef()
                    .set(GitAgent.HEAD, refName)
                    .build()
                    .run();

            eventPublisher.publish(new DefaultBranchChangedEvent(repository, refName));
        }

        @Nonnull
        @Override
        public RpcTransaction.Builder populateTransaction(@Nonnull RpcTransaction.Builder builder) {
            return builder.setSetDefaultBranch(RpcSetDefaultBranchRequest.newBuilder()
                    .setRepository(repository.getId())
                    .setBranchId(ByteString.copyFromUtf8(refName)));
        }

        void prepare() {
            //Before we set the branch they've selected, we need to attempt to resolve it. This is the only way to ensure
            //we get the right branch name. Note that we do not assume the branchName provided is absolute, for this call.
            Branch branch = agent.resolveBranch(repository, refName);

            if (branch == null) {
                //If we can't resolve it, then qualify it and hope for the best
                refName = HEADS.qualify(refName);
                log.warn("{}: Updating default branch to {}, which could not be resolved", repository, refName);
            } else {
                //If we did resolve it, then use the absolute name of the resolved branch to ensure the symbolic ref points
                //to the same branch we resolved
                refName = branch.getId();
                log.info("{}: Updating default branch to {}", repository, branch);
            }
        }
    }

    private static class PreparedSetMetadata extends AbstractPreparedAction {

        private final Map<String, String> metadata;
        private UnderLock.LockedFile staged;

        PreparedSetMetadata(Repository repository, Map<String, String> metadata) {
            super(repository);
            this.metadata = metadata;
        }

        private void prepare() {
            Deadline deadline = Context.current().getDeadline();
            Duration timeout = Duration.ofMillis(deadline == null ? 5000L : deadline.timeRemaining(TimeUnit.MILLISECONDS));
            UnderLock underLock = UnderLock.forFile(repository.getPath().resolve(PATH_REPOSITORY_CONFIG), timeout);
            try {
                staged = underLock.lock();
                staged.edit((ignored, writer) -> writeRepositoryConfig(writer, metadata), false);
            } catch (IOException e) {
                throw new SetMetadataFailedException(e);
            }
        }

        @Override
        public void commit() {
            if (staged != null) {
                try {
                    staged.close();
                } catch (IOException e) {
                    throw new TransactionCommitFailedException("Failed to publish metadata changes",
                            RpcTransactionCommitFailed.Reason.PUBLISH_FAILED, e);
                }
            }
        }

        @Override
        public void rollBack() {
            if (staged != null) {
                staged.rollBack();
                try {
                    staged.close();
                } catch (IOException e) {
                    log.warn("[{}] Error releasing lock on repository metadata", repository, e);
                }
            }
        }

        @Nonnull
        @Override
        public RpcTransaction.Builder populateTransaction(@Nonnull RpcTransaction.Builder builder) {
            return builder.setSetMetadata(RpcSetMetadataRequest.newBuilder()
                    .putAllMetadata(metadata)
                    .setRepository(repository.getId()));
        }
    }
}
