public interface VersionHistoryDagDao {

  VersionHistoryDag create(long itemId) throws GroundException;

  VersionHistoryDag retrieveFromDatabase(long itemId) throws GroundException;

  /**
   * Add a new edge between parentId and childId in DAG.
   *
   * @param dag the DAG to update
   * @param parentId the parent's id
   * @param childId the child's id
   * @param itemId the id of the Item whose DAG we're updating
   */
  DbStatements<?> addEdge(VersionHistoryDag dag, long parentId, long childId, long itemId) throws GroundException;

  /**
   * Truncate the DAG to only have a certain number of levels, removing everything before that.
   *
   * TODO: Once we have delta-encoded tags, this should also update the associated tags.
   *
   * @param dag the DAG to truncate
   * @param numLevels the number of levels to keep
   */
  void truncate(VersionHistoryDag dag, int numLevels, Class<? extends Item> itemType) throws GroundException;
} public class GroundException extends Exception {

  private static final long serialVersionUID = 1L;

  private final String message;
  private final ExceptionType exceptionType;

  public enum ExceptionType {
    DB("Database Exception:", "%s"),
    ITEM_NOT_FOUND("GroundItemNotFoundException", "No %s \'%s\' found."),
    VERSION_NOT_FOUND("GroundVersionNotFoundException", "No %s \'%s\' found."),
    ITEM_ALREADY_EXISTS("GroundItemAlreadyExistsException", "%s %s already exists."),
    OTHER("GroundException", "%s");

    String name;
    String description;

    ExceptionType(String name, String description) {
      this.name = name;
      this.description = description;
    }

    public String format(String... values) {
      return String.format(this.description, values);
    }
  }

  public GroundException(ExceptionType exceptionType, String... values) {
    this.exceptionType = exceptionType;
    this.message = this.exceptionType.format(values);
  }

  public GroundException(Exception exception) {
    this.exceptionType = ExceptionType.OTHER;
    this.message = this.exceptionType.format(exception.getMessage());
  }

  @Override
  public String getMessage() {
    return this.message;
  }

  public ExceptionType getExceptionType() {
    return this.exceptionType;
  }
} public class Item {

  @JsonProperty("id")
  private final long id;

  @JsonProperty("tags")
  private final Map<String, Tag> tags;

  @JsonCreator
  public Item(@JsonProperty("id") long id, @JsonProperty("tags") Map<String, Tag> tags) {
    this.id = id;

    if (tags == null) {
      this.tags = new HashMap<>();
    } else {
      this.tags = tags;
    }
  }

  public long getId() {
    return this.id;
  }

  public Map<String, Tag> getTags() {
    return this.tags;
  }
} public class VersionHistoryDag {

  // the id of the Version that's at the rootId of this DAG
  private final long itemId;

  // list of VersionSuccessors that make up this DAG
  private final List<Long> edgeIds;

  // map of parents to children
  private final Map<Long, List<Long>> parentChildMap;

  /**
   * Create a new version history DAG.
   *
   * @param itemId the id of the item of this DAG
   * @param edges the version successors in this DAG
   */
  public VersionHistoryDag(long itemId, List<VersionSuccessor> edges) {
    this.itemId = itemId;
    this.edgeIds = edges.stream().map(VersionSuccessor::getId).collect(Collectors.toList());
    this.parentChildMap = new HashMap<>();

    edges.forEach(edge -> this.addToParentChildMap(edge.getFromId(), edge.getToId()));
  }

  public long getItemId() {
    return this.itemId;
  }

  public List<Long> getEdgeIds() {
    return this.edgeIds;
  }

  /**
   * Checks if a given ID is in the DAG.
   *
   * @param id the ID to be checked
   * @return true if id is in the DAG, false otherwise
   */
  public boolean checkItemInDag(long id) {
    return this.parentChildMap.keySet().contains(id) || this.getLeaves().contains(id);
  }

  /**
   * Adds an edge to this DAG.
   *
   * @param parentId the id of the "from" of the edge
   * @param childId the id of the "to" of the edge
   */
  public void addEdge(long parentId, long childId, long successorId) {
    this.edgeIds.add(successorId);
    this.addToParentChildMap(parentId, childId);
  }

  /**
   * Return the parent(s) of a particular version.
   *
   * @param childId the query id
   * @return the list of parent version(s)
   */
  public List<Long> getParent(long childId) {
    return this.parentChildMap
             .entrySet()
             .stream()
             .filter(entry -> entry.getValue().contains(childId))
             .map(Map.Entry::getKey)
             .collect(Collectors.toList());
  }

  public Map<Long, Long> getParentChildPairs() {
    Map<Long, Long> result = new HashMap<>();

    for (Long parent : this.parentChildMap.keySet()) {
      List<Long> children = this.parentChildMap.get(parent);

      for (Long child : children) {
        result.put(parent, child);
      }
    }

    return result;
  }

  /**
   * Returns the leaves of the DAG (i.e., any version id that is not a parent of another version
   * id).
   *
   * @return the list of the IDs of the leaves of this DAG
   */
  public List<Long> getLeaves() {
    Set<Long> leaves = new HashSet<>();
    this.parentChildMap.values().forEach(leaves::addAll);
    leaves.removeAll(this.parentChildMap.keySet());

    return new ArrayList<>(leaves);
  }

  private void addToParentChildMap(long parent, long child) {
    List<Long> childList = this.parentChildMap.computeIfAbsent(parent, key -> new ArrayList<>());
    childList.add(child);
  }
} public class VersionSuccessor {

  private final long id;

  private final long fromId;

  private final long toId;

  /**
   * Create a version successor.
   *
   * @param id the id of the successor
   * @param fromId the source id
   * @param toId the destination id
   */
  public VersionSuccessor(long id, long fromId, long toId) {
    this.id = id;
    this.fromId = fromId;
    this.toId = toId;
  }

  public long getId() {
    return this.id;
  }

  public long getFromId() {
    return this.fromId;
  }

  public long getToId() {
    return this.toId;
  }
} @Singleton
public class IdGenerator {

