[] Add artifact cleanup Asynq worker for expired artifact deletion and docker-compose MinIO init bucket job — with unit tests

backend/cmd/server/main.go

@@ -631,6 +631,17 @@ func registerHandlers(e *echo.Echo, cfg config.Config, db *sql.DB) (*sshinfra.SS
 		mux := asynq.NewServeMux()
 		mcpWorker := worker.NewMCPVerificationWorker(mcpVerificationRepo, cfg.WebBaseURL)
 		mux.HandleFunc(queue.TypeMCPVerification, mcpWorker.HandleMCPVerification)
+		artifactCleanupWorker := worker.NewArtifactCleanupWorker(artifactRepo, artifactStorage, minioBucket)
+		mux.HandleFunc(queue.TypeArtifactCleanup, artifactCleanupWorker.HandleCleanup)
+		scheduler := asynq.NewScheduler(asynq.RedisClientOpt{Addr: cfg.RedisAddr}, nil)
+		if _, err := scheduler.Register("@hourly", asynq.NewTask(queue.TypeArtifactCleanup, nil)); err != nil {
+			return nil, fmt.Errorf("register artifact cleanup scheduler: %w", err)
+		}
+		go func() {
+			if err := scheduler.Run(); err != nil {
+				middleware.Log().Info("asynq scheduler stopped", "error", err)
+			}
+		}()
 		prMergeableWorker := queue.NewPRMergeableWorker(prRepo, repoRepo, gitSvc)
 		mux.HandleFunc(queue.TypePRMergeableCheck, prMergeableWorker.HandlePRMergeableCheck)
 		go func() {

backend/internal/infrastructure/queue/asynq_client.go

@@ -11,6 +11,7 @@ import (
 const (
 	TypeWebhookDeliver  = "webhook:deliver"
 	TypeMCPVerification = "mcp:verification"
+	TypeArtifactCleanup = "artifact:cleanup"
 	TypeDispatchJob     = "actions:dispatch_job"
 	TypeCancelJob       = "actions:cancel_job"
 )
@@ -71,6 +72,11 @@ func EnqueueWebhookDelivery(ctx context.Context, client *asynq.Client, payload W
 	return client.EnqueueContext(ctx, task, asynq.MaxRetry(5))
 }
 
+func EnqueueArtifactCleanup(ctx context.Context, client *asynq.Client) (*asynq.TaskInfo, error) {
+	task := asynq.NewTask(TypeArtifactCleanup, nil)
+	return client.EnqueueContext(ctx, task, asynq.Queue("default"))
+}
+
 func EnqueueDispatchJob(ctx context.Context, client *asynq.Client, payload DispatchJobPayload) (*asynq.TaskInfo, error) {
 	data, err := json.Marshal(payload)
 	if err != nil {

backend/internal/worker/artifact_cleanup_worker.go

@@ -0,0 +1,57 @@
+package worker
+
+import (
+	"context"
+	"log/slog"
+
+	"github.com/hibiken/asynq"
+
+	domainrepo "github.com/open-git/backend/internal/domain/repository"
+	artifactusecase "github.com/open-git/backend/internal/usecase/artifact"
+)
+
+type ArtifactCleanupWorker struct {
+	artifactRepo domainrepo.IArtifactRepository
+	storage      artifactusecase.ArtifactStorage
+	bucket       string
+}
+
+func NewArtifactCleanupWorker(
+	repo domainrepo.IArtifactRepository,
+	storage artifactusecase.ArtifactStorage,
+	bucket string,
+) *ArtifactCleanupWorker {
+	return &ArtifactCleanupWorker{
+		artifactRepo: repo,
+		storage:      storage,
+		bucket:       bucket,
+	}
+}
+
+func (w *ArtifactCleanupWorker) HandleCleanup(ctx context.Context, task *asynq.Task) error {
+	const batchSize = 100
+	for {
+		artifacts, err := w.artifactRepo.ListExpired(ctx, batchSize)
+		if err != nil {
+			return err
+		}
+		if len(artifacts) == 0 {
+			break
+		}
+
+		for _, artifact := range artifacts {
+			if err := w.storage.DeleteObject(ctx, w.bucket, artifact.StorageKey); err != nil {
+				slog.Error("delete artifact object", "artifact_id", artifact.ID, "error", err)
+				continue
+			}
+			if err := w.artifactRepo.SoftDelete(ctx, artifact.ID, artifact.OrganizationID); err != nil {
+				slog.Error("soft delete artifact", "artifact_id", artifact.ID, "error", err)
+			}
+		}
+
+		if len(artifacts) < batchSize {
+			break
+		}
+	}
+	return nil
+}

backend/internal/worker/artifact_cleanup_worker_test.go

@@ -0,0 +1,164 @@
+package worker
+
+import (
+	"context"
+	"database/sql"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/hibiken/asynq"
+	"github.com/jmoiron/sqlx"
+	_ "github.com/mattn/go-sqlite3"
+
+	"github.com/open-git/backend/internal/domain/entity"
+	infrarepo "github.com/open-git/backend/internal/infrastructure/repository"
+)
+
+const artifactCleanupTestSchema = `
+CREATE TABLE organizations (
+    id TEXT PRIMARY KEY,
+    login TEXT NOT NULL,
+    name TEXT NOT NULL,
+    plan_tier TEXT NOT NULL DEFAULT 'free',
+    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+CREATE TABLE repositories (
+    id TEXT PRIMARY KEY,
+    organization_id TEXT NOT NULL,
+    name TEXT NOT NULL
+);
+CREATE TABLE workflow_runs (
+    id TEXT PRIMARY KEY,
+    organization_id TEXT NOT NULL,
+    repository_id TEXT NOT NULL,
+    workflow TEXT NOT NULL,
+    status TEXT NOT NULL,
+    conclusion TEXT,
+    started_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    completed_at TIMESTAMP
+);
+CREATE TABLE artifacts (
+    id TEXT PRIMARY KEY,
+    organization_id TEXT NOT NULL,
+    repository_id TEXT NOT NULL,
+    workflow_run_id TEXT NOT NULL,
+    name TEXT NOT NULL,
+    storage_key TEXT NOT NULL,
+    size_in_bytes INTEGER NOT NULL DEFAULT 0,
+    status TEXT NOT NULL CHECK(status IN ('pending','uploading','completed','failed','expired')),
+    retention_days INTEGER NOT NULL,
+    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    expires_at TIMESTAMP NOT NULL,
+    deleted_at TIMESTAMP
+);
+CREATE UNIQUE INDEX idx_artifacts_run_name ON artifacts(workflow_run_id, name) WHERE deleted_at IS NULL;
+`
+
+func newArtifactCleanupTestDB(t *testing.T) *sqlx.DB {
+	t.Helper()
+	db, err := sql.Open("sqlite3", ":memory:")
+	if err != nil {
+		t.Fatalf("open sqlite: %v", err)
+	}
+	if _, err := db.Exec(artifactCleanupTestSchema); err != nil {
+		_ = db.Close()
+		t.Fatalf("create schema: %v", err)
+	}
+	t.Cleanup(func() { _ = db.Close() })
+	return sqlx.NewDb(db, "sqlite3")
+}
+
+type mockArtifactCleanupStorage struct {
+	deletedKeys []string
+}
+
+func (m *mockArtifactCleanupStorage) PresignedPutURL(context.Context, string, string, time.Duration) (string, error) {
+	return "", nil
+}
+
+func (m *mockArtifactCleanupStorage) PresignedGetURL(context.Context, string, string, time.Duration) (string, error) {
+	return "", nil
+}
+
+func (m *mockArtifactCleanupStorage) DeleteObject(_ context.Context, _, key string) error {
+	m.deletedKeys = append(m.deletedKeys, key)
+	return nil
+}
+
+func TestArtifactCleanupWorker_HandleCleanup(t *testing.T) {
+	db := newArtifactCleanupTestDB(t)
+	ctx := context.Background()
+
+	orgID := uuid.MustParse("11111111-1111-1111-1111-111111111111")
+	repoID := uuid.MustParse("33333333-3333-3333-3333-333333333333")
+	runID := uuid.MustParse("44444444-4444-4444-4444-444444444444")
+
+	mustExec(t, db.DB, `INSERT INTO organizations (id, login, name, plan_tier) VALUES (?, ?, ?, ?)`,
+		orgID.String(), "acme", "Acme", "free")
+	mustExec(t, db.DB, `INSERT INTO repositories (id, organization_id, name) VALUES (?, ?, ?)`,
+		repoID.String(), orgID.String(), "widgets")
+	mustExec(t, db.DB, `INSERT INTO workflow_runs (id, organization_id, repository_id, workflow, status) VALUES (?, ?, ?, ?, ?)`,
+		runID.String(), orgID.String(), repoID.String(), "ci.yml", "completed")
+
+	now := time.Now().UTC()
+	expiredKey := "org/acme/repo/widgets/runs/1/expired.zip"
+	futureKey := "org/acme/repo/widgets/runs/1/future.zip"
+
+	artifactRepo := infrarepo.NewArtifactRepository(db)
+	expiredArtifact := &entity.Artifact{
+		OrganizationID: orgID,
+		RunID:          runID,
+		Name:           "expired-artifact",
+		StorageKey:     expiredKey,
+		ExpiresAt:      now.Add(-1 * time.Hour),
+	}
+	futureArtifact := &entity.Artifact{
+		OrganizationID: orgID,
+		RunID:          runID,
+		Name:           "future-artifact",
+		StorageKey:     futureKey,
+		ExpiresAt:      now.Add(24 * time.Hour),
+	}
+	if err := artifactRepo.Create(ctx, expiredArtifact); err != nil {
+		t.Fatalf("Create expired artifact: %v", err)
+	}
+	if err := artifactRepo.Create(ctx, futureArtifact); err != nil {
+		t.Fatalf("Create future artifact: %v", err)
+	}
+	if err := artifactRepo.UpdateStatus(ctx, expiredArtifact.ID, entity.ArtifactStatusCompleted, 100); err != nil {
+		t.Fatalf("UpdateStatus expired artifact: %v", err)
+	}
+	if err := artifactRepo.UpdateStatus(ctx, futureArtifact.ID, entity.ArtifactStatusCompleted, 200); err != nil {
+		t.Fatalf("UpdateStatus future artifact: %v", err)
+	}
+
+	storage := &mockArtifactCleanupStorage{}
+	worker := NewArtifactCleanupWorker(artifactRepo, storage, "artifacts")
+	task := asynq.NewTask("artifact:cleanup", nil)
+	if err := worker.HandleCleanup(ctx, task); err != nil {
+		t.Fatalf("HandleCleanup returned error: %v", err)
+	}
+
+	var expiredDeletedAt sql.NullTime
+	if err := db.QueryRowContext(ctx, `SELECT deleted_at FROM artifacts WHERE id = ?`, expiredArtifact.ID.String()).
+		Scan(&expiredDeletedAt); err != nil {
+		t.Fatalf("query expired artifact deleted_at: %v", err)
+	}
+	if !expiredDeletedAt.Valid {
+		t.Fatal("expected expired artifact deleted_at to be set")
+	}
+
+	var futureDeletedAt sql.NullTime
+	if err := db.QueryRowContext(ctx, `SELECT deleted_at FROM artifacts WHERE id = ?`, futureArtifact.ID.String()).
+		Scan(&futureDeletedAt); err != nil {
+		t.Fatalf("query future artifact deleted_at: %v", err)
+	}
+	if futureDeletedAt.Valid {
+		t.Fatal("expected non-expired artifact deleted_at to remain NULL")
+	}
+
+	if len(storage.deletedKeys) != 1 || storage.deletedKeys[0] != expiredKey {
+		t.Fatalf("DeleteObject calls = %v, want [%q]", storage.deletedKeys, expiredKey)
+	}
+}

docker-compose.yml

@@ -92,6 +92,18 @@ services:
       retries: 3
     restart: unless-stopped
 
+  createbuckets:
+    image: minio/mc
+    depends_on:
+      minio:
+        condition: service_healthy
+    entrypoint: >
+      /bin/sh -c "
+      mc alias set myminio http://minio:9000 $${MINIO_ROOT_USER:-minioadmin} $${MINIO_ROOT_PASSWORD:-minioadmin};
+      mc mb --ignore-existing myminio/artifacts;
+      exit 0;
+      "
+
   traefik:
     image: traefik:v3
     command: