quorate/packages/fsharp-runner/oci/Publisher.cs

using System.Net;
using System.Net.Http.Headers;

namespace Quorate.Oci;

internal sealed class OciPublisher(RegistryTransport transport, RepositoryName repository)
{
    private static readonly string[] ManifestTypes =
    [
        "application/vnd.docker.distribution.manifest.v2+json",
        "application/vnd.docker.distribution.manifest.list.v2+json",
        "application/vnd.oci.image.manifest.v1+json",
        "application/vnd.oci.image.index.v1+json"
    ];

    private static readonly TimeSpan UploadCancelTimeout = TimeSpan.FromSeconds(5);

    public async Task<OciDigest> PushAsync(
        OciPublicationPlan plan,
        OciTag target,
        CancellationToken cancellationToken)
    {
        var parallel = new ParallelOptions
        {
            CancellationToken = cancellationToken,
            MaxDegreeOfParallelism = 4
        };

        await Parallel.ForEachAsync(plan.Blobs, parallel, async (blob, token) =>
        {
            await VerifyBlobAsync(blob, token).ConfigureAwait(false);

            if (!await ExistsAsync(blob, token).ConfigureAwait(false))
            {
                await UploadAsync(blob, token).ConfigureAwait(false);
            }
        }).ConfigureAwait(false);

        foreach (OciManifest manifest in plan.Manifests)
        {
            await PutManifestAsync(
                manifest.Digest.ToString(),
                manifest.MediaType,
                manifest.Content,
                manifest.Digest,
                cancellationToken).ConfigureAwait(false);
        }

        await PutManifestAsync(
            target.ToString(),
            plan.RootMediaType,
            plan.RootContent,
            plan.RootDigest,
            cancellationToken).ConfigureAwait(false);
        OciDigest published = await GetDigestAsync(target, cancellationToken).ConfigureAwait(false);

        if (published != plan.RootDigest)
        {
            throw new InvalidDataException($"Published digest {published} does not match {plan.RootDigest}.");
        }

        return published;
    }

    private async Task<bool> ExistsAsync(OciBlob blob, CancellationToken cancellationToken)
    {
        using HttpResponseMessage response = await transport.SendAsync(
            repository,
            () => RegistryTransport.Request(HttpMethod.Head, transport.Api(repository, $"blobs/{blob.Digest}")),
            cancellationToken,
            HttpStatusCode.NotFound).ConfigureAwait(false);

        if (response.StatusCode == HttpStatusCode.NotFound)
        {
            return false;
        }

        OciDigest actual = HeaderDigest(response);
        long? size = response.Content.Headers.ContentLength;

        if (actual != blob.Digest || size != blob.Size)
        {
            throw new InvalidDataException($"Existing blob {blob.Digest} has invalid metadata.");
        }

        return true;
    }

    private async Task UploadAsync(OciBlob blob, CancellationToken cancellationToken)
    {
        using HttpResponseMessage start = await transport.SendAsync(
            repository,
            () => RegistryTransport.Request(HttpMethod.Post, transport.Api(repository, "blobs/uploads/")),
            cancellationToken).ConfigureAwait(false);
        Uri location = RegistryTransport.ResolveLocation(start);

        try
        {
            using HttpResponseMessage completed = await transport.SendAsync(
                repository,
                () =>
                {
                    HttpRequestMessage request = RegistryTransport.Request(
                        HttpMethod.Put,
                        AddDigest(location, blob.Digest));
                    request.Content = new StreamContent(new FileStream(
                        blob.Path,
                        FileMode.Open,
                        FileAccess.Read,
                        FileShare.Read,
                        131072,
                        FileOptions.Asynchronous | FileOptions.SequentialScan));
                    request.Content.Headers.ContentType = new MediaTypeHeaderValue("application/octet-stream");
                    return request;
                },
                cancellationToken).ConfigureAwait(false);
            OciDigest actual = HeaderDigest(completed);

            if (actual != blob.Digest)
            {
                throw new InvalidDataException($"Uploaded digest {actual} does not match {blob.Digest}.");
            }
        }
        catch (Exception)
        {
            await CancelUploadAsync(location).ConfigureAwait(false);
            throw;
        }
    }

    private async Task CancelUploadAsync(Uri location)
    {
        using var timeout = new CancellationTokenSource(UploadCancelTimeout);

        try
        {
            using HttpResponseMessage response = await transport.SendAsync(
                repository,
                () => RegistryTransport.Request(HttpMethod.Delete, location),
                timeout.Token,
                HttpStatusCode.NotFound).ConfigureAwait(false);
        }
        catch (Exception error) when (error is HttpRequestException or InvalidDataException or OperationCanceledException)
        {
        }
    }

    private async Task PutManifestAsync(
        string reference,
        string mediaType,
        byte[] content,
        OciDigest expected,
        CancellationToken cancellationToken)
    {
        using HttpResponseMessage response = await transport.SendAsync(
            repository,
            () =>
            {
                HttpRequestMessage request = RegistryTransport.Request(
                    HttpMethod.Put,
                    transport.Api(repository, $"manifests/{reference}"));
                request.Content = new ByteArrayContent(content);
                request.Content.Headers.ContentType = new MediaTypeHeaderValue(mediaType);
                return request;
            },
            cancellationToken).ConfigureAwait(false);
        OciDigest actual = HeaderDigest(response);

        if (actual != expected)
        {
            throw new InvalidDataException($"Published digest {actual} does not match {expected}.");
        }
    }

    private async Task<OciDigest> GetDigestAsync(OciTag reference, CancellationToken cancellationToken)
    {
        using HttpResponseMessage response = await transport.SendAsync(
            repository,
            () =>
            {
                HttpRequestMessage request = RegistryTransport.Request(
                    HttpMethod.Head,
                    transport.Api(repository, $"manifests/{reference}"));

                foreach (string mediaType in ManifestTypes)
                {
                    request.Headers.Accept.Add(new MediaTypeWithQualityHeaderValue(mediaType));
                }

                return request;
            },
            cancellationToken).ConfigureAwait(false);
        return HeaderDigest(response);
    }

    private static async Task VerifyBlobAsync(OciBlob blob, CancellationToken cancellationToken)
    {
        FileInfo info = new(blob.Path);

        if (!info.Exists || info.Length != blob.Size)
        {
            throw new InvalidDataException($"OCI blob {blob.Digest} has an invalid size.");
        }

        OciDigest actual = await OciDigest.ComputeFileAsync(blob.Path, cancellationToken).ConfigureAwait(false);

        if (actual != blob.Digest)
        {
            throw new InvalidDataException($"Blob digest {actual} does not match {blob.Digest}.");
        }
    }

    private static Uri AddDigest(Uri location, OciDigest digest)
    {
        var builder = new UriBuilder(location);
        string pair = "digest=" + Uri.EscapeDataString(digest.ToString());
        builder.Query = string.IsNullOrEmpty(builder.Query)
            ? pair
            : builder.Query.TrimStart('?') + "&" + pair;
        return builder.Uri;
    }

    private static OciDigest HeaderDigest(HttpResponseMessage response) =>
        response.Headers.TryGetValues("Docker-Content-Digest", out IEnumerable<string>? values)
            ? OciDigest.Parse(values.Single())
            : throw new InvalidDataException("Registry response lacks Docker-Content-Digest.");
}