-
Notifications
You must be signed in to change notification settings - Fork 46
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
ADD: mempool processor & push notifications
- Loading branch information
1 parent
baaae67
commit 432c450
Showing
9 changed files
with
246 additions
and
8 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,3 +1,4 @@ | ||
| web: npm start | ||
| worker: npm run worker-blockprocessor | ||
| worker2: npm run worker-sender | ||
| worker3: npm run worker-processmempool |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,151 @@ | ||
| import "./openapi/api"; | ||
| import "reflect-metadata"; | ||
| import { createConnection, getRepository, Repository } from "typeorm"; | ||
| import { TokenToAddress } from "./entity/TokenToAddress"; | ||
| import { SendQueue } from "./entity/SendQueue"; | ||
| import { KeyValue } from "./entity/KeyValue"; | ||
| require("dotenv").config(); | ||
| const url = require("url"); | ||
| let jayson = require("jayson/promise"); | ||
| let rpc = url.parse(process.env.BITCOIN_RPC); | ||
| let client = jayson.client.http(rpc); | ||
|
|
||
| let processedTxids = {}; | ||
| const parsed = url.parse(process.env.JAWSDB_MARIA_URL); | ||
| if (!process.env.JAWSDB_MARIA_URL || !process.env.BITCOIN_RPC) { | ||
| console.error("not all env variables set"); | ||
| process.exit(); | ||
| } | ||
|
|
||
| process | ||
| .on("unhandledRejection", (reason, p) => { | ||
| console.error(reason, "Unhandled Rejection at Promise", p); | ||
| process.exit(1); | ||
| }) | ||
| .on("uncaughtException", (err) => { | ||
| console.error(err, "Uncaught Exception thrown"); | ||
| process.exit(1); | ||
| }); | ||
|
|
||
| let sendQueueRepository: Repository<SendQueue>; | ||
|
|
||
| async function processMempool() { | ||
| console.log("cached txids=", Object.keys(processedTxids).length); | ||
| const responseGetrawmempool = await client.request("getrawmempool", []); | ||
| console.log(responseGetrawmempool.result.length, "txs in mempool"); | ||
|
|
||
| let addresses: string[] = []; | ||
| let allPotentialPushPayloadsArray: Components.Schemas.PushNotificationOnchainAddressGotUnconfirmedTransaction[] = []; | ||
|
|
||
| let rpcBatch = []; | ||
| const batchSize = 100; | ||
| let countTxidsProcessed = 0; | ||
| for (const txid of responseGetrawmempool.result) { | ||
| countTxidsProcessed++; | ||
| if (!txid) continue; | ||
| if (!processedTxids[txid]) rpcBatch.push(client.request("getrawtransaction", [txid, true], undefined, false)); | ||
| if (rpcBatch.length >= batchSize || countTxidsProcessed === responseGetrawmempool.result.length) { | ||
| const startBatch = +new Date(); | ||
| // got enough txids lets batch fetch them from bitcoind rpc | ||
| const responses = await client.request(rpcBatch); | ||
| for (const response of responses) { | ||
| if (response.result && response.result.vout) { | ||
| for (const output of response.result.vout) { | ||
| if (output.scriptPubKey && output.scriptPubKey.addresses) { | ||
| for (const address of output.scriptPubKey.addresses) { | ||
| addresses.push(address); | ||
| processedTxids[response.result.txid] = true; | ||
| const payload: Components.Schemas.PushNotificationOnchainAddressGotUnconfirmedTransaction = { | ||
| address, | ||
| txid: response.result.txid, | ||
| sat: Math.floor(output.value * 100000000), | ||
| type: 3, | ||
| token: "", | ||
| os: "ios", | ||
| }; | ||
| allPotentialPushPayloadsArray.push(payload); | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // allPotentialPushPayloadsArray.push({ address: "bc1qaemfnglf928kd9ma2jzdypk333au6ctu7h7led", txid: "666", sat: 1488, type: 3, token: "", os: "ios" }); // debug fixme | ||
| // addresses.push("bc1qaemfnglf928kd9ma2jzdypk333au6ctu7h7led"); // debug fixme | ||
|
|
||
| if (addresses.length === 0) { | ||
| allPotentialPushPayloadsArray = []; | ||
| addresses = []; | ||
| rpcBatch = []; | ||
| continue; | ||
| } | ||
|
|
||
| // fetching found addresses from db: | ||
| const query = getRepository(TokenToAddress).createQueryBuilder().where("address IN (:...address)", { address: addresses }); | ||
| for (const t2a of await query.getMany()) { | ||
| // found all addresses that we are tracking on behalf of our users. now, | ||
| // iterating all addresses in a block to see if there is a match. | ||
| // we could only iterate tracked addresses, but that would imply deduplication which is not good (for example, | ||
| // in a single block user could get several incoming payments to different owned addresses) | ||
| // cycle in cycle is less than optimal, but we can live with that for now | ||
| for (let payload of allPotentialPushPayloadsArray) { | ||
| if (t2a.address === payload.address) { | ||
| console.log("enqueueing", payload); | ||
| payload.os = t2a.os === "android" ? "android" : "ios"; // hacky | ||
| payload.token = t2a.token; | ||
| payload.type = 3; | ||
| await sendQueueRepository.save({ | ||
| data: JSON.stringify(payload), | ||
| }); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| allPotentialPushPayloadsArray = []; | ||
| addresses = []; | ||
| rpcBatch = []; | ||
|
|
||
| const endBatch = +new Date(); | ||
| // process.stdout.write('.'); | ||
| console.log("batch took", (endBatch - startBatch) / 1000, "sec"); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| createConnection({ | ||
| type: "mariadb", | ||
| host: parsed.hostname, | ||
| port: parsed.port, | ||
| username: parsed.auth.split(":")[0], | ||
| password: parsed.auth.split(":")[1], | ||
| database: parsed.path.replace("/", ""), | ||
| synchronize: true, | ||
| logging: false, | ||
| entities: ["src/entity/**/*.ts"], | ||
| migrations: ["src/migration/**/*.ts"], | ||
| subscribers: ["src/subscriber/**/*.ts"], | ||
| cli: { | ||
| entitiesDir: "src/entity", | ||
| migrationsDir: "src/migration", | ||
| subscribersDir: "src/subscriber", | ||
| }, | ||
| }) | ||
| .then(async (connection) => { | ||
| // start worker | ||
| console.log("running"); | ||
|
|
||
| sendQueueRepository = getRepository(SendQueue); | ||
|
|
||
| while (1) { | ||
| const start = +new Date(); | ||
| try { | ||
| await processMempool(); | ||
| } catch (error) { | ||
| console.log(error); | ||
| } | ||
| const end = +new Date(); | ||
| console.log("processing mempool took", (end - start) / 1000, "sec"); | ||
| console.log("-----------------------"); | ||
| } | ||
| }) | ||
| .catch((error) => console.log(error)); |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters