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)
+);

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)
+);

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)
+);

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)
+);

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)
+);

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)
+);

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)
+);

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)
+);