diff --git a/docs/.vitepress/config-base.ts b/docs/.vitepress/config-base.ts index 4a4eb654..b5bf73a3 100644 --- a/docs/.vitepress/config-base.ts +++ b/docs/.vitepress/config-base.ts @@ -261,6 +261,7 @@ export const baseConfig = { { text: 'Resource server (.NET)', link: '/integrate/resource-server' }, { text: 'SaaS app walkthrough', link: '/integrate/saas-walkthrough' }, { text: 'Modgud Management API', link: '/integrate/management-api' }, + { text: 'Application change feed', link: '/integrate/application-change-feed' }, { text: 'Secure an MCP server', link: '/integrate/mcp-server' }, { text: 'Native apps (iOS / mobile)', link: '/integrate/native-apps' }, { text: 'OAuth / OpenIddict', link: '/integrate/oauth' }, diff --git a/docs/admin/service-accounts.md b/docs/admin/service-accounts.md index 68a1bae3..d5e89593 100644 --- a/docs/admin/service-accounts.md +++ b/docs/admin/service-accounts.md @@ -171,6 +171,13 @@ dotnet Modgud.Api.dll recover migrate-cc-credentials [--realm ] For each un-linked `client_credentials` client the command auto-provisions a Service Account named `legacy.{clientId}`, links the client to it, and leaves a re-runnable trail (already-linked clients are skipped; existing `legacy.*` SAs are re-used). Defaults to the `system` realm; pass `--realm` to scope to a specific tenant. After migration, rename the SA from the admin UI or merge it into a properly-named one. +## Event history and existing accounts + +Service Account create, update, and delete operations are event-sourced. A +legacy document-only account is upgraded lazily on its first mutation by +seeding its current snapshot as the stream's creation event; operators do not +need a one-time data migration. + ## Related - [OAuth Clients](./oauth-clients) — the global grid that lists user-facing clients alongside SA-managed credentials with an M2M column linking back here. diff --git a/docs/integrate/application-change-feed.md b/docs/integrate/application-change-feed.md new file mode 100644 index 00000000..2fe265a3 --- /dev/null +++ b/docs/integrate/application-change-feed.md @@ -0,0 +1,195 @@ +--- +title: Application change feed +description: Bootstrap and continuously synchronize the Modgud entities assigned to one Application through a resumable, app-scoped contract. +--- + +# Application change feed + +The Application change feed lets a consumer keep a local read model of the +Modgud entities assigned to that Application. It is a general Modgud contract, +not an AlertHub-specific event stream. + +The feed deliberately does **not** expose raw event-sourcing events. Modgud's +event store remains the permanent business history. The consumer contract is a +short-lived, versioned projection with a full-snapshot escape hatch: + +1. take a full snapshot; +2. persist the returned opaque cursor with the imported state; +3. apply incremental changes and advance the cursor; +4. take a new snapshot whenever Modgud says that the cursor can no longer be + resumed. + +SSE is the live transport for that contract. Consumers that cannot keep a +long-lived connection can read the same queue through HTTP polling. The +envelope remains transport-neutral, so another transport can be added later +without changing the synchronization model. + +## Enable and authorize it + +The feed is off by default. In **Administration → Applications**, open the +Application's **Settings → Sync** tab and enable **Consumer change feed**. The +same tab configures the retention union: + +- keep every change newer than the minimum age; and +- also keep at least the newest configured number of changes. + +The defaults are 7 days and 1,000 changes. Because those conditions form a +union, a quiet Application does not lose its last resume window merely because +seven days passed. + +The caller uses the [Management API](./management-api) contract: + +- token scope `modgud.management`; +- audience `urn:modgud:management-api`; +- live Modgud permission `app-scope:read`; and +- the requested Application id in the OAuth client's `AppIds` assignment. + +An empty `AppIds` list grants access to no Application. One OAuth client may be +assigned to multiple Applications, but every snapshot and subscription still +targets exactly one `appId`. + +## Scope + +The feed uses the same Application scope as +`GET /api/app/{appId}/scope`: active groups whose `BoundTo` contains the App +slug or `*` are roots, and their transitive active members are in scope. A +group does not need to grant permissions to act as a scope-only grouping. + +The public projection currently contains these entity kinds: + +| `EntityKind` | Public meaning | +|---|---| +| `principal` | Person, Group, Service Account, or Position in the App scope | +| `terminal` | Non-revoked terminal with at least one allowed Position in scope | +| `position-grant` | Non-revoked user-to-Position grant whose two ends are in scope | +| `staffing-session` | Active session for an in-scope Position and terminal | + +Secrets, credential identifiers, role ids, DPoP thumbprints, and internal OAuth +authorization ids are not part of this contract. Group member lists and +terminal Position lists are filtered to the requested Application scope. + +## Bootstrap snapshot + +```http +GET /api/app/{appId}/change-feed/snapshot +Authorization: Bearer +``` + +The response contains: + +- `ContractVersion` — currently `1`; +- `AppId` and `AppSlug`; +- an opaque `ScopeVersion`; +- an opaque resume `Cursor`; and +- the complete current `Entities` collection. + +Treat the snapshot and cursor as one transaction in the consumer database. Do +not synthesize, parse, compare, or increment cursors; persist them verbatim. + +Immediately after enablement the endpoint can briefly return +`FeedInitializing` (`409`) until the asynchronous projection reaches the +enablement event. Retry with backoff. + +## Subscribe with SSE + +Open the stream with the snapshot cursor and the same Management API token. +Unlike the finite snapshot and polling routes, the long-lived stream is +bearer-only; Modgud admin cookies are not accepted on this endpoint: + +```http +GET /api/app/{appId}/change-feed/stream?cursor=&batchSize=100 +Authorization: Bearer +Accept: text/event-stream +``` + +The server emits standard SSE frames. `id` is always the opaque Modgud cursor; +`data` is the same JSON envelope returned by HTTP polling. + +```text +id: AQ... +event: change +data: {"ContractVersion":1,"Kind":"Change","Cursor":"AQ...",...} +``` + +The event names are `change`, `checkpoint`, `reset-required`, and +`feed-ended`. A comment heartbeat is sent after 15 seconds without data so +proxies can keep the connection alive. + +On reconnect, pass the last cursor committed together with the local state. +Use either `?cursor=...` or the standard `Last-Event-ID` request header. Do not +resume from an in-memory cursor whose corresponding entity changes were not +committed. + +The stream rechecks the token expiry, OAuth client status, `AppIds` assignment, +Service Account or Person status, and live `app-scope:read` permission every +poll cycle. If authorization ceases after the response has started, Modgud +emits `feed-ended` with the reason and closes the connection. Acquire a fresh +token before reconnecting when the reason is `token_expired`. + +Native browser `EventSource` cannot attach an `Authorization` header. The +recommended integration is therefore a backend/M2M HTTP client. A browser that +must connect directly can use streaming `fetch`; do not put access tokens into +the URL. + +## Polling fallback + +```http +GET /api/app/{appId}/change-feed?cursor=&limit=100 +Authorization: Bearer +``` + +`limit` is clamped to 1–500. Continue immediately while `HasMore` is true; +otherwise poll at a consumer-appropriate interval. The HTTP and SSE +paths use the same cursor and message envelope, so changing transport does not +require rebuilding the local model. Even an empty polling response carries the +current `ContractVersion` and `ScopeVersion` alongside `Messages`. + +## Message handling + +| `Kind` | Required consumer action | +|---|---| +| `Change` + `Upsert` | Replace the identified entity with the versioned payload. | +| `Change` + `Deleted` | Remove it; an optional payload describes the terminal state. | +| `Change` + `FellOutOfScope` | Remove it locally without treating it as deleted in Modgud. | +| `Checkpoint` | Commit the cursor even though no entity changed. | +| `ResetRequired` | Stop applying this stream and take a fresh snapshot. | +| `FeedEnded` | Commit preceding changes and stop; the App feed was disabled or removed. | + +Every entity has an `EntityVersion`; version 1 payloads are defined by this +page. Ignore unknown payload properties for forward compatibility. An +incremental envelope may also carry `SourceEventId` and `OriginatedAt` for +correlation. They are metadata, not an idempotency key: the cursor is the +ordering and resume contract. + +Apply each message and its cursor atomically. Replaying the same `Upsert` or +removal must be harmless. + +## Reset conditions + +A new full snapshot is mandatory when the API returns or the stream emits: + +- `ScopeChanged` — an App binding, nested-group structure, or another scope + definition changed; +- `CursorTooOld` — retention already removed part of the required resume + window; or +- a `ResetRequired` message with either reason. + +Malformed cursors, cursors belonging to another App, and cursors pointing past +the server checkpoint are rejected as `InvalidCursor`. Do not fall back to an +empty cursor; take a new snapshot only for an explicit reset condition. + +## Service Accounts and event replay + +Service Accounts now have create, update, and delete streams like the other +Principal kinds. Existing installations can still contain legacy +document-only Service Accounts. Their first mutation seeds a creation snapshot +before recording the mutation, so no manual migration or projection rebuild is +required for the change feed. + +## Related + +- [Management API](./management-api) — token and live-permission setup +- [Applications](/admin/applications) — App catalog and `BoundTo` model +- [Service Accounts](/admin/service-accounts) — unattended credentials +- [Position terminals](./position-terminals) — terminal and staffing wire + contracts diff --git a/docs/integrate/index.md b/docs/integrate/index.md index 15a40d7b..bc771eff 100644 --- a/docs/integrate/index.md +++ b/docs/integrate/index.md @@ -20,6 +20,9 @@ protocol-specific pages. - [Modgud Management API](./management-api) — let a backend Service Account or delegated administrator read deliberately exposed Modgud resources using the same live permissions as the first-party admin UI. +- [Application change feed](./application-change-feed) — bootstrap an + Application's current Principal/Position scope and resume its public changes + over SSE or HTTP polling. ## Protocol pages diff --git a/docs/integrate/management-api.md b/docs/integrate/management-api.md index 00f53b85..aaf76a8f 100644 --- a/docs/integrate/management-api.md +++ b/docs/integrate/management-api.md @@ -249,6 +249,9 @@ scheme does not turn every cookie-only admin route into a remote API. | `GET` | `/api/position` | `position:read` | `PositionTerminals` enabled | | `GET` | `/api/position/{id}` | `position:read` | `PositionTerminals` enabled | | `GET` | `/api/app/{id}/scope` | `app-scope:read` | Full `BoundTo`-derived Principal snapshot | +| `GET` | `/api/app/{id}/change-feed/snapshot` | `app-scope:read` | Feed enabled; full public synchronization snapshot | +| `GET` | `/api/app/{id}/change-feed` | `app-scope:read` | Feed enabled; resumable HTTP read using an opaque cursor | +| `GET` | `/api/app/{id}/change-feed/stream` | `app-scope:read` | Feed enabled; bearer-only resumable SSE stream using the same cursor and envelope | | `POST` | `/api/admin/oauth/clients` | `oauth-client:write` | `position:write` for terminal provisioning; `service-account:write` for an SA link | Direct Position creation, mutation, deletion, grants, terminal enrollment, and @@ -275,6 +278,8 @@ exposed surface. - [Service Accounts](/admin/service-accounts) — machine identity, credentials, rotation, and group membership - [Permissions & gating](/concepts/permissions) — the live authorization model +- [Application change feed](./application-change-feed) — full sync, cursor, + retention, SSE, and HTTP fallback - [OAuth / OpenIddict](./oauth) — token flows and RFC 8707 resource indicators - [Positions & Terminals](/admin/positions) — administering the first exposed resource diff --git a/src/dotnet/Modgud.Api.Tests/Authorization/ApplicationSettingsAdminTests.cs b/src/dotnet/Modgud.Api.Tests/Authorization/ApplicationSettingsAdminTests.cs index e4f31d74..f3f96b8e 100644 --- a/src/dotnet/Modgud.Api.Tests/Authorization/ApplicationSettingsAdminTests.cs +++ b/src/dotnet/Modgud.Api.Tests/Authorization/ApplicationSettingsAdminTests.cs @@ -132,6 +132,43 @@ public async Task Update_Settings_Roundtrips_And_Writes_The_Routing_Map() Assert.Equal(app.Id, mappedAppId); } + [Fact] + public async Task Change_Feed_Settings_Roundtrip_And_Emit_Only_Effective_Changes() + { + var ct = TestContext.Current.CancellationToken; + var app = await SeedAppAsync("as-change-feed"); + var appShort = ShortGuid.Encode(app.Id); + var enabled = new ApplicationSettingsDto + { + ChangeFeed = new ApplicationChangeFeedDto + { + Enabled = true, + MinimumRetentionAgeDays = 14, + MinimumEventCount = 10_000, + }, + }; + + (await PutSettingsAsync(appShort, enabled, ct)).EnsureSuccessStatusCode(); + // Repeating the same full-replace settings is idempotent and must not + // wake feed consumers with a spurious policy-change event. + (await PutSettingsAsync(appShort, enabled, ct)).EnsureSuccessStatusCode(); + + var read = (await GetAppAsync(appShort, ct)).Settings!.ChangeFeed!; + Assert.True(read.Enabled); + Assert.Equal(14, read.MinimumRetentionAgeDays); + Assert.Equal(10_000, read.MinimumEventCount); + + await using var query = GetTenantedSession(); + var events = await query.Events.FetchStreamAsync(app.Id, token: ct); + var configured = events.Select(x => x.Data) + .OfType() + .ToList(); + var change = Assert.Single(configured); + Assert.True(change.Enabled); + Assert.Equal(14, change.MinimumRetentionAgeDays); + Assert.Equal(10_000, change.MinimumEventCount); + } + [Fact] public async Task Update_Rejects_Invalid_Values() { @@ -151,6 +188,17 @@ public async Task Update_Rejects_Invalid_Values() Assert.Equal(HttpStatusCode.BadRequest, (await PutSettingsAsync(appShort, new ApplicationSettingsDto { NativeGrants = new ApplicationNativeGrantsDto { AccessTokenLifetimeMinutes = 9999 } }, ct)).StatusCode); + Assert.Equal(HttpStatusCode.BadRequest, (await PutSettingsAsync(appShort, + new ApplicationSettingsDto + { + ChangeFeed = new ApplicationChangeFeedDto { Enabled = true, MinimumRetentionAgeDays = 0 }, + }, ct)).StatusCode); + Assert.Equal(HttpStatusCode.BadRequest, (await PutSettingsAsync(appShort, + new ApplicationSettingsDto + { + ChangeFeed = new ApplicationChangeFeedDto { Enabled = true, MinimumEventCount = 1_000_001 }, + }, ct)).StatusCode); + // Page theme values are allowlisted CSS colors and bounded radii, never // arbitrary style declarations. Assert.Equal(HttpStatusCode.BadRequest, (await PutSettingsAsync(appShort, diff --git a/src/dotnet/Modgud.Api.Tests/Authorization/ManagementApiAuthorizationTests.cs b/src/dotnet/Modgud.Api.Tests/Authorization/ManagementApiAuthorizationTests.cs index f474cd63..006abedf 100644 --- a/src/dotnet/Modgud.Api.Tests/Authorization/ManagementApiAuthorizationTests.cs +++ b/src/dotnet/Modgud.Api.Tests/Authorization/ManagementApiAuthorizationTests.cs @@ -12,10 +12,13 @@ using Modgud.Application.DTOs.Positions; using Modgud.Application.DTOs.ServiceAccount; using Modgud.Application.Services; +using Modgud.Authorization.Apps; +using Modgud.Authorization.Events; using Modgud.Domain.OAuth.Applications; using Modgud.Domain.OAuth.Common; using Modgud.Domain.OAuth.Management; using Modgud.Domain.OAuth.Scopes; +using Modgud.Infrastructure.ChangeFeed; using Modgud.Infrastructure.OAuth; using OpenIddict.Abstractions; @@ -200,6 +203,124 @@ await CreateClientCredentialsClientAsync(clientId, serviceAccount.Id, Assert.Equal(HttpStatusCode.Unauthorized, response.StatusCode); } + [Fact] + public async Task App_change_feed_requires_both_live_permission_and_client_app_assignment() + { + var ct = TestContext.Current.CancellationToken; + var appId = Guid.CreateVersion7(); + var otherAppId = Guid.CreateVersion7(); + await using (var arrange = GetTenantedDocumentSession()) + { + arrange.Events.StartStream(appId, new AppCreatedEvent( + appId, $"management-feed-{Guid.NewGuid():N}", "Management feed", null, [], false)); + arrange.Events.StartStream(otherAppId, new AppCreatedEvent( + otherAppId, $"management-feed-other-{Guid.NewGuid():N}", "Other app", null, [], false)); + arrange.Store(new AppChangeFeedState + { + Id = appId, + Enabled = true, + Generation = 1, + ScopeVersion = "v1-management-test", + LastProcessedSequence = 1, + }); + await arrange.SaveChangesAsync(ct); + } + + var serviceAccount = await CreateServiceAccountAsync("management-feed-reader"); + Assert.True(ShortGuid.TryParse(serviceAccount.Id, out Guid serviceAccountId)); + await GrantAppScopeReadAsync(serviceAccountId); + + var allowedClientId = $"management-feed-allowed-{Guid.NewGuid():N}"; + await CreateClientCredentialsClientAsync( + allowedClientId, + serviceAccount.Id, + [ModgudManagementApi.Scope], + appIds: [appId]); + var allowedToken = await IssueClientCredentialsTokenAsync( + allowedClientId, ModgudManagementApi.Scope, ModgudManagementApi.Audience); + + string snapshotCursor; + using (var allowed = await SendManagementGetAsync( + allowedToken, + $"/api/app/{ShortGuid.Encode(appId)}/change-feed/snapshot")) + { + var body = await allowed.Content.ReadAsStringAsync(ct); + Assert.True(allowed.IsSuccessStatusCode, + $"assigned feed GET failed ({(int)allowed.StatusCode}): {body}"); + using var snapshot = JsonDocument.Parse(body); + snapshotCursor = snapshot.RootElement.GetProperty("Cursor").GetString()!; + } + + var feedEntityId = Guid.CreateVersion7(); + await using (var append = GetTenantedDocumentSession()) + { + var state = await append.LoadAsync(appId, ct); + state!.LastProcessedSequence = 2; + append.Store(state); + append.Store(new AppChangeFeedEntry + { + Id = Guid.CreateVersion7(), + AppId = appId, + Generation = 1, + SourceSequence = 2, + Ordinal = 0, + ScopeVersion = state.ScopeVersion, + OriginatedAt = DateTimeOffset.UtcNow, + RecordedAt = DateTimeOffset.UtcNow, + ChangeKind = "Upsert", + EntityKind = "principal", + EntityId = feedEntityId, + PayloadJson = "{\"DisplayName\":\"SSE principal\"}", + }); + await append.SaveChangesAsync(ct); + } + + using (var streamCts = CancellationTokenSource.CreateLinkedTokenSource(ct)) + { + streamCts.CancelAfter(TimeSpan.FromSeconds(10)); + using var streamClient = Factory.CreateClient(); + using var streamRequest = new HttpRequestMessage( + HttpMethod.Get, + $"/api/app/{ShortGuid.Encode(appId)}/change-feed/stream"); + streamRequest.Headers.Authorization = new AuthenticationHeaderValue("Bearer", allowedToken); + streamRequest.Headers.Accept.ParseAdd("text/event-stream"); + streamRequest.Headers.TryAddWithoutValidation("Last-Event-ID", snapshotCursor); + using var streamResponse = await streamClient.SendAsync( + streamRequest, HttpCompletionOption.ResponseHeadersRead, streamCts.Token); + Assert.Equal(HttpStatusCode.OK, streamResponse.StatusCode); + Assert.Equal("text/event-stream", streamResponse.Content.Headers.ContentType?.MediaType); + + await using var body = await streamResponse.Content.ReadAsStreamAsync(streamCts.Token); + using var reader = new StreamReader(body); + var idLine = await reader.ReadLineAsync(streamCts.Token); + var eventLine = await reader.ReadLineAsync(streamCts.Token); + var dataLine = await reader.ReadLineAsync(streamCts.Token); + Assert.StartsWith("id: ", idLine); + Assert.Equal("event: change", eventLine); + Assert.StartsWith("data: ", dataLine); + using var message = JsonDocument.Parse(dataLine!["data: ".Length..]); + Assert.Equal("Change", message.RootElement.GetProperty("Kind").GetString()); + Assert.Equal(ShortGuid.Encode(feedEntityId), + message.RootElement.GetProperty("EntityId").GetString()); + } + + var deniedClientId = $"management-feed-denied-{Guid.NewGuid():N}"; + await CreateClientCredentialsClientAsync( + deniedClientId, + serviceAccount.Id, + [ModgudManagementApi.Scope], + appIds: [otherAppId]); + var deniedToken = await IssueClientCredentialsTokenAsync( + deniedClientId, ModgudManagementApi.Scope, ModgudManagementApi.Audience); + + using var denied = await SendManagementGetAsync( + deniedToken, + $"/api/app/{ShortGuid.Encode(appId)}/change-feed/snapshot"); + var deniedBody = await denied.Content.ReadAsStringAsync(ct); + Assert.Equal(HttpStatusCode.Forbidden, denied.StatusCode); + Assert.Contains("Management.ClientAppMismatch", deniedBody); + } + [Fact] public async Task Realm_seeder_reconciles_a_legacy_management_scope_collision() { @@ -281,11 +402,20 @@ await Factory.CreateTestGroupAsync( $"ManagementPositionReaders_{Guid.NewGuid():N}", [principalId], [role.Id]); } + private async Task GrantAppScopeReadAsync(Guid principalId) + { + var role = await Factory.CreateTestRoleAsync( + $"ManagementAppScopeReader_{Guid.NewGuid():N}", [("app-scope", "read")]); + await Factory.CreateTestGroupAsync( + $"ManagementAppScopeReaders_{Guid.NewGuid():N}", [principalId], [role.Id]); + } + private async Task CreateClientCredentialsClientAsync( string clientId, string serviceAccountId, List scopes, - AccessTokenType accessTokenType = AccessTokenType.Jwt) + AccessTokenType accessTokenType = AccessTokenType.Jwt, + List? appIds = null) { using var scope = Factory.Services.CreateScope(); var oauth = scope.ServiceProvider.GetRequiredService(); @@ -302,7 +432,7 @@ private async Task CreateClientCredentialsClientAsync( AllowedGrantTypes = ["client_credentials"], RequireConsent = false, AccessTokenType = accessTokenType, - AppIds = [], + AppIds = appIds?.Select(ShortGuid.Encode).ToList() ?? [], LinkedServiceAccountId = serviceAccountId, }, TestContext.Current.CancellationToken); if (result.IsError) @@ -442,9 +572,11 @@ private async Task CreateScopeAsync(string name, string audience) $"CreateScopeAsync failed: {string.Join(", ", result.Errors.Select(error => $"{error.Code}: {error.Description}"))}"); } - private async Task SendManagementGetAsync(string token) + private async Task SendManagementGetAsync( + string token, + string path = "/api/position") { - var request = new HttpRequestMessage(HttpMethod.Get, "/api/position"); + var request = new HttpRequestMessage(HttpMethod.Get, path); request.Headers.Authorization = new AuthenticationHeaderValue("Bearer", token); using var client = Factory.CreateClient(); return await client.SendAsync(request, TestContext.Current.CancellationToken); diff --git a/src/dotnet/Modgud.Api.Tests/Authorization/ServiceAccountCreateCompletenessTests.cs b/src/dotnet/Modgud.Api.Tests/Authorization/ServiceAccountCreateCompletenessTests.cs index 432ff503..babaed37 100644 --- a/src/dotnet/Modgud.Api.Tests/Authorization/ServiceAccountCreateCompletenessTests.cs +++ b/src/dotnet/Modgud.Api.Tests/Authorization/ServiceAccountCreateCompletenessTests.cs @@ -6,6 +6,7 @@ using Modgud.Api.Tests.Infrastructure; using Modgud.Application.DTOs.ServiceAccount; using Modgud.Application.DTOs.OAuth; +using Modgud.Authorization.Events; using Modgud.Authorization.Principals; using Modgud.Domain.OAuth.Applications; @@ -54,6 +55,10 @@ public async Task Create_can_commit_status_and_initial_credential_together() Assert.False(serviceAccount!.IsActive); Assert.Contains(OAuthApplicationPropertyKeys.Enabled, credential.Properties.Keys); Assert.Equal("900", credential.Settings[OAuthApplicationSettingKeys.AccessTokenLifetime]); + var stream = await query.Events.FetchStreamAsync(serviceAccountId, token: ct); + var createdEvent = Assert.IsType(Assert.Single(stream).Data); + Assert.Equal(accountName, createdEvent.AccountName); + Assert.False(createdEvent.IsActive); } [Fact] @@ -116,5 +121,52 @@ public async Task OAuth_client_inline_service_account_preserves_create_status() var query = scope.ServiceProvider.GetRequiredService(); var serviceAccount = await query.LoadAsync(serviceAccountId, ct); Assert.False(serviceAccount!.IsActive); + var stream = await query.Events.FetchStreamAsync(serviceAccountId, token: ct); + Assert.IsType(Assert.Single(stream).Data); + } + + [Fact] + public async Task First_update_of_legacy_account_seeds_old_snapshot_then_records_update() + { + var ct = TestContext.Current.CancellationToken; + var legacy = new ServiceAccount + { + Id = Guid.CreateVersion7(), + AccountName = $"legacy-{Guid.NewGuid():N}"[..31], + Purpose = "Before event sourcing", + IsActive = true, + }; + await using (var arrange = GetTenantedDocumentSession()) + { + arrange.Store(legacy); + await arrange.SaveChangesAsync(ct); + } + + var response = await Client.PutAsJsonAsync( + $"/api/service-account/{ShortGuid.Encode(legacy.Id)}", + new { Purpose = "After event sourcing", IsActive = false }, + ct); + response.EnsureSuccessStatusCode(); + + await using var query = GetTenantedSession(); + var stream = await query.Events.FetchStreamAsync(legacy.Id, token: ct); + Assert.Collection( + stream, + item => + { + var created = Assert.IsType(item.Data); + Assert.Equal("Before event sourcing", created.Purpose); + Assert.True(created.IsActive); + }, + item => + { + var updated = Assert.IsType(item.Data); + Assert.Equal("After event sourcing", updated.Purpose); + Assert.False(updated.IsActive); + }); + + var persisted = await query.LoadAsync(legacy.Id, ct); + Assert.Equal("After event sourcing", persisted!.Purpose); + Assert.False(persisted.IsActive); } } diff --git a/src/dotnet/Modgud.Api.Tests/ChangeFeed/AppChangeFeedQueryServiceTests.cs b/src/dotnet/Modgud.Api.Tests/ChangeFeed/AppChangeFeedQueryServiceTests.cs new file mode 100644 index 00000000..d4736c52 --- /dev/null +++ b/src/dotnet/Modgud.Api.Tests/ChangeFeed/AppChangeFeedQueryServiceTests.cs @@ -0,0 +1,132 @@ +using Marten; +using Modgud.Api.Features.ChangeFeed; +using Modgud.Api.Tests.Infrastructure; +using Modgud.Authorization.Apps; +using Modgud.Authorization.Events; +using Modgud.Infrastructure.ChangeFeed; + +namespace Modgud.Api.Tests.ChangeFeed; + +[Collection(IntegrationTestCollection.Name)] +public class AppChangeFeedQueryServiceTests(SharedPostgresFixture fixture) + : IntegrationTestBase(fixture) +{ + [Fact] + public async Task Snapshot_and_incremental_read_share_one_opaque_checkpoint_contract() + { + var ct = TestContext.Current.CancellationToken; + var appId = Guid.CreateVersion7(); + var principalId = Guid.CreateVersion7(); + await using (var arrange = GetTenantedDocumentSession()) + { + arrange.Events.StartStream(appId, new AppCreatedEvent( + appId, "feed-query-test", "Feed query test", null, [], false)); + arrange.Store(new AppChangeFeedState + { + Id = appId, + Enabled = true, + Generation = 1, + ScopeVersion = "v1-test", + LastProcessedSequence = 20, + }); + arrange.Store(new AppChangeFeedEntityState + { + Id = Guid.CreateVersion7(), + AppId = appId, + EntityKind = "principal", + EntityId = principalId, + Fingerprint = "test", + PayloadJson = "{\"DisplayName\":\"Current\"}", + }); + arrange.Store(new AppChangeFeedEntry + { + Id = Guid.CreateVersion7(), + AppId = appId, + Generation = 1, + SourceSequence = 11, + Ordinal = 0, + ScopeVersion = "v1-test", + RecordedAt = DateTimeOffset.UtcNow, + OriginatedAt = DateTimeOffset.UtcNow, + ChangeKind = "Upsert", + EntityKind = "principal", + EntityId = principalId, + PayloadJson = "{\"DisplayName\":\"Current\"}", + }); + await arrange.SaveChangesAsync(ct); + } + + await using var query = GetTenantedSession(); + var service = new AppChangeFeedQueryService(query); + var snapshotResult = await service.SnapshotAsync(appId, ct); + + Assert.True(snapshotResult.IsSuccess); + var snapshot = snapshotResult.Value!; + Assert.Equal(1, snapshot.ContractVersion); + Assert.Equal("feed-query-test", snapshot.AppSlug); + Assert.Equal("v1-test", snapshot.ScopeVersion); + var entity = Assert.Single(snapshot.Entities); + Assert.Equal("principal", entity.EntityKind); + Assert.Equal("Current", entity.Payload.GetProperty("DisplayName").GetString()); + Assert.True(AppChangeFeedCursor.TryDecode(snapshot.Cursor, out var checkpoint)); + Assert.Equal(20, checkpoint.Sequence); + Assert.Equal(int.MaxValue, checkpoint.Ordinal); + + var readResult = await service.ReadAsync( + appId, + AppChangeFeedCursor.Encode(appId, 1, 10, int.MaxValue), + limit: 100, + ct); + Assert.True(readResult.IsSuccess); + Assert.Equal(1, readResult.Value!.ContractVersion); + Assert.Equal("v1-test", readResult.Value.ScopeVersion); + Assert.Collection( + readResult.Value.Messages, + change => + { + Assert.Equal("Change", change.Kind); + Assert.Equal("Upsert", change.ChangeKind); + Assert.Equal("principal", change.EntityKind); + }, + finalCheckpoint => + { + Assert.Equal("Checkpoint", finalCheckpoint.Kind); + Assert.Equal(snapshot.Cursor, finalCheckpoint.Cursor); + }); + Assert.False(readResult.Value.HasMore); + } + + [Fact] + public async Task Read_explicitly_distinguishes_scope_reset_from_expired_retention() + { + var ct = TestContext.Current.CancellationToken; + var appId = Guid.CreateVersion7(); + await using (var arrange = GetTenantedDocumentSession()) + { + arrange.Store(new AppChangeFeedState + { + Id = appId, + Enabled = true, + Generation = 4, + ScopeVersion = "v4-test", + LastProcessedSequence = 100, + RetentionFloorSequence = 50, + RetentionFloorOrdinal = 2, + }); + await arrange.SaveChangesAsync(ct); + } + + await using var query = GetTenantedSession(); + var service = new AppChangeFeedQueryService(query); + + var wrongGeneration = await service.ReadAsync( + appId, AppChangeFeedCursor.Encode(appId, 3, 80, 0), 100, ct); + Assert.False(wrongGeneration.IsSuccess); + Assert.Equal("ScopeChanged", wrongGeneration.Error!.Code); + + var expired = await service.ReadAsync( + appId, AppChangeFeedCursor.Encode(appId, 4, 50, 2), 100, ct); + Assert.False(expired.IsSuccess); + Assert.Equal("CursorTooOld", expired.Error!.Code); + } +} diff --git a/src/dotnet/Modgud.Api.Tests/ChangeFeed/AppChangeFeedSubscriptionTests.cs b/src/dotnet/Modgud.Api.Tests/ChangeFeed/AppChangeFeedSubscriptionTests.cs new file mode 100644 index 00000000..fd15d9ba --- /dev/null +++ b/src/dotnet/Modgud.Api.Tests/ChangeFeed/AppChangeFeedSubscriptionTests.cs @@ -0,0 +1,139 @@ +using System.Net.Http.Json; +using System.Text.Json; +using BuildingBlocks.Helper; +using Marten; +using Modgud.Api.Features.Admin.Apps; +using Modgud.Api.Features.ChangeFeed; +using Modgud.Api.Tests.Infrastructure; +using Modgud.Application.DTOs.Applications; +using Modgud.Authorization.Apps; +using Modgud.Authorization.Events; +using Modgud.Authorization.Principals; +using Modgud.Infrastructure.ChangeFeed; + +namespace Modgud.Api.Tests.ChangeFeed; + +[Collection(IntegrationTestCollection.Name)] +public class AppChangeFeedSubscriptionTests(SharedPostgresFixture fixture) + : IntegrationTestBase(fixture) +{ + [Fact] + public async Task Enable_seeds_snapshot_then_emits_net_change_and_scope_reset() + { + var ct = TestContext.Current.CancellationToken; + var suffix = Guid.NewGuid().ToString("N")[..10]; + var appId = Guid.CreateVersion7(); + var appSlug = $"feed-{suffix}"; + var serviceAccountId = Guid.CreateVersion7(); + var groupId = Guid.CreateVersion7(); + + await using (var arrange = GetTenantedDocumentSession()) + { + arrange.Events.StartStream(appId, new AppCreatedEvent( + appId, appSlug, "Feed subscription test", null, [], false)); + arrange.Events.StartStream(serviceAccountId, + new ServiceAccountCreatedEvent( + serviceAccountId, $"feed-agent-{suffix}", "Before", true)); + arrange.Events.StartStream(groupId, new GroupCreatedEvent( + groupId, + $"Feed group {suffix}", + null, + [serviceAccountId], + [], + BoundTo: [appSlug])); + await arrange.SaveChangesAsync(ct); + } + + // SubscribeFromPresent anchors the short-lived queue without replaying + // the permanent history. Establish that anchor before enablement. + await Factory.WaitForProjectionsAsync(); + + var settings = new ApplicationSettingsDto + { + ChangeFeed = new ApplicationChangeFeedDto + { + Enabled = true, + MinimumRetentionAgeDays = 7, + MinimumEventCount = 1_000, + }, + }; + var enable = await Client.PutAsJsonAsync( + $"/api/app/{ShortGuid.Encode(appId)}", + new UpdateAppDto("Feed subscription test", null, [], settings), + JsonOptions, + ct); + enable.EnsureSuccessStatusCode(); + await Factory.WaitForProjectionsAsync(); + + await using (var query = GetTenantedSession()) + { + var state = await query.LoadAsync(appId, ct); + Assert.NotNull(state); + Assert.True(state!.Enabled); + Assert.Equal(1, state.Generation); + + var projected = await query.Query() + .Where(x => x.AppId == appId) + .ToListAsync(ct); + Assert.Contains(projected, x => x.EntityKind == "principal" && x.EntityId == groupId); + Assert.Contains(projected, x => x.EntityKind == "principal" && x.EntityId == serviceAccountId); + + var snapshot = await new AppChangeFeedQueryService(query).SnapshotAsync(appId, ct); + Assert.True(snapshot.IsSuccess); + Assert.Contains(snapshot.Value!.Entities, + x => x.EntityKind == "principal" && x.EntityId == ShortGuid.Encode(groupId)); + Assert.Contains(snapshot.Value.Entities, + x => x.EntityKind == "principal" && x.EntityId == ShortGuid.Encode(serviceAccountId)); + } + + var update = await Client.PutAsJsonAsync( + $"/api/service-account/{ShortGuid.Encode(serviceAccountId)}", + new { Purpose = "After" }, + JsonOptions, + ct); + update.EnsureSuccessStatusCode(); + await Factory.WaitForProjectionsAsync(); + + await using (var query = GetTenantedSession()) + { + var upsert = await query.Query() + .Where(x => x.AppId == appId + && x.Generation == 1 + && x.ChangeKind == "Upsert" + && x.EntityId == serviceAccountId) + .SingleAsync(ct); + Assert.Equal("After", JsonDocument.Parse(upsert.PayloadJson!) + .RootElement.GetProperty("Purpose").GetString()); + } + + await using (var mutate = GetTenantedDocumentSession()) + { + mutate.Events.Append(groupId, new GroupUpdatedEvent( + groupId, + $"Feed group {suffix}", + null, + [serviceAccountId], + [], + BoundTo: [])); + await mutate.SaveChangesAsync(ct); + } + await Factory.WaitForProjectionsAsync(); + + await using (var query = GetTenantedSession()) + { + var state = await query.LoadAsync(appId, ct); + Assert.Equal(2, state!.Generation); + var remaining = await query.Query() + .Where(x => x.AppId == appId) + .ToListAsync(ct); + Assert.DoesNotContain(remaining, + x => x.EntityKind == "principal" && x.EntityId == groupId); + Assert.DoesNotContain(remaining, + x => x.EntityKind == "principal" && x.EntityId == serviceAccountId); + Assert.True(await query.Query() + .AnyAsync(x => x.AppId == appId + && x.Generation == 2 + && x.ChangeKind == "ScopeChanged", ct)); + } + } +} diff --git a/src/dotnet/Modgud.Api.Tests/Infrastructure/ModgudWebApplicationFactory.cs b/src/dotnet/Modgud.Api.Tests/Infrastructure/ModgudWebApplicationFactory.cs index 4766c48b..1d57ad22 100644 --- a/src/dotnet/Modgud.Api.Tests/Infrastructure/ModgudWebApplicationFactory.cs +++ b/src/dotnet/Modgud.Api.Tests/Infrastructure/ModgudWebApplicationFactory.cs @@ -620,8 +620,8 @@ await rebuildDaemon.RebuildProjectionAsync( } /// - /// Rebuilds both event-sourced Principal subtypes with subtype-scoped cleanup, - /// preserving directly stored ServiceAccount rows in mt_doc_principal. + /// Rebuilds every event-sourced Principal subtype with subtype-scoped cleanup, + /// preserving legacy document-only ServiceAccount rows in mt_doc_principal. /// public async Task RebuildPrincipalProjectionsAsync( string tenantId = TenantConstants.SystemTenantId, diff --git a/src/dotnet/Modgud.Api/Features/ChangeFeed/AppChangeFeedEndpoints.cs b/src/dotnet/Modgud.Api/Features/ChangeFeed/AppChangeFeedEndpoints.cs new file mode 100644 index 00000000..2273a51f --- /dev/null +++ b/src/dotnet/Modgud.Api/Features/ChangeFeed/AppChangeFeedEndpoints.cs @@ -0,0 +1,249 @@ +using System.Text.Json; +using BuildingBlocks.Helper; +using Microsoft.AspNetCore.Http.Features; +using Microsoft.Extensions.Options; +using Modgud.Api.Features.Management; +using Modgud.Application.DTOs.ChangeFeed; +using HttpJsonOptions = Microsoft.AspNetCore.Http.Json.JsonOptions; + +namespace Modgud.Api.Features.ChangeFeed; + +public static class AppChangeFeedEndpoints +{ + public static WebApplication MapAppChangeFeedEndpoints(this WebApplication app, string path) + { + app.MapGet($"{path}/app/{{id}}/change-feed/snapshot", async ( + ShortGuid id, + AppChangeFeedQueryService feed, + CancellationToken cancellationToken) => + { + var result = await feed.SnapshotAsync(id.Guid, cancellationToken); + return ToResult(result); + }) + .WithTags("Apps") + .WithName("V2_AppChangeFeed_Snapshot") + .RequiresManagementPermission("app-scope:read", clientAppRouteParameter: "id"); + + app.MapGet($"{path}/app/{{id}}/change-feed", async ( + ShortGuid id, + string cursor, + int? limit, + AppChangeFeedQueryService feed, + CancellationToken cancellationToken) => + { + var result = await feed.ReadAsync(id.Guid, cursor, limit ?? 100, cancellationToken); + return ToResult(result); + }) + .WithTags("Apps") + .WithName("V2_AppChangeFeed_Read") + .RequiresManagementPermission("app-scope:read", clientAppRouteParameter: "id"); + + // A long-running integration stream needs explicit token expiry and + // refresh semantics; first-party admin cookies remain valid for the + // finite snapshot and polling requests only. + app.MapGet($"{path}/app/{{id}}/change-feed/stream", StreamAsync) + .WithTags("Apps") + .WithName("V2_AppChangeFeed_Stream") + .RequiresManagementPermission( + "app-scope:read", + clientAppRouteParameter: "id", + bearerOnly: true); + + return app; + } + + private static async Task StreamAsync( + HttpContext http, + ShortGuid id, + string? cursor, + int? batchSize, + IServiceScopeFactory scopeFactory, + IOptions jsonOptions, + CancellationToken cancellationToken) + { + cursor = string.IsNullOrWhiteSpace(cursor) + ? http.Request.Headers["Last-Event-ID"].FirstOrDefault() + : cursor; + if (string.IsNullOrWhiteSpace(cursor)) + { + await Results.Problem( + statusCode: StatusCodes.Status400BadRequest, + title: "CursorRequired", + detail: "Take a full snapshot and pass its cursor as ?cursor= or Last-Event-ID.", + extensions: new Dictionary { ["code"] = "CursorRequired" }) + .ExecuteAsync(http); + return; + } + + var currentCursor = cursor; + var currentScopeVersion = string.Empty; + var limit = Math.Clamp(batchSize ?? 100, 1, 500); + var responseStarted = false; + var nextHeartbeatAt = DateTimeOffset.UtcNow.AddSeconds(15); + + try + { + while (!cancellationToken.IsCancellationRequested) + { + ManagementBearerAuthorizationError? denied; + AppChangeFeedQueryResult? read = null; + // A streaming request outlives a normal request scope. Resolve a + // fresh tenant-scoped Marten session for every cycle so neither + // authorization nor feed state can be served from an identity map. + await using (var cycle = scopeFactory.CreateAsyncScope()) + { + denied = await cycle.ServiceProvider + .GetRequiredService() + .AuthorizeAsync(http.User, id.Guid, "app-scope:read", cancellationToken); + if (denied is null) + { + read = await cycle.ServiceProvider + .GetRequiredService() + .ReadAsync(id.Guid, currentCursor, limit, cancellationToken); + } + } + + if (denied is not null) + { + if (!responseStarted) + { + await AuthorizationProblem(denied).ExecuteAsync(http); + } + else + { + await WriteSseMessageAsync( + http.Response, + new AppChangeFeedMessageDto + { + ContractVersion = AppChangeFeedContract.Version, + Kind = "FeedEnded", + Cursor = currentCursor, + ScopeVersion = currentScopeVersion, + Reason = denied.Code, + }, + jsonOptions.Value.SerializerOptions, + cancellationToken); + } + return; + } + + if (!read!.IsSuccess) + { + if (!responseStarted) + { + await ToResult(read).ExecuteAsync(http); + } + else + { + await WriteSseMessageAsync( + http.Response, + new AppChangeFeedMessageDto + { + ContractVersion = AppChangeFeedContract.Version, + Kind = "ResetRequired", + Cursor = currentCursor, + ScopeVersion = currentScopeVersion, + Reason = read.Error!.Code, + }, + jsonOptions.Value.SerializerOptions, + cancellationToken); + } + return; + } + + if (!responseStarted) + { + StartSseResponse(http); + await http.Response.StartAsync(cancellationToken); + responseStarted = true; + } + + currentScopeVersion = read.Value!.ScopeVersion; + foreach (var message in read.Value!.Messages) + { + currentCursor = message.Cursor; + if (!string.IsNullOrWhiteSpace(message.ScopeVersion)) + currentScopeVersion = message.ScopeVersion; + await WriteSseMessageAsync( + http.Response, + message, + jsonOptions.Value.SerializerOptions, + cancellationToken); + nextHeartbeatAt = DateTimeOffset.UtcNow.AddSeconds(15); + } + + if (read.Value.FeedEnded) return; + if (read.Value.HasMore) continue; + + if (read.Value.Messages.Count == 0) + { + await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken); + if (DateTimeOffset.UtcNow >= nextHeartbeatAt) + { + await http.Response.WriteAsync(": keep-alive\n\n", cancellationToken); + await http.Response.Body.FlushAsync(cancellationToken); + nextHeartbeatAt = DateTimeOffset.UtcNow.AddSeconds(15); + } + } + } + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + // Normal SSE disconnect or server shutdown. + } + } + + private static void StartSseResponse(HttpContext http) + { + http.Response.StatusCode = StatusCodes.Status200OK; + http.Response.ContentType = "text/event-stream; charset=utf-8"; + http.Response.Headers.CacheControl = "no-cache, no-transform"; + http.Response.Headers["X-Accel-Buffering"] = "no"; + http.Features.Get()?.DisableBuffering(); + } + + private static async Task WriteSseMessageAsync( + HttpResponse response, + AppChangeFeedMessageDto message, + JsonSerializerOptions jsonOptions, + CancellationToken cancellationToken) + { + var eventName = message.Kind switch + { + "Change" => "change", + "Checkpoint" => "checkpoint", + "ResetRequired" => "reset-required", + "FeedEnded" => "feed-ended", + _ => "message", + }; + var json = JsonSerializer.Serialize(message, jsonOptions); + await response.WriteAsync($"id: {message.Cursor}\n", cancellationToken); + await response.WriteAsync($"event: {eventName}\n", cancellationToken); + await response.WriteAsync($"data: {json}\n\n", cancellationToken); + await response.Body.FlushAsync(cancellationToken); + } + + private static IResult AuthorizationProblem(ManagementBearerAuthorizationError error) + { + var unauthorized = error.Code is "invalid_client" or "invalid_subject" + or "inactive_subject" or "token_expired"; + return Results.Problem( + statusCode: unauthorized + ? StatusCodes.Status401Unauthorized + : StatusCodes.Status403Forbidden, + title: unauthorized ? "Unauthorized" : "Forbidden", + detail: error.Detail, + extensions: new Dictionary { ["code"] = error.Code }); + } + + private static IResult ToResult(AppChangeFeedQueryResult result) + { + if (result.IsSuccess) return Results.Ok(result.Value); + var error = result.Error!; + return Results.Problem( + statusCode: error.StatusCode, + title: error.Code, + detail: error.Detail, + extensions: new Dictionary { ["code"] = error.Code }); + } +} diff --git a/src/dotnet/Modgud.Api/Features/ChangeFeed/AppChangeFeedQueryService.cs b/src/dotnet/Modgud.Api/Features/ChangeFeed/AppChangeFeedQueryService.cs new file mode 100644 index 00000000..649568f6 --- /dev/null +++ b/src/dotnet/Modgud.Api/Features/ChangeFeed/AppChangeFeedQueryService.cs @@ -0,0 +1,213 @@ +using System.Buffers.Binary; +using System.Text.Json; +using BuildingBlocks.Helper; +using Marten; +using Microsoft.AspNetCore.WebUtilities; +using Modgud.Application.DTOs.ChangeFeed; +using Modgud.Authorization.Apps; +using Modgud.Infrastructure.ChangeFeed; + +namespace Modgud.Api.Features.ChangeFeed; + +public sealed class AppChangeFeedQueryService(IQuerySession session) +{ + public async Task> SnapshotAsync( + Guid appId, + CancellationToken cancellationToken) + { + var state = await session.LoadAsync(appId, cancellationToken); + if (state is null) + return AppChangeFeedQueryResult.Fail( + "FeedInitializing", "The change-feed projection has not initialized this Application yet.", 409); + if (!state.Enabled) + return AppChangeFeedQueryResult.Fail( + "FeedDisabled", "The change feed is disabled for this Application.", 409); + + var app = await session.LoadAsync(appId, cancellationToken); + if (app is null || app.IsDeleted) + return AppChangeFeedQueryResult.Fail( + "ApplicationNotFound", "Application not found.", 404); + + var rows = await session.Query() + .Where(x => x.AppId == appId) + .OrderBy(x => x.EntityKind) + .ThenBy(x => x.EntityId) + .ToListAsync(cancellationToken); + var entities = rows.Select(row => new AppScopedEntityDto( + row.EntityKind, + ShortGuid.Encode(row.EntityId), + EntityVersion: 1, + ParsePayload(row.PayloadJson))) + .ToList(); + + return AppChangeFeedQueryResult.Ok(new AppChangeFeedSnapshotDto( + AppChangeFeedContract.Version, + ShortGuid.Encode(appId), + app.Slug, + state.ScopeVersion, + AppChangeFeedCursor.EncodeCheckpoint(state), + entities)); + } + + public async Task> ReadAsync( + Guid appId, + string cursorText, + int limit, + CancellationToken cancellationToken) + { + limit = Math.Clamp(limit, 1, 500); + var state = await session.LoadAsync(appId, cancellationToken); + if (state is null) + return AppChangeFeedQueryResult.Fail( + "FeedInitializing", "The change-feed projection has not initialized this Application yet.", 409); + + if (!AppChangeFeedCursor.TryDecode(cursorText, out var cursor)) + return AppChangeFeedQueryResult.Fail( + "InvalidCursor", "The cursor is malformed or uses an unsupported version.", 400); + if (cursor.AppId != appId) + return AppChangeFeedQueryResult.Fail( + "InvalidCursor", "The cursor belongs to a different Application.", 400); + if (cursor.Generation != state.Generation) + return AppChangeFeedQueryResult.Fail( + "ScopeChanged", "The Application scope changed; take a new full snapshot.", 409); + if (Compare(cursor.Sequence, cursor.Ordinal, + state.RetentionFloorSequence, state.RetentionFloorOrdinal) <= 0 + && (state.RetentionFloorSequence > 0 || state.RetentionFloorOrdinal >= 0)) + { + return AppChangeFeedQueryResult.Fail( + "CursorTooOld", "The cursor is outside the retained resume window; take a new full snapshot.", 409); + } + if (cursor.Sequence > state.LastProcessedSequence) + return AppChangeFeedQueryResult.Fail( + "InvalidCursor", "The cursor points beyond the feed checkpoint.", 400); + + var rows = (await session.Query() + .Where(x => x.AppId == appId + && x.Generation == state.Generation + && (x.SourceSequence > cursor.Sequence + || (x.SourceSequence == cursor.Sequence && x.Ordinal > cursor.Ordinal))) + .OrderBy(x => x.SourceSequence) + .ThenBy(x => x.Ordinal) + .Take(limit + 1) + .ToListAsync(cancellationToken)).ToList(); + + var hasMore = rows.Count > limit; + if (hasMore) rows.RemoveAt(rows.Count - 1); + var messages = rows.Select(MapEntry).ToList(); + + if (!hasMore) + { + var lastSequence = rows.Count == 0 ? cursor.Sequence : rows[^1].SourceSequence; + var lastOrdinal = rows.Count == 0 ? cursor.Ordinal : rows[^1].Ordinal; + if (Compare(lastSequence, lastOrdinal, state.LastProcessedSequence, int.MaxValue) < 0) + { + messages.Add(new AppChangeFeedMessageDto + { + ContractVersion = AppChangeFeedContract.Version, + Kind = "Checkpoint", + Cursor = AppChangeFeedCursor.EncodeCheckpoint(state), + ScopeVersion = state.ScopeVersion, + }); + } + } + + if (!state.Enabled && messages.Count == 0) + return AppChangeFeedQueryResult.Fail( + "FeedDisabled", "The change feed is disabled for this Application.", 409); + + return AppChangeFeedQueryResult.Ok(new AppChangeFeedReadDto( + AppChangeFeedContract.Version, + state.ScopeVersion, + messages, + hasMore, + FeedEnded: !state.Enabled && !hasMore)); + } + + private static AppChangeFeedMessageDto MapEntry(AppChangeFeedEntry entry) => new() + { + ContractVersion = AppChangeFeedContract.Version, + Kind = entry.ChangeKind switch + { + AppChangeKinds.ScopeChanged => "ResetRequired", + AppChangeKinds.FeedDisabled => "FeedEnded", + _ => "Change", + }, + Cursor = AppChangeFeedCursor.Encode( + entry.AppId, entry.Generation, entry.SourceSequence, entry.Ordinal), + ScopeVersion = entry.ScopeVersion, + ChangeKind = entry.ChangeKind, + EntityKind = entry.EntityKind, + EntityId = entry.EntityId is { } id ? ShortGuid.Encode(id) : null, + EntityVersion = entry.EntityId.HasValue ? 1 : null, + Payload = entry.PayloadJson is null ? null : ParsePayload(entry.PayloadJson), + SourceEventId = entry.SourceEventId, + OriginatedAt = entry.OriginatedAt, + Reason = entry.Reason, + }; + + private static JsonElement ParsePayload(string json) => + JsonDocument.Parse(json).RootElement.Clone(); + + private static int Compare(long leftSequence, int leftOrdinal, long rightSequence, int rightOrdinal) + { + var sequence = leftSequence.CompareTo(rightSequence); + return sequence != 0 ? sequence : leftOrdinal.CompareTo(rightOrdinal); + } +} + +public sealed record AppChangeFeedQueryResult(T? Value, AppChangeFeedQueryError? Error) +{ + public bool IsSuccess => Error is null; + public static AppChangeFeedQueryResult Ok(T value) => new(value, null); + public static AppChangeFeedQueryResult Fail(string code, string detail, int statusCode) => + new(default, new AppChangeFeedQueryError(code, detail, statusCode)); +} + +public sealed record AppChangeFeedQueryError(string Code, string Detail, int StatusCode); + +internal readonly record struct AppChangeFeedCursorValue( + Guid AppId, + int Generation, + long Sequence, + int Ordinal); + +internal static class AppChangeFeedCursor +{ + private const byte Version = 1; + private const int Size = 1 + 16 + 4 + 8 + 4; + + public static string EncodeCheckpoint(AppChangeFeedState state) => + Encode(state.Id, state.Generation, state.LastProcessedSequence, int.MaxValue); + + public static string Encode(Guid appId, int generation, long sequence, int ordinal) + { + Span bytes = stackalloc byte[Size]; + bytes[0] = Version; + appId.TryWriteBytes(bytes[1..17]); + BinaryPrimitives.WriteInt32BigEndian(bytes[17..21], generation); + BinaryPrimitives.WriteInt64BigEndian(bytes[21..29], sequence); + BinaryPrimitives.WriteInt32BigEndian(bytes[29..33], ordinal); + return WebEncoders.Base64UrlEncode(bytes); + } + + public static bool TryDecode(string? text, out AppChangeFeedCursorValue value) + { + value = default; + if (string.IsNullOrWhiteSpace(text)) return false; + try + { + var bytes = WebEncoders.Base64UrlDecode(text); + if (bytes.Length != Size || bytes[0] != Version) return false; + value = new AppChangeFeedCursorValue( + new Guid(bytes.AsSpan(1, 16)), + BinaryPrimitives.ReadInt32BigEndian(bytes.AsSpan(17, 4)), + BinaryPrimitives.ReadInt64BigEndian(bytes.AsSpan(21, 8)), + BinaryPrimitives.ReadInt32BigEndian(bytes.AsSpan(29, 4))); + return value.Generation > 0 && value.Sequence >= 0 && value.Ordinal >= -1; + } + catch (FormatException) + { + return false; + } + } +} diff --git a/src/dotnet/Modgud.Api/Features/ChangeFeed/AppChangeFeedSubscription.cs b/src/dotnet/Modgud.Api/Features/ChangeFeed/AppChangeFeedSubscription.cs new file mode 100644 index 00000000..c4344071 --- /dev/null +++ b/src/dotnet/Modgud.Api/Features/ChangeFeed/AppChangeFeedSubscription.cs @@ -0,0 +1,587 @@ +using System.Security.Cryptography; +using System.Text; +using System.Text.Json; +using System.Text.Json.Serialization; +using BuildingBlocks.Helper; +using JasperFx.Events; +using JasperFx.Events.Daemon; +using JasperFx.Events.Projections; +using Marten; +using Marten.Subscriptions; +using Modgud.Authentication.Events; +using Modgud.Authentication.Domain.ExternalAuth.Events; +using Modgud.Authorization.Apps; +using Modgud.Authorization.Principals; +using Modgud.Domain.Applications; +using Modgud.Domain.PositionTerminals; +using Modgud.Infrastructure.ChangeFeed; + +namespace Modgud.Api.Features.ChangeFeed; + +/// +/// High-water-anchored projection into the short-lived per-App resume queue. +/// It publishes net changes to the public integration model, never raw domain +/// event payloads and never a second permanent copy of the event store. +/// +public sealed class AppChangeFeedSubscription : SubscriptionBase +{ + private static readonly JsonSerializerOptions PayloadJson = new() + { + PropertyNamingPolicy = null, + DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull, + Converters = { new JsonStringEnumConverter() }, + }; + + public AppChangeFeedSubscription() + { + Name = "AppChangeFeed"; + Options.BatchSize = 250; + Options.MaximumHopperSize = 5_000; + // This queue is an integration resume window, not a replay of the + // permanent event store. Enabling the feed writes a fresh event and + // seeds a full current-state snapshot from that high-water mark. + Options.SubscribeFromPresent(); + } + + public override async Task ProcessEventsAsync( + EventRange page, + ISubscriptionController controller, + IDocumentOperations operations, + CancellationToken cancellationToken) + { + var now = DateTimeOffset.UtcNow; + var states = await operations.Query().ToListAsync(cancellationToken); + var stateByApp = states.ToDictionary(x => x.Id); + var configured = await operations.Query() + .Where(x => x.ChangeFeed != null) + .ToListAsync(cancellationToken); + var settingsByApp = configured.ToDictionary(x => x.Id); + + var configuredInPage = page.Events + .Select(x => x.Data) + .OfType() + .Select(x => x.ApplicationId); + var appIds = states.Where(x => x.Enabled).Select(x => x.Id) + .Concat(configured.Where(x => x.ChangeFeed!.Enabled).Select(x => x.Id)) + .Concat(configuredInPage) + .Distinct() + .OrderBy(x => x) + .ToList(); + + var relevantSources = page.Events.Where(IsPublicStateSource).Take(2).ToList(); + var publicStateMayHaveChanged = relevantSources.Count > 0; + // A subscription page can coalesce several domain events into one net + // public change. Only expose an origin id when attribution is exact. + var relevantSource = relevantSources.Count == 1 ? relevantSources[0] : null; + + foreach (var appId in appIds) + { + settingsByApp.TryGetValue(appId, out var settingsDocument); + var policy = settingsDocument?.ChangeFeed ?? ApplicationChangeFeedSettings.Disabled; + stateByApp.TryGetValue(appId, out var state); + + if (!policy.Enabled) + { + if (state is { Enabled: true }) + { + state.Enabled = false; + state.LastProcessedSequence = page.SequenceCeiling; + StageEntry( + operations, + state, + page.SequenceCeiling, + ordinal: 0, + source: relevantSource, + now, + changeKind: AppChangeKinds.FeedDisabled, + entity: null, + reason: "FeedDisabled"); + operations.Store(state); + } + continue; + } + + var app = await operations.LoadAsync(appId, cancellationToken); + if (app is null || app.IsDeleted) + { + if (state is { Enabled: true }) + { + state.Enabled = false; + state.LastProcessedSequence = page.SequenceCeiling; + StageEntry( + operations, + state, + page.SequenceCeiling, + ordinal: 0, + source: relevantSource, + now, + changeKind: AppChangeKinds.FeedDisabled, + entity: null, + reason: "ApplicationDeleted"); + operations.Store(state); + } + continue; + } + + var retentionDays = (int)policy.MinimumRetentionAge.TotalDays; + if (state is null || !state.Enabled) + { + state ??= new AppChangeFeedState { Id = appId }; + state.Enabled = true; + state.Generation = Math.Max(1, state.Generation + 1); + state.MinimumRetentionAgeDays = retentionDays; + state.MinimumEventCount = policy.MinimumEventCount; + state.LastProcessedSequence = page.SequenceCeiling; + + var snapshot = await BuildSnapshotAsync(operations, app, cancellationToken); + state.ScopeVersion = snapshot.ScopeVersion; + await ReplaceEntityStateAsync(operations, appId, snapshot.Entities, cancellationToken); + StageEntry( + operations, + state, + page.SequenceCeiling, + ordinal: 0, + source: relevantSource, + now, + changeKind: AppChangeKinds.ScopeChanged, + entity: null, + reason: "FeedEnabled"); + operations.Store(state); + continue; + } + + state.MinimumRetentionAgeDays = retentionDays; + state.MinimumEventCount = policy.MinimumEventCount; + state.LastProcessedSequence = page.SequenceCeiling; + + if (publicStateMayHaveChanged) + { + var snapshot = await BuildSnapshotAsync(operations, app, cancellationToken); + if (!string.Equals(state.ScopeVersion, snapshot.ScopeVersion, StringComparison.Ordinal)) + { + state.Generation++; + state.ScopeVersion = snapshot.ScopeVersion; + state.RetentionFloorSequence = 0; + state.RetentionFloorOrdinal = -1; + await ReplaceEntityStateAsync(operations, appId, snapshot.Entities, cancellationToken); + StageEntry( + operations, + state, + page.SequenceCeiling, + ordinal: 0, + source: relevantSource, + now, + changeKind: AppChangeKinds.ScopeChanged, + entity: null, + reason: "ScopeDefinitionChanged"); + } + else + { + await StageNetChangesAsync( + operations, + state, + snapshot, + page.SequenceCeiling, + relevantSource, + now, + cancellationToken); + } + } + + if (state.LastCompactedAt is null || state.LastCompactedAt < now.AddHours(-1)) + { + await CompactAsync(operations, state, now, cancellationToken); + state.LastCompactedAt = now; + } + + operations.Store(state); + } + + return NullChangeListener.Instance; + } + + private static bool IsPublicStateSource(IEvent source) + { + var data = source.Data; + var ns = data.GetType().Namespace ?? string.Empty; + if (data is ApplicationChangeFeedConfiguredEvent) return true; + if (ns == "Modgud.Authorization.Events") return true; + if (ns == "Modgud.Domain.Users.Events") return true; + if (ns == "Modgud.Domain.PositionTerminals") return true; + + return data is UserIdentitySetupEvent + or UserUserNameChangedEvent + or UserActivatedEvent + or UserDeactivatedEvent + or UserExternalIdentityLinkedEvent + or UserExternalIdentityUnlinkedEvent; + } + + private static async Task BuildSnapshotAsync( + IDocumentOperations operations, + App app, + CancellationToken cancellationToken) + { + var directory = await operations.Query().ToListAsync(cancellationToken); + var scope = ApplicationScopeResolver.BuildSnapshot(app, directory); + var scopeIds = scope.Principals.Select(x => x.Id).ToHashSet(); + var rootIds = scope.RootGroups.Select(x => x.Id).ToHashSet(); + var positions = scope.Principals.OfType().Select(x => x.Id).ToHashSet(); + + var current = new Dictionary(); + var presence = new Dictionary(); + + foreach (var principal in directory) + { + var key = new EntityKey(AppEntityKinds.Principal, principal.Id); + presence[key] = new EntityPresence(!principal.IsDeleted, null); + } + + foreach (var principal in scope.Principals) + { + var key = new EntityKey(AppEntityKinds.Principal, principal.Id); + var group = principal as Group; + var person = principal as Person; + var serviceAccount = principal as ServiceAccount; + var position = principal as PositionPrincipal; + var payload = new + { + Id = ShortGuid.Encode(principal.Id), + principal.Type, + principal.DisplayName, + principal.IsActive, + IsScopeRoot = rootIds.Contains(principal.Id), + AccountName = person?.AccountName ?? serviceAccount?.AccountName ?? position?.AccountName, + person?.Firstname, + person?.Lastname, + person?.Acronym, + person?.Email, + Name = group?.Name, + Description = group?.Description, + Purpose = serviceAccount?.Purpose ?? position?.Purpose, + MemberIds = group?.MemberIds + .Where(scopeIds.Contains) + .OrderBy(x => x) + .Select(ShortGuid.Encode) + .ToArray(), + HasPermissions = group is null ? (bool?)null : group.RoleIds.Count > 0, + TerminalPolicy = position is null ? null : new + { + position.TerminalPolicy.Enabled, + position.TerminalPolicy.AllowedActivationProofs, + position.TerminalPolicy.AllowedDeviceBindings, + StaffingSessionLifetimeSeconds = (long)position.TerminalPolicy.StaffingSessionLifetime.TotalSeconds, + MaximumStaffingSessionLifetimeSeconds = (long)position.TerminalPolicy.MaximumStaffingSessionLifetime.TotalSeconds, + }, + }; + current[key] = PublicEntity.Create(key, payload); + } + + var terminals = await operations.Query().ToListAsync(cancellationToken); + var scopedTerminalIds = new HashSet(); + foreach (var terminal in terminals) + { + var key = new EntityKey(AppEntityKinds.Terminal, terminal.Id); + var allowed = terminal.EffectiveAllowedPositionIds.Where(positions.Contains).OrderBy(x => x).ToArray(); + var exists = terminal.Status != TerminalEnrollmentStatus.Revoked; + presence[key] = new EntityPresence( + exists, + exists ? null : Serialize(new { terminal.Status, terminal.RevokedAt })); + if (!exists || allowed.Length == 0) continue; + + scopedTerminalIds.Add(terminal.Id); + current[key] = PublicEntity.Create(key, new + { + Id = ShortGuid.Encode(terminal.Id), + AllowedPositionIds = allowed.Select(ShortGuid.Encode).ToArray(), + terminal.DisplayName, + terminal.Location, + terminal.ClientId, + terminal.WebAuthnRpId, + terminal.Binding, + terminal.Status, + ActiveStaffingSessionId = terminal.ActiveStaffingSessionId is { } active + ? ShortGuid.Encode(active) + : null, + terminal.CreatedAt, + terminal.EnrolledAt, + terminal.DisabledAt, + }); + } + + var grants = await operations.Query().ToListAsync(cancellationToken); + foreach (var grant in grants) + { + var key = new EntityKey(AppEntityKinds.PositionGrant, grant.Id); + var exists = grant.Status != PositionGrantStatus.Revoked; + presence[key] = new EntityPresence( + exists, + exists ? null : Serialize(new { grant.Status, grant.RevokedAt })); + if (!exists || !positions.Contains(grant.PositionPrincipalId) || !scopeIds.Contains(grant.UserId)) + continue; + + current[key] = PublicEntity.Create(key, new + { + Id = ShortGuid.Encode(grant.Id), + PositionId = ShortGuid.Encode(grant.PositionPrincipalId), + UserId = ShortGuid.Encode(grant.UserId), + grant.Status, + grant.CreatedAt, + }); + } + + var staffingSessions = await operations.Query().ToListAsync(cancellationToken); + foreach (var staffing in staffingSessions) + { + var key = new EntityKey(AppEntityKinds.StaffingSession, staffing.Id); + var exists = staffing.Status == StaffingSessionStatus.Active; + presence[key] = new EntityPresence( + exists, + exists ? null : Serialize(new { staffing.Status, staffing.EndedAt, staffing.EndReason })); + if (!exists || !positions.Contains(staffing.PositionPrincipalId) + || !scopedTerminalIds.Contains(staffing.TerminalEnrollmentId)) + continue; + + current[key] = PublicEntity.Create(key, new + { + Id = ShortGuid.Encode(staffing.Id), + PositionId = ShortGuid.Encode(staffing.PositionPrincipalId), + TerminalId = ShortGuid.Encode(staffing.TerminalEnrollmentId), + ActivatedByUserId = scopeIds.Contains(staffing.ActivatedByUserId) + ? ShortGuid.Encode(staffing.ActivatedByUserId) + : null, + MethodId = staffing.GetActivationEvidence().MethodId, + staffing.Status, + staffing.StartedAt, + staffing.AbsoluteExpiresAt, + }); + } + + return new PublicAppSnapshot(scope.ScopeVersion, current, presence); + } + + private static async Task StageNetChangesAsync( + IDocumentOperations operations, + AppChangeFeedState feed, + PublicAppSnapshot snapshot, + long sourceSequence, + IEvent? source, + DateTimeOffset now, + CancellationToken cancellationToken) + { + var persisted = await operations.Query() + .Where(x => x.AppId == feed.Id) + .ToListAsync(cancellationToken); + var oldByKey = persisted.ToDictionary(x => new EntityKey(x.EntityKind, x.EntityId)); + var changes = new List(); + + foreach (var (key, entity) in snapshot.Entities) + { + if (!oldByKey.TryGetValue(key, out var old)) + { + changes.Add(new PendingChange(AppChangeKinds.Upsert, entity, null)); + operations.Store(new AppChangeFeedEntityState + { + Id = EntityStateId(feed.Id, key), + AppId = feed.Id, + EntityKind = key.Kind, + EntityId = key.Id, + Fingerprint = entity.Fingerprint, + PayloadJson = entity.PayloadJson!, + }); + } + else if (old.Fingerprint != entity.Fingerprint) + { + changes.Add(new PendingChange(AppChangeKinds.Upsert, entity, null)); + old.Fingerprint = entity.Fingerprint; + old.PayloadJson = entity.PayloadJson!; + operations.Store(old); + } + oldByKey.Remove(key); + } + + foreach (var (key, old) in oldByKey) + { + snapshot.Presence.TryGetValue(key, out var presence); + var kind = presence.Exists ? AppChangeKinds.FellOutOfScope : AppChangeKinds.Deleted; + var tombstone = presence.RemovalPayloadJson is null + ? null + : new PublicEntity(key, presence.RemovalPayloadJson, Fingerprint(presence.RemovalPayloadJson)); + changes.Add(new PendingChange(kind, tombstone ?? new PublicEntity(key, null, string.Empty), null)); + operations.Delete(old.Id); + } + + var ordinal = 0; + foreach (var change in changes + .OrderBy(x => x.Entity.Key.Kind, StringComparer.Ordinal) + .ThenBy(x => x.Entity.Key.Id)) + { + StageEntry( + operations, + feed, + sourceSequence, + ordinal++, + source, + now, + change.ChangeKind, + change.Entity, + change.Reason); + } + } + + private static async Task ReplaceEntityStateAsync( + IDocumentOperations operations, + Guid appId, + IReadOnlyDictionary entities, + CancellationToken cancellationToken) + { + var old = await operations.Query() + .Where(x => x.AppId == appId) + .ToListAsync(cancellationToken); + var oldById = old.ToDictionary(x => x.Id); + foreach (var entity in entities.Values) + { + var id = EntityStateId(appId, entity.Key); + if (oldById.Remove(id, out var existing)) + { + existing.EntityKind = entity.Key.Kind; + existing.EntityId = entity.Key.Id; + existing.Fingerprint = entity.Fingerprint; + existing.PayloadJson = entity.PayloadJson!; + operations.Store(existing); + } + else + { + operations.Store(new AppChangeFeedEntityState + { + Id = id, + AppId = appId, + EntityKind = entity.Key.Kind, + EntityId = entity.Key.Id, + Fingerprint = entity.Fingerprint, + PayloadJson = entity.PayloadJson!, + }); + } + } + foreach (var row in oldById.Values) + operations.Delete(row.Id); + } + + private static void StageEntry( + IDocumentOperations operations, + AppChangeFeedState state, + long sourceSequence, + int ordinal, + IEvent? source, + DateTimeOffset now, + string changeKind, + PublicEntity? entity, + string? reason) + { + operations.Store(new AppChangeFeedEntry + { + Id = EntryId(state.Id, state.Generation, sourceSequence, ordinal), + AppId = state.Id, + Generation = state.Generation, + SourceSequence = sourceSequence, + Ordinal = ordinal, + ScopeVersion = state.ScopeVersion, + SourceEventId = source?.Id, + OriginatedAt = source?.Timestamp ?? now, + RecordedAt = now, + ChangeKind = changeKind, + EntityKind = entity?.Key.Kind, + EntityId = entity?.Key.Id, + PayloadJson = entity?.PayloadJson, + Reason = reason, + }); + } + + private static async Task CompactAsync( + IDocumentOperations operations, + AppChangeFeedState state, + DateTimeOffset now, + CancellationToken cancellationToken) + { + var countFloor = await operations.Query() + .Where(x => x.AppId == state.Id) + .OrderByDescending(x => x.SourceSequence) + .ThenByDescending(x => x.Ordinal) + .Skip(state.MinimumEventCount - 1) + .FirstOrDefaultAsync(cancellationToken); + if (countFloor is null) return; + + var ageFloor = now.AddDays(-state.MinimumRetentionAgeDays); + var removable = await operations.Query() + .Where(x => x.AppId == state.Id + && x.RecordedAt < ageFloor + && (x.SourceSequence < countFloor.SourceSequence + || (x.SourceSequence == countFloor.SourceSequence + && x.Ordinal < countFloor.Ordinal))) + .ToListAsync(cancellationToken); + + foreach (var entry in removable) + { + operations.Delete(entry.Id); + if (entry.SourceSequence > state.RetentionFloorSequence + || (entry.SourceSequence == state.RetentionFloorSequence + && entry.Ordinal > state.RetentionFloorOrdinal)) + { + state.RetentionFloorSequence = entry.SourceSequence; + state.RetentionFloorOrdinal = entry.Ordinal; + } + } + } + + private static string Serialize(T value) => JsonSerializer.Serialize(value, PayloadJson); + + private static string Fingerprint(string json) => + Convert.ToHexStringLower(SHA256.HashData(Encoding.UTF8.GetBytes(json))); + + private static Guid EntityStateId(Guid appId, EntityKey key) => + DeterministicGuid($"state|{appId:N}|{key.Kind}|{key.Id:N}"); + + private static Guid EntryId(Guid appId, int generation, long sequence, int ordinal) => + DeterministicGuid($"entry|{appId:N}|{generation}|{sequence}|{ordinal}"); + + private static Guid DeterministicGuid(string value) + { + var digest = SHA256.HashData(Encoding.UTF8.GetBytes(value)); + return new Guid(digest.AsSpan(0, 16)); + } + + private readonly record struct EntityKey(string Kind, Guid Id); + + private sealed record PublicEntity(EntityKey Key, string? PayloadJson, string Fingerprint) + { + public static PublicEntity Create(EntityKey key, T payload) + { + var json = Serialize(payload); + return new PublicEntity(key, json, AppChangeFeedSubscription.Fingerprint(json)); + } + } + + private readonly record struct EntityPresence(bool Exists, string? RemovalPayloadJson); + private sealed record PublicAppSnapshot( + string ScopeVersion, + IReadOnlyDictionary Entities, + IReadOnlyDictionary Presence); + private sealed record PendingChange(string ChangeKind, PublicEntity Entity, string? Reason); +} + +internal static class AppEntityKinds +{ + public const string Principal = "principal"; + public const string Terminal = "terminal"; + public const string PositionGrant = "position-grant"; + public const string StaffingSession = "staffing-session"; +} + +internal static class AppChangeKinds +{ + public const string Upsert = "Upsert"; + public const string Deleted = "Deleted"; + public const string FellOutOfScope = "FellOutOfScope"; + public const string ScopeChanged = "ScopeChanged"; + public const string FeedDisabled = "FeedDisabled"; +} diff --git a/src/dotnet/Modgud.Api/Features/Management/ManagementBearerAuthorizationService.cs b/src/dotnet/Modgud.Api/Features/Management/ManagementBearerAuthorizationService.cs new file mode 100644 index 00000000..f61a2aac --- /dev/null +++ b/src/dotnet/Modgud.Api/Features/Management/ManagementBearerAuthorizationService.cs @@ -0,0 +1,102 @@ +using System.Globalization; +using System.Security.Claims; +using System.Text.Json; +using Marten; +using Modgud.Authorization.Apps; +using Modgud.Authorization.Principals; +using Modgud.Authorization.Services; +using Modgud.Domain.OAuth.Applications; +using Modgud.Domain.OAuth.Management; +using OpenIddict.Abstractions; +using static OpenIddict.Abstractions.OpenIddictConstants; + +namespace Modgud.Api.Features.Management; + +/// +/// Reusable bearer half of the Management API boundary. Long-lived transports +/// must repeat these checks after the minimal-API endpoint filter has returned, +/// so the client, subject, AppIds, and permission invariants live in one guard. +/// +public sealed class ManagementBearerAuthorizationService( + IQuerySession session, + IPrincipalLookupService principalLookup, + IPermissionService permissionService) +{ + public async Task AuthorizeAsync( + ClaimsPrincipal caller, + Guid? targetAppId, + string permission, + CancellationToken cancellationToken) + { + var expiration = caller.FindFirstValue("exp"); + if (long.TryParse(expiration, NumberStyles.Integer, CultureInfo.InvariantCulture, out var exp) + && DateTimeOffset.FromUnixTimeSeconds(exp) <= DateTimeOffset.UtcNow) + { + return new("token_expired", "The access token has expired."); + } + + if (!caller.GetAudiences().Contains(ModgudManagementApi.Audience, StringComparer.Ordinal)) + return new("invalid_audience", $"The token is not intended for '{ModgudManagementApi.Audience}'."); + if (!caller.HasScope(ModgudManagementApi.Scope)) + return new("missing_scope", $"The token is missing '{ModgudManagementApi.Scope}'."); + + var clientId = caller.GetClaim(Claims.ClientId) ?? caller.GetClaim(Claims.AuthorizedParty); + if (string.IsNullOrWhiteSpace(clientId)) + return new("invalid_client", "The token does not identify its OAuth client."); + + var subject = caller.GetClaim(Claims.Subject) + ?? caller.FindFirstValue(ClaimTypes.NameIdentifier); + if (!Guid.TryParse(subject, out var principalId)) + return new("invalid_subject", "The token subject is not a Modgud principal."); + + var principal = await principalLookup.GetByIdAsync(principalId, cancellationToken); + if (principal is null || !principal.IsActive || principal.IsDeleted) + return new("inactive_subject", "The token subject is not an active principal."); + if (principal is not Person and not ServiceAccount) + return new("unsupported_subject", "Management tokens must represent a Person or Service Account."); + + var client = await session.Query() + .FirstOrDefaultAsync(x => x.ClientId == clientId && !x.IsDeleted, cancellationToken); + if (client is null || !BooleanProperty(client, OAuthApplicationPropertyKeys.Enabled, true)) + return new("invalid_client", "The OAuth client is missing or disabled."); + if (BooleanProperty(client, OAuthApplicationPropertyKeys.DcrIsDynamicallyRegistered, false)) + return new("admin_registered_client_required", "Dynamically registered clients cannot use the Management API."); + + var managementScopePermission = + OpenIddictConstants.Permissions.Prefixes.Scope + ModgudManagementApi.Scope; + if (!client.Permissions.Contains(managementScopePermission, StringComparer.Ordinal)) + return new("client_scope_revoked", $"The client is no longer allowed to request '{ModgudManagementApi.Scope}'."); + + if (principal is ServiceAccount account && client.LinkedServiceAccountId != account.Id) + return new("service_account_client_mismatch", "The client is not linked to the Service Account in the token."); + if (principal is Person && client.LinkedServiceAccountId.HasValue) + return new("delegated_client_required", "A delegated Person token must use a user-flow OAuth client."); + if (targetAppId is { } appId && !client.AppIds.Contains(appId)) + return new("client_app_mismatch", "The client is not assigned to the requested Application."); + + if (!await permissionService.HasPermissionAsync( + principalId, AppSlugs.Modgud, permission, cancellationToken)) + { + return new("permission_denied", $"Missing '{permission}' in the Modgud Application."); + } + + return null; + } + + private static bool BooleanProperty( + OAuthApplicationState client, + string key, + bool defaultValue) + { + if (!client.Properties.TryGetValue(key, out var raw) || raw is null) return defaultValue; + return raw switch + { + bool value => value, + JsonElement { ValueKind: JsonValueKind.True } => true, + JsonElement { ValueKind: JsonValueKind.False } => false, + _ => defaultValue, + }; + } +} + +public sealed record ManagementBearerAuthorizationError(string Code, string Detail); diff --git a/src/dotnet/Modgud.Api/Features/Management/ManagementPermissionEndpointFilter.cs b/src/dotnet/Modgud.Api/Features/Management/ManagementPermissionEndpointFilter.cs index 5a3c5f6b..5f52db16 100644 --- a/src/dotnet/Modgud.Api/Features/Management/ManagementPermissionEndpointFilter.cs +++ b/src/dotnet/Modgud.Api/Features/Management/ManagementPermissionEndpointFilter.cs @@ -1,16 +1,11 @@ using System.Net.Http.Headers; using System.Security.Claims; -using System.Text.Json; using BuildingBlocks.Helper; -using Marten; using Microsoft.AspNetCore.Authentication; using Microsoft.AspNetCore.Authorization; using Microsoft.AspNetCore.Identity; using Modgud.Authorization.Apps; -using Modgud.Authorization.Principals; using Modgud.Authorization.Services; -using Modgud.Domain.OAuth.Applications; -using Modgud.Domain.OAuth.Management; using OpenIddict.Abstractions; using OpenIddict.Validation.AspNetCore; using static OpenIddict.Abstractions.OpenIddictConstants; @@ -42,8 +37,40 @@ public sealed class ManagementPermissionEndpointFilter( return Results.Unauthorized(); var caller = authentication.Principal; - if (bearerRequest && await ValidateBearerCallerAsync(http, caller) is { } bearerError) - return bearerError; + if (bearerRequest) + { + Guid? targetAppId = null; + if (clientAppRouteParameter is not null) + { + var raw = http.Request.RouteValues.TryGetValue( + clientAppRouteParameter, out var value) + ? value?.ToString() + : null; + if (raw is null || !ShortGuid.TryParse(raw, out Guid parsedAppId)) + { + return Results.Problem( + statusCode: StatusCodes.Status400BadRequest, + title: "Invalid Application id", + detail: $"The route value '{{{clientAppRouteParameter}}}' is not a valid Application id.", + extensions: new Dictionary + { + ["code"] = "Management.InvalidTargetApp", + }); + } + targetAppId = parsedAppId; + } + + var authorization = http.RequestServices + .GetRequiredService(); + var denied = await authorization.AuthorizeAsync( + caller, targetAppId, permission, http.RequestAborted); + if (denied is not null) return BearerDenied(denied); + + // Downstream handlers and audit helpers must see the explicitly + // selected bearer identity, never an accidental cookie merge. + http.User = caller; + return await next(context); + } var subject = caller.GetClaim(Claims.Subject) ?? caller.FindFirstValue(ClaimTypes.NameIdentifier); @@ -55,25 +82,6 @@ public sealed class ManagementPermissionEndpointFilter( if (principal is null || !principal.IsActive) return Results.Unauthorized(); - if (bearerRequest && principal is not Person and not ServiceAccount) - return Forbidden("Management.UnsupportedPrincipal", - "Management access tokens must represent a Person or Service Account."); - - if (bearerRequest && principal is ServiceAccount serviceAccount && - await ValidateServiceAccountClientAsync(http, caller, serviceAccount) is { } clientError) - return clientError; - - if (bearerRequest && principal is Person && - await ValidateDelegatedClientAsync(http, caller) is { } delegatedClientError) - return delegatedClientError; - - if (bearerRequest && clientAppRouteParameter is not null && - await ValidateClientAppBoundaryAsync(http, caller, clientAppRouteParameter) - is { } appBoundaryError) - { - return appBoundaryError; - } - var permissionService = http.RequestServices.GetRequiredService(); if (!await permissionService.HasPermissionAsync( principalId, AppSlugs.Modgud, permission, http.RequestAborted)) @@ -88,159 +96,27 @@ await ValidateClientAppBoundaryAsync(http, caller, clientAppRouteParameter) return await next(context); } - private static async Task ValidateBearerCallerAsync( - HttpContext http, - ClaimsPrincipal caller) - { - if (!caller.GetAudiences().Contains(ModgudManagementApi.Audience, StringComparer.Ordinal)) - { - return Forbidden("Management.InvalidAudience", - $"The access token is not intended for '{ModgudManagementApi.Audience}'."); - } - - if (!caller.HasScope(ModgudManagementApi.Scope)) - { - return Forbidden("Management.MissingScope", - $"The access token is missing the '{ModgudManagementApi.Scope}' scope."); - } - - return string.IsNullOrWhiteSpace(ResolveClientId(caller)) - ? Results.Unauthorized() - : null; - } - - private static async Task ValidateServiceAccountClientAsync( - HttpContext http, - ClaimsPrincipal caller, - ServiceAccount serviceAccount) - { - var client = await LoadClientAsync(http, caller); - if (ValidateRegisteredClient(client) is { } registrationError) - return registrationError; - - if (client!.LinkedServiceAccountId != serviceAccount.Id) - { - return Forbidden("Management.ServiceAccountClientMismatch", - "The OAuth client is not linked to the Service Account represented by the token."); - } - - return null; - } - - private static async Task ValidateDelegatedClientAsync( - HttpContext http, - ClaimsPrincipal caller) - { - var client = await LoadClientAsync(http, caller); - if (ValidateRegisteredClient(client) is { } registrationError) - return registrationError; - - // Grant separation is an issuance invariant, and is repeated here so a - // stale or malformed token can never turn an M2M credential into a - // delegated-user management client. - if (client!.LinkedServiceAccountId.HasValue) - { - return Forbidden("Management.DelegatedClientRequired", - "A delegated user token must be issued to a user-flow OAuth client."); - } - - return null; - } - - private static IResult? ValidateRegisteredClient(OAuthApplicationState? client) + private static IResult BearerDenied(ManagementBearerAuthorizationError error) { - if (client is null || - !GetBooleanProperty(client, OAuthApplicationPropertyKeys.Enabled, defaultValue: true)) - { + if (error.Code is "invalid_client" or "invalid_subject" or "inactive_subject" + or "token_expired") return Results.Unauthorized(); - } - - // Management clients are an administrator-issued trust decision. DCR - // already prevents opting into the protected scope; repeat the invariant - // at the resource boundary so malformed legacy events cannot bypass it. - if (GetBooleanProperty( - client, - OAuthApplicationPropertyKeys.DcrIsDynamicallyRegistered, - defaultValue: false)) - { - return Forbidden("Management.AdminRegisteredClientRequired", - "Dynamically registered clients cannot call the Modgud management API."); - } - - var managementScopePermission = - OpenIddictConstants.Permissions.Prefixes.Scope + ModgudManagementApi.Scope; - if (!client.Permissions.Contains(managementScopePermission, StringComparer.Ordinal)) - { - return Forbidden("Management.ClientScopeRevoked", - $"The OAuth client is no longer allowed to request '{ModgudManagementApi.Scope}'."); - } - - return null; - } - - private static async Task ValidateClientAppBoundaryAsync( - HttpContext http, - ClaimsPrincipal caller, - string routeParameter) - { - var raw = http.Request.RouteValues.TryGetValue(routeParameter, out var value) - ? value?.ToString() - : null; - if (raw is null || !ShortGuid.TryParse(raw, out Guid targetAppId)) - { - return Results.Problem( - statusCode: StatusCodes.Status400BadRequest, - title: "Invalid Application id", - detail: $"The route value '{{{routeParameter}}}' is not a valid Application id.", - extensions: new Dictionary - { - ["code"] = "Management.InvalidTargetApp", - }); - } - var client = await LoadClientAsync(http, caller); - if (client is null || !client.AppIds.Contains(targetAppId)) + var publicCode = error.Code switch { - return Forbidden( - "Management.ClientAppMismatch", - $"The OAuth client is not assigned to Application '{new ShortGuid(targetAppId)}'."); - } - - return null; - } - - private static async Task LoadClientAsync( - HttpContext http, - ClaimsPrincipal caller) - { - var clientId = ResolveClientId(caller); - if (string.IsNullOrWhiteSpace(clientId)) return null; - - var session = http.RequestServices.GetRequiredService(); - return await session.Query() - .FirstOrDefaultAsync( - candidate => candidate.ClientId == clientId && !candidate.IsDeleted, - http.RequestAborted); - } - - private static string? ResolveClientId(ClaimsPrincipal caller) => - caller.GetClaim(Claims.ClientId) ?? caller.GetClaim(Claims.AuthorizedParty); - - private static bool GetBooleanProperty( - OAuthApplicationState client, - string key, - bool defaultValue) - { - if (!client.Properties.TryGetValue(key, out var raw) || raw is null) - return defaultValue; - - return raw switch - { - bool value => value, - JsonElement { ValueKind: JsonValueKind.True } => true, - JsonElement { ValueKind: JsonValueKind.False } => false, - _ => defaultValue, + "invalid_audience" => "Management.InvalidAudience", + "missing_scope" => "Management.MissingScope", + "unsupported_subject" => "Management.UnsupportedPrincipal", + "admin_registered_client_required" => "Management.AdminRegisteredClientRequired", + "client_scope_revoked" => "Management.ClientScopeRevoked", + "service_account_client_mismatch" => "Management.ServiceAccountClientMismatch", + "delegated_client_required" => "Management.DelegatedClientRequired", + "client_app_mismatch" => "Management.ClientAppMismatch", + "permission_denied" => "Management.PermissionDenied", + _ => "Management.Forbidden", }; + + return Forbidden(publicCode, error.Detail); } private static bool IsBearerRequest(HttpRequest request) @@ -272,11 +148,14 @@ public static class ManagementPermissionEndpointExtensions public static RouteHandlerBuilder RequiresManagementPermission( this RouteHandlerBuilder builder, string permission, - string? clientAppRouteParameter = null) + string? clientAppRouteParameter = null, + bool bearerOnly = false) { builder.RequireAuthorization(new AuthorizeAttribute { - AuthenticationSchemes = AuthenticationSchemes, + AuthenticationSchemes = bearerOnly + ? OpenIddictValidationAspNetCoreDefaults.AuthenticationScheme + : AuthenticationSchemes, }); builder.AddEndpointFilter(new ManagementPermissionEndpointFilter( permission, diff --git a/src/dotnet/Modgud.Api/Features/ServiceAccounts/ServiceAccountsEndpoints.cs b/src/dotnet/Modgud.Api/Features/ServiceAccounts/ServiceAccountsEndpoints.cs index 7d51e51c..667a821c 100644 --- a/src/dotnet/Modgud.Api/Features/ServiceAccounts/ServiceAccountsEndpoints.cs +++ b/src/dotnet/Modgud.Api/Features/ServiceAccounts/ServiceAccountsEndpoints.cs @@ -7,6 +7,7 @@ using Modgud.Authentication.ExtensionMethods; using Modgud.Authorization.AspNetCore; using Modgud.Authorization.Principals; +using Modgud.Authorization.Events; using Modgud.Domain.ValueObjects; using Modgud.Domain.OAuth.Common; using Modgud.Infrastructure.OpenIddict; @@ -135,7 +136,8 @@ public static WebApplication MapServiceAccountsEndpoints(this WebApplication app Purpose = string.IsNullOrWhiteSpace(dto.Purpose) ? null : dto.Purpose.Trim(), IsActive = dto.IsActive, }; - session.Store(sa); + session.Events.StartStream(sa.Id, new ServiceAccountCreatedEvent( + sa.Id, sa.AccountName, sa.Purpose, sa.IsActive)); await session.SaveChangesAsync(ct); var created = ToDto(sa); @@ -150,7 +152,11 @@ public static WebApplication MapServiceAccountsEndpoints(this WebApplication app var sa = await session.LoadAsync(id.Guid, ct); if (sa is null || sa.IsDeleted) return Results.NotFound(); - // Prior active-state, read from the persisted record (not the request). + // Preserve the persisted snapshot before applying the request. + // A legacy document-only account needs this exact state as its + // creation event, followed by the real update event below. + var previousAccountName = sa.AccountName; + var previousPurpose = sa.Purpose; var wasActive = sa.IsActive; if (dto.AccountName is { } rawAccountName) @@ -190,7 +196,23 @@ public static WebApplication MapServiceAccountsEndpoints(this WebApplication app if (dto.IsActive.HasValue) sa.IsActive = dto.IsActive.Value; - session.Store(sa); + var stream = await session.Events.FetchStreamStateAsync(sa.Id, ct); + if (stream is null) + { + // Legacy document-only account: establish its event history + // from the persisted snapshot, then preserve this request as + // a distinct mutation instead of folding it into creation. + session.Events.StartStream(sa.Id, + new ServiceAccountCreatedEvent( + sa.Id, previousAccountName, previousPurpose, wasActive), + new ServiceAccountUpdatedEvent( + sa.Id, sa.AccountName, sa.Purpose, sa.IsActive)); + } + else + { + session.Events.Append(sa.Id, new ServiceAccountUpdatedEvent( + sa.Id, sa.AccountName, sa.Purpose, sa.IsActive)); + } await session.SaveChangesAsync(ct); // Audit #6 — deactivating an SA must cut off its live M2M access, not @@ -231,25 +253,28 @@ public static WebApplication MapServiceAccountsEndpoints(this WebApplication app if (sa is null || sa.IsDeleted) return Results.NotFound(); // Phase 2C — cascade-delete every credential owned by this SA - // PLUS soft-delete the SA itself in one unit of work. The SA - // delete is queued via the strongly-typed Delete overload - // (Marten translates this into a soft-delete on the polymorphic - // mt_doc_principal table without triggering an optimistic - // concurrency check against the in-memory ServiceAccount - // instance we loaded above — which Store(sa) would have). + // plus soft-delete the SA itself in one unit of work. var deletedCredentialCount = await oauth .StageDeleteAllServiceAccountCredentialsAsync(id.Guid, ct); - // We want soft-delete semantics (IsDeleted=true) so audit / - // group-membership references stay resolvable. Mutate the - // loaded instance + Update — Marten's Update path skips the - // Store identity-map concurrency dance that Store + Append - // mix-mode runs into. (See Marten 8 polymorphic-store + - // events.Append in one session.) - sa.IsDeleted = true; - session.Update(sa); + var stream = await session.Events.FetchStreamStateAsync(sa.Id, ct); + if (stream is null) + { + // Legacy document-only account: seed its history from the + // persisted snapshot before recording deletion. + session.Events.StartStream(sa.Id, + new ServiceAccountCreatedEvent(sa.Id, sa.AccountName, sa.Purpose, sa.IsActive), + new ServiceAccountDeletedEvent(sa.Id)); + } + else + { + session.Events.Append(sa.Id, new ServiceAccountDeletedEvent(sa.Id)); + } await session.SaveChangesAsync(ct); + sa.IsDeleted = true; + sa.IsActive = false; + // Audit #7 — deleting an SA cascade-deletes its credential clients, // but a deleted client document does NOT invalidate already-issued // M2M tokens. Revoke them by subject (sub = sa.Id) so outstanding diff --git a/src/dotnet/Modgud.Api/Program.cs b/src/dotnet/Modgud.Api/Program.cs index afc8f3ce..e3372e60 100644 --- a/src/dotnet/Modgud.Api/Program.cs +++ b/src/dotnet/Modgud.Api/Program.cs @@ -269,6 +269,9 @@ ); }); + builder.Services.AddScoped(); + builder.Services.AddScoped(); + builder.Services.AddSingleton(); @@ -773,6 +776,7 @@ options.UseModgudAuthentication(); options.UseModgudOAuth(); options.UseModgudPositionTerminals(); + options.Events.Subscribe(new Modgud.Api.Features.ChangeFeed.AppChangeFeedSubscription()); }, // The behavioural integration suite owns projection progress explicitly: // each consistency boundary runs a fresh interactive daemon. Running the @@ -1368,6 +1372,7 @@ // An App is one resource: AppsEndpoints carries the per-App ADR-0011 settings override // inline (POST/PUT/GET /api/app), so there is no separate /settings endpoint. Modgud.Api.Features.Admin.Apps.AppsEndpoints.MapAppsEndpoints(app, "api"); + Modgud.Api.Features.ChangeFeed.AppChangeFeedEndpoints.MapAppChangeFeedEndpoints(app, "api"); // ADR-0012 — app-scoped invite codes (dual-auth: invite:write scope or invite-code:write permission). Modgud.Api.Features.InviteCodes.InviteCodeEndpoints.MapInviteCodeEndpoints(app, "api"); diff --git a/src/dotnet/Modgud.Application/DTOs/Applications/ApplicationSettingsDtos.cs b/src/dotnet/Modgud.Application/DTOs/Applications/ApplicationSettingsDtos.cs index b4907cf7..79dbd68d 100644 --- a/src/dotnet/Modgud.Application/DTOs/Applications/ApplicationSettingsDtos.cs +++ b/src/dotnet/Modgud.Application/DTOs/Applications/ApplicationSettingsDtos.cs @@ -21,6 +21,7 @@ public record ApplicationSettingsDto public ApplicationDcrDto? Dcr { get; init; } public ApplicationCimdDto? Cimd { get; init; } public ApplicationRegistrationFieldsDto? RegistrationFields { get; init; } + public ApplicationChangeFeedDto? ChangeFeed { get; init; } } public record ApplicationOriginDto @@ -127,3 +128,15 @@ public record ApplicationRegistrationFieldsDto public string? Firstname { get; init; } public string? Lastname { get; init; } } + +/// +/// Explicit per-App opt-in and retention policy for the resumable consumer +/// change feed. Retention keeps both the complete age window and at least the +/// newest event count. +/// +public record ApplicationChangeFeedDto +{ + public bool Enabled { get; init; } + public int MinimumRetentionAgeDays { get; init; } = 7; + public int MinimumEventCount { get; init; } = 1_000; +} diff --git a/src/dotnet/Modgud.Application/DTOs/ChangeFeed/AppChangeFeedDtos.cs b/src/dotnet/Modgud.Application/DTOs/ChangeFeed/AppChangeFeedDtos.cs new file mode 100644 index 00000000..77aa885f --- /dev/null +++ b/src/dotnet/Modgud.Application/DTOs/ChangeFeed/AppChangeFeedDtos.cs @@ -0,0 +1,45 @@ +using System.Text.Json; + +namespace Modgud.Application.DTOs.ChangeFeed; + +public static class AppChangeFeedContract +{ + public const int Version = 1; +} + +public sealed record AppScopedEntityDto( + string EntityKind, + string EntityId, + int EntityVersion, + JsonElement Payload); + +public sealed record AppChangeFeedSnapshotDto( + int ContractVersion, + string AppId, + string AppSlug, + string ScopeVersion, + string Cursor, + IReadOnlyList Entities); + +public sealed record AppChangeFeedMessageDto +{ + public required int ContractVersion { get; init; } + public required string Kind { get; init; } + public required string Cursor { get; init; } + public required string ScopeVersion { get; init; } + public string? ChangeKind { get; init; } + public string? EntityKind { get; init; } + public string? EntityId { get; init; } + public int? EntityVersion { get; init; } + public JsonElement? Payload { get; init; } + public Guid? SourceEventId { get; init; } + public DateTimeOffset? OriginatedAt { get; init; } + public string? Reason { get; init; } +} + +public sealed record AppChangeFeedReadDto( + int ContractVersion, + string ScopeVersion, + IReadOnlyList Messages, + bool HasMore, + bool FeedEnded = false); diff --git a/src/dotnet/Modgud.Application/Services/OAuthAdminService.cs b/src/dotnet/Modgud.Application/Services/OAuthAdminService.cs index 056c63df..8c722e87 100644 --- a/src/dotnet/Modgud.Application/Services/OAuthAdminService.cs +++ b/src/dotnet/Modgud.Application/Services/OAuthAdminService.cs @@ -8,6 +8,7 @@ using Modgud.Application.Errors; using Modgud.Authorization.Apps; using Modgud.Authorization.Principals; +using Modgud.Authorization.Events; using Modgud.Domain.OAuth.Apis; using Modgud.Domain.OAuth.Applications; using Modgud.Domain.OAuth.Common; @@ -180,7 +181,12 @@ public async Task> CreateClientAsync( : dto.NewServiceAccount.Purpose.Trim(), IsActive = dto.NewServiceAccount.IsActive, }; - _session.Store(serviceAccount); + _session.Events.StartStream(serviceAccount.Id, + new ServiceAccountCreatedEvent( + serviceAccount.Id, + serviceAccount.AccountName, + serviceAccount.Purpose, + serviceAccount.IsActive)); linkedServiceAccountId = serviceAccount.Id; createdServiceAccount = new ServiceAccountDto { diff --git a/src/dotnet/Modgud.Authentication/Api/Admin/RecoveryCommands.cs b/src/dotnet/Modgud.Authentication/Api/Admin/RecoveryCommands.cs index 5d9b3424..abf1309f 100644 --- a/src/dotnet/Modgud.Authentication/Api/Admin/RecoveryCommands.cs +++ b/src/dotnet/Modgud.Authentication/Api/Admin/RecoveryCommands.cs @@ -581,7 +581,9 @@ public async Task ExecuteAsync(RecoveryCliContext ctx) Purpose = $"Auto-provisioned by migrate-cc-credentials for OAuth client '{client.ClientId}'.", IsActive = true, }; - session.Store(sa); + session.Events.StartStream(sa.Id, + new Modgud.Authorization.Events.ServiceAccountCreatedEvent( + sa.Id, sa.AccountName, sa.Purpose, sa.IsActive)); saCreated++; } diff --git a/src/dotnet/Modgud.Authentication/Applications/ApplicationSettingsService.cs b/src/dotnet/Modgud.Authentication/Applications/ApplicationSettingsService.cs index d8ce538c..42739909 100644 --- a/src/dotnet/Modgud.Authentication/Applications/ApplicationSettingsService.cs +++ b/src/dotnet/Modgud.Authentication/Applications/ApplicationSettingsService.cs @@ -165,6 +165,14 @@ public async Task> PatchAsync( doc.RegistrationFields = r.Value; } + if (dto.ChangeFeed is not null) + { + var r = MapChangeFeed(dto.ChangeFeed); + if (r.IsError) return r.FirstError; + StageChangeFeedConfigurationEvent(applicationId, doc.ChangeFeed, r.Value); + doc.ChangeFeed = r.Value; + } + // Origin / subdomain is special: it also drives the GLOBAL host→App routing // map, validated for child-of-primary-domain + cross-realm uniqueness. if (dto.Origin is not null) @@ -231,6 +239,11 @@ public async Task> StageNonOriginAsync( if (dto.RegistrationFields is null) doc.RegistrationFields = null; else { var r = MapRegistrationFields(dto.RegistrationFields); if (r.IsError) return r.FirstError; doc.RegistrationFields = r.Value; } + var oldChangeFeed = doc.ChangeFeed; + if (dto.ChangeFeed is null) doc.ChangeFeed = null; + else { var r = MapChangeFeed(dto.ChangeFeed); if (r.IsError) return r.FirstError; doc.ChangeFeed = r.Value; } + StageChangeFeedConfigurationEvent(applicationId, oldChangeFeed, doc.ChangeFeed); + doc.UpdatedAt = DateTimeOffset.UtcNow; session.Store(doc); // enrolled on the shared session; the caller commits. return ErrorOr.Result.Success; @@ -419,6 +432,40 @@ private static ErrorOr MapRegistrationFi }; } + private static ErrorOr MapChangeFeed(ApplicationChangeFeedDto dto) + { + if (dto.MinimumRetentionAgeDays is < 1 or > 3650) + return Error.Validation("Application.ChangeFeed.InvalidRetentionAge", + "MinimumRetentionAgeDays must be between 1 and 3650."); + if (dto.MinimumEventCount is < 1 or > 1_000_000) + return Error.Validation("Application.ChangeFeed.InvalidEventCount", + "MinimumEventCount must be between 1 and 1000000."); + + return new ApplicationChangeFeedSettings + { + Enabled = dto.Enabled, + MinimumRetentionAge = TimeSpan.FromDays(dto.MinimumRetentionAgeDays), + MinimumEventCount = dto.MinimumEventCount, + }; + } + + private void StageChangeFeedConfigurationEvent( + Guid applicationId, + ApplicationChangeFeedSettings? previous, + ApplicationChangeFeedSettings? current) + { + var oldValue = previous ?? ApplicationChangeFeedSettings.Disabled; + var newValue = current ?? ApplicationChangeFeedSettings.Disabled; + if (oldValue == newValue) return; + + session.Events.Append(applicationId, new ApplicationChangeFeedConfiguredEvent( + applicationId, + newValue.Enabled, + (int)newValue.MinimumRetentionAge.TotalDays, + newValue.MinimumEventCount, + DateTimeOffset.UtcNow)); + } + private static ErrorOr ParseRequirement(string field, string? raw) { if (string.IsNullOrWhiteSpace(raw)) return (FieldRequirement?)null; @@ -680,6 +727,12 @@ internal static ApplicationSettingsDto ToDto(ApplicationSettings? doc) Firstname = doc.RegistrationFields.Firstname?.ToString(), Lastname = doc.RegistrationFields.Lastname?.ToString(), }, + ChangeFeed = doc.ChangeFeed is null ? null : new ApplicationChangeFeedDto + { + Enabled = doc.ChangeFeed.Enabled, + MinimumRetentionAgeDays = (int)doc.ChangeFeed.MinimumRetentionAge.TotalDays, + MinimumEventCount = doc.ChangeFeed.MinimumEventCount, + }, }; } } diff --git a/src/dotnet/Modgud.Authentication/Projections/PersonProjection.cs b/src/dotnet/Modgud.Authentication/Projections/PersonProjection.cs index 8c80fcf3..5f6b182b 100644 --- a/src/dotnet/Modgud.Authentication/Projections/PersonProjection.cs +++ b/src/dotnet/Modgud.Authentication/Projections/PersonProjection.cs @@ -20,7 +20,7 @@ public PersonProjection() // mt_doc_principal table. Marten's default projection teardown truncates // the root table, so rebuilding PersonProjection by itself would also // delete every Group and directly stored ServiceAccount. The supported - // PrincipalProjectionRebuilder replays both event-sourced principal + // PrincipalProjectionRebuilder replays all event-sourced principal // projections in place, then performs subtype-scoped stale-row cleanup. Options.TeardownDataOnRebuild = false; diff --git a/src/dotnet/Modgud.Authentication/Projections/PrincipalProjectionRebuilder.cs b/src/dotnet/Modgud.Authentication/Projections/PrincipalProjectionRebuilder.cs index c9573ecd..e31ffd71 100644 --- a/src/dotnet/Modgud.Authentication/Projections/PrincipalProjectionRebuilder.cs +++ b/src/dotnet/Modgud.Authentication/Projections/PrincipalProjectionRebuilder.cs @@ -7,12 +7,13 @@ namespace Modgud.Authentication.Projections; /// /// Rebuilds all event-sourced subtypes without deleting -/// directly stored documents from their shared table. +/// legacy document-only rows from their shared table. /// public static class PrincipalProjectionRebuilder { /// - /// Replays both projections in place, then deletes stale Person and Group + /// Replays every Principal projection in place, then deletes stale Person, + /// Group, and Position /// discriminator rows that have no non-archived creation event. Neither /// projection may use Marten's default teardown because it truncates the root /// mt_doc_principal table. @@ -33,13 +34,22 @@ public static async Task RebuildAsync( await daemon.RebuildProjectionAsync(timeout, ct); progress?.Invoke("OK GroupProjection (mt_doc_principal/group)"); + await daemon.RebuildProjectionAsync(timeout, ct); + progress?.Invoke("OK PositionPrincipalProjection (mt_doc_principal/position)"); + + await daemon.RebuildProjectionAsync(timeout, ct); + progress?.Invoke("OK ServiceAccountProjection (mt_doc_principal/service-account)"); + await using var session = store.LightweightSession(tenantId); // Replay happens before cleanup so live reads and inline writes never see // a deliberately emptied principal table. The creation-event aliases are // stable persisted contracts registered in MartenConfiguration and the // authorization slice. Archived streams must not keep a projected row. - // ServiceAccount is excluded by discriminator and remains byte-for-byte - // untouched because it has no event stream to replay. + // Service-account cleanup is deliberately omitted: old installations + // can still contain valid document-only rows without a creation event. + // The teardown-free replay above updates every event-sourced account + // while preserving those legacy snapshots until their first mutation + // seeds a stream. session.QueueSqlCommand( """ delete from mt_doc_principal as principal @@ -52,6 +62,14 @@ and e.type in (?, ?) and coalesce(e.is_archived, false) = false )) or + (principal.mt_doc_type = ? and not exists ( + select 1 + from mt_events as e + where e.stream_id = principal.id + and e.type = ? + and coalesce(e.is_archived, false) = false + )) + or (principal.mt_doc_type = ? and not exists ( select 1 from mt_events as e @@ -64,7 +82,9 @@ and coalesce(e.is_archived, false) = false "user_created", "user_migrated", "group", - "authorization_group_created"); + "authorization_group_created", + "position", + "authorization_position_created"); await session.SaveChangesAsync(ct); } } diff --git a/src/dotnet/Modgud.Authentication/Setup/MartenStoreOptionsExtensions.cs b/src/dotnet/Modgud.Authentication/Setup/MartenStoreOptionsExtensions.cs index 80f65ad6..16b47855 100644 --- a/src/dotnet/Modgud.Authentication/Setup/MartenStoreOptionsExtensions.cs +++ b/src/dotnet/Modgud.Authentication/Setup/MartenStoreOptionsExtensions.cs @@ -9,6 +9,7 @@ using Modgud.Authentication.Gdpr; using Modgud.Domain.Common; using Modgud.Domain.Users.Events; +using Modgud.Domain.Applications; using RealmSettingsDoc = Modgud.Domain.RealmSettings.RealmSettings; using Modgud.Authentication.Domain.Saml; using Modgud.Authentication.Identity.ExternalAuth; @@ -204,6 +205,8 @@ public static StoreOptions UseModgudAuthentication(this StoreOptions options) options.Schema.For() .Identity(x => x.Id); + options.Events.MapEventType("application_change_feed_configured"); + // Tenant-scoped singleton — per-realm SAML SP certificate state // (PFX bytes DataProtection-encrypted, active + previous slots for // rotation overlap). Lazily generated by SamlSpCertificateService. diff --git a/src/dotnet/Modgud.Authorization/Apps/ApplicationScopeResolver.cs b/src/dotnet/Modgud.Authorization/Apps/ApplicationScopeResolver.cs index 4493e576..c90f7e75 100644 --- a/src/dotnet/Modgud.Authorization/Apps/ApplicationScopeResolver.cs +++ b/src/dotnet/Modgud.Authorization/Apps/ApplicationScopeResolver.cs @@ -43,7 +43,7 @@ public sealed class ApplicationScopeResolver(IQuerySession session) : IApplicati return BuildSnapshot(app, principals); } - internal static ApplicationScopeSnapshot BuildSnapshot( + public static ApplicationScopeSnapshot BuildSnapshot( App app, IReadOnlyCollection directory) { diff --git a/src/dotnet/Modgud.Authorization/Events/ServiceAccountEvents.cs b/src/dotnet/Modgud.Authorization/Events/ServiceAccountEvents.cs new file mode 100644 index 00000000..b30dec5c --- /dev/null +++ b/src/dotnet/Modgud.Authorization/Events/ServiceAccountEvents.cs @@ -0,0 +1,23 @@ +using Modgud.Authorization.Principals; + +namespace Modgud.Authorization.Events; + +/// +/// Stream events for . Existing installations can +/// contain legacy document-only service accounts; their first mutation starts +/// a stream from the persisted snapshot before recording the mutation. +/// +public sealed record ServiceAccountCreatedEvent( + Guid Id, + string AccountName, + string? Purpose, + bool IsActive); + +/// Full-replace service-account update. +public sealed record ServiceAccountUpdatedEvent( + Guid Id, + string AccountName, + string? Purpose, + bool IsActive); + +public sealed record ServiceAccountDeletedEvent(Guid Id); diff --git a/src/dotnet/Modgud.Authorization/Projections/GroupProjection.cs b/src/dotnet/Modgud.Authorization/Projections/GroupProjection.cs index 1b896eb8..54dd6a00 100644 --- a/src/dotnet/Modgud.Authorization/Projections/GroupProjection.cs +++ b/src/dotnet/Modgud.Authorization/Projections/GroupProjection.cs @@ -13,8 +13,8 @@ public partial class GroupProjection : SingleStreamProjection { public GroupProjection() { - // Group shares mt_doc_principal with Person and the non-event-sourced - // ServiceAccount subtype. A normal Marten teardown would truncate that + // Group shares mt_doc_principal with the other Principal subtypes. A + // normal Marten teardown would truncate that // entire root table. Cleanup is therefore coordinated explicitly by the // authentication slice's PrincipalProjectionRebuilder. Options.TeardownDataOnRebuild = false; diff --git a/src/dotnet/Modgud.Authorization/Projections/PositionPrincipalProjection.cs b/src/dotnet/Modgud.Authorization/Projections/PositionPrincipalProjection.cs index c8b37bcc..f9a8c5d4 100644 --- a/src/dotnet/Modgud.Authorization/Projections/PositionPrincipalProjection.cs +++ b/src/dotnet/Modgud.Authorization/Projections/PositionPrincipalProjection.cs @@ -14,8 +14,8 @@ public partial class PositionPrincipalProjection : SingleStreamProjection +/// Builds service-account documents inline from their streams. Teardown stays +/// disabled because all principal subtypes share the same physical table and +/// legacy service accounts may not have a stream yet. +/// +public partial class ServiceAccountProjection : SingleStreamProjection +{ + public ServiceAccountProjection() + { + Options.TeardownDataOnRebuild = false; + IncludeType(); + IncludeType(); + IncludeType(); + } + + public ServiceAccount Apply(ServiceAccountCreatedEvent @event, ServiceAccount _) => new() + { + Id = @event.Id, + AccountName = @event.AccountName, + Purpose = @event.Purpose, + IsActive = @event.IsActive, + IsDeleted = false, + }; + + public ServiceAccount Apply(ServiceAccountUpdatedEvent @event, ServiceAccount account) + { + account.AccountName = @event.AccountName; + account.Purpose = @event.Purpose; + account.IsActive = @event.IsActive; + return account; + } + + public ServiceAccount Apply(ServiceAccountDeletedEvent _, ServiceAccount account) + { + account.IsDeleted = true; + account.IsActive = false; + return account; + } +} diff --git a/src/dotnet/Modgud.Authorization/Setup/MartenStoreOptionsExtensions.cs b/src/dotnet/Modgud.Authorization/Setup/MartenStoreOptionsExtensions.cs index e6ad24da..e0aa1919 100644 --- a/src/dotnet/Modgud.Authorization/Setup/MartenStoreOptionsExtensions.cs +++ b/src/dotnet/Modgud.Authorization/Setup/MartenStoreOptionsExtensions.cs @@ -73,6 +73,7 @@ public static StoreOptions UseModgudAuthorization(this StoreOptions martenOpts) martenOpts.Projections.Add(ProjectionLifecycle.Inline); martenOpts.Projections.Add(ProjectionLifecycle.Inline); martenOpts.Projections.Add(ProjectionLifecycle.Inline); + martenOpts.Projections.Add(ProjectionLifecycle.Inline); martenOpts.Projections.Add(ProjectionLifecycle.Inline); // Stable event-type aliases. Marten resolves events through these, so the @@ -87,6 +88,10 @@ public static StoreOptions UseModgudAuthorization(this StoreOptions martenOpts) martenOpts.Events.MapEventType("authorization_position_updated"); martenOpts.Events.MapEventType("authorization_position_deleted"); + martenOpts.Events.MapEventType("authorization_service_account_created"); + martenOpts.Events.MapEventType("authorization_service_account_updated"); + martenOpts.Events.MapEventType("authorization_service_account_deleted"); + martenOpts.Events.MapEventType("permission_role_created"); martenOpts.Events.MapEventType("permission_role_updated"); martenOpts.Events.MapEventType("permission_role_deleted"); diff --git a/src/dotnet/Modgud.Domain/Applications/ApplicationChangeFeedEvents.cs b/src/dotnet/Modgud.Domain/Applications/ApplicationChangeFeedEvents.cs new file mode 100644 index 00000000..6dce6c1f --- /dev/null +++ b/src/dotnet/Modgud.Domain/Applications/ApplicationChangeFeedEvents.cs @@ -0,0 +1,14 @@ +namespace Modgud.Domain.Applications; + +/// +/// Durable source signal for a change-feed policy change. The settings +/// document remains the query model; this event lets the high-water-anchored +/// feed subscription observe enable/disable and retention changes in the same +/// ordered source as all event-sourced domain changes. +/// +public sealed record ApplicationChangeFeedConfiguredEvent( + Guid ApplicationId, + bool Enabled, + int MinimumRetentionAgeDays, + int MinimumEventCount, + DateTimeOffset ChangedAt); diff --git a/src/dotnet/Modgud.Domain/Applications/ApplicationSettings.cs b/src/dotnet/Modgud.Domain/Applications/ApplicationSettings.cs index d845393d..a83f69a6 100644 --- a/src/dotnet/Modgud.Domain/Applications/ApplicationSettings.cs +++ b/src/dotnet/Modgud.Domain/Applications/ApplicationSettings.cs @@ -87,6 +87,15 @@ public class ApplicationSettings /// while an Enterprise App in the same tenant requires given/family name. public ApplicationRegistrationFieldsOverrides? RegistrationFields { get; set; } + /// + /// Consumer-facing, resumable Application change feed. This is an explicit + /// per-App opt-in: null and + /// false both mean that no remote consumer may subscribe. Retention + /// keeps the union of the age window and the newest event count, so a quiet + /// App never loses its last known resume window merely because time passed. + /// + public ApplicationChangeFeedSettings? ChangeFeed { get; set; } + /// LEGACY (pre-ADR-0001): single per-Application PageBuilder /// schema per slot. Retained only for to /// convert on load; cleared on the next save. New reads/writes use @@ -226,3 +235,16 @@ public record ApplicationRegistrationFieldsOverrides public FieldRequirement? Firstname { get; init; } public FieldRequirement? Lastname { get; init; } } + +/// +/// Per-Application retention and enablement policy for the resumable change +/// feed. Defaults are deliberately conservative and may be tuned per App. +/// +public sealed record ApplicationChangeFeedSettings +{ + public bool Enabled { get; init; } + public TimeSpan MinimumRetentionAge { get; init; } = TimeSpan.FromDays(7); + public int MinimumEventCount { get; init; } = 1_000; + + public static ApplicationChangeFeedSettings Disabled { get; } = new(); +} diff --git a/src/dotnet/Modgud.Infrastructure/ChangeFeed/AppChangeFeedDocuments.cs b/src/dotnet/Modgud.Infrastructure/ChangeFeed/AppChangeFeedDocuments.cs new file mode 100644 index 00000000..8477241b --- /dev/null +++ b/src/dotnet/Modgud.Infrastructure/ChangeFeed/AppChangeFeedDocuments.cs @@ -0,0 +1,53 @@ +namespace Modgud.Infrastructure.ChangeFeed; + +/// Per-Application durable progress and retention state. +public sealed class AppChangeFeedState +{ + public Guid Id { get; set; } + public bool Enabled { get; set; } + public int Generation { get; set; } + public string ScopeVersion { get; set; } = string.Empty; + public long LastProcessedSequence { get; set; } + public long RetentionFloorSequence { get; set; } + public int RetentionFloorOrdinal { get; set; } = -1; + public int MinimumRetentionAgeDays { get; set; } = 7; + public int MinimumEventCount { get; set; } = 1_000; + public DateTimeOffset? LastCompactedAt { get; set; } +} + +/// +/// Last public representation emitted for one entity in one Application. This +/// is feed-owned projection state, not a second source of business truth. +/// +public sealed class AppChangeFeedEntityState +{ + public Guid Id { get; set; } + public Guid AppId { get; set; } + public string EntityKind { get; set; } = string.Empty; + public Guid EntityId { get; set; } + public string Fingerprint { get; set; } = string.Empty; + public string PayloadJson { get; set; } = string.Empty; +} + +/// +/// Short-lived integration envelope. The source event store remains the +/// durable business history; these rows exist only to make consumer resume +/// possible for the configured retention window. +/// +public sealed class AppChangeFeedEntry +{ + public Guid Id { get; set; } + public Guid AppId { get; set; } + public int Generation { get; set; } + public long SourceSequence { get; set; } + public int Ordinal { get; set; } + public string ScopeVersion { get; set; } = string.Empty; + public Guid? SourceEventId { get; set; } + public DateTimeOffset OriginatedAt { get; set; } + public DateTimeOffset RecordedAt { get; set; } + public string ChangeKind { get; set; } = string.Empty; + public string? EntityKind { get; set; } + public Guid? EntityId { get; set; } + public string? PayloadJson { get; set; } + public string? Reason { get; set; } +} diff --git a/src/dotnet/Modgud.Infrastructure/ChangeFeed/AppChangeFeedMartenSetup.cs b/src/dotnet/Modgud.Infrastructure/ChangeFeed/AppChangeFeedMartenSetup.cs new file mode 100644 index 00000000..52569f81 --- /dev/null +++ b/src/dotnet/Modgud.Infrastructure/ChangeFeed/AppChangeFeedMartenSetup.cs @@ -0,0 +1,27 @@ +using Marten; + +namespace Modgud.Infrastructure.ChangeFeed; + +public static class AppChangeFeedMartenSetup +{ + public static StoreOptions UseAppChangeFeed(this StoreOptions options) + { + options.Schema.For() + .Identity(x => x.Id); + + options.Schema.For() + .Identity(x => x.Id) + .Index( + x => new { x.AppId, x.EntityKind, x.EntityId }, + x => x.Name = "idx_app_feed_entity"); + + options.Schema.For() + .Identity(x => x.Id) + .Index( + x => new { x.AppId, x.Generation, x.SourceSequence, x.Ordinal }, + x => x.Name = "idx_app_feed_cursor") + .Index(x => x.RecordedAt, x => x.Name = "idx_app_feed_recorded"); + + return options; + } +} diff --git a/src/dotnet/Modgud.Infrastructure/Persistence/Marten/Configuration/MartenConfiguration.cs b/src/dotnet/Modgud.Infrastructure/Persistence/Marten/Configuration/MartenConfiguration.cs index 88e907b1..ae1e6b91 100644 --- a/src/dotnet/Modgud.Infrastructure/Persistence/Marten/Configuration/MartenConfiguration.cs +++ b/src/dotnet/Modgud.Infrastructure/Persistence/Marten/Configuration/MartenConfiguration.cs @@ -5,6 +5,7 @@ using Modgud.Domain.Common; using Modgud.Domain.Users.Events; using Weasel.Core; +using Modgud.Infrastructure.ChangeFeed; namespace Modgud.Infrastructure.Persistence.Marten.Configuration; @@ -47,6 +48,7 @@ public static void ConfigureDocumentStore(this StoreOptions options) // UseSystemTextJsonForSerialization above so the existing configured serializer // is extended (not replaced). options.UseModgudAuthorization(); + options.UseAppChangeFeed(); // Per-tenant RealmSigningKey storage. Keys live IN the tenant DB (not // the master DB) so a master-DB compromise — or a Realm registry leak diff --git a/src/dotnet/Modgud.Tests.Unit/Authorization/PrincipalProjectionRebuildSafetyTests.cs b/src/dotnet/Modgud.Tests.Unit/Authorization/PrincipalProjectionRebuildSafetyTests.cs index 50aed6b5..e87180d7 100644 --- a/src/dotnet/Modgud.Tests.Unit/Authorization/PrincipalProjectionRebuildSafetyTests.cs +++ b/src/dotnet/Modgud.Tests.Unit/Authorization/PrincipalProjectionRebuildSafetyTests.cs @@ -14,6 +14,8 @@ public void Shared_table_projections_disable_destructive_marten_teardown() { Assert.False(new PersonProjection().Options.TeardownDataOnRebuild); Assert.False(new GroupProjection().Options.TeardownDataOnRebuild); + Assert.False(new PositionPrincipalProjection().Options.TeardownDataOnRebuild); + Assert.False(new ServiceAccountProjection().Options.TeardownDataOnRebuild); } [Fact] @@ -44,11 +46,33 @@ public void Explicit_constructors_keep_complete_event_allow_lists() typeof(GroupDeletedEvent), ]; + var positionProjection = new PositionPrincipalProjection(); + Type[] expectedPositionEvents = + [ + typeof(PositionPrincipalCreatedEvent), + typeof(PositionPrincipalUpdatedEvent), + typeof(PositionPrincipalDeletedEvent), + ]; + + var serviceAccountProjection = new ServiceAccountProjection(); + Type[] expectedServiceAccountEvents = + [ + typeof(ServiceAccountCreatedEvent), + typeof(ServiceAccountUpdatedEvent), + typeof(ServiceAccountDeletedEvent), + ]; + Assert.Equal( expectedPersonEvents.OrderBy(type => type.FullName), personProjection.IncludedEventTypes.OrderBy(type => type.FullName)); Assert.Equal( expectedGroupEvents.OrderBy(type => type.FullName), groupProjection.IncludedEventTypes.OrderBy(type => type.FullName)); + Assert.Equal( + expectedPositionEvents.OrderBy(type => type.FullName), + positionProjection.IncludedEventTypes.OrderBy(type => type.FullName)); + Assert.Equal( + expectedServiceAccountEvents.OrderBy(type => type.FullName), + serviceAccountProjection.IncludedEventTypes.OrderBy(type => type.FullName)); } } diff --git a/src/dotnet/Modgud.Tests.Unit/Authorization/Principals/ServiceAccountProjectionTests.cs b/src/dotnet/Modgud.Tests.Unit/Authorization/Principals/ServiceAccountProjectionTests.cs new file mode 100644 index 00000000..c399248c --- /dev/null +++ b/src/dotnet/Modgud.Tests.Unit/Authorization/Principals/ServiceAccountProjectionTests.cs @@ -0,0 +1,54 @@ +using Modgud.Authorization.Events; +using Modgud.Authorization.Principals; +using Modgud.Authorization.Projections; + +namespace Modgud.Tests.Unit.Authorization.Principals; + +public class ServiceAccountProjectionTests +{ + [Fact] + public void Full_lifecycle_replays_to_soft_deleted_inactive_state() + { + var id = Guid.NewGuid(); + var projection = new ServiceAccountProjection(); + + var account = projection.Apply( + new ServiceAccountCreatedEvent(id, "sync-agent", "Initial import", true), + new ServiceAccount()); + projection.Apply( + new ServiceAccountUpdatedEvent(id, "sync-agent-v2", "Change feed", true), + account); + projection.Apply(new ServiceAccountDeletedEvent(id), account); + + Assert.Equal(id, account.Id); + Assert.Equal("sync-agent-v2", account.AccountName); + Assert.Equal("Change feed", account.Purpose); + Assert.False(account.IsActive); + Assert.True(account.IsDeleted); + } + + [Fact] + public void Creation_replaces_a_legacy_snapshot_during_teardown_free_rebuild() + { + var id = Guid.NewGuid(); + var legacy = new ServiceAccount + { + Id = id, + AccountName = "stale-name", + Purpose = "stale-purpose", + IsActive = false, + IsDeleted = true, + }; + + var rebuilt = new ServiceAccountProjection().Apply( + new ServiceAccountCreatedEvent(id, "canonical-name", null, true), + legacy); + + Assert.NotSame(legacy, rebuilt); + Assert.Equal(id, rebuilt.Id); + Assert.Equal("canonical-name", rebuilt.AccountName); + Assert.Null(rebuilt.Purpose); + Assert.True(rebuilt.IsActive); + Assert.False(rebuilt.IsDeleted); + } +} diff --git a/src/dotnet/Modgud.Tests.Unit/ChangeFeed/AppChangeFeedCursorTests.cs b/src/dotnet/Modgud.Tests.Unit/ChangeFeed/AppChangeFeedCursorTests.cs new file mode 100644 index 00000000..9cb52d3c --- /dev/null +++ b/src/dotnet/Modgud.Tests.Unit/ChangeFeed/AppChangeFeedCursorTests.cs @@ -0,0 +1,64 @@ +using Modgud.Api.Features.ChangeFeed; +using Modgud.Infrastructure.ChangeFeed; + +namespace Modgud.Tests.Unit.ChangeFeed; + +public class AppChangeFeedCursorTests +{ + [Fact] + public void Cursor_round_trips_all_boundary_fields() + { + var appId = Guid.NewGuid(); + var encoded = AppChangeFeedCursor.Encode(appId, 3, 9_876_543_210, 42); + + var parsed = AppChangeFeedCursor.TryDecode(encoded, out var cursor); + + Assert.True(parsed); + Assert.Equal(appId, cursor.AppId); + Assert.Equal(3, cursor.Generation); + Assert.Equal(9_876_543_210, cursor.Sequence); + Assert.Equal(42, cursor.Ordinal); + } + + [Fact] + public void Checkpoint_uses_the_state_high_water_and_max_ordinal() + { + var state = new AppChangeFeedState + { + Id = Guid.NewGuid(), + Generation = 7, + LastProcessedSequence = 1234, + }; + + Assert.True(AppChangeFeedCursor.TryDecode( + AppChangeFeedCursor.EncodeCheckpoint(state), out var cursor)); + Assert.Equal(state.Id, cursor.AppId); + Assert.Equal(7, cursor.Generation); + Assert.Equal(1234, cursor.Sequence); + Assert.Equal(int.MaxValue, cursor.Ordinal); + } + + [Theory] + [InlineData("")] + [InlineData("not-base64!")] + [InlineData("AQ")] + public void Malformed_cursors_are_rejected_without_throwing(string value) + { + Assert.False(AppChangeFeedCursor.TryDecode(value, out _)); + } + + [Theory] + [InlineData(0, 1, 0)] + [InlineData(1, -1, 0)] + [InlineData(1, 1, -2)] + public void Semantically_invalid_cursors_are_rejected( + int generation, + long sequence, + int ordinal) + { + var encoded = AppChangeFeedCursor.Encode( + Guid.NewGuid(), generation, sequence, ordinal); + + Assert.False(AppChangeFeedCursor.TryDecode(encoded, out _)); + } +} diff --git a/src/frontend-vue/src/models/application.ts b/src/frontend-vue/src/models/application.ts index fb1e3c61..63de3cad 100644 --- a/src/frontend-vue/src/models/application.ts +++ b/src/frontend-vue/src/models/application.ts @@ -141,6 +141,12 @@ export interface ApplicationRegistrationFieldsOverrideDto { Lastname?: string | null } +export interface ApplicationChangeFeedDto { + Enabled: boolean + MinimumRetentionAgeDays: number + MinimumEventCount: number +} + export interface ApplicationSettingsDto { Origin?: ApplicationOriginDto | null Branding?: ApplicationBrandingSettingsDto | null @@ -153,4 +159,5 @@ export interface ApplicationSettingsDto { Dcr?: ApplicationDcrOverrideDto | null Cimd?: ApplicationGrantOverrideDto | null RegistrationFields?: ApplicationRegistrationFieldsOverrideDto | null + ChangeFeed?: ApplicationChangeFeedDto | null } diff --git a/src/frontend-vue/src/views/admin/apps/AppSettingsSections.vue b/src/frontend-vue/src/views/admin/apps/AppSettingsSections.vue index ffb3cc11..86aa014c 100644 --- a/src/frontend-vue/src/views/admin/apps/AppSettingsSections.vue +++ b/src/frontend-vue/src/views/admin/apps/AppSettingsSections.vue @@ -2,7 +2,7 @@ import { computed, onMounted, reactive, ref, watch } from 'vue' import { CoarNotice, - CoarTextInput, CoarFormField, CoarCheckbox, CoarSelect, CoarButton, useDialog, + CoarTextInput, CoarNumberInput, CoarFormField, CoarCheckbox, CoarSelect, CoarButton, useDialog, CoarTabGroup, CoarTab, CoarMultiSelect, } from '@cocoar/vue-ui' import { useI18n } from '@cocoar/vue-localization' @@ -34,7 +34,7 @@ const props = defineProps<{ const groupStore = useGroupStore() const loginProviderStore = useLoginProviderStore() const appConfig = useAppConfigStore() -const activeTab = ref<'origin' | 'registration' | 'sessions' | 'grants' | 'oauth' | 'pages'>('origin') +const activeTab = ref<'origin' | 'registration' | 'sessions' | 'grants' | 'oauth' | 'sync' | 'pages'>('origin') const groupOptions = ref<{ value: string; label: string }[]>([]) const loginProviderOptions = ref<{ value: string; label: string }[]>([]) @@ -80,6 +80,11 @@ const f = reactive({ reservedNames: [] as string[], perIp: '', perRealm: '', }, cimd: { override: false, enabled: false, access: '', refresh: '' }, + changeFeed: { + enabled: false, + retentionAgeDays: 7 as number | null, + minimumEventCount: 1000 as number | null, + }, }) const postureOptions = [ @@ -219,6 +224,7 @@ function resetForm() { f.dcr.override = false; f.dcr.enabled = false; f.dcr.access = ''; f.dcr.refresh = '' f.dcr.reservedNames = []; f.dcr.perIp = ''; f.dcr.perRealm = '' f.cimd.override = false; f.cimd.enabled = false; f.cimd.access = ''; f.cimd.refresh = '' + f.changeFeed.enabled = false; f.changeFeed.retentionAgeDays = 7; f.changeFeed.minimumEventCount = 1000 } function populate(s?: ApplicationSettingsDto | null) { @@ -304,6 +310,11 @@ function populate(s?: ApplicationSettingsDto | null) { f.cimd.access = numStr(s.Cimd.AccessTokenLifetimeMinutes) f.cimd.refresh = numStr(s.Cimd.RefreshTokenLifetimeDays) } + if (s.ChangeFeed) { + f.changeFeed.enabled = s.ChangeFeed.Enabled + f.changeFeed.retentionAgeDays = s.ChangeFeed.MinimumRetentionAgeDays ?? 7 + f.changeFeed.minimumEventCount = s.ChangeFeed.MinimumEventCount ?? 1000 + } } const orderedLoginProviders = computed(() => f.loginExperience.providerIds.map((id) => ({ @@ -457,6 +468,11 @@ function build(): ApplicationSettingsDto { Cimd: f.cimd.override ? { Enabled: f.cimd.enabled, AccessTokenLifetimeMinutes: parseNum(f.cimd.access), RefreshTokenLifetimeDays: parseNum(f.cimd.refresh) } : null, + ChangeFeed: { + Enabled: f.changeFeed.enabled, + MinimumRetentionAgeDays: f.changeFeed.retentionAgeDays ?? 7, + MinimumEventCount: f.changeFeed.minimumEventCount ?? 1000, + }, } } @@ -545,6 +561,7 @@ watch(() => [activeTab.value, props.applicationId] as const, ([tab]) => { {{ t('admin.appSettings.tabs.sessions', {}, 'Sessions') }} {{ t('admin.appSettings.tabs.grants', {}, 'Native Grants') }} {{ t('admin.appSettings.tabs.oauth', {}, 'OAuth (DCR/CIMD)') }} + {{ t('admin.appSettings.tabs.sync', {}, 'Sync') }} {{ t('admin.appSettings.tabs.pages', {}, 'Pages') }} @@ -806,6 +823,41 @@ watch(() => [activeTab.value, props.applicationId] as const, ([tab]) => { + +
+ + {{ t('admin.appSettings.changeFeed.hintShort', {}, 'Expose this app\'s current scope through a resumable consumer change feed.') }} + + + +
+ + + + + + +
+
+