eduvia/workavia-drive
11
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: removed messages services

Browse Source
This commit is contained in:
montaghanmy
2023-03-31 09:32:35 +01:00
parent cc2f8543a9
commit 06b8584a4b
47 changed files with 0 additions and 6503 deletions
Download Patch File Download Diff File Expand all files Collapse all files
twake/backend/node/src/services/messages/blocks-types.ts
-208
View File
@@ -1,208 +0,0 @@
import { MessageFileMetadata } from "./entities/message-files";
/* Blocks objects */
export type Block = { block_id?: string } & (
| BlockFile
| BlockActions
| BlockContext
| BlockHeader
| BlockDivider
| BlockImage
| BlockInput
| BlockSection
| BlockIframe
| BlockTwacode
);
export type BlockTwacode = {
type: "twacode";
elements: any;
};
export type BlockActions = {
type: "actions";
elements: BlockElement[];
};
export type BlockContext = {
type: "context";
elements: (BlockElementImage | CompositionTextObject | BlockElementProgressBar)[];
};
export type BlockHeader = {
type: "header";
text: CompositionPlainTextObject;
};
export type BlockDivider = {
type: "divider";
};
export type BlockFile = {
type: "file";
external_id: string;
source: string;
metadata?: MessageFileMetadata;
};
export type BlockImage = BlockElementImage & {
title?: CompositionPlainTextObject;
};
export type BlockInput = {
type: "input";
label: string;
element: BlockElement;
dispatch_action?: boolean;
hint?: CompositionPlainTextObject;
optional?: boolean;
};
export type BlockSection = {
type: "section";
text?: CompositionTextObject;
fields?: CompositionTextObject[];
accessory?: BlockElement;
};
export type BlockIframe = {
type: "iframe";
iframe_url: string;
width: number;
height: number;
};
/* Elements objects */
export type BlockElement =
| BlockElementImage
| BlockElementProgressBar
| BlockElementButton
| BlockElementCheckboxes
| BlockElementDatePicker
| BlockElementMultiselectMenu
| BlockElementPlaintextInput
| BlockElementRadioButtonInput
| BlockElementSelectMenus
| BlockElementOverflowMenus
| BlockElementTimePicker;
export type BlockElementImage = {
type: "image";
image_url: string;
alt_text: string;
title?: any;
metadata?: MessageFileMetadata;
};
export type BlockElementProgressBar = {
type: "progress_bar";
value: number; //Between 0 and 100
title: string;
};
export type BlockElementButton = {
type: "button";
text: CompositionPlainTextObject;
action_id: string;
url?: string;
value?: string;
style?: "primary" | "danger" | "default";
confirm?: CompositionConfirmationDialog;
};
export type BlockElementCheckboxes = {
type: "checkboxes";
action_id: string;
options: CompositionOption[];
initial_options?: CompositionOption[];
confirm?: CompositionConfirmationDialog;
};
type BlockElementPlaintextInput = {
type: "plain_text_input";
action_id: string;
placeholder?: CompositionPlainTextObject;
initial_value?: string;
multiline?: boolean;
min_length?: number;
max_length?: number;
dispatch_action_config?: DispatchActionConfiguration;
readonly?: boolean;
copiable?: boolean;
};
export type BlockElementRadioButtonInput = {
type: "radio_buttons";
action_id: string;
options: CompositionOption[];
initial_option?: CompositionOption;
confirm?: CompositionConfirmationDialog;
};
export type BlockElementDatePicker = {
type: "datepicker";
action_id: string;
placeholder?: CompositionPlainTextObject;
initial_date?: string;
confirm?: CompositionConfirmationDialog;
};
export type BlockElementTimePicker = {
type: "timepicker";
action_id: string;
placeholder?: CompositionPlainTextObject;
initial_time?: string;
confirm?: CompositionConfirmationDialog;
};
export type BlockElementOverflowMenus = {
type: "overflow";
action_id: string;
options: CompositionOption[];
confirm?: CompositionConfirmationDialog;
};
export type BlockElementSelectMenus = {
type: "";
}; //TODO
export type BlockElementMultiselectMenu = {
type: any;
}; //TODO
/* Composition objects */
type CompositionTextObject = CompositionPlainTextObject | CompositionMarkdownTextObject;
type CompositionPlainTextObject = {
type: "plain_text";
text: string;
emoji?: boolean;
verbatim?: boolean;
};
type CompositionMarkdownTextObject = {
type: "mrkdwn";
text: string;
};
type CompositionConfirmationDialog = {
title: CompositionPlainTextObject;
text: CompositionMarkdownTextObject;
confirm: CompositionPlainTextObject;
deny: CompositionPlainTextObject;
style: "confirm" | "danger" | "default";
};
type CompositionOption = {
text: CompositionPlainTextObject;
value: string;
description?: CompositionPlainTextObject;
url?: string;
};
type DispatchActionConfiguration = {
trigger_actions_on: ("on_enter_pressed" | "on_character_entered")[];
};
twake/backend/node/src/services/messages/entities/message-channel-marked-refs.ts
-51
View File
@@ -1,51 +0,0 @@
import { Type } from "class-transformer";
import { merge } from "lodash";
import { Column, Entity } from "../../../core/platform/services/database/services/orm/decorators";
export const TYPE = "message_channel_marked_refs";
@Entity(TYPE, {
primaryKey: [["company_id", "workspace_id"], "type", "channel_id", "thread_id", "message_id"],
type: TYPE,
})
export class MessageChannelMarkedRef {
@Type(() => String)
@Column("company_id", "uuid")
company_id: string;
@Type(() => String)
@Column("workspace_id", "string")
workspace_id: string;
@Type(() => String)
@Column("channel_id", "string")
channel_id: string;
@Type(() => String)
@Column("type", "string")
type: "pinned";
@Type(() => String)
@Column("thread_id", "timeuuid", { order: "DESC" })
thread_id: string;
@Type(() => String)
@Column("message_id", "timeuuid", { order: "DESC" })
message_id: string;
@Type(() => Number)
@Column("created_at", "number")
created_at: number;
@Type(() => String)
@Column("created_by", "string")
created_by: string;
}
export type MessageChannelMarkedRefPrimaryKey = Pick<
MessageChannelMarkedRef,
"company_id" | "workspace_id" | "type" | "channel_id" | "thread_id" | "message_id"
>;
export function getInstance(ref: MessageChannelMarkedRef): MessageChannelMarkedRef {
return merge(new MessageChannelMarkedRef(), ref);
}
twake/backend/node/src/services/messages/entities/message-channel-refs-reversed.ts
-39
View File
@@ -1,39 +0,0 @@
import { Type } from "class-transformer";
import { merge } from "lodash";
import { Column, Entity } from "../../../core/platform/services/database/services/orm/decorators";
export const TYPE = "message_channel_refs_reversed";
@Entity(TYPE, {
primaryKey: [["company_id", "workspace_id"], "channel_id", "thread_id"],
type: TYPE,
})
export class MessageChannelRefReversed {
@Type(() => String)
@Column("company_id", "uuid")
company_id: string;
@Type(() => String)
@Column("workspace_id", "string")
workspace_id: string;
@Type(() => String)
@Column("channel_id", "string")
channel_id: string;
@Type(() => String)
@Column("thread_id", "timeuuid", { order: "DESC" })
thread_id: string;
@Type(() => String)
@Column("message_id", "timeuuid")
message_id: string;
}
export type MessageChannelRefReversedPrimaryKey = Pick<
MessageChannelRefReversed,
"company_id" | "workspace_id" | "channel_id" | "thread_id"
>;
export function getInstance(ref: MessageChannelRefReversed): MessageChannelRefReversed {
return merge(new MessageChannelRefReversed(), ref);
}
twake/backend/node/src/services/messages/entities/message-channel-refs.ts
-39
View File
@@ -1,39 +0,0 @@
import { Type } from "class-transformer";
import { merge } from "lodash";
import { Column, Entity } from "../../../core/platform/services/database/services/orm/decorators";
export const TYPE = "message_channel_refs";
@Entity(TYPE, {
primaryKey: [["company_id", "workspace_id"], "channel_id", "message_id"],
type: TYPE,
})
export class MessageChannelRef {
@Type(() => String)
@Column("company_id", "uuid")
company_id: string;
@Type(() => String)
@Column("workspace_id", "string")
workspace_id: string;
@Type(() => String)
@Column("channel_id", "string")
channel_id: string;
@Type(() => String)
@Column("message_id", "timeuuid", { order: "DESC" })
message_id: string;
@Type(() => String)
@Column("thread_id", "timeuuid")
thread_id: string;
}
export type MessageChannelRefPrimaryKey = Pick<
MessageChannelRef,
"company_id" | "workspace_id" | "channel_id" | "message_id"
>;
export function getInstance(ref: MessageChannelRef): MessageChannelRef {
return merge(new MessageChannelRef(), ref);
}
twake/backend/node/src/services/messages/entities/message-file-refs.ts
-59
View File
@@ -1,59 +0,0 @@
import { Type } from "class-transformer";
import { merge } from "lodash";
import { Column, Entity } from "../../../core/platform/services/database/services/orm/decorators";
export const TYPE = "message_file_refs";
@Entity(TYPE, {
primaryKey: [["target_type", "company_id", "target_id"], "id"],
type: TYPE,
})
export class MessageFileRef {
@Type(() => String)
@Column("target_type", "string")
target_type: "channel" | "channel_media" | "channel_file" | "user_upload" | "user_download";
@Type(() => String)
@Column("target_id", "string")
target_id: string;
@Type(() => String)
@Column("id", "timeuuid", { generator: "timeuuid", order: "DESC" })
id: string;
@Type(() => Number)
@Column("created_at", "number")
created_at: number;
@Type(() => String)
@Column("workspace_id", "string")
workspace_id: string;
@Type(() => String)
@Column("channel_id", "string")
channel_id: string;
@Type(() => String)
@Column("thread_id", "timeuuid")
thread_id: string;
@Type(() => String)
@Column("message_id", "timeuuid")
message_id: string;
@Type(() => String)
@Column("message_file_id", "string")
message_file_id: string;
@Type(() => String)
@Column("file_id", "json")
file_id: string;
@Column("company_id", "timeuuid")
company_id: string;
}
export type MessageFileRefPrimaryKey = Pick<MessageFileRef, "target_type" | "target_id" | "id">;
export function getInstance(ref: MessageFileRef): MessageFileRef {
return merge(new MessageFileRef(), ref);
}
twake/backend/node/src/services/messages/entities/message-files.search.ts
-54
View File
@@ -1,54 +0,0 @@
import _ from "lodash";
import { fileIsMedia } from "../../../services/files/utils";
import { MessageFile } from "./message-files";
export const expandFileNameForSearch = (name: string) => {
return name + " " + _.snakeCase(name).replace(/[^a-zA-Z0-9]/gm, " ");
};
export default {
index: "message_files",
source: (entity: MessageFile) => {
const isMedia = fileIsMedia(entity);
const source = {
name: expandFileNameForSearch(entity.metadata?.name || ""),
size: entity.metadata?.size || "",
source: entity.metadata?.source || "",
extension: (entity.metadata?.name || "").split(".").pop() || "",
is_media: isMedia,
is_file: !isMedia,
created_at: entity.created_at,
cache_company_id: entity.cache?.company_id,
cache_workspace_id: entity.cache?.workspace_id,
cache_channel_id: entity.cache?.channel_id,
cache_user_id: entity.cache?.user_id,
};
return source;
},
mongoMapping: {
text: {
name: "text",
},
prefix: {
name: "prefix",
},
},
esMapping: {
properties: {
name: { type: "text", index_prefixes: { min_chars: 1, max_chars: 20 } },
size: { type: "number" },
source: { type: "keyword" },
extension: { type: "keyword" },
is_media: { type: "boolean" },
is_file: { type: "boolean" },
created_at: { type: "number" },
cache_company_id: { type: "keyword" },
cache_workspace_id: { type: "keyword" },
cache_channel_id: { type: "keyword" },
cache_user_id: { type: "keyword" },
},
},
};
twake/backend/node/src/services/messages/entities/message-files.ts
-54
View File
@@ -1,54 +0,0 @@
import { Type } from "class-transformer";
import { Thumbnail } from "../../files/entities/file";
import { Column, Entity } from "../../../core/platform/services/database/services/orm/decorators";
import search from "./message-files.search";
export const TYPE = "message_files";
@Entity(TYPE, {
primaryKey: [["message_id"], "id"],
type: TYPE,
search,
})
export class MessageFile {
@Type(() => String)
@Column("message_id", "timeuuid", { order: "DESC" })
message_id?: string;
//Timeuuid is used please do not change to uuid
@Type(() => String)
@Column("id", "timeuuid", { order: "DESC" })
id: string;
@Type(() => String)
@Column("thread_id", "string")
thread_id?: string;
@Column("company_id", "timeuuid")
company_id: string;
@Column("created_at", "number")
created_at: number;
@Column("metadata", "encoded_json")
metadata: MessageFileMetadata;
@Column("cache", "encoded_json")
cache: null | {
company_id: string;
workspace_id: string;
channel_id: string;
user_id: string;
};
}
export type MessageFileMetadata = {
source: "internal" | "drive" | string; //Uuid of the corresponding connector
external_id: string | any;
name?: string; //Original file name
mime?: string; //Original file mime
size?: number; //Original file weight
thumbnails?: Thumbnail[]; //Url to thumbnail (or set it to undefined if no relevant)
};
export type MessageFilePrimaryKey = Pick<MessageFile, "message_id" | "id">;
twake/backend/node/src/services/messages/entities/message-user-inbox-refs-reversed.ts
-35
View File
@@ -1,35 +0,0 @@
import { Type } from "class-transformer";
import { merge } from "lodash";
import { Column, Entity } from "../../../core/platform/services/database/services/orm/decorators";
export const TYPE = "message_user_inbox_refs_reversed";
@Entity(TYPE, {
primaryKey: [["company_id"], "user_id", "thread_id"],
type: TYPE,
})
export class MessageUserInboxRefReversed {
@Type(() => String)
@Column("company_id", "uuid")
company_id: string;
@Type(() => String)
@Column("user_id", "string")
user_id: string;
@Type(() => String)
@Column("thread_id", "timeuuid", { order: "DESC" })
thread_id: string;
@Type(() => Number)
@Column("last_activity", "number")
last_activity: number;
}
export type MessageUserInboxRefReversedPrimaryKey = Pick<
MessageUserInboxRefReversed,
"company_id" | "user_id" | "thread_id"
>;
export function getInstance(ref: MessageUserInboxRefReversed): MessageUserInboxRefReversed {
return merge(new MessageUserInboxRefReversed(), ref);
}
twake/backend/node/src/services/messages/entities/message-user-inbox-refs.ts
-43
View File
@@ -1,43 +0,0 @@
import { Type } from "class-transformer";
import { merge } from "lodash";
import { Column, Entity } from "../../../core/platform/services/database/services/orm/decorators";
export const TYPE = "message_user_inbox_refs";
@Entity(TYPE, {
primaryKey: [["company_id"], "user_id", "last_activity", "thread_id"],
type: TYPE,
})
export class MessageUserInboxRef {
@Type(() => String)
@Column("company_id", "uuid")
company_id: string;
@Type(() => String)
@Column("user_id", "string")
user_id: string;
@Type(() => Number)
@Column("last_activity", "number", { order: "DESC" })
last_activity: number;
@Type(() => String)
@Column("thread_id", "timeuuid", { order: "DESC" })
thread_id: string;
@Type(() => String)
@Column("workspace_id", "string")
workspace_id: string;
@Type(() => String)
@Column("channel_id", "string")
channel_id: string;
}
export type MessageUserInboxRefPrimaryKey = Pick<
MessageUserInboxRef,
"company_id" | "user_id" | "last_activity" | "thread_id"
>;
export function getInstance(ref: MessageUserInboxRef): MessageUserInboxRef {
return merge(new MessageUserInboxRef(), ref);
}
twake/backend/node/src/services/messages/entities/message-user-marked_refs.ts
-51
View File
@@ -1,51 +0,0 @@
import { Type } from "class-transformer";
import { merge } from "lodash";
import { Column, Entity } from "../../../core/platform/services/database/services/orm/decorators";
export const TYPE = "message_user_marked_refs";
@Entity(TYPE, {
primaryKey: [["company_id"], "user_id", "bookmark_id", "id"],
type: TYPE,
})
export class MessageUserMarkedRef {
@Type(() => String)
@Column("company_id", "uuid")
company_id: string;
@Type(() => String)
@Column("user_id", "string")
user_id: string;
@Type(() => String)
@Column("bookmark_id", "string")
bookmark_id: string;
@Type(() => Number)
@Column("id", "timeuuid", { generator: "timeuuid", order: "DESC" })
id: number;
@Type(() => String)
@Column("workspace_id", "string")
workspace_id: string;
@Type(() => String)
@Column("channel_id", "string")
channel_id: string;
@Type(() => String)
@Column("thread_id", "timeuuid")
thread_id: string;
@Type(() => String)
@Column("message_id", "timeuuid")
message_id: string;
}
export type MessageUserMarkedRefPrimaryKey = Pick<
MessageUserMarkedRef,
"company_id" | "user_id" | "bookmark_id" | "id"
>;
export function getInstance(ref: MessageUserMarkedRef): MessageUserMarkedRef {
return merge(new MessageUserMarkedRef(), ref);
}
twake/backend/node/src/services/messages/entities/messages.search.ts
-59
View File
@@ -1,59 +0,0 @@
import { fileIsMedia } from "../../../services/files/utils";
import { expandFileNameForSearch } from "./message-files.search";
import { Message } from "./messages";
export default {
index: "messages",
source: (entity: Message) => {
const links = (
(entity.text || "").match(
/(https?:\/\/(?:www\.|(?!www))[^\s\.]+\.[^\s]{2,}|www\.[^\s]+\.[^\s]{2,})/gi,
) || []
).join(" ");
const source: any = {
created_at: entity.created_at,
text: entity.text || "",
has_files: (entity.files || []).length > 0,
has_medias: (entity.files || []).filter(f => fileIsMedia(f)).length > 0,
attachments_names: (entity.files || [])
.map(file => expandFileNameForSearch(file.metadata.name))
.join(" "),
links:
links +
" " +
links.replace(/https?:\/\//gm, "") +
" " +
links.replace(/[^A-Z-a-z0-9]/gm, " "),
};
if (entity.cache) {
return {
company_id: entity.cache?.company_id,
workspace_id: entity.cache?.workspace_id,
channel_id: entity.cache?.channel_id,
user_id: entity.user_id,
...source,
};
}
return source;
},
mongoMapping: {
text: {
text: "text",
},
},
esMapping: {
properties: {
text: { type: "text" },
attachments_names: { type: "text", index_prefixes: { min_chars: 1 } },
links: { type: "text", index_prefixes: { min_chars: 1 } },
user_id: { type: "keyword" },
company_id: { type: "keyword" },
workspace_id: { type: "keyword" },
channel_id: { type: "keyword" },
has_files: { type: "boolean" },
has_medias: { type: "boolean" },
created_at: { type: "number" },
},
},
};
twake/backend/node/src/services/messages/entities/messages.ts
-161
View File
@@ -1,161 +0,0 @@
import { Type } from "class-transformer";
import { merge } from "lodash";
import { Column, Entity } from "../../../core/platform/services/database/services/orm/decorators";
import { Block } from "../blocks-types";
import { UserObject } from "../../user/web/types";
import { MessageFile } from "./message-files";
import Application from "../../applications/entities/application";
import search from "./messages.search";
export const TYPE = "messages";
@Entity(TYPE, {
primaryKey: [["thread_id"], "id"],
type: TYPE,
search,
})
export class Message {
@Type(() => String)
@Column("thread_id", "timeuuid", { order: "DESC" })
thread_id: string;
@Type(() => String)
@Column("id", "timeuuid", { generator: "timeuuid", order: "DESC" })
id: string;
@Type(() => String) //Not in database (obviousl y because it is ephemeral)
ephemeral: EphemeralMessage | null; //Used for non-persisted messages (like interractive messages)
@Type(() => String)
@Column("type", "encoded_string")
type:
| "message" // Classic message
| "event"; // Hidden system events
@Column("subtype", "encoded_string")
subtype:
| null
| "application" //Message from a connector
| "deleted" //Message deleted by user
| "system"; // Message from system (channel activity)
@Type(() => Number)
@Column("created_at", "number")
created_at: number;
@Column("updated_at", "number", { onUpsert: _ => new Date().getTime() })
updated_at: number;
@Type(() => String)
@Column("user_id", "uuid")
user_id: string;
@Column("application_id", "encoded_string")
application_id: null | string;
@Type(() => String)
@Column("text", "encoded_string")
text: string;
@Column("blocks", "encoded_json")
blocks: Block[];
@Column("files", "encoded_json")
files: null | Partial<MessageFile>[];
@Column("context", "encoded_json")
context: any;
@Column("edited", "encoded_json")
edited: null | MessageEdited;
@Column("pinned_info", "encoded_json")
pinned_info: null | MessagePinnedInfo;
@Column("quote_message", "encoded_json")
quote_message:
| null
| (Partial<MessageWithUsers> & {
id: string;
thread_id: string;
channel_id: string | null;
workspace_id: string | null;
company_id: string | null;
});
@Column("reactions", "encoded_json")
reactions: null | MessageReaction[];
@Column("bookmarks", "encoded_json")
bookmarks: null | MessageBookmarks[];
@Column("override", "encoded_json")
override: null | MessageOverride;
@Column("cache", "encoded_json")
cache: null | {
company_id: string;
workspace_id: string;
channel_id: string;
};
@Column("links", "encoded_json")
links: null | MessageLinks[];
@Column("status", "encoded_json")
status: null | MessageDeliveryStatus;
}
export type MessageReaction = { count: number; name: string; users: string[] };
export type MessageOverride = { title?: string; picture?: string };
export type MessagePinnedInfo = { pinned_at: number; pinned_by: string };
export type MessageEdited = { edited_at: number };
export type EphemeralMessage = {
id: string; //Identifier of the ephemeral message
version: string; //Version of ephemeral message (to update the view)
recipient: string; //User that will see this ephemeral message
recipient_context_id: string; //Recipient current view/tab/window to send the message to
};
export type MessageBookmarks = {
user_id: string;
bookmark_id: string;
created_at: number;
};
export type MessagePrimaryKey = Pick<Message, "thread_id" | "id">;
export function getInstance(message: Partial<Message>): Message {
return merge(new Message(), {
id: undefined,
ephemeral: null,
type: "message",
created_at: new Date().getTime(),
application_id: null,
text: "",
blocks: [],
...message,
});
}
export type MessageWithUsers = Message & {
users?: UserObject[];
application?: Partial<Application>;
};
export type MessageLinks = {
title: string;
description: string | null;
domain: string;
img: string | null;
favicon: string | null;
img_width: number | null;
img_height: number | null;
url: string;
};
export type MessageDeliveryStatus = "sent" | "delivered" | "read";
twake/backend/node/src/services/messages/entities/threads.ts
-51
View File
@@ -1,51 +0,0 @@
import { Type } from "class-transformer";
import { merge } from "lodash";
import { Column, Entity } from "../../../core/platform/services/database/services/orm/decorators";
export const TYPE = "threads";
@Entity(TYPE, {
primaryKey: [["id"]],
type: TYPE,
})
export class Thread {
@Type(() => String)
@Column("id", "timeuuid", { order: "DESC" })
id: string;
@Type(() => String)
@Column("created_by", "string")
created_by: string;
@Type(() => Number)
@Column("created_at", "number")
created_at: number;
@Column("updated_at", "number", { onUpsert: _ => new Date().getTime() })
updated_at: number;
@Type(() => Number)
@Column("last_activity", "number")
last_activity: number;
@Type(() => Number)
@Column("answers", "number")
answers: number;
@Column("participants", "encoded_json")
participants: ParticipantObject[];
}
export type ParticipantObject = {
type: "user" | "channel";
created_at: number;
created_by: string;
id: string;
company_id: string;
workspace_id?: string;
};
export type ThreadPrimaryKey = Pick<Thread, "id">;
export function getInstance(thread: Thread): Thread {
return merge(new Thread(), thread);
}
twake/backend/node/src/services/messages/entities/user-message-bookmarks.ts
-37
View File
@@ -1,37 +0,0 @@
import { Type } from "class-transformer";
import { merge } from "lodash";
import { Column, Entity } from "../../../core/platform/services/database/services/orm/decorators";
export const TYPE = "user_message_bookmarks";
@Entity(TYPE, {
primaryKey: [["company_id"], "user_id", "id"],
type: TYPE,
})
export class UserMessageBookmark {
@Type(() => String)
@Column("company_id", "uuid")
company_id: string;
@Type(() => String)
@Column("user_id", "uuid")
user_id: string;
@Type(() => String)
@Column("id", "uuid", { generator: "uuid" })
id: string;
@Type(() => String)
@Column("name", "encoded_string")
name = "";
}
export type UserMessageBookmarkPrimaryKey = Pick<
UserMessageBookmark,
"company_id" | "user_id" | "id"
>;
export function getInstance(
bookmark: Pick<UserMessageBookmark, "company_id" | "user_id" | "name">,
): UserMessageBookmark {
return merge(new UserMessageBookmark(), bookmark);
}
twake/backend/node/src/services/messages/index.ts
-23
View File
@@ -1,23 +0,0 @@
import { Prefix, TwakeService } from "../../core/platform/framework";
import WebServerAPI from "../../core/platform/services/webserver/provider";
import web from "./web";
@Prefix("/internal/services/messages/v1")
export default class MessageService extends TwakeService<undefined> {
version = "1";
name = "messages";
public async doInit(): Promise<this> {
const fastify = this.context.getProvider<WebServerAPI>("webserver").getServer();
fastify.register((instance, _opts, next) => {
web(instance, { prefix: this.prefix });
next();
});
return this;
}
// TODO: remove
api(): undefined {
return undefined;
}
}
twake/backend/node/src/services/messages/services/engine/index.ts
-112
View File
@@ -1,112 +0,0 @@
import { localEventBus } from "../../../../core/platform/framework/event-bus";
import { Initializable } from "../../../../core/platform/framework";
import { MessageFileDownloadEvent, MessageLocalEvent } from "../../types";
import { ChannelViewProcessor } from "./processors/channel-view";
import { ChannelMarkedViewProcessor } from "./processors/channel-marked";
import { UserMarkedViewProcessor } from "./processors/user-marked";
import { UserInboxViewProcessor } from "./processors/user-inbox";
import { FilesViewProcessor } from "./processors/files";
import Repository from "../../../../core/platform/services/database/services/orm/repository/repository";
import { Thread } from "../../entities/threads";
import { ChannelSystemActivityMessageProcessor } from "./processors/system-activity-message";
import { MessageToNotificationsProcessor } from "./processors/message-to-notifications";
import { ResourceEventsPayload } from "../../../../utils/types";
import _ from "lodash";
import { StatisticsMessageProcessor } from "../../../statistics/pubsub/messages";
import { MessageToHooksProcessor } from "./processors/message-to-hooks";
import gr from "../../../global-resolver";
import { MessageLinksPreviewFinishedProcessor } from "./processors/links";
import { Message } from "../../entities/messages";
import { ExecutionContext } from "../../../../core/platform/framework/api/crud-service";
export class MessagesEngine implements Initializable {
private channelViewProcessor: ChannelViewProcessor;
private channelMarkedViewProcessor: ChannelMarkedViewProcessor;
private userMarkedViewProcessor: UserMarkedViewProcessor;
private userInboxViewProcessor: UserInboxViewProcessor;
private filesViewProcessor: FilesViewProcessor;
private messageToNotifications: MessageToNotificationsProcessor;
private messageToHooks: MessageToHooksProcessor;
private threadRepository: Repository<Thread>;
private messageRepository: Repository<Message>;
constructor() {
this.channelViewProcessor = new ChannelViewProcessor();
this.channelMarkedViewProcessor = new ChannelMarkedViewProcessor();
this.userMarkedViewProcessor = new UserMarkedViewProcessor();
this.userInboxViewProcessor = new UserInboxViewProcessor();
this.filesViewProcessor = new FilesViewProcessor();
this.messageToNotifications = new MessageToNotificationsProcessor();
this.messageToHooks = new MessageToHooksProcessor();
}
async dispatchMessage(e: MessageLocalEvent, context?: ExecutionContext) {
const thread = await this.threadRepository.findOne(
{
id: e.resource.thread_id,
},
{},
context,
);
if (e.resource.ephemeral) {
await this.channelViewProcessor.process(thread || null, e);
return;
}
await this.channelViewProcessor.process(thread, e);
await this.channelMarkedViewProcessor.process(thread, e);
await this.userInboxViewProcessor.process(thread, e);
await this.userMarkedViewProcessor.process();
await this.filesViewProcessor.process(thread, e);
await this.messageToNotifications.process(thread, e);
await this.messageToHooks.process(thread, e);
if (e.created) {
for (const workspaceId of _.uniq(
thread.participants.filter(p => p.type == "channel").map(p => p.workspace_id),
)) {
localEventBus.publish<ResourceEventsPayload>("channel:message_sent", {
message: {
thread_id: e.resource.thread_id,
sender: e.resource.user_id,
workspace_id: workspaceId,
},
user: e.context.user,
});
}
}
}
async init(): Promise<this> {
this.threadRepository = await gr.database.getRepository<Thread>("threads", Thread);
this.messageRepository = await gr.database.getRepository<Message>("messages", Message);
await this.channelViewProcessor.init();
await this.channelMarkedViewProcessor.init();
await this.userInboxViewProcessor.init();
await this.userMarkedViewProcessor.init();
await this.filesViewProcessor.init();
gr.platformServices.messageQueue.processor.addHandler(
new ChannelSystemActivityMessageProcessor(),
);
gr.platformServices.messageQueue.processor.addHandler(new StatisticsMessageProcessor());
gr.platformServices.messageQueue.processor.addHandler(
new MessageLinksPreviewFinishedProcessor(this.messageRepository, this.threadRepository),
);
localEventBus.subscribe("message:saved", async (e: MessageLocalEvent) => {
this.dispatchMessage(e);
});
localEventBus.subscribe("message:download", async (e: MessageFileDownloadEvent) => {
if (e.user?.id) {
await this.filesViewProcessor.processDownloaded(e.user?.id, e.operation);
}
});
return this;
}
}
twake/backend/node/src/services/messages/services/engine/processors/channel-marked/index.ts
-47
View File
@@ -1,47 +0,0 @@
import { MessageLocalEvent } from "../../../../types";
import { Thread } from "../../../../entities/threads";
import Repository from "../../../../../../core/platform/services/database/services/orm/repository/repository";
import {
getInstance,
MessageChannelMarkedRef,
} from "../../../../entities/message-channel-marked-refs";
import gr from "../../../../../global-resolver";
import { ExecutionContext } from "../../../../../../core/platform/framework/api/crud-service";
export class ChannelMarkedViewProcessor {
repository: Repository<MessageChannelMarkedRef>;
async init() {
this.repository = await gr.database.getRepository<MessageChannelMarkedRef>(
"message_channel_marked_refs",
MessageChannelMarkedRef,
);
}
async process(
thread: Thread,
message: MessageLocalEvent,
context?: ExecutionContext,
): Promise<void> {
for (const participant of thread.participants.filter(p => p.type === "channel")) {
//Pinned messages
const pinRef = getInstance({
company_id: participant.company_id,
workspace_id: participant.workspace_id,
channel_id: participant.id,
thread_id: thread.id,
message_id: message.resource.id,
type: "pinned",
created_at: message.resource.pinned_info?.pinned_at || 0,
created_by: message.resource.pinned_info?.pinned_by || "",
});
if (message.resource.pinned_info) {
this.repository.save(pinRef, context);
} else if (!message.created) {
this.repository.remove(pinRef, context);
}
//TODO add realtime
}
}
}
twake/backend/node/src/services/messages/services/engine/processors/channel-view/index.ts
-143
View File
@@ -1,143 +0,0 @@
import { MessageLocalEvent } from "../../../../types";
import { ParticipantObject, Thread } from "../../../../entities/threads";
import Repository from "../../../../../../core/platform/services/database/services/orm/repository/repository";
import { getInstance, MessageChannelRef } from "../../../../entities/message-channel-refs";
import {
getInstance as getInstanceReversed,
MessageChannelRefReversed,
} from "../../../../entities/message-channel-refs-reversed";
import { ExecutionContext } from "../../../../../../core/platform/framework/api/crud-service";
import gr from "../../../../../global-resolver";
import { publishMessageInRealtime } from "../../../utils";
export class ChannelViewProcessor {
repository: Repository<MessageChannelRef>;
repositoryReversed: Repository<MessageChannelRefReversed>;
async init() {
this.repository = await gr.database.getRepository<MessageChannelRef>(
"message_channel_refs",
MessageChannelRef,
);
this.repositoryReversed = await gr.database.getRepository<MessageChannelRefReversed>(
"message_channel_refs_reversed",
MessageChannelRefReversed,
);
}
async process(
thread: Thread,
message: MessageLocalEvent,
context?: ExecutionContext,
): Promise<void> {
let participants: ParticipantObject[] = thread?.participants || [];
if (participants.length === 0) {
participants = message.context.channel
? [
{
type: "channel",
id: message.context.channel.id,
workspace_id: message.context.workspace.id,
company_id: message.context.company.id,
} as ParticipantObject,
]
: [];
}
for (const participant of (participants || []).filter(p => p.type === "channel")) {
if (!message.resource.ephemeral) {
//Publish message in corresponding channel
if (message.created) {
const pkPrefix = {
company_id: participant.company_id,
workspace_id: participant.workspace_id,
channel_id: participant.id,
};
//If a pointer exists it means the message already exists (it was probably moved and so we need to keep everything in place)
const existingPointer = await this.repository.findOne(
{
...pkPrefix,
message_id: message.resource.id,
},
{},
context,
);
if (!existingPointer) {
await this.repository.save(
getInstance({
...pkPrefix,
thread_id: thread.id,
message_id: message.resource.id,
}),
context,
);
const reversed = await this.repositoryReversed.findOne(
{
...pkPrefix,
thread_id: thread.id,
},
{},
context,
);
if (reversed) {
const existingThreadRef = await this.repository.findOne(
{
...pkPrefix,
message_id: reversed.message_id,
},
{},
context,
);
if (
existingThreadRef &&
`${existingThreadRef.thread_id}` === `${message.resource.thread_id}`
) {
reversed.message_id = message.resource.id;
await this.repositoryReversed.save(reversed, context);
await this.repository.remove(existingThreadRef, context);
}
} else {
await this.repositoryReversed.save(
getInstanceReversed({
...pkPrefix,
thread_id: thread.id,
message_id: message.resource.id,
}),
context,
);
}
}
//Message moved to it own thread
if (existingPointer && message.resource.thread_id === message.resource.id) {
await this.repository.save(
getInstance({
...pkPrefix,
thread_id: thread.id,
message_id: message.resource.id,
}),
context,
);
await this.repositoryReversed.save(
getInstanceReversed({
...pkPrefix,
thread_id: thread.id,
message_id: message.resource.id,
}),
context,
);
}
}
}
//Publish message in realtime
publishMessageInRealtime(message, participant);
}
}
}
twake/backend/node/src/services/messages/services/engine/processors/files/index.ts
-144
View File
@@ -1,144 +0,0 @@
import { MessageLocalEvent } from "../../../../types";
import { Thread } from "../../../../entities/threads";
import Repository from "../../../../../../core/platform/services/database/services/orm/repository/repository";
import { getInstance, MessageFileRef } from "../../../../entities/message-file-refs";
import gr from "../../../../../global-resolver";
import uuid from "node-uuid";
import {
ExecutionContext,
Pagination,
} from "../../../../../../core/platform/framework/api/crud-service";
import { MessageFile } from "../../../../entities/message-files";
import { fileIsMedia } from "../../../../../files/utils";
export class FilesViewProcessor {
repository: Repository<MessageFileRef>;
messageFileRepository: Repository<MessageFile>;
async init() {
this.repository = await gr.database.getRepository<MessageFileRef>(
"message_file_refs",
MessageFileRef,
);
this.messageFileRepository = await gr.database.getRepository<MessageFile>(
"message_files",
MessageFile,
);
}
async processDownloaded(
userId: string,
operation: { message_id: string; thread_id: string; message_file_id: string },
context?: ExecutionContext,
): Promise<void> {
const messageFile = await this.messageFileRepository.findOne(
{
message_id: operation.message_id,
id: operation.message_file_id,
},
{},
context,
);
const thread = await gr.services.messages.threads.get(
{
id: operation.thread_id,
},
context,
);
const refs = await this.repository.find(
{ target_type: "user_download", target_id: userId, company_id: messageFile.company_id },
{
pagination: new Pagination("", "100"),
},
context,
);
if (refs.getEntities().some(r => r.message_file_id === messageFile.id)) {
//File already in the recent list
return;
}
//For the user we add it as downloaded by user
const fileRef = getInstance({
target_type: "user_download",
target_id: userId,
id: uuid.v1(),
created_at: new Date().getTime(),
file_id: messageFile?.metadata?.external_id,
company_id: messageFile.company_id,
workspace_id: thread.participants?.find(p => p.type === "channel")?.workspace_id || "",
channel_id: thread.participants?.find(p => p.type === "channel")?.id || "",
thread_id: messageFile.thread_id,
message_id: messageFile.message_id,
message_file_id: messageFile.id,
});
this.repository.save(fileRef, context);
}
async process(
thread: Thread,
message: MessageLocalEvent,
context?: ExecutionContext,
): Promise<void> {
if (!message.resource.ephemeral) {
for (const file of message.resource.files || []) {
//For each channel, we add the media
for (const participant of (thread.participants || []).filter(p => p.type === "channel")) {
const fileRef = getInstance({
target_type: "channel",
target_id: participant.id,
id: uuid.v1(),
created_at: message.resource.created_at,
workspace_id: participant.workspace_id,
channel_id: participant.id,
thread_id: thread.id,
message_id: message.resource.id,
message_file_id: file.id,
company_id: file.company_id,
file_id: file.metadata.external_id,
});
this.repository.save(fileRef, context);
const isMedia = fileIsMedia(file);
for (const type of [
"channel",
isMedia ? "channel_media" : "channel_file",
] as MessageFileRef["target_type"][]) {
const fileRef = getInstance({
target_type: type,
target_id: participant.id,
id: uuid.v1(),
created_at: message.resource.created_at,
workspace_id: participant.workspace_id,
channel_id: participant.id,
thread_id: thread.id,
message_id: message.resource.id,
message_file_id: file.id,
company_id: file.company_id,
file_id: file.metadata.external_id,
});
this.repository.save(fileRef);
}
}
//For the user we add it as uploaded by user
const fileRef = getInstance({
target_type: "user_upload",
target_id: message.resource.user_id,
id: uuid.v1(),
created_at: message.resource.created_at,
workspace_id:
(thread.participants || []).filter(p => p.type === "channel")[0]?.workspace_id || "",
channel_id: (thread.participants || []).filter(p => p.type === "channel")[0]?.id || "",
thread_id: thread.id,
message_id: message.resource.id,
message_file_id: file.id,
company_id: file.company_id,
file_id: file.metadata.external_id,
});
this.repository.save(fileRef, context);
}
}
}
}
twake/backend/node/src/services/messages/services/engine/processors/links/index.ts
-85
View File
@@ -1,85 +0,0 @@
import { logger } from "../../../../../../core/platform/framework";
import { MessageQueueHandler } from "../../../../../../core/platform/services/message-queue/api";
import { Message } from "../../../../entities/messages";
import Repository from "../../../../../../core/platform/services/database/services/orm/repository/repository";
import { LinkPreviewMessageQueueCallback } from "../../../../../previews/types";
import { Thread } from "../../../../entities/threads";
import { publishMessageInRealtime } from "../../../utils";
import { ExecutionContext } from "../../../../../../core/platform/framework/api/crud-service";
export class MessageLinksPreviewFinishedProcessor
implements MessageQueueHandler<LinkPreviewMessageQueueCallback, string>
{
constructor(
private MessageRepository: Repository<Message>,
private ThreadRepository: Repository<Thread>,
) {}
readonly name = "MessageLinksPreviewFinishedProcessor";
readonly topics = {
in: "services:preview:links:callback",
};
readonly options = {
unique: true,
ack: true,
};
init?(): Promise<this> {
throw new Error("Method not implemented.");
}
validate(message: LinkPreviewMessageQueueCallback): boolean {
return !!(message && message.previews && message.previews.length);
}
async process(
localMessage: LinkPreviewMessageQueueCallback,
context?: ExecutionContext,
): Promise<string> {
logger.info(
`${this.name} - updating message links with generated previews: ${localMessage.previews.length}`,
);
const entity = await this.MessageRepository.findOne(
{
thread_id: localMessage.message.resource.thread_id,
id: localMessage.message.resource.id,
},
{},
context,
);
if (!entity) {
logger.error(`${this.name} - message not found`);
return "";
}
entity.links = localMessage.previews;
await this.MessageRepository.save(entity, context);
const thread: Thread = await this.ThreadRepository.findOne(
{
id: localMessage.message.resource.thread_id,
},
{},
context,
);
if (!thread) {
logger.error(`${this.name} - thread not found`);
return "";
}
const updatedMessage = {
...localMessage.message,
resource: entity,
};
for (const participant of thread.participants.filter(p => p.type === "channel")) {
publishMessageInRealtime(updatedMessage, participant);
}
return "done";
}
}
twake/backend/node/src/services/messages/services/engine/processors/message-to-hooks/index.ts
-107
View File
@@ -1,107 +0,0 @@
import { MessageHook, MessageLocalEvent } from "../../../../types";
import { ParticipantObject, Thread } from "../../../../entities/threads";
import { logger } from "../../../../../../core/platform/framework";
import { Channel } from "../../../../../channels/entities";
import {
ExecutionContext,
Pagination,
} from "../../../../../../core/platform/framework/api/crud-service";
import { Message } from "../../../../entities/messages";
import gr from "../../../../../global-resolver";
export class MessageToHooksProcessor {
private name = "MessageToHooksProcessor";
async init() {
return this;
}
async process(
thread: Thread,
message: MessageLocalEvent,
context?: ExecutionContext,
): Promise<void> {
logger.debug(`${this.name} - Share message with channel microservice for hooks`);
if (message.resource.ephemeral) {
logger.debug(`${this.name} - Cancel because message is ephemeral`);
return;
}
try {
const messageResource = message.resource;
for (const participant of thread.participants) {
if (participant.type !== "channel") {
continue;
}
const channel: Channel = await gr.services.channels.channels.get(
{
id: participant.id,
company_id: participant.company_id,
workspace_id: participant.workspace_id,
},
context,
);
if (!channel) {
continue;
}
for (const appId of channel.connectors) {
if (message.resource.application_id !== appId) {
//Publish hook
await gr.platformServices.messageQueue.publish<MessageHook>(
"application:hook:message",
{
data: {
type: "message",
application_id: appId,
company_id: participant.company_id,
channel: channel,
thread: thread,
message: messageResource,
},
},
);
}
}
}
} catch (err) {
logger.warn({ err }, `${this.name} - Error while publishing`);
}
}
async isLastActivityMessageDeleted(
participant: ParticipantObject,
messageResource: Message,
message: MessageLocalEvent,
): Promise<boolean> {
if (participant.company_id && participant.workspace_id && participant.id) {
const list = await gr.services.messages.views.listChannel(
new Pagination("", "1"),
{
include_users: false,
replies_per_thread: 1,
emojis: false,
},
{
channel: {
company_id: participant.company_id,
workspace_id: participant.workspace_id,
id: participant.id,
},
user: {
id: messageResource.user_id,
},
},
);
return (
list.getEntities().pop()?.id === message.resource.id &&
messageResource.subtype === "deleted"
);
}
}
}
twake/backend/node/src/services/messages/services/engine/processors/message-to-notifications/index.ts
-179
View File
@@ -1,179 +0,0 @@
import { MessageLocalEvent, MessageNotification } from "../../../../types";
import { ParticipantObject, Thread } from "../../../../entities/threads";
import { logger } from "../../../../../../core/platform/framework";
import { Channel } from "../../../../../channels/entities";
import { isDirectChannel } from "../../../../../channels/utils";
import { ChannelActivityNotification } from "../../../../../channels/types";
import { getMentions } from "../../../utils";
import {
ExecutionContext,
Pagination,
} from "../../../../../../core/platform/framework/api/crud-service";
import { Message } from "../../../../../../services/messages/entities/messages";
import gr from "../../../../../global-resolver";
export class MessageToNotificationsProcessor {
private name = "MessageToNotificationsProcessor";
async init() {
//
}
async process(
thread: Thread,
message: MessageLocalEvent,
context?: ExecutionContext,
): Promise<void> {
logger.debug(`${this.name} - Share message with notification microservice`);
if (message.resource.ephemeral) {
logger.debug(`${this.name} - Cancel because message is ephemeral`);
return;
}
try {
const messageResource = message.resource;
for (const participant of thread.participants) {
if (participant.type !== "channel") {
continue;
}
const channel: Channel = await gr.services.channels.channels.get(
{
id: participant.id,
company_id: participant.company_id,
workspace_id: participant.workspace_id,
},
context,
);
if (!channel) {
continue;
}
const company = await gr.services.companies.getCompany({ id: participant.company_id });
const companyName = company?.name || "";
let workspaceName = "";
let senderName = "Twake";
if (messageResource.user_id) {
const user = await gr.services.users.get({ id: messageResource.user_id });
senderName =
`${user?.first_name || ""} ${user?.last_name || ""}`.trim() ||
`@${user?.username_canonical}`;
}
let title = "";
let text = (messageResource.text || "").substr(0, 255);
const body = messageResource.text; // we need original body just in case text gets updated further on
if (isDirectChannel(channel)) {
title = `${senderName} in ${companyName}`;
} else {
const workspace =
(await gr.services.companies.getCompany({ id: participant.workspace_id })) || null;
workspaceName = workspace?.name || workspaceName;
title = `${channel.name} in ${companyName} • ${workspaceName}`;
text = `${senderName}: ${text}`;
}
const mentions = await getMentions(messageResource, async (username: string) => {
return await gr.services.users.getByUsername(username);
});
const messageEvent: MessageNotification = {
company_id: participant.company_id,
workspace_id: participant.workspace_id || "direct",
channel_id: participant.id,
thread_id: messageResource.thread_id,
id: messageResource.id,
sender: messageResource.user_id,
creation_date: messageResource.created_at,
mentions,
sender_name: senderName,
channel_name: channel.name,
company_name: companyName,
workspace_name: workspaceName,
title: title,
text: text,
};
const channelEvent: ChannelActivityNotification = {
company_id: participant.company_id,
workspace_id: participant.workspace_id || "direct",
channel_id: participant.id,
date: messageResource.created_at,
sender: messageResource.user_id,
sender_name: senderName,
title: title,
text: text,
body: body,
};
if (messageResource.type === "message" && messageResource.subtype !== "system") {
logger.info(
`${this.name} - Forward message ${messageResource.id} to channel:activity and message:created / message:updated`,
);
if (
message.created ||
(await this.isLastActivityMessageDeleted(participant, messageResource, message))
) {
await gr.platformServices.messageQueue.publish<ChannelActivityNotification>(
"channel:activity",
{
data: channelEvent,
},
);
}
await gr.platformServices.messageQueue.publish<MessageNotification>(
message.created ? "message:created" : "message:updated",
{
data: messageEvent,
},
);
} else {
logger.debug(`${this.name} - Cancel because this is system message`);
}
}
} catch (err) {
logger.warn({ err }, `${this.name} - Error while publishing`);
}
}
async isLastActivityMessageDeleted(
participant: ParticipantObject,
messageResource: Message,
message: MessageLocalEvent,
): Promise<boolean> {
if (participant.company_id && participant.workspace_id && participant.id) {
const list = await gr.services.messages.views.listChannel(
new Pagination("", "1"),
{
include_users: false,
replies_per_thread: 1,
emojis: false,
},
{
channel: {
company_id: participant.company_id,
workspace_id: participant.workspace_id,
id: participant.id,
},
user: {
id: messageResource.user_id,
},
},
);
return (
list.getEntities().pop()?.id === message.resource.id &&
messageResource.subtype === "deleted"
);
}
}
}
twake/backend/node/src/services/messages/services/engine/processors/system-activity-message/index.ts
-66
View File
@@ -1,66 +0,0 @@
import { MessageQueueHandler } from "../../../../../../core/platform/services/message-queue/api";
import { ActivityPublishedType } from "../../../../../channels/services/channel/activities/types";
import { ParticipantObject } from "../../../../entities/threads";
import { getInstance } from "../../../../entities/messages";
import { logger } from "../../../../../../core/platform/framework";
import gr from "../../../../../global-resolver";
export class ChannelSystemActivityMessageProcessor
implements MessageQueueHandler<ActivityPublishedType, void>
{
readonly topics = {
in: "channel:activity_message",
};
readonly options = {
unique: true,
ack: true,
};
readonly name = "Channel::ChannelSystemActivityMessageProcessor";
validate(message: ActivityPublishedType): boolean {
return !!(message && message.channel_id && message.company_id && message.workspace_id);
}
async process(event: ActivityPublishedType): Promise<void> {
logger.info(
`${this.name} Publish system message to message service in channel ${event.channel_id}`,
);
const message = getInstance({
type: "message",
subtype: "system",
context: {
type: "activity",
activity: event.activity,
},
});
const participants: Pick<ParticipantObject, "type" | "id" | "workspace_id" | "company_id">[] = [
{
type: "channel",
id: event.channel_id,
workspace_id: event.workspace_id,
company_id: event.company_id,
},
];
gr.services.messages.threads.save(
{
id: undefined,
participants,
},
{
message,
},
{
user: {
id: null,
server_request: true,
},
company: { id: event.company_id },
},
);
}
}
twake/backend/node/src/services/messages/services/engine/processors/user-inbox/index.ts
-93
View File
@@ -1,93 +0,0 @@
import { MessageLocalEvent } from "../../../../types";
import { Thread } from "../../../../entities/threads";
import Repository from "../../../../../../core/platform/services/database/services/orm/repository/repository";
import {
getInstance as getInboxRefInstance,
MessageUserInboxRef,
} from "../../../../entities/message-user-inbox-refs";
import {
getInstance as getInboxThreadInstance,
MessageUserInboxRefReversed,
} from "../../../../entities/message-user-inbox-refs-reversed";
import { ExecutionContext } from "../../../../../../core/platform/framework/api/crud-service";
import gr from "../../../../../global-resolver";
import { publishMessageInRealtime } from "../../../utils";
export class UserInboxViewProcessor {
repositoryRef: Repository<MessageUserInboxRef>;
repositoryReversed: Repository<MessageUserInboxRefReversed>;
async init() {
this.repositoryRef = await gr.database.getRepository<MessageUserInboxRef>(
"message_user_inbox_refs",
MessageUserInboxRef,
);
this.repositoryReversed = await gr.database.getRepository<MessageUserInboxRefReversed>(
"message_user_inbox_refs_reversed",
MessageUserInboxRefReversed,
);
}
async process(
thread: Thread,
message: MessageLocalEvent,
context?: ExecutionContext,
): Promise<void> {
for (const userParticipant of thread.participants.filter(p => p.type === "user")) {
for (const channelParticipant of thread.participants.filter(p => p.type === "channel")) {
//Publish message in corresponding channel
if (!userParticipant.id) {
continue;
}
if (message.created) {
const commonPk = {
company_id: channelParticipant.company_id,
user_id: userParticipant.id,
thread_id: thread.id,
};
let threadActivityReversed = await this.repositoryReversed.findOne(commonPk, {}, context);
let currentRef: MessageUserInboxRef = null;
if (threadActivityReversed) {
currentRef = await this.repositoryRef.findOne(
{
...commonPk,
last_activity: threadActivityReversed.last_activity,
},
{},
context,
);
} else {
threadActivityReversed = getInboxThreadInstance({
...commonPk,
last_activity: 0,
});
}
const ref = getInboxRefInstance({
...commonPk,
workspace_id: channelParticipant.workspace_id,
channel_id: channelParticipant.id,
last_activity: message.resource.created_at,
});
if (currentRef) {
await this.repositoryReversed.remove(currentRef, context);
}
await this.repositoryRef.save(ref, context);
threadActivityReversed.last_activity = message.resource.created_at;
await this.repositoryReversed.save(threadActivityReversed, context);
}
//Publish message in realtime
//TODO send a thread object instead of a message object
publishMessageInRealtime(message, channelParticipant);
}
}
}
}
twake/backend/node/src/services/messages/services/engine/processors/user-marked/index.ts
-18
View File
@@ -1,18 +0,0 @@
import Repository from "../../../../../../core/platform/services/database/services/orm/repository/repository";
import { MessageUserMarkedRef } from "../../../../entities/message-user-marked_refs";
import gr from "../../../../../global-resolver";
export class UserMarkedViewProcessor {
repository: Repository<MessageUserMarkedRef>;
async init(): Promise<void> {
this.repository = await gr.database.getRepository<MessageUserMarkedRef>(
"message_user_marked_refs",
MessageUserMarkedRef,
);
}
async process(): Promise<void> {
//
}
}
twake/backend/node/src/services/messages/services/messages-files.ts
-170
View File
@@ -1,170 +0,0 @@
import { Channel } from "../../../services/channels/entities";
import { fileIsMedia } from "../../../services/files/utils";
import { UserObject } from "../../../services/user/web/types";
import { formatUser } from "../../../utils/users";
import { Initializable } from "../../../core/platform/framework";
import Repository from "../../../core/platform/services/database/services/orm/repository/repository";
import gr from "../../../services/global-resolver";
import { MessageFileRef, TYPE as TYPERef } from "../entities/message-file-refs";
import { MessageFile, TYPE } from "../entities/message-files";
import { Message } from "../entities/messages";
export class MessagesFilesService implements Initializable {
version: "1";
msgFilesRepository: Repository<MessageFile>;
msgFilesRefRepository: Repository<MessageFileRef>;
async init(): Promise<this> {
this.msgFilesRepository = await gr.database.getRepository(TYPE, MessageFile);
this.msgFilesRefRepository = await gr.database.getRepository(TYPERef, MessageFileRef);
return this;
}
/**
* Delete a message file and test this files belongs to the right user
* @param message_id
* @param id
* @param user_id
* @returns
*/
async deleteMessageFile(message_id: string, id: string, user_id: string): Promise<MessageFile> {
const msgFile = await this.getMessageFile(message_id, id);
if (!msgFile) return null;
if (msgFile.message.user_id !== user_id) return null;
await this.msgFilesRepository.remove(msgFile);
for (const target_type of ["channel_media", "channel_file", "channel"]) {
const ref = await this.msgFilesRefRepository.findOne({
target_type,
company_id: msgFile.channel.company_id,
target_id: msgFile.channel.id,
id: msgFile.id,
});
if (ref) await this.msgFilesRefRepository.remove(ref);
}
return msgFile;
}
/**
* Get a message file and returns more contextual data
* @param message_id
* @param id
* @returns
*/
async getMessageFile(
message_id: string,
id: string,
): Promise<
MessageFile & {
user: UserObject;
message: Message;
channel: Channel;
navigation: { next: Partial<MessageFile>; previous: Partial<MessageFile> };
}
> {
const msgFile = await this.msgFilesRepository.findOne({ message_id, id });
if (!msgFile) return null;
const message = await gr.services.messages.messages.get({
thread_id: msgFile.thread_id,
id: message_id,
});
const channel = await gr.services.channels.channels.get({
company_id: message.cache.company_id,
workspace_id: message.cache.workspace_id,
id: message.cache.channel_id,
});
const user = await formatUser(await gr.services.users.get({ id: message.user_id }));
const navigationPk = {
target_type: fileIsMedia(msgFile) ? "channel_media" : "channel_file",
company_id: channel.company_id,
target_id: channel.id,
};
const { previous, next } = await this.getMessageFileNavigation(navigationPk, msgFile.id);
return {
...msgFile,
user,
message,
channel,
navigation: {
next: next
? {
id: next.message_file_id,
message_id: next.message_id,
}
: null,
previous: previous
? {
id: previous.message_file_id,
message_id: previous.message_id,
}
: null,
},
};
}
/**
* Message file references are ordered with an id based on the time the file was uploaded
* We cannot get this specific ID directly from the message file right now
*/
private async getMessageFileNavigation(
navigationPk: { target_type: string; company_id: string; target_id: string },
id: string,
) {
const list = [
...(
await this.msgFilesRefRepository.find(navigationPk, {
pagination: {
page_token: null,
limitStr: "5",
reversed: false,
},
$gte: [["id", id]],
})
).getEntities(),
...(
await this.msgFilesRefRepository.find(navigationPk, {
pagination: {
page_token: null,
limitStr: "5",
reversed: true,
},
$lte: [["id", id]],
})
).getEntities(),
];
const offsetRef = list.find(a => a.message_file_id === id) || null;
const next = (
await this.msgFilesRefRepository.find(navigationPk, {
pagination: {
page_token: null,
limitStr: "2",
reversed: true,
},
$gte: [["id", offsetRef?.id || id]],
})
)
.getEntities()
.filter(a => a.message_file_id !== id)?.[0];
const previous = (
await this.msgFilesRefRepository.find(navigationPk, {
pagination: {
page_token: null,
limitStr: "2",
reversed: false,
},
$lte: [["id", offsetRef?.id || id]],
})
)
.getEntities()
.filter(a => a.message_file_id !== id)?.[0];
return { previous, next };
}
}
twake/backend/node/src/services/messages/services/messages-operations.ts
-301
View File
@@ -1,301 +0,0 @@
import { OperationType, SaveResult } from "../../../core/platform/framework/api/crud-service";
import { logger, TwakeContext } from "../../../core/platform/framework";
import { Message, TYPE as MessageTableName } from "../entities/messages";
import {
BookmarkOperation,
DeleteLinkOperation,
MessageFileDownloadEvent,
PinOperation,
ReactionOperation,
ThreadExecutionContext,
UpdateDeliveryStatusOperation,
} from "../types";
import emoji from "emoji-name-map";
import Repository from "../../../core/platform/services/database/services/orm/repository/repository";
import { ThreadMessagesService } from "./messages";
import gr from "../../global-resolver";
import { updateMessageReactions } from "../../../utils/messages";
import { localEventBus } from "../../../core/platform/framework/event-bus";
import { ReactionNotification } from "../../../services/notifications/types";
export class ThreadMessagesOperationsService {
constructor(private threadMessagesService: ThreadMessagesService) {}
repository: Repository<Message>;
async init(_context: TwakeContext): Promise<this> {
this.repository = await gr.database.getRepository<Message>(MessageTableName, Message);
return this;
}
async pin(
operation: PinOperation,
options: Record<string, unknown>,
context: ThreadExecutionContext,
): Promise<SaveResult<Message>> {
if (
!context?.user?.server_request &&
!gr.services.messages.threads.checkAccessToThread(context)
) {
logger.error(`Unable to write in thread ${context.thread.id}`);
throw Error("Can't edit this message.");
}
const message = await this.repository.findOne(
{
thread_id: context.thread.id,
id: operation.id,
},
{},
context,
);
if (!message) {
logger.error("This message doesn't exists");
throw Error("Can't edit this message.");
}
message.pinned_info = operation.pin
? {
pinned_by: context.user.id,
pinned_at: new Date().getTime(),
}
: null;
logger.info(
`Updated message ${operation.id} pin to ${JSON.stringify(message.pinned_info)} thread ${
message.thread_id
}`,
);
await this.repository.save(message, context);
this.threadMessagesService.onSaved(message, { created: false }, context);
return new SaveResult<Message>("message", message, OperationType.UPDATE);
}
async reaction(
operation: ReactionOperation,
options: Record<string, unknown>,
context: ThreadExecutionContext,
): Promise<SaveResult<Message>> {
if (
!context?.user?.server_request &&
!gr.services.messages.threads.checkAccessToThread(context)
) {
logger.error(`Unable to write in thread ${context.thread.id}`);
throw Error("Can't edit this message.");
}
const message = await this.repository.findOne(
{
thread_id: context.thread.id,
id: operation.id,
},
{},
context,
);
if (!message) {
logger.error("This message doesn't exists");
throw Error("Can't edit this message.");
}
//Update message reactions
updateMessageReactions(message, operation.reactions || [], context.user.id);
logger.info(
`Updated message ${operation.id} reactions to ${JSON.stringify(message.reactions)} thread ${
message.thread_id
}`,
);
await this.repository.save(message, context);
if ((operation.reactions || []).length === 1 && context.user.id !== message.user_id) {
await gr.platformServices.messageQueue.publish<ReactionNotification>(
"notification:reaction",
{
data: {
thread_id: context.thread.id,
company_id: message.cache?.company_id || context.company.id,
user_id: message.user_id,
message_id: message.id,
workspace_id: message.cache?.workspace_id,
channel_id: message.cache?.channel_id,
creation_date: new Date().getTime(),
reaction: emoji.get(operation.reactions[0]),
reaction_user_id: context.user.id,
},
},
);
}
this.threadMessagesService.onSaved(message, { created: false }, context);
return new SaveResult<Message>("message", message, OperationType.UPDATE);
}
async bookmark(
operation: BookmarkOperation,
options: Record<string, unknown>,
context: ThreadExecutionContext,
): Promise<SaveResult<Message>> {
const message = await this.repository.findOne(
{
thread_id: context.thread.id,
id: operation.id,
},
{},
context,
);
if (!message) {
logger.error("This message doesn't exists");
throw Error("Can't edit this message.");
}
//TODO add message to user bookmarks
message.bookmarks = message.bookmarks.filter(
b => !(b.user_id === context.user.id && b.bookmark_id === operation.bookmark_id),
);
if (operation.active) {
message.bookmarks.push({
user_id: context.user.id,
bookmark_id: operation.bookmark_id,
created_at: new Date().getTime(),
});
}
logger.info(
`Added bookmark to message ${operation.id} => ${JSON.stringify(
message.bookmarks,
)} to thread ${message.thread_id}`,
);
await this.repository.save(message, context);
this.threadMessagesService.onSaved(message, { created: false }, context);
return new SaveResult<Message>("message", message, OperationType.UPDATE);
}
async download(
operation: { id: string; thread_id: string; message_file_id: string },
options: Record<string, any>,
context: ThreadExecutionContext,
) {
//Register download action for reference
localEventBus.publish("message:download", {
user: context.user,
operation: {
message_id: operation.id,
thread_id: operation.thread_id,
message_file_id: operation.message_file_id,
},
} as MessageFileDownloadEvent);
}
/**
* Delete a link preview from a message
*
* @param {DeleteLinkOperation} operation - params of the operation
* @param {ThreadExecutionContext} context - Thread execution context
* @returns {Promise<SaveResult<Message>>} - Result of the operation
*/
async deleteLinkPreview(
operation: DeleteLinkOperation,
context: ThreadExecutionContext,
): Promise<SaveResult<Message>> {
if (
!context?.user?.server_request &&
!gr.services.messages.threads.checkAccessToThread(context)
) {
logger.error(`no access ${context.thread.id}`);
throw Error("can't remove link preview from message.");
}
const message = await this.repository.findOne(
{
thread_id: context.thread.id,
id: operation.message_id,
},
{},
context,
);
if (!message) {
logger.error("This message doesn't exists");
throw Error("Can't edit message links previews.");
}
if (context.user.id !== message.user_id) {
logger.error("You can't remove link preview from another user message.");
throw Error("Can't edit message links previews.");
}
message.links = message.links.filter(({ url }: { url: string }) => url !== operation.link);
await this.repository.save(message, context);
this.threadMessagesService.onSaved(message, { created: false }, context);
return new SaveResult<Message>("message", message, OperationType.UPDATE);
}
/**
* Update message delivery status operation
*
* @param {UpdateDeliveryStatusOperation} operation - the operation params
* @param {ThreadExecutionContext} context - Thread execution context
* @returns {Promise<SaveResult<Message>>} - Result of the operation
*/
async updateDeliveryStatus(
operation: UpdateDeliveryStatusOperation,
context: ThreadExecutionContext,
): Promise<SaveResult<Message>> {
if (
!context?.user?.server_request &&
!gr.services.messages.threads.checkAccessToThread(context)
) {
logger.error(`no access ${context.thread.id}`);
throw Error("can't update message delivery status.");
}
if (!operation.message_id || !operation.status) {
logger.error("Invalid operation");
throw Error("Invalid operation");
}
if (operation.status !== "delivered" && operation.status !== "read") {
logger.error("Invalid status");
throw Error("Invalid status");
}
const message = await this.repository.findOne(
{
thread_id: context.thread.id,
id: operation.message_id,
},
{},
context,
);
if (!message) {
logger.error("This message doesn't exists");
throw Error("Can't update message delivery status.");
}
if (operation.status === "delivered" && message.status === "delivered") {
logger.error("Message already delivered");
return;
}
if (operation.status === "read" && message.status === "read") {
logger.error("Message already read");
return;
}
if (operation.status === "delivered" && message.status === "read" && !operation.self_message) {
logger.error("Invalid operation");
return;
}
message.status = operation.status;
await this.repository.save(message, context);
this.threadMessagesService.onSaved(message, { created: false }, context);
return new SaveResult<Message>("message", message, OperationType.UPDATE);
}
}
twake/backend/node/src/services/messages/services/messages.ts
-1078
View File
File diff suppressed because it is too large Load Diff
twake/backend/node/src/services/messages/services/threads.ts
-309
View File
@@ -1,309 +0,0 @@
import {
DeleteResult,
ExecutionContext,
ListResult,
OperationType,
SaveResult,
} from "../../../core/platform/framework/api/crud-service";
import {
Initializable,
logger,
RealtimeSaved,
TwakeServiceProvider,
} from "../../../core/platform/framework";
import Repository from "../../../core/platform/services/database/services/orm/repository/repository";
import { ParticipantObject, Thread } from "../entities/threads";
import { CompanyExecutionContext, ThreadExecutionContext } from "../types";
import { Message, MessageWithUsers } from "../entities/messages";
import _ from "lodash";
import { extendExecutionContentWithChannel } from "../web/controllers";
import gr from "../../global-resolver";
import { ResourcePath } from "../../../core/platform/services/realtime/types";
import { getThreadMessagePath, getThreadMessageWebsocketRoom } from "../web/realtime";
export class ThreadsService implements TwakeServiceProvider, Initializable {
version: "1";
name: "ThreadsService";
repository: Repository<Thread>;
async init(): Promise<this> {
this.repository = await gr.database.getRepository<Thread>("threads", Thread);
return this;
}
/**
* Create a thread with its first message in it
* @param item
* @param options
* @param context
* @returns
*/
async save(
item: Pick<Thread, "id"> & {
participants: Pick<ParticipantObject, "id" | "type">[];
},
options: { participants?: ParticipantOperation; message?: Message } = {},
context?: CompanyExecutionContext,
): Promise<SaveResult<Thread>> {
if (item.id) {
//Update
const participantsOperation: ParticipantOperation = options.participants || {
add: [],
remove: [],
};
const thread = await this.repository.findOne({ id: item.id }, {}, context);
if (thread) {
// Add the created_by information
participantsOperation.add = (participantsOperation.add || []).map(p => {
return {
created_by: context.user.id,
created_at: new Date().getTime(),
...p,
};
}) as ParticipantObject[];
thread.participants = _.uniqBy(
_.differenceBy(
[...thread.participants, ...participantsOperation.add],
participantsOperation.remove || [],
p => p.id,
),
p => p.id,
) as ParticipantObject[];
await this.repository.save(thread, context);
//TODO ensure the thread is in all participants views (and removed from deleted participants)
return new SaveResult(
"thread",
await this.getWithMessage({ id: thread.id }),
OperationType.UPDATE,
);
} else {
//Thread to edit does not exists
if (!context.user?.server_request) {
throw new Error("ThreadService: Unable to edit inexistent thread");
}
}
}
//Creation of thread or server edition
if (!options.message) {
if (context.user?.server_request) {
logger.info(`${this.name} - Create empty thread by server itself`);
} else {
throw new Error("You must provide an initial message in the thread.");
}
}
//Enforce current user in the participants list and add the created_by information
const participants: ParticipantObject[] = [
context.user.application_id ? {} : { type: "user", id: context.user.id },
...item.participants,
].map(p => {
return {
created_by: context.user.id,
created_at: new Date().getTime(),
...p,
};
}) as ParticipantObject[];
const message: Message | null = options.message || null;
const thread = new Thread();
thread.created_at = new Date().getTime();
thread.last_activity = thread.created_at;
thread.answers = 0;
thread.created_by = context.user.id;
thread.participants = _.uniqBy(participants, p => p.id);
//If server request, we allow more
if (context.user?.server_request) {
thread.id = item.id;
thread.created_at = (item as Thread)?.created_at || thread.created_at;
}
if (message && message.ephemeral) {
//We should not save if this is an ephemeral message
} else {
await this.repository.save(thread, context);
}
if (message) {
message.status = "sent";
const result = await gr.services.messages.messages.save(
message,
{
threadInitialMessage: true,
},
extendExecutionContentWithChannel(
participants,
Object.assign(context, { thread: { id: thread.id, company_id: context.company.id } }),
),
);
try {
const channelMembersCount = await gr.services.channels.members.getUsersCount({
id: result.entity.cache.channel_id,
company_id: result.entity.cache.company_id,
workspace_id: result.entity.cache.workspace_id,
counter_type: "members",
});
if (channelMembersCount === 1) {
await gr.services.messages.messages.updateDeliveryStatus(
{
message_id: result.entity.id,
self_message: true,
status: "read",
thread_id: result.entity.thread_id,
},
{
...context,
thread: { id: thread.id },
},
);
await gr.services.channels.members.setChannelMemberReadSections(
{
start: result.entity.id,
end: result.entity.id,
},
{
...context,
channel_id: result.entity.cache.channel_id,
workspace_id: result.entity.cache.workspace_id,
},
);
}
} catch (err) {
logger.error("failed to update message delivery status");
}
}
return new SaveResult(
"thread",
await this.getWithMessage({ id: thread.id }),
OperationType.CREATE,
);
}
/**
* Add reply to thread: increase last_activity time and number of answers
* @param threadId
* @param increment
* @param context
*/
async addReply(threadId: string, increment: number = 1, context: ThreadExecutionContext) {
const thread = await this.repository.findOne({ id: threadId }, {}, context);
if (thread) {
thread.answers = Math.max(0, (thread.answers || 0) + increment);
if (increment > 0) {
thread.last_activity = new Date().getTime();
}
await this.repository.save(thread, context);
} else {
throw new Error("Try to add reply count to inexistent thread");
}
await gr.services.messages.messages.shareMessageInRealtime(
{
id: threadId,
thread_id: threadId,
},
{},
context,
);
}
/**
* Add reply to thread: increase last_activity time and number of answers
* @param threadId
*/
async setReplyCount(threadId: string, count: number, context: ExecutionContext) {
const thread = await this.repository.findOne({ id: threadId }, {}, context);
if (thread) {
thread.answers = count;
await this.repository.save(thread, context);
} else {
throw new Error("Try to add reply count to inexistent thread");
}
}
/**
* Check context is allowed to accesss a thread
* @param context
* @returns
*/
async checkAccessToThread(context: ThreadExecutionContext): Promise<boolean> {
if (context?.user?.server_request) {
return true;
}
logger.info(
`Check access to thread ${context.thread.id} in company ${context.company.id} for user ${context.user.id} and app ${context?.user?.application_id}`,
);
const thread = await this.repository.findOne(
{
id: context.thread.id,
},
{},
context,
);
if (!thread) {
logger.info("No such thread");
return false;
}
//User is participant of the thread directly
if (thread.participants.some(p => p.type === "user" && p.id === context.user.id)) {
return true;
}
//Check user is in one of the participant channels
//TODO: get the channel_member from channel micro_service
return false;
}
get(pk: Pick<Thread, "id">, context: ExecutionContext): Promise<Thread> {
return this.repository.findOne(pk, {}, context);
}
async getWithMessage(
pk: Pick<Thread, "id">,
context?: ExecutionContext,
): Promise<Thread & { message: Message }> {
const thread = await this.get(pk, context);
return {
...thread,
message: await gr.services.messages.messages.get(
{ id: pk.id, thread_id: pk.id },
context as ThreadExecutionContext,
),
};
}
async delete(pk: Pick<Thread, "id">, context?: ExecutionContext): Promise<DeleteResult<Thread>> {
const thread = await this.repository.findOne({ id: pk.id }, {}, context);
if (context.user.server_request) {
await this.repository.remove(thread, context);
}
return new DeleteResult("thread", thread, context.user.server_request && !!thread);
}
list(): Promise<ListResult<Thread>> {
throw new Error("CRUD method not used.");
}
}
type ParticipantOperation = {
add: Pick<ParticipantObject, "id" | "type">[];
remove: Pick<ParticipantObject, "id" | "type">[];
};
twake/backend/node/src/services/messages/services/user-bookmarks.ts
-110
View File
@@ -1,110 +0,0 @@
import {
DeleteResult,
ExecutionContext,
ListResult,
OperationType,
Pagination,
SaveResult,
} from "../../../core/platform/framework/api/crud-service";
import {
Initializable,
RealtimeDeleted,
RealtimeSaved,
TwakeServiceProvider,
} from "../../../core/platform/framework";
import Repository from "../../../core/platform/services/database/services/orm/repository/repository";
import { getInstance, UserMessageBookmark } from "../entities/user-message-bookmarks";
import { CompanyExecutionContext } from "../types";
import { ResourcePath } from "../../../core/platform/services/realtime/types";
import { getUserBookmarksWebsocketRoom } from "../web/realtime";
import gr from "../../global-resolver";
export class UserBookmarksService implements TwakeServiceProvider, Initializable {
version: "1";
repository: Repository<UserMessageBookmark>;
async init(): Promise<this> {
this.repository = await gr.database.getRepository<UserMessageBookmark>(
"user_message_bookmarks",
UserMessageBookmark,
);
return this;
}
get(
pk: Pick<UserMessageBookmark, "company_id" | "user_id" | "id">,
context: CompanyExecutionContext,
): Promise<UserMessageBookmark> {
return this.repository.findOne(pk, {}, context);
}
@RealtimeSaved<UserMessageBookmark>((bookmark, context) => [
{
room: ResourcePath.get(getUserBookmarksWebsocketRoom(context as CompanyExecutionContext)),
path: getUserBookmarksWebsocketRoom(context as CompanyExecutionContext) + "/" + bookmark.id,
},
])
async save(
{ company_id, user_id, id, name }: SaveParams,
context: CompanyExecutionContext,
): Promise<SaveResult<UserMessageBookmark>> {
//Disallow duplicates
const entities = (await this.list({ user_id, company_id }, context)).getEntities();
for (const entity of entities) {
if (name === entity.name) {
return new SaveResult<UserMessageBookmark>(
"user_message_bookmark",
entity,
OperationType.EXISTS,
);
}
}
const instance = getInstance({
company_id: company_id,
user_id: user_id,
name,
});
if (id) {
instance.id = id;
}
await this.repository.save(instance, context);
return new SaveResult<UserMessageBookmark>(
"user_message_bookmark",
instance,
id ? OperationType.UPDATE : OperationType.CREATE,
);
}
@RealtimeDeleted<UserMessageBookmark>((bookmark, context) => [
{
room: ResourcePath.get(getUserBookmarksWebsocketRoom(context as CompanyExecutionContext)),
path: getUserBookmarksWebsocketRoom(context as CompanyExecutionContext) + "/" + bookmark.id,
},
])
async delete(
pk: Pick<UserMessageBookmark, "company_id" | "user_id" | "id">,
context?: CompanyExecutionContext,
): Promise<DeleteResult<UserMessageBookmark>> {
const instance = await this.repository.findOne(pk, {}, context);
if (instance) await this.repository.remove(instance, context);
return new DeleteResult<UserMessageBookmark>("user_message_bookmark", instance, !!instance);
}
async list(
{ user_id, company_id, pagination }: ListParams,
context: ExecutionContext,
): Promise<ListResult<UserMessageBookmark>> {
return this.repository.find({ user_id, company_id }, { pagination }, context);
}
}
type SaveParams = { user_id: string; company_id: string; name: string; id: string; test?: string };
type ListParams = {
company_id: string;
user_id: string;
pagination?: Pagination;
};
twake/backend/node/src/services/messages/services/utils.ts
-104
View File
@@ -1,104 +0,0 @@
import { FindOptions } from "../../../core/platform/services/database/services/orm/repository/repository";
import {
CreateResult,
Pagination,
UpdateResult,
} from "../../../core/platform/framework/api/crud-service";
import { Message } from "../entities/messages";
import { MessageLocalEvent, SpecialMention, ThreadExecutionContext } from "../types";
import User from "../../../services/user/entities/user";
import { RealtimeEntityActionType } from "../../../core/platform/services/realtime/types";
import { getThreadMessagePath } from "../web/realtime";
import { ResourcePath } from "../../../core/platform/services/realtime/types";
import { RealtimeLocalBusEvent } from "../../../core/platform/services/realtime/types";
import { localEventBus } from "../../../core/platform/framework/event-bus";
import { ParticipantObject } from "../entities/threads";
export const buildMessageListPagination = (
pagination: Pagination,
messageIdKey: string,
): FindOptions => {
const offset = pagination.page_token;
pagination = { ...pagination, page_token: null };
return {
pagination,
...(!!offset && pagination.reversed && { $gte: [[messageIdKey, offset]] }),
...(!!offset && !pagination.reversed && { $lte: [[messageIdKey, offset]] }),
};
};
export const getMentions = async (
messageResource: Message,
findByUsername: (username: string) => Promise<User>,
) => {
const idsFromUsernames = [];
try {
const usersNoIdOutput = (messageResource.text || "").match(/( |^)@[a-zA-Z0-9-_.]+/gm);
const usernames = (usersNoIdOutput || []).map(u => (u || "").trim().split("@").pop());
for (const username of usernames) {
if (!"all|here|channel|everyone".split("|").includes(username)) {
const user = await findByUsername(username);
if (user) idsFromUsernames.push(user.id);
}
}
} catch (err) {
console.log(err);
}
const usersOutput = (messageResource.text || "").match(/@[^: ]+:([0-f-]{36})/gm);
const globalOutput = (messageResource.text || "").match(
/(^| )@(all|here|channel|everyone)([^a-z]|$)/gm,
);
return {
users: [
...(usersOutput || []).map(u => (u || "").trim().split(":").pop()),
...idsFromUsernames,
],
specials: (globalOutput || []).map(g => (g || "").trim().split("@").pop()) as SpecialMention[],
};
};
/**
* extracts the links from a message
*
* @param {Message} messageResource - The message to be parsed
* @returns {String} - links found in the message
*/
export const getLinks = (messageResource: Message): string[] => {
const cleanText = messageResource.text.replace(/\n+/gm, " ");
const links = (cleanText || "").match(/https?:\/\/[^ ]+/gm);
return links || [];
};
/**
* Publish a message to the realtime bus
*
* @param {MessageLocalEvent} message - The event to be published
* @param {ParticipantObject} participant - The participant
*/
export const publishMessageInRealtime = (
message: MessageLocalEvent,
participant: Omit<ParticipantObject, "created_at" | "created_by">,
): void => {
if (participant.type !== "channel") return;
const room = `/companies/${participant.company_id}/workspaces/${participant.workspace_id}/channels/${participant.id}/feed`;
const type = "message";
const entity = message.resource;
const context = message.context;
localEventBus.publish("realtime:publish", {
topic: message.created ? RealtimeEntityActionType.Created : RealtimeEntityActionType.Updated,
event: {
type,
room: ResourcePath.get(room),
resourcePath: getThreadMessagePath(context as ThreadExecutionContext) + "/" + entity.id,
entity,
result: message.created
? new CreateResult<Message>(type, entity)
: new UpdateResult<Message>(type, entity),
},
} as RealtimeLocalBusEvent<Message>);
};
twake/backend/node/src/services/messages/services/views.ts
-456
View File
@@ -1,456 +0,0 @@
import _, { uniqBy } from "lodash";
import { Initializable, TwakeServiceProvider } from "../../../core/platform/framework";
import {
ExecutionContext,
ListResult,
Paginable,
Pagination,
} from "../../../core/platform/framework/api/crud-service";
import Repository from "../../../core/platform/services/database/services/orm/repository/repository";
import SearchRepository from "../../../core/platform/services/search/repository";
import { fileIsMedia } from "../../../services/files/utils";
import { formatUser } from "../../../utils/users";
import gr from "../../global-resolver";
import { MessageChannelMarkedRef } from "../entities/message-channel-marked-refs";
import { MessageChannelRef } from "../entities/message-channel-refs";
import { MessageFileRef } from "../entities/message-file-refs";
import { MessageFile } from "../entities/message-files";
import { Message } from "../entities/messages";
import { Thread } from "../entities/threads";
import {
ChannelViewExecutionContext,
CompanyExecutionContext,
FlatFileFromMessage,
FlatPinnedFromMessage,
MessageViewListOptions,
MessageWithReplies,
SearchMessageFilesOptions,
SearchMessageOptions,
} from "../types";
import { FileSearchResult } from "../web/controllers/views/search-files";
import { buildMessageListPagination } from "./utils";
export class ViewsServiceImpl implements TwakeServiceProvider, Initializable {
version: "1";
repositoryChannelRefs: Repository<MessageChannelRef>;
repositoryThreads: Repository<Thread>;
repositoryFilesRef: Repository<MessageFileRef>;
repositoryMessageFile: Repository<MessageFile>;
repositoryMarkedRef: Repository<MessageChannelMarkedRef>;
searchRepository: SearchRepository<Message>;
searchFilesRepository: SearchRepository<MessageFile>;
async init(): Promise<this> {
this.searchRepository = gr.platformServices.search.getRepository<Message>("messages", Message);
this.searchFilesRepository = gr.platformServices.search.getRepository<MessageFile>(
"message_files",
MessageFile,
);
this.repositoryThreads = await gr.database.getRepository<Thread>("threads", Thread);
this.repositoryChannelRefs = await gr.database.getRepository<MessageChannelRef>(
"message_channel_refs",
MessageChannelRef,
);
this.repositoryFilesRef = await gr.database.getRepository<MessageFileRef>(
"message_file_refs",
MessageFileRef,
);
this.repositoryMarkedRef = await gr.database.getRepository<MessageChannelMarkedRef>(
"message_channel_marked_refs",
MessageChannelMarkedRef,
);
this.repositoryMessageFile = await gr.database.getRepository<MessageFile>(
"message_files",
MessageFile,
);
return this;
}
async listChannelFiles(
pagination: Pagination,
options?: MessageViewListOptions,
context?: ChannelViewExecutionContext,
): Promise<ListResult<MessageWithReplies | FlatFileFromMessage>> {
const refs = await this.repositoryFilesRef.find(
{
target_type: options?.media_only
? "channel_media"
: options?.file_only
? "channel_file"
: "channel",
target_id: context.channel.id,
company_id: context.channel.company_id,
},
buildMessageListPagination(pagination, "id"),
);
const threads: (MessageWithReplies & { context: MessageFileRef })[] = [];
for (const ref of refs.getEntities()) {
const thread = await this.repositoryThreads.findOne({ id: ref.thread_id }, {});
const extendedThread = await gr.services.messages.messages.getThread(
thread,
{
replies_per_thread: options.replies_per_thread || 1,
},
context,
);
const message = await gr.services.messages.messages.get({
thread_id: ref.thread_id,
id: ref.message_id,
});
if (message && extendedThread) {
extendedThread.highlighted_replies = [message];
threads.push({ ...extendedThread, context: ref });
}
}
if (options.flat) {
let files: FlatFileFromMessage[] = [];
for (const thread of threads) {
for (const reply of thread.highlighted_replies) {
for (const file of reply.files || []) {
if (file.id === thread.context.message_file_id) {
files.push({
file: file as MessageFile,
thread,
context: thread.context,
});
}
}
}
}
files = _.uniqBy(files, f => f.file.id);
refs.nextPage.page_token = files.length > 0 ? files[files.length - 1].context?.id : null;
return new ListResult("file", files, refs.nextPage);
}
return new ListResult("thread", threads, refs.nextPage);
}
async listChannelPinned(
pagination: Pagination,
options?: MessageViewListOptions,
context?: ChannelViewExecutionContext,
): Promise<ListResult<MessageWithReplies | FlatPinnedFromMessage>> {
const refs = await this.repositoryMarkedRef.find(
{
company_id: context.channel.company_id,
workspace_id: context.channel.workspace_id,
type: "pinned",
channel_id: context.channel.id,
},
buildMessageListPagination(pagination, "thread_id"),
context,
);
const threads: MessageWithReplies[] = [];
for (const ref of uniqBy(refs.getEntities(), "thread_id")) {
const thread = await this.repositoryThreads.findOne({ id: ref.thread_id }, {});
const extendedThread = await gr.services.messages.messages.getThread(
thread,
{
replies_per_thread: options.replies_per_thread || 1,
},
context,
);
if (extendedThread) {
threads.push(extendedThread);
}
}
for (const ref of refs.getEntities()) {
const extendedThread = threads.find(th => th.id === ref.thread_id);
if (extendedThread) {
extendedThread.highlighted_replies = extendedThread.highlighted_replies || [];
const message = await gr.services.messages.messages.get({
thread_id: ref.thread_id,
id: ref.message_id,
});
extendedThread.highlighted_replies.push(message);
}
}
if (options.flat) {
const messages: FlatPinnedFromMessage[] = [];
for (const thread of threads) {
for (const message of thread.highlighted_replies) {
messages.push({
message,
thread,
});
}
}
return new ListResult("message", messages, refs.nextPage);
}
return new ListResult("thread", threads, null);
}
async listChannelThreads(
pagination: Pagination,
options?: MessageViewListOptions,
context?: ChannelViewExecutionContext,
): Promise<ListResult<MessageWithReplies>> {
return this.listChannel(pagination, { ...options, replies_per_thread: 0 }, context);
}
/**
* Get last threads in a channel.
*
* We can do great improvements in the number of calls we currently do:
- Get last threads refs (1req)
- For each ref get the corresponding thread (to get number of replies => can be cached in the thread ref itself) (threads x reqs)
- For each thread, get first message (=> first message id can be cached in thread ref too and so replace this request by a $in) (threads x reqs)
- For each thread get last N replies (threads x reqs)
Total requests for "get last 20 threads with their last 3 replies":
+ 1
+ 20 (can be cached in first req)
+ 20 (can be replaced by a $in )
+ 20
----
= 61 reqs
* @param pagination
* @param options
* @param context
* @returns
*/
async listChannel(
pagination: Paginable,
options?: MessageViewListOptions,
context?: ChannelViewExecutionContext,
): Promise<ListResult<MessageWithReplies>> {
const threadsRefs = await this.repositoryChannelRefs.find(
{
company_id: context.channel.company_id,
workspace_id: context.channel.workspace_id,
channel_id: context.channel.id,
},
buildMessageListPagination(Pagination.fromPaginable(pagination), "message_id"),
context,
);
const threads = uniqBy(
(
await this.repositoryThreads.find(
{},
{
$in: [["id", threadsRefs.getEntities().map(ref => ref.thread_id)]],
},
context,
)
).getEntities(),
thread => thread.id,
);
//Get first message for each thread and add last replies for each thread
let threadWithLastMessages: MessageWithReplies[] = [];
if (options.replies_per_thread !== 0) {
await Promise.all(
threads.map(async (thread: Thread) => {
const extendedThread = await gr.services.messages.messages.getThread(
thread,
{
replies_per_thread: options.replies_per_thread || 3,
},
context,
);
if (
extendedThread?.last_replies?.length === 0 &&
extendedThread.created_at > new Date().getTime() - 1000 * 60 //This is important to avoid removing thread if people loads a channel at the same time people create a thread
) {
await gr.services.messages.threads.delete(
{ id: extendedThread.thread_id },
{ user: { id: null, server_request: true } },
);
} else if (extendedThread) {
threadWithLastMessages.push(extendedThread);
}
}),
);
}
threadWithLastMessages = threadWithLastMessages
.filter(m => m.id)
.sort((a, b) => a.stats.last_activity - b.stats.last_activity);
return new ListResult("thread", threadWithLastMessages, threadsRefs.nextPage);
}
async search(
pagination: Pagination,
options: SearchMessageOptions,
context?: ExecutionContext,
): Promise<ListResult<Message>> {
return await this.searchRepository
.search(
{
...(options.hasFiles ? { has_files: true } : {}),
...(options.hasMedias ? { has_medias: true } : {}),
},
{
pagination,
...(options.companyId ? { $in: [["company_id", [options.companyId]]] } : {}),
...(options.workspaceId ? { $in: [["workspace_id", [options.workspaceId]]] } : {}),
...(options.channelId ? { $in: [["channel_id", [options.channelId]]] } : {}),
...(options.sender ? { $in: [["user_id", [options.sender]]] } : {}),
$text: {
$search: options.search,
},
$sort: {
created_at: "desc",
},
},
context,
)
.then(a => {
return a;
});
}
async searchFiles(
pagination: Pagination,
options: SearchMessageFilesOptions,
context?: ExecutionContext,
): Promise<ListResult<MessageFile>> {
const temp = await this.searchFilesRepository.search(
{
...(options.isFile ? { is_file: true } : {}),
...(options.isMedia ? { is_media: true } : {}),
},
{
pagination,
...(options.companyId ? { $in: [["cache_company_id", [options.companyId]]] } : {}),
...(options.workspaceId ? { $in: [["cache_workspace_id", [options.workspaceId]]] } : {}),
...(options.channelId ? { $in: [["cache_channel_id", [options.channelId]]] } : {}),
...(options.sender ? { $in: [["cache_user_id", [options.sender]]] } : {}),
...(options.extension ? { $in: [["extension", [options.extension]]] } : {}),
$text: {
$search: options.search,
},
$sort: {
created_at: "desc",
},
},
context,
);
return new ListResult(temp.type, await this.checkFiles(temp.getEntities()), temp.nextPage);
}
async listUserMarkedFiles(
userId: string,
type: "user_upload" | "user_download" | "both",
media: "file_only" | "media_only" | "both",
context: CompanyExecutionContext,
pagination: Pagination,
): Promise<ListResult<FileSearchResult>> {
let files: (MessageFile & { context: MessageFileRef })[] = [];
let nextPageUploads: Paginable;
let nextPageDownloads: Paginable;
do {
const uploads =
type === "user_upload" || type === "both"
? await this.repositoryFilesRef
.find(
{ target_type: "user_upload", target_id: userId, company_id: context.company.id },
{
pagination: { ...pagination, page_token: nextPageUploads?.page_token },
},
context,
)
.then(a => {
nextPageUploads = a.nextPage;
return a.getEntities();
})
: [];
const downloads =
type === "user_download" || type === "both"
? await this.repositoryFilesRef
.find(
{ target_type: "user_download", target_id: userId, company_id: context.company.id },
{
pagination: { ...pagination, page_token: nextPageDownloads?.page_token },
},
context,
)
.then(a => {
nextPageDownloads = a.nextPage;
return a.getEntities();
})
: [];
const refs = [...uploads, ...downloads];
const messageFilePromises: Promise<MessageFile & { context: MessageFileRef }>[] = refs.map(
async ref => {
try {
const res = {
...(await this.repositoryMessageFile.findOne(
{
message_id: ref.message_id,
id: ref.message_file_id,
},
{},
context,
)),
context: ref,
};
return res;
} catch (e) {
return null;
}
},
);
const messageFiles = _.uniqBy(
(await Promise.all(messageFilePromises)).filter(a => a),
a => a.metadata?.source + JSON.stringify(a.metadata?.external_id),
);
files = [...files, ...messageFiles.filter(a => a)].filter(ref => {
//Apply media filer
const isMedia = fileIsMedia(ref);
return !((media === "file_only" && isMedia) || (media === "media_only" && !isMedia));
});
files = files.sort((a, b) => b.created_at - a.created_at);
} while (
files.length < (parseInt(pagination.limitStr) || 100) &&
(nextPageDownloads?.page_token || nextPageUploads?.page_token)
);
const fileWithUserAndMessagePromise: Promise<FileSearchResult>[] = files.map(
async file =>
({
user: await formatUser(await gr.services.users.getCached({ id: file.cache?.user_id })),
message: await gr.services.messages.messages.get({
id: file.message_id,
thread_id: file.thread_id,
}),
...file,
} as FileSearchResult),
);
const fileWithUserAndMessage = await Promise.all(fileWithUserAndMessagePromise);
return new ListResult<FileSearchResult>(
"file",
await this.checkFiles(fileWithUserAndMessage),
nextPageUploads || nextPageDownloads,
);
}
async checkFiles<T extends MessageFile>(files: T[]): Promise<T[]> {
const results = await Promise.all(
files.map(async file => {
if (file.metadata.source !== "internal") return true;
const ei = file.metadata.external_id;
return await gr.services.files.exists(ei.id, ei.company_id);
}),
);
return files.filter((_v, index) => results[index]);
}
}
twake/backend/node/src/services/messages/types.ts
-188
View File
@@ -1,188 +0,0 @@
import { ExecutionContext } from "../../core/platform/framework/api/crud-service";
import { uuid } from "../../utils/types";
import { HookType } from "../applications-api/types";
import { Channel } from "../channels/entities";
import { UserObject } from "../user/web/types";
import { MessageFileRef } from "./entities/message-file-refs";
import { MessageFile } from "./entities/message-files";
import { Message, MessageWithUsers } from "./entities/messages";
import { Thread } from "./entities/threads";
export type SpecialMention = "all" | "here" | "everyone" | "channel";
export type MessageNotification = {
company_id: uuid;
workspace_id: uuid | "direct";
channel_id: uuid;
thread_id: uuid;
id: uuid;
sender: uuid;
creation_date: number;
mentions?: {
users?: uuid[];
teams?: uuid[];
specials?: SpecialMention[];
};
sender_name?: string;
channel_name?: string;
company_name?: string;
workspace_name?: string;
title: string;
text: string;
};
export type MessageHook = HookType & {
channel: Channel;
thread: Thread;
message: Message;
};
export type MessageWithReplies = Message & {
last_replies: Message[];
thread?: MessageWithReplies;
highlighted_replies?: Message[];
stats: {
last_activity: number;
replies: number;
};
};
export type MessageWithRepliesWithUsers = MessageWithReplies & {
last_replies: MessageWithUsers[];
thread?: MessageWithRepliesWithUsers;
highlighted_replies?: MessageWithUsers[];
users: UserObject[];
};
export interface CompanyExecutionContext extends ExecutionContext {
company: { id: string };
}
export interface ThreadExecutionContext extends CompanyExecutionContext {
thread: { id: string };
company: { id: string };
workspace?: { id: string };
channel?: { id: string };
}
export interface ChannelViewExecutionContext extends ExecutionContext {
channel: {
company_id: string;
workspace_id: string;
id: string;
};
}
export interface MessageLocalEvent {
resource: Message;
context: ThreadExecutionContext;
created: boolean;
}
export interface PaginationQueryParameters {
page_token?: string;
limit?: string;
websockets?: boolean;
direction?: "history" | "future";
}
export interface MessageViewListOptions {
include_users?: boolean;
replies_per_thread?: number;
flat?: boolean;
emojis?: boolean;
media_only?: boolean;
file_only?: boolean;
}
export interface MessageListQueryParameters extends PaginationQueryParameters {
include_users: boolean;
}
export interface PinOperation {
id: string;
pin: boolean;
}
export interface ReactionOperation {
id: string;
reactions: string[];
}
export interface BookmarkOperation {
id: string;
bookmark_id: string;
active: boolean;
}
export type MessageFileDownloadEvent = {
user: { id: string };
operation: { message_id: string; thread_id: string; message_file_id: string };
};
export interface MessagesSaveOptions {
threadInitialMessage?: boolean;
enforceViewPropagation?: boolean;
message_moved?: boolean;
previous_thread?: string; //If message was in a previous thread before (moved) then this indicate when is it from
}
export interface MessagesGetThreadOptions {
replies_per_thread?: number;
includeQuoteInMessage?: boolean;
}
export type SearchMessageOptions = {
search?: string;
companyId?: string;
workspaceId?: string;
channelId?: string;
hasFiles?: boolean;
hasMedias?: boolean;
sender?: string;
};
export type SearchMessageFilesOptions = {
search?: string;
companyId?: string;
workspaceId?: string;
channelId?: string;
sender?: string;
isFile?: boolean;
isMedia?: boolean;
extension?: string;
};
export type InboxOptions = {
companyId: string;
};
export type FlatFileFromMessage = {
file: MessageFile;
thread: MessageWithReplies;
context: MessageFileRef;
};
export type FlatPinnedFromMessage = {
message: any;
thread: any;
};
export interface DeleteLinkOperation {
message_id: string;
thread_id: string;
link: string;
}
export type UpdateDeliveryStatusOperation = {
status: "delivered" | "read";
self_message?: boolean;
} & MessageIdentifier;
export type MessageReadType = {
messages: MessageIdentifier[];
channel_id: string;
};
export type MessageIdentifier = {
message_id: string;
thread_id: string;
};
twake/backend/node/src/services/messages/utils.ts
-32
View File
@@ -1,32 +0,0 @@
import { MessageFile } from "./entities/message-files";
import { Message } from "./entities/messages";
import { MessageWithReplies } from "./types";
export const formatMessageFile = (file: MessageFile): MessageFile => {
return {
...file,
metadata: {
...file.metadata,
thumbnails: [
...file.metadata.thumbnails.map(thumbnail => ({
...thumbnail,
full_url: thumbnail.url.match(/https?:\/\//)
? "/internal/services/files/v1/" + thumbnail.url.replace(/^\//, "")
: thumbnail.url,
})),
],
},
};
};
export const formatMessage = (
message: MessageWithReplies | Message,
): MessageWithReplies | Message => {
return {
...message,
files: message.files.map(formatMessageFile),
...((message as MessageWithReplies).last_replies
? { last_replies: (message as MessageWithReplies).last_replies.map(formatMessage) }
: {}),
};
};
twake/backend/node/src/services/messages/web/controllers/index.ts
-22
View File
@@ -1,22 +0,0 @@
import { ParticipantObject } from "../../entities/threads";
import { ThreadExecutionContext } from "../../types";
export * from "./threads";
export * from "./messages";
export * from "./views";
export * from "./user-bookmarks";
export const extendExecutionContentWithChannel = (
participants: ParticipantObject[],
context: ThreadExecutionContext,
): ThreadExecutionContext => {
const channelInfo = participants.find(a => a.type === "channel");
if (!channelInfo) {
return context;
}
return Object.assign(context, {
company: { id: channelInfo.company_id },
workspace: { id: channelInfo.workspace_id },
channel: { id: channelInfo.id },
});
};
twake/backend/node/src/services/messages/web/controllers/messages-files.ts
-52
View File
@@ -1,52 +0,0 @@
import gr from "../../../global-resolver";
import { FastifyRequest, FastifyReply } from "fastify";
export class MessagesFilesController {
async deleteMessageFile(
request: FastifyRequest<{
Params: {
company_id: string;
message_id: string;
message_file_id: string;
};
}>,
reply: FastifyReply,
) {
const user = request.currentUser;
const resp = await gr.services.messages.messagesFiles.deleteMessageFile(
request.params.message_id,
request.params.message_file_id,
user.id,
);
if (!resp) reply.code(404).send();
reply.send({ resource: resp });
}
async getMessageFile(
request: FastifyRequest<{
Params: {
company_id: string;
message_id: string;
message_file_id: string;
};
}>,
reply: FastifyReply,
) {
const user = request.currentUser;
const resp = await gr.services.messages.messagesFiles.getMessageFile(
request.params.message_id,
request.params.message_file_id,
);
if (!resp) reply.code(404).send();
//Check user has access to this file (check access to channel)
if (!(await gr.services.channels.members.getChannelMember(user, resp.channel))) {
reply.code(403).send();
}
reply.send({ resource: resp });
}
}
twake/backend/node/src/services/messages/web/controllers/messages.ts
-531
View File
@@ -1,531 +0,0 @@
import { FastifyReply, FastifyRequest } from "fastify";
import { CrudController } from "../../../../core/platform/services/webserver/types";
import {
ResourceDeleteResponse,
ResourceGetResponse,
ResourceListResponse,
ResourceUpdateResponse,
} from "../../../../utils/types";
import { getInstance as getMessageInstance, Message } from "../../entities/messages";
import {
CompanyExecutionContext,
MessageListQueryParameters,
MessageReadType,
MessageWithReplies,
ThreadExecutionContext,
} from "../../types";
import { handleError } from "../../../../utils/handleError";
import { CrudException, Pagination } from "../../../../core/platform/framework/api/crud-service";
import { getThreadMessageWebsocketRoom } from "../realtime";
import { ThreadPrimaryKey } from "../../entities/threads";
import { extendExecutionContentWithChannel } from "./index";
import gr from "../../../global-resolver";
import { formatUser } from "../../../../utils/users";
import { UserObject } from "../../../../services/user/web/types";
export class MessagesController
implements
CrudController<
ResourceGetResponse<Message>,
ResourceUpdateResponse<Message>,
ResourceListResponse<Message>,
ResourceDeleteResponse
>
{
async save(
request: FastifyRequest<{
Querystring: {
include_users: boolean;
};
Params: {
company_id: string;
thread_id: string;
message_id?: string;
};
Body: {
resource: Message;
options: {
previous_thread: string;
};
};
}>,
reply: FastifyReply,
): Promise<ResourceUpdateResponse<Message>> {
const context = getThreadExecutionContext(request);
try {
if (request.body.options?.previous_thread) {
//First move the message to another thread, then edit it
await gr.services.messages.messages.move(
{ id: request.params.message_id || undefined },
{
previous_thread: request.body.options?.previous_thread,
},
context,
);
}
const thread = await gr.services.messages.threads.get(
{
id: context.thread.id,
} as ThreadPrimaryKey,
context,
);
if (!thread) {
throw "Message must be in a thread";
}
let hasOneMembership = false;
for (const participant of thread.participants) {
if (thread.created_by === context.user.id) {
hasOneMembership = true;
break;
}
if (participant.type === "channel") {
const isMember = await gr.services.channels.members.getChannelMember(
{ id: context.user.id },
{
company_id: participant.company_id,
workspace_id: participant.workspace_id,
id: participant.id,
},
);
if (isMember) {
hasOneMembership = true;
break;
}
} else if (participant.type === "user") {
if (participant.id === context.user.id) {
hasOneMembership = true;
break;
}
}
}
if (!hasOneMembership) {
throw CrudException.notFound("You can't post in this thread");
}
const result = await gr.services.messages.messages.save(
{
id: request.params.message_id || undefined,
...request.body.resource,
},
{},
extendExecutionContentWithChannel(thread.participants, context),
);
let entity = result.entity;
if (request.query.include_users) {
entity = await gr.services.messages.messages.includeUsersInMessage(entity, context);
}
return {
resource: entity,
};
} catch (err) {
handleError(reply, err);
}
}
async forceDelete(
request: FastifyRequest<{
Params: {
company_id: string;
thread_id: string;
message_id: string;
};
}>,
reply: FastifyReply,
) {
const context = getThreadExecutionContext(request);
try {
await gr.services.messages.messages.forceDelete(
getMessageInstance({
thread_id: request.params.thread_id,
id: request.params.message_id,
}),
context,
);
return {
status: "success",
};
} catch (err) {
handleError(reply, err);
}
}
async delete(
request: FastifyRequest<{
Params: {
company_id: string;
thread_id: string;
message_id: string;
};
}>,
reply: FastifyReply,
): Promise<ResourceDeleteResponse> {
const context = getThreadExecutionContext(request);
try {
await gr.services.messages.messages.delete(
getMessageInstance({
thread_id: request.params.thread_id,
id: request.params.message_id,
}),
context,
);
return {
status: "success",
};
} catch (err) {
handleError(reply, err);
}
}
async get(
request: FastifyRequest<{
Querystring: MessageListQueryParameters;
Params: {
company_id: string;
thread_id: string;
message_id: string;
};
}>,
reply: FastifyReply,
): Promise<ResourceGetResponse<Message>> {
const context = getThreadExecutionContext(request);
try {
let resource = await gr.services.messages.messages.get(
{
thread_id: request.params.thread_id,
id: request.params.message_id,
},
context,
);
if (request.query.include_users) {
if ((resource as MessageWithReplies).last_replies?.length) {
resource = await gr.services.messages.messages.includeUsersInMessageWithReplies(
resource as MessageWithReplies,
context,
);
} else {
resource = await gr.services.messages.messages.includeUsersInMessage(resource, context);
}
}
return {
resource: resource,
};
} catch (err) {
handleError(reply, err);
}
}
async list(
request: FastifyRequest<{
Querystring: MessageListQueryParameters;
Params: {
company_id: string;
thread_id: string;
};
}>,
reply: FastifyReply,
): Promise<ResourceListResponse<Message>> {
const context = getThreadExecutionContext(request);
try {
const resources = await gr.services.messages.messages.list(
new Pagination(
request.query.page_token,
request.query.limit,
request.query.direction !== "history",
),
{ ...request.query, include_users: request.query.include_users || false },
context,
);
let entities = [];
if (request.query.include_users) {
for (const msg of resources.getEntities()) {
entities.push(await gr.services.messages.messages.includeUsersInMessage(msg, context));
}
} else {
entities = resources.getEntities();
}
return {
resources: entities,
...(request.query.websockets && {
websockets: gr.platformServices.realtime.sign(
[{ room: getThreadMessageWebsocketRoom(context) }],
context.user.id,
),
}),
...(resources.page_token && {
next_page_token: resources.page_token,
}),
};
} catch (err) {
handleError(reply, err);
}
}
async reaction(
request: FastifyRequest<{
Params: {
company_id: string;
thread_id: string;
message_id: string;
};
Body: {
reactions: string[];
};
}>,
reply: FastifyReply,
): Promise<ResourceUpdateResponse<Message>> {
const context = getThreadExecutionContext(request);
try {
const result = await gr.services.messages.messages.reaction(
{
id: request.params.message_id,
reactions: request.body.reactions,
},
{},
context,
);
return {
resource: result.entity,
};
} catch (err) {
handleError(reply, err);
}
}
async bookmark(
request: FastifyRequest<{
Params: {
company_id: string;
thread_id: string;
message_id: string;
};
Body: {
active: boolean;
bookmark_id: string;
};
}>,
reply: FastifyReply,
): Promise<ResourceUpdateResponse<Message>> {
const context = getThreadExecutionContext(request);
try {
const result = await gr.services.messages.messages.bookmark(
{
id: request.params.message_id,
bookmark_id: request.body.bookmark_id,
active: request.body.active,
},
{},
context,
);
return {
resource: result.entity,
};
} catch (err) {
handleError(reply, err);
}
}
async pin(
request: FastifyRequest<{
Params: {
company_id: string;
thread_id: string;
message_id: string;
};
Body: {
pin: boolean;
};
}>,
reply: FastifyReply,
): Promise<ResourceUpdateResponse<Message>> {
const context = getThreadExecutionContext(request);
try {
const result = await gr.services.messages.messages.pin(
{
id: request.params.message_id,
pin: request.body.pin,
},
{},
context,
);
return {
resource: result.entity,
};
} catch (err) {
handleError(reply, err);
}
}
async download(
request: FastifyRequest<{
Params: {
company_id: string;
thread_id: string;
message_id: string;
message_file_id: string;
};
}>,
reply: FastifyReply,
): Promise<{ status: "ok" }> {
const context = getThreadExecutionContext(request);
try {
await gr.services.messages.messages.download(
{
id: request.params.message_id,
thread_id: request.params.thread_id,
message_file_id: request.params.message_file_id,
},
{},
context,
);
return {
status: "ok",
};
} catch (err) {
handleError(reply, err);
}
}
/**
* Delete link preview handler
*
* @param {FastifyRequest} request - The request object
* @param {FastifyReply} reply - The reply object
* @returns {Promise<ResourceUpdateResponse<Message>>} - The response object
*/
async deleteLinkPreview(
request: FastifyRequest<{
Params: {
company_id: string;
thread_id: string;
message_id: string;
encoded_url: string;
};
Body: {
url: string;
};
}>,
reply: FastifyReply,
): Promise<ResourceUpdateResponse<Message>> {
const context = getThreadExecutionContext(request);
try {
const result = await gr.services.messages.messages.deleteLinkPreview(
{
message_id: request.params.message_id,
thread_id: request.params.thread_id,
link: request.body.url,
},
context,
);
return {
resource: result.entity,
};
} catch (err) {
handleError(reply, err);
}
}
/**
* Mark messages as seen
*
* @param {FastifyRequest} request - The request object
* @param {FastifyReply} reply - The reply object
* @returns {Promise<boolean>} - The response promise
*/
async read(
request: FastifyRequest<{
Params: {
company_id: string;
workspace_id: string;
};
Body: MessageReadType;
}>,
reply: FastifyReply,
): Promise<boolean> {
try {
const { messages, channel_id } = request.body;
const context: CompanyExecutionContext = {
company: { id: request.params.company_id },
user: request.currentUser,
url: request.url,
method: request.routerMethod,
reqId: request.id,
transport: "http",
};
const result = await gr.services.messages.messages.read(messages, {
...context,
channel_id,
workspace_id: request.params.workspace_id,
});
return !!result;
} catch (err) {
handleError(reply, err);
return false;
}
}
/**
* get users who've seen the message
*
* @param {FastifyRequest} request - the request object
* @param {FastifyReply} reply - the reply object
* @returns {Promise<ResourceListResponse<UserObject>>} - the users list
*/
async seenBy(
request: FastifyRequest<{
Params: {
company_id: string;
workspace_id: string;
thread_id: string;
message_id: string;
};
}>,
reply: FastifyReply,
): Promise<ResourceListResponse<UserObject>> {
try {
const { message_id, workspace_id } = request.params;
const context = getThreadExecutionContext(request);
const userIds = await gr.services.messages.messages.listSeenBy(message_id, {
...context,
workspace: { id: workspace_id },
});
const users = await Promise.all(userIds.map(id => gr.services.users.get({ id }, context)));
const resources = await Promise.all(users.map(user => formatUser(user)));
return {
resources,
};
} catch (error) {
handleError(reply, error);
}
}
}
function getThreadExecutionContext(
request: FastifyRequest<{
Params: { company_id: string; thread_id: string };
}>,
): ThreadExecutionContext {
return {
user: request.currentUser,
thread: { id: request.params.thread_id },
company: { id: request.params.company_id },
url: request.url,
method: request.routerMethod,
reqId: request.id,
transport: "http",
};
}
twake/backend/node/src/services/messages/web/controllers/threads.ts
-98
View File
@@ -1,98 +0,0 @@
import { FastifyReply, FastifyRequest } from "fastify";
import { CrudController } from "../../../../core/platform/services/webserver/types";
import {
ResourceCreateResponse,
ResourceDeleteResponse,
ResourceGetResponse,
ResourceListResponse,
} from "../../../../utils/types";
import { handleError } from "../../../../utils/handleError";
import { CompanyExecutionContext } from "../../types";
import { ParticipantObject, Thread } from "../../entities/threads";
import gr from "../../../global-resolver";
import { CrudException } from "../../../../core/platform/framework/api/crud-service";
export class ThreadsController
implements
CrudController<
ResourceGetResponse<Thread>,
ResourceCreateResponse<Thread>,
ResourceListResponse<Thread>,
ResourceDeleteResponse
>
{
async save(
request: FastifyRequest<{
Params: {
company_id: string;
thread_id: string;
};
Body: {
resource: {
participants: ParticipantObject[];
};
options: {
message: any;
participants: { add: ParticipantObject[]; remove: ParticipantObject[] };
};
};
}>,
reply: FastifyReply,
): Promise<ResourceCreateResponse<Thread>> {
const context = getCompanyExecutionContext(request);
const participants =
(request.body.resource.participants?.length
? request.body.resource?.participants
: request.body.options?.participants?.add) || [];
for (const participant of participants) {
if (participant.type === "channel") {
const isMember = await gr.services.channels.members.getChannelMember(
{ id: context.user.id },
{
company_id: participant.company_id,
workspace_id: participant.workspace_id,
id: participant.id,
},
);
if (!isMember && !context.user.application_id && !context.user.server_request) {
throw CrudException.notFound("Channel not found");
}
}
}
try {
const result = await gr.services.messages.threads.save(
{
id: request.params.thread_id || undefined,
participants: request.body.resource.participants || undefined,
},
{
message: request.body.options.message,
participants: request.body.options.participants,
},
context,
);
return {
resource: result.entity,
};
} catch (err) {
handleError(reply, err);
}
}
}
function getCompanyExecutionContext(
request: FastifyRequest<{
Params: { company_id: string };
}>,
): CompanyExecutionContext {
return {
user: request.currentUser,
company: { id: request.params.company_id },
url: request.url,
method: request.routerMethod,
reqId: request.id,
transport: "http",
};
}
twake/backend/node/src/services/messages/web/controllers/user-bookmarks.ts
-130
View File
@@ -1,130 +0,0 @@
import { FastifyReply, FastifyRequest } from "fastify";
import { CrudController } from "../../../../core/platform/services/webserver/types";
import {
ResourceDeleteResponse,
ResourceGetResponse,
ResourceListResponse,
ResourceUpdateResponse,
} from "../../../../utils/types";
import { UserMessageBookmark } from "../../entities/user-message-bookmarks";
import { handleError } from "../../../../utils/handleError";
import { CompanyExecutionContext } from "../../types";
import { getUserBookmarksWebsocketRoom } from "../realtime";
import gr from "../../../global-resolver";
export class UserBookmarksController
implements
CrudController<
ResourceGetResponse<UserMessageBookmark>,
ResourceUpdateResponse<UserMessageBookmark>,
ResourceListResponse<UserMessageBookmark>,
ResourceDeleteResponse
>
{
async save(
request: FastifyRequest<{
Params: {
company_id: string;
id: string;
};
Body: {
resource: {
name: string;
};
};
}>,
reply: FastifyReply,
): Promise<ResourceUpdateResponse<UserMessageBookmark>> {
const context = getCompanyExecutionContext(request);
try {
const result = await gr.services.messages.userBookmarks.save(
{
user_id: context.user.id,
company_id: request.params.company_id,
name: request.body.resource.name,
id: request.params.id || undefined,
},
context,
);
return {
resource: result.entity,
};
} catch (err) {
handleError(reply, err);
}
}
async delete(
request: FastifyRequest<{
Params: {
company_id: string;
id: string;
};
}>,
reply: FastifyReply,
): Promise<ResourceDeleteResponse> {
const context = getCompanyExecutionContext(request);
try {
const result = await gr.services.messages.userBookmarks.delete(
{
user_id: context.user.id,
company_id: request.params.company_id,
id: request.params.id,
},
context,
);
return {
status: result.deleted ? "success" : "error",
};
} catch (err) {
handleError(reply, err);
}
}
async list(
request: FastifyRequest<{
Params: {
company_id: string;
};
}>,
reply: FastifyReply,
): Promise<ResourceListResponse<UserMessageBookmark>> {
const context = getCompanyExecutionContext(request);
try {
const list = await gr.services.messages.userBookmarks.list(
{
user_id: context.user.id,
company_id: context.company.id,
},
context,
);
return {
websockets: gr.platformServices.realtime.sign(
[{ room: getUserBookmarksWebsocketRoom(context) }],
context.user.id,
),
resources: list.getEntities(),
...(list.page_token && {
next_page_token: list.page_token,
}),
};
} catch (err) {
handleError(reply, err);
}
}
}
function getCompanyExecutionContext(
request: FastifyRequest<{
Params: { company_id: string };
}>,
): CompanyExecutionContext {
return {
user: request.currentUser,
company: { id: request.params.company_id },
url: request.url,
method: request.routerMethod,
reqId: request.id,
transport: "http",
};
}
twake/backend/node/src/services/messages/web/controllers/views.ts
-317
View File
@@ -1,317 +0,0 @@
import { FastifyReply, FastifyRequest } from "fastify";
import { ResourceListResponse } from "../../../../utils/types";
import { Message } from "../../entities/messages";
import { handleError } from "../../../../utils/handleError";
import {
CrudException,
ListResult,
Pagination,
} from "../../../../core/platform/framework/api/crud-service";
import {
ChannelViewExecutionContext,
FlatFileFromMessage,
FlatPinnedFromMessage,
MessageViewListOptions,
MessageWithReplies,
PaginationQueryParameters,
} from "../../types";
import gr from "../../../global-resolver";
import { CompanyExecutionContext } from "../../../applications/web/types";
import searchFiles from "./views/search-files";
import recentFiles from "./views/recent-files";
export class ViewsController {
async feed(
request: FastifyRequest<{
Querystring: MessageViewListQueryParameters;
Params: {
company_id: string;
workspace_id: string;
channel_id: string;
};
}>,
reply: FastifyReply,
): Promise<
ResourceListResponse<MessageWithReplies | FlatFileFromMessage | FlatPinnedFromMessage>
> {
const pagination = new Pagination(
request.query.page_token,
request.query.limit,
request.query.direction !== "history",
);
const query = { ...request.query, include_users: request.query.include_users };
const context = getChannelViewExecutionContext(request);
const isMember = await gr.services.channels.members.getChannelMember(
{ id: context.user.id },
{
company_id: request.params.company_id,
workspace_id: request.params.workspace_id,
id: request.params.channel_id,
},
);
if (!isMember && !context.user.application_id && !context.user.server_request) {
throw CrudException.notFound("Channel not found");
}
let resources: ListResult<MessageWithReplies | FlatFileFromMessage | FlatPinnedFromMessage>;
try {
if (request.query.filter === "files") {
resources = await gr.services.messages.views.listChannelFiles(pagination, query, context);
} else if (request.query.filter === "thread") {
resources = await gr.services.messages.views.listChannelThreads(pagination, query, context);
} else if (request.query.filter === "pinned") {
resources = await gr.services.messages.views.listChannelPinned(pagination, query, context);
} else {
resources = await gr.services.messages.views.listChannel(pagination, query, context);
}
if (!resources) {
throw new Error("No list resources created");
}
let entities = [];
if (request.query.include_users) {
for (const msg of resources.getEntities()) {
if (request.query.flat) {
entities.push({
...msg,
...((msg as any).message
? {
message: await gr.services.messages.messages.includeUsersInMessageWithReplies(
(msg as any).message,
context,
),
}
: {}),
...((msg as any).thread
? {
thread: await gr.services.messages.messages.includeUsersInMessageWithReplies(
(msg as any).thread,
context,
),
}
: {}),
} as FlatFileFromMessage | FlatPinnedFromMessage);
} else {
entities.push(
await gr.services.messages.messages.includeUsersInMessageWithReplies(
msg as MessageWithReplies,
context,
),
);
}
}
} else {
entities = resources.getEntities();
}
return {
resources: entities,
...(request.query.websockets && {
websockets: gr.platformServices.realtime.sign(
[
{
room: `/companies/${context.channel.company_id}/workspaces/${context.channel.workspace_id}/channels/${context.channel.id}/feed`,
},
],
context.user.id,
),
}),
...(resources.page_token && {
next_page_token: resources.page_token,
}),
};
} catch (err) {
handleError(reply, err);
}
}
// Bookmarked messages of user from all over workspace
async bookmarks(): Promise<ResourceListResponse<MessageWithReplies>> {
return { resources: [] };
}
// Uploaded and downloaded files of user from all over workspace
files = recentFiles;
// Latest messages of user from all over workspace
async inbox(
request: FastifyRequest<{
Querystring: MessageViewListQueryParameters;
Params: {
company_id: string;
};
}>,
_reply: FastifyReply,
): Promise<ResourceListResponse<Message>> {
const context = getCompanyExecutionContext(request);
const messages = await gr.services.messages.messages.inbox(
request.currentUser.id,
context,
new Pagination(null, String(request.query.limit)),
);
return {
resources: messages.getEntities(),
};
}
async search(
request: FastifyRequest<{
Querystring: MessageViewSearchQueryParameters & { page_token: string };
Params: {
company_id: string;
};
}>,
context: ChannelViewExecutionContext,
): Promise<ResourceListResponse<Message>> {
if (request.query.q.length < 1) {
return { resources: [] };
}
const limit = +request.query.limit || 100;
async function* getNextMessages(
initialPageToken?: string,
): AsyncIterableIterator<{ msg: Message; pageToken: string }> {
let lastPageToken = initialPageToken;
let messages: Message[] = [];
let hasMoreMessages = true;
do {
messages = await gr.services.messages.views
.search(
new Pagination(lastPageToken, limit.toString()),
{
search: request.query.q,
workspaceId: request.query.workspace_id,
channelId: request.query.channel_id,
companyId: request.params.company_id,
...(request.query.has_files ? { hasFiles: true } : {}),
...(request.query.has_medias ? { hasMedias: true } : {}),
...(request.query.sender ? { sender: request.query.sender } : {}),
},
context,
)
.then((a: ListResult<Message>) => {
lastPageToken = a.nextPage.page_token;
if (!lastPageToken) {
hasMoreMessages = false;
}
return a.getEntities();
});
if (messages.length) {
for (const message of messages) {
yield { msg: message, pageToken: lastPageToken };
}
} else {
hasMoreMessages = false;
}
} while (hasMoreMessages);
}
const messages = [] as Message[];
let lastPageToken = null;
for await (const { msg, pageToken } of getNextMessages(request.query.page_token)) {
lastPageToken = pageToken;
const getChannelMember = await gr.services.channels.members.getChannelMember(
{ id: request.currentUser.id },
{
company_id: msg.cache.company_id,
workspace_id: msg.cache.workspace_id,
id: msg.cache.channel_id,
},
50,
context,
);
if (!getChannelMember) continue;
messages.push(msg);
if (messages.length == limit) {
break;
}
}
const extendedMessages = [];
for (const message of messages) {
const extended = {
...(await gr.services.messages.messages.includeUsersInMessage(message, context)),
channel: await gr.services.channels.channels.get(
{
company_id: message.cache?.company_id,
workspace_id: message.cache?.workspace_id,
id: message.cache?.channel_id,
},
context,
),
};
extendedMessages.push(extended);
}
return {
resources: extendedMessages,
...(lastPageToken && {
next_page_token: lastPageToken,
}),
};
}
searchFiles = searchFiles;
}
export interface MessageViewListQueryParameters
extends PaginationQueryParameters,
MessageViewListOptions {
include_users: boolean;
filter?: "pinned" | "files" | "thread";
}
export interface MessageViewSearchQueryParameters extends PaginationQueryParameters {
q: string;
workspace_id: string;
channel_id: string;
sender: string;
has_files: boolean;
has_medias: boolean;
}
function getChannelViewExecutionContext(
request: FastifyRequest<{
Params: {
company_id: string;
workspace_id: string;
channel_id: string;
};
}>,
): ChannelViewExecutionContext {
return {
user: request.currentUser,
url: request.url,
method: request.routerMethod,
reqId: request.id,
transport: "http",
channel: {
id: request.params.channel_id,
company_id: request.params.company_id,
workspace_id: request.params.workspace_id,
},
};
}
export function getCompanyExecutionContext(
request: FastifyRequest<{
Params: { company_id: string };
}>,
): CompanyExecutionContext {
return {
user: request.currentUser,
company: { id: request.params.company_id },
url: request.url,
method: request.routerMethod,
reqId: request.id,
transport: "http",
};
}
twake/backend/node/src/services/messages/web/controllers/views/recent-files.ts
-31
View File
@@ -1,31 +0,0 @@
import { FastifyRequest } from "fastify";
import { Pagination } from "../../../../../core/platform/framework/api/crud-service";
import { ResourceListResponse } from "../../../../../utils/types";
import gr from "../../../../global-resolver";
import { getCompanyExecutionContext } from "../views";
import { FileSearchResult } from "./search-files";
export default async (
request: FastifyRequest<{
Params: { company_id: string };
Querystring: {
page_token: string;
limit?: string;
type?: "user_upload" | "user_download";
media?: "media_only" | "file_only";
};
}>,
): Promise<ResourceListResponse<FileSearchResult>> => {
const userFiles = await gr.services.messages.views.listUserMarkedFiles(
request.currentUser.id,
request.query.type || "both",
request.query.media || "both",
getCompanyExecutionContext(request),
new Pagination(request.query.page_token, String(request.query.limit || 100)),
);
return {
resources: userFiles.getEntities().filter(a => a?.metadata?.external_id) as FileSearchResult[],
next_page_token: userFiles?.nextPage?.page_token,
};
};
twake/backend/node/src/services/messages/web/controllers/views/search-files.ts
-216
View File
@@ -1,216 +0,0 @@
import { FastifyRequest } from "fastify";
import { ListResult, Pagination } from "../../../../../core/platform/framework/api/crud-service";
import { MessageFile } from "../../../../../services/messages/entities/message-files";
import { Message } from "../../../../../services/messages/entities/messages";
import {
ChannelViewExecutionContext,
FlatFileFromMessage,
} from "../../../../../services/messages/types";
import { UserObject } from "../../../../../services/user/web/types";
import { PaginationQueryParameters, ResourceListResponse } from "../../../../../utils/types";
import { formatUser } from "../../../../../utils/users";
import gr from "../../../../global-resolver";
import { MessageFileRef } from "../../../../../services/messages/entities/message-file-refs";
import recentFiles from "./recent-files";
import { isEmpty } from "lodash";
interface MessageViewSearchFilesQueryParameters extends PaginationQueryParameters {
q: string;
workspace_id: string;
channel_id: string;
sender: string;
is_file: boolean;
is_media: boolean;
extension: string;
}
export type FileSearchResult = MessageFile & {
message: Message;
user: UserObject;
context?: MessageFileRef;
};
export default async (
request: FastifyRequest<{
Querystring: MessageViewSearchFilesQueryParameters & {
page_token: string;
};
Params: {
company_id: string;
};
}>,
context: ChannelViewExecutionContext,
): Promise<ResourceListResponse<MessageFile>> => {
const files = await searchFiles(request, context);
//Include channel in reply everytime
await Promise.all(
files.resources.map((f, i) => {
return (async () => {
const channel = await gr.services.channels.channels.get(
{
id: f.cache.channel_id,
company_id: f.cache.company_id,
workspace_id: f.cache.workspace_id,
},
context,
);
files.resources[i] = { ...f, channel } as MessageFile;
})();
}),
);
return files;
};
const searchFiles = async (
request: FastifyRequest<{
Querystring: MessageViewSearchFilesQueryParameters & {
page_token: string;
};
Params: {
company_id: string;
};
}>,
context: ChannelViewExecutionContext,
): Promise<ResourceListResponse<MessageFile>> => {
const limit = +request.query.limit || 100;
if (isEmpty(request.query?.q)) {
if (request.query.channel_id) {
const tmp = await gr.services.messages.views.listChannelFiles(
new Pagination(request.query.page_token, limit.toString()),
{
flat: true,
...(request.query.is_file ? { file_only: true } : {}),
...(request.query.is_media ? { media_only: true } : {}),
},
{
...context,
channel: {
id: request.query.channel_id,
workspace_id: request.query.workspace_id || context.channel.workspace_id,
company_id: request.params.company_id,
},
},
);
const resources: FileSearchResult[] = [];
for (let flatFile of tmp.getEntities()) {
flatFile = flatFile as FlatFileFromMessage;
resources.push({
...flatFile.file,
message: flatFile.thread,
context: flatFile.context,
user: await formatUser(
await gr.services.users.getCached({ id: flatFile.thread.user_id }),
),
});
}
return {
resources,
next_page_token: tmp.nextPage.page_token,
};
} else {
return recentFiles({
...request,
query: {
...request.query,
media: request.query.is_media
? "media_only"
: request.query.is_file
? "file_only"
: undefined,
},
});
}
}
async function* getNextMessageFiles(initialPageToken?: string): AsyncIterableIterator<{
msgFile: MessageFile;
pageToken: string;
}> {
let lastPageToken = initialPageToken;
let messageFiles: MessageFile[] = [];
let hasMoreMessageFiles = true;
do {
messageFiles = await gr.services.messages.views
.searchFiles(
new Pagination(lastPageToken, limit.toString()),
{
search: request.query.q,
companyId: request.params.company_id,
workspaceId: request.query.workspace_id,
channelId: request.query.channel_id,
...(request.query.is_file ? { isFile: true } : {}),
...(request.query.is_media ? { isMedia: true } : {}),
...(request.query.sender ? { sender: request.query.sender } : {}),
...(request.query.extension ? { extension: request.query.extension } : {}),
},
context,
)
.then((a: ListResult<MessageFile>) => {
lastPageToken = a.nextPage.page_token;
if (!lastPageToken) {
hasMoreMessageFiles = false;
}
return a.getEntities();
});
if (messageFiles.length) {
for (const messageFile of messageFiles) {
yield { msgFile: messageFile, pageToken: lastPageToken };
}
} else {
hasMoreMessageFiles = false;
}
} while (hasMoreMessageFiles);
}
let messageFiles = [] as FileSearchResult[];
let nextPageToken = null;
for await (const { msgFile, pageToken } of getNextMessageFiles(request.query.page_token)) {
nextPageToken = pageToken;
const getChannelMember = await gr.services.channels.members.getChannelMember(
{ id: request.currentUser.id },
{
company_id: msgFile.cache?.company_id,
workspace_id: msgFile.cache?.workspace_id,
id: msgFile.cache?.channel_id,
},
50,
context,
);
if (!getChannelMember) continue;
try {
const message = await gr.services.messages.messages.get({
thread_id: msgFile.thread_id,
id: msgFile.message_id,
});
const user = await formatUser(
await gr.services.users.getCached({ id: msgFile.cache?.user_id }),
);
messageFiles.push({ ...msgFile, user, message });
if (messageFiles.length == limit) {
break;
}
} catch (e) {
continue;
}
}
messageFiles = messageFiles
.filter(mf => mf.message.subtype !== "deleted")
.filter(a => a?.metadata?.external_id);
return {
resources: messageFiles,
...(nextPageToken && {
next_page_token: nextPageToken,
}),
};
};
twake/backend/node/src/services/messages/web/index.ts
-12
View File
@@ -1,12 +0,0 @@
import { FastifyInstance, FastifyRegisterOptions } from "fastify";
import routes from "./routes";
export default (
fastify: FastifyInstance,
options: FastifyRegisterOptions<{
prefix: string;
}>,
): void => {
fastify.log.debug("Configuring /internal/services/messages/v1 routes");
fastify.register(routes, options);
};
twake/backend/node/src/services/messages/web/realtime.ts
-13
View File
@@ -1,13 +0,0 @@
import { CompanyExecutionContext, ThreadExecutionContext } from "../types";
export const getUserBookmarksWebsocketRoom = (context: CompanyExecutionContext): string => {
return "/companies/" + context.company.id + "/messages/bookmarks";
};
export const getThreadMessageWebsocketRoom = (context: ThreadExecutionContext): string => {
return "/companies/" + context.company.id + "/threads/" + context.thread.id;
};
export const getThreadMessagePath = (context: ThreadExecutionContext): string => {
return "/companies/" + context.company.id + "/threads/" + context.thread.id;
};
twake/backend/node/src/services/messages/web/routes.ts
-227
View File
@@ -1,227 +0,0 @@
import { FastifyInstance, FastifyPluginCallback } from "fastify";
import {
MessagesController,
ThreadsController,
UserBookmarksController,
ViewsController,
} from "./controllers";
import { MessagesFilesController } from "./controllers/messages-files";
import { listUserFiles } from "./schemas";
const routes: FastifyPluginCallback = (fastify: FastifyInstance, options, next) => {
const threadsController = new ThreadsController();
const messagesController = new MessagesController();
const userBookmarksController = new UserBookmarksController();
const viewsController = new ViewsController();
const messagesFilesController = new MessagesFilesController();
/**
* User bookmarks collection
*/
fastify.route({
method: "GET",
url: "/companies/:company_id/preferences/bookmarks",
preValidation: [fastify.authenticate],
handler: userBookmarksController.list.bind(userBookmarksController),
});
fastify.route({
method: "POST",
url: "/companies/:company_id/preferences/bookmarks/:id",
preValidation: [fastify.authenticate],
handler: userBookmarksController.save.bind(userBookmarksController),
});
fastify.route({
method: "POST",
url: "/companies/:company_id/preferences/bookmarks",
preValidation: [fastify.authenticate],
handler: userBookmarksController.save.bind(userBookmarksController),
});
fastify.route({
method: "DELETE",
url: "/companies/:company_id/preferences/bookmarks/:id",
preValidation: [fastify.authenticate],
handler: userBookmarksController.delete.bind(userBookmarksController),
});
/**
* Threads creation route
*/
fastify.route({
method: "POST",
url: "/companies/:company_id/threads",
preValidation: [fastify.authenticate],
handler: threadsController.save.bind(threadsController),
});
fastify.route({
method: "POST",
url: "/companies/:company_id/threads/:thread_id",
preValidation: [fastify.authenticate],
handler: threadsController.save.bind(threadsController),
});
/**
* In threads message collection
*/
fastify.route({
method: "POST",
url: "/companies/:company_id/threads/:thread_id/messages",
preValidation: [fastify.authenticate],
handler: messagesController.save.bind(messagesController),
});
fastify.route({
method: "POST",
url: "/companies/:company_id/threads/:thread_id/messages/:message_id",
preValidation: [fastify.authenticate],
handler: messagesController.save.bind(messagesController),
});
fastify.route({
method: "GET",
url: "/companies/:company_id/threads/:thread_id/messages/:message_id",
preValidation: [fastify.authenticate],
handler: messagesController.get.bind(messagesController),
});
fastify.route({
method: "GET",
url: "/companies/:company_id/threads/:thread_id/messages",
preValidation: [fastify.authenticate],
handler: messagesController.list.bind(messagesController),
});
fastify.route({
method: "DELETE",
url: "/companies/:company_id/threads/:thread_id/messages/:message_id",
preValidation: [fastify.authenticate],
handler: messagesController.forceDelete.bind(messagesController),
});
fastify.route({
method: "POST",
url: "/companies/:company_id/threads/:thread_id/messages/:message_id/reaction",
preValidation: [fastify.authenticate],
handler: messagesController.reaction.bind(messagesController),
});
fastify.route({
method: "POST",
url: "/companies/:company_id/threads/:thread_id/messages/:message_id/bookmark",
preValidation: [fastify.authenticate],
handler: messagesController.bookmark.bind(messagesController),
});
fastify.route({
method: "POST",
url: "/companies/:company_id/threads/:thread_id/messages/:message_id/pin",
preValidation: [fastify.authenticate],
handler: messagesController.pin.bind(messagesController),
});
fastify.route({
method: "POST",
url: "/companies/:company_id/threads/:thread_id/messages/:message_id/download/:message_file_id",
preValidation: [fastify.authenticate],
handler: messagesController.download.bind(messagesController),
});
fastify.route({
method: "POST",
url: "/companies/:company_id/threads/:thread_id/messages/:message_id/delete",
preValidation: [fastify.authenticate],
handler: messagesController.delete.bind(messagesController),
});
fastify.route({
method: "POST",
url: "/companies/:company_id/threads/:thread_id/messages/:message_id/deletelink",
preValidation: [fastify.authenticate],
handler: messagesController.deleteLinkPreview.bind(messagesController),
});
fastify.route({
method: "POST",
url: "/companies/:company_id/workspaces/:workspace_id/threads/read",
preValidation: [fastify.authenticate],
handler: messagesController.read.bind(messagesController),
});
fastify.route({
method: "GET",
url: "/companies/:company_id/workspaces/:workspace_id/threads/:thread_id/messages/:message_id/seen",
preValidation: [fastify.authenticate],
handler: messagesController.seenBy.bind(messagesController),
});
/**
* Views routes
*/
fastify.route({
method: "GET",
url: "/companies/:company_id/workspaces/:workspace_id/channels/:channel_id/feed",
preValidation: [fastify.authenticate],
handler: viewsController.feed.bind(viewsController),
});
fastify.route({
method: "GET",
url: "/companies/:company_id/files",
preValidation: [fastify.authenticate],
schema: listUserFiles,
handler: viewsController.files.bind(viewsController),
});
fastify.route({
method: "GET",
url: "/companies/:company_id/bookmarks",
preValidation: [fastify.authenticate],
handler: viewsController.bookmarks.bind(viewsController),
});
fastify.route({
method: "GET",
url: "/companies/:company_id/inbox",
preValidation: [fastify.authenticate],
handler: viewsController.inbox.bind(viewsController),
});
fastify.route({
method: "GET",
url: "/companies/:company_id/search",
preValidation: [fastify.authenticate],
handler: viewsController.search.bind(viewsController),
});
fastify.route({
method: "GET",
url: "/companies/:company_id/files/search",
preValidation: [fastify.authenticate],
handler: viewsController.searchFiles.bind(viewsController),
});
/**
* Messages files routes
*/
fastify.route({
method: "GET",
url: "/companies/:company_id/messages/:message_id/files/:message_file_id",
preValidation: [fastify.authenticate],
handler: messagesFilesController.getMessageFile.bind(messagesFilesController),
});
fastify.route({
method: "DELETE",
url: "/companies/:company_id/messages/:message_id/files/:message_file_id",
preValidation: [fastify.authenticate],
handler: messagesFilesController.deleteMessageFile.bind(messagesFilesController),
});
next();
};
export default routes;
twake/backend/node/src/services/messages/web/schemas.ts
-37
View File
@@ -1,37 +0,0 @@
const publicFileSchema = {
type: "object",
properties: {
id: { type: "string" },
user_id: { type: "string" },
company_id: { type: "string" },
application_id: { type: "string" },
updated_at: { type: "number" },
created_at: { type: "number" },
metadata: { type: "object", additionalProperties: true },
thumbnails: { type: "array" },
upload_data: { type: "object", additionalProperties: true },
user: { type: "object", additionalProperties: true, nullable: true },
context: { type: "object", additionalProperties: true, nullable: true },
},
};
export const listUserFiles = {
request: {
properties: {
type: { type: "string", enum: ["user_upload", "user_download"] },
page_token: { type: "string" },
limit: { type: "number" },
},
required: ["type"],
},
response: {
"2xx": {
type: "object",
properties: {
resources: { type: "array", items: publicFileSchema },
next_page_token: { type: "string" },
},
required: ["resources"],
},
},
};
twake/backend/node/src/utils/messages.ts
-111
View File
@@ -1,111 +0,0 @@
import _ from "lodash";
import { ThreadExecutionContext } from "../services/messages/types";
import { getInstance, Message, MessageReaction } from "../services/messages/entities/messages";
export function getSubtype(
item: Pick<Message, "subtype">,
context?: ThreadExecutionContext,
): null | "application" | "deleted" | "system" {
//Application request
if (context?.user?.application_id) {
return item.subtype === "application" || item.subtype === "deleted" ? item.subtype : null;
}
//System request
else if (context?.user?.server_request) {
return item.subtype;
}
//User cannot set a subtype itself
return null;
}
export function updateMessageReactions(
message: Message,
selectedReactions: string[],
userId: string,
): void {
const reactions: { [key: string]: MessageReaction } = {};
for (const reaction of message.reactions || []) {
reactions[reaction.name] = reaction;
}
for (const reaction of selectedReactions) {
reactions[reaction] = reactions[reaction] || { name: reaction, count: 0, users: [] };
}
for (const key in reactions) {
if (reactions[key].users.includes(userId)) {
reactions[key].count--;
reactions[key].users = reactions[key].users.filter(u => u != userId);
}
if (selectedReactions.includes(key)) {
reactions[key].count++;
reactions[key].users.push(userId);
}
if (reactions[key].count === 0) {
delete reactions[key];
}
}
message.reactions = Object.values(reactions);
}
export function getDefaultMessageInstance(
item: Partial<Message>,
context: ThreadExecutionContext,
): Message {
let instance = getInstance({
id: undefined,
ephemeral:
(context?.user?.application_id || context?.user?.server_request) && item.ephemeral
? item.ephemeral
: null,
thread_id: (context?.user?.server_request ? item.thread_id : null) || context.thread.id,
type: context?.user?.server_request && item.type === "event" ? "event" : "message",
subtype: getSubtype({ subtype: item?.subtype || null }, context),
created_at: (context?.user?.server_request ? item.created_at : null) || new Date().getTime(),
user_id:
(context?.user?.server_request || context?.user?.application_id ? item.user_id : null) ||
context.user.id,
application_id:
(context?.user?.server_request ? item.application_id : null) ||
context?.user?.application_id ||
null,
text: item.text || "",
blocks: item.blocks || [],
files: item.files || null,
context: item.context || null,
edited: (context?.user?.server_request ? item.edited : null) || null, //Message cannot be created with edition status
pinned_info: item.pinned_info
? {
pinned_at: new Date().getTime(),
pinned_by: context.user.id,
}
: null,
quote_message: item.quote_message
? {
channel_id: context.channel.id,
workspace_id: context.workspace.id,
company_id: context.company.id,
...item.quote_message,
}
: null,
reactions: (context?.user?.server_request ? item.reactions : null) || null, // Reactions cannot be set on creation
bookmarks: (context?.user?.server_request ? item.bookmarks : null) || null,
override:
(context?.user?.application_id || context?.user?.server_request) && item.override
? {
title: item.override.title,
picture: item.override.picture,
}
: null, // Only apps and server can set an override on a message
status: item.status || "sent",
});
if (context.user.server_request) {
instance = _.assign(
instance,
_.pickBy(item, v => v !== undefined),
);
}
return instance;
}