消息信封、交付、恢复与错误
按类别和操作分发 v1 消息信封,容忍重复,并在支持回放时使用已存 History。
分发消息信封
每个实时消息信封先按 t 分发,再按 op 分发。处理程序必须幂等,因为历史回放可能返回已实时处理的事件。
定义消息信封
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;
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. Omitted means twitter on legacy frames.
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. On legacy frames, fall back to 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;
// Completes an earlier truncated rendering. This is not an edit signal.
textUpdateType: 'completion';
}
);
type TweetDeleteEvent = {
tweetId: string;
// On legacy frames, fall back to author.platform, then twitter.
platform?: 'twitter' | 'truth_social' | 'binance_square';
eventId?: string;
deletedAt?: number;
receivedAt?: number;
author?: TweetAuthor;
text?: string;
};
type TweetPinEvent = {
tweetId: string;
// On legacy frames, fall back to 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;
// On legacy frames, fall back to 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 CalloutEvent = {
platform: 'pump_fun';
caller: {
username: string;
};
token: {
address: string;
symbol?: string;
name?: string;
// Token artwork when available. This is not an account profile image.
image?: string;
};
url?: string;
marketCapUsd?: number;
// Exact decimal strings preserve small onchain values.
calloutPrice?: string;
multiple?: number;
maxPriceSol?: string;
// The pump.fun Callout UUID.
calloutId: string;
// When pump.fun created the Callout.
createdAt: number;
// When TweetStream received it.
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 AccountEnvelope =
| AccountProfileEnvelope
| AccountFollowEnvelope
| AccountUnfollowEnvelope
| AccountAffiliateEnvelope
| AccountCalloutEnvelope;
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;
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;
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);
}
}分发操作
| 类别 | 操作 | 含义 |
|---|---|---|
| tweet | content | 原始发帖、回复、引用、转推,或支持的 Truth Social 帖子内容 |
| tweet | meta | 为已知帖子添加 OCR、代币检测、CEX 和预测市场数据 |
| tweet | update | 更新已知帖子的字段 |
| tweet | delete, pin, unpin | 将已知帖子标记为已删除、已置顶或已取消置顶 |
| account | profile_update, follow, unfollow, affiliate_update | 被监控账号的状态变化 |
| charity | added | 新增慈善项目。可在控制台开启提醒 |
| news | article | 来自已开启媒体的文章 |
| control | twitter_handles_result | WebSocket 账号管理命令的结果 |
按事件流处理交付
消息信封没有确认、恢复 token 或序列号。不要根据 ts 或 id 推断顺序或恰好一次交付。
处理程序必须幂等,并保留足够的本地状态,以识别应用已经处理的事件。
重连后回补
固定 startDate 和 endDate,在其他筛选条件不变的情况下持续使用 nextCursor,直到返回 null。以幂等方式处理每行。如果中断后的回放需要继续而不是重启,请一起保存时间范围和 cursor。
History 返回已存事件,不保证完整性或保留期限。AFFILIATE 返回已存的关联账号变更,不是当前列表快照。delete、pin 和 unpin 仍只在实时流中发送。
回放恢复窗口
typescript
type HistoryRow = {
time: string;
tweetId: string;
};
type HistoryPage = {
data: Array<HistoryRow>;
metadata: { nextCursor: string | null };
};
async function replayAfterReconnect(
checkpoint: string,
processEvent: (event: HistoryRow) => Promise<void>,
saveCheckpoint: (eventTime: string) => Promise<void>,
) {
const apiKey = process.env.TWEETSTREAM_API_KEY;
if (!apiKey) throw new Error("Missing TWEETSTREAM_API_KEY");
const url = new URL("https://api.tweetstream.io/api/history");
const replayEnd = new Date().toISOString();
url.searchParams.set("handles", "marketdesk");
url.searchParams.set("startDate", checkpoint);
url.searchParams.set("endDate", replayEnd);
url.searchParams.set("limit", "1000");
url.searchParams.set("type", "TWEET");
const pages: Array<Array<HistoryRow>> = [];
let cursor: string | null = null;
do {
if (cursor) url.searchParams.set("cursor", cursor);
const response = await fetch(url, {
headers: { Authorization: `Bearer ${apiKey}` },
});
if (!response.ok) {
throw new Error(`Replay failed (${response.status}): ${await response.text()}`);
}
const page = (await response.json()) as HistoryPage;
pages.push(page.data);
cursor = page.metadata.nextCursor;
} while (cursor);
for (const event of pages.reverse().flatMap((page) => [...page].reverse())) {
await processEvent(event);
}
await saveCheckpoint(replayEnd);
}修复握手错误
WebSocket 连接升级失败时,会在连接被接受前返回 JSON。客户端可以安全记录该响应。
| 状态 | 可能原因 | 修复方式 |
|---|---|---|
| 400 | WebSocket 协议 header 缺失或无效 | 发送 tweetstream.v1 和有效认证 token 协议 |
| 401 | API key 缺失、为空或无效 | 在控制台重新生成或轮换 API key |
| 403 | 订阅状态不允许实时流 | 开始试用、续费或升级账号 |
| 429 | 活跃 WebSocket 连接数达到限制 | 关闭旧连接或升级套餐 |
| 503 | 连接限制器暂时不可用 | 使用退避策略重试 |
读取连接升级拒绝响应
json
{
"error": "WebSocket connection limit reached (3/3 active). Close an existing connection and retry.",
"status": 429
}