package com.atlassian.bitbucket.mesh.transaction;

import com.atlassian.bitbucket.mesh.config.HostingConfig;
import com.atlassian.bitbucket.mesh.execution.ExecutionManager;
import com.atlassian.bitbucket.mesh.git.GitAgent;
import com.atlassian.bitbucket.mesh.git.hosting.GitReceivePackListener;
import com.atlassian.bitbucket.mesh.git.hosting.GitReceivePackFilter;
import com.atlassian.bitbucket.mesh.repository.Repository;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcStreamChunk;
import com.atlassian.bitbucket.mesh.util.SatiableConsumer;
import com.atlassian.bitbucket.mesh.util.ShaUtils;
import com.google.protobuf.ByteString;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nonnull;

import static org.apache.commons.lang3.StringUtils.leftPad;

/**
 * {@link GitReceivePackListener} that starts a new transaction when it detects a PACK being sent to this node.
 * It then streams the pack to the transaction followers.
 * <p>
 * In addition, when the filter detects that a non-atomic initial push is made (to an empty repository), the filter
 * injects the {@code atomic} capability to the first packet line. As a result {@code git} treats the first push
 * as atomic, which significantly reduces the overhead of this initial push.
 */
public class TransactionGitReceivePackListener implements GitReceivePackListener {

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

    private final GitAgent agent;
    private final ExecutionManager executionManager;
    private final HostingConfig hostingConfig;
    private final TransactionManager transactionManager;

    public TransactionGitReceivePackListener(GitAgent agent, ExecutionManager executionManager,
                                             HostingConfig hostingConfig, TransactionManager transactionManager) {
        this.agent = agent;
        this.executionManager = executionManager;
        this.hostingConfig = hostingConfig;
        this.transactionManager = transactionManager;
    }

    @Override
    public GitReceivePackFilter createFilter(@Nonnull Repository repository) {
        //Return a filter that replicates the pack data to all replicas, but ONLY if we have a current execution
        //context. We need an execution context to serve as the encapsulating context of the transaction that may
        //be opened by PackReplicator. Without an execution context, we'd end up with a transaction that never closes.
        return executionManager.getContext().isPresent() ? new PackReplicator(repository) : null;
    }

    private class PackReplicator implements GitReceivePackFilter {

        private final Repository repository;

        private boolean done;
        private boolean firstPacketIntercepted;
        private SatiableConsumer<RpcStreamChunk> packWriter;

        PackReplicator(Repository repository) {
            this.repository = repository;
        }

        @Nonnull
        @Override
        public RpcStreamChunk onPack(@Nonnull RpcStreamChunk value) {
            if (packWriter == null) {
                //Start of the PACK section. Start a transaction and replicate it to the remote nodes if we can.
                packWriter = transactionManager.start()
                        .newPackWriter(repository)
                        .orElse(null);
            }
            if (packWriter == null) {
                //Cannot open a writer for the pack in the current transaction. Abort and opt out of
                //receiving additional PACK data
                log.info("[{}] Not streaming pack to replica because the current transaction does not support it",
                        repository);
                done = true;
            } else {
                packWriter.accept(value);
            }
            return value;
        }

        @Nonnull
        @Override
        public RpcStreamChunk onPacketLine(@Nonnull RpcStreamChunk chunk) {
            firstPacketIntercepted = true;
            if (hostingConfig.isAtomicInitialPush()) {
                String packetLine = chunk.getData().toStringUtf8();
                //The first packet line of a (non-signed) push looks like this:
                //<4 byte prefix><from-hash> <to-hash> <ref-id>\0 <capabilities>
                //
                //The <from-hash> is expected to followed by a space at index 44 of the packet line. However, to future
                //proof this for the switch to sha256, we'll also allow longer hashes.
                if (packetLine.length() >= 44 && packetLine.indexOf('\0') > 0 && !packetLine.contains(" atomic")) {
                    int index = packetLine.indexOf(' ', 4);
                    if (index >= 44 && ShaUtils.isNull(packetLine.substring(4, index)) && agent.isEmpty(repository)) {
                        log.info("[{}] Intercepted non-atomic push and treating it as an atomic push", repository);
                        String enriched = injectAtomic(packetLine.substring(4));
                        enriched = leftPad(Integer.toString(enriched.length() + 4, 16), 4, '0') + enriched;
                        return chunk.toBuilder()
                                .setData(ByteString.copyFromUtf8(enriched))
                                .build();
                    }
                }
            }
            return chunk;
        }

        @Override
        public boolean wantsPack() {
            return !done;
        }

        @Override
        public boolean wantsPacketLines() {
            return !firstPacketIntercepted && hostingConfig.isAtomicInitialPush();
        }

        private String injectAtomic(String commandWithCapabilities) {
            int index = commandWithCapabilities.indexOf('\0');
            if (index != -1) {
                return commandWithCapabilities.substring(0, index + 1) + " atomic" +
                        commandWithCapabilities.substring(index + 1);
            }
            return commandWithCapabilities;
        }
    }
}