  private final long prefix;
  private long versionCounter;
  private long successorCounter;
  private long itemCounter;

  // If true, only one counter will be used. If false, all three counters will be used.
  private final boolean globallyUnique;

  public IdGenerator() {
    this.prefix = 0;
    this.versionCounter = 1;
    this.successorCounter = 1;
    this.itemCounter = 1;
    this.globallyUnique = true;
  }

  /**
   * Create a unique id generator.
   *
   * @param machineId the id of this machine
   * @param numMachines the total number of machines
   * @param globallyUnique if true, only one counter will be used for all version
   */
  public IdGenerator(long machineId, long numMachines, boolean globallyUnique) {
    long machineBits = 1;
    long fence = 2;

    while (fence < numMachines) {
      fence = fence * 2;
      machineBits++;
    }

    this.prefix = machineId << (64 - machineBits);

    // NOTE: Do not change this. The version counter is set to start a 1 because 0 is the default
    // empty version.
    this.versionCounter = 1;
    this.successorCounter = 1;
    this.itemCounter = 1;

    this.globallyUnique = globallyUnique;
  }

  public synchronized long generateVersionId() {
    return prefix | this.versionCounter++;
  }

  /**
   * Generate an id for version successors.
   *
   * @return a new id
   */
  public synchronized long generateSuccessorId() {
    if (this.globallyUnique) {
      return prefix | this.versionCounter++;
    } else {
      return prefix | this.successorCounter++;
    }
  }

