feat: add chunked desktop client release uploads
Deployment Config CI / Deployment Config (push) Successful in 28s
Frontend CI / Frontend (push) Successful in 2m12s
Backend CI / Backend (push) Successful in 14m44s

This commit is contained in:
2026-06-01 01:31:54 +08:00
parent c5da0cf507
commit 76df672133
14 changed files with 1195 additions and 57 deletions
+3 -3
View File
@@ -8,7 +8,7 @@ real_ip_recursive on;
server { server {
listen 80; listen 80;
server_name _; server_name _;
client_max_body_size 600M; client_max_body_size 1100M;
root /usr/share/nginx/html; root /usr/share/nginx/html;
index index.html; index index.html;
@@ -31,8 +31,8 @@ server {
proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $remote_addr; proxy_set_header X-Forwarded-For $remote_addr;
proxy_set_header X-Forwarded-Proto $scheme; proxy_set_header X-Forwarded-Proto $scheme;
proxy_read_timeout 900s; proxy_read_timeout 1800s;
proxy_send_timeout 900s; proxy_send_timeout 1800s;
} }
location ~* \.(js|css|png|jpg|jpeg|gif|ico|svg|woff2?)$ { location ~* \.(js|css|png|jpg|jpeg|gif|ico|svg|woff2?)$ {
+183 -1
View File
@@ -75,7 +75,27 @@ export interface DesktopClientReleaseDownloadURLResult {
expires_in?: number | null 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 = [ export const platformOptions = [
{ label: 'macOS', value: 'darwin' }, { label: 'macOS', value: 'darwin' },
@@ -127,6 +147,20 @@ export const desktopClientReleaseApi = {
channel: string channel: string
version: string version: string
kind?: 'installer' | 'updater' 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() const form = new FormData()
@@ -142,6 +176,15 @@ export const desktopClientReleaseApi = {
data: DesktopClientReleaseUploadResult data: DesktopClientReleaseUploadResult
}>('/desktop-client/releases/upload', form, { }>('/desktop-client/releases/upload', form, {
timeout: DESKTOP_CLIENT_RELEASE_UPLOAD_TIMEOUT_MS, 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 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<DesktopClientReleaseUploadResult> {
const session = await http.post<DesktopClientReleaseUploadSession>('/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<DesktopClientReleaseUploadResult>(
`/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 { export function platformLabel(value: string): string {
return platformOptions.find((item) => item.value === value)?.label ?? value return platformOptions.find((item) => item.value === value)?.label ?? value
} }
@@ -202,7 +202,7 @@
> >
<a-button :loading="uploading"> <a-button :loading="uploading">
<template #icon><UploadOutlined /></template> <template #icon><UploadOutlined /></template>
{{ form.oss_object_key ? '重新上传' : '选择文件并上传' }} {{ uploadButtonText('installer') }}
</a-button> </a-button>
</a-upload> </a-upload>
<div v-if="form.file_name" class="upload-result"> <div v-if="form.file_name" class="upload-result">
@@ -268,7 +268,7 @@
> >
<a-button :loading="updaterUploading"> <a-button :loading="updaterUploading">
<template #icon><UploadOutlined /></template> <template #icon><UploadOutlined /></template>
{{ form.updater_oss_object_key ? '重新上传' : '选择文件并上传' }} {{ uploadButtonText('updater') }}
</a-button> </a-button>
</a-upload> </a-upload>
<div v-if="form.updater_file_name" class="upload-result"> <div v-if="form.updater_file_name" class="upload-result">
@@ -400,6 +400,8 @@ const total = ref(0)
const statusLoadingId = ref<number | null>(null) const statusLoadingId = ref<number | null>(null)
const uploading = ref(false) const uploading = ref(false)
const updaterUploading = ref(false) const updaterUploading = ref(false)
const uploadProgress = ref<number | null>(null)
const updaterUploadProgress = ref<number | null>(null)
const enabledRows = computed(() => rows.value.filter((row) => row.enabled).length) const enabledRows = computed(() => rows.value.filter((row) => row.enabled).length)
const forceRows = computed(() => rows.value.filter((row) => row.force_update).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') { if (kind === 'installer') {
uploading.value = true uploading.value = true
uploadProgress.value = 0
} else { } else {
updaterUploading.value = true updaterUploading.value = true
updaterUploadProgress.value = 0
} }
try { try {
const result = await desktopClientReleaseApi.uploadOSS(file, { const result = await desktopClientReleaseApi.uploadOSS(file, {
@@ -578,6 +582,14 @@ async function beforeUploadReleasePackage(file: File, kind: 'installer' | 'updat
channel: form.channel.trim() || 'stable', channel: form.channel.trim() || 'stable',
version: form.version.trim(), version: form.version.trim(),
kind, kind,
onProgress(progress) {
if (typeof progress.percent !== 'number') return
if (kind === 'installer') {
uploadProgress.value = progress.percent
} else {
updaterUploadProgress.value = progress.percent
}
},
}) })
if (kind === 'installer') { if (kind === 'installer') {
form.oss_object_key = result.oss_object_key form.oss_object_key = result.oss_object_key
@@ -599,10 +611,25 @@ async function beforeUploadReleasePackage(file: File, kind: 'installer' | 'updat
} finally { } finally {
uploading.value = false uploading.value = false
updaterUploading.value = false updaterUploading.value = false
uploadProgress.value = null
updaterUploadProgress.value = null
} }
return false 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) { function beforeUploadInstallerPackage(file: File) {
return beforeUploadReleasePackage(file, 'installer') return beforeUploadReleasePackage(file, 'installer')
} }
+3 -3
View File
@@ -8,7 +8,7 @@ real_ip_recursive on;
server { server {
listen 80; listen 80;
server_name _; server_name _;
client_max_body_size 600M; client_max_body_size 1100M;
root /usr/share/nginx/html; root /usr/share/nginx/html;
index index.html; index index.html;
@@ -31,8 +31,8 @@ server {
proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $remote_addr; proxy_set_header X-Forwarded-For $remote_addr;
proxy_set_header X-Forwarded-Proto $scheme; proxy_set_header X-Forwarded-Proto $scheme;
proxy_read_timeout 900s; proxy_read_timeout 1800s;
proxy_send_timeout 900s; proxy_send_timeout 1800s;
} }
location ~* \.(js|css|png|jpg|jpeg|gif|ico|svg|woff2?)$ { location ~* \.(js|css|png|jpg|jpeg|gif|ico|svg|woff2?)$ {
+696 -18
View File
@@ -1,19 +1,26 @@
package app package app
import ( import (
"bytes"
"context" "context"
"crypto/sha256" "crypto/sha256"
"encoding/hex" "encoding/hex"
"encoding/json"
"errors" "errors"
"fmt" "fmt"
"io"
"net/http" "net/http"
"net/url" "net/url"
"os"
"path" "path"
"path/filepath" "path/filepath"
"regexp" "regexp"
"sort"
"strings" "strings"
"sync"
"time" "time"
"github.com/google/uuid"
"github.com/jackc/pgx/v5/pgconn" "github.com/jackc/pgx/v5/pgconn"
"gopkg.in/yaml.v3" "gopkg.in/yaml.v3"
@@ -41,10 +48,22 @@ const (
DesktopClientAssetKindUpdater = "updater" 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 { type DesktopClientReleaseService struct {
releases *repository.DesktopClientReleaseRepository releases *repository.DesktopClientReleaseRepository
storage objectstorage.Client storage objectstorage.Client
audits *AuditService audits *AuditService
releaseUploadDir string
releaseUploadLocks map[string]*sync.Mutex
releaseUploadMu sync.Mutex
} }
func NewDesktopClientReleaseService( func NewDesktopClientReleaseService(
@@ -52,7 +71,21 @@ func NewDesktopClientReleaseService(
storage objectstorage.Client, storage objectstorage.Client,
audits *AuditService, audits *AuditService,
) *DesktopClientReleaseService { ) *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 { type DesktopClientReleaseView struct {
@@ -147,13 +180,15 @@ type DesktopClientReleaseInput struct {
} }
type DesktopClientReleaseUploadInput struct { type DesktopClientReleaseUploadInput struct {
Platform string Platform string
Arch string Arch string
Channel string Channel string
Version string Version string
Kind string Kind string
FileName string FileName string
Content []byte Content []byte
ContentReader io.ReadSeeker
ContentSizeBytes int64
} }
type DesktopClientReleaseUploadResult struct { type DesktopClientReleaseUploadResult struct {
@@ -163,6 +198,63 @@ type DesktopClientReleaseUploadResult struct {
SHA256 string `json:"sha256"` 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 { type DesktopClientReleaseCheckInput struct {
Platform string Platform string
Arch string Arch string
@@ -402,6 +494,10 @@ func (s *DesktopClientReleaseService) UploadOSS(ctx context.Context, in DesktopC
if err != nil { if err != nil {
return nil, err return nil, err
} }
reader, fileSizeBytes, err := desktopClientReleaseUploadReader(in)
if err != nil {
return nil, response.ErrBadRequest(40071, "release_file_read_failed", "客户端安装包读取失败")
}
kind := normalizeDesktopClientAssetKind(in.Kind) kind := normalizeDesktopClientAssetKind(in.Kind)
if kind == DesktopClientAssetKindUpdater { if kind == DesktopClientAssetKindUpdater {
@@ -409,10 +505,12 @@ func (s *DesktopClientReleaseService) UploadOSS(ctx context.Context, in DesktopC
return nil, err return nil, err
} }
} }
sum := sha256.Sum256(in.Content) sha256Hex, err := hashDesktopClientReleaseUpload(reader)
sha256Hex := hex.EncodeToString(sum[:]) if err != nil {
return nil, response.ErrBadRequest(40071, "release_file_read_failed", "客户端安装包读取失败")
}
objectKey := buildDesktopClientReleaseObjectKey(sha256Hex, fileName) objectKey := buildDesktopClientReleaseObjectKey(sha256Hex, fileName)
contentType := detectDesktopReleaseContentType(fileName, in.Content) contentType := detectDesktopReleaseContentType(fileName, nil)
exists, err := s.storage.Exists(ctx, objectKey) exists, err := s.storage.Exists(ctx, objectKey)
if err != nil { if err != nil {
appErr := response.ErrInternal(50092, "desktop_client_release_upload_failed", err.Error()) 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 return nil, appErr
} }
if !exists { 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 := response.ErrInternal(50092, "desktop_client_release_upload_failed", err.Error())
appErr.Cause = err appErr.Cause = err
return nil, appErr return nil, appErr
@@ -430,11 +528,239 @@ func (s *DesktopClientReleaseService) UploadOSS(ctx context.Context, in DesktopC
return &DesktopClientReleaseUploadResult{ return &DesktopClientReleaseUploadResult{
OSSObjectKey: objectKey, OSSObjectKey: objectKey,
FileName: fileName, FileName: fileName,
FileSizeBytes: int64(len(in.Content)), FileSizeBytes: fileSizeBytes,
SHA256: sha256Hex, SHA256: sha256Hex,
}, nil }, 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) { func (s *DesktopClientReleaseService) CheckLatest(ctx context.Context, in DesktopClientReleaseCheckInput) (*DesktopClientReleaseCheckResult, error) {
target, currentVersion, err := normalizeDesktopClientReleaseCheckInput(in) target, currentVersion, err := normalizeDesktopClientReleaseCheckInput(in)
if err != nil { if err != nil {
@@ -1005,10 +1331,14 @@ func normalizeDesktopReleaseSHA256(value *string) (*string, error) {
} }
func normalizeDesktopClientReleaseUploadInput(in DesktopClientReleaseUploadInput) (repository.DesktopClientReleaseTarget, 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", "上传文件不能为空") 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") return repository.DesktopClientReleaseTarget{}, "", response.ErrBadRequest(40072, "release_file_too_large", "客户端安装包不能超过 1GB")
} }
@@ -1037,6 +1367,354 @@ func normalizeDesktopClientReleaseUploadInput(in DesktopClientReleaseUploadInput
return target, fileName, nil 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 { func normalizeDesktopClientAssetKind(value string) string {
switch normalizeDesktopReleaseEnum(value) { switch normalizeDesktopReleaseEnum(value) {
case DesktopClientAssetKindUpdater: case DesktopClientAssetKindUpdater:
@@ -1,8 +1,13 @@
package app package app
import ( import (
"bytes"
"context" "context"
"crypto/sha256"
"encoding/hex"
"errors" "errors"
"io"
"os"
"testing" "testing"
"time" "time"
@@ -213,9 +218,60 @@ func TestResolveDownloadURLResultUsesSignedDownloadURL(t *testing.T) {
require.Empty(t, storage.publicURLCalls) 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 { type desktopReleaseDownloadURLStorage struct {
publicURLCalls []string publicURLCalls []string
downloadURLCalls []string downloadURLCalls []string
existsByKey map[string]bool
objects map[string][]byte
} }
func (s *desktopReleaseDownloadURLStorage) Validate() error { func (s *desktopReleaseDownloadURLStorage) Validate() error {
@@ -226,11 +282,30 @@ func (s *desktopReleaseDownloadURLStorage) PutBytes(context.Context, string, []b
return nil 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) { func (s *desktopReleaseDownloadURLStorage) GetBytes(context.Context, string) ([]byte, error) {
return nil, errors.New("not implemented") 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 return true, nil
} }
@@ -2,7 +2,6 @@ package transport
import ( import (
"fmt" "fmt"
"io"
"net/http" "net/http"
"strconv" "strconv"
"strings" "strings"
@@ -40,6 +39,17 @@ type setDesktopClientReleaseEnabledRequest struct {
Enabled bool `json:"enabled"` 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 { func listDesktopClientReleasesHandler(svc *app.DesktopClientReleaseService) gin.HandlerFunc {
return func(c *gin.Context) { return func(c *gin.Context) {
page, _ := strconv.Atoi(c.DefaultQuery("page", "1")) page, _ := strconv.Atoi(c.DefaultQuery("page", "1"))
@@ -151,11 +161,28 @@ func resolveDesktopClientReleaseDownloadURLHandler(svc *app.DesktopClientRelease
func uploadDesktopClientReleaseHandler(svc *app.DesktopClientReleaseService) gin.HandlerFunc { func uploadDesktopClientReleaseHandler(svc *app.DesktopClientReleaseService) gin.HandlerFunc {
return func(c *gin.Context) { 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") fileHeader, err := c.FormFile("file")
if err != nil { 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", "请选择要上传的客户端安装包")) response.Error(c, response.ErrBadRequest(40071, "release_file_required", "请选择要上传的客户端安装包"))
return 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() file, err := fileHeader.Open()
if err != nil { if err != nil {
response.Error(c, response.ErrBadRequest(40071, "release_file_open_failed", "客户端安装包读取失败")) response.Error(c, response.ErrBadRequest(40071, "release_file_open_failed", "客户端安装包读取失败"))
@@ -163,20 +190,15 @@ func uploadDesktopClientReleaseHandler(svc *app.DesktopClientReleaseService) gin
} }
defer file.Close() 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{ result, err := svc.UploadOSS(c.Request.Context(), app.DesktopClientReleaseUploadInput{
Platform: c.PostForm("platform"), Platform: c.PostForm("platform"),
Arch: c.PostForm("arch"), Arch: c.PostForm("arch"),
Channel: c.PostForm("channel"), Channel: c.PostForm("channel"),
Version: c.PostForm("version"), Version: c.PostForm("version"),
Kind: c.DefaultPostForm("kind", app.DesktopClientAssetKindInstaller), Kind: c.DefaultPostForm("kind", app.DesktopClientAssetKindInstaller),
FileName: fileHeader.Filename, FileName: fileHeader.Filename,
Content: content, ContentReader: file,
ContentSizeBytes: fileHeader.Size,
}) })
if err != nil { if err != nil {
response.Error(c, err) 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 { func checkDesktopClientReleaseHandler(svc *app.DesktopClientReleaseService) gin.HandlerFunc {
return func(c *gin.Context) { return func(c *gin.Context) {
result, err := svc.CheckLatest(c.Request.Context(), app.DesktopClientReleaseCheckInput{ result, err := svc.CheckLatest(c.Request.Context(), app.DesktopClientReleaseCheckInput{
+4
View File
@@ -173,6 +173,10 @@ func RegisterRoutes(d Deps) {
authed.GET("/desktop-client/releases", listDesktopClientReleasesHandler(d.Releases)) authed.GET("/desktop-client/releases", listDesktopClientReleasesHandler(d.Releases))
authed.POST("/desktop-client/releases/upload", uploadDesktopClientReleaseHandler(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.POST("/desktop-client/releases", createDesktopClientReleaseHandler(d.Releases))
authed.PATCH("/desktop-client/releases/:id", updateDesktopClientReleaseHandler(d.Releases)) authed.PATCH("/desktop-client/releases/:id", updateDesktopClientReleaseHandler(d.Releases))
authed.GET("/desktop-client/releases/:id/download-url", resolveDesktopClientReleaseDownloadURLHandler(d.Releases)) authed.GET("/desktop-client/releases/:id/download-url", resolveDesktopClientReleaseDownloadURLHandler(d.Releases))
+15 -2
View File
@@ -72,12 +72,25 @@ func (c *aliyunClient) Validate() error {
} }
func (c *aliyunClient) PutBytes(ctx context.Context, objectKey string, content []byte, contentType string) 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 { if err := c.Validate(); err != nil {
return err return err
} }
if strings.TrimSpace(objectKey) == "" { if strings.TrimSpace(objectKey) == "" {
return fmt.Errorf("object key is required") return fmt.Errorf("object key is required")
} }
if reader == nil {
return fmt.Errorf("object reader is required")
}
exists, err := c.client.IsBucketExist(c.bucket) exists, err := c.client.IsBucketExist(c.bucket)
if err != nil { if err != nil {
@@ -113,11 +126,11 @@ func (c *aliyunClient) PutBytes(ctx context.Context, objectKey string, content [
if err != nil { if err != nil {
return err 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", c.logError(ctx, "aliyun put object failed",
zap.String("bucket", c.bucket), zap.String("bucket", c.bucket),
zap.String("object_key", objectKey), 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("object_acl", strings.TrimSpace(c.cfg.ObjectACL)),
zap.String("content_type", contentType), zap.String("content_type", contentType),
zap.String("aliyun_error_code", aliyunErrorCode(err)), zap.String("aliyun_error_code", aliyunErrorCode(err)),
@@ -4,6 +4,7 @@ import (
"context" "context"
"errors" "errors"
"fmt" "fmt"
"io"
"strings" "strings"
"time" "time"
@@ -27,6 +28,10 @@ type Client interface {
DownloadURL(objectKey string) (string, error) 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 { type ManagementClient interface {
Client Client
List(ctx context.Context, in ListInput) (*ListResult, error) 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() 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) { func (c disabledClient) GetBytes(context.Context, string) ([]byte, error) {
return nil, c.Validate() return nil, c.Validate()
} }
+15 -2
View File
@@ -103,6 +103,16 @@ func (c *minioClient) PutBytes(
objectKey string, objectKey string,
content []byte, content []byte,
contentType string, 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 { ) error {
if err := c.Validate(); err != nil { if err := c.Validate(); err != nil {
return err return err
@@ -110,6 +120,9 @@ func (c *minioClient) PutBytes(
if strings.TrimSpace(objectKey) == "" { if strings.TrimSpace(objectKey) == "" {
return fmt.Errorf("object key is required") 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) exists, err := c.client.BucketExists(ctx, c.bucket)
if err != nil { 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, ContentType: contentType,
}) })
if err != nil { if err != nil {
c.logError(ctx, "minio put object failed", c.logError(ctx, "minio put object failed",
zap.String("bucket", c.bucket), zap.String("bucket", c.bucket),
zap.String("object_key", objectKey), zap.String("object_key", objectKey),
zap.Int("content_size", len(content)), zap.Int64("content_size", size),
zap.Error(err), zap.Error(err),
) )
return fmt.Errorf("put minio object: %w", err) return fmt.Errorf("put minio object: %w", err)
@@ -2,6 +2,7 @@ package objectstorage
import ( import (
"context" "context"
"io"
"sync" "sync"
) )
@@ -31,6 +32,19 @@ func (c *ReloadableClient) PutBytes(ctx context.Context, objectKey string, conte
return c.snapshot().PutBytes(ctx, objectKey, content, contentType) 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) { func (c *ReloadableClient) GetBytes(ctx context.Context, objectKey string) ([]byte, error) {
return c.snapshot().GetBytes(ctx, objectKey) return c.snapshot().GetBytes(ctx, objectKey)
} }
+15 -11
View File
@@ -99,17 +99,21 @@ var routeDocs = map[string]routeDoc{
"DELETE /api/ops/object-storage/folders": {"删除对象存储目录", "按前缀递归删除目录下对象。"}, "DELETE /api/ops/object-storage/folders": {"删除对象存储目录", "按前缀递归删除目录下对象。"},
// --- Ops:桌面客户端版本 --- // --- Ops:桌面客户端版本 ---
"GET /api/ops/desktop-client/releases": {"桌面客户端版本列表", "分页查询桌面客户端版本配置。"}, "GET /api/ops/desktop-client/releases": {"桌面客户端版本列表", "分页查询桌面客户端版本配置。"},
"POST /api/ops/desktop-client/releases/upload": {"上传桌面客户端包", "Multipart 上传桌面客户端安装包或更新包。"}, "POST /api/ops/desktop-client/releases/upload": {"上传桌面客户端包(兼容)", "兼容旧版单请求 Multipart 上传;新版前端使用分片上传。"},
"POST /api/ops/desktop-client/releases": {"新建桌面客户端版本", "创建桌面客户端版本配置。"}, "POST /api/ops/desktop-client/release-uploads": {"初始化客户端包分片上传", "创建分片上传会话,返回 upload_id、分片大小和分片数量。"},
"PATCH /api/ops/desktop-client/releases/:id": {"更新桌面客户端版本", "更新桌面客户端版本配置。"}, "POST /api/ops/desktop-client/release-uploads/:upload_id/chunks/:chunk_index": {"上传客户端包分片", "Multipart 上传单个客户端安装包或更新包分片。"},
"GET /api/ops/desktop-client/releases/:id/download-url": {"解析桌面客户端下载地址", "解析指定桌面客户端版本的下载地址。"}, "POST /api/ops/desktop-client/release-uploads/:upload_id/complete": {"完成客户端包分片上传", "校验全部分片,流式写入 OSS,并返回 Object Key、文件大小和 SHA256。"},
"POST /api/ops/desktop-client/releases/:id/enabled": {"切换桌面客户端版本状态", "启用或禁用指定桌面客户端版本。"}, "DELETE /api/ops/desktop-client/release-uploads/:upload_id": {"取消客户端包分片上传", "取消分片上传会话并清理临时分片。"},
"DELETE /api/ops/desktop-client/releases/:id": {"删除桌面客户端版本", "删除指定桌面客户端版本配置。"}, "POST /api/ops/desktop-client/releases": {"新建客户端版本", "创建桌面客户端版本配置。"},
"GET /api/ops/desktop-client/bug-reports": {"客户端 Bug 列表", "分页查询桌面客户端崩溃和用户反馈记录。"}, "PATCH /api/ops/desktop-client/releases/:id": {"更新客户端版本", "更新桌面客户端版本配置。"},
"GET /api/ops/desktop-client/bug-reports/:id": {"客户端 Bug 详情", "查看桌面客户端崩溃元数据、运行快照和 dump 对象。"}, "GET /api/ops/desktop-client/releases/:id/download-url": {"解析桌面客户端下载地址", "解析指定桌面客户端版本的下载地址。"},
"PATCH /api/ops/desktop-client/bug-reports/:id": {"更新客户端 Bug 状态", "调整客户端 Bug 状态、级别和处理备注。"}, "POST /api/ops/desktop-client/releases/:id/enabled": {"切换客户端版本状态", "启用或禁用指定桌面客户端版本。"},
"GET /api/ops/desktop-client/bug-reports/:id/dump-url": {"解析客户端 dump 下载地址", "生成指定客户端 Bug dump 文件的短期下载地址。"}, "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:合规 --- // --- Ops:合规 ---
"GET /api/ops/compliance/dictionaries": {"合规词库列表", "分页查询内容合规词库。"}, "GET /api/ops/compliance/dictionaries": {"合规词库列表", "分页查询内容合规词库。"},
@@ -419,6 +419,7 @@ func isMultipartRoute(route gin.RouteInfo) bool {
path == "/api/ops/site-domain-mappings/import" || path == "/api/ops/site-domain-mappings/import" ||
path == "/api/ops/object-storage/objects/upload" || path == "/api/ops/object-storage/objects/upload" ||
path == "/api/ops/desktop-client/releases/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/knowledge/items/file" ||
path == "/api/tenant/kol/manage/profile/avatar" || path == "/api/tenant/kol/manage/profile/avatar" ||
path == "/api/tenant/articles/:id/images" path == "/api/tenant/articles/:id/images"