package org.gcube.datacatalogue.ckanutillibrary.server;

import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;

import java.io.BufferedReader;
import java.io.File;
import java.io.InputStreamReader;
import java.math.BigDecimal;
import java.net.URLEncoder;
import java.nio.charset.Charset;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

import net.htmlparser.jericho.Renderer;
import net.htmlparser.jericho.Segment;
import net.htmlparser.jericho.Source;

import org.gcube.datacatalogue.ckanutillibrary.server.models.CKanUserWrapper;
import org.gcube.datacatalogue.ckanutillibrary.server.models.CkanDatasetRelationship;
import org.gcube.datacatalogue.ckanutillibrary.server.models.DatasetRelationships;
import org.gcube.datacatalogue.ckanutillibrary.server.models.ResourceBean;
import org.gcube.datacatalogue.ckanutillibrary.server.models.State;
import org.gcube.datacatalogue.ckanutillibrary.server.utils.UtilMethods;
import org.gcube.datacatalogue.ckanutillibrary.server.utils.url.EntityContext;
import org.gcube.datacatalogue.ckanutillibrary.shared.RolesCkanGroupOrOrg;
import org.json.simple.JSONArray;
import org.json.simple.JSONObject;
import org.json.simple.JSONValue;
import org.json.simple.parser.JSONParser;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import eu.trentorise.opendata.jackan.CheckedCkanClient;
import eu.trentorise.opendata.jackan.CkanClient;
import eu.trentorise.opendata.jackan.exceptions.JackanException;
import eu.trentorise.opendata.jackan.internal.org.apache.http.HttpEntity;
import eu.trentorise.opendata.jackan.internal.org.apache.http.HttpResponse;
import eu.trentorise.opendata.jackan.internal.org.apache.http.HttpStatus;
import eu.trentorise.opendata.jackan.internal.org.apache.http.client.methods.HttpGet;
import eu.trentorise.opendata.jackan.internal.org.apache.http.client.methods.HttpPost;
import eu.trentorise.opendata.jackan.internal.org.apache.http.entity.ContentType;
import eu.trentorise.opendata.jackan.internal.org.apache.http.entity.StringEntity;
import eu.trentorise.opendata.jackan.internal.org.apache.http.entity.mime.MultipartEntityBuilder;
import eu.trentorise.opendata.jackan.internal.org.apache.http.impl.client.CloseableHttpClient;
import eu.trentorise.opendata.jackan.internal.org.apache.http.impl.client.HttpClientBuilder;
import eu.trentorise.opendata.jackan.internal.org.apache.http.util.EntityUtils;
import eu.trentorise.opendata.jackan.model.CkanDataset;
import eu.trentorise.opendata.jackan.model.CkanGroup;
import eu.trentorise.opendata.jackan.model.CkanLicense;
import eu.trentorise.opendata.jackan.model.CkanOrganization;
import eu.trentorise.opendata.jackan.model.CkanPair;
import eu.trentorise.opendata.jackan.model.CkanResource;
import eu.trentorise.opendata.jackan.model.CkanTag;
import eu.trentorise.opendata.jackan.model.CkanUser;

/**
 * This is the Ckan Utils implementation class.
 * @author Costantino Perciante at ISTI-CNR (costantino.perciante@isti.cnr.it)
 */
public class DataCatalogueImpl implements DataCatalogue{

	private static final Logger logger = LoggerFactory.getLogger(DataCatalogueImpl.class);

	private String CKAN_CATALOGUE_URL;
	private String CKAN_DB_NAME;
	private String CKAN_DB_USER;
	private String CKAN_DB_PASSWORD;
	private String CKAN_DB_URL;
	private Integer CKAN_DB_PORT;
	private String PORTLET_URL_FOR_SCOPE;
	private String CKAN_TOKEN_SYS;
	private String URI_RESOLVER_URL;
	private boolean MANAGE_PRODUCT_BUTTON;
	private String CONTEXT;

	// ckan client
	private CkanClient client;

	// hashmap for ckan api keys
	private ConcurrentHashMap<String, CKANTokenBean> apiKeysMap;

	// apikey bean expires after X minutes in the above map
	private static final int EXPIRE_KEY_TIME = 5 * 60 * 1000;

	private class CKANTokenBean{
		private String apiKey;
		private long timestamp;
		public CKANTokenBean(String apiKey, long timestamp) {
			super();
			this.apiKey = apiKey;
			this.timestamp = timestamp;
		}
	}

	/**
	 * The ckan catalogue url and database will be discovered in this scope
	 * @param scope
	 * @throws Exception if unable to find datacatalogue info
	 */
	public DataCatalogueImpl(String scope) throws Exception{

		DataCatalogueRunningCluster runningInstance = new DataCatalogueRunningCluster(scope);

		// save information
		CKAN_DB_URL = runningInstance.getDatabaseHosts().get(0).trim();
		CKAN_DB_NAME = runningInstance.getDataBaseName().trim();
		CKAN_DB_USER = runningInstance.getDataBaseUser().trim();
		CKAN_DB_PASSWORD = runningInstance.getDataBasePassword().trim();
		CKAN_TOKEN_SYS = runningInstance.getSysAdminToken().trim();
		CKAN_DB_PORT = runningInstance.getDatabasePorts().get(0);
		CKAN_CATALOGUE_URL = runningInstance.getDataCatalogueUrl().get(0).trim();
		PORTLET_URL_FOR_SCOPE = runningInstance.getPortletUrl().trim();
		MANAGE_PRODUCT_BUTTON = runningInstance.isManageProductEnabled();
		URI_RESOLVER_URL = runningInstance.getUrlResolver();

		logger.debug("Plain sys admin token first 3 chars are " + CKAN_TOKEN_SYS.substring(0, 3));
		logger.debug("Plain db password first 3 chars are " + CKAN_DB_PASSWORD.substring(0, 3));

		// build the client
		client = new CkanClient(CKAN_CATALOGUE_URL);

		// init map
		apiKeysMap = new ConcurrentHashMap<String, CKANTokenBean>();

		// save the context
		CONTEXT = scope;

	}

	@Override
	public String getCatalogueUrl() {
		return CKAN_CATALOGUE_URL;
	}


	@Override
	public String getPortletUrl() {
		return PORTLET_URL_FOR_SCOPE;
	}

	@Override
	public String getUriResolverUrl() {
		return URI_RESOLVER_URL;
	}

	/**
	 * Retrieve connection from the pool
	 * @return a connection available within the pool
	 * @throws SQLException 
	 * @throws ClassNotFoundException 
	 */
	private Connection getConnection() throws SQLException, ClassNotFoundException{

		logger.debug("CONNECTION REQUEST");
		// create db connection
		Class.forName("org.postgresql.Driver");
		Connection connection = DriverManager.getConnection(
				"jdbc:postgresql://" + CKAN_DB_URL + ":" + CKAN_DB_PORT + "/" + CKAN_DB_NAME, CKAN_DB_USER, CKAN_DB_PASSWORD);
		return connection;

	}

	/**
	 * Check if the manage product is enabled
	 * @return
	 */
	@Override
	public boolean isManageProductEnabled() {
		return MANAGE_PRODUCT_BUTTON;
	}

	/**
	 * Tries to close a connection
	 * @param connection
	 */
	private void closeConnection(Connection connection){

		if(connection != null){
			try{
				connection.close();
			}catch(Exception e){
				logger.error("Unable to close this connection ", e);
			}
		}

	}

	@Override
	public String getApiKeyFromUsername(String username) {

		logger.debug("Request api key for user = " + username);

		// checks
		checkNotNull(username);
		checkArgument(!username.isEmpty());

		// in order to avoid errors, the username is always converted
		String ckanUsername = UtilMethods.fromUsernameToCKanUsername(username);

		// check in the hashmap first
		if(apiKeysMap.containsKey(ckanUsername)){
			CKANTokenBean bean = apiKeysMap.get(ckanUsername);
			if((bean.timestamp + EXPIRE_KEY_TIME) > System.currentTimeMillis()){ // it's still ok
				return bean.apiKey;
			}
		}

		logger.debug("Api key was not in cache or it expired");

		// the connection
		Connection connection = null;
		String apiToReturn = null;

		try{

			connection = getConnection();

			String query = "SELECT \"apikey\" FROM \"user\" WHERE \"name\"=? and \"state\"=?;";
			PreparedStatement preparedStatement = connection.prepareStatement(query);
			preparedStatement.setString(1, ckanUsername);
			preparedStatement.setString(2, State.ACTIVE.toString().toLowerCase());

			ResultSet rs = preparedStatement.executeQuery();
			while (rs.next()) {
				apiToReturn = rs.getString("apikey");
				logger.debug("Api key retrieved for user " + ckanUsername);
				break;
			}

			// save into the hashmap
			if(apiToReturn != null)
				apiKeysMap.put(ckanUsername, new CKANTokenBean(apiToReturn, System.currentTimeMillis()));

			return apiToReturn;

		}catch(Exception e){
			logger.error("Unable to retrieve key for user " + ckanUsername, e);
			return null;
		}finally{
			closeConnection(connection);
		}
	}

	@Override
	public CKanUserWrapper getUserFromApiKey(String apiKey) {
		logger.debug("Request user whose api key is = " + apiKey.substring(0, 3) + "*************");

		// checks
		checkNotNull(apiKey);
		checkArgument(!apiKey.isEmpty());

		CKanUserWrapper user = null;

		// the connection
		Connection connection = null;

		try{

			connection = getConnection();
			String query = "SELECT * FROM \"user\" WHERE \"apikey\"=? and \"state\"=?;";
			PreparedStatement preparedStatement = connection.prepareStatement(query);
			preparedStatement.setString(1, apiKey);
			preparedStatement.setString(2, State.ACTIVE.toString().toLowerCase());

			ResultSet rs = preparedStatement.executeQuery();
			while (rs.next()) {

				user = new CKanUserWrapper();
				user.setId(rs.getString("id"));
				user.setName(rs.getString("name"));
				user.setApiKey(apiKey);
				user.setCreationTimestamp(rs.getTimestamp("created").getTime());
				user.setAbout(rs.getString("about"));
				user.setOpenId(rs.getString("openid"));
				user.setFullName(rs.getString("fullname"));
				user.setEmail(rs.getString("email"));
				user.setAdmin(rs.getBoolean("sysadmin"));

				logger.debug("User retrieved");
				break;
			}
		}catch(Exception e){
			logger.error("Unable to retrieve user with api key " + apiKey, e);
		}finally{
			closeConnection(connection);
		}
		return user;
	}

