feat(#451): Create cloudfront invalidator (#452)

* feat(#441): Upload entities to s3 for faster api

* Init impl of cloudfront validator

* Fix envs

* Add leading slash

* Fix test

* Rm yarn lock

* Fix build

docker-info.json

@@ -12,7 +12,8 @@
     "account-postgres-sink-service": "./packages/account-postgres-sink-service",
     "iot-premine-data-only-service": "./packages/iot-premine-data-only-service",
     "crons": "./packages/crons",
-    "rewards-oracle-faucet-service": "./packages/rewards-oracle-faucet-service"
+    "rewards-oracle-faucet-service": "./packages/rewards-oracle-faucet-service",
+    "entity-invalidator": "./packages/entity-invalidator"
   },
   "oracle": {
     "distributor-oracle": "./packages/distributor-oracle",

packages/entity-invalidator/.gitignore

@@ -0,0 +1,4 @@
+npm-debug.log
+dist/
+tmp/
+./node_modules

packages/entity-invalidator/Dockerfile

@@ -0,0 +1,32 @@
+# Specify the base image
+FROM node:18-alpine AS BUILD_IMAGE
+
+WORKDIR /usr/src/app
+
+COPY package.json ./
+COPY yarn.deploy.lock ./
+ENV YARN_LOCKFILE_FILENAME=yarn.deploy.lock
+
+RUN corepack enable
+RUN yarn set version stable
+RUN yarn install
+
+COPY src src
+COPY tsconfig.build.json tsconfig.json
+
+RUN yarn run build
+RUN npm prune --production
+
+FROM node:18-alpine
+
+WORKDIR /usr/src/app
+
+COPY --from=BUILD_IMAGE /usr/src/app/lib ./lib
+COPY --from=BUILD_IMAGE /usr/src/app/node_modules ./node_modules
+
+# This isn't actually used, service is read only. But anchor wants a wallet.
+RUN echo "[124,96,181,146,132,165,175,182,60,194,167,230,29,91,110,109,226,38,41,155,207,186,24,33,205,120,108,98,218,67,77,95,13,60,79,204,253,10,183,101,60,94,220,177,117,97,16,29,31,124,35,65,121,147,161,114,159,23,207,202,122,164,170,201]" > id.json
+
+ENV ANCHOR_WALLET=/usr/src/app/id.json
+
+CMD ["node", "lib/src/index.js"]

packages/entity-invalidator/package.json

@@ -0,0 +1,60 @@
+{
+  "name": "@helium/entity-invalidator",
+  "private": true,
+  "publishConfig": {
+    "access": "public",
+    "registry": "https://registry.npmjs.org/"
+  },
+  "license": "Apache-2.0",
+  "version": "0.4.0",
+  "description": "Sync account data to postgres",
+  "repository": {
+    "type": "git",
+    "url": "https://github.com/helium/helium-program-libary"
+  },
+  "main": "./lib/cjs/index.js",
+  "module": "./lib/esm/src/index.js",
+  "types": "./lib/types/src/index.d.ts",
+  "sideEffects": false,
+  "files": [
+    "lib"
+  ],
+  "exports": {
+    "import": "./lib/esm/src/index.js",
+    "require": "./lib/cjs/index.js",
+    "types": "./lib/types/src/index.d.ts"
+  },
+  "scripts": {
+    "format": "prettier --write \"src/**/*.{ts,tsx}\"",
+    "precommit": "npx git-format-staged -f 'prettier --ignore-unknown --stdin --stdin-filepath \"{}\"' .",
+    "build": "tsc -p tsconfig.json",
+    "start": "node lib/esm/server.js",
+    "dev": "npx ts-node --project tsconfig.cjs.json src/index.ts"
+  },
+  "dependencies": {
+    "@coral-xyz/anchor": "^0.28.0",
+    "@helium/account-fetch-cache": "^0.4.0",
+    "@solana/web3.js": "^1.78.4",
+    "aws-sdk": "^2.1344.0",
+    "bn.js": "^5.2.0",
+    "bs58": "^4.0.1",
+    "pg": "^8.9.0",
+    "prom-client": "^14.2.0",
+    "sequelize": "^6.28.0",
+    "yargs": "^17.7.1"
+  },
+  "devDependencies": {
+    "@types/bn.js": "^5.1.1",
+    "@types/deep-equal": "^1.0.1",
+    "@types/node": "^18.11.11",
+    "@types/pg": "^8.6.6",
+    "@types/yargs": "^17.0.24",
+    "git-format-staged": "^2.1.3",
+    "ts-loader": "^9.2.3",
+    "ts-node": "^10.9.1",
+    "ts-node-dev": "^2.0.0",
+    "typescript": "^5.2.2"
+  },
+  "keywords": [],
+  "author": ""
+}

packages/entity-invalidator/src/env.ts

@@ -0,0 +1,13 @@
+export const LOOKBACK_HOURS = process.env.LOOKBACK_HOURS
+  ? Number(process.env.LOOKBACK_HOURS)
+  : 25;
+export const AWS_REGION = process.env.AWS_REGION;
+export const CLOUDFRONT_DISTRIBUTION = process.env.CLOUDFRONT_DISTRIBUTION!;
+
+// Check for required environment variables
+const requiredEnvVars = ["AWS_REGION", "CLOUDFRONT_DISTRIBUTION"];
+for (const varName of requiredEnvVars) {
+  if (!process.env[varName]) {
+    throw new Error(`Environment variable ${varName} is required`);
+  }
+}

packages/entity-invalidator/src/index.ts

@@ -0,0 +1,118 @@
+import { decodeEntityKey } from "@helium/helium-entity-manager-sdk";
+import AWS from "aws-sdk";
+import { Op } from "sequelize";
+import { IotHotspotInfo, KeyToAsset, MobileHotspotInfo } from "./model";
+// @ts-ignore
+import { AWS_REGION, CLOUDFRONT_DISTRIBUTION, LOOKBACK_HOURS } from "./env";
+
+async function run() {
+  const date = new Date();
+  date.setHours(date.getHours() - LOOKBACK_HOURS);
+  AWS.config.update({ region: AWS_REGION });
+  const cloudfront = new AWS.CloudFront();
+
+  const limit = 10000;
+  let lastId = null;
+  let entities;
+
+  const lastIdWhere = {};
+  const whereClause = {
+    [Op.or]: [
+      {
+        refreshedAt: {
+          [Op.gt]: date,
+        },
+      },
+      {
+        "$iot_hotspot_info.refreshed_at$": {
+          [Op.gt]: date,
+        },
+      },
+      {
+        "$mobile_hotspot_info.refreshed_at$": {
+          [Op.gt]: date,
+        },
+      },
+    ],
+  };
+  const totalCount = await KeyToAsset.count({
+    where: whereClause,
+    include: [
+      {
+        model: IotHotspotInfo,
+        required: false,
+      },
+      {
+        model: MobileHotspotInfo,
+        required: false,
+      },
+    ],
+  });
+  console.log(`Found ${totalCount} updated records`);
+  let totalProgress = 0;
+
+  do {
+    if (lastId) {
+      lastIdWhere["address"] = {
+        [Op.gt]: lastId,
+      };
+    }
+    entities = await KeyToAsset.findAll({
+      where: {
+        [Op.and]: [lastIdWhere, whereClause],
+      },
+      include: [
+        {
+          model: IotHotspotInfo,
+          required: false,
+        },
+        {
+          model: MobileHotspotInfo,
+          required: false,
+        },
+      ],
+      limit: limit,
+      order: [["address", "ASC"]],
+    });
+
+    if (entities.length) {
+      entities.forEach((entity) => {
+        entity.entityKeyStr = decodeEntityKey(entity.entityKey, {
+          [entity.keySerialization.toString()]: {},
+        });
+      });
+
+      const paths = entities.flatMap((entity) => getPaths(entity));
+      await cloudfront
+        .createInvalidation({
+          DistributionId: CLOUDFRONT_DISTRIBUTION,
+          InvalidationBatch: {
+            CallerReference: `${new Date().getTime()}`, // unique identifier for this invalidation batch
+            Paths: {
+              Quantity: paths.length,
+              Items: paths,
+            },
+          },
+        })
+        .promise();
+
+      lastId = entities[entities.length - 1].id;
+      totalProgress += entities.length;
+      console.log(`Processed ${totalProgress} / ${totalCount}`);
+    }
+  } while (entities.length === limit);
+}
+
+function getPaths(entity: KeyToAsset): string[] {
+  const v1 = `/v1/${entity.address}`;
+  if ((entity.entityKeyStr?.length || 0) >= 200) {
+    return [v1];
+  }
+
+  return [`/${entity.entityKeyStr!}`, v1];
+}
+
+run().catch((e) => {
+  console.error(e);
+  process.exit(1);
+});

packages/entity-invalidator/src/model.ts

@@ -0,0 +1,155 @@
+import { Sequelize, STRING, Model, DataTypes } from "sequelize";
+import AWS from "aws-sdk";
+import * as pg from "pg";
+
+const host = process.env.PGHOST || "localhost";
+const port = Number(process.env.PGPORT) || 5432;
+export const sequelize = new Sequelize({
+  host: host,
+  dialect: "postgres",
+  port: port,
+  logging: false,
+  dialectModule: pg,
+  username: process.env.PGUSER,
+  database: process.env.PGDATABASE,
+  pool: {
+    max: 5,
+    min: 0,
+    acquire: 30000,
+    idle: 10000,
+  },
+  hooks: {
+    beforeConnect: async (config: any) => {
+      const isRds = host.includes("rds.amazonaws.com");
+
+      let password = process.env.PGPASSWORD;
+      if (isRds && !password) {
+        const signer = new AWS.RDS.Signer({
+          region: process.env.AWS_REGION,
+          hostname: process.env.PGHOST,
+          port,
+          username: process.env.PGUSER,
+        });
+        password = await new Promise((resolve, reject) =>
+          signer.getAuthToken({}, (err, token) => {
+            if (err) {
+              return reject(err);
+            }
+            resolve(token);
+          })
+        );
+        config.dialectOptions = {
+          ssl: {
+            require: false,
+            rejectUnauthorized: false,
+          },
+        };
+      }
+      config.password = password;
+    },
+  },
+});
+
+export class MobileHotspotInfo extends Model {
+  declare address: string;
+  declare asset: string;
+  declare city: string;
+  declare state: string;
+  declare country: string;
+}
+MobileHotspotInfo.init(
+  {
+    address: {
+      type: STRING,
+      primaryKey: true,
+    },
+    city: DataTypes.STRING,
+    state: DataTypes.STRING,
+    country: DataTypes.STRING,
+    asset: DataTypes.STRING,
+    refreshedAt: DataTypes.TIME,
+    isFullHotspot: DataTypes.BOOLEAN,
+    numLocationAsserts: DataTypes.NUMBER,
+    isActive: DataTypes.BOOLEAN,
+    dcOnboardingFeePaid: DataTypes.DECIMAL.UNSIGNED,
+    deviceType: DataTypes.JSONB,
+  },
+  {
+    sequelize,
+    modelName: "mobile_hotspot_infos",
+    tableName: "mobile_hotspot_infos",
+    underscored: true,
+    timestamps: false,
+  }
+);
+
+export class IotHotspotInfo extends Model {
+  declare address: string;
+  declare asset: string;
+  declare city: string;
+  declare state: string;
+  declare country: string;
+}
+IotHotspotInfo.init(
+  {
+    address: {
+      type: STRING,
+      primaryKey: true,
+    },
+    asset: DataTypes.STRING,
+    city: DataTypes.STRING,
+    state: DataTypes.STRING,
+    country: DataTypes.STRING,
+    location: DataTypes.DECIMAL.UNSIGNED,
+    isFullHotspot: DataTypes.BOOLEAN,
+    numLocationAsserts: DataTypes.NUMBER,
+    isActive: DataTypes.BOOLEAN,
+    dcOnboardingFeePaid: DataTypes.DECIMAL.UNSIGNED,
+    refreshedAt: DataTypes.TIME,
+  },
+  {
+    sequelize,
+    modelName: "iot_hotspot_infos",
+    tableName: "iot_hotspot_infos",
+    underscored: true,
+    timestamps: false,
+  }
+);
+
+export class KeyToAsset extends Model {
+  declare address: string;
+  declare asset: string;
+  declare entityKey: Buffer;
+  declare entityKeyStr?: string;
+  declare keySerialization: any;
+  declare mobile_hotspot_info?: MobileHotspotInfo;
+  declare iot_hotspot_info?: IotHotspotInfo;
+}
+KeyToAsset.init(
+  {
+    address: {
+      type: STRING,
+      primaryKey: true,
+    },
+    entityKey: DataTypes.BLOB,
+    keySerialization: DataTypes.JSONB,
+    asset: DataTypes.STRING,
+    refreshedAt: DataTypes.TIME,
+  },
+  {
+    sequelize,
+    modelName: "key_to_assets",
+    tableName: "key_to_assets",
+    underscored: true,
+    timestamps: false,
+  }
+);
+
+KeyToAsset.hasOne(IotHotspotInfo, {
+  sourceKey: "asset",
+  foreignKey: "asset",
+});
+KeyToAsset.hasOne(MobileHotspotInfo, {
+  sourceKey: "asset",
+  foreignKey: "asset",
+});