implemented mongo gridFS for file upload
This commit is contained in:
m.nazemi
@@ -0,0 +1,512 @@
|
||||
package filestore
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"io"
|
||||
"time"
|
||||
|
||||
"tm/pkg/logger"
|
||||
"tm/pkg/mongo"
|
||||
|
||||
"go.mongodb.org/mongo-driver/v2/bson"
|
||||
mongopkg "go.mongodb.org/mongo-driver/v2/mongo"
|
||||
"go.mongodb.org/mongo-driver/v2/mongo/options"
|
||||
)
|
||||
|
||||
// GridFSService implements FileStoreService using MongoDB GridFS
|
||||
type GridFSService struct {
|
||||
database mongopkg.Database
|
||||
logger logger.Logger
|
||||
}
|
||||
|
||||
// NewGridFSService creates a new GridFS file store service
|
||||
func NewGridFSService(mongoManager *mongo.ConnectionManager, logger logger.Logger) (*GridFSService, error) {
|
||||
if !mongoManager.IsConnected() {
|
||||
return nil, NewError(ErrCodeUploadFailed, "MongoDB connection not available")
|
||||
}
|
||||
|
||||
database := mongoManager.GetDatabase()
|
||||
if database == nil {
|
||||
return nil, NewError(ErrCodeUploadFailed, "MongoDB database not available")
|
||||
}
|
||||
|
||||
logger.Info("GridFS service initialized successfully", map[string]interface{}{
|
||||
"storage_type": "MongoDB GridFS",
|
||||
})
|
||||
|
||||
return &GridFSService{
|
||||
database: *database,
|
||||
logger: logger,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// Upload stores a file in GridFS
|
||||
func (s *GridFSService) Upload(fileID, filename string, reader io.Reader, options UploadOptions) (string, error) {
|
||||
// Validate input
|
||||
if filename == "" {
|
||||
return "", ErrInvalidInput
|
||||
}
|
||||
|
||||
if options.ContentType == "" {
|
||||
options.ContentType = "application/octet-stream"
|
||||
}
|
||||
|
||||
// Check file size if max size is set
|
||||
if options.MaxSize > 0 {
|
||||
// Read into buffer to check size
|
||||
data, err := io.ReadAll(reader)
|
||||
if err != nil {
|
||||
s.logger.Error("Failed to read file for size check", map[string]interface{}{
|
||||
"filename": filename,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return "", NewErrorWithCause(ErrCodeUploadFailed, "failed to read file", err)
|
||||
}
|
||||
|
||||
if int64(len(data)) > options.MaxSize {
|
||||
s.logger.Warn("File exceeds maximum size", map[string]interface{}{
|
||||
"filename": filename,
|
||||
"size": len(data),
|
||||
"max_size": options.MaxSize,
|
||||
})
|
||||
return "", ErrFileTooLarge
|
||||
}
|
||||
|
||||
reader = bytes.NewReader(data)
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
|
||||
// Create files and chunks collections
|
||||
filesCollection := s.database.Collection("fs.files")
|
||||
chunksCollection := s.database.Collection("fs.chunks")
|
||||
|
||||
// Generate file ID if not provided
|
||||
var objectID bson.ObjectID
|
||||
var err error
|
||||
if fileID != "" {
|
||||
objectID, err = bson.ObjectIDFromHex(fileID)
|
||||
if err != nil {
|
||||
s.logger.Error("Invalid file ID format", map[string]interface{}{
|
||||
"file_id": fileID,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return "", NewErrorWithCause(ErrCodeInvalidInput, "invalid file ID format", err)
|
||||
}
|
||||
} else {
|
||||
objectID = bson.NewObjectID()
|
||||
}
|
||||
|
||||
// Read file data
|
||||
fileData, err := io.ReadAll(reader)
|
||||
if err != nil {
|
||||
s.logger.Error("Failed to read file data", map[string]interface{}{
|
||||
"filename": filename,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return "", NewErrorWithCause(ErrCodeUploadFailed, "failed to read file", err)
|
||||
}
|
||||
|
||||
// Prepare metadata
|
||||
metadata := bson.M{
|
||||
"uploaded_at": time.Now().Unix(),
|
||||
"content_type": options.ContentType,
|
||||
}
|
||||
|
||||
if options.UploadedBy != "" {
|
||||
metadata["uploaded_by"] = options.UploadedBy
|
||||
}
|
||||
if options.Category != "" {
|
||||
metadata["category"] = options.Category
|
||||
}
|
||||
if len(options.Tags) > 0 {
|
||||
metadata["tags"] = options.Tags
|
||||
}
|
||||
if options.Description != "" {
|
||||
metadata["description"] = options.Description
|
||||
}
|
||||
if len(options.Extra) > 0 {
|
||||
metadata["extra"] = options.Extra
|
||||
}
|
||||
|
||||
// Split file into chunks (255 KB default chunk size)
|
||||
chunkSize := 255 * 1024
|
||||
numChunks := (len(fileData) + chunkSize - 1) / chunkSize
|
||||
|
||||
// Insert chunks
|
||||
for i := 0; i < numChunks; i++ {
|
||||
start := i * chunkSize
|
||||
end := start + chunkSize
|
||||
if end > len(fileData) {
|
||||
end = len(fileData)
|
||||
}
|
||||
|
||||
chunk := bson.M{
|
||||
"files_id": objectID,
|
||||
"n": int32(i),
|
||||
"data": fileData[start:end],
|
||||
}
|
||||
|
||||
_, err := chunksCollection.InsertOne(ctx, chunk)
|
||||
if err != nil {
|
||||
s.logger.Error("Failed to insert chunk", map[string]interface{}{
|
||||
"filename": filename,
|
||||
"chunk": i,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return "", NewErrorWithCause(ErrCodeUploadFailed, "failed to store file chunks", err)
|
||||
}
|
||||
}
|
||||
|
||||
// Insert file record
|
||||
fileRecord := bson.M{
|
||||
"_id": objectID,
|
||||
"filename": filename,
|
||||
"uploadDate": time.Now(),
|
||||
"chunkSize": int32(chunkSize),
|
||||
"length": int32(len(fileData)),
|
||||
"md5": "", // MongoDB would calculate this, we can leave empty
|
||||
"metadata": metadata,
|
||||
}
|
||||
|
||||
_, err = filesCollection.InsertOne(ctx, fileRecord)
|
||||
if err != nil {
|
||||
s.logger.Error("Failed to insert file record", map[string]interface{}{
|
||||
"filename": filename,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return "", NewErrorWithCause(ErrCodeUploadFailed, "failed to upload file", err)
|
||||
}
|
||||
|
||||
uploadedFileID := objectID.Hex()
|
||||
s.logger.Info("File uploaded successfully to GridFS", map[string]interface{}{
|
||||
"file_id": uploadedFileID,
|
||||
"filename": filename,
|
||||
"content_type": options.ContentType,
|
||||
"uploaded_by": options.UploadedBy,
|
||||
"category": options.Category,
|
||||
})
|
||||
|
||||
return uploadedFileID, nil
|
||||
}
|
||||
|
||||
// Download retrieves a file from GridFS
|
||||
func (s *GridFSService) Download(fileID string) (io.ReadCloser, *FileInfo, error) {
|
||||
if fileID == "" {
|
||||
return nil, nil, ErrInvalidInput
|
||||
}
|
||||
|
||||
objectID, err := bson.ObjectIDFromHex(fileID)
|
||||
if err != nil {
|
||||
s.logger.Error("Invalid file ID format", map[string]interface{}{
|
||||
"file_id": fileID,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return nil, nil, NewErrorWithCause(ErrCodeInvalidInput, "invalid file ID format", err)
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
|
||||
// Get file info first
|
||||
fileInfo, err := s.GetFileInfo(fileID)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
// Get all chunks for this file
|
||||
chunksCollection := s.database.Collection("fs.chunks")
|
||||
findOpts := options.Find().SetSort(bson.D{{Key: "n", Value: 1}})
|
||||
cursor, err := chunksCollection.Find(ctx, bson.M{"files_id": objectID}, findOpts)
|
||||
if err != nil {
|
||||
s.logger.Error("Failed to find chunks", map[string]interface{}{
|
||||
"file_id": fileID,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return nil, nil, NewErrorWithCause(ErrCodeDownloadFailed, "failed to download file", err)
|
||||
}
|
||||
defer cursor.Close(ctx)
|
||||
|
||||
// Read all chunks and assemble file
|
||||
var chunks []bson.M
|
||||
if err := cursor.All(ctx, &chunks); err != nil {
|
||||
s.logger.Error("Failed to read chunks", map[string]interface{}{
|
||||
"file_id": fileID,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return nil, nil, NewErrorWithCause(ErrCodeDownloadFailed, "failed to read file data", err)
|
||||
}
|
||||
|
||||
// Assemble file from chunks
|
||||
var fileData bytes.Buffer
|
||||
for _, chunk := range chunks {
|
||||
data, ok := chunk["data"].(bson.Binary)
|
||||
if !ok {
|
||||
return nil, nil, NewError(ErrCodeDownloadFailed, "invalid chunk data format")
|
||||
}
|
||||
fileData.Write(data.Data)
|
||||
}
|
||||
|
||||
s.logger.Info("File downloaded successfully from GridFS", map[string]interface{}{
|
||||
"file_id": fileID,
|
||||
"filename": fileInfo.Filename,
|
||||
})
|
||||
|
||||
return io.NopCloser(&fileData), fileInfo, nil
|
||||
}
|
||||
|
||||
// Delete removes a file from GridFS
|
||||
func (s *GridFSService) Delete(fileID string) error {
|
||||
if fileID == "" {
|
||||
return ErrInvalidInput
|
||||
}
|
||||
|
||||
objectID, err := bson.ObjectIDFromHex(fileID)
|
||||
if err != nil {
|
||||
s.logger.Error("Invalid file ID format", map[string]interface{}{
|
||||
"file_id": fileID,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return NewErrorWithCause(ErrCodeInvalidInput, "invalid file ID format", err)
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
|
||||
// Delete file record
|
||||
filesCollection := s.database.Collection("fs.files")
|
||||
_, err = filesCollection.DeleteOne(ctx, bson.M{"_id": objectID})
|
||||
if err != nil {
|
||||
s.logger.Error("Failed to delete file record", map[string]interface{}{
|
||||
"file_id": fileID,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return NewErrorWithCause(ErrCodeDeleteFailed, "failed to delete file", err)
|
||||
}
|
||||
|
||||
// Delete associated chunks
|
||||
chunksCollection := s.database.Collection("fs.chunks")
|
||||
_, err = chunksCollection.DeleteMany(ctx, bson.M{"files_id": objectID})
|
||||
if err != nil {
|
||||
s.logger.Error("Failed to delete file chunks", map[string]interface{}{
|
||||
"file_id": fileID,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return NewErrorWithCause(ErrCodeDeleteFailed, "failed to delete file chunks", err)
|
||||
}
|
||||
|
||||
s.logger.Info("File deleted successfully from GridFS", map[string]interface{}{
|
||||
"file_id": fileID,
|
||||
})
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// GetFileInfo retrieves metadata for a file
|
||||
func (s *GridFSService) GetFileInfo(fileID string) (*FileInfo, error) {
|
||||
if fileID == "" {
|
||||
return nil, ErrInvalidInput
|
||||
}
|
||||
|
||||
objectID, err := bson.ObjectIDFromHex(fileID)
|
||||
if err != nil {
|
||||
s.logger.Error("Invalid file ID format", map[string]interface{}{
|
||||
"file_id": fileID,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return nil, NewErrorWithCause(ErrCodeInvalidInput, "invalid file ID format", err)
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
|
||||
// Get file record from fs.files collection
|
||||
filesCollection := s.database.Collection("fs.files")
|
||||
var fileRecord bson.M
|
||||
err = filesCollection.FindOne(ctx, bson.M{"_id": objectID}).Decode(&fileRecord)
|
||||
if err != nil {
|
||||
s.logger.Error("Failed to find file", map[string]interface{}{
|
||||
"file_id": fileID,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return nil, NewErrorWithCause(ErrCodeNotFound, "file not found", err)
|
||||
}
|
||||
|
||||
// Extract upload date
|
||||
var uploadDate time.Time
|
||||
if ud, ok := fileRecord["uploadDate"].(time.Time); ok {
|
||||
uploadDate = ud
|
||||
}
|
||||
|
||||
// Extract metadata
|
||||
var metadata map[string]interface{}
|
||||
if md, ok := fileRecord["metadata"].(bson.M); ok {
|
||||
metadata = md
|
||||
}
|
||||
|
||||
fileInfo := &FileInfo{
|
||||
FileID: fileID,
|
||||
Filename: fileRecord["filename"].(string),
|
||||
ContentType: "application/octet-stream",
|
||||
Size: int64(fileRecord["length"].(int32)),
|
||||
UploadedAt: uploadDate,
|
||||
Metadata: metadata,
|
||||
}
|
||||
|
||||
// Extract metadata fields if they exist
|
||||
if metadata != nil {
|
||||
if ct, ok := metadata["content_type"].(string); ok {
|
||||
fileInfo.ContentType = ct
|
||||
}
|
||||
if ub, ok := metadata["uploaded_by"].(string); ok {
|
||||
fileInfo.UploadedBy = ub
|
||||
}
|
||||
if cat, ok := metadata["category"].(string); ok {
|
||||
fileInfo.Category = cat
|
||||
}
|
||||
if tags, ok := metadata["tags"].([]interface{}); ok {
|
||||
fileInfo.Tags = make([]string, len(tags))
|
||||
for i, tag := range tags {
|
||||
if tagStr, ok := tag.(string); ok {
|
||||
fileInfo.Tags[i] = tagStr
|
||||
}
|
||||
}
|
||||
}
|
||||
if desc, ok := metadata["description"].(string); ok {
|
||||
fileInfo.Description = desc
|
||||
}
|
||||
if extra, ok := metadata["extra"].(bson.M); ok {
|
||||
fileInfo.Extra = make(map[string]string)
|
||||
for k, v := range extra {
|
||||
if vs, ok := v.(string); ok {
|
||||
fileInfo.Extra[k] = vs
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return fileInfo, nil
|
||||
}
|
||||
|
||||
// ListFiles lists files by category with optional filtering
|
||||
func (s *GridFSService) ListFiles(category string, tags []string, limit int64, skip int64) ([]*FileInfo, int64, error) {
|
||||
if limit <= 0 {
|
||||
limit = 50
|
||||
}
|
||||
if limit > 1000 {
|
||||
limit = 1000
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
|
||||
// Build filter
|
||||
filter := bson.M{}
|
||||
if category != "" {
|
||||
filter["metadata.category"] = category
|
||||
}
|
||||
if len(tags) > 0 {
|
||||
filter["metadata.tags"] = bson.M{"$in": tags}
|
||||
}
|
||||
|
||||
// Find files
|
||||
filesCollection := s.database.Collection("fs.files")
|
||||
|
||||
// Count total
|
||||
total, err := filesCollection.CountDocuments(ctx, filter)
|
||||
if err != nil {
|
||||
s.logger.Error("Failed to count files", map[string]interface{}{
|
||||
"error": err.Error(),
|
||||
})
|
||||
return nil, 0, NewErrorWithCause(ErrCodeDownloadFailed, "failed to count files", err)
|
||||
}
|
||||
|
||||
// Find and list
|
||||
opts := options.Find().SetSkip(skip).SetLimit(limit)
|
||||
|
||||
cursor, err := filesCollection.Find(ctx, filter, opts)
|
||||
if err != nil {
|
||||
s.logger.Error("Failed to list files", map[string]interface{}{
|
||||
"error": err.Error(),
|
||||
})
|
||||
return nil, 0, NewErrorWithCause(ErrCodeDownloadFailed, "failed to list files", err)
|
||||
}
|
||||
defer cursor.Close(ctx)
|
||||
|
||||
var fileRecords []bson.M
|
||||
if err := cursor.All(ctx, &fileRecords); err != nil {
|
||||
s.logger.Error("Failed to decode files", map[string]interface{}{
|
||||
"error": err.Error(),
|
||||
})
|
||||
return nil, 0, NewErrorWithCause(ErrCodeDownloadFailed, "failed to decode files", err)
|
||||
}
|
||||
|
||||
// Convert to FileInfo
|
||||
var files []*FileInfo
|
||||
for _, record := range fileRecords {
|
||||
fileID := record["_id"].(bson.ObjectID).Hex()
|
||||
|
||||
var uploadDate time.Time
|
||||
if ud, ok := record["uploadDate"].(time.Time); ok {
|
||||
uploadDate = ud
|
||||
}
|
||||
|
||||
var metadata map[string]interface{}
|
||||
if md, ok := record["metadata"].(bson.M); ok {
|
||||
metadata = md
|
||||
}
|
||||
|
||||
fileInfo := &FileInfo{
|
||||
FileID: fileID,
|
||||
Filename: record["filename"].(string),
|
||||
ContentType: "application/octet-stream",
|
||||
Size: int64(record["length"].(int32)),
|
||||
UploadedAt: uploadDate,
|
||||
Metadata: metadata,
|
||||
}
|
||||
|
||||
// Extract metadata fields
|
||||
if metadata != nil {
|
||||
if ct, ok := metadata["content_type"].(string); ok {
|
||||
fileInfo.ContentType = ct
|
||||
}
|
||||
if ub, ok := metadata["uploaded_by"].(string); ok {
|
||||
fileInfo.UploadedBy = ub
|
||||
}
|
||||
if cat, ok := metadata["category"].(string); ok {
|
||||
fileInfo.Category = cat
|
||||
}
|
||||
if tags, ok := metadata["tags"].([]interface{}); ok {
|
||||
fileInfo.Tags = make([]string, len(tags))
|
||||
for i, tag := range tags {
|
||||
if tagStr, ok := tag.(string); ok {
|
||||
fileInfo.Tags[i] = tagStr
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
files = append(files, fileInfo)
|
||||
}
|
||||
|
||||
return files, total, nil
|
||||
}
|
||||
|
||||
// Exists checks if a file exists
|
||||
func (s *GridFSService) Exists(fileID string) (bool, error) {
|
||||
if fileID == "" {
|
||||
return false, ErrInvalidInput
|
||||
}
|
||||
|
||||
_, err := s.GetFileInfo(fileID)
|
||||
if err != nil {
|
||||
if IsNotFound(err) {
|
||||
return false, nil
|
||||
}
|
||||
return false, err
|
||||
}
|
||||
|
||||
return true, nil
|
||||
}
|
||||
Reference in New Issue
Block a user