natsnet/dotnet/src/ZB.MOM.NatsNet.Server/JetStream/FileStore.cs

// 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 (fileStore struct and methods)

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

namespace ZB.MOM.NatsNet.Server;

/// <summary>
/// File-backed implementation of <see cref="IStreamStore"/>.
/// Stores JetStream messages in per-block files on disk with optional
/// encryption and compression.
/// Mirrors the <c>fileStore</c> struct in filestore.go.
/// </summary>
public sealed class JetStreamFileStore : IStreamStore, IDisposable
{
    // -----------------------------------------------------------------------
    // Fields — mirrors fileStore struct fields
    // -----------------------------------------------------------------------

    private readonly ReaderWriterLockSlim _mu = new(LockRecursionPolicy.NoRecursion);

    // State
    private StreamState _state = new();
    private List<ulong>? _tombs;
    private LostStreamData? _ld;

    // Callbacks
    private StorageUpdateHandler? _scb;
    private StorageRemoveMsgHandler? _rmcb;
    private ProcessJetStreamMsgHandler? _pmsgcb;

    // Age-check timer
    private Timer? _ageChk;
    private bool _ageChkRun;
    private long _ageChkTime;

    // Background sync timer
    private Timer? _syncTmr;

    // Configuration
    private FileStreamInfo _cfg;
    private FileStoreConfig _fcfg;
    private readonly KeyGen? _prf;
    private readonly KeyGen? _oldPrf;
    private AeadCipher? _aek;

    // Message block list and index
    private MessageBlock? _lmb;                            // last (active write) block
    private List<MessageBlock> _blks = [];
    private Dictionary<uint, MessageBlock> _bim = [];

    // Per-subject index map
    private SubjectTree<Psi>? _psim;
    private HashWheel? _ttls;
    private MsgScheduling? _scheduling;
    private readonly StreamDeletionMeta _sdm = new();

    // Total subject-list length (sum of subject-string lengths)
    private int _tsl;

    // writeFullState concurrency guard
    private readonly object _wfsmu = new();
    private long _wfsrun;   // Interlocked: is writeFullState running?
    private int _wfsadml;   // Average dmap length (protected by _wfsmu)

    // Quit / load-done channels (Channel<byte> mimics chan struct{})
    private Channel<byte>? _qch;
    private Channel<byte>? _fsld;

    // Consumer list
    private readonly ReaderWriterLockSlim _cmu = new(LockRecursionPolicy.NoRecursion);
    private List<IConsumerStore> _cfs = [];

    // Snapshot-in-progress count
    private int _sips;

    // Dirty-write counter (incremented when writes are pending flush)
    private int _dirty;

    // Lifecycle flags
    private bool _closing;
    private volatile bool _closed;

    // Flush-in-progress flag
    private bool _fip;

    // Whether the store has ever received a message
    private bool _receivedAny;

    // Whether the first sequence has been moved forward
    private bool _firstMoved;

    // Last PurgeEx call time (for throttle logic)
    private DateTime _lpex;

    // In this incremental port stage, file-store logic delegates core stream semantics
    // to the memory store implementation while file-specific APIs are added on top.
    private readonly JetStreamMemStore _memStore;
    private static readonly ArrayPool<byte> MsgBlockBufferPool = ArrayPool<byte>.Shared;
    private static readonly object InitLock = new();
    private static SemaphoreSlim? _diskIoSlots;
    private static int _diskIoCount;
    private const int ConsumerHeaderLength = 2;
    private const int MaxVarIntLength = 10;
    private const long NanosecondsPerSecond = 1_000_000_000L;
    private const byte FullStateMagic = 11;
    private const byte FullStateMinVersion = 1;
    private const byte FullStateVersion = 3;
    private const int FullStateHeaderLength = 2;
    private const int FullStateMinimumLength = 32;
    private const int FullStateChecksumLength = 8;
    private const ulong EmptyRecordLength = 30;
    private const ulong RecordLengthBadThreshold = 32UL * 1024UL * 1024UL;

    static JetStreamFileStore()
    {
        Init();
    }

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

    /// <summary>
    /// Initialises a file-backed stream store using the supplied file-store
    /// configuration and stream information.
    /// </summary>
    /// <param name="fcfg">File-store configuration (block size, cipher, paths, etc.).</param>
    /// <param name="cfg">Stream metadata (created time and stream config).</param>
    /// <exception cref="ArgumentNullException">
    /// Thrown when <paramref name="fcfg"/> or <paramref name="cfg"/> is null.
    /// </exception>
    public JetStreamFileStore(FileStoreConfig fcfg, FileStreamInfo cfg)
        : this(fcfg, cfg, null, null)
    {
    }

    internal JetStreamFileStore(FileStoreConfig fcfg, FileStreamInfo cfg, KeyGen? prf, KeyGen? oldPrf)
    {
        ArgumentNullException.ThrowIfNull(fcfg);
        ArgumentNullException.ThrowIfNull(cfg);
        ArgumentNullException.ThrowIfNull(cfg.Config);

        if (string.IsNullOrWhiteSpace(cfg.Config.Name))
            throw new ArgumentException("name required", nameof(cfg));
        if (cfg.Config.Storage != StorageType.FileStorage)
            throw new ArgumentException("file store requires file storage config", nameof(cfg));

        _fcfg = fcfg;
        _cfg = cfg;
        _prf = prf;
        _oldPrf = oldPrf;

        // Apply defaults (mirrors newFileStoreWithCreated in filestore.go).
        if (_fcfg.BlockSize == 0)
            _fcfg.BlockSize = DynBlkSize(cfg.Config.Retention, cfg.Config.MaxBytes, _prf != null);
        if (_fcfg.BlockSize > FileStoreDefaults.MaxBlockSize)
            throw new InvalidOperationException($"filestore max block size is {FileStoreDefaults.MaxBlockSize} bytes");
        if (_fcfg.CacheExpire == TimeSpan.Zero)
            _fcfg.CacheExpire = FileStoreDefaults.DefaultCacheBufferExpiration;
        if (_fcfg.SubjectStateExpire == TimeSpan.Zero)
            _fcfg.SubjectStateExpire = FileStoreDefaults.DefaultFssExpiration;
        if (_fcfg.SyncInterval == TimeSpan.Zero)
            _fcfg.SyncInterval = FileStoreDefaults.DefaultSyncInterval;

        EnsureStoreDirectoryWritable(_fcfg.StoreDir);
        Directory.CreateDirectory(Path.Combine(_fcfg.StoreDir, FileStoreDefaults.MsgDir));
        Directory.CreateDirectory(Path.Combine(_fcfg.StoreDir, FileStoreDefaults.ConsumerDir));

        _psim = new SubjectTree<Psi>();
        _bim = new Dictionary<uint, MessageBlock>();
        _qch = Channel.CreateUnbounded<byte>();
        _fsld = Channel.CreateUnbounded<byte>();

        var memCfg = cfg.Config.Clone();
        memCfg.Storage = StorageType.MemoryStorage;
        _memStore = new JetStreamMemStore(memCfg);

        var keyFile = Path.Combine(_fcfg.StoreDir, FileStoreDefaults.JetStreamMetaFileKey);
        if (_prf == null && File.Exists(keyFile))
            throw new InvalidOperationException("encrypted store requires encryption key function");
        if (_prf != null && File.Exists(keyFile))
            RecoverAEK();

        var meta = Path.Combine(_fcfg.StoreDir, FileStoreDefaults.JetStreamMetaFile);
        if (!File.Exists(meta) || new FileInfo(meta).Length == 0)
            WriteStreamMeta();
        else if (_prf != null && !File.Exists(keyFile))
            WriteStreamMeta();
    }

    /// <summary>
    /// Creates a file store using the current UTC timestamp.
    /// Mirrors Go <c>newFileStore</c>.
    /// </summary>
    public static JetStreamFileStore NewFileStore(FileStoreConfig fcfg, StreamConfig cfg)
        => NewFileStoreWithCreated(fcfg, cfg, DateTime.UtcNow, null, null);

    /// <summary>
    /// Creates a file store with an explicit creation timestamp and optional key generators.
    /// Mirrors Go <c>newFileStoreWithCreated</c>.
    /// </summary>
    public static JetStreamFileStore NewFileStoreWithCreated(
        FileStoreConfig fcfg,
        StreamConfig cfg,
        DateTime created,
        KeyGen? prf,
        KeyGen? oldPrf)
    {
        ArgumentNullException.ThrowIfNull(fcfg);
        ArgumentNullException.ThrowIfNull(cfg);

        var ccfg = cfg.Clone();
        return new JetStreamFileStore(
            fcfg,
            new FileStreamInfo { Created = created, Config = ccfg },
            prf,
            oldPrf);
    }

    internal void LockAllMsgBlocks()
    {
        foreach (var mb in _blks)
            mb.Mu.EnterWriteLock();
    }

    internal void UnlockAllMsgBlocks()
    {
        foreach (var mb in _blks)
        {
            if (mb.Mu.IsWriteLockHeld)
                mb.Mu.ExitWriteLock();
        }
    }

    internal static ulong DynBlkSize(RetentionPolicy retention, long maxBytes, bool encrypted)
    {
        if (maxBytes > 0)
        {
            var blkSize = (maxBytes / 4) + 1;
            if (blkSize % 100 != 0)
                blkSize += 100 - (blkSize % 100);

            if (blkSize <= (long)FileStoreDefaults.FileStoreMinBlkSize)
                blkSize = (long)FileStoreDefaults.FileStoreMinBlkSize;
            else if (blkSize >= (long)FileStoreDefaults.FileStoreMaxBlkSize)
                blkSize = (long)FileStoreDefaults.FileStoreMaxBlkSize;
            else
                blkSize = (long)FileStoreDefaults.DefaultMediumBlockSize;

            if (encrypted && blkSize > (long)FileStoreDefaults.MaximumEncryptedBlockSize)
                blkSize = (long)FileStoreDefaults.MaximumEncryptedBlockSize;

            return (ulong)blkSize;
        }

        if (encrypted)
            return FileStoreDefaults.MaximumEncryptedBlockSize;

        return retention == RetentionPolicy.LimitsPolicy
            ? FileStoreDefaults.DefaultLargeBlockSize
            : FileStoreDefaults.DefaultMediumBlockSize;
    }

    internal LostStreamData? LostData()
    {
        _mu.EnterReadLock();
        try
        {
            if (_ld == null)
                return null;

            return new LostStreamData
            {
                Msgs = [.. _ld.Msgs],
                Bytes = _ld.Bytes,
            };
        }
        finally
        {
            _mu.ExitReadLock();
        }
    }

