Replicator.hh

Go to the documentation of this file.

//
//  Replicator.hh
//
// Copyright (c) 2019 Couchbase, Inc All rights reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
 
#pragma once
#include "cbl++/Document.hh"
#include "cbl/CBLReplicator.h"
#include "cbl/CBLDefaults.h"
#include <functional>
#include <optional>
#include <string>
#include <string_view>
#include <vector>
#include <unordered_map>
 
 
CBL_ASSUME_NONNULL_BEGIN
 
namespace cbl {

    class Endpoint {
    public:        
        static Endpoint urlEndpoint(std::string_view url) {
            CBLError error {};
            auto endpoint = CBLEndpoint_CreateWithURL(slice(url), &error);
            internal::check(endpoint, error);
            return Endpoint(endpoint);
        }
        
#ifdef COUCHBASE_ENTERPRISE
        static Endpoint databaseEndpoint(Database db) {
            return Endpoint(CBLEndpoint_CreateWithLocalDB(db.ref()));
        }
#endif
        
    protected:
        friend class ReplicatorConfiguration;
        
        CBLEndpoint* _cbl_nullable ref() const {return _ref.get();}
 
    private:
        Endpoint() = default;
        
        Endpoint(CBLEndpoint* ref) {
            _ref = std::shared_ptr<CBLEndpoint>(ref, [](auto r) {
                CBLEndpoint_Free(r);
            });
        }
        
        std::shared_ptr<CBLEndpoint> _ref;
    };

    class Authenticator {
    public:
        static Authenticator basicAuthenticator(std::string_view username, std::string_view password) {
            return Authenticator(CBLAuth_CreatePassword(slice(username), slice(password)));
        }

        static Authenticator sessionAuthenticator(std::string_view sessionId, std::optional<std::string_view> cookieName =std::nullopt) {
            slice cname;
            if ( cookieName ) cname = *cookieName;
            return Authenticator(CBLAuth_CreateSession(slice(sessionId), cname));
        }
 
    protected:
        friend class ReplicatorConfiguration;
        
        CBLAuthenticator* _cbl_nullable ref() const {return _ref.get();}
        
    private:
        Authenticator() = default;
        
        Authenticator(CBLAuthenticator* ref) {
            _ref = std::shared_ptr<CBLAuthenticator>(ref, [](auto r) {
                CBLAuth_Free(r);
            });
        }
 
        std::shared_ptr<CBLAuthenticator> _ref;
    };

    using ReplicationFilter = std::function<bool(Document, CBLDocumentFlags flags)>;

    using ConflictResolver = std::function<Document(std::string_view docID,
                                                    const Document localDoc,
                                                    const Document remoteDoc)>;
 
#ifdef COUCHBASE_ENTERPRISE
    struct EncryptionResult {
        fleece::alloc_slice ciphertext;          
        std::optional<std::string> algorithm;   
        std::optional<std::string> keyID;       
        CBLError error {};              
    };

    struct DecryptionResult {
        fleece::alloc_slice plaintext;  
        CBLError error {};              
    };

    using PropertyEncryptor = std::function<EncryptionResult(
        fleece::slice scope,
        fleece::slice collection,
        fleece::slice documentID,
        fleece::Dict properties,
        fleece::slice keyPath,
        fleece::slice input)>;

    using PropertyDecryptor = std::function<DecryptionResult(
        fleece::slice scope,
        fleece::slice collection,
        fleece::slice documentID,
        fleece::Dict properties,
        fleece::slice keyPath,
        fleece::slice input,
        std::optional<std::string_view> algorithm,
        std::optional<std::string_view> keyID)>;
#endif

    class CollectionConfiguration {
    public:
        CollectionConfiguration(Collection collection)
        :_collection(collection)
        { }
        
        //-- Accessors:
        Collection collection() const       {return _collection;}
        
        //-- Filtering:
        fleece::MutableArray channels       = fleece::MutableArray::newArray();
        
        fleece::MutableArray documentIDs    = fleece::MutableArray::newArray();

        ReplicationFilter pushFilter;
        
        ReplicationFilter pullFilter;
        
        //-- Conflict Resolver:
        ConflictResolver conflictResolver;
        
