package org.gcube.gcat.persistence.ckan;

import java.io.StringWriter;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import javax.ws.rs.BadRequestException;
import javax.ws.rs.ForbiddenException;
import javax.ws.rs.InternalServerErrorException;
import javax.ws.rs.NotAllowedException;
import javax.ws.rs.WebApplicationException;
import javax.ws.rs.core.MultivaluedMap;

import org.apache.http.MethodNotSupportedException;
import org.gcube.com.fasterxml.jackson.databind.JsonNode;
import org.gcube.com.fasterxml.jackson.databind.node.ArrayNode;
import org.gcube.com.fasterxml.jackson.databind.node.ObjectNode;
import org.gcube.common.scope.impl.ScopeBean.Type;
import org.gcube.gcat.api.GCatConstants;
import org.gcube.gcat.api.configuration.CatalogueConfiguration;
import org.gcube.gcat.api.moderation.CMItemStatus;
import org.gcube.gcat.api.moderation.CMItemVisibility;
import org.gcube.gcat.api.moderation.Moderated;
import org.gcube.gcat.api.moderation.ModerationContent;
import org.gcube.gcat.api.roles.Role;
import org.gcube.gcat.configuration.CatalogueConfigurationFactory;
import org.gcube.gcat.moderation.thread.ModerationThread;
import org.gcube.gcat.oldutils.Validator;
import org.gcube.gcat.profile.MetadataUtility;
import org.gcube.gcat.social.SocialPost;
import org.gcube.gcat.utils.URIResolver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author Luca Frosini (ISTI - CNR)
 */
public class CKANPackage extends CKAN implements Moderated {
	
	private static final Logger logger = LoggerFactory.getLogger(CKANPackage.class);
	
	/*
	// see https://docs.ckan.org/en/latest/api/#ckan.logic.action.get.package_list
	public static final String ITEM_LIST = CKAN.CKAN_API_PATH + "package_list";
	*/
	// see https://docs.ckan.org/en/latest/api/index.html#ckan.logic.action.get.package_search
	public static final String ITEM_LIST = CKAN.CKAN_API_PATH + "package_search";
	// see https://docs.ckan.org/en/latest/api/#ckan.logic.action.create.package_create
	public static final String ITEM_CREATE = CKAN.CKAN_API_PATH + "package_create";
	// see https://docs.ckan.org/en/latest/api/#ckan.logic.action.get.package_show
	public static final String ITEM_SHOW = CKAN.CKAN_API_PATH + "package_show";
	// see https://docs.ckan.org/en/latest/api/#ckan.logic.action.update.package_update
	public static final String ITEM_UPDATE = CKAN.CKAN_API_PATH + "package_update";
	// see https://docs.ckan.org/en/latest/api/#ckan.logic.action.patch.package_patch
	public static final String ITEM_PATCH = CKAN.CKAN_API_PATH + "package_patch";
	// see https://docs.ckan.org/en/latest/api/#ckan.logic.action.delete.package_delete
	public static final String ITEM_DELETE = CKAN.CKAN_API_PATH + "package_delete";
	// see https://docs.ckan.org/en/latest/api/#ckan.logic.action.delete.dataset_purge
	public static final String ITEM_PURGE = CKAN.CKAN_API_PATH + "dataset_purge";
	
	
	// limit in https://docs.ckan.org/en/latest/api/index.html#ckan.logic.action.get.package_search
	protected static final String ROWS_KEY = "rows";
	// offset in https://docs.ckan.org/en/latest/api/index.html#ckan.logic.action.get.package_search
	protected static final String START_KEY = "start";
	
	protected static final String ORGANIZATION_FILTER_TEMPLATE = GCatConstants.ORGANIZATION_PARAMETER + ":%s";
	
	protected static final String ORGANIZATION_REGEX = GCatConstants.ORGANIZATION_PARAMETER + ":[a-zA-Z0-9_\\-\"]*";
	
	protected static final Pattern ORGANIZATION_REGEX_PATTERN;
	
	static {
		ORGANIZATION_REGEX_PATTERN = Pattern.compile(ORGANIZATION_REGEX);
	}
	
	protected static final String LICENSE_KEY = "license_id";
	
	protected static final String EXTRAS_ITEM_URL_KEY = "Item URL";
	
	protected static final String AUTHOR_KEY = "author";
	protected static final String AUTHOR_EMAIL_KEY = "author_email";
	
	protected static final String MAINTAINER_KEY = "maintainer";
	protected static final String MAINTAINER_EMAIL_KEY = "maintainer_email";
	
	protected static final String OWNER_ORG_KEY = "owner_org";
	protected static final String RESOURCES_KEY = "resources";
	protected static final String TITLE_KEY = "title";
	
	public static final String EXTRAS_KEY = "extras";
	public static final String EXTRAS_KEY_KEY = "key";
	public static final String EXTRAS_KEY_VALUE_SYSTEM_TYPE = "system:type";
	public static final String EXTRAS_VALUE_KEY = "value";
	
	// The 'results' array is included in the 'result' object for package_search
	private static final String RESULTS_KEY = "results";
	
	protected static final String PRIVATE_KEY = "private";
	protected static final String SEARCHABLE_KEY = "searchable";
	protected static final String CAPACITY_KEY = "capacity";
	
	protected static final String CM_STATUS_QUERY_FILTER_KEY = "extras_systemcm_item_status";
	
	protected static final String INCLUDE_PRIVATE_KEY = "include_private";
	// protected static final String INCLUDE_DRAFTS_KEY = "include_drafts";
	
	public static final String GROUPS_KEY = "groups";
	public static final String TAGS_KEY = "tags";
	
	protected final List<CKANResource> managedResources;
	
	protected String itemID;
	protected String itemTitle;
	protected String itemURL;
	
	protected final CKANUser ckanUser;
	
	protected final CatalogueConfiguration configuration;
	
	protected ModerationThread moderationThread;
	
	public CKANPackage() {
		super();
		
		LIST = ITEM_LIST;
		CREATE = ITEM_CREATE;
		READ = ITEM_SHOW;
		UPDATE = ITEM_UPDATE;
		PATCH = ITEM_PATCH;
		DELETE = ITEM_DELETE;
		PURGE = ITEM_PURGE;
		managedResources = new ArrayList<CKANResource>();
		
		ckanUser = CKANUserCache.getCurrrentCKANUser();
		
		configuration = CatalogueConfigurationFactory.getInstance();
		
		for(String supportedOrganization : configuration.getSupportedOrganizations()) {
			ckanUser.addUserToOrganization(supportedOrganization);
		}
		
	}
	
