Ably Real-time Collaboration

Ably Real-time Collaboration

When to Use This Skill

Activate this skill when working on:

• Implementing real-time features
• Setting up Ably channels
• Processing real-time messages
• Building collaborative editing features
• Implementing drag-and-drop with real-time sync
• Handling WebSocket connections
• Managing real-time state updates
• Optimizing real-time performance

Core Patterns

Channel Management

Channel Naming Convention:

// Pattern: `board:{boardId}`
const channelName = `board:${boardId}`;

Setting Up Channels:

"use client";

import { useChannel } from "ably/​react";
import { useEffect } from "react";

export function PostChannelComponent({ boardId }: { boardId: string }) {
  const { channel } = useChannel(`board:${boardId}`, (message) => {
    processMessage(message);
  });

  return <PostList boardId={boardId} />;
}

Message Processors with Validation

Critical Pattern: Always validate messages before processing

// lib/​realtime/​messageProcessors.ts
import { z } from "zod";

// Define message schema
const PostUpdateMessageSchema = z.object({
  type: z.literal("post:update"),
  postId: z.string(),
  content: z.string().min(1).max(1000),
  userId: z.string(),
  timestamp: z.number(),
});

// Message processor
export const processPostUpdate = (rawData: unknown) => {
  try {
    // ✅ Validate message structure
    const data = PostUpdateMessageSchema.parse(rawData);

    // ✅ Check staleness (30s threshold)
    const now = Date.now();
    const age = now - data.timestamp;

    if (age > 30000) {
      console.warn("Stale message discarded", {
        type: data.type,
        age,
        postId: data.postId,
      });
      return;
    }

    // ✅ Process validated, fresh message
    updatePostContent(data.postId, data.content);
  } catch (error) {
    if (error instanceof z.ZodError) {
      console.error("Invalid message structure", {
        details: error.errors,
        rawData,
      });
    } else {
      console.error("Message processing error", error);
    }
  }
};

Staleness Filtering

30-Second Threshold: Prevents processing old messages after reconnection

const STALENESS_THRESHOLD_MS = 30000;

export function isMessageStale(timestamp: number): boolean {
  const age = Date.now() - timestamp;
  return age > STALENESS_THRESHOLD_MS;
}

// Usage in message handler
useChannel(`board:${boardId}`, (message) => {
  const { timestamp } = message.data;

  if (isMessageStale(timestamp)) {
    console.warn("Dropping stale message", { age: Date.now() - timestamp });
    return;
  }

  processMessage(message.data);
});

Error Recovery Strategies

Connection Error Handling:

import { useConnectionStateListener } from "ably/​react";

export function RealtimeProvider({ children }: { children: React.ReactNode }) {
  const [connectionState, setConnectionState] = useState<string>("initialized");

  useConnectionStateListener((stateChange) => {
    setConnectionState(stateChange.current);

    switch (stateChange.current) {
      case "connected":
        console.log("✅ Connected to Ably");
        break;

      case "disconnected":
        console.warn("⚠️ Disconnected from Ably");
        break;

      case "suspended":
        console.error("❌ Connection suspended");
        // Optionally show user notification
        break;

      case "failed":
        console.error("❌ Connection failed");
        // Show error message to user
        break;
    }
  });

  return (
    <>
      {connectionState !== "connected" && (
        <ConnectionBanner state={connectionState} />
      )}
      {children}
    </>
  );
}

Publishing Messages

Always Include Timestamp:

import { useChannel } from "ably/​react";

export function usePublishPostUpdate() {
  const { channel } = useChannel(`board:${boardId}`);

  const publishUpdate = async (postId: string, content: string) => {
    await channel.publish("post:update", {
      type: "post:update",
      postId,
      content,
      userId: currentUserId,
      timestamp: Date.now(), // ✅ Always include timestamp
    });
  };

  return publishUpdate;
}

Optimistic Updates for Voting

Pattern: Update UI immediately, sync in background

"use client";

import { voteSignal } from "@/​lib/​signal/​postSignals";
import { useChannel } from "ably/​react";

export function VoteButton({
  postId,
  boardId,
}: {
  postId: string;
  boardId: string;
}) {
  const { channel } = useChannel(`board:${boardId}`);

  const handleVote = async () => {
    // ✅ Optimistic update (immediate UI feedback)
    voteSignal.value = {
      ...voteSignal.value,
      [postId]: (voteSignal.value[postId] || 0) + 1,
    };

    try {
      // Persist to database
      await submitVote(postId);

      // Broadcast to other users
      await channel.publish("post:vote", {
        type: "post:vote",
        postId,
        increment: 1,
        timestamp: Date.now(),
      });
    } catch (error) {
      // ❌ Rollback on error
      voteSignal.value = {
        ...voteSignal.value,
        [postId]: voteSignal.value[postId] - 1,
      };
      console.error("Vote failed", error);
    }
  };

  return <button onClick={handleVote}>Vote</​button>;
}