    internal void AddLostData(LostStreamData? ld)
    {
        _mu.EnterWriteLock();
        try
        {
            AddLostDataLocked(ld);
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal void RemoveFromLostData(ulong seq)
    {
        _mu.EnterWriteLock();
        try
        {
            if (_ld == null)
                return;

            var index = Array.IndexOf(_ld.Msgs, seq);
            if (index < 0)
                return;

            var msgs = new ulong[_ld.Msgs.Length - 1];
            if (index > 0)
                Array.Copy(_ld.Msgs, 0, msgs, 0, index);
            if (index < _ld.Msgs.Length - 1)
                Array.Copy(_ld.Msgs, index + 1, msgs, index, _ld.Msgs.Length - index - 1);

            if (msgs.Length == 0)
                _ld = null;
            else
                _ld.Msgs = msgs;
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal void RebuildState(LostStreamData? ld)
    {
        _mu.EnterWriteLock();
        try
        {
            RebuildStateLocked(ld);
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal void RebuildStateLocked(LostStreamData? ld)
    {
        AddLostDataLocked(ld);

        _state.Msgs = 0;
        _state.Bytes = 0;
        _state.FirstSeq = 0;
        _state.LastSeq = 0;
        _state.FirstTime = default;
        _state.LastTime = default;

        foreach (var mb in _blks)
        {
            mb.Mu.EnterReadLock();
            try
            {
                _state.Msgs += mb.Msgs;
                _state.Bytes += mb.Bytes;

                var firstSeq = mb.First.Seq;
                if (_state.FirstSeq == 0 || (firstSeq < _state.FirstSeq && mb.First.Ts != 0))
                {
                    _state.FirstSeq = firstSeq;
                    _state.FirstTime = mb.First.Ts == 0 ? default : FromUnixNanosUtc(mb.First.Ts);
                }

                // Preserve the highest last-seq timestamp even when a block's terminal record is erased.
                var lastSeq = mb.Last.Seq;
                if (lastSeq >= _state.LastSeq)
                {
                    _state.LastSeq = lastSeq;
                    _state.LastTime = mb.Last.Ts == 0 ? default : FromUnixNanosUtc(mb.Last.Ts);
                }
            }
            finally
            {
                mb.Mu.ExitReadLock();
            }
        }

        _state.Lost = _ld == null
            ? null
            : new LostStreamData { Msgs = [.. _ld.Msgs], Bytes = _ld.Bytes };
    }

    internal static void UpdateTrackingState(StreamState state, MessageBlock mb)
    {
        ArgumentNullException.ThrowIfNull(state);
        ArgumentNullException.ThrowIfNull(mb);

        var first = mb.First.Seq;
        var last = mb.Last.Seq;

        if (state.FirstSeq == 0)
            state.FirstSeq = first;
        else if (first < state.FirstSeq && mb.First.Ts != 0)
            state.FirstSeq = first;

        if (last > state.LastSeq)
            state.LastSeq = last;

        state.Msgs += mb.Msgs;
        state.Bytes += mb.Bytes;
    }

    internal static bool TrackingStatesEqual(StreamState fs, StreamState mb)
    {
        ArgumentNullException.ThrowIfNull(fs);
        ArgumentNullException.ThrowIfNull(mb);

        // Brand-new state may use FirstSeq=0 while tracked block state starts at 1.
        if ((fs.FirstSeq > 1 && mb.FirstSeq > 1) || mb.FirstSeq > 1)
            return fs.Msgs == mb.Msgs && fs.FirstSeq == mb.FirstSeq && fs.LastSeq == mb.LastSeq && fs.Bytes == mb.Bytes;

        return fs.Msgs == mb.Msgs && fs.LastSeq == mb.LastSeq && fs.Bytes == mb.Bytes;
    }

    internal static List<MessageBlock>? CopyMsgBlocks(List<MessageBlock>? src)
    {
        if (src == null)
            return null;

        return [.. src];
    }

    internal static void KickFlusher(Channel<byte>? fch)
    {
        if (fch == null)
            return;

        _ = fch.Writer.TryWrite(0);
    }

    internal static ulong FileStoreMsgSizeRaw(int slen, int hlen, int mlen)
    {
        if (hlen == 0)
            return (ulong)(22 + slen + mlen + 8);

        return (ulong)(22 + slen + 4 + hlen + mlen + 8);
    }

    internal static ulong FileStoreMsgSize(string subj, byte[]? hdr, byte[]? msg)
        => FileStoreMsgSizeRaw(subj.Length, hdr?.Length ?? 0, msg?.Length ?? 0);

    internal static ulong FileStoreMsgSizeEstimate(int slen, int maxPayload)
        => (ulong)(30 + slen + 4 + maxPayload);

    internal static Exception? CheckNewHeader(byte[]? hdr)
    {
        if (hdr == null || hdr.Length < 2)
            return new InvalidDataException("corrupt state");

        if (hdr[0] != FileStoreDefaults.FileStoreMagic ||
            (hdr[1] != FileStoreDefaults.FileStoreVersion && hdr[1] != FileStoreDefaults.NewVersion))
        {
            return new InvalidDataException("corrupt state");
        }

        return null;
    }

    internal static bool SubjectsEqual(string a, string b) => a == b;

    internal static bool SubjectsAll(string a, string b) => true;

    internal static Func<string, string, bool> CompareFn(string subject)
    {
        if (string.IsNullOrEmpty(subject) || subject == ">")
            return SubjectsAll;

        if (SubscriptionIndex.SubjectHasWildcard(subject))
            return SubscriptionIndex.SubjectIsSubsetMatch;

        return SubjectsEqual;
    }

    internal static long TimestampNormalized(DateTime t)
    {
        if (t == default)
            return 0;

        var utc = t.Kind == DateTimeKind.Utc ? t : t.ToUniversalTime();
        return (utc - DateTime.UnixEpoch).Ticks * 100L;
    }

    internal static void Init()
    {
        var mp = Environment.ProcessorCount;
        var nIo = Math.Min(16, Math.Max(4, mp));
        if (mp > 32)
            nIo = Math.Max(16, Math.Min(mp, mp / 2));

        lock (InitLock)
        {
            if (_diskIoSlots != null && _diskIoCount == nIo)
                return;

            _diskIoSlots?.Dispose();
            _diskIoSlots = new SemaphoreSlim(nIo, nIo);
            _diskIoCount = nIo;
        }
    }

    internal static (byte Version, Exception? Error) CheckConsumerHeader(byte[]? hdr)
    {
        if (hdr is not { Length: >= ConsumerHeaderLength } || hdr[0] != FileStoreDefaults.FileStoreMagic)
            return (0, new InvalidDataException("corrupt state"));

        var version = hdr[1];
        return version switch
        {
            FileStoreDefaults.FileStoreVersion or FileStoreDefaults.NewVersion => (version, null),
            _ => (0, new InvalidDataException($"unsupported version: {version}")),
        };
    }

    internal static (ConsumerState? State, Exception? Error) DecodeConsumerState(byte[]? buf)
    {
        if (buf == null)
            return (null, new InvalidDataException("corrupt state"));

        var (version, headerErr) = CheckConsumerHeader(buf);
        if (headerErr != null)
            return (null, headerErr);

        var index = ConsumerHeaderLength;
        if (!TryReadUVarInt(buf, ref index, out var ackConsumer) ||
            !TryReadUVarInt(buf, ref index, out var ackStream) ||
            !TryReadUVarInt(buf, ref index, out var deliveredConsumer) ||
            !TryReadUVarInt(buf, ref index, out var deliveredStream))
        {
            return (null, new InvalidDataException("corrupt state"));
        }

        var state = new ConsumerState
        {
            AckFloor = new SequencePair { Consumer = ackConsumer, Stream = ackStream },
            Delivered = new SequencePair { Consumer = deliveredConsumer, Stream = deliveredStream },
        };

        if (version == FileStoreDefaults.FileStoreVersion)
        {
            if (state.AckFloor.Consumer > 1)
                state.Delivered.Consumer += state.AckFloor.Consumer - 1;
            if (state.AckFloor.Stream > 1)
                state.Delivered.Stream += state.AckFloor.Stream - 1;
        }

        const ulong highBit = 1UL << 63;
        if ((state.AckFloor.Stream & highBit) != 0 || (state.Delivered.Stream & highBit) != 0)
            return (null, new InvalidDataException("corrupt state"));

        if (!TryReadUVarInt(buf, ref index, out var pendingCount))
            return (null, new InvalidDataException("corrupt state"));

        if (pendingCount > 0)
        {
            if (!TryReadVarInt(buf, ref index, out var minTs))
                return (null, new InvalidDataException("corrupt state"));

            state.Pending = new Dictionary<ulong, Pending>((int)pendingCount);
            for (var i = 0; i < (int)pendingCount; i++)
            {
                if (!TryReadUVarInt(buf, ref index, out var sseq))
                    return (null, new InvalidDataException("corrupt state"));

                var dseq = 0UL;
                if (version == FileStoreDefaults.NewVersion && !TryReadUVarInt(buf, ref index, out dseq))
                    return (null, new InvalidDataException("corrupt state"));

                if (!TryReadVarInt(buf, ref index, out var ts))
                    return (null, new InvalidDataException("corrupt state"));

                sseq += state.AckFloor.Stream;
                if (sseq == 0)
                    return (null, new InvalidDataException("corrupt state"));

                if (version == FileStoreDefaults.NewVersion)
                    dseq += state.AckFloor.Consumer;

                var adjustedTs = version == FileStoreDefaults.FileStoreVersion
                    ? (ts + minTs) * NanosecondsPerSecond
                    : (minTs - ts) * NanosecondsPerSecond;

                state.Pending[sseq] = new Pending
                {
                    Sequence = dseq,
                    Timestamp = adjustedTs,
                };
            }
        }

        if (!TryReadUVarInt(buf, ref index, out var redeliveredCount))
            return (null, new InvalidDataException("corrupt state"));

        if (redeliveredCount > 0)
        {
            state.Redelivered = new Dictionary<ulong, ulong>((int)redeliveredCount);
            for (var i = 0; i < (int)redeliveredCount; i++)
            {
                if (!TryReadUVarInt(buf, ref index, out var seq) ||
                    !TryReadUVarInt(buf, ref index, out var count))
                {
                    return (null, new InvalidDataException("corrupt state"));
                }

                if (seq > 0 && count > 0)
                {
                    seq += state.AckFloor.Stream;
                    state.Redelivered[seq] = count;
                }
            }
        }

        return (state, null);
    }

    internal static Exception? WriteFileWithSync(string name, byte[] data, UnixFileMode perm)
        => WriteAtomically(name, data, perm, sync: true);

    internal static Exception? WriteAtomically(string name, byte[] data, UnixFileMode perm, bool sync)
    {
        ArgumentException.ThrowIfNullOrWhiteSpace(name);
        ArgumentNullException.ThrowIfNull(data);

        Init();
        var slots = _diskIoSlots!;
        var tmp = name + ".tmp";

        slots.Wait();
        try
        {
            var options = sync ? FileOptions.WriteThrough : FileOptions.None;
            using (var stream = new FileStream(
                       tmp,
                       FileMode.Create,
                       FileAccess.Write,
                       FileShare.None,
                       bufferSize: 4096,
                       options))
            {
                stream.Write(data, 0, data.Length);
                stream.Flush(sync);
            }

            try
            {
                File.SetUnixFileMode(tmp, perm);
            }
            catch (PlatformNotSupportedException)
            {
            }

            File.Move(tmp, name, overwrite: true);

            if (sync)
            {
                var dir = Path.GetDirectoryName(Path.GetFullPath(name));
                if (!string.IsNullOrEmpty(dir))
                {
                    try
                    {
                        using var handle = File.OpenHandle(dir, FileMode.Open, FileAccess.Read, FileShare.ReadWrite | FileShare.Delete);
                        RandomAccess.FlushToDisk(handle);
                    }
                    catch
                    {
                        // Best-effort directory metadata sync.
                    }
                }
            }

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

            return ex;
        }
        finally
        {
            slots.Release();
        }
    }

    internal static AeadCipher GenEncryptionKey(StoreCipher sc, byte[] seed)
    {
        ArgumentNullException.ThrowIfNull(seed);
        return sc switch
        {
            StoreCipher.ChaCha => new AeadCipher(seed),
            StoreCipher.Aes => new AeadCipher(seed),
            _ => throw new InvalidOperationException("unknown cipher"),
        };
    }

    internal (AeadCipher Aek, byte[] Seed, byte[] Encrypted) GenEncryptionKeys(string context)
    {
        if (_prf == null)
            throw new InvalidOperationException("encryption key function required");

        var rb = _prf(Encoding.UTF8.GetBytes(context));
        var kek = GenEncryptionKey(_fcfg.Cipher, rb);

        var seed = new byte[32];
        RandomNumberGenerator.Fill(seed);
        var aek = GenEncryptionKey(_fcfg.Cipher, seed);

        var nonce = new byte[kek.NonceSize];
        RandomNumberGenerator.Fill(nonce);
        var encryptedSeed = kek.Seal(nonce, seed);

        var encrypted = new byte[nonce.Length + encryptedSeed.Length];
        Buffer.BlockCopy(nonce, 0, encrypted, 0, nonce.Length);
        Buffer.BlockCopy(encryptedSeed, 0, encrypted, nonce.Length, encryptedSeed.Length);
        return (aek, seed, encrypted);
    }

    internal static XorStreamCipher GenBlockEncryptionKey(StoreCipher sc, byte[] seed, byte[] nonce)
    {
        ArgumentNullException.ThrowIfNull(seed);
        ArgumentNullException.ThrowIfNull(nonce);
        return sc switch
        {
            StoreCipher.ChaCha => new XorStreamCipher(seed, nonce),
            StoreCipher.Aes => new XorStreamCipher(seed, nonce),
            _ => throw new InvalidOperationException("unknown cipher"),
        };
    }

    private void AddLostDataLocked(LostStreamData? ld)
    {
        if (ld == null)
            return;

        if (_ld != null)
        {
            var known = new HashSet<ulong>(_ld.Msgs);
            var merged = new List<ulong>(_ld.Msgs);
            var added = false;

            foreach (var seq in ld.Msgs)
            {
                if (known.Add(seq))
                {
                    merged.Add(seq);
                    added = true;
                }
            }

            if (added)
            {
                merged.Sort();
                _ld.Msgs = [.. merged];
                _ld.Bytes += ld.Bytes;
            }
        }
        else
        {
            _ld = new LostStreamData
            {
                Msgs = [.. ld.Msgs],
                Bytes = ld.Bytes,
            };
        }
    }

    private static DateTime FromUnixNanosUtc(long nanos)
    {
        var seconds = nanos / 1_000_000_000L;
        var remainderNanos = nanos % 1_000_000_000L;
        return DateTimeOffset.FromUnixTimeSeconds(seconds).AddTicks(remainderNanos / 100L).UtcDateTime;
    }

    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;
    }

    internal void RecoverAEK()
    {
        if (_prf == null || _aek != null)
            return;

        var keyFile = Path.Combine(_fcfg.StoreDir, FileStoreDefaults.JetStreamMetaFileKey);
        var ekey = File.ReadAllBytes(keyFile);
        if (ekey.Length < FileStoreDefaults.MinMetaKeySize)
            throw new InvalidDataException("bad key size");

        var rb = _prf(Encoding.UTF8.GetBytes(_cfg.Config.Name));
        var kek = GenEncryptionKey(_fcfg.Cipher, rb);
        var ns = kek.NonceSize;
        if (ekey.Length <= ns)
            throw new InvalidDataException("malformed encrypted key");

        var nonce = ekey.AsSpan(0, ns);
        var payload = ekey.AsSpan(ns);
        var seed = kek.Open(nonce, payload);
        _aek = GenEncryptionKey(_fcfg.Cipher, seed);
    }

    internal void SetupAEK()
    {
        if (_prf == null || _aek != null)
            return;

        var (key, _, encrypted) = GenEncryptionKeys(_cfg.Config.Name);
        var keyFile = Path.Combine(_fcfg.StoreDir, FileStoreDefaults.JetStreamMetaFileKey);
        WriteFileWithOptionalSync(keyFile, encrypted);
        _aek = key;
    }

    internal void WriteStreamMeta()
    {
        SetupAEK();

        var payload = JsonSerializer.SerializeToUtf8Bytes(_cfg);
        if (_aek != null)
        {
            var nonce = new byte[_aek.NonceSize];
            RandomNumberGenerator.Fill(nonce);
            var encrypted = _aek.Seal(nonce, payload);
            var sealedPayload = new byte[nonce.Length + encrypted.Length];
            Buffer.BlockCopy(nonce, 0, sealedPayload, 0, nonce.Length);
            Buffer.BlockCopy(encrypted, 0, sealedPayload, nonce.Length, encrypted.Length);
            payload = sealedPayload;
        }

        var meta = Path.Combine(_fcfg.StoreDir, FileStoreDefaults.JetStreamMetaFile);
        WriteFileWithOptionalSync(meta, payload);

        var checksum = Convert.ToHexString(SHA256.HashData(payload)).ToLowerInvariant();
        var sum = Path.Combine(_fcfg.StoreDir, FileStoreDefaults.JetStreamMetaFileSum);
        WriteFileWithOptionalSync(sum, Encoding.ASCII.GetBytes(checksum));
    }

    internal static byte[] GetMsgBlockBuf(int sz)
    {
        var target = sz switch
        {
            <= (int)FileStoreDefaults.DefaultTinyBlockSize => (int)FileStoreDefaults.DefaultTinyBlockSize,
            <= (int)FileStoreDefaults.DefaultSmallBlockSize => (int)FileStoreDefaults.DefaultSmallBlockSize,
            <= (int)FileStoreDefaults.DefaultMediumBlockSize => (int)FileStoreDefaults.DefaultMediumBlockSize,
            <= (int)FileStoreDefaults.DefaultLargeBlockSize => (int)FileStoreDefaults.DefaultLargeBlockSize,
            _ => sz,
        };

        return MsgBlockBufferPool.Rent(target);
    }

    internal static void RecycleMsgBlockBuf(byte[]? buf)
    {
        if (buf == null)
            return;

        var cap = buf.Length;
        if (cap == (int)FileStoreDefaults.DefaultTinyBlockSize ||
            cap == (int)FileStoreDefaults.DefaultSmallBlockSize ||
            cap == (int)FileStoreDefaults.DefaultMediumBlockSize ||
            cap == (int)FileStoreDefaults.DefaultLargeBlockSize)
        {
            MsgBlockBufferPool.Return(buf, clearArray: false);
        }
    }

    internal bool NoTrackSubjects()
    {
        var hasSubjectIndex = _psim is { } psim && psim.Size() > 0;
        var hasSubjects = _cfg.Config.Subjects is { Length: > 0 };
        var hasMirror = _cfg.Config.Mirror != null;
        var hasSources = _cfg.Config.Sources is { Length: > 0 };
        return !(hasSubjectIndex || hasSubjects || hasMirror || hasSources);
    }

    internal MessageBlock InitMsgBlock(uint index)
    {
        var mb = new MessageBlock
        {
            Fs = this,
            Index = index,
            Cexp = _fcfg.CacheExpire,
            Fexp = _fcfg.SubjectStateExpire,
            NoTrack = NoTrackSubjects(),
            SyncAlways = _fcfg.SyncAlways,
        };

        var mdir = Path.Combine(_fcfg.StoreDir, FileStoreDefaults.MsgDir);
        Directory.CreateDirectory(mdir);
        mb.Mfn = Path.Combine(mdir, string.Format(FileStoreDefaults.BlkScan, index));
        mb.Kfn = Path.Combine(mdir, string.Format(FileStoreDefaults.KeyScan, index));
        return mb;
    }

    internal void LoadEncryptionForMsgBlock(MessageBlock mb)
    {
        ArgumentNullException.ThrowIfNull(mb);

        if (_prf == null)
            return;

        var keyPath = string.IsNullOrWhiteSpace(mb.Kfn)
            ? Path.Combine(_fcfg.StoreDir, FileStoreDefaults.MsgDir, string.Format(FileStoreDefaults.KeyScan, mb.Index))
            : mb.Kfn;

        byte[] ekey;
        if (!File.Exists(keyPath))
        {
            var (_, generatedSeed, encrypted) = GenEncryptionKeys($"{_cfg.Config.Name}:{mb.Index}");
            WriteFileWithOptionalSync(keyPath, encrypted);
            mb.Seed = generatedSeed;
            var nonceSize = GenEncryptionKey(_fcfg.Cipher, _prf(Encoding.UTF8.GetBytes($"{_cfg.Config.Name}:{mb.Index}"))).NonceSize;
            mb.Nonce = encrypted[..nonceSize];
            mb.Aek = GenEncryptionKey(_fcfg.Cipher, mb.Seed);
            mb.Bek = GenBlockEncryptionKey(_fcfg.Cipher, mb.Seed, mb.Nonce);
            return;
        }

        ekey = File.ReadAllBytes(keyPath);
        if (ekey.Length < FileStoreDefaults.MinBlkKeySize)
            throw new InvalidDataException("bad key size");

        var rb = _prf(Encoding.UTF8.GetBytes($"{_cfg.Config.Name}:{mb.Index}"));
        var kek = GenEncryptionKey(_fcfg.Cipher, rb);
        var ns = kek.NonceSize;
        if (ekey.Length <= ns)
            throw new InvalidDataException("malformed encrypted key");

        var seed = kek.Open(ekey.AsSpan(0, ns), ekey.AsSpan(ns));
        mb.Seed = seed;
        mb.Nonce = ekey[..ns];
        mb.Aek = GenEncryptionKey(_fcfg.Cipher, mb.Seed);
        mb.Bek = GenBlockEncryptionKey(_fcfg.Cipher, mb.Seed, mb.Nonce);
    }

    internal MessageBlock RecoverMsgBlock(uint index)
    {
        var mb = InitMsgBlock(index);
        if (!File.Exists(mb.Mfn))
            throw new FileNotFoundException("message block file not found", mb.Mfn);

        using var file = new FileStream(mb.Mfn, FileMode.Open, FileAccess.Read, FileShare.ReadWrite);
        mb.RBytes = (ulong)file.Length;

        try
        {
            LoadEncryptionForMsgBlock(mb);
        }
        catch
        {
            // For parity with the Go flow, recovery keeps going and falls back to rebuild paths.
        }

        var lchk = new byte[FileStoreDefaults.RecordHashSize];
        if (mb.RBytes >= (ulong)FileStoreDefaults.RecordHashSize)
        {
            file.Seek(-(long)FileStoreDefaults.RecordHashSize, SeekOrigin.End);
            file.ReadExactly(lchk, 0, lchk.Length);
        }

        var readIndexOk = TryReadBlockIndexInfo(mb, lchk);
        if (!readIndexOk)
        {
            mb.Lchk = lchk;
        }

        mb.CloseFDs();
        AddMsgBlock(mb);
        return mb;
    }

    internal void Warn(string format, params object?[] args)
    {
        if (_fcfg.Server is not NatsServer server)
            return;

        server.Warnf("Filestore [{0}] " + format, BuildLogArgs(args));
    }

    internal void Debug(string format, params object?[] args)
    {
        if (_fcfg.Server is not NatsServer server)
            return;

        server.Debugf("Filestore [{0}] " + format, BuildLogArgs(args));
    }

    internal Exception? RecoverFullState()
    {
        _mu.EnterWriteLock();
        try
        {
            RecoverPartialPurge();

            var fn = Path.Combine(_fcfg.StoreDir, FileStoreDefaults.MsgDir, FileStoreDefaults.StreamStateFile);
            byte[] raw;
            try
            {
                raw = File.ReadAllBytes(fn);
            }
            catch (FileNotFoundException ex)
            {
                return ex;
            }
            catch (DirectoryNotFoundException ex)
            {
                return ex;
            }
            catch (Exception ex)
            {
                Warn("Could not read stream state file: {0}", ex);
                return ex;
            }

            if (raw.Length < FullStateMinimumLength)
            {
                TryDeleteFile(fn);
                Warn("Stream state too short ({0} bytes)", raw.Length);
                return new InvalidDataException("corrupt state");
            }

            if (!TryValidateFullStateChecksum(raw, out var buf))
            {
                TryDeleteFile(fn);
                Warn("Stream state checksum did not match");
                return new InvalidDataException("corrupt state");
            }

            if (_prf != null && _aek != null)
            {
                var ns = _aek.NonceSize;
                if (buf.Length <= ns)
                {
                    Warn("Stream state error reading encryption key: malformed ciphertext");
                    return new InvalidDataException("corrupt state");
                }

                try
                {
                    buf = _aek.Open(buf.AsSpan(0, ns), buf.AsSpan(ns));
                }
                catch (Exception ex)
                {
                    Warn("Stream state error reading encryption key: {0}", ex);
                    return ex;
                }
            }

            if (buf.Length < FullStateHeaderLength)
            {
                TryDeleteFile(fn);
                Warn("Stream state missing header");
                return new InvalidDataException("corrupt state");
            }

            var version = buf[1];
            if (buf[0] != FullStateMagic || version < FullStateMinVersion || version > FullStateVersion)
            {
                TryDeleteFile(fn);
                Warn("Stream state magic and version mismatch");
                return new InvalidDataException("corrupt state");
            }

            var bi = FullStateHeaderLength;
            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 baseTime) ||
                !TryReadUVarInt(buf, ref bi, out var lastSeq) ||
                !TryReadVarInt(buf, ref bi, out var lastTs))
            {
                TryDeleteFile(fn);
                Warn("Stream state could not decode stream summary");
                return new InvalidDataException("corrupt state");
            }

            var recoveredState = new StreamState
            {
                Msgs = msgs,
                Bytes = bytes,
                FirstSeq = firstSeq,
                LastSeq = lastSeq,
                FirstTime = baseTime == 0 ? default : FromUnixNanosUtc(baseTime),
                LastTime = lastTs == 0 ? default : FromUnixNanosUtc(lastTs),
            };

            _psim ??= new SubjectTree<Psi>();
            _psim.Reset();
            _tsl = 0;

            if (!TryReadUVarInt(buf, ref bi, out var numSubjects))
                return CorruptStateWithDelete(fn, "Stream state missing subject metadata");

            for (var i = 0UL; i < numSubjects; i++)
            {
                if (!TryReadUVarInt(buf, ref bi, out var subjectLength))
                    return CorruptStateWithDelete(fn, "Stream state subject length decode failed");

                if (subjectLength == 0)
                    continue;

                if (bi < 0 || bi + (int)subjectLength > buf.Length)
                    return CorruptStateWithDelete(fn, $"Stream state bad subject len ({subjectLength})");

                var subject = Encoding.Latin1.GetString(buf, bi, (int)subjectLength);
                if (!SubscriptionIndex.IsValidLiteralSubject(subject))
                    return CorruptStateWithDelete(fn, "Stream state corrupt subject detected");

                bi += (int)subjectLength;
                if (!TryReadUVarInt(buf, ref bi, out var total) || !TryReadUVarInt(buf, ref bi, out var fblk))
                    return CorruptStateWithDelete(fn, "Stream state could not decode subject index");

                ulong lblk = fblk;
                if (total > 1)
                {
                    if (!TryReadUVarInt(buf, ref bi, out lblk))
                        return CorruptStateWithDelete(fn, "Stream state could not decode subject last block");
                }

                _psim.Insert(Encoding.Latin1.GetBytes(subject), new Psi
                {
                    Total = total,
                    Fblk = (uint)fblk,
                    Lblk = (uint)lblk,
                });
                _tsl += (int)subjectLength;
            }

            if (!TryReadUVarInt(buf, ref bi, out var numBlocks))
                return CorruptStateWithDelete(fn, "Stream state could not decode block count");

            var parsedBlocks = new List<MessageBlock>((int)numBlocks);
            var parsedMap = new Dictionary<uint, MessageBlock>((int)numBlocks);
            var mstate = new StreamState();
            var lastBlockIndex = numBlocks > 0 ? numBlocks - 1 : 0;

            for (var i = 0UL; i < numBlocks; i++)
            {
                if (!TryReadUVarInt(buf, ref bi, out var idx) ||
                    !TryReadUVarInt(buf, ref bi, out var nbytes) ||
                    !TryReadUVarInt(buf, ref bi, out var fseq) ||
                    !TryReadVarInt(buf, ref bi, out var fts) ||
                    !TryReadUVarInt(buf, ref bi, out var lseq) ||
                    !TryReadVarInt(buf, ref bi, out var lts) ||
                    !TryReadUVarInt(buf, ref bi, out var numDeleted))
                {
                    return CorruptStateWithDelete(fn, "Stream state block decode failed");
                }

                ulong ttls = 0;
                if (version >= 2)
                {
                    if (!TryReadUVarInt(buf, ref bi, out ttls))
                        return CorruptStateWithDelete(fn, "Stream state TTL metadata decode failed");
                }

                ulong schedules = 0;
                if (version >= 3)
                {
                    if (!TryReadUVarInt(buf, ref bi, out schedules))
                        return CorruptStateWithDelete(fn, "Stream state schedule metadata decode failed");
                }

                var mb = InitMsgBlock((uint)idx);
                mb.First = new MsgId { Seq = fseq, Ts = fts + baseTime };
                mb.Last = new MsgId { Seq = lseq, Ts = lts + baseTime };
                mb.Bytes = nbytes;
                mb.Msgs = lseq >= fseq ? (lseq - fseq + 1) : 0;
                mb.Ttls = ttls;
                mb.Schedules = schedules;
                mb.Closed = true;

                if (numDeleted > 0)
                {
                    try
                    {
                        var (dmap, consumed) = SequenceSet.Decode(buf.AsSpan(bi));
                        if (consumed <= 0)
                            return CorruptStateWithDelete(fn, "Stream state error decoding deleted map");

                        mb.Dmap = dmap;
                        mb.Msgs = mb.Msgs > numDeleted ? (mb.Msgs - numDeleted) : 0;
                        bi += consumed;
                    }
                    catch (Exception ex)
                    {
                        Warn("Stream state error decoding deleted map: {0}", ex);
                        return CorruptStateWithDelete(fn, "Stream state error decoding deleted map");
                    }
                }

                if (mb.Msgs > 0 || i == lastBlockIndex)
                {
                    parsedBlocks.Add(mb);
                    parsedMap[mb.Index] = mb;
                    UpdateTrackingState(mstate, mb);
                }
                else
                {
                    _dirty++;
                }
            }

            if (!TryReadUVarInt(buf, ref bi, out var blkIndex))
                return CorruptStateWithDelete(fn, "Stream state has no block index");

            if (bi < 0 || bi + FullStateChecksumLength > buf.Length)
                return CorruptStateWithDelete(fn, "Stream state has no checksum present");

            var lchk = buf.AsSpan(bi, FullStateChecksumLength).ToArray();

            _state = recoveredState;
            _blks = parsedBlocks;
            _bim = parsedMap;
            _lmb = _blks.Count > 0 ? _blks[^1] : null;

            if (_lmb == null || _lmb.Index != (uint)blkIndex)
                return CorruptStateWithDelete(fn, "Stream state block does not exist or index mismatch");

            if (!File.Exists(_lmb.Mfn))
            {
                Warn("Stream state detected prior state, could not locate msg block {0}", blkIndex);
                return new InvalidOperationException("prior stream state detected");
            }

            if (!LastChecksumMatches(_lmb, lchk))
            {
                Warn("Stream state outdated, last block has additional entries, will rebuild");
                return new InvalidOperationException("prior stream state detected");
            }

            var mdir = Path.Combine(_fcfg.StoreDir, FileStoreDefaults.MsgDir);
            if (Directory.Exists(mdir))
            {
                foreach (var path in Directory.EnumerateFiles(mdir, "*" + FileStoreDefaults.BlkSuffix, SearchOption.TopDirectoryOnly))
                {
                    if (!TryParseBlockIndex(Path.GetFileName(path), out var index))
                        continue;

                    if (index > blkIndex)
                    {
                        Warn("Stream state outdated, found extra blocks, will rebuild");
                        return new InvalidOperationException("prior stream state detected");
                    }

                    if (index <= uint.MaxValue && _bim.TryGetValue((uint)index, out var mb))
                        mb.Closed = false;
                }
            }

            var rebuildRequired = false;
            foreach (var mb in _blks)
            {
                if (!mb.Closed)
                    continue;

                rebuildRequired = true;
                Warn("Stream state detected prior state, could not locate msg block {0}", mb.Index);
            }

            if (rebuildRequired)
                return new InvalidOperationException("prior stream state detected");

            if (!TrackingStatesEqual(_state, mstate))
                return CorruptStateWithDelete(fn, "Stream state encountered internal inconsistency on recover");

            return null;
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    private object?[] BuildLogArgs(object?[] args)
    {
        if (args.Length == 0)
            return [_cfg.Config.Name];

        var formatted = new object?[args.Length + 1];
        formatted[0] = _cfg.Config.Name;
        Array.Copy(args, 0, formatted, 1, args.Length);
        return formatted;
    }

    private Exception CorruptStateWithDelete(string fileName, string message)
    {
        TryDeleteFile(fileName);
        Warn("{0}", message);
        return new InvalidDataException("corrupt state");
    }

    private static void TryDeleteFile(string fileName)
    {
        try
        {
            if (File.Exists(fileName))
                File.Delete(fileName);
        }
        catch
        {
            // Best effort to drop unusable snapshots.
        }
    }

    private bool TryValidateFullStateChecksum(byte[] raw, out byte[] payload)
    {
        payload = Array.Empty<byte>();
        if (raw.Length <= FullStateChecksumLength)
            return false;

        var contentLength = raw.Length - FullStateChecksumLength;
        payload = raw[..contentLength];
        var checksum = raw.AsSpan(contentLength, FullStateChecksumLength);

        var key = SHA256.HashData(Encoding.UTF8.GetBytes(_cfg.Config.Name));
        using var hmac = new HMACSHA256(key);
        var digest = hmac.ComputeHash(payload);
        return checksum.SequenceEqual(digest.AsSpan(0, FullStateChecksumLength));
    }

    private bool LastChecksumMatches(MessageBlock mb, byte[] expected)
    {
        try
        {
            using var fs = new FileStream(mb.Mfn, FileMode.Open, FileAccess.Read, FileShare.ReadWrite);
            if (fs.Length < FileStoreDefaults.RecordHashSize)
                return expected.AsSpan().SequenceEqual(new byte[FileStoreDefaults.RecordHashSize]);

            var lchk = new byte[FileStoreDefaults.RecordHashSize];
            fs.Seek(-FileStoreDefaults.RecordHashSize, SeekOrigin.End);
            fs.ReadExactly(lchk, 0, lchk.Length);
            return expected.AsSpan().SequenceEqual(lchk);
        }
        catch
        {
            return false;
        }
    }

    private static bool TryParseBlockIndex(string fileName, out ulong index)
    {
        index = 0;
        if (!fileName.EndsWith(FileStoreDefaults.BlkSuffix, StringComparison.Ordinal))
            return false;

        var stem = Path.GetFileNameWithoutExtension(fileName);
        return ulong.TryParse(stem, out index);
    }

    private static bool TryParseIndexedFile(string fileName, string format, out uint index)
    {
        index = 0;
        var suffix = Path.GetExtension(string.Format(format, 0));
        if (string.IsNullOrEmpty(suffix) || !fileName.EndsWith(suffix, StringComparison.Ordinal))
            return false;

        var stem = fileName[..^suffix.Length];
        return uint.TryParse(stem, out index);
    }

    private void RemoveRecoveredBlockLocked(MessageBlock mb)
    {
        _blks.Remove(mb);
        _bim.Remove(mb.Index);
        if (ReferenceEquals(_lmb, mb))
            _lmb = _blks.Count > 0 ? _blks[^1] : null;

        mb.CloseFDs();
        TryDeleteFile(mb.Mfn);

        var mdir = Path.Combine(_fcfg.StoreDir, FileStoreDefaults.MsgDir);
        TryDeleteFile(Path.Combine(mdir, string.Format(FileStoreDefaults.IndexScan, mb.Index)));
        TryDeleteFile(Path.Combine(mdir, string.Format(FileStoreDefaults.KeyScan, mb.Index)));
    }

    private void RecoverPartialPurge()
    {
        var storeDir = _fcfg.StoreDir;
        var msgDir = Path.Combine(storeDir, FileStoreDefaults.MsgDir);
        var purgeDir = Path.Combine(storeDir, FileStoreDefaults.PurgeDir);
        var newMsgDir = Path.Combine(storeDir, FileStoreDefaults.NewMsgDir);

        if (!Directory.Exists(msgDir))
        {
            if (Directory.Exists(newMsgDir))
            {
                Directory.Move(newMsgDir, msgDir);
            }
            else if (Directory.Exists(purgeDir))
            {
                Directory.Move(purgeDir, msgDir);
            }

            return;
        }

        if (Directory.Exists(newMsgDir))
            Directory.Delete(newMsgDir, recursive: true);
        if (Directory.Exists(purgeDir))
            Directory.Delete(purgeDir, recursive: true);
    }

    private void StartAgeChk()
    {
        if (_ageChk != null)
            return;

        if (_cfg.Config.MaxAge == TimeSpan.Zero && _ttls == null)
            return;

        var fireIn = _cfg.Config.MaxAge > TimeSpan.Zero
            ? _cfg.Config.MaxAge
            : TimeSpan.FromMilliseconds(250);
        _ageChk = new Timer(_ => ExpireMsgs(), null, fireIn, Timeout.InfiniteTimeSpan);
    }

    private void ResetAgeChk(long delta)
    {
        if (_ageChkRun)
            return;

        long next = long.MaxValue;
        if (_ttls != null)
            next = _ttls.GetNextExpiration(next);

        if (_cfg.Config.MaxAge <= TimeSpan.Zero && next == long.MaxValue)
        {
            CancelAgeChk();
            return;
        }

        var fireIn = _cfg.Config.MaxAge;
        if (delta == 0 && _state.Msgs > 0)
        {
            var until = TimeSpan.FromSeconds(2);
            if (fireIn == TimeSpan.Zero || until < fireIn)
                fireIn = until;
        }

        if (next < long.MaxValue)
        {
            var nextTicks = DateTime.UnixEpoch.Ticks + next / 100L;
            var nextUtc = new DateTime(Math.Max(nextTicks, DateTime.UnixEpoch.Ticks), DateTimeKind.Utc);
            var until = nextUtc - DateTime.UtcNow;
            if (fireIn == TimeSpan.Zero || until < fireIn)
                fireIn = until;
        }

        if (delta > 0)
        {
            var deltaDur = TimeSpan.FromTicks(delta / 100L);
            if (fireIn == TimeSpan.Zero || deltaDur < fireIn)
                fireIn = deltaDur;
        }

        if (fireIn < TimeSpan.FromMilliseconds(250))
            fireIn = TimeSpan.FromMilliseconds(250);

        var expires = DateTime.UtcNow.Ticks + fireIn.Ticks;
        if (_ageChkTime > 0 && expires > _ageChkTime)
            return;

        _ageChkTime = expires;
        if (_ageChk != null)
            _ageChk.Change(fireIn, Timeout.InfiniteTimeSpan);
        else
            _ageChk = new Timer(_ => ExpireMsgs(), null, fireIn, Timeout.InfiniteTimeSpan);
    }

    private void CancelAgeChk()
    {
        _ageChk?.Dispose();
        _ageChk = null;
        _ageChkTime = 0;
    }

    private void RunMsgScheduling()
    {
        _mu.EnterWriteLock();
        try
        {
            if (_scheduling == null)
                return;
            if (_pmsgcb == null)
            {
                _scheduling.ResetTimer();
                return;
            }

            _scheduling?.ResetTimer();
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    private void ExpireMsgs()
    {
        StoreMsg? sm = null;
        long ageDelta = 0;
        long nextTtl = long.MaxValue;
        var removeByTtl = new List<(ulong Seq, long Expires)>();

        _mu.EnterWriteLock();
        long maxAge;
        long minAge;
        var sdmTtl = (long)_cfg.Config.SubjectDeleteMarkerTTL.TotalSeconds;
        var sdmEnabled = sdmTtl > 0;
        try
        {
            maxAge = _cfg.Config.MaxAge.Ticks * 100L;
            minAge = TimestampNormalized(DateTime.UtcNow) - maxAge;
            if (sdmEnabled && (_rmcb == null || _pmsgcb == null))
            {
                ResetAgeChk(0);
                return;
            }

            _ageChkRun = true;
        }
        finally
        {
            _mu.ExitWriteLock();
        }

        if (maxAge > 0)
        {
            ulong seq = 0;
            while (true)
            {
                var (candidate, skip) = LoadNextMsg(">", wc: true, seq, null);
                if (candidate == null || candidate.Ts > minAge)
                    break;

                sm = candidate;
                seq = skip == ulong.MaxValue ? ulong.MaxValue : skip + 1;

                if (sm.Hdr is { Length: > 0 })
                {
                    var (ttl, err) = JetStreamHeaderHelpers.GetMessageTtl(sm.Hdr);
                    if (err == null && ttl < 0)
                    {
                        minAge = TimestampNormalized(DateTime.UtcNow) - maxAge;
                        continue;
                    }
                }

                if (sdmEnabled)
                {
                    var (last, shouldProcess) = ShouldProcessSdm(skip, sm.Subject);
                    if (shouldProcess)
                    {
                        var sdm = last && !StreamDeletionMeta.IsSubjectDeleteMarker(sm.Hdr);
                        HandleRemovalOrSdm(skip, sm.Subject, sdm, sdmTtl);
                    }
                }
                else
                {
                    _mu.EnterWriteLock();
                    try
                    {
                        _ = RemoveMsgViaLimits(sm.Seq);
                    }
                    finally
                    {
                        _mu.ExitWriteLock();
                    }
                }

                minAge = TimestampNormalized(DateTime.UtcNow) - maxAge;
                if (seq == ulong.MaxValue)
                    break;
            }
        }

        if (sm != null)
            ageDelta = sm.Ts - minAge;

        _mu.EnterWriteLock();
        try
        {
            if (_ttls != null)
            {
                _ttls.ExpireTasks((seq, ts) =>
                {
                    removeByTtl.Add((seq, ts));
                    return false;
                });
                nextTtl = _ttls.GetNextExpiration(long.MaxValue);
            }
        }
        finally
        {
            _mu.ExitWriteLock();
        }

        if (!sdmEnabled)
        {
            foreach (var rm in removeByTtl)
            {
                _mu.EnterWriteLock();
                try
                {
                    _ = RemoveMsgInternal(rm.Seq, secure: false, viaLimits: false, needFsLock: false);
                    try { _ttls?.Remove(rm.Seq, rm.Expires); } catch { }
                }
                finally
                {
                    _mu.ExitWriteLock();
                }
            }
        }
        else
        {
            removeByTtl.Sort(static (a, b) => a.Seq.CompareTo(b.Seq));
            foreach (var rm in removeByTtl)
            {
                _mu.EnterWriteLock();
                StoreMsg? msg;
                (bool IsLast, bool ShouldProcess) p;
                try
                {
                    msg = _memStore.LoadMsg(rm.Seq, null);
                    if (msg == null)
                    {
                        try { _ttls?.Remove(rm.Seq, rm.Expires); } catch { }
                        continue;
                    }

                    p = ShouldProcessSdmLocked(rm.Seq, msg.Subject);
                }
                finally
                {
                    _mu.ExitWriteLock();
                }

                if (p.ShouldProcess && msg != null)
                {
                    var sdm = p.IsLast && !StreamDeletionMeta.IsSubjectDeleteMarker(msg.Hdr);
                    HandleRemovalOrSdm(rm.Seq, msg.Subject, sdm, sdmTtl);
                }
            }
        }

        _mu.EnterWriteLock();
        try
        {
            _ageChkRun = false;
            _ageChkTime = 0;
            _state = _memStore.State();

            if (_state.Msgs == 0 && nextTtl == long.MaxValue)
                CancelAgeChk();
            else
                ResetAgeChk(ageDelta);
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    private (bool IsLast, bool ShouldProcess) ShouldProcessSdm(ulong seq, string subj)
    {
        _mu.EnterWriteLock();
        try
        {
            return ShouldProcessSdmLocked(seq, subj);
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    // Lock should be held.
    private (bool IsLast, bool ShouldProcess) ShouldProcessSdmLocked(ulong seq, string subj)
    {
        if (_sdm.TryGetPending(seq, out var pending))
        {
            var elapsed = TimestampNormalized(DateTime.UtcNow) - pending.Ts;
            if (elapsed < 2_000_000_000L)
                return (pending.Last, false);

            var last = pending.Last;
            if (last)
            {
                var msgs = SubjectsTotalsLocked(subj)?.GetValueOrDefault(subj, 0UL) ?? 0UL;
                var numPending = _sdm.GetSubjectTotal(subj);
                if (msgs > numPending)
                    last = false;
            }

            _sdm.SetPending(seq, new SdmBySeq { Last = last, Ts = TimestampNormalized(DateTime.UtcNow) });
            return (last, true);
        }

        var msgCount = SubjectsTotalsLocked(subj)?.GetValueOrDefault(subj, 0UL) ?? 0UL;
        if (msgCount == 0)
            return (false, true);

        var pendingCount = _sdm.GetSubjectTotal(subj);
        var remaining = msgCount > pendingCount ? msgCount - pendingCount : 0;
        return (_sdm.TrackPending(seq, subj, remaining == 1), true);
    }

    private void HandleRemovalOrSdm(ulong seq, string subj, bool sdm, long sdmTtl)
    {
        if (!sdm)
        {
            _rmcb?.Invoke(seq);
            return;
        }

        var hdr = Encoding.ASCII.GetBytes(
            $"{NatsHeaderConstants.HdrLine}" +
            $"{NatsHeaderConstants.JsMarkerReason}: {NatsHeaderConstants.JsMarkerReasonMaxAge}\r\n" +
            $"{NatsHeaderConstants.JsMessageTtl}: {TimeSpan.FromSeconds(sdmTtl)}\r\n" +
            $"{NatsHeaderConstants.JsMsgRollup}: {NatsHeaderConstants.JsMsgRollupSubject}\r\n\r\n");
        _pmsgcb?.Invoke(new StoreMsg
        {
            Subject = subj,
            Hdr = hdr,
            Msg = Array.Empty<byte>(),
            Seq = 0,
            Ts = 0,
        });
    }

    internal Exception? RecoverTTLState()
    {
        _mu.EnterWriteLock();
        try
        {
            var fn = Path.Combine(_fcfg.StoreDir, FileStoreDefaults.MsgDir, FileStoreDefaults.TtlStreamStateFile);
            byte[]? buf = null;

            try
            {
                if (File.Exists(fn))
                    buf = File.ReadAllBytes(fn);
            }
            catch (Exception ex)
            {
                return ex;
            }

            _ttls = HashWheel.NewHashWheel();

            ulong ttlSeq = 0;
            if (buf != null)
            {
                try
                {
                    ttlSeq = _ttls.Decode(buf);
                }
                catch (Exception ex)
                {
                    Warn("Error decoding TTL state: {0}", ex);
                    TryDeleteFile(fn);
                }
            }

            if (ttlSeq < _state.FirstSeq)
                ttlSeq = _state.FirstSeq;

            try
            {
                if (_state.Msgs > 0 && ttlSeq <= _state.LastSeq)
                {
                    Warn("TTL state is outdated; attempting to recover using linear scan (seq {0} to {1})", ttlSeq, _state.LastSeq);
                    foreach (var mb in _blks)
                    {
                        mb.Mu.EnterReadLock();
                        try
                        {
                            if (mb.Ttls == 0 || mb.Last.Seq < ttlSeq)
                                continue;

                            var start = Math.Max(ttlSeq, mb.First.Seq);
                            var end = mb.Last.Seq;
                            for (var seq = start; seq <= end; seq++)
                            {
                                StoreMsg? sm = null;
                                try
                                {
                                    sm = LoadMsg(seq, null);
                                }
                                catch (Exception ex)
                                {
                                    Warn("Error loading msg seq {0} for recovering TTL: {1}", seq, ex);
                                }

                                if (sm?.Hdr is not { Length: > 0 })
                                {
                                    if (seq == ulong.MaxValue)
                                        break;
                                    continue;
                                }

                                var (ttl, _) = JetStreamHeaderHelpers.GetMessageTtl(sm.Hdr);
                                if (ttl > 0)
                                {
                                    var expires = sm.Ts + (ttl * NanosecondsPerSecond);
                                    _ttls.Add(seq, expires);
                                }

                                if (seq == ulong.MaxValue)
                                    break;
                            }
                        }
                        finally
                        {
                            mb.Mu.ExitReadLock();
                        }
                    }
                }
            }
            finally
            {
                ResetAgeChk(0);
            }

            return null;
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal Exception? RecoverMsgSchedulingState()
    {
        _mu.EnterWriteLock();
        try
        {
            var fn = Path.Combine(_fcfg.StoreDir, FileStoreDefaults.MsgDir, FileStoreDefaults.MsgSchedulingStreamStateFile);
            byte[]? buf = null;

            try
            {
                if (File.Exists(fn))
                    buf = File.ReadAllBytes(fn);
            }
            catch (Exception ex)
            {
                return ex;
            }

            _scheduling = new MsgScheduling(RunMsgScheduling);

            ulong schedSeq = 0;
            if (buf != null)
            {
                var (decodedSeq, err) = _scheduling.Decode(buf);
                schedSeq = decodedSeq;
                if (err != null)
                {
                    Warn("Error decoding message scheduling state: {0}", err);
                    TryDeleteFile(fn);
                    schedSeq = 0;
                }
            }

            if (schedSeq < _state.FirstSeq)
                schedSeq = _state.FirstSeq;

            try
            {
                if (_state.Msgs > 0 && schedSeq <= _state.LastSeq)
                {
                    Warn("Message scheduling state is outdated; attempting to recover using linear scan (seq {0} to {1})", schedSeq, _state.LastSeq);
                    foreach (var mb in _blks)
                    {
                        mb.Mu.EnterReadLock();
                        try
                        {
                            if (mb.Schedules == 0 || mb.Last.Seq < schedSeq)
                                continue;

                            var start = Math.Max(schedSeq, mb.First.Seq);
                            var end = mb.Last.Seq;
                            for (var seq = start; seq <= end; seq++)
                            {
                                StoreMsg? sm = null;
                                try
                                {
                                    sm = LoadMsg(seq, null);
                                }
                                catch (Exception ex)
                                {
                                    Warn("Error loading msg seq {0} for recovering message schedules: {1}", seq, ex);
                                }

                                if (sm?.Hdr is not { Length: > 0 })
                                {
                                    if (seq == ulong.MaxValue)
                                        break;
                                    continue;
                                }

                                var (schedule, ok) = JetStreamHeaderHelpers.NextMessageSchedule(sm.Hdr, sm.Ts);
                                if (ok && schedule != default)
                                    _scheduling.Init(seq, sm.Subject, schedule.Ticks * 100L);

                                if (seq == ulong.MaxValue)
                                    break;
                            }
                        }
                        finally
                        {
                            mb.Mu.ExitReadLock();
                        }
                    }
                }
            }
            finally
            {
                _scheduling.ResetTimer();
            }

            return null;
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal void CleanupOldMeta()
    {
        string mdir;
        _mu.EnterReadLock();
        try
        {
            mdir = Path.Combine(_fcfg.StoreDir, FileStoreDefaults.MsgDir);
        }
        finally
        {
            _mu.ExitReadLock();
        }

        if (!Directory.Exists(mdir))
            return;

        foreach (var file in Directory.EnumerateFiles(mdir, "*", SearchOption.TopDirectoryOnly))
        {
            var name = Path.GetFileName(file);
            if (!name.EndsWith(".idx", StringComparison.Ordinal) &&
                !name.EndsWith(".fss", StringComparison.Ordinal))
            {
                continue;
            }

            TryDeleteFile(file);
        }
    }

    internal Exception? RecoverMsgs()
    {
        _mu.EnterWriteLock();
        try
        {
            RecoverPartialPurge();
            var mdir = Path.Combine(_fcfg.StoreDir, FileStoreDefaults.MsgDir);
            if (!Directory.Exists(mdir))
                return new InvalidOperationException("message directory is not readable");

            var indices = new List<uint>();
            foreach (var file in Directory.EnumerateFiles(mdir, "*" + FileStoreDefaults.BlkSuffix, SearchOption.TopDirectoryOnly))
            {
                if (!TryParseBlockIndex(Path.GetFileName(file), out var index) || index > uint.MaxValue)
                    continue;

                indices.Add((uint)index);
            }

            indices.Sort();

            _blks.Clear();
            _bim.Clear();
            _lmb = null;
            _state = new StreamState();

            foreach (var index in indices)
            {
                MessageBlock mb;
                try
                {
                    mb = RecoverMsgBlock(index);
                }
                catch (Exception ex)
                {
                    return ex;
                }

                mb.Mu.EnterWriteLock();
                try
                {
                    if (mb.First.Seq == 0)
                    {
                        RemoveRecoveredBlockLocked(mb);
                        continue;
                    }

                    if (_state.FirstSeq == 0 || (mb.First.Seq < _state.FirstSeq && mb.First.Ts != 0))
                    {
                        _state.FirstSeq = mb.First.Seq;
                        _state.FirstTime = mb.First.Ts == 0 ? default : FromUnixNanosUtc(mb.First.Ts);
                    }

                    if (mb.Last.Seq >= _state.LastSeq)
                    {
                        _state.LastSeq = mb.Last.Seq;
                        _state.LastTime = mb.Last.Ts == 0 ? default : FromUnixNanosUtc(mb.Last.Ts);
                    }

                    _state.Msgs += mb.Msgs;
                    _state.Bytes += mb.Bytes;

                    if (mb.Msgs == 0)
                    {
                        mb.First.Seq = _state.LastSeq + 1;
                        mb.First.Ts = 0;
                        mb.Last.Seq = _state.LastSeq;
                        mb.Last.Ts = TimestampNormalized(_state.LastTime);
                    }
                }
                finally
                {
                    if (mb.Mu.IsWriteLockHeld)
                        mb.Mu.ExitWriteLock();
                }
            }

            if (_blks.Count == 0)
            {
                AddMsgBlock(InitMsgBlock(1));
            }
            else
            {
                _lmb = _blks[^1];
            }

            var valid = new HashSet<uint>(_blks.Select(b => b.Index));
            foreach (var file in Directory.EnumerateFiles(mdir, "*" + Path.GetExtension(string.Format(FileStoreDefaults.KeyScan, 0)), SearchOption.TopDirectoryOnly))
            {
                var name = Path.GetFileName(file);
                if (!TryParseIndexedFile(name, FileStoreDefaults.KeyScan, out var index) || !valid.Contains(index))
                    TryDeleteFile(file);
            }

            return null;
        }
        catch (Exception ex)
        {
            return ex;
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal Exception? ExpireMsgsOnRecover()
    {
        _mu.EnterWriteLock();
        try
        {
            if (_state.Msgs == 0)
                return null;
            if (_cfg.Config.SubjectDeleteMarkerTTL > TimeSpan.Zero)
                return null;
            if (_cfg.Config.MaxAge <= TimeSpan.Zero)
                return null;

            var minAge = (DateTime.UtcNow - DateTime.UnixEpoch).Ticks * 100L - (_cfg.Config.MaxAge.Ticks * 100L);

            ulong purged = 0;
            ulong bytes = 0;
            long nts = 0;
            var removed = new List<MessageBlock>();
            MsgId last = default;

            foreach (var mb in _blks)
            {
                mb.Mu.EnterWriteLock();
                try
                {
                    if (minAge < mb.First.Ts)
                    {
                        nts = mb.First.Ts;
                        break;
                    }

                    if (mb.Last.Ts > minAge)
                    {
                        nts = mb.First.Ts;
                        break;
                    }

                    purged += mb.Msgs;
                    bytes += mb.Bytes;
                    if (ReferenceEquals(mb, _lmb))
                        last = mb.Last;

                    RemoveRecoveredBlockLocked(mb);
                    removed.Add(mb);
                }
                finally
                {
                    if (mb.Mu.IsWriteLockHeld)
                        mb.Mu.ExitWriteLock();
                }
            }

            if (removed.Count > 0)
            {
                _blks = [.. _blks.Where(mb => !removed.Contains(mb))];
                if (_bim.Count > 0)
                {
                    foreach (var mb in removed)
                        _bim.Remove(mb.Index);
                }

                _lmb = _blks.Count > 0 ? _blks[^1] : null;

                if (_lmb == null && last.Seq != 0)
                {
                    var tomb = InitMsgBlock(1);
                    tomb.First = new MsgId { Seq = last.Seq + 1, Ts = 0 };
                    tomb.Last = new MsgId { Seq = last.Seq, Ts = last.Ts };
                    tomb.Msgs = 0;
                    tomb.Bytes = 0;
                    AddMsgBlock(tomb);
                    _psim?.Reset();
                    _tsl = 0;
                }
            }

            RebuildStateLocked(null);

            if (nts > 0)
                ResetAgeChk(nts - minAge);
            if (purged > 0 || bytes > 0)
                _dirty++;

            return null;
        }
        catch (Exception ex)
        {
            return ex;
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    private static void EnsureStoreDirectoryWritable(string storeDir)
    {
        if (string.IsNullOrWhiteSpace(storeDir))
            throw new ArgumentException("store directory required", nameof(storeDir));

        Directory.CreateDirectory(storeDir);
        var dirInfo = new DirectoryInfo(storeDir);
        if (!dirInfo.Exists)
            throw new InvalidOperationException("storage directory is not accessible");

        var probe = Path.Combine(storeDir, $"_test_{Guid.NewGuid():N}.tmp");
        using (File.Create(probe))
        {
        }

        File.Delete(probe);
    }

    private void WriteFileWithOptionalSync(string path, byte[] payload)
        => WriteFileWithOptionalSync(path, payload, UnixFileMode.UserRead | UnixFileMode.UserWrite);

    private void WriteFileWithOptionalSync(string path, byte[] payload, UnixFileMode perm)
    {
        Directory.CreateDirectory(Path.GetDirectoryName(path)!);
        var err = WriteAtomically(path, payload, perm, _fcfg.SyncAlways);
        if (err != null)
            throw err;
    }

    private bool TryReadBlockIndexInfo(MessageBlock mb, byte[] lchk)
    {
        var idxFile = Path.Combine(_fcfg.StoreDir, FileStoreDefaults.MsgDir, string.Format(FileStoreDefaults.IndexScan, mb.Index));
        if (!File.Exists(idxFile))
            return false;

        try
        {
            var data = JsonSerializer.Deserialize<MessageBlockIndexFile>(File.ReadAllBytes(idxFile));
            if (data == null || data.LastChecksum == null)
                return false;
            if (!lchk.AsSpan().SequenceEqual(data.LastChecksum))
                return false;

            mb.Msgs = data.Msgs;
            mb.Bytes = data.Bytes;
            mb.RBytes = data.RawBytes;
            mb.NoTrack = data.NoTrack;
            mb.First = new MsgId { Seq = data.FirstSeq, Ts = data.FirstTs };
            mb.Last = new MsgId { Seq = data.LastSeq, Ts = data.LastTs };
            mb.Lchk = data.LastChecksum;
            return true;
        }
        catch
        {
            return false;
        }
    }

    private void AddMsgBlock(MessageBlock mb)
    {
        if (!_bim.ContainsKey(mb.Index))
            _blks.Add(mb);
        _bim[mb.Index] = mb;
        _blks.Sort((a, b) => a.Index.CompareTo(b.Index));
        _lmb = _blks.Count > 0 ? _blks[^1] : null;
    }

    // -----------------------------------------------------------------------
    // Write/lifecycle helper methods (Batch 14 Group 1)
    // -----------------------------------------------------------------------

    // Lock should be held.
    private Dictionary<string, ulong>? SubjectsTotalsLocked(string filter)
    {
        if (_psim == null || _psim.Size() == 0)
            return null;

        if (string.IsNullOrEmpty(filter))
            filter = ">";

        var totals = new Dictionary<string, ulong>(StringComparer.Ordinal);
        _psim.Match(Encoding.UTF8.GetBytes(filter), (subj, psi) =>
        {
            totals[Encoding.UTF8.GetString(subj)] = psi.Total;
            return true;
        });

        return totals;
    }

    // Lock should be held.
    private byte[] HashKeyForBlock(uint index)
        => Encoding.UTF8.GetBytes($"{_cfg.Config.Name}-{index}");

    // Lock should be held.
    private (MessageBlock? Mb, Exception? Error) NewMsgBlockForWrite()
    {
        uint index = 1;
        byte[]? recycledBuf = null;

        if (_lmb != null)
        {
            _lmb.Mu.EnterWriteLock();
            try
            {
                index = _lmb.Index + 1;
                _lmb.Qch = null;
                _lmb.CloseFDs();
                if (_lmb.CacheData?.Buf is { Length: > 0 } buf)
                {
                    recycledBuf = buf;
                    _lmb.CacheData = null;
                }
            }
            finally
            {
                _lmb.Mu.ExitWriteLock();
            }
        }

        var mb = InitMsgBlock(index);
        mb.Mu.EnterWriteLock();
        try
        {
            mb.Fss = new SubjectTree<SimpleState>();
            mb.Llts = 0;
            mb.Lwts = TimestampNormalized(DateTime.UtcNow);
            mb.First = new MsgId { Seq = _state.LastSeq + 1, Ts = 0 };
            mb.Last = new MsgId { Seq = _state.LastSeq, Ts = TimestampNormalized(_state.LastTime) };
            mb.CacheData = new Cache
            {
                Buf = recycledBuf ?? GetMsgBlockBuf((int)FileStoreDefaults.DefaultTinyBlockSize),
                Fseq = mb.First.Seq,
                Idx = Array.Empty<uint>(),
                Nra = false,
                Wp = 0,
            };
        }
        finally
        {
            mb.Mu.ExitWriteLock();
        }

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

        if (_prf != null)
        {
            var encErr = GenEncryptionKeysForBlock(mb);
            if (encErr != null)
                return (null, encErr);
        }

        _ = SHA256.HashData(HashKeyForBlock(index));
        AddMsgBlock(mb);
        return (mb, null);
    }

    // Lock should be held.
    private Exception? GenEncryptionKeysForBlock(MessageBlock? mb)
    {
        if (mb == null)
            return null;

        try
        {
            var (key, seed, encrypted) = GenEncryptionKeys($"{_cfg.Config.Name}:{mb.Index}");
            mb.Seed = seed;
            mb.Nonce = encrypted[..key.NonceSize];
            mb.Aek = GenEncryptionKey(_fcfg.Cipher, mb.Seed);
            mb.Bek = GenBlockEncryptionKey(_fcfg.Cipher, mb.Seed, mb.Nonce);

            var mdir = Path.Combine(_fcfg.StoreDir, FileStoreDefaults.MsgDir);
            var keyFile = Path.Combine(mdir, string.Format(FileStoreDefaults.KeyScan, mb.Index));
            WriteFileWithOptionalSync(keyFile, encrypted);
            mb.Kfn = keyFile;
            return null;
        }
        catch (Exception ex)
        {
            return ex;
        }
    }

    internal Exception? RegenerateEncryptionKeysForBlock(MessageBlock mb)
    {
        ArgumentNullException.ThrowIfNull(mb);

        _mu.EnterWriteLock();
        try
        {
            return GenEncryptionKeysForBlock(mb);
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    // Lock should be held.
    private Exception? StoreRawMsgInternal(string subject, byte[]? hdr, byte[]? msg, ulong seq, long ts, long ttl, bool discardNewCheck)
    {
        if (IsClosed())
            return StoreErrors.ErrStoreClosed;

        hdr ??= Array.Empty<byte>();
        msg ??= Array.Empty<byte>();

        if (_lmb == null)
        {
            var (newMb, newErr) = NewMsgBlockForWrite();
            if (newErr != null)
                return newErr;
            _lmb = newMb;
        }

        var projectedSize = _lmb!.RBytes + FileStoreMsgSize(subject, hdr, msg);
        if (_fcfg.BlockSize > 0 && projectedSize > _fcfg.BlockSize)
        {
            var (newMb, newErr) = NewMsgBlockForWrite();
            if (newErr != null)
                return newErr;
            _lmb = newMb;
        }

        try
        {
            _memStore.StoreRawMsg(subject, hdr, msg, seq, ts, ttl, discardNewCheck);
        }
        catch (Exception ex)
        {
            return ex;
        }

        _state = _memStore.State();
        _receivedAny = true;

        var record = BuildMessageRecord(subject, hdr, msg, seq, ts);
        _lmb!.Mu.EnterWriteLock();
        try
        {
            _lmb.Mfd ??= new FileStream(_lmb.Mfn, FileMode.OpenOrCreate, FileAccess.ReadWrite, FileShare.Read);
            _lmb.Mfd.Seek(0, SeekOrigin.End);
            _lmb.Mfd.Write(record, 0, record.Length);
            if (_fcfg.SyncAlways)
                _lmb.Mfd.Flush(true);

            var msgSize = FileStoreMsgSize(subject, hdr, msg);
            _lmb.Msgs++;
            _lmb.Bytes += msgSize;
            _lmb.RBytes += (ulong)record.Length;
            _lmb.CBytes = _lmb.Bytes;
            _lmb.NeedSync = !_fcfg.SyncAlways;
            _lmb.Lwts = ts;
            _lmb.Lchk = record.AsSpan(record.Length - FileStoreDefaults.RecordHashSize).ToArray();

            if (_lmb.First.Seq == 0 || _lmb.Msgs == 1 || seq < _lmb.First.Seq)
                _lmb.First = new MsgId { Seq = seq, Ts = ts };
            if (seq >= _lmb.Last.Seq)
                _lmb.Last = new MsgId { Seq = seq, Ts = ts };

            _lmb.Fss ??= new SubjectTree<SimpleState>();
            if (!string.IsNullOrEmpty(subject))
            {
                var subj = Encoding.UTF8.GetBytes(subject);
                var (ss, ok) = _lmb.Fss.Find(subj);
                if (ok && ss != null)
                {
                    ss.Msgs++;
                    if (ss.First == 0 || seq < ss.First)
                        ss.First = seq;
                    if (seq > ss.Last)
                        ss.Last = seq;
                    ss.LastNeedsUpdate = false;
                }
                else
                {
                    _lmb.Fss.Insert(subj, new SimpleState
                    {
                        Msgs = 1,
                        First = seq,
                        Last = seq,
                        LastNeedsUpdate = false,
                    });
                }
            }
        }
        finally
        {
            _lmb.Mu.ExitWriteLock();
        }

        RefreshPsiFromMemStoreLocked();
        EnforceMsgLimit();
        EnforceBytesLimit();
        EnforceMsgPerSubjectLimit(fireCallback: false);
        return null;
    }

    // Lock should be held.
    private void RebuildFirst()
    {
        _state = _memStore.State();
        if (_state.Msgs == 0)
        {
            _state.FirstSeq = _state.LastSeq + 1;
            _state.FirstTime = default;
            return;
        }

        var ss = _memStore.FilteredState(_state.FirstSeq, ">");
        if (ss.Msgs == 0 || ss.First == 0)
        {
            _state.FirstSeq = _state.LastSeq + 1;
            _state.FirstTime = default;
            return;
        }

        _state.FirstSeq = ss.First;
        var sm = _memStore.LoadMsg(ss.First, null);
        if (sm != null)
            _state.FirstTime = FromUnixNanosUtc(sm.Ts);
        SelectNextFirst();
    }

    // Lock should be held.
    private (ulong Seq, Exception? Error) FirstSeqForSubj(string subj)
    {
        if (string.IsNullOrEmpty(subj))
            return (0, null);

        try
        {
            var ss = _memStore.FilteredState(_state.FirstSeq, subj);
            return (ss.Msgs > 0 ? ss.First : 0, null);
        }
        catch (Exception ex)
        {
            return (0, ex);
        }
    }

    // Lock should be held.
    private void EnforceMsgLimit()
    {
        if (_cfg.Config.MaxMsgs <= 0)
            return;

        var maxMsgs = (ulong)_cfg.Config.MaxMsgs;
        while (_state.Msgs > maxMsgs)
        {
            var (removed, err) = DeleteFirstMsg();
            if (err != null || !removed)
            {
                RebuildFirst();
                break;
            }
        }
    }

    // Lock should be held.
    private void EnforceBytesLimit()
    {
        if (_cfg.Config.MaxBytes <= 0)
            return;

        var maxBytes = (ulong)_cfg.Config.MaxBytes;
        while (_state.Bytes > maxBytes)
        {
            var (removed, err) = DeleteFirstMsg();
            if (err != null || !removed)
            {
                RebuildFirst();
                break;
            }
        }
    }

    // Lock should be held.
    private void EnforceMsgPerSubjectLimit(bool fireCallback)
    {
        if (_cfg.Config.MaxMsgsPer <= 0)
            return;

        var limit = (ulong)_cfg.Config.MaxMsgsPer;
        var totals = _memStore.SubjectsTotals(">");
        foreach (var entry in totals.ToArray())
        {
            var subj = entry.Key;
            var remaining = entry.Value;
            while (remaining > limit)
            {
                var (fseq, seqErr) = FirstSeqForSubj(subj);
                if (seqErr != null || fseq == 0)
                    break;

                var (removed, remErr) = RemoveMsgViaLimits(fseq);
                if (remErr != null || !removed)
                    break;

                remaining--;
            }
        }

        if (fireCallback)
            _state = _memStore.State();
    }

    // Lock should be held.
    private (bool Removed, Exception? Error) DeleteFirstMsg()
        => RemoveMsgViaLimits(_state.FirstSeq);

    // Lock should be held.
    private (bool Removed, Exception? Error) RemoveMsgViaLimits(ulong seq)
        => RemoveMsgInternal(seq, secure: false, viaLimits: true, needFsLock: false);

    // Lock should be held.
    private ulong RemovePerSubject(string subj)
    {
        if (_psim == null || string.IsNullOrEmpty(subj))
            return 0;

        var subjBytes = Encoding.UTF8.GetBytes(subj);
        var (psi, ok) = _psim.Find(subjBytes);
        if (!ok || psi == null)
            return 0;

        _psim.Delete(subjBytes);
        _tsl = Math.Max(0, _tsl - subj.Length);
        return psi.Total;
    }

    private (bool Removed, Exception? Error) RemoveMsgInternal(ulong seq, bool secure, bool viaLimits, bool needFsLock)
    {
        if (needFsLock)
            _mu.EnterWriteLock();

        try
        {
            if (IsClosed())
                return (false, StoreErrors.ErrStoreClosed);

            var sm = _memStore.LoadMsg(seq, null);
            var (removed, err) = secure ? _memStore.EraseMsg(seq) : _memStore.RemoveMsg(seq);
            if (err != null || !removed)
                return (removed, err);

            _state = _memStore.State();

            var mb = FindMsgBlockForSeq(seq);
            if (mb != null)
            {
                var (_, blockErr) = RemoveMsgFromBlock(mb, seq, secure, viaLimits, sm);
                if (blockErr != null)
                    return (true, blockErr);
            }

            RefreshPsiFromMemStoreLocked();
            if (_state.Msgs == 0)
                _firstMoved = true;
            return (true, null);
        }
        finally
        {
            if (needFsLock)
                _mu.ExitWriteLock();
        }
    }

    // Lock should be held.
    private (bool Removed, Exception? Error) RemoveMsgFromBlock(MessageBlock mb, ulong seq, bool secure, bool viaLimits)
        => RemoveMsgFromBlock(mb, seq, secure, viaLimits, null);

    // Lock should be held.
    private (bool Removed, Exception? Error) RemoveMsgFromBlock(MessageBlock mb, ulong seq, bool secure, bool viaLimits, StoreMsg? sm)
    {
        ArgumentNullException.ThrowIfNull(mb);

        mb.Mu.EnterWriteLock();
        try
        {
            if (seq < mb.First.Seq || seq > mb.Last.Seq)
                return (false, null);

            if (mb.Msgs > 0)
                mb.Msgs--;

            mb.Dmap.Insert(seq);

            if (sm != null)
            {
                var size = FileStoreMsgSize(sm.Subject, sm.Hdr, sm.Msg);
                if (size <= mb.Bytes)
                    mb.Bytes -= size;
                else
                    mb.Bytes = 0;

                if (!mb.NoTrack && mb.Fss != null && !string.IsNullOrEmpty(sm.Subject))
                {
                    var subjBytes = Encoding.UTF8.GetBytes(sm.Subject);
                    var (ss, ok) = mb.Fss.Find(subjBytes);
                    if (ok && ss != null)
                    {
                        if (ss.Msgs > 0)
                            ss.Msgs--;
                        if (ss.Msgs == 0)
                            mb.Fss.Delete(subjBytes);
                        else if (seq == ss.First || seq == ss.Last)
                            ss.LastNeedsUpdate = true;
                    }
                }
            }

            mb.SelectNextFirst();
            if (mb.Msgs == 0)
            {
                mb.First = new MsgId { Seq = mb.Last.Seq + 1, Ts = 0 };
                mb.Last = new MsgId { Seq = mb.First.Seq - 1, Ts = 0 };
            }

            mb.NeedSync = mb.NeedSync || viaLimits || secure;
            return (true, null);
        }
        finally
        {
            mb.Mu.ExitWriteLock();
        }
    }

    // Lock should be held.
    private void RemoveMsgsInRange(ulong first, ulong last, bool viaLimits)
    {
        if (first == 0 || last == 0 || last < first)
            return;

        for (var seq = first; seq <= last; seq++)
        {
            _ = RemoveMsgInternal(seq, secure: false, viaLimits: viaLimits, needFsLock: false);
            if (seq == ulong.MaxValue)
                break;
        }

        RebuildFirst();
    }

    private bool IsClosed()
        => _closed;

    // Lock should be held.
    private void SelectNextFirst()
    {
        foreach (var mb in _blks)
        {
            mb.Mu.EnterWriteLock();
            try
            {
                mb.SelectNextFirst();
                if (mb.Msgs == 0)
                    continue;

                _state.FirstSeq = mb.First.Seq;
                var sm = _memStore.LoadMsg(_state.FirstSeq, null);
                _state.FirstTime = sm == null ? default : FromUnixNanosUtc(sm.Ts);
                return;
            }
            finally
            {
                mb.Mu.ExitWriteLock();
            }
        }

        _state.FirstSeq = _state.LastSeq + 1;
        _state.FirstTime = default;
    }

    // Lock should be held.
    private MessageBlock? FindMsgBlockForSeq(ulong seq)
    {
        foreach (var mb in _blks)
        {
            if (seq >= mb.First.Seq && seq <= mb.Last.Seq)
                return mb;
        }

        return null;
    }

    // Lock should be held.
    private void RefreshPsiFromMemStoreLocked()
    {
        _psim ??= new SubjectTree<Psi>();
        _psim.Reset();
        _tsl = 0;

        var totals = _memStore.SubjectsTotals(">");
        if (totals.Count == 0)
            return;

        var fblk = _blks.Count > 0 ? _blks[0].Index : 1U;
        var lblk = _lmb?.Index ?? fblk;

        foreach (var kvp in totals)
        {
            if (kvp.Value == 0)
                continue;

            var subject = kvp.Key;
            _psim.Insert(Encoding.UTF8.GetBytes(subject), new Psi
            {
                Total = kvp.Value,
                Fblk = fblk,
                Lblk = lblk,
            });
            _tsl += subject.Length;
        }
    }

    private static byte[] BuildMessageRecord(string subject, ReadOnlySpan<byte> hdr, ReadOnlySpan<byte> msg, ulong seq, long ts)
    {
        var subjectBytes = Encoding.UTF8.GetBytes(subject);
        var bodyLength = (8 + 8 + 4 + 4 + 4) + subjectBytes.Length + hdr.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..], hdr.Length);
        index += 4;
        BinaryPrimitives.WriteInt32LittleEndian(span[index..], msg.Length);
        index += 4;

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

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

    // Lock should be held.
    private void CheckAndFlushLastBlock()
    {
        var lmb = _lmb;
        if (lmb == null)
            return;

        lmb.Mu.EnterWriteLock();
        try
        {
            if (!lmb.NeedSync)
                return;

            if (lmb.Mfd == null)
            {
                lmb.Mfd = new FileStream(lmb.Mfn, FileMode.OpenOrCreate, FileAccess.ReadWrite, FileShare.Read);
            }

            lmb.Mfd.Flush(_fcfg.SyncAlways);
            lmb.NeedSync = false;
        }
        catch (Exception ex)
        {
            lmb.Werr = ex;
        }
        finally
        {
            lmb.Mu.ExitWriteLock();
        }
    }

    private LostStreamData? CheckMsgs()
    {
        _mu.EnterWriteLock();
        try
        {
            CheckAndFlushLastBlock();
            _state = _memStore.State();
            RefreshPsiFromMemStoreLocked();
            return _ld;
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    // Lock should be held.
    private (MessageBlock? Mb, Exception? Error) CheckLastBlock(ulong recordLength)
    {
        var lmb = _lmb;
        var rbytes = lmb?.RBytes ?? 0UL;
        if (lmb == null || (_fcfg.BlockSize > 0 && rbytes > 0 && rbytes + recordLength > _fcfg.BlockSize))
            return NewMsgBlockForWrite();

        return (lmb, null);
    }

    // Lock should be held.
    private (ulong RecordLength, Exception? Error) WriteMsgRecord(ulong seq, long ts, string subj, byte[]? hdr, byte[]? msg)
    {
        hdr ??= Array.Empty<byte>();
        msg ??= Array.Empty<byte>();

        var recordLength = FileStoreMsgSize(subj, hdr, msg);
        if (recordLength > RecordLengthBadThreshold)
            return (0, StoreErrors.ErrMsgTooLarge);

        var (_, blockErr) = CheckLastBlock(recordLength);
        if (blockErr != null)
            return (0, blockErr);

        _dirty++;
        var err = StoreRawMsgInternal(subj, hdr, msg, seq, ts, ttl: 0, discardNewCheck: false);
        return (recordLength, err);
    }

    // Lock should be held.
    private Exception? WriteTombstone(ulong seq, long ts)
    {
        var err = WriteTombstoneNoFlush(seq, ts);
        if (err != null)
            return err;

        var lmb = _lmb;
        if (lmb == null)
            return null;

        lmb.Mu.EnterWriteLock();
        try
        {
            lmb.Mfd?.Flush(_fcfg.SyncAlways);
            lmb.NeedSync = false;
            return null;
        }
        catch (Exception ex)
        {
            lmb.Werr = ex;
            return ex;
        }
        finally
        {
            lmb.Mu.ExitWriteLock();
        }
    }

    // Lock should be held.
    private Exception? WriteTombstoneNoFlush(ulong seq, long ts)
    {
        var (lmb, blockErr) = CheckLastBlock(EmptyRecordLength);
        if (blockErr != null)
            return blockErr;
        if (lmb == null)
            return StoreErrors.ErrStoreClosed;

        lmb.Mu.EnterWriteLock();
        try
        {
            lmb.Mfd ??= new FileStream(lmb.Mfn, FileMode.OpenOrCreate, FileAccess.ReadWrite, FileShare.Read);
            lmb.Mfd.Seek(0, SeekOrigin.End);
            var record = BuildMessageRecord(string.Empty, ReadOnlySpan<byte>.Empty, ReadOnlySpan<byte>.Empty, seq, ts);
            lmb.Mfd.Write(record, 0, record.Length);
            lmb.RBytes += (ulong)record.Length;
            lmb.Lwts = ts;
            lmb.NeedSync = !_fcfg.SyncAlways;
            lmb.NoCompact = true;
            lmb.Lchk = record.AsSpan(record.Length - FileStoreDefaults.RecordHashSize).ToArray();
            return null;
        }
        catch (Exception ex)
        {
            lmb.Werr = ex;
            return ex;
        }
        finally
        {
            lmb.Mu.ExitWriteLock();
        }
    }

    private void SyncBlocks()
    {
        if (IsClosed())
            return;

        List<MessageBlock> blks;
        bool firstMoved;
        _mu.EnterWriteLock();
        try
        {
            blks = [.. _blks];
            firstMoved = _firstMoved;
            _firstMoved = false;
        }
        finally
        {
            _mu.ExitWriteLock();
        }

        var markDirty = false;
        foreach (var mb in blks)
        {
            mb.Mu.EnterWriteLock();
            try
            {
                if (mb.Closed)
                    continue;

                if (firstMoved && mb.NoCompact)
                {
                    mb.NoCompact = false;
                    markDirty = true;
                }

                if (!mb.NeedSync)
                    continue;

                var opened = false;
                if (mb.Mfd == null)
                {
                    mb.Mfd = new FileStream(mb.Mfn, FileMode.OpenOrCreate, FileAccess.ReadWrite, FileShare.Read);
                    opened = true;
                }

                try
                {
                    mb.Mfd.Flush(true);
                    mb.NeedSync = false;
                }
                finally
                {
                    if (opened)
                    {
                        mb.Mfd.Dispose();
                        mb.Mfd = null;
                    }
                }
            }
            finally
            {
                mb.Mu.ExitWriteLock();
            }
        }

        if (IsClosed())
            return;

        _mu.EnterWriteLock();
        try
        {
            SetSyncTimerLocked();
            if (markDirty)
                _dirty++;

            if (!_fcfg.SyncAlways)
            {
                var fn = Path.Combine(_fcfg.StoreDir, FileStoreDefaults.MsgDir, FileStoreDefaults.StreamStateFile);
                if (File.Exists(fn))
                {
                    using var fd = new FileStream(fn, FileMode.Open, FileAccess.ReadWrite, FileShare.ReadWrite);
                    fd.Flush(true);
                }
            }
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    // Lock should be held.
    private MessageBlock? SelectMsgBlock(ulong seq)
    {
        var (_, mb) = SelectMsgBlockWithIndex(seq);
        return mb;
    }

    // Lock should be held.
    private (int Index, MessageBlock? Block) SelectMsgBlockWithIndex(ulong seq)
    {
        if (seq < _state.FirstSeq || seq > _state.LastSeq || _state.Msgs == 0)
            return (-1, null);

        const int linearThreshold = 32;
        var nb = _blks.Count - 1;
        if (nb < 0)
            return (-1, null);

        if (nb < linearThreshold)
        {
            for (var i = 0; i < _blks.Count; i++)
            {
                if (seq <= _blks[i].Last.Seq)
                    return (i, _blks[i]);
            }

            return (-1, null);
        }

        var low = 0;
        var high = nb;
        while (low <= high)
        {
            var mid = (low + high) / 2;
            var mb = _blks[mid];
            var first = mb.First.Seq;
            var last = mb.Last.Seq;
            if (seq > last)
            {
                low = mid + 1;
            }
            else if (seq < first)
            {
                if (mid == 0 || seq > _blks[mid - 1].Last.Seq)
                    return (mid, mb);
                high = mid - 1;
            }
            else
            {
                return (mid, mb);
            }
        }

        return (-1, null);
    }

    private MessageBlock? SelectMsgBlockForStart(DateTime minTime)
    {
        var target = TimestampNormalized(minTime);
        _mu.EnterReadLock();
        try
        {
            var low = 0;
            var high = _blks.Count - 1;
            var candidate = _blks.Count;

            while (low <= high)
            {
                var mid = (low + high) / 2;
                var mb = _blks[mid];
                long lastTs;
                mb.Mu.EnterReadLock();
                try
                {
                    lastTs = mb.Last.Ts;
                }
                finally
                {
                    mb.Mu.ExitReadLock();
                }

                if (lastTs >= target)
                {
                    candidate = mid;
                    high = mid - 1;
                }
                else
                {
                    low = mid + 1;
                }
            }

            return candidate < _blks.Count ? _blks[candidate] : null;
        }
        finally
        {
            _mu.ExitReadLock();
        }
    }

    // Lock should be held.
    private void SetSyncTimerLocked()
        => SetSyncTimer();

    // Lock should be held.
    private void SetSyncTimer()
    {
        if (_fcfg.SyncInterval <= TimeSpan.Zero || IsClosed())
            return;

        if (_syncTmr != null)
        {
            _syncTmr.Change(_fcfg.SyncInterval, Timeout.InfiniteTimeSpan);
            return;
        }

        var half = TimeSpan.FromTicks(Math.Max(1, _fcfg.SyncInterval.Ticks / 2));
        var jitterTicks = Random.Shared.NextInt64(Math.Max(1, half.Ticks));
        var start = half + TimeSpan.FromTicks(jitterTicks);
        _syncTmr = new Timer(_ => SyncBlocks(), null, start, Timeout.InfiniteTimeSpan);
    }

    // Lock should be held.
    private void CancelSyncTimer()
    {
        _syncTmr?.Dispose();
        _syncTmr = null;
    }

    private async Task FlushStreamStateLoop(ChannelReader<byte> qch, ChannelWriter<byte> done)
    {
        try
        {
            var writeThreshold = TimeSpan.FromMinutes(2);
            var writeJitter = TimeSpan.FromSeconds(Random.Shared.Next(0, 30));
            using var ticker = new PeriodicTimer(writeThreshold + writeJitter);

            while (true)
            {
                var tickTask = ticker.WaitForNextTickAsync().AsTask();
                var quitTask = qch.Completion;
                var completed = await Task.WhenAny(tickTask, quitTask).ConfigureAwait(false);
                if (ReferenceEquals(completed, quitTask))
                    break;

                if (!await tickTask.ConfigureAwait(false))
                    break;

                var err = WriteFullState();
                if (err is UnauthorizedAccessException)
                {
                    Warn("File system permission denied when flushing stream state: {0}", err.Message);
                    break;
                }
            }
        }
        finally
        {
            done.TryWrite(0);
            done.TryComplete();
        }
    }

    // -----------------------------------------------------------------------
    // Lifecycle/query helper methods (Batch 14 Group 3)
    // -----------------------------------------------------------------------

    private int SizeForSeq(ulong seq)
    {
        if (seq == 0)
            return 0;

        var (sm, err) = MsgForSeq(seq, null);
        if (err != null || sm == null)
            return 0;

        var size = FileStoreMsgSize(sm.Subject, sm.Hdr, sm.Msg);
        return size >= int.MaxValue ? int.MaxValue : (int)size;
    }

    private (StoreMsg? Sm, Exception? Error) MsgForSeq(ulong seq, StoreMsg? sm)
        => MsgForSeqLocked(seq, sm, needFsLock: true);

    private (StoreMsg? Sm, Exception? Error) MsgForSeqLocked(ulong seq, StoreMsg? sm, bool needFsLock)
    {
        if (IsClosed())
            return (null, StoreErrors.ErrStoreClosed);

        if (needFsLock)
            _mu.EnterReadLock();

        try
        {
            var state = _memStore.State();

            if (seq == 0)
                seq = state.FirstSeq;

            if (state.Msgs == 0 || seq == 0)
                return (null, StoreErrors.ErrStoreEOF);

            try
            {
                var loaded = _memStore.LoadMsg(seq, sm);
                return loaded == null ? (null, StoreErrors.ErrStoreMsgNotFound) : (loaded, null);
            }
            catch (Exception ex) when (ReferenceEquals(ex, StoreErrors.ErrStoreEOF))
            {
                if (seq <= state.LastSeq)
                    return (null, StoreErrors.ErrStoreMsgNotFound);
                return (null, ex);
            }
            catch (Exception ex)
            {
                return (null, ex);
            }
        }
        finally
        {
            if (needFsLock)
                _mu.ExitReadLock();
        }
    }

    private (StoreMsg? Sm, Exception? Error) LoadLast(string subj, StoreMsg? sm)
    {
        if (IsClosed())
            return (null, StoreErrors.ErrStoreClosed);

        _mu.EnterReadLock();
        try
        {
            return LoadLastLocked(subj, sm);
        }
        finally
        {
            _mu.ExitReadLock();
        }
    }

    // Lock should be held.
    private (StoreMsg? Sm, Exception? Error) LoadLastLocked(string subj, StoreMsg? sm)
    {
        if (IsClosed())
            return (null, StoreErrors.ErrStoreClosed);

        var state = _memStore.State();
        if (state.Msgs == 0)
            return (null, StoreErrors.ErrStoreMsgNotFound);

        if (string.IsNullOrEmpty(subj) || subj == ">")
            return MsgForSeqLocked(state.LastSeq, sm, needFsLock: false);

        var wc = SubscriptionIndex.SubjectHasWildcard(subj);
        if (!wc)
        {
            try
            {
                var loaded = _memStore.LoadLastMsg(subj, sm);
                return loaded == null ? (null, StoreErrors.ErrStoreMsgNotFound) : (loaded, null);
            }
            catch (Exception ex)
            {
                return (null, ex);
            }
        }

        StoreMsg? last = null;
        var lastSeq = 0UL;
        foreach (var candidate in _memStore.SubjectsTotals(">").Keys)
        {
            if (!SubscriptionIndex.SubjectIsSubsetMatch(subj, candidate))
                continue;

            try
            {
                var cmsg = _memStore.LoadLastMsg(candidate, null);
                if (cmsg == null || cmsg.Seq < lastSeq)
                    continue;

                if (last == null)
                    last = sm ?? new StoreMsg();

                last.CopyFrom(cmsg);
                lastSeq = cmsg.Seq;
            }
            catch
            {
                // Ignore candidate misses and continue scanning.
            }
        }

        return last == null ? (null, StoreErrors.ErrStoreMsgNotFound) : (last, null);
    }

    private int NumSubjects()
    {
        _mu.EnterReadLock();
        try
        {
            if (_psim != null)
                return _psim.Size();
            return _memStore.SubjectsTotals(">").Count;
        }
        finally
        {
            _mu.ExitReadLock();
        }
    }

    private int NumConsumers()
    {
        _cmu.EnterReadLock();
        try
        {
            return _cfs.Count;
        }
        finally
        {
            _cmu.ExitReadLock();
        }
    }

    private ulong CacheLoads()
    {
        ulong total = 0;
        _mu.EnterReadLock();
        try
        {
            foreach (var mb in _blks)
            {
                mb.Mu.EnterReadLock();
                try
                {
                    total += mb.Cloads;
                }
                finally
                {
                    mb.Mu.ExitReadLock();
                }
            }
        }
        finally
        {
            _mu.ExitReadLock();
        }

        return total;
    }

    private ulong CacheSize()
    {
        ulong total = 0;
        _mu.EnterReadLock();
        try
        {
            foreach (var mb in _blks)
            {
                mb.Mu.EnterReadLock();
                try
                {
                    if (mb.CacheData?.Buf is { Length: > 0 } buf)
                        total += (ulong)buf.Length;
                }
                finally
                {
                    mb.Mu.ExitReadLock();
                }
            }
        }
        finally
        {
            _mu.ExitReadLock();
        }

        return total;
    }

    private int DmapEntries()
    {
        var total = 0;
        _mu.EnterReadLock();
        try
        {
            foreach (var mb in _blks)
                total += mb.Dmap.Size;
        }
        finally
        {
            _mu.ExitReadLock();
        }

        return total;
    }

    private (ulong Purged, Exception? Error) PurgeInternal(ulong fseq)
    {
        if (IsClosed())
            return (0, StoreErrors.ErrStoreClosed);

        _mu.EnterWriteLock();
        try
        {
            var (purged, err) = fseq > 0 ? _memStore.Compact(fseq) : _memStore.Purge();
            if (err != null)
                return (purged, err);

            _state = _memStore.State();

            foreach (var mb in _blks)
                mb.CloseFDs();

            _blks.Clear();
            _bim.Clear();
            _lmb = null;
            _psim ??= new SubjectTree<Psi>();
            _psim.Reset();
            _tsl = 0;
            _sdm.Empty();

            var (lmb, blockErr) = NewMsgBlockForWrite();
            if (blockErr != null)
                return (purged, blockErr);

            _lmb = lmb;
            _dirty++;

            if (_state.LastSeq > 0)
                _ = WriteTombstone(_state.LastSeq, TimestampNormalized(_state.LastTime));

            return (purged, null);
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    private (ulong Purged, Exception? Error) CompactInternal(ulong seq)
    {
        if (seq == 0)
            return PurgeInternal(0);

        if (IsClosed())
            return (0, StoreErrors.ErrStoreClosed);

        _mu.EnterWriteLock();
        try
        {
            var (purged, err) = _memStore.Compact(seq);
            if (err != null)
                return (purged, err);

            _state = _memStore.State();

            if (_state.Msgs == 0)
            {
                foreach (var mb in _blks.ToArray())
                    ForceRemoveMsgBlock(mb);

                _blks.Clear();
                _bim.Clear();
                _lmb = null;
                var (newMb, blockErr) = NewMsgBlockForWrite();
                if (blockErr != null)
                    return (purged, blockErr);
                _lmb = newMb;
            }
            else
            {
                foreach (var mb in _blks.ToArray())
                {
                    if (mb.Last.Seq < _state.FirstSeq)
                        ForceRemoveMsgBlock(mb);
                }

                _lmb = _blks.Count > 0 ? _blks[^1] : null;
                if (_lmb == null)
                {
                    var (newMb, blockErr) = NewMsgBlockForWrite();
                    if (blockErr != null)
                        return (purged, blockErr);
                    _lmb = newMb;
                }
            }

            RefreshPsiFromMemStoreLocked();
            _dirty++;
            return (purged, null);
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    private Exception? Reset()
    {
        if (IsClosed())
            return StoreErrors.ErrStoreClosed;

        _mu.EnterWriteLock();
        try
        {
            var err = _memStore.Reset();
            if (err != null)
                return err;

            foreach (var mb in _blks.ToArray())
                ForceRemoveMsgBlock(mb);

            _blks.Clear();
            _bim.Clear();
            _lmb = null;
            _state = _memStore.State();
            _psim ??= new SubjectTree<Psi>();
            _psim.Reset();
            _tsl = 0;
            _sdm.Empty();
            _dirty++;
            return null;
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    private ulong LastSeq()
    {
        _mu.EnterReadLock();
        try
        {
            return _memStore.State().LastSeq;
        }
        finally
        {
            _mu.ExitReadLock();
        }
    }

    private int NumMsgBlocks()
    {
        _mu.EnterReadLock();
        try
        {
            if (_blks.Count > 0)
                return _blks.Count;

            return _memStore.State().Msgs > 0 ? 1 : 0;
        }
        finally
        {
            _mu.ExitReadLock();
        }
    }

    // Both locks should be held by caller.
    private void RemoveMsgBlockFromList(MessageBlock mb)
    {
        var idx = _blks.FindIndex(existing => ReferenceEquals(existing, mb));
        if (idx < 0)
            return;

        _blks.RemoveAt(idx);
        _bim.Remove(mb.Index);
        _lmb = _blks.Count > 0 ? _blks[^1] : null;
        _dirty++;
    }

    // Both locks should be held by caller.
    private void RemoveMsgBlock(MessageBlock mb)
    {
        if (ReferenceEquals(mb, _lmb))
        {
            var lseq = mb.Last.Seq;
            var lts = mb.Last.Ts;
            var (newMb, _) = NewMsgBlockForWrite();
            _lmb = newMb;
            if (lseq > 0)
                _ = WriteTombstone(lseq, lts);
        }
        else if (mb.Last.Seq == _state.LastSeq && mb.Last.Seq > 0)
        {
            _ = WriteTombstone(mb.Last.Seq, mb.Last.Ts);
        }

        ForceRemoveMsgBlock(mb);
    }

    // Both locks should be held by caller.
    private void ForceRemoveMsgBlock(MessageBlock mb)
    {
        mb.CloseFDs();
        RemoveMsgBlockFromList(mb);

        TryDeleteFile(mb.Mfn);
        TryDeleteFile(mb.Kfn);

        var mdir = Path.Combine(_fcfg.StoreDir, FileStoreDefaults.MsgDir);
        TryDeleteFile(Path.Combine(mdir, string.Format(FileStoreDefaults.IndexScan, mb.Index)));
    }

    // -----------------------------------------------------------------------
    // Read/query helper methods (Batch 13)
    // -----------------------------------------------------------------------

    // Lock should be held by caller.
    private (int Next, Exception? Error) CheckSkipFirstBlock(string filter, bool wc, int bi)
    {
        if (string.IsNullOrEmpty(filter) || filter == ">")
            return (bi + 1, null);

        var start = uint.MaxValue;
        uint stop = 0;

        if (_psim is { } psim)
        {
            if (wc)
            {
                psim.Match(Encoding.UTF8.GetBytes(filter), (_, psi) =>
                {
                    if (psi.Fblk < start)
                        start = psi.Fblk;
                    if (psi.Lblk > stop)
                        stop = psi.Lblk;
                    return true;
                });
            }
            else
            {
                var (psi, ok) = psim.Find(Encoding.UTF8.GetBytes(filter));
                if (ok && psi != null)
                {
                    start = psi.Fblk;
                    stop = psi.Lblk;
                }
            }
        }

        if (start == uint.MaxValue)
            return (-1, StoreErrors.ErrStoreEOF);

        return SelectSkipFirstBlock(bi, start, stop);
    }

    // Lock should be held by caller.
    private (int Next, Exception? Error) CheckSkipFirstBlockMulti(SimpleSublist? sl, int bi)
    {
        if (_psim == null || sl == null)
            return (-1, StoreErrors.ErrStoreEOF);

        var start = uint.MaxValue;
        uint stop = 0;

        _psim.IterFast((subj, psi) =>
        {
            var matched = false;
            sl.Match(Encoding.UTF8.GetString(subj), _ => matched = true);
            if (matched)
            {
                if (psi.Fblk < start)
                    start = psi.Fblk;
                if (psi.Lblk > stop)
                    stop = psi.Lblk;
            }
            return true;
        });

        if (start == uint.MaxValue)
            return (-1, StoreErrors.ErrStoreEOF);

        return SelectSkipFirstBlock(bi, start, stop);
    }

    // Lock should be held by caller.
    private (int Next, Exception? Error) SelectSkipFirstBlock(int bi, uint start, uint stop)
    {
        if (bi < 0 || bi >= _blks.Count)
            return (-1, StoreErrors.ErrStoreEOF);

        var mbi = _blks[bi].Index;
        if (stop <= mbi)
            return (-1, StoreErrors.ErrStoreEOF);

        if (start > mbi && _bim.TryGetValue(start, out var mb) && mb != null)
        {
            var ni = -1;
            for (var i = 0; i < _blks.Count; i++)
            {
                if (mb.Last.Seq <= _blks[i].Last.Seq)
                {
                    ni = i;
                    break;
                }
            }

            return (ni, null);
        }

        return (bi + 1, null);
    }

    // Lock should be held by caller.
    private void NumFilteredPending(string filter, SimpleState ss)
        => NumFilteredPendingWithLast(filter, includeLast: true, ss);

    // Lock should be held by caller.
    private void NumFilteredPendingNoLast(string filter, SimpleState ss)
        => NumFilteredPendingWithLast(filter, includeLast: false, ss);

    // Lock should be held by caller.
    private void NumFilteredPendingWithLast(string filter, bool includeLast, SimpleState ss)
    {
        ArgumentNullException.ThrowIfNull(ss);

        var isAll = string.IsNullOrEmpty(filter) || filter == ">";
        if (isAll)
        {
            ss.First = _state.FirstSeq;
            ss.Last = _state.LastSeq;
            ss.Msgs = _state.Msgs;
            return;
        }

        ss.First = 0;
        ss.Last = 0;
        ss.Msgs = 0;

        if (_psim == null)
            return;

        var wc = SubscriptionIndex.SubjectHasWildcard(filter);
        var filterBytes = Encoding.UTF8.GetBytes(filter);
        var start = uint.MaxValue;
        uint stop = 0;

        if (wc)
        {
            _psim.Match(filterBytes, (_, psi) =>
            {
                ss.Msgs += psi.Total;
                if (psi.Fblk < start)
                    start = psi.Fblk;
                if (psi.Lblk > stop)
                    stop = psi.Lblk;
                return true;
            });
        }
        else
        {
            var (psi, ok) = _psim.Find(filterBytes);
            if (ok && psi != null)
            {
                ss.Msgs += psi.Total;
                start = psi.Fblk;
                stop = psi.Lblk;
            }
        }

        if (stop == 0 || start == uint.MaxValue)
            return;

        if (_bim.TryGetValue(start, out var mb) && mb != null)
        {
            var (_, first, _) = FilteredPendingInBlock(mb, filter, wc);
            ss.First = first;
        }

        if (ss.First == 0)
        {
            uint? correctedFirstBlock = null;
            for (ulong i = (ulong)start + 1; i <= stop; i++)
            {
                if (!_bim.TryGetValue((uint)i, out var candidate) || candidate == null)
                    continue;

                var (_, first, _) = FilteredPendingInBlock(candidate, filter, wc);
                if (first > 0)
                {
                    ss.First = first;
                    correctedFirstBlock = (uint)i;
                    break;
                }
            }

            if (correctedFirstBlock is { } corrected)
                SchedulePsiFirstBlockCorrection(filter, wc, corrected);
        }

        if (includeLast && _bim.TryGetValue(stop, out var lastMb) && lastMb != null)
        {
            var (_, _, last) = FilteredPendingInBlock(lastMb, filter, wc);
            ss.Last = last;
        }
    }

    // Caller should hold a read lock on fs + message block lock if concurrent mutations are possible.
    private static (ulong Total, ulong First, ulong Last) FilteredPendingInBlock(MessageBlock mb, string filter, bool wc)
    {
        if (mb.Fss == null)
            return (0, 0, 0);

        var filterBytes = Encoding.UTF8.GetBytes(filter);
        ulong total = 0;
        ulong first = 0;
        ulong last = 0;

        if (wc)
        {
            mb.Fss.Match(filterBytes, (_, ss) =>
            {
                total += ss.Msgs;
                if (first == 0 || (ss.First > 0 && ss.First < first))
                    first = ss.First;
                if (ss.Last > last)
                    last = ss.Last;
                return true;
            });
        }
        else
        {
            var (ss, ok) = mb.Fss.Find(filterBytes);
            if (ok && ss != null)
            {
                total = ss.Msgs;
                first = ss.First;
                last = ss.Last;
            }
        }

        return (total, first, last);
    }

    private void SchedulePsiFirstBlockCorrection(string filter, bool wc, uint firstBlock)
    {
        if (_psim == null)
            return;

        var filterBytes = Encoding.UTF8.GetBytes(filter);

        _ = Task.Run(() =>
        {
            _mu.EnterWriteLock();
            try
            {
                if (_psim == null)
                    return;

                if (!wc)
                {
                    var (info, ok) = _psim.Find(filterBytes);
                    if (ok && info != null && firstBlock > info.Fblk)
                        info.Fblk = firstBlock;
                    return;
                }

                _psim.Match(filterBytes, (_, psi) =>
                {
                    if (firstBlock > psi.Fblk)
                        psi.Fblk = firstBlock;
                    return true;
                });
            }
            finally
            {
                _mu.ExitWriteLock();
            }
        });
    }

    // Lock should be held by caller.
    private (ulong[] Seqs, Exception? Error) AllLastSeqsLocked()
    {
        if (_state.Msgs == 0 || NoTrackSubjects() || _psim == null)
            return (Array.Empty<ulong>(), null);

        var numSubjects = _psim.Size();
        if (numSubjects == 0)
            return (Array.Empty<ulong>(), null);

        var seqs = new List<ulong>(numSubjects);
        var seen = new HashSet<string>(numSubjects, StringComparer.Ordinal);

        for (var i = _blks.Count - 1; i >= 0; i--)
        {
            if (seen.Count == numSubjects)
                break;

            var mb = _blks[i];
            if (mb.Fss == null)
                continue;

            mb.Fss.IterFast((subjectBytes, ss) =>
            {
                var subject = Encoding.UTF8.GetString(subjectBytes);
                if (!seen.Add(subject))
                    return true;

                if (ss.LastNeedsUpdate)
                    RecalculateLastForSubject(subject, ss);

                seqs.Add(ss.Last);
                return true;
            });
        }

        seqs.Sort();
        return (seqs.ToArray(), null);
    }

    // Lock should be held by caller.
    private bool FilterIsAll(string[] filters)
    {
        ArgumentNullException.ThrowIfNull(filters);

        var streamSubjects = _cfg.Config.Subjects ?? [];
        if (filters.Length != streamSubjects.Length)
            return false;

        var sortedFilters = filters.ToArray();
        var sortedSubjects = streamSubjects.ToArray();
        Array.Sort(sortedFilters, StringComparer.Ordinal);
        Array.Sort(sortedSubjects, StringComparer.Ordinal);

        for (var i = 0; i < sortedFilters.Length; i++)
        {
            if (!SubscriptionIndex.SubjectIsSubsetMatch(sortedSubjects[i], sortedFilters[i]))
                return false;
        }

        return true;
    }

    // Lock should be held by caller.
    private void RecalculateLastForSubject(string subject, SimpleState ss)
    {
        var subjectBytes = Encoding.UTF8.GetBytes(subject);
        for (var i = _blks.Count - 1; i >= 0; i--)
        {
            var fss = _blks[i].Fss;
            if (fss == null)
                continue;

            var (candidate, ok) = fss.Find(subjectBytes);
            if (!ok || candidate == null || candidate.Last == 0)
                continue;

            ss.Last = candidate.Last;
            ss.LastNeedsUpdate = false;
            return;
        }

        ss.Last = 0;
        ss.LastNeedsUpdate = false;
    }

    // -----------------------------------------------------------------------
    // Lifecycle helpers (Batch 14 Group 4)
    // -----------------------------------------------------------------------

    // Lock should be held.
    private void PurgeMsgBlock(MessageBlock mb)
    {
        ArgumentNullException.ThrowIfNull(mb);

        mb.Mu.EnterWriteLock();
        ulong msgs;
        ulong bytes;
        try
        {
            msgs = Math.Min(mb.Msgs, _state.Msgs);
            bytes = Math.Min(mb.Bytes, _state.Bytes);

            if (_scheduling != null && mb.Schedules > 0)
            {
                for (var seq = mb.First.Seq; seq <= mb.Last.Seq; seq++)
                {
                    _scheduling.Remove(seq);
                    if (seq == ulong.MaxValue)
                        break;
                }
            }

            _state.Msgs -= msgs;
            _state.Bytes -= bytes;
            RemoveMsgBlock(mb);
            mb.TryForceExpireCacheLocked();
        }
        finally
        {
            mb.Mu.ExitWriteLock();
        }

        SelectNextFirst();

        if ((msgs > 0 || bytes > 0) && _scb != null)
            _scb(-(long)msgs, -(long)bytes, 0, string.Empty);
    }

    // Lock should be held.
    private void ResetGlobalPerSubjectInfo()
    {
        _psim ??= new SubjectTree<Psi>();
        _psim.Reset();
        _tsl = 0;

        if (NoTrackSubjects())
            return;

        foreach (var mb in _blks)
            PopulateGlobalPerSubjectInfo(mb);
    }

    // Lock should be held.
    private void PopulateGlobalPerSubjectInfo(MessageBlock mb)
    {
        ArgumentNullException.ThrowIfNull(mb);

        mb.Mu.EnterReadLock();
        try
        {
            var fss = mb.Fss;
            if (fss == null)
                return;

            _psim ??= new SubjectTree<Psi>();

            fss.IterFast((bsubj, ss) =>
            {
                if (bsubj.Length == 0)
                    return true;

                var (psi, ok) = _psim.Find(bsubj);
                if (ok && psi != null)
                {
                    psi.Total += ss.Msgs;
                    if (mb.Index > psi.Lblk)
                        psi.Lblk = mb.Index;
                }
                else
                {
                    _psim.Insert(bsubj, new Psi
                    {
                        Total = ss.Msgs,
                        Fblk = mb.Index,
                        Lblk = mb.Index,
                    });
                    _tsl += bsubj.Length;
                }

                return true;
            });
        }
        finally
        {
            mb.Mu.ExitReadLock();
        }
    }

    private void CloseAllMsgBlocks(bool sync)
    {
        foreach (var mb in _blks)
            CloseMsgBlock(mb, sync);
    }

    private static void CloseMsgBlock(MessageBlock mb, bool sync)
    {
        if (mb == null)
            return;

        mb.Mu.EnterWriteLock();
        try
        {
            if (mb.Closed)
                return;

            mb.Ctmr?.Dispose();
            mb.Ctmr = null;
            mb.Fss = null;
            mb.TryForceExpireCacheLocked();

            if (mb.Qch != null)
            {
                mb.Qch.Writer.TryComplete();
                mb.Qch = null;
            }

            if (mb.Mfd != null)
            {
                if (sync || mb.SyncAlways || mb.NeedSync)
                    mb.Mfd.Flush(true);
                mb.Mfd.Dispose();
                mb.Mfd = null;
            }

            mb.NeedSync = false;
            mb.Closed = true;
        }
        catch (Exception ex)
        {
            mb.Werr = ex;
        }
        finally
        {
            mb.Mu.ExitWriteLock();
        }
    }

    private Exception? WriteFullState()
        => WriteFullStateInternal(force: false);

    private Exception? ForceWriteFullState()
        => WriteFullStateInternal(force: true);

    private Exception? WriteFullStateInternal(bool force)
    {
        if (IsClosed())
            return null;

        if (Interlocked.Increment(ref _wfsrun) > 1 && !force)
        {
            Interlocked.Decrement(ref _wfsrun);
            return null;
        }

        try
        {
            lock (_wfsmu)
            {
                try
                {
                    StreamState state;
                    StreamState tracked = new();
                    List<(byte[] Subject, Psi Info)> subjects = [];
                    List<(uint Index, ulong Bytes, ulong FirstSeq, long FirstTs, ulong LastSeq, long LastTs, ulong NumDeleted, ulong Ttls, ulong Schedules, byte[] Dmap, byte[] Lchk, bool IsLast)> blocks = [];
                    int priorDirty;

                    _mu.EnterReadLock();
                    try
                    {
                        if (_dirty == 0 && !force)
                            return null;

                        state = _memStore.State();
                        priorDirty = _dirty;

                        if (_prf != null && _aek == null)
                        {
                            _mu.ExitReadLock();
                            _mu.EnterWriteLock();
                            try { SetupAEK(); }
                            finally { _mu.ExitWriteLock(); }
                            _mu.EnterReadLock();
                        }

                        if (_psim != null && _psim.Size() > 0)
                        {
                            _psim.Match(Encoding.UTF8.GetBytes(">"), (subject, psi) =>
                            {
                                if (psi != null)
                                {
                                    subjects.Add((subject.ToArray(), new Psi
                                    {
                                        Total = psi.Total,
                                        Fblk = psi.Fblk,
                                        Lblk = psi.Lblk,
                                    }));
                                }

                                return true;
                            });
                        }

                        foreach (var mb in _blks)
                        {
                            mb.Mu.EnterReadLock();
                            try
                            {
                                var numDeleted = (ulong)mb.Dmap.Size;
                                var dmap = numDeleted > 0 ? mb.Dmap.Encode(null) : Array.Empty<byte>();
                                var lchk = mb.Lchk.Length >= FullStateChecksumLength
                                    ? mb.Lchk.AsSpan(0, FullStateChecksumLength).ToArray()
                                    : new byte[FullStateChecksumLength];

                                blocks.Add((
                                    mb.Index,
                                    mb.Bytes,
                                    mb.First.Seq,
                                    mb.First.Ts,
                                    mb.Last.Seq,
                                    mb.Last.Ts,
                                    numDeleted,
                                    mb.Ttls,
                                    mb.Schedules,
                                    dmap,
                                    lchk,
                                    ReferenceEquals(mb, _lmb)));

                                UpdateTrackingState(tracked, mb);
                            }
                            finally
                            {
                                mb.Mu.ExitReadLock();
                            }
                        }
                    }
                    finally
                    {
                        if (_mu.IsReadLockHeld)
                            _mu.ExitReadLock();
                    }

                    if (!TrackingStatesEqual(state, tracked))
                    {
                        Warn("Stream state encountered internal inconsistency on write");
                        RebuildState(null);
                        state = _memStore.State();
                    }

                    var baseTime = TimestampNormalized(state.FirstTime);
                    var buf = new List<byte>(4096)
                    {
                        FullStateMagic,
                        FullStateVersion,
                    };

                    AppendUVarInt(buf, state.Msgs);
                    AppendUVarInt(buf, state.Bytes);
                    AppendUVarInt(buf, state.FirstSeq);
                    AppendVarInt(buf, baseTime);
                    AppendUVarInt(buf, state.LastSeq);
                    AppendVarInt(buf, TimestampNormalized(state.LastTime));

                    AppendUVarInt(buf, (ulong)subjects.Count);
                    foreach (var (subject, info) in subjects)
                    {
                        AppendUVarInt(buf, (ulong)subject.Length);
                        buf.AddRange(subject);
                        AppendUVarInt(buf, info.Total);
                        AppendUVarInt(buf, info.Fblk);
                        if (info.Total > 1)
                            AppendUVarInt(buf, info.Lblk);
                    }

                    AppendUVarInt(buf, (ulong)blocks.Count);

                    uint lastBlockIndex = 0;
                    var lastChecksum = new byte[FullStateChecksumLength];

                    foreach (var block in blocks)
                    {
                        AppendUVarInt(buf, block.Index);
                        AppendUVarInt(buf, block.Bytes);
                        AppendUVarInt(buf, block.FirstSeq);
                        AppendVarInt(buf, block.FirstTs - baseTime);
                        AppendUVarInt(buf, block.LastSeq);
                        AppendVarInt(buf, block.LastTs - baseTime);
                        AppendUVarInt(buf, block.NumDeleted);
                        AppendUVarInt(buf, block.Ttls);
                        AppendUVarInt(buf, block.Schedules);
                        if (block.Dmap.Length > 0)
                            buf.AddRange(block.Dmap);

                        if (block.IsLast)
                        {
                            lastBlockIndex = block.Index;
                            Buffer.BlockCopy(block.Lchk, 0, lastChecksum, 0, Math.Min(lastChecksum.Length, block.Lchk.Length));
                        }
                    }

                    AppendUVarInt(buf, lastBlockIndex);
                    buf.AddRange(lastChecksum);

                    var payload = buf.ToArray();
                    if (_aek != null)
                    {
                        var nonce = new byte[_aek.NonceSize];
                        RandomNumberGenerator.Fill(nonce);
                        var encrypted = _aek.Seal(nonce, payload);
                        var sealedPayload = new byte[nonce.Length + encrypted.Length];
                        Buffer.BlockCopy(nonce, 0, sealedPayload, 0, nonce.Length);
                        Buffer.BlockCopy(encrypted, 0, sealedPayload, nonce.Length, encrypted.Length);
                        payload = sealedPayload;
                    }

                    var key = SHA256.HashData(Encoding.UTF8.GetBytes(_cfg.Config.Name));
                    using var hmac = new HMACSHA256(key);
                    var digest = hmac.ComputeHash(payload);

                    var output = new byte[payload.Length + FullStateChecksumLength];
                    Buffer.BlockCopy(payload, 0, output, 0, payload.Length);
                    Buffer.BlockCopy(digest, 0, output, payload.Length, FullStateChecksumLength);

                    var fn = Path.Combine(_fcfg.StoreDir, FileStoreDefaults.MsgDir, FileStoreDefaults.StreamStateFile);
                    WriteFileWithOptionalSync(fn, output);

                    _mu.EnterWriteLock();
                    try
                    {
                        _dirty = Math.Max(0, _dirty - priorDirty);
                    }
                    finally
                    {
                        _mu.ExitWriteLock();
                    }

                    var ttlErr = WriteTTLState();
                    if (ttlErr != null)
                        return ttlErr;

                    var schedErr = WriteMsgSchedulingState();
                    if (schedErr != null)
                        return schedErr;

                    return null;
                }
                catch (Exception ex)
                {
                    return ex;
                }
            }
        }
        finally
        {
            Interlocked.Decrement(ref _wfsrun);
        }
    }

    private Exception? WriteTTLState()
    {
        _mu.EnterReadLock();
        try
        {
            if (_ttls == null)
                return null;

            var fn = Path.Combine(_fcfg.StoreDir, FileStoreDefaults.MsgDir, FileStoreDefaults.TtlStreamStateFile);
            var buf = _ttls.Encode(_state.LastSeq + 1);
            WriteFileWithOptionalSync(fn, buf);
            return null;
        }
        catch (Exception ex)
        {
            return ex;
        }
        finally
        {
            _mu.ExitReadLock();
        }
    }

    private Exception? WriteMsgSchedulingState()
    {
        _mu.EnterReadLock();
        try
        {
            if (_scheduling == null)
                return null;

            var fn = Path.Combine(_fcfg.StoreDir, FileStoreDefaults.MsgDir, FileStoreDefaults.MsgSchedulingStreamStateFile);
            var buf = _scheduling.Encode(_state.LastSeq + 1);
            WriteFileWithOptionalSync(fn, buf);
            return null;
        }
        catch (Exception ex)
        {
            return ex;
        }
        finally
        {
            _mu.ExitReadLock();
        }
    }

    private Exception? StopInternal(bool delete, bool writeState)
    {
        if (IsClosed())
            return StoreErrors.ErrStoreClosed;

        _mu.EnterWriteLock();
        try
        {
            if (_closing)
                return StoreErrors.ErrStoreClosed;

            _closing = true;

            if (_qch != null)
            {
                _qch.Writer.TryComplete();
                _qch = null;
            }

            if (writeState)
                CheckAndFlushLastBlock();

            CloseAllMsgBlocks(sync: false);
            CancelSyncTimer();
            CancelAgeChk();

            if (writeState)
            {
                _mu.ExitWriteLock();
                _ = ForceWriteFullState();
                _mu.EnterWriteLock();
            }

            _closed = true;
            _lmb = null;

            var cb = _scb;
            var bytes = (long)_state.Bytes;
            _mu.ExitWriteLock();

            _cmu.EnterWriteLock();
            var cfs = _cfs.ToArray();
            _cfs.Clear();
            _cmu.ExitWriteLock();

            foreach (var consumer in cfs)
            {
                if (delete)
                    consumer.StreamDelete();
                else
                    consumer.Stop();
            }

            _memStore.Stop();

            if (bytes > 0 && cb != null)
                cb(0, -bytes, 0, string.Empty);

            return null;
        }
        finally
        {
            if (_mu.IsWriteLockHeld)
                _mu.ExitWriteLock();
        }
    }

    private (MemoryStream? Snapshot, Exception? Error) StreamSnapshot(bool includeConsumers)
    {
        try
        {
            _mu.EnterReadLock();
            var state = _memStore.State();
            var meta = JsonSerializer.SerializeToUtf8Bytes(_cfg);
            var blocks = _blks.ToArray();
            _mu.ExitReadLock();

            var files = new Dictionary<string, byte[]>(StringComparer.Ordinal)
            {
                [FileStoreDefaults.JetStreamMetaFile] = meta,
                [FileStoreDefaults.JetStreamMetaFileSum] = Encoding.ASCII.GetBytes(Convert.ToHexString(SHA256.HashData(meta)).ToLowerInvariant()),
            };

            foreach (var mb in blocks)
            {
                if (!File.Exists(mb.Mfn))
                    continue;
                files[$"{FileStoreDefaults.MsgDir}/{string.Format(FileStoreDefaults.BlkScan, mb.Index)}"] = File.ReadAllBytes(mb.Mfn);
            }

            _ = ForceWriteFullState();
            var stateFile = Path.Combine(_fcfg.StoreDir, FileStoreDefaults.MsgDir, FileStoreDefaults.StreamStateFile);
            if (File.Exists(stateFile))
                files[$"{FileStoreDefaults.MsgDir}/{FileStoreDefaults.StreamStateFile}"] = File.ReadAllBytes(stateFile);

            if (includeConsumers)
            {
                _cmu.EnterReadLock();
                var consumers = _cfs.ToArray();
                _cmu.ExitReadLock();

                var i = 0;
                foreach (var consumer in consumers)
                {
                    var (consumerState, err) = consumer.State();
                    if (err == null && consumerState != null)
                        files[$"{FileStoreDefaults.ConsumerDir}/consumer-{i++}.json"] = JsonSerializer.SerializeToUtf8Bytes(consumerState);
                }
            }

            var payload = JsonSerializer.SerializeToUtf8Bytes(new SnapshotEnvelope
            {
                State = state,
                Files = files,
            });

            return (new MemoryStream(payload, writable: false), null);
        }
        catch (Exception ex)
        {
            return (null, ex);
        }
        finally
        {
            if (_mu.IsReadLockHeld)
                _mu.ExitReadLock();
            if (_cmu.IsReadLockHeld)
                _cmu.ExitReadLock();
        }
    }

    private FileStoreConfig FileStoreConfig()
    {
        _mu.EnterReadLock();
        try
        {
            return new FileStoreConfig
            {
                StoreDir = _fcfg.StoreDir,
                BlockSize = _fcfg.BlockSize,
                CacheExpire = _fcfg.CacheExpire,
                SubjectStateExpire = _fcfg.SubjectStateExpire,
                SyncInterval = _fcfg.SyncInterval,
                SyncAlways = _fcfg.SyncAlways,
                AsyncFlush = _fcfg.AsyncFlush,
                Cipher = _fcfg.Cipher,
                Compression = _fcfg.Compression,
                Server = _fcfg.Server,
            };
        }
        finally
        {
            _mu.ExitReadLock();
        }
    }

    // Lock should be held.
    private void ReadLockAllMsgBlocks()
    {
        foreach (var mb in _blks)
            mb.Mu.EnterReadLock();
    }

    // Lock should be held.
    private void ReadUnlockAllMsgBlocks()
    {
        foreach (var mb in _blks)
        {
            if (mb.Mu.IsReadLockHeld)
                mb.Mu.ExitReadLock();
        }
    }

    // All blocks should be at least read-locked.
    private DeleteBlocks DeleteBlocks()
    {
        var dbs = new DeleteBlocks();
        ulong prevLast = 0;
        DeleteRange? prevRange = null;
        var msgsSinceGap = false;

        foreach (var mb in _blks)
        {
            var fseq = mb.First.Seq;
            if (prevLast > 0 && prevLast + 1 != fseq)
            {
                var gap = fseq - prevLast - 1;
                if (prevRange != null && !msgsSinceGap)
                {
                    prevRange.Num += gap;
                }
                else
                {
                    prevRange = new DeleteRange { First = prevLast + 1, Num = gap };
                    msgsSinceGap = false;
                    dbs.Add(prevRange);
                }
            }

            if (mb.Dmap.Size > 0)
            {
                var deleted = new List<ulong>(mb.Dmap.Size);
                mb.Dmap.Range(seq =>
                {
                    deleted.Add(seq);
                    return true;
                });
                dbs.Add(new DeleteSlice([.. deleted]));
                prevRange = null;
            }

            prevLast = mb.Last.Seq;
            msgsSinceGap = msgsSinceGap || mb.Msgs > 0;
        }

        return dbs;
    }

    private SequenceSet DeleteMap()
    {
        var dmap = new SequenceSet();

        _mu.EnterReadLock();
        try
        {
            ReadLockAllMsgBlocks();
            try
            {
                foreach (var mb in _blks)
                {
                    if (mb.Dmap.Size == 0)
                        continue;

                    mb.Dmap.Range(seq =>
                    {
                        dmap.Insert(seq);
                        return true;
                    });
                }
            }
            finally
            {
                ReadUnlockAllMsgBlocks();
            }
        }
        finally
        {
            _mu.ExitReadLock();
        }

        return dmap;
    }

    private static void AppendUVarInt(List<byte> buffer, ulong value)
    {
        do
        {
            var b = (byte)(value & 0x7Fu);
            value >>= 7;
            if (value != 0)
                b |= 0x80;
            buffer.Add(b);
        }
        while (value != 0);
    }

    private static void AppendVarInt(List<byte> buffer, long value)
    {
        var encoded = (ulong)value << 1;
        if (value < 0)
            encoded = ~encoded;
        AppendUVarInt(buffer, encoded);
    }

    // -----------------------------------------------------------------------
    // IStreamStore — type / state
    // -----------------------------------------------------------------------

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

    /// <inheritdoc/>
    public StreamState State()
        => _memStore.State();

    /// <inheritdoc/>
    public void FastState(StreamState state)
        => _memStore.FastState(state);

    // -----------------------------------------------------------------------
    // IStreamStore — callback registration
    // -----------------------------------------------------------------------

    /// <inheritdoc/>
    public void RegisterStorageUpdates(StorageUpdateHandler cb)
    {
        _scb = cb;
        _memStore.RegisterStorageUpdates(cb);
    }

    /// <inheritdoc/>
    public void RegisterStorageRemoveMsg(StorageRemoveMsgHandler cb)
    {
        _rmcb = cb;
        _memStore.RegisterStorageRemoveMsg(cb);
    }

    /// <inheritdoc/>
    public void RegisterProcessJetStreamMsg(ProcessJetStreamMsgHandler cb)
    {
        _pmsgcb = cb;
        _memStore.RegisterProcessJetStreamMsg(cb);
    }

    // -----------------------------------------------------------------------
    // IStreamStore — lifecycle
    // -----------------------------------------------------------------------

    /// <inheritdoc/>
    public void Stop()
    {
        var err = StopInternal(delete: false, writeState: true);
        if (err != null && !ReferenceEquals(err, StoreErrors.ErrStoreClosed))
            throw err;
    }

    /// <inheritdoc/>
    public void Dispose() => Stop();

    // -----------------------------------------------------------------------
    // IStreamStore — store / load (all stubs)
    // -----------------------------------------------------------------------

    /// <inheritdoc/>
    public (ulong Seq, long Ts) StoreMsg(string subject, byte[]? hdr, byte[]? msg, long ttl)
        => _memStore.StoreMsg(subject, hdr, msg, ttl);

    /// <inheritdoc/>
    public void StoreRawMsg(string subject, byte[]? hdr, byte[]? msg, ulong seq, long ts, long ttl, bool discardNewCheck)
        => _memStore.StoreRawMsg(subject, hdr, msg, seq, ts, ttl, discardNewCheck);

    /// <inheritdoc/>
    public (ulong Seq, Exception? Error) SkipMsg(ulong seq)
        => _memStore.SkipMsg(seq);

    /// <inheritdoc/>
    public void SkipMsgs(ulong seq, ulong num)
        => _memStore.SkipMsgs(seq, num);

    /// <inheritdoc/>
    public void FlushAllPending()
        => _memStore.FlushAllPending();

    /// <inheritdoc/>
    public StoreMsg? LoadMsg(ulong seq, StoreMsg? sm)
        => _memStore.LoadMsg(seq, sm);

    /// <inheritdoc/>
    public (StoreMsg? Sm, ulong Skip) LoadNextMsg(string filter, bool wc, ulong start, StoreMsg? smp)
        => _memStore.LoadNextMsg(filter, wc, start, smp);

    /// <inheritdoc/>
    public (StoreMsg? Sm, ulong Skip) LoadNextMsgMulti(object? sl, ulong start, StoreMsg? smp)
        => _memStore.LoadNextMsgMulti(sl, start, smp);

    /// <inheritdoc/>
    public StoreMsg? LoadLastMsg(string subject, StoreMsg? sm)
    {
        (StoreMsg? Sm, Exception? Error) result =
            string.IsNullOrEmpty(subject) || subject == ">"
                ? MsgForSeq(LastSeq(), sm)
                : LoadLast(subject, sm);

        if (result.Error != null)
            throw result.Error;
        return result.Sm ?? throw StoreErrors.ErrStoreMsgNotFound;
    }

    /// <inheritdoc/>
    public (StoreMsg? Sm, Exception? Error) LoadPrevMsg(ulong start, StoreMsg? smp)
        => _memStore.LoadPrevMsg(start, smp);

    /// <inheritdoc/>
    public (StoreMsg? Sm, ulong Skip, Exception? Error) LoadPrevMsgMulti(object? sl, ulong start, StoreMsg? smp)
        => _memStore.LoadPrevMsgMulti(sl, start, smp);

    /// <inheritdoc/>
    public (bool Removed, Exception? Error) RemoveMsg(ulong seq)
        => _memStore.RemoveMsg(seq);

    /// <inheritdoc/>
    public (bool Removed, Exception? Error) EraseMsg(ulong seq)
        => _memStore.EraseMsg(seq);

    /// <inheritdoc/>
    public (ulong Purged, Exception? Error) Purge()
        => PurgeInternal(0);

    /// <inheritdoc/>
    public (ulong Purged, Exception? Error) PurgeEx(string subject, ulong seq, ulong keep)
        => _memStore.PurgeEx(subject, seq, keep);

    /// <inheritdoc/>
    public (ulong Purged, Exception? Error) Compact(ulong seq)
        => CompactInternal(seq);

    /// <inheritdoc/>
    public void Truncate(ulong seq)
        => _memStore.Truncate(seq);

    // -----------------------------------------------------------------------
    // IStreamStore — query methods (all stubs)
    // -----------------------------------------------------------------------

    /// <inheritdoc/>
    public ulong GetSeqFromTime(DateTime t)
        => _memStore.GetSeqFromTime(t);

    /// <inheritdoc/>
    public SimpleState FilteredState(ulong seq, string subject)
        => _memStore.FilteredState(seq, subject);

    /// <inheritdoc/>
    public Dictionary<string, SimpleState> SubjectsState(string filterSubject)
        => _memStore.SubjectsState(filterSubject);

    /// <inheritdoc/>
    public Dictionary<string, ulong> SubjectsTotals(string filterSubject)
        => _memStore.SubjectsTotals(filterSubject);

    /// <inheritdoc/>
    public (ulong[] Seqs, Exception? Error) AllLastSeqs()
        => _memStore.AllLastSeqs();

    /// <inheritdoc/>
    public (ulong[] Seqs, Exception? Error) MultiLastSeqs(string[] filters, ulong maxSeq, int maxAllowed)
        => _memStore.MultiLastSeqs(filters, maxSeq, maxAllowed);

    /// <inheritdoc/>
    public (string Subject, Exception? Error) SubjectForSeq(ulong seq)
        => _memStore.SubjectForSeq(seq);

    /// <inheritdoc/>
    public (ulong Total, ulong ValidThrough, Exception? Error) NumPending(ulong sseq, string filter, bool lastPerSubject)
        => _memStore.NumPending(sseq, filter, lastPerSubject);

    /// <inheritdoc/>
    public (ulong Total, ulong ValidThrough, Exception? Error) NumPendingMulti(ulong sseq, object? sl, bool lastPerSubject)
        => _memStore.NumPendingMulti(sseq, sl, lastPerSubject);

    // -----------------------------------------------------------------------
    // IStreamStore — stream state encoding (stubs)
    // -----------------------------------------------------------------------

    /// <inheritdoc/>
    public (byte[] Enc, Exception? Error) EncodedStreamState(ulong failed)
        => _memStore.EncodedStreamState(failed);

    /// <inheritdoc/>
    public void SyncDeleted(DeleteBlocks dbs)
    {
        if (dbs.Count == 0)
            return;

        _mu.EnterReadLock();
        try
        {
            if (_state.LastSeq == 0)
                return;
        }
        finally
        {
            _mu.ExitReadLock();
        }

        _memStore.SyncDeleted(dbs);
    }

    // -----------------------------------------------------------------------
    // IStreamStore — config / admin (stubs)
    // -----------------------------------------------------------------------

    /// <inheritdoc/>
    public void UpdateConfig(StreamConfig cfg)
    {
        _cfg.Config = cfg.Clone();
        _memStore.UpdateConfig(cfg);
    }

    /// <inheritdoc/>
    public void Delete(bool inline)
    {
        if (IsClosed())
        {
            try { Directory.Delete(_fcfg.StoreDir, recursive: true); } catch { }
            return;
        }

        _ = StopInternal(delete: true, writeState: false);

        var remove = () =>
        {
            try { Directory.Delete(_fcfg.StoreDir, recursive: true); } catch { }
        };

        if (inline)
            remove();
        else
            _ = Task.Run(remove);
    }

    /// <inheritdoc/>
    public void ResetState()
        => _memStore.ResetState();

    // -----------------------------------------------------------------------
    // IStreamStore — consumer management (stubs)
    // -----------------------------------------------------------------------

    /// <inheritdoc/>
    public IConsumerStore ConsumerStore(string name, DateTime created, ConsumerConfig cfg)
    {
        var cfi = new FileConsumerInfo
        {
            Name = name,
            Created = created,
            Config = cfg,
        };
        var odir = Path.Combine(_fcfg.StoreDir, FileStoreDefaults.ConsumerDir, name);
        Directory.CreateDirectory(odir);
        var cs = new ConsumerFileStore(this, cfi, name, odir);
        AddConsumer(cs);
        return cs;
    }

    /// <inheritdoc/>
    public void AddConsumer(IConsumerStore o)
    {
        _cmu.EnterWriteLock();
        try
        {
            _cfs.Add(o);
            _memStore.AddConsumer(o);
        }
        finally { _cmu.ExitWriteLock(); }
    }

    /// <inheritdoc/>
    public void RemoveConsumer(IConsumerStore o)
    {
        _cmu.EnterWriteLock();
        try
        {
            _cfs.Remove(o);
            _memStore.RemoveConsumer(o);
        }
        finally { _cmu.ExitWriteLock(); }
    }

    // -----------------------------------------------------------------------
    // IStreamStore — snapshot / utilization (stubs)
    // -----------------------------------------------------------------------

    /// <inheritdoc/>
    public (SnapshotResult? Result, Exception? Error) Snapshot(TimeSpan deadline, bool includeConsumers, bool checkMsgs)
    {
        if (IsClosed())
            return (null, StoreErrors.ErrStoreClosed);

        if (checkMsgs)
        {
            var ld = CheckMsgs();
            if (ld is { Msgs.Length: > 0 })
                return (null, new InvalidDataException($"snapshot check detected {ld.Msgs.Length} bad messages"));
        }

        var (reader, err) = StreamSnapshot(includeConsumers);
        if (err != null || reader == null)
            return (null, err ?? new InvalidOperationException("snapshot generation failed"));

        return (new SnapshotResult
        {
            Reader = reader,
            State = _memStore.State(),
        }, null);
    }

    /// <inheritdoc/>
    public (ulong Total, ulong Reported, Exception? Error) Utilization()
        => _memStore.Utilization();

    private sealed class SnapshotEnvelope
    {
        public StreamState State { get; set; } = new();
        public Dictionary<string, byte[]> Files { get; set; } = new(StringComparer.Ordinal);
    }

    internal sealed class XorStreamCipher
    {
        private readonly byte[] _keySeed;

        public XorStreamCipher(byte[] seed, byte[] nonce)
        {
            using var sha = SHA256.Create();
            var input = new byte[seed.Length + nonce.Length];
            Buffer.BlockCopy(seed, 0, input, 0, seed.Length);
            Buffer.BlockCopy(nonce, 0, input, seed.Length, nonce.Length);
            _keySeed = sha.ComputeHash(input);
        }

        public void XorKeyStream(Span<byte> buffer)
        {
            for (var i = 0; i < buffer.Length; i++)
                buffer[i] ^= _keySeed[i % _keySeed.Length];
        }
    }

    internal sealed class AeadCipher
    {
        private const int AeadNonceSize = 12;
        private const int AeadTagSize = 16;

        private readonly byte[] _key;

        public AeadCipher(byte[] seed)
        {
            using var sha = SHA256.Create();
            _key = sha.ComputeHash(seed);
        }

        public int NonceSize => AeadNonceSize;
        public int Overhead => AeadTagSize;

        public byte[] Seal(ReadOnlySpan<byte> nonce, ReadOnlySpan<byte> plaintext)
        {
            if (nonce.Length != AeadNonceSize)
                throw new ArgumentException("invalid nonce size", nameof(nonce));

            var ciphertext = new byte[plaintext.Length];
            var tag = new byte[AeadTagSize];
            using var aes = new AesGcm(_key, AeadTagSize);
            aes.Encrypt(nonce, plaintext, ciphertext, tag);

            var sealedPayload = new byte[ciphertext.Length + tag.Length];
            Buffer.BlockCopy(ciphertext, 0, sealedPayload, 0, ciphertext.Length);
            Buffer.BlockCopy(tag, 0, sealedPayload, ciphertext.Length, tag.Length);
            return sealedPayload;
        }

        public byte[] Open(ReadOnlySpan<byte> nonce, ReadOnlySpan<byte> ciphertextAndTag)
        {
            if (nonce.Length != AeadNonceSize)
                throw new ArgumentException("invalid nonce size", nameof(nonce));
            if (ciphertextAndTag.Length < AeadTagSize)
                throw new ArgumentException("invalid ciphertext size", nameof(ciphertextAndTag));

            var ciphertextLength = ciphertextAndTag.Length - AeadTagSize;
            var ciphertext = ciphertextAndTag[..ciphertextLength];
            var tag = ciphertextAndTag[ciphertextLength..];

            var plaintext = new byte[ciphertextLength];
            using var aes = new AesGcm(_key, AeadTagSize);
            aes.Decrypt(nonce, ciphertext, tag, plaintext);
            return plaintext;
        }
    }

    private sealed class MessageBlockIndexFile
    {
        public ulong Msgs { get; set; }
        public ulong Bytes { get; set; }
        public ulong RawBytes { get; set; }
        public ulong FirstSeq { get; set; }
        public long FirstTs { get; set; }
        public ulong LastSeq { get; set; }
        public long LastTs { get; set; }
        public byte[]? LastChecksum { get; set; }
        public bool NoTrack { get; set; }
    }
}