Feat/atomic data publisher

Cargo.toml

@@ -19,6 +19,7 @@ exclude = [
   "utils/generate-test-gateway-txn",
   "utils/standardize-hotspot-metadata",
   "utils/pyth_solana_receiver_sdk",
+  "utils/atomic-data-publisher",
 ]
 
 [workspace.dependencies]

packages/account-postgres-sink-service/src/constants.ts

@@ -1 +1 @@
-export const OMIT_KEYS = ["refreshed_at", "createdAt"];
+export const OMIT_KEYS = ["createdAt", "refreshedAt", "lastBlock"];

packages/account-postgres-sink-service/src/plugins/encodeEntityKey.ts

@@ -36,10 +36,15 @@ export const EncodeEntityKeyPlugin = ((): IPlugin => {
       };
     };
 
-    const processAccount = async (account: { [key: string]: any }) => {
+    const processAccount = async (
+      account: { [key: string]: any },
+      transaction?: any,
+      lastBlock?: number | null
+    ) => {
       try {
         const entityKey = account[camelize(config.field || "entity_key", true)];
-        const keySerializationRaw = account[camelize("key_serialization", true)];
+        const keySerializationRaw =
+          account[camelize("key_serialization", true)];
         const keySerialization =
           typeof keySerializationRaw === "string"
             ? keySerializationRaw.trim().toLowerCase()
@@ -48,7 +53,10 @@ export const EncodeEntityKeyPlugin = ((): IPlugin => {
         if (entityKey && keySerialization) {
           if (keySerialization === "utf8") {
             encodedEntityKey = Buffer.from(entityKey, "utf8").toString("utf8");
-          } else if (keySerialization === "b58" || keySerialization === "bs58") {
+          } else if (
+            keySerialization === "b58" ||
+            keySerialization === "bs58"
+          ) {
             encodedEntityKey = bs58.encode(entityKey);
           }
         }

packages/account-postgres-sink-service/src/plugins/explodeMiniFanoutOwnership.ts

@@ -1,12 +1,11 @@
-import BN from "bn.js";
-import { cellToLatLng } from "h3-js";
 import { camelize } from "inflection";
 import _omit from "lodash/omit";
 import { DataTypes, Model, QueryTypes } from "sequelize";
 import { IPlugin } from "../types";
 import { database } from "../utils/database";
-import { MapboxService } from "../utils/mapboxService";
 import { PublicKey } from "@solana/web3.js";
+import sequelize from "sequelize";
+import { syncTableWithViews } from "../utils/syncTableWithViews";
 
 export class RewardsRecipient extends Model {
   declare asset: string;
@@ -16,10 +15,10 @@ export class RewardsRecipient extends Model {
   declare destination: string;
   declare entityKey: string;
   declare encodedEntityKey: string;
-  declare shares: number
-  declare totalShares: number
-  declare fixedAmount: number
-  declare type: 'direct' | 'fanout'
+  declare shares: number;
+  declare totalShares: number;
+  declare fixedAmount: number;
+  declare type: "direct" | "fanout";
 }
 
 RewardsRecipient.init(
@@ -47,7 +46,7 @@ RewardsRecipient.init(
       field: "encoded_entity_key",
     },
     keySerialization: {
-      type: DataTypes.STRING,
+      type: "TEXT",
       allowNull: false,
     },
     shares: {
@@ -66,6 +65,11 @@ RewardsRecipient.init(
       type: DataTypes.STRING,
       allowNull: false,
     },
+    lastBlock: {
+      type: DataTypes.DECIMAL.UNSIGNED,
+      allowNull: true,
+      defaultValue: null,
+    },
   },
   {
     sequelize: database,
@@ -74,40 +78,46 @@ RewardsRecipient.init(
     underscored: true,
     timestamps: true,
     indexes: [
-      {
-        fields: ["asset"],
-      },
       {
         fields: ["destination"],
       },
       {
         fields: ["owner"],
       },
       {
-        fields: ["type"],
+        fields: ["last_block"],
+      },
+      {
+        fields: ["encoded_entity_key"],
+      },
+      {
+        fields: ["asset", "last_block"],
+      },
+      {
+        fields: ["type", "last_block"],
       },
     ],
   }
 );
 
 type MiniFanoutShare = {
-  wallet: string,
-  delegate: string,
-  share: Share,
-  totalDust: number,
-  totalOwed: number,
-}
+  wallet: string;
+  delegate: string;
+  share: Share;
+  totalDust: number;
+  totalOwed: number;
+};
 
 type Share = {
-  share?: { amount: number },
-  fixed?: { amount: number },
-}
+  share?: { amount: number };
+  fixed?: { amount: number };
+};
 
 export class Recipient extends Model {
-  declare address: string
-  declare asset: string
-  declare destination: string
-  declare lazyDistributor: string
+  declare address: string;
+  declare asset: string;
+  declare destination: string;
+  declare lazyDistributor: string;
 }
 
 Recipient.init(
@@ -133,15 +143,15 @@ Recipient.init(
     underscored: true,
     timestamps: false,
   }
-)
+);
 
 export class KeyToAsset extends Model {
-  declare address: string
-  declare asset: string
-  declare dao: string
-  declare entityKey: Buffer
-  declare keySerialization: string
-  declare encodedEntityKey: string
+  declare address: string;
+  declare asset: string;
+  declare dao: string;
+  declare entityKey: Buffer;
+  declare keySerialization: string;
+  declare encodedEntityKey: string;
 }
 
 KeyToAsset.init(
@@ -152,12 +162,13 @@ KeyToAsset.init(
     },
     asset: {
       type: DataTypes.STRING,
+      allowNull: false,
     },
     dao: {
       type: DataTypes.STRING,
     },
     keySerialization: {
-      type: DataTypes.STRING,
+      type: DataTypes.JSONB,
     },
     entityKey: {
       type: "BYTEA",
@@ -177,7 +188,7 @@ KeyToAsset.init(
     underscored: true,
     timestamps: false,
   }
-)
+);
 
 export class MiniFanout extends Model {
   declare owner: string;
@@ -262,29 +273,44 @@ MiniFanout.init(
   }
 );
 
-export const HNT_LAZY_DISTRIBUTOR = "6gcZXjHgKUBMedc2V1aZLFPwh8M1rPVRw7kpo2KqNrFq"
+export const HNT_LAZY_DISTRIBUTOR =
+  "6gcZXjHgKUBMedc2V1aZLFPwh8M1rPVRw7kpo2KqNrFq";
 
-export async function handleMiniFanout(asset: string, account: { [key: string]: any }, transaction: any) {
-  const prevAccount = await MiniFanout.findByPk(account.address, { transaction })
-  const oldShares = prevAccount?.shares || []
-  const newShares = (account.shares || []) as MiniFanoutShare[]
+export async function handleMiniFanout(
+  asset: string,
+  account: { [key: string]: any },
+  transaction: any,
+  lastBlock?: number | null
+) {
+  const prevAccount = await MiniFanout.findByPk(account.address, {
+    transaction,
+  });
+  const oldShares = prevAccount?.shares || [];
+  const newShares = (account.shares || []) as MiniFanoutShare[];
 
   function getEffectiveDestination(share: MiniFanoutShare) {
-    return (!share.delegate || share.delegate === PublicKey.default.toBase58()) ? share.wallet : share.delegate
+    return !share.delegate || share.delegate === PublicKey.default.toBase58()
+      ? share.wallet
+      : share.delegate;
   }
 
   function getShareKey(share: MiniFanoutShare) {
-    return `${asset}-${getEffectiveDestination(share)}-${JSON.stringify(share.share)}`
+    return `${asset}-${getEffectiveDestination(share)}-${JSON.stringify(
+      share.share
+    )}`;
   }
   // Create a map of wallet+delegate to share for easy lookup
   const oldSharesMap = new Map(
-    oldShares.map(share => [getShareKey(share), share])
-  )
+    oldShares.map((share) => [getShareKey(share), share])
+  );
   const newSharesMap = new Map(
-    newShares.map(share => [getShareKey(share), share])
-  )
+    newShares.map((share) => [getShareKey(share), share])
+  );
 
-  const totalShares = newShares.reduce((acc, share) => acc + (share.share?.share?.amount || 0), 0)
+  const totalShares = newShares.reduce(
+    (acc, share) => acc + (share.share?.share?.amount || 0),
+    0
+  );
 
   // Handle deletions - remove shares that exist in old but not in new
   for (const [key, oldShare] of oldSharesMap) {
@@ -295,20 +321,20 @@ export async function handleMiniFanout(asset: string, account: { [key: string]:
           asset,
           shares: oldShare.share?.share?.amount || 0,
         },
-        transaction
-      })
+        transaction,
+      });
     }
   }
 
   // Handle updates and additions
   for (const [key, newShare] of newSharesMap) {
     const kta = await KeyToAsset.findOne({
       where: {
-        dao: 'BQ3MCuTT5zVBhNfQ4SjMh3NPVhFy73MPV8rjfq5d1zie',
+        dao: "BQ3MCuTT5zVBhNfQ4SjMh3NPVhFy73MPV8rjfq5d1zie",
         asset: asset,
       },
-      transaction
-    })
+      transaction,
+    });
 
     const toCreate = {
       asset,
@@ -320,20 +346,20 @@ export async function handleMiniFanout(asset: string, account: { [key: string]:
       entityKey: kta?.entityKey,
       encodedEntityKey: kta?.encodedEntityKey,
       keySerialization: kta?.keySerialization,
-      type: 'fanout'
-    }
+      type: "fanout",
+      lastBlock,
+    };
 
-    await RewardsRecipient.upsert(toCreate, { transaction })
+    await RewardsRecipient.upsert(toCreate, { transaction });
   }
 
