54 lines
5.0 KiB
C#
54 lines
5.0 KiB
C#
using Amazon.S3;
|
|
using Amazon.S3.Model;
|
|
using FileService.Application.Ports;
|
|
using FileService.Application.StorageContracts;
|
|
using FileService.Domain.ValueObjects;
|
|
using IM.InitCommon;
|
|
|
|
namespace FileService.Infrastructure.Storage;
|
|
public sealed class S3StorageAdapter(string code, StorageProviderOptions options) : IObjectStoragePort, IDisposable
|
|
{
|
|
private readonly AmazonS3Client client = new(options.AccessKeyId, options.AccessKeySecret, new AmazonS3Config { ServiceURL = options.Endpoint, AuthenticationRegion = string.IsNullOrWhiteSpace(options.Region) ? "us-east-1" : options.Region, ForcePathStyle = true });
|
|
public string ProviderCode => code;
|
|
public async Task<UploadPart> WritePartAsync(UploadRuntimeCache task, int partNumber, Stream content, long size, CancellationToken ct) {
|
|
var part = await client.UploadPartAsync(new UploadPartRequest { BucketName = task.Bucket, Key = task.ObjectKey, UploadId = task.UploadSessionId, PartNumber = partNumber, InputStream = content, PartSize = size }, ct);
|
|
return new(partNumber, part.ETag, size);
|
|
}
|
|
public async Task<InitiateUploadResult> InitUploadAsync(InitiateUploadCommand command, CancellationToken token) {
|
|
var r = await client.InitiateMultipartUploadAsync(new InitiateMultipartUploadRequest { BucketName = command.Bucket, Key = command.ObjectKey, ContentType = command.ContentType }, token);
|
|
return new(r.UploadId, new StorageLocation(code, command.Bucket, command.ObjectKey, options.Region));
|
|
}
|
|
public Task<PresignedUrl> GenerateUploadUrlAsync(GenerateUploadUrlCommand command, CancellationToken token) {
|
|
var expires = DateTimeOffset.UtcNow.Add(command.ExpiresIn);
|
|
var request = new GetPreSignedUrlRequest { BucketName = command.Bucket, Key = command.ObjectKey, Verb = HttpVerb.PUT, Expires = expires.UtcDateTime, UploadId = command.UploadSessionId, PartNumber = command.PartNumber ?? 1 };
|
|
return Task.FromResult(new PresignedUrl(client.GetPreSignedURL(request), "PUT", new Dictionary<string, string>(), expires));
|
|
}
|
|
public async Task<CompleteUploadResult> CompleteUploadAsync(CompleteUploadCommand command, CancellationToken token) {
|
|
var response = await client.CompleteMultipartUploadAsync(new CompleteMultipartUploadRequest { BucketName = command.Bucket, Key = command.ObjectKey, UploadId = command.UploadSessionId, PartETags = command.Parts.OrderBy(x => x.PartNumber).Select(x => new PartETag(x.PartNumber, x.ETag)).ToList() }, token);
|
|
var meta = await client.GetObjectMetadataAsync(command.Bucket, command.ObjectKey, token);
|
|
return new(new StorageLocation(code, command.Bucket, command.ObjectKey, options.Region), response.ETag, meta.ContentLength, VersionId: response.VersionId);
|
|
}
|
|
public async Task<StorageLocation> PutObjectAsync(PutObjectCommand command, CancellationToken token) {
|
|
await client.PutObjectAsync(new PutObjectRequest { BucketName = command.Bucket, Key = command.ObjectKey, ContentType = command.ContentType, InputStream = command.Content, AutoCloseStream = false }, token);
|
|
return new(code, command.Bucket, command.ObjectKey, options.Region);
|
|
}
|
|
// Public rendering also passes through FileService. Private access is always authorized there.
|
|
public string? GetPublicUrl(StorageLocation location) => null;
|
|
public async Task<Stream> OpenReadAsync(StorageLocation location, CancellationToken token) { var response = await client.GetObjectAsync(location.Bucket, location.ObjectKey, token); return new ResponseStream(response); }
|
|
public async Task Test(CancellationToken ct) {
|
|
var key = "im-admin-connectivity/" + Guid.NewGuid().ToString("N");
|
|
try { await client.PutObjectAsync(new PutObjectRequest { BucketName = options.Bucket, Key = key, ContentBody = "IM connectivity test" }, ct); using var read = await client.GetObjectAsync(options.Bucket, key, ct); }
|
|
finally { await client.DeleteObjectAsync(options.Bucket, key, CancellationToken.None); }
|
|
}
|
|
public void Dispose() => client.Dispose();
|
|
sealed class ResponseStream(GetObjectResponse response) : Stream {
|
|
readonly Stream inner = response.ResponseStream;
|
|
public override bool CanRead => inner.CanRead; public override bool CanSeek => inner.CanSeek; public override bool CanWrite => false;
|
|
public override long Length => response.ContentLength; public override long Position { get => inner.Position; set => inner.Position = value; }
|
|
public override void Flush() => inner.Flush(); public override int Read(byte[] b, int o, int c) => inner.Read(b, o, c);
|
|
public override ValueTask<int> ReadAsync(Memory<byte> b, CancellationToken ct = default) => inner.ReadAsync(b, ct);
|
|
public override long Seek(long o, SeekOrigin origin) => inner.Seek(o, origin); public override void SetLength(long v) => throw new NotSupportedException(); public override void Write(byte[] b, int o, int c) => throw new NotSupportedException();
|
|
protected override void Dispose(bool disposing) { if (disposing) response.Dispose(); base.Dispose(disposing); }
|
|
}
|
|
}
|