diff --git a/src/MeshWeaver.Hosting.PostgreSql/PostgreSqlMeshQuery.cs b/src/MeshWeaver.Hosting.PostgreSql/PostgreSqlMeshQuery.cs index 30155f06a..36b4d0cab 100644 --- a/src/MeshWeaver.Hosting.PostgreSql/PostgreSqlMeshQuery.cs +++ b/src/MeshWeaver.Hosting.PostgreSql/PostgreSqlMeshQuery.cs @@ -592,7 +592,7 @@ public IObservable> Query(MeshQueryRequest request, Json // The Initial snapshot itself uses request.Queries via QueryAsync, but the // change-detection layer must observe the union of all branches' shapes. var effectiveQueries = request.EffectiveQueries; - var parsedFilters = new List<(string BasePath, QueryScope Scope)>(effectiveQueries.Count); + var parsedFilters = new List<(string BasePath, QueryScope Scope, IReadOnlyList Namespaces)>(effectiveQueries.Count); ParsedQuery firstParsed = null!; for (var qi = 0; qi < effectiveQueries.Count; qi++) { @@ -606,7 +606,10 @@ public IObservable> Query(MeshQueryRequest request, Json effectiveScope = QueryScope.Children; } var normalizedBasePath = effectivePath?.Trim('/') ?? ""; - parsedFilters.Add((normalizedBasePath, effectiveScope)); + // Capture the query's NAMESPACE filters too: a namespace-only query (no path:) leaves + // BasePath empty, so the path/scope check alone judges deep changes out of scope. The + // namespaces let a delete/update/create under {owner}/*_Thread trigger a re-query. + parsedFilters.Add((normalizedBasePath, effectiveScope, pq.ExtractNamespacePatterns())); if (qi == 0) firstParsed = pq; } @@ -651,7 +654,7 @@ public IObservable> Query(MeshQueryRequest request, Json var earlySubscription = _adapter.Changes .Where(n => parsedFilters.Any(f => - PathMatcher.ShouldNotify(n.Path, f.BasePath, f.Scope))) + PathMatcher.ShouldNotifyForQuery(n.Path, f.BasePath, f.Scope, f.Namespaces))) .Subscribe(n => { lock (earlyLock) @@ -680,7 +683,7 @@ public IObservable> Query(MeshQueryRequest request, Json disposables.Add( _adapter.Changes .Where(n => parsedFilters.Any(f => - PathMatcher.ShouldNotify(n.Path, f.BasePath, f.Scope))) + PathMatcher.ShouldNotifyForQuery(n.Path, f.BasePath, f.Scope, f.Namespaces))) .Subscribe(changeBuffer)); // 🚨 Strict unit-of-work + zero debounce: every change // triggers its own RunQuery, serialised via Concat so the diff --git a/src/MeshWeaver.Hosting/Persistence/Query/PathMatcher.cs b/src/MeshWeaver.Hosting/Persistence/Query/PathMatcher.cs index f519897bf..7ec4ce366 100644 --- a/src/MeshWeaver.Hosting/Persistence/Query/PathMatcher.cs +++ b/src/MeshWeaver.Hosting/Persistence/Query/PathMatcher.cs @@ -138,4 +138,70 @@ private static string[] GetSegments(string path) return path.Split('/', StringSplitOptions.RemoveEmptyEntries); } + + /// + /// Relevance test for a live synced query: a CRUD change at should + /// trigger a re-query when EITHER its path matches the query's path + + /// (), OR the changed node's NAMESPACE matches one of the query's namespace + /// filters (). + /// The path-only is blind to NAMESPACE-filtered queries: the + /// "open threads" catalog queries namespace:{owner}/*_Thread with NO path: term, so its + /// base path is empty and ShouldNotify degrades to "direct children of root". A thread deleted three + /// levels deep ({owner}/_Thread/{id}) is then judged out of scope → no re-query → the catalog + /// never refreshes on delete. Matching the changed node's namespace (the path minus its last segment) + /// against the query namespaces closes that gap for create/update/delete alike. Broadening relevance + /// is correctness-safe: the worst case is a redundant re-query, never a missed change. + /// + public static bool ShouldNotifyForQuery( + string changedPath, string queryBasePath, QueryScope scope, IReadOnlyList? queryNamespaces) + => ShouldNotify(changedPath, queryBasePath, scope) + || NamespaceInScope(NamespaceOf(changedPath), queryNamespaces); + + /// The namespace a node at lives in — the path minus its last segment + /// (a thread {owner}/_Thread/{id} → {owner}/_Thread). Empty for a top-level node. + public static string NamespaceOf(string path) + { + var normalized = NormalizePath(path); + var i = normalized.LastIndexOf('/'); + return i > 0 ? normalized[..i] : ""; + } + + /// + /// True when matches any of . A pattern may carry a + /// single * glob ({owner}/*_Thread → starts-with the pre-* part AND ends-with the + /// post-* part); a pattern without * matches when equals it OR is a + /// sub-namespace under it (robust to whether the parser hands back the glob pattern or its base). + /// + public static bool NamespaceInScope(string ns, IReadOnlyList? patterns) + { + if (patterns is null || patterns.Count == 0 || string.IsNullOrEmpty(ns)) + return false; + var value = NormalizePath(ns); + for (var i = 0; i < patterns.Count; i++) + if (GlobMatch(value, NormalizePath(patterns[i]))) + return true; + return false; + } + + private static bool GlobMatch(string value, string pattern) + { + if (string.IsNullOrEmpty(pattern)) + return false; + // The query parser emits a namespace wildcard as SQL-LIKE '%' (it rewrites the user's '*' → + // '%'); accept either. A single wildcard splits the pattern into a prefix + suffix (e.g. + // "{owner}/%_Thread" → "{owner}/" … "_Thread"); the surrounding literals (incl. the satellite + // "_" in "_Thread") are matched verbatim, which is what real satellite namespaces need. + var star = pattern.IndexOfAny(['*', '%']); + if (star >= 0) + { + var prefix = pattern[..star]; + var suffix = pattern[(star + 1)..]; + return value.Length >= prefix.Length + suffix.Length + && value.StartsWith(prefix, StringComparison.OrdinalIgnoreCase) + && value.EndsWith(suffix, StringComparison.OrdinalIgnoreCase); + } + // No glob: exact namespace, or a sub-namespace under it. + return string.Equals(value, pattern, StringComparison.OrdinalIgnoreCase) + || value.StartsWith(pattern + "/", StringComparison.OrdinalIgnoreCase); + } } diff --git a/src/MeshWeaver.Mesh.Contract/Query/ParsedQuery.cs b/src/MeshWeaver.Mesh.Contract/Query/ParsedQuery.cs index e88066681..c3421391a 100644 --- a/src/MeshWeaver.Mesh.Contract/Query/ParsedQuery.cs +++ b/src/MeshWeaver.Mesh.Contract/Query/ParsedQuery.cs @@ -94,6 +94,39 @@ private static void ExtractNamespacesFromNode(QueryNode node, List colle } } + /// + /// Namespace filter values INCLUDING wildcard (Like) patterns — e.g. {owner}/*_Thread. + /// deliberately collects only concrete Equal/In + /// namespaces (the aggregator ROUTES on those, and can't route a glob); this variant ALSO returns the + /// Like glob patterns, for the live change-relevance filter that decides whether a CRUD event + /// under a namespace should refresh a synced query (see PathMatcher.ShouldNotifyForQuery). The + /// glob is preserved verbatim (the parser keeps the *), so a matcher can apply it directly. + /// + public IReadOnlyList ExtractNamespacePatterns() + { + if (Filter == null) return Array.Empty(); + var collected = new List(); + ExtractNamespacePatternsFromNode(Filter, collected); + return collected; + } + + private static void ExtractNamespacePatternsFromNode(QueryNode node, List collected) + { + switch (node) + { + case QueryComparison c when c.Condition.Selector.Equals("namespace", StringComparison.OrdinalIgnoreCase) + && c.Condition.Operator is QueryOperator.Equal or QueryOperator.In or QueryOperator.Like: + collected.AddRange(c.Condition.Values); + break; + case QueryAnd and: + foreach (var child in and.Children) ExtractNamespacePatternsFromNode(child, collected); + break; + case QueryOr or: + foreach (var child in or.Children) ExtractNamespacePatternsFromNode(child, collected); + break; + } + } + /// /// Projects an item down to only the requested properties. /// Returns a dictionary with the selected property names and their values. diff --git a/test/MeshWeaver.Hosting.Test/QueryChangeRelevanceTest.cs b/test/MeshWeaver.Hosting.Test/QueryChangeRelevanceTest.cs new file mode 100644 index 000000000..d89396826 --- /dev/null +++ b/test/MeshWeaver.Hosting.Test/QueryChangeRelevanceTest.cs @@ -0,0 +1,90 @@ +using System.Collections.Generic; +using MeshWeaver.Hosting.Persistence.Query; +using MeshWeaver.Mesh; +using MeshWeaver.Mesh.Services; +using Xunit; + +namespace MeshWeaver.Hosting.Test; + +/// +/// The reactive change-feed RELEVANCE test — "does this CRUD broadcast event fit the query, so we should +/// re-run it?" (). Pins the fix for the "open threads" +/// catalog not refreshing on delete. +/// +/// The bug. The catalog queries by NAMESPACE — namespace:{owner}/*_Thread with NO +/// path: term — so its base path is empty and the old path-only +/// degraded to "direct children of root". A thread deleted three levels deep at +/// {owner}/_Thread/{id} was judged out of scope → no re-query → the catalog kept showing the +/// deleted thread. also matches the changed node's +/// namespace against the query namespaces, so create / update / delete of a thread all refresh it. +/// +/// Pure logic (path + parsed query), so every CRUD case is a unit test — no Postgres, no change +/// feed. The PG provider only supplies the path in its broadcast events (pg_notify carries the key, not +/// the row), which is exactly what this test drives. +/// +public class QueryChangeRelevanceTest +{ + private static readonly QueryParser Parser = new(); + private const string Owner = "rbuergi"; + private const string ThreadPath = Owner + "/_Thread/do-you-have-guardrails-6b4c"; + + // The exact "open threads" catalog query (UserActivityLayoutAreas.BuildOpenThreads). + private static (string BasePath, QueryScope Scope, IReadOnlyList Namespaces) CatalogFilter() + { + var pq = Parser.Parse($"namespace:{Owner}/*_Thread nodeType:Thread -content.status:Done sort:LastModified-desc"); + // Namespace-only query → no path: term → empty base path (mirrors PostgreSqlMeshQuery's derivation). + return (pq.Path ?? "", pq.Scope, pq.ExtractNamespacePatterns()); + } + + private static bool InScope(DataChangeNotification change) + { + var (basePath, scope, namespaces) = CatalogFilter(); + return PathMatcher.ShouldNotifyForQuery(change.Path, basePath, scope, namespaces); + } + + // ── All CRUD, in-scope (a thread under {owner}/_Thread) → must refresh the catalog ── + + [Fact] + public void Created_thread_refreshes_catalog() + => Assert.True(InScope(DataChangeNotification.Created(ThreadPath, entity: null))); + + [Fact] + public void Updated_thread_refreshes_catalog() // e.g. status → Done: re-query then drops it (leave-scope) + => Assert.True(InScope(DataChangeNotification.Updated(ThreadPath, entity: null))); + + [Fact] + public void Deleted_thread_refreshes_catalog() // ← the reported bug + => Assert.True(InScope(DataChangeNotification.Deleted(ThreadPath))); + + // ── Out of scope → must NOT trigger a redundant re-query ── + + [Fact] + public void Change_to_non_thread_node_same_partition_is_out_of_scope() + => Assert.False(InScope(DataChangeNotification.Updated($"{Owner}/Docs/readme", entity: null))); + + [Fact] + public void Delete_of_non_thread_node_same_partition_is_out_of_scope() + => Assert.False(InScope(DataChangeNotification.Deleted($"{Owner}/Docs/readme"))); + + [Fact] + public void Change_in_a_different_partition_is_out_of_scope() + => Assert.False(InScope(DataChangeNotification.Deleted("acme/_Thread/other"))); + + // ── The namespace glob directly ── + + [Fact] + public void Namespace_glob_matches_satellite_thread_namespace() + => Assert.True(PathMatcher.NamespaceInScope($"{Owner}/_Thread", new[] { $"{Owner}/*_Thread" })); + + [Fact] + public void Namespace_glob_rejects_non_matching_namespace() + => Assert.False(PathMatcher.NamespaceInScope($"{Owner}/Docs", new[] { $"{Owner}/*_Thread" })); + + [Fact] + public void NamespaceOf_strips_the_last_segment() + => Assert.Equal($"{Owner}/_Thread", PathMatcher.NamespaceOf(ThreadPath)); + + [Fact] + public void Namespace_glob_matches_the_parser_emitted_like_pattern() // parser rewrites * → SQL-LIKE % + => Assert.True(PathMatcher.NamespaceInScope($"{Owner}/_Thread", new[] { $"{Owner}/%_Thread" })); +}