implement watchService#watchRepo

Makefile

@@ -107,9 +107,22 @@ setup-test-data:
 	  -H 'Content-Type: application/json' \
 	  -d @test/data/project4.json \
 	  http://localhost:36462/api/v1/projects
+	# ----------
+	# project5
+	curl -X POST \
+	  -H 'Authorization: Bearer anonymous' \
+	  -H 'Content-Type: application/json' \
+	  -d @test/data/project5.json \
+	  http://localhost:36462/api/v1/projects
+	# project5 - repo1
+	curl -X POST \
+	  -H 'Authorization: Bearer anonymous' \
+	  -H 'Content-Type: application/json' \
+	  -d @test/data/project5_repo1.json \
+	  http://localhost:36462/api/v1/projects/project5/repos
 
-.PHONY: update-test-data
-update-test-data:
+.PHONY: update-test-data-for-watchFile
+update-test-data-for-watchFile:
 	curl -X PUT \
 	  -H 'Authorization: Bearer anonymous' \
 	  -H 'Content-Type: application/json' \
@@ -128,6 +141,33 @@ update-test-data:
 	  -d @test/data/project2_repo2_content2_update3.json \
 	  http://localhost:36462/api/v0/projects/project2/repositories/repo2/files/revisions/head
 
+.PHONY: update-test-data-for-watchRepo
+update-test-data-for-watchRepo:
+	# project5 - repo1 - content1
+	curl -X POST \
+	  -H 'Authorization: Bearer anonymous' \
+	  -H 'Content-Type: application/json' \
+	  -d @test/data/project5_repo1_content1.json \
+  	  http://localhost:36462/api/v0/projects/project5/repositories/repo1/files/revisions/head
+	sleep 3
+	curl -X PUT \
+	  -H 'Authorization: Bearer anonymous' \
+	  -H 'Content-Type: application/json' \
+	  -d @test/data/project5_repo1_content1_update1.json \
+	  http://localhost:36462/api/v0/projects/project5/repositories/repo1/files/revisions/head
+	sleep 3
+	curl -X PUT \
+	  -H 'Authorization: Bearer anonymous' \
+	  -H 'Content-Type: application/json' \
+	  -d @test/data/project5_repo1_content1_update2.json \
+	  http://localhost:36462/api/v0/projects/project5/repositories/repo1/files/revisions/head
+	sleep 3
+	curl -X PUT \
+	  -H 'Authorization: Bearer anonymous' \
+	  -H 'Content-Type: application/json' \
+	  -d @test/data/project5_repo1_content1_update3.json \
+	  http://localhost:36462/api/v0/projects/project5/repositories/repo1/files/revisions/head
+
 .PHONY: clean-build
 clean-build:
 	yarn clean && yarn fix && yarn lint && yarn test && yarn build

lib/watchService.ts

