s3 peer ui v0

flow/connectors/core.go

@@ -230,8 +230,12 @@ func GetConnector(ctx context.Context, peer *protos.Peer) (Connector, error) {
 			return nil, fmt.Errorf("missing sqlserver config for %s peer %s", peer.Type.String(), peer.Name)
 		}
 		return connsqlserver.NewSQLServerConnector(ctx, sqlServerConfig)
-	// case protos.DBType_S3:
-	// 	return conns3.NewS3Connector(ctx, config.GetS3Config())
+	case protos.DBType_S3:
+		s3Config := peer.GetS3Config()
+		if s3Config == nil {
+			return nil, fmt.Errorf("missing s3 config for %s peer %s", peer.Type.String(), peer.Name)
+		}
+		return conns3.NewS3Connector(ctx, s3Config)
 	// case protos.DBType_EVENTHUB:
 	// 	return connsqlserver.NewSQLServerConnector(ctx, config.GetSqlserverConfig())
 	default:

ui/app/api/peers/route.ts

@@ -52,6 +52,12 @@ const constructPeer = (
         type: DBType.BIGQUERY,
         bigqueryConfig: config as BigqueryConfig,
       };
+    case 'S3':
+      return {
+        name,
+        type: DBType.S3,
+        s3Config: config as S3Config,
+      };
     default:
       return;
   }

ui/app/mirrors/create/helpers/common.ts

@@ -59,7 +59,7 @@ export const blankQRepSetting = {
   waitBetweenBatchesSeconds: 30,
   writeMode: undefined,
   stagingPath: '',
-  numRowsPerPartition: 0,
+  numRowsPerPartition: 100000,
   setupWatermarkTableOnDestination: false,
   dstTableFullResync: false,
 };

ui/app/mirrors/create/page.tsx

