From db5c1610b3e6121412067491aba1b59b9670641d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=97=D0=B0=D0=B8=D0=B4=20=D0=9E=D0=BC=D0=B0=D1=80=20?= =?UTF-8?q?=D0=9C=D0=B5=D0=B4=D1=85=D0=B0=D1=82=20=7C=20Zaid=20Omar=20Medh?= =?UTF-8?q?at?= Date: Fri, 10 Jul 2026 18:11:25 +0500 Subject: [PATCH] Phase 6: media messages + avatars via MinIO presigned uploads - Media (voice/video/image/file): private bucket, presigned PUT upload, reference-only publish, membership-gated presigned GET for viewing - Avatars: public bucket, presigned PUT + direct public URL, profile.update event - messages.media_key + media_meta (migration 1720000000005_media) - Explicit MinIO region on presigning client (avoids getBucketRegion network call) - Event type values sourced from EventType constants (no raw string literals) - Web: attach button, MediaView renderer, avatar upload in topbar Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01BFRyKxkKEjfAgpXygzoNiD --- .env.example | 6 + docker-compose.yml | 5 +- infra/centrifugo/config.json | 7 + .../migrations/1720000000005_media.cjs | 16 +++ packages/backend/src/app.ts | 12 +- packages/backend/src/config.ts | 6 + packages/backend/src/db/schema.ts | 4 + .../conversations/conversations.routes.ts | 30 +---- .../conversations/conversations.service.ts | 20 +-- packages/backend/src/modules/media/index.ts | 3 + .../backend/src/modules/media/media.routes.ts | 60 +++++++++ .../src/modules/media/media.service.ts | 77 +++++++++++ .../src/modules/messages/messages.mapper.ts | 1 + .../modules/messages/messages.repository.ts | 10 +- .../src/modules/messages/messages.routes.ts | 21 +++ .../src/modules/messages/messages.service.ts | 30 ++++- .../src/modules/realtime/realtime.service.ts | 11 +- .../src/modules/users/users.repository.ts | 9 ++ .../backend/src/modules/users/users.routes.ts | 25 +++- .../src/modules/users/users.service.ts | 28 +++- packages/backend/src/plugins/minio.ts | 28 ++-- packages/core/src/api/conversations.ts | 8 -- packages/core/src/api/index.ts | 6 +- packages/core/src/api/media.ts | 55 ++++++++ packages/core/src/api/users.ts | 9 +- packages/core/src/channels/index.ts | 10 ++ packages/core/src/events/index.ts | 49 ++++--- packages/core/src/realtime/client.ts | 39 +++++- packages/core/src/realtime/index.ts | 7 +- packages/core/src/schemas/conversation.ts | 9 +- packages/core/src/schemas/entities.ts | 4 + packages/core/src/schemas/index.ts | 17 ++- packages/core/src/schemas/live.ts | 10 -- packages/core/src/schemas/media.ts | 83 ++++++++++++ packages/core/src/types/index.ts | 2 +- packages/core/src/types/message.ts | 4 + packages/web/src/app/App.tsx | 73 +++++----- packages/web/src/app/index.css | 52 ++++++++ packages/web/src/entities/session/model.tsx | 9 +- .../web/src/features/conversations/model.ts | 126 +++++++++++++++--- packages/web/src/features/messaging/model.ts | 77 ++++++++--- .../src/features/messaging/ui/ChatView.tsx | 100 +++++++++++++- packages/web/src/features/realtime/model.tsx | 102 ++++++++++---- 43 files changed, 1031 insertions(+), 229 deletions(-) create mode 100644 packages/backend/migrations/1720000000005_media.cjs create mode 100644 packages/backend/src/modules/media/index.ts create mode 100644 packages/backend/src/modules/media/media.routes.ts create mode 100644 packages/backend/src/modules/media/media.service.ts create mode 100644 packages/core/src/api/media.ts create mode 100644 packages/core/src/schemas/media.ts diff --git a/.env.example b/.env.example index 30cba81..c46d436 100644 --- a/.env.example +++ b/.env.example @@ -54,6 +54,12 @@ MINIO_PORT=9000 MINIO_USE_SSL=false MINIO_BUCKET_MEDIA=media MINIO_BUCKET_AVATARS=avatars +# Browser-facing MinIO URL — presigned upload/download URLs are signed for this +# host, so it must match what the browser uses (dev: the exposed host port). +MINIO_PUBLIC_URL=http://localhost:9000 +# S3 region used for SigV4 presigning (MinIO default is us-east-1). Set explicitly +# so presigning never makes a network region-lookup call. +MINIO_REGION=us-east-1 # --- nginx (public entrypoint) --- NGINX_HTTP_PORT=8080 diff --git a/docker-compose.yml b/docker-compose.yml index 2068407..a34d6e5 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -38,6 +38,8 @@ services: environment: MINIO_ROOT_USER: ${MINIO_ROOT_USER} MINIO_ROOT_PASSWORD: ${MINIO_ROOT_PASSWORD} + # Allow browser (cross-origin) presigned PUT/GET uploads. + MINIO_API_CORS_ALLOW_ORIGIN: '*' volumes: - minio-data:/data healthcheck: @@ -61,7 +63,8 @@ services: mc alias set local http://minio:9000 "$${MINIO_ROOT_USER}" "$${MINIO_ROOT_PASSWORD}" && mc mb --ignore-existing "local/$${MINIO_BUCKET_MEDIA}" && mc mb --ignore-existing "local/$${MINIO_BUCKET_AVATARS}" && - echo "minio buckets ready" + mc anonymous set download "local/$${MINIO_BUCKET_AVATARS}" && + echo "minio buckets ready (avatars public)" restart: 'no' networks: [altricade] diff --git a/infra/centrifugo/config.json b/infra/centrifugo/config.json index 1940188..989bd73 100644 --- a/infra/centrifugo/config.json +++ b/infra/centrifugo/config.json @@ -19,6 +19,7 @@ "presence": true, "join_leave": true, "force_push_join_leave": true, + "allow_presence_for_subscriber": true, "history_size": 100, "history_ttl": "300s", "force_recovery": true, @@ -34,6 +35,12 @@ "force_recovery": true, "subscribe_proxy_enabled": true, "subscribe_proxy_name": "backend" + }, + { + "name": "eph", + "allow_publish_for_subscriber": true, + "subscribe_proxy_enabled": true, + "subscribe_proxy_name": "backend" } ] }, diff --git a/packages/backend/migrations/1720000000005_media.cjs b/packages/backend/migrations/1720000000005_media.cjs new file mode 100644 index 0000000..d640f5d --- /dev/null +++ b/packages/backend/migrations/1720000000005_media.cjs @@ -0,0 +1,16 @@ +// Phase 6 — media messages. The media reference is kept in dedicated columns +// (server-visible object key + metadata) separate from `content` (the caption), +// so E2EE can later encrypt the caption without hiding the storage pointer. + +/** @param {import('node-pg-migrate').MigrationBuilder} pgm */ +exports.up = (pgm) => { + pgm.addColumns('messages', { + media_key: { type: 'text' }, + media_meta: { type: 'jsonb' }, + }); +}; + +/** @param {import('node-pg-migrate').MigrationBuilder} pgm */ +exports.down = (pgm) => { + pgm.dropColumns('messages', ['media_key', 'media_meta']); +}; diff --git a/packages/backend/src/app.ts b/packages/backend/src/app.ts index 63b923b..9468c2e 100644 --- a/packages/backend/src/app.ts +++ b/packages/backend/src/app.ts @@ -25,6 +25,7 @@ import { import { createMessagesRepository, createMessagesService, messagesRoutes } from './modules/messages'; import { createContactsRepository, createContactsService, contactsRoutes } from './modules/contacts'; import { createPresenceService, presenceRoutes } from './modules/presence'; +import { createMediaService, mediaRoutes } from './modules/media'; import { realtimeRoutes } from './modules/realtime'; import type { Publisher } from './shared/publisher'; @@ -89,8 +90,15 @@ export const buildApp = async (config: AppConfig = loadConfig()): Promise mediaService.downloadUrl(objectKey), }), ); app.decorate( @@ -146,6 +155,7 @@ export const buildApp = async (config: AppConfig = loadConfig()): Promise ({ useSSL: parseBoolean(optional('MINIO_USE_SSL', 'false')), accessKey: required('MINIO_ROOT_USER'), secretKey: required('MINIO_ROOT_PASSWORD'), + region: optional('MINIO_REGION', 'us-east-1'), + publicUrl: optional('MINIO_PUBLIC_URL', 'http://localhost:9000'), buckets: { media: optional('MINIO_BUCKET_MEDIA', 'media'), avatars: optional('MINIO_BUCKET_AVATARS', 'avatars'), diff --git a/packages/backend/src/db/schema.ts b/packages/backend/src/db/schema.ts index 04513fd..e4d330a 100644 --- a/packages/backend/src/db/schema.ts +++ b/packages/backend/src/db/schema.ts @@ -1,4 +1,5 @@ import type { ColumnType, Generated } from 'kysely'; +import type { MediaRef } from '@altricade/core'; // Kysely database registry: one interface per table. Grows with each migration. @@ -58,6 +59,9 @@ export interface MessagesTable { created_at: Generated; edited_at: Date | null; deleted_at: Date | null; + media_key: string | null; + // jsonb: parsed to a MediaRef on read, JSON string on write. + media_meta: ColumnType; } export interface ContactsTable { diff --git a/packages/backend/src/modules/conversations/conversations.routes.ts b/packages/backend/src/modules/conversations/conversations.routes.ts index 39255a3..a8a93b0 100644 --- a/packages/backend/src/modules/conversations/conversations.routes.ts +++ b/packages/backend/src/modules/conversations/conversations.routes.ts @@ -4,19 +4,12 @@ import { createGroupBodySchema, addMemberBodySchema, readBodySchema, - typingBodySchema, conversationSchema, conversationListSchema, conversationMemberListSchema, errorSchema, } from '@altricade/core'; -import type { - CreateDirectBody, - CreateGroupBody, - AddMemberBody, - ReadBody, - TypingBody, -} from '@altricade/core'; +import type { CreateDirectBody, CreateGroupBody, AddMemberBody, ReadBody } from '@altricade/core'; const bearerAuth = [{ bearerAuth: [] }]; const idParamsSchema = { @@ -197,26 +190,5 @@ export const conversationsRoutes = (app: FastifyInstance): Promise => { }, ); - app.post<{ Params: { id: string }; Body: TypingBody }>( - '/conversations/:id/typing', - { - schema: { - tags: ['live'], - summary: 'Send a typing indicator (ephemeral)', - security: bearerAuth, - params: idParamsSchema, - body: typingBodySchema, - response: { 401: errorSchema, 403: errorSchema }, - }, - preHandler: app.authenticate, - }, - async (request, reply) => { - const user = request.authUser; - if (user === undefined) return reply.code(401).send({ error: 'unauthorized' }); - await app.conversationsService.setTyping(request.params.id, user.id, request.body.state); - return reply.code(204).send(); - }, - ); - return Promise.resolve(); }; diff --git a/packages/backend/src/modules/conversations/conversations.service.ts b/packages/backend/src/modules/conversations/conversations.service.ts index a0bd188..c3a6c76 100644 --- a/packages/backend/src/modules/conversations/conversations.service.ts +++ b/packages/backend/src/modules/conversations/conversations.service.ts @@ -4,9 +4,8 @@ import type { ConversationNewEvent, ConversationMembershipEvent, ReadReceiptEvent, - TypingEvent, } from '@altricade/core'; -import { userChannel } from '@altricade/core'; +import { userChannel, EventType } from '@altricade/core'; import { HttpError } from '../../shared/http-error'; import type { Publisher } from '../../shared/publisher'; import { toPublicUser } from '../users'; @@ -35,7 +34,6 @@ export interface ConversationsService { isMember(conversationId: string, userId: string): Promise; getDeliveryInfo(conversationId: string): Promise; markRead(conversationId: string, userId: string, seq: number): Promise; - setTyping(conversationId: string, userId: string, state: 'start' | 'stop'): Promise; } export const createConversationsService = ( @@ -56,7 +54,7 @@ export const createConversationsService = ( }; const conversationNew = (conversation: Conversation): ConversationNewEvent => ({ - type: 'conversation.new', + type: EventType.ConversationNew, conversation, }); @@ -64,7 +62,7 @@ export const createConversationsService = ( action: ConversationMembershipEvent['action'], conversationId: string, userId: string, - ): ConversationMembershipEvent => ({ type: 'conversation.membership', action, conversationId, userId }); + ): ConversationMembershipEvent => ({ type: EventType.ConversationMembership, action, conversationId, userId }); return { createDirect: async (actorId, username) => { @@ -179,17 +177,7 @@ export const createConversationsService = ( markRead: async (conversationId, userId, seq) => { await assertMember(conversationId, userId); await readState.setRead(conversationId, userId, seq); - const event: ReadReceiptEvent = { type: 'read.receipt', conversationId, userId, seq }; - await deliver(conversationId, event); - }, - - setTyping: async (conversationId, userId, state) => { - await assertMember(conversationId, userId); - const event: TypingEvent = { - type: state === 'start' ? 'typing.start' : 'typing.stop', - conversationId, - userId, - }; + const event: ReadReceiptEvent = { type: EventType.ReadReceipt, conversationId, userId, seq }; await deliver(conversationId, event); }, }; diff --git a/packages/backend/src/modules/media/index.ts b/packages/backend/src/modules/media/index.ts new file mode 100644 index 0000000..a7f30a2 --- /dev/null +++ b/packages/backend/src/modules/media/index.ts @@ -0,0 +1,3 @@ +export { createMediaService } from './media.service'; +export type { MediaService, MediaServiceDeps, UploadTarget, AvatarTarget } from './media.service'; +export { mediaRoutes } from './media.routes'; diff --git a/packages/backend/src/modules/media/media.routes.ts b/packages/backend/src/modules/media/media.routes.ts new file mode 100644 index 0000000..29dc6cc --- /dev/null +++ b/packages/backend/src/modules/media/media.routes.ts @@ -0,0 +1,60 @@ +import type { FastifyInstance } from 'fastify'; +import { + uploadUrlBodySchema, + avatarUploadBodySchema, + uploadTargetSchema, + avatarTargetSchema, + errorSchema, +} from '@altricade/core'; +import type { UploadUrlBody, AvatarUploadBody } from '@altricade/core'; + +const bearerAuth = [{ bearerAuth: [] }]; + +export const mediaRoutes = (app: FastifyInstance): Promise => { + app.post<{ Body: UploadUrlBody }>( + '/media/upload-url', + { + schema: { + tags: ['media'], + summary: 'Get a presigned upload URL for a message attachment', + security: bearerAuth, + body: uploadUrlBodySchema, + response: { 200: uploadTargetSchema, 400: errorSchema, 401: errorSchema }, + }, + preHandler: app.authenticate, + }, + async (request, reply) => { + const user = request.authUser; + if (user === undefined) return reply.code(401).send({ error: 'unauthorized' }); + const target = await app.mediaService.createUploadUrl( + user.id, + request.body.kind, + request.body.mime, + request.body.size, + ); + return reply.send(target); + }, + ); + + app.post<{ Body: AvatarUploadBody }>( + '/media/avatar-url', + { + schema: { + tags: ['media'], + summary: 'Get a presigned upload URL for an avatar', + security: bearerAuth, + body: avatarUploadBodySchema, + response: { 200: avatarTargetSchema, 400: errorSchema, 401: errorSchema }, + }, + preHandler: app.authenticate, + }, + async (request, reply) => { + const user = request.authUser; + if (user === undefined) return reply.code(401).send({ error: 'unauthorized' }); + const target = await app.mediaService.createAvatarUploadUrl(user.id, request.body.mime); + return reply.send(target); + }, + ); + + return Promise.resolve(); +}; diff --git a/packages/backend/src/modules/media/media.service.ts b/packages/backend/src/modules/media/media.service.ts new file mode 100644 index 0000000..d5b3267 --- /dev/null +++ b/packages/backend/src/modules/media/media.service.ts @@ -0,0 +1,77 @@ +import { randomUUID } from 'node:crypto'; +import type { Client } from 'minio'; +import type { MediaKind } from '@altricade/core'; +import { HttpError } from '../../shared/http-error'; + +export interface MediaServiceDeps { + // The presigning (browser-facing) MinIO client. + minio: Client; + mediaBucket: string; + avatarsBucket: string; + publicUrl: string; +} + +export interface UploadTarget { + uploadUrl: string; + objectKey: string; +} + +export interface AvatarTarget extends UploadTarget { + publicUrl: string; +} + +export interface MediaService { + createUploadUrl(userId: string, kind: MediaKind, mime: string, size: number): Promise; + createAvatarUploadUrl(userId: string, mime: string): Promise; + downloadUrl(objectKey: string): Promise; + avatarPublicUrl(objectKey: string): string; +} + +const UPLOAD_EXPIRY = 3600; +const DOWNLOAD_EXPIRY = 3600; + +const kindMatches = (kind: MediaKind, mime: string): boolean => { + if (kind === 'image') return mime.startsWith('image/'); + if (kind === 'video') return mime.startsWith('video/'); + if (kind === 'voice') return mime.startsWith('audio/'); + return true; +}; + +export const createMediaService = (deps: MediaServiceDeps): MediaService => ({ + createUploadUrl: async (userId, kind, mime, _size) => { + if (!kindMatches(kind, mime)) { + throw new HttpError(400, 'invalid_media', `Content type ${mime} does not match kind ${kind}`); + } + const objectKey = `${userId}/${randomUUID()}`; + const uploadUrl = await deps.minio.presignedPutObject(deps.mediaBucket, objectKey, UPLOAD_EXPIRY); + return { uploadUrl, objectKey }; + }, + + createAvatarUploadUrl: async (userId, mime) => { + if (!mime.startsWith('image/')) { + throw new HttpError(400, 'invalid_media', 'Avatar must be an image'); + } + const objectKey = `${userId}/${randomUUID()}`; + const uploadUrl = await deps.minio.presignedPutObject( + deps.avatarsBucket, + objectKey, + UPLOAD_EXPIRY, + ); + return { + uploadUrl, + objectKey, + publicUrl: `${deps.publicUrl}/${deps.avatarsBucket}/${objectKey}`, + }; + }, + + downloadUrl: (objectKey) => + deps.minio.presignedGetObject(deps.mediaBucket, objectKey, DOWNLOAD_EXPIRY), + + avatarPublicUrl: (objectKey) => `${deps.publicUrl}/${deps.avatarsBucket}/${objectKey}`, +}); + +declare module 'fastify' { + interface FastifyInstance { + mediaService: MediaService; + } +} diff --git a/packages/backend/src/modules/messages/messages.mapper.ts b/packages/backend/src/modules/messages/messages.mapper.ts index e57473a..c5e5c47 100644 --- a/packages/backend/src/modules/messages/messages.mapper.ts +++ b/packages/backend/src/modules/messages/messages.mapper.ts @@ -23,4 +23,5 @@ export const toMessage = ( editedAt: row.edited_at === null ? null : row.edited_at.toISOString(), deletedAt: row.deleted_at === null ? null : row.deleted_at.toISOString(), reactions, + media: row.media_meta, }); diff --git a/packages/backend/src/modules/messages/messages.repository.ts b/packages/backend/src/modules/messages/messages.repository.ts index fe66981..9a9b111 100644 --- a/packages/backend/src/modules/messages/messages.repository.ts +++ b/packages/backend/src/modules/messages/messages.repository.ts @@ -1,6 +1,6 @@ import { sql } from 'kysely'; import type { Kysely } from 'kysely'; -import type { ReactionSummary } from '@altricade/core'; +import type { ReactionSummary, MediaRef } from '@altricade/core'; import type { Database } from '../../db/schema'; export interface MessageWithSenderRow { @@ -15,6 +15,8 @@ export interface MessageWithSenderRow { created_at: Date; edited_at: Date | null; deleted_at: Date | null; + media_key: string | null; + media_meta: MediaRef | null; sender_username: string; sender_display_name: string; sender_avatar_ref: string | null; @@ -27,6 +29,8 @@ export interface NewMessage { content: string; contentType: string; encryption: string | null; + mediaKey: string | null; + media: MediaRef | null; } export interface MessagesRepository { @@ -71,6 +75,8 @@ export const createMessagesRepository = (db: Kysely): MessagesReposito 'messages.created_at as created_at', 'messages.edited_at as edited_at', 'messages.deleted_at as deleted_at', + 'messages.media_key as media_key', + 'messages.media_meta as media_meta', 'users.username as sender_username', 'users.display_name as sender_display_name', 'users.avatar_ref as sender_avatar_ref', @@ -87,6 +93,8 @@ export const createMessagesRepository = (db: Kysely): MessagesReposito content: input.content, content_type: input.contentType, encryption: input.encryption, + media_key: input.mediaKey, + media_meta: input.media === null ? null : JSON.stringify(input.media), }) .onConflict((oc) => oc.columns(['conversation_id', 'sender_id', 'client_msg_id']).doNothing(), diff --git a/packages/backend/src/modules/messages/messages.routes.ts b/packages/backend/src/modules/messages/messages.routes.ts index fa4d33d..eb4226b 100644 --- a/packages/backend/src/modules/messages/messages.routes.ts +++ b/packages/backend/src/modules/messages/messages.routes.ts @@ -5,6 +5,7 @@ import { reactionBodySchema, messageSchema, messageListSchema, + mediaUrlSchema, errorSchema, } from '@altricade/core'; import type { SendMessageBody, EditMessageBody, ReactionBody } from '@altricade/core'; @@ -161,6 +162,26 @@ export const messagesRoutes = (app: FastifyInstance): Promise => { }, ); + app.get<{ Params: { id: string; messageId: string } }>( + '/conversations/:id/messages/:messageId/media-url', + { + schema: { + tags: ['media'], + summary: 'Get a presigned download URL for a message attachment (member only)', + security: bearerAuth, + params: messageParamsSchema, + response: { 200: mediaUrlSchema, 401: errorSchema, 403: errorSchema, 404: errorSchema }, + }, + preHandler: app.authenticate, + }, + async (request, reply) => { + const user = request.authUser; + if (user === undefined) return reply.code(401).send({ error: 'unauthorized' }); + const url = await app.messagesService.mediaUrl(request.params.id, request.params.messageId, user.id); + return reply.send({ url }); + }, + ); + app.delete<{ Params: { id: string; messageId: string; emoji: string } }>( '/conversations/:id/messages/:messageId/reactions/:emoji', { diff --git a/packages/backend/src/modules/messages/messages.service.ts b/packages/backend/src/modules/messages/messages.service.ts index 86f2d54..aabce28 100644 --- a/packages/backend/src/modules/messages/messages.service.ts +++ b/packages/backend/src/modules/messages/messages.service.ts @@ -1,3 +1,4 @@ +import { EventType } from '@altricade/core'; import type { Message, MessageNewEvent, @@ -16,6 +17,7 @@ export interface MessagesServiceDeps { messages: MessagesRepository; conversations: ConversationsRepository; deliver: Deliver; + mediaDownloadUrl: (objectKey: string) => Promise; } export interface SentMessage { @@ -50,10 +52,11 @@ export interface MessagesService { userId: string, emoji: string, ): Promise; + mediaUrl(conversationId: string, messageId: string, userId: string): Promise; } export const createMessagesService = (deps: MessagesServiceDeps): MessagesService => { - const { messages, conversations, deliver } = deps; + const { messages, conversations, deliver, mediaDownloadUrl } = deps; const assertMember = async (conversationId: string, userId: string): Promise => { if (!(await conversations.isMember(conversationId, userId))) { @@ -84,9 +87,11 @@ export const createMessagesService = (deps: MessagesServiceDeps): MessagesServic conversationId, senderId, clientMsgId: input.clientMsgId, - content: input.content, + content: input.content ?? '', contentType: input.contentType ?? 'text', encryption: input.encryption ?? null, + mediaKey: input.mediaKey ?? null, + media: input.media ?? null, }); const id = @@ -103,7 +108,7 @@ export const createMessagesService = (deps: MessagesServiceDeps): MessagesServic if (inserted !== undefined) { await conversations.touchLastMessage(conversationId); - const event: MessageNewEvent = { type: 'message.new', message }; + const event: MessageNewEvent = { type: EventType.MessageNew, message }; await deliver(conversationId, event); } return { message, created: inserted !== undefined }; @@ -127,7 +132,7 @@ export const createMessagesService = (deps: MessagesServiceDeps): MessagesServic throw new HttpError(403, 'not_editable', 'You can only edit your own messages'); } const message = await loadMessage(messageId, userId); - const event: MessageEditEvent = { type: 'message.edit', message }; + const event: MessageEditEvent = { type: EventType.MessageEdit, message }; await deliver(conversationId, event); return message; }, @@ -139,7 +144,7 @@ export const createMessagesService = (deps: MessagesServiceDeps): MessagesServic if (deleted === undefined) { throw new HttpError(403, 'not_deletable', 'You can only delete your own messages'); } - const event: MessageDeleteEvent = { type: 'message.delete', conversationId, messageId }; + const event: MessageDeleteEvent = { type: EventType.MessageDelete, conversationId, messageId }; await deliver(conversationId, event); }, @@ -148,7 +153,7 @@ export const createMessagesService = (deps: MessagesServiceDeps): MessagesServic await assertMessageIn(conversationId, messageId); await messages.addReaction(messageId, userId, emoji); const event: ReactionEvent = { - type: 'reaction.add', + type: EventType.ReactionAdd, conversationId, messageId, emoji, @@ -162,7 +167,7 @@ export const createMessagesService = (deps: MessagesServiceDeps): MessagesServic await assertMessageIn(conversationId, messageId); await messages.removeReaction(messageId, userId, emoji); const event: ReactionEvent = { - type: 'reaction.remove', + type: EventType.ReactionRemove, conversationId, messageId, emoji, @@ -170,6 +175,17 @@ export const createMessagesService = (deps: MessagesServiceDeps): MessagesServic }; await deliver(conversationId, event); }, + + mediaUrl: async (conversationId, messageId, userId) => { + await assertMember(conversationId, userId); + await assertMessageIn(conversationId, messageId); + const row = await messages.getWithSenderById(messageId); + const mediaKey = row?.media_key ?? null; + if (mediaKey === null) { + throw new HttpError(404, 'not_found', 'No media on this message'); + } + return mediaDownloadUrl(mediaKey); + }, }; }; diff --git a/packages/backend/src/modules/realtime/realtime.service.ts b/packages/backend/src/modules/realtime/realtime.service.ts index f605f18..4090cbb 100644 --- a/packages/backend/src/modules/realtime/realtime.service.ts +++ b/packages/backend/src/modules/realtime/realtime.service.ts @@ -1,4 +1,9 @@ -import { isUserChannel, conversationChannelId, userChannel } from '@altricade/core'; +import { + isUserChannel, + conversationChannelId, + ephemeralChannelId, + userChannel, +} from '@altricade/core'; export type SubscribeDecision = | { allowed: true } @@ -26,7 +31,9 @@ export const authorizeSubscription = async ( return channel === userChannel(userId) ? { allowed: true } : deny('permission denied'); } - const conversationId = conversationChannelId(channel); + // Group message channel (conv:) and the ephemeral typing channel (eph:) are + // both authorized by conversation membership. + const conversationId = conversationChannelId(channel) ?? ephemeralChannelId(channel); if (conversationId !== null) { if (userId === '') { return deny('permission denied'); diff --git a/packages/backend/src/modules/users/users.repository.ts b/packages/backend/src/modules/users/users.repository.ts index 65cef91..8a5d53d 100644 --- a/packages/backend/src/modules/users/users.repository.ts +++ b/packages/backend/src/modules/users/users.repository.ts @@ -24,6 +24,7 @@ export interface UsersRepository { findManyByIds(ids: string[]): Promise; searchByPrefix(query: string, excludeUserId: string, limit: number): Promise; updateProfile(id: string, patch: ProfilePatch): Promise; + setAvatar(id: string, avatarRef: string): Promise; touchLastSeen(id: string): Promise; } @@ -82,6 +83,14 @@ export const createUsersRepository = (db: Kysely): UsersRepository => .executeTakeFirst(); }, + setAvatar: (id, avatarRef) => + db + .updateTable('users') + .set({ avatar_ref: avatarRef, updated_at: new Date() }) + .where('id', '=', id) + .returningAll() + .executeTakeFirst(), + touchLastSeen: async (id) => { await db.updateTable('users').set({ last_seen_at: new Date() }).where('id', '=', id).execute(); }, diff --git a/packages/backend/src/modules/users/users.routes.ts b/packages/backend/src/modules/users/users.routes.ts index 816793c..f2d0a7d 100644 --- a/packages/backend/src/modules/users/users.routes.ts +++ b/packages/backend/src/modules/users/users.routes.ts @@ -1,12 +1,13 @@ import type { FastifyInstance } from 'fastify'; import { updateMeBodySchema, + setAvatarBodySchema, userSchema, publicUserSchema, publicUserListSchema, errorSchema, } from '@altricade/core'; -import type { UpdateMeBody } from '@altricade/core'; +import type { UpdateMeBody, SetAvatarBody } from '@altricade/core'; const searchQuerySchema = { type: 'object', @@ -87,6 +88,28 @@ export const usersRoutes = (app: FastifyInstance): Promise => { }, ); + app.post<{ Body: SetAvatarBody }>( + '/me/avatar', + { + schema: { + tags: ['users'], + summary: 'Set the current user avatar (from an uploaded object key)', + security: bearerAuth, + body: setAvatarBodySchema, + response: { 200: userSchema, 401: errorSchema }, + }, + preHandler: app.authenticate, + }, + async (request, reply) => { + const user = request.authUser; + if (user === undefined) { + return reply.code(401).send({ error: 'unauthorized' }); + } + const avatarUrl = app.mediaService.avatarPublicUrl(request.body.objectKey); + return reply.send(await app.usersService.setAvatar(user.id, avatarUrl)); + }, + ); + app.post( '/me/heartbeat', { diff --git a/packages/backend/src/modules/users/users.service.ts b/packages/backend/src/modules/users/users.service.ts index c013435..5da2684 100644 --- a/packages/backend/src/modules/users/users.service.ts +++ b/packages/backend/src/modules/users/users.service.ts @@ -1,12 +1,20 @@ -import type { User, PublicUser, UpdateMeBody } from '@altricade/core'; +import type { User, PublicUser, UpdateMeBody, ProfileUpdateEvent } from '@altricade/core'; +import { userChannel, EventType } from '@altricade/core'; import { HttpError } from '../../shared/http-error'; +import type { Publisher } from '../../shared/publisher'; import type { UsersRepository } from './users.repository'; import { toUser, toPublicUser } from './users.mapper'; +export interface UsersServiceDeps { + users: UsersRepository; + publish: Publisher; +} + export interface UsersService { getMe(userId: string): Promise; getPublicProfile(username: string): Promise; updateProfile(userId: string, patch: UpdateMeBody): Promise; + setAvatar(userId: string, avatarUrl: string): Promise; search(query: string, excludeUserId: string): Promise; heartbeat(userId: string): Promise; } @@ -16,7 +24,7 @@ const SEARCH_LIMIT = 20; // Escape LIKE wildcards so user input is matched literally. const escapeLike = (value: string): string => value.replace(/[\\%_]/g, '\\$&'); -export const createUsersService = (users: UsersRepository): UsersService => ({ +export const createUsersService = ({ users, publish }: UsersServiceDeps): UsersService => ({ getMe: async (userId) => { const row = await users.findById(userId); if (row === undefined) { @@ -41,6 +49,22 @@ export const createUsersService = (users: UsersRepository): UsersService => ({ return toUser(row); }, + setAvatar: async (userId, avatarUrl) => { + const row = await users.setAvatar(userId, avatarUrl); + if (row === undefined) { + throw new HttpError(404, 'not_found', 'User not found'); + } + const user = toUser(row); + const event: ProfileUpdateEvent = { + type: EventType.ProfileUpdate, + userId, + displayName: user.displayName, + avatarUrl: user.avatarUrl, + }; + await publish(userChannel(userId), event); + return user; + }, + search: async (query, excludeUserId) => { const trimmed = query.trim(); if (trimmed.length === 0) { diff --git a/packages/backend/src/plugins/minio.ts b/packages/backend/src/plugins/minio.ts index 62f998f..19351b8 100644 --- a/packages/backend/src/plugins/minio.ts +++ b/packages/backend/src/plugins/minio.ts @@ -4,24 +4,34 @@ import { Client } from 'minio'; declare module 'fastify' { interface FastifyInstance { minio: Client; + // Client configured with the browser-facing host — used ONLY for presigning + // upload/download URLs so the signature matches the host the browser hits. + minioPublic: Client; } } // S3-compatible object storage client (media messages + avatars). Buckets are -// bootstrapped by the compose `minio-setup` one-shot; this client is used for -// readiness checks now and presigned uploads from Phase 6. +// bootstrapped by the compose `minio-setup` one-shot. export const minioPlugin = fp( (app) => { - const { endpoint, port, useSSL, accessKey, secretKey } = app.config.minio; - const client = new Client({ - endPoint: endpoint, - port, - useSSL, + const { endpoint, port, useSSL, accessKey, secretKey, region, publicUrl } = app.config.minio; + const client = new Client({ endPoint: endpoint, port, useSSL, accessKey, secretKey, region }); + app.decorate('minio', client); + + const parsed = new URL(publicUrl); + const publicSecure = parsed.protocol === 'https:'; + const publicPort = parsed.port === '' ? (publicSecure ? 443 : 80) : Number.parseInt(parsed.port, 10); + // Explicit region so presigning is purely computational — no getBucketRegion + // network call to the browser-facing host (unreachable from inside the container). + const publicClient = new Client({ + endPoint: parsed.hostname, + port: publicPort, + useSSL: publicSecure, accessKey, secretKey, + region, }); - - app.decorate('minio', client); + app.decorate('minioPublic', publicClient); return Promise.resolve(); }, diff --git a/packages/core/src/api/conversations.ts b/packages/core/src/api/conversations.ts index 20b9c40..0821151 100644 --- a/packages/core/src/api/conversations.ts +++ b/packages/core/src/api/conversations.ts @@ -61,11 +61,3 @@ export const markRead = async ( ): Promise => { await requestJson(config, 'POST', `/conversations/${id}/read`, { seq }); }; - -export const sendTyping = async ( - config: ApiClientConfig, - id: string, - state: 'start' | 'stop', -): Promise => { - await requestJson(config, 'POST', `/conversations/${id}/typing`, { state }); -}; diff --git a/packages/core/src/api/index.ts b/packages/core/src/api/index.ts index 5b21c80..9861cb3 100644 --- a/packages/core/src/api/index.ts +++ b/packages/core/src/api/index.ts @@ -13,7 +13,7 @@ export { getCentrifugoToken, } from './auth'; export { sendEcho } from './realtime'; -export { searchUsers } from './users'; +export { searchUsers, setAvatar } from './users'; export { listConversations, getConversation, @@ -33,5 +33,7 @@ export { removeReaction, } from './messages'; export type { HistoryOptions } from './messages'; -export { markRead, sendTyping } from './conversations'; +export { markRead } from './conversations'; export { getPresence, heartbeat } from './presence'; +export { getUploadUrl, getAvatarUploadUrl, getMediaUrl, uploadToUrl } from './media'; +export type { UploadTarget, AvatarTarget } from './media'; diff --git a/packages/core/src/api/media.ts b/packages/core/src/api/media.ts new file mode 100644 index 0000000..5897a39 --- /dev/null +++ b/packages/core/src/api/media.ts @@ -0,0 +1,55 @@ +import { uploadTargetSchema, avatarTargetSchema, mediaUrlSchema } from '../schemas/index'; +import type { UploadUrlBody, AvatarUploadBody } from '../schemas/index'; +import { compileValidator, parse, requestJson } from './http'; +import type { ApiClientConfig } from './http'; + +export interface UploadTarget { + uploadUrl: string; + objectKey: string; +} + +export interface AvatarTarget { + uploadUrl: string; + objectKey: string; + publicUrl: string; +} + +const uploadTargetV = compileValidator(uploadTargetSchema); +const avatarTargetV = compileValidator(avatarTargetSchema); +const mediaUrlV = compileValidator<{ url: string }>(mediaUrlSchema); + +export const getUploadUrl = async ( + config: ApiClientConfig, + body: UploadUrlBody, +): Promise => + parse(uploadTargetV, await requestJson(config, 'POST', '/media/upload-url', body)); + +export const getAvatarUploadUrl = async ( + config: ApiClientConfig, + body: AvatarUploadBody, +): Promise => + parse(avatarTargetV, await requestJson(config, 'POST', '/media/avatar-url', body)); + +export const getMediaUrl = async ( + config: ApiClientConfig, + conversationId: string, + messageId: string, +): Promise => { + const result = parse( + mediaUrlV, + await requestJson(config, 'GET', `/conversations/${conversationId}/messages/${messageId}/media-url`), + ); + return result.url; +}; + +// Direct PUT of the file bytes to the presigned MinIO URL (no auth/credentials). +export const uploadToUrl = async (uploadUrl: string, file: Blob, mime: string): Promise => { + const response = await fetch(uploadUrl, { + method: 'PUT', + headers: { 'content-type': mime }, + body: file, + }); + if (!response.ok) { + throw new Error(`Upload failed with status ${String(response.status)}`); + } +}; diff --git a/packages/core/src/api/users.ts b/packages/core/src/api/users.ts index ea5be01..849cf65 100644 --- a/packages/core/src/api/users.ts +++ b/packages/core/src/api/users.ts @@ -1,9 +1,11 @@ -import { publicUserListSchema } from '../schemas/index'; -import type { PublicUser } from '../types/index'; +import { publicUserListSchema, userSchema } from '../schemas/index'; +import type { SetAvatarBody } from '../schemas/index'; +import type { PublicUser, User } from '../types/index'; import { compileValidator, parse, requestJson } from './http'; import type { ApiClientConfig } from './http'; const publicUserListV = compileValidator(publicUserListSchema); +const userV = compileValidator(userSchema); export const searchUsers = async ( config: ApiClientConfig, @@ -13,3 +15,6 @@ export const searchUsers = async ( publicUserListV, await requestJson(config, 'GET', `/users/search?q=${encodeURIComponent(query)}`), ); + +export const setAvatar = async (config: ApiClientConfig, body: SetAvatarBody): Promise => + parse(userV, await requestJson(config, 'POST', '/me/avatar', body)); diff --git a/packages/core/src/channels/index.ts b/packages/core/src/channels/index.ts index 50c949d..6cefe43 100644 --- a/packages/core/src/channels/index.ts +++ b/packages/core/src/channels/index.ts @@ -10,12 +10,18 @@ export type UserId = string; const CONV_PREFIX = 'conv:'; const USER_PREFIX = 'user:'; +// Ephemeral channel (typing/recording): clients may publish here directly. Only +// throwaway events are handled from it, so a forged publish is harmless. +const EPH_PREFIX = 'eph:'; export const conversationChannel = (conversationId: ConversationId): string => `${CONV_PREFIX}${conversationId}`; export const userChannel = (userId: UserId): string => `${USER_PREFIX}${userId}`; +export const ephemeralChannel = (conversationId: ConversationId): string => + `${EPH_PREFIX}${conversationId}`; + /** True for a `conv:` channel name. */ export const isConversationChannel = (channel: string): boolean => channel.startsWith(CONV_PREFIX); @@ -29,3 +35,7 @@ export const conversationChannelId = (channel: string): string | null => /** Extract the user id from a `user:` channel, or null. */ export const userChannelId = (channel: string): string | null => channel.startsWith(USER_PREFIX) ? channel.slice(USER_PREFIX.length) : null; + +/** Extract the conversation id from an `eph:` channel, or null. */ +export const ephemeralChannelId = (channel: string): string | null => + channel.startsWith(EPH_PREFIX) ? channel.slice(EPH_PREFIX.length) : null; diff --git a/packages/core/src/events/index.ts b/packages/core/src/events/index.ts index 1e9b29e..bb7fc13 100644 --- a/packages/core/src/events/index.ts +++ b/packages/core/src/events/index.ts @@ -5,6 +5,9 @@ // Bucket A — durable, must be correct → through the backend (persist, publish) // Bucket B — ephemeral, throwaway → published, not persisted // Bucket C — connection-derived → Centrifugo built-in presence +// +// EventType is the single source of truth for every event `type` value — never +// write the raw string literal anywhere; reference EventType.* instead. import type { Message } from '../types/message'; import type { Conversation } from '../types/conversation'; @@ -35,44 +38,43 @@ export const EventType = { export type EventType = (typeof EventType)[keyof typeof EventType]; // --- Realtime event payloads (published by the backend, handled by clients) --- +// Each `type` field is derived from EventType so a typo is a compile error. export interface MessageNewEvent { - type: 'message.new'; + type: typeof EventType.MessageNew; message: Message; } +export interface MessageEditEvent { + type: typeof EventType.MessageEdit; + message: Message; +} + +export interface MessageDeleteEvent { + type: typeof EventType.MessageDelete; + conversationId: string; + messageId: string; +} + // Published to a user's personal channel when a new conversation involving them // is created (a DM someone started, or a group they were added to) so it appears // in their list immediately. export interface ConversationNewEvent { - type: 'conversation.new'; + type: typeof EventType.ConversationNew; conversation: Conversation; } export type ConversationMembershipAction = 'added' | 'removed'; export interface ConversationMembershipEvent { - type: 'conversation.membership'; + type: typeof EventType.ConversationMembership; action: ConversationMembershipAction; conversationId: string; userId: string; } -export interface MessageEditEvent { - type: 'message.edit'; - message: Message; -} - -export interface MessageDeleteEvent { - type: 'message.delete'; - conversationId: string; - messageId: string; -} - -export type ReactionAction = 'reaction.add' | 'reaction.remove'; - export interface ReactionEvent { - type: ReactionAction; + type: typeof EventType.ReactionAdd | typeof EventType.ReactionRemove; conversationId: string; messageId: string; emoji: string; @@ -80,16 +82,21 @@ export interface ReactionEvent { } export interface ReadReceiptEvent { - type: 'read.receipt'; + type: typeof EventType.ReadReceipt; conversationId: string; userId: string; seq: number; } -export type TypingState = 'typing.start' | 'typing.stop'; - export interface TypingEvent { - type: TypingState; + type: typeof EventType.TypingStart | typeof EventType.TypingStop; conversationId: string; userId: string; } + +export interface ProfileUpdateEvent { + type: typeof EventType.ProfileUpdate; + userId: string; + displayName: string; + avatarUrl: string | null; +} diff --git a/packages/core/src/realtime/client.ts b/packages/core/src/realtime/client.ts index 122eda1..880ca35 100644 --- a/packages/core/src/realtime/client.ts +++ b/packages/core/src/realtime/client.ts @@ -8,6 +8,8 @@ export interface RealtimeEvent { data: unknown; } +export type PresenceAction = 'join' | 'leave'; + export interface RealtimeClientOptions { /** Centrifugo WebSocket URL, e.g. '/connection/websocket' (same-origin). */ url: string; @@ -15,11 +17,12 @@ export interface RealtimeClientOptions { getToken: () => Promise; onState?: (state: ConnectionState) => void; onEvent?: (event: RealtimeEvent) => void; + /** Fired when a user joins/leaves a subscribed channel (channel presence). */ + onPresence?: (channel: string, action: PresenceAction, userId: string) => void; } -// UI-agnostic wrapper over the Centrifugo SDK. Receive-only: the app never -// publishes through it (sends go over REST). Uses the platform's global -// WebSocket (browser + React Native). +// UI-agnostic wrapper over the Centrifugo SDK. Receive-only for messages; also +// surfaces channel presence (join/leave + snapshot) for online indicators. export class RealtimeClient { private readonly centrifuge: Centrifuge; private readonly options: RealtimeClientOptions; @@ -58,6 +61,12 @@ export class RealtimeClient { const data: unknown = ctx.data; this.options.onEvent?.({ channel, data }); }); + subscription.on('join', (ctx) => { + this.options.onPresence?.(channel, 'join', ctx.info.user); + }); + subscription.on('leave', (ctx) => { + this.options.onPresence?.(channel, 'leave', ctx.info.user); + }); subscription.subscribe(); this.subscriptions.set(channel, subscription); } @@ -71,4 +80,28 @@ export class RealtimeClient { this.centrifuge.removeSubscription(subscription); this.subscriptions.delete(channel); } + + // Constrained client-side publish — used ONLY for ephemeral events (typing) on + // channels that permit subscriber publish. Durable events never use this. + async publish(channel: string, data: unknown): Promise { + const subscription = this.subscriptions.get(channel); + if (subscription === undefined) { + return; + } + await subscription.publish(data); + } + + // Current set of user ids present in a channel (presence snapshot). + async presence(channel: string): Promise { + const subscription = this.subscriptions.get(channel); + if (subscription === undefined) { + return []; + } + const result = await subscription.presence(); + const users = new Set(); + for (const client of Object.values(result.clients)) { + users.add(client.user); + } + return [...users]; + } } diff --git a/packages/core/src/realtime/index.ts b/packages/core/src/realtime/index.ts index 2dda14a..6e97489 100644 --- a/packages/core/src/realtime/index.ts +++ b/packages/core/src/realtime/index.ts @@ -1,2 +1,7 @@ export { RealtimeClient } from './client'; -export type { ConnectionState, RealtimeEvent, RealtimeClientOptions } from './client'; +export type { + ConnectionState, + RealtimeEvent, + RealtimeClientOptions, + PresenceAction, +} from './client'; diff --git a/packages/core/src/schemas/conversation.ts b/packages/core/src/schemas/conversation.ts index ffe7830..47c280d 100644 --- a/packages/core/src/schemas/conversation.ts +++ b/packages/core/src/schemas/conversation.ts @@ -1,4 +1,5 @@ import type { FromSchema } from 'json-schema-to-ts'; +import { mediaRefSchema } from './media'; const USERNAME_PATTERN = '^[a-zA-Z0-9_]{3,32}$'; @@ -37,12 +38,16 @@ export const addMemberBodySchema = { export const sendMessageBodySchema = { type: 'object', additionalProperties: false, - required: ['content', 'clientMsgId'], + required: ['clientMsgId'], properties: { - content: { type: 'string', minLength: 1, maxLength: 16000 }, + // Caption; may be empty for a media-only message. + content: { type: 'string', maxLength: 16000 }, clientMsgId: { type: 'string', minLength: 1, maxLength: 64 }, contentType: { type: 'string', enum: ['text', 'text/ciphertext'] }, encryption: { type: ['string', 'null'] }, + // Object key returned by the media upload URL + its metadata. + mediaKey: { type: 'string', minLength: 1, maxLength: 255 }, + media: mediaRefSchema, }, } as const; diff --git a/packages/core/src/schemas/entities.ts b/packages/core/src/schemas/entities.ts index 2688428..df79f06 100644 --- a/packages/core/src/schemas/entities.ts +++ b/packages/core/src/schemas/entities.ts @@ -2,6 +2,8 @@ // OpenAPI docs and response serialization, so the documented contract cannot // drift from what the API returns. +import { mediaRefSchema } from './media'; + export const publicUserSchema = { type: 'object', additionalProperties: false, @@ -192,6 +194,7 @@ export const messageSchema = { 'editedAt', 'deletedAt', 'reactions', + 'media', ], properties: { id: { type: 'string', format: 'uuid' }, @@ -207,6 +210,7 @@ export const messageSchema = { editedAt: { type: ['string', 'null'], format: 'date-time' }, deletedAt: { type: ['string', 'null'], format: 'date-time' }, reactions: { type: 'array', items: reactionSummarySchema }, + media: { oneOf: [mediaRefSchema, { type: 'null' }] }, }, } as const; diff --git a/packages/core/src/schemas/index.ts b/packages/core/src/schemas/index.ts index aec3952..eb58626 100644 --- a/packages/core/src/schemas/index.ts +++ b/packages/core/src/schemas/index.ts @@ -14,13 +14,18 @@ export type { SendMessageBody, AddContactBody, } from './conversation'; +export { editMessageBodySchema, reactionBodySchema, readBodySchema } from './live'; +export type { EditMessageBody, ReactionBody, ReadBody } from './live'; export { - editMessageBodySchema, - reactionBodySchema, - readBodySchema, - typingBodySchema, -} from './live'; -export type { EditMessageBody, ReactionBody, ReadBody, TypingBody } from './live'; + mediaRefSchema, + uploadUrlBodySchema, + avatarUploadBodySchema, + setAvatarBodySchema, + uploadTargetSchema, + avatarTargetSchema, + mediaUrlSchema, +} from './media'; +export type { MediaRef, MediaKind, UploadUrlBody, AvatarUploadBody, SetAvatarBody } from './media'; export { publicUserSchema, publicUserListSchema, diff --git a/packages/core/src/schemas/live.ts b/packages/core/src/schemas/live.ts index 208adec..8a50419 100644 --- a/packages/core/src/schemas/live.ts +++ b/packages/core/src/schemas/live.ts @@ -27,16 +27,6 @@ export const readBodySchema = { }, } as const; -export const typingBodySchema = { - type: 'object', - additionalProperties: false, - required: ['state'], - properties: { - state: { type: 'string', enum: ['start', 'stop'] }, - }, -} as const; - export type EditMessageBody = FromSchema; export type ReactionBody = FromSchema; export type ReadBody = FromSchema; -export type TypingBody = FromSchema; diff --git a/packages/core/src/schemas/media.ts b/packages/core/src/schemas/media.ts new file mode 100644 index 0000000..a0a9b1f --- /dev/null +++ b/packages/core/src/schemas/media.ts @@ -0,0 +1,83 @@ +import type { FromSchema } from 'json-schema-to-ts'; + +const MEDIA_KIND = { type: 'string', enum: ['image', 'video', 'voice', 'file'] } as const; + +// Metadata for an attached media object (no object key — that stays server-side). +export const mediaRefSchema = { + type: 'object', + additionalProperties: false, + required: ['kind', 'mime', 'size', 'name'], + properties: { + kind: MEDIA_KIND, + mime: { type: 'string', minLength: 1, maxLength: 255 }, + size: { type: 'integer', minimum: 0 }, + name: { type: 'string', minLength: 1, maxLength: 255 }, + width: { type: 'integer', minimum: 0 }, + height: { type: 'integer', minimum: 0 }, + durationSec: { type: 'number', minimum: 0 }, + }, +} as const; + +export const uploadUrlBodySchema = { + type: 'object', + additionalProperties: false, + required: ['kind', 'mime', 'size'], + properties: { + kind: MEDIA_KIND, + mime: { type: 'string', minLength: 1, maxLength: 255 }, + size: { type: 'integer', minimum: 1, maximum: 104857600 }, + }, +} as const; + +export const avatarUploadBodySchema = { + type: 'object', + additionalProperties: false, + required: ['mime', 'size'], + properties: { + mime: { type: 'string', minLength: 1, maxLength: 255 }, + size: { type: 'integer', minimum: 1, maximum: 10485760 }, + }, +} as const; + +export const setAvatarBodySchema = { + type: 'object', + additionalProperties: false, + required: ['objectKey'], + properties: { + objectKey: { type: 'string', minLength: 1, maxLength: 255 }, + }, +} as const; + +export const uploadTargetSchema = { + type: 'object', + additionalProperties: false, + required: ['uploadUrl', 'objectKey'], + properties: { + uploadUrl: { type: 'string' }, + objectKey: { type: 'string' }, + }, +} as const; + +export const avatarTargetSchema = { + type: 'object', + additionalProperties: false, + required: ['uploadUrl', 'objectKey', 'publicUrl'], + properties: { + uploadUrl: { type: 'string' }, + objectKey: { type: 'string' }, + publicUrl: { type: 'string' }, + }, +} as const; + +export const mediaUrlSchema = { + type: 'object', + additionalProperties: false, + required: ['url'], + properties: { url: { type: 'string' } }, +} as const; + +export type MediaRef = FromSchema; +export type MediaKind = MediaRef['kind']; +export type UploadUrlBody = FromSchema; +export type AvatarUploadBody = FromSchema; +export type SetAvatarBody = FromSchema; diff --git a/packages/core/src/types/index.ts b/packages/core/src/types/index.ts index 1750ff1..ef2bd53 100644 --- a/packages/core/src/types/index.ts +++ b/packages/core/src/types/index.ts @@ -5,5 +5,5 @@ export type { User, PublicUser } from './user'; export type { AuthResult, Session, CentrifugoToken } from './auth'; export type { Conversation, ConversationType, ConversationMember } from './conversation'; export type { Contact } from './contact'; -export type { Message, ReactionSummary } from './message'; +export type { Message, ReactionSummary, MediaRef } from './message'; export type { Presence } from './presence'; diff --git a/packages/core/src/types/message.ts b/packages/core/src/types/message.ts index 62968ba..ddd8c70 100644 --- a/packages/core/src/types/message.ts +++ b/packages/core/src/types/message.ts @@ -1,4 +1,7 @@ import type { PublicUser } from './user'; +import type { MediaRef } from '../schemas/media'; + +export type { MediaRef } from '../schemas/media'; export interface ReactionSummary { emoji: string; @@ -23,4 +26,5 @@ export interface Message { editedAt: string | null; deletedAt: string | null; reactions: ReactionSummary[]; + media: MediaRef | null; } diff --git a/packages/web/src/app/App.tsx b/packages/web/src/app/App.tsx index 89b9c2a..9b45a4e 100644 --- a/packages/web/src/app/App.tsx +++ b/packages/web/src/app/App.tsx @@ -1,7 +1,7 @@ import { useEffect, useState } from 'react'; -import type { ReactElement } from 'react'; +import type { ReactElement, SyntheticEvent } from 'react'; import type { Conversation, PublicUser, User } from '@altricade/core'; -import { getPresence, heartbeat } from '@altricade/core/api'; +import { heartbeat, getAvatarUploadUrl, uploadToUrl, setAvatar } from '@altricade/core/api'; import { SessionProvider, useSession } from '../entities/session'; import { AuthForm } from '../features/auth'; import { RealtimeProvider, useRealtime } from '../features/realtime'; @@ -41,45 +41,28 @@ const Dashboard = ({ user, onLogout }: { user: User; onLogout: () => void }): Re displayName: user.displayName, avatarUrl: user.avatarUrl, }; + const { updateUser } = useSession(); const { state } = useRealtime(); - const { conversations, startDirect, createGroupChat } = useConversations(user.id); - const [current, setCurrent] = useState(null); - const [onlineMap, setOnlineMap] = useState>({}); - const peerKey = conversations - .flatMap((conversation) => (conversation.peer === null ? [] : [conversation.peer.id])) - .join(','); - - // Poll presence for the people we have DMs with. - useEffect(() => { - const ids = peerKey.split(',').filter((id) => id.length > 0); - if (ids.length === 0) { - return undefined; + const onAvatar = (event: SyntheticEvent): void => { + const input = event.currentTarget; + const file = input.files?.[0]; + input.value = ''; + if (file === undefined) { + return; } - let cancelled = false; - const poll = async (): Promise => { - try { - const presences = await getPresence(apiConfig, ids); - if (!cancelled) { - const map: Record = {}; - for (const presence of presences) { - map[presence.userId] = presence.online; - } - setOnlineMap(map); - } - } catch { - // presence is best-effort - } - }; - void poll(); - const interval = setInterval(() => { - void poll(); - }, 20000); - return () => { - cancelled = true; - clearInterval(interval); - }; - }, [peerKey]); + const mime = file.type.length > 0 ? file.type : 'application/octet-stream'; + void (async () => { + const target = await getAvatarUploadUrl(apiConfig, { mime, size: file.size }); + await uploadToUrl(target.uploadUrl, file, mime); + updateUser(await setAvatar(apiConfig, { objectKey: target.objectKey })); + })(); + }; + const [current, setCurrent] = useState(null); + const { conversations, onlineMap, startDirect, createGroupChat } = useConversations( + user.id, + current?.id ?? null, + ); // Keep-alive heartbeat so our own last-seen stays fresh while connected. useEffect(() => { @@ -94,8 +77,18 @@ const Dashboard = ({ user, onLogout }: { user: User; onLogout: () => void }): Re return (
- - @{user.username} · socket: {state} + + + + @{user.username} · socket: {state} + diff --git a/packages/web/src/app/index.css b/packages/web/src/app/index.css index d5e67be..9323fb3 100644 --- a/packages/web/src/app/index.css +++ b/packages/web/src/app/index.css @@ -451,3 +451,55 @@ body { font-style: italic; padding: 0 0.25rem; } + +.topbar-user { + display: flex; + align-items: center; + gap: 0.6rem; +} + +.avatar { + width: 32px; + height: 32px; + border-radius: 50%; + object-fit: cover; + display: inline-flex; + align-items: center; + justify-content: center; +} + +.avatar-placeholder { + background: var(--color-accent); + color: #fff; + text-transform: uppercase; + font-weight: 600; +} + +.avatar-edit { + cursor: pointer; +} + +.attach-btn { + display: inline-flex; + align-items: center; + padding: 0 0.5rem; + cursor: pointer; + font-size: 1.2rem; +} + +.media-img { + max-width: 260px; + max-height: 260px; + border-radius: 8px; + display: block; +} + +.media-video { + max-width: 320px; + border-radius: 8px; + display: block; +} + +.media-file { + color: var(--color-accent); +} diff --git a/packages/web/src/entities/session/model.tsx b/packages/web/src/entities/session/model.tsx index 3acf75e..65cd785 100644 --- a/packages/web/src/entities/session/model.tsx +++ b/packages/web/src/entities/session/model.tsx @@ -18,6 +18,7 @@ export interface SessionContextValue { register: (body: RegisterBody) => Promise; login: (body: LoginBody) => Promise; logout: () => Promise; + updateUser: (user: User) => void; } const SessionContext = createContext(null); @@ -81,9 +82,13 @@ export const SessionProvider = ({ children }: { children: ReactNode }): ReactEle } }, [clear]); + const updateUser = useCallback((next: User): void => { + setUser(next); + }, []); + const value = useMemo( - () => ({ status, user, register, login, logout }), - [status, user, register, login, logout], + () => ({ status, user, register, login, logout, updateUser }), + [status, user, register, login, logout, updateUser], ); return {children}; diff --git a/packages/web/src/features/conversations/model.ts b/packages/web/src/features/conversations/model.ts index f3179ac..9797a51 100644 --- a/packages/web/src/features/conversations/model.ts +++ b/packages/web/src/features/conversations/model.ts @@ -1,6 +1,6 @@ -import { useCallback, useEffect, useState } from 'react'; -import type { Conversation } from '@altricade/core'; -import { userChannel } from '@altricade/core'; +import { useCallback, useEffect, useRef, useState } from 'react'; +import type { Conversation, Message } from '@altricade/core'; +import { conversationChannel, userChannel, EventType } from '@altricade/core'; import { listConversations, createDirect, createGroup } from '@altricade/core/api'; import { apiConfig } from '../../shared/api'; import { useRealtime } from '../realtime'; @@ -8,28 +8,72 @@ import { useRealtime } from '../realtime'; export interface UseConversations { conversations: Conversation[]; loading: boolean; + onlineMap: Record; startDirect: (username: string) => Promise; createGroupChat: (title: string, members: string[]) => Promise; } +const hasType = (data: unknown): data is { type: string } => + typeof data === 'object' && data !== null && 'type' in data && typeof data.type === 'string'; + const isConversationNew = ( data: unknown, -): data is { type: 'conversation.new'; conversation: Conversation } => { - if (typeof data !== 'object' || data === null) { - return false; - } - return 'type' in data && data.type === 'conversation.new' && 'conversation' in data; -}; +): data is { type: 'conversation.new'; conversation: Conversation } => + hasType(data) && data.type === EventType.ConversationNew && 'conversation' in data; + +const isMessageNew = (data: unknown): data is { type: 'message.new'; message: Message } => + hasType(data) && data.type === EventType.MessageNew && 'message' in data; const upsert = (list: Conversation[], conversation: Conversation): Conversation[] => { const rest = list.filter((item) => item.id !== conversation.id); return [conversation, ...rest]; }; -export const useConversations = (userId: string): UseConversations => { - const { subscribe } = useRealtime(); +const byRecency = (a: Conversation, b: Conversation): number => + b.lastMessageAt.localeCompare(a.lastMessageAt); + +export const useConversations = (userId: string, currentId: string | null): UseConversations => { + const { subscribe, onPresence, presence } = useRealtime(); const [conversations, setConversations] = useState([]); const [loading, setLoading] = useState(true); + const [onlineMap, setOnlineMap] = useState>({}); + const presentByChannel = useRef>>(new Map()); + const currentIdRef = useRef(currentId); + currentIdRef.current = currentId; + + const recomputeOnline = useCallback((): void => { + const online: Record = {}; + for (const set of presentByChannel.current.values()) { + for (const uid of set) { + online[uid] = true; + } + } + setOnlineMap(online); + }, []); + + const bumpUnread = useCallback( + (message: Message): void => { + if (message.senderId === userId) { + return; + } + setConversations((prev) => + prev + .map((c) => { + if (c.id !== message.conversationId) { + return c; + } + const isCurrent = currentIdRef.current === c.id; + return { + ...c, + lastMessageAt: message.createdAt, + unreadCount: isCurrent ? 0 : c.unreadCount + 1, + }; + }) + .sort(byRecency), + ); + }, + [userId], + ); useEffect(() => { let cancelled = false; @@ -51,16 +95,68 @@ export const useConversations = (userId: string): UseConversations => { }; }, []); - // New conversations (a DM someone started with me, or a group I was added to) - // arrive on my personal channel. + // Personal channel: new conversations + direct-message arrivals (for unread). useEffect(() => { return subscribe(userChannel(userId), (event) => { if (isConversationNew(event.data)) { const { conversation } = event.data; setConversations((prev) => upsert(prev, conversation)); + } else if (isMessageNew(event.data)) { + bumpUnread(event.data.message); } }); - }, [subscribe, userId]); + }, [subscribe, userId, bumpUnread]); + + // One subscription per conversation channel for LIVE presence (join/leave) and + // group message arrivals (for unread). Re-runs when the set of conversations changes. + const convKey = conversations + .map((c) => c.id) + .sort((a, b) => a.localeCompare(b)) + .join(','); + useEffect(() => { + const ids = convKey.split(',').filter((id) => id.length > 0); + const cleanups: (() => void)[] = []; + for (const id of ids) { + const channel = conversationChannel(id); + cleanups.push( + subscribe(channel, (event) => { + if (isMessageNew(event.data)) { + bumpUnread(event.data.message); + } + }), + ); + cleanups.push( + onPresence(channel, (action, uid) => { + const set = presentByChannel.current.get(channel) ?? new Set(); + if (action === 'join') { + set.add(uid); + } else { + set.delete(uid); + } + presentByChannel.current.set(channel, set); + recomputeOnline(); + }), + ); + void presence(channel).then((uids) => { + presentByChannel.current.set(channel, new Set(uids)); + recomputeOnline(); + }); + } + return () => { + for (const cleanup of cleanups) { + cleanup(); + } + }; + }, [convKey, subscribe, onPresence, presence, recomputeOnline, bumpUnread]); + + // Opening a conversation clears its unread badge immediately (markRead persists it). + useEffect(() => { + if (currentId !== null) { + setConversations((prev) => + prev.map((c) => (c.id === currentId ? { ...c, unreadCount: 0 } : c)), + ); + } + }, [currentId]); const startDirect = useCallback(async (username: string): Promise => { const conversation = await createDirect(apiConfig, { username }); @@ -77,5 +173,5 @@ export const useConversations = (userId: string): UseConversations => { [], ); - return { conversations, loading, startDirect, createGroupChat }; + return { conversations, loading, onlineMap, startDirect, createGroupChat }; }; diff --git a/packages/web/src/features/messaging/model.ts b/packages/web/src/features/messaging/model.ts index 50532cd..4299c9a 100644 --- a/packages/web/src/features/messaging/model.ts +++ b/packages/web/src/features/messaging/model.ts @@ -10,8 +10,17 @@ import type { TypingEvent, PublicUser, ReactionSummary, + MediaRef, + MediaKind, +} from '@altricade/core'; +import { + conversationChannel, + userChannel, + ephemeralChannel, + mergeMessages, + OPTIMISTIC_SEQ, + EventType, } from '@altricade/core'; -import { conversationChannel, userChannel, mergeMessages, OPTIMISTIC_SEQ } from '@altricade/core'; import { getHistory, sendMessage, @@ -20,8 +29,10 @@ import { addReaction, removeReaction, markRead, - sendTyping, + getUploadUrl, + uploadToUrl, } from '@altricade/core/api'; +import type { RealtimeEvent } from '@altricade/core/realtime'; import { apiConfig } from '../../shared/api'; import { useRealtime } from '../realtime'; @@ -32,24 +43,32 @@ export interface UseConversationMessages { /** For a direct conversation: the peer's last-read seq (drives ✓✓). */ peerReadSeq: number; send: (content: string) => Promise; + sendMedia: (file: File, caption: string) => Promise; edit: (messageId: string, content: string) => Promise; remove: (messageId: string) => Promise; toggleReaction: (message: Message, emoji: string) => Promise; notifyTyping: () => void; } +const mimeToKind = (mime: string): MediaKind => { + if (mime.startsWith('image/')) return 'image'; + if (mime.startsWith('video/')) return 'video'; + if (mime.startsWith('audio/')) return 'voice'; + return 'file'; +}; + const hasType = (data: unknown): data is { type: string } => typeof data === 'object' && data !== null && 'type' in data && typeof data.type === 'string'; const isMessageEvent = (d: unknown): d is MessageNewEvent | MessageEditEvent => - hasType(d) && (d.type === 'message.new' || d.type === 'message.edit') && 'message' in d; + hasType(d) && (d.type === EventType.MessageNew || d.type === EventType.MessageEdit) && 'message' in d; const isDeleteEvent = (d: unknown): d is MessageDeleteEvent => - hasType(d) && d.type === 'message.delete' && 'messageId' in d && 'conversationId' in d; + hasType(d) && d.type === EventType.MessageDelete && 'messageId' in d && 'conversationId' in d; const isReactionEvent = (d: unknown): d is ReactionEvent => - hasType(d) && (d.type === 'reaction.add' || d.type === 'reaction.remove'); -const isReadEvent = (d: unknown): d is ReadReceiptEvent => hasType(d) && d.type === 'read.receipt'; + hasType(d) && (d.type === EventType.ReactionAdd || d.type === EventType.ReactionRemove); +const isReadEvent = (d: unknown): d is ReadReceiptEvent => hasType(d) && d.type === EventType.ReadReceipt; const isTypingEvent = (d: unknown): d is TypingEvent => - hasType(d) && (d.type === 'typing.start' || d.type === 'typing.stop'); + hasType(d) && (d.type === EventType.TypingStart || d.type === EventType.TypingStop); const applyReaction = ( reactions: ReactionSummary[], @@ -76,7 +95,7 @@ export const useConversationMessages = ( conversation: Conversation, me: PublicUser, ): UseConversationMessages => { - const { subscribe } = useRealtime(); + const { subscribe, publish } = useRealtime(); const [messages, setMessages] = useState([]); const [loading, setLoading] = useState(true); const [typingUserIds, setTypingUserIds] = useState([]); @@ -100,7 +119,7 @@ export const useConversationMessages = ( setTypingUserIds((prev) => prev.filter((id) => id !== userId)); }; - const unsubscribe = subscribe(channel, (event) => { + const handler = (event: RealtimeEvent): void => { const data = event.data; if (isMessageEvent(data)) { @@ -124,7 +143,7 @@ export const useConversationMessages = ( if (isReactionEvent(data)) { if (data.conversationId === conversationId) { const { messageId, emoji, userId } = data; - const delta = data.type === 'reaction.add' ? 1 : -1; + const delta = data.type === EventType.ReactionAdd ? 1 : -1; setMessages((prev) => prev.map((m) => m.id === messageId @@ -147,7 +166,7 @@ export const useConversationMessages = ( return; } const { userId } = data; - if (data.type === 'typing.stop') { + if (data.type === EventType.TypingStop) { clearTyping(userId); return; } @@ -163,7 +182,11 @@ export const useConversationMessages = ( }, 4000), ); } - }); + }; + + // Durable events arrive on the message channel; typing on the ephemeral channel. + const unsubMessages = subscribe(channel, handler); + const unsubEphemeral = subscribe(ephemeralChannel(conversationId), handler); const load = async (): Promise => { try { @@ -181,7 +204,8 @@ export const useConversationMessages = ( return () => { cancelled = true; - unsubscribe(); + unsubMessages(); + unsubEphemeral(); for (const timer of timers.values()) { clearTimeout(timer); } @@ -216,6 +240,7 @@ export const useConversationMessages = ( editedAt: null, deletedAt: null, reactions: [], + media: null, }; setMessages((prev) => mergeMessages(prev, [optimistic])); const confirmed = await sendMessage(apiConfig, conversationId, { content, clientMsgId }); @@ -224,6 +249,25 @@ export const useConversationMessages = ( [conversationId, me], ); + const sendMedia = useCallback( + async (file: File, caption: string): Promise => { + const mime = file.type.length > 0 ? file.type : 'application/octet-stream'; + const kind = mimeToKind(mime); + const { uploadUrl, objectKey } = await getUploadUrl(apiConfig, { kind, mime, size: file.size }); + await uploadToUrl(uploadUrl, file, mime); + const media: MediaRef = { kind, mime, size: file.size, name: file.name }; + const clientMsgId = crypto.randomUUID(); + const confirmed = await sendMessage(apiConfig, conversationId, { + content: caption, + clientMsgId, + mediaKey: objectKey, + media, + }); + setMessages((prev) => mergeMessages(prev, [confirmed])); + }, + [conversationId], + ); + const edit = useCallback( async (messageId: string, content: string): Promise => { const updated = await apiEdit(apiConfig, conversationId, messageId, { content }); @@ -255,9 +299,11 @@ export const useConversationMessages = ( const now = Date.now(); if (now - lastTypingSent.current > 2500) { lastTypingSent.current = now; - void sendTyping(apiConfig, conversationId, 'start'); + // Client-side publish straight to Centrifugo (ephemeral, no backend round-trip). + const event: TypingEvent = { type: EventType.TypingStart, conversationId, userId: me.id }; + void publish(ephemeralChannel(conversationId), event); } - }, [conversationId]); + }, [conversationId, me.id, publish]); return { messages, @@ -265,6 +311,7 @@ export const useConversationMessages = ( typingUserIds, peerReadSeq, send, + sendMedia, edit, remove, toggleReaction, diff --git a/packages/web/src/features/messaging/ui/ChatView.tsx b/packages/web/src/features/messaging/ui/ChatView.tsx index 156ffda..6bdfc73 100644 --- a/packages/web/src/features/messaging/ui/ChatView.tsx +++ b/packages/web/src/features/messaging/ui/ChatView.tsx @@ -1,6 +1,8 @@ -import { useState } from 'react'; +import { useEffect, useState } from 'react'; import type { ReactElement, SyntheticEvent } from 'react'; import type { Conversation, Message, PublicUser } from '@altricade/core'; +import { getMediaUrl } from '@altricade/core/api'; +import { apiConfig } from '../../../shared/api'; import { useConversationMessages } from '../model'; interface Props { @@ -17,9 +19,70 @@ const headerTitle = (conversation: Conversation): string => { return conversation.peer === null ? 'Direct' : `@${conversation.peer.username}`; }; +const MediaView = ({ + conversationId, + message, +}: { + conversationId: string; + message: Message; +}): ReactElement | null => { + const [url, setUrl] = useState(null); + const media = message.media; + + useEffect(() => { + if (media === null) { + return undefined; + } + let cancelled = false; + getMediaUrl(apiConfig, conversationId, message.id) + .then((resolved) => { + if (!cancelled) { + setUrl(resolved); + } + }) + .catch(() => { + /* media may be unavailable */ + }); + return () => { + cancelled = true; + }; + }, [conversationId, message.id, media]); + + if (media === null) { + return null; + } + if (url === null) { + return loading media…; + } + if (media.kind === 'image') { + return {media.name}; + } + if (media.kind === 'video') { + return