diff --git a/internal/db/models/file_chunk_dao.go b/internal/db/models/file_chunk_dao.go index 0a3220f1..a78c018d 100644 --- a/internal/db/models/file_chunk_dao.go +++ b/internal/db/models/file_chunk_dao.go @@ -5,6 +5,7 @@ import ( _ "github.com/go-sql-driver/mysql" "github.com/iwind/TeaGo/Tea" "github.com/iwind/TeaGo/dbs" + "github.com/iwind/TeaGo/types" ) type FileChunkDAO dbs.DAO @@ -29,16 +30,19 @@ func init() { } // 创建文件Chunk -func (this *FileChunkDAO) CreateFileChunk(fileId int, data []byte) error { +func (this *FileChunkDAO) CreateFileChunk(fileId int64, data []byte) (int64, error) { op := NewFileChunkOperator() op.FileId = fileId op.Data = data _, err := this.Save(op) - return err + if err != nil { + return 0, err + } + return types.Int64(op.Id), nil } // 列出所有的文件Chunk -func (this *FileChunkDAO) FindAllFileChunks(fileId int) (result []*FileChunk, err error) { +func (this *FileChunkDAO) FindAllFileChunks(fileId int64) (result []*FileChunk, err error) { _, err = this.Query(). Attr("fileId", fileId). AscPk(). @@ -47,8 +51,25 @@ func (this *FileChunkDAO) FindAllFileChunks(fileId int) (result []*FileChunk, er return } +// 读取文件的所有片段ID +func (this *FileChunkDAO) FindAllFileChunkIds(fileId int64) ([]int64, error) { + ones, err := this.Query(). + Attr("fileId", fileId). + AscPk(). + ResultPk(). + FindAll() + if err != nil { + return nil, err + } + result := []int64{} + for _, one := range ones { + result = append(result, int64(one.(*FileChunk).Id)) + } + return result, nil +} + // 删除以前的文件 -func (this *FileChunkDAO) DeleteFileChunks(fileId int) error { +func (this *FileChunkDAO) DeleteFileChunks(fileId int64) error { if fileId <= 0 { return errors.New("invalid fileId") } @@ -57,3 +78,17 @@ func (this *FileChunkDAO) DeleteFileChunks(fileId int) error { Delete() return err } + +// 根据ID查找片段 +func (this *FileChunkDAO) FindFileChunk(chunkId int64) (*FileChunk, error) { + one, err := this.Query(). + Pk(chunkId). + Find() + if err != nil { + return nil, err + } + if one == nil { + return nil, nil + } + return one.(*FileChunk), nil +} diff --git a/internal/db/models/file_dao.go b/internal/db/models/file_dao.go index 0fff9085..ab92dbab 100644 --- a/internal/db/models/file_dao.go +++ b/internal/db/models/file_dao.go @@ -5,8 +5,6 @@ import ( "github.com/iwind/TeaGo/Tea" "github.com/iwind/TeaGo/dbs" "github.com/iwind/TeaGo/types" - "mime/multipart" - "os" ) const ( @@ -66,82 +64,26 @@ func (this *FileDAO) FindEnabledFile(id int64) (*File, error) { } // 创建文件 -func (this *FileDAO) CreateFileFromReader(businessType, description string, filename string, body *multipart.FileHeader, order int) (int, error) { - file, err := body.Open() - if err != nil { - return 0, err - } - +func (this *FileDAO) CreateFile(businessType, description string, filename string, size int64) (int64, error) { op := NewFileOperator() op.Type = businessType op.Description = description op.State = FileStateEnabled - op.Size = body.Size - op.Order = order + op.Size = size op.Filename = filename - _, err = this.Save(op) + _, err := this.Save(op) if err != nil { return 0, err } - fileId := types.Int(op.Id) - - // 保存chunk - buf := make([]byte, 512*1024) - for { - n, err := file.Read(buf) - if n > 0 { - err1 := SharedFileChunkDAO.CreateFileChunk(fileId, buf[:n]) - if err1 != nil { - return 0, err1 - } - } - if err != nil { - break - } - } - - return fileId, nil + return types.Int64(op.Id), nil } -// 创建一个空文件 -func (this *FileDAO) UploadLocalFile(businessType string, localFile string, filename string) (fileId int, err error) { - reader, err := os.Open(localFile) - if err != nil { - return 0, err - } - defer func() { - _ = reader.Close() - }() - stat, err := reader.Stat() - if err != nil { - return 0, err - } - - op := NewFileOperator() - op.Type = businessType - op.Filename = filename - op.Size = stat.Size() - op.State = FileStateEnabled - _, err = this.Save(op) - if err != nil { - return - } - fileId = types.Int(op.Id) - - buf := make([]byte, 512*1024) - for { - n, err := reader.Read(buf) - if n > 0 { - err1 := SharedFileChunkDAO.CreateFileChunk(fileId, buf[:n]) - if err1 != nil { - return 0, err1 - } - } - if err != nil { - break - } - } - - return fileId, nil +// 将文件置为已完成 +func (this *FileDAO) UpdateFileIsFinished(fileId int64) error { + _, err := this.Query(). + Pk(fileId). + Set("isFinished", true). + Update() + return err } diff --git a/internal/db/models/file_model.go b/internal/db/models/file_model.go index e8c93bbf..75cc6988 100644 --- a/internal/db/models/file_model.go +++ b/internal/db/models/file_model.go @@ -12,6 +12,7 @@ type File struct { Order uint32 `field:"order"` // 排序 Type string `field:"type"` // 类型 State uint8 `field:"state"` // 状态 + IsFinished uint8 `field:"isFinished"` // 是否已完成上传 } type FileOperator struct { @@ -25,6 +26,7 @@ type FileOperator struct { Order interface{} // 排序 Type interface{} // 类型 State interface{} // 状态 + IsFinished interface{} // 是否已完成上传 } func NewFileOperator() *FileOperator { diff --git a/internal/db/models/ip_library_dao.go b/internal/db/models/ip_library_dao.go new file mode 100644 index 00000000..b5935e42 --- /dev/null +++ b/internal/db/models/ip_library_dao.go @@ -0,0 +1,104 @@ +package models + +import ( + _ "github.com/go-sql-driver/mysql" + "github.com/iwind/TeaGo/Tea" + "github.com/iwind/TeaGo/dbs" + "github.com/iwind/TeaGo/types" +) + +const ( + IPLibraryStateEnabled = 1 // 已启用 + IPLibraryStateDisabled = 0 // 已禁用 +) + +type IPLibraryDAO dbs.DAO + +func NewIPLibraryDAO() *IPLibraryDAO { + return dbs.NewDAO(&IPLibraryDAO{ + DAOObject: dbs.DAOObject{ + DB: Tea.Env, + Table: "edgeIPLibraries", + Model: new(IPLibrary), + PkName: "id", + }, + }).(*IPLibraryDAO) +} + +var SharedIPLibraryDAO *IPLibraryDAO + +func init() { + dbs.OnReady(func() { + SharedIPLibraryDAO = NewIPLibraryDAO() + }) +} + +// 启用条目 +func (this *IPLibraryDAO) EnableIPLibrary(id int64) error { + _, err := this.Query(). + Pk(id). + Set("state", IPLibraryStateEnabled). + Update() + return err +} + +// 禁用条目 +func (this *IPLibraryDAO) DisableIPLibrary(id int64) error { + _, err := this.Query(). + Pk(id). + Set("state", IPLibraryStateDisabled). + Update() + return err +} + +// 查找启用中的条目 +func (this *IPLibraryDAO) FindEnabledIPLibrary(id int64) (*IPLibrary, error) { + result, err := this.Query(). + Pk(id). + Attr("state", IPLibraryStateEnabled). + Find() + if result == nil { + return nil, err + } + return result.(*IPLibrary), err +} + +// 查找某个类型的IP库列表 +func (this *IPLibraryDAO) FindAllEnabledIPLibrariesWithType(libraryType string) (result []*IPLibrary, err error) { + _, err = this.Query(). + State(IPLibraryStateEnabled). + Attr("type", libraryType). + DescPk(). + Slice(&result). + FindAll() + return +} + +// 查找某个类型的最新的IP库 +func (this *IPLibraryDAO) FindLatestIPLibraryWithType(libraryType string) (*IPLibrary, error) { + one, err := this.Query(). + State(IPLibraryStateEnabled). + Attr("type", libraryType). + DescPk(). + Find() + if err != nil { + return nil, err + } + if one == nil { + return nil, nil + } + return one.(*IPLibrary), nil +} + +// 创建新的IP库 +func (this *IPLibraryDAO) CreateIPLibrary(libraryType string, fileId int64) (int64, error) { + op := NewIPLibraryOperator() + op.Type = libraryType + op.FileId = fileId + op.State = IPLibraryStateEnabled + _, err := this.Save(op) + if err != nil { + return 0, err + } + return types.Int64(op.Id), nil +} diff --git a/internal/db/models/ip_library_dao_test.go b/internal/db/models/ip_library_dao_test.go new file mode 100644 index 00000000..97c24b56 --- /dev/null +++ b/internal/db/models/ip_library_dao_test.go @@ -0,0 +1,5 @@ +package models + +import ( + _ "github.com/go-sql-driver/mysql" +) diff --git a/internal/db/models/ip_library_model.go b/internal/db/models/ip_library_model.go new file mode 100644 index 00000000..908c9555 --- /dev/null +++ b/internal/db/models/ip_library_model.go @@ -0,0 +1,24 @@ +package models + +// IP库 +type IPLibrary struct { + Id uint32 `field:"id"` // ID + AdminId uint32 `field:"adminId"` // 管理员ID + FileId uint32 `field:"fileId"` // 文件ID + Type string `field:"type"` // 类型 + State uint8 `field:"state"` // 状态 + CreatedAt uint64 `field:"createdAt"` // 创建时间 +} + +type IPLibraryOperator struct { + Id interface{} // ID + AdminId interface{} // 管理员ID + FileId interface{} // 文件ID + Type interface{} // 类型 + State interface{} // 状态 + CreatedAt interface{} // 创建时间 +} + +func NewIPLibraryOperator() *IPLibraryOperator { + return &IPLibraryOperator{} +} diff --git a/internal/db/models/ip_library_model_ext.go b/internal/db/models/ip_library_model_ext.go new file mode 100644 index 00000000..2640e7f9 --- /dev/null +++ b/internal/db/models/ip_library_model_ext.go @@ -0,0 +1 @@ +package models diff --git a/internal/nodes/api_node.go b/internal/nodes/api_node.go index f5f005b1..1c9f1370 100644 --- a/internal/nodes/api_node.go +++ b/internal/nodes/api_node.go @@ -173,6 +173,9 @@ func (this *APINode) listenRPC(listener net.Listener, tlsConfig *tls.Config) err pb.RegisterMessageServiceServer(rpcServer, &services.MessageService{}) pb.RegisterNodeGroupServiceServer(rpcServer, &services.NodeGroupService{}) pb.RegisterServerGroupServiceServer(rpcServer, &services.ServerGroupService{}) + pb.RegisterIPLibraryServiceServer(rpcServer, &services.IPLibraryService{}) + pb.RegisterFileChunkServiceServer(rpcServer, &services.FileChunkService{}) + pb.RegisterFileServiceServer(rpcServer, &services.FileService{}) err := rpcServer.Serve(listener) if err != nil { return errors.New("[API]start rpc failed: " + err.Error()) diff --git a/internal/rpc/services/service_file.go b/internal/rpc/services/service_file.go new file mode 100644 index 00000000..4c9d8bef --- /dev/null +++ b/internal/rpc/services/service_file.go @@ -0,0 +1,42 @@ +package services + +import ( + "context" + "github.com/TeaOSLab/EdgeAPI/internal/db/models" + rpcutils "github.com/TeaOSLab/EdgeAPI/internal/rpc/utils" + "github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb" +) + +// 文件相关服务 +type FileService struct { +} + +// 创建文件 +func (this *FileService) CreateFile(ctx context.Context, req *pb.CreateFileRequest) (*pb.CreateFileResponse, error) { + // 校验请求 + _, _, err := rpcutils.ValidateRequest(ctx, rpcutils.UserTypeAdmin) + if err != nil { + return nil, err + } + + fileId, err := models.SharedFileDAO.CreateFile("ipLibrary", "", req.Filename, req.Size) + if err != nil { + return nil, err + } + return &pb.CreateFileResponse{FileId: fileId}, nil +} + +// 将文件置为已完成 +func (this *FileService) UpdateFileFinished(ctx context.Context, req *pb.UpdateFileFinishedRequest) (*pb.RPCUpdateSuccess, error) { + // 校验请求 + _, _, err := rpcutils.ValidateRequest(ctx, rpcutils.UserTypeAdmin) + if err != nil { + return nil, err + } + + err = models.SharedFileDAO.UpdateFileIsFinished(req.FileId) + if err != nil { + return nil, err + } + return rpcutils.RPCUpdateSuccess() +} diff --git a/internal/rpc/services/service_file_chunk.go b/internal/rpc/services/service_file_chunk.go new file mode 100644 index 00000000..ff399e83 --- /dev/null +++ b/internal/rpc/services/service_file_chunk.go @@ -0,0 +1,60 @@ +package services + +import ( + "context" + "github.com/TeaOSLab/EdgeAPI/internal/db/models" + rpcutils "github.com/TeaOSLab/EdgeAPI/internal/rpc/utils" + "github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb" +) + +// 文件片段相关服务 +type FileChunkService struct { +} + +// 创建文件片段 +func (this *FileChunkService) CreateFileChunk(ctx context.Context, req *pb.CreateFileChunkRequest) (*pb.CreateFileChunkResponse, error) { + // 校验请求 + _, _, err := rpcutils.ValidateRequest(ctx, rpcutils.UserTypeAdmin) + if err != nil { + return nil, err + } + + chunkId, err := models.SharedFileChunkDAO.CreateFileChunk(req.FileId, req.Data) + if err != nil { + return nil, err + } + return &pb.CreateFileChunkResponse{FileChunkId: chunkId}, nil +} + +// 获取的一个文件的所有片段IDs +func (this *FileChunkService) FindAllFileChunkIds(ctx context.Context, req *pb.FindAllFileChunkIdsRequest) (*pb.FindAllFileChunkIdsResponse, error) { + // 校验请求 + _, _, err := rpcutils.ValidateRequest(ctx, rpcutils.UserTypeNode, rpcutils.UserTypeAdmin) + if err != nil { + return nil, err + } + + chunkIds, err := models.SharedFileChunkDAO.FindAllFileChunkIds(req.FileId) + if err != nil { + return nil, err + } + return &pb.FindAllFileChunkIdsResponse{FileChunkIds: chunkIds}, nil +} + +// 下载文件片段 +func (this *FileChunkService) DownloadFileChunk(ctx context.Context, req *pb.DownloadFileChunkRequest) (*pb.DownloadFileChunkResponse, error) { + // 校验请求 + _, _, err := rpcutils.ValidateRequest(ctx, rpcutils.UserTypeNode, rpcutils.UserTypeAdmin) + if err != nil { + return nil, err + } + + chunk, err := models.SharedFileChunkDAO.FindFileChunk(req.FileChunkId) + if err != nil { + return nil, err + } + if chunk == nil { + return &pb.DownloadFileChunkResponse{FileChunk: nil}, nil + } + return &pb.DownloadFileChunkResponse{FileChunk: &pb.FileChunk{Data: []byte(chunk.Data)}}, nil +} diff --git a/internal/rpc/services/service_ip_library.go b/internal/rpc/services/service_ip_library.go new file mode 100644 index 00000000..0ea9d58a --- /dev/null +++ b/internal/rpc/services/service_ip_library.go @@ -0,0 +1,163 @@ +package services + +import ( + "context" + "github.com/TeaOSLab/EdgeAPI/internal/db/models" + rpcutils "github.com/TeaOSLab/EdgeAPI/internal/rpc/utils" + "github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb" +) + +// IP库服务 +type IPLibraryService struct { +} + +// 创建IP库 +func (this *IPLibraryService) CreateIPLibrary(ctx context.Context, req *pb.CreateIPLibraryRequest) (*pb.CreateIPLibraryResponse, error) { + // 校验请求 + _, _, err := rpcutils.ValidateRequest(ctx, rpcutils.UserTypeAdmin) + if err != nil { + return nil, err + } + + ipLibraryId, err := models.SharedIPLibraryDAO.CreateIPLibrary(req.Type, req.FileId) + if err != nil { + return nil, err + } + + return &pb.CreateIPLibraryResponse{ + IpLibraryId: ipLibraryId, + }, nil +} + +// 查找单个IP库 +func (this *IPLibraryService) FindEnabledIPLibrary(ctx context.Context, req *pb.FindEnabledIPLibraryRequest) (*pb.FindEnabledIPLibraryResponse, error) { + // 校验请求 + _, _, err := rpcutils.ValidateRequest(ctx, rpcutils.UserTypeAdmin) + if err != nil { + return nil, err + } + + ipLibrary, err := models.SharedIPLibraryDAO.FindEnabledIPLibrary(req.IpLibraryId) + if err != nil { + return nil, err + } + if ipLibrary == nil { + return &pb.FindEnabledIPLibraryResponse{IpLibrary: nil}, nil + } + + // 文件相关 + var pbFile *pb.File = nil + file, err := models.SharedFileDAO.FindEnabledFile(int64(ipLibrary.FileId)) + if err != nil { + return nil, err + } + if file != nil { + pbFile = &pb.File{ + Id: int64(file.Id), + Filename: file.Filename, + Size: int64(file.Size), + } + } + + return &pb.FindEnabledIPLibraryResponse{ + IpLibrary: &pb.IPLibrary{ + Id: int64(ipLibrary.Id), + Type: ipLibrary.Type, + File: pbFile, + CreatedAt: int64(ipLibrary.CreatedAt), + }, + }, nil +} + +// 查找最新的IP库 +func (this *IPLibraryService) FindLatestIPLibraryWithType(ctx context.Context, req *pb.FindLatestIPLibraryWithTypeRequest) (*pb.FindLatestIPLibraryWithTypeResponse, error) { + // 校验请求 + _, _, err := rpcutils.ValidateRequest(ctx, rpcutils.UserTypeNode) + if err != nil { + return nil, err + } + + ipLibrary, err := models.SharedIPLibraryDAO.FindLatestIPLibraryWithType(req.Type) + if err != nil { + return nil, err + } + if ipLibrary == nil { + return &pb.FindLatestIPLibraryWithTypeResponse{IpLibrary: nil}, nil + } + + // 文件相关 + var pbFile *pb.File = nil + file, err := models.SharedFileDAO.FindEnabledFile(int64(ipLibrary.FileId)) + if err != nil { + return nil, err + } + if file != nil { + pbFile = &pb.File{ + Id: int64(file.Id), + Filename: file.Filename, + Size: int64(file.Size), + } + } + + return &pb.FindLatestIPLibraryWithTypeResponse{ + IpLibrary: &pb.IPLibrary{ + Id: int64(ipLibrary.Id), + Type: ipLibrary.Type, + File: pbFile, + CreatedAt: int64(ipLibrary.CreatedAt), + }, + }, nil +} + +// 列出某个类型的所有IP库 +func (this *IPLibraryService) FindAllEnabledIPLibrariesWithType(ctx context.Context, req *pb.FindAllEnabledIPLibrariesWithTypeRequest) (*pb.FindAllEnabledIPLibrariesWithTypeResponse, error) { + // 校验请求 + _, _, err := rpcutils.ValidateRequest(ctx, rpcutils.UserTypeAdmin) + if err != nil { + return nil, err + } + + ipLibraries, err := models.SharedIPLibraryDAO.FindAllEnabledIPLibrariesWithType(req.Type) + if err != nil { + return nil, err + } + result := []*pb.IPLibrary{} + for _, library := range ipLibraries { + // 文件相关 + var pbFile *pb.File = nil + file, err := models.SharedFileDAO.FindEnabledFile(int64(library.FileId)) + if err != nil { + return nil, err + } + if file != nil { + pbFile = &pb.File{ + Id: int64(file.Id), + Filename: file.Filename, + Size: int64(file.Size), + } + } + + result = append(result, &pb.IPLibrary{ + Id: int64(library.Id), + Type: library.Type, + File: pbFile, + CreatedAt: int64(library.CreatedAt), + }) + } + return &pb.FindAllEnabledIPLibrariesWithTypeResponse{IpLibraries: result}, nil +} + +// 删除IP库 +func (this *IPLibraryService) DeleteIPLibrary(ctx context.Context, req *pb.DeleteIPLibraryRequest) (*pb.RPCDeleteSuccess, error) { + // 校验请求 + _, _, err := rpcutils.ValidateRequest(ctx, rpcutils.UserTypeAdmin) + if err != nil { + return nil, err + } + + err = models.SharedIPLibraryDAO.DisableIPLibrary(req.IpLibraryId) + if err != nil { + return nil, err + } + return rpcutils.RPCDeleteSuccess() +}