Kamu UI 499 flow config separation

CHANGELOG.md

@@ -12,6 +12,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   - `demo` mode: unimplemented features are disabled
   - `develop` mode: unimplemented features are available
 - Made a tile element clickable
+- Made `force update` link for a flow table
+### Changed
+- Flow configuration separation
 ### Fixed
 - Fixed error handling for the query explainer
 

resources/schema.graphql

@@ -80,18 +80,6 @@ type AccountEdge {
 	node: Account!
 }
 
-type AccountFlowConfigs {
-	"""
-	Checks if all configs of all datasets in account are disabled
-	"""
-	allPaused: Boolean!
-}
-
-type AccountFlowConfigsMut {
-	resumeAccountDatasetFlows: Boolean!
-	pauseAccountDatasetFlows: Boolean!
-}
-
 input AccountFlowFilters {
 	byFlowType: DatasetFlowType
 	byStatus: FlowStatus
@@ -104,19 +92,31 @@ type AccountFlowRuns {
 	listDatasetsWithFlow: DatasetConnection!
 }
 
+type AccountFlowTriggers {
+	"""
+	Checks if all triggers of all datasets in account are disabled
+	"""
+	allPaused: Boolean!
+}
+
+type AccountFlowTriggersMut {
+	resumeAccountDatasetFlows: Boolean!
+	pauseAccountDatasetFlows: Boolean!
+}
+
 type AccountFlows {
 	"""
 	Returns interface for flow runs queries
 	"""
 	runs: AccountFlowRuns!
 	"""
-	Returns interface for flow configurations queries
+	Returns interface for flow triggers queries
 	"""
-	configs: AccountFlowConfigs!
+	triggers: AccountFlowTriggers!
 }
 
 type AccountFlowsMut {
-	configs: AccountFlowConfigsMut!
+	triggers: AccountFlowTriggersMut!
 }
 
 scalar AccountID
@@ -200,6 +200,11 @@ type AuthMut {
 	revokeAccessToken(tokenId: AccessTokenID!): RevokeResult!
 }
 
+input BatchingInput {
+	minRecordsToAwait: Int!
+	maxBatchingInterval: TimeDeltaInput!
+}
+
 type BlockRef {
 	name: String!
 	blockHash: Multihash!
@@ -565,18 +570,10 @@ type DatasetFlowConfigs {
 	Returns defined configuration for a flow of specified type
 	"""
 	byType(datasetFlowType: DatasetFlowType!): FlowConfiguration
-	"""
-	Checks if all configs of this dataset are disabled
-	"""
-	allPaused: Boolean!
 }
 
 type DatasetFlowConfigsMut {
-	setConfigIngest(datasetFlowType: DatasetFlowType!, paused: Boolean!, ingest: IngestConditionInput!): SetFlowConfigResult!
-	setConfigTransform(datasetFlowType: DatasetFlowType!, paused: Boolean!, transform: TransformConditionInput!): SetFlowTransformConfigResult!
-	setConfigCompaction(datasetFlowType: DatasetFlowType!, compactionArgs: CompactionConditionInput!): SetFlowCompactionConfigResult!
-	pauseFlows(datasetFlowType: DatasetFlowType): Boolean!
-	resumeFlows(datasetFlowType: DatasetFlowType): Boolean!
+	setConfig(datasetFlowType: DatasetFlowType!, configInput: FlowConfigurationInput!): SetFlowConfigResult!
 }
 
 input DatasetFlowFilters {
@@ -596,6 +593,23 @@ type DatasetFlowRunsMut {
 	cancelScheduledTasks(flowId: FlowID!): CancelScheduledTasksResult!
 }
 
+type DatasetFlowTriggers {
+	"""
+	Returns defined trigger for a flow of specified type
+	"""
+	byType(datasetFlowType: DatasetFlowType!): FlowTrigger
+	"""
+	Checks if all triggers of this dataset are disabled
+	"""
+	allPaused: Boolean!
+}
+
+type DatasetFlowTriggersMut {
+	setTrigger(datasetFlowType: DatasetFlowType!, paused: Boolean!, triggerInput: FlowTriggerInput!): SetFlowTriggerResult!
+	pauseFlows(datasetFlowType: DatasetFlowType): Boolean!
+	resumeFlows(datasetFlowType: DatasetFlowType): Boolean!
+}
+
 enum DatasetFlowType {
 	INGEST
 	EXECUTE_TRANSFORM
@@ -609,6 +623,10 @@ type DatasetFlows {
 	"""
 	configs: DatasetFlowConfigs!
 	"""
+	Returns interface for flow triggers queries
+	"""
+	triggers: DatasetFlowTriggers!
+	"""
 	Returns interface for flow runs queries
 	"""
 	runs: DatasetFlowRuns!
@@ -617,6 +635,7 @@ type DatasetFlows {
 type DatasetFlowsMut {
 	configs: DatasetFlowConfigsMut!
 	runs: DatasetFlowRunsMut!
+	triggers: DatasetFlowTriggersMut!
 }
 
 scalar DatasetID
@@ -978,7 +997,7 @@ type Flow {
 	"""
 	Primary flow trigger
 	"""
-	primaryTrigger: FlowTrigger!
+	primaryTrigger: FlowTriggerType!
 	"""
 	Start condition
 	"""
@@ -993,10 +1012,14 @@ type FlowAbortedResult {
 	message: String!
 }
 
+type FlowConfigSnapshotModified implements FlowEvent {
+	eventId: EventID!
+	eventTime: DateTime!
+	configSnapshot: FlowConfigurationSnapshot!
+}
+
 type FlowConfiguration {
-	paused: Boolean!
 	ingest: FlowConfigurationIngest
-	transform: FlowConfigurationTransform
 	compaction: FlowConfigurationCompaction
 	reset: FlowConfigurationReset
 }
@@ -1008,10 +1031,14 @@ type FlowConfigurationCompactionRule {
 }
 
 type FlowConfigurationIngest {
-	schedule: FlowConfigurationSchedule!
 	fetchUncacheable: Boolean!
 }
 
+input FlowConfigurationInput @oneOf {
+	ingest: IngestConditionInput
+	compaction: CompactionConditionInput
+}
+
 type FlowConfigurationReset {
 	mode: SnapshotPropagationMode!
 	oldHeadHash: Multihash
@@ -1026,14 +1053,7 @@ input FlowConfigurationResetToSeedDummy {
 	dummy: String!
 }
 
-union FlowConfigurationSchedule = TimeDelta | Cron5ComponentExpression
-
-union FlowConfigurationSnapshot = FlowConfigurationTransform | FlowConfigurationCompactionRule | FlowConfigurationIngest | FlowConfigurationReset
-
-type FlowConfigurationTransform {
-	minRecordsToAwait: Int!
-	maxBatchingInterval: TimeDelta!
-}
+union FlowConfigurationSnapshot = FlowConfigurationCompactionRule | FlowConfigurationIngest | FlowConfigurationReset
 
 type FlowConnection {
 	"""
@@ -1133,7 +1153,7 @@ type FlowEventAborted implements FlowEvent {
 type FlowEventInitiated implements FlowEvent {
 	eventId: EventID!
 	eventTime: DateTime!
-	trigger: FlowTrigger!
+	trigger: FlowTriggerType!
 }
 
 type FlowEventScheduledForActivation implements FlowEvent {
@@ -1159,7 +1179,7 @@ type FlowEventTaskChanged implements FlowEvent {
 type FlowEventTriggerAdded implements FlowEvent {
 	eventId: EventID!
 	eventTime: DateTime!
-	trigger: FlowTrigger!
+	trigger: FlowTriggerType!
 }
 
 type FlowFailedError {
@@ -1179,13 +1199,13 @@ type FlowFailureReasonInputDatasetCompacted {
 
 scalar FlowID
 
-type FlowIncompatibleDatasetKind implements SetFlowConfigResult & SetFlowTransformConfigResult & SetFlowCompactionConfigResult & TriggerFlowResult {
+type FlowIncompatibleDatasetKind implements SetFlowConfigResult & TriggerFlowResult & SetFlowTriggerResult {
 	expectedDatasetKind: DatasetKind!
 	actualDatasetKind: DatasetKind!
 	message: String!
 }
 
-type FlowInvalidCompactionConfig implements SetFlowCompactionConfigResult {
+type FlowInvalidConfigInputError implements SetFlowConfigResult {
 	reason: String!
 	message: String!
 }
@@ -1195,7 +1215,7 @@ type FlowInvalidRunConfigurations implements TriggerFlowResult {
 	message: String!
 }
 
-type FlowInvalidTransformConfig implements SetFlowTransformConfigResult {
+type FlowInvalidTriggerInputError implements SetFlowTriggerResult {
 	reason: String!
 	message: String!
 }
@@ -1207,13 +1227,12 @@ type FlowNotFound implements GetFlowResult & CancelScheduledTasksResult {
 
 union FlowOutcome = FlowSuccessResult | FlowFailedError | FlowAbortedResult
 
-type FlowPreconditionsNotMet implements SetFlowConfigResult & SetFlowTransformConfigResult & TriggerFlowResult {
+type FlowPreconditionsNotMet implements SetFlowConfigResult & TriggerFlowResult & SetFlowTriggerResult {
 	preconditions: String!
 	message: String!
 }
 
 input FlowRunConfiguration @oneOf {
-	transform: TransformConditionInput
 	compaction: CompactionConditionInput
 	ingest: IngestConditionInput
 	reset: ResetConditionInput
@@ -1222,7 +1241,7 @@ input FlowRunConfiguration @oneOf {
 union FlowStartCondition = FlowStartConditionSchedule | FlowStartConditionThrottling | FlowStartConditionBatching | FlowStartConditionExecutor
 
 type FlowStartConditionBatching {
-	activeTransformRule: FlowConfigurationTransform!
+	activeBatchingRule: FlowTriggerBatchingRule!
 	batchingDeadline: DateTime!
 	accumulatedRecordsCount: Int!
 	watermarkModified: Boolean!
@@ -1268,12 +1287,26 @@ type FlowTimingRecords {
 	finishedAt: DateTime
 }
 
-union FlowTrigger = FlowTriggerManual | FlowTriggerAutoPolling | FlowTriggerPush | FlowTriggerInputDatasetFlow
+type FlowTrigger {
+	paused: Boolean!
+	schedule: FlowTriggerScheduleRule
+	batching: FlowTriggerBatchingRule
+}
 
 type FlowTriggerAutoPolling {
 	dummy: Boolean!
 }
 
+type FlowTriggerBatchingRule {
+	minRecordsToAwait: Int!
+	maxBatchingInterval: TimeDelta!
+}
+
+input FlowTriggerInput @oneOf {
+	schedule: ScheduleInput
+	batching: BatchingInput
+}
+
 type FlowTriggerInputDatasetFlow {
 	dataset: Dataset!
 	flowType: DatasetFlowType!
@@ -1288,7 +1321,11 @@ type FlowTriggerPush {
 	dummy: Boolean!
 }
 
-type FlowTypeIsNotSupported implements SetFlowConfigResult & SetFlowTransformConfigResult & SetFlowCompactionConfigResult {
+union FlowTriggerScheduleRule = TimeDelta | Cron5ComponentExpression
+
+union FlowTriggerType = FlowTriggerManual | FlowTriggerAutoPolling | FlowTriggerPush | FlowTriggerInputDatasetFlow
+
+type FlowTypeIsNotSupported implements SetFlowConfigResult & SetFlowTriggerResult {
 	message: String!
 }
 
@@ -1307,7 +1344,6 @@ input IngestConditionInput {
 	Flag indicates to ignore cache during ingest step for API calls
 	"""
 	fetchUncacheable: Boolean!
-	schedule: ScheduleInput!
 }
 
 input InitiatorFilterInput @oneOf {
@@ -1727,20 +1763,21 @@ type SetDataSchema {
 	schema: DataSchema!
 }
 
-interface SetFlowCompactionConfigResult {
+interface SetFlowConfigResult {
 	message: String!
 }
 
-interface SetFlowConfigResult {
+type SetFlowConfigSuccess implements SetFlowConfigResult {
+	config: FlowConfiguration!
 	message: String!
 }
 
-type SetFlowConfigSuccess implements SetFlowConfigResult & SetFlowTransformConfigResult & SetFlowCompactionConfigResult {
-	config: FlowConfiguration!
+interface SetFlowTriggerResult {
 	message: String!
 }
 
-interface SetFlowTransformConfigResult {
+type SetFlowTriggerSuccess implements SetFlowTriggerResult {
+	trigger: FlowTrigger!
 	message: String!
 }
 
@@ -1925,11 +1962,6 @@ enum TimeUnit {
 
 union Transform = TransformSql
 
-input TransformConditionInput {
-	minRecordsToAwait: Int!
-	maxBatchingInterval: TimeDeltaInput!
-}
-
 type TransformInput {
 	datasetRef: DatasetRef!
 	alias: String!
@@ -2030,4 +2062,4 @@ directive @specifiedBy(url: String!) on SCALAR
 schema {
 	query: Query
 	mutation: Mutation
-}
+}

src/app/api/account.api.spec.ts

@@ -145,8 +145,8 @@ describe("AccountApi", () => {
 
     it("should check account flows paused", () => {
         service.accountFlowsPaused(ACCOUNT_NAME).subscribe((state: AccountDatasetFlowsPausedQuery) => {
-            expect(state.accounts.byName?.flows?.configs.allPaused).toEqual(
-                mockAccountDatasetFlowsPausedQuery.accounts.byName?.flows?.configs.allPaused,
+            expect(state.accounts.byName?.flows?.triggers.allPaused).toEqual(
+                mockAccountDatasetFlowsPausedQuery.accounts.byName?.flows?.triggers.allPaused,
             );
         });
 
@@ -160,8 +160,8 @@ describe("AccountApi", () => {
 
     it("should check account pause flows", () => {
         service.accountPauseFlows(ACCOUNT_NAME).subscribe((state: AccountPauseFlowsMutation) => {
-            expect(state.accounts.byName?.flows.configs.pauseAccountDatasetFlows).toEqual(
-                mockAccountPauseFlowsMutationSuccess.accounts.byName?.flows.configs.pauseAccountDatasetFlows,
+            expect(state.accounts.byName?.flows.triggers.pauseAccountDatasetFlows).toEqual(
+                mockAccountPauseFlowsMutationSuccess.accounts.byName?.flows.triggers.pauseAccountDatasetFlows,
             );
         });
 
@@ -175,8 +175,8 @@ describe("AccountApi", () => {
 
     it("should check account resume flows", () => {
         service.accountResumeFlows(ACCOUNT_NAME).subscribe((state: AccountResumeFlowsMutation) => {
-            expect(state.accounts.byName?.flows.configs.resumeAccountDatasetFlows).toEqual(
-                mockAccountResumeFlowsMutationSuccess.accounts.byName?.flows.configs.resumeAccountDatasetFlows,
+            expect(state.accounts.byName?.flows.triggers.resumeAccountDatasetFlows).toEqual(
+                mockAccountResumeFlowsMutationSuccess.accounts.byName?.flows.triggers.resumeAccountDatasetFlows,
             );
         });