| | | 1 | | using Newtonsoft.Json; |
| | | 2 | | using Syki.Back.Events; |
| | | 3 | | using System.Diagnostics; |
| | | 4 | | using Syki.Back.Database; |
| | | 5 | | using Syki.Daemon.Configs; |
| | | 6 | | using System.Collections.Concurrent; |
| | | 7 | | |
| | | 8 | | namespace Syki.Daemon.Events; |
| | | 9 | | |
| | 2658 | 10 | | public class DomainEventsProcessor(IServiceScopeFactory serviceScopeFactory) |
| | | 11 | | { |
| | 2 | 12 | | private static readonly ActivitySource _activitySource = new (OpenTelemetryConfigs.DomainEventsProcessing); |
| | | 13 | | |
| | | 14 | | public async Task Run() |
| | | 15 | | { |
| | 2658 | 16 | | using var scope = serviceScopeFactory.CreateScope(); |
| | 2658 | 17 | | var ctx = scope.ServiceProvider.GetRequiredService<SykiDbContext>(); |
| | | 18 | | |
| | 2658 | 19 | | await Process(scope, ctx, Guid.CreateVersion7()); |
| | 2658 | 20 | | } |
| | | 21 | | |
| | | 22 | | private static async Task Process(IServiceScope scope, SykiDbContext ctx, Guid processorId) |
| | | 23 | | { |
| | 5184 | 24 | | var events = await ctx.DomainEvents.FromSqlRaw(Sql, processorId).ToListAsync(); |
| | 7842 | 25 | | if (events.Count == 0) return; |
| | | 26 | | |
| | 2526 | 27 | | await ctx.Database.BeginTransactionAsync(); |
| | 2526 | 28 | | ctx.Database.AutoSavepointsEnabled = false; |
| | 2526 | 29 | | var sw = Stopwatch.StartNew(); |
| | | 30 | | |
| | 10364 | 31 | | foreach (var evt in events) |
| | | 32 | | { |
| | 2656 | 33 | | using (var activity = _activitySource.StartActivity($"Process {evt.Type.Split('.').Last()}", ActivityKind.Co |
| | | 34 | | { |
| | 2656 | 35 | | activity?.SetTag("event.id", evt.Id); |
| | 2656 | 36 | | activity?.SetTag("event.type", evt.Type.Split('.').Last()); |
| | 2656 | 37 | | activity?.SetTag("event.institutionId", evt.InstitutionId); |
| | | 38 | | |
| | | 39 | | try |
| | | 40 | | { |
| | 2656 | 41 | | sw.Restart(); |
| | 2656 | 42 | | dynamic data = GetData(evt); |
| | 2656 | 43 | | dynamic handler = GetHandler(scope, evt); |
| | 2656 | 44 | | await handler.Handle(evt.InstitutionId, evt.Id, data); |
| | 2656 | 45 | | } |
| | 0 | 46 | | catch (Exception ex) |
| | | 47 | | { |
| | 0 | 48 | | evt.Error = ex.Message + ex.InnerException?.Message; |
| | 0 | 49 | | activity?.SetStatus(ActivityStatusCode.Error, ex.Message); |
| | 0 | 50 | | activity?.AddException(ex); |
| | 0 | 51 | | } |
| | | 52 | | finally |
| | | 53 | | { |
| | 2656 | 54 | | sw.Stop(); |
| | 2656 | 55 | | evt.Processed(sw.Elapsed.TotalMilliseconds); |
| | 2656 | 56 | | activity?.SetTag("event.duration", evt.Duration); |
| | | 57 | | } |
| | 2656 | 58 | | } |
| | 2656 | 59 | | } |
| | | 60 | | |
| | 2526 | 61 | | await ctx.SaveChangesAsync(); |
| | 2526 | 62 | | await ctx.Database.CommitTransactionAsync(); |
| | | 63 | | |
| | 2526 | 64 | | ctx.ChangeTracker.Clear(); |
| | | 65 | | |
| | 2526 | 66 | | await Process(scope, ctx, processorId); |
| | 5184 | 67 | | } |
| | | 68 | | |
| | 2 | 69 | | private static readonly ConcurrentDictionary<string, Type> _types = new(); |
| | | 70 | | private static dynamic GetData(DomainEvent evt) |
| | | 71 | | { |
| | 2656 | 72 | | var type = _types.GetOrAdd(evt.Type, typeof(DomainEvent).Assembly.GetType(evt.Type)!); |
| | 2656 | 73 | | dynamic data = JsonConvert.DeserializeObject(evt.Data, type)!; |
| | 2656 | 74 | | return data; |
| | | 75 | | } |
| | | 76 | | |
| | 2 | 77 | | private static readonly ConcurrentDictionary<string, Type> _handlers = new(); |
| | | 78 | | private static dynamic GetHandler(IServiceScope scope, DomainEvent evt) |
| | | 79 | | { |
| | 2656 | 80 | | var handlerType = _handlers.GetOrAdd(evt.Type, typeof(IDomainEvent).Assembly.GetType($"{evt.Type}Handler")!); |
| | 2656 | 81 | | dynamic handler = scope.ServiceProvider.GetRequiredService(handlerType); |
| | 2656 | 82 | | return handler; |
| | | 83 | | } |
| | | 84 | | |
| | 2 | 85 | | private static readonly string Sql = @" |
| | 2 | 86 | | UPDATE syki.domain_events |
| | 2 | 87 | | SET processor_id = {0}, status = 'Processing' |
| | 2 | 88 | | WHERE ctid IN ( |
| | 2 | 89 | | SELECT ctid |
| | 2 | 90 | | FROM syki.domain_events |
| | 2 | 91 | | WHERE processor_id IS NULL |
| | 2 | 92 | | ORDER BY occurred_at |
| | 2 | 93 | | FOR UPDATE SKIP LOCKED |
| | 2 | 94 | | LIMIT 100 |
| | 2 | 95 | | ) |
| | 2 | 96 | | RETURNING *; |
| | 2 | 97 | | "; |
| | | 98 | | } |