TimeseriesDAO.java

package de.dlr.shepard.data.timeseries.daos;

import static de.dlr.shepard.common.util.CypherDslHelper.internalIdIs;
import static de.dlr.shepard.common.util.CypherDslHelper.notDeleted;
import static de.dlr.shepard.common.util.Neo4jLabels.DELETED;
import static de.dlr.shepard.common.util.Neo4jLabels.DEVICE;
import static de.dlr.shepard.common.util.Neo4jLabels.FIELD;
import static de.dlr.shepard.common.util.Neo4jLabels.HAS_TIMESERIES_TUPLE;
import static de.dlr.shepard.common.util.Neo4jLabels.IS_IN_CONTAINER;
import static de.dlr.shepard.common.util.Neo4jLabels.LOCATION;
import static de.dlr.shepard.common.util.Neo4jLabels.MEASUREMENT;
import static de.dlr.shepard.common.util.Neo4jLabels.SYMBOLIC_NAME;
import static de.dlr.shepard.common.util.Neo4jLabels.TIMESERIES;
import static de.dlr.shepard.common.util.Neo4jLabels.TIMESERIES_CONTAINER;
import static de.dlr.shepard.common.util.Neo4jLabels.TIMESERIES_ID;
import static de.dlr.shepard.common.util.Neo4jLabels.TIMESERIES_TUPLE;
import static org.neo4j.cypherdsl.core.Cypher.match;
import static org.neo4j.cypherdsl.core.Cypher.node;

import de.dlr.shepard.common.neo4j.daos.GenericDAO;
import de.dlr.shepard.common.search.query.TimeseriesInContainerQueryBuilder;
import de.dlr.shepard.common.util.CypherDslHelper;
import de.dlr.shepard.data.timeseries.model.Timeseries;
import de.dlr.shepard.data.timeseries.model.TimeseriesContainer;
import de.dlr.shepard.data.timeseries.model.TimeseriesTuple;
import jakarta.enterprise.context.RequestScoped;
import java.util.Collections;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Optional;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;
import org.neo4j.cypherdsl.core.Condition;
import org.neo4j.cypherdsl.core.Cypher;
import org.neo4j.cypherdsl.core.Node;
import org.neo4j.cypherdsl.core.Relationship;

@RequestScoped
public class TimeseriesDAO extends GenericDAO<Timeseries> {

  @Override
  public Class<Timeseries> getEntityType() {
    return Timeseries.class;
  }

  /**
   * Run a custom Cypher query to get a Stream of Timeseries including
   *
   * @param query The cypher query.
   *              It needs to return a TimeseriesContainer named "tsc", a Timeseries "ts" and a TimeseriesTuple named "tst".
   *              The Objects are required to be node entities known to neo4j ogm.
   * @return Stream of Timeseries with related Objects.
   */
  private Stream<Timeseries> queryCypherWithRelations(String query) {
    var result = session.query(query, Map.of(), true);
    return StreamSupport.stream(result.spliterator(), false).map(resultEntry -> {
      Timeseries tsObj = (Timeseries) resultEntry.get("ts");
      TimeseriesContainer tscObj = (TimeseriesContainer) resultEntry.get("tsc");
      TimeseriesTuple tsTupleObj = (TimeseriesTuple) resultEntry.get("tst");
      tsObj.setContainer(tscObj);
      tsObj.setTimeseriesTuple(tsTupleObj);
      return tsObj;
    });
  }

  public Stream<Timeseries> getAllTimeseriesInContainer(long containerId) {
    var tsc = node(TIMESERIES_CONTAINER).named("tsc");
    var ts = node(TIMESERIES).named("ts");
    var tsTuple = node(TIMESERIES_TUPLE).named("tst");
    var isInContainer = ts.relationshipTo(tsc, IS_IN_CONTAINER);
    var hasTuple = ts.relationshipTo(tsTuple, HAS_TIMESERIES_TUPLE);
    var query = match(isInContainer, hasTuple)
      .where(internalIdIs(tsc, containerId).and(notDeleted(ts)))
      .returning(ts, tsc, tsTuple)
      .build()
      .getCypher();
    return queryCypherWithRelations(query);
  }

  public long getCurrentMaximumTimeseriesId() {
    var ts = node(TIMESERIES);
    var query = match(ts)
      .returning(ts.property(TIMESERIES_ID))
      .orderBy(ts.property(TIMESERIES_ID).descending())
      .limit(1)
      .build()
      .getCypher();
    try {
      return session.query(Long.class, query, Collections.emptyMap()).iterator().next();
    } catch (NoSuchElementException e) {
      // If no Timeseries is found we can assume a "fresh" database and the timeseries IDs can start anew.
      return 0;
    }
  }

