/*
 * 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.ode.daohib.bpel;

import java.util.*;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.ode.bpel.common.CorrelationKey;
import org.apache.ode.bpel.common.CorrelationKeySet;
import org.apache.ode.bpel.dao.*;
import org.apache.ode.bpel.iapi.Scheduler;
import org.apache.ode.daohib.SessionManager;
import org.apache.ode.daohib.bpel.hobj.HCorrelator;
import org.apache.ode.daohib.bpel.hobj.HCorrelatorMessage;
import org.apache.ode.daohib.bpel.hobj.HCorrelatorSelector;
import org.apache.ode.daohib.bpel.hobj.HMessageExchange;
import org.apache.ode.daohib.bpel.hobj.HProcess;
import org.apache.ode.daohib.bpel.hobj.HProcessInstance;
import org.hibernate.Hibernate;
import org.hibernate.LockMode;
import org.hibernate.Query;
import org.hibernate.Session;
import org.hibernate.exception.LockAcquisitionException;

import javax.xml.namespace.QName;

/**
 * Hibernate-based {@link CorrelatorDAO} implementation.
 */
class CorrelatorDaoImpl extends HibernateDao implements CorrelatorDAO {
    static Logger __log = LoggerFactory.getLogger(CorrelatorDaoImpl.class);

    /** filter for finding a matching selector. */
    private static final String LOCK_SELECTORS = "update from HCorrelatorSelector as hs set hs.lock = hs.lock+1 where hs.processType = :processType";
    private static final String CHECK_SELECTORS = "from HCorrelatorSelector as hs where hs.processType = :processType and hs.correlator.correlatorId = :correlatorId";
    //private static final String FLTR_SELECTORS = "from HCorrelatorSelector as hs where hs.processType = :processType and hs.correlator.correlatorId = :correlatorId";
    private static final String FLTR_SELECTORS = "from HCorrelatorSelector as hs where hs.correlator = :correlator";
    private static final String FLTR_SELECTORS_SUBQUERY = ("from HCorrelatorSelector as hs where hs.processType = :processType and hs.correlatorId = " +
            "(select hc.id from HCorrelator as hc where hc.correlatorId = :correlatorId)").intern();

    /** Query for removing routes. */
    private static final String QRY_DELSELECTORS = "delete from HCorrelatorSelector where instance = ? and groupId = ?";

    private HCorrelator _hobj;

    public CorrelatorDaoImpl(SessionManager sm, HCorrelator hobj) {
        super(sm, hobj);
        entering("CorrelatorDaoImpl.CorrelatorDaoImpl");
        _hobj = hobj;
    }

    @SuppressWarnings("unchecked")
    public MessageExchangeDAO dequeueMessage(CorrelationKeySet keySet) {
        entering("CorrelatorDaoImpl.dequeueMessage");

        MessageExchangeDAO mex = null;
        CorrelationKeySet keyInLoop;

        __log.debug("dequeueMessage({}): ",keySet);

        Query qry = getSession().createFilter(_hobj.getMessageCorrelations()," where this.correlationKey = :s0");
        qry.setString("s0", keySet.toCanonicalString());

        // We really should consider the possibility of multiple messages matching a criteria.
        // When the message is handled, its not too convenient to attempt to determine if the
        // received message conflicts with one already received.
        Iterator mcors;
        try {
            mcors = qry.setLockMode("this", LockMode.UPGRADE).iterate();
        } catch (LockAcquisitionException e) {
            throw new Scheduler.JobProcessorException(e, true);
        }
        try {
            if (!mcors.hasNext()) {
                __log.debug("dequeueMessage({}): did not find a MESSAGE entry.",keySet);
            } else {
                boolean keysNotMatched = true;

                for( ; mcors.hasNext() && keysNotMatched ; ) {
                    HCorrelatorMessage mcor = (HCorrelatorMessage) mcors.next();
                    keyInLoop = new CorrelationKeySet(mcor.getCorrelationKey());

                    __log.debug("KeySet in enqueued message is {} and KeySet to be dequeued {}",keyInLoop,keySet);

                    if(keyInLoop.isRoutableTo(keySet, false)) {
                        __log.debug("dequeueMessage({}): found MESSAGE entry {}", keySet, mcor.getMessageExchange());

                        mex = new MessageExchangeDaoImpl(_sm, mcor.getMessageExchange());
                        keysNotMatched = false;
                    }
                }
            }
        } finally {
            Hibernate.close(mcors);
        }

        return mex;
    }

