stella-ops.org/git.stella-ops.org
1
0
Fork 0
Code Issues Pull Requests Actions 2 Releases Wiki Activity

save development progress

Browse Source
This commit is contained in:
StellaOps Bot
2025-12-25 23:09:58 +02:00
parent d71853ad7e
commit aa70af062e
351 changed files with 37683 additions and 150156 deletions
Download Patch File Download Diff File Expand all files Collapse all files

218
src/Concelier/__Libraries/StellaOps.Concelier.Cache.Valkey/AdvisoryCacheKeys.cs Normal file
View File

@@ -0,0 +1,218 @@
// -----------------------------------------------------------------------------
// AdvisoryCacheKeys.cs
// Sprint: SPRINT_8200_0013_0001_GW_valkey_advisory_cache
// Task: VCACHE-8200-004, VCACHE-8200-005, VCACHE-8200-006, VCACHE-8200-007, VCACHE-8200-008
// Description: Key schema for Concelier Valkey cache
// -----------------------------------------------------------------------------
using System.Text;
namespace StellaOps.Concelier.Cache.Valkey;
/// <summary>
/// Static class for generating Valkey cache keys for canonical advisories.
/// </summary>
/// <remarks>
/// Key Schema:
/// <code>
/// advisory:{merge_hash} → JSON(CanonicalAdvisory) - TTL based on interest_score
/// rank:hot → ZSET { merge_hash: interest_score } - max 10,000 entries
/// by:purl:{normalized_purl} → SET { merge_hash, ... } - TTL 24h
/// by:cve:{cve_id} → STRING merge_hash - TTL 24h
/// cache:stats:hits → INCR counter
/// cache:stats:misses → INCR counter
/// cache:warmup:last → STRING ISO8601 timestamp
/// </code>
/// </remarks>
public static class AdvisoryCacheKeys
{
/// <summary>
/// Default key prefix for all cache keys.
/// </summary>
public const string DefaultPrefix = "concelier:";
/// <summary>
/// Key for advisory by merge hash.
/// Format: {prefix}advisory:{mergeHash}
/// </summary>
public static string Advisory(string mergeHash, string prefix = DefaultPrefix)
=> $"{prefix}advisory:{mergeHash}";
/// <summary>
/// Key for the hot advisory sorted set.
/// Format: {prefix}rank:hot
/// </summary>
public static string HotSet(string prefix = DefaultPrefix)
=> $"{prefix}rank:hot";
/// <summary>
/// Key for PURL index set.
/// Format: {prefix}by:purl:{normalizedPurl}
/// </summary>
/// <param name="purl">The PURL (will be normalized).</param>
/// <param name="prefix">Key prefix.</param>
public static string ByPurl(string purl, string prefix = DefaultPrefix)
=> $"{prefix}by:purl:{NormalizePurl(purl)}";
/// <summary>
/// Key for CVE mapping.
/// Format: {prefix}by:cve:{cveId}
/// </summary>
/// <param name="cve">The CVE identifier (case-insensitive).</param>
/// <param name="prefix">Key prefix.</param>
public static string ByCve(string cve, string prefix = DefaultPrefix)
=> $"{prefix}by:cve:{cve.ToUpperInvariant()}";
/// <summary>
/// Key for cache hit counter.
/// Format: {prefix}cache:stats:hits
/// </summary>
public static string StatsHits(string prefix = DefaultPrefix)
=> $"{prefix}cache:stats:hits";
/// <summary>
/// Key for cache miss counter.
/// Format: {prefix}cache:stats:misses
/// </summary>
public static string StatsMisses(string prefix = DefaultPrefix)
=> $"{prefix}cache:stats:misses";
/// <summary>
/// Key for last warmup timestamp.
/// Format: {prefix}cache:warmup:last
/// </summary>
public static string WarmupLast(string prefix = DefaultPrefix)
=> $"{prefix}cache:warmup:last";
/// <summary>
/// Key for warmup lock (for distributed coordination).
/// Format: {prefix}cache:warmup:lock
/// </summary>
public static string WarmupLock(string prefix = DefaultPrefix)
=> $"{prefix}cache:warmup:lock";
/// <summary>
/// Key for total cached advisories gauge.
/// Format: {prefix}cache:stats:count
/// </summary>
public static string StatsCount(string prefix = DefaultPrefix)
=> $"{prefix}cache:stats:count";
/// <summary>
/// Pattern to match all advisory keys (for scanning/cleanup).
/// Format: {prefix}advisory:*
/// </summary>
public static string AdvisoryPattern(string prefix = DefaultPrefix)
=> $"{prefix}advisory:*";
/// <summary>
/// Pattern to match all PURL index keys (for scanning/cleanup).
/// Format: {prefix}by:purl:*
/// </summary>
public static string PurlIndexPattern(string prefix = DefaultPrefix)
=> $"{prefix}by:purl:*";
/// <summary>
/// Pattern to match all CVE mapping keys (for scanning/cleanup).
/// Format: {prefix}by:cve:*
/// </summary>
public static string CveMappingPattern(string prefix = DefaultPrefix)
=> $"{prefix}by:cve:*";
/// <summary>
/// Normalizes a PURL for use as a cache key.
/// </summary>
/// <param name="purl">The PURL to normalize.</param>
/// <returns>Normalized PURL safe for use in cache keys.</returns>
/// <remarks>
/// Normalization:
/// 1. Lowercase the entire PURL
/// 2. Replace special characters that may cause issues in keys
/// 3. Truncate very long PURLs to prevent oversized keys
/// </remarks>
public static string NormalizePurl(string purl)
{
if (string.IsNullOrWhiteSpace(purl))
{
return string.Empty;
}
// Normalize to lowercase
var normalized = purl.ToLowerInvariant();
// Replace characters that could cause issues in Redis keys
// Redis keys should avoid spaces and some special chars for simplicity
var sb = new StringBuilder(normalized.Length);
foreach (var c in normalized)
{
// Allow alphanumeric, standard PURL chars: : / @ . - _ %
if (char.IsLetterOrDigit(c) ||
c is ':' or '/' or '@' or '.' or '-' or '_' or '%')
{
sb.Append(c);
}
else
{
// Replace other chars with underscore
sb.Append('_');
}
}
// Truncate if too long (Redis keys can be up to 512MB, but we want reasonable sizes)
const int MaxKeyLength = 500;
if (sb.Length > MaxKeyLength)
{
return sb.ToString(0, MaxKeyLength);
}
return sb.ToString();
}
/// <summary>
/// Extracts the merge hash from an advisory key.
/// </summary>
/// <param name="key">The full advisory key.</param>
/// <param name="prefix">The key prefix used.</param>
/// <returns>The merge hash, or null if key doesn't match expected format.</returns>
public static string? ExtractMergeHash(string key, string prefix = DefaultPrefix)
{
var expectedStart = $"{prefix}advisory:";
if (key.StartsWith(expectedStart, StringComparison.Ordinal))
{
return key[expectedStart.Length..];
}
return null;
}
/// <summary>
/// Extracts the PURL from a PURL index key.
/// </summary>
/// <param name="key">The full PURL index key.</param>
/// <param name="prefix">The key prefix used.</param>
/// <returns>The normalized PURL, or null if key doesn't match expected format.</returns>
public static string? ExtractPurl(string key, string prefix = DefaultPrefix)
{
var expectedStart = $"{prefix}by:purl:";
if (key.StartsWith(expectedStart, StringComparison.Ordinal))
{
return key[expectedStart.Length..];
}
return null;
}
/// <summary>
/// Extracts the CVE from a CVE mapping key.
/// </summary>
/// <param name="key">The full CVE mapping key.</param>
/// <param name="prefix">The key prefix used.</param>
/// <returns>The CVE identifier, or null if key doesn't match expected format.</returns>
public static string? ExtractCve(string key, string prefix = DefaultPrefix)
{
var expectedStart = $"{prefix}by:cve:";
if (key.StartsWith(expectedStart, StringComparison.Ordinal))
{
return key[expectedStart.Length..];
}
return null;
}
}

64
src/Concelier/__Libraries/StellaOps.Concelier.Cache.Valkey/CacheWarmupHostedService.cs Normal file
View File