    private:
        Collection _collection;
    };

    using ReplicationCollection = CollectionConfiguration;

    class ReplicatorConfiguration {
    public:
        ReplicatorConfiguration(std::vector<CollectionConfiguration>collections, Endpoint endpoint)
        :_collections(collections)
        ,_endpoint(endpoint)
        { }
        
        //-- Accessors:
        
        std::vector<CollectionConfiguration> collections() const  {return _collections;}
        
        Endpoint endpoint() const           {return _endpoint;}
        
        //-- Types:
        CBLReplicatorType replicatorType    = kCBLReplicatorTypePushAndPull;
        bool continuous                     = false;
        
        //-- Auto Purge:
        bool enableAutoPurge                = true;
        
        //-- Retry Logic:
        unsigned maxAttempts                = 0;
        unsigned maxAttemptWaitTime         = 0;
        
        //-- WebSocket:
        unsigned heartbeat                  = 0;
        
    #ifdef __CBL_REPLICATOR_NETWORK_INTERFACE__
        std::string networkInterface;
    #endif
 
        //-- HTTP settings:
        Authenticator authenticator;
        CBLProxySettings* _cbl_nullable proxy = nullptr;
        fleece::MutableDict headers         = fleece::MutableDict::newDict();
 
        //-- Advance HTTP settings:
        bool acceptParentDomainCookies      = kCBLDefaultReplicatorAcceptParentCookies;
 
        //-- TLS settings:
        std::string pinnedServerCertificate;
        std::string trustedRootCertificates;
#ifdef COUCHBASE_ENTERPRISE
        bool acceptOnlySelfSignedServerCertificate = false;
#endif
 
#ifdef COUCHBASE_ENTERPRISE
        //-- Property Encryption (Enterprise Edition only):
        PropertyEncryptor documentPropertyEncryptor;
        PropertyDecryptor documentPropertyDecryptor;
#endif
 
    protected:
        friend class Replicator;
        
        operator CBLReplicatorConfiguration() const {
            CBLReplicatorConfiguration conf = {};
            conf.endpoint = _endpoint.ref();
            assert(conf.endpoint);
            conf.replicatorType = replicatorType;
            conf.continuous = continuous;
            conf.disableAutoPurge = !enableAutoPurge;
            conf.maxAttempts = maxAttempts;
            conf.maxAttemptWaitTime = maxAttemptWaitTime;
            conf.heartbeat = heartbeat;
            conf.authenticator = authenticator.ref();
            conf.acceptParentDomainCookies = acceptParentDomainCookies;
            conf.proxy = proxy;
            if (!headers.empty())
                conf.headers = headers;
        #ifdef __CBL_REPLICATOR_NETWORK_INTERFACE__
            if (!networkInterface.empty())
                conf.networkInterface = slice(networkInterface);
        #endif
            if (!pinnedServerCertificate.empty())
                conf.pinnedServerCertificate = slice(pinnedServerCertificate);
            if (!trustedRootCertificates.empty())
                conf.trustedRootCertificates = slice(trustedRootCertificates);
#ifdef COUCHBASE_ENTERPRISE
            conf.acceptOnlySelfSignedServerCertificate = acceptOnlySelfSignedServerCertificate;
#endif
            return conf;
        }
        
    private:
        Endpoint _endpoint;
        std::vector<CollectionConfiguration> _collections;
    };

