|
1 | 1 | using System.Text.Json; |
| 2 | +using System.Text.Json.Serialization; |
2 | 3 | using Abacus.Run.Abstractions; |
3 | 4 | using Abacus.Run.Core; |
4 | 5 | using Microsoft.AspNetCore.Builder; |
@@ -41,6 +42,49 @@ public sealed record CancelRequestDto(string? Reason); |
41 | 42 |
|
42 | 43 | public sealed record RerunRequestDto(string? Mode, JsonElement? Context, string? FromCheckpointId, string? Reason); |
43 | 44 |
|
| 45 | +/// <summary> |
| 46 | +/// One row of the event log. <c>PayloadJson</c> is re-emitted as a raw JSON value rather than an |
| 47 | +/// escaped string, so a caller reads the payload directly instead of parsing it twice. |
| 48 | +/// </summary> |
| 49 | +public sealed record EventLogDto( |
| 50 | + string InstanceId, |
| 51 | + string? WorkflowName, |
| 52 | + long Sequence, |
| 53 | + string EventType, |
| 54 | + string? ExecutorId, |
| 55 | + int? Superstep, |
| 56 | + string? TenantId, |
| 57 | + [property: JsonConverter(typeof(RawJsonConverter))] string PayloadJson, |
| 58 | + DateTimeOffset OccurredAt); |
| 59 | + |
| 60 | +/// <summary>Writes an already-serialised JSON string through untouched.</summary> |
| 61 | +internal sealed class RawJsonConverter : JsonConverter<string> |
| 62 | +{ |
| 63 | + public override string Read(ref Utf8JsonReader reader, Type typeToConvert, JsonSerializerOptions options) |
| 64 | + => reader.GetString() ?? "{}"; |
| 65 | + |
| 66 | + public override void Write(Utf8JsonWriter writer, string value, JsonSerializerOptions options) |
| 67 | + { |
| 68 | + if (string.IsNullOrWhiteSpace(value)) |
| 69 | + { |
| 70 | + writer.WriteNullValue(); |
| 71 | + return; |
| 72 | + } |
| 73 | + |
| 74 | + try |
| 75 | + { |
| 76 | + using JsonDocument document = JsonDocument.Parse(value); |
| 77 | + document.RootElement.WriteTo(writer); |
| 78 | + } |
| 79 | + catch (JsonException) |
| 80 | + { |
| 81 | + // A payload that is not valid JSON is written as the string it is, rather than failing |
| 82 | + // the whole response — one malformed row must not hide the rest of the log. |
| 83 | + writer.WriteStringValue(value); |
| 84 | + } |
| 85 | + } |
| 86 | +} |
| 87 | + |
44 | 88 | public sealed record PublishEventRequestDto( |
45 | 89 | string? Topic, JsonElement? Payload, string? CorrelationKey, string? Scope); |
46 | 90 |
|
@@ -391,6 +435,17 @@ private static void MapEvents(IEndpointRouteBuilder app) |
391 | 435 | return Results.NotFound(); |
392 | 436 | } |
393 | 437 |
|
| 438 | + // A log-only workflow would hold this stream open and never write to it, which a client |
| 439 | + // cannot tell apart from a stalled run. Refuse, and say where the events actually are. |
| 440 | + if (NotificationsOf(services, instance) is { IsLogOnly: true }) |
| 441 | + { |
| 442 | + return Results.Problem( |
| 443 | + title: "This workflow does not stream events", |
| 444 | + detail: $"'{instance.WorkflowName}' is configured for log-only delivery. Read its events at " + |
| 445 | + $"/v2/workflows/{instance.WorkflowName}/instances/{id}/events.", |
| 446 | + statusCode: StatusCodes.Status409Conflict); |
| 447 | + } |
| 448 | + |
394 | 449 | long from = 0; |
395 | 450 | string? lastEventId = http.Request.Headers["Last-Event-ID"].FirstOrDefault(); |
396 | 451 | if (long.TryParse(lastEventId, out long parsed)) |
@@ -593,6 +648,59 @@ private static void MapApprovals(IEndpointRouteBuilder app) |
593 | 648 | return Results.Accepted(value: new { messageId = message.MessageId, topic = message.Topic }); |
594 | 649 | }); |
595 | 650 |
|
| 651 | + // The event log as a queryable record rather than a stream. This is the read path for a |
| 652 | + // log-only workflow, and works just as well for a streamed one — the same rows either way. |
| 653 | + // Scoped by workflow name, matching the instance-state route, so the caller states which |
| 654 | + // workflow they believe they are reading and a mismatch is caught rather than assumed. |
| 655 | + app.MapGet("/v2/workflows/{name}/instances/{id}/events", async ( |
| 656 | + string name, |
| 657 | + string id, |
| 658 | + [FromQuery] long? from, |
| 659 | + [FromQuery] long? to, |
| 660 | + [FromQuery] int? limit, |
| 661 | + [FromQuery] string? types, |
| 662 | + IInstanceStore instances, |
| 663 | + IEventStore store, |
| 664 | + CancellationToken cancellationToken) => |
| 665 | + { |
| 666 | + WorkflowInstance? instance = await instances.GetAsync(id, cancellationToken).ConfigureAwait(false); |
| 667 | + if (instance is null) |
| 668 | + { |
| 669 | + return Results.NotFound(); |
| 670 | + } |
| 671 | + |
| 672 | + if (!string.Equals(instance.WorkflowName, name, StringComparison.OrdinalIgnoreCase)) |
| 673 | + { |
| 674 | + return Results.Problem( |
| 675 | + title: "Workflow mismatch", |
| 676 | + detail: $"Instance '{id}' belongs to '{instance.WorkflowName}', not '{name}'.", |
| 677 | + statusCode: StatusCodes.Status404NotFound); |
| 678 | + } |
| 679 | + |
| 680 | + string[]? typeFilter = types? |
| 681 | + .Split(',', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries); |
| 682 | + |
| 683 | + Page<EventEnvelope> page = await store.QueryAsync( |
| 684 | + new EventQuery(id, from ?? 0, to, Math.Clamp(limit ?? 100, 1, 1000), typeFilter), cancellationToken) |
| 685 | + .ConfigureAwait(false); |
| 686 | + |
| 687 | + return Results.Ok(new |
| 688 | + { |
| 689 | + instanceId = id, |
| 690 | + workflowName = instance.WorkflowName, |
| 691 | + workflowVersion = instance.WorkflowVersion, |
| 692 | + tenantId = instance.TenantId, |
| 693 | + total = page.Total, |
| 694 | + nextCursor = page.NextCursor, |
| 695 | + |
| 696 | + // WorkflowName is denormalised onto each row too, so a caller collecting events from |
| 697 | + // several instances keeps the attribution without carrying the envelope's context. |
| 698 | + items = page.Items.Select(e => new EventLogDto( |
| 699 | + e.InstanceId, e.WorkflowName ?? instance.WorkflowName, e.Sequence, e.EventType, |
| 700 | + e.ExecutorId, e.Superstep, e.TenantId, e.PayloadJson, e.OccurredAt)) |
| 701 | + }); |
| 702 | + }); |
| 703 | + |
596 | 704 | // What is listening, and what is waiting. An instance parked on an event with no visible |
597 | 705 | // reason is the worst version of this feature. |
598 | 706 | app.MapGet("/subscriptions", async ( |
@@ -730,6 +838,21 @@ internal static bool TryParseOutcome(string? value, out ApprovalOutcomeKind outc |
730 | 838 | approval.Assignees, approval.RequiredApprovers, approval.AllowModification, approval.CreatedAt, |
731 | 839 | approval.ExpiresAt, $"/approvals/{approval.ApprovalId}/decision"); |
732 | 840 |
|
| 841 | + /// <summary> |
| 842 | + /// The notification policy the instance's own workflow version declared, or null when it |
| 843 | + /// declared none. Resolved by the instance's version, not the newest, so an in-flight run keeps |
| 844 | + /// the behaviour it started under. |
| 845 | + /// </summary> |
| 846 | + private static NotificationPolicy? NotificationsOf(IServiceProvider services, WorkflowInstance instance) |
| 847 | + { |
| 848 | + var registry = services.GetService<IWorkflowRegistry>(); |
| 849 | + |
| 850 | + return registry?.Resolve(instance.WorkflowName, instance.WorkflowVersion)?.Definition |
| 851 | + is INotifyingWorkflow notifying |
| 852 | + ? notifying.Notifications |
| 853 | + : null; |
| 854 | + } |
| 855 | + |
733 | 856 | /// <summary> |
734 | 857 | /// The delivered payload is deliberately absent: it is domain data that has already been |
735 | 858 | /// redacted on its way to the event stream, and repeating it unredacted here would undo that. |
|
0 commit comments