🚚 Move io related functions to fs internal package

pkg/internal/fs/README.md

@@ -0,0 +1,3 @@
+# File System
+
+The reason of why this package exists is because "cycle import was not allowed"

pkg/internal/fs/merger.go

@@ -0,0 +1,72 @@
+package fs
+
+import (
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+
+	"git.solsynth.dev/hypernet/paperclip/pkg/internal/database"
+	"git.solsynth.dev/hypernet/paperclip/pkg/internal/models"
+	"github.com/spf13/viper"
+)
+
+func MergeFileChunks(meta models.AttachmentFragment, arrange []string) (models.Attachment, error) {
+	attachment := meta.ToAttachment()
+
+	// Fetch destination from config
+	destMap := viper.GetStringMapString("destinations.0")
+
+	var dest models.LocalDestination
+	dest.Path = destMap["path"]
+
+	// Create the destination file
+	destPath := filepath.Join(dest.Path, meta.Uuid)
+	destFile, err := os.Create(destPath)
+	if err != nil {
+		return attachment, err
+	}
+	defer destFile.Close()
+
+	// 32KB buffer
+	buf := make([]byte, 32*1024)
+
+	// Merge the chunks into the destination file
+	for _, chunk := range arrange {
+		chunkPath := filepath.Join(dest.Path, fmt.Sprintf("%s.part%s", meta.Uuid, chunk))
+		chunkFile, err := os.Open(chunkPath)
+		if err != nil {
+			return attachment, err
+		}
+
+		defer chunkFile.Close() // Ensure the file is closed after reading
+
+		for {
+			n, err := chunkFile.Read(buf)
+			if err != nil && err != io.EOF {
+				return attachment, err
+			}
+			if n == 0 {
+				break
+			}
+
+			if _, err := destFile.Write(buf[:n]); err != nil {
+				return attachment, err
+			}
+		}
+	}
+
+	// Clean up: remove chunk files
+	go DeleteFragment(meta)
+	for _, chunk := range arrange {
+		chunkPath := filepath.Join(dest.Path, fmt.Sprintf("%s.part%s", meta.Uuid, chunk))
+		if err := os.Remove(chunkPath); err != nil {
+			return attachment, err
+		}
+	}
+
+	// Clean up: remove fragment record
+	database.C.Delete(&meta)
+
+	return attachment, nil
+}

pkg/internal/fs/recycler.go

@@ -0,0 +1,137 @@
+package fs
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"time"
+
+	"git.solsynth.dev/hypernet/paperclip/pkg/internal/database"
+	"github.com/samber/lo"
+
+	"git.solsynth.dev/hypernet/paperclip/pkg/internal/models"
+	jsoniter "github.com/json-iterator/go"
+	"github.com/minio/minio-go/v7"
+	"github.com/minio/minio-go/v7/pkg/credentials"
+	"github.com/rs/zerolog/log"
+	"github.com/spf13/viper"
+)
+
+func RunMarkLifecycleDeletionTask() {
+	var pools []models.AttachmentPool
+	if err := database.C.Find(&pools).Error; err != nil {
+		return
+	}
+
+	var pendingPools []models.AttachmentPool
+	for _, pool := range pools {
+		if pool.Config.Data().ExistLifecycle != nil {
+			pendingPools = append(pendingPools, pool)
+		}
+	}
+
+	for _, pool := range pendingPools {
+		lifecycle := time.Now().Add(-time.Duration(*pool.Config.Data().ExistLifecycle) * time.Second)
+		tx := database.C.
+			Where("pool_id = ?", pool.ID).
+			Where("created_at < ?", lifecycle).
+			Where("cleaned_at IS NULL").
+			Updates(&models.Attachment{CleanedAt: lo.ToPtr(time.Now())})
+		log.Info().
+			Str("pool", pool.Alias).
+			Int64("count", tx.RowsAffected).
+			Err(tx.Error).
+			Msg("Marking attachments as clean needed due to pool's lifecycle configuration...")
+	}
+}
+
+func RunMarkMultipartDeletionTask() {
+	lifecycle := time.Now().Add(-60 * time.Minute)
+	tx := database.C.
+		Where("created_at < ?", lifecycle).
+		Where("is_uploaded = ?", false).
+		Where("cleaned_at IS NULL").
+		Updates(&models.Attachment{CleanedAt: lo.ToPtr(time.Now())})
+	log.Info().
+		Int64("count", tx.RowsAffected).
+		Err(tx.Error).
+		Msg("Marking attachments as clean needed due to multipart lifecycle...")
+}
+
+func RunScheduleDeletionTask() {
+	var attachments []models.Attachment
+	if err := database.C.Where("cleaned_at IS NOT NULL").Find(&attachments).Error; err != nil {
+		return
+	}
+
+	for _, attachment := range attachments {
+		if attachment.RefID != nil {
+			continue
+		}
+		if err := DeleteFile(attachment); err != nil {
+			log.Error().
+				Uint("id", attachment.ID).
+				Msg("An error occurred when deleting marked clean up attachments...")
+		}
+	}
+
+	database.C.Where("cleaned_at IS NOT NULL").Delete(&models.Attachment{})
+}
+
+func DeleteFragment(meta models.AttachmentFragment) error {
+	destMap := viper.GetStringMap("destinations.0")
+	var dest models.LocalDestination
+	rawDest, _ := jsoniter.Marshal(destMap)
+	_ = jsoniter.Unmarshal(rawDest, &dest)
+
+	for cid := range meta.FileChunks {
+		path := filepath.Join(dest.Path, fmt.Sprintf("%s.part%s", meta.Uuid, cid))
+		_ = os.Remove(path)
+	}
+
+	return nil
+}
+
+func DeleteFile(meta models.Attachment) error {
+	destMap := viper.GetStringMap(fmt.Sprintf("destinations.%d", meta.Destination))
+
+	var dest models.BaseDestination
+	rawDest, _ := jsoniter.Marshal(destMap)
+	_ = jsoniter.Unmarshal(rawDest, &dest)
+
+	switch dest.Type {
+	case models.DestinationTypeLocal:
+		var destConfigured models.LocalDestination
+		_ = jsoniter.Unmarshal(rawDest, &destConfigured)
+		return DeleteFileFromLocal(destConfigured, meta)
+	case models.DestinationTypeS3:
+		var destConfigured models.S3Destination
+		_ = jsoniter.Unmarshal(rawDest, &destConfigured)
+		return DeleteFileFromS3(destConfigured, meta)
+	default:
+		return fmt.Errorf("invalid destination: unsupported protocol %s", dest.Type)
+	}
+}
+
+func DeleteFileFromLocal(config models.LocalDestination, meta models.Attachment) error {
+	fullpath := filepath.Join(config.Path, meta.Uuid)
+	return os.Remove(fullpath)
+}
+
+func DeleteFileFromS3(config models.S3Destination, meta models.Attachment) error {
+	client, err := minio.New(config.Endpoint, &minio.Options{
+		Creds:  credentials.NewStaticV4(config.SecretID, config.SecretKey, ""),
+		Secure: config.EnableSSL,
+	})
+	if err != nil {
+		return fmt.Errorf("unable to configure s3 client: %v", err)
+	}
+
+	err = client.RemoveObject(context.Background(), config.Bucket, filepath.Join(config.Path, meta.Uuid), minio.RemoveObjectOptions{})
+	if err != nil {
+		return fmt.Errorf("unable to upload file to s3: %v", err)
+	}
+
+	return nil
+}