feat(rpc): handle SA returning file

packages/datasource-rpc/package.json

@@ -32,6 +32,7 @@
     "build:watch": "tsc --watch",
     "clean": "rm -rf coverage dist",
     "lint": "eslint src",
-    "publish:package": "semantic-release"
+    "publish:package": "semantic-release",
+    "test": "jest"
   }
 }

packages/datasource-rpc/src/collection.ts

@@ -1,4 +1,5 @@
 import {
+  ActionResult,
   Aggregation,
   BaseCollection,
   Caller,
@@ -12,6 +13,8 @@ import {
   Projection,
   RecordData,
 } from '@forestadmin/datasource-toolkit';
+import { IncomingHttpHeaders } from 'http';
+import { Readable } from 'stream';
 import superagent from 'superagent';
 
 import { RpcDataSourceOptions } from './types';
@@ -115,7 +118,12 @@ export default class RpcCollection extends BaseCollection {
     return response.body;
   }
 
-  override async execute(caller: Caller, name: string, formValues: RecordData, filter?: Filter) {
+  override async execute(
+    caller: Caller,
+    name: string,
+    formValues: RecordData,
+    filter?: Filter,
+  ): Promise<ActionResult> {
     const url = `${this.rpcCollectionUri}/action-execute`;
 
     this.logger(
@@ -125,17 +133,45 @@ export default class RpcCollection extends BaseCollection {
 
     const request = superagent.post(url);
     appendHeaders(request, this.options.authSecret, caller);
+    // Buffer the response ourselves to branch between binary (File) and JSON paths on the
+    // response headers — superagent's default JSON parser would error on binary bodies.
+    request.buffer(true);
+    request.parse((res, callback) => {
+      const chunks: Uint8Array[] = [];
+      res.on('data', (chunk: Uint8Array) => chunks.push(chunk));
+      res.once('end', () =>
+        callback(null, { headers: res.headers, buffer: Buffer.concat(chunks) }),
+      );
+      res.once('error', err => callback(err, null));
+    });
+
     const response = await request.send({
       action: name,
       filter: keysToSnake(filter),
       data: formValues,
     });
-
-    const body = keysToCamel(response.body);
+    const { headers, buffer } = response.body as {
+      headers: IncomingHttpHeaders;
+      buffer: Buffer;
+    };
+
+    if (headers['x-forest-action-type'] === 'File') {
+      const responseHeaders = headers['x-forest-action-response-headers'] as string | undefined;
+      const fileNameHeader = (headers['x-forest-action-file-name'] as string) || '';
+
+      return {
+        type: 'File',
+        mimeType: headers['content-type'] as string,
+        name: decodeURIComponent(fileNameHeader),
+        stream: Readable.from(buffer),
+        ...(responseHeaders ? { responseHeaders: JSON.parse(responseHeaders) } : {}),
+      };
+    }
+
+    const raw = buffer.toString('utf-8');
+    const body = keysToCamel(raw ? JSON.parse(raw) : {});
     body.invalidated = new Set(body.invalidated);
 
-    // TODO action with file
-
     return body;
   }
 

packages/datasource-rpc/test/collection.test.ts

@@ -0,0 +1,129 @@
+import http, { IncomingMessage, Server, ServerResponse } from 'http';
+import { AddressInfo } from 'net';
+import { Readable } from 'stream';
+
+import RpcCollection from '../src/collection';
+
+type Handler = (req: IncomingMessage, res: ServerResponse) => void;
+
+async function startServer(handler: Handler): Promise<{ server: Server; uri: string }> {
+  const server = http.createServer(handler);
+  await new Promise<void>(resolve => server.listen(0, '127.0.0.1', resolve));
+  const { address, port } = server.address() as AddressInfo;
+
+  return { server, uri: `http://${address}:${port}` };
+}
+
+async function stopServer(server: Server) {
+  await new Promise<void>(resolve => server.close(() => resolve()));
+}
+
+function readBody(req: IncomingMessage): Promise<string> {
+  return new Promise((resolve, reject) => {
+    const chunks: Uint8Array[] = [];
+    req.on('data', chunk => chunks.push(chunk));
+    req.on('end', () => resolve(Buffer.concat(chunks).toString('utf-8')));
+    req.on('error', reject);
+  });
+}
+
+function buildCollection(uri: string) {
+  const logger = jest.fn();
+  const datasource = { collections: [], schema: { charts: [] } };
+  const options = { uri, authSecret: 'secret' };
+  const schema = {
+    actions: {},
+    charts: [],
+    fields: {},
+    segments: [],
+    aggregationCapabilities: { supportedDateOperations: new Set(), supportGroups: false },
+    countable: false,
+    searchable: false,
+  };
+
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  return new RpcCollection(logger as any, datasource as any, options, 'books', schema as any);
+}
+
+describe('RpcCollection.execute', () => {
+  let server: Server;
+  let uri: string;
+
+  afterEach(async () => {
+    if (server) await stopServer(server);
+  });
+
+  it('parses a JSON Success response and rebuilds invalidated as a Set', async () => {
+    const received: { body?: unknown } = {};
+
+    ({ server, uri } = await startServer(async (req, res) => {
+      received.body = JSON.parse(await readBody(req));
+      res.setHeader('Content-Type', 'application/json');
+      res.end(
+        JSON.stringify({
+          type: 'Success',
+          message: 'ok',
+          invalidated: ['books'],
+          response_headers: { 'x-foo': 'bar' },
+        }),
+      );
+    }));
+
+    const collection = buildCollection(uri);
+    const result = await collection.execute({ id: 1 } as never, 'noop', { foo: 'bar' }, undefined);
+
+    expect(received.body).toEqual({ action: 'noop', filter: undefined, data: { foo: 'bar' } });
+    expect(result).toEqual({
+      type: 'Success',
+      message: 'ok',
+      invalidated: new Set(['books']),
+      responseHeaders: { 'x-foo': 'bar' },
+    });
+  });
+
+  it('returns a FileResult with a Readable stream when X-Forest-Action-Type=File', async () => {
+    ({ server, uri } = await startServer(async (req, res) => {
+      await readBody(req);
+      res.setHeader('Content-Type', 'application/pdf');
+      res.setHeader('Content-Disposition', 'attachment; filename="report%20final.pdf"');
+      res.setHeader('X-Forest-Action-Type', 'File');
+      res.setHeader('X-Forest-Action-File-Name', 'report%20final.pdf');
+      res.setHeader(
+        'X-Forest-Action-Response-Headers',
+        JSON.stringify({ 'set-cookie': 'token=xyz' }),
+      );
+      Readable.from([Buffer.from('hello-pdf')]).pipe(res);
+    }));
+
+    const collection = buildCollection(uri);
+    const result = await collection.execute({ id: 1 } as never, 'download', {}, undefined);
+
+    expect(result.type).toBe('File');
+    if (result.type !== 'File') throw new Error('unreachable');
+    expect(result.mimeType).toBe('application/pdf');
+    expect(result.name).toBe('report final.pdf');
+    expect(result.responseHeaders).toEqual({ 'set-cookie': 'token=xyz' });
+
+    const chunks: Uint8Array[] = [];
+    for await (const chunk of result.stream) chunks.push(chunk as Uint8Array);
+    expect(Buffer.concat(chunks).toString('utf-8')).toBe('hello-pdf');
+  });
+
+  it('omits responseHeaders when the header is absent', async () => {
+    ({ server, uri } = await startServer(async (req, res) => {
+      await readBody(req);
+      res.setHeader('Content-Type', 'text/plain');
+      res.setHeader('X-Forest-Action-Type', 'File');
+      res.setHeader('X-Forest-Action-File-Name', 'note.txt');
+      res.end('hi');
+    }));
+
+    const collection = buildCollection(uri);
+    const result = await collection.execute({ id: 1 } as never, 'download', {}, undefined);
+
+    expect(result.type).toBe('File');
+    if (result.type !== 'File') throw new Error('unreachable');
+    expect(result.responseHeaders).toBeUndefined();
+    expect(result.name).toBe('note.txt');
+  });
+});

packages/rpc-agent/package.json

@@ -26,6 +26,7 @@
     "build:watch": "tsc --watch",
     "clean": "rm -rf coverage dist",
     "lint": "eslint src",
-    "publish:package": "semantic-release"
+    "publish:package": "semantic-release",
+    "test": "jest"
   }
 }

packages/rpc-agent/src/routes/action.ts

@@ -19,7 +19,25 @@ export default class RpcActionRoute extends CollectionRoute {
       parseFilter(this.collection, filter),
     );
 
-    // TODO action with file
+    if (actionResult.type === 'File') {
+      const encodedName = encodeURIComponent(actionResult.name);
+
+      context.set('Content-Type', actionResult.mimeType);
+      context.set('Content-Disposition', `attachment; filename="${encodedName}"`);
+      context.set('X-Forest-Action-Type', 'File');
+      context.set('X-Forest-Action-File-Name', encodedName);
+
+      if (actionResult.responseHeaders) {
+        context.set(
+          'X-Forest-Action-Response-Headers',
+          JSON.stringify(actionResult.responseHeaders),
+        );
+      }
+
+      context.body = actionResult.stream;
+
+      return;
+    }
 
     context.response.body = {
       ...keysToSnake(actionResult),

packages/rpc-agent/test/routes/action.test.ts

@@ -0,0 +1,106 @@
+import { Readable } from 'stream';
+
+import RpcActionRoute from '../../src/routes/action';
+
+type FakeContext = {
+  request: { body: unknown; headers: Record<string, string> };
+  headers: Record<string, string>;
+  response: { body?: unknown };
+  body?: unknown;
+  set: jest.Mock;
+};
+
+function createContext(body: unknown): FakeContext {
+  return {
+    request: { body, headers: {} },
+    headers: {},
+    response: {},
+    set: jest.fn(),
+  };
+}
+
+function createRoute(execute: jest.Mock) {
+  const route = Object.create(RpcActionRoute.prototype) as RpcActionRoute & {
+    collection: { execute: jest.Mock };
+  };
+
+  Object.defineProperty(route, 'collection', {
+    value: { execute },
+    configurable: true,
+  });
+
+  return route;
+}
+
+describe('RpcActionRoute', () => {
+  describe('handleExecute', () => {
+    it('serialises a Success result as JSON with invalidated as array', async () => {
+      const execute = jest.fn().mockResolvedValue({
+        type: 'Success',
+        message: 'done',
+        invalidated: new Set(['books', 'authors']),
+        responseHeaders: { 'x-foo': 'bar' },
+      });
+      const route = createRoute(execute);
+      const ctx = createContext({ action: 'noop', filter: null, data: { id: 1 } });
+
+      await route.handleExecute(ctx);
+
+      expect(execute).toHaveBeenCalledTimes(1);
+      expect(ctx.set).not.toHaveBeenCalled();
+      expect(ctx.response.body).toEqual({
+        type: 'Success',
+        message: 'done',
+        response_headers: { 'x-foo': 'bar' },
+        invalidated: ['books', 'authors'],
+      });
+    });
+
+    it('streams a File result with the expected headers', async () => {
+      const stream = Readable.from(['hello']);
+      const execute = jest.fn().mockResolvedValue({
+        type: 'File',
+        mimeType: 'application/pdf',
+        name: 'report final.pdf',
+        stream,
+      });
+      const route = createRoute(execute);
+      const ctx = createContext({ action: 'download', filter: null, data: {} });
+
+      await route.handleExecute(ctx);
+
+      expect(ctx.set).toHaveBeenCalledWith('Content-Type', 'application/pdf');
+      expect(ctx.set).toHaveBeenCalledWith(
+        'Content-Disposition',
+        'attachment; filename="report%20final.pdf"',
+      );
+      expect(ctx.set).toHaveBeenCalledWith('X-Forest-Action-Type', 'File');
+      expect(ctx.set).toHaveBeenCalledWith('X-Forest-Action-File-Name', 'report%20final.pdf');
+      expect(ctx.set).not.toHaveBeenCalledWith(
+        'X-Forest-Action-Response-Headers',
+        expect.anything(),
+      );
+      expect(ctx.body).toBe(stream);
+      expect(ctx.response.body).toBeUndefined();
+    });
+
+    it('forwards responseHeaders on File result via X-Forest-Action-Response-Headers', async () => {
+      const execute = jest.fn().mockResolvedValue({
+        type: 'File',
+        mimeType: 'text/csv',
+        name: 'export.csv',
+        stream: Readable.from(['a,b']),
+        responseHeaders: { 'set-cookie': 'token=xyz' },
+      });
+      const route = createRoute(execute);
+      const ctx = createContext({ action: 'export', filter: null, data: {} });
+
+      await route.handleExecute(ctx);
+
+      expect(ctx.set).toHaveBeenCalledWith(
+        'X-Forest-Action-Response-Headers',
+        JSON.stringify({ 'set-cookie': 'token=xyz' }),
+      );
+    });
+  });
+});