Address PR feedback

Author: tohhsinpei

npm-shrinkwrap.json

@@ -120,7 +120,6 @@
     "google-auth-library": "^7.11.0",
     "inquirer": "^8.2.0",
     "js-yaml": "^3.13.1",
-    "JSONStream": "^1.3.5",
     "jsonwebtoken": "^9.0.0",
     "leven": "^3.1.0",
     "libsodium-wrappers": "^0.7.10",

package.json

@@ -7,16 +7,27 @@ import DatabaseImporter from "../database/import";
 import { Emulators } from "../emulator/types";
 import { FirebaseError } from "../error";
 import { logger } from "../logger";
+import { needProjectId } from "../projectUtils";
+import { Options } from "../options";
 import { printNoticeIfEmulated } from "../emulator/commandUtils";
 import { promptOnce } from "../prompt";
-import { populateInstanceDetails } from "../management/database";
+import { DatabaseInstance, populateInstanceDetails } from "../management/database";
 import { realtimeOriginOrEmulatorOrCustomUrl } from "../database/api";
 import { requireDatabaseInstance } from "../requireDatabaseInstance";
 import { requirePermissions } from "../requirePermissions";
 
+interface DatabaseImportOptions extends Options {
+  instance: string;
+  instanceDetails: DatabaseInstance;
+  disableTriggers?: boolean;
+  filter?: string;
+}
+
 export const command = new Command("database:import <path> [infile]")
-  .description("non-atomically import JSON file to the specified path")
-  .option("-f, --force", "pass this option to bypass confirmation prompt")
+  .description(
+    "non-atomically import the contents of a JSON file to the specified path in Realtime Database"
+  )
+  .withForce()
   .option(
     "--instance <instance>",
     "use the database <instance>.firebaseio.com (if omitted, use default database instance)"
@@ -27,26 +38,23 @@ export const command = new Command("database:import <path> [infile]")
     true
   )
   .option(
-    "--only <dataPath>",
+    "--filter <dataPath>",
     "import only data at this path in the JSON file (if omitted, import entire file)"
   )
   .before(requirePermissions, ["firebasedatabase.instances.update"])
   .before(requireDatabaseInstance)
   .before(populateInstanceDetails)
   .before(printNoticeIfEmulated, Emulators.DATABASE)
