How to Build a Messaging Platform with Replit

1// Type this into Replit Agent:
2// Build a channel-based messaging platform with Express and PostgreSQL using Drizzle ORM.
3// Tables:
4// - workspaces: id serial pk, name text not null, owner_id text not null, created_at timestamp
5// - workspace_members: id serial pk, workspace_id integer FK workspaces, user_id text not null,
6//   role text default 'member' (enum: owner/admin/member), display_name text,
7//   avatar_url text, joined_at timestamp, unique(workspace_id, user_id)
8// - channels: id serial pk, workspace_id integer FK workspaces, name text not null,
9//   type text default 'public' (enum: public/private/direct), description text,
10//   created_by text not null, created_at timestamp
11// - channel_members: id serial pk, channel_id integer FK channels, user_id text not null,
12//   last_read_at timestamp default now(), unique(channel_id, user_id)
13// - messages: id serial pk, channel_id integer FK channels, sender_id text not null,
14//   content text not null, type text default 'text' (enum: text/image/file),
15//   file_url text, parent_id integer FK messages (null for top-level),
16//   edited_at timestamp, created_at timestamp default now()
17// - reactions: id serial pk, message_id integer FK messages, user_id text not null,
18//   emoji text not null, unique(message_id, user_id, emoji)
19// Create a PostgreSQL trigger: after INSERT on messages, fire
20// pg_notify('new_message', row_to_json(NEW)::text).
21// Routes: POST /api/workspaces, POST /api/workspaces/:id/invite,
22// POST /api/workspaces/:id/join, GET /api/workspaces/:id/channels,
23// POST /api/channels, GET /api/channels/:id/messages,
24// POST /api/channels/:id/messages, PATCH /api/messages/:id,
25// DELETE /api/messages/:id, POST /api/messages/:id/reactions,
26// PATCH /api/channels/:id/read, GET /api/channels/:id/stream (SSE).
27// React frontend with 3-panel layout: workspace sidebar, channel list with unread badges,
28// message list with reactions. Use Replit Auth. Bind server to 0.0.0.0.

1const { Pool } = require('pg');
2
3// Dedicated connection for LISTEN/NOTIFY (cannot use the regular pool)
4const listenClient = new Pool({
5  connectionString: process.env.DATABASE_URL,
6  max: 1,
7});
8
9// Map of channel subscriptions: channelId -> Set of SSE response objects
10const subscriptions = new Map();
11
12// Initialize LISTEN once on startup
13async function initNotifyListener() {
14  const client = await listenClient.connect();
15  await client.query('LISTEN new_message');
16
17  client.on('notification', (msg) => {
18    try {
19      const message = JSON.parse(msg.payload);
20      const channelId = message.channel_id;
21      const subscribers = subscriptions.get(channelId);
22      if (subscribers) {
23        const data = `data: ${JSON.stringify(message)}\n\n`;
24        subscribers.forEach(res => {
25          try { res.write(data); } catch (e) { /* client disconnected */ }
26        });
27      }
28    } catch (err) {
29      console.error('NOTIFY parse error:', err.message);
30    }
31  });
32
33  client.on('error', (err) => {
34    console.error('LISTEN client error:', err.message);
35    // Reconnect after 5 seconds
36    setTimeout(initNotifyListener, 5000);
37  });
38}
39
40initNotifyListener();
41
42// GET /api/channels/:id/stream — SSE endpoint
43router.get('/:id/stream', async (req, res) => {
44  const channelId = parseInt(req.params.id);
45
46  res.writeHead(200, {
47    'Content-Type': 'text/event-stream',
48    'Cache-Control': 'no-cache',
49    'Connection': 'keep-alive',
50    'X-Accel-Buffering': 'no',
51  });
52  res.write('data: {"type":"connected"}\n\n');
53
54  // Add to subscriptions
55  if (!subscriptions.has(channelId)) subscriptions.set(channelId, new Set());
56  subscriptions.get(channelId).add(res);
57
58  // Cleanup on disconnect
59  req.on('close', () => {
60    subscriptions.get(channelId)?.delete(res);
61    if (subscriptions.get(channelId)?.size === 0) subscriptions.delete(channelId);
62  });
63});

