package com.atlassian.stash.internal.integrity;

import com.atlassian.bitbucket.cluster.ClusterNode;
import com.atlassian.bitbucket.cluster.ClusterService;
import com.atlassian.bitbucket.util.MoreCollectors;
import com.atlassian.stash.internal.spring.SpringTransactionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataIntegrityViolationException;
import org.springframework.security.access.prepost.PreAuthorize;
import org.springframework.stereotype.Component;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.TransactionCallbackWithoutResult;
import org.springframework.transaction.support.TransactionTemplate;

import javax.annotation.Nonnull;
import java.util.List;

@Component
public class DefaultIntegrityCheckReporter implements IntegrityCheckReporter {

    private static final Logger log = LoggerFactory.getLogger(DefaultIntegrityCheckReporter.class);
    private final ClusterService clusterService;
    private final ApplicationIntegrityDao integrityDao;
    private final TransactionTemplate transactionTemplate;

    @Autowired
    public DefaultIntegrityCheckReporter(ApplicationIntegrityDao integrityDao,
                                         ClusterService clusterService,
                                         PlatformTransactionManager transactionManager) {
        this.integrityDao = integrityDao;
        this.clusterService = clusterService;
        this.transactionTemplate = new TransactionTemplate(transactionManager, SpringTransactionUtils.REQUIRES_NEW);
    }

    @Override
    public void acknowledge() {
        recordEvent(IntegrityEventKey.ACKNOWLEDGED);
    }

    @Override
    public void debug(@Nonnull String message, @Nonnull Object... objects) {
        log.debug(message, objects);
    }

    @Override
    public void error(@Nonnull String message, @Nonnull Object... objects) {
        log.error(message, objects);
    }

    @Override
    @Nonnull
    @PreAuthorize("hasGlobalPermission('SYS_ADMIN')")
    @Transactional(readOnly = true)
    public List<IntegrityEventKey> getLatestEvents(int limit) {
        return integrityDao.getLatestEvents(limit)
                .stream()
                .map(InternalIntegrityEvent::getKey)
                .collect(MoreCollectors.toImmutableList());
    }

    @Override
    public void info(@Nonnull String message, @Nonnull Object... objects) {
        log.info(message, objects);
    }

    @Override
    public void markChecksComplete() {
        log.info("Integrity checking complete");
        recordEvent(IntegrityEventKey.COMPLETED);
    }

    @Override
    public boolean markChecksStarted() {
        if (recordEventIfFirstClusterNode(IntegrityEventKey.STARTED)) {
            log.info("Integrity checking started");
            return true;
        } else {
            return false;
        }
    }

    @Override
    public void reportInconsistency(@Nonnull String message, @Nonnull Object... objects) {
        info(message, objects);
        recordEvent(IntegrityEventKey.INCONSISTENCY);
    }

    @Override
    public void warn(@Nonnull String message, @Nonnull Object... objects) {
        log.warn(message, objects);
    }

    private boolean clusterNodeExists(String nodeVmId) {
        return clusterService.getInformation().getNodes().stream()
                .map(ClusterNode::getVmId)
                .anyMatch(vmId -> vmId.equals(nodeVmId));
    }

    // Set the timestamp on the named Integrity Check related event.
    private void recordEvent(@Nonnull IntegrityEventKey key) {
        String nodeVmId = clusterService.getInformation().getLocalNode().getVmId();

        // There is a race condition inside integrityDao.setTimestamp() the very first time two or more threads try to
        // set the timestamp for the same key.  Try to set it optimistically, and handle any
        // DataIntegrityViolationException raised by retrying once (so that integrityDao.setTimestamp() will execute
        // an UPDATE statement instead of an INSERT).
        for (int tries = 1;; tries++) {
            try {
                transactionTemplate.execute(new TransactionCallbackWithoutResult() {
                    @Override
                    protected void doInTransactionWithoutResult(TransactionStatus status) {
                        integrityDao.setTimestamp(key, nodeVmId);
                    }
                });
                break;
            } catch (DataIntegrityViolationException e) {
                // Another thread probably did an INSERT of the same key.  Retrying should reload the new record from
                // the database and then succeed on the UPDATE path.
                if (tries >= 2) {
                    throw e;
                }
            }
        }
    }

    private boolean recordEventIfFirstClusterNode(@Nonnull IntegrityEventKey key) {
        String nodeVmId = clusterService.getInformation().getLocalNode().getVmId();

        try {
            return transactionTemplate.execute(status -> {
                // If the event with the specified key doesn't exist, or exists but the node that persisted it
                // doesn't appear to be a member of the cluster any more, then go ahead and try to persist the
                // event (with our own cluster node identifier) and return true.
                InternalIntegrityEvent event = integrityDao.getById(key);
                if (event == null) {
                    integrityDao.createWithTimestamp(key, nodeVmId);
                    return true;
                } else if (!clusterNodeExists(event.getNode())) {
                    return integrityDao.updateTimestampIfSameNode(key, event.getNode(), nodeVmId);
                }
                log.info("Integrity checks not {} on this node, node {} has already {} them", key, event.getNode(),
                        key);
                return false;
            });
        } catch (DataIntegrityViolationException e) {
            // Another node got in first and successfully INSERTed the same key.
            log.info("Integrity checks not {} on this node, database insert failed", key);
            log.debug("Exception", e);
            return false;
        }
    }
}
