1use chrono::Utc;
2use sea_orm::sea_query::IntoCondition;
3
4use super::*;
5
6impl Database {
7 pub async fn get_extensions(
8 &self,
9 filter: Option<&str>,
10 max_schema_version: i32,
11 limit: usize,
12 ) -> Result<Vec<ExtensionMetadata>> {
13 self.transaction(|tx| async move {
14 let mut condition = Condition::all()
15 .add(
16 extension::Column::LatestVersion
17 .into_expr()
18 .eq(extension_version::Column::Version.into_expr()),
19 )
20 .add(extension_version::Column::SchemaVersion.lte(max_schema_version));
21 if let Some(filter) = filter {
22 let fuzzy_name_filter = Self::fuzzy_like_string(filter);
23 condition = condition.add(Expr::cust_with_expr("name ILIKE $1", fuzzy_name_filter));
24 }
25
26 self.get_extensions_where(condition, Some(limit as u64), &tx)
27 .await
28 })
29 .await
30 }
31
32 pub async fn get_extensions_by_ids(
33 &self,
34 ids: &[&str],
35 max_schema_version: i32,
36 ) -> Result<Vec<ExtensionMetadata>> {
37 self.transaction(|tx| async move {
38 let condition = Condition::all()
39 .add(
40 extension::Column::LatestVersion
41 .into_expr()
42 .eq(extension_version::Column::Version.into_expr()),
43 )
44 .add(extension::Column::ExternalId.is_in(ids.iter().copied()))
45 .add(extension_version::Column::SchemaVersion.lte(max_schema_version));
46
47 self.get_extensions_where(condition, None, &tx).await
48 })
49 .await
50 }
51
52 /// Returns all of the versions for the extension with the given ID.
53 pub async fn get_extension_versions(
54 &self,
55 extension_id: &str,
56 ) -> Result<Vec<ExtensionMetadata>> {
57 self.transaction(|tx| async move {
58 let condition = extension::Column::ExternalId
59 .eq(extension_id)
60 .into_condition();
61
62 self.get_extensions_where(condition, None, &tx).await
63 })
64 .await
65 }
66
67 async fn get_extensions_where(
68 &self,
69 condition: Condition,
70 limit: Option<u64>,
71 tx: &DatabaseTransaction,
72 ) -> Result<Vec<ExtensionMetadata>> {
73 let extensions = extension::Entity::find()
74 .inner_join(extension_version::Entity)
75 .select_also(extension_version::Entity)
76 .filter(condition)
77 .order_by_desc(extension::Column::TotalDownloadCount)
78 .order_by_asc(extension::Column::Name)
79 .limit(limit)
80 .all(tx)
81 .await?;
82
83 Ok(extensions
84 .into_iter()
85 .filter_map(|(extension, version)| {
86 Some(metadata_from_extension_and_version(extension, version?))
87 })
88 .collect())
89 }
90
91 pub async fn get_extension(&self, extension_id: &str) -> Result<Option<ExtensionMetadata>> {
92 self.transaction(|tx| async move {
93 let extension = extension::Entity::find()
94 .filter(extension::Column::ExternalId.eq(extension_id))
95 .filter(
96 extension::Column::LatestVersion
97 .into_expr()
98 .eq(extension_version::Column::Version.into_expr()),
99 )
100 .inner_join(extension_version::Entity)
101 .select_also(extension_version::Entity)
102 .one(&*tx)
103 .await?;
104
105 Ok(extension.and_then(|(extension, version)| {
106 Some(metadata_from_extension_and_version(extension, version?))
107 }))
108 })
109 .await
110 }
111
112 pub async fn get_extension_version(
113 &self,
114 extension_id: &str,
115 version: &str,
116 ) -> Result<Option<ExtensionMetadata>> {
117 self.transaction(|tx| async move {
118 let extension = extension::Entity::find()
119 .filter(extension::Column::ExternalId.eq(extension_id))
120 .filter(extension_version::Column::Version.eq(version))
121 .inner_join(extension_version::Entity)
122 .select_also(extension_version::Entity)
123 .one(&*tx)
124 .await?;
125
126 Ok(extension.and_then(|(extension, version)| {
127 Some(metadata_from_extension_and_version(extension, version?))
128 }))
129 })
130 .await
131 }
132
133 pub async fn get_known_extension_versions<'a>(&self) -> Result<HashMap<String, Vec<String>>> {
134 self.transaction(|tx| async move {
135 let mut extension_external_ids_by_id = HashMap::default();
136
137 let mut rows = extension::Entity::find().stream(&*tx).await?;
138 while let Some(row) = rows.next().await {
139 let row = row?;
140 extension_external_ids_by_id.insert(row.id, row.external_id);
141 }
142 drop(rows);
143
144 let mut known_versions_by_extension_id: HashMap<String, Vec<String>> =
145 HashMap::default();
146 let mut rows = extension_version::Entity::find().stream(&*tx).await?;
147 while let Some(row) = rows.next().await {
148 let row = row?;
149
150 let Some(extension_id) = extension_external_ids_by_id.get(&row.extension_id) else {
151 continue;
152 };
153
154 let versions = known_versions_by_extension_id
155 .entry(extension_id.clone())
156 .or_default();
157 if let Err(ix) = versions.binary_search(&row.version) {
158 versions.insert(ix, row.version);
159 }
160 }
161 drop(rows);
162
163 Ok(known_versions_by_extension_id)
164 })
165 .await
166 }
167
168 pub async fn insert_extension_versions(
169 &self,
170 versions_by_extension_id: &HashMap<&str, Vec<NewExtensionVersion>>,
171 ) -> Result<()> {
172 self.transaction(|tx| async move {
173 for (external_id, versions) in versions_by_extension_id {
174 if versions.is_empty() {
175 continue;
176 }
177
178 let latest_version = versions
179 .iter()
180 .max_by_key(|version| &version.version)
181 .unwrap();
182
183 let insert = extension::Entity::insert(extension::ActiveModel {
184 name: ActiveValue::Set(latest_version.name.clone()),
185 external_id: ActiveValue::Set(external_id.to_string()),
186 id: ActiveValue::NotSet,
187 latest_version: ActiveValue::Set(latest_version.version.to_string()),
188 total_download_count: ActiveValue::NotSet,
189 })
190 .on_conflict(
191 OnConflict::columns([extension::Column::ExternalId])
192 .update_column(extension::Column::ExternalId)
193 .to_owned(),
194 );
195
196 let extension = if tx.support_returning() {
197 insert.exec_with_returning(&*tx).await?
198 } else {
199 // Sqlite
200 insert.exec_without_returning(&*tx).await?;
201 extension::Entity::find()
202 .filter(extension::Column::ExternalId.eq(*external_id))
203 .one(&*tx)
204 .await?
205 .ok_or_else(|| anyhow!("failed to insert extension"))?
206 };
207
208 extension_version::Entity::insert_many(versions.iter().map(|version| {
209 extension_version::ActiveModel {
210 extension_id: ActiveValue::Set(extension.id),
211 published_at: ActiveValue::Set(version.published_at),
212 version: ActiveValue::Set(version.version.to_string()),
213 authors: ActiveValue::Set(version.authors.join(", ")),
214 repository: ActiveValue::Set(version.repository.clone()),
215 description: ActiveValue::Set(version.description.clone()),
216 schema_version: ActiveValue::Set(version.schema_version),
217 wasm_api_version: ActiveValue::Set(version.wasm_api_version.clone()),
218 download_count: ActiveValue::NotSet,
219 }
220 }))
221 .on_conflict(OnConflict::new().do_nothing().to_owned())
222 .exec_without_returning(&*tx)
223 .await?;
224
225 if let Ok(db_version) = semver::Version::parse(&extension.latest_version) {
226 if db_version >= latest_version.version {
227 continue;
228 }
229 }
230
231 let mut extension = extension.into_active_model();
232 extension.latest_version = ActiveValue::Set(latest_version.version.to_string());
233 extension.name = ActiveValue::set(latest_version.name.clone());
234 extension::Entity::update(extension).exec(&*tx).await?;
235 }
236
237 Ok(())
238 })
239 .await
240 }
241
242 pub async fn record_extension_download(&self, extension: &str, version: &str) -> Result<bool> {
243 self.transaction(|tx| async move {
244 #[derive(Copy, Clone, Debug, EnumIter, DeriveColumn)]
245 enum QueryId {
246 Id,
247 }
248
249 let extension_id: Option<ExtensionId> = extension::Entity::find()
250 .filter(extension::Column::ExternalId.eq(extension))
251 .select_only()
252 .column(extension::Column::Id)
253 .into_values::<_, QueryId>()
254 .one(&*tx)
255 .await?;
256 let Some(extension_id) = extension_id else {
257 return Ok(false);
258 };
259
260 extension_version::Entity::update_many()
261 .col_expr(
262 extension_version::Column::DownloadCount,
263 extension_version::Column::DownloadCount.into_expr().add(1),
264 )
265 .filter(
266 extension_version::Column::ExtensionId
267 .eq(extension_id)
268 .and(extension_version::Column::Version.eq(version)),
269 )
270 .exec(&*tx)
271 .await?;
272
273 extension::Entity::update_many()
274 .col_expr(
275 extension::Column::TotalDownloadCount,
276 extension::Column::TotalDownloadCount.into_expr().add(1),
277 )
278 .filter(extension::Column::Id.eq(extension_id))
279 .exec(&*tx)
280 .await?;
281
282 Ok(true)
283 })
284 .await
285 }
286}
287
288fn metadata_from_extension_and_version(
289 extension: extension::Model,
290 version: extension_version::Model,
291) -> ExtensionMetadata {
292 ExtensionMetadata {
293 id: extension.external_id.into(),
294 manifest: rpc::ExtensionApiManifest {
295 name: extension.name,
296 version: version.version.into(),
297 authors: version
298 .authors
299 .split(',')
300 .map(|author| author.trim().to_string())
301 .collect::<Vec<_>>(),
302 description: Some(version.description),
303 repository: version.repository,
304 schema_version: Some(version.schema_version),
305 wasm_api_version: version.wasm_api_version,
306 },
307
308 published_at: convert_time_to_chrono(version.published_at),
309 download_count: extension.total_download_count as u64,
310 }
311}
312
313pub fn convert_time_to_chrono(time: time::PrimitiveDateTime) -> chrono::DateTime<Utc> {
314 chrono::DateTime::from_naive_utc_and_offset(
315 chrono::NaiveDateTime::from_timestamp_opt(time.assume_utc().unix_timestamp(), 0).unwrap(),
316 Utc,
317 )
318}