@@ -0,0 +1,64 @@
// -----------------------------------------------------------------------------
// CacheWarmupHostedService.cs
// Sprint: SPRINT_8200_0013_0001_GW_valkey_advisory_cache
// Task: VCACHE-8200-024
// Description: Background service for cache warmup on startup
// -----------------------------------------------------------------------------
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
namespace StellaOps.Concelier.Cache.Valkey;
/// <summary>
/// Background hosted service that warms the advisory cache on application startup.
/// </summary>
public sealed class CacheWarmupHostedService : BackgroundService
{
private readonly IAdvisoryCacheService _cacheService;
private readonly ConcelierCacheOptions _options;
private readonly ILogger<CacheWarmupHostedService>? _logger;
/// <summary>
/// Initializes a new instance of <see cref="CacheWarmupHostedService"/>.
/// </summary>
public CacheWarmupHostedService(
IAdvisoryCacheService cacheService,
IOptions<ConcelierCacheOptions> options,
ILogger<CacheWarmupHostedService>? logger = null)
{
_cacheService = cacheService;
_options = options.Value;
_logger = logger;
}
/// <inheritdoc />
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
if (!_options.Enabled || !_options.EnableWarmup)
{
_logger?.LogInformation("Cache warmup is disabled");
return;
}
// Wait a short time for the application to fully start
await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken).ConfigureAwait(false);
_logger?.LogInformation("Starting cache warmup with limit {Limit}", _options.WarmupLimit);
try
{
await _cacheService.WarmupAsync(_options.WarmupLimit, stoppingToken).ConfigureAwait(false);
_logger?.LogInformation("Cache warmup completed successfully");
}
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
{
_logger?.LogInformation("Cache warmup cancelled");
}
catch (Exception ex)
{
_logger?.LogError(ex, "Cache warmup failed");
}
}
}

173
src/Concelier/__Libraries/StellaOps.Concelier.Cache.Valkey/ConcelierCacheConnectionFactory.cs Normal file
View File

@@ -0,0 +1,173 @@
// -----------------------------------------------------------------------------
// ConcelierCacheConnectionFactory.cs
// Sprint: SPRINT_8200_0013_0001_GW_valkey_advisory_cache
// Task: VCACHE-8200-003
// Description: Connection factory for Concelier Valkey cache
// -----------------------------------------------------------------------------
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using StackExchange.Redis;
namespace StellaOps.Concelier.Cache.Valkey;
/// <summary>
/// Factory for creating and managing Valkey/Redis connections for the Concelier cache.
/// Thread-safe with lazy connection initialization and automatic reconnection.
/// </summary>
public sealed class ConcelierCacheConnectionFactory : IAsyncDisposable
{
private readonly ConcelierCacheOptions _options;
private readonly ILogger<ConcelierCacheConnectionFactory>? _logger;
private readonly SemaphoreSlim _connectionLock = new(1, 1);
private readonly Func<ConfigurationOptions, Task<IConnectionMultiplexer>> _connectionFactory;
private IConnectionMultiplexer? _connection;
private bool _disposed;
/// <summary>
/// Initializes a new instance of <see cref="ConcelierCacheConnectionFactory"/>.
/// </summary>
/// <param name="options">Cache configuration options.</param>
/// <param name="logger">Optional logger.</param>
/// <param name="connectionFactory">Optional connection factory for testing.</param>
public ConcelierCacheConnectionFactory(
IOptions<ConcelierCacheOptions> options,
ILogger<ConcelierCacheConnectionFactory>? logger = null,
Func<ConfigurationOptions, Task<IConnectionMultiplexer>>? connectionFactory = null)
{
_options = options.Value;
_logger = logger;
_connectionFactory = connectionFactory ??
(config => Task.FromResult<IConnectionMultiplexer>(ConnectionMultiplexer.Connect(config)));
}
/// <summary>
/// Gets whether caching is enabled.
/// </summary>
public bool IsEnabled => _options.Enabled;
/// <summary>
/// Gets a database connection for cache operations.
/// </summary>
/// <param name="cancellationToken">Cancellation token.</param>
/// <returns>The Valkey database.</returns>
public async ValueTask<IDatabase> GetDatabaseAsync(CancellationToken cancellationToken = default)
{
var connection = await GetConnectionAsync(cancellationToken).ConfigureAwait(false);
return connection.GetDatabase(_options.Database);
}
/// <summary>
/// Gets the underlying connection multiplexer.
/// </summary>
/// <param name="cancellationToken">Cancellation token.</param>
/// <returns>The connection multiplexer.</returns>
public async ValueTask<IConnectionMultiplexer> GetConnectionAsync(CancellationToken cancellationToken = default)
{
ObjectDisposedException.ThrowIf(_disposed, this);
if (_connection is not null && _connection.IsConnected)
{
return _connection;
}
await _connectionLock.WaitAsync(cancellationToken).ConfigureAwait(false);
try
{
if (_connection is null || !_connection.IsConnected)
{
if (_connection is not null)
{
_logger?.LogDebug("Reconnecting to Valkey (previous connection lost)");
await _connection.CloseAsync().ConfigureAwait(false);
_connection.Dispose();
}
var config = ConfigurationOptions.Parse(_options.ConnectionString);
config.AbortOnConnectFail = _options.AbortOnConnectFail;
config.ConnectTimeout = (int)_options.ConnectTimeout.TotalMilliseconds;
config.SyncTimeout = (int)_options.SyncTimeout.TotalMilliseconds;
config.AsyncTimeout = (int)_options.AsyncTimeout.TotalMilliseconds;
config.ConnectRetry = _options.ConnectRetry;
config.DefaultDatabase = _options.Database;
_logger?.LogDebug("Connecting to Valkey at {Endpoint} (database {Database})",
_options.ConnectionString, _options.Database);
_connection = await _connectionFactory(config).ConfigureAwait(false);
_logger?.LogInformation("Connected to Valkey for Concelier cache");
}
}
finally
{
_connectionLock.Release();
}
return _connection;
}
/// <summary>
/// Tests the connection by sending a PING command.
/// </summary>
/// <param name="cancellationToken">Cancellation token.</param>
/// <returns>True if connection is healthy.</returns>
public async ValueTask<bool> PingAsync(CancellationToken cancellationToken = default)
{
if (!_options.Enabled)
{
return false;
}
try
{
var db = await GetDatabaseAsync(cancellationToken).ConfigureAwait(false);
var pong = await db.PingAsync().ConfigureAwait(false);
_logger?.LogDebug("Valkey PING response: {Latency}ms", pong.TotalMilliseconds);
return true;
}
catch (Exception ex)
{
_logger?.LogWarning(ex, "Valkey PING failed");
return false;
}
}
/// <summary>
/// Gets the key prefix for cache keys.
/// </summary>
public string KeyPrefix => _options.KeyPrefix;
/// <summary>
/// Gets the maximum hot set size.
/// </summary>
public int MaxHotSetSize => _options.MaxHotSetSize;
/// <summary>
/// Gets the TTL policy.
/// </summary>
public CacheTtlPolicy TtlPolicy => _options.TtlPolicy;
/// <summary>
/// Disposes the connection factory and releases the connection.
/// </summary>
public async ValueTask DisposeAsync()
{
if (_disposed)
{
return;
}
_disposed = true;
if (_connection is not null)
{
_logger?.LogDebug("Closing Valkey connection");
await _connection.CloseAsync().ConfigureAwait(false);
_connection.Dispose();
}
_connectionLock.Dispose();
}
}

195
src/Concelier/__Libraries/StellaOps.Concelier.Cache.Valkey/ConcelierCacheMetrics.cs Normal file
View File

