Paperclip/pkg/internal/services/recycler.go

160 lines
4.5 KiB
Go
Raw Normal View History

2024-05-17 07:59:51 +00:00
package services
import (
"context"
"fmt"
"git.solsynth.dev/hydrogen/paperclip/pkg/internal/database"
"github.com/samber/lo"
2024-05-17 07:59:51 +00:00
"os"
"path/filepath"
"time"
2024-05-17 07:59:51 +00:00
2024-06-22 04:18:54 +00:00
"git.solsynth.dev/hydrogen/paperclip/pkg/internal/models"
2024-05-17 07:59:51 +00:00
jsoniter "github.com/json-iterator/go"
"github.com/minio/minio-go/v7"
"github.com/minio/minio-go/v7/pkg/credentials"
2024-07-28 16:53:40 +00:00
"github.com/rs/zerolog/log"
2024-05-17 07:59:51 +00:00
"github.com/spf13/viper"
)
2024-07-28 16:53:40 +00:00
var fileDeletionQueue = make(chan models.Attachment, 256)
func PublishDeleteFileTask(file models.Attachment) {
fileDeletionQueue <- file
}
func StartConsumeDeletionTask() {
for {
task := <-fileDeletionQueue
start := time.Now()
2024-07-28 16:53:40 +00:00
if err := DeleteFile(task); err != nil {
log.Error().Err(err).Any("task", task).Msg("A file deletion task failed...")
} else {
2024-07-29 05:22:57 +00:00
log.Info().Dur("elapsed", time.Since(start)).Uint("id", task.ID).Msg("A file deletion task was completed.")
2024-07-28 16:53:40 +00:00
}
}
}
2024-08-20 09:08:40 +00:00
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 {
2024-08-18 09:20:05 +00:00
lifecycle := time.Now().Add(-time.Duration(*pool.Config.Data().ExistLifecycle) * time.Second)
2024-08-18 07:58:07 +00:00
tx := database.C.
Where("pool_id = ?", pool.ID).
2024-08-18 09:20:05 +00:00
Where("created_at < ?", lifecycle).
2024-08-20 09:08:40 +00:00
Where("cleaned_at IS NULL").
2024-08-18 07:58:07 +00:00
Updates(&models.Attachment{CleanedAt: lo.ToPtr(time.Now())})
log.Info().
Str("pool", pool.Alias).
2024-08-18 07:58:07 +00:00
Int64("count", tx.RowsAffected).
Err(tx.Error).
Msg("Marking attachments as clean needed due to pool's lifecycle configuration...")
}
}
2024-08-20 09:08:40 +00:00
func RunMarkMultipartDeletionTask() {
2024-08-20 14:55:58 +00:00
lifecycle := time.Now().Add(-60 * time.Minute)
2024-08-20 09:08:40 +00:00
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...")
}
2024-08-18 07:58:07 +00:00
func RunScheduleDeletionTask() {
var attachments []models.Attachment
if err := database.C.Where("cleaned_at IS NOT NULL").Find(&attachments).Error; err != nil {
return
}
2024-08-20 09:08:40 +00:00
for _, attachment := range attachments {
if attachment.RefID != nil {
continue
}
2024-08-18 07:58:07 +00:00
if err := DeleteFile(attachment); err != nil {
log.Error().
Uint("id", attachment.ID).
Msg("An error occurred when deleting marked clean up attachments...")
}
}
2024-08-18 07:58:07 +00:00
database.C.Where("cleaned_at IS NOT NULL").Delete(&models.Attachment{})
}
2024-05-17 07:59:51 +00:00
func DeleteFile(meta models.Attachment) error {
2024-08-20 09:08:40 +00:00
if !meta.IsUploaded {
destMap := viper.GetStringMap("destinations.temporary")
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.%s", meta.Uuid, cid))
_ = os.Remove(path)
}
return nil
}
2024-07-28 13:03:56 +00:00
var destMap map[string]any
if meta.Destination == models.AttachmentDstTemporary {
destMap = viper.GetStringMap("destinations.temporary")
} else {
destMap = viper.GetStringMap("destinations.permanent")
2024-05-17 07:59:51 +00:00
}
2024-07-28 13:03:56 +00:00
var dest models.BaseDestination
rawDest, _ := jsoniter.Marshal(destMap)
_ = jsoniter.Unmarshal(rawDest, &dest)
2024-05-17 07:59:51 +00:00
2024-07-28 13:03:56 +00:00
switch dest.Type {
2024-05-17 07:59:51 +00:00
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:
2024-07-28 13:03:56 +00:00
return fmt.Errorf("invalid destination: unsupported protocol %s", dest.Type)
2024-05-17 07:59:51 +00:00
}
}
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
}