425 lines
16 KiB
TypeScript
425 lines
16 KiB
TypeScript
import { randomUUID } from 'node:crypto';
|
|
import { EventType, userChannel } from '@altricade/core';
|
|
import type {
|
|
Message,
|
|
MessageNewEvent,
|
|
MessageEditEvent,
|
|
MessageDeleteEvent,
|
|
MessageHiddenEvent,
|
|
MessagePinEvent,
|
|
ReactionEvent,
|
|
SendMessageBody,
|
|
MediaTab,
|
|
ForwardOrigin,
|
|
} from '@altricade/core';
|
|
import { HttpError } from '../../shared/http-error';
|
|
import type { Publisher } from '../../shared/publisher';
|
|
import type { ConversationsRepository } from '../conversations';
|
|
import type { Deliver } from '../conversations';
|
|
import type { MessagesRepository } from './messages.repository';
|
|
import { toMessage } from './messages.mapper';
|
|
|
|
export interface MessagesServiceDeps {
|
|
messages: MessagesRepository;
|
|
conversations: ConversationsRepository;
|
|
deliver: Deliver;
|
|
/** Personal-channel publisher for per-user view-state events. */
|
|
publish: Publisher;
|
|
mediaDownloadUrl: (objectKey: string) => Promise<string>;
|
|
/** Fire-and-forget push-notification hook, called for each newly-created message. */
|
|
notify: (message: Message) => void;
|
|
}
|
|
|
|
// Telegram-style shared-media tabs → media kinds. Round video messages live in
|
|
// the Voice tab alongside voice notes (both are "instant" messages), not Media.
|
|
const TAB_KINDS: Record<MediaTab, string[]> = {
|
|
media: ['image', 'video'],
|
|
files: ['file'],
|
|
voice: ['voice', 'video_note'],
|
|
};
|
|
|
|
export interface SentMessage {
|
|
message: Message;
|
|
created: boolean;
|
|
}
|
|
|
|
export interface MessagesService {
|
|
send(conversationId: string, senderId: string, input: SendMessageBody): Promise<SentMessage>;
|
|
history(
|
|
conversationId: string,
|
|
userId: string,
|
|
beforeSeq: number | null,
|
|
limit: number,
|
|
): Promise<Message[]>;
|
|
historyAfter(
|
|
conversationId: string,
|
|
userId: string,
|
|
afterSeq: number,
|
|
limit: number,
|
|
): Promise<Message[]>;
|
|
/** Window centered on a message — jump-to-message / search result context. */
|
|
context(
|
|
conversationId: string,
|
|
userId: string,
|
|
messageId: string,
|
|
limit: number,
|
|
): Promise<Message[]>;
|
|
edit(
|
|
conversationId: string,
|
|
messageId: string,
|
|
userId: string,
|
|
content: string,
|
|
): Promise<Message>;
|
|
remove(conversationId: string, messageId: string, userId: string): Promise<void>;
|
|
hide(conversationId: string, messageId: string, userId: string): Promise<void>;
|
|
forward(
|
|
targetConversationId: string,
|
|
userId: string,
|
|
sourceConversationId: string,
|
|
messageId: string,
|
|
hideSender: boolean,
|
|
): Promise<Message>;
|
|
pin(conversationId: string, messageId: string, userId: string): Promise<void>;
|
|
unpin(conversationId: string, messageId: string, userId: string): Promise<void>;
|
|
pinned(conversationId: string, userId: string): Promise<Message[]>;
|
|
media(
|
|
conversationId: string,
|
|
userId: string,
|
|
tab: MediaTab,
|
|
beforeSeq: number | null,
|
|
limit: number,
|
|
): Promise<Message[]>;
|
|
addReaction(
|
|
conversationId: string,
|
|
messageId: string,
|
|
userId: string,
|
|
emoji: string,
|
|
): Promise<void>;
|
|
removeReaction(
|
|
conversationId: string,
|
|
messageId: string,
|
|
userId: string,
|
|
emoji: string,
|
|
): Promise<void>;
|
|
mediaUrl(conversationId: string, messageId: string, userId: string): Promise<string>;
|
|
}
|
|
|
|
export const createMessagesService = (deps: MessagesServiceDeps): MessagesService => {
|
|
const { messages, conversations, deliver, publish, mediaDownloadUrl, notify } = deps;
|
|
|
|
const assertMember = async (conversationId: string, userId: string): Promise<void> => {
|
|
if (!(await conversations.isMember(conversationId, userId))) {
|
|
throw new HttpError(403, 'not_a_member', 'You are not a member of this conversation');
|
|
}
|
|
};
|
|
|
|
const assertMessageIn = async (conversationId: string, messageId: string): Promise<void> => {
|
|
if ((await messages.getConversationId(messageId)) !== conversationId) {
|
|
throw new HttpError(404, 'not_found', 'Message not found');
|
|
}
|
|
};
|
|
|
|
const loadMessage = async (messageId: string, userId: string): Promise<Message> => {
|
|
const row = await messages.getWithSenderById(messageId);
|
|
if (row === undefined) {
|
|
throw new HttpError(404, 'not_found', 'Message not found');
|
|
}
|
|
const reactions = await messages.reactionsFor([messageId], userId);
|
|
return toMessage(row, reactions.get(messageId) ?? []);
|
|
};
|
|
|
|
// Channels are broadcast: only the owner and admins may post.
|
|
const assertCanPost = async (conversationId: string, userId: string): Promise<void> => {
|
|
const conversation = await conversations.findById(conversationId);
|
|
if (conversation?.type !== 'channel') {
|
|
return;
|
|
}
|
|
const role = await conversations.getRole(conversationId, userId);
|
|
if (role !== 'owner' && role !== 'admin') {
|
|
throw new HttpError(403, 'not_channel_admin', 'Only channel admins can post');
|
|
}
|
|
};
|
|
|
|
return {
|
|
send: async (conversationId, senderId, input) => {
|
|
await assertMember(conversationId, senderId);
|
|
await assertCanPost(conversationId, senderId);
|
|
|
|
// Only honor a reply target that lives in this same conversation.
|
|
let replyToId: string | null = null;
|
|
if (input.replyToId !== undefined) {
|
|
if ((await messages.getConversationId(input.replyToId)) === conversationId) {
|
|
replyToId = input.replyToId;
|
|
}
|
|
}
|
|
|
|
const inserted = await messages.insert({
|
|
conversationId,
|
|
senderId,
|
|
clientMsgId: input.clientMsgId,
|
|
content: input.content ?? '',
|
|
contentType: input.contentType ?? 'text',
|
|
encryption: input.encryption ?? null,
|
|
mediaKey: input.mediaKey ?? null,
|
|
media: input.media ?? null,
|
|
replyToId,
|
|
forwarded: false,
|
|
forwardedFrom: null,
|
|
});
|
|
|
|
const id =
|
|
inserted?.id ??
|
|
(await messages.findIdByDedupe(conversationId, senderId, input.clientMsgId));
|
|
if (id === undefined) {
|
|
throw new HttpError(500, 'internal_error', 'Message could not be persisted');
|
|
}
|
|
const row = await messages.getWithSenderById(id);
|
|
if (row === undefined) {
|
|
throw new HttpError(500, 'internal_error', 'Message not found after insert');
|
|
}
|
|
const message = toMessage(row, []);
|
|
|
|
if (inserted !== undefined) {
|
|
await conversations.touchLastMessage(conversationId);
|
|
const event: MessageNewEvent = { type: EventType.MessageNew, message };
|
|
await deliver(conversationId, event);
|
|
// Enqueue a push job (offline recipients only; online ones get it live).
|
|
notify(message);
|
|
}
|
|
return { message, created: inserted !== undefined };
|
|
},
|
|
|
|
history: async (conversationId, userId, beforeSeq, limit) => {
|
|
await assertMember(conversationId, userId);
|
|
const rows = await messages.listHistory(conversationId, userId, beforeSeq, limit);
|
|
const reactions = await messages.reactionsFor(
|
|
rows.map((row) => row.id),
|
|
userId,
|
|
);
|
|
return rows.map((row) => toMessage(row, reactions.get(row.id) ?? []));
|
|
},
|
|
|
|
historyAfter: async (conversationId, userId, afterSeq, limit) => {
|
|
await assertMember(conversationId, userId);
|
|
const rows = await messages.listAfter(conversationId, userId, afterSeq, limit);
|
|
const reactions = await messages.reactionsFor(
|
|
rows.map((row) => row.id),
|
|
userId,
|
|
);
|
|
return rows.map((row) => toMessage(row, reactions.get(row.id) ?? []));
|
|
},
|
|
|
|
context: async (conversationId, userId, messageId, limit) => {
|
|
await assertMember(conversationId, userId);
|
|
const target = await messages.getWithSenderById(messageId);
|
|
if (target === undefined) {
|
|
throw new HttpError(404, 'not_found', 'Message not found');
|
|
}
|
|
if (target.conversation_id !== conversationId) {
|
|
throw new HttpError(404, 'not_found', 'Message not found');
|
|
}
|
|
const half = Math.max(1, Math.floor(limit / 2));
|
|
const rows = await messages.listContext(conversationId, userId, target.seq, half);
|
|
const reactions = await messages.reactionsFor(
|
|
rows.map((row) => row.id),
|
|
userId,
|
|
);
|
|
return rows.map((row) => toMessage(row, reactions.get(row.id) ?? []));
|
|
},
|
|
|
|
media: async (conversationId, userId, tab, beforeSeq, limit) => {
|
|
await assertMember(conversationId, userId);
|
|
const rows = await messages.listMedia(conversationId, userId, TAB_KINDS[tab], beforeSeq, limit);
|
|
return rows.map((row) => toMessage(row, []));
|
|
},
|
|
|
|
edit: async (conversationId, messageId, userId, content) => {
|
|
await assertMember(conversationId, userId);
|
|
await assertMessageIn(conversationId, messageId);
|
|
const edited = await messages.editContent(messageId, userId, content);
|
|
if (edited === undefined) {
|
|
throw new HttpError(403, 'not_editable', 'You can only edit your own messages');
|
|
}
|
|
const message = await loadMessage(messageId, userId);
|
|
const event: MessageEditEvent = { type: EventType.MessageEdit, message };
|
|
await deliver(conversationId, event);
|
|
return message;
|
|
},
|
|
|
|
remove: async (conversationId, messageId, userId) => {
|
|
await assertMember(conversationId, userId);
|
|
await assertMessageIn(conversationId, messageId);
|
|
let deleted = await messages.softDelete(messageId, userId);
|
|
if (deleted === undefined) {
|
|
// Not the sender: the owner of a group/channel may moderate any message in it.
|
|
const conversation = await conversations.findById(conversationId);
|
|
const role = await conversations.getRole(conversationId, userId);
|
|
if (conversation === undefined || conversation.type === 'direct' || role !== 'owner') {
|
|
throw new HttpError(
|
|
403,
|
|
'not_deletable',
|
|
'You can only delete your own messages (group owners can delete any)',
|
|
);
|
|
}
|
|
deleted = await messages.softDeleteAny(messageId);
|
|
}
|
|
if (deleted === undefined) {
|
|
// Already deleted — idempotent success, nothing to broadcast.
|
|
return;
|
|
}
|
|
const event: MessageDeleteEvent = { type: EventType.MessageDelete, conversationId, messageId };
|
|
await deliver(conversationId, event);
|
|
},
|
|
|
|
hide: async (conversationId, messageId, userId) => {
|
|
await assertMember(conversationId, userId);
|
|
await assertMessageIn(conversationId, messageId);
|
|
await messages.hide(messageId, userId);
|
|
// Personal channel only — other participants are unaffected by design.
|
|
const event: MessageHiddenEvent = { type: EventType.MessageHidden, conversationId, messageId };
|
|
await publish(userChannel(userId), event);
|
|
},
|
|
|
|
forward: async (targetConversationId, userId, sourceConversationId, messageId, hideSender) => {
|
|
// Must belong to both ends, and be able to post into the target.
|
|
await assertMember(sourceConversationId, userId);
|
|
await assertMember(targetConversationId, userId);
|
|
await assertCanPost(targetConversationId, userId);
|
|
|
|
const source = await messages.getWithSenderById(messageId);
|
|
if (source === undefined) {
|
|
throw new HttpError(404, 'not_found', 'Message not found');
|
|
}
|
|
if (source.conversation_id !== sourceConversationId) {
|
|
throw new HttpError(404, 'not_found', 'Message not found');
|
|
}
|
|
if (source.deleted_at !== null) {
|
|
throw new HttpError(400, 'invalid_target', 'Cannot forward a deleted message');
|
|
}
|
|
|
|
// Attribution (Telegram-style): forwarding an already-forwarded message
|
|
// preserves the ORIGINAL origin; forwarding from a channel credits the
|
|
// channel (no profile link); otherwise credits the author (clickable).
|
|
let origin: ForwardOrigin | null;
|
|
if (source.forwarded) {
|
|
origin = source.forwarded_from;
|
|
} else {
|
|
const sourceConv = await conversations.findById(sourceConversationId);
|
|
if (sourceConv?.type === 'channel') {
|
|
origin = { name: sourceConv.title ?? 'Channel', user: null };
|
|
} else {
|
|
origin = {
|
|
name: source.sender_display_name,
|
|
user: {
|
|
id: source.sender_id,
|
|
username: source.sender_username,
|
|
displayName: source.sender_display_name,
|
|
avatarUrl: source.sender_avatar_ref,
|
|
},
|
|
};
|
|
}
|
|
}
|
|
|
|
const clientMsgId = randomUUID();
|
|
const inserted = await messages.insert({
|
|
conversationId: targetConversationId,
|
|
senderId: userId,
|
|
clientMsgId,
|
|
content: source.content,
|
|
contentType: source.content_type,
|
|
encryption: source.encryption,
|
|
// Object storage is shared — reference the same media object.
|
|
mediaKey: source.media_key,
|
|
media: source.media_meta,
|
|
replyToId: null,
|
|
forwarded: true,
|
|
forwardedFrom: hideSender ? null : origin,
|
|
});
|
|
const id =
|
|
inserted?.id ??
|
|
(await messages.findIdByDedupe(targetConversationId, userId, clientMsgId));
|
|
if (id === undefined) {
|
|
throw new HttpError(500, 'internal_error', 'Forward could not be persisted');
|
|
}
|
|
const row = await messages.getWithSenderById(id);
|
|
if (row === undefined) {
|
|
throw new HttpError(500, 'internal_error', 'Message not found after forward');
|
|
}
|
|
const message = toMessage(row, []);
|
|
await conversations.touchLastMessage(targetConversationId);
|
|
const event: MessageNewEvent = { type: EventType.MessageNew, message };
|
|
await deliver(targetConversationId, event);
|
|
notify(message);
|
|
return message;
|
|
},
|
|
|
|
pin: async (conversationId, messageId, userId) => {
|
|
await assertMember(conversationId, userId);
|
|
await assertMessageIn(conversationId, messageId);
|
|
await assertCanPost(conversationId, userId); // channels: admins only
|
|
await messages.pin(conversationId, messageId, userId);
|
|
const event: MessagePinEvent = { type: EventType.MessagePin, conversationId, messageId };
|
|
await deliver(conversationId, event);
|
|
},
|
|
|
|
unpin: async (conversationId, messageId, userId) => {
|
|
await assertMember(conversationId, userId);
|
|
await assertMessageIn(conversationId, messageId);
|
|
await assertCanPost(conversationId, userId);
|
|
await messages.unpin(conversationId, messageId);
|
|
const event: MessagePinEvent = { type: EventType.MessageUnpin, conversationId, messageId };
|
|
await deliver(conversationId, event);
|
|
},
|
|
|
|
pinned: async (conversationId, userId) => {
|
|
await assertMember(conversationId, userId);
|
|
const rows = await messages.listPinned(conversationId);
|
|
return rows.map((row) => toMessage(row, []));
|
|
},
|
|
|
|
addReaction: async (conversationId, messageId, userId, emoji) => {
|
|
await assertMember(conversationId, userId);
|
|
await assertMessageIn(conversationId, messageId);
|
|
await messages.addReaction(messageId, userId, emoji);
|
|
const event: ReactionEvent = {
|
|
type: EventType.ReactionAdd,
|
|
conversationId,
|
|
messageId,
|
|
emoji,
|
|
userId,
|
|
};
|
|
await deliver(conversationId, event);
|
|
},
|
|
|
|
removeReaction: async (conversationId, messageId, userId, emoji) => {
|
|
await assertMember(conversationId, userId);
|
|
await assertMessageIn(conversationId, messageId);
|
|
await messages.removeReaction(messageId, userId, emoji);
|
|
const event: ReactionEvent = {
|
|
type: EventType.ReactionRemove,
|
|
conversationId,
|
|
messageId,
|
|
emoji,
|
|
userId,
|
|
};
|
|
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);
|
|
},
|
|
};
|
|
};
|
|
|
|
declare module 'fastify' {
|
|
interface FastifyInstance {
|
|
messagesService: MessagesService;
|
|
}
|
|
}
|