package eu.dnetlib.openaire.hadoop.utils;

import java.util.Collection;
import java.util.Set;

import com.google.common.base.Function;
import com.google.common.base.Predicate;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import eu.dnetlib.data.proto.DatasourceOrganizationProtos.DatasourceOrganization.Provision;
import eu.dnetlib.data.proto.DedupProtos.Dedup;
import eu.dnetlib.data.proto.DedupSimilarityProtos.DedupSimilarity;
import eu.dnetlib.data.proto.PersonPersonProtos.PersonPerson.CoAuthorship;
import eu.dnetlib.data.proto.PersonResultProtos.PersonResult.Authorship;
import eu.dnetlib.data.proto.ProjectOrganizationProtos.ProjectOrganization.Participation;
import eu.dnetlib.data.proto.ProjectPersonProtos.ProjectPerson.ContactPerson;
import eu.dnetlib.data.proto.RelTypeProtos.RelType;
import eu.dnetlib.data.proto.RelTypeProtos.SubRelType;
import eu.dnetlib.data.proto.ResultOrganizationProtos.ResultOrganization.Affiliation;
import eu.dnetlib.data.proto.ResultProjectProtos.ResultProject.Outcome;
import eu.dnetlib.data.proto.ResultResultProtos.ResultResult.Part;
import eu.dnetlib.data.proto.ResultResultProtos.ResultResult.PublicationDataset;
import eu.dnetlib.data.proto.ResultResultProtos.ResultResult.Similarity;
import eu.dnetlib.data.proto.ResultResultProtos.ResultResult.Supplement;
import eu.dnetlib.data.proto.TypeProtos.Type;

/**
 * Common static utility methods to manage the hbase tables
 *
 * @author claudio
 */
public class HBaseTableUtils {

	private static final String SEPARATOR = "_";

	private static final Function<Type, String> typeName = new Function<Type, String>() {

		@Override
		public String apply(final Type type) {
			return type.toString();
		}
	};

	public static Set<String> listAllColumns() {
		final Set<String> union = Sets.union(listEntities(), listRelationships());
		return Sets.union(union, listDedupRelationships());
	}

	public static Set<String> listDedupColumns(final Collection<Type> entityTypes) {
		final Set<String> entities = listEntities(Lists.newArrayList(Iterables.transform(entityTypes, typeName)));
		return Sets.union(entities, listDedupRelationships());
	}

	private static Set<String> listDedupRelationships() {
		final Set<String> cfs = Sets.newHashSet();
		cfs.add(RelType.organizationOrganization + SEPARATOR + SubRelType.dedup + SEPARATOR + Dedup.RelName.merges);
		cfs.add(RelType.organizationOrganization + SEPARATOR + SubRelType.dedup + SEPARATOR + Dedup.RelName.isMergedIn);
		cfs.add(RelType.organizationOrganization + SEPARATOR + SubRelType.dedupSimilarity + SEPARATOR + DedupSimilarity.RelName.isSimilarTo);

		cfs.add(RelType.personPerson + SEPARATOR + SubRelType.dedup + SEPARATOR + Dedup.RelName.merges);
		cfs.add(RelType.personPerson + SEPARATOR + SubRelType.dedup + SEPARATOR + Dedup.RelName.isMergedIn);
		cfs.add(RelType.personPerson + SEPARATOR + SubRelType.dedupSimilarity + SEPARATOR + DedupSimilarity.RelName.isSimilarTo);

		cfs.add(RelType.resultResult + SEPARATOR + SubRelType.dedup + SEPARATOR + Dedup.RelName.merges);
		cfs.add(RelType.resultResult + SEPARATOR + SubRelType.dedup + SEPARATOR + Dedup.RelName.isMergedIn);
		cfs.add(RelType.resultResult + SEPARATOR + SubRelType.dedupSimilarity + SEPARATOR + DedupSimilarity.RelName.isSimilarTo);

		return cfs;
	}

