D-Net

package eu.dnetlib.validator.service.impl;

import eu.dnetlib.domain.functionality.validator.JobForValidation;

import eu.dnetlib.domain.functionality.validator.RuleSet;

import eu.dnetlib.domain.functionality.validator.StoredJob;

import eu.dnetlib.enabling.tools.blackboard.BlackboardJob;

import eu.dnetlib.enabling.tools.blackboard.BlackboardNotificationHandler;

import eu.dnetlib.enabling.tools.blackboard.BlackboardServerHandler;

import eu.dnetlib.validator.commons.dao.DaoException;

import eu.dnetlib.validator.commons.dao.jobs.JobsDAO;

import eu.dnetlib.validator.commons.dao.rules.RulesDAO;

import eu.dnetlib.validator.commons.dao.rules.RulesetsDAO;

import eu.dnetlib.validator.engine.Validator;

import eu.dnetlib.validator.engine.ValidatorException;

import eu.dnetlib.validator.engine.data.Rule;

import eu.dnetlib.validator.engine.execution.Job;

import eu.dnetlib.validator.service.impls.ValidatorRestore;

import eu.dnetlib.validator.service.impls.listeners.*;

import eu.dnetlib.validator.service.impls.providers.DnetProvider;

import eu.dnetlib.validator.service.impls.providers.OAIPMHRecordProvider;

import eu.dnetlib.validator.service.impls.providers.OAIPMHSinglePageVerbProvider;

import eu.dnetlib.validator.service.impls.rules.ChainRule;

import eu.dnetlib.validator.service.impls.rules.RuleTypes;

import org.apache.log4j.Logger;

import org.springframework.transaction.annotation.Propagation;

import org.springframework.transaction.annotation.Transactional;

import java.util.*;

@Transactional

public class ValidatorManagerImpl implements ValidatorManager {

	private Validator validator;

	private JobsDAO jobsDao;

	private RulesetsDAO rulesetsDao;

	private RulesDAO rulesDao;

	private ValidatorRestore valRestore;

	private static Logger logger = Logger.getLogger(ValidatorManagerImpl.class);

	private ListenersManager listenersManager;

//	this.restartJobs();

	public void start() {

		logger.info("Initializing Validator Manager module");

		if (valRestore.isAutoRestoreOnStartup()) {

			logger.info("auto-restoring OpenAire Validator is enabled..");

			List<StoredJob> jobs = valRestore.deleteUncompleted();

			valRestore.restartJobs(jobs);

			logger.info("finished restoring OpenAire Validator..");

		} else {

			logger.info("auto-restoring OpenAire Validator is disabled..");

		}

	}

	public StoredJob getStoredJob(int jobId, String groupBy)

			throws ValidatorException {

		try {

			logger.info("Getting job summary for job " + jobId +" with groupBy: "+groupBy);

			return jobsDao.getJobSummary(jobId,groupBy);

		} catch (Exception e) {

			logger.error("error getting job summary for job " + jobId, e);

			throw new ValidatorException(e);

		}

	}

	public List<StoredJob> getStoredJobs(String userMail, String jobType,

			Integer offset, Integer limit, String dateFrom, String dateTo)

			throws ValidatorException {

		return this.getStoredJobs(userMail, jobType, offset, limit, dateFrom, dateTo, null);

	}

	public List<StoredJob> getStoredJobs(String userMail, String jobType,

										 Integer offset, Integer limit, String dateFrom, String dateTo, String jobStatus)

			throws ValidatorException {

		try {

			logger.debug("Getting jobs of user " + userMail);

			return jobsDao.getJobs(userMail, jobType, offset, limit, dateFrom, dateTo, jobStatus);

		} catch (Exception e) {

			logger.error("error Getting jobs of user " + userMail, e);

			throw new ValidatorException(e);

		}

	}

	public int getStoredJobsTotalNumber(String userMail, String jobType)

			throws ValidatorException {

		return this.getStoredJobsTotalNumber(userMail, jobType, null);

	}

	public int getStoredJobsTotalNumber(String userMail, String jobType, String jobStatus)

			throws ValidatorException {

		try {

			logger.debug("Getting jobs total sum of user " + userMail);

			return jobsDao.getJobsTotalNumber(userMail, jobType, jobStatus);

		} catch (Exception e) {

			logger.error("error Getting jobs sum of user " + userMail, e);

			throw new ValidatorException(e);

		}

	}

	public List<RuleSet> getRuleSets() throws ValidatorException {

		try {

			logger.info("Getting rulesets");

			return rulesetsDao.getRuleSets();

		} catch (Exception e) {

			logger.error("error Getting rulesets ", e);

			throw new ValidatorException(e);

		}

	}

	@Transactional(readOnly = false, propagation = Propagation.REQUIRES_NEW)

	public StoredJob beginDataJobForWorkflow(String mdstoreId, String guidelines, String groupBy, int records, int workers, BlackboardJob bJob, BlackboardNotificationHandler<BlackboardServerHandler> blackboardHandler, int jobStatusUpdateInterval, boolean outputEpr, boolean blacklistedRecords, String blacklistGuidelines) throws ValidatorException{

		try {

			logger.info("Submitting data job for workflow");

//			String desiredCompatibility = "openaire3.0";

			Set<Integer> ruleIdsForBlacklist = new HashSet<Integer>();

			Set<Integer> ruleIdsContent = new HashSet<Integer>();

			for (RuleSet ruleset : rulesetsDao.getRuleSets()) {

				if (ruleset.getGuidelinesAcronym().equals(guidelines)) {

					ruleIdsContent = ruleset.getContentRulesIds();

				}	

				if (blacklistedRecords && ruleset.getGuidelinesAcronym().equals(blacklistGuidelines)) {

					ruleIdsForBlacklist = ruleset.getContentRulesIds();

					ruleIdsContent.addAll(ruleIdsForBlacklist);

				}

			}

			Properties props = new Properties();

			props.setProperty(DnetProvider.MDSTORE_ID, mdstoreId);

			props.setProperty(DnetProvider.BATCH_SIZE, "50");

			if (bJob.getParameters().get("batchSize") != null )

				props.setProperty(DnetProvider.BATCH_SIZE, bJob.getParameters().get("batchSize"));

			props.setProperty(DnetProvider.RECORDS, Integer.toString(records));

			StoredJob newJob = new StoredJob();

			newJob.setUserEmail("Workflow Service");

			newJob.setValidationType("C");

			newJob.setDesiredCompatibilityLevel(guidelines);

			newJob.setContentJobStatus("ongoing");

			newJob.setUsageJobStatus("none");

			newJob.setJobType("Workflow Request");

			newJob.setDuration("--");

			newJob.setBaseUrl(mdstoreId);

			newJob.setRules(ruleIdsContent);

			newJob.setRecords(records);

			newJob.setValidationSet("dnet-workflow");

			newJob.setGroupByXpath(groupBy);

			newJob.setMetadataPrefix("oai_dc");

			newJob.setId(-1);

			int jobIdStored = jobsDao.save(newJob);

			Set<Rule> rulesContent = new HashSet<Rule>();

			logger.debug("Selected content rules number: " + ruleIdsContent.size());

			for (Integer ruleId : ruleIdsContent){

				eu.dnetlib.domain.functionality.validator.Rule tempRule=rulesDao.get(ruleId);

				//special behaviour type of rule is chain

				if(tempRule.getType().equals("ChainRule")) {

					ChainRule<Rule> chainRule = this.handleChain(tempRule);

					rulesContent.add(chainRule);

				}

				else {

					rulesContent.add((Rule) this.getRuleClassInstanceByType(tempRule.getType(), tempRule.getConfiguration(), tempRule.getId()));

				}

			}

			Job jobContent = new Job(jobIdStored, 3, rulesContent, props);

			ValidatorJobListener listenerContent = listenersManager.createListener();

			listenerContent.setJobSubmittedId(jobIdStored);

			listenerContent.setJobSubmittedUser("Workflow Service");

			listenerContent.setGroupBy_xpath(groupBy);

			listenerContent.setValidationType("content");

			listenerContent.setInternalJobsSum(workers);

			listenerContent.setBlacklistRuleIds(ruleIdsForBlacklist);

			DnetListener dnetListener = listenersManager.createDnetListener();

			dnetListener.setJob(bJob);

			dnetListener.setBlackboardHandler(blackboardHandler);

			dnetListener.setInternalJobsSum(workers);

			dnetListener.setValidationJobId(jobIdStored);

			dnetListener.setJobStatusUpdateInterval(jobStatusUpdateInterval);

			if (blacklistedRecords) {

				dnetListener.setEnableOutputToDisk(true);

			}

			if (outputEpr)

				dnetListener.setEnableOutputToRS(true);

			validator.submitJob(jobContent, workers, listenerContent, dnetListener);

			return newJob;

		} catch (Exception e) {

			logger.error("Error Submitting content job", e);

			throw new ValidatorException(e);

		}

	}

