Add initial PostgreSQL implementation for get & put operations

app/build.gradle

@@ -1,39 +1,101 @@
 buildscript {
-    ext.gradleVersion = '7.5.1'
-    ext.protobufPlugInVersion = '0.8.12'
-    ext.protobufVersion = '3.21.7'
-    ext.jerseyVersion = '3.1.0'
-    ext.junitVersion = '5.9.0'
+  ext.gradleVersion = '7.5.1'
+  ext.protobufPlugInVersion = '0.8.12'
+  ext.protobufVersion = '3.21.7'
+  ext.jerseyVersion = '3.1.0'
+  ext.junitVersion = '5.9.0'
+  ext.postgresVersion = '42.5.1'
+  ext.jooqVersion = '3.17.7'
+  ext.guiceVersion = '5.1.0'
 }
 
 plugins {
-    id 'java'
-    id 'com.google.protobuf' version "${protobufPlugInVersion}"
-    id 'war'
-    id 'idea'
+  id 'java'
+  id 'com.google.protobuf' version "${protobufPlugInVersion}"
+  id 'war'
+  id 'idea'
+  id 'nu.studer.jooq' version '8.0'
 }
 
 repositories {
-    mavenCentral()
+  mavenCentral()
 }
 
 idea {
-    module {
-        generatedSourceDirs.add(file("build/generated/proto/main"))
-    }
+  module {
+    generatedSourceDirs.add(file("build/generated/proto/main"))
+  }
 }
 
 group 'org.vss'
 version '1.0'
 
-
 dependencies {
-    implementation "com.google.protobuf:protobuf-java:$protobufVersion"
+  implementation "com.google.protobuf:protobuf-java:$protobufVersion"
+
+  //jOOQ & Postgres impl deps
+  implementation "org.jooq:jooq:$jooqVersion"
+  implementation "org.jooq:jooq-meta:$jooqVersion"
+  implementation "org.jooq:jooq-codegen:$jooqVersion"
+  runtimeOnly "org.postgresql:postgresql:$postgresVersion"
+  jooqGenerator "org.postgresql:postgresql:$postgresVersion"
 
-    testImplementation "org.junit.jupiter:junit-jupiter-api:$junitVersion"
-    testRuntimeOnly "org.junit.jupiter:junit-jupiter-engine:$junitVersion"
+  implementation "com.google.inject:guice:$guiceVersion"
+
+  testImplementation "org.junit.jupiter:junit-jupiter-api:$junitVersion"
+  testRuntimeOnly "org.junit.jupiter:junit-jupiter-engine:$junitVersion"
+  testImplementation "org.hamcrest:hamcrest-library:2.2"
+  testImplementation "org.testcontainers:junit-jupiter:1.17.6"
+  testImplementation "org.testcontainers:postgresql:1.17.6"
 }
 
 test {
-    useJUnitPlatform()
-}
+  useJUnitPlatform()
+}
+
+protobuf {
+  protoc {
+    artifact = "com.google.protobuf:protoc:$protobufVersion"
+  }
+}
+
+jooq {
+  configurations {
+    main {
+      generateSchemaSourceOnCompilation = true
+
+      generationTool {
+        jdbc {
+          driver = 'org.postgresql.Driver'
+          url = 'jdbc:postgresql://localhost:5432/postgres'
+          user = 'postgres'
+          password = ''
+          properties {
+            property {
+              key = 'ssl'
+              value = 'false'
+            }
+          }
+        }
+        generator {
+          name = 'org.jooq.codegen.DefaultGenerator'
+          database {
+            name = 'org.jooq.meta.postgres.PostgresDatabase'
+            inputSchema = 'public'
+          }
+          generate {
+            deprecated = false
+            records = true
+            immutablePojos = true
+            fluentSetters = true
+          }
+          target {
+            packageName = 'org.vss.postgres'
+            directory = 'build/generated-src/jooq/main'
+          }
+          strategy.name = 'org.jooq.codegen.DefaultGeneratorStrategy'
+        }
+      }
+    }
+  }
+}

app/src/main/java/org/vss/KVStore.java

@@ -0,0 +1,12 @@
+package org.vss;
+
+public interface KVStore {
+
+  String GLOBAL_VERSION_KEY = "vss_global_version";
+
+  GetObjectResponse get(GetObjectRequest request);
+
+  PutObjectResponse put(PutObjectRequest request);
+
+  ListKeyVersionsResponse listKeyVersions(ListKeyVersionsRequest request);
+}

app/src/main/java/org/vss/exception/ConflictException.java

@@ -0,0 +1,7 @@
+package org.vss.exception;
+
+public class ConflictException extends RuntimeException {
+  public ConflictException(String message) {
+    super(message);
+  }
+}

app/src/main/java/org/vss/impl/postgres/PostgresBackendImpl.java

