package com.atlassian.bitbucket.mesh.repository;

import com.atlassian.bitbucket.mesh.event.EventPublisher;
import com.atlassian.bitbucket.mesh.git.builder.GitProcessBuilderFactory;
import com.atlassian.bitbucket.mesh.git.builder.config.GitConfig;
import com.atlassian.bitbucket.mesh.git.exception.FailedPreconditionException;
import com.atlassian.bitbucket.mesh.gprc.GrpcSupport;
import com.atlassian.bitbucket.mesh.io.MoreFiles;
import com.atlassian.bitbucket.mesh.io.UnderLock;
import com.atlassian.bitbucket.mesh.process.handler.LoggingStdioHandler;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcTransaction;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcTransactionCommitFailed;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcTransactionRollbackFailed;
import com.atlassian.bitbucket.mesh.transaction.PreparedAction;
import com.atlassian.bitbucket.mesh.transaction.Transaction;
import com.atlassian.bitbucket.mesh.transaction.TransactionCommitFailedException;
import com.atlassian.bitbucket.mesh.transaction.TransactionRollbackFailedException;
import com.google.common.collect.ImmutableMap;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nonnull;
import java.io.BufferedReader;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.nio.channels.OverlappingFileLockException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.NoSuchFileException;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.attribute.BasicFileAttributes;
import java.time.Duration;
import java.util.Map;

import static com.atlassian.bitbucket.mesh.git.GitConstants.*;

class CreateRepositoryOperation extends ConcreteRepositoryVisitor<Void> {

    /**
     * Defines the maximum number of times repositories can be recursively forked.
     * <p>
     * {@code git} internally has a recursion limit of 5, when processing alternates, which forms the basis for this
     * limit. <i>However, due to how Git's limit is applied, the functional limit there is 6.</i> The limit here is
     * <i>intentionally</i> one less than Git's limit to account for the temporary clone which is used to merge pull
     * requests. That clone uses alternates back to the pull request's repository (or repositories). As a result, if
     * the limit here matched {@code git}'s limit, pull requests created between repositories at the limit would not
     * be mergeable--they would be one alternate too deep.
     *
     * @see #calculateDepth(Repository)
     */
    static final int MAX_DEPTH = 5;
    /**
     * Only used in Bitbucket Server data stores to reconfigure the fork's parent, not the fork itself, to disable
     * pruning objects during GC. Due to the use of alternates, repositories which have been forked no longer know
     * which objects are or are not in use.
     */
    private static final Map<String, String> ORIGIN_CONFIG = ImmutableMap.<String, String>builder()
            .put(CONFIG_GC_PRUNE_EXPIRE, CONFIG_GC_PRUNE_EXPIRE_NEVER)
            .build();

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

    private final GitProcessBuilderFactory builderFactory;
    private final EventPublisher eventPublisher;
    private final Repository origin;
    private final CreateRepositoryParameters parameters;
    private final RepositoryHashService repositoryHashService;
    private final Transaction transaction;

    CreateRepositoryOperation(GitProcessBuilderFactory builderFactory, EventPublisher eventPublisher,
                              CreateRepositoryParameters parameters, RepositoryHashService repositoryHashService,
                              Transaction transaction) {
        this.builderFactory = builderFactory;
        this.eventPublisher = eventPublisher;
        this.parameters = parameters;
        this.repositoryHashService = repositoryHashService;
        this.transaction = transaction;

        origin = parameters.getOrigin();
    }

    @Override
    public Void defaultValue(Repository repository) {
        throw new FailedPreconditionException("Repository creation is not supported for " +
                repository.getClass().getSimpleName());
    }

    @Override
    Void visit(@Nonnull DefaultBitbucketServerRepository repository) {
        if (origin == null) {
            initBareRepository(repository);
            eventPublisher.publish(new RepositoryCreatedEvent(repository));
        } else {
            checkDepth(repository);
            initBareFork(repository);

            //Apply additional configuration to origin to disable pruning unreachable objects. This is
            //necessitated by the use of alternates; parent repositories don't know which of their objects
            //forks are borrowing, so they cannot remove any objects. Note that the _fork_ still garbage
            //collects like normal.
            configureOrigin();

            eventPublisher.publish(new ForkCreatedEvent(repository, origin));
        }

        return null;
    }