	@Override
	public List<CkanOrganization> getOrganizationsByUser(String username) {

		logger.debug("Requested organizations for user " + username);

		// checks
		checkNotNull(username);

		// in order to avoid errors, the username is always converted
		String ckanUsername = UtilMethods.fromUsernameToCKanUsername(username);

		// list to return
		List<CkanOrganization> toReturn = new ArrayList<CkanOrganization>();

		try{

			// get the list of all organizations
			List<CkanOrganization> organizations = client.getOrganizationList();

			// iterate over them
			for (CkanOrganization ckanOrganization : organizations) {

				// get the list of users in it (if you try ckanOrganization.getUsers() it returns null.. maybe a bug TODO)
				List<CkanUser> users = client.getOrganization(ckanOrganization.getName()).getUsers();

				// check if the current user is among them
				for (CkanUser ckanUser : users) {
					if(ckanUser.getName().equals(ckanUsername)){

						logger.debug("User " + ckanUsername + " is into " + ckanOrganization.getName());
						toReturn.add(ckanOrganization);
						break;

					}
				}
			}
		}catch(Exception e){
			logger.error("Unable to get user's organizations", e);
		}
		return toReturn;
	}

	@Override
	public List<CkanGroup> getGroupsByUser(String username) {
		logger.debug("Requested groups for user " + username);

		// checks
		checkNotNull(username);

		// in order to avoid errors, the username is always converted
		String ckanUsername = UtilMethods.fromUsernameToCKanUsername(username);

		// list to return
		List<CkanGroup> toReturn = new ArrayList<CkanGroup>();

		try{

			// get the list of all organizations
			List<CkanGroup> groups = client.getGroupList();

			// iterate over them
			for (CkanGroup ckanGroup : groups) {

				List<CkanUser> users = client.getGroup(ckanGroup.getName()).getUsers();

				// check if the current user is among them
				for (CkanUser ckanUser : users) {
					if(ckanUser.getName().equals(ckanUsername)){

						logger.debug("User " + ckanUsername + " is into " + ckanGroup.getName());
						toReturn.add(ckanGroup);
						break;

					}
				}
			}
		}catch(Exception e){
			logger.error("Unable to get user's groups", e);
		}
		return toReturn;
	}


	@Override
	public Map<String, List<RolesCkanGroupOrOrg>> getOrganizationsAndRolesByUser(
			String username, List<RolesCkanGroupOrOrg> rolesToMatch) {

		// checks
		checkNotNull(username);
		checkArgument(!rolesToMatch.isEmpty());

		logger.debug("Requested roles that the user " + username + " has into his organizations");
		logger.debug("Roles to check are " + rolesToMatch);

		Map<String, List<RolesCkanGroupOrOrg>> toReturn = new HashMap<String, List<RolesCkanGroupOrOrg>>();

		try{

			// in order to avoid errors, the username is always converted
			String ckanUsername = UtilMethods.fromUsernameToCKanUsername(username);

			// get the list of organizations in which the user is present
			List<CkanOrganization> organizationsByUser = getOrganizationsByUser(ckanUsername);

			// iterate over them
			for (CkanOrganization ckanOrganization : organizationsByUser) {

				// get the list of users in it (if you try ckanOrganization.getUsers() it returns null.. maybe a bug TODO)
				List<CkanUser> users = client.getOrganization(ckanOrganization.getName()).getUsers();

				// check if the current user is among them
				for (CkanUser ckanUser : users) {
					if(ckanUser.getName().equals(ckanUsername)){

						// the role (admin, editor, member)
						String capacity = ckanUser.getCapacity();

						// list of roles
						List<RolesCkanGroupOrOrg> rolesIntoOrg = new ArrayList<RolesCkanGroupOrOrg>();

						if(rolesToMatch.contains(RolesCkanGroupOrOrg.convertFromCapacity(capacity))){
							RolesCkanGroupOrOrg enumRole = RolesCkanGroupOrOrg.convertFromCapacity(capacity);
							rolesIntoOrg.add(enumRole);
							logger.debug("User " + ckanUsername + " has role " + enumRole +  
									" into organization with name " + ckanOrganization.getName());
						}

						// save it
						if(!rolesIntoOrg.isEmpty()){
							String orgName = ckanOrganization.getName();
							toReturn.put(orgName, rolesIntoOrg);
						}
						break;

					}
				}
			}

			logger.debug("Result is " + toReturn);

			return toReturn;
		}catch(Exception e){
			logger.error("Unable to analyze user's roles", e);
		}

		return null;
	}

	@Override
	public List<String> getOrganizationsIds(){

		List<String> toReturn = new ArrayList<String>();
		List<CkanOrganization> orgs = client.getOrganizationList();

		for (CkanOrganization ckanOrganization : orgs) {
			logger.debug("Retrieved org " + ckanOrganization.getName());
			toReturn.add(ckanOrganization.getId());
		}

		return toReturn;
	}

	@Override
	public List<String> getOrganizationsNames(){

		List<String> toReturn = new ArrayList<String>();
		List<CkanOrganization> orgs = client.getOrganizationList();

		for (CkanOrganization ckanOrganization : orgs) {
			logger.debug("Retrieved org " + ckanOrganization.getName());
			toReturn.add(ckanOrganization.getName());
		}

		return toReturn;
	}

	@Override
	public List<String> getOrganizationsNamesByUser(String username) {

		logger.debug("Requested organizations for user " + username);

		// checks
		checkNotNull(username);

		// in order to avoid errors, the username is always converted
		String ckanUsername = UtilMethods.fromUsernameToCKanUsername(username);

		List<CkanOrganization> orgs = getOrganizationsByUser(ckanUsername);

		List<String> orgsName = new ArrayList<String>();

		for (CkanOrganization ckanOrganization : orgs) {
			orgsName.add(ckanOrganization.getName());
			logger.debug("Organization name is " + ckanOrganization.getName());
		}

		return orgsName;

	}

	@Override
	public String findLicenseIdByLicenseTitle(String chosenLicense) {
		logger.debug("Requested license id");

		// checks
		checkNotNull(chosenLicense);

		//retrieve the list of available licenses
		List<CkanLicense> licenses = client.getLicenseList();

		for (CkanLicense ckanLicense : licenses) {
			if(ckanLicense.getTitle().equals(chosenLicense))
				return ckanLicense.getId();
		}

		return null;
	}

	@Override
	public List<String> getLicenseTitles() {

		logger.debug("Request for CKAN licenses");

		// get the url and the api key of the user
		List<String> result = new ArrayList<String>();

		//retrieve the list of available licenses
		List<CkanLicense> licenses = client.getLicenseList();

		for (CkanLicense ckanLicense : licenses) {

			result.add(ckanLicense.getTitle());
			logger.debug("License is " + ckanLicense.getTitle() + " and id " + ckanLicense.getId());

		}

		return result;
	}

	@Override
	public List<CkanLicense> getLicenses() {
		logger.debug("Request for CKAN licenses (original jackan objects are going to be retrieved)");

		//retrieve the list of available licenses
		return client.getLicenseList();
	}

	/**
	 * Set dataset private
	 * @param priv
	 * @param organizationId (NOTE: The ID, not the name!)
	 * @param datasetId (NOTE: The ID, not the name!)
	 * @param apiKey the user's api key
	 * @return true on success, false otherwise
	 */
	protected boolean setDatasetPrivate(boolean priv, String organizationId,
			String datasetId, String apiKey) {

		// checks
		checkNotNull(organizationId);
		checkNotNull(apiKey);
		checkNotNull(datasetId);
		checkArgument(!apiKey.isEmpty());
		checkArgument(!datasetId.isEmpty());
		checkArgument(!organizationId.isEmpty());

		String pathSetPrivate = "/api/3/action/bulk_update_private";
		String pathSetPublic = "/api/3/action/bulk_update_public";

		// Request parameters to be replaced
		String parameter = "{"
				+ "\"org_id\":\"ORGANIZATION_ID\","
				+ "\"datasets\":[\"DATASET_ID\"]"
				+ "}";

		// replace with right data
		parameter = parameter.replace("ORGANIZATION_ID", organizationId);
		parameter = parameter.replace("DATASET_ID", datasetId);

		if(priv){
			try(CloseableHttpClient httpClient = HttpClientBuilder.create().build();) {
				HttpPost request = new HttpPost(CKAN_CATALOGUE_URL + pathSetPrivate);
				request.addHeader("Authorization", apiKey);
				StringEntity params = new StringEntity(parameter);
				request.setEntity(params);
				HttpResponse response = httpClient.execute(request);
				logger.debug("[PRIVATE]Response code is " + response.getStatusLine().getStatusCode() + " and response message is " + response.getStatusLine().getReasonPhrase());

				if(response.getStatusLine().getStatusCode() == HttpStatus.SC_OK)
					return true;

			}catch (Exception ex) {
				logger.error("Error while trying to set private the dataset ", ex);
			}
		}else
		{
			try(CloseableHttpClient httpClient = HttpClientBuilder.create().build();) {
				HttpPost request = new HttpPost(CKAN_CATALOGUE_URL + pathSetPublic);
				StringEntity params = new StringEntity(parameter);
				request.addHeader("Authorization", apiKey);
				request.setEntity(params);
				HttpResponse response = httpClient.execute(request);
				logger.debug("[PUBLIC]Response code is " + response.getStatusLine().getStatusCode() + " and response message is " + response.getStatusLine().getReasonPhrase());

				if(response.getStatusLine().getStatusCode() == HttpStatus.SC_OK)
					return true;

			}catch (Exception ex) {
				logger.error("Error while trying to set public the dataset ", ex);
			}
		}

		return false;
	}

	@Override
	public String addResourceToDataset(ResourceBean resourceBean, String apiKey) {

		logger.debug("Request to add a resource described by this bean " + resourceBean);

		// checks
		checkNotNull(resourceBean);
		checkNotNull(apiKey);
		checkArgument(!apiKey.isEmpty());

		try{
			if(UtilMethods.resourceExists(resourceBean.getUrl())){

				// in order to avoid errors, the username is always converted
				String ckanUsername = UtilMethods.fromUsernameToCKanUsername(resourceBean.getOwner());

				CkanResource resource = new CkanResource(CKAN_CATALOGUE_URL, resourceBean.getDatasetId());
				resource.setName(resourceBean.getName());

				// escape description
				Source description = new Source(resourceBean.getDescription());
				Segment htmlSeg = new Segment(description, 0, description.length());
				Renderer htmlRend = new Renderer(htmlSeg);

				resource.setDescription(htmlRend.toString());
				resource.setUrl(resourceBean.getUrl());
				resource.setOwner(ckanUsername);

				// Checked client
				CheckedCkanClient client = new CheckedCkanClient(CKAN_CATALOGUE_URL, apiKey);
				CkanResource createdRes = client.createResource(resource);

				if(createdRes != null){

					logger.debug("Resource  " + createdRes.getName() + " is now available");

					return createdRes.getId();

				}
			}else
				logger.error("There is no resource at this url " + resourceBean.getUrl());
		}catch(Exception e){
			logger.error("Unable to create the resource described by the bean " + resourceBean, e);
		}
		return null;
	}

