feat: Deliver parcels to public gateway (#243)

Fixes  #10

TODO:

- [x] Start when the daemon starts
- [x] Test logging to external files

packages/daemon/package.json

@@ -17,6 +17,7 @@
     "@relaycorp/relaynet-poweb": "^1.5.2",
     "abort-controller": "^3.0.0",
     "abortable-iterator": "^3.0.0",
+    "bson": "^4.4.0",
     "buffer-to-arraybuffer": "0.0.6",
     "bufferutil": "^4.0.3",
     "date-fns": "^2.22.1",

packages/daemon/src/bin/subprocess.ts

@@ -3,4 +3,4 @@
 import './_setup';
 
 import main from '../subprocessEntrypoint';
-main();
+main(process.argv[2]);

packages/daemon/src/daemon.spec.ts

@@ -16,7 +16,7 @@ mockLoggerToken();
 test('Startup routine should be called', async () => {
   await daemon();
 
-  expect(startup).toBeCalled();
+  expect(startup).toBeCalledWith('daemon');
 });
 
 test('Server should be run', async () => {

packages/daemon/src/daemon.ts

@@ -6,7 +6,7 @@ import runSync from './sync';
 import { LOGGER } from './tokens';
 
 export default async function (): Promise<void> {
-  await startup();
+  await startup('daemon');
 
   const logger = Container.get(LOGGER);
   const server = await makeServer(logger);

packages/daemon/src/fileStore.spec.ts

@@ -4,6 +4,7 @@ import { tmpdir } from 'os';
 import { join } from 'path';
 
 import { FileStore, FileStoreError } from './fileStore';
+import { asyncIterableToArray } from './testUtils/iterables';
 import { getPromiseRejection } from './testUtils/promises';
 
 let tempDir: string;
@@ -83,3 +84,81 @@ describe('putObject', () => {
     await expect(fs.readFile(join(tempAppDirs.data, OBJECT_KEY))).resolves.toEqual(contentV2);
   });
 });
+
+describe('deleteObject', () => {
+  test('Non-existing object should be ignored', async () => {
+    const store = new FileStore(tempAppDirs);
+
+    await store.deleteObject(OBJECT_KEY);
+  });
+
+  test('Error deleting file should be propagated', async () => {
+    const store = new FileStore(tempAppDirs);
+    const unlinkError = new Error('oh no');
+    const unlinkSpy = jest.spyOn(fs, 'unlink');
+    unlinkSpy.mockRejectedValueOnce(unlinkError);
+
+    const error = await getPromiseRejection(store.deleteObject(OBJECT_KEY), FileStoreError);
+
+    expect(error.message).toMatch(/^Failed to delete object: /);
+    expect(error.cause()).toEqual(unlinkError);
+  });
+
+  test('Existing object should be deleted', async () => {
+    const store = new FileStore(tempAppDirs);
+    await store.putObject(OBJECT_CONTENT, OBJECT_KEY);
+
+    await store.deleteObject(OBJECT_KEY);
+
+    await expect(store.getObject(OBJECT_KEY)).resolves.toBeNull();
+  });
+});
+
+describe('listObjects', () => {
+  const keyPrefix = 'sub';
+
+  test('Non-existing key prefixes should be ignored', async () => {
+    const store = new FileStore(tempAppDirs);
+
+    await expect(asyncIterableToArray(store.listObjects(keyPrefix))).resolves.toHaveLength(0);
+  });
+
+  test('Error reading existing key prefix should be propagated', async () => {
+    const store = new FileStore(tempAppDirs);
+    const readError = new Error('oh no');
+    const readdirSpy = jest.spyOn(fs, 'readdir');
+    readdirSpy.mockRejectedValueOnce(readError);
+
+    const error = await getPromiseRejection(
+      asyncIterableToArray(store.listObjects(keyPrefix)),
+      FileStoreError,
+    );
+
+    expect(error.message).toMatch(/^Failed to read directory: /);
+    expect(error.cause()).toEqual(readError);
+  });
+
+  test('No objects should be output if there are none', async () => {
+    const store = new FileStore(tempAppDirs);
+    const subdirectoryPath = join(tempAppDirs.data, keyPrefix);
+    await fs.mkdir(subdirectoryPath, { recursive: true });
+
+    await expect(asyncIterableToArray(store.listObjects(keyPrefix))).resolves.toHaveLength(0);
+  });
+
+  test('Objects at the root should be output', async () => {
+    const store = new FileStore(tempAppDirs);
+    const key = join(keyPrefix, 'thingy');
+    await store.putObject(OBJECT_CONTENT, key);
+
+    await expect(asyncIterableToArray(store.listObjects(keyPrefix))).resolves.toEqual([key]);
+  });
+
+  test('Objects in subdirectories should be output', async () => {
+    const store = new FileStore(tempAppDirs);
+    const key = join(keyPrefix, 'another-sub', 'thingy');
+    await store.putObject(OBJECT_CONTENT, key);
+
+    await expect(asyncIterableToArray(store.listObjects(keyPrefix))).resolves.toEqual([key]);
+  });
+});

packages/daemon/src/fileStore.ts

@@ -1,6 +1,6 @@
 // tslint:disable:max-classes-per-file
 import { Paths } from 'env-paths';
-import { promises as fs } from 'fs';
+import { Dirent, promises as fs } from 'fs';
 import { dirname, join } from 'path';
 import { Inject, Service } from 'typedi';
 
@@ -18,7 +18,7 @@ export class FileStore {
   }
 
   public async getObject(key: string): Promise<Buffer | null> {
-    const objectPath = join(this.dataPath, key);
+    const objectPath = this.getObjectPath(key);
     try {
       return await fs.readFile(objectPath);
     } catch (err) {
@@ -31,9 +31,46 @@ export class FileStore {
   }
 
   public async putObject(objectContent: Buffer, key: string): Promise<void> {
-    const objectPath = join(this.dataPath, key);
+    const objectPath = this.getObjectPath(key);
     const objectDirPath = dirname(objectPath);
     await fs.mkdir(objectDirPath, { recursive: true });
     await fs.writeFile(objectPath, objectContent);
   }
+
+  public async deleteObject(key: string): Promise<void> {
+    const objectPath = this.getObjectPath(key);
+    try {
+      await fs.unlink(objectPath);
+    } catch (err) {
+      if (err.code !== 'ENOENT') {
+        throw new FileStoreError(err, 'Failed to delete object');
+      }
+    }
+  }
+
+  public async *listObjects(keyPrefix: string): AsyncIterable<string> {
+    const directoryPath = this.getObjectPath(keyPrefix);
+    let directoryContents: readonly Dirent[];
+    try {
+      directoryContents = await fs.readdir(directoryPath, { withFileTypes: true });
+    } catch (err) {
+      if (err.code === 'ENOENT') {
+        directoryContents = [];
+      } else {
+        throw new FileStoreError(err, 'Failed to read directory');
+      }
+    }
+    for (const directoryItem of directoryContents) {
+      const itemRelativePath = join(keyPrefix, directoryItem.name);
+      if (directoryItem.isDirectory()) {
+        yield* await this.listObjects(itemRelativePath);
+      } else {
+        yield itemRelativePath;
+      }
+    }
+  }
+
+  protected getObjectPath(key: string): string {
+    return join(this.dataPath, key);
+  }
 }