huge queue refactoring:

3 queues -> 1 queue send delete validate updates
2026-04-23 03:36:31 +00:00 · 2025-05-17 19:25:33 +03:00
parent 87ceaa805b
commit d02550f5a0
20 changed files with 246 additions and 193 deletions
@@ -30,13 +30,16 @@ public class AnnotationService : IAnnotationService, INotificationHandler<Annota
    private readonly QueueConfig _queueConfig;
    private Consumer _consumer = null!;
    private readonly UIConfig _uiConfig;
+    private readonly DirectoriesConfig _dirConfig;
    private static readonly Guid SaveTaskId = Guid.NewGuid();
+    private static readonly Guid SaveQueueOffsetTaskId = Guid.NewGuid();

    public AnnotationService(
        IDbFactory dbFactory,
        FailsafeAnnotationsProducer producer,
        IOptions<QueueConfig> queueConfig,
        IOptions<UIConfig> uiConfig,
+        IOptions<DirectoriesConfig> directoriesConfig,
        IGalleryService galleryService,
        IMediator mediator,
        IAzaionApi api)
@@ -48,11 +51,12 @@ public class AnnotationService : IAnnotationService, INotificationHandler<Annota
        _api = api;
        _queueConfig = queueConfig.Value;
        _uiConfig = uiConfig.Value;
+        _dirConfig = directoriesConfig.Value;

-        Task.Run(async () => await Init()).Wait();
+        Task.Run(async () => await InitQueueConsumer()).Wait();
    }

-    private async Task Init(CancellationToken cancellationToken = default)
+    private async Task InitQueueConsumer(CancellationToken cancellationToken = default)
    {
        if (!_api.CurrentUser.Role.IsValidator())
            return;
@@ -72,29 +76,40 @@ public class AnnotationService : IAnnotationService, INotificationHandler<Annota
            OffsetSpec = new OffsetTypeOffset(offsets.AnnotationsOffset + 1),
            MessageHandler = async (_, _, context, message) =>
            {
-                var msg = MessagePackSerializer.Deserialize<AnnotationCreatedMessage>(message.Data.Contents);
+                var email = (string)message.ApplicationProperties[nameof(User.Email)]!;
+                if (email == _api.CurrentUser.Email) //Don't process messages by yourself
+                    return;
+                var annotationStatus = (AnnotationStatus)message.ApplicationProperties[nameof(AnnotationStatus)];
+                if (annotationStatus.In(AnnotationStatus.Created, AnnotationStatus.ValidatedEdited))
+                {
+                    var msg = MessagePackSerializer.Deserialize<AnnotationMessage>(message.Data.Contents);
+                    await SaveAnnotationInner(
+                        msg.CreatedDate,
+                        msg.OriginalMediaName,
+                        msg.Time,
+                        JsonConvert.DeserializeObject<List<Detection>>(msg.Detections) ?? [],
+                        msg.Source,
+                        new MemoryStream(msg.Image),
+                        msg.Role,
+                        msg.Email,
+                        fromQueue: true,
+                        token: cancellationToken);
+                }
+                else
+                {
+                    var msg = MessagePackSerializer.Deserialize<AnnotationBulkMessage>(message.Data.Contents);
+                    if (annotationStatus == AnnotationStatus.Validated)
+                        await ValidateAnnotations(msg.AnnotationNames.ToList(), true, cancellationToken);
+                    if (annotationStatus == AnnotationStatus.Deleted)
+                        await _mediator.Publish(new AnnotationsDeletedEvent(msg.AnnotationNames.ToList(), fromQueue:true), cancellationToken);
+                }

                offsets.AnnotationsOffset = context.Offset;
                ThrottleExt.Throttle(() =>
                {
                    _api.UpdateOffsets(offsets);
                    return Task.CompletedTask;
-                }, SaveTaskId, TimeSpan.FromSeconds(10), scheduleCallAfterCooldown: true);
-
-                if (msg.CreatedEmail == _api.CurrentUser.Email) //Don't process messages by yourself
-                    return;
-
-                await SaveAnnotationInner(
-                    msg.CreatedDate,
-                    msg.OriginalMediaName,
-                    msg.Time,
-                    JsonConvert.DeserializeObject<List<Detection>>(msg.Detections) ?? [],
-                    msg.Source,
-                    new MemoryStream(msg.Image),
-                    msg.CreatedRole,
-                    msg.CreatedEmail,
-                    fromQueue: true,
-                    token: cancellationToken);
+                }, SaveQueueOffsetTaskId, TimeSpan.FromSeconds(10), scheduleCallAfterCooldown: true);
            }
        });
    }
