Unifying Twitch and YouTube Live Chat: Ordered, De-duplicated

Why a local event stream instead of stitching in the UI?

Use a local event stream so there is a single ordered, durable feed that every consumer can replay. It cleanly separates ingestion (Twitch/YouTube) from delivery (UI, analytics, moderation), which keeps downstream code simple and predictable. You get idempotent processing and straightforward recovery. Ordering, de-duplication, and backpressure live in one place instead of being re-implemented in every client.

Twitch vs. YouTube chat at a glance

Aspect	Twitch (IRC/tmi.js)	YouTube (LiveChat API)
ID	IRC tag id	items[i].id
Timestamp	IRC tag tmi-sent-ts (ms)	snippet.publishedAt (ms)
Delivery	Realtime socket	Polling with nextPageToken + pollingIntervalMillis
Ordering	Mostly by send ts	Roughly by publishAt; polling introduces jitter
Rate limits	High	Strict quota costs

How do you merge Twitch and YouTube chats into one ordered stream?

Normalize both feeds to one schema, assign each event a Hybrid Logical Clock (HLC), and insert into a min-heap keyed by event time. Track a watermark as the minimum last-seen time across sources minus an allowed lateness window. Drain the heap while items are at or before that watermark, de-duplicate, then append to the local stream. The heap handles near-ties and jitter; the HLC breaks ties deterministically.

Normalized schema + HLC (TypeScript)

// types.ts
export type Platform = 'twitch' | 'youtube';

export interface ChatEvent {
  id: string;                // stable, dedup key (platform:id)
  platform: Platform;
  authorId: string;
  authorName: string;
  text: string;
  wallTs: number;            // wall clock ms (source-provided if trustworthy)
  hlc: string;               // hybrid logical clock string for tie-breaks
  sourceOffset: string;      // per-source offset/checkpoint
  meta?: Record<string, any>;
}

// Minimal HLC for a single node. Real implementation should persist nodeId.
export class HLC {
  private lastPhysical = 0;
  private logical = 0;
  constructor(private nodeId = 'n1') {}
  now(physical: number = Date.now()): string {
    if (physical > this.lastPhysical) {
      this.lastPhysical = physical; this.logical = 0;
    } else {
      this.logical += 1; // clock went backward or same tick
    }
    // Encode as ms-logic-nodeId for lexicographic order
    return `${this.lastPhysical.toString(36)}-${this.logical.toString(36)}-${this.nodeId}`;
  }
  static compare(a: string, b: string): number {
    // lexicographic compare is okay with same formatting
    return a < b ? -1 : a > b ? 1 : 0;
  }
}

How do you de-duplicate live chat messages reliably?

Use the platform’s message ID where available and prefix it with the platform for global uniqueness. If the ID is missing, fall back to a rolling content hash built from author, text, and a coarse timestamp, guarded by a TTL window to absorb replays. Back that with an LRU or a Bloom filter for O(1) checks, and require idempotency at storage.

Ingestion: Twitch (tmi.js) and YouTube (googleapis)

// ingress.ts
import tmi from 'tmi.js';
import { google } from 'googleapis';
import crypto from 'crypto';
import { ChatEvent, HLC } from './types';

const hlc = new HLC('ingress');

export function twitchIngress(opts: { channel: string; identity?: { username: string; password: string } }, push: (e: ChatEvent)=>void) {
  const client = new tmi.Client({
    connection: { secure: true, reconnect: true },
    identity: opts.identity,
    channels: [opts.channel]
  });
  client.on('message', (channel, tags, text, self) => {
    if (self) return;
    const id = tags['id'] as string | undefined;
    const sentTs = Number(tags['tmi-sent-ts'] || Date.now());
    const event: ChatEvent = {
      id: id ? `twitch:${id}` : `twitch:${crypto.createHash('sha1').update(`${tags['user-id']}:${sentTs}:${text}`).digest('hex')}`,
      platform: 'twitch',
      authorId: String(tags['user-id'] || ''),
      authorName: String(tags['display-name'] || tags['username'] || ''),
      text,
      wallTs: sentTs,
      hlc: hlc.now(sentTs),
      sourceOffset: id || String(sentTs),
      meta: { badges: tags['badges'], mod: tags['mod'], subscriber: tags['subscriber'] }
    };
    push(event);
  });
  client.connect();
}

