feat: initial mIRCcloud.com scaffold
- Express + TypeScript backend with IRC connection manager - React + Vite frontend with mIRC-inspired theme - Drizzle ORM + PostgreSQL schema (users, connections, channels, messages) - Redis pub/sub for real-time WebSocket delivery across replicas - IRC color code parser (16 + 99 extended palette) - Kubernetes manifests for homecloud deployment - Multi-stage Dockerfile with non-root user - docker-compose for local development
This commit is contained in:
Jason Preston
@@ -0,0 +1,400 @@
|
||||
import { WebSocketServer, WebSocket } from 'ws';
|
||||
import { Server as HttpServer } from 'http';
|
||||
import Redis from 'ioredis';
|
||||
import { config } from '../config.js';
|
||||
import { authenticateWs, JwtPayload } from '../auth/index.js';
|
||||
import { connectionManager } from '../irc/connection-manager.js';
|
||||
import { db } from '../db/index.js';
|
||||
import { connections, messages, readMarkers } from '../db/schema.js';
|
||||
import { eq, and, lt, desc } from 'drizzle-orm';
|
||||
|
||||
interface AuthenticatedSocket {
|
||||
ws: WebSocket;
|
||||
user: JwtPayload;
|
||||
subscriber: Redis;
|
||||
alive: boolean;
|
||||
}
|
||||
|
||||
const activeSockets: Map<number, Set<AuthenticatedSocket>> = new Map();
|
||||
|
||||
export function createWebSocketServer(server: HttpServer): WebSocketServer {
|
||||
const wss = new WebSocketServer({ server, path: '/ws' });
|
||||
|
||||
const heartbeatInterval = setInterval(() => {
|
||||
for (const [, sockets] of activeSockets) {
|
||||
for (const socket of sockets) {
|
||||
if (!socket.alive) {
|
||||
socket.ws.terminate();
|
||||
continue;
|
||||
}
|
||||
socket.alive = false;
|
||||
socket.ws.ping();
|
||||
}
|
||||
}
|
||||
}, 30000);
|
||||
|
||||
wss.on('close', () => {
|
||||
clearInterval(heartbeatInterval);
|
||||
});
|
||||
|
||||
wss.on('connection', async (ws, req) => {
|
||||
const user = authenticateWs(req);
|
||||
|
||||
if (!user) {
|
||||
ws.close(4001, 'Authentication failed');
|
||||
return;
|
||||
}
|
||||
|
||||
const subscriber = new Redis(config.REDIS_URL);
|
||||
|
||||
const socket: AuthenticatedSocket = {
|
||||
ws,
|
||||
user,
|
||||
subscriber,
|
||||
alive: true,
|
||||
};
|
||||
|
||||
let userSockets = activeSockets.get(user.userId);
|
||||
if (!userSockets) {
|
||||
userSockets = new Set();
|
||||
activeSockets.set(user.userId, userSockets);
|
||||
}
|
||||
userSockets.add(socket);
|
||||
|
||||
await subscriber.subscribe(`irc:events:${user.userId}`);
|
||||
|
||||
subscriber.on('message', (_channel: string, message: string) => {
|
||||
if (ws.readyState === WebSocket.OPEN) {
|
||||
ws.send(message);
|
||||
}
|
||||
});
|
||||
|
||||
ws.on('pong', () => {
|
||||
socket.alive = true;
|
||||
});
|
||||
|
||||
ws.on('message', async (data) => {
|
||||
try {
|
||||
const msg = JSON.parse(data.toString());
|
||||
await handleClientCommand(socket, msg);
|
||||
} catch (err) {
|
||||
const errorMsg = err instanceof Error ? err.message : 'Invalid message';
|
||||
ws.send(JSON.stringify({ type: 'error', error: errorMsg }));
|
||||
}
|
||||
});
|
||||
|
||||
ws.on('close', async () => {
|
||||
await subscriber.unsubscribe(`irc:events:${user.userId}`);
|
||||
await subscriber.quit();
|
||||
|
||||
const sockets = activeSockets.get(user.userId);
|
||||
if (sockets) {
|
||||
sockets.delete(socket);
|
||||
if (sockets.size === 0) {
|
||||
activeSockets.delete(user.userId);
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
ws.on('error', () => {
|
||||
ws.terminate();
|
||||
});
|
||||
|
||||
await sendInitialState(socket);
|
||||
});
|
||||
|
||||
return wss;
|
||||
}
|
||||
|
||||
async function sendInitialState(socket: AuthenticatedSocket): Promise<void> {
|
||||
const { user, ws } = socket;
|
||||
|
||||
const userConnections = await db.query.connections.findMany({
|
||||
where: eq(connections.userId, user.userId),
|
||||
with: {
|
||||
channels: true,
|
||||
},
|
||||
});
|
||||
|
||||
const connectionsState = userConnections.map((conn) => ({
|
||||
id: conn.id,
|
||||
name: conn.name,
|
||||
hostname: conn.hostname,
|
||||
port: conn.port,
|
||||
tls: conn.tls,
|
||||
nick: conn.nick,
|
||||
connected: conn.connected,
|
||||
channels: conn.channels.map((ch) => ({
|
||||
id: ch.id,
|
||||
name: ch.name,
|
||||
joined: ch.joined,
|
||||
topic: ch.topic,
|
||||
})),
|
||||
}));
|
||||
|
||||
ws.send(
|
||||
JSON.stringify({
|
||||
type: 'initial_state',
|
||||
connections: connectionsState,
|
||||
})
|
||||
);
|
||||
|
||||
for (const conn of userConnections) {
|
||||
for (const ch of conn.channels) {
|
||||
if (ch.joined) {
|
||||
const recentMessages = await db.query.messages.findMany({
|
||||
where: and(
|
||||
eq(messages.connectionId, conn.id),
|
||||
eq(messages.channelName, ch.name)
|
||||
),
|
||||
orderBy: [desc(messages.timestamp)],
|
||||
limit: 50,
|
||||
});
|
||||
|
||||
ws.send(
|
||||
JSON.stringify({
|
||||
type: 'backlog',
|
||||
connectionId: conn.id,
|
||||
buffer: ch.name,
|
||||
messages: recentMessages.reverse(),
|
||||
})
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
const consoleMessages = await db.query.messages.findMany({
|
||||
where: and(
|
||||
eq(messages.connectionId, conn.id),
|
||||
eq(messages.bufferType, 'console')
|
||||
),
|
||||
orderBy: [desc(messages.timestamp)],
|
||||
limit: 50,
|
||||
});
|
||||
|
||||
if (consoleMessages.length > 0) {
|
||||
ws.send(
|
||||
JSON.stringify({
|
||||
type: 'backlog',
|
||||
connectionId: conn.id,
|
||||
buffer: null,
|
||||
messages: consoleMessages.reverse(),
|
||||
})
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async function handleClientCommand(
|
||||
socket: AuthenticatedSocket,
|
||||
msg: Record<string, unknown>
|
||||
): Promise<void> {
|
||||
const { user, ws } = socket;
|
||||
|
||||
switch (msg.command) {
|
||||
case 'send_message': {
|
||||
const connectionId = msg.connectionId as number;
|
||||
const target = msg.target as string;
|
||||
const text = msg.text as string;
|
||||
|
||||
const client = connectionManager.getConnection(user.userId, connectionId);
|
||||
if (!client) {
|
||||
ws.send(JSON.stringify({ type: 'error', error: 'Connection not found or not active' }));
|
||||
return;
|
||||
}
|
||||
|
||||
client.say(target, text);
|
||||
break;
|
||||
}
|
||||
|
||||
case 'join_channel': {
|
||||
const connectionId = msg.connectionId as number;
|
||||
const channel = msg.channel as string;
|
||||
const key = msg.key as string | undefined;
|
||||
|
||||
const client = connectionManager.getConnection(user.userId, connectionId);
|
||||
if (!client) {
|
||||
ws.send(JSON.stringify({ type: 'error', error: 'Connection not found or not active' }));
|
||||
return;
|
||||
}
|
||||
|
||||
if (key) {
|
||||
client.join(channel, key);
|
||||
} else {
|
||||
client.join(channel);
|
||||
}
|
||||
break;
|
||||
}
|
||||
|
||||
case 'part_channel': {
|
||||
const connectionId = msg.connectionId as number;
|
||||
const channel = msg.channel as string;
|
||||
const reason = msg.reason as string | undefined;
|
||||
|
||||
const client = connectionManager.getConnection(user.userId, connectionId);
|
||||
if (!client) {
|
||||
ws.send(JSON.stringify({ type: 'error', error: 'Connection not found or not active' }));
|
||||
return;
|
||||
}
|
||||
|
||||
client.part(channel, reason || '');
|
||||
break;
|
||||
}
|
||||
|
||||
case 'get_backlog': {
|
||||
const connectionId = msg.connectionId as number;
|
||||
const buffer = msg.buffer as string | null;
|
||||
const before = msg.before as string | undefined;
|
||||
const limit = Math.min((msg.limit as number) || 100, 500);
|
||||
|
||||
const conditions = [eq(messages.connectionId, connectionId)];
|
||||
|
||||
if (buffer) {
|
||||
conditions.push(eq(messages.channelName, buffer));
|
||||
} else {
|
||||
conditions.push(eq(messages.bufferType, 'console'));
|
||||
}
|
||||
|
||||
if (before) {
|
||||
conditions.push(lt(messages.id, BigInt(before)));
|
||||
}
|
||||
|
||||
const result = await db.query.messages.findMany({
|
||||
where: and(...conditions),
|
||||
orderBy: [desc(messages.timestamp)],
|
||||
limit,
|
||||
});
|
||||
|
||||
ws.send(
|
||||
JSON.stringify({
|
||||
type: 'backlog',
|
||||
connectionId,
|
||||
buffer,
|
||||
messages: result.reverse(),
|
||||
})
|
||||
);
|
||||
break;
|
||||
}
|
||||
|
||||
case 'mark_read': {
|
||||
const connectionId = msg.connectionId as number;
|
||||
const channelName = (msg.channel as string) || null;
|
||||
const lastReadId = msg.lastReadId as string;
|
||||
|
||||
const existing = await db.query.readMarkers.findFirst({
|
||||
where: and(
|
||||
eq(readMarkers.userId, user.userId),
|
||||
eq(readMarkers.connectionId, connectionId),
|
||||
channelName
|
||||
? eq(readMarkers.channelName, channelName)
|
||||
: eq(readMarkers.channelName, '')
|
||||
),
|
||||
});
|
||||
|
||||
if (existing) {
|
||||
await db
|
||||
.update(readMarkers)
|
||||
.set({ lastReadId: BigInt(lastReadId), updatedAt: new Date() })
|
||||
.where(eq(readMarkers.id, existing.id));
|
||||
} else {
|
||||
await db.insert(readMarkers).values({
|
||||
userId: user.userId,
|
||||
connectionId,
|
||||
channelName,
|
||||
lastReadId: BigInt(lastReadId),
|
||||
});
|
||||
}
|
||||
|
||||
ws.send(JSON.stringify({ type: 'mark_read_ack', connectionId, channel: channelName }));
|
||||
break;
|
||||
}
|
||||
|
||||
case 'add_connection': {
|
||||
const connData = msg.data as Record<string, unknown>;
|
||||
|
||||
const [newConn] = await db
|
||||
.insert(connections)
|
||||
.values({
|
||||
userId: user.userId,
|
||||
name: connData.name as string,
|
||||
hostname: connData.hostname as string,
|
||||
port: (connData.port as number) || 6667,
|
||||
tls: (connData.tls as boolean) || false,
|
||||
nick: connData.nick as string,
|
||||
username: (connData.username as string) || null,
|
||||
realname: (connData.realname as string) || null,
|
||||
password: (connData.password as string) || null,
|
||||
saslUsername: (connData.saslUsername as string) || null,
|
||||
saslPassword: (connData.saslPassword as string) || null,
|
||||
autoConnect: (connData.autoConnect as boolean) || false,
|
||||
})
|
||||
.returning();
|
||||
|
||||
ws.send(JSON.stringify({ type: 'connection_added', connection: newConn }));
|
||||
break;
|
||||
}
|
||||
|
||||
case 'remove_connection': {
|
||||
const connectionId = msg.connectionId as number;
|
||||
|
||||
const conn = await db.query.connections.findFirst({
|
||||
where: and(
|
||||
eq(connections.id, connectionId),
|
||||
eq(connections.userId, user.userId)
|
||||
),
|
||||
});
|
||||
|
||||
if (!conn) {
|
||||
ws.send(JSON.stringify({ type: 'error', error: 'Connection not found' }));
|
||||
return;
|
||||
}
|
||||
|
||||
await connectionManager.disconnect(user.userId, connectionId);
|
||||
await db.delete(connections).where(eq(connections.id, connectionId));
|
||||
|
||||
ws.send(JSON.stringify({ type: 'connection_removed', connectionId }));
|
||||
break;
|
||||
}
|
||||
|
||||
case 'list_connections': {
|
||||
const userConnections = await db.query.connections.findMany({
|
||||
where: eq(connections.userId, user.userId),
|
||||
with: { channels: true },
|
||||
});
|
||||
|
||||
ws.send(
|
||||
JSON.stringify({
|
||||
type: 'connections_list',
|
||||
connections: userConnections.map((conn) => ({
|
||||
id: conn.id,
|
||||
name: conn.name,
|
||||
hostname: conn.hostname,
|
||||
port: conn.port,
|
||||
tls: conn.tls,
|
||||
nick: conn.nick,
|
||||
connected: conn.connected,
|
||||
channels: conn.channels.map((ch) => ({
|
||||
id: ch.id,
|
||||
name: ch.name,
|
||||
joined: ch.joined,
|
||||
topic: ch.topic,
|
||||
})),
|
||||
})),
|
||||
})
|
||||
);
|
||||
break;
|
||||
}
|
||||
|
||||
default:
|
||||
ws.send(JSON.stringify({ type: 'error', error: `Unknown command: ${msg.command}` }));
|
||||
}
|
||||
}
|
||||
|
||||
export function closeAllSockets(): void {
|
||||
for (const [, sockets] of activeSockets) {
|
||||
for (const socket of sockets) {
|
||||
socket.subscriber.quit();
|
||||
socket.ws.close(1001, 'Server shutting down');
|
||||
}
|
||||
}
|
||||
activeSockets.clear();
|
||||
}
|
||||
Reference in New Issue
Block a user