Files
Nima Nakhostin 1a56070556 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.
2025-11-09 09:52:09 +03:30

502 lines
14 KiB
Go

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
}