-  return account
+  return account;
 }
 
 export const ExplodeMiniFanoutOwnershipPlugin = ((): IPlugin => {
   const name = "ExplodeMiniFanoutOwnership";
   const init = async (config: { [key: string]: any }) => {
     const updateOnDuplicateFields: string[] = [];
-
     const existingColumns = (
       await database.query(
         `
@@ -353,34 +379,48 @@ export const ExplodeMiniFanoutOwnershipPlugin = ((): IPlugin => {
       !existingColumns.length ||
       !columns.every((col) => existingColumns.includes(col))
     ) {
-      await RewardsRecipient.sync({ alter: true });
+      console.log("Syncing rewards_recipients table");
+      await syncTableWithViews(
+        database,
+        "rewards_recipients",
+        "public",
+        async () => {
+          await RewardsRecipient.sync({ alter: true });
+        }
+      );
     }
 
-    const addFields = () => { };
+    const addFields = () => {};
 
     const processAccount = async (
       account: { [key: string]: any },
-      transaction?: any
+      transaction?: any,
+      lastBlock?: number | null
     ) => {
       try {
-        const asset = account.preTask?.remoteV0?.url?.replace("https://hnt-rewards.oracle.helium.io/v1/tuktuk/asset/", "").replace("https://hnt-rewards.oracle.test-helium.com/v1/tuktuk/asset/", "")
+        const asset = account.preTask?.remoteV0?.url
+          ?.replace("https://hnt-rewards.oracle.helium.io/v1/tuktuk/asset/", "")
+          .replace(
+            "https://hnt-rewards.oracle.test-helium.com/v1/tuktuk/asset/",
+            ""
+          );
         if (!asset) {
-          return account
+          return account;
         }
         const recipient = await Recipient.findOne({
           where: {
             destination: account.address,
             asset,
-            lazyDistributor: HNT_LAZY_DISTRIBUTOR
-          }
-        })
+            lazyDistributor: HNT_LAZY_DISTRIBUTOR,
+          },
+        });
         if (!recipient) {
-          return account
+          return account;
         }
-        return handleMiniFanout(asset, account, transaction)
+        return handleMiniFanout(asset, account, transaction, lastBlock);
       } catch (err) {
-        console.error("Error exploding mini fanout ownership", err)
-        throw err
+        console.error("Error exploding mini fanout ownership", err);
+        throw err;
       }
     };