/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

package org.apache.streampipes.storage.rdf4j.impl;

import io.fogsy.empire.core.empire.impl.RdfQuery;
import org.apache.streampipes.model.SpDataSet;
import org.apache.streampipes.model.SpDataStream;
import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
import org.apache.streampipes.model.base.NamedStreamPipesEntity;
import org.apache.streampipes.model.graph.DataProcessorDescription;
import org.apache.streampipes.model.graph.DataSinkDescription;
import org.apache.streampipes.model.staticproperty.StaticProperty;
import org.apache.streampipes.storage.api.IPipelineElementDescriptionStorage;
import org.apache.streampipes.storage.rdf4j.sparql.QueryBuilder;
import org.apache.streampipes.storage.rdf4j.util.Transformer;
import org.eclipse.rdf4j.repository.RepositoryException;
import org.eclipse.rdf4j.rio.RDFParseException;
import org.eclipse.rdf4j.rio.UnsupportedRDFormatException;

import javax.persistence.EntityManager;
import javax.persistence.Query;
import java.io.IOException;
import java.net.URI;
import java.util.List;

public class PipelineElementStorageRequests implements IPipelineElementDescriptionStorage {

  private EntityManager entityManager;

  public PipelineElementStorageRequests(EntityManager entityManager) {
    this.entityManager = entityManager;
  }

  //TODO: exception handling

  @Override
  public boolean storeDataStream(SpDataStream sep) {
    if (exists(sep)) {
      return false;
    }
    entityManager.persist(sep);
    return true;
  }

  @Override
  public boolean storeDataStream(String jsonld) {
    SpDataStream sep;
    try {
      sep = Transformer.fromJsonLd(SpDataStream.class, jsonld);
      return storeDataStream(sep);
    } catch (RDFParseException | IOException | RepositoryException | UnsupportedRDFormatException e) {
      // TODO Auto-generated catch block
      e.printStackTrace();
    }
    return false;
  }

  @Override
  public boolean storeDataProcessor(DataProcessorDescription sepa) {
    if (existsDataProcessor(sepa.getElementId())) {
      return false;
    }
    entityManager.persist(sepa);
    return true;
  }

  @Override
  public boolean existsDataProcessor(String rdfId) {
    return entityManager.find(DataProcessorDescription.class, rdfId) != null;
  }

  @Override
  public boolean existsDataStream(String rdfId) {
    return entityManager.find(SpDataStream.class, rdfId) != null;
  }

  @Override
  public boolean existsDataSink(String rdfId) {
    return entityManager.find(DataSinkDescription.class, rdfId) != null;
  }

  @Override
  public boolean storeDataProcessor(String jsonld) {
    DataProcessorDescription sepa;
    try {
      sepa = Transformer.fromJsonLd(DataProcessorDescription.class, jsonld);
      return storeDataProcessor(sepa);
    } catch (RDFParseException | IOException | RepositoryException | UnsupportedRDFormatException e) {
      // TODO Auto-generated catch block
      e.printStackTrace();
    }
    return false;
  }

  @Override
  public SpDataStream getDataStreamById(URI rdfId) {
    return entityManager.find(SpDataStream.class, rdfId);
  }

  @Override
  public SpDataStream getDataStreamByAppId(String appId) {
    return getByAppId(getAllDataStreams(), appId);
  }

  @Override
  public SpDataStream getDataStreamById(String rdfId) {
    return getDataStreamById(URI.create(rdfId));
  }

  @SuppressWarnings("unchecked")
  @Override
  public List<SpDataStream> getAllDataStreams() {
    Query query = entityManager.createQuery(QueryBuilder.buildListDataStreamQuery());
    query.setHint(RdfQuery.HINT_ENTITY_CLASS, SpDataStream.class);
    List<SpDataStream> allStreams = query.getResultList();
    allStreams.addAll(getAllDataSets());
    return allStreams;
  }

  private List<SpDataSet> getAllDataSets() {
    Query query = entityManager.createQuery(QueryBuilder.buildListDataSetQuery());
    query.setHint(RdfQuery.HINT_ENTITY_CLASS, SpDataSet.class);
    return query.getResultList();
  }

  @SuppressWarnings("unchecked")
  @Override
  public List<DataProcessorDescription> getAllDataProcessors() {
    Query query = entityManager.createQuery(QueryBuilder.buildListSEPAQuery());
    query.setHint(RdfQuery.HINT_ENTITY_CLASS, DataProcessorDescription.class);
    return query.getResultList();
  }