	@Transactional(readOnly = false, propagation = Propagation.REQUIRED)

	public void submitJob(JobForValidation job) throws ValidatorException {

		try {

			logger.info("Submiting validation job requested by user: " + job.getUserEmail());

			if (job.isRegistration()) {

				logger.debug("initiating preregistration validations on repo " + job.getBaseUrl() + " for user " + job.getUserEmail() + "and desired compatibility: " + job.getDesiredCompatibilityLevel());

				for (RuleSet ruleset : rulesetsDao.getRuleSets()) {

					if (ruleset.getGuidelinesAcronym().equals(job.getDesiredCompatibilityLevel())) {

						job.setSelectedContentRules(ruleset.getContentRulesIds());

						job.setSelectedUsageRules(ruleset.getUsageRulesIds());

					}	

				}	

			}

			if ( job.getDesiredCompatibilityLevel().equalsIgnoreCase("openaire2.0") || 

					job.getDesiredCompatibilityLevel().equalsIgnoreCase("openaire3.0") || 

					job.getDesiredCompatibilityLevel().equalsIgnoreCase("driver") ) {

				logger.debug("Chosen set: OpenAIRE For Literature Repositories");

				logger.debug("Setting METADATA_PREFIX to: oai_dc");

				job.setMetadataPrefix("oai_dc");

			} else if ( job.getDesiredCompatibilityLevel().equalsIgnoreCase("openaire2.0_data")) {

				logger.debug("Chosen set: OpenAIRE For Data Archives");

				logger.debug("Setting METADATA_PREFIX to: oai_datacite");

				job.setMetadataPrefix("oai_datacite");			

			} else if ( job.getDesiredCompatibilityLevel().equalsIgnoreCase("openaire3.0_cris")) {

				logger.debug("Chosen set: OpenAIRE For Cris");

				logger.debug("Setting METADATA_PREFIX to: oai_CERIF_openaire");

				job.setMetadataPrefix("oai_CERIF_openaire");			

			}

			StoredJob newJob = new StoredJob(job);

			int usageJobs = 0;

			int contentJobs = 0;

			Set<Integer> totalRules = new HashSet<Integer>();

			Map<String,Set<Integer>> verbRuleMap = null;

			String validationType = "";

			if (job.getSelectedContentRules() != null) {

				contentJobs = 1;

				validationType +="C";

				totalRules.addAll(job.getSelectedContentRules());

				newJob.setContentJobStatus("ongoing");

			} else {

				newJob.setContentJobStatus("none");

			}

			if (job.getSelectedUsageRules() != null) {

				logger.debug("Creating map for provider information");

				verbRuleMap= new HashMap<String,Set<Integer>>();

				Set<Integer> old,temp = null;

				for (Integer id : job.getSelectedUsageRules()){

					eu.dnetlib.domain.functionality.validator.Rule ruleStored = rulesDao.get(id);

					logger.debug("Checking for verb : "+ruleStored.getProvider_information());

			        if((old=verbRuleMap.get(ruleStored.getProvider_information())) == null){

			        	logger.debug("Verb doesn't exist");

			            temp = new HashSet<Integer>();

			            temp.add(ruleStored.getId());

			            verbRuleMap.put(ruleStored.getProvider_information(),temp);

			        }else{

			        	logger.debug("Verb exists");

			            old.add(ruleStored.getId());

			        }

				}

				usageJobs = verbRuleMap.size();

				validationType +="U";

				totalRules.addAll(job.getSelectedUsageRules());

				newJob.setUsageJobStatus("ongoing");

			} else {

				newJob.setUsageJobStatus("none");

			}

			Map<String, Set<Rule>> entityChosenRulesMap = null;

			Map<String, Set<Rule>> entityChosenRulesMapReferential = null;

			if (job.isCris()) {

				entityChosenRulesMap = new HashMap<String, Set<Rule>>();

				entityChosenRulesMapReferential = new HashMap<String, Set<Rule>>();

				this.prepareCrisJobs(job, entityChosenRulesMap, entityChosenRulesMapReferential);

				newJob.setGroupByXpath("//header/setSpec");

				contentJobs = entityChosenRulesMap.keySet().size() + entityChosenRulesMapReferential.keySet().size();

				//TODO move in uoa-domain

				newJob.setCris(true);

				newJob.setSelectedCrisEntities(job.getSelectedCrisEntities());

				newJob.setCrisReferentialChecks(job.isCrisReferentialChecks());

			}

			newJob.setValidationType(validationType);

			if (job.isRegistration()) {

				newJob.setJobType("Registration Request");

				//TODO move in uoa-domain

				newJob.setActivationId(job.getActivationId());

				newJob.setRegistration(true);

				newJob.setUpdateExisting(job.isUpdateExisting());

				newJob.setOfficialName(job.getOfficialName());

				newJob.setAdminEmails(job.getAdminEmails());

				newJob.setDatasourceId(job.getDatasourceId());

				newJob.setInterfaceId(job.getInterfaceId());

				newJob.setInterfaceIdOld(job.getInterfaceIdOld());

				newJob.setRepoType(job.getRepoType());

			} else

				newJob.setJobType("Compatibility Test");

			newJob.setDuration("--");

			newJob.setRules(totalRules);

			newJob.setId(-1);

			int jobIdStored = jobsDao.save(newJob);

			RegistrationListener regListener = null;

			if (job.isRegistration()) {

				regListener = listenersManager.createRegListener();

				regListener.setActivationId(job.getActivationId());

				regListener.setBaseUrl(job.getBaseUrl());

				regListener.setActivationId(job.getActivationId());

				regListener.setAdminEmails(job.getAdminEmails());

				regListener.setOfficialName(job.getOfficialName());

				regListener.setUserMail(job.getUserEmail());

				regListener.setDatasourceId(job.getDatasourceId());

				regListener.setInterfaceId(job.getInterfaceId());

				regListener.setValidationSet(job.getValidationSet());

				regListener.setDesiredCompLevel(job.getDesiredCompatibilityLevel());

				regListener.setRepoType(job.getRepoType());

				regListener.setInterfaceIdOld(job.getInterfaceIdOld());

				regListener.setUpdateExisting(job.isUpdateExisting());	

				regListener.setTotalJobs(usageJobs + contentJobs);

			}

			CompatibilityTestListener compTestListener = listenersManager.createCompTestListener();

			compTestListener.setValidationSet(job.getValidationSet());

			compTestListener.setGuidelines(job.getDesiredCompatibilityLevel());

			compTestListener.setTotalJobs(usageJobs + contentJobs);

//CONTENT

			if (job.getSelectedContentRules() != null) {

				Set<Rule> rulesContent = new HashSet<Rule>();

				Properties props = new Properties();

				props.setProperty(OAIPMHRecordProvider.BASEURL, job.getBaseUrl());

				props.setProperty(OAIPMHRecordProvider.METADATA_PREFIX, newJob.getMetadataPrefix());

				props.setProperty(OAIPMHRecordProvider.RECORDS,Integer.toString(job.getRecords()));

				props.setProperty(OAIPMHRecordProvider.SET,job.getValidationSet());

				Job jobContent = null;

				ValidatorJobListener listenerContent = listenersManager.createListener();

				listenerContent.setJobSubmittedId(jobIdStored);

				listenerContent.setJobSubmittedUser(job.getUserEmail());

				listenerContent.setGroupBy_xpath(newJob.getGroupByXpath());

				listenerContent.setValidationType("content");

				listenerContent.setInternalJobsSum(contentJobs);

				if (job.isCris()) {

					logger.debug("Submiting job for cris.");

					logger.debug("Total content jobs: " + contentJobs);

//					logger.debug("Selected content rules number: " + job.getSelectedContentRules().size());

					jobContent = new Job(jobIdStored, 4, rulesContent, props);

					validator.submitJobForCris(jobContent, entityChosenRulesMap, entityChosenRulesMapReferential, listenerContent, compTestListener);

				} else {

					logger.debug("Selected content rules number: " + job.getSelectedContentRules().size());

					for (Integer ruleId : job.getSelectedContentRules()){

						eu.dnetlib.domain.functionality.validator.Rule tempRule=rulesDao.get(ruleId);

						//special behaviour type of rule is chain

						if(tempRule.getType().equals("ChainRule")) {

							ChainRule<Rule> chainRule = this.handleChain(tempRule);

							rulesContent.add(chainRule);

						}

						else {

							rulesContent.add((Rule) this.getRuleClassInstanceByType(tempRule.getType(), tempRule.getConfiguration(), tempRule.getId()));

						}

					}

					jobContent = new Job(jobIdStored, 1, rulesContent, props);

					if (job.isRegistration()) {

						validator.submitJob(jobContent, 1, listenerContent, regListener);

					} else {

						validator.submitJob(jobContent, 1, listenerContent, compTestListener);

					}	

				}

			}

			//USAGE	

			if (job.getSelectedUsageRules() != null) {

				ValidatorJobListener listenerUsage = listenersManager.createListener();

				listenerUsage.setJobSubmittedId(jobIdStored);

				listenerUsage.setValidationType("usage");

				listenerUsage.setJobSubmittedUser(job.getUserEmail());

				listenerUsage.setInternalJobsSum(usageJobs);

				List <Job> jobsUsage = new ArrayList<Job>();

				for (Map.Entry<String, Set<Integer>> entry : verbRuleMap.entrySet()) {

					Properties pros = new Properties();

					pros.setProperty(OAIPMHSinglePageVerbProvider.VERB,entry.getKey());

					pros.setProperty(OAIPMHSinglePageVerbProvider.BASEURL, job.getBaseUrl());

					Set<Rule> rulesUsage = new HashSet<Rule>();

					for (Integer ruleId : entry.getValue()){

						eu.dnetlib.domain.functionality.validator.Rule tempRule = rulesDao.get(ruleId);

						logger.debug("prepare to add rule to registry with regexp: "+tempRule.getConfiguration().getProperty("regexp"));

						rulesUsage.add((Rule)this.getRuleClassInstanceByType(tempRule.getType(), tempRule.getConfiguration(), tempRule.getId()));

					}

					jobsUsage.add(new Job(jobIdStored, 2, rulesUsage, pros));

				}

				for (Job jobUsage : jobsUsage ) {

					if (job.isRegistration()) {

						validator.submitJob(jobUsage, 1, listenerUsage, regListener);

					} else {

						validator.submitJob(jobUsage, 1, listenerUsage, compTestListener);

					}

				}

			}			

		} catch (Exception e) {

			logger.error("error submitting job ", e);

			throw new ValidatorException(e);

		}

	}

