feat(#435): read-only per-cell projection progress query on IEventDatabase (2.x)

src/EventTests/Daemon/ReadProjectionProgressDefaultsTests.cs

@@ -0,0 +1,143 @@
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using JasperFx.Events;
+using JasperFx.Events.Daemon;
+using JasperFx.Events.Projections;
+using Shouldly;
+using Xunit;
+
+namespace EventTests.Daemon;
+
+/// <summary>
+/// jasperfx#435 — ReadProjectionProgressAsync ships as a default interface implementation so the
+/// out-of-tree IEventDatabase implementors (Marten, Polecat) keep compiling on the 1.x line. These
+/// pin the contract that matters to a consumer: "not implemented" and "no row yet" are different
+/// answers, and the default must not conflate them.
+/// </summary>
+public class ReadProjectionProgressDefaultsTests
+{
+    // A bare IEventDatabase that does NOT override ReadProjectionProgressAsync, so the call below
+    // exercises the default. Everything else throws — those members are never invoked here.
+    private sealed class BareEventDatabase : IEventDatabase
+    {
+        public string Identifier => throw new NotImplementedException();
+        public Uri DatabaseUri => throw new NotImplementedException();
+        public ShardStateTracker Tracker => throw new NotImplementedException();
+        public string StorageIdentifier => throw new NotImplementedException();
+
+        public Task StoreDeadLetterEventAsync(object storage, DeadLetterEvent deadLetterEvent, CancellationToken token)
+            => throw new NotImplementedException();
+
+        public Task EnsureStorageExistsAsync(Type storageType, CancellationToken token)
+            => throw new NotImplementedException();
+
+        public Task WaitForNonStaleProjectionDataAsync(TimeSpan timeout)
+            => throw new NotImplementedException();
+
+        public Task<long> ProjectionProgressFor(ShardName name, CancellationToken token = default)
+            => throw new NotImplementedException();
+
+        public Task<long?> FindEventStoreFloorAtTimeAsync(DateTimeOffset timestamp, CancellationToken token)
+            => throw new NotImplementedException();
+
+        public Task<long> FetchHighestEventSequenceNumber(CancellationToken token)
+            => throw new NotImplementedException();
+
+        public Task<IReadOnlyList<ShardState>> AllProjectionProgress(CancellationToken token = default)
+            => throw new NotImplementedException();
+    }
+
+    // An IEventDatabase that DOES implement the query, standing in for Marten/Polecat. Proves the
+    // member is overridable and that a real implementation's null survives as "no row".
+    private sealed class ProgressionEventDatabase : BareDatabaseBase
+    {
+        public override ValueTask<ProjectionProgressRow?> ReadProjectionProgressAsync(
+            string projectionName, string? tenantId, CancellationToken token)
+            => projectionName == "Orders"
+                ? new ValueTask<ProjectionProgressRow?>(
+                    new ProjectionProgressRow("Orders", tenantId, 42, "Running", null))
+                : new ValueTask<ProjectionProgressRow?>((ProjectionProgressRow?)null);
+    }
+
+    // Split out so the overriding stub does not have to restate every unrelated member.
+    private abstract class BareDatabaseBase : IEventDatabase
+    {
+        public virtual ValueTask<ProjectionProgressRow?> ReadProjectionProgressAsync(
+            string projectionName, string? tenantId, CancellationToken token)
+            => throw new NotImplementedException();
+
+        public string Identifier => throw new NotImplementedException();
+        public Uri DatabaseUri => throw new NotImplementedException();
+        public ShardStateTracker Tracker => throw new NotImplementedException();
+        public string StorageIdentifier => throw new NotImplementedException();
+
+        public Task StoreDeadLetterEventAsync(object storage, DeadLetterEvent deadLetterEvent, CancellationToken token)
+            => throw new NotImplementedException();
+
+        public Task EnsureStorageExistsAsync(Type storageType, CancellationToken token)
+            => throw new NotImplementedException();
+
+        public Task WaitForNonStaleProjectionDataAsync(TimeSpan timeout)
+            => throw new NotImplementedException();
+
+        public Task<long> ProjectionProgressFor(ShardName name, CancellationToken token = default)
+            => throw new NotImplementedException();
+
+        public Task<long?> FindEventStoreFloorAtTimeAsync(DateTimeOffset timestamp, CancellationToken token)
+            => throw new NotImplementedException();
+
+        public Task<long> FetchHighestEventSequenceNumber(CancellationToken token)
+            => throw new NotImplementedException();
+
+        public Task<IReadOnlyList<ShardState>> AllProjectionProgress(CancellationToken token = default)
+            => throw new NotImplementedException();
+    }
+
+    private readonly IEventDatabase theBareDatabase = new BareEventDatabase();
+    private readonly IEventDatabase theProgressionDatabase = new ProgressionEventDatabase();
+
+    [Fact]
+    public async Task default_throws_rather_than_reporting_a_live_cell_as_absent()
+    {
+        // The critical distinction: null is the documented "no row for this pair yet" answer, so an
+        // unimplemented store must not return it. A monitoring UI would render a running projection
+        // as having no progress at all.
+        await Should.ThrowAsync<NotSupportedException>(async () =>
+            await theBareDatabase.ReadProjectionProgressAsync("Orders", null, CancellationToken.None));
+    }
+
+    [Fact]
+    public async Task default_throws_for_a_tenant_scoped_read_too()
+    {
+        await Should.ThrowAsync<NotSupportedException>(async () =>
+            await theBareDatabase.ReadProjectionProgressAsync("Orders", "tenant-1", CancellationToken.None));
+    }
+
+    [Fact]
+    public async Task an_implementing_store_can_return_a_row()
+    {
+        var row = await theProgressionDatabase.ReadProjectionProgressAsync("Orders", "tenant-1", CancellationToken.None);
+
+        row.ShouldNotBeNull();
+        row.ProjectionName.ShouldBe("Orders");
+        row.TenantId.ShouldBe("tenant-1");
+        row.Sequence.ShouldBe(42);
+        row.AgentStatus.ShouldBe("Running");
+        row.LastHeartbeat.ShouldBeNull();
+    }
+
+    [Fact]
+    public async Task an_implementing_store_returns_null_when_no_row_exists_for_the_pair()
+    {
+        var row = await theProgressionDatabase.ReadProjectionProgressAsync("Unobserved", null, CancellationToken.None);
+        row.ShouldBeNull();
+    }
+
+    [Fact]
+    public void null_tenant_id_means_store_global_or_default_tenant()
+    {
+        new ProjectionProgressRow("Orders", null, 1, "Running", null).TenantId.ShouldBeNull();
+    }
+}

src/JasperFx.Events/IEventDatabase.cs

@@ -174,4 +174,31 @@ Task<IReadOnlyList<DeadLetterShardCount>> FetchDeadLetterCountsAsync(string? ten
             ? FetchDeadLetterCountsAsync(token)
             : throw new NotSupportedException(
                 "Per-tenant FetchDeadLetterCountsAsync is not implemented on this IEventDatabase. Use an event store that implements per-tenant partitioning.");
+
+    /// <summary>
+    ///     Read the current sequence + lifecycle state for a single (projection, tenant) cell directly
+    ///     from the progression table, without spinning up an <see cref="Daemon.IProjectionDaemon" />.
+    ///     Returns null when no row exists for the pair yet — e.g. the daemon has not yet observed this
+    ///     projection for this tenant. A null <paramref name="tenantId" /> means store-global on a
+    ///     non-tenanted store, or the default-tenant row on a tenanted store.
+    ///     <para>
+    ///     This is the targeted counterpart to <see cref="AllProjectionProgress(CancellationToken)" />,
+    ///     which returns every row across every projection × tenant pair and leaves the caller to filter.
+    ///     For a per-cell UI polling loop (e.g. 1Hz against a single visible batch) the targeted query is
+    ///     materially cheaper.
+    ///     </para>
+    ///     The default implementation throws <see cref="NotSupportedException" /> rather than returning
+    ///     null: null is the meaningful "no row yet" answer, so a store that simply has not implemented
+    ///     this must not borrow it and report a live cell as absent. Event stores (Marten, Polecat)
+    ///     override this against their progression table. See jasperfx#435.
+    /// </summary>
+    /// <param name="projectionName">Name of the projection whose cell to read.</param>
+    /// <param name="tenantId">Tenant owning the cell. Null means store-global / default tenant.</param>
+    /// <param name="token"></param>
+    ValueTask<ProjectionProgressRow?> ReadProjectionProgressAsync(
+        string projectionName,
+        string? tenantId,
+        CancellationToken token)
+        => throw new NotSupportedException(
+            "ReadProjectionProgressAsync is not implemented on this IEventDatabase. Use an event store (Marten or Polecat) that supports reading a single projection progression row.");
 }

src/JasperFx.Events/ProjectionProgressRow.cs

@@ -0,0 +1,29 @@
+namespace JasperFx.Events;
+
+/// <summary>
+/// Point in time state of a single (projection, tenant) progression cell, read straight from the
+/// event store's progression table. This is the targeted, per-cell counterpart to
+/// <see cref="IEventDatabase.AllProjectionProgress" /> — it exists so a monitoring tool polling one
+/// visible cell does not have to fetch and filter every projection × tenant row on each tick.
+/// See jasperfx#435.
+/// </summary>
+/// <param name="ProjectionName">Name of the projection this cell tracks.</param>
+/// <param name="TenantId">
+/// Tenant the cell belongs to. Null means store-global on a non-tenanted store, or the
+/// default-tenant row on a tenanted store.
+/// </param>
+/// <param name="Sequence">Event sequence number this cell has processed through.</param>
+/// <param name="AgentStatus">
+/// Lifecycle state of the agent driving this cell. Left as a string rather than the
+/// <see cref="JasperFx.AgentStatus" /> enum on purpose: this is a diagnostic read of whatever the
+/// store persisted, and a store may report a state outside the enum's Running/Stopped/Paused.
+/// </param>
+/// <param name="LastHeartbeat">
+/// Timestamp the cell last reported progress; null when the store does not track a heartbeat for it.
+/// </param>
+public record ProjectionProgressRow(
+    string ProjectionName,
+    string? TenantId,
+    long Sequence,
+    string AgentStatus,
+    DateTimeOffset? LastHeartbeat);