diff --git a/apps/backend/drizzle/0001_messaging.sql b/apps/backend/drizzle/0001_messaging.sql new file mode 100644 index 00000000..f5ae586d --- /dev/null +++ b/apps/backend/drizzle/0001_messaging.sql @@ -0,0 +1,27 @@ +CREATE TYPE "public"."conversation_type" AS ENUM('dm', 'group');--> statement-breakpoint +CREATE TABLE "conversation_members" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL, + "conversation_id" uuid NOT NULL, + "user_id" uuid NOT NULL, + "joined_at" timestamp DEFAULT now() NOT NULL +); +--> statement-breakpoint +CREATE TABLE "conversations" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL, + "type" "conversation_type" DEFAULT 'dm' NOT NULL, + "name" text, + "created_at" timestamp DEFAULT now() NOT NULL +); +--> statement-breakpoint +CREATE TABLE "messages" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL, + "conversation_id" uuid NOT NULL, + "sender_id" uuid NOT NULL, + "content" text NOT NULL, + "created_at" timestamp DEFAULT now() NOT NULL +); +--> statement-breakpoint +ALTER TABLE "conversation_members" ADD CONSTRAINT "conversation_members_conversation_id_conversations_id_fk" FOREIGN KEY ("conversation_id") REFERENCES "public"."conversations"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint +ALTER TABLE "conversation_members" ADD CONSTRAINT "conversation_members_user_id_users_id_fk" FOREIGN KEY ("user_id") REFERENCES "public"."users"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint +ALTER TABLE "messages" ADD CONSTRAINT "messages_conversation_id_conversations_id_fk" FOREIGN KEY ("conversation_id") REFERENCES "public"."conversations"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint +ALTER TABLE "messages" ADD CONSTRAINT "messages_sender_id_users_id_fk" FOREIGN KEY ("sender_id") REFERENCES "public"."users"("id") ON DELETE cascade ON UPDATE no action; \ No newline at end of file diff --git a/apps/backend/drizzle/meta/0001_snapshot.json b/apps/backend/drizzle/meta/0001_snapshot.json new file mode 100644 index 00000000..19501d4f --- /dev/null +++ b/apps/backend/drizzle/meta/0001_snapshot.json @@ -0,0 +1,329 @@ +{ + "id": "f0633220-541a-4c99-9b99-8242c7f3420f", + "prevId": "ccda4e45-b486-4477-892a-b4ff308b3326", + "version": "7", + "dialect": "postgresql", + "tables": { + "public.conversation_members": { + "name": "conversation_members", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "conversation_id": { + "name": "conversation_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "user_id": { + "name": "user_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "joined_at": { + "name": "joined_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "conversation_members_conversation_id_conversations_id_fk": { + "name": "conversation_members_conversation_id_conversations_id_fk", + "tableFrom": "conversation_members", + "tableTo": "conversations", + "columnsFrom": [ + "conversation_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "conversation_members_user_id_users_id_fk": { + "name": "conversation_members_user_id_users_id_fk", + "tableFrom": "conversation_members", + "tableTo": "users", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.conversations": { + "name": "conversations", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "type": { + "name": "type", + "type": "conversation_type", + "typeSchema": "public", + "primaryKey": false, + "notNull": true, + "default": "'dm'" + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.messages": { + "name": "messages", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "conversation_id": { + "name": "conversation_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "sender_id": { + "name": "sender_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "content": { + "name": "content", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "messages_conversation_id_conversations_id_fk": { + "name": "messages_conversation_id_conversations_id_fk", + "tableFrom": "messages", + "tableTo": "conversations", + "columnsFrom": [ + "conversation_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "messages_sender_id_users_id_fk": { + "name": "messages_sender_id_users_id_fk", + "tableFrom": "messages", + "tableTo": "users", + "columnsFrom": [ + "sender_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.users": { + "name": "users", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "username": { + "name": "username", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "avatar_url": { + "name": "avatar_url", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "users_username_unique": { + "name": "users_username_unique", + "nullsNotDistinct": false, + "columns": [ + "username" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.wallets": { + "name": "wallets", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "user_id": { + "name": "user_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "address": { + "name": "address", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "is_primary": { + "name": "is_primary", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "wallets_user_id_users_id_fk": { + "name": "wallets_user_id_users_id_fk", + "tableFrom": "wallets", + "tableTo": "users", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "wallets_address_unique": { + "name": "wallets_address_unique", + "nullsNotDistinct": false, + "columns": [ + "address" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + } + }, + "enums": { + "public.conversation_type": { + "name": "conversation_type", + "schema": "public", + "values": [ + "dm", + "group" + ] + } + }, + "schemas": {}, + "sequences": {}, + "roles": {}, + "policies": {}, + "views": {}, + "_meta": { + "columns": {}, + "schemas": {}, + "tables": {} + } +} \ No newline at end of file diff --git a/apps/backend/drizzle/meta/_journal.json b/apps/backend/drizzle/meta/_journal.json index 335366f9..233d680d 100644 --- a/apps/backend/drizzle/meta/_journal.json +++ b/apps/backend/drizzle/meta/_journal.json @@ -8,6 +8,13 @@ "when": 1778589700461, "tag": "0000_init", "breakpoints": true + }, + { + "idx": 1, + "version": "7", + "when": 1778597011664, + "tag": "0001_messaging", + "breakpoints": true } ] } \ No newline at end of file diff --git a/apps/backend/src/db/schema.ts b/apps/backend/src/db/schema.ts index 404006f8..00a14055 100644 --- a/apps/backend/src/db/schema.ts +++ b/apps/backend/src/db/schema.ts @@ -1,4 +1,4 @@ -import { pgTable, text, timestamp, uuid, boolean } from 'drizzle-orm/pg-core'; +import { pgTable, text, timestamp, uuid, boolean, pgEnum } from 'drizzle-orm/pg-core'; import { relations } from 'drizzle-orm'; export const users = pgTable('users', { @@ -19,15 +19,81 @@ export const wallets = pgTable('wallets', { createdAt: timestamp('created_at').notNull().defaultNow(), }); +// ─── Conversations ──────────────────────────────────────────────────────────── + +export const conversationTypeEnum = pgEnum('conversation_type', ['dm', 'group']); + +export const conversations = pgTable('conversations', { + id: uuid('id').primaryKey().defaultRandom(), + type: conversationTypeEnum('type').notNull().default('dm'), + name: text('name'), + createdAt: timestamp('created_at').notNull().defaultNow(), +}); + +export const conversationMembers = pgTable('conversation_members', { + id: uuid('id').primaryKey().defaultRandom(), + conversationId: uuid('conversation_id') + .notNull() + .references(() => conversations.id, { onDelete: 'cascade' }), + userId: uuid('user_id') + .notNull() + .references(() => users.id, { onDelete: 'cascade' }), + joinedAt: timestamp('joined_at').notNull().defaultNow(), +}); + +export const messages = pgTable('messages', { + id: uuid('id').primaryKey().defaultRandom(), + conversationId: uuid('conversation_id') + .notNull() + .references(() => conversations.id, { onDelete: 'cascade' }), + senderId: uuid('sender_id') + .notNull() + .references(() => users.id, { onDelete: 'cascade' }), + content: text('content').notNull(), + createdAt: timestamp('created_at').notNull().defaultNow(), +}); + +// ─── Relations ──────────────────────────────────────────────────────────────── + export const usersRelations = relations(users, ({ many }) => ({ wallets: many(wallets), + memberships: many(conversationMembers), + messages: many(messages), })); export const walletsRelations = relations(wallets, ({ one }) => ({ user: one(users, { fields: [wallets.userId], references: [users.id] }), })); +export const conversationsRelations = relations(conversations, ({ many }) => ({ + members: many(conversationMembers), + messages: many(messages), +})); + +export const conversationMembersRelations = relations(conversationMembers, ({ one }) => ({ + conversation: one(conversations, { + fields: [conversationMembers.conversationId], + references: [conversations.id], + }), + user: one(users, { fields: [conversationMembers.userId], references: [users.id] }), +})); + +export const messagesRelations = relations(messages, ({ one }) => ({ + conversation: one(conversations, { + fields: [messages.conversationId], + references: [conversations.id], + }), + sender: one(users, { fields: [messages.senderId], references: [users.id] }), +})); + +// ─── Types ──────────────────────────────────────────────────────────────────── + export type User = typeof users.$inferSelect; export type NewUser = typeof users.$inferInsert; export type Wallet = typeof wallets.$inferSelect; export type NewWallet = typeof wallets.$inferInsert; +export type Conversation = typeof conversations.$inferSelect; +export type NewConversation = typeof conversations.$inferInsert; +export type ConversationMember = typeof conversationMembers.$inferSelect; +export type Message = typeof messages.$inferSelect; +export type NewMessage = typeof messages.$inferInsert; diff --git a/apps/backend/src/index.ts b/apps/backend/src/index.ts index 2fb96d75..642150a9 100644 --- a/apps/backend/src/index.ts +++ b/apps/backend/src/index.ts @@ -7,8 +7,10 @@ import morgan from 'morgan'; import { sql } from 'drizzle-orm'; import { db } from './db/index.js'; import { authRouter } from './routes/auth.js'; +import { conversationsRouter } from './routes/conversations.js'; import { requireAuth } from './middleware/auth.js'; import { socketAuthMiddleware, type AuthSocket } from './middleware/socketAuth.js'; +import { registerMessagingHandlers } from './socket/messaging.js'; dotenv.config(); @@ -32,6 +34,7 @@ app.get('/health', async (_req, res) => { }); app.use('/auth', authRouter); +app.use('/conversations', conversationsRouter); // Protected route example app.get('/me', requireAuth, (req, res) => { @@ -42,6 +45,7 @@ io.use(socketAuthMiddleware); io.on('connection', (socket: AuthSocket) => { console.log('User connected:', socket.auth?.userId, socket.id); + registerMessagingHandlers(io, socket); socket.on('disconnect', () => { console.log('User disconnected:', socket.auth?.userId); }); diff --git a/apps/backend/src/routes/conversations.ts b/apps/backend/src/routes/conversations.ts new file mode 100644 index 00000000..72d1954d --- /dev/null +++ b/apps/backend/src/routes/conversations.ts @@ -0,0 +1,26 @@ +import { Router } from 'express'; +import { eq } from 'drizzle-orm'; +import { db } from '../db/index.js'; +import { conversationMembers } from '../db/schema.js'; +import { requireAuth, type AuthRequest } from '../middleware/auth.js'; + +export const conversationsRouter = Router(); + +conversationsRouter.use(requireAuth); + +// List all conversations the authenticated user belongs to +conversationsRouter.get('/', async (req: AuthRequest, res) => { + const userId = req.auth!.userId; + + const memberships = await db.query.conversationMembers.findMany({ + where: eq(conversationMembers.userId, userId), + with: { + conversation: { + with: { members: { with: { user: { columns: { id: true, username: true, avatarUrl: true } } } } }, + }, + }, + }); + + const result = memberships.map((m) => m.conversation); + res.json(result); +}); diff --git a/apps/backend/src/socket/messaging.ts b/apps/backend/src/socket/messaging.ts new file mode 100644 index 00000000..71a494ca --- /dev/null +++ b/apps/backend/src/socket/messaging.ts @@ -0,0 +1,137 @@ +import type { Server } from 'socket.io'; +import { and, eq, lt, desc } from 'drizzle-orm'; +import { db } from '../db/index.js'; +import { + conversations, + conversationMembers, + messages, +} from '../db/schema.js'; +import type { AuthSocket } from '../middleware/socketAuth.js'; + +const PAGE_SIZE = 30; + +export function registerMessagingHandlers(io: Server, socket: AuthSocket): void { + const userId = socket.auth!.userId; + + // ── join_room ────────────────────────────────────────────────────────────── + // Payload: { conversationId: string } + // Guards that the caller is a member before subscribing them to the room. + socket.on('join_room', async (payload: { conversationId: string }) => { + const { conversationId } = payload; + + const membership = await db.query.conversationMembers.findFirst({ + where: and( + eq(conversationMembers.conversationId, conversationId), + eq(conversationMembers.userId, userId), + ), + }); + + if (!membership) { + socket.emit('error', { event: 'join_room', message: 'Not a member of this conversation' }); + return; + } + + await socket.join(conversationId); + socket.emit('room_joined', { conversationId }); + }); + + // ── send_message ─────────────────────────────────────────────────────────── + // Payload: { conversationId: string; content: string } + // Persists the message and broadcasts it to all room members. + socket.on('send_message', async (payload: { conversationId: string; content: string }) => { + const { conversationId, content } = payload; + + if (!content?.trim()) { + socket.emit('error', { event: 'send_message', message: 'Content must not be empty' }); + return; + } + + const membership = await db.query.conversationMembers.findFirst({ + where: and( + eq(conversationMembers.conversationId, conversationId), + eq(conversationMembers.userId, userId), + ), + }); + + if (!membership) { + socket.emit('error', { event: 'send_message', message: 'Not a member of this conversation' }); + return; + } + + const [message] = await db + .insert(messages) + .values({ conversationId, senderId: userId, content: content.trim() }) + .returning(); + + io.to(conversationId).emit('new_message', message); + }); + + // ── message_history ──────────────────────────────────────────────────────── + // Payload: { conversationId: string; before?: string } (before = message id cursor) + // Returns the last PAGE_SIZE messages, optionally before a cursor for pagination. + socket.on( + 'message_history', + async (payload: { conversationId: string; before?: string }) => { + const { conversationId, before } = payload; + + const membership = await db.query.conversationMembers.findFirst({ + where: and( + eq(conversationMembers.conversationId, conversationId), + eq(conversationMembers.userId, userId), + ), + }); + + if (!membership) { + socket.emit('error', { event: 'message_history', message: 'Not a member of this conversation' }); + return; + } + + let cursor: Date | undefined; + if (before) { + const ref = await db.query.messages.findFirst({ + where: eq(messages.id, before), + }); + cursor = ref?.createdAt; + } + + const history = await db.query.messages.findMany({ + where: cursor + ? and(eq(messages.conversationId, conversationId), lt(messages.createdAt, cursor)) + : eq(messages.conversationId, conversationId), + orderBy: desc(messages.createdAt), + limit: PAGE_SIZE, + with: { sender: { columns: { id: true, username: true, avatarUrl: true } } }, + }); + + socket.emit('message_history', { conversationId, messages: history.reverse() }); + }, + ); + + // ── create_conversation ──────────────────────────────────────────────────── + // Payload: { type: 'dm'|'group'; name?: string; memberIds: string[] } + // Creates a conversation and adds all members (including caller). + socket.on( + 'create_conversation', + async (payload: { type: 'dm' | 'group'; name?: string; memberIds: string[] }) => { + const { type, name, memberIds } = payload; + + const allMembers = Array.from(new Set([userId, ...memberIds])); + + const [conversation] = await db + .insert(conversations) + .values({ type, name }) + .returning(); + + if (!conversation) { + socket.emit('error', { event: 'create_conversation', message: 'Failed to create conversation' }); + return; + } + + await db.insert(conversationMembers).values( + allMembers.map((uid) => ({ conversationId: conversation.id, userId: uid })), + ); + + socket.emit('conversation_created', conversation); + }, + ); +}