<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.Streams;

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

            // 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.
            var maxRecords = count == int.MaxValue ? count - 1 : count;
            var streamVersion = fromStreamVersion;
            if(streamVersion < StreamVersion.Start) streamVersion = StreamVersion.Start;
            
            using (var connection = OpenConnection())
            using (var command = connection.CreateCommand())
            {
                var idInfo = new StreamIdInfo(streamId);
                int streamIdInternal = 0;
                long lastStreamPosition = 0;
                var maxAge = default(int?);
                command.CommandText = @"SELECT streams.id_internal, 
                                                streams.[version], 
                                                streams.[position], 
                                                coalesce(max_age, (SELECT s.max_age FROM streams s WHERE s.id = @metaId)), 
                                                coalesce(max_count, (SELECT s.max_count FROM streams s WHERE s.id = @metaId))
                                        FROM streams
                                        where streams.id = @streamId";
                command.Parameters.Clear();
                command.Parameters.AddWithValue("@streamId", idInfo.SqlStreamId.Id);
                command.Parameters.AddWithValue("@metaId", idInfo.MetadataSqlStreamId.Id);
                using(var reader = command.ExecuteReader())
                {
                    if(!reader.Read())
                    {
                        // not found.
                        return Task.FromResult(new ReadStreamPage(
                            streamId,
                            PageReadStatus.StreamNotFound,
                            fromStreamVersion,
                            StreamVersion.End,
                            StreamVersion.End,
                            StreamVersion.End,
                            ReadDirection.Forward,
                            true,
                            readNext));
                    }

                    streamIdInternal = reader.ReadScalar<int>(0);
                    lastStreamPosition = reader.ReadScalar<int>(2);
                    maxAge = reader.ReadScalar(3, default(int?));

                    var streamsMaxRecords = reader.ReadScalar(4, StreamVersion.Start);
                    maxRecords = Math.Min(maxRecords, (streamsMaxRecords == StreamVersion.Start ? maxRecords : streamsMaxRecords));
                }

                long? position;
                // command.CommandText = @"SELECT messages.position
                //                         FROM messages
                //                         WHERE messages.stream_id_internal = @streamIdInternal
                //                             AND messages.stream_version = @streamVersion;";
                // command.Parameters.Clear();
                // command.Parameters.AddWithValue("@streamIdInternal", streamIdInternal);
                // command.Parameters.AddWithValue("@streamVersion", streamVersion);
                // position = command.ExecuteScalar<long?>();

                // if(position == null)
                // {
                    command.CommandText = @"SELECT MIN(messages.position)
                                            FROM messages
                                            WHERE messages.stream_id_internal = @streamIdInternal
                                                AND messages.stream_version >= @streamVersion;";
                    command.Parameters.Clear();
                    command.Parameters.AddWithValue("@streamIdInternal", streamIdInternal);
                    command.Parameters.AddWithValue("@streamVersion", streamVersion);
                    position = command.ExecuteScalar<long?>() ?? Position.Start;
                // }

                command.CommandText = @"SELECT COUNT(*)
                        FROM messages <commit_msg>Fixes issue where the read position should initially come from the messages table, and if it is not discoverable from there, to go into the streams table and extract the information.<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.Streams;

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

            // 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.
            var maxRecords = count == int.MaxValue ? count - 1 : count;
            var streamVersion = fromStreamVersion;
            if(streamVersion < StreamVersion.Start) streamVersion = StreamVersion.Start;
            
            using (var connection = OpenConnection())
            using (var command = connection.CreateCommand())
            {
                var idInfo = new StreamIdInfo(streamId);
                int streamIdInternal = 0;
                long lastStreamPosition = 0;
                var maxAge = default(int?);
                command.CommandText = @"SELECT streams.id_internal, 
                                                streams.[version], 
                                                streams.[position], 
                                                coalesce(max_age, (SELECT s.max_age FROM streams s WHERE s.id = @metaId)), 
                                                coalesce(max_count, (SELECT s.max_count FROM streams s WHERE s.id = @metaId))
                                        FROM streams
                                        where streams.id = @streamId";
                command.Parameters.Clear();
                command.Parameters.AddWithValue("@streamId", idInfo.SqlStreamId.Id);
                command.Parameters.AddWithValue("@metaId", idInfo.MetadataSqlStreamId.Id);
                using(var reader = command.ExecuteReader())
                {
                    if(!reader.Read())
                    {
                        // not found.
                        return Task.FromResult(new ReadStreamPage(
                            streamId,
                            PageReadStatus.StreamNotFound,
                            fromStreamVersion,
                            StreamVersion.End,
                            StreamVersion.End,
                            StreamVersion.End,
                            ReadDirection.Forward,
                            true,
                            readNext));
                    }

                    streamIdInternal = reader.ReadScalar<int>(0);
                    lastStreamPosition = reader.ReadScalar<int>(2);
                    maxAge = reader.ReadScalar(3, default(int?));

                    var streamsMaxRecords = reader.ReadScalar(4, StreamVersion.Start);
                    maxRecords = Math.Min(maxRecords, (streamsMaxRecords == StreamVersion.Start ? maxRecords : streamsMaxRecords));
                }

                long? position;
                command.CommandText = @"SELECT messages.position
                                        FROM messages
                                        WHERE messages.stream_id_internal = @streamIdInternal
                                            AND messages.stream_version = @streamVersion;";
                command.Parameters.Clear();
                command.Parameters.AddWithValue("@streamIdInternal", streamIdInternal);
                command.Parameters.AddWithValue("@streamVersion", streamVersion);
                position = command.ExecuteScalar<long?>();

                if(position == null)
                {
                    command.CommandText = @"SELECT MIN(messages.position)
                                            FROM messages
                                            WHERE messages.stream_id_internal = @streamIdInternal
                                                AND messages.stream_version >= @streamVersion;";
                    command.Parameters.Clear();
                    command.Parameters.AddWithValue("@streamIdInternal", streamIdInternal);
                    command.Parameters.AddWithValue("@streamVersion", streamVersion);
                    position = command.ExecuteScalar<long?>() ?? Position.Start;
                }

                command.CommandText = @"SELECT COUNT(*)
                        FROM messages 
{'fraction_non_alphanumeric': 0.0743075651095494, 'fraction_numerical': 0.0014468788755684168, 'mean_word_length': 1.305694543721706, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 45, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2007953', 'n_tokens_mistral': 2029, 'n_tokens_neox': 1911, 'n_words': 450}