	protected CKANOrganization checkGotOrganization(String gotOrganization) throws ForbiddenException {
		if(!configuration.getSupportedOrganizations().contains(gotOrganization)) {
			String error = String.format(
					"IS Configuration does not allow to publish in %s organizations. Allowed organization are: %s",
					gotOrganization, configuration.getSupportedOrganizations());
			throw new ForbiddenException(error);
		}
		
		CKANOrganization ckanOrganization = new CKANOrganization();
		ckanOrganization.setName(gotOrganization);
		ckanOrganization.read();
		
		return ckanOrganization;
	}
	
	protected CKANOrganization getPublishingOrganization(ObjectNode objectNode) throws ForbiddenException {
		CKANOrganization ckanOrganization = null;
		if(objectNode.has(OWNER_ORG_KEY)) {
			String gotOrganizationName = objectNode.get(OWNER_ORG_KEY).asText();
			ckanOrganization = checkGotOrganization(gotOrganizationName);
		}
		
		if(ckanOrganization == null) {
			// owner organization must be specified if the token belongs to a VRE
			String organizationFromContext = configuration.getDefaultOrganization();
			ckanOrganization = checkGotOrganization(organizationFromContext);
			objectNode.put(OWNER_ORG_KEY, organizationFromContext);
		}
		
		return ckanOrganization;
	}
	
	public ObjectNode checkBaseInformation(String json) throws Exception {
		return checkBaseInformation(json, false);
	}
	
	public JsonNode cleanResult(JsonNode jsonNode) {
		if(jsonNode.has(OWNER_ORG_KEY)) {
			((ObjectNode) jsonNode).remove(OWNER_ORG_KEY);
		}
		
		// Removing all Content Moderation Keys
		if(jsonNode.has(EXTRAS_KEY)) {
			ArrayNode extras = (ArrayNode) jsonNode.get(EXTRAS_KEY);
			// It is not possible to remove the element of an array while iterating it.
			// We need to create a new array only with valie elements;
			ArrayNode newExtras = mapper.createArrayNode();
			boolean foundOne = false;
			for(int i=0; i<extras.size(); i++) {
				JsonNode extra = extras.get(i);
				if(extra.has(EXTRAS_KEY_KEY) && 
				   extra.get(EXTRAS_KEY_KEY)!=null && 
				   extra.get(EXTRAS_KEY_KEY).asText().startsWith(Moderated.SYSTEM_CM_PREFIX)) {
					foundOne = true;
				}else {
					newExtras.add(extra.deepCopy());
				}
			}
			
			if(foundOne) {
				((ObjectNode) jsonNode).replace(EXTRAS_KEY, newExtras);
			}
		}
		return jsonNode;
	}
	
	protected String getAsCleanedString(JsonNode node) {
		JsonNode jsonNode = cleanResult(node);
		return getAsString(jsonNode);
	}
	
	/**
	 * @param json The json to check
	 * @param allowPartialInfo used for patch method which provide only partial information (i.e. the info to patch) 
	 * @return ObjectNode from json
	 * @throws Exception
	 */
	public ObjectNode checkBaseInformation(String json, boolean allowPartialInfo) throws Exception {
		ObjectNode objectNode = (ObjectNode) mapper.readTree(json);
		
		try {
			objectNode = (ObjectNode) checkName(objectNode);
		}catch (Exception e) {
			if(!allowPartialInfo) {
				throw e;
			}
		}
		
		// We need to enforce the itemID to properly manage resource persistence
		if(objectNode.has(ID_KEY)) {
			itemID = objectNode.get(ID_KEY).asText();
		}
		
		// To include private item in search result (e.g. listing) a private package must be searchable
		// The package it is just included in the search and listing results but remain private and cannot be accessed
		// if not authorized
		if(objectNode.has(PRIVATE_KEY)) {
			boolean privatePackage = objectNode.get(PRIVATE_KEY).asBoolean();
			if(privatePackage) {
				/*
				 * This version is not properly managed by CKAN.
				 * It is converted to: 
				 * 		searchable: "true" 
				 * which is incidentally considered as true value even is not correct. 
				 * We need to provide a string with T as capitol letters to make it working 
				 * objectNode.put(SEARCHABLE_KEY, true);
				 */
				objectNode.put(SEARCHABLE_KEY, "True");
			}
		}
		
		// check license
		String licenseId = null;
		if(objectNode.has(LICENSE_KEY)) {
			licenseId = objectNode.get(LICENSE_KEY).asText();
		}
		
		if(licenseId!=null && !licenseId.isEmpty()) {
			try {
				CKANLicense.checkLicenseId(licenseId);
			} catch(Exception e) {
				throw new BadRequestException(
						"You must specify an existing license identifier for the item. License list can be retrieved using licence collection");
			}
		}else if(!allowPartialInfo) {
			throw new BadRequestException(
					"You must specify a license identifier for the item. License list can be retrieved using licence collection");
		}
		
		if(objectNode.has(CAPACITY_KEY)) {
			/*
			 * When a client provides the 'capacity' field as 'private', the item is not counted in the 
			 * total number of items in the GUI. We want to avoid such a behavior
			 *  See https://support.d4science.org/issues/16410
			 */
			objectNode.remove(CAPACITY_KEY);
		}
		
		String authorName = ckanUser.getSurnameName();
		if(authorName==null || authorName.compareTo("")==0) {
			authorName = ckanUser.getName();
		}
		objectNode.put(AUTHOR_KEY, authorName);
		String authorEmail = ckanUser.getEMail();
		objectNode.put(AUTHOR_EMAIL_KEY, authorEmail);
		
		
		if(!objectNode.has(MAINTAINER_KEY)) {
			if(!objectNode.has(MAINTAINER_EMAIL_KEY)) {
				objectNode.put(MAINTAINER_KEY, authorName);
				objectNode.put(MAINTAINER_EMAIL_KEY, authorEmail);
			}else {
				objectNode.put(MAINTAINER_KEY, objectNode.get(MAINTAINER_EMAIL_KEY).toString());
			}
		}
		
		getPublishingOrganization(objectNode);
		
		return objectNode;
	}
	
	protected JsonNode validateJson(String json) {
		try {
			// check base information (and set them if needed)
			ObjectNode objectNode = checkBaseInformation(json);
			
			// Validating against profiles if any
			MetadataUtility metadataUtility = new MetadataUtility();
			if(!metadataUtility.getMetadataProfiles().isEmpty()) {
				Validator validator = new Validator(mapper);
				objectNode = validator.validateAgainstProfile(objectNode, metadataUtility);
			}
			
			return objectNode;
		} catch(BadRequestException e) {
			throw e;
		} catch(Exception e) {
			throw new BadRequestException(e);
		}
	}
	
