Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
100 changes: 89 additions & 11 deletions desktop/src-tauri/src/commands/channels.rs
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,9 @@ use crate::{
// the directory cursor, the not-modified hash, and member-count collection)
// lives in the `fetch` submodule to keep this file under the per-file line cap.
mod fetch;
use fetch::{compute_channels_hash, fetch_channels, DirectoryScope};
use fetch::{
compute_channels_hash, fetch_channels, fetch_open_directory_from_writer, DirectoryScope,
};

const STARTER_CHANNEL_NAMESPACE: uuid::Uuid = uuid::uuid!("3ce33bea-8f09-5f1b-9c85-8a7d2659e6b0");

Expand Down Expand Up @@ -230,19 +232,37 @@ fn is_duplicate_channel_rejection(error: &str) -> bool {
error.contains("relay rejected event:") && error.contains("duplicate: channel already exists")
}

/// A starter create the relay will refuse on every retry, with HTTP 400:
/// - another listed channel, archived ones included, holds its name:
/// `invalid: an open channel named "general" already exists`;
/// - an archived channel holds its deterministic id, which the relay checks
/// before duplicates: `invalid: channel is archived`.
fn is_permanent_starter_rejection(error: &str) -> bool {
error.starts_with("relay returned 400")
&& ((error.contains("invalid: an open channel named") && error.contains("already exists"))
|| error.contains("invalid: channel is archived"))
}

fn is_matching_starter_channel(channel: &ChannelInfo, spec: &StarterChannelSpec) -> bool {
normalize_channel_name(&channel.name) == normalize_channel_name(spec.name)
&& channel.channel_type == "stream"
&& channel.visibility == "open"
&& channel.archived_at.is_none()
}

fn has_all_starter_channels(channels: &[ChannelInfo]) -> bool {
STARTER_CHANNELS.iter().all(|spec| {
channels
.iter()
.any(|channel| is_matching_starter_channel(channel, spec))
})
/// Every starter that was not skipped has a usable channel.
fn has_all_starter_channels(
channels: &[ChannelInfo],
skipped: &std::collections::HashSet<&str>,
) -> bool {
STARTER_CHANNELS
.iter()
.filter(|spec| !skipped.contains(spec.slug))
.all(|spec| {
channels
.iter()
.any(|channel| is_matching_starter_channel(channel, spec))
})
}

async fn ensure_starter_channel_memberships(
Expand Down Expand Up @@ -413,6 +433,14 @@ async fn ensure_starter_channels_inner(
let creator_pubkey = creator_keys.public_key().to_hex();
let mut starter_ids = Vec::with_capacity(STARTER_CHANNELS.len());
let mut created_ids = std::collections::HashSet::new();
// Starters whose deterministic id the relay already holds.
let mut duplicate_starters = Vec::new();
// Starters that can never be created here: their name or id belongs to a
// channel that cannot serve as the starter, such as an archived one. They
// are left out instead of blocking the remaining starters.
let mut skipped = std::collections::HashSet::new();
// Starters the relay refused for good, waiting on a second look.
let mut refused = Vec::new();

for spec in STARTER_CHANNELS {
if existing_channels
Expand All @@ -424,7 +452,6 @@ async fn ensure_starter_channels_inner(

let channel_uuid = starter_channel_uuid(&relay_scope, spec.slug);
let channel_uuid_string = channel_uuid.to_string();
starter_ids.push(channel_uuid_string.clone());
let builder = events::build_create_channel(
channel_uuid,
spec.name,
Expand All @@ -442,9 +469,40 @@ async fn ensure_starter_channels_inner(
}
Err(error) if is_duplicate_channel_rejection(&error) => {
state.mark_pending_owned_channel(&creator_pubkey, &channel_uuid_string);
duplicate_starters.push((spec, channel_uuid_string.clone()));
}
// A retry would be refused the same way. The name may still
// belong to a live starter the first read missed, such as one
// created since, so it is looked up again below.
Err(error) if is_permanent_starter_rejection(&error) => {
refused.push(spec);
continue;
}
Err(error) => return Err(error),
}
starter_ids.push(channel_uuid_string);
}

if !refused.is_empty() {
// Read from the writer: the relay refused the create against the
// writer's state, and a lagging replica could miss the same starter
// again.
let mut directory = fetch_open_directory_from_writer(state).await?;
for spec in refused {
match directory
.iter()
.position(|channel| is_matching_starter_channel(channel, spec))
{
Some(index) => {
let channel = directory.swap_remove(index);
existing_channels.retain(|existing| existing.id != channel.id);
existing_channels.push(channel);
}
None => {
skipped.insert(spec.slug);
}
}
}
}

for _ in 0..3 {
Expand All @@ -460,17 +518,37 @@ async fn ensure_starter_channels_inner(
existing_channels.push(channel);
}
}
if has_all_starter_channels(&existing_channels) {
// A live channel at the starter id that cannot serve as the starter
// (renamed or retyped) stays that way, so stop waiting on it.
for (spec, id) in &duplicate_starters {
if existing_channels
.iter()
.any(|channel| channel.id == *id && !is_matching_starter_channel(channel, spec))
{
state.clear_pending_owned_channel(&creator_pubkey, id);
skipped.insert(spec.slug);
}
}
if has_all_starter_channels(&existing_channels, &skipped) {
break;
}
tokio::time::sleep(std::time::Duration::from_millis(150)).await;
}

if !has_all_starter_channels(&existing_channels) {
if !has_all_starter_channels(&existing_channels, &skipped) {
existing_channels = fetch_channels(state, DirectoryScope::IncludeOpenDirectory).await?;
// A starter id the relay holds but still does not serve, even from
// the writer, most likely belongs to a channel this user cannot see,
// such as a deleted one.
for (spec, id) in &duplicate_starters {
if !existing_channels.iter().any(|channel| channel.id == *id) {
state.clear_pending_owned_channel(&creator_pubkey, id);
skipped.insert(spec.slug);
}
}
}

if !has_all_starter_channels(&existing_channels) {
if !has_all_starter_channels(&existing_channels, &skipped) {
return Err("starter channels created but metadata not yet available".to_string());
}

Expand Down
17 changes: 17 additions & 0 deletions desktop/src-tauri/src/commands/channels/fetch.rs
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,23 @@ async fn query_relay_all(
}
}

/// Every open channel's metadata, read from the writer so a lagging replica
/// cannot hide a channel the relay already holds. Membership is not resolved:
/// each channel is reported as not joined.
pub(super) async fn fetch_open_directory_from_writer(
state: &AppState,
) -> Result<Vec<ChannelInfo>, String> {
let events = query_relay_all(
state,
serde_json::json!({"kinds": [39000], "consistency": "strong"}),
)
.await?;
Ok(events
.iter()
.filter_map(|ev| nostr_convert::channel_info_from_event(ev, None, Some(false)).ok())
.collect())
}

/// Whether an open channel not yet in the real member set should still be
/// classified `is_member=true` via the pending-owner overlay. Pulled out of
/// `get_channels`'s open-channel branch so the exact `(d_tag, my_pubkey,
Expand Down
Loading
Loading