FAI-441 - Added revisions & users streams to Phabricator source + bump version

destinations/faros-destination/package.json

@@ -1,6 +1,6 @@
 {
   "name": "faros-destination",
-  "version": "0.1.21",
+  "version": "0.1.22",
   "private": true,
   "description": "Faros Destination for Airbyte",
   "keywords": [
@@ -30,7 +30,7 @@
     "watch": "tsc -b -w src test"
   },
   "dependencies": {
-    "faros-airbyte-cdk": "^0.1.21",
+    "faros-airbyte-cdk": "^0.1.22",
     "faros-feeds-sdk": "^0.8.30",
     "jsonata": "^1.8.5",
     "object-sizeof": "^1.6.1",

faros-airbyte-cdk/package.json

@@ -1,6 +1,6 @@
 {
   "name": "faros-airbyte-cdk",
-  "version": "0.1.21",
+  "version": "0.1.22",
   "description": "Airbyte Connector Development Kit (CDK) for JavaScript/TypeScript",
   "keywords": [
     "airbyte",

lerna.json

@@ -1,5 +1,5 @@
 {
-  "version": "0.1.21",
+  "version": "0.1.22",
   "packages": [
     "faros-airbyte-cdk",
     "destinations/**",

package.json

@@ -37,4 +37,4 @@
     "ts-node": "^10.2.1",
     "typescript": "^4.4.3"
   }
-}
+}

sources/example-source/package.json

@@ -1,6 +1,6 @@
 {
   "name": "example-source",
-  "version": "0.1.21",
+  "version": "0.1.22",
   "description": "Example Airbyte source",
   "keywords": [
     "airbyte",
@@ -31,7 +31,7 @@
   "dependencies": {
     "axios": "^0.21.4",
     "commander": "^8.2.0",
-    "faros-airbyte-cdk": "^0.1.21",
+    "faros-airbyte-cdk": "^0.1.22",
     "verror": "^1.10.0"
   },
   "jest": {

sources/jenkins-source/package.json

@@ -1,6 +1,6 @@
 {
   "name": "jenkins-source",
-  "version": "0.1.21",
+  "version": "0.1.22",
   "description": "Jenkins Airbyte source",
   "keywords": [
     "airbyte",
@@ -31,7 +31,7 @@
   },
   "dependencies": {
     "axios": "^0.21.4",
-    "faros-airbyte-cdk": "^0.1.21",
+    "faros-airbyte-cdk": "^0.1.22",
     "jenkins": "^0.28.1",
     "typescript-memoize": "^1.0.1",
     "verror": "^1.10.0"

sources/jenkins-source/src/index.ts

@@ -26,11 +26,13 @@ export class JenkinsSource extends AirbyteSourceBase {
   async checkConnection(
     config: JenkinsConfig
   ): Promise<[boolean, VError | undefined]> {
-    const [client, errorMessage] = await Jenkins.validateClient(config);
-    if (client) {
-      return [true, undefined];
+    try {
+      const jenkins = Jenkins.instance(config as JenkinsConfig, this.logger);
+      await jenkins.checkConnection();
+    } catch (err: any) {
+      return [false, err];
     }
-    return [false, new VError(errorMessage)];
+    return [true, undefined];
   }
   streams(config: JenkinsConfig): AirbyteStreamBase[] {
     return [new Builds(config, this.logger), new Jobs(config, this.logger)];

sources/jenkins-source/src/jenkins.ts

@@ -1,10 +1,11 @@
 import {AirbyteConfig, AirbyteLogger} from 'faros-airbyte-cdk';
-import jenkinsClient, {JenkinsPromisifiedAPI} from 'jenkins';
+import jenkinsClient from 'jenkins';
 import {Memoize} from 'typescript-memoize';
 import {URL} from 'url';
 import util from 'util';
+import {VError} from 'verror';
 
-export const DEFAULT_PAGE_SIZE = 10;
+const DEFAULT_PAGE_SIZE = 10;
 const FEED_ALL_FIELDS_PATTERN = `name,fullName,url,lastCompletedBuild[number],%s[id,displayName,number,building,result,timestamp,duration,url,actions[lastBuiltRevision[SHA1],remoteUrls],fullName,fullDisplayName],jobs[*]`;
 const FEED_JOBS_COUNT_PATTERN = 'jobs[name]';
 const FEED_MAX_DEPTH_CALC_PATTERN = 'fullName,jobs[*]';
@@ -51,16 +52,16 @@ export interface JenkinsState {
 
 export class Jenkins {
   constructor(
-    private readonly client: any,
+    private readonly client: any, // It should be 'JenkinsPromisifiedAPI' instead of any, but we could not make it work
     private readonly logger: AirbyteLogger
   ) {}
 
-  static parse(str: string, ...args: any[]): string {
+  private static parse(str: string, ...args: any[]): string {
     let i = 0;
     return str.replace(/%s/g, () => args[i++]);
   }
 
-  static validateInteger(
+  private static validateInteger(
     value: number
   ): [true | undefined, string | undefined] {
     if (value) {
@@ -72,44 +73,30 @@ export class Jenkins {
     return [true, undefined];
   }
 
-  static async make(
-    config: JenkinsConfig,
-    logger: AirbyteLogger
-  ): Promise<Jenkins | undefined> {
-    const [client, errorMessage] = await Jenkins.validateClient(config);
-    if (!client) {
-      logger.error(errorMessage || '');
-      return undefined;
-    }
-    return new Jenkins(client, logger);
-  }
-
-  static async validateClient(
-    config: JenkinsConfig
-  ): Promise<[JenkinsPromisifiedAPI | undefined, string | undefined]> {
+  static instance(config: JenkinsConfig, logger: AirbyteLogger): Jenkins {
     if (typeof config.server_url !== 'string') {
-      return [undefined, 'server_url: must be a string'];
+      throw new VError('server_url: must be a string');
     }
     if (typeof config.user !== 'string') {
-      return [undefined, 'User: must be a string'];
+      throw new VError('user: must be a string');
     }
     if (typeof config.token !== 'string') {
-      return [undefined, 'Token: must be a string'];
+      throw new VError('token: must be a string');
     }
     const depthCheck = Jenkins.validateInteger(config.depth);
     if (!depthCheck[0]) {
-      return [undefined, depthCheck[1]];
+      throw new VError(depthCheck[1]);
     }
     const pageSizeCheck = Jenkins.validateInteger(config.pageSize);
     if (!pageSizeCheck[0]) {
-      return [undefined, pageSizeCheck[1]];
+      throw new VError(pageSizeCheck[1]);
     }
 
-    let jenkinsUrl;
+    let jenkinsUrl: URL;
     try {
       jenkinsUrl = new URL(config.server_url);
     } catch (error) {
-      return [undefined, 'server_url: must be a valid url'];
+      throw new VError('server_url: must be a valid url');
     }
 
     jenkinsUrl.username = config.user;
@@ -121,15 +108,18 @@ export class Jenkins {
       promisify: true,
     });
 
+    return new Jenkins(client, logger);
+  }
+
+  async checkConnection(): Promise<void> {
     try {
-      await client.info();
-    } catch (error) {
-      return [
-        undefined,
-        'Please verify your server_url and user/token are correct',
-      ];
+      await this.client.info();
+    } catch (error: any) {
+      const err = error?.message ?? JSON.stringify(error);
+      throw new VError(
+        `Please verify your server_url and user/token are correct. Error: ${err}`
+      );
     }
-    return [client, undefined];
   }
 
   async *syncBuilds(

sources/jenkins-source/src/streams/builds.ts

@@ -6,13 +6,7 @@ import {
 } from 'faros-airbyte-cdk';
 import {Dictionary} from 'ts-essentials';
 
-import {
-  Build,
-  DEFAULT_PAGE_SIZE,
-  Jenkins,
-  JenkinsConfig,
-  JenkinsState,
-} from '../jenkins';
+import {Build, Jenkins, JenkinsConfig, JenkinsState} from '../jenkins';
 
 export class Builds extends AirbyteStreamBase {
   constructor(readonly config: JenkinsConfig, logger: AirbyteLogger) {
@@ -28,26 +22,17 @@ export class Builds extends AirbyteStreamBase {
   get cursorField(): string | string[] {
     return 'number';
   }
-  get stateCheckpointInterval(): number {
-    return 10 * (this.config.pageSize ?? DEFAULT_PAGE_SIZE);
-  }
-
   async *readRecords(
     syncMode: SyncMode,
     cursorField?: string[],
     streamSlice?: Build,
     streamState?: JenkinsState
   ): AsyncGenerator<Build, any, any> {
-    const jenkins = await Jenkins.make(this.config, this.logger);
-    if (!jenkins) return;
+    const jenkins = Jenkins.instance(this.config, this.logger);
+    const state =
+      syncMode === SyncMode.INCREMENTAL ? streamState ?? null : null;
 
-    let iter: AsyncGenerator<Build, any, unknown>;
-    if (syncMode === SyncMode.INCREMENTAL) {
-      iter = jenkins.syncBuilds(this.config, streamState ?? null);
-    } else {
-      iter = jenkins.syncBuilds(this.config, null);
-    }
-    yield* iter;
+    yield* jenkins.syncBuilds(this.config, state);
   }
 
   getUpdatedState(

sources/jenkins-source/src/streams/jobs.ts

@@ -6,7 +6,7 @@ import {
 } from 'faros-airbyte-cdk';
 import {Dictionary} from 'ts-essentials';
 
-import {DEFAULT_PAGE_SIZE, Jenkins, JenkinsConfig, Job} from '../jenkins';
+import {Jenkins, JenkinsConfig, Job} from '../jenkins';
 
 export class Jobs extends AirbyteStreamBase {
   constructor(readonly config: JenkinsConfig, logger: AirbyteLogger) {
@@ -22,9 +22,6 @@ export class Jobs extends AirbyteStreamBase {
   get cursorField(): string | string[] {
     return ['url'];
   }
-  get stateCheckpointInterval(): number {
-    return 10 * (this.config.pageSize ?? DEFAULT_PAGE_SIZE);
-  }
   async *streamSlices(
     syncMode: SyncMode,
     cursorField?: string[],
@@ -52,15 +49,11 @@ export class Jobs extends AirbyteStreamBase {
     streamSlice?: Job,
     streamState?: any
   ): AsyncGenerator<Job, any, any> {
-    const jenkins = await Jenkins.make(this.config, this.logger);
-    if (!jenkins) return;
+    const jenkins = Jenkins.instance(this.config, this.logger);
+    const state =
+      syncMode === SyncMode.INCREMENTAL ? streamSlice || null : null;
 
-    let jobs: Job[];
-    if (syncMode === SyncMode.INCREMENTAL) {
-      jobs = await jenkins.syncJobs(this.config, streamSlice || null);
-    } else {
-      jobs = await jenkins.syncJobs(this.config, null);
-    }
+    const jobs: Job[] = await jenkins.syncJobs(this.config, state);
     for (const job of jobs) {
       yield job;
     }

sources/jenkins-source/test/index.test.ts

@@ -27,7 +27,7 @@ describe('index', () => {
 
   test('spec', async () => {
     const source = new sut.JenkinsSource(logger);
-    expect(source.spec()).resolves.toStrictEqual(
+    await expect(source.spec()).resolves.toStrictEqual(
       new AirbyteSpec(readTestResourceFile('spec.json'))
     );
   });
@@ -40,28 +40,30 @@ describe('index', () => {
       info: jest.fn().mockRejectedValue({}),
     } as any);
     const source = new sut.JenkinsSource(logger);
-    expect(
+    await expect(
       source.checkConnection({
         user: '123',
         token: 'token',
         server_url: 'http://example.com',
       })
     ).resolves.toStrictEqual([true, undefined]);
-    expect(
+    await expect(
       source.checkConnection({
         user: '123',
         token: 'token',
         server_url: 'http://example.com',
       })
     ).resolves.toStrictEqual([
       false,
-      new VError('Please verify your server_url and user/token are correct'),
+      new VError(
+        'Please verify your server_url and user/token are correct. Error: {}'
+      ),
     ]);
-    expect(source.checkConnection({} as any)).resolves.toStrictEqual([
+    await expect(source.checkConnection({} as any)).resolves.toStrictEqual([
       false,
       new VError('server_url: must be a string'),
     ]);
-    expect(
+    await expect(
       source.checkConnection({server_url: '111', user: '', token: ''})
     ).resolves.toStrictEqual([
       false,
@@ -128,16 +130,20 @@ describe('index', () => {
     ]);
   });
 
-  test('streams - return undefined if config not correct', async () => {
+  test('streams - error out if config not correct', async () => {
     mocked(jenkinsClient).mockReturnValue({
       info: jest.fn().mockResolvedValue({}),
     } as any);
     const source = new sut.JenkinsSource(logger);
     const [jobStream, buildStream] = source.streams({} as any);
     const jobIter = jobStream.readRecords(SyncMode.FULL_REFRESH);
     const buildIter = buildStream.readRecords(SyncMode.FULL_REFRESH);
-    expect((await jobIter.next()).value).toBeUndefined();
-    expect((await buildIter.next()).value).toBeUndefined();
+    await expect(jobIter.next()).rejects.toStrictEqual(
+      new VError('server_url: must be a string')
+    );
+    await expect(buildIter.next()).rejects.toStrictEqual(
+      new VError('server_url: must be a string')
+    );
   });
 
   test('streams - builds, use incremental sync mode', async () => {