	protected Map<String,String> addFieldsFilters(Map<String,String> parameters, String... requiredFields){
		StringBuffer stringBuffer = new StringBuffer();
		stringBuffer.append("[");
		stringBuffer.append("'");
		stringBuffer.append(ID_KEY);
		stringBuffer.append("'");
		stringBuffer.append(",");
		stringBuffer.append("'");
		stringBuffer.append(NAME_KEY);
		stringBuffer.append("'");
		for(String requiredField : requiredFields) {
			if(requiredField!=null && requiredField.compareTo("")!=0) {
				stringBuffer.append(",");
				stringBuffer.append("'");
				stringBuffer.append(requiredField);
				stringBuffer.append("'");
			}
		}
		stringBuffer.append("]");
		parameters.put("fl", stringBuffer.toString());
		return parameters;
	}
	
	protected Map<String,String> getListingParameters(int limit, int offset, String... requiredFields) {
		Map<String,String> parameters = new HashMap<>();
		if(limit <= 0) {
			// According to CKAN documentation
			// the number of matching rows to return. There is a hard limit of 1000 datasets per query.
			// see https://docs.ckan.org/en/2.6/api/index.html#ckan.logic.action.get.package_search
			limit = 1000;
		}
		parameters.put(ROWS_KEY, String.valueOf(limit));
		
		if(offset < 0) {
			offset = 0;
		}
		parameters.put(START_KEY, String.valueOf(offset));
		// parameters.put(START_KEY, String.valueOf(pageNumber * limit));
		
		MultivaluedMap<String,String> queryParameters = uriInfo.getQueryParameters();
		parameters = checkListParameters(queryParameters, parameters);
		
		parameters = addFieldsFilters(parameters, requiredFields);
		
		parameters = addModerationStatusFilter(parameters);
		
		return parameters;
	}
	
	
	protected void reuseInstance() {
		this.name = null;
		this.result = null;
		this.itemID = null;
		this.itemURL = null;
		this.itemTitle = null;
	}
	
	/**
	 * @param purge indicate if the item 
	 * @return the name list of deleted items
	 */
	public String deleteAll(boolean purge){
		MultivaluedMap<String,String> queryParameters = uriInfo.getQueryParameters();
		if(queryParameters.containsKey(GCatConstants.OWN_ONLY_QUERY_PARAMETER)) {
			if(ckanUser.getRole().ordinal() < Role.ADMIN.ordinal()) {
				queryParameters.remove(GCatConstants.OWN_ONLY_QUERY_PARAMETER);
				queryParameters.add(GCatConstants.OWN_ONLY_QUERY_PARAMETER, Boolean.TRUE.toString());
			}		
		}else {
			queryParameters.add(GCatConstants.OWN_ONLY_QUERY_PARAMETER, Boolean.TRUE.toString());
		}
		
		int limit = 25;
		int offset = 0;
		Map<String,String> parameters = getListingParameters(limit,offset, "resources");
		ObjectNode objectNode = mapper.createObjectNode();
		
		ArrayNode deleted = mapper.createArrayNode();
		ArrayNode notDeleted = mapper.createArrayNode();
		
		sendGetRequest(LIST, parameters);
		ArrayNode results = (ArrayNode) result.get(RESULTS_KEY);
		
		Set<String> notDeletedSet = new HashSet<>();
		
		while(results.size()>0) {
			int alreadyTriedAndNotDeletedAgain = 0;
			for(JsonNode node : results) {
				try {
					this.reuseInstance();
					this.result = node;
					this.name = node.get(NAME_KEY).asText();
					this.itemID = node.get(ID_KEY).asText();
					delete(purge);
					deleted.add(name);
					if(notDeletedSet.contains(name)) {
						notDeletedSet.remove(name);
					}
					try {
						Thread.sleep(TimeUnit.MILLISECONDS.toMillis(300));
					} catch (InterruptedException e) {
						
					}
				} catch(Exception e) {
					try {
						if(name!=null) {
							if(notDeletedSet.contains(name)) {
								alreadyTriedAndNotDeletedAgain++;
							}else {
								notDeleted.add(name);
								notDeletedSet.add(name);
							}
							logger.error("Error while trying to delete item with name {}", name);
						}else {
							logger.error("Unable to get the name of {}.",mapper.writeValueAsString(node));
						}
					} catch(Exception ex) {
						logger.error("", ex);
					}
				}
			}
			
			try {
				Thread.sleep(TimeUnit.SECONDS.toMillis(5));
			} catch (InterruptedException e) {
				
			}
			if(purge) {
				setApiKey(CKANUtility.getApiKey());
			}
			
			if(limit==alreadyTriedAndNotDeletedAgain) {
				offset++;
				parameters = getListingParameters(limit,offset, "resources");
			}
			
			sendGetRequest(LIST, parameters);
			results = (ArrayNode) result.get(RESULTS_KEY);
			
		}
		
		if(notDeleted.size()!=notDeletedSet.size()) {
			notDeleted = mapper.createArrayNode();
			for(String name : notDeletedSet) {
				notDeleted.add(name);
			}
		}
		
		objectNode.set("deleted", deleted);
		objectNode.set("failed", notDeleted);
		
		return getAsString(objectNode);
	}
	
	public String list(Map<String,String> parameters) {
		sendGetRequest(LIST, parameters);
		
		ArrayNode results = (ArrayNode) result.get(RESULTS_KEY);
		
		ArrayNode arrayNode = mapper.createArrayNode();
		for(JsonNode node : results) {
			try {
				String name = node.get(NAME_KEY).asText();
				arrayNode.add(name);
			} catch(Exception e) {
				try {
					logger.error("Unable to get the name of {}. The Item will not be included in the result",
							mapper.writeValueAsString(node));
				} catch(Exception ex) {
					logger.error("", ex);
				}
			}
		}
		
		return getAsString(arrayNode);
	}
	
	@Override
	public String list(int limit, int offset) {
		Map<String,String> parameters = getListingParameters(limit, offset);
		return list(parameters);
	}
	
	public int count() {
		Map<String,String> parameters = getListingParameters(1, 0);

		sendGetRequest(LIST, parameters);
		
		int count = result.get(GCatConstants.COUNT_KEY).asInt();
		return count;
	}

	protected Set<String> checkOrganizationFilter(String q) {
		Matcher m = ORGANIZATION_REGEX_PATTERN.matcher(q);
		
		Set<String> matches = new HashSet<>();
		while(m.find()) {
			matches.add(q.substring(m.start(), m.end()).replace(GCatConstants.ORGANIZATION_PARAMETER+":", ""));
		}
		
		return matches;
	}
	
	protected static String[] allowedListQueryParameters = new String[] {"fq", "fq_list", "sort",
			/* "facet", "facet.mincount", "facet.limit", "facet.field", */
			"include_drafts", "include_private", "ext_bbox"};
	
	protected String getFilterForOrganizations() {
		StringWriter stringWriter = new StringWriter();
		int i=1;
		for(String organizationName : configuration.getSupportedOrganizations()) {
			stringWriter.append(String.format(GCatConstants.ORGANIZATION_FILTER_TEMPLATE, organizationName));
			if(i!=configuration.getSupportedOrganizations().size()) {
				// Please note that an item can only belong to a single organization. 
				// Hence the query must put supported organizations in OR.
				stringWriter.append(" OR ");
			}
			i++;
		}
		return stringWriter.toString();
	}
	
	protected Map<String,String> checkListParameters(MultivaluedMap<String,String> queryParameters,
			Map<String,String> parameters) {
		
		String q = null;
		if(queryParameters.containsKey(GCatConstants.Q_KEY)) {
			q = queryParameters.getFirst(GCatConstants.Q_KEY);
		}
		
		if(q != null) {
			Set<String> organizations = checkOrganizationFilter(q);
			
			if(organizations.size()==0) {
				// Adding organization filter to q
				String filter = getFilterForOrganizations();
				q = String.format("%s AND %s", q, filter);
			}else {
				organizations.removeAll(configuration.getSupportedOrganizations());
				if(organizations.size()>0) {
					String error = String.format("It is not possible to query the following organizations %s. Supported organization in this context are %s", organizations.toString(), configuration.getSupportedOrganizations().toString());
					throw new ForbiddenException(error);
				}
			}
			
		} else {
			String filter = getFilterForOrganizations();
			q = filter;
		}
		
		if(queryParameters.containsKey(GCatConstants.OWN_ONLY_QUERY_PARAMETER)) {
			if(!queryParameters.get(GCatConstants.OWN_ONLY_QUERY_PARAMETER).isEmpty() && Boolean.parseBoolean(queryParameters.get(GCatConstants.OWN_ONLY_QUERY_PARAMETER).get(0))) {
				String filter =  String.format("%s:%s", AUTHOR_EMAIL_KEY, ckanUser.getEMail());
				q = String.format("%s AND %s", q, filter);
			}
		}
		
		parameters.put(GCatConstants.Q_KEY, q);
		
		for(String key : allowedListQueryParameters) {
			if(queryParameters.containsKey(key)) {
				parameters.put(key, queryParameters.getFirst(key));
			}
		}
		
		// parameters.put(INCLUDE_PRIVATE_KEY, String.valueOf(true));
		
		// By default not including draft
		// parameters.put(INCLUDE_DRAFTS_KEY, String.valueOf(false));
		
		return parameters;
	}
	
	protected void rollbackManagedResources() {
		for(CKANResource ckanResource : managedResources) {
			try {
				ckanResource.rollback();
			} catch(Exception e) {
				logger.error("Unable to rollback resource {} to the original value", ckanResource.getResourceID());
			}
			
		}
	}
	
	protected ArrayNode createResources(ArrayNode resourcesToBeCreated) {
		ArrayNode created = mapper.createArrayNode();
		for(JsonNode resourceNode : resourcesToBeCreated) {
			CKANResource ckanResource = new CKANResource(itemID);
			String json = ckanResource.create(getAsString(resourceNode));
			created.add(getAsJsonNode(json));
			managedResources.add(ckanResource);
		}
		return created;
	}
	
	protected JsonNode addExtraField(JsonNode jsonNode, String key, String value) {
		ArrayNode extras = null;
		boolean found = false;
		if(jsonNode.has(EXTRAS_KEY)) {
			extras = (ArrayNode) jsonNode.get(EXTRAS_KEY);
			for(JsonNode extra : extras) {
				if(extra.has(EXTRAS_KEY_KEY) && extra.get(EXTRAS_KEY_KEY)!=null && extra.get(EXTRAS_KEY_KEY).asText().compareTo(key) == 0) {
					((ObjectNode) extra).put(EXTRAS_VALUE_KEY, value);
					found = true;
					break;
				}
			}
		} else {
			extras = ((ObjectNode) jsonNode).putArray(EXTRAS_KEY);
		}
		
		if(!found) {
			ObjectNode extra = mapper.createObjectNode();
			extra.put(EXTRAS_KEY_KEY, key);
			extra.put(EXTRAS_VALUE_KEY, value);
			extras.add(extra);
		}
		
		return jsonNode;
	}
	
	protected JsonNode getExtraField(JsonNode jsonNode, String key) {
		if(jsonNode.has(EXTRAS_KEY)) {
			ArrayNode extras = (ArrayNode) jsonNode.get(EXTRAS_KEY);
			for(JsonNode extra : extras) {
				if(extra.has(EXTRAS_KEY_KEY) && extra.get(EXTRAS_KEY_KEY)!=null && extra.get(EXTRAS_KEY_KEY).asText().compareTo(key) == 0) {
					return extra.get(EXTRAS_VALUE_KEY);
				}
			}
		}
		return null;
	}
	
	
	protected String addItemURLViaResolver(JsonNode jsonNode) {
		// Adding Item URL via Resolver
		itemURL = URIResolver.getCatalogueItemURL(name);
		addExtraField(jsonNode, EXTRAS_ITEM_URL_KEY, itemURL);
		return itemURL;
	}
	
	protected void sendSocialPost() {
		try {
			boolean makePost = false;
			try {
				MultivaluedMap<String,String> queryParameters = uriInfo.getQueryParameters();
				if(queryParameters.containsKey(GCatConstants.SOCIAL_POST_QUERY_PARAMETER)) {
					makePost = Boolean.parseBoolean(queryParameters.getFirst(GCatConstants.SOCIAL_POST_QUERY_PARAMETER));
				}
			} catch(Exception e) {
				makePost = false;
			}
			
			if(makePost) {
				ArrayNode arrayNode = (ArrayNode) result.get(TAGS_KEY);
				SocialPost socialPost = new SocialPost();
				socialPost.setItemID(itemID);
				socialPost.setItemURL(itemURL);
				socialPost.setItemTitle(itemTitle);
				socialPost.setTags(arrayNode);
				
				Boolean notification = null;
				try {
					MultivaluedMap<String,String> queryParameters = uriInfo.getQueryParameters();
					if(queryParameters.containsKey(GCatConstants.SOCIAL_POST_NOTIFICATION_QUERY_PARAMETER)) {
						notification = Boolean.parseBoolean(queryParameters.getFirst(GCatConstants.SOCIAL_POST_NOTIFICATION_QUERY_PARAMETER));
					}
				} catch(Exception e) {
					
				}
				socialPost.setNotification(notification);
				socialPost.start();
			} else {
				logger.info("The request explicitly disabled the Social Post.");
			}
		} catch(Exception e) {
			logger.warn(
					"error dealing with Social Post. The service will not raise the exception belove. Please contact the administrator to let him know about this message.",
					e);
		}
	}
	
	protected boolean isItemCreator() {
		return result.get(AUTHOR_EMAIL_KEY).asText().compareTo(ckanUser.getEMail())==0;
	}
	
	protected void parseResult() {
		if(this.itemID == null) {
			this.itemID = result.get(ID_KEY).asText();
		}
		this.itemTitle = result.get(TITLE_KEY).asText();
		this.itemURL = getExtraField(result, EXTRAS_ITEM_URL_KEY).asText();
	}
	
	
	protected void readItem() throws Exception {
		if(this.result == null) {
			String ret = super.read();
			this.result = mapper.readTree(ret);
		}
		parseResult();
	}
	
	
	@Override
	public String read() {
		try {
			readItem();
			checkModerationRead();
			return getAsCleanedString(result);
		} catch(WebApplicationException e) {
			throw e;
		} catch(Exception e) {
			throw new InternalServerErrorException(e);
		}
	}

	// see https://docs.ckan.org/en/latest/api/#ckan.logic.action.create.package_create
	@Override
	public String create(String json) {
		try {
			logger.debug("Going to create Item {}", json);
			
			if(ckanUser.getRole().ordinal() < Role.EDITOR.ordinal()) {
				StringBuffer stringBuffer = new StringBuffer();
				stringBuffer.append("Only ");
				stringBuffer.append(Role.EDITOR.getPortalRole());
				stringBuffer.append(" and ");
				stringBuffer.append(Role.ADMIN.getPortalRole());
				stringBuffer.append(" are entitled to create items. ");
				stringBuffer.append("Please contact the VRE Manager to request your grant.");
				throw new ForbiddenException(stringBuffer.toString()); 
			}
			
			JsonNode jsonNode = validateJson(json);
			
			setItemToPending(jsonNode);
			
			ArrayNode resourcesToBeCreated = mapper.createArrayNode();
			if(jsonNode.has(RESOURCES_KEY)) {
				resourcesToBeCreated = (ArrayNode) jsonNode.get(RESOURCES_KEY);
				((ObjectNode) jsonNode).remove(RESOURCES_KEY);
			}
			
			if(configuration.getScopeBean().is(Type.VRE)) {
				addItemURLViaResolver(jsonNode);
			}
			
			super.create(getAsString(jsonNode));
			
			parseResult();
			ArrayNode created = createResources(resourcesToBeCreated);
			((ObjectNode) result).replace(RESOURCES_KEY, created);
			
			postItemCreated();
			
			if(!isModerationEnabled()) {
				if(configuration.getScopeBean().is(Type.VRE)) {
					// Actions performed after a package has been correctly created on ckan.
					sendSocialPost();
				}
			}
			
			return getAsCleanedString(result);
		} catch(WebApplicationException e) {
			rollbackManagedResources();
			throw e;
		} catch(Exception e) {
			rollbackManagedResources();
			throw new InternalServerErrorException(e);
		}
	}
	
	// see https://docs.ckan.org/en/latest/api/#ckan.logic.action.update.package_update
	@Override
	public String update(String json) {
		try {
			JsonNode jsonNode = validateJson(json);
			
			/*
			 * Going to read the item from CKAN just to check the item status.
			 * I need to reset the result first because the current contains 
			 * the extras as sent by the client which are not trusted 
			 */
			this.result = null;
			readItem();
			
			jsonNode = checkModerationUpdate(jsonNode);
			
			Map<String,CKANResource> originalResources = new HashMap<>();
			ArrayNode originalResourcesarrayNode = (ArrayNode) result.get(RESOURCES_KEY);
			if(originalResources != null) {
				for(JsonNode resourceNode : originalResourcesarrayNode) {
					CKANResource ckanResource = new CKANResource(itemID);
					ckanResource.setPreviousRepresentation(resourceNode);
					String resourceID = ckanResource.getResourceID();
					originalResources.put(resourceID, ckanResource);
				}
			}
			
			if(jsonNode.has(RESOURCES_KEY)) {
				ArrayNode resourcesToBeSend = mapper.createArrayNode();
				ArrayNode receivedResources = (ArrayNode) jsonNode.get(RESOURCES_KEY);
				for(JsonNode resourceNode : receivedResources) {
					CKANResource ckanResource = new CKANResource(itemID);
					String resourceId = CKANResource.extractResourceID(resourceNode);
					if(resourceId != null) {
						if(originalResources.containsKey(resourceId)) {
							ckanResource = originalResources.get(resourceId);
							originalResources.remove(resourceId);
						} else {
							throw new BadRequestException(
									"The content contains a resource with id " + resourceId + " which does not exists");
						}
					}
					if(originalResources.get(resourceId) != null
							&& (!originalResources.get(resourceId).getPreviousRepresentation().equals(resourceNode))) {
						resourceNode = ckanResource.createOrUpdate(resourceNode);
					}
					resourcesToBeSend.add(resourceNode);
					managedResources.add(ckanResource);
					
				}
				((ObjectNode) jsonNode).replace(RESOURCES_KEY, resourcesToBeSend);
			}
			
			addItemURLViaResolver(jsonNode);
			
			sendPostRequest(ITEM_UPDATE, getAsString(jsonNode));
			
			for(String resourceId : originalResources.keySet()) {
				CKANResource ckanResource = originalResources.get(resourceId);
				ckanResource.deleteFile();
			}
			
			postItemUpdated();
			
			return getAsCleanedString(result);
		} catch(WebApplicationException e) {
			rollbackManagedResources();
			throw e;
		} catch(Exception e) {
			rollbackManagedResources();
			throw new InternalServerErrorException(e);
		}
	}
	
