feat: add Compare protos

Signed-off-by: tison <[email protected]>

zeronos-proto/src/main/proto/rpc.proto

@@ -77,8 +77,40 @@ message ResponseOp {
   }
 }
 
+message Compare {
+  enum CompareResult {
+    EQUAL = 0;
+    GREATER = 1;
+    LESS = 2;
+    NOT_EQUAL = 3;
+    GREATER_OR_EQUAL = 4;
+    LESS_OR_EQUAL = 5;
+  }
+
+  enum CompareTarget {
+    VERSION = 0;
+    VALUE = 1;
+  }
+
+  // result is logical comparison operation for this comparison.
+  CompareResult result = 1;
+  // target is the key-value field to inspect for the comparison.
+  CompareTarget target = 2;
+  // key is the subject key for the comparison operation.
+  bytes key = 3;
+
+  oneof target_union {
+    // version is the version of the given key
+    int64 version = 4;
+    // value is the value of the given key, in bytes.
+    bytes value = 5;
+  }
+}
+
 message TxnRequest {
+  repeated Compare compare = 1;
   repeated RequestOp success = 2;
+  repeated RequestOp failure = 3;
 }
 
 message TxnResponse {

zeronos-server/src/main/java/io/korandoru/zeronos/server/state/ZeroStateMachine.java

@@ -16,6 +16,7 @@
 
 package io.korandoru.zeronos.server.state;
 
+import io.korandoru.zeronos.proto.Compare;
 import io.korandoru.zeronos.proto.DeleteRangeRequest;
 import io.korandoru.zeronos.proto.DeleteRangeResponse;
 import io.korandoru.zeronos.proto.KeyBytes;
@@ -42,6 +43,7 @@
 import java.io.File;
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Comparator;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicLong;
@@ -156,6 +158,51 @@ private DeleteRangeResponse deleteRange(WriteTxn writeTxn, DeleteRangeRequest re
         return DeleteRangeResponse.newBuilder().setDeleted(r.getTotal()).build();
     }
 
+    private boolean applyCompares(ReadTxn readTxn, List<Compare> compareList, long revision)
+            throws InvalidProtocolBufferException {
+        for (Compare compare : compareList) {
+            final RangeRequest req = RangeRequest.newBuilder().setKey(compare.getKey()).build();
+            final RangeResponse resp = range(readTxn, req, revision);
+            if (!applyCompare(compare, resp.getKvsList())) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    private boolean applyCompare(Compare compare, List<KeyValue> kvs) {
+        if (kvs.isEmpty()) {
+            // always fail if comparing a value on a key/keys that doesn't exist;
+            // null == empty string in gRPC; no way to represent missing value
+            return false;
+        }
+
+        final Comparator<ByteString> comparator = ByteString.unsignedLexicographicalComparator();
+        for (KeyValue kv : kvs) {
+            final int result = switch (compare.getTarget()) {
+                case VERSION -> Long.compare(kv.getVersion(), compare.getVersion());
+                case VALUE -> comparator.compare(kv.getValue(), compare.getValue());
+                default -> throw new UnsupportedOperationException(compare.getTarget().name());
+            };
+
+            final boolean match = switch (compare.getResult()) {
+                case EQUAL -> result == 0;
+                case GREATER -> result > 0;
+                case LESS -> result < 0;
+                case NOT_EQUAL -> result != 0;
+                case GREATER_OR_EQUAL -> result >= 0;
+                case LESS_OR_EQUAL -> result <= 0;
+                default -> throw new UnsupportedOperationException(compare.getResult().name());
+            };
+
+            if (!match) {
+                return false;
+            }
+        }
+
+        return true;
+    }
+
     @Override
     public CompletableFuture<Message> query(Message request) {
         final List<RequestOp> requestList;