/*
 * JBoss, Home of Professional Open Source.
 * Copyright 2009, Red Hat Middleware LLC, and individual contributors
 * as indicated by the @author tags. See the copyright.txt file in the
 * distribution for a full listing of individual contributors.
 *
 * This is free software; you can redistribute it and/or modify it
 * under the terms of the GNU Lesser General Public License as
 * published by the Free Software Foundation; either version 2.1 of
 * the License, or (at your option) any later version.
 *
 * This software is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this software; if not, write to the Free
 * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
 * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
 */
package org.infinispan.schematic.internal;

import java.util.Collection;
import org.infinispan.AdvancedCache;
import org.infinispan.Cache;
import org.infinispan.context.Flag;
import org.infinispan.schematic.SchematicDb;
import org.infinispan.schematic.SchematicEntry;
import org.infinispan.schematic.SchematicEntry.FieldName;
import org.infinispan.schematic.document.Document;
import org.infinispan.schematic.document.EditableDocument;
import org.infinispan.schematic.internal.document.DocumentEditor;
import org.infinispan.schematic.internal.document.MutableDocument;
import org.infinispan.transaction.LockingMode;

public class CacheSchematicDb implements SchematicDb {

    private final String name;
    private final AdvancedCache<String, SchematicEntry> store;
    private final AdvancedCache<String, SchematicEntry> lockingStore;
    private final boolean explicitLocking;

    public CacheSchematicDb( AdvancedCache<String, SchematicEntry> store ) {
        this.store = store;
        this.name = store.getName();
        this.explicitLocking = store.getCacheConfiguration().transaction().lockingMode() == LockingMode.PESSIMISTIC;
        if (this.explicitLocking) {
            // the FAIL SILENTLY flag is required here because without it ISPN will rollback the active transaction on the first
            // TimeoutException, while ModeShape has built-in logic for retrying....
            this.lockingStore = store.withFlags(Flag.FAIL_SILENTLY).getAdvancedCache();
        } else {
            this.lockingStore = store;
        }
    }

    @Override
    public void start() {
        this.store.start();
    }

    @Override
    public void stop() {
        this.store.stop();
    }

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

    @Override
    public Cache<String, SchematicEntry> getCache() {
        return store;
    }

    @Override
    public SchematicEntry get( String key ) {
        return store.get(key);
    }

    @Override
    public boolean containsKey( String key ) {
        return store.containsKey(key);
    }

    @Override
    public SchematicEntry put( String key,
                               Document document ) {
        SchematicEntry newEntry = new SchematicEntryLiteral(key, document);
        SchematicEntry oldValue = store.put(key, newEntry);
        return oldValue != null ? removedResult(key, oldValue) : null;
    }

    protected SchematicEntry removedResult( String key,
                                            SchematicEntry entry ) {
        SchematicEntryLiteral literal = (SchematicEntryLiteral)entry;
        literal.markRemoved(true);
        return literal;
    }

    @Override
    public SchematicEntry put( Document entryDocument ) {
        Document metadata = entryDocument.getDocument(FieldName.METADATA);
        Object content = entryDocument.get(FieldName.CONTENT);
        if (metadata == null || content == null) {
            throw new IllegalArgumentException("The supplied document is not of the required format");
        }
        String key = metadata.getString(FieldName.ID);
        if (key == null) {
            throw new IllegalArgumentException("The supplied document is not of the required format");
        }
        SchematicEntry newEntry = null;
        if (content instanceof Document) {
            newEntry = new SchematicEntryLiteral(metadata, (Document)content);
        }
        SchematicEntry oldValue = store.put(key, newEntry);
        return oldValue != null ? removedResult(key, oldValue) : null;
    }

    @Override
    public SchematicEntry putIfAbsent( String key,
                                       Document document ) {
        SchematicEntryLiteral newEntry = new SchematicEntryLiteral(key, document);
        SchematicEntry existingEntry = store.putIfAbsent(key, newEntry);
        return existingEntry;
    }

    @Override
    public SchematicEntry replace( String key,
                                   Document document ) {
        SchematicEntryLiteral newEntry = new SchematicEntryLiteral(key, document);
        return removedResult(key, store.replace(key, newEntry));
    }

    @Override
    public SchematicEntry remove( String key ) {
        SchematicEntry existing = store.remove(key);
        return existing == null ? null : removedResult(key, existing);
    }

    @Override
    public EditableDocument editContent( String key,
                                         boolean createIfMissing ) {
        return editContent(key, createIfMissing, true);
    }

    @Override
    public EditableDocument editContent( String key,
                                         boolean createIfMissing,
                                         boolean acquireLock ) {
        // Get the literal ...
        SchematicEntryLiteral literal = (SchematicEntryLiteral)store.get(key);
        if (literal == null) {
            if (!createIfMissing) return null;
            literal = new SchematicEntryLiteral(key);
            store.put(key, literal);
            return new DocumentEditor((MutableDocument)literal.getContent());
        }
        // this makes a copy and puts the new copy into the store ...
        return literal.edit(key, store, shouldAcquireLock(acquireLock));
    }

    private AdvancedCache<String, SchematicEntry> shouldAcquireLock( boolean acquireLockRequested ) {
        return explicitLocking && acquireLockRequested ? lockingStore : null;
    }

    @Override
    public boolean lock( Collection<String> keys ) {
        return !explicitLocking || keys.isEmpty() ? true : lockingStore.lock(keys);
    }

    @Override
    public boolean lock( String key ) {
        return !explicitLocking ? true : lockingStore.lock(key);
    }
}
