diff --git a/backend/cmd/server/main.go b/backend/cmd/server/main.go index 3d95f132..8e8c7e26 100644 --- a/backend/cmd/server/main.go +++ b/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() { diff --git a/backend/internal/infrastructure/queue/asynq_client.go b/backend/internal/infrastructure/queue/asynq_client.go index bdee0294..6aa7dcc3 100644 --- a/backend/internal/infrastructure/queue/asynq_client.go +++ b/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 { diff --git a/backend/internal/worker/artifact_cleanup_worker.go b/backend/internal/worker/artifact_cleanup_worker.go new file mode 100644 index 00000000..fea58fd1 --- /dev/null +++ b/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 +} diff --git a/backend/internal/worker/artifact_cleanup_worker_test.go b/backend/internal/worker/artifact_cleanup_worker_test.go new file mode 100644 index 00000000..2f755cce --- /dev/null +++ b/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) + } +} diff --git a/docker-compose.yml b/docker-compose.yml index 6b05dc56..54e49695 100644 --- a/docker-compose.yml +++ b/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: