root/geo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: report desktop release upload completion progress

Browse Source
This commit is contained in:
liangxu
2026-06-23 00:04:26 +08:00
parent f0de364063
commit d5e592e290
6 changed files with 395 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/internal/ops/app/desktop_client_release.go
+204
View File
@@ -66,6 +66,7 @@ type DesktopClientReleaseService struct {
audits *AuditService
releaseUploadDir string
releaseUploadLocks map[string]*sync.Mutex
completeJobs map[string]*desktopClientReleaseCompleteJob
releaseUploadMu sync.Mutex
}
@@ -80,6 +81,7 @@ func NewDesktopClientReleaseService(
audits: audits,
releaseUploadDir: defaultDesktopClientReleaseUploadDir(),
releaseUploadLocks: make(map[string]*sync.Mutex),
completeJobs: make(map[string]*desktopClientReleaseCompleteJob),
}
}
@@ -201,6 +203,19 @@ type DesktopClientReleaseUploadResult struct {
SHA256 string `json:"sha256"`
}
type DesktopClientReleaseCompleteJobResult struct {
JobID string `json:"job_id"`
Status string `json:"status"`
Stage string `json:"stage"`
Message string `json:"message"`
Percent int `json:"percent"`
Result *DesktopClientReleaseUploadResult `json:"result,omitempty"`
ErrorCode string `json:"error_code,omitempty"`
Error string `json:"error,omitempty"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
}
type DesktopClientReleaseChunkedUploadInitInput struct {
Platform string
Arch string
@@ -270,6 +285,20 @@ type DesktopClientReleaseDirectUploadCompleteInput struct {
SHA256 string
}
type desktopClientReleaseCompleteJob struct {
mu sync.RWMutex
id string
status string
stage string
message string
percent int
result *DesktopClientReleaseUploadResult
errorCode string
err string
createdAt time.Time
updatedAt time.Time
}
type desktopClientReleaseUploadManifest struct {
UploadID string `json:"upload_id"`
Platform string `json:"platform"`
@@ -658,6 +687,14 @@ func (s *DesktopClientReleaseService) InitiateDirectUpload(
func (s *DesktopClientReleaseService) CompleteDirectUpload(
ctx context.Context,
in DesktopClientReleaseDirectUploadCompleteInput,
) (*DesktopClientReleaseUploadResult, error) {
return s.completeDirectUpload(ctx, in, nil)
}
func (s *DesktopClientReleaseService) completeDirectUpload(
ctx context.Context,
in DesktopClientReleaseDirectUploadCompleteInput,
onProgress func(stage string, percent int, message string),
) (*DesktopClientReleaseUploadResult, error) {
if s.storage == nil {
return nil, response.ErrInternal(50091, "object_storage_unavailable", "object storage is unavailable")
@@ -699,6 +736,7 @@ func (s *DesktopClientReleaseService) CompleteDirectUpload(
return nil, response.ErrBadRequest(40066, "invalid_oss_object_key", "OSS Object Key 与上传文件校验信息不匹配")
}
reportDesktopReleaseUploadProgress(onProgress, "checking_object", 20, "正在检查 OSS 对象")
if management, ok := s.storage.(objectstorage.ManagementClient); ok {
info, err := management.Stat(ctx, objectKey)
if err != nil {
@@ -723,6 +761,7 @@ func (s *DesktopClientReleaseService) CompleteDirectUpload(
return nil, response.ErrBadRequest(40078, "direct_upload_not_found", "OSS 直传文件尚未完成")
}
}
reportDesktopReleaseUploadProgress(onProgress, "hashing_object", 55, "正在计算 OSS 文件 SHA256")
actualSHA, actualSize, err := hashDesktopClientReleaseStorageObject(ctx, s.storage, objectKey)
if err != nil {
if errors.Is(err, objectstorage.ErrObjectNotFound) {
@@ -738,6 +777,7 @@ func (s *DesktopClientReleaseService) CompleteDirectUpload(
if actualSHA != sha256Hex {
return nil, response.ErrBadRequest(40070, "invalid_sha256", "OSS 直传文件 SHA256 不匹配，请重新上传")
}
reportDesktopReleaseUploadProgress(onProgress, "finalizing", 95, "正在确认上传结果")
return &DesktopClientReleaseUploadResult{
OSSObjectKey: objectKey,
@@ -775,6 +815,157 @@ func hashDesktopClientReleaseStorageObject(
return hex.EncodeToString(actualSHA[:]), int64(len(content)), nil
}
func (s *DesktopClientReleaseService) StartDirectUploadCompleteJob(
in DesktopClientReleaseDirectUploadCompleteInput,
) *DesktopClientReleaseCompleteJobResult {
job := newDesktopReleaseCompleteJob()
s.storeDesktopReleaseCompleteJob(job)
go func() {
job.setProgress("checking_object", 10, "正在检查 OSS 对象")
result, err := s.completeDirectUpload(context.Background(), in, job.setProgress)
if err != nil {
job.fail(err)
return
}
job.succeed(result)
}()
return job.snapshot()
}
func (s *DesktopClientReleaseService) StartChunkedUploadCompleteJob(uploadID string) (*DesktopClientReleaseCompleteJobResult, error) {
if _, err := normalizeDesktopClientReleaseUploadID(uploadID); err != nil {
return nil, err
}
job := newDesktopReleaseCompleteJob()
s.storeDesktopReleaseCompleteJob(job)
go func() {
job.setProgress("validating_chunks", 10, "正在校验上传分片")
result, err := s.completeChunkedUpload(context.Background(), uploadID, job.setProgress)
if err != nil {
job.fail(err)
return
}
job.succeed(result)
}()
return job.snapshot(), nil
}
func (s *DesktopClientReleaseService) GetCompleteJob(jobID string) (*DesktopClientReleaseCompleteJobResult, error) {
jobID = strings.TrimSpace(jobID)
if jobID == "" {
return nil, response.ErrBadRequest(40081, "invalid_upload_job_id", "上传任务 ID 无效")
}
s.releaseUploadMu.Lock()
defer s.releaseUploadMu.Unlock()
job, ok := s.completeJobs[jobID]
if !ok {
return nil, response.ErrNotFound(40481, "upload_job_not_found", "上传任务不存在或已过期")
}
return job.snapshot(), nil
}
func (s *DesktopClientReleaseService) storeDesktopReleaseCompleteJob(job *desktopClientReleaseCompleteJob) {
s.releaseUploadMu.Lock()
defer s.releaseUploadMu.Unlock()
if s.completeJobs == nil {
s.completeJobs = map[string]*desktopClientReleaseCompleteJob{}
}
cutoff := time.Now().UTC().Add(-desktopClientReleaseUploadSessionTTL)
for id, existing := range s.completeJobs {
if existing.snapshot().UpdatedAt.Before(cutoff) {
delete(s.completeJobs, id)
}
}
s.completeJobs[job.id] = job
}
func newDesktopReleaseCompleteJob() *desktopClientReleaseCompleteJob {
now := time.Now().UTC()
return &desktopClientReleaseCompleteJob{
id: uuid.NewString(),
status: "running",
stage: "queued",
message: "上传确认任务已创建",
percent: 1,
createdAt: now,
updatedAt: now,
}
}
func (j *desktopClientReleaseCompleteJob) setProgress(stage string, percent int, message string) {
if j == nil {
return
}
j.mu.Lock()
defer j.mu.Unlock()
if j.status == "succeeded" || j.status == "failed" {
return
}
j.status = "running"
j.stage = strings.TrimSpace(stage)
j.message = strings.TrimSpace(message)
if percent < 1 {
percent = 1
}
if percent > 99 {
percent = 99
}
j.percent = percent
j.updatedAt = time.Now().UTC()
}
func (j *desktopClientReleaseCompleteJob) succeed(result *DesktopClientReleaseUploadResult) {
j.mu.Lock()
defer j.mu.Unlock()
j.status = "succeeded"
j.stage = "complete"
j.message = "上传完成"
j.percent = 100
j.result = result
j.updatedAt = time.Now().UTC()
}
func (j *desktopClientReleaseCompleteJob) fail(err error) {
appErr := response.Normalize(err)
j.mu.Lock()
defer j.mu.Unlock()
j.status = "failed"
j.stage = "failed"
j.message = appErr.Detail
if j.message == "" {
j.message = appErr.Message
}
j.errorCode = appErr.Message
j.err = j.message
if j.percent < 1 {
j.percent = 1
}
j.updatedAt = time.Now().UTC()
}
func (j *desktopClientReleaseCompleteJob) snapshot() *DesktopClientReleaseCompleteJobResult {
j.mu.RLock()
defer j.mu.RUnlock()
return &DesktopClientReleaseCompleteJobResult{
JobID: j.id,
Status: j.status,
Stage: j.stage,
Message: j.message,
Percent: j.percent,
Result: j.result,
ErrorCode: j.errorCode,
Error: j.err,
CreatedAt: j.createdAt,
UpdatedAt: j.updatedAt,
}
}
func reportDesktopReleaseUploadProgress(onProgress func(stage string, percent int, message string), stage string, percent int, message string) {
if onProgress != nil {
onProgress(stage, percent, message)
}
}
func MaxDesktopClientReleaseMultipartBytes() int64 {
return MaxDesktopClientReleaseUploadBytes + desktopClientReleaseMultipartOverheadBytes
}
@@ -934,6 +1125,14 @@ func (s *DesktopClientReleaseService) UploadChunk(
func (s *DesktopClientReleaseService) CompleteChunkedUpload(
ctx context.Context,
uploadIDValue string,
) (*DesktopClientReleaseUploadResult, error) {
return s.completeChunkedUpload(ctx, uploadIDValue, nil)
}
func (s *DesktopClientReleaseService) completeChunkedUpload(
ctx context.Context,
uploadIDValue string,
onProgress func(stage string, percent int, message string),
) (*DesktopClientReleaseUploadResult, error) {
uploadID, err := normalizeDesktopClientReleaseUploadID(uploadIDValue)
if err != nil {
@@ -950,9 +1149,11 @@ func (s *DesktopClientReleaseService) CompleteChunkedUpload(
if time.Now().UTC().After(manifest.ExpiresAt) {
return nil, response.ErrBadRequest(40075, "release_upload_expired", "上传会话已过期，请重新选择文件上传")
}
reportDesktopReleaseUploadProgress(onProgress, "validating_chunks", 20, "正在校验上传分片")
if err := manifest.validateComplete(); err != nil {
return nil, err
}
reportDesktopReleaseUploadProgress(onProgress, "hashing_chunks", 45, "正在计算完整文件 SHA256")
sha256Hex, err := s.hashDesktopReleaseUploadChunks(manifest)
if err != nil {
return nil, err
@@ -960,6 +1161,7 @@ func (s *DesktopClientReleaseService) CompleteChunkedUpload(
objectKey := buildDesktopClientReleaseObjectKey(sha256Hex, manifest.FileName)
contentType := detectDesktopReleaseContentType(manifest.FileName, nil)
reportDesktopReleaseUploadProgress(onProgress, "checking_object", 65, "正在检查 OSS 对象")
exists, err := s.storage.Exists(ctx, objectKey)
if err != nil {
appErr := response.ErrInternal(50092, "desktop_client_release_upload_failed", err.Error())
@@ -967,6 +1169,7 @@ func (s *DesktopClientReleaseService) CompleteChunkedUpload(
return nil, appErr
}
if !exists {
reportDesktopReleaseUploadProgress(onProgress, "uploading_object", 75, "正在合并并上传到 OSS")
reader, err := s.openDesktopReleaseUploadChunks(manifest)
if err != nil {
return nil, err
@@ -978,6 +1181,7 @@ func (s *DesktopClientReleaseService) CompleteChunkedUpload(
return nil, appErr
}
}
reportDesktopReleaseUploadProgress(onProgress, "finalizing", 95, "正在确认上传结果")
result := &DesktopClientReleaseUploadResult{
OSSObjectKey: objectKey,
server/internal/ops/app/desktop_client_release_test.go
+35
View File
@@ -424,6 +424,41 @@ func TestDesktopClientReleaseDirectUploadRejectsSHA256Mismatch(t *testing.T) {
require.Contains(t, err.Error(), "invalid_sha256")
}
func TestDesktopClientReleaseDirectUploadCompleteJobReportsStatus(t *testing.T) {
t.Parallel()
content := []byte("release-package")
sum := sha256.Sum256(content)
sha256Hex := hex.EncodeToString(sum[:])
objectKey := buildDesktopClientReleaseObjectKey(sha256Hex, "client.zip")
storage := &desktopReleaseDownloadURLStorage{
existsByKey: map[string]bool{objectKey: true},
objects: map[string][]byte{objectKey: content},
objectSizes: map[string]int64{objectKey: int64(len(content))},
}
svc := NewDesktopClientReleaseService(nil, storage, nil)
job := svc.StartDirectUploadCompleteJob(DesktopClientReleaseDirectUploadCompleteInput{
Platform: "darwin",
Arch: "arm64",
Channel: "stable",
Version: "1.2.3",
Kind: DesktopClientAssetKindUpdater,
OSSObjectKey: objectKey,
FileName: "client.zip",
FileSizeBytes: int64(len(content)),
SHA256: sha256Hex,
})
require.NotEmpty(t, job.JobID)
require.Equal(t, "running", job.Status)
require.Eventually(t, func() bool {
current, err := svc.GetCompleteJob(job.JobID)
require.NoError(t, err)
return current.Status == "succeeded" && current.Result != nil && current.Result.SHA256 == sha256Hex
}, time.Second, 10*time.Millisecond)
}
type desktopReleaseDownloadURLStorage struct {
publicURLCalls []string
downloadURLCalls []string
server/internal/ops/transport/desktop_client_release_handler.go
+44
View File
@@ -284,6 +284,39 @@ func completeDesktopClientReleaseDirectUploadHandler(svc *app.DesktopClientRelea
}
}
func startDesktopClientReleaseDirectUploadCompleteJobHandler(svc *app.DesktopClientReleaseService) gin.HandlerFunc {
return func(c *gin.Context) {
var body desktopClientReleaseDirectUploadCompleteRequest
if err := c.ShouldBindJSON(&body); err != nil {
response.Error(c, response.ErrBadRequest(40000, "invalid_payload", err.Error()))
return
}
result := svc.StartDirectUploadCompleteJob(app.DesktopClientReleaseDirectUploadCompleteInput{
Platform: body.Platform,
Arch: body.Arch,
Channel: body.Channel,
Version: body.Version,
Kind: body.Kind,
OSSObjectKey: body.OSSObjectKey,
FileName: body.FileName,
FileSizeBytes: body.FileSizeBytes,
SHA256: body.SHA256,
})
response.Success(c, result)
}
}
func getDesktopClientReleaseCompleteJobHandler(svc *app.DesktopClientReleaseService) gin.HandlerFunc {
return func(c *gin.Context) {
result, err := svc.GetCompleteJob(c.Param("job_id"))
if err != nil {
response.Error(c, err)
return
}
response.Success(c, result)
}
}
func initiateDesktopClientReleaseUploadHandler(svc *app.DesktopClientReleaseService) gin.HandlerFunc {
return func(c *gin.Context) {
var body desktopClientReleaseUploadInitRequest
@@ -370,6 +403,17 @@ func completeDesktopClientReleaseUploadHandler(svc *app.DesktopClientReleaseServ
}
}
func startDesktopClientReleaseUploadCompleteJobHandler(svc *app.DesktopClientReleaseService) gin.HandlerFunc {
return func(c *gin.Context) {
result, err := svc.StartChunkedUploadCompleteJob(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 {
server/internal/ops/transport/router.go
+3
View File
@@ -175,9 +175,12 @@ func RegisterRoutes(d Deps) {
authed.POST("/desktop-client/releases/upload", uploadDesktopClientReleaseHandler(d.Releases))
authed.POST("/desktop-client/releases/direct-upload", initiateDesktopClientReleaseDirectUploadHandler(d.Releases))
authed.POST("/desktop-client/releases/direct-upload/complete", completeDesktopClientReleaseDirectUploadHandler(d.Releases))
authed.POST("/desktop-client/releases/direct-upload/complete-jobs", startDesktopClientReleaseDirectUploadCompleteJobHandler(d.Releases))
authed.GET("/desktop-client/release-upload-complete-jobs/:job_id", getDesktopClientReleaseCompleteJobHandler(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.POST("/desktop-client/release-uploads/:upload_id/complete-jobs", startDesktopClientReleaseUploadCompleteJobHandler(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))