package com.atlassian.stash.internal.maintenance;

import com.atlassian.bitbucket.i18n.I18nService;
import com.atlassian.stash.internal.hazelcast.NodeIdMemberSelector;
import com.atlassian.stash.internal.maintenance.latch.ResultCollectingExecutionCallback;
import com.atlassian.bitbucket.user.ApplicationUser;
import com.google.common.base.Throwables;
import com.google.common.collect.Sets;
import com.hazelcast.core.IExecutorService;
import com.hazelcast.core.Member;
import com.hazelcast.spring.context.SpringAware;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import javax.annotation.Nonnull;
import java.io.Serializable;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;

import static com.google.common.base.Preconditions.checkNotNull;

/**
 * {@link MaintenanceLock} implementation that locks and unlocks all nodes in a cluster for maintenance.
 */
@SpringAware
public class ClusterMaintenanceLock implements MaintenanceLock, Serializable {
    private static final long serialVersionUID = 1L;

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

    // services that will be injected by Hazelcast when the instance is deserialized
    private transient IExecutorService executorService;
    private transient I18nService i18nService;
    private transient InternalMaintenanceService maintenanceService;

    private final ApplicationUser owner;
    private final String token;

    public ClusterMaintenanceLock(@Nonnull ApplicationUser owner, @Nonnull String token, IExecutorService executorService,
                                  I18nService i18nService, InternalMaintenanceService maintenanceService) {
        this.executorService = executorService;
        this.i18nService = i18nService;
        this.maintenanceService = maintenanceService;
        this.owner = checkNotNull(owner, "owner");
        this.token = checkNotNull(token, "token");
    }

    public void lock() {
        FailureTrackingCallback callback = new FailureTrackingCallback();
        try {
            // lock all the nodes in the cluster
            executorService.submitToAllMembers(new LockTask(this), callback);
            callback.await(1, TimeUnit.MINUTES);
        } catch (InterruptedException e) {
            throw Throwables.propagate(e);
        } finally {
            // if one of the nodes failed to lock, unlock them all
            if (!callback.isSuccess()) {
                unlock(token);
                String failedNodeIds = StringUtils.join(callback.getFailedNodes(), ", ");
                throw new LockFailedMaintenanceException(
                        i18nService.createKeyedMessage("bitbucket.service.maintenance.lock.failed", failedNodeIds));
            }
        }
    }

    @Nonnull
    @Override
    public ApplicationUser getOwner() {
        return owner;
    }

    @Nonnull
    @Override
    public String getUnlockToken() {
        return token;
    }

    @Autowired
    public void setExecutorService(IExecutorService executorService) {
        this.executorService = executorService;
    }

    @Autowired
    public void setI18nService(I18nService i18nService) {
        this.i18nService = i18nService;
    }

    @Autowired
    public void setMaintenanceService(InternalMaintenanceService maintenanceService) {
        this.maintenanceService = maintenanceService;
    }

    @Override
    public void unlock(@Nonnull String token) {
        if (checkNotNull(token, "token").equals(this.token)) {
            FailureTrackingCallback callback = new FailureTrackingCallback();
            executorService.submitToAllMembers(new UnlockTask(token), callback);
            try {
                if (!callback.await(1, TimeUnit.MINUTES)) {
                    log.info("Timed out waiting for all nodes to unlock");
                }
                if (!callback.isSuccess()) {
                    log.info("Failed to unlock some nodes: {}. Retrying", StringUtils.join(callback.getFailedNodes(), ", "));
                    executorService.submitToMembers(new UnlockTask(token), new NodeIdMemberSelector(callback.getFailedNodes()));
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt(); // retain interrupted flag
                log.info("Interrupted while waiting for nodes to unlock");
            }
            maintenanceService.clearClusterLock();
        } else {
            log.warn("An invalid token ({}) was supplied to attempt to unlock the system", token);
            throw new IncorrectTokenMaintenanceException(
                    i18nService.createKeyedMessage("bitbucket.service.maintenance.lock.incorrectunlocktoken"), token);
        }
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) {
            return true;
        }
        if (o == null || getClass() != o.getClass()) {
            return false;
        }

        ClusterMaintenanceLock that = (ClusterMaintenanceLock) o;

        return token.equals(that.token);
    }

    @Override
    public int hashCode() {
        return token.hashCode();
    }

    private static class FailureTrackingCallback extends ResultCollectingExecutionCallback<Void> {
        private final Set<String> failedNodes = Sets.newHashSet();

        public Set<String> getFailedNodes() {
            return failedNodes;
        }

        @Override
        protected void onError(Member member, Throwable throwable) {
            failedNodes.add(member.getUuid());
        }
    }

    /**
     * Locks a node for maintenance
     */
    @SpringAware
    private static class LockTask implements Runnable, Serializable {
        private static final long serialVersionUID = 1L;

        private final MaintenanceLock lockInfo;
        private transient InternalMaintenanceService maintenanceService;

        private LockTask(MaintenanceLock lockInfo) {
            this.lockInfo = lockInfo;
        }

        @Override
        public void run() {
            maintenanceService.lockNode(lockInfo);
        }

        @Autowired
        public void setMaintenanceService(InternalMaintenanceService maintenanceService) {
            this.maintenanceService = maintenanceService;
        }
    }

    /**
     * Unlocks a node from maintenance mode
     */
    @SpringAware
    private static class UnlockTask implements Callable<Void>, Serializable {
        private static final long serialVersionUID = 1L;

        private transient InternalMaintenanceService maintenanceService;
        private final String unlockToken;

        private UnlockTask(String unlockToken) {
            this.unlockToken = unlockToken;
        }

        @Override
        public Void call() {
            MaintenanceLock lock = maintenanceService.getNodeLock();
            if (lock != null) {
                lock.unlock(unlockToken);
            }
            return null;
        }

        @Autowired
        public void setMaintenanceService(InternalMaintenanceService maintenanceService) {
            this.maintenanceService = maintenanceService;
        }
    }
}
