Understanding the Social Media Feed Problem
A social media feed is deceptively complex. At first glance it looks like a simple list of posts sorted by time, but under the hood it involves multiple read models, cross-service data aggregation, real-time updates, and personalized content. Traditional CRUD architectures struggle here because they tightly couple the write model (how data is stored) to the read model (how data is queried). When you need to serve a feed that combines posts, comments, likes, follow relationships, and trending signals — all with different access patterns — that coupling becomes a bottleneck.
CQRS (Command Query Responsibility Segregation) and Event Sourcing offer a compelling alternative. Together they decouple writes from reads, give you an immutable audit trail of every user action, and let you build specialized, highly performant read models tailored to each feed variant.
What Is CQRS in the Context of a Social Feed
🚀 Deploy your AI agent in 10 minutes
Managed Hermes hosting. Zero DevOps. 100M tokens/mo included.
Try it free →CQRS splits the application into two distinct paths:
- Command Side — handles write operations (create post, add comment, like, follow). It validates business rules, persists state, and emits events.
- Query Side — handles read operations (load home feed, load user profile feed, load trending feed). It queries pre-built, optimized read models called projections.
The critical insight: the command side never directly serves feed queries. Instead, every write operation produces an event. Those events are consumed asynchronously by projection builders that maintain read-optimized views. This means your feed can be denormalized, pre-sorted, and personalized without ever touching the canonical write storage.
What Is Event Sourcing
Event Sourcing replaces mutable state with an append-only log of events. Instead of storing "current state of a post" in a row and overwriting it on edits, you store a sequence of fine-grained events:
// Event stream for a single post (aggregate: Post)
PostCreated { postId, authorId, content, timestamp }
PostEdited { postId, newContent, editTimestamp }
PostLiked { postId, userId, timestamp }
PostUnliked { postId, userId, timestamp }
CommentAdded { postId, commentId, authorId, text, timestamp }
The current state of any aggregate is derived by replaying its event stream from the beginning. This gives you a complete, auditable history — invaluable for features like "edit history," undo/redo, and machine learning pipelines that analyze behavior over time.
Why This Combination Matters for Feed Design
- Multiple Read Models Without Contention — You can build a chronological feed projection, a trending feed projection, and a "friends-first" feed projection all independently, each with its own schema, indexes, and caching strategy. The write side never bottlenecks reads.
- Real-Time Feed Updates — Events naturally plug into streaming infrastructure (Kafka, Pulsar, CDC streams). When a user posts, the event triggers projection updates and pushes changes to connected feed consumers via WebSockets or Server-Sent Events.
- Time-Travel Queries — Because events are immutable, you can reconstruct a user's feed exactly as it appeared at any point in time — useful for debugging, audit, or "on this day" nostalgia features.
- Resilience Against Failures — If a projection becomes corrupt, you don't lose data. You simply discard it and replay all events from the beginning to rebuild a correct read model.
- Machine Learning Integration — The raw event stream is gold for ML. Recommendation models can consume every like, skip, dwell-time event, and share as training data without imposing any additional load on the production write path.
Architecture Overview
Here's the high-level flow for a social media feed system built with CQRS and Event Sourcing:
┌──────────────┐ ┌─────────────────┐ ┌──────────────────┐
│ Client App │──POST─▶│ Command API │──────▶│ Event Store │
│ (write) │ │ /api/v1/posts │ │ (append-only) │
└──────────────┘ └─────────────────┘ └────────┬─────────┘
│
│ events (Kafka/stream)
▼
┌──────────────┐ ┌─────────────────┐ ┌──────────────────┐
│ Client App │──GET──│ Query API │◀──────│ Projection DB │
│ (read) │ │ /api/v1/feed │ │ (read-optimized)│
└──────────────┘ └─────────────────┘ └──────────────────┘
▲
│
┌───────┴──────────┐
│ Projection │
│ Workers │
│ (build views) │
└──────────────────┘
Modeling the Domain Events
Start by defining the events that matter for your feed. Think in terms of what users do, not what state changes look like. Use a strongly typed event envelope:
// events/types.ts
export interface EventEnvelope {
eventId: string; // UUID, unique per event
aggregateId: string; // e.g., postId or userId
aggregateType: string; // "Post" | "User" | "Comment"
eventType: string; // "PostCreated" | "PostLiked" etc.
payload: object;
metadata: {
timestamp: number; // epoch millis
userId: string; // who caused this event
correlationId: string; // trace ID
version: number; // aggregate version, for optimistic concurrency
};
}
// events/post-events.ts
export interface PostCreated extends EventEnvelope {
eventType: 'PostCreated';
payload: {
postId: string;
authorId: string;
content: string;
mediaUrls: string[];
visibility: 'public' | 'friends' | 'private';
};
}
export interface PostLiked extends EventEnvelope {
eventType: 'PostLiked';
payload: {
postId: string;
likedByUserId: string;
};
}
export interface FollowRelationCreated extends EventEnvelope {
eventType: 'FollowRelationCreated';
payload: {
followerId: string;
followedId: string;
};
}
Every event is immutable. Once appended to the event store, it can never be modified or deleted. This is the foundation that makes everything else work reliably.
Implementing the Command Side
The command side accepts user intents (commands), validates them, and produces events. A common pattern uses the Aggregate pattern from Domain-Driven Design. Each aggregate is responsible for a consistency boundary — typically one aggregate per post, per user profile, and per follow relationship.
// aggregates/post-aggregate.ts
import { PostCreated, PostEdited, PostLiked, PostUnliked, CommentAdded } from '../events/post-events';
import { EventStore } from '../event-store/event-store';
export class PostAggregate {
private postId: string;
private authorId: string;
private content: string;
private mediaUrls: string[];
private visibility: string;
private likes: Set; // set of userIds who liked
private comments: Array<{ commentId: string; authorId: string; text: string; timestamp: number }>;
private version: number;
// Rehydrate from event history
static async load(postId: string, eventStore: EventStore): Promise {
const events = await eventStore.getEventsForAggregate('Post', postId);
const aggregate = new PostAggregate(postId);
for (const event of events) {
aggregate.apply(event);
}
return aggregate;
}
constructor(postId: string) {
this.postId = postId;
this.likes = new Set();
this.comments = [];
this.version = 0;
}
private apply(event: EventEnvelope): void {
switch (event.eventType) {
case 'PostCreated':
const created = event.payload as PostCreated['payload'];
this.authorId = created.authorId;
this.content = created.content;
this.mediaUrls = created.mediaUrls;
this.visibility = created.visibility;
break;
case 'PostEdited':
const edited = event.payload as PostEdited['payload'];
this.content = edited.newContent;
break;
case 'PostLiked':
const liked = event.payload as PostLiked['payload'];
this.likes.add(liked.likedByUserId);
break;
case 'PostUnliked':
const unliked = event.payload as PostUnliked['payload'];
this.likes.delete(unliked.unlikedByUserId);
break;
case 'CommentAdded':
const commentAdded = event.payload as CommentAdded['payload'];
this.comments.push({
commentId: commentAdded.commentId,
authorId: commentAdded.authorId,
text: commentAdded.text,
timestamp: commentAdded.timestamp
});
break;
}
this.version = event.metadata.version;
}
// Command: Create a new post
public createPost(
authorId: string,
content: string,
mediaUrls: string[],
visibility: string
): PostCreated {
if (this.version !== 0) {
throw new Error('Post already exists');
}
if (!content || content.length > 5000) {
throw new Error('Content must be between 1 and 5000 characters');
}
const event: PostCreated = {
eventId: generateUUID(),
aggregateId: this.postId,
aggregateType: 'Post',
eventType: 'PostCreated',
payload: { postId: this.postId, authorId, content, mediaUrls, visibility },
metadata: {
timestamp: Date.now(),
userId: authorId,
correlationId: getCurrentCorrelationId(),
version: this.version + 1
}
};
this.apply(event);
return event;
}
// Command: Like a post
public likePost(userId: string): PostLiked {
if (this.likes.has(userId)) {
throw new Error('User already liked this post');
}
const event: PostLiked = {
eventId: generateUUID(),
aggregateId: this.postId,
aggregateType: 'Post',
eventType: 'PostLiked',
payload: { postId: this.postId, likedByUserId: userId },
metadata: {
timestamp: Date.now(),
userId: userId,
correlationId: getCurrentCorrelationId(),
version: this.version + 1
}
};
this.apply(event);
return event;
}
}
The command handler ties it together — loads the aggregate, executes the command, and persists the resulting events atomically:
// handlers/post-command-handler.ts
export class PostCommandHandler {
constructor(private eventStore: EventStore, private eventBus: EventBus) {}
async handleCreatePost(command: CreatePostCommand): Promise {
const aggregate = new PostAggregate(command.postId);
const event = aggregate.createPost(
command.authorId,
command.content,
command.mediaUrls,
command.visibility
);
// Atomically append to event store with optimistic concurrency check
await this.eventStore.append(
'Post',
command.postId,
[event],
-1 // expected version: -1 means "new stream"
);
// Publish to event bus for projections and downstream consumers
await this.eventBus.publish(event);
}
async handleLikePost(command: LikePostCommand): Promise {
const aggregate = await PostAggregate.load(command.postId, this.eventStore);
const event = aggregate.likePost(command.userId);
await this.eventStore.append(
'Post',
command.postId,
[event],
aggregate.version // optimistic concurrency check
);
await this.eventBus.publish(event);
}
}
The Event Store
The event store is the system of record. It must guarantee atomic, ordered appends per aggregate stream. A production implementation might use PostgreSQL with a dedicated events table, or a specialized database like EventStoreDB. Here's a minimal PostgreSQL-backed event store:
// event-store/postgres-event-store.ts
import { Pool } from 'pg';
export class PostgresEventStore implements EventStore {
private pool: Pool;
constructor(pool: Pool) {
this.pool = pool;
}
async append(
aggregateType: string,
aggregateId: string,
events: EventEnvelope[],
expectedVersion: number
): Promise {
const client = await this.pool.connect();
try {
await client.query('BEGIN');
// Lock the stream to prevent concurrent appends
const current = await client.query(
`SELECT MAX(metadata->>'version')::int as version
FROM events
WHERE aggregate_id = $1 AND aggregate_type = $2`,
[aggregateId, aggregateType]
);
const currentVersion = current.rows[0]?.version || 0;
if (expectedVersion !== -1 && currentVersion !== expectedVersion) {
throw new Error(
`Concurrency conflict: expected version ${expectedVersion}, got ${currentVersion}`
);
}
let nextVersion = currentVersion;
for (const event of events) {
nextVersion++;
event.metadata.version = nextVersion;
await client.query(
`INSERT INTO events (event_id, aggregate_id, aggregate_type, event_type, payload, metadata, occurred_at)
VALUES ($1, $2, $3, $4, $5, $6, $7)`,
[
event.eventId,
event.aggregateId,
event.aggregateType,
event.eventType,
JSON.stringify(event.payload),
JSON.stringify(event.metadata),
new Date(event.metadata.timestamp)
]
);
}
await client.query('COMMIT');
} catch (err) {
await client.query('ROLLBACK');
throw err;
} finally {
client.release();
}
}
async getEventsForAggregate(
aggregateType: string,
aggregateId: string
): Promise {
const result = await this.pool.query(
`SELECT event_id, event_type, payload, metadata
FROM events
WHERE aggregate_id = $1 AND aggregate_type = $2
ORDER BY metadata->>'version' ASC`,
[aggregateId, aggregateType]
);
return result.rows.map(row => ({
eventId: row.event_id,
aggregateId,
aggregateType,
eventType: row.event_type,
payload: row.payload,
metadata: row.metadata
}));
}
async getAllEventsSince(timestamp: number, batchSize: number = 500): Promise {
const result = await this.pool.query(
`SELECT event_id, aggregate_id, aggregate_type, event_type, payload, metadata
FROM events
WHERE occurred_at > $1
ORDER BY occurred_at ASC
LIMIT $2`,
[new Date(timestamp), batchSize]
);
return result.rows.map(row => ({
eventId: row.event_id,
aggregateId: row.aggregate_id,
aggregateType: row.aggregate_type,
eventType: row.event_type,
payload: row.payload,
metadata: row.metadata
}));
}
}
The events table schema is straightforward and append-only:
CREATE TABLE events (
event_id UUID PRIMARY KEY,
aggregate_id UUID NOT NULL,
aggregate_type VARCHAR(50) NOT NULL,
event_type VARCHAR(100) NOT NULL,
payload JSONB NOT NULL,
metadata JSONB NOT NULL,
occurred_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE INDEX idx_events_aggregate ON events (aggregate_id, aggregate_type, occurred_at);
CREATE INDEX idx_events_type_occurred ON events (event_type, occurred_at);
CREATE INDEX idx_events_occurred ON events (occurred_at);
Building Feed Projections
Projections are the heart of the read side. They consume events and build denormalized views optimized for feed queries. A projection worker typically runs as a background process, listening to the event bus and incrementally updating a dedicated read database (Redis, PostgreSQL, Elasticsearch, or a combination).
Home Feed Projection (Chronological)
The home feed shows posts from users the viewer follows, sorted by recency. This projection maintains a per-user feed list in Redis for lightning-fast retrieval:
// projections/home-feed-projection.ts
import { Redis } from 'ioredis';
import { EventBus } from '../event-bus/event-bus';
import {
PostCreated, FollowRelationCreated, PostEdited,
PostVisibilityChanged, PostDeleted
} from '../events/post-events';
export class HomeFeedProjection {
private redis: Redis;
private eventBus: EventBus;
private checkpoint: number; // last processed event timestamp
constructor(redis: Redis, eventBus: EventBus) {
this.redis = redis;
this.eventBus = eventBus;
this.checkpoint = 0;
}
async start(): Promise {
// 1. Load checkpoint from Redis
const saved = await this.redis.get('projection:home-feed:checkpoint');
if (saved) {
this.checkpoint = parseInt(saved, 10);
}
// 2. Subscribe to relevant event types
this.eventBus.subscribe('PostCreated', this.handlePostCreated.bind(this));
this.eventBus.subscribe('PostEdited', this.handlePostEdited.bind(this));
this.eventBus.subscribe('PostVisibilityChanged', this.handleVisibilityChanged.bind(this));
this.eventBus.subscribe('PostDeleted', this.handlePostDeleted.bind(this));
this.eventBus.subscribe('FollowRelationCreated', this.handleFollowCreated.bind(this));
this.eventBus.subscribe('FollowRelationRemoved', this.handleFollowRemoved.bind(this));
// 3. Catch up on missed events if needed (handled by event bus subscription replay)
console.log('HomeFeedProjection started');
}
// When a new post is created, fan-out to all followers' feeds
private async handlePostCreated(event: PostCreated): Promise {
const { postId, authorId, content, visibility } = event.payload;
const timestamp = event.metadata.timestamp;
if (visibility === 'private') {
// Private posts don't go into follower feeds
await this.saveCheckpoint(timestamp);
return;
}
// Get all followers of the author
const followers = await this.getFollowers(authorId);
// Build the feed entry
const feedEntry = JSON.stringify({
postId,
authorId,
content: content.substring(0, 200), // preview
timestamp,
type: 'post'
});
// Fan-out: push to each follower's feed sorted set in Redis
const pipeline = this.redis.pipeline();
for (const followerId of followers) {
const feedKey = `feed:home:${followerId}`;
pipeline.zadd(feedKey, timestamp, feedEntry);
// Keep only the latest 1000 entries to bound memory
pipeline.zremrangebyrank(feedKey, 0, -1001);
}
await pipeline.exec();
await this.saveCheckpoint(timestamp);
}
// When a follow relationship is created, backfill the new follower's feed
private async handleFollowCreated(event: FollowRelationCreated): Promise {
const { followerId, followedId } = event.payload;
// Get recent public posts from the newly followed user
const recentPosts = await this.getRecentPublicPosts(followedId, 50);
if (recentPosts.length > 0) {
const feedKey = `feed:home:${followerId}`;
const pipeline = this.redis.pipeline();
for (const post of recentPosts) {
const entry = JSON.stringify({
postId: post.postId,
authorId: post.authorId,
content: post.content.substring(0, 200),
timestamp: post.timestamp,
type: 'post'
});
pipeline.zadd(feedKey, post.timestamp, entry);
}
pipeline.zremrangebyrank(feedKey, 0, -1001);
await pipeline.exec();
}
// Also store the follow relationship for future fan-out
await this.redis.sadd(`user:${followedId}:followers`, followerId);
await this.redis.sadd(`user:${followerId}:following`, followedId);
await this.saveCheckpoint(event.metadata.timestamp);
}
private async handleFollowRemoved(event: FollowRelationRemoved): Promise {
const { followerId, followedId } = event.payload;
// Remove all posts by followedId from follower's feed
const feedKey = `feed:home:${followerId}`;
const members = await this.redis.zrange(feedKey, 0, -1);
const pipeline = this.redis.pipeline();
for (const member of members) {
const entry = JSON.parse(member);
if (entry.authorId === followedId) {
pipeline.zrem(feedKey, member);
}
}
await pipeline.exec();
await this.redis.srem(`user:${followedId}:followers`, followerId);
await this.redis.srem(`user:${followerId}:following`, followedId);
await this.saveCheckpoint(event.metadata.timestamp);
}
private async handlePostEdited(event: PostEdited): Promise {
// Update the feed entry content across all follower feeds
const { postId, newContent } = event.payload;
const followers = await this.redis.smembers(`post:${postId}:followers-cached`);
const pipeline = this.redis.pipeline();
for (const followerId of followers) {
const feedKey = `feed:home:${followerId}`;
const members = await this.redis.zrange(feedKey, 0, -1);
for (const member of members) {
const entry = JSON.parse(member);
if (entry.postId === postId) {
entry.content = newContent.substring(0, 200);
// ZADD with same score updates the member value in Redis sorted sets
pipeline.zadd(feedKey, entry.timestamp, JSON.stringify(entry));
}
}
}
await pipeline.exec();
await this.saveCheckpoint(event.metadata.timestamp);
}
private async handlePostDeleted(event: PostDeleted): Promise {
const { postId } = event.payload;
const followers = await this.redis.smembers(`post:${postId}:followers-cached`);
const pipeline = this.redis.pipeline();
for (const followerId of followers) {
const feedKey = `feed:home:${followerId}`;
const members = await this.redis.zrange(feedKey, 0, -1);
for (const member of members) {
const entry = JSON.parse(member);
if (entry.postId === postId) {
pipeline.zrem(feedKey, member);
}
}
}
await pipeline.exec();
await this.saveCheckpoint(event.metadata.timestamp);
}
private async getFollowers(userId: string): Promise {
return this.redis.smembers(`user:${userId}:followers`);
}
private async getRecentPublicPosts(userId: string, limit: number): Promise {
// This could query a dedicated post-detail projection
const posts = await this.redis.lrange(`user:${userId}:posts`, 0, limit - 1);
return posts.map(p => JSON.parse(p));
}
private async saveCheckpoint(timestamp: number): Promise {
this.checkpoint = Math.max(this.checkpoint, timestamp);
await this.redis.set('projection:home-feed:checkpoint', this.checkpoint.toString());
}
}
Trending Feed Projection (Materialized View in PostgreSQL)
A trending feed ranks posts by a composite score combining recency, likes, and comments. This projection uses PostgreSQL as the read store because it supports complex ranking queries with indexes:
// projections/trending-feed-projection.ts
import { Pool } from 'pg';
export class TrendingFeedProjection {
private pool: Pool;
constructor(pool: Pool) {
this.pool = pool;
}
async initialize(): Promise {
await this.pool.query(`
CREATE TABLE IF NOT EXISTS trending_posts (
post_id UUID PRIMARY KEY,
author_id UUID NOT NULL,
content TEXT NOT NULL,
like_count INT DEFAULT 0,
comment_count INT DEFAULT 0,
created_at TIMESTAMPTZ NOT NULL,
last_updated TIMESTAMPTZ NOT NULL,
trending_score DOUBLE PRECISION DEFAULT 0
);
CREATE INDEX IF NOT EXISTS idx_trending_score
ON trending_posts (trending_score DESC, created_at DESC);
`);
}
async handlePostCreated(event: PostCreated): Promise {
const { postId, authorId, content } = event.payload;
await this.pool.query(
`INSERT INTO trending_posts (post_id, author_id, content, created_at, last_updated)
VALUES ($1, $2, $3, $4, $4)
ON CONFLICT (post_id) DO NOTHING`,
[postId, authorId, content, new Date(event.metadata.timestamp)]
);
await this.recalculateScore(postId);
}
async handlePostLiked(event: PostLiked): Promise {
const { postId } = event.payload;
await this.pool.query(
`UPDATE trending_posts
SET like_count = like_count + 1, last_updated = NOW()
WHERE post_id = $1`,
[postId]
);
await this.recalculateScore(postId);
}
async handleCommentAdded(event: CommentAdded): Promise {
const { postId } = event.payload;
await this.pool.query(
`UPDATE trending_posts
SET comment_count = comment_count + 1, last_updated = NOW()
WHERE post_id = $1`,
[postId]
);
await this.recalculateScore(postId);
}
private async recalculateScore(postId: string): Promise {
// Trending score = weighted combination of recency, likes, and comments
// Uses a time-decay factor so older posts naturally fade
await this.pool.query(`
UPDATE trending_posts
SET trending_score = (
(like_count * 2.0 + comment_count * 3.0) *
EXP(-EXTRACT(EPOCH FROM (NOW() - created_at)) / 86400.0 * 0.1)
)
WHERE post_id = $1
`, [postId]);
}
async getTrendingFeed(limit: number = 20, offset: number = 0): Promise {
const result = await this.pool.query(
`SELECT post_id, author_id, content, like_count, comment_count, trending_score
FROM trending_posts
ORDER BY trending_score DESC, created_at DESC
LIMIT $1 OFFSET $2`,
[limit, offset]
);
return result.rows;
}
}
The Query API — Serving the Feed
The query API is thin and fast. It simply reads from projections with no business logic, no validation, and no joins across write-side tables:
// query-api/feed-query-handler.ts
import { Redis } from 'ioredis';
import { TrendingFeedProjection } from '../projections/trending-feed-projection';
export class FeedQueryService {
private redis: Redis;
private trendingProjection: TrendingFeedProjection;
constructor(redis: Redis, trendingProjection: TrendingFeedProjection) {
this.redis = redis;
this.trendingProjection = trendingProjection;
}
async getHomeFeed(userId: string, cursor?: number, limit: number = 20): Promise {
const feedKey = `feed:home:${userId}`;
// Redis ZREVRANGE returns highest score (most recent) first
const start = cursor ? await this.redis.zrevrank(feedKey, cursor) + 1 : 0;
const entries = await this.redis.zrevrange(feedKey, start, start + limit - 1);
// entries are JSON strings; parse and enrich if needed
const feedItems = entries.map(e => JSON.parse(e));
const nextCursor = entries.length === limit
? entries[entries.length - 1]
: null;
return {
items: feedItems,
nextCursor: nextCursor ? JSON.stringify(nextCursor) : null,
hasMore: entries.length === limit
};
}
async getTrendingFeed(userId: string, page: number = 1, limit: number = 20): Promise {
const offset = (page - 1) * limit;
const posts = await this.trendingProjection.getTrendingFeed(limit, offset);
return {
items: posts,
nextCursor: page.toString(),
hasMore: posts.length === limit
};
}
}
interface FeedResponse {
items: any[];
nextCursor: string | null;
hasMore: boolean;
}
Real-Time Feed Updates with WebSockets
For truly live feeds, combine the projection system with WebSocket push. When a projection updates a follower's feed, it also publishes a notification to a WebSocket hub:
// realtime/feed-notifier.ts
import { WebSocketServer, WebSocket } from 'ws';
import { EventBus } from '../event-bus/event-bus';
export class FeedNotifier {
private wss: WebSocketServer;
private connections: Map; // userId -> active sockets
constructor(server: any) {
this.wss = new WebSocketServer({ server });
this.connections = new Map();
this.wss.on('connection', (ws, req) => {
// Authenticate and extract userId from token in query string or header
const userId = extractUserIdFromRequest(req);
if (!userId) {
ws.close(4001, 'Authentication required');
return;
}
// Register connection
const existing = this.connections.get(userId) || [];
existing.push(ws);
this.connections.set(userId, existing);
ws.on('close', () => {
const sockets = this.connections.get(userId) || [];
this.connections.set(userId, sockets.filter(s => s !== ws));
});
// Send initial snapshot
this.sendInitialFeed(userId, ws);
});
}
private async sendInitialFeed(userId: string, ws: WebSocket): Promise {
// Fetch current feed snapshot and send as first message
// (Implementation depends on your feed query service)
ws.send(JSON.stringify({ type: 'feed:snapshot', items: [] }));
}
// Called by projection after updating a user's feed
notifyFeedUpdate(userId: string, newEntry: object): void {
const sockets = this.connections.get(userId) || [];
const message = JSON.stringify({
type: 'feed:new-item',
item: newEntry
});
for (const ws of sockets) {
if (ws.readyState === WebSocket.OPEN) {
ws.send(message);
}
}
}
// Called when a user is actively viewing their feed and scrolls
notifyFeedInsert(userId: string, entries: object[], position: 'top' | 'bottom'): void {
const sockets = this.connections.get(userId) || [];
const message = JSON.stringify({
type: 'feed:insert',
entries,
position
});
for (const ws of sockets) {
if (ws.readyState === WebSocket.OPEN) {
ws.send(message);
}
}
}
}
Wire the notifier into the home feed projection so that when a new post is fanned-out to followers, those currently online get instant updates:
// Inside HomeFeedProjection.handlePostCreated, after pipeline.exec():
for (const followerId of followers) {
const feedKey = `feed:home:${followerId}`;
// Check if follower is online via presence service or connection map
if (this.presenceService.isOnline(followerId)) {
this.feedNotifier.notifyFeedUpdate(followerId, {
postId,
authorId,
content: content.substring(0, 200),
timestamp,
type: 'post'
});
}
}
Handling Consistency and Lag
Eventual consistency is inherent to CQRS. There is always a delay between a write being acknowledged and the projection being updated. For a social feed, this is usually acceptable — users understand that feeds take a moment to update. However, you can minimize perceived lag:
- Write-side read-through — After a user creates a post, immediately insert it into their own feed projection synchronously (or via a fast-path cache) so the author sees their post instantly.
- Optimistic UI updates — The client app can optimistically show the user's own actions (their new post, their like) immediately in the UI while the projection catches up in the background.
- Versioned feed responses — Include a
lastEventTimestampin feed API responses. The client can poll with?since={timestamp}to get only new items, reducing the need for full feed reloads.
Snapshotting and Rehydration Performance
Loading an aggregate by replaying thousands of events on every command becomes expensive. Introduce snapshots — periodically save the aggregate's full state at a given version, then load only events since that snapshot:
// snapshotting/snapshot-manager.ts
export class SnapshotManager {
private pool: Pool;
private snapshotFrequency: number = 50; // every 50 events
async getAggregate(
aggregateType: string,
aggregateId: string,
eventStore: EventStore
): Promise<{ state: any; version: number }> {
// Try to load latest snapshot
const snapshot = await this.pool.query(
`SELECT state, version FROM snapshots
WHERE aggregate_id = $1 AND aggregate_type = $2
ORDER BY version DESC LIMIT 1`,
[aggregateId, aggregateType]
);
let state: any = {};
let fromVersion = 0;
if (snapshot.rows.length > 0) {
state = snapshot.rows