/*
 * Copyright (c) 2013-2015 Cinchapi, Inc.
 * 
 * Licensed 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.cinchapi.concourse.server.storage.db;

import java.io.File;
import java.lang.reflect.Field;
import java.util.Iterator;
import java.util.List;

import org.cinchapi.concourse.server.io.FileSystem;
import org.cinchapi.concourse.server.model.PrimaryKey;
import org.cinchapi.concourse.server.model.Text;
import org.cinchapi.concourse.server.model.Value;
import org.cinchapi.concourse.server.storage.Store;
import org.cinchapi.concourse.server.storage.StoreTest;
import org.cinchapi.concourse.server.storage.temp.Write;
import org.cinchapi.concourse.thrift.Operator;
import org.cinchapi.concourse.thrift.TObject;
import org.cinchapi.concourse.time.Time;
import org.cinchapi.concourse.util.Convert;
import org.cinchapi.concourse.util.TestData;
import org.junit.Assert;
import org.junit.Test;

import com.google.common.collect.Lists;

/**
 * Unit tests for the {@link Database}.
 * 
 * @author Jeff Nelson
 */
public class DatabaseTest extends StoreTest {

    private String current;

    @Test
    public void testDatabaseRemovesUnbalancedBlocksOnStartup() throws Exception {
        Database db = (Database) store;
        db.accept(Write.add(TestData.getString(), TestData.getTObject(),
                TestData.getLong()));
        db.triggerSync();
        db.stop();
        FileSystem.deleteDirectory(current + File.separator + "csb");
        FileSystem.mkdirs(current + File.separator + "csb");
        db = new Database(db.getBackingStore()); // simulate server restart
        db.start();
        Field cpb = db.getClass().getDeclaredField("cpb");
        Field csb = db.getClass().getDeclaredField("csb");
        Field ctb = db.getClass().getDeclaredField("ctb");
        cpb.setAccessible(true);
        csb.setAccessible(true);
        ctb.setAccessible(true);
        Assert.assertEquals(1, ((List<?>) ctb.get(db)).size());
        Assert.assertEquals(1, ((List<?>) csb.get(db)).size());
        Assert.assertEquals(1, ((List<?>) cpb.get(db)).size());
    }

    @Test
    public void testDatabaseAppendsToCachedPartialPrimaryRecords() {
        Database db = (Database) store;
        String key = TestData.getString();
        long record = TestData.getLong();
        int count = TestData.getScaleCount();
        for (int i = 0; i < count; i++) {
            db.accept(Write.add(key, Convert.javaToThrift(i), record));
        }
        db.select(key, record);
        int increase = TestData.getScaleCount();
        db.accept(Write.add(key, Convert.javaToThrift(count * increase), record));
        Assert.assertTrue(db.select(key, record).contains(
                Convert.javaToThrift(count * increase)));
    }

    @Test
    public void testDatabaseAppendsToCachedSecondaryRecords() {
        Database db = (Database) store;
        String key = TestData.getString();
        TObject value = TestData.getTObject();
        int count = TestData.getScaleCount();
        for (int i = 0; i < count; i++) {
            db.accept(Write.add(key, value, i));
        }
        db.find(key, Operator.EQUALS, value);
        int increase = TestData.getScaleCount();
        db.accept(Write.add(key, value, count * increase));
        Assert.assertTrue(db.find(key, Operator.EQUALS, value).contains(
                (long) count * increase));
    }

    @Test
    public void testOnDiskStreamingIterator() {
        Database db = (Database) store;
        int count = TestData.getScaleCount() * 5;
        List<Revision<PrimaryKey, Text, Value>> expected = Lists
                .newArrayListWithCapacity(count);
        for (int i = 0; i < count; ++i) {
            Write write = Write.add(TestData.getSimpleString(),
                    TestData.getTObject(), i);
            db.accept(write);
            Revision<PrimaryKey, Text, Value> revision = Revision
                    .createPrimaryRevision(write.getRecord(), write.getKey(),
                            write.getValue(), write.getVersion(),
                            write.getType());
            expected.add(revision);
            if(i % 100 == 0){
                db.triggerSync();
            }
        }
        db.triggerSync();
        List<Revision<PrimaryKey, Text, Value>> stored = Lists
                .newArrayListWithCapacity(count);
        Iterator<Revision<PrimaryKey, Text, Value>> it = Database
                .onDiskStreamingIterator(db.getBackingStore());
        while (it.hasNext()) {
            stored.add(it.next());
        }
        Assert.assertEquals(expected, stored);
    }

    @Override
    protected void add(String key, TObject value, long record) {
        if(!store.verify(key, value, record)) {
            ((Database) store).accept(Write.add(key, value, record));
        }
    }

    @Override
    protected void cleanup(Store store) {
        FileSystem.deleteDirectory(current);
    }

    @Override
    protected Database getStore() {
        current = TestData.DATA_DIR + File.separator + Time.now();
        return new Database(current);
    }

    @Override
    protected void remove(String key, TObject value, long record) {
        if(store.verify(key, value, record)) {
            ((Database) store).accept(Write.remove(key, value, record));
        }
    }

}

@@ -752,16 +752,38 @@
 il.List;
+%0Aimport java.util.Set;
 %0A%0Aimport
@@ -1452,19 +1452,18 @@
 collect.
-Lis
+Se
 ts;%0A%0A/**
@@ -4039,35 +4039,34 @@
 () * 5;%0A        
-Lis
+Se
 t%3CRevision%3CPrima
@@ -4093,27 +4093,26 @@
  expected = 
-Lis
+Se
 ts%0A         
@@ -4122,37 +4122,45 @@
     .new
-ArrayListWithCapacity
+LinkedHashSetWithExpectedSize
 (count);
@@ -4663,16 +4663,17 @@
 00 == 0)
+ 
 %7B%0A      
@@ -4754,121 +4754,8 @@
 ();%0A
-        List%3CRevision%3CPrimaryKey, Text, Value%3E%3E stored = Lists%0A                .newArrayListWithCapacity(count);%0A
     
@@ -4892,122 +4892,175 @@
     
-while (it.hasNext()) %7B%0A            stored.add(it.next());%0A        %7D%0A        Assert.assertEquals(expected, stored);
+Iterator%3CRevision%3CPrimaryKey, Text, Value%3E%3E it2 = expected.iterator();%0A        while (it.hasNext()) %7B%0A            Assert.assertEquals(it.next(), it2.next());%0A        %7D
 %0A   

test for consistent ordering