From 223e9b3a2fdae4576a540fd7cf6cc38b9e802d0b Mon Sep 17 00:00:00 2001 From: onenewcode Date: Thu, 30 Jul 2026 15:56:56 +0800 Subject: [PATCH] fix(redis): paginate stream entries --- .../RedisValueViewer.streamMonitoring.spec.ts | 31 +- .../src/components/redis/RedisValueViewer.vue | 65 +++- apps/desktop/src/i18n/locales/en.ts | 1 + apps/desktop/src/i18n/locales/es.ts | 1 + apps/desktop/src/i18n/locales/it.ts | 1 + apps/desktop/src/i18n/locales/ja.ts | 1 + apps/desktop/src/i18n/locales/ko.ts | 1 + apps/desktop/src/i18n/locales/pt-BR.ts | 1 + apps/desktop/src/i18n/locales/zh-CN.ts | 1 + apps/desktop/src/i18n/locales/zh-TW.ts | 1 + .../redis/redisValuePresentation.spec.ts | 32 ++ apps/desktop/src/lib/backend/api.ts | 2 + apps/desktop/src/lib/backend/http.ts | 5 + apps/desktop/src/lib/backend/tauri.ts | 11 +- .../src/lib/redis/redisValuePresentation.ts | 2 +- crates/dbx-core/src/db/redis_driver.rs | 283 +++++++++++++++++- crates/dbx-core/src/redis_ops.rs | 32 +- crates/dbx-web/src/main.rs | 1 + crates/dbx-web/src/routes/redis.rs | 25 ++ src-tauri/src/commands/redis_cmd.rs | 14 +- src-tauri/src/lib.rs | 1 + 21 files changed, 479 insertions(+), 33 deletions(-) diff --git a/apps/desktop/src/components/redis/RedisValueViewer.streamMonitoring.spec.ts b/apps/desktop/src/components/redis/RedisValueViewer.streamMonitoring.spec.ts index 789ab776b..e2b28e588 100644 --- a/apps/desktop/src/components/redis/RedisValueViewer.streamMonitoring.spec.ts +++ b/apps/desktop/src/components/redis/RedisValueViewer.streamMonitoring.spec.ts @@ -6,6 +6,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; const mocks = vi.hoisted(() => ({ redisGetValue: vi.fn(), + redisGetStreamEntries: vi.fn(), redisGetStreamGroups: vi.fn(), redisGetStreamConsumers: vi.fn(), redisGetStreamPending: vi.fn(), @@ -16,6 +17,7 @@ const mocks = vi.hoisted(() => ({ vi.mock("@/lib/backend/api", () => ({ redisGetValue: mocks.redisGetValue, + redisGetStreamEntries: mocks.redisGetStreamEntries, redisGetStreamGroups: mocks.redisGetStreamGroups, redisGetStreamConsumers: mocks.redisGetStreamConsumers, redisGetStreamPending: mocks.redisGetStreamPending, @@ -40,11 +42,7 @@ vi.mock("vue-virtual-scroller", async () => { const DynamicScroller = defineComponent({ props: { items: { type: Array, default: () => [] } }, setup(props, { slots }) { - return () => - h( - "div", - (props.items as unknown[]).map((item, index) => slots.default?.({ item, active: true, index })), - ); + return () => h("div", [...(props.items as unknown[]).map((item, index) => slots.default?.({ item, active: true, index })), ...(slots.after?.() ?? [])]); }, }); const DynamicScrollerItem = defineComponent({ @@ -84,13 +82,17 @@ async function settle() { await nextTick(); } -function streamValue() { +function streamEntry(id: string) { + return { id, fields: [{ field: "event", value: "login" }] }; +} + +function streamValue(entries = [] as ReturnType[], nextCursor?: string, total = entries.length) { return { key_display: "orders", key_raw: "b3JkZXJz", ttl: -1, redis_type: "stream", - data: { kind: "stream" as const, entries: [] }, + data: { kind: "stream" as const, entries, total, ...(nextCursor ? { next_cursor: nextCursor } : {}) }, }; } @@ -158,6 +160,21 @@ function openGroups(host: HTMLElement) { } describe("RedisValueViewer stream monitoring", () => { + it("loads more Stream entries from the returned cursor", async () => { + mocks.redisGetValue.mockResolvedValue(streamValue([streamEntry("1714470000000-0")], "1714470000000-0")); + mocks.redisGetStreamEntries.mockResolvedValue({ entries: [streamEntry("1714470000001-0")] }); + const host = mountViewer(); + await settle(); + + expect(host.querySelectorAll("[data-redis-stream-entry]")).toHaveLength(1); + host.querySelector("[data-redis-stream-entries-more]")!.click(); + await settle(); + + expect(mocks.redisGetStreamEntries).toHaveBeenCalledWith("redis-1", 2, "b3JkZXJz", "1714470000000-0"); + expect(host.querySelectorAll("[data-redis-stream-entry]")).toHaveLength(2); + expect(host.querySelector("[data-redis-stream-entries-more]")).toBeNull(); + }); + it("renders large counter values transported as decimal strings", async () => { const unsafeMetric = "9007199254740992"; mocks.redisGetValue.mockResolvedValue(streamValue()); diff --git a/apps/desktop/src/components/redis/RedisValueViewer.vue b/apps/desktop/src/components/redis/RedisValueViewer.vue index 20261a3d3..404cb0876 100644 --- a/apps/desktop/src/components/redis/RedisValueViewer.vue +++ b/apps/desktop/src/components/redis/RedisValueViewer.vue @@ -82,6 +82,9 @@ const loading = ref(false); const loadingMore = ref(false); let loadRequestId = 0; const streamTab = ref<"entries" | "groups">("entries"); +const streamEntries = ref([]); +const streamEntriesCursor = ref(); +const streamEntriesLoadingMore = ref(false); const streamGroups = ref([]); const streamGroupsLoaded = ref(false); const streamGroupsLoading = ref(false); @@ -112,6 +115,7 @@ let streamGroupsRequestId = 0; let streamGroupDetailRequestId = 0; let streamConsumersRequestId = 0; let streamPendingRequestId = 0; +let streamEntriesRequestId = 0; const editValue = ref(""); const savingString = ref(false); const savingJson = ref(false); @@ -316,8 +320,8 @@ const metadataSizeLabel = computed(() => { return String(size); }); const streamRows = computed(() => { - if (data.value?.data.kind !== "stream") return []; - return data.value.data.entries.map((entry, index) => ({ + if (redisKind.value !== "stream") return []; + return streamEntries.value.map((entry, index) => ({ id: `${index}:${entry.id}`, index, entry, @@ -434,6 +438,25 @@ type RedisStreamRow = { entry: RedisStreamEntry; }; +function replaceStreamEntries(value: RedisValue) { + streamEntriesRequestId++; + streamEntriesLoadingMore.value = false; + if (value.data.kind === "stream") { + streamEntries.value = [...value.data.entries]; + streamEntriesCursor.value = value.data.next_cursor; + return; + } + streamEntries.value = []; + streamEntriesCursor.value = undefined; +} + +function resetStreamEntries() { + streamEntriesRequestId++; + streamEntries.value = []; + streamEntriesCursor.value = undefined; + streamEntriesLoadingMore.value = false; +} + function isSelectedStreamGroup(group: RedisStreamGroup, requestId = streamGroupDetailRequestId): boolean { return requestId === streamGroupDetailRequestId && selectedStreamGroup.value?.name.raw_base64 === group.name.raw_base64; } @@ -464,6 +487,25 @@ function resetStreamMonitoring() { streamGroupsError.value = ""; } +async function loadMoreStreamEntries() { + const cursor = streamEntriesCursor.value; + if (redisKind.value !== "stream" || !cursor || loading.value || streamEntriesLoadingMore.value) return; + + const requestId = ++streamEntriesRequestId; + streamEntriesLoadingMore.value = true; + try { + const page = await api.redisGetStreamEntries(props.connectionId, props.db, props.keyRaw, cursor); + if (requestId !== streamEntriesRequestId || redisKind.value !== "stream" || streamEntriesCursor.value !== cursor) return; + + streamEntries.value = [...streamEntries.value, ...page.entries]; + streamEntriesCursor.value = page.next_cursor; + } catch (error) { + if (requestId === streamEntriesRequestId) toast(errorMessage(error), 3000); + } finally { + if (requestId === streamEntriesRequestId) streamEntriesLoadingMore.value = false; + } +} + async function loadStreamGroups(force = false): Promise { if (redisKind.value !== "stream") return false; if (!force && (streamGroupsLoaded.value || streamGroupsLoading.value)) return streamGroupsLoaded.value; @@ -902,6 +944,7 @@ async function load(options: { selectDefaultMember?: boolean; preserveDraft?: bo data.value = null; collectionItems.value = []; scanCursor.value = undefined; + resetStreamEntries(); resetStreamMonitoring(); stopAutoRefresh(); emit("deleted", props.keyRaw); @@ -929,6 +972,7 @@ async function load(options: { selectDefaultMember?: boolean; preserveDraft?: bo emit("loaded", loadedValue); scanCursor.value = redisValueCollectionScanCursor(loadedValue); collectionItems.value = redisValueCollectionItems(loadedValue); + replaceStreamEntries(loadedValue); if (loadedValue.data.kind !== "stream") resetStreamMonitoring(); // A foreground load replaces the current value, so it also starts a new @@ -1050,7 +1094,8 @@ function requestDeleteKey() { async function copyValue() { if (!data.value) return; - const text = redisValueCopyText(data.value, collectionItems.value); + const value = data.value.data.kind === "stream" ? { ...data.value, data: { ...data.value.data, entries: streamEntries.value } } : data.value; + const text = redisValueCopyText(value, collectionItems.value); try { await copyToClipboard(text); toast(t("redis.copied"), 2000); @@ -1129,7 +1174,7 @@ function generateInsertStatements(): string | null { break; } case "stream": { - for (const entry of data.value.data.entries) { + for (const entry of streamEntries.value) { const fields = entry.fields.map(({ field, value }) => `${escapeRedisArg(field)} ${escapeRedisArg(value)}`).join(" "); commands.push(`XADD ${escapeRedisArg(key)} * ${fields}`); } @@ -1772,6 +1817,7 @@ watch( () => { resetValueSearch(); valueViewerSearchActive.value = false; + resetStreamEntries(); resetStreamMonitoring(); }, ); @@ -2227,9 +2273,6 @@ defineExpose({ focusSearch }); -
- {{ t("redis.entries", { count: streamRows.length }) }} -
+ diff --git a/apps/desktop/src/i18n/locales/en.ts b/apps/desktop/src/i18n/locales/en.ts index 7bde841d0..5a3ded4cb 100644 --- a/apps/desktop/src/i18n/locales/en.ts +++ b/apps/desktop/src/i18n/locales/en.ts @@ -3069,6 +3069,7 @@ export default { noPendingEntries: "No pending entries", entryId: "Entry ID", deliveries: "Deliveries", + loadMoreEntries: "Load more entries", loadMorePending: "Load more pending entries", noExpiry: "no expiry", expiry: "Expiration", diff --git a/apps/desktop/src/i18n/locales/es.ts b/apps/desktop/src/i18n/locales/es.ts index fded4174a..2dc93b8a0 100644 --- a/apps/desktop/src/i18n/locales/es.ts +++ b/apps/desktop/src/i18n/locales/es.ts @@ -2930,6 +2930,7 @@ export default withEnglishFallback({ noPendingEntries: "No hay entradas pendientes", entryId: "ID de entrada", deliveries: "Entregas", + loadMoreEntries: "Cargar más entradas", loadMorePending: "Cargar más entradas pendientes", noExpiry: "sin expiración", expiry: "Expiración", diff --git a/apps/desktop/src/i18n/locales/it.ts b/apps/desktop/src/i18n/locales/it.ts index dbbeb4ac2..000f31ee9 100644 --- a/apps/desktop/src/i18n/locales/it.ts +++ b/apps/desktop/src/i18n/locales/it.ts @@ -2928,6 +2928,7 @@ export default withEnglishFallback({ noPendingEntries: "Nessuna voce in sospeso", entryId: "ID voce", deliveries: "Consegne", + loadMoreEntries: "Carica altre voci", loadMorePending: "Carica altre voci in sospeso", noExpiry: "nessuna scadenza", expiry: "Scadenza", diff --git a/apps/desktop/src/i18n/locales/ja.ts b/apps/desktop/src/i18n/locales/ja.ts index 44966cfe8..8afe4464c 100644 --- a/apps/desktop/src/i18n/locales/ja.ts +++ b/apps/desktop/src/i18n/locales/ja.ts @@ -2929,6 +2929,7 @@ export default withEnglishFallback({ noPendingEntries: "保留中のエントリはありません", entryId: "エントリ ID", deliveries: "配信回数", + loadMoreEntries: "さらにエントリを読み込む", loadMorePending: "保留中のエントリをさらに読み込む", noExpiry: "期限なし", expiry: "有効期限", diff --git a/apps/desktop/src/i18n/locales/ko.ts b/apps/desktop/src/i18n/locales/ko.ts index 4755a1fa9..e41e80dd3 100644 --- a/apps/desktop/src/i18n/locales/ko.ts +++ b/apps/desktop/src/i18n/locales/ko.ts @@ -3032,6 +3032,7 @@ export default withEnglishFallback({ noPendingEntries: "대기 항목이 없습니다", entryId: "항목 ID", deliveries: "전달", + loadMoreEntries: "항목 더 불러오기", loadMorePending: "대기 항목 더 불러오기", noExpiry: "만료 없음", expiry: "만료", diff --git a/apps/desktop/src/i18n/locales/pt-BR.ts b/apps/desktop/src/i18n/locales/pt-BR.ts index d47b0b982..eab0472f6 100644 --- a/apps/desktop/src/i18n/locales/pt-BR.ts +++ b/apps/desktop/src/i18n/locales/pt-BR.ts @@ -2930,6 +2930,7 @@ export default withEnglishFallback({ noPendingEntries: "Nenhuma entrada pendente", entryId: "ID da entrada", deliveries: "Entregas", + loadMoreEntries: "Carregar mais entradas", loadMorePending: "Carregar mais entradas pendentes", noExpiry: "sem expiração", expiry: "Expiração", diff --git a/apps/desktop/src/i18n/locales/zh-CN.ts b/apps/desktop/src/i18n/locales/zh-CN.ts index 43d8d7154..a97c750f6 100644 --- a/apps/desktop/src/i18n/locales/zh-CN.ts +++ b/apps/desktop/src/i18n/locales/zh-CN.ts @@ -3069,6 +3069,7 @@ export default withEnglishFallback({ noPendingEntries: "暂无待处理条目", entryId: "条目 ID", deliveries: "投递次数", + loadMoreEntries: "加载更多条目", loadMorePending: "加载更多待处理条目", noExpiry: "永不过期", expiry: "过期时间", diff --git a/apps/desktop/src/i18n/locales/zh-TW.ts b/apps/desktop/src/i18n/locales/zh-TW.ts index ed358081a..0bc359e3c 100644 --- a/apps/desktop/src/i18n/locales/zh-TW.ts +++ b/apps/desktop/src/i18n/locales/zh-TW.ts @@ -2601,6 +2601,7 @@ export default withEnglishFallback({ noPendingEntries: "暫無待處理項目", entryId: "項目 ID", deliveries: "投遞次數", + loadMoreEntries: "載入更多項目", loadMorePending: "載入更多待處理項目", noExpiry: "永不過期", expiry: "到期時間", diff --git a/apps/desktop/src/lib/__tests__/redis/redisValuePresentation.spec.ts b/apps/desktop/src/lib/__tests__/redis/redisValuePresentation.spec.ts index ec31e12b6..681509ea9 100644 --- a/apps/desktop/src/lib/__tests__/redis/redisValuePresentation.spec.ts +++ b/apps/desktop/src/lib/__tests__/redis/redisValuePresentation.spec.ts @@ -146,6 +146,38 @@ describe("redisValuePresentation", () => { }`); }); + it("uses the Stream's Redis length instead of the loaded page size", () => { + const value = { + key_display: "orders", + key_raw: "b3JkZXJz", + ttl: -1, + redis_type: "stream", + data: { + kind: "stream" as const, + entries: [{ id: "1714470000000-0", fields: [{ field: "event", value: "login" }] }], + total: 177, + next_cursor: "1714470000000-0", + }, + }; + + expect(redisValueSize(value)).toBe(177); + }); + + it("uses loaded Stream entries when the total is unavailable", () => { + const value = { + key_display: "orders", + key_raw: "b3JkZXJz", + ttl: -1, + redis_type: "stream", + data: { + kind: "stream" as const, + entries: [{ id: "1714470000000-0", fields: [{ field: "event", value: "login" }] }], + }, + }; + + expect(redisValueSize(value)).toBe(1); + }); + it("labels raw text views by encoding instead of generic raw text", () => { expect(formatRedisMemberDetail("plain-text").rawLabel).toBe("ASCII"); expect( diff --git a/apps/desktop/src/lib/backend/api.ts b/apps/desktop/src/lib/backend/api.ts index 7d6e64850..ab1c6658e 100644 --- a/apps/desktop/src/lib/backend/api.ts +++ b/apps/desktop/src/lib/backend/api.ts @@ -392,6 +392,7 @@ export const redisScanKeys = forward("redisScanKeys"); export const redisScanKeysBatch = forward("redisScanKeysBatch"); export const redisScanValues = forward("redisScanValues"); export const redisGetValue = forward("redisGetValue"); +export const redisGetStreamEntries = forward("redisGetStreamEntries"); export const redisGetStreamGroups = forward("redisGetStreamGroups"); export const redisGetStreamConsumers = forward("redisGetStreamConsumers"); export const redisGetStreamPending = forward("redisGetStreamPending"); @@ -640,6 +641,7 @@ export type { RedisStreamField, RedisStreamGroup, RedisStreamMetric, + RedisStreamPage, RedisStreamPendingEntry, RedisStreamPendingPage, RedisValue, diff --git a/apps/desktop/src/lib/backend/http.ts b/apps/desktop/src/lib/backend/http.ts index c9813ce3b..0a4ff1dc1 100644 --- a/apps/desktop/src/lib/backend/http.ts +++ b/apps/desktop/src/lib/backend/http.ts @@ -70,6 +70,7 @@ import type { RedisDatabaseInfo, RedisStreamConsumer, RedisStreamGroup, + RedisStreamPage, RedisStreamPendingPage, RedisValue, RedisScanResult, @@ -2253,6 +2254,10 @@ export async function redisGetValue(connectionId: string, db: number, keyRaw: st return post("/api/redis/get-value", { connectionId, db, keyRaw }); } +export async function redisGetStreamEntries(connectionId: string, db: number, keyRaw: string, cursor?: string): Promise { + return post("/api/redis/get-stream-entries", { connectionId, db, keyRaw, cursor }); +} + export async function redisGetStreamGroups(connectionId: string, db: number, keyRaw: string): Promise { return post("/api/redis/get-stream-groups", { connectionId, db, keyRaw }); } diff --git a/apps/desktop/src/lib/backend/tauri.ts b/apps/desktop/src/lib/backend/tauri.ts index 3a58b2c1e..f1b52c00c 100644 --- a/apps/desktop/src/lib/backend/tauri.ts +++ b/apps/desktop/src/lib/backend/tauri.ts @@ -1949,6 +1949,11 @@ export interface RedisStreamEntry { fields: RedisStreamField[]; } +export interface RedisStreamPage { + entries: RedisStreamEntry[]; + next_cursor?: string; +} + // Redis counters above Number.MAX_SAFE_INTEGER are transported as decimal strings. export type RedisStreamMetric = number | string; @@ -2002,7 +2007,7 @@ export type RedisValueData = total: number; scan_cursor?: number; } - | { kind: "stream"; entries: RedisStreamEntry[] } + | { kind: "stream"; entries: RedisStreamEntry[]; total?: number; next_cursor?: string } | { kind: "unknown" }; export interface RedisValue { @@ -2085,6 +2090,10 @@ export async function redisGetValue(connectionId: string, db: number, keyRaw: st return invoke("redis_get_value", { connectionId, db, keyRaw }); } +export async function redisGetStreamEntries(connectionId: string, db: number, keyRaw: string, cursor?: string): Promise { + return invoke("redis_get_stream_entries", { connectionId, db, keyRaw, cursor }); +} + export async function redisGetStreamGroups(connectionId: string, db: number, keyRaw: string): Promise { return invoke("redis_get_stream_groups", { connectionId, db, keyRaw }); } diff --git a/apps/desktop/src/lib/redis/redisValuePresentation.ts b/apps/desktop/src/lib/redis/redisValuePresentation.ts index 2c0072dca..bc42e25f7 100644 --- a/apps/desktop/src/lib/redis/redisValuePresentation.ts +++ b/apps/desktop/src/lib/redis/redisValuePresentation.ts @@ -327,7 +327,7 @@ export function redisValueSize(value: RedisValue): number { case "zset": return value.data.total; case "stream": - return value.data.entries.length; + return value.data.total ?? value.data.entries.length; default: return 0; } diff --git a/crates/dbx-core/src/db/redis_driver.rs b/crates/dbx-core/src/db/redis_driver.rs index f80ae4a01..9ee53a016 100644 --- a/crates/dbx-core/src/db/redis_driver.rs +++ b/crates/dbx-core/src/db/redis_driver.rs @@ -14,7 +14,7 @@ use tokio::sync::{Mutex, MutexGuard}; use super::json_value_for_js; -const STREAM_ENTRY_LIMIT: usize = 100; +const STREAM_ENTRY_PAGE_SIZE: usize = 50; const STREAM_PENDING_PAGE_SIZE: usize = 100; const COLLECTION_PAGE_SIZE: usize = 200; const HASH_FILTER_SCAN_MAX_ITERATIONS: usize = 10; @@ -124,6 +124,13 @@ pub struct RedisStreamEntry { pub fields: Vec, } +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct RedisStreamPage { + pub entries: Vec, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub next_cursor: Option, +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct RedisStreamGroup { pub name: RedisBlob, @@ -224,6 +231,14 @@ pub enum RedisValueData { }, Stream { entries: Vec, + #[serde( + default, + skip_serializing_if = "Option::is_none", + serialize_with = "serialize_optional_redis_u64_for_js" + )] + total: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + next_cursor: Option, }, Unknown, } @@ -2281,7 +2296,14 @@ where { let redis_type: String = redis::cmd("TYPE").arg(key).query_async(con).await.map_err(|e| e.to_string())?; - let ttl: i64 = redis::cmd("TTL").arg(key).query_async(con).await.unwrap_or(-1); + // For Streams, fetch metadata and the initial page together after TYPE. + // This keeps the accurate XLEN without adding another Redis round trip. + let stream_initial_page = + if redis_type == "stream" { Some(get_stream_initial_page(con, key).await?) } else { None }; + let ttl: i64 = match stream_initial_page.as_ref() { + Some((ttl, _, _)) => *ttl, + None => redis::cmd("TTL").arg(key).query_async(con).await.unwrap_or(-1), + }; let data = match redis_type.as_str() { "string" => { @@ -2315,7 +2337,10 @@ where let (cursor, items) = hscan_page_raw(con, key, 0, COLLECTION_PAGE_SIZE, None).await?; RedisValueData::Hash { items, total: len, scan_cursor: (cursor > 0).then_some(cursor) } } - "stream" => RedisValueData::Stream { entries: get_stream_entries(con, key).await? }, + "stream" => { + let (_, total, page) = stream_initial_page.expect("Stream page is loaded with its Stream type"); + RedisValueData::Stream { entries: page.entries, total, next_cursor: page.next_cursor } + } key_type if is_redis_json_type(key_type) => { let raw: RedisRawValue = redis::cmd("JSON.GET").arg(key).query_async(con).await.map_err(|e| e.to_string())?; @@ -2370,7 +2395,7 @@ fn redis_search_value_text(value: &RedisValueData) -> String { .flat_map(|item| [item.score.clone(), redis_blob_display_text(&item.member)]) .collect::>() .join(" "), - RedisValueData::Stream { entries } => entries + RedisValueData::Stream { entries, .. } => entries .iter() .flat_map(|entry| { entry.fields.iter().flat_map(|field| [field.field.clone(), field.value.clone()]).collect::>() @@ -2403,26 +2428,77 @@ fn redis_search_value_size(value: &RedisValue) -> u64 { | RedisValueData::Set { total, .. } | RedisValueData::Hash { total, .. } | RedisValueData::Zset { total, .. } => *total, - RedisValueData::Stream { entries } => entries.len() as u64, + RedisValueData::Stream { entries, total, .. } => total.unwrap_or(entries.len() as u64), RedisValueData::Unknown => 0, } } -async fn get_stream_entries(con: &mut C, key: &[u8]) -> Result, String> +pub async fn get_stream_entries_page( + con: &mut C, + key: &[u8], + cursor: Option<&str>, +) -> Result where C: ConnectionLike + Send + Sync + Unpin, { + let cursor = cursor.filter(|value| !value.is_empty()); + let start = cursor.unwrap_or("-"); + // XRANGE starts inclusively. Fetch a duplicate candidate and a lookahead + // row so this works with Redis 5 and does not skip records after trimming. + let requested_count = stream_entry_page_request_count(cursor); let raw: RedisRawValue = redis::cmd("XRANGE") .arg(key) - .arg("-") + .arg(start) .arg("+") .arg("COUNT") - .arg(STREAM_ENTRY_LIMIT) + .arg(requested_count) .query_async(con) .await .map_err(|e| e.to_string())?; - Ok(parse_stream_entries(raw)) + Ok(stream_entries_page_from_raw(raw, cursor)) +} + +async fn get_stream_initial_page(con: &mut C, key: &[u8]) -> Result<(i64, Option, RedisStreamPage), String> +where + C: ConnectionLike + Send + Sync + Unpin, +{ + let mut pipe = redis::pipe(); + pipe.cmd("TTL").arg(key); + pipe.cmd("XLEN").arg(key); + pipe.cmd("XRANGE").arg(key).arg("-").arg("+").arg("COUNT").arg(stream_entry_page_request_count(None)); + match pipe.query_async::<(i64, u64, RedisRawValue)>(con).await { + Ok((ttl, total, raw)) => Ok((ttl, Some(total), stream_entries_page_from_raw(raw, None))), + Err(_) => { + // An ACL may allow XRANGE while rejecting the optional XLEN + // metadata. Keep the Stream readable and omit the unknown size. + let ttl: i64 = redis::cmd("TTL").arg(key).query_async(con).await.unwrap_or(-1); + let total = redis::cmd("XLEN").arg(key).query_async::(con).await.ok(); + let page = get_stream_entries_page(con, key, None).await?; + Ok((ttl, total, page)) + } + } +} + +fn stream_entry_page_request_count(cursor: Option<&str>) -> usize { + STREAM_ENTRY_PAGE_SIZE + 1 + usize::from(cursor.is_some()) +} + +fn stream_entries_page_from_raw(raw: RedisRawValue, cursor: Option<&str>) -> RedisStreamPage { + let mut entries = parse_stream_entries(raw); + if let Some(cursor) = cursor { + if entries.first().is_some_and(|entry| entry.id == cursor) { + entries.remove(0); + } + } + let next_cursor = if entries.len() > STREAM_ENTRY_PAGE_SIZE { + entries.truncate(STREAM_ENTRY_PAGE_SIZE); + entries.last().map(|entry| entry.id.clone()) + } else { + None + }; + + RedisStreamPage { entries, next_cursor } } pub async fn get_stream_groups(con: &mut C, key: &[u8]) -> Result, String> @@ -3216,11 +3292,15 @@ mod tests { fn req_packed_commands<'a>( &'a mut self, - _cmd: &'a Pipeline, + cmd: &'a Pipeline, _offset: usize, - _count: usize, + count: usize, ) -> RedisFuture<'a, Vec> { - Box::pin(async move { Ok(Vec::new()) }) + self.commands.push(String::from_utf8_lossy(&cmd.get_packed_pipeline()).into_owned()); + let responses = (0..count) + .map(|_| self.responses.pop_front().unwrap_or(Ok(RedisRawValue::Nil))) + .collect::>>(); + Box::pin(async move { responses }) } fn get_db(&self) -> i64 { @@ -3248,11 +3328,15 @@ mod tests { fn req_packed_commands<'a>( &'a mut self, - _cmd: &'a Pipeline, + cmd: &'a Pipeline, _offset: usize, - _count: usize, + count: usize, ) -> RedisFuture<'a, Vec> { - Box::pin(async move { Ok(Vec::new()) }) + self.commands.lock().unwrap().push(String::from_utf8_lossy(&cmd.get_packed_pipeline()).into_owned()); + let responses = (0..count) + .map(|_| self.responses.pop_front().unwrap_or(Ok(RedisRawValue::Nil))) + .collect::>>(); + Box::pin(async move { responses }) } fn get_db(&self) -> i64 { @@ -3514,6 +3598,175 @@ mod tests { assert_eq!(entry_json["deliveries"], unsafe_value.to_string()); } + #[tokio::test] + async fn stream_value_uses_xlen_for_its_total_size() { + let entries = RedisRawValue::Array(vec![RedisRawValue::Array(vec![ + bulk("1714470000000-0"), + RedisRawValue::Array(vec![bulk("event"), bulk("login")]), + ])]); + let mut con = + FakeRedisConnection::new(vec![bulk("stream"), RedisRawValue::Int(-1), RedisRawValue::Int(177), entries]); + + let value = super::get_value(&mut con, b"orders").await.unwrap(); + + let RedisValueData::Stream { entries, total, next_cursor } = &value.data else { + panic!("expected a Stream value"); + }; + assert_eq!(*total, Some(177)); + assert_eq!(entries.len(), 1); + assert!(next_cursor.is_none()); + assert_eq!(super::redis_search_value_size(&value), 177); + assert_eq!(con.command_count("TYPE"), 1); + assert_eq!(con.command_count("TTL"), 1); + assert_eq!(con.command_count("XLEN"), 1); + assert_eq!(con.command_count("XRANGE"), 1); + assert_eq!(con.commands.len(), 2); + } + + #[tokio::test] + async fn stream_value_still_loads_when_xlen_is_not_permitted() { + let entries = RedisRawValue::Array(vec![RedisRawValue::Array(vec![ + bulk("1714470000000-0"), + RedisRawValue::Array(vec![bulk("event"), bulk("login")]), + ])]); + let xlen_denied = || { + redis::make_extension_error( + "NOPERM".to_string(), + Some("this user has no permissions to run the 'xlen' command".to_string()), + ) + }; + let mut con = FakeRedisConnection::with_results(vec![ + Ok(bulk("stream")), + Ok(RedisRawValue::Int(-1)), + Err(xlen_denied()), + Ok(RedisRawValue::Int(-1)), + Err(xlen_denied()), + Ok(entries), + ]); + + let value = super::get_value(&mut con, b"orders").await.unwrap(); + + let RedisValueData::Stream { entries, total, .. } = &value.data else { + panic!("expected a Stream value"); + }; + assert_eq!(entries.len(), 1); + assert_eq!(*total, None); + assert_eq!(super::redis_search_value_size(&value), 1); + assert!(serde_json::to_value(&value).unwrap()["data"].get("total").is_none()); + assert_eq!(con.command_count("XLEN"), 2); + assert_eq!(con.command_count("XRANGE"), 2); + } + + #[tokio::test] + async fn stream_entry_first_page_exposes_a_cursor_when_more_entries_exist() { + let entries = RedisRawValue::Array( + (0..=50) + .map(|index| { + RedisRawValue::Array(vec![ + bulk(&format!("1714470000000-{index}")), + RedisRawValue::Array(vec![bulk("event"), bulk("login")]), + ]) + }) + .collect(), + ); + let mut con = FakeRedisConnection::new(vec![entries]); + + let page = super::get_stream_entries_page(&mut con, b"orders", None).await.unwrap(); + + assert_eq!(page.entries.len(), 50); + assert_eq!(page.entries.first().map(|entry| entry.id.as_str()), Some("1714470000000-0")); + assert_eq!(page.entries.last().map(|entry| entry.id.as_str()), Some("1714470000000-49")); + assert_eq!(page.next_cursor.as_deref(), Some("1714470000000-49")); + assert_eq!(con.command_count("XRANGE"), 1); + assert!(con.commands[0].contains("\r\n-\r\n")); + assert!(con.commands[0].contains("\r\n51\r\n")); + } + + #[tokio::test] + async fn stream_entry_pagination_reads_all_entries_across_sequential_pages() { + let stream_entries = |start: u64, end: u64| { + RedisRawValue::Array( + (start..=end) + .map(|index| { + RedisRawValue::Array(vec![ + bulk(&format!("1714470000000-{index}")), + RedisRawValue::Array(vec![bulk("event"), bulk("login")]), + ]) + }) + .collect(), + ) + }; + let mut con = + FakeRedisConnection::new(vec![stream_entries(0, 50), stream_entries(49, 100), stream_entries(99, 120)]); + + let first = super::get_stream_entries_page(&mut con, b"orders", None).await.unwrap(); + let second = super::get_stream_entries_page(&mut con, b"orders", first.next_cursor.as_deref()).await.unwrap(); + let third = super::get_stream_entries_page(&mut con, b"orders", second.next_cursor.as_deref()).await.unwrap(); + + let ids = first + .entries + .iter() + .chain(&second.entries) + .chain(&third.entries) + .map(|entry| entry.id.clone()) + .collect::>(); + let expected = (0..=120).map(|index| format!("1714470000000-{index}")).collect::>(); + + assert_eq!(first.next_cursor.as_deref(), Some("1714470000000-49")); + assert_eq!(second.next_cursor.as_deref(), Some("1714470000000-99")); + assert_eq!(third.next_cursor, None); + assert_eq!(ids, expected); + assert_eq!(con.command_count("XRANGE"), 3); + } + + #[tokio::test] + async fn stream_entry_pagination_uses_a_lookahead_and_skips_the_duplicate_cursor() { + let entries = RedisRawValue::Array( + (17..=68) + .map(|index| { + RedisRawValue::Array(vec![ + bulk(&format!("1714470000000-{index}")), + RedisRawValue::Array(vec![bulk("event"), bulk("login")]), + ]) + }) + .collect(), + ); + let mut con = FakeRedisConnection::new(vec![entries]); + + let page = super::get_stream_entries_page(&mut con, b"orders", Some("1714470000000-17")).await.unwrap(); + + assert_eq!(page.entries.len(), 50); + assert_eq!(page.entries.first().map(|entry| entry.id.as_str()), Some("1714470000000-18")); + assert_eq!(page.entries.last().map(|entry| entry.id.as_str()), Some("1714470000000-67")); + assert_eq!(page.next_cursor.as_deref(), Some("1714470000000-67")); + assert_eq!(con.command_count("XRANGE"), 1); + assert!(con.commands[0].contains("\r\n1714470000000-17\r\n")); + assert!(!con.commands[0].contains("\r\n(1714470000000-17\r\n")); + assert!(con.commands[0].contains("\r\n52\r\n")); + } + + #[tokio::test] + async fn stream_entry_pagination_keeps_the_first_entry_when_the_cursor_was_trimmed() { + let entries = RedisRawValue::Array( + (18..=68) + .map(|index| { + RedisRawValue::Array(vec![ + bulk(&format!("1714470000000-{index}")), + RedisRawValue::Array(vec![bulk("event"), bulk("login")]), + ]) + }) + .collect(), + ); + let mut con = FakeRedisConnection::new(vec![entries]); + + let page = super::get_stream_entries_page(&mut con, b"orders", Some("1714470000000-17")).await.unwrap(); + + assert_eq!(page.entries.len(), 50); + assert_eq!(page.entries.first().map(|entry| entry.id.as_str()), Some("1714470000000-18")); + assert_eq!(page.entries.last().map(|entry| entry.id.as_str()), Some("1714470000000-67")); + assert_eq!(page.next_cursor.as_deref(), Some("1714470000000-67")); + } + #[tokio::test] async fn stream_monitoring_commands_are_read_only_and_pending_pagination_supports_redis_5() { let groups = RedisRawValue::Array(vec![RedisRawValue::Array(vec![ diff --git a/crates/dbx-core/src/redis_ops.rs b/crates/dbx-core/src/redis_ops.rs index d7635f70d..ff4887354 100644 --- a/crates/dbx-core/src/redis_ops.rs +++ b/crates/dbx-core/src/redis_ops.rs @@ -1,7 +1,7 @@ use crate::connection::{AppState, PoolKind}; use crate::db::redis_driver::{ self, RedisCollectionPage, RedisCommandResult, RedisConnection, RedisDatabaseInfo, RedisScanResult, - RedisStreamConsumer, RedisStreamGroup, RedisStreamPendingPage, RedisValue, + RedisStreamConsumer, RedisStreamGroup, RedisStreamPage, RedisStreamPendingPage, RedisValue, }; async fn ensure_redis_pool(state: &AppState, connection_id: &str) -> Result<(), String> { @@ -137,6 +137,36 @@ pub async fn redis_get_value_in_db_core( } } +pub async fn redis_stream_entries_in_db_core( + state: &AppState, + connection_id: &str, + db: u32, + key_raw: &str, + cursor: Option<&str>, +) -> Result { + ensure_redis_pool(state, connection_id).await?; + let connections = state.connections.read().await; + let pool = connections.get(connection_id).ok_or("Connection not found")?; + match pool { + PoolKind::Redis(redis) => { + let key = redis_driver::redis_key_raw_to_bytes(key_raw)?; + match redis { + RedisConnection::Direct(con) => { + let mut con = con.lock().await; + redis_driver::select_db(&mut *con, db).await?; + redis_driver::get_stream_entries_page(&mut *con, &key, cursor).await + } + RedisConnection::Cluster(cluster) => { + redis_driver::ensure_cluster_db(db)?; + let mut con = redis_driver::cluster_key_connection(cluster, &key).await?; + redis_driver::get_stream_entries_page(&mut con, &key, cursor).await + } + } + } + _ => Err("Not a Redis connection".to_string()), + } +} + pub async fn redis_stream_groups_in_db_core( state: &AppState, connection_id: &str, diff --git a/crates/dbx-web/src/main.rs b/crates/dbx-web/src/main.rs index 0a6895e5f..c61f49040 100644 --- a/crates/dbx-web/src/main.rs +++ b/crates/dbx-web/src/main.rs @@ -493,6 +493,7 @@ async fn main() { .route("/redis/scan-keys-batch", post(routes::redis::scan_keys_batch)) .route("/redis/scan-values", post(routes::redis::scan_values)) .route("/redis/get-value", post(routes::redis::get_value)) + .route("/redis/get-stream-entries", post(routes::redis::get_stream_entries)) .route("/redis/get-stream-groups", post(routes::redis::get_stream_groups)) .route("/redis/get-stream-consumers", post(routes::redis::get_stream_consumers)) .route("/redis/get-stream-pending", post(routes::redis::get_stream_pending)) diff --git a/crates/dbx-web/src/routes/redis.rs b/crates/dbx-web/src/routes/redis.rs index 7c572a24d..ae5408b24 100644 --- a/crates/dbx-web/src/routes/redis.rs +++ b/crates/dbx-web/src/routes/redis.rs @@ -76,6 +76,15 @@ pub struct RedisKeyRequest { pub key_raw: String, } +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct RedisStreamEntriesRequest { + pub connection_id: String, + pub db: u32, + pub key_raw: String, + pub cursor: Option, +} + #[derive(Deserialize)] #[serde(rename_all = "camelCase")] pub struct RedisStreamGroupRequest { @@ -322,6 +331,22 @@ pub async fn get_value( Ok(Json(serde_json::to_value(result).map_err(|e| AppError::from(e.to_string()))?)) } +pub async fn get_stream_entries( + State(state): State>, + Json(req): Json, +) -> Result, AppError> { + let result = dbx_core::redis_ops::redis_stream_entries_in_db_core( + &state.app, + &req.connection_id, + req.db, + &req.key_raw, + req.cursor.as_deref(), + ) + .await + .map_err(AppError::from)?; + Ok(Json(serde_json::to_value(result).map_err(|e| AppError::from(e.to_string()))?)) +} + pub async fn get_stream_groups( State(state): State>, Json(req): Json, diff --git a/src-tauri/src/commands/redis_cmd.rs b/src-tauri/src/commands/redis_cmd.rs index 0321c8d9b..447dd55f0 100644 --- a/src-tauri/src/commands/redis_cmd.rs +++ b/src-tauri/src/commands/redis_cmd.rs @@ -4,7 +4,8 @@ use tauri::State; use crate::commands::connection::{ensure_connection_writable, AppState}; use dbx_core::db::redis_driver::{ classify_command, parse_command_argv, RedisCollectionPage, RedisCommandResult, RedisCommandSafety, - RedisDatabaseInfo, RedisScanResult, RedisStreamConsumer, RedisStreamGroup, RedisStreamPendingPage, RedisValue, + RedisDatabaseInfo, RedisScanResult, RedisStreamConsumer, RedisStreamGroup, RedisStreamPage, RedisStreamPendingPage, + RedisValue, }; #[tauri::command] @@ -86,6 +87,17 @@ pub async fn redis_get_value( dbx_core::redis_ops::redis_get_value_in_db_core(&state, &connection_id, db, &key_raw).await } +#[tauri::command] +pub async fn redis_get_stream_entries( + state: State<'_, Arc>, + connection_id: String, + db: u32, + key_raw: String, + cursor: Option, +) -> Result { + dbx_core::redis_ops::redis_stream_entries_in_db_core(&state, &connection_id, db, &key_raw, cursor.as_deref()).await +} + #[tauri::command] pub async fn redis_get_stream_groups( state: State<'_, Arc>, diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 4579a8782..c9cc34ab2 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -1765,6 +1765,7 @@ pub fn run() { commands::redis_cmd::redis_scan_keys_batch, commands::redis_cmd::redis_scan_values, commands::redis_cmd::redis_get_value, + commands::redis_cmd::redis_get_stream_entries, commands::redis_cmd::redis_get_stream_groups, commands::redis_cmd::redis_get_stream_consumers, commands::redis_cmd::redis_get_stream_pending,