D-Net

package eu.dnetlib.msro.openaireplus.workflows.nodes.consistency;

import java.io.IOException;

import java.io.StringReader;

import java.util.Date;

import java.util.List;

import java.util.Objects;

import java.util.Set;

import java.util.stream.Collectors;

import javax.annotation.Resource;

import org.antlr.stringtemplate.StringTemplate;

import org.apache.commons.io.IOUtils;

import org.apache.commons.lang.math.NumberUtils;

import org.apache.commons.lang3.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.dom4j.Document;

import org.dom4j.DocumentException;

import org.dom4j.io.SAXReader;

import org.springframework.beans.factory.annotation.Autowired;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.data.mdstore.MDStoreServiceException;

import eu.dnetlib.data.mdstore.modular.connector.MDStoreDao;

import eu.dnetlib.data.objectstore.modular.connector.ObjectStoreDao;

import eu.dnetlib.data.objectstore.rmi.ObjectStoreServiceException;

import eu.dnetlib.enabling.datasources.LocalOpenaireDatasourceManager;

import eu.dnetlib.enabling.datasources.common.Api;

import eu.dnetlib.enabling.datasources.common.ApiParam;

import eu.dnetlib.enabling.datasources.common.DsmException;

import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpException;

import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpService;

import eu.dnetlib.enabling.is.registry.rmi.ISRegistryException;

import eu.dnetlib.enabling.is.registry.rmi.ISRegistryService;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.miscutils.datetime.DateUtils;

import eu.dnetlib.msro.workflows.nodes.SimpleJobNode;

import eu.dnetlib.msro.workflows.util.ProgressProvider;

import eu.dnetlib.msro.workflows.util.WorkflowsConstants;

public class FixRepoMdstoreSizesJobNode extends SimpleJobNode implements ProgressProvider {

	@Autowired

	private LocalOpenaireDatasourceManager dsManager;

	@Autowired

	private UniqueServiceLocator serviceLocator;

	@Resource(name = "mongodbMDStoreDao")

	private MDStoreDao mdstoreDao;

	@Autowired

	private ObjectStoreDao objectStoreDao;

	private final DateUtils dateUtils = new DateUtils();

	private int current = 0;

	private int total = 0;

	private ISRegistryService registry;

	private ISLookUpService lookup;

	private boolean alwaysUpdate = false;

	private static final Log log = LogFactory.getLog(FixRepoMdstoreSizesJobNode.class);

	public void init(final int total) {

		this.current = 0;

		this.total = total;

		this.lookup = serviceLocator.getService(ISLookUpService.class);

		this.registry = serviceLocator.getService(ISRegistryService.class);

	}

	@Override

	protected String execute(final NodeToken token) throws Exception {

		final Set<String> list = dsManager.listManagedDatasourceIds();

		init(list.size());

		for (final String dsId : list) {

			log.info("Processing ds: " + dsId);

			current++;

			try {

				for (final Api<ApiParam> api : dsManager.getApis(dsId)) {

					verifyApi(dsId, api);

				}

			} catch (final Throwable e) {

				log.error("Error processing ds: " + dsId, e);

				token.getEnv().setAttribute(WorkflowsConstants.SYSTEM_ERROR + "::" + dsId, e.getMessage());

			}

		}

		return Arc.DEFAULT_ARC;

	}

	private void verifyApi(final String dsId, final Api<ApiParam> api)