  @Override
  public boolean deleteDataStream(SpDataStream stream) {
    deleteDataStream(stream.getElementId());
    return true;
  }

  @Override
  public boolean deleteDataStream(String rdfId) {
    SpDataStream sep = entityManager.find(SpDataStream.class, rdfId);
    entityManager.remove(sep);
    return true;
  }

  @Override
  public boolean deleteDataProcessor(DataProcessorDescription sepa) {
    deleteDataProcessor(sepa.getElementId());
    return true;
  }

  @Override
  public boolean deleteDataProcessor(String rdfId) {
    DataProcessorDescription sepa = entityManager.find(DataProcessorDescription.class, rdfId);
    entityManager.remove(sepa);
    return true;
  }

  @Override
  public boolean exists(SpDataStream sep) {
    SpDataStream storedSEP = entityManager.find(SpDataStream.class, sep.getElementId());
    return storedSEP != null;
  }

  @Override
  public boolean exists(DataProcessorDescription sepa) {
    DataProcessorDescription storedSEPA = entityManager.find(DataProcessorDescription.class, sepa.getElementId());
    return storedSEPA != null;
  }

  @Override
  public boolean update(SpDataStream sep) {
    return deleteDataStream(sep) && storeDataStream(sep);
  }

  @Override
  public boolean update(DataProcessorDescription sepa) {
    return deleteDataProcessor(sepa) && storeDataProcessor(sepa);
  }

  @Override
  public DataProcessorDescription getDataProcessorById(String rdfId) {
    return getDataProcessorById(URI.create(rdfId));
  }

  @Override
  public DataProcessorDescription getDataProcessorById(URI rdfId) {
    return entityManager.find(DataProcessorDescription.class, rdfId);
  }

  @Override
  public DataProcessorDescription getDataProcessorByAppId(String appId) {
    return getByAppId(getAllDataProcessors(), appId);
  }

  @Override
  public DataSinkDescription getDataSinkById(String rdfId) {
    return getDataSinkById(URI.create(rdfId));
  }

  @Override
  public DataSinkDescription getDataSinkById(URI rdfId) {
    return entityManager.find(DataSinkDescription.class, rdfId);
  }

  @Override
  public DataSinkDescription getDataSinkByAppId(String appId) {
    return getByAppId(getAllDataSinks(), appId);
  }

  @Override
  public boolean exists(DataSinkDescription sec) {
    DataSinkDescription storedSEC = entityManager.find(DataSinkDescription.class, sec.getElementId());
    return storedSEC != null;
  }

  @Override
  public boolean update(DataSinkDescription sec) {
    return deleteDataSink(sec) && storeDataSink(sec);

  }

  @Override
  public boolean deleteDataSink(DataSinkDescription sec) {
    return deleteDataSink(sec.getElementId());
  }

  @Override
  public boolean deleteDataSink(String rdfId) {
    DataSinkDescription sec = entityManager.find(DataSinkDescription.class, rdfId);
    entityManager.remove(sec);
    return true;
  }

  @Override
  public boolean storeDataSink(DataSinkDescription sec) {
    if (exists(sec)) {
      return false;
    }
    entityManager.persist(sec);
    return true;
  }

  @SuppressWarnings("unchecked")
  @Override
  public List<DataSinkDescription> getAllDataSinks() {
    Query query = entityManager.createQuery(QueryBuilder.buildListSECQuery());
    query.setHint(RdfQuery.HINT_ENTITY_CLASS, DataSinkDescription.class);
    return query.getResultList();
  }

  @Override
  public StaticProperty getStaticPropertyById(String rdfId) {
    return entityManager.find(StaticProperty.class, URI.create(rdfId));
  }

  @Override
  public boolean storeInvocablePipelineElement(InvocableStreamPipesEntity element) {
    entityManager.persist(element);
    return true;
  }

  @Override
  public SpDataStream getEventStreamById(String rdfId) {
    return entityManager.find(SpDataStream.class, URI.create(rdfId));
  }

  private <T extends NamedStreamPipesEntity> T getByAppId(List<T> elements, String appId) {
    return elements
            .stream()
            .filter(e -> e.getAppId().equals(appId))
            .findFirst()
            .orElse(null);
  }


}
