1package main
2
3import "fmt"
4
5func main() {
6    for i := 0; i < 3; i++ {
7        fmt.Println("Value of i:", i)
8    }
9}

package net.lbruun.dbleaderelect.internal.core;

import static net.lbruun.dbleaderelect.LeaderElector.NO_LEADER_LASTSEENTIMESTAMP_MS;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.SQLRecoverableException;
import java.sql.SQLTransientException;
import java.time.Instant;
import javax.sql.DataSource;
import net.lbruun.dbleaderelect.LeaderElectorConfiguration;
import net.lbruun.dbleaderelect.LeaderElectorListener;
import net.lbruun.dbleaderelect.exception.LeaderElectorException;
import net.lbruun.dbleaderelect.exception.LeaderElectorExceptionNonRecoverable;
import net.lbruun.dbleaderelect.exception.LeaderElectorExceptionRecoverable;
import net.lbruun.dbleaderelect.internal.core.RowInLockTable.CurrentLeaderDbStatus;
import net.lbruun.dbleaderelect.internal.events.EventHelpers;
import net.lbruun.dbleaderelect.internal.sql.SQLCmds;
import net.lbruun.dbleaderelect.internal.utils.SQLUtils;

/**
 *
 */
public class SQLLeaderElect {

    private final LeaderElectorConfiguration configuration;
    private final SQLCmds sqlCmds;
    private final String myRoleId;
    private final String myCandidateId;
    private final DataSource dataSource;
    private final String tableNameDisplay;
    private volatile boolean currentlyAmLeader = false;
    private int noOfConsecutiveTransientErrors = 0;
    private volatile boolean hasHadSuccessfulExection = false;
    private boolean hasRelinquishedLeadership = false;

    public SQLLeaderElect(LeaderElectorConfiguration configuration, DataSource dataSource, String tableNameDisplay) {
        this.dataSource = dataSource;
        this.configuration = configuration;
        this.myRoleId = configuration.getRoleId();
        this.myCandidateId = configuration.getCandidateId();
        this.sqlCmds = SQLCmds.getSQL(configuration);
        this.tableNameDisplay = tableNameDisplay;
    }
    
    public boolean isLeader() {
        return currentlyAmLeader;
    }
    
    public void ensureTable() throws SQLException {
        try (Connection connnection = this.dataSource.getConnection()) {
            if (!SQLUtils.tableExists(connnection, configuration.getSchemaName(), configuration.getTableName())) {
                configuration.getLeaderElectorLogger().logInfo(this.getClass(), "Creating table " + tableNameDisplay);
                
                try (PreparedStatement createTableStmt = sqlCmds.getCreateTableStmt(connnection)) {
                    createTableStmt.execute();
                } catch (SQLException ex) {
                    if (!sqlCmds.isTableAlreadyExistException(ex)) {
                        throw ex;
                    }
                }
            }
        }
    }
    
    public void ensureRoleRow() throws SQLException {
        try (Connection connnection = this.dataSource.getConnection()) {
            try (PreparedStatement p = sqlCmds.getInsertRoleStmt(connnection, this.configuration.getRoleId())) {
                p.execute();
            }
        }
    }
    
    private void affirmLeadership(Connection connection, String roleId, String candidateId) 
            throws SQLException, LeaderElectorExceptionNonRecoverable {
        try (PreparedStatement pstmt = sqlCmds.getAffirmLeadershipStmt(connection, roleId, candidateId)) {
            executeUpdate(pstmt);
        }
    }
    
    private void assumeLeadership(Connection connection, String roleId, String candidateId, long newLeaseCounter) 
            throws SQLException, LeaderElectorExceptionNonRecoverable {
        try (PreparedStatement pstmt = sqlCmds.getAssumeLeadershipStmt(connection, roleId, candidateId, newLeaseCounter)) {
            executeUpdate(pstmt);
        }
    }
    
    private void relinquishLeadership(Connection connection, String roleId, String candidateId) 
            throws SQLException, LeaderElectorExceptionNonRecoverable {
        try (PreparedStatement pstmt = sqlCmds.getRelinquishLeadershipStmt(connection, roleId, candidateId)) {
            executeUpdate(pstmt);
        }
    }
    
