✨ feat(redis-stream): 支持 Redis Stream 类型查看与消息增删

- 后端扩展 RedisClient 接口，新增 StreamEntry 与 Stream 操作定义 - Redis 实现新增 XADD/XDEL/XRANGE 封装并接入 RedisGetValue 的 stream 分支 - App 层新增 RedisStreamAdd 与 RedisStreamDelete 方法并返回操作结果 - 前端新增 stream 类型视图，支持消息新增、删除与字段复制 - refs #92
2026-06-26 00:11:43 +08:00 · 2026-02-11 10:41:22 +08:00
parent ab92e94bf8
commit e31dc4e7f1
7 changed files with 378 additions and 3 deletions
--- a/frontend/src/components/RedisViewer.tsx
+++ b/frontend/src/components/RedisViewer.tsx
@@ -2,7 +2,7 @@ import React, { useState, useEffect, useCallback, useMemo, useRef } from 'react'
 import { Table, Input, Button, Space, Tag, Tree, Spin, message, Modal, Form, InputNumber, Popconfirm, Tooltip, Radio } from 'antd';
 import { ReloadOutlined, DeleteOutlined, PlusOutlined, EditOutlined, SearchOutlined, ClockCircleOutlined, CopyOutlined, FolderOpenOutlined, KeyOutlined } from '@ant-design/icons';
 import { useStore } from '../store';
-import { RedisKeyInfo, RedisValue } from '../types';
+import { RedisKeyInfo, RedisValue, StreamEntry } from '../types';
 import Editor from '@monaco-editor/react';
 import type { DataNode } from 'antd/es/tree';

@@ -625,6 +625,7 @@ const RedisViewer: React.FC<RedisViewerProps> = ({ connectionId, redisDB }) => {
            case 'list': return 'orange';
            case 'set': return 'purple';
            case 'zset': return 'magenta';
+            case 'stream': return 'cyan';
            default: return 'default';
        }
    };
@@ -1468,6 +1469,212 @@ const RedisViewer: React.FC<RedisViewerProps> = ({ connectionId, redisDB }) => {
            );
        };

