Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Adds support for tenantId to MartenOps #1091

Draft
wants to merge 1 commit into
base: main
Choose a base branch
from
+340 −29
Draft

Adds support for tenantId to MartenOps #1091

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
150 changes: 150 additions & 0 deletions ...tence/MartenTests/handler_actions_with_implied_marten_operations_with_tenant_switching.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,150 @@
using IntegrationTests;
using Marten;
using Marten.Events;
using Marten.Exceptions;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Shouldly;
using Wolverine;
using Wolverine.Marten;
using Wolverine.Tracking;

namespace MartenTests;

public class handler_actions_with_implied_marten_operations_with_tenant_switching : PostgresqlContext, IAsyncLifetime
{
private IHost _host;
private IDocumentStore _store;

public async Task InitializeAsync()
{
_host = await Host.CreateDefaultBuilder()
.UseWolverine(opts =>
{
opts.Services
.AddMarten(o =>
{
o.Connection(Servers.PostgresConnectionString);
o.Policies.AllDocumentsAreMultiTenanted();
o.Events.StreamIdentity = StreamIdentity.AsString;
})
.IntegrateWithWolverine();


opts.Policies.AutoApplyTransactions();
}).StartAsync();

_store = _host.Services.GetRequiredService<IDocumentStore>();

await _store.Advanced.Clean.DeleteDocumentsByTypeAsync(typeof(TenantNamedDocument));
}

public async Task DisposeAsync()
{
await _host.StopAsync();
_host.Dispose();
}

[Fact]
public async Task storing_document()
{
var tracked = await _host.InvokeMessageAndWaitAsync(new TenantCreateMartenDocument("Aubrey", "green"));

tracked.Sent.MessagesOf<StoreDoc<TenantNamedDocument>>().ShouldHaveNoMessages();
tracked.Sent.SingleMessage<TenantMartenMessage2>().Name.ShouldBe("Aubrey");

using var session = _store.LightweightSession("green");
var doc = await session.LoadAsync<TenantNamedDocument>("Aubrey");
doc.ShouldNotBeNull();
}

[Fact]
public async Task insert_document()
{
await _host.InvokeMessageAndWaitAsync(new TenantInsertMartenDocument("Declan", "green"));

using var session = _store.LightweightSession("green");
var doc = await session.LoadAsync<TenantNamedDocument>("Declan");
doc.ShouldNotBeNull();

await Should.ThrowAsync<DocumentAlreadyExistsException>(() =>
_host.InvokeMessageAndWaitAsync(new TenantInsertMartenDocument("Declan", "green")));
}

[Fact]
public async Task update_document_happy_path()
{
await _host.InvokeMessageAndWaitAsync(new TenantInsertMartenDocument("Max", "green"));
await _host.InvokeMessageAndWaitAsync(new TenantUpdateMartenDocument("Max", 10, "green"));


using var session = _store.LightweightSession("green");
var doc = await session.LoadAsync<TenantNamedDocument>("Max");
doc.Number.ShouldBe(10);


}

[Fact]
public async Task update_document_sad_path()
{
await Should.ThrowAsync<NonExistentDocumentException>(() =>
_host.InvokeMessageAndWaitAsync(new TenantUpdateMartenDocument("Max", 10, "green")));
}

[Fact]
public async Task delete_document()
{
await _host.InvokeMessageAndWaitAsync(new TenantInsertMartenDocument("Max", "green"));
await _host.InvokeMessageAndWaitAsync(new TenantDeleteMartenDocument("Max", "green"));

using var session = _store.LightweightSession("green");
var doc = await session.LoadAsync<TenantNamedDocument>("Max");
doc.ShouldBeNull();
}

}


public record TenantCreateMartenDocument(string Name, string TenantId);
public record TenantInsertMartenDocument(string Name, string TenantId);
public record TenantUpdateMartenDocument(string Name, int Number, string TenantId);
public record TenantDeleteMartenDocument(string Name, string TenantId);

public record TenantMartenMessage2(string Name, string TenantId);

public static class TenantMartenCommandHandler
{
public static (TenantMartenMessage2, DocumentOp) Handle(TenantCreateMartenDocument command)
{
return (new TenantMartenMessage2(command.Name, command.TenantId), MartenOps.Store(new TenantNamedDocument { Id = command.Name }, command.TenantId));
}

public static DocumentOp Handle(TenantInsertMartenDocument command)
{
return MartenOps.Insert(new TenantNamedDocument { Id = command.Name }, command.TenantId);
}

public static async Task<DocumentOp> Handle(TenantUpdateMartenDocument command, IDocumentSession session)
{
return MartenOps.Update(new TenantNamedDocument{Id = command.Name, Number = command.Number}, command.TenantId);
}

public static async Task<DocumentOp> Handle(TenantDeleteMartenDocument command, IDocumentSession session)
{
var doc = await session.ForTenant(command.TenantId).LoadAsync<TenantNamedDocument>(command.Name);

return MartenOps.Delete(doc, command.TenantId);
}

public static void Handle(TenantMartenMessage2 message)
{
// Nothing yet
}
}

