feat: Scaffold plugin server (#27)

Author: erezrokah

.eslintrc

@@ -1,7 +1,7 @@
 {
   "root": true,
   "parser": "@typescript-eslint/parser",
-  "plugins": ["@typescript-eslint", "prettier", "unicorn"],
+  "plugins": ["@typescript-eslint", "prettier", "unicorn", "unused-imports"],
   "parserOptions": {
     "project": "./tsconfig.json"
   },
@@ -19,6 +19,7 @@
     "plugin:you-dont-need-lodash-underscore/all"
   ],
   "rules": {
+    "unused-imports/no-unused-imports": "error",
     "no-console": "error",
     "@typescript-eslint/no-unused-vars": 0,
     "require-await": "off",

package-lock.json

@@ -59,6 +59,7 @@
     "eslint-plugin-prettier": "^5.0.0",
     "eslint-plugin-promise": "^6.1.1",
     "eslint-plugin-unicorn": "^48.0.1",
+    "eslint-plugin-unused-imports": "^3.0.0",
     "eslint-plugin-you-dont-need-lodash-underscore": "^6.12.0",
     "prettier": "^3.0.1",
     "ts-node": "^10.9.1",
@@ -78,7 +79,7 @@
   },
   "dependencies": {
     "@apache-arrow/esnext-esm": "^12.0.1",
-    "@cloudquery/plugin-pb-javascript": "^0.0.6",
+    "@cloudquery/plugin-pb-javascript": "^0.0.7",
     "boolean": "^3.2.0",
     "winston": "^3.10.0",
     "yargs": "^17.7.2"

package.json

@@ -1,6 +1,5 @@
 import { discovery1 } from '@cloudquery/plugin-pb-javascript';
 import grpc = require('@grpc/grpc-js');
-import winston from 'winston';
 
 const SUPPORTED_VERSIONS = [3];
 

src/grpc/discovery.ts

@@ -1,16 +1,25 @@
 import { pluginV3 } from '@cloudquery/plugin-pb-javascript';
 import grpc = require('@grpc/grpc-js');
-import winston from 'winston';
+
+import { Plugin } from '../plugin/plugin.js';
 
 export class PluginServer extends pluginV3.cloudquery.plugin.v3.UnimplementedPluginService {
+  // Needed due to some TypeScript nonsense
+  private plugin: Plugin & grpc.UntypedHandleCall;
+
+  constructor(plugin: Plugin) {
+    super();
+    this.plugin = plugin as Plugin & grpc.UntypedHandleCall;
+  }
+
   GetName(
     call: grpc.ServerUnaryCall<
       pluginV3.cloudquery.plugin.v3.GetName.Request,
       pluginV3.cloudquery.plugin.v3.GetName.Response
     >,
     callback: grpc.sendUnaryData<pluginV3.cloudquery.plugin.v3.GetName.Response>,
   ): void {
-    throw new Error('Method not implemented.');
+    return callback(null, new pluginV3.cloudquery.plugin.v3.GetName.Response({ name: this.plugin.name() }));
   }
   GetVersion(
     call: grpc.ServerUnaryCall<
@@ -19,13 +28,27 @@ export class PluginServer extends pluginV3.cloudquery.plugin.v3.UnimplementedPlu
     >,
     callback: grpc.sendUnaryData<pluginV3.cloudquery.plugin.v3.GetVersion.Response>,
   ): void {
-    throw new Error('Method not implemented.');
+    return callback(null, new pluginV3.cloudquery.plugin.v3.GetVersion.Response({ version: this.plugin.version() }));
   }
   Init(
     call: grpc.ServerUnaryCall<pluginV3.cloudquery.plugin.v3.Init.Request, pluginV3.cloudquery.plugin.v3.Init.Response>,
     callback: grpc.sendUnaryData<pluginV3.cloudquery.plugin.v3.Init.Response>,
   ): void {
-    throw new Error('Method not implemented.');
+    const {
+      request: { spec, no_connection: noConnection },
+    } = call;
+
+    const stringSpec = new TextDecoder().decode(spec);
+    this.plugin
+      .init(stringSpec, { noConnection })
+      .then(() => {
+        // eslint-disable-next-line promise/no-callback-in-promise
+        return callback(null, new pluginV3.cloudquery.plugin.v3.Init.Response());
+      })
+      .catch((error) => {
+        // eslint-disable-next-line promise/no-callback-in-promise
+        return callback(error, null);
+      });
   }
   GetTables(
     call: grpc.ServerUnaryCall<
@@ -34,23 +57,54 @@ export class PluginServer extends pluginV3.cloudquery.plugin.v3.UnimplementedPlu
     >,
     callback: grpc.sendUnaryData<pluginV3.cloudquery.plugin.v3.GetTables.Response>,
   ): void {
-    throw new Error('Method not implemented.');
+    const {
+      request: { tables, skip_tables: skipTables, skip_dependent_tables: skipDependentTables },
+    } = call;
+
+    this.plugin
+      .tables({ tables, skipTables, skipDependentTables })
+      .then((tables) => {
+        const encodedTables = tables.map((table) => new TextEncoder().encode(table));
+        // eslint-disable-next-line promise/no-callback-in-promise
+        return callback(null, new pluginV3.cloudquery.plugin.v3.GetTables.Response({ tables: encodedTables }));
+      })
+      .catch((error) => {
+        // eslint-disable-next-line promise/no-callback-in-promise
+        return callback(error, null);
+      });
   }
   Sync(
     call: grpc.ServerWritableStream<
       pluginV3.cloudquery.plugin.v3.Sync.Request,
       pluginV3.cloudquery.plugin.v3.Sync.Response
     >,
   ): void {
-    throw new Error('Method not implemented.');
+    const {
+      request: {
+        tables,
+        skip_tables: skipTables,
+        skip_dependent_tables: skipDependentTables,
+        deterministic_cq_id: deterministicCQId,
+        backend: { connection, table_name: tableName },
+      },
+    } = call;
+
+    this.plugin.sync({
+      tables,
+      skipTables,
+      skipDependentTables,
+      deterministicCQId,
+      backendOptions: { connection, tableName },
+      stream: call,
+    });
   }
   Read(
     call: grpc.ServerWritableStream<
       pluginV3.cloudquery.plugin.v3.Read.Request,
       pluginV3.cloudquery.plugin.v3.Read.Response
     >,
   ): void {
-    throw new Error('Method not implemented.');
+    this.plugin.read(call);
   }
   Write(
     call: grpc.ServerReadableStream<
@@ -59,7 +113,8 @@ export class PluginServer extends pluginV3.cloudquery.plugin.v3.UnimplementedPlu
     >,
     callback: grpc.sendUnaryData<pluginV3.cloudquery.plugin.v3.Write.Response>,
   ): void {
-    throw new Error('Method not implemented.');
+    this.plugin.write(call);
+    callback(null, new pluginV3.cloudquery.plugin.v3.Write.Response());
   }
   Close(
     call: grpc.ServerUnaryCall<
@@ -68,6 +123,15 @@ export class PluginServer extends pluginV3.cloudquery.plugin.v3.UnimplementedPlu
     >,
     callback: grpc.sendUnaryData<pluginV3.cloudquery.plugin.v3.Close.Response>,
   ): void {
-    throw new Error('Method not implemented.');
+    this.plugin
+      .close()
+      .then(() => {
+        // eslint-disable-next-line promise/no-callback-in-promise
+        return callback(null, new pluginV3.cloudquery.plugin.v3.Close.Response());
+      })
+      .catch((error) => {
+        // eslint-disable-next-line promise/no-callback-in-promise
+        return callback(error, null);
+      });
   }
 }

src/grpc/plugin.ts

@@ -2,29 +2,26 @@ import { pluginV3, discovery1 } from '@cloudquery/plugin-pb-javascript';
 import grpc = require('@grpc/grpc-js');
 import winston from 'winston';
 
+import { Plugin } from '../plugin/plugin.js';
+
 import { DiscoveryServer } from './discovery.js';
 import { PluginServer } from './plugin.js';
 
 export enum Network {
   TCP = 'tcp',
 }
 
-export const getServer = () => {
+export const startServer = (logger: winston.Logger, address: string, plugin: Plugin) => {
   const server = new grpc.Server();
-  server.addService(pluginV3.cloudquery.plugin.v3.UnimplementedPluginService.definition, new PluginServer());
+  server.addService(pluginV3.cloudquery.plugin.v3.UnimplementedPluginService.definition, new PluginServer(plugin));
   server.addService(discovery1.cloudquery.discovery.v1.UnimplementedDiscoveryService.definition, new DiscoveryServer());
 
-  return server;
-};
-
-export const startServer = (logger: winston.Logger, address: string) => {
-  const server = getServer();
   server.bindAsync(address, grpc.ServerCredentials.createInsecure(), (error, port) => {
     if (error) {
       logger.error(error);
       return;
     }
     server.start();
-    logger.info('server running on port', port);
+    logger.info(`server running on port: ${port}`);
   });
 };

src/grpc/server.test.ts

@@ -1,3 +1,4 @@
-import { serve } from './serve/serve.js';
+import { newPlugin, newUnimplementedClient } from './plugin/plugin.js';
+import { createServeCommand } from './plugin/serve.js';
 
-serve.parse();
+createServeCommand(newPlugin('test', 'v1.0.0', newUnimplementedClient)).parse();