drop flow: update status with flow name, not workflow id

flow/internal/workflow.go

@@ -42,3 +42,13 @@ func UpdateFlowStatusInCatalog(ctx context.Context, pool shared.CatalogPool,
 	}
 	return status, nil
 }
+
+func UpdateFlowStatusWithNameInCatalog(ctx context.Context, pool shared.CatalogPool,
+	flowName string, status protos.FlowStatus,
+) (protos.FlowStatus, error) {
+	if _, err := pool.Exec(ctx, "UPDATE flows SET status=$1,updated_at=now() WHERE name=$2", status, flowName); err != nil {
+		slog.ErrorContext(ctx, "failed to update flow status", slog.Any("error", err), slog.String("flowName", flowName))
+		return status, fmt.Errorf("failed to update flow status: %w", err)
+	}
+	return status, nil
+}

flow/workflows/cdc_flow.go

@@ -67,14 +67,32 @@ func NewCDCFlowWorkflowState(ctx workflow.Context, logger log.Logger, cfg *proto
 	return &state
 }
 
+func syncStatusToCatalogWithFlowName(ctx workflow.Context, logger log.Logger, status protos.FlowStatus, flowName string) {
+	updateCtx := workflow.WithLocalActivityOptions(ctx, workflow.LocalActivityOptions{
+		StartToCloseTimeout: 1 * time.Minute,
+	})
+
+	if err := workflow.ExecuteLocalActivity(
+		updateCtx,
+		updateFlowStatusWithNameInCatalogActivity,
+		flowName,
+		status,
+	).Get(updateCtx, nil); err != nil {
+		logger.Error("Failed to update flow status in catalog", slog.Any("error", err), slog.String("flowStatus", status.String()))
+	}
+}
+
 func syncStatusToCatalog(ctx workflow.Context, logger log.Logger, status protos.FlowStatus) {
 	updateCtx := workflow.WithLocalActivityOptions(ctx, workflow.LocalActivityOptions{
 		StartToCloseTimeout: 1 * time.Minute,
 	})
 
-	updateFuture := workflow.ExecuteLocalActivity(updateCtx,
-		updateFlowStatusInCatalogActivity, workflow.GetInfo(ctx).WorkflowExecution.ID, status)
-	if err := updateFuture.Get(updateCtx, nil); err != nil {
+	if err := workflow.ExecuteLocalActivity(
+		updateCtx,
+		updateFlowStatusInCatalogActivity,
+		workflow.GetInfo(ctx).WorkflowExecution.ID,
+		status,
+	).Get(updateCtx, nil); err != nil {
 		logger.Error("Failed to update flow status in catalog", slog.Any("error", err), slog.String("flowStatus", status.String()))
 	}
 }

flow/workflows/drop_flow.go

@@ -142,7 +142,7 @@ func DropFlowWorkflow(ctx workflow.Context, input *protos.DropFlowInput) error {
 		status = protos.FlowStatus_STATUS_RESYNC
 	}
 	logger := workflow.GetLogger(ctx)
-	syncStatusToCatalog(ctx, logger, status)
+	syncStatusToCatalogWithFlowName(ctx, logger, status, input.FlowJobName)
 
 	ctx = workflow.WithValue(ctx, shared.FlowNameKey, input.FlowJobName)
 	logger.Info("performing cleanup for flow",

flow/workflows/local_activities.go

@@ -25,3 +25,15 @@ func updateFlowStatusInCatalogActivity(
 	}
 	return internal.UpdateFlowStatusInCatalog(ctx, pool, workflowID, status)
 }
+
+func updateFlowStatusWithNameInCatalogActivity(
+	ctx context.Context,
+	flowName string,
+	status protos.FlowStatus,
+) (protos.FlowStatus, error) {
+	pool, err := internal.GetCatalogConnectionPoolFromEnv(ctx)
+	if err != nil {
+		return status, fmt.Errorf("failed to get catalog connection pool: %w", err)
+	}
+	return internal.UpdateFlowStatusWithNameInCatalog(ctx, pool, flowName, status)
+}