File size: 4,517 Bytes
52a9af3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 | use crate::backend::BackendBundleClient;
use crate::backend::BundleClient;
use crate::service::CLOUD_CONFIG_BUNDLE_TIMEOUT;
use crate::service::CloudConfigBundleService;
use codex_config::CloudConfigBundleLoader;
use codex_http_client::HttpClientFactory;
use codex_login::AuthConfig;
use codex_login::AuthManager;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::Mutex;
use std::sync::OnceLock;
use tokio::task::AbortHandle;
use tokio::task::JoinHandle;
fn refresher_task_slot() -> &'static Mutex<Option<AbortHandle>> {
static REFRESHER_TASK: OnceLock<Mutex<Option<AbortHandle>>> = OnceLock::new();
REFRESHER_TASK.get_or_init(|| Mutex::new(None))
}
pub(crate) fn replace_refresh_task(slot: &Mutex<Option<AbortHandle>>, next: AbortHandle) {
let mut guard = slot.lock().unwrap_or_else(|err| {
tracing::warn!("cloud config bundle refresher task slot was poisoned");
err.into_inner()
});
if let Some(previous) = guard.replace(next) {
previous.abort();
}
}
struct CloudConfigBundleLoaderLifetime<C> {
service: Arc<CloudConfigBundleService<C>>,
refresh_task: JoinHandle<()>,
}
impl<C> Drop for CloudConfigBundleLoaderLifetime<C> {
fn drop(&mut self) {
self.refresh_task.abort();
}
}
pub fn cloud_config_bundle_loader(
auth_manager: Arc<AuthManager>,
chatgpt_base_url: String,
codex_home: PathBuf,
http_client_factory: HttpClientFactory,
) -> CloudConfigBundleLoader {
let service = CloudConfigBundleService::new(
auth_manager,
Arc::new(BackendBundleClient::new(
chatgpt_base_url,
http_client_factory,
)),
codex_home,
CLOUD_CONFIG_BUNDLE_TIMEOUT,
);
let (loader, refresh_task) = cloud_config_bundle_loader_for_service(service);
replace_refresh_task(refresher_task_slot(), refresh_task);
loader
}
pub(crate) fn cloud_config_bundle_loader_for_service<C>(
service: CloudConfigBundleService<C>,
) -> (CloudConfigBundleLoader, AbortHandle)
where
C: BundleClient + 'static,
{
let service = Arc::new(service);
let background_service = Arc::clone(&service);
let refresh_task = tokio::spawn(async move {
let _ = background_service.get_latest().await;
background_service.refresh_cache_in_background().await;
});
let abort_handle = refresh_task.abort_handle();
let lifetime = Arc::new(CloudConfigBundleLoaderLifetime {
service,
refresh_task,
});
let loader = CloudConfigBundleLoader::from_getter(move || {
let lifetime = Arc::clone(&lifetime);
async move { lifetime.service.get_latest().await }
});
(loader, abort_handle)
}
pub async fn cloud_config_bundle_loader_for_storage(
auth_config: AuthConfig,
enable_codex_api_key_env: bool,
) -> std::io::Result<CloudConfigBundleLoader> {
let service =
cloud_config_bundle_service_for_storage(auth_config, enable_codex_api_key_env).await?;
let (loader, refresh_task) = cloud_config_bundle_loader_for_service(service);
replace_refresh_task(refresher_task_slot(), refresh_task);
Ok(loader)
}
/// Fetches directly from the network on each load, without reading or writing
/// the disk cache or starting a background refresher.
pub async fn cloud_config_bundle_loader_for_storage_without_cache(
auth_config: AuthConfig,
enable_codex_api_key_env: bool,
) -> std::io::Result<CloudConfigBundleLoader> {
let service = Arc::new(
cloud_config_bundle_service_for_storage(auth_config, enable_codex_api_key_env)
.await?
.without_cache(),
);
Ok(CloudConfigBundleLoader::from_getter(move || {
let service = Arc::clone(&service);
async move { service.load_startup_bundle_with_timeout().await }
}))
}
async fn cloud_config_bundle_service_for_storage(
auth_config: AuthConfig,
enable_codex_api_key_env: bool,
) -> std::io::Result<CloudConfigBundleService<BackendBundleClient>> {
let auth_manager =
AuthManager::shared_from_auth_config(auth_config.clone(), enable_codex_api_key_env).await?;
Ok(CloudConfigBundleService::new(
auth_manager,
Arc::new(BackendBundleClient::new(
auth_config
.chatgpt_base_url
.unwrap_or_else(|| "https://chatgpt.com/backend-api/".to_string()),
auth_config.auth_route_config.http_client_factory().clone(),
)),
auth_config.codex_home,
CLOUD_CONFIG_BUNDLE_TIMEOUT,
))
}
|