    private void executeUpdate(PreparedStatement pstmt) 
            throws SQLException, LeaderElectorExceptionNonRecoverable {
        int rowsAffected = pstmt.executeUpdate();
        if (rowsAffected != 1) {
            throw new LeaderElectorExceptionNonRecoverable(rowsAffected + " rows was affected by UPDATE statement. Expected exactly 1 (one) row to be affected.");
        }
    }
    
    private long getNewLeaseCounter(long existingLeaseCounter) {
        return (existingLeaseCounter == Long.MAX_VALUE) ? 0 : existingLeaseCounter + 1;
    }
    
    public LeaderElectorListener.Event electLeader(boolean relinquish) {
        boolean wasLeaderAtStartOfElection = currentlyAmLeader;
        EventHelpers.ErrorEventsBuilder errorHolder = null;
        LeaderElectorListener.Event event = null;
        Instant startTime = Instant.now();

        try ( Connection connection = dataSource.getConnection()) {
            boolean originalAutoCommit = connection.getAutoCommit();
            connection.setAutoCommit(false);

            try ( PreparedStatement preparedStatement = sqlCmds.getSelectStmt(connection, myRoleId)) {
                preparedStatement.setQueryTimeout(configuration.getQueryTimeoutSecs());

                try ( ResultSet rs = preparedStatement.executeQuery()) {
                    event = executeInsideTableLock(connection, rs, wasLeaderAtStartOfElection, relinquish);
                }
                connection.commit();
                noOfConsecutiveTransientErrors = 0;
            } catch (SQLTransientException | SQLRecoverableException ex) {
                noOfConsecutiveTransientErrors++;
                if (noOfConsecutiveTransientErrors == 3) {
                    noOfConsecutiveTransientErrors = 0;
                    errorHolder = addError(errorHolder, new LeaderElectorExceptionNonRecoverable(ex));
                } else {
                    errorHolder = addError(errorHolder, new LeaderElectorExceptionRecoverable(ex));
                }
            } catch (Exception ex) {
                errorHolder = addError(errorHolder, new LeaderElectorExceptionNonRecoverable(ex));
                try {
                    connection.rollback();
                } catch (SQLException ex2) {
                    errorHolder = addError(errorHolder, new LeaderElectorExceptionNonRecoverable("Error performing rollback", ex2));
                }
            } finally {
                try {
                    connection.setAutoCommit(originalAutoCommit);
                } catch (SQLException ex) {
                    errorHolder = addError(errorHolder, new LeaderElectorExceptionNonRecoverable("Error resetting auto-commit", ex));
                }
            }
        } catch (SQLTransientException ex) {
            if (hasHadSuccessfulExection) {
                errorHolder = addError(errorHolder, new LeaderElectorExceptionRecoverable("No longer able to connect to database", ex));
            } else {
                errorHolder = addError(errorHolder, new LeaderElectorExceptionNonRecoverable("Cannot connect to database (first time)", ex));
            }
        } catch (Exception ex) {
            errorHolder = addError(errorHolder, new LeaderElectorExceptionNonRecoverable(ex));
        }

        if (errorHolder != null) {
            currentlyAmLeader = false;
            return errorHolder.build(startTime, wasLeaderAtStartOfElection, myRoleId);
        } else if (event != null) {
            if (!hasHadSuccessfulExection) {
                hasHadSuccessfulExection = true;
            }
            return event;
        }

        throw new RuntimeException("Unexpected event. Neither 'event' nor 'errorHolder' has a value");
    }