			throws DsmException, ISRegistryException, IOException, ISLookUpException, MDStoreServiceException, ObjectStoreServiceException {

		for (final Document doc : listCollectionMdStores(dsId, api.getId())) {

			final String mdId = doc.valueOf("//RESOURCE_IDENTIFIER/@value");

			final int actualSize = NumberUtils.toInt(doc.valueOf("//NUMBER_OF_RECORDS"), 0);

			final int size = mdstoreDao.getMDStore(mdId).getSize();

			if (alwaysUpdate || size != actualSize) {

				log.info("  -- Updating size of api " + api.getId() + ", new value = " + size);

				updateMdStoreProfile(mdId, doc, size);

				dsManager.setLastCollectionInfo(dsId, api.getId(), mdId, size, calculateLastDate(doc));

			}

		}

		for (final Document doc : listTransformationMdStores(dsId, api.getId())) {

			final String mdId = doc.valueOf("//RESOURCE_IDENTIFIER/@value");

			final int actualSize = NumberUtils.toInt(doc.valueOf("//NUMBER_OF_RECORDS"), 0);

			final int size = mdstoreDao.getMDStore(mdId).getSize();

			if (alwaysUpdate || size != actualSize) {

				updateMdStoreProfile(mdId, doc, size);

				dsManager.setLastAggregationInfo(dsId, api.getId(), mdId, size, calculateLastDate(doc));

			}

		}

		for (final Document doc : listDownloadObjectStores(dsId, api.getId())) {

			final String objId = doc.valueOf("//RESOURCE_IDENTIFIER/@value");

			final int actualSize = NumberUtils.toInt(doc.valueOf("//STORE_SIZE"), 0);

			final int size = objectStoreDao.getObjectStore(objId).getSize();

			if (alwaysUpdate || size != actualSize) {

				updateObjStoreProfile(objId, doc, size);

				dsManager.setLastDownloadInfo(dsId, api.getId(), objId, size, calculateLastDate(doc));

			}

		}

	}

	private Date calculateLastDate(final Document doc) {

		final String dateS = doc.valueOf("//LAST_STORAGE_DATE");

		final Date date = StringUtils.isNoneBlank(dateS) ? dateUtils.parse(dateS) : new Date();

		return date;

	}

	private List<Document> listCollectionMdStores(final String dsId, final String apiId) throws ISLookUpException, IOException {

		return executeXquery("listCollectionMdStores.xquery.st", dsId, apiId);

	}

	private List<Document> listTransformationMdStores(final String dsId, final String apiId) throws ISLookUpException, IOException {

		return executeXquery("listTransformationMdStores.xquery.st", dsId, apiId);

	}

	private List<Document> listDownloadObjectStores(final String dsId, final String apiId) throws ISLookUpException, IOException {

		return executeXquery("listDownloadObjectStores.xquery.st", dsId, apiId);

	}

	private List<Document> executeXquery(final String template, final String dsId, final String apiId) throws ISLookUpException, IOException {

		final StringTemplate st = new StringTemplate(IOUtils.toString(getClass().getResourceAsStream(template)));

		st.setAttribute("dsId", dsId);

		st.setAttribute("apiId", apiId);

		final SAXReader reader = new SAXReader();

		return lookup.quickSearchProfile(st.toString())

				.stream()

				.map(s -> {

					try {

						return reader.read(new StringReader(s));

					} catch (final DocumentException e) {

						return null;

					}

				})

				.filter(Objects::nonNull)

				.collect(Collectors.toList());

	}

	private void updateMdStoreProfile(final String mdId, final Document doc, final int size) throws ISRegistryException {

		doc.selectSingleNode("//NUMBER_OF_RECORDS").setText(Integer.toString(size));

		registry.updateProfile(mdId, doc.asXML(), "MDStoreDSResourceType");

	}

	private void updateObjStoreProfile(final String objId, final Document doc, final int size) throws ISRegistryException {

		doc.selectSingleNode("//COUNT_STORE").setText(Integer.toString(size));

		doc.selectSingleNode("//STORE_SIZE").setText(Integer.toString(size));

		registry.updateProfile(objId, doc.asXML(), "ObjectStoreDSResourceType");

	}

	@Override

	public int getTotalValue() {

		return total;

	}

	@Override

	public int getCurrentValue() {

		return current;

	}

	@Override

	public boolean isInaccurate() {

		return false;

	}

	public boolean isAlwaysUpdate() {

		return alwaysUpdate;

	}

	public void setAlwaysUpdate(final boolean alwaysUpdate) {

		this.alwaysUpdate = alwaysUpdate;

	}

}