tcjlj/OpenFlare
1
0
Fork 0
mirror of https://github.com/Rain-kl/OpenFlare.git synced 2026-09-29 05:56:38 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(storage): add dynamic storage config and migration

Browse Source 
Move storage backend configuration from startup YAML to system_config-backed runtime configuration. Add local, S3-compatible, R2, MinIO, OSS, and WebDAV backend support.

Add a storage migration async task using the existing task dispatch framework. Migration target config is carried in task payload, and maintenance mode is derived from task execution state.

Split upload file management and storage operations, add the admin storage configuration tab, and update migrations and Swagger docs.
This commit is contained in:
ryan
2026-06-13 14:55:15 +08:00
parent 4bf8a4806e
commit 9a18bea324
46 changed files with 2005 additions and 743 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+1
View File
@@ -56,3 +56,4 @@ s3_cache
/.idea/
/uploads/
/*-source/
/.cache/
config.example.yaml
-16
View File
@@ -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://<account-id>.r2.cloudflarestorage.com"
region: "auto"
bucket: "<bucket-name>"
access_key_id: "<S3_ACCESS_KEY_ID>"
secret_access_key: "<S3_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
docs/docs.go
+2 -4
View File
@@ -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",
docs/swagger.json
+2 -4
View File
@@ -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",
docs/swagger.yaml
-2
View File
@@ -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:
frontend/components/common/admin/files.tsx
+9 -3
View File
@@ -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"
>
{/* 顶部标题区 */}
<div className="flex flex-col md:flex-row md:items-center justify-between gap-4 border-b pb-5">
@@ -29,10 +30,11 @@ export function FilesMain() {
</div>
</div>
<Tabs value={activeTab} onValueChange={setActiveTab} className="w-full space-y-6">
<TabsList className="grid w-fit grid-cols-2">
<Tabs value={activeTab} onValueChange={setActiveTab} className="flex w-full flex-col gap-6">
<TabsList className="grid w-fit grid-cols-3">
<TabsTrigger value="stats">文件存储信息</TabsTrigger>
<TabsTrigger value="list">文件列表</TabsTrigger>
<TabsTrigger value="storage">存储配置</TabsTrigger>
</TabsList>
{/* ──────── TAB 1: 统计看板 ──────── */}
@@ -44,6 +46,10 @@ export function FilesMain() {
<TabsContent value="list" className="outline-hidden">
<FileList />
</TabsContent>
<TabsContent value="storage" className="outline-hidden">
<StorageConfigTab />
</TabsContent>
</Tabs>
</motion.div>
)
frontend/components/common/admin/storage-config-tab.tsx
+339
View File
@@ -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<StorageDriver, string> = {
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<StorageMigrationPayload>
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<StorageConfig | null>(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 <div className="flex justify-center py-20"><Loader2 className="animate-spin" /></div>
}
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<ObjectStorageConfig>) => {
setConfig((current) => current ? {
...current,
[driver]: {...current[driver], ...patch},
} : current)
}
return (
<div className="flex w-full flex-col gap-6">
{isReadOnly && migration && (
<Card>
<CardHeader>
<CardTitle className="flex items-center gap-2">
存储维护模式
<Badge variant={migration.state === "failed" ? "destructive" : "secondary"}>
{migration.state}
</Badge>
</CardTitle>
<CardDescription>
{driverLabel(migration.source_driver)} → {driverLabel(migration.target_driver)}。迁移期间文件只允许读取，禁止上传、删除和清理。
</CardDescription>
</CardHeader>
<CardContent className="flex flex-col gap-3">
<Progress value={migration.state === "succeeded" ? 100 : undefined} />
<p className="text-sm text-muted-foreground">
迁移进度请查看任务执行日志：{migration.task_id || "尚未下发"}
</p>
{migration.error && <p className="text-sm text-destructive">{migration.error}</p>}
</CardContent>
<CardFooter>
<Button
variant="outline"
disabled={migration.state !== "failed" || !migration.id || runMutation.isPending}
onClick={() => migration.id && runMutation.mutate(migration.id)}
>
{runMutation.isPending ? <Loader2 data-icon="inline-start" className="animate-spin" /> : <Play data-icon="inline-start" />}
重试迁移
</Button>
</CardFooter>
</Card>
)}
<Card>
<CardHeader>
<CardTitle className="flex items-center gap-2">
<Database />
文件存储
</CardTitle>
<CardDescription>
默认使用本地存储。切换存储类型且已有文件时，系统会自动进入维护模式并迁移文件。
</CardDescription>
</CardHeader>
<CardContent>
<FieldGroup>
<Field>
<FieldLabel>存储类型</FieldLabel>
<Select
value={config.driver}
disabled={isFormDisabled}
onValueChange={(value) => setConfig({...config, driver: value as StorageDriver})}
>
<SelectTrigger className="w-full">
<SelectValue />
</SelectTrigger>
<SelectContent>
<SelectGroup>
{(Object.keys(driverLabels) as StorageDriver[]).map((driver) => (
<SelectItem key={driver} value={driver}>{driverLabels[driver]}</SelectItem>
))}
</SelectGroup>
</SelectContent>
</Select>
</Field>
{config.driver === "local" && (
<TextField
label="根目录"
value={config.local.root}
placeholder="."
onChange={(root) => setConfig({...config, local: {root}})}
/>
)}
{(config.driver === "s3" || config.driver === "r2" || config.driver === "minio" || config.driver === "oss") && (
<ObjectFields
driver={config.driver as "s3" | "r2" | "minio" | "oss"}
value={config[config.driver as "s3" | "r2" | "minio" | "oss"]}
onChange={(patch) => updateObject(config.driver as "s3" | "r2" | "minio" | "oss", patch)}
/>
)}
{config.driver === "webdav" && (
<>
<TextField label="服务地址" value={config.webdav.endpoint} placeholder="https://dav.example.com" onChange={(endpoint) => setConfig({...config, webdav: {...config.webdav, endpoint}})} />
<TextField label="用户名" value={config.webdav.username} onChange={(username) => setConfig({...config, webdav: {...config.webdav, username}})} />
<TextField label="密码" type="password" value={config.webdav.password} onChange={(password) => setConfig({...config, webdav: {...config.webdav, password}})} />
<TextField label="基础路径" value={config.webdav.base_path} placeholder="wavelet" onChange={(base_path) => setConfig({...config, webdav: {...config.webdav, base_path}})} />
</>
)}
</FieldGroup>
</CardContent>
<CardFooter className="justify-end gap-2">
<Button disabled={isFormDisabled || saveMutation.isPending} onClick={() => saveMutation.mutate(config)}>
{saveMutation.isPending ? <Loader2 data-icon="inline-start" className="animate-spin" /> : <Save data-icon="inline-start" />}
保存配置
</Button>
</CardFooter>
</Card>
</div>
)
}
function ObjectFields({
driver,
value,
onChange,
}: {
driver: "s3" | "r2" | "minio" | "oss"
value: ObjectStorageConfig
onChange: (patch: Partial<ObjectStorageConfig>) => void
}) {
return (
<>
{driver === "r2" && <TextField label="Account ID" value={value.account_id || ""} onChange={(account_id) => onChange({account_id})} />}
{driver !== "s3" && <TextField label="Endpoint" value={value.endpoint} placeholder="https://..." onChange={(endpoint) => onChange({endpoint})} />}
<TextField label="Region" value={value.region} onChange={(region) => onChange({region})} />
<TextField label="Bucket" value={value.bucket} onChange={(bucket) => onChange({bucket})} />
<TextField label="Access Key ID" value={value.access_key_id} onChange={(access_key_id) => onChange({access_key_id})} />
<TextField label="Secret Access Key" type="password" value={value.secret_access_key} onChange={(secret_access_key) => onChange({secret_access_key})} />
<TextField label="对象前缀" value={value.key_prefix} placeholder="uploads" onChange={(key_prefix) => onChange({key_prefix})} />
<TextField label="CDN 地址" value={value.cdn_url} placeholder="https://cdn.example.com" onChange={(cdn_url) => onChange({cdn_url})} />
{(driver === "s3" || driver === "minio") && (
<Field orientation="horizontal">
<FieldLabel>Path Style</FieldLabel>
<Switch checked={value.path_style} onCheckedChange={(path_style) => onChange({path_style})} />
<FieldDescription>MinIO 等自托管 S3 通常需要开启。</FieldDescription>
</Field>
)}
</>
)
}
function TextField({
label,
value,
onChange,
placeholder,
type = "text",
}: {
label: string
value: string
onChange: (value: string) => void
placeholder?: string
type?: React.HTMLInputTypeAttribute
}) {
return (
<Field>
<FieldLabel>{label}</FieldLabel>
<Input type={type} value={value} placeholder={placeholder} onChange={(event) => onChange(event.target.value)} />
</Field>
)
}
frontend/lib/services/admin/admin.service.ts
-2
View File
@@ -155,8 +155,6 @@ export class AdminService extends BaseService {
return this.get<string[]>('/uploads/types');
}
// ==================== 认证源管理 ====================
static async listAuthSources(): Promise<AuthSource[]> {
frontend/lib/services/admin/types.ts
+29 -2
View File
@@ -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;
};
}
// ==================== 任务管理 ====================
go.mod
+2
View File
@@ -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
go.sum
+4
View File
@@ -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=
internal/apps/admin/system_config/routers.go
+53 -8
View File
@@ -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), &currentCfg); 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
}
internal/apps/admin/system_config/routers_test.go
+1 -1
View File
@@ -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)
internal/apps/upload/constants.go
+3 -1
View File
@@ -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)
)
internal/apps/upload/errs.go
+1
View File
@@ -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 /"
internal/apps/upload/file_management.go
+167
View File
@@ -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))
}
internal/apps/upload/file_server.go
+2 -31
View File
@@ -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)
}
internal/apps/upload/routers.go
+37 -190
View File
@@ -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))
}
internal/apps/upload/routers_test.go
+2 -3
View File
@@ -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["文档"])
}
}
internal/apps/upload/storage_migration_task.go
+268
View File
@@ -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
}
internal/apps/upload/storage_migration_task_test.go
+110
View File
@@ -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)
}
}
internal/apps/upload/storage_ops.go
+87
View File
@@ -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 ""
}
}
internal/apps/upload/tasks.go
+12 -2
View File
@@ -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
}
internal/apps/upload/tasks_test.go
+9 -2
View File
@@ -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})
internal/config/config.go
-10
View File
@@ -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)
internal/config/model.go
-20
View File
@@ -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"`
}
internal/db/migrator/goose/postgres/202606090001_initial_schema.sql
+3 -2
View File
@@ -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)
internal/db/migrator/goose/postgres/202606130002_expand_system_config_value.sql
+5
View File
@@ -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);
internal/db/migrator/goose/postgres/202606130003_add_storage_config.sql
+15
View File
@@ -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';
internal/db/migrator/goose/sqlite/202606090001_initial_schema.sql
+3 -2
View File
@@ -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)
internal/db/migrator/goose/sqlite/202606130002_expand_system_config_value.sql
+6
View File
@@ -0,0 +1,6 @@
-- +goose Up
-- SQLite stores VARCHAR and TEXT with the same TEXT affinity.
SELECT 1;
-- +goose Down
SELECT 1;
internal/db/migrator/goose/sqlite/202606130003_add_storage_config.sql
+15
View File
@@ -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';
internal/db/migrator/migrator_test.go
+1 -1
View File
@@ -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{
internal/model/system_configs.go
+2 -2
View File
@@ -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"`
internal/storage/cache.go
-205
View File
@@ -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
}
internal/storage/config.go
+234
View File
@@ -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
}
internal/storage/errs.go
-29
View File
@@ -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"
)
internal/storage/http.go
+39
View File
@@ -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
}
internal/storage/local.go
+120
View File
@@ -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
}
internal/storage/local_test.go
+51
View File
@@ -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)
}
}
internal/storage/oss.go
+98
View File
@@ -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
}
internal/storage/s3.go
+63 -201
View File
@@ -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
}
internal/storage/storage.go
+127
View File
@@ -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)
}
}
internal/storage/webdav.go
+74
View File
@@ -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, "/"))
}
internal/task/handlers/register.go
+3
View File
@@ -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)
internal/testhelper/test_helper.go
+6
View File
@@ -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）",
},
}
}