// Copyright 2019-2026 The NATS Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// Adapted from server/filestore.go (msgBlock struct and consumerFileStore struct)

using System.Text.Json;
using System.Text;
using System.Threading.Channels;
using System.Buffers.Binary;
using System.Security.Cryptography;
using ZB.MOM.NatsNet.Server.Internal.DataStructures;

namespace ZB.MOM.NatsNet.Server;

// ---------------------------------------------------------------------------
// MessageBlock
// ---------------------------------------------------------------------------

/// <summary>
/// Represents a single on-disk message block file together with its
/// in-memory cache and index state.
/// Mirrors the <c>msgBlock</c> struct in filestore.go.
/// </summary>
internal sealed class MessageBlock
{
    private const int MaxVarIntLength = 10;
    private static readonly Exception ErrNoCache = new InvalidOperationException("no message cache");
    private static readonly Exception ErrPartialCache = new InvalidOperationException("partial cache");
    private static readonly Exception ErrDeletedMsg = new InvalidOperationException("deleted message");

    // ------------------------------------------------------------------
    // Identity fields — first/last use volatile-style access in Go via
    // atomic.LoadUint64 on the embedded msgId structs.
    // We replicate those as plain fields; callers must acquire _mu before
    // reading/writing unless using the Interlocked helpers on Seq/Ts.
    // ------------------------------------------------------------------

    /// <summary>First message in this block (sequence + nanosecond timestamp).</summary>
    public MsgId First;

    /// <summary>Last message in this block (sequence + nanosecond timestamp).</summary>
    public MsgId Last;

    // ------------------------------------------------------------------
    // Lock
    // ------------------------------------------------------------------

    /// <summary>Guards all mutable fields in this block.</summary>
    public readonly ReaderWriterLockSlim Mu = new(LockRecursionPolicy.NoRecursion);

    // ------------------------------------------------------------------
    // Back-reference
    // ------------------------------------------------------------------

    /// <summary>Owning file store.</summary>
    public JetStreamFileStore? Fs { get; set; }

    // ------------------------------------------------------------------
    // File I/O
    // ------------------------------------------------------------------

    /// <summary>Path to the <c>.blk</c> message data file.</summary>
    public string Mfn { get; set; } = string.Empty;

    /// <summary>Open file stream for the block data file (null when closed/idle).</summary>
    public FileStream? Mfd { get; set; }

    /// <summary>Path to the per-block encryption key file.</summary>
    public string Kfn { get; set; } = string.Empty;

    /// <summary>Decrypted per-block seed material (when encryption is enabled).</summary>
    public byte[]? Seed { get; set; }

    /// <summary>Per-block nonce recovered from the encrypted key payload.</summary>
    public byte[]? Nonce { get; set; }

    /// <summary>Per-block AEAD key derived from <see cref="Seed"/>.</summary>
    public JetStreamFileStore.AeadCipher? Aek { get; set; }

    /// <summary>Per-block stream cipher derived from <see cref="Seed"/> and <see cref="Nonce"/>.</summary>
    public JetStreamFileStore.XorStreamCipher? Bek { get; set; }

    // ------------------------------------------------------------------
    // Compression
    // ------------------------------------------------------------------

    /// <summary>Effective compression algorithm when the block was last loaded.</summary>
    public StoreCompression Cmp { get; set; }

    /// <summary>
    /// Last index write size in bytes; used to detect whether re-compressing
    /// the block would save meaningful space.
    /// </summary>
    public long Liwsz { get; set; }

    // ------------------------------------------------------------------
    // Block identity
    // ------------------------------------------------------------------

    /// <summary>Monotonically increasing block index number (used in file names).</summary>
    public uint Index { get; set; }

    // ------------------------------------------------------------------
    // Counters
    // ------------------------------------------------------------------

    /// <summary>User-visible byte count (excludes deleted-message bytes).</summary>
    public ulong Bytes { get; set; }

    /// <summary>
    /// Total raw byte count including deleted messages.
    /// Used to decide when to roll to a new block.
    /// </summary>
    public ulong RBytes { get; set; }

    /// <summary>Byte count captured at the last compaction (0 if never compacted).</summary>
    public ulong CBytes { get; set; }

    /// <summary>User-visible message count (excludes deleted messages).</summary>
    public ulong Msgs { get; set; }

    // ------------------------------------------------------------------
    // Per-subject state
    // ------------------------------------------------------------------

    /// <summary>
    /// Optional per-subject state tree for this block.
    /// Lazily populated and expired when idle.
    /// Mirrors <c>mb.fss</c> in filestore.go.
    /// </summary>
    public SubjectTree<SimpleState>? Fss { get; set; }

    // ------------------------------------------------------------------
    // Deleted-sequence tracking
    // ------------------------------------------------------------------

    /// <summary>
    /// Set of deleted sequence numbers within this block.
    /// Uses the AVL-backed <see cref="SequenceSet"/> to match Go's <c>avl.SequenceSet</c>.
    /// </summary>
    public SequenceSet Dmap { get; set; } = new();

    // ------------------------------------------------------------------
    // Timestamps (nanosecond Unix times, matches Go int64)
    // ------------------------------------------------------------------

    /// <summary>Nanosecond timestamp of the last write to this block.</summary>
    public long Lwts { get; set; }

    /// <summary>Nanosecond timestamp of the last load (cache fill) of this block.</summary>
    public long Llts { get; set; }

    /// <summary>Nanosecond timestamp of the last read from this block.</summary>
    public long Lrts { get; set; }

    /// <summary>Nanosecond timestamp of the last subject-state (fss) access.</summary>
    public long Lsts { get; set; }

    /// <summary>Last sequence that was looked up; used to detect linear scans.</summary>
    public ulong Llseq { get; set; }

    // ------------------------------------------------------------------
    // Cache
    // ------------------------------------------------------------------

    /// <summary>
    /// Active in-memory cache. May be null when evicted.
    /// Mirrors <c>mb.cache</c>; the elastic-pointer field (<c>mb.ecache</c>) is
    /// not ported — a plain nullable field is sufficient here.
    /// </summary>
    public Cache? CacheData { get; set; }

    /// <summary>Number of times the cache has been (re)loaded from disk.</summary>
    public ulong Cloads { get; set; }

    /// <summary>Cache buffer expiration duration for this block.</summary>
    public TimeSpan Cexp { get; set; }

    /// <summary>Per-block subject-state (fss) expiration duration.</summary>
    public TimeSpan Fexp { get; set; }

    /// <summary>Timer used to expire the cache buffer when idle.</summary>
    public Timer? Ctmr { get; set; }

    // ------------------------------------------------------------------
    // State flags
    // ------------------------------------------------------------------

    /// <summary>Whether the in-memory cache is currently populated.</summary>
    public bool HaveCache => CacheData != null;

    /// <summary>Whether this block has been closed and must not be written to.</summary>
    public bool Closed { get; set; }

    /// <summary>Whether this block has unflushed data that must be synced to disk.</summary>
    public bool NeedSync { get; set; }

    /// <summary>When true every write is immediately synced (SyncAlways mode).</summary>
    public bool SyncAlways { get; set; }

    /// <summary>When true compaction is suppressed for this block.</summary>
    public bool NoCompact { get; set; }

    /// <summary>
    /// When true the block's messages are not tracked in the per-subject index.
    /// Used for blocks that only contain tombstone or deleted markers.
    /// </summary>
    public bool NoTrack { get; set; }

    /// <summary>Whether a background flusher goroutine equivalent is running.</summary>
    public bool Flusher { get; set; }

    /// <summary>Whether a cache-load is currently in progress.</summary>
    public bool Loading { get; set; }

    // ------------------------------------------------------------------
    // Write error
    // ------------------------------------------------------------------

    /// <summary>Captured write error; non-null means the block is in a bad state.</summary>
    public Exception? Werr { get; set; }

    // ------------------------------------------------------------------
    // TTL / scheduling counters
    // ------------------------------------------------------------------

    /// <summary>Number of messages in this block that have a TTL set.</summary>
    public ulong Ttls { get; set; }

    /// <summary>Number of messages in this block that have a schedule set.</summary>
    public ulong Schedules { get; set; }

    // ------------------------------------------------------------------
    // Channels (equivalent to Go chan struct{})
    // ------------------------------------------------------------------

    /// <summary>Flush-request channel: signals the flusher to run.</summary>
    public Channel<byte>? Fch { get; set; }

    /// <summary>Quit channel: closing signals background goroutine equivalents to stop.</summary>
    public Channel<byte>? Qch { get; set; }

    // ------------------------------------------------------------------
    // Checksum
    // ------------------------------------------------------------------

    /// <summary>
    /// Last-check hash: 8-byte rolling checksum of the last validated record.
    /// Mirrors <c>mb.lchk [8]byte</c>.
    /// </summary>
    public byte[] Lchk { get; set; } = new byte[8];

    // ------------------------------------------------------------------
    // Test hook
    // ------------------------------------------------------------------

    /// <summary>When true, simulates a write failure. Used by unit tests only.</summary>
    public bool MockWriteErr { get; set; }

    internal Exception? CloseFDs()
    {
        if (Mu.IsWriteLockHeld)
            return CloseFDsLocked();

        Mu.EnterWriteLock();
        try
        {
            return CloseFDsLocked();
        }
        finally
        {
            Mu.ExitWriteLock();
        }
    }

    internal Exception? CloseFDsLocked()
    {
        var (pending, pendingErr) = BytesPending();
        if (pendingErr == null && pending != null && pending.Length > 0)
            return new InvalidOperationException("pending data");

        CloseFDsLockedNoCheck();
        return null;
    }

    internal void CloseFDsLockedNoCheck()
    {
        Mfd?.Dispose();
        Mfd = null;
    }

    internal Exception? CheckAndLoadEncryption()
    {
        var fs = Fs;
        if (fs == null)
            return null;

        if (Aek != null && Bek != null)
            return null;

        try
        {
            fs.LoadEncryptionForMsgBlock(this);
            return null;
        }
        catch (Exception ex)
        {
            return ex;
        }
    }

    internal void EnsureLastChecksumLoaded()
    {
        if (HasChecksum(Lchk))
            return;

        var lchk = LastChecksum();
        if (lchk.Length == FileStoreDefaults.RecordHashSize)
            Lchk = lchk;
    }

    internal Exception? ConvertCipher()
    {
        var fs = Fs;
        if (fs == null)
            return new InvalidOperationException("message block has no parent file store");

        var checkErr = CheckAndLoadEncryption();
        if (checkErr != null)
            return checkErr;

        if (Bek == null)
            return null;

        byte[] encrypted;
        try
        {
            encrypted = File.Exists(Mfn) ? File.ReadAllBytes(Mfn) : Array.Empty<byte>();
        }
        catch (Exception ex)
        {
            return ex;
        }

        if (encrypted.Length == 0)
            return null;

        var plaintext = (byte[])encrypted.Clone();
        Bek.XorKeyStream(plaintext.AsSpan());

        if (!TryValidateRecordBuffer(plaintext.AsSpan(), out _))
            return new InvalidDataException("unable to recover keys");

        var regenErr = fs.RegenerateEncryptionKeysForBlock(this);
        if (regenErr != null)
            return regenErr;

        var reloadErr = CheckAndLoadEncryption();
        if (reloadErr != null)
            return reloadErr;

        if (Bek == null)
            return new InvalidOperationException("missing block encryption key after regeneration");

        Bek.XorKeyStream(plaintext.AsSpan());

        try
        {
            File.WriteAllBytes(Mfn, plaintext);
            return null;
        }
        catch (Exception ex)
        {
            return ex;
        }
    }

    internal Exception? ConvertToEncrypted()
    {
        var checkErr = CheckAndLoadEncryption();
        if (checkErr != null)
            return checkErr;

        if (Bek == null)
            return null;

        byte[] buf;
        try
        {
            buf = File.Exists(Mfn) ? File.ReadAllBytes(Mfn) : Array.Empty<byte>();
        }
        catch (Exception ex)
        {
            return ex;
        }

        if (buf.Length == 0)
            return null;

        if (!TryValidateRecordBuffer(buf.AsSpan(), out _))
            return new InvalidDataException("unable to convert block to encrypted format");

        Bek.XorKeyStream(buf.AsSpan());

        try
        {
            File.WriteAllBytes(Mfn, buf);
            return null;
        }
        catch (Exception ex)
        {
            return ex;
        }
    }

    internal uint GetIndex()
    {
        Mu.EnterReadLock();
        try
        {
            return Index;
        }
        finally
        {
            Mu.ExitReadLock();
        }
    }

    internal (LostStreamData? LostData, ulong[] Tombstones, Exception? Error) RebuildState()
    {
        Mu.EnterWriteLock();
        try
        {
            return RebuildStateLocked();
        }
        finally
        {
            Mu.ExitWriteLock();
        }
    }

    internal (LostStreamData? LostData, ulong[] Tombstones, Exception? Error) RebuildStateLocked()
    {
        var startFirst = First.Seq;
        var startLast = Last.Seq;
        var previousDeleted = SnapshotDeletedSequences();

        TryForceExpireCacheLocked();
        Fss = null;

        byte[] buf;
        try
        {
            buf = File.Exists(Mfn) ? File.ReadAllBytes(Mfn) : Array.Empty<byte>();
            RBytes = (ulong)buf.Length;
        }
        catch (Exception ex)
        {
            return (null, Array.Empty<ulong>(), ex);
        }

        if (buf.Length == 0)
        {
            var ld = BuildLostData(startFirst, startLast, previousDeleted, bytes: Bytes);

            Msgs = 0;
            Bytes = 0;
            RBytes = 0;
            Dmap.Empty();
            First = new MsgId { Seq = startLast + 1, Ts = 0 };
            Last = new MsgId { Seq = startLast, Ts = 0 };

            return (ld, Array.Empty<ulong>(), null);
        }

        var encErr = CheckAndLoadEncryption();
        if (encErr != null)
            return (null, Array.Empty<ulong>(), encErr);

        if (Bek != null)
            Bek.XorKeyStream(buf.AsSpan());

        return RebuildStateFromBufLocked(buf, allowTruncate: true);
    }

