feat(CO2): add column in db for emissions

Author: fabienzucchet

package.json

@@ -13,6 +13,7 @@
     "data:dump": "node dist/cmd/dataDump.js",
     "data:download": "node dist/cmd/dataDownload.js",
     "data:load": "node dist/cmd/dataLoad.js",
+    "data:patchEmissions": "node dist/cmd/dataPatchEmissions.js",
     "data:status": "node dist/cmd/dataStatus.js",
     "job:init": "npm run db:setup && npm run job:update",
     "job:update": "npm run data:download && npm run data:load",
@@ -21,6 +22,7 @@
     "data:dump:dev": "ts-node src/cmd/dataDump.ts",
     "data:download:dev": "ts-node src/cmd/dataDownload.ts",
     "data:load:dev": "ts-node src/cmd/dataLoad.ts",
+    "data:patchEmissions:dev": "ts-node src/cmd/dataPatchEmissions.ts",
     "data:status:dev": "ts-node src/cmd/dataStatus.ts",
     "job:init:dev": "npm run db:setup:dev && npm run job:update:dev",
     "job:update:dev": "npm run data:download:dev && npm run data:load:dev",

src/cmd/dataPatchEmissions.ts

@@ -0,0 +1,98 @@
+import ProgressBar from 'progress';
+import config from '../config';
+import { PoolClient } from 'pg';
+
+const batchSize = 10000;
+
+async function run(): Promise<void> {
+    const pool = await config.pg();
+
+    const client = await pool.connect();
+    try {
+        // Fetch all product hashes
+        config.logger.info('Fetching all product hashes');
+        const productHashes = await getProductHashes(client);
+
+        const progressBar = new ProgressBar(
+            '-> loading [:bar] :percent (:etas remaining)',
+            {
+              width: 40,
+              complete: '=',
+              incomplete: ' ',
+              renderThrottle: 500,
+              total: productHashes.length,
+            }
+          );
+
+        // For each product hash, insert by batch the emissions data into the product table
+        config.logger.info('Updating product emissions');
+        for (let i = 0; i < productHashes.length; i += batchSize) {
+            const batch = productHashes.slice(i, i + batchSize);
+
+            await client.query('BEGIN');
+
+            await Promise.all(batch.map((productHash) => {
+                updateProductEmissions(client, productHash);
+                progressBar.tick();
+            }));
+
+            await client.query('COMMIT');
+        }
+
+        // await setEmissionsUpdateSuccessful(client);
+    } catch (e) {
+        await client.query('ROLLBACK');
+
+        // await setEmissionsUpdateFailed(client);
+
+        throw e;
+    }
+
+}
+
+async function getProductHashes(client: PoolClient): Promise<string[]> {
+    const result = await client.query(
+        `
+        SELECT "productHash"
+        FROM "products"
+        `
+    );
+
+    return result.rows.map((row) => row.productHash);
+}
+
+async function updateProductEmissions(client: PoolClient, productHash: string): Promise<void> {
+    const emissionsData = generateEmissionsData();
+    await client.query(
+        `
+        UPDATE "products"
+        SET "emissions" = $1
+        WHERE "productHash" = $2
+        `,
+        [emissionsData, productHash]
+    );
+}
+
+const generateEmissionsData = () => {
+    return JSON.stringify([{
+        emissionHash: 'SampleEmissionHash',
+        unit: 'kgeqCO2',
+        CO2e: Math.random() * 100,
+        effectiveDateStart: '2020-01-01',
+        effectiveDateEnd: '2024-12-31',
+        startUsageAmount: 0,
+        endUsageAmount: 100,
+        description: 'Sample Description',
+    }]);
+};
+
+config.logger.info('Starting: loading data into DB');
+run()
+  .then(() => {
+    config.logger.info('Completed: loading data into DB');
+    process.exit(0);
+  })
+  .catch((err) => {
+    config.logger.error(err);
+    process.exit(1);
+  });

src/db/query.ts

@@ -1,6 +1,6 @@
 // In order to make upserting more efficient, prices in postgres are stored as a map of priceHash -> prices.
 import format from 'pg-format';