    @Override
    Void visit(@Nonnull DefaultManagedRepository repository) {
        PreparedCreate create = new PreparedCreate(repository);
        create.prepare();

        //Enroll the create in the transaction
        if (!transaction.prepared(create).isSuccess()) {
            throw new CreateRepositoryFailedException(repository.getId(), "Coordinated repository creation failed");
        }

        return null;
    }

    @Override
    Void visit(@Nonnull DefaultSharedObjectsRepository repository) {
        Path sharedObjectsDir = repository.getPath();
        try {
            BasicFileAttributes attributes = Files.readAttributes(sharedObjectsDir, BasicFileAttributes.class);
            if (attributes.isDirectory()) {
                //It exists, nothing left to do
                return null;
            }

            //If we're able to read attributes, it means something other than a directory exists on disk at that path.
            throw new RepositoryAlreadyExistsException(repository.getId(), sharedObjectsDir,
                    "Cannot create shared-objects repository for " + repository.getId() + " at " +
                            sharedObjectsDir + "; a file exists at the same location");
        } catch (FileNotFoundException | NoSuchFileException ignored) {
            //It does not exist yet, we'll create it outside the try to let any exceptions propagate
        } catch (IOException e) {
            //This shouldn't happen, but we allow repository creation to continue because we can't be certain
            //a failure here is going to translate to overall creation failing
            log.warn("{}: Failed to check target path: {}", repository.getId(), sharedObjectsDir, e);
        }
        initBareRepository(repository);

        return null;
    }

    /**
     * Walks recursive alternates to calculate the <i>current</i> fork depth. {@code git} has an internal <i>hard-
     * coded</i> limit of walking no more than 5 layers of alternates. That means objects that are only available
     * in repositories above that depth will not be reachable from the fork, resulting in the creation of a "broken"
     * fork.
     * <p>
     * <i>Direct</i> alternates (those listed within the fork repository) are "free". The parent {@link Repository}
     * will be a direct alternate, so the initial depth here is considered to be 0, not 1. This means forks of a
     * repository with no origin (a "root" repository) will have a calculated depth  of 0. The first fork of a fork
     * will have depth 1, a fork of a fork of a fork will have depth 2, and so on.
     * <p>
     * For more details, see {@code sha1_file.c} in Git. {@code link_alt_odb_entries} applies the recursion limit.
     * Due to how the functions call each other, the maximum <i>depth</i> ends up being 7 object stores:
     * <ol>
     *     <li>"My" object store has no calculated depth</li>
     *     <li>Parent's object store has a depth of 0</li>
     *     <li>Grandparent's object store has a depth of 1</li>
     *     <li>Great grandparent's object store has a depth of 2</li>
     *     <li>Great great grandparent's object store has a depth of 3</li>
     *     <li>Great great great grandparent's object store has a depth of 4</li>
     *     <li>Great great great great grandparent's object store has a depth of 5</li>
     * </ol>
     * Git's processing blocks strictly greater than the limit, so alternates <i>at</i> 5 deep <i>are</i> valid.
     *
     * @return the calculated fork depth
     */
    private int calculateDepth(Repository repository) {
        int depth = 0;

        Path alternates = MoreFiles.resolve(origin.getPath(), PATH_OBJECTS, PATH_INFO, PATH_ALTERNATES);
        while (Files.isRegularFile(alternates)) {
            //If an alternate exists, it points to the objects directory for a "parent". This may _not_ align with
            //origins in the database, so those are ignored here. Instead, this will traverse up all the alternates
            //to compute the _real_ depth.
            ++depth;

            try (BufferedReader reader = Files.newBufferedReader(alternates, StandardCharsets.UTF_8)) {
                //Alternates point to object directories, so they can be used as the "base" part of a construct
                //call to assemble a path to their recursive alternates, if present.
                String first = reader.readLine();
                if (StringUtils.isBlank(first)) {
                    //The system should never trigger the creation of empty alternates files, but just in case a
                    //customer ends up with one, if the first line is blank (empty or null), break out
                    break;
                }

                alternates = Paths.get(first, PATH_INFO, PATH_ALTERNATES);
            } catch (IOException e) {
                throw new CreateRepositoryFailedException(repository.getId(), e.getMessage(), e);
            }
        }

        return depth;
    }