    internal (LostStreamData? LostData, ulong[] Tombstones, Exception? Error) RebuildStateFromBufLocked(byte[] buf, bool allowTruncate)
    {
        ArgumentNullException.ThrowIfNull(buf);

        var startLast = Last.Seq;
        var tombstones = new List<ulong>();
        var oldDeleted = SnapshotDeletedSequences();

        Msgs = 0;
        Bytes = 0;
        RBytes = (ulong)buf.Length;
        First = new MsgId { Seq = 0, Ts = 0 };
        Last = new MsgId { Seq = 0, Ts = 0 };
        Dmap.Empty();

        ulong highestSeq = 0;
        ulong maxTombstoneSeq = 0;
        long maxTombstoneTs = 0;

        var index = 0;
        while (index < buf.Length)
        {
            if (!TryReadRecordHeader(buf.AsSpan(index), out var seqRaw, out var ts, out var subjectLength, out var headerLength, out var msgLength, out var recordLength, out var headerError))
            {
                var ld = BuildLostData(Last.Seq + 1, startLast, oldDeleted, bytes: (ulong)(buf.Length - index));
                return (ld, tombstones.ToArray(), headerError ?? new InvalidDataException("corrupt message header"));
            }

            if (recordLength <= 0 || index + recordLength > buf.Length)
            {
                var ld = BuildLostData(Last.Seq + 1, startLast, oldDeleted, bytes: (ulong)(buf.Length - index));
                return (ld, tombstones.ToArray(), new InvalidDataException("record extends past block length"));
            }

            var record = buf.AsSpan(index, recordLength);
            var payloadLength = recordLength - FileStoreDefaults.RecordHashSize;
            var payload = record[..payloadLength];
            var checksum = record[payloadLength..];
            var computed = SHA256.HashData(payload);
            if (!computed.AsSpan(0, FileStoreDefaults.RecordHashSize).SequenceEqual(checksum))
            {
                var ld = BuildLostData(Last.Seq + 1, startLast, oldDeleted, bytes: (ulong)(buf.Length - index));
                return (ld, tombstones.ToArray(), new ErrBadMsg(Mfn, "invalid checksum"));
            }

            Lchk = checksum.ToArray();

            var seq = seqRaw & ~FileStoreDefaults.Tbit & ~FileStoreDefaults.Ebit;
            var isTombstone = (seqRaw & FileStoreDefaults.Tbit) != 0 ||
                              (subjectLength == 0 && headerLength == 0 && msgLength == 0);

            if (isTombstone && seq > 0)
            {
                tombstones.Add(seq);
                Dmap.Insert(seq);
                if (seq > maxTombstoneSeq)
                {
                    maxTombstoneSeq = seq;
                    maxTombstoneTs = ts;
                }

                if (seq > Last.Seq)
                    Last = new MsgId { Seq = seq, Ts = ts };

                index += recordLength;
                continue;
            }

            if (seq == 0)
            {
                index += recordLength;
                continue;
            }

            if (First.Seq == 0 || seq < First.Seq)
                First = new MsgId { Seq = seq, Ts = ts };

            if (seq > highestSeq)
            {
                highestSeq = seq;
                Last = new MsgId { Seq = seq, Ts = ts };
            }

            if (!Dmap.Exists(seq))
            {
                Msgs++;
                Bytes += (ulong)recordLength;
            }

            index += recordLength;
        }

        if (Msgs == 0)
        {
            if (maxTombstoneSeq > 0)
            {
                First = new MsgId { Seq = maxTombstoneSeq + 1, Ts = 0 };
                if (Last.Seq == 0)
                    Last = new MsgId { Seq = maxTombstoneSeq, Ts = maxTombstoneTs };
            }
            else if (First.Seq > 0)
            {
                Last = new MsgId { Seq = First.Seq - 1, Ts = 0 };
            }
            else
            {
                First = new MsgId { Seq = startLast + 1, Ts = 0 };
                Last = new MsgId { Seq = startLast, Ts = 0 };
            }
        }

        return (null, tombstones.ToArray(), null);
    }

    internal byte[] LastChecksum()
    {
        byte[] checksum = new byte[FileStoreDefaults.RecordHashSize];
        if (string.IsNullOrWhiteSpace(Mfn) || !File.Exists(Mfn))
            return Array.Empty<byte>();

        try
        {
            using var file = new FileStream(Mfn, FileMode.Open, FileAccess.Read, FileShare.ReadWrite);
            RBytes = (ulong)file.Length;
            if (RBytes < FileStoreDefaults.RecordHashSize)
                return checksum;

            var encErr = CheckAndLoadEncryption();
            if (encErr != null)
                return Array.Empty<byte>();

            if (Bek != null)
            {
                var buf = File.ReadAllBytes(Mfn);
                if (buf.Length < FileStoreDefaults.RecordHashSize)
                    return checksum;

                Bek.XorKeyStream(buf.AsSpan());
                Buffer.BlockCopy(buf, buf.Length - FileStoreDefaults.RecordHashSize, checksum, 0, FileStoreDefaults.RecordHashSize);
            }
            else
            {
                file.Seek(-FileStoreDefaults.RecordHashSize, SeekOrigin.End);
                file.ReadExactly(checksum, 0, checksum.Length);
            }

            return checksum;
        }
        catch
        {
            return Array.Empty<byte>();
        }
    }

    internal (StoreMsg? Message, bool DidLoad, Exception? Error) FirstMatchingMulti(SimpleSublist? sl, ulong start, StoreMsg? sm)
    {
        Mu.EnterWriteLock();
        try
        {
            if (Fs == null || sl == null)
                return (null, false, StoreErrors.ErrStoreMsgNotFound);

            var floor = Math.Max(start, First.Seq);
            if (floor > Last.Seq)
                return (null, false, StoreErrors.ErrStoreMsgNotFound);

            var (msg, _) = Fs.LoadNextMsgMulti(sl, floor, sm);
            if (msg == null || msg.Seq > Last.Seq)
                return (null, false, StoreErrors.ErrStoreMsgNotFound);

            Llseq = msg.Seq;
            Llts = JetStreamFileStore.TimestampNormalized(DateTime.UtcNow);
            return (msg, false, null);
        }
        catch (Exception ex)
        {
            return (null, false, ex);
        }
        finally
        {
            Mu.ExitWriteLock();
        }
    }

    internal (StoreMsg? Message, bool DidLoad, Exception? Error) FirstMatching(string filter, bool wc, ulong start, StoreMsg? sm)
    {
        Mu.EnterWriteLock();
        try
        {
            if (Fs == null)
                return (null, false, StoreErrors.ErrStoreMsgNotFound);

            var floor = Math.Max(start, First.Seq);
            if (floor > Last.Seq)
                return (null, false, StoreErrors.ErrStoreMsgNotFound);

            var effectiveFilter = string.IsNullOrEmpty(filter) ? ">" : filter;
            var useWildcard = wc || effectiveFilter == ">";
            var (msg, _) = Fs.LoadNextMsg(effectiveFilter, useWildcard, floor, sm);
            if (msg == null || msg.Seq > Last.Seq)
                return (null, false, StoreErrors.ErrStoreMsgNotFound);

            Llseq = msg.Seq;
            Llts = JetStreamFileStore.TimestampNormalized(DateTime.UtcNow);
            return (msg, false, null);
        }
        catch (Exception ex)
        {
            return (null, false, ex);
        }
        finally
        {
            Mu.ExitWriteLock();
        }
    }

    internal (StoreMsg? Message, bool DidLoad, Exception? Error) PrevMatchingMulti(SimpleSublist? sl, ulong start, StoreMsg? sm)
    {
        Mu.EnterWriteLock();
        try
        {
            if (Fs == null || sl == null)
                return (null, false, StoreErrors.ErrStoreMsgNotFound);

            var ceiling = Math.Min(start, Last.Seq);
            if (ceiling < First.Seq)
                return (null, false, StoreErrors.ErrStoreMsgNotFound);

            var (msg, _, err) = Fs.LoadPrevMsgMulti(sl, ceiling, sm);
            if (err != null)
                return (null, false, err);

            if (msg == null || msg.Seq < First.Seq)
                return (null, false, StoreErrors.ErrStoreMsgNotFound);

            Llseq = msg.Seq;
            Llts = JetStreamFileStore.TimestampNormalized(DateTime.UtcNow);
            return (msg, false, null);
        }
        catch (Exception ex)
        {
            return (null, false, ex);
        }
        finally
        {
            Mu.ExitWriteLock();
        }
    }

    internal (ulong Total, ulong First, ulong Last) FilteredPending(string filter, bool wc, ulong seq)
    {
        Mu.EnterWriteLock();
        try
        {
            return FilteredPendingLocked(filter, wc, seq);
        }
        finally
        {
            Mu.ExitWriteLock();
        }
    }

    internal (ulong Total, ulong First, ulong Last) FilteredPendingLocked(string filter, bool wc, ulong sseq)
    {
        if (Fs == null)
            return (0, 0, 0);

        var floor = Math.Max(sseq, First.Seq);
        if (floor > Last.Seq)
            return (0, 0, 0);

        var effectiveFilter = string.IsNullOrEmpty(filter) ? ">" : filter;
        var useWildcard = wc || effectiveFilter == ">";

        ulong total = 0;
        ulong first = 0;
        ulong last = 0;
        var cursor = floor;
        while (cursor <= Last.Seq)
        {
            var (msg, _) = Fs.LoadNextMsg(effectiveFilter, useWildcard, cursor, null);
            if (msg == null || msg.Seq > Last.Seq)
                break;

            total++;
            if (first == 0)
                first = msg.Seq;
            last = msg.Seq;

            if (msg.Seq == ulong.MaxValue)
                break;

            cursor = msg.Seq + 1;
        }

        return (total, first, last);
    }

    internal Exception? SetupWriteCache(byte[]? initialBuffer = null)
    {
        if (CacheData != null)
            return null;

        if (!string.IsNullOrWhiteSpace(Mfn) && File.Exists(Mfn))
        {
            try
            {
                var existing = File.ReadAllBytes(Mfn);
                CacheData = new Cache
                {
                    Buf = existing,
                    Wp = existing.Length,
                    Fseq = First.Seq == 0 ? 1UL : First.Seq,
                };
                Llts = JetStreamFileStore.TimestampNormalized(DateTime.UtcNow);
                return null;
            }
            catch (Exception ex)
            {
                return ex;
            }
        }

        var buf = initialBuffer ?? JetStreamFileStore.GetMsgBlockBuf((int)FileStoreDefaults.DefaultTinyBlockSize);
        CacheData = new Cache
        {
            Buf = buf,
            Wp = 0,
            Fseq = First.Seq == 0 ? 1UL : First.Seq,
        };
        Llts = JetStreamFileStore.TimestampNormalized(DateTime.UtcNow);
        return null;
    }

    internal void ClearCache()
    {
        if (Ctmr != null)
        {
            var sinceLast = SinceLastActivity();
            if (Fss == null || sinceLast >= Fexp)
            {
                Fss = null;
                Ctmr.Dispose();
                Ctmr = null;
            }
            else
            {
                ResetCacheExpireTimerLocked(Fexp - sinceLast);
            }
        }

        var cache = CacheData;
        if (cache == null)
            return;

        var buf = cache.Buf;
        CacheData = null;
        if (!cache.Nra && buf.Length > 0)
            JetStreamFileStore.RecycleMsgBlockBuf(buf);
    }

    internal void ExpireCache()
    {
        Mu.EnterWriteLock();
        try
        {
            ExpireCacheLocked();
        }
        finally
        {
            Mu.ExitWriteLock();
        }
    }

    internal void TryForceExpireCache()
    {
        Mu.EnterWriteLock();
        try
        {
            TryForceExpireCacheLocked();
        }
        finally
        {
            Mu.ExitWriteLock();
        }
    }

    internal byte[]? TryExpireWriteCache()
    {
        if (CacheData == null)
            return null;

        var lwts = Lwts;
        var llts = Llts;
        var prior = CacheData.Buf;
        var priorNra = CacheData.Nra;

        Lwts = 0;
        CacheData.Nra = true;
        ExpireCacheLocked();

        Lwts = lwts;
        if (CacheData != null)
            CacheData.Nra = priorNra;

        if (llts == 0 && CacheData?.Buf is null or { Length: 0 })
        {
            Lwts = 0;
            return Array.Empty<byte>();
        }

        return null;
    }

    internal void ExpireCacheLocked()
    {
        var cache = CacheData;
        if (cache == null && Fss == null)
        {
            if (Ctmr != null)
            {
                Ctmr.Dispose();
                Ctmr = null;
            }
            return;
        }

        if (cache != null && PendingWriteSizeLocked() > 0)
        {
            ResetCacheExpireTimerLocked(Cexp);
            return;
        }

        var now = JetStreamFileStore.TimestampNormalized(DateTime.UtcNow);
        var bufts = Math.Max(Llts, Lwts);
        var cacheNs = TimeSpanToNanoseconds(Cexp);
        if (cacheNs > 0 && now - bufts <= cacheNs)
        {
            ResetCacheExpireTimerLocked(NanosecondsToTimeSpan(cacheNs - (now - bufts)));
            return;
        }

        if (cache != null)
        {
            if (!cache.Nra && cache.Buf.Length > 0)
                JetStreamFileStore.RecycleMsgBlockBuf(cache.Buf);

            cache.Buf = Array.Empty<byte>();
            cache.Idx = Array.Empty<uint>();
            cache.Wp = 0;
        }

        ClearCache();
    }

    internal Exception? EnableForWriting(bool fip)
    {
        if (Mfd != null)
            return null;

        try
        {
            Mfd = new FileStream(Mfn, FileMode.OpenOrCreate, FileAccess.ReadWrite, FileShare.Read);
            if (!fip)
                SpinUpFlushLoopLocked();
            return null;
        }
        catch (Exception ex)
        {
            return ex;
        }
    }

    internal Exception? WriteTombstone(ulong seq, long ts)
        => WriteMsgRecord(0, seq | FileStoreDefaults.Tbit, string.Empty, null, null, ts, true);

    internal Exception? WriteTombstoneNoFlush(ulong seq, long ts)
    {
        Mu.EnterWriteLock();
        try
        {
            return WriteMsgRecordLocked(0, seq | FileStoreDefaults.Tbit, string.Empty, null, null, ts, flush: false, kick: false);
        }
        finally
        {
            Mu.ExitWriteLock();
        }
    }

    internal Exception? WriteMsgRecord(ulong rl, ulong seq, string subj, byte[]? mhdr, byte[]? msg, long ts, bool flush)
    {
        Mu.EnterWriteLock();
        try
        {
            return WriteMsgRecordLocked(rl, seq, subj, mhdr, msg, ts, flush, kick: true);
        }
        finally
        {
            Mu.ExitWriteLock();
        }
    }

