drop mirror and more ui

PeerDB-io · Oct 31, 2023 · 3b1f4d6 · 3b1f4d6
1 parent 243ff7b
commit 3b1f4d6
Show file tree

Hide file tree

Showing 12 changed files with 337 additions and 86 deletions.
diff --git a/flow/generated/protos/route.pb.go b/flow/generated/protos/route.pb.go
diff --git a/flow/generated/protos/route.pb.gw.go b/flow/generated/protos/route.pb.gw.go
diff --git a/protos/route.proto b/protos/route.proto
@@ -209,7 +209,9 @@ service FlowService {
   rpc GetStatInfo(PostgresPeerActivityInfoRequest) returns (PeerStatResponse) {
     option (google.api.http) = { get: "/v1/peers/stats/{peer_name}" };
   }
-  rpc ShutdownFlow(ShutdownRequest) returns (ShutdownResponse) {}
+  rpc ShutdownFlow(ShutdownRequest) returns (ShutdownResponse) {
+    option (google.api.http) = { post: "/v1/mirrors/drop", body: "*" };
+  }
   rpc MirrorStatus(MirrorStatusRequest) returns (MirrorStatusResponse) {
     option (google.api.http) = { get: "/v1/mirrors/{flow_job_name}" };
   }

diff --git a/ui/app/api/mirrors/drop/route.ts b/ui/app/api/mirrors/drop/route.ts
@@ -0,0 +1,30 @@
+import { UDropMirrorResponse } from '@/app/dto/MirrorsDTO';
+import { ShutdownRequest, ShutdownResponse } from '@/grpc_generated/route';
+import { GetFlowHttpAddressFromEnv } from '@/rpc/http';
+
+export async function POST(request: Request) {
+  const body = await request.json();
+  const { workflowId, flowJobName, sourcePeer, destinationPeer } = body;
+  const flowServiceAddr = GetFlowHttpAddressFromEnv();
+  const req: ShutdownRequest = {
+    workflowId,
+    flowJobName,
+    sourcePeer,
+    destinationPeer,
+  };
+  console.log('/drop/mirror: req:', req);
+  const dropStatus: ShutdownResponse = await fetch(
+    `${flowServiceAddr}/v1/mirrors/drop`,
+    {
+      method: 'POST',
+      body: JSON.stringify(req),
+    }
+  ).then((res) => {
+    return res.json();
+  });
+  let response: UDropMirrorResponse = {
+    dropped: dropStatus.ok,
+  };
+
+  return new Response(JSON.stringify(response));
+}
diff --git a/ui/app/api/peers/route.ts b/ui/app/api/peers/route.ts
@@ -1,4 +1,5 @@
 import {
+  CatalogPeer,
   PeerConfig,
   UCreatePeerResponse,
   UValidatePeerResponse,
@@ -94,56 +95,58 @@ export async function POST(request: Request) {
   }
 }
 
+export const getTruePeer = (peer: CatalogPeer) => {
+  const newPeer: Peer = {
+    name: peer.name,
+    type: peer.type,
+  };
+  const options = peer.options;
+  let config:
+    | BigqueryConfig
+    | SnowflakeConfig
+    | PostgresConfig
+    | EventHubConfig
+    | S3Config
+    | SqlServerConfig
+    | EventHubGroupConfig;
+  switch (peer.type) {
+    case 0:
+      config = BigqueryConfig.decode(options);
+      newPeer.bigqueryConfig = config;
+      break;
+    case 1:
+      config = SnowflakeConfig.decode(options);
+      newPeer.snowflakeConfig = config;
+      break;
+    case 3:
+      config = PostgresConfig.decode(options);
+      newPeer.postgresConfig = config;
+      break;
+    case 4:
+      config = EventHubConfig.decode(options);
+      newPeer.eventhubConfig = config;
+      break;
+    case 5:
+      config = S3Config.decode(options);
+      newPeer.s3Config = config;
+      break;
+    case 6:
+      config = SqlServerConfig.decode(options);
+      newPeer.sqlserverConfig = config;
+      break;
+    case 7:
+      config = EventHubGroupConfig.decode(options);
+      newPeer.eventhubGroupConfig = config;
+      break;
+    default:
+      return newPeer;
+  }
+  return newPeer;
+};
+
 // GET all the peers from the database
 export async function GET(request: Request) {
   const peers = await prisma.peers.findMany();
-  const truePeers: Peer[] = peers.map((peer) => {
-    const newPeer: Peer = {
-      name: peer.name,
-      type: peer.type,
-    };
-    const options = peer.options;
-    let config:
-      | BigqueryConfig
-      | SnowflakeConfig
-      | PostgresConfig
-      | EventHubConfig
-      | S3Config
-      | SqlServerConfig
-      | EventHubGroupConfig;
-    switch (peer.type) {
-      case 0:
-        config = BigqueryConfig.decode(options);
-        newPeer.bigqueryConfig = config;
-        break;
-      case 1:
-        config = SnowflakeConfig.decode(options);
-        newPeer.snowflakeConfig = config;
-        break;
-      case 3:
-        config = PostgresConfig.decode(options);
-        newPeer.postgresConfig = config;
-        break;
-      case 4:
-        config = EventHubConfig.decode(options);
-        newPeer.eventhubConfig = config;
-        break;
-      case 5:
-        config = S3Config.decode(options);
-        newPeer.s3Config = config;
-        break;
-      case 6:
-        config = SqlServerConfig.decode(options);
-        newPeer.sqlserverConfig = config;
-        break;
-      case 7:
-        config = EventHubGroupConfig.decode(options);
-        newPeer.eventhubGroupConfig = config;
-        break;
-      default:
-        return newPeer;
-    }
-    return newPeer;
-  });
+  const truePeers: Peer[] = peers.map((peer) => getTruePeer(peer));
   return new Response(JSON.stringify(truePeers));
 }
diff --git a/ui/app/dto/MirrorsDTO.ts b/ui/app/dto/MirrorsDTO.ts
@@ -5,6 +5,10 @@ export type UCreateMirrorResponse = {
   created: boolean;
 };
 
+export type UDropMirrorResponse = {
+  dropped: boolean;
+};
+
 export type CDCConfig = FlowConnectionConfigs;
 export type MirrorConfig = CDCConfig | QRepConfig;
 export type MirrorSetter = Dispatch<SetStateAction<CDCConfig | QRepConfig>>;

diff --git a/ui/app/dto/PeersDTO.ts b/ui/app/dto/PeersDTO.ts
@@ -23,3 +23,9 @@ export type UColumnsResponse = {
 };
 
 export type PeerConfig = PostgresConfig | SnowflakeConfig;
+export type CatalogPeer = {
+  id: number;
+  name: string;
+  type: number;
+  options: Buffer;
+};
diff --git a/ui/app/mirrors/edit/[mirrorId]/cdc.tsx b/ui/app/mirrors/edit/[mirrorId]/cdc.tsx
@@ -195,6 +195,7 @@ export function CDCMirror({ cdc, syncStatusChild }: CDCMirrorStatusProps) {
       className='flex flex-col w-full'
       defaultValue={selectedTab}
       onValueChange={setSelectedTab}
+      style={{ marginTop: '2rem' }}
     >
       <Tabs.List className='flex border-b' aria-label='Details'>
         <Trigger