diff --git a/.gitignore b/.gitignore index fa108dd9..a7af12bd 100644 --- a/.gitignore +++ b/.gitignore @@ -56,3 +56,4 @@ s3_cache /.idea/ /uploads/ /*-source/ +/.cache/ diff --git a/config.example.yaml b/config.example.yaml index 6d165661..fc0a1d14 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -96,22 +96,6 @@ worker: otel: sampling_rate: 0.0 # Trace sampling rate (0.0 – 1.0) -# ─── S3-Compatible File Storage ───────────────────────────────────────────────── -# Compatible with AWS S3, MinIO, Cloudflare R2, Tencent COS, etc. -s3: - enabled: false - endpoint: "https://.r2.cloudflarestorage.com" - region: "auto" - bucket: "" - access_key_id: "" - secret_access_key: "" - path_style: false # Set true for self-hosted S3 (e.g. MinIO) - key_prefix: "" # Optional prefix for all object keys, e.g. "uploads/" - cdn_url: "" # CDN base URL (e.g. https://cdn.example.com); falls back to S3 if empty - local_cache: - enabled: false - cache_dir: "./s3_cache" - # ─── ClickHouse (optional) ────────────────────────────────────────────────────── clickhouse: enabled: false diff --git a/docs/docs.go b/docs/docs.go index 67be53ae..c532a46d 100644 --- a/docs/docs.go +++ b/docs/docs.go @@ -5601,8 +5601,7 @@ const docTemplate = `{ ] }, "value": { - "type": "string", - "maxLength": 255 + "type": "string" }, "visibility": { "type": "integer", @@ -5668,8 +5667,7 @@ const docTemplate = `{ "maxLength": 255 }, "value": { - "type": "string", - "maxLength": 255 + "type": "string" }, "visibility": { "type": "integer", diff --git a/docs/swagger.json b/docs/swagger.json index cb214ff6..53d90ddf 100644 --- a/docs/swagger.json +++ b/docs/swagger.json @@ -5594,8 +5594,7 @@ ] }, "value": { - "type": "string", - "maxLength": 255 + "type": "string" }, "visibility": { "type": "integer", @@ -5661,8 +5660,7 @@ "maxLength": 255 }, "value": { - "type": "string", - "maxLength": 255 + "type": "string" }, "visibility": { "type": "integer", diff --git a/docs/swagger.yaml b/docs/swagger.yaml index 7f66c3f6..bc9a83c7 100644 --- a/docs/swagger.yaml +++ b/docs/swagger.yaml @@ -659,7 +659,6 @@ definitions: - business type: string value: - maxLength: 255 type: string visibility: enum: @@ -708,7 +707,6 @@ definitions: maxLength: 255 type: string value: - maxLength: 255 type: string visibility: enum: diff --git a/frontend/components/common/admin/files.tsx b/frontend/components/common/admin/files.tsx index 878a874a..550f6433 100644 --- a/frontend/components/common/admin/files.tsx +++ b/frontend/components/common/admin/files.tsx @@ -6,6 +6,7 @@ import {motion} from "motion/react" import {Tabs, TabsContent, TabsList, TabsTrigger} from "@/components/ui/tabs" import {FileStats} from "./file-stats" import {FileList} from "./file-list" +import {StorageConfigTab} from "./storage-config-tab" export function FilesMain() { const [activeTab, setActiveTab] = React.useState("stats") @@ -15,7 +16,7 @@ export function FilesMain() { initial={{ opacity: 0, y: 15 }} animate={{ opacity: 1, y: 0 }} transition={{ duration: 0.35, ease: "easeOut" }} - className="py-6 space-y-6 w-full" + className="flex w-full flex-col gap-6 py-6" > {/* 顶部标题区 */}
@@ -29,10 +30,11 @@ export function FilesMain() {
- - + + 文件存储信息 文件列表 + 存储配置 {/* ──────── TAB 1: 统计看板 ──────── */} @@ -44,6 +46,10 @@ export function FilesMain() { + + + + ) diff --git a/frontend/components/common/admin/storage-config-tab.tsx b/frontend/components/common/admin/storage-config-tab.tsx new file mode 100644 index 00000000..417e64e9 --- /dev/null +++ b/frontend/components/common/admin/storage-config-tab.tsx @@ -0,0 +1,339 @@ +"use client" + +import * as React from "react" +import {useMutation, useQuery, useQueryClient} from "@tanstack/react-query" +import {Database, Loader2, Play, Save} from "lucide-react" +import {toast} from "sonner" + +import {Badge} from "@/components/ui/badge" +import {Button} from "@/components/ui/button" +import {Card, CardContent, CardDescription, CardFooter, CardHeader, CardTitle} from "@/components/ui/card" +import {Field, FieldDescription, FieldGroup, FieldLabel} from "@/components/ui/field" +import {Input} from "@/components/ui/input" +import {Progress} from "@/components/ui/progress" +import {Select, SelectContent, SelectGroup, SelectItem, SelectTrigger, SelectValue} from "@/components/ui/select" +import {Switch} from "@/components/ui/switch" +import {AdminService} from "@/lib/services/admin/admin.service" +import type { + ObjectStorageConfig, + StorageConfig, + StorageDriver, + TaskExecution, + TaskExecutionStatus +} from "@/lib/services/admin/types" + +const storageConfigKey = "storage_config" +const storageMigrationTaskType = "storage_migration" + +const driverLabels: Record = { + local: "本地文件系统", + s3: "AWS S3", + r2: "Cloudflare R2", + minio: "MinIO", + oss: "阿里云 OSS", + webdav: "WebDAV", +} + +const emptyObjectConfig: ObjectStorageConfig = { + endpoint: "", + region: "", + bucket: "", + access_key_id: "", + secret_access_key: "", + path_style: false, + key_prefix: "", + cdn_url: "", +} + +function normalizeConfig(config: StorageConfig): StorageConfig { + const s3 = {...emptyObjectConfig, ...config.s3} + const r2 = {...emptyObjectConfig, ...config.r2} + const minio = {...emptyObjectConfig, ...config.minio} + const oss = {...emptyObjectConfig, ...config.oss} + const webdav = {...config.webdav} + return { + ...config, + local: {root: config.local?.root || "."}, + s3: {...s3, region: s3.region || "us-east-1"}, + r2: {...r2, region: r2.region || "auto"}, + minio: {...minio, region: minio.region || "us-east-1", path_style: config.minio?.path_style ?? true}, + oss, + webdav: { + endpoint: webdav.endpoint || "", + username: webdav.username || "", + password: webdav.password || "", + base_path: webdav.base_path || "", + }, + } +} + +type StorageMigrationView = { + state: "idle" | TaskExecutionStatus + id?: string + task_id?: string + source_driver?: StorageDriver + target_driver?: StorageDriver + error?: string +} + +type StorageMigrationPayload = { + target: StorageConfig +} + +function parseMigrationPayload(payload: string): StorageConfig | undefined { + try { + const parsed = JSON.parse(payload) as Partial + return parsed.target ? normalizeConfig(parsed.target) : undefined + } catch { + return undefined + } +} + +function driverLabel(driver?: StorageDriver): string { + return driver ? driverLabels[driver] : "未知" +} + +function latestMigration(executions: TaskExecution[], current?: StorageConfig): StorageMigrationView { + const execution = executions[0] + if (!execution || execution.status === "succeeded") { + return {state: "idle"} + } + const target = parseMigrationPayload(execution.payload) + return { + state: execution.status, + id: execution.id, + task_id: execution.task_id, + source_driver: current?.driver, + target_driver: target?.driver, + error: execution.error_message, + } +} + +export function StorageConfigTab() { + const queryClient = useQueryClient() + const [config, setConfig] = React.useState(null) + const query = useQuery({ + queryKey: ["admin", "storage-config"], + queryFn: async () => { + const [configRecord, executions] = await Promise.all([ + AdminService.getSystemConfig(storageConfigKey), + AdminService.listTaskExecutions({task_type: storageMigrationTaskType, page: 1, page_size: 1}), + ]) + const current = normalizeConfig(JSON.parse(configRecord.value) as StorageConfig) + const migration = latestMigration(executions.items, current) + return { + config: current, + migration, + } + }, + refetchInterval: (result) => { + const state = result.state.data?.migration.state + return state === "pending" || state === "running" ? 2000 : false + }, + }) + + React.useEffect(() => { + if (query.data?.config) { + setConfig(normalizeConfig(query.data.config)) + } + }, [query.data?.config]) + + const saveMutation = useMutation({ + mutationFn: async (value: StorageConfig) => { + if (query.data?.config.driver === value.driver) { + await AdminService.updateSystemConfig(storageConfigKey, { + value: JSON.stringify(value), + }) + return "" + } + const payload: StorageMigrationPayload = {target: value} + return AdminService.dispatchTask({task_type: storageMigrationTaskType, payload: JSON.stringify(payload)}) + }, + onSuccess: (taskID) => { + toast.success(taskID ? "配置已保存，存储迁移任务已下发" : "存储配置已保存") + void queryClient.invalidateQueries({queryKey: ["admin", "storage-config"]}) + }, + onError: (error: Error) => { + toast.error(error.message || "保存存储配置失败") + void queryClient.invalidateQueries({queryKey: ["admin", "storage-config"]}) + }, + }) + const runMutation = useMutation({ + mutationFn: (executionID: string) => AdminService.retryTaskExecution(executionID), + onSuccess: () => { + toast.success("存储迁移任务已重新下发") + void queryClient.invalidateQueries({queryKey: ["admin", "storage-config"]}) + }, + onError: (error: Error) => toast.error(error.message || "运行存储迁移失败"), + }) + + if (query.isPending || !config) { + return
+ } + + const migration = query.data?.migration + const isReadOnly = migration ? migration.state !== "idle" : false + const isFormDisabled = migration ? (migration.state === "pending" || migration.state === "running") : false + + const updateObject = (driver: "s3" | "r2" | "minio" | "oss", patch: Partial) => { + setConfig((current) => current ? { + ...current, + [driver]: {...current[driver], ...patch}, + } : current) + } + + return ( +
+ {isReadOnly && migration && ( + + + + 存储维护模式 + + {migration.state} + + + + {driverLabel(migration.source_driver)} → {driverLabel(migration.target_driver)}。迁移期间文件只允许读取，禁止上传、删除和清理。 + + + + +

+ 迁移进度请查看任务执行日志：{migration.task_id || "尚未下发"} +

+ {migration.error &&

{migration.error}

} + + + + + + )} + + + + + + 文件存储 + + + 默认使用本地存储。切换存储类型且已有文件时，系统会自动进入维护模式并迁移文件。 + + + + + + 存储类型 + + + + {config.driver === "local" && ( + setConfig({...config, local: {root}})} + /> + )} + + {(config.driver === "s3" || config.driver === "r2" || config.driver === "minio" || config.driver === "oss") && ( + updateObject(config.driver as "s3" | "r2" | "minio" | "oss", patch)} + /> + )} + + {config.driver === "webdav" && ( + <> + setConfig({...config, webdav: {...config.webdav, endpoint}})} /> + setConfig({...config, webdav: {...config.webdav, username}})} /> + setConfig({...config, webdav: {...config.webdav, password}})} /> + setConfig({...config, webdav: {...config.webdav, base_path}})} /> + + )} + + + + + + +
+ ) +} + +function ObjectFields({ + driver, + value, + onChange, +}: { + driver: "s3" | "r2" | "minio" | "oss" + value: ObjectStorageConfig + onChange: (patch: Partial) => void +}) { + return ( + <> + {driver === "r2" && onChange({account_id})} />} + {driver !== "s3" && onChange({endpoint})} />} + onChange({region})} /> + onChange({bucket})} /> + onChange({access_key_id})} /> + onChange({secret_access_key})} /> + onChange({key_prefix})} /> + onChange({cdn_url})} /> + {(driver === "s3" || driver === "minio") && ( + + Path Style + onChange({path_style})} /> + MinIO 等自托管 S3 通常需要开启。 + + )} + + ) +} + +function TextField({ + label, + value, + onChange, + placeholder, + type = "text", +}: { + label: string + value: string + onChange: (value: string) => void + placeholder?: string + type?: React.HTMLInputTypeAttribute +}) { + return ( + + {label} + onChange(event.target.value)} /> + + ) +} diff --git a/frontend/lib/services/admin/admin.service.ts b/frontend/lib/services/admin/admin.service.ts index 0736a171..ad973125 100644 --- a/frontend/lib/services/admin/admin.service.ts +++ b/frontend/lib/services/admin/admin.service.ts @@ -155,8 +155,6 @@ export class AdminService extends BaseService { return this.get('/uploads/types'); } - - // ==================== 认证源管理 ==================== static async listAuthSources(): Promise { diff --git a/frontend/lib/services/admin/types.ts b/frontend/lib/services/admin/types.ts index 324887c2..3206c60e 100644 --- a/frontend/lib/services/admin/types.ts +++ b/frontend/lib/services/admin/types.ts @@ -26,7 +26,7 @@ export interface SystemConfig { export interface CreateSystemConfigRequest { /** 配置键（最大64字符） */ key: string; - /** 配置值（最大255字符） */ + /** 配置值 */ value: string; /** 配置类型：'system' | 'business' */ type: 'system' | 'business'; @@ -40,7 +40,7 @@ export interface CreateSystemConfigRequest { * 更新系统配置请求参数 */ export interface UpdateSystemConfigRequest { - /** 配置值（最大255字符） */ + /** 配置值 */ value: string; /** 是否对公共配置接口可见：0 不可见，1 可见 */ visibility?: 0 | 1; @@ -48,7 +48,34 @@ export interface UpdateSystemConfigRequest { description?: string; } +export type StorageDriver = 'local' | 's3' | 'r2' | 'minio' | 'oss' | 'webdav'; +export interface ObjectStorageConfig { + endpoint: string; + region: string; + bucket: string; + access_key_id: string; + secret_access_key: string; + account_id?: string; + path_style: boolean; + key_prefix: string; + cdn_url: string; +} + +export interface StorageConfig { + driver: StorageDriver; + local: { root: string }; + s3: ObjectStorageConfig; + r2: ObjectStorageConfig; + minio: ObjectStorageConfig; + oss: ObjectStorageConfig; + webdav: { + endpoint: string; + username: string; + password: string; + base_path: string; + }; +} // ==================== 任务管理 ==================== diff --git a/go.mod b/go.mod index d932d63d..2a0b5d23 100644 --- a/go.mod +++ b/go.mod @@ -52,6 +52,7 @@ require ( filippo.io/edwards25519 v1.1.0 // indirect github.com/ClickHouse/ch-go v0.66.1 // indirect github.com/KyleBanks/depth v1.2.1 // indirect + github.com/aliyun/alibabacloud-oss-go-sdk-v2 v1.5.1 // indirect github.com/andybalholm/brotli v1.2.0 // indirect github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.8 // indirect github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.21 // indirect @@ -142,6 +143,7 @@ require ( github.com/spf13/afero v1.15.0 // indirect github.com/spf13/cast v1.10.0 // indirect github.com/spf13/pflag v1.0.10 // indirect + github.com/studio-b12/gowebdav v0.12.0 // indirect github.com/subosito/gotenv v1.6.0 // indirect github.com/twitchyliquid64/golang-asm v0.15.1 // indirect github.com/ugorji/go/codec v1.3.1 // indirect diff --git a/go.sum b/go.sum index 0b15482d..976efd9b 100644 --- a/go.sum +++ b/go.sum @@ -8,6 +8,8 @@ github.com/KyleBanks/depth v1.2.1 h1:5h8fQADFrWtarTdtDudMmGsC7GPbOAu6RVB3ffsVFHc github.com/KyleBanks/depth v1.2.1/go.mod h1:jzSb9d0L43HxTQfT+oSA1EEp2q+ne2uh6XgeJcm8brE= github.com/alicebob/miniredis/v2 v2.38.0 h1:nZAzCR+Lj+Vxk4ZXzm2NuKq2O33RXj1XxJ2e2uP9jiw= github.com/alicebob/miniredis/v2 v2.38.0/go.mod h1:TcL7YfarKPGDAthEtl5NBeHZfeUQj6OXMm/+iu5cLMM= +github.com/aliyun/alibabacloud-oss-go-sdk-v2 v1.5.1 h1:vtiFd0hhPAbyYJjztl0wYUq/PqEGkIlDmVuTIy6zw8Y= +github.com/aliyun/alibabacloud-oss-go-sdk-v2 v1.5.1/go.mod h1:FTzydeQVmR24FI0D6XWUOMKckjXehM/jgMn1xC+DA9M= github.com/andybalholm/brotli v1.2.0 h1:ukwgCxwYrmACq68yiUqwIWnGY0cTPox/M94sVwToPjQ= github.com/andybalholm/brotli v1.2.0/go.mod h1:rzTDkvFWvIrjDXZHkuS16NPggd91W3kUSvPlQ1pLaKY= github.com/aws/aws-sdk-go-v2 v1.41.5 h1:dj5kopbwUsVUVFgO4Fi5BIT3t4WyqIDjGKCangnV/yY= @@ -289,6 +291,8 @@ github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXl github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +github.com/studio-b12/gowebdav v0.12.0 h1:kFRtQECt8jmVAvA6RHBz3geXUGJHUZA6/IKpOVUs5kM= +github.com/studio-b12/gowebdav v0.12.0/go.mod h1:bHA7t77X/QFExdeAnDzK6vKM34kEZAcE1OX4MfiwjkE= github.com/subosito/gotenv v1.6.0 h1:9NlTDc1FTs4qu0DDq7AEtTPNw6SVm7uBMsUCUjABIf8= github.com/subosito/gotenv v1.6.0/go.mod h1:Dk4QP5c2W3ibzajGcXpNraDfq2IrhjMIvMSWPKKo0FU= github.com/swaggo/files v1.0.1 h1:J1bVJ4XHZNq0I46UU90611i9/YzdrF7x92oX1ig5IdE= diff --git a/internal/apps/admin/system_config/routers.go b/internal/apps/admin/system_config/routers.go index 764d0eac..d949a673 100644 --- a/internal/apps/admin/system_config/routers.go +++ b/internal/apps/admin/system_config/routers.go @@ -5,11 +5,15 @@ package system_config import ( + "encoding/json" "errors" "net/http" + "time" "github.com/Rain-kl/Wavelet/internal/db" + "github.com/Rain-kl/Wavelet/internal/logger" "github.com/Rain-kl/Wavelet/internal/model" + "github.com/Rain-kl/Wavelet/internal/storage" "github.com/Rain-kl/Wavelet/internal/util" mail "github.com/Rain-kl/Wavelet/internal/util/mail" "github.com/gin-gonic/gin" @@ -21,7 +25,7 @@ const maskedConfigValue = "******" // CreateSystemConfigRequest 创建系统配置请求 type CreateSystemConfigRequest struct { Key string `json:"key" binding:"required,max=64"` - Value string `json:"value" binding:"required,max=255"` + Value string `json:"value" binding:"required"` Type string `json:"type" binding:"required,oneof=system business"` Visibility int `json:"visibility" binding:"oneof=0 1"` Description string `json:"description" binding:"max=255"` @@ -29,7 +33,7 @@ type CreateSystemConfigRequest struct { // UpdateSystemConfigRequest 更新系统配置请求 type UpdateSystemConfigRequest struct { - Value string `json:"value" binding:"required,max=255"` + Value string `json:"value" binding:"required"` Visibility *int `json:"visibility" binding:"omitempty,oneof=0 1"` Description string `json:"description" binding:"max=255"` } @@ -118,9 +122,7 @@ func ListSystemConfigs(c *gin.Context) { } for i := range configs { - if configs[i].Key == model.ConfigKeySMTPPassword && configs[i].Value != "" { - configs[i].Value = maskedConfigValue - } + configs[i].Value = maskSensitiveConfig(configs[i].Key, configs[i].Value) } c.JSON(http.StatusOK, util.OK(configs)) @@ -150,9 +152,7 @@ func GetSystemConfig(c *gin.Context) { return } - if config.Key == model.ConfigKeySMTPPassword && config.Value != "" { - config.Value = maskedConfigValue - } + config.Value = maskSensitiveConfig(config.Key, config.Value) c.JSON(http.StatusOK, util.OK(config)) } @@ -193,6 +193,14 @@ func UpdateSystemConfig(c *gin.Context) { return } + var originalDriver storage.Driver + if key == model.ConfigKeyStorageConfig { + var currentCfg storage.Config + if err := json.Unmarshal([]byte(config.Value), ¤tCfg); err == nil { + originalDriver = currentCfg.Driver + } + } + if err := db.DB(c.Request.Context()).Transaction(func(tx *gorm.DB) error { // 更新配置 updates := map[string]interface{}{ @@ -214,6 +222,24 @@ func UpdateSystemConfig(c *gin.Context) { return err } + if key == model.ConfigKeyStorageConfig && originalDriver != "" { + var newCfg storage.Config + if err := json.Unmarshal([]byte(req.Value), &newCfg); err == nil { + if newCfg.Driver == originalDriver { + // Mark failed storage:migrate task execution as succeeded + if err := tx.Model(&model.TaskExecution{}). + Where("task_type = ? AND status = ?", "storage:migrate", model.TaskExecutionStatusFailed). + Updates(map[string]any{ + "status": model.TaskExecutionStatusSucceeded, + "result": "存储配置直接更新，故障迁移任务自动标记为已解决", + "finished_at": time.Now(), + }).Error; err != nil { + logger.ErrorF(c.Request.Context(), "自动更新迁移任务状态失败: %v", err) + } + } + } + } + return nil }); err != nil { c.JSON(http.StatusInternalServerError, util.Err(err.Error())) @@ -288,3 +314,22 @@ func TestSMTP(c *gin.Context) { c.JSON(http.StatusOK, util.OK(resp)) } + +func maskSensitiveConfig(key, value string) string { + if value == "" { + return value + } + switch key { + case model.ConfigKeySMTPPassword: + return maskedConfigValue + case model.ConfigKeyStorageConfig: + var cfg storage.Config + if err := json.Unmarshal([]byte(value), &cfg); err == nil { + masked := storage.MaskSecrets(cfg) + if val, err := json.Marshal(masked); err == nil { + return string(val) + } + } + } + return value +} diff --git a/internal/apps/admin/system_config/routers_test.go b/internal/apps/admin/system_config/routers_test.go index 80ee5e22..268aa907 100644 --- a/internal/apps/admin/system_config/routers_test.go +++ b/internal/apps/admin/system_config/routers_test.go @@ -23,7 +23,7 @@ import ( "github.com/gin-gonic/gin" ) -const expectedDefaultConfigsCount = 29 +const expectedDefaultConfigsCount = 30 func setupTestRouter(authUser *model.User) *gin.Engine { gin.SetMode(gin.TestMode) diff --git a/internal/apps/upload/constants.go b/internal/apps/upload/constants.go index 2276d8fc..b927a86b 100644 --- a/internal/apps/upload/constants.go +++ b/internal/apps/upload/constants.go @@ -3,6 +3,8 @@ package upload +import "github.com/Rain-kl/Wavelet/internal/storage" + const ( maxUploadSize = 32 * 1024 * 1024 // 32MB detectContentBytes = 512 // http.DetectContentType 需要的最小字节数 @@ -12,5 +14,5 @@ const ( imageQualityMedium = "medium" imageQualityHigh = "high" imageQualityOrigin = "origin" - storageDriverLocal = "local" + storageDriverLocal = string(storage.DriverLocal) ) diff --git a/internal/apps/upload/errs.go b/internal/apps/upload/errs.go index 1660dc74..5b2cafd6 100644 --- a/internal/apps/upload/errs.go +++ b/internal/apps/upload/errs.go @@ -27,6 +27,7 @@ const ( ErrQueryFileCountFailed = "查询文件数量失败" ErrQueryFileListFailed = "查询文件列表失败" ErrDeleteFileFailed = "删除文件失败" + ErrStorageReadOnly = "存储迁移维护中，当前仅允许读取文件" ErrS3KeyRequired = "s3 key must not be empty" ErrS3KeyTooLongFormat = "s3 key exceeds maximum length of %d" ErrS3KeyStartsWithSlash = "s3 key must not start with /" diff --git a/internal/apps/upload/file_management.go b/internal/apps/upload/file_management.go new file mode 100644 index 00000000..925c4a65 --- /dev/null +++ b/internal/apps/upload/file_management.go @@ -0,0 +1,167 @@ +// Copyright 2026 Arctel.net +// SPDX-License-Identifier: Apache-2.0 + +package upload + +import ( + "errors" + "net/http" + "sort" + "strconv" + "strings" + + "github.com/Rain-kl/Wavelet/internal/apps/oauth" + "github.com/Rain-kl/Wavelet/internal/db" + "github.com/Rain-kl/Wavelet/internal/model" + "github.com/Rain-kl/Wavelet/internal/util" + "github.com/gin-gonic/gin" + "gorm.io/gorm" +) + +type listMyFilesRequest struct { + Page int `form:"page"` + PageSize int `form:"page_size"` + Keyword string `form:"keyword"` + Type string `form:"type"` + Extension string `form:"extension"` +} + +type listMyFilesResponse struct { + Total int64 `json:"total"` + Page int `json:"page"` + PageSize int `json:"page_size"` + Items []model.Upload `json:"items"` +} + +// ListMyFiles 获取当前用户上传的文件列表 +// @Summary 获取我的文件列表 +// @Description 分页获取当前登录用户上传的文件，支持文件名关键词、业务类型、扩展名过滤 +// @Tags upload +// @Produce json +// @Param page query int false "页码（默认 1）" +// @Param page_size query int false "每页数量（默认 20，最大 100）" +// @Param keyword query string false "文件名关键词（模糊匹配）" +// @Param type query string false "业务分类过滤" +// @Param extension query string false "扩展名过滤" +// @Security SessionCookie +// @Success 200 {object} util.ResponseAny{data=listMyFilesResponse} "查询成功" +// @Failure 401 {object} util.ResponseAny "未登录" +// @Router /api/v1/upload/my [get] +func ListMyFiles(c *gin.Context) { + currUser, _ := util.GetFromContext[*model.User](c, oauth.UserObjKey) + ctx := c.Request.Context() + + var req listMyFilesRequest + if err := c.ShouldBindQuery(&req); err != nil { + c.JSON(http.StatusOK, util.Err(ErrInvalidParams)) + return + } + if req.Page <= 0 { + req.Page = 1 + } + if req.PageSize <= 0 || req.PageSize > 100 { + req.PageSize = 20 + } + + query := db.DB(ctx).Model(&model.Upload{}). + Where("user_id = ? AND status != ?", currUser.ID, model.UploadStatusDeleted) + + if req.Keyword != "" { + query = query.Where("LOWER(file_name) LIKE ?", "%"+strings.ToLower(req.Keyword)+"%") + } + if req.Type != "" { + query = query.Where("type = ?", req.Type) + } + if req.Extension != "" { + query = query.Where("extension = ?", strings.ToLower(req.Extension)) + } + + var total int64 + if err := query.Count(&total).Error; err != nil { + c.JSON(http.StatusOK, util.Err(ErrQueryFileCountFailed)) + return + } + + var items []model.Upload + offset := (req.Page - 1) * req.PageSize + if err := query.Order("created_at DESC").Offset(offset).Limit(req.PageSize).Find(&items).Error; err != nil { + c.JSON(http.StatusOK, util.Err(ErrQueryFileListFailed)) + return + } + + c.JSON(http.StatusOK, util.OK(listMyFilesResponse{ + Total: total, + Page: req.Page, + PageSize: req.PageSize, + Items: items, + })) +} + +// DeleteFile 软删除文件记录 +// @Summary 删除文件 +// @Description 将文件状态置为 deleted（软删除），不会立即清理底层存储对象 +// @Tags upload +// @Produce json +// @Param id path string true "文件 ID" +// @Security SessionCookie +// @Success 200 {object} util.ResponseAny "删除成功" +// @Failure 403 {object} util.ResponseAny "无权操作" +// @Failure 404 {object} util.ResponseAny "文件不存在" +// @Router /api/v1/upload/{id} [delete] +func DeleteFile(c *gin.Context) { + currUser, _ := util.GetFromContext[*model.User](c, oauth.UserObjKey) + ctx := c.Request.Context() + if storageReadOnly(ctx) { + c.JSON(http.StatusConflict, util.Err(ErrStorageReadOnly)) + return + } + + uploadID, err := strconv.ParseUint(c.Param("id"), 10, 64) + if err != nil { + c.JSON(http.StatusOK, util.Err(ErrInvalidFileID)) + return + } + + var upload model.Upload + if err := db.DB(ctx).Where("id = ? AND status != ?", uploadID, model.UploadStatusDeleted).First(&upload).Error; err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + c.AbortWithStatus(http.StatusNotFound) + return + } + c.JSON(http.StatusOK, util.Err(ErrQueryUploadRecordFailed)) + return + } + if upload.UserID != currUser.ID && !currUser.IsAdmin { + c.AbortWithStatus(http.StatusForbidden) + return + } + if err := db.DB(ctx).Model(&upload).Update("status", model.UploadStatusDeleted).Error; err != nil { + c.JSON(http.StatusOK, util.Err(ErrDeleteFileFailed)) + return + } + c.JSON(http.StatusOK, util.OKNil()) +} + +// GetDistinctUploadTypes 获取数据库中所有已存在的文件业务类型 +// @Summary 获取文件业务类型列表 +// @Description 返回数据库中所有已上传文件实际拥有的业务类型列表 +// @Tags admin +// @Produce json +// @Security SessionCookie +// @Success 200 {object} util.ResponseAny{data=[]string} "业务类型列表" +// @Failure 401 {object} util.ResponseAny "未登录" +// @Failure 403 {object} util.ResponseAny "无管理员权限" +// @Failure 500 {object} util.ResponseAny "内部错误" +// @Router /api/v1/admin/uploads/types [get] +func GetDistinctUploadTypes(c *gin.Context) { + var dbTypes []string + if err := db.DB(c.Request.Context()).Model(&model.Upload{}). + Where("type IS NOT NULL AND type != ''"). + Distinct(). + Pluck("type", &dbTypes).Error; err != nil { + c.JSON(http.StatusInternalServerError, util.Err(err.Error())) + return + } + sort.Strings(dbTypes) + c.JSON(http.StatusOK, util.OK(dbTypes)) +} diff --git a/internal/apps/upload/file_server.go b/internal/apps/upload/file_server.go index e2f90f2b..d7eb4754 100644 --- a/internal/apps/upload/file_server.go +++ b/internal/apps/upload/file_server.go @@ -12,7 +12,6 @@ import ( "fmt" "io" "net/http" - "os" "strconv" "strings" @@ -22,7 +21,6 @@ import ( "github.com/Rain-kl/Wavelet/internal/diskcache" "github.com/Rain-kl/Wavelet/internal/logger" "github.com/Rain-kl/Wavelet/internal/model" - "github.com/Rain-kl/Wavelet/internal/storage" "github.com/Rain-kl/Wavelet/internal/util" "github.com/gin-gonic/gin" "gorm.io/gorm" @@ -236,48 +234,21 @@ func normalizeImageQuality(quality string) string { // serveOriginal 原始文件的流式响应逻辑 func serveOriginal(c *gin.Context, upload *model.Upload) { - if upload.StorageDriver == storageDriverLocal || (upload.StorageDriver == "" && !storage.IsEnabled()) { - c.File(upload.FilePath) - return - } - - // Retrieve file from S3 (via CDN if configured) - obj, err := storage.GetObjectViaCache(c.Request.Context(), upload.FilePath) + obj, err := openStoredObject(c.Request.Context(), upload) if err != nil { c.AbortWithStatus(http.StatusNotFound) return } - - // Cachefile - if obj.CachePath != "" { - c.File(obj.CachePath) - return - } - - // Stream from CDN/S3 defer func() { _ = obj.Body.Close() }() - - // Respond with the file content c.DataFromReader(http.StatusOK, obj.ContentLength, obj.ContentType, obj.Body, nil) } // getOriginalFileBytes 获取原始文件所有字节 func getOriginalFileBytes(ctx context.Context, upload *model.Upload) ([]byte, error) { - if upload.StorageDriver == storageDriverLocal || (upload.StorageDriver == "" && !storage.IsEnabled()) { - return os.ReadFile(upload.FilePath) - } - - // Retrieve file from S3 (via CDN if configured) - obj, err := storage.GetObjectViaCache(ctx, upload.FilePath) + obj, err := openStoredObject(ctx, upload) if err != nil { return nil, err } - - // Cachefile - if obj.CachePath != "" { - return os.ReadFile(obj.CachePath) - } - defer func() { _ = obj.Body.Close() }() return io.ReadAll(obj.Body) } diff --git a/internal/apps/upload/routers.go b/internal/apps/upload/routers.go index 85a05b00..e11e7891 100644 --- a/internal/apps/upload/routers.go +++ b/internal/apps/upload/routers.go @@ -17,16 +17,13 @@ import ( "mime/multipart" "net/http" "net/url" - "os" "path/filepath" - "sort" "strconv" "strings" "time" "github.com/Rain-kl/Wavelet/internal/apps/oauth" "github.com/Rain-kl/Wavelet/internal/common" - "github.com/Rain-kl/Wavelet/internal/config" "github.com/Rain-kl/Wavelet/internal/db" "github.com/Rain-kl/Wavelet/internal/db/idgen" "github.com/Rain-kl/Wavelet/internal/logger" @@ -157,8 +154,8 @@ func UploadFile(c *gin.Context) { id := idgen.NextUint64ID() subPath := fmt.Sprintf("uploads/%s/%d.%s", time.Now().Format("2006/01/02"), id, ext) - // 8. 写入底层存储驱动 (优先 S3 驱动，无配置或未开启则 fallback 至本地文件) - storageDriver, subPath, errMsg := storeUploadFile(ctx, id, ext, subPath, size, mimeType, &buf, &meta) + // 8. 写入当前活动存储驱动。 + storageDriver, subPath, errMsg := storeUploadFile(ctx, subPath, size, mimeType, &buf, &meta) if errMsg != "" { c.JSON(http.StatusOK, util.Err(errMsg)) return @@ -323,21 +320,12 @@ func BatchDownloadFiles(c *gin.Context) { // 打开底层文件数据源 var rc io.ReadCloser - if upload.StorageDriver == storageDriverLocal || (upload.StorageDriver == "" && !storage.IsEnabled()) { - fileSrc, err := os.Open(upload.FilePath) - if err != nil { - logger.ErrorF(ctx, "打包时读取本地文件失败: %v", err) - continue - } - rc = fileSrc - } else { - obj, err := storage.GetObject(ctx, upload.FilePath) - if err != nil { - logger.ErrorF(ctx, "打包时拉取 S3 文件失败: %v", err) - continue - } - rc = obj.Body + obj, err := openStoredObject(ctx, &upload) + if err != nil { + logger.ErrorF(ctx, "打包时读取文件失败: %v", err) + continue } + rc = obj.Body // 流式拷贝到 ZIP entry _, err = io.Copy(zipFileEntry, rc) @@ -348,131 +336,6 @@ func BatchDownloadFiles(c *gin.Context) { } } -type listMyFilesRequest struct { - Page int `form:"page"` - PageSize int `form:"page_size"` - Keyword string `form:"keyword"` - Type string `form:"type"` - Extension string `form:"extension"` -} - -type listMyFilesResponse struct { - Total int64 `json:"total"` - Page int `json:"page"` - PageSize int `json:"page_size"` - Items []model.Upload `json:"items"` -} - -// ListMyFiles 获取当前用户上传的文件列表 -// @Summary 获取我的文件列表 -// @Description 分页获取当前登录用户上传的文件，支持文件名关键词、业务类型、扩展名过滤 -// @Tags upload -// @Produce json -// @Param page query int false "页码（默认 1）" -// @Param page_size query int false "每页数量（默认 20，最大 100）" -// @Param keyword query string false "文件名关键词（模糊匹配）" -// @Param type query string false "业务分类过滤" -// @Param extension query string false "扩展名过滤" -// @Security SessionCookie -// @Success 200 {object} util.ResponseAny{data=listMyFilesResponse} "查询成功" -// @Failure 401 {object} util.ResponseAny "未登录" -// @Router /api/v1/upload/my [get] -func ListMyFiles(c *gin.Context) { - currUser, _ := util.GetFromContext[*model.User](c, oauth.UserObjKey) - ctx := c.Request.Context() - - var req listMyFilesRequest - if err := c.ShouldBindQuery(&req); err != nil { - c.JSON(http.StatusOK, util.Err(ErrInvalidParams)) - return - } - if req.Page <= 0 { - req.Page = 1 - } - if req.PageSize <= 0 || req.PageSize > 100 { - req.PageSize = 20 - } - - query := db.DB(ctx).Model(&model.Upload{}). - Where("user_id = ? AND status != ?", currUser.ID, model.UploadStatusDeleted) - - if req.Keyword != "" { - query = query.Where("LOWER(file_name) LIKE ?", "%"+strings.ToLower(req.Keyword)+"%") - } - if req.Type != "" { - query = query.Where("type = ?", req.Type) - } - if req.Extension != "" { - query = query.Where("extension = ?", strings.ToLower(req.Extension)) - } - - var total int64 - if err := query.Count(&total).Error; err != nil { - c.JSON(http.StatusOK, util.Err(ErrQueryFileCountFailed)) - return - } - - var items []model.Upload - offset := (req.Page - 1) * req.PageSize - if err := query.Order("created_at DESC").Offset(offset).Limit(req.PageSize).Find(&items).Error; err != nil { - c.JSON(http.StatusOK, util.Err(ErrQueryFileListFailed)) - return - } - - c.JSON(http.StatusOK, util.OK(listMyFilesResponse{ - Total: total, - Page: req.Page, - PageSize: req.PageSize, - Items: items, - })) -} - -// DeleteFile 软删除文件记录 -// @Summary 删除文件 -// @Description 将文件状态置为 deleted（软删除），不会立即清理底层存储对象 -// @Tags upload -// @Produce json -// @Param id path string true "文件 ID" -// @Security SessionCookie -// @Success 200 {object} util.ResponseAny "删除成功" -// @Failure 403 {object} util.ResponseAny "无权操作" -// @Failure 404 {object} util.ResponseAny "文件不存在" -// @Router /api/v1/upload/{id} [delete] -func DeleteFile(c *gin.Context) { - currUser, _ := util.GetFromContext[*model.User](c, oauth.UserObjKey) - ctx := c.Request.Context() - - idStr := c.Param("id") - uploadID, err := strconv.ParseUint(idStr, 10, 64) - if err != nil { - c.JSON(http.StatusOK, util.Err(ErrInvalidFileID)) - return - } - - var upload model.Upload - if err := db.DB(ctx).Where("id = ? AND status != ?", uploadID, model.UploadStatusDeleted).First(&upload).Error; err != nil { - if errors.Is(err, gorm.ErrRecordNotFound) { - c.AbortWithStatus(http.StatusNotFound) - return - } - c.JSON(http.StatusOK, util.Err(ErrQueryUploadRecordFailed)) - return - } - - // 仅允许文件所有者或管理员删除 - if upload.UserID != currUser.ID && !currUser.IsAdmin { - c.AbortWithStatus(http.StatusForbidden) - return - } - - if err := db.DB(ctx).Model(&upload).Update("status", model.UploadStatusDeleted).Error; err != nil { - c.JSON(http.StatusOK, util.Err(ErrDeleteFileFailed)) - return - } - - c.JSON(http.StatusOK, util.OKNil()) -} - // validateUploadExtension 校验文件后缀是否在系统允许的上传扩展名列表中 func validateUploadExtension(ctx context.Context, ext string) string { var sc model.SystemConfig @@ -499,6 +362,10 @@ func tryInstantUpload(ctx context.Context, c *gin.Context, currUser *model.User, if err != nil { return false, err } + if storageReadOnly(ctx) { + c.JSON(http.StatusConflict, util.Err(ErrStorageReadOnly)) + return true, nil + } id := idgen.NextUint64ID() newUpload := model.Upload{ @@ -527,30 +394,32 @@ func tryInstantUpload(ctx context.Context, c *gin.Context, currUser *model.User, return true, nil } -// storeUploadFile 将文件写入底层存储驱动（S3 或本地磁盘） -func storeUploadFile(ctx context.Context, id uint64, ext, subPath string, size int64, mimeType string, buf *bytes.Buffer, meta *model.UploadMetadata) (string, string, string) { - if storage.IsEnabled() { - meta.Bucket = config.Config.S3.Bucket - fullKey := storage.BuildKey(subPath) - if err := storage.PutObject(ctx, fullKey, bytes.NewReader(buf.Bytes()), size, mimeType); err != nil { - logger.ErrorF(ctx, "S3 存储上传失败: %v", err) - return "", "", ErrSaveFileFailed - } - return "s3", subPath, "" +// storeUploadFile 将文件写入当前活动存储驱动。 +func storeUploadFile(ctx context.Context, subPath string, size int64, mimeType string, buf *bytes.Buffer, meta *model.UploadMetadata) (string, string, string) { + if storageReadOnly(ctx) { + return "", "", ErrStorageReadOnly } - - localDir := filepath.Join("uploads", time.Now().Format("2006/01/02")) - if err := os.MkdirAll(localDir, uploadDirPerm); err != nil { - logger.ErrorF(ctx, "创建本地上传目录失败: %v", err) + driver, backend, err := storage.Active(ctx) + if err != nil { + logger.ErrorF(ctx, "初始化活动存储失败: %v", err) return "", "", ErrSaveFileFailed } - - localPath := filepath.Join(localDir, fmt.Sprintf("%d.%s", id, ext)) - if err := os.WriteFile(localPath, buf.Bytes(), uploadFilePerm); err != nil { - logger.ErrorF(ctx, "本地磁盘写入文件失败: %v", err) + storedPath, err := backend.Put(ctx, subPath, bytes.NewReader(buf.Bytes()), size, mimeType) + if err != nil { + logger.ErrorF(ctx, "写入 %s 存储失败: %v", driver, err) return "", "", ErrSaveFileFailed } - return storageDriverLocal, localPath, "" + switch driver { + case storage.DriverS3: + meta.Bucket = cfgBucket(ctx, driver) + case storage.DriverR2: + meta.Bucket = cfgBucket(ctx, driver) + case storage.DriverMinIO: + meta.Bucket = cfgBucket(ctx, driver) + case storage.DriverOSS: + meta.Bucket = cfgBucket(ctx, driver) + } + return string(driver), storedPath, "" } // isImageExtension 判断文件扩展名是否属于常见图片格式 @@ -590,35 +459,13 @@ func detectMimeType(buf *bytes.Buffer, header *multipart.FileHeader, size int64) // saveUploadRecord 保存上传记录到数据库，失败时清理本地垃圾文件 func saveUploadRecord(ctx context.Context, upload *model.Upload, storageDriver, filePath string) string { if err := db.DB(ctx).Create(upload).Error; err != nil { - if storageDriver == storageDriverLocal { - _ = os.Remove(filePath) + backend, backendErr := storage.ForDriver(ctx, storage.Driver(storageDriver)) + if backendErr == nil { + if deleteErr := backend.Delete(ctx, filePath); deleteErr != nil { + logger.WarnF(ctx, "清理未写入数据库的上传对象失败: %v", deleteErr) + } } return ErrSaveUploadRecordFailed } return "" } - -// GetDistinctUploadTypes 获取数据库中所有已存在的文件业务类型 -// @Summary 获取文件业务类型列表 -// @Description 返回数据库中所有已上传文件实际拥有的业务类型列表 -// @Tags admin -// @Produce json -// @Security SessionCookie -// @Success 200 {object} util.ResponseAny{data=[]string} "业务类型列表" -// @Failure 401 {object} util.ResponseAny "未登录" -// @Failure 403 {object} util.ResponseAny "无管理员权限" -// @Failure 500 {object} util.ResponseAny "内部错误" -// @Router /api/v1/admin/uploads/types [get] -func GetDistinctUploadTypes(c *gin.Context) { - var dbTypes []string - if err := db.DB(c.Request.Context()).Model(&model.Upload{}). - Where("type IS NOT NULL AND type != ''"). - Distinct(). - Pluck("type", &dbTypes).Error; err != nil { - c.JSON(http.StatusInternalServerError, util.Err(err.Error())) - return - } - - sort.Strings(dbTypes) - c.JSON(http.StatusOK, util.OK(dbTypes)) -} diff --git a/internal/apps/upload/routers_test.go b/internal/apps/upload/routers_test.go index 4a9b3a02..1106722e 100644 --- a/internal/apps/upload/routers_test.go +++ b/internal/apps/upload/routers_test.go @@ -105,12 +105,12 @@ func TestUploadFile(t *testing.T) { putCount++ return nil }, - func(ctx context.Context, key string) (*storage.ObjectInfo, error) { + func(ctx context.Context, key string) (*storage.Object, error) { data, ok := mockFiles[key] if !ok { return nil, os.ErrNotExist } - return &storage.ObjectInfo{ + return &storage.Object{ Body: io.NopCloser(bytes.NewReader(data)), ContentLength: int64(len(data)), ContentType: "application/octet-stream", @@ -821,4 +821,3 @@ func TestGetFileStats(t *testing.T) { t.Errorf("expected 1 document category, got %d", categoryMap["文档"]) } } - diff --git a/internal/apps/upload/storage_migration_task.go b/internal/apps/upload/storage_migration_task.go new file mode 100644 index 00000000..28bbd62d --- /dev/null +++ b/internal/apps/upload/storage_migration_task.go @@ -0,0 +1,268 @@ +// Copyright 2026 Arctel.net +// SPDX-License-Identifier: Apache-2.0 + +package upload + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "os" + "strings" + + "github.com/Rain-kl/Wavelet/internal/db" + "github.com/Rain-kl/Wavelet/internal/model" + "github.com/Rain-kl/Wavelet/internal/storage" + "github.com/Rain-kl/Wavelet/internal/task" + "gorm.io/gorm" +) + +const ( + // StorageMigrationTask is the Asynq task name for storage migration. + StorageMigrationTask = "storage:migrate" + // TaskTypeStorageMigration is the task metadata type for storage migration. + TaskTypeStorageMigration = "storage_migration" +) + +// StorageMigrationMeta describes the manually dispatchable migration task. +var StorageMigrationMeta = task.TaskMeta{ + Type: TaskTypeStorageMigration, + AsynqTask: StorageMigrationTask, + Name: "迁移文件存储", + Description: "将活动存储中的文件迁移到待切换的目标存储，迁移期间文件系统保持只读", + SupportsTime: false, + MaxRetry: task.DefaultMaxRetry, + Queue: task.QueueDefault, + Retryable: true, +} + +// MigrationHandler copies stored objects and activates the target backend. +type MigrationHandler struct{} + +type storageMigrationPayload struct { + Target storage.Config `json:"target"` +} + +// ValidatePayload rejects duplicate active migrations through the task framework. +func (h *MigrationHandler) ValidatePayload(payload []byte) ([]byte, error) { + normalized, _, err := normalizeStorageMigrationPayload(context.Background(), payload) + if err != nil { + return payload, err + } + active, err := hasUnresolvedMigrationTask(context.Background()) + if err != nil { + return payload, err + } + if active { + return payload, fmt.Errorf("storage migration task is already unresolved") + } + return normalized, nil +} + +// Execute migrates all unique active-storage objects to the pending backend. +func (h *MigrationHandler) Execute(ctx context.Context, payload []byte) (*task.TaskResult, error) { + active, err := storage.LoadConfig(ctx) + if err != nil { + return nil, fmt.Errorf("load active storage config: %w", err) + } + target, err := parseMigrationTargetConfig(ctx, payload) + if err != nil { + return nil, err + } + if target.Driver == active.Driver { + if err := storage.SaveActiveConfig(ctx, target); err != nil { + return nil, fmt.Errorf("activate same-driver storage config: %w", err) + } + message := fmt.Sprintf("存储配置已更新，活动存储保持为 %s", target.Driver) + task.AppendLog(ctx, "%s", message) + return &task.TaskResult{Message: message}, nil + } + + total, err := countStorageObjects(ctx, active.Driver) + if err != nil { + return nil, fmt.Errorf("count source objects: %w", err) + } + if total == 0 { + if err := storage.SaveActiveConfig(ctx, target); err != nil { + return nil, fmt.Errorf("activate empty storage config: %w", err) + } + message := fmt.Sprintf("当前存储没有需要迁移的对象，活动存储已切换为 %s", target.Driver) + task.AppendLog(ctx, "%s", message) + return &task.TaskResult{Message: message}, nil + } + + sourceBackend, err := storage.NewBackend(ctx, active, active.Driver) + if err != nil { + return nil, fmt.Errorf("create source storage: %w", err) + } + targetBackend, err := storage.NewBackend(ctx, target, target.Driver) + if err != nil { + return nil, fmt.Errorf("create target storage: %w", err) + } + + task.AppendLog(ctx, "开始存储迁移: %s -> %s，总对象数: %d", active.Driver, target.Driver, total) + migrated, err := migrateObjects(ctx, sourceBackend, targetBackend, active.Driver, target.Driver, total) + if err != nil { + return nil, err + } + + if err := storage.SaveActiveConfig(ctx, target); err != nil { + return nil, fmt.Errorf("activate target storage: %w", err) + } + message := fmt.Sprintf("存储迁移完成，共迁移 %d 个对象，活动存储已切换为 %s", migrated, target.Driver) + task.AppendLog(ctx, "%s", message) + return &task.TaskResult{Message: message}, nil +} + +func normalizeStorageMigrationPayload(ctx context.Context, payload []byte) ([]byte, storage.Config, error) { + target, err := parseMigrationTargetConfig(ctx, payload) + if err != nil { + return nil, storage.Config{}, err + } + normalized, err := json.Marshal(storageMigrationPayload{Target: target}) + if err != nil { + return nil, storage.Config{}, fmt.Errorf("marshal storage migration payload: %w", err) + } + return normalized, target, nil +} + +func parseMigrationTargetConfig(ctx context.Context, payload []byte) (storage.Config, error) { + if strings.TrimSpace(string(payload)) == "" { + return storage.Config{}, errors.New("storage migration target payload is required") + } + var req storageMigrationPayload + if err := json.Unmarshal(payload, &req); err != nil { + return storage.Config{}, fmt.Errorf("parse storage migration payload: %w", err) + } + current, err := storage.LoadConfig(ctx) + if err != nil { + return storage.Config{}, fmt.Errorf("load active storage config: %w", err) + } + target := storage.MergeMaskedSecrets(req.Target, current) + if err := storage.ValidateConfig(target); err != nil { + return storage.Config{}, fmt.Errorf("validate target storage config: %w", err) + } + return target, nil +} + +func countStorageObjects(ctx context.Context, driver storage.Driver) (int64, error) { + var count int64 + err := db.DB(ctx).Model(&model.Upload{}). + Where("storage_driver = ? AND status != ?", driver, model.UploadStatusDeleted). + Distinct("file_path"). + Count(&count).Error + return count, err +} + +func hasUnresolvedMigrationTask(ctx context.Context) (bool, error) { + execution, ok, err := latestStorageMigrationExecution(ctx) + if err != nil || !ok { + return false, err + } + return execution.Status == model.TaskExecutionStatusPending || execution.Status == model.TaskExecutionStatusRunning, nil +} + +func latestStorageMigrationExecution(ctx context.Context) (*model.TaskExecution, bool, error) { + var execution model.TaskExecution + err := db.DB(ctx). + Where("task_type = ?", StorageMigrationTask). + Order("id DESC"). + First(&execution).Error + if err == nil { + return &execution, true, nil + } + if !errors.Is(err, gorm.ErrRecordNotFound) { + return nil, false, err + } + return nil, false, nil +} + +func migrateObjects( + ctx context.Context, + sourceBackend storage.Backend, + targetBackend storage.Backend, + sourceDriver storage.Driver, + targetDriver storage.Driver, + total int64, +) (int64, error) { + const batchSize = 50 + var migrated int64 + for { + if err := ctx.Err(); err != nil { + return migrated, fmt.Errorf("storage migration canceled: %w", err) + } + + var objects []struct { + FilePath string `gorm:"column:file_path"` + FileSize int64 `gorm:"column:file_size"` + MimeType string `gorm:"column:mime_type"` + } + if err := db.DB(ctx).Model(&model.Upload{}). + Select("file_path, MAX(file_size) AS file_size, MAX(mime_type) AS mime_type"). + Where("storage_driver = ? AND status != ?", sourceDriver, model.UploadStatusDeleted). + Group("file_path"). + Order("file_path ASC"). + Limit(batchSize). + Scan(&objects).Error; err != nil { + return migrated, fmt.Errorf("query source objects: %w", err) + } + if len(objects) == 0 { + break + } + + for _, object := range objects { + source, err := sourceBackend.Get(ctx, object.FilePath) + if err != nil { + if isNotFoundError(err) { + task.AppendLog(ctx, "警告: 源存储中物理文件不存在，标记为已删除并跳过: %s (错误: %v)", object.FilePath, err) + if updateErr := db.DB(ctx).Model(&model.Upload{}). + Where("storage_driver = ? AND file_path = ?", sourceDriver, object.FilePath). + Updates(map[string]any{ + "status": model.UploadStatusDeleted, + "storage_driver": targetDriver, + }).Error; updateErr != nil { + return migrated, fmt.Errorf("update missing object %q: %w", object.FilePath, updateErr) + } + continue + } + return migrated, fmt.Errorf("open source object %q: %w", object.FilePath, err) + } + targetPath, putErr := targetBackend.Put(ctx, object.FilePath, source.Body, object.FileSize, object.MimeType) + closeErr := source.Body.Close() + if putErr != nil { + return migrated, fmt.Errorf("copy object %q: %w", object.FilePath, putErr) + } + if closeErr != nil { + return migrated, fmt.Errorf("close source object %q: %w", object.FilePath, closeErr) + } + if err := db.DB(ctx).Model(&model.Upload{}). + Where("storage_driver = ? AND file_path = ?", sourceDriver, object.FilePath). + Updates(map[string]any{ + "storage_driver": targetDriver, + "file_path": targetPath, + }).Error; err != nil { + return migrated, fmt.Errorf("update migrated object %q: %w", object.FilePath, err) + } + migrated++ + } + task.AppendLog(ctx, "迁移进度: %d/%d", migrated, total) + } + return migrated, nil +} + +func isNotFoundError(err error) bool { + if err == nil { + return false + } + if errors.Is(err, os.ErrNotExist) { + return true + } + errStr := strings.ToLower(err.Error()) + for _, sub := range []string{"not found", "nosuchkey", "nosuchbucket", "404", "does not exist"} { + if strings.Contains(errStr, sub) { + return true + } + } + return false +} diff --git a/internal/apps/upload/storage_migration_task_test.go b/internal/apps/upload/storage_migration_task_test.go new file mode 100644 index 00000000..53a9953e --- /dev/null +++ b/internal/apps/upload/storage_migration_task_test.go @@ -0,0 +1,110 @@ +// Copyright 2026 Arctel.net +// SPDX-License-Identifier: Apache-2.0 + +package upload + +import ( + "bytes" + "context" + "encoding/json" + "io" + "os" + "path/filepath" + "testing" + + "github.com/Rain-kl/Wavelet/internal/model" + "github.com/Rain-kl/Wavelet/internal/storage" + "github.com/Rain-kl/Wavelet/internal/testhelper" +) + +func TestMigrationHandlerExecute(t *testing.T) { + dbConn, _, cleanup := testhelper.SetupTestEnvironment(t) + defer cleanup() + + sourceRoot := t.TempDir() + sourcePath := filepath.Join(sourceRoot, "uploads", "test.txt") + if err := os.MkdirAll(filepath.Dir(sourcePath), 0755); err != nil { + t.Fatalf("MkdirAll(%q) returned error: %v", sourcePath, err) + } + const content = "storage migration" + if err := os.WriteFile(sourcePath, []byte(content), 0644); err != nil { + t.Fatalf("WriteFile(%q) returned error: %v", sourcePath, err) + } + + ctx := context.Background() + active := storage.DefaultConfig() + active.Local.Root = sourceRoot + if err := storage.SaveActiveConfig(ctx, active); err != nil { + t.Fatalf("SaveActiveConfig() returned error: %v", err) + } + target := storage.DefaultConfig() + target.Driver = storage.DriverS3 + target.S3 = storage.ObjectConfig{ + Region: "us-east-1", + Bucket: "target", + AccessKeyID: "key", + SecretAccessKey: "secret", + } + payload, err := json.Marshal(storageMigrationPayload{Target: target}) + if err != nil { + t.Fatalf("Marshal(storageMigrationPayload) returned error: %v", err) + } + + upload := model.Upload{ + ID: 99101, + UserID: 1, + FileName: "test.txt", + FilePath: "uploads/test.txt", + FileSize: int64(len(content)), + MimeType: "text/plain", + Extension: "txt", + Hash: "hash", + StorageDriver: string(storage.DriverLocal), + Type: "attachment", + Status: model.UploadStatusUsed, + } + if err := dbConn.Create(&upload).Error; err != nil { + t.Fatalf("Create(upload) returned error: %v", err) + } + + var copied bytes.Buffer + restore := storage.MockStorage( + func(_ context.Context, _ string, body io.Reader, _ int64, _ string) error { + _, err := io.Copy(&copied, body) + return err + }, + func(context.Context, string) (*storage.Object, error) { + return nil, nil + }, + func(context.Context, string) error { + return nil + }, + ) + defer restore() + + result, err := (&MigrationHandler{}).Execute(ctx, payload) + if err != nil { + t.Fatalf("Execute() returned error: %v", err) + } + if result == nil { + t.Fatal("Execute() result = nil, want non-nil") + } + if copied.String() != content { + t.Errorf("migrated content = %q, want %q", copied.String(), content) + } + + var migrated model.Upload + if err := dbConn.First(&migrated, upload.ID).Error; err != nil { + t.Fatalf("First(upload) returned error: %v", err) + } + if migrated.StorageDriver != string(storage.DriverS3) { + t.Errorf("StorageDriver = %q, want %q", migrated.StorageDriver, storage.DriverS3) + } + current, err := storage.LoadConfig(ctx) + if err != nil { + t.Fatalf("LoadConfig() returned error: %v", err) + } + if current.Driver != storage.DriverS3 { + t.Errorf("active driver = %q, want %q", current.Driver, storage.DriverS3) + } +} diff --git a/internal/apps/upload/storage_ops.go b/internal/apps/upload/storage_ops.go new file mode 100644 index 00000000..ec01d2a5 --- /dev/null +++ b/internal/apps/upload/storage_ops.go @@ -0,0 +1,87 @@ +// Copyright 2026 Arctel.net +// SPDX-License-Identifier: Apache-2.0 + +package upload + +import ( + "context" + "fmt" + + "github.com/Rain-kl/Wavelet/internal/logger" + "github.com/Rain-kl/Wavelet/internal/model" + "github.com/Rain-kl/Wavelet/internal/storage" +) + +func storageReadOnly(ctx context.Context) bool { + execution, ok, err := latestStorageMigrationExecution(ctx) + if err != nil { + logger.ErrorF(ctx, "读取存储维护状态失败: %v", err) + return true + } + if !ok { + return false + } + return execution.Status != model.TaskExecutionStatusSucceeded +} + +func openStoredObject(ctx context.Context, upload *model.Upload) (*storage.Object, error) { + driver := storage.Driver(upload.StorageDriver) + if driver == "" { + driver = storage.DriverLocal + } + backend, err := backendForStoredDriver(ctx, driver) + if err != nil { + return nil, err + } + return backend.Get(ctx, upload.FilePath) +} + +func backendForStoredDriver(ctx context.Context, driver storage.Driver) (storage.Backend, error) { + backend, err := storage.ForDriver(ctx, driver) + if err == nil { + return backend, nil + } + + target, ok, targetErr := currentMigrationTargetConfig(ctx) + if targetErr != nil { + return nil, targetErr + } + if ok && target.Driver == driver { + return storage.NewBackend(ctx, target, driver) + } + return nil, fmt.Errorf("storage configuration for driver %q is unavailable", driver) +} + +func currentMigrationTargetConfig(ctx context.Context) (storage.Config, bool, error) { + execution, ok, err := latestStorageMigrationExecution(ctx) + if err != nil || !ok { + return storage.Config{}, false, err + } + if execution.Status == model.TaskExecutionStatusSucceeded { + return storage.Config{}, false, nil + } + target, err := parseMigrationTargetConfig(ctx, []byte(execution.Payload)) + if err != nil { + return storage.Config{}, false, err + } + return target, true, nil +} + +func cfgBucket(ctx context.Context, driver storage.Driver) string { + cfg, err := storage.LoadConfig(ctx) + if err != nil { + return "" + } + switch driver { + case storage.DriverS3: + return cfg.S3.Bucket + case storage.DriverR2: + return cfg.R2.Bucket + case storage.DriverMinIO: + return cfg.MinIO.Bucket + case storage.DriverOSS: + return cfg.OSS.Bucket + default: + return "" + } +} diff --git a/internal/apps/upload/tasks.go b/internal/apps/upload/tasks.go index 3d5e7dd2..73fca16e 100644 --- a/internal/apps/upload/tasks.go +++ b/internal/apps/upload/tasks.go @@ -81,6 +81,9 @@ type WarmImageCacheHandler struct{} // Execute 执行清理未使用上传文件的业务逻辑 func (h *CleanupUnusedUploadsHandler) Execute(ctx context.Context, _ []byte) (*task.TaskResult, error) { + if storageReadOnly(ctx) { + return nil, errors.New(ErrStorageReadOnly) + } const batchSize = 100 // 每批处理100个文件 var lastID uint64 var totalProcessed int @@ -122,8 +125,15 @@ func (h *CleanupUnusedUploadsHandler) Execute(ctx context.Context, _ []byte) (*t return err } - // Delete from S3 - if err := storage.DeleteObject(ctx, upload.FilePath); err != nil { + driver := storage.Driver(upload.StorageDriver) + if driver == "" { + driver = storage.DriverLocal + } + backend, err := storage.ForDriver(ctx, driver) + if err != nil { + return err + } + if err := backend.Delete(ctx, upload.FilePath); err != nil { return err } diff --git a/internal/apps/upload/tasks_test.go b/internal/apps/upload/tasks_test.go index 71cf3555..600d65ae 100644 --- a/internal/apps/upload/tasks_test.go +++ b/internal/apps/upload/tasks_test.go @@ -36,7 +36,7 @@ func TestCleanupUnusedUploadsHandler_Execute(t *testing.T) { func(ctx context.Context, key string, body io.Reader, size int64, contentType string) error { return nil }, - func(ctx context.Context, key string) (*storage.ObjectInfo, error) { return nil, nil }, + func(ctx context.Context, key string) (*storage.Object, error) { return nil, nil }, func(ctx context.Context, key string) error { return nil }, ) defer storageMock() @@ -113,7 +113,7 @@ func TestCleanupUnusedUploadsHandler_ExecuteNoFiles(t *testing.T) { func(ctx context.Context, key string, body io.Reader, size int64, contentType string) error { return nil }, - func(ctx context.Context, key string) (*storage.ObjectInfo, error) { return nil, nil }, + func(ctx context.Context, key string) (*storage.Object, error) { return nil, nil }, func(ctx context.Context, key string) error { return nil }, ) defer storageMock() @@ -204,6 +204,13 @@ func TestWarmImageCacheHandlerExecute(t *testing.T) { }) testDir := t.TempDir() + ctx := context.Background() + active := storage.DefaultConfig() + active.Local.Root = testDir + if err := storage.SaveActiveConfig(ctx, active); err != nil { + t.Fatalf("SaveActiveConfig() returned error: %v", err) + } + firstPath := filepath.Join(testDir, "first.png") secondPath := filepath.Join(testDir, "second.jpg") writeTaskTestPNG(t, firstPath, color.RGBA{R: 255, A: 255}) diff --git a/internal/config/config.go b/internal/config/config.go index 14bdc10e..a399d514 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -224,16 +224,6 @@ func applyEnvOverrides(c *configModel) { // ─── OTel ─── c.Otel.SamplingRate = envFloat64("OTEL_SAMPLING_RATE", c.Otel.SamplingRate) - // ─── S3 ─── - c.S3.Endpoint = envStr("S3_ENDPOINT", c.S3.Endpoint) - c.S3.Region = envStr("S3_REGION", c.S3.Region) - c.S3.Bucket = envStr("S3_BUCKET", c.S3.Bucket) - c.S3.AccessKeyID = envStr("S3_ACCESS_KEY_ID", c.S3.AccessKeyID) - c.S3.SecretAccessKey = envStr("S3_SECRET_ACCESS_KEY", c.S3.SecretAccessKey) - c.S3.CdnURL = envStr("S3_CDN_URL", c.S3.CdnURL) - c.S3.PathStyle = envBool("S3_PATH_STYLE", c.S3.PathStyle) - c.S3.Enabled = envBool("S3_ENABLED", c.S3.Enabled) - // ─── Worker ─── c.Worker.Concurrency = envInt("WORKER_CONCURRENCY", c.Worker.Concurrency) c.Worker.StrictPriority = envBool("WORKER_STRICT_PRIORITY", c.Worker.StrictPriority) diff --git a/internal/config/model.go b/internal/config/model.go index ca89db3b..ba44553b 100644 --- a/internal/config/model.go +++ b/internal/config/model.go @@ -15,7 +15,6 @@ type configModel struct { Worker workerConfig `mapstructure:"worker"` ClickHouse clickHouseConfig `mapstructure:"clickhouse"` Otel otelConfig `mapstructure:"otel"` - S3 s3Config `mapstructure:"s3"` } // appConfig 应用基本配置 @@ -139,22 +138,3 @@ type QueueConfig struct { type otelConfig struct { SamplingRate float64 `mapstructure:"sampling_rate"` } - -// s3Config S3 compatible storage configuration -type s3Config struct { - Enabled bool `mapstructure:"enabled"` - Endpoint string `mapstructure:"endpoint"` - Region string `mapstructure:"region"` - Bucket string `mapstructure:"bucket"` - AccessKeyID string `mapstructure:"access_key_id" json:"-"` - SecretAccessKey string `mapstructure:"secret_access_key" json:"-"` - PathStyle bool `mapstructure:"path_style"` - KeyPrefix string `mapstructure:"key_prefix"` - CdnURL string `mapstructure:"cdn_url"` - LocalCache localCacheConfig `mapstructure:"local_cache"` -} - -type localCacheConfig struct { - Enabled bool `mapstructure:"enabled"` - CacheDir string `mapstructure:"cache_dir"` -} diff --git a/internal/db/migrator/goose/postgres/202606090001_initial_schema.sql b/internal/db/migrator/goose/postgres/202606090001_initial_schema.sql index 95487010..5628860d 100644 --- a/internal/db/migrator/goose/postgres/202606090001_initial_schema.sql +++ b/internal/db/migrator/goose/postgres/202606090001_initial_schema.sql @@ -54,7 +54,7 @@ CREATE UNIQUE INDEX IF NOT EXISTS idx_external_accounts_source_external ON exter CREATE TABLE IF NOT EXISTS system_configs ( key VARCHAR(64) PRIMARY KEY, - value VARCHAR(255) NOT NULL, + value TEXT NOT NULL, type VARCHAR(32) NOT NULL DEFAULT 'system', visibility INTEGER NOT NULL DEFAULT 0, description VARCHAR(255), @@ -160,7 +160,8 @@ INSERT INTO system_configs (key, value, type, visibility, description, created_a ('email_register_verification_enabled', 'false', 'system', 1, '是否开启邮箱注册验证（true/false）', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP), ('menu_display_config', '{}', 'system', 1, '目录显示配置（JSON 字符串，格式为 {url: enabled}）', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP), ('search_engine_indexing_enabled', 'false', 'system', 1, '是否允许搜索引擎爬取/检索该站点（true/false）', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP), - ('update_upstream_repository', 'Rain-kl/Wavelet', 'system', 0, 'GitHub Actions Release 上游仓库（owner/repo 或 GitHub 仓库地址）', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) + ('update_upstream_repository', 'Rain-kl/Wavelet', 'system', 0, 'GitHub Actions Release 上游仓库（owner/repo 或 GitHub 仓库地址）', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP), + ('storage_config', '{"driver":"local","local":{"root":"."},"s3":{"region":"us-east-1"},"r2":{"region":"auto"},"minio":{"region":"us-east-1","path_style":true},"oss":{},"webdav":{}}', 'system', 0, '文件存储驱动及连接配置（JSON）', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) ON CONFLICT (key) DO NOTHING; INSERT INTO users (id, username, password, nickname, avatar_url, is_active, is_admin, last_login_at, created_at, updated_at) diff --git a/internal/db/migrator/goose/postgres/202606130002_expand_system_config_value.sql b/internal/db/migrator/goose/postgres/202606130002_expand_system_config_value.sql new file mode 100644 index 00000000..0d806e27 --- /dev/null +++ b/internal/db/migrator/goose/postgres/202606130002_expand_system_config_value.sql @@ -0,0 +1,5 @@ +-- +goose Up +ALTER TABLE w_system_configs ALTER COLUMN value TYPE TEXT; + +-- +goose Down +ALTER TABLE w_system_configs ALTER COLUMN value TYPE VARCHAR(255); diff --git a/internal/db/migrator/goose/postgres/202606130003_add_storage_config.sql b/internal/db/migrator/goose/postgres/202606130003_add_storage_config.sql new file mode 100644 index 00000000..4d49aa64 --- /dev/null +++ b/internal/db/migrator/goose/postgres/202606130003_add_storage_config.sql @@ -0,0 +1,15 @@ +-- +goose Up +INSERT INTO w_system_configs (key, value, type, visibility, description, created_at, updated_at) +VALUES ( + 'storage_config', + '{"driver":"local","local":{"root":"."},"s3":{"region":"us-east-1"},"r2":{"region":"auto"},"minio":{"region":"us-east-1","path_style":true},"oss":{},"webdav":{}}', + 'system', + 0, + '文件存储驱动及连接配置（JSON）', + CURRENT_TIMESTAMP, + CURRENT_TIMESTAMP +) +ON CONFLICT (key) DO NOTHING; + +-- +goose Down +DELETE FROM w_system_configs WHERE key = 'storage_config'; diff --git a/internal/db/migrator/goose/sqlite/202606090001_initial_schema.sql b/internal/db/migrator/goose/sqlite/202606090001_initial_schema.sql index 82defab4..c54b5caa 100644 --- a/internal/db/migrator/goose/sqlite/202606090001_initial_schema.sql +++ b/internal/db/migrator/goose/sqlite/202606090001_initial_schema.sql @@ -54,7 +54,7 @@ CREATE UNIQUE INDEX IF NOT EXISTS idx_external_accounts_source_external ON exter CREATE TABLE IF NOT EXISTS system_configs ( key VARCHAR(64) PRIMARY KEY, - value VARCHAR(255) NOT NULL, + value TEXT NOT NULL, type VARCHAR(32) NOT NULL DEFAULT 'system', visibility INTEGER NOT NULL DEFAULT 0, description VARCHAR(255), @@ -160,7 +160,8 @@ INSERT INTO system_configs (key, value, type, visibility, description, created_a ('email_register_verification_enabled', 'false', 'system', 1, '是否开启邮箱注册验证（true/false）', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP), ('menu_display_config', '{}', 'system', 1, '目录显示配置（JSON 字符串，格式为 {url: enabled}）', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP), ('search_engine_indexing_enabled', 'false', 'system', 1, '是否允许搜索引擎爬取/检索该站点（true/false）', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP), - ('update_upstream_repository', 'Rain-kl/Wavelet', 'system', 0, 'GitHub Actions Release 上游仓库（owner/repo 或 GitHub 仓库地址）', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) + ('update_upstream_repository', 'Rain-kl/Wavelet', 'system', 0, 'GitHub Actions Release 上游仓库（owner/repo 或 GitHub 仓库地址）', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP), + ('storage_config', '{"driver":"local","local":{"root":"."},"s3":{"region":"us-east-1"},"r2":{"region":"auto"},"minio":{"region":"us-east-1","path_style":true},"oss":{},"webdav":{}}', 'system', 0, '文件存储驱动及连接配置（JSON）', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) ON CONFLICT (key) DO NOTHING; INSERT INTO users (id, username, password, nickname, avatar_url, is_active, is_admin, last_login_at, created_at, updated_at) diff --git a/internal/db/migrator/goose/sqlite/202606130002_expand_system_config_value.sql b/internal/db/migrator/goose/sqlite/202606130002_expand_system_config_value.sql new file mode 100644 index 00000000..ac50e2a4 --- /dev/null +++ b/internal/db/migrator/goose/sqlite/202606130002_expand_system_config_value.sql @@ -0,0 +1,6 @@ +-- +goose Up +-- SQLite stores VARCHAR and TEXT with the same TEXT affinity. +SELECT 1; + +-- +goose Down +SELECT 1; diff --git a/internal/db/migrator/goose/sqlite/202606130003_add_storage_config.sql b/internal/db/migrator/goose/sqlite/202606130003_add_storage_config.sql new file mode 100644 index 00000000..4d49aa64 --- /dev/null +++ b/internal/db/migrator/goose/sqlite/202606130003_add_storage_config.sql @@ -0,0 +1,15 @@ +-- +goose Up +INSERT INTO w_system_configs (key, value, type, visibility, description, created_at, updated_at) +VALUES ( + 'storage_config', + '{"driver":"local","local":{"root":"."},"s3":{"region":"us-east-1"},"r2":{"region":"auto"},"minio":{"region":"us-east-1","path_style":true},"oss":{},"webdav":{}}', + 'system', + 0, + '文件存储驱动及连接配置（JSON）', + CURRENT_TIMESTAMP, + CURRENT_TIMESTAMP +) +ON CONFLICT (key) DO NOTHING; + +-- +goose Down +DELETE FROM w_system_configs WHERE key = 'storage_config'; diff --git a/internal/db/migrator/migrator_test.go b/internal/db/migrator/migrator_test.go index 0b577ecb..a355e2ec 100644 --- a/internal/db/migrator/migrator_test.go +++ b/internal/db/migrator/migrator_test.go @@ -17,7 +17,7 @@ import ( "gorm.io/gorm" ) -const expectedMigratedSystemConfigCount = 29 +const expectedMigratedSystemConfigCount = 30 func TestMigrateInitializesSQLiteDatabase(t *testing.T) { sqliteDB, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{ diff --git a/internal/model/system_configs.go b/internal/model/system_configs.go index 4fd9e37e..66de8e84 100644 --- a/internal/model/system_configs.go +++ b/internal/model/system_configs.go @@ -1,4 +1,3 @@ -// Copyright 2025 linux.do // Copyright 2026 Arctel.net // SPDX-License-Identifier: Apache-2.0 @@ -49,6 +48,7 @@ const ( ConfigKeyDiskCacheLRUEnabled = "disk_cache_lru_enabled" // 是否启用 LRU 淘汰机制 ConfigKeyLoginSessionTTLHours = "login_session_ttl_hours" // 登录会话过期时间 (小时，0表示浏览器关闭后自动退出登录，-1表示永不过期) ConfigKeyUpdateUpstreamRepository = "update_upstream_repository" // GitHub Actions Release 上游仓库 + ConfigKeyStorageConfig = "storage_config" // 文件存储配置 (JSON) ) const ( @@ -66,7 +66,7 @@ const ( // SystemConfig 系统配置实体 type SystemConfig struct { Key string `json:"key" gorm:"primaryKey;size:64;not null"` - Value string `json:"value" gorm:"size:255;not null"` + Value string `json:"value" gorm:"type:text;not null"` Type string `json:"type" gorm:"size:32;not null;default:'system'"` Visibility int `json:"visibility" gorm:"not null;default:0"` Description string `json:"description" gorm:"size:255"` diff --git a/internal/storage/cache.go b/internal/storage/cache.go deleted file mode 100644 index 07838331..00000000 --- a/internal/storage/cache.go +++ /dev/null @@ -1,205 +0,0 @@ -// Copyright 2025 linux.do -// Copyright 2026 Arctel.net -// SPDX-License-Identifier: Apache-2.0 - -package storage - -import ( - "bytes" - "context" - "encoding/json" - "fmt" - "io" - "log" - "os" - "path/filepath" - "strings" - - "github.com/Rain-kl/Wavelet/internal/config" - "github.com/Rain-kl/Wavelet/internal/logger" - "github.com/Rain-kl/Wavelet/internal/otel_trace" - "go.opentelemetry.io/otel/codes" - "go.opentelemetry.io/otel/trace" - "golang.org/x/sync/singleflight" -) - -var localCacheEnabled = false -var localCacheDir = "" -var cacheFilePath = "%s/%s" -var cacheMetaFilePath = "%s/%s.meta" -var group singleflight.Group - -type metaInfo struct { - ContentType string `json:"content_type"` - ContentLength int64 `json:"content_length"` -} - -// cacheDirPerm 缓存目录权限 -const cacheDirPerm = 0755 - -func init() { - cfg := config.Config.S3.LocalCache - localCacheEnabled = cfg.Enabled && cfg.CacheDir != "" - localCacheDir = strings.TrimSuffix(cfg.CacheDir, "/") - if localCacheEnabled { - if err := os.MkdirAll(cfg.CacheDir, cacheDirPerm); err != nil { - log.Fatalf("[Storage] failed to create local cache directory: %v\n", err) - } - } -} - -// GetObjectViaCache 通过本地缓存获取对象，缓存未命中时从 S3/CDN 拉取 -func GetObjectViaCache(ctx context.Context, key string) (*ObjectInfo, error) { - // 没有开启本地缓存 - if !localCacheEnabled { - return GetObjectViaProxy(ctx, key) - } - - // 初始化 Trace - ctx, span := otel_trace.Start(ctx, "S3.GetObjectViaCache", trace.WithSpanKind(trace.SpanKindClient)) - defer span.End() - - // 检查本地缓存 - key = strings.TrimPrefix(key, "/") - localPath := fmt.Sprintf(cacheFilePath, localCacheDir, key) - metaPath := fmt.Sprintf(cacheMetaFilePath, localCacheDir, key) - objInfo, err := getLocalCacheFile(ctx, localPath, metaPath) - if err != nil { - return nil, err - } - if objInfo != nil { - return objInfo, nil - } - - // 使用 singleflight 确保同一时间只有一个请求会触发 CDN 获取和本地缓存保存 - _, err, _ = group.Do(key, func() (interface{}, error) { - ctx := context.WithoutCancel(ctx) - - // 没有缓存，通过 CDN 获取 - objInfo, err := GetObjectViaProxy(ctx, key) - if err != nil { - return nil, err - } - - // 保存到本地 - if err := saveToLocalCache(ctx, localPath, metaPath, objInfo); err != nil { - return nil, err - } - - return nil, nil - }) - if err != nil { - logger.ErrorF(ctx, "Failed to get object via singleflight for key %s: %v", key, err) - return nil, LocalCacheError{} - } - - return GetObjectViaCache(ctx, key) -} - -func getLocalCacheFile(ctx context.Context, localPath, metaPath string) (*ObjectInfo, error) { - _, span := otel_trace.Start(ctx, "S3.GetLocalCacheFile", trace.WithSpanKind(trace.SpanKindClient)) - defer span.End() - - // 尝试打开本地缓存文件 - file, err := os.Open(localPath) //nolint:gosec // localPath is internally managed cache path - if err == nil { - defer func() { _ = file.Close() }() - } - - // 文件不存在 - if err != nil && os.IsNotExist(err) { - return nil, nil - } - - // 判断是否为其他异常 - if err != nil { - span.SetStatus(codes.Error, err.Error()) - return nil, err - } - - // 读取元信息 - metaData, err := os.ReadFile(metaPath) //nolint:gosec // metaPath is internally managed cache path - - // 文件不存在 - if err != nil && os.IsNotExist(err) { - return nil, nil - } - - // 判断是否为其他异常 - if err != nil { - span.SetStatus(codes.Error, err.Error()) - return nil, err - } - - // 解析元信息 - meta := &metaInfo{} - if err := json.Unmarshal(metaData, meta); err != nil { - span.SetStatus(codes.Error, err.Error()) - return nil, err - } - - return &ObjectInfo{CachePath: localPath, ContentLength: meta.ContentLength, ContentType: meta.ContentType}, nil -} - -func saveToLocalCache(ctx context.Context, localPath, metaPath string, objInfo *ObjectInfo) error { - _, span := otel_trace.Start(ctx, "S3.SaveToLocalCache", trace.WithSpanKind(trace.SpanKindClient)) - defer span.End() - - // 创建目录 - localDir := filepath.Dir(localPath) - if err := os.MkdirAll(localDir, cacheDirPerm); err != nil { - span.SetStatus(codes.Error, err.Error()) - return err - } - - // 创建文件 - if err := saveFile(localPath, objInfo.Body); err != nil { - span.SetStatus(codes.Error, err.Error()) - return err - } - - // 创建元信息文件 - meta := &metaInfo{ContentType: objInfo.ContentType, ContentLength: objInfo.ContentLength} - metaData, err := json.Marshal(meta) - if err != nil { - span.SetStatus(codes.Error, err.Error()) - return err - } - if err := saveFile(metaPath, bytes.NewReader(metaData)); err != nil { - span.SetStatus(codes.Error, err.Error()) - return err - } - - return nil -} - -func saveFile(localPath string, data io.Reader) error { - // 创建临时文件 - tempFile, err := os.CreateTemp(filepath.Dir(localPath), "cache_temp_*") - if err != nil { - return err - } - defer func() { _ = os.Remove(tempFile.Name()) }() - - // 将内容写入临时文件 - if _, err := tempFile.ReadFrom(data); err != nil { - return err - } - - // 确保数据写入磁盘 - if err := tempFile.Sync(); err != nil { - return err - } - - // 关闭临时文件 - if err := tempFile.Close(); err != nil { - return err - } - - // 将临时文件重命名为最终文件 - if err := os.Rename(tempFile.Name(), localPath); err != nil { - return err - } - - return nil -} diff --git a/internal/storage/config.go b/internal/storage/config.go new file mode 100644 index 00000000..eca85327 --- /dev/null +++ b/internal/storage/config.go @@ -0,0 +1,234 @@ +// Copyright 2026 Arctel.net +// SPDX-License-Identifier: Apache-2.0 + +// Package storage provides dynamically configured file storage backends. +package storage + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + + "github.com/Rain-kl/Wavelet/internal/db" + "github.com/Rain-kl/Wavelet/internal/model" + "gorm.io/gorm" +) + +// Driver identifies a supported storage backend. +type Driver string + +const ( + // DriverLocal stores files on the local filesystem. + DriverLocal Driver = "local" + // DriverS3 stores files in an S3-compatible object store. + DriverS3 Driver = "s3" + // DriverR2 stores files in Cloudflare R2. + DriverR2 Driver = "r2" + // DriverMinIO stores files in MinIO. + DriverMinIO Driver = "minio" + // DriverOSS stores files in Aliyun OSS. + DriverOSS Driver = "oss" + // DriverWebDAV stores files through WebDAV. + DriverWebDAV Driver = "webdav" + + // ConfigMask replaces secrets returned to the frontend. + ConfigMask = "******" +) + +// LocalConfig configures local filesystem storage. +type LocalConfig struct { + Root string `json:"root"` +} + +// ObjectConfig configures S3-compatible or OSS object storage. +type ObjectConfig struct { + Endpoint string `json:"endpoint"` + Region string `json:"region"` + Bucket string `json:"bucket"` + AccessKeyID string `json:"access_key_id"` + SecretAccessKey string `json:"secret_access_key"` + AccountID string `json:"account_id,omitempty"` + PathStyle bool `json:"path_style"` + KeyPrefix string `json:"key_prefix"` + CDNURL string `json:"cdn_url"` +} + +// WebDAVConfig configures WebDAV storage. +type WebDAVConfig struct { + Endpoint string `json:"endpoint"` + Username string `json:"username"` + Password string `json:"password"` + BasePath string `json:"base_path"` +} + +// Config contains all storage backends and the currently active driver. +type Config struct { + Driver Driver `json:"driver"` + Local LocalConfig `json:"local"` + S3 ObjectConfig `json:"s3"` + R2 ObjectConfig `json:"r2"` + MinIO ObjectConfig `json:"minio"` + OSS ObjectConfig `json:"oss"` + WebDAV WebDAVConfig `json:"webdav"` +} + +// DefaultConfig returns the local-storage default configuration. +func DefaultConfig() Config { + return Config{ + Driver: DriverLocal, + Local: LocalConfig{Root: "."}, + S3: ObjectConfig{Region: "us-east-1"}, + R2: ObjectConfig{Region: "auto"}, + MinIO: ObjectConfig{Region: "us-east-1", PathStyle: true}, + } +} + +// LoadConfig loads the active storage configuration. +func LoadConfig(ctx context.Context) (Config, error) { + return loadConfigByKey(ctx, model.ConfigKeyStorageConfig, DefaultConfig()) +} + +func loadConfigByKey(ctx context.Context, key string, fallback Config) (Config, error) { + var sc model.SystemConfig + if err := sc.GetByKey(ctx, key); err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + return fallback, nil + } + return Config{}, err + } + if strings.TrimSpace(sc.Value) == "" { + return fallback, nil + } + if err := json.Unmarshal([]byte(sc.Value), &fallback); err != nil { + return Config{}, fmt.Errorf("parse %s: %w", key, err) + } + return fallback, nil +} + +// ValidateConfig validates the selected backend configuration. +func ValidateConfig(cfg Config) error { + switch cfg.Driver { + case DriverLocal: + if strings.TrimSpace(cfg.Local.Root) == "" { + return errors.New("local root is required") + } + case DriverS3: + return validateObjectConfig(cfg.S3, false) + case DriverR2: + if strings.TrimSpace(cfg.R2.AccountID) == "" && strings.TrimSpace(cfg.R2.Endpoint) == "" { + return errors.New("R2 account ID or endpoint is required") + } + return validateObjectConfig(cfg.R2, false) + case DriverMinIO: + if strings.TrimSpace(cfg.MinIO.Endpoint) == "" { + return errors.New("MinIO endpoint is required") + } + return validateObjectConfig(cfg.MinIO, true) + case DriverOSS: + if strings.TrimSpace(cfg.OSS.Endpoint) == "" { + return errors.New("OSS endpoint is required") + } + return validateObjectConfig(cfg.OSS, true) + case DriverWebDAV: + if strings.TrimSpace(cfg.WebDAV.Endpoint) == "" { + return errors.New("WebDAV endpoint is required") + } + default: + return fmt.Errorf("unsupported storage driver %q", cfg.Driver) + } + return nil +} + +func validateObjectConfig(cfg ObjectConfig, endpointRequired bool) error { + if endpointRequired && strings.TrimSpace(cfg.Endpoint) == "" { + return errors.New("endpoint is required") + } + if strings.TrimSpace(cfg.Region) == "" { + return errors.New("region is required") + } + if strings.TrimSpace(cfg.Bucket) == "" { + return errors.New("bucket is required") + } + if strings.TrimSpace(cfg.AccessKeyID) == "" || strings.TrimSpace(cfg.SecretAccessKey) == "" { + return errors.New("access key ID and secret access key are required") + } + return nil +} + +// SaveActiveConfig persists the active storage configuration. +func SaveActiveConfig(ctx context.Context, cfg Config) error { + return saveSystemConfig(ctx, model.ConfigKeyStorageConfig, cfg, "文件存储驱动及连接配置（JSON）") +} + +func saveSystemConfig(ctx context.Context, key string, value any, description string) error { + return db.DB(ctx).Transaction(func(tx *gorm.DB) error { + return upsertSystemConfig(ctx, tx, key, value, description) + }) +} + +func upsertSystemConfig(ctx context.Context, tx *gorm.DB, key string, value any, description string) error { + data, err := json.Marshal(value) + if err != nil { + return fmt.Errorf("marshal %s: %w", key, err) + } + sc := model.SystemConfig{ + Key: key, + Value: string(data), + Type: "system", + Visibility: model.ConfigVisibilityHidden, + Description: description, + } + if err := tx.Where("key = ?", key). + Assign(map[string]any{"value": sc.Value, "description": description, "visibility": model.ConfigVisibilityHidden}). + FirstOrCreate(&sc).Error; err != nil { + return err + } + if db.Redis != nil { + if err := db.HSetJSON(ctx, model.SystemConfigRedisHashKey, key, &sc); err != nil { + return err + } + } + return nil +} + +// MergeMaskedSecrets restores unchanged secrets from the current configuration. +func MergeMaskedSecrets(next, current Config) Config { + mergeObjectSecret := func(dst *ObjectConfig, src ObjectConfig) { + if dst.AccessKeyID == ConfigMask { + dst.AccessKeyID = src.AccessKeyID + } + if dst.SecretAccessKey == ConfigMask { + dst.SecretAccessKey = src.SecretAccessKey + } + } + mergeObjectSecret(&next.S3, current.S3) + mergeObjectSecret(&next.R2, current.R2) + mergeObjectSecret(&next.MinIO, current.MinIO) + mergeObjectSecret(&next.OSS, current.OSS) + if next.WebDAV.Password == ConfigMask { + next.WebDAV.Password = current.WebDAV.Password + } + return next +} + +// MaskSecrets replaces stored credentials with placeholders for API responses. +func MaskSecrets(cfg Config) Config { + maskObject := func(value *ObjectConfig) { + if value.AccessKeyID != "" { + value.AccessKeyID = ConfigMask + } + if value.SecretAccessKey != "" { + value.SecretAccessKey = ConfigMask + } + } + maskObject(&cfg.S3) + maskObject(&cfg.R2) + maskObject(&cfg.MinIO) + maskObject(&cfg.OSS) + if cfg.WebDAV.Password != "" { + cfg.WebDAV.Password = ConfigMask + } + return cfg +} diff --git a/internal/storage/errs.go b/internal/storage/errs.go deleted file mode 100644 index 4a72d5d5..00000000 --- a/internal/storage/errs.go +++ /dev/null @@ -1,29 +0,0 @@ -// Copyright 2026 Arctel.net -// SPDX-License-Identifier: Apache-2.0 - -// Package storage 提供文件存储抽象层，包括 S3 兼容存储和本地缓存。 -package storage - -// ErrS3InitializationFailed S3 存储初始化失败错误 -type ErrS3InitializationFailed struct{} - -func (e ErrS3InitializationFailed) Error() string { - return errS3InitializationFailed -} - -// LocalCacheError 本地缓存错误 -type LocalCacheError struct{} - -func (e LocalCacheError) Error() string { - return errLocalCache -} - -const ( - errS3InitializationFailed = "S3存储初始化失败" - errLocalCache = "本地缓存错误" - errS3PutObjectFailed = "s3 put object failed: %w" - errS3GetObjectFailed = "s3 get object failed: %w" - errCDNRequestFailed = "cdn request failed: %w" - errCDNStatusFailed = "cdn returned status %d" - errS3DeleteObjectFailed = "s3 delete object failed: %w" -) diff --git a/internal/storage/http.go b/internal/storage/http.go new file mode 100644 index 00000000..add2f472 --- /dev/null +++ b/internal/storage/http.go @@ -0,0 +1,39 @@ +// Copyright 2026 Arctel.net +// SPDX-License-Identifier: Apache-2.0 + +package storage + +import ( + "context" + "fmt" + "net/http" + "net/url" +) + +func getHTTPObject(ctx context.Context, baseURL, key string) (*Object, error) { + objectURL, err := url.JoinPath(baseURL, key) + if err != nil { + return nil, fmt.Errorf("build CDN object URL: %w", err) + } + request, err := http.NewRequestWithContext(ctx, http.MethodGet, objectURL, nil) + if err != nil { + return nil, fmt.Errorf("create CDN request: %w", err) + } + response, err := http.DefaultClient.Do(request) + if err != nil { + return nil, fmt.Errorf("get CDN object: %w", err) + } + if response.StatusCode != http.StatusOK { + _ = response.Body.Close() + return nil, fmt.Errorf("get CDN object: unexpected status %d", response.StatusCode) + } + contentType := response.Header.Get("Content-Type") + if contentType == "" { + contentType = defaultContentType + } + return &Object{ + Body: response.Body, + ContentLength: response.ContentLength, + ContentType: contentType, + }, nil +} diff --git a/internal/storage/local.go b/internal/storage/local.go new file mode 100644 index 00000000..4b870cbb --- /dev/null +++ b/internal/storage/local.go @@ -0,0 +1,120 @@ +// Copyright 2026 Arctel.net +// SPDX-License-Identifier: Apache-2.0 + +package storage + +import ( + "context" + "fmt" + "io" + "mime" + "os" + "path/filepath" + "strings" +) + +type localBackend struct { + root string +} + +func newLocalBackend(cfg LocalConfig) (*localBackend, error) { + root := filepath.Clean(cfg.Root) + if root == "" { + return nil, fmt.Errorf("local root is required") + } + return &localBackend{root: root}, nil +} + +func (b *localBackend) Put(_ context.Context, key string, body io.Reader, _ int64, _ string) (string, error) { + path, err := b.path(key) + if err != nil { + return "", err + } + if err := os.MkdirAll(filepath.Dir(path), storageDirPerm); err != nil { + return "", err + } + file, err := os.OpenFile( //nolint:gosec // path is constrained to the configured storage root. + path, + os.O_CREATE|os.O_TRUNC|os.O_WRONLY, + storageFilePerm, + ) + if err != nil { + return "", err + } + if _, err := io.Copy(file, body); err != nil { + _ = file.Close() + _ = os.Remove(path) + return "", err + } + if err := file.Close(); err != nil { + _ = os.Remove(path) + return "", err + } + return filepath.ToSlash(key), nil +} + +func (b *localBackend) Get(_ context.Context, key string) (*Object, error) { + path, err := b.path(key) + if err != nil { + return nil, err + } + file, err := os.Open(path) //nolint:gosec // path is constrained to the configured storage root. + if err != nil { + return nil, err + } + info, err := file.Stat() + if err != nil { + _ = file.Close() + return nil, err + } + contentType := mime.TypeByExtension(filepath.Ext(path)) + if contentType == "" { + contentType = defaultContentType + } + return &Object{Body: file, ContentLength: info.Size(), ContentType: contentType}, nil +} + +func (b *localBackend) Delete(_ context.Context, key string) error { + path, err := b.path(key) + if err != nil { + return err + } + err = os.Remove(path) + if os.IsNotExist(err) { + return nil + } + return err +} + +func (b *localBackend) Test(_ context.Context) error { + return os.MkdirAll(b.root, storageDirPerm) +} + +func (b *localBackend) path(key string) (string, error) { + if filepath.IsAbs(key) { + cleanPath := filepath.Clean(key) + absRoot, err := filepath.Abs(b.root) + if err != nil { + return "", err + } + absPath, err := filepath.Abs(cleanPath) + if err != nil { + return "", err + } + rel, err := filepath.Rel(absRoot, absPath) + if err != nil || strings.HasPrefix(rel, "..") { + return "", fmt.Errorf("storage key escapes local root") + } + return cleanPath, nil + } + cleanKey := filepath.Clean(filepath.FromSlash(strings.TrimPrefix(key, "/"))) + if cleanKey == "." || cleanKey == "" || strings.HasPrefix(cleanKey, "..") { + return "", fmt.Errorf("invalid local storage key %q", key) + } + path := filepath.Join(b.root, cleanKey) + rel, err := filepath.Rel(b.root, path) + if err != nil || strings.HasPrefix(rel, "..") { + return "", fmt.Errorf("storage key escapes local root") + } + return path, nil +} diff --git a/internal/storage/local_test.go b/internal/storage/local_test.go new file mode 100644 index 00000000..8dccd07f --- /dev/null +++ b/internal/storage/local_test.go @@ -0,0 +1,51 @@ +// Copyright 2026 Arctel.net +// SPDX-License-Identifier: Apache-2.0 + +package storage + +import ( + "bytes" + "context" + "io" + "testing" +) + +func TestLocalBackendRoundTrip(t *testing.T) { + backend, err := newLocalBackend(LocalConfig{Root: t.TempDir()}) + if err != nil { + t.Fatalf("newLocalBackend() returned error: %v", err) + } + ctx := context.Background() + const key = "uploads/2026/06/13/test.txt" + const content = "wavelet storage" + + storedKey, err := backend.Put(ctx, key, bytes.NewBufferString(content), int64(len(content)), "text/plain") + if err != nil { + t.Fatalf("Put(%q) returned error: %v", key, err) + } + if storedKey != key { + t.Errorf("Put(%q) key = %q, want %q", key, storedKey, key) + } + + object, err := backend.Get(ctx, key) + if err != nil { + t.Fatalf("Get(%q) returned error: %v", key, err) + } + got, err := io.ReadAll(object.Body) + if err != nil { + t.Fatalf("ReadAll(Get(%q)) returned error: %v", key, err) + } + if err := object.Body.Close(); err != nil { + t.Fatalf("Close(Get(%q)) returned error: %v", key, err) + } + if string(got) != content { + t.Errorf("Get(%q) content = %q, want %q", key, got, content) + } + + if err := backend.Delete(ctx, key); err != nil { + t.Fatalf("Delete(%q) returned error: %v", key, err) + } + if _, err := backend.Get(ctx, key); err == nil { + t.Errorf("Get(%q) after Delete() returned nil error", key) + } +} diff --git a/internal/storage/oss.go b/internal/storage/oss.go new file mode 100644 index 00000000..866e8fdd --- /dev/null +++ b/internal/storage/oss.go @@ -0,0 +1,98 @@ +// Copyright 2026 Arctel.net +// SPDX-License-Identifier: Apache-2.0 + +package storage + +import ( + "context" + "fmt" + "io" + "strings" + + "github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss" + "github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/credentials" +) + +type ossBackend struct { + client *oss.Client + bucket string + keyPrefix string + cdnURL string +} + +func newOSSBackend(cfg ObjectConfig) (*ossBackend, error) { + options := oss.LoadDefaultConfig(). + WithCredentialsProvider(credentials.NewStaticCredentialsProvider(cfg.AccessKeyID, cfg.SecretAccessKey)). + WithRegion(cfg.Region) + if cfg.Endpoint != "" { + options.WithEndpoint(cfg.Endpoint) + } + return &ossBackend{ + client: oss.NewClient(options), + bucket: cfg.Bucket, + keyPrefix: strings.Trim(cfg.KeyPrefix, "/"), + cdnURL: strings.TrimRight(cfg.CDNURL, "/"), + }, nil +} + +func (b *ossBackend) Put(ctx context.Context, key string, body io.Reader, _ int64, _ string) (string, error) { + key = b.key(key) + _, err := b.client.PutObject(ctx, &oss.PutObjectRequest{ + Bucket: oss.Ptr(b.bucket), + Key: oss.Ptr(key), + Body: body, + }) + if err != nil { + return "", fmt.Errorf("put OSS object: %w", err) + } + return key, nil +} + +func (b *ossBackend) Get(ctx context.Context, key string) (*Object, error) { + key = b.key(key) + if b.cdnURL != "" { + return getHTTPObject(ctx, b.cdnURL, key) + } + output, err := b.client.GetObject(ctx, &oss.GetObjectRequest{ + Bucket: oss.Ptr(b.bucket), + Key: oss.Ptr(key), + }) + if err != nil { + return nil, fmt.Errorf("get OSS object: %w", err) + } + contentType := defaultContentType + if output.ContentType != nil { + contentType = *output.ContentType + } + return &Object{Body: output.Body, ContentLength: output.ContentLength, ContentType: contentType}, nil +} + +func (b *ossBackend) Delete(ctx context.Context, key string) error { + _, err := b.client.DeleteObject(ctx, &oss.DeleteObjectRequest{ + Bucket: oss.Ptr(b.bucket), + Key: oss.Ptr(b.key(key)), + }) + if err != nil { + return fmt.Errorf("delete OSS object: %w", err) + } + return nil +} + +func (b *ossBackend) Test(ctx context.Context) error { + ok, err := b.client.IsBucketExist(ctx, b.bucket) + if err != nil { + return fmt.Errorf("access OSS bucket: %w", err) + } + if !ok { + return fmt.Errorf("OSS bucket %q does not exist", b.bucket) + } + return nil +} + +func (b *ossBackend) key(key string) string { + key = strings.TrimLeft(key, "/") + if b.keyPrefix == "" || strings.HasPrefix(key, b.keyPrefix+"/") { + return key + } + return b.keyPrefix + "/" + key +} diff --git a/internal/storage/s3.go b/internal/storage/s3.go index 787cd7c8..af7c3e59 100644 --- a/internal/storage/s3.go +++ b/internal/storage/s3.go @@ -1,4 +1,3 @@ -// Copyright 2025 linux.do // Copyright 2026 Arctel.net // SPDX-License-Identifier: Apache-2.0 @@ -8,253 +7,116 @@ import ( "context" "fmt" "io" - "log" - "net/http" "strings" - "github.com/Rain-kl/Wavelet/internal/config" - "github.com/Rain-kl/Wavelet/internal/otel_trace" - "github.com/Rain-kl/Wavelet/internal/util" "github.com/aws/aws-sdk-go-v2/aws" awsconfig "github.com/aws/aws-sdk-go-v2/config" "github.com/aws/aws-sdk-go-v2/credentials" "github.com/aws/aws-sdk-go-v2/service/s3" - "go.opentelemetry.io/otel/attribute" - "go.opentelemetry.io/otel/codes" - "go.opentelemetry.io/otel/trace" ) -var ( +type s3Backend struct { client *s3.Client bucket string keyPrefix string cdnURL string -) +} -func init() { - cfg := config.Config.S3 - if !cfg.Enabled { - log.Println("[Storage] S3 storage disabled") - return - } - - bucket = cfg.Bucket - keyPrefix = cfg.KeyPrefix - cdnURL = strings.TrimRight(cfg.CdnURL, "/") - - awsCfg, err := awsconfig.LoadDefaultConfig(context.Background(), +func newS3Backend(ctx context.Context, cfg ObjectConfig) (*s3Backend, error) { + awsCfg, err := awsconfig.LoadDefaultConfig(ctx, awsconfig.WithRegion(cfg.Region), - awsconfig.WithCredentialsProvider( - credentials.NewStaticCredentialsProvider(cfg.AccessKeyID, cfg.SecretAccessKey, ""), - ), + awsconfig.WithCredentialsProvider(credentials.NewStaticCredentialsProvider( + cfg.AccessKeyID, + cfg.SecretAccessKey, + "", + )), ) if err != nil { - log.Fatalf("[Storage] failed to load AWS config: %v\n", err) + return nil, fmt.Errorf("load S3 config: %w", err) } - - client = s3.NewFromConfig(awsCfg, func(o *s3.Options) { + client := s3.NewFromConfig(awsCfg, func(options *s3.Options) { if cfg.Endpoint != "" { - o.BaseEndpoint = aws.String(cfg.Endpoint) + options.BaseEndpoint = aws.String(strings.TrimRight(cfg.Endpoint, "/")) } - o.UsePathStyle = cfg.PathStyle + options.UsePathStyle = cfg.PathStyle }) - - log.Printf("[Storage] S3 storage initialized (bucket: %s, prefix: %s, cdn: %s)\n", bucket, keyPrefix, cdnURL) + return &s3Backend{ + client: client, + bucket: cfg.Bucket, + keyPrefix: strings.Trim(cfg.KeyPrefix, "/"), + cdnURL: strings.TrimRight(cfg.CDNURL, "/"), + }, nil } -// IsEnabledFunc 检查 S3 存储是否已初始化（可替换用于测试） -var IsEnabledFunc = func() bool { - return client != nil -} - -// IsEnabled 检查 S3 存储是否可用 -func IsEnabled() bool { - return IsEnabledFunc() -} - -// BuildKey constructs a full S3 object key with the configured prefix. -func BuildKey(path string) string { - return keyPrefix + path -} - -var ( - // PutObjectFunc enables mocking S3 uploads in tests. - PutObjectFunc = putObjectDefault - // GetObjectFunc enables mocking S3 downloads in tests. - GetObjectFunc = getObjectDefault - // DeleteObjectFunc enables mocking S3 deletion in tests. - DeleteObjectFunc = deleteObjectDefault -) - -// MockStorage is a test helper to mock S3 storage operations. -// It returns a function that restores original implementations. -func MockStorage( - mockPut func(ctx context.Context, key string, body io.Reader, size int64, contentType string) error, - mockGet func(ctx context.Context, key string) (*ObjectInfo, error), - mockDelete func(ctx context.Context, key string) error, -) func() { - origPut, origGet, origDelete := PutObjectFunc, GetObjectFunc, DeleteObjectFunc - PutObjectFunc = mockPut - GetObjectFunc = mockGet - DeleteObjectFunc = mockDelete - return func() { - PutObjectFunc = origPut - GetObjectFunc = origGet - DeleteObjectFunc = origDelete +func newR2Backend(ctx context.Context, cfg ObjectConfig) (*s3Backend, error) { + if cfg.Endpoint == "" { + cfg.Endpoint = fmt.Sprintf("https://%s.r2.cloudflarestorage.com", cfg.AccountID) } + cfg.Region = "auto" + return newS3Backend(ctx, cfg) } -// PutObject uploads a file to S3. -func PutObject(ctx context.Context, key string, body io.Reader, size int64, contentType string) error { - return PutObjectFunc(ctx, key, body, size, contentType) -} - -func putObjectDefault(ctx context.Context, key string, body io.Reader, size int64, contentType string) error { - ctx, span := otel_trace.Start(ctx, "S3.PutObject", trace.WithSpanKind(trace.SpanKindClient)) - defer span.End() - - span.SetAttributes( - attribute.String("s3.key", key), - attribute.Int64("s3.content_length", size), - attribute.String("s3.content_type", contentType), - ) - - if !IsEnabled() { - span.SetStatus(codes.Error, "S3 not initialized") - return ErrS3InitializationFailed{} - } - - input := &s3.PutObjectInput{ - Bucket: aws.String(bucket), +func (b *s3Backend) Put(ctx context.Context, key string, body io.Reader, size int64, contentType string) (string, error) { + key = b.key(key) + _, err := b.client.PutObject(ctx, &s3.PutObjectInput{ + Bucket: aws.String(b.bucket), Key: aws.String(key), Body: body, ContentLength: aws.Int64(size), ContentType: aws.String(contentType), - } - - _, err := client.PutObject(ctx, input) + }) if err != nil { - span.SetStatus(codes.Error, fmt.Sprintf("S3 put object failed: %v", err)) - return fmt.Errorf(errS3PutObjectFailed, err) + return "", fmt.Errorf("put S3 object: %w", err) } - return nil + return key, nil } -// ObjectInfo holds metadata about a retrieved object. -type ObjectInfo struct { - CachePath string - Body io.ReadCloser - ContentLength int64 - ContentType string -} - -// GetObject retrieves a file directly from S3. -func GetObject(ctx context.Context, key string) (*ObjectInfo, error) { - return GetObjectFunc(ctx, key) -} - -func getObjectDefault(ctx context.Context, key string) (*ObjectInfo, error) { - ctx, span := otel_trace.Start(ctx, "S3.GetObject", trace.WithSpanKind(trace.SpanKindClient)) - defer span.End() - - span.SetAttributes(attribute.String("s3.key", key)) - - if !IsEnabled() { - span.SetStatus(codes.Error, "S3 not initialized") - return nil, ErrS3InitializationFailed{} +func (b *s3Backend) Get(ctx context.Context, key string) (*Object, error) { + key = b.key(key) + if b.cdnURL != "" { + return getHTTPObject(ctx, b.cdnURL, key) } - - output, err := client.GetObject(ctx, &s3.GetObjectInput{ - Bucket: aws.String(bucket), + output, err := b.client.GetObject(ctx, &s3.GetObjectInput{ + Bucket: aws.String(b.bucket), Key: aws.String(key), }) if err != nil { - span.SetStatus(codes.Error, fmt.Sprintf("S3 get object failed: %v", err)) - return nil, fmt.Errorf(errS3GetObjectFailed, err) + return nil, fmt.Errorf("get S3 object: %w", err) } - - contentType := "application/octet-stream" + contentType := defaultContentType if output.ContentType != nil { contentType = *output.ContentType } - - var contentLength int64 + var size int64 if output.ContentLength != nil { - contentLength = *output.ContentLength + size = *output.ContentLength } - - return &ObjectInfo{ - Body: output.Body, - ContentLength: contentLength, - ContentType: contentType, - }, nil + return &Object{Body: output.Body, ContentLength: size, ContentType: contentType}, nil } -// GetObjectViaProxy retrieves a file via CDN if configured, otherwise falls back to S3. -func GetObjectViaProxy(ctx context.Context, key string) (*ObjectInfo, error) { - ctx, span := otel_trace.Start(ctx, "S3.GetObjectViaProxy", trace.WithSpanKind(trace.SpanKindClient)) - defer span.End() - - span.SetAttributes(attribute.String("s3.key", key)) - - if !IsEnabled() { - span.SetStatus(codes.Error, "S3 not initialized") - return nil, ErrS3InitializationFailed{} - } - - if cdnURL == "" { - return GetObject(ctx, key) - } - - url := cdnURL + "/" + key - span.SetAttributes(attribute.Bool("s3.use_cdn", true)) - - resp, err := util.Request(ctx, http.MethodGet, url, nil, nil, nil) - if err != nil { - span.SetStatus(codes.Error, fmt.Sprintf("cdn request failed: %v", err)) - return nil, fmt.Errorf(errCDNRequestFailed, err) - } - - if resp.StatusCode != http.StatusOK { - _ = resp.Body.Close() - span.SetStatus(codes.Error, fmt.Sprintf("cdn returned status %d", resp.StatusCode)) - return nil, fmt.Errorf(errCDNStatusFailed, resp.StatusCode) - } - - contentType := resp.Header.Get("Content-Type") - if contentType == "" { - contentType = "application/octet-stream" - } - - return &ObjectInfo{ - Body: resp.Body, - ContentLength: resp.ContentLength, - ContentType: contentType, - }, nil -} - -// DeleteObject deletes a file from S3. -func DeleteObject(ctx context.Context, key string) error { - return DeleteObjectFunc(ctx, key) -} - -func deleteObjectDefault(ctx context.Context, key string) error { - ctx, span := otel_trace.Start(ctx, "S3.DeleteObject", trace.WithSpanKind(trace.SpanKindClient)) - defer span.End() - - span.SetAttributes(attribute.String("s3.key", key)) - - if !IsEnabled() { - return ErrS3InitializationFailed{} - } - - _, err := client.DeleteObject(ctx, &s3.DeleteObjectInput{ - Bucket: aws.String(bucket), - Key: aws.String(key), +func (b *s3Backend) Delete(ctx context.Context, key string) error { + _, err := b.client.DeleteObject(ctx, &s3.DeleteObjectInput{ + Bucket: aws.String(b.bucket), + Key: aws.String(b.key(key)), }) if err != nil { - span.SetStatus(codes.Error, fmt.Sprintf("S3 delete object failed: %v", err)) - return fmt.Errorf(errS3DeleteObjectFailed, err) + return fmt.Errorf("delete S3 object: %w", err) } return nil } + +func (b *s3Backend) Test(ctx context.Context) error { + _, err := b.client.HeadBucket(ctx, &s3.HeadBucketInput{Bucket: aws.String(b.bucket)}) + if err != nil { + return fmt.Errorf("access S3 bucket: %w", err) + } + return nil +} + +func (b *s3Backend) key(key string) string { + key = strings.TrimLeft(key, "/") + if b.keyPrefix == "" || strings.HasPrefix(key, b.keyPrefix+"/") { + return key + } + return b.keyPrefix + "/" + key +} diff --git a/internal/storage/storage.go b/internal/storage/storage.go new file mode 100644 index 00000000..65492e60 --- /dev/null +++ b/internal/storage/storage.go @@ -0,0 +1,127 @@ +// Copyright 2026 Arctel.net +// SPDX-License-Identifier: Apache-2.0 + +package storage + +import ( + "context" + "fmt" + "io" +) + +const ( + defaultContentType = "application/octet-stream" + storageDirPerm = 0o750 + storageFilePerm = 0o600 +) + +// Object describes a readable stored object. +type Object struct { + CachePath string + Body io.ReadCloser + ContentLength int64 + ContentType string +} + +// Backend defines storage operations used by the upload domain. +type Backend interface { + Put(ctx context.Context, key string, body io.Reader, size int64, contentType string) (string, error) + Get(ctx context.Context, key string) (*Object, error) + Delete(ctx context.Context, key string) error + Test(ctx context.Context) error +} + +var ( + // IsEnabledFunc preserves the legacy S3 test hook while tests migrate to backend injection. + IsEnabledFunc = func() bool { return false } + mockBackend Backend +) + +// Active returns the configured active driver and backend. +func Active(ctx context.Context) (Driver, Backend, error) { + if IsEnabledFunc() && mockBackend != nil { + return DriverS3, mockBackend, nil + } + cfg, err := LoadConfig(ctx) + if err != nil { + return "", nil, err + } + backend, err := NewBackend(ctx, cfg, cfg.Driver) + return cfg.Driver, backend, err +} + +// ForDriver returns the active or pending backend for an upload record. +func ForDriver(ctx context.Context, driver Driver) (Backend, error) { + if driver == DriverS3 && mockBackend != nil { + return mockBackend, nil + } + cfg, err := LoadConfig(ctx) + if err != nil { + return nil, err + } + if cfg.Driver == driver { + return NewBackend(ctx, cfg, driver) + } + return nil, fmt.Errorf("storage configuration for driver %q is unavailable", driver) +} + +type functionBackend struct { + put func(context.Context, string, io.Reader, int64, string) error + get func(context.Context, string) (*Object, error) + delete func(context.Context, string) error +} + +func (b *functionBackend) Put(ctx context.Context, key string, body io.Reader, size int64, contentType string) (string, error) { + if err := b.put(ctx, key, body, size, contentType); err != nil { + return "", err + } + return key, nil +} + +func (b *functionBackend) Get(ctx context.Context, key string) (*Object, error) { + return b.get(ctx, key) +} + +func (b *functionBackend) Delete(ctx context.Context, key string) error { + return b.delete(ctx, key) +} + +func (b *functionBackend) Test(context.Context) error { + return nil +} + +// MockStorage replaces object operations for package tests and returns a restore function. +func MockStorage( + put func(context.Context, string, io.Reader, int64, string) error, + get func(context.Context, string) (*Object, error), + deleteObject func(context.Context, string) error, +) func() { + previous := mockBackend + mockBackend = &functionBackend{put: put, get: get, delete: deleteObject} + return func() { + mockBackend = previous + } +} + +// NewBackend constructs a concrete backend from configuration. +func NewBackend(ctx context.Context, cfg Config, driver Driver) (Backend, error) { + if driver == DriverS3 && mockBackend != nil { + return mockBackend, nil + } + switch driver { + case DriverLocal: + return newLocalBackend(cfg.Local) + case DriverS3: + return newS3Backend(ctx, cfg.S3) + case DriverR2: + return newR2Backend(ctx, cfg.R2) + case DriverMinIO: + return newS3Backend(ctx, cfg.MinIO) + case DriverOSS: + return newOSSBackend(cfg.OSS) + case DriverWebDAV: + return newWebDAVBackend(cfg.WebDAV) + default: + return nil, fmt.Errorf("unsupported storage driver %q", driver) + } +} diff --git a/internal/storage/webdav.go b/internal/storage/webdav.go new file mode 100644 index 00000000..f8619452 --- /dev/null +++ b/internal/storage/webdav.go @@ -0,0 +1,74 @@ +// Copyright 2026 Arctel.net +// SPDX-License-Identifier: Apache-2.0 + +package storage + +import ( + "context" + "fmt" + "io" + "path" + "strings" + + "github.com/studio-b12/gowebdav" +) + +type webDAVBackend struct { + client *gowebdav.Client + basePath string +} + +func newWebDAVBackend(cfg WebDAVConfig) (*webDAVBackend, error) { + return &webDAVBackend{ + client: gowebdav.NewClient(strings.TrimRight(cfg.Endpoint, "/"), cfg.Username, cfg.Password), + basePath: strings.Trim(cfg.BasePath, "/"), + }, nil +} + +func (b *webDAVBackend) Put(_ context.Context, key string, body io.Reader, size int64, _ string) (string, error) { + key = b.key(key) + if dir := path.Dir(key); dir != "." && dir != "/" { + if err := b.client.MkdirAll(dir, storageDirPerm); err != nil { + return "", fmt.Errorf("create WebDAV directory: %w", err) + } + } + if err := b.client.WriteStreamWithLength(key, body, size, storageFilePerm); err != nil { + return "", fmt.Errorf("put WebDAV object: %w", err) + } + return key, nil +} + +func (b *webDAVBackend) Get(_ context.Context, key string) (*Object, error) { + key = b.key(key) + info, err := b.client.Stat(key) + if err != nil { + return nil, fmt.Errorf("stat WebDAV object: %w", err) + } + body, err := b.client.ReadStream(key) + if err != nil { + return nil, fmt.Errorf("get WebDAV object: %w", err) + } + contentType := defaultContentType + if typed, ok := info.(interface{ ContentType() string }); ok && typed.ContentType() != "" { + contentType = typed.ContentType() + } + return &Object{Body: body, ContentLength: info.Size(), ContentType: contentType}, nil +} + +func (b *webDAVBackend) Delete(_ context.Context, key string) error { + if err := b.client.Remove(b.key(key)); err != nil { + return fmt.Errorf("delete WebDAV object: %w", err) + } + return nil +} + +func (b *webDAVBackend) Test(_ context.Context) error { + if err := b.client.Connect(); err != nil { + return fmt.Errorf("connect WebDAV: %w", err) + } + return nil +} + +func (b *webDAVBackend) key(key string) string { + return "/" + path.Join(b.basePath, strings.TrimLeft(key, "/")) +} diff --git a/internal/task/handlers/register.go b/internal/task/handlers/register.go index 9d7304e7..01c3e767 100644 --- a/internal/task/handlers/register.go +++ b/internal/task/handlers/register.go @@ -13,6 +13,9 @@ import ( // Register registers all built-in task handlers and their metadata. func Register() { + task.RegisterHandler(upload.StorageMigrationTask, &upload.MigrationHandler{}) + task.RegisterTaskMeta(upload.StorageMigrationMeta) + // upload task.RegisterHandler(upload.CleanupUnusedUploadsTask, &upload.CleanupUnusedUploadsHandler{}) task.RegisterTaskMeta(upload.CleanupUnusedUploadsMeta) diff --git a/internal/testhelper/test_helper.go b/internal/testhelper/test_helper.go index dae5323e..db938c4d 100644 --- a/internal/testhelper/test_helper.go +++ b/internal/testhelper/test_helper.go @@ -265,6 +265,12 @@ func getSeedConfigsPart2() []model.SystemConfig { Type: configTypeSystem, Description: "GitHub Actions Release 上游仓库（owner/repo 或 GitHub 仓库地址）", }, + { + Key: model.ConfigKeyStorageConfig, + Value: `{"driver":"local","local":{"root":"."},"s3":{"region":"us-east-1"},"r2":{"region":"auto"},"minio":{"region":"us-east-1","path_style":true},"oss":{},"webdav":{}}`, + Type: configTypeSystem, + Description: "文件存储驱动及连接配置（JSON）", + }, } }