sqlite_provider_call_broker.h

#pragma once
 
#include <neograph/graph/configuration_failure.h>
#include <neograph/program/core_provider_call.h>
#include <neograph/program/result.h>
#include <neograph/types.h>
#include <neograph/provider.h>
#include <core/native_archive.h>
#include <runtime/client.h>
 
#include <cstdint>
#include <exception>
#include <stdexcept>
#include <utility>
#include <memory>
#include <optional>
#include <string>
#include <string_view>
#include <vector>
 
namespace neograph::program {
 
class NEOGRAPH_PROGRAM_API ProgramProviderOutcomePersistenceError final : public ProviderOutcomeError {
public:
    ProgramProviderOutcomePersistenceError(
        sp::runtime::Result outcome, std::exception_ptr cause,
        std::exception_ptr observer_error = {})
        : ProviderOutcomeError("Program provider outcome could not be persisted",
                               std::move(outcome), std::move(cause)),
          observer_error_(std::move(observer_error)) {}
    const std::exception_ptr& observer_error() const noexcept { return observer_error_; }
private:
    std::exception_ptr observer_error_;
};
 
class NEOGRAPH_PROGRAM_API ProgramProviderDispatchUncertainError final : public ProviderOutcomeError {
public:
    explicit ProgramProviderDispatchUncertainError(std::string message,
                                                   std::exception_ptr cause = {})
        : ProviderOutcomeError(message.c_str(), nullptr, std::move(cause)) {}
};
 
class NEOGRAPH_PROGRAM_API ProgramProviderForkResendRefusedError final
    : public std::runtime_error,
      public graph::ConfigurationFailure {
public:
    enum class Reason : std::uint8_t {
        Unverifiable,
        EncoderDiffers,
        DeploymentDiffers,
    };
    ProgramProviderForkResendRefusedError(Reason reason, std::string call_id,
                                          std::string source_run_id, std::string source_call_id);
    Reason             reason() const noexcept { return reason_; }
    const std::string& call_id() const noexcept { return call_id_; }
    const std::string& source_run_id() const noexcept { return source_run_id_; }
    const std::string& source_call_id() const noexcept { return source_call_id_; }
private:
    Reason      reason_;
    std::string call_id_, source_run_id_, source_call_id_;
};
 
class NEOGRAPH_PROGRAM_API ProgramProviderUploadResendRequiredError final
    : public std::runtime_error,
      public graph::ConfigurationFailure {
public:
    ProgramProviderUploadResendRequiredError(std::string call_id, std::string later_call_id);
    const std::string& call_id() const noexcept { return call_id_; }
    const std::string& later_call_id() const noexcept { return later_call_id_; }
private:
    std::string call_id_, later_call_id_;
};
 
class NEOGRAPH_PROGRAM_API SQLiteProgramProviderCallJournal final {
public:
    explicit SQLiteProgramProviderCallJournal(
        std::string database_path, std::shared_ptr<sp::NativeArchive> native_archive = {});
    ~SQLiteProgramProviderCallJournal();
    SQLiteProgramProviderCallJournal(const SQLiteProgramProviderCallJournal&) = delete;
    SQLiteProgramProviderCallJournal& operator=(const SQLiteProgramProviderCallJournal&) = delete;
 
    ProgramCoreProviderCallBinding bind(const ProgramCoreProviderCallContext& context,
                                        std::string provider_binding_identity);
    ProgramCoreProviderCallBinding bind_priced(const ProgramCoreProviderCallContext& context,
                                               std::string provider_binding_identity);
    static std::string logical_call_id(const ProgramCoreProviderCallContext& context,
                                       const graph::ProviderCallIdentity& call);
 
    enum class State { Dispatched, Succeeded, ReconciliationRequired, Failed };
    struct Record {
        State state;
        std::string request_digest;
        std::string provider_binding_identity;
        std::string original_attempt;
        std::string outcome_digest;
        std::string provider_name;
        std::string model;
        std::string family;
        std::string mode;
        std::string original_deadline;
        std::uint32_t encoding_version;
        std::uint64_t claim_amount;
        std::uint64_t committed_amount;
        std::optional<ProgramMonetaryRecord> monetary_record;
        std::uint32_t identity_version = 2;
        std::string content_digest;
        std::string policy_identity;
        std::string producer;
        bool opens_upload_session = false;
        std::uint32_t restarts = 0;
        bool superseded = false;
    };
    std::optional<Record> inspect(std::string_view owner_scope,
                                  std::string_view dispatch_id) const;
 
    struct PolicyChange {
        enum class Via : std::uint8_t {
            Replay,
            Adoption,
        };
        std::string   journaled_policy_identity;
        std::string   observed_policy_identity;
        std::uint64_t attempt = 0;
        Via           via     = Via::Replay;
    };
    std::vector<PolicyChange> policy_changes(std::string_view owner_scope,
                                             std::string_view dispatch_id) const;
 
    struct Restart {
        enum class Reason : std::uint8_t {
            SessionNotStored,
            OutcomeUnknown,
        };
        Reason        reason = Reason::OutcomeUnknown;
        std::uint64_t attempt = 0;
        State         replaced_state = State::Dispatched;
        std::string   replaced_outcome_digest;
    };
    std::vector<Restart> restarts(std::string_view owner_scope,
                                  std::string_view dispatch_id) const;
 
    void reconcile_success(std::string_view owner_scope, std::string_view dispatch_id,
                           std::string evidence_identity, sp::runtime::Result completion);
 
    struct Impl;
private:
    std::shared_ptr<Impl> impl_;
};
 
}  // namespace neograph::program