/*
 * Copyright (c) 2015 Cisco Systems, Inc. and others.  All rights reserved.
 *
 * This program and the accompanying materials are made available under the
 * terms of the Eclipse Public License v1.0 which accompanies this distribution,
 * and is available at http://www.eclipse.org/legal/epl-v10.html
 */

package org.opendaylight.controller.cluster.raft.behaviors;

import akka.actor.ActorRef;
import com.google.common.base.Preconditions;
import org.opendaylight.controller.cluster.raft.base.messages.FollowerInitialSyncUpStatus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The SyncStatusTracker tracks if a Follower is in sync with any given Leader or not
 * When an update is received from the Leader and the update happens to be the first update
 * from that Leader then the SyncStatusTracker will not mark the Follower as not in-sync till the
 * Followers commitIndex matches the commitIndex that the Leader sent in it's very first update.
 * Subsequently when an update is received the tracker will consider the Follower to be out of
 * sync if it is behind by 'syncThreshold' commits.
 */
public class SyncStatusTracker {
    private static final class LeaderInfo {
        final long minimumCommitIndex;
        final String leaderId;

        LeaderInfo(final String leaderId, final long minimumCommitIndex) {
            this.leaderId = Preconditions.checkNotNull(leaderId);
            this.minimumCommitIndex = minimumCommitIndex;
        }
    }

    private static final Logger LOG = LoggerFactory.getLogger(SyncStatusTracker.class);

    private static final boolean IN_SYNC = true;
    private static final boolean NOT_IN_SYNC = false;

    private final String id;
    private final ActorRef actor;
    private final int syncThreshold;

    private LeaderInfo syncTarget;
    private boolean syncStatus;

    public SyncStatusTracker(final ActorRef actor, final String id, final int syncThreshold) {
        this.actor = Preconditions.checkNotNull(actor, "actor should not be null");
        this.id = Preconditions.checkNotNull(id, "id should not be null");
        Preconditions.checkArgument(syncThreshold >= 0, "syncThreshold should be greater than or equal to 0");
        this.syncThreshold = syncThreshold;
    }

    public void update(final String leaderId, final long leaderCommit, final long commitIndex) {
        Preconditions.checkNotNull(leaderId, "leaderId should not be null");

        if (syncTarget == null || !leaderId.equals(syncTarget.leaderId)) {
            LOG.debug("Last sync leader does not match current leader {}, need to catch up to {}",
                leaderId, leaderCommit);
            changeSyncStatus(NOT_IN_SYNC, true);
            syncTarget = new LeaderInfo(leaderId, leaderCommit);
            return;
        }

        final long lag = leaderCommit - commitIndex;
        if (lag > syncThreshold) {
            LOG.debug("Lagging {} entries behind leader {}", lag, leaderId);
            changeSyncStatus(NOT_IN_SYNC, false);
        } else if (commitIndex >= syncTarget.minimumCommitIndex) {
            LOG.debug("Lagging {} entries behind leader and reached {} (of expected {})", lag, leaderId, commitIndex,
                syncTarget.minimumCommitIndex);
            changeSyncStatus(IN_SYNC, false);
        }
    }

    private void changeSyncStatus(final boolean newSyncStatus, final boolean forceStatusChange) {
        if (forceStatusChange || newSyncStatus != syncStatus) {
            actor.tell(new FollowerInitialSyncUpStatus(newSyncStatus, id), ActorRef.noSender());
            syncStatus = newSyncStatus;
        } else {
            LOG.trace("No change in sync status of {}, dampening message", actor);
        }
    }
}

@@ -1688,24 +1688,33 @@
 e final 
-String i
+long syncThreshol
 d;%0A    p
@@ -1760,32 +1760,24 @@
 e final 
-int syncThreshol
+String i
 d;%0A%0A    
@@ -1918,11 +1918,12 @@
 nal 
-int
+long
  syn
@@ -2530,16 +2530,20 @@
 .debug(%22
+%7B%7D: 
 Last syn
@@ -2609,16 +2609,20 @@
  to %7B%7D%22,
+ id,
 %0A       
@@ -2903,32 +2903,36 @@
      LOG.debug(%22
+%7B%7D: 
 Lagging %7B%7D entri
@@ -2952,16 +2952,20 @@
 der %7B%7D%22,
+ id,
  lag, le
@@ -3113,16 +3113,20 @@
 .debug(%22
+%7B%7D: 
 Lagging 
@@ -3183,16 +3183,20 @@
 ed %7B%7D)%22,
+ id,
  lag, le
@@ -3206,21 +3206,8 @@
 rId,
- commitIndex,
 %0A   
@@ -3210,32 +3210,45 @@
 %0A               
+ commitIndex,
  syncTarget.mini
@@ -3666,16 +3666,20 @@
 .trace(%22
+%7B%7D: 
 No chang
@@ -3697,19 +3697,16 @@
 tatus of
- %7B%7D
 , dampen
@@ -3719,21 +3719,18 @@
 ssage%22, 
-actor
+id
 );%0A     

improve debug logs