	@Override
	public boolean deleteResourceFromDataset(String resourceId, String apiKey) {

		logger.error("Request to delete a resource with id " + resourceId + " coming by user with key " + apiKey);

		// checks
		checkNotNull(apiKey);
		checkNotNull(resourceId);
		checkArgument(!apiKey.isEmpty());
		checkArgument(!resourceId.isEmpty());

		try{

			CheckedCkanClient client = new CheckedCkanClient(CKAN_CATALOGUE_URL, apiKey);
			client.deleteResource(resourceId);

			return true;

		}catch(Exception e){
			logger.error("Unable to delete resource whose id is " + resourceId, e);
		}

		return false;
	}

	@Override
	public String createCKanDataset(String apiKey,
			String title, String name, String organizationNameOrId, String author,
			String authorMail, String maintainer, String maintainerMail,
			long version, String description, String licenseId,
			List<String> tags, Map<String, String> customFields,
			List<ResourceBean> resources, boolean setPublic) {

		// delegate the private method
		return createCkanDatasetBody(apiKey,
				title, name, organizationNameOrId, author,
				authorMail, maintainer,maintainerMail,
				version, description, licenseId,
				tags, customFields, null,
				resources,  setPublic);
	}

	@Override
	public String createCKanDatasetMultipleCustomFields(String apiKey,
			String title, String name, String organizationNameOrId, String author,
			String authorMail, String maintainer, String maintainerMail,
			long version, String description, String licenseId,
			List<String> tags, Map<String, List<String>> customFieldsMultiple,
			List<ResourceBean> resources, boolean setPublic) {

		// delegate the private method
		return createCkanDatasetBody(apiKey,
				title, name, organizationNameOrId, author,
				authorMail, maintainer,maintainerMail,
				version, description, licenseId,
				tags, null, customFieldsMultiple,
				resources,  setPublic);
	}

	// the body of the actual dataset creation methods
	private String createCkanDatasetBody(String apiKey,
			String title, String name, String organizationNameOrId, String author,
			String authorMail, String maintainer, String maintainerMail,
			long version, String description, String licenseId,
			List<String> tags, Map<String, String> customFields,
			Map<String, List<String>> customFieldsMultipleValues,
			List<ResourceBean> resources, boolean setPublic){

		// checks (minimum)
		checkNotNull(apiKey);
		checkNotNull(organizationNameOrId);
		checkArgument(!apiKey.isEmpty());
		checkArgument(!organizationNameOrId.isEmpty());
		checkArgument(!((title == null && name == null) || (title.isEmpty() && name.isEmpty())), "Name and Title cannot be empty/null at the same time!");

		logger.debug("Request for dataset creation");

		CheckedCkanClient client = new CheckedCkanClient(CKAN_CATALOGUE_URL, apiKey);

		// get client from apiKey
		String ckanUsername = getUserFromApiKey(apiKey).getName();

		// create the base dataset and fill it
		CkanDataset dataset = new CkanDataset();

		// get the name from the title
		String nameToUse = name;
		if(nameToUse == null)
			nameToUse = UtilMethods.fromProductTitleToName(title);

		logger.debug("Name of the dataset is going to be " + nameToUse + ". Title is going to be " + title);
		dataset.setName(nameToUse);
		dataset.setTitle(title);

		CkanOrganization orgOwner = client.getOrganization(organizationNameOrId);
		dataset.setOwnerOrg(orgOwner.getId());
		dataset.setAuthor(author);
		dataset.setAuthorEmail(authorMail);
		dataset.setMaintainer(maintainer);
		dataset.setMaintainerEmail(maintainerMail);
		dataset.setVersion(String.valueOf(version));

		// description must be escaped
		if(description != null && !description.isEmpty()){
			Source descriptionEscaped = new Source(description);
			Segment htmlSeg = new Segment(descriptionEscaped, 0, descriptionEscaped.length());
			Renderer htmlRend = new Renderer(htmlSeg);
			dataset.setNotes(htmlRend.toString());

			logger.debug("Description (escaped is ) " + htmlRend.toString());
		}

		dataset.setLicenseId(licenseId);

		// set the tags, if any
		if(tags != null && !tags.isEmpty()){

			// convert to ckan tags
			List<CkanTag> ckanTags = new ArrayList<CkanTag>(tags.size());
			for (String stringTag : tags) {
				ckanTags.add(new CkanTag(stringTag));
			}

			dataset.setTags(ckanTags);
		}

		// set the custom fields, if any
		if(customFields != null && !customFields.isEmpty()){

			// iterate and create 
			Iterator<Entry<String, String>> iterator = customFields.entrySet().iterator();

			List<CkanPair> extras = new ArrayList<CkanPair>();

			while (iterator.hasNext()) {

				Map.Entry<String, String> entry = (Map.Entry<String, String>) iterator.next();
				extras.add(new CkanPair(entry.getKey(), entry.getValue()));

			}

			dataset.setExtras(extras);
		}else if(customFieldsMultipleValues != null && !customFieldsMultipleValues.isEmpty()){

			// iterate and create 
			Iterator<Entry<String, List<String>>> iterator = customFieldsMultipleValues.entrySet().iterator();

			List<CkanPair> extras = new ArrayList<CkanPair>();

			while (iterator.hasNext()) {

				Map.Entry<String, List<String>> entry = (Map.Entry<String, List<String>>) iterator.next();
				List<String> valuesForEntry = entry.getValue();
				for (String value : valuesForEntry) {
					extras.add(new CkanPair(entry.getKey(), value));
				}
			}

			// sort them
			Collections.sort(extras, new Comparator<CkanPair>() {
				@Override public int compare(CkanPair b1, CkanPair b2) {
					return b1.getKey().compareTo(b2.getKey());
				}

			});

			dataset.setExtras(extras);

		}

		// check if we need to add the resources
		if(resources != null && !resources.isEmpty()){

			logger.debug("We need to add resources to the dataset");

			try{

				List<CkanResource> resourcesCkan = new ArrayList<CkanResource>();


				for(ResourceBean resource: resources){

					CkanResource newResource = new CkanResource();
					newResource.setDescription(resource.getDescription());
					newResource.setId(resource.getId());
					newResource.setUrl(resource.getUrl());
					newResource.setName(resource.getName());
					newResource.setMimetype(resource.getMimeType());
					newResource.setOwner(ckanUsername);
					resourcesCkan.add(newResource);
				}

				// add to the dataset
				dataset.setResources(resourcesCkan);

			}catch(Exception e){
				logger.error("Unable to add those resources to the dataset", e);
			}

		}

		// try to create
		CkanDataset res = null;
		try{

			res = client.createDataset(dataset);

			if(res != null){

				logger.debug("Dataset with name " + res.getName() + " has been created. Setting visibility");

				// set visibility
				boolean visibilitySet = setDatasetPrivate(
						!setPublic, // swap to private 
						res.getOrganization().getId(), 
						res.getId(), 
						CKAN_TOKEN_SYS); // use sysadmin api key to be sure it will be set

				logger.info("Was visibility set to " + (setPublic ? "public" : "private") + "? " + visibilitySet);

				// set searchable to true if dataset visibility is private
				if(!setPublic){ // (swap to private)
					boolean searchableSet = setSearchableField(res.getId(), true);
					logger.info("Was searchable set to True? " + searchableSet);
				}

				return res.getId();
			}



		}catch(Exception e){

			// try to update
			logger.error("Error while creating the dataset.", e);

		}

		return null;

	}

	@Override
	public String getUrlFromDatasetIdOrName(String datasetIdOrName) {

		logger.debug("Request coming for getting dataset url (encrypted) of dataset with name/id " + datasetIdOrName);

		// checks
		checkNotNull(datasetIdOrName);
		checkArgument(!datasetIdOrName.isEmpty());
		String url = null;
		try{

			// get the dataset from name
			CheckedCkanClient client = new CheckedCkanClient(CKAN_CATALOGUE_URL, CKAN_TOKEN_SYS);
			CkanDataset dataset = client.getDataset(datasetIdOrName);
			String name = dataset.getName();

			if(dataset != null){

				if(getUriResolverUrl() != null)
					url = getUrlForProduct(CONTEXT, EntityContext.PRODUCT, name, false);

				if(url == null || url.isEmpty())
					url = getPortletUrl() + "?" + URLEncoder.encode("path=/dataset/"  + name, "UTF-8");

			}
		}catch(Exception e){
			logger.error("Error while retrieving dataset with id/name=" + datasetIdOrName, e);
		}
		return url;
	}

	@Override
	public String getUnencryptedUrlFromDatasetIdOrName(String datasetIdOrName) {
		logger.debug("Request coming for getting dataset url (not encrypted) of dataset with name/id " + datasetIdOrName);

		// checks
		checkNotNull(datasetIdOrName);
		checkArgument(!datasetIdOrName.isEmpty());
		String url = null;
		try{

			// get the dataset from name
			CheckedCkanClient client = new CheckedCkanClient(CKAN_CATALOGUE_URL, CKAN_TOKEN_SYS);
			CkanDataset dataset = client.getDataset(datasetIdOrName);
			String name = dataset.getName();

			if(dataset != null){

				if(getUriResolverUrl() != null)
					url = getUrlForProduct(CONTEXT, EntityContext.PRODUCT, name, true);

				if(url == null || url.isEmpty())
					url = getPortletUrl() + "?" + URLEncoder.encode("path=/dataset/"  + name, "UTF-8");

			}
		}catch(Exception e){
			logger.error("Error while retrieving dataset with id/name=" + datasetIdOrName, e);
		}
		return url;
	}

	/**
	 * Retrieve an url for the tuple scope, entity, entity name
	 * @param context
	 * @param entityContext
	 * @param entityName
	 * @return the url for the product
	 */
	@SuppressWarnings("unchecked")
	private String getUrlForProduct(String context, EntityContext entityContext, String entityName, boolean unencrypted){

		String toReturn = null;

		try (CloseableHttpClient httpClient = HttpClientBuilder.create().build()) {

			HttpPost httpPostRequest = new HttpPost(getUriResolverUrl());

			JSONObject requestEntity = new JSONObject();
			requestEntity.put("gcube_scope", context);
			requestEntity.put("entity_context", entityContext.toString());
			requestEntity.put("entity_name", entityName);
			requestEntity.put("clear_url", Boolean.toString(unencrypted));

			StringEntity params = new StringEntity(requestEntity.toJSONString(), ContentType.APPLICATION_JSON);
			httpPostRequest.setEntity(params);

			HttpResponse response = httpClient.execute(httpPostRequest);

			if(response.getStatusLine().getStatusCode() != 200)
				throw new Exception("There was an error while creating an url " + response.getStatusLine());

			toReturn = EntityUtils.toString(response.getEntity());
			logger.debug("Result is " + toReturn);

		}catch(Exception e){
			logger.error("Failed to get an url for this product", e);
		}

		return  toReturn;
	}

