Add MongoDB storage library and update acceptance tests with deterministic stubs
- Created StellaOps.Notify.Storage.Mongo project with initial configuration. - Added expected output files for acceptance tests (at1.txt to at10.txt). - Added fixture input files for acceptance tests (at1 to at10). - Created input and signature files for test cases fc1 to fc5.
This commit is contained in:
StellaOps Bot
@@ -0,0 +1,945 @@
|
||||
using System.Collections.Concurrent;
|
||||
using System.Text.Json;
|
||||
using System.Text.Json.Nodes;
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.DependencyInjection.Extensions;
|
||||
using StellaOps.Notify.Models;
|
||||
|
||||
namespace StellaOps.Notify.Storage.Mongo.Documents;
|
||||
|
||||
public sealed class NotifyAuditEntryDocument
|
||||
{
|
||||
public required string TenantId { get; init; }
|
||||
public required string Action { get; init; }
|
||||
public string? Actor { get; init; }
|
||||
public string? EntityId { get; init; }
|
||||
public string? EntityType { get; init; }
|
||||
public string? CorrelationId { get; init; }
|
||||
public JsonObject? Payload { get; init; }
|
||||
public DateTimeOffset Timestamp { get; init; }
|
||||
}
|
||||
|
||||
public sealed class NotifyDigestDocument
|
||||
{
|
||||
public required string TenantId { get; init; }
|
||||
public required string ActionKey { get; init; }
|
||||
public string? Content { get; init; }
|
||||
public DateTimeOffset CreatedAt { get; init; } = DateTimeOffset.UtcNow;
|
||||
}
|
||||
|
||||
public sealed class PackApprovalDocument
|
||||
{
|
||||
public required string TenantId { get; init; }
|
||||
public required Guid EventId { get; init; }
|
||||
public required string PackId { get; init; }
|
||||
public string? Kind { get; init; }
|
||||
public string? Decision { get; init; }
|
||||
public string? Actor { get; init; }
|
||||
public DateTimeOffset? IssuedAt { get; init; }
|
||||
public string? PolicyId { get; init; }
|
||||
public string? PolicyVersion { get; init; }
|
||||
public string? ResumeToken { get; init; }
|
||||
public string? Summary { get; init; }
|
||||
public IDictionary<string, string>? Labels { get; init; }
|
||||
public DateTimeOffset CreatedAt { get; init; } = DateTimeOffset.UtcNow;
|
||||
}
|
||||
|
||||
public sealed class NotifyInboxMessage
|
||||
{
|
||||
public required string MessageId { get; init; }
|
||||
public required string TenantId { get; init; }
|
||||
public required string UserId { get; init; }
|
||||
public required string Title { get; init; }
|
||||
public required string Body { get; init; }
|
||||
public string? Summary { get; init; }
|
||||
public string? Category { get; init; }
|
||||
public int Priority { get; init; }
|
||||
public IDictionary<string, string>? Metadata { get; init; }
|
||||
public DateTimeOffset CreatedAt { get; init; }
|
||||
public DateTimeOffset? ExpiresAt { get; init; }
|
||||
public DateTimeOffset? ReadAt { get; set; }
|
||||
public string? SourceChannel { get; init; }
|
||||
public string? DeliveryId { get; init; }
|
||||
}
|
||||
|
||||
namespace StellaOps.Notify.Storage.Mongo.Repositories;
|
||||
|
||||
public interface INotifyMongoInitializer
|
||||
{
|
||||
Task EnsureIndexesAsync(CancellationToken cancellationToken = default);
|
||||
}
|
||||
|
||||
public interface INotifyMongoMigration { }
|
||||
|
||||
public interface INotifyMongoMigrationRunner { }
|
||||
|
||||
public interface INotifyRuleRepository
|
||||
{
|
||||
Task UpsertAsync(NotifyRule rule, CancellationToken cancellationToken = default);
|
||||
Task<NotifyRule?> GetAsync(string tenantId, string ruleId, CancellationToken cancellationToken = default);
|
||||
Task<IReadOnlyList<NotifyRule>> ListAsync(string tenantId, CancellationToken cancellationToken = default);
|
||||
Task DeleteAsync(string tenantId, string ruleId, CancellationToken cancellationToken = default);
|
||||
}
|
||||
|
||||
public interface INotifyChannelRepository
|
||||
{
|
||||
Task UpsertAsync(NotifyChannel channel, CancellationToken cancellationToken = default);
|
||||
Task<NotifyChannel?> GetAsync(string tenantId, string channelId, CancellationToken cancellationToken = default);
|
||||
Task<IReadOnlyList<NotifyChannel>> ListAsync(string tenantId, CancellationToken cancellationToken = default);
|
||||
Task DeleteAsync(string tenantId, string channelId, CancellationToken cancellationToken = default);
|
||||
}
|
||||
|
||||
public interface INotifyTemplateRepository
|
||||
{
|
||||
Task UpsertAsync(NotifyTemplate template, CancellationToken cancellationToken = default);
|
||||
Task<NotifyTemplate?> GetAsync(string tenantId, string templateId, CancellationToken cancellationToken = default);
|
||||
Task<IReadOnlyList<NotifyTemplate>> ListAsync(string tenantId, CancellationToken cancellationToken = default);
|
||||
Task DeleteAsync(string tenantId, string templateId, CancellationToken cancellationToken = default);
|
||||
}
|
||||
|
||||
public interface INotifyDeliveryRepository
|
||||
{
|
||||
Task AppendAsync(NotifyDelivery delivery, CancellationToken cancellationToken = default);
|
||||
Task UpdateAsync(NotifyDelivery delivery, CancellationToken cancellationToken = default);
|
||||
Task<NotifyDelivery?> GetAsync(string tenantId, string deliveryId, CancellationToken cancellationToken = default);
|
||||
Task<NotifyDeliveryQueryResult> QueryAsync(
|
||||
string tenantId,
|
||||
DateTimeOffset? since,
|
||||
string? status,
|
||||
int? limit,
|
||||
string? continuationToken = null,
|
||||
CancellationToken cancellationToken = default);
|
||||
}
|
||||
|
||||
public sealed record NotifyDeliveryQueryResult(IReadOnlyList<NotifyDelivery> Items, string? ContinuationToken);
|
||||
|
||||
public interface INotifyDigestRepository
|
||||
{
|
||||
Task<NotifyDigestDocument?> GetAsync(string tenantId, string actionKey, CancellationToken cancellationToken = default);
|
||||
Task UpsertAsync(NotifyDigestDocument document, CancellationToken cancellationToken = default);
|
||||
Task RemoveAsync(string tenantId, string actionKey, CancellationToken cancellationToken = default);
|
||||
}
|
||||
|
||||
public interface INotifyLockRepository
|
||||
{
|
||||
Task<bool> TryAcquireAsync(string tenantId, string resource, string owner, TimeSpan ttl, CancellationToken cancellationToken = default);
|
||||
Task ReleaseAsync(string tenantId, string resource, string owner, CancellationToken cancellationToken = default);
|
||||
}
|
||||
|
||||
public interface INotifyAuditRepository
|
||||
{
|
||||
Task AppendAsync(NotifyAuditEntryDocument entry, CancellationToken cancellationToken = default);
|
||||
Task AppendAsync(string tenantId, string action, IReadOnlyDictionary<string, string> payload, string? actor = null, CancellationToken cancellationToken = default);
|
||||
Task<IReadOnlyList<NotifyAuditEntryDocument>> QueryAsync(string tenantId, DateTimeOffset? since, int? limit, CancellationToken cancellationToken = default);
|
||||
}
|
||||
|
||||
public interface INotifyPackApprovalRepository
|
||||
{
|
||||
Task UpsertAsync(PackApprovalDocument document, CancellationToken cancellationToken = default);
|
||||
bool Exists(string tenantId, Guid eventId, string packId);
|
||||
}
|
||||
|
||||
public interface INotifyQuietHoursRepository
|
||||
{
|
||||
Task<IReadOnlyList<NotifyQuietHoursSchedule>> ListEnabledAsync(string tenantId, string? channelId = null, CancellationToken cancellationToken = default);
|
||||
}
|
||||
|
||||
public interface INotifyMaintenanceWindowRepository
|
||||
{
|
||||
Task<IReadOnlyList<NotifyMaintenanceWindow>> GetActiveAsync(string tenantId, DateTimeOffset timestamp, CancellationToken cancellationToken = default);
|
||||
}
|
||||
|
||||
public interface INotifyOperatorOverrideRepository
|
||||
{
|
||||
Task<IReadOnlyList<NotifyOperatorOverride>> ListActiveAsync(
|
||||
string tenantId,
|
||||
DateTimeOffset asOf,
|
||||
NotifyOverrideType? type = null,
|
||||
string? channelId = null,
|
||||
CancellationToken cancellationToken = default);
|
||||
}
|
||||
|
||||
public interface INotifyThrottleConfigRepository
|
||||
{
|
||||
Task<IReadOnlyList<NotifyThrottleConfig>> ListAsync(string tenantId, CancellationToken cancellationToken = default);
|
||||
Task<NotifyThrottleConfig?> GetAsync(string tenantId, string configId, CancellationToken cancellationToken = default);
|
||||
Task UpsertAsync(NotifyThrottleConfig config, CancellationToken cancellationToken = default);
|
||||
Task DeleteAsync(string tenantId, string configId, CancellationToken cancellationToken = default);
|
||||
}
|
||||
|
||||
public interface INotifyLocalizationRepository
|
||||
{
|
||||
Task<NotifyLocalizationBundle?> GetByKeyAndLocaleAsync(string tenantId, string bundleKey, string locale, CancellationToken cancellationToken = default);
|
||||
Task<NotifyLocalizationBundle?> GetDefaultAsync(string tenantId, string bundleKey, CancellationToken cancellationToken = default);
|
||||
}
|
||||
|
||||
public interface INotifyEscalationPolicyRepository
|
||||
{
|
||||
Task<IReadOnlyList<NotifyEscalationPolicy>> ListAsync(string tenantId, bool? enabled = null, CancellationToken cancellationToken = default);
|
||||
Task<NotifyEscalationPolicy?> GetAsync(string tenantId, string policyId, CancellationToken cancellationToken = default);
|
||||
Task UpsertAsync(NotifyEscalationPolicy policy, CancellationToken cancellationToken = default);
|
||||
Task DeleteAsync(string tenantId, string policyId, CancellationToken cancellationToken = default);
|
||||
}
|
||||
|
||||
public interface INotifyEscalationStateRepository
|
||||
{
|
||||
Task<NotifyEscalationState?> GetAsync(string tenantId, string stateId, CancellationToken cancellationToken = default);
|
||||
Task<NotifyEscalationState?> GetByIncidentAsync(string tenantId, string incidentId, CancellationToken cancellationToken = default);
|
||||
Task<IReadOnlyList<NotifyEscalationState>> ListDueForEscalationAsync(string tenantId, DateTimeOffset asOf, int batchSize, CancellationToken cancellationToken = default);
|
||||
Task UpsertAsync(NotifyEscalationState state, CancellationToken cancellationToken = default);
|
||||
Task AcknowledgeAsync(string tenantId, string stateId, string acknowledgedBy, DateTimeOffset acknowledgedAt, CancellationToken cancellationToken = default);
|
||||
Task ResolveAsync(string tenantId, string stateId, string resolvedBy, DateTimeOffset resolvedAt, CancellationToken cancellationToken = default);
|
||||
Task DeleteAsync(string tenantId, string stateId, CancellationToken cancellationToken = default);
|
||||
}
|
||||
|
||||
public interface INotifyOnCallScheduleRepository
|
||||
{
|
||||
Task<IReadOnlyList<NotifyOnCallSchedule>> ListAsync(string tenantId, CancellationToken cancellationToken = default);
|
||||
Task<NotifyOnCallSchedule?> GetAsync(string tenantId, string scheduleId, CancellationToken cancellationToken = default);
|
||||
Task UpsertAsync(NotifyOnCallSchedule schedule, CancellationToken cancellationToken = default);
|
||||
Task DeleteAsync(string tenantId, string scheduleId, CancellationToken cancellationToken = default);
|
||||
}
|
||||
|
||||
public interface INotifyInboxRepository
|
||||
{
|
||||
Task StoreAsync(NotifyInboxMessage message, CancellationToken cancellationToken = default);
|
||||
Task<IReadOnlyList<NotifyInboxMessage>> GetForUserAsync(string tenantId, string userId, int limit = 50, CancellationToken cancellationToken = default);
|
||||
Task<NotifyInboxMessage?> GetAsync(string tenantId, string messageId, CancellationToken cancellationToken = default);
|
||||
Task MarkReadAsync(string tenantId, string messageId, CancellationToken cancellationToken = default);
|
||||
Task MarkAllReadAsync(string tenantId, string userId, CancellationToken cancellationToken = default);
|
||||
Task DeleteAsync(string tenantId, string messageId, CancellationToken cancellationToken = default);
|
||||
Task<int> GetUnreadCountAsync(string tenantId, string userId, CancellationToken cancellationToken = default);
|
||||
}
|
||||
|
||||
internal sealed class InMemoryRuleRepository : INotifyRuleRepository
|
||||
{
|
||||
private readonly ConcurrentDictionary<string, ConcurrentDictionary<string, NotifyRule>> _rules = new(StringComparer.Ordinal);
|
||||
|
||||
public Task UpsertAsync(NotifyRule rule, CancellationToken cancellationToken = default)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(rule);
|
||||
var tenantRules = _rules.GetOrAdd(rule.TenantId, _ => new ConcurrentDictionary<string, NotifyRule>(StringComparer.Ordinal));
|
||||
tenantRules[rule.RuleId] = rule;
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public Task<NotifyRule?> GetAsync(string tenantId, string ruleId, CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (_rules.TryGetValue(tenantId, out var rules) && rules.TryGetValue(ruleId, out var rule))
|
||||
{
|
||||
return Task.FromResult<NotifyRule?>(rule);
|
||||
}
|
||||
|
||||
return Task.FromResult<NotifyRule?>(null);
|
||||
}
|
||||
|
||||
public Task<IReadOnlyList<NotifyRule>> ListAsync(string tenantId, CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (_rules.TryGetValue(tenantId, out var rules))
|
||||
{
|
||||
return Task.FromResult<IReadOnlyList<NotifyRule>>(rules.Values.ToArray());
|
||||
}
|
||||
|
||||
return Task.FromResult<IReadOnlyList<NotifyRule>>(Array.Empty<NotifyRule>());
|
||||
}
|
||||
|
||||
public Task DeleteAsync(string tenantId, string ruleId, CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (_rules.TryGetValue(tenantId, out var rules))
|
||||
{
|
||||
rules.TryRemove(ruleId, out _);
|
||||
}
|
||||
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
}
|
||||
|
||||
internal sealed class InMemoryChannelRepository : INotifyChannelRepository
|
||||
{
|
||||
private readonly ConcurrentDictionary<string, ConcurrentDictionary<string, NotifyChannel>> _channels = new(StringComparer.Ordinal);
|
||||
|
||||
public Task UpsertAsync(NotifyChannel channel, CancellationToken cancellationToken = default)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(channel);
|
||||
var map = _channels.GetOrAdd(channel.TenantId, _ => new ConcurrentDictionary<string, NotifyChannel>(StringComparer.Ordinal));
|
||||
map[channel.ChannelId] = channel;
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public Task<NotifyChannel?> GetAsync(string tenantId, string channelId, CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (_channels.TryGetValue(tenantId, out var map) && map.TryGetValue(channelId, out var channel))
|
||||
{
|
||||
return Task.FromResult<NotifyChannel?>(channel);
|
||||
}
|
||||
|
||||
return Task.FromResult<NotifyChannel?>(null);
|
||||
}
|
||||
|
||||
public Task<IReadOnlyList<NotifyChannel>> ListAsync(string tenantId, CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (_channels.TryGetValue(tenantId, out var map))
|
||||
{
|
||||
return Task.FromResult<IReadOnlyList<NotifyChannel>>(map.Values.ToArray());
|
||||
}
|
||||
|
||||
return Task.FromResult<IReadOnlyList<NotifyChannel>>(Array.Empty<NotifyChannel>());
|
||||
}
|
||||
|
||||
public Task DeleteAsync(string tenantId, string channelId, CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (_channels.TryGetValue(tenantId, out var map))
|
||||
{
|
||||
map.TryRemove(channelId, out _);
|
||||
}
|
||||
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
}
|
||||
|
||||
internal sealed class InMemoryTemplateRepository : INotifyTemplateRepository
|
||||
{
|
||||
private readonly ConcurrentDictionary<(string TenantId, string TemplateId), NotifyTemplate> _templates = new();
|
||||
|
||||
public Task UpsertAsync(NotifyTemplate template, CancellationToken cancellationToken = default)
|
||||
{
|
||||
_templates[(template.TenantId, template.TemplateId)] = template;
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public Task<NotifyTemplate?> GetAsync(string tenantId, string templateId, CancellationToken cancellationToken = default)
|
||||
{
|
||||
_templates.TryGetValue((tenantId, templateId), out var tpl);
|
||||
return Task.FromResult(tpl);
|
||||
}
|
||||
|
||||
public Task<IReadOnlyList<NotifyTemplate>> ListAsync(string tenantId, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var list = _templates.Where(kv => kv.Key.TenantId == tenantId).Select(kv => kv.Value).ToList();
|
||||
return Task.FromResult<IReadOnlyList<NotifyTemplate>>(list);
|
||||
}
|
||||
|
||||
public Task DeleteAsync(string tenantId, string templateId, CancellationToken cancellationToken = default)
|
||||
{
|
||||
_templates.TryRemove((tenantId, templateId), out _);
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
}
|
||||
|
||||
internal sealed class InMemoryDeliveryRepository : INotifyDeliveryRepository
|
||||
{
|
||||
private readonly ConcurrentDictionary<string, List<NotifyDelivery>> _deliveries = new(StringComparer.Ordinal);
|
||||
|
||||
public Task AppendAsync(NotifyDelivery delivery, CancellationToken cancellationToken = default)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(delivery);
|
||||
var list = _deliveries.GetOrAdd(delivery.TenantId, _ => new List<NotifyDelivery>());
|
||||
lock (list)
|
||||
{
|
||||
list.Add(delivery);
|
||||
}
|
||||
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public Task UpdateAsync(NotifyDelivery delivery, CancellationToken cancellationToken = default)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(delivery);
|
||||
var list = _deliveries.GetOrAdd(delivery.TenantId, _ => new List<NotifyDelivery>());
|
||||
lock (list)
|
||||
{
|
||||
var index = list.FindIndex(existing => existing.DeliveryId == delivery.DeliveryId);
|
||||
if (index >= 0)
|
||||
{
|
||||
list[index] = delivery;
|
||||
}
|
||||
else
|
||||
{
|
||||
list.Add(delivery);
|
||||
}
|
||||
}
|
||||
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public Task<NotifyDelivery?> GetAsync(string tenantId, string deliveryId, CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (_deliveries.TryGetValue(tenantId, out var list))
|
||||
{
|
||||
lock (list)
|
||||
{
|
||||
return Task.FromResult<NotifyDelivery?>(list.FirstOrDefault(delivery => delivery.DeliveryId == deliveryId));
|
||||
}
|
||||
}
|
||||
|
||||
return Task.FromResult<NotifyDelivery?>(null);
|
||||
}
|
||||
|
||||
public Task<NotifyDeliveryQueryResult> QueryAsync(
|
||||
string tenantId,
|
||||
DateTimeOffset? since,
|
||||
string? status,
|
||||
int? limit,
|
||||
string? continuationToken = null,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (_deliveries.TryGetValue(tenantId, out var list))
|
||||
{
|
||||
lock (list)
|
||||
{
|
||||
var items = list
|
||||
.Where(d => (!since.HasValue || d.CreatedAt >= since) &&
|
||||
(string.IsNullOrWhiteSpace(status) || string.Equals(d.Status.ToString(), status, StringComparison.OrdinalIgnoreCase)))
|
||||
.OrderByDescending(d => d.CreatedAt)
|
||||
.Take(limit ?? 50)
|
||||
.ToArray();
|
||||
|
||||
return Task.FromResult(new NotifyDeliveryQueryResult(items, null));
|
||||
}
|
||||
}
|
||||
|
||||
return Task.FromResult(new NotifyDeliveryQueryResult(Array.Empty<NotifyDelivery>(), null));
|
||||
}
|
||||
}
|
||||
|
||||
internal sealed class InMemoryDigestRepository : INotifyDigestRepository
|
||||
{
|
||||
private readonly ConcurrentDictionary<(string TenantId, string ActionKey), NotifyDigestDocument> _digests = new();
|
||||
|
||||
public Task<NotifyDigestDocument?> GetAsync(string tenantId, string actionKey, CancellationToken cancellationToken = default)
|
||||
{
|
||||
_digests.TryGetValue((tenantId, actionKey), out var doc);
|
||||
return Task.FromResult(doc);
|
||||
}
|
||||
|
||||
public Task UpsertAsync(NotifyDigestDocument document, CancellationToken cancellationToken = default)
|
||||
{
|
||||
_digests[(document.TenantId, document.ActionKey)] = document;
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public Task RemoveAsync(string tenantId, string actionKey, CancellationToken cancellationToken = default)
|
||||
{
|
||||
_digests.TryRemove((tenantId, actionKey), out _);
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
}
|
||||
|
||||
internal sealed class InMemoryLockRepository : INotifyLockRepository
|
||||
{
|
||||
private readonly object _sync = new();
|
||||
private readonly Dictionary<(string TenantId, string Resource), (string Owner, DateTimeOffset Expiry)> _locks = new();
|
||||
|
||||
public Task<bool> TryAcquireAsync(string tenantId, string resource, string owner, TimeSpan ttl, CancellationToken cancellationToken = default)
|
||||
{
|
||||
ArgumentException.ThrowIfNullOrWhiteSpace(tenantId);
|
||||
ArgumentException.ThrowIfNullOrWhiteSpace(resource);
|
||||
ArgumentException.ThrowIfNullOrWhiteSpace(owner);
|
||||
|
||||
lock (_sync)
|
||||
{
|
||||
var key = (tenantId, resource);
|
||||
var now = DateTimeOffset.UtcNow;
|
||||
if (_locks.TryGetValue(key, out var existing) && existing.Expiry > now)
|
||||
{
|
||||
return Task.FromResult(false);
|
||||
}
|
||||
|
||||
_locks[key] = (owner, now + ttl);
|
||||
return Task.FromResult(true);
|
||||
}
|
||||
}
|
||||
|
||||
public Task ReleaseAsync(string tenantId, string resource, string owner, CancellationToken cancellationToken = default)
|
||||
{
|
||||
lock (_sync)
|
||||
{
|
||||
var key = (tenantId, resource);
|
||||
_locks.Remove(key);
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
internal sealed class InMemoryAuditRepository : INotifyAuditRepository
|
||||
{
|
||||
private readonly ConcurrentDictionary<string, List<NotifyAuditEntryDocument>> _entries = new(StringComparer.Ordinal);
|
||||
|
||||
public Task AppendAsync(NotifyAuditEntryDocument entry, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var list = _entries.GetOrAdd(entry.TenantId, _ => new List<NotifyAuditEntryDocument>());
|
||||
lock (list)
|
||||
{
|
||||
list.Add(entry);
|
||||
}
|
||||
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public Task AppendAsync(string tenantId, string action, IReadOnlyDictionary<string, string> payload, string? actor = null, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var entry = new NotifyAuditEntryDocument
|
||||
{
|
||||
TenantId = tenantId,
|
||||
Action = action,
|
||||
Actor = actor,
|
||||
EntityType = "audit",
|
||||
Timestamp = DateTimeOffset.UtcNow,
|
||||
Payload = JsonSerializer.SerializeToNode(payload) as JsonObject
|
||||
};
|
||||
|
||||
return AppendAsync(entry, cancellationToken);
|
||||
}
|
||||
|
||||
public Task<IReadOnlyList<NotifyAuditEntryDocument>> QueryAsync(string tenantId, DateTimeOffset? since, int? limit, CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (_entries.TryGetValue(tenantId, out var list))
|
||||
{
|
||||
lock (list)
|
||||
{
|
||||
var items = list
|
||||
.Where(e => !since.HasValue || e.Timestamp >= since.Value)
|
||||
.OrderByDescending(e => e.Timestamp)
|
||||
.ToList();
|
||||
|
||||
if (limit is > 0)
|
||||
{
|
||||
items = items.Take(limit.Value).ToList();
|
||||
}
|
||||
|
||||
return Task.FromResult<IReadOnlyList<NotifyAuditEntryDocument>>(items);
|
||||
}
|
||||
}
|
||||
|
||||
return Task.FromResult<IReadOnlyList<NotifyAuditEntryDocument>>(Array.Empty<NotifyAuditEntryDocument>());
|
||||
}
|
||||
}
|
||||
|
||||
internal sealed class InMemoryPackApprovalRepository : INotifyPackApprovalRepository
|
||||
{
|
||||
private readonly ConcurrentDictionary<(string TenantId, Guid EventId, string PackId), PackApprovalDocument> _records = new();
|
||||
|
||||
public Task UpsertAsync(PackApprovalDocument document, CancellationToken cancellationToken = default)
|
||||
{
|
||||
_records[(document.TenantId, document.EventId, document.PackId)] = document;
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public bool Exists(string tenantId, Guid eventId, string packId)
|
||||
=> _records.ContainsKey((tenantId, eventId, packId));
|
||||
}
|
||||
|
||||
internal sealed class InMemoryQuietHoursRepository : INotifyQuietHoursRepository
|
||||
{
|
||||
private readonly ConcurrentDictionary<string, List<NotifyQuietHoursSchedule>> _schedules = new(StringComparer.Ordinal);
|
||||
|
||||
public Task<IReadOnlyList<NotifyQuietHoursSchedule>> ListEnabledAsync(string tenantId, string? channelId = null, CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (_schedules.TryGetValue(tenantId, out var list))
|
||||
{
|
||||
var filtered = list
|
||||
.Where(s => s.Enabled)
|
||||
.Where(s => channelId is null || s.ChannelId is null || s.ChannelId == channelId)
|
||||
.ToList();
|
||||
return Task.FromResult<IReadOnlyList<NotifyQuietHoursSchedule>>(filtered);
|
||||
}
|
||||
|
||||
return Task.FromResult<IReadOnlyList<NotifyQuietHoursSchedule>>(Array.Empty<NotifyQuietHoursSchedule>());
|
||||
}
|
||||
|
||||
public void Seed(string tenantId, params NotifyQuietHoursSchedule[] schedules)
|
||||
{
|
||||
var list = _schedules.GetOrAdd(tenantId, _ => new List<NotifyQuietHoursSchedule>());
|
||||
lock (list)
|
||||
{
|
||||
list.AddRange(schedules);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
internal sealed class InMemoryMaintenanceWindowRepository : INotifyMaintenanceWindowRepository
|
||||
{
|
||||
private readonly ConcurrentDictionary<string, List<NotifyMaintenanceWindow>> _windows = new(StringComparer.Ordinal);
|
||||
|
||||
public Task<IReadOnlyList<NotifyMaintenanceWindow>> GetActiveAsync(string tenantId, DateTimeOffset timestamp, CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (_windows.TryGetValue(tenantId, out var list))
|
||||
{
|
||||
var active = list.Where(w => w.IsActiveAt(timestamp)).ToList();
|
||||
return Task.FromResult<IReadOnlyList<NotifyMaintenanceWindow>>(active);
|
||||
}
|
||||
|
||||
return Task.FromResult<IReadOnlyList<NotifyMaintenanceWindow>>(Array.Empty<NotifyMaintenanceWindow>());
|
||||
}
|
||||
|
||||
public void Seed(string tenantId, params NotifyMaintenanceWindow[] windows)
|
||||
{
|
||||
var list = _windows.GetOrAdd(tenantId, _ => new List<NotifyMaintenanceWindow>());
|
||||
lock (list)
|
||||
{
|
||||
list.AddRange(windows);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
internal sealed class InMemoryOperatorOverrideRepository : INotifyOperatorOverrideRepository
|
||||
{
|
||||
private readonly ConcurrentDictionary<string, List<NotifyOperatorOverride>> _overrides = new(StringComparer.Ordinal);
|
||||
|
||||
public Task<IReadOnlyList<NotifyOperatorOverride>> ListActiveAsync(
|
||||
string tenantId,
|
||||
DateTimeOffset asOf,
|
||||
NotifyOverrideType? type = null,
|
||||
string? channelId = null,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (_overrides.TryGetValue(tenantId, out var list))
|
||||
{
|
||||
var items = list
|
||||
.Where(o => o.IsActiveAt(asOf))
|
||||
.Where(o => type is null || o.Type == type)
|
||||
.Where(o => channelId is null || o.ChannelId is null || o.ChannelId == channelId)
|
||||
.ToList();
|
||||
return Task.FromResult<IReadOnlyList<NotifyOperatorOverride>>(items);
|
||||
}
|
||||
|
||||
return Task.FromResult<IReadOnlyList<NotifyOperatorOverride>>(Array.Empty<NotifyOperatorOverride>());
|
||||
}
|
||||
|
||||
public void Seed(string tenantId, params NotifyOperatorOverride[] overrides)
|
||||
{
|
||||
var list = _overrides.GetOrAdd(tenantId, _ => new List<NotifyOperatorOverride>());
|
||||
lock (list)
|
||||
{
|
||||
list.AddRange(overrides);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
internal sealed class InMemoryThrottleConfigRepository : INotifyThrottleConfigRepository
|
||||
{
|
||||
private readonly ConcurrentDictionary<(string TenantId, string ConfigId), NotifyThrottleConfig> _configs = new();
|
||||
|
||||
public Task<IReadOnlyList<NotifyThrottleConfig>> ListAsync(string tenantId, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var list = _configs
|
||||
.Where(kv => kv.Key.TenantId == tenantId)
|
||||
.Select(kv => kv.Value)
|
||||
.ToList();
|
||||
return Task.FromResult<IReadOnlyList<NotifyThrottleConfig>>(list);
|
||||
}
|
||||
|
||||
public Task<NotifyThrottleConfig?> GetAsync(string tenantId, string configId, CancellationToken cancellationToken = default)
|
||||
{
|
||||
_configs.TryGetValue((tenantId, configId), out var cfg);
|
||||
return Task.FromResult(cfg);
|
||||
}
|
||||
|
||||
public Task UpsertAsync(NotifyThrottleConfig config, CancellationToken cancellationToken = default)
|
||||
{
|
||||
_configs[(config.TenantId, config.ConfigId)] = config;
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public Task DeleteAsync(string tenantId, string configId, CancellationToken cancellationToken = default)
|
||||
{
|
||||
_configs.TryRemove((tenantId, configId), out _);
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
}
|
||||
|
||||
internal sealed class InMemoryLocalizationRepository : INotifyLocalizationRepository
|
||||
{
|
||||
private readonly ConcurrentDictionary<(string TenantId, string BundleKey, string Locale), NotifyLocalizationBundle> _bundles = new();
|
||||
|
||||
public Task<NotifyLocalizationBundle?> GetByKeyAndLocaleAsync(string tenantId, string bundleKey, string locale, CancellationToken cancellationToken = default)
|
||||
{
|
||||
_bundles.TryGetValue((tenantId, bundleKey, locale), out var bundle);
|
||||
return Task.FromResult(bundle);
|
||||
}
|
||||
|
||||
public Task<NotifyLocalizationBundle?> GetDefaultAsync(string tenantId, string bundleKey, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var match = _bundles.FirstOrDefault(kv => kv.Key.TenantId == tenantId && kv.Key.BundleKey == bundleKey);
|
||||
return Task.FromResult(match.Value);
|
||||
}
|
||||
}
|
||||
|
||||
internal sealed class InMemoryEscalationPolicyRepository : INotifyEscalationPolicyRepository
|
||||
{
|
||||
private readonly ConcurrentDictionary<(string TenantId, string PolicyId), NotifyEscalationPolicy> _policies = new();
|
||||
|
||||
public Task<IReadOnlyList<NotifyEscalationPolicy>> ListAsync(string tenantId, bool? enabled = null, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var list = _policies
|
||||
.Where(kv => kv.Key.TenantId == tenantId)
|
||||
.Select(kv => kv.Value)
|
||||
.Where(p => !enabled.HasValue || p.Enabled == enabled.Value)
|
||||
.ToList();
|
||||
return Task.FromResult<IReadOnlyList<NotifyEscalationPolicy>>(list);
|
||||
}
|
||||
|
||||
public Task<NotifyEscalationPolicy?> GetAsync(string tenantId, string policyId, CancellationToken cancellationToken = default)
|
||||
{
|
||||
_policies.TryGetValue((tenantId, policyId), out var policy);
|
||||
return Task.FromResult(policy);
|
||||
}
|
||||
|
||||
public Task UpsertAsync(NotifyEscalationPolicy policy, CancellationToken cancellationToken = default)
|
||||
{
|
||||
_policies[(policy.TenantId, policy.PolicyId)] = policy;
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public Task DeleteAsync(string tenantId, string policyId, CancellationToken cancellationToken = default)
|
||||
{
|
||||
_policies.TryRemove((tenantId, policyId), out _);
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
}
|
||||
|
||||
internal sealed class InMemoryEscalationStateRepository : INotifyEscalationStateRepository
|
||||
{
|
||||
private readonly ConcurrentDictionary<(string TenantId, string StateId), NotifyEscalationState> _states = new();
|
||||
|
||||
public Task<NotifyEscalationState?> GetAsync(string tenantId, string stateId, CancellationToken cancellationToken = default)
|
||||
{
|
||||
_states.TryGetValue((tenantId, stateId), out var state);
|
||||
return Task.FromResult(state);
|
||||
}
|
||||
|
||||
public Task<NotifyEscalationState?> GetByIncidentAsync(string tenantId, string incidentId, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var match = _states.FirstOrDefault(kv => kv.Key.TenantId == tenantId && kv.Value.IncidentId == incidentId);
|
||||
return Task.FromResult(match.Value);
|
||||
}
|
||||
|
||||
public Task<IReadOnlyList<NotifyEscalationState>> ListDueForEscalationAsync(string tenantId, DateTimeOffset asOf, int batchSize, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var states = _states
|
||||
.Where(kv => kv.Key.TenantId == tenantId && kv.Value.Status == NotifyEscalationStatus.Active)
|
||||
.Where(kv => kv.Value.NextEscalationAt is null || kv.Value.NextEscalationAt <= asOf)
|
||||
.Select(kv => kv.Value)
|
||||
.Take(batchSize)
|
||||
.ToList();
|
||||
return Task.FromResult<IReadOnlyList<NotifyEscalationState>>(states);
|
||||
}
|
||||
|
||||
public Task UpsertAsync(NotifyEscalationState state, CancellationToken cancellationToken = default)
|
||||
{
|
||||
_states[(state.TenantId, state.StateId)] = state;
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public Task AcknowledgeAsync(string tenantId, string stateId, string acknowledgedBy, DateTimeOffset acknowledgedAt, CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (_states.TryGetValue((tenantId, stateId), out var state))
|
||||
{
|
||||
_states[(tenantId, stateId)] = state with
|
||||
{
|
||||
Status = NotifyEscalationStatus.Acknowledged,
|
||||
AcknowledgedAt = acknowledgedAt,
|
||||
AcknowledgedBy = acknowledgedBy
|
||||
};
|
||||
}
|
||||
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public Task ResolveAsync(string tenantId, string stateId, string resolvedBy, DateTimeOffset resolvedAt, CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (_states.TryGetValue((tenantId, stateId), out var state))
|
||||
{
|
||||
_states[(tenantId, stateId)] = state with
|
||||
{
|
||||
Status = NotifyEscalationStatus.Resolved,
|
||||
ResolvedAt = resolvedAt,
|
||||
ResolvedBy = resolvedBy
|
||||
};
|
||||
}
|
||||
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public Task DeleteAsync(string tenantId, string stateId, CancellationToken cancellationToken = default)
|
||||
{
|
||||
_states.TryRemove((tenantId, stateId), out _);
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
}
|
||||
|
||||
internal sealed class InMemoryOnCallScheduleRepository : INotifyOnCallScheduleRepository
|
||||
{
|
||||
private readonly ConcurrentDictionary<(string TenantId, string ScheduleId), NotifyOnCallSchedule> _schedules = new();
|
||||
|
||||
public Task<IReadOnlyList<NotifyOnCallSchedule>> ListAsync(string tenantId, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var list = _schedules.Where(kv => kv.Key.TenantId == tenantId).Select(kv => kv.Value).ToList();
|
||||
return Task.FromResult<IReadOnlyList<NotifyOnCallSchedule>>(list);
|
||||
}
|
||||
|
||||
public Task<NotifyOnCallSchedule?> GetAsync(string tenantId, string scheduleId, CancellationToken cancellationToken = default)
|
||||
{
|
||||
_schedules.TryGetValue((tenantId, scheduleId), out var schedule);
|
||||
return Task.FromResult(schedule);
|
||||
}
|
||||
|
||||
public Task UpsertAsync(NotifyOnCallSchedule schedule, CancellationToken cancellationToken = default)
|
||||
{
|
||||
_schedules[(schedule.TenantId, schedule.ScheduleId)] = schedule;
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public Task DeleteAsync(string tenantId, string scheduleId, CancellationToken cancellationToken = default)
|
||||
{
|
||||
_schedules.TryRemove((tenantId, scheduleId), out _);
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
}
|
||||
|
||||
internal sealed class InMemoryInboxRepository : INotifyInboxRepository
|
||||
{
|
||||
private readonly ConcurrentDictionary<string, List<NotifyInboxMessage>> _messages = new(StringComparer.Ordinal);
|
||||
|
||||
public Task StoreAsync(NotifyInboxMessage message, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var list = _messages.GetOrAdd(message.TenantId, _ => new List<NotifyInboxMessage>());
|
||||
lock (list)
|
||||
{
|
||||
list.Add(message);
|
||||
}
|
||||
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public Task<IReadOnlyList<NotifyInboxMessage>> GetForUserAsync(string tenantId, string userId, int limit = 50, CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (_messages.TryGetValue(tenantId, out var list))
|
||||
{
|
||||
lock (list)
|
||||
{
|
||||
return Task.FromResult<IReadOnlyList<NotifyInboxMessage>>(list
|
||||
.Where(m => m.UserId == userId)
|
||||
.OrderByDescending(m => m.CreatedAt)
|
||||
.Take(limit)
|
||||
.ToList());
|
||||
}
|
||||
}
|
||||
|
||||
return Task.FromResult<IReadOnlyList<NotifyInboxMessage>>(Array.Empty<NotifyInboxMessage>());
|
||||
}
|
||||
|
||||
public Task<NotifyInboxMessage?> GetAsync(string tenantId, string messageId, CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (_messages.TryGetValue(tenantId, out var list))
|
||||
{
|
||||
lock (list)
|
||||
{
|
||||
return Task.FromResult<NotifyInboxMessage?>(list.FirstOrDefault(m => m.MessageId == messageId));
|
||||
}
|
||||
}
|
||||
|
||||
return Task.FromResult<NotifyInboxMessage?>(null);
|
||||
}
|
||||
|
||||
public Task MarkReadAsync(string tenantId, string messageId, CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (_messages.TryGetValue(tenantId, out var list))
|
||||
{
|
||||
lock (list)
|
||||
{
|
||||
var msg = list.FirstOrDefault(m => m.MessageId == messageId);
|
||||
if (msg is not null)
|
||||
{
|
||||
msg.ReadAt = DateTimeOffset.UtcNow;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public Task MarkAllReadAsync(string tenantId, string userId, CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (_messages.TryGetValue(tenantId, out var list))
|
||||
{
|
||||
lock (list)
|
||||
{
|
||||
foreach (var msg in list.Where(m => m.UserId == userId))
|
||||
{
|
||||
msg.ReadAt ??= DateTimeOffset.UtcNow;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public Task DeleteAsync(string tenantId, string messageId, CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (_messages.TryGetValue(tenantId, out var list))
|
||||
{
|
||||
lock (list)
|
||||
{
|
||||
var idx = list.FindIndex(m => m.MessageId == messageId);
|
||||
if (idx >= 0) list.RemoveAt(idx);
|
||||
}
|
||||
}
|
||||
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public Task<int> GetUnreadCountAsync(string tenantId, string userId, CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (_messages.TryGetValue(tenantId, out var list))
|
||||
{
|
||||
lock (list)
|
||||
{
|
||||
return Task.FromResult(list.Count(m => m.UserId == userId && m.ReadAt is null));
|
||||
}
|
||||
}
|
||||
|
||||
return Task.FromResult(0);
|
||||
}
|
||||
}
|
||||
|
||||
namespace StellaOps.Notify.Storage.Mongo.Internal;
|
||||
|
||||
public sealed class NotifyMongoInitializer : INotifyMongoInitializer
|
||||
{
|
||||
public Task EnsureIndexesAsync(CancellationToken cancellationToken = default) => Task.CompletedTask;
|
||||
}
|
||||
|
||||
namespace StellaOps.Notify.Storage.Mongo;
|
||||
|
||||
using Documents;
|
||||
using Internal;
|
||||
using Repositories;
|
||||
|
||||
public static class ServiceCollectionExtensions
|
||||
{
|
||||
public static IServiceCollection AddNotifyMongoStorage(this IServiceCollection services, IConfiguration configuration)
|
||||
{
|
||||
services.TryAddSingleton<INotifyMongoInitializer, NotifyMongoInitializer>();
|
||||
services.TryAddSingleton<INotifyRuleRepository, InMemoryRuleRepository>();
|
||||
services.TryAddSingleton<INotifyChannelRepository, InMemoryChannelRepository>();
|
||||
services.TryAddSingleton<INotifyTemplateRepository, InMemoryTemplateRepository>();
|
||||
services.TryAddSingleton<INotifyDeliveryRepository, InMemoryDeliveryRepository>();
|
||||
services.TryAddSingleton<INotifyDigestRepository, InMemoryDigestRepository>();
|
||||
services.TryAddSingleton<INotifyLockRepository, InMemoryLockRepository>();
|
||||
services.TryAddSingleton<INotifyAuditRepository, InMemoryAuditRepository>();
|
||||
services.TryAddSingleton<INotifyPackApprovalRepository, InMemoryPackApprovalRepository>();
|
||||
services.TryAddSingleton<INotifyQuietHoursRepository, InMemoryQuietHoursRepository>();
|
||||
services.TryAddSingleton<INotifyMaintenanceWindowRepository, InMemoryMaintenanceWindowRepository>();
|
||||
services.TryAddSingleton<INotifyOperatorOverrideRepository, InMemoryOperatorOverrideRepository>();
|
||||
services.TryAddSingleton<INotifyThrottleConfigRepository, InMemoryThrottleConfigRepository>();
|
||||
services.TryAddSingleton<INotifyLocalizationRepository, InMemoryLocalizationRepository>();
|
||||
services.TryAddSingleton<INotifyEscalationPolicyRepository, InMemoryEscalationPolicyRepository>();
|
||||
services.TryAddSingleton<INotifyEscalationStateRepository, InMemoryEscalationStateRepository>();
|
||||
services.TryAddSingleton<INotifyOnCallScheduleRepository, InMemoryOnCallScheduleRepository>();
|
||||
services.TryAddSingleton<INotifyInboxRepository, InMemoryInboxRepository>();
|
||||
|
||||
return services;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,13 @@
|
||||
<?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>
|
||||
</PropertyGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="../StellaOps.Notify.Models/StellaOps.Notify.Models.csproj" />
|
||||
</ItemGroup>
|
||||
</Project>
|
||||
Reference in New Issue
Block a user