checkPlanCompletion checks if all tasks in a plan are complete and successful. If so, sends PIPELINE_COMPLETED webhook and auto-resolves issues for deferred rollout plans. Deferred rollout plans (exportDataConfig, createDatabaseConfig) auto-resolve when tasks complete. Called when tasks are marked D
(ctx context.Context, ref bus.PlanRef)
| 195 | // Deferred rollout plans (exportDataConfig, createDatabaseConfig) auto-resolve when tasks complete. |
| 196 | // Called when tasks are marked DONE/SKIPPED, or when tasks are skipped/canceled via API. |
| 197 | func (s *Scheduler) checkPlanCompletion(ctx context.Context, ref bus.PlanRef) { |
| 198 | planID := ref.PlanID |
| 199 | plan, err := s.store.GetPlan(ctx, &store.FindPlanMessage{ProjectID: ref.ProjectID, UID: &planID}) |
| 200 | if err != nil || plan == nil { |
| 201 | slog.Error("failed to get plan for completion check", log.BBError(err)) |
| 202 | return |
| 203 | } |
| 204 | |
| 205 | // Get all tasks for this plan |
| 206 | tasks, err := s.store.ListTasks(ctx, &store.TaskFind{ProjectID: ref.ProjectID, PlanID: &planID}) |
| 207 | if err != nil { |
| 208 | slog.Error("failed to list tasks for plan completion check", log.BBError(err)) |
| 209 | return |
| 210 | } |
| 211 | |
| 212 | // Check if all tasks are complete (DONE or SKIPPED) |
| 213 | for _, task := range tasks { |
| 214 | status := task.LatestTaskRunStatus |
| 215 | |
| 216 | // Only DONE and SKIPPED are considered complete |
| 217 | // FAILED and CANCELED are not complete states |
| 218 | isComplete := status == storepb.TaskRun_DONE || |
| 219 | status == storepb.TaskRun_SKIPPED || |
| 220 | task.Payload.GetSkipped() |
| 221 | |
| 222 | if !isComplete { |
| 223 | // Not all tasks complete - no webhook |
| 224 | return |
| 225 | } |
| 226 | } |
| 227 | |
| 228 | // All tasks complete and successful - try to claim completion notification |
| 229 | claimed, err := s.store.ClaimPipelineCompletionNotification(ctx, ref.ProjectID, planID) |
| 230 | if err != nil { |
| 231 | slog.Error("failed to claim pipeline completion notification", log.BBError(err)) |
| 232 | return |
| 233 | } |
| 234 | if !claimed { |
| 235 | return // Already sent |
| 236 | } |
| 237 | |
| 238 | project, err := s.store.GetProjectByResourceID(ctx, plan.ProjectID) |
| 239 | if err != nil || project == nil { |
| 240 | slog.Error("failed to get project for completion webhook", log.BBError(err)) |
| 241 | return |
| 242 | } |
| 243 | |
| 244 | // Use environment from the first task (all tasks should be in the same environment for a rollout) |
| 245 | environment := "" |
| 246 | if len(tasks) > 0 { |
| 247 | environment = tasks[0].Environment |
| 248 | } |
| 249 | |
| 250 | // Send PIPELINE_COMPLETED webhook |
| 251 | s.webhookManager.CreateEvent(ctx, &webhook.Event{ |
| 252 | Type: storepb.Activity_PIPELINE_COMPLETED, |
| 253 | Project: webhook.NewProject(project), |
| 254 | RolloutCompleted: &webhook.EventRolloutCompleted{ |
no test coverage detected