	@Override
	public boolean checkRoleIntoOrganization(String username, String organizationName,
			RolesCkanGroupOrOrg correspondentRoleToCheck) {

		logger.debug("Request for checking if " + username + " into organization " + organizationName + " has role " + correspondentRoleToCheck);

		// checks
		checkNotNull(username);
		checkNotNull(organizationName);
		checkNotNull(correspondentRoleToCheck);
		checkArgument(!username.isEmpty());
		checkArgument(!organizationName.isEmpty());

		// convert ckan username
		String ckanUsername = UtilMethods.fromUsernameToCKanUsername(username);

		// check if this role is already present in ckan for this user within the organization
		String organizationNameToCheck = organizationName.toLowerCase();

		try{
			boolean alreadyPresent = isRoleAlreadySet(ckanUsername, organizationNameToCheck, correspondentRoleToCheck, false);

			if(alreadyPresent)
				return true; // just return
			else{

				// we need to use the apis to make it
				String path = "/api/3/action/organization_member_create";

				// Request parameters to be replaced
				String parameter = "{"
						+ "\"id\":\"ORGANIZATION_ID_NAME\","
						+ "\"username\":\"USERNAME_ID_NAME\","
						+ "\"role\":\"ROLE\""
						+ "}";

				// replace those values
				parameter = parameter.replace("ORGANIZATION_ID_NAME", organizationNameToCheck);
				parameter = parameter.replace("USERNAME_ID_NAME", ckanUsername);
				parameter = parameter.replace("ROLE", RolesCkanGroupOrOrg.convertToCkanCapacity(correspondentRoleToCheck));

				logger.debug("API request for organization membership is going to be " + parameter);

				try(CloseableHttpClient httpClient = HttpClientBuilder.create().build();) {
					HttpPost request = new HttpPost(CKAN_CATALOGUE_URL + path);
					request.addHeader("Authorization", CKAN_TOKEN_SYS); // sys token
					StringEntity params = new StringEntity(parameter);
					request.setEntity(params);
					HttpResponse response = httpClient.execute(request);
					logger.debug("Response code is " + response.getStatusLine().getStatusCode() + " and response message is " + response.getStatusLine().getReasonPhrase());

					return (response.getStatusLine().getStatusCode() == HttpStatus.SC_OK);

				}catch (Exception ex) {
					logger.error("Error while trying to change the role for this user ", ex);
				}
			}
		}catch (Exception ex) {
			logger.error("Unable to check if this role was already set, please check your parameters! ", ex);
		}

		return false;
	}

	/**
	 * Check if the user has this role into the organization/group with groupOrOrganization name
	 * @param ckanUsername
	 * @param organizationName
	 * @param correspondentRoleToCheck
	 * @return true if he has the role, false otherwise
	 */
	protected boolean isRoleAlreadySet(String ckanUsername, String groupOrOrganization, RolesCkanGroupOrOrg correspondentRoleToCheck, boolean group) throws Exception{

		// get the users (if you try ckanOrganization.getUsers() it returns null.. maybe a bug TODO)
		List<CkanUser> users;

		if(group)
			users = client.getGroup(groupOrOrganization).getUsers();
		else
			users = client.getOrganization(groupOrOrganization).getUsers();

		for (CkanUser ckanUser : users) {
			if(ckanUser.getName().equals(ckanUsername))
				if(ckanUser.getCapacity().equals(RolesCkanGroupOrOrg.convertToCkanCapacity(correspondentRoleToCheck)))
					return true;
				else
					break;
		}

		return false;
	}

	@Override
	public boolean checkRoleIntoGroup(String username, String groupName, RolesCkanGroupOrOrg correspondentRoleToCheck) {
		logger.debug("Request for checking if " + username + " into group " + groupName + " has role " + correspondentRoleToCheck);

		// checks
		checkNotNull(username);
		checkNotNull(groupName);
		checkNotNull(correspondentRoleToCheck);
		checkArgument(!username.isEmpty());
		checkArgument(!groupName.isEmpty());

		// convert ckan username
		String ckanUsername = UtilMethods.fromUsernameToCKanUsername(username);

		// check if this role is already present in ckan for this user within the group
		String groupNameToCheck = UtilMethods.fromGroupTitleToName(groupName);

		try{
			boolean alreadyPresent = isRoleAlreadySet(ckanUsername, groupNameToCheck, correspondentRoleToCheck, true);

			if(alreadyPresent)
				return true; // just return
			else{

				// we need to use the apis to make it
				String path = "/api/3/action/group_member_create";

				// Request parameters to be replaced
				String parameter = "{"
						+ "\"id\":\"ORGANIZATION_ID_NAME\","
						+ "\"username\":\"USERNAME_ID_NAME\","
						+ "\"role\":\"ROLE\""
						+ "}";

				// replace those values
				parameter = parameter.replace("ORGANIZATION_ID_NAME", groupNameToCheck);
				parameter = parameter.replace("USERNAME_ID_NAME", ckanUsername);
				parameter = parameter.replace("ROLE", RolesCkanGroupOrOrg.convertToCkanCapacity(correspondentRoleToCheck));

				logger.debug("API request for organization membership is going to be " + parameter);

				try(CloseableHttpClient httpClient = HttpClientBuilder.create().build();) {
					HttpPost request = new HttpPost(CKAN_CATALOGUE_URL + path);
					request.addHeader("Authorization", CKAN_TOKEN_SYS); // sys token
					StringEntity params = new StringEntity(parameter);
					request.setEntity(params);
					HttpResponse response = httpClient.execute(request);
					logger.debug("Response code is " + response.getStatusLine().getStatusCode() + " and response message is " + response.getStatusLine().getReasonPhrase());

					return (response.getStatusLine().getStatusCode() == HttpStatus.SC_OK);

				}catch (Exception ex) {
					logger.error("Error while trying to change the role for this user ", ex);
				}
			}
		}catch (Exception ex) {
			logger.error("Unable to check if this role was already set, please check your parameters! ", ex);
		}

		return false;
	}

	@Override
	public boolean isSysAdmin(String username) {

		// checks
		checkNotNull(username);
		checkArgument(!username.isEmpty());

		// in order to avoid errors, the username is always converted
		String ckanUsername = UtilMethods.fromUsernameToCKanUsername(username);
		try{
			return client.getUser(ckanUsername).isSysadmin();
		}catch(Exception e){
			logger.error("Failed to check if the user " + username + " has role sysadmin", e);
		}

		return false;
	}

	@Override
	public boolean createDatasetRelationship(String datasetIdSubject,
			String datasetIdObject, DatasetRelationships relation, String relationComment, String apiKey) {

		// checks
		checkNotNull(datasetIdSubject);
		checkNotNull(datasetIdObject);
		checkNotNull(relation);
		checkNotNull(apiKey);
		checkArgument(!datasetIdSubject.isEmpty());
		checkArgument(!datasetIdObject.isEmpty());
		checkArgument(!apiKey.isEmpty());

		try(CloseableHttpClient httpClient = HttpClientBuilder.create().build();){

			// we need to use the apis to make it
			String path = "/api/3/action/package_relationship_create";

			// Request parameters to be replaced
			String parameter = "{"
					+ "\"subject\":\"SUBJECT\","
					+ "\"object\":\"OBJECT\","
					+ "\"type\":\"RELATIONSHIP\","
					+ "\"comment\" : \"COMMENT\""
					+ "}";

			// replace those values
			parameter = parameter.replace("SUBJECT", datasetIdSubject);
			parameter = parameter.replace("OBJECT", datasetIdObject);
			parameter = parameter.replace("RELATIONSHIP", relation.toString());
			if(relationComment != null && !relationComment.isEmpty())
				parameter = parameter.replace("COMMENT", relationComment);

			logger.debug("API request for relationship create is going to be " + parameter);

			HttpPost request = new HttpPost(CKAN_CATALOGUE_URL + path);
			request.addHeader("Authorization", apiKey);
			StringEntity params = new StringEntity(parameter);
			request.setEntity(params);
			HttpResponse response = httpClient.execute(request);
			logger.debug("Response code is " + response.getStatusLine().getStatusCode() + " and response message is " + response.getStatusLine().getReasonPhrase());

			return (response.getStatusLine().getStatusCode() == HttpStatus.SC_OK);

		}catch(Exception e){
			logger.error("Failed to create the relationship between dataset subject = "  + datasetIdSubject 
					+ " and " + " dataset subject " + datasetIdObject, e);
		}
		return false;
	}

	@Override
	public boolean deleteDatasetRelationship(String datasetIdSubject,
			String datasetIdObject, DatasetRelationships relation, String apiKey) {

		// checks
		checkNotNull(datasetIdSubject);
		checkNotNull(datasetIdObject);
		checkNotNull(relation);
		checkNotNull(apiKey);
		checkArgument(!datasetIdSubject.isEmpty());
		checkArgument(!datasetIdObject.isEmpty());
		checkArgument(!apiKey.isEmpty());

		try(CloseableHttpClient httpClient = HttpClientBuilder.create().build();){

			// we need to use the apis to make it
			String path = "/api/3/action/package_relationship_delete";

			// Request parameters to be replaced
			String parameter = "{"
					+ "\"subject\":\"SUBJECT\","
					+ "\"object\":\"OBJECT\","
					+ "\"type\":\"RELATIONSHIP\""
					+ "}";

			// replace those values
			parameter = parameter.replace("SUBJECT", datasetIdSubject);
			parameter = parameter.replace("OBJECT", datasetIdObject);
			parameter = parameter.replace("RELATIONSHIP", relation.toString());

			logger.debug("API request for delete relationship is going to be " + parameter);

			HttpPost request = new HttpPost(CKAN_CATALOGUE_URL + path);
			request.addHeader("Authorization", apiKey);
			StringEntity params = new StringEntity(parameter);
			request.setEntity(params);
			HttpResponse response = httpClient.execute(request);
			logger.debug("Response code is " + response.getStatusLine().getStatusCode() + " and response message is " + response.getStatusLine().getReasonPhrase());

			return (response.getStatusLine().getStatusCode() == HttpStatus.SC_OK);
		}catch(Exception e){
			logger.error("Failed to delete the relationship between dataset subject = "  + datasetIdSubject 
					+ " and " + " dataset subject " + datasetIdObject, e);
		}
		return false;
	}

