loose attempt at using the much more performant REST interface for re…

…ports

package.json

@@ -18,10 +18,15 @@
   "license": "MIT",
   "dependencies": {
     "@isaacs/ttlcache": "^1.2.2",
+    "@types/stream-json": "^1.7.7",
+    "axios": "^1.6.7",
+    "decamelize": "^5.0.0",
     "google-ads-node": "^12.0.2",
     "google-auth-library": "^7.1.0",
     "google-gax": "^4.0.3",
-    "long": "^4.0.0"
+    "long": "^4.0.0",
+    "map-obj": "^4.0.0",
+    "stream-json": "^1.8.0"
   },
   "devDependencies": {
     "@types/jest": "^29.0.1",

scripts/fields.ts

@@ -158,6 +158,15 @@ export async function compileFields(): Promise<void> {
   });
 
   stream.write(`}`);
+
+  stream.write(`\n\n/*  -- Field types (used in parsing) --  */`);
+  stream.write(`\nexport const fieldDataTypes = new Map([ `);
+
+  for (const field of fields) {
+    stream.write(`\n['${field.name}','${field.data_type}'], `);
+  }
+
+  stream.write(`\n])`);
   stream.end();
 }
 

src/customer.ts

@@ -1,4 +1,9 @@
 import { CancellableStream } from "google-gax";
+import axios from "axios";
+import { chain } from "stream-chain";
+import { parser } from "stream-json";
+import { streamArray } from "stream-json/streamers/StreamArray";
+
 import { ClientOptions } from "./client";
 import {
   BaseMutationHookArgs,
@@ -8,7 +13,8 @@ import {
   Hooks,
 } from "./hooks";
 import { parse } from "./parser";
-import { services } from "./protos";
+import { decamelizeKeys } from "./parserRest";
+import { services, errors } from "./protos";
 import ServiceFactory from "./protos/autogen/serviceFactory";
 import { buildQuery } from "./query";
 import {
@@ -69,7 +75,7 @@ export class Customer extends ServiceFactory {
     options: Readonly<ReportOptions>
   ): Promise<T> {
     const { gaqlQuery, requestOptions } = buildQuery(options);
-    const { response } = await this.querier<T>(
+    const { response } = await this.querierRest<T>(
       gaqlQuery,
       requestOptions,
       options
@@ -225,6 +231,99 @@ export class Customer extends ServiceFactory {
     return { response, totalResultsCount };
   }
 
+  private async querierRest<T = services.IGoogleAdsRow[]>(
+    gaqlQuery: string,
+    requestOptions: RequestOptions = {},
+    reportOptions?: Readonly<ReportOptions>,
+    useHooks = true
+  ): Promise<{ response: T; totalResultsCount?: number }> {
+    const accessToken = await this.getAccessToken();
+    const args = {
+      method: "POST",
+      url: `https://googleads.googleapis.com/v14/customers/${this.customerOptions.customer_id}/googleAds:searchStream`,
+      headers: {
+        Authorization: `Bearer ${accessToken}`,
+        "developer-token": this.clientOptions.developer_token ?? "",
+        "login-customer-id": this.credentials.login_customer_id ?? "",
+      },
+      responseType: "stream",
+      data: {
+        query: gaqlQuery,
+        // summary_row_setting: "SUMMARY_ROW_ONLY",
+      },
+    };
+
+    console.time("request");
+    try {
+      // @ts-ignore
+      const response = await axios(args);
+
+      const stream = response.data as any;
+      let waste = 0;
+
+      let rows: any = [];
+
+      const pipeline = chain([stream, parser(), streamArray()]);
+
+      stream.once("data", () => {
+        console.timeEnd("request");
+        console.time("parsing");
+      });
+
+      pipeline.on("data", (data) => {
+        console.log(data.value.results);
+        const now = Date.now();
+        const parsed = data.value.results.map((row: any) => {
+          return decamelizeKeys(row);
+        });
+        rows = [...rows, ...parsed];
+        waste += Date.now() - now;
+      });
+
+      await new Promise<void>((resolve, reject) => {
+        pipeline.on("end", () => resolve());
+        pipeline.on("error", (err) => reject(err));
+      });
+
+      console.timeEnd("parsing");
+      console.log({ waste });
+
+      return { response: rows as T };
+    } catch (e: any) {
+      // console.log("ERROR INCOMING");
+      // console.log(e.toJSON());
+      const stream = e.response.data as any;
+
+      const pipeline = chain([stream, parser(), streamArray()]);
+
+      stream.once("data", () => {
+        console.timeEnd("request");
+        console.time("parsing");
+      });
+
+      let googleAdsFailure: errors.GoogleAdsFailure | undefined;
+      pipeline.on("data", (data) => {
+        // console.log(data);
+
+        googleAdsFailure = new errors.GoogleAdsFailure(
+          decamelizeKeys(data.value.error.details[0])
+        );
+
+        // console.log(e);
+        // console.dir(data.value, { depth: null });
+        // new Error(data.error.message);
+      });
+
+      await new Promise<void>((resolve, reject) => {
+        pipeline.on("end", () => reject(googleAdsFailure));
+        pipeline.on("error", (err) => reject(err));
+      });
+      // console.log("AFTER ERROR");
+    }
+
+    return { response: [] as T };
+  }
+
   private async querier<T = services.IGoogleAdsRow[]>(
     gaqlQuery: string,
     requestOptions: RequestOptions = {},

src/parserRest.ts

@@ -0,0 +1,79 @@
+/**
+ * JSON Rest parsing
+ */
+
+import mapObject from "map-obj";
+import decamelize from "decamelize";
+
+import { enums, fields, fieldDataTypes } from "./protos";
+
+const cache = new Map<string, string>();
+
+const isObject = (value: unknown) =>
+  typeof value === "object" && value !== null;
+
+const decamelcaseConvert = (input: any) => {
+  if (!isObject(input)) {
+    return input;
+  }
+
+  //   const stopPathsSet = new Set(stopPaths);
+
+  const makeMapper = (parentPath?: string) => (key: string, value: any) => {
+    if (isObject(value)) {
+      const path = parentPath === undefined ? key : `${parentPath}.${key}`;
+
+      // @ts-ignore
+      value = mapObject(value, makeMapper(path));
+    }
+
+    key = cachedDecamelize(key);
+    value = cachedValueParser(key, parentPath, value);
+
+    return [key, value];
+  };
+
+  // @ts-ignore
+  return mapObject(input, makeMapper(undefined));
+};
+
+const cachedDecamelize = (key: string) => {
+  if (cache.has(key)) {
+    return cache.get(key) as string;
+  }
+
+  const newKey = decamelize(key);
+
+  cache.set(key, newKey);
+
+  return newKey;
+};
+
+const cachedValueParser = (
+  key: string,
+  parentPath: string | undefined,
+  value: any
+) => {
+  let newValue = value;
+
+  const fullPath = cachedDecamelize(
+    parentPath ? `${parentPath}.${key}` : key
+  ) as keyof typeof fields.enumFields;
+
+  const dataType = fieldDataTypes.get(fullPath);
+
+  if (dataType === "INT64") {
+    newValue = Number(value);
+  }
+
+  if (dataType === "ENUM") {
+    // @ts-expect-error typescript doesn't like accessing items in a namespace with a string
+    newValue = enums[fields.enumFields[fullPath]][value]; // e.g. enums['CampaignStatus'][ENABLED] = "2"
+  }
+
+  return newValue;
+};
+
+export const decamelizeKeys = (input: Record<string, any>) => {
+  return decamelcaseConvert(input);
+};