	private void prepareCrisJobs(JobForValidation job,

			Map<String, Set<Rule>> entityChosenRulesMap,

			Map<String, Set<Rule>> entityChosenRulesMapReferential) throws ValidatorException, DaoException {

		//getting rules per entity and leave only chosen ones

		logger.debug("Selected Entities: " + job.getSelectedCrisEntities());

		for (String entity : RuleTypes.getEntities().keySet()) {

			logger.debug("checking entity: " + entity);

			Set<Rule> rulesBasic = new HashSet<Rule>();

			Set<Rule> rulesReferential = new HashSet<Rule>();

			if (job.getSelectedCrisEntities().contains(entity)) {

				logger.debug("entity: " + entity + " is selected");

				for (eu.dnetlib.domain.functionality.validator.Rule rule : rulesDao.getAllRulesByJobTypeEntityType("content", entity)) {

					if (job.getSelectedContentRules().contains(rule.getId())) {

						eu.dnetlib.domain.functionality.validator.Rule tempRule=rulesDao.get(rule.getId());

						if (rule.getName().contains("Relationship")) {

							if (job.isCrisReferentialChecks()) {

								rulesReferential.add((Rule) this.getRuleClassInstanceByType(tempRule.getType(), tempRule.getConfiguration(), tempRule.getId()));

							} 

						} else {

							rulesBasic.add((Rule) this.getRuleClassInstanceByType(tempRule.getType(), tempRule.getConfiguration(), tempRule.getId()));

						}

					}

				}

				logger.debug("Basic rules: " + rulesBasic.size());

				logger.debug("Referential rules: " + rulesReferential.size());

				entityChosenRulesMap.put(RuleTypes.getSetOfEntity(entity), rulesBasic);

				if (job.isCrisReferentialChecks() && !rulesReferential.isEmpty())

					entityChosenRulesMapReferential.put(RuleTypes.getSetOfEntity(entity), rulesReferential);

			} else {

				logger.debug("entity: " + entity + " is not selected");

			}

		}

		logger.debug("Return entities: " + entityChosenRulesMap.keySet());

	}