    private LeaderElectorListener.Event executeInsideTableLock(
            Connection connection, 
            ResultSet rs, 
            boolean wasLeaderAtStartOfElection,
            boolean relinquish) throws SQLException, LeaderElectorExceptionNonRecoverable {
        Instant startTime = Instant.now();
        LeaderElectorListener.Event event = null;

        int rows = 0;

        while (rs.next()) {
            rows++;
            if (rows > 1) {
                throw new LeaderElectorExceptionNonRecoverable("Table " + tableNameDisplay + " has more than one row. This is unexpected. It must contain exactly one row.");
            }
            final RowInLockTable row = new RowInLockTable(rs, myCandidateId);
            final long lastSeenTimestampMillis = row.getLastSeenTimestampMillis();
            final long nowUTCMillis = row.getNowUTCMillis();
            final long leaseCounter = row.getLeaseCounter();
            final RowInLockTable.CurrentLeaderDbStatus currentLeader = row.getCurrentLeaderDbStatus();
            final long leaseAgeMillis = nowUTCMillis - lastSeenTimestampMillis;
            final boolean leaseExpired = (leaseAgeMillis >= configuration.getAssumeDeadMs());

            checkRowValidity(row);

            switch (currentLeader) {
                case ME: {
                    if (relinquish) {
                        relinquishLeadership(connection, myRoleId, myCandidateId);
                        currentlyAmLeader = false;
                        hasRelinquishedLeadership = true;
                        if (wasLeaderAtStartOfElection) {
                            event = EventHelpers.createLeadershipLostEvent(startTime, myRoleId, row.getNodeId(), lastSeenTimestampMillis, leaseCounter);
                        }
                    } else {
                        affirmLeadership(connection, myRoleId, myCandidateId);
                        event = EventHelpers.createLeadershipConfirmedEvent(startTime, myRoleId, myCandidateId, lastSeenTimestampMillis, leaseCounter);
                    }
                }
                break;
                case SOMEONE_ELSE:
                    if (!leaseExpired) {
                        hasRelinquishedLeadership = false;
                    } 
                case NOBODY: {
                    if (leaseExpired && (!hasRelinquishedLeadership)) {
                        long newLeaseCounter = getNewLeaseCounter(leaseCounter);
                        assumeLeadership(connection, myRoleId, myCandidateId, newLeaseCounter);
                        event = EventHelpers.createLeadershipAssumedEvent(startTime, myRoleId, row.getNodeId(), lastSeenTimestampMillis, newLeaseCounter);
                        if (!wasLeaderAtStartOfElection) {
                            currentlyAmLeader = true;
                        }
                    } else {
                        event = EventHelpers.createLeadershipNoOp(startTime, myRoleId, row.getNodeId(), lastSeenTimestampMillis, leaseCounter);
                    }
                }
                break;
                default:
                    throw new LeaderElectorExceptionNonRecoverable("Unexpected value for 'currentLeader' : " + currentLeader);
            }

            if (event == null) {
                event = EventHelpers.createLeadershipNoOp(startTime, myRoleId, row.getNodeId(), lastSeenTimestampMillis, leaseCounter);
            }
        }
        
        if (rows == 0) {
            throw new LeaderElectorExceptionNonRecoverable("No row for lock_id='" + myRoleId + "' in table " + tableNameDisplay);
        }
        
        return event;
    }
                    
    private void checkRowValidity(RowInLockTable row) throws LeaderElectorExceptionNonRecoverable {
        String prefix = "ERROR: Unexpected: Table " + tableNameDisplay + " with content " + row ;
        if (row.getCurrentLeaderDbStatus()== CurrentLeaderDbStatus.ME && (!currentlyAmLeader)) {
            throw new LeaderElectorExceptionNonRecoverable(prefix
                    + ", says current candidate is leader but 'currentlyAmLeader' is false. "
                    + "Possibly table content was altered by an unsolicated process.");
        }

        if (row.getCurrentLeaderDbStatus() != CurrentLeaderDbStatus.ME && (currentlyAmLeader)) {
            throw new LeaderElectorExceptionNonRecoverable(prefix
                    + ", says current \"" + row.getNodeId() + "\" is leader, not me, but 'currentlyAmLeader' is true. "
                    + "In effect leadership was stolen. "        
                    + "Possible cause is if current process has not kept its lease alive. Perhaps the process has been dormant? "
                    + "Another possible cause is if candidates do not use the same configuration values for their Leader Elector process "
                    + "(for example, they use different values for 'assumeDeadMs')"
            );
        }
        if (row.getCurrentLeaderDbStatus() == CurrentLeaderDbStatus.NOBODY && row.getLastSeenTimestampMillis() != NO_LEADER_LASTSEENTIMESTAMP_MS) {
            throw new LeaderElectorExceptionNonRecoverable(prefix
                    + ", is inconsistent. "
                    + "Possibly table content was altered by an unsolicated process.");
        }
    }
    
    private EventHelpers.ErrorEventsBuilder addError(final EventHelpers.ErrorEventsBuilder errorEventsBuilder, LeaderElectorException error) {
        EventHelpers.ErrorEventsBuilder e = (errorEventsBuilder == null) ? new EventHelpers.ErrorEventsBuilder() : errorEventsBuilder;
        e.add(error);
        return e;
    }
    
}