feat(galaxyrepo): hierarchy cache + snapshot + refresh service + projector
This commit is contained in:
Joseph Doherty
+19
@@ -0,0 +1,19 @@
|
|||||||
|
using ZB.MOM.WW.GalaxyRepository.Grpc;
|
||||||
|
|
||||||
|
namespace ZB.MOM.WW.GalaxyRepository;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Result of one <see cref="GalaxyBrowseProjector.ProjectChildren"/> call. Holds a
|
||||||
|
/// materialized page of direct children for the requested parent, along with a
|
||||||
|
/// parallel-indexed <see cref="ChildHasChildren"/> hint and the total post-filter
|
||||||
|
/// sibling count for paging.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="Children">The page of direct children, sorted areas-first then by display name.</param>
|
||||||
|
/// <param name="ChildHasChildren">Parallel array indicating whether each child has at least one matching descendant under the same filter set.</param>
|
||||||
|
/// <param name="TotalChildCount">Total matching direct children of the parent (post-filter).</param>
|
||||||
|
/// <param name="FilterSignature">Stable signature of the filter and parent selector, used to bind page tokens.</param>
|
||||||
|
public sealed record GalaxyBrowseChildrenResult(
|
||||||
|
IReadOnlyList<GalaxyObject> Children,
|
||||||
|
IReadOnlyList<bool> ChildHasChildren,
|
||||||
|
int TotalChildCount,
|
||||||
|
string FilterSignature);
|
||||||
@@ -0,0 +1,281 @@
|
|||||||
|
using System.Collections.Concurrent;
|
||||||
|
using System.Runtime.CompilerServices;
|
||||||
|
using System.Security.Cryptography;
|
||||||
|
using System.Text;
|
||||||
|
using Grpc.Core;
|
||||||
|
using ZB.MOM.WW.GalaxyRepository.Grpc;
|
||||||
|
|
||||||
|
namespace ZB.MOM.WW.GalaxyRepository;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Projects one level of children of a parent object out of an immutable
|
||||||
|
/// <see cref="GalaxyHierarchyCacheEntry"/>. Pure and side-effect free. Memoizes the
|
||||||
|
/// filtered child list per cache-entry instance so repeated paging is an O(pageSize)
|
||||||
|
/// slice rather than an O(siblings) filter scan per page. The memo is keyed on the
|
||||||
|
/// immutable cache entry, so when the cache publishes a new entry the stale memo
|
||||||
|
/// becomes unreachable and is reclaimed with it.
|
||||||
|
/// </summary>
|
||||||
|
public static class GalaxyBrowseProjector
|
||||||
|
{
|
||||||
|
private static readonly ConditionalWeakTable<
|
||||||
|
GalaxyHierarchyCacheEntry,
|
||||||
|
ConcurrentDictionary<string, FilteredChildren>> FilteredChildrenCache = new();
|
||||||
|
|
||||||
|
/// <summary>Projects one page of direct children of the resolved parent.</summary>
|
||||||
|
/// <param name="entry">The Galaxy hierarchy cache entry to query.</param>
|
||||||
|
/// <param name="request">The browse-children request.</param>
|
||||||
|
/// <param name="browseSubtreeGlobs">Optional API-key browse-subtree constraints.</param>
|
||||||
|
/// <param name="offset">Zero-based offset into the filtered child list.</param>
|
||||||
|
/// <param name="pageSize">Maximum number of children to return.</param>
|
||||||
|
public static GalaxyBrowseChildrenResult ProjectChildren(
|
||||||
|
GalaxyHierarchyCacheEntry entry,
|
||||||
|
BrowseChildrenRequest request,
|
||||||
|
IReadOnlyList<string>? browseSubtreeGlobs,
|
||||||
|
int offset,
|
||||||
|
int pageSize)
|
||||||
|
{
|
||||||
|
ArgumentNullException.ThrowIfNull(entry);
|
||||||
|
ArgumentNullException.ThrowIfNull(request);
|
||||||
|
if (offset < 0)
|
||||||
|
{
|
||||||
|
throw new ArgumentOutOfRangeException(nameof(offset), offset, "Offset must be greater than or equal to zero.");
|
||||||
|
}
|
||||||
|
if (pageSize <= 0)
|
||||||
|
{
|
||||||
|
throw new ArgumentOutOfRangeException(nameof(pageSize), pageSize, "Page size must be greater than zero.");
|
||||||
|
}
|
||||||
|
|
||||||
|
int parentId = ResolveParentId(entry, request);
|
||||||
|
string filterSignature = ComputeFilterSignature(request, browseSubtreeGlobs, parentId);
|
||||||
|
FilteredChildren filtered = GetFilteredChildren(entry, request, browseSubtreeGlobs, parentId, filterSignature);
|
||||||
|
|
||||||
|
bool includeAttributes = IncludeAttributes(request);
|
||||||
|
int end = (int)Math.Min((long)offset + pageSize, filtered.Children.Count);
|
||||||
|
List<GalaxyObject> page = new(Math.Max(0, end - offset));
|
||||||
|
List<bool> hasChildren = new(Math.Max(0, end - offset));
|
||||||
|
for (int index = offset; index < end; index++)
|
||||||
|
{
|
||||||
|
page.Add(CloneObject(filtered.Children[index].Object, includeAttributes));
|
||||||
|
hasChildren.Add(filtered.HasMatchingDescendant[index]);
|
||||||
|
}
|
||||||
|
|
||||||
|
return new GalaxyBrowseChildrenResult(page, hasChildren, filtered.Children.Count, filterSignature);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Resolves the request's parent oneof to a gobject id, throwing
|
||||||
|
/// <see cref="RpcException"/> with <see cref="StatusCode.NotFound"/> when the
|
||||||
|
/// parent does not exist. Public so the gRPC handler can compute the same
|
||||||
|
/// parent id (needed for the page-token signature) without reimplementing the
|
||||||
|
/// resolution rules.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="entry">The Galaxy hierarchy cache entry to query.</param>
|
||||||
|
/// <param name="request">The browse-children request.</param>
|
||||||
|
public static int ResolveParentId(GalaxyHierarchyCacheEntry entry, BrowseChildrenRequest request)
|
||||||
|
{
|
||||||
|
switch (request.ParentCase)
|
||||||
|
{
|
||||||
|
case BrowseChildrenRequest.ParentOneofCase.None:
|
||||||
|
return 0;
|
||||||
|
case BrowseChildrenRequest.ParentOneofCase.ParentGobjectId:
|
||||||
|
if (request.ParentGobjectId == 0)
|
||||||
|
{
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
if (!entry.Index.ObjectViewsById.ContainsKey(request.ParentGobjectId))
|
||||||
|
{
|
||||||
|
throw new RpcException(new Status(StatusCode.NotFound, "BrowseChildren parent was not found."));
|
||||||
|
}
|
||||||
|
return request.ParentGobjectId;
|
||||||
|
case BrowseChildrenRequest.ParentOneofCase.ParentTagName:
|
||||||
|
{
|
||||||
|
if (!entry.Index.ObjectViewsByTagName.TryGetValue(request.ParentTagName, out GalaxyObjectView? match))
|
||||||
|
{
|
||||||
|
throw new RpcException(new Status(StatusCode.NotFound, "BrowseChildren parent was not found."));
|
||||||
|
}
|
||||||
|
return match.Object.GobjectId;
|
||||||
|
}
|
||||||
|
case BrowseChildrenRequest.ParentOneofCase.ParentContainedPath:
|
||||||
|
{
|
||||||
|
if (!entry.Index.ObjectViewsByContainedPath.TryGetValue(request.ParentContainedPath, out GalaxyObjectView? match))
|
||||||
|
{
|
||||||
|
throw new RpcException(new Status(StatusCode.NotFound, "BrowseChildren parent was not found."));
|
||||||
|
}
|
||||||
|
return match.Object.GobjectId;
|
||||||
|
}
|
||||||
|
default:
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private static FilteredChildren GetFilteredChildren(
|
||||||
|
GalaxyHierarchyCacheEntry entry,
|
||||||
|
BrowseChildrenRequest request,
|
||||||
|
IReadOnlyList<string>? browseSubtreeGlobs,
|
||||||
|
int parentId,
|
||||||
|
string filterSignature)
|
||||||
|
{
|
||||||
|
ConcurrentDictionary<string, FilteredChildren> memo =
|
||||||
|
FilteredChildrenCache.GetValue(entry, static _ => new ConcurrentDictionary<string, FilteredChildren>(StringComparer.Ordinal));
|
||||||
|
|
||||||
|
return memo.GetOrAdd(
|
||||||
|
filterSignature,
|
||||||
|
static (_, state) =>
|
||||||
|
{
|
||||||
|
IReadOnlyDictionary<int, IReadOnlyList<GalaxyObjectView>> map = state.Entry.Index.ChildrenByParent;
|
||||||
|
IReadOnlyList<GalaxyObjectView> directChildren = map.TryGetValue(state.ParentId, out IReadOnlyList<GalaxyObjectView>? list)
|
||||||
|
? list
|
||||||
|
: Array.Empty<GalaxyObjectView>();
|
||||||
|
|
||||||
|
List<GalaxyObjectView> matched = [];
|
||||||
|
List<bool> hasMatching = [];
|
||||||
|
foreach (GalaxyObjectView view in directChildren)
|
||||||
|
{
|
||||||
|
if (!MatchesBrowseSubtrees(view, state.BrowseSubtreeGlobs))
|
||||||
|
{
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
if (!MatchesFilters(view.Object, state.Request))
|
||||||
|
{
|
||||||
|
// Even if the direct child itself fails the filter, a matching
|
||||||
|
// descendant should still surface its ancestor — but only when
|
||||||
|
// there is one. Mirror the dashboard browse-tree semantics: if a
|
||||||
|
// descendant matches, include the parent with has-children true.
|
||||||
|
if (HasMatchingDescendant(view, state.Entry.Index, state.Request, state.BrowseSubtreeGlobs))
|
||||||
|
{
|
||||||
|
matched.Add(view);
|
||||||
|
hasMatching.Add(true);
|
||||||
|
}
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
matched.Add(view);
|
||||||
|
hasMatching.Add(HasMatchingDescendant(view, state.Entry.Index, state.Request, state.BrowseSubtreeGlobs));
|
||||||
|
}
|
||||||
|
|
||||||
|
return new FilteredChildren(matched, hasMatching);
|
||||||
|
},
|
||||||
|
(Entry: entry, ParentId: parentId, Request: request, BrowseSubtreeGlobs: browseSubtreeGlobs));
|
||||||
|
}
|
||||||
|
|
||||||
|
private static bool HasMatchingDescendant(
|
||||||
|
GalaxyObjectView parent,
|
||||||
|
GalaxyHierarchyIndex index,
|
||||||
|
BrowseChildrenRequest request,
|
||||||
|
IReadOnlyList<string>? browseSubtreeGlobs)
|
||||||
|
{
|
||||||
|
if (!index.ChildrenByParent.TryGetValue(parent.Object.GobjectId, out IReadOnlyList<GalaxyObjectView>? children))
|
||||||
|
{
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
|
// Defend against pathological cycles in Galaxy data (e.g. a corrupt A→B→A chain).
|
||||||
|
// BuildContainedPath uses the same visited-id pattern; mirror it so this walk
|
||||||
|
// terminates even when ChildrenByParent forms a cycle.
|
||||||
|
HashSet<int> visited = new() { parent.Object.GobjectId };
|
||||||
|
Stack<GalaxyObjectView> stack = new();
|
||||||
|
foreach (GalaxyObjectView child in children)
|
||||||
|
{
|
||||||
|
if (visited.Add(child.Object.GobjectId))
|
||||||
|
{
|
||||||
|
stack.Push(child);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
while (stack.Count > 0)
|
||||||
|
{
|
||||||
|
GalaxyObjectView candidate = stack.Pop();
|
||||||
|
if (MatchesBrowseSubtrees(candidate, browseSubtreeGlobs)
|
||||||
|
&& MatchesFilters(candidate.Object, request))
|
||||||
|
{
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
if (index.ChildrenByParent.TryGetValue(candidate.Object.GobjectId, out IReadOnlyList<GalaxyObjectView>? grandchildren))
|
||||||
|
{
|
||||||
|
foreach (GalaxyObjectView grandchild in grandchildren)
|
||||||
|
{
|
||||||
|
if (visited.Add(grandchild.Object.GobjectId))
|
||||||
|
{
|
||||||
|
stack.Push(grandchild);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
|
private static bool MatchesBrowseSubtrees(GalaxyObjectView view, IReadOnlyList<string>? browseSubtreeGlobs)
|
||||||
|
{
|
||||||
|
return browseSubtreeGlobs is null
|
||||||
|
|| browseSubtreeGlobs.Count == 0
|
||||||
|
|| browseSubtreeGlobs.Any(glob => GalaxyGlobMatcher.IsMatch(view.ContainedPath, glob));
|
||||||
|
}
|
||||||
|
|
||||||
|
private static bool MatchesFilters(GalaxyObject obj, BrowseChildrenRequest request)
|
||||||
|
{
|
||||||
|
if (request.CategoryIds.Count > 0 && !request.CategoryIds.Contains(obj.CategoryId))
|
||||||
|
{
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
foreach (string templateFilter in request.TemplateChainContains)
|
||||||
|
{
|
||||||
|
if (!obj.TemplateChain.Any(template => template.Contains(templateFilter, StringComparison.OrdinalIgnoreCase)))
|
||||||
|
{
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (!string.IsNullOrWhiteSpace(request.TagNameGlob)
|
||||||
|
&& !GalaxyGlobMatcher.IsMatch(obj.TagName, request.TagNameGlob))
|
||||||
|
{
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
if (request.AlarmBearingOnly && !obj.Attributes.Any(attribute => attribute.IsAlarm))
|
||||||
|
{
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
if (request.HistorizedOnly && !obj.Attributes.Any(attribute => attribute.IsHistorized))
|
||||||
|
{
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
|
||||||
|
private static bool IncludeAttributes(BrowseChildrenRequest request)
|
||||||
|
{
|
||||||
|
return !request.HasIncludeAttributes || request.IncludeAttributes;
|
||||||
|
}
|
||||||
|
|
||||||
|
private static GalaxyObject CloneObject(GalaxyObject source, bool includeAttributes)
|
||||||
|
{
|
||||||
|
GalaxyObject clone = source.Clone();
|
||||||
|
if (!includeAttributes)
|
||||||
|
{
|
||||||
|
clone.Attributes.Clear();
|
||||||
|
}
|
||||||
|
return clone;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>Computes a stable filter signature for memoization purposes.</summary>
|
||||||
|
/// <param name="request">The browse-children request.</param>
|
||||||
|
/// <param name="browseSubtreeGlobs">Optional API-key browse-subtree constraints.</param>
|
||||||
|
/// <param name="parentId">Resolved parent gobject id (0 for roots).</param>
|
||||||
|
public static string ComputeFilterSignature(
|
||||||
|
BrowseChildrenRequest request,
|
||||||
|
IReadOnlyList<string>? browseSubtreeGlobs,
|
||||||
|
int parentId)
|
||||||
|
{
|
||||||
|
StringBuilder builder = new();
|
||||||
|
builder.Append("parent=").Append(parentId.ToString(System.Globalization.CultureInfo.InvariantCulture));
|
||||||
|
builder.Append("|cat=").AppendJoin(',', request.CategoryIds.Order());
|
||||||
|
builder.Append("|tpl=").AppendJoin(',', request.TemplateChainContains.Order(StringComparer.OrdinalIgnoreCase));
|
||||||
|
builder.Append("|glob=").Append(request.TagNameGlob);
|
||||||
|
builder.Append("|attrs=").Append(request.HasIncludeAttributes ? request.IncludeAttributes.ToString() : "unset");
|
||||||
|
builder.Append("|alarm=").Append(request.AlarmBearingOnly);
|
||||||
|
builder.Append("|hist=").Append(request.HistorizedOnly);
|
||||||
|
builder.Append("|browse=").AppendJoin(',', (browseSubtreeGlobs ?? Array.Empty<string>()).Order(StringComparer.OrdinalIgnoreCase));
|
||||||
|
byte[] hash = SHA256.HashData(Encoding.UTF8.GetBytes(builder.ToString()));
|
||||||
|
return Convert.ToHexString(hash, 0, 12);
|
||||||
|
}
|
||||||
|
|
||||||
|
private sealed record FilteredChildren(
|
||||||
|
IReadOnlyList<GalaxyObjectView> Children,
|
||||||
|
IReadOnlyList<bool> HasMatchingDescendant);
|
||||||
|
}
|
||||||
@@ -0,0 +1,18 @@
|
|||||||
|
namespace ZB.MOM.WW.GalaxyRepository;
|
||||||
|
|
||||||
|
/// <summary>Freshness state of the shared Galaxy hierarchy cache entry.</summary>
|
||||||
|
public enum GalaxyCacheStatus
|
||||||
|
{
|
||||||
|
/// <summary>Cache has never completed a refresh.</summary>
|
||||||
|
Unknown = 0,
|
||||||
|
|
||||||
|
/// <summary>Cache holds data from a recent successful refresh.</summary>
|
||||||
|
Healthy = 1,
|
||||||
|
|
||||||
|
/// <summary>Cache holds data, but the most recent refresh attempt failed
|
||||||
|
/// or no successful refresh has happened within the staleness threshold.</summary>
|
||||||
|
Stale = 2,
|
||||||
|
|
||||||
|
/// <summary>Latest refresh failed and no prior data is available.</summary>
|
||||||
|
Unavailable = 3,
|
||||||
|
}
|
||||||
@@ -0,0 +1,19 @@
|
|||||||
|
namespace ZB.MOM.WW.GalaxyRepository;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// A single Galaxy deploy notification. Published by <see cref="GalaxyHierarchyCache"/>
|
||||||
|
/// whenever a refresh detects that <c>galaxy.time_of_last_deploy</c> has changed (or on
|
||||||
|
/// the first successful refresh). Consumed by <see cref="IGalaxyDeployNotifier"/>
|
||||||
|
/// subscribers (the streaming gRPC RPC).
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="Sequence">Monotonically increasing per process start; gaps indicate dropped events.</param>
|
||||||
|
/// <param name="ObservedAt">Server wall-clock when the cache observed the deploy.</param>
|
||||||
|
/// <param name="TimeOfLastDeploy">The <c>galaxy.time_of_last_deploy</c> value, or <see langword="null"/> when the Galaxy table reports none.</param>
|
||||||
|
/// <param name="ObjectCount">Number of objects in the hierarchy at the time of the event.</param>
|
||||||
|
/// <param name="AttributeCount">Number of attributes in the hierarchy at the time of the event.</param>
|
||||||
|
public sealed record GalaxyDeployEventInfo(
|
||||||
|
long Sequence,
|
||||||
|
DateTimeOffset ObservedAt,
|
||||||
|
DateTimeOffset? TimeOfLastDeploy,
|
||||||
|
int ObjectCount,
|
||||||
|
int AttributeCount);
|
||||||
@@ -0,0 +1,79 @@
|
|||||||
|
using System.Collections.Concurrent;
|
||||||
|
using System.Runtime.CompilerServices;
|
||||||
|
using System.Threading.Channels;
|
||||||
|
|
||||||
|
namespace ZB.MOM.WW.GalaxyRepository;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Channel-based fan-out of Galaxy deploy events to streaming gRPC subscribers. Each
|
||||||
|
/// subscriber gets a private bounded channel so a slow client cannot back-pressure
|
||||||
|
/// other subscribers or the publisher. When a subscriber's channel is full the oldest
|
||||||
|
/// event is dropped — clients use the sequence field to detect gaps.
|
||||||
|
/// </summary>
|
||||||
|
public sealed class GalaxyDeployNotifier : IGalaxyDeployNotifier
|
||||||
|
{
|
||||||
|
private const int SubscriberQueueCapacity = 16;
|
||||||
|
|
||||||
|
private readonly ConcurrentDictionary<Guid, Channel<GalaxyDeployEventInfo>> _subscribers = new();
|
||||||
|
private GalaxyDeployEventInfo? _latest;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// The most recent deploy event, or null if none has been published.
|
||||||
|
/// </summary>
|
||||||
|
public GalaxyDeployEventInfo? Latest => Volatile.Read(ref _latest);
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public void Publish(GalaxyDeployEventInfo info)
|
||||||
|
{
|
||||||
|
ArgumentNullException.ThrowIfNull(info);
|
||||||
|
|
||||||
|
Volatile.Write(ref _latest, info);
|
||||||
|
|
||||||
|
foreach (Channel<GalaxyDeployEventInfo> channel in _subscribers.Values)
|
||||||
|
{
|
||||||
|
// BoundedChannelFullMode.DropOldest -> writes never wait; we only fail if the
|
||||||
|
// channel was completed by the subscriber side, which we ignore.
|
||||||
|
channel.Writer.TryWrite(info);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public async IAsyncEnumerable<GalaxyDeployEventInfo> SubscribeAsync(
|
||||||
|
[EnumeratorCancellation] CancellationToken cancellationToken)
|
||||||
|
{
|
||||||
|
Guid subscriberId = Guid.NewGuid();
|
||||||
|
Channel<GalaxyDeployEventInfo> channel = Channel.CreateBounded<GalaxyDeployEventInfo>(
|
||||||
|
new BoundedChannelOptions(SubscriberQueueCapacity)
|
||||||
|
{
|
||||||
|
FullMode = BoundedChannelFullMode.DropOldest,
|
||||||
|
SingleReader = true,
|
||||||
|
SingleWriter = false,
|
||||||
|
});
|
||||||
|
|
||||||
|
_subscribers[subscriberId] = channel;
|
||||||
|
|
||||||
|
// Bootstrap: emit the latest known event so subscribers don't need to wait for
|
||||||
|
// the next deploy to know current state.
|
||||||
|
GalaxyDeployEventInfo? bootstrap = Volatile.Read(ref _latest);
|
||||||
|
if (bootstrap is not null)
|
||||||
|
{
|
||||||
|
channel.Writer.TryWrite(bootstrap);
|
||||||
|
}
|
||||||
|
|
||||||
|
try
|
||||||
|
{
|
||||||
|
while (await channel.Reader.WaitToReadAsync(cancellationToken).ConfigureAwait(false))
|
||||||
|
{
|
||||||
|
while (channel.Reader.TryRead(out GalaxyDeployEventInfo? next))
|
||||||
|
{
|
||||||
|
yield return next;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
finally
|
||||||
|
{
|
||||||
|
_subscribers.TryRemove(subscriberId, out _);
|
||||||
|
channel.Writer.TryComplete();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,131 @@
|
|||||||
|
using System.Collections.Concurrent;
|
||||||
|
using System.Text;
|
||||||
|
using System.Text.RegularExpressions;
|
||||||
|
|
||||||
|
namespace ZB.MOM.WW.GalaxyRepository;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Anchored, case-insensitive glob matcher (<c>*</c> and <c>?</c> wildcards) used by the
|
||||||
|
/// hierarchy and browse projectors to filter object tag names and browse subtrees.
|
||||||
|
/// Compiled regexes are cached and the cache is bounded so an unbounded stream of distinct
|
||||||
|
/// client-supplied globs cannot grow memory without limit.
|
||||||
|
/// </summary>
|
||||||
|
public static class GalaxyGlobMatcher
|
||||||
|
{
|
||||||
|
/// <summary>
|
||||||
|
/// Maximum number of compiled-regex entries retained in <see cref="RegexCache"/>.
|
||||||
|
/// The cache is keyed by glob pattern and patterns flow in from two sources:
|
||||||
|
/// admin-controlled API-key constraints (naturally bounded) and the
|
||||||
|
/// client-supplied <c>DiscoverHierarchyRequest.TagNameGlob</c> (unbounded — a
|
||||||
|
/// client can iterate through generated names and create millions of distinct
|
||||||
|
/// globs over the process lifetime). Capping the cache bounds memory while
|
||||||
|
/// keeping the hot working set hit-cached.
|
||||||
|
/// </summary>
|
||||||
|
internal const int RegexCacheCapacity = 256;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Bounded compiled-regex cache keyed by glob pattern. <c>IsMatch</c> is called
|
||||||
|
/// once per object per <c>DiscoverHierarchy</c>/<c>WatchDeployEvents</c>
|
||||||
|
/// evaluation, so the same handful of glob patterns are translated
|
||||||
|
/// repeatedly; caching avoids rebuilding and recompiling the regex on every
|
||||||
|
/// call. Beyond <see cref="RegexCacheCapacity"/> entries the oldest insertion
|
||||||
|
/// is evicted so a client cannot grow the cache without bound by submitting
|
||||||
|
/// unique patterns. Eviction is approximate (FIFO over insertion order, not
|
||||||
|
/// true LRU) because we only need the bound, not exact recency tracking.
|
||||||
|
/// </summary>
|
||||||
|
private static readonly ConcurrentDictionary<string, Regex> RegexCache = new(StringComparer.Ordinal);
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Insertion-order queue used to evict the oldest cache entry when the cache
|
||||||
|
/// exceeds <see cref="RegexCacheCapacity"/>. A separate queue keeps the
|
||||||
|
/// <see cref="RegexCache"/> reads lock-free; the lock below only guards the
|
||||||
|
/// eviction path.
|
||||||
|
/// </summary>
|
||||||
|
private static readonly ConcurrentQueue<string> InsertionOrder = new();
|
||||||
|
private static readonly object EvictionLock = new();
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Current cache size, exposed for tests asserting the cap is honoured.
|
||||||
|
/// </summary>
|
||||||
|
internal static int CurrentCacheSize => RegexCache.Count;
|
||||||
|
|
||||||
|
/// <summary>Determines whether a value matches a glob pattern (with * and ? wildcards).</summary>
|
||||||
|
/// <param name="value">The value to test against the glob pattern.</param>
|
||||||
|
/// <param name="glob">The glob pattern with * and ? wildcards.</param>
|
||||||
|
public static bool IsMatch(string value, string glob)
|
||||||
|
{
|
||||||
|
if (string.IsNullOrWhiteSpace(glob))
|
||||||
|
{
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
|
||||||
|
return GetOrCreateRegex(glob).IsMatch(value ?? string.Empty);
|
||||||
|
}
|
||||||
|
|
||||||
|
private static Regex GetOrCreateRegex(string glob)
|
||||||
|
{
|
||||||
|
if (RegexCache.TryGetValue(glob, out Regex? existing))
|
||||||
|
{
|
||||||
|
return existing;
|
||||||
|
}
|
||||||
|
|
||||||
|
Regex compiled = new(
|
||||||
|
BuildRegex(glob),
|
||||||
|
RegexOptions.CultureInvariant | RegexOptions.IgnoreCase | RegexOptions.Compiled,
|
||||||
|
TimeSpan.FromMilliseconds(100));
|
||||||
|
|
||||||
|
// GetOrAdd atomically returns whichever instance is in the cache after the
|
||||||
|
// call — either the locally-compiled regex (we won the race) or the regex
|
||||||
|
// another thread inserted (we lost). It also avoids the TryAdd-then-indexer
|
||||||
|
// pattern where the key could be evicted between the failed TryAdd and the
|
||||||
|
// indexer read, producing a KeyNotFoundException under contention near the cap.
|
||||||
|
Regex result = RegexCache.GetOrAdd(glob, compiled);
|
||||||
|
if (ReferenceEquals(result, compiled))
|
||||||
|
{
|
||||||
|
// We were the inserter — track for FIFO eviction and bound the cache.
|
||||||
|
InsertionOrder.Enqueue(glob);
|
||||||
|
EvictIfOverCapacity();
|
||||||
|
}
|
||||||
|
return result;
|
||||||
|
}
|
||||||
|
|
||||||
|
private static void EvictIfOverCapacity()
|
||||||
|
{
|
||||||
|
if (RegexCache.Count <= RegexCacheCapacity)
|
||||||
|
{
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
// Serialize eviction so two threads do not race past the cap together.
|
||||||
|
lock (EvictionLock)
|
||||||
|
{
|
||||||
|
while (RegexCache.Count > RegexCacheCapacity && InsertionOrder.TryDequeue(out string? oldest))
|
||||||
|
{
|
||||||
|
RegexCache.TryRemove(oldest, out _);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private static string BuildRegex(string glob)
|
||||||
|
{
|
||||||
|
StringBuilder builder = new("^", glob.Length + 2);
|
||||||
|
foreach (char character in glob)
|
||||||
|
{
|
||||||
|
switch (character)
|
||||||
|
{
|
||||||
|
case '*':
|
||||||
|
builder.Append(".*");
|
||||||
|
break;
|
||||||
|
case '?':
|
||||||
|
builder.Append('.');
|
||||||
|
break;
|
||||||
|
default:
|
||||||
|
builder.Append(Regex.Escape(character.ToString()));
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
builder.Append('$');
|
||||||
|
return builder.ToString();
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,355 @@
|
|||||||
|
using Microsoft.Extensions.Logging;
|
||||||
|
using ZB.MOM.WW.GalaxyRepository.Grpc;
|
||||||
|
|
||||||
|
namespace ZB.MOM.WW.GalaxyRepository;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Server-side cache of Galaxy Repository browse data. All gRPC clients share the same
|
||||||
|
/// entry — the materialized object list is produced once per refresh and reused across
|
||||||
|
/// requests. Refreshes are deploy-time gated: every tick queries
|
||||||
|
/// <c>galaxy.time_of_last_deploy</c> (cheap), and the heavy hierarchy + attributes rowsets
|
||||||
|
/// are pulled only when that timestamp has advanced.
|
||||||
|
/// Each successful heavy refresh is persisted to disk through
|
||||||
|
/// <see cref="IGalaxyHierarchySnapshotStore"/>; the first refresh restores that
|
||||||
|
/// snapshot (as <see cref="GalaxyCacheStatus.Stale"/>) so clients can browse
|
||||||
|
/// last-known data when the Galaxy database is unreachable on a cold start.
|
||||||
|
/// </summary>
|
||||||
|
public sealed class GalaxyHierarchyCache : IGalaxyHierarchyCache
|
||||||
|
{
|
||||||
|
private static readonly TimeSpan StaleThreshold = TimeSpan.FromMinutes(5);
|
||||||
|
|
||||||
|
private readonly IGalaxyRepository _repository;
|
||||||
|
private readonly IGalaxyDeployNotifier _notifier;
|
||||||
|
private readonly IGalaxyHierarchySnapshotStore? _snapshotStore;
|
||||||
|
private readonly TimeProvider _timeProvider;
|
||||||
|
private readonly ILogger<GalaxyHierarchyCache>? _logger;
|
||||||
|
private readonly TaskCompletionSource _firstLoad = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||||
|
private readonly SemaphoreSlim _refreshGate = new(1, 1);
|
||||||
|
private GalaxyHierarchyCacheEntry _current = GalaxyHierarchyCacheEntry.Empty;
|
||||||
|
private bool _restoreAttempted;
|
||||||
|
|
||||||
|
/// <summary>Initializes a new instance of the <see cref="GalaxyHierarchyCache"/> class.</summary>
|
||||||
|
/// <param name="repository">Galaxy Repository client for SQL queries.</param>
|
||||||
|
/// <param name="notifier">Galaxy deploy event notifier.</param>
|
||||||
|
/// <param name="timeProvider">Provider for current time; defaults to system time.</param>
|
||||||
|
/// <param name="logger">Optional logger for diagnostic output.</param>
|
||||||
|
/// <param name="snapshotStore">
|
||||||
|
/// Optional on-disk snapshot store. When supplied, the cache persists each
|
||||||
|
/// successful refresh and restores the last snapshot on first load.
|
||||||
|
/// </param>
|
||||||
|
public GalaxyHierarchyCache(
|
||||||
|
IGalaxyRepository repository,
|
||||||
|
IGalaxyDeployNotifier notifier,
|
||||||
|
TimeProvider? timeProvider = null,
|
||||||
|
ILogger<GalaxyHierarchyCache>? logger = null,
|
||||||
|
IGalaxyHierarchySnapshotStore? snapshotStore = null)
|
||||||
|
{
|
||||||
|
_repository = repository;
|
||||||
|
_notifier = notifier;
|
||||||
|
_timeProvider = timeProvider ?? TimeProvider.System;
|
||||||
|
_logger = logger;
|
||||||
|
_snapshotStore = snapshotStore;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>Gets the current Galaxy hierarchy cache entry with projected status.</summary>
|
||||||
|
public GalaxyHierarchyCacheEntry Current
|
||||||
|
{
|
||||||
|
get
|
||||||
|
{
|
||||||
|
GalaxyHierarchyCacheEntry snapshot = Volatile.Read(ref _current);
|
||||||
|
GalaxyCacheStatus projected = ProjectStatus(snapshot);
|
||||||
|
return projected == snapshot.Status
|
||||||
|
? snapshot
|
||||||
|
: snapshot with { Status = projected };
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>Refreshes the Galaxy hierarchy cache if the deploy time has advanced.</summary>
|
||||||
|
/// <param name="cancellationToken">Token to cancel the asynchronous operation.</param>
|
||||||
|
/// <returns>Asynchronous task representing the refresh operation.</returns>
|
||||||
|
public async Task RefreshAsync(CancellationToken cancellationToken)
|
||||||
|
{
|
||||||
|
await _refreshGate.WaitAsync(cancellationToken).ConfigureAwait(false);
|
||||||
|
try
|
||||||
|
{
|
||||||
|
await RefreshCoreAsync(cancellationToken).ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
finally
|
||||||
|
{
|
||||||
|
_refreshGate.Release();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>Waits for the Galaxy hierarchy cache to complete its first load.</summary>
|
||||||
|
/// <param name="cancellationToken">Token to cancel the asynchronous operation.</param>
|
||||||
|
/// <returns>Asynchronous task representing the wait operation.</returns>
|
||||||
|
public Task WaitForFirstLoadAsync(CancellationToken cancellationToken)
|
||||||
|
{
|
||||||
|
return _firstLoad.Task.WaitAsync(cancellationToken);
|
||||||
|
}
|
||||||
|
|
||||||
|
private async Task RefreshCoreAsync(CancellationToken cancellationToken)
|
||||||
|
{
|
||||||
|
// First refresh only: seed the cache from the on-disk snapshot before
|
||||||
|
// querying SQL, so a cold start with an unreachable Galaxy database can
|
||||||
|
// still serve last-known browse data. Runs under the refresh gate.
|
||||||
|
if (!_restoreAttempted)
|
||||||
|
{
|
||||||
|
_restoreAttempted = true;
|
||||||
|
await TryRestoreFromDiskAsync(cancellationToken).ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
|
||||||
|
GalaxyHierarchyCacheEntry previous = Volatile.Read(ref _current);
|
||||||
|
DateTimeOffset queriedAt = _timeProvider.GetUtcNow();
|
||||||
|
|
||||||
|
try
|
||||||
|
{
|
||||||
|
DateTime? deployRaw = await _repository.GetLastDeployTimeAsync(cancellationToken).ConfigureAwait(false);
|
||||||
|
DateTimeOffset? deployTime = deployRaw.HasValue
|
||||||
|
? new DateTimeOffset(DateTime.SpecifyKind(deployRaw.Value, DateTimeKind.Utc))
|
||||||
|
: null;
|
||||||
|
|
||||||
|
bool hasPriorData = previous.HasData;
|
||||||
|
bool deployChanged = !hasPriorData || deployTime != previous.LastDeployTime;
|
||||||
|
|
||||||
|
if (!deployChanged)
|
||||||
|
{
|
||||||
|
// No deploy change — skip heavy queries; just bump LastSuccessAt.
|
||||||
|
GalaxyHierarchyCacheEntry refreshed = previous with
|
||||||
|
{
|
||||||
|
Status = GalaxyCacheStatus.Healthy,
|
||||||
|
LastQueriedAt = queriedAt,
|
||||||
|
LastSuccessAt = queriedAt,
|
||||||
|
LastError = null,
|
||||||
|
};
|
||||||
|
Volatile.Write(ref _current, refreshed);
|
||||||
|
_firstLoad.TrySetResult();
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
Task<List<GalaxyHierarchyRow>> hierarchyTask = _repository.GetHierarchyAsync(cancellationToken);
|
||||||
|
Task<List<GalaxyAttributeRow>> attributesTask = _repository.GetAttributesAsync(cancellationToken);
|
||||||
|
await Task.WhenAll(hierarchyTask, attributesTask).ConfigureAwait(false);
|
||||||
|
|
||||||
|
List<GalaxyHierarchyRow> hierarchy = hierarchyTask.Result;
|
||||||
|
List<GalaxyAttributeRow> attributes = attributesTask.Result;
|
||||||
|
|
||||||
|
long nextSequence = previous.Sequence + 1;
|
||||||
|
GalaxyHierarchyCacheEntry next = BuildEntry(
|
||||||
|
status: GalaxyCacheStatus.Healthy,
|
||||||
|
sequence: nextSequence,
|
||||||
|
lastQueriedAt: queriedAt,
|
||||||
|
lastSuccessAt: queriedAt,
|
||||||
|
lastDeployTime: deployTime,
|
||||||
|
lastError: null,
|
||||||
|
hierarchy: hierarchy,
|
||||||
|
attributes: attributes);
|
||||||
|
|
||||||
|
Volatile.Write(ref _current, next);
|
||||||
|
_firstLoad.TrySetResult();
|
||||||
|
|
||||||
|
_notifier.Publish(new GalaxyDeployEventInfo(
|
||||||
|
Sequence: nextSequence,
|
||||||
|
ObservedAt: queriedAt,
|
||||||
|
TimeOfLastDeploy: deployTime,
|
||||||
|
ObjectCount: hierarchy.Count,
|
||||||
|
AttributeCount: attributes.Count));
|
||||||
|
|
||||||
|
await PersistSnapshotAsync(deployTime, queriedAt, hierarchy, attributes, cancellationToken).ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
|
||||||
|
{
|
||||||
|
throw;
|
||||||
|
}
|
||||||
|
catch (Exception exception)
|
||||||
|
{
|
||||||
|
// Catch every non-cancellation failure — not just SqlException /
|
||||||
|
// InvalidOperationException. A TimeoutException or Win32Exception
|
||||||
|
// from connection establishment, or another DbException subtype,
|
||||||
|
// must still degrade gracefully to Stale/Unavailable and complete
|
||||||
|
// _firstLoad rather than escape and fault the refresh BackgroundService.
|
||||||
|
_logger?.LogWarning(exception, "Galaxy hierarchy cache refresh failed.");
|
||||||
|
GalaxyHierarchyCacheEntry failed = previous with
|
||||||
|
{
|
||||||
|
Status = previous.HasData ? GalaxyCacheStatus.Stale : GalaxyCacheStatus.Unavailable,
|
||||||
|
LastQueriedAt = queriedAt,
|
||||||
|
LastError = exception.Message,
|
||||||
|
};
|
||||||
|
Volatile.Write(ref _current, failed);
|
||||||
|
_firstLoad.TrySetResult();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Materializes a complete <see cref="GalaxyHierarchyCacheEntry"/> from raw
|
||||||
|
/// hierarchy and attribute rowsets. Shared by the live refresh path and the
|
||||||
|
/// on-disk restore path so both produce an identical object list and index.
|
||||||
|
/// </summary>
|
||||||
|
private static GalaxyHierarchyCacheEntry BuildEntry(
|
||||||
|
GalaxyCacheStatus status,
|
||||||
|
long sequence,
|
||||||
|
DateTimeOffset? lastQueriedAt,
|
||||||
|
DateTimeOffset? lastSuccessAt,
|
||||||
|
DateTimeOffset? lastDeployTime,
|
||||||
|
string? lastError,
|
||||||
|
IReadOnlyList<GalaxyHierarchyRow> hierarchy,
|
||||||
|
IReadOnlyList<GalaxyAttributeRow> attributes)
|
||||||
|
{
|
||||||
|
IReadOnlyList<GalaxyObject> objects = BuildObjects(hierarchy, attributes);
|
||||||
|
GalaxyHierarchyIndex index = GalaxyHierarchyIndex.Build(objects);
|
||||||
|
|
||||||
|
int areaCount = hierarchy.Count(row => row.IsArea);
|
||||||
|
int historized = attributes.Count(row => row.IsHistorized);
|
||||||
|
int alarms = attributes.Count(row => row.IsAlarm);
|
||||||
|
|
||||||
|
return new GalaxyHierarchyCacheEntry(
|
||||||
|
Status: status,
|
||||||
|
Sequence: sequence,
|
||||||
|
LastQueriedAt: lastQueriedAt,
|
||||||
|
LastSuccessAt: lastSuccessAt,
|
||||||
|
LastDeployTime: lastDeployTime,
|
||||||
|
LastError: lastError,
|
||||||
|
Objects: objects,
|
||||||
|
Index: index,
|
||||||
|
ObjectCount: hierarchy.Count,
|
||||||
|
AreaCount: areaCount,
|
||||||
|
AttributeCount: attributes.Count,
|
||||||
|
HistorizedAttributeCount: historized,
|
||||||
|
AlarmAttributeCount: alarms);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Seeds the cache from the on-disk snapshot when no live data has loaded yet.
|
||||||
|
/// The restored entry is marked <see cref="GalaxyCacheStatus.Stale"/> — it is
|
||||||
|
/// last-known data, not live. A later refresh that observes the same deploy
|
||||||
|
/// time promotes it to healthy; one that observes a newer deploy replaces it.
|
||||||
|
/// </summary>
|
||||||
|
private async Task TryRestoreFromDiskAsync(CancellationToken cancellationToken)
|
||||||
|
{
|
||||||
|
if (_snapshotStore is null)
|
||||||
|
{
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (Volatile.Read(ref _current).HasData)
|
||||||
|
{
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
GalaxyHierarchySnapshot? snapshot;
|
||||||
|
try
|
||||||
|
{
|
||||||
|
snapshot = await _snapshotStore.TryLoadAsync(cancellationToken).ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
|
||||||
|
{
|
||||||
|
throw;
|
||||||
|
}
|
||||||
|
catch (Exception exception)
|
||||||
|
{
|
||||||
|
_logger?.LogWarning(exception, "Failed to restore the Galaxy hierarchy from the on-disk snapshot.");
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (snapshot is null)
|
||||||
|
{
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
long sequence = Volatile.Read(ref _current).Sequence + 1;
|
||||||
|
GalaxyHierarchyCacheEntry restored = BuildEntry(
|
||||||
|
status: GalaxyCacheStatus.Stale,
|
||||||
|
sequence: sequence,
|
||||||
|
lastQueriedAt: snapshot.SavedAt,
|
||||||
|
lastSuccessAt: snapshot.SavedAt,
|
||||||
|
lastDeployTime: snapshot.LastDeployTime,
|
||||||
|
lastError: null,
|
||||||
|
hierarchy: snapshot.Hierarchy,
|
||||||
|
attributes: snapshot.Attributes);
|
||||||
|
Volatile.Write(ref _current, restored);
|
||||||
|
|
||||||
|
// Restored data is a valid completed first load: unblock callers waiting on
|
||||||
|
// the bootstrap gate immediately, rather than making them wait out the full
|
||||||
|
// wait budget for a live query that — when the database is unreachable, the
|
||||||
|
// scenario this restore exists for — may not return for seconds.
|
||||||
|
_firstLoad.TrySetResult();
|
||||||
|
|
||||||
|
_notifier.Publish(new GalaxyDeployEventInfo(
|
||||||
|
Sequence: sequence,
|
||||||
|
ObservedAt: _timeProvider.GetUtcNow(),
|
||||||
|
TimeOfLastDeploy: snapshot.LastDeployTime,
|
||||||
|
ObjectCount: snapshot.Hierarchy.Count,
|
||||||
|
AttributeCount: snapshot.Attributes.Count));
|
||||||
|
|
||||||
|
_logger?.LogInformation(
|
||||||
|
"Restored Galaxy hierarchy from on-disk snapshot saved {SavedAt:o}: {ObjectCount} objects, {AttributeCount} attributes (status Stale until the Galaxy database confirms).",
|
||||||
|
snapshot.SavedAt,
|
||||||
|
snapshot.Hierarchy.Count,
|
||||||
|
snapshot.Attributes.Count);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Persists a successful refresh to disk. Persistence failures are logged and
|
||||||
|
/// swallowed — a cache that cannot write its backup is still fully usable.
|
||||||
|
/// </summary>
|
||||||
|
private async Task PersistSnapshotAsync(
|
||||||
|
DateTimeOffset? deployTime,
|
||||||
|
DateTimeOffset savedAt,
|
||||||
|
IReadOnlyList<GalaxyHierarchyRow> hierarchy,
|
||||||
|
IReadOnlyList<GalaxyAttributeRow> attributes,
|
||||||
|
CancellationToken cancellationToken)
|
||||||
|
{
|
||||||
|
if (_snapshotStore is null)
|
||||||
|
{
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
try
|
||||||
|
{
|
||||||
|
await _snapshotStore.SaveAsync(
|
||||||
|
new GalaxyHierarchySnapshot(deployTime, savedAt, hierarchy, attributes),
|
||||||
|
cancellationToken).ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
|
||||||
|
{
|
||||||
|
// The refresh was cancelled (service shutdown) before the write finished.
|
||||||
|
// That is not a persistence failure — do not log it as a warning.
|
||||||
|
}
|
||||||
|
catch (Exception exception)
|
||||||
|
{
|
||||||
|
_logger?.LogWarning(exception, "Failed to persist the Galaxy hierarchy snapshot to disk.");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private static IReadOnlyList<GalaxyObject> BuildObjects(
|
||||||
|
IReadOnlyList<GalaxyHierarchyRow> hierarchy,
|
||||||
|
IReadOnlyList<GalaxyAttributeRow> attributes)
|
||||||
|
{
|
||||||
|
Dictionary<int, List<GalaxyAttributeRow>> attributesByGobjectId = attributes
|
||||||
|
.GroupBy(a => a.GobjectId)
|
||||||
|
.ToDictionary(g => g.Key, g => g.ToList());
|
||||||
|
|
||||||
|
List<GalaxyObject> objects = new(hierarchy.Count);
|
||||||
|
foreach (GalaxyHierarchyRow row in hierarchy)
|
||||||
|
{
|
||||||
|
objects.Add(GalaxyProtoMapper.MapObject(row, attributesByGobjectId));
|
||||||
|
}
|
||||||
|
return objects;
|
||||||
|
}
|
||||||
|
|
||||||
|
private GalaxyCacheStatus ProjectStatus(GalaxyHierarchyCacheEntry snapshot)
|
||||||
|
{
|
||||||
|
if (snapshot.Status is GalaxyCacheStatus.Unknown or GalaxyCacheStatus.Unavailable)
|
||||||
|
{
|
||||||
|
return snapshot.Status;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (snapshot.LastSuccessAt is { } success
|
||||||
|
&& _timeProvider.GetUtcNow() - success > StaleThreshold)
|
||||||
|
{
|
||||||
|
return GalaxyCacheStatus.Stale;
|
||||||
|
}
|
||||||
|
|
||||||
|
return snapshot.Status;
|
||||||
|
}
|
||||||
|
}
|
||||||
+56
@@ -0,0 +1,56 @@
|
|||||||
|
using ZB.MOM.WW.GalaxyRepository.Grpc;
|
||||||
|
|
||||||
|
namespace ZB.MOM.WW.GalaxyRepository;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Immutable snapshot of the Galaxy Repository browse data held by
|
||||||
|
/// <see cref="GalaxyHierarchyCache"/>. Multiple gRPC clients share the same
|
||||||
|
/// materialized object list and precomputed hierarchy index.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="Status">The cache freshness state at the time the entry was produced.</param>
|
||||||
|
/// <param name="Sequence">Monotonically increasing per process start; bumped on each heavy refresh.</param>
|
||||||
|
/// <param name="LastQueriedAt">UTC wall-clock of the most recent refresh attempt.</param>
|
||||||
|
/// <param name="LastSuccessAt">UTC wall-clock of the most recent successful refresh.</param>
|
||||||
|
/// <param name="LastDeployTime">The <c>galaxy.time_of_last_deploy</c> the data was pulled at.</param>
|
||||||
|
/// <param name="LastError">The most recent refresh error message, or <see langword="null"/>.</param>
|
||||||
|
/// <param name="Objects">The materialized Galaxy object list.</param>
|
||||||
|
/// <param name="Index">Precomputed lookup structures over <paramref name="Objects"/>.</param>
|
||||||
|
/// <param name="ObjectCount">Number of objects in the hierarchy.</param>
|
||||||
|
/// <param name="AreaCount">Number of area objects in the hierarchy.</param>
|
||||||
|
/// <param name="AttributeCount">Number of attributes across all objects.</param>
|
||||||
|
/// <param name="HistorizedAttributeCount">Number of historized attributes.</param>
|
||||||
|
/// <param name="AlarmAttributeCount">Number of alarm-bearing attributes.</param>
|
||||||
|
public sealed record GalaxyHierarchyCacheEntry(
|
||||||
|
GalaxyCacheStatus Status,
|
||||||
|
long Sequence,
|
||||||
|
DateTimeOffset? LastQueriedAt,
|
||||||
|
DateTimeOffset? LastSuccessAt,
|
||||||
|
DateTimeOffset? LastDeployTime,
|
||||||
|
string? LastError,
|
||||||
|
IReadOnlyList<GalaxyObject> Objects,
|
||||||
|
GalaxyHierarchyIndex Index,
|
||||||
|
int ObjectCount,
|
||||||
|
int AreaCount,
|
||||||
|
int AttributeCount,
|
||||||
|
int HistorizedAttributeCount,
|
||||||
|
int AlarmAttributeCount)
|
||||||
|
{
|
||||||
|
/// <summary>Gets an empty Galaxy hierarchy cache entry.</summary>
|
||||||
|
public static GalaxyHierarchyCacheEntry Empty { get; } = new(
|
||||||
|
Status: GalaxyCacheStatus.Unknown,
|
||||||
|
Sequence: 0,
|
||||||
|
LastQueriedAt: null,
|
||||||
|
LastSuccessAt: null,
|
||||||
|
LastDeployTime: null,
|
||||||
|
LastError: null,
|
||||||
|
Objects: Array.Empty<GalaxyObject>(),
|
||||||
|
Index: GalaxyHierarchyIndex.Empty,
|
||||||
|
ObjectCount: 0,
|
||||||
|
AreaCount: 0,
|
||||||
|
AttributeCount: 0,
|
||||||
|
HistorizedAttributeCount: 0,
|
||||||
|
AlarmAttributeCount: 0);
|
||||||
|
|
||||||
|
/// <summary>Gets a value indicating whether the cache entry contains usable data.</summary>
|
||||||
|
public bool HasData => Status is GalaxyCacheStatus.Healthy or GalaxyCacheStatus.Stale;
|
||||||
|
}
|
||||||
@@ -0,0 +1,206 @@
|
|||||||
|
using ZB.MOM.WW.GalaxyRepository.Grpc;
|
||||||
|
|
||||||
|
namespace ZB.MOM.WW.GalaxyRepository;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Precomputed lookup structures over a materialized Galaxy object list. Built once per
|
||||||
|
/// cache entry so browse/discover handlers can resolve roots/parents by id, tag name, or
|
||||||
|
/// contained path in O(1), enumerate direct children, and resolve tag addresses to objects
|
||||||
|
/// or attributes without rescanning the full object list.
|
||||||
|
/// </summary>
|
||||||
|
public sealed class GalaxyHierarchyIndex
|
||||||
|
{
|
||||||
|
private GalaxyHierarchyIndex(
|
||||||
|
IReadOnlyList<GalaxyObjectView> objectViews,
|
||||||
|
IReadOnlyDictionary<int, GalaxyObjectView> objectViewsById,
|
||||||
|
IReadOnlyDictionary<string, GalaxyTagLookup> tagsByAddress,
|
||||||
|
IReadOnlyDictionary<int, IReadOnlyList<GalaxyObjectView>> childrenByParent,
|
||||||
|
IReadOnlyDictionary<string, GalaxyObjectView> objectViewsByTagName,
|
||||||
|
IReadOnlyDictionary<string, GalaxyObjectView> objectViewsByContainedPath)
|
||||||
|
{
|
||||||
|
ObjectViews = objectViews;
|
||||||
|
ObjectViewsById = objectViewsById;
|
||||||
|
TagsByAddress = tagsByAddress;
|
||||||
|
ChildrenByParent = childrenByParent;
|
||||||
|
ObjectViewsByTagName = objectViewsByTagName;
|
||||||
|
ObjectViewsByContainedPath = objectViewsByContainedPath;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>Gets an empty Galaxy hierarchy index.</summary>
|
||||||
|
public static GalaxyHierarchyIndex Empty { get; } = new(
|
||||||
|
Array.Empty<GalaxyObjectView>(),
|
||||||
|
new Dictionary<int, GalaxyObjectView>(),
|
||||||
|
new Dictionary<string, GalaxyTagLookup>(StringComparer.OrdinalIgnoreCase),
|
||||||
|
new Dictionary<int, IReadOnlyList<GalaxyObjectView>>(),
|
||||||
|
new Dictionary<string, GalaxyObjectView>(StringComparer.OrdinalIgnoreCase),
|
||||||
|
new Dictionary<string, GalaxyObjectView>(StringComparer.OrdinalIgnoreCase));
|
||||||
|
|
||||||
|
/// <summary>Gets the object views.</summary>
|
||||||
|
public IReadOnlyList<GalaxyObjectView> ObjectViews { get; }
|
||||||
|
|
||||||
|
/// <summary>Gets the object views indexed by gobject id.</summary>
|
||||||
|
public IReadOnlyDictionary<int, GalaxyObjectView> ObjectViewsById { get; }
|
||||||
|
|
||||||
|
/// <summary>Gets tags indexed by address.</summary>
|
||||||
|
public IReadOnlyDictionary<string, GalaxyTagLookup> TagsByAddress { get; }
|
||||||
|
|
||||||
|
/// <summary>Gets direct children grouped by parent gobject id. Root objects (no parent, or self-parented) live under key 0. Each list is sorted areas-first, then by display name (OrdinalIgnoreCase).</summary>
|
||||||
|
public IReadOnlyDictionary<int, IReadOnlyList<GalaxyObjectView>> ChildrenByParent { get; }
|
||||||
|
|
||||||
|
/// <summary>Gets object views indexed by <see cref="GalaxyObject.TagName"/> (OrdinalIgnoreCase). Lets browse/discover handlers resolve parents/roots by tag name in O(1) instead of scanning <see cref="ObjectViews"/>.</summary>
|
||||||
|
public IReadOnlyDictionary<string, GalaxyObjectView> ObjectViewsByTagName { get; }
|
||||||
|
|
||||||
|
/// <summary>Gets object views indexed by contained path (OrdinalIgnoreCase). Lets browse/discover handlers resolve parents/roots by path in O(1) instead of scanning <see cref="ObjectViews"/>.</summary>
|
||||||
|
public IReadOnlyDictionary<string, GalaxyObjectView> ObjectViewsByContainedPath { get; }
|
||||||
|
|
||||||
|
/// <summary>Builds a Galaxy hierarchy index from the given objects.</summary>
|
||||||
|
/// <param name="objects">The Galaxy objects to index.</param>
|
||||||
|
/// <returns>A new Galaxy hierarchy index.</returns>
|
||||||
|
public static GalaxyHierarchyIndex Build(IReadOnlyList<GalaxyObject> objects)
|
||||||
|
{
|
||||||
|
if (objects.Count == 0)
|
||||||
|
{
|
||||||
|
return Empty;
|
||||||
|
}
|
||||||
|
|
||||||
|
Dictionary<int, GalaxyObject> objectsById = new();
|
||||||
|
foreach (GalaxyObject obj in objects)
|
||||||
|
{
|
||||||
|
objectsById.TryAdd(obj.GobjectId, obj);
|
||||||
|
}
|
||||||
|
|
||||||
|
List<GalaxyObjectView> views = new(objects.Count);
|
||||||
|
Dictionary<int, GalaxyObjectView> viewsById = new();
|
||||||
|
Dictionary<string, GalaxyTagLookup> tagsByAddress = new(StringComparer.OrdinalIgnoreCase);
|
||||||
|
Dictionary<string, GalaxyObjectView> viewsByTagName = new(StringComparer.OrdinalIgnoreCase);
|
||||||
|
Dictionary<string, GalaxyObjectView> viewsByContainedPath = new(StringComparer.OrdinalIgnoreCase);
|
||||||
|
|
||||||
|
foreach (GalaxyObject obj in objects)
|
||||||
|
{
|
||||||
|
string path = BuildContainedPath(obj, objectsById);
|
||||||
|
int depth = string.IsNullOrWhiteSpace(path) ? 0 : path.Count(character => character == '/');
|
||||||
|
GalaxyObjectView view = new(obj, path, depth);
|
||||||
|
views.Add(view);
|
||||||
|
viewsById.TryAdd(obj.GobjectId, view);
|
||||||
|
|
||||||
|
if (!string.IsNullOrWhiteSpace(obj.TagName))
|
||||||
|
{
|
||||||
|
tagsByAddress.TryAdd(obj.TagName, new GalaxyTagLookup(obj, Attribute: null, path));
|
||||||
|
viewsByTagName.TryAdd(obj.TagName, view);
|
||||||
|
}
|
||||||
|
|
||||||
|
if (!string.IsNullOrWhiteSpace(path))
|
||||||
|
{
|
||||||
|
viewsByContainedPath.TryAdd(path, view);
|
||||||
|
}
|
||||||
|
|
||||||
|
foreach (GalaxyAttribute attribute in obj.Attributes)
|
||||||
|
{
|
||||||
|
if (!string.IsNullOrWhiteSpace(attribute.FullTagReference))
|
||||||
|
{
|
||||||
|
tagsByAddress.TryAdd(attribute.FullTagReference, new GalaxyTagLookup(obj, attribute, path));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
Dictionary<int, List<GalaxyObjectView>> childrenByParent = new();
|
||||||
|
foreach (GalaxyObjectView view in views)
|
||||||
|
{
|
||||||
|
int parentKey = view.Object.ParentGobjectId;
|
||||||
|
// Treat self-parented (corrupt) rows as roots.
|
||||||
|
if (parentKey == view.Object.GobjectId)
|
||||||
|
{
|
||||||
|
parentKey = 0;
|
||||||
|
}
|
||||||
|
// Re-root orphans whose parent object is absent from the set (e.g. a deleted or
|
||||||
|
// never-loaded container area). Otherwise they bucket under a phantom parent id
|
||||||
|
// that is never reached from the root, so they vanish from browse entirely.
|
||||||
|
else if (parentKey != 0 && !objectsById.ContainsKey(parentKey))
|
||||||
|
{
|
||||||
|
parentKey = 0;
|
||||||
|
}
|
||||||
|
if (!childrenByParent.TryGetValue(parentKey, out List<GalaxyObjectView>? bucket))
|
||||||
|
{
|
||||||
|
bucket = [];
|
||||||
|
childrenByParent[parentKey] = bucket;
|
||||||
|
}
|
||||||
|
bucket.Add(view);
|
||||||
|
}
|
||||||
|
|
||||||
|
foreach (List<GalaxyObjectView> bucket in childrenByParent.Values)
|
||||||
|
{
|
||||||
|
bucket.Sort(CompareByAreaThenDisplayName);
|
||||||
|
}
|
||||||
|
|
||||||
|
Dictionary<int, IReadOnlyList<GalaxyObjectView>> readOnlyChildren = new(childrenByParent.Count);
|
||||||
|
foreach (KeyValuePair<int, List<GalaxyObjectView>> kvp in childrenByParent)
|
||||||
|
{
|
||||||
|
readOnlyChildren[kvp.Key] = kvp.Value;
|
||||||
|
}
|
||||||
|
|
||||||
|
return new GalaxyHierarchyIndex(
|
||||||
|
views,
|
||||||
|
viewsById,
|
||||||
|
tagsByAddress,
|
||||||
|
readOnlyChildren,
|
||||||
|
viewsByTagName,
|
||||||
|
viewsByContainedPath);
|
||||||
|
}
|
||||||
|
|
||||||
|
private static string BuildContainedPath(
|
||||||
|
GalaxyObject obj,
|
||||||
|
IReadOnlyDictionary<int, GalaxyObject> objectsById)
|
||||||
|
{
|
||||||
|
Stack<string> names = new();
|
||||||
|
HashSet<int> seen = [];
|
||||||
|
GalaxyObject? current = obj;
|
||||||
|
while (current is not null && seen.Add(current.GobjectId))
|
||||||
|
{
|
||||||
|
names.Push(ResolvePathSegment(current));
|
||||||
|
current = current.ParentGobjectId != 0
|
||||||
|
&& objectsById.TryGetValue(current.ParentGobjectId, out GalaxyObject? parent)
|
||||||
|
? parent
|
||||||
|
: null;
|
||||||
|
}
|
||||||
|
|
||||||
|
return string.Join('/', names.Where(name => !string.IsNullOrWhiteSpace(name)));
|
||||||
|
}
|
||||||
|
|
||||||
|
private static string ResolvePathSegment(GalaxyObject obj)
|
||||||
|
{
|
||||||
|
if (!string.IsNullOrWhiteSpace(obj.ContainedName))
|
||||||
|
{
|
||||||
|
return obj.ContainedName;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (!string.IsNullOrWhiteSpace(obj.BrowseName))
|
||||||
|
{
|
||||||
|
return obj.BrowseName;
|
||||||
|
}
|
||||||
|
|
||||||
|
return obj.TagName;
|
||||||
|
}
|
||||||
|
|
||||||
|
private static int CompareByAreaThenDisplayName(GalaxyObjectView left, GalaxyObjectView right)
|
||||||
|
{
|
||||||
|
if (left.Object.IsArea != right.Object.IsArea)
|
||||||
|
{
|
||||||
|
return left.Object.IsArea ? -1 : 1;
|
||||||
|
}
|
||||||
|
return string.Compare(DisplayNameOf(left), DisplayNameOf(right), StringComparison.OrdinalIgnoreCase);
|
||||||
|
}
|
||||||
|
|
||||||
|
private static string DisplayNameOf(GalaxyObjectView view)
|
||||||
|
{
|
||||||
|
GalaxyObject obj = view.Object;
|
||||||
|
if (!string.IsNullOrWhiteSpace(obj.BrowseName))
|
||||||
|
{
|
||||||
|
return obj.BrowseName;
|
||||||
|
}
|
||||||
|
if (!string.IsNullOrWhiteSpace(obj.ContainedName))
|
||||||
|
{
|
||||||
|
return obj.ContainedName;
|
||||||
|
}
|
||||||
|
return obj.TagName;
|
||||||
|
}
|
||||||
|
}
|
||||||
+317
@@ -0,0 +1,317 @@
|
|||||||
|
using System.Collections.Concurrent;
|
||||||
|
using System.Runtime.CompilerServices;
|
||||||
|
using System.Security.Cryptography;
|
||||||
|
using System.Text;
|
||||||
|
using Grpc.Core;
|
||||||
|
using ZB.MOM.WW.GalaxyRepository.Grpc;
|
||||||
|
|
||||||
|
namespace ZB.MOM.WW.GalaxyRepository;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Projects a <c>DiscoverHierarchy</c> request against an immutable
|
||||||
|
/// <see cref="GalaxyHierarchyCacheEntry"/>: applies the root/depth/category/template/glob
|
||||||
|
/// filters, pages the result, and memoizes the filtered list per cache-entry instance so
|
||||||
|
/// paging is O(pageSize) rather than O(total) per page. Pure and side-effect free.
|
||||||
|
/// </summary>
|
||||||
|
public static class GalaxyHierarchyProjector
|
||||||
|
{
|
||||||
|
/// <summary>
|
||||||
|
/// Per-cache-entry memo of filtered, ordered <see cref="GalaxyObjectView"/> lists
|
||||||
|
/// keyed by filter signature. Without it, paging through a large hierarchy
|
||||||
|
/// re-applies every filter and re-scans the full <see cref="GalaxyHierarchyIndex.ObjectViews"/>
|
||||||
|
/// collection on every page — O(total) per page, O(total²/pageSize) end-to-end.
|
||||||
|
/// With it, the first page builds the filtered list and each subsequent page is an
|
||||||
|
/// O(pageSize) slice. The table is keyed on the immutable cache-entry instance, so
|
||||||
|
/// when the cache publishes a new entry the stale memo becomes unreachable and is
|
||||||
|
/// reclaimed with it — no explicit invalidation needed.
|
||||||
|
/// </summary>
|
||||||
|
private static readonly ConditionalWeakTable<GalaxyHierarchyCacheEntry, ConcurrentDictionary<string, IReadOnlyList<GalaxyObjectView>>> FilteredViewCache = new();
|
||||||
|
|
||||||
|
/// <summary>Projects a discovery request against a cache entry and returns all matching objects.</summary>
|
||||||
|
/// <param name="entry">The Galaxy hierarchy cache entry.</param>
|
||||||
|
/// <param name="request">The discovery hierarchy request.</param>
|
||||||
|
/// <param name="browseSubtreeGlobs">Optional glob patterns to filter browse subtrees.</param>
|
||||||
|
public static GalaxyHierarchyQueryResult Project(
|
||||||
|
GalaxyHierarchyCacheEntry entry,
|
||||||
|
DiscoverHierarchyRequest request,
|
||||||
|
IReadOnlyList<string>? browseSubtreeGlobs = null)
|
||||||
|
{
|
||||||
|
return Project(
|
||||||
|
entry,
|
||||||
|
request,
|
||||||
|
browseSubtreeGlobs,
|
||||||
|
offset: 0,
|
||||||
|
pageSize: int.MaxValue);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>Projects a discovery request with paging against a cache entry and returns a page of matching objects.</summary>
|
||||||
|
/// <param name="entry">The Galaxy hierarchy cache entry.</param>
|
||||||
|
/// <param name="request">The discovery hierarchy request.</param>
|
||||||
|
/// <param name="browseSubtreeGlobs">Optional glob patterns to filter browse subtrees.</param>
|
||||||
|
/// <param name="offset">The zero-based offset into the result set.</param>
|
||||||
|
/// <param name="pageSize">The maximum number of results to return.</param>
|
||||||
|
public static GalaxyHierarchyQueryResult Project(
|
||||||
|
GalaxyHierarchyCacheEntry entry,
|
||||||
|
DiscoverHierarchyRequest request,
|
||||||
|
IReadOnlyList<string>? browseSubtreeGlobs,
|
||||||
|
int offset,
|
||||||
|
int pageSize)
|
||||||
|
{
|
||||||
|
ArgumentNullException.ThrowIfNull(entry);
|
||||||
|
ArgumentNullException.ThrowIfNull(request);
|
||||||
|
if (offset < 0)
|
||||||
|
{
|
||||||
|
throw new ArgumentOutOfRangeException(nameof(offset), offset, "Offset must be greater than or equal to zero.");
|
||||||
|
}
|
||||||
|
|
||||||
|
if (pageSize <= 0)
|
||||||
|
{
|
||||||
|
throw new ArgumentOutOfRangeException(nameof(pageSize), pageSize, "Page size must be greater than zero.");
|
||||||
|
}
|
||||||
|
|
||||||
|
int? maxDepth = request.MaxDepth;
|
||||||
|
if (maxDepth < 0)
|
||||||
|
{
|
||||||
|
throw new RpcException(new Status(
|
||||||
|
StatusCode.InvalidArgument,
|
||||||
|
"DiscoverHierarchy max_depth must be greater than or equal to zero when provided."));
|
||||||
|
}
|
||||||
|
|
||||||
|
string filterSignature = ComputeFilterSignature(request, browseSubtreeGlobs);
|
||||||
|
IReadOnlyList<GalaxyObjectView> matchedViews = GetFilteredViews(
|
||||||
|
entry,
|
||||||
|
request,
|
||||||
|
browseSubtreeGlobs,
|
||||||
|
maxDepth,
|
||||||
|
filterSignature);
|
||||||
|
|
||||||
|
bool includeAttributes = IncludeAttributes(request);
|
||||||
|
List<GalaxyObject> page = new(Math.Min(pageSize, Math.Max(0, matchedViews.Count - offset)));
|
||||||
|
int end = (int)Math.Min((long)offset + pageSize, matchedViews.Count);
|
||||||
|
for (int index = offset; index < end; index++)
|
||||||
|
{
|
||||||
|
page.Add(CloneObject(matchedViews[index].Object, includeAttributes));
|
||||||
|
}
|
||||||
|
|
||||||
|
return new GalaxyHierarchyQueryResult(
|
||||||
|
page,
|
||||||
|
matchedViews.Count,
|
||||||
|
filterSignature);
|
||||||
|
}
|
||||||
|
|
||||||
|
private static IReadOnlyList<GalaxyObjectView> GetFilteredViews(
|
||||||
|
GalaxyHierarchyCacheEntry entry,
|
||||||
|
DiscoverHierarchyRequest request,
|
||||||
|
IReadOnlyList<string>? browseSubtreeGlobs,
|
||||||
|
int? maxDepth,
|
||||||
|
string filterSignature)
|
||||||
|
{
|
||||||
|
// ResolveRoot can throw RpcException(NotFound); run it before consulting the
|
||||||
|
// memo so a bad root surfaces consistently regardless of cache state.
|
||||||
|
IReadOnlyList<GalaxyObjectView> views = entry.Index.ObjectViews;
|
||||||
|
GalaxyObjectView? root = ResolveRoot(request, entry.Index);
|
||||||
|
|
||||||
|
ConcurrentDictionary<string, IReadOnlyList<GalaxyObjectView>> memo =
|
||||||
|
FilteredViewCache.GetValue(entry, static _ => new ConcurrentDictionary<string, IReadOnlyList<GalaxyObjectView>>(StringComparer.Ordinal));
|
||||||
|
|
||||||
|
return memo.GetOrAdd(
|
||||||
|
filterSignature,
|
||||||
|
static (_, state) =>
|
||||||
|
{
|
||||||
|
List<GalaxyObjectView> matched = [];
|
||||||
|
foreach (GalaxyObjectView view in state.Views)
|
||||||
|
{
|
||||||
|
if (MatchesRoot(view, state.Root, state.MaxDepth)
|
||||||
|
&& MatchesBrowseSubtrees(view, state.BrowseSubtreeGlobs)
|
||||||
|
&& MatchesFilters(view.Object, state.Request))
|
||||||
|
{
|
||||||
|
matched.Add(view);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return matched;
|
||||||
|
},
|
||||||
|
(Views: views, Root: root, MaxDepth: maxDepth, BrowseSubtreeGlobs: browseSubtreeGlobs, Request: request));
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>Finds an object in the hierarchy by its tag address.</summary>
|
||||||
|
/// <param name="entry">The Galaxy hierarchy cache entry.</param>
|
||||||
|
/// <param name="tagAddress">The tag address to search for.</param>
|
||||||
|
public static GalaxyObject? FindObjectForTag(
|
||||||
|
GalaxyHierarchyCacheEntry entry,
|
||||||
|
string tagAddress)
|
||||||
|
{
|
||||||
|
if (string.IsNullOrWhiteSpace(tagAddress))
|
||||||
|
{
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
|
||||||
|
return entry.Index.TagsByAddress.TryGetValue(tagAddress, out GalaxyTagLookup? lookup)
|
||||||
|
? lookup.Object
|
||||||
|
: null;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>Finds an attribute in the hierarchy by its tag address.</summary>
|
||||||
|
/// <param name="entry">The Galaxy hierarchy cache entry.</param>
|
||||||
|
/// <param name="tagAddress">The tag address to search for.</param>
|
||||||
|
public static GalaxyAttribute? FindAttributeForTag(
|
||||||
|
GalaxyHierarchyCacheEntry entry,
|
||||||
|
string tagAddress)
|
||||||
|
{
|
||||||
|
if (string.IsNullOrWhiteSpace(tagAddress))
|
||||||
|
{
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
|
||||||
|
return entry.Index.TagsByAddress.TryGetValue(tagAddress, out GalaxyTagLookup? lookup)
|
||||||
|
? lookup.Attribute
|
||||||
|
: null;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>Gets the contained path for an object by its gobject ID.</summary>
|
||||||
|
/// <param name="entry">The Galaxy hierarchy cache entry.</param>
|
||||||
|
/// <param name="gobjectId">The Galaxy object ID.</param>
|
||||||
|
public static string GetContainedPath(
|
||||||
|
GalaxyHierarchyCacheEntry entry,
|
||||||
|
int gobjectId)
|
||||||
|
{
|
||||||
|
return entry.Index.ObjectViewsById.TryGetValue(gobjectId, out GalaxyObjectView? view)
|
||||||
|
? view.ContainedPath
|
||||||
|
: string.Empty;
|
||||||
|
}
|
||||||
|
|
||||||
|
private static GalaxyObjectView? ResolveRoot(
|
||||||
|
DiscoverHierarchyRequest request,
|
||||||
|
GalaxyHierarchyIndex index)
|
||||||
|
{
|
||||||
|
GalaxyObjectView? root = request.RootCase switch
|
||||||
|
{
|
||||||
|
DiscoverHierarchyRequest.RootOneofCase.None => null,
|
||||||
|
DiscoverHierarchyRequest.RootOneofCase.RootGobjectId =>
|
||||||
|
index.ObjectViewsById.TryGetValue(request.RootGobjectId, out GalaxyObjectView? byId) ? byId : null,
|
||||||
|
DiscoverHierarchyRequest.RootOneofCase.RootTagName =>
|
||||||
|
index.ObjectViewsByTagName.TryGetValue(request.RootTagName, out GalaxyObjectView? byTag) ? byTag : null,
|
||||||
|
DiscoverHierarchyRequest.RootOneofCase.RootContainedPath =>
|
||||||
|
index.ObjectViewsByContainedPath.TryGetValue(request.RootContainedPath, out GalaxyObjectView? byPath) ? byPath : null,
|
||||||
|
_ => null,
|
||||||
|
};
|
||||||
|
|
||||||
|
if (request.RootCase != DiscoverHierarchyRequest.RootOneofCase.None && root is null)
|
||||||
|
{
|
||||||
|
throw new RpcException(new Status(StatusCode.NotFound, "DiscoverHierarchy root was not found."));
|
||||||
|
}
|
||||||
|
|
||||||
|
return root;
|
||||||
|
}
|
||||||
|
|
||||||
|
private static bool MatchesRoot(
|
||||||
|
GalaxyObjectView view,
|
||||||
|
GalaxyObjectView? root,
|
||||||
|
int? maxDepth)
|
||||||
|
{
|
||||||
|
if (root is null)
|
||||||
|
{
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
|
||||||
|
bool isRoot = view.Object.GobjectId == root.Object.GobjectId;
|
||||||
|
bool isDescendant = view.ContainedPath.StartsWith(root.ContainedPath + "/", StringComparison.OrdinalIgnoreCase);
|
||||||
|
if (!isRoot && !isDescendant)
|
||||||
|
{
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
|
return maxDepth is null || view.Depth - root.Depth <= maxDepth.Value;
|
||||||
|
}
|
||||||
|
|
||||||
|
private static bool MatchesBrowseSubtrees(
|
||||||
|
GalaxyObjectView view,
|
||||||
|
IReadOnlyList<string>? browseSubtreeGlobs)
|
||||||
|
{
|
||||||
|
return browseSubtreeGlobs is null
|
||||||
|
|| browseSubtreeGlobs.Count == 0
|
||||||
|
|| browseSubtreeGlobs.Any(glob => GalaxyGlobMatcher.IsMatch(view.ContainedPath, glob));
|
||||||
|
}
|
||||||
|
|
||||||
|
private static bool MatchesFilters(
|
||||||
|
GalaxyObject obj,
|
||||||
|
DiscoverHierarchyRequest request)
|
||||||
|
{
|
||||||
|
if (request.CategoryIds.Count > 0 && !request.CategoryIds.Contains(obj.CategoryId))
|
||||||
|
{
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
|
foreach (string templateFilter in request.TemplateChainContains)
|
||||||
|
{
|
||||||
|
if (!obj.TemplateChain.Any(template => template.Contains(templateFilter, StringComparison.OrdinalIgnoreCase)))
|
||||||
|
{
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if (!string.IsNullOrWhiteSpace(request.TagNameGlob)
|
||||||
|
&& !GalaxyGlobMatcher.IsMatch(obj.TagName, request.TagNameGlob))
|
||||||
|
{
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (request.AlarmBearingOnly && !obj.Attributes.Any(attribute => attribute.IsAlarm))
|
||||||
|
{
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (request.HistorizedOnly && !obj.Attributes.Any(attribute => attribute.IsHistorized))
|
||||||
|
{
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
|
||||||
|
private static bool IncludeAttributes(DiscoverHierarchyRequest request)
|
||||||
|
{
|
||||||
|
return !request.HasIncludeAttributes || request.IncludeAttributes;
|
||||||
|
}
|
||||||
|
|
||||||
|
private static GalaxyObject CloneObject(GalaxyObject source, bool includeAttributes)
|
||||||
|
{
|
||||||
|
GalaxyObject clone = source.Clone();
|
||||||
|
if (!includeAttributes)
|
||||||
|
{
|
||||||
|
clone.Attributes.Clear();
|
||||||
|
}
|
||||||
|
|
||||||
|
return clone;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>Computes a stable filter signature for memoization purposes.</summary>
|
||||||
|
/// <param name="request">The discovery hierarchy request.</param>
|
||||||
|
/// <param name="browseSubtreeGlobs">Optional glob patterns to filter browse subtrees.</param>
|
||||||
|
public static string ComputeFilterSignature(
|
||||||
|
DiscoverHierarchyRequest request,
|
||||||
|
IReadOnlyList<string>? browseSubtreeGlobs)
|
||||||
|
{
|
||||||
|
StringBuilder builder = new();
|
||||||
|
builder.Append("root=").Append(request.RootCase).Append('|');
|
||||||
|
builder.Append(request.RootCase switch
|
||||||
|
{
|
||||||
|
DiscoverHierarchyRequest.RootOneofCase.RootGobjectId => request.RootGobjectId.ToString(
|
||||||
|
System.Globalization.CultureInfo.InvariantCulture),
|
||||||
|
DiscoverHierarchyRequest.RootOneofCase.RootTagName => request.RootTagName,
|
||||||
|
DiscoverHierarchyRequest.RootOneofCase.RootContainedPath => request.RootContainedPath,
|
||||||
|
_ => string.Empty,
|
||||||
|
});
|
||||||
|
builder.Append("|max=").Append(request.MaxDepth?.ToString(System.Globalization.CultureInfo.InvariantCulture) ?? "");
|
||||||
|
builder.Append("|cat=").AppendJoin(',', request.CategoryIds.Order());
|
||||||
|
builder.Append("|tpl=").AppendJoin(',', request.TemplateChainContains.Order(StringComparer.OrdinalIgnoreCase));
|
||||||
|
builder.Append("|glob=").Append(request.TagNameGlob);
|
||||||
|
builder.Append("|attrs=").Append(request.HasIncludeAttributes ? request.IncludeAttributes.ToString() : "unset");
|
||||||
|
builder.Append("|alarm=").Append(request.AlarmBearingOnly);
|
||||||
|
builder.Append("|hist=").Append(request.HistorizedOnly);
|
||||||
|
builder.Append("|browse=").AppendJoin(',', (browseSubtreeGlobs ?? Array.Empty<string>()).Order(StringComparer.OrdinalIgnoreCase));
|
||||||
|
|
||||||
|
byte[] hash = SHA256.HashData(Encoding.UTF8.GetBytes(builder.ToString()));
|
||||||
|
return Convert.ToHexString(hash, 0, 12);
|
||||||
|
}
|
||||||
|
}
|
||||||
+16
@@ -0,0 +1,16 @@
|
|||||||
|
using ZB.MOM.WW.GalaxyRepository.Grpc;
|
||||||
|
|
||||||
|
namespace ZB.MOM.WW.GalaxyRepository;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Result of one <see cref="GalaxyHierarchyProjector.Project(GalaxyHierarchyCacheEntry, DiscoverHierarchyRequest, System.Collections.Generic.IReadOnlyList{string}, int, int)"/>
|
||||||
|
/// call: a materialized page of matching objects, the total post-filter object count, and
|
||||||
|
/// the stable filter signature used to bind page tokens.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="Objects">The page of matching objects.</param>
|
||||||
|
/// <param name="TotalObjectCount">Total matching objects across the whole hierarchy (post-filter).</param>
|
||||||
|
/// <param name="FilterSignature">Stable signature of the filter set, used to bind page tokens.</param>
|
||||||
|
public sealed record GalaxyHierarchyQueryResult(
|
||||||
|
IReadOnlyList<GalaxyObject> Objects,
|
||||||
|
int TotalObjectCount,
|
||||||
|
string FilterSignature);
|
||||||
+62
@@ -0,0 +1,62 @@
|
|||||||
|
using Microsoft.Extensions.Hosting;
|
||||||
|
using Microsoft.Extensions.Logging;
|
||||||
|
using Microsoft.Extensions.Options;
|
||||||
|
|
||||||
|
namespace ZB.MOM.WW.GalaxyRepository;
|
||||||
|
|
||||||
|
/// <summary>Background service that periodically refreshes the Galaxy Repository hierarchy cache off the request path.</summary>
|
||||||
|
public sealed class GalaxyHierarchyRefreshService(
|
||||||
|
IGalaxyHierarchyCache cache,
|
||||||
|
IOptions<GalaxyRepositoryOptions> options,
|
||||||
|
ILogger<GalaxyHierarchyRefreshService> logger,
|
||||||
|
TimeProvider? timeProvider = null) : BackgroundService
|
||||||
|
{
|
||||||
|
private readonly TimeProvider _timeProvider = timeProvider ?? TimeProvider.System;
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||||
|
{
|
||||||
|
TimeSpan interval = TimeSpan.FromSeconds(Math.Max(1, options.Value.DashboardRefreshIntervalSeconds));
|
||||||
|
|
||||||
|
try
|
||||||
|
{
|
||||||
|
await cache.RefreshAsync(stoppingToken).ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
|
||||||
|
{
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
catch (Exception exception)
|
||||||
|
{
|
||||||
|
// A transient first-load failure (e.g. a TimeoutException or
|
||||||
|
// Win32Exception from connection establishment, or a DbException
|
||||||
|
// subtype the cache does not catch) must not fault this
|
||||||
|
// BackgroundService and stop the whole host. The cache records
|
||||||
|
// its own Unavailable/Stale status; the periodic tick below retries.
|
||||||
|
logger.LogWarning(exception, "Initial Galaxy hierarchy cache load failed; will retry on the refresh interval.");
|
||||||
|
}
|
||||||
|
|
||||||
|
using PeriodicTimer timer = new(interval, _timeProvider);
|
||||||
|
try
|
||||||
|
{
|
||||||
|
while (await timer.WaitForNextTickAsync(stoppingToken).ConfigureAwait(false))
|
||||||
|
{
|
||||||
|
try
|
||||||
|
{
|
||||||
|
await cache.RefreshAsync(stoppingToken).ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
|
||||||
|
{
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
catch (Exception exception)
|
||||||
|
{
|
||||||
|
logger.LogWarning(exception, "Galaxy hierarchy cache refresh tick failed.");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
|
||||||
|
{
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,24 @@
|
|||||||
|
namespace ZB.MOM.WW.GalaxyRepository;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// A serializable point-in-time copy of the Galaxy Repository browse data.
|
||||||
|
/// Holds the raw hierarchy and attribute rowsets — not the materialized
|
||||||
|
/// protobuf objects — so the restore path runs the exact same
|
||||||
|
/// materialization as a live refresh. Persisted by
|
||||||
|
/// <see cref="IGalaxyHierarchySnapshotStore"/> after a successful refresh
|
||||||
|
/// and reloaded at startup when the Galaxy database is unreachable.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="LastDeployTime">
|
||||||
|
/// The <c>galaxy.time_of_last_deploy</c> the rowsets were pulled at, or
|
||||||
|
/// <see langword="null"/> when the Galaxy table reported no deploy. A later
|
||||||
|
/// live refresh that observes this same timestamp can promote the restored
|
||||||
|
/// entry to healthy without re-running the heavy queries.
|
||||||
|
/// </param>
|
||||||
|
/// <param name="SavedAt">UTC wall-clock when the snapshot was written to disk.</param>
|
||||||
|
/// <param name="Hierarchy">The persisted object-hierarchy rowset.</param>
|
||||||
|
/// <param name="Attributes">The persisted attribute rowset.</param>
|
||||||
|
public sealed record GalaxyHierarchySnapshot(
|
||||||
|
DateTimeOffset? LastDeployTime,
|
||||||
|
DateTimeOffset SavedAt,
|
||||||
|
IReadOnlyList<GalaxyHierarchyRow> Hierarchy,
|
||||||
|
IReadOnlyList<GalaxyAttributeRow> Attributes);
|
||||||
+143
@@ -0,0 +1,143 @@
|
|||||||
|
using System.Text.Json;
|
||||||
|
using Microsoft.Extensions.Logging;
|
||||||
|
using Microsoft.Extensions.Options;
|
||||||
|
|
||||||
|
namespace ZB.MOM.WW.GalaxyRepository;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// JSON-file implementation of <see cref="IGalaxyHierarchySnapshotStore"/>.
|
||||||
|
/// Writes the on-disk snapshot atomically (temp file + rename) so a crash
|
||||||
|
/// mid-write can never leave a torn file, and ignores files whose schema
|
||||||
|
/// version it does not recognize. When
|
||||||
|
/// <see cref="GalaxyRepositoryOptions.PersistSnapshot"/> is <see langword="false"/>
|
||||||
|
/// — or <see cref="GalaxyRepositoryOptions.SnapshotCachePath"/> is empty —
|
||||||
|
/// both operations are no-ops. The snapshot path is fully consumer-supplied;
|
||||||
|
/// this store imposes no platform-specific default, so it is cross-platform.
|
||||||
|
/// </summary>
|
||||||
|
public sealed class GalaxyHierarchySnapshotStore : IGalaxyHierarchySnapshotStore
|
||||||
|
{
|
||||||
|
/// <summary>
|
||||||
|
/// On-disk format version. Bump this whenever the persisted shape changes
|
||||||
|
/// in a way an older or newer consumer cannot read; a mismatched file is
|
||||||
|
/// ignored rather than misparsed.
|
||||||
|
/// </summary>
|
||||||
|
private const int CurrentSchemaVersion = 1;
|
||||||
|
|
||||||
|
private static readonly JsonSerializerOptions SerializerOptions = new()
|
||||||
|
{
|
||||||
|
WriteIndented = false,
|
||||||
|
};
|
||||||
|
|
||||||
|
private readonly string? _path;
|
||||||
|
private readonly TimeSpan _writeTimeout;
|
||||||
|
private readonly ILogger<GalaxyHierarchySnapshotStore>? _logger;
|
||||||
|
private readonly SemaphoreSlim _ioGate = new(1, 1);
|
||||||
|
|
||||||
|
/// <summary>Initializes a new instance of the <see cref="GalaxyHierarchySnapshotStore"/> class.</summary>
|
||||||
|
/// <param name="options">Galaxy repository options carrying the snapshot path and enable flag.</param>
|
||||||
|
/// <param name="logger">Optional logger for diagnostic output.</param>
|
||||||
|
public GalaxyHierarchySnapshotStore(
|
||||||
|
IOptions<GalaxyRepositoryOptions> options,
|
||||||
|
ILogger<GalaxyHierarchySnapshotStore>? logger = null)
|
||||||
|
{
|
||||||
|
GalaxyRepositoryOptions value = options.Value;
|
||||||
|
_path = value.PersistSnapshot && !string.IsNullOrWhiteSpace(value.SnapshotCachePath)
|
||||||
|
? value.SnapshotCachePath
|
||||||
|
: null;
|
||||||
|
_writeTimeout = TimeSpan.FromSeconds(Math.Max(1, value.CommandTimeoutSeconds));
|
||||||
|
_logger = logger;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public async Task SaveAsync(GalaxyHierarchySnapshot snapshot, CancellationToken cancellationToken)
|
||||||
|
{
|
||||||
|
ArgumentNullException.ThrowIfNull(snapshot);
|
||||||
|
if (_path is null)
|
||||||
|
{
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
PersistedFile file = new(CurrentSchemaVersion, snapshot);
|
||||||
|
|
||||||
|
await _ioGate.WaitAsync(cancellationToken).ConfigureAwait(false);
|
||||||
|
try
|
||||||
|
{
|
||||||
|
// Bound the write so a stuck disk — e.g. a SnapshotCachePath on an
|
||||||
|
// unresponsive network share — cannot stall the caller. On the cache
|
||||||
|
// refresh path that would otherwise pin the whole refresh loop.
|
||||||
|
using CancellationTokenSource writeCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
|
||||||
|
writeCts.CancelAfter(_writeTimeout);
|
||||||
|
|
||||||
|
string? directory = Path.GetDirectoryName(_path);
|
||||||
|
if (!string.IsNullOrEmpty(directory))
|
||||||
|
{
|
||||||
|
Directory.CreateDirectory(directory);
|
||||||
|
}
|
||||||
|
|
||||||
|
string tempPath = _path + ".tmp";
|
||||||
|
await using (FileStream stream = new(tempPath, FileMode.Create, FileAccess.Write, FileShare.None))
|
||||||
|
{
|
||||||
|
await JsonSerializer.SerializeAsync(stream, file, SerializerOptions, writeCts.Token).ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
|
||||||
|
File.Move(tempPath, _path, overwrite: true);
|
||||||
|
_logger?.LogDebug(
|
||||||
|
"Persisted Galaxy hierarchy snapshot to {Path} ({ObjectCount} objects, {AttributeCount} attributes).",
|
||||||
|
_path,
|
||||||
|
snapshot.Hierarchy.Count,
|
||||||
|
snapshot.Attributes.Count);
|
||||||
|
}
|
||||||
|
finally
|
||||||
|
{
|
||||||
|
_ioGate.Release();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public async Task<GalaxyHierarchySnapshot?> TryLoadAsync(CancellationToken cancellationToken)
|
||||||
|
{
|
||||||
|
if (_path is null || !File.Exists(_path))
|
||||||
|
{
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
|
||||||
|
await _ioGate.WaitAsync(cancellationToken).ConfigureAwait(false);
|
||||||
|
try
|
||||||
|
{
|
||||||
|
PersistedFile? file;
|
||||||
|
await using (FileStream stream = new(_path, FileMode.Open, FileAccess.Read, FileShare.Read))
|
||||||
|
{
|
||||||
|
file = await JsonSerializer.DeserializeAsync<PersistedFile>(
|
||||||
|
stream, SerializerOptions, cancellationToken).ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
|
||||||
|
if (file is null || file.SchemaVersion != CurrentSchemaVersion || file.Snapshot is null)
|
||||||
|
{
|
||||||
|
_logger?.LogWarning(
|
||||||
|
"Ignoring Galaxy hierarchy snapshot at {Path}: unrecognized or empty schema version.",
|
||||||
|
_path);
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
|
||||||
|
return file.Snapshot;
|
||||||
|
}
|
||||||
|
catch (Exception exception) when (exception is JsonException or IOException or UnauthorizedAccessException)
|
||||||
|
{
|
||||||
|
// A corrupt, truncated, locked, or access-denied snapshot file is an
|
||||||
|
// expected failure mode for a disk cache — honor the Try contract and
|
||||||
|
// return null rather than throwing.
|
||||||
|
_logger?.LogWarning(
|
||||||
|
exception,
|
||||||
|
"Ignoring Galaxy hierarchy snapshot at {Path}: the file is unreadable or not valid JSON.",
|
||||||
|
_path);
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
finally
|
||||||
|
{
|
||||||
|
_ioGate.Release();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>On-disk envelope: a schema version plus the snapshot payload.</summary>
|
||||||
|
private sealed record PersistedFile(int SchemaVersion, GalaxyHierarchySnapshot? Snapshot);
|
||||||
|
}
|
||||||
@@ -0,0 +1,16 @@
|
|||||||
|
using ZB.MOM.WW.GalaxyRepository.Grpc;
|
||||||
|
|
||||||
|
namespace ZB.MOM.WW.GalaxyRepository;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// A <see cref="GalaxyObject"/> paired with its computed contained path and hierarchy
|
||||||
|
/// depth. Materialized once per cache entry by <see cref="GalaxyHierarchyIndex"/> so
|
||||||
|
/// browse/discover projection can filter and page without recomputing paths.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="Object">The projected Galaxy object.</param>
|
||||||
|
/// <param name="ContainedPath">The slash-delimited contained path from the hierarchy root.</param>
|
||||||
|
/// <param name="Depth">The number of path segments from the root (zero for top-level objects).</param>
|
||||||
|
public sealed record GalaxyObjectView(
|
||||||
|
GalaxyObject Object,
|
||||||
|
string ContainedPath,
|
||||||
|
int Depth);
|
||||||
@@ -0,0 +1,76 @@
|
|||||||
|
using ZB.MOM.WW.GalaxyRepository.Grpc;
|
||||||
|
|
||||||
|
namespace ZB.MOM.WW.GalaxyRepository;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Maps <see cref="GalaxyHierarchyRow"/> + <see cref="GalaxyAttributeRow"/> rows produced
|
||||||
|
/// by <see cref="GalaxyRepository"/> into <c>galaxy_repository.v1</c> proto messages.
|
||||||
|
/// Pure function, separated so it can be unit-tested without a SQL connection.
|
||||||
|
/// </summary>
|
||||||
|
public static class GalaxyProtoMapper
|
||||||
|
{
|
||||||
|
/// <summary>Maps Galaxy hierarchy and attribute rows to Galaxy object protos.</summary>
|
||||||
|
/// <param name="hierarchy">Hierarchy rows from Galaxy Repository.</param>
|
||||||
|
/// <param name="attributes">Attribute rows from Galaxy Repository.</param>
|
||||||
|
public static IEnumerable<GalaxyObject> MapHierarchy(
|
||||||
|
IReadOnlyList<GalaxyHierarchyRow> hierarchy,
|
||||||
|
IReadOnlyList<GalaxyAttributeRow> attributes)
|
||||||
|
{
|
||||||
|
Dictionary<int, List<GalaxyAttributeRow>> attributesByGobjectId = attributes
|
||||||
|
.GroupBy(a => a.GobjectId)
|
||||||
|
.ToDictionary(g => g.Key, g => g.ToList());
|
||||||
|
|
||||||
|
foreach (GalaxyHierarchyRow row in hierarchy)
|
||||||
|
{
|
||||||
|
yield return MapObject(row, attributesByGobjectId);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>Maps a Galaxy hierarchy row to a Galaxy object proto.</summary>
|
||||||
|
/// <param name="row">Hierarchy row from Galaxy Repository.</param>
|
||||||
|
/// <param name="attributesByGobjectId">Attributes indexed by gobject ID.</param>
|
||||||
|
public static GalaxyObject MapObject(
|
||||||
|
GalaxyHierarchyRow row,
|
||||||
|
IReadOnlyDictionary<int, List<GalaxyAttributeRow>> attributesByGobjectId)
|
||||||
|
{
|
||||||
|
GalaxyObject obj = new()
|
||||||
|
{
|
||||||
|
GobjectId = row.GobjectId,
|
||||||
|
TagName = row.TagName,
|
||||||
|
ContainedName = row.ContainedName,
|
||||||
|
BrowseName = row.BrowseName,
|
||||||
|
ParentGobjectId = row.ParentGobjectId,
|
||||||
|
IsArea = row.IsArea,
|
||||||
|
CategoryId = row.CategoryId,
|
||||||
|
HostedByGobjectId = row.HostedByGobjectId,
|
||||||
|
};
|
||||||
|
obj.TemplateChain.AddRange(row.TemplateChain);
|
||||||
|
|
||||||
|
if (attributesByGobjectId.TryGetValue(row.GobjectId, out List<GalaxyAttributeRow>? attrs))
|
||||||
|
{
|
||||||
|
foreach (GalaxyAttributeRow attr in attrs)
|
||||||
|
{
|
||||||
|
obj.Attributes.Add(MapAttribute(attr));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return obj;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>Maps a Galaxy attribute row to a Galaxy attribute proto.</summary>
|
||||||
|
/// <param name="row">Attribute row from Galaxy Repository.</param>
|
||||||
|
public static GalaxyAttribute MapAttribute(GalaxyAttributeRow row) => new()
|
||||||
|
{
|
||||||
|
AttributeName = row.AttributeName,
|
||||||
|
FullTagReference = row.FullTagReference,
|
||||||
|
MxDataType = row.MxDataType,
|
||||||
|
DataTypeName = row.DataTypeName ?? string.Empty,
|
||||||
|
IsArray = row.IsArray,
|
||||||
|
ArrayDimension = row.ArrayDimension ?? 0,
|
||||||
|
ArrayDimensionPresent = row.ArrayDimension.HasValue,
|
||||||
|
MxAttributeCategory = row.MxAttributeCategory,
|
||||||
|
SecurityClassification = row.SecurityClassification,
|
||||||
|
IsHistorized = row.IsHistorized,
|
||||||
|
IsAlarm = row.IsAlarm,
|
||||||
|
};
|
||||||
|
}
|
||||||
@@ -0,0 +1,16 @@
|
|||||||
|
using ZB.MOM.WW.GalaxyRepository.Grpc;
|
||||||
|
|
||||||
|
namespace ZB.MOM.WW.GalaxyRepository;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Resolution result for a tag address: the owning <see cref="GalaxyObject"/>, the
|
||||||
|
/// specific <see cref="GalaxyAttribute"/> when the address names an attribute (otherwise
|
||||||
|
/// <see langword="null"/>), and the object's contained path.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="Object">The Galaxy object that owns the looked-up address.</param>
|
||||||
|
/// <param name="Attribute">The matched attribute, or <see langword="null"/> when the address names an object.</param>
|
||||||
|
/// <param name="ContainedPath">The owning object's contained path.</param>
|
||||||
|
public sealed record GalaxyTagLookup(
|
||||||
|
GalaxyObject Object,
|
||||||
|
GalaxyAttribute? Attribute,
|
||||||
|
string ContainedPath);
|
||||||
@@ -0,0 +1,17 @@
|
|||||||
|
namespace ZB.MOM.WW.GalaxyRepository;
|
||||||
|
|
||||||
|
/// <summary>Publishes Galaxy repository deploy events to subscribers.</summary>
|
||||||
|
public interface IGalaxyDeployNotifier
|
||||||
|
{
|
||||||
|
/// <summary>The most recently published event, or null if no event has fired yet.</summary>
|
||||||
|
GalaxyDeployEventInfo? Latest { get; }
|
||||||
|
|
||||||
|
/// <summary>Publishes a deploy event to all current subscribers and stores it as Latest.</summary>
|
||||||
|
/// <param name="info">The deploy event to publish.</param>
|
||||||
|
void Publish(GalaxyDeployEventInfo info);
|
||||||
|
|
||||||
|
/// <summary>Subscribes to deploy events. The sequence yields the latest event first (if available) then streams new events as they fire.</summary>
|
||||||
|
/// <param name="cancellationToken">Token to cancel the asynchronous operation.</param>
|
||||||
|
/// <returns>Async enumerable of deploy events.</returns>
|
||||||
|
IAsyncEnumerable<GalaxyDeployEventInfo> SubscribeAsync(CancellationToken cancellationToken);
|
||||||
|
}
|
||||||
@@ -0,0 +1,25 @@
|
|||||||
|
namespace ZB.MOM.WW.GalaxyRepository;
|
||||||
|
|
||||||
|
/// <summary>Cache for Galaxy Repository hierarchy data.</summary>
|
||||||
|
public interface IGalaxyHierarchyCache
|
||||||
|
{
|
||||||
|
/// <summary>The latest cache entry. Status freshness is recomputed against the clock.</summary>
|
||||||
|
GalaxyHierarchyCacheEntry Current { get; }
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Forces a refresh against the Galaxy Repository. Performs a cheap
|
||||||
|
/// <c>time_of_last_deploy</c> probe first and only re-queries the heavy hierarchy +
|
||||||
|
/// attributes rowsets when the deploy time has changed since the last successful
|
||||||
|
/// refresh.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="cancellationToken">Token to cancel the asynchronous operation.</param>
|
||||||
|
Task RefreshAsync(CancellationToken cancellationToken);
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Awaits the first completed refresh attempt (success or failure). Useful for
|
||||||
|
/// gRPC handlers that want to serve from cache without returning Unavailable on the
|
||||||
|
/// very first request after the service starts.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="cancellationToken">Token to cancel the asynchronous operation.</param>
|
||||||
|
Task WaitForFirstLoadAsync(CancellationToken cancellationToken);
|
||||||
|
}
|
||||||
+28
@@ -0,0 +1,28 @@
|
|||||||
|
namespace ZB.MOM.WW.GalaxyRepository;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Persists the latest Galaxy Repository browse dataset to disk and reloads
|
||||||
|
/// it at startup. Lets <see cref="GalaxyHierarchyCache"/> serve last-known
|
||||||
|
/// browse data when the Galaxy database is unreachable on a cold start.
|
||||||
|
/// </summary>
|
||||||
|
public interface IGalaxyHierarchySnapshotStore
|
||||||
|
{
|
||||||
|
/// <summary>
|
||||||
|
/// Writes <paramref name="snapshot"/> to disk, replacing any previous
|
||||||
|
/// snapshot atomically. A no-op when snapshot persistence is disabled.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="snapshot">The browse dataset to persist.</param>
|
||||||
|
/// <param name="cancellationToken">Token to cancel the asynchronous operation.</param>
|
||||||
|
Task SaveAsync(GalaxyHierarchySnapshot snapshot, CancellationToken cancellationToken);
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Reads the persisted Galaxy browse dataset.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="cancellationToken">Token to cancel the asynchronous operation.</param>
|
||||||
|
/// <returns>
|
||||||
|
/// The persisted snapshot, or <see langword="null"/> when none exists,
|
||||||
|
/// persistence is disabled, or the on-disk file uses an unrecognized
|
||||||
|
/// schema version.
|
||||||
|
/// </returns>
|
||||||
|
Task<GalaxyHierarchySnapshot?> TryLoadAsync(CancellationToken cancellationToken);
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user