	public Rule getRuleClassInstanceByType(String type, Properties pros, int id) throws ValidatorException {

		logger.debug("getting rule object of type "+type);

		String classname = RuleTypes.getClassOfType(type);

		if (classname == null){

			logger.debug("error getting rule object of type "+type+" classname=null");

			return null;

		}

		else {

			try {

				Class<?> clazz = Class.forName(classname);

				logger.debug("classname: "+clazz.getName());

				return (Rule) clazz.getConstructor(new Class[]{Properties.class,Integer.TYPE}).newInstance(pros,id);

			} catch (Exception e) {

				logger.debug("error getting rule object: "+e);

				return null;

			} 

		}

	}

	private ChainRule<Rule> handleChain(eu.dnetlib.domain.functionality.validator.Rule tempRule) throws ValidatorException, DaoException {

		logger.debug("chain rule found");

		List<Rule> rules_chain = new ArrayList<Rule>();

		eu.dnetlib.domain.functionality.validator.Rule tempRuleChain1=rulesDao.get(Integer.parseInt(tempRule.getConfiguration().getProperty("rule_1")));

		if(tempRuleChain1.getType().equals("ChainRule")) {

			ChainRule<Rule> chainRule1 = this.handleChain(tempRuleChain1);

			rules_chain.add(chainRule1);

		} else {

			rules_chain.add((Rule) this.getRuleClassInstanceByType(tempRuleChain1.getType(), tempRuleChain1.getConfiguration(),tempRuleChain1.getId()));

		}

		eu.dnetlib.domain.functionality.validator.Rule tempRuleChain2=rulesDao.get(Integer.parseInt(tempRule.getConfiguration().getProperty("rule_2")));

		if(tempRuleChain2.getType().equals("ChainRule")) {

			ChainRule<Rule> chainRule2 = this.handleChain(tempRuleChain2);

			rules_chain.add(chainRule2);

		} else {

			rules_chain.add((Rule) this.getRuleClassInstanceByType(tempRuleChain2.getType(), tempRuleChain2.getConfiguration(), tempRuleChain2.getId()));

		}

		Properties chainPros = new Properties();

		chainPros.setProperty(ChainRule.TYPE,tempRule.getConfiguration().getProperty(ChainRule.TYPE));

		ChainRule<Rule> chainRule = new ChainRule<Rule>(chainPros, tempRule.getId(), rules_chain);

		return chainRule;

	}

	public Validator getValidator() {

		return validator;

	}

	public ListenersManager getListenersManager() {

		return listenersManager;

	}

	public void setListenersManager(ListenersManager listenersManager) {

		this.listenersManager = listenersManager;

	}

	public void setValidator(Validator validator) {

		this.validator = validator;

	}

	public JobsDAO getJobsDao() {

		return jobsDao;

	}

	public void setJobsDao(JobsDAO jobsDao) {

		this.jobsDao = jobsDao;

	}

	public RulesetsDAO getRulesetsDao() {

		return rulesetsDao;

	}

	public void setRulesetsDao(RulesetsDAO rulesetsDao) {

		this.rulesetsDao = rulesetsDao;

	}

	public RulesDAO getRulesDao() {

		return rulesDao;

	}

	public void setRulesDao(RulesDAO rulesDao) {

		this.rulesDao = rulesDao;

	}

	public ValidatorRestore getValRestore() {

		return valRestore;

	}

	public void setValRestore(ValidatorRestore valRestore) {

		this.valRestore = valRestore;

	}

}