1const { messages, reactions } = require('../../shared/schema');
2const { eq, isNull, and, desc } = require('drizzle-orm');
3
4// GET /api/channels/:id/messages — cursor-based pagination
5router.get('/:id/messages', async (req, res) => {
6  const channelId = parseInt(req.params.id);
7  const { before, limit = 50 } = req.query;
8
9  const conditions = [
10    eq(messages.channelId, channelId),
11    isNull(messages.parentId), // Top-level only — threads fetched separately
12  ];
13  if (before) conditions.push(sql`${messages.id} < ${parseInt(before)}`);
14
15  const rows = await db.select().from(messages)
16    .where(and(...conditions))
17    .orderBy(desc(messages.createdAt))
18    .limit(parseInt(limit));
19
20  // Get reaction counts for all messages
21  const messageIds = rows.map(m => m.id);
22  const reactionCounts = messageIds.length > 0 ? await db.execute(
23    sql`SELECT message_id, emoji, COUNT(*) as count FROM reactions
24        WHERE message_id = ANY(${messageIds}) GROUP BY message_id, emoji`
25  ) : { rows: [] };
26
27  // Attach reactions to messages
28  const withReactions = rows.map(msg => ({
29    ...msg,
30    reactions: reactionCounts.rows.filter(r => r.message_id === msg.id),
31  }));
32
33  res.json(withReactions.reverse()); // Chronological order for the UI
34});
35
36// PATCH /api/messages/:id — edit own message
37router.patch('/:id', async (req, res) => {
38  const senderId = req.user?.id;
39  const [updated] = await db.update(messages)
40    .set({ content: req.body.content, editedAt: new Date() })
41    .where(and(eq(messages.id, parseInt(req.params.id)), eq(messages.senderId, senderId)))
42    .returning();
43  if (!updated) return res.status(403).json({ error: 'Cannot edit this message' });
44  res.json(updated);
45});
46
47// POST /api/messages/:id/reactions — add or remove emoji
48router.post('/:id/reactions', async (req, res) => {
49  const userId = req.user?.id;
50  const messageId = parseInt(req.params.id);
51  const { emoji } = req.body;
52
53  const existing = await db.select().from(reactions)
54    .where(and(
55      eq(reactions.messageId, messageId),
56      eq(reactions.userId, userId),
57      eq(reactions.emoji, emoji)
58    ));
59
60  if (existing.length > 0) {
61    await db.delete(reactions).where(eq(reactions.id, existing[0].id));
62    return res.json({ added: false, emoji });
63  }
64
65  await db.insert(reactions).values({ messageId, userId, emoji });
66  res.json({ added: true, emoji });
67});

1// PATCH /api/channels/:id/read — mark channel as read
2router.patch('/:id/read', async (req, res) => {
3  const userId = req.user?.id;
4  const channelId = parseInt(req.params.id);
5
6  await db.update(channelMembers)
7    .set({ lastReadAt: new Date() })
8    .where(and(
9      eq(channelMembers.channelId, channelId),
10      eq(channelMembers.userId, userId)
11    ));
12
13  res.json({ ok: true });
14});
15
16// GET /api/workspaces/:id/channels — channel list with unread counts
17router.get('/:workspaceId/channels', async (req, res) => {
18  const userId = req.user?.id;
19  const workspaceId = parseInt(req.params.workspaceId);
20
21  const channelsWithUnread = await db.execute(sql`
22    SELECT
23      c.id, c.name, c.type, c.description,
24      COALESCE(unread.count, 0) AS unread_count
25    FROM channels c
26    INNER JOIN channel_members cm ON cm.channel_id = c.id AND cm.user_id = ${userId}
27    LEFT JOIN LATERAL (
28      SELECT COUNT(*) as count
29      FROM messages m
30      WHERE m.channel_id = c.id
31        AND m.created_at > cm.last_read_at
32        AND m.sender_id != ${userId}
33        AND m.parent_id IS NULL
34    ) unread ON true
35    WHERE c.workspace_id = ${workspaceId}
36    ORDER BY c.type, c.name ASC
37  `);
38
39  res.json(channelsWithUnread.rows);
40});

1// server/index.js — complete setup for messaging platform
2const express = require('express');
3const path = require('path');
4const { requireAuth } = require('@replit/repl-auth');
5
6const channelsRouter = require('./routes/channels');
7const messagesRouter = require('./routes/messages');
8const workspacesRouter = require('./routes/workspaces');
9const streamRouter = require('./routes/stream');
10
11const app = express();
12app.use(express.json());
13app.use(requireAuth);
14
15// SSE stream routes (no JSON body parsing needed)
16app.use('/api/channels', streamRouter); // GET /:id/stream
17app.use('/api/channels', messagesRouter);
18app.use('/api/workspaces', workspacesRouter);
19app.use('/api/workspaces', channelsRouter);
20
21// Serve React frontend
22app.use(express.static(path.join(__dirname, '../client/dist')));
23app.get('*', (req, res) => {
24  res.sendFile(path.join(__dirname, '../client/dist/index.html'));
25});
26
27// IMPORTANT: bind to 0.0.0.0 for Replit
28const server = app.listen(5000, '0.0.0.0', () => {
29  console.log('Messaging platform running on port 5000');
30});
31
32// Keep SSE connections alive with longer timeout
33server.keepAliveTimeout = 120000;
34server.headersTimeout = 121000;