using System.Collections.Immutable; using Connected.Caching; using Connected.Entities; using Connected.Entities.Storage; using Connected.Notifications.Events; using Connected.ServiceModel; using Connected.Services; namespace Logistics.Types.Serials; internal sealed class SerialOps { /// public sealed class Delete : ServiceAction> { public Delete(IStorageProvider storage, ISerialService serials, IEventService events, ICachingService cache) { Storage = storage; Serials = serials; Events = events; Cache = cache; } private IStorageProvider Storage { get; } private ISerialService Serials { get; } private IEventService Events { get; } private ICachingService Cache { get; } protected override async Task OnInvoke() { if (await Serials.Select(Arguments.Id) is not ISerial serial) return; /* * Setting state of the entity enable other middleware to use the entity even after it * is deleted. For example, IEventListener will receive the state of this operation. */ SetState(serial); /* * Perform delete. */ await Storage.Open().Update(Arguments.AsEntity(State.Deleted)); } protected override async Task OnCommitted() { /* * Remove entity from the cache. */ await Cache.Remove(Serial.EntityKey, Arguments.Id); /* * Enqueue event so event listeners can respond to the transaction. */ await Events.Enqueue(this, Events, nameof(ISerialService.Deleted), Arguments); } } /// public sealed class Insert : ServiceFunction { public Insert(IStorageProvider storage, IEventService events) { Storage = storage; Events = events; } private IStorageProvider Storage { get; } private IEventService Events { get; } protected override async Task OnInvoke() { /* * Perform insert and return the newly inserted id. */ return (await Storage.Open().Update(Arguments.AsEntity(State.New))).Id; } protected override async Task OnCommitted() { await Events.Enqueue(this, Events, nameof(ISerialService.Inserted), Arguments); } } /// public sealed class Query : ServiceFunction> { public Query(IStorageProvider storage) { Storage = storage; } private IStorageProvider Storage { get; } protected override async Task> OnInvoke() { /* * For non cached entities query always hits the storage. */ return await (from e in Storage.Open() select e).WithArguments(Arguments).AsEntities(); } } /// public sealed class Lookup : ServiceFunction, ImmutableList> { public Lookup(IStorageProvider storage) { Storage = storage; } private IStorageProvider Storage { get; } protected override async Task> OnInvoke() { return await (from e in Storage.Open() where Arguments.IdList.Any(f => f == e.Id) select e).AsEntities(); } } /// public sealed class Select : NullableServiceFunction, ISerial> { public Select(IStorageProvider storage, ICachingService cache) { Storage = storage; Cache = cache; } private IStorageProvider Storage { get; } private ICachingService Cache { get; } protected override async Task OnInvoke() { /* * First, try to receive entity from the cache. If it doesn't exist in the cache * load it from storage. If storage returns non null value, store it in the cache * for subsequent calls. The entity gets remove either because of inactivity or when * updating or deleting it. */ return await Cache.Get(Serial.EntityKey, Arguments.Id, async (f) => { /* * Doesn't exist in the cache. Let's do the storage action. */ return await (from e in Storage.Open() where e.Id == Arguments.Id select e).AsEntity(); }); } } /// public sealed class SelectByValue : NullableServiceFunction { public SelectByValue(IStorageProvider storage, ICachingService cache) { Storage = storage; Cache = cache; } private IStorageProvider Storage { get; } private ICachingService Cache { get; } protected override async Task OnInvoke() { return await Cache.Get(Serial.EntityKey, f => string.Equals(f.Value, Arguments.Value, StringComparison.OrdinalIgnoreCase), async (f) => { return await (from e in Storage.Open() where string.Equals(e.Value, Arguments.Value, StringComparison.OrdinalIgnoreCase) select e).AsEntity(); }); } } /// public sealed class Update : ServiceAction { public Update(IStorageProvider storage, ICachingService cache, ISerialService packingService, IEventService events) { Storage = storage; Cache = cache; Serials = packingService; Events = events; } private IStorageProvider Storage { get; } private ICachingService Cache { get; } private ISerialService Serials { get; } private IEventService Events { get; } protected override async Task OnInvoke() { /* * Set the state of the unchanged entity. This enable other middleware to * calculate quantity delta, for example because they receive the state of * this operation. Not all middleware supports this, the primary example of * such state client is IEventListener. */ if (SetState(await Serials.Select(Arguments.Id)) is not Serial entity) return; /* * Sinc ethis is concurrent entity we must perform retry if the concurrency fails. */ await Storage.Open().Update(entity.Merge(Arguments, State.Default), Arguments, async () => { /* * The update failed because of concurrency. Remove the entity from the cache to ensure * it gets loaded from the storage next time with fresh values and try again. */ await Cache.Remove(Serial.EntityKey, Arguments.Id); /* * Since the entity reloaded we must overwrite its state. */ return SetState(await Serials.Select(Arguments.Id)) as Serial; }); } protected override async Task OnCommitted() { await Cache.Remove(Serial.EntityKey, Arguments.Id); await Events.Enqueue(this, Serials, nameof(Serials.Updated), Arguments); } } }