using Deal.Grpc.Storage;
using Deal.Storage.Services;
using Grpc.Core;
namespace Deal.Storage;
///
/// Реализация сервиса данных.
///
public sealed class StorageServiceImpl : StorageService.StorageServiceBase
{
private const int HeadSize = 512;
private const int DownloadChunkSize = 64 * 1024;
private readonly MinioObjectStore _store;
private readonly ILogger _logger;
///
/// Создаёт реализацию поверх хранилища.
///
public StorageServiceImpl(MinioObjectStore store, ILogger logger)
{
ArgumentNullException.ThrowIfNull(store);
ArgumentNullException.ThrowIfNull(logger);
_store = store;
_logger = logger;
}
///
public override async Task Upload(
IAsyncStreamReader requestStream,
ServerCallContext context)
{
CancellationToken ct = context.CancellationToken;
await using MemoryStream buffer = new();
byte[] head = new byte[HeadSize];
int headLength = 0;
UploadMeta? meta = null;
while (await requestStream.MoveNext(ct))
{
UploadRequest message = requestStream.Current;
if (meta is null && message.Meta is not null)
{
meta = message.Meta;
}
if (message.Data.Length == 0)
{
continue;
}
if (headLength < HeadSize)
{
int take = Math.Min(HeadSize - headLength, message.Data.Length);
message.Data.Span[..take].CopyTo(head.AsSpan(headLength));
headLength += take;
}
await buffer.WriteAsync(message.Data.Memory, ct);
}
if (meta is null || string.IsNullOrWhiteSpace(meta.TenantId))
{
throw new RpcException(new Status(StatusCode.InvalidArgument, "tenant_id обязателен"));
}
(string kind, string mimeType) = FileKindSniffer.Detect(head.AsSpan(0, headLength), meta.ContentType);
string key = $"{meta.TenantId.Trim()}/{Guid.NewGuid():N}";
buffer.Position = 0;
long size = await _store.PutAsync(key, buffer, mimeType, ct);
_logger.LogInformation("Объект сохранён: {Key} ({Kind}, {Size} байт)", key, kind, size);
return new UploadReply
{
Id = key,
Ref = key,
Kind = kind,
MimeType = mimeType,
FileName = meta.FileName ?? string.Empty,
Size = size,
};
}
///
public override async Task Download(
DownloadRequest request,
IServerStreamWriter responseStream,
ServerCallContext context)
{
CancellationToken ct = context.CancellationToken;
await using Stream? content = await _store.GetAsync(request.Id, ct);
if (content is null)
{
throw new RpcException(new Status(StatusCode.NotFound, "объект не найден"));
}
byte[] chunk = new byte[DownloadChunkSize];
int read;
while ((read = await content.ReadAsync(chunk, ct)) > 0)
{
await responseStream.WriteAsync(new DownloadChunk { Data = Google.Protobuf.ByteString.CopyFrom(chunk, 0, read) }, ct);
}
}
///
public override async Task Stat(StatRequest request, ServerCallContext context)
{
(long Size, string ContentType)? stat = await _store.StatAsync(request.Id, context.CancellationToken);
if (stat is null)
{
return new StatReply { Found = false };
}
return new StatReply
{
Found = true,
Info = new UploadReply
{
Id = request.Id,
Ref = request.Id,
MimeType = stat.Value.ContentType,
Size = stat.Value.Size,
},
};
}
///
public override async Task Delete(DeleteRequest request, ServerCallContext context)
{
await _store.DeleteAsync(request.Id, context.CancellationToken);
return new DeleteReply();
}
}