soa/service/s3.h

/* s3.h                                                            -*- C++ -*-
   Jeremy Barnes, 3 July 2012
   Copyright (c) 2012 Datacratic.  All rights reserved.

   Class to deal with doing s3.
*/

#pragma once

#include <string>
#include <vector>
#include <map>
#include "jml/arch/exception.h"
#include "jml/utils/unnamed_bool.h"
#include "tinyxml2/tinyxml2.h"
#include "soa/service/http_endpoint.h"
#include <memory>

namespace Datacratic {

struct S3Api {
    static std::string sign(const std::string & stringToSign,
                            const std::string & accessKey);

    static double defaultBandwidthToServiceMbps;

    S3Api();

    S3Api(const std::string & accessKeyId,
          const std::string & accessKey,
          double bandwidthToServiceMbps = defaultBandwidthToServiceMbps,
          const std::string & defaultProtocol = "http",
          const std::string & serviceUri = "s3.amazonaws.com");

    void init(const std::string & accessKeyId,
              const std::string & accessKey,
              double bandwidthToServiceMbps = defaultBandwidthToServiceMbps,
              const std::string & defaultProtocol = "http",
              const std::string & serviceUri = "s3.amazonaws.com");

    std::string accessKeyId;
    std::string accessKey;
    std::string defaultProtocol;
    std::string serviceUri;
    double bandwidthToServiceMbps;

    typedef std::vector<std::pair<std::string, std::string> > StrPairVector;

    struct Content {
        Content()
            : data(0), size(0), hasContent(false)
        {
        }

        Content(const char * data, uint64_t size,
                const std::string & contentType = "",
                const std::string & contentMd5 = "")
            : data(data), size(size), hasContent(true),
              contentType(contentType), contentMd5(contentMd5)
        {
        }

        Content(const tinyxml2::XMLDocument & xml);

        const char * data;
        uint64_t size;
        bool hasContent;

        std::string str;
        std::string contentType;
        std::string contentMd5;
    };

    struct RequestParams {

        RequestParams()
            : expectedBytesToDownload(0)
        {
        }

        std::string verb;
        std::string bucket;
        std::string resource;
        std::string subResource;
        std::string date;

        std::string contentType;
        std::string contentMd5;
        Content content;
        uint64_t expectedBytesToDownload;

        StrPairVector headers;
        StrPairVector queryParams;
    };

    struct Response {
        Response()
            : code_(0)
        {
        }

        std::string body() const
        {
            if (code_ < 200 || code_ >= 300)
                throw ML::Exception("invalid http code returned");
            return body_;
        }

        std::unique_ptr<tinyxml2::XMLDocument> bodyXml() const
        {
            std::unique_ptr<tinyxml2::XMLDocument> result(new tinyxml2::XMLDocument());
            result->Parse(body_.c_str());
            return result;
        }

        operator std::unique_ptr<tinyxml2::XMLDocument>() const
        {
            return bodyXml();
        }

        std::string bodyXmlStr() const
        {
            auto x = bodyXml();
            tinyxml2::XMLPrinter printer;
            x->Print(&printer);
            return printer.CStr();
        }

        std::string getHeader(const std::string & name) const
        {
            auto it = header_.headers.find(name);
            if (it == header_.headers.end())
                throw ML::Exception("required header " + name + " not found");
            return it->second;
        }

        long code_;
        std::string body_;
        HttpHeader header_;
    };

    enum Redundancy {
        REDUNDANCY_STANDARD,
        REDUNDANCY_REDUCED,
        REDUNDANCY_GLACIER
    };

    enum ServerSideEncryption {
        SSE_NONE,
        SSE_AES256
    };

    struct ObjectMetadata {
        ObjectMetadata()
            : redundancy(REDUNDANCY_STANDARD),
              serverSideEncryption(SSE_NONE)
        {
        }

        ObjectMetadata(const Redundancy & redundancy)
            : redundancy(redundancy),
              serverSideEncryption(SSE_NONE)
        {
        }

        std::vector<std::pair<std::string, std::string> >
        getRequestHeaders() const;