	@SuppressWarnings("unchecked")
	@Override
	public List<CkanDatasetRelationship> getRelationshipDatasets(
			String datasetIdSubject, String datasetIdObject, String apiKey) {

		// checks
		checkNotNull(datasetIdSubject);
		checkNotNull(apiKey);
		checkArgument(!datasetIdSubject.isEmpty());
		checkArgument(!apiKey.isEmpty());

		try(CloseableHttpClient httpClient = HttpClientBuilder.create().build();){

			List<CkanDatasetRelationship> toReturn = new ArrayList<CkanDatasetRelationship>();

			// we need to use the apis to make it
			String path = "/api/3/action/package_relationships_list";

			String parameter;

			// Request parameters to be replaced
			if(datasetIdObject == null || datasetIdObject.isEmpty())
				parameter = "{"
						+ "\"id\":\"SUBJECT\""
						+ "}";
			else
				parameter = "{"
						+ "\"id\":\"SUBJECT\","
						+ "\"id2\":\"OBJECT\""
						+ "}";

			// replace those values
			parameter = parameter.replace("SUBJECT", datasetIdSubject);
			if(datasetIdObject != null && !datasetIdObject.isEmpty())
				parameter = parameter.replace("OBJECT", datasetIdObject);

			logger.debug("API request for getting relationship is going to be " + parameter);

			HttpPost request = new HttpPost(CKAN_CATALOGUE_URL + path);
			request.addHeader("Authorization", apiKey);
			StringEntity params = new StringEntity(parameter);
			request.setEntity(params);
			HttpResponse response = httpClient.execute(request);
			logger.debug("Response code is " + response.getStatusLine().getStatusCode() + " and response message is " + response.getStatusLine().getReasonPhrase());

			if(response.getStatusLine().getStatusCode() == HttpStatus.SC_OK){

				// parse the json and convert to java beans
				BufferedReader br = new BufferedReader(
						new InputStreamReader((response.getEntity().getContent())));

				String output;
				String res = "";
				while ((output = br.readLine()) != null) {
					res += output;
				}

				if(res == "")
					return toReturn;

				// parse the json object returned
				JSONParser parser = new JSONParser();
				JSONObject json = (JSONObject) parser.parse(res);
				JSONArray resultJson = (JSONArray) json.get("result");
				Iterator<JSONObject> it = resultJson.iterator();

				while (it.hasNext()) {
					JSONObject object = (JSONObject) it.next();
					try{
						CkanDatasetRelationship relation = new CkanDatasetRelationship(object);
						toReturn.add(relation);
					}catch(Exception e){
						logger.error("Error while building CkanRelationship bean from object " +  object, e);
					}
				}
			}

			return toReturn;

		}catch(Exception e){
			logger.error("Failed to retrieve the relationship between dataset subject = "  + datasetIdSubject 
					+ " and " + " dataset object " + datasetIdObject, e);
		}
		return null;
	}

	@Override
	public boolean existProductWithNameOrId(String nameOrId) {

		// checks
		checkNotNull(nameOrId);
		checkArgument(!nameOrId.isEmpty());

		try{
			CheckedCkanClient client = new CheckedCkanClient(CKAN_CATALOGUE_URL, CKAN_TOKEN_SYS);
			CkanDataset product = client.getDataset(nameOrId);
			return product != null;
		}catch(Exception e){
			logger.debug("A dataset with name " + nameOrId + " doesn't exist");
			return false;
		}
	}

	@Override
	public CkanGroup createGroup(String nameOrId, String title, String description) {

		// checks
		checkNotNull(nameOrId);
		checkArgument(!nameOrId.trim().isEmpty());

		// check if it exists
		CkanGroup toCreate = null;
		CheckedCkanClient client = new CheckedCkanClient(CKAN_CATALOGUE_URL, CKAN_TOKEN_SYS);

		logger.debug("Request for creating group with name " + nameOrId + " title " + title + " and description " + description);

		String name = UtilMethods.fromGroupTitleToName(nameOrId);
		if((toCreate = groupExists(name, client))!= null)
			return toCreate;
		else{
			try{

				CkanGroup group = new CkanGroup(name);
				group.setTitle(title);
				group.setDisplayName(title);
				group.setDescription(description);
				toCreate = client.createGroup(group);

			}catch(JackanException je){
				logger.error("Unable to create such a group", je);
			}
		}

		return toCreate;
	}

	/**
	 * Just check if the group exists
	 * @param nameOrId
	 * @param client
	 * @return
	 */
	private CkanGroup groupExists(String nameOrId, CheckedCkanClient client){

		CkanGroup toReturn = null;

		try{

			toReturn = client.getGroup(nameOrId);

		}catch(JackanException je){
			logger.error("This group doesn't exist");
		}

		return toReturn;
	}

	@SuppressWarnings("unchecked")
	@Override
	public boolean assignDatasetToGroup(String groupNameOrId, String datasetNameOrId, String apiKey) {

		// checks
		checkNotNull(groupNameOrId);
		checkArgument(!groupNameOrId.isEmpty());
		checkNotNull(datasetNameOrId);
		checkArgument(!datasetNameOrId.isEmpty());
		checkNotNull(apiKey);
		checkArgument(!apiKey.isEmpty());

		String groupNameToCheck = UtilMethods.fromGroupTitleToName(groupNameOrId);

		try(CloseableHttpClient httpClient = HttpClientBuilder.create().build();){

			CheckedCkanClient client = new CheckedCkanClient(CKAN_CATALOGUE_URL, CKAN_TOKEN_SYS);

			// check the group exists
			CkanGroup group = client.getGroup(groupNameToCheck);

			// we need to use the apis to make it
			String pathPackageShow = CKAN_CATALOGUE_URL + "/api/3/action/package_show?id=" + datasetNameOrId;
			HttpGet getRequest = new HttpGet(pathPackageShow);
			getRequest.addHeader("Authorization", CKAN_TOKEN_SYS);
			HttpResponse response = httpClient.execute(getRequest);
			List<String> fetchedGroups = new ArrayList<String>();

			logger.debug("Response is " + response.getStatusLine().getStatusCode() + " and message is " + response.getStatusLine().getReasonPhrase());

			// read the json dataset and fetch the groups and fetch the groups' names, if any
			if(response.getStatusLine().getStatusCode() == HttpStatus.SC_OK){

				// parse the json and convert to java beans
				String jsonAsString = EntityUtils.toString(response.getEntity());
				JSONParser parser = new JSONParser();
				JSONObject json = (JSONObject) parser.parse(jsonAsString);
				JSONObject resultJson =  (JSONObject) json.get("result");
				JSONArray groupsJson = (JSONArray)resultJson.get("groups");
				Iterator<JSONObject> it =  groupsJson.iterator();

				while (it.hasNext()) {
					JSONObject object = (JSONObject) it.next();
					try{
						if(object.containsKey("name"))
							fetchedGroups.add((String)object.get("name"));
					}catch(Exception e){
						logger.error("Error while building CkanRelationship bean from object " +  object, e);
					}
				}

				// add the new one
				if(!fetchedGroups.contains(group.getName()))
					fetchedGroups.add(group.getName());

				logger.debug("Groups to be added are " + fetchedGroups);

				// now we patch the dataset with the new group
				String pathUpdatePatch = CKAN_CATALOGUE_URL + "/api/3/action/package_patch";
				String parameterPostPatch = "{\"id\":\"PACKAGE_ID\", \"groups\":[GROUPS]}";
				parameterPostPatch = parameterPostPatch.replace("PACKAGE_ID", datasetNameOrId);
				String singleGroup = "{\"name\":\"GROUP_ID\"}";

				// evaluate parameterPostPatch
				String replaceGROUPS = "";
				for (int i = 0;  i < fetchedGroups.size(); i++) {
					replaceGROUPS += singleGroup.replace("GROUP_ID", fetchedGroups.get(i));
					if(i != fetchedGroups.size() - 1)
						replaceGROUPS += ",";
				}

				// replace this into parameterPostPatch
				parameterPostPatch = parameterPostPatch.replace("GROUPS", replaceGROUPS);

				logger.debug("Request for patch is going to be " + parameterPostPatch);

				HttpPost request = new HttpPost(pathUpdatePatch);
				request.addHeader("Authorization", CKAN_TOKEN_SYS);
				StringEntity params = new StringEntity(parameterPostPatch);
				request.setEntity(params);
				HttpResponse responsePatch = httpClient.execute(request);
				logger.debug("Response code is " + responsePatch.getStatusLine().getStatusCode() + " and response message is " + responsePatch.getStatusLine().getReasonPhrase());

				if(responsePatch.getStatusLine().getStatusCode() == HttpStatus.SC_OK){
					logger.info("Dataset Added to the group!!");
					return true;
				}

			}

		}catch(Exception e){
			logger.error("Unable to make this association", e);
		}

		return false;
	}

	@SuppressWarnings("unchecked")
	@Override
	public boolean removeDatasetFromGroup(String groupNameOrId,
			String datasetNameOrId, String apiKey) {

		// checks
		checkNotNull(groupNameOrId);
		checkNotNull(datasetNameOrId);
		checkNotNull(apiKey);

		try (CloseableHttpClient httpClient = HttpClientBuilder.create().build()) {
			String apiRequestUrl = CKAN_CATALOGUE_URL + "/api/3/action/package_patch";
			HttpPost httpPostRequest = new HttpPost(apiRequestUrl);
			httpPostRequest.setHeader("Authorization", CKAN_TOKEN_SYS);

			// Request parameters to be replaced			
			JSONObject jsonRequest = new JSONObject();
			JSONArray groupsAsJson = new JSONArray();

			CheckedCkanClient client = new CheckedCkanClient(CKAN_CATALOGUE_URL, CKAN_TOKEN_SYS);
			CkanDataset dataset = client.getDataset(datasetNameOrId);
			List<CkanGroup> currentGroups = dataset.getGroups();

			// build the json array for the "groups" field.. each object looks like 
			//			{
			//
			//			    "display_name": "Fishery activity",
			//			    "description": "",
			//			    "image_display_url": "",
			//			    "title": "Fishery activity",
			//			    "id": "20b8d23c-d2dd-4613-9e1e-856311862e87",
			//			    "name": "fishery-activity"
			//
			//			}

			// get the id of the group to be removed
			String groupId = client.getGroup(groupNameOrId).getId();

			for(CkanGroup ckanGroup : currentGroups){
				if(!ckanGroup.getId().equals(groupId)){
					JSONObject obj = new JSONObject();
					obj.put("display_name", ckanGroup.getDisplayName());
					obj.put("description", ckanGroup.getDescription());
					obj.put("image_display_url", ckanGroup.getImageDisplayUrl());
					obj.put("title", ckanGroup.getTitle());
					obj.put("id", ckanGroup.getId());
					obj.put("name", ckanGroup.getName());
					groupsAsJson.add(obj);
				}
			}

			// perform the request
			jsonRequest.put("id", datasetNameOrId);
			jsonRequest.put("groups", groupsAsJson);

			logger.debug("Request param is going to be " + jsonRequest);

			StringEntity params = new StringEntity(jsonRequest.toJSONString(), ContentType.APPLICATION_JSON);
			httpPostRequest.setEntity(params);

			HttpResponse response = httpClient.execute(httpPostRequest);

			if (response.getStatusLine().getStatusCode() < 200 || response.getStatusLine().getStatusCode() >= 300) {
				throw new RuntimeException("failed to patch the product. response status line from "
						+ apiRequestUrl + " was: " + response.getStatusLine());
			}

			return true;

		}catch(Exception e){
			logger.error("Failed to remove the group " + groupNameOrId + " from product " + datasetNameOrId, e);
		}

		return false;
	}