	// see https://docs.ckan.org/en/latest/api/#ckan.logic.action.patch.package_patch
	@Override
	public String patch(String json) {
		try {
			readItem();
			
			JsonNode jsonNode = checkBaseInformation(json, true);
			((ObjectNode)jsonNode).put(ID_KEY, this.itemID);
			
			jsonNode = checkModerationUpdate(jsonNode);
			
			Map<String,CKANResource> originalResources = new HashMap<>();
			ArrayNode originalResourcesarrayNode = (ArrayNode) result.get(RESOURCES_KEY);
			if(originalResources != null) {
				for(JsonNode resourceNode : originalResourcesarrayNode) {
					CKANResource ckanResource = new CKANResource(itemID);
					ckanResource.setPreviousRepresentation(resourceNode);
					String resourceID = ckanResource.getResourceID();
					originalResources.put(resourceID, ckanResource);
				}
			}
			
			if(jsonNode.has(RESOURCES_KEY)) {
				ArrayNode resourcesToBeSend = mapper.createArrayNode();
				ArrayNode receivedResources = (ArrayNode) jsonNode.get(RESOURCES_KEY);
				for(JsonNode resourceNode : receivedResources) {
					CKANResource ckanResource = new CKANResource(itemID);
					String resourceId = CKANResource.extractResourceID(resourceNode);
					if(resourceId != null) {
						if(originalResources.containsKey(resourceId)) {
							ckanResource = originalResources.get(resourceId);
							originalResources.remove(resourceId);
						} else {
							throw new BadRequestException(
									"The content contains a resource with id " + resourceId + " which does not exists");
						}
					}
					if(originalResources.get(resourceId) != null
							&& (!originalResources.get(resourceId).getPreviousRepresentation().equals(resourceNode))) {
						resourceNode = ckanResource.createOrUpdate(resourceNode);
					}
					resourcesToBeSend.add(resourceNode);
					managedResources.add(ckanResource);
					
				}
				((ObjectNode) jsonNode).replace(RESOURCES_KEY, resourcesToBeSend);
			}
			
			addItemURLViaResolver(jsonNode);
			
			sendPostRequest(ITEM_PATCH, getAsString(jsonNode));
			
			parseResult();
			
			for(String resourceId : originalResources.keySet()) {
				CKANResource ckanResource = originalResources.get(resourceId);
				ckanResource.deleteFile();
			}
			
			postItemUpdated();
			
			return getAsCleanedString(result);
		} catch(WebApplicationException e) {
			rollbackManagedResources();
			throw e;
		} catch(Exception e) {
			rollbackManagedResources();
			throw new InternalServerErrorException(e);
		}
	}
	
	@Override
	protected void delete() {
		checkModerationDelete();
		super.delete();
	}
	
	@Override
	public void purge() {
		try {
			setApiKey(CKANUtility.getSysAdminAPI());
			readItem();
			
			checkModerationDelete();
			
			if(ckanUser.getRole().ordinal() < Role.ADMIN.ordinal() && !isItemCreator()) {
				throw new ForbiddenException("Only " + Role.ADMIN.getPortalRole() + "s and item creator are entitled to purge an item");
			}
			
			if(result.has(RESOURCES_KEY)) {
				itemID = result.get(ID_KEY).asText();
				ArrayNode arrayNode = (ArrayNode) result.get(RESOURCES_KEY);
				for(JsonNode jsonNode : arrayNode) {
					CKANResource ckanResource = new CKANResource(itemID);
					ckanResource.setPreviousRepresentation(jsonNode);
					ckanResource.deleteFile(); // Only delete file is required because the item will be purged at the end
				}
			}
			super.purge();
		} catch(WebApplicationException e) {
			throw e;
		} catch(Exception e) {
			throw new InternalServerErrorException(e);
		}
	}
	
	/**
	 * Used for bulk purging. Internal use only
	 */
	protected void purgeNoCheckNoDeleteFiles() {
//		setApiKey(CKANUtility.getSysAdminAPI());
//		super.purge();
	}
	
	/* 
	 * --------------------------------------------------------------------------------------------------------
	 * Moderation Related functions
	 * --------------------------------------------------------------------------------------------------------
	 * 
	 */
	
	protected CMItemStatus getCMItemStatus() {
		
		String cmItemStatusString = CMItemStatus.APPROVED.getValue();
		boolean found = false;
		if(result.has(EXTRAS_KEY)) {
			ArrayNode extras = (ArrayNode) result.get(EXTRAS_KEY);
			for(JsonNode extra : extras) {
				if(extra.has(EXTRAS_KEY_KEY) && extra.get(EXTRAS_KEY_KEY).asText().compareTo(Moderated.SYSTEM_CM_ITEM_STATUS) == 0) {
					cmItemStatusString = extra.get(EXTRAS_VALUE_KEY).asText();
					found = true;
					break;
				}
			}
		}
		
		CMItemStatus cmItemStatus = CMItemStatus.getCMItemStatusFromValue(cmItemStatusString);
		
		if(!found) {
			// The item was published before activating the moderation. 
			// The item is considered as approved and the item representation must be updated
			setToApproved(result);
			String ret = sendPostRequest(ITEM_UPDATE, getAsString(result));
			try {
				result = mapper.readTree(ret);
			}catch (Exception e) {
				new InternalServerErrorException(e);
			}
		}
		
		return cmItemStatus;
	}
	
	protected CMItemStatus getRequestedCMItemStatus() {
		CMItemStatus cmItemStatus = null;
		try {
			MultivaluedMap<String,String> queryParameters = uriInfo.getQueryParameters();
			if(queryParameters.containsKey(Moderated.CM_ITEM_STATUS_QUERY_PARAMETER)) {
				String cmItemStatusString = queryParameters.getFirst(Moderated.CM_ITEM_STATUS_QUERY_PARAMETER);
				cmItemStatus = CMItemStatus.getCMItemStatusFromValue(cmItemStatusString);
			}
		}catch (Exception e) {
			cmItemStatus = null;
		}
		return cmItemStatus;
	}
	
	protected boolean isModerationEnabled() {
		boolean moderationEnabled = configuration.isModerationEnabled();
		if(moderationEnabled && moderationThread==null) {
			moderationThread = ModerationThread.getDefaultInstance();
			moderationThread.setCKANUser(ckanUser);
		}
		return moderationEnabled;
	}
	
	protected Map<String,String> addModerationStatusFilter(Map<String,String> parameters){
		if(isModerationEnabled()) {
			String q = parameters.get(GCatConstants.Q_KEY);
			
			CMItemStatus cmItemStatus = getRequestedCMItemStatus();
			
			this.apiKey = CKANUtility.getSysAdminAPI();
			
			if(!ckanUser.isCatalogueModerator()) {
				q = String.format("%s AND %s:%s", q, AUTHOR_EMAIL_KEY, ckanUser.getEMail());
				parameters.put(GCatConstants.Q_KEY,	q);
				
				switch (ckanUser.getRole()) {
					case ADMIN:
					case MANAGER:
					case EDITOR:
						break;
						
					case MEMBER:
						if(cmItemStatus!=null && cmItemStatus!=CMItemStatus.APPROVED) {
							throw new ForbiddenException("You are only authorized to list " + CMItemStatus.APPROVED.getValue() + " items");
						}
						break;
						
					default:
						break;
				}
			}
			
			if(cmItemStatus==null) {
				cmItemStatus = CMItemStatus.APPROVED;
			}
			
			StringBuffer stringBuffer = new StringBuffer();
			stringBuffer.append("(");
			stringBuffer.append(CM_STATUS_QUERY_FILTER_KEY);
			stringBuffer.append(":");
			stringBuffer.append(cmItemStatus.getValue());
			
			if(cmItemStatus == CMItemStatus.APPROVED) {
				stringBuffer.append(" OR (*:* -");
				stringBuffer.append(CM_STATUS_QUERY_FILTER_KEY);
				stringBuffer.append(":[* TO *])");
			}
			stringBuffer.append(")");
			q = String.format("%s AND %s", q, stringBuffer.toString());
			parameters.put(GCatConstants.Q_KEY,	q);
		
			
			parameters.put(INCLUDE_PRIVATE_KEY, String.valueOf(true));
		}else{
			if(ckanUser.getRole().ordinal()>=Role.ADMIN.ordinal()) {
				parameters.put(INCLUDE_PRIVATE_KEY, String.valueOf(true));
			}
		}
		
		return parameters;
	}
	
