Basic pause/unpause ui for qrep

flow/workflows/qrep_flow.go

@@ -46,7 +46,6 @@ func newQRepFlowState() *protos.QRepFlowState {
 	}
 }
 
-// newQRepFlowExecution creates a new instance of QRepFlowExecution.
 func newQRepFlowExecution(ctx workflow.Context, config *protos.QRepConfig, runUUID string) *QRepFlowExecution {
 	return &QRepFlowExecution{
 		config:          config,
@@ -57,7 +56,6 @@ func newQRepFlowExecution(ctx workflow.Context, config *protos.QRepConfig, runUU
 	}
 }
 
-// NewQRepFlowExecution creates a new instance of QRepFlowExecution.
 func newQRepPartitionFlowExecution(ctx workflow.Context,
 	config *protos.QRepConfig, runUUID string,
 ) *QRepPartitionFlowExecution {
@@ -451,14 +449,6 @@ func setWorkflowQueries(ctx workflow.Context, state *protos.QRepFlowState) error
 		return fmt.Errorf("failed to set `%s` query handler: %w", shared.FlowStatusQuery, err)
 	}
 
-	// Support an Update for the current status of the qrep flow.
-	err = workflow.SetUpdateHandler(ctx, shared.FlowStatusUpdate, func(status *protos.FlowStatus) error {
-		state.CurrentFlowStatus = *status
-		return nil
-	})
-	if err != nil {
-		return fmt.Errorf("failed to register query handler: %w", err)
-	}
 	return nil
 }
 

ui/app/api/mirrors/state_change/route.ts

@@ -12,9 +12,7 @@ export async function POST(request: Request) {
         method: 'POST',
         body: JSON.stringify(body),
       }