    internal Exception? WriteMsgRecordLocked(ulong rl, ulong seq, string subj, byte[]? mhdr, byte[]? msg, long ts, bool flush, bool kick)
    {
        var enableErr = EnableForWriting(flush && kick);
        if (enableErr != null)
            return enableErr;

        var cacheErr = SetupWriteCache();
        if (cacheErr != null)
            return cacheErr;

        var cache = CacheData;
        if (cache == null)
            return new InvalidOperationException("cache was not initialized");

        mhdr ??= Array.Empty<byte>();
        msg ??= Array.Empty<byte>();
        rl = rl == 0 ? (ulong)CalculateRecordLength(subj, mhdr, msg) : rl;

        var index = cache.Buf.Length;
        var record = BuildMessageRecord(subj, mhdr, msg, seq, ts);
        cache.Buf = AppendBuffer(cache.Buf, record);
        Lwts = ts;
        NeedSync = true;
        Lchk = record.AsSpan(record.Length - FileStoreDefaults.RecordHashSize).ToArray();

        var isTombstone = (seq & FileStoreDefaults.Tbit) != 0;
        if (!isTombstone)
        {
            var last = Last.Seq;
            UpdateAccounting(seq, ts, rl);
            seq &= ~FileStoreDefaults.Ebit;

            if (cache.Idx.Length > 0 && last + 1 < seq)
            {
                var fill = (int)Math.Min(int.MaxValue, seq - (last + 1));
                if (fill > 0)
                {
                    var next = new uint[cache.Idx.Length + fill];
                    cache.Idx.AsSpan().CopyTo(next);
                    for (var i = cache.Idx.Length; i < next.Length; i++)
                        next[i] = FileStoreDefaults.Dbit;
                    cache.Idx = next;
                }
            }

            var slot = (uint)index | FileStoreDefaults.Cbit;
            var expanded = new uint[cache.Idx.Length + 1];
            cache.Idx.AsSpan().CopyTo(expanded);
            expanded[^1] = slot;
            cache.Idx = expanded;
            if (cache.Idx.Length == 1)
                cache.Fseq = seq;
        }
        else
        {
            var tombstoneSeq = seq & ~FileStoreDefaults.Tbit;
            if (Msgs == 0 && tombstoneSeq > Last.Seq)
            {
                Last = new MsgId { Seq = tombstoneSeq, Ts = ts };
                First = new MsgId { Seq = tombstoneSeq + 1, Ts = 0 };
            }
            RBytes += rl;
            Dmap.Insert(tombstoneSeq);
        }

        if (flush || Werr != null)
        {
            var (lostData, flushErr) = FlushPendingMsgsLocked();
            if (lostData != null && Fs != null)
                Fs.RebuildState(lostData);
            if (flushErr != null)
                return flushErr;
        }
        else if (kick)
        {
            JetStreamFileStore.KickFlusher(Fch);
        }

        return null;
    }

    internal void FinishedWithCache()
    {
        if (CacheData != null && PendingWriteSizeLocked() == 0)
            CacheData = null;
    }

    internal (ulong Seq, Exception? Error) SkipMsg(ulong seq)
    {
        var fs = Fs;
        if (fs == null)
            return (0, StoreErrors.ErrStoreClosed);

        return fs.SkipMsg(seq);
    }

    internal bool ShouldCompactInline()
    {
        if (NoCompact || Msgs == 0 || Dmap.Size == 0)
            return false;

        var deletedBytes = RBytes > Bytes ? RBytes - Bytes : 0;
        return deletedBytes >= (ulong)FileStoreDefaults.CompactMinimum ||
               Dmap.Size >= Math.Max(1, (int)(Msgs / 2));
    }

    internal bool ShouldCompactSync()
        => ShouldCompactInline() && SyncAlways;

    internal Exception? Compact()
        => CompactWithFloor(0);

    internal Exception? CompactWithFloor(ulong floor)
    {
        var fs = Fs;
        if (fs == null)
            return StoreErrors.ErrStoreClosed;

        var start = floor == 0 ? Math.Max(First.Seq, 1UL) : floor;
        var (_, err) = fs.Compact(start);
        return err;
    }

    internal (uint Slot, bool Deleted) SlotInfo(ulong seq)
    {
        var cache = CacheData;
        if (cache == null || cache.Idx.Length == 0 || seq < cache.Fseq)
            return (0, false);

        var slot = seq - cache.Fseq;
        if (slot >= (ulong)cache.Idx.Length)
            return (0, false);

        var raw = cache.Idx[slot];
        var deleted = (raw & FileStoreDefaults.Dbit) != 0;
        return (raw & ~(FileStoreDefaults.Dbit | FileStoreDefaults.Cbit), deleted);
    }

    // Will do a lookup from cache.
    // This will copy the msg from the cache.
    // Lock should be held.
    internal (StoreMsg? Message, Exception? Error) CacheLookup(ulong seq, StoreMsg? sm)
        => CacheLookupEx(seq, sm, doCopy: true);

    // Will do a lookup from cache.
    // This will NOT copy the msg from the cache.
    // Lock should be held.
    internal (StoreMsg? Message, Exception? Error) CacheLookupNoCopy(ulong seq, StoreMsg? sm)
        => CacheLookupEx(seq, sm, doCopy: false);

    // Will do a lookup from cache.
    // Lock should be held.
    internal (StoreMsg? Message, Exception? Error) CacheLookupEx(ulong seq, StoreMsg? sm, bool doCopy)
    {
        var fseq = First.Seq;
        var lseq = Last.Seq;
        if ((fseq > 0 && lseq == fseq - 1) || seq < fseq || seq > lseq)
            return (null, StoreErrors.ErrStoreMsgNotFound);

        if (Llseq == 0 || seq < Llseq || seq == Llseq + 1 || seq + 1 == Llseq)
            Llseq = seq;

        if (Dmap.Exists(seq))
        {
            Llts = JetStreamFileStore.TimestampNormalized(DateTime.UtcNow);
            return (null, ErrDeletedMsg);
        }

        var cache = CacheData;
        if (cache == null || cache.Fseq == 0 || cache.Idx.Length == 0 || cache.Buf.Length == 0)
        {
            if (cache != null)
                TryForceExpireCacheLocked();
            return (null, ErrNoCache);
        }

        if (seq < cache.Fseq)
            return (null, ErrPartialCache);

        var slot = seq - cache.Fseq;
        if (slot >= (ulong)cache.Idx.Length)
            return (null, StoreErrors.ErrStoreMsgNotFound);

        var raw = cache.Idx[(int)slot];
        var deleted = (raw & FileStoreDefaults.Dbit) != 0;
        var hashChecked = (raw & FileStoreDefaults.Cbit) != 0;
        if (deleted)
            return (null, ErrDeletedMsg);

        Llts = JetStreamFileStore.TimestampNormalized(DateTime.UtcNow);

        var offset = (int)(raw & ~(FileStoreDefaults.Dbit | FileStoreDefaults.Cbit));
        if ((uint)offset >= (uint)cache.Buf.Length)
            return (null, ErrPartialCache);

        var (fsm, parseErr) = MsgFromBufEx(cache.Buf.AsSpan(offset), sm, verifyChecksum: !hashChecked, doCopy: doCopy);
        if (parseErr != null || fsm == null)
            return (null, parseErr ?? StoreErrors.ErrStoreMsgNotFound);

        if (fsm.Seq == 0)
            return (null, ErrDeletedMsg);

        if (seq != fsm.Seq)
        {
            TryForceExpireCacheLocked();
            return (null, new InvalidOperationException($"sequence numbers for cache load did not match, {seq} vs {fsm.Seq}"));
        }

        if (!hashChecked)
            cache.Idx[(int)slot] = raw | FileStoreDefaults.Cbit;

        return (fsm, null);
    }

    // Internal function to return msg parts from a raw buffer.
    // Raw buffer will be copied into sm.
    // Lock should be held.
    internal (StoreMsg? Message, Exception? Error) MsgFromBuf(ReadOnlySpan<byte> buf, StoreMsg? sm, bool verifyChecksum)
        => MsgFromBufEx(buf, sm, verifyChecksum, doCopy: true);

    // Internal function to return msg parts from a raw buffer.
    // Raw buffer will NOT be copied into sm.
    // Lock should be held.
    internal (StoreMsg? Message, Exception? Error) MsgFromBufNoCopy(ReadOnlySpan<byte> buf, StoreMsg? sm, bool verifyChecksum)
        => MsgFromBufEx(buf, sm, verifyChecksum, doCopy: false);

    // Internal function to return msg parts from a raw buffer.
    // copy boolean determines if message/header buffers are copied.
    // Lock should be held.
    internal (StoreMsg? Message, Exception? Error) MsgFromBufEx(ReadOnlySpan<byte> buf, StoreMsg? sm, bool verifyChecksum, bool doCopy)
    {
        if (!TryReadRecordHeader(buf, out var seqRaw, out var ts, out var subjectLength, out var headerLength, out var msgLength, out var recordLength, out var parseErr))
            return (null, parseErr ?? new ErrBadMsg(Mfn, "record too short"));

        if (recordLength > buf.Length)
            return (null, new ErrBadMsg(Mfn, "record extends past buffer"));

        var record = buf[..recordLength];
        var payloadLength = recordLength - FileStoreDefaults.RecordHashSize;
        var payload = record[..payloadLength];
        var checksum = record[payloadLength..];

        if (verifyChecksum)
        {
            var computed = SHA256.HashData(payload);
            if (!computed.AsSpan(0, FileStoreDefaults.RecordHashSize).SequenceEqual(checksum))
                return (null, new ErrBadMsg(Mfn, "invalid checksum"));
        }

        var seq = seqRaw;
        if ((seq & FileStoreDefaults.Ebit) != 0)
            seq = 0;
        seq &= ~FileStoreDefaults.Tbit;

        sm ??= new StoreMsg();
        sm.Clear();
        sm.Seq = seq;
        sm.Ts = ts;

        var subjectOffset = 8 + 8 + 4 + 4 + 4;
        var headerOffset = subjectOffset + subjectLength;
        var msgOffset = headerOffset + headerLength;
        if (msgOffset + msgLength > record.Length)
            return (null, new ErrBadMsg(Mfn, "invalid message bounds"));

        sm.Subject = subjectLength > 0
            ? Encoding.UTF8.GetString(record.Slice(subjectOffset, subjectLength))
            : string.Empty;

        var headerSlice = record.Slice(headerOffset, headerLength);
        var msgSlice = record.Slice(msgOffset, msgLength);

        if (doCopy)
        {
            sm.Hdr = headerSlice.ToArray();
            sm.Msg = msgSlice.ToArray();
        }
        else
        {
            // StoreMsg uses byte[] fields, so no-copy mode still exposes slices as arrays.
            sm.Hdr = headerSlice.ToArray();
            sm.Msg = msgSlice.ToArray();
        }

        sm.Buf = new byte[sm.Hdr.Length + sm.Msg.Length];
        if (sm.Hdr.Length > 0)
            Buffer.BlockCopy(sm.Hdr, 0, sm.Buf, 0, sm.Hdr.Length);
        if (sm.Msg.Length > 0)
            Buffer.BlockCopy(sm.Msg, 0, sm.Buf, sm.Hdr.Length, sm.Msg.Length);

        return (sm, null);
    }

    // readIndexInfo will read in the index information for this message block.
    internal Exception? ReadIndexInfo()
    {
        var dir = Path.GetDirectoryName(Mfn);
        if (string.IsNullOrWhiteSpace(dir))
            return new InvalidOperationException("message block path is missing directory");

        var ifn = Path.Combine(dir, string.Format(FileStoreDefaults.IndexScan, Index));
        byte[] buf;
        try
        {
            buf = File.ReadAllBytes(ifn);
        }
        catch (Exception ex)
        {
            return ex;
        }

        if (Liwsz == 0)
            Liwsz = buf.Length;

        if (Aek != null)
        {
            if (Nonce == null || Nonce.Length == 0)
                return new InvalidOperationException("missing block nonce");

            try
            {
                buf = Aek.Open(Nonce, buf);
            }
            catch (Exception ex)
            {
                return ex;
            }
        }

        var headerErr = JetStreamFileStore.CheckNewHeader(buf);
        if (headerErr != null)
        {
            TryDeleteFile(ifn);
            return new InvalidDataException("bad index file");
        }

        var bi = FileStoreDefaults.HdrLen;
        if (!TryReadUVarInt(buf, ref bi, out var msgs) ||
            !TryReadUVarInt(buf, ref bi, out var bytes) ||
            !TryReadUVarInt(buf, ref bi, out var firstSeq) ||
            !TryReadVarInt(buf, ref bi, out var firstTs) ||
            !TryReadUVarInt(buf, ref bi, out var lastSeq) ||
            !TryReadVarInt(buf, ref bi, out var lastTs) ||
            !TryReadUVarInt(buf, ref bi, out var dmapLen))
        {
            TryDeleteFile(ifn);
            return new InvalidDataException("short index file");
        }

        Msgs = msgs;
        Bytes = bytes;
        First = new MsgId { Seq = firstSeq, Ts = firstTs };
        Last = new MsgId { Seq = lastSeq, Ts = lastTs };

        var expected = Last.Seq >= First.Seq ? Last.Seq - First.Seq + 1 : 0;
        var activeExpected = expected >= dmapLen ? expected - dmapLen : 0;
        if (Msgs != activeExpected)
        {
            TryDeleteFile(ifn);
            return new InvalidDataException("accounting inconsistent");
        }

        if (bi < 0 || bi + FileStoreDefaults.RecordHashSize > buf.Length)
        {
            TryDeleteFile(ifn);
            return new InvalidDataException("short index file");
        }

        Lchk = buf.AsSpan(bi, FileStoreDefaults.RecordHashSize).ToArray();
        bi += FileStoreDefaults.RecordHashSize;

        if (dmapLen > 0)
        {
            if (buf[1] == FileStoreDefaults.NewVersion)
            {
                try
                {
                    var (decoded, consumed) = SequenceSet.Decode(buf.AsSpan(bi));
                    if (consumed <= 0)
                        return new InvalidDataException("could not decode dmap");

                    Dmap = decoded;
                }
                catch (Exception ex)
                {
                    return ex;
                }
            }
            else
            {
                var fseq = First.Seq;
                for (var i = 0UL; i < dmapLen; i++)
                {
                    if (!TryReadUVarInt(buf, ref bi, out var relSeq))
                        break;

                    Dmap.Insert(relSeq + fseq);
                }
            }
        }

        return null;
    }

