From b058e7dc5332413e23eceff7278a0abc7abc978d Mon Sep 17 00:00:00 2001
From: Raimund Hirz <hirzraimund@users.noreply.github.com>
Date: Wed, 13 Mar 2019 01:52:32 +0100
Subject: [PATCH] When deleting persistence journal entries through the common
 sql implementation also delete the entries in the metadata table. This solves
 #3205 (violation of primary key constraint when the value already exists) and
 ensures that there exists at most one entry per PersistenceId value. (#3468)

---
 .../Journal/BatchingSqlJournal.cs                         | 8 +++++---
 .../Akka.Persistence.Sql.Common/Journal/QueryExecutor.cs  | 4 +++-
 2 files changed, 8 insertions(+), 4 deletions(-)

diff --git a/src/contrib/persistence/Akka.Persistence.Sql.Common/Journal/BatchingSqlJournal.cs b/src/contrib/persistence/Akka.Persistence.Sql.Common/Journal/BatchingSqlJournal.cs
index bbc347fc5d6..2fe578cdb64 100644
--- a/src/contrib/persistence/Akka.Persistence.Sql.Common/Journal/BatchingSqlJournal.cs
+++ b/src/contrib/persistence/Akka.Persistence.Sql.Common/Journal/BatchingSqlJournal.cs
@@ -570,7 +570,9 @@ SELECT MAX(u.SeqNr) as SequenceNr
                     SELECT m.{conventions.SequenceNrColumnName} as SeqNr FROM {conventions.FullMetaTableName} m WHERE m.{conventions.PersistenceIdColumnName} = @PersistenceId) as u";
 
             DeleteBatchSql = $@"
-                DELETE FROM {conventions.FullJournalTableName} 
+                DELETE FROM {conventions.FullJournalTableName}
+                WHERE {conventions.PersistenceIdColumnName} = @PersistenceId AND {conventions.SequenceNrColumnName} <= @ToSequenceNr;
+                DELETE FROM {conventions.FullMetaTableName}
                 WHERE {conventions.PersistenceIdColumnName} = @PersistenceId AND {conventions.SequenceNrColumnName} <= @ToSequenceNr;";
 
             UpdateSequenceNrSql = $@"
@@ -908,8 +910,8 @@ protected virtual async Task HandleDeleteMessagesTo(DeleteMessagesTo req, TComma
 
                 command.CommandText = DeleteBatchSql;
                 command.Parameters.Clear();
-                AddParameter(command, "PersistenceId", DbType.String, persistenceId);
-                AddParameter(command, "ToSequenceNr", DbType.Int64, toSequenceNr);
+                AddParameter(command, "@PersistenceId", DbType.String, persistenceId);
+                AddParameter(command, "@ToSequenceNr", DbType.Int64, toSequenceNr);
 
                 await command.ExecuteNonQueryAsync();
 
diff --git a/src/contrib/persistence/Akka.Persistence.Sql.Common/Journal/QueryExecutor.cs b/src/contrib/persistence/Akka.Persistence.Sql.Common/Journal/QueryExecutor.cs
index a4e41f6b8c6..2b0b0f25a4b 100644
--- a/src/contrib/persistence/Akka.Persistence.Sql.Common/Journal/QueryExecutor.cs
+++ b/src/contrib/persistence/Akka.Persistence.Sql.Common/Journal/QueryExecutor.cs
@@ -329,7 +329,9 @@ SELECT MAX(u.SeqNr) as SequenceNr
                     SELECT m.{Configuration.SequenceNrColumnName} as SeqNr FROM {Configuration.FullMetaTableName} m WHERE m.{Configuration.PersistenceIdColumnName} = @PersistenceId) as u";
 
             DeleteBatchSql = $@"
-                DELETE FROM {Configuration.FullJournalTableName} 
+                DELETE FROM {Configuration.FullJournalTableName}
+                WHERE {Configuration.PersistenceIdColumnName} = @PersistenceId AND {Configuration.SequenceNrColumnName} <= @ToSequenceNr;
+                DELETE FROM {Configuration.FullMetaTableName}
                 WHERE {Configuration.PersistenceIdColumnName} = @PersistenceId AND {Configuration.SequenceNrColumnName} <= @ToSequenceNr;";
 
             UpdateSequenceNrSql = $@"