        Redundancy redundancy;
        ServerSideEncryption serverSideEncryption;
        std::string contentType;
        std::string contentEncoding;
        std::map<std::string, std::string> metadata;
    };

    struct SignedRequest : public RequestParams {
        RequestParams params;
        std::string auth;
        std::string uri;
        double bandwidthToServiceMbps;

        Response performSync() const;
    };

    std::string signature(const RequestParams & request) const;

    SignedRequest prepare(const RequestParams & request) const;

    Response get(const std::string & bucket,
                 const std::string & resource,
                 uint64_t expectedBytesToTransfer,
                 const std::string & subResource = "",
                 const StrPairVector & headers = StrPairVector(),
                 const StrPairVector & queryParams = StrPairVector()) const;


    Response post(const std::string & bucket,
                  const std::string & resource,
                  const std::string & subResource = "",
                  const StrPairVector & headers = StrPairVector(),
                  const StrPairVector & queryParams = StrPairVector(),
                  const Content & content = Content()) const;

    Response put(const std::string & bucket,
                 const std::string & resource,
                 const std::string & subResource = "",
                 const StrPairVector & headers = StrPairVector(),
                 const StrPairVector & queryParams = StrPairVector(),
                 const Content & content = Content()) const;

     Response erase(const std::string & bucket,
                    const std::string & resource,
                    const std::string & subResource = "",
                    const StrPairVector & headers = StrPairVector(),
                    const StrPairVector & queryParams = StrPairVector(),
                    const Content & content = Content()) const;


    enum CheckMethod {
        CM_SIZE,     
        CM_MD5_ETAG, 
        CM_ASSUME_INVALID  
    };

    std::string upload(const char * data,
                       size_t bytes,
                       const std::string & bucket,
                       const std::string & resource,
                       CheckMethod check = CM_SIZE,
                       const ObjectMetadata & md = ObjectMetadata(),
                       int numInParallel = -1);

    std::string upload(const char * data,
                       size_t bytes,
                       const std::string & uri,
                       CheckMethod check = CM_SIZE,
                       const ObjectMetadata & md = ObjectMetadata(),
                       int numInParallel = -1);

    typedef std::function<void (const char * chunk,
                                size_t size,
                                int chunkIndex,
                                uint64_t offset,
                                uint64_t totalSize) >
        OnChunk;

    static OnChunk writeToFile(const std::string & filename);

    void download(const std::string & bucket,
                  const std::string & object,
                  const OnChunk & onChunk,
                  ssize_t startOffset = 0,
                  ssize_t endOffset = -1) const;

    void download(const std::string & uri,
                  const OnChunk & onChunk,
                  ssize_t startOffset = 0,
                  ssize_t endOffset = -1) const;

    void downloadToFile(const std::string & uri,
                  const std::string & outfile,
                  ssize_t endOffset = -1) const;

    struct StreamingDownloadStreambuf;

    std::auto_ptr<std::streambuf>
    streamingDownload(const std::string & bucket,
                      const std::string & object,
                      ssize_t startOffset = 0,
                      ssize_t endOffset = -1,
                      const OnChunk & onChunk = OnChunk()) const;

    std::auto_ptr<std::streambuf>
    streamingDownload(const std::string & uri,
                      ssize_t startOffset = 0,
                      ssize_t endOffset = -1,
                      const OnChunk & onChunk = OnChunk()) const;

    std::auto_ptr<std::streambuf>
    streamingUpload(const std::string & uri,
                    const ObjectMetadata & md = ObjectMetadata()) const;

    std::auto_ptr<std::streambuf>
    streamingUpload(const std::string & bucket,
                    const std::string & object,
                    const ObjectMetadata & md = ObjectMetadata()) const;

    struct ObjectInfo {
        ObjectInfo();
        ObjectInfo(tinyxml2::XMLNode * element);

        JML_IMPLEMENT_OPERATOR_BOOL(exists);

        std::string key;
        uint64_t size;
        bool exists;
        std::string etag;
        std::string ownerId;
        std::string ownerName;
        Date lastModified;
        std::string storageClass;
    };

    typedef std::function<bool (const std::string & prefix,
                                const std::string & objectName,
                                const ObjectInfo & info,
                                int depth)>
        OnObject;

