-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Added 20k dev wallets Updated to an sdk like feel Added caching checks Check for existing conversation Moved to weighted batches
- Loading branch information
Alex Risch
authored and
Alex Risch
committed
May 10, 2024
1 parent
a344282
commit 39be7a5
Showing
9 changed files
with
20,300 additions
and
71 deletions.
There are no files selected for viewing
Large diffs are not rendered by default.
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,7 @@ | ||
| import * as fs from "fs"; | ||
|
|
||
| export const getDevWalletAddresses = (): string[] => { | ||
| // Read from addresses.json | ||
| const addresses = JSON.parse(fs.readFileSync("addresses.json", "utf-8")); | ||
| return addresses; | ||
| }; |
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 |
|---|---|---|
| @@ -1,76 +1,81 @@ | ||
| import type { Client } from "@xmtp/xmtp-js"; | ||
| import { | ||
| finishBroadcast, | ||
| updateBroadcastFromBatch, | ||
| incrementBroadcastSent, | ||
| updateBroadcastStatus, | ||
| } from "./broadcasts"; | ||
| import { Broadcast } from "./utils/Broadcast"; | ||
| const XMTP_RATE_LIMIT = 1000; | ||
| const XMTP_RATE_LIMIT_TIME = 60 * 1000; // 1 minute | ||
| const XMTP_RATE_LIMIT_TIME_INCREASE = XMTP_RATE_LIMIT_TIME * 5; // 5 minutes | ||
|
|
||
| const delay = (ms: number) => | ||
| new Promise<void>((resolve) => setTimeout(resolve, ms)); | ||
|
|
||
| let sendCount = 0; | ||
| let errorCount = 0; | ||
| let startTime: number; | ||
| export const startBroadcast = async ( | ||
| client: Client, | ||
| broadcastAddresses: string[], | ||
| message: string, | ||
| broadcastId: string | ||
| ): Promise<void> => { | ||
| const batches: string[][] = []; | ||
| let batch: string[] = []; | ||
| const canMessageAddresses = await client.canMessage(broadcastAddresses); | ||
| let errorCount = 0; | ||
| for (let i = 0; i < canMessageAddresses.length; i++) { | ||
| if (canMessageAddresses[i]) { | ||
| batch.push(broadcastAddresses[i]); | ||
| } | ||
| // Add a batch of 500 addresses to the batches array | ||
| // An introduction message is sent for new contacts, so each new message will actually be 2 messages in this case | ||
| // We want to send 1000 messages per minute, so we split the batches in half | ||
| // Additional optimization can be done to send messages to contacts that have already been introduced | ||
| if (batch.length === XMTP_RATE_LIMIT / 2) { | ||
| batches.push(batch); | ||
| batch = []; | ||
| } | ||
| } | ||
| if (batch.length > 0) { | ||
| batches.push(batch); | ||
| } | ||
|
|
||
| for (let i = 0; i < batches.length; i++) { | ||
| const batch: string[] = []; | ||
| const sentAddresses: string[] = []; | ||
| updateBroadcastStatus(broadcastId, "sending"); | ||
| await Promise.all( | ||
| batches[i].map(async (address, index) => { | ||
| const conversation = await client.conversations.newConversation( | ||
| address | ||
| ); | ||
| try { | ||
| await conversation.send(message); | ||
| console.log( | ||
| `Sent message for batch ${i} index ${index} to ${address}` | ||
| ); | ||
| sentAddresses.push(address); | ||
| } catch (err) { | ||
| errorCount++; | ||
| console.error(err); | ||
| batch.push(address); | ||
| // Add error handling here | ||
| } | ||
| }) | ||
| const onBroadcastComplete = () => { | ||
| let endTime = Date.now(); | ||
| console.log( | ||
| `Broadcast ${broadcastId} completed Total time ${endTime - startTime}ms` | ||
| ); | ||
| updateBroadcastFromBatch(broadcastId, sentAddresses); | ||
| if (i !== batches.length - 1) { | ||
| updateBroadcastStatus(broadcastId, "waiting"); | ||
| // Wait between batches | ||
| console.log(`Waiting between batches ${i} and ${i + 1}`); | ||
| await delay(XMTP_RATE_LIMIT_TIME_INCREASE); | ||
| } | ||
| if (batch.length > 0) { | ||
| batches.push(batch); | ||
| } | ||
| finishBroadcast(broadcastId); | ||
| }; | ||
| const onMessageFailed = (address: string) => { | ||
| errorCount++; | ||
| console.log(`Message failed for address ${errorCount} : ${address}`); | ||
| }; | ||
| const onMessageSent = () => { | ||
| sendCount++; | ||
| incrementBroadcastSent(broadcastId); | ||
| }; | ||
| const onCanMessageAddressesUpdate = (addresses: string[]) => { | ||
| console.log(`Can message addresses updated to ${addresses.length}`); | ||
| }; | ||
| const onBatchComplete = (addresses: string[]) => { | ||
| console.log(`Batch complete for ${addresses.length} addresses`); | ||
| }; | ||
| const onCantMessageAddress = (address: string) => { | ||
| console.log(`Can't message address ${address}`); | ||
| }; | ||
| const onCanMessageAddreses = (addresses: string[]) => { | ||
| console.log(`Can message addresses ${addresses.length}`); | ||
| }; | ||
| const onDelay = (ms: number) => { | ||
| console.log(`Delaying for ${ms}ms`); | ||
| updateBroadcastStatus(broadcastId, "waiting"); | ||
| }; | ||
| const onBatchStart = (addresses: string[]) => { | ||
| console.log(`Batch start for ${addresses.length} addresses`); | ||
| updateBroadcastStatus(broadcastId, "sending"); | ||
| }; | ||
| errorCount = 0; | ||
| sendCount = 0; | ||
| startTime = Date.now(); | ||
| const broadcast = new Broadcast({ | ||
| client, | ||
| addresses: broadcastAddresses, | ||
| cachedCanMessageAddresses: [], | ||
| rateLimitAmount: XMTP_RATE_LIMIT, | ||
| rateLimitTime: XMTP_RATE_LIMIT_TIME_INCREASE, | ||
| onBatchStart, | ||
| onBatchComplete, | ||
| onBroadcastComplete, | ||
| onCantMessageAddress, | ||
| onCanMessageAddreses, | ||
| onMessageFailed, | ||
| onMessageSent, | ||
| onCanMessageAddressesUpdate, | ||
| onDelay, | ||
| }); | ||
| try { | ||
| await broadcast.broadcast({ message }); | ||
| } catch (err) { | ||
| console.error(`Error broadcasting: ${err}`); | ||
| updateBroadcastStatus(broadcastId, "failed"); | ||
| } | ||
| finishBroadcast(broadcastId); | ||
| }; |
Oops, something went wrong.