	protected void checkModerationRead() {
		if(isModerationEnabled()) {
			CMItemStatus cmItemStatus = getCMItemStatus();
			if(cmItemStatus == CMItemStatus.APPROVED) {
				return;
			}
			
			if(isItemCreator()) {
				// The author is entitled to read its own items independently from the status
				return;
			}
			
			if(ckanUser.getRole().ordinal() >= Role.ADMIN.ordinal() || ckanUser.isCatalogueModerator()) {
				// Catalogue-Admin and Catalogue-Moderator are entitled to read items with any statues
				return;
			}		
			
			throw new ForbiddenException("You are not entitled to read the item");
		}
	}
	
	protected JsonNode checkModerationUpdate(JsonNode jsonNode) {
		if(isModerationEnabled()) {
			CMItemStatus cmItemStatus = getCMItemStatus();
			
			boolean setToPending = true;
			
			switch (cmItemStatus) {
				case APPROVED:
					if(ckanUser.getRole().ordinal() < Role.ADMIN.ordinal() && !isItemCreator()) {
						throw new ForbiddenException("Only " + Role.ADMIN.getPortalRole() + "s and item creator are entitled to update an " + cmItemStatus.getValue() + " item");
					}
					if(ckanUser.getRole().ordinal() >= Role.ADMIN.ordinal()) {
						setToApproved(jsonNode);
						setToPending = false;
					}
					break;
					
				case PENDING:
					if(isItemCreator()) {
						break;
					}
					if(ckanUser.isCatalogueModerator()) {
						break;
					}
					throw new ForbiddenException("You are not entitled to update a " + cmItemStatus.getValue() + " item");
					
				case REJECTED:
					if(isItemCreator()) {
						break;
					}
					if(ckanUser.isCatalogueModerator()) {
						break;
					}
					throw new ForbiddenException("You are not entitled to update a " + cmItemStatus.getValue() + " item");
					
				default:
					break;
			}
			
			if(setToPending) {
				setItemToPending(jsonNode);
			}
			
		}
		
		return jsonNode;
	}
	
	protected void checkModerationDelete() {
		try {
			if(isModerationEnabled()) {
				readItem();
				
				if(ckanUser.getRole().ordinal() >= Role.ADMIN.ordinal()) {
					// Ad Admin can delete any item independently from the status
					return;
				}
				
				CMItemStatus cmItemStatus = getCMItemStatus();
				
				switch (cmItemStatus) {
					case APPROVED:
						if(isItemCreator()) {
							break;
						}
						throw new ForbiddenException("Only " + Role.ADMIN.getPortalRole() + "s and item creator are entitled to delete an " + cmItemStatus.getValue() + " item");
						
					case REJECTED:
						if(isItemCreator()) {
							break;
						}
						if(ckanUser.isCatalogueModerator()) {
							break;
						}
						throw new ForbiddenException("You are not entitled to delete a " + cmItemStatus.getValue() + " item");
						
					case PENDING:
						if(isItemCreator()) {
							break;
						}
						if(ckanUser.isCatalogueModerator()) {
							break;
						}
						throw new ForbiddenException("You are not entitled to update a " + cmItemStatus.getValue() + " item");
					
					default:
						break;
				}		
			}
		} catch(WebApplicationException e) {
			throw e;
		} catch(Exception e) {
			throw new InternalServerErrorException(e);
		}
	}
	
	protected void setToRejected(JsonNode jsonNode) {
		addExtraField(jsonNode, Moderated.SYSTEM_CM_ITEM_STATUS, CMItemStatus.REJECTED.getValue());
	}
	
	protected void setItemToPending(JsonNode jsonNode) {
		if(isModerationEnabled()) {
			addExtraField(jsonNode, Moderated.SYSTEM_CM_ITEM_STATUS, CMItemStatus.PENDING.getValue());
			
			CMItemVisibility cmItemVisibility = CMItemVisibility.PUBLIC;
			
			if(jsonNode.has(PRIVATE_KEY)) {
				boolean privatePackage = jsonNode.get(PRIVATE_KEY).asBoolean();
				if(privatePackage) {
					cmItemVisibility = CMItemVisibility.RESTRICTED;
				}
			}
			addExtraField(jsonNode, Moderated.SYSTEM_CM_ITEM_VISIBILITY, cmItemVisibility.getValue());
			
			((ObjectNode) jsonNode).put(PRIVATE_KEY, true);
			/*
			 * This version is not properly managed by CKAN.
			 * It is converted to: 
			 * 		searchable: "false" 
			 * which is considered as true value. 
			 * We need to provide a string with F as capitol letters to make it working 
			 * ((ObjectNode) jsonNode).put(SEARCHABLE_KEY, false);
			 */
			((ObjectNode) jsonNode).put(SEARCHABLE_KEY, "False");
		}
	}
	