    typedef std::function<bool (const std::string & prefix,
                                const std::string & dirName,
                                int depth)>
        OnSubdir;

    void forEachObject(const std::string & bucket,
                       const std::string & prefix = "",
                       const OnObject & onObject = OnObject(),
                       const OnSubdir & onSubdir = OnSubdir(),
                       const std::string & delimiter = "/",
                       int depth = 1) const;

    ObjectInfo tryGetObjectInfo(const std::string & bucket,
                                const std::string & object) const;

    ObjectInfo tryGetObjectInfo(const std::string & uri) const;


    ObjectInfo getObjectInfo(const std::string & bucket,
                             const std::string & object) const;

    ObjectInfo getObjectInfo(const std::string & uri) const;

    typedef std::function<bool (std::string bucket)> OnBucket;

    bool forEachBucket(const OnBucket & bucket) const;

    static std::string
    getDigestMulti(const std::string & verb,
                   const std::string & bucket,
                   const std::string & resource,
                   const std::string & subResource,
                   const std::string & contentType,
                   const std::string & contentMd5,
                   const std::string & date,
                   const std::vector<std::pair<std::string, std::string> > & headers);

    static std::string
    getDigest(const std::string & verb,
              const std::string & bucket,
              const std::string & resource,
              const std::string & subResource,
              const std::string & contentType,
              const std::string & contentMd5,
              const std::string & date,
              const std::map<std::string, std::string> & headers);

    static std::pair<std::string, std::string>
    parseUri(const std::string & uri);

    struct MultiPartUploadPart {
        MultiPartUploadPart()
            : partNumber(0), done(false)
        {
        }

        int partNumber;
        uint64_t startOffset;
        uint64_t size;
        std::string lastModified;
        std::string contentMd5;
        std::string etag;
        bool done;

        void fromXml(tinyxml2::XMLElement * element);
    };

    struct MultiPartUpload {
        std::string id;
        std::vector<MultiPartUploadPart> parts;
    };

    MultiPartUpload
    obtainMultiPartUpload(const std::string & bucket,
                          const std::string & resource,
                          const ObjectMetadata & metadata) const;

    std::string
    finishMultiPartUpload(const std::string & bucket,
                          const std::string & resource,
                          const std::string & uploadId,
                          const std::vector<std::string> & etags) const;

    void uploadRecursive(std::string dirSrc,
                         std::string bucketDest,
                         bool includeDir);

    //easy handle for v8 wrapping
    void setDefaultBandwidthToServiceMbps(double mpbs);

};

struct S3IStream : public std::istream {
    S3IStream();
    S3IStream(const S3Api & s3, const std::string & uri);

    void open(const std::string & uri);
    void open(const std::string & bucket,
              const std::string & object);
};

struct S3Handle{
    S3Api s3;
    std::string s3UriPrefix;

    void initS3(const std::string & accessKeyId,
                const std::string & accessKey,
                const std::string & uriPrefix);

    size_t getS3Buffer(const std::string & filename, char** outBuffer);
};

class BucketAlreadyRegistered : public ML::Exception{
    public:
        BucketAlreadyRegistered(const std::string & bucketName) : 
                ML::Exception("s3 bucket %s already registered",
                              bucketName.c_str()){}
};

void registerS3Bucket(const std::string & bucketName,
                      const std::string & accessKeyId,
                      const std::string & accessKey,
                      double bandwidthToServiceMbps = S3Api::defaultBandwidthToServiceMbps,
                      const std::string & protocol = "http",
                      const std::string & serviceUri = "s3.amazonaws.com");

void registerS3Buckets(const std::string & accessKeyId,
                       const std::string & accessKey,
                       double bandwidthToServiceMbps = S3Api::defaultBandwidthToServiceMbps,
                       const std::string & protocol = "http",
                       const std::string & serviceUri = "s3.amazonaws.com");

std::shared_ptr<S3Api> getS3ApiForBucket(const std::string & bucketName);

// Return an URI for either a file or an s3 object
size_t getUriSize(const std::string & filename);

} // namespace Datacratic