diff --git a/cmd/main/main.go b/cmd/main/main.go index e0e49aa..b9b84e5 100644 --- a/cmd/main/main.go +++ b/cmd/main/main.go @@ -1,7 +1,13 @@ package main import ( - "fmt" + "context" + "errors" + "net/http" + "os" + "os/signal" + "syscall" + "time" "git.apinb.com/bsm-sdk/core/infra" "git.apinb.com/bsm-sdk/core/middleware" @@ -16,12 +22,33 @@ import ( const ServiceKey = "Files" func main() { + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + config.New(ServiceKey) impl.NewImpl() - if err := models.InitData(); err != nil { + if err := models.RequireSchemaVersion(impl.DBService, "files"); err != nil { panic(err) } - jobs.StartPendingUploadCleanup() + if err := models.RequireStorageProvider(impl.DBService, impl.StorageService.Provider()); err != nil { + panic(err) + } + if err := jobs.StartPendingUploadCleanup(ctx); err != nil { + if closeErr := impl.Close(); closeErr != nil { + panic(errors.Join(err, closeErr)) + } + panic(err) + } + if err := jobs.StartIntegrity(ctx); err != nil { + stop() + waitCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + waitErr := jobs.Wait(waitCtx) + cancel() + if closeErr := impl.Close(); closeErr != nil { + panic(errors.Join(err, waitErr, closeErr)) + } + panic(errors.Join(err, waitErr)) + } app := gin.Default() middleware.Mode(app) @@ -30,7 +57,30 @@ func main() { app.HEAD("/", infra.Health) routers.Register(ServiceKey, app) - if err := app.Run(fmt.Sprintf(":%s", config.Spec.Port)); err != nil { + serveErr := serve(ctx, app) + stop() + waitCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + waitErr := jobs.Wait(waitCtx) + cancel() + closeErr := impl.Close() + if err := errors.Join(serveErr, waitErr, closeErr); err != nil { panic(err) } } + +func serve(ctx context.Context, app http.Handler) error { + server := &http.Server{Addr: config.Spec.Addr, Handler: app, ReadHeaderTimeout: 10 * time.Second} + errCh := make(chan error, 1) + go func() { errCh <- server.ListenAndServe() }() + select { + case err := <-errCh: + if errors.Is(err, http.ErrServerClosed) { + return nil + } + return err + case <-ctx.Done(): + shutdownCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + return server.Shutdown(shutdownCtx) + } +} diff --git a/etc/files_dev.yaml b/etc/files_dev.yaml index 125e0ac..cdc3075 100644 --- a/etc/files_dev.yaml +++ b/etc/files_dev.yaml @@ -1,11 +1,12 @@ Service: files Port: 12452 +BindIP: 0.0.0.0 SecretKey: ops-files Databases: Driver: postgres Source: - - host=8.137.107.29 user=postgres password=Weidong2023~! dbname=ops_dev port=19432 sslmode=disable TimeZone=Asia/Shanghai + - "${OPS_FILES_DB_DSN}" Cache: redis://null:Weidong2023~!@8.137.107.29:19379/ @@ -19,17 +20,26 @@ ServiceClients: dc-control: ${DC_CONTROL_FILES_SECRET} visual: ${VISUAL_FILES_SECRET} -ObjectStorage: - Provider: aliyun - Endpoint: https://oss-cn-beijing.aliyuncs.com - Region: cn-beijing - Bucket: ops-app - PublicBaseURL: https://ops-app.oss-cn-beijing.aliyuncs.com - AccessKeyID: ${FILES_OSS_ACCESS_KEY_ID} - AccessKeySecret: ${FILES_OSS_ACCESS_KEY_SECRET} - PresignTTLSeconds: 600 +Storage: + Provider: local + ExternalBaseURL: http://127.0.0.1:12452/Files + AccessTTLSeconds: 600 + SigningSecret: local-dev-files-signing-secret + Local: + RootPath: D:/work/ops/.data/files + MinimumFreeSpaceMB: 1024 + Aliyun: + Endpoint: "" + Region: "" + Bucket: "" + AccessKeyID: "" + AccessKeySecret: "" Namespaces: + device-config-backups: + Prefix: device-config-backups + MaxSizeMB: 1 + AllowedExtensions: [.txt] reports: Prefix: reports MaxSizeMB: 512 @@ -45,3 +55,5 @@ Namespaces: Cleanup: IntervalSeconds: 600 + IntegrityIntervalSeconds: 3600 + IntegrityBatchSize: 100 diff --git a/etc/files_prod.yaml b/etc/files_prod.yaml index 125e0ac..da0568d 100644 --- a/etc/files_prod.yaml +++ b/etc/files_prod.yaml @@ -1,13 +1,14 @@ Service: files Port: 12452 -SecretKey: ops-files +BindIP: 127.0.0.1 +SecretKey: ${FILES_SERVICE_SECRET} Databases: Driver: postgres Source: - - host=8.137.107.29 user=postgres password=Weidong2023~! dbname=ops_dev port=19432 sslmode=disable TimeZone=Asia/Shanghai + - "${OPS_FILES_DB_DSN}" -Cache: redis://null:Weidong2023~!@8.137.107.29:19379/ +Cache: ${OPS_CACHE_DSN} MicroService: Enable: false @@ -19,17 +20,26 @@ ServiceClients: dc-control: ${DC_CONTROL_FILES_SECRET} visual: ${VISUAL_FILES_SECRET} -ObjectStorage: - Provider: aliyun - Endpoint: https://oss-cn-beijing.aliyuncs.com - Region: cn-beijing - Bucket: ops-app - PublicBaseURL: https://ops-app.oss-cn-beijing.aliyuncs.com - AccessKeyID: ${FILES_OSS_ACCESS_KEY_ID} - AccessKeySecret: ${FILES_OSS_ACCESS_KEY_SECRET} - PresignTTLSeconds: 600 +Storage: + Provider: ${FILES_STORAGE_PROVIDER} + ExternalBaseURL: ${FILES_PUBLIC_BASE_URL} + AccessTTLSeconds: 600 + SigningSecret: ${FILES_STORAGE_SIGNING_SECRET} + Local: + RootPath: ${FILES_LOCAL_ROOT} + MinimumFreeSpaceMB: 10240 + Aliyun: + Endpoint: ${FILES_OSS_ENDPOINT} + Region: ${FILES_OSS_REGION} + Bucket: ${FILES_OSS_BUCKET} + AccessKeyID: ${FILES_OSS_ACCESS_KEY_ID} + AccessKeySecret: ${FILES_OSS_ACCESS_KEY_SECRET} Namespaces: + device-config-backups: + Prefix: device-config-backups + MaxSizeMB: 1 + AllowedExtensions: [.txt] reports: Prefix: reports MaxSizeMB: 512 @@ -45,3 +55,5 @@ Namespaces: Cleanup: IntervalSeconds: 600 + IntegrityIntervalSeconds: 3600 + IntegrityBatchSize: 100 diff --git a/etc/files_test.yaml b/etc/files_test.yaml index 125e0ac..541b04a 100644 --- a/etc/files_test.yaml +++ b/etc/files_test.yaml @@ -1,5 +1,6 @@ Service: files Port: 12452 +BindIP: 127.0.0.1 SecretKey: ops-files Databases: @@ -19,17 +20,26 @@ ServiceClients: dc-control: ${DC_CONTROL_FILES_SECRET} visual: ${VISUAL_FILES_SECRET} -ObjectStorage: - Provider: aliyun - Endpoint: https://oss-cn-beijing.aliyuncs.com - Region: cn-beijing - Bucket: ops-app - PublicBaseURL: https://ops-app.oss-cn-beijing.aliyuncs.com - AccessKeyID: ${FILES_OSS_ACCESS_KEY_ID} - AccessKeySecret: ${FILES_OSS_ACCESS_KEY_SECRET} - PresignTTLSeconds: 600 +Storage: + Provider: local + ExternalBaseURL: http://127.0.0.1:12452/Files + AccessTTLSeconds: 600 + SigningSecret: local-test-files-signing-secret + Local: + RootPath: D:/work/ops/.data/files-test + MinimumFreeSpaceMB: 1 + Aliyun: + Endpoint: "" + Region: "" + Bucket: "" + AccessKeyID: "" + AccessKeySecret: "" Namespaces: + device-config-backups: + Prefix: device-config-backups + MaxSizeMB: 1 + AllowedExtensions: [.txt] reports: Prefix: reports MaxSizeMB: 512 @@ -45,3 +55,5 @@ Namespaces: Cleanup: IntervalSeconds: 600 + IntegrityIntervalSeconds: 3600 + IntegrityBatchSize: 100 diff --git a/etc/ops-files.service b/etc/ops-files.service index c05b6e8..8b08ee2 100644 --- a/etc/ops-files.service +++ b/etc/ops-files.service @@ -13,11 +13,11 @@ Environment=BSM_RuntimeMode=prod Environment=RUN_MODE=prod Environment=BSM_Prefix=/data/app ExecStart=/data/app/ops-files -ExecStartPost=/data/app/systemd/wait-http.sh ops-files http://127.0.0.1:12452/Files/v1/ping/hello 60 +ExecStartPost=/data/app/systemd/wait-http.sh ops-files http://127.0.0.1:12452/ready 60 Restart=on-failure RestartSec=5s TimeoutStartSec=75s -TimeoutStopSec=30s +TimeoutStopSec=75s KillSignal=SIGTERM StandardOutput=append:/data/app/logs/files.log StandardError=inherit diff --git a/go.mod b/go.mod index 162435a..226648b 100644 --- a/go.mod +++ b/go.mod @@ -23,6 +23,7 @@ require ( github.com/gabriel-vasile/mimetype v1.4.11 // indirect github.com/gin-contrib/cors v1.7.6 // indirect github.com/gin-contrib/sse v1.1.0 // indirect + github.com/go-ole/go-ole v1.2.6 // indirect github.com/go-playground/locales v0.14.1 // indirect github.com/go-playground/universal-translator v0.18.1 // indirect github.com/go-playground/validator/v10 v10.28.0 // indirect @@ -38,11 +39,13 @@ require ( github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect github.com/modern-go/reflect2 v1.0.2 // indirect github.com/pelletier/go-toml/v2 v2.2.4 // indirect + github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect github.com/quic-go/qpack v0.5.1 // indirect github.com/quic-go/quic-go v0.55.0 // indirect github.com/skip2/go-qrcode v0.0.0-20200617195104-da1b6568686e // indirect github.com/twitchyliquid64/golang-asm v0.15.1 // indirect github.com/ugorji/go/codec v1.3.1 // indirect + github.com/yusufpapurcu/wmi v1.2.4 // indirect go.etcd.io/etcd/api/v3 v3.6.5 // indirect go.uber.org/mock v0.6.0 // indirect go.uber.org/multierr v1.11.0 // indirect @@ -68,6 +71,7 @@ require ( github.com/jinzhu/now v1.1.5 // indirect github.com/oklog/ulid/v2 v2.1.1 // indirect github.com/redis/go-redis/v9 v9.16.0 // indirect + github.com/shirou/gopsutil/v3 v3.24.5 go.etcd.io/etcd/client/pkg/v3 v3.6.5 // indirect go.etcd.io/etcd/client/v3 v3.6.5 golang.org/x/crypto v0.43.0 // indirect diff --git a/go.sum b/go.sum index 7e0f9b3..0ce6a72 100644 --- a/go.sum +++ b/go.sum @@ -41,6 +41,8 @@ github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= +github.com/go-ole/go-ole v1.2.6 h1:/Fpf6oFPoeFik9ty7siob0G6Ke8QvQEuVcuChpwXzpY= +github.com/go-ole/go-ole v1.2.6/go.mod h1:pprOEPIfldk/42T2oK7lQ4v4JSDwmV0As9GaiUsvbm0= github.com/go-playground/assert/v2 v2.2.0 h1:JvknZsQTYeFEAhQwI4qEt9cyV5ONwRHC+lYKSsYSR8s= github.com/go-playground/assert/v2 v2.2.0/go.mod h1:VDjEfimB/XKnb+ZQfWdccd7VUvScMdVu0Titje2rxJ4= github.com/go-playground/locales v0.14.1 h1:EWaQ/wswjilfKLTECiXz7Rh+3BjFhfDFKv/oXslEjJA= @@ -106,6 +108,8 @@ github.com/pelletier/go-toml/v2 v2.2.4 h1:mye9XuhQ6gvn5h28+VilKrrPoQVanw5PMw/TB0 github.com/pelletier/go-toml/v2 v2.2.4/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c h1:ncq/mPwQF4JjgDlrVEn3C11VoGHZN7m8qihwgMEtzYw= +github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c/go.mod h1:OmDBASR4679mdNQnz2pUhc2G8CO2JrUAVFDRBDP/hJE= github.com/quic-go/qpack v0.5.1 h1:giqksBPnT/HDtZ6VhtFKgoLOWmlyo9Ei6u9PqzIMbhI= github.com/quic-go/qpack v0.5.1/go.mod h1:+PC4XFrEskIVkcLzpEkbLqq1uCoxPhQuvK5rH1ZgaEg= github.com/quic-go/quic-go v0.55.0 h1:zccPQIqYCXDt5NmcEabyYvOnomjs8Tlwl7tISjJh9Mk= @@ -114,6 +118,8 @@ github.com/redis/go-redis/v9 v9.16.0 h1:OotgqgLSRCmzfqChbQyG1PHC3tLNR89DG4jdOERS github.com/redis/go-redis/v9 v9.16.0/go.mod h1:u410H11HMLoB+TP67dz8rL9s6QW2j76l0//kSOd3370= github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= +github.com/shirou/gopsutil/v3 v3.24.5 h1:i0t8kL+kQTvpAYToeuiVk3TgDeKOFioZO3Ztz/iZ9pI= +github.com/shirou/gopsutil/v3 v3.24.5/go.mod h1:bsoOS1aStSs9ErQ1WWfxllSeS1K5D+U30r2NfcubMVk= github.com/skip2/go-qrcode v0.0.0-20200617195104-da1b6568686e h1:MRM5ITcdelLK2j1vwZ3Je0FKVCfqOLp5zO6trqMLYs0= github.com/skip2/go-qrcode v0.0.0-20200617195104-da1b6568686e/go.mod h1:XV66xRDqSt+GTGFMVlhk3ULuV0y9ZmzeVGR4mloJI3M= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= @@ -134,6 +140,8 @@ github.com/ugorji/go/codec v1.3.1 h1:waO7eEiFDwidsBN6agj1vJQ4AG7lh2yqXyOXqhgQuyY github.com/ugorji/go/codec v1.3.1/go.mod h1:pRBVtBSKl77K30Bv8R2P+cLSGaTtex6fsA2Wjqmfxj4= github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= +github.com/yusufpapurcu/wmi v1.2.4 h1:zFUKzehAFReQwLys1b/iSMl+JQGSCSjtVqQn9bBrPo0= +github.com/yusufpapurcu/wmi v1.2.4/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0= go.etcd.io/etcd/api/v3 v3.6.5 h1:pMMc42276sgR1j1raO/Qv3QI9Af/AuyQUW6CBAWuntA= go.etcd.io/etcd/api/v3 v3.6.5/go.mod h1:ob0/oWA/UQQlT1BmaEkWQzI0sJ1M0Et0mMpaABxguOQ= go.etcd.io/etcd/client/pkg/v3 v3.6.5 h1:Duz9fAzIZFhYWgRjp/FgNq2gO1jId9Yae/rLn3RrBP8= @@ -184,7 +192,9 @@ golang.org/x/sync v0.17.0 h1:l60nONMj9l5drqw6jlhIELNv9I0A4OFgRsG9k2oT9Ug= golang.org/x/sync v0.17.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20201204225414-ed752295db88/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.37.0 h1:fdNQudmxPjkdUTPnLn5mdQv7Zwvbvpaxqs831goi9kQ= golang.org/x/sys v0.37.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= diff --git a/internal/config/config.go b/internal/config/config.go index 2442c43..b5bfc7e 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -1,8 +1,11 @@ package config import ( - "log" + "fmt" + "math" "net" + "net/url" + "path/filepath" "regexp" "strconv" "strings" @@ -20,21 +23,32 @@ type SrvConfig struct { Databases *conf.DBConf `yaml:"Databases"` MicroService *conf.MicroServiceConf `yaml:"MicroService"` Etcd *conf.EtcdConf `yaml:"Etcd"` - ObjectStorage ObjectStorageConf `yaml:"ObjectStorage"` + Storage StorageConf `yaml:"Storage"` Namespaces map[string]NamespaceConf `yaml:"Namespaces"` ServiceClients map[string]string `yaml:"ServiceClients"` Cleanup CleanupConf `yaml:"Cleanup"` } -type ObjectStorageConf struct { - Provider string `yaml:"Provider"` - Endpoint string `yaml:"Endpoint"` - Region string `yaml:"Region"` - Bucket string `yaml:"Bucket"` - PublicBaseURL string `yaml:"PublicBaseURL"` - AccessKeyID string `yaml:"AccessKeyID"` - AccessKeySecret string `yaml:"AccessKeySecret"` - PresignTTLSeconds int64 `yaml:"PresignTTLSeconds"` +type StorageConf struct { + Provider string `yaml:"Provider"` + ExternalBaseURL string `yaml:"ExternalBaseURL"` + AccessTTLSeconds int64 `yaml:"AccessTTLSeconds"` + SigningSecret string `yaml:"SigningSecret"` + Local LocalStorageConf `yaml:"Local"` + Aliyun AliyunStorageConf `yaml:"Aliyun"` +} + +type LocalStorageConf struct { + RootPath string `yaml:"RootPath"` + MinimumFreeSpaceMB uint64 `yaml:"MinimumFreeSpaceMB"` +} + +type AliyunStorageConf struct { + Endpoint string `yaml:"Endpoint"` + Region string `yaml:"Region"` + Bucket string `yaml:"Bucket"` + AccessKeyID string `yaml:"AccessKeyID"` + AccessKeySecret string `yaml:"AccessKeySecret"` } type NamespaceConf struct { @@ -44,60 +58,111 @@ type NamespaceConf struct { } type CleanupConf struct { - IntervalSeconds int64 `yaml:"IntervalSeconds"` + IntervalSeconds int64 `yaml:"IntervalSeconds"` + IntegrityIntervalSeconds int64 `yaml:"IntegrityIntervalSeconds"` + IntegrityBatchSize int `yaml:"IntegrityBatchSize"` } func New(srvKey string) { conf.New(srvKey, &Spec) - - Spec.Port = conf.CheckPort(Spec.Port) - Spec.BindIP = conf.CheckIP(Spec.BindIP) + if err := Validate(); err != nil { + panic(fmt.Errorf("files 配置校验失败: %w", err)) + } Spec.Addr = net.JoinHostPort(Spec.BindIP, Spec.Port) - - validate() conf.PrintInfo(Spec.Addr) } -func validate() { - validateRequired("Service", Spec.Service) - validateRequired("Cache", Spec.Cache) - - if Spec.ObjectStorage.Provider != "aliyun" { - configError("ObjectStorage.Provider", "必须为 aliyun") +// Validate 校验文件服务配置，不生成随机监听参数或对象存储默认值。 +func Validate() error { + Spec.Service = strings.TrimSpace(Spec.Service) + Spec.Port = strings.TrimSpace(Spec.Port) + Spec.BindIP = strings.TrimSpace(Spec.BindIP) + if err := validateRequired("Service", Spec.Service); err != nil { + return err + } + if err := validateRequired("Cache", Spec.Cache); err != nil { + return err + } + port, err := strconv.Atoi(Spec.Port) + if err != nil || port < 1 || port > 65535 { + return fmt.Errorf("配置项 Port 无效：必须是 1 到 65535 的整数") + } + if net.ParseIP(Spec.BindIP) == nil { + return fmt.Errorf("配置项 BindIP 无效：必须是明确的 IPv4 或 IPv6 地址") + } + if Spec.Databases == nil || strings.TrimSpace(Spec.Databases.Driver) == "" || len(Spec.Databases.Source) == 0 { + return fmt.Errorf("配置项 Databases 无效：Driver 和 Source 不能为空") } - validateRequired("ObjectStorage.Endpoint", Spec.ObjectStorage.Endpoint) - validateRequired("ObjectStorage.Region", Spec.ObjectStorage.Region) - validateRequired("ObjectStorage.Bucket", Spec.ObjectStorage.Bucket) - validateRequired("ObjectStorage.PublicBaseURL", Spec.ObjectStorage.PublicBaseURL) - validateRequired("ObjectStorage.AccessKeyID", Spec.ObjectStorage.AccessKeyID) - validateRequired("ObjectStorage.AccessKeySecret", Spec.ObjectStorage.AccessKeySecret) + Spec.Storage.Provider = strings.ToLower(strings.TrimSpace(Spec.Storage.Provider)) + for path, value := range map[string]string{ + "Storage.Provider": Spec.Storage.Provider, + "Storage.ExternalBaseURL": Spec.Storage.ExternalBaseURL, + "Storage.SigningSecret": Spec.Storage.SigningSecret, + } { + if err := validateRequired(path, value); err != nil { + return err + } + } + if err := validateHTTPURL("Storage.ExternalBaseURL", Spec.Storage.ExternalBaseURL); err != nil { + return err + } + if Spec.Storage.AccessTTLSeconds < 60 || Spec.Storage.AccessTTLSeconds > 3600 { + return configError("Storage.AccessTTLSeconds", "必须在 60 到 3600 秒之间") + } - if Spec.ObjectStorage.PresignTTLSeconds < 60 || Spec.ObjectStorage.PresignTTLSeconds > 3600 { - configError("ObjectStorage.PresignTTLSeconds", "必须在 60 到 3600 秒之间") + switch Spec.Storage.Provider { + case "local": + Spec.Storage.Local.RootPath = strings.TrimSpace(Spec.Storage.Local.RootPath) + if !filepath.IsAbs(Spec.Storage.Local.RootPath) { + return configError("Storage.Local.RootPath", "必须是绝对路径") + } + if Spec.Storage.Local.MinimumFreeSpaceMB == 0 { + return configError("Storage.Local.MinimumFreeSpaceMB", "必须大于 0") + } + if Spec.Storage.Local.MinimumFreeSpaceMB > math.MaxUint64/(1024*1024) { + return configError("Storage.Local.MinimumFreeSpaceMB", "换算为字节后溢出") + } + case "aliyun": + for path, value := range map[string]string{ + "Storage.Aliyun.Endpoint": Spec.Storage.Aliyun.Endpoint, + "Storage.Aliyun.Region": Spec.Storage.Aliyun.Region, + "Storage.Aliyun.Bucket": Spec.Storage.Aliyun.Bucket, + "Storage.Aliyun.AccessKeyID": Spec.Storage.Aliyun.AccessKeyID, + "Storage.Aliyun.AccessKeySecret": Spec.Storage.Aliyun.AccessKeySecret, + } { + if err := validateRequired(path, value); err != nil { + return err + } + } + if err := validateHTTPURL("Storage.Aliyun.Endpoint", Spec.Storage.Aliyun.Endpoint); err != nil { + return err + } + default: + return configError("Storage.Provider", "必须为 local 或 aliyun") } if len(Spec.Namespaces) == 0 { - configError("Namespaces", "不能为空") + return configError("Namespaces", "不能为空") } for name, namespace := range Spec.Namespaces { path := "Namespaces." + name if !configNameRegexp.MatchString(name) { - configError(path, "名称仅允许小写字母、数字和连字符") + return configError(path, "名称仅允许小写字母、数字和连字符") } if !isSafeRelativeObjectPath(namespace.Prefix) { - configError(path+".Prefix", "必须是安全的相对对象路径") + return configError(path+".Prefix", "必须是安全的相对对象路径") } if namespace.MaxSizeMB <= 0 { - configError(path+".MaxSizeMB", "必须大于 0") + return configError(path+".MaxSizeMB", "必须大于 0") } if len(namespace.AllowedExtensions) == 0 { - configError(path+".AllowedExtensions", "不能为空") + return configError(path+".AllowedExtensions", "不能为空") } for index, extension := range namespace.AllowedExtensions { extensionPath := path + ".AllowedExtensions[" + stringIndex(index) + "]" if !strings.HasPrefix(extension, ".") { - configError(extensionPath, "必须以 . 开头") + return configError(extensionPath, "必须以 . 开头") } namespace.AllowedExtensions[index] = strings.ToLower(extension) } @@ -105,25 +170,43 @@ func validate() { } if len(Spec.ServiceClients) == 0 { - configError("ServiceClients", "不能为空") + return configError("ServiceClients", "不能为空") } for name, secret := range Spec.ServiceClients { path := "ServiceClients." + name if !configNameRegexp.MatchString(name) { - configError(path, "服务名仅允许小写字母、数字和连字符") + return configError(path, "服务名仅允许小写字母、数字和连字符") + } + if err := validateRequired(path, secret); err != nil { + return err } - validateRequired(path, secret) } if Spec.Cleanup.IntervalSeconds <= 0 { - configError("Cleanup.IntervalSeconds", "必须大于 0") + return configError("Cleanup.IntervalSeconds", "必须大于 0") } + if Spec.Cleanup.IntegrityIntervalSeconds <= 0 { + return configError("Cleanup.IntegrityIntervalSeconds", "必须大于 0") + } + if Spec.Cleanup.IntegrityBatchSize <= 0 || Spec.Cleanup.IntegrityBatchSize > 1000 { + return configError("Cleanup.IntegrityBatchSize", "必须在 1 到 1000 之间") + } + return nil } -func validateRequired(path, value string) { - if strings.TrimSpace(value) == "" { - configError(path, "不能为空") +func validateHTTPURL(path, raw string) error { + parsed, err := url.Parse(strings.TrimSpace(raw)) + if err != nil || parsed.Host == "" || (parsed.Scheme != "http" && parsed.Scheme != "https") { + return configError(path, "必须是完整的 HTTP 或 HTTPS 地址") } + return nil +} + +func validateRequired(path, value string) error { + if strings.TrimSpace(value) == "" { + return configError(path, "不能为空") + } + return nil } func isSafeRelativeObjectPath(path string) bool { @@ -138,8 +221,8 @@ func isSafeRelativeObjectPath(path string) bool { return true } -func configError(path, message string) { - log.Fatalf("配置项 %s 无效：%s", path, message) +func configError(path, message string) error { + return fmt.Errorf("配置项 %s 无效：%s", path, message) } func stringIndex(index int) string { diff --git a/internal/errors/errors.go b/internal/errors/errors.go index d52ef63..9609620 100644 --- a/internal/errors/errors.go +++ b/internal/errors/errors.go @@ -3,15 +3,20 @@ package errors import "git.apinb.com/bsm-sdk/core/errcode" var ( - ErrInvalidParameter = errcode.NewError(2101, "参数错误") - ErrNamespaceNotAllowed = errcode.NewError(2102, "命名空间不允许") - ErrExtensionNotAllowed = errcode.NewError(2103, "扩展名不允许") - ErrInvalidFileSize = errcode.NewError(2104, "文件大小不合法") - ErrFileNotFound = errcode.NewError(2105, "文件不存在") - ErrUploadStatusConflict = errcode.NewError(2106, "上传状态冲突") - ErrUploadExpired = errcode.NewError(2107, "上传已过期") - ErrObjectInfoMismatch = errcode.NewError(2108, "OSS 对象信息不匹配") - ErrUnauthorizedOperation = errcode.NewError(2109, "无权操作") - ErrObjectStorageOperation = errcode.NewError(2110, "OSS 操作失败") - ErrDatabaseOperation = errcode.NewError(2111, "数据库操作失败") + ErrInvalidParameter = errcode.NewError(2101, "参数错误") + ErrNamespaceNotAllowed = errcode.NewError(2102, "命名空间不允许") + ErrExtensionNotAllowed = errcode.NewError(2103, "扩展名不允许") + ErrInvalidFileSize = errcode.NewError(2104, "文件大小不合法") + ErrFileNotFound = errcode.NewError(2105, "文件不存在") + ErrUploadStatusConflict = errcode.NewError(2106, "上传状态冲突") + ErrUploadExpired = errcode.NewError(2107, "上传已过期") + ErrObjectInfoMismatch = errcode.NewError(2108, "文件对象信息不匹配") + ErrUnauthorizedOperation = errcode.NewError(2109, "无权操作") + ErrStorageOperation = errcode.NewError(2110, "文件存储操作失败") + ErrDatabaseOperation = errcode.NewError(2111, "数据库操作失败") + ErrSignatureInvalid = errcode.NewError(2112, "文件访问签名无效或已过期") + ErrStorageSpaceLow = errcode.NewError(2113, "文件存储空间不足") + ErrVisibilityNotAllowed = errcode.NewError(2114, "该文件不允许公开访问") + ErrRangeNotSupported = errcode.NewError(2115, "不支持多段文件读取") + ErrUploadLocked = errcode.NewError(2116, "文件正在上传") ) diff --git a/internal/health/readiness.go b/internal/health/readiness.go new file mode 100644 index 0000000..e70c114 --- /dev/null +++ b/internal/health/readiness.go @@ -0,0 +1,143 @@ +package health + +import ( + "context" + "fmt" + "net/http" + "runtime/debug" + "strings" + "time" + + "git.apinb.com/ops/files/internal/config" + "git.apinb.com/ops/files/internal/impl" + "git.apinb.com/ops/files/internal/jobs" + "git.apinb.com/ops/files/internal/runtimeinfo" + "git.apinb.com/ops/files/internal/storage" + "github.com/gin-gonic/gin" + "gorm.io/gorm" +) + +type CheckResult struct { + Name string `json:"name"` + Required bool `json:"required"` + Ready bool `json:"ready"` + Message string `json:"message,omitempty"` +} +type Readiness struct { + Ready bool `json:"ready"` + CheckedAt time.Time `json:"checked_at"` + Checks []CheckResult `json:"checks,omitempty"` + Version string `json:"version"` +} + +type FileCounts struct { + Total int64 `json:"total"` + Ready int64 `json:"ready"` + Pending int64 `json:"pending"` +} + +func Evaluate(ctx context.Context) Readiness { + result, _ := evaluate(ctx) + return result +} + +func evaluate(ctx context.Context) (Readiness, storage.RuntimeStatus) { + result := Readiness{Ready: true, CheckedAt: time.Now().UTC(), Version: runtimeVersion()} + result.add("config", true, config.Validate()) + result.add("database", true, pingDatabase(ctx, impl.DBService)) + storageStatus, storageErr := impl.StorageService.Status(ctx) + if storageErr == nil && !storageStatus.Writable { + storageErr = fmt.Errorf("文件存储当前不可写") + } + result.add("storage", true, storageErr) + worker := jobs.Status() + if worker.Running { + result.add("pending_upload_cleanup", true, nil) + } else { + result.add("pending_upload_cleanup", true, fmt.Errorf("过期上传清理任务未运行")) + } + integrity := jobs.IntegrityStatusSnapshot() + if integrity.Running { + result.add("integrity_worker", true, nil) + } else { + result.add("integrity_worker", true, fmt.Errorf("文件完整性任务未运行")) + } + if integrity.Missing+integrity.SizeMismatch+integrity.ETagMismatch > 0 { + result.add("file_integrity", false, fmt.Errorf("发现文件缺失或校验不一致")) + } else { + result.add("file_integrity", false, nil) + } + return result, storageStatus +} + +func runtimeVersion() string { + info, ok := debug.ReadBuildInfo() + if !ok || strings.TrimSpace(info.Main.Version) == "" { + return "unknown" + } + return info.Main.Version +} +func Ready(c *gin.Context) { + result := Evaluate(c.Request.Context()) + status := http.StatusOK + if !result.Ready { + status = http.StatusServiceUnavailable + } + c.JSON(status, gin.H{"ready": result.Ready}) +} +func Status(c *gin.Context) { + result, storageStatus := evaluate(c.Request.Context()) + status := http.StatusOK + if !result.Ready { + status = http.StatusServiceUnavailable + } + counts, countErr := fileCounts(c.Request.Context()) + if countErr != nil { + result.add("file_ledger", true, countErr) + status = http.StatusServiceUnavailable + } + c.JSON(status, gin.H{ + "readiness": result, + "workers": gin.H{ + "pending_upload_cleanup": jobs.Status(), + "integrity": jobs.IntegrityStatusSnapshot(), + }, + "storage": storageStatus, + "disk": storageStatus.Disk, + "files": counts, + "activity": runtimeinfo.Snapshot(), + }) +} + +func fileCounts(ctx context.Context) (FileCounts, error) { + var counts FileCounts + err := impl.DBService.WithContext(ctx).Raw(` +SELECT COUNT(*) AS total, + COUNT(*) FILTER (WHERE status = 'ready') AS ready, + COUNT(*) FILTER (WHERE status = 'pending') AS pending +FROM files_object +WHERE deleted_at IS NULL`).Scan(&counts).Error + return counts, err +} +func (r *Readiness) add(name string, required bool, err error) { + check := CheckResult{Name: name, Required: required, Ready: err == nil} + if err != nil { + check.Message = strings.TrimSpace(err.Error()) + if required { + r.Ready = false + } + } + r.Checks = append(r.Checks, check) +} +func pingDatabase(ctx context.Context, db *gorm.DB) error { + if db == nil { + return fmt.Errorf("数据库未初始化") + } + sqlDB, err := db.DB() + if err != nil { + return err + } + checkCtx, cancel := context.WithTimeout(ctx, 2*time.Second) + defer cancel() + return sqlDB.PingContext(checkCtx) +} diff --git a/internal/impl/impl.go b/internal/impl/impl.go index 3779a70..b50f18e 100644 --- a/internal/impl/impl.go +++ b/internal/impl/impl.go @@ -1,6 +1,8 @@ package impl import ( + "errors" + "git.apinb.com/bsm-sdk/core/cache/redis" "git.apinb.com/bsm-sdk/core/logger" "git.apinb.com/bsm-sdk/core/types" @@ -8,6 +10,7 @@ import ( "git.apinb.com/bsm-sdk/core/with" "git.apinb.com/ops/files/internal/config" "git.apinb.com/ops/files/internal/storage" + "git.apinb.com/ops/files/internal/storage/factory" clientv3 "go.etcd.io/etcd/client/v3" "gorm.io/gorm" ) @@ -16,7 +19,7 @@ var ( RedisService *redis.RedisClient EtcdService *clientv3.Client DBService *gorm.DB - StorageService *storage.Client + StorageService storage.Backend Logger *logger.Logger ) @@ -31,7 +34,7 @@ func NewImpl() { }) var err error - StorageService, err = storage.New(config.Spec.ObjectStorage) + StorageService, err = factory.New(config.Spec.Storage) if err != nil { panic(err) } @@ -40,3 +43,26 @@ func NewImpl() { EtcdService = with.Etcd(config.Spec.Etcd) logger.New(nil) } + +// Close 关闭数据库、Redis 和 Etcd 连接。 +func Close() error { + var closeErrors []error + if DBService != nil { + if sqlDB, err := DBService.DB(); err != nil { + closeErrors = append(closeErrors, err) + } else if err := sqlDB.Close(); err != nil { + closeErrors = append(closeErrors, err) + } + } + if RedisService != nil && RedisService.Client != nil { + if err := RedisService.Client.Close(); err != nil { + closeErrors = append(closeErrors, err) + } + } + if EtcdService != nil { + if err := EtcdService.Close(); err != nil { + closeErrors = append(closeErrors, err) + } + } + return errors.Join(closeErrors...) +} diff --git a/internal/jobs/integrity.go b/internal/jobs/integrity.go new file mode 100644 index 0000000..aa35c35 --- /dev/null +++ b/internal/jobs/integrity.go @@ -0,0 +1,151 @@ +package jobs + +import ( + "context" + "errors" + "fmt" + "sync" + "time" + + "git.apinb.com/bsm-sdk/core/logger" + "git.apinb.com/ops/files/internal/config" + "git.apinb.com/ops/files/internal/impl" + "git.apinb.com/ops/files/internal/models" + "git.apinb.com/ops/files/internal/storage" +) + +type IntegrityStatus struct { + Running bool `json:"running"` + LastStarted time.Time `json:"last_started,omitempty"` + LastSuccess time.Time `json:"last_success,omitempty"` + LastError string `json:"last_error,omitempty"` + Checked int64 `json:"checked"` + Missing int64 `json:"missing"` + SizeMismatch int64 `json:"size_mismatch"` + ETagMismatch int64 `json:"etag_mismatch"` +} + +var integrityRuntime struct { + sync.RWMutex + IntegrityStatus + wg sync.WaitGroup +} + +func StartIntegrity(ctx context.Context) error { + integrityRuntime.Lock() + if integrityRuntime.Running { + integrityRuntime.Unlock() + return fmt.Errorf("文件完整性任务已启动") + } + integrityRuntime.Unlock() + if err := runIntegrity(ctx); err != nil { + return fmt.Errorf("首次核对文件完整性失败: %w", err) + } + integrityRuntime.Lock() + integrityRuntime.Running = true + integrityRuntime.Unlock() + integrityRuntime.wg.Add(1) + go func() { + defer integrityRuntime.wg.Done() + defer func() { + integrityRuntime.Lock() + integrityRuntime.Running = false + integrityRuntime.Unlock() + }() + ticker := time.NewTicker(time.Duration(config.Spec.Cleanup.IntegrityIntervalSeconds) * time.Second) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + if err := runIntegrity(ctx); err != nil && !errors.Is(err, context.Canceled) { + logger.Errorf("stage=files_integrity error=%v", err) + } + } + } + }() + return nil +} + +func IntegrityStatusSnapshot() IntegrityStatus { + integrityRuntime.RLock() + defer integrityRuntime.RUnlock() + return integrityRuntime.IntegrityStatus +} + +func waitIntegrity(ctx context.Context) error { + done := make(chan struct{}) + go func() { + integrityRuntime.wg.Wait() + close(done) + }() + select { + case <-done: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + +func runIntegrity(ctx context.Context) error { + startedAt := time.Now().UTC() + status := IntegrityStatus{Running: true, LastStarted: startedAt} + integrityRuntime.Lock() + status.LastSuccess = integrityRuntime.LastSuccess + integrityRuntime.IntegrityStatus = status + integrityRuntime.Unlock() + + var lastID uint + for { + var fileObjects []models.FileObject + result := impl.DBService.WithContext(ctx). + Where("status = ? AND id > ?", models.FileStatusReady, lastID). + Order("id ASC").Limit(config.Spec.Cleanup.IntegrityBatchSize).Find(&fileObjects) + if result.Error != nil { + return finishIntegrity(status, result.Error) + } + if len(fileObjects) == 0 { + break + } + for _, fileObject := range fileObjects { + if err := ctx.Err(); err != nil { + return finishIntegrity(status, err) + } + status.Checked++ + objectInfo, err := impl.StorageService.Stat(ctx, fileObject.ObjectKey) + if err != nil { + if errors.Is(err, storage.ErrObjectNotFound) { + status.Missing++ + continue + } + return finishIntegrity(status, err) + } + if objectInfo.Size != fileObject.ActualSize { + status.SizeMismatch++ + } + if objectInfo.StorageETag != fileObject.StorageETag { + status.ETagMismatch++ + } + } + lastID = fileObjects[len(fileObjects)-1].ID + if len(fileObjects) < config.Spec.Cleanup.IntegrityBatchSize { + break + } + } + status.LastSuccess = time.Now().UTC() + status.Running = integrityRuntime.Running + integrityRuntime.Lock() + integrityRuntime.IntegrityStatus = status + integrityRuntime.Unlock() + return nil +} + +func finishIntegrity(status IntegrityStatus, err error) error { + status.LastError = err.Error() + status.Running = integrityRuntime.Running + integrityRuntime.Lock() + integrityRuntime.IntegrityStatus = status + integrityRuntime.Unlock() + return err +} diff --git a/internal/jobs/pending_upload.go b/internal/jobs/pending_upload.go index 5bfd2a5..239b4f5 100644 --- a/internal/jobs/pending_upload.go +++ b/internal/jobs/pending_upload.go @@ -2,6 +2,9 @@ package jobs import ( "context" + "errors" + "fmt" + "sync" "time" "git.apinb.com/bsm-sdk/core/logger" @@ -9,24 +12,108 @@ import ( "git.apinb.com/ops/files/internal/impl" "git.apinb.com/ops/files/internal/lifecycle" "git.apinb.com/ops/files/internal/models" + "git.apinb.com/ops/files/internal/storage" ) -func StartPendingUploadCleanup() { +// CleanupStatus 描述过期上传清理任务的运行状态。 +type CleanupStatus struct { + Running bool `json:"running"` + LastStarted time.Time `json:"last_started,omitempty"` + LastSuccess time.Time `json:"last_success,omitempty"` + LastError string `json:"last_error,omitempty"` +} + +var cleanupRuntime struct { + sync.RWMutex + CleanupStatus + wg sync.WaitGroup +} + +// StartPendingUploadCleanup 启动可取消的过期上传清理任务。 +func StartPendingUploadCleanup(ctx context.Context) error { + cleanupRuntime.Lock() + if cleanupRuntime.Running { + cleanupRuntime.Unlock() + return fmt.Errorf("过期上传清理任务已启动") + } + cleanupRuntime.Unlock() + + if err := runCleanup(ctx); err != nil { + return fmt.Errorf("首次清理过期上传失败: %w", err) + } + + cleanupRuntime.Lock() + cleanupRuntime.Running = true + cleanupRuntime.Unlock() + cleanupRuntime.wg.Add(1) go func() { - cleanupPendingUploads() + defer cleanupRuntime.wg.Done() + defer func() { + cleanupRuntime.Lock() + cleanupRuntime.Running = false + cleanupRuntime.Unlock() + }() ticker := time.NewTicker(time.Duration(config.Spec.Cleanup.IntervalSeconds) * time.Second) defer ticker.Stop() - for range ticker.C { - cleanupPendingUploads() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + if err := runCleanup(ctx); err != nil && !errors.Is(err, context.Canceled) { + logger.Errorf("stage=cleanup error=%v", err) + } + } } }() + return nil } -func cleanupPendingUploads() { +// Status 返回清理任务状态快照。 +func Status() CleanupStatus { + cleanupRuntime.RLock() + defer cleanupRuntime.RUnlock() + return cleanupRuntime.CleanupStatus +} + +// Wait 等待清理任务退出。 +func Wait(ctx context.Context) error { + done := make(chan struct{}) + go func() { + cleanupRuntime.wg.Wait() + close(done) + }() + select { + case <-done: + return waitIntegrity(ctx) + case <-ctx.Done(): + return ctx.Err() + } +} + +func runCleanup(ctx context.Context) error { + startedAt := time.Now().UTC() + cleanupRuntime.Lock() + cleanupRuntime.LastStarted = startedAt + cleanupRuntime.Unlock() + + err := cleanupPendingUploads(ctx) + cleanupRuntime.Lock() + defer cleanupRuntime.Unlock() + if err != nil { + cleanupRuntime.LastError = err.Error() + return err + } + cleanupRuntime.LastSuccess = time.Now().UTC() + cleanupRuntime.LastError = "" + return nil +} + +func cleanupPendingUploads(ctx context.Context) error { var fileObjects []models.FileObject now := time.Now() - result := impl.DBService. + result := impl.DBService.WithContext(ctx). Where( "(status = ? AND expires_at <= ?) OR (status = ? AND (delete_lease_until IS NULL OR delete_lease_until <= ?))", models.FileStatusPending, @@ -38,10 +125,10 @@ func cleanupPendingUploads() { Limit(100). Find(&fileObjects) if result.Error != nil { - logger.Error("stage=scan") - return + return fmt.Errorf("扫描待清理文件失败: %w", result.Error) } + var cleanupErrors []error for _, fileObject := range fileObjects { var lease lifecycle.DeletionLease var claimed bool @@ -54,26 +141,37 @@ func cleanupPendingUploads() { } if err != nil { logger.Errorf("identity=%s stage=claim", fileObject.Identity) + cleanupErrors = append(cleanupErrors, fmt.Errorf("文件 %s 获取删除租约失败: %w", fileObject.Identity, err)) continue } if !claimed { continue } - cleanupClaimedFile(fileObject, lease) + if err := cleanupClaimedFile(ctx, fileObject, lease); err != nil { + cleanupErrors = append(cleanupErrors, err) + } } + return errors.Join(cleanupErrors...) } -func cleanupClaimedFile(fileObject models.FileObject, lease lifecycle.DeletionLease) { - operationCtx, cancel := context.WithTimeout(context.Background(), lifecycle.DeleteOperationTimeout) +func cleanupClaimedFile(ctx context.Context, fileObject models.FileObject, lease lifecycle.DeletionLease) error { + operationCtx, cancel := context.WithTimeout(ctx, lifecycle.DeleteOperationTimeout) defer cancel() + if receiver, ok := impl.StorageService.(storage.UploadReceiver); ok { + if err := receiver.DiscardUpload(operationCtx, fileObject.Identity); err != nil { + return fmt.Errorf("文件 %s 清理暂存内容失败: %w", fileObject.Identity, err) + } + } if err := impl.StorageService.Delete(operationCtx, fileObject.ObjectKey); err != nil { - logger.Errorf("identity=%s stage=oss_delete", fileObject.Identity) - return + logger.Errorf("identity=%s stage=storage_delete", fileObject.Identity) + return fmt.Errorf("文件 %s 删除对象失败: %w", fileObject.Identity, err) } if err := lifecycle.FinalizeDeletion(impl.DBService.WithContext(operationCtx), fileObject.ID, lease.Token); err != nil { logger.Errorf("identity=%s stage=finalize", fileObject.Identity) + return fmt.Errorf("文件 %s 完成删除失败: %w", fileObject.Identity, err) } + return nil } diff --git a/internal/logic/files/access.go b/internal/logic/files/access.go new file mode 100644 index 0000000..1af2d90 --- /dev/null +++ b/internal/logic/files/access.go @@ -0,0 +1,154 @@ +package files + +import ( + "errors" + "fmt" + "net/http" + "strconv" + "strings" + "time" + + "git.apinb.com/bsm-sdk/core/infra" + "git.apinb.com/ops/files/internal/auth" + "git.apinb.com/ops/files/internal/config" + fileerrors "git.apinb.com/ops/files/internal/errors" + "git.apinb.com/ops/files/internal/impl" + "git.apinb.com/ops/files/internal/models" + "git.apinb.com/ops/files/internal/signing" + "github.com/gin-gonic/gin" + "gorm.io/gorm" +) + +func Access(ctx *gin.Context) { + fileObject, ok := loadOwnedFile(ctx) + if !ok { + return + } + if fileObject.Status != models.FileStatusReady { + infra.Response.Error(ctx, fileerrors.ErrFileNotFound) + return + } + infra.Response.Success(ctx, newAccessInstruction(fileObject, normalizeDisposition(ctx.Query("disposition"), "inline"))) +} + +func BatchAccess(ctx *gin.Context) { + actor, ok := auth.FromContext(ctx) + if !ok || actor.Type != auth.ActorTypeService { + infra.Response.Error(ctx, fileerrors.ErrUnauthorizedOperation) + return + } + var request BatchAccessRequest + if err := ctx.ShouldBindJSON(&request); err != nil || len(request.FileIDs) == 0 || len(request.FileIDs) > 100 { + infra.Response.Error(ctx, fileerrors.ErrInvalidParameter) + return + } + fileIDs := uniqueFileIDs(request.FileIDs) + if len(fileIDs) != len(request.FileIDs) { + infra.Response.Error(ctx, fileerrors.ErrInvalidParameter) + return + } + var fileObjects []models.FileObject + if err := impl.DBService.Where( + "identity IN ? AND owner_type = ? AND owner_identity = ? AND status = ?", + fileIDs, auth.ActorTypeService, actor.Identity, models.FileStatusReady, + ).Find(&fileObjects).Error; err != nil { + infra.Response.Error(ctx, fileerrors.ErrDatabaseOperation) + return + } + if len(fileObjects) != len(fileIDs) { + infra.Response.Error(ctx, fileerrors.ErrFileNotFound) + return + } + byIdentity := make(map[string]models.FileObject, len(fileObjects)) + for _, fileObject := range fileObjects { + byIdentity[fileObject.Identity] = fileObject + } + disposition := normalizeDisposition(request.Disposition, "inline") + response := make([]AccessInstruction, 0, len(fileIDs)) + for _, fileID := range fileIDs { + response = append(response, newAccessInstruction(byIdentity[fileID], disposition)) + } + infra.Response.Success(ctx, response) +} + +func SignedAccess(ctx *gin.Context) { + fileID := strings.TrimSpace(ctx.Param("identity")) + disposition := normalizeDisposition(ctx.Query("disposition"), "inline") + expiresUnix, err := strconv.ParseInt(ctx.Query("expires"), 10, 64) + if err != nil || !signing.VerifyAccess( + config.Spec.Storage.SigningSecret, fileID, time.Unix(expiresUnix, 0), disposition, + ctx.Query("signature"), time.Now(), + ) { + ctx.JSON(http.StatusForbidden, gin.H{"message": "文件访问签名无效或已过期"}) + return + } + fileObject, ok := loadReadyFile(ctx, fileID, "") + if !ok { + return + } + streamFile(ctx, fileObject, disposition) +} + +func PublicAccess(ctx *gin.Context) { + fileObject, ok := loadReadyFile(ctx, strings.TrimSpace(ctx.Param("identity")), models.FileVisibilityPublic) + if !ok { + return + } + streamFile(ctx, fileObject, normalizeDisposition(ctx.Query("disposition"), "inline")) +} + +func newAccessInstruction(fileObject models.FileObject, disposition string) AccessInstruction { + baseURL := strings.TrimRight(config.Spec.Storage.ExternalBaseURL, "/") + result := AccessInstruction{ + FileID: fileObject.Identity, Filename: fileObject.OriginalName, ContentType: fileObject.ContentType, + Size: fileObject.ActualSize, Disposition: disposition, + } + if fileObject.Visibility == models.FileVisibilityPublic { + result.URL = fmt.Sprintf("%s/v1/public/files/%s/content?disposition=%s", baseURL, fileObject.Identity, disposition) + return result + } + expiresAt := time.Now().Add(time.Duration(config.Spec.Storage.AccessTTLSeconds) * time.Second).UTC() + signature := signing.SignAccess(config.Spec.Storage.SigningSecret, fileObject.Identity, expiresAt, disposition) + result.ExpiresAt = expiresAt + result.URL = fmt.Sprintf("%s/v1/access/%s/content?expires=%d&disposition=%s&signature=%s", + baseURL, fileObject.Identity, expiresAt.Unix(), disposition, signature) + return result +} + +func loadReadyFile(ctx *gin.Context, fileID string, visibility models.FileVisibility) (models.FileObject, bool) { + if fileID == "" { + ctx.Status(http.StatusNotFound) + return models.FileObject{}, false + } + query := impl.DBService.Where("identity = ? AND status = ?", fileID, models.FileStatusReady) + if visibility != "" { + query = query.Where("visibility = ?", visibility) + } + var fileObject models.FileObject + if err := query.First(&fileObject).Error; err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + ctx.Status(http.StatusNotFound) + } else { + ctx.Status(http.StatusInternalServerError) + } + return models.FileObject{}, false + } + return fileObject, true +} + +func uniqueFileIDs(values []string) []string { + seen := make(map[string]struct{}, len(values)) + result := make([]string, 0, len(values)) + for _, value := range values { + value = strings.TrimSpace(value) + if value == "" { + continue + } + if _, exists := seen[value]; exists { + continue + } + seen[value] = struct{}{} + result = append(result, value) + } + return result +} diff --git a/internal/logic/files/complete.go b/internal/logic/files/complete.go index 7736a19..5550564 100644 --- a/internal/logic/files/complete.go +++ b/internal/logic/files/complete.go @@ -7,10 +7,19 @@ import ( fileerrors "git.apinb.com/ops/files/internal/errors" "git.apinb.com/ops/files/internal/impl" "git.apinb.com/ops/files/internal/models" + "git.apinb.com/ops/files/internal/runtimeinfo" "github.com/gin-gonic/gin" ) func CompleteUpload(ctx *gin.Context) { + success := false + defer func() { + if success { + runtimeinfo.RecordUploadSuccess() + } else { + runtimeinfo.RecordUploadFailure() + } + }() fileObject, ok := loadOwnedFile(ctx) if !ok { return @@ -27,7 +36,7 @@ func CompleteUpload(ctx *gin.Context) { objectInfo, err := impl.StorageService.Stat(ctx.Request.Context(), fileObject.ObjectKey) if err != nil { - infra.Response.Error(ctx, fileerrors.ErrObjectStorageOperation) + infra.Response.Error(ctx, fileerrors.ErrStorageOperation) return } if objectInfo.Size != fileObject.ExpectedSize { @@ -41,7 +50,7 @@ func CompleteUpload(ctx *gin.Context) { Updates(map[string]any{ "status": models.FileStatusReady, "actual_size": objectInfo.Size, - "e_tag": objectInfo.ETag, + "storage_etag": objectInfo.StorageETag, "completed_at": completedAt, }) if result.Error != nil { @@ -55,7 +64,8 @@ func CompleteUpload(ctx *gin.Context) { fileObject.Status = models.FileStatusReady fileObject.ActualSize = objectInfo.Size - fileObject.ETag = objectInfo.ETag + fileObject.StorageETag = objectInfo.StorageETag fileObject.CompletedAt = &completedAt + success = true infra.Response.Success(ctx, newFileResponse(fileObject)) } diff --git a/internal/logic/files/delete.go b/internal/logic/files/delete.go index ad6a9c0..d3796ea 100644 --- a/internal/logic/files/delete.go +++ b/internal/logic/files/delete.go @@ -9,6 +9,7 @@ import ( "git.apinb.com/ops/files/internal/impl" "git.apinb.com/ops/files/internal/lifecycle" "git.apinb.com/ops/files/internal/models" + "git.apinb.com/ops/files/internal/runtimeinfo" "github.com/gin-gonic/gin" ) @@ -37,14 +38,17 @@ func Delete(ctx *gin.Context) { defer cancel() if err := impl.StorageService.Delete(operationCtx, fileObject.ObjectKey); err != nil { - infra.Response.Error(ctx, fileerrors.ErrObjectStorageOperation) + runtimeinfo.RecordDeleteFailure() + infra.Response.Error(ctx, fileerrors.ErrStorageOperation) return } if err := lifecycle.FinalizeDeletion(impl.DBService.WithContext(operationCtx), fileObject.ID, lease.Token); err != nil { + runtimeinfo.RecordDeleteFailure() infra.Response.Error(ctx, fileerrors.ErrDatabaseOperation) return } + runtimeinfo.RecordDeleteSuccess() infra.Response.Success(ctx, nil) } diff --git a/internal/logic/files/detail.go b/internal/logic/files/detail.go index 2921997..536cb47 100644 --- a/internal/logic/files/detail.go +++ b/internal/logic/files/detail.go @@ -26,6 +26,19 @@ func Detail(ctx *gin.Context) { infra.Response.Success(ctx, newFileResponse(fileObject)) } +// Download 通过文件服务鉴权读取文件内容。 +func Download(ctx *gin.Context) { + fileObject, ok := loadOwnedFile(ctx) + if !ok { + return + } + if fileObject.Status != models.FileStatusReady { + infra.Response.Error(ctx, fileerrors.ErrFileNotFound) + return + } + streamFile(ctx, fileObject, normalizeDisposition(ctx.Query("disposition"), "attachment")) +} + func loadOwnedFile(ctx *gin.Context) (models.FileObject, bool) { actor, ok := auth.FromContext(ctx) if !ok { @@ -57,12 +70,10 @@ func loadOwnedFile(ctx *gin.Context) (models.FileObject, bool) { func newFileResponse(fileObject models.FileObject) FileResponse { return FileResponse{ FileID: fileObject.Identity, - ObjectKey: fileObject.ObjectKey, - URL: impl.StorageService.PublicURL(fileObject.ObjectKey), Filename: fileObject.OriginalName, Size: fileObject.ActualSize, ContentType: fileObject.ContentType, - ETag: fileObject.ETag, + StorageETag: fileObject.StorageETag, Status: fileObject.Status, } } diff --git a/internal/logic/files/init.go b/internal/logic/files/init.go index 280fd75..9f8723d 100644 --- a/internal/logic/files/init.go +++ b/internal/logic/files/init.go @@ -1,6 +1,7 @@ package files import ( + "errors" "math" "path/filepath" "strings" @@ -14,6 +15,7 @@ import ( fileerrors "git.apinb.com/ops/files/internal/errors" "git.apinb.com/ops/files/internal/impl" "git.apinb.com/ops/files/internal/models" + "git.apinb.com/ops/files/internal/storage" "github.com/gin-gonic/gin" ) @@ -54,34 +56,43 @@ func InitUpload(ctx *gin.Context) { } contentType := strings.TrimSpace(request.ContentType) - if contentType == "" || utf8.RuneCountInString(contentType) > 255 { + if contentType == "" || utf8.RuneCountInString(contentType) > 255 || strings.ContainsAny(contentType, "\r\n") { infra.Response.Error(ctx, fileerrors.ErrInvalidParameter) return } fileID := utils.ULID() objectKey := buildObjectKey(namespaceConfig.Prefix, time.Now(), fileID, extension) - upload, err := impl.StorageService.PresignPut(ctx.Request.Context(), objectKey, contentType) + expiresAt := time.Now().Add(time.Duration(config.Spec.Storage.AccessTTLSeconds) * time.Second) + upload, err := impl.StorageService.PrepareUpload(ctx.Request.Context(), storage.UploadRequest{ + FileID: fileID, ObjectKey: objectKey, ContentType: contentType, + ExpectedSize: request.Size, ExpiresAt: expiresAt, + }) if err != nil { - infra.Response.Error(ctx, fileerrors.ErrObjectStorageOperation) + if errors.Is(err, storage.ErrInsufficientSpace) { + infra.Response.Error(ctx, fileerrors.ErrStorageSpaceLow) + } else { + infra.Response.Error(ctx, fileerrors.ErrStorageOperation) + } return } fileObject := models.FileObject{ - Identity: fileID, - Namespace: namespace, - Provider: config.Spec.ObjectStorage.Provider, - Bucket: impl.StorageService.Bucket(), - ObjectKey: objectKey, - OriginalName: filename, - Extension: extension, - ContentType: contentType, - ExpectedSize: request.Size, - Status: models.FileStatusPending, - OwnerType: actor.Type, - OwnerID: actor.ID, - OwnerIdentity: actor.Identity, - ExpiresAt: &upload.ExpiresAt, + Identity: fileID, + Namespace: namespace, + Provider: impl.StorageService.Provider(), + StorageContainer: impl.StorageService.Container(), + ObjectKey: objectKey, + OriginalName: filename, + Extension: extension, + ContentType: contentType, + ExpectedSize: request.Size, + Status: models.FileStatusPending, + Visibility: models.FileVisibilityPrivate, + OwnerType: actor.Type, + OwnerID: actor.ID, + OwnerIdentity: actor.Identity, + ExpiresAt: &upload.ExpiresAt, } if err := impl.DBService.Create(&fileObject).Error; err != nil { infra.Response.Error(ctx, fileerrors.ErrDatabaseOperation) @@ -89,9 +100,8 @@ func InitUpload(ctx *gin.Context) { } infra.Response.Success(ctx, InitUploadResponse{ - FileID: fileID, - ObjectKey: objectKey, - Upload: upload, + FileID: fileID, + Upload: upload, }) } diff --git a/internal/logic/files/local_upload.go b/internal/logic/files/local_upload.go new file mode 100644 index 0000000..523ce03 --- /dev/null +++ b/internal/logic/files/local_upload.go @@ -0,0 +1,78 @@ +package files + +import ( + "errors" + "net/http" + "strconv" + "strings" + "time" + + "git.apinb.com/ops/files/internal/config" + "git.apinb.com/ops/files/internal/impl" + "git.apinb.com/ops/files/internal/models" + "git.apinb.com/ops/files/internal/signing" + "git.apinb.com/ops/files/internal/storage" + "github.com/gin-gonic/gin" + "gorm.io/gorm" +) + +func LocalUpload(ctx *gin.Context) { + receiver, ok := impl.StorageService.(storage.UploadReceiver) + if !ok || impl.StorageService.Provider() != "local" { + ctx.Status(http.StatusNotFound) + return + } + fileID := strings.TrimSpace(ctx.Param("identity")) + expiresUnix, err := strconv.ParseInt(ctx.Query("expires"), 10, 64) + if err != nil || fileID == "" { + ctx.Status(http.StatusForbidden) + return + } + var fileObject models.FileObject + if err := impl.DBService.Where("identity = ?", fileID).First(&fileObject).Error; err != nil { + ctx.Status(http.StatusForbidden) + return + } + request := storage.UploadRequest{ + FileID: fileID, ObjectKey: fileObject.ObjectKey, ContentType: fileObject.ContentType, + ExpectedSize: fileObject.ExpectedSize, ExpiresAt: time.Unix(expiresUnix, 0), + } + if !signing.VerifyUpload(config.Spec.Storage.SigningSecret, ctx.Query("signature"), request, time.Now()) || + ctx.GetHeader("Content-Type") != fileObject.ContentType || ctx.Request.ContentLength != fileObject.ExpectedSize { + ctx.Status(http.StatusForbidden) + return + } + release, err := receiver.AcquireUpload(ctx.Request.Context(), fileID) + if err != nil { + if errors.Is(err, storage.ErrUploadLocked) { + ctx.Status(http.StatusConflict) + } else { + ctx.Status(http.StatusInternalServerError) + } + return + } + defer release() + + if err := impl.DBService.Where("id = ?", fileObject.ID).First(&fileObject).Error; err != nil { + ctx.Status(http.StatusConflict) + return + } + if fileObject.Status != models.FileStatusPending || fileObject.ExpiresAt == nil || !fileObject.ExpiresAt.After(time.Now()) { + ctx.Status(http.StatusConflict) + return + } + request.ObjectKey = fileObject.ObjectKey + request.ContentType = fileObject.ContentType + request.ExpectedSize = fileObject.ExpectedSize + if _, err := receiver.ReceiveUpload(ctx.Request.Context(), request, ctx.Request.Body); err != nil { + if errors.Is(err, storage.ErrInsufficientSpace) { + ctx.Status(http.StatusInsufficientStorage) + } else if errors.Is(err, gorm.ErrRecordNotFound) { + ctx.Status(http.StatusConflict) + } else { + ctx.Status(http.StatusInternalServerError) + } + return + } + ctx.Status(http.StatusNoContent) +} diff --git a/internal/logic/files/range.go b/internal/logic/files/range.go new file mode 100644 index 0000000..88d866f --- /dev/null +++ b/internal/logic/files/range.go @@ -0,0 +1,52 @@ +package files + +import ( + "fmt" + "strconv" + "strings" + + "git.apinb.com/ops/files/internal/storage" +) + +func parseByteRange(value string, size int64) (*storage.ByteRange, error) { + value = strings.TrimSpace(value) + if value == "" { + return nil, nil + } + if size <= 0 || !strings.HasPrefix(value, "bytes=") { + return nil, fmt.Errorf("文件读取范围无效") + } + rangeValue := strings.TrimPrefix(value, "bytes=") + if strings.Contains(rangeValue, ",") { + return nil, fmt.Errorf("不支持多段文件读取") + } + parts := strings.SplitN(rangeValue, "-", 2) + if len(parts) != 2 { + return nil, fmt.Errorf("文件读取范围无效") + } + if parts[0] == "" { + suffixLength, err := strconv.ParseInt(parts[1], 10, 64) + if err != nil || suffixLength <= 0 { + return nil, fmt.Errorf("文件读取范围无效") + } + if suffixLength > size { + suffixLength = size + } + return &storage.ByteRange{Start: size - suffixLength, End: size - 1}, nil + } + start, err := strconv.ParseInt(parts[0], 10, 64) + if err != nil || start < 0 || start >= size { + return nil, fmt.Errorf("文件读取范围无效") + } + end := size - 1 + if parts[1] != "" { + end, err = strconv.ParseInt(parts[1], 10, 64) + if err != nil || end < start { + return nil, fmt.Errorf("文件读取范围无效") + } + if end >= size { + end = size - 1 + } + } + return &storage.ByteRange{Start: start, End: end}, nil +} diff --git a/internal/logic/files/stream.go b/internal/logic/files/stream.go new file mode 100644 index 0000000..b09f042 --- /dev/null +++ b/internal/logic/files/stream.go @@ -0,0 +1,59 @@ +package files + +import ( + "fmt" + "mime" + "net/http" + "strings" + + "git.apinb.com/ops/files/internal/impl" + "git.apinb.com/ops/files/internal/models" + "git.apinb.com/ops/files/internal/runtimeinfo" + "github.com/gin-gonic/gin" +) + +func streamFile(ctx *gin.Context, fileObject models.FileObject, disposition string) { + byteRange, err := parseByteRange(ctx.GetHeader("Range"), fileObject.ActualSize) + if err != nil { + ctx.Header("Content-Range", fmt.Sprintf("bytes */%d", fileObject.ActualSize)) + ctx.JSON(http.StatusRequestedRangeNotSatisfiable, gin.H{"message": err.Error()}) + return + } + body, err := impl.StorageService.Open(ctx.Request.Context(), fileObject.ObjectKey, byteRange) + if err != nil { + runtimeinfo.RecordReadFailure() + ctx.JSON(http.StatusNotFound, gin.H{"message": "文件不存在"}) + return + } + defer body.Close() + + status := http.StatusOK + length := fileObject.ActualSize + if byteRange != nil { + status = http.StatusPartialContent + length = byteRange.Length() + ctx.Header("Content-Range", fmt.Sprintf("bytes %d-%d/%d", byteRange.Start, byteRange.End, fileObject.ActualSize)) + } + ctx.Header("Accept-Ranges", "bytes") + ctx.Header("Cache-Control", "no-store") + if fileObject.StorageETag != "" { + ctx.Header("ETag", fmt.Sprintf("\"%s\"", strings.ReplaceAll(fileObject.StorageETag, "\"", ""))) + } + if fileObject.CompletedAt != nil { + ctx.Header("Last-Modified", fileObject.CompletedAt.UTC().Format(http.TimeFormat)) + } + contentDisposition := mime.FormatMediaType(disposition, map[string]string{"filename": fileObject.OriginalName}) + ctx.Header("Content-Disposition", contentDisposition) + ctx.DataFromReader(status, length, fileObject.ContentType, body, nil) +} + +func normalizeDisposition(value, defaultValue string) string { + switch strings.ToLower(strings.TrimSpace(value)) { + case "inline": + return "inline" + case "attachment": + return "attachment" + default: + return defaultValue + } +} diff --git a/internal/logic/files/types.go b/internal/logic/files/types.go index 4b44c97..a9af314 100644 --- a/internal/logic/files/types.go +++ b/internal/logic/files/types.go @@ -3,6 +3,7 @@ package files import ( "git.apinb.com/ops/files/internal/models" "git.apinb.com/ops/files/internal/storage" + "time" ) type InitUploadRequest struct { @@ -13,18 +14,34 @@ type InitUploadRequest struct { } type InitUploadResponse struct { - FileID string `json:"file_id"` - ObjectKey string `json:"object_key"` - Upload storage.UploadInstruction `json:"upload"` + FileID string `json:"file_id"` + Upload storage.UploadInstruction `json:"upload"` } type FileResponse struct { FileID string `json:"file_id"` - ObjectKey string `json:"object_key"` - URL string `json:"url"` Filename string `json:"filename"` Size int64 `json:"size"` ContentType string `json:"content_type"` - ETag string `json:"etag"` + StorageETag string `json:"storage_etag"` Status models.FileStatus `json:"status"` } + +type AccessInstruction struct { + FileID string `json:"file_id"` + URL string `json:"url"` + Filename string `json:"filename"` + ContentType string `json:"content_type"` + Size int64 `json:"size"` + Disposition string `json:"disposition"` + ExpiresAt time.Time `json:"expires_at,omitempty"` +} + +type BatchAccessRequest struct { + FileIDs []string `json:"file_ids" binding:"required"` + Disposition string `json:"disposition"` +} + +type SetVisibilityRequest struct { + Visibility models.FileVisibility `json:"visibility" binding:"required"` +} diff --git a/internal/logic/files/visibility.go b/internal/logic/files/visibility.go new file mode 100644 index 0000000..2d3d0c1 --- /dev/null +++ b/internal/logic/files/visibility.go @@ -0,0 +1,39 @@ +package files + +import ( + "git.apinb.com/bsm-sdk/core/infra" + fileerrors "git.apinb.com/ops/files/internal/errors" + "git.apinb.com/ops/files/internal/impl" + "git.apinb.com/ops/files/internal/models" + "github.com/gin-gonic/gin" +) + +func SetVisibility(ctx *gin.Context) { + fileObject, ok := loadOwnedFile(ctx) + if !ok { + return + } + var request SetVisibilityRequest + if err := ctx.ShouldBindJSON(&request); err != nil || + (request.Visibility != models.FileVisibilityPrivate && request.Visibility != models.FileVisibilityPublic) { + infra.Response.Error(ctx, fileerrors.ErrInvalidParameter) + return + } + if fileObject.Namespace != "visual" || fileObject.Status != models.FileStatusReady { + infra.Response.Error(ctx, fileerrors.ErrVisibilityNotAllowed) + return + } + result := impl.DBService.Model(&models.FileObject{}). + Where("id = ? AND status = ?", fileObject.ID, models.FileStatusReady). + Update("visibility", request.Visibility) + if result.Error != nil { + infra.Response.Error(ctx, fileerrors.ErrDatabaseOperation) + return + } + if result.RowsAffected != 1 { + infra.Response.Error(ctx, fileerrors.ErrUploadStatusConflict) + return + } + fileObject.Visibility = request.Visibility + infra.Response.Success(ctx, newFileResponse(fileObject)) +} diff --git a/internal/models/file_object.go b/internal/models/file_object.go index 8c0e7c4..d914176 100644 --- a/internal/models/file_object.go +++ b/internal/models/file_object.go @@ -7,6 +7,7 @@ import ( ) type FileStatus string +type FileVisibility string const ( FileStatusPending FileStatus = "pending" @@ -15,20 +16,26 @@ const ( FileStatusExpired FileStatus = "expired" ) +const ( + FileVisibilityPrivate FileVisibility = "private" + FileVisibilityPublic FileVisibility = "public" +) + type FileObject struct { ID uint `gorm:"primaryKey" json:"id"` Identity string `gorm:"size:64;not null;uniqueIndex" json:"identity"` Namespace string `gorm:"size:64;not null" json:"namespace"` Provider string `gorm:"size:32;not null" json:"provider"` - Bucket string `gorm:"size:255;not null" json:"bucket"` - ObjectKey string `gorm:"size:512;not null;uniqueIndex" json:"object_key"` + StorageContainer string `gorm:"size:255;not null" json:"-"` + ObjectKey string `gorm:"size:512;not null;uniqueIndex" json:"-"` OriginalName string `gorm:"size:255;not null" json:"original_name"` Extension string `gorm:"size:32" json:"extension"` ContentType string `gorm:"size:255" json:"content_type"` ExpectedSize int64 `gorm:"not null" json:"expected_size"` ActualSize int64 `json:"actual_size"` - ETag string `gorm:"size:255" json:"etag"` - Status FileStatus `gorm:"size:16;not null;index:idx_files_object_status_expires_at,priority:1;index:idx_files_object_status_delete_lease,priority:1" json:"status"` + StorageETag string `gorm:"column:storage_etag;size:255" json:"storage_etag"` + Visibility FileVisibility `gorm:"size:16;not null;default:private;index:idx_files_object_visibility_status,priority:1" json:"visibility"` + Status FileStatus `gorm:"size:16;not null;index:idx_files_object_status_expires_at,priority:1;index:idx_files_object_status_delete_lease,priority:1;index:idx_files_object_visibility_status,priority:2" json:"status"` OwnerType string `gorm:"size:64;not null;index:idx_files_object_owner_type_identity,priority:1" json:"owner_type"` OwnerID uint `gorm:"not null" json:"owner_id"` OwnerIdentity string `gorm:"size:64;not null;index:idx_files_object_owner_type_identity,priority:2" json:"owner_identity"` diff --git a/internal/models/schema_version.go b/internal/models/schema_version.go new file mode 100644 index 0000000..0b37408 --- /dev/null +++ b/internal/models/schema_version.go @@ -0,0 +1,66 @@ +package models + +import ( + "fmt" + + "gorm.io/gorm" +) + +const requiredSchemaScript = "0001_baseline.up.sql" + +var requiredSchema = struct { + checksum string + tables []string +}{ + checksum: "ee621597df0d1c80e98cf38b941c1ac00e62f570223b9818bdab20c4e6f96689", + tables: []string{"files_object"}, +} + +// RequireSchemaVersion 确认当前数据库已经由受控迁移脚本初始化。 +func RequireSchemaVersion(db *gorm.DB, service string) error { + if db == nil { + return fmt.Errorf("%s 数据库连接未初始化", service) + } + + var count int64 + err := db.Raw(` +SELECT COUNT(*) +FROM public.ops_schema_migrations +WHERE service = ? AND script_name = ? AND checksum = ? AND status = 'success'`, + service, requiredSchemaScript, requiredSchema.checksum).Scan(&count).Error + if err != nil { + return fmt.Errorf("%s 数据库版本检查失败，请先执行 migrate-all.sh up --service %s: %w", service, service, err) + } + if count != 1 { + return fmt.Errorf("%s 数据库迁移版本不匹配，请执行 migrate-all.sh status --service %s", service, service) + } + + for _, table := range requiredSchema.tables { + var valid bool + err = db.Raw(` +SELECT to_regclass(?) IS NOT NULL AND EXISTS ( + SELECT 1 + FROM pg_constraint c + JOIN pg_class t ON t.oid = c.conrelid + JOIN pg_namespace n ON n.oid = t.relnamespace + WHERE n.nspname = 'public' AND t.relname = ? AND c.contype = 'p' +)`, "public."+table, table).Scan(&valid).Error + if err != nil || !valid { + return fmt.Errorf("%s 数据库关键表结构不完整：%s", service, table) + } + } + return nil +} + +func RequireStorageProvider(db *gorm.DB, provider string) error { + var providers []string + if err := db.Model(&FileObject{}).Distinct("provider").Pluck("provider", &providers).Error; err != nil { + return fmt.Errorf("检查文件存储类型失败: %w", err) + } + for _, storedProvider := range providers { + if storedProvider != provider { + return fmt.Errorf("数据库存在 %s 存储的文件，当前配置为 %s，禁止混用存储后端", storedProvider, provider) + } + } + return nil +} diff --git a/internal/routers/register.go b/internal/routers/register.go index 44892fd..6a73171 100644 --- a/internal/routers/register.go +++ b/internal/routers/register.go @@ -5,6 +5,7 @@ import ( "git.apinb.com/bsm-sdk/core/middleware" "git.apinb.com/ops/files/internal/auth" + "git.apinb.com/ops/files/internal/health" "git.apinb.com/ops/files/internal/logic/files" "git.apinb.com/ops/files/internal/logic/ping" "github.com/gin-gonic/gin" @@ -13,15 +14,28 @@ import ( // Register 注册基础服务路由。 func Register(srvKey string, engine *gin.Engine) { v1Group := engine.Group(fmt.Sprintf("/%s/v1", srvKey)) + engine.GET("/ready", health.Ready) v1Group.GET("/ping/hello", ping.Hello) + v1Group.GET("/runtime/status", middleware.JwtAuth(true), health.Status) + v1Group.PUT("/local/uploads/:identity", files.LocalUpload) + v1Group.GET("/access/:identity/content", files.SignedAccess) + v1Group.GET("/public/files/:identity/content", files.PublicAccess) - registerFileRoutes(v1Group.Group("", middleware.JwtAuth(true))) - registerFileRoutes(v1Group.Group("/internal", auth.ServiceAuth())) + userGroup := v1Group.Group("", middleware.JwtAuth(true)) + registerFileRoutes(userGroup) + userGroup.GET("/files/:identity/access", files.Access) + + internalGroup := v1Group.Group("/internal", auth.ServiceAuth()) + internalGroup.GET("/runtime/status", health.Status) + registerFileRoutes(internalGroup) + internalGroup.POST("/files/access", files.BatchAccess) + internalGroup.PUT("/files/:identity/visibility", files.SetVisibility) } func registerFileRoutes(group *gin.RouterGroup) { group.POST("/uploads/init", files.InitUpload) group.POST("/uploads/:identity/complete", files.CompleteUpload) group.GET("/files/:identity", files.Detail) + group.GET("/files/:identity/content", files.Download) group.DELETE("/files/:identity", files.Delete) } diff --git a/internal/runtimeinfo/storage.go b/internal/runtimeinfo/storage.go new file mode 100644 index 0000000..1f2f785 --- /dev/null +++ b/internal/runtimeinfo/storage.go @@ -0,0 +1,67 @@ +package runtimeinfo + +import ( + "sync/atomic" + "time" +) + +type Activity struct { + UploadSuccess uint64 `json:"upload_success"` + UploadFailure uint64 `json:"upload_failure"` + ReadFailure uint64 `json:"read_failure"` + DeleteSuccess uint64 `json:"delete_success"` + DeleteFailure uint64 `json:"delete_failure"` + LastUploadAt time.Time `json:"last_upload_at,omitempty"` + LastReadFailAt time.Time `json:"last_read_failure_at,omitempty"` + LastDeleteAt time.Time `json:"last_delete_at,omitempty"` +} + +var counters struct { + uploadSuccess atomic.Uint64 + uploadFailure atomic.Uint64 + readFailure atomic.Uint64 + deleteSuccess atomic.Uint64 + deleteFailure atomic.Uint64 + lastUpload atomic.Int64 + lastReadFail atomic.Int64 + lastDelete atomic.Int64 +} + +func RecordUploadSuccess() { + counters.uploadSuccess.Add(1) + counters.lastUpload.Store(time.Now().UTC().UnixMilli()) +} + +func RecordUploadFailure() { counters.uploadFailure.Add(1) } + +func RecordReadFailure() { + counters.readFailure.Add(1) + counters.lastReadFail.Store(time.Now().UTC().UnixMilli()) +} + +func RecordDeleteSuccess() { + counters.deleteSuccess.Add(1) + counters.lastDelete.Store(time.Now().UTC().UnixMilli()) +} + +func RecordDeleteFailure() { counters.deleteFailure.Add(1) } + +func Snapshot() Activity { + return Activity{ + UploadSuccess: counters.uploadSuccess.Load(), + UploadFailure: counters.uploadFailure.Load(), + ReadFailure: counters.readFailure.Load(), + DeleteSuccess: counters.deleteSuccess.Load(), + DeleteFailure: counters.deleteFailure.Load(), + LastUploadAt: timestamp(counters.lastUpload.Load()), + LastReadFailAt: timestamp(counters.lastReadFail.Load()), + LastDeleteAt: timestamp(counters.lastDelete.Load()), + } +} + +func timestamp(milliseconds int64) time.Time { + if milliseconds <= 0 { + return time.Time{} + } + return time.UnixMilli(milliseconds).UTC() +} diff --git a/internal/signing/signature.go b/internal/signing/signature.go new file mode 100644 index 0000000..c753db6 --- /dev/null +++ b/internal/signing/signature.go @@ -0,0 +1,64 @@ +package signing + +import ( + "crypto/hmac" + "crypto/sha256" + "crypto/subtle" + "encoding/hex" + "fmt" + "strconv" + "strings" + "time" + + "git.apinb.com/ops/files/internal/storage" +) + +func SignUpload(secret string, request storage.UploadRequest) string { + canonical := strings.Join([]string{ + "PUT", + request.FileID, + strconv.FormatInt(request.ExpiresAt.Unix(), 10), + strconv.FormatInt(request.ExpectedSize, 10), + request.ContentType, + }, "\n") + return sign(secret, canonical) +} + +func VerifyUpload(secret, signature string, request storage.UploadRequest, now time.Time) bool { + return request.ExpiresAt.After(now) && verify(secret, SignUpload(secret, request), signature) +} + +func SignAccess(secret, fileID string, expiresAt time.Time, disposition string) string { + canonical := strings.Join([]string{"GET", fileID, strconv.FormatInt(expiresAt.Unix(), 10), disposition}, "\n") + return sign(secret, canonical) +} + +func VerifyAccess(secret, fileID string, expiresAt time.Time, disposition, signature string, now time.Time) bool { + return expiresAt.After(now) && verify(secret, SignAccess(secret, fileID, expiresAt, disposition), signature) +} + +func sign(secret, canonical string) string { + mac := hmac.New(sha256.New, []byte(secret)) + _, _ = mac.Write([]byte(canonical)) + return hex.EncodeToString(mac.Sum(nil)) +} + +func verify(secret, expected, actual string) bool { + if strings.TrimSpace(secret) == "" || len(actual) != sha256.Size*2 { + return false + } + expectedBytes, expectedErr := hex.DecodeString(expected) + actualBytes, actualErr := hex.DecodeString(actual) + if expectedErr != nil || actualErr != nil { + return false + } + return subtle.ConstantTimeCompare(expectedBytes, actualBytes) == 1 +} + +func UploadNonce(secret string, request storage.UploadRequest) (string, error) { + signature := SignUpload(secret, request) + if len(signature) < 16 { + return "", fmt.Errorf("上传签名长度无效") + } + return signature[:16], nil +} diff --git a/internal/storage/client.go b/internal/storage/aliyun/client.go similarity index 56% rename from internal/storage/client.go rename to internal/storage/aliyun/client.go index c97c18f..02405d5 100644 --- a/internal/storage/client.go +++ b/internal/storage/aliyun/client.go @@ -1,21 +1,18 @@ -package storage +package aliyun import ( - "time" - "git.apinb.com/ops/files/internal/config" "github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss" "github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/credentials" ) type Client struct { - client *oss.Client - bucket string - publicBaseURL string - presignTTL time.Duration + client *oss.Client + bucket string + presignTTL int64 } -func New(cfg config.ObjectStorageConf) (*Client, error) { +func New(cfg config.AliyunStorageConf, accessTTLSeconds int64) (*Client, error) { ossConfig := oss.LoadDefaultConfig(). WithRegion(cfg.Region). WithEndpoint(cfg.Endpoint). @@ -23,9 +20,12 @@ func New(cfg config.ObjectStorageConf) (*Client, error) { WithSignatureVersion(oss.SignatureVersionV4) return &Client{ - client: oss.NewClient(ossConfig), - bucket: cfg.Bucket, - publicBaseURL: cfg.PublicBaseURL, - presignTTL: time.Duration(cfg.PresignTTLSeconds) * time.Second, + client: oss.NewClient(ossConfig), + bucket: cfg.Bucket, + presignTTL: accessTTLSeconds, }, nil } + +func (c *Client) Provider() string { return "aliyun" } + +func (c *Client) Container() string { return c.bucket } diff --git a/internal/storage/aliyun/object.go b/internal/storage/aliyun/object.go new file mode 100644 index 0000000..6fa9a7b --- /dev/null +++ b/internal/storage/aliyun/object.go @@ -0,0 +1,99 @@ +package aliyun + +import ( + "context" + "errors" + "fmt" + "io" + "strings" + + "git.apinb.com/ops/files/internal/storage" + "github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss" +) + +// Open 返回对象存储中的只读内容流。 +func (c *Client) Open(ctx context.Context, objectKey string, byteRange *storage.ByteRange) (io.ReadCloser, error) { + if strings.TrimSpace(objectKey) == "" { + return nil, fmt.Errorf("对象键不能为空") + } + request := &oss.GetObjectRequest{ + Bucket: oss.Ptr(c.bucket), + Key: oss.Ptr(objectKey), + } + if byteRange != nil { + rangeHeader := fmt.Sprintf("bytes=%d-%d", byteRange.Start, byteRange.End) + request.Range = &rangeHeader + } + result, err := c.client.GetObject(ctx, request) + if err != nil { + return nil, fmt.Errorf("读取对象: %w", err) + } + if result == nil || result.Body == nil { + return nil, fmt.Errorf("对象内容为空") + } + return result.Body, nil +} + +func (c *Client) Stat(ctx context.Context, objectKey string) (storage.ObjectInfo, error) { + if strings.TrimSpace(objectKey) == "" { + return storage.ObjectInfo{}, fmt.Errorf("对象键不能为空") + } + + result, err := c.client.GetObjectMeta(ctx, &oss.GetObjectMetaRequest{ + Bucket: oss.Ptr(c.bucket), + Key: oss.Ptr(objectKey), + }) + if err != nil { + var serviceError *oss.ServiceError + if errors.As(err, &serviceError) && serviceError.Code == "NoSuchKey" { + return storage.ObjectInfo{}, storage.ErrObjectNotFound + } + return storage.ObjectInfo{}, fmt.Errorf("查询对象元数据: %w", err) + } + if result == nil { + return storage.ObjectInfo{}, fmt.Errorf("对象元数据为空") + } + if result.ContentLength < 0 { + return storage.ObjectInfo{}, fmt.Errorf("对象大小无效: %d", result.ContentLength) + } + if result.ETag == nil { + return storage.ObjectInfo{}, fmt.Errorf("对象 ETag 缺失") + } + etag := strings.TrimSpace(strings.Trim(strings.TrimSpace(*result.ETag), "\"")) + if etag == "" { + return storage.ObjectInfo{}, fmt.Errorf("对象 ETag 无效") + } + if result.LastModified == nil || result.LastModified.IsZero() { + return storage.ObjectInfo{}, fmt.Errorf("对象最后修改时间缺失") + } + + return storage.ObjectInfo{ + Size: result.ContentLength, + StorageETag: etag, + LastModified: *result.LastModified, + }, nil +} + +func (c *Client) Delete(ctx context.Context, objectKey string) error { + if strings.TrimSpace(objectKey) == "" { + return fmt.Errorf("对象键不能为空") + } + + _, err := c.client.DeleteObject(ctx, &oss.DeleteObjectRequest{ + Bucket: oss.Ptr(c.bucket), + Key: oss.Ptr(objectKey), + }) + if err != nil { + return fmt.Errorf("删除对象: %w", err) + } + + return nil +} + +func (c *Client) Status(ctx context.Context) (storage.RuntimeStatus, error) { + _, err := c.client.ListObjectsV2(ctx, &oss.ListObjectsV2Request{Bucket: oss.Ptr(c.bucket), MaxKeys: 1}) + if err != nil { + return storage.RuntimeStatus{Provider: c.Provider(), Container: c.Container()}, fmt.Errorf("检查 OSS 存储状态: %w", err) + } + return storage.RuntimeStatus{Provider: c.Provider(), Container: c.Container(), Writable: true}, nil +} diff --git a/internal/storage/aliyun/upload.go b/internal/storage/aliyun/upload.go new file mode 100644 index 0000000..256aedf --- /dev/null +++ b/internal/storage/aliyun/upload.go @@ -0,0 +1,37 @@ +package aliyun + +import ( + "context" + "fmt" + "strings" + "time" + + "git.apinb.com/ops/files/internal/storage" + "github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss" +) + +func (c *Client) PrepareUpload(ctx context.Context, request storage.UploadRequest) (storage.UploadInstruction, error) { + if strings.TrimSpace(request.ObjectKey) == "" { + return storage.UploadInstruction{}, fmt.Errorf("对象键不能为空") + } + if strings.TrimSpace(request.ContentType) == "" { + return storage.UploadInstruction{}, fmt.Errorf("内容类型不能为空") + } + + result, err := c.client.Presign(ctx, &oss.PutObjectRequest{ + Bucket: oss.Ptr(c.bucket), + Key: oss.Ptr(request.ObjectKey), + ContentType: oss.Ptr(request.ContentType), + ForbidOverwrite: oss.Ptr("true"), + }, oss.PresignExpires(time.Duration(c.presignTTL)*time.Second)) + if err != nil { + return storage.UploadInstruction{}, fmt.Errorf("生成上传预签名: %w", err) + } + + return storage.UploadInstruction{ + Method: result.Method, + URL: result.URL, + Headers: result.SignedHeaders, + ExpiresAt: result.Expiration, + }, nil +} diff --git a/internal/storage/backend.go b/internal/storage/backend.go new file mode 100644 index 0000000..23a588e --- /dev/null +++ b/internal/storage/backend.go @@ -0,0 +1,77 @@ +package storage + +import ( + "context" + "errors" + "io" + "time" +) + +var ( + ErrInsufficientSpace = errors.New("文件存储空间不足") + ErrUploadLocked = errors.New("文件正在上传") + ErrObjectNotFound = errors.New("文件对象不存在") +) + +type UploadRequest struct { + FileID string + ObjectKey string + ContentType string + ExpectedSize int64 + ExpiresAt time.Time +} + +type UploadInstruction struct { + Method string `json:"method"` + URL string `json:"url"` + Headers map[string]string `json:"headers"` + ExpiresAt time.Time `json:"expires_at"` +} + +type ByteRange struct { + Start int64 + End int64 +} + +func (r ByteRange) Length() int64 { + return r.End - r.Start + 1 +} + +type ObjectInfo struct { + Size int64 + StorageETag string + LastModified time.Time +} + +type DiskStatus struct { + Path string `json:"path"` + TotalBytes uint64 `json:"total_bytes"` + UsedBytes uint64 `json:"used_bytes"` + FreeBytes uint64 `json:"free_bytes"` + UsedPercent float64 `json:"used_percent"` + MinimumFreeBytes uint64 `json:"minimum_free_bytes"` + Level string `json:"level"` +} + +type RuntimeStatus struct { + Provider string `json:"provider"` + Container string `json:"container"` + Writable bool `json:"writable"` + Disk *DiskStatus `json:"disk,omitempty"` +} + +type Backend interface { + Provider() string + Container() string + PrepareUpload(context.Context, UploadRequest) (UploadInstruction, error) + Open(context.Context, string, *ByteRange) (io.ReadCloser, error) + Stat(context.Context, string) (ObjectInfo, error) + Delete(context.Context, string) error + Status(context.Context) (RuntimeStatus, error) +} + +type UploadReceiver interface { + AcquireUpload(context.Context, string) (release func() error, err error) + ReceiveUpload(context.Context, UploadRequest, io.Reader) (ObjectInfo, error) + DiscardUpload(context.Context, string) error +} diff --git a/internal/storage/factory/factory.go b/internal/storage/factory/factory.go new file mode 100644 index 0000000..5acf5a4 --- /dev/null +++ b/internal/storage/factory/factory.go @@ -0,0 +1,21 @@ +package factory + +import ( + "fmt" + + "git.apinb.com/ops/files/internal/config" + "git.apinb.com/ops/files/internal/storage" + "git.apinb.com/ops/files/internal/storage/aliyun" + "git.apinb.com/ops/files/internal/storage/local" +) + +func New(storageConfig config.StorageConf) (storage.Backend, error) { + switch storageConfig.Provider { + case "local": + return local.New(storageConfig) + case "aliyun": + return aliyun.New(storageConfig.Aliyun, storageConfig.AccessTTLSeconds) + default: + return nil, fmt.Errorf("不支持的文件存储类型: %s", storageConfig.Provider) + } +} diff --git a/internal/storage/local/capacity.go b/internal/storage/local/capacity.go new file mode 100644 index 0000000..894c841 --- /dev/null +++ b/internal/storage/local/capacity.go @@ -0,0 +1,46 @@ +package local + +import ( + "fmt" + "math" + + "git.apinb.com/ops/files/internal/storage" + "github.com/shirou/gopsutil/v3/disk" +) + +func (c *Client) diskStatus() (*storage.DiskStatus, error) { + usage, err := disk.Usage(c.rootPath) + if err != nil { + return nil, fmt.Errorf("读取本地存储磁盘容量: %w", err) + } + level := "normal" + if usage.Free < c.minimumFreeBytes { + level = "critical" + } else if c.minimumFreeBytes <= math.MaxUint64/2 && usage.Free < c.minimumFreeBytes*2 { + level = "warning" + } + return &storage.DiskStatus{ + Path: c.rootPath, + TotalBytes: usage.Total, + UsedBytes: usage.Used, + FreeBytes: usage.Free, + UsedPercent: usage.UsedPercent, + MinimumFreeBytes: c.minimumFreeBytes, + Level: level, + }, nil +} + +func (c *Client) ensureCapacity(expectedSize int64) error { + if expectedSize < 0 { + return fmt.Errorf("文件大小不能为负数") + } + diskStatus, err := c.diskStatus() + if err != nil { + return err + } + requested := uint64(expectedSize) + if requested > math.MaxUint64-c.minimumFreeBytes || diskStatus.FreeBytes < requested+c.minimumFreeBytes { + return storage.ErrInsufficientSpace + } + return nil +} diff --git a/internal/storage/local/client.go b/internal/storage/local/client.go new file mode 100644 index 0000000..2c52570 --- /dev/null +++ b/internal/storage/local/client.go @@ -0,0 +1,60 @@ +package local + +import ( + "fmt" + "os" + "path/filepath" + "strings" + + "git.apinb.com/ops/files/internal/config" +) + +const containerName = "local" + +type Client struct { + rootPath string + objectsPath string + stagingPath string + locksPath string + externalBaseURL string + signingSecret string + accessTTLSeconds int64 + minimumFreeBytes uint64 +} + +func New(storageConfig config.StorageConf) (*Client, error) { + rootPath, err := filepath.Abs(storageConfig.Local.RootPath) + if err != nil { + return nil, fmt.Errorf("解析本地存储根目录: %w", err) + } + if err := os.MkdirAll(rootPath, 0o750); err != nil { + return nil, fmt.Errorf("创建本地存储根目录: %w", err) + } + rootPath, err = filepath.EvalSymlinks(rootPath) + if err != nil { + return nil, fmt.Errorf("解析本地存储真实路径: %w", err) + } + client := &Client{ + rootPath: filepath.Clean(rootPath), + objectsPath: filepath.Join(rootPath, "objects"), + stagingPath: filepath.Join(rootPath, ".staging"), + locksPath: filepath.Join(rootPath, ".locks"), + externalBaseURL: strings.TrimRight(storageConfig.ExternalBaseURL, "/"), + signingSecret: storageConfig.SigningSecret, + accessTTLSeconds: storageConfig.AccessTTLSeconds, + minimumFreeBytes: storageConfig.Local.MinimumFreeSpaceMB * 1024 * 1024, + } + for _, directory := range []string{client.objectsPath, client.stagingPath, client.locksPath} { + if err := client.ensureDirectory(directory); err != nil { + return nil, err + } + } + if _, err := client.Status(nil); err != nil { + return nil, err + } + return client, nil +} + +func (c *Client) Provider() string { return "local" } + +func (c *Client) Container() string { return containerName } diff --git a/internal/storage/local/object.go b/internal/storage/local/object.go new file mode 100644 index 0000000..01879c6 --- /dev/null +++ b/internal/storage/local/object.go @@ -0,0 +1,152 @@ +package local + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "io" + "os" + "path/filepath" + "strings" + + "git.apinb.com/ops/files/internal/storage" +) + +func (c *Client) Open(ctx context.Context, objectKey string, byteRange *storage.ByteRange) (io.ReadCloser, error) { + path, err := c.resolveObjectPath(objectKey) + if err != nil { + return nil, err + } + file, err := os.Open(path) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + return nil, storage.ErrObjectNotFound + } + return nil, fmt.Errorf("打开本地文件: %w", err) + } + if byteRange == nil { + return &contextReadCloser{ctx: ctx, reader: file, closer: file}, nil + } + info, err := file.Stat() + if err != nil { + _ = file.Close() + return nil, err + } + if byteRange.Start < 0 || byteRange.End < byteRange.Start || byteRange.End >= info.Size() { + _ = file.Close() + return nil, fmt.Errorf("文件读取范围无效") + } + if _, err := file.Seek(byteRange.Start, io.SeekStart); err != nil { + _ = file.Close() + return nil, err + } + return &contextReadCloser{ctx: ctx, reader: io.LimitReader(file, byteRange.Length()), closer: file}, nil +} + +func (c *Client) Stat(ctx context.Context, objectKey string) (storage.ObjectInfo, error) { + path, err := c.resolveObjectPath(objectKey) + if err != nil { + return storage.ObjectInfo{}, err + } + file, err := os.Open(path) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + return storage.ObjectInfo{}, storage.ErrObjectNotFound + } + return storage.ObjectInfo{}, fmt.Errorf("打开本地文件: %w", err) + } + defer file.Close() + info, err := file.Stat() + if err != nil { + return storage.ObjectInfo{}, err + } + hasher := sha256.New() + if _, err := io.Copy(hasher, &contextReader{ctx: ctx, reader: file}); err != nil { + return storage.ObjectInfo{}, fmt.Errorf("计算本地文件摘要: %w", err) + } + return storage.ObjectInfo{ + Size: info.Size(), + StorageETag: "sha256:" + hex.EncodeToString(hasher.Sum(nil)), + LastModified: info.ModTime().UTC(), + }, nil +} + +func (c *Client) Delete(_ context.Context, objectKey string) error { + path, err := c.resolveObjectPath(objectKey) + if err != nil { + return err + } + if err := os.Remove(path); err != nil && !os.IsNotExist(err) { + return fmt.Errorf("删除本地文件: %w", err) + } + c.removeEmptyParents(filepath.Dir(path)) + return nil +} + +func (c *Client) removeEmptyParents(directory string) { + boundary := filepath.Clean(c.objectsPath) + for current := filepath.Clean(directory); current != boundary; current = filepath.Dir(current) { + relative, err := filepath.Rel(boundary, current) + if err != nil || relative == ".." || strings.HasPrefix(relative, ".."+string(filepath.Separator)) { + return + } + if err := os.Remove(current); err != nil { + return + } + } +} + +func (c *Client) Status(_ context.Context) (storage.RuntimeStatus, error) { + diskStatus, err := c.diskStatus() + if err != nil { + return storage.RuntimeStatus{Provider: c.Provider(), Container: c.Container()}, err + } + probe, err := os.CreateTemp(c.stagingPath, ".write-probe-") + if err != nil { + return storage.RuntimeStatus{Provider: c.Provider(), Container: c.Container(), Disk: diskStatus}, fmt.Errorf("本地存储目录不可写: %w", err) + } + probePath := probe.Name() + removeProbe := func() { _ = os.Remove(probePath) } + defer removeProbe() + if err := probe.Chmod(0o640); err != nil { + _ = probe.Close() + return storage.RuntimeStatus{Provider: c.Provider(), Container: c.Container(), Disk: diskStatus}, err + } + if _, err := probe.Write([]byte("files-storage-probe")); err != nil { + _ = probe.Close() + return storage.RuntimeStatus{Provider: c.Provider(), Container: c.Container(), Disk: diskStatus}, err + } + if err := probe.Sync(); err != nil { + _ = probe.Close() + return storage.RuntimeStatus{Provider: c.Provider(), Container: c.Container(), Disk: diskStatus}, err + } + if err := probe.Close(); err != nil { + return storage.RuntimeStatus{Provider: c.Provider(), Container: c.Container(), Disk: diskStatus}, err + } + if err := os.Remove(probePath); err != nil { + return storage.RuntimeStatus{Provider: c.Provider(), Container: c.Container(), Disk: diskStatus}, err + } + return storage.RuntimeStatus{ + Provider: c.Provider(), + Container: c.Container(), + Writable: diskStatus.Level != "critical", + Disk: diskStatus, + }, nil +} + +type contextReadCloser struct { + ctx context.Context + reader io.Reader + closer io.Closer +} + +func (r *contextReadCloser) Read(buffer []byte) (int, error) { + if err := r.ctx.Err(); err != nil { + return 0, err + } + return r.reader.Read(buffer) +} + +func (r *contextReadCloser) Close() error { return r.closer.Close() } diff --git a/internal/storage/local/path.go b/internal/storage/local/path.go new file mode 100644 index 0000000..ff951a6 --- /dev/null +++ b/internal/storage/local/path.go @@ -0,0 +1,79 @@ +package local + +import ( + "fmt" + "os" + "path/filepath" + "strings" +) + +func (c *Client) resolveObjectPath(objectKey string) (string, error) { + if strings.TrimSpace(objectKey) == "" || filepath.IsAbs(objectKey) || strings.Contains(objectKey, "\\") { + return "", fmt.Errorf("对象键不是安全的相对路径") + } + for _, segment := range strings.Split(objectKey, "/") { + if segment == "" || segment == "." || segment == ".." { + return "", fmt.Errorf("对象键包含无效路径段") + } + } + target := filepath.Join(c.objectsPath, filepath.FromSlash(objectKey)) + relative, err := filepath.Rel(c.objectsPath, target) + if err != nil || relative == ".." || strings.HasPrefix(relative, ".."+string(filepath.Separator)) { + return "", fmt.Errorf("对象路径超出存储目录") + } + if err := c.ensureNoSymlink(filepath.Dir(target), c.objectsPath); err != nil { + return "", err + } + return target, nil +} + +func (c *Client) ensureDirectory(directory string) error { + if err := os.MkdirAll(directory, 0o750); err != nil { + return fmt.Errorf("创建本地存储目录 %s: %w", directory, err) + } + if err := c.ensureNoSymlink(directory, c.rootPath); err != nil { + return err + } + return os.Chmod(directory, 0o750) +} + +func (c *Client) ensureNoSymlink(target, boundary string) error { + target = filepath.Clean(target) + boundary = filepath.Clean(boundary) + relative, err := filepath.Rel(boundary, target) + if err != nil || relative == ".." || strings.HasPrefix(relative, ".."+string(filepath.Separator)) { + return fmt.Errorf("路径超出本地存储边界") + } + current := boundary + if relative == "." { + return rejectSymlink(current) + } + if err := rejectSymlink(current); err != nil { + return err + } + for _, segment := range strings.Split(relative, string(filepath.Separator)) { + current = filepath.Join(current, segment) + info, statErr := os.Lstat(current) + if os.IsNotExist(statErr) { + continue + } + if statErr != nil { + return fmt.Errorf("检查路径 %s: %w", current, statErr) + } + if info.Mode()&os.ModeSymlink != 0 { + return fmt.Errorf("本地存储路径不允许符号链接: %s", current) + } + } + return nil +} + +func rejectSymlink(path string) error { + info, err := os.Lstat(path) + if err != nil { + return err + } + if info.Mode()&os.ModeSymlink != 0 { + return fmt.Errorf("本地存储路径不允许符号链接: %s", path) + } + return nil +} diff --git a/internal/storage/local/upload.go b/internal/storage/local/upload.go new file mode 100644 index 0000000..7381b68 --- /dev/null +++ b/internal/storage/local/upload.go @@ -0,0 +1,190 @@ +package local + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "io" + "os" + "path/filepath" + "strings" + "time" + + "git.apinb.com/ops/files/internal/signing" + "git.apinb.com/ops/files/internal/storage" +) + +func (c *Client) PrepareUpload(_ context.Context, request storage.UploadRequest) (storage.UploadInstruction, error) { + if err := c.ensureCapacity(request.ExpectedSize); err != nil { + return storage.UploadInstruction{}, err + } + signature := signing.SignUpload(c.signingSecret, request) + return storage.UploadInstruction{ + Method: "PUT", + URL: fmt.Sprintf("%s/v1/local/uploads/%s?expires=%d&signature=%s", + c.externalBaseURL, request.FileID, request.ExpiresAt.Unix(), signature), + Headers: map[string]string{ + "Content-Type": request.ContentType, + "Content-Length": fmt.Sprintf("%d", request.ExpectedSize), + }, + ExpiresAt: request.ExpiresAt, + }, nil +} + +func (c *Client) AcquireUpload(ctx context.Context, fileID string) (func() error, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + if !validFileID(fileID) { + return nil, fmt.Errorf("文件标识无效") + } + lockPath := filepath.Join(c.locksPath, fileID+".lock") + lockFile, err := os.OpenFile(lockPath, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o640) + if err != nil { + if os.IsExist(err) { + return nil, storage.ErrUploadLocked + } + return nil, fmt.Errorf("获取上传锁: %w", err) + } + return func() error { + closeErr := lockFile.Close() + removeErr := os.Remove(lockPath) + if os.IsNotExist(removeErr) { + removeErr = nil + } + if closeErr != nil { + return closeErr + } + return removeErr + }, nil +} + +func (c *Client) ReceiveUpload(ctx context.Context, request storage.UploadRequest, source io.Reader) (storage.ObjectInfo, error) { + if err := c.ensureCapacity(request.ExpectedSize); err != nil { + return storage.ObjectInfo{}, err + } + targetPath, err := c.resolveObjectPath(request.ObjectKey) + if err != nil { + return storage.ObjectInfo{}, err + } + if err := c.ensureDirectory(filepath.Dir(targetPath)); err != nil { + return storage.ObjectInfo{}, err + } + if existing, statErr := c.Stat(ctx, request.ObjectKey); statErr == nil { + if existing.Size != request.ExpectedSize { + return storage.ObjectInfo{}, fmt.Errorf("已存在文件的大小与上传请求不一致") + } + return existing, nil + } else if !errors.Is(statErr, storage.ErrObjectNotFound) { + return storage.ObjectInfo{}, statErr + } + nonce, err := signing.UploadNonce(c.signingSecret, request) + if err != nil { + return storage.ObjectInfo{}, err + } + stagingPath := filepath.Join(c.stagingPath, request.FileID+"."+nonce+".part") + staging, err := os.OpenFile(stagingPath, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o640) + if err != nil { + return storage.ObjectInfo{}, fmt.Errorf("创建上传暂存文件: %w", err) + } + committed := false + defer func() { + _ = staging.Close() + if !committed { + _ = os.Remove(stagingPath) + } + }() + + hasher := sha256.New() + written, err := io.Copy(io.MultiWriter(staging, hasher), io.LimitReader(&contextReader{ctx: ctx, reader: source}, request.ExpectedSize+1)) + if err != nil { + return storage.ObjectInfo{}, fmt.Errorf("写入上传暂存文件: %w", err) + } + if written != request.ExpectedSize { + return storage.ObjectInfo{}, fmt.Errorf("上传文件大小不匹配: 期望 %d，实际 %d", request.ExpectedSize, written) + } + if err := staging.Sync(); err != nil { + return storage.ObjectInfo{}, fmt.Errorf("同步上传暂存文件: %w", err) + } + if err := staging.Close(); err != nil { + return storage.ObjectInfo{}, fmt.Errorf("关闭上传暂存文件: %w", err) + } + if err := os.Link(stagingPath, targetPath); err != nil { + return storage.ObjectInfo{}, fmt.Errorf("提交上传文件: %w", err) + } + if err := os.Chmod(targetPath, 0o640); err != nil { + _ = os.Remove(targetPath) + return storage.ObjectInfo{}, fmt.Errorf("设置上传文件权限: %w", err) + } + if err := os.Remove(stagingPath); err != nil { + _ = os.Remove(targetPath) + return storage.ObjectInfo{}, fmt.Errorf("清理上传暂存文件: %w", err) + } + if err := syncDirectory(filepath.Dir(targetPath)); err != nil { + _ = os.Remove(targetPath) + return storage.ObjectInfo{}, err + } + committed = true + return storage.ObjectInfo{ + Size: written, + StorageETag: "sha256:" + hex.EncodeToString(hasher.Sum(nil)), + LastModified: time.Now().UTC(), + }, nil +} + +func (c *Client) DiscardUpload(_ context.Context, fileID string) error { + if !validFileID(fileID) { + return fmt.Errorf("文件标识无效") + } + entries, err := os.ReadDir(c.stagingPath) + if err != nil { + return err + } + prefix := fileID + "." + for _, entry := range entries { + if !entry.IsDir() && strings.HasPrefix(entry.Name(), prefix) && strings.HasSuffix(entry.Name(), ".part") { + if err := os.Remove(filepath.Join(c.stagingPath, entry.Name())); err != nil && !os.IsNotExist(err) { + return err + } + } + } + return nil +} + +type contextReader struct { + ctx context.Context + reader io.Reader +} + +func (r *contextReader) Read(buffer []byte) (int, error) { + if err := r.ctx.Err(); err != nil { + return 0, err + } + return r.reader.Read(buffer) +} + +func validFileID(fileID string) bool { + if fileID == "" { + return false + } + for _, character := range fileID { + if (character < '0' || character > '9') && (character < 'A' || character > 'Z') && (character < 'a' || character > 'z') && character != '-' { + return false + } + } + return true +} + +func syncDirectory(path string) error { + directory, err := os.Open(path) + if err != nil { + return fmt.Errorf("打开目录进行同步: %w", err) + } + defer directory.Close() + if err := directory.Sync(); err != nil { + return fmt.Errorf("同步文件目录: %w", err) + } + return nil +} diff --git a/internal/storage/object.go b/internal/storage/object.go deleted file mode 100644 index 11f5a0d..0000000 --- a/internal/storage/object.go +++ /dev/null @@ -1,76 +0,0 @@ -package storage - -import ( - "context" - "fmt" - "strings" - "time" - - "github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss" -) - -type ObjectInfo struct { - Size int64 - ETag string - LastModified time.Time -} - -func (c *Client) Stat(ctx context.Context, objectKey string) (ObjectInfo, error) { - if strings.TrimSpace(objectKey) == "" { - return ObjectInfo{}, fmt.Errorf("对象键不能为空") - } - - result, err := c.client.GetObjectMeta(ctx, &oss.GetObjectMetaRequest{ - Bucket: oss.Ptr(c.bucket), - Key: oss.Ptr(objectKey), - }) - if err != nil { - return ObjectInfo{}, fmt.Errorf("查询对象元数据: %w", err) - } - if result == nil { - return ObjectInfo{}, fmt.Errorf("对象元数据为空") - } - if result.ContentLength < 0 { - return ObjectInfo{}, fmt.Errorf("对象大小无效: %d", result.ContentLength) - } - if result.ETag == nil { - return ObjectInfo{}, fmt.Errorf("对象 ETag 缺失") - } - etag := strings.TrimSpace(strings.Trim(strings.TrimSpace(*result.ETag), "\"")) - if etag == "" { - return ObjectInfo{}, fmt.Errorf("对象 ETag 无效") - } - if result.LastModified == nil || result.LastModified.IsZero() { - return ObjectInfo{}, fmt.Errorf("对象最后修改时间缺失") - } - - return ObjectInfo{ - Size: result.ContentLength, - ETag: etag, - LastModified: *result.LastModified, - }, nil -} - -func (c *Client) Delete(ctx context.Context, objectKey string) error { - if strings.TrimSpace(objectKey) == "" { - return fmt.Errorf("对象键不能为空") - } - - _, err := c.client.DeleteObject(ctx, &oss.DeleteObjectRequest{ - Bucket: oss.Ptr(c.bucket), - Key: oss.Ptr(objectKey), - }) - if err != nil { - return fmt.Errorf("删除对象: %w", err) - } - - return nil -} - -func (c *Client) PublicURL(objectKey string) string { - return strings.TrimRight(c.publicBaseURL, "/") + "/" + strings.TrimLeft(objectKey, "/") -} - -func (c *Client) Bucket() string { - return c.bucket -} diff --git a/internal/storage/presign.go b/internal/storage/presign.go deleted file mode 100644 index ce47921..0000000 --- a/internal/storage/presign.go +++ /dev/null @@ -1,43 +0,0 @@ -package storage - -import ( - "context" - "fmt" - "strings" - "time" - - "github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss" -) - -type UploadInstruction struct { - Method string `json:"method"` - URL string `json:"url"` - Headers map[string]string `json:"headers"` - ExpiresAt time.Time `json:"expires_at"` -} - -func (c *Client) PresignPut(ctx context.Context, objectKey, contentType string) (UploadInstruction, error) { - if strings.TrimSpace(objectKey) == "" { - return UploadInstruction{}, fmt.Errorf("对象键不能为空") - } - if strings.TrimSpace(contentType) == "" { - return UploadInstruction{}, fmt.Errorf("内容类型不能为空") - } - - result, err := c.client.Presign(ctx, &oss.PutObjectRequest{ - Bucket: oss.Ptr(c.bucket), - Key: oss.Ptr(objectKey), - ContentType: oss.Ptr(contentType), - ForbidOverwrite: oss.Ptr("true"), - }, oss.PresignExpires(c.presignTTL)) - if err != nil { - return UploadInstruction{}, fmt.Errorf("生成上传预签名: %w", err) - } - - return UploadInstruction{ - Method: result.Method, - URL: result.URL, - Headers: result.SignedHeaders, - ExpiresAt: result.Expiration, - }, nil -}