<commit_before>namespace SqlStreamStore
{
    using System;
    using System.Collections.Generic;
    using System.Linq;
    using System.Threading;
    using System.Threading.Tasks;
    using Microsoft.Data.Sqlite;
    using SqlStreamStore.Infrastructure;
    using SqlStreamStore.Streams;
    using StreamStoreStore.Json;

    public partial class SqliteStreamStore
    {
        protected override async Task<AppendResult> AppendToStreamInternal(
            string streamId,
            int expectedVersion,
            NewStreamMessage[] messages,
            CancellationToken cancellationToken)
        {
            GuardAgainstDisposed();
            cancellationToken.ThrowIfCancellationRequested();
            
            SqliteAppendResult result;
            StreamIdInfo streamIdInfo = new StreamIdInfo(streamId);
            
            switch(expectedVersion)
            {
                case ExpectedVersion.Any:
                    result = await AppendToStreamAnyVersion(streamIdInfo.SqlStreamId, messages, cancellationToken).NotOnCapturedContext();
                    break;
                case ExpectedVersion.EmptyStream:
                    result = await AppendToStreamEmpty(streamIdInfo.SqlStreamId, messages, cancellationToken).NotOnCapturedContext();
                    break;
                case ExpectedVersion.NoStream:
                    result = await AppendToNonexistentStream(streamIdInfo.SqlStreamId, messages, cancellationToken).NotOnCapturedContext();
                    break;
                default:
                    result = AppendToStreamExpectedVersion(streamIdInfo.SqlStreamId, expectedVersion, messages, cancellationToken);
                    break;
            }

            if(result.MaxCount.HasValue)
            {
                await CheckStreamMaxCount(streamId, result.MaxCount, cancellationToken).NotOnCapturedContext();
            }

            await TryScavengeAsync(streamId, cancellationToken);

            return result;
        }

        private Task<SqliteAppendResult> AppendToStreamAnyVersion(SqliteStreamId streamId, NewStreamMessage[] messages, CancellationToken cancellationToken)
        {
            using(var conn = OpenConnection(false))
            using(var txn = conn.BeginTransaction())
            using(var cmd = conn.CreateCommand())
            {
                cmd.Transaction = txn;

                var stored = AppendToStreamAnyVersion(cmd, streamId, messages, cancellationToken);
                
                txn.Commit();
                
                return Task.FromResult(stored);
            }
        }

        private SqliteAppendResult AppendToStreamAnyVersion(
            SqliteCommand cmd,
            SqliteStreamId streamId,
            NewStreamMessage[] messages,
            CancellationToken cancellationToken)
        {
            var internalId = ResolveInternalStreamId(streamId.IdOriginal, cmd, false) ?? CreateStream(cmd, streamId);
            // unit tests state that if there is a single message, and it exists already, then
            // do not add the second message into the system.  Send the SqliteAppendResult back
            // as if you had done the insert.
            if(messages.Length == 1)
            {
                var msg = messages[0];

                // if the message's event id exists in the database...
                cmd.CommandText = @"SELECT count(*) 
                                        FROM messages
                                        WHERE event_id = @eventId AND stream_id_internal = @idInternal;";
                cmd.Parameters.Clear();
                cmd.Parameters.AddWithValue("@idInternal", internalId);
                cmd.Parameters.AddWithValue("@eventId", msg.MessageId);

                var existsInStream = cmd.ExecuteScalar<long>(0) > 0;
                if(existsInStream)
                {
                    cmd.CommandText = @"SELECT [version], [position]
                                        FROM streams
                                        WHERE id_internal = @idInternal;";
                    cmd.Parameters.Clear();
                    cmd.Parameters.AddWithValue("@idInternal", internalId);

                    using(var reader = cmd.ExecuteReader())
                    if(reader.Read())
                    {
                        var ver = reader.ReadScalar<int>(0);
                        var pos = reader.ReadScalar<long>(1);

                        {
                            return new SqliteAppendResult(ver, pos, null);
                        }
                    }
                }
            }
            else if (messages.Any())
            {
                var msg1 = messages.First();
                cmd.CommandText = @"SELECT event_id
                                    FROM messages
                                    WHERE [position] >= (SELECT [position] 
                                                      FROM messages 
                                                      WHERE messages.event_id = @messageId
                                                        AND messages.stream_id_internal = @internalId)
                                        AND stream_id_internal = @internalId;";
                cmd.Parameters.Clear();
                cmd.Parameters.AddWithValue("@messageId", msg1.MessageId);
                cmd.Parameters.AddWithValue("@internalId", internalId);
                cmd.Parameters.AddWithValue("@count", messages.Length);

                var eventIds = new List<Guid>();
                using(var reader = cmd.ExecuteReader())
                {
                    while(reader.Read())
                    {
                        eventIds.Add(reader.ReadScalar<Guid>(0, Guid.Empty));
                    }

                    eventIds.RemoveAll(x => x == Guid.Empty);
                }

                if(eventIds.Count > 0)
                {
                    for(var i = 0; i < Math.Min(eventIds.Count, messages.Length); i++)
                    {
                        if(eventIds[i] != messages[i].MessageId)
                        {
                            throw new WrongExpectedVersionException(
                                ErrorMessages.AppendFailedWrongExpectedVersion(
                                    streamId.IdOriginal,
                                    StreamVersion.Start),
                                streamId.IdOriginal,
                                StreamVersion.Start);
                        }
                    }

                    if(eventIds.Count < messages.Length && eventIds.Count > 0)
                    {
                        throw new WrongExpectedVersionException(
                            ErrorMessages.AppendFailedWrongExpectedVersion(
                                streamId.IdOriginal,
                                StreamVersion.Start),
                            streamId.IdOriginal,
                            StreamVersion.Start);
                    }

                    cmd.CommandText = @"SELECT [version], [position]
                                        FROM streams
                                        WHERE id_internal = @idInternal;";
                    cmd.Parameters.Clear();
                    cmd.Parameters.AddWithValue("@idInternal", internalId);

                    using(var reader = cmd.ExecuteReader())
                        if(reader.Read())
                        {
                            var ver = reader.ReadScalar<int>(0);
                            var pos = reader.ReadScalar<long>(1);

                            {
                                return new SqliteAppendResult(ver, pos, null);
                            }
                        }
                }
            }

            var stored = StoreMessages(messages, cmd, streamId);
            
            return stored;
        }

        private Task<SqliteAppendResult> AppendToStreamEmpty(SqliteStreamId streamId, NewStreamMessage[] messages, CancellationToken cancellationToken)
        {
            var internalId = ResolveInternalStreamId(streamId.IdOriginal);
            
            using(var conn = OpenConnection(false))
            using(var txn = conn.BeginTransaction())
            using(var cmd = conn.CreateCommand())
            {
                cmd.Transaction = txn;

                var lengthOfStream = GetStreamLength(cmd, internalId, out var result);
                if(lengthOfStream > StreamVersion.Start)
                {
                    throw new WrongExpectedVersionException(
                        ErrorMessages.AppendFailedWrongExpectedVersion(
                            streamId.IdOriginal,
                            StreamVersion.Start),
                        streamId.IdOriginal,
                        StreamVersion.Start);
                }

                var stored = StoreMessages(messages, cmd, streamId);
                
                txn.Commit();

                return Task.FromResult(stored);
            }
        }

        private static int GetStreamLength(SqliteCommand cmd, int? internalId, out SqliteAppendResult appendResult)
        {
            // check to see if the stream has records.  if so, throw wrongexpectedversion exception.
            cmd.CommandText = @"SELECT COUNT(*)
                                    FROM messages
                                    WHERE stream_id_internal = @internalId";
            cmd.Parameters.Clear();
            cmd.Parameters.AddWithValue("@internalId", internalId);
            var streamLength = StreamVersion.End;
            
            if(streamLength > 0)
            {
                // we has information in the stream.
                {
                    appendResult = new SqliteAppendResult(ExpectedVersion.NoStream,
                        ExpectedVersion.NoStream,
                        internalId ?? -1);
                    return streamLength;
                }
            }

            appendResult = default;
            return StreamVersion.Start;
        }

        private Task<SqliteAppendResult> AppendToNonexistentStream(SqliteStreamId streamId, NewStreamMessage[] messages, CancellationToken cancellationToken)
        {
            var internalId = ResolveInternalStreamId(streamId.IdOriginal, throwIfNotExists: false);
            if(internalId != null)
            {
                using(var connection = OpenConnection())
                using(var command = connection.CreateCommand())
                {
                    command.CommandText = @"SELECT event_id
                                    FROM messages
                                    WHERE messages.stream_id_internal = @internalId
                                    ORDER BY messages.position;";
                    command.Parameters.Clear();
                    command.Parameters.AddWithValue("@internalId", internalId);

                    var eventIds = new List<Guid>();
                    using(var reader = command.ExecuteReader())
                    {
                        while(reader.Read())
                        {
                            eventIds.Add(reader.ReadScalar<Guid>(0, Guid.Empty));
                        }

                        eventIds.RemoveAll(x => x == Guid.Empty);
                    }

                    if(eventIds.Count > 0)
                    {
                        for(var i = 0; i < Math.Min(eventIds.Count, messages.Length); i++)
                        {
                            if(eventIds[i] != messages[i].MessageId)
                            {
                                throw new WrongExpectedVersionException(
                                    ErrorMessages.AppendFailedWrongExpectedVersion(
                                        streamId.IdOriginal,
                                        ExpectedVersion.NoStream),
                                    streamId.IdOriginal,
                                    ExpectedVersion.NoStream);
                            }
                        }

                        if(eventIds.Count < messages.Length)
                        {
                            throw new WrongExpectedVersionException(
                                ErrorMessages.AppendFailedWrongExpectedVersion(
                                    streamId.IdOriginal,
                                    ExpectedVersion.NoStream),
                                streamId.IdOriginal,
                                ExpectedVersion.NoStream);
                        }

                        command.CommandText = @"SELECT [version], [position]
                                            FROM streams
                                            WHERE id_internal = @idInternal;";
                        command.Parameters.Clear();
                        command.Parameters.AddWithValue("@idInternal", internalId);

                        using(var reader = command.ExecuteReader())
                            if(reader.Read())
                            {
                                var ver = reader.ReadScalar<int>(0);
                                var pos = reader.ReadScalar<long>(1);

                                {
                                    return Task.FromResult(new SqliteAppendResult(ver, pos, null));
                                }
                            }
                    }
                }
            }

            CreateStream(streamId);
            
            return AppendToStreamEmpty(streamId, messages, cancellationToken);
        }

        private SqliteAppendResult AppendToStreamExpectedVersion(SqliteStreamId streamId, int expectedVersion, NewStreamMessage[] messages, CancellationToken cancellationToken)
        {
            var internalId = ResolveInternalStreamId(streamId.IdOriginal, throwIfNotExists: false);

            if(internalId == null)
            {
                throw new WrongExpectedVersionException(
                    ErrorMessages.AppendFailedWrongExpectedVersion(
                        streamId.IdOriginal,
                        expectedVersion),
                    streamId.IdOriginal,
                    expectedVersion);
            }
            
            using(var connection = OpenConnection(false))
            using(var transaction = connection.BeginTransaction())
            using(var command = connection.CreateCommand())
            {
                command.Transaction = transaction;
                
                if(messages.Length == 1)
                {
                    var msg = messages.First();
                    
                    // tries to fix "When_append_single_message_to_stream_with_correct_expected_version_second_time_with_same_initial_messages_then_should_have_expected_result"
                    command.CommandText = @"SELECT COUNT(*)
                                            FROM messages
                                            WHERE messages.stream_id_internal = @internalId
                                                AND messages.stream_version = @expected
                                                AND messages.event_id = @eventId;";
                    command.Parameters.Clear();
                    command.Parameters.AddWithValue("@internalId", internalId);
                    command.Parameters.AddWithValue("@expected", expectedVersion + 1);
                    command.Parameters.AddWithValue("@eventId", msg.MessageId);
                    var duplicateMessage = command.ExecuteScalar<int?>(0);
                    if(duplicateMessage > 0)
                    {
                        command.CommandText = @"SELECT messages.stream_version, messages.[position]
                                        FROM messages
                                        WHERE messages.stream_id_internal = @stream_id_internal
                                        ORDER BY messages.[position] DESC
                                        LIMIT 1";
                        command.Parameters.Clear();
                        command.Parameters.AddWithValue("@stream_id_internal", internalId ?? -1);
                        using(var reader = command.ExecuteReader())
                        {
                            if (reader.Read() && reader.HasRows)
                                return new SqliteAppendResult(
                                    reader.ReadScalar(0, StreamVersion.End),
                                    reader.ReadScalar(1, Position.End),
                                    internalId ?? -1
                                );
                        }
                    }
                    // end - tries to fix "When_append_single_message_to_stream_with_correct_expected_version_second_time_with_same_initial_messages_then_should_have_expected_result"

                    
                    // tries to fix "When_append_stream_with_expected_version_and_duplicate_message_Id_then_should_throw"
                    command.CommandText = @"SELECT COUNT(*)
                                            FROM messages
                                            WHERE messages.stream_id_internal = @internalId
                                                AND messages.stream_version <= @expected
                                                AND messages.event_id = @eventId;";
                    command.Parameters.Clear();
                    command.Parameters.AddWithValue("@internalId", internalId);
                    command.Parameters.AddWithValue("@expected", expectedVersion);
                    command.Parameters.AddWithValue("@eventId", msg.MessageId);
                    duplicateMessage = command.ExecuteScalar<int?>(0);
                    if(duplicateMessage > 0)
                    {
                        throw new WrongExpectedVersionException(
                            ErrorMessages.AppendFailedWrongExpectedVersion(
                                streamId.IdOriginal,
                                expectedVersion),
                            streamId.IdOriginal,
                            expectedVersion);
                    }
                    // end - tries to fix "When_append_stream_with_expected_version_and_duplicate_message_Id_then_should_throw"
                    
                    command.CommandText = @"SELECT event_id
                            FROM messages
                            WHERE messages.stream_id_internal = @internalId
                                AND messages.stream_version >= @expected
                            ORDER BY messages.position;";
                    command.Parameters.Clear();
                    command.Parameters.AddWithValue("@internalId", internalId);
                    command.Parameters.AddWithValue("@expected", expectedVersion);

                    var eventIds = new List<Guid>();
                    using(var reader = command.ExecuteReader())
                    {
                        while(reader.Read())
                        {
                            eventIds.Add(reader.ReadScalar<Guid>(0, Guid.Empty));
                        }

                        eventIds.RemoveAll(x => x == Guid.Empty);
                    }
                    //
                    // if(!eventIds.Any())
                    // {
                    //     throw new WrongExpectedVersionException(
                    //         ErrorMessages.AppendFailedWrongExpectedVersion(
                    //             streamId.IdOriginal,
                    //             expectedVersion),
                    //         streamId.IdOriginal,
                    //         expectedVersion);
                    // }

                    if(eventIds.Contains(msg.MessageId))
                    {
                        if(eventIds.Count > messages.Length)
                        {
                            throw new WrongExpectedVersionException(
                                ErrorMessages.AppendFailedWrongExpectedVersion(
                                    streamId.IdOriginal,
                                    ExpectedVersion.NoStream),
                                streamId.IdOriginal,
                                ExpectedVersion.NoStream);<commit_msg>Removes commented code.<commit_after>namespace SqlStreamStore
{
    using System;
    using System.Collections.Generic;
    using System.Linq;
    using System.Threading;
    using System.Threading.Tasks;
    using Microsoft.Data.Sqlite;
    using SqlStreamStore.Infrastructure;
    using SqlStreamStore.Streams;
    using StreamStoreStore.Json;

    public partial class SqliteStreamStore
    {
        protected override async Task<AppendResult> AppendToStreamInternal(
            string streamId,
            int expectedVersion,
            NewStreamMessage[] messages,
            CancellationToken cancellationToken)
        {
            GuardAgainstDisposed();
            cancellationToken.ThrowIfCancellationRequested();
            
            SqliteAppendResult result;
            StreamIdInfo streamIdInfo = new StreamIdInfo(streamId);
            
            switch(expectedVersion)
            {
                case ExpectedVersion.Any:
                    result = await AppendToStreamAnyVersion(streamIdInfo.SqlStreamId, messages, cancellationToken).NotOnCapturedContext();
                    break;
                case ExpectedVersion.EmptyStream:
                    result = await AppendToStreamEmpty(streamIdInfo.SqlStreamId, messages, cancellationToken).NotOnCapturedContext();
                    break;
                case ExpectedVersion.NoStream:
                    result = await AppendToNonexistentStream(streamIdInfo.SqlStreamId, messages, cancellationToken).NotOnCapturedContext();
                    break;
                default:
                    result = AppendToStreamExpectedVersion(streamIdInfo.SqlStreamId, expectedVersion, messages, cancellationToken);
                    break;
            }

            if(result.MaxCount.HasValue)
            {
                await CheckStreamMaxCount(streamId, result.MaxCount, cancellationToken).NotOnCapturedContext();
            }

            await TryScavengeAsync(streamId, cancellationToken);

            return result;
        }

        private Task<SqliteAppendResult> AppendToStreamAnyVersion(SqliteStreamId streamId, NewStreamMessage[] messages, CancellationToken cancellationToken)
        {
            using(var conn = OpenConnection(false))
            using(var txn = conn.BeginTransaction())
            using(var cmd = conn.CreateCommand())
            {
                cmd.Transaction = txn;

                var stored = AppendToStreamAnyVersion(cmd, streamId, messages, cancellationToken);
                
                txn.Commit();
                
                return Task.FromResult(stored);
            }
        }

        private SqliteAppendResult AppendToStreamAnyVersion(
            SqliteCommand cmd,
            SqliteStreamId streamId,
            NewStreamMessage[] messages,
            CancellationToken cancellationToken)
        {
            var internalId = ResolveInternalStreamId(streamId.IdOriginal, cmd, false) ?? CreateStream(cmd, streamId);
            // unit tests state that if there is a single message, and it exists already, then
            // do not add the second message into the system.  Send the SqliteAppendResult back
            // as if you had done the insert.
            if(messages.Length == 1)
            {
                var msg = messages[0];

                // if the message's event id exists in the database...
                cmd.CommandText = @"SELECT count(*) 
                                        FROM messages
                                        WHERE event_id = @eventId AND stream_id_internal = @idInternal;";
                cmd.Parameters.Clear();
                cmd.Parameters.AddWithValue("@idInternal", internalId);
                cmd.Parameters.AddWithValue("@eventId", msg.MessageId);

                var existsInStream = cmd.ExecuteScalar<long>(0) > 0;
                if(existsInStream)
                {
                    cmd.CommandText = @"SELECT [version], [position]
                                        FROM streams
                                        WHERE id_internal = @idInternal;";
                    cmd.Parameters.Clear();
                    cmd.Parameters.AddWithValue("@idInternal", internalId);

                    using(var reader = cmd.ExecuteReader())
                    if(reader.Read())
                    {
                        var ver = reader.ReadScalar<int>(0);
                        var pos = reader.ReadScalar<long>(1);

                        {
                            return new SqliteAppendResult(ver, pos, null);
                        }
                    }
                }
            }
            else if (messages.Any())
            {
                var msg1 = messages.First();
                cmd.CommandText = @"SELECT event_id
                                    FROM messages
                                    WHERE [position] >= (SELECT [position] 
                                                      FROM messages 
                                                      WHERE messages.event_id = @messageId
                                                        AND messages.stream_id_internal = @internalId)
                                        AND stream_id_internal = @internalId;";
                cmd.Parameters.Clear();
                cmd.Parameters.AddWithValue("@messageId", msg1.MessageId);
                cmd.Parameters.AddWithValue("@internalId", internalId);
                cmd.Parameters.AddWithValue("@count", messages.Length);

                var eventIds = new List<Guid>();
                using(var reader = cmd.ExecuteReader())
                {
                    while(reader.Read())
                    {
                        eventIds.Add(reader.ReadScalar<Guid>(0, Guid.Empty));
                    }

                    eventIds.RemoveAll(x => x == Guid.Empty);
                }

                if(eventIds.Count > 0)
                {
                    for(var i = 0; i < Math.Min(eventIds.Count, messages.Length); i++)
                    {
                        if(eventIds[i] != messages[i].MessageId)
                        {
                            throw new WrongExpectedVersionException(
                                ErrorMessages.AppendFailedWrongExpectedVersion(
                                    streamId.IdOriginal,
                                    StreamVersion.Start),
                                streamId.IdOriginal,
                                StreamVersion.Start);
                        }
                    }

                    if(eventIds.Count < messages.Length && eventIds.Count > 0)
                    {
                        throw new WrongExpectedVersionException(
                            ErrorMessages.AppendFailedWrongExpectedVersion(
                                streamId.IdOriginal,
                                StreamVersion.Start),
                            streamId.IdOriginal,
                            StreamVersion.Start);
                    }

                    cmd.CommandText = @"SELECT [version], [position]
                                        FROM streams
                                        WHERE id_internal = @idInternal;";
                    cmd.Parameters.Clear();
                    cmd.Parameters.AddWithValue("@idInternal", internalId);

                    using(var reader = cmd.ExecuteReader())
                        if(reader.Read())
                        {
                            var ver = reader.ReadScalar<int>(0);
                            var pos = reader.ReadScalar<long>(1);

                            {
                                return new SqliteAppendResult(ver, pos, null);
                            }
                        }
                }
            }

            var stored = StoreMessages(messages, cmd, streamId);
            
            return stored;
        }

        private Task<SqliteAppendResult> AppendToStreamEmpty(SqliteStreamId streamId, NewStreamMessage[] messages, CancellationToken cancellationToken)
        {
            var internalId = ResolveInternalStreamId(streamId.IdOriginal);
            
            using(var conn = OpenConnection(false))
            using(var txn = conn.BeginTransaction())
            using(var cmd = conn.CreateCommand())
            {
                cmd.Transaction = txn;

                var lengthOfStream = GetStreamLength(cmd, internalId, out var result);
                if(lengthOfStream > StreamVersion.Start)
                {
                    throw new WrongExpectedVersionException(
                        ErrorMessages.AppendFailedWrongExpectedVersion(
                            streamId.IdOriginal,
                            StreamVersion.Start),
                        streamId.IdOriginal,
                        StreamVersion.Start);
                }

                var stored = StoreMessages(messages, cmd, streamId);
                
                txn.Commit();

                return Task.FromResult(stored);
            }
        }

        private static int GetStreamLength(SqliteCommand cmd, int? internalId, out SqliteAppendResult appendResult)
        {
            // check to see if the stream has records.  if so, throw wrongexpectedversion exception.
            cmd.CommandText = @"SELECT COUNT(*)
                                    FROM messages
                                    WHERE stream_id_internal = @internalId";
            cmd.Parameters.Clear();
            cmd.Parameters.AddWithValue("@internalId", internalId);
            var streamLength = StreamVersion.End;
            
            if(streamLength > 0)
            {
                // we has information in the stream.
                {
                    appendResult = new SqliteAppendResult(ExpectedVersion.NoStream,
                        ExpectedVersion.NoStream,
                        internalId ?? -1);
                    return streamLength;
                }
            }

            appendResult = default;
            return StreamVersion.Start;
        }

        private Task<SqliteAppendResult> AppendToNonexistentStream(SqliteStreamId streamId, NewStreamMessage[] messages, CancellationToken cancellationToken)
        {
            var internalId = ResolveInternalStreamId(streamId.IdOriginal, throwIfNotExists: false);
            if(internalId != null)
            {
                using(var connection = OpenConnection())
                using(var command = connection.CreateCommand())
                {
                    command.CommandText = @"SELECT event_id
                                    FROM messages
                                    WHERE messages.stream_id_internal = @internalId
                                    ORDER BY messages.position;";
                    command.Parameters.Clear();
                    command.Parameters.AddWithValue("@internalId", internalId);

                    var eventIds = new List<Guid>();
                    using(var reader = command.ExecuteReader())
                    {
                        while(reader.Read())
                        {
                            eventIds.Add(reader.ReadScalar<Guid>(0, Guid.Empty));
                        }

                        eventIds.RemoveAll(x => x == Guid.Empty);
                    }

                    if(eventIds.Count > 0)
                    {
                        for(var i = 0; i < Math.Min(eventIds.Count, messages.Length); i++)
                        {
                            if(eventIds[i] != messages[i].MessageId)
                            {
                                throw new WrongExpectedVersionException(
                                    ErrorMessages.AppendFailedWrongExpectedVersion(
                                        streamId.IdOriginal,
                                        ExpectedVersion.NoStream),
                                    streamId.IdOriginal,
                                    ExpectedVersion.NoStream);
                            }
                        }

                        if(eventIds.Count < messages.Length)
                        {
                            throw new WrongExpectedVersionException(
                                ErrorMessages.AppendFailedWrongExpectedVersion(
                                    streamId.IdOriginal,
                                    ExpectedVersion.NoStream),
                                streamId.IdOriginal,
                                ExpectedVersion.NoStream);
                        }

                        command.CommandText = @"SELECT [version], [position]
                                            FROM streams
                                            WHERE id_internal = @idInternal;";
                        command.Parameters.Clear();
                        command.Parameters.AddWithValue("@idInternal", internalId);

                        using(var reader = command.ExecuteReader())
                            if(reader.Read())
                            {
                                var ver = reader.ReadScalar<int>(0);
                                var pos = reader.ReadScalar<long>(1);

                                {
                                    return Task.FromResult(new SqliteAppendResult(ver, pos, null));
                                }
                            }
                    }
                }
            }

            CreateStream(streamId);
            
            return AppendToStreamEmpty(streamId, messages, cancellationToken);
        }

        private SqliteAppendResult AppendToStreamExpectedVersion(SqliteStreamId streamId, int expectedVersion, NewStreamMessage[] messages, CancellationToken cancellationToken)
        {
            var internalId = ResolveInternalStreamId(streamId.IdOriginal, throwIfNotExists: false);

            if(internalId == null)
            {
                throw new WrongExpectedVersionException(
                    ErrorMessages.AppendFailedWrongExpectedVersion(
                        streamId.IdOriginal,
                        expectedVersion),
                    streamId.IdOriginal,
                    expectedVersion);
            }
            
            using(var connection = OpenConnection(false))
            using(var transaction = connection.BeginTransaction())
            using(var command = connection.CreateCommand())
            {
                command.Transaction = transaction;
                
                if(messages.Length == 1)
                {
                    var msg = messages.First();
                    
                    // tries to fix "When_append_single_message_to_stream_with_correct_expected_version_second_time_with_same_initial_messages_then_should_have_expected_result"
                    command.CommandText = @"SELECT COUNT(*)
                                            FROM messages
                                            WHERE messages.stream_id_internal = @internalId
                                                AND messages.stream_version = @expected
                                                AND messages.event_id = @eventId;";
                    command.Parameters.Clear();
                    command.Parameters.AddWithValue("@internalId", internalId);
                    command.Parameters.AddWithValue("@expected", expectedVersion + 1);
                    command.Parameters.AddWithValue("@eventId", msg.MessageId);
                    var duplicateMessage = command.ExecuteScalar<int?>(0);
                    if(duplicateMessage > 0)
                    {
                        command.CommandText = @"SELECT messages.stream_version, messages.[position]
                                        FROM messages
                                        WHERE messages.stream_id_internal = @stream_id_internal
                                        ORDER BY messages.[position] DESC
                                        LIMIT 1";
                        command.Parameters.Clear();
                        command.Parameters.AddWithValue("@stream_id_internal", internalId ?? -1);
                        using(var reader = command.ExecuteReader())
                        {
                            if (reader.Read() && reader.HasRows)
                                return new SqliteAppendResult(
                                    reader.ReadScalar(0, StreamVersion.End),
                                    reader.ReadScalar(1, Position.End),
                                    internalId ?? -1
                                );
                        }
                    }
                    // end - tries to fix "When_append_single_message_to_stream_with_correct_expected_version_second_time_with_same_initial_messages_then_should_have_expected_result"

                    
                    // tries to fix "When_append_stream_with_expected_version_and_duplicate_message_Id_then_should_throw"
                    command.CommandText = @"SELECT COUNT(*)
                                            FROM messages
                                            WHERE messages.stream_id_internal = @internalId
                                                AND messages.stream_version <= @expected
                                                AND messages.event_id = @eventId;";
                    command.Parameters.Clear();
                    command.Parameters.AddWithValue("@internalId", internalId);
                    command.Parameters.AddWithValue("@expected", expectedVersion);
                    command.Parameters.AddWithValue("@eventId", msg.MessageId);
                    duplicateMessage = command.ExecuteScalar<int?>(0);
                    if(duplicateMessage > 0)
                    {
                        throw new WrongExpectedVersionException(
                            ErrorMessages.AppendFailedWrongExpectedVersion(
                                streamId.IdOriginal,
                                expectedVersion),
                            streamId.IdOriginal,
                            expectedVersion);
                    }
                    // end - tries to fix "When_append_stream_with_expected_version_and_duplicate_message_Id_then_should_throw"
                    
                    command.CommandText = @"SELECT event_id
                            FROM messages
                            WHERE messages.stream_id_internal = @internalId
                                AND messages.stream_version >= @expected
                            ORDER BY messages.position;";
                    command.Parameters.Clear();
                    command.Parameters.AddWithValue("@internalId", internalId);
                    command.Parameters.AddWithValue("@expected", expectedVersion);

                    var eventIds = new List<Guid>();
                    using(var reader = command.ExecuteReader())
                    {
                        while(reader.Read())
                        {
                            eventIds.Add(reader.ReadScalar<Guid>(0, Guid.Empty));
                        }

                        eventIds.RemoveAll(x => x == Guid.Empty);
                    }

                    if(eventIds.Contains(msg.MessageId))
                    {
                        if(eventIds.Count > messages.Length)
                        {
                            throw new WrongExpectedVersionException(
                                ErrorMessages.AppendFailedWrongExpectedVersion(
                                    streamId.IdOriginal,
                                    ExpectedVersion.NoStream),
                                streamId.IdOriginal,
                                ExpectedVersion.NoStream);
{'fraction_non_alphanumeric': 0.06867166670819524, 'fraction_numerical': 0.0016445319313283333, 'mean_word_length': 1.1117600631412785, 'pattern_counts': {'":': 0, '<': 51, '<?xml version=': 0, '>': 69, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 146, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4225373', 'n_tokens_mistral': 8371, 'n_tokens_neox': 7799, 'n_words': 1510}