Designing a Video Streaming Platform with CQRS and Event Sourcing
Modern video streaming platforms like YouTube, Netflix, or Twitch handle enormous complexity behind the scenes. Videos must be uploaded, transcoded into multiple resolutions, analyzed for content, processed for thumbnails, distributed via CDNs, and served to millions of concurrent users — all while maintaining accurate view counts, engagement metrics, and recommendation data. Traditional CRUD-based architectures struggle with this combination of heavy processing pipelines, high read/write ratios, and audit requirements. This is where CQRS (Command Query Responsibility Segregation) and Event Sourcing shine. This tutorial walks you through designing such a system from the ground up, with practical code examples you can adapt to your own projects.
What Is CQRS and Event Sourcing?
🚀 Deploy your AI agent in 10 minutes
Managed Hermes hosting. Zero DevOps. 100M tokens/mo included.
Try it free →Command Query Responsibility Segregation (CQRS)
CQRS is an architectural pattern that separates read operations (queries) from write operations (commands). Instead of a single model that handles both, you build distinct paths:
- Command side (write model): Validates business rules, processes commands, and produces events.
- Query side (read model): Optimized for fast data retrieval, often using denormalized projections stored in separate databases.
This separation lets you scale reads and writes independently, optimize each database for its workload, and apply different security and performance characteristics to each side.
Event Sourcing
Event Sourcing replaces traditional state persistence with an append-only log of events. Instead of storing "current state" in a row and overwriting it, you store every state-changing event that occurred. The current state is derived by replaying all events in order. This gives you:
- A complete, immutable audit trail of everything that happened
- The ability to reconstruct any historical state (temporal queries)
- Natural support for event-driven architectures and eventual consistency
Why These Patterns Matter for Video Streaming
Video streaming platforms face unique challenges that make CQRS and Event Sourcing particularly valuable:
- Complex processing pipelines: A single video upload triggers transcoding (1080p, 720p, 480p, 360p), thumbnail generation, content moderation, and more. Each step produces events that other services react to — a perfect fit for event-driven design.
- Massive read/write asymmetry: View counts are updated millions of times per minute, while video metadata changes rarely. Separating the read and write models lets you scale view tracking independently from the video catalog.
- Full audit requirements: Content ownership disputes, copyright claims, and moderation decisions demand a complete history. Event Sourcing provides this natively.
- Rebuild capabilities: If your analytics projections become corrupted, you can rebuild them entirely by replaying the event log — no data migration scripts needed.
- Async processing resilience: Transcoding can take minutes or hours. Events decouple the upload from processing, letting the system remain responsive while heavy work happens in the background.
Architecture Overview
Here's the high-level architecture we'll build. The system consists of these key components:
- Command API: Receives commands like UploadVideo, TranscodeVideo, RecordView
- Aggregates: Domain objects that validate commands and emit events (Video, Channel, Playlist)
- Event Store: Append-only storage for all events (e.g., PostgreSQL, EventStoreDB, Kafka)
- Event Bus: Distributes events to interested consumers
- Projectors / Event Handlers: Build read models by reacting to events
- Read Databases: Specialized stores for different queries (Elasticsearch for search, Redis for view counts, PostgreSQL for user playlists)
- Query API: Serves read-optimized data to the frontend
- Processing Workers: Perform actual video transcoding, thumbnail extraction, etc.
Step 1: Designing the Domain Model and Events
Start by defining your domain events. These are the immutable facts that capture every meaningful change in the system. For a video streaming platform, key events include:
// domain-events.ts
export interface VideoEvent {
readonly eventId: string;
readonly aggregateId: string; // videoId
readonly aggregateType: 'Video';
readonly eventType: string;
readonly timestamp: Date;
readonly version: number; // aggregate version for optimistic concurrency
readonly metadata: {
userId: string;
ipAddress?: string;
userAgent?: string;
};
}
// --- Video Lifecycle Events ---
export interface VideoUploaded extends VideoEvent {
eventType: 'VideoUploaded';
payload: {
title: string;
description: string;
tags: string[];
originalFileName: string;
fileSizeBytes: number;
containerFormat: string; // e.g., 'mp4', 'mov', 'avi'
durationSeconds?: number;
storageBucket: string;
storageKey: string;
visibility: 'public' | 'private' | 'unlisted';
};
}
export interface TranscodingRequested extends VideoEvent {
eventType: 'TranscodingRequested';
payload: {
targetResolutions: string[]; // ['1080p', '720p', '480p', '360p']
transcodingJobId: string;
priority: 'normal' | 'high';
};
}
export interface TranscodingCompleted extends VideoEvent {
eventType: 'TranscodingCompleted';
payload: {
resolution: string;
codec: string;
bitrateKbps: number;
outputStorageKey: string;
fileSizeBytes: number;
durationSeconds: number;
checksum: string;
};
}
export interface ThumbnailGenerated extends VideoEvent {
eventType: 'ThumbnailGenerated';
payload: {
resolution: string;
storageKey: string;
timestampInVideo: number; // seconds from start
dominantColor?: string;
};
}
export interface ContentModerationCompleted extends VideoEvent {
eventType: 'ContentModerationCompleted';
payload: {
status: 'approved' | 'flagged' | 'rejected';
flags?: string[]; // e.g., ['copyright_music', 'adult_content']
confidenceScore: number; // 0.0 to 1.0
reviewedBy?: string; // 'auto' or moderator ID
};
}
export interface VideoPublished extends VideoEvent {
eventType: 'VideoPublished';
payload: {
publishedAt: Date;
};
}
export interface ViewRecorded extends VideoEvent {
eventType: 'ViewRecorded';
payload: {
viewerId?: string;
watchDurationSeconds: number;
watchedAt: Date;
source: string; // e.g., 'web', 'ios', 'android', 'embed'
};
}
export interface VideoMetadataUpdated extends VideoEvent {
eventType: 'VideoMetadataUpdated';
payload: {
changes: {
title?: string;
description?: string;
tags?: string[];
visibility?: string;
};
};
}
export interface VideoDeleted extends VideoEvent {
eventType: 'VideoDeleted';
payload: {
reason: string;
deletedBy: string;
};
}Step 2: Implementing the Video Aggregate (Command Side)
The aggregate is the domain object that receives commands, applies business rules, and produces events. It does NOT store current state in a traditional way — instead, it maintains state in memory by replaying past events, then validates commands against that state.
// video-aggregate.ts
import { v4 as uuidv4 } from 'uuid';
interface VideoState {
videoId: string;
title: string;
description: string;
tags: string[];
status: 'draft' | 'processing' | 'published' | 'failed' | 'deleted';
visibility: 'public' | 'private' | 'unlisted';
originalFile: { bucket: string; key: string; sizeBytes: number } | null;
transcodingJobs: Map;
completedTranscodings: { resolution: string; storageKey: string }[];
thumbnails: { resolution: string; storageKey: string }[];
moderationStatus: 'pending' | 'approved' | 'flagged' | 'rejected';
version: number;
createdAt: Date;
publishedAt: Date | null;
deletedAt: Date | null;
}
type VideoEvent =
| VideoUploaded
| TranscodingRequested
| TranscodingCompleted
| ThumbnailGenerated
| ContentModerationCompleted
| VideoPublished
| ViewRecorded
| VideoMetadataUpdated
| VideoDeleted;
export class VideoAggregate {
private state: VideoState;
private pendingEvents: VideoEvent[] = [];
constructor(videoId: string) {
// Initialize empty state for a brand new video
this.state = {
videoId,
title: '',
description: '',
tags: [],
status: 'draft',
visibility: 'public',
originalFile: null,
transcodingJobs: new Map(),
completedTranscodings: [],
thumbnails: [],
moderationStatus: 'pending',
version: 0,
createdAt: new Date(),
publishedAt: null,
deletedAt: null,
};
}
// Replay events to rebuild state (used when loading from event store)
replayEvents(events: VideoEvent[]): void {
for (const event of events) {
this.applyEvent(event);
this.state.version = event.version;
}
}
private applyEvent(event: VideoEvent): void {
switch (event.eventType) {
case 'VideoUploaded':
this.state.title = event.payload.title;
this.state.description = event.payload.description;
this.state.tags = event.payload.tags;
this.state.visibility = event.payload.visibility;
this.state.originalFile = {
bucket: event.payload.storageBucket,
key: event.payload.storageKey,
sizeBytes: event.payload.fileSizeBytes,
};
this.state.status = 'draft';
break;
case 'TranscodingRequested':
this.state.transcodingJobs.set(event.payload.transcodingJobId, {
status: 'processing',
resolutions: event.payload.targetResolutions,
});
this.state.status = 'processing';
break;
case 'TranscodingCompleted':
this.state.completedTranscodings.push({
resolution: event.payload.resolution,
storageKey: event.payload.outputStorageKey,
});
break;
case 'ContentModerationCompleted':
this.state.moderationStatus = event.payload.status;
if (event.payload.status === 'rejected') {
this.state.status = 'failed';
}
break;
case 'VideoPublished':
this.state.status = 'published';
this.state.publishedAt = event.payload.publishedAt;
break;
case 'VideoMetadataUpdated':
if (event.payload.changes.title)
this.state.title = event.payload.changes.title;
if (event.payload.changes.description)
this.state.description = event.payload.changes.description;
if (event.payload.changes.tags)
this.state.tags = event.payload.changes.tags;
if (event.payload.changes.visibility)
this.state.visibility = event.payload.changes.visibility as any;
break;
case 'VideoDeleted':
this.state.status = 'deleted';
this.state.deletedAt = new Date();
break;
case 'ViewRecorded':
case 'ThumbnailGenerated':
// View events and thumbnails don't change core state,
// but we track them for event count
break;
}
}
// --- Command Handlers ---
uploadVideo(command: {
userId: string;
title: string;
description: string;
tags: string[];
fileName: string;
fileSizeBytes: number;
containerFormat: string;
storageBucket: string;
storageKey: string;
visibility: 'public' | 'private' | 'unlisted';
}): void {
if (this.state.status !== 'draft') {
throw new Error(`Cannot upload: video is in ${this.state.status} state`);
}
if (!command.title || command.title.length > 100) {
throw new Error('Title is required and must be <= 100 characters');
}
const event: VideoUploaded = {
eventId: uuidv4(),
aggregateId: this.state.videoId,
aggregateType: 'Video',
eventType: 'VideoUploaded',
timestamp: new Date(),
version: this.state.version + 1,
metadata: { userId: command.userId },
payload: {
title: command.title,
description: command.description,
tags: command.tags,
originalFileName: command.fileName,
fileSizeBytes: command.fileSizeBytes,
containerFormat: command.containerFormat,
storageBucket: command.storageBucket,
storageKey: command.storageKey,
visibility: command.visibility,
},
};
this.applyEvent(event);
this.pendingEvents.push(event);
}
requestTranscoding(targetResolutions: string[], jobId: string): void {
if (!['draft', 'processing'].includes(this.state.status)) {
throw new Error(`Cannot request transcoding in ${this.state.status} state`);
}
const event: TranscodingRequested = {
eventId: uuidv4(),
aggregateId: this.state.videoId,
aggregateType: 'Video',
eventType: 'TranscodingRequested',
timestamp: new Date(),
version: this.state.version + 1,
metadata: { userId: 'system' },
payload: {
targetResolutions,
transcodingJobId: jobId,
priority: 'normal',
},
};
this.applyEvent(event);
this.pendingEvents.push(event);
}
completeTranscoding(resolution: string, outputKey: string,
codec: string, bitrate: number, checksum: string): void {
const event: TranscodingCompleted = {
eventId: uuidv4(),
aggregateId: this.state.videoId,
aggregateType: 'Video',
eventType: 'TranscodingCompleted',
timestamp: new Date(),
version: this.state.version + 1,
metadata: { userId: 'system' },
payload: {
resolution,
codec,
bitrateKbps: bitrate,
outputStorageKey: outputKey,
fileSizeBytes: 0, // would come from storage metadata
durationSeconds: 0,
checksum,
},
};
this.applyEvent(event);
this.pendingEvents.push(event);
}
publishVideo(): void {
if (this.state.moderationStatus !== 'approved') {
throw new Error('Video must be approved by moderation before publishing');
}
if (this.state.completedTranscodings.length === 0) {
throw new Error('At least one transcoding must be completed');
}
const event: VideoPublished = {
eventId: uuidv4(),
aggregateId: this.state.videoId,
aggregateType: 'Video',
eventType: 'VideoPublished',
timestamp: new Date(),
version: this.state.version + 1,
metadata: { userId: 'system' },
payload: { publishedAt: new Date() },
};
this.applyEvent(event);
this.pendingEvents.push(event);
}
// Returns all uncommitted events for persistence
getUncommittedEvents(): VideoEvent[] {
return [...this.pendingEvents];
}
// Clears pending events after successful persistence
markEventsCommitted(): void {
this.pendingEvents = [];
}
getCurrentState(): Readonly {
return Object.freeze({ ...this.state });
}
} Step 3: The Event Store — Persisting Events
The event store is the single source of truth. It stores events immutably and allows replaying them by aggregate ID. Here's an implementation using PostgreSQL, which is practical for many teams:
// event-store.ts
import { Pool } from 'pg';
export class PostgresEventStore {
private pool: Pool;
constructor(connectionString: string) {
this.pool = new Pool({ connectionString });
}
async initialize(): Promise {
await this.pool.query(`
CREATE TABLE IF NOT EXISTS events (
event_id VARCHAR(36) PRIMARY KEY,
aggregate_id VARCHAR(36) NOT NULL,
aggregate_type VARCHAR(100) NOT NULL,
event_type VARCHAR(100) NOT NULL,
version INTEGER NOT NULL,
payload JSONB NOT NULL,
metadata JSONB NOT NULL,
timestamp TIMESTAMPTZ NOT NULL DEFAULT NOW(),
position BIGSERIAL,
UNIQUE (aggregate_id, version)
);
CREATE INDEX IF NOT EXISTS idx_events_aggregate
ON events (aggregate_id, version);
CREATE INDEX IF NOT EXISTS idx_events_position
ON events (position);
CREATE INDEX IF NOT EXISTS idx_events_type
ON events (event_type);
`);
}
async appendEvents(
aggregateId: string,
expectedVersion: number,
events: VideoEvent[]
): Promise {
const client = await this.pool.connect();
try {
await client.query('BEGIN');
// Optimistic concurrency check
const currentVersion = await client.query(
'SELECT MAX(version) as current_version FROM events WHERE aggregate_id = $1',
[aggregateId]
);
const dbVersion = currentVersion.rows[0]?.current_version || 0;
if (dbVersion !== expectedVersion) {
throw new Error(
`Concurrency conflict: expected version ${expectedVersion}, ` +
`but aggregate is at version ${dbVersion}`
);
}
// Insert all events atomically
for (const event of events) {
await client.query(
`INSERT INTO events (event_id, aggregate_id, aggregate_type,
event_type, version, payload, metadata, timestamp)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)`,
[
event.eventId,
event.aggregateId,
event.aggregateType,
event.eventType,
event.version,
JSON.stringify((event as any).payload),
JSON.stringify(event.metadata),
event.timestamp,
]
);
}
await client.query('COMMIT');
} catch (error) {
await client.query('ROLLBACK');
throw error;
} finally {
client.release();
}
}
async getEventsByAggregateId(aggregateId: string): Promise {
const result = await this.pool.query(
`SELECT * FROM events WHERE aggregate_id = $1 ORDER BY version ASC`,
[aggregateId]
);
return result.rows.map(row => ({
eventId: row.event_id,
aggregateId: row.aggregate_id,
aggregateType: row.aggregate_type,
eventType: row.event_type,
version: row.version,
timestamp: row.timestamp,
payload: row.payload,
metadata: row.metadata,
})) as VideoEvent[];
}
async getAllEventsSince(position: number, limit: number = 500): Promise {
const result = await this.pool.query(
`SELECT * FROM events WHERE position > $1 ORDER BY position ASC LIMIT $2`,
[position, limit]
);
return result.rows.map(row => ({
eventId: row.event_id,
aggregateId: row.aggregate_id,
aggregateType: row.aggregate_type,
eventType: row.event_type,
version: row.version,
timestamp: row.timestamp,
payload: row.payload,
metadata: row.metadata,
})) as VideoEvent[];
}
} Step 4: Building the Command Handler (Application Service)
The command handler coordinates between the API layer, aggregates, and the event store. It loads the aggregate from its event history, invokes the command method, and persists the new events:
// command-handler.ts
export class VideoCommandHandler {
constructor(
private eventStore: PostgresEventStore,
private eventBus: EventBus
) {}
async handleUploadVideo(command: {
videoId: string;
userId: string;
title: string;
description: string;
tags: string[];
fileName: string;
fileSizeBytes: number;
containerFormat: string;
storageBucket: string;
storageKey: string;
visibility: 'public' | 'private' | 'unlisted';
}): Promise {
// Load aggregate from event history
const aggregate = new VideoAggregate(command.videoId);
const pastEvents = await this.eventStore.getEventsByAggregateId(command.videoId);
aggregate.replayEvents(pastEvents);
const currentVersion = aggregate.getCurrentState().version;
// Execute command
aggregate.uploadVideo(command);
// Persist new events with optimistic concurrency
const newEvents = aggregate.getUncommittedEvents();
try {
await this.eventStore.appendEvents(command.videoId, currentVersion, newEvents);
aggregate.markEventsCommitted();
// Publish events to the bus for projection updates
for (const event of newEvents) {
await this.eventBus.publish(event);
}
} catch (error) {
// If concurrency conflict, retry or return error
if (error.message.includes('Concurrency conflict')) {
// Implement retry logic with exponential backoff
throw new Error('Concurrency conflict, please retry');
}
throw error;
}
}
async handlePublishVideo(videoId: string): Promise {
const aggregate = new VideoAggregate(videoId);
const pastEvents = await this.eventStore.getEventsByAggregateId(videoId);
aggregate.replayEvents(pastEvents);
const currentVersion = aggregate.getCurrentState().version;
aggregate.publishVideo();
const newEvents = aggregate.getUncommittedEvents();
await this.eventStore.appendEvents(videoId, currentVersion, newEvents);
aggregate.markEventsCommitted();
for (const event of newEvents) {
await this.eventBus.publish(event);
}
}
} Step 5: Building Read Models (Projections)
The query side listens to events and builds specialized read models. For a video streaming platform, you typically need multiple projections:
Video Catalog Projection (for browsing and search)
// projections/video-catalog-projection.ts
export class VideoCatalogProjection {
private db: any; // Your read database client (e.g., Elasticsearch, PostgreSQL)
async handleVideoUploaded(event: VideoUploaded): Promise {
await this.db.upsert('video_catalog', event.aggregateId, {
videoId: event.aggregateId,
title: event.payload.title,
description: event.payload.description,
tags: event.payload.tags,
status: 'draft',
visibility: event.payload.visibility,
uploadedAt: event.timestamp,
updatedAt: event.timestamp,
});
}
async handleVideoPublished(event: VideoPublished): Promise {
await this.db.update('video_catalog', event.aggregateId, {
status: 'published',
publishedAt: event.payload.publishedAt,
updatedAt: event.timestamp,
});
}
async handleVideoMetadataUpdated(event: VideoMetadataUpdated): Promise {
const updates: any = { updatedAt: event.timestamp };
if (event.payload.changes.title) updates.title = event.payload.changes.title;
if (event.payload.changes.description) updates.description = event.payload.changes.description;
if (event.payload.changes.tags) updates.tags = event.payload.changes.tags;
if (event.payload.changes.visibility) updates.visibility = event.payload.changes.visibility;
await this.db.update('video_catalog', event.aggregateId, updates);
}
async handleVideoDeleted(event: VideoDeleted): Promise {
await this.db.delete('video_catalog', event.aggregateId);
}
} View Count Projection (high-write, optimized in Redis)
// projections/view-counter-projection.ts
import Redis from 'ioredis';
export class ViewCounterProjection {
private redis: Redis;
constructor() {
this.redis = new Redis({
host: process.env.REDIS_HOST || 'localhost',
port: 6379,
});
}
async handleViewRecorded(event: ViewRecorded): Promise {
const videoId = event.aggregateId;
const date = event.payload.watchedAt.toISOString().split('T')[0];
// Use Redis sorted sets for real-time view counting
// Total all-time views
await this.redis.hincrby(`video:${videoId}:stats`, 'totalViews', 1);
// Daily views (for trend analysis)
await this.redis.hincrby(`video:${videoId}:stats`, `daily:${date}`, 1);
// Source breakdown
await this.redis.hincrby(
`video:${videoId}:sources`,
event.payload.source,
1
);
// Watch duration tracking
await this.redis.hincrbyfloat(
`video:${videoId}:stats`,
'totalWatchSeconds',
event.payload.watchDurationSeconds
);
// Set expiration for daily keys after 90 days
await this.redis.expire(`video:${videoId}:stats`, 7776000); // 90 days
}
async getVideoStats(videoId: string): Promise {
const stats = await this.redis.hgetall(`video:${videoId}:stats`);
const sources = await this.redis.hgetall(`video:${videoId}:sources`);
return { stats, sources };
}
} User Watch History Projection
// projections/user-watch-history-projection.ts
export class UserWatchHistoryProjection {
private db: any; // PostgreSQL read replica
async handleViewRecorded(event: ViewRecorded): Promise {
if (!event.payload.viewerId) return; // Anonymous views not stored in history
await this.db.query(
`INSERT INTO user_watch_history
(user_id, video_id, watched_at, watch_duration_seconds, source)
VALUES ($1, $2, $3, $4, $5)
ON CONFLICT (user_id, video_id) DO UPDATE SET
watched_at = $3,
watch_duration_seconds = $4,
source = $5`,
[
event.payload.viewerId,
event.aggregateId,
event.payload.watchedAt,
event.payload.watchDurationSeconds,
event.payload.source,
]
);
}
} Step 6: The Event Bus — Connecting Commands to Projections
After persisting events, you need to distribute them to all interested projections. For production systems, consider Kafka, RabbitMQ, or AWS SNS/SQS. Here's a simplified in-process bus for development:
// event-bus.ts
type EventHandler = (event: VideoEvent) => Promise;
export class EventBus {
private handlers: Map = new Map();
private deadLetterQueue: VideoEvent[] = [];
subscribe(eventType: string, handler: EventHandler): void {
const existing = this.handlers.get(eventType) || [];
existing.push(handler);
this.handlers.set(eventType, existing);
}
async publish(event: VideoEvent): Promise {
const handlers = this.handlers.get(event.eventType) || [];
for (const handler of handlers) {
try {
await handler(event);
} catch (error) {
console.error(
`Handler failed for event ${event.eventType}: ${error.message}`
);
// In production, send to dead letter queue for retry
this.deadLetterQueue.push(event);
}
}
}
// Wire up all projections
registerProjections(
catalogProjection: VideoCatalogProjection,
viewCounter: ViewCounterProjection,
watchHistory: UserWatchHistoryProjection
): void {
this.subscribe('VideoUploaded', (e) => catalogProjection.handleVideoUploaded(e as VideoUploaded));
this.subscribe('VideoPublished', (e) => catalogProjection.handleVideoPublished(e as VideoPublished));
this.subscribe('VideoMetadataUpdated', (e) => catalogProjection.handleVideoMetadataUpdated(e as VideoMetadataUpdated));
this.subscribe('VideoDeleted', (e) => catalogProjection.handleVideoDeleted(e as VideoDeleted));
this.subscribe('ViewRecorded', (e) => viewCounter.handleViewRecorded(e as ViewRecorded));
this.subscribe('ViewRecorded', (e) => watchHistory.handleViewRecorded(e as ViewRecorded));
}
} Step 7: Integrating the Video Processing Pipeline
Video processing (transcoding, thumbnail generation, content moderation) is inherently asynchronous and long-running. Events are perfect for orchestrating this. Here's how a transcoding worker integrates:
// workers/transcoding-worker.ts
import { spawn } from 'child_process';
export class TranscodingWorker {
constructor(
private eventStore: PostgresEventStore,
private eventBus: EventBus,
private commandHandler: VideoCommandHandler
) {}
async start(): Promise {
// Subscribe to TranscodingRequested events
// In production, this would be a queue consumer (Kafka, SQS)
this.eventBus.subscribe('TranscodingRequested', async (event) => {
const transcodingEvent = event as TranscodingRequested;
await this.processTranscoding(transcodingEvent);
});
}
private async processTranscoding(event: TranscodingRequested): Promise {
const { aggregateId, payload } = event;
console.log(`Starting transcoding for video ${aggregateId}`);
for (const resolution of payload.targetResolutions) {
try {
// Actual FFmpeg transcoding call
const outputKey = await this.transcodeWithFFmpeg(
aggregateId,
resolution,
payload.transcodingJobId
);
// Emit completion event via command handler
// This ensures proper aggregate state management
const aggregate = new VideoAggregate(aggregateId);
const pastEvents = await this.eventStore.getEventsByAggregateId(aggregateId);
aggregate.replayEvents(pastEvents);
aggregate.completeTranscoding(
resolution,
outputKey,
'h264',
this.getBitrateForResolution(resolution),
'checksum-here' // Would calculate actual checksum
);
const newEvents = aggregate.getUncommittedEvents();
await this.eventStore.appendEvents(
aggregateId,
aggregate.getCurrentState().version - newEvents.length,
newEvents
);
aggregate.markEventsCommitted();
for (const e of newEvents) {
await this.eventBus.publish(e);
}
} catch (error) {
console.error(`Transcoding failed for ${resolution}: ${error.message}`);
// Emit a TranscodingFailed event (not shown here for brevity)
}
}
}
private transcodeWithFFmpeg(
videoId: string,
resolution: string,
jobId: string
): Promise {
return new Promise((resolve, reject) => {
const resolutionMap: Record = {
'1080p': '1920x1080',
'720p': '1280x720',
'480p': '854x480',
'360p': '640x360',
};
const outputKey = `transcoded/${videoId}/${resolution}/output_${jobId}.mp4`;
const ffmpeg = spawn('ffmpeg', [
'-i', `input/${videoId}/original.mp4`,
'-vf', `scale=${resolutionMap[resolution] || '1280x720'}`,
'-c:v', 'libx264',
'-preset', 'medium',
'-crf', '23',
'-c:a', 'aac',
'-b:a', '128k',
'-movflags', '+faststart',
outputKey,
]);
ffmpeg.on('close', (code) => {
if (code === 0) {
resolve(outputKey);
} else {
reject(new Error(`FFmpeg exited with code ${code}`));
}
});
ffmpeg.stderr.on('data', (data) => {
// Parse FFmpeg progress for monitoring
console.log(`FFmpeg progress: ${data.toString()}`);
});
});
}
private getBitrateForResolution(resolution: string): number {
const bitrates: Record = {
'1080p': 8000,
'720p': 5000,
'480p': 2500,
'360p': 1000,
};
return bitrates[resolution] || 2500;
}
} Step 8: The Query API
With read models built, the query API is straightforward — just query the appropriate read store:
// query-api.ts
import express from 'express';
export class VideoQueryAPI {
private app = express();
private catalogDb: any; // Elasticsearch or PostgreSQL
private viewCounter: ViewCounterProjection;
private watchHistoryDb: any;
constructor() {
this.app.get('/api/videos/:videoId', this.getVideo.bind(this));
this.app.get('/api/videos/:videoId/stats', this.getVideoStats.bind(this));
this.app.get('/api/users/:userId/history', this.getWatchHistory.bind(this));
this.app.get('/api/search', this.searchVideos.bind(this));
}
private async getVideo(req: any, res: any): Promise
🚀 Need a reliable AI agent for your project?
Deploy Hermes Agent in 10 minutes. Managed hosting, zero DevOps.
Get Started — $23.99/mo