lmxopcua/src/Core/ZB.MOM.WW.OtOpcUa.Configuration/LocalCache/LiteDbConfigCache.cs

using LiteDB;

namespace ZB.MOM.WW.OtOpcUa.Configuration.LocalCache;

/// <summary>
///     LiteDB-backed <see cref="ILocalConfigCache"/>. One file per node (default
///     <c>config_cache.db</c>), one collection per snapshot. Corruption surfaces as
///     <see cref="LocalConfigCacheCorruptException"/> on construction or read — callers should
///     delete and re-fetch from the central DB.
/// </summary>
public sealed class LiteDbConfigCache : ILocalConfigCache, IDisposable
{
    private const string CollectionName = "generations";

    // LiteDB's default BsonMapper.Global is a process-wide singleton whose per-type member
    // registration is lazy and NOT thread-safe across concurrently-constructed LiteDatabase
    // instances. When several caches (this one + GenerationSealedCache) initialise in parallel
    // the global mapper races, surfacing as "Member ClusterId not found on BsonMapper" or a
    // bogus "duplicate key _id = 0" (the int auto-id mapping was lost so Insert writes a literal
    // 0 twice). Give each database a private, pre-registered mapper so member
    // resolution happens once, single-threaded, at construction and never touches the global.
    private static BsonMapper BuildMapper()
    {
        var mapper = new BsonMapper();
        mapper.Entity<GenerationSnapshot>();
        return mapper;
    }

    private readonly LiteDatabase _db;
    private readonly ILiteCollection<GenerationSnapshot> _col;
    // PutAsync is a find-then-insert/update; without serialization, two concurrent puts for the
    // same (ClusterId, GenerationId) can both observe `existing is null` and both Insert,
    // producing duplicate rows. Serialize writes through this semaphore so
    // the read-modify-write block is atomic for a given instance. LiteDB itself only locks the
    // page-level write, not the find-then-insert window.
    private readonly SemaphoreSlim _writeGate = new(initialCount: 1, maxCount: 1);

    /// <summary>Initializes a new instance of the <see cref="LiteDbConfigCache"/> class.</summary>
    /// <param name="dbPath">Path to the LiteDB database file.</param>
    public LiteDbConfigCache(string dbPath)
    {
        // LiteDB can be tolerant of header-only corruption at construction time (it may overwrite
        // the header and "recover"), so we force a write + read probe to fail fast on real corruption.
        try
        {
            _db = new LiteDatabase(new ConnectionString { Filename = dbPath, Upgrade = true }, BuildMapper());
            _col = _db.GetCollection<GenerationSnapshot>(CollectionName);
            _col.EnsureIndex(s => s.ClusterId);
            _col.EnsureIndex(s => s.GenerationId);
            _ = _col.Count();
        }
        catch (Exception ex) when (ex is LiteException or InvalidDataException or IOException
                                      or NotSupportedException or UnauthorizedAccessException
                                      or ArgumentOutOfRangeException or FormatException)
        {
            _db?.Dispose();
            throw new LocalConfigCacheCorruptException(
                $"LiteDB cache at '{dbPath}' is corrupt or unreadable — delete the file and refetch from the central DB.",
                ex);
        }
    }

    /// <inheritdoc />
    public Task<GenerationSnapshot?> GetMostRecentAsync(string clusterId, CancellationToken ct = default)
    {
        ct.ThrowIfCancellationRequested();
        var snapshot = _col
            .Find(s => s.ClusterId == clusterId)
            .OrderByDescending(s => s.GenerationId)
            .FirstOrDefault();
        return Task.FromResult<GenerationSnapshot?>(snapshot);
    }

    /// <inheritdoc />
    public async Task PutAsync(GenerationSnapshot snapshot, CancellationToken ct = default)
    {
        ct.ThrowIfCancellationRequested();
        // Serialize the find-then-insert/update so concurrent callers do not observe a stale
        // `existing is null` and both Insert. LiteDB's per-call lock is not enough — the
        // read and the write are independent calls.
        await _writeGate.WaitAsync(ct).ConfigureAwait(false);
        try
        {
            // upsert by (ClusterId, GenerationId) — replace in place if already cached
            var existing = _col
                .Find(s => s.ClusterId == snapshot.ClusterId && s.GenerationId == snapshot.GenerationId)
                .FirstOrDefault();

            if (existing is null)
                _col.Insert(snapshot);
            else
            {
                snapshot.Id = existing.Id;
                _col.Update(snapshot);
            }
        }
        finally
        {
            _writeGate.Release();
        }
    }

    /// <inheritdoc />
    public Task PruneOldGenerationsAsync(string clusterId, int keepLatest = 10, CancellationToken ct = default)
    {
        ct.ThrowIfCancellationRequested();
        var doomed = _col
            .Find(s => s.ClusterId == clusterId)
            .OrderByDescending(s => s.GenerationId)
            .Skip(keepLatest)
            .Select(s => s.Id)
            .ToList();

        foreach (var id in doomed)
            _col.Delete(id);

        return Task.CompletedTask;
    }

    /// <summary>Releases all resources used by the cache.</summary>
    public void Dispose()
    {
        _writeGate.Dispose();
        _db.Dispose();
    }
}

public sealed class LocalConfigCacheCorruptException(string message, Exception inner)
    : Exception(message, inner);