diff --git a/.gitignore b/.gitignore index 4c119ef8fd..099728498a 100644 --- a/.gitignore +++ b/.gitignore @@ -123,3 +123,8 @@ normalized-*.jsonl # Local worktrees (not the backend-managed tree under ~/.kandev) /.worktrees/ + +# PR screenshot/asset scratch dirs — never committed; drag-and-drop into the +# GitHub PR description instead (see apps/web/e2e/scripts/upload-pr-assets.ts). +/.pr-screenshots/ +/apps/web/.pr-assets/ diff --git a/apps/backend/internal/automation/handlers.go b/apps/backend/internal/automation/handlers.go index cceeee3155..f9e74d51d8 100644 --- a/apps/backend/internal/automation/handlers.go +++ b/apps/backend/internal/automation/handlers.go @@ -31,6 +31,8 @@ func registerWSHandlers(dispatcher *ws.Dispatcher, svc *Service, log *logger.Log dispatcher.RegisterFunc(ws.ActionAutomationTriggerDelete, wsDeleteTrigger(svc, log)) dispatcher.RegisterFunc(ws.ActionAutomationTriggerTypes, wsTriggerTypes()) dispatcher.RegisterFunc(ws.ActionAutomationWebhookRevealSecret, wsRevealWebhookSecret(svc, log)) + dispatcher.RegisterFunc(ws.ActionAutomationRunDelete, wsDeleteRun(svc, log)) + dispatcher.RegisterFunc(ws.ActionAutomationRunsDeleteAll, wsDeleteAllRuns(svc, log)) } func registerHTTPRoutes(router *gin.Engine, svc *Service, log *logger.Logger) { @@ -290,3 +292,67 @@ func wsRevealWebhookSecret(svc *Service, _ *logger.Logger) func(ctx context.Cont return ws.NewResponse(msg.ID, msg.Action, &RevealWebhookSecretResponse{WebhookSecret: a.WebhookSecret}) } } + +func wsDeleteRun(svc *Service, _ *logger.Logger) func(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + return func(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + payload, _ := parseMap(msg) + runID, _ := payload["run_id"].(string) + if runID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "run_id required", nil) + } + workspaceID, _ := payload["workspace_id"].(string) + if workspaceID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "workspace_id required", nil) + } + // The WS layer is not itself workspace-scoped, so we resolve the run's + // automation and reject the request when it belongs to a different + // workspace — otherwise any authenticated client who knows a foreign + // run_id could permanently delete it. NotFound (not Forbidden) avoids + // disclosing whether the id exists at all, mirroring wsRevealWebhookSecret. + run, err := svc.GetRun(ctx, runID) + if err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, err.Error(), nil) + } + if run == nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeNotFound, "run not found", nil) + } + a, err := svc.GetAutomation(ctx, run.AutomationID) + if err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, err.Error(), nil) + } + if a == nil || a.WorkspaceID != workspaceID { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeNotFound, "run not found", nil) + } + if err := svc.DeleteRun(ctx, runID); err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, err.Error(), nil) + } + return ws.NewResponse(msg.ID, msg.Action, map[string]bool{"deleted": true}) + } +} + +func wsDeleteAllRuns(svc *Service, _ *logger.Logger) func(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + return func(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + payload, _ := parseMap(msg) + automationID, _ := payload["automation_id"].(string) + if automationID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "automation_id required", nil) + } + workspaceID, _ := payload["workspace_id"].(string) + if workspaceID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "workspace_id required", nil) + } + // See wsDeleteRun: the WS layer isn't workspace-scoped, so this + // destructive bulk delete must verify ownership itself. + a, err := svc.GetAutomation(ctx, automationID) + if err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, err.Error(), nil) + } + if a == nil || a.WorkspaceID != workspaceID { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeNotFound, "automation not found", nil) + } + if err := svc.DeleteAllRuns(ctx, automationID); err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, err.Error(), nil) + } + return ws.NewResponse(msg.ID, msg.Action, map[string]bool{"deleted": true}) + } +} diff --git a/apps/backend/internal/automation/handlers_test.go b/apps/backend/internal/automation/handlers_test.go index b8f950b614..88d62553b6 100644 --- a/apps/backend/internal/automation/handlers_test.go +++ b/apps/backend/internal/automation/handlers_test.go @@ -186,3 +186,227 @@ func TestWsRevealWebhookSecret_RejectsCrossWorkspace(t *testing.T) { t.Errorf("expected NOT_FOUND to avoid disclosing existence, got %q", ep.Code) } } + +func TestWsDeleteRun_DeletesRun(t *testing.T) { + svc := newTestService(t) + log, _ := logger.NewFromZap(zap.NewNop()) + ctx := context.Background() + + // Create an automation and a run. + a := &Automation{WorkspaceID: "ws-1", Name: "X", WorkflowID: "wf-1", WorkflowStepID: "s-1", Enabled: true} + if err := svc.store.CreateAutomation(ctx, a); err != nil { + t.Fatal(err) + } + run := &AutomationRun{ + AutomationID: a.ID, + TriggerType: TriggerTypeScheduled, + Status: RunStatusSkipped, + TriggerData: json.RawMessage(`{}`), + } + if err := svc.store.CreateRun(ctx, run); err != nil { + t.Fatal(err) + } + + req, err := ws.NewRequest("req-del", ws.ActionAutomationRunDelete, map[string]string{"run_id": run.ID, "workspace_id": "ws-1"}) + if err != nil { + t.Fatal(err) + } + resp, err := wsDeleteRun(svc, log)(ctx, req) + if err != nil { + t.Fatal(err) + } + if resp.Type != ws.MessageTypeResponse { + t.Fatalf("expected response, got %v: %s", resp.Type, string(resp.Payload)) + } + + // Run should be gone. + got, _ := svc.store.GetRun(ctx, run.ID) + if got != nil { + t.Error("expected run to be deleted") + } +} + +func TestWsDeleteRun_RequiresRunID(t *testing.T) { + svc := newTestService(t) + log, _ := logger.NewFromZap(zap.NewNop()) + ctx := context.Background() + + req, err := ws.NewRequest("req-1", ws.ActionAutomationRunDelete, map[string]string{"workspace_id": "ws-1"}) + if err != nil { + t.Fatal(err) + } + resp, err := wsDeleteRun(svc, log)(ctx, req) + if err != nil { + t.Fatal(err) + } + var ep struct { + Code string `json:"code"` + } + _ = json.Unmarshal(resp.Payload, &ep) + if ep.Code != ws.ErrorCodeBadRequest { + t.Errorf("expected BAD_REQUEST, got %q", ep.Code) + } +} + +func TestWsDeleteRun_RequiresWorkspaceID(t *testing.T) { + svc := newTestService(t) + log, _ := logger.NewFromZap(zap.NewNop()) + ctx := context.Background() + + req, err := ws.NewRequest("req-1", ws.ActionAutomationRunDelete, map[string]string{"run_id": "run-1"}) + if err != nil { + t.Fatal(err) + } + resp, err := wsDeleteRun(svc, log)(ctx, req) + if err != nil { + t.Fatal(err) + } + var ep ws.ErrorPayload + _ = json.Unmarshal(resp.Payload, &ep) + if ep.Code != ws.ErrorCodeBadRequest { + t.Errorf("expected BAD_REQUEST, got %q", ep.Code) + } +} + +func TestWsDeleteRun_RejectsCrossWorkspace(t *testing.T) { + svc := newTestService(t) + log, _ := logger.NewFromZap(zap.NewNop()) + ctx := context.Background() + + // Run belongs to an automation in workspace A. + a := &Automation{WorkspaceID: "ws-A", Name: "X", WorkflowID: "wf-1", WorkflowStepID: "s-1", Enabled: true} + if err := svc.store.CreateAutomation(ctx, a); err != nil { + t.Fatal(err) + } + run := &AutomationRun{ + AutomationID: a.ID, + TriggerType: TriggerTypeScheduled, + Status: RunStatusSkipped, + TriggerData: json.RawMessage(`{}`), + } + if err := svc.store.CreateRun(ctx, run); err != nil { + t.Fatal(err) + } + + // A caller in workspace B must not be able to delete it. + req, _ := ws.NewRequest("req-1", ws.ActionAutomationRunDelete, map[string]any{"run_id": run.ID, "workspace_id": "ws-B"}) + resp, err := wsDeleteRun(svc, log)(ctx, req) + if err != nil { + t.Fatal(err) + } + if resp.Type != ws.MessageTypeError { + t.Fatalf("expected error response for cross-workspace delete, got %v: %s", resp.Type, string(resp.Payload)) + } + var ep ws.ErrorPayload + if err := json.Unmarshal(resp.Payload, &ep); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if ep.Code != ws.ErrorCodeNotFound { + t.Errorf("expected NOT_FOUND to avoid disclosing existence, got %q", ep.Code) + } + + // The run must survive the rejected request. + got, _ := svc.store.GetRun(ctx, run.ID) + if got == nil { + t.Error("run was deleted despite cross-workspace mismatch — regression") + } +} + +func TestWsDeleteAllRuns_ClearsAllRuns(t *testing.T) { + svc := newTestService(t) + log, _ := logger.NewFromZap(zap.NewNop()) + ctx := context.Background() + + a := &Automation{WorkspaceID: "ws-1", Name: "Y", WorkflowID: "wf-1", WorkflowStepID: "s-1", Enabled: true} + if err := svc.store.CreateAutomation(ctx, a); err != nil { + t.Fatal(err) + } + for range 3 { + if err := svc.store.CreateRun(ctx, &AutomationRun{ + AutomationID: a.ID, + TriggerType: TriggerTypeScheduled, + Status: RunStatusSkipped, + TriggerData: json.RawMessage(`{}`), + }); err != nil { + t.Fatal(err) + } + } + + req, err := ws.NewRequest("req-all", ws.ActionAutomationRunsDeleteAll, map[string]string{"automation_id": a.ID, "workspace_id": "ws-1"}) + if err != nil { + t.Fatal(err) + } + resp, err := wsDeleteAllRuns(svc, log)(ctx, req) + if err != nil { + t.Fatal(err) + } + if resp.Type != ws.MessageTypeResponse { + t.Fatalf("expected response, got %v: %s", resp.Type, string(resp.Payload)) + } + + runs, _ := svc.store.ListRuns(ctx, a.ID, 50) + if len(runs) != 0 { + t.Errorf("expected 0 runs after delete-all, got %d", len(runs)) + } +} + +func TestWsDeleteAllRuns_RequiresWorkspaceID(t *testing.T) { + svc := newTestService(t) + log, _ := logger.NewFromZap(zap.NewNop()) + ctx := context.Background() + + req, err := ws.NewRequest("req-1", ws.ActionAutomationRunsDeleteAll, map[string]string{"automation_id": "auto-1"}) + if err != nil { + t.Fatal(err) + } + resp, err := wsDeleteAllRuns(svc, log)(ctx, req) + if err != nil { + t.Fatal(err) + } + var ep ws.ErrorPayload + _ = json.Unmarshal(resp.Payload, &ep) + if ep.Code != ws.ErrorCodeBadRequest { + t.Errorf("expected BAD_REQUEST, got %q", ep.Code) + } +} + +func TestWsDeleteAllRuns_RejectsCrossWorkspace(t *testing.T) { + svc := newTestService(t) + log, _ := logger.NewFromZap(zap.NewNop()) + ctx := context.Background() + + a := &Automation{WorkspaceID: "ws-A", Name: "Y", WorkflowID: "wf-1", WorkflowStepID: "s-1", Enabled: true} + if err := svc.store.CreateAutomation(ctx, a); err != nil { + t.Fatal(err) + } + if err := svc.store.CreateRun(ctx, &AutomationRun{ + AutomationID: a.ID, + TriggerType: TriggerTypeScheduled, + Status: RunStatusSkipped, + TriggerData: json.RawMessage(`{}`), + }); err != nil { + t.Fatal(err) + } + + req, _ := ws.NewRequest("req-1", ws.ActionAutomationRunsDeleteAll, map[string]any{"automation_id": a.ID, "workspace_id": "ws-B"}) + resp, err := wsDeleteAllRuns(svc, log)(ctx, req) + if err != nil { + t.Fatal(err) + } + if resp.Type != ws.MessageTypeError { + t.Fatalf("expected error response for cross-workspace delete-all, got %v: %s", resp.Type, string(resp.Payload)) + } + var ep ws.ErrorPayload + if err := json.Unmarshal(resp.Payload, &ep); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if ep.Code != ws.ErrorCodeNotFound { + t.Errorf("expected NOT_FOUND to avoid disclosing existence, got %q", ep.Code) + } + + // Runs must survive the rejected request. + runs, _ := svc.store.ListRuns(ctx, a.ID, 50) + if len(runs) != 1 { + t.Errorf("expected the run to survive a rejected cross-workspace delete-all, got %d runs", len(runs)) + } +} diff --git a/apps/backend/internal/automation/service.go b/apps/backend/internal/automation/service.go index be76abbedd..8e75237190 100644 --- a/apps/backend/internal/automation/service.go +++ b/apps/backend/internal/automation/service.go @@ -3,7 +3,9 @@ package automation import ( "context" "encoding/json" + "errors" "fmt" + "sync" "time" "go.uber.org/zap" @@ -13,11 +15,36 @@ import ( "github.com/kandev/kandev/internal/events/bus" ) +// ErrTaskNotFound is the sentinel that run-cleanup paths check to distinguish +// "the task is already gone — fine, drop the run row anyway" from a real +// upstream failure. TaskDeleter implementations should wrap this when the +// task domain reports a missing row, so this package can recognize the case +// via errors.Is without importing the task repository package (see +// backendapp's taskDeleterAdapter for the production wiring). +var ErrTaskNotFound = errors.New("automation: task not found for cleanup") + +// TaskDeleter deletes a task and cleans up its resources. +// Satisfied by *taskservice.Service; injected to avoid a cyclic import. +// Implementations should return errors wrapping ErrTaskNotFound when the +// task is already gone. +type TaskDeleter interface { + DeleteTask(ctx context.Context, id string) error +} + // Service coordinates automation operations. type Service struct { - store *Store - eventBus bus.EventBus - logger *logger.Logger + store *Store + eventBus bus.EventBus + logger *logger.Logger + taskDeleter TaskDeleter // optional; nil-safe + + // runLocks serializes run creation (RecordRun, the concurrency-cap skip + // insert) against DeleteAllRuns per automation ID. Without this, a run + // created between DeleteAllRuns' task-id snapshot and its final row + // purge would have its row deleted without its task ever reaching the + // TaskDeleter — orphaning the task. Entries are never removed: growth is + // bounded by the number of distinct automation IDs (~160 B per entry). + runLocks sync.Map // automationID (string) -> *sync.Mutex } // NewService creates a new automation service. @@ -34,6 +61,12 @@ func (s *Service) Store() *Store { return s.store } +// SetTaskDeleter wires the task deletion handler for run cleanup. +// Optional: when nil, run deletion skips task teardown. +func (s *Service) SetTaskDeleter(d TaskDeleter) { + s.taskDeleter = d +} + // --- Automation CRUD --- // CreateAutomation creates an automation with its initial triggers. @@ -174,6 +207,73 @@ func (s *Service) ListRuns(ctx context.Context, automationID string, limit int) return s.store.ListRuns(ctx, automationID, limit) } +// GetRun returns a single run by ID, or nil if not found. +func (s *Service) GetRun(ctx context.Context, id string) (*AutomationRun, error) { + return s.store.GetRun(ctx, id) +} + +// automationRunLock returns an unlock func for the per-automation mutex that +// serializes run creation (createRunLocked) against DeleteAllRuns. +func (s *Service) automationRunLock(automationID string) func() { + v, _ := s.runLocks.LoadOrStore(automationID, &sync.Mutex{}) + mu := v.(*sync.Mutex) + mu.Lock() + return mu.Unlock +} + +// createRunLocked persists a run row while holding the per-automation lock +// that DeleteAllRuns also acquires. Without this, a run created between +// DeleteAllRuns' task-id snapshot and its final row purge would be deleted +// without its task ever reaching the TaskDeleter, orphaning the task. +func (s *Service) createRunLocked(ctx context.Context, run *AutomationRun) error { + defer s.automationRunLock(run.AutomationID)() + return s.store.CreateRun(ctx, run) +} + +// DeleteRun removes a single run and its associated task (if any). +// Task deletion is best-effort: a not-found error is silently ignored so +// stale/orphaned run rows are always removable by the user. +func (s *Service) DeleteRun(ctx context.Context, runID string) error { + run, err := s.store.GetRun(ctx, runID) + if err != nil { + return fmt.Errorf("get run: %w", err) + } + if run != nil && run.TaskID != "" && s.taskDeleter != nil { + if delErr := s.taskDeleter.DeleteTask(ctx, run.TaskID); delErr != nil { + if !errors.Is(delErr, ErrTaskNotFound) { + return fmt.Errorf("delete task: %w", delErr) + } + s.logger.Debug("run task already gone, continuing delete", + zap.String("run_id", runID), + zap.String("task_id", run.TaskID)) + } + } + return s.store.DeleteRun(ctx, runID) +} + +// DeleteAllRuns removes every run for an automation, deleting each associated +// task first. Task deletion is best-effort: not-found errors are ignored. +func (s *Service) DeleteAllRuns(ctx context.Context, automationID string) error { + defer s.automationRunLock(automationID)() + if s.taskDeleter != nil { + taskIDs, err := s.store.ListRunTaskIDs(ctx, automationID) + if err != nil { + return fmt.Errorf("list run task ids: %w", err) + } + for _, taskID := range taskIDs { + if delErr := s.taskDeleter.DeleteTask(ctx, taskID); delErr != nil { + if !errors.Is(delErr, ErrTaskNotFound) { + return fmt.Errorf("delete task %s: %w", taskID, delErr) + } + s.logger.Debug("run task already gone, skipping", + zap.String("automation_id", automationID), + zap.String("task_id", taskID)) + } + } + } + return s.store.DeleteAllRuns(ctx, automationID) +} + // --- Trigger firing --- // FireTrigger publishes an AutomationTriggered event for the given trigger. @@ -248,7 +348,7 @@ func (s *Service) FireTrigger(ctx context.Context, automationID, triggerID strin // RecordRun records a trigger run outcome. func (s *Service) RecordRun(ctx context.Context, run *AutomationRun) error { - return s.store.CreateRun(ctx, run) + return s.createRunLocked(ctx, run) } // maybeSkipForConcurrencyCap enforces max_concurrent_runs. Returns (skipped, @@ -278,7 +378,7 @@ func (s *Service) maybeSkipForConcurrencyCap(ctx context.Context, automationID, TriggerData: triggerData, ErrorMessage: fmt.Sprintf("max_concurrent_runs=%d reached", a.MaxConcurrentRuns), } - if recErr := s.store.CreateRun(ctx, skipRun); recErr != nil { + if recErr := s.createRunLocked(ctx, skipRun); recErr != nil { s.logger.Warn("failed to record skipped run", zap.Error(recErr)) } s.logger.Info("automation trigger skipped: concurrency cap reached", diff --git a/apps/backend/internal/automation/service_test.go b/apps/backend/internal/automation/service_test.go new file mode 100644 index 0000000000..a9f7c57b0d --- /dev/null +++ b/apps/backend/internal/automation/service_test.go @@ -0,0 +1,347 @@ +package automation + +import ( + "context" + "encoding/json" + "sync" + "testing" + "time" + + "github.com/jmoiron/sqlx" + "go.uber.org/zap" + + "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/internal/events/bus" +) + +// fakeTaskDeleter records deletions and can inject errors per task ID. +type fakeTaskDeleter struct { + deleted []string + errors map[string]error +} + +func (f *fakeTaskDeleter) DeleteTask(_ context.Context, id string) error { + f.deleted = append(f.deleted, id) + if f.errors != nil { + if err, ok := f.errors[id]; ok { + return err + } + } + return nil +} + +func TestService_DeleteRun_CallsTaskDeleter(t *testing.T) { + svc := newTestService(t) + deleter := &fakeTaskDeleter{} + svc.SetTaskDeleter(deleter) + ctx := context.Background() + + a := &Automation{WorkspaceID: "ws-1", Name: "A", WorkflowID: "wf-1", WorkflowStepID: "s-1", Enabled: true} + if err := svc.store.CreateAutomation(ctx, a); err != nil { + t.Fatal(err) + } + run := &AutomationRun{ + AutomationID: a.ID, + TriggerType: TriggerTypeScheduled, + Status: RunStatusTaskCreated, + TaskID: "task-xyz", + TriggerData: json.RawMessage(`{}`), + } + if err := svc.store.CreateRun(ctx, run); err != nil { + t.Fatal(err) + } + + if err := svc.DeleteRun(ctx, run.ID); err != nil { + t.Fatalf("DeleteRun: %v", err) + } + + // Task deleter must have been called. + if len(deleter.deleted) != 1 || deleter.deleted[0] != "task-xyz" { + t.Errorf("expected DeleteTask(task-xyz), got %v", deleter.deleted) + } + // Run row must be gone. + got, _ := svc.store.GetRun(ctx, run.ID) + if got != nil { + t.Error("expected run row to be removed") + } +} + +func TestService_DeleteRun_TaskNotFound_StillDeletesRun(t *testing.T) { + svc := newTestService(t) + deleter := &fakeTaskDeleter{ + errors: map[string]error{"task-gone": ErrTaskNotFound}, + } + svc.SetTaskDeleter(deleter) + ctx := context.Background() + + a := &Automation{WorkspaceID: "ws-1", Name: "B", WorkflowID: "wf-1", WorkflowStepID: "s-1", Enabled: true} + if err := svc.store.CreateAutomation(ctx, a); err != nil { + t.Fatal(err) + } + run := &AutomationRun{ + AutomationID: a.ID, + TriggerType: TriggerTypeScheduled, + Status: RunStatusSkipped, + TaskID: "task-gone", + TriggerData: json.RawMessage(`{}`), + } + if err := svc.store.CreateRun(ctx, run); err != nil { + t.Fatal(err) + } + + // Must succeed even though the task is not found. + if err := svc.DeleteRun(ctx, run.ID); err != nil { + t.Fatalf("DeleteRun with not-found task: %v", err) + } + + // Run row must still be gone. + got, _ := svc.store.GetRun(ctx, run.ID) + if got != nil { + t.Error("expected run row to be removed despite task-not-found") + } +} + +func TestService_DeleteAllRuns_CallsTaskDeleterForEach(t *testing.T) { + svc := newTestService(t) + deleter := &fakeTaskDeleter{} + svc.SetTaskDeleter(deleter) + ctx := context.Background() + + a := &Automation{WorkspaceID: "ws-1", Name: "C", WorkflowID: "wf-1", WorkflowStepID: "s-1", Enabled: true} + if err := svc.store.CreateAutomation(ctx, a); err != nil { + t.Fatal(err) + } + taskIDs := []string{"task-1", "task-2", "task-3"} + for _, tid := range taskIDs { + if err := svc.store.CreateRun(ctx, &AutomationRun{ + AutomationID: a.ID, + TriggerType: TriggerTypeScheduled, + Status: RunStatusTaskCreated, + TaskID: tid, + TriggerData: json.RawMessage(`{}`), + }); err != nil { + t.Fatal(err) + } + } + // Also one run with no task_id (fire-and-forget / skipped). + if err := svc.store.CreateRun(ctx, &AutomationRun{ + AutomationID: a.ID, + TriggerType: TriggerTypeScheduled, + Status: RunStatusSkipped, + TriggerData: json.RawMessage(`{}`), + }); err != nil { + t.Fatal(err) + } + + if err := svc.DeleteAllRuns(ctx, a.ID); err != nil { + t.Fatalf("DeleteAllRuns: %v", err) + } + + // All three task IDs must have been passed to DeleteTask. + if len(deleter.deleted) != 3 { + t.Errorf("expected 3 task deletions, got %d: %v", len(deleter.deleted), deleter.deleted) + } + // All run rows gone. + runs, _ := svc.store.ListRuns(ctx, a.ID, 50) + if len(runs) != 0 { + t.Errorf("expected 0 runs, got %d", len(runs)) + } +} + +func TestService_DeleteAllRuns_TaskNotFound_StillClearsRuns(t *testing.T) { + svc := newTestService(t) + deleter := &fakeTaskDeleter{ + errors: map[string]error{"task-stale": ErrTaskNotFound}, + } + svc.SetTaskDeleter(deleter) + ctx := context.Background() + + a := &Automation{WorkspaceID: "ws-1", Name: "D", WorkflowID: "wf-1", WorkflowStepID: "s-1", Enabled: true} + if err := svc.store.CreateAutomation(ctx, a); err != nil { + t.Fatal(err) + } + for _, tid := range []string{"task-stale", "task-ok"} { + if err := svc.store.CreateRun(ctx, &AutomationRun{ + AutomationID: a.ID, + TriggerType: TriggerTypeScheduled, + Status: RunStatusTaskCreated, + TaskID: tid, + TriggerData: json.RawMessage(`{}`), + }); err != nil { + t.Fatal(err) + } + } + + if err := svc.DeleteAllRuns(ctx, a.ID); err != nil { + t.Fatalf("DeleteAllRuns with not-found task: %v", err) + } + + runs, _ := svc.store.ListRuns(ctx, a.ID, 50) + if len(runs) != 0 { + t.Errorf("expected 0 runs after delete-all, got %d", len(runs)) + } +} + +// TestService_DeleteAllRuns_SerializesAgainstConcurrentRecordRun is a +// regression guard for the orphaned-task race: DeleteAllRuns snapshots task +// IDs via ListRunTaskIDs and then issues a broad DELETE by automation_id. If +// a run were recorded in between, its row would be purged by the broad +// DELETE without its task ever reaching the TaskDeleter. The per-automation +// run lock must make RecordRun block for the full duration of DeleteAllRuns. +func TestService_DeleteAllRuns_SerializesAgainstConcurrentRecordRun(t *testing.T) { + svc := newTestService(t) + svc.SetTaskDeleter(&fakeTaskDeleter{}) + ctx := context.Background() + + a := &Automation{WorkspaceID: "ws-1", Name: "Race", WorkflowID: "wf-1", WorkflowStepID: "s-1", Enabled: true} + if err := svc.store.CreateAutomation(ctx, a); err != nil { + t.Fatal(err) + } + if err := svc.store.CreateRun(ctx, &AutomationRun{ + AutomationID: a.ID, + TriggerType: TriggerTypeScheduled, + Status: RunStatusTaskCreated, + TaskID: "task-existing", + TriggerData: json.RawMessage(`{}`), + }); err != nil { + t.Fatal(err) + } + + // Simulate DeleteAllRuns being mid-flight by holding its lock directly. + // Register cleanup immediately, before any t.Fatal path, so a failure in + // the select below (RecordRun completing too early) can't leak the lock + // for the rest of the test binary's life. sync.Once makes the explicit + // unlock() call below and this cleanup mutually safe against a double + // Unlock() panic. + unlock := svc.automationRunLock(a.ID) + var unlockOnce sync.Once + safeUnlock := func() { unlockOnce.Do(unlock) } + t.Cleanup(safeUnlock) + + done := make(chan error, 1) + go func() { + done <- svc.RecordRun(ctx, &AutomationRun{ + AutomationID: a.ID, + TriggerType: TriggerTypeScheduled, + Status: RunStatusTaskCreated, + TaskID: "task-new", + TriggerData: json.RawMessage(`{}`), + }) + }() + + select { + case <-done: + t.Fatal("RecordRun completed while the automation run lock was held — DeleteAllRuns is not serialized against run creation") + case <-time.After(50 * time.Millisecond): + // Expected: RecordRun is still blocked on the lock. + } + + safeUnlock() + + if err := <-done; err != nil { + t.Fatalf("RecordRun: %v", err) + } + + runs, _ := svc.store.ListRuns(ctx, a.ID, 50) + if len(runs) != 2 { + t.Errorf("expected 2 runs (pre-existing + the one recorded after unlock), got %d", len(runs)) + } +} + +// TestDeleteAllRuns_AutomationSurvives is a regression guard: deleting all run +// rows — including issuing real DELETE SQL against task rows in the shared +// in-memory DB — must never delete the parent automation row. A real DB-level +// deleter catches SQL trigger / ON DELETE CASCADE regressions. Note: event +// handler side-effects are not covered here (no orchestrator runs in this test). +func TestDeleteAllRuns_AutomationSurvives(t *testing.T) { + store := setupTestStore(t) + ctx := context.Background() + + // Create a minimal tasks table in the same in-memory DB so the + // real-deleter can insert and then DELETE task rows. + if _, err := store.db.ExecContext(ctx, + `CREATE TABLE IF NOT EXISTS tasks (id TEXT PRIMARY KEY, title TEXT, state TEXT)`); err != nil { + t.Fatal("create tasks table:", err) + } + + // sqliteTaskDeleter deletes from the real tasks table in the same DB — + // any SQL cascade or trigger that touched automations would fire here. + realDeleter := &sqliteTaskDeleter{db: store.db} + + log, _ := logger.NewFromZap(zap.NewNop()) + eb := bus.NewMemoryEventBus(log) + svc := NewService(store, eb, log) + svc.SetTaskDeleter(realDeleter) + + a := &Automation{WorkspaceID: "ws-1", Name: "Survives", WorkflowID: "wf-1", WorkflowStepID: "s-1", Enabled: true} + if err := store.CreateAutomation(ctx, a); err != nil { + t.Fatal(err) + } + + // Insert real task rows and create runs referencing them. + taskIDs := []string{"task-a", "task-b", "task-c"} + for _, tid := range taskIDs { + if _, err := store.db.ExecContext(ctx, + `INSERT INTO tasks (id, title, state) VALUES (?, 'Test task', 'running')`, tid); err != nil { + t.Fatal("insert task:", err) + } + if err := store.CreateRun(ctx, &AutomationRun{ + AutomationID: a.ID, + TriggerType: TriggerTypeScheduled, + Status: RunStatusTaskCreated, + TaskID: tid, + TriggerData: json.RawMessage(`{}`), + }); err != nil { + t.Fatal(err) + } + } + // Skipped runs without task IDs. + for range 3 { + if err := store.CreateRun(ctx, &AutomationRun{ + AutomationID: a.ID, TriggerType: TriggerTypeScheduled, + Status: RunStatusSkipped, TriggerData: json.RawMessage(`{}`), + }); err != nil { + t.Fatal(err) + } + } + + if err := svc.DeleteAllRuns(ctx, a.ID); err != nil { + t.Fatalf("DeleteAllRuns: %v", err) + } + + // Automation row must still exist after real task DELETEs fired. + got, err := store.GetAutomation(ctx, a.ID) + if err != nil { + t.Fatalf("GetAutomation after DeleteAllRuns: %v", err) + } + if got == nil { + t.Error("automation was deleted by DeleteAllRuns — regression") + return + } + if got.Name != "Survives" { + t.Errorf("unexpected automation name %q", got.Name) + } + + // Runs must be gone. + runs, _ := store.ListRuns(ctx, a.ID, 50) + if len(runs) != 0 { + t.Errorf("expected 0 runs, got %d", len(runs)) + } + // Task rows should have been deleted. + if len(realDeleter.deleted) != 3 { + t.Errorf("expected 3 task deletions, got %d: %v", len(realDeleter.deleted), realDeleter.deleted) + } +} + +// sqliteTaskDeleter deletes from the real tasks table in the same in-memory +// DB, so any SQL trigger or ON DELETE CASCADE that touches automations fires. +type sqliteTaskDeleter struct { + db *sqlx.DB + deleted []string +} + +func (d *sqliteTaskDeleter) DeleteTask(ctx context.Context, id string) error { + d.deleted = append(d.deleted, id) + _, err := d.db.ExecContext(ctx, `DELETE FROM tasks WHERE id = ?`, id) + return err +} diff --git a/apps/backend/internal/automation/store.go b/apps/backend/internal/automation/store.go index f70823195e..4f8fb28f99 100644 --- a/apps/backend/internal/automation/store.go +++ b/apps/backend/internal/automation/store.go @@ -471,6 +471,43 @@ func (s *Store) CountActiveRuns(ctx context.Context, automationID string) (int, return count, err } +// GetRun returns a single run by ID, or nil if not found. +func (s *Store) GetRun(ctx context.Context, id string) (*AutomationRun, error) { + var r AutomationRun + err := s.ro.GetContext(ctx, &r, + `SELECT * FROM automation_runs WHERE id = ?`, id) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + return nil, nil + } + return nil, err + } + r.TriggerData = json.RawMessage(r.TriggerDataJSON) + return &r, nil +} + +// DeleteRun removes a single run row. +func (s *Store) DeleteRun(ctx context.Context, id string) error { + _, err := s.db.ExecContext(ctx, `DELETE FROM automation_runs WHERE id = ?`, id) + return err +} + +// ListRunTaskIDs returns all non-empty task_id values for an automation's runs. +// Used by DeleteAllRuns so the service can clean up tasks before purging rows. +func (s *Store) ListRunTaskIDs(ctx context.Context, automationID string) ([]string, error) { + var ids []string + err := s.ro.SelectContext(ctx, &ids, + `SELECT task_id FROM automation_runs WHERE automation_id = ? AND task_id != ''`, + automationID) + return ids, err +} + +// DeleteAllRuns removes every run row for an automation. +func (s *Store) DeleteAllRuns(ctx context.Context, automationID string) error { + _, err := s.db.ExecContext(ctx, `DELETE FROM automation_runs WHERE automation_id = ?`, automationID) + return err +} + // DeleteAutomationsByWorkspace removes all automations (and their triggers/runs) for a workspace. // Used by e2e reset. func (s *Store) DeleteAutomationsByWorkspace(ctx context.Context, workspaceID string) (int, error) { diff --git a/apps/backend/internal/automation/store_test.go b/apps/backend/internal/automation/store_test.go index 373be78a7f..a97108e991 100644 --- a/apps/backend/internal/automation/store_test.go +++ b/apps/backend/internal/automation/store_test.go @@ -212,3 +212,89 @@ func TestDeleteAutomation(t *testing.T) { t.Error("expected nil after delete") } } + +func TestDeleteRun(t *testing.T) { + store := setupTestStore(t) + ctx := context.Background() + + a := &Automation{WorkspaceID: "ws-1", Name: "A", WorkflowID: "wf-1", WorkflowStepID: "s-1", Enabled: true} + if err := store.CreateAutomation(ctx, a); err != nil { + t.Fatal(err) + } + run := &AutomationRun{ + AutomationID: a.ID, + TriggerType: TriggerTypeScheduled, + Status: RunStatusSkipped, + TaskID: "task-abc", + TriggerData: json.RawMessage(`{}`), + } + if err := store.CreateRun(ctx, run); err != nil { + t.Fatal(err) + } + + // GetRun finds it. + got, err := store.GetRun(ctx, run.ID) + if err != nil { + t.Fatal(err) + } + if got == nil { + t.Error("expected run, got nil") + return + } + if got.TaskID != "task-abc" { + t.Errorf("expected task_id 'task-abc', got %q", got.TaskID) + } + + // DeleteRun removes it. + if err := store.DeleteRun(ctx, run.ID); err != nil { + t.Fatal(err) + } + got, err = store.GetRun(ctx, run.ID) + if err != nil { + t.Fatal(err) + } + if got != nil { + t.Error("expected nil after delete, got run") + } +} + +func TestDeleteAllRuns(t *testing.T) { + store := setupTestStore(t) + ctx := context.Background() + + a := &Automation{WorkspaceID: "ws-1", Name: "B", WorkflowID: "wf-1", WorkflowStepID: "s-1", Enabled: true} + if err := store.CreateAutomation(ctx, a); err != nil { + t.Fatal(err) + } + for i := 0; i < 3; i++ { + if err := store.CreateRun(ctx, &AutomationRun{ + AutomationID: a.ID, + TriggerType: TriggerTypeScheduled, + Status: RunStatusSkipped, + TaskID: "task-" + string(rune('0'+i)), + TriggerData: json.RawMessage(`{}`), + }); err != nil { + t.Fatal(err) + } + } + + taskIDs, err := store.ListRunTaskIDs(ctx, a.ID) + if err != nil { + t.Fatal(err) + } + if len(taskIDs) != 3 { + t.Fatalf("expected 3 task IDs, got %d", len(taskIDs)) + } + + if err := store.DeleteAllRuns(ctx, a.ID); err != nil { + t.Fatal(err) + } + + runs, err := store.ListRuns(ctx, a.ID, 50) + if err != nil { + t.Fatal(err) + } + if len(runs) != 0 { + t.Errorf("expected 0 runs after delete, got %d", len(runs)) + } +} diff --git a/apps/backend/internal/backendapp/e2e_reset.go b/apps/backend/internal/backendapp/e2e_reset.go index 581a70ac53..d10baf17b7 100644 --- a/apps/backend/internal/backendapp/e2e_reset.go +++ b/apps/backend/internal/backendapp/e2e_reset.go @@ -18,6 +18,7 @@ import ( // errKey is the JSON field used for error responses from the E2E endpoints. const errKey = "error" +const statusKey = "status" // registerE2EResetRoutes registers the E2E test-only endpoints. // The endpoints are available when KANDEV_MOCK_AGENT is "true" or "only" (dev/E2E modes). @@ -40,6 +41,12 @@ func registerE2EResetRoutes( // workflow path (e.g. improve-kandev) without depending on the real // bootstrap endpoint, which clones from GitHub and shells out to gh. api.POST("/hidden-workflow", handleE2ECreateHiddenWorkflow(taskSvc, log)) + // Automation seeding helpers for E2E tests that need runs without going + // through the WS API (works on any Node version). + if automationSvc != nil { + api.POST("/automations", handleE2ECreateAutomation(automationSvc, log)) + api.POST("/automation-runs", handleE2ECreateAutomationRun(automationSvc, log)) + } log.Info("registered E2E endpoints (test-only)") } @@ -215,3 +222,67 @@ func handleE2ECreateHiddenWorkflow(taskSvc *taskservice.Service, log *logger.Log }) } } + +type e2eCreateAutomationRequest struct { + WorkspaceID string `json:"workspace_id"` + Name string `json:"name"` + WorkflowID string `json:"workflow_id"` + WorkflowStepID string `json:"workflow_step_id"` +} + +// handleE2ECreateAutomation seeds an automation for E2E tests via HTTP so tests +// don't require Node 24 (WS API requires a global WebSocket that isn't in Node 20). +func handleE2ECreateAutomation(svc *automation.Service, log *logger.Logger) gin.HandlerFunc { + return func(c *gin.Context) { + var body e2eCreateAutomationRequest + if err := c.ShouldBindJSON(&body); err != nil || body.WorkspaceID == "" || body.Name == "" { + c.JSON(http.StatusBadRequest, gin.H{errKey: "workspace_id and name are required"}) + return + } + a, err := svc.CreateAutomation(c.Request.Context(), &automation.CreateAutomationRequest{ + WorkspaceID: body.WorkspaceID, + Name: body.Name, + WorkflowID: body.WorkflowID, + WorkflowStepID: body.WorkflowStepID, + MaxConcurrentRuns: 10, + }) + if err != nil { + log.Error("e2e: failed to create automation", zap.Error(err)) + c.JSON(http.StatusInternalServerError, gin.H{errKey: err.Error()}) + return + } + c.JSON(http.StatusCreated, gin.H{"id": a.ID, "workspace_id": a.WorkspaceID, "name": a.Name}) + } +} + +type e2eCreateAutomationRunRequest struct { + AutomationID string `json:"automation_id"` + Status string `json:"status"` +} + +// handleE2ECreateAutomationRun seeds an automation run row for E2E tests. +func handleE2ECreateAutomationRun(svc *automation.Service, log *logger.Logger) gin.HandlerFunc { + return func(c *gin.Context) { + var body e2eCreateAutomationRunRequest + if err := c.ShouldBindJSON(&body); err != nil || body.AutomationID == "" { + c.JSON(http.StatusBadRequest, gin.H{errKey: "automation_id is required"}) + return + } + status := automation.RunStatus(body.Status) + if status == "" { + status = automation.RunStatusSkipped + } + run := &automation.AutomationRun{ + AutomationID: body.AutomationID, + TriggerType: automation.TriggerTypeScheduled, + Status: status, + TriggerData: []byte(`{}`), + } + if err := svc.RecordRun(c.Request.Context(), run); err != nil { + log.Error("e2e: failed to create automation run", zap.Error(err)) + c.JSON(http.StatusInternalServerError, gin.H{errKey: err.Error()}) + return + } + c.JSON(http.StatusCreated, gin.H{"id": run.ID, "automation_id": run.AutomationID, statusKey: run.Status}) + } +} diff --git a/apps/backend/internal/backendapp/services.go b/apps/backend/internal/backendapp/services.go index 81a0a80705..2a691eb2fb 100644 --- a/apps/backend/internal/backendapp/services.go +++ b/apps/backend/internal/backendapp/services.go @@ -120,6 +120,9 @@ func provideServices(cfg *config.Config, log *logger.Logger, repos *Repositories if automationErr != nil { log.Warn("Automation service initialization failed (non-fatal)", zap.Error(automationErr)) } + if automationComponents != nil { + automationComponents.Service.SetTaskDeleter(&automationTaskDeleterAdapter{svc: taskSvc}) + } return &Services{ Task: taskSvc, diff --git a/apps/backend/internal/backendapp/turn_adapters.go b/apps/backend/internal/backendapp/turn_adapters.go index 1d3c786cd4..6937257d12 100644 --- a/apps/backend/internal/backendapp/turn_adapters.go +++ b/apps/backend/internal/backendapp/turn_adapters.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" + "github.com/kandev/kandev/internal/automation" "github.com/kandev/kandev/internal/github" "github.com/kandev/kandev/internal/task/models" taskrepo "github.com/kandev/kandev/internal/task/repository/sqlite" @@ -123,6 +124,26 @@ func (a *taskDeleterAdapter) translateDeleteErr(err error) error { return err } +// automationTaskDeleterAdapter satisfies automation.TaskDeleter and +// translates the task repository's ErrTaskNotFound sentinel to +// automation.ErrTaskNotFound so the automation run-cleanup paths can +// classify the "already gone" case via errors.Is without importing the task +// repository's package. +type automationTaskDeleterAdapter struct { + svc *taskservice.Service +} + +func (a *automationTaskDeleterAdapter) DeleteTask(ctx context.Context, taskID string) error { + err := a.svc.DeleteTask(ctx, taskID) + if err == nil { + return nil + } + if errors.Is(err, taskrepo.ErrTaskNotFound) { + return fmt.Errorf("%w: %w", automation.ErrTaskNotFound, err) + } + return err +} + // repositoryLookupAdapter satisfies the linear/jira/sentry RepositoryLookup // interface over the task service. It is the validation seam for a watcher's // optional repository binding. The task service's GetRepository filters diff --git a/apps/backend/pkg/websocket/actions.go b/apps/backend/pkg/websocket/actions.go index 333e89ddc8..53134d7e93 100644 --- a/apps/backend/pkg/websocket/actions.go +++ b/apps/backend/pkg/websocket/actions.go @@ -553,6 +553,8 @@ const ( ActionAutomationTriggerDelete = "automation.trigger.delete" ActionAutomationTriggerTypes = "automation.trigger_types" ActionAutomationWebhookRevealSecret = "automation.webhook.reveal_secret" + ActionAutomationRunDelete = "automation.run.delete" + ActionAutomationRunsDeleteAll = "automation.runs.delete_all" ) // Error codes diff --git a/apps/web/components/automations/automation-editor.tsx b/apps/web/components/automations/automation-editor.tsx index 826069fcb8..ea27dea345 100644 --- a/apps/web/components/automations/automation-editor.tsx +++ b/apps/web/components/automations/automation-editor.tsx @@ -353,23 +353,7 @@ export function AutomationEditor({ workspaceId, automationId }: AutomationEditor return (
- Enter an automation name to enable saving. -
- )} -+ Enter an automation name to enable saving. +
+ )} +