/*
 * Copyright (c) 2016 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.sharding;

import static com.google.common.base.Preconditions.checkState;
import static java.util.Objects.requireNonNull;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
import org.checkerframework.checker.lock.qual.GuardedBy;
import org.opendaylight.controller.cluster.databroker.actors.dds.DataStoreClient;
import org.opendaylight.controller.cluster.datastore.DistributedDataStoreInterface;
import org.opendaylight.mdsal.dom.api.DOMDataTreeChangeListener;
import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
import org.opendaylight.mdsal.dom.spi.AbstractDOMDataTreeChangeListenerRegistration;
import org.opendaylight.mdsal.dom.spi.AbstractRegistrationTree;
import org.opendaylight.mdsal.dom.spi.RegistrationTreeNode;
import org.opendaylight.mdsal.dom.spi.shard.ChildShardContext;
import org.opendaylight.mdsal.dom.spi.store.DOMStoreTreeChangePublisher;
import org.opendaylight.yangtools.concepts.ListenerRegistration;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.PathArgument;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidateNode;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidateNodes;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidates;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeConfiguration;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataValidationFailedException;
import org.opendaylight.yangtools.yang.data.impl.schema.tree.InMemoryDataTreeFactory;
import org.opendaylight.yangtools.yang.data.impl.schema.tree.SchemaValidationFailedException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@Deprecated(forRemoval = true)
public class DistributedShardChangePublisher
        extends AbstractRegistrationTree<AbstractDOMDataTreeChangeListenerRegistration<?>>
        implements DOMStoreTreeChangePublisher {

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

    private final DistributedDataStoreInterface distributedDataStore;
    private final YangInstanceIdentifier shardPath;

    private final Map<DOMDataTreeIdentifier, ChildShardContext> childShards;

    @GuardedBy("this")
    private final DataTree dataTree;

    public DistributedShardChangePublisher(final DataStoreClient client,
                                           final DistributedDataStoreInterface distributedDataStore,
                                           final DOMDataTreeIdentifier prefix,
                                           final Map<DOMDataTreeIdentifier, ChildShardContext> childShards) {
        this.distributedDataStore = distributedDataStore;
        // TODO keeping the whole dataTree thats contained in subshards doesn't seem like a good idea
        // maybe the whole listener logic would be better in the backend shards where we have direct access to the
        // dataTree and wont have to cache it redundantly.

        final DataTreeConfiguration baseConfig;
        switch (prefix.getDatastoreType()) {
            case CONFIGURATION:
                baseConfig = DataTreeConfiguration.DEFAULT_CONFIGURATION;
                break;
            case OPERATIONAL:
                baseConfig = DataTreeConfiguration.DEFAULT_OPERATIONAL;
                break;
            default:
                throw new UnsupportedOperationException("Unknown prefix type " + prefix.getDatastoreType());
        }

        this.dataTree = new InMemoryDataTreeFactory().create(new DataTreeConfiguration.Builder(baseConfig.getTreeType())
                .setMandatoryNodesValidation(baseConfig.isMandatoryNodesValidationEnabled())
                .setUniqueIndexes(baseConfig.isUniqueIndexEnabled())
                .setRootPath(prefix.getRootIdentifier())
                .build());

        // XXX: can we guarantee that the root is present in the schemacontext?
        this.dataTree.setEffectiveModelContext(distributedDataStore.getActorUtils().getSchemaContext());
        this.shardPath = prefix.getRootIdentifier();
        this.childShards = childShards;
    }

    protected void registrationRemoved(final AbstractDOMDataTreeChangeListenerRegistration<?> registration) {
        LOG.debug("Closing registration {}", registration);
    }

    @Override
    public <L extends DOMDataTreeChangeListener> AbstractDOMDataTreeChangeListenerRegistration<L>
            registerTreeChangeListener(final YangInstanceIdentifier path, final L listener) {
        takeLock();
        try {
            return setupListenerContext(path, listener);
        } finally {
            releaseLock();
        }
    }

    private <L extends DOMDataTreeChangeListener> AbstractDOMDataTreeChangeListenerRegistration<L>
            setupListenerContext(final YangInstanceIdentifier listenerPath, final L listener) {
        // we need to register the listener registration path based on the shards root
        // we have to strip the shard path from the listener path and then register
        YangInstanceIdentifier strippedIdentifier = listenerPath;
        if (!shardPath.isEmpty()) {
            strippedIdentifier = YangInstanceIdentifier.create(stripShardPath(shardPath, listenerPath));
        }

        final DOMDataTreeListenerWithSubshards subshardListener =
                new DOMDataTreeListenerWithSubshards(strippedIdentifier, listener);
        final AbstractDOMDataTreeChangeListenerRegistration<L> reg =
                setupContextWithoutSubshards(listenerPath, strippedIdentifier, subshardListener);

        for (final ChildShardContext maybeAffected : childShards.values()) {
            if (listenerPath.contains(maybeAffected.getPrefix().getRootIdentifier())) {
                // consumer has initialDataChangeEvent subshard somewhere on lower level
                // register to the notification manager with snapshot and forward child notifications to parent
                LOG.debug("Adding new subshard{{}} to listener at {}", maybeAffected.getPrefix(), listenerPath);
                subshardListener.addSubshard(maybeAffected);
            } else if (maybeAffected.getPrefix().getRootIdentifier().contains(listenerPath)) {
                // bind path is inside subshard
                // TODO can this happen? seems like in ShardedDOMDataTree we are
                // already registering to the lowest shard possible
                throw new UnsupportedOperationException("Listener should be registered directly "
                        + "into initialDataChangeEvent subshard");
            }
        }

        return reg;
    }

    private <L extends DOMDataTreeChangeListener> AbstractDOMDataTreeChangeListenerRegistration<L>
            setupContextWithoutSubshards(final YangInstanceIdentifier shardLookup,
                                         final YangInstanceIdentifier listenerPath,
                                         final DOMDataTreeListenerWithSubshards listener) {

        LOG.debug("Registering root listener full path: {}, path inside shard: {}", shardLookup, listenerPath);

        // register in the shard tree
        final RegistrationTreeNode<AbstractDOMDataTreeChangeListenerRegistration<?>> node =
                findNodeFor(listenerPath.getPathArguments());

        // register listener in CDS
        ListenerRegistration<DOMDataTreeChangeListener> listenerReg = distributedDataStore
                .registerProxyListener(shardLookup, listenerPath, listener);

        @SuppressWarnings("unchecked")
        final AbstractDOMDataTreeChangeListenerRegistration<L> registration =
            new AbstractDOMDataTreeChangeListenerRegistration<>((L) listener) {
                @Override
                protected void removeRegistration() {
                    listener.close();
                    DistributedShardChangePublisher.this.removeRegistration(node, this);
                    registrationRemoved(this);
                    listenerReg.close();
                }
            };
        addRegistration(node, registration);

        return registration;
    }

    private static Iterable<PathArgument> stripShardPath(final YangInstanceIdentifier shardPath,
                                                         final YangInstanceIdentifier listenerPath) {
        if (shardPath.isEmpty()) {
            return listenerPath.getPathArguments();
        }

        final List<PathArgument> listenerPathArgs = new ArrayList<>(listenerPath.getPathArguments());
        final Iterator<PathArgument> shardIter = shardPath.getPathArguments().iterator();
        final Iterator<PathArgument> listenerIter = listenerPathArgs.iterator();

        while (shardIter.hasNext()) {
            if (shardIter.next().equals(listenerIter.next())) {
                listenerIter.remove();
            } else {
                break;
            }
        }

        return listenerPathArgs;
    }

    synchronized DataTreeCandidate applyChanges(final YangInstanceIdentifier listenerPath,
            final Collection<DataTreeCandidate> changes) throws DataValidationFailedException {
        final DataTreeModification modification = dataTree.takeSnapshot().newModification();
        for (final DataTreeCandidate change : changes) {
            try {
                DataTreeCandidates.applyToModification(modification, change);
            } catch (SchemaValidationFailedException e) {
                LOG.error("Validation failed", e);
            }
        }

        modification.ready();

        final DataTreeCandidate candidate;

        dataTree.validate(modification);

        // strip nodes we dont need since this listener doesn't have to be registered at the root of the DataTree
        candidate = dataTree.prepare(modification);
        dataTree.commit(candidate);


        DataTreeCandidateNode modifiedChild = candidate.getRootNode();

        for (final PathArgument pathArgument : listenerPath.getPathArguments()) {
            modifiedChild = modifiedChild.getModifiedChild(pathArgument).orElse(null);
        }


        if (modifiedChild == null) {
            modifiedChild = DataTreeCandidateNodes.empty(dataTree.getRootPath().getLastPathArgument());
        }

        return DataTreeCandidates.newDataTreeCandidate(dataTree.getRootPath(), modifiedChild);
    }


    private final class DOMDataTreeListenerWithSubshards implements DOMDataTreeChangeListener {

        private final YangInstanceIdentifier listenerPath;
        private final DOMDataTreeChangeListener delegate;
        private final Map<YangInstanceIdentifier, ListenerRegistration<DOMDataTreeChangeListener>> registrations =
                new ConcurrentHashMap<>();

        @GuardedBy("this")
        private final Collection<DataTreeCandidate> stashedDataTreeCandidates = new LinkedList<>();

        DOMDataTreeListenerWithSubshards(final YangInstanceIdentifier listenerPath,
                                         final DOMDataTreeChangeListener delegate) {
            this.listenerPath = requireNonNull(listenerPath);
            this.delegate = requireNonNull(delegate);
        }

        @Override
        public synchronized void onDataTreeChanged(final Collection<DataTreeCandidate> changes) {
            LOG.debug("Received data changed {}", changes);

            if (!stashedDataTreeCandidates.isEmpty()) {
                LOG.debug("Adding stashed subshards' changes {}", stashedDataTreeCandidates);
                changes.addAll(stashedDataTreeCandidates);
                stashedDataTreeCandidates.clear();
            }

            try {
                applyChanges(listenerPath, changes);
            } catch (final DataValidationFailedException e) {
                // TODO should we fail here? What if stashed changes
                // (changes from subshards) got ahead more than one generation
                // from current shard. Than we can fail to apply this changes
                // upon current data tree, but once we get respective changes
                // from current shard, we can apply also changes from
                // subshards.
                //
                // However, we can loose ability to notice and report some
                // errors then. For example, we cannot detect potential lost
                // changes from current shard.
                LOG.error("Validation failed for modification built from changes {}, current data tree: {}",
                        changes, dataTree, e);
                throw new RuntimeException("Notification validation failed", e);
            }

            delegate.onDataTreeChanged(changes);
        }

        synchronized void onDataTreeChanged(final YangInstanceIdentifier pathFromRoot,
                                            final Collection<DataTreeCandidate> changes) {
            final YangInstanceIdentifier changeId =
                    YangInstanceIdentifier.create(stripShardPath(dataTree.getRootPath(), pathFromRoot));

            final List<DataTreeCandidate> newCandidates = changes.stream()
                    .map(candidate -> DataTreeCandidates.newDataTreeCandidate(changeId, candidate.getRootNode()))
                    .collect(Collectors.toList());

            try {
                delegate.onDataTreeChanged(Collections.singleton(applyChanges(listenerPath, newCandidates)));
            } catch (final DataValidationFailedException e) {
                // We cannot apply changes from subshard to current data tree.
                // Maybe changes from current shard haven't been applied to
                // data tree yet. Postpone processing of these changes till we
                // receive changes from current shard.
                LOG.debug("Validation for modification built from subshard {} changes {} failed, current data tree {}.",
                        pathFromRoot, changes, dataTree, e);
                stashedDataTreeCandidates.addAll(newCandidates);
            }
        }

        void addSubshard(final ChildShardContext context) {
            checkState(context.getShard() instanceof DOMStoreTreeChangePublisher,
                    "All subshards that are initialDataChangeEvent part of ListenerContext need to be listenable");

            final DOMStoreTreeChangePublisher listenableShard = (DOMStoreTreeChangePublisher) context.getShard();
            // since this is going into subshard we want to listen for ALL changes in the subshard
            registrations.put(context.getPrefix().getRootIdentifier(),
                    listenableShard.registerTreeChangeListener(
                            context.getPrefix().getRootIdentifier(), changes -> onDataTreeChanged(
                                    context.getPrefix().getRootIdentifier(), changes)));
        }

        void close() {
            for (final ListenerRegistration<DOMDataTreeChangeListener> registration : registrations.values()) {
                registration.close();
            }
            registrations.clear();
        }
    }
}