diff --git a/Directory.Packages.props b/Directory.Packages.props index 26c8128..d1ffd37 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -49,6 +49,10 @@ + + + diff --git a/src/CritterMart.Catalog/Program.cs b/src/CritterMart.Catalog/Program.cs index 5f322db..a9c47f9 100644 --- a/src/CritterMart.Catalog/Program.cs +++ b/src/CritterMart.Catalog/Program.cs @@ -47,11 +47,15 @@ // event stream in the console's store) — must be unique across monitored services. opts.ServiceName = "Catalog"; - // RabbitMQ solely as the CritterWatch telemetry channel — Catalog has no cross-BC - // message flows (no UseConventionalRouting on purpose). Aspire injects the "rabbitmq" - // connection string via WithReference. + // RabbitMQ as the CritterWatch telemetry channel. On `main` Catalog has no cross-BC message + // flows (no UseConventionalRouting on purpose). Aspire injects the "rabbitmq" connection string + // via WithReference. opts.UseRabbitMqUsingNamedConnection("rabbitmq") - .AutoProvision(); + .AutoProvision() + // CW-TELEMETRY SPIKE (research/cw-telemetry-spike) — NOT round-one baseline: conventional + // routing so Catalog can subscribe to the OrderPlacedSignal broadcast (fan-out target #2), + // giving it its first inbound Topology edge in the console. See docs/research/. + .UseConventionalRouting(); // Metrics/health flow to the shared `critterwatch` queue; the console sends control // commands (pause listeners, chaos monkey, …) back on this service's private queue. diff --git a/src/CritterMart.Catalog/Spike/CatalogOrderPlacedSignalHandler.cs b/src/CritterMart.Catalog/Spike/CatalogOrderPlacedSignalHandler.cs new file mode 100644 index 0000000..18abcab --- /dev/null +++ b/src/CritterMart.Catalog/Spike/CatalogOrderPlacedSignalHandler.cs @@ -0,0 +1,22 @@ +using CritterMart.Contracts; +using Microsoft.Extensions.Logging; + +namespace CritterMart.Catalog.Spike; + +// ════════════════════════════════════════════════════════════════════════════════════════════════ +// CW-TELEMETRY SPIKE (research/cw-telemetry-spike) — NOT round-one baseline. +// See docs/research/cw-telemetry-fodder.md. +// ════════════════════════════════════════════════════════════════════════════════════════════════ + +// The SECOND subscriber to the OrderPlacedSignal broadcast — the one that makes it a genuine fan-out. +// On `main` Catalog has no cross-BC message flows (RabbitMQ is wired solely as the CritterWatch +// telemetry channel, no conventional routing). The spike adds conventional routing to Catalog's +// Program.cs so this handler binds an inbound queue — giving Catalog its first Topology edge in the +// console. The handler does no catalog work; the value is the edge itself. +public static class CatalogOrderPlacedSignalHandler +{ + public static void Handle(OrderPlacedSignal message, ILogger logger) => + logger.LogInformation( + "CW-spike: Catalog observed OrderPlacedSignal for order {OrderId} (total {Total})", + message.OrderId, message.Total); +} diff --git a/src/CritterMart.Contracts/OrderPlacedSignal.cs b/src/CritterMart.Contracts/OrderPlacedSignal.cs new file mode 100644 index 0000000..e8c8759 --- /dev/null +++ b/src/CritterMart.Contracts/OrderPlacedSignal.cs @@ -0,0 +1,14 @@ +namespace CritterMart.Contracts; + +// ════════════════════════════════════════════════════════════════════════════════════════════════ +// CW-TELEMETRY SPIKE (research/cw-telemetry-spike) — NOT round-one baseline. +// See docs/research/cw-telemetry-fodder.md. +// ════════════════════════════════════════════════════════════════════════════════════════════════ + +// A BROADCAST integration event published by Orders when an order is placed (gated on Cw:Telemetry). +// Unlike the request/reply ReserveStock flow (one sender, one handler), this notification fans OUT to +// MULTIPLE subscribers — Inventory AND Catalog — over RabbitMQ. The point is purely topological: it +// thickens CritterWatch's Topology edges, adds Listeners queues, and produces Durability inbox/outbox +// rows, and it gives Catalog (which has no cross-BC flows of its own on `main`) its first inbound edge. +// Carries only the notification shape — no command semantics, no reply expected. +public record OrderPlacedSignal(string OrderId, string CustomerId, decimal Total); diff --git a/src/CritterMart.CritterWatch/CritterMart.CritterWatch.csproj b/src/CritterMart.CritterWatch/CritterMart.CritterWatch.csproj index baa3f6c..2579893 100644 --- a/src/CritterMart.CritterWatch/CritterMart.CritterWatch.csproj +++ b/src/CritterMart.CritterWatch/CritterMart.CritterWatch.csproj @@ -8,6 +8,8 @@ + + diff --git a/src/CritterMart.CritterWatch/Program.cs b/src/CritterMart.CritterWatch/Program.cs index 93219a9..c052872 100644 --- a/src/CritterMart.CritterWatch/Program.cs +++ b/src/CritterMart.CritterWatch/Program.cs @@ -1,3 +1,4 @@ +using CritterWatch.Mcp; using CritterWatch.Services.Hosting; using Wolverine.RabbitMQ; @@ -39,9 +40,20 @@ // listener above already gives one node the same per-service ordering guarantee. enableClusterPartitioning: false); +// CritterWatch.Mcp — the cross-application MCP server (spike DX exploration for the JasperFx feedback +// round). Stateless HTTP transport is mandatory: in stateful mode the HttpContext reachable for RBAC is +// the one that opened the SignalR/MCP session, not the per-tool-call one, so HttpContext.User would be +// stale. AddCritterWatchMcp() sets stateless mode itself. The tool surface is license-gated (paid tier), +// which our Trial license satisfies; action tools additionally check ICritterWatchAuthorizer (open by +// default here — no RBAC wired). NOT a round-one feature; see docs/research/cw-feedback-jasperfx-deep.md DX-6. +builder.Services.AddCritterWatchMcp(); + var app = builder.Build(); // Maps the Wolverine HTTP endpoints, the SignalR hub at /api/messages, and the embedded SPA. app.UseCritterWatch(); +// Mounts the MCP streamable-HTTP endpoint (default route /api/mcp). +app.MapCritterWatchMcp(); + await app.RunAsync(); diff --git a/src/CritterMart.Inventory/Spike/InventoryOrderPlacedSignalHandler.cs b/src/CritterMart.Inventory/Spike/InventoryOrderPlacedSignalHandler.cs new file mode 100644 index 0000000..f2bed63 --- /dev/null +++ b/src/CritterMart.Inventory/Spike/InventoryOrderPlacedSignalHandler.cs @@ -0,0 +1,21 @@ +using CritterMart.Contracts; +using Microsoft.Extensions.Logging; + +namespace CritterMart.Inventory.Spike; + +// ════════════════════════════════════════════════════════════════════════════════════════════════ +// CW-TELEMETRY SPIKE (research/cw-telemetry-spike) — NOT round-one baseline. +// See docs/research/cw-telemetry-fodder.md. +// ════════════════════════════════════════════════════════════════════════════════════════════════ + +// One of two subscribers (Catalog is the other) to the OrderPlacedSignal broadcast. Its only purpose +// is to make Inventory bind a conventional listening queue for the signal — that queue is the fan-out +// edge CritterWatch's Topology renders. The handler does no domain work (Inventory's real reservation +// flow is the ReserveStock request/reply path). Inert unless Orders broadcasts (Cw:Telemetry on). +public static class InventoryOrderPlacedSignalHandler +{ + public static void Handle(OrderPlacedSignal message, ILogger logger) => + logger.LogInformation( + "CW-spike: Inventory observed OrderPlacedSignal for order {OrderId} (total {Total})", + message.OrderId, message.Total); +} diff --git a/src/CritterMart.Orders/Analytics/OrderLineItemsProjection.cs b/src/CritterMart.Orders/Analytics/OrderLineItemsProjection.cs new file mode 100644 index 0000000..31e7add --- /dev/null +++ b/src/CritterMart.Orders/Analytics/OrderLineItemsProjection.cs @@ -0,0 +1,70 @@ +using CritterMart.Orders.Ordering; +using JasperFx.Events; +using Marten; +using Marten.Events.Projections; +using Weasel.Postgresql.Tables; + +namespace CritterMart.Orders.Analytics; + +// ════════════════════════════════════════════════════════════════════════════════════════════════ +// CW-TELEMETRY SPIKE (research/cw-telemetry-spike) — NOT round-one baseline. Async; only runs with +// the daemon on (Cw:Telemetry). See docs/research/cw-telemetry-fodder.md. +// ════════════════════════════════════════════════════════════════════════════════════════════════ + +// A FLAT-TABLE read model — raw SQL rows, not a JSONB document: one row per (order, line) in +// orders.order_line_items, the kind of table a BI tool or EF Core report would consume. +// +// Implemented as an EventProjection (not the declarative FlatTableProjection) for two reasons the +// DSL can't satisfy: (1) we want event METADATA — placed_at (the event timestamp) and event_sequence +// — which FlatTableProjection deliberately can't reach; (2) one OrderPlaced must produce MANY rows +// (one per line), which the per-event single-row DSL doesn't fan out to. QueueSqlCommand batches the +// inserts into the projection's unit of work. +// +// Its real job here is to probe whether CritterWatch's Store Inspector / Event Store Explorer renders +// a non-document projection AT ALL — a question the baseline (JSONB documents only) can't ask. +public partial class OrderLineItemsProjection : EventProjection +{ + public OrderLineItemsProjection() + { + // Marten manages this table's schema (create + migrate) alongside the Orders store objects. + var table = new Table("orders.order_line_items"); + table.AddColumn("order_id").AsPrimaryKey(); + table.AddColumn("sku").AsPrimaryKey(); + table.AddColumn("product_name").NotNull(); + table.AddColumn("quantity").NotNull(); + table.AddColumn("unit_price").NotNull(); + table.AddColumn("line_total").NotNull(); + table.AddColumn("customer_id").NotNull(); + table.AddColumn("placed_at").NotNull(); + table.AddColumn("event_sequence").NotNull(); + SchemaObjects.Add(table); + + // Clear the table on rebuild so a "rebuild this projection" from CritterWatch is clean. + Options.DeleteDataInTableOnTeardown(table.Identifier.QualifiedName); + } + + public void Project(IEvent e, IDocumentOperations ops) + { + foreach (var line in e.Data.Items) + { + // ON CONFLICT DO NOTHING keeps the insert idempotent against at-least-once replays and + // async-daemon rebuilds — the (order, sku) pair is the natural key. + ops.QueueSqlCommand( + """ + INSERT INTO orders.order_line_items + (order_id, sku, product_name, quantity, unit_price, line_total, customer_id, placed_at, event_sequence) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT (order_id, sku) DO NOTHING + """, + e.Data.OrderId, + line.Sku, + line.Name, + line.Quantity, + line.Price, + line.Quantity * line.Price, + e.Data.CustomerId, + e.Timestamp, + e.Sequence); + } + } +} diff --git a/src/CritterMart.Orders/Analytics/ProductSalesLeaderboard.cs b/src/CritterMart.Orders/Analytics/ProductSalesLeaderboard.cs new file mode 100644 index 0000000..6c140b6 --- /dev/null +++ b/src/CritterMart.Orders/Analytics/ProductSalesLeaderboard.cs @@ -0,0 +1,63 @@ +using CritterMart.Orders.Ordering; +using JasperFx.Events; +using Marten.Events.Projections; + +namespace CritterMart.Orders.Analytics; + +// ════════════════════════════════════════════════════════════════════════════════════════════════ +// CW-TELEMETRY SPIKE (research/cw-telemetry-spike) — NOT round-one baseline. ADR 008 (no async +// daemon) still holds on `main`; this projection only MOVES when the daemon is flipped on behind +// the Cw:Telemetry flag. See docs/research/cw-telemetry-fodder.md. +// ════════════════════════════════════════════════════════════════════════════════════════════════ + +// A per-SKU sales leaderboard. Where CartAbandonmentReport folds MANY Cart streams into ONE document +// (many-streams → one-doc), this is the mirror topology — ONE OrderPlaced event fans OUT to MANY +// documents (one-event → many-docs), one ProductSalesLeaderboard per SKU on the order. That fan-out +// is exactly the multi-stream shape CritterWatch's Projection Stepper "Stream Slice" / "Tag Query" +// source modes exist for but that the baseline (inline-only, no daemon) never produces. +public class ProductSalesLeaderboard +{ + public string Id { get; set; } = string.Empty; // the SKU — the document identity + public string ProductName { get; set; } = string.Empty; + public int UnitsSold { get; set; } + public decimal GrossRevenue { get; set; } + public int OrderCount { get; set; } // orders that included this SKU +} + +// `partial` is load-bearing (Marten 9 source-gen convention, same as CartAbandonmentReportProjection): +// conventional Apply methods are dispatched by the compile-time JasperFx generator, which extends a +// partial class. Without it the host refuses to boot with InvalidProjectionException. +public partial class ProductSalesLeaderboardProjection + : MultiStreamProjection +{ + public ProductSalesLeaderboardProjection() + { + // Fan-out routing: each OrderPlaced is routed to one document per DISTINCT SKU it contains. + // Identities (plural) is Marten's one-event-updates-many-documents primitive. + Identities>(e => e.Data.Items.Select(i => i.Sku).Distinct().ToList()); + } + + // view.Id is the SKU this particular document tracks (Marten assigns the routed identity to the + // document key), so accumulate only the order line(s) matching it. + public void Apply(IEvent e, ProductSalesLeaderboard view) + { + var lines = e.Data.Items.Where(i => i.Sku == view.Id).ToList(); + if (lines.Count == 0) + { + return; + } + + foreach (var line in lines) + { + if (string.IsNullOrEmpty(view.ProductName)) + { + view.ProductName = line.Name; + } + + view.UnitsSold += line.Quantity; + view.GrossRevenue += line.Quantity * line.Price; + } + + view.OrderCount++; + } +} diff --git a/src/CritterMart.Orders/Features/PlaceOrder.cs b/src/CritterMart.Orders/Features/PlaceOrder.cs index e9fe068..bb5938a 100644 --- a/src/CritterMart.Orders/Features/PlaceOrder.cs +++ b/src/CritterMart.Orders/Features/PlaceOrder.cs @@ -1,6 +1,7 @@ using CritterMart.Orders.Customers; using CritterMart.Orders.Ordering; using CritterMart.Orders.Shopping; +using CritterMart.Orders.Spike; using Marten; using Microsoft.AspNetCore.Http; using Microsoft.AspNetCore.Mvc; @@ -27,15 +28,20 @@ public static class PlaceOrderEndpoint // goes out, and the payment deadline is set in the same step that placed the order (the Bruun // temporal automation's starting gun; Workshop slice 4.1 writes-to). On a rejection there is no // order, so both cascades are null (Wolverine skips null cascading messages). + // CW-TELEMETRY SPIKE: the fourth tuple member, Contracts.OrderPlacedSignal?, is a broadcast + // notification fanned out to Inventory + Catalog when Cw:Telemetry is on (null otherwise, and on + // every rejection path — Wolverine skips null cascades). It is pure CritterWatch topology fodder + // and changes no order behaviour. See docs/research/cw-telemetry-fodder.md. [WolverinePost("/orders")] - public static async Task<(IResult, Contracts.ReserveStock?, DeliveryMessage?)> Post( + public static async Task<(IResult, Contracts.ReserveStock?, DeliveryMessage?, Contracts.OrderPlacedSignal?)> Post( [FromHeader(Name = "X-Customer-Id")] string? customerId, - IDocumentSession session, [FromServices] PaymentDeadline deadline) + IDocumentSession session, [FromServices] PaymentDeadline deadline, + [FromServices] CwTelemetryFlag cwTelemetry) { // A missing/blank header is a malformed request — 400, consistent with GET /orders/mine and // GET /carts/mine (ADR 009; the Polecat promotion swaps the header for a claim). if (string.IsNullOrWhiteSpace(customerId)) - return (Results.BadRequest("X-Customer-Id header is required."), null, null); + return (Results.BadRequest("X-Customer-Id header is required."), null, null, null); // Resolve the customer's open cart — the same indexed Cart query AddToCart uses. // A cart that was already checked out has IsOpen=false, so a repeat PlaceOrder finds no @@ -50,7 +56,7 @@ public static class PlaceOrderEndpoint return (Results.Problem( title: "NoOpenCart", detail: $"Customer '{customerId}' has no open cart to place.", - statusCode: StatusCodes.Status409Conflict), null, null); + statusCode: StatusCodes.Status409Conflict), null, null, null); } // Defensive guard for the workshop's CartEmpty path. Unreachable in 4.1 (a cart is @@ -61,7 +67,7 @@ public static class PlaceOrderEndpoint return (Results.Problem( title: "CartEmpty", detail: $"Customer '{customerId}' has an empty cart.", - statusCode: StatusCodes.Status409Conflict), null, null); + statusCode: StatusCodes.Status409Conflict), null, null, null); } var orderId = Guid.NewGuid().ToString(); @@ -90,7 +96,13 @@ public static class PlaceOrderEndpoint // terminal guard makes it a no-op; if not, the order is cancelled and its stock released. var paymentTimeout = new OrderPaymentTimeout(orderId).DelayedFor(deadline.Duration); - return (Results.Created($"/orders/{orderId}", new PlaceOrderResponse(orderId)), reserveStock, paymentTimeout); + // CW-TELEMETRY SPIKE: broadcast the placed order to Inventory + Catalog (fan-out topology) + // only when the flag is on — null keeps the baseline order flow byte-for-byte unchanged. + var orderPlacedSignal = cwTelemetry.Enabled + ? new Contracts.OrderPlacedSignal(orderId, customerId, total) + : null; + + return (Results.Created($"/orders/{orderId}", new PlaceOrderResponse(orderId)), reserveStock, paymentTimeout, orderPlacedSignal); } } diff --git a/src/CritterMart.Orders/Program.cs b/src/CritterMart.Orders/Program.cs index 2676ea3..7d6b901 100644 --- a/src/CritterMart.Orders/Program.cs +++ b/src/CritterMart.Orders/Program.cs @@ -1,6 +1,9 @@ +using CritterMart.Orders.Analytics; using CritterMart.Orders.Ordering; using CritterMart.Orders.Shopping; +using CritterMart.Orders.Spike; using JasperFx.Events; +using JasperFx.Events.Daemon; using JasperFx.Events.Projections; using JasperFx.OpenTelemetry; using Marten; @@ -39,7 +42,7 @@ // "now" — injected as TimeProvider so tests can drive the clock instead of waiting real time. builder.Services.AddSingleton(TimeProvider.System); -builder.Services.AddMarten(opts => +var martenConfig = builder.Services.AddMarten(opts => { opts.Connection(connectionString); @@ -85,6 +88,15 @@ // talk's teaching beat, not a bug. opts.Projections.Add(ProjectionLifecycle.Async); + // ── CW-TELEMETRY SPIKE projections (research/cw-telemetry-spike) — NOT round-one baseline ── + // Two ASYNC read models that only MOVE when the daemon is on (Cw:Telemetry, below). They give + // CritterWatch the live async telemetry the baseline never produces: ProductSalesLeaderboard is + // a fan-out multi-stream projection (one OrderPlaced → one doc per SKU); OrderLineItemsProjection + // is a flat SQL table (orders.order_line_items). Registered unconditionally — exactly like the + // CartAbandonment teaser above, they sit inert until a daemon turns. See docs/research/. + opts.Projections.Add(ProjectionLifecycle.Async); + opts.Projections.Add(ProjectionLifecycle.Async); + // The open-cart invariant lives on the Cart AGGREGATE (ADR 020 — it is a write-side rule): // a partial-unique index on Cart.CustomerId, scoped to open carts, enforces "one open cart // per customer" at the DB. The write paths resolve the customer's open cart against this index; a @@ -125,6 +137,21 @@ .IntegrateWithWolverine() .ApplyAllDatabaseChangesOnStartup(); +// ── CW-TELEMETRY SPIKE: async daemon (research/cw-telemetry-spike) — NOT round-one baseline ──────── +// ADR 008 keeps the baseline daemon-free; the spike flips it ON behind Cw:Telemetry so the async +// projections above (plus the previously-inert CartAbandonment teaser) actually run and stream +// shard / lag / rebuild telemetry to CritterWatch — letting you WATCH lag climb then drain in the +// console. Solo mode = single node, right for the single-instance Aspire host. Flag OFF = exact +// baseline behaviour (the "before" CritterWatch picture). See docs/research/cw-telemetry-fodder.md. +var cwTelemetry = builder.Configuration.GetValue("Cw:Telemetry"); +if (cwTelemetry) +{ + martenConfig.AddAsyncDaemon(DaemonMode.Solo); +} + +// Expose the toggle so the PlaceOrder endpoint can decide whether to broadcast OrderPlacedSignal. +builder.Services.AddSingleton(new CwTelemetryFlag(cwTelemetry)); + builder.Host.UseWolverine(opts => { // Pin handler/endpoint discovery to this service's assembly. Explicit (not auto-detected) diff --git a/src/CritterMart.Orders/Spike/CwTelemetryFlag.cs b/src/CritterMart.Orders/Spike/CwTelemetryFlag.cs new file mode 100644 index 0000000..9f95622 --- /dev/null +++ b/src/CritterMart.Orders/Spike/CwTelemetryFlag.cs @@ -0,0 +1,13 @@ +namespace CritterMart.Orders.Spike; + +// ════════════════════════════════════════════════════════════════════════════════════════════════ +// CW-TELEMETRY SPIKE (research/cw-telemetry-spike) — NOT round-one baseline. +// ════════════════════════════════════════════════════════════════════════════════════════════════ + +// The one runtime toggle (config key Cw:Telemetry, env Cw__Telemetry) gating the spike's ACTIVE +// behaviour: the async daemon (Program.cs) and the OrderPlacedSignal broadcast (PlaceOrder). Flag +// OFF reproduces the round-one baseline CritterWatch picture (inline-only, no async progress, no +// cross-BC topology beyond stock reservation); flag ON lights the dark surfaces. Registered as a +// singleton so the PlaceOrder endpoint can decide whether to broadcast. Mirrors the +// PaymentDeadline / PaymentDeclinePolicy config-singleton pattern already used in this service. +public record CwTelemetryFlag(bool Enabled); diff --git a/src/CritterMart.Orders/Spike/PoisonPing.cs b/src/CritterMart.Orders/Spike/PoisonPing.cs new file mode 100644 index 0000000..f41d8df --- /dev/null +++ b/src/CritterMart.Orders/Spike/PoisonPing.cs @@ -0,0 +1,34 @@ +using Microsoft.AspNetCore.Http; +using Wolverine.Http; + +namespace CritterMart.Orders.Spike; + +// ════════════════════════════════════════════════════════════════════════════════════════════════ +// CW-TELEMETRY SPIKE (research/cw-telemetry-spike) — NOT round-one baseline. On-demand only. +// See docs/research/cw-telemetry-fodder.md. +// ════════════════════════════════════════════════════════════════════════════════════════════════ + +// A self-contained POISON message for exercising the two CritterWatch surfaces a healthy system +// never populates: the Dead Letters tab and the Projection-Statuses "Error" column (always `—` on +// the baseline). The handler always throws; Wolverine exhausts its attempts and moves the message to +// the durable dead-letter store the console reads. A DEDICATED message type so it never corrupts a +// real domain stream — the failure is isolated and repeatable. +public record PoisonPing(string Note); + +public static class PoisonPingHandler +{ + // Always throws. With no matching retry policy Wolverine dead-letters it — exactly the signal + // we want CritterWatch to surface. + public static void Handle(PoisonPing message) => + throw new InvalidOperationException( + $"CW-telemetry spike: intentional poison message — {message.Note}"); +} + +public static class PoisonEndpoint +{ + // Cascade a PoisonPing onto the bus. Wolverine.Http treats the IResult as the HTTP response and + // publishes the tuple's message member; the local handler above then throws and dead-letters it. + [WolverinePost("/spike/poison")] + public static (IResult, PoisonPing) Post() => + (Results.Accepted(), new PoisonPing($"triggered at {DateTimeOffset.UtcNow:O}")); +}