feat: Scaffold plugin server

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,19 +2,16 @@ 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 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);

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();

src/grpc/server.ts

@@ -0,0 +1,101 @@
+import { Readable, Writable } from 'node:stream';
+
+import { Logger } from 'winston';
+
+export type BackendOptions = {
+  tableName: string;
+  connection: string;
+};
+
+export type TableOptions = {
+  tables: string[];
+  skipTables: string[];
+  skipDependentTables: boolean;
+};
+
+export type SyncOptions = {
+  tables: string[];
+  skipTables: string[];
+  skipDependentTables: boolean;
+  deterministicCQId: boolean;
+  backendOptions: BackendOptions;
+  stream: Writable;
+};
+
+export type NewClientOptions = {
+  noConnection: boolean;
+};
+
+export type NewClientFunction = (logger: Logger, spec: string, options: NewClientOptions) => Promise<Client>;
+
+export interface SourceClient {
+  close: () => Promise<void>;
+  tables: (options: TableOptions) => Promise<string[]>;
+  sync: (options: SyncOptions) => void;
+}
+
+export interface DestinationClient {
+  close: () => Promise<void>;
+  read: (stream: Writable) => void;
+  write: (stream: Readable) => void;
+}
+
+export interface Client extends SourceClient, DestinationClient {}
+
+export interface Plugin {
+  name: () => string;
+  version: () => string;
+  write: (stream: Readable) => void;
+  read: (stream: Writable) => void;
+  setLogger: (logger: Logger) => void;
+  sync: (options: SyncOptions) => void;
+  tables: (options: TableOptions) => Promise<string[]>;
+  init: (spec: string, options: NewClientOptions) => Promise<void>;
+  close: () => Promise<void>;
+}
+
+export const newUnimplementedSourceClient = (): SourceClient => {
+  return {
+    close: () => Promise.reject(new Error('unimplemented')),
+    tables: () => Promise.reject(new Error('unimplemented')),
+    sync: () => Promise.reject(new Error('unimplemented')),
+  };
+};
+
+export const newUnimplementedDestinationClient = (): DestinationClient => {
+  return {
+    close: () => Promise.reject(new Error('unimplemented')),
+    read: () => Promise.reject(new Error('unimplemented')),
+    write: () => Promise.reject(new Error('unimplemented')),
+  };
+};
+
+export const newUnimplementedClient: NewClientFunction = (logger: Logger, spec: string, options: NewClientOptions) => {
+  return Promise.resolve({
+    ...newUnimplementedSourceClient(),
+    ...newUnimplementedDestinationClient(),
+  });
+};
+
+export const newPlugin = (name: string, version: string, newClient: NewClientFunction): Plugin => {
+  const plugin = {
+    client: undefined as Client | undefined,
+    logger: undefined as Logger | undefined,
+    name: () => name,
+    version: () => version,
+    write: (stream: Readable) => plugin.client?.write(stream) ?? new Error('client not initialized'),
+    read: (stream: Writable) => plugin.client?.read(stream) ?? new Error('client not initialized'),
+    setLogger: (logger: Logger) => {
+      plugin.logger = logger;
+    },
+    sync: (options: SyncOptions) => plugin.client?.sync(options) ?? new Error('client not initialized'),
+    tables: (options: TableOptions) =>
+      plugin.client?.tables(options) ?? Promise.reject(new Error('client not initialized')),
+    init: async (spec: string, options: NewClientOptions) => {
+      plugin.client = await newClient(plugin.logger!, spec, options);
+    },
+    close: () => plugin.client?.close() ?? Promise.reject(new Error('client not initialized')),
+  };
+
+  return plugin;
+};