refactor(core): make initConnectors check existing connectors in DB once (#451)

* refactor(core): make initConnectors check existing connectors in DB once

* refactor(core): remove unnecessary hasConnector

packages/core/src/connectors/index.test.ts

@@ -68,14 +68,12 @@ const findConnectorById = jest.fn(async (id: string) => {
 
   return connector;
 });
-const hasConnector = jest.fn(async () => true);
 const insertConnector = jest.fn(async (connector: Connector) => connector);
 
 jest.mock('@/queries/connector', () => ({
   ...jest.requireActual('@/queries/connector'),
   findAllConnectors: async () => findAllConnectors(),
   findConnectorById: async (id: string) => findConnectorById(id),
-  hasConnector: async () => hasConnector(),
   insertConnector: async (connector: Connector) => insertConnector(connector),
 }));
 
@@ -159,31 +157,23 @@ describe('getConnectorInstanceByType', () => {
 });
 
 describe('initConnectors', () => {
-  beforeEach(() => {
-    insertConnector.mockClear();
-    hasConnector.mockClear();
-  });
-
-  afterEach(() => {
-    insertConnector.mockClear();
-    hasConnector.mockClear();
-  });
-
   test('should insert the necessary connector if it does not exist in DB', async () => {
+    findAllConnectors.mockImplementationOnce(async () => []);
     await expect(initConnectors()).resolves.not.toThrow();
-    expect(hasConnector).toHaveBeenCalledTimes(connectors.length);
-    expect(insertConnector).not.toHaveBeenCalled();
-  });
-
-  test('should not insert the connector if it exists in DB', async () => {
-    hasConnector.mockImplementation(async () => false);
-
-    await expect(initConnectors()).resolves.not.toThrow();
-    expect(hasConnector).toHaveBeenCalledTimes(connectors.length);
+    expect(insertConnector).toHaveBeenCalledTimes(connectors.length);
 
     for (const [i, connector] of connectors.entries()) {
       const { id } = connector;
       expect(insertConnector).toHaveBeenNthCalledWith(i + 1, { id });
     }
   });
+
+  test('should not insert the connector if it exists in DB', async () => {
+    await expect(initConnectors()).resolves.not.toThrow();
+    expect(insertConnector).not.toHaveBeenCalled();
+  });
+
+  afterEach(() => {
+    insertConnector.mockClear();
+  });
 });

packages/core/src/connectors/index.ts

@@ -1,10 +1,5 @@
 import RequestError from '@/errors/RequestError';
-import {
-  findAllConnectors,
-  findConnectorById,
-  hasConnector,
-  insertConnector,
-} from '@/queries/connector';
+import { findAllConnectors, findConnectorById, insertConnector } from '@/queries/connector';
 
 import * as AliyunDM from './aliyun-dm';
 import * as AliyunSMS from './aliyun-sms';
@@ -99,15 +94,16 @@ export const getConnectorInstanceByType = async <T extends ConnectorInstance>(
 };
 
 export const initConnectors = async () => {
+  const connectors = await findAllConnectors();
+  const existingConnectorIds = new Set(connectors.map((connector) => connector.id));
+
   await Promise.all(
     allConnectors.map(async ({ metadata: { id } }) => {
-      if (await hasConnector(id)) {
+      if (existingConnectorIds.has(id)) {
         return;
       }
 
-      await insertConnector({
-        id,
-      });
+      await insertConnector({ id });
     })
   );
 };

packages/core/src/queries/connector.test.ts

@@ -7,7 +7,6 @@ import { expectSqlAssert, QueryType } from '@/utils/test-utils';
 import {
   findAllConnectors,
   findConnectorById,
-  hasConnector,
   insertConnector,
   updateConnector,
 } from './connector';
@@ -63,27 +62,6 @@ describe('connector queries', () => {
     await expect(findConnectorById(id)).resolves.toEqual(rowData);
   });
 
-  it('hasConnector', async () => {
-    const id = 'foo';
-
-    const expectSql = sql`
-      SELECT EXISTS(
-        select ${sql.join(Object.values(fields), sql`, `)}
-        from ${table}
-        where ${fields.id}=$1
-      )
-    `;
-
-    mockQuery.mockImplementationOnce(async (sql, values) => {
-      expectSqlAssert(sql, expectSql.sql);
-      expect(values).toEqual([id]);
-
-      return createMockQueryResult([{ exists: true }]);
-    });
-
-    await expect(hasConnector(id)).resolves.toEqual(true);
-  });
-
   it('insertConnector', async () => {
     const connector: CreateConnector & QueryResultRowType = {
       id: 'foo',

packages/core/src/queries/connector.ts

@@ -22,13 +22,6 @@ export const findConnectorById = async (id: string) =>
     where ${fields.id}=${id}
   `);
 
-export const hasConnector = async (id: string) =>
-  pool.exists(sql`
-    select ${sql.join(Object.values(fields), sql`, `)}
-    from ${table}
-    where ${fields.id}=${id}
-  `);
-
 export const insertConnector = buildInsertInto<CreateConnector, Connector>(pool, Connectors, {
   returning: true,
 });