/*
 * 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.bookkeeper.metastore;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.util.NavigableMap;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import org.apache.bookkeeper.metastore.MSException.Code;
import org.apache.bookkeeper.versioning.Version;
import org.apache.bookkeeper.versioning.Versioned;

/**
 * An in-memory implementation of a Metastore table.
 */
public class InMemoryMetastoreTable implements MetastoreScannableTable {

    /**
     * An implementation of the Version interface for metadata.
     */
    public static class MetadataVersion implements Version {
        int version;

        public MetadataVersion(int v) {
            this.version = v;
        }

        public MetadataVersion(MetadataVersion v) {
            this.version = v.version;
        }

        public synchronized MetadataVersion incrementVersion() {
            ++version;
            return this;
        }

        @Override
        public Occurred compare(Version v) {
            if (null == v) {
                throw new NullPointerException("Version is not allowed to be null.");
            }
            if (v == Version.NEW) {
                return Occurred.AFTER;
            } else if (v == Version.ANY) {
                return Occurred.CONCURRENTLY;
            } else if (!(v instanceof MetadataVersion)) {
                throw new IllegalArgumentException("Invalid version type");
            }
            MetadataVersion mv = (MetadataVersion) v;
            int res = version - mv.version;
            if (res == 0) {
                return Occurred.CONCURRENTLY;
            } else if (res < 0) {
                return Occurred.BEFORE;
            } else {
                return Occurred.AFTER;
            }
        }

        @Override
        public boolean equals(Object obj) {
            if (null == obj || !(obj instanceof MetadataVersion)) {
                return false;
            }
            MetadataVersion v = (MetadataVersion) obj;
            return 0 == (version - v.version);
        }

        @Override
        public String toString() {
            return "version=" + version;
        }

        @Override
        public int hashCode() {
            return version;
        }
    }

    private String name;
    private TreeMap<String, Versioned<Value>> map = null;
    private TreeMap<String, MetastoreWatcher> watcherMap = null;
    private ScheduledExecutorService scheduler;

    public InMemoryMetastoreTable(InMemoryMetaStore metastore, String name) {
        this.map = new TreeMap<String, Versioned<Value>>();
        this.watcherMap = new TreeMap<String, MetastoreWatcher>();
        this.name = name;
        String thName = "InMemoryMetastore-Table(" + name + ")-Scheduler-%d";
        ThreadFactoryBuilder tfb = new ThreadFactoryBuilder()
                .setNameFormat(thName);
        this.scheduler = Executors
                .newSingleThreadScheduledExecutor(tfb.build());
    }

    @Override
    public String getName () {
        return this.name;
    }

    static Versioned<Value> cloneValue(Value value, Version version, Set<String> fields) {
        if (null != value) {
            Value newValue = new Value();
            if (ALL_FIELDS == fields) {
                fields = value.getFields();
            }
            for (String f : fields) {
                newValue.setField(f, value.getField(f));
            }
            value = newValue;
        }

        if (null == version) {
            throw new NullPointerException("Version isn't allowed to be null.");
        }
        if (Version.ANY != version && Version.NEW != version) {
            if (version instanceof MetadataVersion) {
                version = new MetadataVersion(((MetadataVersion) version).version);
            } else {
                throw new IllegalStateException("Wrong version type.");
            }
        }
        return new Versioned<Value>(value, version);
    }

    @Override
    public void get(final String key, final MetastoreCallback<Versioned<Value>> cb, final Object ctx) {
        scheduler.submit(new Runnable() {
            @Override
            public void run() {
                scheduleGet(key, ALL_FIELDS, cb, ctx);
            }
        });
    }

    @Override
    public void get(final String key, final MetastoreWatcher watcher, final MetastoreCallback<Versioned<Value>> cb,
            final Object ctx) {
        scheduler.submit(new Runnable() {
            @Override
            public void run() {
                scheduleGet(key, ALL_FIELDS, cb, ctx);
                synchronized (watcherMap) {
                    watcherMap.put(key, watcher);
                }
            }
        });
    }

    @Override
    public void get(final String key, final Set<String> fields, final MetastoreCallback<Versioned<Value>> cb,
            final Object ctx) {
        scheduler.submit(new Runnable() {
            @Override
            public void run() {
                scheduleGet(key, fields, cb, ctx);
            }
        });
    }

    public synchronized void scheduleGet(String key, Set<String> fields, MetastoreCallback<Versioned<Value>> cb,
            Object ctx) {
        if (null == key) {
            cb.complete(Code.IllegalOp.getCode(), null, ctx);
            return;
        }
        Versioned<Value> vv = get(key);
        int rc = null == vv ? Code.NoKey.getCode() : Code.OK.getCode();
        if (vv != null) {
            vv = cloneValue(vv.getValue(), vv.getVersion(), fields);
        }
        cb.complete(rc, vv, ctx);
    }

    @Override
    public void put(final String key, final Value value, final Version version, final MetastoreCallback<Version> cb,
            final Object ctx) {
        scheduler.submit(new Runnable() {
            @Override
            public void run() {
                if (null == key || null == value || null == version) {
                    cb.complete(Code.IllegalOp.getCode(), null, ctx);
                    return;
                }
                Result<Version> result = put(key, value, version);
                cb.complete(result.code.getCode(), result.value, ctx);

                /*
                 * If there is a watcher set for this key, we need
                 * to trigger it.
                 */
                if (result.code == MSException.Code.OK) {
                    triggerWatch(key, MSWatchedEvent.EventType.CHANGED);
                }
            }
        });
    }