	protected void setToApproved(JsonNode jsonNode) {
		ArrayNode extras = (ArrayNode) jsonNode.get(EXTRAS_KEY);
		
		boolean approvedSet = false;
		CMItemVisibility cmItemVisibility = null;
		
		for(JsonNode extra : extras) {
			if(extra.has(EXTRAS_KEY_KEY) && extra.get(EXTRAS_KEY_KEY)!=null && extra.get(EXTRAS_KEY_KEY).asText().compareTo(Moderated.SYSTEM_CM_ITEM_STATUS) == 0) {
				((ObjectNode) extra).put(EXTRAS_VALUE_KEY, CMItemStatus.APPROVED.getValue());
				approvedSet = true;
			}
			
			if(extra.has(EXTRAS_KEY_KEY) && extra.get(EXTRAS_KEY_KEY)!=null && extra.get(EXTRAS_KEY_KEY).asText().compareTo(Moderated.SYSTEM_CM_ITEM_VISIBILITY) == 0) {
				cmItemVisibility = CMItemVisibility.getCMItemStatusFromValue(extra.get(EXTRAS_VALUE_KEY).asText());
			}
		}
		
		if(!approvedSet) {
			addExtraField(jsonNode, Moderated.SYSTEM_CM_ITEM_STATUS, CMItemStatus.APPROVED.getValue());
		}
		
		if(cmItemVisibility==null) {
			cmItemVisibility = CMItemVisibility.PUBLIC;
			addExtraField(jsonNode, Moderated.SYSTEM_CM_ITEM_VISIBILITY, cmItemVisibility.getValue());
		}
		
		
		boolean privateItem = cmItemVisibility == CMItemVisibility.RESTRICTED ? true :false;
		((ObjectNode) jsonNode).put(PRIVATE_KEY, privateItem);
		if(privateItem) {
			((ObjectNode) jsonNode).put(SEARCHABLE_KEY, "True");
		}else {
			((ObjectNode) jsonNode).remove(SEARCHABLE_KEY);
		}
	}
	
	private void postItemCreated() throws Exception {
		try {
			if(isModerationEnabled()) {
				moderationThread.setItemCoordinates(itemID, name, itemTitle, itemURL);
				moderationThread.postItemCreated();
			}
		} catch(WebApplicationException e) {
			throw e;
		} catch(Exception e) {
			throw new InternalServerErrorException(e);
		}
	}
	
	private void postItemUpdated() {
		try {
			if(isModerationEnabled()) {
				moderationThread.setItemCoordinates(itemID, name, itemTitle, itemURL);
				moderationThread.postItemUpdated();
			}
		} catch(WebApplicationException e) {
			throw e;
		} catch(Exception e) {
			throw new InternalServerErrorException(e);
		}
	}
	
	@Override
	public String approve(String moderatorMessage) {
		try {
			if(isModerationEnabled()) {
				readItem();
				CMItemStatus cmItemStatus = getCMItemStatus();
				switch (cmItemStatus) {
					case APPROVED:
						// Nothing TO DO
						break;
	
					case REJECTED:
						throw new MethodNotSupportedException("You can't approve a rejected item. The item must be updated first. The update will set the item in pending, than it can be approved/rejected.");
					
					case PENDING:
						if(!ckanUser.isCatalogueModerator()) {
							throw new MethodNotSupportedException("Only catalogue moderator can approve a pending item.");
						}
						setToApproved(result);
						
						// Need to use sysadmin because the user could not have the right to modify the item
						setApiKey(CKANUtility.getSysAdminAPI());
						String ret = sendPostRequest(ITEM_UPDATE, getAsString(result));
						// Resetting the api key
						setApiKey(null);
						
						result = mapper.readTree(ret);
						
						parseResult();
						moderationThread.setItemCoordinates(itemID, name, itemTitle,  itemURL);
						moderationThread.postItemApproved(moderatorMessage);
						
						if(configuration.getScopeBean().is(Type.VRE)) {
							// Actions performed after a package has been correctly created on ckan.
							sendSocialPost();
						}
						
						break;
						
					default:
						break;
				}
				return getAsCleanedString(result);
			}
			throw new MethodNotSupportedException("The approve operation is available only in moderation mode");
		}catch(WebApplicationException e) {
			throw e;
		} catch(Exception e) {
			throw new InternalServerErrorException(e);
		}
	}
	
	@Override
	public String reject(String moderatorMessage) {
		try {
			if(isModerationEnabled()) {
				readItem();
				CMItemStatus cmItemStatus = getCMItemStatus();
				switch (cmItemStatus) {
					case APPROVED:
						throw new MethodNotSupportedException("You can't rejected an approved item. The item must be updated first. The update will set the item in pending, than it can be approved/rejected.");
	
					case REJECTED:
						// Nothing TO DO
						break;
					
					case PENDING:
						if(!ckanUser.isCatalogueModerator()) {
							throw new MethodNotSupportedException("Only catalogue moderator can reject a pending item.");
						}
						
						setToRejected(result);
						
						// Need to use sysadmin because the user could not have the right to modify the item
						setApiKey(CKANUtility.getSysAdminAPI());
						String ret = sendPostRequest(ITEM_PATCH, getAsString(result));
						// Resetting the api key
						setApiKey(null);
						
						result = mapper.readTree(ret);
						parseResult();
						moderationThread.setItemCoordinates(itemID, name, itemTitle, itemURL);
						moderationThread.postItemRejected(moderatorMessage);
						break;
						
					default:
						break;
				}
				return getAsCleanedString(result);
			}
			throw new MethodNotSupportedException("The reject operation is available only in moderation mode");
		}catch(WebApplicationException e) {
			throw e;
		} catch(Exception e) {
			throw new InternalServerErrorException(e);
		}
	}
	
	@Override
	public void message(String message) {
		try {
			if(isModerationEnabled()) {
				if(message==null || message.compareTo("")==0) {
					return;
				}
				
				readItem();
				
				// Catalogue Moderators are allowed to post message to the dedicated Stream
				if(!ckanUser.isCatalogueModerator()) {
					// Users that are not 
					if(!isItemCreator()) {
						throw new NotAllowedException("Only item creator and " + Moderated.CATALOGUE_MODERATOR + "s are entitled to partecipate to the moderation discussion thread.");
					}else {
						moderationThread.setItemAuthor(true);
					}
				}

				CMItemStatus cmItemStatus = getCMItemStatus();
				moderationThread.setItemCoordinates(itemID, name, itemTitle, itemURL);
				moderationThread.postUserMessage(cmItemStatus, message);
				return;
			}
			throw new MethodNotSupportedException("The message operation is available only in moderation mode");
		}catch(WebApplicationException e) {
			throw e;
		} catch(Exception e) {
			throw new InternalServerErrorException(e);
		}
	}
	
	public String moderate(String json) {
		try {
			ModerationContent moderationContent = mapper.readValue(json, ModerationContent.class);
			String message = moderationContent.getMessage(); 
			if(moderationContent.getCMItemStatus() !=null) {
				CMItemStatus cmItemStatus = moderationContent.getCMItemStatus();
				switch (cmItemStatus) {
					case APPROVED:
						return approve(message);
					case REJECTED:
						return reject(message);
					default:
						throw new BadRequestException("Allowed moderation operations are approve, reject and message");
				}
			}else {
				if(message==null || message.length()==0) {
					throw new BadRequestException("Allowed moderation operations are approve, reject and message");
				}
				message(message);
				return null;
			}
		}catch(WebApplicationException e) {
			throw e;
		} catch(Exception e) {
			throw new InternalServerErrorException(e);
		}
	}

	
	
}