	@Override
	public Map<String, List<String>> getRolesAndUsersOrganization(String organizationName) {

		// checks
		checkNotNull(organizationName);
		checkArgument(!organizationName.isEmpty());

		Map<String, List<String>> capacityAndUsers = new HashMap<String, List<String>>();
		CkanOrganization org = client.getOrganization(organizationName);
		List<CkanUser> users = org.getUsers();
		for (CkanUser ckanUser : users) {

			logger.debug(ckanUser.getName());

			List<String> listUsers;
			if(capacityAndUsers.containsKey(ckanUser.getCapacity())){
				listUsers = capacityAndUsers.get(ckanUser.getCapacity());
			}else
				listUsers = new ArrayList<String>();

			listUsers.add(ckanUser.getName());
			capacityAndUsers.put(ckanUser.getCapacity(), listUsers);

		}

		return capacityAndUsers;
	}

	@Override
	public Map<RolesCkanGroupOrOrg, List<String>> getRolesAndUsersGroup(String groupName) {

		// checks
		checkNotNull(groupName);
		checkArgument(!groupName.isEmpty());
		Map<RolesCkanGroupOrOrg, List<String>> capacityAndUsers = null;
		String groupNameToCheck = UtilMethods.fromGroupTitleToName(groupName);

		CkanGroup group = client.getGroup(groupNameToCheck);

		if(group != null){
			capacityAndUsers = new HashMap<RolesCkanGroupOrOrg, List<String>>();
			List<CkanUser> users = group.getUsers();
			for (CkanUser ckanUser : users) {
				List<String> listUsers;
				if(capacityAndUsers.containsKey(RolesCkanGroupOrOrg.convertFromCapacity(ckanUser.getCapacity()))){
					listUsers = capacityAndUsers.get(RolesCkanGroupOrOrg.convertFromCapacity(ckanUser.getCapacity()));
				}else
					listUsers = new ArrayList<String>();

				listUsers.add(ckanUser.getName());
				capacityAndUsers.put(RolesCkanGroupOrOrg.convertFromCapacity(ckanUser.getCapacity()), listUsers);

			}

			logger.info("Returning " + capacityAndUsers);
		}
		return capacityAndUsers;
	}

	@Override
	public String getRoleOfUserInOrganization(String username, String orgName, String apiKey) {

		String toReturn = null;

		String usernameCkan = UtilMethods.fromUsernameToCKanUsername(username);

		try{

			CheckedCkanClient client = new CheckedCkanClient(CKAN_CATALOGUE_URL, apiKey);
			List<CkanUser> users = client.getOrganization(orgName).getUsers();
			for (CkanUser ckanUser : users) {
				if(ckanUser.getName().equals(usernameCkan)){
					toReturn = ckanUser.getCapacity();
					break;
				}
			}

		}catch(Exception e){
			logger.error("Unable to retrieve the role the user has into this organization", e);
		}

		return toReturn;
	}

	@Override
	public String getRoleOfUserInGroup(String username, String groupName, String apiKey) {

		String toReturn = null;

		String usernameCkan = UtilMethods.fromUsernameToCKanUsername(username);

		try{

			CheckedCkanClient client = new CheckedCkanClient(CKAN_CATALOGUE_URL, apiKey);
			List<CkanUser> users = client.getGroup(groupName).getUsers();
			for (CkanUser ckanUser : users) {
				if(ckanUser.getName().equals(usernameCkan)){
					toReturn = ckanUser.getCapacity();
					break;
				}
			}

		}catch(Exception e){
			logger.error("Unable to retrieve the role the user has into this group", e);
		}

		return toReturn;
	}

	@Override
	public boolean deleteProduct(String datasetId, String apiKey, boolean purge) {

		// checks
		checkNotNull(datasetId);
		checkArgument(!datasetId.isEmpty());
		checkNotNull(apiKey);
		checkArgument(!apiKey.isEmpty());

		logger.debug("Incoming request of deleting dataset with id " + datasetId);

		try{

			CheckedCkanClient client = new CheckedCkanClient(CKAN_CATALOGUE_URL, apiKey);
			client.deleteDataset(datasetId);
			logger.info("Dataset with id " + datasetId + " deleted!");

			if(purge){

				logger.debug("Purging also ....");
				try(CloseableHttpClient httpClient = HttpClientBuilder.create().build();){
					String path = CKAN_CATALOGUE_URL + "/api/3/action/dataset_purge";
					HttpPost request = new HttpPost(path);
					request.addHeader("Authorization", CKAN_TOKEN_SYS); // this must be a sys_admin key
					String entityBody = "{\"id\": \"" + datasetId + "\"}";
					StringEntity params = new StringEntity(entityBody);
					request.setEntity(params);
					HttpResponse response = httpClient.execute(request);

					logger.debug("Response code is " + response.getStatusLine().getStatusCode() + " and response message is " + response.getStatusLine().getReasonPhrase());

					if(response.getStatusLine().getStatusCode() == 200){

						logger.info("Dataset with id " + datasetId + " delete and purged!");
						return true;
					}

					return false;
				}

			}
			return true;

		}catch(Exception e){
			logger.error("Unable to delete such dataset ", e);
		}
		return false;
	}

	@Override
	public CkanDataset getDataset(String datasetId, String apiKey) {

		logger.info("Request ckan dataset with id " + datasetId);

		// checks
		checkNotNull(datasetId);
		checkArgument(!datasetId.isEmpty());
		checkNotNull(apiKey);
		checkArgument(!apiKey.isEmpty());

		try{
			CheckedCkanClient client = new CheckedCkanClient(CKAN_CATALOGUE_URL, apiKey);
			return client.getDataset(datasetId);
		}catch(Exception e){
			logger.error("Unable to retrieve such dataset, returning null ...", e);
		}

		return null;
	}

	@Override
	public boolean setSearchableField(String datasetId, boolean searchable) {

		// checks
		checkNotNull(datasetId);
		checkArgument(!datasetId.isEmpty());
		String searchableAsString = searchable ? "True" : "False";

		// Patch package path
		String patchPackage = CKAN_CATALOGUE_URL + "/api/3/action/package_patch";

		// Request parameters to be replaced
		String parameter = "{"
				+ "\"id\":\"DATASET_ID\","
				+ "\"searchable\":\"SEARCHABLE\""
				+ "}";

		// replace with right data
		parameter = parameter.replace("DATASET_ID", datasetId);
		parameter = parameter.replace("SEARCHABLE", searchableAsString);

		try(CloseableHttpClient httpClient = HttpClientBuilder.create().build();) {
			HttpPost request = new HttpPost(patchPackage);
			request.addHeader("Authorization", CKAN_TOKEN_SYS);
			StringEntity params = new StringEntity(parameter);
			request.setEntity(params);
			HttpResponse response = httpClient.execute(request);
			logger.debug("Response code is " + response.getStatusLine().getStatusCode() + " and response message is " + response.getStatusLine().getReasonPhrase());

			if(response.getStatusLine().getStatusCode() == HttpStatus.SC_OK)
				return true;

		}catch (Exception ex) {
			logger.error("Error while trying to set searchable the dataset ", ex);
		}

		return false;
	}

	@Override
	public List<CkanGroup> getGroups() {

		try{
			CkanClient client = new CkanClient(CKAN_CATALOGUE_URL);
			return client.getGroupList();
		}catch(Exception e){
			logger.error("Failed to retrieve the list groups", e);
			return null;
		}

	}

	@Override
	public CkanResource uploadResourceFile(File file, String packageId,
			String token, String name, String description) {

		// checks
		checkNotNull(file);
		checkNotNull(packageId);
		checkNotNull(token);
		checkNotNull(name);

		String returnedId = null;

		try (CloseableHttpClient httpClient = HttpClientBuilder.create().build()) {
			String apiRequestUrl = CKAN_CATALOGUE_URL + "/api/3/action/resource_create";
			HttpPost httpPostRequest = new HttpPost(apiRequestUrl);
			httpPostRequest.setHeader("Authorization", token);
			HttpEntity mpEntity = 
					MultipartEntityBuilder.create()
					.addTextBody("package_id", packageId, ContentType.TEXT_PLAIN)
					.addTextBody("url", "upload", ContentType.TEXT_PLAIN)
					.addTextBody("description", description == null ? "" : description, ContentType.TEXT_PLAIN)
					.addTextBody("name", name, ContentType.TEXT_PLAIN)
					.addBinaryBody("upload", file, 
							ContentType.create(
									"application/octet-stream",
									Charset.forName("UTF-8")), 
									name)
									.build();

			httpPostRequest.setEntity(mpEntity);
			HttpResponse response = httpClient.execute(httpPostRequest);

			if (response.getStatusLine().getStatusCode() < 200 || response.getStatusLine().getStatusCode() >= 300) {
				throw new RuntimeException("failed to add the file to CKAN storage. response status line from "
						+ apiRequestUrl + " was: " + response.getStatusLine());
			}

			logger.info("Returned message is " + response.getStatusLine());

			String json = EntityUtils.toString(response.getEntity());
			Object obj = JSONValue.parse(json);
			JSONObject finalResult=(JSONObject)obj;
			JSONObject result = (JSONObject)finalResult.get("result");
			logger.debug("Returned json is " + result.get("id"));
			returnedId = (String) result.get("id");
			return new CheckedCkanClient(CKAN_CATALOGUE_URL, token).getResource(returnedId);
		} catch (Exception e) {
			logger.error("Error while uploading file");
			return null;
		}
	}

