Parital completed cloud cache token provider interface.

7 files changed, 197 insertions, 273 deletions

diff --git a/zenserver/auth/authmgr.cpp b/zenserver/auth/authmgr.cpp
index 28e128fc0..4d19316dd 100644
--- a/zenserver/auth/authmgr.cpp
+++ b/zenserver/auth/authmgr.cpp
@@ -124,7 +124,7 @@ public:
 		{
 			const OpenIdToken& Token = It->second;
 
-			return {.AccessToken = Token.AccessToken};
+			return {.AccessToken = fmt::format("Bearer {}", Token.AccessToken)};
 		}
 
 		return {};
diff --git a/zenserver/compute/apply.cpp b/zenserver/compute/apply.cpp
index fe2889c7f..15d3acacb 100644
--- a/zenserver/compute/apply.cpp
+++ b/zenserver/compute/apply.cpp
@@ -337,10 +337,12 @@ HttpFunctionService::HttpFunctionService(CasStore& Store, CidStore& InCidStore,
 {
 	m_UpstreamApply = MakeUpstreamApply({}, m_CasStore, m_CidStore);
 
+	CloudCacheAccessToken AccessToken{.Value = "ServiceAccount 0f8056b30bd0df0959be55fc3338159b6f938456d3474aed0087fb965268d079"};
+
 	CloudCacheClientOptions Options = {.ServiceUrl		   = "https://horde.devtools-dev.epicgames.com"sv,
 									   .DdcNamespace	   = "default"sv,
 									   .BlobStoreNamespace = "default"sv,
-									   .AccessToken = "ServiceAccount 0f8056b30bd0df0959be55fc3338159b6f938456d3474aed0087fb965268d079"sv};
+									   .TokenProvider	   = CloudCacheTokenProvider::MakeFromStaticToken(AccessToken)};
 
 	auto HordeUpstreamEndpoint = MakeHordeUpstreamEndpoint(Options, m_CasStore, m_CidStore);
 	m_UpstreamApply->RegisterEndpoint(std::move(HordeUpstreamEndpoint));
diff --git a/zenserver/upstream/jupiter.cpp b/zenserver/upstream/jupiter.cpp
index c952d8e10..86c7e1611 100644
--- a/zenserver/upstream/jupiter.cpp
+++ b/zenserver/upstream/jupiter.cpp
@@ -31,37 +31,26 @@ namespace zen {
 namespace detail {
 	struct CloudCacheSessionState
 	{
-		CloudCacheSessionState(CloudCacheClient& Client) : OwnerClient(Client) {}
-		~CloudCacheSessionState() {}
+		CloudCacheSessionState(CloudCacheClient& Client) : m_Client(Client) {}
 
-		const CloudCacheAccessToken& GetAccessToken()
-		{
-			if (!AccessToken.IsValid())
-			{
-				AccessToken = OwnerClient.AcquireAccessToken();
-			}
-			return AccessToken;
-		}
+		const CloudCacheAccessToken& GetAccessToken() { return m_AccessToken; }
 
-		void InvalidateAccessToken() { AccessToken = {}; }
+		cpr::Session& GetSession() { return m_Session; }
 
 		void Reset(std::chrono::milliseconds ConnectTimeout, std::chrono::milliseconds Timeout)
 		{
-			Session.SetBody({});
-			Session.SetHeader({});
-			Session.SetConnectTimeout(ConnectTimeout);
-			Session.SetTimeout(Timeout);
-			AccessToken = GetAccessToken();
+			m_Session.SetBody({});
+			m_Session.SetHeader({});
+			m_Session.SetConnectTimeout(ConnectTimeout);
+			m_Session.SetTimeout(Timeout);
 		}
 
-		cpr::Session& GetSession() { return Session; }
-
 	private:
 		friend class zen::CloudCacheClient;
 
-		CloudCacheClient&	  OwnerClient;
-		CloudCacheAccessToken AccessToken;
-		cpr::Session		  Session;
+		CloudCacheClient&	  m_Client;
+		CloudCacheAccessToken m_AccessToken;
+		cpr::Session		  m_Session;
 	};
 
 }  // namespace detail
@@ -88,16 +77,11 @@ CloudCacheSession::GetDerivedData(std::string_view BucketId, std::string_view Ke
 {
 	ZEN_TRACE_CPU("HordeClient::GetDerivedData");
 
-	const CloudCacheAccessToken& AccessToken = GetAccessToken();
-	if (!AccessToken.IsValid())
-	{
-		return {.ErrorCode = 401, .Reason = std::string("Invalid access token")};
-	}
-
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/c/ddc/" << m_CacheClient->DdcNamespace() << "/" << BucketId << "/" << Key;
 
-	cpr::Session& Session = m_SessionState->GetSession();
+	cpr::Session&				 Session	 = GetSession();
+	const CloudCacheAccessToken& AccessToken = GetAccessToken();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Accept", "application/octet-stream"}});
@@ -129,19 +113,14 @@ CloudCacheSession::GetDerivedData(std::string_view BucketId, const IoHash& Key)
 CloudCacheResult
 CloudCacheSession::GetRef(std::string_view BucketId, const IoHash& Key, ZenContentType RefType)
 {
-	const CloudCacheAccessToken& AccessToken = GetAccessToken();
-	if (!AccessToken.IsValid())
-	{
-		return {.ErrorCode = 401, .Reason = std::string("Invalid access token")};
-	}
-
 	const std::string ContentType = RefType == ZenContentType::kCbObject ? "application/x-ue-cb" : "application/octet-stream";
 
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/refs/" << m_CacheClient->BlobStoreNamespace() << "/" << BucketId << "/"
 		<< Key.ToHexString();
 
-	cpr::Session& Session = m_SessionState->GetSession();
+	cpr::Session&				 Session	 = GetSession();
+	const CloudCacheAccessToken& AccessToken = GetAccessToken();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Accept", ContentType}});
@@ -168,16 +147,11 @@ CloudCacheSession::GetRef(std::string_view BucketId, const IoHash& Key, ZenConte
 CloudCacheResult
 CloudCacheSession::GetBlob(const IoHash& Key)
 {
-	const CloudCacheAccessToken& AccessToken = GetAccessToken();
-	if (!AccessToken.IsValid())
-	{
-		return {.ErrorCode = 401, .Reason = std::string("Invalid access token")};
-	}
-
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/blobs/" << m_CacheClient->BlobStoreNamespace() << "/" << Key.ToHexString();
 
-	cpr::Session& Session = m_SessionState->GetSession();
+	cpr::Session&				 Session	 = GetSession();
+	const CloudCacheAccessToken& AccessToken = GetAccessToken();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Accept", "application/octet-stream"}});
@@ -207,16 +181,11 @@ CloudCacheSession::GetCompressedBlob(const IoHash& Key)
 {
 	ZEN_TRACE_CPU("HordeClient::GetCompressedBlob");
 
-	const CloudCacheAccessToken& AccessToken = GetAccessToken();
-	if (!AccessToken.IsValid())
-	{
-		return {.ErrorCode = 401, .Reason = std::string("Invalid access token")};
-	}
-
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/compressed-blobs/" << m_CacheClient->BlobStoreNamespace() << "/" << Key.ToHexString();
 
-	cpr::Session& Session = m_SessionState->GetSession();
+	cpr::Session&				 Session	 = GetSession();
+	const CloudCacheAccessToken& AccessToken = GetAccessToken();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Accept", "application/x-ue-comp"}});
@@ -245,16 +214,11 @@ CloudCacheSession::GetObject(const IoHash& Key)
 {
 	ZEN_TRACE_CPU("HordeClient::GetObject");
 
-	const CloudCacheAccessToken& AccessToken = GetAccessToken();
-	if (!AccessToken.IsValid())
-	{
-		return {.ErrorCode = 401, .Reason = std::string("Invalid access token")};
-	}
-
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/objects/" << m_CacheClient->BlobStoreNamespace() << "/" << Key.ToHexString();
 
-	cpr::Session& Session = m_SessionState->GetSession();
+	cpr::Session&				 Session	 = GetSession();
+	const CloudCacheAccessToken& AccessToken = GetAccessToken();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Accept", "application/x-ue-cb"}});
@@ -283,18 +247,13 @@ CloudCacheSession::PutDerivedData(std::string_view BucketId, std::string_view Ke
 {
 	ZEN_TRACE_CPU("HordeClient::PutDerivedData");
 
-	const CloudCacheAccessToken& AccessToken = GetAccessToken();
-	if (!AccessToken.IsValid())
-	{
-		return {.ErrorCode = 401, .Reason = std::string("Invalid access token")};
-	}
-
 	IoHash Hash = IoHash::HashBuffer(DerivedData.Data(), DerivedData.Size());
 
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/c/ddc/" << m_CacheClient->DdcNamespace() << "/" << BucketId << "/" << Key;
 
-	auto& Session = m_SessionState->GetSession();
+	cpr::Session&				 Session	 = GetSession();
+	const CloudCacheAccessToken& AccessToken = GetAccessToken();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value},
@@ -330,15 +289,6 @@ CloudCacheSession::PutRef(std::string_view BucketId, const IoHash& Key, IoBuffer
 {
 	ZEN_TRACE_CPU("HordeClient::PutRef");
 
-	const CloudCacheAccessToken& AccessToken = GetAccessToken();
-	if (!AccessToken.IsValid())
-	{
-		PutRefResult Result;
-		Result.ErrorCode = 401;
-		Result.Reason	 = "Invalid access token"sv;
-		return Result;
-	}
-
 	IoHash Hash = IoHash::HashBuffer(Ref.Data(), Ref.Size());
 
 	const std::string ContentType = RefType == ZenContentType::kCbObject ? "application/x-ue-cb" : "application/octet-stream";
@@ -347,7 +297,8 @@ CloudCacheSession::PutRef(std::string_view BucketId, const IoHash& Key, IoBuffer
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/refs/" << m_CacheClient->BlobStoreNamespace() << "/" << BucketId << "/"
 		<< Key.ToHexString();
 
-	cpr::Session& Session = m_SessionState->GetSession();
+	cpr::Session&				 Session	 = GetSession();
+	const CloudCacheAccessToken& AccessToken = GetAccessToken();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(
@@ -399,20 +350,12 @@ CloudCacheSession::FinalizeRef(std::string_view BucketId, const IoHash& Key, con
 {
 	ZEN_TRACE_CPU("HordeClient::FinalizeRef");
 
-	const CloudCacheAccessToken& AccessToken = GetAccessToken();
-	if (!AccessToken.IsValid())
-	{
-		FinalizeRefResult Result;
-		Result.ErrorCode = 401;
-		Result.Reason	 = "Invalid access token"sv;
-		return Result;
-	}
-
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/refs/" << m_CacheClient->BlobStoreNamespace() << "/" << BucketId << "/"
 		<< Key.ToHexString() << "/finalize/" << RefHash.ToHexString();
 
-	cpr::Session& Session = m_SessionState->GetSession();
+	cpr::Session&				 Session	 = GetSession();
+	const CloudCacheAccessToken& AccessToken = GetAccessToken();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value},
@@ -465,16 +408,11 @@ CloudCacheSession::PutBlob(const IoHash& Key, IoBuffer Blob)
 {
 	ZEN_TRACE_CPU("HordeClient::PutBlob");
 
-	const CloudCacheAccessToken& AccessToken = GetAccessToken();
-	if (!AccessToken.IsValid())
-	{
-		return {.ErrorCode = 401, .Reason = std::string("Invalid access token")};
-	}
-
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/blobs/" << m_CacheClient->BlobStoreNamespace() << "/" << Key.ToHexString();
 
-	cpr::Session& Session = m_SessionState->GetSession();
+	cpr::Session&				 Session	 = GetSession();
+	const CloudCacheAccessToken& AccessToken = GetAccessToken();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Content-Type", "application/octet-stream"}});
@@ -502,16 +440,11 @@ CloudCacheSession::PutCompressedBlob(const IoHash& Key, IoBuffer Blob)
 {
 	ZEN_TRACE_CPU("HordeClient::PutCompressedBlob");
 
-	const CloudCacheAccessToken& AccessToken = GetAccessToken();
-	if (!AccessToken.IsValid())
-	{
-		return {.ErrorCode = 401, .Reason = std::string("Invalid access token")};
-	}
-
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/compressed-blobs/" << m_CacheClient->BlobStoreNamespace() << "/" << Key.ToHexString();
 
-	cpr::Session& Session = m_SessionState->GetSession();
+	cpr::Session&				 Session	 = GetSession();
+	const CloudCacheAccessToken& AccessToken = GetAccessToken();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Content-Type", "application/x-ue-comp"}});
@@ -539,16 +472,11 @@ CloudCacheSession::PutObject(const IoHash& Key, IoBuffer Object)
 {
 	ZEN_TRACE_CPU("HordeClient::PutObject");
 
-	const CloudCacheAccessToken& AccessToken = GetAccessToken();
-	if (!AccessToken.IsValid())
-	{
-		return {.ErrorCode = 401, .Reason = std::string("Invalid access token")};
-	}
-
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/objects/" << m_CacheClient->BlobStoreNamespace() << "/" << Key.ToHexString();
 
-	cpr::Session& Session = m_SessionState->GetSession();
+	cpr::Session&				 Session	 = GetSession();
+	const CloudCacheAccessToken& AccessToken = GetAccessToken();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Content-Type", "application/x-ue-cb"}});
@@ -576,17 +504,12 @@ CloudCacheSession::RefExists(std::string_view BucketId, const IoHash& Key)
 {
 	ZEN_TRACE_CPU("HordeClient::RefExists");
 
-	const CloudCacheAccessToken& AccessToken = GetAccessToken();
-	if (!AccessToken.IsValid())
-	{
-		return {.ErrorCode = 401, .Reason = std::string("Invalid access token")};
-	}
-
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/refs/" << m_CacheClient->BlobStoreNamespace() << "/" << BucketId << "/"
 		<< Key.ToHexString();
 
-	cpr::Session& Session = m_SessionState->GetSession();
+	cpr::Session&				 Session	 = GetSession();
+	const CloudCacheAccessToken& AccessToken = GetAccessToken();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}});
@@ -648,16 +571,11 @@ CloudCacheSession::PostComputeTasks(std::string_view ChannelId, IoBuffer TasksDa
 {
 	ZEN_TRACE_CPU("HordeClient::PostComputeTasks");
 
-	const CloudCacheAccessToken& AccessToken = GetAccessToken();
-	if (!AccessToken.IsValid())
-	{
-		return {.ErrorCode = 401, .Reason = std::string("Invalid access token")};
-	}
-
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/compute/" << ChannelId;
 
-	auto& Session = m_SessionState->GetSession();
+	cpr::Session&				 Session	 = GetSession();
+	const CloudCacheAccessToken& AccessToken = GetAccessToken();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Content-Type", "application/x-ue-cb"}});
@@ -681,16 +599,11 @@ CloudCacheSession::PostComputeTasks(std::string_view ChannelId, IoBuffer TasksDa
 CloudCacheResult
 CloudCacheSession::GetComputeUpdates(std::string_view ChannelId, const uint32_t WaitSeconds)
 {
-	const CloudCacheAccessToken& AccessToken = GetAccessToken();
-	if (!AccessToken.IsValid())
-	{
-		return {.ErrorCode = 401, .Reason = std::string("Invalid access token")};
-	}
-
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/compute/" << ChannelId << "/updates?wait=" << WaitSeconds;
 
-	auto& Session = m_SessionState->GetSession();
+	cpr::Session&				 Session	 = GetSession();
+	const CloudCacheAccessToken& AccessToken = GetAccessToken();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Accept", "application/x-ue-cb"}});
@@ -719,16 +632,11 @@ CloudCacheSession::GetObjectTree(const IoHash& Key)
 {
 	ZEN_TRACE_CPU("HordeClient::GetObjectTree");
 
-	const CloudCacheAccessToken& AccessToken = GetAccessToken();
-	if (!AccessToken.IsValid())
-	{
-		return {.ErrorCode = 401, .Reason = std::string("Invalid access token")};
-	}
-
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/objects/" << m_CacheClient->BlobStoreNamespace() << "/" << Key.ToHexString() << "/tree";
 
-	cpr::Session& Session = m_SessionState->GetSession();
+	cpr::Session&				 Session	 = GetSession();
+	const CloudCacheAccessToken& AccessToken = GetAccessToken();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Accept", "application/octet-stream"}});
@@ -764,7 +672,13 @@ CloudCacheSession::Filter(std::string_view BucketId, const std::vector<IoHash>&
 	return {};
 }
 
-const CloudCacheAccessToken&
+cpr::Session&
+CloudCacheSession::GetSession()
+{
+	return m_SessionState->GetSession();
+}
+
+CloudCacheAccessToken
 CloudCacheSession::GetAccessToken()
 {
 	return m_SessionState->GetAccessToken();
@@ -773,27 +687,17 @@ CloudCacheSession::GetAccessToken()
 bool
 CloudCacheSession::VerifyAccessToken(long StatusCode)
 {
-	if (StatusCode == 401)
-	{
-		m_SessionState->InvalidateAccessToken();
-		return false;
-	}
-	return true;
+	return StatusCode != 401;
 }
 
 CloudCacheResult
 CloudCacheSession::CacheTypeExists(std::string_view TypeId, const IoHash& Key)
 {
-	const CloudCacheAccessToken& AccessToken = GetAccessToken();
-	if (!AccessToken.IsValid())
-	{
-		return {.ErrorCode = 401, .Reason = std::string("Invalid access token")};
-	}
-
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/" << TypeId << "/" << m_CacheClient->BlobStoreNamespace() << "/" << Key.ToHexString();
 
-	cpr::Session& Session = m_SessionState->GetSession();
+	cpr::Session&				 Session	 = GetSession();
+	const CloudCacheAccessToken& AccessToken = GetAccessToken();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}});
@@ -817,12 +721,6 @@ CloudCacheSession::CacheTypeExists(std::string_view TypeId, const IoHash& Key)
 CloudCacheExistsResult
 CloudCacheSession::CacheTypeExists(std::string_view TypeId, const std::set<IoHash>& Keys)
 {
-	const CloudCacheAccessToken& AccessToken = GetAccessToken();
-	if (!AccessToken.IsValid())
-	{
-		return {CloudCacheResult{.ErrorCode = 401, .Reason = std::string("Invalid access token")}};
-	}
-
 	ExtendableStringBuilder<256> Query;
 	for (const auto& Key : Keys)
 	{
@@ -832,7 +730,8 @@ CloudCacheSession::CacheTypeExists(std::string_view TypeId, const std::set<IoHas
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/" << TypeId << "/" << m_CacheClient->BlobStoreNamespace() << "/exists?" << Query;
 
-	cpr::Session& Session = m_SessionState->GetSession();
+	cpr::Session&				 Session	 = GetSession();
+	const CloudCacheAccessToken& AccessToken = GetAccessToken();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Accept", "application/x-ue-cb"}});
@@ -869,65 +768,90 @@ CloudCacheSession::CacheTypeExists(std::string_view TypeId, const std::set<IoHas
 	return Result;
 }
 
-//////////////////////////////////////////////////////////////////////////
-//
-// ServiceUrl:		https://jupiter.devtools.epicgames.com
-// DdcNamespace:		ue4.ddc
-// OAuthClientId:	0oao91lrhqPiAlaGD0x7
-// OAuthProvider:	https://epicgames.okta.com/oauth2/auso645ojjWVdRI3d0x7/v1/token
-// OAuthSecret:		-GBWjjenhCgOwhxL5yBKNJECVIoDPH0MK4RDuN7d
-//
+/**
+ * An access token provider that holds a token that will never change.
+ */
+class StaticTokenProvider final : public CloudCacheTokenProvider
+{
+public:
+	StaticTokenProvider(CloudCacheAccessToken Token) : m_Token(std::move(Token)) {}
 
-CloudCacheClient::CloudCacheClient(const CloudCacheClientOptions& Options)
-: m_Log(zen::logging::Get("jupiter"))
-, m_ServiceUrl(Options.ServiceUrl)
-, m_OAuthFullUri(Options.OAuthProvider)
-, m_DdcNamespace(Options.DdcNamespace)
-, m_BlobStoreNamespace(Options.BlobStoreNamespace)
-, m_OAuthClientId(Options.OAuthClientId)
-, m_OAuthSecret(Options.OAuthSecret)
-, m_AccessToken(Options.AccessToken)
-, m_ConnectTimeout(Options.ConnectTimeout)
-, m_Timeout(Options.Timeout)
+	virtual ~StaticTokenProvider() = default;
+
+	virtual CloudCacheAccessToken GetAccessToken() final override { return m_Token; }
+
+private:
+	CloudCacheAccessToken m_Token;
+};
+
+std::unique_ptr<CloudCacheTokenProvider>
+CloudCacheTokenProvider::MakeFromStaticToken(CloudCacheAccessToken Token)
 {
-	if (!Options.AccessToken.empty())
+	return std::make_unique<StaticTokenProvider>(std::move(Token));
+}
+
+class OAuthClientCredentialsTokenProvider final : public CloudCacheTokenProvider
+{
+public:
+	OAuthClientCredentialsTokenProvider(const CloudCacheTokenProvider::OAuthClientCredentialsParams& Params)
 	{
-		// If an access token was provided, OAuth settings are not used.
-		return;
+		m_Url		   = std::string(Params.Url);
+		m_ClientId	   = std::string(Params.ClientId);
+		m_ClientSecret = std::string(Params.ClientSecret);
 	}
 
-	if (!Options.OAuthProvider.starts_with("http://"sv) && !Options.OAuthProvider.starts_with("https://"sv))
-	{
-		ZEN_WARN("bad provider specification: '{}' - must be fully qualified", Options.OAuthProvider);
-		m_IsValid = false;
+	virtual ~OAuthClientCredentialsTokenProvider() = default;
 
-		return;
-	}
+	virtual CloudCacheAccessToken GetAccessToken() final override
+	{
+		using namespace std::chrono;
 
-	// Split into host and Uri substrings
+		std::string Body =
+			fmt::format("client_id={}&scope=cache_access&grant_type=client_credentials&client_secret={k}", m_ClientId, m_ClientSecret);
 
-	auto SchemePos = Options.OAuthProvider.find("://"sv);
+		cpr::Response Response =
+			cpr::Post(cpr::Url{m_Url}, cpr::Header{{"Content-Type", "application/x-www-form-urlencoded"}}, cpr::Body{std::move(Body)});
 
-	if (SchemePos == std::string::npos)
-	{
-		ZEN_WARN("Bad service URL passed to cloud cache client: '{}'", Options.ServiceUrl);
-		m_IsValid = false;
+		if (Response.error || Response.status_code != 200)
+		{
+			return {};
+		}
 
-		return;
-	}
+		std::string	 JsonError;
+		json11::Json Json = json11::Json::parse(Response.text, JsonError);
 
-	auto DomainEnd = Options.OAuthProvider.find('/', /* also skip the :// */ SchemePos + 3);
+		if (JsonError.empty() == false)
+		{
+			return {};
+		}
 
-	if (DomainEnd == std::string::npos)
-	{
-		ZEN_WARN("Bad service URL passed to cloud cache client: '{}' no path delimiter found", Options.ServiceUrl);
-		m_IsValid = false;
+		std::string						 Token			  = Json["access_token"].string_value();
+		int64_t							 ExpiresInSeconds = static_cast<int64_t>(Json["expires_in"].int_value());
+		CloudCacheAccessToken::TimePoint ExpireTime		  = CloudCacheAccessToken::Clock::now() + seconds(ExpiresInSeconds);
 
-		return;
+		return {.Value = fmt::format("Bearer {}", Token), .ExpireTime = ExpireTime};
 	}
 
-	m_OAuthDomain  = Options.OAuthProvider.substr(SchemePos + 3, DomainEnd - SchemePos - 3);  // epicgames.okta.com
-	m_OAuthUriPath = Options.OAuthProvider.substr(DomainEnd + 1);							  // oauth2/..../v1/token
+private:
+	std::string m_Url;
+	std::string m_ClientId;
+	std::string m_ClientSecret;
+};
+
+std::unique_ptr<CloudCacheTokenProvider>
+CloudCacheTokenProvider::MakeFromOAuthClientCredentials(const OAuthClientCredentialsParams& Params)
+{
+	return std::make_unique<OAuthClientCredentialsTokenProvider>(Params);
+}
+
+CloudCacheClient::CloudCacheClient(const CloudCacheClientOptions& Options)
+: m_Log(zen::logging::Get("jupiter"))
+, m_ServiceUrl(Options.ServiceUrl)
+, m_DdcNamespace(Options.DdcNamespace)
+, m_BlobStoreNamespace(Options.BlobStoreNamespace)
+, m_ConnectTimeout(Options.ConnectTimeout)
+, m_Timeout(Options.Timeout)
+{
 }
 
 CloudCacheClient::~CloudCacheClient()
@@ -945,36 +869,7 @@ CloudCacheClient::AcquireAccessToken()
 {
 	ZEN_TRACE_CPU("HordeClient::AcquireAccessToken");
 
-	using namespace std::chrono;
-
-	// If an access token was provided, return it instead of querying OAuth
-	if (!m_AccessToken.empty())
-	{
-		return {m_AccessToken, steady_clock::time_point::max()};
-	}
-
-	ExtendableStringBuilder<128> OAuthFormData;
-	OAuthFormData << "client_id=" << m_OAuthClientId << "&scope=cache_access&grant_type=client_credentials&client_secret=" << m_OAuthSecret;
-
-	std::string Body{OAuthFormData};
-
-	cpr::Response Response =
-		cpr::Post(cpr::Url{m_OAuthFullUri}, cpr::Header{{"Content-Type", "application/x-www-form-urlencoded"}}, cpr::Body{Body});
-
-	Body = std::move(Response.text);
-
-	std::string	 JsonError;
-	json11::Json JsonResponse = json11::Json::parse(Body, JsonError);
-	if (!JsonError.empty())
-	{
-		return {};
-	}
-
-	std::string				 AccessToken	  = std::string("Bearer ") + JsonResponse["access_token"].string_value();
-	int64_t					 ExpiresInSeconds = static_cast<int64_t>(JsonResponse["expires_in"].int_value());
-	steady_clock::time_point ExpireTime		  = steady_clock::now() + seconds(ExpiresInSeconds);
-
-	return {std::move(AccessToken), ExpireTime};
+	return m_TokenProvider->GetAccessToken();
 }
 
 detail::CloudCacheSessionState*
@@ -1001,19 +896,8 @@ CloudCacheClient::AllocSessionState()
 void
 CloudCacheClient::FreeSessionState(detail::CloudCacheSessionState* State)
 {
-	const bool IsTokenValid = State->AccessToken.IsValid();
-
 	RwLock::ExclusiveLockScope _(m_SessionStateLock);
 	m_SessionStateCache.push_front(State);
-
-	// Invalidate all cached access tokens if any one fails
-	if (!IsTokenValid)
-	{
-		for (auto& CachedState : m_SessionStateCache)
-		{
-			CachedState->AccessToken = {};
-		}
-	}
 }
 
 }  // namespace zen
diff --git a/zenserver/upstream/jupiter.h b/zenserver/upstream/jupiter.h
index bb1797393..057488bcc 100644
--- a/zenserver/upstream/jupiter.h
+++ b/zenserver/upstream/jupiter.h
@@ -17,6 +17,10 @@
 
 struct ZenCacheValue;
 
+namespace cpr {
+class Session;
+}
+
 namespace zen {
 namespace detail {
 	struct CloudCacheSessionState;
@@ -32,16 +36,18 @@ class CbObjectView;
  */
 struct CloudCacheAccessToken
 {
+	using Clock		= std::chrono::system_clock;
+	using TimePoint = Clock::time_point;
+
 	static constexpr int64_t ExpireMarginInSeconds = 30;
 
-	std::string							  Value;
-	std::chrono::steady_clock::time_point ExpireTime;
+	std::string Value;
+	TimePoint	ExpireTime;
 
 	bool IsValid() const
 	{
-		return !Value.empty() &&
-			   ExpireMarginInSeconds <
-				   std::chrono::duration_cast<std::chrono::seconds>(ExpireTime - std::chrono::steady_clock::now()).count();
+		return Value.empty() == false &&
+			   ExpireMarginInSeconds < std::chrono::duration_cast<std::chrono::seconds>(ExpireTime - Clock::now()).count();
 	}
 };
 
@@ -117,9 +123,10 @@ public:
 	std::vector<IoHash> Filter(std::string_view BucketId, const std::vector<IoHash>& ChunkHashes);
 
 private:
-	inline spdlog::logger&		 Log() { return m_Log; }
-	const CloudCacheAccessToken& GetAccessToken();
-	bool						 VerifyAccessToken(long StatusCode);
+	inline spdlog::logger& Log() { return m_Log; }
+	cpr::Session&		   GetSession();
+	CloudCacheAccessToken  GetAccessToken();
+	bool				   VerifyAccessToken(long StatusCode);
 
 	CloudCacheResult CacheTypeExists(std::string_view TypeId, const IoHash& Key);
 
@@ -130,19 +137,38 @@ private:
 	detail::CloudCacheSessionState* m_SessionState;
 };
 
+/**
+ * Access token provider interface
+ */
+class CloudCacheTokenProvider
+{
+public:
+	virtual ~CloudCacheTokenProvider() = default;
+
+	virtual CloudCacheAccessToken GetAccessToken() = 0;
+
+	static std::unique_ptr<CloudCacheTokenProvider> MakeFromStaticToken(CloudCacheAccessToken Token);
+
+	struct OAuthClientCredentialsParams
+	{
+		std::string_view Url;
+		std::string_view ClientId;
+		std::string_view ClientSecret;
+	};
+
+	static std::unique_ptr<CloudCacheTokenProvider> MakeFromOAuthClientCredentials(const OAuthClientCredentialsParams& Params);
+};
+
 struct CloudCacheClientOptions
 {
-	std::string_view		  Name;
-	std::string_view		  ServiceUrl;
-	std::string_view		  DdcNamespace;
-	std::string_view		  BlobStoreNamespace;
-	std::string_view		  OAuthProvider;
-	std::string_view		  OAuthClientId;
-	std::string_view		  OAuthSecret;
-	std::string_view		  AccessToken;
-	std::chrono::milliseconds ConnectTimeout{5000};
-	std::chrono::milliseconds Timeout{};
-	bool					  UseLegacyDdc = false;
+	std::string_view						 Name;
+	std::string_view						 ServiceUrl;
+	std::string_view						 DdcNamespace;
+	std::string_view						 BlobStoreNamespace;
+	std::chrono::milliseconds				 ConnectTimeout{5000};
+	std::chrono::milliseconds				 Timeout{};
+	std::unique_ptr<CloudCacheTokenProvider> TokenProvider;
+	bool									 UseLegacyDdc = false;
 };
 
 /**
@@ -159,23 +185,19 @@ public:
 	std::string_view	  BlobStoreNamespace() const { return m_BlobStoreNamespace; }
 	std::string_view	  ServiceUrl() const { return m_ServiceUrl; }
 	bool				  IsValid() const { return m_IsValid; }
+	void				  SetAccessToken(CloudCacheAccessToken Token);
 
 	spdlog::logger& Logger() { return m_Log; }
 
 private:
-	spdlog::logger&			  m_Log;
-	std::string				  m_ServiceUrl;
-	std::string				  m_OAuthDomain;
-	std::string				  m_OAuthUriPath;
-	std::string				  m_OAuthFullUri;
-	std::string				  m_DdcNamespace;
-	std::string				  m_BlobStoreNamespace;
-	std::string				  m_OAuthClientId;
-	std::string				  m_OAuthSecret;
-	std::string				  m_AccessToken;
-	std::chrono::milliseconds m_ConnectTimeout{};
-	std::chrono::milliseconds m_Timeout{};
-	bool					  m_IsValid = false;
+	spdlog::logger&							 m_Log;
+	std::unique_ptr<CloudCacheTokenProvider> m_TokenProvider;
+	std::string								 m_ServiceUrl;
+	std::string								 m_DdcNamespace;
+	std::string								 m_BlobStoreNamespace;
+	std::chrono::milliseconds				 m_ConnectTimeout{};
+	std::chrono::milliseconds				 m_Timeout{};
+	bool									 m_IsValid = false;
 
 	RwLock									   m_SessionStateLock;
 	std::list<detail::CloudCacheSessionState*> m_SessionStateCache;
diff --git a/zenserver/upstream/upstreamcache.cpp b/zenserver/upstream/upstreamcache.cpp
index 206787bf7..44a8274da 100644
--- a/zenserver/upstream/upstreamcache.cpp
+++ b/zenserver/upstream/upstreamcache.cpp
@@ -109,6 +109,8 @@ namespace detail {
 					return {.State = UpstreamEndpointState::kOk};
 				}
 
+				const AuthMgr::OpenIdAccessToken Token = m_AuthMgr.GetOpenIdAccessToken("Okta");
+
 				CloudCacheSession	   Session(m_Client);
 				const CloudCacheResult Result = Session.Authenticate();
 
diff --git a/zenserver/upstream/upstreamservice.cpp b/zenserver/upstream/upstreamservice.cpp
index c8176779e..5f248678a 100644
--- a/zenserver/upstream/upstreamservice.cpp
+++ b/zenserver/upstream/upstreamservice.cpp
@@ -172,6 +172,7 @@ HttpUpstreamService::HttpUpstreamService(UpstreamCache& Upstream, AuthMgr& Mgr)
 					}
 				}
 
+				/*
 				const zen::CloudCacheClientOptions Options = {.Name				  = Name,
 															  .ServiceUrl		  = Url,
 															  .DdcNamespace		  = Namespace,
@@ -182,6 +183,7 @@ HttpUpstreamService::HttpUpstreamService(UpstreamCache& Upstream, AuthMgr& Mgr)
 															  .AccessToken		  = OAuthToken};
 
 				Endpoint = zen::MakeJupiterUpstreamEndpoint(Options, m_AuthMgr);
+				*/
 			}
 
 			m_Upstream.RegisterEndpoint(std::move(Endpoint));
diff --git a/zenserver/zenserver.cpp b/zenserver/zenserver.cpp
index bd4a3c1cf..4227a296c 100644
--- a/zenserver/zenserver.cpp
+++ b/zenserver/zenserver.cpp
@@ -742,6 +742,7 @@ ZenServer::InitializeStructuredCache(const ZenServerOptions& ServerOptions)
 
 		// Jupiter upstream
 		{
+			/*
 			zen::CloudCacheClientOptions Options;
 			if (UpstreamConfig.JupiterConfig.UseProductionSettings)
 			{
@@ -788,7 +789,18 @@ ZenServer::InitializeStructuredCache(const ZenServerOptions& ServerOptions)
 												 .Timeout			 = std::chrono::milliseconds(UpstreamConfig.TimeoutMilliseconds),
 												 .UseLegacyDdc		 = false};
 			}
-
+			*/
+
+			const AuthMgr::OpenIdAccessToken Token = m_AuthMgr->GetOpenIdAccessToken("Okta");
+
+			zen::CloudCacheClientOptions Options =
+				zen::CloudCacheClientOptions{.Name				 = "Jupiter-Prod"sv,
+											 .ServiceUrl		 = "https://jupiter.devtools.epicgames.com"sv,
+											 .DdcNamespace		 = "ue.ddc"sv,
+											 .BlobStoreNamespace = "ue.ddc"sv,
+											 .ConnectTimeout	 = std::chrono::milliseconds(UpstreamConfig.ConnectTimeoutMilliseconds),
+											 .Timeout			 = std::chrono::milliseconds(UpstreamConfig.TimeoutMilliseconds),
+											 .UseLegacyDdc		 = false};
 			if (!Options.ServiceUrl.empty())
 			{
 				std::unique_ptr<zen::UpstreamEndpoint> JupiterEndpoint = zen::MakeJupiterUpstreamEndpoint(Options, *m_AuthMgr);