export async function youtubeIngress(opts: { oauthClient: any; liveChatId: string }, push: (e: ChatEvent)=>void) {
  const youtube = google.youtube({ version: 'v3', auth: opts.oauthClient });
  let pageToken: string | undefined = undefined;
  while (true) {
    const res = await youtube.liveChatMessages.list({
      liveChatId: opts.liveChatId,
      part: ['id', 'snippet', 'authorDetails'],
      pageToken
    });
    pageToken = res.data.nextPageToken || pageToken;
    const waitMs = Number(res.data.pollingIntervalMillis || 1500);
    for (const item of res.data.items || []) {
      const id = item.id!;
      const sn = item.snippet!;
      const ad = item.authorDetails!;
      const ts = new Date(sn.publishedAt!).getTime();
      const text = sn.textMessageDetails?.messageText || '';
      const event: ChatEvent = {
        id: `youtube:${id}`,
        platform: 'youtube',
        authorId: ad.channelId!,
        authorName: ad.displayName!,
        text,
        wallTs: ts,
        hlc: hlc.now(ts),
        sourceOffset: pageToken || id,
        meta: { isChatOwner: ad.isChatOwner, isChatModerator: ad.isChatModerator, isChatSponsor: ad.isChatSponsor }
      };
      push(event);
    }
    await new Promise(r => setTimeout(r, waitMs));
  }
}

What about clocks, jitter, and watermarks?

Wall clocks drift. Polling introduces jitter. HLC gives you strictly monotonic timestamps within the process, but you still need to model lateness. Compute a watermark from the minimum last-seen wallTs across sources, minus an allowedLateness (for example, 2s). Emit only events with wallTs at or before the watermark, and keep the rest in a min-heap until they’re safe.

Ordered merge with dedupe, watermark, and delivery to Redis Streams

// merge.ts
import FastPriorityQueue from 'fastpriorityqueue';
import LRUCache from 'lru-cache';
import IORedis from 'ioredis';
import { ChatEvent, HLC } from './types';

const redis = new IORedis(process.env.REDIS_URL || 'redis://localhost:6379');

type HeapItem = ChatEvent;
const allowedLatenessMs = 2000; // tune with production metrics

// Min-heap by (wallTs, hlc)
const heap = new FastPriorityQueue<HeapItem>((a, b) =>
  a.wallTs === b.wallTs ? HLC.compare(a.hlc, b.hlc) < 0 : a.wallTs < b.wallTs);

// Dedupe: prefer platform IDs; fallback content hash bucketed by 3s
const seen = new LRUCache<string, number>({ max: 200_000, ttl: 5 * 60 * 1000 }); // 5 min TTL

// Track last-seen wallTs per source for watermarking
const lastSeen: Record<string, number> = { twitch: 0, youtube: 0 };

function dedupeKey(e: ChatEvent): string {
  if (e.id) return e.id; // platform:id
  const bucket = Math.floor(e.wallTs / 3000);
  const h = require('crypto').createHash('sha1').update(`${e.platform}:${e.authorId}:${bucket}:${e.text}`).digest('hex');
  return `${e.platform}:h:${h}`;
}

async function emit(e: ChatEvent) {
  // Persist to a local ordered bus. Redis Streams is easy to operate.
  // Include a deterministic idempotency key so consumers can enforce exactly-once.
  await redis.xadd('chat:merged', '*', 'id', e.id, 'platform', e.platform, 'authorId', e.authorId, 'authorName', e.authorName, 'text', e.text, 'wallTs', String(e.wallTs), 'hlc', e.hlc);
}

function watermark(): number {
  const vals = Object.values(lastSeen).filter(Boolean);
  if (vals.length === 0) return 0;
  return Math.min(...vals) - allowedLatenessMs;
}