    class Replicator : private RefCounted {
    public:
        Replicator(const ReplicatorConfiguration& config)
        {
            // Get the current configured collections and populate one for the
            // default collection if the config is configured with the database:
            auto collections = config.collections();
 
            // Create a shared context. Its pointer is passed as the C-level context so
            // that all captureless C callbacks (filters, conflict resolver, encryptor,
            // decryptor) can reach both the collection map and the crypto functions.
            _context = std::make_shared<ReplicatorContext>();
 
            // Get base C config:
            CBLReplicatorConfiguration c_config = config;
 
            // Construct C replication collections to set to the c_config:
            std::vector<CBLCollectionConfiguration> colConfigs;
            for (int i = 0; i < collections.size(); i++) {
                CollectionConfiguration& col = collections[i];
 
                CBLCollectionConfiguration colConfig {};
                colConfig.collection = col.collection().ref();
 
                if (!col.channels.empty()) {
                    colConfig.channels = col.channels;
                }
 
                if (!col.documentIDs.empty()) {
                    colConfig.documentIDs = col.documentIDs;
                }
 
                if (col.pushFilter) {
                    colConfig.pushFilter = [](void* context,
                                              CBLDocument* cDoc,
                                              CBLDocumentFlags flags) -> bool {
                        auto doc = Document(cDoc);
                        auto ctx = (ReplicatorContext*)context;
                        return ctx->collectionMap.find(doc.collection())->second.pushFilter(doc, flags);
                    };
                }
 
                if (col.pullFilter) {
                    colConfig.pullFilter = [](void* context,
                                              CBLDocument* cDoc,
                                              CBLDocumentFlags flags) -> bool {
                        auto doc = Document(cDoc);
                        auto ctx = (ReplicatorContext*)context;
                        return ctx->collectionMap.find(doc.collection())->second.pullFilter(doc, flags);
                    };
                }
 
                if (col.conflictResolver) {
                    colConfig.conflictResolver = [](void* context,
                                                    FLString docID,
                                                    const CBLDocument* cLocalDoc,
                                                    const CBLDocument* cRemoteDoc) -> const CBLDocument*
                    {
                        auto localDoc = Document(cLocalDoc);
                        auto remoteDoc = Document(cRemoteDoc);
                        auto collection = localDoc ? localDoc.collection() : remoteDoc.collection();
 
                        auto ctx = (ReplicatorContext*)context;
                        auto resolved = ctx->collectionMap.find(collection)->second.
                            conflictResolver((std::string_view)slice(docID), localDoc, remoteDoc);
 
                        auto ref = resolved.ref();
                        if (ref && ref != cLocalDoc && ref != cRemoteDoc) {
                            CBLDocument_Retain(ref);
                        }
                        return ref;
                    };
                }
                colConfigs.push_back(colConfig);
                _context->collectionMap.insert({col.collection(), col});
            }
 
            c_config.collections = colConfigs.data();
            c_config.collectionCount = colConfigs.size();
            c_config.context = _context.get();
 
#ifdef COUCHBASE_ENTERPRISE
            if (config.documentPropertyEncryptor) {
                _context->encryptor = config.documentPropertyEncryptor;
                c_config.documentPropertyEncryptor = [](void* context,
                                                        FLString scope,
                                                        FLString collection,
                                                        FLString documentID,
                                                        FLDict properties,
                                                        FLString keyPath,
                                                        FLSlice input,
                                                        FLStringResult* algorithm,
                                                        FLStringResult* keyID,
                                                        CBLError* error) -> FLSliceResult {
                    auto ctx = (ReplicatorContext*)context;
                    auto r = ctx->encryptor(scope, collection, documentID, properties, keyPath, input);
                    if (error)     *error     = r.error;
                    if (algorithm) *algorithm = r.algorithm ? FLStringResult(slice(*r.algorithm)) : FLStringResult{};
                    if (keyID)     *keyID     = r.keyID ? FLStringResult(slice(*r.keyID)) : FLStringResult{};
                    return FLSliceResult(r.ciphertext);
                };
            }
 
            if (config.documentPropertyDecryptor) {
                _context->decryptor = config.documentPropertyDecryptor;
                c_config.documentPropertyDecryptor = [](void* context,
                                                        FLString scope,
                                                        FLString collection,
                                                        FLString documentID,
                                                        FLDict properties,
                                                        FLString keyPath,
                                                        FLSlice input,
                                                        FLString algorithm,
                                                        FLString keyID,
                                                        CBLError* error) -> FLSliceResult {
                    auto ctx = (ReplicatorContext*)context;
                    auto toOpt = [](FLString s) -> std::optional<std::string_view> {
                        return s.buf ? std::optional<std::string_view>{{(const char*)s.buf, s.size}} : std::nullopt;
                    };
                    auto r = ctx->decryptor(scope, collection, documentID, properties, keyPath, input,
                                            toOpt(algorithm), toOpt(keyID));
                    if (error) *error = r.error;
                    return FLSliceResult(r.plaintext);
                };
            }
#endif
 
            CBLError error {};
            _ref = (CBLRefCounted*) CBLReplicator_Create(&c_config, &error);
            internal::check(_ref, error);
        }