@@ -63,7 +63,6 @@ export default function CreateMirrors() {
   const [config, setConfig] = useState<CDCConfig | QRepConfig>(blankCDCSetting);
   const [peers, setPeers] = useState<Peer[]>([]);
   const [rows, setRows] = useState<TableMapRow[]>([]);
-  const [validSource, setValidSource] = useState<boolean>(false);
   const [sourceSchema, setSourceSchema] = useState('public');
   const [qrepQuery, setQrepQuery] =
     useState<string>(`-- Here's a sample template:

ui/app/peers/create/[peerType]/handlers.ts

@@ -3,6 +3,7 @@ import {
   UCreatePeerResponse,
   UValidatePeerResponse,
 } from '@/app/dto/PeersDTO';
+import { S3Config } from '@/grpc_generated/peers';
 import { Dispatch, SetStateAction } from 'react';
 import { bqSchema, pgSchema, s3Schema, sfSchema } from './schema';
 
@@ -17,6 +18,15 @@ const validateFields = (
     setMessage({ ok: false, msg: 'Peer name is required' });
     return false;
   }
+
+  if (type === 'S3') {
+    const s3Valid = S3Validation(config as S3Config);
+    if (s3Valid.length > 0) {
+      setMessage({ ok: false, msg: s3Valid });
+      return false;
+    }
+  }
+
   let validationErr: string | undefined;
   switch (type) {
     case 'POSTGRES':
@@ -74,6 +84,13 @@ export const handleValidate = async (
   setLoading(false);
 };
 
+const S3Validation = (config: S3Config): string => {
+  if (!config.secretAccessKey && !config.accessKeyId && !config.roleArn) {
+    return 'Either both access key and secret or role ARN is required';
+  }
+  return '';
+};
+
 // API call to create peer
 export const handleCreate = async (
   type: string,

ui/app/peers/create/[peerType]/helpers/common.ts

@@ -2,6 +2,7 @@ import { PeerConfig } from '@/app/dto/PeersDTO';
 import { PeerSetter } from '@/components/ConfigForm';
 import { blankBigquerySetting } from './bq';
 import { blankPostgresSetting } from './pg';
+import { blankS3Setting } from './s3';
 import { blankSnowflakeSetting } from './sf';
 
 export interface PeerSetting {
@@ -22,6 +23,8 @@ export const getBlankSetting = (dbType: string): PeerConfig => {
       return blankSnowflakeSetting;
     case 'BIGQUERY':
       return blankBigquerySetting;
+    case 'S3':
+      return blankS3Setting;
     default:
       return blankPostgresSetting;
   }

ui/app/peers/create/[peerType]/helpers/s3.ts

@@ -6,54 +6,60 @@ export const s3Setting: PeerSetting[] = [
     label: 'Bucket URL',
     stateHandler: (value, setter) =>
       setter((curr) => ({ ...curr, url: value })),
-    tips: 'The URL of the S3/GCS bucket. It begins with s3://',
+    tips: 'The URL of your existing S3/GCS bucket along with a prefix of your choice. It begins with s3://',
+    helpfulLink:
+      'https://docs.aws.amazon.com/AmazonS3/latest/userguide/access-bucket-intro.html#accessing-a-bucket-using-S3-format',
+    default: 's3://<bucket_name>/<prefix_name>',
   },
   {
     label: 'Access Key ID',
     stateHandler: (value, setter) =>
-      setter((curr) => ({ ...curr, accessKeyID: parseInt(value, 10) })),
-    tips: 'Specifies the TCP/IP port or local Unix domain socket file extension on which postgres is listening for connections from client applications.',
+      setter((curr) => ({ ...curr, accessKeyId: value })),
+    optional: true,
+    tips: 'The AWS access key ID associated with your account. In case of GCS, this is the HMAC access key ID.',
+    helpfulLink:
+      'https://docs.aws.amazon.com/IAM/latest/UserGuide/id_credentials_access-keys.html',
   },
   {
     label: 'Secret Access Key',
     stateHandler: (value, setter) =>
       setter((curr) => ({ ...curr, secretAccessKey: value })),
-    tips: 'Specify the user that we should use to connect to this host.',
-    helpfulLink: 'https://www.postgresql.org/docs/8.0/user-manag.html',
-  },
-  {
-    label: 'Role ARN',
-    stateHandler: (value, setter) =>
-      setter((curr) => ({ ...curr, roleArn: value })),
-    type: 'password',
-    tips: 'Password associated with the user you provided.',
-    helpfulLink: 'https://www.postgresql.org/docs/current/auth-password.html',
+    tips: 'The AWS secret access key associated with your account. In case of GCS, this is the HMAC secret.',
+    helpfulLink:
+      'https://docs.aws.amazon.com/IAM/latest/UserGuide/id_credentials_access-keys.html',
     optional: true,
   },
   {
     label: 'Region',
     stateHandler: (value, setter) =>
       setter((curr) => ({ ...curr, region: value })),
-    tips: 'Specify which database to associate with this peer.',
-    helpfulLink:
-      'https://www.postgresql.org/docs/current/sql-createdatabase.html',
-    optional: true,
+    tips: 'The region where your bucket is located. For example, us-east-1. In case of GCS, this will be set to auto, which detects where your bucket it.',
   },
   {
-    label: 'Endpoint',
+    label: 'Role ARN',
     stateHandler: (value, setter) =>
-      setter((curr) => ({ ...curr, endpoint: value })),
+      setter((curr) => ({ ...curr, roleArn: value })),
+    type: 'password',
+    tips: 'You may set this instead of the access key ID and secret.',
+    helpfulLink:
+      'https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_identifiers.html#identifiers-arns',
     optional: true,
-    tips: '',
   },
 ];
 
-export const blankS3Config: S3Config = {
-  url: '',
+export const blankS3Setting: S3Config = {
+  url: 's3://<bucket_name>/<prefix_name>',
   accessKeyId: undefined,
   secretAccessKey: undefined,
   roleArn: undefined,
   region: undefined,
-  endpoint: undefined,
-  metadataDb: undefined,
+  endpoint: '',
+  metadataDb: {
+    host: '',
+    port: 5432,
+    user: 'postgres',
+    password: '',
+    database: 'postgres',
+    transactionSnapshot: '',
+  },
 };

ui/app/peers/create/[peerType]/page.tsx

@@ -5,7 +5,7 @@ import S3ConfigForm from '@/components/S3Form';
 import { Button } from '@/lib/Button';
 import { ButtonGroup } from '@/lib/ButtonGroup';
 import { Label } from '@/lib/Label';
-import { LayoutMain, RowWithTextField } from '@/lib/Layout';
+import { RowWithTextField } from '@/lib/Layout';
 import { Panel } from '@/lib/Panel';
 import { TextField } from '@/lib/TextField';
 import { Tooltip } from '@/lib/Tooltip';
@@ -58,7 +58,15 @@ export default function CreateConfig({
   };
 
   return (
-    <LayoutMain alignSelf='center' justifySelf='center' width='xxLarge'>
+    <div
+      style={{
+        display: 'flex',
+        flexDirection: 'column',
+        alignSelf: 'center',
+        justifySelf: 'center',
+        width: '45%',
+      }}
+    >
       <Panel>
         <Label variant='title3'>
           Setup a new{' '}
@@ -147,6 +155,6 @@ export default function CreateConfig({
           )}
         </Panel>
       </Panel>
-    </LayoutMain>
+    </div>
   );
 }

ui/app/peers/create/[peerType]/schema.ts

@@ -193,8 +193,8 @@ export const s3Schema = z.object({
       required_error: 'URL is required',
     })
     .min(1, { message: 'URL must be non-empty' })
-    .refine((url) => url.startsWith('s3:/'), {
-      message: 'URL must start with s3:/',
+    .refine((url) => url.startsWith('s3://'), {
+      message: 'URL must start with s3://',
     }),
   accessKeyId: z
     .string({