Fix rest of ui error messages

Now to actually make it work
PeerDB-io · Jun 17, 2024 · 640396c · 640396c
1 parent 4685b84
commit 640396c
Show file tree

Hide file tree

Showing 9 changed files with 160 additions and 152 deletions.
diff --git a/flow/connectors/postgres/cdc.go b/flow/connectors/postgres/cdc.go
@@ -342,8 +342,7 @@ func PullCdcRecords[Items model.Items](
 	nextStandbyMessageDeadline := time.Now().Add(standbyMessageTimeout)
 
 	addRecordWithKey := func(key model.TableWithPkey, rec model.Record[Items]) error {
-		err := cdcRecordsStorage.Set(logger, key, rec)
-		if err != nil {
+		if err := cdcRecordsStorage.Set(logger, key, rec); err != nil {
 			return err
 		}
 		if err := records.AddRecord(ctx, rec); err != nil {

diff --git a/ui/app/api/peers/route.ts b/ui/app/api/peers/route.ts
@@ -1,6 +1,4 @@
-import { getTruePeer } from '@/app/api/peers/getTruePeer';
 import {
-  CatalogPeer,
   PeerConfig,
   UCreatePeerResponse,
   UValidatePeerResponse,
@@ -151,8 +149,9 @@ export async function POST(request: Request) {
 }
 
 // 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: CatalogPeer) => getTruePeer(peer));
-  return new Response(JSON.stringify(truePeers));
+export async function GET(_request: Request) {
+  const peers = await prisma.peers.findMany({
+    select: { name: true, type: true },
+  });
+  return new Response(JSON.stringify(peers));
 }
diff --git a/ui/app/dto/PeersDTO.ts b/ui/app/dto/PeersDTO.ts
@@ -1,6 +1,7 @@
 import {
   BigqueryConfig,
   ClickhouseConfig,
+  DBType,
   ElasticsearchConfig,
   EventHubConfig,
   EventHubGroupConfig,
@@ -60,6 +61,7 @@ export type CatalogPeer = {
   type: number;
   options: Buffer;
 };
+export type PeerRef = { name: string; type: DBType };
 export type PeerSetter = React.Dispatch<React.SetStateAction<PeerConfig>>;
 
 export type SlotLagPoint = {

diff --git a/ui/app/mirrors/create/cdc/cdc.tsx b/ui/app/mirrors/create/cdc/cdc.tsx
@@ -12,6 +12,8 @@ import TableMapping from './tablemapping';
 interface MirrorConfigProps {
   settings: MirrorSetting[];
   mirrorConfig: CDCConfig;
+  destinationType: DBType;
+  sourceType: DBType;
   setter: MirrorSetter;
   rows: TableMapRow[];
   setRows: Dispatch<SetStateAction<TableMapRow[]>>;
@@ -33,6 +35,8 @@ export const defaultSyncMode = (dtype: DBType | undefined) => {
 export default function CDCConfigForm({
   settings,
   mirrorConfig,
+  destinationType,
+  sourceType,
   setter,
   rows,
   setRows,
@@ -45,22 +49,24 @@ export default function CDCConfigForm({
     setting.stateHandler(stateVal, setter);
   };
 
-  const normalSettings = useMemo(() => {
-    return settings!.filter(
-      (setting) =>
-        !(
-          (IsQueuePeer(mirrorConfig.destination?.type) &&
-            setting.advanced === AdvancedSettingType.QUEUE) ||
-          setting.advanced === AdvancedSettingType.ALL
-        )
-    );
-  }, [settings, mirrorConfig.destination?.type]);
+  const normalSettings = useMemo(
+    () =>
+      settings!.filter(
+        (setting) =>
+          !(
+            (IsQueuePeer(destinationType) &&
+              setting.advanced === AdvancedSettingType.QUEUE) ||
+            setting.advanced === AdvancedSettingType.ALL
+          )
+      ),
+    [settings, destinationType]
+  );
 
   const advancedSettings = useMemo(() => {
     return settings!
       .map((setting) => {
         if (
-          IsQueuePeer(mirrorConfig.destination?.type) &&
+          IsQueuePeer(destinationType) &&
           setting.advanced === AdvancedSettingType.QUEUE
         ) {
           setting.stateHandler(600, setter);
@@ -71,30 +77,29 @@ export default function CDCConfigForm({
         }
       })
       .filter((setting) => setting !== undefined);
-  }, [settings, mirrorConfig.destination?.type, setter]);
+  }, [settings, destinationType, setter]);
 
   const paramDisplayCondition = (setting: MirrorSetting) => {
     const label = setting.label.toLowerCase();
-    const isQueue = IsQueuePeer(mirrorConfig.destination?.type);
+    const isQueue = IsQueuePeer(destinationType);
     if (
       (label.includes('snapshot') && mirrorConfig.doInitialSnapshot !== true) ||
       (label === 'replication slot name' &&
         mirrorConfig.doInitialSnapshot === true) ||
       (label.includes('staging path') &&
-        defaultSyncMode(mirrorConfig.destination?.type) !== 'AVRO') ||
+        defaultSyncMode(destinationType) !== 'AVRO') ||
       (isQueue && label.includes('soft delete')) ||
-      (mirrorConfig.destination?.type === DBType.EVENTHUBS &&
+      (destinationType === DBType.EVENTHUBS &&
         (label.includes('initial copy') ||
           label.includes('initial load') ||
           label.includes('snapshot'))) ||
-      ((mirrorConfig.source?.type !== DBType.POSTGRES ||
-        mirrorConfig.destination?.type !== DBType.POSTGRES) &&
+      ((sourceType !== DBType.POSTGRES ||
+        destinationType !== DBType.POSTGRES) &&
         label.includes('type system')) ||
-      (mirrorConfig.destination?.type !== DBType.BIGQUERY &&
-        label.includes('column name')) ||
+      (destinationType !== DBType.BIGQUERY && label.includes('column name')) ||
       (label.includes('soft delete') &&
         ![DBType.BIGQUERY, DBType.POSTGRES, DBType.SNOWFLAKE].includes(
-          mirrorConfig.destination?.type ?? DBType.UNRECOGNIZED
+          destinationType ?? DBType.UNRECOGNIZED
         ))
     ) {
       return false;
@@ -104,17 +109,17 @@ export default function CDCConfigForm({
 
   useEffect(() => {
     setPubLoading(true);
-    fetchPublications(mirrorConfig.source?.name || '').then((pubs) => {
+    fetchPublications(mirrorConfig.source || '').then((pubs) => {
       setPublications(pubs);
       setPubLoading(false);
     });
-  }, [mirrorConfig.source?.name]);
+  }, [mirrorConfig.source]);
 
   if (mirrorConfig.source != undefined && mirrorConfig.destination != undefined)
     return (
       <>
-        {normalSettings!.map((setting, id) => {
-          return (
+        {normalSettings!.map(
+          (setting, id) =>
             paramDisplayCondition(setting!) && (
               <CDCField
                 key={id}
@@ -128,8 +133,7 @@ export default function CDCConfigForm({
                 optionsLoading={pubLoading}
               />
             )
-          );
-        })}
+        )}
         <Button
           className='IconButton'
           aria-label='collapse'
@@ -150,7 +154,7 @@ export default function CDCConfigForm({
         </Button>
 
         {show &&
-          advancedSettings!.map((setting, id) => {
+          advancedSettings!.map((setting) => {
             return (
               paramDisplayCondition(setting!) && (
                 <CDCField
@@ -163,10 +167,10 @@ export default function CDCConfigForm({
           })}
 
         <TableMapping
-          sourcePeerName={mirrorConfig.source?.name}
+          sourcePeerName={mirrorConfig.source}
           rows={rows}
           setRows={setRows}
-          peerType={mirrorConfig.destination?.type}
+          peerType={destinationType}
           omitAdditionalTablesMapping={new Map<string, string[]>()}
         />
       </>

diff --git a/ui/app/mirrors/create/handlers.ts b/ui/app/mirrors/create/handlers.ts
@@ -14,9 +14,11 @@ import {
   QRepConfig,
   QRepWriteType,
 } from '@/grpc_generated/flow';
-import { DBType, Peer, dBTypeToJSON } from '@/grpc_generated/peers';
+import { DBType, dBTypeToJSON } from '@/grpc_generated/peers';
 import { Dispatch, SetStateAction } from 'react';
+
 import { CDCConfig, TableMapRow } from '../../dto/MirrorsDTO';
+
 import {
   cdcSchema,
   flowNameSchema,
@@ -33,31 +35,11 @@ export const IsQueuePeer = (peerType?: DBType): boolean => {
   );
 };
 
-export const handlePeer = (
-  peer: Peer | null,
-  peerEnd: 'src' | 'dst',
-  setConfig: (value: SetStateAction<CDCConfig | QRepConfig>) => void
-) => {
-  if (!peer) return;
-  if (peerEnd === 'dst') {
-    setConfig((curr) => ({
-      ...curr,
-      destination: peer,
-      destinationPeer: peer,
-    }));
-  } else {
-    setConfig((curr) => ({
-      ...curr,
-      source: peer,
-      sourcePeer: peer,
-    }));
-  }
-};
-
 const CDCCheck = (
   flowJobName: string,
   rows: TableMapRow[],
-  config: CDCConfig
+  config: CDCConfig,
+  destinationType: DBType
 ) => {
   const flowNameValid = flowNameSchema.safeParse(flowJobName);
   if (!flowNameValid.success) {
@@ -82,7 +64,7 @@ const CDCCheck = (
     config.replicationSlotName = '';
   }
 
-  if (IsQueuePeer(config.destination?.type)) {
+  if (IsQueuePeer(destinationType)) {
     config.softDelete = false;
   }
 
@@ -163,10 +145,11 @@ export const handleCreateCDC = async (
   flowJobName: string,
   rows: TableMapRow[],
   config: CDCConfig,
+  destinationType: DBType,
   setLoading: Dispatch<SetStateAction<boolean>>,
   route: RouteCallback
 ) => {
-  const err = CDCCheck(flowJobName, rows, config);
+  const err = CDCCheck(flowJobName, rows, config, destinationType);
   if (err != '') {
     notifyErr(err);
     return;
@@ -202,6 +185,7 @@ export const handleCreateQRep = async (
   flowJobName: string,
   query: string,
   config: QRepConfig,
+  destinationType: DBType,
   setLoading: Dispatch<SetStateAction<boolean>>,
   route: RouteCallback,
   xmin?: boolean
@@ -255,14 +239,12 @@ export const handleCreateQRep = async (
   config.query = query;
 
   const isSchemaLessPeer =
-    config.destinationPeer?.type === DBType.BIGQUERY ||
-    config.destinationPeer?.type === DBType.CLICKHOUSE;
-  if (config.destinationPeer?.type !== DBType.ELASTICSEARCH) {
+    destinationType === DBType.BIGQUERY ||
+    destinationType === DBType.CLICKHOUSE;
+  if (destinationType !== DBType.ELASTICSEARCH) {
     if (isSchemaLessPeer && config.destinationTableIdentifier?.includes('.')) {
       notifyErr(
-        'Destination table should not be schema qualified for ' +
-          DBTypeToGoodText(config.destinationPeer?.type) +
-          ' targets'
+        `Destination table should not be schema qualified for ${DBTypeToGoodText(destinationType)} targets`
       );
       return;
     }
@@ -271,9 +253,7 @@ export const handleCreateQRep = async (
       !config.destinationTableIdentifier?.includes('.')
     ) {
       notifyErr(
-        'Destination table should be schema qualified for ' +
-          DBTypeToGoodText(config.destinationPeer?.type) +
-          ' targets'
+        `Destination table should be schema qualified for ${DBTypeToGoodText(destinationType)} targets`
       );
       return;
     }
@@ -428,10 +408,11 @@ export const handleValidateCDC = async (
   flowJobName: string,
   rows: TableMapRow[],
   config: CDCConfig,
+  destinationType: DBType,
   setLoading: Dispatch<SetStateAction<boolean>>
 ) => {
   setLoading(true);
-  const err = CDCCheck(flowJobName, rows, config);
+  const err = CDCCheck(flowJobName, rows, config, destinationType);
   if (err != '') {
     notifyErr(err);
     setLoading(false);

diff --git a/ui/app/mirrors/create/helpers/common.ts b/ui/app/mirrors/create/helpers/common.ts
@@ -21,8 +21,8 @@ export interface MirrorSetting {
 }
 
 export const blankCDCSetting: CDCConfig = {
-  source: undefined,
-  destination: undefined,
+  source: '',
+  destination: '',
   flowJobName: '',
   tableMappings: [],
   maxBatchSize: 1000000,