	private static Set<String> listEntities(final Collection<String> entityType) {
		return Sets.newHashSet(Iterables.filter(Iterables.transform(Lists.newArrayList(Type.values()), typeName), new Predicate<String>() {

			@Override
			public boolean apply(final String s) {
				return entityType.contains(s);
			}
		}));
	}

	public static Set<String> listEntities() {
		return Sets.newHashSet(Iterables.transform(Lists.newArrayList(Type.values()), typeName));
	}

	public static Set<String> listRelationships() {
		final Set<String> cfs = Sets.newHashSet();
		cfs.add(RelType.datasourceOrganization + SEPARATOR + SubRelType.provision + SEPARATOR + Provision.RelName.isProvidedBy);
		cfs.add(RelType.datasourceOrganization + SEPARATOR + SubRelType.provision + SEPARATOR + Provision.RelName.provides);

		cfs.add(RelType.personPerson + SEPARATOR + SubRelType.coauthorship + SEPARATOR + CoAuthorship.RelName.isCoauthorOf);

		cfs.add(RelType.personResult + SEPARATOR + SubRelType.authorship + SEPARATOR + Authorship.RelName.isAuthorOf);
		cfs.add(RelType.personResult + SEPARATOR + SubRelType.authorship + SEPARATOR + Authorship.RelName.hasAuthor);

		cfs.add(RelType.projectOrganization + SEPARATOR + SubRelType.participation + SEPARATOR + Participation.RelName.hasParticipant);
		cfs.add(RelType.projectOrganization + SEPARATOR + SubRelType.participation + SEPARATOR + Participation.RelName.isParticipant);

		cfs.add(RelType.projectPerson + SEPARATOR + SubRelType.contactPerson + SEPARATOR + ContactPerson.RelName.isContact);
		cfs.add(RelType.projectPerson + SEPARATOR + SubRelType.contactPerson + SEPARATOR + ContactPerson.RelName.hasContact);

		cfs.add(RelType.resultProject + SEPARATOR + SubRelType.outcome + SEPARATOR + Outcome.RelName.isProducedBy);
		cfs.add(RelType.resultProject + SEPARATOR + SubRelType.outcome + SEPARATOR + Outcome.RelName.produces);

		cfs.add(RelType.resultResult + SEPARATOR + SubRelType.similarity + SEPARATOR + Similarity.RelName.hasAmongTopNSimilarDocuments);
		cfs.add(RelType.resultResult + SEPARATOR + SubRelType.similarity + SEPARATOR + Similarity.RelName.isAmongTopNSimilarDocuments);

		cfs.add(RelType.resultResult + SEPARATOR + SubRelType.supplement + SEPARATOR + Supplement.RelName.isSupplementedBy);
		cfs.add(RelType.resultResult + SEPARATOR + SubRelType.supplement + SEPARATOR + Supplement.RelName.isSupplementTo);
		cfs.add(RelType.resultResult + SEPARATOR + SubRelType.part + SEPARATOR + Part.RelName.isPartOf);
		cfs.add(RelType.resultResult + SEPARATOR + SubRelType.part + SEPARATOR + Part.RelName.hasPart);

		cfs.add(RelType.resultResult + SEPARATOR + SubRelType.publicationDataset + SEPARATOR + PublicationDataset.RelName.isRelatedTo);

		cfs.add(RelType.resultOrganization + SEPARATOR + SubRelType.affiliation + SEPARATOR + Affiliation.RelName.isAuthorInstitutionOf);
		cfs.add(RelType.resultOrganization + SEPARATOR + SubRelType.affiliation + SEPARATOR + Affiliation.RelName.hasAuthorInstitution);

		return cfs;
	}

	public enum VolatileColumnFamily {
		dedup, dedupPerson; // instance is here to remove the old protos

		public static boolean isVolatile(final String columnName) {
			try {
				return VolatileColumnFamily.valueOf(columnName) != null;
			} catch (final Throwable e) {
				return false;
			}
		}
	}

}