构建生产环境消费者
分发已知操作,通过退避策略重连,并确保每个副作用都可安全重复。
通过退避策略重连
从可重连的 Node.js 或 Python 消费者开始,再用一个完整的处理程序分发所有操作。
分发所有服务端操作
此 TypeScript 分发器覆盖全部 18 种服务端操作。它会先验证输入,并在遇到未知操作时忽略该消息,而不是关闭 WebSocket。
让每个处理程序保持简短。在调用策略、通知或执行边界前,先保存回执和本地投影。
完整 TypeScript 分发器
显示完整定义(632 行)
完整 TypeScript 分发器typescript
type VerifiedType = 'blue' | 'business' | 'government' | 'verified' | 'none';
type TweetVerifiedLabel = {
badge: string | null;
description: string;
url: string | null;
};
type TweetAuthor = {
banner?: string;
bio?: string;
followersCount?: number;
followingCount?: number;
id?: string;
joinedAt?: number;
location?: string;
metrics?: {
likes?: number;
tweets?: number;
};
// Includes a leading @ when present, for example "@elonmusk".
handle?: string;
name?: string;
platform?: 'twitter' | 'truth_social' | 'binance_square';
profileImage?: string;
// The account's own website, when it has one.
url?: string;
verifiedLabel?: TweetVerifiedLabel;
verifiedType?: VerifiedType;
};
type Media = {
url: string;
} & (
| {
type: 'video';
// Every video is a progressive MP4. A public still poster is included
// when available.
thumbnail?: string;
}
| {
type?: 'image' | 'gif';
thumbnail?: string;
}
);
type TweetUrl = {
url: string;
name?: string;
tco?: string;
};
type TweetMention = {
handle?: string;
id?: string;
name?: string;
};
type TweetArticle = {
description?: string;
id?: string;
publishedAt?: number;
text?: string;
thumbnail?: string;
title: string;
updatedAt?: number;
url: string;
};
type TweetPollChoice = {
id?: string;
image?: string;
label: string;
votes?: number;
};
type TweetPoll = {
choices: TweetPollChoice[];
endsAt?: number;
totalVotes?: number;
updatedAt?: number;
};
type TweetContentKind = 'post' | 'reply' | 'quote' | 'retweet';
type TweetReference = {
article?: TweetArticle;
type: 'reply' | 'quote' | 'retweet';
tweetId: string;
text?: string;
translatedText?: string;
author?: TweetAuthor;
media?: Media[];
poll?: TweetPoll;
quoted?: TweetReference;
subtweet?: TweetReference;
};
type TweetContent = {
tweetId: string;
kind: TweetContentKind;
// Original tweet text when the event includes both original and translated text.
text: string;
// Translation, present only when available.
translatedText?: string;
createdAt: number;
author: TweetAuthor;
article?: TweetArticle;
link?: string;
media?: Media[];
mentions?: TweetMention[];
poll?: TweetPoll;
receivedAt?: number;
urls?: TweetUrl[];
ref?: TweetReference;
};
type TweetMeta = {
tweetId: string;
// Pair with tweetId when merging enrichment. When omitted, use twitter.
platform?: 'twitter' | 'truth_social' | 'binance_square';
ocr?: {
text: string;
};
detected?: {
tokens?: Array<{
symbol?: string;
name?: string;
contract?: string;
chain?: string;
networkId?: number;
priceUsd?: number;
sources: Array<'text' | 'ocr'>;
}>;
cex?: Array<{
exchange: 'bybit' | 'binance' | 'hyperliquid';
symbol?: string;
priceUsd?: number;
url?: string;
baseAsset?: string;
quoteAsset?: string;
sources: Array<'text' | 'ocr'>;
}>;
prediction?: Array<{
exchange: 'polymarket' | 'kalshi';
marketId?: string;
title?: string;
priceUsd?: number;
url?: string;
sources: Array<'text' | 'ocr'>;
}>;
};
};
type TweetUpdate = {
tweetId: string;
// Pair with tweetId. When absent, use author.platform, then twitter.
platform?: 'twitter' | 'truth_social' | 'binance_square';
article?: TweetArticle;
kind?: TweetContentKind;
translatedText?: string;
author?: TweetAuthor;
media?: Media[];
mentions?: TweetMention[];
poll?: TweetPoll;
receivedAt?: number;
urls?: TweetUrl[];
ref?: TweetReference;
} & (
| {
text?: string;
textUpdateType?: never;
}
| {
text: string;
// Supplies the full text of a previously shortened post. Not an edit.
textUpdateType: 'completion';
}
);
type TweetDeleteEvent = {
tweetId: string;
// When absent, use author.platform, then twitter.
platform?: 'twitter' | 'truth_social' | 'binance_square';
eventId?: string;
deletedAt?: number;
receivedAt?: number;
author?: TweetAuthor;
text?: string;
};
type TweetPinEvent = {
tweetId: string;
// When absent, use tweet.author.platform or author.platform, then twitter.
platform?: 'twitter' | 'truth_social' | 'binance_square';
eventId: string;
observedAt: number;
receivedAt?: number;
action: 'pin';
author: TweetAuthor;
text?: string;
tweet?: TweetContent;
};
type TweetUnpinEvent = {
tweetId: string;
// When absent, use tweet.author.platform or author.platform, then twitter.
platform?: 'twitter' | 'truth_social' | 'binance_square';
eventId: string;
observedAt: number;
receivedAt?: number;
action: 'unpin';
author: TweetAuthor;
text?: string;
tweet?: TweetContent;
};
type AccountEventActor = TweetAuthor & {
websiteUrl?: string;
};
type ProfileUpdateEvent = {
kind: 'PROFILE';
eventId: string;
observedAt: number;
receivedAt?: number;
actor: AccountEventActor;
changes: {
avatar?: string;
banner?: string;
bio?: string;
handle?: string;
location?: string;
name?: string;
verifiedLabel?: TweetVerifiedLabel | null;
websiteUrl?: string | null;
};
previous?: {
avatar?: string;
banner?: string;
bio?: string;
handle?: string;
location?: string;
name?: string;
verifiedLabel?: TweetVerifiedLabel | null;
websiteUrl?: string | null;
};
};
type FollowEvent = {
kind: 'FOLLOW';
eventId: string;
observedAt: number;
receivedAt?: number;
actor: AccountEventActor;
target: AccountEventActor & {
handle: string;
};
};
type UnfollowEvent = {
kind: 'UNFOLLOW';
eventId: string;
observedAt: number;
receivedAt?: number;
actor: AccountEventActor;
target: AccountEventActor & {
handle: string;
};
};
type AffiliateAccountIdentity = AccountEventActor & {
id: string;
};
type AffiliateUpdateEvent = {
action: 'added' | 'removed';
eventId: string;
observedAt: number;
receivedAt?: number;
organization: AffiliateAccountIdentity;
member: AffiliateAccountIdentity;
};
type CalloutCaller = {
username: string;
image?: string;
};
type CalloutReply = {
caller: CalloutCaller;
replyId: string;
parentReplyId?: string;
rootReplyId?: string;
text?: string;
mediaUrl?: string;
};
type CalloutEvent = {
platform: 'pump_fun';
caller: CalloutCaller;
kind?: 'callout' | 'author_update' | 'reply' | 'repost';
text?: string;
mediaUrl?: string;
// Required for replies and author updates.
replyId?: string;
parentReplyId?: string;
rootReplyId?: string;
reference?: {
caller: CalloutCaller;
text?: string;
mediaUrl?: string;
reply?: CalloutReply;
};
position?: {
amountHeld?: string;
valueUsd?: string;
costBasisUsd?: string;
pnlUsd?: string;
pnlPercentage?: string;
realizedPnlUsd?: string;
};
token: {
address: string;
chainId?: string;
symbol?: string;
name?: string;
image?: string;
};
url?: string;
marketCapUsd?: number;
// SOL, exact decimal.
calloutPrice?: string;
// USD, exact decimal.
calloutPriceUsd?: string;
multiple?: number;
// SOL peak, exact decimal.
maxPriceSol?: string;
// USD peak, exact decimal.
maxPriceUsd?: string;
calloutId: string;
createdAt: number;
receivedAt?: number;
};
type CalloutUpdateEvent = CalloutEvent & {
// Original envelope id, not calloutId.
updateOf: string;
};
type FomoCoin = {
address: string;
networkId: number;
name?: string;
symbol?: string;
image?: string;
metadataObservedAt?: number;
priceUsd?: string;
marketCapUsd?: string;
liquidityUsd?: string;
volume24hUsd?: string;
change24hPercent?: string;
};
type FomoThesisEvent = {
platform: 'fomo';
coin: FomoCoin;
author: {
id: string;
username: string;
verified: boolean;
displayName?: string;
avatar?: string;
twitter?: string;
};
thesis: {
id: string;
text: string;
createdAt: number;
likes: number;
replies: number;
tradeId?: string;
parentId?: string;
segments?: Array<{ text: string; link?: string }>;
};
position?: {
tokenAmount?: string;
usdValue?: string;
unrealizedPnlUsd?: string;
realizedPnlUsd?: string;
unrealizedPnlPercent?: string;
realizedPnlPercent?: string;
closedAt?: number;
};
observedAt?: number;
receivedAt?: number;
};
type FomoTokenVerifiedEvent = {
platform: 'fomo';
coin: FomoCoin;
observedAt: number;
receivedAt?: number;
};
type CharityAddedEvent = {
id: string;
name: string;
slug: string;
enabled: boolean;
taxId?: string;
location?: {
city?: string;
state?: string;
country?: string;
};
observedAt: number;
};
type TwitterHandlesResult = {
action: 'follow' | 'unfollow';
requestId: string | null;
results: Array<{
input: string;
state:
| 'added'
| 'already_following'
| 'invalid_input'
| 'duplicate'
| 'not_found'
| 'failed'
| 'removed'
| 'not_following';
message?: string;
}>;
error: string | null;
};
type NewsSource = {
id: string;
name: string;
};
type NewsMedia = {
url: string;
type?: 'image' | 'video';
caption?: string;
};
type NewsArticle = {
source: NewsSource;
url: string;
title: string;
publishedAt: string;
receivedAt: number;
modifiedAt?: string;
primaryCategory?: string;
categories: string[];
author?: string;
keywords: string[];
description?: string;
summary?: string;
media: NewsMedia[];
content?: string;
language?: string;
copyright?: string;
};
type EnvelopeBase<
TFamily extends 'tweet' | 'account' | 'charity' | 'control' | 'news',
TOp extends string,
TPayload extends object,
> = {
v: 1;
t: TFamily;
op: TOp;
id?: string;
ts: number;
d: TPayload;
};
type TweetContentEnvelope = EnvelopeBase<'tweet', 'content', TweetContent>;
type TweetMetaEnvelope = EnvelopeBase<'tweet', 'meta', TweetMeta>;
type TweetUpdateEnvelope = EnvelopeBase<'tweet', 'update', TweetUpdate>;
type TweetDeleteEnvelope = EnvelopeBase<'tweet', 'delete', TweetDeleteEvent>;
type TweetPinEnvelope = EnvelopeBase<'tweet', 'pin', TweetPinEvent>;
type TweetUnpinEnvelope = EnvelopeBase<'tweet', 'unpin', TweetUnpinEvent>;
type AccountProfileEnvelope = EnvelopeBase<
'account',
'profile_update',
ProfileUpdateEvent
>;
type AccountFollowEnvelope = EnvelopeBase<'account', 'follow', FollowEvent>;
type AccountUnfollowEnvelope = EnvelopeBase<'account', 'unfollow', UnfollowEvent>;
type AccountAffiliateEnvelope = EnvelopeBase<
'account',
'affiliate_update',
AffiliateUpdateEvent
>;
type AccountCalloutEnvelope = EnvelopeBase<'account', 'callout', CalloutEvent> & {
id: string;
};
type AccountCalloutUpdateEnvelope = EnvelopeBase<
'account',
'callout_update',
CalloutUpdateEvent
> & {
id: string;
};
type AccountThesisEnvelope = EnvelopeBase<'account', 'thesis', FomoThesisEvent> & {
id: string;
};
type AccountTokenVerifiedEnvelope = EnvelopeBase<
'account',
'token_verified',
FomoTokenVerifiedEvent
> & {
id: string;
};
type AccountEnvelope =
| AccountProfileEnvelope
| AccountFollowEnvelope
| AccountUnfollowEnvelope
| AccountAffiliateEnvelope
| AccountCalloutEnvelope
| AccountCalloutUpdateEnvelope
| AccountThesisEnvelope
| AccountTokenVerifiedEnvelope;
type CharityAddedEnvelope = EnvelopeBase<'charity', 'added', CharityAddedEvent> & {
id: string;
};
type EmptyPayload = Record<string, never>;
type AuthPingCommand = { op: 'auth_ping' };
type AuthPongEnvelope = EnvelopeBase<'control', 'auth_pong', EmptyPayload>;
type TwitterHandlesResultEnvelope = EnvelopeBase<
'control',
'twitter_handles_result',
TwitterHandlesResult
>;
type ControlEnvelope = AuthPongEnvelope | TwitterHandlesResultEnvelope;
type NewsEnvelope = EnvelopeBase<'news', 'article', NewsArticle> & {
id: string;
};
type TweetStreamEnvelope =
| TweetContentEnvelope
| TweetMetaEnvelope
| TweetUpdateEnvelope
| TweetDeleteEnvelope
| TweetPinEnvelope
| TweetUnpinEnvelope
| AccountEnvelope
| CharityAddedEnvelope
| NewsEnvelope
| ControlEnvelope;
// Feed this router only values returned by a complete runtime decoder.
// See the reference consumer linked below for the executable decoder.
type DecodedEnvelope =
| { kind: 'known'; event: TweetStreamEnvelope }
| { kind: 'unknown' };
type ProtocolHandlers = {
tweetContent: (event: TweetContentEnvelope) => void;
tweetMeta: (event: TweetMetaEnvelope) => void;
tweetUpdate: (event: TweetUpdateEnvelope) => void;
tweetDelete: (event: TweetDeleteEnvelope) => void;
tweetPin: (event: TweetPinEnvelope) => void;
tweetUnpin: (event: TweetUnpinEnvelope) => void;
accountProfileUpdate: (event: AccountProfileEnvelope) => void;
accountFollow: (event: AccountFollowEnvelope) => void;
accountUnfollow: (event: AccountUnfollowEnvelope) => void;
accountAffiliateUpdate: (event: AccountAffiliateEnvelope) => void;
accountCallout: (event: AccountCalloutEnvelope) => void;
accountCalloutUpdate: (event: AccountCalloutUpdateEnvelope) => void;
accountThesis: (event: AccountThesisEnvelope) => void;
accountTokenVerified: (event: AccountTokenVerifiedEnvelope) => void;
charityAdded: (event: CharityAddedEnvelope) => void;
newsArticle: (event: NewsEnvelope) => void;
authPong: (event: AuthPongEnvelope) => void;
twitterHandlesResult: (event: TwitterHandlesResultEnvelope) => void;
// Count and discard unknown input without logging its raw payload.
unknown: () => void;
};
function assertNever(value: never): never {
throw new Error('Unhandled validated envelope');
}
function route(decoded: DecodedEnvelope, handlers: ProtocolHandlers): void {
if (decoded.kind === 'unknown') {
handlers.unknown();
return;
}
const event = decoded.event;
switch (event.t) {
case 'tweet':
switch (event.op) {
case 'content': handlers.tweetContent(event); return;
case 'meta': handlers.tweetMeta(event); return;
case 'update': handlers.tweetUpdate(event); return;
case 'delete': handlers.tweetDelete(event); return;
case 'pin': handlers.tweetPin(event); return;
case 'unpin': handlers.tweetUnpin(event); return;
default: return assertNever(event);
}
case 'account':
switch (event.op) {
case 'profile_update': handlers.accountProfileUpdate(event); return;
case 'follow': handlers.accountFollow(event); return;
case 'unfollow': handlers.accountUnfollow(event); return;
case 'affiliate_update': handlers.accountAffiliateUpdate(event); return;
case 'callout': handlers.accountCallout(event); return;
case 'callout_update': handlers.accountCalloutUpdate(event); return;
case 'thesis': handlers.accountThesis(event); return;
case 'token_verified': handlers.accountTokenVerified(event); return;
default: return assertNever(event);
}
case 'charity':
handlers.charityAdded(event);
return;
case 'news':
handlers.newsArticle(event);
return;
case 'control':
switch (event.op) {
case 'auth_pong': handlers.authPong(event); return;
case 'twitter_handles_result': handlers.twitterHandlesResult(event); return;
default: return assertNever(event);
}
default:
return assertNever(event);
}
}准备生产环境
- 连接关闭或发生网络错误后,使用退避策略重连。
- 只连接文档列出的 TweetStream 端点,并使用上文说明的认证方式。路由由 TweetStream 处理,不需要基础设施专用 Header。
- 使用
(platform, tweetId)作为帖子键,而不是消息信封 id。content 从author.platform读取platform,后续操作从d.platform读取。 - 旧版操作缺少
d.platform时,使用其中的 author platform;如果没有,则按twitter处理。 - 幂等应用每个
tweet/update。更新包含ref时,用该快照替换已存引用,并沿ref.subtweet读取下一条被引用的帖子。 - 只有对完整消息信封计算指纹后,才丢弃完全相同的重放。
- 把
meta视为某条已路由推文的迟到富化信息。 - 将回执和投影存入带索引的持久化存储。根据回放窗口和存储预算制定保留或归档策略;不要删除待处理任务。
- 记录操作失败时不要写入 payload,并通过退避策略重试。使用相同的回执 ID 作为幂等键。
- 跟踪活跃 WebSocket 连接数,避免超过套餐限制。
测量信号延迟
在消息回调开始时记录本地接收时间。对于 X/Twitter 事件,解码 tweet snowflake 时间戳,测量从发布到接收的时间。主机时钟必须与 NTP 同步,wall-clock 结果才有意义。接收后的处理阶段使用单调时钟。
- 报告监控列表、客户端区域、UTC 测试时段、样本量、p50 和 p95。
- 记录重连、缺失事件、时钟同步状态和每条排除规则。
- 单独报告冷启动和重连后的样本,不要将它们混入热路径结果。
- 只有发布点、接收点、地理位置、样本和百分位边界一致时,才比较结果。
| 阶段 | 起点 | 终点 |
|---|---|---|
| 发布到接收 | X snowflake 时间戳 | WebSocket 回调开始时记录的本地时间 |
| 接收到决策 | 本地 WebSocket 接收 | 策略与风控决策就绪 |
| 决策到交易场所确认 | 通过风控的决策 | 单独的交易场所响应或拒绝 |
测量 Snowflake 延迟
typescripttype ContentEvent = {
t?: string;
op?: string;
d?: {
tweetId?: string;
};
};
const TWITTER_EPOCH_MS = 1_288_834_974_657n;
function tweetIdToTimestampMs(tweetId: string) {
const id = BigInt(tweetId);
return Number((id >> 22n) + TWITTER_EPOCH_MS);
}
function measureSnowflakeLatency(tweetId: string, arrivedAtMs: number) {
const tweetedAtMs = tweetIdToTimestampMs(tweetId);
return arrivedAtMs - tweetedAtMs;
}
ws.on("message", (raw) => {
const arrivedAtMs = Date.now();
const event = JSON.parse(raw.toString()) as ContentEvent;
const tweetId = event.d?.tweetId;
if (event.t === "tweet" && event.op === "content" && tweetId) {
const latencyMs = measureSnowflakeLatency(tweetId, arrivedAtMs);
console.log("publication-to-receipt ms", latencyMs);
}
});构建参考消费者
构建自己的消费者时,这些页面分别讲解每个部分。先使用上方的分发器,再添加一种重连客户端、持久恢复和账号管理。