@@ -11,7 +11,23 @@ const {
 
 const REQUEST_HEADER_PREFER_SECONDS_DEFAULT = 60;
 
+export type ParamsWatchFile = {
+    project: string;
+    repo: string;
+    filePath: string;
+    timeoutSeconds?: number;
+};
+
+export type ParamsWatchRepo = {
+    project: string;
+    repo: string;
+    pathPattern: string;
+    lastKnownRevision: number;
+    timeoutSeconds?: number;
+};
+
 export type WatchResult = {
+    revision: number;
     entry: Entry;
 };
 
@@ -24,21 +40,21 @@ export class WatchService {
         this.contentService = contentService;
     }
 
-    watchFile(
-        project: string,
-        repo: string,
-        path: string,
-        timeoutSeconds?: number
-    ): EventEmitter {
+    watchFile(params: ParamsWatchFile): EventEmitter {
         const emitter = new EventEmitter();
 
         setImmediate(async () => {
             // get the current entry
-            const entry = await this.contentService.getFile(project, repo, {
-                path,
-                type: QueryTypes.Identity,
-            });
+            const entry = await this.contentService.getFile(
+                params.project,
+                params.repo,
+                {
+                    path: params.filePath,
+                    type: QueryTypes.Identity,
+                }
+            );
             const currentEntry: WatchResult = {
+                revision: entry.revision ?? 1,
                 entry,
             };
             emitter.emit('data', currentEntry);
@@ -48,13 +64,14 @@ export class WatchService {
                 let currentRevision = revision;
                 try {
                     const watchResult = await this.watchFileInner(
-                        project,
-                        repo,
-                        path,
+                        params.project,
+                        params.repo,
+                        params.filePath,
                         currentRevision,
-                        timeoutSeconds ?? REQUEST_HEADER_PREFER_SECONDS_DEFAULT
+                        params.timeoutSeconds ??
+                            REQUEST_HEADER_PREFER_SECONDS_DEFAULT
                     );
-                    currentRevision = watchResult.entry.revision ?? -1;
+                    currentRevision = watchResult.revision;
                     emitter.emit('data', watchResult);
                 } catch (e) {
                     // TODO: implement exponential backoff with jitter
@@ -76,6 +93,43 @@ export class WatchService {
         return emitter;
     }
 
+    watchRepo(params: ParamsWatchRepo): EventEmitter {
+        const emitter = new EventEmitter();
+
+        setImmediate(async () => {
+            // start watching
+            const watch = async (revision: number) => {
+                let currentRevision = revision;
+                try {
+                    const watchResult = await this.watchFileInner(
+                        params.project,
+                        params.repo,
+                        params.pathPattern,
+                        currentRevision,
+                        params.timeoutSeconds ??
+                            REQUEST_HEADER_PREFER_SECONDS_DEFAULT
+                    );
+                    currentRevision = watchResult.revision;
+                    emitter.emit('data', watchResult);
+                } catch (e) {
+                    // TODO: implement exponential backoff with jitter
+                    if (e.statusCode !== HTTP_STATUS_NOT_MODIFIED) {
+                        emitter.emit('error', e);
+                    }
+                } finally {
+                    setImmediate(() => {
+                        watch(currentRevision);
+                    });
+                }
+            };
+            setImmediate(() => {
+                watch(params.lastKnownRevision);
+            });
+        });
+
+        return emitter;
+    }
+
     private async watchFileInner(
         project: string,
         repo: string,
@@ -92,15 +146,6 @@ export class WatchService {
             [HTTP2_HEADER_PREFER]: prefer,
         };
         const response = await this.httpClient.get(requestPath, headers);
-        const entry: Entry = response.data
-            ? JSON.parse(response.data).entry ?? {}
-            : {};
-        return {
-            entry,
-        };
-    }
-
-    async watchRepo(): Promise<WatchResult> {
-        throw new Error('not implemented');
+        return response.data ? JSON.parse(response.data) : {};
     }
 }

test/data/project5.json

@@ -0,0 +1,3 @@
+{
+    "name": "project5"
+}

test/data/project5_repo1.json

@@ -0,0 +1,3 @@
+{
+    "name": "repo1"
+}

test/data/project5_repo1_content1.json

@@ -0,0 +1,15 @@
+{
+    "file": {
+        "name": "a.json",
+        "type": "JSON",
+        "path": "/a.json",
+        "content": "{\"field1\": \"foo\"}"
+    },
+    "commitMessage": {
+        "summary": "Add /a.json",
+        "detail": {
+            "content": "",
+            "markup": "PLAINTEXT"
+        }
+    }
+}

test/data/project5_repo1_content1_update1.json

@@ -0,0 +1,16 @@
+{
+    "file": {
+        "revision": "HEAD",
+        "name": "a.json",
+        "type": "JSON",
+        "path": "/a.json",
+        "content": "{\"field1\": \"foo1\"}"
+    },
+    "commitMessage": {
+        "summary": "Edit /a.json",
+        "detail": {
+            "content": "",
+            "markup": "PLAINTEXT"
+        }
+    }
+}

test/data/project5_repo1_content1_update2.json

@@ -0,0 +1,16 @@
+{
+    "file": {
+        "revision": "HEAD",
+        "name": "a.json",
+        "type": "JSON",
+        "path": "/a.json",
+        "content": "{\"field1\": \"foo2\"}"
+    },
+    "commitMessage": {
+        "summary": "Edit /a.json",
+        "detail": {
+            "content": "",
+            "markup": "PLAINTEXT"
+        }
+    }
+}

test/data/project5_repo1_content1_update3.json

@@ -0,0 +1,16 @@
+{
+    "file": {
+        "revision": "HEAD",
+        "name": "a.json",
+        "type": "JSON",
+        "path": "/a.json",
+        "content": "{\"field1\": \"foo3\"}"
+    },
+    "commitMessage": {
+        "summary": "Edit /a.json",
+        "detail": {
+            "content": "",
+            "markup": "PLAINTEXT"
+        }
+    }
+}

test/watchService.test.ts

@@ -1,15 +1,21 @@
 import { constants as http2constants } from 'http2';
 import { exec } from 'child_process';
 import { HttpClient } from '../lib/internal/httpClient';
-import { ContentService, WatchResult, WatchService } from '../lib';
-import { QueryTypes } from '../lib/contentService';
+import {
+    ContentService,
+    RepositoryService,
+    WatchResult,
+    WatchService,
+} from '../lib';
+import { ChangeTypes, QueryTypes } from '../lib/contentService';
 
 const { HTTP_STATUS_NOT_MODIFIED } = http2constants;
 
 const client = new HttpClient({
     baseURL: 'http://localhost:36462',
 });
 const contentService = new ContentService(client);
+const repositoryService = new RepositoryService(client);
 const sut = new WatchService(client, contentService);
 
 describe('WatchService', () => {
@@ -50,16 +56,20 @@ describe('WatchService', () => {
     it('watchFile', async () => {
         const project = 'project2';
         const repo = 'repo2';
-        const path = '/test8.json';
+        const filePath = '/test8.json';
 
-        const emitter = await sut.watchFile(project, repo, path);
+        const emitter = await sut.watchFile({
+            project,
+            repo,
+            filePath,
+        });
 
         let count = 0;
         emitter.on('data', (data: WatchResult) => {
             count++;
             console.log(`data=${JSON.stringify(data)}`);
 
-            expect(data.entry.path).toBe(path);
+            expect(data.entry.path).toBe(filePath);
         });
         emitter.on('error', (e) => {
             console.log(`error=${JSON.stringify(e)}`);
@@ -68,7 +78,7 @@ describe('WatchService', () => {
 
         setTimeout(() => {
             // The target updates the json three times
-            exec('make update-test-data', (e) => {
+            exec('make update-test-data-for-watchFile', (e) => {
                 if (e) {
                     // fail
                     expect(true).toBe(false);
@@ -80,4 +90,64 @@ describe('WatchService', () => {
 
         expect(count).toBe(4); // initial entry + updated three times = 4 times
     }, 30_000);
+
+    it('watchRepo', async () => {
+        const project = 'project5';
+        const repoName = 'repo1';
+        const pathPattern = '/**';
+
+        const repos = await repositoryService.list(project);
+        expect(repos.length).toBe(3);
+
+        const repo = repos.filter((repo) => repo.name === repoName);
+        const lastKnownRevision = repo[0].headRevision ?? 1;
+
+        const emitter = await sut.watchRepo({
+            project,
+            repo: repoName,
+            pathPattern,
+            lastKnownRevision,
+        });
+
+        let count = 0;
+        emitter.on('data', (data: WatchResult) => {
+            count++;
+            console.log(`data=${JSON.stringify(data)}`);
+
+            // expect(data.entry.path).toBe(filePath);
+        });
+        emitter.on('error', (e) => {
+            console.log(`error=${JSON.stringify(e)}`);
+            throw e;
+        });
+
+        setTimeout(() => {
+            // The target updates the json three times
+            exec('make update-test-data-for-watchRepo', (e) => {
+                if (e) {
+                    // fail
+                    expect(true).toBe(false);
+                }
+            });
+        }, 1_000);
+
+        await sleep(20_000);
+
+        expect(count).toBe(4); // initial entry + updated three times = 4 times
+
+        await contentService.push({
+            project,
+            repo: repoName,
+            baseRevision: 'HEAD',
+            commitMessage: {
+                summary: 'Remove the test file',
+            },
+            changes: [
+                {
+                    path: '/a.json',
+                    type: ChangeTypes.Remove,
+                },
+            ],
+        });
+    }, 30_000);
 });