From 4ab024ce542f2f20d5d8c3a83e5eb3439f98d43f Mon Sep 17 00:00:00 2001
From: Kevin Biju <kevin@peerdb.io>
Date: Wed, 21 Feb 2024 03:08:05 +0530
Subject: [PATCH] making UI for edit mirror simpler

---
 flow/workflows/cdc_flow.go              | 22 ++++++++++++----------
 ui/app/mirrors/[mirrorId]/edit/page.tsx | 16 ++++++++++------
 ui/app/mirrors/[mirrorId]/page.tsx      | 10 ++++++++--
 ui/components/EditButton.tsx            |  8 +++++---
 4 files changed, 35 insertions(+), 21 deletions(-)

diff --git a/flow/workflows/cdc_flow.go b/flow/workflows/cdc_flow.go
index afc17927ce..40edabea89 100644
--- a/flow/workflows/cdc_flow.go
+++ b/flow/workflows/cdc_flow.go
@@ -45,7 +45,7 @@ type CDCFlowWorkflowState struct {
 	SrcTableIdNameMapping  map[uint32]string
 	TableNameSchemaMapping map[string]*protos.TableSchema
 	// flow config update request, set to nil after processed
-	FlowConfigUpdates []*protos.CDCFlowConfigUpdate
+	FlowConfigUpdate *protos.CDCFlowConfigUpdate
 	// options passed to all SyncFlows
 	SyncFlowOptions *protos.SyncFlowOptions
 	// initially copied from config, all changes are made here though
@@ -76,7 +76,7 @@ func NewCDCFlowWorkflowState(cfgTableMappings []*protos.TableMapping) *CDCFlowWo
 		CurrentFlowStatus:      protos.FlowStatus_STATUS_SETUP,
 		SrcTableIdNameMapping:  nil,
 		TableNameSchemaMapping: nil,
-		FlowConfigUpdates:      nil,
+		FlowConfigUpdate:       nil,
 		SyncFlowOptions:        nil,
 		TableMappings:          tableMappings,
 	}