    // Return all active tombstones in this block.
    internal MsgId[] Tombs()
    {
        Mu.EnterWriteLock();
        try
        {
            return TombsLocked();
        }
        finally
        {
            Mu.ExitWriteLock();
        }
    }

    // Return all active tombstones in this block.
    // Write lock should be held.
    internal MsgId[] TombsLocked()
    {
        if (CacheNotLoaded())
        {
            var loadErr = LoadMsgsWithLock();
            if (loadErr != null)
                return Array.Empty<MsgId>();
        }

        try
        {
            var cache = CacheData;
            if (cache == null || cache.Buf.Length == 0)
                return Array.Empty<MsgId>();

            var tombs = new List<MsgId>();
            var offset = 0;
            while (offset < cache.Buf.Length)
            {
                if (!TryReadRecordHeader(cache.Buf.AsSpan(offset), out var seqRaw, out var ts, out _, out _, out _, out var recordLength, out _))
                    break;

                if (recordLength <= 0 || offset + recordLength > cache.Buf.Length)
                    break;

                if ((seqRaw & FileStoreDefaults.Tbit) != 0)
                    tombs.Add(new MsgId { Seq = seqRaw & ~FileStoreDefaults.Tbit, Ts = ts });

                offset += recordLength;
            }

            return tombs.ToArray();
        }
        finally
        {
            FinishedWithCache();
        }
    }

    // fs lock should be held.
    internal int NumPriorTombs()
    {
        Mu.EnterWriteLock();
        try
        {
            return NumPriorTombsLocked();
        }
        finally
        {
            Mu.ExitWriteLock();
        }
    }

    // Return number of tombstones for messages prior to this block.
    // Write lock should be held for block.
    internal int NumPriorTombsLocked()
    {
        if (CacheNotLoaded())
        {
            var loadErr = LoadMsgsWithLock();
            if (loadErr != null)
                return 0;
        }

        try
        {
            var cache = CacheData;
            if (cache == null || cache.Buf.Length == 0)
                return 0;

            var fsFirstSeq = Fs?.State().FirstSeq ?? 0;
            ulong blockFirst = 0;
            var tombs = 0;
            var offset = 0;
            while (offset < cache.Buf.Length)
            {
                if (!TryReadRecordHeader(cache.Buf.AsSpan(offset), out var seqRaw, out _, out _, out _, out _, out var recordLength, out _))
                    break;

                if (recordLength <= 0 || offset + recordLength > cache.Buf.Length)
                    break;

                if ((seqRaw & FileStoreDefaults.Tbit) != 0)
                {
                    var tseq = seqRaw & ~FileStoreDefaults.Tbit;
                    if (tseq >= fsFirstSeq && (blockFirst == 0 || tseq < blockFirst))
                        tombs++;

                    offset += recordLength;
                    continue;
                }

                var seq = seqRaw & ~FileStoreDefaults.Ebit;
                if (seq != 0 && (seqRaw & FileStoreDefaults.Ebit) == 0 && blockFirst == 0)
                    blockFirst = seq;

                offset += recordLength;
            }

            return tombs;
        }
        finally
        {
            FinishedWithCache();
        }
    }

    // Called by purge to simply get rid of cache and close fds.
    internal void DirtyClose()
    {
        Mu.EnterWriteLock();
        try
        {
            _ = DirtyCloseWithRemove(remove: false);
        }
        finally
        {
            Mu.ExitWriteLock();
        }
    }

    // Should be called with write lock held.
    internal Exception? DirtyCloseWithRemove(bool remove)
    {
        if (Ctmr != null)
        {
            Ctmr.Dispose();
            Ctmr = null;
        }

        ClearCacheAndOffset();

        Qch?.Writer.TryComplete();
        Qch = null;

        Mfd?.Dispose();
        Mfd = null;

        if (!remove)
            return null;

        Fss = null;
        if (!string.IsNullOrEmpty(Mfn))
        {
            try
            {
                File.Delete(Mfn);
                Mfn = string.Empty;
            }
            catch (Exception ex)
            {
                return ex;
            }
        }

        if (!string.IsNullOrEmpty(Kfn))
        {
            try
            {
                File.Delete(Kfn);
            }
            catch (Exception ex)
            {
                return ex;
            }
        }

        return null;
    }

    // Close the message block.
    internal void Close(bool sync)
    {
        Mu.EnterWriteLock();
        try
        {
            if (Closed)
                return;

            if (Ctmr != null)
            {
                Ctmr.Dispose();
                Ctmr = null;
            }

            Fss = null;
            _ = FlushPendingMsgsLocked();
            ClearCacheAndOffset();

            Qch?.Writer.TryComplete();
            Qch = null;

            if (Mfd != null)
            {
                if (sync)
                    Mfd.Flush(flushToDisk: true);
                Mfd.Dispose();
            }

            Mfd = null;
            Closed = true;
        }
        finally
        {
            Mu.ExitWriteLock();
        }
    }

    // Remove a sequence from per-subject state and track whether first/last need recompute.
    // Lock should be held.
    internal ulong RemoveSeqPerSubject(string subj, ulong seq)
    {
        _ = EnsurePerSubjectInfoLoaded();
        if (Fss == null || string.IsNullOrEmpty(subj))
            return 0;

        var bsubj = Encoding.UTF8.GetBytes(subj);
        var (ss, ok) = Fss.Find(bsubj);
        if (!ok || ss == null)
            return 0;

        if (ss.Msgs == 1)
        {
            Fss.Delete(bsubj);
            return 0;
        }

        ss.Msgs--;

        if (ss.Msgs == 1)
        {
            if (!ss.LastNeedsUpdate && seq != ss.Last)
            {
                ss.First = ss.Last;
                ss.FirstNeedsUpdate = false;
                return 1;
            }

            if (!ss.FirstNeedsUpdate && seq != ss.First)
            {
                ss.Last = ss.First;
                ss.LastNeedsUpdate = false;
                return 1;
            }
        }

        ss.FirstNeedsUpdate = seq == ss.First || ss.FirstNeedsUpdate;
        ss.LastNeedsUpdate = seq == ss.Last || ss.LastNeedsUpdate;
        return ss.Msgs;
    }

    // Recalculate first/last sequence values for a subject in this block.
    // Lock should be held.
    internal void RecalculateForSubj(string subj, SimpleState ss)
    {
        if (string.IsNullOrEmpty(subj) || ss == null)
            return;

        var loadedHere = false;
        if (CacheNotLoaded())
        {
            if (LoadMsgsWithLock() != null)
                return;
            loadedHere = true;
        }

        try
        {
            var cache = CacheData;
            if (cache == null || cache.Idx.Length == 0 || cache.Buf.Length == 0)
                return;

            var startSlot = (int)Math.Max(0, (long)ss.First - (long)cache.Fseq);
            if (startSlot >= cache.Idx.Length)
            {
                ss.First = ss.Last;
                ss.FirstNeedsUpdate = false;
                ss.LastNeedsUpdate = false;
                return;
            }

            var endSlot = (int)Math.Max(0, (long)ss.Last - (long)cache.Fseq);
            if (endSlot >= cache.Idx.Length)
                endSlot = cache.Idx.Length - 1;
            if (startSlot > endSlot)
                return;

            if (ss.FirstNeedsUpdate)
            {
                ss.FirstNeedsUpdate = false;
                var minSeq = ss.First + 1;
                if (minSeq < First.Seq)
                    minSeq = First.Seq;

                var scratch = new StoreMsg();
                for (var slot = startSlot; slot < cache.Idx.Length; slot++)
                {
                    var raw = cache.Idx[slot] & ~FileStoreDefaults.Cbit;
                    if (raw == FileStoreDefaults.Dbit)
                        continue;

                    var offset = (int)raw;
                    if ((uint)offset >= (uint)cache.Buf.Length)
                    {
                        ss.First = ss.Last;
                        ss.LastNeedsUpdate = false;
                        return;
                    }

                    var (sm, err) = MsgFromBufNoCopy(cache.Buf.AsSpan(offset), scratch, verifyChecksum: false);
                    if (err != null || sm == null || sm.Seq == 0)
                        continue;
                    if (sm.Subject != subj || sm.Seq < minSeq || Dmap.Exists(sm.Seq))
                        continue;

                    ss.First = sm.Seq;
                    if (ss.Msgs == 1)
                    {
                        ss.Last = sm.Seq;
                        ss.LastNeedsUpdate = false;
                        return;
                    }

                    startSlot = slot;
                    break;
                }
            }

            if (ss.LastNeedsUpdate)
            {
                ss.LastNeedsUpdate = false;
                var maxSeq = ss.Last > 0 ? ss.Last - 1 : 0;
                if (maxSeq > Last.Seq)
                    maxSeq = Last.Seq;

                var scratch = new StoreMsg();
                for (var slot = endSlot; slot >= startSlot; slot--)
                {
                    var raw = cache.Idx[slot] & ~FileStoreDefaults.Cbit;
                    if (raw == FileStoreDefaults.Dbit)
                        continue;

                    var offset = (int)raw;
                    if ((uint)offset >= (uint)cache.Buf.Length)
                        return;

                    var (sm, err) = MsgFromBufNoCopy(cache.Buf.AsSpan(offset), scratch, verifyChecksum: false);
                    if (err != null || sm == null || sm.Seq == 0)
                        continue;
                    if (sm.Subject != subj || sm.Seq > maxSeq || Dmap.Exists(sm.Seq))
                        continue;

                    ss.Last = sm.Seq < ss.First ? ss.First : sm.Seq;
                    if (ss.Msgs == 1)
                    {
                        ss.First = ss.Last;
                        ss.FirstNeedsUpdate = false;
                    }
                    return;
                }
            }
        }
        finally
        {
            if (loadedHere)
                FinishedWithCache();
        }
    }

    // Lock should be held.
    internal Exception? ResetPerSubjectInfo()
    {
        Fss = null;
        return GeneratePerSubjectInfo();
    }

    // generatePerSubjectInfo rebuilds per-subject state from the cached block data.
    // Lock should be held.
    internal Exception? GeneratePerSubjectInfo()
    {
        if (Msgs == 0)
            return null;

        var loadedHere = false;
        if (CacheNotLoaded())
        {
            var loadErr = LoadMsgsWithLock();
            if (loadErr != null)
                return loadErr;

            if (Fss != null)
                return null;

            loadedHere = true;
        }

        try
        {
            Fss = new SubjectTree<SimpleState>();
            var scratch = new StoreMsg();
            for (var seq = First.Seq; seq <= Last.Seq; seq++)
            {
                if (Dmap.Exists(seq))
                {
                    if (seq == ulong.MaxValue)
                        break;
                    continue;
                }

                var (sm, err) = CacheLookupNoCopy(seq, scratch);
                if (err != null)
                {
                    if (ReferenceEquals(err, StoreErrors.ErrStoreMsgNotFound) || IsDeletedMsgError(err))
                    {
                        if (seq == ulong.MaxValue)
                            break;
                        continue;
                    }

                    if (ReferenceEquals(err, ErrNoCache))
                        return null;

                    return err;
                }

                if (sm != null && !string.IsNullOrEmpty(sm.Subject))
                {
                    var subjBytes = Encoding.UTF8.GetBytes(sm.Subject);
                    var (state, exists) = Fss.Find(subjBytes);
                    if (exists && state != null)
                    {
                        state.Msgs++;
                        state.Last = seq;
                        state.LastNeedsUpdate = false;
                    }
                    else
                    {
                        Fss.Insert(subjBytes, new SimpleState
                        {
                            Msgs = 1,
                            First = seq,
                            Last = seq,
                        });
                    }
                }

                if (seq == ulong.MaxValue)
                    break;
            }

            if (Fss.Size() > 0)
            {
                Llts = JetStreamFileStore.TimestampNormalized(DateTime.UtcNow);
                Lsts = Llts;
                StartCacheExpireTimer();
            }
            return null;
        }
        finally
        {
            if (loadedHere)
                FinishedWithCache();
        }
    }

    // Helper to make sure per-subject state is loaded if we are tracking subjects.
    // Lock should be held.
    internal Exception? EnsurePerSubjectInfoLoaded()
    {
        if (Fss != null || NoTrack)
        {
            if (Fss != null)
                Lsts = JetStreamFileStore.TimestampNormalized(DateTime.UtcNow);
            return null;
        }

        if (Msgs == 0)
        {
            Fss = new SubjectTree<SimpleState>();
            return null;
        }

        return GeneratePerSubjectInfo();
    }

    internal void SpinUpFlushLoop()
    {
        Mu.EnterWriteLock();
        try
        {
            SpinUpFlushLoopLocked();
        }
        finally
        {
            Mu.ExitWriteLock();
        }
    }

    internal void SpinUpFlushLoopLocked()
    {
        if (Flusher || Closed)
            return;

        Flusher = true;
        Fch = Channel.CreateBounded<byte>(1);
        Qch = Channel.CreateUnbounded<byte>();
        var fch = Fch;
        var qch = Qch;

        _ = Task.Run(() => FlushLoop(fch, qch));
    }

    internal void KickFlusher()
    {
        Mu.EnterReadLock();
        try
        {
            _ = Fch?.Writer.TryWrite(0);
        }
        finally
        {
            Mu.ExitReadLock();
        }
    }

    internal void SetInFlusher()
    {
        Mu.EnterWriteLock();
        try
        {
            Flusher = true;
        }
        finally
        {
            Mu.ExitWriteLock();
        }
    }

    internal void ClearInFlusher()
    {
        Mu.EnterWriteLock();
        try
        {
            Flusher = false;
            Qch?.Writer.TryComplete();
            Qch = null;
            Fch?.Writer.TryComplete();
            Fch = null;
        }
        finally
        {
            Mu.ExitWriteLock();
        }
    }