@@ -0,0 +1,195 @@
// -----------------------------------------------------------------------------
// ConcelierCacheMetrics.cs
// Sprint: SPRINT_8200_0013_0001_GW_valkey_advisory_cache
// Task: VCACHE-8200-027, VCACHE-8200-028
// Description: OpenTelemetry metrics for cache operations
// -----------------------------------------------------------------------------
using System.Diagnostics;
using System.Diagnostics.Metrics;
namespace StellaOps.Concelier.Cache.Valkey;
/// <summary>
/// Metrics instrumentation for the Concelier advisory cache.
/// </summary>
public sealed class ConcelierCacheMetrics : IDisposable
{
/// <summary>
/// Activity source name for cache operations.
/// </summary>
public const string ActivitySourceName = "StellaOps.Concelier.Cache";
/// <summary>
/// Meter name for cache metrics.
/// </summary>
public const string MeterName = "StellaOps.Concelier.Cache";
private readonly Meter _meter;
private readonly Counter<long> _hitsCounter;
private readonly Counter<long> _missesCounter;
private readonly Counter<long> _evictionsCounter;
private readonly Histogram<double> _latencyHistogram;
private readonly ObservableGauge<long> _hotSetSizeGauge;
private long _lastKnownHotSetSize;
/// <summary>
/// Activity source for tracing cache operations.
/// </summary>
public static ActivitySource ActivitySource { get; } = new(ActivitySourceName, "1.0.0");
/// <summary>
/// Initializes a new instance of <see cref="ConcelierCacheMetrics"/>.
/// </summary>
public ConcelierCacheMetrics()
{
_meter = new Meter(MeterName, "1.0.0");
_hitsCounter = _meter.CreateCounter<long>(
"concelier_cache_hits_total",
unit: "{hits}",
description: "Total number of cache hits");
_missesCounter = _meter.CreateCounter<long>(
"concelier_cache_misses_total",
unit: "{misses}",
description: "Total number of cache misses");
_evictionsCounter = _meter.CreateCounter<long>(
"concelier_cache_evictions_total",
unit: "{evictions}",
description: "Total number of cache evictions");
_latencyHistogram = _meter.CreateHistogram<double>(
"concelier_cache_latency_ms",
unit: "ms",
description: "Cache operation latency in milliseconds");
_hotSetSizeGauge = _meter.CreateObservableGauge(
"concelier_cache_hot_set_size",
() => _lastKnownHotSetSize,
unit: "{entries}",
description: "Current number of entries in the hot advisory set");
}
/// <summary>
/// Records a cache hit.
/// </summary>
public void RecordHit() => _hitsCounter.Add(1);
/// <summary>
/// Records a cache miss.
/// </summary>
public void RecordMiss() => _missesCounter.Add(1);
/// <summary>
/// Records a cache eviction.
/// </summary>
/// <param name="reason">The reason for eviction.</param>
public void RecordEviction(string reason = "ttl")
{
_evictionsCounter.Add(1, new KeyValuePair<string, object?>("reason", reason));
}
/// <summary>
/// Records operation latency.
/// </summary>
/// <param name="milliseconds">Latency in milliseconds.</param>
/// <param name="operation">The operation type (get, set, invalidate).</param>
public void RecordLatency(double milliseconds, string operation)
{
_latencyHistogram.Record(milliseconds, new KeyValuePair<string, object?>("operation", operation));
}
/// <summary>
/// Updates the hot set size gauge.
/// </summary>
/// <param name="size">Current hot set size.</param>
public void UpdateHotSetSize(long size)
{
_lastKnownHotSetSize = size;
}
/// <summary>
/// Starts an activity for tracing a cache operation.
/// </summary>
/// <param name="operationName">Name of the operation.</param>
/// <returns>The activity, or null if tracing is disabled.</returns>
public static Activity? StartActivity(string operationName)
{
return ActivitySource.StartActivity(operationName, ActivityKind.Internal);
}
/// <summary>
/// Starts an activity with tags.
/// </summary>
/// <param name="operationName">Name of the operation.</param>
/// <param name="tags">Tags to add to the activity.</param>
/// <returns>The activity, or null if tracing is disabled.</returns>
public static Activity? StartActivity(string operationName, params (string Key, object? Value)[] tags)
{
var activity = ActivitySource.StartActivity(operationName, ActivityKind.Internal);
if (activity is not null)
{
foreach (var (key, value) in tags)
{
activity.SetTag(key, value);
}
}
return activity;
}
/// <inheritdoc />
public void Dispose()
{
_meter.Dispose();
ActivitySource.Dispose();
}
}
/// <summary>
/// Extension methods for timing cache operations.
/// </summary>
public static class CacheMetricsExtensions
{
/// <summary>
/// Times an async operation and records the latency.
/// </summary>
public static async Task<T> TimeAsync<T>(
this ConcelierCacheMetrics metrics,
string operation,
Func<Task<T>> action)
{
var sw = Stopwatch.StartNew();
try
{
return await action().ConfigureAwait(false);
}
finally
{
sw.Stop();
metrics.RecordLatency(sw.Elapsed.TotalMilliseconds, operation);
}
}
/// <summary>
/// Times an async operation and records the latency.
/// </summary>
public static async Task TimeAsync(
this ConcelierCacheMetrics metrics,
string operation,
Func<Task> action)
{
var sw = Stopwatch.StartNew();
try
{
await action().ConfigureAwait(false);
}
finally
{
sw.Stop();
metrics.RecordLatency(sw.Elapsed.TotalMilliseconds, operation);
}
}
}

145
src/Concelier/__Libraries/StellaOps.Concelier.Cache.Valkey/ConcelierCacheOptions.cs Normal file
View File

@@ -0,0 +1,145 @@
// -----------------------------------------------------------------------------
// ConcelierCacheOptions.cs
// Sprint: SPRINT_8200_0013_0001_GW_valkey_advisory_cache
// Task: VCACHE-8200-002
// Description: Configuration options for Concelier Valkey cache
// -----------------------------------------------------------------------------
namespace StellaOps.Concelier.Cache.Valkey;
/// <summary>
/// Configuration options for the Concelier Valkey advisory cache.
/// </summary>
public sealed class ConcelierCacheOptions
{
/// <summary>
/// Configuration section name.
/// </summary>
public const string SectionName = "Concelier:Cache";
/// <summary>
/// Whether Valkey caching is enabled.
/// </summary>
public bool Enabled { get; set; } = true;
/// <summary>
/// Valkey connection string (e.g., "localhost:6379" or "valkey:6379,password=secret").
/// </summary>
public string ConnectionString { get; set; } = "localhost:6379";
/// <summary>
/// Valkey database number (0-15).
/// </summary>
public int Database { get; set; } = 1;
/// <summary>
/// Key prefix for all cache keys.
/// </summary>
public string KeyPrefix { get; set; } = "concelier:";
/// <summary>
/// Maximum hot set size.
/// </summary>
public int MaxHotSetSize { get; set; } = 10_000;
/// <summary>
/// Connection timeout.
/// </summary>
public TimeSpan ConnectTimeout { get; set; } = TimeSpan.FromSeconds(5);
/// <summary>
/// Synchronous operation timeout.
/// </summary>
public TimeSpan SyncTimeout { get; set; } = TimeSpan.FromMilliseconds(100);
/// <summary>
/// Async operation timeout.
/// </summary>
public TimeSpan AsyncTimeout { get; set; } = TimeSpan.FromMilliseconds(200);
/// <summary>
/// Number of connection retries.
/// </summary>
public int ConnectRetry { get; set; } = 3;
/// <summary>
/// Whether to abort on connect fail.
/// </summary>
public bool AbortOnConnectFail { get; set; } = false;
/// <summary>
/// TTL policy configuration.
/// </summary>
public CacheTtlPolicy TtlPolicy { get; set; } = new();
/// <summary>
/// Whether to enable cache warmup on startup.
/// </summary>
public bool EnableWarmup { get; set; } = true;
/// <summary>
/// Number of advisories to preload during warmup.
/// </summary>
public int WarmupLimit { get; set; } = 1000;
}
/// <summary>
/// TTL policy for cached advisories based on interest score.
/// </summary>
public sealed class CacheTtlPolicy
{
/// <summary>
/// TTL for high interest advisories (score >= 0.7).
/// </summary>
public TimeSpan HighScoreTtl { get; set; } = TimeSpan.FromHours(24);
/// <summary>
/// TTL for medium interest advisories (score >= 0.4).
/// </summary>
public TimeSpan MediumScoreTtl { get; set; } = TimeSpan.FromHours(4);
/// <summary>
/// TTL for low interest advisories (score < 0.4).
/// </summary>
public TimeSpan LowScoreTtl { get; set; } = TimeSpan.FromHours(1);
/// <summary>
/// Threshold for high interest score.
/// </summary>
public double HighScoreThreshold { get; set; } = 0.7;
/// <summary>
/// Threshold for medium interest score.
/// </summary>
public double MediumScoreThreshold { get; set; } = 0.4;
/// <summary>
/// TTL for PURL index entries.
/// </summary>
public TimeSpan PurlIndexTtl { get; set; } = TimeSpan.FromHours(24);
/// <summary>
/// TTL for CVE mapping entries.
/// </summary>
public TimeSpan CveMappingTtl { get; set; } = TimeSpan.FromHours(24);
/// <summary>
/// Gets the appropriate TTL for an advisory based on its interest score.
/// </summary>
/// <param name="score">Interest score (0.0 - 1.0), or null for default.</param>
/// <returns>TTL for the advisory.</returns>
public TimeSpan GetTtl(double? score)
{
if (!score.HasValue)
{
return LowScoreTtl;
}
return score.Value switch
{
>= 0.7 => HighScoreTtl, // High interest: 24h
>= 0.4 => MediumScoreTtl, // Medium interest: 4h
_ => LowScoreTtl // Low interest: 1h
};
}
}

