/// <summary>
        /// Updates the saga data
        /// </summary>
        public async Task Update(ISagaData sagaData, IEnumerable<ISagaCorrelationProperty> correlationProperties)
        {
            var id = GetId(sagaData);

            lock (_lock)
            {
                if (!_data.ContainsKey(id))
                {
                    throw new ConcurrencyException("Saga data with ID {0} no longer exists and cannot be updated", id);
                }

                VerifyCorrelationPropertyUniqueness(sagaData, correlationProperties);

                var existingCopy = _data[id];

                if (existingCopy.Revision != sagaData.Revision)
                {
                    throw new ConcurrencyException("Attempted to update saga data with ID {0} with revision {1}, but the existing data was updated to revision {2}",
                        id, sagaData.Revision, existingCopy.Revision);
                }

                var clone = Clone(sagaData);
                clone.Revision++;
                _data[id] = clone;
                sagaData.Revision++;
            }
        }

        public async Task Insert(ISagaData sagaData, IEnumerable<ISagaCorrelationProperty> correlationProperties)
        {
            if (sagaData.Id == Guid.Empty)
            {
                throw new InvalidOperationException($"Saga data {sagaData.GetType()} has an uninitialized Id property!");
            }

            if (sagaData.Revision != 0)
            {
                throw new InvalidOperationException($"Attempted to insert saga data with ID {sagaData.Id} and revision {sagaData.Revision}, but revision must be 0 on first insert!");
            }

            using (var session = _documentStore.OpenAsyncSession())
            {
                session.Advanced.UseOptimisticConcurrency = true;

                var sagaDataDocumentId = SagaDataDocument.GetIdFromGuid(sagaData.Id);

                var existingSagaDataDocument = await session.LoadAsync<SagaDataDocument>(sagaDataDocumentId);

                if (existingSagaDataDocument != null)
                {
                    throw new ConcurrencyException("Cannot insert document with an id that already exists");
                }

                var sagaDataDocument = new SagaDataDocument(sagaData);
                await session.StoreAsync(sagaDataDocument, sagaDataDocumentId);

                var correlationPropertyDocumentIds = await SaveCorrelationProperties(session, sagaData, correlationProperties, sagaDataDocumentId);
                sagaDataDocument.SagaCorrelationPropertyDocumentIds = correlationPropertyDocumentIds;

                await session.SaveChangesAsync();
            }
        }

        /// <summary>
        /// Saves a snapshot of the saga data along with the given metadata
        /// </summary>
        public async Task Save(ISagaData sagaData, Dictionary<string, string> sagaAuditMetadata)
        {
            using (var connection = await _connectionProvider.GetConnection())
            {
                using (var command = connection.CreateCommand())
                {
                    command.CommandText = string.Format(@"

INSERT INTO [{0}] (
    [id],
    [revision],
    [data],
    [metadata]
) VALUES (
    @id, 
    @revision, 
    @data,
    @metadata
)

", _tableName);
                    command.Parameters.Add("id", SqlDbType.UniqueIdentifier).Value = sagaData.Id;
                    command.Parameters.Add("revision", SqlDbType.Int).Value = sagaData.Revision;
                    command.Parameters.Add("data", SqlDbType.NVarChar).Value = JsonConvert.SerializeObject(sagaData, DataSettings);
                    command.Parameters.Add("metadata", SqlDbType.NVarChar).Value = JsonConvert.SerializeObject(sagaAuditMetadata, MetadataSettings);

                    await command.ExecuteNonQueryAsync();
                }

                await connection.Complete();
            }
        }

        public async Task Save(ISagaData sagaData, Dictionary<string, string> sagaAuditMetadata)
        {
            using (var connection = await _connectionHelper.GetConnection())
            {
                using (var command = connection.CreateCommand())
                {
                    command.CommandText =
                        [email protected]"

INSERT
    INTO ""{_tableName}"" (""id"", ""revision"", ""data"", ""metadata"")
    VALUES (@id, @revision, @data, @metadata);

";
                    command.Parameters.Add("id", NpgsqlDbType.Uuid).Value = sagaData.Id;
                    command.Parameters.Add("revision", NpgsqlDbType.Integer).Value = sagaData.Revision;
                    command.Parameters.Add("data", NpgsqlDbType.Bytea).Value = _objectSerializer.Serialize(sagaData);
                    command.Parameters.Add("metadata", NpgsqlDbType.Jsonb).Value =
                        _dictionarySerializer.SerializeToString(sagaAuditMetadata);

                    await command.ExecuteNonQueryAsync();
                }
                
                connection.Complete();
            }
        }

 public void Insert(ISagaData sagaData, IEnumerable<ISagaCorrelationProperty> correlationProperties)
 {
     var indexItems = ReadIndexItems();
     indexItems.RemoveAll(i => i.SagaId == sagaData.Id);
     indexItems.AddRange(GetPropertiesToIndex(sagaData, correlationProperties));
     var path = SagaFilePath(sagaData.Id);
     File.WriteAllText(path, JsonConvert.SerializeObject(sagaData, _serializerSettings));
     WriteSagaIndexItems(indexItems);
 }

        public async Task Save(ISagaData sagaData, Dictionary<string, string> sagaAuditMetadata)
        {
            var document = new BsonDocument
            {
                {"_id", new {Id = sagaData.Id, Revision = sagaData.Revision}.ToBsonDocument()},
                {"Metadata", sagaAuditMetadata.ToBsonDocument()},
                {"Data", sagaData.ToBsonDocument()}
            };

            await _snapshots.InsertOneAsync(document);
        }

        public async Task Insert(ISagaData sagaData, IEnumerable<ISagaCorrelationProperty> correlationProperties)
        {
            if (sagaData.Id == Guid.Empty)
            {
                throw new InvalidOperationException(string.Format("Attempted to insert saga data {0} without an ID", sagaData.GetType()));
            }

            var collection = GetCollection(sagaData.GetType());

            await collection.InsertOneAsync(sagaData.ToBsonDocument()).ConfigureAwait(false);
        }

 internal void AddInstance(ISagaData sagaData)
 {
     lock (_lock)
     {
         var instance = Clone(sagaData);
         if (instance.Id == Guid.Empty)
         {
             instance.Id = Guid.NewGuid();
         }
         _data[instance.Id] = instance;
     }
 }

 Dictionary<string, string> GetMetadata(ISagaData sagaData, object handler, Message message)
 {
     return new Dictionary<string, string>
     {
         {SagaAuditingMetadataKeys.HandleQueue, _transport.Address},
         {SagaAuditingMetadataKeys.SagaDataType, sagaData.GetType().GetSimpleAssemblyQualifiedName()},
         {SagaAuditingMetadataKeys.SagaHandlerType, handler.GetType().GetSimpleAssemblyQualifiedName()},
         {SagaAuditingMetadataKeys.MessageType, message.GetMessageType()},
         {SagaAuditingMetadataKeys.MessageId, message.GetMessageId()},
         {SagaAuditingMetadataKeys.MachineName, Environment.MachineName},
     };
 }

        public async Task Save(ISagaData sagaData, Dictionary<string, string> sagaAuditMetadata)
        {
            var logData = new
            {
                Data = sagaData,
                Metadata = sagaAuditMetadata
            };
            
            var jsonText = JsonConvert.SerializeObject(logData, Formatting.None);

            _log.Info(jsonText);
        }

        public async Task Delete(ISagaData sagaData)
        {
            var collection = GetCollection(sagaData.GetType());

            var result = await collection.DeleteManyAsync(new BsonDocument("_id", sagaData.Id)).ConfigureAwait(false);

            if (result.DeletedCount != 1)
            {
                throw new ConcurrencyException("Saga data {0} with ID {1} in collection {2} could not be deleted", 
                    sagaData.GetType(), sagaData.Id, collection.CollectionNamespace);
            }
        }

 /// <summary>
 /// Archives the given saga data under its current ID and revision
 /// </summary>
 public async Task Save(ISagaData sagaData, Dictionary<string, string> sagaAuditMetadata)
 {
     var dataRef = $"{sagaData.Id:N}/{sagaData.Revision:0000000000}/data.json";
     var metaDataRef = $"{sagaData.Id:N}/{sagaData.Revision:0000000000}/metadata.json";
     var dataBlob = _container.GetBlockBlobReference(dataRef);
     var metaDataBlob = _container.GetBlockBlobReference(metaDataRef);
     dataBlob.Properties.ContentType = "application/json";
     metaDataBlob.Properties.ContentType = "application/json";
     await dataBlob.UploadTextAsync(JsonConvert.SerializeObject(sagaData, DataSettings), TextEncoding, DefaultAccessCondition, DefaultRequestOptions, DefaultOperationContext);
     await metaDataBlob.UploadTextAsync(JsonConvert.SerializeObject(sagaAuditMetadata, MetadataSettings), TextEncoding, DefaultAccessCondition, DefaultRequestOptions, DefaultOperationContext);
     await dataBlob.SetPropertiesAsync();
     await metaDataBlob.SetPropertiesAsync();
 }

        public void Delete(ISagaData sagaData)
        {
            var collection = database.GetCollection(collectionName);

            if (Transaction.Current != null)
            {
                var hack = new AmbientTxHack(() => collection.Remove(Query.EQ("_id", sagaData.Id)));
                Transaction.Current.EnlistVolatile(hack, EnlistmentOptions.None);
            }
            else
            {
                collection.Remove(Query.EQ("_id", sagaData.Id));
            }
        }

        public void Delete(ISagaData sagaData)
        {
            var collection = database.GetCollection(collectionName);

            var query = Query.And(Query.EQ("_id", sagaData.Id),
                                  Query.EQ("_rev", sagaData.Revision));

            var safeModeResult = collection.Remove(query, SafeMode.True);

            EnsureResultIsGood(safeModeResult,
                               "delete saga data of type {0} with _id {1} and _rev {2}",
                               sagaData.GetType(),
                               sagaData.Id,
                               sagaData.Revision);
        }

        public void Delete(ISagaData sagaData)
        {
            using (var connection = new SqlConnection(connectionString))
            {
                connection.Open();

                using (var command = connection.CreateCommand())
                {
                    command.CommandText = @"delete from sagas where id = @id;
                                            delete from saga_index where saga_id = @id";
                    command.Parameters.AddWithValue("id", sagaData.Id);
                    command.ExecuteNonQuery();
                }
            }
        }

        public async Task Delete(ISagaData sagaData)
        {
            var collection = GetCollection(sagaData.GetType());

            var result = collection.Remove(Query.EQ("_id", sagaData.Id));

            try
            {
                CheckResult(result, 1);
            }
            catch (Exception exception)
            {
                throw new ConcurrencyException(exception, "Saga data {0} with ID {1} in collection {2} could not be deleted", 
                    sagaData.GetType(), sagaData.Id, collection.Name);
            }
        }

        public async Task Insert(ISagaData sagaData, IEnumerable<ISagaCorrelationProperty> correlationProperties)
        {
            if (sagaData.Id == Guid.Empty)
            {
                throw new InvalidOperationException($"Attempted to insert saga data {sagaData.GetType()} without an ID");
            }

            if (sagaData.Revision != 0)
            {
                throw new InvalidOperationException($"Attempted to insert saga data with ID {sagaData.Id} and revision {sagaData.Revision}, but revision must be 0 on first insert!");
            }

            var collection = GetCollection(sagaData.GetType());

            await collection.InsertOneAsync(sagaData.ToBsonDocument()).ConfigureAwait(false);
        }

        public async Task Update(ISagaData sagaData, IEnumerable<ISagaCorrelationProperty> correlationProperties)
        {
            var collection = GetCollection(sagaData.GetType());

            var criteria = Builders<BsonDocument>.Filter.And(Builders<BsonDocument>.Filter.Eq("_id", sagaData.Id),
                Builders<BsonDocument>.Filter.Eq("Revision", sagaData.Revision));

            sagaData.Revision++;

            var result = await collection.ReplaceOneAsync(criteria, sagaData.ToBsonDocument(sagaData.GetType())).ConfigureAwait(false);

            if (!result.IsModifiedCountAvailable || result.ModifiedCount != 1)
            {
                throw new ConcurrencyException("Saga data {0} with ID {1} in collection {2} could not be updated!",
                    sagaData.GetType(), sagaData.Id, collection.CollectionNamespace);
            }
        }

        public async Task Update(ISagaData sagaData, IEnumerable<ISagaCorrelationProperty> correlationProperties)
        {
            using (var session = _documentStore.OpenAsyncSession())
            {
                var documentId = SagaDataDocument.GetIdFromGuid(sagaData.Id);
                var existingSagaData = await session.LoadAsync<SagaDataDocument>(documentId);
                sagaData.Revision++;
                existingSagaData.SagaData = sagaData;
                
                await DeleteCorrelationPropertyDataForSaga(existingSagaData, session);

                //add the new saga correlation documents
                var correlationPropertyDocumentIds = await SaveCorrelationProperties(session, sagaData, correlationProperties, existingSagaData.Id);
                existingSagaData.SagaCorrelationPropertyDocumentIds = correlationPropertyDocumentIds;

                await session.SaveChangesAsync();
            }
        }

        /// <summary>
        /// Saves a snapshot of the saga data along with the given metadata
        /// </summary>
        public async Task Save(ISagaData sagaData, Dictionary<string, string> sagaAuditMetadata)
        {
            var jsonText = JsonConvert.SerializeObject(new Snapshot
            {
                Data = sagaData,
                Metadata = sagaAuditMetadata
            }, Formatting.Indented);

            var snapshotFilePath = Path.Combine(_snapshotDirectory, string.Format("{0:N}-{1}.json", sagaData.Id, sagaData.Revision));

            using (var file = File.OpenWrite(snapshotFilePath))
            {
                using (var writer = new StreamWriter(file, Encoding.UTF8))
                {
                    await writer.WriteAsync(jsonText);
                }
            }
        }