Add high_watermark data to metadata db table

- high_watermark_epoch and high_watermark_slot
- CRUD operations in MetadataDao
 - find
 - update (assumes GVR metadata is already inserted)
 - delete

slashing-protection/src/main/java/tech/pegasys/web3signer/slashingprotection/dao/DatabaseVersionDao.java

@@ -15,7 +15,7 @@
 import org.jdbi.v3.core.Handle;
 
 public class DatabaseVersionDao {
-  public static final int EXPECTED_DATABASE_VERSION = 11;
+  public static final int EXPECTED_DATABASE_VERSION = 12;
   public static final int VALIDATOR_ENABLE_FLAG_VERSION = 10;
 
   public Integer findDatabaseVersion(final Handle handle) {

slashing-protection/src/main/java/tech/pegasys/web3signer/slashingprotection/dao/HighWatermark.java

@@ -0,0 +1,60 @@
+/*
+ * Copyright 2023 ConsenSys AG.
+ *
+ * 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 tech.pegasys.web3signer.slashingprotection.dao;
+
+import java.util.Objects;
+
+import org.apache.tuweni.units.bigints.UInt64;
+
+public class HighWatermark {
+
+  private UInt64 slot;
+  private UInt64 epoch;
+
+  // needed for JDBI
+  public HighWatermark() {}
+
+  public HighWatermark(final UInt64 slot, final UInt64 epoch) {
+    this.slot = slot;
+    this.epoch = epoch;
+  }
+
+  public UInt64 getSlot() {
+    return slot;
+  }
+
+  public UInt64 getEpoch() {
+    return epoch;
+  }
+
+  public void setSlot(final UInt64 slot) {
+    this.slot = slot;
+  }
+
+  public void setEpoch(final UInt64 epoch) {
+    this.epoch = epoch;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) return true;
+    if (o == null || getClass() != o.getClass()) return false;
+    HighWatermark that = (HighWatermark) o;
+    return Objects.equals(slot, that.slot) && Objects.equals(epoch, that.epoch);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(slot, epoch);
+  }
+}

slashing-protection/src/main/java/tech/pegasys/web3signer/slashingprotection/dao/MetadataDao.java

@@ -36,4 +36,32 @@ public void insertGenesisValidatorsRoot(
         .bind(1, genesisValidatorsRoot)
         .execute();
   }
+
+  public Optional<HighWatermark> findHighWatermark(Handle handle) {
+    return handle
+        .createQuery(
+            "SELECT high_watermark_epoch as epoch, high_watermark_slot as slot FROM metadata WHERE id = ?")
+        .bind(0, METADATA_ROW_ID)
+        .mapToBean(HighWatermark.class)
+        .filter(h -> h.getEpoch() != null && h.getSlot() != null)
+        .findFirst();
+  }
+
+  public int updateHighWatermark(final Handle handle, final HighWatermark highWatermark) {
+    return handle
+        .createUpdate(
+            "UPDATE metadata set high_watermark_epoch=:epoch, high_watermark_slot=:slot WHERE id =:id")
+        .bind("id", METADATA_ROW_ID)
+        .bind("epoch", highWatermark.getEpoch())
+        .bind("slot", highWatermark.getSlot())
+        .execute();
+  }
+
+  public void deleteHighWatermark(final Handle handle) {
+    handle
+        .createUpdate(
+            "UPDATE metadata set high_watermark_epoch=NULL, high_watermark_slot=NULL WHERE id =:id")
+        .bind("id", METADATA_ROW_ID)
+        .execute();
+  }
 }

slashing-protection/src/main/resources/migrations/postgresql/V00012__add_highwatermark_metadata.sql

@@ -0,0 +1,5 @@
+ALTER TABLE metadata
+    ADD COLUMN high_watermark_epoch NUMERIC(20),
+    ADD COLUMN high_watermark_slot NUMERIC(20);
+
+UPDATE database_version SET version = 12 WHERE id = 1;

slashing-protection/src/test/java/tech/pegasys/web3signer/slashingprotection/dao/MetadataDaoTest.java

@@ -21,6 +21,7 @@
 import db.DatabaseSetupExtension;
 import org.apache.tuweni.bytes.Bytes;
 import org.apache.tuweni.bytes.Bytes32;
+import org.apache.tuweni.units.bigints.UInt64;
 import org.jdbi.v3.core.Handle;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -66,10 +67,99 @@ public void failsInsertingMultipleGvrIntoDb(final Handle handle) {
         .hasMessageContaining("duplicate key value violates unique constraint");
   }
 
+  @Test
+  public void findsExistingHighWatermark(final Handle handle) {
+    insertGvr(handle, Bytes32.leftPad(Bytes.of(3)));
+    updateHighWatermark(handle, 1, 2);
+
+    final Optional<HighWatermark> existingHighWatermark = metadataDao.findHighWatermark(handle);
+
+    assertThat(existingHighWatermark).isNotEmpty();
+    assertThat(existingHighWatermark)
+        .contains(new HighWatermark(UInt64.valueOf(2), UInt64.valueOf(1)));
+  }
+
+  @Test
+  public void returnsEmptyForNonExistingHighWatermark(final Handle handle) {
+    assertThat(metadataDao.findHighWatermark(handle)).isEmpty();
+  }
+
+  @Test
+  public void returnsEmptyForNonExistingHighWatermarkWhenGvrSet(final Handle handle) {
+    insertGvr(handle, Bytes32.leftPad(Bytes.of(3)));
+    assertThat(metadataDao.findHighWatermark(handle)).isEmpty();
+  }
+
+  @Test
+  public void insertsHighWatermark(final Handle handle) {
+    insertGvr(handle, Bytes32.leftPad(Bytes.of(3)));
+    HighWatermark highWatermark = new HighWatermark(UInt64.valueOf(2), UInt64.valueOf(1));
+
+    int updateCount = metadataDao.updateHighWatermark(handle, highWatermark);
+
+    assertThat(updateCount).isEqualTo(1);
+    final List<HighWatermark> highWatermarks =
+        handle
+            .createQuery(
+                "SELECT high_watermark_epoch as epoch, high_watermark_slot as slot FROM metadata")
+            .mapToBean(HighWatermark.class)
+            .list();
+    assertThat(highWatermarks.size()).isEqualTo(1);
+    assertThat(highWatermarks.get(0)).isEqualTo(highWatermark);
+  }
+
+  @Test
+  public void updatesHighWatermark(final Handle handle) {
+    insertGvr(handle, Bytes32.leftPad(Bytes.of(3)));
+    updateHighWatermark(handle, 1, 2);
+    HighWatermark highWatermark = createHighWatermark(3, 3);
+
+    int updateCount = metadataDao.updateHighWatermark(handle, highWatermark);
+
+    assertThat(updateCount).isEqualTo(1);
+    final List<HighWatermark> highWatermarks =
+        handle
+            .createQuery(
+                "SELECT high_watermark_epoch as epoch, high_watermark_slot as slot FROM metadata")
+            .mapToBean(HighWatermark.class)
+            .list();
+    assertThat(highWatermarks.size()).isEqualTo(1);
+    assertThat(highWatermarks.get(0)).isEqualTo(highWatermark);
+  }
+
+  @Test
+  public void updateHighWatermarkWhenNoGvrHasNoEffect(final Handle handle) {
+    int updateCount = metadataDao.updateHighWatermark(handle, createHighWatermark(1, 1));
+    assertThat(updateCount).isEqualTo(0);
+  }
+
+  @Test
+  public void deletesHighWatermark(final Handle handle) {
+    insertGvr(handle, Bytes32.leftPad(Bytes.of(3)));
+    updateHighWatermark(handle, 2, 2);
+    assertThat(metadataDao.findHighWatermark(handle)).isNotEmpty();
+
+    metadataDao.deleteHighWatermark(handle);
+
+    assertThat(metadataDao.findHighWatermark(handle)).isEmpty();
+  }
+
   private void insertGvr(final Handle handle, final Bytes genesisValidatorsRoot) {
     handle.execute(
         "INSERT INTO metadata (id, genesis_validators_root) VALUES (?, ?)",
         1,
         genesisValidatorsRoot);
   }
+
+  private void updateHighWatermark(final Handle handle, final int epoch, final int slot) {
+    handle
+        .createUpdate("UPDATE metadata set high_watermark_epoch=:epoch, high_watermark_slot=:slot")
+        .bind("epoch", epoch)
+        .bind("slot", slot)
+        .execute();
+  }
+
+  private HighWatermark createHighWatermark(final int epoch, final int slot) {
+    return new HighWatermark(UInt64.valueOf(epoch), UInt64.valueOf(slot));
+  }
 }