    @SuppressWarnings("unchecked")
    public List<MessageRouteDAO> findRoute(CorrelationKeySet keySet) {
        return findRoute(keySet,false);
    }

    private String generateUnmatchedQuery(List<CorrelationKeySet> subSets) {
        StringBuffer filterQuery = new StringBuffer();

        if( subSets.size() == 1 ) {
            filterQuery.append(" where this.correlationKey = :s0");
        } else if( subSets.size() > 1 ) {
            filterQuery.append(" where this.correlationKey in(");
            for( int i = 0; i < subSets.size(); i++ ) {
                if( i > 0 ) {
                    filterQuery.append(", ");
                }
                filterQuery.append(":s").append(i);
            }
            filterQuery.append(")");
        }

        return filterQuery.toString();
    }

    private String generateSelectorQuery(String header, List<CorrelationKeySet> subSets) {
        StringBuffer filterQuery = new StringBuffer(header);

        if( subSets.size() == 1 ) {
            filterQuery.append(" and hs.correlationKey = :s0");
        } else if( subSets.size() > 1 ) {
            filterQuery.append(" and hs.correlationKey in(");
            for( int i = 0; i < subSets.size(); i++ ) {
                if( i > 0 ) {
                    filterQuery.append(", ");
                }
                filterQuery.append(":s").append(i);
            }
            filterQuery.append(")");
        }

        return filterQuery.toString();
    }

    public void enqueueMessage(MessageExchangeDAO mex, CorrelationKeySet correlationKeySet) {
        entering("CorrelatorDaoImpl.enqueueMessage");
        String hdr = "enqueueMessage(mex=" + ((MessageExchangeDaoImpl) mex)._hobj.getId() + " keySet="
                + correlationKeySet.toCanonicalString() + "): ";

        if (__log.isDebugEnabled())
            __log.debug(hdr);

        for( CorrelationKeySet aSubSet : correlationKeySet.findSubSets() ) {
            HCorrelatorMessage mcor = new HCorrelatorMessage();
            mcor.setCorrelator(_hobj);
            mcor.setCreated(new Date());
            mcor.setMessageExchange((HMessageExchange) ((MessageExchangeDaoImpl) mex)._hobj);
            mcor.setCorrelationKey(aSubSet.toCanonicalString());
            getSession().save(mcor);

            if (__log.isDebugEnabled())
                __log.debug(hdr + "saved " + mcor);
        }
    }

    public void addRoute(String routeGroupId, ProcessInstanceDAO target, int idx, CorrelationKeySet correlationKeySet, String routePolicy) {
        entering("CorrelatorDaoImpl.addRoute");
        String hdr = "addRoute(" + routeGroupId + ", iid=" + target.getInstanceId() + ", idx=" + idx + ", ckeySet="
                + correlationKeySet + "): ";

        if (__log.isDebugEnabled())
            __log.debug(hdr);
        HCorrelatorSelector hsel = new HCorrelatorSelector();
        hsel.setGroupId(routeGroupId);
        hsel.setIndex(idx);
        hsel.setLock(0);
        hsel.setCorrelationKey(correlationKeySet.toCanonicalString());
        hsel.setInstance((HProcessInstance) ((ProcessInstanceDaoImpl) target).getHibernateObj());
        hsel.setProcessType(target.getProcess().getType().toString());
        hsel.setCorrelator(_hobj);
        hsel.setCreated(new Date());
        hsel.setRoute(routePolicy);
        try {
            getSession().save(hsel);
        } catch (LockAcquisitionException e) {
            throw new Scheduler.JobProcessorException(e, true);
        }

        if (__log.isDebugEnabled())
            __log.debug(hdr + "saved " + hsel);
    }

    public boolean checkRoute(CorrelationKeySet correlationKeySet) {
        entering("CorrelatorDaoImpl.checkRoute");
        Query q = getSession().getNamedQuery(HCorrelatorSelector.SELECT_MESSAGE_ROUTE);
        q.setEntity("corr",_hobj);
        q.setString("ckey", correlationKeySet.toCanonicalString());
        q.setReadOnly(true);
        return q.list().isEmpty();
    }

