diff --git a/apps/ops-web/nginx.conf b/apps/ops-web/nginx.conf index a8ee2a6..eed3e6d 100644 --- a/apps/ops-web/nginx.conf +++ b/apps/ops-web/nginx.conf @@ -8,7 +8,7 @@ real_ip_recursive on; server { listen 80; server_name _; - client_max_body_size 600M; + client_max_body_size 1100M; root /usr/share/nginx/html; index index.html; @@ -31,8 +31,8 @@ server { proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $remote_addr; proxy_set_header X-Forwarded-Proto $scheme; - proxy_read_timeout 900s; - proxy_send_timeout 900s; + proxy_read_timeout 1800s; + proxy_send_timeout 1800s; } location ~* \.(js|css|png|jpg|jpeg|gif|ico|svg|woff2?)$ { diff --git a/apps/ops-web/src/lib/desktop-client-releases.ts b/apps/ops-web/src/lib/desktop-client-releases.ts index ce0b12b..4b5de32 100644 --- a/apps/ops-web/src/lib/desktop-client-releases.ts +++ b/apps/ops-web/src/lib/desktop-client-releases.ts @@ -75,7 +75,27 @@ export interface DesktopClientReleaseDownloadURLResult { expires_in?: number | null } -const DESKTOP_CLIENT_RELEASE_UPLOAD_TIMEOUT_MS = 15 * 60 * 1000 +const DESKTOP_CLIENT_RELEASE_UPLOAD_TIMEOUT_MS = 30 * 60 * 1000 +const DESKTOP_CLIENT_RELEASE_CHUNK_TIMEOUT_MS = 5 * 60 * 1000 +const DESKTOP_CLIENT_RELEASE_CHUNK_SIZE_BYTES = 16 * 1024 * 1024 +const DESKTOP_CLIENT_RELEASE_CHUNK_RETRIES = 2 + +export interface DesktopClientReleaseUploadProgress { + loaded: number + total?: number + percent?: number + stage?: 'uploading' | 'complete' + chunkIndex?: number + chunkCount?: number +} + +interface DesktopClientReleaseUploadSession { + upload_id: string + chunk_size_bytes: number + chunk_count: number + uploaded_chunks: number[] + expires_at: string +} export const platformOptions = [ { label: 'macOS', value: 'darwin' }, @@ -127,6 +147,20 @@ export const desktopClientReleaseApi = { channel: string version: string kind?: 'installer' | 'updater' + onProgress?: (progress: DesktopClientReleaseUploadProgress) => void + }, + ) { + return uploadOSSInChunks(file, target) + }, + async uploadOSSLegacy( + file: File, + target: { + platform: string + arch: string + channel: string + version: string + kind?: 'installer' | 'updater' + onProgress?: (progress: DesktopClientReleaseUploadProgress) => void }, ) { const form = new FormData() @@ -142,6 +176,15 @@ export const desktopClientReleaseApi = { data: DesktopClientReleaseUploadResult }>('/desktop-client/releases/upload', form, { timeout: DESKTOP_CLIENT_RELEASE_UPLOAD_TIMEOUT_MS, + onUploadProgress(event) { + const total = event.total + target.onProgress?.({ + loaded: event.loaded, + total, + percent: total ? Math.min(100, Math.round((event.loaded / total) * 100)) : undefined, + stage: 'uploading', + }) + }, }) return response.data.data }, @@ -155,6 +198,145 @@ export const desktopClientReleaseApi = { }, } +async function uploadOSSInChunks( + file: File, + target: { + platform: string + arch: string + channel: string + version: string + kind?: 'installer' | 'updater' + onProgress?: (progress: DesktopClientReleaseUploadProgress) => void + }, +): Promise { + const session = await http.post('/desktop-client/release-uploads', { + platform: target.platform, + arch: target.arch, + channel: target.channel, + version: target.version, + kind: target.kind ?? 'installer', + file_name: file.name, + file_size_bytes: file.size, + chunk_size_bytes: DESKTOP_CLIENT_RELEASE_CHUNK_SIZE_BYTES, + }) + + const chunkSize = session.chunk_size_bytes || DESKTOP_CLIENT_RELEASE_CHUNK_SIZE_BYTES + const chunkCount = session.chunk_count || Math.ceil(file.size / chunkSize) + const uploaded = new Set(session.uploaded_chunks ?? []) + let confirmedBytes = 0 + for (const index of uploaded) { + confirmedBytes += chunkByteSize(file.size, chunkSize, index, chunkCount) + } + emitChunkProgress(target, confirmedBytes, file.size, chunkCount) + + try { + for (let index = 0; index < chunkCount; index += 1) { + if (uploaded.has(index)) continue + await uploadChunkWithRetry(file, session.upload_id, index, chunkSize, chunkCount, confirmedBytes, target) + confirmedBytes += chunkByteSize(file.size, chunkSize, index, chunkCount) + emitChunkProgress(target, confirmedBytes, file.size, chunkCount, index) + } + + target.onProgress?.({ + loaded: file.size, + total: file.size, + percent: 100, + stage: 'complete', + chunkCount, + }) + return await http.post( + `/desktop-client/release-uploads/${session.upload_id}/complete`, + ) + } catch (error) { + void http.delete(`/desktop-client/release-uploads/${session.upload_id}`).catch(() => undefined) + throw error + } +} + +async function uploadChunkWithRetry( + file: File, + uploadID: string, + index: number, + chunkSize: number, + chunkCount: number, + confirmedBytes: number, + target: { + onProgress?: (progress: DesktopClientReleaseUploadProgress) => void + }, +) { + let lastError: unknown + for (let attempt = 0; attempt <= DESKTOP_CLIENT_RELEASE_CHUNK_RETRIES; attempt += 1) { + try { + await uploadChunk(file, uploadID, index, chunkSize, chunkCount, confirmedBytes, target) + return + } catch (error) { + lastError = error + if (attempt === DESKTOP_CLIENT_RELEASE_CHUNK_RETRIES) break + await delay(400 * (attempt + 1)) + } + } + throw lastError +} + +async function uploadChunk( + file: File, + uploadID: string, + index: number, + chunkSize: number, + chunkCount: number, + confirmedBytes: number, + target: { + onProgress?: (progress: DesktopClientReleaseUploadProgress) => void + }, +) { + const start = index * chunkSize + const end = Math.min(start + chunkSize, file.size) + const chunk = file.slice(start, end) + const form = new FormData() + form.append('file', chunk, file.name) + await http.raw.post(`/desktop-client/release-uploads/${uploadID}/chunks/${index}`, form, { + timeout: DESKTOP_CLIENT_RELEASE_CHUNK_TIMEOUT_MS, + onUploadProgress(event) { + const loaded = confirmedBytes + Math.min(event.loaded, chunk.size) + target.onProgress?.({ + loaded, + total: file.size, + percent: file.size ? Math.min(99, Math.round((loaded / file.size) * 100)) : undefined, + stage: 'uploading', + chunkIndex: index, + chunkCount, + }) + }, + }) +} + +function emitChunkProgress( + target: { onProgress?: (progress: DesktopClientReleaseUploadProgress) => void }, + loaded: number, + total: number, + chunkCount: number, + chunkIndex?: number, +) { + target.onProgress?.({ + loaded, + total, + percent: total ? Math.min(99, Math.round((loaded / total) * 100)) : undefined, + stage: 'uploading', + chunkIndex, + chunkCount, + }) +} + +function chunkByteSize(fileSize: number, chunkSize: number, index: number, chunkCount: number) { + if (index < 0 || index >= chunkCount) return 0 + const start = index * chunkSize + return Math.max(0, Math.min(chunkSize, fileSize - start)) +} + +function delay(ms: number) { + return new Promise((resolve) => window.setTimeout(resolve, ms)) +} + export function platformLabel(value: string): string { return platformOptions.find((item) => item.value === value)?.label ?? value } diff --git a/apps/ops-web/src/views/DesktopClientReleasesView.vue b/apps/ops-web/src/views/DesktopClientReleasesView.vue index d215ac4..611fa4f 100644 --- a/apps/ops-web/src/views/DesktopClientReleasesView.vue +++ b/apps/ops-web/src/views/DesktopClientReleasesView.vue @@ -202,7 +202,7 @@ > - {{ form.oss_object_key ? '重新上传' : '选择文件并上传' }} + {{ uploadButtonText('installer') }}
@@ -268,7 +268,7 @@ > - {{ form.updater_oss_object_key ? '重新上传' : '选择文件并上传' }} + {{ uploadButtonText('updater') }}
@@ -400,6 +400,8 @@ const total = ref(0) const statusLoadingId = ref(null) const uploading = ref(false) const updaterUploading = ref(false) +const uploadProgress = ref(null) +const updaterUploadProgress = ref(null) const enabledRows = computed(() => rows.value.filter((row) => row.enabled).length) const forceRows = computed(() => rows.value.filter((row) => row.force_update).length) @@ -568,8 +570,10 @@ async function beforeUploadReleasePackage(file: File, kind: 'installer' | 'updat } if (kind === 'installer') { uploading.value = true + uploadProgress.value = 0 } else { updaterUploading.value = true + updaterUploadProgress.value = 0 } try { const result = await desktopClientReleaseApi.uploadOSS(file, { @@ -578,6 +582,14 @@ async function beforeUploadReleasePackage(file: File, kind: 'installer' | 'updat channel: form.channel.trim() || 'stable', version: form.version.trim(), kind, + onProgress(progress) { + if (typeof progress.percent !== 'number') return + if (kind === 'installer') { + uploadProgress.value = progress.percent + } else { + updaterUploadProgress.value = progress.percent + } + }, }) if (kind === 'installer') { form.oss_object_key = result.oss_object_key @@ -599,10 +611,25 @@ async function beforeUploadReleasePackage(file: File, kind: 'installer' | 'updat } finally { uploading.value = false updaterUploading.value = false + uploadProgress.value = null + updaterUploadProgress.value = null } return false } +function uploadButtonText(kind: 'installer' | 'updater') { + const isUploading = kind === 'installer' ? uploading.value : updaterUploading.value + const progress = kind === 'installer' ? uploadProgress.value : updaterUploadProgress.value + const hasUploaded = kind === 'installer' ? Boolean(form.oss_object_key) : Boolean(form.updater_oss_object_key) + if (isUploading && typeof progress === 'number' && progress < 100) { + return `上传中 ${progress}%` + } + if (isUploading) { + return '写入 OSS...' + } + return hasUploaded ? '重新上传' : '选择文件并上传' +} + function beforeUploadInstallerPackage(file: File) { return beforeUploadReleasePackage(file, 'installer') } diff --git a/deploy/k3s/config/ops-web-nginx.conf b/deploy/k3s/config/ops-web-nginx.conf index a8ee2a6..eed3e6d 100644 --- a/deploy/k3s/config/ops-web-nginx.conf +++ b/deploy/k3s/config/ops-web-nginx.conf @@ -8,7 +8,7 @@ real_ip_recursive on; server { listen 80; server_name _; - client_max_body_size 600M; + client_max_body_size 1100M; root /usr/share/nginx/html; index index.html; @@ -31,8 +31,8 @@ server { proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $remote_addr; proxy_set_header X-Forwarded-Proto $scheme; - proxy_read_timeout 900s; - proxy_send_timeout 900s; + proxy_read_timeout 1800s; + proxy_send_timeout 1800s; } location ~* \.(js|css|png|jpg|jpeg|gif|ico|svg|woff2?)$ { diff --git a/server/internal/ops/app/desktop_client_release.go b/server/internal/ops/app/desktop_client_release.go index 5c53a64..8af0b45 100644 --- a/server/internal/ops/app/desktop_client_release.go +++ b/server/internal/ops/app/desktop_client_release.go @@ -1,19 +1,26 @@ package app import ( + "bytes" "context" "crypto/sha256" "encoding/hex" + "encoding/json" "errors" "fmt" + "io" "net/http" "net/url" + "os" "path" "path/filepath" "regexp" + "sort" "strings" + "sync" "time" + "github.com/google/uuid" "github.com/jackc/pgx/v5/pgconn" "gopkg.in/yaml.v3" @@ -41,10 +48,22 @@ const ( DesktopClientAssetKindUpdater = "updater" ) +const ( + MaxDesktopClientReleaseUploadBytes int64 = 1 * 1024 * 1024 * 1024 + desktopClientReleaseMultipartOverheadBytes = 4 * 1024 * 1024 + DefaultDesktopClientReleaseChunkSizeBytes = 16 * 1024 * 1024 + MinDesktopClientReleaseChunkSizeBytes = 1 * 1024 * 1024 + MaxDesktopClientReleaseChunkSizeBytes = 64 * 1024 * 1024 + desktopClientReleaseUploadSessionTTL = 24 * time.Hour +) + type DesktopClientReleaseService struct { - releases *repository.DesktopClientReleaseRepository - storage objectstorage.Client - audits *AuditService + releases *repository.DesktopClientReleaseRepository + storage objectstorage.Client + audits *AuditService + releaseUploadDir string + releaseUploadLocks map[string]*sync.Mutex + releaseUploadMu sync.Mutex } func NewDesktopClientReleaseService( @@ -52,7 +71,21 @@ func NewDesktopClientReleaseService( storage objectstorage.Client, audits *AuditService, ) *DesktopClientReleaseService { - return &DesktopClientReleaseService{releases: releases, storage: storage, audits: audits} + return &DesktopClientReleaseService{ + releases: releases, + storage: storage, + audits: audits, + releaseUploadDir: defaultDesktopClientReleaseUploadDir(), + releaseUploadLocks: make(map[string]*sync.Mutex), + } +} + +func (s *DesktopClientReleaseService) WithReleaseUploadDir(dir string) *DesktopClientReleaseService { + if s == nil { + return s + } + s.releaseUploadDir = strings.TrimSpace(dir) + return s } type DesktopClientReleaseView struct { @@ -147,13 +180,15 @@ type DesktopClientReleaseInput struct { } type DesktopClientReleaseUploadInput struct { - Platform string - Arch string - Channel string - Version string - Kind string - FileName string - Content []byte + Platform string + Arch string + Channel string + Version string + Kind string + FileName string + Content []byte + ContentReader io.ReadSeeker + ContentSizeBytes int64 } type DesktopClientReleaseUploadResult struct { @@ -163,6 +198,63 @@ type DesktopClientReleaseUploadResult struct { SHA256 string `json:"sha256"` } +type DesktopClientReleaseChunkedUploadInitInput struct { + Platform string + Arch string + Channel string + Version string + Kind string + FileName string + FileSizeBytes int64 + ChunkSizeBytes int64 +} + +type DesktopClientReleaseChunkedUploadInitResult struct { + UploadID string `json:"upload_id"` + ChunkSizeBytes int64 `json:"chunk_size_bytes"` + ChunkCount int `json:"chunk_count"` + UploadedChunks []int `json:"uploaded_chunks"` + ExpiresAt time.Time `json:"expires_at"` +} + +type DesktopClientReleaseChunkUploadInput struct { + UploadID string + ChunkIndex int + ContentReader io.Reader + ContentSizeBytes int64 +} + +type DesktopClientReleaseChunkUploadResult struct { + UploadID string `json:"upload_id"` + ChunkIndex int `json:"chunk_index"` + ReceivedBytes int64 `json:"received_bytes"` + UploadedChunks []int `json:"uploaded_chunks"` +} + +type desktopClientReleaseUploadManifest struct { + UploadID string `json:"upload_id"` + Platform string `json:"platform"` + Arch string `json:"arch"` + Channel string `json:"channel"` + Version string `json:"version"` + Kind string `json:"kind"` + FileName string `json:"file_name"` + FileSizeBytes int64 `json:"file_size_bytes"` + ChunkSizeBytes int64 `json:"chunk_size_bytes"` + ChunkCount int `json:"chunk_count"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` + ExpiresAt time.Time `json:"expires_at"` + Chunks map[int]desktopClientReleaseUploadChunk `json:"chunks"` +} + +type desktopClientReleaseUploadChunk struct { + Index int `json:"index"` + SizeBytes int64 `json:"size_bytes"` + SHA256 string `json:"sha256"` + UploadedAt time.Time `json:"uploaded_at"` +} + type DesktopClientReleaseCheckInput struct { Platform string Arch string @@ -402,6 +494,10 @@ func (s *DesktopClientReleaseService) UploadOSS(ctx context.Context, in DesktopC if err != nil { return nil, err } + reader, fileSizeBytes, err := desktopClientReleaseUploadReader(in) + if err != nil { + return nil, response.ErrBadRequest(40071, "release_file_read_failed", "客户端安装包读取失败") + } kind := normalizeDesktopClientAssetKind(in.Kind) if kind == DesktopClientAssetKindUpdater { @@ -409,10 +505,12 @@ func (s *DesktopClientReleaseService) UploadOSS(ctx context.Context, in DesktopC return nil, err } } - sum := sha256.Sum256(in.Content) - sha256Hex := hex.EncodeToString(sum[:]) + sha256Hex, err := hashDesktopClientReleaseUpload(reader) + if err != nil { + return nil, response.ErrBadRequest(40071, "release_file_read_failed", "客户端安装包读取失败") + } objectKey := buildDesktopClientReleaseObjectKey(sha256Hex, fileName) - contentType := detectDesktopReleaseContentType(fileName, in.Content) + contentType := detectDesktopReleaseContentType(fileName, nil) exists, err := s.storage.Exists(ctx, objectKey) if err != nil { appErr := response.ErrInternal(50092, "desktop_client_release_upload_failed", err.Error()) @@ -420,7 +518,7 @@ func (s *DesktopClientReleaseService) UploadOSS(ctx context.Context, in DesktopC return nil, appErr } if !exists { - if err := s.storage.PutBytes(ctx, objectKey, in.Content, contentType); err != nil { + if err := putDesktopClientReleaseUpload(ctx, s.storage, objectKey, reader, fileSizeBytes, contentType); err != nil { appErr := response.ErrInternal(50092, "desktop_client_release_upload_failed", err.Error()) appErr.Cause = err return nil, appErr @@ -430,11 +528,239 @@ func (s *DesktopClientReleaseService) UploadOSS(ctx context.Context, in DesktopC return &DesktopClientReleaseUploadResult{ OSSObjectKey: objectKey, FileName: fileName, - FileSizeBytes: int64(len(in.Content)), + FileSizeBytes: fileSizeBytes, SHA256: sha256Hex, }, nil } +func MaxDesktopClientReleaseMultipartBytes() int64 { + return MaxDesktopClientReleaseUploadBytes + desktopClientReleaseMultipartOverheadBytes +} + +func MaxDesktopClientReleaseChunkMultipartBytes() int64 { + return MaxDesktopClientReleaseChunkSizeBytes + desktopClientReleaseMultipartOverheadBytes +} + +func (s *DesktopClientReleaseService) InitiateChunkedUpload( + ctx context.Context, + in DesktopClientReleaseChunkedUploadInitInput, +) (*DesktopClientReleaseChunkedUploadInitResult, error) { + if s.storage == nil { + return nil, response.ErrInternal(50091, "object_storage_unavailable", "object storage is unavailable") + } + if err := s.storage.Validate(); err != nil { + return nil, response.ErrInternal(50091, "object_storage_unavailable", "object storage is unavailable") + } + + uploadInput := DesktopClientReleaseUploadInput{ + Platform: in.Platform, + Arch: in.Arch, + Channel: in.Channel, + Version: in.Version, + Kind: in.Kind, + FileName: in.FileName, + ContentSizeBytes: in.FileSizeBytes, + } + target, fileName, err := normalizeDesktopClientReleaseUploadInput(uploadInput) + if err != nil { + return nil, err + } + chunkSize := normalizeDesktopClientReleaseChunkSize(in.ChunkSizeBytes) + chunkCount := desktopClientReleaseChunkCount(in.FileSizeBytes, chunkSize) + if chunkCount <= 0 { + return nil, response.ErrBadRequest(40071, "empty_release_file", "上传文件不能为空") + } + + _ = s.cleanupExpiredDesktopReleaseUploads(time.Now()) + + uploadID := uuid.NewString() + now := time.Now().UTC() + manifest := desktopClientReleaseUploadManifest{ + UploadID: uploadID, + Platform: target.Platform, + Arch: target.Arch, + Channel: target.Channel, + Version: strings.TrimSpace(in.Version), + Kind: normalizeDesktopClientAssetKind(in.Kind), + FileName: fileName, + FileSizeBytes: in.FileSizeBytes, + ChunkSizeBytes: chunkSize, + ChunkCount: chunkCount, + CreatedAt: now, + UpdatedAt: now, + ExpiresAt: now.Add(desktopClientReleaseUploadSessionTTL), + Chunks: map[int]desktopClientReleaseUploadChunk{}, + } + lock := s.desktopReleaseUploadLock(uploadID) + lock.Lock() + defer lock.Unlock() + + if err := os.MkdirAll(s.desktopReleaseUploadChunkDir(uploadID), 0o700); err != nil { + return nil, response.ErrInternal(50092, "desktop_client_release_upload_failed", "failed to create upload session") + } + if err := s.saveDesktopReleaseUploadManifest(manifest); err != nil { + return nil, response.ErrInternal(50092, "desktop_client_release_upload_failed", "failed to save upload session") + } + return &DesktopClientReleaseChunkedUploadInitResult{ + UploadID: uploadID, + ChunkSizeBytes: chunkSize, + ChunkCount: chunkCount, + UploadedChunks: []int{}, + ExpiresAt: manifest.ExpiresAt, + }, nil +} + +func (s *DesktopClientReleaseService) UploadChunk( + ctx context.Context, + in DesktopClientReleaseChunkUploadInput, +) (*DesktopClientReleaseChunkUploadResult, error) { + uploadID, err := normalizeDesktopClientReleaseUploadID(in.UploadID) + if err != nil { + return nil, err + } + if in.ContentReader == nil { + return nil, response.ErrBadRequest(40071, "release_file_required", "请选择要上传的客户端安装包") + } + lock := s.desktopReleaseUploadLock(uploadID) + lock.Lock() + defer lock.Unlock() + + manifest, err := s.loadDesktopReleaseUploadManifest(uploadID) + if err != nil { + return nil, err + } + if time.Now().UTC().After(manifest.ExpiresAt) { + return nil, response.ErrBadRequest(40075, "release_upload_expired", "上传会话已过期，请重新选择文件上传") + } + if in.ChunkIndex < 0 || in.ChunkIndex >= manifest.ChunkCount { + return nil, response.ErrBadRequest(40076, "invalid_release_upload_chunk", "上传分片序号无效") + } + expectedSize := manifest.expectedChunkSize(in.ChunkIndex) + if in.ContentSizeBytes > 0 && in.ContentSizeBytes != expectedSize { + return nil, response.ErrBadRequest(40076, "invalid_release_upload_chunk", "上传分片大小无效") + } + + chunkDir := s.desktopReleaseUploadChunkDir(uploadID) + if err := os.MkdirAll(chunkDir, 0o700); err != nil { + return nil, response.ErrInternal(50092, "desktop_client_release_upload_failed", "failed to create upload chunk directory") + } + chunkPath := desktopReleaseUploadChunkPath(chunkDir, in.ChunkIndex) + tmpPath := chunkPath + ".tmp" + chunkFile, err := os.OpenFile(tmpPath, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o600) + if err != nil { + return nil, response.ErrInternal(50092, "desktop_client_release_upload_failed", "failed to write upload chunk") + } + hasher := sha256.New() + written, copyErr := io.Copy(io.MultiWriter(chunkFile, hasher), io.LimitReader(in.ContentReader, expectedSize+1)) + closeErr := chunkFile.Close() + if copyErr != nil { + _ = os.Remove(tmpPath) + return nil, response.ErrBadRequest(40071, "release_file_read_failed", "客户端安装包读取失败") + } + if closeErr != nil { + _ = os.Remove(tmpPath) + return nil, response.ErrInternal(50092, "desktop_client_release_upload_failed", "failed to write upload chunk") + } + if written != expectedSize { + _ = os.Remove(tmpPath) + return nil, response.ErrBadRequest(40076, "invalid_release_upload_chunk", "上传分片大小无效") + } + if err := os.Rename(tmpPath, chunkPath); err != nil { + _ = os.Remove(tmpPath) + return nil, response.ErrInternal(50092, "desktop_client_release_upload_failed", "failed to save upload chunk") + } + + now := time.Now().UTC() + manifest.UpdatedAt = now + manifest.Chunks[in.ChunkIndex] = desktopClientReleaseUploadChunk{ + Index: in.ChunkIndex, + SizeBytes: written, + SHA256: hex.EncodeToString(hasher.Sum(nil)), + UploadedAt: now, + } + if err := s.saveDesktopReleaseUploadManifest(manifest); err != nil { + return nil, response.ErrInternal(50092, "desktop_client_release_upload_failed", "failed to update upload session") + } + return &DesktopClientReleaseChunkUploadResult{ + UploadID: uploadID, + ChunkIndex: in.ChunkIndex, + ReceivedBytes: written, + UploadedChunks: manifest.uploadedChunkIndexes(), + }, nil +} + +func (s *DesktopClientReleaseService) CompleteChunkedUpload( + ctx context.Context, + uploadIDValue string, +) (*DesktopClientReleaseUploadResult, error) { + uploadID, err := normalizeDesktopClientReleaseUploadID(uploadIDValue) + if err != nil { + return nil, err + } + lock := s.desktopReleaseUploadLock(uploadID) + lock.Lock() + defer lock.Unlock() + + manifest, err := s.loadDesktopReleaseUploadManifest(uploadID) + if err != nil { + return nil, err + } + if time.Now().UTC().After(manifest.ExpiresAt) { + return nil, response.ErrBadRequest(40075, "release_upload_expired", "上传会话已过期，请重新选择文件上传") + } + if err := manifest.validateComplete(); err != nil { + return nil, err + } + sha256Hex, err := s.hashDesktopReleaseUploadChunks(manifest) + if err != nil { + return nil, err + } + + objectKey := buildDesktopClientReleaseObjectKey(sha256Hex, manifest.FileName) + contentType := detectDesktopReleaseContentType(manifest.FileName, nil) + exists, err := s.storage.Exists(ctx, objectKey) + if err != nil { + appErr := response.ErrInternal(50092, "desktop_client_release_upload_failed", err.Error()) + appErr.Cause = err + return nil, appErr + } + if !exists { + reader, err := s.openDesktopReleaseUploadChunks(manifest) + if err != nil { + return nil, err + } + defer reader.Close() + if err := putDesktopClientReleaseReader(ctx, s.storage, objectKey, reader, manifest.FileSizeBytes, contentType); err != nil { + appErr := response.ErrInternal(50092, "desktop_client_release_upload_failed", err.Error()) + appErr.Cause = err + return nil, appErr + } + } + + result := &DesktopClientReleaseUploadResult{ + OSSObjectKey: objectKey, + FileName: manifest.FileName, + FileSizeBytes: manifest.FileSizeBytes, + SHA256: sha256Hex, + } + _ = os.RemoveAll(s.desktopReleaseUploadSessionDir(uploadID)) + s.forgetDesktopReleaseUploadLock(uploadID) + return result, nil +} + +func (s *DesktopClientReleaseService) AbortChunkedUpload(uploadIDValue string) error { + uploadID, err := normalizeDesktopClientReleaseUploadID(uploadIDValue) + if err != nil { + return err + } + lock := s.desktopReleaseUploadLock(uploadID) + lock.Lock() + defer lock.Unlock() + _ = os.RemoveAll(s.desktopReleaseUploadSessionDir(uploadID)) + s.forgetDesktopReleaseUploadLock(uploadID) + return nil +} + func (s *DesktopClientReleaseService) CheckLatest(ctx context.Context, in DesktopClientReleaseCheckInput) (*DesktopClientReleaseCheckResult, error) { target, currentVersion, err := normalizeDesktopClientReleaseCheckInput(in) if err != nil { @@ -1005,10 +1331,14 @@ func normalizeDesktopReleaseSHA256(value *string) (*string, error) { } func normalizeDesktopClientReleaseUploadInput(in DesktopClientReleaseUploadInput) (repository.DesktopClientReleaseTarget, string, error) { - if len(in.Content) == 0 { + size, err := desktopClientReleaseUploadSize(in) + if err != nil { + return repository.DesktopClientReleaseTarget{}, "", response.ErrBadRequest(40071, "release_file_read_failed", "客户端安装包读取失败") + } + if size == 0 { return repository.DesktopClientReleaseTarget{}, "", response.ErrBadRequest(40071, "empty_release_file", "上传文件不能为空") } - if len(in.Content) > 1024*1024*1024 { + if size > MaxDesktopClientReleaseUploadBytes { return repository.DesktopClientReleaseTarget{}, "", response.ErrBadRequest(40072, "release_file_too_large", "客户端安装包不能超过 1GB") } @@ -1037,6 +1367,354 @@ func normalizeDesktopClientReleaseUploadInput(in DesktopClientReleaseUploadInput return target, fileName, nil } +func desktopClientReleaseUploadSize(in DesktopClientReleaseUploadInput) (int64, error) { + if in.ContentReader == nil { + if in.ContentSizeBytes > 0 { + return in.ContentSizeBytes, nil + } + return int64(len(in.Content)), nil + } + if in.ContentSizeBytes > 0 { + return in.ContentSizeBytes, nil + } + current, err := in.ContentReader.Seek(0, io.SeekCurrent) + if err != nil { + return 0, err + } + end, err := in.ContentReader.Seek(0, io.SeekEnd) + if restoreErr := seekDesktopClientReleaseUpload(in.ContentReader, current); err == nil { + err = restoreErr + } + if err != nil { + return 0, err + } + return end, nil +} + +func desktopClientReleaseUploadReader(in DesktopClientReleaseUploadInput) (io.ReadSeeker, int64, error) { + size, err := desktopClientReleaseUploadSize(in) + if err != nil { + return nil, 0, err + } + if in.ContentReader != nil { + if err := seekDesktopClientReleaseUpload(in.ContentReader, 0); err != nil { + return nil, 0, err + } + return in.ContentReader, size, nil + } + return bytes.NewReader(in.Content), size, nil +} + +func hashDesktopClientReleaseUpload(reader io.ReadSeeker) (string, error) { + if err := seekDesktopClientReleaseUpload(reader, 0); err != nil { + return "", err + } + hasher := sha256.New() + if _, err := io.Copy(hasher, reader); err != nil { + return "", err + } + return hex.EncodeToString(hasher.Sum(nil)), nil +} + +func putDesktopClientReleaseUpload( + ctx context.Context, + storage objectstorage.Client, + objectKey string, + reader io.ReadSeeker, + size int64, + contentType string, +) error { + if err := seekDesktopClientReleaseUpload(reader, 0); err != nil { + return err + } + return putDesktopClientReleaseReader(ctx, storage, objectKey, reader, size, contentType) +} + +func putDesktopClientReleaseReader( + ctx context.Context, + storage objectstorage.Client, + objectKey string, + reader io.Reader, + size int64, + contentType string, +) error { + if readerClient, ok := storage.(objectstorage.ReaderClient); ok { + return readerClient.PutReader(ctx, objectKey, reader, size, contentType) + } + content, err := io.ReadAll(io.LimitReader(reader, MaxDesktopClientReleaseUploadBytes+1)) + if err != nil { + return err + } + if int64(len(content)) > MaxDesktopClientReleaseUploadBytes { + return response.ErrBadRequest(40072, "release_file_too_large", "客户端安装包不能超过 1GB") + } + return storage.PutBytes(ctx, objectKey, content, contentType) +} + +func defaultDesktopClientReleaseUploadDir() string { + return filepath.Join(os.TempDir(), "geo-rankly-desktop-release-uploads") +} + +func normalizeDesktopClientReleaseChunkSize(value int64) int64 { + switch { + case value <= 0: + return DefaultDesktopClientReleaseChunkSizeBytes + case value < MinDesktopClientReleaseChunkSizeBytes: + return MinDesktopClientReleaseChunkSizeBytes + case value > MaxDesktopClientReleaseChunkSizeBytes: + return MaxDesktopClientReleaseChunkSizeBytes + default: + return value + } +} + +func desktopClientReleaseChunkCount(fileSizeBytes int64, chunkSizeBytes int64) int { + if fileSizeBytes <= 0 || chunkSizeBytes <= 0 { + return 0 + } + return int((fileSizeBytes + chunkSizeBytes - 1) / chunkSizeBytes) +} + +func normalizeDesktopClientReleaseUploadID(value string) (string, error) { + uploadID := strings.TrimSpace(value) + if _, err := uuid.Parse(uploadID); err != nil { + return "", response.ErrBadRequest(40074, "invalid_release_upload_id", "上传会话不存在或已失效") + } + return uploadID, nil +} + +func (s *DesktopClientReleaseService) desktopReleaseUploadBaseDir() string { + dir := strings.TrimSpace(s.releaseUploadDir) + if dir == "" { + return defaultDesktopClientReleaseUploadDir() + } + return dir +} + +func (s *DesktopClientReleaseService) desktopReleaseUploadSessionDir(uploadID string) string { + return filepath.Join(s.desktopReleaseUploadBaseDir(), uploadID) +} + +func (s *DesktopClientReleaseService) desktopReleaseUploadChunkDir(uploadID string) string { + return filepath.Join(s.desktopReleaseUploadSessionDir(uploadID), "chunks") +} + +func (s *DesktopClientReleaseService) desktopReleaseUploadManifestPath(uploadID string) string { + return filepath.Join(s.desktopReleaseUploadSessionDir(uploadID), "manifest.json") +} + +func desktopReleaseUploadChunkPath(chunkDir string, index int) string { + return filepath.Join(chunkDir, fmt.Sprintf("%06d.part", index)) +} + +func (s *DesktopClientReleaseService) desktopReleaseUploadLock(uploadID string) *sync.Mutex { + s.releaseUploadMu.Lock() + defer s.releaseUploadMu.Unlock() + if s.releaseUploadLocks == nil { + s.releaseUploadLocks = make(map[string]*sync.Mutex) + } + if lock := s.releaseUploadLocks[uploadID]; lock != nil { + return lock + } + lock := &sync.Mutex{} + s.releaseUploadLocks[uploadID] = lock + return lock +} + +func (s *DesktopClientReleaseService) forgetDesktopReleaseUploadLock(uploadID string) { + s.releaseUploadMu.Lock() + defer s.releaseUploadMu.Unlock() + delete(s.releaseUploadLocks, uploadID) +} + +func (s *DesktopClientReleaseService) saveDesktopReleaseUploadManifest(manifest desktopClientReleaseUploadManifest) error { + if manifest.Chunks == nil { + manifest.Chunks = map[int]desktopClientReleaseUploadChunk{} + } + sessionDir := s.desktopReleaseUploadSessionDir(manifest.UploadID) + if err := os.MkdirAll(sessionDir, 0o700); err != nil { + return err + } + data, err := json.MarshalIndent(manifest, "", " ") + if err != nil { + return err + } + manifestPath := s.desktopReleaseUploadManifestPath(manifest.UploadID) + tmpPath := manifestPath + ".tmp" + if err := os.WriteFile(tmpPath, data, 0o600); err != nil { + return err + } + return os.Rename(tmpPath, manifestPath) +} + +func (s *DesktopClientReleaseService) loadDesktopReleaseUploadManifest(uploadID string) (desktopClientReleaseUploadManifest, error) { + data, err := os.ReadFile(s.desktopReleaseUploadManifestPath(uploadID)) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + return desktopClientReleaseUploadManifest{}, response.ErrBadRequest(40074, "invalid_release_upload_id", "上传会话不存在或已失效") + } + return desktopClientReleaseUploadManifest{}, response.ErrInternal(50092, "desktop_client_release_upload_failed", "failed to read upload session") + } + var manifest desktopClientReleaseUploadManifest + if err := json.Unmarshal(data, &manifest); err != nil { + return desktopClientReleaseUploadManifest{}, response.ErrBadRequest(40074, "invalid_release_upload_id", "上传会话不存在或已失效") + } + if manifest.UploadID != uploadID || manifest.Chunks == nil { + return desktopClientReleaseUploadManifest{}, response.ErrBadRequest(40074, "invalid_release_upload_id", "上传会话不存在或已失效") + } + return manifest, nil +} + +func (s *DesktopClientReleaseService) cleanupExpiredDesktopReleaseUploads(now time.Time) error { + baseDir := s.desktopReleaseUploadBaseDir() + entries, err := os.ReadDir(baseDir) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + return nil + } + return err + } + for _, entry := range entries { + if !entry.IsDir() { + continue + } + uploadID := entry.Name() + if _, err := normalizeDesktopClientReleaseUploadID(uploadID); err != nil { + continue + } + manifest, err := s.loadDesktopReleaseUploadManifest(uploadID) + if err != nil || now.UTC().After(manifest.ExpiresAt) { + _ = os.RemoveAll(s.desktopReleaseUploadSessionDir(uploadID)) + s.forgetDesktopReleaseUploadLock(uploadID) + } + } + return nil +} + +func (m desktopClientReleaseUploadManifest) expectedChunkSize(index int) int64 { + if index < 0 || index >= m.ChunkCount || m.ChunkSizeBytes <= 0 { + return 0 + } + if index == m.ChunkCount-1 { + return m.FileSizeBytes - int64(index)*m.ChunkSizeBytes + } + return m.ChunkSizeBytes +} + +func (m desktopClientReleaseUploadManifest) uploadedChunkIndexes() []int { + indexes := make([]int, 0, len(m.Chunks)) + for index := range m.Chunks { + indexes = append(indexes, index) + } + sort.Ints(indexes) + return indexes +} + +func (m desktopClientReleaseUploadManifest) validateComplete() error { + if m.ChunkCount <= 0 || len(m.Chunks) != m.ChunkCount { + return response.ErrBadRequest(40077, "release_upload_incomplete", "上传分片尚未全部完成") + } + for index := 0; index < m.ChunkCount; index++ { + chunk, ok := m.Chunks[index] + if !ok || chunk.SizeBytes != m.expectedChunkSize(index) { + return response.ErrBadRequest(40077, "release_upload_incomplete", "上传分片尚未全部完成") + } + } + return nil +} + +func (s *DesktopClientReleaseService) hashDesktopReleaseUploadChunks( + manifest desktopClientReleaseUploadManifest, +) (string, error) { + hasher := sha256.New() + chunkDir := s.desktopReleaseUploadChunkDir(manifest.UploadID) + for index := 0; index < manifest.ChunkCount; index++ { + chunk := manifest.Chunks[index] + if err := copyDesktopReleaseChunkToWriter(hasher, desktopReleaseUploadChunkPath(chunkDir, index), manifest.expectedChunkSize(index), chunk.SHA256); err != nil { + return "", err + } + } + return hex.EncodeToString(hasher.Sum(nil)), nil +} + +func copyDesktopReleaseChunkToWriter(dst io.Writer, chunkPath string, expectedSize int64, expectedSHA256 string) error { + file, err := os.Open(chunkPath) + if err != nil { + return response.ErrBadRequest(40077, "release_upload_incomplete", "上传分片尚未全部完成") + } + defer file.Close() + hasher := sha256.New() + written, err := io.Copy(io.MultiWriter(dst, hasher), file) + if err != nil { + return response.ErrBadRequest(40071, "release_file_read_failed", "客户端安装包读取失败") + } + if written != expectedSize { + return response.ErrBadRequest(40077, "release_upload_incomplete", "上传分片尚未全部完成") + } + if sha256Hex := hex.EncodeToString(hasher.Sum(nil)); expectedSHA256 != "" && sha256Hex != expectedSHA256 { + return response.ErrBadRequest(40077, "release_upload_incomplete", "上传分片校验失败，请重新上传") + } + return nil +} + +func (s *DesktopClientReleaseService) openDesktopReleaseUploadChunks( + manifest desktopClientReleaseUploadManifest, +) (*desktopClientReleaseChunkReadCloser, error) { + chunkDir := s.desktopReleaseUploadChunkDir(manifest.UploadID) + closers := make([]io.Closer, 0, manifest.ChunkCount) + readers := make([]io.Reader, 0, manifest.ChunkCount) + for index := 0; index < manifest.ChunkCount; index++ { + chunkPath := desktopReleaseUploadChunkPath(chunkDir, index) + info, err := os.Stat(chunkPath) + if err != nil || info.Size() != manifest.expectedChunkSize(index) { + for _, closer := range closers { + _ = closer.Close() + } + return nil, response.ErrBadRequest(40077, "release_upload_incomplete", "上传分片尚未全部完成") + } + file, err := os.Open(chunkPath) + if err != nil { + for _, closer := range closers { + _ = closer.Close() + } + return nil, response.ErrBadRequest(40077, "release_upload_incomplete", "上传分片尚未全部完成") + } + closers = append(closers, file) + readers = append(readers, file) + } + return &desktopClientReleaseChunkReadCloser{ + reader: io.MultiReader(readers...), + closers: closers, + }, nil +} + +type desktopClientReleaseChunkReadCloser struct { + reader io.Reader + closers []io.Closer +} + +func (r *desktopClientReleaseChunkReadCloser) Read(p []byte) (int, error) { + return r.reader.Read(p) +} + +func (r *desktopClientReleaseChunkReadCloser) Close() error { + var closeErr error + for _, closer := range r.closers { + if err := closer.Close(); err != nil && closeErr == nil { + closeErr = err + } + } + return closeErr +} + +func seekDesktopClientReleaseUpload(reader io.ReadSeeker, offset int64) error { + if reader == nil { + return fmt.Errorf("release upload reader is required") + } + _, err := reader.Seek(offset, io.SeekStart) + return err +} + func normalizeDesktopClientAssetKind(value string) string { switch normalizeDesktopReleaseEnum(value) { case DesktopClientAssetKindUpdater: diff --git a/server/internal/ops/app/desktop_client_release_test.go b/server/internal/ops/app/desktop_client_release_test.go index 0500cde..748c191 100644 --- a/server/internal/ops/app/desktop_client_release_test.go +++ b/server/internal/ops/app/desktop_client_release_test.go @@ -1,8 +1,13 @@ package app import ( + "bytes" "context" + "crypto/sha256" + "encoding/hex" "errors" + "io" + "os" "testing" "time" @@ -213,9 +218,60 @@ func TestResolveDownloadURLResultUsesSignedDownloadURL(t *testing.T) { require.Empty(t, storage.publicURLCalls) } +func TestDesktopClientReleaseChunkedUploadCompletesOutOfOrder(t *testing.T) { + t.Parallel() + + storage := &desktopReleaseDownloadURLStorage{ + existsByKey: map[string]bool{}, + objects: map[string][]byte{}, + } + svc := NewDesktopClientReleaseService(nil, storage, nil).WithReleaseUploadDir(t.TempDir()) + content := bytes.Repeat([]byte("release-package-"), 1024*1024) + sum := sha256.Sum256(content) + wantSHA := hex.EncodeToString(sum[:]) + + initResult, err := svc.InitiateChunkedUpload(context.Background(), DesktopClientReleaseChunkedUploadInitInput{ + Platform: "darwin", + Arch: "arm64", + Channel: "stable", + Version: "1.2.3", + Kind: DesktopClientAssetKindInstaller, + FileName: "client.dmg", + FileSizeBytes: int64(len(content)), + ChunkSizeBytes: 5 * 1024 * 1024, + }) + require.NoError(t, err) + require.Equal(t, 4, initResult.ChunkCount) + + order := []int{2, 0, 3, 1} + for _, index := range order { + start := int64(index) * initResult.ChunkSizeBytes + end := min(start+initResult.ChunkSizeBytes, int64(len(content))) + result, err := svc.UploadChunk(context.Background(), DesktopClientReleaseChunkUploadInput{ + UploadID: initResult.UploadID, + ChunkIndex: index, + ContentReader: bytes.NewReader(content[start:end]), + ContentSizeBytes: end - start, + }) + require.NoError(t, err) + require.Equal(t, index, result.ChunkIndex) + } + + result, err := svc.CompleteChunkedUpload(context.Background(), initResult.UploadID) + require.NoError(t, err) + require.Equal(t, wantSHA, result.SHA256) + require.Equal(t, int64(len(content)), result.FileSizeBytes) + require.Equal(t, "desktop-client/releases/blobs/"+wantSHA+".dmg", result.OSSObjectKey) + require.Equal(t, content, storage.objects[result.OSSObjectKey]) + _, err = os.Stat(svc.desktopReleaseUploadSessionDir(initResult.UploadID)) + require.True(t, errors.Is(err, os.ErrNotExist)) +} + type desktopReleaseDownloadURLStorage struct { publicURLCalls []string downloadURLCalls []string + existsByKey map[string]bool + objects map[string][]byte } func (s *desktopReleaseDownloadURLStorage) Validate() error { @@ -226,11 +282,30 @@ func (s *desktopReleaseDownloadURLStorage) PutBytes(context.Context, string, []b return nil } +func (s *desktopReleaseDownloadURLStorage) PutReader(_ context.Context, objectKey string, reader io.Reader, _ int64, _ string) error { + if s.objects == nil { + s.objects = map[string][]byte{} + } + content, err := io.ReadAll(reader) + if err != nil { + return err + } + s.objects[objectKey] = content + if s.existsByKey == nil { + s.existsByKey = map[string]bool{} + } + s.existsByKey[objectKey] = true + return nil +} + func (s *desktopReleaseDownloadURLStorage) GetBytes(context.Context, string) ([]byte, error) { return nil, errors.New("not implemented") } -func (s *desktopReleaseDownloadURLStorage) Exists(context.Context, string) (bool, error) { +func (s *desktopReleaseDownloadURLStorage) Exists(_ context.Context, objectKey string) (bool, error) { + if s.existsByKey != nil { + return s.existsByKey[objectKey], nil + } return true, nil } diff --git a/server/internal/ops/transport/desktop_client_release_handler.go b/server/internal/ops/transport/desktop_client_release_handler.go index 16c94f2..7684aa4 100644 --- a/server/internal/ops/transport/desktop_client_release_handler.go +++ b/server/internal/ops/transport/desktop_client_release_handler.go @@ -2,7 +2,6 @@ package transport import ( "fmt" - "io" "net/http" "strconv" "strings" @@ -40,6 +39,17 @@ type setDesktopClientReleaseEnabledRequest struct { Enabled bool `json:"enabled"` } +type desktopClientReleaseUploadInitRequest struct { + Platform string `json:"platform" binding:"required"` + Arch string `json:"arch"` + Channel string `json:"channel"` + Version string `json:"version" binding:"required"` + Kind string `json:"kind"` + FileName string `json:"file_name" binding:"required"` + FileSizeBytes int64 `json:"file_size_bytes" binding:"required"` + ChunkSizeBytes int64 `json:"chunk_size_bytes"` +} + func listDesktopClientReleasesHandler(svc *app.DesktopClientReleaseService) gin.HandlerFunc { return func(c *gin.Context) { page, _ := strconv.Atoi(c.DefaultQuery("page", "1")) @@ -151,11 +161,28 @@ func resolveDesktopClientReleaseDownloadURLHandler(svc *app.DesktopClientRelease func uploadDesktopClientReleaseHandler(svc *app.DesktopClientReleaseService) gin.HandlerFunc { return func(c *gin.Context) { + if c.Request.ContentLength > app.MaxDesktopClientReleaseMultipartBytes() { + response.Error(c, response.ErrBadRequest(40072, "release_file_too_large", "客户端安装包不能超过 1GB")) + return + } + c.Request.Body = http.MaxBytesReader(c.Writer, c.Request.Body, app.MaxDesktopClientReleaseMultipartBytes()) + fileHeader, err := c.FormFile("file") if err != nil { + if strings.Contains(err.Error(), "request body too large") { + response.Error(c, response.ErrBadRequest(40072, "release_file_too_large", "客户端安装包不能超过 1GB")) + return + } response.Error(c, response.ErrBadRequest(40071, "release_file_required", "请选择要上传的客户端安装包")) return } + if c.Request.MultipartForm != nil { + defer c.Request.MultipartForm.RemoveAll() + } + if fileHeader.Size > app.MaxDesktopClientReleaseUploadBytes { + response.Error(c, response.ErrBadRequest(40072, "release_file_too_large", "客户端安装包不能超过 1GB")) + return + } file, err := fileHeader.Open() if err != nil { response.Error(c, response.ErrBadRequest(40071, "release_file_open_failed", "客户端安装包读取失败")) @@ -163,20 +190,15 @@ func uploadDesktopClientReleaseHandler(svc *app.DesktopClientReleaseService) gin } defer file.Close() - content, err := io.ReadAll(file) - if err != nil { - response.Error(c, response.ErrBadRequest(40071, "release_file_read_failed", "客户端安装包读取失败")) - return - } - result, err := svc.UploadOSS(c.Request.Context(), app.DesktopClientReleaseUploadInput{ - Platform: c.PostForm("platform"), - Arch: c.PostForm("arch"), - Channel: c.PostForm("channel"), - Version: c.PostForm("version"), - Kind: c.DefaultPostForm("kind", app.DesktopClientAssetKindInstaller), - FileName: fileHeader.Filename, - Content: content, + Platform: c.PostForm("platform"), + Arch: c.PostForm("arch"), + Channel: c.PostForm("channel"), + Version: c.PostForm("version"), + Kind: c.DefaultPostForm("kind", app.DesktopClientAssetKindInstaller), + FileName: fileHeader.Filename, + ContentReader: file, + ContentSizeBytes: fileHeader.Size, }) if err != nil { response.Error(c, err) @@ -186,6 +208,102 @@ func uploadDesktopClientReleaseHandler(svc *app.DesktopClientReleaseService) gin } } +func initiateDesktopClientReleaseUploadHandler(svc *app.DesktopClientReleaseService) gin.HandlerFunc { + return func(c *gin.Context) { + var body desktopClientReleaseUploadInitRequest + if err := c.ShouldBindJSON(&body); err != nil { + response.Error(c, response.ErrBadRequest(40000, "invalid_payload", err.Error())) + return + } + result, err := svc.InitiateChunkedUpload(c.Request.Context(), app.DesktopClientReleaseChunkedUploadInitInput{ + Platform: body.Platform, + Arch: body.Arch, + Channel: body.Channel, + Version: body.Version, + Kind: body.Kind, + FileName: body.FileName, + FileSizeBytes: body.FileSizeBytes, + ChunkSizeBytes: body.ChunkSizeBytes, + }) + if err != nil { + response.Error(c, err) + return + } + response.Success(c, result) + } +} + +func uploadDesktopClientReleaseChunkHandler(svc *app.DesktopClientReleaseService) gin.HandlerFunc { + return func(c *gin.Context) { + if c.Request.ContentLength > app.MaxDesktopClientReleaseChunkMultipartBytes() { + response.Error(c, response.ErrBadRequest(40076, "invalid_release_upload_chunk", "上传分片不能超过 64MB")) + return + } + c.Request.Body = http.MaxBytesReader(c.Writer, c.Request.Body, app.MaxDesktopClientReleaseChunkMultipartBytes()) + + chunkIndex, err := strconv.Atoi(c.Param("chunk_index")) + if err != nil { + response.Error(c, response.ErrBadRequest(40076, "invalid_release_upload_chunk", "上传分片序号无效")) + return + } + fileHeader, err := c.FormFile("file") + if err != nil { + if strings.Contains(err.Error(), "request body too large") { + response.Error(c, response.ErrBadRequest(40076, "invalid_release_upload_chunk", "上传分片不能超过 64MB")) + return + } + response.Error(c, response.ErrBadRequest(40071, "release_file_required", "请选择要上传的客户端安装包")) + return + } + if c.Request.MultipartForm != nil { + defer c.Request.MultipartForm.RemoveAll() + } + if fileHeader.Size > app.MaxDesktopClientReleaseChunkSizeBytes { + response.Error(c, response.ErrBadRequest(40076, "invalid_release_upload_chunk", "上传分片不能超过 64MB")) + return + } + file, err := fileHeader.Open() + if err != nil { + response.Error(c, response.ErrBadRequest(40071, "release_file_open_failed", "客户端安装包读取失败")) + return + } + defer file.Close() + + result, err := svc.UploadChunk(c.Request.Context(), app.DesktopClientReleaseChunkUploadInput{ + UploadID: c.Param("upload_id"), + ChunkIndex: chunkIndex, + ContentReader: file, + ContentSizeBytes: fileHeader.Size, + }) + if err != nil { + response.Error(c, err) + return + } + response.Success(c, result) + } +} + +func completeDesktopClientReleaseUploadHandler(svc *app.DesktopClientReleaseService) gin.HandlerFunc { + return func(c *gin.Context) { + result, err := svc.CompleteChunkedUpload(c.Request.Context(), c.Param("upload_id")) + if err != nil { + response.Error(c, err) + return + } + response.Success(c, result) + } +} + +func abortDesktopClientReleaseUploadHandler(svc *app.DesktopClientReleaseService) gin.HandlerFunc { + return func(c *gin.Context) { + if err := svc.AbortChunkedUpload(c.Param("upload_id")); err != nil { + response.Error(c, err) + return + } + response.Success(c, gin.H{"upload_id": c.Param("upload_id")}) + } +} + func checkDesktopClientReleaseHandler(svc *app.DesktopClientReleaseService) gin.HandlerFunc { return func(c *gin.Context) { result, err := svc.CheckLatest(c.Request.Context(), app.DesktopClientReleaseCheckInput{ diff --git a/server/internal/ops/transport/router.go b/server/internal/ops/transport/router.go index 8bf55a4..96169ed 100644 --- a/server/internal/ops/transport/router.go +++ b/server/internal/ops/transport/router.go @@ -173,6 +173,10 @@ func RegisterRoutes(d Deps) { authed.GET("/desktop-client/releases", listDesktopClientReleasesHandler(d.Releases)) authed.POST("/desktop-client/releases/upload", uploadDesktopClientReleaseHandler(d.Releases)) + authed.POST("/desktop-client/release-uploads", initiateDesktopClientReleaseUploadHandler(d.Releases)) + authed.POST("/desktop-client/release-uploads/:upload_id/chunks/:chunk_index", uploadDesktopClientReleaseChunkHandler(d.Releases)) + authed.POST("/desktop-client/release-uploads/:upload_id/complete", completeDesktopClientReleaseUploadHandler(d.Releases)) + authed.DELETE("/desktop-client/release-uploads/:upload_id", abortDesktopClientReleaseUploadHandler(d.Releases)) authed.POST("/desktop-client/releases", createDesktopClientReleaseHandler(d.Releases)) authed.PATCH("/desktop-client/releases/:id", updateDesktopClientReleaseHandler(d.Releases)) authed.GET("/desktop-client/releases/:id/download-url", resolveDesktopClientReleaseDownloadURLHandler(d.Releases)) diff --git a/server/internal/shared/objectstorage/aliyun.go b/server/internal/shared/objectstorage/aliyun.go index d357129..712c903 100644 --- a/server/internal/shared/objectstorage/aliyun.go +++ b/server/internal/shared/objectstorage/aliyun.go @@ -72,12 +72,25 @@ func (c *aliyunClient) Validate() error { } func (c *aliyunClient) PutBytes(ctx context.Context, objectKey string, content []byte, contentType string) error { + return c.PutReader(ctx, objectKey, bytes.NewReader(content), int64(len(content)), contentType) +} + +func (c *aliyunClient) PutReader( + ctx context.Context, + objectKey string, + reader io.Reader, + size int64, + contentType string, +) error { if err := c.Validate(); err != nil { return err } if strings.TrimSpace(objectKey) == "" { return fmt.Errorf("object key is required") } + if reader == nil { + return fmt.Errorf("object reader is required") + } exists, err := c.client.IsBucketExist(c.bucket) if err != nil { @@ -113,11 +126,11 @@ func (c *aliyunClient) PutBytes(ctx context.Context, objectKey string, content [ if err != nil { return err } - if err := bucket.PutObject(objectKey, bytes.NewReader(content), options...); err != nil { + if err := bucket.PutObject(objectKey, reader, options...); err != nil { c.logError(ctx, "aliyun put object failed", zap.String("bucket", c.bucket), zap.String("object_key", objectKey), - zap.Int("content_size", len(content)), + zap.Int64("content_size", size), zap.String("object_acl", strings.TrimSpace(c.cfg.ObjectACL)), zap.String("content_type", contentType), zap.String("aliyun_error_code", aliyunErrorCode(err)), diff --git a/server/internal/shared/objectstorage/client.go b/server/internal/shared/objectstorage/client.go index c212e3e..7417496 100644 --- a/server/internal/shared/objectstorage/client.go +++ b/server/internal/shared/objectstorage/client.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "io" "strings" "time" @@ -27,6 +28,10 @@ type Client interface { DownloadURL(objectKey string) (string, error) } +type ReaderClient interface { + PutReader(ctx context.Context, objectKey string, reader io.Reader, size int64, contentType string) error +} + type ManagementClient interface { Client List(ctx context.Context, in ListInput) (*ListResult, error) @@ -106,6 +111,10 @@ func (c disabledClient) PutBytes(context.Context, string, []byte, string) error return c.Validate() } +func (c disabledClient) PutReader(context.Context, string, io.Reader, int64, string) error { + return c.Validate() +} + func (c disabledClient) GetBytes(context.Context, string) ([]byte, error) { return nil, c.Validate() } diff --git a/server/internal/shared/objectstorage/minio.go b/server/internal/shared/objectstorage/minio.go index 825f7e6..35c0e1f 100644 --- a/server/internal/shared/objectstorage/minio.go +++ b/server/internal/shared/objectstorage/minio.go @@ -103,6 +103,16 @@ func (c *minioClient) PutBytes( objectKey string, content []byte, contentType string, +) error { + return c.PutReader(ctx, objectKey, bytes.NewReader(content), int64(len(content)), contentType) +} + +func (c *minioClient) PutReader( + ctx context.Context, + objectKey string, + reader io.Reader, + size int64, + contentType string, ) error { if err := c.Validate(); err != nil { return err @@ -110,6 +120,9 @@ func (c *minioClient) PutBytes( if strings.TrimSpace(objectKey) == "" { return fmt.Errorf("object key is required") } + if reader == nil { + return fmt.Errorf("object reader is required") + } exists, err := c.client.BucketExists(ctx, c.bucket) if err != nil { @@ -131,14 +144,14 @@ func (c *minioClient) PutBytes( } } - _, err = c.client.PutObject(ctx, c.bucket, objectKey, bytes.NewReader(content), int64(len(content)), minio.PutObjectOptions{ + _, err = c.client.PutObject(ctx, c.bucket, objectKey, reader, size, minio.PutObjectOptions{ ContentType: contentType, }) if err != nil { c.logError(ctx, "minio put object failed", zap.String("bucket", c.bucket), zap.String("object_key", objectKey), - zap.Int("content_size", len(content)), + zap.Int64("content_size", size), zap.Error(err), ) return fmt.Errorf("put minio object: %w", err) diff --git a/server/internal/shared/objectstorage/reloadable.go b/server/internal/shared/objectstorage/reloadable.go index ce5473c..3c1143b 100644 --- a/server/internal/shared/objectstorage/reloadable.go +++ b/server/internal/shared/objectstorage/reloadable.go @@ -2,6 +2,7 @@ package objectstorage import ( "context" + "io" "sync" ) @@ -31,6 +32,19 @@ func (c *ReloadableClient) PutBytes(ctx context.Context, objectKey string, conte return c.snapshot().PutBytes(ctx, objectKey, content, contentType) } +func (c *ReloadableClient) PutReader(ctx context.Context, objectKey string, reader io.Reader, size int64, contentType string) error { + client := c.snapshot() + readerClient, ok := client.(ReaderClient) + if !ok { + content, err := io.ReadAll(reader) + if err != nil { + return err + } + return client.PutBytes(ctx, objectKey, content, contentType) + } + return readerClient.PutReader(ctx, objectKey, reader, size, contentType) +} + func (c *ReloadableClient) GetBytes(ctx context.Context, objectKey string) ([]byte, error) { return c.snapshot().GetBytes(ctx, objectKey) } diff --git a/server/internal/shared/swagger/descriptions.go b/server/internal/shared/swagger/descriptions.go index 2301bb5..193ba11 100644 --- a/server/internal/shared/swagger/descriptions.go +++ b/server/internal/shared/swagger/descriptions.go @@ -99,17 +99,21 @@ var routeDocs = map[string]routeDoc{ "DELETE /api/ops/object-storage/folders": {"删除对象存储目录", "按前缀递归删除目录下对象。"}, // --- Ops：桌面客户端版本 --- - "GET /api/ops/desktop-client/releases": {"桌面客户端版本列表", "分页查询桌面客户端版本配置。"}, - "POST /api/ops/desktop-client/releases/upload": {"上传桌面客户端包", "Multipart 上传桌面客户端安装包或更新包。"}, - "POST /api/ops/desktop-client/releases": {"新建桌面客户端版本", "创建桌面客户端版本配置。"}, - "PATCH /api/ops/desktop-client/releases/:id": {"更新桌面客户端版本", "更新桌面客户端版本配置。"}, - "GET /api/ops/desktop-client/releases/:id/download-url": {"解析桌面客户端下载地址", "解析指定桌面客户端版本的下载地址。"}, - "POST /api/ops/desktop-client/releases/:id/enabled": {"切换桌面客户端版本状态", "启用或禁用指定桌面客户端版本。"}, - "DELETE /api/ops/desktop-client/releases/:id": {"删除桌面客户端版本", "删除指定桌面客户端版本配置。"}, - "GET /api/ops/desktop-client/bug-reports": {"客户端 Bug 列表", "分页查询桌面客户端崩溃和用户反馈记录。"}, - "GET /api/ops/desktop-client/bug-reports/:id": {"客户端 Bug 详情", "查看桌面客户端崩溃元数据、运行快照和 dump 对象。"}, - "PATCH /api/ops/desktop-client/bug-reports/:id": {"更新客户端 Bug 状态", "调整客户端 Bug 状态、级别和处理备注。"}, - "GET /api/ops/desktop-client/bug-reports/:id/dump-url": {"解析客户端 dump 下载地址", "生成指定客户端 Bug dump 文件的短期下载地址。"}, + "GET /api/ops/desktop-client/releases": {"桌面客户端版本列表", "分页查询桌面客户端版本配置。"}, + "POST /api/ops/desktop-client/releases/upload": {"上传桌面客户端包（兼容）", "兼容旧版单请求 Multipart 上传；新版前端使用分片上传。"}, + "POST /api/ops/desktop-client/release-uploads": {"初始化客户端包分片上传", "创建分片上传会话，返回 upload_id、分片大小和分片数量。"}, + "POST /api/ops/desktop-client/release-uploads/:upload_id/chunks/:chunk_index": {"上传客户端包分片", "Multipart 上传单个客户端安装包或更新包分片。"}, + "POST /api/ops/desktop-client/release-uploads/:upload_id/complete": {"完成客户端包分片上传", "校验全部分片，流式写入 OSS，并返回 Object Key、文件大小和 SHA256。"}, + "DELETE /api/ops/desktop-client/release-uploads/:upload_id": {"取消客户端包分片上传", "取消分片上传会话并清理临时分片。"}, + "POST /api/ops/desktop-client/releases": {"新建客户端版本", "创建桌面客户端版本配置。"}, + "PATCH /api/ops/desktop-client/releases/:id": {"更新客户端版本", "更新桌面客户端版本配置。"}, + "GET /api/ops/desktop-client/releases/:id/download-url": {"解析桌面客户端下载地址", "解析指定桌面客户端版本的下载地址。"}, + "POST /api/ops/desktop-client/releases/:id/enabled": {"切换客户端版本状态", "启用或禁用指定桌面客户端版本。"}, + "DELETE /api/ops/desktop-client/releases/:id": {"删除客户端版本", "删除指定桌面客户端版本配置。"}, + "GET /api/ops/desktop-client/bug-reports": {"客户端 Bug 列表", "分页查询桌面客户端崩溃和用户反馈记录。"}, + "GET /api/ops/desktop-client/bug-reports/:id": {"客户端 Bug 详情", "查看桌面客户端崩溃元数据、运行快照和 dump 对象。"}, + "PATCH /api/ops/desktop-client/bug-reports/:id": {"更新客户端 Bug 状态", "调整客户端 Bug 状态、级别和处理备注。"}, + "GET /api/ops/desktop-client/bug-reports/:id/dump-url": {"解析客户端 dump 下载地址", "生成指定客户端 Bug dump 文件的短期下载地址。"}, // --- Ops：合规 --- "GET /api/ops/compliance/dictionaries": {"合规词库列表", "分页查询内容合规词库。"}, diff --git a/server/internal/shared/swagger/swagger.go b/server/internal/shared/swagger/swagger.go index 0c9584d..e6d8753 100644 --- a/server/internal/shared/swagger/swagger.go +++ b/server/internal/shared/swagger/swagger.go @@ -419,6 +419,7 @@ func isMultipartRoute(route gin.RouteInfo) bool { path == "/api/ops/site-domain-mappings/import" || path == "/api/ops/object-storage/objects/upload" || path == "/api/ops/desktop-client/releases/upload" || + path == "/api/ops/desktop-client/release-uploads/:upload_id/chunks/:chunk_index" || path == "/api/tenant/knowledge/items/file" || path == "/api/tenant/kol/manage/profile/avatar" || path == "/api/tenant/articles/:id/images"