  public Optional<Timeseries> findTimeseries(long containerId, TimeseriesTuple tsTuple) {
    var tsTupleNode = node(TIMESERIES_TUPLE)
      .withProperties(
        MEASUREMENT,
        Cypher.literalOf(tsTuple.getMeasurement()),
        DEVICE,
        Cypher.literalOf(tsTuple.getDevice()),
        LOCATION,
        Cypher.literalOf(tsTuple.getLocation()),
        SYMBOLIC_NAME,
        Cypher.literalOf(tsTuple.getSymbolicName()),
        FIELD,
        Cypher.literalOf(tsTuple.getField())
      )
      .named("tst");
    var tsc = node(TIMESERIES_CONTAINER).named("tsc");
    var ts = node(TIMESERIES).named("ts");
    var query = match(tsTupleNode.relationshipFrom(ts, HAS_TIMESERIES_TUPLE).relationshipTo(tsc, IS_IN_CONTAINER))
      .where(internalIdIs(tsc, containerId).and(notDeleted(ts)))
      .returning(tsTupleNode, tsc, ts)
      .build()
      .getCypher();
    return queryCypherWithRelations(query).findFirst();
  }

  public Optional<Timeseries> findByTimeseriesId(long timeseriesId) {
    var ts = node(TIMESERIES).withProperties(TIMESERIES_ID, Cypher.literalOf(timeseriesId));
    var related = Cypher.anyNode();
    var rels = ts.relationshipTo(related);
    var query = match(ts, rels, related).where(notDeleted(ts)).returning(ts, rels, related).build().getCypher();
    return this.findByQuery(query).findFirst();
  }

  public Stream<Timeseries> findByQuintupleInContainer(
    long containerId,
    String measurement,
    String device,
    String location,
    String symbolicName,
    String field
  ) {
    Node container = Cypher.node(TIMESERIES_CONTAINER).named("tsc");
    Node timeseries = Cypher.node(TIMESERIES).named("ts");
    Node timeseriesTuple = Cypher.node(TIMESERIES_TUPLE).named("tsp");
    Node neighbor = Cypher.anyNode().named("nb");
    Node timeseriesDummy = Cypher.node(TIMESERIES).named("tsd");
    Relationship neighborhood = timeseriesDummy.relationshipBetween(neighbor).named("nh");
    Relationship hasTuple = timeseries.relationshipTo(timeseriesTuple, HAS_TIMESERIES_TUPLE).named("htt");
    Relationship isInContainer = timeseries.relationshipTo(container, IS_IN_CONTAINER).named("iic");
    Condition wherePart = CypherDslHelper.notDeleted(timeseries).and(
      CypherDslHelper.isNotDeletedWithInternalId(container, containerId)
    );
    if (measurement != null) wherePart = wherePart.and(
      TimeseriesInContainerQueryBuilder.hasStringProperty(timeseriesTuple, MEASUREMENT, measurement, "eq")
    );
    if (device != null) wherePart = wherePart.and(
      TimeseriesInContainerQueryBuilder.hasStringProperty(timeseriesTuple, DEVICE, device, "eq")
    );
    if (location != null) wherePart = wherePart.and(
      TimeseriesInContainerQueryBuilder.hasStringProperty(timeseriesTuple, LOCATION, location, "eq")
    );
    if (symbolicName != null) wherePart = wherePart.and(
      TimeseriesInContainerQueryBuilder.hasStringProperty(timeseriesTuple, SYMBOLIC_NAME, symbolicName, "eq")
    );
    if (field != null) wherePart = wherePart.and(
      TimeseriesInContainerQueryBuilder.hasStringProperty(timeseriesTuple, FIELD, field, "eq")
    );
    String query = Cypher.match(timeseries, container, timeseriesTuple, hasTuple, isInContainer)
      .where(wherePart)
      .optionalMatch(neighborhood)
      .where(timeseriesDummy.internalId().eq(timeseries.internalId()))
      .returning(timeseries, container, timeseriesTuple, neighborhood, hasTuple, isInContainer, neighbor)
      .build()
      .getCypher();
    return findByQuery(query);
  }

  public void deleteAllTimeseriesInContainer(long containerId) {
    var ts = node(TIMESERIES);
    var tsc = node(TIMESERIES_CONTAINER);
    var query = match(ts.relationshipTo(tsc, IS_IN_CONTAINER))
      .where(internalIdIs(tsc, containerId))
      .set(ts.property(DELETED), Cypher.literalOf(true))
      .build()
      .getCypher();
    session.query(query, Collections.emptyMap());
  }
}