@@ -0,0 +1,132 @@
+package org.vss.impl.postgres;
+
+import com.google.inject.Inject;
+import com.google.protobuf.ByteString;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import javax.inject.Singleton;
+import org.jooq.DSLContext;
+import org.jooq.Insert;
+import org.jooq.Query;
+import org.jooq.Update;
+import org.vss.GetObjectRequest;
+import org.vss.GetObjectResponse;
+import org.vss.KVStore;
+import org.vss.KeyValue;
+import org.vss.ListKeyVersionsRequest;
+import org.vss.ListKeyVersionsResponse;
+import org.vss.PutObjectRequest;
+import org.vss.PutObjectResponse;
+import org.vss.exception.ConflictException;
+import org.vss.postgres.tables.records.VssDbRecord;
+
+import static org.jooq.impl.DSL.val;
+import static org.vss.postgres.tables.VssDb.VSS_DB;
+
+@Singleton
+public class PostgresBackendImpl implements KVStore {
+
+  private final DSLContext context;
+
+  @Inject
+  public PostgresBackendImpl(DSLContext context) {
+    this.context = context;
+  }
+
+  @Override
+  public GetObjectResponse get(GetObjectRequest request) {
+
+    VssDbRecord vssDbRecord = context.selectFrom(VSS_DB)
+        .where(VSS_DB.STORE_ID.eq(request.getStoreId())
+            .and(VSS_DB.KEY.eq(request.getKey())))
+        .fetchOne();
+
+    final KeyValue keyValue;
+
+    if (vssDbRecord != null) {
+      keyValue = KeyValue.newBuilder()
+          .setKey(vssDbRecord.getKey())
+          .setValue(ByteString.copyFrom(vssDbRecord.getValue()))
+          .setVersion(vssDbRecord.getVersion())
+          .build();
+    } else {
+      keyValue = KeyValue.newBuilder()
+          .setKey(request.getKey()).build();
+    }
+
+    return GetObjectResponse.newBuilder()
+        .setValue(keyValue)
+        .build();
+  }
+
+  @Override
+  public PutObjectResponse put(PutObjectRequest request) {
+
+    String storeId = request.getStoreId();
+
+    List<VssDbRecord> vssRecords = new ArrayList<>(request.getTransactionItemsList().stream()
+        .map(kv -> buildVssRecord(storeId, kv)).toList());
+
+    if (request.hasGlobalVersion()) {
+      VssDbRecord globalVersionRecord = buildVssRecord(storeId,
+          KeyValue.newBuilder()
+              .setKey(GLOBAL_VERSION_KEY)
+              .setVersion(request.getGlobalVersion())
+              .build());
+
+      vssRecords.add(globalVersionRecord);
+    }
+
+    context.transaction((ctx) -> {
+      DSLContext dsl = ctx.dsl();
+      List<Query> batchQueries = vssRecords.stream()
+          .map(vssRecord -> buildPutObjectQuery(dsl, vssRecord)).toList();
+
+      int[] batchResult = dsl.batch(batchQueries).execute();
+
+      for (int numOfRowsUpdated : batchResult) {
+        if (numOfRowsUpdated == 0) {
+          throw new ConflictException(
+              "Transaction could not be completed due to a possible conflict");
+        }
+      }
+    });
+
+    return PutObjectResponse.newBuilder().build();
+  }
+
+  private Query buildPutObjectQuery(DSLContext dsl, VssDbRecord vssRecord) {
+    return vssRecord.getVersion() == 0 ? buildInsertRecordQuery(dsl, vssRecord)
+        : buildUpdateRecordQuery(dsl, vssRecord);
+  }
+
+  private Insert<VssDbRecord> buildInsertRecordQuery(DSLContext dsl, VssDbRecord vssRecord) {
+    return dsl.insertInto(VSS_DB)
+        .values(vssRecord.getStoreId(), vssRecord.getKey(),
+            vssRecord.getValue(), 1)
+        .onDuplicateKeyIgnore();
+  }
+
+  private Update<VssDbRecord> buildUpdateRecordQuery(DSLContext dsl, VssDbRecord vssRecord) {
+    return dsl.update(VSS_DB)
+        .set(Map.of(VSS_DB.VALUE, vssRecord.getValue(),
+            VSS_DB.VERSION, vssRecord.getVersion() + 1))
+        .where(VSS_DB.STORE_ID.eq(vssRecord.getStoreId())
+            .and(VSS_DB.KEY.eq(vssRecord.getKey()))
+            .and(VSS_DB.VERSION.eq(vssRecord.getVersion())));
+  }
+
+  private VssDbRecord buildVssRecord(String storeId, KeyValue kv) {
+    return new VssDbRecord()
+        .setStoreId(storeId)
+        .setKey(kv.getKey())
+        .setValue(kv.getValue().toByteArray())
+        .setVersion(kv.getVersion());
+  }
+
+  @Override
+  public ListKeyVersionsResponse listKeyVersions(ListKeyVersionsRequest request) {
+    throw new UnsupportedOperationException("Operation not implemented");
+  }
+}

app/src/main/java/org/vss/impl/postgres/sql/v0_create_vss_db.sql

@@ -0,0 +1,7 @@
+CREATE TABLE vss_db (
+    store_id character varying(120) NOT NULL CHECK (store_id <> ''),
+    key character varying(120) NOT NULL,
+    value bytea NULL,
+    version bigint NOT NULL,
+    PRIMARY KEY (store_id, key)
+);