155
src/Concelier/__Libraries/StellaOps.Concelier.Cache.Valkey/IAdvisoryCacheService.cs Normal file
View File

@@ -0,0 +1,155 @@
// -----------------------------------------------------------------------------
// IAdvisoryCacheService.cs
// Sprint: SPRINT_8200_0013_0001_GW_valkey_advisory_cache
// Task: VCACHE-8200-010
// Description: Interface for Valkey-based canonical advisory caching
// -----------------------------------------------------------------------------
using StellaOps.Concelier.Core.Canonical;
namespace StellaOps.Concelier.Cache.Valkey;
/// <summary>
/// Valkey-based cache for canonical advisories.
/// Provides read-through caching with TTL based on interest score.
/// </summary>
public interface IAdvisoryCacheService
{
// === Read Operations ===
/// <summary>
/// Get canonical advisory by merge hash (cache-first).
/// </summary>
/// <param name="mergeHash">The merge hash identifying the canonical.</param>
/// <param name="cancellationToken">Cancellation token.</param>
/// <returns>The cached advisory, or null if not found.</returns>
Task<CanonicalAdvisory?> GetAsync(string mergeHash, CancellationToken cancellationToken = default);
/// <summary>
/// Get canonical advisories by PURL (uses index).
/// </summary>
/// <param name="purl">The PURL to lookup.</param>
/// <param name="cancellationToken">Cancellation token.</param>
/// <returns>List of advisories affecting this PURL.</returns>
Task<IReadOnlyList<CanonicalAdvisory>> GetByPurlAsync(string purl, CancellationToken cancellationToken = default);
/// <summary>
/// Get canonical advisory by CVE (uses mapping).
/// </summary>
/// <param name="cve">The CVE identifier.</param>
/// <param name="cancellationToken">Cancellation token.</param>
/// <returns>The primary canonical for this CVE, or null.</returns>
Task<CanonicalAdvisory?> GetByCveAsync(string cve, CancellationToken cancellationToken = default);
/// <summary>
/// Get hot advisories (top N by interest score).
/// </summary>
/// <param name="limit">Maximum number to return.</param>
/// <param name="cancellationToken">Cancellation token.</param>
/// <returns>List of hot advisories in descending score order.</returns>
Task<IReadOnlyList<CanonicalAdvisory>> GetHotAsync(int limit = 100, CancellationToken cancellationToken = default);
// === Write Operations ===
/// <summary>
/// Cache canonical advisory with TTL based on interest score.
/// </summary>
/// <param name="advisory">The advisory to cache.</param>
/// <param name="interestScore">Optional interest score (0.0-1.0) for TTL calculation.</param>
/// <param name="cancellationToken">Cancellation token.</param>
Task SetAsync(CanonicalAdvisory advisory, double? interestScore = null, CancellationToken cancellationToken = default);
/// <summary>
/// Invalidate cached advisory.
/// </summary>
/// <param name="mergeHash">The merge hash to invalidate.</param>
/// <param name="cancellationToken">Cancellation token.</param>
Task InvalidateAsync(string mergeHash, CancellationToken cancellationToken = default);
/// <summary>
/// Update interest score (affects TTL and hot set membership).
/// </summary>
/// <param name="mergeHash">The merge hash to update.</param>
/// <param name="score">The new interest score (0.0-1.0).</param>
/// <param name="cancellationToken">Cancellation token.</param>
Task UpdateScoreAsync(string mergeHash, double score, CancellationToken cancellationToken = default);
// === Index Operations ===
/// <summary>
/// Add merge hash to PURL index.
/// </summary>
/// <param name="purl">The PURL to index.</param>
/// <param name="mergeHash">The merge hash to add.</param>
/// <param name="cancellationToken">Cancellation token.</param>
Task IndexPurlAsync(string purl, string mergeHash, CancellationToken cancellationToken = default);
/// <summary>
/// Remove merge hash from PURL index.
/// </summary>
/// <param name="purl">The PURL to unindex.</param>
/// <param name="mergeHash">The merge hash to remove.</param>
/// <param name="cancellationToken">Cancellation token.</param>
Task UnindexPurlAsync(string purl, string mergeHash, CancellationToken cancellationToken = default);
/// <summary>
/// Set CVE to merge hash mapping.
/// </summary>
/// <param name="cve">The CVE identifier.</param>
/// <param name="mergeHash">The canonical merge hash.</param>
/// <param name="cancellationToken">Cancellation token.</param>
Task IndexCveAsync(string cve, string mergeHash, CancellationToken cancellationToken = default);
// === Maintenance ===
/// <summary>
/// Warm cache with hot advisories from database.
/// </summary>
/// <param name="limit">Maximum number of advisories to preload.</param>
/// <param name="cancellationToken">Cancellation token.</param>
Task WarmupAsync(int limit = 1000, CancellationToken cancellationToken = default);
/// <summary>
/// Get cache statistics.
/// </summary>
/// <param name="cancellationToken">Cancellation token.</param>
/// <returns>Current cache statistics.</returns>
Task<CacheStatistics> GetStatisticsAsync(CancellationToken cancellationToken = default);
/// <summary>
/// Check if the cache service is healthy.
/// </summary>
/// <param name="cancellationToken">Cancellation token.</param>
/// <returns>True if the cache is reachable and operational.</returns>
Task<bool> IsHealthyAsync(CancellationToken cancellationToken = default);
}
/// <summary>
/// Cache statistics for monitoring and debugging.
/// </summary>
public sealed record CacheStatistics
{
/// <summary>Total cache hits.</summary>
public long Hits { get; init; }
/// <summary>Total cache misses.</summary>
public long Misses { get; init; }
/// <summary>Cache hit rate (0.0-1.0).</summary>
public double HitRate => Hits + Misses > 0 ? (double)Hits / (Hits + Misses) : 0;
/// <summary>Current size of the hot advisory set.</summary>
public long HotSetSize { get; init; }
/// <summary>Approximate total cached advisories.</summary>
public long TotalCachedAdvisories { get; init; }
/// <summary>When the cache was last warmed up.</summary>
public DateTimeOffset? LastWarmup { get; init; }
/// <summary>Whether the cache service is currently healthy.</summary>
public bool IsHealthy { get; init; }
/// <summary>Valkey server info string.</summary>
public string? ServerInfo { get; init; }
}

169
src/Concelier/__Libraries/StellaOps.Concelier.Cache.Valkey/ServiceCollectionExtensions.cs Normal file
View File

