use super::*;
use nostr::{EventBuilder, JsonUtil, Kind, Tag, TagKind};
pub(super) fn bounded_servers(servers: &[String]) -> Vec<String> {
let mut result = Vec::new();
for server in servers {
if server.len() > 512 {
continue;
}
let Ok(url) = url::Url::parse(server) else {
continue;
};
if !matches!(url.scheme(), "http" | "https")
|| url.host_str().is_none()
|| !url.username().is_empty()
|| url.password().is_some()
|| url.query().is_some()
|| url.fragment().is_some()
{
continue;
}
let server = url.as_str().trim_end_matches('/').to_string();
if !result.contains(&server) {
result.push(server);
}
if result.len() == local_store::MAX_SERVERS {
break;
}
}
result
}
async fn upload_block(
client: &reqwest::Client,
keys: &nostr::Keys,
server: &str,
hash: &Hash,
data: &[u8],
) -> anyhow::Result<()> {
anyhow::ensure!(
hashtree_core::sha256(data) == *hash,
"attachment hash mismatch"
);
let hash = to_hex(hash);
let auth = EventBuilder::new(Kind::Custom(24242), "Upload")
.tags([
Tag::custom(TagKind::custom("t"), ["upload"]),
Tag::custom(TagKind::custom("x"), [hash.as_str()]),
Tag::custom(
TagKind::custom("expiration"),
[(local_store::now() + 300).to_string()],
),
])
.sign_with_keys(keys)?;
let mut response = client
.put(format!("{server}/upload"))
.header(
"Authorization",
format!(
"Nostr {}",
base64::engine::general_purpose::STANDARD.encode(auth.as_json())
),
)
.header("Content-Type", "application/octet-stream")
.header("x-sha-256", &hash)
.body(data.to_vec())
.send()
.await?;
anyhow::ensure!(
response.status().is_success(),
"attachment server rejected upload"
);
anyhow::ensure!(
response.content_length().is_none_or(|n| n <= 4096),
"upload acknowledgment is too large"
);
let mut body = Vec::new();
while let Some(part) = response.chunk().await? {
anyhow::ensure!(
body.len() + part.len() <= 4096,
"upload acknowledgment is too large"
);
body.extend_from_slice(&part);
}
let descriptor = serde_json::from_slice::<serde_json::Value>(&body).ok();
if let Some(remote_hash) = descriptor
.as_ref()
.and_then(|v| v.get("sha256"))
.and_then(|v| v.as_str())
{
anyhow::ensure!(
remote_hash == hash,
"server acknowledged a different attachment hash"
);
} else {
anyhow::ensure!(
download_attachment_blob(client, server, &hash, MAX_ATTACHMENT_BYTES + 64)
.await?
.is_some(),
"server did not confirm the uploaded ciphertext"
);
}
Ok(())
}
pub(super) async fn replicate_tree(
store: &Arc<AttachmentChunkStore>,
client: &reqwest::Client,
keys: &nostr::Keys,
root: Hash,
server: &str,
) -> anyhow::Result<()> {
let lease = store.lease(&root)?;
let recovery = store.confirmed_servers(&root)?;
for hash in &lease.hashes {
let mut data = store.get(hash).await.ok().flatten();
if data.is_none() {
for previous in &recovery {
if let Ok(Some(bytes)) = download_attachment_blob(
client,
previous,
&to_hex(hash),
MAX_ATTACHMENT_BYTES + 64,
)
.await
{
store
.put(*hash, bytes.clone())
.await
.map_err(|e| anyhow::anyhow!(e.to_string()))?;
data = Some(bytes);
break;
}
}
}
let data = data.ok_or_else(|| anyhow::anyhow!("attachment block is unavailable"))?;
upload_block(client, keys, server, hash, &data).await?;
}
anyhow::ensure!(
download_attachment_blob(client, server, &to_hex(&root), MAX_ATTACHMENT_BYTES + 64)
.await?
.is_some(),
"root ciphertext URL is unavailable"
);
store.finish_replica(&root, server, true)?;
Ok(())
}
pub(in crate::core) struct AttachmentService {
owner: String,
pub(in crate::core) store: Arc<AttachmentChunkStore>,
task: tokio::task::JoinHandle<()>,
}
impl Drop for AttachmentService {
fn drop(&mut self) {
self.task.abort();
self.store.close();
}
}
impl AppCore {
pub(in crate::core) fn ensure_attachment_service(
&mut self,
) -> anyhow::Result<Arc<AttachmentChunkStore>> {
#[cfg(any(target_os = "ios", test))]
anyhow::ensure!(!self.suspended, "attachment storage is suspended");
let account = self
.logged_in
.as_ref()
.ok_or_else(|| anyhow::anyhow!("no active attachment account"))?;
let owner = account.owner_pubkey.to_hex();
if let Some(service) = &self.attachment_service {
anyhow::ensure!(
service.owner == owner,
"attachment service belongs to another account"
);
return Ok(service.store.clone());
}
let keys = account
.owner_keys
.as_ref()
.unwrap_or(&account.device_keys)
.clone();
let store = AttachmentChunkStore::open(
&self.data_dir.join("attachments"),
local_store::DEFAULT_BUDGET,
)?;
store.bind_owner(&owner)?;
self.resolve_attachment_staging(&store, None)?;
let configured_servers = bounded_servers(&blossom_servers_from_config().1);
store.configure_pending(&configured_servers)?;
let client = reqwest::Client::builder()
.timeout(Duration::from_secs(15))
.redirect(reqwest::redirect::Policy::none())
.build()?;
let worker_store = store.clone();
let sender = self.core_sender.clone();
let task = self.runtime.spawn(async move {
let mut configured_servers = configured_servers;
let mut checked_configuration = std::time::Instant::now();
loop {
if worker_store.is_suspended() {
worker_store.wake.notified().await;
continue;
}
if checked_configuration.elapsed() >= Duration::from_secs(30) {
checked_configuration = std::time::Instant::now();
let current = bounded_servers(&blossom_servers_from_config().1);
if current != configured_servers {
if let Err(error) = worker_store.configure_pending(¤t) {
if error.is::<local_store::StorageSuspended>() {
continue;
}
break;
}
configured_servers = current;
}
}
match worker_store.next_replica() {
Ok(Some((root, server))) => {
let complete = tokio::time::timeout(
Duration::from_secs(600),
replicate_tree(&worker_store, &client, &keys, root, &server),
)
.await
.is_ok_and(|result| result.is_ok());
if complete {
let _ = sender.send(CoreMsg::Internal(Box::new(
InternalEvent::AttachmentReceiptsUpdated,
)));
} else if let Err(error) =
worker_store.finish_replica(&root, &server, false)
{
if error.is::<local_store::StorageSuspended>() {
continue;
}
break;
}
}
Ok(None) => tokio::select! {
_ = worker_store.wake.notified() => (),
_ = tokio::time::sleep(Duration::from_secs(30)) => (),
},
Err(error) if error.is::<local_store::StorageSuspended>() => continue,
Err(_) => break,
}
}
});
*active_local_store()
.write()
.unwrap_or_else(|p| p.into_inner()) = Some(Arc::downgrade(&store));
self.attachment_service = Some(AttachmentService {
owner,
store: store.clone(),
task,
});
Ok(store)
}
pub(in crate::core) fn stop_attachment_service(&mut self) {
if let Some(service) = self.attachment_service.take() {
service.task.abort();
service.store.close();
let mut active = active_local_store()
.write()
.unwrap_or_else(|p| p.into_inner());
if active
.as_ref()
.and_then(Weak::upgrade)
.is_some_and(|store| store.id == service.store.id)
{
*active = None;
}
}
}
pub(in crate::core) fn clear_attachment_service(&mut self) {
if let Some(service) = self.attachment_service.take() {
service.task.abort();
let result = service.store.close_and_clear();
service.store.close();
let mut active = active_local_store()
.write()
.unwrap_or_else(|p| p.into_inner());
if active
.as_ref()
.and_then(Weak::upgrade)
.is_some_and(|store| store.id == service.store.id)
{
*active = None;
}
drop(active);
if let Err(error) = result {
self.push_debug_log("attachment.storage.clear.error", error.to_string());
}
return;
}
let directory = self.data_dir.join("attachments");
if !directory.join("index.sqlite3").exists() {
return;
}
let result = AttachmentChunkStore::open(&directory, local_store::DEFAULT_BUDGET)
.and_then(|store| store.close_and_clear());
if let Err(error) = result {
self.push_debug_log("attachment.storage.clear.error", error.to_string());
}
}
}