    internal void FlushLoop(Channel<byte>? flushChannel = null, Channel<byte>? quitChannel = null)
    {
        SetInFlusher();
        try
        {
            var fch = flushChannel ?? Fch;
            var qch = quitChannel ?? Qch;
            if (fch == null)
                return;

            while (true)
            {
                if (qch?.Reader.Completion.IsCompleted == true)
                    break;

                var canRead = fch.Reader.WaitToReadAsync().AsTask().GetAwaiter().GetResult();
                if (!canRead)
                    break;

                while (fch.Reader.TryRead(out _)) { }

                Mu.EnterWriteLock();
                try
                {
                    if (Closed)
                        break;

                    if (Mfd != null)
                    {
                        Mfd.Flush(SyncAlways);
                        NeedSync = false;
                    }
                }
                catch (Exception ex)
                {
                    Werr = ex;
                }
                finally
                {
                    Mu.ExitWriteLock();
                }
            }
        }
        finally
        {
            ClearInFlusher();
        }
    }

    internal (bool Removed, Exception? Error) EraseMsg(ulong seq)
    {
        var fs = Fs;
        if (fs == null)
            return (false, StoreErrors.ErrStoreClosed);

        return fs.EraseMsg(seq);
    }

    internal void Truncate(ulong seq)
    {
        Fs?.Truncate(seq);
    }

    internal bool IsEmpty()
    {
        Mu.EnterReadLock();
        try
        {
            return Msgs == 0;
        }
        finally
        {
            Mu.ExitReadLock();
        }
    }

    internal void ResetCacheExpireTimer()
    {
        Mu.EnterWriteLock();
        try
        {
            ResetCacheExpireTimerLocked(Cexp);
        }
        finally
        {
            Mu.ExitWriteLock();
        }
    }

    internal void StartCacheExpireTimer()
    {
        if (Cexp <= TimeSpan.Zero)
            return;

        ResetCacheExpireTimerLocked(Cexp);
    }

    internal void ClearCacheAndOffset()
    {
        CacheData = null;
        Fss = null;
        Ctmr?.Dispose();
        Ctmr = null;

        try
        {
            if (!string.IsNullOrWhiteSpace(Mfn))
            {
                var dir = Path.GetDirectoryName(Mfn);
                if (!string.IsNullOrWhiteSpace(dir))
                {
                    var idxPath = Path.Combine(dir, string.Format(FileStoreDefaults.IndexScan, Index));
                    if (File.Exists(idxPath))
                        File.Delete(idxPath);

                    var fssPath = Path.Combine(dir, $"{Index}.fss");
                    if (File.Exists(fssPath))
                        File.Delete(fssPath);
                }
            }
        }
        catch
        {
            // best effort cleanup for stale metadata files
        }
    }

    internal ulong PendingWriteSize()
    {
        Mu.EnterReadLock();
        try
        {
            return PendingWriteSizeLocked();
        }
        finally
        {
            Mu.ExitReadLock();
        }
    }

    internal ulong PendingWriteSizeLocked()
    {
        if (Closed || Mfd == null)
            return 0;

        var cache = CacheData;
        if (cache == null || cache.Buf.Length <= cache.Wp)
            return 0;

        var wp = Math.Clamp(cache.Wp, 0, cache.Buf.Length);
        return (ulong)(cache.Buf.Length - wp);
    }

    internal (byte[]? Buffer, Exception? Error) BytesPending()
    {
        if (Mfd == null)
            return (null, new InvalidOperationException("no pending data"));

        var cache = CacheData;
        if (cache == null)
            return (null, new InvalidOperationException("no cache"));

        if (cache.Buf.Length <= cache.Wp)
            return (null, new InvalidOperationException("no pending data"));

        var wp = Math.Clamp(cache.Wp, 0, cache.Buf.Length);
        var pending = cache.Buf[wp..];
        return pending.Length == 0
            ? (null, new InvalidOperationException("no pending data"))
            : (pending, null);
    }

    internal ulong BlkSize()
    {
        Mu.EnterReadLock();
        try
        {
            return RBytes;
        }
        finally
        {
            Mu.ExitReadLock();
        }
    }

    internal void UpdateAccounting(ulong seq, long ts, ulong rl)
    {
        var isDeleted = (seq & FileStoreDefaults.Ebit) != 0;
        if (isDeleted)
            seq &= ~FileStoreDefaults.Ebit;

        var fseq = First.Seq;
        if ((fseq == 0 || First.Ts == 0) && seq >= fseq)
            First = new MsgId { Seq = seq, Ts = ts };

        Last = new MsgId { Seq = seq, Ts = ts };
        RBytes += rl;
        if (!isDeleted)
        {
            Bytes += rl;
            Msgs++;
        }
    }

    internal Exception? RecompressOnDiskIfNeeded()
    {
        byte[] original;
        try
        {
            original = File.Exists(Mfn) ? File.ReadAllBytes(Mfn) : Array.Empty<byte>();
        }
        catch (Exception ex)
        {
            return new IOException("failed to read original block from disk", ex);
        }

        if (original.Length == 0)
            return null;

        var meta = new CompressionInfo();
        var consumed = meta.UnmarshalMetadata(original);
        var desired = Cmp;
        if (consumed > 0 && meta.Type == desired && desired == StoreCompression.NoCompression)
            return null;

        byte[] rewritten = original;
        if (consumed > 0)
        {
            if (meta.Type == StoreCompression.NoCompression)
            {
                rewritten = original[consumed..];
            }
            else
            {
                var (decompressed, decompressErr) = DecompressIfNeeded(original);
                if (decompressErr != null)
                    return decompressErr;

                rewritten = decompressed ?? Array.Empty<byte>();
            }
        }

        return AtomicOverwriteFile(rewritten, allowCompress: true);
    }

    // Lock should be held.
    internal Exception? AtomicOverwriteFile(byte[] buf, bool allowCompress)
    {
        ArgumentNullException.ThrowIfNull(buf);

        var hadOpenFile = Mfd != null;
        if (hadOpenFile)
            CloseFDsLockedNoCheck();

        var origFileName = Mfn;
        var tempFileName = Mfn + FileStoreDefaults.BlkTmpSuffix;

        try
        {
            var algorithm = StoreCompression.NoCompression;
            if (allowCompress && Cmp != StoreCompression.NoCompression)
            {
                if (Cmp != StoreCompression.S2Compression)
                    return new NotSupportedException($"unsupported compression algorithm: {Cmp}");

                return new NotSupportedException("message block compression is not available");
            }

            var writeBuffer = buf;
            if (algorithm != StoreCompression.NoCompression)
            {
                return new NotSupportedException("message block compression is not available");
            }

            var encryptErr = EncryptOrDecryptIfNeeded(writeBuffer);
            if (encryptErr != null)
                return encryptErr;

            using var tmp = new FileStream(tempFileName, FileMode.Create, FileAccess.Write, FileShare.None);
            tmp.Write(writeBuffer, 0, writeBuffer.Length);
            tmp.Flush(flushToDisk: true);
            tmp.Close();

            File.Move(tempFileName, origFileName, overwrite: true);

            Cmp = algorithm;
            RBytes = (ulong)writeBuffer.Length;

            if (hadOpenFile)
            {
                var enableErr = EnableForWriting(false);
                if (enableErr != null)
                    return enableErr;
            }

            return null;
        }
        catch (Exception ex)
        {
            try
            {
                if (File.Exists(tempFileName))
                    File.Delete(tempFileName);
            }
            catch
            {
                // Best-effort cleanup only.
            }

            if (hadOpenFile && Mfd == null)
                _ = EnableForWriting(false);

            return ex;
        }
    }

    // Lock should be held.
    internal (byte[]? Buffer, Exception? Error) DecompressIfNeeded(byte[] buf)
    {
        ArgumentNullException.ThrowIfNull(buf);

        var hasCompressionHeader = buf.Length >= 3 &&
                                   buf[0] == (byte)'c' &&
                                   buf[1] == (byte)'m' &&
                                   buf[2] == (byte)'p';

        var meta = new CompressionInfo();
        var consumed = meta.UnmarshalMetadata(buf);
        if (consumed == 0)
        {
            if (hasCompressionHeader)
                return (null, new InvalidDataException("invalid compression metadata"));

            return (buf, null);
        }

        if (meta.Type == StoreCompression.NoCompression)
            return (buf[consumed..], null);

        return (null, new NotSupportedException("message block decompression is not available"));
    }

    // Lock should be held.
    internal Exception? EncryptOrDecryptIfNeeded(byte[] buf)
    {
        ArgumentNullException.ThrowIfNull(buf);

        if (Bek != null && buf.Length > 0)
            Bek.XorKeyStream(buf.AsSpan());

        return null;
    }

    // Lock should be held.
    internal Exception? EnsureRawBytesLoaded()
    {
        if (RBytes > 0)
            return null;

        var (file, openErr) = OpenBlock();
        if (openErr != null || file == null)
            return openErr ?? new InvalidOperationException("unable to open message block");

        try
        {
            RBytes = (ulong)file.Length;
            return null;
        }
        finally
        {
            file.Dispose();
        }
    }

    // Index a raw message buffer.
    // Lock should be held.
    internal Exception? IndexCacheBuf(byte[] buf)
    {
        ArgumentNullException.ThrowIfNull(buf);

        var blockFirstSeq = First.Seq;
        var blockLastSeq = Last.Seq;

        if (blockFirstSeq > blockLastSeq + 1)
        {
            Fs?.Warn("indexCacheBuf corrupt state in {0}: first {1} last {2}", Mfn, blockFirstSeq, blockLastSeq);
            return new InvalidDataException("corrupt state");
        }

        var idxSize = blockLastSeq - blockFirstSeq + 1;
        if (idxSize > int.MaxValue)
            return new InvalidDataException("corrupt state");

        if (CacheData != null && CacheData.Buf.Length > 0)
            JetStreamFileStore.RecycleMsgBlockBuf(CacheData.Buf);

        CacheData ??= new Cache();
        CacheData.Buf = Array.Empty<byte>();
        CacheData.Idx = Array.Empty<uint>();
        CacheData.Fseq = 0;
        CacheData.Wp = 0;
        CacheData.Nra = false;

        var idx = new List<uint>((int)idxSize);

        var populateFss = false;
        if (FssNotLoaded())
        {
            Fss = new SubjectTree<SimpleState>();
            populateFss = true;
        }

        Lsts = JetStreamFileStore.TimestampNormalized(DateTime.UtcNow);
        Ttls = 0;
        Schedules = 0;

        var offset = 0;
        ulong firstSeen = 0;
        ulong lastSeen = 0;
        while (offset < buf.Length)
        {
            if (!TryReadRecordHeader(buf.AsSpan(offset), out var seqRaw, out _, out var subjectLength, out var headerLength, out var msgLength, out var recordLength, out _))
                return new InvalidDataException("corrupt state");

            if (recordLength <= 0 || offset + recordLength > buf.Length)
                return new InvalidDataException("corrupt state");

            if ((seqRaw & FileStoreDefaults.Tbit) != 0)
            {
                offset += recordLength;
                continue;
            }

            var erased = (seqRaw & FileStoreDefaults.Ebit) != 0;
            var seq = seqRaw & ~FileStoreDefaults.Ebit & ~FileStoreDefaults.Tbit;
            if (seq <= lastSeen)
            {
                offset += recordLength;
                continue;
            }

            if (seq >= blockFirstSeq)
            {
                lastSeen = seq;
                if (firstSeen == 0)
                {
                    firstSeen = seq;
                    if (firstSeen != blockFirstSeq)
                        return new InvalidDataException("corrupt state");
                }

                var expectedSlot = (int)(seq - blockFirstSeq);
                if (expectedSlot != idx.Count)
                {
                    for (var dseq = blockFirstSeq + (ulong)idx.Count; dseq < seq; dseq++)
                    {
                        idx.Add(FileStoreDefaults.Dbit);
                        if (dseq != 0)
                            Dmap.Insert(dseq);
                    }
                }

                idx.Add((uint)offset);

                if (erased && seq != 0 && !Dmap.Exists(seq))
                    return new InvalidDataException("corrupt state");

                if (populateFss && subjectLength > 0 && !NoTrack && !erased && !Dmap.Exists(seq))
                {
                    var subjectOffset = offset + 8 + 8 + 4 + 4 + 4;
                    if (subjectOffset < 0 || subjectOffset + subjectLength > buf.Length)
                        return new InvalidDataException("corrupt state");

                    var subjectSpan = buf.AsSpan(subjectOffset, subjectLength);
                    var (state, exists) = Fss!.Find(subjectSpan);
                    if (exists && state != null)
                    {
                        state.Msgs++;
                        state.Last = seq;
                        state.LastNeedsUpdate = false;
                    }
                    else
                    {
                        Fss!.Insert(subjectSpan, new SimpleState
                        {
                            Msgs = 1,
                            First = seq,
                            Last = seq,
                            LastNeedsUpdate = false,
                        });
                    }
                }
            }

            offset += recordLength;
        }

        if (idx.Count != (int)idxSize ||
            (firstSeen > 0 && firstSeen != blockFirstSeq) ||
            (lastSeen > 0 && lastSeen != blockLastSeq))
            return new InvalidDataException("corrupt state");

        CacheData.Buf = buf;
        CacheData.Idx = idx.ToArray();
        CacheData.Fseq = blockFirstSeq;
        CacheData.Wp = buf.Length;
        return null;
    }

    // Flushes pending messages for this block.
    internal Exception? FlushPendingMsgs()
    {
        Mu.EnterWriteLock();
        try
        {
            var (lostData, err) = FlushPendingMsgsLocked();
            if (lostData != null && Fs != null)
                Fs.RebuildState(lostData);

            return err;
        }
        finally
        {
            Mu.ExitWriteLock();
        }
    }

    // Writes the buffer to disk at an explicit offset.
    // Lock should be held.
    internal (int Written, Exception? Error) WriteAt(byte[] buf, long writeOffset)
    {
        ArgumentNullException.ThrowIfNull(buf);

        if (MockWriteErr)
        {
            MockWriteErr = false;
            return (0, new IOException("mock write error"));
        }

        if (Mfd == null)
            return (0, new InvalidOperationException("message block file is not open"));

        try
        {
            Mfd.Seek(writeOffset, SeekOrigin.Begin);
            Mfd.Write(buf, 0, buf.Length);
            return (buf.Length, null);
        }
        catch (Exception ex)
        {
            return (0, ex);
        }
    }

