diff --git a/README.md b/README.md index 39ae54b..1715975 100644 --- a/README.md +++ b/README.md @@ -346,6 +346,121 @@ public class CrmCustomerResolver : IExternalResolver } ``` +Example: resolve an address from an external source by its id: + +```json +{ + "entityType": "customer", + "pipeline": [ + { + "project": { + "id": 1, + "name": 1, + "attributes.addressId": 1 + } + }, + { + "resolve": { + "source": "crm.address", + "localPath": "attributes.addressId", + "as": "address" + } + }, + { + "page": { + "limit": 25 + } + } + ] +} +``` + +If a customer document contains `attributes.addressId = "addr-42"`, the resolver registered for `crm.address` receives `"addr-42"` and the resolved address object is added to the result as `address`. + +Example: complete OxQL query with one cross-service resolve stage against `contact-api/v1`: + +```json +{ + "entityType": "invoice", + "pipeline": [ + { + "match": { + "id": { "eq": "invoice-1001" } + } + }, + { + "project": { + "id": 1, + "entityType": 1, + "attributes.contactId": 1, + "attributes.totalAmount": 1 + } + }, + { + "resolve": { + "source": "contact-api/v1", + "localPath": "attributes.contactId", + "as": "contact" + } + }, + { + "page": { + "limit": 1 + } + } + ] +} +``` + +When this query is executed with `services=contact-api/v1=https://contact-api.internal/`, and the invoice contains `attributes.contactId = "contact-42"`, the resolver calls `contact-api/v1` to load `contact-42` by id and adds the returned contact object to the result as `contact`. + +Example: resolve with a parameterized subquery sent to `contact-api/v1`: + +```json +{ + "entityType": "invoice", + "pipeline": [ + { + "match": { + "id": { "eq": "invoice-1001" } + } + }, + { + "resolve": { + "source": "contact-api/v1", + "localPath": "attributes.contactId", + "parameters": { + "contactId": "attributes.contactId" + }, + "subquery": { + "entityType": "contact", + "pipeline": [ + { + "match": { + "id": { "eq": { "$var": "contactId" } } + } + }, + { + "page": { + "limit": 1 + } + } + ] + }, + "as": "contact" + } + }, + { + "page": { + "limit": 1 + } + } + ] +} +``` + +The resolver uses `parameters` to map local document values into subquery variables, then forwards the `subquery` to the remote OxQL endpoint. + ### IQueryAdapter Implement for non-MongoDB backends: diff --git a/src/.github/copilot-instructions.md b/src/.github/copilot-instructions.md new file mode 100644 index 0000000..83db254 --- /dev/null +++ b/src/.github/copilot-instructions.md @@ -0,0 +1,4 @@ +# Copilot Instructions + +## Project Guidelines +- When the user asks for a resolver example, 'dynamic resolver system' refers to a cross-service resolver setup, not Microsoft Dynamics. \ No newline at end of file diff --git a/src/OxQL.Core/Interfaces/IExternalResolver.cs b/src/OxQL.Core/Interfaces/IExternalResolver.cs index 5aacc50..dddf6d2 100644 --- a/src/OxQL.Core/Interfaces/IExternalResolver.cs +++ b/src/OxQL.Core/Interfaces/IExternalResolver.cs @@ -1,5 +1,23 @@ +using OxQL.Core.Models; + namespace OxQL.Core.Interfaces; +/// +/// Describes a resolve request sent to an . +/// +public sealed record ExternalResolveRequest +{ + /// + /// The keys to resolve in key-based mode. + /// + public IReadOnlyList? Keys { get; init; } + + /// + /// The query request to execute in subquery mode. + /// + public QueryRequest? Query { get; init; } +} + /// /// Resolves data from external sources (e.g., CRM, external APIs). /// @@ -19,4 +37,31 @@ public interface IExternalResolver Task> ResolveAsync( IReadOnlyList keys, CancellationToken cancellationToken = default); + + /// + /// Resolves one object using either key-based or query-based request context. + /// + /// + /// Default behavior maps key-based requests to . + /// Override this in resolvers that support subquery forwarding. + /// + async Task ResolveOneAsync( + ExternalResolveRequest request, + CancellationToken cancellationToken = default) + { + ArgumentNullException.ThrowIfNull(request); + + if (request.Query is not null) + { + throw new NotSupportedException( + $"Resolver '{Source}' does not support query-based resolve requests."); + } + + if (request.Keys is null || request.Keys.Count == 0) + return null; + + var resolved = await ResolveAsync(request.Keys, cancellationToken); + var firstKey = request.Keys[0]; + return resolved.TryGetValue(firstKey, out var value) ? value : null; + } } diff --git a/src/OxQL.Core/Models/LookupStage.cs b/src/OxQL.Core/Models/LookupStage.cs index b00b5c2..c4658b1 100644 --- a/src/OxQL.Core/Models/LookupStage.cs +++ b/src/OxQL.Core/Models/LookupStage.cs @@ -49,6 +49,18 @@ public sealed record ResolveStage [JsonPropertyName("localPath")] public required string LocalPath { get; init; } + /// + /// Optional parameter mappings for subquery mode (variable name -> local path). + /// + [JsonPropertyName("parameters")] + public IReadOnlyDictionary? Parameters { get; init; } + + /// + /// Optional subquery sent to the external source. + /// + [JsonPropertyName("subquery")] + public QueryRequest? Subquery { get; init; } + /// /// The alias for the resolved result. /// diff --git a/src/OxQL.Core/Validation/QueryValidator.cs b/src/OxQL.Core/Validation/QueryValidator.cs index 6c53f10..7419c95 100644 --- a/src/OxQL.Core/Validation/QueryValidator.cs +++ b/src/OxQL.Core/Validation/QueryValidator.cs @@ -255,6 +255,45 @@ private void ValidateResolve(ResolveStage resolve, List er ValidatePath(resolve.LocalPath, "resolve.localPath", errors); + if (resolve.Parameters is not null) + { + foreach (var (name, path) in resolve.Parameters) + { + if (string.IsNullOrWhiteSpace(name)) + { + errors.Add(new QueryValidationError + { + Code = "INVALID_RESOLVE_PARAMETER_NAME", + Message = "Resolve parameter name cannot be empty." + }); + continue; + } + + ValidatePath(path, $"resolve.parameters.{name}", errors); + } + } + + if (resolve.Subquery is not null) + { + if (string.IsNullOrWhiteSpace(resolve.Subquery.EntityType)) + { + errors.Add(new QueryValidationError + { + Code = "INVALID_RESOLVE_SUBQUERY_ENTITY", + Message = "Resolve subquery entityType is required." + }); + } + + if (resolve.Subquery.Pipeline is null || resolve.Subquery.Pipeline.Count == 0) + { + errors.Add(new QueryValidationError + { + Code = "INVALID_RESOLVE_SUBQUERY_PIPELINE", + Message = "Resolve subquery pipeline must contain at least one stage." + }); + } + } + if (string.IsNullOrWhiteSpace(resolve.As)) { errors.Add(new QueryValidationError diff --git a/src/OxQL.Mongo/MongoQueryAdapter.cs b/src/OxQL.Mongo/MongoQueryAdapter.cs index 09d885d..b67f1f6 100644 --- a/src/OxQL.Mongo/MongoQueryAdapter.cs +++ b/src/OxQL.Mongo/MongoQueryAdapter.cs @@ -1,3 +1,4 @@ +using System.Globalization; using OxQL.Core.Cursor; using OxQL.Core.Interfaces; using OxQL.Core.Models; @@ -14,13 +15,20 @@ public sealed class MongoQueryAdapter : IQueryAdapter { private readonly ICursorSerializer _cursorSerializer; private readonly Func> _collectionResolver; + private readonly IReadOnlyDictionary _externalResolvers; + private readonly Func>> _executor; public MongoQueryAdapter( Func> collectionResolver, - ICursorSerializer cursorSerializer) + ICursorSerializer cursorSerializer, + IEnumerable? externalResolvers = null, + Func>>? executor = null) { _collectionResolver = collectionResolver ?? throw new ArgumentNullException(nameof(collectionResolver)); _cursorSerializer = cursorSerializer ?? throw new ArgumentNullException(nameof(cursorSerializer)); + _externalResolvers = (externalResolvers ?? []) + .ToDictionary(resolver => resolver.Source, StringComparer.OrdinalIgnoreCase); + _executor = executor ?? ExecuteAggregationAsync; } public async Task> ExecuteAsync( @@ -38,19 +46,7 @@ public async Task> ExecuteAsync( cursorPayload = _cursorSerializer.Deserialize(pageStage.Cursor, plan.Sort); } - // Build main pipeline - var pipeline = pipelineBuilder.Build(plan, cursorPayload); - - // Resolve the collection for this entity type - var collection = _collectionResolver(plan.EntityType); - - // Execute aggregation - var pipelineDef = pipeline.Select(doc => (PipelineStageDefinition)doc).ToList(); - var aggPipeline = PipelineDefinition.Create(pipelineDef); - - var results = await collection - .Aggregate(aggPipeline, cancellationToken: cancellationToken) - .ToListAsync(cancellationToken); + var results = await _executor(plan, variables, cancellationToken); // Determine if there's a next page (we fetched limit+1) var hasNextPage = results.Count > pageStage.Limit; @@ -77,6 +73,7 @@ public async Task> ExecuteAsync( var countPipeline = pipelineBuilder.BuildCountPipeline(plan); var countPipelineDef = countPipeline.Select(doc => (PipelineStageDefinition)doc).ToList(); var countAgg = PipelineDefinition.Create(countPipelineDef); + var collection = _collectionResolver(plan.EntityType); var countResult = await collection .Aggregate(countAgg, cancellationToken: cancellationToken) @@ -92,6 +89,8 @@ public async Task> ExecuteAsync( } } + results = await ApplyResolveStagesAsync(results, plan.Pipeline, cancellationToken); + return new QueryResponse { Items = results, @@ -104,6 +103,266 @@ public async Task> ExecuteAsync( }; } + private async Task> ExecuteAggregationAsync( + QueryPlan plan, + QueryVariables? variables, + CancellationToken cancellationToken) + { + var pipelineBuilder = new MongoPipelineBuilder(variables); + + CursorPayload? cursorPayload = null; + var pageStage = plan.Page; + if (!string.IsNullOrEmpty(pageStage.Cursor)) + { + cursorPayload = _cursorSerializer.Deserialize(pageStage.Cursor, plan.Sort); + } + + var pipeline = pipelineBuilder.Build(plan, cursorPayload); + var collection = _collectionResolver(plan.EntityType); + var pipelineDef = pipeline.Select(doc => (PipelineStageDefinition)doc).ToList(); + var aggPipeline = PipelineDefinition.Create(pipelineDef); + + return await collection + .Aggregate(aggPipeline, cancellationToken: cancellationToken) + .ToListAsync(cancellationToken); + } + + private async Task> ApplyResolveStagesAsync( + List documents, + IReadOnlyList pipeline, + CancellationToken cancellationToken) + { + var resolveStages = pipeline + .Where(stage => stage.Resolve is not null) + .Select(stage => stage.Resolve!) + .ToList(); + + if (documents.Count == 0 || resolveStages.Count == 0) + return documents; + + var cache = new Dictionary>( + StringComparer.OrdinalIgnoreCase); + + foreach (var resolveStage in resolveStages) + { + await ApplyResolveStageAsync(documents, resolveStage, cache, cancellationToken); + } + + return documents; + } + + private async Task ApplyResolveStageAsync( + List documents, + ResolveStage resolveStage, + Dictionary> cache, + CancellationToken cancellationToken) + { + if (!_externalResolvers.TryGetValue(resolveStage.Source, out var resolver)) + { + throw new QueryValidationException( + $"No external resolver is registered for source '{resolveStage.Source}'."); + } + + if (!cache.TryGetValue(resolveStage.Source, out var sourceCache)) + { + sourceCache = new Dictionary(StringComparer.Ordinal); + cache[resolveStage.Source] = sourceCache; + } + + if (resolveStage.Subquery is null) + { + await ResolveByKeysAsync(documents, resolveStage, resolver, sourceCache, cancellationToken); + return; + } + + await ResolveBySubqueryAsync(documents, resolveStage, resolver, sourceCache, cancellationToken); + } + + private static async Task ResolveByKeysAsync( + List documents, + ResolveStage resolveStage, + IExternalResolver resolver, + Dictionary sourceCache, + CancellationToken cancellationToken) + { + var keys = documents + .Select(document => GetResolverKey(document, resolveStage.LocalPath)) + .Where(key => !string.IsNullOrWhiteSpace(key)) + .Distinct(StringComparer.Ordinal) + .ToList()!; + + var missingKeys = keys + .Where(key => !sourceCache.ContainsKey(key)) + .ToList(); + + if (missingKeys.Count > 0) + { + var resolvedValues = await resolver.ResolveAsync(missingKeys, cancellationToken); + + foreach (var key in missingKeys) + { + sourceCache[key] = resolvedValues.TryGetValue(key, out var value) + ? new ExternalResolutionCacheEntry(true, value) + : new ExternalResolutionCacheEntry(false, null); + } + } + + foreach (var document in documents) + { + var key = GetResolverKey(document, resolveStage.LocalPath); + var value = key is not null && sourceCache.TryGetValue(key, out var entry) && entry.Found + ? entry.Value + : null; + + SetFieldValue(document, resolveStage.As, value); + } + } + + private static async Task ResolveBySubqueryAsync( + List documents, + ResolveStage resolveStage, + IExternalResolver resolver, + Dictionary sourceCache, + CancellationToken cancellationToken) + { + foreach (var document in documents) + { + var request = BuildSubqueryRequest(document, resolveStage); + var cacheKey = BuildSubqueryCacheKey(request); + + if (!sourceCache.TryGetValue(cacheKey, out var entry)) + { + var value = await resolver.ResolveOneAsync(request, cancellationToken); + entry = value is null + ? new ExternalResolutionCacheEntry(false, null) + : new ExternalResolutionCacheEntry(true, value); + sourceCache[cacheKey] = entry; + } + + SetFieldValue(document, resolveStage.As, entry.Found ? entry.Value : null); + } + } + + private static ExternalResolveRequest BuildSubqueryRequest(BsonDocument document, ResolveStage resolveStage) + { + var baseSubquery = resolveStage.Subquery + ?? throw new QueryValidationException("Resolve subquery configuration is missing."); + + var variableValues = new Dictionary(StringComparer.OrdinalIgnoreCase); + if (resolveStage.Parameters is not null) + { + foreach (var (name, path) in resolveStage.Parameters) + { + variableValues[name] = GetFieldValue(document, path); + } + } + + var subquery = new QueryRequest + { + EntityType = baseSubquery.EntityType, + Pipeline = baseSubquery.Pipeline, + Variables = variableValues.Count == 0 + ? baseSubquery.Variables + : MergeVariables(baseSubquery.Variables, variableValues) + }; + + return new ExternalResolveRequest + { + Query = subquery, + Keys = null + }; + } + + private static QueryVariables MergeVariables(QueryVariables? baseVariables, Dictionary overrides) + { + var values = new Dictionary(StringComparer.OrdinalIgnoreCase); + + if (baseVariables is not null) + { + foreach (var (name, value) in baseVariables.Values) + { + values[name] = value; + } + } + + foreach (var (name, value) in overrides) + { + values[name] = value; + } + + return new QueryVariables { Values = values }; + } + + private static string BuildSubqueryCacheKey(ExternalResolveRequest request) + { + var query = request.Query + ?? throw new QueryValidationException("Resolve subquery request is missing query payload."); + + var serializedVariables = query.Variables is null + ? string.Empty + : string.Join("|", query.Variables.Values + .OrderBy(kv => kv.Key, StringComparer.OrdinalIgnoreCase) + .Select(kv => $"{kv.Key}:{SerializeCacheValue(kv.Value)}")); + + return $"subquery:{query.EntityType}:{serializedVariables}"; + } + + private static string SerializeCacheValue(object? value) + { + return value switch + { + null => "null", + IFormattable formattable => formattable.ToString(null, CultureInfo.InvariantCulture), + _ => value.ToString() ?? string.Empty + }; + } + + private static string? GetResolverKey(BsonDocument doc, string path) + { + var value = GetFieldValue(doc, path); + return value switch + { + null => null, + string s when string.IsNullOrWhiteSpace(s) => null, + IFormattable formattable => formattable.ToString(null, CultureInfo.InvariantCulture), + _ => value.ToString() + }; + } + + private static void SetFieldValue(BsonDocument doc, string path, object? value) + { + var segments = path.Split('.'); + var current = doc; + + for (var i = 0; i < segments.Length - 1; i++) + { + var segment = segments[i]; + if (!current.TryGetValue(segment, out var existing) || existing is not BsonDocument nested) + { + nested = new BsonDocument(); + current[segment] = nested; + } + + current = nested; + } + + current[segments[^1]] = ToBsonValue(value); + } + + private static BsonValue ToBsonValue(object? value) + { + if (value is null) + return BsonNull.Value; + + if (value is BsonValue bsonValue) + return bsonValue; + + if (BsonTypeMapper.TryMapToBsonValue(value, out var mapped)) + return mapped; + + return value.ToBsonDocument(); + } + private static object? GetFieldValue(BsonDocument doc, string path) { var mongoPath = path == "id" ? "_id" : path; @@ -135,4 +394,6 @@ public async Task> ExecuteAsync( _ => current.ToString() }; } + + private readonly record struct ExternalResolutionCacheEntry(bool Found, object? Value); } diff --git a/src/OxQL.Mongo/MongoQueryExecutor.cs b/src/OxQL.Mongo/MongoQueryExecutor.cs index df5e53f..b25efb2 100644 --- a/src/OxQL.Mongo/MongoQueryExecutor.cs +++ b/src/OxQL.Mongo/MongoQueryExecutor.cs @@ -20,7 +20,8 @@ public MongoQueryExecutor( IMongoCollection collection, OxQLOptions options, ICursorSerializer? cursorSerializer = null, - IQueryPlanCache? cache = null) + IQueryPlanCache? cache = null, + IEnumerable? externalResolvers = null) { ArgumentNullException.ThrowIfNull(collection); ArgumentNullException.ThrowIfNull(options); @@ -30,7 +31,7 @@ public MongoQueryExecutor( _normalizer = new Core.Normalization.QueryRequestNormalizer(options); _planner = new Core.Planning.QueryPlanner(_normalizer); _cache = cache ?? new Core.Caching.QueryPlanCache(options); - _adapter = new MongoQueryAdapter(_ => collection, cursor); + _adapter = new MongoQueryAdapter(_ => collection, cursor, externalResolvers); } public MongoQueryExecutor( diff --git a/src/OxQL.Mongo/ServiceCollectionExtensions.cs b/src/OxQL.Mongo/ServiceCollectionExtensions.cs index 139bd7e..b2379b0 100644 --- a/src/OxQL.Mongo/ServiceCollectionExtensions.cs +++ b/src/OxQL.Mongo/ServiceCollectionExtensions.cs @@ -129,9 +129,10 @@ IMongoCollection Resolve(string entityType) services.AddSingleton, MongoQueryAdapter>(sp => { - var resolver = sp.GetRequiredService(); - var cursorSerializer = sp.GetRequiredService(); - return new MongoQueryAdapter(resolver.Resolve, cursorSerializer); + var resolver = sp.GetRequiredService(); + var cursorSerializer = sp.GetRequiredService(); + var externalResolvers = sp.GetServices(); + return new MongoQueryAdapter(resolver.Resolve, cursorSerializer, externalResolvers); }); services.AddSingleton, MongoQueryExecutor>(sp => diff --git a/src/OxQL.Tests/Core/QueryRequestDeserializationTests.cs b/src/OxQL.Tests/Core/QueryRequestDeserializationTests.cs index 33128dc..3fdf0ca 100644 --- a/src/OxQL.Tests/Core/QueryRequestDeserializationTests.cs +++ b/src/OxQL.Tests/Core/QueryRequestDeserializationTests.cs @@ -102,6 +102,54 @@ public void Deserialize_LookupStage_Succeeds() request.Pipeline[0].Lookup.As.Should().Be("customer"); } + [Fact] + public void Deserialize_ResolveStageWithSubquery_Succeeds() + { + var json = """ + { + "entityType": "invoice", + "pipeline": [ + { + "resolve": { + "source": "contact-api/v1", + "localPath": "attributes.contactId", + "parameters": { + "contactId": "attributes.contactId" + }, + "subquery": { + "entityType": "contact", + "pipeline": [ + { + "match": { + "id": { "eq": { "$var": "contactId" } } + } + }, + { + "page": { + "limit": 1 + } + } + ] + }, + "as": "contact" + } + } + ] + } + """; + + var request = JsonSerializer.Deserialize(json, JsonOptions); + + request.Should().NotBeNull(); + request!.Pipeline[0].Resolve.Should().NotBeNull(); + request.Pipeline[0].Resolve!.Source.Should().Be("contact-api/v1"); + request.Pipeline[0].Resolve.Parameters.Should().ContainKey("contactId"); + request.Pipeline[0].Resolve.Parameters!["contactId"].Should().Be("attributes.contactId"); + request.Pipeline[0].Resolve.Subquery.Should().NotBeNull(); + request.Pipeline[0].Resolve.Subquery!.EntityType.Should().Be("contact"); + request.Pipeline[0].Resolve.Subquery.Pipeline.Should().HaveCount(2); + } + [Fact] public void Deserialize_UnwindStage_Succeeds() { diff --git a/src/OxQL.Tests/Core/QueryValidatorTests.cs b/src/OxQL.Tests/Core/QueryValidatorTests.cs index 81354fd..e6ee616 100644 --- a/src/OxQL.Tests/Core/QueryValidatorTests.cs +++ b/src/OxQL.Tests/Core/QueryValidatorTests.cs @@ -306,6 +306,41 @@ public void Validate_DisallowedResolveSource_Fails() result.Errors.Should().Contain(e => e.Code == "DISALLOWED_RESOLVE_SOURCE"); } + [Fact] + public void Validate_ResolveSubqueryWithoutEntityType_Fails() + { + var request = new QueryRequest + { + EntityType = "invoice", + Pipeline = + [ + new PipelineStage + { + Resolve = new ResolveStage + { + Source = "crm.customer", + LocalPath = "attributes.customerId", + Parameters = new Dictionary + { + ["customerId"] = "attributes.customerId" + }, + Subquery = new QueryRequest + { + EntityType = "", + Pipeline = [new PipelineStage { Page = new PageStage { Limit = 1 } }] + }, + As = "customer" + } + } + ] + }; + + var result = CreateValidator().Validate(request); + + result.IsValid.Should().BeFalse(); + result.Errors.Should().Contain(e => e.Code == "INVALID_RESOLVE_SUBQUERY_ENTITY"); + } + [Fact] public void Validate_ExceedMaxProjectionFields_Fails() { diff --git a/src/OxQL.Tests/Fakes/FakeExternalResolver.cs b/src/OxQL.Tests/Fakes/FakeExternalResolver.cs index 575ed85..670f7d1 100644 --- a/src/OxQL.Tests/Fakes/FakeExternalResolver.cs +++ b/src/OxQL.Tests/Fakes/FakeExternalResolver.cs @@ -8,8 +8,11 @@ namespace OxQL.Tests.Fakes; public sealed class FakeExternalResolver : IExternalResolver { private readonly Dictionary _data = new(); + private readonly List> _requestedKeyBatches = []; public string Source { get; } + public int CallCount { get; private set; } + public IReadOnlyList> RequestedKeyBatches => _requestedKeyBatches; public FakeExternalResolver(string source) { @@ -29,6 +32,9 @@ public FakeExternalResolver WithData(string key, object? value) IReadOnlyList keys, CancellationToken cancellationToken = default) { + CallCount++; + _requestedKeyBatches.Add(keys.ToArray()); + var results = new Dictionary(); foreach (var key in keys) { diff --git a/src/OxQL.Tests/Mongo/MongoQueryAdapterResolveTests.cs b/src/OxQL.Tests/Mongo/MongoQueryAdapterResolveTests.cs new file mode 100644 index 0000000..16b5319 --- /dev/null +++ b/src/OxQL.Tests/Mongo/MongoQueryAdapterResolveTests.cs @@ -0,0 +1,209 @@ +using FluentAssertions; +using OxQL.Core.Cursor; +using OxQL.Core.Interfaces; +using OxQL.Core.Models; +using OxQL.Tests.Fakes; +using MongoDB.Bson; +using MongoDB.Driver; +using Xunit; + +namespace OxQL.Tests.Mongo; + +public class MongoQueryAdapterResolveTests +{ + private sealed class SubqueryResolver : IExternalResolver + { + private readonly List _queries = []; + + public string Source { get; } + public int QueryCallCount => _queries.Count; + public IReadOnlyList Queries => _queries; + + public SubqueryResolver(string source) + { + Source = source; + } + + public Task> ResolveAsync( + IReadOnlyList keys, + CancellationToken cancellationToken = default) + { + return Task.FromResult>(new Dictionary()); + } + + public Task ResolveOneAsync( + ExternalResolveRequest request, + CancellationToken cancellationToken = default) + { + if (request.Query is null) + return Task.FromResult(null); + + _queries.Add(request.Query); + var id = request.Query.Variables?.GetValue("contactId")?.ToString(); + if (string.IsNullOrWhiteSpace(id)) + return Task.FromResult(null); + + return Task.FromResult(new Dictionary + { + ["id"] = id, + ["name"] = $"Contact {id}" + }); + } + } + + [Fact] + public async Task ExecuteAsync_ResolveStage_CachesResponsesPerQuery() + { + var resolver = new FakeExternalResolver("crm.customer") + .WithData("cust-1", new Dictionary + { + ["id"] = "cust-1", + ["name"] = "Contoso" + }); + + var adapter = new global::OxQL.Mongo.MongoQueryAdapter( + _ => throw new NotSupportedException(), + new CursorSerializer(), + [resolver], + (_, _, _) => Task.FromResult(new List + { + new() { ["_id"] = "1", ["attributes"] = new BsonDocument("customerId", "cust-1") }, + new() { ["_id"] = "2", ["attributes"] = new BsonDocument("customerId", "cust-1") } + })); + + var response = await adapter.ExecuteAsync( + CreatePlan( + new PipelineStage + { + Resolve = new ResolveStage + { + Source = "crm.customer", + LocalPath = "attributes.customerId", + As = "crmCustomer" + } + }), + null); + + response.Items.Should().HaveCount(2); + response.Items[0]["crmCustomer"].AsBsonDocument["name"].AsString.Should().Be("Contoso"); + response.Items[1]["crmCustomer"].AsBsonDocument["name"].AsString.Should().Be("Contoso"); + resolver.CallCount.Should().Be(1); + resolver.RequestedKeyBatches.Should().ContainSingle(); + resolver.RequestedKeyBatches[0].Should().Equal("cust-1"); + } + + [Fact] + public async Task ExecuteAsync_RepeatedResolveStage_ReusesCachedKeys() + { + var resolver = new FakeExternalResolver("crm.customer") + .WithData("cust-1", new Dictionary { ["id"] = "cust-1" }); + + var adapter = new global::OxQL.Mongo.MongoQueryAdapter( + _ => throw new NotSupportedException(), + new CursorSerializer(), + [resolver], + (_, _, _) => Task.FromResult(new List + { + new() { ["_id"] = "1", ["attributes"] = new BsonDocument("customerId", "cust-1") } + })); + + var response = await adapter.ExecuteAsync( + CreatePlan( + new PipelineStage + { + Resolve = new ResolveStage + { + Source = "crm.customer", + LocalPath = "attributes.customerId", + As = "crmCustomer" + } + }, + new PipelineStage + { + Resolve = new ResolveStage + { + Source = "crm.customer", + LocalPath = "attributes.customerId", + As = "crmCustomerAgain" + } + }), + null); + + response.Items[0].Contains("crmCustomer").Should().BeTrue(); + response.Items[0].Contains("crmCustomerAgain").Should().BeTrue(); + resolver.CallCount.Should().Be(1); + resolver.RequestedKeyBatches.Should().ContainSingle(); + resolver.RequestedKeyBatches[0].Should().Equal("cust-1"); + } + + [Fact] + public async Task ExecuteAsync_ResolveSubquery_UsesParametersAndCachesPerRequest() + { + var resolver = new SubqueryResolver("contact-api/v1"); + + var adapter = new global::OxQL.Mongo.MongoQueryAdapter( + _ => throw new NotSupportedException(), + new CursorSerializer(), + [resolver], + (_, _, _) => Task.FromResult(new List + { + new() { ["_id"] = "inv-1", ["attributes"] = new BsonDocument("contactId", "contact-42") }, + new() { ["_id"] = "inv-2", ["attributes"] = new BsonDocument("contactId", "contact-42") } + })); + + var response = await adapter.ExecuteAsync( + CreatePlan( + new PipelineStage + { + Resolve = new ResolveStage + { + Source = "contact-api/v1", + LocalPath = "attributes.contactId", + Parameters = new Dictionary + { + ["contactId"] = "attributes.contactId" + }, + Subquery = new QueryRequest + { + EntityType = "contact", + Pipeline = + [ + new PipelineStage + { + Match = new MatchStage + { + Condition = new FilterCondition + { + Path = "id", + Op = "eq", + Value = System.Text.Json.JsonDocument.Parse("{\"$var\":\"contactId\"}").RootElement + } + } + }, + new PipelineStage { Page = new PageStage { Limit = 1 } } + ] + }, + As = "contact" + } + }), + null); + + response.Items.Should().HaveCount(2); + response.Items[0]["contact"].AsBsonDocument["id"].AsString.Should().Be("contact-42"); + response.Items[1]["contact"].AsBsonDocument["id"].AsString.Should().Be("contact-42"); + + resolver.QueryCallCount.Should().Be(1); + resolver.Queries.Should().ContainSingle(); + resolver.Queries[0].Variables.Should().NotBeNull(); + resolver.Queries[0].Variables!.GetValue("contactId")!.ToString().Should().Be("contact-42"); + } + + private static QueryPlan CreatePlan(params PipelineStage[] stages) => new() + { + EntityType = "invoice", + Pipeline = [.. stages, new PipelineStage { Page = new PageStage { Limit = 50 } }], + Sort = [new SortField { Path = "id", Direction = "asc" }], + Page = new PageStage { Limit = 50 }, + CacheKey = "test" + }; +} diff --git a/src/README.md b/src/README.md new file mode 100644 index 0000000..e69de29 diff --git a/src/docs/cross-service-resolver.md b/src/docs/cross-service-resolver.md new file mode 100644 index 0000000..090accf --- /dev/null +++ b/src/docs/cross-service-resolver.md @@ -0,0 +1,306 @@ +# Cross-Service Resolver for OxQL + +Enable an OxQL-powered microservice to resolve data from **other** microservices at query time — with no static knowledge of peers required. Callers supply the target services dynamically via a `?services=` query parameter. + +--- + +## How it works + +``` +GET /oxql/query + ?q=match id in ["abc"] | resolve vehicle-api/v1.vehicleId + &services=vehicle-api/v1=https://vehicle-api.internal/,erp-api/v2=https://erp-api.internal/ +``` + +1. The receiving service parses `?services=` into `(source, baseAddress)` pairs. +2. `DynamicResolverFactory` creates one `ExternalServiceResolver` per pair — no startup registration needed. +3. Each resolver forwards the OxQL sub-query to the target service as `POST /oxql/query?q=`. +4. Results are merged back into the parent query result. + +--- + +## Files + +Place everything in a single file, e.g. `ExternalServiceResolver.cs`, inside your project. + +```csharp +using System.Net.Http.Json; +using System.Text.Json; +using System.Web; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; +using OxQL.Core.Interfaces; + +namespace YourService.CrossService; // <-- replace with your actual namespace + +// --------------------------------------------------------------------------- +// Options +// --------------------------------------------------------------------------- + +public sealed class OxQLCrossServiceOptions +{ + /// Maximum number of external services a caller may pass per request. + public int MaxServicesPerRequest { get; set; } = 20; + + /// Allow plain HTTP base addresses. Keep false in production. + public bool AllowHttp { get; set; } = false; +} + +// --------------------------------------------------------------------------- +// ExternalServiceResolver (implements IExternalResolver) +// --------------------------------------------------------------------------- + +/// +/// Resolves data from a remote OxQL service. +/// Source format: "{service-name}/{version}" e.g. "vehicle-api/v1" +/// The sub-query is sent as the ?q= query parameter. +/// +public sealed class ExternalServiceResolver : IExternalResolver +{ + private readonly HttpClient _httpClient; + private readonly ILogger _logger; + + public ExternalServiceResolver( + HttpClient httpClient, + string source, + ILogger logger) + { + _httpClient = httpClient ?? throw new ArgumentNullException(nameof(httpClient)); + Source = source ?? throw new ArgumentNullException(nameof(source)); + _logger = logger ?? throw new ArgumentNullException(nameof(logger)); + } + + public string Source { get; } + + public async Task> ResolveAsync( + IReadOnlyList keys, + CancellationToken cancellationToken = default) + { + if (keys.Count == 0) + return new Dictionary(); + + // Build: match id in ["key1","key2"] + var inList = string.Join(", ", keys.Select(k => $"\"{k}\"")); + var query = $"match id in [{inList}]"; + var url = $"/oxql/query?q={HttpUtility.UrlEncode(query)}"; + + try + { + var response = await _httpClient.PostAsync(url, content: null, cancellationToken); + response.EnsureSuccessStatusCode(); + + var items = await response.Content + .ReadFromJsonAsync>(cancellationToken: cancellationToken) ?? []; + + var result = new Dictionary(items.Count); + foreach (var item in items) + { + if (item.TryGetProperty("id", out var idProp) && idProp.GetString() is { } id) + result[id] = item; + } + + return result; + } + catch (HttpRequestException ex) + { + _logger.LogError(ex, "Failed to resolve keys from external service '{Source}'", Source); + return new Dictionary(); + } + } +} + +// --------------------------------------------------------------------------- +// ServiceRegistryParser +// --------------------------------------------------------------------------- + +/// +/// Parses the ?services= query parameter. +/// Format: name/version=https://base-url/,name2/version2=https://... +/// +public static class ServiceRegistryParser +{ + public static IReadOnlyList<(string Source, Uri BaseAddress)> Parse(string? input) + { + if (string.IsNullOrWhiteSpace(input)) + return []; + + var entries = input.Split(',', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries); + var result = new List<(string, Uri)>(entries.Length); + + foreach (var entry in entries) + { + var idx = entry.IndexOf('='); + if (idx <= 0) + throw new ArgumentException($"Malformed services entry (missing '='): '{entry}'"); + + var source = entry[..idx].Trim(); + var rawUrl = entry[(idx + 1)..].Trim(); + + if (!Uri.TryCreate(rawUrl, UriKind.Absolute, out var uri)) + throw new ArgumentException($"Invalid base address for '{source}': '{rawUrl}'"); + + result.Add((source, uri)); + } + + return result; + } +} + +// --------------------------------------------------------------------------- +// DynamicResolverFactory +// --------------------------------------------------------------------------- + +/// +/// Creates ExternalServiceResolver instances on-the-fly from a caller-supplied +/// service map. No static peer registration is required at startup. +/// +public sealed class DynamicResolverFactory +{ + private readonly IHttpClientFactory _httpClientFactory; + private readonly ILogger _resolverLogger; + private readonly OxQLCrossServiceOptions _options; + + public DynamicResolverFactory( + IHttpClientFactory httpClientFactory, + ILogger resolverLogger, + IOptions options) + { + _httpClientFactory = httpClientFactory; + _resolverLogger = resolverLogger; + _options = options.Value; + } + + /// + /// Too many services, or HTTP used while AllowHttp is false. + /// + public IReadOnlyList CreateResolvers( + IReadOnlyList<(string Source, Uri BaseAddress)> serviceMap) + { + if (serviceMap.Count > _options.MaxServicesPerRequest) + throw new InvalidOperationException( + $"Request exceeds the maximum of {_options.MaxServicesPerRequest} external services."); + + var resolvers = new List(serviceMap.Count); + + foreach (var (source, baseAddress) in serviceMap) + { + if (!_options.AllowHttp && baseAddress.Scheme.Equals("http", StringComparison.OrdinalIgnoreCase)) + throw new InvalidOperationException( + $"Plain HTTP is not allowed for '{source}'. Use HTTPS."); + + var client = _httpClientFactory.CreateClient($"oxql-external-{source}"); + client.BaseAddress = baseAddress; + + resolvers.Add(new ExternalServiceResolver(client, source, _resolverLogger)); + } + + return resolvers; + } +} + +// --------------------------------------------------------------------------- +// DI registration +// --------------------------------------------------------------------------- + +public static class OxQLCrossServiceExtensions +{ + /// + /// Registers the dynamic cross-service resolver infrastructure. + /// No target services need to be known at startup. + /// + public static IServiceCollection AddOxQLCrossServiceQuerying( + this IServiceCollection services, + Action? configure = null) + { + services.AddHttpClient(); + + if (configure is not null) + services.Configure(configure); + else + services.Configure(_ => { }); + + services.AddScoped(); + + return services; + } +} +``` + +--- + +## Registration — `Program.cs` + +```csharp +builder.Services.AddOxQLCrossServiceQuerying(opts => +{ + opts.AllowHttp = builder.Environment.IsDevelopment(); + opts.MaxServicesPerRequest = 10; +}); +``` + +No peer services are registered here. The caller decides which services to include at request time. + +--- + +## Endpoint + +```csharp +app.MapGet("/oxql/query", async ( + [FromQuery] string q, + [FromQuery] string? services, + DynamicResolverFactory factory, + IOxQLQueryService queryService, + CancellationToken ct) => +{ + var serviceMap = ServiceRegistryParser.Parse(services); + var resolvers = factory.CreateResolvers(serviceMap); + + // Register resolvers into your execution context, then run the query + var request = BuildQueryRequest(q); + var result = await queryService.ExecuteAsync(request, ct); + + return Results.Ok(result); +}); +``` + +--- + +## Example request + +``` +GET /oxql/query + ?q=match id in ["abc123"] | resolve vehicle-api/v1.vehicleId + &services=vehicle-api/v1=https://vehicle-api.internal/,erp-api/v2=https://erp-api.internal/ +``` + +The resolver for `vehicle-api/v1` will call: + +``` +POST https://vehicle-api.internal/oxql/query?q=match%20id%20in%20%5B%22abc123%22%5D +``` + +--- + +## `?services=` format + +``` +{service-name}/{version}={https://base-url/} +``` + +Multiple services are comma-separated: + +``` +vehicle-api/v1=https://vehicle-api.internal/,erp-api/v2=https://erp-api.internal/ +``` + +--- + +## Security + +| Rule | Default | +|---|---| +| HTTPS required | `AllowHttp = false` | +| Max services per request | `MaxServicesPerRequest = 20` | +| Invalid base URI | throws `ArgumentException` | +| HTTP request failure | logs error, returns empty result (no crash) | diff --git a/src/docs/oxql-query-syntax.md b/src/docs/oxql-query-syntax.md index ad1017e..15c27f2 100644 --- a/src/docs/oxql-query-syntax.md +++ b/src/docs/oxql-query-syntax.md @@ -224,8 +224,40 @@ Fetches a related document from a configured external source (e.g. a CRM, extern |---|---|---| | `source` | ✅ | External source identifier. Must be in the server's allowed list. | | `localPath` | ✅ | Local field containing the lookup key. | +| `parameters` | ❌ | Variable mapping (`name -> localPath`) used for subquery resolve mode. | +| `subquery` | ❌ | Query sent to the external resolver endpoint. Supports `$var` references from `parameters`. | | `as` | ✅ | Alias for the resolved result. | +Subquery resolve example: + +```json +{ + "resolve": { + "source": "contact-api/v1", + "localPath": "attributes.contactId", + "parameters": { + "contactId": "attributes.contactId" + }, + "subquery": { + "entityType": "contact", + "pipeline": [ + { + "match": { + "id": { "eq": { "$var": "contactId" } } + } + }, + { + "page": { + "limit": 1 + } + } + ] + }, + "as": "contact" + } +} +``` + --- ## Stage: `unwind`