admin/nexus-terminal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #34 from Heavrnl/sftp

Browse Source 
Sftp
This commit is contained in:
Baobhan Sith
2025-05-15 20:47:33 +08:00
committed by GitHub
parent b9d1a70b9a 23d5ba7ac6
commit 01284f662f
7 changed files with 145 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/backend/src/services/sftp.service.ts
+48 -30
View File
@@ -64,6 +64,7 @@ interface ActiveUpload {
stream: WriteStream;
sessionId: string; // Link back to the session for cleanup
relativePath?: string;
drainPromise?: Promise<void> | null; // +++ For managing drain event listeners +++
}
export class SftpService {
@@ -1489,6 +1490,7 @@ export class SftpService {
stream,
sessionId,
relativePath, // +++ 存储 relativePath +++
drainPromise: null // +++ Initialize drainPromise +++
};
this.activeUploads.set(uploadId, uploadState);
@@ -1566,46 +1568,36 @@ export class SftpService {
const chunkBuffer = Buffer.from(dataBase64, 'base64');
const writeSuccess = uploadState.stream.write(chunkBuffer, (err) => {
if (err) {
// This callback handles errors specifically related to *this* write operation.
console.error(`[SFTP Upload ${uploadId}] Error writing chunk ${chunkIndex} to ${uploadState.remotePath}:`, err);
state.ws.send(JSON.stringify({ type: 'sftp:upload:error', payload: { uploadId, message: `写入块 ${chunkIndex} 失败: ${err.message}` } }));
// Consider cancelling the upload on write error
this.cancelUploadInternal(uploadId, `Write error on chunk ${chunkIndex}`);
}
});
if (!writeSuccess) {
// console.warn(`[SFTP Upload ${uploadId}] Write stream buffer full after chunk ${chunkIndex}. Pausing chunk processing until 'drain'.`);
try {
await new Promise<void>(resolve => uploadState.stream.once('drain', resolve));
// console.log(`[SFTP Upload ${uploadId}] Write stream drained after chunk ${chunkIndex}. Resuming chunk processing.`);
} catch (drainError) {
// Should not happen with .once, but handle defensively
console.error(`[SFTP Upload ${uploadId}] Error waiting for drain event:`, drainError);
// Consider cancelling upload if waiting for drain fails critically
this.cancelUploadInternal(uploadId, 'Error waiting for drain');
throw drainError; // Re-throw to stop further processing in this chunk handler
}
}
} else {
uploadState.bytesWritten += chunkBuffer.length;
if (state.ws && state.ws.readyState === WebSocket.OPEN) {
const progressPercent = Math.round((uploadState.bytesWritten / uploadState.totalSize) * 100);
state.ws.send(JSON.stringify({
type: 'sftp:upload:progress',
uploadId: uploadId,
payload: {
bytesWritten: uploadState.bytesWritten,
totalSize: uploadState.totalSize,
progress: Math.min(100, progressPercent)
}
}));
}
// Check if upload is complete
if (uploadState.bytesWritten > uploadState.totalSize) {
console.error(`[SFTP Upload ${uploadId}] Bytes written (${uploadState.bytesWritten}) exceeded total size (${uploadState.totalSize}) for ${uploadState.remotePath}.`);
state.ws.send(JSON.stringify({ type: 'sftp:upload:error', payload: { uploadId, message: '写入字节数超过文件总大小' } }));
this.cancelUploadInternal(uploadId, 'Bytes written exceeded total size');
} else if (uploadState.bytesWritten >= uploadState.totalSize) {
if (uploadState.bytesWritten >= uploadState.totalSize) {
if (!uploadState.stream.writableEnded) {
uploadState.stream.end((endErr: Error & { code?: string } | undefined) => { // Added code to Error type hint
const streamStateInEndCallback = uploadState?.stream; // Re-fetch in case it changed
uploadState.stream.end((endErr: Error & { code?: string } | undefined) => {
const streamStateInEndCallback = uploadState?.stream;
if (endErr) {
// Check if it's the specific ERR_STREAM_DESTROYED error and all bytes were written
if (endErr.code === 'ERR_STREAM_DESTROYED' && uploadState && uploadState.bytesWritten >= uploadState.totalSize) {
console.warn(`[SFTP Upload ${uploadId}] stream.end() CALLBACK reported ERR_STREAM_DESTROYED, but all bytes written. UploadId: ${uploadId}. Error:`, endErr);
console.log(`[SFTP Upload ${uploadId}] Treating ERR_STREAM_DESTROYED as non-fatal for this upload. Expecting 'close' event to finalize success for ${uploadState.remotePath}.`);
@@ -1620,6 +1612,32 @@ export class SftpService {
});
}
}
}
});
if (!writeSuccess) {
if (!uploadState.drainPromise) {
uploadState.drainPromise = new Promise<void>(resolve => {
uploadState.stream.once('drain', () => {
uploadState.drainPromise = null;
resolve();
});
});
}
try {
await uploadState.drainPromise;
} catch (drainError) {
console.error(`[SFTP Upload ${uploadId}] Error awaiting drain promise for chunk ${chunkIndex}:`, drainError);
this.cancelUploadInternal(uploadId, 'Error waiting for drain promise');
throw drainError;
}
}
} catch (error: any) {
console.error(`[SFTP Upload ${uploadId}] Error handling chunk ${chunkIndex} for ${uploadState?.remotePath}:`, error);
packages/backend/src/websocket/connection.ts
+1 -1
View File
@@ -245,7 +245,7 @@ export function initializeConnectionHandler(wss: WebSocketServer, sshSuspendServ
}));
console.log(`[WebSocket Handler][SSH_SUSPEND_RESUME_REQUEST] 已发送 ssh:connected 给 ${newFrontendSessionId}。`);
}
// +++ 结束新增 +++
const responseNotification: SshSuspendResumedNotification = { // 确保变量名不冲突且类型正确
type: 'SSH_SUSPEND_RESUMED_NOTIF', // 改回与前端和新类型定义一致
packages/backend/src/websocket/types.ts
+21 -20
View File
@@ -6,7 +6,7 @@ export interface AuthenticatedWebSocket extends WebSocket {
isAlive?: boolean;
userId?: number;
username?: string;
sessionId?: string; // 用于关联 ClientState 的唯一 ID
sessionId?: string;
}
// 中心化的客户端状态接口 (统一版本)
@@ -15,11 +15,11 @@ export interface ClientState { // 导出以便 Service 可以导入
sshClient: Client;
sshShellStream?: ClientChannel;
dbConnectionId: number;
connectionName?: string; // 添加连接名称字段
sftp?: SFTPWrapper; // 添加 sftp 实例 (由 SftpService 管理)
statusIntervalId?: NodeJS.Timeout; // 添加状态轮询 ID (由 StatusMonitorService 管理)
connectionName?: string; // 连接名称字段
sftp?: SFTPWrapper; // sftp 实例 (由 SftpService 管理)
statusIntervalId?: NodeJS.Timeout; // 状态轮询 ID (由 StatusMonitorService 管理)
dockerStatusIntervalId?: NodeJS.Timeout; // Docker 状态轮询 ID
ipAddress?: string; // 添加 IP 地址字段
ipAddress?: string; // IP 地址字段
isShellReady?: boolean; // 标记 Shell 是否已准备好处理输入和调整大小
isSuspendedByService?: boolean; // 标记此会话是否已被 SshSuspendService 接管
isMarkedForSuspend?: boolean; // 标记此会话是否已被用户请求挂起（等待断开连接）
@@ -37,14 +37,14 @@ export interface PortInfo {
// --- Docker Interfaces (Ensure this matches frontend and DockerService) ---
// Stats 接口
export interface DockerStats {
ID: string; // 来自 docker stats
Name: string; // 来自 docker stats
CPUPerc: string; // 来自 docker stats
MemUsage: string; // 来自 docker stats
MemPerc: string; // 来自 docker stats
NetIO: string; // 来自 docker stats
BlockIO: string; // 来自 docker stats
PIDs: string; // 来自 docker stats
ID: string;
Name: string;
CPUPerc: string;
MemUsage: string;
MemPerc: string;
NetIO: string;
BlockIO: string;
PIDs: string;
}
// Container 接口 (包含 stats)
@@ -246,13 +246,7 @@ export type SshSuspendServerToClientMessages =
| SshMarkedForSuspendAck
| SshUnmarkedForSuspendAck;
// It might be useful to have a general type for incoming messages if not already present
// For example, if you have a main message handler:
// export type WebSocketMessage = BaseMessageType | SshSuspendClientToServerMessages | OtherFeatureMessages;
// And for outgoing:
// export type WebSocketResponse = BaseResponseType | SshSuspendServerToClientMessages | OtherFeatureResponses;
// This part depends on the existing structure, so I'm providing the specific types for now.
// --- SFTP Compress/Decompress Message Types ---
// C -> S: Request to compress files/directories
export interface SftpCompressRequestPayload {
@@ -294,3 +288,10 @@ export interface SftpDecompressErrorPayload {
details?: string; // Stderr output or specific error details
requestId: string;
}
// S -> C: SFTP Upload Progress (New)
export interface SftpUploadProgressPayload {
uploadId: string; // To correlate with the specific upload
bytesWritten: number;
totalSize: number;
progress: number; // Calculated percentage (0-100)
}
packages/frontend/src/components/FileEditorContainer.vue
+1 -2
View File
@@ -309,7 +309,7 @@ const handleKeyDown = (event: KeyboardEvent) => {
</select>
</div>
<span v-else-if="activeTab" class="encoding-select-placeholder">{{ t('fileManager.loadingEncoding', '加载中...') }}</span>
<!-- +++ 结束新增 +++ -->
<span v-if="currentTabSaveStatus === 'saving'" class="save-status saving">{{ t('fileManager.saving') }}...</span>
<span v-if="currentTabSaveStatus === 'success'" class="save-status success">✅ {{ t('fileManager.saveSuccess') }}</span>
@@ -317,7 +317,6 @@ const handleKeyDown = (event: KeyboardEvent) => {
<button @click="handleSaveRequest" :disabled="currentTabIsSaving || currentTabIsLoading || !!currentTabLoadingError || !activeTab || !currentTabIsModified" class="save-btn">
{{ t('fileManager.actions.save') }}
</button>
<!-- 关闭/最小化按钮已移除 -->
</div>
</div>
<!-- 如果没有活动标签页，显示简化头部 -->
packages/frontend/src/components/FileEditorOverlay.vue
+1 -1
View File
@@ -520,7 +520,7 @@ onBeforeUnmount(() => {
</select>
</div>
<span v-else-if="activeTab" class="encoding-select-placeholder">{{ t('fileManager.loadingEncoding', '加载中...') }}</span>
<!-- +++ 结束新增 +++ -->
++ -->
<span v-if="currentTabSaveStatus === 'saving'" class="save-status saving">{{ t('fileManager.saving') }}...</span>
<span v-if="currentTabSaveStatus === 'success'" class="save-status success">✅ {{ t('fileManager.saveSuccess') }}</span>
packages/frontend/src/composables/useFileUploader.ts
+50 -46
View File
@@ -1,21 +1,19 @@
import { ref, reactive, nextTick, onUnmounted, readonly, type Ref } from 'vue'; // 再次修正导入，移除大写 Readonly
import { createWebSocketConnectionManager } from './useWebSocketConnection'; // 导入工厂函数
import { ref, reactive, nextTick, onUnmounted, readonly, type Ref } from 'vue';
import { createWebSocketConnectionManager } from './useWebSocketConnection';
import { useI18n } from 'vue-i18n';
import type { FileListItem } from '../types/sftp.types'; // 从 sftp 类型文件导入
import type { UploadItem } from '../types/upload.types'; // 从 upload 类型文件导入
import type { WebSocketMessage, MessagePayload } from '../types/websocket.types'; // 从类型文件导入
import type { FileListItem } from '../types/sftp.types';
import type { UploadItem } from '../types/upload.types';
import type { WebSocketMessage, MessagePayload } from '../types/websocket.types';
// --- 接口定义 (已移至 upload.types.ts) ---
import type { WebSocketDependencies } from './useSftpActions'; // 导入 WebSocketDependencies 类型
import type { WebSocketDependencies } from './useSftpActions';
// 辅助函数 (从 FileManager.vue 复制)
const generateUploadId = (): string => {
// 如果需要，可以使用稍微不同的格式作为上传 ID
return `upload-${Date.now()}-${Math.random().toString(36).substring(2, 9)}`;
};
// 辅助函数 (从 FileManager.vue 复制)
const joinPath = (base: string, name: string): string => {
if (base === '/') return `/${name}`;
if (base.endsWith('/')) return `${base}${name}`;
@@ -25,14 +23,9 @@ const joinPath = (base: string, name: string): string => {
export function useFileUploader(
currentPathRef: Ref<string>,
fileListRef: Readonly<Ref<readonly FileListItem[]>>, // 使用 Readonly 类型
// refreshDirectory: () => void, // 不再需要此回调
// sessionId: string, // 不再需要，因为 wsDeps 包含了会话上下文
// dbConnectionId: string, // 不再需要
wsDeps: WebSocketDependencies // 注入 WebSocket 依赖项
) {
const { t } = useI18n();
// 不再创建独立的连接管理器，而是使用注入的依赖项
// const { sendMessage, onMessage, isConnected } = createWebSocketConnectionManager(sessionId, dbConnectionId, t);
const { sendMessage, onMessage, isConnected } = wsDeps; // 使用注入的依赖项
// 对 uploads 字典使用 reactive 以获得更好的深度响应性
@@ -70,24 +63,24 @@ export function useFileUploader(
sendMessage({
type: 'sftp:upload:chunk',
payload: { uploadId, chunkIndex: chunkIndex++, data: chunkBase64, isLast } // Add and increment chunkIndex
payload: { uploadId, chunkIndex: chunkIndex++, data: chunkBase64, isLast }
});
// --- FIX: Update offset based on the actual chunk size that was read ---
offset += currentChunkSize; // Use the stored size of the slice
currentUpload.progress = Math.min(100, Math.round((offset / file.size) * 100));
offset += currentChunkSize;
if (!isLast) {
// 使用 requestAnimationFrame 或 nextTick 在块之间添加轻微延迟
// 以潜在地改善 UI 响应性并减少负载。
nextTick(readNextChunk);
} else {
console.log(`[文件上传模块] 已发送 ${uploadId} 的最后一个块`);
// 后端将在收到最后一个块后发送 sftp:upload:success
}
} else {
console.error(`[文件上传模块] FileReader 为 ${uploadId} 返回了意外结果:`, chunkResult);
// 处理错误：更新上传状态，也许重试？
currentUpload.status = 'error';
currentUpload.error = t('fileManager.errors.readFileError');
}
@@ -106,7 +99,7 @@ export function useFileUploader(
// 读取下一个块之前再次检查状态
if (offset < file.size && uploads[uploadId]?.status === 'uploading') {
const slice = file.slice(offset, offset + chunkSize);
currentChunkSize = slice.size; // Store the actual size of the slice being read
currentChunkSize = slice.size;
reader.readAsDataURL(slice);
}
};
@@ -120,23 +113,23 @@ export function useFileUploader(
// Send chunkIndex 0 for zero-byte file
sendMessage({ type: 'sftp:upload:chunk', payload: { uploadId, chunkIndex: 0, data: '', isLast: true } });
upload.progress = 100;
// Backend should send success message shortly after this
}
};
const startFileUpload = (file: File, relativePath?: string) => { // 保持签名修改
const startFileUpload = (file: File, relativePath?: string) => {
if (!isConnected.value) {
console.warn('[文件上传模块] 无法开始上传：WebSocket 未连接。');
// 可以选择向用户显示错误消息
return;
}
const uploadId = generateUploadId();
// --- 修正：直接构建最终远程路径 ---
let finalRemotePath: string;
if (relativePath) {
// 确保 currentPathRef.value 结尾有斜杠
const basePath = currentPathRef.value.endsWith('/') ? currentPathRef.value : `${currentPathRef.value}/`;
// 确保 relativePath 开头没有斜杠，末尾有斜杠 (如果非空)
let cleanRelativePath = relativePath.startsWith('/') ? relativePath.substring(1) : relativePath;
@@ -215,20 +208,12 @@ export function useFileUploader(
upload.status = 'success';
upload.progress = 100;
// 不再调用 refreshDirectory()，由 useSftpActions 处理列表更新
// refreshDirectory();
// 立即删除记录
if (uploads[uploadId]) { // 确保记录仍然存在
delete uploads[uploadId];
}
// 延迟后从列表中移除
// setTimeout(() => {
// if (uploads[uploadId]?.status === 'success') {
// delete uploads[uploadId];
// }
// }, 2000); // 成功状态显示时间短一些
} else {
console.warn(`[文件上传模块] 收到未知上传 ID 的 upload:success 消息: ${uploadId}`);
}
@@ -277,10 +262,7 @@ export function useFileUploader(
if (upload && upload.status === 'paused') {
console.log(`[文件上传模块] 恢复上传 ${uploadId}`);
upload.status = 'uploading';
// 恢复发送块（后端应该告知从哪里恢复，
// 但现在假设我们重新开始或后端处理了它）
// 更健壮的实现需要后端发送最后接收到的字节偏移量。
sendFileChunks(uploadId, upload.file); // 为简单起见，现在重新开始发送块
sendFileChunks(uploadId, upload.file);
}
};
@@ -303,6 +285,30 @@ export function useFileUploader(
}
};
// +++ 处理上传进度更新 +++
const onUploadProgress = (payload: MessagePayload, message: WebSocketMessage) => {
const uploadId = message.uploadId || payload?.uploadId; // 从顶层获取 uploadId
if (!uploadId) {
console.warn(`[文件上传模块] 收到缺少 uploadId 的 upload:progress 消息:`, message);
return;
}
const upload = uploads[uploadId];
if (upload && upload.status === 'uploading') {
// payload 现在应该包含 bytesWritten 和 totalSize
if (typeof payload?.bytesWritten === 'number' && typeof payload?.totalSize === 'number') {
upload.progress = Math.min(100, Math.round((payload.bytesWritten / payload.totalSize) * 100));
} else {
console.warn(`[文件上传模块] 收到 upload:progress 消息，但 payload 格式不正确:`, payload);
}
} else if (upload) {
} else {
console.warn(`[文件上传模块] 收到未知上传 ID 的 upload:progress 消息: ${uploadId}`);
}
};
// --- 注册处理器 ---
const unregisterUploadReady = onMessage('sftp:upload:ready', onUploadReady);
@@ -311,6 +317,7 @@ export function useFileUploader(
const unregisterUploadPause = onMessage('sftp:upload:pause', onUploadPause);
const unregisterUploadResume = onMessage('sftp:upload:resume', onUploadResume);
const unregisterUploadCancelled = onMessage('sftp:upload:cancelled', onUploadCancelled);
const unregisterUploadProgress = onMessage('sftp:upload:progress', onUploadProgress); // +++ 注册新处理器 +++
// --- 清理 ---
onUnmounted(() => {
@@ -321,6 +328,7 @@ export function useFileUploader(
unregisterUploadPause?.();
unregisterUploadResume?.();
unregisterUploadCancelled?.();
unregisterUploadProgress?.();
// 当使用此 composable 的组件卸载时，取消任何正在进行的上传
Object.keys(uploads).forEach(uploadId => {
@@ -329,12 +337,8 @@ export function useFileUploader(
});
return {
uploads, // 暴露响应式字典
uploads,
startFileUpload,
cancelUpload,
// 如果拖放/选择处理程序要在这里管理，则暴露它们，
// 或者将它们保留在组件中并调用 startFileUpload。
// 为简单起见，假设组件处理 UI 事件
// 并为每个文件调用 startFileUpload(file)。
};
}
packages/frontend/src/types/websocket.types.ts
+11
View File
@@ -15,6 +15,17 @@ export interface WebSocketMessage {
// 消息处理器函数类型
export type MessageHandler = (payload: MessagePayload, message: WebSocketMessage) => void; // 恢复 message 参数为必需
export interface SftpUploadProgressPayload {
uploadId: string; // 虽然 uploadId 在 WebSocketMessage 顶层，payload 里也包含以明确关联
bytesWritten: number;
totalSize: number;
progress: number; // 0-100
}
export interface SftpUploadProgressMessage extends WebSocketMessage {
type: 'sftp:upload:progress';
uploadId: string; // uploadId 也在顶层消息中，这里为了明确 payload 关联
payload: SftpUploadProgressPayload;
}
// --- SSH Suspend Mode WebSocket Message Types ---