@@ -120,7 +135,7 @@ public class AnnotationService : IAnnotationService, INotificationHandler<Annota
        bool fromQueue = false,
        CancellationToken token = default)
    {
-        AnnotationStatus status;
+        var status = AnnotationStatus.Created;
        var fName = originalMediaName.ToTimeName(time);
        var annotation = await _dbFactory.Run(async db =>
        {
@@ -128,25 +143,41 @@ public class AnnotationService : IAnnotationService, INotificationHandler<Annota
                .LoadWith(x => x.Detections)
                .FirstOrDefaultAsync(x => x.Name == fName, token: token);

-            status = userRole.IsValidator() && source == SourceEnum.Manual
-                ? AnnotationStatus.Validated
-                : AnnotationStatus.Created;
+            if (userRole.IsValidator() && source == SourceEnum.Manual)
+                status = AnnotationStatus.Validated;

            if (fromQueue && ann is { AnnotationStatus: AnnotationStatus.Validated })
                return ann;

            await db.Detections.DeleteAsync(x => x.AnnotationName == fName, token: token);

-            if (ann != null)
+            if (ann != null) //Annotation is already exists
            {
-                await db.Annotations
+                var annotationUpdatable = db.Annotations
                    .Where(x => x.Name == fName)
                    .Set(x => x.Source, source)
+                    .Set(x => x.CreatedRole, userRole);
+
+                if (status == AnnotationStatus.Validated)
+                {
+                    if (status == AnnotationStatus.Validated)
+                        status = AnnotationStatus.ValidatedEdited; //For further processing mark Annotations *edited* by Validator, not just simply Validated by button.
+
+                    annotationUpdatable = annotationUpdatable
+                        .Set(x => x.ValidateDate, createdDate)
+                        .Set(x => x.ValidateEmail, createdEmail);
+                }
+                else
+                {
+                    annotationUpdatable = annotationUpdatable
+                        .Set(x => x.CreatedDate, createdDate)
+                        .Set(x => x.CreatedEmail, createdEmail);
+                }
+
+                await annotationUpdatable
                    .Set(x => x.AnnotationStatus, status)
-                    .Set(x => x.CreatedDate, createdDate)
-                    .Set(x => x.CreatedEmail, createdEmail)
-                    .Set(x => x.CreatedRole, userRole)
                    .UpdateAsync(token: token);
+
                ann.Detections = detections;
            }
            else
@@ -184,10 +215,11 @@ public class AnnotationService : IAnnotationService, INotificationHandler<Annota
        if (_uiConfig.GenerateAnnotatedImage)
            await _galleryService.CreateAnnotatedImage(annotation, token);

-        if (!fromQueue && !_uiConfig.SilentDetection) //Send to queue only if we're not getting from queue already
-            await _producer.SendToInnerQueue(annotation, token);
-
        await _mediator.Publish(new AnnotationCreatedEvent(annotation), token);
+
+        if (!fromQueue && !_uiConfig.SilentDetection) //Send to queue only if we're not getting from queue already
+            await _producer.SendToInnerQueue([annotation.Name], status, token);
+
        ThrottleExt.Throttle(async () =>
        {
            _dbFactory.SaveToDisk();
@@ -196,12 +228,12 @@ public class AnnotationService : IAnnotationService, INotificationHandler<Annota
        return annotation;
    }

-    public async Task ValidateAnnotations(List<Annotation> annotations, CancellationToken token = default)
+    public async Task ValidateAnnotations(List<string> annotationNames, bool fromQueue = false, CancellationToken token = default)
    {
        if (!_api.CurrentUser.Role.IsValidator())
            return;

-        var annNames = annotations.Select(x => x.Name).ToHashSet();
+        var annNames = annotationNames.ToHashSet();
        await _dbFactory.Run(async db =>
        {
            await db.Annotations
@@ -211,6 +243,9 @@ public class AnnotationService : IAnnotationService, INotificationHandler<Annota
                .Set(x => x.ValidateEmail, _api.CurrentUser.Email)
                .UpdateAsync(token: token);
        });
+        if (!fromQueue)
+            await _producer.SendToInnerQueue(annotationNames, AnnotationStatus.Validated, token);
+
        ThrottleExt.Throttle(async () =>
        {
            _dbFactory.SaveToDisk();
@@ -218,15 +253,25 @@ public class AnnotationService : IAnnotationService, INotificationHandler<Annota
        }, SaveTaskId, TimeSpan.FromSeconds(5), true);
    }

-    public async Task Handle(AnnotationsDeletedEvent notification, CancellationToken cancellationToken)
+    public async Task Handle(AnnotationsDeletedEvent notification, CancellationToken ct)
    {
-        await _dbFactory.DeleteAnnotations(notification.Annotations, cancellationToken);
-        foreach (var annotation in notification.Annotations)
+        await _dbFactory.DeleteAnnotations(notification.AnnotationNames, ct);
+        foreach (var name in notification.AnnotationNames)
        {
-            File.Delete(annotation.ImagePath);
-            File.Delete(annotation.LabelPath);
-            File.Delete(annotation.ThumbPath);
+            File.Delete(Path.Combine(_dirConfig.ImagesDirectory, $"{name}{Constants.JPG_EXT}"));
+            File.Delete(Path.Combine(_dirConfig.LabelsDirectory, $"{name}{Constants.TXT_EXT}"));
+            File.Delete(Path.Combine(_dirConfig.ThumbnailsDirectory, $"{name}{Constants.THUMBNAIL_PREFIX}{Constants.JPG_EXT}"));
+            File.Delete(Path.Combine(_dirConfig.ResultsDirectory, $"{name}{Constants.RESULT_PREFIX}{Constants.JPG_EXT}"));
        }
+
+        if (!notification.FromQueue)
+            await _producer.SendToInnerQueue(notification.AnnotationNames, AnnotationStatus.Deleted, ct);
+
+        ThrottleExt.Throttle(async () =>
+        {
+            _dbFactory.SaveToDisk();
+            await Task.CompletedTask;
+        }, SaveTaskId, TimeSpan.FromSeconds(5), true);
    }
 }

@@ -234,6 +279,5 @@ public interface IAnnotationService
 {
    Task<Annotation> SaveAnnotation(AnnotationImage a, CancellationToken ct = default);
    Task<Annotation> SaveAnnotation(string originalMediaName, TimeSpan time, List<Detection> detections, Stream? stream = null, CancellationToken token = default);
-    Task ValidateAnnotations(List<Annotation> annotations, CancellationToken token = default);
-
+    Task ValidateAnnotations(List<string> annotationNames, bool fromQueue = false, CancellationToken token = default);
 }
@@ -3,12 +3,16 @@ using System.Net;
 using Azaion.Common.Database;
 using Azaion.Common.DTO.Config;
 using Azaion.Common.DTO.Queue;
+using Azaion.Common.Extensions;
+using Azaion.CommonSecurity.DTO;
+using Azaion.CommonSecurity.Services;
 using LinqToDB;
 using MessagePack;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
 using Newtonsoft.Json;
 using RabbitMQ.Stream.Client;
+using RabbitMQ.Stream.Client.AMQP;
 using RabbitMQ.Stream.Client.Reliable;

 namespace Azaion.Common.Services;
@@ -17,16 +21,15 @@ public class FailsafeAnnotationsProducer
 {
    private readonly ILogger<FailsafeAnnotationsProducer> _logger;
    private readonly IDbFactory _dbFactory;
+    private readonly IAzaionApi _azaionApi;
    private readonly QueueConfig _queueConfig;
-
    private Producer _annotationProducer = null!;
-    private Producer _annotationConfirmProducer = null!;

-
-    public FailsafeAnnotationsProducer(ILogger<FailsafeAnnotationsProducer> logger, IDbFactory dbFactory, IOptions<QueueConfig> queueConfig)
+    public FailsafeAnnotationsProducer(ILogger<FailsafeAnnotationsProducer> logger, IDbFactory dbFactory, IOptions<QueueConfig> queueConfig, IAzaionApi azaionApi)
    {
        _logger = logger;
        _dbFactory = dbFactory;
+        _azaionApi = azaionApi;
        _queueConfig = queueConfig.Value;
        Task.Run(async () => await ProcessQueue());
    }
@@ -41,107 +44,104 @@ public class FailsafeAnnotationsProducer
        });
    }

-    private async Task Init(CancellationToken cancellationToken = default)
+    private async Task ProcessQueue(CancellationToken ct = default)
    {
        _annotationProducer = await Producer.Create(new ProducerConfig(await GetProducerQueueConfig(), Constants.MQ_ANNOTATIONS_QUEUE));
-        _annotationConfirmProducer = await Producer.Create(new ProducerConfig(await GetProducerQueueConfig(), Constants.MQ_ANNOTATIONS_CONFIRM_QUEUE));
-    }
-
-    private async Task ProcessQueue(CancellationToken cancellationToken = default)
-    {
-        await Init(cancellationToken);
-        while (!cancellationToken.IsCancellationRequested)
+        while (!ct.IsCancellationRequested)
        {
-            var messages = await GetFromInnerQueue(cancellationToken);
-            foreach (var messagesChunk in messages.Chunk(10)) //Sending by 10
-            {
-                var sent = false;
-                while (!sent || cancellationToken.IsCancellationRequested) //Waiting for send
-                {
-                    try
-                    {
-                        var createdMessages = messagesChunk
-                            .Where(x => x.Status == AnnotationStatus.Created)
-                            .Select(x => new Message(MessagePackSerializer.Serialize(x)))
-                            .ToList();
-                        if (createdMessages.Any())
-                            await _annotationProducer.Send(createdMessages, CompressionType.Gzip);
-
-                        var validatedMessages = messagesChunk
-                            .Where(x => x.Status == AnnotationStatus.Validated)
-                            .Select(x => new Message(MessagePackSerializer.Serialize(x)))
-                            .ToList();
-                        if (validatedMessages.Any())
-                            await _annotationConfirmProducer.Send(validatedMessages, CompressionType.Gzip);
-
-                        await _dbFactory.Run(async db =>
-                            await db.AnnotationsQueue.DeleteAsync(aq => messagesChunk.Any(x => aq.Name == x.Name), token: cancellationToken));
-                        sent = true;
-                        _dbFactory.SaveToDisk();
-                    }
-                    catch (Exception e)
-                    {
-                        _logger.LogError(e, e.Message);
-                        await Task.Delay(TimeSpan.FromSeconds(10), cancellationToken);
-                    }
-                    await Task.Delay(TimeSpan.FromSeconds(10), cancellationToken);
-                }
-            }
-            await Task.Delay(TimeSpan.FromSeconds(5), cancellationToken);
-        }
-    }
-
-    private async Task<List<AnnotationCreatedMessage>> GetFromInnerQueue(CancellationToken cancellationToken = default)
-    {
-        return await _dbFactory.Run(async db =>
-        {
-            var annotations = await db.AnnotationsQueue.Join(
-                    db.Annotations.LoadWith(x => x.Detections), aq => aq.Name, a => a.Name, (aq, a) => a)
-                .ToListAsync(token: cancellationToken);
-
-            var messages = new List<AnnotationCreatedMessage>();
-            var badImages = new List<string>();
-            foreach (var annotation in annotations)
+            var sent = false;
+            while (!sent || !ct.IsCancellationRequested) //Waiting for send
            {
                try
                {
-                    var image = await File.ReadAllBytesAsync(annotation.ImagePath, cancellationToken);
-                    var annCreateMessage = new AnnotationCreatedMessage
+                    var result = await _dbFactory.Run(async db =>
                    {
-                        Name = annotation.Name,
-                        OriginalMediaName = annotation.OriginalMediaName,
-                        Time = annotation.Time,
-                        CreatedRole = annotation.CreatedRole,
-                        CreatedEmail = annotation.CreatedEmail,
-                        CreatedDate = annotation.CreatedDate,
-                        Status = annotation.AnnotationStatus,
+                        var records = await db.AnnotationsQueueRecords.OrderBy(x => x.DateTime).ToListAsync(token: ct);
+                        var editedCreatedNames = records
+                            .Where(x => x.Operation.In(AnnotationStatus.Created, AnnotationStatus.ValidatedEdited))
+                            .Select(x => x.AnnotationNames.FirstOrDefault())
+                            .ToList();

-                        ImageExtension = annotation.ImageExtension,
-                        Image = image,
-                        Detections = JsonConvert.SerializeObject(annotation.Detections),
-                        Source = annotation.Source,
-                    };
-                    messages.Add(annCreateMessage);
+                        var annotationsDict = await db.Annotations.LoadWith(x => x.Detections)
+                            .Where(x => editedCreatedNames.Contains(x.Name))
+                            .ToDictionaryAsync(a => a.Name, token: ct);
+
+                        var messages = new List<Message>();
+                        foreach (var record in records)
+                        {
+                            var appProperties = new ApplicationProperties
+                            {
+                                { nameof(AnnotationStatus), record.Operation },
+                                { nameof(User.Email), _azaionApi.CurrentUser.Email }
+                            };
+
+                            if (record.Operation.In(AnnotationStatus.Validated, AnnotationStatus.Deleted))
+                            {
+                                var message = new Message(MessagePackSerializer.Serialize(new AnnotationBulkMessage
+                                {
+                                    AnnotationNames = record.AnnotationNames.ToArray()
+                                })) { ApplicationProperties = appProperties };
+
+                                messages.Add(message);
+                            }
+                            else
+                            {
+                                var annotation = annotationsDict.GetValueOrDefault(record.AnnotationNames.FirstOrDefault());
+                                if (annotation == null)
+                                    continue;
+
+                                var image = await File.ReadAllBytesAsync(annotation.ImagePath, ct);
+                                var annMessage = new AnnotationMessage
+                                {
+                                    Name = annotation.Name,
+                                    OriginalMediaName = annotation.OriginalMediaName,
+                                    Time = annotation.Time,
+                                    Role = annotation.CreatedRole,
+                                    Email = annotation.CreatedEmail,
+                                    CreatedDate = annotation.CreatedDate,
+                                    Status = annotation.AnnotationStatus,
+
+                                    ImageExtension = annotation.ImageExtension,
+                                    Image = image,
+                                    Detections = JsonConvert.SerializeObject(annotation.Detections),
+                                    Source = annotation.Source,
+                                };
+                                var message = new Message(MessagePackSerializer.Serialize(annMessage)) { ApplicationProperties = appProperties };
+
+                                messages.Add(message);
+                            }
+                        }
+
+                        return (messages, records);
+                    });
+
+                    if (result.messages.Any())
+                    {
+                        await _annotationProducer.Send(result.messages, CompressionType.Gzip);
+                        await _dbFactory.Run(async db => await db.DeleteAsync(result.records, token: ct));
+                        sent = true;
+                        _dbFactory.SaveToDisk();
+                    }
                }
                catch (Exception e)
                {
                    _logger.LogError(e, e.Message);
-                    badImages.Add(annotation.Name);
+                    await Task.Delay(TimeSpan.FromSeconds(10), ct);
                }
+                await Task.Delay(TimeSpan.FromSeconds(10), ct);
            }
-
-            if (badImages.Any())
-            {
-                await db.AnnotationsQueue.Where(x => badImages.Contains(x.Name)).DeleteAsync(token: cancellationToken);
-                _dbFactory.SaveToDisk();
-            }
-            return messages;
-        });
+        }
+        await Task.Delay(TimeSpan.FromSeconds(5), ct);
    }

-    public async Task SendToInnerQueue(Annotation annotation, CancellationToken cancellationToken = default)
+    public async Task SendToInnerQueue(List<string> annotationNames, AnnotationStatus status, CancellationToken cancellationToken = default)
    {
        await _dbFactory.Run(async db =>
-            await db.InsertAsync(new AnnotationName { Name = annotation.Name }, token: cancellationToken));
+            await db.InsertAsync(new AnnotationQueueRecord
+            {
+                DateTime = DateTime.UtcNow,
+                Operation = status,
+                AnnotationNames = annotationNames
+            }, token: cancellationToken));
    }
 }
@@ -60,7 +60,7 @@ public class GpsMatcherClient : IGpsMatcherClient

            process.OutputDataReceived += (_, e) => { if (e.Data != null) Console.WriteLine(e.Data); };
            process.ErrorDataReceived += (_, e) => { if (e.Data != null) Console.WriteLine(e.Data); };
-            process.Start();
+            //process.Start();
        }
        catch (Exception e)
        {