Add SQL replay checkpoint object to SQL (#868)

* Add SQL replay checkpoint object to Datastore

This will be part of the asynchronous commit-log replay to SQL. Whenever
we successfully export commits up to a particular time, we should
persist that time so we don't replay the same commits again (it is not
idempotent)

* Move SqlReplayCheckpoint from DS to SQL

* Responses to CR

core/src/main/java/google/registry/schema/replay/SqlReplayCheckpoint.java

@@ -0,0 +1,58 @@
+// Copyright 2020 The Nomulus Authors. All Rights Reserved.
+//
+// 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
+//
+//     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 google.registry.schema.replay;
+
+import static google.registry.model.common.CrossTldSingleton.SINGLETON_ID;
+import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
+import static google.registry.util.DateTimeUtils.START_OF_TIME;
+
+import com.google.common.collect.ImmutableList;
+import javax.persistence.Entity;
+import javax.persistence.Id;
+import org.joda.time.DateTime;
+
+@Entity
+public class SqlReplayCheckpoint implements SqlEntity {
+
+  // Hibernate doesn't allow us to have a converted DateTime as our primary key so we need this
+  @Id private long revisionId = SINGLETON_ID;
+
+  private DateTime lastReplayTime;
+
+  @Override
+  public ImmutableList<DatastoreEntity> toDatastoreEntities() {
+    return ImmutableList.of(); // not necessary to persist in Datastore
+  }
+
+  public static DateTime get() {
+    jpaTm().assertInTransaction();
+    return jpaTm()
+        .getEntityManager()
+        .createQuery("FROM SqlReplayCheckpoint", SqlReplayCheckpoint.class)
+        .setMaxResults(1)
+        .getResultStream()
+        .findFirst()
+        .map(checkpoint -> checkpoint.lastReplayTime)
+        .orElse(START_OF_TIME);
+  }
+
+  public static void set(DateTime lastReplayTime) {
+    jpaTm().assertInTransaction();
+    SqlReplayCheckpoint checkpoint = new SqlReplayCheckpoint();
+    checkpoint.lastReplayTime = lastReplayTime;
+    // this will overwrite the existing object due to the constant revisionId
+    jpaTm().put(checkpoint);
+  }
+}

core/src/main/resources/META-INF/persistence.xml

@@ -70,6 +70,7 @@
     <class>google.registry.persistence.transaction.TransactionEntity</class>
     <class>google.registry.schema.cursor.Cursor</class>
     <class>google.registry.schema.domain.RegistryLock</class>
+    <class>google.registry.schema.replay.SqlReplayCheckpoint</class>
     <class>google.registry.schema.server.Lock</class>
     <class>google.registry.schema.tld.PremiumEntry</class>