public class TenantNamedDocument
{
public string Id { get; set; }
public int Number { get; set; }
}
114 changes: 112 additions & 2 deletions src/Persistence/MartenTests/handler_actions_with_returned_StartStream.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
using IntegrationTests;
using Marten;
using Marten.Events;
using Marten.Storage;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Oakton.Resources;
Expand Down Expand Up @@ -55,6 +56,57 @@ public async Task start_stream_by_guid1()
events[1].Data.ShouldBeOfType<BEvent>();
}
}
public class handler_actions_with_returned_StartStream_with_tenant_switching : PostgresqlContext, IAsyncLifetime
{
private IHost _host;
private IDocumentStore _store;

public async Task InitializeAsync()
{
_host = await Host.CreateDefaultBuilder()
.UseWolverine(opts =>
{
opts.Services
.AddMarten(o =>
{
o.Connection(Servers.PostgresConnectionString);
o.Policies.AllDocumentsAreMultiTenanted();
o.Events.TenancyStyle = TenancyStyle.Conjoined;
o.DatabaseSchemaName = "martenops_events_guid";
})
.IntegrateWithWolverine();

opts.Policies.AutoApplyTransactions();

opts.Services.AddResourceSetupOnStartup();
}).StartAsync();

_store = _host.Services.GetRequiredService<IDocumentStore>();

await _store.Advanced.Clean.DeleteDocumentsByTypeAsync(typeof(NamedDocument));
await _store.Advanced.Clean.DeleteAllEventDataAsync();
}

public async Task DisposeAsync()
{
await _host.StopAsync();
_host.Dispose();
}

[Fact]
public async Task start_stream_by_guid1()
{
var id = Guid.NewGuid();

await _host.InvokeMessageAndWaitAsync(new StartStreamMessage(id, "green"));

using var session = _store.LightweightSession("green");
var events = await session.Events.FetchStreamAsync(id);
events.Count.ShouldBe(2);
events[0].Data.ShouldBeOfType<AEvent>();
events[1].Data.ShouldBeOfType<BEvent>();
}
}

public class start_stream_by_string_from_return_value : PostgresqlContext, IAsyncLifetime
{
Expand Down Expand Up @@ -106,18 +158,76 @@ public async Task start_stream_by_string()
}
}

public record StartStreamMessage(Guid Id);
public record StartStreamMessage2(string Id);
public class start_stream_by_string_from_return_value_with_tenant_switching : PostgresqlContext, IAsyncLifetime
{
private IHost _host;
private IDocumentStore _store;

public async Task InitializeAsync()
{
_host = await Host.CreateDefaultBuilder()
.UseWolverine(opts =>
{
opts.Services
.AddMarten(m =>
{
m.Connection(Servers.PostgresConnectionString);
m.Policies.AllDocumentsAreMultiTenanted();
m.Events.StreamIdentity = StreamIdentity.AsString;
m.DatabaseSchemaName = "martenops_string_identity";
m.Events.TenancyStyle = TenancyStyle.Conjoined;
})
.IntegrateWithWolverine();

opts.Policies.AutoApplyTransactions();

opts.Services.AddResourceSetupOnStartup();
}).StartAsync();

_store = _host.Services.GetRequiredService<IDocumentStore>();

await _store.Advanced.Clean.DeleteDocumentsByTypeAsync(typeof(NamedDocument));
}

public async Task DisposeAsync()
{
await _host.StopAsync();
_host.Dispose();
}

[Fact]
public async Task start_stream_by_string()
{
var id = Guid.NewGuid().ToString();

await _host.InvokeMessageAndWaitAsync(new StartStreamMessage2(id, "green"));

using var session = _store.LightweightSession("green");
var events = await session.Events.FetchStreamAsync(id);
events.Count.ShouldBe(2);
events[0].Data.ShouldBeOfType<CEvent>();
events[1].Data.ShouldBeOfType<BEvent>();
}
}

public record StartStreamMessage(Guid Id, string? TenantId = null);
public record StartStreamMessage2(string Id, string? TenantId = null);

public static class StartStreamMessageHandler
{
public static IStartStream Handle(StartStreamMessage message)
{
if (message.TenantId is not null)
return MartenOps.StartStream<NamedDocument>(message.Id, message.TenantId, new AEvent(), new BEvent());

return MartenOps.StartStream<NamedDocument>(message.Id, new AEvent(), new BEvent());
}

public static IStartStream Handle(StartStreamMessage2 message)
{
if (message.TenantId is not null)
return MartenOps.StartStream<NamedDocument>(message.Id, message.TenantId, new CEvent(), new BEvent());

return MartenOps.StartStream<NamedDocument>(message.Id, new CEvent(), new BEvent());
}
}
Loading
Loading