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.HAS_TIMESERIES_TUPLE;
import static de.dlr.shepard.common.util.Neo4jLabels.IS_IN_CONTAINER;
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_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.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.Cypher;

@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("timeseriesId"))
      .orderBy(ts.property("timeseriesId").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()),
        "symbolicName",
        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("timeseriesId", 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 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());
  }
}