Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 21 additions & 1 deletion .agents/skills/aquila/SKILL.md
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@ Aquila decouples business domain semantics (sessions, units-of-work, aggregates,

### Storage SPI Contracts
- [`IDocumentStorageProvider`](file:///home/chad/source/dotnet/Aquila/src/Aquila.Core/Storage/StorageContracts.cs#L78): Atomic reads, queries, upserts, deletes, and batch execution of `StorageOperation`s.
- [`IEventStorageProvider`](file:///home/chad/source/dotnet/Aquila/src/Aquila.Core/Storage/StorageContracts.cs#L92): Append stream events, fetch streams/global sequences, get stream headers, save/get aggregate snapshots.
- [`IEventStorageProvider`](file:///home/chad/source/dotnet/Aquila/src/Aquila.Core/Storage/StorageContracts.cs#L92): Append stream events, fetch streams/global sequences/by-tag, get stream headers, save/get aggregate snapshots.
- [`IProjectionStorageProvider`](file:///home/chad/source/dotnet/Aquila/src/Aquila.Core/Storage/StorageContracts.cs#L153): Materialized read models, point views, high-throughput batch updates, and native instantaneous zero-RU rebuilds ([`PurgeProjectionAsync`](file:///home/chad/source/dotnet/Aquila/src/Aquila.Core/Storage/StorageContracts.cs#L159)).
- [`IProjectionCheckpointStore`](file:///home/chad/source/dotnet/Aquila/src/Aquila.Core/Projections/Daemon/IProjectionCheckpointStore.cs): Durable checkpoint sequence persistence for async projection daemons.

Expand Down Expand Up @@ -222,6 +222,26 @@ catch (AquilaConcurrencyException ex)
}
```

### Event Tagging
Tag individual events with `TaggedEvent`, then query across all streams by tag via `FetchEventsByTagAsync` — independent of `Apply`-based aggregate rehydration:
```csharp
using Aquila.Core.Events;

session.Events.StartStreamTagged<OrderAggregate>(streamId,
[
new TaggedEvent(new OrderPlaced(streamId, "CUST-1", 150.00m), tags: ["audit"])
]);
session.Events.AppendTagged(streamId, expectedVersion: 1,
[
new TaggedEvent(new ItemAdded(streamId, "SKU-99", 25.00m), tags: ["audit", "large-item"])
]);
await session.SaveChangesAsync();

// Global-sequence stream of every "audit"-tagged event, across all streams
IReadOnlyList<IEvent> auditEvents = await session.Events.FetchEventsByTagAsync("audit");
```
Untagged `StartStream`/`Append` overloads still work unchanged and produce events with an empty `Tags` set.

### Aggregate Rehydration
Aggregates rehydrate their state by declaring public or internal `Apply(TEvent)` methods:

Expand Down
3 changes: 2 additions & 1 deletion .github/workflows/build-and-test.yml
Original file line number Diff line number Diff line change
Expand Up @@ -39,13 +39,14 @@ jobs:
- name: Build Solution
run: dotnet build Aquila.slnx --configuration Release --no-restore

- name: Run Unit & Integration Tests
- name: Run Unit Tests
run: >
dotnet test Aquila.slnx
--configuration Release
--no-build
--verbosity normal
--settings codecoverage.runsettings
--filter "Category!=Integration"
--collect:"XPlat Code Coverage"
--results-directory ./TestResults
--logger "trx;LogFileName=test-results.trx"
Expand Down
3 changes: 2 additions & 1 deletion ARCHITECTURE.md
Original file line number Diff line number Diff line change
Expand Up @@ -58,7 +58,7 @@ Aquila/

Aquila decouples business domain semantics (sessions, units-of-work, aggregates, and projections) from physical storage engines using a **Tripartite Polyglot Storage Architecture** comprising three independent, first-class SPI contracts defined in [`StorageContracts.cs`](file:///home/chad/source/dotnet/Aquila/src/Aquila.Core/Storage/StorageContracts.cs):

1. **`IEventStorageProvider`**: Append-only event streams, aggregate rehydration, global sequence streaming, and aggregate snapshots (e.g. Azure Cosmos DB, In-Memory).
1. **`IEventStorageProvider`**: Append-only event streams, aggregate rehydration, global sequence streaming, per-event tagging (`FetchEventsByTagAsync`), and aggregate snapshots (e.g. Azure Cosmos DB, In-Memory).
2. **`IDocumentStorageProvider`**: Primary domain documents, dirty tracking, units of work, optimistic concurrency, and LINQ querying (e.g. Azure Cosmos DB, Redis, In-Memory).
3. **`IProjectionStorageProvider`**: Materialized read models, point views, high-throughput batch updates, native instantaneous zero-RU rebuilds ([`PurgeProjectionAsync`](file:///home/chad/source/dotnet/Aquila/src/Aquila.Core/Storage/StorageContracts.cs#L159)), and ultra-low latency reads (e.g. **Redis**, dedicated Cosmos DB read containers).

Expand All @@ -77,6 +77,7 @@ classDiagram
+string ProviderName
+AppendEventsAsync(streamId, events, expectedVersion) Task
+FetchEventsAsync(streamId, tenantId, fromVersion) Task~IReadOnlyList~IEvent~~
+FetchEventsByTagAsync(tag, fromGlobalSequence, batchSize, tenantId) Task~IReadOnlyList~IEvent~~
+GetStreamHeaderAsync(streamId, tenantId) Task~EventStreamHeader~
+SaveSnapshotAsync~TAggregate~(streamId, version, snapshot) Task
+GetSnapshotAsync~TAggregate~(streamId, tenantId) Task
Expand Down
1 change: 1 addition & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
- 🔀 **$O(1)$ Zero-Allocation Type Routing**: Unified session APIs (`session.LoadAsync<T>()`, `session.QueryAsync<T>()`, `session.Store<T>()`) automatically route read models to `ProjectionStorage` and domain documents to `DocumentStorage` via an immutable `FrozenSet<Type>` registry compiled on store freeze.
- ⚡ **1-RU Point Reads**: High-efficiency point reads (`LoadAsync<T>`) executing direct `ReadItemAsync` operations on Cosmos DB (~1 RU) or sub-millisecond string gets on Redis.
- 📜 **Event Sourcing & CQRS**: First-class stream append operations (`StartStream`, `Append`), expected version concurrency checks, stream fetching (`FetchStreamAsync`), and aggregate rehydration (`AggregateStreamAsync`).
- 🏷️ **Event Tagging**: Attach arbitrary string tags to individual events via [`TaggedEvent`](file:///home/chad/source/dotnet/Aquila/src/Aquila.Core/Events/TaggedEvent.cs) (`StartStreamTagged`, `AppendTagged`), then stream every event carrying a given tag across all streams in global sequence order via `FetchEventsByTagAsync` (Cosmos DB, In-Memory).
- 🔁 **Event Upcasting & Snapshotting**: Transparent schema evolution via [`IEventUpcaster`](file:///home/chad/source/dotnet/Aquila/src/Aquila.Core/Events/IEventUpcaster.cs) chains, plus [`ISnapshotStrategy<TAggregate>`](file:///home/chad/source/dotnet/Aquila/src/Aquila.Core/Events/ISnapshotStrategy.cs)-driven aggregate snapshots to avoid full-stream replay on rehydration.
- 📊 **Projections**: Read-model generation via [`SingleStreamProjection<TAggregate>`](file:///home/chad/source/dotnet/Aquila/src/Aquila.Core/Projections/SingleStreamProjection.cs) and [`MultiStreamProjection<TDoc,TId>`](file:///home/chad/source/dotnet/Aquila/src/Aquila.Core/Projections/MultiStreamProjection.cs), offering `Inline` (transaction-scoped for mono-stores), `Async` (background daemon), and `Live` (on-the-fly, unpersisted) execution lifecycles.
- 🛰️ **Async Projection Daemon & Zero-Downtime Rebuilds**: A background [`IProjectionDaemon`](file:///home/chad/source/dotnet/Aquila/src/Aquila.Core/Projections/Daemon/IProjectionDaemon.cs) with durable checkpointing ([`IProjectionCheckpointStore`](file:///home/chad/source/dotnet/Aquila/src/Aquila.Core/Projections/Daemon/IProjectionCheckpointStore.cs)), `CatchUpAsync()`, and zero-downtime `RebuildProjectionAsync()` with instant key purging — plus a Cosmos DB Change Feed-aware variant ([`CosmosProjectionDaemon`](file:///home/chad/source/dotnet/Aquila/src/Aquila.Cosmos/Projections/CosmosProjectionDaemon.cs)) via `AddCosmosDaemon()`.
Expand Down
32 changes: 31 additions & 1 deletion USAGE.md
Original file line number Diff line number Diff line change
Expand Up @@ -142,7 +142,37 @@ IReadOnlyList<IEvent> allEvents = await session.Events.FetchStreamAsync(streamId
IReadOnlyList<IEvent> partialEvents = await session.Events.FetchStreamAsync(streamId, fromVersion: 2);
```

### 4. Aggregate Rehydration
### 4. Tagging Events

Individual events can carry a set of string tags, independent of the stream they belong to, via [`TaggedEvent`](file:///home/chad/source/dotnet/Aquila/src/Aquila.Core/Events/TaggedEvent.cs). Use the `*Tagged` counterparts of `StartStream`/`Append` to attach tags at write time, and [`FetchEventsByTagAsync`](file:///home/chad/source/dotnet/Aquila/src/Aquila.Core/Abstractions/IDocumentStore.cs#L45) to stream every event carrying a given tag across the global sequence, regardless of which stream it came from.

```csharp
using Aquila.Core.Events;

using var session = store.OpenSession();

// Start a stream with per-event tags
session.Events.StartStreamTagged<AccountAggregate>(streamId,
[
new TaggedEvent(new AccountOpened(streamId, Owner: "Alice", InitialBalance: 500.00m), tags: ["audit", "onboarding"]),
new TaggedEvent(new MoneyDeposited(streamId, Amount: 200.00m), tags: ["audit"])
]);

// Append a tagged event to an existing stream (with optional expected version)
session.Events.AppendTagged(streamId, expectedVersion: 2,
[
new TaggedEvent(new MoneyWithdrawn(streamId, Amount: 50.00m), tags: ["audit", "large-withdrawal"])
]);

await session.SaveChangesAsync();

// Read every "audit"-tagged event across all streams, in global sequence order
IReadOnlyList<IEvent> auditEvents = await session.Events.FetchEventsByTagAsync("audit");
```

Every [`IEvent`](file:///home/chad/source/dotnet/Aquila/src/Aquila.Core/Events/IEvent.cs) exposes its tags via the `Tags` property (an `IReadOnlySet<string>`, empty by default). The plain, untagged `StartStream`/`Append` overloads are unaffected and simply produce events with an empty tag set.

### 5. Aggregate Rehydration

Aggregates rehydrate their state by defining `Apply(TEvent)` methods for each domain event.

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@ public void Apply(IntegrationItemAddedEvent @event)
}

[Collection("CosmosIntegration")]
[Trait("Category", "Integration")]
public sealed class CosmosIntegrationTests
{
private readonly CosmosContainerFixture _fixture;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -167,6 +167,7 @@ public override bool Apply(IEvent @event, IntegrationCustomerSummaryReadModel do
// ─── Integration Tests ─────────────────────────────────────────────────────

[Collection("CosmosIntegration")]
[Trait("Category", "Integration")]
public sealed class CosmosMultiStreamProjectionIntegrationTests
{
private readonly CosmosContainerFixture _fixture;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -110,6 +110,7 @@ public IntegrationAsyncSingleStreamProjection()
// ─── Integration Tests ─────────────────────────────────────────────────────

[Collection("CosmosIntegration")]
[Trait("Category", "Integration")]
public sealed class CosmosSingleStreamProjectionIntegrationTests
{
private readonly CosmosContainerFixture _fixture;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -139,6 +139,7 @@ public SegregatedSpecialProjection()
// ─── Integration Tests ─────────────────────────────────────────────────────

[Collection("CosmosIntegration")]
[Trait("Category", "Integration")]
public sealed class CosmosStorageSegregationIntegrationTests
{
private readonly CosmosContainerFixture _fixture;
Expand Down
Loading