diff --git a/.github/release-notes/v1.2.35.md b/.github/release-notes/v1.2.35.md new file mode 100644 index 0000000..e3bce05 --- /dev/null +++ b/.github/release-notes/v1.2.35.md @@ -0,0 +1,7 @@ +- 修复完整备份上传 Google 云端硬盘失败：由小文件上传改为官方可恢复分块上传，支持大型邮箱备份。 +- 同一份本地加密备份只显示一次文件名，下方分别显示 Telegram 与 Google 云端硬盘的上传百分比、已上传大小和结果。 +- 手动发送改为后台任务，刷新或离开页面后上传仍会继续，返回备份页可继续查看进度。 +- Google 授权失效、空间不足、请求限流、Drive API 未启用及网络超时会显示对应中文处理建议。 +- 定时备份的云端推送失败也会直接显示具体原因，不再只提示查看服务器日志。 + +**完整更新日志**：[v1.2.34...v1.2.35](https://github.com/zxyszx/NewSzxcn-Email/compare/v1.2.34...v1.2.35) diff --git a/VERSION b/VERSION index e8c4ceb..0ca6e94 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -1.2.34 +1.2.35 diff --git a/apps/api/internal/app/app.go b/apps/api/internal/app/app.go index 084fc83..0f37634 100644 --- a/apps/api/internal/app/app.go +++ b/apps/api/internal/app/app.go @@ -40,6 +40,7 @@ type App struct { telegramDeliveryMu sync.Mutex backupMu sync.Mutex backupJob *backupJob + backupTransfers map[string]*backupTransfer } const ( @@ -83,7 +84,7 @@ func New(cfg Config, logger *slog.Logger) (*App, error) { } db.SetMaxOpenConns(1) - a := &App{cfg: cfg, db: db, log: logger, now: time.Now, policy: NewHTMLPolicy(), maildirHealth: newMaildirSyncHealthTracker(), telegramURL: "https://api.telegram.org", telegramPairs: map[string]telegramPairing{}} + a := &App{cfg: cfg, db: db, log: logger, now: time.Now, policy: NewHTMLPolicy(), maildirHealth: newMaildirSyncHealthTracker(), telegramURL: "https://api.telegram.org", telegramPairs: map[string]telegramPairing{}, backupTransfers: map[string]*backupTransfer{}} a.externalIMAP = a if err := a.configureSQLite(context.Background()); err != nil { db.Close() diff --git a/apps/api/internal/app/backup_handlers.go b/apps/api/internal/app/backup_handlers.go index c7f73e1..7d8e1d9 100644 --- a/apps/api/internal/app/backup_handlers.go +++ b/apps/api/internal/app/backup_handlers.go @@ -17,7 +17,6 @@ import ( "mime/multipart" "net" "net/http" - "net/textproto" "net/url" "os" "os/exec" @@ -31,6 +30,7 @@ import ( ) const backupTelegramLimit = 49 << 20 +const googleDriveUploadChunkSize = 8 << 20 type backupJob struct { Status string `json:"status"` @@ -45,6 +45,17 @@ type backupItem struct { SHA256 string `json:"sha256,omitempty"` } +type backupTransfer struct { + Provider string `json:"provider"` + Name string `json:"name"` + Status string `json:"status"` + Uploaded int64 `json:"uploaded"` + Total int64 `json:"total"` + StartedAt time.Time `json:"startedAt"` + FinishedAt time.Time `json:"finishedAt,omitempty"` + Error string `json:"error,omitempty"` +} + type backupListResponse struct { Enabled bool `json:"enabled"` TelegramSet bool `json:"telegramSet"` @@ -53,6 +64,7 @@ type backupListResponse struct { Items []backupItem `json:"items"` Schedule backupSchedule `json:"schedule"` GoogleDrive googleDriveStatus `json:"googleDrive"` + Transfers []backupTransfer `json:"transfers"` } type createBackupRequest struct { @@ -177,7 +189,13 @@ func (a *App) handleListBackups(w http.ResponseWriter, r *http.Request) { copy := *job job = © } + transfers := make([]backupTransfer, 0, len(a.backupTransfers)) + for _, transfer := range a.backupTransfers { + copy := *transfer + transfers = append(transfers, copy) + } a.backupMu.Unlock() + sort.Slice(transfers, func(i, j int) bool { return transfers[i].StartedAt.After(transfers[j].StartedAt) }) schedule, _ := a.loadBackupSchedule(r.Context()) schedule.ServerIP = detectPublicServerIP(r.Context(), a.config().PublicHostname) telegramToken, telegramDestination, _ := a.backupTelegramCredentials(r.Context(), schedule) @@ -185,7 +203,7 @@ func (a *App) handleListBackups(w http.ResponseWriter, r *http.Request) { Enabled: a.backupAssetsAvailable(), TelegramSet: strings.TrimSpace(telegramToken) != "" && validTelegramPrivateChatID(telegramDestination), TelegramLimit: backupTelegramLimit, Job: job, Items: items, Schedule: schedule, - GoogleDrive: a.loadGoogleDriveStatus(r.Context()), + GoogleDrive: a.loadGoogleDriveStatus(r.Context()), Transfers: transfers, }) } @@ -250,16 +268,32 @@ func (a *App) handleCreateBackup(w http.ResponseWriter, r *http.Request) { defer cancel() path, err := a.createDisasterBackup(ctx, password) password = "" + var deliveryMessages []string if err == nil { var deliveryErrors []error + if uploadGoogleDrive { + a.queueBackupTransfer("googleDrive", path) + } + if sendTelegram { + a.queueBackupTransfer("telegram", path) + } if uploadGoogleDrive { if driveErr := a.uploadBackupToGoogleDrive(ctx, path); driveErr != nil { + message := googleDriveUploadMessage(driveErr) + a.finishBackupTransfer("googleDrive", path, message) + deliveryMessages = append(deliveryMessages, message) deliveryErrors = append(deliveryErrors, fmt.Errorf("google drive: %w", driveErr)) + } else { + a.finishBackupTransfer("googleDrive", path, "") } } if sendTelegram { if telegramErr := a.sendBackupToTelegram(ctx, path); telegramErr != nil { + a.finishBackupTransfer("telegram", path, telegramErr.Error()) + deliveryMessages = append(deliveryMessages, "Telegram 发送失败："+telegramErr.Error()) deliveryErrors = append(deliveryErrors, fmt.Errorf("telegram: %w", telegramErr)) + } else { + a.finishBackupTransfer("telegram", path, "") } } err = errors.Join(deliveryErrors...) @@ -267,7 +301,11 @@ func (a *App) handleCreateBackup(w http.ResponseWriter, r *http.Request) { a.backupMu.Lock() if err != nil { a.backupJob.Status = "failed" - a.backupJob.Error = "本地备份或所选推送未全部完成，请查看服务日志" + if len(deliveryMessages) > 0 { + a.backupJob.Error = strings.Join(deliveryMessages, "；") + } else { + a.backupJob.Error = "本地备份创建失败，请检查服务器存储空间" + } a.log.Error("create disaster backup", "error", err) } else { a.backupJob.Status = "success" @@ -810,6 +848,10 @@ func (a *App) handleDeleteBackup(w http.ResponseWriter, r *http.Request) { return } _ = os.Remove(path + ".sha256") + a.backupMu.Lock() + delete(a.backupTransfers, backupTransferKey("telegram", path)) + delete(a.backupTransfers, backupTransferKey("googleDrive", path)) + a.backupMu.Unlock() respondJSON(w, 200, map[string]any{"ok": true}) } @@ -822,12 +864,21 @@ func (a *App) handleSendBackupTelegram(w http.ResponseWriter, r *http.Request) { respondError(w, 404, "备份不存在") return } - if err := a.sendBackupToTelegram(r.Context(), path); err != nil { - a.log.Error("send backup telegram", "error", err) - respondError(w, 502, err.Error()) + if !a.startBackupTransfer("telegram", path) { + respondError(w, http.StatusConflict, "该备份正在发送到 Telegram") return } - respondJSON(w, 200, map[string]any{"ok": true}) + go func() { + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Hour) + defer cancel() + if err := a.sendBackupToTelegram(ctx, path); err != nil { + a.finishBackupTransfer("telegram", path, err.Error()) + a.log.Error("send backup telegram", "error", err) + return + } + a.finishBackupTransfer("telegram", path, "") + }() + respondJSON(w, http.StatusAccepted, map[string]any{"ok": true}) } func (a *App) handleSendBackupGoogleDrive(w http.ResponseWriter, r *http.Request) { @@ -839,12 +890,89 @@ func (a *App) handleSendBackupGoogleDrive(w http.ResponseWriter, r *http.Request respondError(w, 404, "备份不存在") return } - if err := a.uploadBackupToGoogleDrive(r.Context(), path); err != nil { - a.log.Error("upload backup to google drive", "error", err) - respondError(w, 502, "上传 Google 云端硬盘失败") + if !a.startBackupTransfer("googleDrive", path) { + respondError(w, http.StatusConflict, "该备份正在上传到 Google 云端硬盘") return } - respondJSON(w, 200, map[string]bool{"ok": true}) + go func() { + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Hour) + defer cancel() + if err := a.uploadBackupToGoogleDrive(ctx, path); err != nil { + message := googleDriveUploadMessage(err) + a.finishBackupTransfer("googleDrive", path, message) + a.log.Error("upload backup to google drive", "error", err) + return + } + a.finishBackupTransfer("googleDrive", path, "") + }() + respondJSON(w, http.StatusAccepted, map[string]bool{"ok": true}) +} + +func backupTransferKey(provider, path string) string { + return provider + ":" + filepath.Base(path) +} + +func (a *App) startBackupTransfer(provider, path string) bool { + info, err := os.Stat(path) + if err != nil { + return false + } + key := backupTransferKey(provider, path) + a.backupMu.Lock() + defer a.backupMu.Unlock() + if transfer := a.backupTransfers[key]; transfer != nil && transfer.Status == "running" { + return false + } + a.backupTransfers[key] = &backupTransfer{Provider: provider, Name: filepath.Base(path), Status: "running", Total: info.Size(), StartedAt: a.now().UTC()} + return true +} + +func (a *App) ensureBackupTransfer(provider, path string) { + info, err := os.Stat(path) + if err != nil { + return + } + key := backupTransferKey(provider, path) + a.backupMu.Lock() + defer a.backupMu.Unlock() + if transfer := a.backupTransfers[key]; transfer == nil { + a.backupTransfers[key] = &backupTransfer{Provider: provider, Name: filepath.Base(path), Status: "running", Total: info.Size(), StartedAt: a.now().UTC()} + } else if transfer.Status == "queued" { + transfer.Status = "running" + } +} + +func (a *App) queueBackupTransfer(provider, path string) { + info, err := os.Stat(path) + if err != nil { + return + } + a.backupMu.Lock() + defer a.backupMu.Unlock() + a.backupTransfers[backupTransferKey(provider, path)] = &backupTransfer{Provider: provider, Name: filepath.Base(path), Status: "queued", Total: info.Size(), StartedAt: a.now().UTC()} +} + +func (a *App) updateBackupTransfer(provider, path string, uploaded int64) { + a.backupMu.Lock() + defer a.backupMu.Unlock() + if transfer := a.backupTransfers[backupTransferKey(provider, path)]; transfer != nil { + transfer.Uploaded = uploaded + } +} + +func (a *App) finishBackupTransfer(provider, path, message string) { + a.backupMu.Lock() + defer a.backupMu.Unlock() + if transfer := a.backupTransfers[backupTransferKey(provider, path)]; transfer != nil { + transfer.FinishedAt = a.now().UTC() + if message == "" { + transfer.Status = "success" + transfer.Uploaded = transfer.Total + } else { + transfer.Status = "failed" + transfer.Error = message + } + } } func (a *App) googleDriveClient(ctx context.Context) (*http.Client, error) { @@ -898,7 +1026,7 @@ func (a *App) googleDriveFolderID(ctx context.Context, client *http.Client, name defer resp.Body.Close() if resp.StatusCode < 200 || resp.StatusCode >= 300 { raw, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) - return "", fmt.Errorf("drive folder lookup %s: %s", resp.Status, raw) + return "", &googleDriveAPIError{Operation: "folder lookup", StatusCode: resp.StatusCode, Body: string(raw)} } var list struct { Files []struct { @@ -921,7 +1049,7 @@ func (a *App) googleDriveFolderID(ctx context.Context, client *http.Client, name defer resp.Body.Close() if resp.StatusCode < 200 || resp.StatusCode >= 300 { raw, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) - return "", fmt.Errorf("drive folder create %s: %s", resp.Status, raw) + return "", &googleDriveAPIError{Operation: "folder create", StatusCode: resp.StatusCode, Body: string(raw)} } var created struct { ID string `json:"id"` @@ -936,6 +1064,7 @@ func (a *App) googleDriveFolderID(ctx context.Context, client *http.Client, name } func (a *App) uploadBackupToGoogleDrive(ctx context.Context, path string) error { + a.ensureBackupTransfer("googleDrive", path) client, err := a.googleDriveClient(ctx) if err != nil { return err @@ -945,7 +1074,7 @@ func (a *App) uploadBackupToGoogleDrive(ctx context.Context, path string) error if err != nil { return err } - req, err := newGoogleDriveUploadRequest(ctx, path, folderID) + req, size, err := newGoogleDriveResumableRequest(ctx, path, folderID) if err != nil { return err } @@ -956,53 +1085,107 @@ func (a *App) uploadBackupToGoogleDrive(ctx context.Context, path string) error defer resp.Body.Close() if resp.StatusCode < 200 || resp.StatusCode >= 300 { raw, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) - return fmt.Errorf("drive upload %s: %s", resp.Status, raw) + return &googleDriveAPIError{Operation: "start upload", StatusCode: resp.StatusCode, Body: string(raw)} } - return nil + location := strings.TrimSpace(resp.Header.Get("Location")) + if location == "" { + return errors.New("Google 云端硬盘未返回可恢复上传地址") + } + return a.uploadGoogleDriveChunks(ctx, client, location, path, size) } -func newGoogleDriveUploadRequest(ctx context.Context, path, folderID string) (*http.Request, error) { +type googleDriveAPIError struct { + Operation string + StatusCode int + Body string +} + +func (e *googleDriveAPIError) Error() string { + return fmt.Sprintf("google drive %s returned %d: %s", e.Operation, e.StatusCode, e.Body) +} + +func newGoogleDriveResumableRequest(ctx context.Context, path, folderID string) (*http.Request, int64, error) { + info, err := os.Stat(path) + if err != nil { + return nil, 0, err + } + metadata, _ := json.Marshal(map[string]any{"name": filepath.Base(path), "parents": []string{folderID}}) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, "https://www.googleapis.com/upload/drive/v3/files?uploadType=resumable&fields=id,name", strings.NewReader(string(metadata))) + if err != nil { + return nil, 0, err + } + req.Header.Set("Content-Type", "application/json; charset=UTF-8") + req.Header.Set("X-Upload-Content-Type", "application/octet-stream") + req.Header.Set("X-Upload-Content-Length", fmt.Sprint(info.Size())) + return req, info.Size(), nil +} + +func (a *App) uploadGoogleDriveChunks(ctx context.Context, client *http.Client, location, path string, size int64) error { file, err := os.Open(path) if err != nil { - return nil, err + return err } - reader, writerSide := io.Pipe() - mw := multipart.NewWriter(writerSide) - go func() { - var writeErr error - defer func() { _ = file.Close(); _ = mw.Close(); _ = writerSide.CloseWithError(writeErr) }() - head := textproto.MIMEHeader{} - head.Set("Content-Type", "application/json; charset=UTF-8") - part, err := mw.CreatePart(head) + defer file.Close() + for offset := int64(0); offset < size; { + length := int64(googleDriveUploadChunkSize) + if remaining := size - offset; remaining < length { + length = remaining + } + end := offset + length - 1 + req, err := http.NewRequestWithContext(ctx, http.MethodPut, location, io.NewSectionReader(file, offset, length)) if err != nil { - writeErr = err - return + return err } - metadata, _ := json.Marshal(map[string]any{"name": filepath.Base(path), "parents": []string{folderID}}) - if _, err = part.Write(metadata); err != nil { - writeErr = err - return - } - head = textproto.MIMEHeader{} - head.Set("Content-Type", "application/octet-stream") - part, err = mw.CreatePart(head) + req.ContentLength = length + req.Header.Set("Content-Type", "application/octet-stream") + req.Header.Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", offset, end, size)) + resp, err := client.Do(req) if err != nil { - writeErr = err - return + return err } - _, writeErr = io.Copy(part, file) - }() - req, err := http.NewRequestWithContext(ctx, http.MethodPost, "https://www.googleapis.com/upload/drive/v3/files?uploadType=multipart&fields=id,name", reader) - if err != nil { - _ = file.Close() - _ = writerSide.CloseWithError(err) - return nil, err + raw, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) + _ = resp.Body.Close() + if resp.StatusCode == http.StatusPermanentRedirect { + offset += length + a.updateBackupTransfer("googleDrive", path, offset) + continue + } + if resp.StatusCode >= 200 && resp.StatusCode < 300 && end+1 == size { + a.updateBackupTransfer("googleDrive", path, size) + return nil + } + return &googleDriveAPIError{Operation: "upload chunk", StatusCode: resp.StatusCode, Body: string(raw)} } - req.Header.Set("Content-Type", "multipart/related; boundary="+mw.Boundary()) - return req, nil + return errors.New("Google 云端硬盘不能上传空备份文件") +} + +func googleDriveUploadMessage(err error) string { + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + return "Google 云端硬盘上传超时，请检查服务器网络后重试" + } + var apiErr *googleDriveAPIError + if errors.As(err, &apiErr) { + body := strings.ToLower(apiErr.Body) + switch { + case apiErr.StatusCode == http.StatusUnauthorized || strings.Contains(body, "invalid_grant"): + return "Google 授权已失效，请打开配置，断开后重新连接" + case strings.Contains(body, "storagequota") || strings.Contains(body, "storage quota"): + return "Google 云端硬盘空间不足，请清理空间后重试" + case apiErr.StatusCode == http.StatusTooManyRequests || strings.Contains(body, "ratelimit"): + return "Google 云端硬盘请求过于频繁，请稍后重试" + case apiErr.StatusCode == http.StatusForbidden: + return "Google 云端硬盘无上传权限，请确认 Drive API 已启用并重新连接" + } + } + message := strings.ToLower(err.Error()) + if strings.Contains(message, "oauth2") || strings.Contains(message, "token") { + return "Google 授权已失效，请打开配置，断开后重新连接" + } + return "上传 Google 云端硬盘失败，请检查服务器网络或重新连接 Google 账号" } func (a *App) sendBackupToTelegram(ctx context.Context, path string) error { + a.ensureBackupTransfer("telegram", path) schedule, _ := a.loadBackupSchedule(ctx) token, chatID, err := a.backupTelegramCredentials(ctx, schedule) if err != nil { @@ -1137,7 +1320,7 @@ func (a *App) sendTelegramDocument(ctx context.Context, token, chatID, path stri return } defer file.Close() - _, writeErr = io.Copy(part, file) + _, writeErr = io.Copy(part, &backupProgressReader{reader: file, onProgress: func(uploaded int64) { a.updateBackupTransfer("telegram", path, uploaded) }}) }() endpoint := strings.TrimRight(a.telegramURL, "/") + "/bot" + token + "/sendDocument" req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, reader) @@ -1157,6 +1340,21 @@ func (a *App) sendTelegramDocument(ctx context.Context, token, chatID, path stri return nil } +type backupProgressReader struct { + reader io.Reader + uploaded int64 + onProgress func(int64) +} + +func (r *backupProgressReader) Read(p []byte) (int, error) { + n, err := r.reader.Read(p) + if n > 0 { + r.uploaded += int64(n) + r.onProgress(r.uploaded) + } + return n, err +} + func (a *App) listBackups() ([]backupItem, error) { dir := a.config().BackupDir if dir == "" { @@ -1302,18 +1500,34 @@ func (a *App) runScheduledBackup(ctx context.Context) { path, runErr := a.createDisasterBackup(ctx, password) password = "" localBackupSucceeded := runErr == nil + var deliveryMessages []string if localBackupSucceeded { now := a.now().UTC().Format(time.RFC3339Nano) _, _ = a.db.ExecContext(ctx, `INSERT INTO system_settings(key,value,updated_at) VALUES('backupScheduleLastRun',?,?) ON CONFLICT(key) DO UPDATE SET value=excluded.value,updated_at=excluded.updated_at`, now, now) var deliveryErrors []error + if schedule.GoogleDriveEnabled { + a.queueBackupTransfer("googleDrive", path) + } + if schedule.TelegramEnabled { + a.queueBackupTransfer("telegram", path) + } if schedule.GoogleDriveEnabled { if driveErr := a.uploadBackupToGoogleDrive(ctx, path); driveErr != nil { + message := googleDriveUploadMessage(driveErr) + a.finishBackupTransfer("googleDrive", path, message) + deliveryMessages = append(deliveryMessages, message) deliveryErrors = append(deliveryErrors, fmt.Errorf("google drive: %w", driveErr)) + } else { + a.finishBackupTransfer("googleDrive", path, "") } } if schedule.TelegramEnabled { if telegramErr := a.sendBackupToTelegram(ctx, path); telegramErr != nil { + a.finishBackupTransfer("telegram", path, telegramErr.Error()) + deliveryMessages = append(deliveryMessages, "Telegram 发送失败："+telegramErr.Error()) deliveryErrors = append(deliveryErrors, fmt.Errorf("telegram: %w", telegramErr)) + } else { + a.finishBackupTransfer("telegram", path, "") } } runErr = errors.Join(deliveryErrors...) @@ -1322,7 +1536,11 @@ func (a *App) runScheduledBackup(ctx context.Context) { publicError := "" if runErr != nil { status = "failed" - publicError = "定时备份或云端推送失败，请查看服务日志" + if len(deliveryMessages) > 0 { + publicError = strings.Join(deliveryMessages, "；") + } else { + publicError = "定时备份创建失败，请检查服务器存储空间" + } a.log.Error("scheduled backup", "error", runErr) } a.backupMu.Lock() diff --git a/apps/api/internal/app/backup_handlers_test.go b/apps/api/internal/app/backup_handlers_test.go index 756c0ca..f6447e3 100644 --- a/apps/api/internal/app/backup_handlers_test.go +++ b/apps/api/internal/app/backup_handlers_test.go @@ -3,9 +3,8 @@ package app import ( "context" "encoding/json" + "fmt" "io" - "mime" - "mime/multipart" "net" "net/http" "net/http/httptest" @@ -58,42 +57,109 @@ func TestDiscoverTelegramGroupsReturnsUniqueCandidates(t *testing.T) { } } -func TestGoogleDriveUploadRequestUsesMultipartRelated(t *testing.T) { +func TestGoogleDriveResumableRequest(t *testing.T) { dir := t.TempDir() path := filepath.Join(dir, "newszxcn-backup-test.tar.zst.enc") if err := os.WriteFile(path, []byte("encrypted backup"), 0o600); err != nil { t.Fatal(err) } - req, err := newGoogleDriveUploadRequest(context.Background(), path, "folder-123") + req, size, err := newGoogleDriveResumableRequest(context.Background(), path, "folder-123") if err != nil { t.Fatal(err) } - mediaType, params, err := mime.ParseMediaType(req.Header.Get("Content-Type")) - if err != nil || mediaType != "multipart/related" || params["boundary"] == "" { - t.Fatalf("content type = %q, %v", req.Header.Get("Content-Type"), err) + if size != int64(len("encrypted backup")) { + t.Fatalf("upload size = %d", size) } - reader := multipart.NewReader(req.Body, params["boundary"]) - metadataPart, err := reader.NextPart() - if err != nil { - t.Fatal(err) + if req.URL.Query().Get("uploadType") != "resumable" || req.Header.Get("X-Upload-Content-Length") != fmt.Sprint(size) { + t.Fatalf("resumable request = %s headers=%v", req.URL, req.Header) } var metadata struct { Name string `json:"name"` Parents []string `json:"parents"` } - if err := json.NewDecoder(metadataPart).Decode(&metadata); err != nil { + if err := json.NewDecoder(req.Body).Decode(&metadata); err != nil { t.Fatal(err) } if metadata.Name != filepath.Base(path) || len(metadata.Parents) != 1 || metadata.Parents[0] != "folder-123" { t.Fatalf("metadata = %+v", metadata) } - filePart, err := reader.NextPart() +} + +func TestBackupProgressReaderReportsBytes(t *testing.T) { + var updates []int64 + reader := &backupProgressReader{reader: strings.NewReader("encrypted backup"), onProgress: func(uploaded int64) { + updates = append(updates, uploaded) + }} + raw, err := io.ReadAll(reader) + if err != nil || string(raw) != "encrypted backup" { + t.Fatalf("read = %q, %v", raw, err) + } + if len(updates) == 0 || updates[len(updates)-1] != int64(len(raw)) { + t.Fatalf("progress updates = %v", updates) + } +} + +func TestGoogleDriveUploadMessage(t *testing.T) { + tests := []struct { + status int + body string + want string + }{ + {http.StatusUnauthorized, `{}`, "授权已失效"}, + {http.StatusForbidden, `{"reason":"storageQuotaExceeded"}`, "空间不足"}, + {http.StatusForbidden, `{}`, "无上传权限"}, + {http.StatusTooManyRequests, `{}`, "请求过于频繁"}, + } + for _, test := range tests { + message := googleDriveUploadMessage(&googleDriveAPIError{Operation: "upload", StatusCode: test.status, Body: test.body}) + if !strings.Contains(message, test.want) { + t.Fatalf("message %q does not contain %q", message, test.want) + } + } +} + +func TestGoogleDriveChunkUploadAndProgress(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "large-backup.tar.zst.enc") + size := int64(googleDriveUploadChunkSize + 3) + file, err := os.Create(path) if err != nil { t.Fatal(err) } - raw, err := io.ReadAll(filePart) - if err != nil || string(raw) != "encrypted backup" { - t.Fatalf("uploaded bytes = %q, %v", raw, err) + if err := file.Truncate(size); err != nil { + t.Fatal(err) + } + _ = file.Close() + var ranges []string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + ranges = append(ranges, r.Header.Get("Content-Range")) + _, _ = io.Copy(io.Discard, r.Body) + if len(ranges) == 1 { + w.WriteHeader(http.StatusPermanentRedirect) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = io.WriteString(w, `{"id":"uploaded"}`) + })) + defer server.Close() + a := newTestApp(t) + stopTestWorkers(a) + if !a.startBackupTransfer("googleDrive", path) { + t.Fatal("failed to start transfer") + } + if err := a.uploadGoogleDriveChunks(context.Background(), server.Client(), server.URL, path, size); err != nil { + t.Fatal(err) + } + wantRanges := []string{ + fmt.Sprintf("bytes 0-%d/%d", googleDriveUploadChunkSize-1, size), + fmt.Sprintf("bytes %d-%d/%d", googleDriveUploadChunkSize, size-1, size), + } + if len(ranges) != len(wantRanges) || ranges[0] != wantRanges[0] || ranges[1] != wantRanges[1] { + t.Fatalf("content ranges = %v, want %v", ranges, wantRanges) + } + transfer := a.backupTransfers[backupTransferKey("googleDrive", path)] + if transfer == nil || transfer.Uploaded != size { + t.Fatalf("transfer = %+v", transfer) } } diff --git a/apps/web/src/lib/api-types.ts b/apps/web/src/lib/api-types.ts index 955ae0a..82119af 100644 --- a/apps/web/src/lib/api-types.ts +++ b/apps/web/src/lib/api-types.ts @@ -204,9 +204,10 @@ export type SystemUpdateResult = { } export type BackupItem = { name: string; size: number; createdAt: string; sha256?: string } export type BackupJob = { status: "running" | "success" | "failed"; startedAt: string; error?: string } +export type BackupTransfer = { provider: "telegram" | "googleDrive"; name: string; status: "queued" | "running" | "success" | "failed"; uploaded: number; total: number; startedAt: string; finishedAt?: string; error?: string } export type BackupSchedule = { enabled: boolean; days: number; passwordSet: boolean; passwordHint?: string; serverIp: string; chatId: string; telegramMode: "system" | "custom"; telegramEnabled: boolean; googleDriveEnabled: boolean } export type GoogleDriveBackupStatus = { clientId: string; clientSecretSet: boolean; connected: boolean; folderName: string } -export type BackupList = { enabled: boolean; telegramSet: boolean; telegramLimit: number; job?: BackupJob; items: BackupItem[]; schedule: BackupSchedule; googleDrive: GoogleDriveBackupStatus } +export type BackupList = { enabled: boolean; telegramSet: boolean; telegramLimit: number; job?: BackupJob; items: BackupItem[]; schedule: BackupSchedule; googleDrive: GoogleDriveBackupStatus; transfers: BackupTransfer[] } export type SystemSettings = { publicHostname: string publicBaseUrl: string diff --git a/apps/web/src/pages/admin.tsx b/apps/web/src/pages/admin.tsx index 0b50744..98f2c74 100644 --- a/apps/web/src/pages/admin.tsx +++ b/apps/web/src/pages/admin.tsx @@ -24,7 +24,7 @@ import { SystemVersionDialog } from "@/components/system-version-dialog" import { useMe } from "@/hooks/use-me" import { useToast } from "@/hooks/use-toast" import { hasAnyPermission, hasPermission } from "@/lib/permissions" -import type { PermissionKey, TelegramPairing } from "@/lib/api-types" +import type { BackupTransfer, PermissionKey, TelegramPairing } from "@/lib/api-types" type Section = "overview" | "users" | "permissionGroups" | "domains" | "mailboxes" | "aliases" | "messages" | "sendAudit" | "backups" | "settings" type SettingsTab = "base" | "smtp" | "storage" | "mail" | "notifications" | "externalImap" | "templates" | "security" | "about" @@ -271,7 +271,7 @@ function BackupsSection() { const backups = useQuery({ queryKey: ["admin", "backups"], queryFn: api.backups, - refetchInterval: (query) => query.state.data?.job?.status === "running" ? 2000 : false, + refetchInterval: (query) => query.state.data?.job?.status === "running" || query.state.data?.transfers?.some((item) => item.status === "running" || item.status === "queued") ? 2000 : false, }) const [createOpen, setCreateOpen] = React.useState(false) const [password, setPassword] = React.useState("") @@ -346,7 +346,7 @@ function BackupsSection() { }) const sendTelegram = useMutation({ mutationFn: api.sendBackupTelegram, - onSuccess: () => toast({ title: "已发送到 Telegram" }), + onSuccess: async () => { await qc.invalidateQueries({ queryKey: ["admin", "backups"] }); toast({ title: "已开始发送到 Telegram" }) }, onError: (error) => toast({ title: "发送失败", description: error instanceof Error ? error.message : "请稍后重试" }), }) const testBackupTelegram = useMutation({ @@ -370,7 +370,7 @@ function BackupsSection() { }) const sendDrive = useMutation({ mutationFn: api.sendBackupGoogleDrive, - onSuccess: () => toast({ title: "已上传到 Google 云端硬盘" }), + onSuccess: async () => { await qc.invalidateQueries({ queryKey: ["admin", "backups"] }); toast({ title: "已开始上传到 Google 云端硬盘" }) }, onError: (error) => toast({ title: "上传失败", description: error instanceof Error ? error.message : "请稍后重试" }), }) const connectDrive = useMutation({ @@ -445,6 +445,10 @@ function BackupsSection() { if (schedulePassword !== scheduleConfirmPassword) { toast({ title: "两次输入的备份密码不一致" }); return } savePassword.mutate() } + const transferGroups = Object.values((backups.data?.transfers || []).reduce>((groups, transfer) => { + ;(groups[transfer.name] ||= []).push(transfer) + return groups + }, {})).sort((left, right) => Date.parse(right[0]?.startedAt || "") - Date.parse(left[0]?.startedAt || "")) return (
@@ -461,6 +465,18 @@ function BackupsSection() { {job?.status === "failed" &&
{job.error || "备份生成失败"}
} {job?.status === "success" &&
最近一次备份已完成。
} {!job &&

手动备份与定时备份共用同一个恢复密码，避免不同备份使用不同密码。

} + {transferGroups.slice(0, 3).map((transfers) =>
+
{transfers[0].name}
+ {transfers.sort((a, b) => (a.provider === "telegram" ? 0 : 1) - (b.provider === "telegram" ? 0 : 1)).map((transfer) => { + const percent = transfer.total > 0 ? Math.min(100, Math.round(transfer.uploaded * 100 / transfer.total)) : 0 + const label = transfer.provider === "telegram" ? "Telegram" : "Google 云端硬盘" + return
+
{label}{transfer.status === "queued" ? "等待上传" : transfer.status === "running" ? `${percent}% · ${formatBytes(transfer.uploaded)} / ${formatBytes(transfer.total)}` : transfer.status === "success" ? "上传完成" : "上传失败"}
+
+ {transfer.error &&

{transfer.error}

} +
+ })} +
)}
本地备份保留最近 10 份