feat: implement basic frame of eos action watcher

server/package.json

@@ -25,6 +25,9 @@
     "@nestjs/platform-express": "^6.0.0",
     "@nestjs/typeorm": "^6.1.1",
     "class-validator": "^0.9.1",
+    "demux": "^4.0.0",
+    "demux-eos": "4.0.1",
+    "eosjs": "^20.0.0",
     "nestjs-config": "^1.4.0",
     "nestjs-typeorm-paginate": "^0.1.7",
     "reflect-metadata": "^0.1.12",

server/src/config/watcher.ts

@@ -0,0 +1,7 @@
+export default {
+    endPoint: 'http://211.195.229.86:8888',
+    irreversible: 'false',
+    startAt: 3136735,
+    stopAt: 0,
+    contract: 'ghostghost12',
+};

server/src/port/adapter/service/blockchain/eos/watcher/actionHandler/ObjectActionHandler.ts

@@ -0,0 +1,119 @@
+import {
+    AbstractActionHandler,
+    IndexState,
+    Block,
+    NextBlock,
+    VersionedAction,
+} from 'demux';
+import { State } from '../types/types';
+
+export class ObjectActionHandler extends AbstractActionHandler {
+    constructor([handleVersion]: any, stopAt: number) {
+        super([handleVersion]);
+        this.stopAt = stopAt;
+        this.state = {
+            trx_id: '',
+            indexState: {
+                blockNumber: 0,
+                blockHash: '',
+                isReplay: false,
+                handlerVersionName: 'v1',
+            },
+        };
+    }
+    public stopService = (blockNumber: number) => {
+        // Function stop the service when meet the stopAt block number
+        if (blockNumber >= this.stopAt) {
+            // console.log('\n####################\n# STOP AT: ', blockNumber);
+            // console.log('####################\n');
+            process.exit(1);
+        }
+    }
+
+    private state: State;
+    private hashHistory: { [key: number]: string } = { 0: '' };
+    private stopAt: number;
+
+    get _handlerVersionName() {
+        return this.handlerVersionName;
+    }
+
+    // tslint:disable-next-line
+    public async handleWithState(handle: (state: any) => void) {
+        try {
+            await handle(this.state);
+        } catch (err) {
+            throw new Error('handle state err');
+        }
+    }
+
+    public async rollbackTo(blockNumber: number) {
+        this.setLastProcessedBlockNumber(blockNumber);
+        this.setLastProcessedBlockHash(this.hashHistory[blockNumber]);
+        this.state.indexState = {
+            ...this.state.indexState,
+            blockNumber,
+            blockHash: this.hashHistory[blockNumber],
+        };
+    }
+
+    public setLastProcessedBlockHash(hash: string) {
+        this.lastProcessedBlockHash = hash;
+    }
+
+    public setLastProcessedBlockNumber(num: number) {
+        this.lastProcessedBlockNumber = num;
+    }
+
+    public async _applyUpdaters(
+        state: any,
+        block: Block,
+        context: any,
+        isReplay: boolean,
+    ): Promise<VersionedAction[]> {
+        return this.applyUpdaters(state, block, context, isReplay);
+    }
+
+    public _runEffects(
+        versionedActions: VersionedAction[],
+        context: any,
+        nextBlock: NextBlock,
+    ) {
+        this.runEffects(versionedActions, context, nextBlock);
+    }
+
+    protected async loadIndexState(): Promise<IndexState> {
+        return this.state.indexState;
+    }
+
+    public async handleBlock(
+        nextBlock: NextBlock,
+        isReplay: boolean,
+    ): Promise<number | null> {
+        const { blockNumber, blockHash } = nextBlock.block.blockInfo;
+        this.hashHistory[blockNumber] = blockHash;
+        return super.handleBlock(nextBlock, isReplay);
+    }
+
+    protected async updateIndexState(
+        state: any,
+        block: Block,
+        isReplay: boolean,
+        handlerVersionName: string,
+    ) {
+        // console.log("Processing block: ", block.blockInfo.blockNumber);
+        const { blockNumber, blockHash } = block.blockInfo;
+
+        state.indexState = {
+            blockNumber,
+            blockHash,
+            isReplay,
+            handlerVersionName,
+        };
+        if (this.stopAt) {
+            this.stopService(blockNumber);
+        }
+    }
+
+    protected async setup(): Promise<void> {}
+}

server/src/port/adapter/service/blockchain/eos/watcher/actionHandler/handlerVersions/v1/index.ts

