<commit_before>/*   Copyright (c) 2015 Magnet Systems, 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 com.magnet.mmx.server.plugin.mmxmgmt.handler;

import com.magnet.mmx.protocol.Constants;
import com.magnet.mmx.protocol.MMXAttribute;
import com.magnet.mmx.protocol.MMXStatus;
import com.magnet.mmx.protocol.MMXTopicId;
import com.magnet.mmx.protocol.MMXTopicOptions;
import com.magnet.mmx.protocol.OSType;
import com.magnet.mmx.protocol.SendLastPublishedItems;
import com.magnet.mmx.protocol.TagSearch;
import com.magnet.mmx.protocol.TopicAction;
import com.magnet.mmx.protocol.TopicAction.ListType;
import com.magnet.mmx.protocol.TopicAction.MMXPublishedItem;
import com.magnet.mmx.protocol.TopicInfo;
import com.magnet.mmx.protocol.TopicSummary;
import com.magnet.mmx.server.api.v1.protocol.TopicCreateInfo;
import com.magnet.mmx.server.common.data.AppEntity;
import com.magnet.mmx.server.plugin.mmxmgmt.MMXException;
import com.magnet.mmx.server.plugin.mmxmgmt.db.ConnectionProvider;
import com.magnet.mmx.server.plugin.mmxmgmt.db.DbInteractionException;
import com.magnet.mmx.server.plugin.mmxmgmt.db.OpenFireDBConnectionProvider;
import com.magnet.mmx.server.plugin.mmxmgmt.db.QueryBuilderResult;
import com.magnet.mmx.server.plugin.mmxmgmt.db.SearchResult;
import com.magnet.mmx.server.plugin.mmxmgmt.db.TagDAO;
import com.magnet.mmx.server.plugin.mmxmgmt.db.TopicDAO;
import com.magnet.mmx.server.plugin.mmxmgmt.db.TopicEntity;
import com.magnet.mmx.server.plugin.mmxmgmt.db.UserDAO;
import com.magnet.mmx.server.plugin.mmxmgmt.db.UserDAOImpl;
import com.magnet.mmx.server.plugin.mmxmgmt.db.UserEntity;
import com.magnet.mmx.server.plugin.mmxmgmt.handler.ConfigureForm.PublishModel;
import com.magnet.mmx.server.plugin.mmxmgmt.pubsub.PubSubPersistenceManagerExt;
import com.magnet.mmx.server.plugin.mmxmgmt.pubsub.TopicQueryBuilder;
import com.magnet.mmx.server.plugin.mmxmgmt.search.PaginationInfo;
import com.magnet.mmx.server.plugin.mmxmgmt.topic.TopicNode;
import com.magnet.mmx.server.plugin.mmxmgmt.util.DBUtil;
import com.magnet.mmx.server.plugin.mmxmgmt.util.JIDUtil;
import com.magnet.mmx.server.plugin.mmxmgmt.util.MMXServerConstants;
import com.magnet.mmx.util.AppTopic;
import com.magnet.mmx.util.TopicHelper;
import com.magnet.mmx.util.Utils;
import org.dom4j.Element;
import org.jivesoftware.database.DbConnectionManager;
import org.jivesoftware.openfire.XMPPServer;
import org.jivesoftware.openfire.pubsub.CollectionNode;
import org.jivesoftware.openfire.pubsub.LeafNode;
import org.jivesoftware.openfire.pubsub.Node;
import org.jivesoftware.openfire.pubsub.NodeAffiliate;
import org.jivesoftware.openfire.pubsub.NodeSubscription;
import org.jivesoftware.openfire.pubsub.NotAcceptableException;
import org.jivesoftware.openfire.pubsub.PubSubService;
import org.jivesoftware.openfire.pubsub.PublishedItem;
import org.jivesoftware.openfire.pubsub.cluster.RefreshNodeTask;
import org.jivesoftware.openfire.pubsub.models.AccessModel;
import org.jivesoftware.openfire.pubsub.models.PublisherModel;
import org.jivesoftware.util.LocaleUtils;
import org.jivesoftware.util.StringUtils;
import org.jivesoftware.util.XMPPDateTimeFormat;
import org.jivesoftware.util.cache.CacheFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.xmpp.forms.DataForm;
import org.xmpp.packet.JID;
import org.xmpp.packet.Message;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Date;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map.Entry;
import java.util.TreeMap;
import java.util.TreeSet;

public class MMXTopicManager {

  private static final Logger LOGGER = LoggerFactory.getLogger(MMXAppManager.class);
  private static final boolean EXCLUDE_USER_TOPICS = true;
  private XMPPServer mServer = XMPPServer.getInstance();
  private PubSubService mPubSubModule = mServer.getPubSubModule();

  private static MMXTopicManager sInstance = null;

  public static final String MISSING_APP_ROOT_TOPIC = "Root topic not found for selected app.";
  public static final String INVALID_TOPIC_ID = "Supplied topic id is invalid.";
  public static final String INVALID_TOPIC_NAME = "Supplied topic name is invalid.";
  public static final String DUPLICATE_TOPIC_ID = "Topic with supplied id exists.";

  protected MMXTopicManager() {
  }

  public static MMXTopicManager getInstance() {
    if (sInstance == null) {
      sInstance = new MMXTopicManager();
    }
    return sInstance;
  }

  List<String> listTopics(String parentNode) {
    Collection<Node> nodes = mPubSubModule.getNodes();
    ArrayList<String> result = new ArrayList<String>();
    for (Node node : nodes) {
      result.add(node.getNodeID());
    }
    return result;
  }

  /**
   * Get a list of Leaf Topic nodes for the passed in appId
   * (collection nodes aren't returned)
   * @param appId
   * @return
   */
  public List<TopicNode> listTopicsForAppId(String appId) {
    ArrayList<TopicNode> result = new ArrayList<TopicNode>();
    CollectionNode rootNode = getRootAppTopic(appId);
    if (rootNode != null) {
      Collection<Node> nodes = rootNode.getNodes();
      for (Node node : nodes) {
        //For fixing: https://magneteng.atlassian.net/browse/MOB-833
        if (node.isCollectionNode()) {
          continue;
        }
        String identifier = node.getNodeID();
        boolean isAppTopic = TopicHelper.isAppTopic(identifier, appId);
        if (isAppTopic) {
          TopicNode tn = TopicNode.build(appId, node);
          result.add(tn);
        }
      }
    }
    return result;
  }

  /**
   * Get a list of Leaf Topic nodes for the passed in appId
   * (collection nodes aren't returned)
   * @param appId
   * @return
   */
  public List<com.magnet.mmx.server.api.v1.protocol.TopicInfo> getTopicInfo(String appId) {
    ArrayList<com.magnet.mmx.server.api.v1.protocol.TopicInfo> result = 
        new ArrayList<com.magnet.mmx.server.api.v1.protocol.TopicInfo>();
    CollectionNode rootNode = getRootAppTopic(appId);
    if (rootNode != null) {
      Collection<Node> nodes = rootNode.getNodes();
      for (Node node : nodes) {
        //For fixing: https://magneteng.atlassian.net/browse/MOB-833
        if (node.isCollectionNode()) {
          continue;
        }
        String identifier = node.getNodeID();
        boolean isAppTopic = TopicHelper.isAppTopic(identifier, appId);
        if (isAppTopic) {
          com.magnet.mmx.server.api.v1.protocol.TopicInfo info = 
              getTopicInfoFromNode(appId, node);
          result.add(info);
        }
      }
    }
    return result;
  }

  public com.magnet.mmx.server.api.v1.protocol.TopicInfo getTopicInfoFromNode(
                  String appId, Node node) {
    com.magnet.mmx.server.api.v1.protocol.TopicInfo info = new 
        com.magnet.mmx.server.api.v1.protocol.TopicInfo();
    info.setDescription(node.getDescription());
    info.setTopicName(node.getName());
    if(node instanceof  LeafNode) {
      LeafNode lnode = (LeafNode) node;
      info.setSubscriptionEnabled(lnode.isSubscriptionEnabled());
      info.setMaxItems(lnode.isPersistPublishedItems() ?
          lnode.getMaxPublishedItems() : 0);
    }
    info.setTopicName(TopicHelper.parseNode(node.getNodeID()).getName());
    info.setPublisherType(node.getPublisherModel().getName());
    return info;
  }
  
  /**
   * Get a list of subscriptions for a specific topic
   * @param topicId
   * @return
   */
  public List<NodeSubscription> listSubscriptionsForTopic(String topicId) {
    String lowerCase = topicId.toLowerCase();
    LeafNode node = (LeafNode) mPubSubModule.getNode(lowerCase);
    Collection<NodeSubscription> rv = Collections.emptyList();
    if (node != null) {
      rv = node.getAllSubscriptions();
    }
    ArrayList<NodeSubscription> returnList = new ArrayList<NodeSubscription>(rv);
    return returnList;
  }

  public TopicActionResult createTopic(AppEntity entity, TopicCreateInfo topicInfo) {
    TopicActionResult result = new TopicActionResult();
    if (topicInfo.getTopicName() == null || topicInfo.getTopicName().isEmpty()) {
      result.setSuccess(false);
      result.setCode(TopicFailureCode.INVALID_TOPIC_NAME);
      result.setMessage("Topic id is either null or empty");
      return result;
    }
    if (topicInfo.getTopicName().length() > MMXServerConstants.MAX_TOPIC_NAME_LEN) {
      result.setSuccess(false);
      result.setCode(TopicFailureCode.INVALID_TOPIC_ID);
      result.setMessage("Supplied topic id exceeds the permitted maximum length of:" + MMXServerConstants.MAX_TOPIC_NAME_LEN);
      return result;
    }
    String topicName = topicInfo.getTopicName();
    topicName = TopicHelper.normalizePath(topicName);

    String appId = entity.getAppId();
    String topicId = TopicHelper.makeTopic(appId, null, topicName);

    /**
     * Get the appRootNode
     */
    CollectionNode appRootNode = getRootAppTopic(appId);
    if (appRootNode == null) {
      result = new TopicActionResult();
      result.setSuccess(false);
      result.setMessage(MISSING_APP_ROOT_TOPIC);
      result.setCode(TopicFailureCode.NOTFOUND);
      return result;
    }
    // create the requested topic
    Node newAppTopic = null;

    try {
      newAppTopic = createLeafNode(entity.getServerUserId(), topicId, appRootNode, topicInfo);
      result.setSuccess(true);
      result.setNode(TopicNode.build(appId, newAppTopic));
    } catch (NotAcceptableException e) {
      result.setSuccess(false);
      result.setCode(TopicFailureCode.UNKNOWN);
      result.setMessage(e.getMessage());
    } catch (TopicExistsException e) {
      result.setSuccess(false);
      result.setCode(TopicFailureCode.DUPLICATE);
      result.setMessage(DUPLICATE_TOPIC_ID);
    }
    return result;
  }
  /**
   * Delete a topic identified by a topic id.
   * @param appId The app ID for error message.
   * @param topicId The node ID.
   * @return
   */
  public TopicActionResult deleteTopic (String appId, String topicId) {

    TopicActionResult result = new TopicActionResult();

    Node gonner = mPubSubModule.getNode(topicId);

    if (gonner == null) {
      result.setSuccess(false);
      result.setCode(TopicFailureCode.NOTFOUND);
      result.setMessage(INVALID_TOPIC_ID);
    } else {
      LOGGER.trace("Deleting topic with id:" + topicId);
      gonner.delete();
      result.setSuccess(true);
      result.setNode(TopicNode.build(appId, gonner));
      result.setMessage(INVALID_TOPIC_ID);
    }
    return result;
  }

  /**
   * Get the root collection node for an app
   *
   * @param appId
   * @return
   */
  public CollectionNode getRootAppTopic(String appId) {
    Node result = mPubSubModule.getNode(appId);
    if (result != null && result.isCollectionNode()) {
      return (CollectionNode) result;
    } else {
      return null;
    }
  }

  /**
   * Get a Node representing the topic with the specified topicId.
   * @param topicId
   * @return Node if one exists false other wise.
   */
  public Node getTopicNode (String topicId){
    Node result = mPubSubModule.getNode(topicId);
    return result;
  }

  CollectionNode createCollectionNode(String creatorUserId, String nodeId, CollectionNode parentNode) {
    CollectionNode parent = (parentNode == null) ? mPubSubModule.getRootCollectionNode() : parentNode;
    Node result = mPubSubModule.getNode(nodeId);
    if (result != null && !result.isCollectionNode()) {
      result.delete();
      result = null;
      // TODO cleanup existing published Items for this node
    }

    if (result == null) {
      ConfigureForm form = new ConfigureForm(DataForm.Type.submit);
      form.setSendItemSubscribe(true);
      form.setDeliverPayloads(true);
      form.setNotifyRetract(false);
      form.setNotifyDelete(false);
      form.setNotifyConfig(false);
      form.setNodeType(ConfigureForm.NodeType.collection);
      form.setPublishModel(ConfigureForm.PublishModel.open);
      LOGGER.trace("Collection config form: " + form);
      
      JID jid = new JID(creatorUserId, mServer.getServerInfo().getXMPPDomain(), null);
      CollectionNode node = new CollectionNode(mPubSubModule, parent, nodeId, jid);
      node.addOwner(jid);
      try {
        node.configure(form);
      } catch (NotAcceptableException e) {
        LOGGER.warn("NotAcceptableException", e);
      }
      node.saveToDB();
      CacheFactory.doClusterTask(new RefreshNodeTask(node));
      result = node;
    }
    return (CollectionNode) result;
  }

  /**
   * Create a topic using the topic request.
   *
   * @param creatorUserId
   * @param topicId
   * @param parentNode
   * @throws TopicExistsException   -- if topic already exists
   * @throws NotAcceptableException -- if an exception is thrown by openfire during topic creation.
   */
  private LeafNode createLeafNode(String creatorUserId, String topicId, 
                    CollectionNode parentNode, TopicCreateInfo topicInfo)
                        throws TopicExistsException, NotAcceptableException {

    String topicName = topicInfo.getTopicName();
    String topicDescription = topicInfo.getDescription();
    int maxItems = topicInfo.getMaxItems();

    Node result = mPubSubModule.getNode(topicId);
    if (result != null) {
      throw new TopicExistsException("Topic with id: " + topicId + " exists");
    }
    LeafNode createdNode = null;
    if (result == null) {
      // This config form should use the same values from setOptions().
      ConfigureForm form = new ConfigureForm(DataForm.Type.submit);
      form.setAccessModel(TopicHelper.isUserTopic(topicId) ?
          ConfigureForm.AccessModel.authorize : ConfigureForm.AccessModel.open);
      form.setPersistentItems(maxItems != 0);
      form.setMaxItems(maxItems);
      form.setSendItemSubscribe(true);
      form.setMaxPayloadSize(Constants.MAX_PAYLOAD_SIZE);
      form.setDeliverPayloads(true);
      form.setNotifyRetract(false);
      form.setNotifyDelete(false);
      form.setNotifyConfig(false);
      form.setNodeType(ConfigureForm.NodeType.leaf);
      // ;TODO: a workaround for MOB-1763
      form.setPublishModel(PublishModel.open);
      form.setTitle(topicName);
      form.setSubscribe(topicInfo.isSubscriptionEnabled());
      if (topicDescription != null) {
        form.setDescription(topicDescription);
      }
//      LOGGER.trace("Leaf config form: "+form);

      JID jid = new JID(creatorUserId, mServer.getServerInfo().getXMPPDomain(), null);
      LeafNode node = new LeafNode(mPubSubModule, parentNode, topicId, jid);
      node.addOwner(jid);
      try {
        node.configure(form);
      } catch (NotAcceptableException e) {
        LOGGER.warn("NotAcceptableException", e);
        throw e;
      }
      
      if (topicInfo.isSubscribeOnCreate()) {
        NodeSubscription subscription = subscribeToNode(node, jid, jid);
        // TODO: not returning the subscription ID yet.
      }
      
      node.saveToDB();
      CacheFactory.doClusterTask(new RefreshNodeTask(node));
      createdNode = node;
    }
    return createdNode;
  }

  String getAppIdPrefix(String appId) {
    return TopicHelper.TOPIC_DELIM + appId + TopicHelper.TOPIC_DELIM + "*" + TopicHelper.TOPIC_DELIM;
  }


  /**
   * Class that represents a result of a topic related action.
   */
  public static class TopicActionResult {
    boolean success;
    private TopicNode node;
    private String message;
    private TopicFailureCode code;

    public boolean isSuccess() {
      return success;
    }

    public void setSuccess(boolean success) {
      this.success = success;
    }

    public TopicNode getNode() {
      return node;
    }

    public void setNode(TopicNode node) {
      this.node = node;
    }

    public String getMessage() {
      return message;
    }

    public void setMessage(String message) {
      this.message = message;
    }

    public TopicFailureCode getCode() {
      return code;
    }

    public void setCode(TopicFailureCode code) {
      this.code = code;
    }
  }

  public static enum TopicFailureCode {
    DUPLICATE,
    NOTFOUND,
    UNKNOWN,
    INVALID_TOPIC_ID,
    INVALID_TOPIC_NAME
  }


  public static class TopicExistsException extends RuntimeException {
    public TopicExistsException(String message, Throwable cause) {
      super(message, cause);
    }

    public TopicExistsException(String message) {
      super(message);
    }
  }
  
  private void setOptions(String topicName, Node node, MMXTopicOptions options) {
    ConfigureForm form = new ConfigureForm(DataForm.Type.submit);
    form.setSendItemSubscribe(true);
    form.setMaxPayloadSize(Constants.MAX_PAYLOAD_SIZE);
    form.setDeliverPayloads(true);
    form.setNotifyRetract(false);
    form.setNotifyDelete(false);
    form.setNotifyConfig(false);
    form.setAccessModel(TopicHelper.isUserTopic(node.getNodeID()) ?
        ConfigureForm.AccessModel.authorize : ConfigureForm.AccessModel.open);
    form.setNodeType(node.isCollectionNode() ?
        ConfigureForm.NodeType.collection : ConfigureForm.NodeType.leaf);
    if (options == null) {
      form.setPublishModel(ConfigureForm.PublishModel.open);
      form.setPersistentItems(true);
      form.setMaxItems(-1);
      form.setSubscribe(true);
      form.setTitle(topicName);
    } else {
      // Set the default values if not specified.
      options.fillWithDefaults();
      
      form.setPublishModel(ConfigureForm.convert(options.getPublisherType()));
      form.setPersistentItems(options.getMaxItems() != 0);
      form.setMaxItems(options.getMaxItems());
      form.setSubscribe(options.isSubscriptionEnabled());
      form.setTitle(topicName);
      form.setDescription(options.getDescription());
    }
    try {
      node.configure(form);
    } catch (NotAcceptableException e) {
      e.printStackTrace();
    }
  }
  
  // Create the collection node and its ancestors recursively.
  private CollectionNode createCollectionNode(int prefix, String nodeId, 
                              JID creator, JID owner) throws IllegalArgumentException {
//    LOGGER.trace("createCollectionNode: prefix="+prefix+", nodeId="+nodeId);
    if (nodeId == null) {
      return null;
    }
    // Check if the collection node exists and is valid.
    Node node;
    if ((node = mPubSubModule.getNode(nodeId)) != null) {
      if (!node.isCollectionNode()) {
        throw new IllegalArgumentException(nodeId.substring(prefix)+
            " exists, but not a collection node.");
      }
//      LOGGER.trace("return existing collection node="+nodeId);
      return (CollectionNode) node;
    }
    String parentNodeId = TopicHelper.getParent(prefix, nodeId);
    CollectionNode parentNode;
    if (parentNodeId == null) {
      parentNode = getRootAppTopic(TopicHelper.getRootNodeId(nodeId));
    } else {
      parentNode = createCollectionNode(prefix, parentNodeId, creator, owner);
    }
    
    synchronized(nodeId.intern()) {
      if ((node = mPubSubModule.getNode(nodeId)) == null) {
        LOGGER.trace("create collection node=" + nodeId + ", parent=" +
                ((parentNode == null) ? "" : parentNode.getNodeID()));
        node = new CollectionNode(mPubSubModule, parentNode, nodeId, creator);
        setOptions(null, node, null);
        node.addOwner(owner);
        node.saveToDB();
        
        CacheFactory.doClusterTask(new RefreshNodeTask(node));
      }
    }
    LOGGER.trace("return new collection node=" + nodeId);
    return (CollectionNode) node;
  }
  
  /**
   * Create an application-wide or personal topic and create all parent nodes
   * if needed.
   * @param from
   * @param appId
   * @param rqt
   * @return
   * @throws MMXException
   */
  public MMXStatus createTopic(JID from, String appId, TopicAction.CreateRequest rqt) 
                          throws MMXException {
    String topic = rqt.getTopicName();
    try {
      // Note, TopicHelper.checkPathAllowed(topic) should not be called here;
      // it is a restriction in the client SDK.
      topic = TopicHelper.normalizePath(topic);
    } catch (IllegalArgumentException e) {
      throw new MMXException(e.getMessage(), StatusCode.BAD_REQUEST.getCode());
    }

    boolean isValid = TopicHelper.validateApplicationTopicName(topic);
    if (!isValid) {
      throw new MMXException(StatusCode.INVALID_TOPIC_NAME.getMessage(),
          StatusCode.INVALID_TOPIC_NAME.getCode());
    }
    String userId = JIDUtil.getUserId(from);
    JID owner = from.asBareJID();
    String topicId = TopicHelper.makeTopic(appId, rqt.isPersonal() ?
        userId : null, topic);
//    LOGGER.trace("createTopic realTopic="+realTopic+", topic="+topic);

    if (!mPubSubModule.canCreateNode(from)) {
      throw new MMXException(StatusCode.FORBIDDEN.getMessage(topic),
          StatusCode.FORBIDDEN.getCode());
    }
    if (mPubSubModule.getNode(topicId) != null) {
      throw new MMXException(StatusCode.TOPIC_EXISTS.getMessage(topic),
          StatusCode.TOPIC_EXISTS.getCode());
    }
    if (!mPubSubModule.isInstantNodeSupported()) {
      throw new MMXException(StatusCode.NOT_ACCEPTABLE.getMessage(topic),
          StatusCode.NOT_ACCEPTABLE.getCode());
    }
    int prefix = TopicHelper.getPrefixLength(topicId);
    String parentId = TopicHelper.getParent(prefix, topicId);
    CollectionNode parent;
    if (parentId == null)
      parent = getRootAppTopic(appId);
    else {
      try {
        // Recursively create the parent nodes if they don't exist.
        parent = createCollectionNode(prefix, parentId, from, owner);
      } catch (Throwable e) {
        throw new MMXException(e.getMessage(), StatusCode.BAD_REQUEST.getCode());
      }
    }
    if (parent != null && !rqt.isCollection()) {
      if (!parent.isAssociationAllowed(from)) {
        // Check if requester is allowed to add a new leaf node to the parent.
        throw new MMXException(StatusCode.FORBIDDEN.getMessage(topic), 
            StatusCode.FORBIDDEN.getCode());
      }
      if (parent.isMaxLeafNodeReached()) {
        throw new MMXException("Max nodes exceeded in the parent of "+topic,
            StatusCode.CONFLICT.getCode());
      }
    }
    
    synchronized(topicId.intern()) {
      if (mPubSubModule.getNode(topicId) != null) {
        throw new MMXException(StatusCode.TOPIC_EXISTS.getMessage(topic),
            StatusCode.TOPIC_EXISTS.getCode());
      }
//      LOGGER.trace("create node="+realTopic+", parent="+parent);
      Node node;
      if (rqt.isCollection())
        node = new CollectionNode(mPubSubModule, parent, topicId, from);
      else {
        node = new LeafNode(mPubSubModule, parent, topicId, from);
      }
      // Add the creator as the owner.
      node.addOwner(owner);
      setOptions(topic, node, rqt.getOptions());

      // Do the auto-subscription for creator.
      if (rqt.getOptions() != null && rqt.getOptions().isSubscribeOnCreate()) {
        NodeSubscription subscription = subscribeToNode(node, owner, owner);
        // TODO: not returning the subscription ID yet.
      }
      
      node.saveToDB();
      CacheFactory.doClusterTask(new RefreshNodeTask(node));
    }
//    LOGGER.trace("create node="+realTopic+" success");

    MMXStatus status = (new MMXStatus())
        .setCode(StatusCode.SUCCESS.getCode())
        .setMessage(StatusCode.SUCCESS.getMessage());
    return status;
  }
  
  private NodeSubscription subscribeToNode(Node node, JID owner, JID subscriber) {
    SubscribeForm optionsForm = new SubscribeForm(DataForm.Type.submit);
    // Receive notification of new items only.
    optionsForm.setSubscriptionType(NodeSubscription.Type.items);
    optionsForm.setSubscriptionDepth("all");
    // Don't set other options; it will change the subscription state to
    // pending because it will wait for the owner's approval.
    
    // Need a modified Node.java from mmx-openfire repo.
    NodeSubscription subscription = node.createSubscription(null, owner,
        subscriber, false, optionsForm);
    return subscription;
  }
  
  private int deleteNode(Node node, JID owner) throws MMXException {
    if (!node.getOwners().contains(owner)) {
      AppTopic topic = TopicHelper.parseTopic(node.getNodeID());
      throw new MMXException(StatusCode.FORBIDDEN.getMessage(topic.getName()),
          StatusCode.FORBIDDEN.getCode());
    }
    int count = 0;
    if (node.isCollectionNode()) {
      for (Node child : node.getNodes()) {
        if (child.isCollectionNode()) {
          count += deleteNode(child, owner);
        } else {
          if (!child.getOwners().contains(owner)) {
            AppTopic topic = TopicHelper.parseTopic(child.getNodeID());
            throw new MMXException(StatusCode.FORBIDDEN.getMessage(topic.getName()),
                StatusCode.FORBIDDEN.getCode());
          }
          
//          LOGGER.trace("delete leaf node=" + node.getNodeID());
          child.delete();
          ++count;
        }
      }
    }
//    LOGGER.trace("delete node="+node.getNodeID());
    node.delete();
    ++count;
    return count;
  }
  
  public MMXStatus deleteTopic(JID from, String appId, 
                          TopicAction.DeleteRequest rqt) throws MMXException {
    String topic = TopicHelper.normalizePath(rqt.getTopic());
    String userId = JIDUtil.getUserId(from);
    JID owner = from.asBareJID();
    String realTopic = TopicHelper.makeTopic(appId, rqt.isPersonal() ?
        userId : null, topic);
    Node node = mPubSubModule.getNode(realTopic);
    if (node == null) {
      throw new MMXException(StatusCode.TOPIC_NOT_FOUND.getMessage(topic), 
          StatusCode.TOPIC_NOT_FOUND.getCode());
    }
    int count = deleteNode(node, owner);    
    MMXStatus status = (new MMXStatus())
        .setCode(StatusCode.SUCCESS.getCode())
        .setMessage(count+" topic"+((count==1)?" is":"s are")+" deleted");
    return status;
  }
  
  public TopicInfo getTopic(JID from, String appId, MMXTopicId topic)
                          throws MMXException {
    String realTopic = TopicHelper.makeTopic(appId, topic.getEscUserId(),
        TopicHelper.normalizePath(topic.getName()));
    Node node = mPubSubModule.getNode(realTopic);
    if (node == null) {
      throw new MMXException(StatusCode.TOPIC_NOT_FOUND.getMessage(topic.getName()), 
          StatusCode.TOPIC_NOT_FOUND.getCode());
    }
    return nodeToInfo(topic.getUserId(), topic.getName(), node);
  }
  
  public MMXStatus retractAllFromTopic(JID from, String appId,
      TopicAction.RetractAllRequest rqt) throws MMXException {
    String topic = TopicHelper.normalizePath(rqt.getTopic());
    String userId = JIDUtil.getUserId(from);
    JID owner = from.asBareJID();
    String realTopic = TopicHelper.makeTopic(appId, rqt.isPersonal() ? 
        userId : null, topic);
    Node node = mPubSubModule.getNode(realTopic);
    if (node == null) {
      throw new MMXException(StatusCode.TOPIC_NOT_FOUND.getMessage(topic), 
          StatusCode.TOPIC_NOT_FOUND.getCode());
    }
    if (!node.getOwners().contains(owner)) {
      throw new MMXException(StatusCode.FORBIDDEN.getMessage(topic), 
          StatusCode.FORBIDDEN.getCode());
    }
    LeafNode leafNode = (LeafNode) node;
    List<PublishedItem> pubItems = leafNode.getPublishedItems();
    leafNode.deleteItems(pubItems);
    
    int count = (pubItems == null) ? 0 : pubItems.size();
    MMXStatus status = (new MMXStatus())
        .setCode(StatusCode.SUCCESS.getCode())
        .setMessage(count+" item"+((count==1)?" is":"s are")+" retracted");
    return status;
  }
  
  public MMXStatus retractFromTopic(JID from, String appId, 
      TopicAction.RetractRequest rqt) throws MMXException {
    String topic = TopicHelper.normalizePath(rqt.getTopic());
    String realTopic = TopicHelper.makeTopic(appId, rqt.getUserId(), topic);
    Node node = mPubSubModule.getNode(realTopic);
    if (node == null) {
      throw new MMXException(StatusCode.TOPIC_NOT_FOUND.getMessage(topic), 
          StatusCode.TOPIC_NOT_FOUND.getCode());
    }
    if (node.isCollectionNode()) {
      throw new MMXException("Cannot retract items from a collection topic",
          StatusCode.NOT_IMPLEMENTED.getCode());
    }
    
    LeafNode leafNode = (LeafNode) node;
    List<String> itemIds = rqt.getItemIds();
    if (itemIds == null) {
      throw new MMXException(StatusCode.BAD_REQUEST.getMessage("no item ID's"),
          StatusCode.BAD_REQUEST.getCode());
    }
    if (!leafNode.isItemRequired()) {
      // Cannot delete items from a leaf node that doesn't handle itemIDs
      throw new MMXException("Items required in this topic",
          StatusCode.NOT_IMPLEMENTED.getCode());
    }
    
    List<PublishedItem> pubItems = new ArrayList<PublishedItem>(itemIds.size());
    for (String itemId : itemIds) {
      if (itemId == null) {
        throw new MMXException(StatusCode.BAD_REQUEST.getMessage("null item ID"), 
            StatusCode.BAD_REQUEST.getCode());
      }
      PublishedItem item = leafNode.getPublishedItem(itemId);
      if (item == null) {
        throw new MMXException(StatusCode.ITEM_NOT_FOUND.getMessage(itemId), 
            StatusCode.ITEM_NOT_FOUND.getCode());
      }
      if (!item.canDelete(from)) {
        throw new MMXException(StatusCode.FORBIDDEN.getMessage(itemId), 
            StatusCode.FORBIDDEN.getCode());
      }
      pubItems.add(item);
    }
    leafNode.deleteItems(pubItems);
    
    int count = (pubItems == null) ? 0 : pubItems.size();
    MMXStatus status = (new MMXStatus())
        .setCode(StatusCode.SUCCESS.getCode())
        .setMessage(count+" item"+((count==1)?" is":"s are")+" retracted");
    return status;
  }
  
  public TopicAction.SubscribeResponse subscribeTopic(JID from, String appId, 
              TopicAction.SubscribeRequest rqt) throws MMXException {    
    String topic = TopicHelper.normalizePath(rqt.getTopic());
    String realTopic = TopicHelper.makeTopic(appId, rqt.getUserId(), topic);
    Node node = mPubSubModule.getNode(realTopic);
    if (node == null) {
      throw new MMXException(StatusCode.TOPIC_NOT_FOUND.getMessage(topic), 
          StatusCode.TOPIC_NOT_FOUND.getCode());
    }
    
    JID owner = from.asBareJID();
    // The subscriber can specify a different resource or without resource.
    JID subscriber = new JID(from.getNode(), from.getDomain(), rqt.getDevId());

    AccessModel accessModel = node.getAccessModel();
    if (!accessModel.canSubscribe(node, owner, subscriber)) {
      throw new MMXException(StatusCode.FORBIDDEN.getMessage(topic),
          StatusCode.FORBIDDEN.getCode());
    }
    
    // Check if the subscription owner is a user with outcast affiliation
    NodeAffiliate nodeAffiliate = node.getAffiliate(owner);
    if (nodeAffiliate != null &&
        nodeAffiliate.getAffiliation() == NodeAffiliate.Affiliation.outcast) {
      throw new MMXException(StatusCode.FORBIDDEN.getMessage(topic),
          StatusCode.FORBIDDEN.getCode());
    }
    
    // Check that subscriptions to the node are enabled
    if (!node.isSubscriptionEnabled()) {
      throw new MMXException(StatusCode.FORBIDDEN.getMessage(topic),
          StatusCode.FORBIDDEN.getCode());
    }

    // Check for duplicated subscription; return error or existing subscription.
    NodeSubscription subscription;
    if ((subscription = node.getSubscription(subscriber)) != null) {
      if (rqt.isErrorOnDup()) {
        throw new MMXException(StatusCode.SUBSCRIPTION_EXISTS.getMessage(topic),
          StatusCode.SUBSCRIPTION_EXISTS.getCode());
      } else {
        TopicAction.SubscribeResponse resp = new TopicAction.SubscribeResponse(
            subscription.getID(), StatusCode.SUCCESS.getCode(),
            StatusCode.SUCCESS.getMessage());
        return resp;
      }
    }
    
    subscription = subscribeToNode(node, owner, subscriber);
    
    TopicAction.SubscribeResponse resp = new TopicAction.SubscribeResponse(
        subscription.getID(), StatusCode.SUCCESS.getCode(), 
        StatusCode.SUCCESS.getMessage());
    return resp;
  }
  
  public MMXStatus unsubscribeTopic(JID from, String appId, 
                  TopicAction.UnsubscribeRequest rqt) throws MMXException {    
    String topic = TopicHelper.normalizePath(rqt.getTopic());
    String realTopic = TopicHelper.makeTopic(appId, rqt.getUserId(), topic);
    Node node = mPubSubModule.getNode(realTopic);
    if (node == null) {
      throw new MMXException(StatusCode.TOPIC_NOT_FOUND.getMessage(topic),
          StatusCode.TOPIC_NOT_FOUND.getCode());
    }
    int count = 0;
    JID owner = from.asBareJID();
    String subId = rqt.getSubId();
    for (NodeSubscription subscription : node.getSubscriptions(owner)) {
      if (subId == null || subscription.getID().equals(subId)) {
        ++count;
        node.cancelSubscription(subscription);
      }
    }
    if (count == 0) {
      throw new MMXException(StatusCode.GONE.getMessage(),
          StatusCode.GONE.getCode());
    }
    
    MMXStatus status = (new MMXStatus())
        .setCode(StatusCode.SUCCESS.getCode())
        .setMessage(count+" subscription"+((count==1)?" is":"s are")+" cancelled");
    return status;
  }
  
  public MMXStatus unsubscribeForDev(JID from, String appId, 
                  TopicAction.UnsubscribeForDevRequest rqt) throws MMXException {
    String prefix = TopicHelper.TOPIC_DELIM + appId + TopicHelper.TOPIC_DELIM;
    int count = 0;
    JID owner = from.asBareJID();
    String devId = rqt.getDevId();
    for (Node node : mPubSubModule.getNodes()) {
      if (!node.getNodeID().startsWith(prefix)) {
        continue;
      }
      for (NodeSubscription subscription : node.getSubscriptions(owner)) {
        if (devId.equals(subscription.getJID().getResource())) {
          ++count;
          node.cancelSubscription(subscription);
        }
      }
    }
    MMXStatus status = (new MMXStatus())
        .setCode(StatusCode.SUCCESS.getCode())
        .setMessage(count+" subscription"+((count==1)?" is":"s are")+" cancelled");
    return status;
  }
  
  public TopicAction.ListResponse listTopics(JID from, String appId,
                  TopicAction.ListRequest rqt) throws MMXException {
    TopicAction.ListResponse resp = new TopicAction.ListResponse();
    Integer maxLimit = rqt.getLimit();
    boolean recursive = rqt.isRecursive();
    String start = rqt.getStart();
    ListType type = rqt.getType();
    int limit = (maxLimit == null || maxLimit == -1) ? Integer.MAX_VALUE : maxLimit;
    String realTopic;
    if (start == null || start.isEmpty()) {
      // Get the global root node.
      Node node = mPubSubModule.getNode(appId);
      if (node == null) {
        throw new MMXException(StatusCode.APP_NODE_NOT_FOUND.getMessage(appId),
            StatusCode.APP_NODE_NOT_FOUND.getCode());
      }
      // Get its top level children nodes filtered by the search type.
      String userId = JIDUtil.getUserId(from);
      limit = getTopChildNodes(recursive, node, resp, limit, type, userId);
    } else {
      // Filter by the global topics.
      if (type == ListType.global || type == ListType.both) {
        realTopic = TopicHelper.makeTopic(appId, null, start);
        // Get the global topic node first.
        Node node = mPubSubModule.getNode(realTopic);
        if (node == null) {
          throw new MMXException(StatusCode.TOPIC_NOT_FOUND.getMessage(start),
                StatusCode.TOPIC_NOT_FOUND.getCode());
        }
        // Get its children nodes without search filter.
        limit = getChildNodes(recursive, node, resp, limit);
      }
      // Filter by the user topics.
      if (type == ListType.personal || type == ListType.both) {
        realTopic = TopicHelper.makeTopic(appId, JIDUtil.getUserId(from), start);
        // Get the personal topic node first.
        Node node = mPubSubModule.getNode(realTopic);
        if (node == null) {
          throw new MMXException(StatusCode.TOPIC_NOT_FOUND.getMessage(start),
                StatusCode.TOPIC_NOT_FOUND.getCode());
        }
        // Get its children nodes without search filter.
        limit = getChildNodes(recursive, node, resp, limit);
      }
    }
    return resp;
  }
  
  private TopicInfo nodeToInfo(String userId, String topic, Node node) {
    TopicInfo info = new TopicInfo(
        userId, node.getName() != null ? node.getName() : topic, node.isCollectionNode())
      .setCreationDate(node.getCreationDate())
      .setDescription(node.getDescription())
      .setModifiedDate(node.getModificationDate())
      .setCreator(node.getCreator().toString())
      .setSubscriptionEnabled(node.isSubscriptionEnabled());
    if (!node.isCollectionNode()) {
      LeafNode leafNode = (LeafNode) node;
      info.setMaxItems(leafNode.isPersistPublishedItems() ?
            leafNode.getMaxPublishedItems() : 0)
          .setMaxPayloadSize(leafNode.getMaxPayloadSize())
          .setPersistent(leafNode.isPersistPublishedItems())
          .setPublisherType(ConfigureForm.convert(leafNode.getPublisherModel()));
    } else {
      info.setMaxItems(0)
          .setMaxPayloadSize(0)
          .setPersistent(false)
          .setPublisherType(null);
    }
    return info;
  }
  
  // Get all top level children nodes which are either global topics or user
  // topics.  Filter out the nodes by the search type.
  private int getTopChildNodes(boolean recursive, Node root, 
      TopicAction.ListResponse resp, int limit, ListType type, String userId) {
    boolean globalOnly = (type == ListType.global);
    for (Node child : root.getNodes()) {
      AppTopic topic = TopicHelper.parseTopic(child.getNodeID());
      if (topic == null) {
        LOGGER.warn("Ignore malformed topic: " + child.getNodeID());
        continue;
      }
      // Brain teaser: the xor below is actually same as
      //    (type == ListType.global && !topic.isUserTopic()) ||
      //    (type == ListType.personal && topic.isUserTopic()))
      if ((type == ListType.both) || (globalOnly ^ topic.isUserTopic())) {
        if (topic.isUserTopic() && !topic.getEscUserId().equals(userId)) {
          continue;
        }
        if (--limit < 0) {
          return limit;
        }
        resp.add(nodeToInfo(topic.getUserId(), topic.getName(), child));        
        if (recursive && child.isCollectionNode()) {
          limit = getChildNodes(recursive, child, resp, limit);
        }
      }
    }
    return limit;
  }
  
  // Get all children nodes recursively below the top level.  
  // @return < 0 if exceeding the limit, >= 0 if within the limit.
  private int getChildNodes(boolean recursive, Node node, 
      TopicAction.ListResponse resp, int limit) {
    if (!node.isCollectionNode()) {
      AppTopic topic = TopicHelper.parseTopic(node.getNodeID());
      if (topic == null) {
        LOGGER.warn("Ignore malformed topic: " + node.getNodeID());
      } else {
        if (--limit < 0) {
          return limit;
        }
        resp.add(nodeToInfo(topic.getUserId(), topic.getName(), node));
      }
    } else {
      for (Node child : node.getNodes()) {
        AppTopic topic = TopicHelper.parseTopic(child.getNodeID());
        if (topic == null) {
          LOGGER.warn("Ignore malformed topic: " + child.getNodeID());
          continue;
        }
        if (--limit < 0) {
          return limit;
        }
        resp.add(nodeToInfo(topic.getUserId(), topic.getName(), child));
        if (recursive && child.isCollectionNode()) {
              limit = getChildNodes(recursive, child, resp, limit);
        }
      }
    }
    return limit;
  }
  
  private String[] getTopics(String appId, List<MMXTopicId> list, 
                              int begin, int size) {
    if (list == null) {
      return null;
    }
    size = Math.min(size, list.size() - begin);
    String[] topics = new String[size];
    for (int i = 0, index = begin; --size >= 0; index++, i++) {
      MMXTopicId topicNode = list.get(index);
      topics[i] = TopicHelper.makeTopic(appId, topicNode.getEscUserId(),
                                        topicNode.getName());
    }
    return topics;
  }
  
  private static class SQLHelper {
    public static String generateArgList(int numOfArgs) {
      if (numOfArgs == 0) {
        return "";
      }
      StringBuilder sb = new StringBuilder((numOfArgs-1) * 2 + 1);
      while (--numOfArgs > 0) {
        sb.append("?,");
      }
      sb.append('?');
      return sb.toString();
    }

    public static void bindArgList(PreparedStatement pstmt, int index,
                                      String[] args) throws SQLException {
      for (String arg : args) {
        pstmt.setString(index++, arg);
      }
    }
    
    public static void bindArgList(PreparedStatement pstmt, int index, 
                                      List<String> args) throws SQLException {
      for (String arg : args) {
        pstmt.setString(index++, arg);
      }
    }
  }
  
  public TopicAction.SummaryResponse getSummary(JID from, String appId,
          TopicAction.SummaryRequest rqt) throws MMXException {
    // Build a collection of topic ID's from the request; it contains topics
    // without any published items.
    HashSet<String> tpNoItems = new HashSet<String>(rqt.getTopicNodes().size());
    for (MMXTopicId topicId : rqt.getTopicNodes()) {
      tpNoItems.add(topicId.toString());
    }
    TopicAction.SummaryResponse resp = new TopicAction.SummaryResponse(
        rqt.getTopicNodes().size());
    Connection con = null;
    PreparedStatement pstmt = null;
    ResultSet rs = null;
    try {
      con = DbConnectionManager.getConnection();
      int start = 0;
      int numOfTopics = rqt.getTopicNodes().size();
      StringBuilder dateRange = new StringBuilder();
      ArrayList<String> dates = new ArrayList<String>();
      if (rqt.getSince() != null) {
        dateRange.append("AND ofPubsubItem.creationDate >= ? ");
        dates.add(StringUtils.dateToMillis(rqt.getSince()));
      }
      if (rqt.getUntil() != null) {
        dateRange.append("AND ofPubsubItem.creationDate <= ? ");
        dates.add(StringUtils.dateToMillis(rqt.getUntil()));
      }
      do {
        // Limit to 128 topics per query because some DBMS cannot handle more
        // than 255 arguments in the IN clause.
        String[] topics = getTopics(appId, rqt.getTopicNodes(), start, 128);
        if (topics == null || topics.length == 0) {
          break;
        }
        String argList = SQLHelper.generateArgList(topics.length);
        String sql = "SELECT ofPubsubNode.maxItems,count(*),max(ofPubsubItem.creationDate),ofPubsubNode.name "+
                     "FROM ofPubsubItem, ofPubsubNode  " +
                     "WHERE ofPubsubItem.serviceID=? AND ofPubsubItem.nodeID = ofPubsubNode.nodeId AND ofPubsubItem.nodeID IN ("+argList+") "+dateRange+
                     "GROUP BY ofPubsubItem.nodeID";
        pstmt = con.prepareStatement(sql);
        pstmt.setString(1, mPubSubModule.getServiceID());
        SQLHelper.bindArgList(pstmt, 2, topics);
        SQLHelper.bindArgList(pstmt, 2+topics.length, dates);
        rs = pstmt.executeQuery();
        while (rs.next()) {
          int maxItems = rs.getInt(1);
          int count = rs.getInt(2);
          Date creationDate = new Date(Long.parseLong(rs.getString(3).trim()));
          String topicName = rs.getString(4);
          MMXTopicId topic = new MMXTopicId(topicName);
          resp.add(new TopicSummary(topic)
            .setCount((maxItems < 0) ? count : Math.min(maxItems, count))
            .setLastPubTime(creationDate));
          // This topic has published items; remove it from the collection.
          tpNoItems.remove(topic.toString());
        }
        start += topics.length;
      } while (start < numOfTopics);
      // Fill the response with the topics having no published items.
      Iterator<String> it = tpNoItems.iterator();
      while (it.hasNext()) {
        String topicId = it.next();
        resp.add(new TopicSummary(MMXTopicId.parse(topicId)).setCount(0));
      }
      return resp;
    } catch (Exception sqle) {
      LOGGER.error(sqle.getMessage(), sqle);<commit_msg>Fix a bug that getting topic summary ignored user topics.<commit_after>/*   Copyright (c) 2015 Magnet Systems, 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 com.magnet.mmx.server.plugin.mmxmgmt.handler;

import com.magnet.mmx.protocol.Constants;
import com.magnet.mmx.protocol.MMXAttribute;
import com.magnet.mmx.protocol.MMXStatus;
import com.magnet.mmx.protocol.MMXTopicId;
import com.magnet.mmx.protocol.MMXTopicOptions;
import com.magnet.mmx.protocol.OSType;
import com.magnet.mmx.protocol.SendLastPublishedItems;
import com.magnet.mmx.protocol.TagSearch;
import com.magnet.mmx.protocol.TopicAction;
import com.magnet.mmx.protocol.TopicAction.ListType;
import com.magnet.mmx.protocol.TopicAction.MMXPublishedItem;
import com.magnet.mmx.protocol.TopicInfo;
import com.magnet.mmx.protocol.TopicSummary;
import com.magnet.mmx.server.api.v1.protocol.TopicCreateInfo;
import com.magnet.mmx.server.common.data.AppEntity;
import com.magnet.mmx.server.plugin.mmxmgmt.MMXException;
import com.magnet.mmx.server.plugin.mmxmgmt.db.ConnectionProvider;
import com.magnet.mmx.server.plugin.mmxmgmt.db.DbInteractionException;
import com.magnet.mmx.server.plugin.mmxmgmt.db.OpenFireDBConnectionProvider;
import com.magnet.mmx.server.plugin.mmxmgmt.db.QueryBuilderResult;
import com.magnet.mmx.server.plugin.mmxmgmt.db.SearchResult;
import com.magnet.mmx.server.plugin.mmxmgmt.db.TagDAO;
import com.magnet.mmx.server.plugin.mmxmgmt.db.TopicDAO;
import com.magnet.mmx.server.plugin.mmxmgmt.db.TopicEntity;
import com.magnet.mmx.server.plugin.mmxmgmt.db.UserDAO;
import com.magnet.mmx.server.plugin.mmxmgmt.db.UserDAOImpl;
import com.magnet.mmx.server.plugin.mmxmgmt.db.UserEntity;
import com.magnet.mmx.server.plugin.mmxmgmt.handler.ConfigureForm.PublishModel;
import com.magnet.mmx.server.plugin.mmxmgmt.pubsub.PubSubPersistenceManagerExt;
import com.magnet.mmx.server.plugin.mmxmgmt.pubsub.TopicQueryBuilder;
import com.magnet.mmx.server.plugin.mmxmgmt.search.PaginationInfo;
import com.magnet.mmx.server.plugin.mmxmgmt.topic.TopicNode;
import com.magnet.mmx.server.plugin.mmxmgmt.util.DBUtil;
import com.magnet.mmx.server.plugin.mmxmgmt.util.JIDUtil;
import com.magnet.mmx.server.plugin.mmxmgmt.util.MMXServerConstants;
import com.magnet.mmx.util.AppTopic;
import com.magnet.mmx.util.TopicHelper;
import com.magnet.mmx.util.Utils;
import org.dom4j.Element;
import org.jivesoftware.database.DbConnectionManager;
import org.jivesoftware.openfire.XMPPServer;
import org.jivesoftware.openfire.pubsub.CollectionNode;
import org.jivesoftware.openfire.pubsub.LeafNode;
import org.jivesoftware.openfire.pubsub.Node;
import org.jivesoftware.openfire.pubsub.NodeAffiliate;
import org.jivesoftware.openfire.pubsub.NodeSubscription;
import org.jivesoftware.openfire.pubsub.NotAcceptableException;
import org.jivesoftware.openfire.pubsub.PubSubService;
import org.jivesoftware.openfire.pubsub.PublishedItem;
import org.jivesoftware.openfire.pubsub.cluster.RefreshNodeTask;
import org.jivesoftware.openfire.pubsub.models.AccessModel;
import org.jivesoftware.openfire.pubsub.models.PublisherModel;
import org.jivesoftware.util.LocaleUtils;
import org.jivesoftware.util.StringUtils;
import org.jivesoftware.util.XMPPDateTimeFormat;
import org.jivesoftware.util.cache.CacheFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.xmpp.forms.DataForm;
import org.xmpp.packet.JID;
import org.xmpp.packet.Message;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Date;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map.Entry;
import java.util.TreeMap;
import java.util.TreeSet;

public class MMXTopicManager {

  private static final Logger LOGGER = LoggerFactory.getLogger(MMXAppManager.class);
  private static final boolean EXCLUDE_USER_TOPICS = true;
  private XMPPServer mServer = XMPPServer.getInstance();
  private PubSubService mPubSubModule = mServer.getPubSubModule();

  private static MMXTopicManager sInstance = null;

  public static final String MISSING_APP_ROOT_TOPIC = "Root topic not found for selected app.";
  public static final String INVALID_TOPIC_ID = "Supplied topic id is invalid.";
  public static final String INVALID_TOPIC_NAME = "Supplied topic name is invalid.";
  public static final String DUPLICATE_TOPIC_ID = "Topic with supplied id exists.";

  protected MMXTopicManager() {
  }

  public static MMXTopicManager getInstance() {
    if (sInstance == null) {
      sInstance = new MMXTopicManager();
    }
    return sInstance;
  }

  List<String> listTopics(String parentNode) {
    Collection<Node> nodes = mPubSubModule.getNodes();
    ArrayList<String> result = new ArrayList<String>();
    for (Node node : nodes) {
      result.add(node.getNodeID());
    }
    return result;
  }

  /**
   * Get a list of Leaf Topic nodes for the passed in appId
   * (collection nodes aren't returned)
   * @param appId
   * @return
   */
  public List<TopicNode> listTopicsForAppId(String appId) {
    ArrayList<TopicNode> result = new ArrayList<TopicNode>();
    CollectionNode rootNode = getRootAppTopic(appId);
    if (rootNode != null) {
      Collection<Node> nodes = rootNode.getNodes();
      for (Node node : nodes) {
        //For fixing: https://magneteng.atlassian.net/browse/MOB-833
        if (node.isCollectionNode()) {
          continue;
        }
        String identifier = node.getNodeID();
        boolean isAppTopic = TopicHelper.isAppTopic(identifier, appId);
        if (isAppTopic) {
          TopicNode tn = TopicNode.build(appId, node);
          result.add(tn);
        }
      }
    }
    return result;
  }

  /**
   * Get a list of Leaf Topic nodes for the passed in appId
   * (collection nodes aren't returned)
   * @param appId
   * @return
   */
  public List<com.magnet.mmx.server.api.v1.protocol.TopicInfo> getTopicInfo(String appId) {
    ArrayList<com.magnet.mmx.server.api.v1.protocol.TopicInfo> result = 
        new ArrayList<com.magnet.mmx.server.api.v1.protocol.TopicInfo>();
    CollectionNode rootNode = getRootAppTopic(appId);
    if (rootNode != null) {
      Collection<Node> nodes = rootNode.getNodes();
      for (Node node : nodes) {
        //For fixing: https://magneteng.atlassian.net/browse/MOB-833
        if (node.isCollectionNode()) {
          continue;
        }
        String identifier = node.getNodeID();
        boolean isAppTopic = TopicHelper.isAppTopic(identifier, appId);
        if (isAppTopic) {
          com.magnet.mmx.server.api.v1.protocol.TopicInfo info = 
              getTopicInfoFromNode(appId, node);
          result.add(info);
        }
      }
    }
    return result;
  }

  public com.magnet.mmx.server.api.v1.protocol.TopicInfo getTopicInfoFromNode(
                  String appId, Node node) {
    com.magnet.mmx.server.api.v1.protocol.TopicInfo info = new 
        com.magnet.mmx.server.api.v1.protocol.TopicInfo();
    info.setDescription(node.getDescription());
    info.setTopicName(node.getName());
    if(node instanceof  LeafNode) {
      LeafNode lnode = (LeafNode) node;
      info.setSubscriptionEnabled(lnode.isSubscriptionEnabled());
      info.setMaxItems(lnode.isPersistPublishedItems() ?
          lnode.getMaxPublishedItems() : 0);
    }
    info.setTopicName(TopicHelper.parseNode(node.getNodeID()).getName());
    info.setPublisherType(node.getPublisherModel().getName());
    return info;
  }
  
  /**
   * Get a list of subscriptions for a specific topic
   * @param topicId
   * @return
   */
  public List<NodeSubscription> listSubscriptionsForTopic(String topicId) {
    String lowerCase = topicId.toLowerCase();
    LeafNode node = (LeafNode) mPubSubModule.getNode(lowerCase);
    Collection<NodeSubscription> rv = Collections.emptyList();
    if (node != null) {
      rv = node.getAllSubscriptions();
    }
    ArrayList<NodeSubscription> returnList = new ArrayList<NodeSubscription>(rv);
    return returnList;
  }

  public TopicActionResult createTopic(AppEntity entity, TopicCreateInfo topicInfo) {
    TopicActionResult result = new TopicActionResult();
    if (topicInfo.getTopicName() == null || topicInfo.getTopicName().isEmpty()) {
      result.setSuccess(false);
      result.setCode(TopicFailureCode.INVALID_TOPIC_NAME);
      result.setMessage("Topic id is either null or empty");
      return result;
    }
    if (topicInfo.getTopicName().length() > MMXServerConstants.MAX_TOPIC_NAME_LEN) {
      result.setSuccess(false);
      result.setCode(TopicFailureCode.INVALID_TOPIC_ID);
      result.setMessage("Supplied topic id exceeds the permitted maximum length of:" + MMXServerConstants.MAX_TOPIC_NAME_LEN);
      return result;
    }
    String topicName = topicInfo.getTopicName();
    topicName = TopicHelper.normalizePath(topicName);

    String appId = entity.getAppId();
    String topicId = TopicHelper.makeTopic(appId, null, topicName);

    /**
     * Get the appRootNode
     */
    CollectionNode appRootNode = getRootAppTopic(appId);
    if (appRootNode == null) {
      result = new TopicActionResult();
      result.setSuccess(false);
      result.setMessage(MISSING_APP_ROOT_TOPIC);
      result.setCode(TopicFailureCode.NOTFOUND);
      return result;
    }
    // create the requested topic
    Node newAppTopic = null;

    try {
      newAppTopic = createLeafNode(entity.getServerUserId(), topicId, appRootNode, topicInfo);
      result.setSuccess(true);
      result.setNode(TopicNode.build(appId, newAppTopic));
    } catch (NotAcceptableException e) {
      result.setSuccess(false);
      result.setCode(TopicFailureCode.UNKNOWN);
      result.setMessage(e.getMessage());
    } catch (TopicExistsException e) {
      result.setSuccess(false);
      result.setCode(TopicFailureCode.DUPLICATE);
      result.setMessage(DUPLICATE_TOPIC_ID);
    }
    return result;
  }
  /**
   * Delete a topic identified by a topic id.
   * @param appId The app ID for error message.
   * @param topicId The node ID.
   * @return
   */
  public TopicActionResult deleteTopic (String appId, String topicId) {

    TopicActionResult result = new TopicActionResult();

    Node gonner = mPubSubModule.getNode(topicId);

    if (gonner == null) {
      result.setSuccess(false);
      result.setCode(TopicFailureCode.NOTFOUND);
      result.setMessage(INVALID_TOPIC_ID);
    } else {
      LOGGER.trace("Deleting topic with id:" + topicId);
      gonner.delete();
      result.setSuccess(true);
      result.setNode(TopicNode.build(appId, gonner));
      result.setMessage(INVALID_TOPIC_ID);
    }
    return result;
  }

  /**
   * Get the root collection node for an app
   *
   * @param appId
   * @return
   */
  public CollectionNode getRootAppTopic(String appId) {
    Node result = mPubSubModule.getNode(appId);
    if (result != null && result.isCollectionNode()) {
      return (CollectionNode) result;
    } else {
      return null;
    }
  }

  /**
   * Get a Node representing the topic with the specified topicId.
   * @param topicId
   * @return Node if one exists false other wise.
   */
  public Node getTopicNode (String topicId){
    Node result = mPubSubModule.getNode(topicId);
    return result;
  }

  CollectionNode createCollectionNode(String creatorUserId, String nodeId, CollectionNode parentNode) {
    CollectionNode parent = (parentNode == null) ? mPubSubModule.getRootCollectionNode() : parentNode;
    Node result = mPubSubModule.getNode(nodeId);
    if (result != null && !result.isCollectionNode()) {
      result.delete();
      result = null;
      // TODO cleanup existing published Items for this node
    }

    if (result == null) {
      ConfigureForm form = new ConfigureForm(DataForm.Type.submit);
      form.setSendItemSubscribe(true);
      form.setDeliverPayloads(true);
      form.setNotifyRetract(false);
      form.setNotifyDelete(false);
      form.setNotifyConfig(false);
      form.setNodeType(ConfigureForm.NodeType.collection);
      form.setPublishModel(ConfigureForm.PublishModel.open);
      LOGGER.trace("Collection config form: " + form);
      
      JID jid = new JID(creatorUserId, mServer.getServerInfo().getXMPPDomain(), null);
      CollectionNode node = new CollectionNode(mPubSubModule, parent, nodeId, jid);
      node.addOwner(jid);
      try {
        node.configure(form);
      } catch (NotAcceptableException e) {
        LOGGER.warn("NotAcceptableException", e);
      }
      node.saveToDB();
      CacheFactory.doClusterTask(new RefreshNodeTask(node));
      result = node;
    }
    return (CollectionNode) result;
  }

  /**
   * Create a topic using the topic request.
   *
   * @param creatorUserId
   * @param topicId
   * @param parentNode
   * @throws TopicExistsException   -- if topic already exists
   * @throws NotAcceptableException -- if an exception is thrown by openfire during topic creation.
   */
  private LeafNode createLeafNode(String creatorUserId, String topicId, 
                    CollectionNode parentNode, TopicCreateInfo topicInfo)
                        throws TopicExistsException, NotAcceptableException {

    String topicName = topicInfo.getTopicName();
    String topicDescription = topicInfo.getDescription();
    int maxItems = topicInfo.getMaxItems();

    Node result = mPubSubModule.getNode(topicId);
    if (result != null) {
      throw new TopicExistsException("Topic with id: " + topicId + " exists");
    }
    LeafNode createdNode = null;
    if (result == null) {
      // This config form should use the same values from setOptions().
      ConfigureForm form = new ConfigureForm(DataForm.Type.submit);
      form.setAccessModel(TopicHelper.isUserTopic(topicId) ?
          ConfigureForm.AccessModel.authorize : ConfigureForm.AccessModel.open);
      form.setPersistentItems(maxItems != 0);
      form.setMaxItems(maxItems);
      form.setSendItemSubscribe(true);
      form.setMaxPayloadSize(Constants.MAX_PAYLOAD_SIZE);
      form.setDeliverPayloads(true);
      form.setNotifyRetract(false);
      form.setNotifyDelete(false);
      form.setNotifyConfig(false);
      form.setNodeType(ConfigureForm.NodeType.leaf);
      // ;TODO: a workaround for MOB-1763
      form.setPublishModel(PublishModel.open);
      form.setTitle(topicName);
      form.setSubscribe(topicInfo.isSubscriptionEnabled());
      if (topicDescription != null) {
        form.setDescription(topicDescription);
      }
//      LOGGER.trace("Leaf config form: "+form);

      JID jid = new JID(creatorUserId, mServer.getServerInfo().getXMPPDomain(), null);
      LeafNode node = new LeafNode(mPubSubModule, parentNode, topicId, jid);
      node.addOwner(jid);
      try {
        node.configure(form);
      } catch (NotAcceptableException e) {
        LOGGER.warn("NotAcceptableException", e);
        throw e;
      }
      
      if (topicInfo.isSubscribeOnCreate()) {
        NodeSubscription subscription = subscribeToNode(node, jid, jid);
        // TODO: not returning the subscription ID yet.
      }
      
      node.saveToDB();
      CacheFactory.doClusterTask(new RefreshNodeTask(node));
      createdNode = node;
    }
    return createdNode;
  }

  String getAppIdPrefix(String appId) {
    return TopicHelper.TOPIC_DELIM + appId + TopicHelper.TOPIC_DELIM + "*" + TopicHelper.TOPIC_DELIM;
  }


  /**
   * Class that represents a result of a topic related action.
   */
  public static class TopicActionResult {
    boolean success;
    private TopicNode node;
    private String message;
    private TopicFailureCode code;

    public boolean isSuccess() {
      return success;
    }

    public void setSuccess(boolean success) {
      this.success = success;
    }

    public TopicNode getNode() {
      return node;
    }

    public void setNode(TopicNode node) {
      this.node = node;
    }

    public String getMessage() {
      return message;
    }

    public void setMessage(String message) {
      this.message = message;
    }

    public TopicFailureCode getCode() {
      return code;
    }

    public void setCode(TopicFailureCode code) {
      this.code = code;
    }
  }

  public static enum TopicFailureCode {
    DUPLICATE,
    NOTFOUND,
    UNKNOWN,
    INVALID_TOPIC_ID,
    INVALID_TOPIC_NAME
  }


  public static class TopicExistsException extends RuntimeException {
    public TopicExistsException(String message, Throwable cause) {
      super(message, cause);
    }

    public TopicExistsException(String message) {
      super(message);
    }
  }
  
  private void setOptions(String topicName, Node node, MMXTopicOptions options) {
    ConfigureForm form = new ConfigureForm(DataForm.Type.submit);
    form.setSendItemSubscribe(true);
    form.setMaxPayloadSize(Constants.MAX_PAYLOAD_SIZE);
    form.setDeliverPayloads(true);
    form.setNotifyRetract(false);
    form.setNotifyDelete(false);
    form.setNotifyConfig(false);
    form.setAccessModel(TopicHelper.isUserTopic(node.getNodeID()) ?
        ConfigureForm.AccessModel.authorize : ConfigureForm.AccessModel.open);
    form.setNodeType(node.isCollectionNode() ?
        ConfigureForm.NodeType.collection : ConfigureForm.NodeType.leaf);
    if (options == null) {
      form.setPublishModel(ConfigureForm.PublishModel.open);
      form.setPersistentItems(true);
      form.setMaxItems(-1);
      form.setSubscribe(true);
      form.setTitle(topicName);
    } else {
      // Set the default values if not specified.
      options.fillWithDefaults();
      
      form.setPublishModel(ConfigureForm.convert(options.getPublisherType()));
      form.setPersistentItems(options.getMaxItems() != 0);
      form.setMaxItems(options.getMaxItems());
      form.setSubscribe(options.isSubscriptionEnabled());
      form.setTitle(topicName);
      form.setDescription(options.getDescription());
    }
    try {
      node.configure(form);
    } catch (NotAcceptableException e) {
      e.printStackTrace();
    }
  }
  
  // Create the collection node and its ancestors recursively.
  private CollectionNode createCollectionNode(int prefix, String nodeId, 
                              JID creator, JID owner) throws IllegalArgumentException {
//    LOGGER.trace("createCollectionNode: prefix="+prefix+", nodeId="+nodeId);
    if (nodeId == null) {
      return null;
    }
    // Check if the collection node exists and is valid.
    Node node;
    if ((node = mPubSubModule.getNode(nodeId)) != null) {
      if (!node.isCollectionNode()) {
        throw new IllegalArgumentException(nodeId.substring(prefix)+
            " exists, but not a collection node.");
      }
//      LOGGER.trace("return existing collection node="+nodeId);
      return (CollectionNode) node;
    }
    String parentNodeId = TopicHelper.getParent(prefix, nodeId);
    CollectionNode parentNode;
    if (parentNodeId == null) {
      parentNode = getRootAppTopic(TopicHelper.getRootNodeId(nodeId));
    } else {
      parentNode = createCollectionNode(prefix, parentNodeId, creator, owner);
    }
    
    synchronized(nodeId.intern()) {
      if ((node = mPubSubModule.getNode(nodeId)) == null) {
        LOGGER.trace("create collection node=" + nodeId + ", parent=" +
                ((parentNode == null) ? "" : parentNode.getNodeID()));
        node = new CollectionNode(mPubSubModule, parentNode, nodeId, creator);
        setOptions(null, node, null);
        node.addOwner(owner);
        node.saveToDB();
        
        CacheFactory.doClusterTask(new RefreshNodeTask(node));
      }
    }
    LOGGER.trace("return new collection node=" + nodeId);
    return (CollectionNode) node;
  }
  
  /**
   * Create an application-wide or personal topic and create all parent nodes
   * if needed.
   * @param from
   * @param appId
   * @param rqt
   * @return
   * @throws MMXException
   */
  public MMXStatus createTopic(JID from, String appId, TopicAction.CreateRequest rqt) 
                          throws MMXException {
    String topic = rqt.getTopicName();
    try {
      // Note, TopicHelper.checkPathAllowed(topic) should not be called here;
      // it is a restriction in the client SDK.
      topic = TopicHelper.normalizePath(topic);
    } catch (IllegalArgumentException e) {
      throw new MMXException(e.getMessage(), StatusCode.BAD_REQUEST.getCode());
    }

    boolean isValid = TopicHelper.validateApplicationTopicName(topic);
    if (!isValid) {
      throw new MMXException(StatusCode.INVALID_TOPIC_NAME.getMessage(),
          StatusCode.INVALID_TOPIC_NAME.getCode());
    }
    String userId = JIDUtil.getUserId(from);
    JID owner = from.asBareJID();
    String topicId = TopicHelper.makeTopic(appId, rqt.isPersonal() ?
        userId : null, topic);
//    LOGGER.trace("createTopic realTopic="+realTopic+", topic="+topic);

    if (!mPubSubModule.canCreateNode(from)) {
      throw new MMXException(StatusCode.FORBIDDEN.getMessage(topic),
          StatusCode.FORBIDDEN.getCode());
    }
    if (mPubSubModule.getNode(topicId) != null) {
      throw new MMXException(StatusCode.TOPIC_EXISTS.getMessage(topic),
          StatusCode.TOPIC_EXISTS.getCode());
    }
    if (!mPubSubModule.isInstantNodeSupported()) {
      throw new MMXException(StatusCode.NOT_ACCEPTABLE.getMessage(topic),
          StatusCode.NOT_ACCEPTABLE.getCode());
    }
    int prefix = TopicHelper.getPrefixLength(topicId);
    String parentId = TopicHelper.getParent(prefix, topicId);
    CollectionNode parent;
    if (parentId == null)
      parent = getRootAppTopic(appId);
    else {
      try {
        // Recursively create the parent nodes if they don't exist.
        parent = createCollectionNode(prefix, parentId, from, owner);
      } catch (Throwable e) {
        throw new MMXException(e.getMessage(), StatusCode.BAD_REQUEST.getCode());
      }
    }
    if (parent != null && !rqt.isCollection()) {
      if (!parent.isAssociationAllowed(from)) {
        // Check if requester is allowed to add a new leaf node to the parent.
        throw new MMXException(StatusCode.FORBIDDEN.getMessage(topic), 
            StatusCode.FORBIDDEN.getCode());
      }
      if (parent.isMaxLeafNodeReached()) {
        throw new MMXException("Max nodes exceeded in the parent of "+topic,
            StatusCode.CONFLICT.getCode());
      }
    }
    
    synchronized(topicId.intern()) {
      if (mPubSubModule.getNode(topicId) != null) {
        throw new MMXException(StatusCode.TOPIC_EXISTS.getMessage(topic),
            StatusCode.TOPIC_EXISTS.getCode());
      }
//      LOGGER.trace("create node="+realTopic+", parent="+parent);
      Node node;
      if (rqt.isCollection())
        node = new CollectionNode(mPubSubModule, parent, topicId, from);
      else {
        node = new LeafNode(mPubSubModule, parent, topicId, from);
      }
      // Add the creator as the owner.
      node.addOwner(owner);
      setOptions(topic, node, rqt.getOptions());

      // Do the auto-subscription for creator.
      if (rqt.getOptions() != null && rqt.getOptions().isSubscribeOnCreate()) {
        NodeSubscription subscription = subscribeToNode(node, owner, owner);
        // TODO: not returning the subscription ID yet.
      }
      
      node.saveToDB();
      CacheFactory.doClusterTask(new RefreshNodeTask(node));
    }
//    LOGGER.trace("create node="+realTopic+" success");

    MMXStatus status = (new MMXStatus())
        .setCode(StatusCode.SUCCESS.getCode())
        .setMessage(StatusCode.SUCCESS.getMessage());
    return status;
  }
  
  private NodeSubscription subscribeToNode(Node node, JID owner, JID subscriber) {
    SubscribeForm optionsForm = new SubscribeForm(DataForm.Type.submit);
    // Receive notification of new items only.
    optionsForm.setSubscriptionType(NodeSubscription.Type.items);
    optionsForm.setSubscriptionDepth("all");
    // Don't set other options; it will change the subscription state to
    // pending because it will wait for the owner's approval.
    
    // Need a modified Node.java from mmx-openfire repo.
    NodeSubscription subscription = node.createSubscription(null, owner,
        subscriber, false, optionsForm);
    return subscription;
  }
  
  private int deleteNode(Node node, JID owner) throws MMXException {
    if (!node.getOwners().contains(owner)) {
      AppTopic topic = TopicHelper.parseTopic(node.getNodeID());
      throw new MMXException(StatusCode.FORBIDDEN.getMessage(topic.getName()),
          StatusCode.FORBIDDEN.getCode());
    }
    int count = 0;
    if (node.isCollectionNode()) {
      for (Node child : node.getNodes()) {
        if (child.isCollectionNode()) {
          count += deleteNode(child, owner);
        } else {
          if (!child.getOwners().contains(owner)) {
            AppTopic topic = TopicHelper.parseTopic(child.getNodeID());
            throw new MMXException(StatusCode.FORBIDDEN.getMessage(topic.getName()),
                StatusCode.FORBIDDEN.getCode());
          }
          
//          LOGGER.trace("delete leaf node=" + node.getNodeID());
          child.delete();
          ++count;
        }
      }
    }
//    LOGGER.trace("delete node="+node.getNodeID());
    node.delete();
    ++count;
    return count;
  }
  
  public MMXStatus deleteTopic(JID from, String appId, 
                          TopicAction.DeleteRequest rqt) throws MMXException {
    String topic = TopicHelper.normalizePath(rqt.getTopic());
    String userId = JIDUtil.getUserId(from);
    JID owner = from.asBareJID();
    String realTopic = TopicHelper.makeTopic(appId, rqt.isPersonal() ?
        userId : null, topic);
    Node node = mPubSubModule.getNode(realTopic);
    if (node == null) {
      throw new MMXException(StatusCode.TOPIC_NOT_FOUND.getMessage(topic), 
          StatusCode.TOPIC_NOT_FOUND.getCode());
    }
    int count = deleteNode(node, owner);    
    MMXStatus status = (new MMXStatus())
        .setCode(StatusCode.SUCCESS.getCode())
        .setMessage(count+" topic"+((count==1)?" is":"s are")+" deleted");
    return status;
  }
  
  public TopicInfo getTopic(JID from, String appId, MMXTopicId topic)
                          throws MMXException {
    String realTopic = TopicHelper.makeTopic(appId, topic.getEscUserId(),
        TopicHelper.normalizePath(topic.getName()));
    Node node = mPubSubModule.getNode(realTopic);
    if (node == null) {
      throw new MMXException(StatusCode.TOPIC_NOT_FOUND.getMessage(topic.getName()), 
          StatusCode.TOPIC_NOT_FOUND.getCode());
    }
    return nodeToInfo(topic.getUserId(), topic.getName(), node);
  }
  
  public MMXStatus retractAllFromTopic(JID from, String appId,
      TopicAction.RetractAllRequest rqt) throws MMXException {
    String topic = TopicHelper.normalizePath(rqt.getTopic());
    String userId = JIDUtil.getUserId(from);
    JID owner = from.asBareJID();
    String realTopic = TopicHelper.makeTopic(appId, rqt.isPersonal() ? 
        userId : null, topic);
    Node node = mPubSubModule.getNode(realTopic);
    if (node == null) {
      throw new MMXException(StatusCode.TOPIC_NOT_FOUND.getMessage(topic), 
          StatusCode.TOPIC_NOT_FOUND.getCode());
    }
    if (!node.getOwners().contains(owner)) {
      throw new MMXException(StatusCode.FORBIDDEN.getMessage(topic), 
          StatusCode.FORBIDDEN.getCode());
    }
    LeafNode leafNode = (LeafNode) node;
    List<PublishedItem> pubItems = leafNode.getPublishedItems();
    leafNode.deleteItems(pubItems);
    
    int count = (pubItems == null) ? 0 : pubItems.size();
    MMXStatus status = (new MMXStatus())
        .setCode(StatusCode.SUCCESS.getCode())
        .setMessage(count+" item"+((count==1)?" is":"s are")+" retracted");
    return status;
  }
  
  public MMXStatus retractFromTopic(JID from, String appId, 
      TopicAction.RetractRequest rqt) throws MMXException {
    String topic = TopicHelper.normalizePath(rqt.getTopic());
    String realTopic = TopicHelper.makeTopic(appId, rqt.getUserId(), topic);
    Node node = mPubSubModule.getNode(realTopic);
    if (node == null) {
      throw new MMXException(StatusCode.TOPIC_NOT_FOUND.getMessage(topic), 
          StatusCode.TOPIC_NOT_FOUND.getCode());
    }
    if (node.isCollectionNode()) {
      throw new MMXException("Cannot retract items from a collection topic",
          StatusCode.NOT_IMPLEMENTED.getCode());
    }
    
    LeafNode leafNode = (LeafNode) node;
    List<String> itemIds = rqt.getItemIds();
    if (itemIds == null) {
      throw new MMXException(StatusCode.BAD_REQUEST.getMessage("no item ID's"),
          StatusCode.BAD_REQUEST.getCode());
    }
    if (!leafNode.isItemRequired()) {
      // Cannot delete items from a leaf node that doesn't handle itemIDs
      throw new MMXException("Items required in this topic",
          StatusCode.NOT_IMPLEMENTED.getCode());
    }
    
    List<PublishedItem> pubItems = new ArrayList<PublishedItem>(itemIds.size());
    for (String itemId : itemIds) {
      if (itemId == null) {
        throw new MMXException(StatusCode.BAD_REQUEST.getMessage("null item ID"), 
            StatusCode.BAD_REQUEST.getCode());
      }
      PublishedItem item = leafNode.getPublishedItem(itemId);
      if (item == null) {
        throw new MMXException(StatusCode.ITEM_NOT_FOUND.getMessage(itemId), 
            StatusCode.ITEM_NOT_FOUND.getCode());
      }
      if (!item.canDelete(from)) {
        throw new MMXException(StatusCode.FORBIDDEN.getMessage(itemId), 
            StatusCode.FORBIDDEN.getCode());
      }
      pubItems.add(item);
    }
    leafNode.deleteItems(pubItems);
    
    int count = (pubItems == null) ? 0 : pubItems.size();
    MMXStatus status = (new MMXStatus())
        .setCode(StatusCode.SUCCESS.getCode())
        .setMessage(count+" item"+((count==1)?" is":"s are")+" retracted");
    return status;
  }
  
  public TopicAction.SubscribeResponse subscribeTopic(JID from, String appId, 
              TopicAction.SubscribeRequest rqt) throws MMXException {    
    String topic = TopicHelper.normalizePath(rqt.getTopic());
    String realTopic = TopicHelper.makeTopic(appId, rqt.getUserId(), topic);
    Node node = mPubSubModule.getNode(realTopic);
    if (node == null) {
      throw new MMXException(StatusCode.TOPIC_NOT_FOUND.getMessage(topic), 
          StatusCode.TOPIC_NOT_FOUND.getCode());
    }
    
    JID owner = from.asBareJID();
    // The subscriber can specify a different resource or without resource.
    JID subscriber = new JID(from.getNode(), from.getDomain(), rqt.getDevId());

    AccessModel accessModel = node.getAccessModel();
    if (!accessModel.canSubscribe(node, owner, subscriber)) {
      throw new MMXException(StatusCode.FORBIDDEN.getMessage(topic),
          StatusCode.FORBIDDEN.getCode());
    }
    
    // Check if the subscription owner is a user with outcast affiliation
    NodeAffiliate nodeAffiliate = node.getAffiliate(owner);
    if (nodeAffiliate != null &&
        nodeAffiliate.getAffiliation() == NodeAffiliate.Affiliation.outcast) {
      throw new MMXException(StatusCode.FORBIDDEN.getMessage(topic),
          StatusCode.FORBIDDEN.getCode());
    }
    
    // Check that subscriptions to the node are enabled
    if (!node.isSubscriptionEnabled()) {
      throw new MMXException(StatusCode.FORBIDDEN.getMessage(topic),
          StatusCode.FORBIDDEN.getCode());
    }

    // Check for duplicated subscription; return error or existing subscription.
    NodeSubscription subscription;
    if ((subscription = node.getSubscription(subscriber)) != null) {
      if (rqt.isErrorOnDup()) {
        throw new MMXException(StatusCode.SUBSCRIPTION_EXISTS.getMessage(topic),
          StatusCode.SUBSCRIPTION_EXISTS.getCode());
      } else {
        TopicAction.SubscribeResponse resp = new TopicAction.SubscribeResponse(
            subscription.getID(), StatusCode.SUCCESS.getCode(),
            StatusCode.SUCCESS.getMessage());
        return resp;
      }
    }
    
    subscription = subscribeToNode(node, owner, subscriber);
    
    TopicAction.SubscribeResponse resp = new TopicAction.SubscribeResponse(
        subscription.getID(), StatusCode.SUCCESS.getCode(), 
        StatusCode.SUCCESS.getMessage());
    return resp;
  }
  
  public MMXStatus unsubscribeTopic(JID from, String appId, 
                  TopicAction.UnsubscribeRequest rqt) throws MMXException {    
    String topic = TopicHelper.normalizePath(rqt.getTopic());
    String realTopic = TopicHelper.makeTopic(appId, rqt.getUserId(), topic);
    Node node = mPubSubModule.getNode(realTopic);
    if (node == null) {
      throw new MMXException(StatusCode.TOPIC_NOT_FOUND.getMessage(topic),
          StatusCode.TOPIC_NOT_FOUND.getCode());
    }
    int count = 0;
    JID owner = from.asBareJID();
    String subId = rqt.getSubId();
    for (NodeSubscription subscription : node.getSubscriptions(owner)) {
      if (subId == null || subscription.getID().equals(subId)) {
        ++count;
        node.cancelSubscription(subscription);
      }
    }
    if (count == 0) {
      throw new MMXException(StatusCode.GONE.getMessage(),
          StatusCode.GONE.getCode());
    }
    
    MMXStatus status = (new MMXStatus())
        .setCode(StatusCode.SUCCESS.getCode())
        .setMessage(count+" subscription"+((count==1)?" is":"s are")+" cancelled");
    return status;
  }
  
  public MMXStatus unsubscribeForDev(JID from, String appId, 
                  TopicAction.UnsubscribeForDevRequest rqt) throws MMXException {
    String prefix = TopicHelper.TOPIC_DELIM + appId + TopicHelper.TOPIC_DELIM;
    int count = 0;
    JID owner = from.asBareJID();
    String devId = rqt.getDevId();
    for (Node node : mPubSubModule.getNodes()) {
      if (!node.getNodeID().startsWith(prefix)) {
        continue;
      }
      for (NodeSubscription subscription : node.getSubscriptions(owner)) {
        if (devId.equals(subscription.getJID().getResource())) {
          ++count;
          node.cancelSubscription(subscription);
        }
      }
    }
    MMXStatus status = (new MMXStatus())
        .setCode(StatusCode.SUCCESS.getCode())
        .setMessage(count+" subscription"+((count==1)?" is":"s are")+" cancelled");
    return status;
  }
  
  public TopicAction.ListResponse listTopics(JID from, String appId,
                  TopicAction.ListRequest rqt) throws MMXException {
    TopicAction.ListResponse resp = new TopicAction.ListResponse();
    Integer maxLimit = rqt.getLimit();
    boolean recursive = rqt.isRecursive();
    String start = rqt.getStart();
    ListType type = rqt.getType();
    int limit = (maxLimit == null || maxLimit == -1) ? Integer.MAX_VALUE : maxLimit;
    String realTopic;
    if (start == null || start.isEmpty()) {
      // Get the global root node.
      Node node = mPubSubModule.getNode(appId);
      if (node == null) {
        throw new MMXException(StatusCode.APP_NODE_NOT_FOUND.getMessage(appId),
            StatusCode.APP_NODE_NOT_FOUND.getCode());
      }
      // Get its top level children nodes filtered by the search type.
      String userId = JIDUtil.getUserId(from);
      limit = getTopChildNodes(recursive, node, resp, limit, type, userId);
    } else {
      // Filter by the global topics.
      if (type == ListType.global || type == ListType.both) {
        realTopic = TopicHelper.makeTopic(appId, null, start);
        // Get the global topic node first.
        Node node = mPubSubModule.getNode(realTopic);
        if (node == null) {
          throw new MMXException(StatusCode.TOPIC_NOT_FOUND.getMessage(start),
                StatusCode.TOPIC_NOT_FOUND.getCode());
        }
        // Get its children nodes without search filter.
        limit = getChildNodes(recursive, node, resp, limit);
      }
      // Filter by the user topics.
      if (type == ListType.personal || type == ListType.both) {
        realTopic = TopicHelper.makeTopic(appId, JIDUtil.getUserId(from), start);
        // Get the personal topic node first.
        Node node = mPubSubModule.getNode(realTopic);
        if (node == null) {
          throw new MMXException(StatusCode.TOPIC_NOT_FOUND.getMessage(start),
                StatusCode.TOPIC_NOT_FOUND.getCode());
        }
        // Get its children nodes without search filter.
        limit = getChildNodes(recursive, node, resp, limit);
      }
    }
    return resp;
  }
  
  private TopicInfo nodeToInfo(String userId, String topic, Node node) {
    TopicInfo info = new TopicInfo(
        userId, node.getName() != null ? node.getName() : topic, node.isCollectionNode())
      .setCreationDate(node.getCreationDate())
      .setDescription(node.getDescription())
      .setModifiedDate(node.getModificationDate())
      .setCreator(node.getCreator().toString())
      .setSubscriptionEnabled(node.isSubscriptionEnabled());
    if (!node.isCollectionNode()) {
      LeafNode leafNode = (LeafNode) node;
      info.setMaxItems(leafNode.isPersistPublishedItems() ?
            leafNode.getMaxPublishedItems() : 0)
          .setMaxPayloadSize(leafNode.getMaxPayloadSize())
          .setPersistent(leafNode.isPersistPublishedItems())
          .setPublisherType(ConfigureForm.convert(leafNode.getPublisherModel()));
    } else {
      info.setMaxItems(0)
          .setMaxPayloadSize(0)
          .setPersistent(false)
          .setPublisherType(null);
    }
    return info;
  }
  
  // Get all top level children nodes which are either global topics or user
  // topics.  Filter out the nodes by the search type.
  private int getTopChildNodes(boolean recursive, Node root, 
      TopicAction.ListResponse resp, int limit, ListType type, String userId) {
    boolean globalOnly = (type == ListType.global);
    for (Node child : root.getNodes()) {
      AppTopic topic = TopicHelper.parseTopic(child.getNodeID());
      if (topic == null) {
        LOGGER.warn("Ignore malformed topic: " + child.getNodeID());
        continue;
      }
      // Brain teaser: the xor below is actually same as
      //    (type == ListType.global && !topic.isUserTopic()) ||
      //    (type == ListType.personal && topic.isUserTopic()))
      if ((type == ListType.both) || (globalOnly ^ topic.isUserTopic())) {
        if (topic.isUserTopic() && !topic.getEscUserId().equals(userId)) {
          continue;
        }
        if (--limit < 0) {
          return limit;
        }
        resp.add(nodeToInfo(topic.getUserId(), topic.getName(), child));        
        if (recursive && child.isCollectionNode()) {
          limit = getChildNodes(recursive, child, resp, limit);
        }
      }
    }
    return limit;
  }
  
  // Get all children nodes recursively below the top level.  
  // @return < 0 if exceeding the limit, >= 0 if within the limit.
  private int getChildNodes(boolean recursive, Node node, 
      TopicAction.ListResponse resp, int limit) {
    if (!node.isCollectionNode()) {
      AppTopic topic = TopicHelper.parseTopic(node.getNodeID());
      if (topic == null) {
        LOGGER.warn("Ignore malformed topic: " + node.getNodeID());
      } else {
        if (--limit < 0) {
          return limit;
        }
        resp.add(nodeToInfo(topic.getUserId(), topic.getName(), node));
      }
    } else {
      for (Node child : node.getNodes()) {
        AppTopic topic = TopicHelper.parseTopic(child.getNodeID());
        if (topic == null) {
          LOGGER.warn("Ignore malformed topic: " + child.getNodeID());
          continue;
        }
        if (--limit < 0) {
          return limit;
        }
        resp.add(nodeToInfo(topic.getUserId(), topic.getName(), child));
        if (recursive && child.isCollectionNode()) {
              limit = getChildNodes(recursive, child, resp, limit);
        }
      }
    }
    return limit;
  }
  
  private String[] getTopics(String appId, List<MMXTopicId> list, 
                              int begin, int size) {
    if (list == null) {
      return null;
    }
    size = Math.min(size, list.size() - begin);
    String[] topics = new String[size];
    for (int i = 0, index = begin; --size >= 0; index++, i++) {
      MMXTopicId topicNode = list.get(index);
      topics[i] = TopicHelper.makeTopic(appId, topicNode.getEscUserId(),
                                        topicNode.getName());
    }
    return topics;
  }
  
  private static class SQLHelper {
    public static String generateArgList(int numOfArgs) {
      if (numOfArgs == 0) {
        return "";
      }
      StringBuilder sb = new StringBuilder((numOfArgs-1) * 2 + 1);
      while (--numOfArgs > 0) {
        sb.append("?,");
      }
      sb.append('?');
      return sb.toString();
    }

    public static void bindArgList(PreparedStatement pstmt, int index,
                                      String[] args) throws SQLException {
      for (String arg : args) {
        pstmt.setString(index++, arg);
      }
    }
    
    public static void bindArgList(PreparedStatement pstmt, int index, 
                                      List<String> args) throws SQLException {
      for (String arg : args) {
        pstmt.setString(index++, arg);
      }
    }
  }
  
  public TopicAction.SummaryResponse getSummary(JID from, String appId,
          TopicAction.SummaryRequest rqt) throws MMXException {
    // Build a collection of topic ID's from the request; it contains topics
    // without any published items.
    HashSet<String> tpNoItems = new HashSet<String>(rqt.getTopicNodes().size());
    for (MMXTopicId topicId : rqt.getTopicNodes()) {
      tpNoItems.add(topicId.toString());
    }
    TopicAction.SummaryResponse resp = new TopicAction.SummaryResponse(
        rqt.getTopicNodes().size());
    Connection con = null;
    PreparedStatement pstmt = null;
    ResultSet rs = null;
    try {
      con = DbConnectionManager.getConnection();
      int start = 0;
      int numOfTopics = rqt.getTopicNodes().size();
      StringBuilder dateRange = new StringBuilder();
      ArrayList<String> dates = new ArrayList<String>();
      if (rqt.getSince() != null) {
        dateRange.append("AND ofPubsubItem.creationDate >= ? ");
        dates.add(StringUtils.dateToMillis(rqt.getSince()));
      }
      if (rqt.getUntil() != null) {
        dateRange.append("AND ofPubsubItem.creationDate <= ? ");
        dates.add(StringUtils.dateToMillis(rqt.getUntil()));
      }
      do {
        // Limit to 128 topics per query because some DBMS cannot handle more
        // than 255 arguments in the IN clause.
        String[] topics = getTopics(appId, rqt.getTopicNodes(), start, 128);
        if (topics == null || topics.length == 0) {
          break;
        }
        String argList = SQLHelper.generateArgList(topics.length);
        String sql = "SELECT ofPubsubNode.maxItems,count(*),max(ofPubsubItem.creationDate),ofPubsubNode.nodeId "+
                     "FROM ofPubsubItem, ofPubsubNode " +
                     "WHERE ofPubsubItem.serviceID=? AND " +
                     "      ofPubsubItem.nodeID = ofPubsubNode.nodeId AND " +
                     "      ofPubsubItem.nodeID IN ("+argList+") "+dateRange+
                     "GROUP BY ofPubsubItem.nodeID";
        pstmt = con.prepareStatement(sql);
        pstmt.setString(1, mPubSubModule.getServiceID());
        SQLHelper.bindArgList(pstmt, 2, topics);
        SQLHelper.bindArgList(pstmt, 2+topics.length, dates);
        rs = pstmt.executeQuery();
        while (rs.next()) {
          int maxItems = rs.getInt(1);
          int count = rs.getInt(2);
          Date creationDate = new Date(Long.parseLong(rs.getString(3).trim()));
          String nodeId = rs.getString(4);
          MMXTopicId topic = TopicHelper.parseNode(nodeId);
          resp.add(new TopicSummary(topic)
            .setCount((maxItems < 0) ? count : Math.min(maxItems, count))
            .setLastPubTime(creationDate));
          // This topic has published items; remove it from the collection.
          tpNoItems.remove(topic.toString());
        }
        start += topics.length;
      } while (start < numOfTopics);
      // Fill the response with the topics having no published items.
      Iterator<String> it = tpNoItems.iterator();
      while (it.hasNext()) {
        String topicId = it.next();
        resp.add(new TopicSummary(MMXTopicId.parse(topicId)).setCount(0));
      }
      return resp;
    } catch (Exception sqle) {
      LOGGER.error(sqle.getMessage(), sqle);
{'fraction_non_alphanumeric': 0.11172136637642331, 'fraction_numerical': 0.0017861129716454567, 'mean_word_length': 3.0326370757180157, 'pattern_counts': {'":': 10, '<': 71, '<?xml version=': 0, '>': 67, 'https://': 4, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '425234', 'n_tokens_mistral': 26871, 'n_tokens_neox': 26736, 'n_words': 5709}