package com.atlassian.bitbucket.mesh.transaction;

import com.atlassian.bitbucket.mesh.io.MoreFiles;
import com.atlassian.bitbucket.mesh.io.UnderLock;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcTransaction;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Throwables;
import com.google.common.collect.ImmutableList;
import com.google.protobuf.CodedInputStream;
import com.google.protobuf.CodedOutputStream;
import org.apache.commons.lang3.mutable.MutableObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nonnull;
import java.io.*;
import java.nio.file.*;
import java.nio.file.attribute.BasicFileAttributes;
import java.time.Duration;
import java.time.Instant;
import java.util.*;
import java.util.stream.Stream;

import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.collect.ImmutableList.toImmutableList;
import static java.util.Collections.emptySet;
import static java.util.Objects.requireNonNull;
import static org.apache.commons.lang3.StringUtils.isBlank;

/**
 * {@link TransactionLog} implementation that writes {@link RpcTransaction}s to disk in one or more files, and edits
 * said file(s) under a file lock to allow the transaction log to be safely used by multiple processes concurrently.
 * <p>
 * This is necessary because a {@code FileTransactionLog} on NFS is used to record all repository changes to a
 * "classic" Bitbucket Server repository while migration of the repository to Mesh is in progress.
 * <p>
 * Note that when transactions are {@link #append(RpcTransaction) appended} to or {@link #process(Processor, int) taken}
 * from the log, the transaction log is rewritten under a file lock. To limit the amount of data needed to be rewritten,
 * the transaction log is split into segments and only the relevant segment is rewritten (under lock). In addition, if
 * a very large transaction is appended to the log, only the summary information is appended to the log and the
 * transaction details are written to a separate file.
 * <p>
 * The most recent segment is always stored under the path passed into the {@link #FileTransactionLog(Path) constructor}.
 * Older segments are stored under the same path, but with their segment ID appended as a suffix.
 * <p>
 * The file layout is as follows:
 * <ul>
 *     <li>variable length uint32 - version of the transaction log format (currently 1)</li>
 *     <li>variable length uint64 - previous segment ID</li>
 *     <li>variable length uint64 - this segment ID</li>
 *     <li>RpcTransaction summaries in protobuf delimited format:
 *       <ul>
 *         <li>size in bytes as a variable length uint64</li>
 *         <li>RpcTransaction in protobuf format, with the details field(s) stripped off</li>
 *       </ul>
 *     </li>
 * </ul>
 */
public class FileTransactionLog implements TransactionLog {

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

    private static final Duration DEFAULT_LOCK_TIMEOUT = Duration.ofSeconds(5L);
    private static final int DEFAULT_TRANSACTIONS_PER_SEGMENT = 1024;
    private static final long MAX_SIZE_INLINE = 1024;
    private static final int VERSION = 1;

    private final Path file;
    private final int maxTransactionsPerSegment;

    public FileTransactionLog(Path file) {
        this(file, DEFAULT_TRANSACTIONS_PER_SEGMENT);
    }

    public FileTransactionLog(Path file, int maxTransactionsPerSegment) {
        this.file = file;
        this.maxTransactionsPerSegment = maxTransactionsPerSegment;
    }

    @Override
    public void append(@Nonnull RpcTransaction transaction) {
        internalAppend(requireNonNull(transaction, "transaction"));
    }

    /**
     * Creates an empty transaction log file, with a valid header if one does not yet exist
     */
    public void init() {
        internalAppend(null);
    }

    @Override
    public void process(@Nonnull Processor processor, int preferredBatchSize) {
        requireNonNull(processor, "processor");
        checkArgument(preferredBatchSize > 0, "preferredBatchSize must be non-negative");

        for (Path segment : getSegments()) {
            Result result = processSegment(segment, processor, preferredBatchSize);
            if (result.getUnprocessed() == 0) {
                //Previous segment(s) completely processed and is now empty
                MoreFiles.deleteQuietly(segment);
            }
            //Rethrow any exception the processor threw _after_ the unprocessed transactions have
            //been persisted in the segment
            result.getException().ifPresent(exception -> {
                Throwables.throwIfUnchecked(exception);
                throw new TransactionLogException("Error while processing the transaction log", exception);
            });
            if (result.isAbort()) {
                break;
            }
        }
    }

    /**
     * @return {@link Stream stream} of {@link RpcTransaction transactions} that are part of the log
     */
    @Nonnull
    @VisibleForTesting
    public Stream<RpcTransaction> stream() {
        return streamSegment(file);
    }