        void start(bool resetCheckpoint =false) {CBLReplicator_Start(ref(), resetCheckpoint);}
        
        void stop()                         {CBLReplicator_Stop(ref());}

        void setHostReachable(bool r)       {CBLReplicator_SetHostReachable(ref(), r);}
        
        void setSuspended(bool s)           {CBLReplicator_SetSuspended(ref(), s);}

        CBLReplicatorStatus status() const  {return CBLReplicator_Status(ref());}

        std::string correlationID() const  {return internal::asString(CBLReplicator_CorrelationID(ref()));}

        fleece::Dict pendingDocumentIDs(Collection& collection) const {
            CBLError error;
            fleece::Dict result = CBLReplicator_PendingDocumentIDs(ref(), collection.ref(), &error);
            internal::check(result != nullptr, error);
            return result;
        }
        
        bool isDocumentPending(std::string_view docID, Collection& collection) const {
            CBLError error;
            bool pending = CBLReplicator_IsDocumentPending(ref(), slice(docID), collection.ref(), &error);
            internal::check(pending || error.code == 0, error);
            return pending;
        }
        
        using ChangeListener = cbl::ListenerToken<Replicator, const CBLReplicatorStatus&>;
        
        [[nodiscard]] ChangeListener addChangeListener(ChangeListener::Callback callback) {
            auto l = ChangeListener(callback);
            l.setToken( CBLReplicator_AddChangeListener(ref(), &_callChangeListener, l.context()) );
            return l;
        }
        
        using DocumentReplicationListener = cbl::ListenerToken<Replicator, bool,
            const std::vector<CBLReplicatedDocument>>;

        [[nodiscard]] DocumentReplicationListener addDocumentReplicationListener(DocumentReplicationListener::Callback callback) {
            auto l = DocumentReplicationListener(callback);
            l.setToken( CBLReplicator_AddDocumentReplicationListener(ref(), &_callDocListener, l.context()) );
            return l;
        }
        
    private:
        static void _callChangeListener(void* _cbl_nullable context,
                                        CBLReplicator *repl,
                                        const CBLReplicatorStatus *status)
        {
            ChangeListener::call(context, Replicator(repl), *status);
        }
 
        static void _callDocListener(void* _cbl_nullable context,
                                     CBLReplicator *repl,
                                     bool isPush,
                                     unsigned numDocuments,
                                     const CBLReplicatedDocument* documents)
        {
            std::vector<CBLReplicatedDocument> docs(&documents[0], &documents[numDocuments]);
            DocumentReplicationListener::call(context, Replicator(repl), isPush, docs);
        }
        
        using CollectionToReplCollectionMap = std::unordered_map<Collection, CollectionConfiguration>;
 
        struct ReplicatorContext {
            CollectionToReplCollectionMap collectionMap;
#ifdef COUCHBASE_ENTERPRISE
            PropertyEncryptor encryptor;
            PropertyDecryptor decryptor;
#endif
        };
        std::shared_ptr<ReplicatorContext> _context;
        
        CBL_REFCOUNTED_WITHOUT_COPY_MOVE_BOILERPLATE(Replicator, RefCounted, CBLReplicator)
        
    public:
        Replicator(const Replicator &other) noexcept
        :RefCounted(other)
        ,_context(other._context)
        { }

        Replicator(Replicator &&other) noexcept
        :RefCounted((RefCounted&&)other)
        ,_context(std::move(other._context))
        { }

        Replicator& operator=(const Replicator &other) noexcept {
            RefCounted::operator=(other);
            _context = other._context;
            return *this;
        }

        Replicator& operator=(Replicator &&other) noexcept {
            RefCounted::operator=((RefCounted&&)other);
            _context = std::move(other._context);
            return *this;
        }
        
        void clear() {
            RefCounted::clear();
            _context.reset();
        }
    };
}
 
CBL_ASSUME_NONNULL_END