feat: initial JaniChat messenger — PWA, WebSocket, admin panel
This commit is contained in:
Ai
202
backend/src/ws.ts
Normal file
202
backend/src/ws.ts
Normal file
@@ -0,0 +1,202 @@
|
||||
import { FastifyInstance } from 'fastify';
|
||||
import { WebSocket } from 'ws';
|
||||
import { pool } from './db.js';
|
||||
import webpush from 'web-push';
|
||||
|
||||
// userId -> Set of WebSocket connections
|
||||
export const connections = new Map<string, Set<WebSocket>>();
|
||||
|
||||
function broadcast(userIds: string[], data: object) {
|
||||
const msg = JSON.stringify(data);
|
||||
for (const uid of userIds) {
|
||||
const sockets = connections.get(uid);
|
||||
if (sockets) {
|
||||
sockets.forEach(ws => {
|
||||
if (ws.readyState === WebSocket.OPEN) ws.send(msg);
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async function getChatMemberIds(chatId: string): Promise<string[]> {
|
||||
const { rows } = await pool.query(
|
||||
'SELECT user_id FROM chat_members WHERE chat_id = $1', [chatId]
|
||||
);
|
||||
return rows.map(r => r.user_id);
|
||||
}
|
||||
|
||||
async function sendPushToOfflineUsers(userIds: string[], payload: object) {
|
||||
const online = new Set(connections.keys());
|
||||
const offline = userIds.filter(id => !online.has(id));
|
||||
if (offline.length === 0) return;
|
||||
|
||||
const { rows } = await pool.query(
|
||||
'SELECT endpoint, auth_key, p256dh FROM push_subscriptions WHERE user_id = ANY($1)',
|
||||
[offline]
|
||||
);
|
||||
|
||||
const body = JSON.stringify(payload);
|
||||
for (const sub of rows) {
|
||||
try {
|
||||
await webpush.sendNotification(
|
||||
{ endpoint: sub.endpoint, keys: { auth: sub.auth_key, p256dh: sub.p256dh } },
|
||||
body
|
||||
);
|
||||
} catch (e: any) {
|
||||
if (e.statusCode === 410) {
|
||||
await pool.query('DELETE FROM push_subscriptions WHERE endpoint = $1', [sub.endpoint]);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export function setupWebSocket(app: FastifyInstance) {
|
||||
app.get('/ws', { websocket: true }, async (socket, req) => {
|
||||
let userId: string | null = null;
|
||||
|
||||
try {
|
||||
const token = (req.query as any).token;
|
||||
const decoded = app.jwt.verify<{ id: string }>(token);
|
||||
userId = decoded.id;
|
||||
} catch {
|
||||
socket.close(1008, 'Unauthorized');
|
||||
return;
|
||||
}
|
||||
|
||||
// Register connection
|
||||
if (!connections.has(userId)) connections.set(userId, new Set());
|
||||
connections.get(userId)!.add(socket);
|
||||
|
||||
// Update last_seen
|
||||
await pool.query('UPDATE users SET last_seen = NOW() WHERE id = $1', [userId]);
|
||||
|
||||
// Notify others that user is online
|
||||
const { rows: memberChats } = await pool.query(
|
||||
'SELECT DISTINCT chat_id FROM chat_members WHERE user_id = $1', [userId]
|
||||
);
|
||||
for (const row of memberChats) {
|
||||
const ids = await getChatMemberIds(row.chat_id);
|
||||
broadcast(ids.filter(id => id !== userId), { type: 'user_online', payload: { userId, online: true } });
|
||||
}
|
||||
|
||||
socket.on('message', async (raw) => {
|
||||
try {
|
||||
const { type, payload } = JSON.parse(raw.toString());
|
||||
|
||||
if (type === 'typing') {
|
||||
const members = await getChatMemberIds(payload.chatId);
|
||||
broadcast(members.filter(id => id !== userId), {
|
||||
type: 'typing',
|
||||
payload: { chatId: payload.chatId, userId, typing: payload.typing }
|
||||
});
|
||||
}
|
||||
|
||||
if (type === 'read_messages') {
|
||||
await pool.query(
|
||||
`UPDATE chat_members SET last_read_at = NOW() WHERE chat_id = $1 AND user_id = $2`,
|
||||
[payload.chatId, userId]
|
||||
);
|
||||
const members = await getChatMemberIds(payload.chatId);
|
||||
broadcast(members.filter(id => id !== userId), {
|
||||
type: 'messages_read',
|
||||
payload: { chatId: payload.chatId, userId }
|
||||
});
|
||||
}
|
||||
|
||||
if (type === 'send_message') {
|
||||
// Check membership
|
||||
const { rows: [member] } = await pool.query(
|
||||
'SELECT role, can_send_messages FROM chat_members WHERE chat_id = $1 AND user_id = $2',
|
||||
[payload.chatId, userId]
|
||||
);
|
||||
if (!member) return;
|
||||
|
||||
const { rows: [chat] } = await pool.query('SELECT type FROM chats WHERE id = $1', [payload.chatId]);
|
||||
// Channels: only owner/admin can send
|
||||
if (chat.type === 'channel' && !['owner', 'admin'].includes(member.role)) return;
|
||||
if (!member.can_send_messages && member.role === 'member') return;
|
||||
|
||||
const { rows: [msg] } = await pool.query(
|
||||
`INSERT INTO messages (chat_id, sender_id, content, type, reply_to_id)
|
||||
VALUES ($1, $2, $3, $4, $5)
|
||||
RETURNING *`,
|
||||
[payload.chatId, userId, payload.content, payload.type || 'text', payload.replyToId || null]
|
||||
);
|
||||
|
||||
// Get sender info
|
||||
const { rows: [sender] } = await pool.query(
|
||||
'SELECT id, username, display_name, avatar_color FROM users WHERE id = $1', [userId]
|
||||
);
|
||||
|
||||
const fullMsg = { ...msg, sender };
|
||||
|
||||
const members = await getChatMemberIds(payload.chatId);
|
||||
broadcast(members, { type: 'new_message', payload: fullMsg });
|
||||
|
||||
// Push to offline users
|
||||
await sendPushToOfflineUsers(members, {
|
||||
title: sender.display_name,
|
||||
body: payload.content.substring(0, 100),
|
||||
chatId: payload.chatId
|
||||
});
|
||||
|
||||
// Update sender's last_read_at
|
||||
await pool.query(
|
||||
'UPDATE chat_members SET last_read_at = NOW() WHERE chat_id = $1 AND user_id = $2',
|
||||
[payload.chatId, userId]
|
||||
);
|
||||
}
|
||||
|
||||
if (type === 'delete_message') {
|
||||
const { rows: [msg] } = await pool.query(
|
||||
'SELECT sender_id, chat_id FROM messages WHERE id = $1', [payload.messageId]
|
||||
);
|
||||
if (!msg) return;
|
||||
|
||||
const { rows: [member] } = await pool.query(
|
||||
'SELECT role FROM chat_members WHERE chat_id = $1 AND user_id = $2', [msg.chat_id, userId]
|
||||
);
|
||||
if (msg.sender_id !== userId && !['owner', 'admin'].includes(member?.role)) return;
|
||||
|
||||
await pool.query('UPDATE messages SET is_deleted = TRUE WHERE id = $1', [payload.messageId]);
|
||||
const members = await getChatMemberIds(msg.chat_id);
|
||||
broadcast(members, { type: 'message_deleted', payload: { messageId: payload.messageId, chatId: msg.chat_id } });
|
||||
}
|
||||
|
||||
if (type === 'edit_message') {
|
||||
const { rows: [msg] } = await pool.query(
|
||||
'SELECT sender_id, chat_id FROM messages WHERE id = $1', [payload.messageId]
|
||||
);
|
||||
if (!msg || msg.sender_id !== userId) return;
|
||||
|
||||
const { rows: [updated] } = await pool.query(
|
||||
'UPDATE messages SET content = $1, is_edited = TRUE, edited_at = NOW() WHERE id = $2 RETURNING *',
|
||||
[payload.content, payload.messageId]
|
||||
);
|
||||
const members = await getChatMemberIds(msg.chat_id);
|
||||
broadcast(members, { type: 'message_edited', payload: updated });
|
||||
}
|
||||
|
||||
} catch (e) {
|
||||
console.error('WS message error:', e);
|
||||
}
|
||||
});
|
||||
|
||||
socket.on('close', async () => {
|
||||
if (!userId) return;
|
||||
connections.get(userId)!.delete(socket);
|
||||
if (connections.get(userId)!.size === 0) {
|
||||
connections.delete(userId);
|
||||
await pool.query('UPDATE users SET last_seen = NOW() WHERE id = $1', [userId]);
|
||||
|
||||
const { rows: memberChats } = await pool.query(
|
||||
'SELECT DISTINCT chat_id FROM chat_members WHERE user_id = $1', [userId]
|
||||
);
|
||||
for (const row of memberChats) {
|
||||
const ids = await getChatMemberIds(row.chat_id);
|
||||
broadcast(ids, { type: 'user_online', payload: { userId, online: false } });
|
||||
}
|
||||
}
|
||||
});
|
||||
});
|
||||
}
|
||||
Reference in New Issue
Block a user