    /**
     * Ensures the depth of the would-be fork does not exceed the {@link #MAX_DEPTH maximum}.
     *
     * @throws ForkHierarchyTooDeepException if the fork would exceed the {@link #MAX_DEPTH maximum} depth
     * @see #calculateDepth(Repository)
     */
    private void checkDepth(Repository repository) {
        //Git's application of the recursion limit is strictly greater than, meaning the limit itself is a valid
        //depth. As a result, we apply less-than-or-equal to our limit. This will leave one more layer of forks
        //before hitting Git's hard-coded limit, allowing room for the temporary clone used to merge pull requests.
        int depth = calculateDepth(repository);
        if (depth < MAX_DEPTH) {
            return;
        }
        throw new ForkHierarchyTooDeepException(repository.getId(), depth,
                repository.getId() + " cannot be forked. For performance reasons, fork hierarchies are " +
                        "limited to a maximum depth of " + MAX_DEPTH + " beneath a given root repository.");
    }

    private void configureOrigin() {
        log.debug("Applying {} configuration setting(s) to {}", ORIGIN_CONFIG.size(), origin.getId());

        GitConfig configBuilder = builderFactory.builder(origin).config();
        ORIGIN_CONFIG.forEach((key, value) -> {
            log.trace("Configuring {} -> {}", key, value);

            parameters.configureAndRun(configBuilder.set(key, value).build());
        });
    }

    private void initBareFork(AbstractRepository repository) {
        log.trace("{}: Initializing fork", repository.getId());
        parameters.configureAndRun(builderFactory.builder().clone()
                .bare()
                .directory(repository.getPath())
                .origin(origin.getPath())
                .shared(true)
                .template(repository.getTemplateDir())
                .build());
    }

    private void initBareRepository(AbstractRepository repository) {
        log.trace("{}: Initializing empty repository at {}", repository.getId(), repository.getPath());
        parameters.configureAndRun(builderFactory.builder().init()
                .bare(true)
                .directory(repository.getPath())
                .quiet(true)
                .shared("false")
                .template(repository.getTemplateDir())
                .build());
    }

    private class PreparedCreate implements PreparedAction {

        private final DefaultManagedRepository repository;

        private volatile UnderLock.LockedFile lockedFile;

        private PreparedCreate(DefaultManagedRepository repository) {
            this.repository = repository;
        }

        @Override
        public void commit() {
            try {
                int maxAttempts = 5;
                for (int attempt = 1; attempt <= maxAttempts; ++attempt) {
                    try {
                        Files.delete(repository.getDeletedMarker());
                        log.trace("[{}] Published the repository on commit", repository);
                        break;
                    } catch (FileNotFoundException | NoSuchFileException e) {
                        //Nothing to delete
                        break;
                    } catch (IOException e) {
                        if (attempt == maxAttempts) {
                            throw new TransactionCommitFailedException(
                                    "Could not make repository visible on commit",
                                    RpcTransactionCommitFailed.Reason.PUBLISH_FAILED, e);
                        } else {
                            try {
                                Thread.sleep(100L);
                            } catch (InterruptedException interruptedException) {
                                Thread.currentThread().interrupt();
                                throw new TransactionCommitFailedException(
                                        "Could not make repository visible on commit",
                                        RpcTransactionCommitFailed.Reason.INTERRUPTED);
                            }
                        }
                    }
                }
            } finally {
                releaseLock();
            }

            if (origin == null) {
                eventPublisher.publish(new RepositoryCreatedEvent(repository));
            } else {
                //Initialize a fork's refs _after_ 'publishing' the bare repository and releasing the hierarchy lock
                //file to minimize how long other repository creation/deletion operations in the same repository need
                //to wait. Once the bare repository has been 'published' it's safe for those other operations to
                //proceed; the hierarchy lock is meant to protect against a race between the delete of the last repo
                //in a hierarchy and a concurrent fork operation of the same repository. Without the locking, the
                //delete could end up removing the hierarchy's shared-objects, leaving the fork in a bad state.
                parameters.configureAndRun(builderFactory.builder(repository).fetch()
                        .quiet(true)
                        .refspecs("+refs/heads/*:refs/heads/*", "+refs/tags/*:refs/tags/*")
                        .repository(origin)
                        .withConfiguration(CONFIG_CORE_HOOKS_PATH, "") //Don't call hooks for the initial fetch
                        .build(new LoggingStdioHandler()));

                //Since we're not calling the hooks while fetching, the content-hash will end up being stale
                //Clear the content hash after the fetch to force a recalculation.
                repositoryHashService.resetContentHash(repository);
                eventPublisher.publish(new ForkCreatedEvent(repository, origin));
            }
        }

