Files
olons_server/web/src/mobile/im.ts
T
Your Name db18514e6d
Deploy olons-server production / deploy (push) Successful in 3m38s
fix: make support IM delivery recoverable
2026-10-19 17:23:34 +08:00

399 lines
14 KiB
TypeScript

import { Drafty, Tinode } from 'tinode-sdk';
export interface CustomerServiceSession {
tenant_code: string;
api_base_url: string;
file_url: string;
websocket_url: string;
api_key: string;
token: string;
current_user_id: string;
topic: string;
}
export interface ChatMessage {
seq: number;
from: string;
text: string;
sent_at: string;
mine: boolean;
sending: boolean;
failed: boolean;
type: 'text' | 'image' | 'video';
media_url?: string;
preview_url?: string;
file_name?: string;
}
export type CustomerServiceConnectionState = 'connecting' | 'online' | 'reconnecting' | 'offline';
interface CustomerServiceChatOptions {
origin?: string;
onConnectionState?: (state: CustomerServiceConnectionState) => void;
}
export function resolveCustomerServiceEndpoints(session: CustomerServiceSession, origin: string) {
const pageOrigin = new URL(origin);
const sameOriginURL = (value: string, fallback: string) => {
const configured = new URL(value || fallback, pageOrigin);
const resolved = new URL(`${configured.pathname}${configured.search}`, pageOrigin);
return resolved.toString().replace(/\/$/, '');
};
const websocket = new URL(sameOriginURL(session.websocket_url, '/v0/channels'));
websocket.protocol = pageOrigin.protocol === 'https:' ? 'wss:' : 'ws:';
return {
apiBaseURL: sameOriginURL(session.api_base_url, '/v1/im'),
fileURL: sameOriginURL(session.file_url, '/v0/file'),
websocketURL: websocket.toString().replace(/\/$/, ''),
};
}
export class CustomerServiceChat {
private static readonly reconnectPageSize = 50;
private static readonly maxReconnectPages = 100;
private tinode: any;
private topic: any;
private currentUserID: string;
private onMessages: (messages: ChatMessage[]) => void;
private session: CustomerServiceSession;
private endpoints: ReturnType<typeof resolveCustomerServiceEndpoints>;
private token = '';
private sessionActive = false;
private intentionalDisconnect = false;
private recovery: Promise<void> | null = null;
private onConnectionState: (state: CustomerServiceConnectionState) => void;
constructor(
session: CustomerServiceSession,
onMessages: (messages: ChatMessage[]) => void,
options: CustomerServiceChatOptions = {},
) {
const origin = options.origin || window.location.origin;
this.endpoints = resolveCustomerServiceEndpoints(session, origin);
const endpoint = new URL(this.endpoints.websocketURL);
this.session = session;
this.token = session.token;
this.currentUserID = session.current_user_id;
this.onMessages = onMessages;
this.onConnectionState = options.onConnectionState || (() => undefined);
this.tinode = new Tinode({
appName: '悦选',
host: endpoint.host,
apiKey: session.api_key,
transport: 'ws',
secure: endpoint.protocol === 'wss:',
persist: false,
});
this.installTenantPatch(session.tenant_code);
this.topic = this.tinode.getTopic(session.topic);
this.topic.onData = () => this.refreshMessages();
this.tinode.onDisconnect = () => {
if (!this.intentionalDisconnect && this.sessionActive) {
this.onConnectionState('reconnecting');
}
};
this.tinode.onAutoreconnectIteration = (timeoutMs: number) => {
if (!this.intentionalDisconnect && this.sessionActive) {
this.onConnectionState(timeoutMs >= 0 ? 'reconnecting' : 'offline');
}
};
this.tinode.onConnect = () => {
if (!this.intentionalDisconnect && this.sessionActive) {
return this.restoreAfterReconnect();
}
};
}
async connect(token: string) {
this.intentionalDisconnect = false;
this.token = token;
this.onConnectionState('connecting');
await this.tinode.connect();
const login = await this.tinode.loginToken(this.token);
if (login?.code >= 300) throw new Error('客服会话登录失败');
this.acceptRotatedToken(login);
const query = this.topic.startMetaQuery().withData(undefined, undefined, 50).build();
await this.requestMessagePage(() => this.topic.subscribe(query), '客服会话订阅失败');
this.sessionActive = true;
this.onConnectionState('online');
this.refreshMessages();
}
async send(text: string) {
const value = text.trim();
if (!value) return;
const message = this.topic.createMessage(value, false);
message.head = Object.assign(message.head || {}, { 'x-im-client-mid': clientMID() });
const result = await this.topic.publishDraft(message, Promise.resolve());
if (!result || result.code >= 300 || message._failed || message._fatal) {
throw new Error(result?.text || '消息发送失败,请重试');
}
}
async sendMedia(file: File, type: 'image' | 'video', onProgress?: (progress: number) => void) {
const dimensions = await mediaDimensions(file, type);
const ref = await this.upload(file, onProgress);
const descriptor = {
mime: file.type,
// Keep the server-issued relative reference. Tinode only includes
// relative references in extra.attachments, which lets the server bind
// the uploaded file to this message and authorize the recipient.
refurl: ref,
filename: file.name,
size: file.size,
width: dimensions.width,
height: dimensions.height,
duration: dimensions.duration,
};
const content = type === 'image'
? Drafty.appendImage(null, descriptor)
: Drafty.insertVideo(null, 0, descriptor);
const message = this.topic.createMessage(content, false);
message.head = Object.assign(message.head || {}, {
'x-im-client-mid': clientMID(),
'x-im-type': type,
});
const result = await this.topic.publishDraft(message, Promise.resolve());
if (!result || result.code >= 300 || message._failed || message._fatal) {
throw new Error(result?.text || `${type === 'image' ? '照片' : '视频'}发送失败,请重试`);
}
}
disconnect() {
this.intentionalDisconnect = true;
this.sessionActive = false;
this.onConnectionState('offline');
try {
this.topic?.leave?.(false);
} finally {
this.tinode?.disconnect?.();
}
}
private restoreAfterReconnect() {
if (this.recovery) return this.recovery;
this.recovery = (async () => {
this.onConnectionState('reconnecting');
const login = await this.tinode.loginToken(this.token);
if (login?.code >= 300) throw new Error('客服会话重新登录失败');
this.acceptRotatedToken(login);
await this.restoreSubscriptionAndMessages();
this.refreshMessages();
this.onConnectionState('online');
})()
.catch(() => {
this.onConnectionState('offline');
})
.finally(() => {
this.recovery = null;
});
return this.recovery;
}
private acceptRotatedToken(login: any) {
const rotated = String(login?.params?.token || '');
if (!rotated) return;
this.token = rotated;
this.session.token = rotated;
}
private async restoreSubscriptionAndMessages() {
const pageSize = CustomerServiceChat.reconnectPageSize;
const firstQuery = this.topic.startMetaQuery().withLaterData(pageSize).build();
let count = await this.requestMessagePage(() => this.topic.subscribe(firstQuery), '客服会话重新订阅失败');
for (let page = 1; count >= pageSize; page += 1) {
if (page >= CustomerServiceChat.maxReconnectPages) {
throw new Error('客服消息积压过多,请重新进入客服会话');
}
const previousMax = Number(this.topic.maxMsgSeq?.() || 0);
const query = this.topic.startMetaQuery().withLaterData(pageSize).build();
count = await this.requestMessagePage(() => this.topic.getMeta(query), '客服历史消息补拉失败');
const currentMax = Number(this.topic.maxMsgSeq?.() || 0);
if (count > 0 && currentMax <= previousMax) {
throw new Error('客服历史消息补拉未推进');
}
}
}
private async requestMessagePage(request: () => Promise<any>, errorMessage: string) {
let observedCount: number | undefined;
let resolveObserved: ((count: number) => void) | undefined;
const observed = new Promise<number>((resolve) => {
resolveObserved = resolve;
});
const previousHandler = this.topic.onAllMessagesReceived;
this.topic.onAllMessagesReceived = (count: number) => {
previousHandler?.(count);
observedCount = Number(count || 0);
resolveObserved?.(observedCount);
};
let timeout: ReturnType<typeof setTimeout> | undefined;
try {
const response = await request();
if (response?.code >= 300) throw new Error(errorMessage);
if (observedCount !== undefined) return observedCount;
return await Promise.race([
observed,
new Promise<number>((_, reject) => {
// A Tinode request promise resolves before the SDK finishes routing
// its data packets. Only this completion callback guarantees that
// maxMsgSeq and the local message cache belong to the finished page.
timeout = setTimeout(() => reject(new Error(errorMessage)), 5_000);
}),
]);
} finally {
if (timeout) clearTimeout(timeout);
this.topic.onAllMessagesReceived = previousHandler;
}
}
private installTenantPatch(tenantCode: string) {
const connection = this.tinode._connection;
if (!connection || connection.__olonsTenantPatched) return;
const original = connection.sendText.bind(connection);
connection.sendText = (packet: unknown) => {
if (typeof packet === 'string' && packet.includes('"hi":')) {
try {
const parsed = JSON.parse(packet);
if (parsed.hi) {
parsed.hi.tenant = tenantCode;
packet = JSON.stringify(parsed);
}
} catch {
// Preserve the original packet when it is not JSON.
}
}
return original(packet);
};
connection.__olonsTenantPatched = true;
}
private refreshMessages() {
const rows: ChatMessage[] = [];
this.topic.messages((message: any) => {
const parsed = messageContent(message?.content, (ref) => this.mediaURL(ref));
if (!parsed.text && !parsed.media_url) return;
rows.push({
seq: Number(message.seq || 0),
from: String(message.from || ''),
text: parsed.text,
sent_at: message.ts ? new Date(message.ts).toISOString() : '',
mine: message.from === this.currentUserID,
sending: Boolean(message._sending),
failed: Boolean(message._failed || message._fatal),
type: parsed.type,
media_url: parsed.media_url,
preview_url: parsed.preview_url,
file_name: parsed.file_name,
});
});
this.onMessages(rows.sort((a, b) => a.seq - b.seq));
}
private async upload(file: File, onProgress?: (progress: number) => void) {
const uploadURL = new URL(this.endpoints.fileURL);
uploadURL.pathname = `${uploadURL.pathname.replace(/\/$/, '').replace(/\/u$/, '')}/u/`;
const form = new FormData();
form.set('id', clientMID());
form.set('file', file, file.name);
onProgress?.(0.1);
let response: Response;
try {
// Keep uploads on fetch so API and token credentials stay in headers and
// never leak into the URL, browser history, or proxy access logs.
response = await fetch(uploadURL.toString(), {
method: 'POST',
headers: {
'X-MeetingIM-APIKey': this.session.api_key,
Authorization: `Token ${this.token}`,
},
body: form,
});
} catch {
throw new Error('媒体上传失败,请检查网络后重试');
}
let payload: any;
try {
payload = await response.json();
} catch {
throw new Error('媒体上传响应格式不正确');
}
const ref = String(payload?.ctrl?.params?.url || '');
if (!response.ok || !ref) {
throw new Error(payload?.ctrl?.text || '媒体上传失败');
}
onProgress?.(1);
return ref;
}
private mediaURL(ref?: string) {
if (!ref) return '';
const apiBase = new URL(this.endpoints.apiBaseURL);
const resolved = new URL(ref, apiBase);
// Historical messages may contain the old absolute IM host. Route only
// known IM file paths through the current frontend origin; leave unrelated
// external media untouched.
if (!resolved.pathname.startsWith('/v0/file/') && resolved.origin !== apiBase.origin) return ref;
const signable = `${resolved.pathname}${resolved.search}`;
const authorized = this.tinode.authorizeURL(signable);
return new URL(authorized, this.endpoints.apiBaseURL).toString();
}
}
function messageContent(content: unknown, mediaURL: (ref?: string) => string) {
if (typeof content === 'string') return { text: content, type: 'text' as const };
if (content && typeof content === 'object') {
const drafty = content as { txt?: unknown; ent?: Array<{ tp?: string; data?: Record<string, unknown> }> };
const media = drafty.ent?.find((entity) => entity?.tp === 'IM' || entity?.tp === 'VD');
if (media?.data) {
const type = media.tp === 'VD' ? 'video' : 'image';
return {
text: '',
type,
media_url: mediaURL(String(media.data.ref || '')),
preview_url: mediaURL(String(media.data.preref || '')),
file_name: String(media.data.name || ''),
} as const;
}
return { text: String(drafty.txt || ''), type: 'text' as const };
}
return { text: '', type: 'text' as const };
}
function mediaDimensions(file: File, type: 'image' | 'video') {
return new Promise<{ width: number; height: number; duration: number }>((resolve, reject) => {
const url = URL.createObjectURL(file);
const cleanup = () => URL.revokeObjectURL(url);
if (type === 'image') {
const image = new Image();
image.onload = () => {
cleanup();
resolve({ width: image.naturalWidth, height: image.naturalHeight, duration: 0 });
};
image.onerror = () => {
cleanup();
reject(new Error('无法读取所选照片'));
};
image.src = url;
return;
}
const video = document.createElement('video');
video.preload = 'metadata';
video.onloadedmetadata = () => {
cleanup();
resolve({ width: video.videoWidth, height: video.videoHeight, duration: Math.round(video.duration * 1000) });
};
video.onerror = () => {
cleanup();
reject(new Error('无法读取所选视频'));
};
video.src = url;
});
}
function clientMID() {
return crypto.randomUUID?.() || `mid_${Date.now()}_${Math.random().toString(36).slice(2)}`;
}