Skip to content
Merged
Show file tree
Hide file tree
Changes from 2 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
40 changes: 33 additions & 7 deletions desktop/src-tauri/src/commands/channels.rs
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,9 @@ use crate::{
// ── Reads (pure-nostr via /query) ────────────────────────────────────────────

const DIRECTORY_PAGE_SIZE: usize = 500;
// Keep this aligned with the relay's aggregate explicit-`#h` request bound.
// Each filter carries one channel so the relay can use its channel_id index.
const LAST_MESSAGE_QUERY_CHANNEL_BATCH_SIZE: usize = 128;
const STARTER_CHANNEL_NAMESPACE: uuid::Uuid = uuid::uuid!("3ce33bea-8f09-5f1b-9c85-8a7d2659e6b0");

struct StarterChannelSpec {
Expand Down Expand Up @@ -147,6 +150,23 @@ fn compute_channels_hash(channels: &[ChannelInfo]) -> String {

// ── Core fetch implementation ─────────────────────────────────────────────────

fn last_message_filter_batches(filters: &[serde_json::Value]) -> Vec<&[serde_json::Value]> {
filters
.chunks(LAST_MESSAGE_QUERY_CHANNEL_BATCH_SIZE)
.collect()
}

async fn query_last_messages(
state: &AppState,
filters: &[serde_json::Value],
) -> Result<Vec<nostr::Event>, String> {
let mut messages = Vec::with_capacity(filters.len());
for batch in last_message_filter_batches(filters) {
messages.extend(query_relay(state, batch).await?);
}
Ok(messages)
}

/// Fetch the full channel list from the relay. Called by both `get_channels`
/// (the Tauri command, which wraps the result with hash-based short-circuit
/// logic) and `ensure_starter_channels` (which needs the raw list directly).
Expand All @@ -155,7 +175,9 @@ fn compute_channels_hash(channels: &[ChannelInfo]) -> String {
/// - Phase 1 (parallel): member-chain (kind:39002→kind:39000), open directory
/// (kind:39000 all-open), and hidden-DM snapshot (kind:30622).
/// - Phase 2 (parallel): member counts (kind:39002 batch) and last-message
/// timestamps (per-channel kind:9/40002).
/// timestamps (bounded per-channel kind:9/40002 batches). Member-count
/// failures degrade to zero; timestamp failures abort so cached recency is
/// never replaced by a false authoritative empty result.
async fn fetch_channels(state: &AppState) -> Result<Vec<ChannelInfo>, String> {
#[cfg(debug_assertions)]
let _profile_start = std::time::Instant::now();
Expand Down Expand Up @@ -305,8 +327,8 @@ async fn fetch_channels(state: &AppState) -> Result<Vec<ChannelInfo>, String> {
}

// Phase 2 — concurrent: member counts (step 4) and last-message timestamps
// (step 5). Both tolerate failures — empty defaults leave counts at 0 and
// timestamps at None rather than aborting.
// (step 5). Member-count failures degrade to zero. Timestamp failures
// abort this refresh so the frontend keeps its previous Recent ordering.
let all_channel_ids: Vec<String> = channels.iter().map(|c| c.id.clone()).collect();
if !all_channel_ids.is_empty() {
let last_msg_filters: Vec<serde_json::Value> = all_channel_ids
Expand All @@ -330,10 +352,14 @@ async fn fetch_channels(state: &AppState) -> Result<Vec<ChannelInfo>, String> {
let (members_result, message_result) = tokio::join!(
// Step 4: batch-fetch kind:39002 for member counts.
query_relay(state, &member_count_filters),
// Step 5: per-channel last-message filter. Uses per-channel `#h`
// so the relay can push each query to its indexed channel_id column.
query_relay(state, &last_msg_filters),
// Step 5: preserve one indexed filter per channel while keeping
// every relay request within its aggregate explicit-channel cap.
query_last_messages(state, &last_msg_filters),
);
// Message timestamps drive the user-selected Recent ordering. Unlike
// member counts, a failed query must not masquerade as an authoritative
// empty result and clear every cached timestamp in the frontend.
let messages = message_result?;

let membership = collect_members_by_channel(&members_result.unwrap_or_default());
for channel in &mut channels {
Expand All @@ -345,7 +371,7 @@ async fn fetch_channels(state: &AppState) -> Result<Vec<ChannelInfo>, String> {

let mut last_message_by_channel: std::collections::HashMap<String, u64> =
std::collections::HashMap::new();
for ev in &message_result.unwrap_or_default() {
for ev in &messages {
if let Some(ch_id) = ev.tags.iter().find_map(|t| {
let s = t.as_slice();
(s.len() >= 2 && s[0] == "h").then(|| s[1].clone())
Expand Down
15 changes: 15 additions & 0 deletions desktop/src-tauri/src/commands/channels_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -427,3 +427,18 @@ fn starter_match_requires_open_unarchived_stream_by_normalized_name() {
channel.archived_at = Some("2026-07-16T00:00:00Z".to_string());
assert!(!is_matching_starter_channel(&channel, spec));
}

#[test]
fn last_message_filters_stay_within_relay_channel_cap() {
let filters: Vec<serde_json::Value> = (0..257)
.map(|index| serde_json::json!({"#h": [format!("channel-{index}")]}))
.collect();

let batches = last_message_filter_batches(&filters);

assert_eq!(
batches.iter().map(|batch| batch.len()).collect::<Vec<_>>(),
[128, 128, 1]
);
assert_eq!(batches.concat(), filters);
}
49 changes: 33 additions & 16 deletions desktop/src/features/channels/hooks.ts
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@ import type {
UpdateChannelInput,
} from "@/shared/api/types";
import type { OpenDmInput } from "@/shared/api/tauriChannels";
import { mergeConcurrentChannelRecency } from "@/features/channels/lib/channelRecencyMerge";
import { useIdentityQuery } from "@/shared/api/hooks";
import { useFocusedRefetchInterval } from "@/shared/lib/useDocumentVisible";
import { useCommunities } from "@/features/communities/useCommunities";
Expand Down Expand Up @@ -360,6 +361,8 @@ export function useChannelsQuery(options?: { enabled?: boolean }) {
initialSnapshotPair;
const knownHash = cachedPair?.hash ?? null;

const channelsAtRequestStart =
queryClient.getQueryData<Channel[]>(channelsQueryKey);
const payload = await getChannels(knownHash);

// A not-modified response is usable only when it echoes the exact hash
Expand All @@ -377,13 +380,22 @@ export function useChannelsQuery(options?: { enabled?: boolean }) {
// Missing cache or a mismatched not-modified response: discard the hash
// and fetch a complete authoritative list before updating persistence.
const full = await getChannels(null);
const sorted = sortChannels(
const authoritativeChannels = sortChannels(
applyLastMessages(
requireFullChannelList(full.channels),
full.lastMessages,
),
);
const pair = { channels: sorted, hash: full.hash };
const displayedAtSettlement =
queryClient.getQueryData<Channel[]>(channelsQueryKey);
const sorted = sortChannels(
mergeConcurrentChannelRecency(
authoritativeChannels,
displayedAtSettlement,
channelsAtRequestStart,
),
);
const pair = { channels: authoritativeChannels, hash: full.hash };
queryClient.setQueryData(channelsSnapshotPairKey, pair);
if (relayUrl && ownerPubkey) {
writeChannelSnapshot(relayUrl, ownerPubkey, pair.channels, pair.hash);
Expand All @@ -399,22 +411,27 @@ export function useChannelsQuery(options?: { enabled?: boolean }) {
hash: payload.hash,
};
queryClient.setQueryData(channelsSnapshotPairKey, pair);
// A matching not-modified result must merge timestamps into whatever is
// displayed at completion time. Reading through setQueryData avoids
// clobbering an optimistic mutation that landed while the request ran.
const sorted =
// Merge against the displayed cache at settlement so a newer live
// timestamp cannot be rolled back by an older request result. This is
// required for both full-list and matching not-modified responses.
const displayedAtSettlement =
queryClient.getQueryData<Channel[]>(channelsQueryKey);
const refreshedForDisplay =
payload.channels === null
? (queryClient.setQueryData<Channel[]>(
channelsQueryKey,
(displayedChannels) =>
sortChannels(
applyLastMessages(
displayedChannels ?? authoritativeChannels,
payload.lastMessages,
),
),
) ?? authoritativeChannels)
? sortChannels(
applyLastMessages(
displayedAtSettlement ?? authoritativeChannels,
payload.lastMessages,
),
)
: authoritativeChannels;
const sorted = sortChannels(
mergeConcurrentChannelRecency(
refreshedForDisplay,
displayedAtSettlement,
channelsAtRequestStart,
),
);
if (relayUrl && ownerPubkey) {
writeChannelSnapshot(relayUrl, ownerPubkey, pair.channels, pair.hash);
}
Expand Down
101 changes: 101 additions & 0 deletions desktop/src/features/channels/lib/channelRecency.test.mjs
Original file line number Diff line number Diff line change
@@ -0,0 +1,101 @@
import assert from "node:assert/strict";
import test from "node:test";

import { applyChannelLastMessageAt } from "./channelRecency.ts";
import { mergeConcurrentChannelRecency } from "./channelRecencyMerge.ts";

function makeChannel(id, lastMessageAt = null) {
return { id, lastMessageAt };
}

test("applyChannelLastMessageAt advances only the matching channel", () => {
const general = makeChannel("general", "2026-01-01T00:00:00.000Z");
const design = makeChannel("design", "2026-01-01T00:00:00.000Z");

const result = applyChannelLastMessageAt(
[general, design],
"design",
1_767_225_660,
);

assert.notStrictEqual(result, undefined);
assert.strictEqual(result[0], general);
assert.notStrictEqual(result[1], design);
assert.equal(result[1].lastMessageAt, "2026-01-01T00:01:00.000Z");
});

test("applyChannelLastMessageAt ignores stale or equal timestamps", () => {
const design = makeChannel("design", "2026-01-01T00:01:00.000Z");
const channels = [design];

assert.strictEqual(
applyChannelLastMessageAt(channels, "design", 1_767_225_600),
channels,
);
assert.strictEqual(
applyChannelLastMessageAt(channels, "design", "2026-01-01T00:01:00.000Z"),
channels,
);
});

test("applyChannelLastMessageAt preserves the list for invalid or unknown updates", () => {
const channels = [makeChannel("design")];

assert.strictEqual(
applyChannelLastMessageAt(channels, "design", "not-a-date"),
channels,
);
assert.strictEqual(
applyChannelLastMessageAt(channels, "unknown", 1_767_225_660),
channels,
);
assert.strictEqual(
applyChannelLastMessageAt(undefined, "design", 1_767_225_660),
undefined,
);
});

function mergeRecency(start, displayed, refreshed) {
return mergeConcurrentChannelRecency(
[makeChannel("general", refreshed)],
[makeChannel("general", displayed)],
[makeChannel("general", start)],
)[0];
}

test("mergeConcurrentChannelRecency preserves live recency for both refresh response shapes", () => {
// A full-list response and a not-modified response both reduce to the same
// refreshed/displayed/start inputs before settlement.
for (const responseShape of ["full-list", "not-modified"]) {
const result = mergeRecency(
"2026-01-01T00:00:00Z",
"2026-01-01T00:02:00Z",
"2026-01-01T00:01:00Z",
);
assert.equal(result.lastMessageAt, "2026-01-01T00:02:00Z", responseShape);
}
});

test("mergeConcurrentChannelRecency preserves monotonic and absence semantics", () => {
assert.equal(
mergeRecency(
"2026-01-01T00:02:00Z",
"2026-01-01T00:02:00Z",
"2026-01-01T00:01:00Z",
).lastMessageAt,
"2026-01-01T00:02:00Z",
);
assert.equal(
mergeRecency("2026-01-01T00:01:00Z", "2026-01-01T00:01:00Z", null)
.lastMessageAt,
null,
);
assert.equal(
mergeRecency(
"2026-01-01T00:00:00Z",
"2026-01-01T00:01:00Z",
"2026-01-01T00:02:00Z",
).lastMessageAt,
"2026-01-01T00:02:00Z",
);
});
63 changes: 63 additions & 0 deletions desktop/src/features/channels/lib/channelRecency.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,63 @@
import type { QueryClient } from "@tanstack/react-query";

import { channelsQueryKey } from "@/features/channels/hooks";
import type { Channel } from "@/shared/api/types";

function parseTimestamp(value: number | string | null | undefined) {
if (typeof value === "number") {
return Number.isFinite(value) ? value * 1_000 : null;
}

if (!value) {
return null;
}

const timestamp = Date.parse(value);
return Number.isNaN(timestamp) ? null : timestamp;
}

export function applyChannelLastMessageAt(
current: Channel[] | undefined,
channelId: string,
lastMessageAt: number | string | null | undefined,
): Channel[] | undefined {
if (!current) {
return current;
}

const candidateTimestamp = parseTimestamp(lastMessageAt);
if (candidateTimestamp === null) {
return current;
}

let didUpdate = false;
const normalizedLastMessageAt = new Date(candidateTimestamp).toISOString();
const nextChannels = current.map((channel) => {
if (channel.id !== channelId) {
return channel;
}

const currentTimestamp = parseTimestamp(channel.lastMessageAt);
if (currentTimestamp !== null && candidateTimestamp <= currentTimestamp) {
return channel;
}

didUpdate = true;
return {
...channel,
lastMessageAt: normalizedLastMessageAt,
};
});

return didUpdate ? nextChannels : current;
}

export function updateChannelLastMessageAt(
queryClient: QueryClient,
channelId: string,
lastMessageAt: number | string | null | undefined,
) {
queryClient.setQueryData<Channel[]>(channelsQueryKey, (current) =>
applyChannelLastMessageAt(current, channelId, lastMessageAt),
);
}
39 changes: 39 additions & 0 deletions desktop/src/features/channels/lib/channelRecencyMerge.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,39 @@
export type RecencyChannel = {
id: string;
lastMessageAt: string | null;
};

function timestamp(value: string | null | undefined): number | null {
if (!value) return null;
const parsed = Date.parse(value);
return Number.isNaN(parsed) ? null : parsed;
}

/**
* Keeps recency monotonic when a refresh settles. An authoritative absence may
* clear an unchanged value, but never a live value added during the request.
*/
export function mergeConcurrentChannelRecency<T extends RecencyChannel>(
refreshed: T[],
displayed: T[] | undefined,
atRequestStart: T[] | undefined,
): T[] {
if (!displayed) return refreshed;
const displayedById = new Map(displayed.map((c) => [c.id, c.lastMessageAt]));
const startById = new Map(
atRequestStart?.map((c) => [c.id, c.lastMessageAt]) ?? [],
);

return refreshed.map((channel) => {
const displayedValue = displayedById.get(channel.id);
const displayedAt = timestamp(displayedValue);
const refreshedAt = timestamp(channel.lastMessageAt);
const changed = displayedValue !== startById.get(channel.id);
const keepDisplayed =
displayedAt !== null &&
(refreshedAt !== null ? displayedAt > refreshedAt : changed);
return keepDisplayed
? { ...channel, lastMessageAt: displayedValue ?? null }
: channel;
});
}
Loading
Loading