-  .action(async (path: string, infile, options) => {
+  .action(async (path: string, infile: string | undefined, options: DatabaseImportOptions) => {
     if (!path.startsWith("/")) {
       throw new FirebaseError("Path must begin with /");
     }
 
-    if (options.only && !options.only.startsWith("/")) {
-      throw new FirebaseError("Data path must begin with /");
-    }
-
     if (!infile) {
       throw new FirebaseError("No file supplied");
     }
 
+    const projectId = needProjectId(options);
     const origin = realtimeOriginOrEmulatorOrCustomUrl(options.instanceDetails.databaseUrl);
     const dbPath = utils.getDatabaseUrl(origin, options.instance, path);
     const dbUrl = new URL(dbPath);
@@ -68,7 +76,7 @@ export const command = new Command("database:import <path> [infile]")
     }
 
     const inStream = fs.createReadStream(infile);
-    const dataPath = options.only || "/";
+    const dataPath = options.filter || "";
     const importer = new DatabaseImporter(dbUrl, inStream, dataPath);
 
     let responses;
@@ -91,6 +99,6 @@ export const command = new Command("database:import <path> [infile]")
     logger.info();
     logger.info(
       clc.bold("View data at:"),
-      utils.getDatabaseViewDataUrl(origin, options.project, options.instance, path)
+      utils.getDatabaseViewDataUrl(origin, projectId, options.instance, path)
     );
   });

src/commands/database-import.ts

@@ -1,13 +1,15 @@
+import * as Chain from "stream-chain";
 import * as clc from "colorette";
+import * as Filter from "stream-json/filters/Filter";
 import * as stream from "stream";
+import * as StreamObject from "stream-json/streamers/StreamObject";
+
 import pLimit from "p-limit";
 
 import { URL } from "url";
-import { Client } from "../apiv2";
+import { Client, ClientResponse } from "../apiv2";
 import { FirebaseError } from "../error";
 
-const JSONStream = require("JSONStream");
-
 const MAX_CHUNK_SIZE = 1024 * 1024;
 const CONCURRENCY_LIMIT = 5;
 
@@ -27,17 +29,15 @@ type ChunkedData = {
  * The data is parsed and chunked into subtrees of ~1 MB, to be subsequently written in parallel.
  */
 export default class DatabaseImporter {
-  private jsonPath: string;
   private client: Client;
   private limit = pLimit(CONCURRENCY_LIMIT);
 
   constructor(
     private dbUrl: URL,
-    private inStream: NodeJS.ReadableStream,
-    dataPath: string,
+    private inStream: stream.Readable,
+    private dataPath: string,
     private chunkSize = MAX_CHUNK_SIZE
   ) {
-    this.jsonPath = this.computeJsonPath(dataPath);
     this.client = new Client({ urlPrefix: dbUrl.origin, auth: true });
   }
 
@@ -46,7 +46,7 @@ export default class DatabaseImporter {
    */
   async execute(): Promise<any> {
     await this.checkLocationIsEmpty();
-    return this.readAndWriteChunks(this.inStream);
+    return this.readAndWriteChunks();
   }
 
   private async checkLocationIsEmpty(): Promise<void> {
@@ -66,7 +66,7 @@ export default class DatabaseImporter {
     }
   }
 
-  private readAndWriteChunks(inStream: NodeJS.ReadableStream): Promise<any> {
+  private readAndWriteChunks(): Promise<any> {
     const { dbUrl } = this;
     const chunkData = this.chunkData.bind(this);
     const writeChunk = this.writeChunk.bind(this);
@@ -90,14 +90,24 @@ export default class DatabaseImporter {
 
     return new Promise((resolve, reject) => {
       const responses: any[] = [];
-      inStream
-        .pipe(JSONStream.parse(this.jsonPath))
+      const pipeline = new Chain([
+        this.inStream,
+        Filter.withParser({
+          filter: this.computeFilterString(this.dataPath) || (() => true),
+          pathSeparator: "/",
+        }),
+        StreamObject.streamObject(),
+      ]);
+      pipeline
         .on("error", (err: any) =>
           reject(
-            new FirebaseError("Invalid data; couldn't parse JSON object, array, or value.", {
-              original: err,
-              exit: 2,
-            })
+            new FirebaseError(
+              `Invalid data; couldn't parse JSON object, array, or value. ${err.message}`,
+              {
+                original: err,
+                exit: 2,
+              }
+            )
           )
         )
         .pipe(readChunks)
@@ -108,7 +118,7 @@ export default class DatabaseImporter {
     });
   }
 
-  private writeChunk(chunk: Data): Promise<any> {
+  private writeChunk(chunk: Data): Promise<ClientResponse<any>> {
     return this.limit(() =>
       this.client.request({
         method: "PUT",
@@ -152,12 +162,8 @@ export default class DatabaseImporter {
     }
   }
 
-  private computeJsonPath(dataPath: string): string {
-    if (dataPath === "/") {
-      return "$*";
-    } else {
-      return `${dataPath.split("/").slice(1).join(".")}.$*`;
-    }
+  private computeFilterString(dataPath: string): string {
+    return dataPath.split("/").filter(Boolean).join("/");
   }
 
   private joinPath(root: string, key: string): string {

src/database/import.ts

@@ -1,4 +1,5 @@
 import * as nock from "nock";
+import * as stream from "stream";
 import * as utils from "../../utils";
 import { expect } from "chai";
 
@@ -9,21 +10,21 @@ const dbUrl = new URL("https://test-db.firebaseio.com/foo");
 
 describe("DatabaseImporter", () => {
   const DATA = { a: 100, b: [true, "bar", { f: { g: 0, h: 1 }, i: "baz" }] };
-  let DATA_STREAM: NodeJS.ReadableStream;
+  let DATA_STREAM: stream.Readable;
 
   beforeEach(() => {
     DATA_STREAM = utils.stringToStream(JSON.stringify(DATA))!;
   });
 
   it("throws FirebaseError when JSON is invalid", async () => {
     nock("https://test-db.firebaseio.com").get("/foo.json?shallow=true").reply(200);
-
     const INVALID_JSON = '{"a": {"b"}}';
     const importer = new DatabaseImporter(
       dbUrl,
       utils.stringToStream(INVALID_JSON)!,
       /* importPath= */ "/"
     );
+
     await expect(importer.execute()).to.be.rejectedWith(
       FirebaseError,
       "Invalid data; couldn't parse JSON object, array, or value."
@@ -36,9 +37,10 @@ describe("DatabaseImporter", () => {
     nock("https://test-db.firebaseio.com")
       .put("/foo/b.json", JSON.stringify([true, "bar", { f: { g: 0, h: 1 }, i: "baz" }]))
       .reply(200);
-
     const importer = new DatabaseImporter(dbUrl, DATA_STREAM, /* importPath= */ "/");
+
     const responses = await importer.execute();
+
     expect(responses).to.have.length(2);
     expect(nock.isDone()).to.be.true;
   });
@@ -52,35 +54,36 @@ describe("DatabaseImporter", () => {
       .put("/foo/b/2/f.json", JSON.stringify({ g: 0, h: 1 }))
       .reply(200);
     nock("https://test-db.firebaseio.com").put("/foo/b/2/i.json", '"baz"').reply(200);
-
     const importer = new DatabaseImporter(
       dbUrl,
       DATA_STREAM,
       /* importPath= */ "/",
       /* chunkSize= */ 20
     );
+
     const responses = await importer.execute();
+
     expect(responses).to.have.length(5);
     expect(nock.isDone()).to.be.true;
   });
 
   it("imports from data path", async () => {
     nock("https://test-db.firebaseio.com").get("/foo.json?shallow=true").reply(200);
-    nock("https://test-db.firebaseio.com").put("/foo/0.json", "true").reply(200);
-    nock("https://test-db.firebaseio.com").put("/foo/1.json", '"bar"').reply(200);
     nock("https://test-db.firebaseio.com")
-      .put("/foo/2.json", JSON.stringify({ f: { g: 0, h: 1 }, i: "baz" }))
+      .put("/foo/b.json", JSON.stringify([true, "bar", { f: { g: 0, h: 1 }, i: "baz" }]))
       .reply(200);
-
     const importer = new DatabaseImporter(dbUrl, DATA_STREAM, /* importPath= */ "/b");
+
     const responses = await importer.execute();
-    expect(responses).to.have.length(3);
+
+    expect(responses).to.have.length(1);
     expect(nock.isDone()).to.be.true;
   });
 
   it("throws FirebaseError when data location is nonempty", async () => {
     nock("https://test-db.firebaseio.com").get("/foo.json?shallow=true").reply(200, { a: "foo" });
     const importer = new DatabaseImporter(dbUrl, DATA_STREAM, /* importPath= */ "/");
+
     await expect(importer.execute()).to.be.rejectedWith(
       FirebaseError,
       /Importing is only allowed for an empty location./