add MediaHash. Step1

Azaion.Common/Services/AnnotationService.cs

@@ -62,7 +62,8 @@ public class AnnotationService : IAnnotationService
 
     private async Task InitQueueConsumer(CancellationToken token = default)
     {
-        if (!_api.CurrentUser.Role.IsValidator())
+        var currentUser = await _api.GetCurrentUserAsync();
+        if (!currentUser.Role.IsValidator())
             return;
 
         var consumerSystem = await StreamSystem.Create(new StreamSystemConfig
@@ -72,11 +73,11 @@ public class AnnotationService : IAnnotationService
             Password = _queueConfig.ConsumerPassword
         });
 
-        var offsets = _api.CurrentUser.UserConfig?.QueueOffsets ?? new UserQueueOffsets();
+        var offsets = currentUser.UserConfig?.QueueOffsets ?? new UserQueueOffsets();
 
         _consumer = await Consumer.Create(new ConsumerConfig(consumerSystem, Constants.MQ_ANNOTATIONS_QUEUE)
         {
-            Reference = _api.CurrentUser.Email,
+            Reference = currentUser.Email,
             OffsetSpec = new OffsetTypeOffset(offsets.AnnotationsOffset),
             MessageHandler = async (_, _, context, message) =>
             {
@@ -87,13 +88,14 @@ public class AnnotationService : IAnnotationService
                     if (!Enum.TryParse<AnnotationStatus>((string)message.ApplicationProperties[nameof(AnnotationStatus)], out var annotationStatus))
                         return;
 
-                    if (email != _api.CurrentUser.Email) //Don't process messages by yourself
+                    if (email != currentUser.Email)
                     {
                         if (annotationStatus.In(AnnotationStatus.Created, AnnotationStatus.Edited))
                         {
                             var msg = MessagePackSerializer.Deserialize<AnnotationMessage>(message.Data.Contents);
                             await SaveAnnotationInner(
                                 msg.CreatedDate,
+                                msg.MediaHash,
                                 msg.OriginalMediaName,
                                 msg.Name,
                                 msg.Time,
@@ -115,12 +117,9 @@ public class AnnotationService : IAnnotationService
                         }
                     }
                     
-                    offsets.AnnotationsOffset = context.Offset + 1; //to consume on the next launch from the next message
-                    ThrottleExt.Throttle(() =>
-                    {
-                        _api.UpdateOffsets(offsets);
-                        return Task.CompletedTask;
-                    }, SaveQueueOffsetTaskId, TimeSpan.FromSeconds(10), scheduleCallAfterCooldown: true);
+                    offsets.AnnotationsOffset = context.Offset + 1;
+                    ThrottleExt.Throttle(async () => await _api.UpdateOffsetsAsync(offsets), 
+                        SaveQueueOffsetTaskId, TimeSpan.FromSeconds(10), scheduleCallAfterCooldown: true);
                 }
                 catch (Exception e)
                 {
@@ -137,17 +136,21 @@ public class AnnotationService : IAnnotationService
     //AI
     public async Task<Annotation> SaveAnnotation(AnnotationImage a, CancellationToken ct = default)
     {
+        var currentUser = await _api.GetCurrentUserAsync();
         a.Time = TimeSpan.FromMilliseconds(a.Milliseconds);
-        return await SaveAnnotationInner(DateTime.UtcNow, a.OriginalMediaName, a.Name, a.Time, a.Detections.ToList(),
-            SourceEnum.AI, new MemoryStream(a.Image), _api.CurrentUser.Role, _api.CurrentUser.Email, token: ct);
+        return await SaveAnnotationInner(DateTime.UtcNow, a.MediaHash, a.OriginalMediaName, a.Name, a.Time, a.Detections.ToList(),
+            SourceEnum.AI, new MemoryStream(a.Image), currentUser.Role, currentUser.Email, token: ct);
     }
 
     //Manual
-    public async Task<Annotation> SaveAnnotation(string originalMediaName, string annotationName, TimeSpan time, List<Detection> detections, Stream? stream = null, CancellationToken token = default) =>
-        await SaveAnnotationInner(DateTime.UtcNow, originalMediaName, annotationName, time, detections, SourceEnum.Manual, stream,
-            _api.CurrentUser.Role, _api.CurrentUser.Email, token: token);
+    public async Task<Annotation> SaveAnnotation(string mediaHash, string originalMediaName, string annotationName, TimeSpan time, List<Detection> detections,
+        Stream? stream = null, CancellationToken token = default)
+    {
+        var currentUser = await _api.GetCurrentUserAsync();
+        return await SaveAnnotationInner(DateTime.UtcNow, mediaHash, originalMediaName, annotationName, time, detections, SourceEnum.Manual, stream, currentUser.Role, currentUser.Email, token: token);
+    }
 
-    private async Task<Annotation> SaveAnnotationInner(DateTime createdDate, string originalMediaName, string annotationName, TimeSpan time,
+    private async Task<Annotation> SaveAnnotationInner(DateTime createdDate, string mediaHash, string originalMediaName, string annotationName, TimeSpan time,
         List<Detection> detections, SourceEnum source, Stream? stream,
         RoleEnum userRole,
         string createdEmail,
@@ -163,13 +166,17 @@ public class AnnotationService : IAnnotationService
 
             await db.Detections.DeleteAsync(x => x.AnnotationName == annotationName, token: token);
 
-            if (ann != null) //Annotation is already exists
+            if (ann != null)
             {
                 status = AnnotationStatus.Edited;
 
                 var annotationUpdatable = db.Annotations
                     .Where(x => x.Name == annotationName)
-                    .Set(x => x.Source, source);
+                    .Set(x => x.Source, source)
+                    .Set(x => x.OriginalMediaName, originalMediaName);
+
+                if (!string.IsNullOrEmpty(mediaHash))
+                    annotationUpdatable = annotationUpdatable.Set(x => x.MediaHash, mediaHash);
 
                 if (userRole.IsValidator() && source == SourceEnum.Manual)
                 {
@@ -183,6 +190,7 @@ public class AnnotationService : IAnnotationService
                     .UpdateAsync(token: token);
 
                 ann.Detections = detections;
+                ann.MediaHash = mediaHash;
             }
             else
             {
@@ -190,6 +198,7 @@ public class AnnotationService : IAnnotationService
                 {
                     CreatedDate = createdDate,
                     Name = annotationName,
+                    MediaHash = mediaHash,
                     OriginalMediaName = originalMediaName,
                     Time = time,
                     ImageExtension = Constants.JPG_EXT,
@@ -244,7 +253,8 @@ public class AnnotationService : IAnnotationService
 
     public async Task ValidateAnnotations(List<string> annotationNames, bool fromQueue = false, CancellationToken token = default)
     {
-        if (!_api.CurrentUser.Role.IsValidator())
+        var currentUser = await _api.GetCurrentUserAsync();
+        if (!currentUser.Role.IsValidator())
             return;
 
         var annNames = annotationNames.ToHashSet();
@@ -254,7 +264,7 @@ public class AnnotationService : IAnnotationService
                 .Where(x => annNames.Contains(x.Name))
                 .Set(x => x.AnnotationStatus, AnnotationStatus.Validated)
                 .Set(x => x.ValidateDate, DateTime.UtcNow)
-                .Set(x => x.ValidateEmail, _api.CurrentUser.Email)
+                .Set(x => x.ValidateEmail, currentUser.Email)
                 .UpdateAsync(token: token);
         });
         if (!fromQueue)
@@ -265,6 +275,6 @@ public class AnnotationService : IAnnotationService
 public interface IAnnotationService
 {
     Task<Annotation> SaveAnnotation(AnnotationImage a, CancellationToken ct = default);
-    Task<Annotation> SaveAnnotation(string originalMediaName, string annotationName, TimeSpan time, List<Detection> detections, Stream? stream = null, CancellationToken token = default);
+    Task<Annotation> SaveAnnotation(string mediaHash, string originalMediaName, string annotationName, TimeSpan time, List<Detection> detections, Stream? stream = null, CancellationToken token = default);
     Task ValidateAnnotations(List<string> annotationNames, bool fromQueue = false, CancellationToken token = default);
 }

Azaion.Common/Services/AuthProvider.cs

@@ -1,8 +1,9 @@
-using System.IO;
-using System.Net;
+using System.Net;
 using System.Net.Http;
 using System.Net.Http.Headers;
 using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
 using Azaion.Common.DTO;
 
 using Newtonsoft.Json;
@@ -13,115 +14,103 @@ namespace Azaion.Common.Services;
 public interface IAzaionApi
 {
     ApiCredentials Credentials { get; }
-    User CurrentUser { get; }
-    void UpdateOffsets(UserQueueOffsets offsets);
-    //Stream GetResource(string filename, string folder);
+    Task<User> GetCurrentUserAsync();
+    Task UpdateOffsetsAsync(UserQueueOffsets offsets);
 }
 
 public class AzaionApi(ILogger logger, HttpClient client, ICache cache, ApiCredentials credentials) : IAzaionApi
 {
-    private string _jwtToken = null!;
+    private readonly SemaphoreSlim _authLock = new(1, 1);
+    private string? _jwtToken;
     const string APP_JSON = "application/json";
+    
     public ApiCredentials Credentials => credentials;
 
-    public User CurrentUser
+    public async Task<User> GetCurrentUserAsync()
     {
-        get
-        {
-            var user = cache.GetFromCache(Constants.CURRENT_USER_CACHE_KEY,
-                () => Get<User>("users/current"));
-            if (user == null)
-                throw new Exception("Can't get current user");
-            return user;
-        }
+        var user = await cache.GetFromCacheAsync(Constants.CURRENT_USER_CACHE_KEY,
+            async () => await GetAsync<User>("users/current"));
+        return user ?? throw new Exception("Can't get current user");
     }
 
-    public void UpdateOffsets(UserQueueOffsets offsets)
+    public async Task UpdateOffsetsAsync(UserQueueOffsets offsets)
     {
-        Put($"/users/queue-offsets/set", new
-        {
-            Email = CurrentUser.Email,
-            Offsets = offsets
-        });
+        var user = await GetCurrentUserAsync();
+        await PutAsync("/users/queue-offsets/set", new { Email = user.Email, Offsets = offsets });
     }
 
-    private HttpResponseMessage Send(HttpRequestMessage request)
+    private async Task<T?> GetAsync<T>(string url)
+    {
+        var response = await SendWithAuthAsync(() => client.GetAsync(url));
+        return JsonConvert.DeserializeObject<T>(await response.Content.ReadAsStringAsync());
+    }
+
+    private async Task PutAsync<T>(string url, T obj)
+    {
+        var content = new StringContent(JsonConvert.SerializeObject(obj), Encoding.UTF8, APP_JSON);
+        await SendWithAuthAsync(() => client.PutAsync(url, content));
+    }
+
+    private async Task<HttpResponseMessage> SendWithAuthAsync(Func<Task<HttpResponseMessage>> sendAction)
     {
         if (string.IsNullOrEmpty(_jwtToken))
-            Authorize();
-        request.Headers.Authorization = new AuthenticationHeaderValue("Bearer", _jwtToken);
-        var response = client.Send(request);
+            await AuthorizeAsync();
+
+        client.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Bearer", _jwtToken);
+        var response = await sendAction();
 
         if (response.StatusCode == HttpStatusCode.Unauthorized)
         {
-            Authorize();
-            request.Headers.Authorization = new AuthenticationHeaderValue("Bearer", _jwtToken);
-            response = client.Send(request);
+            await AuthorizeAsync();
+            client.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Bearer", _jwtToken);
+            response = await sendAction();
         }
 
-        if (response.IsSuccessStatusCode)
-            return response;
-
-        var stream = response.Content.ReadAsStream();
-        var content = new StreamReader(stream).ReadToEnd();
-        if (response.StatusCode == HttpStatusCode.Conflict)
+        if (!response.IsSuccessStatusCode)
         {
-            var result = JsonConvert.DeserializeObject<BusinessExceptionDto>(content);
-            throw new Exception($"Failed: {response.StatusCode}! Error Code: {result?.ErrorCode}. Message: {result?.Message}");
+            var content = await response.Content.ReadAsStringAsync();
+            if (response.StatusCode == HttpStatusCode.Conflict)
+            {
+                var error = JsonConvert.DeserializeObject<BusinessExceptionDto>(content);
+                throw new Exception($"Failed: {response.StatusCode}! Error Code: {error?.ErrorCode}. Message: {error?.Message}");
+            }
+            throw new Exception($"Failed: {response.StatusCode}! Result: {content}");
         }
-        throw new Exception($"Failed: {response.StatusCode}! Result: {content}");
+
+        return response;
     }
 
-    private T? Get<T>(string url)
-    {
-        var response = Send(new HttpRequestMessage(HttpMethod.Get, url));
-        var stream = response.Content.ReadAsStream();
-        var json = new StreamReader(stream).ReadToEnd();
-        return JsonConvert.DeserializeObject<T>(json);
-    }
-
-    private void Put<T>(string url, T obj)
-    {
-        Send(new HttpRequestMessage(HttpMethod.Put, url)
-        {
-            Content = new StringContent(JsonConvert.SerializeObject(obj), Encoding.UTF8, APP_JSON)
-        });
-    }
-
-    private void Authorize()
+    private async Task AuthorizeAsync()
     {
+        await _authLock.WaitAsync();
         try
         {
             if (string.IsNullOrEmpty(credentials.Email) || credentials.Password.Length == 0)
-                throw new Exception("Email or password is empty! Please do EnterCredentials first!");
+                throw new Exception("Email or password is empty!");
 
-            var payload = new
-            {
-                email = credentials.Email,
-                password = credentials.Password
-            };
-            var content = new StringContent(JsonConvert.SerializeObject(payload), Encoding.UTF8, APP_JSON);
-            var message = new HttpRequestMessage(HttpMethod.Post, "login") { Content = content };
-            var response = client.Send(message);
+            var content = new StringContent(
+                JsonConvert.SerializeObject(new { email = credentials.Email, password = credentials.Password }), 
+                Encoding.UTF8, APP_JSON);
+            
+            client.DefaultRequestHeaders.Authorization = null;
+            var response = await client.PostAsync("login", content);
 
             if (!response.IsSuccessStatusCode)
-                throw new Exception($"EnterCredentials failed: {response.StatusCode}");
+                throw new Exception($"Authorization failed: {response.StatusCode}");
 
-            var stream = response.Content.ReadAsStream();
-            var json = new StreamReader(stream).ReadToEnd();
-            var result = JsonConvert.DeserializeObject<LoginResponse>(json);
+            var result = JsonConvert.DeserializeObject<LoginResponse>(await response.Content.ReadAsStringAsync())
+                ?? throw new Exception("JWT Token not found in response");
 
-            if (string.IsNullOrEmpty(result?.Token))
-                throw new Exception("JWT Token not found in response");
-
-            _jwtToken = result.Token;
+            _jwtToken = result.Token ?? throw new Exception("JWT Token not found in response");
         }
         catch (Exception e)
         {
             logger.Error(e, e.Message);
             throw;
         }
+        finally
+        {
+            _authLock.Release();
+        }
     }
-
-
 }

Azaion.Common/Services/Cache.cs

@@ -5,6 +5,7 @@ namespace Azaion.Common.Services;
 public interface ICache
 {
     T GetFromCache<T>(string key, Func<T> fetchFunc, TimeSpan? expiration = null);
+    Task<T> GetFromCacheAsync<T>(string key, Func<Task<T>> fetchFunc, TimeSpan? expiration = null);
     void Invalidate(string key);
 }
 
@@ -23,5 +24,16 @@ public class MemoryCache : ICache
         });
     }
 
+    public async Task<T> GetFromCacheAsync<T>(string key, Func<Task<T>> fetchFunc, TimeSpan? expiration = null)
+    {
+        expiration ??= TimeSpan.FromHours(4);
+        return await _cache.GetOrAddAsync(key, async entry =>
+        {
+            var result = await fetchFunc();
+            entry.AbsoluteExpirationRelativeToNow = expiration;
+            return result;
+        });
+    }
+
     public void Invalidate(string key) => _cache.Remove(key);
 }

Azaion.Common/Services/FailsafeProducer.cs

@@ -54,6 +54,7 @@ public class FailsafeAnnotationsProducer
     private async Task ProcessQueue(CancellationToken ct = default)
     {
         _annotationProducer = await Producer.Create(new ProducerConfig(await GetProducerQueueConfig(), Constants.MQ_ANNOTATIONS_QUEUE));
+        var currentUser = await _azaionApi.GetCurrentUserAsync();
         while (!ct.IsCancellationRequested)
         {
             var sent = false;
@@ -81,7 +82,7 @@ public class FailsafeAnnotationsProducer
                         var appProperties = new ApplicationProperties
                         {
                             { nameof(AnnotationStatus), record.Operation.ToString() },
-                            { nameof(User.Email), _azaionApi.CurrentUser.Email }
+                            { nameof(User.Email), currentUser.Email }
                         };
 
                         if (record.Operation.In(AnnotationStatus.Validated, AnnotationStatus.Deleted))
@@ -90,7 +91,7 @@ public class FailsafeAnnotationsProducer
                             {
                                 AnnotationNames = record.AnnotationNames.ToArray(),
                                 AnnotationStatus = record.Operation,
-                                Email = _azaionApi.CurrentUser.Email,
+                                Email = currentUser.Email,
                                 CreatedDate = record.DateTime
                             })) { ApplicationProperties = appProperties };
 

Azaion.Common/Services/HashExtensions.cs

@@ -0,0 +1,63 @@
+using System.IO;
+using System.IO.Hashing;
+using System.Text;
+
+namespace Azaion.Common.Services;
+
+public static class HashExtensions
+{
+    private const int SampleSize = 1024;
+    private const int HashLength = 9;
+    private const string VirtualHashPrefix = "V";
+
+    public static string CalcHash(this FileInfo file)
+    {
+        if (!file.Exists)
+            return CalcVirtualHash(file.Name);
+
+        var fileSize = file.Length;
+        if (fileSize < SampleSize * 8)
+            return CalcHashInner(File.ReadAllBytes(file.FullName));
+        
+        var combinedData = new byte[8 + SampleSize * 3];
+        var offset = 0;
+
+        var fileSizeBytes = BitConverter.GetBytes(fileSize);
+        Array.Copy(fileSizeBytes, 0, combinedData, offset, 8);
+        offset += 8;
+
+        using var stream = new FileStream(file.FullName, FileMode.Open, FileAccess.Read, FileShare.Read);
+
+        stream.ReadExactly(combinedData, offset, SampleSize);
+        offset += SampleSize;
+
+        stream.Position = (fileSize - SampleSize) / 2;
+        stream.ReadExactly(combinedData, offset, SampleSize);
+        offset += SampleSize;
+
+        stream.Position = fileSize - SampleSize;
+        stream.ReadExactly(combinedData, offset, SampleSize);
+
+        return CalcHashInner(combinedData);
+    }
+
+    public static string CalcVirtualHash(string filename)
+    {
+        var bytes = Encoding.UTF8.GetBytes($"VIRTUAL:{filename}");
+        var hash = XxHash64.HashToUInt64(bytes);
+        var hashBytes = BitConverter.GetBytes(hash);
+        var base64 = Convert.ToBase64String(hashBytes);
+        return VirtualHashPrefix + base64[..HashLength];
+    }
+
+    public static bool IsVirtualHash(this string? hash) => 
+        !string.IsNullOrEmpty(hash) && hash.StartsWith(VirtualHashPrefix);
+
+    private static string CalcHashInner(byte[] data)
+    {
+        var hash = XxHash64.HashToUInt64(data);
+        var hashBytes = BitConverter.GetBytes(hash);
+        var base64 = Convert.ToBase64String(hashBytes);
+        return base64[..HashLength];
+    }
+}

Azaion.Common/Services/Inference/InferenceClient.cs

@@ -77,6 +77,7 @@ public class InferenceClient : IInferenceClient
             {
                 case CommandType.InferenceData:
                     var annotationImage = MessagePackSerializer.Deserialize<AnnotationImage>(remoteCommand.Data, cancellationToken: ct);
+                    _logger.LogInformation("Received command: {AnnotationImage}", annotationImage.ToString());
                     await _mediator.Publish(new InferenceDataEvent(annotationImage), ct);
                     break;
                 case CommandType.InferenceStatus:

Azaion.Common/Services/Inference/InferenceServiceEventHandler.cs

@@ -1,18 +1,16 @@
-using Azaion.Common.Database;
-using Azaion.Common.DTO;
 using Azaion.Common.Events;
 using MediatR;
-using MessagePack;
-using Microsoft.Extensions.Logging;
 
 namespace Azaion.Common.Services.Inference;
 
-public class InferenceServiceEventHandler(IInferenceService inferenceService, IAnnotationService annotationService, IMediator mediator) :
+public class InferenceServiceEventHandler(
+    IInferenceService inferenceService, 
+    IAnnotationService annotationService, 
+    IMediator mediator) :
         INotificationHandler<InferenceDataEvent>,
         INotificationHandler<InferenceStatusEvent>,
         INotificationHandler<InferenceDoneEvent>
 {
-
     public async Task Handle(InferenceDataEvent e, CancellationToken ct)
     {
         var annotation = await annotationService.SaveAnnotation(e.AnnotationImage, ct);

Azaion.Common/Services/SatelliteDownloader.cs

@@ -1,5 +1,4 @@
 using System.Collections.Concurrent;
-using System.Diagnostics;
 using System.IO;
 using System.Net.Http;
 using System.Net.Http.Json;
@@ -7,7 +6,6 @@ using Azaion.Common.DTO;
 using Azaion.Common.DTO.Config;
 using Azaion.Common.Events;
 using Azaion.Common.Extensions;
-using Azaion.CommonSecurity;
 using MediatR;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;