    public String getCorrelatorId() {
        return _hobj.getCorrelatorId();
    }

    public void setCorrelatorId(String newId) {
        _hobj.setCorrelatorId(newId);
    }

    public void removeRoutes(String routeGroupId, ProcessInstanceDAO target) {
        entering("CorrelatorDaoImpl.removeRoutes");
        String hdr = "removeRoutes(" + routeGroupId + ", iid=" + target.getInstanceId() + "): ";
        __log.debug(hdr);
        Session session = getSession();
        Query q = session.createQuery(QRY_DELSELECTORS);
        q.setEntity(0, ((ProcessInstanceDaoImpl) target).getHibernateObj()); // instance
        q.setString(1, routeGroupId); // groupId
        int updates = q.executeUpdate();
        session.flush(); // explicit flush to ensure route removed
        if (__log.isDebugEnabled())
            __log.debug(hdr + "deleted " + updates + " rows");
    }

    public Collection<CorrelatorMessageDAO> getAllMessages() {
        Collection<CorrelatorMessageDAO> msgs = new ArrayList<CorrelatorMessageDAO>();
        for (HCorrelatorMessage correlatorMessage : _hobj.getMessageCorrelations())
            msgs.add(new CorrelatorMessageDaoImpl(_sm, correlatorMessage));
        return msgs;
    }

    public Collection<MessageRouteDAO> getAllRoutes() {
        Collection<MessageRouteDAO> routes = new ArrayList<MessageRouteDAO>();
        for (HCorrelatorSelector selector : _hobj.getSelectors())
            routes.add(new MessageRouteDaoImpl(_sm, selector));
        return routes;
    }

    public List<MessageRouteDAO> findRoute(CorrelationKeySet keySet,boolean isCorrleationKeySetPreInitialized) {
        List<MessageRouteDAO> routes = new ArrayList<MessageRouteDAO>();

        entering("CorrelatorDaoImpl.findRoute");
        __log.debug("findRoute(keySet={})",keySet);

        //Query q = getSession().createQuery(generateSelectorQuery(_sm.canJoinForUpdate() ? FLTR_SELECTORS : FLTR_SELECTORS_SUBQUERY, subSets));
        Query q = null;
        if(isCorrleationKeySetPreInitialized){
            q = getSession().createQuery(FLTR_SELECTORS + " and hs.correlationKey = :s0");
            q.setEntity("correlator", getHibernateObj());
            q.setString("s0", keySet.toCanonicalString());
        } else {
            List<CorrelationKeySet> subSets = keySet.findSubSets();
            q = getSession().createQuery(generateSelectorQuery(FLTR_SELECTORS, subSets));
            q.setEntity("correlator", getHibernateObj());

            for( int i = 0; i < subSets.size(); i++ ) {
                q.setString("s" + i, subSets.get(i).toCanonicalString());
            }
        }
        // Make sure we obtain a lock for the selector we want to find.
        q.setLockMode("hs", LockMode.UPGRADE);

        List<HProcessInstance> targets = new ArrayList<HProcessInstance>();
        List<HCorrelatorSelector> list;
        try {
            list = (List<HCorrelatorSelector>) q.list();
        } catch (LockAcquisitionException e) {
            throw new Scheduler.JobProcessorException(e, true);
        }
        for (HCorrelatorSelector selector : list) {
            __log.debug("selector returned form findRoute {} and targets {}", selector.getInstance().getId(),targets);
            if (selector != null) {
                boolean isRoutePolicyOne = selector.getRoute() == null || "one".equals(selector.getRoute());
                if ("all".equals(selector.getRoute()) ||
                        (isRoutePolicyOne && !targets.contains(selector.getInstance()))) {
                    __log.debug("selector added for targets {}", selector.getInstance().getId());
                    routes.add(new MessageRouteDaoImpl(_sm, selector));
                    targets.add(selector.getInstance());
                }
            }
        }

        __log.debug("findRoute(keySet={}) found {}",keySet,routes);

        return routes;
    }
}
