80 lines
2.0 KiB
Go
80 lines
2.0 KiB
Go
|
package jobs
|
||
|
|
|
||
|
|
import (
|
||
|
|
"context"
|
||
|
|
"time"
|
||
|
|
|
||
|
|
"git.apinb.com/bsm-sdk/core/logger"
|
||
|
|
"git.apinb.com/ops/files/internal/config"
|
||
|
|
"git.apinb.com/ops/files/internal/impl"
|
||
|
|
"git.apinb.com/ops/files/internal/lifecycle"
|
||
|
|
"git.apinb.com/ops/files/internal/models"
|
||
|
|
)
|
||
|
|
|
||
|
|
func StartPendingUploadCleanup() {
|
||
|
|
go func() {
|
||
|
|
cleanupPendingUploads()
|
||
|
|
|
||
|
|
ticker := time.NewTicker(time.Duration(config.Spec.Cleanup.IntervalSeconds) * time.Second)
|
||
|
|
defer ticker.Stop()
|
||
|
|
for range ticker.C {
|
||
|
|
cleanupPendingUploads()
|
||
|
|
}
|
||
|
|
}()
|
||
|
|
}
|
||
|
|
|
||
|
|
func cleanupPendingUploads() {
|
||
|
|
var fileObjects []models.FileObject
|
||
|
|
now := time.Now()
|
||
|
|
result := impl.DBService.
|
||
|
|
Where(
|
||
|
|
"(status = ? AND expires_at <= ?) OR (status = ? AND (delete_lease_until IS NULL OR delete_lease_until <= ?))",
|
||
|
|
models.FileStatusPending,
|
||
|
|
now,
|
||
|
|
models.FileStatusDeleting,
|
||
|
|
now,
|
||
|
|
).
|
||
|
|
Order("id ASC").
|
||
|
|
Limit(100).
|
||
|
|
Find(&fileObjects)
|
||
|
|
if result.Error != nil {
|
||
|
|
logger.Error("stage=scan")
|
||
|
|
return
|
||
|
|
}
|
||
|
|
|
||
|
|
for _, fileObject := range fileObjects {
|
||
|
|
var lease lifecycle.DeletionLease
|
||
|
|
var claimed bool
|
||
|
|
var err error
|
||
|
|
now = time.Now()
|
||
|
|
if fileObject.Status == models.FileStatusPending {
|
||
|
|
lease, claimed, err = lifecycle.ClaimExpiredPendingDeletion(impl.DBService, fileObject.ID, now)
|
||
|
|
} else {
|
||
|
|
lease, claimed, err = lifecycle.ClaimDeletionRetry(impl.DBService, fileObject.ID, now)
|
||
|
|
}
|
||
|
|
if err != nil {
|
||
|
|
logger.Errorf("identity=%s stage=claim", fileObject.Identity)
|
||
|
|
continue
|
||
|
|
}
|
||
|
|
if !claimed {
|
||
|
|
continue
|
||
|
|
}
|
||
|
|
|
||
|
|
cleanupClaimedFile(fileObject, lease)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
func cleanupClaimedFile(fileObject models.FileObject, lease lifecycle.DeletionLease) {
|
||
|
|
operationCtx, cancel := context.WithTimeout(context.Background(), lifecycle.DeleteOperationTimeout)
|
||
|
|
defer cancel()
|
||
|
|
|
||
|
|
if err := impl.StorageService.Delete(operationCtx, fileObject.ObjectKey); err != nil {
|
||
|
|
logger.Errorf("identity=%s stage=oss_delete", fileObject.Identity)
|
||
|
|
return
|
||
|
|
}
|
||
|
|
|
||
|
|
if err := lifecycle.FinalizeDeletion(impl.DBService.WithContext(operationCtx), fileObject.ID, lease.Token); err != nil {
|
||
|
|
logger.Errorf("identity=%s stage=finalize", fileObject.Identity)
|
||
|
|
}
|
||
|
|
}
|