export function push(e: ChatEvent) {
  lastSeen[e.platform] = Math.max(lastSeen[e.platform] || 0, e.wallTs);
  heap.add(e);
}

export function startDrainLoop() {
  setInterval(async () => {
    const w = watermark();
    while (!heap.isEmpty()) {
      const head = heap.peek();
      if (!head || head.wallTs > w) break; // not safe to emit yet
      const e = heap.poll()!;
      const k = dedupeKey(e);
      if (seen.has(k)) continue;
      seen.set(k, e.wallTs);
      await emit(e);
    }
  }, 25);
}

Wiring it together (one process example)

// index.ts
import { twitchIngress, youtubeIngress } from './ingress';
import { push, startDrainLoop } from './merge';

async function main() {
  startDrainLoop();

  twitchIngress({ channel: 'your_twitch_channel' }, push);

  // YouTube OAuth and liveChatId discovery omitted for brevity
  // Acquire liveChatId from liveBroadcasts.list(part=snippet,contentDetails,status)
  // where status.lifecycleStatus = 'live'.
  const oauthClient = /* ... */ null as any;
  const liveChatId = process.env.YT_LIVE_CHAT_ID!;
  youtubeIngress({ oauthClient, liveChatId }, push).catch(console.error);
}

main().catch(console.error);

How do you persist and deliver the merged stream locally?

Pick one append-only local bus (Redis Streams, Kafka, SQLite WAL) as the source of truth. Write merged events in order with an idempotency key, and let consumers read via consumer groups. Store per-source checkpoints so a restart resumes cleanly without replays.

Consuming from Redis Streams with consumer groups

// consumer.ts
import IORedis from 'ioredis';
const redis = new IORedis(process.env.REDIS_URL || 'redis://localhost:6379');
const stream = 'chat:merged';
const group = 'ui';
const consumer = `c-${process.pid}`;

async function ensureGroup() {
  try { await redis.xgroup('CREATE', stream, group, '$', 'MKSTREAM'); } catch {}
}

async function run() {
  await ensureGroup();
  while (true) {
    const resp = await redis.xreadgroup('GROUP', group, consumer, 'BLOCK', 5000, 'COUNT', 100, 'STREAMS', stream, '>');
    if (!resp) continue;
    for (const [, entries] of resp) {
      for (const [id, fields] of entries) {
        const m = Object.fromEntries(chunks(fields, 2) as any) as Record<string,string>;
        // render to UI, store in MongoDB, etc.
        // Acknowledge after successful processing to avoid re-delivery
        await redis.xack(stream, group, id);
      }
    }
  }
}

function* chunks<T>(arr: T[], n: number) { for (let i=0;i<arr.length;i+=n) yield arr.slice(i, i+n); }
run().catch(console.error);

Operational notes and hard-won opinions

• Size the lateness window with measurements, not guesses. Start with P99 source skew plus the polling interval. Log head.wallTs − watermark so you can see actual buffering delay.
• Exactly-once is the consumer’s job. Put idempotency keys on the stream and write consumers to honor them.
• Quotas: YouTube LiveChat reads burn quota; follow pollingIntervalMillis. Batch writes and avoid fan-out loops over entire chats.
• Backpressure: if the heap grows past a threshold, temporarily raise allowed lateness or drop non-critical sources first.
• Recovery: snapshot lastSeen and the LRU periodically if you need tighter guarantees; otherwise TTL-based dedupe will cover duplicates after restart.

Testing the ordering and dedupe logic

• Fuzz late arrivals: randomly hold back 5–10% of inputs by 1–3s and assert total order by (wallTs, hlc).
• Inject duplicates: replay 1% of previously emitted IDs and verify xadd count does not increase.
• Clock skew: prefer source timestamps when available and rely on HLC for monotonicity to clamp negative drift.

Where to go from here

• Add moderation events (timeouts, deletes) to the same stream and treat them as commands over message IDs.
• Keep a small SQLite table for offsets and a checksum of the last 1k IDs to speed up warm restarts.
• Export metrics: heap size, watermark lag, emit rate, dedupe hit rate, per-source offsets, Redis pending counts.