mirror of https://github.com/Squidex/squidex.git
Sebastian Stehle
8 years ago
committed by
GitHub
GitHub
58 changed files with 1125 additions and 740 deletions
@ -0,0 +1,26 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using Squidex.Domain.Apps.Events; |
|||
using Squidex.Infrastructure.Commands; |
|||
using Squidex.Infrastructure.EventSourcing; |
|||
|
|||
namespace Squidex.Domain.Apps.Entities |
|||
{ |
|||
public abstract class SquidexDomainObjectBase<T> : DomainObjectBase<T> where T : IDomainState, new() |
|||
{ |
|||
public override void RaiseEvent(Envelope<IEvent> @event) |
|||
{ |
|||
if (@event.Payload is AppEvent appEvent) |
|||
{ |
|||
@event.SetAppId(appEvent.AppId.Id); |
|||
} |
|||
|
|||
base.RaiseEvent(@event); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,28 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using System; |
|||
using System.Globalization; |
|||
using Squidex.Infrastructure.EventSourcing; |
|||
|
|||
namespace Squidex.Domain.Apps.Events |
|||
{ |
|||
public static class SquidexHeaderExtensions |
|||
{ |
|||
public static Guid AppId(this EnvelopeHeaders headers) |
|||
{ |
|||
return headers[SquidexHeaders.AppId].ToGuid(CultureInfo.InvariantCulture); |
|||
} |
|||
|
|||
public static Envelope<T> SetAppId<T>(this Envelope<T> envelope, Guid value) where T : class |
|||
{ |
|||
envelope.Headers.Set(SquidexHeaders.AppId, value); |
|||
|
|||
return envelope; |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,14 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
namespace Squidex.Domain.Apps.Events |
|||
{ |
|||
public static class SquidexHeaders |
|||
{ |
|||
public static readonly string AppId = "AppId"; |
|||
} |
|||
} |
|||
@ -0,0 +1,142 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschränkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using System; |
|||
using System.Collections.Concurrent; |
|||
using System.Linq; |
|||
using System.Net; |
|||
using System.Net.Sockets; |
|||
using System.Threading.Tasks; |
|||
using EventStore.ClientAPI; |
|||
using EventStore.ClientAPI.Exceptions; |
|||
using EventStore.ClientAPI.Projections; |
|||
|
|||
namespace Squidex.Infrastructure.EventSourcing |
|||
{ |
|||
public sealed class ProjectionClient |
|||
{ |
|||
private readonly ConcurrentDictionary<string, bool> projections = new ConcurrentDictionary<string, bool>(); |
|||
private readonly IEventStoreConnection connection; |
|||
private readonly string prefix; |
|||
private readonly string projectionHost; |
|||
private ProjectionsManager projectionsManager; |
|||
|
|||
public ProjectionClient(IEventStoreConnection connection, string prefix, string projectionHost) |
|||
{ |
|||
this.connection = connection; |
|||
|
|||
this.prefix = prefix; |
|||
this.projectionHost = projectionHost; |
|||
} |
|||
|
|||
private string CreateFilterProjectionName(string filter) |
|||
{ |
|||
return $"by-{prefix.Slugify()}-{filter.Slugify()}"; |
|||
} |
|||
|
|||
private string CreatePropertyProjectionName(string property) |
|||
{ |
|||
return $"by-{prefix.Slugify()}-{property.Slugify()}-property"; |
|||
} |
|||
|
|||
public async Task<string> CreateProjectionAsync(string property, object value) |
|||
{ |
|||
var name = CreatePropertyProjectionName(property); |
|||
|
|||
var query = |
|||
$@"fromAll()
|
|||
.when({{ |
|||
$any: function (s, e) {{ |
|||
if (e.streamId.indexOf('{prefix}') === 0 && e.metadata.{property}) {{ |
|||
linkTo('{name}-' + e.metadata.{property}, e); |
|||
}} |
|||
}} |
|||
}});";
|
|||
|
|||
await CreateProjectionAsync(name, query); |
|||
|
|||
return $"{name}-{value}"; |
|||
} |
|||
|
|||
public async Task<string> CreateProjectionAsync(string streamFilter = null) |
|||
{ |
|||
streamFilter = streamFilter ?? ".*"; |
|||
|
|||
var name = CreateFilterProjectionName(streamFilter); |
|||
|
|||
var query = |
|||
$@"fromAll()
|
|||
.when({{ |
|||
$any: function (s, e) {{ |
|||
if (e.streamId.indexOf('{prefix}') === 0 && /{streamFilter}/.test(e.streamId.substring({prefix.Length + 1}))) {{ |
|||
linkTo('{name}', e); |
|||
}} |
|||
}} |
|||
}});";
|
|||
|
|||
await CreateProjectionAsync(name, query); |
|||
|
|||
return name; |
|||
} |
|||
|
|||
private async Task CreateProjectionAsync(string name, string query) |
|||
{ |
|||
if (projections.TryAdd(name, true)) |
|||
{ |
|||
try |
|||
{ |
|||
var credentials = connection.Settings.DefaultUserCredentials; |
|||
|
|||
await projectionsManager.CreateContinuousAsync(name, query, credentials); |
|||
} |
|||
catch (Exception ex) |
|||
{ |
|||
if (!ex.Is<ProjectionCommandConflictException>()) |
|||
{ |
|||
throw; |
|||
} |
|||
} |
|||
} |
|||
} |
|||
|
|||
public async Task ConnectAsync() |
|||
{ |
|||
var addressParts = projectionHost.Split(':'); |
|||
|
|||
if (addressParts.Length < 2 || !int.TryParse(addressParts[1], out var port)) |
|||
{ |
|||
port = 2113; |
|||
} |
|||
|
|||
var endpoints = await Dns.GetHostAddressesAsync(addressParts[0]); |
|||
var endpoint = new IPEndPoint(endpoints.First(x => x.AddressFamily == AddressFamily.InterNetwork), port); |
|||
|
|||
projectionsManager = |
|||
new ProjectionsManager( |
|||
connection.Settings.Log, endpoint, |
|||
connection.Settings.OperationTimeout); |
|||
try |
|||
{ |
|||
await projectionsManager.ListAllAsync(connection.Settings.DefaultUserCredentials); |
|||
} |
|||
catch (Exception ex) |
|||
{ |
|||
throw new ConfigurationException($"Cannot connect to event store projections: {projectionHost}.", ex); |
|||
} |
|||
} |
|||
|
|||
public long? ParsePositionOrNull(string position) |
|||
{ |
|||
return long.TryParse(position, out var parsedPosition) ? (long?)parsedPosition : null; |
|||
} |
|||
|
|||
public long ParsePosition(string position) |
|||
{ |
|||
return long.TryParse(position, out var parsedPosition) ? parsedPosition : 0; |
|||
} |
|||
} |
|||
} |
|||
@ -1,97 +0,0 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschränkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using System; |
|||
using System.Collections.Concurrent; |
|||
using System.Globalization; |
|||
using System.Linq; |
|||
using System.Net; |
|||
using System.Net.Sockets; |
|||
using System.Threading.Tasks; |
|||
using EventStore.ClientAPI; |
|||
using EventStore.ClientAPI.Exceptions; |
|||
using EventStore.ClientAPI.Projections; |
|||
|
|||
namespace Squidex.Infrastructure.EventSourcing |
|||
{ |
|||
public static class ProjectionHelper |
|||
{ |
|||
private const string ProjectionName = "by-{0}-{1}"; |
|||
private static readonly ConcurrentDictionary<string, bool> SubscriptionsCreated = new ConcurrentDictionary<string, bool>(); |
|||
|
|||
private static string ParseFilter(string prefix, string filter) |
|||
{ |
|||
return string.Format(CultureInfo.InvariantCulture, ProjectionName, prefix.Slugify(), filter.Slugify()); |
|||
} |
|||
|
|||
public static async Task<string> CreateProjectionAsync(this IEventStoreConnection connection, ProjectionsManager projectionsManager, string prefix, string streamFilter = null) |
|||
{ |
|||
streamFilter = streamFilter ?? ".*"; |
|||
|
|||
var streamName = ParseFilter(prefix, streamFilter); |
|||
|
|||
if (SubscriptionsCreated.TryAdd(streamName, true)) |
|||
{ |
|||
var projectionConfig = |
|||
$@"fromAll()
|
|||
.when({{ |
|||
$any: function (s, e) {{ |
|||
if (e.streamId.indexOf('{prefix}') === 0 && /{streamFilter}/.test(e.streamId.substring({prefix.Length + 1}))) {{ |
|||
linkTo('{streamName}', e); |
|||
}} |
|||
}} |
|||
}});";
|
|||
|
|||
try |
|||
{ |
|||
var credentials = connection.Settings.DefaultUserCredentials; |
|||
|
|||
await projectionsManager.CreateContinuousAsync($"${streamName}", projectionConfig, credentials); |
|||
} |
|||
catch (Exception ex) |
|||
{ |
|||
if (!ex.Is<ProjectionCommandConflictException>()) |
|||
{ |
|||
throw; |
|||
} |
|||
} |
|||
} |
|||
|
|||
return streamName; |
|||
} |
|||
|
|||
public static async Task<ProjectionsManager> GetProjectionsManagerAsync(this IEventStoreConnection connection, string projectionHost) |
|||
{ |
|||
var addressParts = projectionHost.Split(':'); |
|||
|
|||
if (addressParts.Length < 2 || !int.TryParse(addressParts[1], out var port)) |
|||
{ |
|||
port = 2113; |
|||
} |
|||
|
|||
var endpoints = await Dns.GetHostAddressesAsync(addressParts[0]); |
|||
var endpoint = new IPEndPoint(endpoints.First(x => x.AddressFamily == AddressFamily.InterNetwork), port); |
|||
|
|||
var projectionsManager = |
|||
new ProjectionsManager( |
|||
connection.Settings.Log, endpoint, |
|||
connection.Settings.OperationTimeout); |
|||
|
|||
return projectionsManager; |
|||
} |
|||
|
|||
public static long? ParsePositionOrNull(string position) |
|||
{ |
|||
return long.TryParse(position, out var parsedPosition) ? (long?)parsedPosition : null; |
|||
} |
|||
|
|||
public static long ParsePosition(string position) |
|||
{ |
|||
return long.TryParse(position, out var parsedPosition) ? parsedPosition : 0; |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,173 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschränkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using System; |
|||
using System.Collections.Generic; |
|||
using System.Reactive.Linq; |
|||
using System.Threading; |
|||
using System.Threading.Tasks; |
|||
using MongoDB.Driver; |
|||
using Squidex.Infrastructure.MongoDb; |
|||
|
|||
namespace Squidex.Infrastructure.EventSourcing |
|||
{ |
|||
public partial class MongoEventStore : MongoRepositoryBase<MongoEventCommit>, IEventStore |
|||
{ |
|||
public Task CreateIndexAsync(string property) |
|||
{ |
|||
return Collection.Indexes.CreateOneAsync(Index.Ascending(CreateIndexPath(property))); |
|||
} |
|||
|
|||
public IEventSubscription CreateSubscription(IEventSubscriber subscriber, string streamFilter, string position = null) |
|||
{ |
|||
Guard.NotNull(subscriber, nameof(subscriber)); |
|||
Guard.NotNullOrEmpty(streamFilter, nameof(streamFilter)); |
|||
|
|||
return new PollingSubscription(this, notifier, subscriber, streamFilter, position); |
|||
} |
|||
|
|||
public async Task<IReadOnlyList<StoredEvent>> QueryAsync(string streamName, long streamPosition = 0) |
|||
{ |
|||
var commits = |
|||
await Collection.Find( |
|||
Filter.And( |
|||
Filter.Eq(EventStreamField, streamName), |
|||
Filter.Gte(EventStreamOffsetField, streamPosition - 1))) |
|||
.Sort(Sort.Ascending(TimestampField)).ToListAsync(); |
|||
|
|||
var result = new List<StoredEvent>(); |
|||
|
|||
foreach (var commit in commits) |
|||
{ |
|||
var eventStreamOffset = (int)commit.EventStreamOffset; |
|||
|
|||
var commitTimestamp = commit.Timestamp; |
|||
var commitOffset = 0; |
|||
|
|||
foreach (var e in commit.Events) |
|||
{ |
|||
eventStreamOffset++; |
|||
|
|||
if (eventStreamOffset >= streamPosition) |
|||
{ |
|||
var eventData = e.ToEventData(); |
|||
var eventToken = new StreamPosition(commitTimestamp, commitOffset, commit.Events.Length); |
|||
|
|||
result.Add(new StoredEvent(eventToken, eventStreamOffset, eventData)); |
|||
} |
|||
} |
|||
} |
|||
|
|||
return result; |
|||
} |
|||
|
|||
public Task QueryAsync(Func<StoredEvent, Task> callback, string property, object value, string position = null, CancellationToken ct = default(CancellationToken)) |
|||
{ |
|||
Guard.NotNull(callback, nameof(callback)); |
|||
|
|||
StreamPosition lastPosition = position; |
|||
|
|||
var filter = CreateFilter(property, value, lastPosition); |
|||
|
|||
return QueryAsync(callback, lastPosition, filter, ct); |
|||
} |
|||
|
|||
public Task QueryAsync(Func<StoredEvent, Task> callback, string streamFilter = null, string position = null, CancellationToken ct = default(CancellationToken)) |
|||
{ |
|||
Guard.NotNull(callback, nameof(callback)); |
|||
|
|||
StreamPosition lastPosition = position; |
|||
|
|||
var filter = CreateFilter(streamFilter, lastPosition); |
|||
|
|||
return QueryAsync(callback, lastPosition, filter, ct); |
|||
} |
|||
|
|||
private async Task QueryAsync(Func<StoredEvent, Task> callback, StreamPosition lastPosition, FilterDefinition<MongoEventCommit> filter, CancellationToken ct) |
|||
{ |
|||
await Collection.Find(filter).Sort(Sort.Ascending(TimestampField)).ForEachAsync(async commit => |
|||
{ |
|||
var eventStreamOffset = (int)commit.EventStreamOffset; |
|||
|
|||
var commitTimestamp = commit.Timestamp; |
|||
var commitOffset = 0; |
|||
|
|||
foreach (var e in commit.Events) |
|||
{ |
|||
eventStreamOffset++; |
|||
|
|||
if (commitOffset > lastPosition.CommitOffset || commitTimestamp > lastPosition.Timestamp) |
|||
{ |
|||
var eventData = e.ToEventData(); |
|||
var eventToken = new StreamPosition(commitTimestamp, commitOffset, commit.Events.Length); |
|||
|
|||
await callback(new StoredEvent(eventToken, eventStreamOffset, eventData)); |
|||
|
|||
commitOffset++; |
|||
} |
|||
} |
|||
}, ct); |
|||
} |
|||
|
|||
private static FilterDefinition<MongoEventCommit> CreateFilter(string property, object value, StreamPosition streamPosition) |
|||
{ |
|||
var filters = new List<FilterDefinition<MongoEventCommit>>(); |
|||
|
|||
AddPositionFilter(streamPosition, filters); |
|||
AddPropertyFitler(property, value, filters); |
|||
|
|||
return Filter.And(filters); |
|||
} |
|||
|
|||
private static FilterDefinition<MongoEventCommit> CreateFilter(string streamFilter, StreamPosition streamPosition) |
|||
{ |
|||
var filters = new List<FilterDefinition<MongoEventCommit>>(); |
|||
|
|||
AddPositionFilter(streamPosition, filters); |
|||
AddStreamFilter(streamFilter, filters); |
|||
|
|||
return Filter.And(filters); |
|||
} |
|||
|
|||
private static void AddPropertyFitler(string property, object value, List<FilterDefinition<MongoEventCommit>> filters) |
|||
{ |
|||
filters.Add(Filter.Eq(CreateIndexPath(property), value)); |
|||
} |
|||
|
|||
private static void AddStreamFilter(string streamFilter, List<FilterDefinition<MongoEventCommit>> filters) |
|||
{ |
|||
if (!string.IsNullOrWhiteSpace(streamFilter) && !string.Equals(streamFilter, ".*", StringComparison.OrdinalIgnoreCase)) |
|||
{ |
|||
if (streamFilter.Contains("^")) |
|||
{ |
|||
filters.Add(Filter.Regex(EventStreamField, streamFilter)); |
|||
} |
|||
else |
|||
{ |
|||
filters.Add(Filter.Eq(EventStreamField, streamFilter)); |
|||
} |
|||
} |
|||
} |
|||
|
|||
private static void AddPositionFilter(StreamPosition streamPosition, List<FilterDefinition<MongoEventCommit>> filters) |
|||
{ |
|||
if (streamPosition.IsEndOfCommit) |
|||
{ |
|||
filters.Add(Filter.Gt(TimestampField, streamPosition.Timestamp)); |
|||
} |
|||
else |
|||
{ |
|||
filters.Add(Filter.Gte(TimestampField, streamPosition.Timestamp)); |
|||
} |
|||
} |
|||
|
|||
private static string CreateIndexPath(string property) |
|||
{ |
|||
return $"Events.Metadata.{property}"; |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,129 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschränkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using System; |
|||
using System.Collections.Generic; |
|||
using System.Reactive.Linq; |
|||
using System.Threading.Tasks; |
|||
using MongoDB.Bson; |
|||
using MongoDB.Driver; |
|||
|
|||
namespace Squidex.Infrastructure.EventSourcing |
|||
{ |
|||
public partial class MongoEventStore |
|||
{ |
|||
private const int MaxWriteAttempts = 20; |
|||
private static readonly BsonTimestamp EmptyTimestamp = new BsonTimestamp(0); |
|||
|
|||
public Task AppendAsync(Guid commitId, string streamName, ICollection<EventData> events) |
|||
{ |
|||
return AppendAsync(commitId, streamName, EtagVersion.Any, events); |
|||
} |
|||
|
|||
public async Task AppendAsync(Guid commitId, string streamName, long expectedVersion, ICollection<EventData> events) |
|||
{ |
|||
Guard.GreaterEquals(expectedVersion, EtagVersion.Any, nameof(expectedVersion)); |
|||
Guard.NotNullOrEmpty(streamName, nameof(streamName)); |
|||
Guard.NotNull(events, nameof(events)); |
|||
|
|||
if (events.Count == 0) |
|||
{ |
|||
return; |
|||
} |
|||
|
|||
var currentVersion = await GetEventStreamOffset(streamName); |
|||
|
|||
if (expectedVersion != EtagVersion.Any && expectedVersion != currentVersion) |
|||
{ |
|||
throw new WrongEventVersionException(currentVersion, expectedVersion); |
|||
} |
|||
|
|||
var commit = BuildCommit(commitId, streamName, expectedVersion >= -1 ? expectedVersion : currentVersion, events); |
|||
|
|||
for (var attempt = 0; attempt < MaxWriteAttempts; attempt++) |
|||
{ |
|||
try |
|||
{ |
|||
await Collection.InsertOneAsync(commit); |
|||
|
|||
notifier.NotifyEventsStored(streamName); |
|||
|
|||
return; |
|||
} |
|||
catch (MongoWriteException ex) |
|||
{ |
|||
if (ex.WriteError?.Category == ServerErrorCategory.DuplicateKey) |
|||
{ |
|||
currentVersion = await GetEventStreamOffset(streamName); |
|||
|
|||
if (expectedVersion != EtagVersion.Any) |
|||
{ |
|||
throw new WrongEventVersionException(currentVersion, expectedVersion); |
|||
} |
|||
|
|||
if (attempt < MaxWriteAttempts) |
|||
{ |
|||
expectedVersion = currentVersion; |
|||
} |
|||
else |
|||
{ |
|||
throw new TimeoutException("Could not acquire a free slot for the commit within the provided time."); |
|||
} |
|||
} |
|||
else |
|||
{ |
|||
throw; |
|||
} |
|||
} |
|||
} |
|||
} |
|||
|
|||
private async Task<long> GetEventStreamOffset(string streamName) |
|||
{ |
|||
var document = |
|||
await Collection.Find(Filter.Eq(EventStreamField, streamName)) |
|||
.Project<BsonDocument>(Projection |
|||
.Include(EventStreamOffsetField) |
|||
.Include(EventsCountField)) |
|||
.Sort(Sort.Descending(EventStreamOffsetField)).Limit(1) |
|||
.FirstOrDefaultAsync(); |
|||
|
|||
if (document != null) |
|||
{ |
|||
return document[nameof(MongoEventCommit.EventStreamOffset)].ToInt64() + document[nameof(MongoEventCommit.EventsCount)].ToInt64(); |
|||
} |
|||
|
|||
return EtagVersion.Empty; |
|||
} |
|||
|
|||
private static MongoEventCommit BuildCommit(Guid commitId, string streamName, long expectedVersion, ICollection<EventData> events) |
|||
{ |
|||
var commitEvents = new MongoEvent[events.Count]; |
|||
|
|||
var i = 0; |
|||
|
|||
foreach (var e in events) |
|||
{ |
|||
var mongoEvent = MongoEvent.FromEventData(e); |
|||
|
|||
commitEvents[i++] = mongoEvent; |
|||
} |
|||
|
|||
var mongoCommit = new MongoEventCommit |
|||
{ |
|||
Id = commitId, |
|||
Events = commitEvents, |
|||
EventsCount = events.Count, |
|||
EventStream = streamName, |
|||
EventStreamOffset = expectedVersion, |
|||
Timestamp = EmptyTimestamp |
|||
}; |
|||
|
|||
return mongoCommit; |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,32 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using MongoDB.Bson.Serialization; |
|||
using MongoDB.Bson.Serialization.Serializers; |
|||
using Newtonsoft.Json.Linq; |
|||
|
|||
namespace Squidex.Infrastructure.MongoDb |
|||
{ |
|||
public sealed class JTokenSerializer<T> : ClassSerializerBase<T> where T : JToken |
|||
{ |
|||
public static readonly JTokenSerializer<T> Instance = new JTokenSerializer<T>(); |
|||
|
|||
protected override T DeserializeValue(BsonDeserializationContext context, BsonDeserializationArgs args) |
|||
{ |
|||
var jsonReader = new BsonJsonReader(context.Reader); |
|||
|
|||
return (T)JToken.ReadFrom(jsonReader); |
|||
} |
|||
|
|||
protected override void SerializeValue(BsonSerializationContext context, BsonSerializationArgs args, T value) |
|||
{ |
|||
var jsonWriter = new BsonJsonWriter(context.Writer); |
|||
|
|||
value.WriteTo(jsonWriter); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,16 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using System.Collections.Generic; |
|||
|
|||
namespace Squidex.Infrastructure.Migrations |
|||
{ |
|||
public interface IMigrationPath |
|||
{ |
|||
(int Version, IEnumerable<IMigration> Migrations) GetNext(int version); |
|||
} |
|||
} |
|||
@ -1,38 +0,0 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschränkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using System.Collections.Generic; |
|||
using System.Linq; |
|||
using System.Threading.Tasks; |
|||
using Squidex.Infrastructure.Migrations; |
|||
|
|||
namespace Migrate_01 |
|||
{ |
|||
public sealed class Migration05_RebuildForNewCommands : IMigration |
|||
{ |
|||
private readonly Rebuilder rebuilder; |
|||
|
|||
public int FromVersion { get; } = 4; |
|||
|
|||
public int ToVersion { get; } = 5; |
|||
|
|||
public Migration05_RebuildForNewCommands(Rebuilder rebuilder) |
|||
{ |
|||
this.rebuilder = rebuilder; |
|||
} |
|||
|
|||
public async Task UpdateAsync(IEnumerable<IMigration> previousMigrations) |
|||
{ |
|||
if (!previousMigrations.Any(x => x is Migration01_FromCqrs)) |
|||
{ |
|||
await rebuilder.RebuildConfigAsync(); |
|||
await rebuilder.RebuildContentAsync(); |
|||
await rebuilder.RebuildAssetsAsync(); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,56 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using System; |
|||
using System.Collections.Generic; |
|||
using Microsoft.Extensions.DependencyInjection; |
|||
using Migrate_01.Migrations; |
|||
using Squidex.Infrastructure.Migrations; |
|||
|
|||
namespace Migrate_01 |
|||
{ |
|||
public sealed class MigrationPath : IMigrationPath |
|||
{ |
|||
private const int CurrentVersion = 6; |
|||
private readonly IServiceProvider serviceProvider; |
|||
|
|||
public MigrationPath(IServiceProvider serviceProvider) |
|||
{ |
|||
this.serviceProvider = serviceProvider; |
|||
} |
|||
|
|||
public (int Version, IEnumerable<IMigration> Migrations) GetNext(int version) |
|||
{ |
|||
if (version == CurrentVersion) |
|||
{ |
|||
return (CurrentVersion, null); |
|||
} |
|||
|
|||
var migrations = new List<IMigration>(); |
|||
|
|||
// Version 6: Convert Event store. Must always be executed first.
|
|||
if (version < 6) |
|||
{ |
|||
migrations.Add(serviceProvider.GetRequiredService<ConvertEventStore>()); |
|||
} |
|||
|
|||
// Version 5: Fixes the broken command architecture and requires a rebuild of all snapshots.
|
|||
if (version < 5) |
|||
{ |
|||
migrations.Add(serviceProvider.GetRequiredService<RebuildSnapshots>()); |
|||
} |
|||
|
|||
// Version 1: Introduce App patterns.
|
|||
if (version <= 1) |
|||
{ |
|||
migrations.Add(serviceProvider.GetRequiredService<AddPatterns>()); |
|||
} |
|||
|
|||
return (CurrentVersion, migrations); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,59 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using System; |
|||
using System.Threading.Tasks; |
|||
using MongoDB.Bson; |
|||
using MongoDB.Driver; |
|||
using Newtonsoft.Json.Linq; |
|||
using Squidex.Domain.Apps.Events; |
|||
using Squidex.Infrastructure; |
|||
using Squidex.Infrastructure.EventSourcing; |
|||
using Squidex.Infrastructure.Migrations; |
|||
using Squidex.Infrastructure.MongoDb; |
|||
|
|||
namespace Migrate_01.Migrations |
|||
{ |
|||
public sealed class ConvertEventStore : IMigration |
|||
{ |
|||
private readonly IEventStore eventStore; |
|||
|
|||
public ConvertEventStore(IEventStore eventStore) |
|||
{ |
|||
this.eventStore = eventStore; |
|||
} |
|||
|
|||
public async Task UpdateAsync() |
|||
{ |
|||
if (eventStore is MongoEventStore mongoEventStore) |
|||
{ |
|||
var collection = mongoEventStore.RawCollection; |
|||
|
|||
var filter = Builders<BsonDocument>.Filter; |
|||
|
|||
await collection.Find(new BsonDocument()).ForEachAsync(async commit => |
|||
{ |
|||
foreach (BsonDocument @event in commit["Events"].AsBsonArray) |
|||
{ |
|||
var meta = JObject.Parse(@event["Metadata"].AsString); |
|||
var data = JObject.Parse(@event["Payload"].AsString); |
|||
|
|||
if (data.TryGetValue("appId", out var appId)) |
|||
{ |
|||
meta[SquidexHeaders.AppId] = NamedId<Guid>.Parse(appId.ToString(), Guid.TryParse).Id; |
|||
} |
|||
|
|||
@event.Remove("EventId"); |
|||
@event["Metadata"] = meta.ToBson(); |
|||
} |
|||
|
|||
await collection.ReplaceOneAsync(filter.Eq("_id", commit["_id"].AsString), commit); |
|||
}); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
Loading…
Reference in new issue