/*
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
 * agreements. See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance with the License. You may obtain a
 * copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */

package org.apache.fluo.accumulo.iterators;

import java.io.IOException;
import java.util.Collection;
import java.util.Map;

import org.apache.accumulo.core.client.IteratorSetting;
import org.apache.accumulo.core.data.ByteSequence;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.PartialKey;
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.iterators.IteratorEnvironment;
import org.apache.accumulo.core.iterators.IteratorUtil;
import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
import org.apache.fluo.accumulo.util.ColumnConstants;
import org.apache.fluo.accumulo.util.ColumnType;
import org.apache.fluo.accumulo.values.WriteValue;

public class RollbackCheckIterator implements SortedKeyValueIterator<Key, Value> {
  private static final String TIMESTAMP_OPT = "timestampOpt";

  private TimestampSkippingIterator source;
  private long lockTime;

  boolean hasTop = false;
  boolean checkAck = false;

  public static void setLocktime(IteratorSetting cfg, long time) {
    if (time < 0 || (ColumnConstants.PREFIX_MASK & time) != 0) {
      throw new IllegalArgumentException();
    }
    cfg.addOption(TIMESTAMP_OPT, time + "");
  }

  @Override
  public void init(SortedKeyValueIterator<Key, Value> source, Map<String, String> options,
      IteratorEnvironment env) throws IOException {
    this.source = new TimestampSkippingIterator(source);
    this.lockTime = Long.parseLong(options.get(TIMESTAMP_OPT));
  }

  @Override
  public boolean hasTop() {
    return hasTop && source.hasTop();
  }

  @Override
  public void next() throws IOException {
    hasTop = false;
  }

  @Override
  public void seek(Range range, Collection<ByteSequence> columnFamilies, boolean inclusive)
      throws IOException {
    range = IteratorUtil.maximizeStartKeyTimeStamp(range);

    if (columnFamilies.isEmpty() && !inclusive) {
      source.seek(range, SnapshotIterator.NOTIFY_CF_SET, false);
    } else {
      source.seek(range, columnFamilies, inclusive);
    }

    Key curCol = new Key();

    if (source.hasTop()) {
      curCol.set(source.getTopKey());

      // TODO can this optimization cause problems?
      if (!curCol.equals(range.getStartKey(), PartialKey.ROW_COLFAM_COLQUAL_COLVIS)) {
        return;
      }
    }

    long invalidationTime = -1;

    hasTop = false;
    while (source.hasTop()
        && curCol.equals(source.getTopKey(), PartialKey.ROW_COLFAM_COLQUAL_COLVIS)) {
      ColumnType colType = ColumnType.from(source.getTopKey());
      long ts = source.getTopKey().getTimestamp() & ColumnConstants.TIMESTAMP_MASK;

      switch (colType) {
        case TX_DONE:
          source.skipToPrefix(curCol, ColumnType.WRITE);
          continue;
        case WRITE: {
          long timePtr = WriteValue.getTimestamp(source.getTopValue().get());

          if (timePtr > invalidationTime) {
            invalidationTime = timePtr;
          }

          if (lockTime == timePtr) {
            hasTop = true;
            return;
          }

          if (lockTime > timePtr) {
            source.skipToPrefix(curCol, ColumnType.DEL_LOCK);
            continue;
          }
          break;
        }
        case DEL_LOCK: {
          if (ts > invalidationTime) {
            invalidationTime = ts;
          }

          if (ts == lockTime) {
            hasTop = true;
            return;
          }

          if (lockTime > ts) {
            source.skipToPrefix(curCol, ColumnType.LOCK);
            continue;
          }
          break;
        }
        case RLOCK: {
          source.skipToPrefix(curCol, ColumnType.LOCK);
          continue;
        }
        case LOCK: {
          if (ts > invalidationTime) {
            // nothing supersedes this lock, therefore the column is locked
            hasTop = true;
            return;
          }
          break;
        }
        case DATA: {
          // can stop looking
          return;
        }
        case ACK: {
          // do nothing if ACK
          break;
        }
        default:
          throw new IllegalArgumentException();
      }

      source.next();
    }
  }

  @Override
  public Key getTopKey() {
    return source.getTopKey();
  }

  @Override
  public Value getTopValue() {
    return source.getTopValue();
  }

  @Override
  public SortedKeyValueIterator<Key, Value> deepCopy(IteratorEnvironment env) {
    // TODO Auto-generated method stub
    return null;
  }
}