        @Nonnull
        @Override
        public Repository getRepository() {
            //We treat forking a repository as an operation on the origin repository to ensure
            //that all replicas initialize their fork refs while holding the write lock on the _origin_
            //and therefore end up with the same set of refs.
            return origin == null ? repository : origin;
        }

        @Nonnull
        @Override
        public RpcTransaction.Builder populateTransaction(@Nonnull RpcTransaction.Builder builder) {
            return builder.setCreateRepository(parameters.toCreateRepositoryRequest());
        }

        @Override
        public void rollBack() {
            try {
                int maxAttempts = 5;
                for (int attempt = 1; attempt <= maxAttempts; ++attempt) {
                    try {
                        try {
                            MoreFiles.deleteRecursively(repository.getPath());
                        } catch (FileNotFoundException | NoSuchFileException e) {
                            //The repository was never created on disk, nothing to do
                        }
                        try {
                            Files.delete(repository.getDeletedMarker());
                        } catch (FileNotFoundException | NoSuchFileException e) {
                            //The deleted marker doesn't exist, nothing to do
                        }
                    } catch (IOException e) {
                        if (attempt == maxAttempts) {
                            throw new TransactionRollbackFailedException(
                                    "Could not delete partially created repository",
                                    RpcTransactionRollbackFailed.Reason.CLEANUP_FAILED, e);
                        } else {
                            try {
                                Thread.sleep(100L);
                            } catch (InterruptedException interruptedException) {
                                Thread.currentThread().interrupt();
                                throw new TransactionRollbackFailedException(
                                        "Could not delete partially created repository",
                                        RpcTransactionRollbackFailed.Reason.INTERRUPTED);
                            }
                        }
                    }
                }
            } finally {
                releaseLock();
            }
        }

        void prepare() {
            boolean prepared = false;
            try {
                Path hierarchyLockFile = repository.getHierarchyLockFile();
                MoreFiles.mkdir(hierarchyLockFile.getParent());
                try {
                    lockedFile = UnderLock.forFile(hierarchyLockFile, getLockTimeout()).lock();
                } catch (OverlappingFileLockException e) {
                    throw new CreateRepositoryFailedException(repository.getId(), "Hierarchy is already locked", e);
                } catch (IOException e) {
                    throw new CreateRepositoryFailedException(repository.getId(), "Hierarchy lock failed", e);
                }

                //Ensure the shared-objects repository exists
                visit(repository.getSharedObjects());

                //Create the deleted marker to hide the repository until the transaction commits
                try {
                    MoreFiles.touch(repository.getDeletedMarker());
                } catch (IOException e) {
                    //Failed to created the deleted marker
                    log.warn("[{}] Failed to hide the repository until the transaction commits", repository, e);
                }

                //Initialize the repository. In case of a fork, we postpone fetching the refs from the origin
                //until we're holding the write lock on origin on all replicas and commit. This ensures that all
                //replicas end up fetching the same refs.
                initBareRepository(repository);
                prepared = true;
            } finally {
                if (!prepared) {
                    releaseLock();
                }
            }
        }

        void releaseLock() {
            UnderLock.LockedFile locked = lockedFile;
            if (locked != null) {
                try {
                    locked.close();
                    lockedFile = null;
                } catch (IOException e) {
                    log.warn("[{}] Failed to release hierarchy lock", repository.getHierarchyId(), e);
                }
            }
        }

        private Duration getLockTimeout() {
            long timeoutSec = parameters.getExecutionTimeout();
            //Apply a 30s timeout if no execution timeout is set
            Duration timeout = Duration.ofSeconds(timeoutSec <= 0L ? 30L : timeoutSec);
            return GrpcSupport.maybeReduceToDeadline(timeout);
        }
    }
}
