feat: implement configurable background jobs and toggle flags
Introduce environment variables to globally and individually control background jobs: - `WARPBOX_JOBS_ENABLED` to toggle all background workers. - `WARPBOX_CLEANUP_ENABLED` to toggle the expired box cleanup job. - `WARPBOX_THUMBNAIL_ENABLED` to toggle the thumbnail generation job. Refactor background tasks into a dedicated `backend/libs/jobs` package, allowing jobs to be registered, scheduled, and conditionally run based on the new configuration flags. Additionally, update the default maximum upload size in `.env.example` to 16GB and document the new settings in the README.
This commit is contained in:
57
backend/libs/jobs/cleanup.go
Normal file
57
backend/libs/jobs/cleanup.go
Normal file
@@ -0,0 +1,57 @@
|
||||
package jobs
|
||||
|
||||
import (
|
||||
"log/slog"
|
||||
"time"
|
||||
|
||||
"warpbox.dev/backend/libs/config"
|
||||
"warpbox.dev/backend/libs/services"
|
||||
)
|
||||
|
||||
func newCleanupJob(cfg config.Config, logger *slog.Logger, uploadService *services.UploadService) job {
|
||||
return job{
|
||||
name: "cleanup",
|
||||
enabled: cfg.CleanupEnabled,
|
||||
interval: cfg.CleanupEvery,
|
||||
run: func() {
|
||||
cleaned, err := cleanupUnavailableBoxes(uploadService, logger)
|
||||
if err != nil {
|
||||
logger.Warn("cleanup job failed", "source", "housekeeping", "severity", "warn", "code", 4202, "error", err.Error())
|
||||
return
|
||||
}
|
||||
if cleaned > 0 {
|
||||
logger.Info("cleanup job complete", "source", "housekeeping", "severity", "user_activity", "code", 2202, "cleaned", cleaned)
|
||||
}
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func cleanupUnavailableBoxes(uploadService *services.UploadService, logger *slog.Logger) (int, error) {
|
||||
boxes, err := uploadService.ListBoxes(0)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
now := time.Now().UTC()
|
||||
cleaned := 0
|
||||
for _, box := range boxes {
|
||||
if !shouldDeleteBox(box, now) {
|
||||
continue
|
||||
}
|
||||
if err := uploadService.DeleteBoxWithSource(box.ID, "housekeeping"); err != nil {
|
||||
return cleaned, err
|
||||
}
|
||||
cleaned++
|
||||
}
|
||||
if cleaned > 0 {
|
||||
logger.Info("unavailable boxes cleaned", "source", "housekeeping", "severity", "user_activity", "code", 2201, "cleaned", cleaned)
|
||||
}
|
||||
return cleaned, nil
|
||||
}
|
||||
|
||||
func shouldDeleteBox(box services.Box, now time.Time) bool {
|
||||
if !box.ExpiresAt.After(now) {
|
||||
return true
|
||||
}
|
||||
return box.MaxDownloads > 0 && box.DownloadCount >= box.MaxDownloads
|
||||
}
|
||||
50
backend/libs/jobs/cleanup_test.go
Normal file
50
backend/libs/jobs/cleanup_test.go
Normal file
@@ -0,0 +1,50 @@
|
||||
package jobs
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"warpbox.dev/backend/libs/services"
|
||||
)
|
||||
|
||||
func TestShouldDeleteBox(t *testing.T) {
|
||||
now := time.Date(2026, 5, 29, 12, 0, 0, 0, time.UTC)
|
||||
|
||||
tests := map[string]struct {
|
||||
box services.Box
|
||||
want bool
|
||||
}{
|
||||
"expired": {
|
||||
box: services.Box{ExpiresAt: now.Add(-time.Second)},
|
||||
want: true,
|
||||
},
|
||||
"expires now": {
|
||||
box: services.Box{ExpiresAt: now},
|
||||
want: true,
|
||||
},
|
||||
"download limit reached": {
|
||||
box: services.Box{ExpiresAt: now.Add(time.Hour), MaxDownloads: 3, DownloadCount: 3},
|
||||
want: true,
|
||||
},
|
||||
"download limit exceeded": {
|
||||
box: services.Box{ExpiresAt: now.Add(time.Hour), MaxDownloads: 3, DownloadCount: 4},
|
||||
want: true,
|
||||
},
|
||||
"active unlimited": {
|
||||
box: services.Box{ExpiresAt: now.Add(time.Hour)},
|
||||
want: false,
|
||||
},
|
||||
"active under limit": {
|
||||
box: services.Box{ExpiresAt: now.Add(time.Hour), MaxDownloads: 3, DownloadCount: 2},
|
||||
want: false,
|
||||
},
|
||||
}
|
||||
|
||||
for name, tt := range tests {
|
||||
t.Run(name, func(t *testing.T) {
|
||||
if got := shouldDeleteBox(tt.box, now); got != tt.want {
|
||||
t.Fatalf("shouldDeleteBox() = %v, want %v", got, tt.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
75
backend/libs/jobs/jobs.go
Normal file
75
backend/libs/jobs/jobs.go
Normal file
@@ -0,0 +1,75 @@
|
||||
package jobs
|
||||
|
||||
import (
|
||||
"log/slog"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"warpbox.dev/backend/libs/config"
|
||||
"warpbox.dev/backend/libs/services"
|
||||
)
|
||||
|
||||
type job struct {
|
||||
name string
|
||||
enabled bool
|
||||
interval time.Duration
|
||||
run func()
|
||||
}
|
||||
|
||||
func StartAll(cfg config.Config, logger *slog.Logger, uploadService *services.UploadService) func() {
|
||||
if !cfg.JobsEnabled {
|
||||
logger.Info("background jobs disabled", "source", "jobs", "severity", "dev")
|
||||
return func() {}
|
||||
}
|
||||
|
||||
stops := []func(){
|
||||
start(newCleanupJob(cfg, logger, uploadService), logger),
|
||||
start(newThumbnailsJob(cfg, logger, uploadService), logger),
|
||||
}
|
||||
|
||||
var once sync.Once
|
||||
return func() {
|
||||
once.Do(func() {
|
||||
for _, stop := range stops {
|
||||
stop()
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func start(j job, logger *slog.Logger) func() {
|
||||
if !j.enabled {
|
||||
logger.Info("background job disabled", "source", "jobs", "severity", "dev", "job", j.name)
|
||||
return func() {}
|
||||
}
|
||||
if j.interval <= 0 {
|
||||
logger.Info("background job disabled by interval", "source", "jobs", "severity", "dev", "job", j.name, "interval", j.interval.String())
|
||||
return func() {}
|
||||
}
|
||||
|
||||
stop := make(chan struct{})
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
defer close(done)
|
||||
j.run()
|
||||
|
||||
ticker := time.NewTicker(j.interval)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
j.run()
|
||||
case <-stop:
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
var once sync.Once
|
||||
return func() {
|
||||
once.Do(func() {
|
||||
close(stop)
|
||||
<-done
|
||||
})
|
||||
}
|
||||
}
|
||||
161
backend/libs/jobs/thumbnails.go
Normal file
161
backend/libs/jobs/thumbnails.go
Normal file
@@ -0,0 +1,161 @@
|
||||
package jobs
|
||||
|
||||
import (
|
||||
"image"
|
||||
_ "image/gif"
|
||||
"image/jpeg"
|
||||
_ "image/jpeg"
|
||||
_ "image/png"
|
||||
"log/slog"
|
||||
"os"
|
||||
"os/exec"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
_ "golang.org/x/image/webp"
|
||||
"warpbox.dev/backend/libs/config"
|
||||
"warpbox.dev/backend/libs/services"
|
||||
)
|
||||
|
||||
type thumbnailJobResult struct {
|
||||
Scanned int
|
||||
Generated int
|
||||
Failed int
|
||||
}
|
||||
|
||||
func newThumbnailsJob(cfg config.Config, logger *slog.Logger, uploadService *services.UploadService) job {
|
||||
return job{
|
||||
name: "thumbnail",
|
||||
enabled: cfg.ThumbnailEnabled,
|
||||
interval: cfg.ThumbnailEvery,
|
||||
run: func() {
|
||||
result, err := generateMissingThumbnails(uploadService, logger)
|
||||
if err != nil {
|
||||
logger.Warn("thumbnail job failed", "source", "thumbnail", "severity", "warn", "code", 4203, "error", err.Error())
|
||||
return
|
||||
}
|
||||
if result.Generated > 0 || result.Failed > 0 {
|
||||
logger.Info("thumbnail job complete", "source", "thumbnail", "severity", "user_activity", "code", 2204, "generated", result.Generated, "failed", result.Failed)
|
||||
}
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func generateMissingThumbnails(uploadService *services.UploadService, logger *slog.Logger) (thumbnailJobResult, error) {
|
||||
boxes, err := uploadService.ListBoxes(0)
|
||||
if err != nil {
|
||||
return thumbnailJobResult{}, err
|
||||
}
|
||||
|
||||
var result thumbnailJobResult
|
||||
now := time.Now().UTC()
|
||||
for _, box := range boxes {
|
||||
if !box.ExpiresAt.After(now) {
|
||||
continue
|
||||
}
|
||||
|
||||
changed := false
|
||||
for i := range box.Files {
|
||||
file := &box.Files[i]
|
||||
if file.Thumbnail != "" || !needsThumbnail(*file) {
|
||||
continue
|
||||
}
|
||||
result.Scanned++
|
||||
|
||||
thumbnail, err := generateThumbnail(uploadService, box, *file)
|
||||
if err != nil {
|
||||
logger.Warn("thumbnail generation failed", "source", "thumbnail", "severity", "warn", "code", 4101, "file_id", file.ID, "error", err.Error())
|
||||
result.Failed++
|
||||
continue
|
||||
}
|
||||
if thumbnail == "" {
|
||||
result.Failed++
|
||||
continue
|
||||
}
|
||||
|
||||
file.Thumbnail = thumbnail
|
||||
changed = true
|
||||
result.Generated++
|
||||
}
|
||||
|
||||
if changed {
|
||||
if err := uploadService.SaveBox(box); err != nil {
|
||||
return result, err
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func needsThumbnail(file services.File) bool {
|
||||
return file.PreviewKind == "image" || file.PreviewKind == "video"
|
||||
}
|
||||
|
||||
func generateThumbnail(uploadService *services.UploadService, box services.Box, file services.File) (string, error) {
|
||||
thumbnailName := "@thumb@" + file.ID + ".jpg"
|
||||
thumbnailPath := uploadService.ThumbnailPath(box, services.File{Thumbnail: thumbnailName})
|
||||
sourcePath := uploadService.FilePath(box, file)
|
||||
|
||||
switch {
|
||||
case strings.HasPrefix(file.ContentType, "image/"):
|
||||
return thumbnailName, createImageThumbnail(sourcePath, thumbnailPath)
|
||||
case strings.HasPrefix(file.ContentType, "video/"):
|
||||
return thumbnailName, createVideoThumbnail(sourcePath, thumbnailPath)
|
||||
default:
|
||||
return "", nil
|
||||
}
|
||||
}
|
||||
|
||||
func createImageThumbnail(sourcePath, targetPath string) error {
|
||||
source, err := os.Open(sourcePath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer source.Close()
|
||||
|
||||
img, _, err := image.Decode(source)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
thumb := resizeNearest(img, 360, 240)
|
||||
target, err := os.OpenFile(targetPath, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o644)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer target.Close()
|
||||
|
||||
return jpeg.Encode(target, thumb, &jpeg.Options{Quality: 82})
|
||||
}
|
||||
|
||||
func createVideoThumbnail(sourcePath, targetPath string) error {
|
||||
return exec.Command("ffmpeg", "-y", "-loglevel", "error", "-ss", "00:00:01", "-i", sourcePath, "-frames:v", "1", "-vf", "scale=360:-1", targetPath).Run()
|
||||
}
|
||||
|
||||
func resizeNearest(src image.Image, maxWidth, maxHeight int) *image.RGBA {
|
||||
bounds := src.Bounds()
|
||||
width := bounds.Dx()
|
||||
height := bounds.Dy()
|
||||
if width <= 0 || height <= 0 {
|
||||
return image.NewRGBA(image.Rect(0, 0, 1, 1))
|
||||
}
|
||||
|
||||
scale := min(float64(maxWidth)/float64(width), float64(maxHeight)/float64(height))
|
||||
if scale > 1 {
|
||||
scale = 1
|
||||
}
|
||||
targetWidth := max(1, int(float64(width)*scale))
|
||||
targetHeight := max(1, int(float64(height)*scale))
|
||||
dst := image.NewRGBA(image.Rect(0, 0, targetWidth, targetHeight))
|
||||
|
||||
for y := 0; y < targetHeight; y++ {
|
||||
for x := 0; x < targetWidth; x++ {
|
||||
srcX := bounds.Min.X + int(float64(x)/scale)
|
||||
srcY := bounds.Min.Y + int(float64(y)/scale)
|
||||
dst.Set(x, y, src.At(srcX, srcY))
|
||||
}
|
||||
}
|
||||
|
||||
return dst
|
||||
}
|
||||
Reference in New Issue
Block a user