@@ -0,0 +1,190 @@
+import * as State from '../../../types/types';
+import { BlockInfo } from 'demux';
+
+const account = process.env.CONTRACT;
+
+const parseTokenString = (
+    tokenString: string,
+): { amount: number; symbol: string } => {
+    const [amountString, symbol] = tokenString.split(' ');
+    const amount = parseFloat(amountString);
+    return { amount, symbol };
+};
+
+const updateCreateEggData = (
+    state: State.CreateEggState,
+    payload: any,
+    blockInfo: BlockInfo,
+    context: any,
+): void => {
+    state.gene = payload.data.gene;
+    state.owner = payload.data.owner;
+    state.trx_id = payload.transactionId;
+    state.indexState.blockNumber = blockInfo.blockNumber;
+    state.indexState.blockHash = blockInfo.blockHash;
+
+    // database save
+
+    context.stateCopy = JSON.parse(JSON.stringify(state));
+};
+
+const updateSendData = (
+    state: State.SendState,
+    payload: any,
+    blockInfo: BlockInfo,
+    context: any,
+): void => {
+    state.from = payload.data.from;
+    state.to = payload.data.to;
+    state.gene = payload.data.gene;
+    state.trx_id = payload.transactionId;
+    state.indexState.blockNumber = blockInfo.blockNumber;
+    state.indexState.blockHash = blockInfo.blockHash;
+
+    // database save
+
+    context.stateCopy = JSON.parse(JSON.stringify(state));
+};
+
+const updateLevelUpData = (
+    state: State.LevelUpState,
+    payload: any,
+    blockInfo: BlockInfo,
+    context: any,
+): void => {
+    state.owner = payload.data.owner;
+    state.gene = payload.data.gene;
+    state.level = payload.data.level;
+    state.trx_id = payload.transactionId;
+    state.indexState.blockNumber = blockInfo.blockNumber;
+    state.indexState.blockHash = blockInfo.blockHash;
+
+    // database save
+
+    context.stateCopy = JSON.parse(JSON.stringify(state));
+};
+
+const updateAuctionData = (
+    state: State.AuctionState,
+    payload: any,
+    blockInfo: BlockInfo,
+    context: any,
+): void => {
+    const { amount, symbol } = parseTokenString(payload.data.min_price);
+    state.auctioneer = payload.data.auctioneer;
+    state.gene = payload.data.gene;
+    state.minPrice = amount;
+    state.deadline = payload.data.sec + (Date.now() / 1000);
+    state.trx_id = payload.transactionId;
+    state.indexState.blockNumber = blockInfo.blockNumber;
+    state.indexState.blockHash = blockInfo.blockHash;
+
+    // database save
+
+    context.stateCopy = JSON.parse(JSON.stringify(state));
+};
+
+const updateBidData = (
+    state: State.BidState,
+    payload: any,
+    blockInfo: BlockInfo,
+    context: any,
+): void => {
+    const { amount, symbol } = parseTokenString(payload.data.bid);
+    state.bidder = payload.data.bidder;
+    state.gene = payload.data.gene;
+    state.bid = amount;
+    state.trx_id = payload.transactionId;
+    state.indexState.blockNumber = blockInfo.blockNumber;
+    state.indexState.blockHash = blockInfo.blockHash;
+
+    // database save
+
+    context.stateCopy = JSON.parse(JSON.stringify(state));
+};
+
+const updateClaimData = (
+    state: State.ClaimState,
+    payload: any,
+    blockInfo: BlockInfo,
+    context: any,
+): void => {
+    state.requester = payload.data.requester;
+    state.gene = payload.data.gene;
+    state.trx_id = payload.transactionId;
+    state.indexState.blockNumber = blockInfo.blockNumber;
+    state.indexState.blockHash = blockInfo.blockHash;
+
+    // database save
+
+    context.stateCopy = JSON.parse(JSON.stringify(state));
+};
+
+const updaters = [
+    {
+        actionType: `${account}::createegg`,
+        apply: updateCreateEggData,
+    },
+    {
+        actionType: `${account}::send`,
+        apply: updateSendData,
+    },
+    {
+        actionType: `${account}::levelup`,
+        apply: updateLevelUpData,
+    },
+    {
+        actionType: `${account}::auction`,
+        apply: updateAuctionData,
+    },
+    {
+        actionType: `${account}::bid`,
+        apply: updateBidData,
+    },
+    {
+        actionType: `${account}::claim`,
+        apply: updateClaimData,
+    },
+];
+
+const logUpdate = (payload: any, blockInfo: BlockInfo, context: any): void => {
+    // console.info(
+    //     'State updated:\n',
+    //     JSON.stringify(context.stateCopy, null, 2),
+    // );
+};
+
+const effects = [
+    {
+        actionType: `${account}::createegg`,
+        run: logUpdate,
+    },
+    {
+        actionType: `${account}::send`,
+        run: logUpdate,
+    },
+    {
+        actionType: `${account}::levelup`,
+        run: logUpdate,
+    },
+    {
+        actionType: `${account}::auction`,
+        run: logUpdate,
+    },
+    {
+        actionType: `${account}::bid`,
+        run: logUpdate,
+    },
+    {
+        actionType: `${account}::claim`,
+        run: logUpdate,
+    },
+];
+
+const handlerVersion = {
+    versionName: 'v1',
+    updaters,
+    effects,
+};
+
+export default handlerVersion;

server/src/port/adapter/service/blockchain/eos/watcher/test/config/watcher.ts

@@ -0,0 +1,7 @@
+export default {
+    endPoint: 'http://211.195.229.86:8888',
+    irreversible: 'false',
+    startAt: 3136735,
+    stopAt: 0,
+    contract: 'ghostghost12',
+};

server/src/port/adapter/service/blockchain/eos/watcher/types/types.d.ts

@@ -0,0 +1,45 @@
+export interface State {
+    trx_id: string;
+    indexState: {
+        blockNumber: number;
+        blockHash: string;
+        isReplay: boolean;
+        handlerVersionName: string;
+    };
+}
+export interface CreateEggState extends State {
+    gene: string;
+    owner: string;
+}
+export interface SendState extends State {
+    from: string;
+    to: string;
+    gene: string;
+}
+export interface LevelUpState extends State {
+    owner: string;
+    gene: string;
+    level: number;
+}
+export interface AuctionState extends State {
+    auctioneer: string;
+    gene: string;
+    minPrice: number;
+    deadline: number;
+}
+export interface BidState extends State {
+    bidder: string;
+    gene: string;
+    bid: number;
+}
+export interface ClaimState extends State {
+    requester: string;
+    gene: string;
+}
+export interface NodeosActionReaderOptions extends ActionReaderOptions {
+    nodeosEndpoint?: string;
+}
+export interface ActionReaderOptions {
+    startAtBlock?: number;
+    onlyIrreversible?: boolean;
+}