package uk.gov;

import org.postgresql.util.PGobject;

import java.nio.charset.Charset;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;

class DestinationPostgresDB extends PostgresDB {
    private final String indexedEntriesTableName;
    private final String waterMarkTableName;

    public DestinationPostgresDB(String connectionString) throws SQLException {
        super(connectionString);
        this.indexedEntriesTableName = "ordered_entry_index";
        this.waterMarkTableName = "streamed_entries";
    }

    public void write(ResultSet resultSet) throws SQLException {
        while (resultSet.next()) {
            try (PreparedStatement statement = connection.prepareStatement("INSERT INTO " + indexedEntriesTableName + "(ENTRY) VALUES(?)")) {
                statement.setObject(1, jsonbObject(resultSet.getBytes("ENTRY")));
                statement.executeUpdate();
            }
        }
    }

    public int currentWaterMark() throws SQLException {
        try (Statement statement = connection.createStatement()) {
            try(ResultSet resultSet = statement.executeQuery("SELECT ID FROM " + waterMarkTableName)){
                return resultSet.next() ? resultSet.getInt("ID") : 0;
            }
        }
    }

    private PGobject jsonbObject(byte[] value) {
        try {
            PGobject pGobject = new PGobject();
            pGobject.setType("jsonb");
            pGobject.setValue(new String(value, Charset.forName("UTF-8")));
            return pGobject;
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }
    }
}

@@ -678,24 +678,69 @@
 t.next()) %7B%0A
+            connection.setAutoCommit(false);%0A
             
@@ -1000,32 +1000,264 @@
 ;%0A            %7D%0A
+            try (PreparedStatement statement = connection.prepareStatement(%22UPDATE %22 + waterMarkTableName + %22 SET ID = ID + 1%22)) %7B%0A                statement.executeUpdate();%0A            %7D%0A            connection.setAutoCommit(true);%0A
         %7D%0A    %7D%0A

Increase watermark by one everytime when an entry is written