-    ).then((res) => {
-      return res.json();
-    });
+    ).then((res) => res.json());
 
     return new Response(JSON.stringify(res));
   } catch (e) {

ui/app/mirrors/[mirrorId]/handlers.ts

@@ -14,9 +14,8 @@ export const getMirrorState = async (mirrorId: string) => {
 };
 
 export const getCurrentIdleTimeout = async (mirrorId: string) => {
-  return await getMirrorState(mirrorId).then((res) => {
-    return (res as MirrorStatusResponse).cdcStatus?.config?.idleTimeoutSeconds;
-  });
+  const res = await getMirrorState(mirrorId);
+  return (res as MirrorStatusResponse).cdcStatus?.config?.idleTimeoutSeconds;
 };
 
 export const changeFlowState = async (

ui/app/mirrors/[mirrorId]/page.tsx

@@ -7,9 +7,11 @@ import { MirrorStatusResponse } from '@/grpc_generated/route';
 import { Header } from '@/lib/Header';
 import { LayoutMain } from '@/lib/Layout';
 import { GetFlowHttpAddressFromEnv } from '@/rpc/http';
-import { redirect } from 'next/navigation';
 import { CDCMirror } from './cdc';
 import NoMirror from './nomirror';
+import QrepGraph from './qrepGraph';
+import QRepStatusButtons from './qrepStatusButtons';
+import QRepStatusTable, { QRepPartitionStatus } from './qrepStatusTable';
 import SyncStatus from './syncStatus';
 
 type EditMirrorProps = {
@@ -38,7 +40,7 @@ export default async function ViewMirror({
     return <div>No mirror status found!</div>;
   }
 
-  let mirrorInfo = await prisma.flows.findFirst({
+  const mirrorInfo = await prisma.flows.findFirst({
     select: {
       created_at: true,
       workflow_id: true,
@@ -49,7 +51,7 @@ export default async function ViewMirror({
     },
   });
 
-  let syncs = await prisma.cdc_batches.findMany({
+  const syncs = await prisma.cdc_batches.findMany({
     where: {
       flow_name: mirrorId,
       start_time: {
@@ -113,29 +115,77 @@ export default async function ViewMirror({
         isNotPaused={isNotPaused}
       />
     );
+
+    return (
+      <LayoutMain alignSelf='flex-start' justifySelf='flex-start' width='full'>
+        <div
+          style={{
+            display: 'flex',
+            alignItems: 'center',
+            justifyContent: 'space-between',
+            paddingRight: '2rem',
+          }}
+        >
+          <Header variant='title2'>{mirrorId}</Header>
+          {actionsDropdown}
+        </div>
+        <CDCMirror
+          rows={rows}
+          createdAt={mirrorInfo?.created_at}
+          syncStatusChild={syncStatusChild}
+          status={mirrorStatus}
+        />
+      </LayoutMain>
+    );
   } else {
-    redirect(`/mirrors/status/qrep/${mirrorId}`);
-  }
+    const runs = await prisma.qrep_partitions.findMany({
+      where: {
+        flow_name: mirrorId,
+        start_time: {
+          not: null,
+        },
+        rows_in_partition: {
+          not: 0,
+        },
+      },
+      orderBy: {
+        start_time: 'desc',
+      },
+    });
 
-  return (
-    <LayoutMain alignSelf='flex-start' justifySelf='flex-start' width='full'>
-      <div
-        style={{
-          display: 'flex',
-          alignItems: 'center',
-          justifyContent: 'space-between',
-          paddingRight: '2rem',
-        }}
-      >
-        <Header variant='title2'>{mirrorId}</Header>
-        {actionsDropdown}
-      </div>
-      <CDCMirror
-        rows={rows}
-        createdAt={mirrorInfo?.created_at}
-        syncStatusChild={syncStatusChild}
-        status={mirrorStatus}
-      />
-    </LayoutMain>
-  );
+    const partitions: QRepPartitionStatus[] = runs.map((run) => ({
+      partitionId: run.partition_uuid,
+      startTime: run.start_time,
+      endTime: run.end_time,
+      pulledRows: run.rows_in_partition,
+      syncedRows: Number(run.rows_synced),
+    }));
+
+    return (
+      <LayoutMain alignSelf='flex-start' justifySelf='flex-start' width='full'>
+        <div
+          style={{
+            display: 'flex',
+            alignItems: 'center',
+            justifyContent: 'space-between',
+            paddingRight: '2rem',
+          }}
+        >
+          <Header variant='title2'>{mirrorId}</Header>
+          <QRepStatusButtons mirrorId={mirrorId} />
+        </div>
+        <div>Status: {mirrorStatus.currentFlowState}</div>
+        <QrepGraph
+          syncs={partitions.map((partition) => ({
+            partitionID: partition.partitionId,
+            startTime: partition.startTime,
+            endTime: partition.endTime,
+            numRows: partition.pulledRows,
+          }))}
+        />
+        <br></br>
+        <QRepStatusTable flowJobName={mirrorId} partitions={partitions} />
+      </LayoutMain>
+    );
+  }
 }

ui/app/mirrors/status/qrep/[mirrorId]/qrepGraph.tsx → ui/app/mirrors/[mirrorId]/qrepGraph.tsx

@@ -9,7 +9,7 @@ import { Label } from '@/lib/Label';
 import { BarChart } from '@tremor/react';
 import { useEffect, useState } from 'react';
 import ReactSelect from 'react-select';
-import aggregateCountsByInterval from '../../../[mirrorId]/aggregatedCountsByInterval';
+import aggregateCountsByInterval from './aggregatedCountsByInterval';
 
 type QrepStatusRow = {
   partitionID: string;

ui/app/mirrors/[mirrorId]/qrepStatusButtons.tsx

@@ -0,0 +1,34 @@
+'use client';
+import { FlowStatus } from '@/grpc_generated/flow';
+
+async function setFlowState(
+  flowJobName: string,
+  requestedFlowState: FlowStatus
+) {
+  await fetch(`/api/mirrors/state_change`, {
+    method: 'POST',
+    body: JSON.stringify({
+      flowJobName,
+      requestedFlowState,
+    }),
+    cache: 'no-store',
+  });
+  window.location.reload();
+}
+
+export default function qrepStatusButtons(props: { mirrorId: string }) {
+  return (
+    <>
+      <input
+        type='button'
+        value='Pause'
+        onClick={() => setFlowState(props.mirrorId, FlowStatus.STATUS_PAUSED)}
+      />
+      <input
+        type='button'
+        value='Resume'
+        onClick={() => setFlowState(props.mirrorId, FlowStatus.STATUS_RUNNING)}
+      />
+    </>
+  );
+}