	@SuppressWarnings("unchecked")
	@Override
	public boolean patchResource(String resourceId, String url,
			String name, String description, String urlType, String apiKey) {

		// checks
		checkNotNull(resourceId);
		checkNotNull(apiKey);

		logger.debug("Going to change resource with id " + resourceId +"."
				+ " Request comes from user with key " + apiKey.substring(0, 5) + "****************");


		try (CloseableHttpClient httpClient = HttpClientBuilder.create().build()) {
			String apiRequestUrl = CKAN_CATALOGUE_URL + "/api/3/action/resource_patch";
			HttpPost httpPostRequest = new HttpPost(apiRequestUrl);
			httpPostRequest.setHeader("Authorization", apiKey);

			// Request parameters to be replaced			
			JSONObject jsonRequest = new JSONObject();
			Map<String, String> requestMap = new HashMap<String, String>();

			requestMap.put("id", resourceId);
			if(url != null && !url.isEmpty())
				requestMap.put("url", url);
			if(description != null)
				requestMap.put("description", description);
			if(name != null && !name.isEmpty())
				requestMap.put("name", name);
			if(urlType != null)
				requestMap.put("url_type", urlType);

			jsonRequest.putAll(requestMap);
			StringEntity params = new StringEntity(jsonRequest.toJSONString());
			httpPostRequest.setEntity(params);

			HttpResponse response = httpClient.execute(httpPostRequest);

			if (response.getStatusLine().getStatusCode() < 200 || response.getStatusLine().getStatusCode() >= 300) {
				throw new RuntimeException("failed to patch the resource. response status line from "
						+ apiRequestUrl + " was: " + response.getStatusLine());
			}

			return true;

		}catch(Exception e){
			logger.error("Failed to update the resource ", e);
		}

		return false;
	}

	@SuppressWarnings("unchecked")
	@Override
	public boolean patchProductCustomFields(String productId, String apiKey,
			Map<String, List<String>> customFieldsToChange) {

		// checks
		checkNotNull(productId);
		checkNotNull(apiKey);

		if(customFieldsToChange == null || customFieldsToChange.isEmpty()) // TODO.. remove all custom fields maybe?!
			return true;

		logger.info("Going to change product with id " + productId +"."
				+ " Request comes from user with key " + apiKey.substring(0, 5) + "****************");

		logger.info("The new values are " + customFieldsToChange);

		// Get already available custom fields
		Map<String, List<String>> fromCKANCustomFields = new HashMap<String, List<String>>();
		CheckedCkanClient client = new CheckedCkanClient(CKAN_CATALOGUE_URL, apiKey);
		List<CkanPair> extras = client.getDataset(productId).getExtras();

		if(extras == null)
			extras = new ArrayList<CkanPair>();

		// fill the above map with these values
		for (CkanPair ckanPair : extras) {
			List<String> forThisValue = null;
			String key = ckanPair.getKey();
			if(fromCKANCustomFields.containsKey(key))
				forThisValue = fromCKANCustomFields.get(key);
			else
				forThisValue = 	new ArrayList<String>();
			forThisValue.add(ckanPair.getValue());
			fromCKANCustomFields.put(key, forThisValue);
		}

		logger.info("The generated map from jackan looks like " + fromCKANCustomFields + ". Going to merge them");

		// merge them with the new values
		Iterator<Entry<String, List<String>>> iteratorUserMap = customFieldsToChange.entrySet().iterator();
		while (iteratorUserMap.hasNext()) {
			Map.Entry<java.lang.String, java.util.List<java.lang.String>> entry = (Map.Entry<java.lang.String, java.util.List<java.lang.String>>) iteratorUserMap
					.next();

			String key = entry.getKey();
			List<String> newValues = entry.getValue();

			// get the unique set of values
			Set<String> uniqueValues = new HashSet<String>();

			if(fromCKANCustomFields.containsKey(key))
				uniqueValues.addAll(fromCKANCustomFields.get(key));

			uniqueValues.addAll(newValues);
			fromCKANCustomFields.put(key, new ArrayList<String>(uniqueValues));
		}

		logger.info("After merging it is " + fromCKANCustomFields);

		try (CloseableHttpClient httpClient = HttpClientBuilder.create().build()) {
			String apiRequestUrl = CKAN_CATALOGUE_URL + "/api/3/action/package_patch";
			HttpPost httpPostRequest = new HttpPost(apiRequestUrl);
			httpPostRequest.setHeader("Authorization", apiKey);

			// Request parameters to be replaced			
			JSONObject jsonRequest = new JSONObject();

			// build the json array for the "extras" field.. each object looks like {"key": ..., "value": ...}
			JSONArray extrasObject = new JSONArray();

			Iterator<Entry<String, List<String>>> iteratorNewFields = fromCKANCustomFields.entrySet().iterator();
			while (iteratorNewFields.hasNext()) {
				Map.Entry<java.lang.String, java.util.List<java.lang.String>> entry = (Map.Entry<java.lang.String, java.util.List<java.lang.String>>) iteratorNewFields
						.next();

				String key = entry.getKey();
				List<String> values = entry.getValue();

				for (String value : values) {
					JSONObject obj = new JSONObject();
					obj.put("value", value);
					obj.put("key", key);
					extrasObject.add(obj);
				}
			}

			// perform the request
			jsonRequest.put("id", productId);
			jsonRequest.put("extras", extrasObject);

			logger.debug("Request param is going to be " + jsonRequest);

			StringEntity params = new StringEntity(jsonRequest.toJSONString(), ContentType.APPLICATION_JSON);
			httpPostRequest.setEntity(params);

			HttpResponse response = httpClient.execute(httpPostRequest);

			if (response.getStatusLine().getStatusCode() < 200 || response.getStatusLine().getStatusCode() >= 300) {
				throw new RuntimeException("failed to patch the product. response status line from "
						+ apiRequestUrl + " was: " + response.getStatusLine());
			}

			return true;

		}catch(Exception e){
			logger.error("Failed to patch the product ", e);
		}

		return false;
	}

	@SuppressWarnings("unchecked")
	@Override
	public boolean removeCustomField(String productId, String key,
			String value, String apiKey) {

		// checks
		checkNotNull(productId);
		checkNotNull(apiKey);
		checkNotNull(value);
		checkNotNull(key);

		logger.info("Going to change product with id " + productId +"."
				+ " Request comes from user with key " + apiKey.substring(0, 5) + "****************");

		logger.info("The couple key/value to remove from custom fields is  [" + key + "," + value +"]");

		CheckedCkanClient client = new CheckedCkanClient(CKAN_CATALOGUE_URL, apiKey);
		List<CkanPair> extras = client.getDataset(productId).getExtras();

		Iterator<CkanPair> iterator = extras.iterator();
		while (iterator.hasNext()) {
			CkanPair ckanPair = (CkanPair) iterator.next();

			if(ckanPair.getKey().equals(key) && ckanPair.getValue().equals(value)){
				logger.info("Removed from the ckan pairs list");
				iterator.remove();
				break;
			}

		}

		try (CloseableHttpClient httpClient = HttpClientBuilder.create().build()) {
			String apiRequestUrl = CKAN_CATALOGUE_URL + "/api/3/action/package_patch";
			HttpPost httpPostRequest = new HttpPost(apiRequestUrl);
			httpPostRequest.setHeader("Authorization", apiKey);

			// Request parameters to be replaced			
			JSONObject jsonRequest = new JSONObject();

			// build the json array for the "extras" field.. each object looks like {"key": ..., "value": ...}
			JSONArray extrasObject = new JSONArray();

			for(CkanPair extra: extras){
				JSONObject obj = new JSONObject();
				obj.put("value", extra.getValue());
				obj.put("key", extra.getKey());
				extrasObject.add(obj);
			}

			// perform the request
			jsonRequest.put("id", productId);
			jsonRequest.put("extras", extrasObject);

			logger.debug("Request param is going to be " + jsonRequest);

			StringEntity params = new StringEntity(jsonRequest.toJSONString(), ContentType.APPLICATION_JSON);
			httpPostRequest.setEntity(params);

			HttpResponse response = httpClient.execute(httpPostRequest);

			if (response.getStatusLine().getStatusCode() < 200 || response.getStatusLine().getStatusCode() >= 300) {
				throw new RuntimeException("failed to patch the product. response status line from "
						+ apiRequestUrl + " was: " + response.getStatusLine());
			}

			return true;

		}catch(Exception e){
			logger.error("Failed to remove the custom field for this product ", e);
		}

		return false;

	}

	@SuppressWarnings("unchecked")
	@Override
	public boolean removeTag(String productId, String apiKey, String tagToRemove) {

		// checks
		checkNotNull(productId);
		checkNotNull(apiKey);
		checkNotNull(tagToRemove);

		try (CloseableHttpClient httpClient = HttpClientBuilder.create().build()) {
			String apiRequestUrl = CKAN_CATALOGUE_URL + "/api/3/action/package_patch";
			HttpPost httpPostRequest = new HttpPost(apiRequestUrl);
			httpPostRequest.setHeader("Authorization", apiKey);

			// Request parameters to be replaced			
			JSONObject jsonRequest = new JSONObject();
			JSONArray tagsAsJson = new JSONArray();

			CheckedCkanClient client = new CheckedCkanClient(CKAN_CATALOGUE_URL, apiKey);
			CkanDataset dataset = client.getDataset(productId);
			List<CkanTag> currentTags = dataset.getTags();

			// build the json array for the "tags" field.. each object looks like 
			//			{
			//
			//			    "vocabulary_id": null,
			//			    "state": "active",
			//			    "display_name": "....",
			//			    "id": "....",
			//			    "name": "..."
			//
			//			}

			for(CkanTag ckanTag : currentTags){
				if(!ckanTag.getName().equals(tagToRemove)){
					JSONObject obj = new JSONObject();
					obj.put("vocabulary_id", ckanTag.getVocabularyId());
					obj.put("state", ckanTag.getState().toString());
					obj.put("display_name", ckanTag.getDisplayName());
					obj.put("id", ckanTag.getId());
					obj.put("name", ckanTag.getName());
					tagsAsJson.add(obj);
				}
			}

			// perform the request
			jsonRequest.put("id", productId);
			jsonRequest.put("tags", tagsAsJson);

			logger.debug("Request param is going to be " + jsonRequest);

			StringEntity params = new StringEntity(jsonRequest.toJSONString(), ContentType.APPLICATION_JSON);
			httpPostRequest.setEntity(params);

			HttpResponse response = httpClient.execute(httpPostRequest);

			if (response.getStatusLine().getStatusCode() < 200 || response.getStatusLine().getStatusCode() >= 300) {
				throw new RuntimeException("failed to patch the product. response status line from "
						+ apiRequestUrl + " was: " + response.getStatusLine());
			}

			return true;

		}catch(Exception e){
			logger.error("Failed to remove the tag " + tagToRemove, e);
		}

		return false;
	}

