数据源和 oss

This commit is contained in:
2026-08-13 23:17:55 +08:00
parent f1a0b17bb0
commit c4bf7ee03b
40 changed files with 1975 additions and 922 deletions
+50 -92
View File
@@ -2,126 +2,84 @@ package data
import (
"context"
"errors"
"time"
"github.com/go-kratos/aip-go/ents"
"gorm.io/gorm"
"kra/internal/biz"
"kra/internal/data/ent"
"kra/internal/data/ent/admin"
)
func convertAdmin(po *ent.Admin) *biz.Admin {
return &biz.Admin{
ID: po.ID,
Name: po.Name,
Email: po.Email,
Avatar: po.Avatar,
Access: po.Access,
Password: po.Password,
CreateTime: po.CreateTime,
UpdateTime: po.UpdateTime,
type adminPO struct {
ID int64 `gorm:"primaryKey;autoIncrement"`
Name string `gorm:"uniqueIndex"`
Email string `gorm:"uniqueIndex"`
Password string
Access string
Avatar string
CreateTime time.Time
UpdateTime time.Time
}
func (adminPO) TableName() string { return "admins" }
func convertAdmin(po *adminPO) *biz.Admin {
return &biz.Admin{ID: po.ID, Name: po.Name, Email: po.Email, Password: po.Password, Access: po.Access, Avatar: po.Avatar, CreateTime: po.CreateTime, UpdateTime: po.UpdateTime}
}
type adminRepo struct{ data *Data }
func NewAdminRepo(data *Data) biz.AdminRepo { return &adminRepo{data: data} }
func (r *adminRepo) one(ctx context.Context, query *gorm.DB) (*biz.Admin, error) {
var po adminPO
if err := query.WithContext(ctx).First(&po).Error; err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
return nil, biz.ErrAdminNotFound
}
return nil, err
}
return convertAdmin(&po), nil
}
type adminRepo struct {
data *Data
}
// NewAdminRepo creates a new AdminRepo instance.
func NewAdminRepo(data *Data) biz.AdminRepo {
return &adminRepo{
data: data,
}
}
func (r *adminRepo) FindByID(ctx context.Context, id int64) (*biz.Admin, error) {
po, err := r.data.db.Admin.Get(ctx, id)
if err != nil {
if ent.IsNotFound(err) {
return nil, biz.ErrAdminNotFound
}
return nil, err
}
return convertAdmin(po), nil
return r.one(ctx, r.data.gormDB.Where("id = ?", id))
}
func (r *adminRepo) FindByName(ctx context.Context, name string) (*biz.Admin, error) {
po, err := r.data.db.Admin.Query().Where(admin.NameEQ(name)).Only(ctx)
if err != nil {
if ent.IsNotFound(err) {
return nil, biz.ErrAdminNotFound
}
return nil, err
}
return convertAdmin(po), nil
return r.one(ctx, r.data.gormDB.Where("name = ?", name))
}
func (r *adminRepo) FindByEmail(ctx context.Context, email string) (*biz.Admin, error) {
po, err := r.data.db.Admin.Query().Where(admin.EmailEQ(email)).Only(ctx)
if err != nil {
if ent.IsNotFound(err) {
return nil, biz.ErrAdminNotFound
}
return nil, err
}
return convertAdmin(po), nil
return r.one(ctx, r.data.gormDB.Where("email = ?", email))
}
func (r *adminRepo) ListAdmins(ctx context.Context, opts ...biz.ListOption) ([]*biz.Admin, error) {
o := biz.ListOptions{Limit: 20}
for _, opt := range opts {
opt(&o)
}
pos, err := r.data.db.Admin.Query().
Where(ents.ApplyFilter(o.Filter)).
Order(ents.ApplyOrderBy(o.OrderBy)).
Offset(o.Offset).
Limit(o.Limit).
All(ctx)
if err != nil {
var pos []adminPO
if err := r.data.gormDB.WithContext(ctx).Offset(o.Offset).Limit(o.Limit).Order("id asc").Find(&pos).Error; err != nil {
return nil, err
}
var admins []*biz.Admin
for _, po := range pos {
admins = append(admins, convertAdmin(po))
out := make([]*biz.Admin, 0, len(pos))
for i := range pos {
out = append(out, convertAdmin(&pos[i]))
}
return admins, nil
return out, nil
}
func (r *adminRepo) CreateAdmin(ctx context.Context, admin *biz.Admin) (*biz.Admin, error) {
po, err := r.data.db.Admin.Create().
SetName(admin.Name).
SetEmail(admin.Email).
SetAvatar(admin.Avatar).
SetAccess(admin.Access).
SetPassword(admin.Password).
SetCreateTime(time.Now()).
SetUpdateTime(time.Now()).
Save(ctx)
if err != nil {
func (r *adminRepo) CreateAdmin(ctx context.Context, a *biz.Admin) (*biz.Admin, error) {
now := time.Now()
po := adminPO{Name: a.Name, Email: a.Email, Password: a.Password, Access: a.Access, Avatar: a.Avatar, CreateTime: now, UpdateTime: now}
if err := r.data.gormDB.WithContext(ctx).Create(&po).Error; err != nil {
return nil, err
}
return convertAdmin(po), nil
return convertAdmin(&po), nil
}
func (r *adminRepo) UpdateAdmin(ctx context.Context, admin *biz.Admin) (*biz.Admin, error) {
update := r.data.db.Admin.UpdateOneID(admin.ID).
SetName(admin.Name).
SetEmail(admin.Email).
SetAccess(admin.Access).
SetAvatar(admin.Avatar).
SetUpdateTime(time.Now())
// Only update the password if it's not empty
if admin.Password != "" {
update.SetPassword(admin.Password)
func (r *adminRepo) UpdateAdmin(ctx context.Context, a *biz.Admin) (*biz.Admin, error) {
values := map[string]any{"name": a.Name, "email": a.Email, "access": a.Access, "avatar": a.Avatar, "update_time": time.Now()}
if a.Password != "" {
values["password"] = a.Password
}
po, err := update.Save(ctx)
if err != nil {
if err := r.data.gormDB.WithContext(ctx).Model(&adminPO{}).Where("id = ?", a.ID).Updates(values).Error; err != nil {
return nil, err
}
return convertAdmin(po), nil
return r.FindByID(ctx, a.ID)
}
func (r *adminRepo) DeleteAdmin(ctx context.Context, id int64) error {
return r.data.db.Admin.DeleteOneID(id).Exec(ctx)
return r.data.gormDB.WithContext(ctx).Delete(&adminPO{}, id).Error
}
+103
View File
@@ -0,0 +1,103 @@
package data
import (
"encoding/json"
"fmt"
"os"
"path/filepath"
"google.golang.org/protobuf/encoding/protojson"
"google.golang.org/protobuf/proto"
"gopkg.in/yaml.v3"
)
func protoMap(message proto.Message) (map[string]any, error) {
raw, err := protojson.MarshalOptions{UseProtoNames: true}.Marshal(message)
if err != nil {
return nil, err
}
var value map[string]any
if err = json.Unmarshal(raw, &value); err != nil {
return nil, err
}
delete(value, "config_path")
return value, nil
}
func setYAMLMapping(node *yaml.Node, key string, value any) error {
if node.Kind == yaml.DocumentNode {
node = node.Content[0]
}
if node.Kind != yaml.MappingNode {
return fmt.Errorf("configuration root is not a mapping")
}
raw, err := yaml.Marshal(value)
if err != nil {
return err
}
var replacement yaml.Node
if err = yaml.Unmarshal(raw, &replacement); err != nil {
return err
}
for i := 0; i < len(node.Content); i += 2 {
if node.Content[i].Value == key {
node.Content[i+1] = replacement.Content[0]
return nil
}
}
node.Content = append(node.Content, &yaml.Node{Kind: yaml.ScalarNode, Value: key}, replacement.Content[0])
return nil
}
func (d *Data) persistConfig() error {
if d.admin == nil || d.admin.ConfigPath == "" {
return nil
}
raw, err := os.ReadFile(d.admin.ConfigPath)
if err != nil {
return err
}
var document yaml.Node
if err = yaml.Unmarshal(raw, &document); err != nil {
return err
}
dataValue, err := protoMap(d.config)
if err != nil {
return err
}
adminValue, err := protoMap(d.admin)
if err != nil {
return err
}
if err = setYAMLMapping(&document, "data", dataValue); err != nil {
return err
}
if err = setYAMLMapping(&document, "admin", adminValue); err != nil {
return err
}
output, err := yaml.Marshal(&document)
if err != nil {
return err
}
if err = os.MkdirAll(filepath.Dir(d.admin.ConfigPath), 0o755); err != nil {
return err
}
temporary, err := os.CreateTemp(filepath.Dir(d.admin.ConfigPath), ".kra-config-*.yaml")
if err != nil {
return err
}
tempName := temporary.Name()
defer os.Remove(tempName)
if _, err = temporary.Write(output); err != nil {
_ = temporary.Close()
return err
}
if err = temporary.Chmod(0o600); err != nil {
_ = temporary.Close()
return err
}
if err = temporary.Close(); err != nil {
return err
}
return os.Rename(tempName, d.admin.ConfigPath)
}
+55 -48
View File
@@ -4,55 +4,47 @@ import (
"context"
"fmt"
"log"
"os"
"sync"
"time"
_ "github.com/go-sql-driver/mysql"
"github.com/google/wire"
"github.com/redis/go-redis/v9"
"gorm.io/driver/mysql"
"gorm.io/gorm"
"kra/internal/conf"
"kra/internal/data/ent"
"kra/internal/data/ent/migrate"
)
// ProviderSet is data providers.
var ProviderSet = wire.NewSet(NewData, NewAdminRepo, NewSystemRepo, NewAccessRepo, NewSettingsRepo, NewVersionRepo, NewExportRepo, NewAuditRepo, NewTaskRepo, NewMediaRepo, NewAnnouncementRepo, NewEmailRepo, NewCache, NewFileStorage)
// Data is a struct that contains the database client.
type Data struct {
db *ent.Client
gormDB *gorm.DB
redis *redis.Client
mu sync.RWMutex
gormDB *gorm.DB
redis *redis.Client
database *conf.Data_Database
config *conf.Data
admin *conf.AdminBackend
}
// NewData creates a new Data instance.
func NewData(c *conf.Data) (*Data, func(), error) {
func NewData(c *conf.Data, admin *conf.AdminBackend) (*Data, func(), error) {
if c == nil || c.Database == nil {
return nil, nil, fmt.Errorf("database configuration is required")
}
db, err := ent.Open(c.Database.Driver, c.Database.Source)
d := &Data{database: c.Database, config: c, admin: admin}
db, err := openDatabase(c.Database, false)
if err != nil {
return nil, nil, fmt.Errorf("open ent database: %w", err)
}
gormDB, err := gorm.Open(mysql.Open(c.Database.Source), &gorm.Config{})
if err != nil {
_ = db.Close()
return nil, nil, err
}
// The announcement module owns its table lifecycle independently of the
// first-run database initializer.
if err = gormDB.AutoMigrate(&announcementPO{}); err != nil {
_ = db.Close()
return nil, nil, fmt.Errorf("migrate announcement table: %w", err)
}
var redisClient *redis.Client
if c.Redis != nil && c.Redis.Addr != "" {
options := &redis.Options{Addr: c.Redis.Addr}
if c.Redis.Network != "" {
options.Network = c.Redis.Network
// The initialization endpoint must remain available when the configured
// target database has not been created yet.
log.Printf("configured database unavailable before initialization: %v", err)
db, err = openFallbackDatabase()
if err != nil {
return nil, nil, fmt.Errorf("open bootstrap database: %w", err)
}
}
d.gormDB = db
if err = db.AutoMigrate(&announcementPO{}, &adminPO{}); err != nil {
return nil, nil, fmt.Errorf("migrate bootstrap tables: %w", err)
}
if c.Redis != nil && c.Redis.Addr != "" {
options := &redis.Options{Addr: c.Redis.Addr, Network: c.Redis.Network}
if c.Redis.ReadTimeout != nil {
options.ReadTimeout = c.Redis.ReadTimeout.AsDuration()
}
@@ -65,27 +57,42 @@ func NewData(c *conf.Data) (*Data, func(), error) {
log.Printf("redis unavailable, using in-memory cache: %v", pingErr)
_ = candidate.Close()
} else {
redisClient = candidate
d.redis = candidate
}
cancel()
}
if os.Getenv("DEPLOY_ENV") == "dev" {
// Enable debug mode for detailed logging.
db = db.Debug()
// Run the auto migration tool.
if err = db.Schema.Create(context.Background(), migrate.WithDropIndex(true)); err != nil {
return nil, nil, err
}
}
cleanup := func() {
_ = db.Close()
if redisClient != nil {
_ = redisClient.Close()
d.mu.RLock()
db := d.gormDB
d.mu.RUnlock()
if sqlDB, closeErr := db.DB(); closeErr == nil {
_ = sqlDB.Close()
}
if d.redis != nil {
_ = d.redis.Close()
}
}
return &Data{
db: db,
gormDB: gormDB,
redis: redisClient,
}, cleanup, nil
return d, cleanup, nil
}
func (d *Data) switchDatabase(config *conf.Data_Database) error {
db, err := openDatabase(config, true)
if err != nil {
return err
}
d.mu.Lock()
old := d.gormDB
d.gormDB = db
d.database = config
d.config.Database = config
d.mu.Unlock()
if err := d.persistConfig(); err != nil {
return fmt.Errorf("persist database configuration: %w", err)
}
if old != nil {
if sqlDB, e := old.DB(); e == nil {
_ = sqlDB.Close()
}
}
return nil
}
+179
View File
@@ -0,0 +1,179 @@
package data
import (
"fmt"
"net"
"net/url"
"os"
"path/filepath"
"regexp"
"strings"
oracle "github.com/dzwvip/gorm-oracle"
"github.com/glebarez/sqlite"
"gorm.io/driver/mysql"
"gorm.io/driver/postgres"
"gorm.io/driver/sqlserver"
"gorm.io/gorm"
"kra/internal/conf"
)
var databaseNamePattern = regexp.MustCompile(`^[A-Za-z][A-Za-z0-9_$-]*$`)
func normalizedDriver(driver string) string {
switch strings.ToLower(strings.TrimSpace(driver)) {
case "postgres", "postgresql", "pgsql":
return "pgsql"
case "sqlserver", "mssql":
return "mssql"
case "sqlite", "sqlite3":
return "sqlite"
case "oracle":
return "oracle"
default:
return "mysql"
}
}
func databaseDSN(c *conf.Data_Database, name string) (string, error) {
if c.Source != "" && c.Host == "" && c.Path == "" {
return c.Source, nil
}
driver := normalizedDriver(c.Driver)
if name == "" {
name = c.Name
}
host := c.Host
if host == "" {
host = "127.0.0.1"
}
switch driver {
case "mysql":
port := c.Port
if port == "" {
port = "3306"
}
query := c.Config
if query == "" {
query = "timeout=5s&parseTime=True&loc=Local&charset=utf8mb4"
}
return fmt.Sprintf("%s:%s@tcp(%s)/%s?%s", c.User, c.Password, net.JoinHostPort(host, port), name, query), nil
case "pgsql":
port := c.Port
if port == "" {
port = "5432"
}
extra := c.Config
if extra == "" {
extra = "sslmode=disable TimeZone=Local"
}
return fmt.Sprintf("host=%s port=%s user=%s password=%s dbname=%s %s", host, port, c.User, c.Password, name, extra), nil
case "mssql":
port := c.Port
if port == "" {
port = "1433"
}
q := url.Values{"database": {name}, "encrypt": {"disable"}}
return (&url.URL{Scheme: "sqlserver", User: url.UserPassword(c.User, c.Password), Host: net.JoinHostPort(host, port), RawQuery: q.Encode()}).String(), nil
case "oracle":
port := c.Port
if port == "" {
port = "1521"
}
return fmt.Sprintf("oracle://%s:%s@%s/%s?%s", url.PathEscape(c.User), url.PathEscape(c.Password), net.JoinHostPort(host, port), url.PathEscape(name), c.Config), nil
case "sqlite":
path := c.Path
if path == "" {
path = "."
}
if name == "" {
name = "kra"
}
if filepath.Ext(name) == "" {
name += ".db"
}
return filepath.Join(path, name), nil
}
return "", fmt.Errorf("unsupported database driver %q", c.Driver)
}
func openWithDriver(driver, dsn string) (*gorm.DB, error) {
switch normalizedDriver(driver) {
case "mysql":
return gorm.Open(mysql.Open(dsn), &gorm.Config{})
case "pgsql":
return gorm.Open(postgres.Open(dsn), &gorm.Config{})
case "mssql":
return gorm.Open(sqlserver.Open(dsn), &gorm.Config{})
case "oracle":
return gorm.Open(oracle.Open(dsn), &gorm.Config{})
case "sqlite":
return gorm.Open(sqlite.Open(dsn), &gorm.Config{})
default:
return nil, fmt.Errorf("unsupported database driver %q", driver)
}
}
func openDatabase(c *conf.Data_Database, create bool) (*gorm.DB, error) {
driver := normalizedDriver(c.Driver)
if driver == "sqlite" {
dsn, err := databaseDSN(c, "")
if err != nil {
return nil, err
}
if err = os.MkdirAll(filepath.Dir(dsn), 0o755); err != nil {
return nil, err
}
return openWithDriver(driver, dsn)
}
if create && driver != "oracle" {
if !databaseNamePattern.MatchString(c.Name) {
return nil, fmt.Errorf("invalid database name %q", c.Name)
}
bootstrap := ""
switch driver {
case "pgsql":
bootstrap = "postgres"
case "mssql":
bootstrap = "master"
}
dsn, err := databaseDSN(c, bootstrap)
if err != nil {
return nil, err
}
adminDB, err := openWithDriver(driver, dsn)
if err != nil {
return nil, fmt.Errorf("connect database server: %w", err)
}
var statement string
switch driver {
case "mysql":
statement = "CREATE DATABASE IF NOT EXISTS `" + c.Name + "` CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci"
case "pgsql":
var count int64
if err = adminDB.Raw("SELECT count(*) FROM pg_database WHERE datname = ?", c.Name).Scan(&count).Error; err == nil && count == 0 {
statement = `CREATE DATABASE "` + c.Name + `"`
}
case "mssql":
statement = "IF DB_ID(N'" + c.Name + "') IS NULL CREATE DATABASE [" + c.Name + "]"
}
if statement != "" {
err = adminDB.Exec(statement).Error
}
if sqlDB, e := adminDB.DB(); e == nil {
_ = sqlDB.Close()
}
if err != nil {
return nil, fmt.Errorf("create database: %w", err)
}
}
dsn, err := databaseDSN(c, "")
if err != nil {
return nil, err
}
return openWithDriver(driver, dsn)
}
func openFallbackDatabase() (*gorm.DB, error) {
return openWithDriver("sqlite", "file:kra-bootstrap?mode=memory&cache=shared")
}
+65 -34
View File
@@ -21,6 +21,33 @@ type fileStorage struct {
}
func NewFileStorage(config *conf.AdminBackend) (biz.FileStorage, error) {
storageType := "local"
if config != nil && config.Storage != nil && config.Storage.Type != "" {
storageType = strings.ToLower(config.Storage.Type)
}
if storageType == "qiniu" {
return newQiniuStorage(config.Storage.Qiniu)
}
if storageType != "local" {
var object *conf.AdminBackend_ObjectStore
switch storageType {
case "aliyun-oss":
object = config.Storage.AliyunOss
case "huawei-obs":
object = config.Storage.HuaweiObs
case "tencent-cos":
object = config.Storage.TencentCos
case "aws-s3":
object = config.Storage.AwsS3
case "cloudflare-r2":
object = config.Storage.CloudflareR2
case "minio":
object = config.Storage.Minio
default:
return nil, fmt.Errorf("unsupported storage type %q", storageType)
}
return newS3Storage(storageType, object)
}
root, prefix := "uploads/file", "uploads/file"
if config != nil && config.Local != nil {
if config.Local.StorePath != "" {
@@ -37,6 +64,43 @@ func NewFileStorage(config *conf.AdminBackend) (biz.FileStorage, error) {
return &fileStorage{root: root, urlPrefix: "/" + strings.Trim(prefix, "/")}, nil
}
func composeFiles(ctx context.Context, storage biz.FileStorage, names []string, destination string) (*biz.StoredFile, string, error) {
reader, writer := io.Pipe()
hash := md5.New()
errCh := make(chan error, 1)
go func() {
defer writer.Close()
for _, name := range names {
if ctx.Err() != nil {
errCh <- ctx.Err()
return
}
file, err := storage.Open(ctx, name)
if err != nil {
errCh <- err
return
}
_, err = io.Copy(io.MultiWriter(writer, hash), file)
_ = file.Close()
if err != nil {
errCh <- err
return
}
}
errCh <- nil
}()
stored, err := storage.Put(ctx, destination, reader)
composeErr := <-errCh
if err != nil {
return nil, "", err
}
if composeErr != nil {
_ = storage.Delete(ctx, destination)
return nil, "", composeErr
}
return stored, hex.EncodeToString(hash.Sum(nil)), nil
}
func (s *fileStorage) resolve(name string) (string, error) {
clean := filepath.Clean(strings.TrimPrefix(name, "/"))
if clean == "." || filepath.IsAbs(clean) || clean == ".." || strings.HasPrefix(clean, ".."+string(filepath.Separator)) {
@@ -88,40 +152,7 @@ func (s *fileStorage) Delete(ctx context.Context, name string) error {
}
func (s *fileStorage) Compose(ctx context.Context, names []string, destination string) (*biz.StoredFile, string, error) {
reader, writer := io.Pipe()
hash := md5.New()
errCh := make(chan error, 1)
go func() {
defer writer.Close()
for _, name := range names {
if ctx.Err() != nil {
errCh <- ctx.Err()
return
}
file, err := s.Open(ctx, name)
if err != nil {
errCh <- err
return
}
_, err = io.Copy(io.MultiWriter(writer, hash), file)
_ = file.Close()
if err != nil {
errCh <- err
return
}
}
errCh <- nil
}()
stored, err := s.Put(ctx, destination, reader)
composeErr := <-errCh
if err != nil {
return nil, "", err
}
if composeErr != nil {
_ = s.Delete(ctx, destination)
return nil, "", composeErr
}
return stored, hex.EncodeToString(hash.Sum(nil)), nil
return composeFiles(ctx, s, names, destination)
}
func (s *fileStorage) DeletePrefix(ctx context.Context, prefix string) error {
path, err := s.resolve(strings.TrimSuffix(prefix, "/") + "/placeholder")
+1 -1
View File
@@ -143,7 +143,7 @@ func (r *mediaRepo) DeleteMedia(ctx context.Context, id uint) error {
}
func (r *mediaRepo) MediaKeyReferences(ctx context.Context, key string) (int64, error) {
var count int64
err := r.data.gormDB.WithContext(ctx).Model(&mediaPO{}).Where("`key` = ?", key).Count(&count).Error
err := r.data.gormDB.WithContext(ctx).Model(&mediaPO{}).Where(map[string]any{"key": key}).Count(&count).Error
return count, err
}
func (r *mediaRepo) CreateMediaBatch(ctx context.Context, items []*biz.MediaFile) error {
+126
View File
@@ -0,0 +1,126 @@
package data
import (
"context"
"fmt"
"io"
"net/http"
"os"
"path"
"strings"
"time"
"github.com/qiniu/go-sdk/v7/auth/qbox"
qstorage "github.com/qiniu/go-sdk/v7/storage"
"kra/internal/biz"
"kra/internal/conf"
)
type qiniuStorage struct {
config *conf.AdminBackend_Qiniu
upload *qstorage.FormUploader
manager *qstorage.BucketManager
mac *qbox.Mac
token string
}
func newQiniuStorage(config *conf.AdminBackend_Qiniu) (biz.FileStorage, error) {
if config == nil || config.Bucket == "" || config.AccessKey == "" || config.SecretKey == "" {
return nil, fmt.Errorf("qiniu storage configuration is incomplete")
}
cfg := qstorage.Config{UseHTTPS: config.UseHttps, UseCdnDomains: config.UseCdnDomains}
switch config.Zone {
case "ZoneHuadong":
cfg.Zone = &qstorage.ZoneHuadong
case "ZoneHuabei":
cfg.Zone = &qstorage.ZoneHuabei
case "ZoneHuanan":
cfg.Zone = &qstorage.ZoneHuanan
case "ZoneBeimei":
cfg.Zone = &qstorage.ZoneBeimei
case "ZoneXinjiapo":
cfg.Zone = &qstorage.ZoneXinjiapo
}
mac := qbox.NewMac(config.AccessKey, config.SecretKey)
policy := qstorage.PutPolicy{Scope: config.Bucket}
token := policy.UploadToken(mac)
return &qiniuStorage{config: config, upload: qstorage.NewFormUploader(&cfg), manager: qstorage.NewBucketManager(mac, &cfg), mac: mac, token: token}, nil
}
func (s *qiniuStorage) file(key string, size int64) *biz.StoredFile {
return &biz.StoredFile{Name: path.Base(key), Path: key, URL: strings.TrimSuffix(s.config.BaseUrl, "/") + "/" + key, Size: size}
}
func (s *qiniuStorage) Put(ctx context.Context, name string, reader io.Reader) (*biz.StoredFile, error) {
temporary, err := os.CreateTemp("", "kra-qiniu-upload-*")
if err != nil {
return nil, err
}
temporaryName := temporary.Name()
defer os.Remove(temporaryName)
size, err := io.Copy(temporary, reader)
if err != nil {
_ = temporary.Close()
return nil, err
}
if _, err = temporary.Seek(0, io.SeekStart); err != nil {
_ = temporary.Close()
return nil, err
}
defer temporary.Close()
ret := qstorage.PutRet{}
if err = s.upload.Put(ctx, &ret, s.token, strings.TrimPrefix(name, "/"), temporary, size, &qstorage.PutExtra{}); err != nil {
return nil, err
}
return s.file(ret.Key, size), nil
}
func (s *qiniuStorage) Open(ctx context.Context, name string) (io.ReadCloser, error) {
downloadURL := qstorage.MakePrivateURL(s.mac, strings.TrimSuffix(s.config.BaseUrl, "/"), strings.TrimPrefix(name, "/"), time.Now().Add(time.Hour).Unix())
req, err := http.NewRequestWithContext(ctx, http.MethodGet, downloadURL, nil)
if err != nil {
return nil, err
}
resp, err := http.DefaultClient.Do(req)
if err != nil {
return nil, err
}
if resp.StatusCode >= 300 {
_ = resp.Body.Close()
return nil, fmt.Errorf("qiniu download failed: %s", resp.Status)
}
return resp.Body, nil
}
func (s *qiniuStorage) Delete(ctx context.Context, name string) error {
return s.manager.Delete(s.config.Bucket, strings.TrimPrefix(name, "/"))
}
func (s *qiniuStorage) Compose(ctx context.Context, names []string, destination string) (*biz.StoredFile, string, error) {
return composeFiles(ctx, s, names, destination)
}
func (s *qiniuStorage) DeletePrefix(ctx context.Context, prefix string) error {
for {
items, _, more, err := s.List(ctx, prefix, "", 1000)
if err != nil {
return err
}
for _, item := range items {
if err = s.Delete(ctx, item.Path); err != nil {
return err
}
}
if !more {
return nil
}
}
}
func (s *qiniuStorage) List(ctx context.Context, prefix, cursor string, limit int) ([]*biz.StoredFile, string, bool, error) {
if limit < 1 || limit > 1000 {
limit = 100
}
entries, _, marker, more, err := s.manager.ListFiles(s.config.Bucket, prefix, "", cursor, limit)
if err != nil {
return nil, "", false, err
}
out := make([]*biz.StoredFile, 0, len(entries))
for _, entry := range entries {
out = append(out, s.file(entry.Key, entry.Fsize))
}
return out, marker, more, nil
}
+138
View File
@@ -0,0 +1,138 @@
package data
import (
"context"
"fmt"
"io"
"net/url"
"path"
"strings"
"github.com/minio/minio-go/v7"
"github.com/minio/minio-go/v7/pkg/credentials"
"kra/internal/biz"
"kra/internal/conf"
)
type s3Storage struct {
client *minio.Client
bucket, baseURL, prefix string
}
func newS3Storage(provider string, config *conf.AdminBackend_ObjectStore) (biz.FileStorage, error) {
if config == nil || config.Bucket == "" || config.AccessKey == "" || config.SecretKey == "" {
return nil, fmt.Errorf("%s storage configuration is incomplete", provider)
}
endpoint := config.Endpoint
if endpoint == "" {
switch provider {
case "aws-s3":
endpoint = "s3." + config.Region + ".amazonaws.com"
case "cloudflare-r2":
endpoint = config.AccountId + ".r2.cloudflarestorage.com"
case "aliyun-oss":
endpoint = "oss-" + config.Region + ".aliyuncs.com"
case "huawei-obs":
endpoint = "obs." + config.Region + ".myhuaweicloud.com"
case "tencent-cos":
endpoint = "cos." + config.Region + ".myqcloud.com"
}
}
secure := config.UseSsl
if parsed, err := url.Parse(endpoint); err == nil && parsed.Host != "" {
secure = parsed.Scheme == "https"
endpoint = parsed.Host
}
if endpoint == "" {
return nil, fmt.Errorf("%s endpoint is required", provider)
}
client, err := minio.New(strings.TrimSuffix(endpoint, "/"), &minio.Options{Creds: credentials.NewStaticV4(config.AccessKey, config.SecretKey, ""), Secure: secure, Region: config.Region, BucketLookup: func() minio.BucketLookupType {
if config.ForcePathStyle {
return minio.BucketLookupPath
}
return minio.BucketLookupAuto
}()})
if err != nil {
return nil, err
}
return &s3Storage{client: client, bucket: config.Bucket, baseURL: strings.TrimSuffix(config.BaseUrl, "/"), prefix: strings.Trim(config.PathPrefix, "/")}, nil
}
func (s *s3Storage) key(name string) string {
if s.prefix == "" {
return strings.TrimPrefix(name, "/")
}
return path.Join(s.prefix, strings.TrimPrefix(name, "/"))
}
func (s *s3Storage) unkey(key string) string {
return strings.TrimPrefix(strings.TrimPrefix(key, s.prefix), "/")
}
func (s *s3Storage) file(key string, size int64) *biz.StoredFile {
name := s.unkey(key)
rawURL := s.baseURL + "/" + key
if s.baseURL == "" {
rawURL = key
}
return &biz.StoredFile{Name: path.Base(name), Path: name, URL: rawURL, Size: size}
}
func (s *s3Storage) Put(ctx context.Context, name string, reader io.Reader) (*biz.StoredFile, error) {
key := s.key(name)
info, err := s.client.PutObject(ctx, s.bucket, key, reader, -1, minio.PutObjectOptions{})
if err != nil {
return nil, err
}
return s.file(key, info.Size), nil
}
func (s *s3Storage) Open(ctx context.Context, name string) (io.ReadCloser, error) {
obj, err := s.client.GetObject(ctx, s.bucket, s.key(name), minio.GetObjectOptions{})
if err != nil {
return nil, err
}
if _, err = obj.Stat(); err != nil {
_ = obj.Close()
return nil, err
}
return obj, nil
}
func (s *s3Storage) Delete(ctx context.Context, name string) error {
return s.client.RemoveObject(ctx, s.bucket, s.key(name), minio.RemoveObjectOptions{})
}
func (s *s3Storage) Compose(ctx context.Context, names []string, destination string) (*biz.StoredFile, string, error) {
return composeFiles(ctx, s, names, destination)
}
func (s *s3Storage) DeletePrefix(ctx context.Context, prefix string) error {
items := s.client.ListObjects(ctx, s.bucket, minio.ListObjectsOptions{Prefix: s.key(prefix), Recursive: true})
for item := range items {
if item.Err != nil {
return item.Err
}
if err := s.client.RemoveObject(ctx, s.bucket, item.Key, minio.RemoveObjectOptions{}); err != nil {
return err
}
}
return nil
}
func (s *s3Storage) List(ctx context.Context, prefix, cursor string, limit int) ([]*biz.StoredFile, string, bool, error) {
if limit < 1 || limit > 1000 {
limit = 100
}
items := s.client.ListObjects(ctx, s.bucket, minio.ListObjectsOptions{Prefix: s.key(prefix), Recursive: true, StartAfter: s.key(cursor)})
out := make([]*biz.StoredFile, 0, limit+1)
for item := range items {
if item.Err != nil {
return nil, "", false, item.Err
}
out = append(out, s.file(item.Key, item.Size))
if len(out) > limit {
break
}
}
more := len(out) > limit
if more {
out = out[:limit]
}
next := ""
if len(out) > 0 {
next = out[len(out)-1].Path
}
return out, next, more, nil
}
+3 -2
View File
@@ -10,6 +10,7 @@ import (
"kra/internal/biz"
"gorm.io/gorm"
"gorm.io/gorm/clause"
)
type dictionaryPO struct {
@@ -334,7 +335,7 @@ func (r *settingsRepo) FindParameter(ctx context.Context, id uint, key string) (
if id != 0 {
err = db.First(&po, id).Error
} else {
err = db.Where("`key` = ?", key).First(&po).Error
err = db.Where(map[string]any{"key": key}).First(&po).Error
}
if err != nil {
return nil, err
@@ -354,7 +355,7 @@ func (r *settingsRepo) ListParameters(ctx context.Context, page, size int, q *bi
db = db.Where("name LIKE ?", "%"+q.Name+"%")
}
if q.Key != "" {
db = db.Where("`key` LIKE ?", "%"+q.Key+"%")
db = db.Where(clause.Like{Column: clause.Column{Name: "key"}, Value: "%" + q.Key + "%"})
}
}
var total int64
+18 -5
View File
@@ -7,6 +7,7 @@ import (
"time"
"kra/internal/biz"
"kra/internal/conf"
"github.com/google/uuid"
"golang.org/x/crypto/bcrypt"
@@ -87,14 +88,22 @@ func (authorityMenuPO) TableName() string { return "sys_authority_menus" }
type systemRepo struct{ data *Data }
func NewSystemRepo(data *Data) biz.SystemRepo { return &systemRepo{data: data} }
func NewSystemRepo(data *Data) biz.SystemRepo { return &systemRepo{data: data} }
func (r *systemRepo) PersistConfig(context.Context) error { return r.data.persistConfig() }
func (r *systemRepo) IsInitialized(ctx context.Context) (bool, error) {
return r.data.gormDB.WithContext(ctx).Migrator().HasTable(&userPO{}), nil
}
func (r *systemRepo) Initialize(ctx context.Context) error {
func (r *systemRepo) Initialize(ctx context.Context, input *biz.DatabaseConfig) error {
config := &conf.Data_Database{Driver: input.Driver, Host: input.Host, Port: input.Port, User: input.User, Password: input.Password, Name: input.Name, Path: input.Path, Config: input.Config}
if err := r.data.switchDatabase(config); err != nil {
return err
}
db := r.data.gormDB.WithContext(ctx)
if err := db.AutoMigrate(&adminPO{}); err != nil {
return err
}
if err := db.AutoMigrate(&userPO{}, &authorityPO{}, &menuPO{}, &userAuthorityPO{}, &authorityMenuPO{}, &apiPO{}, &ignoredAPIPO{}, &authorityAPIPO{}, &menuButtonPO{}, &authorityButtonPO{}, &departmentPO{}, &positionPO{}, &userDepartmentPO{}, &userPositionPO{}, &authorityDepartmentPO{}, &dictionaryPO{}, &dictionaryDetailPO{}, &parameterPO{}, &apiTokenPO{}, &securityConfigPO{}, &versionPO{}, &exportTemplatePO{}, &exportConditionPO{}, &exportJoinPO{}, &operationPO{}, &loginLogPO{}, &dataAccessLogPO{}, &errorRecordPO{}, &taskPO{}, &taskLogPO{}, &mediaPO{}, &categoryPO{}, &uploadSessionPO{}, &uploadChunkPO{}, &announcementPO{}); err != nil {
return err
}
@@ -140,7 +149,7 @@ func (r *systemRepo) Initialize(ctx context.Context) error {
return err
}
if count == 0 {
hash, err := bcrypt.GenerateFromPassword([]byte("123456"), bcrypt.DefaultCost)
hash, err := bcrypt.GenerateFromPassword([]byte(input.AdminPassword), bcrypt.DefaultCost)
if err != nil {
return err
}
@@ -349,9 +358,13 @@ func (r *systemRepo) DeleteUser(ctx context.Context, id uint) error {
return tx.Delete(&userPO{}, id).Error
})
}
func (r *systemRepo) UpdatePassword(ctx context.Context, id uint, password string) error {
func (r *systemRepo) UpdatePassword(ctx context.Context, id uint, password string, clearMustChange bool) error {
now := time.Now()
return r.data.gormDB.WithContext(ctx).Model(&userPO{}).Where("id = ?", id).Updates(map[string]any{"password": password, "password_updated_at": now, "must_change_password": false}).Error
updates := map[string]any{"password": password, "password_updated_at": now}
if clearMustChange {
updates["must_change_password"] = false
}
return r.data.gormDB.WithContext(ctx).Model(&userPO{}).Where("id = ?", id).Updates(updates).Error
}
func (r *systemRepo) ListAuthorities(ctx context.Context) ([]*biz.Authority, error) {
var pos []authorityPO