integrated scheduling tab

resources/schema.graphql

@@ -1012,6 +1012,12 @@ type FlowAbortedResult {
 	message: String!
 }
 
+type FlowConfigSnapshotModified implements FlowEvent {
+	eventId: EventID!
+	eventTime: DateTime!
+	configSnapshot: FlowConfigurationSnapshot!
+}
+
 type FlowConfiguration {
 	ingest: FlowConfigurationIngest
 	compaction: FlowConfigurationCompaction

src/app/api/dataset-flow.api.ts

@@ -17,14 +17,19 @@ import {
     DatasetTriggerFlowMutation,
     FlowConfigurationInput,
     FlowRunConfiguration,
+    FlowTriggerInput,
     GetDatasetFlowConfigsGQL,
     GetDatasetFlowConfigsQuery,
+    GetDatasetFlowTriggersGQL,
+    GetDatasetFlowTriggersQuery,
     GetDatasetListFlowsGQL,
     GetDatasetListFlowsQuery,
     GetFlowByIdGQL,
     GetFlowByIdQuery,
     SetDatasetFlowConfigGQL,
     SetDatasetFlowConfigMutation,
+    SetDatasetFlowTriggersGQL,
+    SetDatasetFlowTriggersMutation,
 } from "./kamu.graphql.interface";
 import { Observable, first, map } from "rxjs";
 import { ApolloQueryResult } from "@apollo/client";
@@ -46,6 +51,8 @@ export class DatasetFlowApi {
     // private datasetFlowCompactionGQL = inject(DatasetFlowCompactionGQL);
     private datasetFlowsInitiatorsGQL = inject(DatasetFlowsInitiatorsGQL);
     private setDatasetFlowConfigGQL = inject(SetDatasetFlowConfigGQL);
+    private setDatasetFlowTriggersGQL = inject(SetDatasetFlowTriggersGQL);
+    private getDatasetFlowTriggersGQL = inject(GetDatasetFlowTriggersGQL);
 
     public datasetTriggerFlow(params: {
         accountId: string;
@@ -103,6 +110,43 @@ export class DatasetFlowApi {
             );
     }
 
+    public setDatasetFlowTriggers(params: {
+        datasetId: string;
+        datasetFlowType: DatasetFlowType;
+        paused: boolean;
+        triggerInput: FlowTriggerInput;
+    }): Observable<SetDatasetFlowTriggersMutation> {
+        return this.setDatasetFlowTriggersGQL.mutate(params).pipe(
+            first(),
+            map((result: MutationResult<SetDatasetFlowTriggersMutation>) => {
+                /* istanbul ignore else */
+                if (result.data) {
+                    return result.data;
+                } else {
+                    throw new DatasetOperationError(result.errors ?? []);
+                }
+            }),
+        );
+    }
+
+    public getDatasetFlowTriggers(params: {
+        datasetId: string;
+        datasetFlowType: DatasetFlowType;
+    }): Observable<GetDatasetFlowTriggersQuery> {
+        return this.getDatasetFlowTriggersGQL
+            .watch(params, {
+                ...noCacheFetchPolicy,
+                context: {
+                    skipLoading: true,
+                },
+            })
+            .valueChanges.pipe(
+                map((result: ApolloQueryResult<GetDatasetFlowTriggersQuery>) => {
+                    return result.data;
+                }),
+            );
+    }
+
     // public setDatasetFlowSchedule(params: {
     //     accountId: string;
     //     datasetId: string;

src/app/api/gql/flows-dataset/fragments/fragment-flow-history-data.graphql

@@ -5,6 +5,7 @@ fragment FlowHistoryData on FlowEvent {
   ... on FlowEventAborted {
     __typename
   }
+
   ... on FlowEventInitiated {
     trigger {
       __typename
@@ -90,4 +91,10 @@ fragment FlowHistoryData on FlowEvent {
       }
     }
   }
+
+  ... on FlowConfigSnapshotModified {
+    configSnapshot {
+      __typename
+    }
+  }
 }

src/app/api/gql/scheduling-dataset/dataset-flow-set-triggers.graphql

@@ -0,0 +1,39 @@
+mutation setDatasetFlowTriggers(
+  $datasetId: DatasetID!
+  $datasetFlowType: DatasetFlowType!
+  $paused: Boolean!
+  $triggerInput: FlowTriggerInput!
+) {
+  datasets {
+    byId(datasetId: $datasetId) {
+      flows {
+        triggers {
+          setTrigger(datasetFlowType: $datasetFlowType, paused: $paused, triggerInput: $triggerInput) {
+            ... on SetFlowTriggerSuccess {
+              message
+            }
+
+            ... on FlowIncompatibleDatasetKind {
+              message
+              expectedDatasetKind
+              actualDatasetKind
+            }
+
+            ... on FlowPreconditionsNotMet {
+              message
+            }
+
+            ... on FlowTypeIsNotSupported {
+              message
+            }
+
+            ... on FlowInvalidTriggerInputError {
+              message
+              reason
+            }
+          }
+        }
+      }
+    }
+  }
+}

src/app/api/gql/scheduling-dataset/dataset-flow-triggers.graphql

@@ -0,0 +1,29 @@
+query getDatasetFlowTriggers($datasetId: DatasetID!, $datasetFlowType: DatasetFlowType!) {
+  datasets {
+    byId(datasetId: $datasetId) {
+      flows {
+        triggers {
+          byType(datasetFlowType: $datasetFlowType) {
+            paused
+            schedule {
+              ... on TimeDelta {
+                ...TimeDeltaData
+              }
+
+              ... on Cron5ComponentExpression {
+                cron5ComponentExpression
+              }
+            }
+
+            batching {
+              maxBatchingInterval {
+                ...TimeDeltaData
+              }
+              minRecordsToAwait
+            }
+          }
+        }
+      }
+    }
+  }
+}