	@Override
	public boolean addTag(String productId, String apiKey, String tagToAdd) {

		// checks
		checkNotNull(productId);
		checkNotNull(apiKey);
		checkNotNull(tagToAdd);

		try{

			CheckedCkanClient client = new CheckedCkanClient(CKAN_CATALOGUE_URL, apiKey);
			CkanDataset dataset = client.getDataset(productId);
			List<CkanTag> currentTags = dataset.getTags();

			Iterator<CkanTag> tagsIterator = currentTags.iterator();

			boolean added = true;

			// check if it is already there ...
			while (tagsIterator.hasNext()) {
				CkanTag ckanTag = (CkanTag) tagsIterator.next();
				if(ckanTag.getName().equals(tagToAdd)){
					added = false;
					break;
				}
			}


			if(added){
				currentTags.add(new CkanTag(tagToAdd));
				dataset.setTags(currentTags);
				client.patchUpdateDataset(dataset);
			}

			return true;
		}catch(Exception e){
			logger.error("Failed to add the tag " + tagToAdd, e);
		}

		return false;
	}

	@Override
	public List<CkanGroup> getParentGroups(String groupName, String apiKey) {
		// checks
		checkNotNull(groupName);
		checkNotNull(apiKey);

		try{
			CheckedCkanClient client = new CheckedCkanClient(CKAN_CATALOGUE_URL, apiKey);
			return client.getGroup(groupName).getGroups();
		}catch(Exception e){
			logger.error("Something went wrong, returning null", e);
		}

		return null;
	}

	@SuppressWarnings("unchecked")
	@Override
	public boolean setGroupParent(String parentName, String groupName) {

		// checks
		checkNotNull(parentName);
		checkNotNull(groupName);

		try (CloseableHttpClient httpClient = HttpClientBuilder.create().build()) {
			String apiRequestUrl = CKAN_CATALOGUE_URL + "/api/3/action/group_patch";
			HttpPost httpPostRequest = new HttpPost(apiRequestUrl);
			httpPostRequest.setHeader("Authorization", CKAN_TOKEN_SYS);

			CheckedCkanClient client = new CheckedCkanClient(CKAN_CATALOGUE_URL, CKAN_TOKEN_SYS);
			List<CkanGroup> alreadyAvailableParents = client.getGroup(groupName).getGroups();

			Set<String> parentsNames = new HashSet<String>();
			parentsNames.add(parentName);

			for (CkanGroup alreadyAvailableParent : alreadyAvailableParents) {
				parentsNames.add(alreadyAvailableParent.getName());
			}

			logger.info("Setting as parents of group " + groupName + " :" + parentsNames);

			// Request parameters to be replaced			
			JSONObject jsonRequest = new JSONObject();
			JSONArray parentsAsJson = new JSONArray();

			for(String parent : parentsNames){
				JSONObject obj = new JSONObject();
				obj.put("name", parent);
				parentsAsJson.add(obj);
			}

			// perform the request
			jsonRequest.put("id", groupName);
			jsonRequest.put("groups", parentsAsJson);

			logger.debug("Request param is going to be " + jsonRequest);

			StringEntity params = new StringEntity(jsonRequest.toJSONString(), ContentType.APPLICATION_JSON);
			httpPostRequest.setEntity(params);

			HttpResponse response = httpClient.execute(httpPostRequest);

			if (response.getStatusLine().getStatusCode() < 200 || response.getStatusLine().getStatusCode() >= 300) {
				throw new RuntimeException("failed to patch the group. response status line from "
						+ apiRequestUrl + " was: " + response.getStatusLine());
			}

			return true;

		}catch(Exception e){
			logger.error("Failed to set parents groups", e);
		}

		return false;
	}

	@Override
	public boolean isDatasetInGroup(String groupName, String datasetId) {

		// checks
		checkNotNull(datasetId);
		checkNotNull(groupName);

		CheckedCkanClient client = new CheckedCkanClient(CKAN_CATALOGUE_URL, CKAN_TOKEN_SYS);
		List<CkanGroup> groups = client.getDataset(datasetId).getGroups();

		for (CkanGroup ckanGroup : groups) {
			if(ckanGroup.getName().equals(groupName))
				return true;
		}

		return false;
	}

	@Override
	public List<CkanDataset> getProductsInGroup(String groupName) {

		checkNotNull(groupName);

		List<CkanDataset> toReturn = null;

		try (CloseableHttpClient httpClient = HttpClientBuilder.create().build()) {

			CheckedCkanClient client = new CheckedCkanClient(CKAN_CATALOGUE_URL, CKAN_TOKEN_SYS);
			String apiRequestUrl = CKAN_CATALOGUE_URL + "/api/3/action/group_show" + "?id=" + groupName + "&include_datasets=true";
			HttpGet httpGetRequest = new HttpGet(apiRequestUrl);
			httpGetRequest.setHeader("Authorization", CKAN_TOKEN_SYS);
			HttpResponse response = httpClient.execute(httpGetRequest);
			String jsonString = EntityUtils.toString(response.getEntity());
			JSONParser parser = new JSONParser();
			JSONObject parsedJson = (JSONObject)parser.parse(jsonString);
			logger.debug("JSONObject looks like " + parsedJson);

			// get "packages" array
			toReturn = new ArrayList<CkanDataset>();
			JSONObject result = (JSONObject)parsedJson.get("result");
			JSONArray packages = (JSONArray)result.get("packages");

			logger.debug("Packages looks like " + packages);

			for (int i = 0, size = packages.size(); i < size; i++){
				JSONObject objectInArray = (JSONObject)packages.get(i);
				String packageId = (String)objectInArray.get("id");
				toReturn.add(client.getDataset(packageId));
			}

		}catch(Exception e){
			logger.error("Failed to get groups information", e);
		}

		return  toReturn;

	}

	@Override
	public String patchProductWithJSON(String productId, JSONObject jsonRequest,
			String apiKey) {

		checkNotNull(productId);
		checkNotNull(jsonRequest);
		checkNotNull(apiKey);

		logger.info("Request of patching product " + productId + " with json " + jsonRequest);

		try(CloseableHttpClient httpClient = HttpClientBuilder.create().build();){

			String pathUpdatePatch = getCatalogueUrl() + "/api/3/action/package_patch";
			HttpPost httpPostRequest = new HttpPost(pathUpdatePatch);
			httpPostRequest.setHeader("Authorization", CKAN_TOKEN_SYS);

			StringEntity params = new StringEntity(jsonRequest.toJSONString(), ContentType.APPLICATION_JSON);
			httpPostRequest.setEntity(params);

			HttpResponse response = httpClient.execute(httpPostRequest);

			if (response.getStatusLine().getStatusCode() < 200 || response.getStatusLine().getStatusCode() >= 300) {
				throw new RuntimeException("failed to patch the product. response status line from "
						+ pathUpdatePatch + " was: " + response.getStatusLine());
			}

			return null;

		}catch(Exception e){
			logger.error("Error while trying to patch grsf record " + e.getMessage());
			return e.getMessage();
		}
	}

	@Override
	public List<String> getProductsIdsInGroupOrOrg(String orgOrGroupName, boolean isOrganization, int start, int rows) throws ClassNotFoundException, SQLException {

		List<String> toReturn = new ArrayList<String>();

		checkNotNull(orgOrGroupName);
		checkArgument(start >= 0);
		checkArgument(rows >= 0);

		Connection connection = getConnection();
		try{

			ResultSet rs;
			if(isOrganization){

				String joinQuery = "SELECT \"package\".\"id\" AS \"dataset_id\" FROM \"package\" INNER JOIN \"group\" ON"
						+ " \"package\".\"owner_org\"=\"group\".\"id\" WHERE \"group\".\"name\"=? "
						+ "AND \"group\".\"is_organization\"=? AND \"package\".\"type\"=? AND \"package\".\"state\"=? LIMIT ? OFFSET ?; ";

				PreparedStatement preparedStatement = connection.prepareStatement(joinQuery);
				preparedStatement.setString(1, orgOrGroupName);
				preparedStatement.setBoolean(2, isOrganization);
				preparedStatement.setString(3, "dataset");
				preparedStatement.setString(4, "active");
				preparedStatement.setBigDecimal(5, new BigDecimal(rows));
				preparedStatement.setBigDecimal(6, new BigDecimal(start));

				rs = preparedStatement.executeQuery();

			}else{

				/**
				 * Inner join between the member table and the package table.
				 * Basically every time a dataset is added to a group, a new row is added to the table
				 * where table_id is the package_id, and group_id is the owner group identifier (not the name)
				 */
				String groupId = client.getGroup(orgOrGroupName).getId();
				String joinQuery = "SELECT \"table_id\" AS \"dataset_id\" FROM \"package\" INNER JOIN \"member\" ON"
						+ " \"member\".\"table_id\"=\"package\".\"id\" WHERE \"group_id\"=? "
						+ "AND \"member\".\"state\"=? LIMIT ? OFFSET ?;";

				PreparedStatement preparedStatement = connection.prepareStatement(joinQuery);
				preparedStatement.setString(1, groupId);
				preparedStatement.setString(2, "active");
				preparedStatement.setBigDecimal(3, new BigDecimal(rows));
				preparedStatement.setBigDecimal(4, new BigDecimal(start));

				rs = preparedStatement.executeQuery();

			}

			while (rs.next()) {
				toReturn.add(rs.getString("dataset_id"));
			}

		}catch(Exception e){
			logger.error("Failed to retrieve the ids of products in group/org. Error is " + e.getMessage());
			return null;
		}finally{
			closeConnection(connection);
		}
		return toReturn;


	}

	@Override
	public boolean deleteGroup(String groupName, boolean purge) {
		
		checkNotNull(groupName);

		logger.info("Request of deleting group " + groupName + ". Purge is " + Boolean.toString(purge));

		try(CloseableHttpClient httpClient = HttpClientBuilder.create().build();){

			String deletePath = getCatalogueUrl() + "/api/3/action/group_delete";
			String purgePath = getCatalogueUrl() + "/api/3/action/group_purge";
			
			String requestToPerform = purge ? purgePath : deletePath;
			
			HttpPost httpPostRequest = new HttpPost(requestToPerform);
			httpPostRequest.setHeader("Authorization", CKAN_TOKEN_SYS);
			
			JSONObject obj = new JSONObject();
			obj.put("id", groupName);

			StringEntity params = new StringEntity(obj.toJSONString(), ContentType.APPLICATION_JSON);
			httpPostRequest.setEntity(params);

			HttpResponse response = httpClient.execute(httpPostRequest);

			if (response.getStatusLine().getStatusCode() < 200 || response.getStatusLine().getStatusCode() >= 300) {
				throw new RuntimeException("failed to delete/purge the group. response status line from "
						+ requestToPerform + " was: " + response.getStatusLine());
			}

			return true;

		}catch(Exception e){
			logger.error("Error while trying to delete/purge the group " + e.getMessage());
			return false;
		}
	}
}