    // flushPendingMsgsLocked writes pending bytes to disk.
    // Lock should be held.
    internal (LostStreamData? LostData, Exception? Error) FlushPendingMsgsLocked()
    {
        LostStreamData? fsLostData = null;

        if (CacheData == null || Mfd == null)
            return (null, new InvalidOperationException("no message cache"));

        var (pending, pendingErr) = BytesPending();
        if (pendingErr != null || pending == null || pending.Length == 0)
        {
            if (pendingErr is InvalidOperationException)
                return (null, null);

            return (null, pendingErr);
        }

        var writeOffset = (long)CacheData.Wp;
        var writeBuffer = pending;

        var checkErr = CheckAndLoadEncryption();
        if (checkErr != null)
            return (null, checkErr);

        if (Bek != null && writeBuffer.Length > 0)
        {
            var encrypted = new byte[writeBuffer.Length];
            writeBuffer.AsSpan().CopyTo(encrypted);
            Bek.XorKeyStream(encrypted.AsSpan());
            writeBuffer = encrypted;
        }

        while (writeBuffer.Length > 0)
        {
            var (written, writeErr) = WriteAt(writeBuffer, writeOffset);
            if (writeErr != null)
            {
                CloseFDsLockedNoCheck();
                var (lostData, _, _) = RebuildStateLocked();
                Werr = writeErr;
                return (lostData, writeErr);
            }

            writeOffset += written;
            writeBuffer = writeBuffer[written..];
        }

        Werr = null;

        if (CacheData == null || Mfd == null)
            return (fsLostData, Werr);

        CacheData.Wp = (int)writeOffset;
        if (SyncAlways)
            Mfd.Flush(flushToDisk: true);
        else
            NeedSync = true;

        if (PendingWriteSizeLocked() > 0)
            return (fsLostData, Werr);

        var now = JetStreamFileStore.TimestampNormalized(DateTime.UtcNow);
        var cexpNs = TimeSpanToNanoseconds(Cexp);
        if (now < Llts || (cexpNs > 0 && (now - Llts) <= cexpNs))
        {
            ResetCacheExpireTimerLocked(TimeSpan.Zero);
            return (fsLostData, Werr);
        }

        CacheData.Nra = false;
        ExpireCacheLocked();
        return (fsLostData, Werr);
    }

    // Lock should be held.
    internal void ClearLoading()
    {
        Loading = false;
    }

    // Loads messages from disk while holding the block lock.
    internal Exception? LoadMsgs()
    {
        Mu.EnterWriteLock();
        try
        {
            return LoadMsgsWithLock();
        }
        finally
        {
            Mu.ExitWriteLock();
        }
    }

    // Lock should be held.
    internal bool CacheAlreadyLoaded()
    {
        var cache = CacheData;
        if (cache == null || cache.Fseq == 0 || cache.Buf.Length == 0)
            return false;

        if (First.Seq < cache.Fseq)
            return false;

        var expectedEntries = Msgs + (ulong)Dmap.Size + (First.Seq - cache.Fseq);
        return expectedEntries == (ulong)cache.Idx.Length;
    }

    // Lock should be held.
    internal bool CacheNotLoaded()
        => !CacheAlreadyLoaded();

    // Lock should be held.
    internal bool FssNotLoaded()
        => Fss == null && !NoTrack;

    // Lock should be held.
    internal (FileStream? File, Exception? Error) OpenBlock()
    {
        try
        {
            return (new FileStream(Mfn, FileMode.Open, FileAccess.Read, FileShare.ReadWrite), null);
        }
        catch (Exception ex)
        {
            return (null, ex);
        }
    }

    // Loads a block file into memory.
    // Lock should be held.
    internal (byte[]? Buffer, Exception? Error) LoadBlock(byte[]? buf)
    {
        FileStream? file = null;
        var closeWhenDone = false;

        try
        {
            if (Mfd != null)
            {
                file = Mfd;
                if (file.Seek(0, SeekOrigin.Begin) != 0)
                {
                    file = null;
                    CloseFDsLockedNoCheck();
                }
            }

            if (file == null)
            {
                var (opened, openErr) = OpenBlock();
                if (openErr != null || opened == null)
                {
                    if (openErr is FileNotFoundException)
                        return (null, new FileNotFoundException("message block data missing", Mfn, openErr));

                    return (null, openErr ?? new IOException("failed to open message block"));
                }

                file = opened;
                closeWhenDone = true;
            }

            var size64 = file.Length;
            if (size64 > int.MaxValue)
                return (null, new InvalidOperationException("message block size exceeded int capacity"));

            var size = (int)size64;
            if (buf == null || buf.Length < size)
                buf = JetStreamFileStore.GetMsgBlockBuf(size);

            var readTotal = 0;
            while (readTotal < size)
            {
                var read = file.Read(buf, readTotal, size - readTotal);
                if (read <= 0)
                    return (buf[..readTotal], new EndOfStreamException("short read while loading message block"));

                readTotal += read;
            }

            RBytes = (ulong)readTotal;
            return (buf[..readTotal], null);
        }
        catch (Exception ex)
        {
            return (null, ex);
        }
        finally
        {
            if (closeWhenDone)
                file?.Dispose();
        }
    }

    // Lock should be held.
    internal Exception? LoadMsgsWithLock()
    {
        var checkErr = CheckAndLoadEncryption();
        if (checkErr != null)
            return checkErr;

        if (Loading)
            return null;

        Loading = true;
        try
        {
            var checks = 0;
            while (true)
            {
                checks++;
                if (checks > 8)
                    return new InvalidOperationException("no message cache");

                if (CacheAlreadyLoaded())
                    return null;

                Llts = JetStreamFileStore.TimestampNormalized(DateTime.UtcNow);

                var (pending, _) = BytesPending();
                if (pending is { Length: > 0 })
                {
                    var (lostData, flushErr) = FlushPendingMsgsLocked();
                    if (lostData != null && Fs != null)
                        Fs.RebuildState(lostData);
                    if (flushErr != null)
                        return flushErr;

                    continue;
                }

                var (buffer, loadErr) = LoadBlock(null);
                if (loadErr != null || buffer == null)
                {
                    Fs?.Warn("loadBlock error: {0}", loadErr?.Message ?? "unknown");
                    if (loadErr is FileNotFoundException)
                    {
                        var (lostData, _, rebuildErr) = RebuildStateLocked();
                        if (rebuildErr != null && lostData != null && Fs != null)
                            Fs.RebuildState(lostData);
                    }

                    return loadErr ?? new IOException("failed to load message block");
                }

                var cryptErr = EncryptOrDecryptIfNeeded(buffer);
                if (cryptErr != null)
                    return cryptErr;

                var (decompressed, decompressErr) = DecompressIfNeeded(buffer);
                if (decompressErr != null || decompressed == null)
                    return decompressErr ?? new InvalidDataException("failed to decompress message block");

                var indexErr = IndexCacheBuf(decompressed);
                if (indexErr != null)
                {
                    if (indexErr is InvalidDataException)
                    {
                        var (lostData, _, rebuildErr) = RebuildStateLocked();
                        if (Fs != null)
                            Fs.RebuildState(lostData);
                        if (rebuildErr != null)
                            return rebuildErr;
                        continue;
                    }

                    return indexErr;
                }

                if (decompressed.Length > 0)
                {
                    Cloads++;
                    StartCacheExpireTimer();
                }

                return null;
            }
        }
        finally
        {
            ClearLoading();
        }
    }

    // Fetches a message from this block.
    internal (StoreMsg? Message, bool ExpireOk, Exception? Error) FetchMsg(ulong seq, StoreMsg? sm)
        => FetchMsgEx(seq, sm, doCopy: true);

    // Fetches a message from this block without copy preference.
    internal (StoreMsg? Message, bool ExpireOk, Exception? Error) FetchMsgNoCopy(ulong seq, StoreMsg? sm)
        => FetchMsgEx(seq, sm, doCopy: false);

    // Fetches a message from this block.
    internal (StoreMsg? Message, bool ExpireOk, Exception? Error) FetchMsgEx(ulong seq, StoreMsg? sm, bool doCopy)
    {
        Mu.EnterWriteLock();
        try
        {
            var firstSeq = First.Seq;
            var lastSeq = Last.Seq;
            if (seq < firstSeq || seq > lastSeq)
                return (null, false, StoreErrors.ErrStoreMsgNotFound);

            if (Dmap.Exists(seq))
            {
                var previous = Llseq;
                if (Llseq == 0 || seq < Llseq || seq == Llseq + 1 || seq + 1 == Llseq)
                    Llseq = seq;

                var deletedExpireOk = (seq == lastSeq && previous == seq - 1) || (seq == firstSeq && previous == seq + 1);
                return (null, deletedExpireOk, new InvalidOperationException("deleted message"));
            }

            if (CacheNotLoaded())
            {
                var loadErr = LoadMsgsWithLock();
                if (loadErr != null)
                    return (null, false, loadErr);
            }

            var cache = CacheData;
            if (cache == null || cache.Buf.Length == 0 || cache.Idx.Length == 0 || seq < cache.Fseq)
                return (null, false, StoreErrors.ErrStoreMsgNotFound);

            var slot = seq - cache.Fseq;
            if (slot >= (ulong)cache.Idx.Length)
                return (null, false, StoreErrors.ErrStoreMsgNotFound);

            var raw = cache.Idx[(int)slot];
            if ((raw & FileStoreDefaults.Dbit) != 0)
                return (null, false, new InvalidOperationException("deleted message"));

            var offset = (int)(raw & ~(FileStoreDefaults.Dbit | FileStoreDefaults.Cbit));
            if (offset < 0 || offset >= cache.Buf.Length)
                return (null, false, new ErrBadMsg(Mfn, "invalid cache index"));

            if (!TryReadRecordHeader(cache.Buf.AsSpan(offset), out var seqRaw, out var ts, out var subjectLength, out var headerLength, out var msgLength, out var recordLength, out var parseErr))
                return (null, false, parseErr ?? new ErrBadMsg(Mfn, "malformed record"));

            if (offset + recordLength > cache.Buf.Length)
                return (null, false, new ErrBadMsg(Mfn, "record extends past cache buffer"));

            var record = cache.Buf.AsSpan(offset, recordLength);
            var payloadLength = recordLength - FileStoreDefaults.RecordHashSize;
            var payload = record[..payloadLength];
            var checksum = record[payloadLength..];
            var storedSeq = seqRaw & ~FileStoreDefaults.Ebit & ~FileStoreDefaults.Tbit;
            if (storedSeq != seq)
                return (null, false, StoreErrors.ErrStoreMsgNotFound);

            if ((raw & FileStoreDefaults.Cbit) == 0)
            {
                var computed = SHA256.HashData(payload);
                if (!computed.AsSpan(0, FileStoreDefaults.RecordHashSize).SequenceEqual(checksum))
                    return (null, false, new ErrBadMsg(Mfn, "invalid checksum"));

                cache.Idx[(int)slot] = raw | FileStoreDefaults.Cbit;
            }

            var subjectOffset = 8 + 8 + 4 + 4 + 4;
            var headerOffset = subjectOffset + subjectLength;
            var msgOffset = headerOffset + headerLength;

            var parsed = sm ?? new StoreMsg();
            parsed.Seq = seq;
            parsed.Ts = ts;
            parsed.Subject = subjectLength > 0
                ? Encoding.UTF8.GetString(record.Slice(subjectOffset, subjectLength))
                : string.Empty;

            var headerSlice = record.Slice(headerOffset, headerLength);
            var msgSlice = record.Slice(msgOffset, msgLength);
            if (doCopy)
            {
                parsed.Hdr = headerSlice.ToArray();
                parsed.Msg = msgSlice.ToArray();
            }
            else
            {
                // StoreMsg uses byte[] fields; preserving that shape still requires arrays.
                parsed.Hdr = headerSlice.ToArray();
                parsed.Msg = msgSlice.ToArray();
            }

            parsed.Buf = new byte[parsed.Hdr.Length + parsed.Msg.Length];
            if (parsed.Hdr.Length > 0)
                Buffer.BlockCopy(parsed.Hdr, 0, parsed.Buf, 0, parsed.Hdr.Length);
            if (parsed.Msg.Length > 0)
                Buffer.BlockCopy(parsed.Msg, 0, parsed.Buf, parsed.Hdr.Length, parsed.Msg.Length);

            var prevSeq = Llseq;
            if (Llseq == 0 || seq < Llseq || seq == Llseq + 1 || seq + 1 == Llseq)
                Llseq = seq;

            Lrts = JetStreamFileStore.TimestampNormalized(DateTime.UtcNow);
            var expireOk = (seq == lastSeq && prevSeq == seq - 1) || (seq == firstSeq && prevSeq == seq + 1);
            return (parsed, expireOk, null);
        }
        finally
        {
            FinishedWithCache();
            Mu.ExitWriteLock();
        }
    }

    private Exception? FlushPendingToDiskLocked()
    {
        var (_, err) = FlushPendingMsgsLocked();
        return err;
    }

    private void ResetCacheExpireTimerLocked(TimeSpan due)
    {
        if (Cexp <= TimeSpan.Zero)
            return;

        if (due <= TimeSpan.Zero)
            due = TimeSpan.FromMilliseconds(1);

        if (Ctmr == null)
        {
            Ctmr = new Timer(_ =>
            {
                Mu.EnterWriteLock();
                try
                {
                    if (!Closed)
                        ExpireCacheLocked();
                }
                finally
                {
                    Mu.ExitWriteLock();
                }
            }, null, due, Timeout.InfiniteTimeSpan);
        }
        else
        {
            _ = Ctmr.Change(due, Timeout.InfiniteTimeSpan);
        }
    }

    internal TimeSpan SinceLastActivity()
    {
        if (Closed)
            return TimeSpan.Zero;

        var now = JetStreamFileStore.TimestampNormalized(DateTime.UtcNow);
        var last = Math.Max(Math.Max(Lrts, Llts), Math.Max(Lwts, Lsts));
        var delta = now - last;
        return NanosecondsToTimeSpan(delta);
    }

    // Determine time since last write or remove of a message.
    // Read lock should be held.
    internal TimeSpan SinceLastWriteActivity()
    {
        if (Closed)
            return TimeSpan.Zero;

        var now = JetStreamFileStore.TimestampNormalized(DateTime.UtcNow);
        var last = Math.Max(Lwts, Lrts);
        return NanosecondsToTimeSpan(now - last);
    }

    private static long TimeSpanToNanoseconds(TimeSpan value)
        => value <= TimeSpan.Zero ? 0 : checked(value.Ticks * 100L);

