mirror of https://github.com/Squidex/squidex.git
Browse Source
* Temp * Temp. * Rename messages. * Some progress. * Some progress. * Some fixes in registration of dependencies. * Infrastructure tests * More tests. * More tests. * Naming fixes. * Compile fix * More fixes. * Tests green :) * More tests and distributed cache. * Compile and bug fixes. * Fix recreation. * Fix tests and better cancellation token flow. * Flow the cancellation token. * Compile fix. * Build fix. * Fixes and better tests. * Fix threading. * File was not saved. * More cancellation tokens. * Fix method name. * Fix tests. * Update yml. * Better check. * Increase time for text indexer.pull/898/head
Sebastian Stehle
4 years ago
committed by
GitHub
GitHub
467 changed files with 7220 additions and 11061 deletions
@ -1,30 +0,0 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using Orleans; |
|||
using Squidex.Infrastructure.EventSourcing.Grains; |
|||
using Squidex.Infrastructure.Migrations; |
|||
using Squidex.Infrastructure.Orleans; |
|||
|
|||
namespace Migrations.Migrations |
|||
{ |
|||
public sealed class StartEventConsumers : IMigration |
|||
{ |
|||
private readonly IEventConsumerManagerGrain eventConsumerManager; |
|||
|
|||
public StartEventConsumers(IGrainFactory grainFactory) |
|||
{ |
|||
eventConsumerManager = grainFactory.GetGrain<IEventConsumerManagerGrain>(SingleGrain.Id); |
|||
} |
|||
|
|||
public Task UpdateAsync( |
|||
CancellationToken ct) |
|||
{ |
|||
return eventConsumerManager.StartAllAsync(); |
|||
} |
|||
} |
|||
} |
|||
@ -1,30 +0,0 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using Orleans; |
|||
using Squidex.Infrastructure.EventSourcing.Grains; |
|||
using Squidex.Infrastructure.Migrations; |
|||
using Squidex.Infrastructure.Orleans; |
|||
|
|||
namespace Migrations.Migrations |
|||
{ |
|||
public sealed class StopEventConsumers : IMigration |
|||
{ |
|||
private readonly IEventConsumerManagerGrain eventConsumerManager; |
|||
|
|||
public StopEventConsumers(IGrainFactory grainFactory) |
|||
{ |
|||
eventConsumerManager = grainFactory.GetGrain<IEventConsumerManagerGrain>(SingleGrain.Id); |
|||
} |
|||
|
|||
public Task UpdateAsync( |
|||
CancellationToken ct) |
|||
{ |
|||
return eventConsumerManager.StopAllAsync(); |
|||
} |
|||
} |
|||
} |
|||
@ -1,120 +0,0 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using Orleans.Core; |
|||
using Squidex.Infrastructure; |
|||
using Squidex.Infrastructure.Json.Objects; |
|||
using Squidex.Infrastructure.Orleans; |
|||
using Squidex.Infrastructure.States; |
|||
|
|||
namespace Squidex.Domain.Apps.Entities.Apps |
|||
{ |
|||
public sealed class AppUISettingsGrain : GrainBase, IAppUISettingsGrain |
|||
{ |
|||
private readonly IGrainState<State> state; |
|||
|
|||
[CollectionName("UISettings")] |
|||
public sealed class State |
|||
{ |
|||
public JsonObject Settings { get; set; } = new JsonObject(); |
|||
} |
|||
|
|||
public AppUISettingsGrain(IGrainIdentity identity, IGrainState<State> state) |
|||
: base(identity) |
|||
{ |
|||
this.state = state; |
|||
} |
|||
|
|||
public Task<JsonObject> GetAsync() |
|||
{ |
|||
return Task.FromResult(state.Value.Settings); |
|||
} |
|||
|
|||
public Task ClearAsync() |
|||
{ |
|||
TryDeactivateOnIdle(); |
|||
|
|||
return state.ClearAsync(); |
|||
} |
|||
|
|||
public Task SetAsync(JsonObject settings) |
|||
{ |
|||
state.Value.Settings = settings; |
|||
|
|||
return state.WriteAsync(); |
|||
} |
|||
|
|||
public Task SetAsync(string path, JsonValue value) |
|||
{ |
|||
var container = GetContainer(path, true, out var key); |
|||
|
|||
if (container == null) |
|||
{ |
|||
ThrowHelper.InvalidOperationException("Path does not lead to an object."); |
|||
return Task.CompletedTask; |
|||
} |
|||
|
|||
container[key] = value; |
|||
|
|||
return state.WriteAsync(); |
|||
} |
|||
|
|||
public async Task RemoveAsync(string path) |
|||
{ |
|||
var container = GetContainer(path, false, out var key); |
|||
|
|||
if (container?.ContainsKey(key) == true) |
|||
{ |
|||
container.Remove(key); |
|||
|
|||
await state.WriteAsync(); |
|||
} |
|||
} |
|||
|
|||
private JsonObject? GetContainer(string path, bool add, out string key) |
|||
{ |
|||
Guard.NotNullOrEmpty(path); |
|||
|
|||
var segments = path.Split('.'); |
|||
|
|||
key = segments[^1]; |
|||
|
|||
var current = state.Value.Settings; |
|||
|
|||
if (segments.Length > 1) |
|||
{ |
|||
foreach (var segment in segments.Take(segments.Length - 1)) |
|||
{ |
|||
if (!current.TryGetValue(segment, out var found)) |
|||
{ |
|||
if (add) |
|||
{ |
|||
found = new JsonObject(); |
|||
|
|||
current[segment] = found; |
|||
} |
|||
else |
|||
{ |
|||
return null; |
|||
} |
|||
} |
|||
|
|||
if (found.Value is JsonObject o) |
|||
{ |
|||
current = o; |
|||
} |
|||
else |
|||
{ |
|||
return null; |
|||
} |
|||
} |
|||
} |
|||
|
|||
return current; |
|||
} |
|||
} |
|||
} |
|||
@ -1,37 +0,0 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using Microsoft.Extensions.Diagnostics.HealthChecks; |
|||
using Orleans; |
|||
using Squidex.Domain.Apps.Entities.Apps.Indexes; |
|||
using Squidex.Infrastructure.Orleans; |
|||
|
|||
namespace Squidex.Domain.Apps.Entities.Apps.Diagnostics |
|||
{ |
|||
public sealed class OrleansAppsHealthCheck : IHealthCheck |
|||
{ |
|||
private readonly IGrainFactory grainFactory; |
|||
|
|||
public OrleansAppsHealthCheck(IGrainFactory grainFactory) |
|||
{ |
|||
this.grainFactory = grainFactory; |
|||
} |
|||
|
|||
public async Task<HealthCheckResult> CheckHealthAsync(HealthCheckContext context, |
|||
CancellationToken cancellationToken = default) |
|||
{ |
|||
await GetGrain().GetAppIdsAsync(new[] { "test" }); |
|||
|
|||
return HealthCheckResult.Healthy("Orleans must establish communication."); |
|||
} |
|||
|
|||
private IAppsCacheGrain GetGrain() |
|||
{ |
|||
return grainFactory.GetGrain<IAppsCacheGrain>(SingleGrain.Id); |
|||
} |
|||
} |
|||
} |
|||
@ -1,27 +0,0 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using Orleans.Core; |
|||
using Squidex.Infrastructure.Commands; |
|||
|
|||
namespace Squidex.Domain.Apps.Entities.Apps.DomainObject |
|||
{ |
|||
public sealed class AppDomainObjectGrain : DomainObjectGrain<AppDomainObject, AppDomainObject.State>, IAppGrain |
|||
{ |
|||
public AppDomainObjectGrain(IGrainIdentity identity, IDomainObjectFactory factory) |
|||
: base(identity, factory) |
|||
{ |
|||
} |
|||
|
|||
public async Task<IAppEntity> GetStateAsync() |
|||
{ |
|||
await DomainObject.EnsureLoadedAsync(); |
|||
|
|||
return Snapshot; |
|||
} |
|||
} |
|||
} |
|||
@ -1,16 +0,0 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using Squidex.Infrastructure.Commands; |
|||
|
|||
namespace Squidex.Domain.Apps.Entities.Apps.DomainObject |
|||
{ |
|||
public interface IAppGrain : IDomainObjectGrain |
|||
{ |
|||
Task<IAppEntity> GetStateAsync(); |
|||
} |
|||
} |
|||
@ -1,108 +0,0 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using Orleans.Concurrency; |
|||
using Orleans.Core; |
|||
using Squidex.Domain.Apps.Entities.Apps.Repositories; |
|||
using Squidex.Infrastructure; |
|||
using Squidex.Infrastructure.Orleans.Indexes; |
|||
|
|||
namespace Squidex.Domain.Apps.Entities.Apps.Indexes |
|||
{ |
|||
[Reentrant] |
|||
public sealed class AppsCacheGrain : UniqueNameGrain<DomainId>, IAppsCacheGrain |
|||
{ |
|||
private readonly IAppRepository appRepository; |
|||
private readonly Dictionary<string, DomainId> appIds = new Dictionary<string, DomainId>(); |
|||
|
|||
public AppsCacheGrain(IGrainIdentity identity, IAppRepository appRepository) |
|||
: base(identity) |
|||
{ |
|||
this.appRepository = appRepository; |
|||
} |
|||
|
|||
public override async Task<string?> ReserveAsync(DomainId id, string name) |
|||
{ |
|||
var token = await base.ReserveAsync(id, name); |
|||
|
|||
if (token == null) |
|||
{ |
|||
return null; |
|||
} |
|||
|
|||
var ids = await GetAppIdsAsync(new[] { name }); |
|||
|
|||
if (ids.Any()) |
|||
{ |
|||
await RemoveReservationAsync(token); |
|||
return null; |
|||
} |
|||
|
|||
return token; |
|||
} |
|||
|
|||
public async Task<IReadOnlyCollection<DomainId>> GetAppIdsAsync(string[] names) |
|||
{ |
|||
var result = new List<DomainId>(); |
|||
|
|||
List<string>? pendingNames = null; |
|||
|
|||
foreach (var name in names) |
|||
{ |
|||
if (!appIds.TryGetValue(name, out var cachedId)) |
|||
{ |
|||
pendingNames ??= new List<string>(); |
|||
pendingNames.Add(name); |
|||
} |
|||
else if (cachedId != DomainId.Empty) |
|||
{ |
|||
result.Add(cachedId); |
|||
} |
|||
} |
|||
|
|||
if (pendingNames != null) |
|||
{ |
|||
var foundIds = await appRepository.QueryIdsAsync(pendingNames); |
|||
|
|||
foreach (var name in pendingNames) |
|||
{ |
|||
if (foundIds.TryGetValue(name, out var id)) |
|||
{ |
|||
appIds[name] = id; |
|||
|
|||
result.Add(id); |
|||
} |
|||
else |
|||
{ |
|||
appIds[name] = default; |
|||
} |
|||
} |
|||
} |
|||
|
|||
return result; |
|||
} |
|||
|
|||
public Task AddAsync(DomainId id, string name) |
|||
{ |
|||
appIds[name] = id; |
|||
|
|||
return Task.CompletedTask; |
|||
} |
|||
|
|||
public Task RemoveAsync(DomainId id) |
|||
{ |
|||
var name = appIds.FirstOrDefault(x => x.Value == id).Key; |
|||
|
|||
if (name != null) |
|||
{ |
|||
appIds.Remove(name); |
|||
} |
|||
|
|||
return Task.CompletedTask; |
|||
} |
|||
} |
|||
} |
|||
@ -1,21 +0,0 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using Squidex.Infrastructure; |
|||
using Squidex.Infrastructure.Orleans.Indexes; |
|||
|
|||
namespace Squidex.Domain.Apps.Entities.Apps.Indexes |
|||
{ |
|||
public interface IAppsCacheGrain : IUniqueNameGrain<DomainId> |
|||
{ |
|||
Task<IReadOnlyCollection<DomainId>> GetAppIdsAsync(string[] names); |
|||
|
|||
Task AddAsync(DomainId id, string name); |
|||
|
|||
Task RemoveAsync(DomainId id); |
|||
} |
|||
} |
|||
@ -1,18 +0,0 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using Orleans; |
|||
using Orleans.Concurrency; |
|||
|
|||
namespace Squidex.Domain.Apps.Entities.Apps.Plans |
|||
{ |
|||
public interface IUsageNotifierGrain : IGrainWithStringKey |
|||
{ |
|||
[OneWay] |
|||
Task NotifyAsync(UsageNotification notification); |
|||
} |
|||
} |
|||
@ -1,40 +0,0 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using Orleans.Core; |
|||
using Squidex.Infrastructure; |
|||
using Squidex.Infrastructure.Commands; |
|||
using Squidex.Infrastructure.Orleans; |
|||
|
|||
namespace Squidex.Domain.Apps.Entities.Assets.DomainObject |
|||
{ |
|||
public sealed class AssetDomainObjectGrain : DomainObjectGrain<AssetDomainObject, AssetDomainObject.State>, IAssetGrain |
|||
{ |
|||
private static readonly TimeSpan Lifetime = TimeSpan.FromMinutes(5); |
|||
|
|||
public AssetDomainObjectGrain(IGrainIdentity identity, IDomainObjectFactory factory, |
|||
IActivationLimit limit) |
|||
: base(identity, factory) |
|||
{ |
|||
limit?.SetLimit(5000, Lifetime); |
|||
} |
|||
|
|||
public override Task OnActivateAsync() |
|||
{ |
|||
TryDelayDeactivation(Lifetime); |
|||
|
|||
return base.OnActivateAsync(); |
|||
} |
|||
|
|||
public async Task<IAssetEntity> GetStateAsync(long version = EtagVersion.Any) |
|||
{ |
|||
await DomainObject.EnsureLoadedAsync(); |
|||
|
|||
return await DomainObject.GetSnapshotAsync(version); |
|||
} |
|||
} |
|||
} |
|||
@ -1,39 +0,0 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using Orleans.Core; |
|||
using Squidex.Infrastructure.Commands; |
|||
using Squidex.Infrastructure.Orleans; |
|||
|
|||
namespace Squidex.Domain.Apps.Entities.Assets.DomainObject |
|||
{ |
|||
public sealed class AssetFolderDomainObjectGrain : DomainObjectGrain<AssetFolderDomainObject, AssetFolderDomainObject.State>, IAssetFolderGrain |
|||
{ |
|||
private static readonly TimeSpan Lifetime = TimeSpan.FromMinutes(5); |
|||
|
|||
public AssetFolderDomainObjectGrain(IGrainIdentity grainIdentity, IDomainObjectFactory factory, |
|||
IActivationLimit limit) |
|||
: base(grainIdentity, factory) |
|||
{ |
|||
limit?.SetLimit(5000, Lifetime); |
|||
} |
|||
|
|||
public override Task OnActivateAsync() |
|||
{ |
|||
TryDelayDeactivation(Lifetime); |
|||
|
|||
return base.OnActivateAsync(); |
|||
} |
|||
|
|||
public async Task<IAssetFolderEntity> GetStateAsync() |
|||
{ |
|||
await DomainObject.EnsureLoadedAsync(); |
|||
|
|||
return Snapshot; |
|||
} |
|||
} |
|||
} |
|||
@ -1,17 +0,0 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using Squidex.Infrastructure; |
|||
using Squidex.Infrastructure.Commands; |
|||
|
|||
namespace Squidex.Domain.Apps.Entities.Assets.DomainObject |
|||
{ |
|||
public interface IAssetGrain : IDomainObjectGrain |
|||
{ |
|||
Task<IAssetEntity> GetStateAsync(long version = EtagVersion.Any); |
|||
} |
|||
} |
|||
@ -1,285 +0,0 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using System.Text.RegularExpressions; |
|||
using Microsoft.Extensions.Logging; |
|||
using NodaTime; |
|||
using Orleans.Concurrency; |
|||
using Orleans.Core; |
|||
using Squidex.Domain.Apps.Entities.Backup.State; |
|||
using Squidex.Domain.Apps.Events; |
|||
using Squidex.Infrastructure; |
|||
using Squidex.Infrastructure.EventSourcing; |
|||
using Squidex.Infrastructure.Orleans; |
|||
using Squidex.Infrastructure.Tasks; |
|||
using Squidex.Infrastructure.Translations; |
|||
using Squidex.Shared.Users; |
|||
|
|||
namespace Squidex.Domain.Apps.Entities.Backup |
|||
{ |
|||
[Reentrant] |
|||
public sealed class BackupGrain : GrainBase, IBackupGrain |
|||
{ |
|||
private const int MaxBackups = 10; |
|||
private static readonly Duration UpdateDuration = Duration.FromSeconds(1); |
|||
private readonly IBackupArchiveLocation backupArchiveLocation; |
|||
private readonly IBackupArchiveStore backupArchiveStore; |
|||
private readonly IBackupHandlerFactory backupHandlers; |
|||
private readonly IClock clock; |
|||
private readonly IEventFormatter eventFormatter; |
|||
private readonly IEventStore eventStore; |
|||
private readonly IGrainState<BackupState> state; |
|||
private readonly IUserResolver userResolver; |
|||
private readonly ILogger<BackupGrain> log; |
|||
private CancellationTokenSource? currentJobToken; |
|||
private BackupJob? currentJob; |
|||
|
|||
public BackupGrain(IGrainIdentity identity, |
|||
IBackupArchiveLocation backupArchiveLocation, |
|||
IBackupArchiveStore backupArchiveStore, |
|||
IBackupHandlerFactory backupHandlers, |
|||
IClock clock, |
|||
IEventFormatter eventFormatter, |
|||
IEventStore eventStore, |
|||
IGrainState<BackupState> state, |
|||
IUserResolver userResolver, |
|||
ILogger<BackupGrain> log) |
|||
: base(identity) |
|||
{ |
|||
this.backupArchiveLocation = backupArchiveLocation; |
|||
this.backupArchiveStore = backupArchiveStore; |
|||
this.backupHandlers = backupHandlers; |
|||
this.clock = clock; |
|||
this.eventFormatter = eventFormatter; |
|||
this.eventStore = eventStore; |
|||
this.state = state; |
|||
this.userResolver = userResolver; |
|||
|
|||
this.log = log; |
|||
} |
|||
|
|||
public override Task OnActivateAsync() |
|||
{ |
|||
RecoverAfterRestartAsync().Forget(); |
|||
|
|||
return Task.CompletedTask; |
|||
} |
|||
|
|||
private async Task RecoverAfterRestartAsync() |
|||
{ |
|||
state.Value.Jobs.RemoveAll(x => x.Stopped == null); |
|||
|
|||
await state.WriteAsync(); |
|||
} |
|||
|
|||
public async Task ClearAsync() |
|||
{ |
|||
foreach (var backup in state.Value.Jobs) |
|||
{ |
|||
await backupArchiveStore.DeleteAsync(backup.Id, default); |
|||
} |
|||
|
|||
TryDeactivateOnIdle(); |
|||
|
|||
await state.ClearAsync(); |
|||
} |
|||
|
|||
public async Task BackupAsync(RefToken actor) |
|||
{ |
|||
if (currentJobToken != null) |
|||
{ |
|||
throw new DomainException(T.Get("backups.alreadyRunning")); |
|||
} |
|||
|
|||
if (state.Value.Jobs.Count >= MaxBackups) |
|||
{ |
|||
throw new DomainException(T.Get("backups.maxReached", new { max = MaxBackups })); |
|||
} |
|||
|
|||
var job = new BackupJob |
|||
{ |
|||
Id = DomainId.NewGuid(), |
|||
Started = clock.GetCurrentInstant(), |
|||
Status = JobStatus.Started |
|||
}; |
|||
|
|||
currentJobToken = new CancellationTokenSource(); |
|||
currentJob = job; |
|||
|
|||
state.Value.Jobs.Insert(0, job); |
|||
|
|||
await state.WriteAsync(); |
|||
|
|||
#pragma warning disable MA0042 // Do not use blocking calls in an async method
|
|||
Process(job, actor, currentJobToken.Token); |
|||
#pragma warning restore MA0042 // Do not use blocking calls in an async method
|
|||
} |
|||
|
|||
private void Process(BackupJob job, RefToken actor, |
|||
CancellationToken ct) |
|||
{ |
|||
ProcessAsync(job, actor, ct).Forget(); |
|||
} |
|||
|
|||
private async Task ProcessAsync(BackupJob job, RefToken actor, |
|||
CancellationToken ct) |
|||
{ |
|||
var handlers = backupHandlers.CreateMany(); |
|||
|
|||
var lastTimestamp = job.Started; |
|||
|
|||
try |
|||
{ |
|||
await using (var stream = backupArchiveLocation.OpenStream(job.Id)) |
|||
{ |
|||
using (var writer = await backupArchiveLocation.OpenWriterAsync(stream)) |
|||
{ |
|||
await writer.WriteVersionAsync(); |
|||
|
|||
var userMapping = new UserMapping(actor); |
|||
|
|||
var context = new BackupContext(Key, userMapping, writer); |
|||
|
|||
await foreach (var storedEvent in eventStore.QueryAllAsync(GetFilter(), ct: ct)) |
|||
{ |
|||
var @event = eventFormatter.Parse(storedEvent); |
|||
|
|||
if (@event.Payload is SquidexEvent squidexEvent && squidexEvent.Actor != null) |
|||
{ |
|||
context.UserMapping.Backup(squidexEvent.Actor); |
|||
} |
|||
|
|||
foreach (var handler in handlers) |
|||
{ |
|||
await handler.BackupEventAsync(@event, context, ct); |
|||
} |
|||
|
|||
writer.WriteEvent(storedEvent, ct); |
|||
|
|||
job.HandledEvents = writer.WrittenEvents; |
|||
job.HandledAssets = writer.WrittenAttachments; |
|||
|
|||
lastTimestamp = await WritePeriodically(lastTimestamp); |
|||
} |
|||
|
|||
foreach (var handler in handlers) |
|||
{ |
|||
ct.ThrowIfCancellationRequested(); |
|||
|
|||
await handler.BackupAsync(context, ct); |
|||
} |
|||
|
|||
foreach (var handler in handlers) |
|||
{ |
|||
ct.ThrowIfCancellationRequested(); |
|||
|
|||
await handler.CompleteBackupAsync(context); |
|||
} |
|||
|
|||
await userMapping.StoreAsync(writer, userResolver, ct); |
|||
} |
|||
|
|||
stream.Position = 0; |
|||
|
|||
ct.ThrowIfCancellationRequested(); |
|||
|
|||
await backupArchiveStore.UploadAsync(job.Id, stream, ct); |
|||
} |
|||
|
|||
job.Status = JobStatus.Completed; |
|||
} |
|||
catch (OperationCanceledException) |
|||
{ |
|||
await RemoveAsync(job); |
|||
} |
|||
catch (Exception ex) |
|||
{ |
|||
log.LogError(ex, "Faield to make backup with backup id '{backupId}'.", job.Id); |
|||
|
|||
job.Status = JobStatus.Failed; |
|||
} |
|||
finally |
|||
{ |
|||
job.Stopped = clock.GetCurrentInstant(); |
|||
|
|||
await state.WriteAsync(); |
|||
|
|||
currentJobToken?.Dispose(); |
|||
currentJobToken = null; |
|||
currentJob = null; |
|||
} |
|||
} |
|||
|
|||
private string GetFilter() |
|||
{ |
|||
return $"^[^\\-]*-{Regex.Escape(Key.ToString())}"; |
|||
} |
|||
|
|||
private async Task<Instant> WritePeriodically(Instant lastTimestamp) |
|||
{ |
|||
var now = clock.GetCurrentInstant(); |
|||
|
|||
if ((now - lastTimestamp) >= UpdateDuration) |
|||
{ |
|||
lastTimestamp = now; |
|||
|
|||
await state.WriteAsync(); |
|||
} |
|||
|
|||
return lastTimestamp; |
|||
} |
|||
|
|||
public async Task DeleteAsync(DomainId id) |
|||
{ |
|||
var job = state.Value.Jobs.Find(x => x.Id == id); |
|||
|
|||
if (job == null) |
|||
{ |
|||
throw new DomainObjectNotFoundException(id.ToString()); |
|||
} |
|||
|
|||
if (currentJob == job) |
|||
{ |
|||
try |
|||
{ |
|||
currentJobToken?.Cancel(); |
|||
} |
|||
catch (ObjectDisposedException) |
|||
{ |
|||
return; |
|||
} |
|||
} |
|||
else |
|||
{ |
|||
await RemoveAsync(job); |
|||
} |
|||
} |
|||
|
|||
private async Task RemoveAsync(BackupJob job) |
|||
{ |
|||
try |
|||
{ |
|||
#pragma warning disable MA0040 // Flow the cancellation token
|
|||
await backupArchiveStore.DeleteAsync(job.Id); |
|||
#pragma warning restore MA0040 // Flow the cancellation token
|
|||
} |
|||
catch (Exception ex) |
|||
{ |
|||
log.LogError(ex, "Failed to make remove with backup id '{backupId}'.", job.Id); |
|||
} |
|||
|
|||
state.Value.Jobs.Remove(job); |
|||
|
|||
await state.WriteAsync(); |
|||
} |
|||
|
|||
public Task<List<IBackupJob>> GetStateAsync() |
|||
{ |
|||
return Task.FromResult(state.Value.Jobs.OfType<IBackupJob>().ToList()); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,305 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using System.Text.RegularExpressions; |
|||
using Microsoft.Extensions.Logging; |
|||
using NodaTime; |
|||
using Squidex.Domain.Apps.Entities.Backup.State; |
|||
using Squidex.Domain.Apps.Events; |
|||
using Squidex.Infrastructure; |
|||
using Squidex.Infrastructure.EventSourcing; |
|||
using Squidex.Infrastructure.States; |
|||
using Squidex.Infrastructure.Tasks; |
|||
using Squidex.Infrastructure.Translations; |
|||
using Squidex.Shared.Users; |
|||
|
|||
#pragma warning disable MA0040 // Flow the cancellation token
|
|||
|
|||
namespace Squidex.Domain.Apps.Entities.Backup |
|||
{ |
|||
public sealed class BackupProcessor |
|||
{ |
|||
private static readonly Duration UpdateDuration = Duration.FromSeconds(1); |
|||
private readonly IBackupArchiveLocation backupArchiveLocation; |
|||
private readonly IBackupArchiveStore backupArchiveStore; |
|||
private readonly IBackupHandlerFactory backupHandlerFactory; |
|||
private readonly IEventFormatter eventFormatter; |
|||
private readonly IEventStore eventStore; |
|||
private readonly IUserResolver userResolver; |
|||
private readonly ILogger<BackupProcessor> log; |
|||
private readonly SimpleState<BackupState> state; |
|||
private readonly ReentrantScheduler scheduler = new ReentrantScheduler(1); |
|||
private readonly DomainId appId; |
|||
private Run? currentRun; |
|||
|
|||
// Use a run to store all state that is necessary for a single run.
|
|||
private sealed class Run : IDisposable |
|||
{ |
|||
private readonly CancellationTokenSource cancellationSource = new CancellationTokenSource(); |
|||
private readonly CancellationTokenSource cancellationLinked; |
|||
|
|||
public IEnumerable<IBackupHandler> Handlers { get; init; } |
|||
|
|||
public RefToken Actor { get; init; } |
|||
|
|||
public BackupJob Job { get; init; } |
|||
|
|||
public CancellationToken CancellationToken => cancellationLinked.Token; |
|||
|
|||
public Run(CancellationToken ct) |
|||
{ |
|||
cancellationLinked = CancellationTokenSource.CreateLinkedTokenSource(ct, cancellationSource.Token); |
|||
} |
|||
|
|||
public void Dispose() |
|||
{ |
|||
cancellationSource.Dispose(); |
|||
cancellationLinked.Dispose(); |
|||
} |
|||
|
|||
public void Cancel() |
|||
{ |
|||
try |
|||
{ |
|||
cancellationSource.Cancel(); |
|||
} |
|||
catch (ObjectDisposedException) |
|||
{ |
|||
// Cancellation token might have been disposed, if the run is completed.
|
|||
} |
|||
} |
|||
} |
|||
|
|||
public IClock Clock { get; set; } = SystemClock.Instance; |
|||
|
|||
public BackupProcessor( |
|||
DomainId appId, |
|||
IBackupArchiveLocation backupArchiveLocation, |
|||
IBackupArchiveStore backupArchiveStore, |
|||
IBackupHandlerFactory backupHandlerFactory, |
|||
IEventFormatter eventFormatter, |
|||
IEventStore eventStore, |
|||
IPersistenceFactory<BackupState> persistenceFactory, |
|||
IUserResolver userResolver, |
|||
ILogger<BackupProcessor> log) |
|||
{ |
|||
this.appId = appId; |
|||
this.backupArchiveLocation = backupArchiveLocation; |
|||
this.backupArchiveStore = backupArchiveStore; |
|||
this.backupHandlerFactory = backupHandlerFactory; |
|||
this.eventFormatter = eventFormatter; |
|||
this.eventStore = eventStore; |
|||
this.userResolver = userResolver; |
|||
this.log = log; |
|||
|
|||
state = new SimpleState<BackupState>(persistenceFactory, GetType(), appId); |
|||
} |
|||
|
|||
public async Task LoadAsync( |
|||
CancellationToken ct) |
|||
{ |
|||
await state.LoadAsync(ct); |
|||
|
|||
if (state.Value.Jobs.RemoveAll(x => x.Stopped == null) > 0) |
|||
{ |
|||
await state.WriteAsync(ct); |
|||
} |
|||
} |
|||
|
|||
public Task ClearAsync() |
|||
{ |
|||
return scheduler.ScheduleAsync(async _ => |
|||
{ |
|||
foreach (var backup in state.Value.Jobs) |
|||
{ |
|||
await backupArchiveStore.DeleteAsync(backup.Id, default); |
|||
} |
|||
|
|||
await state.ClearAsync(); |
|||
}); |
|||
} |
|||
|
|||
public Task BackupAsync(RefToken actor, |
|||
CancellationToken ct) |
|||
{ |
|||
return scheduler.ScheduleAsync(async _ => |
|||
{ |
|||
if (currentRun != null) |
|||
{ |
|||
throw new DomainException(T.Get("backups.alreadyRunning")); |
|||
} |
|||
|
|||
state.Value.EnsureCanStart(); |
|||
|
|||
// Set the current run first to indicate that we are running a rule at the moment.
|
|||
var run = currentRun = new Run(ct) |
|||
{ |
|||
Actor = actor, |
|||
Job = new BackupJob |
|||
{ |
|||
Id = DomainId.NewGuid(), |
|||
Started = Clock.GetCurrentInstant(), |
|||
Status = JobStatus.Started |
|||
}, |
|||
Handlers = backupHandlerFactory.CreateMany() |
|||
}; |
|||
|
|||
state.Value.Jobs.Insert(0, run.Job); |
|||
try |
|||
{ |
|||
await state.WriteAsync(run.CancellationToken); |
|||
|
|||
await ProcessAsync(run, run.CancellationToken); |
|||
} |
|||
finally |
|||
{ |
|||
// Unset the run to indicate that we are done.
|
|||
currentRun.Dispose(); |
|||
currentRun = null; |
|||
} |
|||
}); |
|||
} |
|||
|
|||
private async Task ProcessAsync(Run run, |
|||
CancellationToken ct) |
|||
{ |
|||
var lastTimestamp = run.Job.Started; |
|||
try |
|||
{ |
|||
await using (var stream = backupArchiveLocation.OpenStream(run.Job.Id)) |
|||
{ |
|||
using (var writer = await backupArchiveLocation.OpenWriterAsync(stream, ct)) |
|||
{ |
|||
await writer.WriteVersionAsync(); |
|||
|
|||
var backupUsers = new UserMapping(run.Actor); |
|||
var backupContext = new BackupContext(appId, backupUsers, writer); |
|||
|
|||
await foreach (var storedEvent in eventStore.QueryAllAsync(GetFilter(), ct: ct)) |
|||
{ |
|||
var @event = eventFormatter.Parse(storedEvent); |
|||
|
|||
if (@event.Payload is SquidexEvent squidexEvent && squidexEvent.Actor != null) |
|||
{ |
|||
backupUsers.Backup(squidexEvent.Actor); |
|||
} |
|||
|
|||
foreach (var handler in run.Handlers) |
|||
{ |
|||
await handler.BackupEventAsync(@event, backupContext, ct); |
|||
} |
|||
|
|||
writer.WriteEvent(storedEvent, ct); |
|||
|
|||
run.Job.HandledEvents = writer.WrittenEvents; |
|||
run.Job.HandledAssets = writer.WrittenAttachments; |
|||
|
|||
lastTimestamp = await WritePeriodically(lastTimestamp); |
|||
} |
|||
|
|||
foreach (var handler in run.Handlers) |
|||
{ |
|||
ct.ThrowIfCancellationRequested(); |
|||
|
|||
await handler.BackupAsync(backupContext, ct); |
|||
} |
|||
|
|||
foreach (var handler in run.Handlers) |
|||
{ |
|||
ct.ThrowIfCancellationRequested(); |
|||
|
|||
await handler.CompleteBackupAsync(backupContext); |
|||
} |
|||
|
|||
await backupUsers.StoreAsync(writer, userResolver, ct); |
|||
} |
|||
|
|||
stream.Position = 0; |
|||
|
|||
ct.ThrowIfCancellationRequested(); |
|||
|
|||
await backupArchiveStore.UploadAsync(run.Job.Id, stream, ct); |
|||
} |
|||
|
|||
run.Job.Status = JobStatus.Completed; |
|||
} |
|||
catch (OperationCanceledException) |
|||
{ |
|||
await RemoveAsync(run.Job); |
|||
} |
|||
catch (Exception ex) |
|||
{ |
|||
log.LogError(ex, "Faield to make backup with backup id '{backupId}'.", run.Job.Id); |
|||
|
|||
run.Job.Status = JobStatus.Failed; |
|||
} |
|||
finally |
|||
{ |
|||
run.Job.Stopped = Clock.GetCurrentInstant(); |
|||
|
|||
await state.WriteAsync(); |
|||
} |
|||
} |
|||
|
|||
private string GetFilter() |
|||
{ |
|||
return $"^[^\\-]*-{Regex.Escape(appId.ToString())}"; |
|||
} |
|||
|
|||
private async Task<Instant> WritePeriodically(Instant lastTimestamp) |
|||
{ |
|||
var now = Clock.GetCurrentInstant(); |
|||
|
|||
if ((now - lastTimestamp) >= UpdateDuration) |
|||
{ |
|||
lastTimestamp = now; |
|||
|
|||
await state.WriteAsync(); |
|||
} |
|||
|
|||
return lastTimestamp; |
|||
} |
|||
|
|||
public Task DeleteAsync(DomainId id) |
|||
{ |
|||
return scheduler.ScheduleAsync(async _ => |
|||
{ |
|||
var job = state.Value.Jobs.Find(x => x.Id == id); |
|||
|
|||
if (job == null) |
|||
{ |
|||
throw new DomainObjectNotFoundException(id.ToString()); |
|||
} |
|||
|
|||
if (currentRun?.Job == job) |
|||
{ |
|||
currentRun.Cancel(); |
|||
} |
|||
else |
|||
{ |
|||
await RemoveAsync(job); |
|||
} |
|||
}); |
|||
} |
|||
|
|||
private async Task RemoveAsync(BackupJob job) |
|||
{ |
|||
try |
|||
{ |
|||
await backupArchiveStore.DeleteAsync(job.Id); |
|||
} |
|||
catch (Exception ex) |
|||
{ |
|||
log.LogError(ex, "Failed to make remove with backup id '{backupId}'.", job.Id); |
|||
} |
|||
|
|||
state.Value.Jobs.Remove(job); |
|||
|
|||
await state.WriteAsync(); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,90 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using System.Collections.Concurrent; |
|||
using Microsoft.Extensions.DependencyInjection; |
|||
using Squidex.Hosting; |
|||
using Squidex.Infrastructure; |
|||
using Squidex.Messaging; |
|||
|
|||
namespace Squidex.Domain.Apps.Entities.Backup |
|||
{ |
|||
public sealed class BackupWorker : |
|||
IMessageHandler<BackupRestore>, |
|||
IMessageHandler<BackupStart>, |
|||
IMessageHandler<BackupDelete>, |
|||
IMessageHandler<BackupClear>, |
|||
IInitializable |
|||
{ |
|||
private readonly Dictionary<DomainId, Task<BackupProcessor>> backupProcessors = new Dictionary<DomainId, Task<BackupProcessor>>(); |
|||
private readonly Func<DomainId, BackupProcessor> backupFactory; |
|||
private readonly RestoreProcessor restoreProcessor; |
|||
|
|||
public BackupWorker(IServiceProvider serviceProvider) |
|||
{ |
|||
var objectFactory = ActivatorUtilities.CreateFactory(typeof(BackupProcessor), new[] { typeof(DomainId) }); |
|||
|
|||
backupFactory = key => |
|||
{ |
|||
return (BackupProcessor)objectFactory(serviceProvider, new object[] { key }); |
|||
}; |
|||
|
|||
restoreProcessor = serviceProvider.GetRequiredService<RestoreProcessor>(); |
|||
} |
|||
|
|||
public Task InitializeAsync( |
|||
CancellationToken ct) |
|||
{ |
|||
return restoreProcessor.LoadAsync(ct); |
|||
} |
|||
|
|||
public Task HandleAsync(BackupRestore message, |
|||
CancellationToken ct) |
|||
{ |
|||
return restoreProcessor.RestoreAsync(message.Url, message.Actor, message.NewAppName, ct); |
|||
} |
|||
|
|||
public async Task HandleAsync(BackupStart message, |
|||
CancellationToken ct) |
|||
{ |
|||
var processor = await GetBackupProcessorAsync(message.AppId); |
|||
|
|||
await processor.BackupAsync(message.Actor, ct); |
|||
} |
|||
|
|||
public async Task HandleAsync(BackupDelete message, |
|||
CancellationToken ct) |
|||
{ |
|||
var processor = await GetBackupProcessorAsync(message.AppId); |
|||
|
|||
await processor.DeleteAsync(message.Id); |
|||
} |
|||
|
|||
public async Task HandleAsync(BackupClear message, |
|||
CancellationToken ct) |
|||
{ |
|||
var processor = await GetBackupProcessorAsync(message.AppId); |
|||
|
|||
await processor.ClearAsync(); |
|||
} |
|||
|
|||
private Task<BackupProcessor> GetBackupProcessorAsync(DomainId appId) |
|||
{ |
|||
lock (backupProcessors) |
|||
{ |
|||
return backupProcessors.GetOrAdd(appId, async key => |
|||
{ |
|||
var processor = backupFactory(key); |
|||
|
|||
await processor.LoadAsync(default); |
|||
|
|||
return processor; |
|||
}); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
@ -1,435 +0,0 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using System.Threading.Tasks.Dataflow; |
|||
using Microsoft.Extensions.Logging; |
|||
using NodaTime; |
|||
using Orleans.Core; |
|||
using Squidex.Domain.Apps.Core.Apps; |
|||
using Squidex.Domain.Apps.Entities.Apps.Commands; |
|||
using Squidex.Domain.Apps.Entities.Backup.State; |
|||
using Squidex.Domain.Apps.Events; |
|||
using Squidex.Domain.Apps.Events.Apps; |
|||
using Squidex.Infrastructure; |
|||
using Squidex.Infrastructure.Commands; |
|||
using Squidex.Infrastructure.EventSourcing; |
|||
using Squidex.Infrastructure.Orleans; |
|||
using Squidex.Infrastructure.States; |
|||
using Squidex.Infrastructure.Tasks; |
|||
using Squidex.Infrastructure.Translations; |
|||
using Squidex.Shared.Users; |
|||
|
|||
namespace Squidex.Domain.Apps.Entities.Backup |
|||
{ |
|||
public sealed class RestoreGrain : GrainBase, IRestoreGrain |
|||
{ |
|||
private readonly IBackupArchiveLocation backupArchiveLocation; |
|||
private readonly IBackupHandlerFactory backupHandlers; |
|||
private readonly IClock clock; |
|||
private readonly ICommandBus commandBus; |
|||
private readonly IEventFormatter eventFormatter; |
|||
private readonly IEventStore eventStore; |
|||
private readonly IEventStreamNames eventStreams; |
|||
private readonly IGrainState<BackupRestoreState> state; |
|||
private readonly ILogger<RestoreGrain> log; |
|||
private readonly IUserResolver userResolver; |
|||
private RestoreContext runningContext; |
|||
private StreamMapper runningStreamMapper; |
|||
|
|||
private RestoreJob CurrentJob |
|||
{ |
|||
get => state.Value.Job; |
|||
} |
|||
|
|||
public RestoreGrain( |
|||
IBackupArchiveLocation backupArchiveLocation, |
|||
IBackupHandlerFactory backupHandlers, |
|||
IClock clock, |
|||
ICommandBus commandBus, |
|||
IEventFormatter eventFormatter, |
|||
IEventStore eventStore, |
|||
IEventStreamNames eventStreams, |
|||
IGrainIdentity identity, |
|||
IGrainState<BackupRestoreState> state, |
|||
IUserResolver userResolver, |
|||
ILogger<RestoreGrain> log) |
|||
: base(identity) |
|||
{ |
|||
this.backupArchiveLocation = backupArchiveLocation; |
|||
this.backupHandlers = backupHandlers; |
|||
this.clock = clock; |
|||
this.commandBus = commandBus; |
|||
this.eventFormatter = eventFormatter; |
|||
this.eventStore = eventStore; |
|||
this.eventStreams = eventStreams; |
|||
this.state = state; |
|||
this.userResolver = userResolver; |
|||
this.log = log; |
|||
} |
|||
|
|||
public override Task OnActivateAsync() |
|||
{ |
|||
RecoverAfterRestartAsync().Forget(); |
|||
|
|||
return Task.CompletedTask; |
|||
} |
|||
|
|||
private async Task RecoverAfterRestartAsync() |
|||
{ |
|||
if (CurrentJob?.Status == JobStatus.Started) |
|||
{ |
|||
Log("Failed due application restart"); |
|||
|
|||
CurrentJob.Status = JobStatus.Failed; |
|||
|
|||
await state.WriteAsync(); |
|||
} |
|||
} |
|||
|
|||
public async Task RestoreAsync(Uri url, RefToken actor, string? newAppName = null) |
|||
{ |
|||
Guard.NotNull(url); |
|||
Guard.NotNull(actor); |
|||
|
|||
if (!string.IsNullOrWhiteSpace(newAppName)) |
|||
{ |
|||
Guard.ValidSlug(newAppName); |
|||
} |
|||
|
|||
if (CurrentJob?.Status == JobStatus.Started) |
|||
{ |
|||
throw new DomainException(T.Get("backups.restoreRunning")); |
|||
} |
|||
|
|||
state.Value.Job = new RestoreJob |
|||
{ |
|||
Id = DomainId.NewGuid(), |
|||
NewAppName = newAppName, |
|||
Actor = actor, |
|||
Started = clock.GetCurrentInstant(), |
|||
Status = JobStatus.Started, |
|||
Url = url |
|||
}; |
|||
|
|||
await state.WriteAsync(); |
|||
|
|||
#pragma warning disable MA0042 // Do not use blocking calls in an async method
|
|||
Process(); |
|||
#pragma warning restore MA0042 // Do not use blocking calls in an async method
|
|||
} |
|||
|
|||
private void Process() |
|||
{ |
|||
ProcessAsync().Forget(); |
|||
} |
|||
|
|||
private async Task ProcessAsync() |
|||
{ |
|||
var handlers = backupHandlers.CreateMany(); |
|||
|
|||
var ct = default(CancellationToken); |
|||
|
|||
using (Telemetry.Activities.StartActivity("RestoreBackup")) |
|||
{ |
|||
try |
|||
{ |
|||
Log("Started. The restore process has the following steps:"); |
|||
Log(" * Download backup"); |
|||
Log(" * Restore events and attachments."); |
|||
Log(" * Restore all objects like app, schemas and contents"); |
|||
Log(" * Complete the restore operation for all objects"); |
|||
|
|||
log.LogInformation("Backup with job id {backupId} with from URL '{url}' started.", |
|||
CurrentJob.Id, |
|||
CurrentJob.Url); |
|||
|
|||
using (var reader = await DownloadAsync()) |
|||
{ |
|||
await reader.CheckCompatibilityAsync(); |
|||
|
|||
using (Telemetry.Activities.StartActivity("ReadEvents")) |
|||
{ |
|||
await ReadEventsAsync(reader, handlers); |
|||
} |
|||
|
|||
foreach (var handler in handlers) |
|||
{ |
|||
using (Telemetry.Activities.StartActivity($"{handler.GetType().Name}/RestoreAsync")) |
|||
{ |
|||
await handler.RestoreAsync(runningContext, ct); |
|||
} |
|||
|
|||
Log($"Restored {handler.Name}"); |
|||
} |
|||
|
|||
foreach (var handler in handlers) |
|||
{ |
|||
using (Telemetry.Activities.StartActivity($"{handler.GetType().Name}/CompleteRestoreAsync")) |
|||
{ |
|||
await handler.CompleteRestoreAsync(runningContext, CurrentJob.NewAppName!); |
|||
} |
|||
|
|||
Log($"Completed {handler.Name}"); |
|||
} |
|||
} |
|||
|
|||
await AssignContributorAsync(); |
|||
|
|||
CurrentJob.Status = JobStatus.Completed; |
|||
|
|||
Log("Completed, Yeah!"); |
|||
|
|||
log.LogInformation("Backup with job id {backupId} from URL '{url}' completed.", |
|||
CurrentJob.Id, |
|||
CurrentJob.Url); |
|||
} |
|||
catch (Exception ex) |
|||
{ |
|||
switch (ex) |
|||
{ |
|||
case BackupRestoreException backupException: |
|||
Log(backupException.Message); |
|||
break; |
|||
case FileNotFoundException fileNotFoundException: |
|||
Log(fileNotFoundException.Message); |
|||
break; |
|||
default: |
|||
Log("Failed with internal error"); |
|||
break; |
|||
} |
|||
|
|||
await CleanupAsync(handlers); |
|||
|
|||
CurrentJob.Status = JobStatus.Failed; |
|||
|
|||
log.LogError(ex, "Backup with job id {backupId} from URL '{url}' failed.", |
|||
CurrentJob.Id, |
|||
CurrentJob.Url); |
|||
} |
|||
finally |
|||
{ |
|||
CurrentJob.Stopped = clock.GetCurrentInstant(); |
|||
|
|||
await state.WriteAsync(); |
|||
|
|||
runningStreamMapper = null!; |
|||
runningContext = null!; |
|||
} |
|||
} |
|||
} |
|||
|
|||
private async Task AssignContributorAsync() |
|||
{ |
|||
var actor = CurrentJob.Actor; |
|||
|
|||
if (actor?.IsUser == true) |
|||
{ |
|||
try |
|||
{ |
|||
await commandBus.PublishAsync(new AssignContributor |
|||
{ |
|||
Actor = actor, |
|||
AppId = CurrentJob.AppId, |
|||
ContributorId = actor.Identifier, |
|||
Restoring = true, |
|||
Role = Role.Owner |
|||
}); |
|||
|
|||
Log("Assigned current user."); |
|||
} |
|||
catch (DomainException ex) |
|||
{ |
|||
Log($"Failed to assign contributor: {ex.Message}"); |
|||
} |
|||
} |
|||
else |
|||
{ |
|||
Log("Current user not assigned because restore was triggered by client."); |
|||
} |
|||
} |
|||
|
|||
private async Task CleanupAsync(IEnumerable<IBackupHandler> handlers) |
|||
{ |
|||
if (CurrentJob.AppId != null) |
|||
{ |
|||
var appId = CurrentJob.AppId.Id; |
|||
|
|||
foreach (var handler in handlers) |
|||
{ |
|||
try |
|||
{ |
|||
await handler.CleanupRestoreErrorAsync(appId); |
|||
} |
|||
catch (Exception ex) |
|||
{ |
|||
log.LogError(ex, "Failed to clean up restore."); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
|
|||
private async Task<IBackupReader> DownloadAsync() |
|||
{ |
|||
using (Telemetry.Activities.StartActivity("Download")) |
|||
{ |
|||
Log("Downloading Backup"); |
|||
|
|||
var reader = await backupArchiveLocation.OpenReaderAsync(CurrentJob.Url, CurrentJob.Id); |
|||
|
|||
Log("Downloaded Backup"); |
|||
|
|||
return reader; |
|||
} |
|||
} |
|||
|
|||
private async Task ReadEventsAsync(IBackupReader reader, IEnumerable<IBackupHandler> handlers) |
|||
{ |
|||
const int BatchSize = 100; |
|||
|
|||
var handled = 0; |
|||
|
|||
var writeBlock = new ActionBlock<(string, Envelope<IEvent>)[]>(async batch => |
|||
{ |
|||
try |
|||
{ |
|||
var commits = new List<EventCommit>(batch.Length); |
|||
|
|||
foreach (var (stream, @event) in batch) |
|||
{ |
|||
var offset = runningStreamMapper.GetStreamOffset(stream); |
|||
|
|||
commits.Add(EventCommit.Create(stream, offset, @event, eventFormatter)); |
|||
} |
|||
|
|||
await eventStore.AppendUnsafeAsync(commits); |
|||
|
|||
handled += commits.Count; |
|||
|
|||
Log($"Reading {reader.ReadEvents}/{handled} events and {reader.ReadAttachments} attachments completed.", true); |
|||
} |
|||
catch (OperationCanceledException ex) |
|||
{ |
|||
// Dataflow swallows operation cancelled exception.
|
|||
throw new AggregateException(ex); |
|||
} |
|||
}, new ExecutionDataflowBlockOptions |
|||
{ |
|||
MaxDegreeOfParallelism = 1, |
|||
MaxMessagesPerTask = 1, |
|||
BoundedCapacity = 2 |
|||
}); |
|||
|
|||
var batchBlock = new BatchBlock<(string, Envelope<IEvent>)>(BatchSize, new GroupingDataflowBlockOptions |
|||
{ |
|||
BoundedCapacity = BatchSize * 2 |
|||
}); |
|||
|
|||
batchBlock.BidirectionalLinkTo(writeBlock); |
|||
|
|||
await foreach (var job in reader.ReadEventsAsync(eventStreams, eventFormatter)) |
|||
{ |
|||
var newStream = await HandleEventAsync(reader, handlers, job.Stream, job.Event); |
|||
|
|||
if (newStream != null) |
|||
{ |
|||
if (!await batchBlock.SendAsync((newStream, job.Event))) |
|||
{ |
|||
break; |
|||
} |
|||
} |
|||
} |
|||
|
|||
batchBlock.Complete(); |
|||
|
|||
await writeBlock.Completion; |
|||
} |
|||
|
|||
private async Task<string?> HandleEventAsync(IBackupReader reader, IEnumerable<IBackupHandler> handlers, string stream, Envelope<IEvent> @event, |
|||
CancellationToken ct = default) |
|||
{ |
|||
if (@event.Payload is AppCreated appCreated) |
|||
{ |
|||
var previousAppId = appCreated.AppId.Id; |
|||
|
|||
if (!string.IsNullOrWhiteSpace(CurrentJob.NewAppName)) |
|||
{ |
|||
appCreated.Name = CurrentJob.NewAppName; |
|||
|
|||
CurrentJob.AppId = NamedId.Of(DomainId.NewGuid(), CurrentJob.NewAppName); |
|||
} |
|||
else |
|||
{ |
|||
CurrentJob.AppId = NamedId.Of(DomainId.NewGuid(), appCreated.Name); |
|||
} |
|||
|
|||
await CreateContextAsync(reader, previousAppId); |
|||
} |
|||
|
|||
if (@event.Payload is SquidexEvent squidexEvent && squidexEvent.Actor != null) |
|||
{ |
|||
if (runningContext.UserMapping.TryMap(squidexEvent.Actor, out var newUser)) |
|||
{ |
|||
squidexEvent.Actor = newUser; |
|||
} |
|||
} |
|||
|
|||
if (@event.Payload is AppEvent appEvent) |
|||
{ |
|||
appEvent.AppId = CurrentJob.AppId; |
|||
} |
|||
|
|||
var (newStream, id) = runningStreamMapper.Map(stream); |
|||
|
|||
@event.SetAggregateId(id); |
|||
@event.SetRestored(); |
|||
|
|||
foreach (var handler in handlers) |
|||
{ |
|||
if (!await handler.RestoreEventAsync(@event, runningContext, ct)) |
|||
{ |
|||
return null; |
|||
} |
|||
} |
|||
|
|||
return newStream; |
|||
} |
|||
|
|||
private async Task CreateContextAsync(IBackupReader reader, DomainId previousAppId) |
|||
{ |
|||
var userMapping = new UserMapping(CurrentJob.Actor); |
|||
|
|||
using (Telemetry.Activities.StartActivity("CreateUsers")) |
|||
{ |
|||
Log("Creating Users"); |
|||
|
|||
await userMapping.RestoreAsync(reader, userResolver); |
|||
|
|||
Log("Created Users"); |
|||
} |
|||
|
|||
runningContext = new RestoreContext(CurrentJob.AppId.Id, userMapping, reader, previousAppId); |
|||
runningStreamMapper = new StreamMapper(runningContext); |
|||
} |
|||
|
|||
private void Log(string message, bool replace = false) |
|||
{ |
|||
if (replace && CurrentJob.Log.Count > 0) |
|||
{ |
|||
CurrentJob.Log[^1] = $"{clock.GetCurrentInstant()}: {message}"; |
|||
} |
|||
else |
|||
{ |
|||
CurrentJob.Log.Add($"{clock.GetCurrentInstant()}: {message}"); |
|||
} |
|||
} |
|||
|
|||
public Task<IRestoreJob> GetStateAsync() |
|||
{ |
|||
return Task.FromResult<IRestoreJob>(CurrentJob); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,465 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using System.Threading.Tasks.Dataflow; |
|||
using Microsoft.Extensions.Logging; |
|||
using NodaTime; |
|||
using Squidex.Domain.Apps.Core.Apps; |
|||
using Squidex.Domain.Apps.Entities.Apps.Commands; |
|||
using Squidex.Domain.Apps.Entities.Backup.State; |
|||
using Squidex.Domain.Apps.Events; |
|||
using Squidex.Domain.Apps.Events.Apps; |
|||
using Squidex.Infrastructure; |
|||
using Squidex.Infrastructure.Commands; |
|||
using Squidex.Infrastructure.EventSourcing; |
|||
using Squidex.Infrastructure.States; |
|||
using Squidex.Infrastructure.Tasks; |
|||
using Squidex.Infrastructure.Translations; |
|||
using Squidex.Shared.Users; |
|||
|
|||
namespace Squidex.Domain.Apps.Entities.Backup |
|||
{ |
|||
public sealed class RestoreProcessor |
|||
{ |
|||
private readonly IBackupArchiveLocation backupArchiveLocation; |
|||
private readonly IBackupHandlerFactory backupHandlerFactory; |
|||
private readonly ICommandBus commandBus; |
|||
private readonly IEventFormatter eventFormatter; |
|||
private readonly IEventStore eventStore; |
|||
private readonly IEventStreamNames eventStreamNames; |
|||
private readonly IUserResolver userResolver; |
|||
private readonly ILogger<RestoreProcessor> log; |
|||
private readonly ReentrantScheduler scheduler = new ReentrantScheduler(1); |
|||
private readonly SimpleState<BackupRestoreState> state; |
|||
private Run? currentRun; |
|||
|
|||
// Use a run to store all state that is necessary for a single run.
|
|||
private sealed class Run : IDisposable |
|||
{ |
|||
private readonly CancellationTokenSource cancellationSource = new CancellationTokenSource(); |
|||
private readonly CancellationTokenSource cancellationLinked; |
|||
private readonly IClock clock; |
|||
|
|||
public IEnumerable<IBackupHandler> Handlers { get; init; } |
|||
|
|||
public IBackupReader Reader { get; set; } |
|||
|
|||
public RestoreJob Job { get; init; } |
|||
|
|||
public RestoreContext Context { get; set; } |
|||
|
|||
public StreamMapper StreamMapper { get; set; } |
|||
|
|||
public CancellationToken CancellationToken => cancellationLinked.Token; |
|||
|
|||
public Run(IClock clock, CancellationToken ct) |
|||
{ |
|||
cancellationLinked = CancellationTokenSource.CreateLinkedTokenSource(ct, cancellationSource.Token); |
|||
|
|||
this.clock = clock; |
|||
} |
|||
|
|||
public void Log(string message, bool replace = false) |
|||
{ |
|||
if (replace && Job.Log.Count > 0) |
|||
{ |
|||
Job.Log[^1] = $"{clock.GetCurrentInstant()}: {message}"; |
|||
} |
|||
else |
|||
{ |
|||
Job.Log.Add($"{clock.GetCurrentInstant()}: {message}"); |
|||
} |
|||
} |
|||
|
|||
public void Dispose() |
|||
{ |
|||
Reader?.Dispose(); |
|||
|
|||
cancellationSource.Dispose(); |
|||
cancellationLinked.Dispose(); |
|||
} |
|||
|
|||
public void Cancel() |
|||
{ |
|||
try |
|||
{ |
|||
cancellationSource.Cancel(); |
|||
} |
|||
catch (ObjectDisposedException) |
|||
{ |
|||
// Cancellation token might have been disposed, if the run is completed.
|
|||
} |
|||
} |
|||
} |
|||
|
|||
public IClock Clock { get; set; } = SystemClock.Instance; |
|||
|
|||
public RestoreProcessor( |
|||
IBackupArchiveLocation backupArchiveLocation, |
|||
IBackupHandlerFactory backupHandlerFactory, |
|||
ICommandBus commandBus, |
|||
IEventFormatter eventFormatter, |
|||
IEventStore eventStore, |
|||
IEventStreamNames eventStreamNames, |
|||
IPersistenceFactory<BackupRestoreState> persistenceFactory, |
|||
IUserResolver userResolver, |
|||
ILogger<RestoreProcessor> log) |
|||
{ |
|||
this.backupArchiveLocation = backupArchiveLocation; |
|||
this.backupHandlerFactory = backupHandlerFactory; |
|||
this.commandBus = commandBus; |
|||
this.eventFormatter = eventFormatter; |
|||
this.eventStore = eventStore; |
|||
this.eventStreamNames = eventStreamNames; |
|||
this.userResolver = userResolver; |
|||
this.log = log; |
|||
|
|||
state = new SimpleState<BackupRestoreState>(persistenceFactory, GetType(), "Default"); |
|||
} |
|||
|
|||
public async Task LoadAsync( |
|||
CancellationToken ct) |
|||
{ |
|||
await state.LoadAsync(ct); |
|||
|
|||
if (state.Value.Job?.Status == JobStatus.Started) |
|||
{ |
|||
state.Value.Job.Status = JobStatus.Failed; |
|||
|
|||
await state.WriteAsync(ct); |
|||
} |
|||
} |
|||
|
|||
public Task RestoreAsync(Uri url, RefToken actor, string? newAppName, |
|||
CancellationToken ct) |
|||
{ |
|||
Guard.NotNull(url); |
|||
Guard.NotNull(actor); |
|||
|
|||
if (!string.IsNullOrWhiteSpace(newAppName)) |
|||
{ |
|||
Guard.ValidSlug(newAppName); |
|||
} |
|||
|
|||
return scheduler.ScheduleAsync(async ct => |
|||
{ |
|||
if (currentRun != null) |
|||
{ |
|||
throw new DomainException(T.Get("backups.restoreRunning")); |
|||
} |
|||
|
|||
state.Value.Job?.EnsureCanStart(); |
|||
|
|||
// Set the current run first to indicate that we are running a rule at the moment.
|
|||
var run = currentRun = new Run(Clock, ct) |
|||
{ |
|||
Job = new RestoreJob |
|||
{ |
|||
Id = DomainId.NewGuid(), |
|||
NewAppName = newAppName, |
|||
Actor = actor, |
|||
Started = Clock.GetCurrentInstant(), |
|||
Status = JobStatus.Started, |
|||
Url = url |
|||
}, |
|||
Handlers = backupHandlerFactory.CreateMany() |
|||
}; |
|||
|
|||
state.Value.Job = run.Job; |
|||
try |
|||
{ |
|||
await state.WriteAsync(run.CancellationToken); |
|||
|
|||
await ProcessAsync(run, run.CancellationToken); |
|||
} |
|||
finally |
|||
{ |
|||
// Unset the run to indicate that we are done.
|
|||
currentRun.Dispose(); |
|||
currentRun = null; |
|||
} |
|||
}, ct); |
|||
} |
|||
|
|||
private async Task ProcessAsync(Run run, |
|||
CancellationToken ct) |
|||
{ |
|||
using (Telemetry.Activities.StartActivity("RestoreBackup")) |
|||
{ |
|||
try |
|||
{ |
|||
run.Log("Started. The restore process has the following steps:"); |
|||
run.Log(" * Download backup"); |
|||
run.Log(" * Restore events and attachments."); |
|||
run.Log(" * Restore all objects like app, schemas and contents"); |
|||
run.Log(" * Complete the restore operation for all objects"); |
|||
|
|||
log.LogInformation("Backup with job id {backupId} with from URL '{url}' started.", run.Job.Id, run.Job.Url); |
|||
|
|||
run.Reader = await DownloadAsync(run, ct); |
|||
|
|||
await run.Reader.CheckCompatibilityAsync(); |
|||
|
|||
using (Telemetry.Activities.StartActivity("ReadEvents")) |
|||
{ |
|||
await ReadEventsAsync(run, ct); |
|||
} |
|||
|
|||
foreach (var handler in run.Handlers) |
|||
{ |
|||
using (Telemetry.Activities.StartActivity($"{handler.GetType().Name}/RestoreAsync")) |
|||
{ |
|||
await handler.RestoreAsync(run.Context, ct); |
|||
} |
|||
|
|||
run.Log($"Restored {handler.Name}"); |
|||
} |
|||
|
|||
foreach (var handler in run.Handlers) |
|||
{ |
|||
using (Telemetry.Activities.StartActivity($"{handler.GetType().Name}/CompleteRestoreAsync")) |
|||
{ |
|||
await handler.CompleteRestoreAsync(run.Context, run.Job.NewAppName!); |
|||
} |
|||
|
|||
run.Log($"Completed {handler.Name}"); |
|||
} |
|||
|
|||
await AssignContributorAsync(run); |
|||
|
|||
run.Job.Status = JobStatus.Completed; |
|||
run.Log("Completed, Yeah!"); |
|||
|
|||
log.LogInformation("Backup with job id {backupId} from URL '{url}' completed.", run.Job.Id, run.Job.Url); |
|||
} |
|||
catch (Exception ex) |
|||
{ |
|||
switch (ex) |
|||
{ |
|||
case BackupRestoreException backupException: |
|||
run.Log(backupException.Message); |
|||
break; |
|||
case FileNotFoundException fileNotFoundException: |
|||
run.Log(fileNotFoundException.Message); |
|||
break; |
|||
default: |
|||
run.Log("Failed with internal error"); |
|||
break; |
|||
} |
|||
|
|||
await CleanupAsync(run); |
|||
|
|||
run.Job.Status = JobStatus.Failed; |
|||
|
|||
log.LogError(ex, "Backup with job id {backupId} from URL '{url}' failed.", run.Job.Id, run.Job.Url); |
|||
} |
|||
finally |
|||
{ |
|||
run.Job.Stopped = Clock.GetCurrentInstant(); |
|||
|
|||
await state.WriteAsync(ct); |
|||
} |
|||
} |
|||
} |
|||
|
|||
private async Task AssignContributorAsync(Run run) |
|||
{ |
|||
if (run.Job.Actor?.IsUser != true) |
|||
{ |
|||
run.Log("Current user not assigned because restore was triggered by client."); |
|||
return; |
|||
} |
|||
|
|||
try |
|||
{ |
|||
var command = new AssignContributor |
|||
{ |
|||
Actor = run.Job.Actor, |
|||
AppId = run.Job.AppId, |
|||
ContributorId = run.Job.Actor.Identifier, |
|||
Restoring = true, |
|||
Role = Role.Owner |
|||
}; |
|||
|
|||
await commandBus.PublishAsync(command, default); |
|||
|
|||
run.Log("Assigned current user."); |
|||
} |
|||
catch (DomainException ex) |
|||
{ |
|||
run.Log($"Failed to assign contributor: {ex.Message}"); |
|||
} |
|||
} |
|||
|
|||
private async Task CleanupAsync(Run run) |
|||
{ |
|||
if (run.Job.AppId == null) |
|||
{ |
|||
return; |
|||
} |
|||
|
|||
foreach (var handler in run.Handlers) |
|||
{ |
|||
try |
|||
{ |
|||
await handler.CleanupRestoreErrorAsync(run.Job.AppId.Id); |
|||
} |
|||
catch (Exception ex) |
|||
{ |
|||
log.LogError(ex, "Failed to clean up restore."); |
|||
} |
|||
} |
|||
} |
|||
|
|||
private async Task<IBackupReader> DownloadAsync(Run run, |
|||
CancellationToken ct) |
|||
{ |
|||
using (Telemetry.Activities.StartActivity("Download")) |
|||
{ |
|||
run.Log("Downloading Backup"); |
|||
|
|||
var reader = await backupArchiveLocation.OpenReaderAsync(run.Job.Url, run.Job.Id, ct); |
|||
|
|||
run.Log("Downloaded Backup"); |
|||
|
|||
return reader; |
|||
} |
|||
} |
|||
|
|||
private async Task ReadEventsAsync(Run run, |
|||
CancellationToken ct) |
|||
{ |
|||
const int BatchSize = 100; |
|||
|
|||
var handled = 0; |
|||
|
|||
var writeBlock = new ActionBlock<(string, Envelope<IEvent>)[]>(async batch => |
|||
{ |
|||
try |
|||
{ |
|||
var commits = new List<EventCommit>(batch.Length); |
|||
|
|||
foreach (var (stream, @event) in batch) |
|||
{ |
|||
var offset = run.StreamMapper.GetStreamOffset(stream); |
|||
|
|||
commits.Add(EventCommit.Create(stream, offset, @event, eventFormatter)); |
|||
} |
|||
|
|||
await eventStore.AppendUnsafeAsync(commits, ct); |
|||
|
|||
handled += commits.Count; |
|||
|
|||
run.Log($"Reading {run.Reader.ReadEvents}/{handled} events and {run.Reader.ReadAttachments} attachments completed.", true); |
|||
} |
|||
catch (OperationCanceledException ex) |
|||
{ |
|||
// Dataflow swallows operation cancelled exception.
|
|||
throw new AggregateException(ex); |
|||
} |
|||
}, new ExecutionDataflowBlockOptions |
|||
{ |
|||
MaxDegreeOfParallelism = 1, |
|||
MaxMessagesPerTask = 1, |
|||
BoundedCapacity = 2 |
|||
}); |
|||
|
|||
var batchBlock = new BatchBlock<(string, Envelope<IEvent>)>(BatchSize, new GroupingDataflowBlockOptions |
|||
{ |
|||
BoundedCapacity = BatchSize * 2 |
|||
}); |
|||
|
|||
batchBlock.BidirectionalLinkTo(writeBlock); |
|||
|
|||
await foreach (var job in run.Reader.ReadEventsAsync(eventStreamNames, eventFormatter, ct)) |
|||
{ |
|||
var newStream = await HandleEventAsync(run, job.Stream, job.Event, ct); |
|||
|
|||
if (newStream != null) |
|||
{ |
|||
if (!await batchBlock.SendAsync((newStream, job.Event))) |
|||
{ |
|||
break; |
|||
} |
|||
} |
|||
} |
|||
|
|||
batchBlock.Complete(); |
|||
|
|||
await writeBlock.Completion; |
|||
} |
|||
|
|||
private async Task<string?> HandleEventAsync(Run run, string stream, Envelope<IEvent> @event, |
|||
CancellationToken ct = default) |
|||
{ |
|||
if (@event.Payload is AppCreated appCreated) |
|||
{ |
|||
var previousAppId = appCreated.AppId.Id; |
|||
|
|||
if (!string.IsNullOrWhiteSpace(run.Job.NewAppName)) |
|||
{ |
|||
appCreated.Name = run.Job.NewAppName; |
|||
|
|||
run.Job.AppId = NamedId.Of(DomainId.NewGuid(), run.Job.NewAppName); |
|||
} |
|||
else |
|||
{ |
|||
run.Job.AppId = NamedId.Of(DomainId.NewGuid(), appCreated.Name); |
|||
} |
|||
|
|||
await CreateContextAsync(run, previousAppId, ct); |
|||
|
|||
run.StreamMapper = new StreamMapper(run.Context); |
|||
} |
|||
|
|||
if (@event.Payload is SquidexEvent squidexEvent && squidexEvent.Actor != null) |
|||
{ |
|||
if (run.Context.UserMapping.TryMap(squidexEvent.Actor, out var newUser)) |
|||
{ |
|||
squidexEvent.Actor = newUser; |
|||
} |
|||
} |
|||
|
|||
if (@event.Payload is AppEvent appEvent) |
|||
{ |
|||
appEvent.AppId = run.Job.AppId; |
|||
} |
|||
|
|||
var (newStream, id) = run.StreamMapper.Map(stream); |
|||
|
|||
@event.SetAggregateId(id); |
|||
@event.SetRestored(); |
|||
|
|||
foreach (var handler in run.Handlers) |
|||
{ |
|||
if (!await handler.RestoreEventAsync(@event, run.Context, ct)) |
|||
{ |
|||
return null; |
|||
} |
|||
} |
|||
|
|||
return newStream; |
|||
} |
|||
|
|||
private async Task CreateContextAsync(Run run, DomainId previousAppId, |
|||
CancellationToken ct) |
|||
{ |
|||
var userMapping = new UserMapping(run.Job.Actor); |
|||
|
|||
using (Telemetry.Activities.StartActivity("CreateUsers")) |
|||
{ |
|||
run.Log("Creating Users"); |
|||
|
|||
await userMapping.RestoreAsync(run.Reader, userResolver, ct); |
|||
|
|||
run.Log("Created Users"); |
|||
} |
|||
|
|||
run.Context = new RestoreContext(run.Job.AppId.Id, userMapping, run.Reader, previousAppId); |
|||
} |
|||
} |
|||
} |
|||
@ -1,21 +0,0 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using Orleans; |
|||
using Squidex.Domain.Apps.Entities.Comments.Commands; |
|||
using Squidex.Infrastructure; |
|||
using Squidex.Infrastructure.Commands; |
|||
|
|||
namespace Squidex.Domain.Apps.Entities.Comments.DomainObject |
|||
{ |
|||
public interface ICommentsGrain : IGrainWithStringKey |
|||
{ |
|||
Task<CommandResult> ExecuteAsync(CommentsCommand command); |
|||
|
|||
Task<CommentsResult> GetCommentsAsync(long sinceVersion = EtagVersion.Any); |
|||
} |
|||
} |
|||
@ -1,35 +0,0 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using Orleans; |
|||
using Squidex.Infrastructure; |
|||
|
|||
namespace Squidex.Domain.Apps.Entities.Comments |
|||
{ |
|||
public sealed class GrainWatchingService : IWatchingService |
|||
{ |
|||
private readonly IGrainFactory grainFactory; |
|||
|
|||
public GrainWatchingService(IGrainFactory grainFactory) |
|||
{ |
|||
this.grainFactory = grainFactory; |
|||
} |
|||
|
|||
public Task<string[]> GetWatchingUsersAsync(DomainId appId, string resource, string userId) |
|||
{ |
|||
Guard.NotNullOrEmpty(resource); |
|||
Guard.NotNullOrEmpty(userId); |
|||
|
|||
return GetGrain(appId).GetWatchingUsersAsync(resource, userId); |
|||
} |
|||
|
|||
private IWatchingGrain GetGrain(DomainId appId) |
|||
{ |
|||
return grainFactory.GetGrain<IWatchingGrain>(appId.ToString()); |
|||
} |
|||
} |
|||
} |
|||
@ -1,16 +0,0 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using Orleans; |
|||
|
|||
namespace Squidex.Domain.Apps.Entities.Comments |
|||
{ |
|||
public interface IWatchingGrain : IGrainWithStringKey |
|||
{ |
|||
Task<string[]> GetWatchingUsersAsync(string resource, string userId); |
|||
} |
|||
} |
|||
@ -1,86 +0,0 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using NodaTime; |
|||
using Orleans.Core; |
|||
using Squidex.Infrastructure; |
|||
using Squidex.Infrastructure.Orleans; |
|||
|
|||
namespace Squidex.Domain.Apps.Entities.Comments |
|||
{ |
|||
public sealed class WatchingGrain : GrainBase, IWatchingGrain |
|||
{ |
|||
private static readonly Duration Timeout = Duration.FromMinutes(1); |
|||
private readonly Dictionary<string, Dictionary<string, Instant>> users = new Dictionary<string, Dictionary<string, Instant>>(); |
|||
private readonly IClock clock; |
|||
|
|||
public WatchingGrain(IGrainIdentity grainIdentity, IClock clock) |
|||
: base(grainIdentity) |
|||
{ |
|||
this.clock = clock; |
|||
} |
|||
|
|||
public override Task OnActivateAsync() |
|||
{ |
|||
var time = TimeSpan.FromSeconds(30); |
|||
|
|||
RegisterTimer(x => |
|||
{ |
|||
Cleanup(); |
|||
|
|||
return Task.CompletedTask; |
|||
}, null, time, time); |
|||
|
|||
return Task.CompletedTask; |
|||
} |
|||
|
|||
public Task<string[]> GetWatchingUsersAsync(string resource, string userId) |
|||
{ |
|||
Guard.NotNullOrEmpty(resource); |
|||
Guard.NotNullOrEmpty(userId); |
|||
|
|||
var usersByResource = users.GetOrAddNew(resource); |
|||
|
|||
usersByResource[userId] = clock.GetCurrentInstant(); |
|||
|
|||
return Task.FromResult(usersByResource.Keys.ToArray()); |
|||
} |
|||
|
|||
public void Cleanup() |
|||
{ |
|||
if (users.Count == 0) |
|||
{ |
|||
return; |
|||
} |
|||
|
|||
var now = clock.GetCurrentInstant(); |
|||
|
|||
foreach (var (resource, usersByResource) in users.ToList()) |
|||
{ |
|||
foreach (var (userId, lastSeen) in usersByResource.ToList()) |
|||
{ |
|||
var timeSinceLastSeen = now - lastSeen; |
|||
|
|||
if (timeSinceLastSeen > Timeout) |
|||
{ |
|||
usersByResource.Remove(userId); |
|||
} |
|||
} |
|||
|
|||
if (usersByResource.Count == 0) |
|||
{ |
|||
users.Remove(resource); |
|||
} |
|||
} |
|||
|
|||
if (users.Count == 0) |
|||
{ |
|||
TryDeactivateOnIdle(); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,62 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using NodaTime; |
|||
using Squidex.Infrastructure; |
|||
using Squidex.Infrastructure.States; |
|||
|
|||
namespace Squidex.Domain.Apps.Entities.Comments |
|||
{ |
|||
public sealed class WatchingService : IWatchingService |
|||
{ |
|||
private readonly IPersistenceFactory<State> persistenceFactory; |
|||
|
|||
[CollectionName("Watches")] |
|||
public sealed class State |
|||
{ |
|||
private static readonly Duration Timeout = Duration.FromMinutes(1); |
|||
|
|||
public Dictionary<string, Instant> Users { get; } = new Dictionary<string, Instant>(); |
|||
|
|||
public string[] Add(string watcherId, IClock clock) |
|||
{ |
|||
var now = clock.GetCurrentInstant(); |
|||
|
|||
foreach (var (userId, lastSeen) in Users.ToList()) |
|||
{ |
|||
var timeSinceLastSeen = now - lastSeen; |
|||
|
|||
if (timeSinceLastSeen > Timeout) |
|||
{ |
|||
Users.Remove(userId); |
|||
} |
|||
} |
|||
|
|||
Users[watcherId] = now; |
|||
|
|||
return Users.Keys.ToArray(); |
|||
} |
|||
} |
|||
|
|||
public IClock Clock { get; set; } = SystemClock.Instance; |
|||
|
|||
public WatchingService(IPersistenceFactory<State> persistenceFactory) |
|||
{ |
|||
this.persistenceFactory = persistenceFactory; |
|||
} |
|||
|
|||
public async Task<string[]> GetWatchingUsersAsync(DomainId appId, string? resource, string userId, |
|||
CancellationToken ct = default) |
|||
{ |
|||
var state = new SimpleState<State>(persistenceFactory, GetType(), $"{appId}_{resource}"); |
|||
|
|||
await state.LoadAsync(ct); |
|||
|
|||
return await state.UpdateAsync(x => x.Add(userId, Clock), ct: ct); |
|||
} |
|||
} |
|||
} |
|||
Some files were not shown because too many files changed in this diff
Loading…
Reference in new issue