<NME> SqliteStreamStore.Read.cs
<BEF> namespace SqlStreamStore
{
    using System;
    using System.Collections.Generic;
    using System.Data;
    using System.Linq;
    using System.Threading;
    using System.Threading.Tasks;

    public partial class SqliteStreamStore
    {
        protected override async Task<ReadStreamPage> ReadStreamForwardsInternal(
            string streamId,
            int fromVersion,
            int count,
            bool prefetch,
            ReadNextStreamPage readNext,
            CancellationToken cancellationToken)
        {
            GuardAgainstDisposed();
            cancellationToken.ThrowIfCancellationRequested();

            using(var connection = OpenConnection())
            {
                var streamProperties = await connection.Streams(streamId)
                    .Properties(initializeIfNotFound:false, cancellationToken);

                if(streamProperties == null)
                {
                    // not found.
                    return new ReadStreamPage(
                        streamId,
                        PageReadStatus.StreamNotFound,
                        fromVersion,
                        StreamVersion.End,
                        StreamVersion.End,
                        StreamVersion.End,
                        ReadDirection.Forward,
                        true,
                        readNext);
                }
                
                return await PrepareStreamResponse(connection,
                    streamId,
                    ReadDirection.Forward,
                    fromVersion,
                    prefetch,
                    readNext,
                    count,
                    streamProperties.MaxAge);
            }
        }

        protected override async Task<ReadStreamPage> ReadStreamBackwardsInternal(
            string streamId,
            int fromStreamVersion,
            int count,
            bool prefetch,
            ReadNextStreamPage readNext,
            CancellationToken cancellationToken)
        {
            GuardAgainstDisposed();
            cancellationToken.ThrowIfCancellationRequested();

            var streamVersion = fromStreamVersion == StreamVersion.End ? int.MaxValue - 1 : fromStreamVersion;
            using(var connection = OpenConnection())
            {
                var streamProperties = await connection.Streams(streamId)
                    .Properties(initializeIfNotFound: false, cancellationToken);

                if(streamProperties == null)
                {
                    // not found.
                    return new ReadStreamPage(
                        streamId,
                        PageReadStatus.StreamNotFound,
                        fromStreamVersion,
                        StreamVersion.End,
                        StreamVersion.End,
                        StreamVersion.End,
                        ReadDirection.Forward,
                        true,
                        readNext);
                }

                var position = connection.Streams(streamId)
                    .AllStreamPosition(ReadDirection.Backward, streamVersion);

                // if no position, then need to return success with end of stream.
                if(position == null)
                {
                    // not found.
                    return new ReadStreamPage(
                        streamId,
                        PageReadStatus.Success,

                using(var command = connection.CreateCommand())
                {
                    command.CommandText = @"SELECT messages.position
                                FROM messages
                                WHERE messages.stream_id_internal = @idOriginal
                                    AND messages.stream_version <= @streamVersion
                                ORDER BY messages.position DESC
                                LIMIT 1;";
                    command.Parameters.Clear();
                    command.Parameters.AddWithValue("@idOriginal", streamProperties.Key);
                    command.Parameters.AddWithValue("@streamVersion", streamVersion);
                    var position = command.ExecuteScalar<long?>();

                    if(position == null)
                    {
                        command.CommandText = @"SELECT streams.position
                            FROM streams
                            WHERE streams.id_internal = @idOriginal;";
                        command.Parameters.Clear();
                        command.Parameters.AddWithValue("@idOriginal", streamProperties.Key);
                        position = command.ExecuteScalar<long?>();
                    }

                    // if no position, then need to return success with end of stream.
                    if(position == null)
        {
            // If the count is int.MaxValue, TSql will see it as a negative number. 
            // Users shouldn't be using int.MaxValue in the first place anyway.
            maxRecords = maxRecords == int.MaxValue ? maxRecords - 1 : maxRecords;
            var streamVersion = fromVersion == StreamVersion.End ? int.MaxValue -1 : fromVersion;
            int nextVersion = 0;
            var stream = connection.Streams(streamId);

            var header = stream
                .Properties()
                .GetAwaiter().GetResult();

            var position = stream
                .AllStreamPosition(direction, streamVersion)
                .GetAwaiter().GetResult();

            var remaining = stream
                .Length(direction, position, CancellationToken.None)
                .GetAwaiter().GetResult();

            var messages = stream
                .Read(direction, position, prefetch, maxRecords)
                .GetAwaiter().GetResult()
                .Select(Message => (Message, maxAge))
                .ToList();

            var filtered = FilterExpired(messages);

            var isEnd = remaining - messages.Count <= 0;

            if(direction == ReadDirection.Forward)
            {
                if(messages.Any())
                {
                    nextVersion = messages.Last().Message.StreamVersion + 1;
                }
            int? maxAge)
        {
            var streamVersion = fromVersion == StreamVersion.End ? int.MaxValue -1 : fromVersion;
            var lastPosition = 0L;
            int lastVersion = 0;
            int nextVersion = 0;

            command.CommandText = @"SELECT [position], [version]
                                    FROM streams
                                    WHERE streams.id_internal = @idInternal;";
            command.Parameters.Clear();
            command.Parameters.AddWithValue("@idInternal", streamIdInternal);
            using(var reader = command.ExecuteReader(CommandBehavior.SingleRow))
            {
                if(reader.Read())
                {
                    lastPosition = reader.ReadScalar<long>(0);
                    lastVersion = reader.ReadScalar<int>(1);
                }
            }

            var position = command.Connection.Streams(streamId)
                .AllStreamPosition(direction, streamVersion)

                if(messages.Any())
                {
                    nextVersion = messages.Last().Message.StreamVersion - 1;
                }
            }

            var page = new ReadStreamPage(
                streamId,
                status: PageReadStatus.Success,
                fromStreamVersion: fromVersion,
                nextStreamVersion: nextVersion,
                lastStreamVersion: header.Version,
                lastStreamPosition: header.Position,
                direction: direction,
                isEnd: isEnd,
                readNext: readNext,
                messages: filtered.ToArray());

            return Task.FromResult(page);
        }
    }
}                }
                else
                {
                    nextVersion = lastVersion + 1;
                }
            }
            else if (direction == ReadDirection.Backward)
                status: PageReadStatus.Success,
                fromStreamVersion: fromVersion,
                nextStreamVersion: nextVersion,
                lastStreamVersion: lastVersion,
                lastStreamPosition: lastPosition,
                direction: direction,
                isEnd: isEnd,
                readNext: readNext,

<MSG> Updates read prior position code to use the centralized 'header' function.

<DFF> @@ -2,7 +2,6 @@ namespace SqlStreamStore
 {
     using System;
     using System.Collections.Generic;
-    using System.Data;
     using System.Linq;
     using System.Threading;
     using System.Threading.Tasks;
@@ -96,26 +95,29 @@ namespace SqlStreamStore
 
                 using(var command = connection.CreateCommand())
                 {
-                    command.CommandText = @"SELECT messages.position
-                                FROM messages
-                                WHERE messages.stream_id_internal = @idOriginal
-                                    AND messages.stream_version <= @streamVersion
-                                ORDER BY messages.position DESC
-                                LIMIT 1;";
-                    command.Parameters.Clear();
-                    command.Parameters.AddWithValue("@idOriginal", streamProperties.Key);
-                    command.Parameters.AddWithValue("@streamVersion", streamVersion);
-                    var position = command.ExecuteScalar<long?>();
+                    // command.CommandText = @"SELECT messages.position
+                    //             FROM messages
+                    //             WHERE messages.stream_id_internal = @idOriginal
+                    //                 AND messages.stream_version <= @streamVersion
+                    //             ORDER BY messages.position DESC
+                    //             LIMIT 1;";
+                    // command.Parameters.Clear();
+                    // command.Parameters.AddWithValue("@idOriginal", streamProperties.Key);
+                    // command.Parameters.AddWithValue("@streamVersion", streamVersion);
+                    // var position = command.ExecuteScalar<long?>();
+                    //
+                    // if(position == null)
+                    // {
+                    //     command.CommandText = @"SELECT streams.position
+                    //         FROM streams
+                    //         WHERE streams.id_internal = @idOriginal;";
+                    //     command.Parameters.Clear();
+                    //     command.Parameters.AddWithValue("@idOriginal", streamProperties.Key);
+                    //     position = command.ExecuteScalar<long?>();
+                    // }
 
-                    if(position == null)
-                    {
-                        command.CommandText = @"SELECT streams.position
-                            FROM streams
-                            WHERE streams.id_internal = @idOriginal;";
-                        command.Parameters.Clear();
-                        command.Parameters.AddWithValue("@idOriginal", streamProperties.Key);
-                        position = command.ExecuteScalar<long?>();
-                    }
+                    var position = command.Connection.Streams(streamId)
+                        .AllStreamPosition(ReadDirection.Backward, streamVersion);
 
                     // if no position, then need to return success with end of stream.
                     if(position == null)
@@ -158,23 +160,11 @@ namespace SqlStreamStore
             int? maxAge)
         {
             var streamVersion = fromVersion == StreamVersion.End ? int.MaxValue -1 : fromVersion;
-            var lastPosition = 0L;
-            int lastVersion = 0;
             int nextVersion = 0;
 
-            command.CommandText = @"SELECT [position], [version]
-                                    FROM streams
-                                    WHERE streams.id_internal = @idInternal;";
-            command.Parameters.Clear();
-            command.Parameters.AddWithValue("@idInternal", streamIdInternal);
-            using(var reader = command.ExecuteReader(CommandBehavior.SingleRow))
-            {
-                if(reader.Read())
-                {
-                    lastPosition = reader.ReadScalar<long>(0);
-                    lastVersion = reader.ReadScalar<int>(1);
-                }
-            }
+            var header = command.Connection.Streams(streamId)
+                .Properties()
+                .GetAwaiter().GetResult();
 
             var position = command.Connection.Streams(streamId)
                 .AllStreamPosition(direction, streamVersion)
@@ -234,7 +224,7 @@ namespace SqlStreamStore
                 }
                 else
                 {
-                    nextVersion = lastVersion + 1;
+                    nextVersion = header.Version + 1;
                 }
             }
             else if (direction == ReadDirection.Backward)
@@ -255,8 +245,8 @@ namespace SqlStreamStore
                 status: PageReadStatus.Success,
                 fromStreamVersion: fromVersion,
                 nextStreamVersion: nextVersion,
-                lastStreamVersion: lastVersion,
-                lastStreamPosition: lastPosition,
+                lastStreamVersion: header.Version,
+                lastStreamPosition: header.Position,
                 direction: direction,
                 isEnd: isEnd,
                 readNext: readNext,

{'addition_count': 28, 'commit_subject': "Updates read prior position code to use the centralized 'header' function.", 'deletion_count': 38, 'file_extension': '.cs', 'lang': 'Sqlite/SqliteStreamStore', 'license': 'mit', 'repo_name': 'SQLStreamStore/SQLStreamStore'}