@@ -0,0 +1,169 @@
// -----------------------------------------------------------------------------
// ServiceCollectionExtensions.cs
// Sprint: SPRINT_8200_0013_0001_GW_valkey_advisory_cache
// Task: VCACHE-8200-011
// Description: DI registration for Concelier Valkey cache services
// -----------------------------------------------------------------------------
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.DependencyInjection.Extensions;
namespace StellaOps.Concelier.Cache.Valkey;
/// <summary>
/// Extension methods for registering Concelier cache services.
/// </summary>
public static class ServiceCollectionExtensions
{
/// <summary>
/// Adds Concelier Valkey cache services to the service collection.
/// </summary>
/// <param name="services">The service collection.</param>
/// <param name="configuration">The configuration root.</param>
/// <param name="enableWarmup">Whether to enable background cache warmup.</param>
/// <returns>The service collection for chaining.</returns>
public static IServiceCollection AddConcelierValkeyCache(
this IServiceCollection services,
IConfiguration configuration,
bool enableWarmup = true)
{
// Bind options from configuration
services.Configure<ConcelierCacheOptions>(
configuration.GetSection(ConcelierCacheOptions.SectionName));
return AddCoreServices(services, enableWarmup);
}
/// <summary>
/// Adds Concelier Valkey cache services with custom options.
/// </summary>
/// <param name="services">The service collection.</param>
/// <param name="configureOptions">Action to configure options.</param>
/// <param name="enableWarmup">Whether to enable background cache warmup.</param>
/// <returns>The service collection for chaining.</returns>
public static IServiceCollection AddConcelierValkeyCache(
this IServiceCollection services,
Action<ConcelierCacheOptions> configureOptions,
bool enableWarmup = true)
{
services.Configure(configureOptions);
return AddCoreServices(services, enableWarmup);
}
private static IServiceCollection AddCoreServices(IServiceCollection services, bool enableWarmup)
{
// Register connection factory as singleton (manages connection lifecycle)
services.TryAddSingleton<ConcelierCacheConnectionFactory>();
// Register metrics
services.TryAddSingleton<ConcelierCacheMetrics>();
// Register cache service
services.TryAddSingleton<IAdvisoryCacheService, ValkeyAdvisoryCacheService>();
// Register warmup hosted service if enabled
if (enableWarmup)
{
services.AddHostedService<CacheWarmupHostedService>();
}
return services;
}
/// <summary>
/// Decorates the registered <see cref="ICanonicalAdvisoryService"/> with Valkey caching.
/// Call this after registering the base ICanonicalAdvisoryService implementation.
/// </summary>
/// <param name="services">The service collection.</param>
/// <returns>The service collection for chaining.</returns>
public static IServiceCollection AddValkeyCachingDecorator(this IServiceCollection services)
{
// Find the existing ICanonicalAdvisoryService registration
var existingDescriptor = services.FirstOrDefault(
d => d.ServiceType == typeof(StellaOps.Concelier.Core.Canonical.ICanonicalAdvisoryService));
if (existingDescriptor is null)
{
throw new InvalidOperationException(
"ICanonicalAdvisoryService must be registered before adding the Valkey caching decorator. " +
"Call AddConcelierCore() or register ICanonicalAdvisoryService first.");
}
// Remove the original registration
services.Remove(existingDescriptor);
// Re-register the original service with a different key for the decorator to use
if (existingDescriptor.ImplementationType is not null)
{
services.Add(new ServiceDescriptor(
existingDescriptor.ImplementationType,
existingDescriptor.ImplementationType,
existingDescriptor.Lifetime));
}
else if (existingDescriptor.ImplementationFactory is not null)
{
services.Add(new ServiceDescriptor(
typeof(StellaOps.Concelier.Core.Canonical.ICanonicalAdvisoryService),
sp => existingDescriptor.ImplementationFactory(sp),
existingDescriptor.Lifetime));
}
// Register the decorator as the new ICanonicalAdvisoryService
services.Add(new ServiceDescriptor(
typeof(StellaOps.Concelier.Core.Canonical.ICanonicalAdvisoryService),
sp =>
{
// Resolve the inner service (the original implementation)
StellaOps.Concelier.Core.Canonical.ICanonicalAdvisoryService innerService;
if (existingDescriptor.ImplementationType is not null)
{
innerService = (StellaOps.Concelier.Core.Canonical.ICanonicalAdvisoryService)
sp.GetRequiredService(existingDescriptor.ImplementationType);
}
else if (existingDescriptor.ImplementationFactory is not null)
{
innerService = (StellaOps.Concelier.Core.Canonical.ICanonicalAdvisoryService)
existingDescriptor.ImplementationFactory(sp);
}
else if (existingDescriptor.ImplementationInstance is not null)
{
innerService = (StellaOps.Concelier.Core.Canonical.ICanonicalAdvisoryService)
existingDescriptor.ImplementationInstance;
}
else
{
throw new InvalidOperationException(
"Unable to resolve inner ICanonicalAdvisoryService for decorator.");
}
var cache = sp.GetRequiredService<IAdvisoryCacheService>();
var logger = sp.GetRequiredService<Microsoft.Extensions.Logging.ILogger<ValkeyCanonicalAdvisoryService>>();
return new ValkeyCanonicalAdvisoryService(innerService, cache, logger);
},
existingDescriptor.Lifetime));
return services;
}
/// <summary>
/// Adds Valkey caching for canonical advisories, including the decorator.
/// This is a convenience method that combines AddConcelierValkeyCache and AddValkeyCachingDecorator.
/// </summary>
/// <param name="services">The service collection.</param>
/// <param name="configuration">The configuration root.</param>
/// <param name="enableWarmup">Whether to enable background cache warmup.</param>
/// <returns>The service collection for chaining.</returns>
public static IServiceCollection AddConcelierValkeyCacheWithDecorator(
this IServiceCollection services,
IConfiguration configuration,
bool enableWarmup = true)
{
services.AddConcelierValkeyCache(configuration, enableWarmup);
services.AddValkeyCachingDecorator();
return services;
}
}

30
src/Concelier/__Libraries/StellaOps.Concelier.Cache.Valkey/StellaOps.Concelier.Cache.Valkey.csproj Normal file
View File

@@ -0,0 +1,30 @@
<?xml version="1.0" encoding="utf-8"?>
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net10.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
<LangVersion>preview</LangVersion>
<TreatWarningsAsErrors>false</TreatWarningsAsErrors>
<RootNamespace>StellaOps.Concelier.Cache.Valkey</RootNamespace>
<AssemblyName>StellaOps.Concelier.Cache.Valkey</AssemblyName>
<Description>Valkey/Redis caching for Concelier canonical advisories</Description>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Microsoft.Extensions.Configuration.Abstractions" Version="10.0.0" />
<PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="10.0.0" />
<PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="10.0.0" />
<PackageReference Include="Microsoft.Extensions.Options" Version="10.0.0" />
<PackageReference Include="Microsoft.Extensions.Options.ConfigurationExtensions" Version="10.0.0" />
<PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="10.0.0" />
<PackageReference Include="StackExchange.Redis" Version="2.8.37" />
<PackageReference Include="System.Diagnostics.DiagnosticSource" Version="9.0.0" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\StellaOps.Concelier.Core\StellaOps.Concelier.Core.csproj" />
</ItemGroup>
</Project>

493
src/Concelier/__Libraries/StellaOps.Concelier.Cache.Valkey/ValkeyAdvisoryCacheService.cs Normal file
View File

