1use crate::{
2 db::{ExtensionMetadata, NewExtensionVersion},
3 executor::Executor,
4 AppState, Error, Result,
5};
6use anyhow::{anyhow, Context as _};
7use aws_sdk_s3::presigning::PresigningConfig;
8use axum::{
9 extract::{Path, Query},
10 http::StatusCode,
11 response::Redirect,
12 routing::get,
13 Extension, Json, Router,
14};
15use collections::HashMap;
16use serde::{Deserialize, Serialize};
17use std::{sync::Arc, time::Duration};
18use time::PrimitiveDateTime;
19use util::ResultExt;
20
21pub fn router() -> Router {
22 Router::new()
23 .route("/extensions", get(get_extensions))
24 .route(
25 "/extensions/:extension_id/download",
26 get(download_latest_extension),
27 )
28 .route(
29 "/extensions/:extension_id/:version/download",
30 get(download_extension),
31 )
32}
33
34#[derive(Debug, Deserialize)]
35struct GetExtensionsParams {
36 filter: Option<String>,
37}
38
39#[derive(Debug, Deserialize)]
40struct DownloadLatestExtensionParams {
41 extension_id: String,
42}
43
44#[derive(Debug, Deserialize)]
45struct DownloadExtensionParams {
46 extension_id: String,
47 version: String,
48}
49
50#[derive(Debug, Serialize)]
51struct GetExtensionsResponse {
52 pub data: Vec<ExtensionMetadata>,
53}
54
55#[derive(Deserialize)]
56struct ExtensionManifest {
57 name: String,
58 version: String,
59 description: Option<String>,
60 authors: Vec<String>,
61 repository: String,
62}
63
64async fn get_extensions(
65 Extension(app): Extension<Arc<AppState>>,
66 Query(params): Query<GetExtensionsParams>,
67) -> Result<Json<GetExtensionsResponse>> {
68 let extensions = app.db.get_extensions(params.filter.as_deref(), 500).await?;
69 Ok(Json(GetExtensionsResponse { data: extensions }))
70}
71
72async fn download_latest_extension(
73 Extension(app): Extension<Arc<AppState>>,
74 Path(params): Path<DownloadLatestExtensionParams>,
75) -> Result<Redirect> {
76 let extension = app
77 .db
78 .get_extension(¶ms.extension_id)
79 .await?
80 .ok_or_else(|| anyhow!("unknown extension"))?;
81 download_extension(
82 Extension(app),
83 Path(DownloadExtensionParams {
84 extension_id: params.extension_id,
85 version: extension.version,
86 }),
87 )
88 .await
89}
90
91async fn download_extension(
92 Extension(app): Extension<Arc<AppState>>,
93 Path(params): Path<DownloadExtensionParams>,
94) -> Result<Redirect> {
95 let Some((blob_store_client, bucket)) = app
96 .blob_store_client
97 .clone()
98 .zip(app.config.blob_store_bucket.clone())
99 else {
100 Err(Error::Http(
101 StatusCode::NOT_IMPLEMENTED,
102 "not supported".into(),
103 ))?
104 };
105
106 let DownloadExtensionParams {
107 extension_id,
108 version,
109 } = params;
110
111 let version_exists = app
112 .db
113 .record_extension_download(&extension_id, &version)
114 .await?;
115
116 if !version_exists {
117 Err(Error::Http(
118 StatusCode::NOT_FOUND,
119 "unknown extension version".into(),
120 ))?;
121 }
122
123 let url = blob_store_client
124 .get_object()
125 .bucket(bucket)
126 .key(format!(
127 "extensions/{extension_id}/{version}/archive.tar.gz"
128 ))
129 .presigned(PresigningConfig::expires_in(EXTENSION_DOWNLOAD_URL_LIFETIME).unwrap())
130 .await
131 .map_err(|e| anyhow!("failed to create presigned extension download url {e}"))?;
132
133 Ok(Redirect::temporary(url.uri()))
134}
135
136const EXTENSION_FETCH_INTERVAL: Duration = Duration::from_secs(5 * 60);
137const EXTENSION_DOWNLOAD_URL_LIFETIME: Duration = Duration::from_secs(3 * 60);
138
139pub fn fetch_extensions_from_blob_store_periodically(app_state: Arc<AppState>, executor: Executor) {
140 let Some(blob_store_client) = app_state.blob_store_client.clone() else {
141 log::info!("no blob store client");
142 return;
143 };
144 let Some(blob_store_bucket) = app_state.config.blob_store_bucket.clone() else {
145 log::info!("no blob store bucket");
146 return;
147 };
148
149 executor.spawn_detached({
150 let executor = executor.clone();
151 async move {
152 loop {
153 fetch_extensions_from_blob_store(
154 &blob_store_client,
155 &blob_store_bucket,
156 &app_state,
157 )
158 .await
159 .log_err();
160 executor.sleep(EXTENSION_FETCH_INTERVAL).await;
161 }
162 }
163 });
164}
165
166async fn fetch_extensions_from_blob_store(
167 blob_store_client: &aws_sdk_s3::Client,
168 blob_store_bucket: &String,
169 app_state: &Arc<AppState>,
170) -> anyhow::Result<()> {
171 log::info!("fetching extensions from blob store");
172
173 let list = blob_store_client
174 .list_objects()
175 .bucket(blob_store_bucket)
176 .prefix("extensions/")
177 .send()
178 .await?;
179
180 let objects = list.contents.unwrap_or_default();
181
182 let mut published_versions = HashMap::<&str, Vec<&str>>::default();
183 for object in &objects {
184 let Some(key) = object.key.as_ref() else {
185 continue;
186 };
187 let mut parts = key.split('/');
188 let Some(_) = parts.next().filter(|part| *part == "extensions") else {
189 continue;
190 };
191 let Some(extension_id) = parts.next() else {
192 continue;
193 };
194 let Some(version) = parts.next() else {
195 continue;
196 };
197 if parts.next() == Some("manifest.json") {
198 published_versions
199 .entry(extension_id)
200 .or_default()
201 .push(version);
202 }
203 }
204
205 let known_versions = app_state.db.get_known_extension_versions().await?;
206
207 let mut new_versions = HashMap::<&str, Vec<NewExtensionVersion>>::default();
208 let empty = Vec::new();
209 for (extension_id, published_versions) in published_versions {
210 let known_versions = known_versions.get(extension_id).unwrap_or(&empty);
211
212 for published_version in published_versions {
213 if known_versions
214 .binary_search_by_key(&published_version, String::as_str)
215 .is_err()
216 {
217 if let Some(extension) = fetch_extension_manifest(
218 blob_store_client,
219 blob_store_bucket,
220 extension_id,
221 published_version,
222 )
223 .await
224 .log_err()
225 {
226 new_versions
227 .entry(extension_id)
228 .or_default()
229 .push(extension);
230 }
231 }
232 }
233 }
234
235 app_state
236 .db
237 .insert_extension_versions(&new_versions)
238 .await?;
239
240 log::info!(
241 "fetched {} new extensions from blob store",
242 new_versions.values().map(|v| v.len()).sum::<usize>()
243 );
244
245 Ok(())
246}
247
248async fn fetch_extension_manifest(
249 blob_store_client: &aws_sdk_s3::Client,
250 blob_store_bucket: &String,
251 extension_id: &str,
252 version: &str,
253) -> Result<NewExtensionVersion, anyhow::Error> {
254 let object = blob_store_client
255 .get_object()
256 .bucket(blob_store_bucket)
257 .key(format!("extensions/{extension_id}/{version}/manifest.json"))
258 .send()
259 .await?;
260 let manifest_bytes = object
261 .body
262 .collect()
263 .await
264 .map(|data| data.into_bytes())
265 .with_context(|| {
266 format!("failed to download manifest for extension {extension_id} version {version}")
267 })?
268 .to_vec();
269 let manifest =
270 serde_json::from_slice::<ExtensionManifest>(&manifest_bytes).with_context(|| {
271 format!(
272 "invalid manifest for extension {extension_id} version {version}: {}",
273 String::from_utf8_lossy(&manifest_bytes)
274 )
275 })?;
276 let published_at = object.last_modified.ok_or_else(|| {
277 anyhow!("missing last modified timestamp for extension {extension_id} version {version}")
278 })?;
279 let published_at = time::OffsetDateTime::from_unix_timestamp_nanos(published_at.as_nanos())?;
280 let published_at = PrimitiveDateTime::new(published_at.date(), published_at.time());
281 let version = semver::Version::parse(&manifest.version).with_context(|| {
282 format!("invalid version for extension {extension_id} version {version}")
283 })?;
284 Ok(NewExtensionVersion {
285 name: manifest.name,
286 version,
287 description: manifest.description.unwrap_or_default(),
288 authors: manifest.authors,
289 repository: manifest.repository,
290 published_at,
291 })
292}