    @Override
    public void remove(final String key, final Version version, final MetastoreCallback<Void> cb, final Object ctx) {
        scheduler.submit(new Runnable() {
            @Override
            public void run() {
                if (null == key || null == version) {
                    cb.complete(Code.IllegalOp.getCode(), null, ctx);
                    return;
                }
                Code code = remove(key, version);
                cb.complete(code.getCode(), null, ctx);

                if (code == MSException.Code.OK) {
                    triggerWatch(key, MSWatchedEvent.EventType.REMOVED);
                }
            }
        });
    }

    @Override
    public void openCursor(MetastoreCallback<MetastoreCursor> cb, Object ctx) {
        openCursor(EMPTY_START_KEY, true, EMPTY_END_KEY, true, Order.ASC,
                   ALL_FIELDS, cb, ctx);
    }

    @Override
    public void openCursor(Set<String> fields,
                           MetastoreCallback<MetastoreCursor> cb, Object ctx) {
        openCursor(EMPTY_START_KEY, true, EMPTY_END_KEY, true, Order.ASC,
                   fields, cb, ctx);
    }

    @Override
    public void openCursor(String firstKey, boolean firstInclusive,
                           String lastKey, boolean lastInclusive,
                           Order order, MetastoreCallback<MetastoreCursor> cb,
                           Object ctx) {
        openCursor(firstKey, firstInclusive, lastKey, lastInclusive,
                   order, ALL_FIELDS, cb, ctx);
    }

    @Override
    public void openCursor(final String firstKey, final boolean firstInclusive,
                           final String lastKey, final boolean lastInclusive,
                           final Order order, final Set<String> fields,
                           final MetastoreCallback<MetastoreCursor> cb, final Object ctx) {
        scheduler.submit(new Runnable() {
            @Override
            public void run() {
                Result<MetastoreCursor> result = openCursor(firstKey, firstInclusive, lastKey, lastInclusive,
                        order, fields);
                cb.complete(result.code.getCode(), result.value, ctx);
            }
        });
    }

    private void triggerWatch(String key, MSWatchedEvent.EventType type) {
        synchronized (watcherMap){
            if (watcherMap.containsKey(key)) {
                MSWatchedEvent event = new MSWatchedEvent(key, type);
                watcherMap.get(key).process(event);
                watcherMap.remove(key);
            }
        }
    }

    private synchronized Versioned<Value> get(String key) {
        return map.get(key);
    }

    private synchronized Code remove(String key, Version version) {
        Versioned<Value> vv = map.get(key);
        if (null == vv) {
            return Code.NoKey;
        }
        if (Version.Occurred.CONCURRENTLY != vv.getVersion().compare(version)) {
            return Code.BadVersion;
        }
        map.remove(key);
        return Code.OK;
    }

    static class Result<T> {
        Code code;
        T value;

        public Result(Code code, T value) {
            this.code = code;
            this.value = value;
        }
    }

    private synchronized Result<Version> put(String key, Value value, Version version) {
        Versioned<Value> vv = map.get(key);
        if (vv == null) {
            if (Version.NEW != version) {
                return new Result<Version>(Code.NoKey, null);
            }
            vv = cloneValue(value, version, ALL_FIELDS);
            vv.setVersion(new MetadataVersion(0));
            map.put(key, vv);
            return new Result<Version>(Code.OK, new MetadataVersion(0));
        }
        if (Version.NEW == version) {
            return new Result<Version>(Code.KeyExists, null);
        }
        if (Version.Occurred.CONCURRENTLY != vv.getVersion().compare(version)) {
            return new Result<Version>(Code.BadVersion, null);
        }
        vv.setVersion(((MetadataVersion) vv.getVersion()).incrementVersion());
        vv.setValue(vv.getValue().merge(value));
        return new Result<Version>(Code.OK, new MetadataVersion((MetadataVersion) vv.getVersion()));
    }

    private synchronized Result<MetastoreCursor> openCursor(
            String firstKey, boolean firstInclusive,
            String lastKey, boolean lastInclusive,
            Order order, Set<String> fields) {
        if (0 == map.size()) {
            return new Result<MetastoreCursor>(Code.OK, MetastoreCursor.EMPTY_CURSOR);
        }

        boolean isLegalCursor = false;
        NavigableMap<String, Versioned<Value>> myMap = null;
        if (Order.ASC == order) {
            myMap = map;
            if (EMPTY_END_KEY == lastKey || lastKey.compareTo(myMap.lastKey()) > 0) {
                lastKey = myMap.lastKey();
                lastInclusive = true;
            }
            if (EMPTY_START_KEY == firstKey || firstKey.compareTo(myMap.firstKey()) < 0) {
                firstKey = myMap.firstKey();
                firstInclusive = true;
            }
            if (firstKey.compareTo(lastKey) <= 0) {
                isLegalCursor = true;
            }
        } else if (Order.DESC == order) {
            myMap = map.descendingMap();
            if (EMPTY_START_KEY == lastKey || lastKey.compareTo(myMap.lastKey()) < 0) {
                lastKey = myMap.lastKey();
                lastInclusive = true;
            }
            if (EMPTY_END_KEY == firstKey || firstKey.compareTo(myMap.firstKey()) > 0) {
                firstKey = myMap.firstKey();
                firstInclusive = true;
            }
            if (firstKey.compareTo(lastKey) >= 0) {
                isLegalCursor = true;
            }
        }

        if (!isLegalCursor || null == myMap) {
            return new Result<MetastoreCursor>(Code.IllegalOp, null);
        }
        MetastoreCursor cursor = new InMemoryMetastoreCursor(
                myMap.subMap(firstKey, firstInclusive, lastKey, lastInclusive), fields, scheduler);
        return new Result<MetastoreCursor>(Code.OK, cursor);
    }

    @Override
    public void close() {
        // do nothing
    }
}
