Ef core connector (#44)

* Implement provider for EF Core

* Improve naming

src/RdbmsEventStore.EFCore.Tests/EventStoreTests/ExtraMetaTests.cs

@@ -0,0 +1,120 @@
+using System;
+using System.Threading.Tasks;
+using RdbmsEventStore.EFCore.Tests.Infrastructure;
+using Xunit;
+using System.Linq;
+using Microsoft.EntityFrameworkCore;
+using RdbmsEventStore.EFCore.Tests.TestData;
+using RdbmsEventStore.EventRegistry;
+using RdbmsEventStore.Serialization;
+
+namespace RdbmsEventStore.EFCore.Tests.EventStoreTests
+{
+    public class ExtraMetaTests : IClassFixture<ExtraMetaEventFactoryFixture>
+    {
+        private readonly ExtraMetaEventFactoryFixture _fixture;
+        private readonly EFCoreEventStoreContext<string, ExtraMetaLongStringPersistedEventMetadata> _dbContext;
+
+        public ExtraMetaTests(ExtraMetaEventFactoryFixture fixture)
+        {
+            _fixture = fixture;
+            var options = new DbContextOptionsBuilder<EFCoreEventStoreContext<string, ExtraMetaLongStringPersistedEventMetadata>>()
+                .UseInMemoryDatabase(databaseName: Guid.NewGuid().ToString())
+                .Options;
+            _dbContext = new EFCoreEventStoreContext<string, ExtraMetaLongStringPersistedEventMetadata>(options);
+
+            var stream1 = _fixture.EventFactory.Create("stream-1", 0, new object[] {
+                    new FooEvent { Foo = "Foo" },
+                    new BarEvent { Bar = "Bar" },
+                    new FooEvent { Foo = "Baz" }
+                })
+                .Select(_fixture.EventSerializer.Serialize);
+            var stream2 = _fixture.EventFactory.Create("stream-2", 0, new object[] {
+                    new FooEvent { Foo = "Boo" },
+                    new BarEvent { Bar = "Far" }
+                })
+                .Select(_fixture.EventSerializer.Serialize);
+
+            _dbContext.Events.AddRange(stream1);
+            _dbContext.Events.AddRange(stream2);
+            _dbContext.SaveChanges();
+        }
+
+        [Theory]
+        [InlineData("stream-1", 3)]
+        [InlineData("stream-2", 2)]
+        public async Task ReturnsEventsFromCorrectStreamOnly(string streamId, long expectedCount)
+        {
+            var store = _fixture.BuildEventStore(_dbContext) as IEventStore<string, ExtraMetaStringEvent, IExtraMeta>;
+            var events = await store.Events(streamId);
+            Assert.Equal(expectedCount, events.Count());
+        }
+
+        [Theory]
+        [InlineData("stream-1", 2)]
+        [InlineData("stream-2", 1)]
+        public async Task ReturnsEventsAccordingToQuery(string streamId, long expectedCount)
+        {
+            var store = _fixture.BuildEventStore(_dbContext) as IEventStore<string, ExtraMetaStringEvent, IExtraMeta>;
+            var events = await store.Events(streamId, es => es.Where(e => e.ExtraMeta.StartsWith("Foo")));
+            Assert.Equal(expectedCount, events.Count());
+        }
+
+        [Theory]
+        [InlineData("stream-1", 2)]
+        [InlineData("stream-2", 1)]
+        public async Task ReturnsEventsWithMetadata(string streamId, long expectedCount)
+        {
+            var store = _fixture.BuildEventStore(_dbContext) as IEventStore<string, ExtraMetaStringEvent, IExtraMeta>;
+            var events = await store
+                .Events(streamId, es => es.Where(e => e.ExtraMeta.StartsWith("Foo")))
+                .ToReadOnlyCollection();
+
+            Assert.Equal(expectedCount, events.Count);
+            Assert.All(events, @event => Assert.StartsWith("Foo", @event.ExtraMeta));
+        }
+
+        [Theory]
+        [InlineData("stream-1", 2)]
+        [InlineData("stream-2", 1)]
+        public async Task CanQueryByExtraMetadata(string streamId, long expectedCount)
+        {
+            var store = _fixture.BuildEventStore(_dbContext) as IEventStore<string, ExtraMetaStringEvent, IExtraMeta>;
+            var events = await store.Events(streamId, es => es.Where(e => e.ExtraMeta.StartsWith("Foo")));
+            Assert.Equal(expectedCount, events.Count());
+        }
+    }
+
+    public class ExtraMetaEventFactory : DefaultEventFactory<string, ExtraMetaStringEvent>
+    {
+        private int _total;
+
+        protected override ExtraMetaStringEvent CreateSingle(string streamId, long version, object payload)
+        {
+            var @event = base.CreateSingle(streamId, version, payload);
+            @event.ExtraMeta = $"{payload.GetType().Name}-{_total++}";
+            return @event;
+        }
+    }
+
+    public class ExtraMetaEventSerializer : DefaultEventSerializer<string, ExtraMetaStringEvent, ExtraMetaLongStringPersistedEventMetadata>
+    {
+        public ExtraMetaEventSerializer(IEventRegistry registry) : base(registry)
+        {
+        }
+
+        public override ExtraMetaLongStringPersistedEventMetadata Serialize(ExtraMetaStringEvent @event)
+        {
+            var serialized = base.Serialize(@event);
+            serialized.ExtraMeta = @event.ExtraMeta;
+            return serialized;
+        }
+
+        public override ExtraMetaStringEvent Deserialize(ExtraMetaLongStringPersistedEventMetadata @event)
+        {
+            var deserialized = base.Deserialize(@event);
+            deserialized.ExtraMeta = @event.ExtraMeta;
+            return deserialized;
+        }
+    }
+}

src/RdbmsEventStore.EFCore.Tests/EventStoreTests/QueryEventsTests.cs

@@ -0,0 +1,66 @@
+using System.Linq;
+using System.Threading.Tasks;
+using RdbmsEventStore.EFCore.Tests.Infrastructure;
+using RdbmsEventStore.EFCore.Tests.TestData;
+using Xunit;
+
+namespace RdbmsEventStore.EFCore.Tests.EventStoreTests
+{
+    public class QueryEventsTests : EventStoreTestBase<long, string, StringEvent, IEventMetadata<string>, LongStringPersistedEvent>
+    {
+        public QueryEventsTests(EventStoreFixture<long, string, StringEvent, IEventMetadata<string>, LongStringPersistedEvent> fixture) : base(fixture)
+        {
+            var stream1 = _fixture.EventFactory.Create("stream-1", 0, new object[] {
+                    new FooEvent { Foo = "Foo" },
+                    new BarEvent { Bar = "Bar" },
+                    new FooEvent { Foo = "Baz" }
+                })
+                .Select(_fixture.EventSerializer.Serialize);
+            var stream2 = _fixture.EventFactory.Create("stream-2", 0, new object[] {
+                    new FooEvent { Foo = "Boo" },
+                    new BarEvent { Bar = "Far" }
+                })
+                .Select(_fixture.EventSerializer.Serialize);
+
+            _dbContext.Events.AddRange(stream1);
+            _dbContext.Events.AddRange(stream2);
+            _dbContext.SaveChanges();
+        }
+
+        [Theory]
+        [InlineData("stream-1", 3)]
+        [InlineData("stream-2", 2)]
+        public async Task ReturnsEventsFromCorrectStreamOnly(string streamId, long expectedCount)
+        {
+            var store = _fixture.BuildEventStore(_dbContext) as IEventStore<string, StringEvent, IEventMetadata<string>>;
+            var events = await store.Events(streamId);
+            Assert.Equal(expectedCount, events.Count());
+        }
+
+        [Theory]
+        [InlineData("stream-1", 2)]
+        [InlineData("stream-2", 1)]
+        public async Task ReturnsEventsAccordingToQuery(string streamId, long expectedCount)
+        {
+            var store = _fixture.BuildEventStore(_dbContext) as IEventStore<string, StringEvent, IEventMetadata<string>>;
+            var events = await store.Events(streamId, es => es.Where(e => e.Version > 1));
+            Assert.Equal(expectedCount, events.Count());
+        }
+
+        [Fact]
+        public async Task ReturnsAllEvents()
+        {
+            var store = _fixture.BuildEventStore(_dbContext) as IEventStore<string, StringEvent, IEventMetadata<string>>;
+            var events = await store.Events();
+            Assert.Equal(5, events.Count());
+        }
+
+        [Fact]
+        public async Task ReturnsAllEventsAccordingToQuery()
+        {
+            var store = _fixture.BuildEventStore(_dbContext) as IEventStore<string, StringEvent, IEventMetadata<string>>;
+            var events = await store.Events(es => es.Where(e => e.Version > 1));
+            Assert.Equal(3, events.Count());
+        }
+    }
+}

src/RdbmsEventStore.EFCore.Tests/EventStoreTests/WriteEventTests.cs

@@ -0,0 +1,99 @@
+using System;
+using System.Linq;
+using System.Threading.Tasks;
+using Microsoft.EntityFrameworkCore;
+using Moq;
+using RdbmsEventStore.EFCore.Tests.Infrastructure;
+using RdbmsEventStore.EFCore.Tests.TestData;
+using Xunit;
+
+namespace RdbmsEventStore.EFCore.Tests.EventStoreTests
+{
+    public class WriteEventTests : EventStoreTestBase<Guid, Guid, GuidEvent, IEventMetadata<Guid>, GuidGuidPersistedEvent>
+    {
+        public WriteEventTests(EventStoreFixture<Guid, Guid, GuidEvent, IEventMetadata<Guid>, GuidGuidPersistedEvent> fixture) : base(fixture)
+        {
+        }
+
+        [Fact]
+        public async Task CommittingEventStoresEventInContext()
+        {
+            var store = _fixture.BuildEventStore(_dbContext);
+            await store.Append(Guid.NewGuid(), 0, new[] { new FooEvent { Foo = "Bar" } });
+            Assert.Equal(1, await _dbContext.Events.CountAsync());
+        }
+
+        [Fact]
+        public async Task CommittingWithOutOfSyncDataThrowsConflictException()
+        {
+            var store = _fixture.BuildEventStore(_dbContext);
+            var stream = Guid.NewGuid();
+            _dbContext.Events.AddRange(_fixture.EventFactory.Create(stream, 0, new[] { new FooEvent { Foo = "Bar" } }).Select(_fixture.EventSerializer.Serialize));
+            await _dbContext.SaveChangesAsync();
+
+            await Assert.ThrowsAsync<ConflictException>(() => store.Append(stream, 0, new[] { new FooEvent { Foo = "Qux" } }));
+        }
+
+        [Fact]
+        public async Task CommittingNoEventsExitsEarly() {
+            var context = new Mock<EFCoreEventStoreContext<Guid, GuidGuidPersistedEvent>>(MockBehavior.Strict);
+            var set = new Mock<DbSet<GuidGuidPersistedEvent>>(MockBehavior.Strict);
+            context.Setup(c => c.Set<GuidGuidPersistedEvent>()).Returns(set.Object);
+            var stream = Guid.NewGuid();
+
+            var store = _fixture.BuildEventStore(context.Object);
+
+            try {
+                await store.Append(stream, 0, new object[] { });
+            } catch (NotImplementedException) {
+                // Thrown by the mock DbSet if we try to query for existing events
+                // This indicates that we didn't exit early
+
+                Assert.False(true, "Expected to exit early, but apparently didn't.");
+            }
+        }
+
+        [Fact]
+        public async Task CommittingMultipleEventsStoresAllEventsInContext()
+        {
+            Assert.Empty(await _dbContext.Events.ToListAsync());
+
+            var store = _fixture.BuildEventStore(_dbContext);
+
+            var events = new[] { new FooEvent { Foo = "Foo" }, new FooEvent { Foo = "Bar" } };
+            await store.Append(Guid.NewGuid(), 0, events);
+
+            Assert.Equal(2, await _dbContext.Events.CountAsync());
+        }
+
+        [Fact]
+        public async Task CommittingMultipleEventsStoresEventsInOrder()
+        {
+            Assert.Empty(await _dbContext.Events.ToListAsync());
+
+            var store = _fixture.BuildEventStore(_dbContext);
+
+            var events = new object[] { new FooEvent { Foo = "Foo" }, new BarEvent { Bar = "Bar" } };
+            await store.Append(Guid.NewGuid(), 0, events);
+
+            Assert.Collection(await _dbContext.Events.OrderBy(e => e.Version).ToListAsync(),
+                foo => Assert.Equal(typeof(FooEvent), _fixture.EventRegistry.TypeFor(foo.Type)),
+                bar => Assert.Equal(typeof(BarEvent), _fixture.EventRegistry.TypeFor(bar.Type)));
+        }
+
+        [Fact]
+        public async Task CommittingMultipleEventsIncrementsVersionForEachEvent()
+        {
+            Assert.Empty(await _dbContext.Events.ToListAsync());
+
+            var store = _fixture.BuildEventStore(_dbContext);
+            var events = new object[] { new FooEvent { Foo = "Foo" }, new BarEvent { Bar = "Bar" } };
+            await store.Append(Guid.NewGuid(), 0, events);
+
+            var storedEvents = await _dbContext.Events.OrderBy(e => e.Timestamp).ToListAsync();
+            Assert.Collection(storedEvents,
+                foo => Assert.Equal(1, foo.Version),
+                bar => Assert.Equal(2, bar.Version));
+        }
+    }
+}

src/RdbmsEventStore.EFCore.Tests/ExtensibilityTests/NonDefaultImplementationsTests.cs

@@ -0,0 +1,99 @@
+using System;
+using System.ComponentModel.DataAnnotations;
+using System.ComponentModel.DataAnnotations.Schema;
+using System.Text;
+using System.Threading.Tasks;
+using Microsoft.EntityFrameworkCore;
+using RdbmsEventStore.EFCore.Tests.Infrastructure;
+using RdbmsEventStore.EFCore.Tests.TestData;
+using RdbmsEventStore.Serialization;
+using Xunit;
+
+namespace RdbmsEventStore.EFCore.Tests.ExtensibilityTests
+{
+    public class NonDefaultEvent : IMutableEvent<long>
+    {
+        public DateTimeOffset Timestamp { get; set; }
+        public long StreamId { get; set; }
+        public long Version { get; set; }
+        public Type Type { get; set; }
+        public object Payload { get; set; }
+    }
+
+    public class NonDefaultPersistedEvent : IPersistedEvent<long>
+    {
+        [Key]
+        [DatabaseGenerated(DatabaseGeneratedOption.Identity)]
+        public long EventId { get; set; }
+        [Required]
+        public long StreamId { get; set; }
+        [Required]
+        public DateTimeOffset Timestamp { get; set; }
+        [Required]
+        public long Version { get; set; }
+        [Required]
+        public string Type { get; set; }
+        [Required]
+        public byte[] Payload { get; set; }
+    }
+
+    public class NonDefaultContext : DbContext, IEFCoreEventStoreContext<NonDefaultPersistedEvent>
+    {
+        public NonDefaultContext(DbContextOptions<NonDefaultContext> options) : base(options)
+        {
+        }
+
+        public DbSet<NonDefaultPersistedEvent> Events { get; set; }
+    }
+
+    public class NonDefaultImplementationsTests : IClassFixture<EventStoreFixture<long, long, NonDefaultEvent, IEventMetadata<long>, NonDefaultPersistedEvent>>, IDisposable
+    {
+        private readonly EventStoreFixture<long, long, NonDefaultEvent, IEventMetadata<long>, NonDefaultPersistedEvent> _fixture;
+        private readonly NonDefaultContext _dbContext;
+
+        public NonDefaultImplementationsTests(EventStoreFixture<long, long, NonDefaultEvent, IEventMetadata<long>, NonDefaultPersistedEvent> fixture)
+        {
+            _fixture = fixture;
+            var options = new DbContextOptionsBuilder<NonDefaultContext>()
+                .UseInMemoryDatabase(databaseName: Guid.NewGuid().ToString())
+                .Options;
+
+            _dbContext = new NonDefaultContext(options);
+        }
+
+        [Fact]
+        public async Task CanCommitEventsToStoreWithDefaultImplementations()
+        {
+            var store = _fixture.BuildEventStore(_dbContext);
+
+            await store.Append(1, 0, new[] { new FooEvent { Foo = "Bar" } });
+        }
+
+        [Fact]
+        public async Task CanReadEventsFromStoreWithNonDefaultImplementations()
+        {
+            _dbContext.Events.AddRange(new[]
+            {
+                new NonDefaultPersistedEvent
+                {
+                    StreamId = 1,
+                    Timestamp = DateTimeOffset.UtcNow,
+                    Version = 1,
+                    Type = "FooEvent",
+                    Payload = Encoding.UTF8.GetBytes(@"{""Foo"":""Bar""}")
+                }
+            });
+            await _dbContext.SaveChangesAsync();
+
+            var store = _fixture.BuildEventStore(_dbContext) as IEventStore<long, NonDefaultEvent, IEventMetadata<long>>;
+
+            var events = await store.Events(1);
+
+            Assert.Single(events);
+        }
+        public void Dispose()
+        {
+            _dbContext?.Dispose();
+        }
+    }
+}