    private static void checkVersionSupported(int version) {
        if (version > VERSION) {
            throw new IllegalStateException("Cannot read transaction logs of version " + version);
        }
    }

    private static String getRepositoryPrefix(String repositoryId) {
        //The repository ID is at the end of the FQ repository, even for path: type repository IDs
        int index = repositoryId.lastIndexOf('/');
        String prefix = index == -1 ? repositoryId : repositoryId.substring(index + 1);
        return MoreFiles.pathSafe(prefix);
    }

    private static RpcTransaction loadTransactionFromFile(Path file) {
        try (InputStream inputStream = Files.newInputStream(file)) {
            CodedInputStream cis = CodedInputStream.newInstance(inputStream, 1);
            checkVersionSupported(cis.readRawVarint32());
            return RpcTransaction.parseDelimitedFrom(inputStream);
        } catch (FileNotFoundException | NoSuchFileException e) {
            return null;
        } catch (IOException e) {
            log.error("Failed to read transaction details from {}", file, e);
            throw new TransactionLogException("Error reading transaction details from " + file.getFileName());
        }
    }

    private static List<RpcTransaction> readTransactions(InputStream input, int numberToRead) throws IOException {
        ImmutableList.Builder<RpcTransaction> builder = ImmutableList.builder();
        RpcTransaction transaction;
        int size = 0;
        while (size < numberToRead && (transaction = RpcTransaction.parseDelimitedFrom(input)) != null) {
            builder.add(transaction);
            size++;
        }
        return builder.build();
    }

    private static void writeTransactionDetails(Path file, RpcTransaction transaction) {
        MoreFiles.mkdir(file.getParent());

        try (OutputStream outputStream = Files.newOutputStream(file)) {
            CodedOutputStream cos = CodedOutputStream.newInstance(outputStream);
            cos.writeInt32NoTag(VERSION);
            cos.flush();
            transaction.writeDelimitedTo(outputStream);
        } catch (IOException e) {
            log.error("Failed to write transaction details to {}", file, e);
            throw new TransactionLogException("Error writing transaction details to " + file.getFileName());
        }
    }

    private static void writeTransactions(OutputStream output, List<RpcTransaction> transactions) throws IOException {
        for (RpcTransaction transaction : transactions) {
            transaction.writeDelimitedTo(output);
        }
    }

    private void deleteTransactionDetails(List<RpcTransaction> transactions) {
        for (RpcTransaction transaction : transactions) {
            Path detailsFile = getTransactionPath(transaction.getId());
            try {
                Files.delete(detailsFile);
            } catch (FileNotFoundException | NoSuchFileException e) {
                //Nothing left to do
            } catch (IOException e) {
                log.warn("Failed to delete transaction details file {}", detailsFile);
            }
        }
    }

    private Path getSegmentPath(long segmentId) {
        return file.resolveSibling(file.getFileName() + "." + segmentId);
    }