+        const renderStreamValue = () => {
+            const processValue = (value: string) => {
+                if (viewMode === 'hex') {
+                    return { displayValue: toHexDisplay(value), isBinary: true, isJson: false, encoding: 'HEX' };
+                } else if (viewMode === 'text') {
+                    return { displayValue: value, isBinary: false, isJson: false, encoding: 'Text' };
+                } else if (viewMode === 'utf8') {
+                    try {
+                        const bytes = new Uint8Array(value.length);
+                        for (let i = 0; i < value.length; i++) {
+                            bytes[i] = value.charCodeAt(i) & 0xFF;
+                        }
+                        const decoded = new TextDecoder('utf-8', { fatal: false }).decode(bytes);
+                        return { displayValue: decoded, isBinary: false, isJson: false, encoding: 'UTF-8' };
+                    } catch (e) {
+                        return { displayValue: value, isBinary: false, isJson: false, encoding: 'UTF-8 (失败)' };
+                    }
+                } else {
+                    return formatStringValue(value);
+                }
+            };
+
+            const data = (keyValue.value as StreamEntry[]).map((item, index) => {
+                const rawFieldsText = JSON.stringify(item.fields ?? {}, null, 2);
+                const { displayValue, isBinary, isJson, encoding } = processValue(rawFieldsText);
+                return {
+                    index,
+                    id: item.id,
+                    rawFieldsText,
+                    displayFields: displayValue,
+                    isBinary,
+                    isJson,
+                    encoding,
+                };
+            });
+
+            const handleAddStreamEntry = async (fieldsText: string, id: string) => {
+                const config = getConfig();
+                if (!config) return;
+
+                let parsed: unknown;
+                try {
+                    parsed = JSON.parse(fieldsText);
+                } catch (e) {
+                    message.error('字段 JSON 格式不正确');
+                    return;
+                }
+
+                if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed)) {
+                    message.error('字段必须是 JSON 对象');
+                    return;
+                }
+
+                const fieldMap: Record<string, string> = {};
+                Object.entries(parsed as Record<string, unknown>).forEach(([field, value]) => {
+                    fieldMap[field] = value == null ? '' : String(value);
+                });
+
+                if (Object.keys(fieldMap).length === 0) {
+                    message.error('至少提供一个字段');
+                    return;
+                }
+
+                try {
+                    const res = await (window as any).go.app.App.RedisStreamAdd(config, selectedKey, fieldMap, id || '*');
+                    if (res.success) {
+                        const newID = res.data?.id ? ` (${res.data.id})` : '';
+                        message.success(`添加成功${newID}`);
+                        loadKeyValue(selectedKey);
+                    } else {
+                        message.error('添加失败: ' + res.message);
+                    }
+                } catch (e: any) {
+                    message.error('添加失败: ' + (e?.message || String(e)));
+                }
+            };
+
+            const handleDeleteStreamEntry = async (id: string) => {
+                const config = getConfig();
+                if (!config) return;
+
+                try {
+                    const res = await (window as any).go.app.App.RedisStreamDelete(config, selectedKey, [id]);
+                    if (res.success) {
+                        const deleted = Number(res.data?.deleted ?? 0);
+                        if (deleted > 0) {
+                            message.success('删除成功');
+                        } else {
+                            message.warning('未删除任何消息，可能已不存在');
+                        }
+                        loadKeyValue(selectedKey);
+                    } else {
+                        message.error('删除失败: ' + res.message);
+                    }
+                } catch (e: any) {
+                    message.error('删除失败: ' + (e?.message || String(e)));
+                }
+            };
+
+            return (
+                <div style={{ display: 'flex', flexDirection: 'column', height: '100%' }}>
+                    <div style={{ marginBottom: 8, display: 'flex', justifyContent: 'space-between', alignItems: 'center' }}>
+                        <Button size="small" icon={<PlusOutlined />} onClick={() => {
+                            Modal.confirm({
+                                title: '添加 Stream 消息',
+                                width: 680,
+                                content: (
+                                    <div>
+                                        <div style={{ marginBottom: 8 }}>
+                                            <label>ID（可选，默认 *）：</label>
+                                            <Input id="new-stream-id" placeholder="例如: * 或 1723110000000-0" />
+                                        </div>
+                                        <div>
+                                            <label>字段 JSON：</label>
+                                            <Input.TextArea id="new-stream-fields" rows={8} defaultValue={'{\n  "field": "value"\n}'} />
+                                        </div>
+                                    </div>
+                                ),
+                                onOk: async () => {
+                                    const id = (document.getElementById('new-stream-id') as HTMLInputElement)?.value?.trim() || '*';
+                                    const fieldsText = (document.getElementById('new-stream-fields') as HTMLTextAreaElement)?.value || '{}';
+                                    await handleAddStreamEntry(fieldsText, id);
+                                }
+                            });
+                        }}>添加消息</Button>
+                        <Radio.Group size="small" value={viewMode} onChange={(e) => setViewMode(e.target.value)}>
+                            <Radio.Button value="auto">自动</Radio.Button>
+                            <Radio.Button value="text">原始文本</Radio.Button>
+                            <Radio.Button value="utf8">UTF-8</Radio.Button>
+                            <Radio.Button value="hex">十六进制</Radio.Button>
+                        </Radio.Group>
+                    </div>
+                    <Table
+                        dataSource={data}
+                        columns={[
+                            {
+                                title: 'ID',
+                                dataIndex: 'id',
+                                key: 'id',
+                                width: 240,
+                                ellipsis: true,
+                            },
+                            {
+                                title: '字段',
+                                dataIndex: 'displayFields',
+                                key: 'fields',
+                                ellipsis: true,
+                                render: (text: string, record: any) => {
+                                    const tooltipContent = record.encoding && record.encoding !== 'UTF-8'
+                                        ? `[${record.encoding}]\n${text}`
+                                        : text;
+
+                                    return (
+                                        <Tooltip title={<pre style={{ maxHeight: 300, overflow: 'auto', margin: 0, fontSize: 12 }}>{tooltipContent}</pre>} styles={{ root: { maxWidth: 720 } }}>
+                                            <span style={{
+                                                color: record.isBinary ? '#d46b08' : (record.isJson ? '#1890ff' : undefined),
+                                                fontFamily: record.isBinary ? 'monospace' : undefined,
+                                                fontSize: record.isBinary ? 11 : undefined
+                                            }}>
+                                                {text}
+                                            </span>
+                                        </Tooltip>
+                                    );
+                                }
+                            },
+                            {
+                                title: '操作',
+                                key: 'action',
+                                width: 140,
+                                render: (_: any, record: any) => (
+                                    <Space size="small">
+                                        <Tooltip title="复制 ID">
+                                            <Button type="text" size="small" icon={<CopyOutlined />} onClick={() => {
+                                                navigator.clipboard.writeText(record.id).then(() => {
+                                                    message.success('已复制');
+                                                }).catch(() => {
+                                                    message.error('复制失败');
+                                                });
+                                            }} />
+                                        </Tooltip>
+                                        <Tooltip title="复制字段 JSON">
+                                            <Button type="text" size="small" icon={<CopyOutlined />} onClick={() => {
+                                                navigator.clipboard.writeText(record.rawFieldsText).then(() => {
+                                                    message.success('已复制');
+                                                }).catch(() => {
+                                                    message.error('复制失败');
+                                                });
+                                            }} />
+                                        </Tooltip>
+                                        <Popconfirm title="确定删除此消息？" onConfirm={() => handleDeleteStreamEntry(record.id)}>
+                                            <Button type="text" size="small" danger icon={<DeleteOutlined />} />
+                                        </Popconfirm>
+                                    </Space>
+                                )
+                            }
+                        ]}
+                        rowKey="id"
+                        size="small"
+                        pagination={{ pageSize: 50 }}
+                        scroll={{ y: 'calc(100vh - 350px)' }}
+                        style={{ flex: 1 }}
+                    />
+                </div>
+            );
+        };
+
        return (
            <div style={{ padding: 12, height: '100%', display: 'flex', flexDirection: 'column' }}>
                <div style={{ marginBottom: 12, display: 'flex', justifyContent: 'space-between', alignItems: 'center', flexShrink: 0 }}>
@@ -1511,6 +1718,7 @@ const RedisViewer: React.FC<RedisViewerProps> = ({ connectionId, redisDB }) => {
                    {keyValue.type === 'list' && renderListValue()}
                    {keyValue.type === 'set' && renderSetValue()}
                    {keyValue.type === 'zset' && renderZSetValue()}
+                    {keyValue.type === 'stream' && renderStreamValue()}
                </div>
            </div>
        );
--- a/frontend/src/types.ts
+++ b/frontend/src/types.ts
@@ -131,7 +131,7 @@ export interface RedisScanResult {
 }

 export interface RedisValue {
-  type: 'string' | 'hash' | 'list' | 'set' | 'zset';
+  type: 'string' | 'hash' | 'list' | 'set' | 'zset' | 'stream';
  ttl: number;
  value: any;
  length: number;
@@ -146,3 +146,8 @@ export interface ZSetMember {
  member: string;
  score: number;
 }
+
+export interface StreamEntry {
+  id: string;
+  fields: Record<string, string>;
+}
--- a/frontend/wailsjs/go/app/App.d.ts
+++ b/frontend/wailsjs/go/app/App.d.ts
@@ -120,6 +120,10 @@ export function RedisSetString(arg1:connection.ConnectionConfig,arg2:string,arg3

 export function RedisSetTTL(arg1:connection.ConnectionConfig,arg2:string,arg3:number):Promise<connection.QueryResult>;

+export function RedisStreamAdd(arg1:connection.ConnectionConfig,arg2:string,arg3:Record<string, string>,arg4:string):Promise<connection.QueryResult>;
+
+export function RedisStreamDelete(arg1:connection.ConnectionConfig,arg2:string,arg3:Array<string>):Promise<connection.QueryResult>;
+
 export function RedisTestConnection(arg1:connection.ConnectionConfig):Promise<connection.QueryResult>;

 export function RedisZSetAdd(arg1:connection.ConnectionConfig,arg2:string,arg3:Array<redis.ZSetMember>):Promise<connection.QueryResult>;
--- a/frontend/wailsjs/go/app/App.js
+++ b/frontend/wailsjs/go/app/App.js
@@ -234,6 +234,14 @@ export function RedisSetTTL(arg1, arg2, arg3) {
  return window['go']['app']['App']['RedisSetTTL'](arg1, arg2, arg3);
 }

+export function RedisStreamAdd(arg1, arg2, arg3, arg4) {
+  return window['go']['app']['App']['RedisStreamAdd'](arg1, arg2, arg3, arg4);
+}
+
+export function RedisStreamDelete(arg1, arg2, arg3) {
+  return window['go']['app']['App']['RedisStreamDelete'](arg1, arg2, arg3);
+}
+
 export function RedisTestConnection(arg1) {
  return window['go']['app']['App']['RedisTestConnection'](arg1);
 }
--- a/internal/app/methods_redis.go
+++ b/internal/app/methods_redis.go
@@ -450,6 +450,40 @@ func (a *App) RedisZSetRemove(config connection.ConnectionConfig, key string, me
 	return connection.QueryResult{Success: true, Message: "删除成功"}
 }

+// RedisStreamAdd adds an entry to a stream
+func (a *App) RedisStreamAdd(config connection.ConnectionConfig, key string, fields map[string]string, id string) connection.QueryResult {
+	config.Type = "redis"
+	client, err := a.getRedisClient(config)
+	if err != nil {
+		return connection.QueryResult{Success: false, Message: err.Error()}
+	}
+
+	newID, err := client.StreamAdd(key, fields, id)
+	if err != nil {
+		logger.Error(err, "RedisStreamAdd 添加失败：key=%s id=%s", key, id)
+		return connection.QueryResult{Success: false, Message: err.Error()}
+	}
+
+	return connection.QueryResult{Success: true, Message: "添加成功", Data: map[string]string{"id": newID}}
+}
+
+// RedisStreamDelete deletes stream entries by IDs
+func (a *App) RedisStreamDelete(config connection.ConnectionConfig, key string, ids []string) connection.QueryResult {
+	config.Type = "redis"
+	client, err := a.getRedisClient(config)
+	if err != nil {
+		return connection.QueryResult{Success: false, Message: err.Error()}
+	}
+
+	deleted, err := client.StreamDelete(key, ids...)
+	if err != nil {
+		logger.Error(err, "RedisStreamDelete 删除失败：key=%s ids=%v", key, ids)
+		return connection.QueryResult{Success: false, Message: err.Error()}
+	}
+
+	return connection.QueryResult{Success: true, Message: "删除成功", Data: map[string]int64{"deleted": deleted}}
+}
+
 // RedisFlushDB flushes the current database
 func (a *App) RedisFlushDB(config connection.ConnectionConfig) connection.QueryResult {
 	config.Type = "redis"
--- a/internal/redis/redis.go
+++ b/internal/redis/redis.go
@@ -4,7 +4,7 @@ import "GoNavi-Wails/internal/connection"

 // RedisValue represents a Redis value with its type and metadata
 type RedisValue struct {
-	Type   string      `json:"type"`   // string, hash, list, set, zset
+	Type   string      `json:"type"`   // string, hash, list, set, zset, stream
 	TTL    int64       `json:"ttl"`    // TTL in seconds, -1 means no expiry, -2 means key doesn't exist
 	Value  interface{} `json:"value"`  // The actual value
 	Length int64       `json:"length"` // Length/size of the value
@@ -72,6 +72,11 @@ type RedisClient interface {
 	ZSetAdd(key string, members ...ZSetMember) error
 	ZSetRemove(key string, members ...string) error

+	// Stream operations
+	GetStream(key, start, stop string, count int64) ([]StreamEntry, error)
+	StreamAdd(key string, fields map[string]string, id string) (string, error)
+	StreamDelete(key string, ids ...string) (int64, error)
+
 	// Command execution
 	ExecuteCommand(args []string) (interface{}, error)

@@ -88,3 +93,9 @@ type ZSetMember struct {
 	Member string  `json:"member"`
 	Score  float64 `json:"score"`
 }
+
+// StreamEntry represents a single stream message
+type StreamEntry struct {
+	ID     string            `json:"id"`
+	Fields map[string]string `json:"fields"`
+}
--- a/internal/redis/redis_impl.go
+++ b/internal/redis/redis_impl.go
@@ -334,6 +334,26 @@ func (r *RedisClientImpl) GetValue(key string) (*RedisValue, error) {
 		result.Value = members
 		result.Length = length

+	case "stream":
+		length, err := r.client.XLen(ctx, key).Result()
+		if err != nil {
+			return nil, err
+		}
+		result.Length = length
+		if length == 0 {
+			result.Value = []StreamEntry{}
+			break
+		}
+		limit := int64(1000)
+		if length < limit {
+			limit = length
+		}
+		val, err := r.client.XRangeN(ctx, key, "-", "+", limit).Result()
+		if err != nil {
+			return nil, err
+		}
+		result.Value = toStreamEntries(val)
+
 	default:
 		return nil, fmt.Errorf("不支持的 Redis 数据类型: %s", keyType)
 	}
@@ -523,6 +543,91 @@ func (r *RedisClientImpl) ZSetRemove(key string, members ...string) error {
 	return r.client.ZRem(ctx, key, args...).Err()
 }

+// GetStream gets stream entries in a range
+func (r *RedisClientImpl) GetStream(key, start, stop string, count int64) ([]StreamEntry, error) {
+	if r.client == nil {
+		return nil, fmt.Errorf("Redis 客户端未连接")
+	}
+	if start == "" {
+		start = "-"
+	}
+	if stop == "" {
+		stop = "+"
+	}
+	if count <= 0 {
+		count = 1000
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	val, err := r.client.XRangeN(ctx, key, start, stop, count).Result()
+	if err != nil {
+		return nil, err
+	}
+	return toStreamEntries(val), nil
+}
+
+// StreamAdd adds an entry to a stream
+func (r *RedisClientImpl) StreamAdd(key string, fields map[string]string, id string) (string, error) {
+	if r.client == nil {
+		return "", fmt.Errorf("Redis 客户端未连接")
+	}
+	if len(fields) == 0 {
+		return "", fmt.Errorf("Stream 字段不能为空")
+	}
+	if id == "" {
+		id = "*"
+	}
+
+	values := make(map[string]interface{}, len(fields))
+	for field, value := range fields {
+		values[field] = value
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	newID, err := r.client.XAdd(ctx, &redis.XAddArgs{
+		Stream: key,
+		ID:     id,
+		Values: values,
+	}).Result()
+	if err != nil {
+		return "", err
+	}
+	return newID, nil
+}
+
+// StreamDelete deletes entries from a stream by IDs
+func (r *RedisClientImpl) StreamDelete(key string, ids ...string) (int64, error) {
+	if r.client == nil {
+		return 0, fmt.Errorf("Redis 客户端未连接")
+	}
+	if len(ids) == 0 {
+		return 0, fmt.Errorf("Stream ID 不能为空")
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	return r.client.XDel(ctx, key, ids...).Result()
+}
+
+func toStreamEntries(messages []redis.XMessage) []StreamEntry {
+	entries := make([]StreamEntry, 0, len(messages))
+	for _, msg := range messages {
+		fields := make(map[string]string, len(msg.Values))
+		for field, value := range msg.Values {
+			fields[field] = fmt.Sprint(value)
+		}
+		entries = append(entries, StreamEntry{
+			ID:     msg.ID,
+			Fields: fields,
+		})
+	}
+	return entries
+}
+
 // ExecuteCommand executes a raw Redis command
 func (r *RedisClientImpl) ExecuteCommand(args []string) (interface{}, error) {
 	if r.client == nil {