@@ -0,0 +1,493 @@
// -----------------------------------------------------------------------------
// ValkeyAdvisoryCacheService.cs
// Sprint: SPRINT_8200_0013_0001_GW_valkey_advisory_cache
// Task: VCACHE-8200-011 to VCACHE-8200-016
// Description: Valkey implementation of advisory cache service
// -----------------------------------------------------------------------------
using System.Diagnostics;
using System.Text.Json;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using StackExchange.Redis;
using StellaOps.Concelier.Core.Canonical;
namespace StellaOps.Concelier.Cache.Valkey;
/// <summary>
/// Valkey-based implementation of the advisory cache service.
/// Provides read-through caching with TTL based on interest scores.
/// </summary>
public sealed class ValkeyAdvisoryCacheService : IAdvisoryCacheService
{
private readonly ConcelierCacheConnectionFactory _connectionFactory;
private readonly ConcelierCacheOptions _options;
private readonly ILogger<ValkeyAdvisoryCacheService>? _logger;
private static readonly JsonSerializerOptions JsonOptions = new()
{
PropertyNamingPolicy = JsonNamingPolicy.CamelCase,
WriteIndented = false
};
/// <summary>
/// Initializes a new instance of <see cref="ValkeyAdvisoryCacheService"/>.
/// </summary>
public ValkeyAdvisoryCacheService(
ConcelierCacheConnectionFactory connectionFactory,
IOptions<ConcelierCacheOptions> options,
ILogger<ValkeyAdvisoryCacheService>? logger = null)
{
_connectionFactory = connectionFactory;
_options = options.Value;
_logger = logger;
}
/// <inheritdoc />
public async Task<CanonicalAdvisory?> GetAsync(string mergeHash, CancellationToken cancellationToken = default)
{
if (!_options.Enabled)
{
return null;
}
try
{
var db = await _connectionFactory.GetDatabaseAsync(cancellationToken).ConfigureAwait(false);
var key = AdvisoryCacheKeys.Advisory(mergeHash, _options.KeyPrefix);
var cached = await db.StringGetAsync(key).ConfigureAwait(false);
if (cached.HasValue)
{
await db.StringIncrementAsync(AdvisoryCacheKeys.StatsHits(_options.KeyPrefix)).ConfigureAwait(false);
_logger?.LogDebug("Cache hit for advisory {MergeHash}", mergeHash);
return JsonSerializer.Deserialize<CanonicalAdvisory>((string)cached!, JsonOptions);
}
await db.StringIncrementAsync(AdvisoryCacheKeys.StatsMisses(_options.KeyPrefix)).ConfigureAwait(false);
_logger?.LogDebug("Cache miss for advisory {MergeHash}", mergeHash);
return null;
}
catch (Exception ex)
{
_logger?.LogWarning(ex, "Failed to get advisory {MergeHash} from cache", mergeHash);
return null;
}
}
/// <inheritdoc />
public async Task<IReadOnlyList<CanonicalAdvisory>> GetByPurlAsync(string purl, CancellationToken cancellationToken = default)
{
if (!_options.Enabled)
{
return [];
}
try
{
var db = await _connectionFactory.GetDatabaseAsync(cancellationToken).ConfigureAwait(false);
var indexKey = AdvisoryCacheKeys.ByPurl(purl, _options.KeyPrefix);
// Get all merge hashes for this PURL
var mergeHashes = await db.SetMembersAsync(indexKey).ConfigureAwait(false);
if (mergeHashes.Length == 0)
{
return [];
}
// Refresh TTL on access
await db.KeyExpireAsync(indexKey, _options.TtlPolicy.PurlIndexTtl).ConfigureAwait(false);
// Fetch all advisories
var results = new List<CanonicalAdvisory>(mergeHashes.Length);
foreach (var hash in mergeHashes)
{
var advisory = await GetAsync(hash!, cancellationToken).ConfigureAwait(false);
if (advisory is not null)
{
results.Add(advisory);
}
}
return results;
}
catch (Exception ex)
{
_logger?.LogWarning(ex, "Failed to get advisories for PURL {Purl}", purl);
return [];
}
}
/// <inheritdoc />
public async Task<CanonicalAdvisory?> GetByCveAsync(string cve, CancellationToken cancellationToken = default)
{
if (!_options.Enabled)
{
return null;
}
try
{
var db = await _connectionFactory.GetDatabaseAsync(cancellationToken).ConfigureAwait(false);
var mappingKey = AdvisoryCacheKeys.ByCve(cve, _options.KeyPrefix);
var mergeHash = await db.StringGetAsync(mappingKey).ConfigureAwait(false);
if (!mergeHash.HasValue)
{
return null;
}
return await GetAsync(mergeHash!, cancellationToken).ConfigureAwait(false);
}
catch (Exception ex)
{
_logger?.LogWarning(ex, "Failed to get advisory for CVE {Cve}", cve);
return null;
}
}
/// <inheritdoc />
public async Task<IReadOnlyList<CanonicalAdvisory>> GetHotAsync(int limit = 100, CancellationToken cancellationToken = default)
{
if (!_options.Enabled)
{
return [];
}
try
{
var db = await _connectionFactory.GetDatabaseAsync(cancellationToken).ConfigureAwait(false);
var hotKey = AdvisoryCacheKeys.HotSet(_options.KeyPrefix);
// Get top N merge hashes by score (descending)
var entries = await db.SortedSetRangeByRankAsync(
hotKey,
start: 0,
stop: limit - 1,
order: Order.Descending).ConfigureAwait(false);
if (entries.Length == 0)
{
return [];
}
// Fetch all advisories
var results = new List<CanonicalAdvisory>(entries.Length);
foreach (var mergeHash in entries)
{
var advisory = await GetAsync(mergeHash!, cancellationToken).ConfigureAwait(false);
if (advisory is not null)
{
results.Add(advisory);
}
}
return results;
}
catch (Exception ex)
{
_logger?.LogWarning(ex, "Failed to get hot advisories");
return [];
}
}
/// <inheritdoc />
public async Task SetAsync(CanonicalAdvisory advisory, double? interestScore = null, CancellationToken cancellationToken = default)
{
if (!_options.Enabled)
{
return;
}
try
{
var db = await _connectionFactory.GetDatabaseAsync(cancellationToken).ConfigureAwait(false);
var key = AdvisoryCacheKeys.Advisory(advisory.MergeHash, _options.KeyPrefix);
var json = JsonSerializer.Serialize(advisory, JsonOptions);
var ttl = _options.TtlPolicy.GetTtl(interestScore);
await db.StringSetAsync(key, json, ttl).ConfigureAwait(false);
_logger?.LogDebug("Cached advisory {MergeHash} with TTL {Ttl}", advisory.MergeHash, ttl);
// Update hot set if score provided
if (interestScore.HasValue)
{
await UpdateScoreAsync(advisory.MergeHash, interestScore.Value, cancellationToken).ConfigureAwait(false);
}
// Index CVE mapping
if (!string.IsNullOrWhiteSpace(advisory.Cve))
{
await IndexCveAsync(advisory.Cve, advisory.MergeHash, cancellationToken).ConfigureAwait(false);
}
// Index by PURL (affects key)
if (!string.IsNullOrWhiteSpace(advisory.AffectsKey))
{
await IndexPurlAsync(advisory.AffectsKey, advisory.MergeHash, cancellationToken).ConfigureAwait(false);
}
}
catch (Exception ex)
{
_logger?.LogWarning(ex, "Failed to cache advisory {MergeHash}", advisory.MergeHash);
}
}
/// <inheritdoc />
public async Task InvalidateAsync(string mergeHash, CancellationToken cancellationToken = default)
{
if (!_options.Enabled)
{
return;
}
try
{
var db = await _connectionFactory.GetDatabaseAsync(cancellationToken).ConfigureAwait(false);
// Remove from advisory cache
var key = AdvisoryCacheKeys.Advisory(mergeHash, _options.KeyPrefix);
await db.KeyDeleteAsync(key).ConfigureAwait(false);
// Remove from hot set
var hotKey = AdvisoryCacheKeys.HotSet(_options.KeyPrefix);
await db.SortedSetRemoveAsync(hotKey, mergeHash).ConfigureAwait(false);
_logger?.LogDebug("Invalidated advisory {MergeHash}", mergeHash);
}
catch (Exception ex)
{
_logger?.LogWarning(ex, "Failed to invalidate advisory {MergeHash}", mergeHash);
}
}
/// <inheritdoc />
public async Task UpdateScoreAsync(string mergeHash, double score, CancellationToken cancellationToken = default)
{
if (!_options.Enabled)
{
return;
}
try
{
var db = await _connectionFactory.GetDatabaseAsync(cancellationToken).ConfigureAwait(false);
var hotKey = AdvisoryCacheKeys.HotSet(_options.KeyPrefix);
// Add/update in hot set
await db.SortedSetAddAsync(hotKey, mergeHash, score).ConfigureAwait(false);
// Trim to max size
var currentSize = await db.SortedSetLengthAsync(hotKey).ConfigureAwait(false);
if (currentSize > _options.MaxHotSetSize)
{
// Remove lowest scoring entries
await db.SortedSetRemoveRangeByRankAsync(
hotKey,
start: 0,
stop: currentSize - _options.MaxHotSetSize - 1).ConfigureAwait(false);
}
// Update advisory TTL if cached
var advisoryKey = AdvisoryCacheKeys.Advisory(mergeHash, _options.KeyPrefix);
if (await db.KeyExistsAsync(advisoryKey).ConfigureAwait(false))
{
var ttl = _options.TtlPolicy.GetTtl(score);
await db.KeyExpireAsync(advisoryKey, ttl).ConfigureAwait(false);
}
_logger?.LogDebug("Updated score for {MergeHash} to {Score}", mergeHash, score);
}
catch (Exception ex)
{
_logger?.LogWarning(ex, "Failed to update score for {MergeHash}", mergeHash);
}
}
/// <inheritdoc />
public async Task IndexPurlAsync(string purl, string mergeHash, CancellationToken cancellationToken = default)
{
if (!_options.Enabled)
{
return;
}
try
{
var db = await _connectionFactory.GetDatabaseAsync(cancellationToken).ConfigureAwait(false);
var indexKey = AdvisoryCacheKeys.ByPurl(purl, _options.KeyPrefix);
await db.SetAddAsync(indexKey, mergeHash).ConfigureAwait(false);
await db.KeyExpireAsync(indexKey, _options.TtlPolicy.PurlIndexTtl).ConfigureAwait(false);
_logger?.LogDebug("Indexed PURL {Purl} -> {MergeHash}", purl, mergeHash);
}
catch (Exception ex)
{
_logger?.LogWarning(ex, "Failed to index PURL {Purl}", purl);
}
}
/// <inheritdoc />
public async Task UnindexPurlAsync(string purl, string mergeHash, CancellationToken cancellationToken = default)
{
if (!_options.Enabled)
{
return;
}
try
{
var db = await _connectionFactory.GetDatabaseAsync(cancellationToken).ConfigureAwait(false);
var indexKey = AdvisoryCacheKeys.ByPurl(purl, _options.KeyPrefix);
await db.SetRemoveAsync(indexKey, mergeHash).ConfigureAwait(false);
_logger?.LogDebug("Unindexed PURL {Purl} -> {MergeHash}", purl, mergeHash);
}
catch (Exception ex)
{
_logger?.LogWarning(ex, "Failed to unindex PURL {Purl}", purl);
}
}
/// <inheritdoc />
public async Task IndexCveAsync(string cve, string mergeHash, CancellationToken cancellationToken = default)
{
if (!_options.Enabled)
{
return;
}
try
{
var db = await _connectionFactory.GetDatabaseAsync(cancellationToken).ConfigureAwait(false);
var mappingKey = AdvisoryCacheKeys.ByCve(cve, _options.KeyPrefix);
await db.StringSetAsync(mappingKey, mergeHash, _options.TtlPolicy.CveMappingTtl).ConfigureAwait(false);
_logger?.LogDebug("Indexed CVE {Cve} -> {MergeHash}", cve, mergeHash);
}
catch (Exception ex)
{
_logger?.LogWarning(ex, "Failed to index CVE {Cve}", cve);
}
}
/// <inheritdoc />
public async Task WarmupAsync(int limit = 1000, CancellationToken cancellationToken = default)
{
if (!_options.Enabled || !_options.EnableWarmup)
{
return;
}
var sw = Stopwatch.StartNew();
_logger?.LogInformation("Starting cache warmup (limit: {Limit})", limit);
try
{
var db = await _connectionFactory.GetDatabaseAsync(cancellationToken).ConfigureAwait(false);
// Try to acquire warmup lock (prevent concurrent warmups)
var lockKey = AdvisoryCacheKeys.WarmupLock(_options.KeyPrefix);
var lockAcquired = await db.StringSetAsync(
lockKey,
"warming",
TimeSpan.FromMinutes(10),
When.NotExists).ConfigureAwait(false);
if (!lockAcquired)
{
_logger?.LogDebug("Warmup already in progress, skipping");
return;
}
try
{
// Record warmup timestamp
var warmupKey = AdvisoryCacheKeys.WarmupLast(_options.KeyPrefix);
await db.StringSetAsync(warmupKey, DateTimeOffset.UtcNow.ToString("o")).ConfigureAwait(false);
// Note: Actual warmup would load from ICanonicalAdvisoryStore
// This is a placeholder - the actual implementation would be in the integration layer
_logger?.LogInformation("Cache warmup completed in {Elapsed}ms", sw.ElapsedMilliseconds);
}
finally
{
// Release lock
await db.KeyDeleteAsync(lockKey).ConfigureAwait(false);
}
}
catch (Exception ex)
{
_logger?.LogWarning(ex, "Cache warmup failed after {Elapsed}ms", sw.ElapsedMilliseconds);
}
}
/// <inheritdoc />
public async Task<CacheStatistics> GetStatisticsAsync(CancellationToken cancellationToken = default)
{
if (!_options.Enabled)
{
return new CacheStatistics { IsHealthy = false };
}
try
{
var db = await _connectionFactory.GetDatabaseAsync(cancellationToken).ConfigureAwait(false);
var hitsKey = AdvisoryCacheKeys.StatsHits(_options.KeyPrefix);
var missesKey = AdvisoryCacheKeys.StatsMisses(_options.KeyPrefix);
var hotKey = AdvisoryCacheKeys.HotSet(_options.KeyPrefix);
var warmupKey = AdvisoryCacheKeys.WarmupLast(_options.KeyPrefix);
var hits = (long)(await db.StringGetAsync(hitsKey).ConfigureAwait(false));
var misses = (long)(await db.StringGetAsync(missesKey).ConfigureAwait(false));
var hotSetSize = await db.SortedSetLengthAsync(hotKey).ConfigureAwait(false);
DateTimeOffset? lastWarmup = null;
var warmupStr = await db.StringGetAsync(warmupKey).ConfigureAwait(false);
if (warmupStr.HasValue && DateTimeOffset.TryParse(warmupStr, out var parsed))
{
lastWarmup = parsed;
}
// Get server info
var connection = await _connectionFactory.GetConnectionAsync(cancellationToken).ConfigureAwait(false);
var server = connection.GetServer(connection.GetEndPoints().First());
var info = (await server.InfoAsync().ConfigureAwait(false))
.FirstOrDefault(g => g.Key == "Server")?
.FirstOrDefault(e => e.Key == "redis_version")
.Value;
return new CacheStatistics
{
Hits = hits,
Misses = misses,
HotSetSize = hotSetSize,
TotalCachedAdvisories = hotSetSize, // Approximation
LastWarmup = lastWarmup,
IsHealthy = true,
ServerInfo = info
};
}
catch (Exception ex)
{
_logger?.LogWarning(ex, "Failed to get cache statistics");
return new CacheStatistics { IsHealthy = false };
}
}
/// <inheritdoc />
public async Task<bool> IsHealthyAsync(CancellationToken cancellationToken = default)
{
if (!_options.Enabled)
{
return false;
}
return await _connectionFactory.PingAsync(cancellationToken).ConfigureAwait(false);
}
}