@@ -143,13 +143,14 @@ func (w *CDCFlowWorkflowExecution) processCDCFlowConfigUpdates(ctx workflow.Cont
 	cfg *protos.FlowConnectionConfigs, state *CDCFlowWorkflowState,
 	mirrorNameSearch map[string]interface{},
 ) error {
-	for _, flowConfigUpdate := range state.FlowConfigUpdates {
+	flowConfigUpdate := state.FlowConfigUpdate
+	if flowConfigUpdate != nil {
 		if len(flowConfigUpdate.AdditionalTables) == 0 {
-			continue
+			return nil
 		}
 		if shared.AdditionalTablesHasOverlap(state.TableMappings, flowConfigUpdate.AdditionalTables) {
 			w.logger.Warn("duplicate source/destination tables found in additionalTables")
-			continue
+			return nil
 		}
 
 		alterPublicationAddAdditionalTablesCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
@@ -206,7 +207,7 @@ func (w *CDCFlowWorkflowExecution) processCDCFlowConfigUpdates(ctx workflow.Cont
 		state.TableMappings = append(state.TableMappings, flowConfigUpdate.AdditionalTables...)
 		state.SyncFlowOptions.TableMappings = state.TableMappings
 		// finished processing, wipe it
-		state.FlowConfigUpdates = nil
+		state.FlowConfigUpdate = nil
 	}
 	return nil
 }
@@ -444,9 +445,8 @@ func CDCFlowWorkflowWithConfig(
 		if cdcConfigUpdate.IdleTimeout > 0 {
 			state.SyncFlowOptions.IdleTimeoutSeconds = cdcConfigUpdate.IdleTimeout
 		}
-		if len(cdcConfigUpdate.AdditionalTables) > 0 {
-			state.FlowConfigUpdates = append(state.FlowConfigUpdates, cdcConfigUpdate)
-		}
+		// do this irrespective of additional tables being present, for auto unpausing
+		state.FlowConfigUpdate = cdcConfigUpdate
 
 		w.logger.Info("CDC Signal received. Parameters on signal reception:",
 			slog.Int("BatchSize", int(state.SyncFlowOptions.BatchSize)),
@@ -470,11 +470,13 @@ func CDCFlowWorkflowWithConfig(
 				w.logger.Info("mirror has been paused", slog.Any("duration", time.Since(startTime)))
 				// only place we block on receive, so signal processing is immediate
 				mainLoopSelector.Select(ctx)
-				if state.ActiveSignal == shared.NoopSignal {
+				if state.FlowConfigUpdate != nil {
 					err = w.processCDCFlowConfigUpdates(ctx, cfg, state, mirrorNameSearch)
 					if err != nil {
 						return state, err
 					}
+					// explicitly unpause
+					state.ActiveSignal = shared.NoopSignal
 				}
 			}
 
diff --git a/ui/app/mirrors/[mirrorId]/edit/page.tsx b/ui/app/mirrors/[mirrorId]/edit/page.tsx
index 1d9c6aa850..4cff1b4538 100644
--- a/ui/app/mirrors/[mirrorId]/edit/page.tsx
+++ b/ui/app/mirrors/[mirrorId]/edit/page.tsx
@@ -114,6 +114,10 @@ const EditMirror = ({ params: { mirrorId } }: EditMirrorProps) => {
     }
   };
 
+  const isNotPaused =
+    mirrorState.currentFlowState.toString() !==
+    FlowStatus[FlowStatus.STATUS_PAUSED];
+
   return (
     <div>
       <Label variant='title3'>Edit {mirrorId}</Label>
@@ -178,6 +182,11 @@ const EditMirror = ({ params: { mirrorId } }: EditMirrorProps) => {
         omitAdditionalTablesMapping={omitAdditionalTablesMapping}
       />
 
+      {isNotPaused ? (
+        <Label>Mirror can only be edited while paused.</Label>
+      ) : (
+        <Label>Editing mirror will automatically unpause it.</Label>
+      )}
       <div style={{ display: 'flex' }}>
         <Button
           style={{
@@ -187,12 +196,7 @@ const EditMirror = ({ params: { mirrorId } }: EditMirrorProps) => {
             height: '2.5rem',
           }}
           variant='normalSolid'
-          disabled={
-            loading ||
-            (additionalTables.length > 0 &&
-              mirrorState.currentFlowState.toString() !==
-                FlowStatus[FlowStatus.STATUS_PAUSED])
-          }
+          disabled={loading || isNotPaused}
           onClick={sendFlowStateChangeRequest}
         >
           {loading ? (
diff --git a/ui/app/mirrors/[mirrorId]/page.tsx b/ui/app/mirrors/[mirrorId]/page.tsx
index 786776d47b..cf171a15f9 100644
--- a/ui/app/mirrors/[mirrorId]/page.tsx
+++ b/ui/app/mirrors/[mirrorId]/page.tsx
@@ -2,7 +2,7 @@ import { SyncStatusRow } from '@/app/dto/MirrorsDTO';
 import prisma from '@/app/utils/prisma';
 import EditButton from '@/components/EditButton';
 import { ResyncDialog } from '@/components/ResyncDialog';
-import { FlowConnectionConfigs } from '@/grpc_generated/flow';
+import { FlowConnectionConfigs, FlowStatus } from '@/grpc_generated/flow';
 import { MirrorStatusResponse } from '@/grpc_generated/route';
 import { Header } from '@/lib/Header';
 import { LayoutMain } from '@/lib/Layout';
@@ -96,9 +96,15 @@ export default async function ViewMirror({
     syncStatusChild = (
       <SyncStatus rowsSynced={rowsSynced} rows={rows} flowJobName={mirrorId} />
     );
+    const isNotPaused =
+      mirrorStatus.currentFlowState.toString() !==
+      FlowStatus[FlowStatus.STATUS_PAUSED];
     editButtonHTML = (
       <div style={{ display: 'flex', alignItems: 'center' }}>
-        <EditButton toLink={`/mirrors/${mirrorId}/edit`} />
+        <EditButton
+          toLink={`/mirrors/${mirrorId}/edit`}
+          disabled={isNotPaused}
+        />
       </div>
     );
   } else {
diff --git a/ui/components/EditButton.tsx b/ui/components/EditButton.tsx
index 4476cdf217..2aed0a8ee9 100644
--- a/ui/components/EditButton.tsx
+++ b/ui/components/EditButton.tsx
@@ -1,11 +1,12 @@
 'use client';
+import { Button } from '@/lib/Button';
 import { Icon } from '@/lib/Icon';
 import { Label } from '@/lib/Label';
 import { ProgressCircle } from '@/lib/ProgressCircle';
 import { useRouter } from 'next/navigation';
 import { useState } from 'react';
 
-const EditButton = ({ toLink }: { toLink: string }) => {
+const EditButton = ({ toLink, disabled }: { toLink: string, disabled: boolean }) => {
   const [loading, setLoading] = useState(false);
   const router = useRouter();
 
@@ -14,7 +15,7 @@ const EditButton = ({ toLink }: { toLink: string }) => {
     router.push(toLink);
   };
   return (
-    <button
+    <Button
       className='IconButton'
       onClick={handleEdit}
       aria-label='sort up'
@@ -26,6 +27,7 @@ const EditButton = ({ toLink }: { toLink: string }) => {
         border: '1px solid rgba(0,0,0,0.1)',
         borderRadius: '0.5rem',
       }}
+      disabled={disabled}
     >
       <Label>Edit Mirror</Label>
       {loading ? (
@@ -33,7 +35,7 @@ const EditButton = ({ toLink }: { toLink: string }) => {
       ) : (
         <Icon name='edit' />
       )}
-    </button>
+    </Button>
   );
 };