  /**
   * Generate an id for items.
   *
   * @return a new id
   */
  public synchronized long generateItemId() {
    if (this.globallyUnique) {
      return prefix | this.versionCounter++;
    } else {
      return prefix | this.itemCounter++;
    }
  }
} public class SqlConstants {

  /* General insert statements */
  public static final String INSERT_GENERIC_ITEM_WITH_NAME = "INSERT INTO %s (item_id, source_key, name) VALUES (%d, \'%s\', \'%s\');";
  public static final String INSERT_GENERIC_ITEM_WITHOUT_NAME = "INSERT INTO %s (item_id, source_key, name) VALUES (%d, \'%s\', null);";

  /* General select statements */
  public static final String SELECT_STAR_BY_SOURCE_KEY = "SELECT * FROM %s WHERE source_key = \'%s\'";
  public static final String SELECT_STAR_ITEM_BY_ID = "SELECT * FROM %s WHERE item_id = %d;";
  public static final String SELECT_STAR_BY_ID = "SELECT * FROM %s WHERE id = %d;";
  public static final String DELETE_BY_ID = "DELETE FROM %s WHERE id = %d";

  /* Version-specific statements */
  public static final String INSERT_VERSION = "INSERT INTO version (id) VALUES (%d);";

  /* Version Successor-specific statements */
  public static final String INSERT_VERSION_SUCCESSOR = "INSERT INTO version_successor (id, from_version_id, to_version_id) VALUES (%d, %d, %d);";
  public static final String SELECT_VERSION_SUCCESSOR = "SELECT * FROM version_successor where id = %d;";
  public static final String SELECT_VERSION_SUCCESSOR_BY_ENDPOINT = "SELECT * FROM version_successor WHERE to_version_id = %d;";
  public static final String DELETE_VERSION_SUCCESSOR = "DELETE FROM version_successor WHERE id = %d;";

  /* Version History DAG-specific statements */
  public static final String INSERT_VERSION_HISTORY_DAG_EDGE = "INSERT INTO version_history_dag (item_id, version_successor_id) VALUES (%d, %d);";
  public static final String SELECT_VERSION_HISTORY_DAG = "SELECT * FROM version_history_dag WHERE item_id = %d;";
  public static final String DELETE_SUCCESSOR_FROM_DAG = "DELETE FROM version_history_dag WHERE version_successor_id = %d;";

  /* Item-specific statements */
  public static final String INSERT_ITEM = "INSERT INTO ITEM (id) VALUES (%d);";
  public static final String INSERT_ITEM_TAG_WITH_VALUE =
    "INSERT INTO item_tag (item_id, key, value, type) VALUES (%d, " + "\'%s\', \'%s\', \'%s\');";
  public static final String INSERT_ITEM_TAG_NO_VALUE = "INSERT INTO item_tag (item_id, key, value, type) VALUES (%d, \'%s\', null, null);";
  public static final String SELECT_ITEM_TAGS = "SELECT * FROM item_tag WHERE item_id = %d;";
  public static final String SELECT_ITEM_TAGS_BY_KEY = "SELECT * FROM item_tag WHERE key = \'%s\';";

  /* Edge-specific statements */
  public static final String INSERT_EDGE_WITH_NAME =
    "INSERT INTO edge (item_id, source_key, from_node_id, to_node_id, name) VALUES (%d, \'%s\', %d, %d, \'%s\');";
  public static final String INSERT_EDGE_WITHOUT_NAME =
    "INSERT INTO edge (item_id, source_key, from_node_id, to_node_id, name) VALUES (%d, \'%s\', %d, %d, null);";
  public static final String INSERT_EDGE_VERSION = "INSERT INTO edge_version (id, edge_id, from_node_version_start_id, from_node_version_end_id, "
                                                     + "to_node_version_start_id, to_node_version_end_id) VALUES (%d, %d, %d, %d, %d, %d);";
  public static final String UPDATE_EDGE_VERSION = "UPDATE edge_version SET from_node_version_end_id = %d, to_node_version_end_id = %d WHERE id = "
                                                     + "%d;";

  /* Graph-specific statements */
  public static final String INSERT_GRAPH_VERSION = "INSERT INTO graph_version (id, graph_id) VALUES (%d, %d);";
  public static final String INSERT_GRAPH_VERSION_EDGE = "INSERT INTO graph_version_edge (graph_version_id, edge_version_id) VALUES (%d, %d);";
  public static final String SELECT_GRAPH_VERSION_EDGES = "SELECT * FROM graph_version_edge WHERE graph_version_id = %d;";
  public static final String DELETE_ALL_GRAPH_VERSION_EDGES = "DELETE FROM %s WHERE %s_version_id = %d";

  /* Node-specific statements */
  public static final String INSERT_NODE_VERSION = "INSERT INTO node_version (id, node_id) VALUES (%d, %d);";
  public static final String SELECT_NODE_VERSION_ADJACENT_LINEAGE = "SELECT * FROM lineage_edge_version WHERE from_rich_version_id = %d;";

  /* Rich Version-specific statements */
  public static final String INSERT_RICH_VERSION_WITH_REFERENCE = "INSERT INTO rich_version (id, structure_version_id, reference) VALUES (%d, %d, "
                                                                    + "\'%s\');";
  public static final String INSERT_RICH_VERSION_WITHOUT_REFERENCE = "INSERT INTO rich_version (id, structure_version_id, reference) VALUES (%d, %d, "
                                                                       + "null);";
  public static final String INSERT_RICH_VERSION_TAG_WITH_VALUE = "INSERT INTO rich_version_tag (rich_version_id, key, value, type) VALUES (%d, "
                                                                    + "\'%s\', \'%s\', \'%s\');";
  public static final String INSERT_RICH_VERSION_TAG_NO_VALUE = "INSERT INTO rich_version_tag (rich_version_id, key, value, type) VALUES (%d, "
                                                                  + "\'%s\', null, null);";
  public static final String INSERT_RICH_VERSION_EXTERNAL_PARAMETER = "INSERT INTO rich_version_external_parameter (rich_version_id, key, value) "
                                                                        + "VALUES (%d, \'%s\', \'%s\');";
  public static final String SELECT_RICH_VERSION_EXTERNAL_PARAMETERS = "SELECT * FROM rich_version_external_parameter WHERE rich_version_id = %d;";
  public static final String SELECT_RICH_VERSION_TAGS = "SELECT * FROM rich_version_tag WHERE rich_version_id = %d;";
  public static final String SELECT_RICH_VERSION_TAGS_BY_KEY = "SELECT * FROM rich_version_tag WHERE key = \'%s\';";
  public static final String DELETE_RICH_VERSION_TAGS = "DELETE FROM rich_version_tag WHERE rich_version_id = %d";
  public static final String DELETE_RICH_EXTERNAL_PARAMETERS = "DELETE FROM rich_version_external_parameter WHERE rich_version_id = %d";

  /* Structure-specific statements */
  public static final String INSERT_STRUCTURE_VERSION = "INSERT INTO structure_version (id, structure_id) VALUES (%d, %d);";
  public static final String INSERT_STRUCTURE_VERSION_ATTRIBUTE = "INSERT INTO structure_version_attribute (structure_version_id, key, type) "
                                                                    + "VALUES (%d, \'%s\', \'%s\');";
  public static final String SELECT_STRUCTURE_VERSION_ATTRIBUTES = "SELECT * FROM structure_version_attribute WHERE structure_version_id = %d;";
  public static final String DELETE_STRUCTURE_VERSION_ATTRIBUTES = "DELETE FROM structure_version_attribute WHERE structure_version_id = %d";

  /* Lineage Edge-specific statements */
  public static final String INSERT_LINEAGE_EDGE_VERSION = "INSERT INTO lineage_edge_version (id, lineage_edge_id, from_rich_version_id, "
                                                             + "to_rich_version_id, principal_id) VALUES (%d, %d, %d, %d, %d);";

  /* Lineage Graph-specific statements */
  public static final String INSERT_LINEAGE_GRAPH_VERSION = "INSERT INTO lineage_graph_version (id, lineage_graph_id) VALUES (%d, %d);";
  public static final String INSERT_LINEAGE_GRAPH_VERSION_EDGE = "INSERT INTO lineage_graph_version_edge (lineage_graph_version_id, "
                                                                   + "lineage_edge_version_id) VALUES (%d, %d);";
  public static final String SELECT_LINEAGE_GRAPH_VERSION_EDGES = "SELECT * FROM lineage_graph_version_edge WHERE lineage_graph_version_id = %d;";
} public final class GroundUtils {

  private GroundUtils() {
  }

  public static Result handleException(Throwable e, Request request) {
    if (e.getCause().getCause() instanceof GroundException) {
      return badRequest(GroundUtils.getClientError(request, e.getCause().getCause(), ExceptionType.ITEM_NOT_FOUND));
    } else {
      return internalServerError(GroundUtils.getServerError(request, e.getCause().getCause()));
    }
  }

  private static ObjectNode getServerError(final Request request, final Throwable e) {
    Logger.error("Error! Request Path: {}\nError Message: {}\n Stack Trace: {}", request.path(), e.getMessage(), e.getStackTrace());

    ObjectNode result = Json.newObject();
    result.put("Error", "Unexpected error while processing request.");
    result.put("Request Path", request.path());

    StringWriter sw = new StringWriter();
    e.printStackTrace(new PrintWriter(sw));
    result.put("Stack Trace", sw.toString());
    return result;
  }

  private static ObjectNode getClientError(final Request request, final Throwable e, ExceptionType type) {
    Logger.error("Error! Request Path: {}\nError Message: {}\n Stack Trace: {}", request.path(), e.getMessage(), e.getStackTrace());

    ObjectNode result = Json.newObject();

    result.put("Error", String.format("The request to %s was invalid.", request.path()));
    result.put("Message", String.format("%s", e.getMessage()));

    return result;
  }

  static String listToJson(final List<Map<String, Object>> objList) {
    try {
      return new ObjectMapper().writeValueAsString(objList);
    } catch (IOException e) {
      throw new RuntimeException("ERROR : listToJson Converting List to JSON." + e.getMessage(), e);
    }
  }

  public static List<Long> getListFromJson(JsonNode jsonNode, String fieldName) {
    List<Long> parents = new ArrayList<>();
    JsonNode listNode = jsonNode.get(fieldName);

    if (listNode != null) {
      listNode.forEach(node -> parents.add(node.asLong()));
    }

    return parents;
  }

  public static VersionDao<?> getVersionDaoFromItemType(Class<?> klass, Database dbSource, IdGenerator idGenerator) throws GroundException {
    if (klass.equals(Node.class)) {
      return new PostgresNodeVersionDao(dbSource, idGenerator);
    } else if (klass.equals(Edge.class)) {
      return new PostgresEdgeVersionDao(dbSource, idGenerator);
    } else if (klass.equals(Graph.class)) {
      return new PostgresGraphVersionDao(dbSource, idGenerator);
    } else if (klass.equals(Structure.class)) {
      return new PostgresStructureVersionDao(dbSource, idGenerator);
    } else if (klass.equals(LineageEdge.class)) {
      return new PostgresLineageEdgeVersionDao(dbSource, idGenerator);
    } else if (klass.equals(LineageGraph.class)) {
      return new PostgresLineageGraphVersionDao(dbSource, idGenerator);
    } else {
      throw new GroundException(ExceptionType.OTHER, String.format("Unknown class :%s.", klass.getSimpleName()));
    }
  }
} public class PostgresStatements implements DbStatements<String> {

  List<String> statements;

  public PostgresStatements() {
    this.statements = new ArrayList<>();
  }

  public PostgresStatements(List<String> statements) {
    this.statements = statements;
  }

  @Override
  public void append(String statement) {
    this.statements.add(statement);
  }

  @Override
  public void merge(DbStatements other) {
    this.statements.addAll(other.getAllStatements());
  }

  @Override
  public List<String> getAllStatements() {
    return this.statements;
  }
} public final class PostgresUtils {

  private PostgresUtils() {
  }

  public static Executor getDbSourceHttpContext(final ActorSystem actorSystem) {
    return HttpExecution.fromThread((Executor) actorSystem.dispatchers().lookup("ground.db.context"));
  }

  public static String executeQueryToJson(Database dbSource, String sql) throws GroundException {
    Logger.debug("executeQueryToJson: {}", sql);

    try {
      Connection con = dbSource.getConnection();
      Statement stmt = con.createStatement();

      final ResultSet resultSet = stmt.executeQuery(sql);
      final long columnCount = resultSet.getMetaData().getColumnCount();
      final List<Map<String, Object>> objList = new ArrayList<>();

      while (resultSet.next()) {
        final Map<String, Object> rowData = new HashMap<>();

        for (int column = 1; column <= columnCount; column++) {
          String key = CaseFormat.LOWER_UNDERSCORE.to(CaseFormat.LOWER_CAMEL, resultSet.getMetaData().getColumnLabel(column));
          rowData.put(key, resultSet.getObject(column));
        }

        objList.add(rowData);
      }

      stmt.close();
      con.close();
      return GroundUtils.listToJson(objList);
    } catch (SQLException e) {
      Logger.error("ERROR:  executeQueryToJson  SQL : {} Message: {} Trace: {}", sql, e.getMessage(), e.getStackTrace());
      throw new GroundException(e);
    }
  }

  public static void executeSqlList(final Database dbSource, final PostgresStatements statements) throws GroundException {
    try {
      Connection con = dbSource.getConnection();
      con.setAutoCommit(false);
      Statement stmt = con.createStatement();

      for (final String sql : statements.getAllStatements()) {
        Logger.debug("executeSqlList sql : {}", sql);

        try {
          stmt.execute(sql);
        } catch (final SQLException e) {
          con.rollback();
          Logger.error("error:  Message: {} Trace: {}", e.getMessage(), e.getStackTrace());

          throw new GroundException(e);
        }
      }

      stmt.close();
      con.commit();
      con.close();
    } catch (SQLException e) {
      Logger.error("error:  executeSqlList SQL : {} Message: {} Trace: {}", statements.getAllStatements(), e.getMessage(), e.getStackTrace());

      throw new GroundException(e);
    }
  }
}
import play.db.Database;
import edu.berkeley.ground.common.dao.version.VersionHistoryDagDao;
import edu.berkeley.ground.common.exception.GroundException;
import edu.berkeley.ground.common.model.version.Item;
import edu.berkeley.ground.common.model.version.VersionHistoryDag;
import edu.berkeley.ground.common.model.version.VersionSuccessor;
import edu.berkeley.ground.common.util.IdGenerator;
import edu.berkeley.ground.postgres.dao.SqlConstants;
import edu.berkeley.ground.postgres.util.GroundUtils;
import edu.berkeley.ground.postgres.util.PostgresStatements;
import edu.berkeley.ground.postgres.util.PostgresUtils;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
/**
 * Licensed 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
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * 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 edu.berkeley.ground.postgres.dao.version;


public class PostgresVersionHistoryDagDao implements VersionHistoryDagDao {

  private PostgresVersionSuccessorDao postgresVersionSuccessorDao;
  private Database dbSource;
  private IdGenerator idGenerator;

  public PostgresVersionHistoryDagDao(Database dbSource, IdGenerator idGenerator) {
    this.dbSource = dbSource;
    this.idGenerator = idGenerator;

    this.postgresVersionSuccessorDao = new PostgresVersionSuccessorDao(this.dbSource, this.idGenerator);
  }

  @Override
  public VersionHistoryDag create(long itemId) throws GroundException {
    return new VersionHistoryDag(itemId, new ArrayList<>());
  }

  /**
   * Retrieve a DAG from the database.
   *
   * @param itemId the id of the item whose dag we are retrieving
   * @return the retrieved DAG
   * @throws GroundException an error retrieving the DAG
   */
  @Override
  public VersionHistoryDag retrieveFromDatabase(long itemId) throws GroundException {
    String sql = String.format(SqlConstants.SELECT_VERSION_HISTORY_DAG, itemId);

    List<VersionSuccessor> edges = new ArrayList<>();
    try (Connection con = dbSource.getConnection()) {

      Statement stmt = con.createStatement();
      final ResultSet resultSet = stmt.executeQuery(sql);

      List<Long> successors = new ArrayList<>();
      while (resultSet.next()) {
        successors.add(resultSet.getLong("version_successor_id"));
      }

      stmt.close();
      con.close();

      for (Long versionSuccessorId : successors) {
        VersionSuccessor versionSuccessor = postgresVersionSuccessorDao.retrieveFromDatabase(versionSuccessorId);
        edges.add(versionSuccessor);
      }
    } catch (Exception e) {
      throw new GroundException(e);
    }
    return new VersionHistoryDag(itemId, edges);
  }

  /**
   * Add an edge to the DAG.
   *
   * @param dag the DAG to update
   * @param parentId the parent's id
   * @param childId the child's id
   * @param itemId the id of the Item whose DAG we're updating
   * @throws GroundException an error adding the edge
   */
  @Override
  public PostgresStatements addEdge(VersionHistoryDag dag, long parentId, long childId, long itemId) throws GroundException {