-import { Price, Product, ProductAttributes } from './types';
+import { Emission, Price, Product, ProductAttributes } from './types';
 import config from '../config';
 
 type AttributeFilter = {
@@ -10,7 +10,7 @@ type AttributeFilter = {
   value_regex?: string;
 };
 
-type ProductWithPriceMap = {
+type ProductWithPriceAndEmissionsMap = {
   productHash: string;
   sku: string;
   vendorName: string;
@@ -19,10 +19,11 @@ type ProductWithPriceMap = {
   productFamily: string;
   attributes: ProductAttributes;
   prices: { [priceHash: string]: Price[] };
+  emissions: { [emissionHash: string]: Emission[] };
 };
 
-function flattenPrices(p: ProductWithPriceMap): Product {
-  return { ...p, prices: Object.values(p.prices).flat() };
+function flattenPrices(p: ProductWithPriceAndEmissionsMap): Product {
+  return { ...p, prices: Object.values(p.prices).flat(), emissions: Object.values(p.emissions).flat() };
 }
 
 // eslint-disable-next-line import/prefer-default-export
@@ -53,7 +54,7 @@ export async function findProducts(
   }
 
   const response = await pool.query(sql);
-  const products = response.rows as ProductWithPriceMap[];
+  const products = response.rows as ProductWithPriceAndEmissionsMap[];
   return products.map((product) => flattenPrices(product));
 }
 

src/db/setup.ts

@@ -18,9 +18,10 @@ export async function createProductsTable(
         service text NOT NULL,
         "productFamily" text DEFAULT ''::text NOT NULL,
         attributes jsonb NOT NULL,
-        prices jsonb NOT NULL, 
+        prices jsonb NOT NULL,
+        emissions jsonb NOT NULL DEFAULT '[]'::jsonb,
         CONSTRAINT %I PRIMARY KEY("productHash")
-      )   
+      )
     `,
       tableName,
       `${tableName}_pkey`
@@ -72,7 +73,7 @@ export async function createStatsTable(
         total_runs bigint DEFAULT 0,
         ci_runs bigint DEFAULT 0,
         non_ci_runs bigint DEFAULT 0
-      )   
+      )
     `,
       tableName
     )
@@ -101,7 +102,7 @@ export async function createInstallsTable(
       (
         install_id uuid PRIMARY KEY NOT NULL,
         created_at timestamp DEFAUlT NOW() NOT NULL
-      )   
+      )
     `,
       tableName
     )

src/db/types.ts

@@ -9,6 +9,7 @@ export type Product = {
   productFamily: string;
   attributes: ProductAttributes;
   prices: Price[];
+  emissions: Emission[];
 };
 
 export type Price = {

src/resolvers.ts

@@ -82,18 +82,12 @@ const getResolvers = <TContext>(
       return prices;
     },
     emissions: async (product: Product, args: EmissionsArgs): Promise<Emission[]> => {
-      const emission = {
-        emissionHash: 'sampleEmissionHash',
-        unit: 'kgeqCO2',
-        CO2e: '10',
-        effectiveDateStart: '2021-01-01',
-        effectiveDateEnd: '2021-12-31',
-        startUsageAmount: '0',
-        endUsageAmount: '100',
-        description: 'test',
-      };
-
-      return [emission];
+
+      const emissions = mingo
+        .find(product.emissions, transformFilter(args.filter))
+        .all() as Emission[];
+
+        return emissions;
     },
   },
   Price:

src/scrapers/awsBulk.ts

@@ -207,6 +207,7 @@ function parseProduct(productJson: ProductJson) {
     sku: productJson.sku,
     attributes: productJson.attributes,
     prices: [],
+    emissions: [],
   };
 
   product.productHash = generateProductHash(product);

src/scrapers/azureRetail.ts

@@ -139,6 +139,7 @@ function parseProduct(productJson: ProductJson): Product {
       meterName: productJson.meterName,
     },
     prices: [],
+    emissions: [],
   };
 
   product.productHash = generateProductHash(product);

src/scrapers/gcpCatalog.ts

@@ -191,6 +191,7 @@ function parseProduct(productJson: ProductJson, region: string): Product {
       resourceGroup: productJson.category.resourceGroup,
     },
     prices: [],
+    emissions: [],
   };
 
   product.productHash = generateProductHash(product);

src/scrapers/gcpMachineTypes.ts

@@ -129,6 +129,7 @@ async function scrape(): Promise<void> {
           machineType: name,
         },
         prices: [],
+        emissions: [],
       };
 
       product.productHash = generateProductHash(product);