    private static TimeSpan NanosecondsToTimeSpan(long value)
    {
        if (value <= 0)
            return TimeSpan.Zero;

        return TimeSpan.FromTicks(Math.Max(1L, value / 100L));
    }

    private static int CalculateRecordLength(string subject, byte[] mhdr, byte[] msg)
        => (8 + 8 + 4 + 4 + 4) + Encoding.UTF8.GetByteCount(subject) + mhdr.Length + msg.Length + FileStoreDefaults.RecordHashSize;

    private static byte[] BuildMessageRecord(string subject, byte[] mhdr, byte[] msg, ulong seq, long ts)
    {
        var subjectBytes = Encoding.UTF8.GetBytes(subject);
        var bodyLength = (8 + 8 + 4 + 4 + 4) + subjectBytes.Length + mhdr.Length + msg.Length;
        var record = new byte[bodyLength + FileStoreDefaults.RecordHashSize];
        var span = record.AsSpan();

        var index = 0;
        BinaryPrimitives.WriteUInt64LittleEndian(span[index..], seq);
        index += 8;
        BinaryPrimitives.WriteInt64LittleEndian(span[index..], ts);
        index += 8;
        BinaryPrimitives.WriteInt32LittleEndian(span[index..], subjectBytes.Length);
        index += 4;
        BinaryPrimitives.WriteInt32LittleEndian(span[index..], mhdr.Length);
        index += 4;
        BinaryPrimitives.WriteInt32LittleEndian(span[index..], msg.Length);
        index += 4;

        subjectBytes.CopyTo(span[index..]);
        index += subjectBytes.Length;
        mhdr.CopyTo(span[index..]);
        index += mhdr.Length;
        msg.CopyTo(span[index..]);
        index += msg.Length;

        var checksum = SHA256.HashData(span[..index]);
        checksum.AsSpan(0, FileStoreDefaults.RecordHashSize).CopyTo(span[index..]);
        return record;
    }

    private static byte[] AppendBuffer(byte[] source, byte[] append)
    {
        if (append.Length == 0)
            return source;

        var merged = new byte[source.Length + append.Length];
        if (source.Length > 0)
            Buffer.BlockCopy(source, 0, merged, 0, source.Length);
        Buffer.BlockCopy(append, 0, merged, source.Length, append.Length);
        return merged;
    }

    private static bool HasChecksum(byte[]? checksum)
    {
        if (checksum == null || checksum.Length != FileStoreDefaults.RecordHashSize)
            return false;

        foreach (var b in checksum)
        {
            if (b != 0)
                return true;
        }

        return false;
    }

    private ulong[] SnapshotDeletedSequences()
    {
        if (Dmap.Size == 0)
            return Array.Empty<ulong>();

        var deleted = new List<ulong>(Dmap.Size);
        Dmap.Range(seq =>
        {
            deleted.Add(seq);
            return true;
        });
        return deleted.ToArray();
    }

    private static LostStreamData? BuildLostData(ulong start, ulong end, ulong[] deleted, ulong bytes)
    {
        if (start == 0 || start > end)
            return null;

        var deletedSet = deleted.Length == 0 ? null : new HashSet<ulong>(deleted);
        var missing = new List<ulong>();
        for (var seq = start; seq <= end; seq++)
        {
            if (deletedSet == null || !deletedSet.Contains(seq))
                missing.Add(seq);

            if (seq == ulong.MaxValue)
                break;
        }

        if (missing.Count == 0)
            return null;

        return new LostStreamData
        {
            Msgs = missing.ToArray(),
            Bytes = bytes,
        };
    }

    private static bool TryReadRecordHeader(ReadOnlySpan<byte> record, out ulong seq, out long ts, out int subjectLength, out int headerLength, out int msgLength, out int totalLength, out Exception? error)
    {
        seq = 0;
        ts = 0;
        subjectLength = 0;
        headerLength = 0;
        msgLength = 0;
        totalLength = 0;
        error = null;

        const int fixedHeaderLength = 8 + 8 + 4 + 4 + 4;
        if (record.Length < fixedHeaderLength + FileStoreDefaults.RecordHashSize)
        {
            error = new InvalidDataException("record shorter than minimum header");
            return false;
        }

        seq = BinaryPrimitives.ReadUInt64LittleEndian(record);
        ts = BinaryPrimitives.ReadInt64LittleEndian(record[8..]);
        subjectLength = BinaryPrimitives.ReadInt32LittleEndian(record[16..]);
        headerLength = BinaryPrimitives.ReadInt32LittleEndian(record[20..]);
        msgLength = BinaryPrimitives.ReadInt32LittleEndian(record[24..]);

        if (subjectLength < 0 || headerLength < 0 || msgLength < 0)
        {
            error = new InvalidDataException("negative message section length");
            return false;
        }

        var payloadLength = fixedHeaderLength + subjectLength + headerLength + msgLength;
        if (payloadLength > record.Length || payloadLength > FileStoreDefaults.RlBadThresh)
        {
            error = new InvalidDataException("record payload length out of bounds");
            return false;
        }

        totalLength = payloadLength + FileStoreDefaults.RecordHashSize;
        if (totalLength > record.Length)
        {
            error = new InvalidDataException("record checksum exceeds block length");
            return false;
        }

        return true;
    }

    private static bool TryReadUVarInt(ReadOnlySpan<byte> source, ref int index, out ulong value)
    {
        value = 0;
        var shift = 0;
        for (var i = 0; i < MaxVarIntLength; i++)
        {
            if ((uint)index >= (uint)source.Length)
            {
                index = -1;
                value = 0;
                return false;
            }

            var b = source[index++];
            if (b < 0x80)
            {
                if (i == MaxVarIntLength - 1 && b > 1)
                {
                    index = -1;
                    value = 0;
                    return false;
                }

                value |= (ulong)b << shift;
                return true;
            }

            value |= (ulong)(b & 0x7F) << shift;
            shift += 7;
        }

        index = -1;
        value = 0;
        return false;
    }

    private static bool TryReadVarInt(ReadOnlySpan<byte> source, ref int index, out long value)
    {
        if (!TryReadUVarInt(source, ref index, out var unsigned))
        {
            value = 0;
            return false;
        }

        value = (long)(unsigned >> 1);
        if ((unsigned & 1) != 0)
            value = ~value;

        return true;
    }

    private static bool IsDeletedMsgError(Exception? ex)
        => ex is InvalidOperationException ioe &&
           string.Equals(ioe.Message, ErrDeletedMsg.Message, StringComparison.Ordinal);

    private static void TryDeleteFile(string path)
    {
        try
        {
            if (File.Exists(path))
                File.Delete(path);
        }
        catch
        {
            // best effort
        }
    }

    private static bool TryValidateRecordBuffer(ReadOnlySpan<byte> buf, out int validLength)
    {
        validLength = 0;
        while (validLength < buf.Length)
        {
            if (!TryReadRecordHeader(buf[validLength..], out _, out _, out _, out _, out _, out var recordLength, out _))
                return false;

            var record = buf.Slice(validLength, recordLength);
            var payloadLength = recordLength - FileStoreDefaults.RecordHashSize;
            var payload = record[..payloadLength];
            var checksum = record[payloadLength..];
            var computed = SHA256.HashData(payload);
            if (!computed.AsSpan(0, FileStoreDefaults.RecordHashSize).SequenceEqual(checksum))
                return false;

            validLength += recordLength;
        }

        return validLength == buf.Length;
    }

    internal void TryForceExpireCacheLocked()
    {
        var llts = Llts;
        var lwts = Lwts;
        Llts = 0;
        Lwts = 0;
        ExpireCacheLocked();
        Fss = null;
        Llts = llts;
        Lwts = lwts;
    }

    // Lock should be held.
    internal void SelectNextFirst()
    {
        if (Msgs == 0 || First.Seq == 0 || Last.Seq == 0)
            return;

        var seq = First.Seq;
        while (seq <= Last.Seq && Dmap.Exists(seq))
        {
            if (seq == ulong.MaxValue)
                break;
            seq++;
        }

        if (seq > Last.Seq)
            First = new MsgId { Seq = Last.Seq + 1, Ts = 0 };
        else
            First = new MsgId { Seq = seq, Ts = First.Ts };
    }
}

// ---------------------------------------------------------------------------
// ConsumerFileStore
// ---------------------------------------------------------------------------

/// <summary>
/// File-backed implementation of <see cref="IConsumerStore"/>.
/// Persists consumer delivery and ack state to a directory under the stream's
/// <c>obs/</c> subdirectory.
/// Mirrors the <c>consumerFileStore</c> struct in filestore.go.
/// </summary>
public sealed class ConsumerFileStore : IConsumerStore
{
    // ------------------------------------------------------------------
    // Fields — mirrors consumerFileStore struct
    // ------------------------------------------------------------------

    private readonly object _mu = new();

    /// <summary>Back-reference to the owning file store.</summary>
    private readonly JetStreamFileStore _fs;

    /// <summary>Consumer metadata (name, created time, config).</summary>
    private FileConsumerInfo _cfg;

    /// <summary>Durable consumer name.</summary>
    private readonly string _name;

    /// <summary>Path to the consumer's state directory.</summary>
    private readonly string _odir;

    /// <summary>Path to the consumer state file (<c>o.dat</c>).</summary>
    private readonly string _ifn;

    /// <summary>Consumer delivery/ack state.</summary>
    private ConsumerState _state = new();

    /// <summary>Flush-request channel.</summary>
    private Channel<byte>? _fch;

    /// <summary>Quit channel.</summary>
    private Channel<byte>? _qch;

    /// <summary>Whether a background flusher is running.</summary>
    private bool _flusher;

    /// <summary>Whether a write is currently in progress.</summary>
    private bool _writing;

    /// <summary>Whether the state is dirty (pending flush).</summary>
    private bool _dirty;

    /// <summary>Whether this consumer store is closed.</summary>
    private bool _closed;

    // ------------------------------------------------------------------
    // Constructor
    // ------------------------------------------------------------------

    /// <summary>
    /// Creates a new file-backed consumer store.
    /// </summary>
    public ConsumerFileStore(JetStreamFileStore fs, FileConsumerInfo cfg, string name, string odir)
    {
        _fs = fs;
        _cfg = cfg;
        _name = name;
        _odir = odir;
        _ifn = Path.Combine(odir, FileStoreDefaults.ConsumerState);
        _fch = Channel.CreateBounded<byte>(1);
        _qch = Channel.CreateUnbounded<byte>();
        _ = Task.Run(() => FlushLoop(_fch, _qch));

        lock (_mu)
        {
            var loadErr = LoadState();
            if (loadErr != null)
                throw loadErr;
        }
    }

    internal Exception? ConvertCipher()
    {
        byte[]? state;
        Exception? err;
        lock (_mu)
        {
            (state, err) = EncodeStateLocked();
        }

        if (err != null || state == null)
            return err ?? new InvalidOperationException("unable to encode consumer state");

        var metaErr = WriteConsumerMeta();
        if (metaErr != null)
            return metaErr;

        return WriteState(state);
    }

    // Kick flusher for this consumer.
    // Lock should be held.
    internal void KickFlusher()
    {
        if (_fch != null)
            _ = _fch.Writer.TryWrite(0);
        _dirty = true;
    }

    // Set in flusher status.
    internal void SetInFlusher()
    {
        lock (_mu)
        {
            _flusher = true;
        }
    }

    // Clear in flusher status.
    internal void ClearInFlusher()
    {
        lock (_mu)
        {
            _flusher = false;
        }
    }

    // Report in flusher status.
    internal bool InFlusher()
    {
        lock (_mu)
        {
            return _flusher;
        }
    }

    // flushLoop watches for consumer updates and the quit channel.
    internal void FlushLoop(Channel<byte>? fch, Channel<byte>? qch)
    {
        SetInFlusher();
        try
        {
            if (fch == null)
                return;

            var minTime = TimeSpan.FromMilliseconds(100);
            var lastWrite = DateTime.MinValue;

            while (true)
            {
                if (qch?.Reader.Completion.IsCompleted == true)
                    return;

                var canRead = fch.Reader.WaitToReadAsync().AsTask().GetAwaiter().GetResult();
                if (!canRead)
                    return;

                while (fch.Reader.TryRead(out _))
                {
                }

                if (lastWrite != DateTime.MinValue)
                {
                    var elapsed = DateTime.UtcNow - lastWrite;
                    if (elapsed < minTime)
                        Thread.Sleep(minTime - elapsed);
                }

                byte[]? buf;
                Exception? encodeErr;
                lock (_mu)
                {
                    if (_closed)
                        return;

                    (buf, encodeErr) = EncodeStateLocked();
                }

                if (encodeErr != null || buf == null)
                    return;

                if (WriteState(buf) == null)
                    lastWrite = DateTime.UtcNow;
            }
        }
        finally
        {
            ClearInFlusher();
        }
    }

    // ------------------------------------------------------------------
    // IConsumerStore
    // ------------------------------------------------------------------

    /// <inheritdoc/>
    public void SetStarting(ulong sseq)
    {
        byte[]? buf;
        Exception? encodeErr;
        lock (_mu)
        {
            _state.Delivered.Stream = sseq;
            _state.AckFloor.Stream = sseq;
            (buf, encodeErr) = EncodeStateLocked();
        }

        if (encodeErr != null || buf == null)
            throw encodeErr ?? new InvalidOperationException("unable to encode consumer state");

        var writeErr = WriteState(buf);
        if (writeErr != null)
            throw writeErr;
    }

    /// <inheritdoc/>
    public void UpdateStarting(ulong sseq)
    {
        lock (_mu)
        {
            if (sseq <= _state.Delivered.Stream)
                return;

            _state.Delivered.Stream = sseq;
            if (_cfg.Config.AckPolicy == AckPolicy.AckNone)
                _state.AckFloor.Stream = sseq;
            KickFlusher();
        }
    }

    /// <inheritdoc/>
    public void Reset(ulong sseq)
    {
        lock (_mu)
        {
            _state = new ConsumerState();
        }

        SetStarting(sseq);
    }

    /// <inheritdoc/>
    public bool HasState()
    {
        lock (_mu)
        {
            if (_state.Delivered.Consumer != 0 || _state.Delivered.Stream != 0)
                return true;

            return File.Exists(_ifn);
        }
    }

