up

src/Policy/StellaOps.Policy.Engine/Caching/MessagingPolicyEvaluationCache.cs

@@ -0,0 +1,202 @@
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+using StellaOps.Messaging;
+using StellaOps.Messaging.Abstractions;
+using StellaOps.Policy.Engine.Options;
+
+namespace StellaOps.Policy.Engine.Caching;
+
+/// <summary>
+/// Policy evaluation cache backed by <see cref="IDistributedCache{TValue}"/>.
+/// Supports any transport (InMemory, Valkey, PostgreSQL) via factory injection.
+/// </summary>
+public sealed class MessagingPolicyEvaluationCache : IPolicyEvaluationCache
+{
+    private readonly IDistributedCache<PolicyEvaluationCacheEntry> _cache;
+    private readonly TimeProvider _timeProvider;
+    private readonly ILogger<MessagingPolicyEvaluationCache> _logger;
+    private readonly TimeSpan _defaultTtl;
+
+    private long _totalRequests;
+    private long _cacheHits;
+    private long _cacheMisses;
+
+    public MessagingPolicyEvaluationCache(
+        IDistributedCacheFactory cacheFactory,
+        ILogger<MessagingPolicyEvaluationCache> logger,
+        TimeProvider timeProvider,
+        IOptions<PolicyEngineOptions> options)
+    {
+        ArgumentNullException.ThrowIfNull(cacheFactory);
+        _logger = logger ?? throw new ArgumentNullException(nameof(logger));
+        _timeProvider = timeProvider ?? throw new ArgumentNullException(nameof(timeProvider));
+
+        var cacheOptions = options?.Value.EvaluationCache ?? new PolicyEvaluationCacheOptions();
+        _defaultTtl = TimeSpan.FromMinutes(cacheOptions.DefaultTtlMinutes);
+
+        _cache = cacheFactory.Create<PolicyEvaluationCacheEntry>(new CacheOptions
+        {
+            KeyPrefix = "pe:",
+            DefaultTtl = _defaultTtl,
+        });
+
+        _logger.LogInformation(
+            "Initialized MessagingPolicyEvaluationCache with provider {Provider}, TTL {Ttl}",
+            _cache.ProviderName,
+            _defaultTtl);
+    }
+
+    public async Task<PolicyEvaluationCacheResult> GetAsync(
+        PolicyEvaluationCacheKey key,
+        CancellationToken cancellationToken = default)
+    {
+        Interlocked.Increment(ref _totalRequests);
+
+        var cacheKey = key.ToCacheKey();
+        var result = await _cache.GetAsync(cacheKey, cancellationToken).ConfigureAwait(false);
+
+        if (result.HasValue)
+        {
+            var entry = result.Value;
+            var now = _timeProvider.GetUtcNow();
+
+            // Check if entry is still valid
+            if (entry.ExpiresAt > now)
+            {
+                Interlocked.Increment(ref _cacheHits);
+                return new PolicyEvaluationCacheResult(entry, true, MapSource(_cache.ProviderName));
+            }
+
+            // Entry expired - remove it
+            await _cache.InvalidateAsync(cacheKey, cancellationToken).ConfigureAwait(false);
+        }
+
+        Interlocked.Increment(ref _cacheMisses);
+        return new PolicyEvaluationCacheResult(null, false, CacheSource.None);
+    }
+
+    public async Task<PolicyEvaluationCacheBatch> GetBatchAsync(
+        IReadOnlyList<PolicyEvaluationCacheKey> keys,
+        CancellationToken cancellationToken = default)
+    {
+        var found = new Dictionary<PolicyEvaluationCacheKey, PolicyEvaluationCacheEntry>();
+        var notFound = new List<PolicyEvaluationCacheKey>();
+        var hits = 0;
+        var misses = 0;
+
+        foreach (var key in keys)
+        {
+            var result = await GetAsync(key, cancellationToken).ConfigureAwait(false);
+            if (result.Entry != null)
+            {
+                found[key] = result.Entry;
+                hits++;
+            }
+            else
+            {
+                notFound.Add(key);
+                misses++;
+            }
+        }
+
+        var source = MapSource(_cache.ProviderName);
+        return new PolicyEvaluationCacheBatch
+        {
+            Found = found,
+            NotFound = notFound,
+            CacheHits = hits,
+            CacheMisses = misses,
+            InMemoryHits = source == CacheSource.InMemory ? hits : 0,
+            RedisHits = source == CacheSource.Redis ? hits : 0,
+        };
+    }
+
+    public async Task SetAsync(
+        PolicyEvaluationCacheKey key,
+        PolicyEvaluationCacheEntry entry,
+        CancellationToken cancellationToken = default)
+    {
+        var cacheKey = key.ToCacheKey();
+        var now = _timeProvider.GetUtcNow();
+        var expiresAt = entry.ExpiresAt > now ? entry.ExpiresAt : now.Add(_defaultTtl);
+
+        var ttl = expiresAt - now;
+        if (ttl <= TimeSpan.Zero)
+        {
+            return;
+        }
+
+        var options = new CacheEntryOptions { TimeToLive = ttl };
+        await _cache.SetAsync(cacheKey, entry, options, cancellationToken).ConfigureAwait(false);
+    }
+
+    public async Task SetBatchAsync(
+        IReadOnlyDictionary<PolicyEvaluationCacheKey, PolicyEvaluationCacheEntry> entries,
+        CancellationToken cancellationToken = default)
+    {
+        var now = _timeProvider.GetUtcNow();
+
+        foreach (var (key, entry) in entries)
+        {
+            var cacheKey = key.ToCacheKey();
+            var expiresAt = entry.ExpiresAt > now ? entry.ExpiresAt : now.Add(_defaultTtl);
+
+            var ttl = expiresAt - now;
+            if (ttl <= TimeSpan.Zero)
+            {
+                continue;
+            }
+
+            var options = new CacheEntryOptions { TimeToLive = ttl };
+            await _cache.SetAsync(cacheKey, entry, options, cancellationToken).ConfigureAwait(false);
+        }
+    }
+
+    public async Task InvalidateAsync(
+        PolicyEvaluationCacheKey key,
+        CancellationToken cancellationToken = default)
+    {
+        var cacheKey = key.ToCacheKey();
+        await _cache.InvalidateAsync(cacheKey, cancellationToken).ConfigureAwait(false);
+    }
+
+    public async Task InvalidateByPolicyDigestAsync(
+        string policyDigest,
+        CancellationToken cancellationToken = default)
+    {
+        // Pattern: pe:<policyDigest>:*
+        var pattern = $"{policyDigest}:*";
+        var count = await _cache.InvalidateByPatternAsync(pattern, cancellationToken).ConfigureAwait(false);
+
+        _logger.LogDebug(
+            "Invalidated {Count} cache entries for policy digest {Digest}",
+            count,
+            policyDigest);
+    }
+
+    public PolicyEvaluationCacheStats GetStats()
+    {
+        var source = MapSource(_cache.ProviderName);
+        var hits = Interlocked.Read(ref _cacheHits);
+
+        return new PolicyEvaluationCacheStats
+        {
+            TotalRequests = Interlocked.Read(ref _totalRequests),
+            CacheHits = hits,
+            CacheMisses = Interlocked.Read(ref _cacheMisses),
+            InMemoryHits = source == CacheSource.InMemory ? hits : 0,
+            RedisHits = source == CacheSource.Redis ? hits : 0,
+            RedisFallbacks = 0,
+            ItemCount = 0, // Not available from IDistributedCache
+            EvictionCount = 0, // Not available from IDistributedCache
+        };
+    }
+
+    private static CacheSource MapSource(string providerName) => providerName.ToLowerInvariant() switch
+    {
+        "inmemory" => CacheSource.InMemory,
+        "valkey" => CacheSource.Redis,
+        "redis" => CacheSource.Redis,
+        _ => CacheSource.None,
+    };
+}