Skip to content
rachel/workerd
File

Blob: src/workerd/api/r2-multipart.c++

10.5 KB
1// Copyright (c) 2017-2022 Cloudflare, Inc.
2// Licensed under the Apache 2.0 license found in the LICENSE file or at:
3//     https://opensource.org/licenses/Apache-2.0
4 
5#include "r2-multipart.h"
6 
7#include "r2-bucket.h"
8#include "r2-rpc.h"
9#include "workerd/jsg/jsg.h"
10 
11#include <workerd/api/r2-api.capnp.h>
12#include <workerd/util/http-util.h>
13 
14#include <capnp/compat/json.h>
15#include <capnp/message.h>
16#include <kj/compat/http.h>
17#include <kj/encoding.h>
18 
19#include <regex>
20 
21namespace workerd::api::public_beta {
22 
23static void addR2ResponseSpanTags(TraceContext& traceContext, R2Result& r2Result) {
24  traceContext.setTag("cloudflare.r2.response.success"_kjc, r2Result.success());
25  KJ_IF_SOME(e, r2Result.getR2ErrorMessage()) {
26    traceContext.setTag("error.type"_kjc, e.asPtr());
27    traceContext.setTag("cloudflare.r2.error.message"_kjc, e.asPtr());
28  }
29  KJ_IF_SOME(v4, r2Result.v4ErrorCode()) {
30    traceContext.setTag("cloudflare.r2.error.code"_kjc, static_cast<int64_t>(v4));
31  }
32}
33 
34jsg::Promise<R2MultipartUpload::UploadedPart> R2MultipartUpload::uploadPart(jsg::Lock& js,
35    int partNumber,
36    R2PutValue value,
37    jsg::Optional<UploadPartOptions> options,
38    const jsg::TypeHandler<jsg::Ref<R2Error>>& errorType) {
39  return js.evalNow([&] {
40    JSG_REQUIRE(partNumber >= 1 && partNumber <= 10000, TypeError,
41        "Part number must be between 1 and 10000 (inclusive). Actual value was: ", partNumber);
42 
43    auto& context = IoContext::current();
44    TraceContext traceContext = context.makeUserTraceSpan("r2_uploadPart"_kjc);
45 
46    traceContext.setTag("cloudflare.binding.type"_kjc, "r2"_kjc);
47    KJ_IF_SOME(b, this->bucket->bindingName()) {
48      traceContext.setTag("cloudflare.binding.name"_kjc, b);
49    }
50    traceContext.setTag("cloudflare.r2.operation"_kjc, "UploadPart"_kjc);
51    KJ_IF_SOME(b, this->bucket->bucketName()) {
52      traceContext.setTag("cloudflare.r2.bucket"_kjc, b);
53    }
54    traceContext.setTag("cloudflare.r2.request.upload_id"_kjc, uploadId.asPtr());
55    traceContext.setTag("cloudflare.r2.request.part_number"_kjc, static_cast<int64_t>(partNumber));
56    traceContext.setTag("cloudflare.r2.request.key"_kjc, key.asPtr());
57 
58    capnp::JsonCodec json;
59    json.handleByAnnotation<R2BindingRequest>();
60    json.setHasMode(capnp::HasMode::NON_DEFAULT);
61    capnp::MallocMessageBuilder requestMessage;
62 
63    auto requestBuilder = requestMessage.initRoot<R2BindingRequest>();
64    requestBuilder.setVersion(VERSION_PUBLIC_BETA);
65    auto payloadBuilder = requestBuilder.initPayload();
66    auto uploadPartBuilder = payloadBuilder.initUploadPart();
67 
68    uploadPartBuilder.setUploadId(uploadId);
69    uploadPartBuilder.setPartNumber(partNumber);
70    uploadPartBuilder.setObject(key);
71    KJ_IF_SOME(options, options) {
72      KJ_IF_SOME(ssecKey, options.ssecKey) {
73        auto ssecBuilder = uploadPartBuilder.initSsec();
74        KJ_SWITCH_ONEOF(ssecKey) {
75          KJ_CASE_ONEOF(keyString, kj::String) {
76            JSG_REQUIRE(
77                std::regex_match(keyString.begin(), keyString.end(), std::regex("^[0-9a-f]+$")),
78                Error, "SSE-C Key has invalid format");
79            JSG_REQUIRE(keyString.size() == 64, Error, "SSE-C Key must be 32 bytes in length");
80            ssecBuilder.setKey(kj::str(keyString));
81            traceContext.setTag("cloudflare.r2.request.ssec_key"_kjc, true);
82          }
83          KJ_CASE_ONEOF(keyBuff, kj::Array<byte>) {
84            JSG_REQUIRE(keyBuff.size() == 32, Error, "SSE-C Key must be 32 bytes in length");
85            ssecBuilder.setKey(kj::encodeHex(keyBuff));
86            traceContext.setTag("cloudflare.r2.request.ssec_key"_kjc, true);
87          }
88        }
89      }
90    }
91 
92    kj::Maybe<int64_t> requestSize = kj::none;
93    KJ_SWITCH_ONEOF(value) {
94      KJ_CASE_ONEOF(stream, jsg::Ref<ReadableStream>) {
95        KJ_IF_SOME(size, stream->tryGetLength(StreamEncoding::IDENTITY)) {
96          requestSize = size;
97        }
98      }
99      KJ_CASE_ONEOF(text, jsg::NonCoercible<kj::String>) {
100        requestSize = text.value.size();
101      }
102      KJ_CASE_ONEOF(data, kj::Array<byte>) {
103        requestSize = data.size();
104      }
105      KJ_CASE_ONEOF(blob, jsg::Ref<Blob>) {
106        requestSize = blob->getSize();
107      }
108    }
109    KJ_IF_SOME(size, requestSize) {
110      traceContext.setTag("cloudflare.r2.request.size"_kjc, size);
111    }
112 
113    auto requestJson = json.encode(requestBuilder);
114    auto bucket = this->bucket->adminBucket.map([](auto&& s) { return kj::str(s); });
115 
116    kj::StringPtr components[1];
117    auto path = fillR2Path(components, this->bucket->adminBucket);
118    auto client = context.getHttpClient(this->bucket->clientIndex, true, kj::none, traceContext);
119    auto promise = doR2HTTPPutRequest(
120        kj::mv(client), kj::mv(value), kj::none, kj::mv(requestJson), path, kj::none);
121 
122    return context.awaitIo(js, kj::mv(promise),
123        [&errorType, partNumber, traceContext = kj::mv(traceContext)](
124            jsg::Lock& js, R2Result r2Result) mutable {
125      addR2ResponseSpanTags(traceContext, r2Result);
126      r2Result.throwIfError("uploadPart", errorType);
127 
128      capnp::MallocMessageBuilder responseMessage;
129      capnp::JsonCodec json;
130      json.handleByAnnotation<R2UploadPartResponse>();
131      auto responseBuilder = responseMessage.initRoot<R2UploadPartResponse>();
132 
133      json.decode(KJ_ASSERT_NONNULL(r2Result.metadataPayload), responseBuilder);
134      kj::StringPtr etag = responseBuilder.getEtag();
135      traceContext.setTag("cloudflare.r2.response.etag"_kjc, etag);
136      UploadedPart uploadedPart = {partNumber, kj::str(etag)};
137      return uploadedPart;
138    });
139  });
140}
141 
142jsg::Promise<jsg::Ref<R2Bucket::HeadResult>> R2MultipartUpload::complete(jsg::Lock& js,
143    kj::Array<UploadedPart> uploadedParts,
144    const jsg::TypeHandler<jsg::Ref<R2Error>>& errorType) {
145  return js.evalNow([&] {
146    auto& context = IoContext::current();
147    TraceContext traceContext = context.makeUserTraceSpan("r2_completeMultipartUpload"_kjc);
148 
149    traceContext.setTag("cloudflare.binding.type"_kjc, "r2"_kjc);
150    KJ_IF_SOME(b, this->bucket->bindingName()) {
151      traceContext.setTag("cloudflare.binding.name"_kjc, b);
152    }
153    traceContext.setTag("cloudflare.r2.operation"_kjc, "CompleteMultipartUpload"_kjc);
154    KJ_IF_SOME(b, this->bucket->bucketName()) {
155      traceContext.setTag("cloudflare.r2.bucket"_kjc, b);
156    }
157    traceContext.setTag("cloudflare.r2.request.upload_id"_kjc, uploadId.asPtr());
158    traceContext.setTag("cloudflare.r2.request.key"_kjc, key.asPtr());
159    kj::String partIds =
160        kj::strArray(KJ_MAP(part, uploadedParts) { return kj::str(part.partNumber); }, ", ");
161    traceContext.setTag("cloudflare.r2.request.uploaded_parts"_kjc, kj::mv(partIds));
162 
163    capnp::JsonCodec json;
164    json.handleByAnnotation<R2BindingRequest>();
165    capnp::MallocMessageBuilder requestMessage;
166 
167    auto requestBuilder = requestMessage.initRoot<R2BindingRequest>();
168    requestBuilder.setVersion(VERSION_PUBLIC_BETA);
169    auto completeMultipartUploadBuilder =
170        requestBuilder.initPayload().initCompleteMultipartUpload();
171 
172    completeMultipartUploadBuilder.setObject(key);
173    completeMultipartUploadBuilder.setUploadId(uploadId);
174 
175    auto partsList = completeMultipartUploadBuilder.initParts(uploadedParts.size());
176    UploadedPart* currentPart = uploadedParts.begin();
177    for (unsigned int i = 0; i < uploadedParts.size(); i++) {
178      int partNumber = currentPart->partNumber;
179      JSG_REQUIRE(partNumber >= 1 && partNumber <= 10000, TypeError,
180          "Part number must be between 1 and 10000 (inclusive). Actual value was: ", partNumber);
181      partsList[i].setPart(partNumber);
182      partsList[i].setEtag(currentPart->etag);
183      currentPart = std::next(currentPart);
184    }
185 
186    auto requestJson = json.encode(requestBuilder);
187 
188    kj::StringPtr components[1];
189    auto path = fillR2Path(components, this->bucket->adminBucket);
190    auto client = context.getHttpClient(this->bucket->clientIndex, true, kj::none, traceContext);
191    auto promise =
192        doR2HTTPPutRequest(kj::mv(client), kj::none, kj::none, kj::mv(requestJson), path, kj::none);
193 
194    return context.awaitIo(js, kj::mv(promise),
195        [&errorType, traceContext = kj::mv(traceContext)](
196            jsg::Lock& js, R2Result r2Result) mutable {
197      addR2ResponseSpanTags(traceContext, r2Result);
198      auto parsedObject =
199          parseHeadResultWrapper(js, "completeMultipartUpload", r2Result, errorType);
200      KJ_IF_SOME(obj, parsedObject) {
201        addHeadResultSpanTags(js, traceContext, *obj.get());
202        return obj.addRef();
203      } else {
204        KJ_FAIL_ASSERT(
205            "Shouldn't happen, multipart completion should either error or return an object");
206      }
207    });
208  });
209}
210 
211jsg::Promise<void> R2MultipartUpload::abort(
212    jsg::Lock& js, const jsg::TypeHandler<jsg::Ref<R2Error>>& errorType) {
213  return js.evalNow([&] {
214    auto& context = IoContext::current();
215    TraceContext traceContext = context.makeUserTraceSpan("r2_abortMultipartUpload"_kjc);
216 
217    traceContext.setTag("cloudflare.binding.type"_kjc, "r2"_kjc);
218    KJ_IF_SOME(b, this->bucket->bindingName()) {
219      traceContext.setTag("cloudflare.binding.name"_kjc, b);
220    }
221    traceContext.setTag("cloudflare.r2.operation"_kjc, "AbortMultipartUpload"_kjc);
222    KJ_IF_SOME(b, this->bucket->bucketName()) {
223      traceContext.setTag("cloudflare.r2.bucket"_kjc, b);
224    }
225    traceContext.setTag("cloudflare.r2.request.upload_id"_kjc, uploadId.asPtr());
226    traceContext.setTag("cloudflare.r2.request.key"_kjc, key.asPtr());
227 
228    capnp::JsonCodec json;
229    json.handleByAnnotation<R2BindingRequest>();
230    capnp::MallocMessageBuilder requestMessage;
231 
232    auto requestBuilder = requestMessage.initRoot<R2BindingRequest>();
233    requestBuilder.setVersion(VERSION_PUBLIC_BETA);
234    auto abortMultipartUploadBuilder = requestBuilder.initPayload().initAbortMultipartUpload();
235 
236    abortMultipartUploadBuilder.setObject(key);
237    abortMultipartUploadBuilder.setUploadId(uploadId);
238 
239    auto requestJson = json.encode(requestBuilder);
240 
241    kj::StringPtr components[1];
242    auto path = fillR2Path(components, this->bucket->adminBucket);
243    auto client = context.getHttpClient(this->bucket->clientIndex, true, kj::none, traceContext);
244    auto promise =
245        doR2HTTPPutRequest(kj::mv(client), kj::none, kj::none, kj::mv(requestJson), path, kj::none);
246 
247    return context.awaitIo(js, kj::mv(promise),
248        [&errorType, traceContext = kj::mv(traceContext)](
249            jsg::Lock& js, R2Result r2Result) mutable {
250      addR2ResponseSpanTags(traceContext, r2Result);
251      if (r2Result.objectNotFound()) {
252        return;
253      }
254 
255      r2Result.throwIfError("abortMultipartUpload", errorType);
256    });
257  });
258}
259}  // namespace workerd::api::public_beta