Add MinIO Package for Object Storage Management
- Introduced a comprehensive MinIO package for managing object storage, including support for hierarchical document storage and general file management. - Implemented core components such as Config, ConnectionManager, and Service, following Clean Architecture principles. - Added functionality for file uploads, downloads, bucket management, and structured logging, enhancing usability and maintainability. - Included detailed error handling and validation for configuration and operations, ensuring robustness. - Updated documentation with usage examples and configuration guidelines to facilitate integration and understanding of the package's functionality.
This commit is contained in:
Nima Nakhostin
@@ -0,0 +1,501 @@
|
||||
package minio
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
"mime"
|
||||
"net/url"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/minio/minio-go/v7"
|
||||
)
|
||||
|
||||
// Service provides MinIO file storage operations
|
||||
type Service struct {
|
||||
connMgr *ConnectionManager
|
||||
config Config
|
||||
logger Logger
|
||||
}
|
||||
|
||||
// NewService creates a new MinIO service
|
||||
func NewService(connMgr *ConnectionManager, config Config, logger Logger) *Service {
|
||||
return &Service{
|
||||
connMgr: connMgr,
|
||||
config: config,
|
||||
logger: logger,
|
||||
}
|
||||
}
|
||||
|
||||
// Upload uploads a file to the specified bucket
|
||||
func (s *Service) Upload(ctx context.Context, bucket, objectName string, reader io.Reader, options UploadOptions) error {
|
||||
if !s.connMgr.IsConnected() {
|
||||
return ErrNotConnected
|
||||
}
|
||||
|
||||
client := s.connMgr.GetClient().(*minio.Client)
|
||||
|
||||
// Set default content type if not provided
|
||||
contentType := options.ContentType
|
||||
if contentType == "" {
|
||||
contentType = mime.TypeByExtension(filepath.Ext(objectName))
|
||||
if contentType == "" {
|
||||
contentType = "application/octet-stream"
|
||||
}
|
||||
}
|
||||
|
||||
// Upload the file
|
||||
_, err := client.PutObject(ctx, bucket, objectName, reader, -1, minio.PutObjectOptions{
|
||||
ContentType: contentType,
|
||||
UserMetadata: options.Metadata,
|
||||
})
|
||||
if err != nil {
|
||||
s.logger.Error("Failed to upload file", map[string]interface{}{
|
||||
"bucket": bucket,
|
||||
"objectName": objectName,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return NewErrorWithCause(ErrCodeUploadFailed, "failed to upload file", err)
|
||||
}
|
||||
|
||||
s.logger.Info("File uploaded successfully", map[string]interface{}{
|
||||
"bucket": bucket,
|
||||
"objectName": objectName,
|
||||
"contentType": contentType,
|
||||
})
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// UploadHierarchical uploads a file to hierarchical storage with classification
|
||||
func (s *Service) UploadHierarchical(ctx context.Context, file io.Reader, filename string, options HierarchicalUploadOptions) (string, error) {
|
||||
if !s.connMgr.IsConnected() {
|
||||
return "", ErrNotConnected
|
||||
}
|
||||
|
||||
// Generate hierarchical path
|
||||
objectPath := s.GetDocumentPath(options.Category, options.SubCategory, filename, options.Tags)
|
||||
|
||||
// Ensure bucket exists
|
||||
err := s.connMgr.CreateBucket(ctx, s.config.HierarchicalBucket)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("failed to create hierarchical bucket: %w", err)
|
||||
}
|
||||
|
||||
// Set metadata for classification
|
||||
metadata := options.Metadata
|
||||
if metadata == nil {
|
||||
metadata = make(map[string]string)
|
||||
}
|
||||
metadata["category"] = options.Category
|
||||
metadata["subcategory"] = options.SubCategory
|
||||
if len(options.Tags) > 0 {
|
||||
metadata["tags"] = strings.Join(options.Tags, ",")
|
||||
}
|
||||
metadata["original_filename"] = filename
|
||||
metadata["uploaded_at"] = fmt.Sprintf("%d", time.Now().Unix())
|
||||
|
||||
uploadOptions := UploadOptions{
|
||||
ContentType: options.ContentType,
|
||||
Metadata: metadata,
|
||||
}
|
||||
|
||||
err = s.Upload(ctx, s.config.HierarchicalBucket, objectPath, file, uploadOptions)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
s.logger.Info("Hierarchical file uploaded successfully", map[string]interface{}{
|
||||
"bucket": s.config.HierarchicalBucket,
|
||||
"objectPath": objectPath,
|
||||
"category": options.Category,
|
||||
"subCategory": options.SubCategory,
|
||||
})
|
||||
|
||||
return objectPath, nil
|
||||
}
|
||||
|
||||
// Download downloads a file from the specified bucket
|
||||
func (s *Service) Download(ctx context.Context, bucket, objectName string, options DownloadOptions) (io.ReadCloser, error) {
|
||||
if !s.connMgr.IsConnected() {
|
||||
return nil, ErrNotConnected
|
||||
}
|
||||
|
||||
client := s.connMgr.GetClient().(*minio.Client)
|
||||
|
||||
getOptions := minio.GetObjectOptions{}
|
||||
if options.Range != nil {
|
||||
getOptions.SetRange(options.Range.Start, options.Range.End)
|
||||
}
|
||||
|
||||
object, err := client.GetObject(ctx, bucket, objectName, getOptions)
|
||||
if err != nil {
|
||||
s.logger.Error("Failed to download file", map[string]interface{}{
|
||||
"bucket": bucket,
|
||||
"objectName": objectName,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return nil, NewErrorWithCause(ErrCodeDownloadFailed, "failed to download file", err)
|
||||
}
|
||||
|
||||
s.logger.Debug("File download initiated", map[string]interface{}{
|
||||
"bucket": bucket,
|
||||
"objectName": objectName,
|
||||
})
|
||||
|
||||
return object, nil
|
||||
}
|
||||
|
||||
// Delete deletes a file from the specified bucket
|
||||
func (s *Service) Delete(ctx context.Context, bucket, objectName string) error {
|
||||
if !s.connMgr.IsConnected() {
|
||||
return ErrNotConnected
|
||||
}
|
||||
|
||||
client := s.connMgr.GetClient().(*minio.Client)
|
||||
|
||||
err := client.RemoveObject(ctx, bucket, objectName, minio.RemoveObjectOptions{})
|
||||
if err != nil {
|
||||
s.logger.Error("Failed to delete file", map[string]interface{}{
|
||||
"bucket": bucket,
|
||||
"objectName": objectName,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return NewErrorWithCause(ErrCodeDeleteFailed, "failed to delete file", err)
|
||||
}
|
||||
|
||||
s.logger.Info("File deleted successfully", map[string]interface{}{
|
||||
"bucket": bucket,
|
||||
"objectName": objectName,
|
||||
})
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// GetInfo gets information about a file
|
||||
func (s *Service) GetInfo(ctx context.Context, bucket, objectName string) (*FileInfo, error) {
|
||||
if !s.connMgr.IsConnected() {
|
||||
return nil, ErrNotConnected
|
||||
}
|
||||
|
||||
client := s.connMgr.GetClient().(*minio.Client)
|
||||
|
||||
stat, err := client.StatObject(ctx, bucket, objectName, minio.StatObjectOptions{})
|
||||
if err != nil {
|
||||
s.logger.Error("Failed to get file info", map[string]interface{}{
|
||||
"bucket": bucket,
|
||||
"objectName": objectName,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return nil, NewErrorWithCause(ErrCodeObjectNotFound, "failed to get file info", err)
|
||||
}
|
||||
|
||||
fileInfo := &FileInfo{
|
||||
Name: filepath.Base(objectName),
|
||||
Path: objectName,
|
||||
Size: stat.Size,
|
||||
ContentType: stat.ContentType,
|
||||
ETag: stat.ETag,
|
||||
LastModified: stat.LastModified.Unix(),
|
||||
Metadata: stat.UserMetadata,
|
||||
}
|
||||
|
||||
return fileInfo, nil
|
||||
}
|
||||
|
||||
// List lists files in a bucket with optional prefix
|
||||
func (s *Service) List(ctx context.Context, bucket string, options ListOptions) (*ListResult, error) {
|
||||
if !s.connMgr.IsConnected() {
|
||||
return nil, ErrNotConnected
|
||||
}
|
||||
|
||||
client := s.connMgr.GetClient().(*minio.Client)
|
||||
|
||||
listOptions := minio.ListObjectsOptions{
|
||||
Prefix: options.Prefix,
|
||||
Recursive: false,
|
||||
}
|
||||
|
||||
if options.Delimiter != "" {
|
||||
listOptions = minio.ListObjectsOptions{
|
||||
Prefix: options.Prefix,
|
||||
Recursive: false,
|
||||
UseV1: true,
|
||||
}
|
||||
}
|
||||
|
||||
if options.MaxKeys > 0 {
|
||||
listOptions.MaxKeys = int(options.MaxKeys)
|
||||
}
|
||||
|
||||
objectCh := client.ListObjects(ctx, bucket, listOptions)
|
||||
|
||||
var objects []FileInfo
|
||||
var commonPrefixes []string
|
||||
|
||||
for object := range objectCh {
|
||||
if object.Err != nil {
|
||||
s.logger.Error("Error listing objects", map[string]interface{}{
|
||||
"bucket": bucket,
|
||||
"prefix": options.Prefix,
|
||||
"error": object.Err.Error(),
|
||||
})
|
||||
return nil, NewErrorWithCause("LIST_FAILED", "failed to list objects", object.Err)
|
||||
}
|
||||
|
||||
if strings.HasSuffix(object.Key, "/") {
|
||||
// This is a "directory" (common prefix)
|
||||
commonPrefixes = append(commonPrefixes, object.Key)
|
||||
} else {
|
||||
// This is a file
|
||||
fileInfo := FileInfo{
|
||||
Name: filepath.Base(object.Key),
|
||||
Path: object.Key,
|
||||
Size: object.Size,
|
||||
ContentType: "", // Would need to get from StatObject if needed
|
||||
ETag: object.ETag,
|
||||
LastModified: object.LastModified.Unix(),
|
||||
Metadata: nil,
|
||||
}
|
||||
objects = append(objects, fileInfo)
|
||||
}
|
||||
}
|
||||
|
||||
result := &ListResult{
|
||||
Objects: objects,
|
||||
CommonPrefixes: commonPrefixes,
|
||||
IsTruncated: false, // MinIO Go client doesn't provide this info directly
|
||||
NextMarker: "",
|
||||
}
|
||||
|
||||
s.logger.Debug("Objects listed successfully", map[string]interface{}{
|
||||
"bucket": bucket,
|
||||
"prefix": options.Prefix,
|
||||
"objects_count": len(objects),
|
||||
"prefixes_count": len(commonPrefixes),
|
||||
})
|
||||
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// Copy copies a file within the same bucket or to another bucket
|
||||
func (s *Service) Copy(ctx context.Context, srcBucket, srcObject, destBucket, destObject string) error {
|
||||
if !s.connMgr.IsConnected() {
|
||||
return ErrNotConnected
|
||||
}
|
||||
|
||||
client := s.connMgr.GetClient().(*minio.Client)
|
||||
|
||||
src := minio.CopySrcOptions{
|
||||
Bucket: srcBucket,
|
||||
Object: srcObject,
|
||||
}
|
||||
dst := minio.CopyDestOptions{
|
||||
Bucket: destBucket,
|
||||
Object: destObject,
|
||||
}
|
||||
|
||||
_, err := client.CopyObject(ctx, dst, src)
|
||||
if err != nil {
|
||||
s.logger.Error("Failed to copy file", map[string]interface{}{
|
||||
"srcBucket": srcBucket,
|
||||
"srcObject": srcObject,
|
||||
"destBucket": destBucket,
|
||||
"destObject": destObject,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return NewErrorWithCause(ErrCodeCopyFailed, "failed to copy file", err)
|
||||
}
|
||||
|
||||
s.logger.Info("File copied successfully", map[string]interface{}{
|
||||
"srcBucket": srcBucket,
|
||||
"srcObject": srcObject,
|
||||
"destBucket": destBucket,
|
||||
"destObject": destObject,
|
||||
})
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Move moves a file (copy + delete)
|
||||
func (s *Service) Move(ctx context.Context, srcBucket, srcObject, destBucket, destObject string) error {
|
||||
err := s.Copy(ctx, srcBucket, srcObject, destBucket, destObject)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return s.Delete(ctx, srcBucket, srcObject)
|
||||
}
|
||||
|
||||
// Exists checks if a file exists
|
||||
func (s *Service) Exists(ctx context.Context, bucket, objectName string) (bool, error) {
|
||||
_, err := s.GetInfo(ctx, bucket, objectName)
|
||||
if err != nil {
|
||||
if strings.Contains(err.Error(), "NoSuchKey") || strings.Contains(err.Error(), "not found") {
|
||||
return false, nil
|
||||
}
|
||||
return false, err
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// GetPresignedURL generates a presigned URL for temporary access
|
||||
func (s *Service) GetPresignedURL(ctx context.Context, bucket, objectName string, expirySeconds int64) (string, error) {
|
||||
if !s.connMgr.IsConnected() {
|
||||
return "", ErrNotConnected
|
||||
}
|
||||
|
||||
client := s.connMgr.GetClient().(*minio.Client)
|
||||
|
||||
reqParams := make(url.Values)
|
||||
presignedURL, err := client.PresignedGetObject(ctx, bucket, objectName, time.Duration(expirySeconds)*time.Second, reqParams)
|
||||
if err != nil {
|
||||
s.logger.Error("Failed to generate presigned URL", map[string]interface{}{
|
||||
"bucket": bucket,
|
||||
"objectName": objectName,
|
||||
"expiry": expirySeconds,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return "", NewErrorWithCause(ErrCodePresignFailed, "failed to generate presigned URL", err)
|
||||
}
|
||||
|
||||
return presignedURL.String(), nil
|
||||
}
|
||||
|
||||
// UploadDocument uploads a document with hierarchical classification
|
||||
func (s *Service) UploadDocument(ctx context.Context, file io.Reader, filename string, options HierarchicalUploadOptions) (string, error) {
|
||||
return s.UploadHierarchical(ctx, file, filename, options)
|
||||
}
|
||||
|
||||
// ListDocuments lists documents in a category/subcategory
|
||||
func (s *Service) ListDocuments(ctx context.Context, category, subCategory string, options ListOptions) (*ListResult, error) {
|
||||
// Build prefix for hierarchical listing
|
||||
prefix := s.buildCategoryPrefix(category, subCategory)
|
||||
if prefix != "" && !strings.HasSuffix(prefix, s.config.PathSeparator) {
|
||||
prefix += s.config.PathSeparator
|
||||
}
|
||||
|
||||
listOptions := ListOptions{
|
||||
Prefix: prefix,
|
||||
Delimiter: s.config.PathSeparator,
|
||||
MaxKeys: options.MaxKeys,
|
||||
Marker: options.Marker,
|
||||
}
|
||||
|
||||
return s.List(ctx, s.config.HierarchicalBucket, listOptions)
|
||||
}
|
||||
|
||||
// GetDocumentPath generates the hierarchical path for a document
|
||||
func (s *Service) GetDocumentPath(category, subCategory, filename string, tags []string) string {
|
||||
var pathParts []string
|
||||
|
||||
if category != "" {
|
||||
pathParts = append(pathParts, category)
|
||||
}
|
||||
|
||||
if subCategory != "" {
|
||||
pathParts = append(pathParts, subCategory)
|
||||
}
|
||||
|
||||
if len(tags) > 0 {
|
||||
// Create a subdirectory based on tags
|
||||
tagDir := strings.Join(tags, "_")
|
||||
if tagDir != "" {
|
||||
pathParts = append(pathParts, tagDir)
|
||||
}
|
||||
}
|
||||
|
||||
// Add timestamp for uniqueness
|
||||
timestamp := fmt.Sprintf("%d", time.Now().Unix())
|
||||
pathParts = append(pathParts, timestamp)
|
||||
|
||||
// Add filename
|
||||
pathParts = append(pathParts, filename)
|
||||
|
||||
return strings.Join(pathParts, s.config.PathSeparator)
|
||||
}
|
||||
|
||||
// SearchDocuments searches documents by tags or metadata (basic implementation)
|
||||
func (s *Service) SearchDocuments(ctx context.Context, query string, category, subCategory string, options ListOptions) (*ListResult, error) {
|
||||
// For now, do a prefix-based search
|
||||
// In a real implementation, you might want to use MinIO's search capabilities or index metadata
|
||||
prefix := s.buildCategoryPrefix(category, subCategory)
|
||||
|
||||
listOptions := ListOptions{
|
||||
Prefix: prefix,
|
||||
Delimiter: s.config.PathSeparator,
|
||||
MaxKeys: options.MaxKeys,
|
||||
Marker: options.Marker,
|
||||
}
|
||||
|
||||
result, err := s.List(ctx, s.config.HierarchicalBucket, listOptions)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Basic filtering based on query (could be enhanced)
|
||||
if query != "" {
|
||||
var filteredObjects []FileInfo
|
||||
for _, obj := range result.Objects {
|
||||
// Check if query matches filename or metadata
|
||||
if strings.Contains(strings.ToLower(obj.Name), strings.ToLower(query)) {
|
||||
filteredObjects = append(filteredObjects, obj)
|
||||
} else {
|
||||
// Check metadata
|
||||
for _, value := range obj.Metadata {
|
||||
if strings.Contains(strings.ToLower(value), strings.ToLower(query)) {
|
||||
filteredObjects = append(filteredObjects, obj)
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
result.Objects = filteredObjects
|
||||
}
|
||||
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// buildCategoryPrefix builds the prefix for category-based searches
|
||||
func (s *Service) buildCategoryPrefix(category, subCategory string) string {
|
||||
var parts []string
|
||||
|
||||
if category != "" {
|
||||
parts = append(parts, category)
|
||||
}
|
||||
|
||||
if subCategory != "" {
|
||||
parts = append(parts, subCategory)
|
||||
}
|
||||
|
||||
if len(parts) == 0 {
|
||||
return ""
|
||||
}
|
||||
|
||||
return strings.Join(parts, s.config.PathSeparator)
|
||||
}
|
||||
|
||||
// validateFileType validates if the file type is allowed
|
||||
func (s *Service) validateFileType(contentType string) error {
|
||||
if len(s.config.AllowedTypes) == 1 && s.config.AllowedTypes[0] == "*" {
|
||||
return nil // Allow all types
|
||||
}
|
||||
|
||||
for _, allowedType := range s.config.AllowedTypes {
|
||||
if allowedType == contentType || strings.HasPrefix(contentType, allowedType) {
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
return NewError(ErrCodeInvalidFileType, fmt.Sprintf("file type %s is not allowed", contentType))
|
||||
}
|
||||
|
||||
// validateFileSize validates if the file size is within limits
|
||||
func (s *Service) validateFileSize(size int64) error {
|
||||
if size > s.config.MaxFileSize {
|
||||
return NewError(ErrCodeFileTooLarge, fmt.Sprintf("file size %d exceeds maximum allowed size %d", size, s.config.MaxFileSize))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
Reference in New Issue
Block a user