use std::collections::HashMap; use std::path::Path; use std::sync::Arc; use chrono::{DateTime, Utc}; use crate::util::future::NonDetachingJoinHandle; use reqwest::Client; use serde::{Deserialize, Serialize}; use tokio::io::AsyncWrite; use ts_rs::TS; use url::Url; use crate::prelude::*; use crate::progress::PhaseProgressTrackerHandle; use crate::registry::signer::AcceptSigners; use crate::s9pk::S9pk; use crate::s9pk::merkle_archive::source::http::HttpSource; use crate::s9pk::merkle_archive::source::{ArchiveSource, Section}; use crate::sign::commitment::merkle_archive::MerkleArchiveCommitment; use crate::sign::commitment::{Commitment, Digestable}; use crate::sign::{AnySignature, AnyVerifyingKey}; use crate::upload::UploadingFile; #[derive(Debug, Deserialize, Serialize, TS)] #[serde(rename_all = "camelCase")] #[ts(export)] pub struct RegistryAsset { #[ts(type = "string")] pub published_at: DateTime, #[ts(type = "string")] pub url: Url, pub commitment: Commitment, pub signatures: HashMap, } impl RegistryAsset { pub fn all_signers(&self) -> AcceptSigners { AcceptSigners::All( self.signatures .keys() .cloned() .map(AcceptSigners::Signer) .collect(), ) } } impl RegistryAsset { pub fn validate(&self, context: &str, mut accept: AcceptSigners) -> Result<&Commitment, Error> { for (signer, signature) in &self.signatures { accept.process_signature(signer, &self.commitment, context, signature)?; } accept.try_accept()?; Ok(&self.commitment) } } impl Commitment<&'a HttpSource>> RegistryAsset { pub async fn download( &self, client: Client, dst: &mut (impl AsyncWrite + Unpin + Send + ?Sized), ) -> Result<(), Error> { self.commitment .copy_to(&HttpSource::new(client, self.url.clone()).await?, dst) .await } } impl RegistryAsset { pub async fn deserialize_s9pk( &self, client: Client, ) -> Result>>, Error> { S9pk::deserialize( &Arc::new(HttpSource::new(client, self.url.clone()).await?), Some(&self.commitment), ) .await } pub async fn deserialize_s9pk_buffered( &self, client: Client, progress: PhaseProgressTrackerHandle, ) -> Result>>, Error> { S9pk::deserialize( &Arc::new(BufferedHttpSource::new(client, self.url.clone(), progress).await?), Some(&self.commitment), ) .await } pub async fn download_to( &self, path: impl AsRef, client: Client, progress: PhaseProgressTrackerHandle, ) -> Result< ( S9pk>>, Arc, ), Error, > { let source = Arc::new( BufferedHttpSource::with_path(path, client, self.url.clone(), progress).await?, ); Ok(( S9pk::deserialize(&source, Some(&self.commitment)).await?, source, )) } } pub struct BufferedHttpSource { _download: NonDetachingJoinHandle<()>, file: UploadingFile, } impl BufferedHttpSource { pub async fn with_path( path: impl AsRef, client: Client, url: Url, progress: PhaseProgressTrackerHandle, ) -> Result { let (mut handle, file) = UploadingFile::with_path(path, progress).await?; let response = client.get(url).send().await?; Ok(Self { _download: tokio::spawn(async move { handle.download(response).await }).into(), file, }) } pub async fn new( client: Client, url: Url, progress: PhaseProgressTrackerHandle, ) -> Result { let (mut handle, file) = UploadingFile::new(progress).await?; let response = client.get(url).send().await?; Ok(Self { _download: tokio::spawn(async move { handle.download(response).await }).into(), file, }) } pub async fn wait_for_buffered(&self) -> Result<(), Error> { self.file.wait_for_complete().await } } impl ArchiveSource for BufferedHttpSource { type FetchReader = ::FetchReader; type FetchAllReader = ::FetchAllReader; async fn size(&self) -> Option { self.file.size().await } async fn fetch_all(&self) -> Result { self.file.fetch_all().await } async fn fetch(&self, position: u64, size: u64) -> Result { self.file.fetch(position, size).await } }