mirror of
https://github.com/block/buzz.git
synced 2026-10-29 05:21:31 +00:00
Serialize profile avatar writes across commands
Signed-off-by: coder 3 <3bc2c73cf6228485ecee3c37d374aa143eab158665389f5168c4cbbf82ef7018@buzz.block.builderlab.xyz> Signed-off-by: Brainy Bumble <607666932e190ac5b35e80f42080da5e94b6b256c1bc6ab5e5b083d68393bdea@buzz.block.builderlab.xyz> Signed-off-by: bb-expert <51cde9442e46eac81c83ec71552708c3b0cb96a88ff8e1581d8fb4de281afbf3@buzz.block.builderlab.xyz>
This commit is contained in:
@@ -15,8 +15,8 @@ use crate::{
|
||||
models::{ProfileInfo, SearchUsersResponse, UserNotesResponse, UsersBatchResponse},
|
||||
nostr_convert,
|
||||
relay::{
|
||||
query_relay, query_relay_at_with_keys, relay_http_base_url, submit_event,
|
||||
submit_event_at_with_keys,
|
||||
query_relay, query_relay_at_with_keys, relay_api_base_url_with_override,
|
||||
relay_http_base_url, submit_event_at_with_keys,
|
||||
},
|
||||
};
|
||||
|
||||
@@ -49,60 +49,39 @@ pub async fn update_profile(
|
||||
nip05_handle: Option<String>,
|
||||
state: State<'_, AppState>,
|
||||
) -> Result<ProfileInfo, String> {
|
||||
// Read-merge-write: kind 0 is a full profile snapshot.
|
||||
let my_pubkey = current_pubkey_hex(&state)?;
|
||||
let prior_events = query_relay(
|
||||
&state,
|
||||
&[serde_json::json!({
|
||||
"kinds": [0],
|
||||
"authors": [my_pubkey],
|
||||
"limit": 1
|
||||
})],
|
||||
update_profile_inner(display_name, name, avatar_url, about, nip05_handle, &state).await
|
||||
}
|
||||
|
||||
async fn update_profile_inner(
|
||||
display_name: Option<String>,
|
||||
name: Option<String>,
|
||||
avatar_url: Option<String>,
|
||||
about: Option<String>,
|
||||
nip05_handle: Option<String>,
|
||||
state: &AppState,
|
||||
) -> Result<ProfileInfo, String> {
|
||||
let signer = state.signing_keys()?;
|
||||
let my_pubkey = signer.public_key().to_hex();
|
||||
let api_base_url = relay_api_base_url_with_override(state)
|
||||
.trim_end_matches('/')
|
||||
.to_string();
|
||||
|
||||
update_profile_with_scope(
|
||||
state,
|
||||
&api_base_url,
|
||||
&signer,
|
||||
&my_pubkey,
|
||||
ProfileUpdateFields {
|
||||
display_name: display_name.as_deref(),
|
||||
name: name.as_deref(),
|
||||
avatar_url: avatar_url.as_deref(),
|
||||
about: about.as_deref(),
|
||||
nip05_handle: nip05_handle.as_deref(),
|
||||
normalize_updates: false,
|
||||
},
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
|
||||
// Pull the current content as a JSON object so we can merge with
|
||||
// the caller's overrides.
|
||||
let current: Value = prior_events
|
||||
.first()
|
||||
.and_then(|ev| serde_json::from_str::<Value>(&ev.content).ok())
|
||||
.unwrap_or(Value::Null);
|
||||
|
||||
let dn = display_name
|
||||
.as_deref()
|
||||
.or_else(|| current.get("display_name").and_then(Value::as_str));
|
||||
let name = name
|
||||
.as_deref()
|
||||
.or_else(|| current.get("name").and_then(Value::as_str));
|
||||
let picture = avatar_url
|
||||
.as_deref()
|
||||
.or_else(|| current.get("picture").and_then(Value::as_str));
|
||||
let ab = about
|
||||
.as_deref()
|
||||
.or_else(|| current.get("about").and_then(Value::as_str));
|
||||
let nip05 = nip05_handle
|
||||
.as_deref()
|
||||
.or_else(|| current.get("nip05").and_then(Value::as_str));
|
||||
|
||||
let builder = events::build_profile(dn, name, picture, ab, nip05)?;
|
||||
submit_event(builder, &state).await?;
|
||||
|
||||
// Re-fetch to return canonical profile.
|
||||
let events = query_relay(
|
||||
&state,
|
||||
&[serde_json::json!({
|
||||
"kinds": [0],
|
||||
"authors": [current_pubkey_hex(&state)?],
|
||||
"limit": 1
|
||||
})],
|
||||
)
|
||||
.await?;
|
||||
|
||||
Ok(events
|
||||
.first()
|
||||
.map(nostr_convert::profile_info_from_event)
|
||||
.transpose()?
|
||||
.unwrap_or_else(|| empty_profile_info(¤t_pubkey_hex_unwrap(&state))))
|
||||
.await
|
||||
}
|
||||
|
||||
fn profile_update_lock(
|
||||
@@ -170,19 +149,59 @@ async fn update_profile_at_relay_inner(
|
||||
let api_base_url = relay_http_base_url(&relay_url)
|
||||
.trim_end_matches('/')
|
||||
.to_string();
|
||||
|
||||
update_profile_with_scope(
|
||||
state,
|
||||
&api_base_url,
|
||||
&signer,
|
||||
&expected_pubkey,
|
||||
ProfileUpdateFields {
|
||||
display_name: display_name.as_deref(),
|
||||
name: name.as_deref(),
|
||||
avatar_url: avatar_url.as_deref(),
|
||||
about: None,
|
||||
nip05_handle: None,
|
||||
normalize_updates: true,
|
||||
},
|
||||
Some(expected_avatar_url.as_deref()),
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
struct ProfileUpdateFields<'a> {
|
||||
display_name: Option<&'a str>,
|
||||
name: Option<&'a str>,
|
||||
avatar_url: Option<&'a str>,
|
||||
about: Option<&'a str>,
|
||||
nip05_handle: Option<&'a str>,
|
||||
normalize_updates: bool,
|
||||
}
|
||||
|
||||
async fn update_profile_with_scope(
|
||||
state: &AppState,
|
||||
api_base_url: &str,
|
||||
signer: &nostr::Keys,
|
||||
expected_pubkey: &str,
|
||||
fields: ProfileUpdateFields<'_>,
|
||||
expected_avatar_url: Option<Option<&str>>,
|
||||
) -> Result<ProfileInfo, String> {
|
||||
if signer.public_key().to_hex() != expected_pubkey {
|
||||
return Err("profile identity changed before avatar save".to_string());
|
||||
}
|
||||
|
||||
let filter = serde_json::json!({
|
||||
"kinds": [0],
|
||||
"authors": [expected_pubkey],
|
||||
"limit": 1
|
||||
});
|
||||
let profile_update_lock = profile_update_lock(&api_base_url, &expected_pubkey);
|
||||
let profile_update_lock = profile_update_lock(api_base_url, expected_pubkey);
|
||||
let _profile_update_guard = profile_update_lock.lock().await;
|
||||
|
||||
let prior_events = query_relay_at_with_keys(
|
||||
&state,
|
||||
&api_base_url,
|
||||
state,
|
||||
api_base_url,
|
||||
std::slice::from_ref(&filter),
|
||||
&signer,
|
||||
signer,
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
@@ -190,42 +209,38 @@ async fn update_profile_at_relay_inner(
|
||||
let current: Value = prior_event
|
||||
.and_then(|event| serde_json::from_str::<Value>(&event.content).ok())
|
||||
.unwrap_or(Value::Null);
|
||||
if should_check_expected_avatar(avatar_url.as_deref()) {
|
||||
assert_expected_avatar(¤t, expected_avatar_url.as_deref())?;
|
||||
if should_check_expected_avatar(fields.avatar_url) {
|
||||
if let Some(expected_avatar_url) = expected_avatar_url {
|
||||
assert_expected_avatar(¤t, expected_avatar_url)?;
|
||||
}
|
||||
}
|
||||
|
||||
let builder = build_scoped_profile_event(
|
||||
¤t,
|
||||
display_name.as_deref(),
|
||||
name.as_deref(),
|
||||
avatar_url.as_deref(),
|
||||
prior_event,
|
||||
)?;
|
||||
submit_event_at_with_keys(builder, &state, &api_base_url, &signer).await?;
|
||||
let builder = build_profile_event_from_current(¤t, fields, prior_event)?;
|
||||
submit_event_at_with_keys(builder, state, api_base_url, signer).await?;
|
||||
|
||||
let events = query_relay_at_with_keys(&state, &api_base_url, &[filter], &signer, None).await?;
|
||||
let events = query_relay_at_with_keys(state, api_base_url, &[filter], signer, None).await?;
|
||||
Ok(events
|
||||
.first()
|
||||
.map(nostr_convert::profile_info_from_event)
|
||||
.transpose()?
|
||||
.unwrap_or_else(|| empty_profile_info(&expected_pubkey)))
|
||||
.unwrap_or_else(|| empty_profile_info(expected_pubkey)))
|
||||
}
|
||||
|
||||
fn build_scoped_profile_event(
|
||||
fn build_profile_event_from_current(
|
||||
current: &Value,
|
||||
display_name: Option<&str>,
|
||||
name: Option<&str>,
|
||||
avatar_url: Option<&str>,
|
||||
fields: ProfileUpdateFields<'_>,
|
||||
prior_event: Option<&nostr::Event>,
|
||||
) -> Result<nostr::EventBuilder, String> {
|
||||
let display_name = normalized_profile_field(display_name)
|
||||
let display_name = profile_update_field(fields.display_name, fields.normalize_updates)
|
||||
.or_else(|| current.get("display_name").and_then(Value::as_str));
|
||||
let name =
|
||||
normalized_profile_field(name).or_else(|| current.get("name").and_then(Value::as_str));
|
||||
let picture = normalized_profile_field(avatar_url)
|
||||
let name = profile_update_field(fields.name, fields.normalize_updates)
|
||||
.or_else(|| current.get("name").and_then(Value::as_str));
|
||||
let picture = profile_update_field(fields.avatar_url, fields.normalize_updates)
|
||||
.or_else(|| current.get("picture").and_then(Value::as_str));
|
||||
let about = current.get("about").and_then(Value::as_str);
|
||||
let nip05 = current.get("nip05").and_then(Value::as_str);
|
||||
let about = profile_update_field(fields.about, fields.normalize_updates)
|
||||
.or_else(|| current.get("about").and_then(Value::as_str));
|
||||
let nip05 = profile_update_field(fields.nip05_handle, fields.normalize_updates)
|
||||
.or_else(|| current.get("nip05").and_then(Value::as_str));
|
||||
|
||||
Ok(
|
||||
events::build_profile(display_name, name, picture, about, nip05)?.custom_created_at(
|
||||
@@ -234,6 +249,14 @@ fn build_scoped_profile_event(
|
||||
)
|
||||
}
|
||||
|
||||
fn profile_update_field(value: Option<&str>, normalize: bool) -> Option<&str> {
|
||||
if normalize {
|
||||
normalized_profile_field(value)
|
||||
} else {
|
||||
value
|
||||
}
|
||||
}
|
||||
|
||||
fn normalized_profile_field(value: Option<&str>) -> Option<&str> {
|
||||
value.map(str::trim).filter(|value| !value.is_empty())
|
||||
}
|
||||
@@ -493,18 +516,24 @@ mod tests {
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::{mpsc, oneshot};
|
||||
|
||||
#[derive(Clone, Copy)]
|
||||
enum ProfileRelayGate {
|
||||
Query,
|
||||
Submit,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
struct ProfileRelayState {
|
||||
current: Arc<tokio::sync::Mutex<Option<nostr::Event>>>,
|
||||
first_submit_gate: Arc<tokio::sync::Mutex<Option<oneshot::Receiver<()>>>>,
|
||||
first_submit_started: mpsc::Sender<()>,
|
||||
gate: Arc<tokio::sync::Mutex<Option<(ProfileRelayGate, oneshot::Receiver<()>)>>>,
|
||||
gate_started: mpsc::Sender<()>,
|
||||
}
|
||||
|
||||
struct ProfileRelay {
|
||||
url: String,
|
||||
current: Arc<tokio::sync::Mutex<Option<nostr::Event>>>,
|
||||
first_submit_release: Option<oneshot::Sender<()>>,
|
||||
first_submit_started: mpsc::Receiver<()>,
|
||||
gate_release: Option<oneshot::Sender<()>>,
|
||||
gate_started: mpsc::Receiver<()>,
|
||||
server: tokio::task::JoinHandle<()>,
|
||||
}
|
||||
|
||||
@@ -534,22 +563,32 @@ mod tests {
|
||||
|
||||
async fn profile_relay(
|
||||
initial: Option<nostr::Event>,
|
||||
delay_first_submit: bool,
|
||||
gate: Option<ProfileRelayGate>,
|
||||
) -> ProfileRelay {
|
||||
let current = Arc::new(tokio::sync::Mutex::new(initial));
|
||||
let (first_submit_started, started_rx) = mpsc::channel(2);
|
||||
let (first_submit_release, release_rx) = oneshot::channel();
|
||||
let (gate_started, started_rx) = mpsc::channel(2);
|
||||
let (gate_release, release_rx) = oneshot::channel();
|
||||
let state = ProfileRelayState {
|
||||
current: Arc::clone(¤t),
|
||||
first_submit_gate: Arc::new(tokio::sync::Mutex::new(
|
||||
delay_first_submit.then_some(release_rx),
|
||||
)),
|
||||
first_submit_started,
|
||||
gate: Arc::new(tokio::sync::Mutex::new(gate.map(|gate| (gate, release_rx)))),
|
||||
gate_started,
|
||||
};
|
||||
let app = Router::new()
|
||||
.route(
|
||||
"/query",
|
||||
post(|AxumState(state): AxumState<ProfileRelayState>| async move {
|
||||
let release = {
|
||||
let mut gate = state.gate.lock().await;
|
||||
if matches!(gate.as_ref().map(|(gate, _)| gate), Some(ProfileRelayGate::Query)) {
|
||||
gate.take().map(|(_, release)| release)
|
||||
} else {
|
||||
None
|
||||
}
|
||||
};
|
||||
if let Some(release) = release {
|
||||
let _ = state.gate_started.send(()).await;
|
||||
let _ = release.await;
|
||||
}
|
||||
let events = state
|
||||
.current
|
||||
.lock()
|
||||
@@ -566,12 +605,20 @@ mod tests {
|
||||
post(
|
||||
|AxumState(state): AxumState<ProfileRelayState>, Json(event): Json<nostr::Event>| async move {
|
||||
assert!(event.verify().is_ok());
|
||||
if let Some(release) = state.first_submit_gate.lock().await.take() {
|
||||
let _ = state.first_submit_started.send(()).await;
|
||||
let release = {
|
||||
let mut gate = state.gate.lock().await;
|
||||
if matches!(gate.as_ref().map(|(gate, _)| gate), Some(ProfileRelayGate::Submit)) {
|
||||
gate.take().map(|(_, release)| release)
|
||||
} else {
|
||||
None
|
||||
}
|
||||
};
|
||||
if let Some(release) = release {
|
||||
let _ = state.gate_started.send(()).await;
|
||||
let _ = release.await;
|
||||
}
|
||||
let event_id = event.id.to_hex();
|
||||
*state.current.lock().await = Some(event);
|
||||
replace_profile_head(&state.current, event).await;
|
||||
Json(serde_json::json!({"event_id": event_id, "accepted": true, "message": ""}))
|
||||
},
|
||||
),
|
||||
@@ -585,12 +632,35 @@ mod tests {
|
||||
ProfileRelay {
|
||||
url: format!("http://{address}"),
|
||||
current,
|
||||
first_submit_release: delay_first_submit.then_some(first_submit_release),
|
||||
first_submit_started: started_rx,
|
||||
gate_release: gate.map(|_| gate_release),
|
||||
gate_started: started_rx,
|
||||
server,
|
||||
}
|
||||
}
|
||||
|
||||
fn profile_state_for_relay(keys: &nostr::Keys, relay_url: &str) -> AppState {
|
||||
let state = profile_state(keys);
|
||||
*state
|
||||
.relay_url_override
|
||||
.lock()
|
||||
.expect("lock relay override") = Some(relay_url.to_string());
|
||||
state
|
||||
}
|
||||
|
||||
async fn replace_profile_head(
|
||||
current: &tokio::sync::Mutex<Option<nostr::Event>>,
|
||||
candidate: nostr::Event,
|
||||
) {
|
||||
let mut current = current.lock().await;
|
||||
let should_replace = current.as_ref().is_none_or(|head| {
|
||||
candidate.created_at > head.created_at
|
||||
|| (candidate.created_at == head.created_at && candidate.id < head.id)
|
||||
});
|
||||
if should_replace {
|
||||
*current = Some(candidate);
|
||||
}
|
||||
}
|
||||
|
||||
fn profile_content(event: &nostr::Event) -> Value {
|
||||
serde_json::from_str(&event.content).expect("profile json")
|
||||
}
|
||||
@@ -602,7 +672,7 @@ mod tests {
|
||||
let keys = nostr::Keys::generate();
|
||||
let state = profile_state(&keys);
|
||||
let pubkey = keys.public_key().to_hex();
|
||||
let relay = profile_relay(None, false).await;
|
||||
let relay = profile_relay(None, None).await;
|
||||
|
||||
let saved = update_profile_at_relay_inner(
|
||||
relay.url.clone(),
|
||||
@@ -663,7 +733,7 @@ mod tests {
|
||||
None,
|
||||
Some("https://example.com/old.png"),
|
||||
)),
|
||||
false,
|
||||
None,
|
||||
)
|
||||
.await;
|
||||
|
||||
@@ -720,7 +790,7 @@ mod tests {
|
||||
Some("old"),
|
||||
Some("https://example.com/avatar.png"),
|
||||
)),
|
||||
false,
|
||||
None,
|
||||
)
|
||||
.await;
|
||||
|
||||
@@ -755,7 +825,7 @@ mod tests {
|
||||
let keys = nostr::Keys::generate();
|
||||
let state = Arc::new(profile_state(&keys));
|
||||
let pubkey = keys.public_key().to_hex();
|
||||
let mut relay = profile_relay(None, true).await;
|
||||
let mut relay = profile_relay(None, Some(ProfileRelayGate::Submit)).await;
|
||||
|
||||
let newer = tokio::spawn({
|
||||
let state = Arc::clone(&state);
|
||||
@@ -775,7 +845,7 @@ mod tests {
|
||||
}
|
||||
});
|
||||
relay
|
||||
.first_submit_started
|
||||
.gate_started
|
||||
.recv()
|
||||
.await
|
||||
.expect("newer save reached relay submit");
|
||||
@@ -799,7 +869,7 @@ mod tests {
|
||||
});
|
||||
|
||||
relay
|
||||
.first_submit_release
|
||||
.gate_release
|
||||
.take()
|
||||
.expect("first submit release")
|
||||
.send(())
|
||||
@@ -822,6 +892,110 @@ mod tests {
|
||||
crate::relay_admission::reset_rate_limit_gate();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn ordinary_profile_update_waits_for_deferred_query_and_wins_canonical_head() {
|
||||
let _serial = crate::relay_admission::TEST_SERIAL.lock().await;
|
||||
crate::relay_admission::reset_rate_limit_gate();
|
||||
let keys = nostr::Keys::generate();
|
||||
let pubkey = keys.public_key().to_hex();
|
||||
let future_created_at = nostr::Timestamp::now().as_secs() + 60;
|
||||
let existing = events::build_profile(
|
||||
Some("Existing"),
|
||||
Some("existing"),
|
||||
Some("https://example.com/old.png"),
|
||||
None,
|
||||
None,
|
||||
)
|
||||
.expect("build existing profile")
|
||||
.custom_created_at(nostr::Timestamp::from(future_created_at))
|
||||
.sign_with_keys(&keys)
|
||||
.expect("sign existing profile");
|
||||
let mut relay = profile_relay(Some(existing), Some(ProfileRelayGate::Query)).await;
|
||||
let state = Arc::new(profile_state_for_relay(&keys, &relay.url));
|
||||
|
||||
let deferred = tokio::spawn({
|
||||
let state = Arc::clone(&state);
|
||||
let relay_url = relay.url.clone();
|
||||
let pubkey = pubkey.clone();
|
||||
async move {
|
||||
update_profile_at_relay_inner(
|
||||
relay_url,
|
||||
pubkey,
|
||||
Some("https://example.com/old.png".to_string()),
|
||||
Some("https://example.com/deferred.png".to_string()),
|
||||
None,
|
||||
None,
|
||||
&state,
|
||||
)
|
||||
.await
|
||||
}
|
||||
});
|
||||
relay
|
||||
.gate_started
|
||||
.recv()
|
||||
.await
|
||||
.expect("deferred save reached held scoped query");
|
||||
|
||||
let settings = tokio::spawn({
|
||||
let state = Arc::clone(&state);
|
||||
async move {
|
||||
update_profile_inner(
|
||||
Some("Settings Name".to_string()),
|
||||
None,
|
||||
Some("https://example.com/settings.png".to_string()),
|
||||
None,
|
||||
None,
|
||||
&state,
|
||||
)
|
||||
.await
|
||||
}
|
||||
});
|
||||
let mut settings = Box::pin(settings);
|
||||
tokio::select! {
|
||||
result = &mut settings => {
|
||||
let completed = result
|
||||
.map(|profile| profile.map(|profile| profile.avatar_url))
|
||||
.map_err(|error| error.to_string());
|
||||
panic!("ordinary Settings update completed before deferred query lock released: {completed:?}");
|
||||
}
|
||||
_ = tokio::time::sleep(std::time::Duration::from_millis(100)) => {}
|
||||
}
|
||||
|
||||
relay
|
||||
.gate_release
|
||||
.take()
|
||||
.expect("deferred query release")
|
||||
.send(())
|
||||
.expect("release deferred query");
|
||||
let deferred_profile = deferred
|
||||
.await
|
||||
.expect("deferred task")
|
||||
.expect("deferred save");
|
||||
assert_eq!(
|
||||
deferred_profile.avatar_url.as_deref(),
|
||||
Some("https://example.com/deferred.png")
|
||||
);
|
||||
let settings_profile = settings
|
||||
.await
|
||||
.expect("settings task")
|
||||
.expect("settings save");
|
||||
assert_eq!(
|
||||
settings_profile.display_name.as_deref(),
|
||||
Some("Settings Name")
|
||||
);
|
||||
assert_eq!(
|
||||
settings_profile.avatar_url.as_deref(),
|
||||
Some("https://example.com/settings.png")
|
||||
);
|
||||
|
||||
let final_event = relay.current.lock().await.clone().expect("profile head");
|
||||
let final_content = profile_content(&final_event);
|
||||
assert_eq!(final_content["display_name"], "Settings Name");
|
||||
assert_eq!(final_content["picture"], "https://example.com/settings.png");
|
||||
assert_eq!(final_event.created_at.as_secs(), future_created_at + 2);
|
||||
crate::relay_admission::reset_rate_limit_gate();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn deferred_profile_signer_is_captured_and_rejects_wrong_identity() {
|
||||
let state = crate::app_state::build_app_state();
|
||||
@@ -855,11 +1029,16 @@ mod tests {
|
||||
.sign_with_keys(&keys)
|
||||
.expect("sign prior profile");
|
||||
|
||||
let builder = build_scoped_profile_event(
|
||||
let builder = build_profile_event_from_current(
|
||||
&serde_json::json!({"display_name": "Larry"}),
|
||||
None,
|
||||
None,
|
||||
Some("https://example.com/avatar.png"),
|
||||
ProfileUpdateFields {
|
||||
display_name: None,
|
||||
name: None,
|
||||
avatar_url: Some("https://example.com/avatar.png"),
|
||||
about: None,
|
||||
nip05_handle: None,
|
||||
normalize_updates: true,
|
||||
},
|
||||
Some(&prior_event),
|
||||
)
|
||||
.expect("build deferred profile");
|
||||
|
||||
Reference in New Issue
Block a user