limit functions reaper to own orgs

Author: disintegrator

server/internal/background/activities.go

@@ -147,6 +147,6 @@ func (a *Activities) ValidateDeployment(ctx context.Context, projectID uuid.UUID
 	return a.validateDeployment.Do(ctx, projectID, deploymentID)
 }
 
-func (a *Activities) ReapFlyApps(ctx context.Context) (*activities.ReapFlyAppsResult, error) {
-	return a.reapFlyApps.Do(ctx)
+func (a *Activities) ReapFlyApps(ctx context.Context, req activities.ReapFlyAppsRequest) (*activities.ReapFlyAppsResult, error) {
+	return a.reapFlyApps.Do(ctx, req)
 }

server/internal/background/activities/reap_functions.go

@@ -4,16 +4,24 @@ import (
 	"context"
 	"log/slog"
 
+	"github.com/google/uuid"
 	"github.com/jackc/pgx/v5/pgtype"
 	"github.com/jackc/pgx/v5/pgxpool"
 	"go.opentelemetry.io/otel/metric"
+	"go.temporal.io/sdk/temporal"
 
 	"github.com/speakeasy-api/gram/server/internal/attr"
 	"github.com/speakeasy-api/gram/server/internal/functions"
 	funcrepo "github.com/speakeasy-api/gram/server/internal/functions/repo"
 	"github.com/speakeasy-api/gram/server/internal/oops"
 )
 
+type ReapFlyAppsRequest struct {
+	Scope FunctionsReaperScope
+
+	ProjectID uuid.NullUUID
+}
+
 type ReapFlyAppsResult struct {
 	Reaped int
 	Errors int
@@ -43,9 +51,22 @@ func NewReapFlyApps(
 	}
 }
 
-func (r *ReapFlyApps) Do(ctx context.Context) (*ReapFlyAppsResult, error) {
+func (r *ReapFlyApps) Do(ctx context.Context, req ReapFlyAppsRequest) (*ReapFlyAppsResult, error) {
 	logger := r.logger
 
+	switch {
+	case req.Scope == FunctionsReaperScopeProject && req.ProjectID.UUID == uuid.Nil:
+		return nil, temporal.NewApplicationErrorWithOptions("project ID must be set for project-scoped reaper", "reaper_error", temporal.ApplicationErrorOptions{
+			NonRetryable: true,
+			Cause:        nil,
+		})
+	case req.Scope == FunctionsReaperScopeGlobal && req.ProjectID.UUID != uuid.Nil:
+		return nil, temporal.NewApplicationErrorWithOptions("project ID must not be set for global reaper", "reaper_error", temporal.ApplicationErrorOptions{
+			NonRetryable: true,
+			Cause:        nil,
+		})
+	}
+
 	repo := funcrepo.New(r.db)
 
 	// Get all apps that should be reaped (keeping only the most recent N per project)

server/internal/background/activities/types.go

@@ -0,0 +1,8 @@
+package activities
+
+type FunctionsReaperScope string
+
+const (
+	FunctionsReaperScopeGlobal  FunctionsReaperScope = "global"
+	FunctionsReaperScopeProject FunctionsReaperScope = "project"
+)

server/internal/background/deployments.go

@@ -184,18 +184,23 @@ func ProcessDeploymentWorkflow(ctx workflow.Context, params ProcessDeploymentWor
 		)
 	}
 
-	// Trigger functions reaper workflow after successful deployment
+	// Trigger project-scoped functions reaper workflow after successful deployment
 	// This runs asynchronously and doesn't block the deployment workflow
 	if finalStatus == "completed" {
-		_ = workflow.ExecuteChildWorkflow(
-			workflow.WithChildOptions(ctx, workflow.ChildWorkflowOptions{
-				WorkflowID:        "v1:functions-reaper-triggered",
-				ParentClosePolicy: enums.PARENT_CLOSE_POLICY_ABANDON,
-			}),
-			FunctionsReaperWorkflow,
-			FunctionsReaperWorkflowParams{},
-		).GetChildWorkflowExecution()
-		// We don't wait for the reaper to complete - it runs independently
+		logger.Info(
+			"starting project-scoped function reaper",
+			string(attr.ProjectIDKey), params.ProjectID,
+			string(attr.DeploymentIDKey), params.DeploymentID,
+		)
+		_, err = ExecuteProjectFunctionsReaperChildWorkflow(ctx, params.ProjectID)
+		if err != nil {
+			logger.Error(
+				"failed to start project-scoped functions reaper workflow",
+				"error", err.Error(),
+				string(attr.ProjectIDKey), params.ProjectID,
+				string(attr.DeploymentIDKey), params.DeploymentID,
+			)
+		}
 	}
 
 	return &ProcessDeploymentWorkflowResult{

server/internal/background/functions_reaper.go

@@ -10,18 +10,34 @@ import (
 	"go.temporal.io/sdk/temporal"
 	"go.temporal.io/sdk/workflow"
 
+	"github.com/google/uuid"
 	"github.com/speakeasy-api/gram/server/internal/background/activities"
 )
 
 type FunctionsReaperWorkflowParams struct {
-	// No parameters needed - reaper runs globally across all projects
+	Scope activities.FunctionsReaperScope
+
+	ProjectID uuid.NullUUID
 }
 
 type FunctionsReaperWorkflowResult struct {
 	AppsReaped int
 	Errors     int
 }
 
+func ExecuteProjectFunctionsReaperChildWorkflow(ctx workflow.Context, projectID uuid.UUID) (workflow.ChildWorkflowFuture, error) {
+	return workflow.ExecuteChildWorkflow(ctx, client.StartWorkflowOptions{
+		ID:                       "v1:functions-reaper:" + projectID.String(),
+		TaskQueue:                string(TaskQueueMain),
+		WorkflowIDConflictPolicy: enums.WORKFLOW_ID_CONFLICT_POLICY_USE_EXISTING,
+		WorkflowIDReusePolicy:    enums.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE,
+		WorkflowRunTimeout:       time.Minute * 5,
+	}, FunctionsReaperWorkflow, FunctionsReaperWorkflowParams{
+		Scope:     activities.FunctionsReaperScopeProject,
+		ProjectID: uuid.NullUUID{UUID: projectID, Valid: projectID != uuid.Nil},
+	}), nil
+}
+
 func ExecuteFunctionsReaperWorkflow(ctx context.Context, temporalClient client.Client, params FunctionsReaperWorkflowParams) (client.WorkflowRun, error) {
 	// Use a fixed workflow ID so that only one reaper workflow can run at a time
 	return temporalClient.ExecuteWorkflow(ctx, client.StartWorkflowOptions{
@@ -54,6 +70,10 @@ func FunctionsReaperWorkflow(ctx workflow.Context, params FunctionsReaperWorkflo
 	err := workflow.ExecuteActivity(
 		ctx,
 		a.ReapFlyApps,
+		activities.ReapFlyAppsRequest{
+			Scope:     params.Scope,
+			ProjectID: params.ProjectID,
+		},
 	).Get(ctx, &result)
 	if err != nil {
 		return nil, fmt.Errorf("failed to reap functions: %w", err)

server/internal/functions/deploy_fly.go

@@ -536,13 +536,51 @@ func (f *FlyRunner) reap(ctx context.Context, logger *slog.Logger, appsRepo *rep
 		return fmt.Errorf("get existing app name: %w", err)
 	}
 
+	enabled, err := appsRepo.IsReapingEnabledForProject(ctx, repo.IsReapingEnabledForProjectParams{
+		ProjectID:       req.ProjectID,
+		OrganizationIds: []string{"5ad61845-b72e-4f0d-9dde-e0bdcf98e374", "5a25158b-24dc-4d49-b03d-e85acfbea59c"},
+	})
+	if err != nil && !errors.Is(err, sql.ErrNoRows) {
+		return fmt.Errorf("check reaping enabled for project: %w", err)
+	}
+	if !enabled {
+		logger.InfoContext(ctx, "reaping is not enabled for project")
+		return nil
+	}
+
 	logger = logger.With(
 		attr.SlogFlyAppName(app.AppName),
 		attr.SlogFlyOrgSlug(app.FlyOrgSlug),
 	)
 
 	logger.InfoContext(ctx, fmt.Sprintf("deleting existing app: %s", app.AppName))
 
+	if app.AppName == "" {
+		logger.InfoContext(ctx, "app name is empty, skipping reap")
+		return nil
+	}
+
+	deleteRequest, err := http.NewRequestWithContext(
+		ctx,
+		http.MethodDelete,
+		fmt.Sprintf("%s/v1/apps/%s", f.machinesAPIBase, app.AppName),
+		nil,
+	)
+	if err != nil {
+		return fmt.Errorf("create delete app request: %w", err)
+	}
+
+	res, err := f.machinesClient.Do(deleteRequest)
+	if err != nil {
+		return fmt.Errorf("send delete app request: %w", err)
+	}
+	defer o11y.LogDefer(ctx, logger, func() error { return res.Body.Close() })
+
+	if res.StatusCode == http.StatusNotFound {
+		logger.InfoContext(ctx, "app not found during delete, assuming already deleted")
+		return nil
+	}
+
 	reapErrorMessage := ""
 	if err := f.client.DeleteApp(ctx, app.AppName); err != nil {
 		// Log error but continue - we still want to mark it as reaped in the database

server/internal/functions/queries.sql

@@ -157,4 +157,11 @@ WHERE
   AND status = 'ready'
   AND reaped_at IS NULL
 ORDER BY created_at DESC
-LIMIT 1;
+LIMIT 1;
+
+-- name: IsReapingEnabledForProject :one
+SELECT true AS enabled
+FROM projects
+WHERE
+  id = @project_id
+  AND organization_id = ANY(@organization_ids);