feat: cascade delete related runs and outputs when delete workflow

internal/workflow/dispatcher/dispatcher.go

@@ -142,11 +142,11 @@ func (w *WorkflowDispatcher) Shutdown() {
 	w.workerMutex.Lock()
 	for _, worker := range w.workers {
 		worker.Cancel()
+		delete(w.workers, worker.Data.WorkflowId)
+		delete(w.workerIdMap, worker.Data.RunId)
 	}
 	w.workerMutex.Unlock()
 	w.wg.Wait()
-	w.workers = make(map[string]*workflowWorker)
-	w.workerIdMap = make(map[string]string)
 }
 
 func (w *WorkflowDispatcher) enqueueWorker(data *WorkflowWorkerData) {

migrations/1739263253_updated_workflow_run.go

@@ -0,0 +1,58 @@
+package migrations
+
+import (
+	"github.com/pocketbase/pocketbase/core"
+	m "github.com/pocketbase/pocketbase/migrations"
+)
+
+func init() {
+	m.Register(func(app core.App) error {
+		collection, err := app.FindCollectionByNameOrId("qjp8lygssgwyqyz")
+		if err != nil {
+			return err
+		}
+
+		// update field
+		if err := collection.Fields.AddMarshaledJSONAt(1, []byte(`{
+			"cascadeDelete": true,
+			"collectionId": "tovyif5ax6j62ur",
+			"hidden": false,
+			"id": "m8xfsyyy",
+			"maxSelect": 1,
+			"minSelect": 0,
+			"name": "workflowId",
+			"presentable": false,
+			"required": false,
+			"system": false,
+			"type": "relation"
+		}`)); err != nil {
+			return err
+		}
+
+		return app.Save(collection)
+	}, func(app core.App) error {
+		collection, err := app.FindCollectionByNameOrId("qjp8lygssgwyqyz")
+		if err != nil {
+			return err
+		}
+
+		// update field
+		if err := collection.Fields.AddMarshaledJSONAt(1, []byte(`{
+			"cascadeDelete": false,
+			"collectionId": "tovyif5ax6j62ur",
+			"hidden": false,
+			"id": "m8xfsyyy",
+			"maxSelect": 1,
+			"minSelect": 0,
+			"name": "workflowId",
+			"presentable": false,
+			"required": false,
+			"system": false,
+			"type": "relation"
+		}`)); err != nil {
+			return err
+		}
+
+		return app.Save(collection)
+	})
+}

migrations/1739263264_updated_workflow_output.go

@@ -0,0 +1,92 @@
+package migrations
+
+import (
+	"github.com/pocketbase/pocketbase/core"
+	m "github.com/pocketbase/pocketbase/migrations"
+)
+
+func init() {
+	m.Register(func(app core.App) error {
+		collection, err := app.FindCollectionByNameOrId("bqnxb95f2cooowp")
+		if err != nil {
+			return err
+		}
+
+		// update field
+		if err := collection.Fields.AddMarshaledJSONAt(1, []byte(`{
+			"cascadeDelete": true,
+			"collectionId": "tovyif5ax6j62ur",
+			"hidden": false,
+			"id": "jka88auc",
+			"maxSelect": 1,
+			"minSelect": 0,
+			"name": "workflowId",
+			"presentable": false,
+			"required": false,
+			"system": false,
+			"type": "relation"
+		}`)); err != nil {
+			return err
+		}
+
+		// update field
+		if err := collection.Fields.AddMarshaledJSONAt(2, []byte(`{
+			"cascadeDelete": true,
+			"collectionId": "qjp8lygssgwyqyz",
+			"hidden": false,
+			"id": "relation821863227",
+			"maxSelect": 1,
+			"minSelect": 0,
+			"name": "runId",
+			"presentable": false,
+			"required": false,
+			"system": false,
+			"type": "relation"
+		}`)); err != nil {
+			return err
+		}
+
+		return app.Save(collection)
+	}, func(app core.App) error {
+		collection, err := app.FindCollectionByNameOrId("bqnxb95f2cooowp")
+		if err != nil {
+			return err
+		}
+
+		// update field
+		if err := collection.Fields.AddMarshaledJSONAt(1, []byte(`{
+			"cascadeDelete": false,
+			"collectionId": "tovyif5ax6j62ur",
+			"hidden": false,
+			"id": "jka88auc",
+			"maxSelect": 1,
+			"minSelect": 0,
+			"name": "workflowId",
+			"presentable": false,
+			"required": false,
+			"system": false,
+			"type": "relation"
+		}`)); err != nil {
+			return err
+		}
+
+		// update field
+		if err := collection.Fields.AddMarshaledJSONAt(2, []byte(`{
+			"cascadeDelete": false,
+			"collectionId": "qjp8lygssgwyqyz",
+			"hidden": false,
+			"id": "relation821863227",
+			"maxSelect": 1,
+			"minSelect": 0,
+			"name": "runId",
+			"presentable": false,
+			"required": false,
+			"system": false,
+			"type": "relation"
+		}`)); err != nil {
+			return err
+		}
+
+		return app.Save(collection)
+	})
+}

ui/src/repository/_pocketbase.ts

@@ -6,3 +6,11 @@ export const getPocketBase = () => {
   pb = new PocketBase("/");
   return pb;
 };
+
+export const COLLECTION_NAME_ADMIN = "_superusers";
+export const COLLECTION_NAME_ACCESS = "access";
+export const COLLECTION_NAME_CERTIFICATE = "certificate";
+export const COLLECTION_NAME_SETTINGS = "settings";
+export const COLLECTION_NAME_WORKFLOW = "workflow";
+export const COLLECTION_NAME_WORKFLOW_RUN = "workflow_run";
+export const COLLECTION_NAME_WORKFLOW_OUTPUT = "workflow_output";

ui/src/repository/access.ts

@@ -1,12 +1,10 @@
 import dayjs from "dayjs";
 
 import { type AccessModel } from "@/domain/access";
-import { getPocketBase } from "./_pocketbase";
+import { COLLECTION_NAME_ACCESS, getPocketBase } from "./_pocketbase";
-
-const COLLECTION_NAME = "access";
 
 export const list = async () => {
-  return await getPocketBase().collection(COLLECTION_NAME).getFullList<AccessModel>({
+  return await getPocketBase().collection(COLLECTION_NAME_ACCESS).getFullList<AccessModel>({
     filter: "deleted=null",
     sort: "-created",
     requestKey: null,
@@ -15,15 +13,15 @@ export const list = async () => {
 
 export const save = async (record: MaybeModelRecord<AccessModel>) => {
   if (record.id) {
-    return await getPocketBase().collection(COLLECTION_NAME).update<AccessModel>(record.id, record);
+    return await getPocketBase().collection(COLLECTION_NAME_ACCESS).update<AccessModel>(record.id, record);
   }
 
-  return await getPocketBase().collection(COLLECTION_NAME).create<AccessModel>(record);
+  return await getPocketBase().collection(COLLECTION_NAME_ACCESS).create<AccessModel>(record);
 };
 
 export const remove = async (record: MaybeModelRecordWithId<AccessModel>) => {
   await getPocketBase()
-    .collection(COLLECTION_NAME)
+    .collection(COLLECTION_NAME_ACCESS)
     .update<AccessModel>(record.id!, { deleted: dayjs.utc().format("YYYY-MM-DD HH:mm:ss") });
   return true;
 };

ui/src/repository/admin.ts

@@ -1,9 +1,7 @@
-import { getPocketBase } from "./_pocketbase";
+import { COLLECTION_NAME_ADMIN, getPocketBase } from "./_pocketbase";
-
-const COLLECTION_NAME = "_superusers";
 
 export const authWithPassword = (username: string, password: string) => {
-  return getPocketBase().collection(COLLECTION_NAME).authWithPassword(username, password);
+  return getPocketBase().collection(COLLECTION_NAME_ADMIN).authWithPassword(username, password);
 };
 
 export const getAuthStore = () => {
@@ -12,6 +10,6 @@ export const getAuthStore = () => {
 
 export const save = (data: { email: string } | { password: string; passwordConfirm: string }) => {
   return getPocketBase()
-    .collection(COLLECTION_NAME)
+    .collection(COLLECTION_NAME_ADMIN)
     .update(getAuthStore().record?.id || "", data);
 };

ui/src/repository/certificate.ts

@@ -2,9 +2,7 @@ import dayjs from "dayjs";
 import { type RecordListOptions } from "pocketbase";
 
 import { type CertificateModel } from "@/domain/certificate";
-import { getPocketBase } from "./_pocketbase";
+import { COLLECTION_NAME_CERTIFICATE, getPocketBase } from "./_pocketbase";
-
-const COLLECTION_NAME = "certificate";
 
 export type ListCertificateRequest = {
   page?: number;
@@ -35,7 +33,7 @@ export const list = async (request: ListCertificateRequest) => {
     });
   }
 
-  return pb.collection(COLLECTION_NAME).getList<CertificateModel>(page, perPage, options);
+  return pb.collection(COLLECTION_NAME_CERTIFICATE).getList<CertificateModel>(page, perPage, options);
 };
 
 export const listByWorkflowRunId = async (workflowRunId: string) => {
@@ -48,7 +46,7 @@ export const listByWorkflowRunId = async (workflowRunId: string) => {
     sort: "-created",
     requestKey: null,
   };
-  const items = await pb.collection(COLLECTION_NAME).getFullList<CertificateModel>(options);
+  const items = await pb.collection(COLLECTION_NAME_CERTIFICATE).getFullList<CertificateModel>(options);
   return {
     totalItems: items.length,
     items: items,
@@ -57,7 +55,7 @@ export const listByWorkflowRunId = async (workflowRunId: string) => {
 
 export const remove = async (record: MaybeModelRecordWithId<CertificateModel>) => {
   await getPocketBase()
-    .collection(COLLECTION_NAME)
+    .collection(COLLECTION_NAME_CERTIFICATE)
     .update<CertificateModel>(record.id!, { deleted: dayjs.utc().format("YYYY-MM-DD HH:mm:ss") });
   return true;
 };

ui/src/repository/settings.ts

@@ -1,13 +1,11 @@
 import { ClientResponseError } from "pocketbase";
 
 import { type SettingsModel, type SettingsNames } from "@/domain/settings";
-import { getPocketBase } from "./_pocketbase";
+import { COLLECTION_NAME_SETTINGS, getPocketBase } from "./_pocketbase";
-
-const COLLECTION_NAME = "settings";
 
 export const get = async <T extends NonNullable<unknown>>(name: SettingsNames) => {
   try {
-    const resp = await getPocketBase().collection(COLLECTION_NAME).getFirstListItem<SettingsModel<T>>(`name='${name}'`, {
+    const resp = await getPocketBase().collection(COLLECTION_NAME_SETTINGS).getFirstListItem<SettingsModel<T>>(`name='${name}'`, {
       requestKey: null,
     });
     return resp;
@@ -25,8 +23,8 @@ export const get = async <T extends NonNullable<unknown>>(name: SettingsNames) =
 
 export const save = async <T extends NonNullable<unknown>>(record: MaybeModelRecordWithId<SettingsModel<T>>) => {
   if (record.id) {
-    return await getPocketBase().collection(COLLECTION_NAME).update<SettingsModel<T>>(record.id, record);
+    return await getPocketBase().collection(COLLECTION_NAME_SETTINGS).update<SettingsModel<T>>(record.id, record);
   }
 
-  return await getPocketBase().collection(COLLECTION_NAME).create<SettingsModel<T>>(record);
+  return await getPocketBase().collection(COLLECTION_NAME_SETTINGS).create<SettingsModel<T>>(record);
 };

ui/src/repository/workflow.ts

@@ -1,9 +1,7 @@
 import { type RecordListOptions, type RecordSubscription } from "pocketbase";
 
 import { type WorkflowModel } from "@/domain/workflow";
-import { getPocketBase } from "./_pocketbase";
+import { COLLECTION_NAME_WORKFLOW, getPocketBase } from "./_pocketbase";
-
-const COLLECTION_NAME = "workflow";
 
 export type ListWorkflowRequest = {
   page?: number;
@@ -26,11 +24,11 @@ export const list = async (request: ListWorkflowRequest) => {
     options.filter = pb.filter("enabled={:enabled}", { enabled: request.enabled });
   }
 
-  return await pb.collection(COLLECTION_NAME).getList<WorkflowModel>(page, perPage, options);
+  return await pb.collection(COLLECTION_NAME_WORKFLOW).getList<WorkflowModel>(page, perPage, options);
 };
 
 export const get = async (id: string) => {
-  return await getPocketBase().collection(COLLECTION_NAME).getOne<WorkflowModel>(id, {
+  return await getPocketBase().collection(COLLECTION_NAME_WORKFLOW).getOne<WorkflowModel>(id, {
     requestKey: null,
   });
 };
@@ -38,21 +36,21 @@ export const get = async (id: string) => {
 export const save = async (record: MaybeModelRecord<WorkflowModel>) => {
   if (record.id) {
     return await getPocketBase()
-      .collection(COLLECTION_NAME)
+      .collection(COLLECTION_NAME_WORKFLOW)
       .update<WorkflowModel>(record.id as string, record);
   }
 
-  return await getPocketBase().collection(COLLECTION_NAME).create<WorkflowModel>(record);
+  return await getPocketBase().collection(COLLECTION_NAME_WORKFLOW).create<WorkflowModel>(record);
 };
 
 export const remove = async (record: MaybeModelRecordWithId<WorkflowModel>) => {
-  return await getPocketBase().collection(COLLECTION_NAME).delete(record.id);
+  return await getPocketBase().collection(COLLECTION_NAME_WORKFLOW).delete(record.id);
 };
 
 export const subscribe = async (id: string, cb: (e: RecordSubscription<WorkflowModel>) => void) => {
-  return getPocketBase().collection(COLLECTION_NAME).subscribe(id, cb);
+  return getPocketBase().collection(COLLECTION_NAME_WORKFLOW).subscribe(id, cb);
 };
 
 export const unsubscribe = async (id: string) => {
-  return getPocketBase().collection(COLLECTION_NAME).unsubscribe(id);
+  return getPocketBase().collection(COLLECTION_NAME_WORKFLOW).unsubscribe(id);
 };

ui/src/repository/workflowRun.ts

@@ -2,9 +2,7 @@
 
 import { type WorkflowRunModel } from "@/domain/workflowRun";
 
-import { getPocketBase } from "./_pocketbase";
+import { COLLECTION_NAME_WORKFLOW_RUN, getPocketBase } from "./_pocketbase";
-
-const COLLECTION_NAME = "workflow_run";
 
 export type ListWorkflowRunsRequest = {
   workflowId?: string;
@@ -25,7 +23,7 @@ export const list = async (request: ListWorkflowRunsRequest) => {
   }
 
   return await getPocketBase()
-    .collection(COLLECTION_NAME)
+    .collection(COLLECTION_NAME_WORKFLOW_RUN)
     .getList<WorkflowRunModel>(page, perPage, {
       filter: getPocketBase().filter(filter, params),
       sort: "-created",
@@ -35,13 +33,13 @@ export const list = async (request: ListWorkflowRunsRequest) => {
 };
 
 export const remove = async (record: MaybeModelRecordWithId<WorkflowRunModel>) => {
-  return await getPocketBase().collection(COLLECTION_NAME).delete(record.id);
+  return await getPocketBase().collection(COLLECTION_NAME_WORKFLOW_RUN).delete(record.id);
 };
 
 export const subscribe = async (id: string, cb: (e: RecordSubscription<WorkflowRunModel>) => void) => {
-  return getPocketBase().collection(COLLECTION_NAME).subscribe(id, cb);
+  return getPocketBase().collection(COLLECTION_NAME_WORKFLOW_RUN).subscribe(id, cb);
 };
 
 export const unsubscribe = async (id: string) => {
-  return getPocketBase().collection(COLLECTION_NAME).unsubscribe(id);
+  return getPocketBase().collection(COLLECTION_NAME_WORKFLOW_RUN).unsubscribe(id);
 };