src, test, doc: allow customizing conflict behavior

Author: louwers

doc/api/sqlite.md

@@ -169,6 +169,11 @@ Creates and attackes a session to the database. This method is a wrapper around
   * `filter` {Function} Optional function that takes the name of a table as the first argument.
      When `true` is returned, includes the change, otherwise, it is discarded. When not provided
      no changes are filtered, and all are changes are attempted.
+  * `onConflict` {number} When provided, must be one of `SQLITE_CHANGESET_OMIT`,
+    `SQLITE_CHANGESET_REPLACE` or `SQLITE_CHANGESET_ABORT`. Determines how conflicts are handled.
+    Conflicting changes are either omitted, changes replace existing values or the
+    call is aborted when the first conflicting change is encountered (this is the default).
+* Returns: {boolean} Whether the changeset was applied succesfully without being aborted.
 
 An exception is thrown if the database is not
 open. This method is a wrapper around [`sqlite3changeset_apply()`][].

src/node_sqlite.cc

@@ -276,8 +276,11 @@ void DatabaseSync::CreateSession(const FunctionCallbackInfo<Value>& args) {
   args.GetReturnValue().Set(session->object());
 }
 
+static std::function<int()> conflictCallback;
+
 static int xConflict(void* pCtx, int eConflict, sqlite3_changeset_iter* pIter) {
-  return SQLITE_CHANGESET_ABORT;
+  if (!conflictCallback) return SQLITE_CHANGESET_ABORT;
+  return conflictCallback();
 }
 
 static std::function<bool(std::string)> filterCallback;
@@ -289,6 +292,7 @@ static int xFilter(void* pCtx, const char* zTab) {
 }
 
 void DatabaseSync::ApplyChangeset(const FunctionCallbackInfo<Value>& args) {
+  conflictCallback = nullptr;
   filterCallback = nullptr;
 
   DatabaseSync* db;
@@ -313,6 +317,24 @@ void DatabaseSync::ApplyChangeset(const FunctionCallbackInfo<Value>& args) {
     }
 
     Local<Object> options = args[1].As<Object>();
+
+    Local<String> conflictKey = String::NewFromUtf8(env->isolate(), "onConflict", v8::NewStringType::kNormal).ToLocalChecked();
+    if (options->HasOwnProperty(env->context(), conflictKey).FromJust()) {
+      Local<Value> conflictValue = options->Get(env->context(), conflictKey).ToLocalChecked();
+
+      if (!conflictValue->IsNumber()) {
+        node::THROW_ERR_INVALID_ARG_TYPE(
+            env->isolate(),
+            "The \"options.onConflict\" argument must be an number.");
+        return;
+      }
+
+      int conflictInt = conflictValue->Int32Value(env->context()).FromJust();
+      conflictCallback = [conflictInt]() -> int {
+        return conflictInt;
+      };
+    }
+
     Local<String> filterKey = String::NewFromUtf8(
       env->isolate(),
       "filter",
@@ -351,7 +373,6 @@ void DatabaseSync::ApplyChangeset(const FunctionCallbackInfo<Value>& args) {
     buf.length(),
     const_cast<void *>(static_cast<const void *>(buf.data())),
     xFilter,
-    // TODO(louwers): allow custom conflict handler
     xConflict,
     nullptr);
   if (r == SQLITE_ABORT) {

test/parallel/test-sqlite.js

@@ -881,7 +881,7 @@ suite('session extension', () => {
     t.assert.strictEqual(database1.prepare(select2).all().length, 2);  // data1 should have values in database1
   });
 
-  test('conflict while applying changeset', (t) => {
+  const prepareConflict = () => {
     const database1 = new DatabaseSync(':memory:');
     const database2 = new DatabaseSync(':memory:');
 
@@ -896,9 +896,57 @@ suite('session extension', () => {
     const insertSql = 'INSERT INTO data (key, value) VALUES (?, ?)';
     const session = database1.createSession();
     database1.prepare(insertSql).run(1, 'hello');
+    database1.prepare(insertSql).run(2, 'foo');
     database2.prepare(insertSql).run(1, 'world');
+    return {
+      database2,
+      changeset: session.changeset()
+    }
+  }
+
+  test('conflict while applying changeset (default abort)', (t) => {
+    const { database2, changeset } = prepareConflict();
     // When changeset is aborted due to a conflict, applyChangeset should return false
-    t.assert.strictEqual(database2.applyChangeset(session.changeset()), false);
+    t.assert.strictEqual(database2.applyChangeset(changeset), false);
+    t.assert.deepStrictEqual(
+      database2.prepare('SELECT value from data').all(),
+      [{ value: 'world' }]);  // unchanged
+  });
+
+  test('conflict while applying changeset (explicit abort)', (t) => {
+    const { database2, changeset } = prepareConflict();
+    const result = database2.applyChangeset(changeset, {
+      onConflict: SQLITE_CHANGESET_ABORT
+    });
+    // When changeset is aborted due to a conflict, applyChangeset should return false
+    t.assert.strictEqual(result, false);
+    t.assert.deepStrictEqual(
+      database2.prepare('SELECT value from data').all(),
+      [{ value: 'world' }]);  // unchanged
+  });
+
+  test('conflict while applying changeset (replacement)', (t) => {
+    const { database2, changeset } = prepareConflict();
+    const result = database2.applyChangeset(changeset, {
+      onConflict: SQLITE_CHANGESET_REPLACE
+    });
+    // Not aborted due to conflict, so should return true
+    t.assert.strictEqual(result, true);
+    t.assert.deepStrictEqual(
+      database2.prepare('SELECT value from data ORDER BY key').all(),
+      [{ value: 'hello'}, { value: 'foo' }]);  // replaced
+  });
+
+  test('conflict while applying changeset (omit)', (t) => {
+    const { database2, changeset } = prepareConflict();
+    const result = database2.applyChangeset(changeset, {
+      onConflict: SQLITE_CHANGESET_OMIT
+    });
+    // Not aborted due to conflict, so should return true
+    t.assert.strictEqual(result, true);
+    t.assert.deepStrictEqual(
+      database2.prepare('SELECT value from data ORDER BY key ASC').all(),
+      [{ value: 'world'}, { value: 'foo' }]);  // conflicting change omitted
   });
 
   test('constants are defined', (t) => {