335
src/Concelier/__Libraries/StellaOps.Concelier.Cache.Valkey/ValkeyCanonicalAdvisoryService.cs Normal file
View File

@@ -0,0 +1,335 @@
// -----------------------------------------------------------------------------
// ValkeyCanonicalAdvisoryService.cs
// Sprint: SPRINT_8200_0013_0001_GW_valkey_advisory_cache
// Task: VCACHE-8200-026
// Description: Decorator that integrates Valkey cache with CanonicalAdvisoryService
// -----------------------------------------------------------------------------
using Microsoft.Extensions.Logging;
using StellaOps.Concelier.Core.Canonical;
namespace StellaOps.Concelier.Cache.Valkey;
/// <summary>
/// Decorator that integrates Valkey distributed cache with the canonical advisory service.
/// Provides cache-first reads with automatic population and invalidation on writes.
/// </summary>
public sealed class ValkeyCanonicalAdvisoryService : ICanonicalAdvisoryService
{
private readonly ICanonicalAdvisoryService _inner;
private readonly IAdvisoryCacheService _cache;
private readonly ILogger<ValkeyCanonicalAdvisoryService> _logger;
/// <summary>
/// Initializes a new instance of <see cref="ValkeyCanonicalAdvisoryService"/>.
/// </summary>
/// <param name="inner">The inner canonical advisory service.</param>
/// <param name="cache">The Valkey cache service.</param>
/// <param name="logger">Logger instance.</param>
public ValkeyCanonicalAdvisoryService(
ICanonicalAdvisoryService inner,
IAdvisoryCacheService cache,
ILogger<ValkeyCanonicalAdvisoryService> logger)
{
_inner = inner ?? throw new ArgumentNullException(nameof(inner));
_cache = cache ?? throw new ArgumentNullException(nameof(cache));
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
}
#region Ingest Operations (Write-through with cache population)
/// <inheritdoc />
public async Task<IngestResult> IngestAsync(
string source,
RawAdvisory rawAdvisory,
CancellationToken ct = default)
{
var result = await _inner.IngestAsync(source, rawAdvisory, ct).ConfigureAwait(false);
// Populate or invalidate cache based on result
if (result.Decision != MergeDecision.Duplicate)
{
await InvalidateAndRefreshCacheAsync(result.CanonicalId, result.MergeHash, rawAdvisory.Cve, ct)
.ConfigureAwait(false);
}
return result;
}
/// <inheritdoc />
public async Task<IReadOnlyList<IngestResult>> IngestBatchAsync(
string source,
IEnumerable<RawAdvisory> advisories,
CancellationToken ct = default)
{
var results = await _inner.IngestBatchAsync(source, advisories, ct).ConfigureAwait(false);
// Invalidate cache for all affected entries (non-duplicates)
var affectedResults = results.Where(r => r.Decision != MergeDecision.Duplicate).ToList();
foreach (var result in affectedResults)
{
try
{
await _cache.InvalidateAsync(result.MergeHash, ct).ConfigureAwait(false);
}
catch (Exception ex)
{
_logger.LogWarning(
ex,
"Failed to invalidate cache for canonical {CanonicalId} during batch ingest",
result.CanonicalId);
}
}
if (affectedResults.Count > 0)
{
_logger.LogDebug(
"Invalidated {Count} cache entries during batch ingest",
affectedResults.Count);
}
return results;
}
#endregion
#region Query Operations (Cache-first with read-through)
/// <inheritdoc />
public async Task<CanonicalAdvisory?> GetByIdAsync(Guid id, CancellationToken ct = default)
{
// For ID-based lookups, we need to fetch from store first to get merge hash
// unless we maintain an ID->merge_hash index
var result = await _inner.GetByIdAsync(id, ct).ConfigureAwait(false);
if (result is not null)
{
// Populate cache for future merge-hash based lookups
try
{
await _cache.SetAsync(result, null, ct).ConfigureAwait(false);
}
catch (Exception ex)
{
_logger.LogWarning(ex, "Failed to populate cache for canonical {CanonicalId}", id);
}
}
return result;
}
/// <inheritdoc />
public async Task<CanonicalAdvisory?> GetByMergeHashAsync(string mergeHash, CancellationToken ct = default)
{
// Try cache first
try
{
var cached = await _cache.GetAsync(mergeHash, ct).ConfigureAwait(false);
if (cached is not null)
{
_logger.LogTrace("Valkey cache hit for merge hash {MergeHash}", mergeHash);
return cached;
}
}
catch (Exception ex)
{
_logger.LogWarning(ex, "Failed to get from cache, falling back to store for {MergeHash}", mergeHash);
}
// Cache miss - fetch from store
var result = await _inner.GetByMergeHashAsync(mergeHash, ct).ConfigureAwait(false);
if (result is not null)
{
// Populate cache
try
{
await _cache.SetAsync(result, null, ct).ConfigureAwait(false);
_logger.LogTrace("Populated cache for merge hash {MergeHash}", mergeHash);
}
catch (Exception ex)
{
_logger.LogWarning(ex, "Failed to populate cache for merge hash {MergeHash}", mergeHash);
}
}
return result;
}
/// <inheritdoc />
public async Task<IReadOnlyList<CanonicalAdvisory>> GetByCveAsync(string cve, CancellationToken ct = default)
{
// Note: The cache stores only the primary canonical per CVE (GetByCveAsync returns single item).
// For full results, we always query the store but use cache to accelerate individual lookups.
// This is intentional - CVE queries may return multiple canonicals (different affected packages).
// Fetch from store (the source of truth for multiple canonicals per CVE)
var results = await _inner.GetByCveAsync(cve, ct).ConfigureAwait(false);
// Populate cache for each result (for future individual lookups)
foreach (var advisory in results)
{
try
{
await _cache.SetAsync(advisory, null, ct).ConfigureAwait(false);
}
catch (Exception ex)
{
_logger.LogWarning(
ex,
"Failed to populate cache for advisory {MergeHash} (CVE: {Cve})",
advisory.MergeHash, cve);
}
}
if (results.Count > 0)
{
_logger.LogTrace("Fetched {Count} advisories for CVE {Cve} and cached them", results.Count, cve);
}
return results;
}
/// <inheritdoc />
public async Task<IReadOnlyList<CanonicalAdvisory>> GetByArtifactAsync(
string artifactKey,
CancellationToken ct = default)
{
// Try cache first (uses PURL index)
try
{
var cached = await _cache.GetByPurlAsync(artifactKey, ct).ConfigureAwait(false);
if (cached.Count > 0)
{
_logger.LogTrace(
"Valkey cache hit for artifact {ArtifactKey} ({Count} items)",
artifactKey, cached.Count);
return cached;
}
}
catch (Exception ex)
{
_logger.LogWarning(ex, "Failed to get from cache for artifact {ArtifactKey}", artifactKey);
}
// Cache miss - fetch from store
var results = await _inner.GetByArtifactAsync(artifactKey, ct).ConfigureAwait(false);
// Populate cache for each result and update PURL index
foreach (var advisory in results)
{
try
{
await _cache.SetAsync(advisory, null, ct).ConfigureAwait(false);
await _cache.IndexPurlAsync(artifactKey, advisory.MergeHash, ct).ConfigureAwait(false);
}
catch (Exception ex)
{
_logger.LogWarning(
ex,
"Failed to populate cache for advisory {MergeHash} (artifact: {ArtifactKey})",
advisory.MergeHash, artifactKey);
}
}
return results;
}
/// <inheritdoc />
public Task<PagedResult<CanonicalAdvisory>> QueryAsync(
CanonicalQueryOptions options,
CancellationToken ct = default)
{
// Complex queries bypass cache - pass through to store
// Individual results could be cached, but we don't cache the query itself
return _inner.QueryAsync(options, ct);
}
#endregion
#region Status Operations (Write-through with cache invalidation)
/// <inheritdoc />
public async Task UpdateStatusAsync(Guid id, CanonicalStatus status, CancellationToken ct = default)
{
await _inner.UpdateStatusAsync(id, status, ct).ConfigureAwait(false);
// Fetch the canonical to get merge hash for cache invalidation
try
{
var canonical = await _inner.GetByIdAsync(id, ct).ConfigureAwait(false);
if (canonical is not null)
{
await _cache.InvalidateAsync(canonical.MergeHash, ct).ConfigureAwait(false);
_logger.LogDebug(
"Invalidated cache for canonical {CanonicalId} after status update to {Status}",
id, status);
}
}
catch (Exception ex)
{
_logger.LogWarning(
ex,
"Failed to invalidate cache for canonical {CanonicalId} after status update",
id);
}
}
/// <inheritdoc />
public Task<int> DegradeToStubsAsync(double scoreThreshold, CancellationToken ct = default)
{
// This may affect many entries - invalidation will happen naturally through TTL
// or could be done through a background job if needed
return _inner.DegradeToStubsAsync(scoreThreshold, ct);
}
#endregion
#region Private Helpers
private async Task InvalidateAndRefreshCacheAsync(
Guid canonicalId,
string mergeHash,
string? cve,
CancellationToken ct)
{
try
{
// Invalidate existing entry
await _cache.InvalidateAsync(mergeHash, ct).ConfigureAwait(false);
// Fetch fresh data from store and populate cache
var canonical = await _inner.GetByIdAsync(canonicalId, ct).ConfigureAwait(false);
if (canonical is not null)
{
await _cache.SetAsync(canonical, null, ct).ConfigureAwait(false);
// Update CVE index
if (!string.IsNullOrWhiteSpace(cve))
{
await _cache.IndexCveAsync(cve, mergeHash, ct).ConfigureAwait(false);
}
// Update PURL index
if (!string.IsNullOrWhiteSpace(canonical.AffectsKey))
{
await _cache.IndexPurlAsync(canonical.AffectsKey, mergeHash, ct).ConfigureAwait(false);
}
_logger.LogDebug(
"Refreshed cache for canonical {CanonicalId} with merge hash {MergeHash}",
canonicalId, mergeHash);
}
}
catch (Exception ex)
{
_logger.LogWarning(
ex,
"Failed to refresh cache for canonical {CanonicalId}",
canonicalId);
}
}
#endregion
}