    /// <inheritdoc/>
    public void UpdateDelivered(ulong dseq, ulong sseq, ulong dc, long ts)
    {
        lock (_mu)
        {
            if (_closed)
                throw StoreErrors.ErrStoreClosed;

            if (dc != 1 && _cfg.Config.AckPolicy == AckPolicy.AckNone)
                throw StoreErrors.ErrNoAckPolicy;

            if (dseq <= _state.AckFloor.Consumer)
                return;

            if (_cfg.Config.AckPolicy != AckPolicy.AckNone)
            {
                _state.Pending ??= new Dictionary<ulong, Pending>();

                if (sseq <= _state.Delivered.Stream)
                {
                    if (_state.Pending.TryGetValue(sseq, out var pending) && pending != null)
                        pending.Timestamp = ts;
                }
                else
                {
                    _state.Pending[sseq] = new Pending { Sequence = dseq, Timestamp = ts };
                }

                if (dseq > _state.Delivered.Consumer)
                    _state.Delivered.Consumer = dseq;
                if (sseq > _state.Delivered.Stream)
                    _state.Delivered.Stream = sseq;

                if (dc > 1)
                {
                    var maxdc = (ulong)_cfg.Config.MaxDeliver;
                    if (maxdc > 0 && dc > maxdc)
                        _state.Pending.Remove(sseq);

                    _state.Redelivered ??= new Dictionary<ulong, ulong>();
                    if (!_state.Redelivered.TryGetValue(sseq, out var cur) || cur < dc - 1)
                        _state.Redelivered[sseq] = dc - 1;
                }
            }
            else
            {
                if (dseq > _state.Delivered.Consumer)
                {
                    _state.Delivered.Consumer = dseq;
                    _state.AckFloor.Consumer = dseq;
                }
                if (sseq > _state.Delivered.Stream)
                {
                    _state.Delivered.Stream = sseq;
                    _state.AckFloor.Stream = sseq;
                }
            }

            KickFlusher();
        }
    }

    /// <inheritdoc/>
    public void UpdateAcks(ulong dseq, ulong sseq)
    {
        lock (_mu)
        {
            if (_closed)
                throw StoreErrors.ErrStoreClosed;

            if (_cfg.Config.AckPolicy == AckPolicy.AckNone)
                throw StoreErrors.ErrNoAckPolicy;

            if (dseq <= _state.AckFloor.Consumer)
                return;

            if (_state.Pending == null || !_state.Pending.ContainsKey(sseq))
            {
                _state.Redelivered?.Remove(sseq);
                throw StoreErrors.ErrStoreMsgNotFound;
            }

            if (_cfg.Config.AckPolicy == AckPolicy.AckAll)
            {
                var sgap = sseq - _state.AckFloor.Stream;
                _state.AckFloor.Consumer = dseq;
                _state.AckFloor.Stream = sseq;

                if (sgap > (ulong)_state.Pending.Count)
                {
                    var toRemove = new List<ulong>();
                    foreach (var kv in _state.Pending)
                        if (kv.Key <= sseq)
                            toRemove.Add(kv.Key);
                    foreach (var key in toRemove)
                    {
                        _state.Pending.Remove(key);
                        _state.Redelivered?.Remove(key);
                    }
                }
                else
                {
                    for (var seq = sseq; seq > sseq - sgap && _state.Pending.Count > 0; seq--)
                    {
                        _state.Pending.Remove(seq);
                        _state.Redelivered?.Remove(seq);
                        if (seq == 0)
                            break;
                    }
                }

                KickFlusher();
                return;
            }

            if (_state.Pending.TryGetValue(sseq, out var pending) && pending != null)
            {
                _state.Pending.Remove(sseq);
                if (dseq > pending.Sequence && pending.Sequence > 0)
                    dseq = pending.Sequence;
            }

            if (_state.Pending.Count == 0)
            {
                _state.AckFloor.Consumer = _state.Delivered.Consumer;
                _state.AckFloor.Stream = _state.Delivered.Stream;
            }
            else if (dseq == _state.AckFloor.Consumer + 1)
            {
                _state.AckFloor.Consumer = dseq;
                _state.AckFloor.Stream = sseq;

                if (_state.Delivered.Consumer > dseq)
                {
                    for (var ss = sseq + 1; ss <= _state.Delivered.Stream; ss++)
                    {
                        if (_state.Pending.TryGetValue(ss, out var p) && p != null)
                        {
                            if (p.Sequence > 0)
                            {
                                _state.AckFloor.Consumer = p.Sequence - 1;
                                _state.AckFloor.Stream = ss - 1;
                            }
                            break;
                        }
                    }
                }
            }

            _state.Redelivered?.Remove(sseq);
            KickFlusher();
        }
    }

    /// <inheritdoc/>
    public void UpdateConfig(ConsumerConfig cfg)
        => _ = UpdateConfigInternal(cfg);

    // Used to update config for recovered ephemerals.
    internal Exception? UpdateConfigInternal(ConsumerConfig cfg)
    {
        lock (_mu)
        {
            _cfg.Config = cfg;
            return WriteConsumerMeta();
        }
    }

    /// <inheritdoc/>
    public void Update(ConsumerState state)
    {
        ArgumentNullException.ThrowIfNull(state);

        if (state.AckFloor.Consumer > state.Delivered.Consumer)
            throw new InvalidOperationException("bad ack floor for consumer");
        if (state.AckFloor.Stream > state.Delivered.Stream)
            throw new InvalidOperationException("bad ack floor for stream");

        lock (_mu)
        {
            if (_closed)
                throw StoreErrors.ErrStoreClosed;

            if (state.Delivered.Consumer < _state.Delivered.Consumer ||
                state.AckFloor.Stream < _state.AckFloor.Stream)
                throw new InvalidOperationException("old update ignored");

            _state = CloneState(state, copyCollections: true);
            KickFlusher();
        }
    }

    /// <inheritdoc/>
    public (ConsumerState? State, Exception? Error) State()
        => StateWithCopy(doCopy: true);

    /// <inheritdoc/>
    public (ConsumerState? State, Exception? Error) BorrowState()
        => StateWithCopy(doCopy: false);

    /// <inheritdoc/>
    public byte[] EncodedState()
    {
        lock (_mu)
        {
            if (_closed)
                throw StoreErrors.ErrStoreClosed;
            var (buf, err) = EncodeStateLocked();
            if (err != null || buf == null)
                throw err ?? new InvalidOperationException("unable to encode consumer state");
            return buf;
        }
    }

    /// <inheritdoc/>
    public StorageType Type() => StorageType.FileStorage;

    /// <inheritdoc/>
    public void Stop()
    {
        byte[]? buf = null;
        lock (_mu)
        {
            if (_closed)
                return;

            _qch?.Writer.TryComplete();
            _qch = null;

            var hasState = _state.Delivered.Consumer != 0 ||
                           _state.Delivered.Stream != 0 ||
                           _state.Pending is { Count: > 0 } ||
                           _state.Redelivered is { Count: > 0 };

            if (_dirty || hasState)
            {
                var (encoded, err) = EncodeStateLocked();
                if (err == null)
                    buf = encoded;
            }

            _closed = true;
        }

        _fs.RemoveConsumer(this);

        if (buf is { Length: > 0 })
        {
            WaitOnFlusher();
            _ = WriteState(buf);
        }
    }

    /// <inheritdoc/>
    public void Delete()
        => _ = DeleteInternal(streamDeleted: false);

    /// <inheritdoc/>
    public void StreamDelete()
        => _ = DeleteInternal(streamDeleted: true);

    internal (byte[]? Buffer, Exception? Error) EncodeState()
    {
        lock (_mu)
        {
            return EncodeStateLocked();
        }
    }

    // Lock should be held.
    internal (byte[]? Buffer, Exception? Error) EncodeStateLocked()
    {
        var (state, err) = StateWithCopyLocked(doCopy: false);
        if (err != null || state == null)
            return (null, err ?? StoreErrors.ErrStoreClosed);

        return (JsonSerializer.SerializeToUtf8Bytes(state), null);
    }

    // Will encrypt the state with the consumer key.
    // Current .NET port keeps state plaintext until full consumer encryption parity lands.
    internal (byte[]? Buffer, Exception? Error) EncryptState(byte[] buf)
    {
        ArgumentNullException.ThrowIfNull(buf);
        return (buf, null);
    }

    internal Exception? WriteState(byte[] buf)
    {
        ArgumentNullException.ThrowIfNull(buf);

        lock (_mu)
        {
            if (_writing || buf.Length == 0)
                return null;

            var (encrypted, encryptErr) = EncryptState(buf);
            if (encryptErr != null || encrypted == null)
                return encryptErr ?? new InvalidOperationException("failed to encrypt consumer state");

            buf = encrypted;
            _writing = true;
            _dirty = false;
        }

        Exception? writeErr = null;
        try
        {
            Directory.CreateDirectory(_odir);
            File.WriteAllBytes(_ifn, buf);
        }
        catch (Exception ex)
        {
            writeErr = ex;
        }

        lock (_mu)
        {
            if (writeErr != null)
                _dirty = true;
            _writing = false;
        }

        return writeErr;
    }

    // Write out consumer meta data (configuration).
    // Lock should be held.
    internal Exception? WriteConsumerMeta()
    {
        try
        {
            Directory.CreateDirectory(_odir);
            var meta = Path.Combine(_odir, FileStoreDefaults.JetStreamMetaFile);
            var b = JsonSerializer.SerializeToUtf8Bytes(_cfg);
            File.WriteAllBytes(meta, b);

            var checksum = Convert.ToHexString(SHA256.HashData(b)).ToLowerInvariant();
            var sum = Path.Combine(_odir, FileStoreDefaults.JetStreamMetaFileSum);
            File.WriteAllText(sum, checksum);
            return null;
        }
        catch (Exception ex)
        {
            return ex;
        }
    }

    internal Dictionary<ulong, Pending>? CopyPending()
    {
        if (_state.Pending is not { Count: > 0 })
            return null;

        var pending = new Dictionary<ulong, Pending>(_state.Pending.Count);
        foreach (var (seq, p) in _state.Pending)
        {
            pending[seq] = new Pending
            {
                Sequence = p.Sequence,
                Timestamp = p.Timestamp,
            };
        }

        return pending;
    }

    internal Dictionary<ulong, ulong>? CopyRedelivered()
    {
        if (_state.Redelivered is not { Count: > 0 })
            return null;

        return new Dictionary<ulong, ulong>(_state.Redelivered);
    }

    internal (ConsumerState? State, Exception? Error) StateWithCopy(bool doCopy)
    {
        lock (_mu)
        {
            return StateWithCopyLocked(doCopy);
        }
    }

    // Lock should be held.
    internal (ConsumerState? State, Exception? Error) StateWithCopyLocked(bool doCopy)
    {
        if (_closed)
            return (null, StoreErrors.ErrStoreClosed);

        if (_state.Delivered.Consumer != 0 || _state.Delivered.Stream != 0 || _state.Pending is { Count: > 0 } || _state.Redelivered is { Count: > 0 })
            return (CloneState(_state, copyCollections: doCopy), null);

        if (File.Exists(_ifn))
        {
            try
            {
                var raw = File.ReadAllBytes(_ifn);
                if (raw.Length > 0)
                {
                    var loaded = JsonSerializer.Deserialize<ConsumerState>(raw);
                    if (loaded != null)
                        _state = CloneState(loaded, copyCollections: true);
                }
            }
            catch (Exception ex)
            {
                return (null, ex);
            }
        }

        return (CloneState(_state, copyCollections: doCopy), null);
    }

    // Lock should be held at startup.
    internal Exception? LoadState()
    {
        if (File.Exists(_ifn))
        {
            var (_, err) = StateWithCopyLocked(doCopy: false);
            return err;
        }

        return null;
    }

    internal void WaitOnFlusher()
    {
        if (!InFlusher())
            return;

        var timeoutAt = DateTime.UtcNow.AddMilliseconds(100);
        while (DateTime.UtcNow < timeoutAt)
        {
            if (!InFlusher())
                return;

            Thread.Sleep(10);
        }
    }

    internal Exception? DeleteInternal(bool streamDeleted)
    {
        string? removeDir = null;
        lock (_mu)
        {
            if (_closed)
                return null;

            _qch?.Writer.TryComplete();
            _qch = null;
            _closed = true;
            removeDir = _odir;
        }

        if (!streamDeleted)
        {
            try
            {
                if (!string.IsNullOrEmpty(removeDir) && Directory.Exists(removeDir))
                    Directory.Delete(removeDir, recursive: true);
            }
            catch (Exception ex)
            {
                return ex;
            }
        }

        if (!streamDeleted)
            _fs.RemoveConsumer(this);

        return null;
    }

    private void TryLoadStateLocked()
    {
        if (!File.Exists(_ifn))
            return;

        try
        {
            var raw = File.ReadAllBytes(_ifn);
            var loaded = JsonSerializer.Deserialize<ConsumerState>(raw);
            if (loaded != null)
                _state = CloneState(loaded, copyCollections: true);
        }
        catch (Exception)
        {
            _state = new ConsumerState();
        }
    }

    private void PersistStateLocked()
    {
        if (_closed)
            return;

        Directory.CreateDirectory(_odir);
        var encoded = JsonSerializer.SerializeToUtf8Bytes(CloneState(_state, copyCollections: true));
        File.WriteAllBytes(_ifn, encoded);
        _dirty = false;
    }

    private static ConsumerState CloneState(ConsumerState state, bool copyCollections)
    {
        var clone = new ConsumerState
        {
            Delivered = new SequencePair
            {
                Consumer = state.Delivered.Consumer,
                Stream = state.Delivered.Stream,
            },
            AckFloor = new SequencePair
            {
                Consumer = state.AckFloor.Consumer,
                Stream = state.AckFloor.Stream,
            },
        };

        if (state.Pending is { Count: > 0 })
        {
            clone.Pending = new Dictionary<ulong, Pending>(state.Pending.Count);
            foreach (var kv in state.Pending)
            {
                clone.Pending[kv.Key] = new Pending
                {
                    Sequence = kv.Value.Sequence,
                    Timestamp = kv.Value.Timestamp,
                };
            }
        }
        else if (!copyCollections)
        {
            clone.Pending = state.Pending;
        }

        if (state.Redelivered is { Count: > 0 })
            clone.Redelivered = new Dictionary<ulong, ulong>(state.Redelivered);
        else if (!copyCollections)
            clone.Redelivered = state.Redelivered;

        return clone;
    }
}
