add offset

fixes
add visual validation border and validate functionality

Azaion.Common/Services/AnnotationService.cs

@@ -5,6 +5,7 @@ using Azaion.Common.Database;
 using Azaion.Common.DTO;
 using Azaion.Common.DTO.Config;
 using Azaion.Common.DTO.Queue;
+using Azaion.Common.Extensions;
 using Azaion.CommonSecurity.DTO;
 using Azaion.CommonSecurity.Services;
 using LinqToDB;
@@ -25,6 +26,7 @@ public class AnnotationService
     private readonly FailsafeAnnotationsProducer _producer;
     private readonly IGalleryService _galleryService;
     private readonly IMediator _mediator;
+    private readonly IHardwareService _hardwareService;
     private readonly QueueConfig _queueConfig;
     private Consumer _consumer = null!;
 
@@ -33,19 +35,21 @@ public class AnnotationService
         FailsafeAnnotationsProducer producer,
         IOptions<QueueConfig> queueConfig,
         IGalleryService galleryService,
-        IMediator mediator)
+        IMediator mediator,
+        IHardwareService hardwareService)
     {
         _apiClient = apiClient;
         _dbFactory = dbFactory;
         _producer = producer;
         _galleryService = galleryService;
         _mediator = mediator;
+        _hardwareService = hardwareService;
         _queueConfig = queueConfig.Value;
 
         Task.Run(async () => await Init()).Wait();
     }
 
-    private async Task Init()
+    private async Task Init(CancellationToken cancellationToken = default)
     {
         var consumerSystem = await StreamSystem.Create(new StreamSystemConfig
         {
@@ -53,11 +57,28 @@ public class AnnotationService
             UserName = _queueConfig.ConsumerUsername,
             Password = _queueConfig.ConsumerPassword
         });
+        var offset = (await _dbFactory.Run(db => db.QueueOffsets.FirstOrDefaultAsync(
+            x => x.QueueName == Constants.MQ_ANNOTATIONS_QUEUE, token: cancellationToken))
+            )?.Offset ?? 0;
         _consumer = await Consumer.Create(new ConsumerConfig(consumerSystem, Constants.MQ_ANNOTATIONS_QUEUE)
         {
-            OffsetSpec = new OffsetTypeFirst(),
-            MessageHandler = async (stream, _, _, message) =>
-                await Consume(MessagePackSerializer.Deserialize<AnnotationCreatedMessage>(message.Data.Contents)),
+            Reference = _hardwareService.GetHardware().Hash,
+            OffsetSpec = new OffsetTypeOffset(offset + 1),
+            MessageHandler = async (stream, consumer, context, message) =>
+            {
+                await Consume(MessagePackSerializer.Deserialize<AnnotationCreatedMessage>(message.Data.Contents), cancellationToken);
+                await _dbFactory.Run(async db => await db.QueueOffsets
+                    .Where(x => x.QueueName == Constants.MQ_ANNOTATIONS_QUEUE)
+                    .Set(x => x.Offset, context.Offset)
+                    .UpdateAsync(token: cancellationToken));
+
+                await ThrottleExt.ThrottleRunAfter(() =>
+                {
+                    _dbFactory.SaveToDisk();
+                    return Task.CompletedTask;
+                }, TimeSpan.FromSeconds(3), cancellationToken);
+
+            }
         });
     }
 
@@ -65,6 +86,10 @@ public class AnnotationService
     public async Task SaveAnnotation(string fName, string imageExtension, List<Detection> detections, SourceEnum source, Stream? stream = null, CancellationToken token = default) =>
         await SaveAnnotationInner(DateTime.UtcNow, fName, imageExtension, detections, source, stream, _apiClient.User.Role, _apiClient.User.Email, token);
 
+    //Manual
+    public async Task ValidateAnnotation(Annotation annotation, CancellationToken token = default) =>
+        await SaveAnnotationInner(DateTime.UtcNow, annotation.Name, annotation.ImageExtension, annotation.Detections.ToList(), SourceEnum.Manual, null, _apiClient.User.Role, _apiClient.User.Email, token);
+
     //Queue (only from operators)
     public async Task Consume(AnnotationCreatedMessage message, CancellationToken cancellationToken = default)
     {
@@ -84,24 +109,20 @@ public class AnnotationService
     }
 
     private async Task SaveAnnotationInner(DateTime createdDate, string fName, string imageExtension, List<Detection> detections, SourceEnum source, Stream? stream,
-        RoleEnum createdRole,
+        RoleEnum userRole,
         string createdEmail,
         CancellationToken token = default)
     {
         //Flow for roles:
-        // Operator:
-        //   sourceEnum: (manual, ai) <AnnotationCreatedMessage>
-        // Validator:
-        //   sourceEnum: (manual)     if was in received.json then <AnnotationValidatedMessage> else <AnnotationCreatedMessage>
-        //   sourceEnum: (queue, AI)  if queue CreatedMessage with the same user - do nothing Add to received.json
+        // Operator or (AI from any role) -> Created
+        // Validator, Admin & Manual -> Validated
 
-        var classes = detections.Select(x => x.ClassNumber).Distinct().ToList() ?? [];
         AnnotationStatus status;
 
         var annotation = await _dbFactory.Run(async db =>
         {
             var ann = await db.Annotations.FirstOrDefaultAsync(x => x.Name == fName, token: token);
-            status = ann?.AnnotationStatus == AnnotationStatus.Created && createdRole == RoleEnum.Validator
+            status = userRole.IsValidator() && source == SourceEnum.Manual
                 ? AnnotationStatus.Validated
                 : AnnotationStatus.Created;
 
@@ -110,7 +131,6 @@ public class AnnotationService
             if (ann != null)
                 await db.Annotations
                     .Where(x => x.Name == fName)
-                    .Set(x => x.Classes, classes)
                     .Set(x => x.Source, source)
                     .Set(x => x.AnnotationStatus, status)
                     .UpdateAsync(token: token);
@@ -122,7 +142,7 @@ public class AnnotationService
                     Name = fName,
                     ImageExtension = imageExtension,
                     CreatedEmail = createdEmail,
-                    CreatedRole = createdRole,
+                    CreatedRole = userRole,
                     AnnotationStatus = status,
                     Source = source,
                     Detections = detections
@@ -142,5 +162,10 @@ public class AnnotationService
         await _producer.SendToQueue(annotation, token);
 
         await _mediator.Publish(new AnnotationCreatedEvent(annotation), token);
+        await ThrottleExt.ThrottleRunAfter(() =>
+        {
+            _dbFactory.SaveToDisk();
+            return Task.CompletedTask;
+        }, TimeSpan.FromSeconds(5), token);
     }
 }

Azaion.Common/Services/GalleryService.cs

@@ -150,11 +150,6 @@ public class GalleryService(
             };
             await dbFactory.Run(async db =>
             {
-                var xx = missedAnnotations.GroupBy(x => x.Name)
-                    .Where(gr => gr.Count() > 1)
-                    .ToList();
-                foreach (var gr in xx)
-                    Console.WriteLine(gr.Key);
                 await db.BulkCopyAsync(copyOptions, missedAnnotations);
                 await db.BulkCopyAsync(copyOptions, missedAnnotations.SelectMany(x => x.Detections));
             });