From 3882dbd90d8e0d6349ba9af4032ce6866ee61873 Mon Sep 17 00:00:00 2001 From: bitbiter-dev Date: Tue, 20 Oct 2026 20:32:04 +1300 Subject: [PATCH 1/3] fix(worker): Move a work item's UpdatedAt on a heartbeat only when its progress changed (#279) UpdatedAt is the change cursor the admin event stream reads. A heartbeat that only extends the lease changed nothing an admin can see, so it no longer moves UpdatedAt; LeaseUntil already records that the Worker is alive. Co-Authored-By: Claude Opus 5.5 --- docs/schema.sql | 2 +- src/Anichron.Core/Data/AnichronDbContext.cs | 4 +- ..._CommentUpdatedAtMovesOnChange.Designer.cs | 693 ++++++++++++++++++ ...020073145_CommentUpdatedAtMovesOnChange.cs | 39 + .../AnichronDbContextModelSnapshot.cs | 5 +- .../Work/WorkItemStoreTests.cs | 50 ++ src/Anichron.Worker/Work/EfWorkItemStore.cs | 6 +- 7 files changed, 793 insertions(+), 6 deletions(-) create mode 100644 src/Anichron.Core/Migrations/20261020073145_CommentUpdatedAtMovesOnChange.Designer.cs create mode 100644 src/Anichron.Core/Migrations/20261020073145_CommentUpdatedAtMovesOnChange.cs diff --git a/docs/schema.sql b/docs/schema.sql index e7af92e..435cf7c 100644 --- a/docs/schema.sql +++ b/docs/schema.sql @@ -95,7 +95,7 @@ COMMENT ON COLUMN "WorkItems"."Attempt" IS 'Incremented by every claim. Every wr COMMENT ON COLUMN "WorkItems"."Stalls" IS 'Lease expiries, counted by lease recovery. A graceful shutdown hands the item back without counting one; the item fails as stalled once the limit is reached.'; COMMENT ON COLUMN "WorkItems"."ClaimedBy" IS 'The claiming Worker for display: its WORKER__USER or ''all'', then its host name.'; COMMENT ON COLUMN "WorkItems"."LeaseUntil" IS 'Extended by every heartbeat. A Running item past it belongs to a dead or paused Worker and goes back to the queue.'; -COMMENT ON COLUMN "WorkItems"."UpdatedAt" IS 'Moves on every insert, claim, heartbeat and transition, so a reader can ask what changed since a point in time.'; +COMMENT ON COLUMN "WorkItems"."UpdatedAt" IS 'Moves on every insert, claim and transition, and on a heartbeat only when the progress snapshot changed, so a reader can ask what changed since a point in time.'; COMMENT ON COLUMN "WorkItems"."ProgressNote" IS 'The latest progress snapshot (done, optional total, note), overwritten by each heartbeat; no history is kept.'; COMMENT ON COLUMN "WorkItems"."Result" IS 'The handler''s results summary, written when the item succeeds.'; diff --git a/src/Anichron.Core/Data/AnichronDbContext.cs b/src/Anichron.Core/Data/AnichronDbContext.cs index 36ed79c..c30a815 100644 --- a/src/Anichron.Core/Data/AnichronDbContext.cs +++ b/src/Anichron.Core/Data/AnichronDbContext.cs @@ -346,8 +346,8 @@ protected override void OnModelCreating(ModelBuilder modelBuilder) "Extended by every heartbeat. A Running item past it belongs to a dead or paused " + "Worker and goes back to the queue."); entity.Property(w => w.UpdatedAt).HasComment( - "Moves on every insert, claim, heartbeat and transition, so a reader can ask what " - + "changed since a point in time."); + "Moves on every insert, claim and transition, and on a heartbeat only when the progress " + + "snapshot changed, so a reader can ask what changed since a point in time."); entity.Property(w => w.ProgressNote).HasComment( "The latest progress snapshot (done, optional total, note), overwritten by each " + "heartbeat; no history is kept."); diff --git a/src/Anichron.Core/Migrations/20261020073145_CommentUpdatedAtMovesOnChange.Designer.cs b/src/Anichron.Core/Migrations/20261020073145_CommentUpdatedAtMovesOnChange.Designer.cs new file mode 100644 index 0000000..df52b70 --- /dev/null +++ b/src/Anichron.Core/Migrations/20261020073145_CommentUpdatedAtMovesOnChange.Designer.cs @@ -0,0 +1,693 @@ +// +using System; +using Anichron.Core.Data; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Infrastructure; +using Microsoft.EntityFrameworkCore.Migrations; +using Microsoft.EntityFrameworkCore.Storage.ValueConversion; +using NodaTime; +using Npgsql.EntityFrameworkCore.PostgreSQL.Metadata; + +#nullable disable + +namespace Anichron.Core.Migrations +{ + [DbContext(typeof(AnichronDbContext))] + [Migration("20261020073145_CommentUpdatedAtMovesOnChange")] + partial class CommentUpdatedAtMovesOnChange + { + /// + protected override void BuildTargetModel(ModelBuilder modelBuilder) + { +#pragma warning disable 612, 618 + modelBuilder + .HasAnnotation("ProductVersion", "10.0.13") + .HasAnnotation("Relational:MaxIdentifierLength", 63); + + NpgsqlModelBuilderExtensions.UseIdentityByDefaultColumns(modelBuilder); + + modelBuilder.Entity("Anichron.Core.Domain.AssetInteraction", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("uuid"); + + b.Property("AssetId") + .HasColumnType("uuid"); + + b.Property("IsHidden") + .HasColumnType("boolean"); + + b.Property("IsLiked") + .HasColumnType("boolean"); + + b.Property("IsStarred") + .HasColumnType("boolean"); + + b.Property("LastViewed") + .HasColumnType("timestamp with time zone"); + + b.Property("UserId") + .HasColumnType("uuid"); + + b.Property("ViewCount") + .HasColumnType("integer"); + + b.HasKey("Id"); + + b.HasIndex("AssetId"); + + b.HasIndex("UserId", "AssetId") + .IsUnique() + .HasDatabaseName("IX_Interactions_UserId_AssetId"); + + b.ToTable("Interactions", t => + { + t.HasComment("Per-user state for one asset. Never shared between users: two people looking at the same file each get their own row. Drives flashback weighting — hidden excludes, starred always includes and sorts first, liked raises DisplayWeight."); + }); + }); + + modelBuilder.Entity("Anichron.Core.Domain.Burst", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("uuid"); + + b.Property("CreatedAt") + .HasColumnType("timestamp with time zone"); + + b.Property("PrimaryAssetId") + .HasColumnType("uuid") + .HasComment("The frame chosen to represent the burst. Restrict on delete: losing the cover would leave a group that cannot be rendered."); + + b.HasKey("Id"); + + b.HasIndex("PrimaryAssetId"); + + b.ToTable("Bursts", t => + { + t.HasComment("Groups a rapid-fire sequence so a flashback shows one memory rather than nine near-identical frames. The table and schema exist; burst DETECTION is not yet implemented in the Worker, so in a running system this table is empty."); + }); + }); + + modelBuilder.Entity("Anichron.Core.Domain.Invite", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("uuid"); + + b.Property("CreatedAt") + .HasColumnType("timestamp with time zone"); + + b.Property("CreatedByUserId") + .HasColumnType("uuid"); + + b.Property("ExpiresAt") + .HasColumnType("timestamp with time zone"); + + b.Property("TokenHash") + .IsRequired() + .HasMaxLength(44) + .HasColumnType("character varying(44)") + .HasComment("Hash, never the token itself. 44 chars — base64 of a 32-byte value."); + + b.Property("UsedAt") + .HasColumnType("timestamp with time zone"); + + b.Property("UsedByUserId") + .HasColumnType("uuid"); + + b.Property("xmin") + .IsConcurrencyToken() + .ValueGeneratedOnAddOrUpdate() + .HasColumnType("xid") + .HasColumnName("xmin"); + + b.HasKey("Id"); + + b.HasIndex("CreatedByUserId"); + + b.HasIndex("TokenHash") + .IsUnique(); + + b.HasIndex("UsedByUserId"); + + b.ToTable("Invites", t => + { + t.HasComment("Admin-issued registration tokens. Registration requires one; the first user (bootstrap admin) is exempt. Redemption is a read-then-write race, which is why this is the only table carrying a concurrency token."); + }); + }); + + modelBuilder.Entity("Anichron.Core.Domain.MediaAsset", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("uuid"); + + b.Property("BurstId") + .HasColumnType("uuid"); + + b.Property("ContentHash") + .IsRequired() + .HasColumnType("text") + .HasComment("XXHash64 of the file contents. Dedup is deliberately CONFIG-SCOPED: the same bytes under two storage configs are two assets, which is what makes multi-user work. A global unique index on this column alone would break that and must never be added."); + + b.Property("DateCaptured") + .HasColumnType("timestamp without time zone"); + + b.Property("Day") + .HasColumnType("integer") + .HasComment("See Month — the pair backs IX_MediaAsset_Flashback."); + + b.Property("FileName") + .IsRequired() + .HasColumnType("text"); + + b.Property("FilePath") + .IsRequired() + .HasColumnType("text") + .HasComment("Path relative to the storage config root. Not to be confused with ProxyFiles.ProxyPath, which is relative to the proxy root."); + + b.Property("IsSoftDeleted") + .HasColumnType("boolean") + .HasComment("Set by reconciliation when a file is gone from the NAS. Soft rather than hard so the user's interactions survive a temporarily unmounted share."); + + b.Property("LastSeenOnNas") + .HasColumnType("timestamp with time zone"); + + b.Property("MediaType") + .IsRequired() + .HasColumnType("text"); + + b.Property("Month") + .HasColumnType("integer") + .HasComment("Stored apart from DateCaptured so \"On This Day\" is an index lookup on (Month, Day) rather than an EXTRACT() scan over every row."); + + b.Property("PairedAssetId") + .HasColumnType("uuid") + .HasComment("Live Photo pairing: the HEIC parent points at its MOV child. SetNull on delete so losing one half leaves the other a usable standalone asset."); + + b.Property("StorageConfigId") + .HasColumnType("uuid"); + + b.Property("Year") + .HasColumnType("integer"); + + b.HasKey("Id"); + + b.HasIndex("BurstId"); + + b.HasIndex("IsSoftDeleted") + .HasDatabaseName("IX_MediaAssets_Active") + .HasFilter("\"IsSoftDeleted\" = false"); + + b.HasIndex("PairedAssetId"); + + b.HasIndex("Month", "Day") + .HasDatabaseName("IX_MediaAsset_Flashback"); + + b.HasIndex("StorageConfigId", "ContentHash") + .IsUnique() + .HasDatabaseName("ix_mediaassets_storageconfigid_contenthash_active") + .HasFilter("\"IsSoftDeleted\" = false"); + + b.HasIndex("StorageConfigId", "FilePath") + .IsUnique() + .HasDatabaseName("IX_MediaAssets_StorageConfigId_FilePath"); + + b.ToTable("MediaAssets", t => + { + t.HasComment("One row per media file, scoped to a storage config. Written ONLY by the Worker — the API never writes media data."); + }); + }); + + modelBuilder.Entity("Anichron.Core.Domain.Metadata", b => + { + b.Property("AssetId") + .HasColumnType("uuid"); + + b.Property("Blurhash") + .HasColumnType("text") + .HasComment("BlurHash placeholder string (4x3 components), computed by the Worker from the upright image. Kept here rather than as a proxy file so a response needs no disk read for it. Null for videos, and for images ingested before the column existed, which are not backfilled."); + + b.Property("CameraMake") + .HasColumnType("text"); + + b.Property("CameraModel") + .HasColumnType("text"); + + b.Property("DurationInSeconds") + .HasColumnType("integer"); + + b.Property("Height") + .HasColumnType("integer"); + + b.Property("Latitude") + .HasColumnType("double precision"); + + b.Property("LensModel") + .HasColumnType("text"); + + b.Property("Longitude") + .HasColumnType("double precision"); + + b.Property("Orientation") + .ValueGeneratedOnAdd() + .HasColumnType("integer") + .HasDefaultValue(1) + .HasComment("Raw EXIF orientation tag, 1-8 (1 = upright, also used when absent or invalid). Stored raw because 5 and 7 combine a rotation with a mirror and cannot be expressed as degrees; a reader converts, and for 5-8 swaps Width/Height. Proxies keep this tag on unrotated pixels, so do not rotate them twice."); + + b.Property("Width") + .HasColumnType("integer"); + + b.HasKey("AssetId"); + + b.ToTable("Metadata", t => + { + t.HasComment("EXIF data, 1:1 with MediaAsset and keyed by it — there is no separate Id. Populated by ExifExtractionMiddleware; every field is best-effort, because a file may carry no EXIF at all."); + }); + }); + + modelBuilder.Entity("Anichron.Core.Domain.ProxyFile", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("uuid"); + + b.Property("AssetId") + .HasColumnType("uuid"); + + b.Property("CreatedAt") + .HasColumnType("timestamp with time zone"); + + b.Property("ProxyPath") + .IsRequired() + .HasColumnType("text") + .HasComment("Path relative to the proxy root, sharded by CONTENT rather than by ingestion attempt so a retry overwrites its own debris instead of stranding it. IProxyDirectoryStrategy is the single place that rule lives."); + + b.Property("ProxyType") + .IsRequired() + .HasColumnType("text") + .HasComment("Thumbnail, WebVideo or FullPreview, persisted as text. The BlurHash placeholder is not a proxy: it is the Metadata.Blurhash column."); + + b.Property("SizeBytes") + .HasColumnType("bigint"); + + b.HasKey("Id"); + + b.HasIndex("AssetId", "ProxyType") + .IsUnique() + .HasDatabaseName("ix_proxyfiles_assetid_proxytype"); + + b.ToTable("ProxyFiles", t => + { + t.HasComment("One row per generated artefact on the local SSD. An asset has several, but at most one per ProxyType, so of two attempts racing to add the same proxy only one row lands. Written only by ProxyStagingWriter, at the moment the file reaches its final name."); + }); + }); + + modelBuilder.Entity("Anichron.Core.Domain.RefreshToken", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("uuid"); + + b.Property("CreatedAt") + .HasColumnType("timestamp with time zone"); + + b.Property("ExpiresAt") + .HasColumnType("timestamp with time zone"); + + b.Property("RevokedAt") + .HasColumnType("timestamp with time zone"); + + b.Property("TokenHash") + .IsRequired() + .HasColumnType("text") + .HasComment("Hash, never the token itself — a database disclosure must not hand out usable sessions."); + + b.Property("UserId") + .HasColumnType("uuid"); + + b.HasKey("Id"); + + b.HasIndex("TokenHash") + .IsUnique(); + + b.HasIndex("UserId", "ExpiresAt"); + + b.ToTable("RefreshTokens", t => + { + t.HasComment("Issued and revoked by the API; expired rows are swept by the Worker's TokenCleanup work item. One of the few tables both services touch."); + }); + }); + + modelBuilder.Entity("Anichron.Core.Domain.User", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("uuid"); + + b.Property("Email") + .IsRequired() + .HasColumnType("text") + .HasComment("Always stored trimmed and lowercased: an EF value converter applies IdentityNormalizer.Normalize on every write and to every equality comparison against this column, so the case-sensitive unique index cannot hold two spellings of one name."); + + b.Property("FailedLoginAttempts") + .HasColumnType("integer"); + + b.Property("IsAdmin") + .HasColumnType("boolean"); + + b.Property("IsDisabled") + .HasColumnType("boolean"); + + b.Property("LockedUntil") + .HasColumnType("timestamp with time zone"); + + b.Property("MustChangePassword") + .HasColumnType("boolean") + .HasComment("Set by an admin reset. MustChangePasswordMiddleware then 403s every route except the handful needed to change it."); + + b.Property("PasswordHash") + .IsRequired() + .HasColumnType("text") + .HasComment("Argon2id. The parameters live with the hasher, not here, so rotating them does not require a migration."); + + b.Property("Username") + .IsRequired() + .HasColumnType("text") + .HasComment("Always stored trimmed and lowercased: an EF value converter applies IdentityNormalizer.Normalize on every write and to every equality comparison against this column, so the case-sensitive unique index cannot hold two spellings of one name."); + + b.HasKey("Id"); + + b.HasIndex("Email") + .IsUnique() + .HasDatabaseName("ix_users_email"); + + b.HasIndex("Username") + .IsUnique() + .HasDatabaseName("ix_users_username"); + + b.ToTable("Users", t => + { + t.HasComment("Owns storage configs and interactions. All identity writes happen in the API; the Worker only reads."); + }); + }); + + modelBuilder.Entity("Anichron.Core.Domain.UserStorageConfig", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("uuid"); + + b.Property("IsActive") + .HasColumnType("boolean"); + + b.Property("RootPath") + .IsRequired() + .HasColumnType("text") + .HasComment("GLOBALLY unique, not per user: a path cannot be assigned to two configs across all users. Two users indexing the same directory would duplicate every asset and let one user's config removal cascade away the other's rows."); + + b.Property("UserId") + .HasColumnType("uuid"); + + b.HasKey("Id"); + + b.HasIndex("RootPath") + .IsUnique() + .HasDatabaseName("IX_StorageConfigs_RootPath"); + + b.HasIndex("UserId"); + + b.ToTable("StorageConfigs", t => + { + t.HasComment("A NAS root assigned to one user. This is the boundary that scopes assets to a user and lets several Workers each monitor a separate config."); + }); + }); + + modelBuilder.Entity("Anichron.Core.Domain.WorkItem", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("uuid"); + + b.Property("Attempt") + .HasColumnType("integer") + .HasComment("Incremented by every claim. Every write a Worker makes while running an item is fenced by (Id, Attempt), so an attempt that lost its lease changes nothing."); + + b.Property("ClaimedBy") + .HasColumnType("text") + .HasComment("The claiming Worker for display: its WORKER__USER or 'all', then its host name."); + + b.Property("DedupKey") + .HasColumnType("text") + .HasComment("Opt-in coalescing. While an item with this key is Queued, enqueueing the same key returns that item instead of adding a second; the partial unique index enforces it."); + + b.Property("Error") + .HasColumnType("text"); + + b.Property("FinishedAt") + .HasColumnType("timestamp with time zone"); + + b.Property("Kind") + .IsRequired() + .HasColumnType("text") + .HasComment("WorkKind, persisted as text. Picks the payload type and the handler through the registry, never through a switch."); + + b.Property("LeaseUntil") + .HasColumnType("timestamp with time zone") + .HasComment("Extended by every heartbeat. A Running item past it belongs to a dead or paused Worker and goes back to the queue."); + + b.Property("Payload") + .IsRequired() + .HasColumnType("jsonb") + .HasComment("The kind's IWorkPayload, serialised with System.Text.Json and NodaTime settings."); + + b.Property("ProgressDone") + .HasColumnType("bigint"); + + b.Property("ProgressNote") + .HasColumnType("text") + .HasComment("The latest progress snapshot (done, optional total, note), overwritten by each heartbeat; no history is kept."); + + b.Property("ProgressTotal") + .HasColumnType("bigint"); + + b.Property("RequestedAt") + .HasColumnType("timestamp with time zone"); + + b.Property("RequestedBy") + .HasColumnType("uuid") + .HasComment("The admin who asked for the work; null when a Worker's scheduler enqueued it."); + + b.Property("Result") + .HasColumnType("jsonb") + .HasComment("The handler's results summary, written when the item succeeds."); + + b.Property("Stalls") + .HasColumnType("integer") + .HasComment("Lease expiries, counted by lease recovery. A graceful shutdown hands the item back without counting one; the item fails as stalled once the limit is reached."); + + b.Property("StartedAt") + .HasColumnType("timestamp with time zone"); + + b.Property("Status") + .IsRequired() + .HasColumnType("text") + .HasComment("Queued, Running, Succeeded, Failed or Cancelled, persisted as text."); + + b.Property("StorageConfigId") + .HasColumnType("uuid") + .HasComment("The storage config the work belongs to, or null for unscoped work. A per-user Worker claims only its user's configs and unscoped work; deleting the config deletes its items."); + + b.Property("UpdatedAt") + .HasColumnType("timestamp with time zone") + .HasComment("Moves on every insert, claim and transition, and on a heartbeat only when the progress snapshot changed, so a reader can ask what changed since a point in time."); + + b.HasKey("Id"); + + b.HasIndex("DedupKey") + .IsUnique() + .HasDatabaseName("ix_workitems_dedupkey_queued") + .HasFilter("\"Status\" = 'Queued' AND \"DedupKey\" IS NOT NULL"); + + b.HasIndex("LeaseUntil") + .HasDatabaseName("ix_workitems_lease") + .HasFilter("\"Status\" = 'Running'"); + + b.HasIndex("RequestedAt") + .HasDatabaseName("ix_workitems_claim") + .HasFilter("\"Status\" = 'Queued'"); + + b.HasIndex("RequestedBy"); + + b.HasIndex("StorageConfigId"); + + b.HasIndex("UpdatedAt"); + + b.ToTable("WorkItems", t => + { + t.HasComment("Durable work for the Worker: periodic maintenance today, admin requests later. Only IWorkQueue inserts rows; every later state change belongs to the Worker's executor. Delivery is at least once, so handlers start from scratch and converge."); + }); + }); + + modelBuilder.Entity("Anichron.Core.Domain.AssetInteraction", b => + { + b.HasOne("Anichron.Core.Domain.MediaAsset", "Asset") + .WithMany("Interactions") + .HasForeignKey("AssetId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired(); + + b.HasOne("Anichron.Core.Domain.User", "User") + .WithMany("Interactions") + .HasForeignKey("UserId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired(); + + b.Navigation("Asset"); + + b.Navigation("User"); + }); + + modelBuilder.Entity("Anichron.Core.Domain.Burst", b => + { + b.HasOne("Anichron.Core.Domain.MediaAsset", "PrimaryAsset") + .WithMany() + .HasForeignKey("PrimaryAssetId") + .OnDelete(DeleteBehavior.Restrict) + .IsRequired(); + + b.Navigation("PrimaryAsset"); + }); + + modelBuilder.Entity("Anichron.Core.Domain.Invite", b => + { + b.HasOne("Anichron.Core.Domain.User", "CreatedBy") + .WithMany() + .HasForeignKey("CreatedByUserId") + .OnDelete(DeleteBehavior.SetNull); + + b.HasOne("Anichron.Core.Domain.User", "UsedBy") + .WithMany() + .HasForeignKey("UsedByUserId") + .OnDelete(DeleteBehavior.SetNull); + + b.Navigation("CreatedBy"); + + b.Navigation("UsedBy"); + }); + + modelBuilder.Entity("Anichron.Core.Domain.MediaAsset", b => + { + b.HasOne("Anichron.Core.Domain.Burst", "Burst") + .WithMany("Assets") + .HasForeignKey("BurstId") + .OnDelete(DeleteBehavior.SetNull); + + b.HasOne("Anichron.Core.Domain.MediaAsset", "PairedAsset") + .WithMany() + .HasForeignKey("PairedAssetId") + .OnDelete(DeleteBehavior.SetNull); + + b.HasOne("Anichron.Core.Domain.UserStorageConfig", "StorageConfig") + .WithMany("Assets") + .HasForeignKey("StorageConfigId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired(); + + b.Navigation("Burst"); + + b.Navigation("PairedAsset"); + + b.Navigation("StorageConfig"); + }); + + modelBuilder.Entity("Anichron.Core.Domain.Metadata", b => + { + b.HasOne("Anichron.Core.Domain.MediaAsset", "Asset") + .WithOne("Metadata") + .HasForeignKey("Anichron.Core.Domain.Metadata", "AssetId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired(); + + b.Navigation("Asset"); + }); + + modelBuilder.Entity("Anichron.Core.Domain.ProxyFile", b => + { + b.HasOne("Anichron.Core.Domain.MediaAsset", "Asset") + .WithMany("ProxyFiles") + .HasForeignKey("AssetId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired(); + + b.Navigation("Asset"); + }); + + modelBuilder.Entity("Anichron.Core.Domain.RefreshToken", b => + { + b.HasOne("Anichron.Core.Domain.User", "User") + .WithMany("RefreshTokens") + .HasForeignKey("UserId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired(); + + b.Navigation("User"); + }); + + modelBuilder.Entity("Anichron.Core.Domain.UserStorageConfig", b => + { + b.HasOne("Anichron.Core.Domain.User", "User") + .WithMany("StorageConfigs") + .HasForeignKey("UserId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired(); + + b.Navigation("User"); + }); + + modelBuilder.Entity("Anichron.Core.Domain.WorkItem", b => + { + b.HasOne("Anichron.Core.Domain.User", null) + .WithMany() + .HasForeignKey("RequestedBy") + .OnDelete(DeleteBehavior.SetNull); + + b.HasOne("Anichron.Core.Domain.UserStorageConfig", null) + .WithMany() + .HasForeignKey("StorageConfigId") + .OnDelete(DeleteBehavior.Cascade); + }); + + modelBuilder.Entity("Anichron.Core.Domain.Burst", b => + { + b.Navigation("Assets"); + }); + + modelBuilder.Entity("Anichron.Core.Domain.MediaAsset", b => + { + b.Navigation("Interactions"); + + b.Navigation("Metadata"); + + b.Navigation("ProxyFiles"); + }); + + modelBuilder.Entity("Anichron.Core.Domain.User", b => + { + b.Navigation("Interactions"); + + b.Navigation("RefreshTokens"); + + b.Navigation("StorageConfigs"); + }); + + modelBuilder.Entity("Anichron.Core.Domain.UserStorageConfig", b => + { + b.Navigation("Assets"); + }); +#pragma warning restore 612, 618 + } + } +} diff --git a/src/Anichron.Core/Migrations/20261020073145_CommentUpdatedAtMovesOnChange.cs b/src/Anichron.Core/Migrations/20261020073145_CommentUpdatedAtMovesOnChange.cs new file mode 100644 index 0000000..7edb84c --- /dev/null +++ b/src/Anichron.Core/Migrations/20261020073145_CommentUpdatedAtMovesOnChange.cs @@ -0,0 +1,39 @@ +using Microsoft.EntityFrameworkCore.Migrations; +using NodaTime; + +#nullable disable + +namespace Anichron.Core.Migrations +{ + /// + public partial class CommentUpdatedAtMovesOnChange : Migration + { + /// + protected override void Up(MigrationBuilder migrationBuilder) + { + migrationBuilder.AlterColumn( + name: "UpdatedAt", + table: "WorkItems", + type: "timestamp with time zone", + nullable: false, + comment: "Moves on every insert, claim and transition, and on a heartbeat only when the progress snapshot changed, so a reader can ask what changed since a point in time.", + oldClrType: typeof(Instant), + oldType: "timestamp with time zone", + oldComment: "Moves on every insert, claim, heartbeat and transition, so a reader can ask what changed since a point in time."); + } + + /// + protected override void Down(MigrationBuilder migrationBuilder) + { + migrationBuilder.AlterColumn( + name: "UpdatedAt", + table: "WorkItems", + type: "timestamp with time zone", + nullable: false, + comment: "Moves on every insert, claim, heartbeat and transition, so a reader can ask what changed since a point in time.", + oldClrType: typeof(Instant), + oldType: "timestamp with time zone", + oldComment: "Moves on every insert, claim and transition, and on a heartbeat only when the progress snapshot changed, so a reader can ask what changed since a point in time."); + } + } +} diff --git a/src/Anichron.Core/Migrations/AnichronDbContextModelSnapshot.cs b/src/Anichron.Core/Migrations/AnichronDbContextModelSnapshot.cs index 347294a..d3c2dd7 100644 --- a/src/Anichron.Core/Migrations/AnichronDbContextModelSnapshot.cs +++ b/src/Anichron.Core/Migrations/AnichronDbContextModelSnapshot.cs @@ -55,7 +55,8 @@ protected override void BuildModel(ModelBuilder modelBuilder) b.HasIndex("AssetId"); b.HasIndex("UserId", "AssetId") - .IsUnique(); + .IsUnique() + .HasDatabaseName("IX_Interactions_UserId_AssetId"); b.ToTable("Interactions", t => { @@ -498,7 +499,7 @@ protected override void BuildModel(ModelBuilder modelBuilder) b.Property("UpdatedAt") .HasColumnType("timestamp with time zone") - .HasComment("Moves on every insert, claim, heartbeat and transition, so a reader can ask what changed since a point in time."); + .HasComment("Moves on every insert, claim and transition, and on a heartbeat only when the progress snapshot changed, so a reader can ask what changed since a point in time."); b.HasKey("Id"); diff --git a/src/Anichron.Tests.Integration/Work/WorkItemStoreTests.cs b/src/Anichron.Tests.Integration/Work/WorkItemStoreTests.cs index f97f92c..7c433b2 100644 --- a/src/Anichron.Tests.Integration/Work/WorkItemStoreTests.cs +++ b/src/Anichron.Tests.Integration/Work/WorkItemStoreTests.cs @@ -91,6 +91,56 @@ public async Task HeartbeatAsync_HoldingAttempt_ExtendsTheLeaseAndStoresTheProgr item.UpdatedAt.Should().Be(At(30)); } + [Fact] + public async Task HeartbeatAsync_SameProgress_ExtendsTheLeaseWithoutMovingUpdatedAtAsync() + { + var work = await WorkDatabase.CreateAsync(server, Ct); + await work.EnqueueAsync(new UnkeyedPayload(1), Ct); + var claimed = (await work.Store.ClaimAsync("worker", null, At(0), Lease, Ct))!; + await work.Store.HeartbeatAsync(claimed.Attempt, Halfway, At(30), Lease, Ct); + + var held = await work.Store.HeartbeatAsync(claimed.Attempt, new WorkProgress(5, 10, "halfway"), At(32), Lease, Ct); + + held.Should().BeTrue(); + var item = await work.ItemAsync(claimed.Attempt.Id, Ct); + item.LeaseUntil.Should().Be(At(92)); + item.UpdatedAt.Should().Be(At(30)); + } + + [Fact] + public async Task HeartbeatAsync_NoProgressSinceTheClaim_LeavesUpdatedAtAtTheClaimAsync() + { + var work = await WorkDatabase.CreateAsync(server, Ct); + await work.EnqueueAsync(new UnkeyedPayload(1), Ct); + var claimed = (await work.Store.ClaimAsync("worker", null, At(0), Lease, Ct))!; + + await work.Store.HeartbeatAsync(claimed.Attempt, WorkProgress.None, At(2), Lease, Ct); + + var item = await work.ItemAsync(claimed.Attempt.Id, Ct); + item.LeaseUntil.Should().Be(At(62)); + item.UpdatedAt.Should().Be(At(0)); + } + + [Theory] + [InlineData(6L, 10L, "halfway")] + [InlineData(5L, null, "halfway")] + [InlineData(5L, 11L, "halfway")] + [InlineData(5L, 10L, null)] + [InlineData(5L, 10L, "later")] + public async Task HeartbeatAsync_AnyPartOfTheProgressChanged_MovesUpdatedAtAsync(long done, long? total, string? note) + { + var work = await WorkDatabase.CreateAsync(server, Ct); + await work.EnqueueAsync(new UnkeyedPayload(1), Ct); + var claimed = (await work.Store.ClaimAsync("worker", null, At(0), Lease, Ct))!; + await work.Store.HeartbeatAsync(claimed.Attempt, Halfway, At(30), Lease, Ct); + + await work.Store.HeartbeatAsync(claimed.Attempt, new WorkProgress(done, total, note), At(32), Lease, Ct); + + var item = await work.ItemAsync(claimed.Attempt.Id, Ct); + (item.ProgressDone, item.ProgressTotal, item.ProgressNote).Should().Be((done, total, note)); + item.UpdatedAt.Should().Be(At(32)); + } + [Fact] public async Task AfterAReclaim_TheOldAttemptsHeartbeatAndCompletionChangeNothingAsync() { diff --git a/src/Anichron.Worker/Work/EfWorkItemStore.cs b/src/Anichron.Worker/Work/EfWorkItemStore.cs index a354b63..492e6cb 100644 --- a/src/Anichron.Worker/Work/EfWorkItemStore.cs +++ b/src/Anichron.Worker/Work/EfWorkItemStore.cs @@ -53,7 +53,11 @@ public async Task HeartbeatAsync( .SetProperty(w => w.ProgressDone, progress.Done) .SetProperty(w => w.ProgressTotal, progress.Total) .SetProperty(w => w.ProgressNote, progress.Note) - .SetProperty(w => w.UpdatedAt, now), ct); + .SetProperty( + w => w.UpdatedAt, + w => w.ProgressDone == progress.Done && w.ProgressTotal == progress.Total && w.ProgressNote == progress.Note + ? w.UpdatedAt + : now), ct); return rows == 1; } From a192680f210e732233e6b2e74a8e2f1a4f01935c Mon Sep 17 00:00:00 2001 From: bitbiter-dev Date: Tue, 20 Oct 2026 20:54:17 +1300 Subject: [PATCH 2/3] feat(api): Admin work-item list, detail and live event stream (#279) GET /api/v1/admin/work-items lists items newest request first, filtered by status and kind and paged, with the stream cursor read before the page. GET .../{id} returns one item or 404. GET .../events streams every changed item as Server-Sent Events, re-querying UpdatedAt at WorkItemEvents: IntervalSeconds (2 s); each event id is the cursor, so Last-Event-ID resumes from it. A 10 s lookback, deduplicated against the previous poll, keeps a change whose commit lands behind the cursor from being skipped. The API reads work items only through IWorkItemReader, as WorkItemView. A result is shown once an item succeeded; an error once it failed or was cancelled. Co-Authored-By: Claude Opus 5.5 --- .../Endpoints/WorkItemEndpointsHttpTests.cs | 121 +++++++++ .../Endpoints/WorkItemEndpointsTests.cs | 231 +++++++++++++++++ .../Endpoints/WorkItemEventsTests.cs | 188 ++++++++++++++ .../Infrastructure/ApiIniDefaultsTests.cs | 3 +- .../ApiServiceRegistrationTests.cs | 52 ++++ .../Infrastructure/AuthenticatedTestApp.cs | 21 +- .../Infrastructure/RouteTableTests.cs | 20 +- src/Anichron.API/Endpoints/WorkItemDto.cs | 24 ++ .../Endpoints/WorkItemDtoMapper.cs | 31 +++ .../Endpoints/WorkItemEndpoints.cs | 61 +++++ src/Anichron.API/Endpoints/WorkItemEvents.cs | 52 ++++ .../Infrastructure/ApiIniDefaults.cs | 1 + src/Anichron.API/Infrastructure/ApiPaths.cs | 7 + .../Infrastructure/ApplicationExtensions.cs | 1 + .../ServiceCollectionExtensions.cs | 12 + src/Anichron.API/Program.cs | 1 + .../Settings/WorkItemEventsSettings.cs | 9 + .../Work/WorkItemReaderTests.cs | 241 ++++++++++++++++++ src/Anichron.Core/Work/WorkItemReader.cs | 94 +++++++ 19 files changed, 1164 insertions(+), 6 deletions(-) create mode 100644 src/Anichron.API.Tests.Unit/Endpoints/WorkItemEndpointsHttpTests.cs create mode 100644 src/Anichron.API.Tests.Unit/Endpoints/WorkItemEndpointsTests.cs create mode 100644 src/Anichron.API.Tests.Unit/Endpoints/WorkItemEventsTests.cs create mode 100644 src/Anichron.API/Endpoints/WorkItemDto.cs create mode 100644 src/Anichron.API/Endpoints/WorkItemDtoMapper.cs create mode 100644 src/Anichron.API/Endpoints/WorkItemEndpoints.cs create mode 100644 src/Anichron.API/Endpoints/WorkItemEvents.cs create mode 100644 src/Anichron.API/Settings/WorkItemEventsSettings.cs create mode 100644 src/Anichron.Core.Tests.Unit/Work/WorkItemReaderTests.cs create mode 100644 src/Anichron.Core/Work/WorkItemReader.cs diff --git a/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEndpointsHttpTests.cs b/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEndpointsHttpTests.cs new file mode 100644 index 0000000..063d565 --- /dev/null +++ b/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEndpointsHttpTests.cs @@ -0,0 +1,121 @@ +using Anichron.API.Endpoints; +using Anichron.API.Settings; +using Anichron.API.Tests.Unit.Infrastructure; +using Anichron.Core.Domain; +using Anichron.Core.Work; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; +using System.Net; +using System.Text.Json; + +namespace Anichron.API.Tests.Unit.Endpoints; + +public sealed class WorkItemEndpointsHttpTests +{ + private static readonly Instant Start = Instant.FromUtc(2026, 10, 18, 9, 0); + private static readonly Guid AdminId = Guid.Parse("dddddddd-dddd-dddd-dddd-dddddddddddd"); + private static readonly Guid ItemId = Guid.Parse("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"); + + private readonly IWorkItemReader reader = Substitute.For(); + + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + private static WorkItemView Succeeded + => new(ItemId, WorkKind.Crawl, null, null, Start, WorkItemStatus.Succeeded, 1, 0, "all@host", + Start, Start + Duration.FromSeconds(9), Start + Duration.FromSeconds(9), 8, 8, "8 found", + /*lang=json,strict*/ """{"found": 8}""", null); + + private Task StartAsync() + => AuthenticatedTestApp.StartAsync(api => api.MapWorkItemEndpoints(), services => + { + services.AddSingleton(reader); + services.AddSingleton(new ImmediateTimers()); + services.AddSingleton(Options.Create(new WorkItemEventsSettings())); + services.AddScoped(); + }); + + private sealed class ImmediateTimers : TimeProvider + { + public override ITimer CreateTimer(TimerCallback callback, object? state, TimeSpan dueTime, TimeSpan period) + => base.CreateTimer(callback, state, TimeSpan.Zero, Timeout.InfiniteTimeSpan); + } + + [Fact] + public async Task List_BindsTheFiltersAndPagingFromTheQueryAsync() + { + reader.ListAsync(Arg.Any(), Arg.Any()).Returns(new WorkItemPage([Succeeded], 11)); + await using var app = await StartAsync(); + using var client = app.Client; + + using var response = await client.SendAsync( + AuthenticatedTestApp.GetAsAdmin("/api/v1/admin/work-items?status=Succeeded&kind=Crawl&page=2&pageSize=10", AdminId), Ct); + + response.StatusCode.Should().Be(HttpStatusCode.OK); + await reader.Received(1).ListAsync( + new WorkItemQuery(WorkItemStatus.Succeeded, WorkKind.Crawl, Skip: 10, Take: 10), Arg.Any()); + using var body = JsonDocument.Parse(await response.Content.ReadAsStringAsync(Ct)); + body.RootElement.GetProperty("total").GetInt32().Should().Be(11); + var item = body.RootElement.GetProperty("items")[0]; + item.GetProperty("status").GetString().Should().Be("Succeeded"); + item.GetProperty("result").GetProperty("found").GetInt32().Should().Be(8); + } + + [Fact] + public async Task List_UnknownStatus_IsABadRequestAsync() + { + await using var app = await StartAsync(); + using var client = app.Client; + + using var response = await client.SendAsync( + AuthenticatedTestApp.GetAsAdmin("/api/v1/admin/work-items?status=Sleeping", AdminId), Ct); + + response.StatusCode.Should().Be(HttpStatusCode.BadRequest); + } + + [Theory] + [InlineData("/api/v1/admin/work-items")] + [InlineData("/api/v1/admin/work-items/aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa")] + [InlineData("/api/v1/admin/work-items/events")] + public async Task EveryRoute_ForANonAdmin_IsForbiddenAsync(string path) + { + await using var app = await StartAsync(); + using var client = app.Client; + + using var response = await client.SendAsync(AuthenticatedTestApp.Get(path, AdminId), Ct); + + response.StatusCode.Should().Be(HttpStatusCode.Forbidden); + } + + [Fact] + public async Task Events_ResumeFromLastEventIdAndFrameEachItemAsync() + { + var polls = 0; + reader.UpdatedSinceAsync(Arg.Any(), Arg.Any()).Returns(_ => ++polls == 1 + ? Task.FromResult>([Succeeded]) + : throw new InvalidOperationException("The stream is read once.")); + await using var app = await StartAsync(); + using var client = app.Client; + using var request = AuthenticatedTestApp.GetAsAdmin("/api/v1/admin/work-items/events", AdminId); + request.Headers.Add("Last-Event-ID", "2026-10-18T09:00:05Z"); + + using var response = await client.SendAsync(request, HttpCompletionOption.ResponseHeadersRead, Ct); + + response.Content.Headers.ContentType!.MediaType.Should().Be("text/event-stream"); + await reader.Received(1).UpdatedSinceAsync(Start - Duration.FromSeconds(5), Arg.Any()); + var frame = await FirstFrameAsync(response); + frame.Should().HaveCount(3).And.Contain("event: work-item").And.Contain("id: 2026-10-18T09:00:09Z"); + var json = frame.Should().ContainSingle(line => line.StartsWith("data: ", StringComparison.Ordinal)).Which; + using var data = JsonDocument.Parse(json["data: ".Length..]); + data.RootElement.GetProperty("id").GetGuid().Should().Be(ItemId); + data.RootElement.GetProperty("progress").GetProperty("note").GetString().Should().Be("8 found"); + } + + private static async Task> FirstFrameAsync(HttpResponseMessage response) + { + using var stream = new StreamReader(await response.Content.ReadAsStreamAsync(Ct)); + var lines = new List(); + while (await stream.ReadLineAsync(Ct) is { Length: > 0 } line) + lines.Add(line); + return lines; + } +} diff --git a/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEndpointsTests.cs b/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEndpointsTests.cs new file mode 100644 index 0000000..6996509 --- /dev/null +++ b/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEndpointsTests.cs @@ -0,0 +1,231 @@ +using Anichron.API.Endpoints; +using Anichron.API.Settings; +using Anichron.Core.Domain; +using Anichron.Core.Work; +using Microsoft.AspNetCore.Http.HttpResults; +using Microsoft.Extensions.Options; + +namespace Anichron.API.Tests.Unit.Endpoints; + +public sealed class WorkItemEndpointsTests +{ + private static readonly Instant Start = Instant.FromUtc(2026, 10, 18, 9, 0); + private static readonly Guid ItemId = Guid.Parse("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"); + private static readonly Guid ConfigId = Guid.Parse("cccccccc-cccc-cccc-cccc-cccccccccccc"); + private static readonly Guid AdminId = Guid.Parse("dddddddd-dddd-dddd-dddd-dddddddddddd"); + + private readonly IWorkItemReader reader = Substitute.For(); + + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + private static Instant At(int seconds) => Start + Duration.FromSeconds(seconds); + + private static WorkItemView View( + WorkItemStatus status, + string? result = /*lang=json,strict*/ """{"found": 8}""", + string? error = "IOException: the root is gone") + => new(ItemId, WorkKind.Crawl, ConfigId, AdminId, At(1), status, 2, 1, "all@host", + At(2), At(9), At(9), 7, 8, "7 found", result, error); + + private void Pages(int total = 1, Instant? latest = null) + { + reader.LatestUpdateAsync(Arg.Any()).Returns(latest ?? At(30)); + reader.ListAsync(Arg.Any(), Arg.Any()) + .Returns(new WorkItemPage([View(WorkItemStatus.Running)], total)); + } + + // ========================================================================== + // ListAsync + // ========================================================================== + + [Fact] + public async Task ListAsync_NoPaging_ReadsTheFirstPageOfFiftyAsync() + { + Pages(total: 137); + + var result = await WorkItemEndpoints.ListAsync(WorkItemStatus.Running, WorkKind.Crawl, null, null, reader, Ct); + + var page = result.Result.Should().BeOfType>().Which.Value!; + (page.Page, page.PageSize, page.Total).Should().Be((1, 50, 137)); + page.Items.Should().ContainSingle().Which.Id.Should().Be(ItemId); + await reader.Received(1).ListAsync( + new WorkItemQuery(WorkItemStatus.Running, WorkKind.Crawl, Skip: 0, Take: 50), Arg.Any()); + } + + [Fact] + public async Task ListAsync_PageAndSize_SkipThePagesBeforeAsync() + { + Pages(); + + var result = await WorkItemEndpoints.ListAsync(null, null, 3, 20, reader, Ct); + + var page = result.Result.Should().BeOfType>().Which.Value!; + (page.Page, page.PageSize).Should().Be((3, 20)); + await reader.Received(1).ListAsync(new WorkItemQuery(null, null, Skip: 40, Take: 20), Arg.Any()); + } + + [Fact] + public async Task ListAsync_CarriesTheCursorReadBeforeThePageAsync() + { + Pages(latest: At(30)); + + var result = await WorkItemEndpoints.ListAsync(null, null, null, null, reader, Ct); + + result.Result.Should().BeOfType>().Which.Value!.Cursor.Should().Be("2026-10-18T09:00:30Z"); + Received.InOrder(() => + { + _ = reader.LatestUpdateAsync(Arg.Any()); + _ = reader.ListAsync(Arg.Any(), Arg.Any()); + }); + } + + [Fact] + public async Task ListAsync_NoItemsYet_CursorIsTheBeginningAsync() + { + reader.LatestUpdateAsync(Arg.Any()).Returns((Instant?)null); + reader.ListAsync(Arg.Any(), Arg.Any()).Returns(new WorkItemPage([], 0)); + + var result = await WorkItemEndpoints.ListAsync(null, null, null, null, reader, Ct); + + result.Result.Should().BeOfType>().Which.Value!.Cursor.Should().Be("1970-01-01T00:00:00Z"); + } + + [Theory] + [InlineData(1, 1, 0)] + [InlineData(1, 200, 0)] + [InlineData(10_737_418, 200, 2_147_483_400)] + [InlineData(2_147_483_647, 1, 2_147_483_646)] + public async Task ListAsync_PagingInRange_IsReadAsync(int page, int pageSize, int skip) + { + Pages(); + + var result = await WorkItemEndpoints.ListAsync(null, null, page, pageSize, reader, Ct); + + result.Result.Should().BeOfType>(); + await reader.Received(1).ListAsync(new WorkItemQuery(null, null, skip, pageSize), Arg.Any()); + } + + [Theory] + [InlineData(0, 50, "page")] + [InlineData(-1, 50, "page")] + [InlineData(10_737_419, 200, "page")] + [InlineData(1, 0, "pageSize")] + [InlineData(1, 201, "pageSize")] + public async Task ListAsync_PagingOutOfRange_IsAValidationProblemAsync(int page, int pageSize, string field) + { + Pages(); + + var result = await WorkItemEndpoints.ListAsync(null, null, page, pageSize, reader, Ct); + + result.Result.Should().BeOfType().Which.ProblemDetails.Errors.Keys.Should().Equal(field); + await reader.DidNotReceive().ListAsync(Arg.Any(), Arg.Any()); + } + + // ========================================================================== + // GetAsync + // ========================================================================== + + [Fact] + public async Task GetAsync_KnownId_ReturnsTheItemAsync() + { + reader.FindAsync(ItemId, Arg.Any()).Returns(View(WorkItemStatus.Running)); + + var result = await WorkItemEndpoints.GetAsync(ItemId, reader, Ct); + + result.Result.Should().BeOfType>().Which.Value!.Id.Should().Be(ItemId); + } + + [Fact] + public async Task GetAsync_UnknownId_IsNotFoundAsync() + { + reader.FindAsync(ItemId, Arg.Any()).Returns((WorkItemView?)null); + + var result = await WorkItemEndpoints.GetAsync(ItemId, reader, Ct); + + result.Result.Should().BeOfType(); + } + + // ========================================================================== + // StreamEvents + // ========================================================================== + + [Fact] + public void StreamEvents_AnswersWithAServerSentEventStream() + { + var events = new WorkItemEvents(reader, TimeProvider.System, Options.Create(new WorkItemEventsSettings())); + + WorkItemEndpoints.StreamEvents(null, events, Ct).Should().BeOfType>(); + } + + // ========================================================================== + // WorkItemDtoMapper + // ========================================================================== + + [Fact] + public void ToDto_ShowsTheItemWithIsoTimesAndNamedEnums() + { + var dto = WorkItemDtoMapper.ToDto(View(WorkItemStatus.Running) with { FinishedAt = null }); + + dto.Should().BeEquivalentTo(new + { + Id = ItemId, + Kind = "Crawl", + StorageConfigId = ConfigId, + RequestedBy = AdminId, + Status = "Running", + Attempt = 2, + Stalls = 1, + ClaimedBy = "all@host", + RequestedAt = "2026-10-18T09:00:01Z", + StartedAt = "2026-10-18T09:00:02Z", + FinishedAt = (string?)null, + UpdatedAt = "2026-10-18T09:00:09Z", + Progress = new WorkProgressDto(7, 8, "7 found"), + }); + } + + [Fact] + public void ToDto_NotStarted_HasNoStartTime() + => WorkItemDtoMapper.ToDto(View(WorkItemStatus.Queued) with { StartedAt = null }).StartedAt.Should().BeNull(); + + [Fact] + public void ToDto_Finished_HasItsFinishTime() + => WorkItemDtoMapper.ToDto(View(WorkItemStatus.Succeeded)).FinishedAt.Should().Be("2026-10-18T09:00:09Z"); + + [Fact] + public void ToDto_Succeeded_ShowsTheResultAsJsonAndNoError() + { + var dto = WorkItemDtoMapper.ToDto(View(WorkItemStatus.Succeeded)); + + dto.Result.Should().NotBeNull(); + dto.Result.Value.GetProperty("found").GetInt32().Should().Be(8); + dto.Error.Should().BeNull(); + } + + [Fact] + public void ToDto_SucceededWithoutAResult_ShowsNoResult() + => WorkItemDtoMapper.ToDto(View(WorkItemStatus.Succeeded, result: null)).Result.Should().BeNull(); + + [Theory] + [InlineData(WorkItemStatus.Failed)] + [InlineData(WorkItemStatus.Cancelled)] + public void ToDto_FailedOrCancelled_ShowsTheErrorAndProgressButNoResult(WorkItemStatus status) + { + var dto = WorkItemDtoMapper.ToDto(View(status)); + + dto.Error.Should().Be("IOException: the root is gone"); + dto.Progress.Should().Be(new WorkProgressDto(7, 8, "7 found")); + dto.Result.Should().BeNull(); + } + + [Theory] + [InlineData(WorkItemStatus.Queued)] + [InlineData(WorkItemStatus.Running)] + public void ToDto_NotFinished_ShowsNeitherResultNorError(WorkItemStatus status) + { + var dto = WorkItemDtoMapper.ToDto(View(status)); + + dto.Result.Should().BeNull(); + dto.Error.Should().BeNull(); + } +} diff --git a/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEventsTests.cs b/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEventsTests.cs new file mode 100644 index 0000000..e3a2402 --- /dev/null +++ b/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEventsTests.cs @@ -0,0 +1,188 @@ +using Anichron.API.Endpoints; +using Anichron.API.Settings; +using Anichron.Core.Domain; +using Anichron.Core.Work; +using Anichron.Testing.Time; +using Microsoft.Extensions.Options; +using NodaTime.Text; + +namespace Anichron.API.Tests.Unit.Endpoints; + +public sealed class WorkItemEventsTests : IAsyncDisposable +{ + private static readonly Instant Start = Instant.FromUtc(2026, 10, 18, 9, 0); + private static readonly Guid A = Guid.Parse("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"); + private static readonly Guid B = Guid.Parse("bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb"); + + private readonly Timeline timeline = new(); + private readonly IWorkItemReader reader = Substitute.For(); + private readonly Queue> polls = new(); + private Instant? latest = At(100); + + public WorkItemEventsTests() + { + reader.LatestUpdateAsync(Arg.Any()).Returns(_ => + { + timeline.Record("latest"); + return latest; + }); + reader.UpdatedSinceAsync(Arg.Any(), Arg.Any()).Returns(call => + { + timeline.Record($"since {Iso(call.Arg())}"); + return polls.TryDequeue(out var poll) ? poll : []; + }); + } + + public ValueTask DisposeAsync() => timeline.DisposeAsync(); + + private static Instant At(int seconds) => Start + Duration.FromSeconds(seconds); + + private static string Iso(Instant instant) => InstantPattern.ExtendedIso.Format(instant); + + private static WorkItemView Changed(Guid id, int updatedAt) + => new(id, WorkKind.Crawl, null, null, Start, WorkItemStatus.Running, 1, 0, "all@host", + Start, null, At(updatedAt), 3, null, "3 found", null, null); + + private void Stream(string? lastEventId = null, double intervalSeconds = 2) + { + var events = new WorkItemEvents( + reader, timeline.Clock, Options.Create(new WorkItemEventsSettings { IntervalSeconds = intervalSeconds })); + timeline.Watch(Task.Run(async () => + { + await foreach (var item in events.StreamAsync(lastEventId, timeline.Lifetime)) + timeline.Record($"{item.EventType} {item.Data.Id:N} {item.Data.UpdatedAt} id {item.EventId}"); + }, CancellationToken.None)); + } + + private static string Event(Guid id, int updatedAt, int cursor) + => $"work-item {id:N} {Iso(At(updatedAt))} id {Iso(At(cursor))}"; + + [Fact] + public Task StreamAsync_NoLastEventId_StartsFromTheLatestChangeLessTheLookbackAsync() + { + Stream(); + + return timeline.ExpectAsync(new Called("latest"), new Called($"since {Iso(At(90))}"), new TimerArmed(TimeSpan.FromSeconds(2))); + } + + [Fact] + public Task StreamAsync_NoItemsYet_StartsFromTheBeginningAsync() + { + latest = null; + + Stream(); + + return timeline.ExpectAsync(new Called("latest"), new Called("since 1969-12-31T23:59:50Z")); + } + + [Fact] + public Task StreamAsync_LastEventId_ResumesFromItWithoutAskingForTheLatestAsync() + { + Stream(lastEventId: Iso(At(40))); + + return timeline.ExpectAsync(new Called($"since {Iso(At(30))}")); + } + + [Theory] + [InlineData("")] + [InlineData("not-a-time")] + [InlineData("2026-10-18")] + public Task StreamAsync_UnreadableLastEventId_StartsFromTheLatestChangeAsync(string lastEventId) + { + Stream(lastEventId); + + return timeline.ExpectAsync(new Called("latest"), new Called($"since {Iso(At(90))}")); + } + + [Fact] + public Task StreamAsync_EmitsEachChangedItemWithTheCursorAsItsIdAsync() + { + polls.Enqueue([Changed(A, 101), Changed(B, 104)]); + + Stream(); + + return timeline.ExpectAsync( + new Called("latest"), + new Called($"since {Iso(At(90))}"), + new Called(Event(A, 101, cursor: 101)), + new Called(Event(B, 104, cursor: 104)), + new TimerArmed(TimeSpan.FromSeconds(2))); + } + + [Fact] + public async Task StreamAsync_AfterTheInterval_ReQueriesFromTheAdvancedCursorAsync() + { + polls.Enqueue([Changed(A, 105)]); + Stream(intervalSeconds: 7.5); + await timeline.ExpectAsync( + new Called("latest"), new Called($"since {Iso(At(90))}"), new Called(Event(A, 105, cursor: 105)), + new TimerArmed(TimeSpan.FromSeconds(7.5))); + + timeline.Clock.Advance(TimeSpan.FromSeconds(7.5)); + + await timeline.ExpectAsync( + new TimerFired(TimeSpan.FromSeconds(7.5)), new Called($"since {Iso(At(95))}"), new TimerArmed(TimeSpan.FromSeconds(7.5))); + } + + [Fact] + public async Task StreamAsync_ItemAlreadySentUnchanged_IsNotSentAgainAsync() + { + polls.Enqueue([Changed(A, 105)]); + polls.Enqueue([Changed(A, 105), Changed(B, 106)]); + Stream(); + await timeline.ExpectAsync( + new Called("latest"), new Called($"since {Iso(At(90))}"), new Called(Event(A, 105, cursor: 105)), + new TimerArmed(TimeSpan.FromSeconds(2))); + + timeline.Clock.Advance(TimeSpan.FromSeconds(2)); + + await timeline.ExpectAsync( + new TimerFired(TimeSpan.FromSeconds(2)), new Called($"since {Iso(At(95))}"), + new Called(Event(B, 106, cursor: 106)), new TimerArmed(TimeSpan.FromSeconds(2))); + } + + [Fact] + public async Task StreamAsync_ItemChangedAgain_IsSentAgainAsync() + { + polls.Enqueue([Changed(A, 105)]); + polls.Enqueue([Changed(A, 107)]); + Stream(); + await timeline.ExpectAsync( + new Called("latest"), new Called($"since {Iso(At(90))}"), new Called(Event(A, 105, cursor: 105)), + new TimerArmed(TimeSpan.FromSeconds(2))); + + timeline.Clock.Advance(TimeSpan.FromSeconds(2)); + + await timeline.ExpectAsync( + new TimerFired(TimeSpan.FromSeconds(2)), new Called($"since {Iso(At(95))}"), + new Called(Event(A, 107, cursor: 107))); + } + + [Fact] + public async Task StreamAsync_ChangeCommittedBehindTheCursor_IsStillSentAndKeepsTheCursorAsync() + { + polls.Enqueue([Changed(A, 105)]); + polls.Enqueue([Changed(B, 103)]); + Stream(); + await timeline.ExpectAsync( + new Called("latest"), new Called($"since {Iso(At(90))}"), new Called(Event(A, 105, cursor: 105)), + new TimerArmed(TimeSpan.FromSeconds(2))); + + timeline.Clock.Advance(TimeSpan.FromSeconds(2)); + + await timeline.ExpectAsync( + new TimerFired(TimeSpan.FromSeconds(2)), new Called($"since {Iso(At(95))}"), + new Called(Event(B, 103, cursor: 105))); + } + + [Fact] + public Task StreamAsync_ItemsBehindTheStartingCursor_KeepTheCursorAsync() + { + polls.Enqueue([Changed(A, 95)]); + + Stream(); + + return timeline.ExpectAsync( + new Called("latest"), new Called($"since {Iso(At(90))}"), new Called(Event(A, 95, cursor: 100))); + } +} diff --git a/src/Anichron.API.Tests.Unit/Infrastructure/ApiIniDefaultsTests.cs b/src/Anichron.API.Tests.Unit/Infrastructure/ApiIniDefaultsTests.cs index 276eef9..7a02433 100644 --- a/src/Anichron.API.Tests.Unit/Infrastructure/ApiIniDefaultsTests.cs +++ b/src/Anichron.API.Tests.Unit/Infrastructure/ApiIniDefaultsTests.cs @@ -16,7 +16,8 @@ public void Entries_ExceptGeneratedSecret_HaveTheDocumentedDefaults() ("Jwt", "Audience", "anichron-client"), ("Cors", "AllowedOrigins", ""), ("ForwardedHeaders", "KnownProxies", ""), - ("ForwardedHeaders", "KnownNetworks", "")); + ("ForwardedHeaders", "KnownNetworks", ""), + ("WorkItemEvents", "IntervalSeconds", "2")); } [Fact] diff --git a/src/Anichron.API.Tests.Unit/Infrastructure/ApiServiceRegistrationTests.cs b/src/Anichron.API.Tests.Unit/Infrastructure/ApiServiceRegistrationTests.cs index 007757c..ac2a541 100644 --- a/src/Anichron.API.Tests.Unit/Infrastructure/ApiServiceRegistrationTests.cs +++ b/src/Anichron.API.Tests.Unit/Infrastructure/ApiServiceRegistrationTests.cs @@ -5,6 +5,7 @@ using Anichron.API.Settings; using Anichron.Core.Data; using Anichron.Core.Data.Repository; +using Anichron.Core.Work; using Anichron.Infrastructure.Storage; using Microsoft.AspNetCore.Authentication; using Microsoft.AspNetCore.Authentication.JwtBearer; @@ -104,6 +105,57 @@ public void AddAuthServices_UnitOfWorkIsTheScopedDbContext() => AuthServices().Should().ContainSingle(d => d.ServiceType == typeof(IUnitOfWork)) .Which.Lifetime.Should().Be(ServiceLifetime.Scoped); + // ========================================================================== + // Work items + // ========================================================================== + + private static ServiceCollection WorkItemServices(Dictionary? values = null) + { + var services = new ServiceCollection(); + services.AddSingleton(new ConfigurationBuilder().AddInMemoryCollection(values ?? []).Build()); + services.AddWorkItemServices(); + return services; + } + + [Theory] + [InlineData(typeof(IWorkItemReader), typeof(EfWorkItemReader))] + [InlineData(typeof(WorkItemEvents), typeof(WorkItemEvents))] + public void AddWorkItemServices_RegistersTheScopedImplementation(Type service, Type implementation) + => WorkItemServices().Should().ContainSingle(d => d.ServiceType == service) + .Which.Should().Match(d => d.ImplementationType == implementation && d.Lifetime == ServiceLifetime.Scoped); + + [Fact] + public void AddWorkItemServices_EventIntervalDefaultsToTwoSeconds() + { + using var provider = WorkItemServices().BuildServiceProvider(); + + provider.GetRequiredService>().Value.IntervalSeconds.Should().Be(2); + } + + [Fact] + public void AddWorkItemServices_BindsTheEventIntervalFromConfiguration() + { + using var provider = WorkItemServices(new() { ["WorkItemEvents:IntervalSeconds"] = "0.5" }).BuildServiceProvider(); + + provider.GetRequiredService>().Value.IntervalSeconds.Should().Be(0.5); + } + + [Theory] + [InlineData("0")] + [InlineData("-1")] + public void AddWorkItemServices_RejectsAnEventIntervalThatIsNotPositive(string seconds) + { + using var provider = WorkItemServices(new() { ["WorkItemEvents:IntervalSeconds"] = seconds }).BuildServiceProvider(); + + var read = () => provider.GetRequiredService>().Value; + + read.Should().Throw(); + } + + [Fact] + public void AddWorkItemServices_ValidatesTheEventIntervalAtStartup() + => WorkItemServices().Should().Contain(d => d.ServiceType == typeof(IStartupValidator)); + // ========================================================================== // Options binding // ========================================================================== diff --git a/src/Anichron.API.Tests.Unit/Infrastructure/AuthenticatedTestApp.cs b/src/Anichron.API.Tests.Unit/Infrastructure/AuthenticatedTestApp.cs index bc5ee76..80cbbb3 100644 --- a/src/Anichron.API.Tests.Unit/Infrastructure/AuthenticatedTestApp.cs +++ b/src/Anichron.API.Tests.Unit/Infrastructure/AuthenticatedTestApp.cs @@ -1,4 +1,5 @@ using Anichron.API.Infrastructure; +using Anichron.API.Security; using Microsoft.AspNetCore.Authentication; using Microsoft.AspNetCore.Builder; using Microsoft.AspNetCore.Routing; @@ -13,12 +14,14 @@ namespace Anichron.API.Tests.Unit.Infrastructure; /// /// Real API endpoints on a test server, under the API base path, where a request is authenticated -/// as the user named in . A test maps only the endpoints it exercises, so -/// the wiring of the others is not in its coverage. +/// as the user named in , as an admin when it also carries +/// . A test maps only the endpoints it exercises, so the wiring of the +/// others is not in its coverage. /// internal sealed class AuthenticatedTestApp : IAsyncDisposable { public const string UserHeader = "X-Test-User"; + public const string AdminHeader = "X-Test-Admin"; private readonly WebApplication app; @@ -32,7 +35,7 @@ public static async Task StartAsync( var builder = ApiTestHost.CreateBuilder(); builder.WebHost.UseTestServer(); builder.Services.AddAuthentication("Test").AddScheme("Test", null); - builder.Services.AddAuthorization(); + builder.Services.AddAuthorizationPolicies(); configureServices(builder.Services); var app = builder.Build(); app.UseAuthentication(); @@ -49,6 +52,13 @@ public static HttpRequestMessage Get(string path, Guid userId) return request; } + public static HttpRequestMessage GetAsAdmin(string path, Guid userId) + { + var request = Get(path, userId); + request.Headers.Add(AdminHeader, "true"); + return request; + } + public ValueTask DisposeAsync() => app.DisposeAsync(); private sealed class HeaderAuthHandler( @@ -60,7 +70,10 @@ protected override Task HandleAuthenticateAsync() if (!Request.Headers.TryGetValue(UserHeader, out var userId)) return Task.FromResult(AuthenticateResult.NoResult()); - var identity = new ClaimsIdentity([new Claim(ClaimTypes.NameIdentifier, userId.ToString())], Scheme.Name); + List claims = [new(ClaimTypes.NameIdentifier, userId.ToString())]; + if (Request.Headers.ContainsKey(AdminHeader)) + claims.Add(new Claim(AppClaimTypes.IsAdmin, "true")); + var identity = new ClaimsIdentity(claims, Scheme.Name); return Task.FromResult(AuthenticateResult.Success(new AuthenticationTicket(new ClaimsPrincipal(identity), Scheme.Name))); } } diff --git a/src/Anichron.API.Tests.Unit/Infrastructure/RouteTableTests.cs b/src/Anichron.API.Tests.Unit/Infrastructure/RouteTableTests.cs index 2e0af9c..b474679 100644 --- a/src/Anichron.API.Tests.Unit/Infrastructure/RouteTableTests.cs +++ b/src/Anichron.API.Tests.Unit/Infrastructure/RouteTableTests.cs @@ -139,11 +139,29 @@ public void FlashbackRoutes_AreTaggedFlashbacks() => Routes.Where(r => r.Pattern.StartsWith("/api/v1/flashbacks", StringComparison.Ordinal)) .Should().NotBeEmpty().And.OnlyContain(r => r.Tags.SequenceEqual(new[] { "Flashbacks" })); + [Theory] + [InlineData("/api/v1/admin/work-items")] + [InlineData("/api/v1/admin/work-items/{workItemId:guid}")] + [InlineData("/api/v1/admin/work-items/events")] + public void WorkItemRoutes_RequireTheAdminPolicyWithoutRateLimit(string pattern) + { + var route = Find("GET", pattern); + + route.Anonymous.Should().BeFalse(); + route.Policies.Should().Contain(AuthPolicies.Admin); + route.RateLimit.Should().BeNull(); + } + + [Fact] + public void WorkItemRoutes_AreTaggedWorkItems() + => Routes.Where(r => r.Pattern.StartsWith("/api/v1/admin/work-items", StringComparison.Ordinal)) + .Should().HaveCount(3).And.OnlyContain(r => r.Tags.SequenceEqual(new[] { "Work items" })); + [Fact] public void Healthz_IsAnonymous() => Find("GET", "/api/v1/healthz").Anonymous.Should().BeTrue(); [Fact] public void EveryMappedRoute_IsCoveredByTheTestsAbove() - => Routes.Should().HaveCount(22); + => Routes.Should().HaveCount(25); } diff --git a/src/Anichron.API/Endpoints/WorkItemDto.cs b/src/Anichron.API/Endpoints/WorkItemDto.cs new file mode 100644 index 0000000..cfc5885 --- /dev/null +++ b/src/Anichron.API/Endpoints/WorkItemDto.cs @@ -0,0 +1,24 @@ +using System.Text.Json; + +namespace Anichron.API.Endpoints; + +public sealed record WorkItemDto( + Guid Id, + string Kind, + Guid? StorageConfigId, + Guid? RequestedBy, + string Status, + int Attempt, + int Stalls, + string? ClaimedBy, + string RequestedAt, + string? StartedAt, + string? FinishedAt, + string UpdatedAt, + WorkProgressDto Progress, + JsonElement? Result, + string? Error); + +public sealed record WorkProgressDto(long Done, long? Total, string? Note); + +public sealed record WorkItemPageDto(IReadOnlyList Items, int Page, int PageSize, int Total, string Cursor); diff --git a/src/Anichron.API/Endpoints/WorkItemDtoMapper.cs b/src/Anichron.API/Endpoints/WorkItemDtoMapper.cs new file mode 100644 index 0000000..c374a84 --- /dev/null +++ b/src/Anichron.API/Endpoints/WorkItemDtoMapper.cs @@ -0,0 +1,31 @@ +using Anichron.Core.Domain; +using Anichron.Core.Work; +using NodaTime.Text; +using System.Text.Json; + +namespace Anichron.API.Endpoints; + +internal static class WorkItemDtoMapper +{ + internal static WorkItemDto ToDto(WorkItemView view) + => new( + view.Id, + view.Kind.ToString(), + view.StorageConfigId, + view.RequestedBy, + view.Status.ToString(), + view.Attempt, + view.Stalls, + view.ClaimedBy, + Format(view.RequestedAt), + view.StartedAt is { } started ? Format(started) : null, + view.FinishedAt is { } finished ? Format(finished) : null, + Format(view.UpdatedAt), + new WorkProgressDto(view.ProgressDone, view.ProgressTotal, view.ProgressNote), + view is { Status: WorkItemStatus.Succeeded, Result: { } result } + ? JsonSerializer.Deserialize(result) + : null, + view.Status is WorkItemStatus.Failed or WorkItemStatus.Cancelled ? view.Error : null); + + internal static string Format(Instant instant) => InstantPattern.ExtendedIso.Format(instant); +} diff --git a/src/Anichron.API/Endpoints/WorkItemEndpoints.cs b/src/Anichron.API/Endpoints/WorkItemEndpoints.cs new file mode 100644 index 0000000..95a33aa --- /dev/null +++ b/src/Anichron.API/Endpoints/WorkItemEndpoints.cs @@ -0,0 +1,61 @@ +using Anichron.API.Infrastructure; +using Anichron.Core.Domain; +using Anichron.Core.Work; +using Microsoft.AspNetCore.Http.HttpResults; +using Microsoft.AspNetCore.Mvc; + +namespace Anichron.API.Endpoints; + +public static class WorkItemEndpoints +{ + internal const int DefaultPageSize = 50; + internal const int MaxPageSize = 200; + + public static IEndpointRouteBuilder MapWorkItemEndpoints(this IEndpointRouteBuilder app) + { + var group = app.MapGroup(ApiPaths.WorkItems.Group).WithTags("Work items") + .RequireAuthorization(AuthPolicies.Admin); + group.MapGet(string.Empty, ListAsync); + group.MapGet(ApiPaths.WorkItems.Events, StreamEvents); + group.MapGet(ApiPaths.WorkItems.ById, GetAsync); + return app; + } + + internal static async Task, ValidationProblem>> ListAsync( + WorkItemStatus? status, + WorkKind? kind, + int? page, + int? pageSize, + [FromServices] IWorkItemReader reader, + CancellationToken ct) + { + var number = page ?? 1; + var size = pageSize ?? DefaultPageSize; + if (size is < 1 or > MaxPageSize) + return Invalid(nameof(pageSize), $"pageSize must be from 1 to {MaxPageSize}."); + if (number < 1 || number > int.MaxValue / size) + return Invalid(nameof(page), "page is out of range."); + + var cursor = await reader.LatestUpdateAsync(ct) ?? NodaConstants.UnixEpoch; + var found = await reader.ListAsync(new WorkItemQuery(status, kind, (number - 1) * size, size), ct); + return TypedResults.Ok(new WorkItemPageDto( + [.. found.Items.Select(WorkItemDtoMapper.ToDto)], number, size, found.Total, WorkItemEvents.Format(cursor))); + } + + internal static async Task, NotFound>> GetAsync( + Guid workItemId, + [FromServices] IWorkItemReader reader, + CancellationToken ct) + => await reader.FindAsync(workItemId, ct) is { } item + ? TypedResults.Ok(WorkItemDtoMapper.ToDto(item)) + : TypedResults.NotFound(); + + internal static ServerSentEventsResult StreamEvents( + [FromHeader(Name = "Last-Event-ID")] string? lastEventId, + [FromServices] WorkItemEvents events, + CancellationToken ct) + => TypedResults.ServerSentEvents(events.StreamAsync(lastEventId, ct)); + + private static ValidationProblem Invalid(string field, string message) + => TypedResults.ValidationProblem(new Dictionary { [field] = [message] }); +} diff --git a/src/Anichron.API/Endpoints/WorkItemEvents.cs b/src/Anichron.API/Endpoints/WorkItemEvents.cs new file mode 100644 index 0000000..61227b6 --- /dev/null +++ b/src/Anichron.API/Endpoints/WorkItemEvents.cs @@ -0,0 +1,52 @@ +using Anichron.API.Settings; +using Anichron.Core.Work; +using Microsoft.Extensions.Options; +using NodaTime.Text; +using System.Net.ServerSentEvents; +using System.Runtime.CompilerServices; + +namespace Anichron.API.Endpoints; + +public sealed class WorkItemEvents(IWorkItemReader reader, TimeProvider time, IOptions settings) +{ + internal const string EventType = "work-item"; + + internal static Duration Lookback => Duration.FromSeconds(10); + + /// + /// Every work item that changes from the cursor on, re-queried at the configured interval. Each + /// event's id is the cursor after it, so a reconnect that sends it as Last-Event-ID resumes there. + /// + /// The cursor to resume from; without a readable one, the latest change. + /// Ends the stream. + /// One event per changed item; an item may arrive twice, never not at all. + public async IAsyncEnumerable> StreamAsync( + string? lastEventId, [EnumeratorCancellation] CancellationToken ct) + { + var cursor = Parse(lastEventId) ?? await reader.LatestUpdateAsync(ct) ?? NodaConstants.UnixEpoch; + var interval = TimeSpan.FromSeconds(settings.Value.IntervalSeconds); + var sent = new HashSet<(Guid, Instant)>(); + while (!ct.IsCancellationRequested) + { + var changed = await reader.UpdatedSinceAsync(cursor - Lookback, ct); + foreach (var item in changed.Where(item => !sent.Contains((item.Id, item.UpdatedAt)))) + { + cursor = Instant.Max(cursor, item.UpdatedAt); + yield return new SseItem(WorkItemDtoMapper.ToDto(item), EventType) + { + EventId = Format(cursor), + }; + } + + sent = [.. changed.Select(item => (item.Id, item.UpdatedAt))]; + await Task.Delay(interval, time, ct); + } + } + + internal static string Format(Instant cursor) => InstantPattern.ExtendedIso.Format(cursor); + + private static Instant? Parse(string? lastEventId) + => lastEventId is not null && InstantPattern.ExtendedIso.Parse(lastEventId) is { Success: true } parsed + ? parsed.Value + : null; +} diff --git a/src/Anichron.API/Infrastructure/ApiIniDefaults.cs b/src/Anichron.API/Infrastructure/ApiIniDefaults.cs index ef44b53..9879772 100644 --- a/src/Anichron.API/Infrastructure/ApiIniDefaults.cs +++ b/src/Anichron.API/Infrastructure/ApiIniDefaults.cs @@ -12,5 +12,6 @@ internal static class ApiIniDefaults new IniEntry("Cors", "AllowedOrigins", () => string.Empty), new IniEntry("ForwardedHeaders", "KnownProxies", () => string.Empty), new IniEntry("ForwardedHeaders", "KnownNetworks", () => string.Empty), + new IniEntry("WorkItemEvents", "IntervalSeconds", () => "2"), ]; } diff --git a/src/Anichron.API/Infrastructure/ApiPaths.cs b/src/Anichron.API/Infrastructure/ApiPaths.cs index d6025f4..76a18f2 100644 --- a/src/Anichron.API/Infrastructure/ApiPaths.cs +++ b/src/Anichron.API/Infrastructure/ApiPaths.cs @@ -51,6 +51,13 @@ internal static string ProxyUrl(Guid assetId, string type) => $"{Group}/{assetId}/proxies/{type}"; } + internal static class WorkItems + { + internal const string Group = "admin/work-items"; + internal const string ById = "{workItemId:guid}"; + internal const string Events = "events"; + } + internal static class Flashbacks { internal const string Group = "flashbacks"; diff --git a/src/Anichron.API/Infrastructure/ApplicationExtensions.cs b/src/Anichron.API/Infrastructure/ApplicationExtensions.cs index 78d666b..6e6788d 100644 --- a/src/Anichron.API/Infrastructure/ApplicationExtensions.cs +++ b/src/Anichron.API/Infrastructure/ApplicationExtensions.cs @@ -20,6 +20,7 @@ public WebApplication MapApiEndpoints() api.MapAdminEndpoints(); api.MapMediaEndpoints(); api.MapFlashbackEndpoints(); + api.MapWorkItemEndpoints(); return app; } } diff --git a/src/Anichron.API/Infrastructure/ServiceCollectionExtensions.cs b/src/Anichron.API/Infrastructure/ServiceCollectionExtensions.cs index 92c84b1..11554f6 100644 --- a/src/Anichron.API/Infrastructure/ServiceCollectionExtensions.cs +++ b/src/Anichron.API/Infrastructure/ServiceCollectionExtensions.cs @@ -4,6 +4,7 @@ using Anichron.API.Settings; using Anichron.Core.Data; using Anichron.Core.Data.Repository; +using Anichron.Core.Work; using Anichron.Infrastructure.Configuration; using Anichron.Infrastructure.Storage; using Microsoft.AspNetCore.Authentication.JwtBearer; @@ -77,6 +78,17 @@ public IServiceCollection AddRateLimiting() }); } + public IServiceCollection AddWorkItemServices() + { + services.AddOptions() + .BindConfiguration("WorkItemEvents") + .ValidateDataAnnotations() + .ValidateOnStart(); + services.AddScoped(); + services.AddScoped(); + return services; + } + public IServiceCollection AddCorsPolicy(IConfiguration configuration) { var allowedOrigins = CorsOrigins.Read(configuration); diff --git a/src/Anichron.API/Program.cs b/src/Anichron.API/Program.cs index 1d26d13..e316758 100644 --- a/src/Anichron.API/Program.cs +++ b/src/Anichron.API/Program.cs @@ -8,6 +8,7 @@ .AddDatabase(builder.Configuration) .AddForwardedHeadersSupport(builder.Configuration) .AddAuthServices(builder.Configuration) + .AddWorkItemServices() .AddAuthorizationPolicies() .AddRateLimiting() .AddCorsPolicy(builder.Configuration) diff --git a/src/Anichron.API/Settings/WorkItemEventsSettings.cs b/src/Anichron.API/Settings/WorkItemEventsSettings.cs new file mode 100644 index 0000000..47172cb --- /dev/null +++ b/src/Anichron.API/Settings/WorkItemEventsSettings.cs @@ -0,0 +1,9 @@ +using System.ComponentModel.DataAnnotations; + +namespace Anichron.API.Settings; + +public sealed record WorkItemEventsSettings +{ + [Range(0.0, double.MaxValue, MinimumIsExclusive = true)] + public double IntervalSeconds { get; init; } = 2; +} diff --git a/src/Anichron.Core.Tests.Unit/Work/WorkItemReaderTests.cs b/src/Anichron.Core.Tests.Unit/Work/WorkItemReaderTests.cs new file mode 100644 index 0000000..82da127 --- /dev/null +++ b/src/Anichron.Core.Tests.Unit/Work/WorkItemReaderTests.cs @@ -0,0 +1,241 @@ +using Anichron.Core.Data; +using Anichron.Core.Domain; +using Anichron.Core.Work; +using Anichron.Testing.Data; +using Microsoft.EntityFrameworkCore; +using NodaTime; + +namespace Anichron.Core.Tests.Unit.Work; + +public sealed class WorkItemReaderTests +{ + private static readonly Instant Start = Instant.FromUtc(2026, 10, 18, 9, 0); + private static readonly Guid First = Guid.Parse("11111111-1111-1111-1111-111111111111"); + private static readonly Guid Second = Guid.Parse("22222222-2222-2222-2222-222222222222"); + private static readonly Guid Third = Guid.Parse("33333333-3333-3333-3333-333333333333"); + + private readonly string database = Guid.NewGuid().ToString(); + + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + private static Instant At(int seconds) => Start + Duration.FromSeconds(seconds); + + private AnichronDbContext Context() => AnichronDbContexts.InMemory(database); + + private EfWorkItemReader Reader() => new(Context()); + + private async Task SeedAsync(params WorkItem[] items) + { + await using var db = Context(); + db.WorkItems.AddRange(items); + await db.SaveChangesAsync(Ct); + } + + private static WorkItem Item( + Guid id, + int requestedAt = 0, + int? updatedAt = null, + WorkItemStatus status = WorkItemStatus.Queued, + WorkKind kind = WorkKind.TokenCleanup) + => new() + { + Id = id, + Kind = kind, + Payload = "{}", + RequestedAt = At(requestedAt), + Status = status, + UpdatedAt = At(updatedAt ?? requestedAt), + }; + + private static IEnumerable Ids(IEnumerable views) => views.Select(v => v.Id); + + // ========================================================================== + // ListAsync + // ========================================================================== + + [Fact] + public async Task ListAsync_NoFilter_ReturnsEveryItemNewestRequestFirstAsync() + { + await SeedAsync(Item(First, requestedAt: 1), Item(Second, requestedAt: 3), Item(Third, requestedAt: 2)); + + var page = await Reader().ListAsync(new WorkItemQuery(null, null, Skip: 0, Take: 10), Ct); + + Ids(page.Items).Should().Equal(Second, Third, First); + page.Total.Should().Be(3); + } + + [Fact] + public async Task ListAsync_OrdersByRequestTimeNotByLastChangeAsync() + { + await SeedAsync(Item(First, requestedAt: 1, updatedAt: 50), Item(Second, requestedAt: 2, updatedAt: 2)); + + var page = await Reader().ListAsync(new WorkItemQuery(null, null, Skip: 0, Take: 10), Ct); + + Ids(page.Items).Should().Equal(Second, First); + } + + [Fact] + public async Task ListAsync_SameRequestTime_PutsTheLargerIdFirstAsync() + { + await SeedAsync(Item(First, requestedAt: 1), Item(Second, requestedAt: 1)); + + var page = await Reader().ListAsync(new WorkItemQuery(null, null, Skip: 0, Take: 10), Ct); + + Ids(page.Items).Should().Equal(Second, First); + } + + [Fact] + public async Task ListAsync_SkipAndTake_ReturnThatPageAndTheFullTotalAsync() + { + await SeedAsync(Item(First, requestedAt: 1), Item(Second, requestedAt: 2), Item(Third, requestedAt: 3)); + + var page = await Reader().ListAsync(new WorkItemQuery(null, null, Skip: 1, Take: 1), Ct); + + Ids(page.Items).Should().Equal(Second); + page.Total.Should().Be(3); + } + + [Fact] + public async Task ListAsync_StatusFilter_ReturnsAndCountsOnlyThatStatusAsync() + { + await SeedAsync( + Item(First, requestedAt: 1, status: WorkItemStatus.Running), + Item(Second, requestedAt: 2, status: WorkItemStatus.Failed), + Item(Third, requestedAt: 3, status: WorkItemStatus.Running)); + + var page = await Reader().ListAsync(new WorkItemQuery(WorkItemStatus.Running, null, Skip: 0, Take: 10), Ct); + + Ids(page.Items).Should().Equal(Third, First); + page.Total.Should().Be(2); + } + + [Fact] + public async Task ListAsync_KindFilter_ReturnsAndCountsOnlyThatKindAsync() + { + await SeedAsync( + Item(First, requestedAt: 1, kind: WorkKind.Crawl), + Item(Second, requestedAt: 2, kind: WorkKind.TokenCleanup), + Item(Third, requestedAt: 3, kind: WorkKind.Crawl)); + + var page = await Reader().ListAsync(new WorkItemQuery(null, WorkKind.Crawl, Skip: 0, Take: 10), Ct); + + Ids(page.Items).Should().Equal(Third, First); + page.Total.Should().Be(2); + } + + [Fact] + public async Task ListAsync_StatusAndKindFilters_BothApplyAsync() + { + await SeedAsync( + Item(First, requestedAt: 1, status: WorkItemStatus.Running, kind: WorkKind.Crawl), + Item(Second, requestedAt: 2, status: WorkItemStatus.Running, kind: WorkKind.TokenCleanup), + Item(Third, requestedAt: 3, status: WorkItemStatus.Failed, kind: WorkKind.Crawl)); + + var page = await Reader().ListAsync(new WorkItemQuery(WorkItemStatus.Running, WorkKind.Crawl, Skip: 0, Take: 10), Ct); + + Ids(page.Items).Should().Equal(First); + page.Total.Should().Be(1); + } + + // ========================================================================== + // FindAsync + // ========================================================================== + + [Fact] + public async Task FindAsync_KnownId_ReturnsEveryShownFieldAsync() + { + var configId = Guid.Parse("44444444-4444-4444-4444-444444444444"); + var adminId = Guid.Parse("55555555-5555-5555-5555-555555555555"); + await SeedAsync(new WorkItem + { + Id = First, + Kind = WorkKind.Crawl, + Payload = /*lang=json,strict*/ """{"storageConfigId": "44444444-4444-4444-4444-444444444444"}""", + DedupKey = "crawl:44444444", + StorageConfigId = configId, + RequestedBy = adminId, + RequestedAt = At(1), + Status = WorkItemStatus.Succeeded, + Attempt = 2, + Stalls = 1, + ClaimedBy = "all@host", + StartedAt = At(2), + LeaseUntil = At(62), + FinishedAt = At(9), + UpdatedAt = At(9), + ProgressDone = 7, + ProgressTotal = 8, + ProgressNote = "7 of 8", + Result = /*lang=json,strict*/ """{"found": 8}""", + Error = "an earlier error", + }); + + var view = await Reader().FindAsync(First, Ct); + + view.Should().Be(new WorkItemView( + First, WorkKind.Crawl, configId, adminId, At(1), WorkItemStatus.Succeeded, 2, 1, "all@host", + At(2), At(9), At(9), 7, 8, "7 of 8", /*lang=json,strict*/ """{"found": 8}""", "an earlier error")); + } + + [Fact] + public async Task FindAsync_UnknownId_ReturnsNullAsync() + { + await SeedAsync(Item(First)); + + (await Reader().FindAsync(Second, Ct)).Should().BeNull(); + } + + // ========================================================================== + // UpdatedSinceAsync + // ========================================================================== + + [Fact] + public async Task UpdatedSinceAsync_ReturnsItemsChangedStrictlyAfterInChangeOrderAsync() + { + await SeedAsync( + Item(First, updatedAt: 12), + Item(Second, updatedAt: 10), + Item(Third, updatedAt: 11)); + + var changed = await Reader().UpdatedSinceAsync(At(10), Ct); + + Ids(changed).Should().Equal(Third, First); + } + + [Fact] + public async Task UpdatedSinceAsync_SameChangeTime_PutsTheSmallerIdFirstAsync() + { + await SeedAsync(Item(Second, updatedAt: 11), Item(First, updatedAt: 11)); + + var changed = await Reader().UpdatedSinceAsync(At(10), Ct); + + Ids(changed).Should().Equal(First, Second); + } + + [Fact] + public void UpdatedSinceAsync_OnPostgres_FiltersAndOrdersInTheQuery() + { + using var db = AnichronDbContexts.NpgsqlModel(); + + var sql = EfWorkItemReader.UpdatedSinceQuery(db, At(10)).ToQueryString(); + + sql.Should().Contain("""WHERE w."UpdatedAt" > @since""") + .And.Contain("""ORDER BY w."UpdatedAt", w."Id" """.TrimEnd()); + } + + // ========================================================================== + // LatestUpdateAsync + // ========================================================================== + + [Fact] + public async Task LatestUpdateAsync_ReturnsTheNewestChangeTimeAsync() + { + await SeedAsync(Item(First, updatedAt: 12), Item(Second, updatedAt: 30), Item(Third, updatedAt: 11)); + + (await Reader().LatestUpdateAsync(Ct)).Should().Be(At(30)); + } + + [Fact] + public async Task LatestUpdateAsync_NoItems_ReturnsNullAsync() + => (await Reader().LatestUpdateAsync(Ct)).Should().BeNull(); +} diff --git a/src/Anichron.Core/Work/WorkItemReader.cs b/src/Anichron.Core/Work/WorkItemReader.cs new file mode 100644 index 0000000..f3bd793 --- /dev/null +++ b/src/Anichron.Core/Work/WorkItemReader.cs @@ -0,0 +1,94 @@ +using Anichron.Core.Data; +using Anichron.Core.Domain; +using Microsoft.EntityFrameworkCore; +using System.Linq.Expressions; + +namespace Anichron.Core.Work; + +public sealed record WorkItemView( + Guid Id, + WorkKind Kind, + Guid? StorageConfigId, + Guid? RequestedBy, + Instant RequestedAt, + WorkItemStatus Status, + int Attempt, + int Stalls, + string? ClaimedBy, + Instant? StartedAt, + Instant? FinishedAt, + Instant UpdatedAt, + long ProgressDone, + long? ProgressTotal, + string? ProgressNote, + string? Result, + string? Error); + +public sealed record WorkItemQuery(WorkItemStatus? Status, WorkKind? Kind, int Skip, int Take); + +public sealed record WorkItemPage(IReadOnlyList Items, int Total); + +public interface IWorkItemReader +{ + /// A page of items, newest request first. + /// The optional status and kind filters, and the page. + /// Cancels the read. + /// The page, and the number of items matching the filters. + Task ListAsync(WorkItemQuery query, CancellationToken ct); + + Task FindAsync(Guid id, CancellationToken ct); + + /// Every item that changed after a point in time, oldest change first. + /// Only items whose is later are returned. + /// Cancels the read. + /// The changed items, ordered by and then id. + Task> UpdatedSinceAsync(Instant since, CancellationToken ct); + + /// The newest of any item. + /// Cancels the read. + /// The newest change time, or null when there are no items. + Task LatestUpdateAsync(CancellationToken ct); +} + +public sealed class EfWorkItemReader(AnichronDbContext db) : IWorkItemReader +{ + public async Task ListAsync(WorkItemQuery query, CancellationToken ct) + { + var matching = db.WorkItems.AsQueryable(); + if (query.Status is { } status) + matching = matching.Where(w => w.Status == status); + if (query.Kind is { } kind) + matching = matching.Where(w => w.Kind == kind); + + var items = await matching + .OrderByDescending(w => w.RequestedAt) + .ThenByDescending(w => w.Id) + .Skip(query.Skip) + .Take(query.Take) + .Select(ToView) + .ToListAsync(ct); + return new WorkItemPage(items, await matching.CountAsync(ct)); + } + + public Task FindAsync(Guid id, CancellationToken ct) + => db.WorkItems.Where(w => w.Id == id).Select(ToView).FirstOrDefaultAsync(ct); + + public async Task> UpdatedSinceAsync(Instant since, CancellationToken ct) + => await UpdatedSinceQuery(db, since).ToListAsync(ct); + + public Task LatestUpdateAsync(CancellationToken ct) + => db.WorkItems.MaxAsync(w => (Instant?)w.UpdatedAt, ct); + + internal static IQueryable UpdatedSinceQuery(AnichronDbContext db, Instant since) + => db.WorkItems + .Where(w => w.UpdatedAt > since) + .OrderBy(w => w.UpdatedAt) + .ThenBy(w => w.Id) + .Select(ToView); + + private static Expression> ToView + => w => new WorkItemView( + w.Id, w.Kind, w.StorageConfigId, w.RequestedBy, w.RequestedAt, w.Status, w.Attempt, w.Stalls, + w.ClaimedBy, w.StartedAt, w.FinishedAt, w.UpdatedAt, w.ProgressDone, w.ProgressTotal, + w.ProgressNote, w.Result, w.Error); +} From 3184b25be67d0878db348c28882ebd019e6bba16 Mon Sep 17 00:00:00 2001 From: bitbiter-dev Date: Tue, 20 Oct 2026 21:07:35 +1300 Subject: [PATCH 3/3] refactor(api): Carry work progress as one value, format instants in one place (#279) WorkItemView holds Core's WorkProgress rather than three loose fields, and the event id and the list cursor share the mapper's ISO formatter, so the two cursors cannot drift apart. Asserts the paging validation messages (two surviving mutants) and makes the HTTP event-stream test independent of when the server reaches its next poll. Documents the admin view and the heartbeat rule in CLAUDE.md. Co-Authored-By: Claude Opus 5.5 --- CLAUDE.md | 3 +- .../Endpoints/WorkItemEndpointsHttpTests.cs | 37 ++++++++++--------- .../Endpoints/WorkItemEndpointsTests.cs | 18 +++++---- .../Endpoints/WorkItemEventsTests.cs | 2 +- .../Endpoints/WorkItemDtoMapper.cs | 2 +- .../Endpoints/WorkItemEndpoints.cs | 2 +- src/Anichron.API/Endpoints/WorkItemEvents.cs | 4 +- .../Work/WorkItemReaderTests.cs | 2 +- src/Anichron.Core/Work/WorkItemReader.cs | 8 ++-- 9 files changed, 39 insertions(+), 39 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index bdc189d..8094eeb 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -199,7 +199,8 @@ Durable work for the Worker lives in `WorkItems` (#276, ADR 0008). Its kinds are - **Producing:** `IWorkQueue.EnqueueAsync` (`EfWorkQueue`, Core) is the only insert. A payload with a `DedupKey` coalesces: while an item with that key is `Queued`, enqueueing returns it (a partial unique index absorbs racing inserts). - **Scope:** a scoped item carries the `StorageConfigId` it belongs to (`IWorkPayload.StorageConfigId`). A per-user Worker (`WORKER__USER`) claims unscoped items and items of its user's configs; an all-user Worker claims everything. The served user is `WorkerState.ResolvedUserId`, read at each claim and tick, never cached at construction. - **Scheduling:** every Worker runs `WorkScheduler`. Each poll (`Worker:WorkPollIntervalSeconds`, 5 s) it recovers expired leases, then asks each `IWorkScheduleSource` for its schedules (a fixed `WorkSchedule` is its own source; `CrawlSchedules` returns one per active config the Worker serves) and whether each is due: nothing queued or running with its key and nothing finished within its interval. No timer state is kept, so overdue work runs right after downtime. -- **Executing:** `WorkExecutor` claims the oldest queued item (`FOR UPDATE SKIP LOCKED`), one at a time, under a lease (`WorkLeaseSeconds`, 60 s) that a heartbeat (`WorkHeartbeatSeconds`, 2 s) extends while writing the progress snapshot. Every write about a running item is fenced by `(Id, Attempt)`. A fenced heartbeat cancels the handler and writes nothing more; a shutdown releases the item without a stall; a handler exception fails it with no retry. An expired lease requeues the item and counts a stall; the `WorkMaxStalls`th (3) fails it as stalled. An item handed back (release or recovery) while another with its key is already queued ends `Cancelled` as superseded, since the queued one does the same work. A Worker claims only the kinds it knows, so a newer kind waits for a newer Worker. `WorkItemRetention` deletes items finished more than `WorkItemRetentionDays` (30) ago. +- **Executing:** `WorkExecutor` claims the oldest queued item (`FOR UPDATE SKIP LOCKED`), one at a time, under a lease (`WorkLeaseSeconds`, 60 s) that a heartbeat (`WorkHeartbeatSeconds`, 2 s) extends while writing the progress snapshot. `UpdatedAt` is the change cursor, so a heartbeat moves it only when the snapshot changed; `LeaseUntil` is what says the Worker is alive. Every write about a running item is fenced by `(Id, Attempt)`. A fenced heartbeat cancels the handler and writes nothing more; a shutdown releases the item without a stall; a handler exception fails it with no retry. An expired lease requeues the item and counts a stall; the `WorkMaxStalls`th (3) fails it as stalled. An item handed back (release or recovery) while another with its key is already queued ends `Cancelled` as superseded, since the queued one does the same work. A Worker claims only the kinds it knows, so a newer kind waits for a newer Worker. `WorkItemRetention` deletes items finished more than `WorkItemRetentionDays` (30) ago. +- **Admin view (#279):** `GET /api/v1/admin/work-items` (filtered by status and kind, newest request first, paged), `.../{id}`, and `.../events`, a Server-Sent Events stream that re-queries `UpdatedAt` every `WorkItemEvents:IntervalSeconds` (2 s). An event's id is the cursor, and the list carries the cursor read before its page, so a client that lists and then subscribes with it as `Last-Event-ID` misses nothing; an item may arrive twice and is merged by id. The API reads items only through `IWorkItemReader` (Core), as `WorkItemView`. A result is shown once an item succeeded, an error once it failed or was cancelled. - **Adding a kind:** a `WorkKind` value, a payload record in Core implementing `IWorkPayload` (static `Kind`, optional `DedupKey`), an `IWorkHandler` in the Worker registered with `AddWorkHandler`, and for a periodic kind a `WorkSchedule` subclass registered as an `IWorkScheduleSource` (or a source that builds several). `WorkHandlerRegistry` fails startup unless every kind has exactly one payload and one handler; nothing switches on the kind. - **Handler contract:** delivery is at least once. Each run starts from scratch and converges, side effects are safe to repeat, and the returned summary becomes `Result`. diff --git a/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEndpointsHttpTests.cs b/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEndpointsHttpTests.cs index 063d565..85950cb 100644 --- a/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEndpointsHttpTests.cs +++ b/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEndpointsHttpTests.cs @@ -3,8 +3,10 @@ using Anichron.API.Tests.Unit.Infrastructure; using Anichron.Core.Domain; using Anichron.Core.Work; +using Anichron.Testing.Time; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Options; +using Microsoft.Extensions.Time.Testing; using System.Net; using System.Text.Json; @@ -22,24 +24,18 @@ public sealed class WorkItemEndpointsHttpTests private static WorkItemView Succeeded => new(ItemId, WorkKind.Crawl, null, null, Start, WorkItemStatus.Succeeded, 1, 0, "all@host", - Start, Start + Duration.FromSeconds(9), Start + Duration.FromSeconds(9), 8, 8, "8 found", + Start, Start + Duration.FromSeconds(9), Start + Duration.FromSeconds(9), new WorkProgress(8, 8, "8 found"), /*lang=json,strict*/ """{"found": 8}""", null); private Task StartAsync() => AuthenticatedTestApp.StartAsync(api => api.MapWorkItemEndpoints(), services => { services.AddSingleton(reader); - services.AddSingleton(new ImmediateTimers()); + services.AddSingleton(new FakeTimeProvider()); services.AddSingleton(Options.Create(new WorkItemEventsSettings())); services.AddScoped(); }); - private sealed class ImmediateTimers : TimeProvider - { - public override ITimer CreateTimer(TimerCallback callback, object? state, TimeSpan dueTime, TimeSpan period) - => base.CreateTimer(callback, state, TimeSpan.Zero, Timeout.InfiniteTimeSpan); - } - [Fact] public async Task List_BindsTheFiltersAndPagingFromTheQueryAsync() { @@ -90,24 +86,29 @@ public async Task EveryRoute_ForANonAdmin_IsForbiddenAsync(string path) public async Task Events_ResumeFromLastEventIdAndFrameEachItemAsync() { var polls = 0; - reader.UpdatedSinceAsync(Arg.Any(), Arg.Any()).Returns(_ => ++polls == 1 - ? Task.FromResult>([Succeeded]) - : throw new InvalidOperationException("The stream is read once.")); + reader.UpdatedSinceAsync(Arg.Any(), Arg.Any()) + .Returns(_ => ++polls == 1 ? [Succeeded] : Array.Empty()); + await using var timeline = new Timeline(); await using var app = await StartAsync(); using var client = app.Client; using var request = AuthenticatedTestApp.GetAsAdmin("/api/v1/admin/work-items/events", AdminId); request.Headers.Add("Last-Event-ID", "2026-10-18T09:00:05Z"); - using var response = await client.SendAsync(request, HttpCompletionOption.ResponseHeadersRead, Ct); + timeline.Watch(Task.Run(async () => timeline.Record(Describe(await FirstFrameAsync(response))), CancellationToken.None)); + + await timeline.ExpectAsync(new Called( + "event: work-item | id: 2026-10-18T09:00:09Z | aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa 8 found")); response.Content.Headers.ContentType!.MediaType.Should().Be("text/event-stream"); await reader.Received(1).UpdatedSinceAsync(Start - Duration.FromSeconds(5), Arg.Any()); - var frame = await FirstFrameAsync(response); - frame.Should().HaveCount(3).And.Contain("event: work-item").And.Contain("id: 2026-10-18T09:00:09Z"); - var json = frame.Should().ContainSingle(line => line.StartsWith("data: ", StringComparison.Ordinal)).Which; - using var data = JsonDocument.Parse(json["data: ".Length..]); - data.RootElement.GetProperty("id").GetGuid().Should().Be(ItemId); - data.RootElement.GetProperty("progress").GetProperty("note").GetString().Should().Be("8 found"); + } + + private static string Describe(List frame) + { + var data = frame.Single(line => line.StartsWith("data: ", StringComparison.Ordinal)); + using var json = JsonDocument.Parse(data["data: ".Length..]); + var item = $"{json.RootElement.GetProperty("id").GetGuid()} {json.RootElement.GetProperty("progress").GetProperty("note").GetString()}"; + return string.Join(" | ", frame.Where(line => line != data).Order(StringComparer.Ordinal).Append(item)); } private static async Task> FirstFrameAsync(HttpResponseMessage response) diff --git a/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEndpointsTests.cs b/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEndpointsTests.cs index 6996509..e3a4b4b 100644 --- a/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEndpointsTests.cs +++ b/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEndpointsTests.cs @@ -25,7 +25,7 @@ private static WorkItemView View( string? result = /*lang=json,strict*/ """{"found": 8}""", string? error = "IOException: the root is gone") => new(ItemId, WorkKind.Crawl, ConfigId, AdminId, At(1), status, 2, 1, "all@host", - At(2), At(9), At(9), 7, 8, "7 found", result, error); + At(2), At(9), At(9), new WorkProgress(7, 8, "7 found"), result, error); private void Pages(int total = 1, Instant? latest = null) { @@ -106,18 +106,20 @@ public async Task ListAsync_PagingInRange_IsReadAsync(int page, int pageSize, in } [Theory] - [InlineData(0, 50, "page")] - [InlineData(-1, 50, "page")] - [InlineData(10_737_419, 200, "page")] - [InlineData(1, 0, "pageSize")] - [InlineData(1, 201, "pageSize")] - public async Task ListAsync_PagingOutOfRange_IsAValidationProblemAsync(int page, int pageSize, string field) + [InlineData(0, 50, "page", "page is out of range.")] + [InlineData(-1, 50, "page", "page is out of range.")] + [InlineData(10_737_419, 200, "page", "page is out of range.")] + [InlineData(1, 0, "pageSize", "pageSize must be from 1 to 200.")] + [InlineData(1, 201, "pageSize", "pageSize must be from 1 to 200.")] + public async Task ListAsync_PagingOutOfRange_IsAValidationProblemAsync(int page, int pageSize, string field, string message) { Pages(); var result = await WorkItemEndpoints.ListAsync(null, null, page, pageSize, reader, Ct); - result.Result.Should().BeOfType().Which.ProblemDetails.Errors.Keys.Should().Equal(field); + var errors = result.Result.Should().BeOfType().Which.ProblemDetails.Errors; + errors.Keys.Should().Equal(field); + errors[field].Should().Equal(message); await reader.DidNotReceive().ListAsync(Arg.Any(), Arg.Any()); } diff --git a/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEventsTests.cs b/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEventsTests.cs index e3a2402..ec3a252 100644 --- a/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEventsTests.cs +++ b/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEventsTests.cs @@ -41,7 +41,7 @@ public WorkItemEventsTests() private static WorkItemView Changed(Guid id, int updatedAt) => new(id, WorkKind.Crawl, null, null, Start, WorkItemStatus.Running, 1, 0, "all@host", - Start, null, At(updatedAt), 3, null, "3 found", null, null); + Start, null, At(updatedAt), new WorkProgress(3, null, "3 found"), null, null); private void Stream(string? lastEventId = null, double intervalSeconds = 2) { diff --git a/src/Anichron.API/Endpoints/WorkItemDtoMapper.cs b/src/Anichron.API/Endpoints/WorkItemDtoMapper.cs index c374a84..e81a823 100644 --- a/src/Anichron.API/Endpoints/WorkItemDtoMapper.cs +++ b/src/Anichron.API/Endpoints/WorkItemDtoMapper.cs @@ -21,7 +21,7 @@ internal static WorkItemDto ToDto(WorkItemView view) view.StartedAt is { } started ? Format(started) : null, view.FinishedAt is { } finished ? Format(finished) : null, Format(view.UpdatedAt), - new WorkProgressDto(view.ProgressDone, view.ProgressTotal, view.ProgressNote), + new WorkProgressDto(view.Progress.Done, view.Progress.Total, view.Progress.Note), view is { Status: WorkItemStatus.Succeeded, Result: { } result } ? JsonSerializer.Deserialize(result) : null, diff --git a/src/Anichron.API/Endpoints/WorkItemEndpoints.cs b/src/Anichron.API/Endpoints/WorkItemEndpoints.cs index 95a33aa..fb63c13 100644 --- a/src/Anichron.API/Endpoints/WorkItemEndpoints.cs +++ b/src/Anichron.API/Endpoints/WorkItemEndpoints.cs @@ -39,7 +39,7 @@ internal static async Task, ValidationProblem>> List var cursor = await reader.LatestUpdateAsync(ct) ?? NodaConstants.UnixEpoch; var found = await reader.ListAsync(new WorkItemQuery(status, kind, (number - 1) * size, size), ct); return TypedResults.Ok(new WorkItemPageDto( - [.. found.Items.Select(WorkItemDtoMapper.ToDto)], number, size, found.Total, WorkItemEvents.Format(cursor))); + [.. found.Items.Select(WorkItemDtoMapper.ToDto)], number, size, found.Total, WorkItemDtoMapper.Format(cursor))); } internal static async Task, NotFound>> GetAsync( diff --git a/src/Anichron.API/Endpoints/WorkItemEvents.cs b/src/Anichron.API/Endpoints/WorkItemEvents.cs index 61227b6..9484a05 100644 --- a/src/Anichron.API/Endpoints/WorkItemEvents.cs +++ b/src/Anichron.API/Endpoints/WorkItemEvents.cs @@ -34,7 +34,7 @@ public async IAsyncEnumerable> StreamAsync( cursor = Instant.Max(cursor, item.UpdatedAt); yield return new SseItem(WorkItemDtoMapper.ToDto(item), EventType) { - EventId = Format(cursor), + EventId = WorkItemDtoMapper.Format(cursor), }; } @@ -43,8 +43,6 @@ public async IAsyncEnumerable> StreamAsync( } } - internal static string Format(Instant cursor) => InstantPattern.ExtendedIso.Format(cursor); - private static Instant? Parse(string? lastEventId) => lastEventId is not null && InstantPattern.ExtendedIso.Parse(lastEventId) is { Success: true } parsed ? parsed.Value diff --git a/src/Anichron.Core.Tests.Unit/Work/WorkItemReaderTests.cs b/src/Anichron.Core.Tests.Unit/Work/WorkItemReaderTests.cs index 82da127..531be3b 100644 --- a/src/Anichron.Core.Tests.Unit/Work/WorkItemReaderTests.cs +++ b/src/Anichron.Core.Tests.Unit/Work/WorkItemReaderTests.cs @@ -174,7 +174,7 @@ await SeedAsync(new WorkItem view.Should().Be(new WorkItemView( First, WorkKind.Crawl, configId, adminId, At(1), WorkItemStatus.Succeeded, 2, 1, "all@host", - At(2), At(9), At(9), 7, 8, "7 of 8", /*lang=json,strict*/ """{"found": 8}""", "an earlier error")); + At(2), At(9), At(9), new WorkProgress(7, 8, "7 of 8"), /*lang=json,strict*/ """{"found": 8}""", "an earlier error")); } [Fact] diff --git a/src/Anichron.Core/Work/WorkItemReader.cs b/src/Anichron.Core/Work/WorkItemReader.cs index f3bd793..687efad 100644 --- a/src/Anichron.Core/Work/WorkItemReader.cs +++ b/src/Anichron.Core/Work/WorkItemReader.cs @@ -18,9 +18,7 @@ public sealed record WorkItemView( Instant? StartedAt, Instant? FinishedAt, Instant UpdatedAt, - long ProgressDone, - long? ProgressTotal, - string? ProgressNote, + WorkProgress Progress, string? Result, string? Error); @@ -89,6 +87,6 @@ internal static IQueryable UpdatedSinceQuery(AnichronDbContext db, private static Expression> ToView => w => new WorkItemView( w.Id, w.Kind, w.StorageConfigId, w.RequestedBy, w.RequestedAt, w.Status, w.Attempt, w.Stalls, - w.ClaimedBy, w.StartedAt, w.FinishedAt, w.UpdatedAt, w.ProgressDone, w.ProgressTotal, - w.ProgressNote, w.Result, w.Error); + w.ClaimedBy, w.StartedAt, w.FinishedAt, w.UpdatedAt, + new WorkProgress(w.ProgressDone, w.ProgressTotal, w.ProgressNote), w.Result, w.Error); }