Drag-and-Drop Integration

Lazy-Loaded with Real-Time Sync:

// components/​board/​PostProvider.tsx
"use client";

import { useChannel } from "ably/​react";
import dynamic from "next/​dynamic";

// ✅ Lazy load drag-and-drop (reduces initial bundle)
const DragDropArea = dynamic(() => import("./​DragDropArea"), { ssr: false });

export function PostProvider({ boardId }: { boardId: string }) {
  useChannel(`board:${boardId}`, (message) => {
    if (message.name === "post:move") {
      handlePostMove(message.data);
    }
  });

  const handleDrop = async (postId: string, newType: PostType) => {
    // Update locally
    movePostSignal(postId, newType);

    // Persist to database
    await updatePostType(postId, newType);

    // Broadcast to other users
    channel.publish("post:move", {
      type: "post:move",
      postId,
      newType,
      timestamp: Date.now(),
    });
  };

  return <DragDropArea onDrop={handleDrop} />;
}

Anti-Patterns

❌ Not Validating Messages

Bad:

useChannel(`board:${boardId}`, (message) => {
  // ❌ Trusts message data completely
  updatePost(message.data.postId, message.data.content);
});

Good:

useChannel(`board:${boardId}`, (message) => {
  // ✅ Validates before processing
  const validated = PostUpdateSchema.safeParse(message.data);
  if (!validated.success) return;
  updatePost(validated.data.postId, validated.data.content);
});

❌ Not Checking Message Staleness

Bad:

useChannel(channelName, (message) => {
  // ❌ Processes all messages, even old ones after reconnect
  processMessage(message.data);
});

Good:

useChannel(channelName, (message) => {
  // ✅ Filters stale messages
  if (isMessageStale(message.data.timestamp)) return;
  processMessage(message.data);
});

❌ Not Handling Connection Errors

Bad:

// ❌ No error handling
const { channel } = useChannel(channelName);

Good:

// ✅ Monitor connection state
useConnectionStateListener((stateChange) => {
  if (stateChange.current === "failed") {
    showErrorNotification("Real-time connection lost");
  }
});

❌ Publishing Without Timestamp

Bad:

channel.publish("update", {
  postId,
  content,
  // ❌ No timestamp for staleness check
});

Good:

channel.publish("update", {
  postId,
  content,
  timestamp: Date.now(), // ✅ Include timestamp
});

❌ Not Handling Race Conditions

Bad:

// ❌ Multiple updates could conflict
const handleVote = async () => {
  const newCount = currentCount + 1;
  await updateVoteCount(postId, newCount);
};

Good:

// ✅ Use atomic increment
const handleVote = async () => {
  await db
    .update(postTable)
    .set({ voteCount: sql`vote_count + 1` })
    .where(eq(postTable.id, postId));
};

Integration with Other Skills

• rbac-security: Validate messages for security
• signal-state-management: Real-time updates to signals
• nextjs-app-router: Client components for real-time features
• testing-patterns: Test message processors with fake timers

Project-Specific Context

Key Files

• lib/​realtime/​messageProcessors.ts - Message validation and processing
• components/​board/​PostProvider.tsx - Real-time channel setup
• components/​board/​PostChannelComponent.tsx - Channel subscription
• lib/​realtime/​__tests__/ - Message processor tests

Message Types

Current Message Types:

type MessageType =
  | "post:create"
  | "post:update"
  | "post:delete"
  | "post:move"
  | "post:vote"
  | "member:join"
  | "member:leave";

Channel Structure

One Channel Per Board:

• Channel name: board:{boardId}
• All board events published to this channel
• Subscribers filter by message type

Performance Optimizations

1. Lazy Loading: Drag-and-drop loaded on demand
2. Staleness Filter: Discards messages >30s old
3. Optimistic Updates: Immediate UI feedback
4. Batching: Vote counts updated atomically
5. Connection Pooling: Reuse Ably client instance

Error Recovery

Automatic Reconnection:

• Ably SDK handles reconnection automatically
• Messages buffered during disconnection
• Staleness filter prevents processing old messages after reconnect

Manual Recovery:

// Refresh data after long disconnection
if (
  stateChange.previous === "suspended" &&
  stateChange.current === "connected"
) {
  await refreshBoardData();
}

Testing

Mock Ably in Tests:

jest.mock("ably/​react", () => ({
  useChannel: jest.fn(() => ({
    channel: {
      publish: jest.fn(),
      subscribe: jest.fn(),
    },
  })),
}));

---

Last Updated: 2026-01-10