    private List<Path> getSegments() {
        List<Path> segments = new ArrayList<>();
        String prefix = file.getFileName().toString();
        try {
            Files.walkFileTree(file.getParent(), emptySet(), 1, new SimpleFileVisitor<Path>() {

                @Override
                public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) {
                    if (attrs.isRegularFile() && file.getFileName().toString().startsWith(prefix)) {
                        segments.add(file);
                    }
                    return FileVisitResult.CONTINUE;
                }
            });
        } catch (FileNotFoundException | NoSuchFileException e) {
            //Parent directory does not exist, nothing to do
        } catch (IOException e) {
            log.warn("Error while searching for segments of {}", file, e);
            throw new TransactionLogException("Error while searching for segments of " + file.getFileName(), e);
        }
        segments.sort((path1, path2) -> {
            if (path1.equals(path2)) {
                return 0;
            }
            //If either of the paths matches file, _that_ path is the most recent segment and should be last in the list
            if (path1.equals(file)) {
                return 1;
            }
            if (path2.equals(file)) {
                return -1;
            }
            //Both file names MUST start with the txLog path, followed by '.' and the segment ID
            long segmentId1 = Long.parseLong(path1.getFileName().toString().substring(prefix.length() + 1));
            long segmentId2 = Long.parseLong(path2.getFileName().toString().substring(prefix.length() + 1));
            return Long.compare(segmentId1, segmentId2);
        });
        return segments;
    }

    private Path getTransactionPath(String transactionId) {
        return file.resolveSibling(transactionId + ".dat");
    }

    private void internalAppend(RpcTransaction transaction) {
        boolean appended = false;
        try {
            MoreFiles.mkdir(file.getParent());
            UnderLock.forFile(file, DEFAULT_LOCK_TIMEOUT).editRaw((input, output) -> {
                long prevSegmentId = 0L;
                long segmentId;
                Queue<RpcTransaction> remaining = new LinkedList<>();
                if (input == null) {
                    segmentId = Instant.now().toEpochMilli();
                } else {
                    CodedInputStream cis = CodedInputStream.newInstance(input, 1);
                    checkVersionSupported(cis.readUInt32());
                    prevSegmentId = cis.readUInt64();
                    segmentId = cis.readUInt64();

                    //Read the existing transactions into memory; these will be retained but possibly written to another
                    //segment file (if the existing + new transactions exceed maxTransactionPerSegment)
                    RpcTransaction existing;
                    input = new BufferedInputStream(input, 1024);
                    while ((existing = RpcTransaction.parseDelimitedFrom(input)) != null) {
                        remaining.add(existing);
                    }
                }
                if (transaction != null) {
                    remaining.add(transaction);
                }

                //Write out separate segment files if we would end up with more than maxTransactionsPerSegment
                //transactions in file
                if (remaining.size() > maxTransactionsPerSegment) {
                    writeSegment(remaining, prevSegmentId, segmentId);
                    prevSegmentId = segmentId;
                    ++segmentId;
                }

                //Write the last 0..maxTransactionsPerSegment current transactions to output
                writeSegment(output, remaining, prevSegmentId, segmentId);
            }, false);
            appended = true;
        } catch (IOException e) {
            log.error("Failed to write to transaction log {}", file, e);
            throw new TransactionLogException("Failed to write to transaction log " + file.getFileName());
        } finally {
            //For posterity, always log the full transaction if we failed to append it to the log file for some reason
            if (!appended) {
                log.error("Failed to record transaction in {}. Details: {}", file, transaction);
            }
        }
    }

    private RpcTransaction maybeLoadDetails(RpcTransaction transaction) {
        if (transaction.getDetailsOneofCase() == RpcTransaction.DetailsOneofCase.DETAILSONEOF_NOT_SET) {
            Path detailPath = getTransactionPath(transaction.getId());
            RpcTransaction fullTransaction = loadTransactionFromFile(detailPath);
            if (fullTransaction != null) {
                return fullTransaction;
            }
        }
        return transaction;
    }

    private List<RpcTransaction> maybeLoadDetails(List<RpcTransaction> transactions) {
        return transactions.stream()
                .map(this::maybeLoadDetails)
                .collect(toImmutableList());
    }

    private Result processSegment(Path path, Processor processor, int preferredBatchSize) {
        try {
            MutableObject<Result> retVal = new MutableObject<>(Result.proceed());
            UnderLock.forFile(path, DEFAULT_LOCK_TIMEOUT).editRaw((input, output) ->
                    retVal.setValue(processSegment(input, output, processor, preferredBatchSize)), false);
            return retVal.getValue();
        } catch (FileNotFoundException | NoSuchFileException e) {
            //The segment does not exist. Treat it the same as a fully consumed segment
            return Result.proceed();
        } catch (IOException e) {
            log.error("Error processing transaction log segment {}", path, e);
            throw new TransactionLogException("Error processing transaction log segment " + path.getFileName(), e);
        }
    }

    private Result processSegment(InputStream input, OutputStream output,
                                  Processor processor, int preferredBatchSize) throws IOException {
        if (input == null) {
            //Nothing to process
            return Result.proceed();
        }

        CodedInputStream cis = CodedInputStream.newInstance(input, 1);
        checkVersionSupported(cis.readUInt32());
        long prevSegmentId = cis.readUInt64();
        long segmentId = cis.readUInt64();

        //Process the contents of the current segment
        CodedOutputStream cos = CodedOutputStream.newInstance(output, 12);
        cos.writeUInt32NoTag(VERSION);
        cos.writeUInt64NoTag(prevSegmentId);
        cos.writeUInt64NoTag(segmentId);
        cos.flush();

        Result result = Result.proceed();
        while (!result.isAbort()) {
            List<RpcTransaction> transactions = readTransactions(input, preferredBatchSize);
            try {
                result = processor.process(maybeLoadDetails(transactions));
            } catch (RuntimeException e) {
                //Assume all transactions need to be retained in the log
                result = Result.error(transactions.size(), e);
            }

            int unprocessed = Math.min(Math.max(0, result.getUnprocessed()), transactions.size());
            int indexFirstUnprocessed = transactions.size() - unprocessed;
            deleteTransactionDetails(transactions.subList(0, indexFirstUnprocessed));
            writeTransactions(output, transactions.subList(indexFirstUnprocessed, transactions.size()));

            if (transactions.size() < preferredBatchSize) {
                //End of segment reached; the outcome of processing the last batch _is_ the outcome for this segment;
                //If the processor voted to abort, this segment should be deleted if result.getUnprocessed() == 0
                return result;
            }
        }

        if (result.isAbort()) {
            //Write the remaining transactions to output without offering them to the processor
            List<RpcTransaction> transactions;
            do {
                transactions = readTransactions(input, 1000);
                if (!transactions.isEmpty()) {
                    //Signal to the caller that this segment is not deleted by setting the unprocessed count to at
                    //least one
                    result = new Result(result.isAbort(), Math.max(1, result.getUnprocessed()),
                            result.getException().orElse(null));
                }
                writeTransactions(output, transactions);
            } while (transactions.size() == 1000);
        }

        return result;
    }

    @Nonnull
    private Stream<RpcTransaction> streamSegment(Path path) {
        try (InputStream input = Files.newInputStream(path)) {
            CodedInputStream cis = CodedInputStream.newInstance(input, 1);
            int version = cis.readUInt32();
            checkVersionSupported(version);
            long prevSegmentId = cis.readUInt64();
            cis.readUInt64(); // this segment ID
            ImmutableList.Builder<RpcTransaction> builder = ImmutableList.builder();
            RpcTransaction transaction;
            while ((transaction = RpcTransaction.parseDelimitedFrom(input)) != null) {
                builder.add(transaction);
            }
            Stream<RpcTransaction> result = builder.build().stream()
                    .map(this::maybeLoadDetails);
            return prevSegmentId == 0L ? result : Stream.concat(streamSegment(getSegmentPath(prevSegmentId)), result);
        } catch (FileNotFoundException | NoSuchFileException e) {
            return Stream.empty();
        } catch (IOException e) {
            log.error("Error streaming transaction log segment {}", path, e);
            throw new TransactionLogException("Error streaming transaction log segment " + path.getFileName(), e);
        }
    }

    private void writeSegment(Queue<RpcTransaction> transactions, long prevSegmentId, long segmentId) {
        Path path = getSegmentPath(segmentId);
        try (OutputStream output = Files.newOutputStream(path, StandardOpenOption.CREATE_NEW)) {
            writeSegment(output, transactions, prevSegmentId, segmentId);
        } catch (IOException e) {
            log.error("Error writing transaction log segment {}", path, e);
            throw new TransactionLogException("Error writing transaction log segment " + path.getFileName(), e);
        }
    }

    private void writeSegment(OutputStream output, Queue<RpcTransaction> transactions,
                              long prevSegmentId, long segmentId) throws IOException {
        BufferedOutputStream bufferedOut = new BufferedOutputStream(output);
        try {
            //Write out the header, consisting of the version and segment IDs
            CodedOutputStream cos = CodedOutputStream.newInstance(bufferedOut, 1);
            cos.writeUInt32NoTag(VERSION);
            cos.writeUInt64NoTag(prevSegmentId);
            cos.writeUInt64NoTag(segmentId);
            cos.flush();

            for (int i = 0; i < maxTransactionsPerSegment; ++i) {
                RpcTransaction transaction = transactions.poll();
                if (transaction == null) {
                    break;
                }

                String prefix = getRepositoryPrefix(transaction.getRepository());
                String transactionId = transaction.getId();
                if (isBlank(transactionId)) {
                    //Generate the transaction ID and inject it into the RpcTransaction
                    transactionId = prefix + "-" + segmentId + "-" + UUID.randomUUID();
                    transaction = transaction.toBuilder()
                            .setId(transactionId)
                            .build();
                }
                if (transaction.getSerializedSize() <= MAX_SIZE_INLINE) {
                    //No separate details file necessary, append the transaction to output as is
                    transaction.writeDelimitedTo(bufferedOut);
                } else {
                    //Transaction was too big to store in the main segment, store the full tx in a separate file and
                    //append a slimmed down version of the transaction to the segment
                    writeTransactionDetails(getTransactionPath(transactionId), transaction);
                    transaction.toBuilder()
                            .clearDetailsOneof()
                            .setId(transactionId)
                            .build()
                            .writeDelimitedTo(bufferedOut);
                }
            }
        } finally {
            bufferedOut.flush();
        }
    }
}
