diff --git a/.vscode/launch.json b/.vscode/launch.json index bbec12c7f5..6923fbaaa3 100644 --- a/.vscode/launch.json +++ b/.vscode/launch.json @@ -736,11 +736,14 @@ "MONGO_URL": "mongodb://localhost:27017", "MONGO_DB": "telegram-bot", "SECRET": "secret", - "ACCOUNTS_URL": "http://localhost:3000", + "ACCOUNTS_URL": "http://huly.local:3000", "SERVICE_ID": "telegram-bot-service", "MINIO_ACCESS_KEY": "minioadmin", "MINIO_SECRET_KEY": "minioadmin", - "MINIO_ENDPOINT": "localhost" + "MINIO_ENDPOINT": "localhost", + "QUEUE_CONFIG": "localhost:19092", + "QUEUE_REGION": "cockroach", + "DB_URL": "postgresql://root@huly.local:26257/defaultdb?sslmode=disable" }, "runtimeArgs": ["--nolazy", "-r", "ts-node/register"], "runtimeVersion": "20", diff --git a/common/config/rush/pnpm-lock.yaml b/common/config/rush/pnpm-lock.yaml index 275848d767..c629f575da 100644 --- a/common/config/rush/pnpm-lock.yaml +++ b/common/config/rush/pnpm-lock.yaml @@ -4778,7 +4778,7 @@ packages: version: 0.0.0 '@rush-temp/pod-telegram-bot@file:projects/pod-telegram-bot.tgz': - resolution: {integrity: sha512-QcNX6QCVKgZ6cV4ZkPBzdPH0Nu1F1sQ7yIMYQ4ikoR/BWWBB2U4E/YWd10I3c34QV3DUbKo3CID8EtkzrXadYQ==, tarball: file:projects/pod-telegram-bot.tgz} + resolution: {integrity: sha512-j08OCWsLAcBq3aElbkz0CbG2Ib0/CgHHHimwLZ7zpU7Php3ddqkAgaZFKoghv+zag8aSXG/eH34AMSSnraZ5HQ==, tarball: file:projects/pod-telegram-bot.tgz} version: 0.0.0 '@rush-temp/pod-telegram@file:projects/pod-telegram.tgz': @@ -5178,11 +5178,11 @@ packages: version: 0.0.0 '@rush-temp/server-telegram-resources@file:projects/server-telegram-resources.tgz': - resolution: {integrity: sha512-VsEVxoPM/hyelQa9/rQFzo+vbS8ixxU4RvVWdM0ae9HJuaEyszX4y9GheG/SDMa0NzCtyOOLf9KqL3ohGpkWRg==, tarball: file:projects/server-telegram-resources.tgz} + resolution: {integrity: sha512-sMiJ2rj5FQC6jNLypYdwCgUSj96/pz96QESWwmn0UoTVbg/SBKgFTZgF4CTUHmnr0ZE2vioiEkSZnsEiNuhU5Q==, tarball: file:projects/server-telegram-resources.tgz} version: 0.0.0 '@rush-temp/server-telegram@file:projects/server-telegram.tgz': - resolution: {integrity: sha512-99mvNi29b6C/lZr6MpKHmXDCm9FmOpg7AoWzLvkBpBd6rQJBK1Y9bRyuqHpH/VE1UgVtBR31wPpvKdMzTyluWA==, tarball: file:projects/server-telegram.tgz} + resolution: {integrity: sha512-KG0Q+p6leEECVhmFiwKm0CCQyFpT0lDnirzpeXW8LiqQfKnxQo2xfcIopr5PrExgkSJMdBe7tc05pqlCmbxZLg==, tarball: file:projects/server-telegram.tgz} version: 0.0.0 '@rush-temp/server-templates@file:projects/server-templates.tgz': @@ -22042,6 +22042,7 @@ snapshots: jest: 29.7.0(@types/node@20.11.19)(ts-node@10.9.2(@types/node@20.11.19)(typescript@5.3.3)) mongodb: 6.12.0(gcp-metadata@5.3.0(encoding@0.1.13))(snappy@7.2.2)(socks@2.8.3) otp-generator: 4.0.1 + postgres: 3.4.5 prettier: 3.2.5 telegraf: 4.16.3(encoding@0.1.13) ts-jest: 29.1.2(@babel/core@7.23.9)(@jest/types@29.6.3)(babel-jest@29.7.0(@babel/core@7.23.9))(esbuild@0.24.2)(jest@29.7.0(@types/node@20.11.19)(ts-node@10.9.2(@types/node@20.11.19)(typescript@5.3.3)))(typescript@5.3.3) diff --git a/desktop/src/ui/platform.ts b/desktop/src/ui/platform.ts index 75dc68f512..b5f6a73e83 100644 --- a/desktop/src/ui/platform.ts +++ b/desktop/src/ui/platform.ts @@ -266,7 +266,7 @@ export async function configurePlatform (): Promise { setMetadata(presentation.metadata.FrontVersion, config.VERSION) } setMetadata(telegram.metadata.TelegramURL, config.TELEGRAM_URL ?? 'http://localhost:8086') - setMetadata(telegram.metadata.BotUrl, config.TELEGRAM_BOT_URL ?? 'http://localhost:4020') + setMetadata(telegram.metadata.BotUrl, config.TELEGRAM_BOT_URL ?? 'http://huly.local:4020') setMetadata(gmail.metadata.GmailURL, config.GMAIL_URL ?? 'http://localhost:8087') setMetadata(calendar.metadata.CalendarServiceURL, config.CALENDAR_URL ?? 'http://localhost:8095') setMetadata(calendar.metadata.PublicScheduleURL, config.PUBLIC_SCHEDULE_URL) diff --git a/dev/prod/src/platform.ts b/dev/prod/src/platform.ts index 77342e6b48..f3571ead40 100644 --- a/dev/prod/src/platform.ts +++ b/dev/prod/src/platform.ts @@ -437,7 +437,7 @@ export async function configurePlatform() { setMetadata(presentation.metadata.FrontVersion, config.VERSION) } setMetadata(telegram.metadata.TelegramURL, config.TELEGRAM_URL ?? 'http://localhost:8086') - setMetadata(telegram.metadata.BotUrl, config.TELEGRAM_BOT_URL) + setMetadata(telegram.metadata.BotUrl, config.TELEGRAM_BOT_URL ?? 'http://huly.local:4020') setMetadata(gmail.metadata.GmailURL, config.GMAIL_URL ?? 'http://localhost:8087') setMetadata(calendar.metadata.CalendarServiceURL, config.CALENDAR_URL ?? 'http://localhost:8095') setMetadata(calendar.metadata.PublicScheduleURL, config.PUBLIC_SCHEDULE_URL) diff --git a/models/server-gmail/src/index.ts b/models/server-gmail/src/index.ts index d1837e989c..cd094d79a9 100644 --- a/models/server-gmail/src/index.ts +++ b/models/server-gmail/src/index.ts @@ -34,11 +34,6 @@ export function createModel (builder: Builder): void { } ) - builder.createDoc(serverNotification.class.NotificationProviderResources, core.space.Model, { - provider: gmail.providers.EmailNotificationProvider, - fn: serverGmail.function.SendEmailNotifications - }) - builder.createDoc(serverCore.class.Trigger, core.space.Model, { trigger: serverGmail.trigger.OnMessageCreate, txMatch: { diff --git a/models/server-notification/src/index.ts b/models/server-notification/src/index.ts index 2e533f4608..2af54c602e 100644 --- a/models/server-notification/src/index.ts +++ b/models/server-notification/src/index.ts @@ -14,21 +14,19 @@ // limitations under the License. // -import { type Builder, Mixin, Model } from '@hcengineering/model' +import { type Builder, Mixin } from '@hcengineering/model' import contact from '@hcengineering/contact' -import core, { type Ref } from '@hcengineering/core' -import { TClass, TDoc } from '@hcengineering/model-core' +import core from '@hcengineering/core' +import { TClass } from '@hcengineering/model-core' import { TNotificationType } from '@hcengineering/model-notification' -import notification, { type NotificationProvider } from '@hcengineering/notification' +import notification from '@hcengineering/notification' import { type Resource } from '@hcengineering/platform' import serverCore from '@hcengineering/server-core' import serverNotification, { type HTMLPresenter, type NotificationContentProvider, type NotificationPresenter, - type NotificationProviderFunc, - type NotificationProviderResources, type Presenter, type TextPresenter, type TypeMatch, @@ -57,20 +55,8 @@ export class TTypeMatch extends TNotificationType implements TypeMatch { func!: TypeMatchFunc } -@Model(serverNotification.class.NotificationProviderResources, core.class.Doc) -export class TNotificationProviderResources extends TDoc implements NotificationProviderResources { - provider!: Ref - fn!: Resource -} - export function createModel (builder: Builder): void { - builder.createModel( - THTMLPresenter, - TTextPresenter, - TTypeMatch, - TNotificationPresenter, - TNotificationProviderResources - ) + builder.createModel(THTMLPresenter, TTextPresenter, TTypeMatch, TNotificationPresenter) builder.createDoc(serverCore.class.Trigger, core.space.Model, { trigger: serverNotification.trigger.OnAttributeCreate, diff --git a/models/server-telegram/src/index.ts b/models/server-telegram/src/index.ts index 7ea53af712..4439c89b7d 100644 --- a/models/server-telegram/src/index.ts +++ b/models/server-telegram/src/index.ts @@ -37,11 +37,6 @@ export function createModel (builder: Builder): void { } ) - builder.createDoc(serverNotification.class.NotificationProviderResources, core.space.Model, { - provider: telegram.providers.TelegramNotificationProvider, - fn: serverTelegram.function.SendTelegramNotifications - }) - builder.createDoc(serverCore.class.Trigger, core.space.Model, { trigger: serverTelegram.trigger.OnMessageCreate, txMatch: { @@ -76,4 +71,20 @@ export function createModel (builder: Builder): void { serverFunc: serverTelegram.function.GetIntegrationOwnerTG } ) + + builder.createDoc(serverCore.class.Trigger, core.space.Model, { + trigger: serverTelegram.trigger.NotificationsHandler, + isAsync: true, + txMatch: { + _class: core.class.TxCreateDoc, + objectClass: notification.class.InboxNotification + } + }) + builder.createDoc(serverCore.class.Trigger, core.space.Model, { + trigger: serverTelegram.trigger.ProviderSettingsHandler, + isAsync: true, + txMatch: { + objectClass: notification.class.NotificationProviderSetting + } + }) } diff --git a/packages/account-client/src/client.ts b/packages/account-client/src/client.ts index 138569fad9..0736861866 100644 --- a/packages/account-client/src/client.ts +++ b/packages/account-client/src/client.ts @@ -20,7 +20,6 @@ import { type Person, type PersonUuid, type PersonInfo, - SocialId, Version, type WorkspaceInfoWithStatus, type WorkspaceMemberInfo, @@ -44,7 +43,8 @@ import type { Integration, IntegrationKey, IntegrationSecret, - IntegrationSecretKey + IntegrationSecretKey, + SocialId } from './types' import { getClientTimezone } from './utils' @@ -108,6 +108,7 @@ export interface AccountClient { findPersonBySocialKey: (socialKey: string, requireAccount?: boolean) => Promise findPersonBySocialId: (socialId: PersonId, requireAccount?: boolean) => Promise findSocialIdBySocialKey: (socialKey: string) => Promise + findFullSocialIdBySocialKey: (socialKey: string) => Promise getMailboxOptions: () => Promise createMailbox: (name: string, domain: string) => Promise<{ mailbox: string, socialId: PersonId }> getMailboxes: () => Promise @@ -142,8 +143,15 @@ export interface AccountClient { firstName: string, lastName: string ) => Promise<{ uuid: PersonUuid, socialId: PersonId }> + addSocialIdToPerson: ( + person: PersonUuid, + type: SocialIdType, + value: string, + confirmed: boolean, + displayValue?: string + ) => Promise + updateSocialId: (personId: PersonId, displayValue: string) => Promise exchangeGuestToken: (token: string) => Promise - addSocialIdToPerson: (person: PersonUuid, type: SocialIdType, value: string, confirmed: boolean) => Promise createIntegration: (integration: Integration) => Promise updateIntegration: (integration: Integration) => Promise deleteIntegration: (integrationKey: IntegrationKey) => Promise @@ -631,6 +639,14 @@ class AccountClientImpl implements AccountClient { return await this.rpc(request) } + async findFullSocialIdBySocialKey (socialKey: string): Promise { + const request = { + method: 'findFullSocialIdBySocialKey' as const, + params: { socialKey } + } + return await this.rpc(request) + } + async listWorkspaces (region?: string | null, mode: WorkspaceMode | null = null): Promise { const request = { method: 'listWorkspaces' as const, @@ -707,16 +723,25 @@ class AccountClientImpl implements AccountClient { person: PersonUuid, type: SocialIdType, value: string, - confirmed: boolean + confirmed: boolean, + displayValue?: string ): Promise { const request = { method: 'addSocialIdToPerson' as const, - params: { person, type, value, confirmed } + params: { person, type, value, confirmed, displayValue } } return await this.rpc(request) } + async updateSocialId (personId: PersonId, displayValue: string): Promise { + const request = { + method: 'updateSocialId' as const, + params: { personId, displayValue } + } + return await this.rpc(request) + } + async getMailboxOptions (): Promise { const request = { method: 'getMailboxOptions' as const, diff --git a/packages/account-client/src/types.ts b/packages/account-client/src/types.ts index e1f67c63e9..65b0d945da 100644 --- a/packages/account-client/src/types.ts +++ b/packages/account-client/src/types.ts @@ -4,7 +4,9 @@ import { WorkspaceDataId, WorkspaceUuid, type AccountRole, - type Timestamp + type Timestamp, + type SocialId as SocialIdBase, + PersonUuid } from '@hcengineering/core' export interface LoginInfo { @@ -63,6 +65,10 @@ export interface Integration { data?: Record } +export interface SocialId extends SocialIdBase { + personUuid: PersonUuid +} + export type IntegrationKey = Omit export interface IntegrationSecret { diff --git a/packages/core/src/classes.ts b/packages/core/src/classes.ts index f365b45c6c..1811557b72 100644 --- a/packages/core/src/classes.ts +++ b/packages/core/src/classes.ts @@ -853,6 +853,8 @@ export interface SocialId { value: string key: string // Calculated from type and value. Just for convenience. + displayValue?: string + // To be used later when person detaches social id from his account by any means // There should always be only one ACTIVE social id with the same key every time // active: boolean diff --git a/plugins/telegram-assets/lang/cs.json b/plugins/telegram-assets/lang/cs.json index 7262756d29..350164f54c 100644 --- a/plugins/telegram-assets/lang/cs.json +++ b/plugins/telegram-assets/lang/cs.json @@ -46,9 +46,9 @@ "ConnectAccount": "Připojit účet přes OTP", "ShowCommandsDetails": "Zobrazit detaily příkazů", "TurnNotificationsOff": "Vypnout oznámení", - "ConnectedDescriptionHtml": "✅ Jste připojeni jako {email}", - "AccountAlreadyConnectedHtml": "Již jste připojeni jako {email}", - "AccountConnectedHtml": "Úspěšně jste se připojili jako {email}", + "ConnectedDescriptionHtml": "✅ Jste připojeni jako {name}", + "AccountAlreadyConnectedHtml": "Již jste připojeni jako {name}", + "AccountConnectedHtml": "Úspěšně jste se připojili jako {name}", "AccountAlreadyConnected": "Účet je již připojen", "InvalidCode": "Neplatný kód", "SomethingWentWrong": "Něco se pokazilo. Zkuste to prosím znovu.", diff --git a/plugins/telegram-assets/lang/de.json b/plugins/telegram-assets/lang/de.json index 7ebf96e9a6..854cb9125b 100644 --- a/plugins/telegram-assets/lang/de.json +++ b/plugins/telegram-assets/lang/de.json @@ -46,9 +46,9 @@ "ConnectAccount": "Konto über OTP verbinden", "ShowCommandsDetails": "Befehlsdetails anzeigen", "TurnNotificationsOff": "Benachrichtigungen ausschalten", - "ConnectedDescriptionHtml": "✅Sie sind als {email} verbunden", - "AccountAlreadyConnectedHtml": "Sie sind bereits als {email} verbunden", - "AccountConnectedHtml": "Sie wurden erfolgreich als {email} verbunden", + "ConnectedDescriptionHtml": "✅Sie sind als {name} verbunden", + "AccountAlreadyConnectedHtml": "Sie sind bereits als {name} verbunden", + "AccountConnectedHtml": "Sie wurden erfolgreich als {name} verbunden", "AccountAlreadyConnected": "Konto bereits verbunden", "InvalidCode": "Ungültiger Code", "SomethingWentWrong": "Etwas ist schiefgelaufen. Bitte versuchen Sie es erneut.", diff --git a/plugins/telegram-assets/lang/en.json b/plugins/telegram-assets/lang/en.json index a4a88f6a48..47003567f7 100644 --- a/plugins/telegram-assets/lang/en.json +++ b/plugins/telegram-assets/lang/en.json @@ -46,9 +46,9 @@ "ConnectAccount": "Connect account via OTP", "ShowCommandsDetails": "Show commands details", "TurnNotificationsOff": "Turn notifications off", - "ConnectedDescriptionHtml": "✅Your are connected as {email}", - "AccountAlreadyConnectedHtml": "You are already connected as {email}", - "AccountConnectedHtml": "You have been successfully connected as {email}", + "ConnectedDescriptionHtml": "✅Your are connected as {name}", + "AccountAlreadyConnectedHtml": "You are already connected as {name}", + "AccountConnectedHtml": "You have been successfully connected as {name}", "AccountAlreadyConnected": "Account already connected", "InvalidCode": "Invalid code", "SomethingWentWrong": "Something went wrong. Please try again.", diff --git a/plugins/telegram-assets/lang/es.json b/plugins/telegram-assets/lang/es.json index ee0be62ea8..93af7373e3 100644 --- a/plugins/telegram-assets/lang/es.json +++ b/plugins/telegram-assets/lang/es.json @@ -46,9 +46,9 @@ "ConnectAccount": "Conectar cuenta a través de OTP", "ShowCommandsDetails": "Mostrar detalles de los comandos", "TurnNotificationsOff": "Desactivar notificaciones", - "ConnectedDescriptionHtml": "✅Estás conectado como {email}", - "AccountAlreadyConnectedHtml": "Ya estás conectado como {email}", - "AccountConnectedHtml": "Te has conectado correctamente como {email}", + "ConnectedDescriptionHtml": "✅Estás conectado como {name}", + "AccountAlreadyConnectedHtml": "Ya estás conectado como {name}", + "AccountConnectedHtml": "Te has conectado correctamente como {name}", "AccountAlreadyConnected": "Cuenta ya conectada", "InvalidCode": "Código no válido", "SomethingWentWrong": "Algo salió mal. Por favor, inténtalo de nuevo.", diff --git a/plugins/telegram-assets/lang/fr.json b/plugins/telegram-assets/lang/fr.json index 05511cbb36..39fa202eef 100644 --- a/plugins/telegram-assets/lang/fr.json +++ b/plugins/telegram-assets/lang/fr.json @@ -46,9 +46,9 @@ "ConnectAccount": "Connecter le compte via OTP", "ShowCommandsDetails": "Afficher les détails des commandes", "TurnNotificationsOff": "Désactiver les notifications", - "ConnectedDescriptionHtml": "✅Vous êtes connecté en tant que {email}", - "AccountAlreadyConnectedHtml": "Vous êtes déjà connecté en tant que {email}", - "AccountConnectedHtml": "Vous avez été connecté avec succès en tant que {email}", + "ConnectedDescriptionHtml": "✅Vous êtes connecté en tant que {name}", + "AccountAlreadyConnectedHtml": "Vous êtes déjà connecté en tant que {name}", + "AccountConnectedHtml": "Vous avez été connecté avec succès en tant que {name}", "AccountAlreadyConnected": "Compte déjà connecté", "InvalidCode": "Code invalide", "SomethingWentWrong": "Quelque chose s'est mal passé. Veuillez réessayer.", diff --git a/plugins/telegram-assets/lang/it.json b/plugins/telegram-assets/lang/it.json index 120ef6ace6..b176d7a9c0 100644 --- a/plugins/telegram-assets/lang/it.json +++ b/plugins/telegram-assets/lang/it.json @@ -46,9 +46,9 @@ "ConnectAccount": "Collega l'account tramite OTP", "ShowCommandsDetails": "Mostra dettagli dei comandi", "TurnNotificationsOff": "Disattiva notifiche", - "ConnectedDescriptionHtml": "✅Sei connesso come {email}", - "AccountAlreadyConnectedHtml": "Sei già connesso come {email}", - "AccountConnectedHtml": "Sei stato connesso con successo come {email}", + "ConnectedDescriptionHtml": "✅Sei connesso come {name}", + "AccountAlreadyConnectedHtml": "Sei già connesso come {name}", + "AccountConnectedHtml": "Sei stato connesso con successo come {name}", "AccountAlreadyConnected": "Account già connesso", "InvalidCode": "Codice non valido", "SomethingWentWrong": "Qualcosa è andato storto. Per favore riprova.", diff --git a/plugins/telegram-assets/lang/pt.json b/plugins/telegram-assets/lang/pt.json index 81f98a50fd..c06d0def80 100644 --- a/plugins/telegram-assets/lang/pt.json +++ b/plugins/telegram-assets/lang/pt.json @@ -46,9 +46,9 @@ "ConnectAccount": "Ligar conta via OTP", "ShowCommandsDetails": "Mostrar detalhes dos comandos", "TurnNotificationsOff": "Desativar notificações", - "ConnectedDescriptionHtml": "✅Está ligado como {email}", - "AccountAlreadyConnectedHtml": "Já está ligado como {email}", - "AccountConnectedHtml": "Foi ligado com sucesso como {email}", + "ConnectedDescriptionHtml": "✅Está ligado como {name}", + "AccountAlreadyConnectedHtml": "Já está ligado como {name}", + "AccountConnectedHtml": "Foi ligado com sucesso como {name}", "AccountAlreadyConnected": "Conta já ligada", "InvalidCode": "Código inválido", "SomethingWentWrong": "Algo correu mal. Por favor, tente novamente.", diff --git a/plugins/telegram-assets/lang/ru.json b/plugins/telegram-assets/lang/ru.json index 11a942f599..f3b65c5986 100644 --- a/plugins/telegram-assets/lang/ru.json +++ b/plugins/telegram-assets/lang/ru.json @@ -46,9 +46,9 @@ "ConnectAccount": "Подключить аккаунт через OTP", "ShowCommandsDetails": "Показать информацию о командах", "TurnNotificationsOff": "Отключить уведомления", - "ConnectedDescriptionHtml": "✅Вы подключены как ${email}", - "AccountAlreadyConnectedHtml": "Вы уже подключены как {email}", - "AccountConnectedHtml": "Вы успешно подключены как {email}", + "ConnectedDescriptionHtml": "✅Вы подключены как ${name}", + "AccountAlreadyConnectedHtml": "Вы уже подключены как {name}", + "AccountConnectedHtml": "Вы успешно подключены как {name}", "AccountAlreadyConnected": "Аккаунт уже подключен", "InvalidCode": "Неверный код", "SomethingWentWrong": "Что-то пошло не так. Пожалуйста, попробуйте снова.", diff --git a/plugins/telegram-assets/lang/zh.json b/plugins/telegram-assets/lang/zh.json index aa05eeabff..a3b4590eca 100644 --- a/plugins/telegram-assets/lang/zh.json +++ b/plugins/telegram-assets/lang/zh.json @@ -46,9 +46,9 @@ "ConnectAccount": "通过 OTP 连接帐户", "ShowCommandsDetails": "显示命令详细信息", "TurnNotificationsOff": "关闭通知", - "ConnectedDescriptionHtml": "✅您已连接为 {email}", - "AccountAlreadyConnectedHtml": "您已连接为 {email}", - "AccountConnectedHtml": "您已成功连接为 {email}", + "ConnectedDescriptionHtml": "✅您已连接为 {name}", + "AccountAlreadyConnectedHtml": "您已连接为 {name}", + "AccountConnectedHtml": "您已成功连接为 {name}", "AccountAlreadyConnected": "帐户已连接", "InvalidCode": "无效代码", "SomethingWentWrong": "出现问题。 请重试。", diff --git a/plugins/telegram-resources/src/components/ConfigureBotPopup.svelte b/plugins/telegram-resources/src/components/ConfigureBotPopup.svelte index 6509b8264d..78e3ddc556 100644 --- a/plugins/telegram-resources/src/components/ConfigureBotPopup.svelte +++ b/plugins/telegram-resources/src/components/ConfigureBotPopup.svelte @@ -27,7 +27,7 @@ let isConnectionEstablished = false let connectionError: Error | undefined - let info: { name: string, username: string, photoUrl: string } | undefined = undefined + let info: { name: string, username: string, photoId: string } | undefined = undefined let isLoading = false const url = getMetadata(telegram.metadata.BotUrl) ?? '' @@ -134,8 +134,8 @@
- {#if info.photoUrl !== ''} - + {#if info.photoId !== ''} + {:else} {/if} diff --git a/plugins/telegram-resources/src/components/NotificationProviderPresenter.svelte b/plugins/telegram-resources/src/components/NotificationProviderPresenter.svelte index aa96d2361a..74e1ccdc91 100644 --- a/plugins/telegram-resources/src/components/NotificationProviderPresenter.svelte +++ b/plugins/telegram-resources/src/components/NotificationProviderPresenter.svelte @@ -16,37 +16,14 @@ {#if enabled} diff --git a/plugins/telegram/src/index.ts b/plugins/telegram/src/index.ts index de471a8e6c..77baff626c 100644 --- a/plugins/telegram/src/index.ts +++ b/plugins/telegram/src/index.ts @@ -13,10 +13,9 @@ // limitations under the License. // -import { ActivityMessage } from '@hcengineering/activity' import { ChannelItem } from '@hcengineering/contact' -import { PersonId, AttachedDoc, Class, Doc, Ref, Timestamp } from '@hcengineering/core' -import { InboxNotification, NotificationProvider, NotificationType } from '@hcengineering/notification' +import { AttachedDoc, Class, Doc, Ref, Timestamp } from '@hcengineering/core' +import { NotificationProvider, NotificationType } from '@hcengineering/notification' import type { IntlString, Metadata, Plugin } from '@hcengineering/platform' import { plugin } from '@hcengineering/platform' import type { Handler, IntegrationType } from '@hcengineering/setting' @@ -59,19 +58,6 @@ export interface SharedTelegramMessages extends AttachedDoc { messages: SharedTelegramMessage[] } -export interface TelegramNotificationRequest { - notificationId: Ref - messageId?: Ref - attachments: boolean - workspace: string - account: PersonId - title: string - quote?: string - body: string - sender: string - link: string -} - /** * @public */ diff --git a/plugins/workbench-resources/src/connect.ts b/plugins/workbench-resources/src/connect.ts index 9f6c41fc37..60babe34fd 100644 --- a/plugins/workbench-resources/src/connect.ts +++ b/plugins/workbench-resources/src/connect.ts @@ -367,7 +367,7 @@ export async function connect (title: string): Promise { Analytics.setUser(account) Analytics.setTag('workspace', wsUrl) - console.log(`Logged in with account ${me.uuid} as ${me.role}`) + console.log('Logged in with account: ', me) setCurrentAccount(me) setCurrentEmployee(employee) diff --git a/pods/server/src/__start.ts b/pods/server/src/__start.ts index 8ff3c653e4..183f211068 100644 --- a/pods/server/src/__start.ts +++ b/pods/server/src/__start.ts @@ -25,7 +25,6 @@ import serverCore, { } from '@hcengineering/server-core' import serverNotification from '@hcengineering/server-notification' import { storageConfigFromEnv } from '@hcengineering/server-storage' -import serverTelegram from '@hcengineering/server-telegram' import serverToken from '@hcengineering/server-token' import { join } from 'path' import { start } from '.' @@ -86,7 +85,6 @@ setMetadata(serverToken.metadata.Secret, config.serverSecret) setMetadata(serverNotification.metadata.MailUrl, config.mailUrl ?? '') setMetadata(serverNotification.metadata.MailAuthToken, config.mailAuthToken) setMetadata(serverNotification.metadata.WebPushUrl, config.webPushUrl) -setMetadata(serverTelegram.metadata.BotUrl, process.env.TELEGRAM_BOT_URL) setMetadata(serverAiBot.metadata.EndpointURL, process.env.AI_BOT_URL) setMetadata(serverCalendar.metadata.EndpointURL, process.env.CALENDAR_URL) diff --git a/server-plugins/activity-resources/src/references.ts b/server-plugins/activity-resources/src/references.ts index 5cf0c83ab4..b1a7544b35 100644 --- a/server-plugins/activity-resources/src/references.ts +++ b/server-plugins/activity-resources/src/references.ts @@ -21,7 +21,6 @@ import core, { Class, Data, Doc, - generateId, Hierarchy, Markup, Ref, @@ -38,13 +37,11 @@ import core, { type MeasureContext, AccountUuid } from '@hcengineering/core' -import notification, { CommonInboxNotification, MentionInboxNotification } from '@hcengineering/notification' +import notification, { MentionInboxNotification } from '@hcengineering/notification' import { getPerson } from '@hcengineering/server-contact' import { StorageAdapter, TriggerControl } from '@hcengineering/server-core' import { - applyNotificationProviders, getCommonNotificationTxes, - getNotificationContent, getNotificationProviderControl, getPushCollaboratorTx, NotifyResult, @@ -213,43 +210,6 @@ export async function getPersonNotificationTxes ( originTx ) res.push(...txes) - } else { - const context = ( - await control.findAll( - ctx, - notification.class.DocNotifyContext, - { objectId: reference.srcDocId, user: receiverAccount }, - { projection: { _id: 1 } } - ) - )[0] - if (context !== undefined) { - const content = await getNotificationContent(originTx, receiverPersonRef, sender, doc, control) - const notificationData: CommonInboxNotification = { - ...data, - ...content, - docNotifyContext: context._id, - _id: generateId(), - _class: notification.class.MentionInboxNotification, - space: receiverSpace._id, - modifiedOn: originTx.modifiedOn, - modifiedBy: senderId - } - - const msg = control.hierarchy.isDerived(data.mentionedInClass, activity.class.ActivityMessage) - ? (await control.findAll(control.ctx, data.mentionedInClass, { _id: data.mentionedIn }))[0] - : undefined - await applyNotificationProviders( - notificationData, - notifyResult, - control, - res, - doc, - receiver, - sender, - notification.class.MentionInboxNotification, - msg as ActivityMessage - ) - } } return res diff --git a/server-plugins/contact/src/utils.ts b/server-plugins/contact/src/utils.ts index c209d39eae..b07ebafda7 100644 --- a/server-plugins/contact/src/utils.ts +++ b/server-plugins/contact/src/utils.ts @@ -81,6 +81,15 @@ export async function getAllSocialStringsByPersonId ( } export async function getPerson (control: TriggerControl, personId: PersonId): Promise { + const { contextData } = control.ctx + const account: AccountUuid | undefined = contextData.account.socialIds.includes(personId) + ? contextData.account.uuid + : contextData.socialStringsToUsers.get(personId) + + if (account !== undefined) { + return (await control.findAll(control.ctx, contact.class.Person, { personUuid: account }))[0] + } + const socialId = ( await control.findAll(control.ctx, contact.class.SocialIdentity, { _id: personId as SocialIdentityRef }) )[0] diff --git a/server-plugins/gmail-resources/src/index.ts b/server-plugins/gmail-resources/src/index.ts index 49f92f808a..71f42678d7 100644 --- a/server-plugins/gmail-resources/src/index.ts +++ b/server-plugins/gmail-resources/src/index.ts @@ -32,11 +32,7 @@ import { import gmail, { Message } from '@hcengineering/gmail' import { TriggerControl } from '@hcengineering/server-core' import { BaseNotificationType, InboxNotification, NotificationType } from '@hcengineering/notification' -import serverNotification, { - NotificationProviderFunc, - ReceiverInfo, - SenderInfo -} from '@hcengineering/server-notification' +import serverNotification, { ReceiverInfo, SenderInfo } from '@hcengineering/server-notification' import { getContentByTemplate } from '@hcengineering/server-notification-resources' import { getMetadata } from '@hcengineering/platform' import { ActivityMessage } from '@hcengineering/activity' @@ -162,7 +158,7 @@ async function notifyByEmail ( // } } -const SendEmailNotifications: NotificationProviderFunc = async ( +const SendEmailNotifications = async ( control: TriggerControl, types: BaseNotificationType[], object: Doc, diff --git a/server-plugins/gmail/src/index.ts b/server-plugins/gmail/src/index.ts index 98eab2f0b9..1ba7fbb620 100644 --- a/server-plugins/gmail/src/index.ts +++ b/server-plugins/gmail/src/index.ts @@ -17,7 +17,7 @@ import type { Plugin, Resource } from '@hcengineering/platform' import { plugin } from '@hcengineering/platform' import { ObjectDDParticipantFunc, TriggerFunc } from '@hcengineering/server-core' -import { NotificationProviderFunc, TypeMatchFunc } from '@hcengineering/server-notification' +import { TypeMatchFunc } from '@hcengineering/server-notification' /** * @public @@ -33,7 +33,6 @@ export default plugin(serverGmailId, { }, function: { IsIncomingMessageTypeMatch: '' as TypeMatchFunc, - FindMessages: '' as Resource, - SendEmailNotifications: '' as Resource + FindMessages: '' as Resource } }) diff --git a/server-plugins/notification-resources/src/index.ts b/server-plugins/notification-resources/src/index.ts index bc4d433cfd..f1f4b45ecc 100644 --- a/server-plugins/notification-resources/src/index.ts +++ b/server-plugins/notification-resources/src/index.ts @@ -60,11 +60,7 @@ import notification, { } from '@hcengineering/notification' import { getResource, translate } from '@hcengineering/platform' import { type TriggerControl } from '@hcengineering/server-core' -import serverNotification, { - NOTIFICATION_BODY_SIZE, - ReceiverInfo, - SenderInfo -} from '@hcengineering/server-notification' +import { NOTIFICATION_BODY_SIZE, ReceiverInfo, SenderInfo } from '@hcengineering/server-notification' import { markupToText, stripTags } from '@hcengineering/text-core' import { Analytics } from '@hcengineering/analytics' import { getAccountBySocialId, getEmployeesBySocialIds } from '@hcengineering/server-contact' @@ -140,7 +136,7 @@ export async function getCommonNotificationTxes ( const res: Tx[] = [] const notifyContexts = await control.findAll(ctx, notification.class.DocNotifyContext, { objectId: attachedTo }) - const notificationTx = await pushInboxNotifications( + await pushInboxNotifications( ctx, control, res, @@ -157,12 +153,6 @@ export async function getCommonNotificationTxes ( tx ) - if (notificationTx !== undefined) { - const notificationData = TxProcessor.createDoc2Doc(notificationTx) - - await applyNotificationProviders(notificationData, notifyResult, control, res, doc, receiver, sender, _class) - } - return res } @@ -534,32 +524,6 @@ export async function pushActivityInboxNotifications ( ) } -export async function applyNotificationProviders ( - data: InboxNotification, - notifyResult: NotifyResult, - control: TriggerControl, - res: Tx[], - object: Doc, - receiver: ReceiverInfo, - sender: SenderInfo, - _class = notification.class.ActivityInboxNotification, - message?: ActivityMessage -): Promise { - const resources = control.modelDb.findAllSync(serverNotification.class.NotificationProviderResources, {}) - for (const [provider, types] of notifyResult.entries()) { - const resource = resources.find((it) => it.provider === provider) - - if (resource === undefined) continue - - const fn = await getResource(resource.fn) - - const txes = await fn(control, types, object, data, receiver, sender, message) - if (txes.length > 0) { - res.push(...txes) - } - } -} - async function createNotifyContext ( ctx: MeasureContext, control: TriggerControl, @@ -651,28 +615,12 @@ export async function getNotificationTxes ( ) if (notificationTx !== undefined) { - const notificationData = TxProcessor.createDoc2Doc(notificationTx) - const current: AvailableProvidersCache = control.contextCache.get(AvailableProvidersCacheKey) ?? new Map() - const providers = Array.from(notifyResult.keys()).filter( - (p) => p !== notification.providers.InboxNotificationProvider - ) + const providers = Array.from(notifyResult.keys()) if (providers.length > 0) { - current.set(notificationData._id, providers) + current.set(notificationTx.objectId, providers) control.contextCache.set('AvailableNotificationProviders', current) } - - await applyNotificationProviders( - notificationData, - notifyResult, - control, - res, - object, - receiver, - sender, - notificationData._class, - message - ) } } else { const context = getDocNotifyContext(control, docNotifyContexts, message.attachedTo, receiver.account) diff --git a/server-plugins/notification-resources/src/utils.ts b/server-plugins/notification-resources/src/utils.ts index aac014aaa0..92c4f01610 100644 --- a/server-plugins/notification-resources/src/utils.ts +++ b/server-plugins/notification-resources/src/utils.ts @@ -82,10 +82,10 @@ export function isUserEmployeeInFieldValueTypeMatch ( type: NotificationType, control: TriggerControl ): boolean { - // TODO: check field type and compare with Ref or PersonId based on that if (type.field === undefined) return false const value = (doc as any)[type.field] if (value == null) return false + if (value === person) return true if (Array.isArray(value)) { return includesAny(value, socialIds) @@ -121,7 +121,7 @@ function escapeRegExp (str: string): string { export async function shouldNotifyCommon ( control: TriggerControl, - user: PersonId[], + socialIds: PersonId[], typeId: Ref, notificationControl: NotificationProviderControl ): Promise { @@ -135,7 +135,7 @@ export async function shouldNotifyCommon ( const providers = await control.modelDb.findAll(notification.class.NotificationProvider, {}) for (const provider of providers) { - const allowed = isAllowed(control, user, type, provider, notificationControl) + const allowed = isAllowed(control, socialIds, type, provider, notificationControl) if (allowed) { const cur = result.get(provider._id) ?? [] diff --git a/server-plugins/notification/src/index.ts b/server-plugins/notification/src/index.ts index 08c33ba972..8ccd112108 100644 --- a/server-plugins/notification/src/index.ts +++ b/server-plugins/notification/src/index.ts @@ -14,16 +14,9 @@ // limitations under the License. // -import { ActivityMessage } from '@hcengineering/activity' import { Employee, Person, PersonSpace } from '@hcengineering/contact' import { PersonId, Class, Doc, Mixin, Ref, Tx, TxCUD, AccountUuid } from '@hcengineering/core' -import { - BaseNotificationType, - InboxNotification, - NotificationContent, - NotificationProvider, - NotificationType -} from '@hcengineering/notification' +import { NotificationContent, NotificationType } from '@hcengineering/notification' import { Metadata, Plugin, Resource, plugin } from '@hcengineering/platform' import type { TriggerControl, TriggerFunc } from '@hcengineering/server-core' @@ -102,21 +95,6 @@ export interface SenderInfo { person?: Person } -export type NotificationProviderFunc = ( - control: TriggerControl, - types: BaseNotificationType[], - object: Doc, - data: InboxNotification, - receiver: ReceiverInfo, - sender: SenderInfo, - message?: ActivityMessage -) => Promise - -export interface NotificationProviderResources extends Doc { - provider: Ref - fn: Resource -} - export const NOTIFICATION_BODY_SIZE = 50 export const PUSH_NOTIFICATION_TITLE_SIZE = 80 @@ -130,9 +108,6 @@ export default plugin(serverNotificationId, { WebPushUrl: '' as Metadata, InboxOnlyNotifications: '' as Metadata }, - class: { - NotificationProviderResources: '' as Ref> - }, mixin: { HTMLPresenter: '' as Ref>, TextPresenter: '' as Ref>, diff --git a/server-plugins/telegram-resources/package.json b/server-plugins/telegram-resources/package.json index 8bbf4d734d..eae97794e4 100644 --- a/server-plugins/telegram-resources/package.json +++ b/server-plugins/telegram-resources/package.json @@ -43,6 +43,7 @@ "@hcengineering/core": "^0.6.32", "@hcengineering/notification": "^0.6.23", "@hcengineering/platform": "^0.6.11", + "@hcengineering/server-contact": "^0.6.1", "@hcengineering/server-core": "^0.6.1", "@hcengineering/server-notification": "^0.6.1", "@hcengineering/server-notification-resources": "^0.6.0", diff --git a/server-plugins/telegram-resources/src/index.ts b/server-plugins/telegram-resources/src/index.ts index 271775c91d..78e1c87729 100644 --- a/server-plugins/telegram-resources/src/index.ts +++ b/server-plugins/telegram-resources/src/index.ts @@ -12,14 +12,12 @@ // See the License for the specific language governing permissions and // limitations under the License. // -/* eslint-disable @typescript-eslint/no-unused-vars */ import activity, { ActivityMessage, DocUpdateMessage } from '@hcengineering/activity' import chunter, { ChatMessage } from '@hcengineering/chunter' -import contact, { Channel, ChannelProvider, Contact, Employee, formatName, Person } from '@hcengineering/contact' -import { +import contact, { Channel, Person } from '@hcengineering/contact' +import core, { PersonId, Class, - concatLink, Doc, DocumentQuery, FindOptions, @@ -28,26 +26,33 @@ import { Ref, Tx, TxCreateDoc, - TxProcessor + TxProcessor, + TxCUD, + TxUpdateDoc } from '@hcengineering/core' import notification, { - BaseNotificationType, + ActivityInboxNotification, InboxNotification, MentionInboxNotification, + NotificationProviderSetting, NotificationType } from '@hcengineering/notification' -import { getMetadata, getResource, translate } from '@hcengineering/platform' -import { TriggerControl } from '@hcengineering/server-core' -import { NotificationProviderFunc, ReceiverInfo, SenderInfo } from '@hcengineering/server-notification' +import { getResource, translate } from '@hcengineering/platform' +import { getAccountBySocialId, getPerson } from '@hcengineering/server-contact' +import { PlatformQueueProducer, QueueTopic, TriggerControl } from '@hcengineering/server-core' import { getNotificationLink, getTextPresenter, - getTranslatedNotificationContent + getTranslatedNotificationContent, + AvailableProvidersCache, + AvailableProvidersCacheKey } from '@hcengineering/server-notification-resources' -import serverTelegram from '@hcengineering/server-telegram' -import { generateToken } from '@hcengineering/server-token' -import setting, { Integration } from '@hcengineering/setting' -import telegram, { TelegramMessage, TelegramNotificationRequest } from '@hcengineering/telegram' +import { + type TelegramNotificationQueueMessage, + TelegramQueueMessageType, + TelegramQueueMessage +} from '@hcengineering/server-telegram' +import telegram, { TelegramMessage } from '@hcengineering/telegram' import { jsonToHTML, markupToJSON } from '@hcengineering/text' /** @@ -149,20 +154,20 @@ export async function GetIntegrationOwnerTG ( return undefined } -async function getContactChannel ( - control: TriggerControl, - value: Contact, - provider: Ref -): Promise { - if (value === undefined) return - const res = ( - await control.findAll(control.ctx, contact.class.Channel, { - attachedTo: value._id, - provider - }) - )[0] - return res?.value ?? '' -} +// async function getContactChannel ( +// control: TriggerControl, +// value: Contact, +// provider: Ref +// ): Promise { +// if (value === undefined) return +// const res = ( +// await control.findAll(control.ctx, contact.class.Channel, { +// attachedTo: value._id, +// provider +// }) +// )[0] +// return res?.value ?? '' +// } async function activityMessageToHtml (control: TriggerControl, message: ActivityMessage): Promise { const { hierarchy } = control @@ -252,66 +257,185 @@ function hasAttachments (doc: ActivityMessage | undefined, hierarchy: Hierarchy) return false } -const telegramNotificationKey = 'telegram.notification.reported' -const SendTelegramNotifications: NotificationProviderFunc = async ( +const telegramNotificationCacheKey = 'telegram.notification.cache' + +async function NotificationsHandler (txes: TxCreateDoc[], control: TriggerControl): Promise { + const queue = control.queue + + if (queue === undefined) { + return [] + } + + const availableProviders: AvailableProvidersCache = control.contextCache.get(AvailableProvidersCacheKey) ?? new Map() + + const all: InboxNotification[] = txes + .map((tx) => TxProcessor.createDoc2Doc(tx)) + .filter( + (it) => + availableProviders.get(it._id)?.find((p) => p === telegram.providers.TelegramNotificationProvider) !== undefined + ) + + if (all.length === 0) { + return [] + } + + const result: Tx[] = [] + const producer = queue.createProducer(control.ctx, QueueTopic.TelegramBot) + try { + for (const inboxNotification of all) { + result.push(...(await processNotification(inboxNotification, control, producer))) + } + } finally { + await producer.close() + } + return result +} + +async function getNotificationMessage ( + n: InboxNotification, control: TriggerControl, - types: BaseNotificationType[], - doc: Doc, - data: InboxNotification, - receiver: ReceiverInfo, - sender: SenderInfo, - message?: ActivityMessage -): Promise => { - // TODO: FIXME - // if (types.length === 0) { - // return [] - // } + cache: Map, Doc> +): Promise { + const { hierarchy } = control + if (hierarchy.isDerived(n._class, notification.class.ActivityInboxNotification)) { + const activityNotification = n as ActivityInboxNotification + const message = + cache.get(activityNotification.attachedTo) ?? + ( + await control.findAll(control.ctx, activityNotification.attachedToClass, { + _id: activityNotification.attachedTo + }) + )[0] + return message as ActivityMessage + } else if (hierarchy.isDerived(n._class, notification.class.MentionInboxNotification)) { + const mentionNotification = n as MentionInboxNotification + if (hierarchy.isDerived(mentionNotification.mentionedInClass, activity.class.ActivityMessage)) { + const message = + cache.get(mentionNotification.mentionedIn) ?? + ( + await control.findAll(control.ctx, mentionNotification.mentionedInClass, { + _id: mentionNotification.mentionedIn + }) + )[0] + return message as ActivityMessage + } + } - // const botUrl = getMetadata(serverTelegram.metadata.BotUrl) + return undefined +} - // if (botUrl === undefined || botUrl === '') { - // const reported = control.cache.get(telegramNotificationKey) - // if (reported === undefined) { - // control.ctx.error('Please provide telegram bot service url to enable telegram notifications.') - // control.cache.set(telegramNotificationKey, true) - // } - // return [] - // } +async function getSenderName (n: InboxNotification, control: TriggerControl): Promise { + const inlineName = n.intlParams?.senderName + if (inlineName != null && inlineName !== '') { + return inlineName.toString() + } + const senderPerson = await getPerson(control, n.createdBy ?? n.modifiedBy) + return senderPerson?.name ?? 'System' +} - // if (!receiver.person.active) { - // return [] - // } +async function processNotification ( + n: InboxNotification, + control: TriggerControl, + producer: PlatformQueueProducer +): Promise { + try { + const cache: Map, Doc> = control.contextCache.get(telegramNotificationCacheKey) ?? new Map() + const doc = cache.get(n.objectId) ?? (await control.findAll(control.ctx, n.objectClass, { _id: n.objectId }))[0] + if (doc === undefined) return [] + const message = await getNotificationMessage(n, control, cache) - // try { - // const { title, body, quote, link } = await getTranslatedData(data, doc, control, message) - // const record: TelegramNotificationRequest = { - // notificationId: data._id, - // messageId: message?._id, - // account: receiver._id, - // workspace: control.workspace, - // sender: data.intlParams?.senderName?.toString() ?? formatName(sender.person?.name ?? 'System'), - // attachments: hasAttachments(message, control.hierarchy), - // title, - // quote, - // body, - // link - // } + cache.set(n.objectId, doc) + control.contextCache.set(telegramNotificationCacheKey, cache) - // await fetch(concatLink(botUrl, '/notify'), { - // method: 'POST', - // headers: { - // Authorization: 'Bearer ' + generateToken(receiver.account.email, control.workspace, { service: 'telegram' }), - // 'Content-Type': 'application/json' - // }, - // body: JSON.stringify([record]) - // }) - // } catch (err) { - // control.ctx.error('Could not send telegram notification', { - // err, - // notificationId: data._id, - // receiver: receiver.account.email - // }) - // } + const { title, body, quote, link } = await getTranslatedData(n, doc, control, message) + const record: TelegramNotificationQueueMessage = { + type: TelegramQueueMessageType.Notification, + notificationId: n._id, + messageId: message?._id, + account: n.user, + sender: await getSenderName(n, control), + attachments: hasAttachments(message, control.hierarchy), + title, + quote, + body, + link + } + + await producer.send(control.workspace.uuid, [record]) + } catch (err) { + control.ctx.error('Could not send telegram notification', { + err, + notificationId: n._id, + account: n.user + }) + } + + return [] +} + +async function updateWorkspaceSubscription ( + producer: PlatformQueueProducer, + enabled: boolean, + socialId: PersonId, + control: TriggerControl +): Promise { + const account = await getAccountBySocialId(control, socialId) + if (account == null) { + return + } + await producer.send(control.workspace.uuid, [ + { + type: TelegramQueueMessageType.WorkspaceSubscription, + account, + subscribe: enabled + } + ]) +} + +async function ProviderSettingsHandler ( + txes: TxCUD[], + control: TriggerControl +): Promise { + const queue = control.queue + + if (queue === undefined) { + return [] + } + + const producer = queue.createProducer(control.ctx, QueueTopic.TelegramBot) + + try { + for (const tx of txes) { + if (tx._class === core.class.TxCreateDoc) { + const createTx = tx as TxCreateDoc + const setting = TxProcessor.createDoc2Doc(createTx) + + if (setting.attachedTo === telegram.providers.TelegramNotificationProvider) { + await updateWorkspaceSubscription(producer, setting.enabled, setting.createdBy ?? setting.modifiedBy, control) + } + } else if (tx._class === core.class.TxUpdateDoc) { + const updateTx = tx as TxUpdateDoc + if (updateTx.operations.enabled !== undefined) { + const setting = ( + await control.findAll(control.ctx, notification.class.NotificationProviderSetting, { + _id: updateTx.objectId + }) + )[0] + + if (setting !== undefined && setting.attachedTo === telegram.providers.TelegramNotificationProvider) { + await updateWorkspaceSubscription( + producer, + updateTx.operations.enabled, + setting.createdBy ?? setting.modifiedBy, + control + ) + } + } + } + } + } finally { + await producer.close() + } return [] } @@ -319,13 +443,14 @@ const SendTelegramNotifications: NotificationProviderFunc = async ( // eslint-disable-next-line @typescript-eslint/explicit-function-return-type export default async () => ({ trigger: { - OnMessageCreate + OnMessageCreate, + NotificationsHandler, + ProviderSettingsHandler }, function: { IsIncomingMessageTypeMatch, FindMessages, GetCurrentEmployeeTG, - GetIntegrationOwnerTG, - SendTelegramNotifications + GetIntegrationOwnerTG } }) diff --git a/server-plugins/telegram/package.json b/server-plugins/telegram/package.json index 7e801af6a8..7711bf7199 100644 --- a/server-plugins/telegram/package.json +++ b/server-plugins/telegram/package.json @@ -38,10 +38,12 @@ "@types/jest": "^29.5.5" }, "dependencies": { + "@hcengineering/activity": "^0.6.0", "@hcengineering/core": "^0.6.32", + "@hcengineering/notification": "^0.6.23", "@hcengineering/platform": "^0.6.11", - "@hcengineering/server-templates": "^0.6.0", + "@hcengineering/server-core": "^0.6.1", "@hcengineering/server-notification": "^0.6.1", - "@hcengineering/server-core": "^0.6.1" + "@hcengineering/server-templates": "^0.6.0" } } diff --git a/server-plugins/telegram/src/index.ts b/server-plugins/telegram/src/index.ts index 9a675102a1..7d34a50ae4 100644 --- a/server-plugins/telegram/src/index.ts +++ b/server-plugins/telegram/src/index.ts @@ -14,11 +14,12 @@ // limitations under the License. // -import { Metadata, Plugin, Resource, plugin } from '@hcengineering/platform' +import { Plugin, Resource, plugin } from '@hcengineering/platform' import { ObjectDDParticipantFunc, TriggerFunc } from '@hcengineering/server-core' -import { NotificationProviderFunc, TypeMatchFunc } from '@hcengineering/server-notification' +import { TypeMatchFunc } from '@hcengineering/server-notification' import { TemplateFieldServerFunc } from '@hcengineering/server-templates' +export * from './types' /** * @public */ @@ -28,17 +29,15 @@ export const serverTelegramId = 'server-telegram' as Plugin * @public */ export default plugin(serverTelegramId, { - metadata: { - BotUrl: '' as Metadata - }, trigger: { - OnMessageCreate: '' as Resource + OnMessageCreate: '' as Resource, + NotificationsHandler: '' as Resource, + ProviderSettingsHandler: '' as Resource }, function: { IsIncomingMessageTypeMatch: '' as TypeMatchFunc, FindMessages: '' as Resource, GetCurrentEmployeeTG: '' as Resource, - GetIntegrationOwnerTG: '' as Resource, - SendTelegramNotifications: '' as Resource + GetIntegrationOwnerTG: '' as Resource } }) diff --git a/server-plugins/telegram/src/types.ts b/server-plugins/telegram/src/types.ts new file mode 100644 index 0000000000..c7350fb055 --- /dev/null +++ b/server-plugins/telegram/src/types.ts @@ -0,0 +1,44 @@ +// +// Copyright © 2025 Hardcore Engineering Inc. +// +// Licensed under the Eclipse Public License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. You may +// obtain a copy of the License at https://www.eclipse.org/legal/epl-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// +// See the License for the specific language governing permissions and +// limitations under the License. +// + +import { ActivityMessage } from '@hcengineering/activity' +import { AccountUuid, Ref } from '@hcengineering/core' +import { InboxNotification } from '@hcengineering/notification' + +export type TelegramQueueMessage = TelegramNotificationQueueMessage | TelegramWorkspaceSubscriptionQueueMessage + +export enum TelegramQueueMessageType { + Notification = 'notification', + WorkspaceSubscription = 'workspaceSubscription' +} + +export interface TelegramNotificationQueueMessage { + type: TelegramQueueMessageType.Notification + notificationId: Ref + messageId?: Ref + attachments: boolean + account: AccountUuid + title: string + quote?: string + body: string + sender: string + link: string +} + +export interface TelegramWorkspaceSubscriptionQueueMessage { + type: TelegramQueueMessageType.WorkspaceSubscription + account: AccountUuid + subscribe: boolean +} diff --git a/server/account/src/__tests__/serviceOperations.test.ts b/server/account/src/__tests__/serviceOperations.test.ts index d8d9827a32..57cafe2d8d 100644 --- a/server/account/src/__tests__/serviceOperations.test.ts +++ b/server/account/src/__tests__/serviceOperations.test.ts @@ -82,13 +82,21 @@ describe('addSocialIdToPerson', () => { person: 'test-person' as PersonUuid, type: SocialIdType.GITHUB, value: 'test-value', - confirmed: true + confirmed: true, + displayValue: 'test-display-value' } const result = await addSocialIdToPerson(mockCtx, mockDb, mockBranding, mockToken, params) expect(result).toBe(newSocialId) - expect(addSocialIdSpy).toHaveBeenCalledWith(mockDb, params.person, params.type, params.value, params.confirmed) + expect(addSocialIdSpy).toHaveBeenCalledWith( + mockDb, + params.person, + params.type, + params.value, + params.confirmed, + params.displayValue + ) }) test('should allow admin to add social id', async () => { @@ -102,13 +110,21 @@ describe('addSocialIdToPerson', () => { person: 'test-person' as PersonUuid, type: SocialIdType.GITHUB, value: 'test-value', - confirmed: false + confirmed: false, + displayValue: 'test-display-value' } const result = await addSocialIdToPerson(mockCtx, mockDb, mockBranding, mockToken, params) expect(result).toBe(newSocialId) - expect(addSocialIdSpy).toHaveBeenCalledWith(mockDb, params.person, params.type, params.value, params.confirmed) + expect(addSocialIdSpy).toHaveBeenCalledWith( + mockDb, + params.person, + params.type, + params.value, + params.confirmed, + params.displayValue + ) }) test('should throw error for unauthorized service', async () => { diff --git a/server/account/src/collections/postgres.ts b/server/account/src/collections/postgres.ts index a91c6f1e35..1b45f53845 100644 --- a/server/account/src/collections/postgres.ts +++ b/server/account/src/collections/postgres.ts @@ -676,7 +676,8 @@ export class PostgresAccountDB implements AccountDB { this.getV4Migration(), this.getV4Migration1(), this.getV5Migration(), - this.getV6Migration() + this.getV6Migration(), + this.getV7Migration() ] } @@ -960,4 +961,14 @@ export class PostgresAccountDB implements AccountDB { ` ] } + + private getV7Migration (): [string, string] { + return [ + 'account_db_v7_add_display_value', + ` + ALTER TABLE ${this.ns}.social_id + ADD COLUMN IF NOT EXISTS display_value TEXT; + ` + ] + } } diff --git a/server/account/src/operations.ts b/server/account/src/operations.ts index 14ce03d890..758728c165 100644 --- a/server/account/src/operations.ts +++ b/server/account/src/operations.ts @@ -14,19 +14,19 @@ // import { Analytics } from '@hcengineering/analytics' import { - AccountRole, AccountInfo, + AccountRole, + type Branding, buildSocialIdString, concatLink, isActiveMode, isWorkspaceCreating, MeasureContext, - SocialIdType, - systemAccountUuid, - type Branding, type Person, type PersonId, type PersonUuid, + SocialIdType, + systemAccountUuid, type WorkspaceMemberInfo, type WorkspaceUuid, type AccountUuid @@ -40,9 +40,9 @@ import type { AccountDB, AccountMethodHandler, LoginInfo, - Meta, Mailbox, MailboxOptions, + Meta, OtpInfo, RegionInfo, SocialId, @@ -51,6 +51,7 @@ import type { WorkspaceLoginInfo } from './types' import { + addSocialId, checkInvite, cleanEmail, confirmEmail, @@ -58,40 +59,39 @@ import { createWorkspaceRecord, doJoinByInvite, EndpointKind, + generatePassword, getAccount, getEmailSocialId, getEndpoint, getFrontUrl, getInviteEmail, + getMailUrl, getPersonName, getRegions, getRolePower, - getMailUrl, getWorkspaceById, getWorkspaceInfoWithStatusById, getWorkspaceInvite, + getWorkspaceRole, GUEST_ACCOUNT, + isEmail, isOtpValid, + normalizeValue, + releaseSocialId, selectWorkspace, sendEmail, sendEmailConfirmation, sendOtp, setPassword, + setTimezoneIfNotDefined, signUpByEmail, - verifyAllowedServices, + updateWorkspaceRole, verifyAllowedRole, + verifyAllowedServices, verifyPassword, wrap, - getWorkspaceRole, - normalizeValue, - isEmail, - generatePassword, - addSocialId, - releaseSocialId, - updateWorkspaceRole, - setTimezoneIfNotDefined, - confirmHulyIds, - getWorkspaceByUrl + getWorkspaceByUrl, + confirmHulyIds } from './utils' import { type AccountServiceMethods, getServiceMethods } from './serviceOperations' @@ -1702,6 +1702,7 @@ export type AccountMethods = | 'findPersonBySocialKey' | 'findPersonBySocialId' | 'findSocialIdBySocialKey' + | 'findFullSocialIdBySocialKey' | 'ensurePerson' | 'exchangeGuestToken' | 'getMailboxOptions' @@ -1709,6 +1710,7 @@ export type AccountMethods = | 'getMailboxes' | 'deleteMailbox' | 'addSocialIdToPerson' + | 'updateSocialId' | 'getAccountInfo' /** diff --git a/server/account/src/serviceOperations.ts b/server/account/src/serviceOperations.ts index 967122b1f2..1022697223 100644 --- a/server/account/src/serviceOperations.ts +++ b/server/account/src/serviceOperations.ts @@ -39,6 +39,7 @@ import type { IntegrationKey, IntegrationSecret, IntegrationSecretKey, + SocialId, Workspace, WorkspaceEvent, WorkspaceInfoWithStatus, @@ -520,14 +521,34 @@ export async function addSocialIdToPerson ( db: AccountDB, branding: Branding | null, token: string, - params: { person: PersonUuid, type: SocialIdType, value: string, confirmed: boolean } + params: { person: PersonUuid, type: SocialIdType, value: string, confirmed: boolean, displayValue?: string } ): Promise { - const { person, type, value, confirmed } = params + const { person, type, value, confirmed, displayValue } = params const { extra } = decodeTokenVerbose(ctx, token) - verifyAllowedServices(['github'], extra) + verifyAllowedServices(['github', 'telegram-bot'], extra) - return await addSocialId(db, person, type, value, confirmed) + return await addSocialId(db, person, type, value, confirmed, displayValue) +} + +export async function updateSocialId ( + ctx: MeasureContext, + db: AccountDB, + branding: Branding | null, + token: string, + params: { personId: PersonId, displayValue: string } +): Promise { + const { personId, displayValue } = params + const { extra } = decodeTokenVerbose(ctx, token) + + verifyAllowedServices(['telegram-bot'], extra) + + const socialId = await db.socialId.findOne({ _id: personId }) + if (socialId != null) { + throw new PlatformError(new Status(Severity.ERROR, platform.status.SocialIdNotFound, { _id: personId })) + } + + await db.socialId.updateOne({ _id: personId }, { displayValue }) } // Move to config? @@ -795,6 +816,21 @@ export async function listIntegrationsSecrets ( return await db.integrationSecret.find({ socialId, kind, workspaceUuid, key }) } +export async function findFullSocialIdBySocialKey ( + ctx: MeasureContext, + db: AccountDB, + branding: Branding | null, + token: string, + params: { socialKey: string } +): Promise { + const { extra } = decodeTokenVerbose(ctx, token) + verifyAllowedServices(['telegram-bot'], extra) + + const { socialKey } = params + + return await db.socialId.findOne({ key: socialKey }) +} + export type AccountServiceMethods = | 'getPendingWorkspace' | 'updateWorkspaceInfo' @@ -805,6 +841,7 @@ export type AccountServiceMethods = | 'performWorkspaceOperation' | 'updateWorkspaceRoleBySocialKey' | 'addSocialIdToPerson' + | 'updateSocialId' | 'getPersonInfo' | 'createIntegration' | 'updateIntegration' @@ -816,6 +853,7 @@ export type AccountServiceMethods = | 'deleteIntegrationSecret' | 'getIntegrationSecret' | 'listIntegrationsSecrets' + | 'findFullSocialIdBySocialKey' /** * @public @@ -831,6 +869,7 @@ export function getServiceMethods (): Partial { const normalizedValue = normalizeValue(value ?? '') @@ -1360,7 +1361,8 @@ export async function addSocialId ( const newSocialId: Omit = { type, value: normalizedValue, - personUuid + personUuid, + displayValue } if (confirmed) { diff --git a/server/core/src/queue/types.ts b/server/core/src/queue/types.ts index b71db2319b..42fc6391e9 100644 --- a/server/core/src/queue/types.ts +++ b/server/core/src/queue/types.ts @@ -11,7 +11,9 @@ export enum QueueTopic { Fulltext = 'fulltext', // A topic about user activity. - Users = 'users' + Users = 'users', + + TelegramBot = 'telegramBot' } export interface ConsumerHandle { diff --git a/server/core/src/types.ts b/server/core/src/types.ts index 40d4bf9aaf..1cce904f89 100644 --- a/server/core/src/types.ts +++ b/server/core/src/types.ts @@ -64,6 +64,7 @@ import { type Readable } from 'stream' import type { DbAdapter, DomainHelper } from './adapter' import type { StatisticsElement } from './stats' import { type StorageAdapter } from './storage' +import { type PlatformQueue } from './queue' export interface ServerFindOptions extends FindOptions { domain?: Domain // Allow to find for Doc's in specified domain only. @@ -193,6 +194,7 @@ export interface PipelineContext { serviceAdapterManager?: ServiceAdaptersManager lowLevelStorage?: LowLevelStorage liveQuery?: LiveQuery + queue?: PlatformQueue // Entry point for derived data procvessing derived?: Middleware @@ -267,6 +269,9 @@ export interface TriggerControl { lowLevel: LowLevelStorage modelDb: ModelDb removedMap: Map, Doc> + + queue?: PlatformQueue + communicationApi: CommunicationApi | null // Cache per workspace diff --git a/server/middleware/src/triggers.ts b/server/middleware/src/triggers.ts index 40f48c910f..918a435eed 100644 --- a/server/middleware/src/triggers.ts +++ b/server/middleware/src/triggers.ts @@ -133,6 +133,7 @@ export class TriggersMiddleware extends BaseMiddleware implements Middleware { storageAdapter: this.storageAdapter, serviceAdaptersManager: this.context.serviceAdapterManager as ServiceAdaptersManager, findAll, + queue: this.context.queue, contextCache: ctx.contextData.contextCache, modelDb: this.context.modelDb, hierarchy: this.context.hierarchy, diff --git a/server/server-pipeline/src/pipeline.ts b/server/server-pipeline/src/pipeline.ts index 8b797343d1..a0f139c99d 100644 --- a/server/server-pipeline/src/pipeline.ts +++ b/server/server-pipeline/src/pipeline.ts @@ -152,6 +152,7 @@ export function createServerPipeline ( branding, modelDb, hierarchy, + queue: opt.queue, storageAdapter: opt.externalStorage, contextVars: opt.pipelineContextVars ?? sharedPipelineContextVars, communicationApi diff --git a/services/telegram-bot/pod-telegram-bot/package.json b/services/telegram-bot/pod-telegram-bot/package.json index 81ca6ec4ee..f24a60d9a2 100644 --- a/services/telegram-bot/pod-telegram-bot/package.json +++ b/services/telegram-bot/pod-telegram-bot/package.json @@ -14,7 +14,7 @@ "build": "compile", "build:watch": "compile", "test": "jest --passWithNoTests --silent", - "_phase:bundle": "rushx bundle", + "_phase:bundle": "rushx bundle --external=ws", "_phase:docker-build": "rushx docker:build", "_phase:docker-staging": "rushx docker:staging", "bundle": "node ../../../common/scripts/esbuild.js", @@ -55,6 +55,7 @@ "dependencies": { "@hcengineering/activity": "^0.6.0", "@hcengineering/analytics": "^0.6.0", + "@hcengineering/account-client": "^0.6.0", "@hcengineering/analytics-service": "^0.6.0", "@hcengineering/attachment": "^0.6.14", "@hcengineering/chunter": "^0.6.20", @@ -62,17 +63,20 @@ "@hcengineering/client-resources": "^0.6.27", "@hcengineering/contact": "^0.6.24", "@hcengineering/core": "^0.6.32", + "@hcengineering/kafka": "^0.6.0", "@hcengineering/mongo": "^0.6.1", "@hcengineering/notification": "^0.6.23", "@hcengineering/platform": "^0.6.11", "@hcengineering/server-client": "^0.6.0", "@hcengineering/server-core": "^0.6.1", "@hcengineering/server-storage": "^0.6.0", + "@hcengineering/server-telegram": "^0.6.0", "@hcengineering/server-token": "^0.6.11", "@hcengineering/setting": "^0.6.17", "@hcengineering/telegram": "^0.6.21", "@hcengineering/telegram-assets": "^0.6.0", "@hcengineering/text": "^0.6.5", + "@hcengineering/api-client": "^0.6.0", "@telegraf/entity": "^0.5.0", "cors": "^2.8.5", "dotenv": "~16.0.0", @@ -80,7 +84,7 @@ "htmlparser2": "^9.0.0", "mongodb": "^6.12.0", "otp-generator": "^4.0.1", - "telegraf": "^4.16.3", - "ws": "^8.18.0" + "postgres": "^3.4.5", + "telegraf": "^4.16.3" } } diff --git a/services/telegram-bot/pod-telegram-bot/src/account.ts b/services/telegram-bot/pod-telegram-bot/src/account.ts new file mode 100644 index 0000000000..e7aabb6313 --- /dev/null +++ b/services/telegram-bot/pod-telegram-bot/src/account.ts @@ -0,0 +1,215 @@ +// +// Copyright © 2025 Hardcore Engineering Inc. +// +// Licensed under the Eclipse Public License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. You may +// obtain a copy of the License at https://www.eclipse.org/legal/epl-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// +// See the License for the specific language governing permissions and +// limitations under the License. +// +import { + AccountUuid, + Person, + PersonId, + SocialId, + SocialIdType, + WorkspaceUuid, + buildSocialIdString +} from '@hcengineering/core' +import { getAccountClient } from '@hcengineering/server-client' +import { generateToken } from '@hcengineering/server-token' +import { Integration } from '@hcengineering/account-client' + +import { serviceToken } from './utils' +import { IntegrationInfo } from './types' + +export async function getAccountPerson (account: AccountUuid): Promise { + try { + const accountClient = getAccountClient(generateToken(account)) + return await accountClient.getPerson() + } catch (e) { + console.error(e) + } + return undefined +} + +export async function getAccountSocialIds (account: AccountUuid): Promise { + try { + const accountClient = getAccountClient(generateToken(account)) + return await accountClient.getSocialIds() + } catch (e) { + console.error(e) + } + return [] +} + +export async function listIntegrationsByAccount (account: AccountUuid): Promise { + const client = getAccountClient(generateToken(account)) + const integrations = await client.listIntegrations({ kind: 'telegram-bot' }) + if (integrations.length === 0) return [] + const socialIds = await getAccountSocialIds(account) + + const result: IntegrationInfo[] = [] + for (const integration of integrations) { + if (integration.workspaceUuid == null) continue + const socialId = socialIds.find((it) => it._id === integration.socialId) + if (socialId === undefined) continue + result.push({ + ...integration, + workspaceUuid: integration.workspaceUuid, + account, + telegramId: Number(socialId.value), + username: socialId.displayValue + }) + } + + return result +} + +export async function listIntegrationsByTelegramId (telegramId: number): Promise { + const client = getAccountClient(serviceToken()) + const socialId = await client.findFullSocialIdBySocialKey( + buildSocialIdString({ type: SocialIdType.TELEGRAM, value: telegramId.toString() }) + ) + if (socialId == null) return [] + + const integrations = await client.listIntegrations({ kind: 'telegram-bot', socialId: socialId._id }) + if (integrations == null) return [] + + return integrations.map((it) => ({ + ...it, + workspaceUuid: it.workspaceUuid as WorkspaceUuid, + account: socialId.personUuid as AccountUuid, + telegramId, + username: socialId.displayValue + })) +} + +export async function getAnyIntegrationByTelegramId ( + telegramId: number, + workspace?: WorkspaceUuid +): Promise { + const client = getAccountClient(serviceToken()) + const socialId = await client.findFullSocialIdBySocialKey( + buildSocialIdString({ type: SocialIdType.TELEGRAM, value: telegramId.toString() }) + ) + if (socialId == null) return undefined + + const integrations = await client.listIntegrations({ kind: 'telegram-bot', socialId: socialId._id }) + if (integrations.length === 0) return undefined + + const integration = workspace != null ? integrations.find((it) => it.workspaceUuid === workspace) : integrations[0] + if (integration == null) return undefined + return { + ...integration, + workspaceUuid: integration.workspaceUuid as WorkspaceUuid, + account: socialId.personUuid as AccountUuid, + telegramId, + username: socialId.displayValue + } +} + +export async function getAnyIntegrationByAccount ( + account: AccountUuid, + workspace?: WorkspaceUuid +): Promise { + const client = getAccountClient(generateToken(account)) + const integrations = await client.listIntegrations({ kind: 'telegram-bot', workspaceUuid: workspace }) + if (integrations.length === 0) return undefined + + const integration = integrations[0] + const socialId = await getAccountClient(serviceToken()).findFullSocialIdBySocialKey(integration.socialId) + if (socialId == null) return undefined + + return { + ...integration, + workspaceUuid: integration.workspaceUuid as WorkspaceUuid, + account, + telegramId: Number(socialId.value), + username: socialId?.displayValue + } +} + +export async function getOrCreateSocialId ( + account: AccountUuid, + telegramId: number, + username?: string +): Promise { + const accountClient = getAccountClient(serviceToken()) + const socialId = await accountClient.findFullSocialIdBySocialKey( + buildSocialIdString({ type: SocialIdType.TELEGRAM, value: telegramId.toString() }) + ) + if (socialId == null) { + return await accountClient.addSocialIdToPerson( + account, + SocialIdType.TELEGRAM, + telegramId.toString(), + true, + username + ) + } + + // TODO: proper handle if connected to other account + if (socialId.personUuid !== account) { + throw new Error('Social id connected to another account') + } + + if (socialId.displayValue !== username) { + await accountClient.updateSocialId(socialId._id, username ?? '') + } + + return socialId._id +} + +export async function createIntegration (socialId: PersonId, workspace: WorkspaceUuid): Promise { + const accountClient = getAccountClient(serviceToken()) + const integration = { + socialId, + kind: 'telegram-bot', + workspaceUuid: workspace + } + await accountClient.createIntegration(integration) + return integration +} + +export async function removeIntegrationsByTg (telegramId: number): Promise { + const accountClient = getAccountClient(serviceToken()) + const socialId = await accountClient.findSocialIdBySocialKey( + buildSocialIdString({ type: SocialIdType.TELEGRAM, value: telegramId.toString() }) + ) + if (socialId == null) return + const integrations = await accountClient.listIntegrations({ kind: 'telegram-bot', socialId }) + for (const integration of integrations) { + await accountClient.deleteIntegration({ socialId, kind: 'telegram-bot', workspaceUuid: integration.workspaceUuid }) + } +} + +export async function addWorkspace (integration: IntegrationInfo, workspace: WorkspaceUuid): Promise { + const client = getAccountClient(serviceToken()) + await client.createIntegration({ ...integration, workspaceUuid: workspace }) +} + +export async function disableIntegration (integration: IntegrationInfo): Promise { + const client = getAccountClient(serviceToken()) + await client.updateIntegration({ + ...integration, + data: { + disabled: true + } + }) +} + +export async function enableIntegration (integration: IntegrationInfo): Promise { + const client = getAccountClient(serviceToken()) + await client.updateIntegration({ + ...integration, + data: { + disabled: false + } + }) +} diff --git a/services/telegram-bot/pod-telegram-bot/src/config.ts b/services/telegram-bot/pod-telegram-bot/src/config.ts index 736bf45a2e..fe0594975c 100644 --- a/services/telegram-bot/pod-telegram-bot/src/config.ts +++ b/services/telegram-bot/pod-telegram-bot/src/config.ts @@ -14,20 +14,23 @@ // export interface Config { - Port: number - BotToken: string - MongoURL: string - MongoDB: string - ServiceId: string - Secret: string - Domain: string - BotPort: number - App: string - OtpTimeToLiveSec: number - OtpRetryDelaySec: number - AccountsUrl: string - SentryDSN: string AccountsURL: string + AccountsUrl: string + App: string + BotPort: number + BotToken: string + DbUrl: string + Domain: string + MongoDB: string + MongoURL: string + OtpRetryDelaySec: number + OtpTimeToLiveSec: number + Port: number + QueueConfig: string + QueueRegion: string + Secret: string + SentryDSN: string + ServiceId: string } const parseNumber = (str: string | undefined): number | undefined => (str !== undefined ? Number(str) : undefined) @@ -36,10 +39,11 @@ const config: Config = (() => { const params: Partial = { Port: parseNumber(process.env.PORT) ?? 4020, BotToken: process.env.BOT_TOKEN, - MongoURL: process.env.MONGO_URL, - MongoDB: process.env.MONGO_DB, + // TODO: remove mongo + MongoURL: process.env.MONGO_URL ?? '', + MongoDB: process.env.MONGO_DB ?? '', AccountsUrl: process.env.ACCOUNTS_URL, - ServiceId: process.env.SERVICE_ID, + ServiceId: process.env.SERVICE_ID ?? 'telegram-bot', Secret: process.env.SECRET, Domain: process.env.DOMAIN ?? '', BotPort: parseNumber(process.env.BOT_PORT) ?? 8443, @@ -48,7 +52,10 @@ const config: Config = (() => { OtpTimeToLiveSec: parseNumber(process.env.OTP_TIME_TO_LIVE_SEC) ?? 5 * 60, OtpRetryDelaySec: parseNumber(process.env.OTP_RETRY_DELAY_SEC) ?? 60, SentryDSN: process.env.SENTRY_DSN ?? '', - AccountsURL: process.env.ACCOUNTS_URL + AccountsURL: process.env.ACCOUNTS_URL, + DbUrl: process.env.DB_URL, + QueueRegion: process.env.QUEUE_REGION, + QueueConfig: process.env.QUEUE_CONFIG } const missingEnv = (Object.keys(params) as Array).filter((key) => params[key] === undefined) diff --git a/services/telegram-bot/pod-telegram-bot/src/db.ts b/services/telegram-bot/pod-telegram-bot/src/db.ts new file mode 100644 index 0000000000..36c2947572 --- /dev/null +++ b/services/telegram-bot/pod-telegram-bot/src/db.ts @@ -0,0 +1,236 @@ +// +// Copyright © 2025 Hardcore Engineering Inc. +// +// Licensed under the Eclipse Public License, Version 2.0 (the 'License'); +// you may not use this file except in compliance with the License. You may +// obtain a copy of the License at https://www.eclipse.org/legal/epl-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an 'AS IS' BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// +// See the License for the specific language governing permissions and +// limitations under the License. +// + +import postgres from 'postgres' +import { AccountUuid, Ref, WorkspaceUuid } from '@hcengineering/core' +import { ActivityMessage } from '@hcengineering/activity' + +import config from './config' +import { ChannelId, ChannelRecord, MessageRecord, OtpRecord, ReplyRecord } from './types' + +export async function getDb (): Promise { + const sql = postgres(config.DbUrl, { + connection: { + application_name: config.ServiceId + }, + fetch_types: true, + prepare: true + }) + + return await PostgresDB.create(sql) +} + +const otpTable = 'telegram_bot.otp' +const messagesTable = 'telegram_bot.messages' +const channelsTable = 'telegram_bot.channels' +const repliesTable = 'telegram_bot.replies' + +export class PostgresDB { + constructor (private readonly client: postgres.Sql) {} + + static async create (client: postgres.Sql): Promise { + await this.init(client) + return new PostgresDB(client) + } + + static async init (client: postgres.Sql): Promise { + const sql = ` + CREATE SCHEMA IF NOT EXISTS telegram_bot; + + CREATE TABLE IF NOT EXISTS ${otpTable} ( + telegram_id INT8 NOT NULL, + telegram_username TEXT NOT NULL, + code VARCHAR(255) NOT NULL, + expires TIMESTAMPTZ NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY (code) + ); + + CREATE TABLE IF NOT EXISTS ${messagesTable} ( + message_id VARCHAR(255) NOT NULL, + workspace UUID NOT NULL, + account UUID NOT NULL, + telegram_message_id INT8 NOT NULL, + PRIMARY KEY (workspace, account, message_id) + ); + + CREATE TABLE IF NOT EXISTS ${channelsTable} ( + rowid INT8 NOT NULL DEFAULT unique_rowid(), + workspace UUID NOT NULL, + _id VARCHAR(255) NOT NULL, + _class VARCHAR(255) NOT NULL, + name TEXT NOT NULL, + account UUID NOT NULL, + PRIMARY KEY (rowid), + UNIQUE (workspace, _id, account) + ); + + CREATE TABLE IF NOT EXISTS ${repliesTable} ( + message_id VARCHAR(255) NOT NULL, + telegram_user_id INT8 NOT NULL, + reply_id INT8 NOT NULL, + PRIMARY KEY (message_id, telegram_user_id, reply_id) + ); + ` + + await client.unsafe(sql) + } + + async insertOtp (otp: OtpRecord): Promise { + const sql = ` + INSERT INTO ${otpTable} (telegram_id, telegram_username, code, expires) + VALUES ($1::int8, $2::text, $3::text, $4::timestamptz)` + await this.client.unsafe(sql, [otp.telegramId, otp.telegramUsername ?? null, otp.code, otp.expires]) + } + + async getOtpByCode (code: string): Promise { + const sql = ` + SELECT * FROM ${otpTable} WHERE code = $1::text LIMIT 1` + const res = await this.client.unsafe(sql, [code]) + return res.map(toOtpRecord)[0] + } + + async getOtpByTelegramId (telegramId: number): Promise { + const sql = ` + SELECT * FROM ${otpTable} WHERE telegram_id = $1::int8 ORDER BY created_at DESC LIMIT 1` + const res = await this.client.unsafe(sql, [telegramId]) + return res.map(toOtpRecord)[0] + } + + async removeOtp (code: string): Promise { + const sql = `DELETE FROM ${otpTable} WHERE code = $1::text` + await this.client.unsafe(sql, [code]) + } + + async removeExpiredOtp (): Promise { + const sql = `DELETE FROM ${otpTable} WHERE expires < NOW();` + await this.client.unsafe(sql) + } + + async getChannels (account: AccountUuid, workspace: WorkspaceUuid): Promise { + const sql = ` + SELECT * FROM ${channelsTable} WHERE account = $1::uuid AND workspace = $2::uuid ORDER BY name ASC` + const res = await this.client.unsafe(sql, [account, workspace]) + return res.map(toChannelRecord) + } + + async getChannel (account: AccountUuid, channelId: ChannelId): Promise { + const sql = ` + SELECT * FROM ${channelsTable} WHERE account = $1::uuid AND _id = $2::varchar` + const res = await this.client.unsafe(sql, [account, channelId]) + return res.map(toChannelRecord)[0] + } + + async insertChannel (records: Omit): Promise { + const sql = ` + INSERT INTO ${channelsTable} ( + workspace, account, _id, _class, name + ) + VALUES ($1::uuid, $2::uuid, $3::varchar, $4::varchar, $5::text)` + await this.client.unsafe(sql, [records.workspace, records.account, records._id, records._class, records.name]) + } + + async removeChannels (ids: ChannelId[]): Promise { + const sql = `DELETE FROM ${channelsTable} WHERE rowid = ANY($1::int8[])` + await this.client.unsafe(sql, [ids]) + } + + async updateChannelName (id: ChannelId, name: string): Promise { + const sql = `UPDATE ${channelsTable} SET name = $2::text WHERE rowid = $1::int8` + await this.client.unsafe(sql, [id, name]) + } + + async insertMessage (record: MessageRecord): Promise { + const sql = ` + INSERT INTO ${messagesTable} ( + message_id, workspace, account, telegram_message_id + ) + VALUES ($1::varchar, $2::uuid, $3::uuid, $4::int8) + ON CONFLICT DO NOTHING` + await this.client.unsafe(sql, [record.messageId, record.workspace, record.account, record.telegramMessageId]) + } + + async getMessageByRef (account: AccountUuid, messageId: Ref): Promise { + const sql = ` + SELECT * FROM ${messagesTable} WHERE account = $1::uuid AND message_id = $2::varchar LIMIT 1` + const res = await this.client.unsafe(sql, [account, messageId]) + return res.map(toMessageRecord)[0] + } + + async getMessageByTgId (account: AccountUuid, telegramId: number): Promise { + const sql = ` + SELECT * FROM ${messagesTable} WHERE telegram_message_id = $1::int8 AND account = $2::uuid LIMIT 1` + const res = await this.client.unsafe(sql, [telegramId, account]) + return res.map(toMessageRecord)[0] + } + + async insertReply (record: ReplyRecord): Promise { + const sql = ` + INSERT INTO ${repliesTable} ( + message_id, telegram_user_id, reply_id + ) + VALUES ($1::varchar, $2::int8, $3::int8)` + await this.client.unsafe(sql, [record.messageId, record.telegramUserId, record.replyId]) + } + + async getReply (tgUserId: number, replyTo: number): Promise { + const sql = ` + SELECT * FROM ${repliesTable} WHERE telegram_user_id = $1::int8 AND reply_id = $2::int8 LIMIT 1` + const res = await this.client.unsafe(sql, [tgUserId, replyTo]) + return res.map(toReplyRecord)[0] + } + + async close (): Promise { + await this.client.end({ timeout: 0 }) + } +} + +function toOtpRecord (raw: any): OtpRecord { + return { + telegramId: Number(raw.telegram_id), + telegramUsername: raw.telegram_username, + code: raw.code, + expires: new Date(raw.expires), + createdAt: new Date(raw.created_at) + } +} + +function toChannelRecord (raw: any): ChannelRecord { + return { + rowId: String(raw.rowid) as ChannelId, + workspace: raw.workspace, + _id: raw._id, + _class: raw._class, + name: raw.name, + account: raw.account + } +} + +function toReplyRecord (raw: any): ReplyRecord { + return { + messageId: raw.message_id, + telegramUserId: Number(raw.telegram_user_id), + replyId: Number(raw.reply_id) + } +} + +function toMessageRecord (raw: any): MessageRecord { + return { + messageId: raw.message_id, + workspace: raw.workspace, + account: raw.account, + telegramMessageId: Number(raw.telegram_message_id) + } +} diff --git a/services/telegram-bot/pod-telegram-bot/src/mongoDb.ts b/services/telegram-bot/pod-telegram-bot/src/mongoDb.ts new file mode 100644 index 0000000000..c74c63afe2 --- /dev/null +++ b/services/telegram-bot/pod-telegram-bot/src/mongoDb.ts @@ -0,0 +1,60 @@ +// +// Copyright © 2025 Hardcore Engineering Inc. +// +// Licensed under the Eclipse Public License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. You may +// obtain a copy of the License at https://www.eclipse.org/legal/epl-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// +// See the License for the specific language governing permissions and +// limitations under the License. +// + +import { Collection, MongoClient } from 'mongodb' + +import config from './config' + +export interface OldUserRecord { + telegramId: number + telegramUsername?: string + email: string +} + +// TODO: remove mongo db after disconnect users +export class MongoDb { + private constructor ( + private readonly client: MongoClient, + private readonly users: Collection + ) {} + + async getAllUsers (): Promise { + return await this.users.find().toArray() + } + + async removeAllUsers (): Promise { + await this.users.deleteMany({}) + } + + static async create (): Promise { + const client = new MongoClient(config.MongoURL) + await client.connect() + + const db = client.db(config.MongoDB) + + const userStorage = db.collection('users') + + await db.dropCollection('messages') + await db.dropCollection('otp') + await db.dropCollection('replies') + await db.dropCollection('channels') + + return new MongoDb(client, userStorage) + } + + async close (): Promise { + await this.client.close() + } +} diff --git a/services/telegram-bot/pod-telegram-bot/src/server.ts b/services/telegram-bot/pod-telegram-bot/src/server.ts index 56ada6f667..cb012ef3b9 100644 --- a/services/telegram-bot/pod-telegram-bot/src/server.ts +++ b/services/telegram-bot/pod-telegram-bot/src/server.ts @@ -12,23 +12,29 @@ // See the License for the specific language governing permissions and // limitations under the License. // -/* eslint-disable @typescript-eslint/no-unused-vars */ import { Token } from '@hcengineering/server-token' import cors from 'cors' import express, { type Express, type NextFunction, type Request, type Response } from 'express' import { type Server } from 'http' import { MeasureContext } from '@hcengineering/core' import { Telegraf } from 'telegraf' -import telegram, { TelegramNotificationRequest } from '@hcengineering/telegram' +import telegram from '@hcengineering/telegram' import { translate } from '@hcengineering/platform' import { extractToken } from '@hcengineering/server-client' +import { Readable } from 'stream' +import type { ReadableStream } from 'stream/web' import { ApiError } from './error' import { PlatformWorker } from './worker' -import { Limiter } from './limiter' import config from './config' -import { toTelegramHtml, toMediaGroups } from './utils' import { TgContext } from './telegraf/types' +import { + addWorkspace, + listIntegrationsByAccount, + getAccountPerson, + enableIntegration, + getAnyIntegrationByAccount +} from './account' type AsyncRequestHandler = (req: Request, res: Response, token: Token, next: NextFunction) => Promise @@ -54,12 +60,7 @@ const wrapRequest = (fn: AsyncRequestHandler) => (req: Request, res: Response, n void handleRequest(fn, req, res, next) } -export function createServer ( - bot: Telegraf, - worker: PlatformWorker, - ctx: MeasureContext, - limiter: Limiter -): Express { +export function createServer (bot: Telegraf, worker: PlatformWorker, ctx: MeasureContext): Express { const app = express() app.use(cors()) @@ -68,166 +69,108 @@ export function createServer ( app.post( '/test', wrapRequest(async (_, res, token) => { - // TODO: FIXME - throw new Error('Not implemented') - // const record = await worker.getUserRecordByEmail(token.email) - // if (record === undefined) { - // throw new ApiError(404) - // } + const integrations = await listIntegrationsByAccount(token.account) + if (integrations.length === 0) { + throw new ApiError(404) + } - // await limiter.add(record.telegramId, async () => { - // ctx.info('Sending test message', { email: token.email, username: record.telegramUsername }) - // const testMessage = await translate(telegram.string.TestMessage, { app: config.App }) - // await bot.telegram.sendMessage(record.telegramId, testMessage) - // }) + const integration = integrations[0] - // res.status(200) - // res.json({}) - }) - ) + await worker.limiter.add(integration.telegramId, async () => { + ctx.info('Sending test message', { account: token.account, username: integration.username }) + const testMessage = await translate(telegram.string.TestMessage, { app: config.App }) + await bot.telegram.sendMessage(integration.telegramId, testMessage) + }) - app.post( - '/updateWorkspace', - wrapRequest(async (req, res, token) => { - // TODO: FIXME - throw new Error('Not implemented') - // if (req.body == null || typeof req.body !== 'object' || req.body.enabled == null) { - // throw new ApiError(400) - // } + const workspaceIntegration = integrations.find((it) => it.workspaceUuid === token.workspace) - // const enabled: boolean = req.body.enabled - // const record = await worker.getUserRecordByEmail(token.email) + if (workspaceIntegration === undefined) { + await addWorkspace(integrations[0], token.workspace) + } else if (workspaceIntegration.data?.disabled === true) { + await enableIntegration(workspaceIntegration) + } - // if (record === undefined) { - // return - // } - - // if (enabled && !record.workspaces.includes(token.workspace.name)) { - // await worker.addWorkspace(token.email, token.workspace.name) - // } - - // if (!enabled && record.workspaces.includes(token.workspace.name)) { - // await worker.removeWorkspace(token.email, token.workspace.name) - // } - - // res.status(200) - // res.json({}) + res.status(200) + res.json({}) }) ) app.post( '/auth', wrapRequest(async (req, res, token) => { - // TODO: FIXME - throw new Error('Not implemented') - // if (req.body == null || typeof req.body !== 'object') { - // throw new ApiError(400) - // } + if (req.body == null || typeof req.body !== 'object') { + throw new ApiError(400) + } - // const { code } = req.body + const { code } = req.body - // if (code == null || code === '' || typeof code !== 'string') { - // throw new ApiError(400) - // } + if (code == null || code === '' || typeof code !== 'string') { + throw new ApiError(400) + } - // const record = await worker.getUserRecordByEmail(token.email) + const integration = await getAnyIntegrationByAccount(token.account) - // if (record !== undefined) { - // throw new ApiError(409, 'User already authorized') - // } + if (integration !== undefined) { + throw new ApiError(409, 'User already authorized') + } - // const newRecord = await worker.authorizeUser(code, token.email, token.workspace.name) + const person = await getAccountPerson(token.account) + if (person === undefined) { + throw new ApiError(404, 'Person not found') + } - // if (newRecord === undefined) { - // throw new ApiError(500) - // } + const newRecord = await worker.authorizeUser(code, token.account, token.workspace) + if (newRecord === undefined) { + throw new ApiError(500) + } - // void limiter.add(newRecord.telegramId, async () => { - // ctx.info('Connected account', { email: token.email, username: newRecord.telegramUsername }) - // const message = await translate(telegram.string.AccountConnectedHtml, { app: config.App, email: token.email }) - // await bot.telegram.sendMessage(newRecord.telegramId, message, { parse_mode: 'HTML' }) - // }) + void worker.limiter.add(newRecord.telegramId, async () => { + ctx.info('Connected account', { account: token.account, username: newRecord.username }) + const message = await translate(telegram.string.AccountConnectedHtml, { + app: config.App, + name: `${person.firstName} ${person.lastName}` + }) + await bot.telegram.sendMessage(newRecord.telegramId, message, { parse_mode: 'HTML' }) + }) - // res.status(200) - // res.json({}) + res.status(200) + res.json({}) }) ) app.get( '/info', - wrapRequest(async (_, res) => { - const me = await bot.telegram.getMe() + wrapRequest(async (req, res) => { + const me = bot.botInfo ?? (await bot.telegram.getMe()) const profilePhotos = await bot.telegram.getUserProfilePhotos(me.id) - const photoId = profilePhotos.photos[0]?.[0]?.file_id + const photoId = profilePhotos.photos[0]?.[0]?.file_id ?? '' - let photoUrl = '' - - if (photoId !== undefined) { - photoUrl = (await bot.telegram.getFileLink(photoId)).toString() - } res.status(200) - res.json({ username: me.username, name: me.first_name, photoUrl }) + res.json({ username: me.username, name: me.first_name, photoId }) }) ) - app.post( - '/notify', - wrapRequest(async (req, res, token) => { - // TODO: FIXME - throw new Error('Not implemented') - // if (req.body == null || !Array.isArray(req.body)) { - // ctx.error('Invalid request body', { body: req.body, email: token.email }) - // throw new ApiError(400) - // } + app.get( + '/photo/:fileId', + wrapRequest(async (req, res) => { + const { fileId } = req.params + const fileLink = await bot.telegram.getFileLink(fileId) - // const notificationRequests = req.body as TelegramNotificationRequest[] - // const userRecord = await worker.getUserRecordByEmail(token.email) + const response = await fetch(fileLink.toString()) + if (!response.ok) { + res.status(response.status).send(response.statusText) + return + } - // if (userRecord === undefined) { - // ctx.error('User not found', { email: token.email }) - // throw new ApiError(404) - // } + if (response.body != null) { + res.setHeader('Content-Type', response.headers.get('Content-Type') ?? 'application/octet-stream') + res.setHeader('Content-Length', response.headers.get('Content-Length') ?? '0') - // ctx.info('Received notification', { - // email: token.email, - // username: userRecord.telegramUsername, - // ids: notificationRequests.map((it) => it.notificationId) - // }) - - // for (const request of notificationRequests) { - // void limiter.add(userRecord.telegramId, async () => { - // const { full: fullMessage, short: shortMessage } = toTelegramHtml(request) - // const files = await worker.getFiles(request) - // const messageIds: number[] = [] - - // if (files.length === 0) { - // const message = await bot.telegram.sendMessage(userRecord.telegramId, fullMessage, { - // parse_mode: 'HTML' - // }) - - // messageIds.push(message.message_id) - // } else { - // const groups = toMediaGroups(files, fullMessage, shortMessage) - // for (const group of groups) { - // const mediaGroup = await bot.telegram.sendMediaGroup(userRecord.telegramId, group) - // messageIds.push(...mediaGroup.map((it) => it.message_id)) - // } - // } - - // for (const messageId of messageIds) { - // await worker.addNotificationRecord({ - // messageId: request.messageId, - // notificationId: request.notificationId, - // email: userRecord.email, - // workspace: request.workspace, - // telegramId: messageId - // }) - // } - // }) - // } - - // res.status(200) - // res.json({}) + const stream = Readable.fromWeb(response.body as ReadableStream) + stream.pipe(res) + } else { + res.status(500).send('Failed to fetch photo') + } }) ) diff --git a/services/telegram-bot/pod-telegram-bot/src/start.ts b/services/telegram-bot/pod-telegram-bot/src/start.ts index 99f1410c8b..5759538ca9 100644 --- a/services/telegram-bot/pod-telegram-bot/src/start.ts +++ b/services/telegram-bot/pod-telegram-bot/src/start.ts @@ -15,24 +15,26 @@ import { Analytics } from '@hcengineering/analytics' import { SplitLogger, configureAnalytics } from '@hcengineering/analytics-service' -import { MeasureMetricsContext, newMetrics } from '@hcengineering/core' +import { MeasureMetricsContext, WorkspaceUuid, newMetrics } from '@hcengineering/core' import { setMetadata, translate } from '@hcengineering/platform' import serverClient from '@hcengineering/server-client' -import { initStatisticsContext, type StorageConfiguration } from '@hcengineering/server-core' -import { buildStorageFromConfig, storageConfigFromEnv } from '@hcengineering/server-storage' +import { initStatisticsContext, QueueTopic } from '@hcengineering/server-core' import serverToken from '@hcengineering/server-token' -import telegram from '@hcengineering/telegram' import { join } from 'path' -import { Telegraf } from 'telegraf' +import { getPlatformQueue } from '@hcengineering/kafka' +import { TelegramQueueMessage, TelegramQueueMessageType } from '@hcengineering/server-telegram' +import telegram from '@hcengineering/telegram' import config from './config' -import { Limiter } from './limiter' import { registerLoaders } from './loaders' import { createServer, listen } from './server' import { setUpBot } from './telegraf/bot' -import { Command } from './telegraf/commands' -import { TgContext } from './telegraf/types' import { PlatformWorker } from './worker' +import { Telegraf } from 'telegraf' +import { TgContext } from './telegraf/types' +import { Limiter } from './limiter' +import { MongoDb } from './mongoDb' +import { Command } from './telegraf/commands' const ctx = initStatisticsContext('telegram-bot', { factory: () => @@ -51,15 +53,16 @@ const ctx = initStatisticsContext('telegram-bot', { configureAnalytics(config.SentryDSN, config) Analytics.setTag('application', 'telegram-bot-service') -export async function requestReconnect ( - bot: Telegraf, - worker: PlatformWorker, - limiter: Limiter -): Promise { - const toReconnect = await worker.getUsersToDisconnect() +export async function requestReconnect (bot: Telegraf, limiter: Limiter): Promise { + if (config.MongoDB === '' || config.MongoURL === '') { + return + } + + const mongoDb = await MongoDb.create() + const toReconnect = await mongoDb.getAllUsers() if (toReconnect.length > 0) { - ctx.info('Disconnecting users', { users: toReconnect.map((it) => it.email) }) + ctx.info('Disconnecting users', { users: toReconnect.map((it) => [it.telegramUsername, it.email]) }) const message = await translate(telegram.string.DisconnectMessage, { app: config.App, command: Command.Connect }) for (const userRecord of toReconnect) { try { @@ -67,10 +70,11 @@ export async function requestReconnect ( await bot.telegram.sendMessage(userRecord.telegramId, message) }) } catch (e) { - ctx.error('Failed to send message', { user: userRecord.email, error: e }) + ctx.error('Failed to send message', { email: userRecord.email, tg: userRecord.telegramUsername, error: e }) } } - await worker.disconnectUsers() + await mongoDb.removeAllUsers() + await mongoDb.close() } } @@ -80,13 +84,11 @@ export const start = async (): Promise => { setMetadata(serverClient.metadata.UserAgent, config.ServiceId) registerLoaders() - const storageConfig: StorageConfiguration = storageConfigFromEnv() - const storageAdapter = buildStorageFromConfig(storageConfig) - - const worker = await PlatformWorker.create(ctx, storageAdapter) + const worker = await PlatformWorker.create(ctx) const bot = await setUpBot(worker) - const limiter = new Limiter() - const app = createServer(bot, worker, ctx, limiter) + + const app = createServer(bot, worker, ctx) + const queue = getPlatformQueue('telegramBotService', config.QueueRegion) if (config.Domain === '') { ctx.info('Starting bot with polling') @@ -108,11 +110,35 @@ export const start = async (): Promise => { res.status(200).send() }) - await requestReconnect(bot, worker, limiter) + await requestReconnect(bot, worker.limiter) const server = listen(app, ctx, config.Port) + const consumer = queue.createConsumer( + ctx, + QueueTopic.TelegramBot, + queue.getClientId(), + async (messages) => { + for (const message of messages) { + const id = message.id as WorkspaceUuid + const records = message.value + for (const record of records) { + switch (record.type) { + case TelegramQueueMessageType.Notification: + await worker.processNotification(id, record, bot) + break + case TelegramQueueMessageType.WorkspaceSubscription: + await worker.processWorkspaceSubscription(id, record) + break + } + } + } + } + ) + const onClose = (): void => { - server.close(() => process.exit()) + void Promise.all([consumer.close(), worker.close(), server.close()]).then(() => { + process.exit() + }) } process.once('SIGINT', () => { diff --git a/services/telegram-bot/pod-telegram-bot/src/storage.ts b/services/telegram-bot/pod-telegram-bot/src/storage.ts deleted file mode 100644 index f724b79820..0000000000 --- a/services/telegram-bot/pod-telegram-bot/src/storage.ts +++ /dev/null @@ -1,31 +0,0 @@ -// -// Copyright © 2024 Hardcore Engineering Inc. -// -// Licensed under the Eclipse Public License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. You may -// obtain a copy of the License at https://www.eclipse.org/legal/epl-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// -// See the License for the specific language governing permissions and -// limitations under the License. -// - -import { MongoClient } from 'mongodb' - -import config from './config' - -export const getDB = (() => { - let client: MongoClient | undefined - - return async () => { - if (client === undefined) { - client = new MongoClient(config.MongoURL) - await client.connect() - } - - return client.db(config.MongoDB) - } -})() diff --git a/services/telegram-bot/pod-telegram-bot/src/telegraf/bot.ts b/services/telegram-bot/pod-telegram-bot/src/telegraf/bot.ts index 3ac64f48f1..ed237b304a 100644 --- a/services/telegram-bot/pod-telegram-bot/src/telegraf/bot.ts +++ b/services/telegram-bot/pod-telegram-bot/src/telegraf/bot.ts @@ -20,15 +20,15 @@ import { htmlToMarkup, isEmptyMarkup, jsonToMarkup, MarkupNodeType } from '@hcen import { toHTML } from '@telegraf/entity' import { CallbackQuery, Message, Update } from 'telegraf/typings/core/types/typegram' import { translate } from '@hcengineering/platform' -import { ObjectId, WithId } from 'mongodb' +import { AccountUuid, WorkspaceUuid } from '@hcengineering/core' import config from '../config' import { PlatformWorker } from '../worker' import { TgContext, ReplyMessage } from './types' import { toTelegramFileInfo } from '../utils' import { Command, defineCommands } from './commands' -import { ChannelRecord, MessageRecord, TelegramFileInfo, UserRecord, WorkspaceInfo } from '../types' -import { WorkspaceUuid } from '@hcengineering/core' +import { ChannelId, ChannelRecord, IntegrationInfo, MessageRecord, TelegramFileInfo, WorkspaceInfo } from '../types' +import { getAnyIntegrationByTelegramId, listIntegrationsByTelegramId } from '../account' function encodeChannelId (channelId: string): string { return `@${channelId}` @@ -44,45 +44,45 @@ const getPrevActionId = (workspace: string, page: number): string => `prev_${wor async function findMessageRecord ( worker: PlatformWorker, - from: number, + fromTgUser: number, replyTo: number, - email: string + account: AccountUuid ): Promise { - const record = await worker.getNotificationRecord(replyTo, email) + const record = await worker.getMessageRecordByTelegramId(account, replyTo) if (record !== undefined) { return record } - const reply = await worker.getReply(from, replyTo) + const reply = await worker.getReply(fromTgUser, replyTo) if (reply === undefined) { return undefined } - return await worker.findMessageRecord(email, reply.notificationId, reply.messageId) + return await worker.getMessageRecordByRef(account, reply.messageId) } async function onReply ( ctx: Context, - from: number, + fromTgUser: number, message: ReplyMessage, messageId: number, replyTo: number, worker: PlatformWorker, username?: string ): Promise { - const userRecord = await worker.getUserRecord(from) + const integration = await getAnyIntegrationByTelegramId(fromTgUser) - if (userRecord === undefined) { + if (integration === undefined) { return false } - if (userRecord.telegramUsername !== username) { - await worker.updateTelegramUsername(userRecord, username) + if (integration.username !== username && username !== undefined) { + void worker.updateTelegramUsername(integration.socialId, username) } - const messageRecord = await findMessageRecord(worker, from, replyTo, userRecord.email) + const messageRecord = await findMessageRecord(worker, fromTgUser, replyTo, integration.account) if (messageRecord === undefined) { return false @@ -90,15 +90,14 @@ async function onReply ( await worker.saveReply({ replyId: messageId, - telegramId: from, - notificationId: messageRecord.notificationId, + telegramUserId: fromTgUser, messageId: messageRecord.messageId }) const file = await toTelegramFileInfo(ctx, message) const files: TelegramFileInfo[] = file !== undefined ? [file] : [] - return await worker.reply(messageRecord, htmlToMarkup(toHTML(message)), files) + return await worker.reply(integration, messageRecord, htmlToMarkup(toHTML(message)), files) } async function handleSelectChannel ( @@ -112,12 +111,12 @@ async function handleSelectChannel ( const id = ctx.chat?.id if (id === undefined) return ['', false] - const userRecord = await worker.getUserRecord(id) - if (userRecord === undefined) return ['', false] + const integration = await getAnyIntegrationByTelegramId(id) + if (integration === undefined) return ['', false] const channelId = decodeChannelId(match) if (channelId === undefined || channelId === '') return ['', false] - const channel = await worker.getChannel(userRecord.email, new ObjectId(channelId)) + const channel = await worker.getChannel(integration.account, channelId as ChannelId) if (channel === undefined) return ['', false] @@ -131,11 +130,19 @@ async function handleSelectChannel ( }) } - return [channel.name, await worker.sendMessage(channel, userMessage.message_id, text, file)] + return [ + channel.name, + await worker.sendMessage(channel, integration.account, integration.socialId, userMessage.message_id, text, file) + ] } -async function showNoChannelsMessage (ctx: Context, worker: PlatformWorker, workspace: WorkspaceUuid): Promise { - const ws = await worker.getWorkspaceInfo(workspace) +async function showNoChannelsMessage ( + ctx: Context, + worker: PlatformWorker, + workspace: WorkspaceUuid, + account: AccountUuid +): Promise { + const ws = await worker.getWorkspaceInfo(account, workspace) await ctx.editMessageText( `No channels found in workspace ${ws?.name ?? workspace}.\nTo sync channels call /${Command.SyncAllChannels} or /${Command.SyncStarredChannels}`, { parse_mode: 'HTML' } @@ -145,13 +152,13 @@ async function showNoChannelsMessage (ctx: Context, worker: PlatformWorker, work async function createSelectChannelKeyboard ( ctx: NarrowedContext, worker: PlatformWorker, - userRecord: UserRecord, + integration: IntegrationInfo, workspace: WorkspaceUuid ): Promise { - const channels = await worker.getChannels(userRecord.email, workspace) + const channels = await worker.getChannels(integration.account, workspace) if (channels.length === 0) { - const ws = await worker.getWorkspaceInfo(workspace) + const ws = await worker.getWorkspaceInfo(integration.account, workspace) await ctx.replyWithHTML( `No channels found in workspace ${ws?.name ?? workspace}.\nTo sync channels call /${Command.SyncAllChannels} or /${Command.SyncStarredChannels}` ) @@ -176,12 +183,13 @@ async function createSelectChannelKeyboard ( async function createSelectWorkspaceKeyboard ( ctx: NarrowedContext, worker: PlatformWorker, - workspaces: WorkspaceUuid[] + workspaces: WorkspaceUuid[], + integration: IntegrationInfo ): Promise { const info: WorkspaceInfo[] = [] for (const workspace of workspaces) { - const workspaceInfo = await worker.getWorkspaceInfo(workspace) + const workspaceInfo = await worker.getWorkspaceInfo(integration.account, workspace) if (workspaceInfo === undefined) continue info.push(workspaceInfo) } @@ -237,19 +245,20 @@ export async function setUpBot (worker: PlatformWorker): Promise it.data?.disabled !== true) + .map((it) => it.workspaceUuid) if (workspaces.length === 0) { await ctx.reply("You don't have any connected workspaces") return } if (workspaces.length === 1) { - await createSelectChannelKeyboard(ctx, worker, userRecord, workspaces[0]) + await createSelectChannelKeyboard(ctx, worker, integrations[0], workspaces[0]) } else { - await createSelectWorkspaceKeyboard(ctx, worker, workspaces) + await createSelectWorkspaceKeyboard(ctx, worker, workspaces, integrations[0]) } }) @@ -279,13 +288,15 @@ export async function setUpBot (worker: PlatformWorker): Promise${info.name}`, { parse_mode: 'HTML' @@ -337,7 +348,7 @@ export async function setUpBot (worker: PlatformWorker): Promise[], page: number): WithId[] => { +const getPageChannels = (channels: ChannelRecord[], page: number): ChannelRecord[] => { return channels.slice(page * channelsPerPage, (page + 1) * channelsPerPage) } @@ -350,13 +361,13 @@ const editChannelKeyboard = async ( const id = ctx.chat?.id if (id === undefined) return - const userRecord = await worker.getUserRecord(id) - if (userRecord === undefined) return + const integration = await getAnyIntegrationByTelegramId(id, workspace) + if (integration === undefined) return - const channels = await worker.getChannels(userRecord.email, workspace) + const channels = await worker.getChannels(integration.account, workspace) if (channels.length === 0) { - await showNoChannelsMessage(ctx, worker, workspace) + await showNoChannelsMessage(ctx, worker, workspace, integration.account) return } diff --git a/services/telegram-bot/pod-telegram-bot/src/telegraf/commands.ts b/services/telegram-bot/pod-telegram-bot/src/telegraf/commands.ts index 9b4463b762..e272abfed4 100644 --- a/services/telegram-bot/pod-telegram-bot/src/telegraf/commands.ts +++ b/services/telegram-bot/pod-telegram-bot/src/telegraf/commands.ts @@ -21,6 +21,13 @@ import { Context, Telegraf } from 'telegraf' import config from '../config' import { PlatformWorker } from '../worker' import { TgContext } from './types' +import { + listIntegrationsByTelegramId, + getAccountPerson, + removeIntegrationsByTg, + getAnyIntegrationByTelegramId +} from '../account' +import { WorkspaceUuid } from '@hcengineering/core' export enum Command { Start = 'start', @@ -68,22 +75,25 @@ export async function getCommandsHelp (lang: string): Promise { async function onStart (ctx: Context, worker: PlatformWorker): Promise { const id = ctx.from?.id const lang = ctx.from?.language_code ?? 'en' - const record = id !== undefined ? await worker.getUserRecord(id) : undefined + const integration = id !== undefined ? await getAnyIntegrationByTelegramId(id) : undefined const commandsHelp = await getCommandsHelp(lang) const welcomeMessage = await translate(telegram.string.WelcomeMessage, { app: config.App }, lang) - if (record !== undefined) { + if (integration !== undefined) { + const person = await getAccountPerson(integration.account) + if (person === undefined) return const connectedMessage = await translate( telegram.string.ConnectedDescriptionHtml, - { email: record.email, app: config.App }, + { name: `${person.firstName} ${person.lastName}`, app: config.App }, lang ) const message = welcomeMessage + '\n\n' + commandsHelp + '\n\n' + connectedMessage await ctx.replyWithHTML(message) - if (record.telegramUsername !== ctx.from?.username) { - await worker.updateTelegramUsername(record, ctx.from?.username) + const username = ctx.from?.username + if (integration.username !== username && username !== undefined) { + await worker.updateTelegramUsername(integration.socialId, username) } } else { const minutes = Math.round(config.OtpTimeToLiveSec / 60) @@ -100,9 +110,9 @@ async function onHelp (ctx: Context): Promise { await ctx.reply(commandsHelp) } -async function onStop (ctx: Context, worker: PlatformWorker): Promise { +async function onStop (ctx: Context): Promise { if (ctx.from?.id !== undefined) { - await worker.removeUserByTelegramId(ctx.from?.id) + await removeIntegrationsByTg(ctx.from?.id) } const lang = ctx.from?.language_code ?? 'en' const message = await translate(telegram.string.StopMessage, { app: config.App }, lang) @@ -117,14 +127,16 @@ async function onSyncChannels (ctx: Context, worker: PlatformWorker, onlyStarred return } - const record = await worker.getUserRecord(id) + const integrations = await listIntegrationsByTelegramId(id) - if (record === undefined) return + if (integrations.length === 0) return - const workspaces = record.workspaces + const workspaces: WorkspaceUuid[] = integrations + .filter((it) => it.data?.disabled !== true) + .map((it) => it.workspaceUuid) for (const workspace of workspaces) { - await worker.syncChannels(record.email, workspace, onlyStarred) + await worker.syncChannels(integrations[0].account, workspace, onlyStarred) } await ctx.reply('List of channels updated') @@ -138,12 +150,14 @@ async function onConnect (ctx: Context, worker: PlatformWorker): Promise { return } - const account = await worker.getUserRecord(id) + const integration = await getAnyIntegrationByTelegramId(id) - if (account !== undefined) { + if (integration !== undefined) { + const person = await getAccountPerson(integration.account) + if (person === undefined) return const reply = await translate( telegram.string.AccountAlreadyConnectedHtml, - { email: account.email, app: config.App }, + { name: `${person.firstName} ${person.lastName}`, app: config.App }, lang ) await ctx.replyWithHTML(reply) @@ -160,7 +174,7 @@ export async function defineCommands (bot: Telegraf, worker: Platform bot.start((ctx) => onStart(ctx, worker)) bot.help(onHelp) - bot.command(Command.Stop, (ctx) => onStop(ctx, worker)) + bot.command(Command.Stop, (ctx) => onStop(ctx)) bot.command(Command.Connect, (ctx) => onConnect(ctx, worker)) bot.command(Command.SyncAllChannels, (ctx) => onSyncChannels(ctx, worker, false)) bot.command(Command.SyncStarredChannels, (ctx) => onSyncChannels(ctx, worker, true)) diff --git a/services/telegram-bot/pod-telegram-bot/src/types.ts b/services/telegram-bot/pod-telegram-bot/src/types.ts index 83601403cb..31a208b864 100644 --- a/services/telegram-bot/pod-telegram-bot/src/types.ts +++ b/services/telegram-bot/pod-telegram-bot/src/types.ts @@ -13,38 +13,32 @@ // limitations under the License. // -import { Class, Ref, Timestamp, WorkspaceUuid } from '@hcengineering/core' -import { InboxNotification } from '@hcengineering/notification' +import { AccountUuid, Class, Ref, WorkspaceUuid } from '@hcengineering/core' import { ChunterSpace } from '@hcengineering/chunter' import { ActivityMessage } from '@hcengineering/activity' +import { Integration } from '@hcengineering/account-client' -export interface UserRecord { - telegramId: number - telegramUsername?: string - email: string - workspaces: WorkspaceUuid[] -} +export type ChannelId = string & { __channelId: true } export interface MessageRecord { - notificationId?: Ref - messageId?: Ref - workspace: string - email: string - telegramId: number + workspace: WorkspaceUuid + account: AccountUuid + messageId: Ref + telegramMessageId: number } export interface ChannelRecord { - workspace: string - channelId: Ref - channelClass: Ref> + rowId: ChannelId + workspace: WorkspaceUuid + _id: Ref + _class: Ref> name: string - email: string + account: AccountUuid } export interface ReplyRecord { - notificationId?: Ref - messageId?: Ref - telegramId: number + messageId: Ref + telegramUserId: number replyId: number } @@ -52,8 +46,8 @@ export interface OtpRecord { telegramId: number telegramUsername?: string code: string - expires: Timestamp - createdOn: Timestamp + expires: Date + createdAt: Date } export interface PlatformFileInfo { @@ -76,3 +70,10 @@ export interface WorkspaceInfo { id: string name: string } + +export type IntegrationInfo = Integration & { + account: AccountUuid + workspaceUuid: WorkspaceUuid + telegramId: number + username?: string +} diff --git a/services/telegram-bot/pod-telegram-bot/src/utils.ts b/services/telegram-bot/pod-telegram-bot/src/utils.ts index cdd754eb03..550deca5d4 100644 --- a/services/telegram-bot/pod-telegram-bot/src/utils.ts +++ b/services/telegram-bot/pod-telegram-bot/src/utils.ts @@ -13,31 +13,33 @@ // limitations under the License. // -import { Collection } from 'mongodb' import otpGenerator from 'otp-generator' import { Message } from 'telegraf/typings/core/types/typegram' -import { TelegramNotificationRequest } from '@hcengineering/telegram' import { Parser } from 'htmlparser2' import { MediaGroup } from 'telegraf/typings/telegram-types' import { InputMediaAudio, InputMediaDocument, InputMediaPhoto, InputMediaVideo } from 'telegraf/src/core/types/typegram' import { Context, Input } from 'telegraf' +import { TelegramNotificationQueueMessage } from '@hcengineering/server-telegram' +import { systemAccountUuid } from '@hcengineering/core' +import { generateToken } from '@hcengineering/server-token' -import { OtpRecord, PlatformFileInfo, TelegramFileInfo } from './types' +import { PlatformFileInfo, TelegramFileInfo } from './types' +import { PostgresDB } from './db' -export async function getNewOtp (otpCollection: Collection): Promise { +export async function getNewOtp (db: PostgresDB): Promise { let otp = otpGenerator.generate(6, { upperCaseAlphabets: false, lowerCaseAlphabets: false, specialChars: false }) - let exist = await otpCollection.findOne({ otp }) + let exist = await db.getOtpByCode(otp) while (exist != null) { otp = otpGenerator.generate(6, { lowerCaseAlphabets: false }) - exist = await otpCollection.findOne({ otp }) + exist = await db.getOtpByCode(otp) } return otp @@ -48,7 +50,7 @@ const maxQuoteLength = 500 const maxBodyLength = 2000 const maxSenderLength = 100 -export function toTelegramHtml (record: TelegramNotificationRequest): { +export function toTelegramHtml (record: TelegramNotificationQueueMessage): { full: string short: string } { @@ -304,3 +306,7 @@ export async function toTelegramFileInfo ( return undefined } + +export function serviceToken (): string { + return generateToken(systemAccountUuid, undefined, { service: 'telegram-bot' }) +} diff --git a/services/telegram-bot/pod-telegram-bot/src/worker.ts b/services/telegram-bot/pod-telegram-bot/src/worker.ts index e929c3d4c8..33691b48c3 100644 --- a/services/telegram-bot/pod-telegram-bot/src/worker.ts +++ b/services/telegram-bot/pod-telegram-bot/src/worker.ts @@ -13,74 +13,84 @@ // limitations under the License. // -import type { Collection, ObjectId, WithId } from 'mongodb' -import { MeasureContext, Ref, SortingOrder, systemAccountUuid, WorkspaceUuid } from '@hcengineering/core' -import { InboxNotification } from '@hcengineering/notification' -import { TelegramNotificationRequest } from '@hcengineering/telegram' -import { StorageAdapter } from '@hcengineering/server-core' +import { AccountUuid, MeasureContext, PersonId, Ref, WorkspaceUuid } from '@hcengineering/core' +import { StorageAdapter, type StorageConfiguration } from '@hcengineering/server-core' import chunter, { ChunterSpace } from '@hcengineering/chunter' import { formatName } from '@hcengineering/contact' -import { generateToken } from '@hcengineering/server-token' import { getAccountClient } from '@hcengineering/server-client' import { ActivityMessage } from '@hcengineering/activity' import { + ChannelId, ChannelRecord, + IntegrationInfo, MessageRecord, - OtpRecord, PlatformFileInfo, ReplyRecord, TelegramFileInfo, - UserRecord, WorkspaceInfo } from './types' -import { getDB } from './storage' import { WorkspaceClient } from './workspace' -import { getNewOtp } from './utils' +import { getNewOtp, serviceToken, toMediaGroups, toTelegramHtml } from './utils' import config from './config' +import { + TelegramNotificationQueueMessage, + TelegramWorkspaceSubscriptionQueueMessage +} from '@hcengineering/server-telegram' +import { generateToken } from '@hcengineering/server-token' +import { Telegraf } from 'telegraf' +import { buildStorageFromConfig, storageConfigFromEnv } from '@hcengineering/server-storage' -const closeWorkspaceTimeout = 10 * 60 * 1000 // 10 minutes +import { Limiter } from './limiter' +import { TgContext } from './telegraf/types' +import { getDb, PostgresDB } from './db' +import { + addWorkspace, + createIntegration, + listIntegrationsByAccount, + getOrCreateSocialId, + disableIntegration, + getAnyIntegrationByTelegramId, + enableIntegration +} from './account' export class PlatformWorker { - private readonly workspacesClients = new Map() - private readonly closeWorkspaceTimeouts: Map = new Map() private readonly otpIntervalId: NodeJS.Timeout | undefined private readonly clearIntervalId: NodeJS.Timeout | undefined - private readonly channelsByWorkspace = new Map[]>() - private readonly channelById = new Map>() + private readonly channelsByWorkspace = new Map() + private readonly channelByRowId = new Map() private readonly workspaceInfoById = new Map() private constructor ( readonly ctx: MeasureContext, - readonly storageAdapter: StorageAdapter, - private readonly usersStorage: Collection, - private readonly messagesStorage: Collection, - private readonly otpStorage: Collection, - private readonly repliesStorage: Collection, - private readonly channelsStorage: Collection + readonly storage: StorageAdapter, + readonly limiter: Limiter, + private readonly db: PostgresDB ) { this.otpIntervalId = setInterval( () => { - void otpStorage.deleteMany({ expires: { $lte: Date.now() } }) + void db.removeExpiredOtp() }, 3 * 60 * 1000 ) this.clearIntervalId = setInterval( () => { this.channelsByWorkspace.clear() - this.channelById.clear() + this.channelByRowId.clear() }, 60 * 60 * 1000 ) } - public async getUsersToDisconnect (): Promise { - return await this.usersStorage.find({ workspaces: { $exists: false } }).toArray() - } + public static async create (ctx: MeasureContext): Promise { + const storageConfig: StorageConfiguration = storageConfigFromEnv() + const storage = buildStorageFromConfig(storageConfig) - public async disconnectUsers (): Promise { - await this.usersStorage.deleteMany({ workspaces: { $exists: false } }) + const db = await getDb() + const limiter = new Limiter() + + return new PlatformWorker(ctx, storage, limiter, db) } async close (): Promise { @@ -90,162 +100,83 @@ export class PlatformWorker { if (this.clearIntervalId !== undefined) { clearInterval(this.clearIntervalId) } + await this.db.close() } - async closeWorkspaceClient (workspace: WorkspaceUuid): Promise { - const timeoutId = this.closeWorkspaceTimeouts.get(workspace) - - if (timeoutId !== undefined) { - clearTimeout(timeoutId) - this.closeWorkspaceTimeouts.delete(workspace) - } - - const client = this.workspacesClients.get(workspace) - - if (client !== undefined) { - await client.close() - this.workspacesClients.delete(workspace) - } - } - - async addUser ( - id: number, - email: string, + private async addIntegration ( + telegramId: number, + account: AccountUuid, workspace: WorkspaceUuid, - telegramUsername?: string - ): Promise { - const emailRes = await this.usersStorage.findOne({ email }) + socialId: PersonId, + username?: string + ): Promise { + const integrationByTg = await getAnyIntegrationByTelegramId(telegramId) - if (emailRes !== null) { - if (emailRes.workspaces.includes(workspace)) { - return - } - if (!emailRes.workspaces.includes(workspace)) { - await this.usersStorage.updateOne({ email }, { $push: { workspaces: workspace } }) - } + if (integrationByTg != null) { + this.ctx.error('Account is already registered', { + telegramId, + account: integrationByTg.account, + requestAccount: account + }) return } - const tRes = await this.usersStorage.findOne({ telegramId: id }) + const integration = await createIntegration(socialId, workspace) - if (tRes !== null) { - if (tRes.email !== email) { - this.ctx.error('Account is already registered', { id, email: tRes.email, newEmail: email }) - } - if (tRes.email === email && !tRes.workspaces.includes(workspace)) { - await this.usersStorage.updateOne({ email }, { $push: { workspaces: workspace } }) - } - return + return { + ...integration, + workspaceUuid: workspace, + account, + telegramId, + username } - - const insertResult = await this.usersStorage.insertOne({ - telegramId: id, - email, - workspaces: [workspace], - telegramUsername - }) - - return (await this.usersStorage.findOne({ _id: insertResult.insertedId })) ?? undefined } - async getFiles (request: TelegramNotificationRequest): Promise { - if (request.messageId === undefined || !request.attachments) { - return [] - } - const wsClient = await this.getWorkspaceClient(request.workspace as any) // TODO: FIXME - return await wsClient.getFiles(request.messageId) + async getFiles ( + workspace: WorkspaceUuid, + message: Ref, + account: AccountUuid + ): Promise { + const wsClient = await WorkspaceClient.create(workspace, account, this.ctx, this.storage) + return await wsClient.getFiles(message) } - async updateTelegramUsername (userRecord: UserRecord, telegramUsername?: string): Promise { - await this.usersStorage.updateOne( - { telegramId: userRecord.telegramId, email: userRecord.email }, - { $set: { telegramUsername } } - ) - } - - async addNotificationRecord (record: MessageRecord): Promise { - await this.messagesStorage.insertOne(record) - } - - async removeUserByTelegramId (id: number): Promise { - await this.usersStorage.deleteOne({ telegramId: id }) + async updateTelegramUsername (personId: PersonId, telegramUsername: string): Promise { + await getAccountClient(serviceToken()).updateSocialId(personId, telegramUsername) } async saveReply (record: ReplyRecord): Promise { - await this.repliesStorage.insertOne(record) + await this.db.insertReply(record) } - async getReply (id: number, replyTo: number): Promise { - return (await this.repliesStorage.findOne({ telegramId: id, replyId: replyTo })) ?? undefined + async getReply (tgUserId: number, replyTo: number): Promise { + return await this.db.getReply(tgUserId, replyTo) } - async getNotificationRecord (id: number, email: string): Promise { - return (await this.messagesStorage.findOne({ telegramId: id, email })) ?? undefined - } - - async findMessageRecord ( - email: string, - notificationId?: Ref, - messageId?: Ref + async getMessageRecordByRef ( + account: AccountUuid, + messageId: Ref ): Promise { - if (notificationId !== undefined) { - return (await this.messagesStorage.findOne({ notificationId, email })) ?? undefined - } - - if (messageId !== undefined) { - return (await this.messagesStorage.findOne({ messageId, email })) ?? undefined - } - - return undefined + return await this.db.getMessageByRef(account, messageId) } - async getUserRecord (id: number): Promise { - return (await this.usersStorage.findOne({ telegramId: id })) ?? undefined + async getMessageRecordByTelegramId (account: AccountUuid, telegramId: number): Promise { + return await this.db.getMessageByTgId(account, telegramId) } - async getUserRecordByEmail (email: string): Promise { - return (await this.usersStorage.findOne({ email })) ?? undefined + async reply ( + integration: IntegrationInfo, + messageRecord: MessageRecord, + text: string, + files: TelegramFileInfo[] + ): Promise { + const client = await WorkspaceClient.create(messageRecord.workspace, integration.account, this.ctx, this.storage) + return await client.replyToMessage(integration.account, integration.socialId, messageRecord, text, files) } - async addWorkspace (email: string, workspace: WorkspaceUuid): Promise { - await this.usersStorage.updateOne({ email }, { $push: { workspaces: workspace } }) - } - - async removeWorkspace (email: string, workspace: WorkspaceUuid): Promise { - await this.usersStorage.updateOne({ email }, { $pull: { workspaces: workspace } }) - } - - async getWorkspaceClient (workspace: WorkspaceUuid): Promise { - const wsClient = - this.workspacesClients.get(workspace) ?? (await WorkspaceClient.create(workspace, this.ctx, this.storageAdapter)) - - if (!this.workspacesClients.has(workspace)) { - this.workspacesClients.set(workspace, wsClient) - } - - const timeoutId = this.closeWorkspaceTimeouts.get(workspace) - - if (timeoutId !== undefined) { - clearTimeout(timeoutId) - } - - const newTimeoutId = setTimeout(() => { - void this.closeWorkspaceClient(workspace) - }, closeWorkspaceTimeout) - - this.closeWorkspaceTimeouts.set(workspace, newTimeoutId) - - return wsClient - } - - async reply (messageRecord: MessageRecord, text: string, files: TelegramFileInfo[]): Promise { - const client = await this.getWorkspaceClient(messageRecord.workspace as any) // TODO: FIXME - return await client.reply(messageRecord, text, files) - } - - async getChannelName (client: WorkspaceClient, channel: ChunterSpace, email: string): Promise { + async getChannelName (client: WorkspaceClient, channel: ChunterSpace, account: AccountUuid): Promise { if (client.hierarchy.isDerived(channel._class, chunter.class.DirectMessage)) { - const persons = await client.getPersons(channel.members, email) + const persons = await client.getPersons(channel.members.filter((it) => it !== account)) return persons .map(({ name }) => formatName(name)) .sort((a, b) => a.localeCompare(b)) @@ -259,110 +190,107 @@ export class PlatformWorker { return channel.name } - async getWorkspaces (email: string): Promise { - return (await this.usersStorage.findOne({ email }))?.workspaces ?? [] - } - - async getChannels (email: string, workspace: WorkspaceUuid): Promise[]> { - const key = `${email}:${workspace}` + async getChannels (account: AccountUuid, workspace: WorkspaceUuid): Promise { + const key = `${account}:${workspace}` if (this.channelsByWorkspace.has(key)) { return this.channelsByWorkspace.get(key) ?? [] } - const res = await this.channelsStorage - .find({ workspace, email }, { sort: { name: SortingOrder.Ascending } }) - .toArray() + const res = await this.db.getChannels(account, workspace) this.channelsByWorkspace.set(key, res) for (const channel of res) { - this.channelById.set(channel._id, channel) + this.channelByRowId.set(channel.rowId, channel) } return res } - async getChannel (email: string, channelId: ObjectId): Promise | undefined> { - if (this.channelById.has(channelId)) { - const channel = this.channelById.get(channelId) + async getChannel (account: AccountUuid, channelId: ChannelId): Promise { + if (this.channelByRowId.has(channelId)) { + const channel = this.channelByRowId.get(channelId) - return channel !== undefined && channel.email === email ? channel : undefined + return channel !== undefined && channel.account === account ? channel : undefined } - const res = (await this.channelsStorage.findOne({ _id: channelId, email })) ?? undefined + const res = await this.db.getChannel(account, channelId) if (res !== undefined) { - this.channelById.set(res._id, res) + this.channelByRowId.set(res.rowId, res) } return res } async sendMessage ( channel: ChannelRecord, + account: AccountUuid, + socialId: PersonId, telegramId: number, text: string, file?: TelegramFileInfo ): Promise { - const client = await this.getWorkspaceClient(channel.workspace as any) // TODO: FIXME - const _id = await client.sendMessage(channel, text, file) + const client = await WorkspaceClient.create(channel.workspace, account, this.ctx, this.storage) + const _id = await client.sendMessage(channel, account, socialId, text, file) - await this.messagesStorage.insertOne({ - email: channel.email, + if (_id === undefined) return false + + await this.db.insertMessage({ workspace: channel.workspace, - telegramId, - messageId: _id + account: channel.account, + messageId: _id, + telegramMessageId: telegramId }) - return _id !== undefined + return true } - async syncChannels (email: string, workspace: WorkspaceUuid, onlyStarred: boolean): Promise { - const client = await this.getWorkspaceClient(workspace as any) // TODO: FIXME - const channels = await client.getChannels(email, onlyStarred) - const existingChannels = await this.channelsStorage.find({ workspace, email }).toArray() + async syncChannels (account: AccountUuid, workspace: WorkspaceUuid, onlyStarred: boolean): Promise { + const client = await WorkspaceClient.create(workspace, account, this.ctx, this.storage) + const channels = await client.getChannels(account, onlyStarred) + const existingChannels = await this.db.getChannels(account, workspace) - const toInsert: ChannelRecord[] = [] - const toDelete: WithId[] = [] + const toInsert: Omit[] = [] + const toDelete: ChannelRecord[] = [] for (const channel of channels) { - const existingChannel = existingChannels.find((c) => c.channelId === channel._id) - const name = await this.getChannelName(client, channel, email) + const existingChannel = existingChannels.find((c) => c._id === channel._id) + const name = await this.getChannelName(client, channel, account) if (existingChannel === undefined) { - toInsert.push({ workspace, email, channelId: channel._id, channelClass: channel._class, name }) + toInsert.push({ workspace, account, _id: channel._id, _class: channel._class, name }) } else if (existingChannel.name !== name) { - await this.channelsStorage.updateOne({ workspace, email, _id: channel._id }, { $set: { name } }) + await this.db.updateChannelName(existingChannel.rowId, name) } } for (const existingChannel of existingChannels) { - const channel = channels.find(({ _id }) => _id === existingChannel.channelId) + const channel = channels.find(({ _id }) => _id === existingChannel._id) if (channel === undefined) { toDelete.push(existingChannel) } } if (toInsert.length > 0) { - await this.channelsStorage.insertMany(toInsert) + await Promise.all(toInsert.map((it) => this.db.insertChannel(it))) } if (toDelete.length > 0) { - await this.channelsStorage.deleteMany({ _id: { $in: toDelete.map((c) => c._id) } }) + await this.db.removeChannels(toDelete.map((c) => c.rowId)) } - this.channelsByWorkspace.delete(`${email}:${workspace}`) - for (const [key, channel] of this.channelById.entries()) { - if (channel.email === email) { - this.channelById.delete(key) + this.channelsByWorkspace.delete(`${account}:${workspace}`) + for (const [key, channel] of this.channelByRowId.entries()) { + if (channel.account === account) { + this.channelByRowId.delete(key) } } } - async getWorkspaceInfo (workspaceId: WorkspaceUuid): Promise { + async getWorkspaceInfo (account: AccountUuid, workspaceId: WorkspaceUuid): Promise { if (this.workspaceInfoById.has(workspaceId)) { return this.workspaceInfoById.get(workspaceId) } try { - const token = generateToken(systemAccountUuid, workspaceId, { service: 'telegram' }) - const accountClient = getAccountClient(token) + const accountClient = getAccountClient(generateToken(account, workspaceId)) const result = await accountClient.getWorkspaceInfo(false) if (result === undefined) { @@ -382,72 +310,119 @@ export class PlatformWorker { } } - async authorizeUser (code: string, email: string, workspace: WorkspaceUuid): Promise { - const otpData = (await this.otpStorage.findOne({ code })) ?? undefined - const isExpired = otpData !== undefined && otpData.expires < Date.now() + async authorizeUser ( + code: string, + account: AccountUuid, + workspace: WorkspaceUuid + ): Promise { + const otpData = await this.db.getOtpByCode(code) + const isExpired = otpData !== undefined && otpData.expires < new Date() const isValid = otpData !== undefined && !isExpired && code === otpData.code - if (!isValid) { + if (!isValid || otpData === undefined) { throw new Error('Invalid OTP') } - return await this.addUser(otpData.telegramId, email, workspace, otpData.telegramUsername) + await this.db.removeOtp(code) + + const socialId = await getOrCreateSocialId(account, otpData.telegramId, otpData.telegramUsername) + return await this.addIntegration(otpData.telegramId, account, workspace, socialId, otpData.telegramUsername) } async generateCode (telegramId: number, telegramUsername?: string): Promise { - const now = Date.now() - const otpData = ( - await this.otpStorage.find({ telegramId }).sort({ createdOn: SortingOrder.Descending }).limit(1).toArray() - )[0] + const now = new Date() + const otpData = await this.db.getOtpByTelegramId(telegramId) const retryDelay = config.OtpRetryDelaySec * 1000 const isValid = otpData !== undefined && otpData.expires > now - const canRetry = otpData !== undefined && otpData.createdOn + retryDelay < now + const canRetry = otpData !== undefined && otpData.createdAt.getTime() + retryDelay < now.getTime() if (isValid && !canRetry) { return otpData.code } - const newCode = await getNewOtp(this.otpStorage) + const newCode = await getNewOtp(this.db) const timeToLive = config.OtpTimeToLiveSec * 1000 - const expires = now + timeToLive + const expires = Date.now() + timeToLive - await this.otpStorage.insertOne({ telegramId, code: newCode, expires, createdOn: now, telegramUsername }) + await this.db.insertOtp({ telegramId, code: newCode, expires: new Date(expires), createdAt: now, telegramUsername }) return newCode } - static async createStorages (): Promise< - [ - Collection, - Collection, - Collection, - Collection, - Collection - ] - > { - const db = await getDB() - const userStorage = db.collection('users') - await db.dropCollection('notifications') - const messagesStorage = db.collection('messages') - const otpStorage = db.collection('otp') - const repliesStorage = db.collection('replies') - const channelsStorage = db.collection('channels') + async processNotification ( + workspace: WorkspaceUuid, + record: TelegramNotificationQueueMessage, + bot: Telegraf + ): Promise { + const integrations = await listIntegrationsByAccount(record.account) - return [userStorage, messagesStorage, otpStorage, repliesStorage, channelsStorage] + if (integrations.length === 0) { + this.ctx.error('Integrations not found', { account: record.account }) + return + } + + const workspaceIntegration = integrations.find((it) => it.workspaceUuid === workspace) + if (workspaceIntegration === undefined) { + await addWorkspace(integrations[0], workspace) + } else if (workspaceIntegration.data?.disabled === true) { + await enableIntegration(workspaceIntegration) + } + + const integration = integrations[0] + + void this.limiter.add(integration.telegramId, async () => { + const { full: fullMessage, short: shortMessage } = toTelegramHtml(record) + const files = + record.attachments && record.messageId != null + ? await this.getFiles(workspace, record.messageId, record.account) + : [] + const tgMessageIds: number[] = [] + + if (files.length === 0) { + const message = await bot.telegram.sendMessage(integration.telegramId, fullMessage, { + parse_mode: 'HTML' + }) + + tgMessageIds.push(message.message_id) + } else { + const groups = toMediaGroups(files, fullMessage, shortMessage) + for (const group of groups) { + const mediaGroup = await bot.telegram.sendMediaGroup(integration.telegramId, group) + tgMessageIds.push(...mediaGroup.map((it) => it.message_id)) + } + } + + for (const messageId of tgMessageIds) { + if (record.messageId === undefined) continue + await this.db.insertMessage({ + messageId: record.messageId, + account: integration.account, + workspace, + telegramMessageId: messageId + }) + } + }) } - static async create (ctx: MeasureContext, storageAdapter: StorageAdapter): Promise { - const [userStorage, messagesStorage, otpStorage, repliesStorage, channelsStorage] = - await PlatformWorker.createStorages() + async processWorkspaceSubscription ( + workspace: WorkspaceUuid, + record: TelegramWorkspaceSubscriptionQueueMessage + ): Promise { + const integrations = await listIntegrationsByAccount(record.account) - return new PlatformWorker( - ctx, - storageAdapter, - userStorage, - messagesStorage, - otpStorage, - repliesStorage, - channelsStorage - ) + if (integrations.length === 0) { + this.ctx.error('Integrations not found', { account: record.account }) + return + } + + const workspaceIntegration = integrations.find((it) => it.workspaceUuid === workspace) + + if (record.subscribe && workspaceIntegration === undefined) { + await addWorkspace(integrations[0], workspace) + } else if (record.subscribe && workspaceIntegration?.data?.disabled === true) { + await enableIntegration(workspaceIntegration) + } else if (!record.subscribe && workspaceIntegration !== undefined) { + await disableIntegration(workspaceIntegration) + } } } diff --git a/services/telegram-bot/pod-telegram-bot/src/workspace.ts b/services/telegram-bot/pod-telegram-bot/src/workspace.ts index 1a7222f173..8e3682edbd 100644 --- a/services/telegram-bot/pod-telegram-bot/src/workspace.ts +++ b/services/telegram-bot/pod-telegram-bot/src/workspace.ts @@ -12,57 +12,55 @@ // See the License for the specific language governing permissions and // limitations under the License. // -/* eslint-disable @typescript-eslint/no-unused-vars */ import core, { - PersonId, + AccountUuid, Blob, Class, - Client, Doc, generateId, Hierarchy, Markup, MeasureContext, + PersonId, Ref, Space, - systemAccountUuid, TxFactory, - WorkspaceUuid, - AccountUuid + WorkspaceUuid } from '@hcengineering/core' -import { generateToken } from '@hcengineering/server-token' -import notification, { ActivityInboxNotification, MentionInboxNotification } from '@hcengineering/notification' +import notification from '@hcengineering/notification' import chunter, { ChatMessage, ChunterSpace, ThreadMessage } from '@hcengineering/chunter' import contact, { Person } from '@hcengineering/contact' -import { createClient, getTransactorEndpoint } from '@hcengineering/server-client' +import { getTransactorEndpoint } from '@hcengineering/server-client' import activity, { ActivityMessage } from '@hcengineering/activity' import attachment, { Attachment } from '@hcengineering/attachment' import { StorageAdapter } from '@hcengineering/server-core' +import { createRestClient, RestClient } from '@hcengineering/api-client' import { isEmptyMarkup } from '@hcengineering/text' +import { generateToken } from '@hcengineering/server-token' import { ChannelRecord, MessageRecord, PlatformFileInfo, TelegramFileInfo } from './types' export class WorkspaceClient { - hierarchy: Hierarchy private constructor ( private readonly ctx: MeasureContext, - private readonly storageAdapter: StorageAdapter, - private readonly client: Client, - private readonly token: string, + private readonly storage: StorageAdapter, + private readonly client: RestClient, + readonly hierarchy: Hierarchy, private readonly workspace: WorkspaceUuid - ) { - this.hierarchy = client.getHierarchy() - } + ) {} static async create ( workspace: WorkspaceUuid, + account: AccountUuid, ctx: MeasureContext, - storageAdapter: StorageAdapter + storage: StorageAdapter ): Promise { - const token = generateToken(systemAccountUuid, workspace) - const client = await connectPlatform(token) + const token = generateToken(account, workspace) + const endpoint = await getTransactorEndpoint(token) + const client = createRestClient(endpoint, workspace, token) + const model = await client.getModel() - return new WorkspaceClient(ctx, storageAdapter, client, token, workspace) + return new WorkspaceClient(ctx, storage, client, model.hierarchy, workspace) } async createAttachments ( @@ -79,7 +77,7 @@ export class WorkspaceClient { const response = await fetch(file.url) const buffer = Buffer.from(await response.arrayBuffer()) const uuid = generateId() - await this.storageAdapter.put(this.ctx, this.workspace as any, uuid, buffer, file.type, file.size) // TODO: FIXME + await this.storage.put(this.ctx, this.workspace as any, uuid, buffer, file.type, file.size) // TODO: FIXME const tx = factory.createTxCollectionCUD( _class, _id, @@ -146,189 +144,99 @@ export class WorkspaceClient { async createThreadMessage ( message: ActivityMessage, - account: any, + account: AccountUuid, + socialId: PersonId, text: string, files: TelegramFileInfo[] ): Promise { - // TODO: FIXME - throw new Error('Not implemented') - // const txFactory = new TxFactory(account._id) - // const hierarchy = this.hierarchy + const txFactory = new TxFactory(socialId) + const hierarchy = this.hierarchy - // const isAvailable = await this.isReplyAvailable(account._id, message) + const isAvailable = await this.isReplyAvailable(account, message) - // if (!isAvailable) { - // return false - // } + if (!isAvailable) { + return false + } - // const messageId = generateId() - // const attachments = await this.createAttachments( - // txFactory, - // messageId, - // chunter.class.ThreadMessage, - // message.space, - // files - // ) + const messageId = generateId() + const attachments = await this.createAttachments( + txFactory, + messageId, + chunter.class.ThreadMessage, + message.space, + files + ) - // if (attachments === 0 && isEmptyMarkup(text)) { - // return false - // } + if (attachments === 0 && isEmptyMarkup(text)) { + return false + } - // if (hierarchy.isDerived(message._class, chunter.class.ThreadMessage)) { - // const thread = message as ThreadMessage - // const collectionTx = txFactory.createTxCollectionCUD( - // thread.attachedToClass, - // thread.attachedTo, - // message.space, - // 'replies', - // txFactory.createTxCreateDoc( - // chunter.class.ThreadMessage, - // message.space, - // { - // attachedTo: thread.attachedTo, - // attachedToClass: thread.attachedToClass, - // objectId: thread.objectId, - // objectClass: thread.objectClass, - // message: text, - // attachments, - // collection: 'replies', - // provider: contact.channelProvider.Telegram - // }, - // messageId - // ) - // ) - // await this.client.tx(collectionTx) - // } else { - // const collectionTx = txFactory.createTxCollectionCUD( - // message._class, - // message._id, - // message.space, - // 'replies', - // txFactory.createTxCreateDoc( - // chunter.class.ThreadMessage, - // message.space, - // { - // attachedTo: message._id, - // attachedToClass: message._class, - // objectId: message.attachedTo, - // objectClass: message.attachedToClass, - // message: text, - // attachments, - // collection: 'replies', - // provider: contact.channelProvider.Telegram - // }, - // messageId - // ) - // ) - // await this.client.tx(collectionTx) - // } + if (hierarchy.isDerived(message._class, chunter.class.ThreadMessage)) { + const thread = message as ThreadMessage + const collectionTx = txFactory.createTxCollectionCUD( + thread.attachedToClass, + thread.attachedTo, + message.space, + 'replies', + txFactory.createTxCreateDoc( + chunter.class.ThreadMessage, + message.space, + { + attachedTo: thread.attachedTo, + attachedToClass: thread.attachedToClass, + objectId: thread.objectId, + objectClass: thread.objectClass, + message: text, + attachments, + collection: 'replies', + provider: contact.channelProvider.Telegram + }, + messageId + ) + ) + await this.client.tx(collectionTx) + } else { + const collectionTx = txFactory.createTxCollectionCUD( + message._class, + message._id, + message.space, + 'replies', + txFactory.createTxCreateDoc( + chunter.class.ThreadMessage, + message.space, + { + attachedTo: message._id, + attachedToClass: message._class, + objectId: message.attachedTo, + objectClass: message.attachedToClass, + message: text, + attachments, + collection: 'replies', + provider: contact.channelProvider.Telegram + }, + messageId + ) + ) + await this.client.tx(collectionTx) + } - // return true + return true } - async replyToActivityNotification ( - it: ActivityInboxNotification, - account: any, - text: string, - files: TelegramFileInfo[] - ): Promise { - // TODO: FIXME - throw new Error('Not implemented') - // const message = await this.client.findOne(it.attachedToClass, { _id: it.attachedTo }) - - // if (message !== undefined) { - // return await this.createThreadMessage(message, account, text, files) - // } - - // return false - } - - async replyToMention ( - it: MentionInboxNotification, - account: any, - text: string, - files: TelegramFileInfo[] - ): Promise { - // TODO: FIXME - throw new Error('Not implemented') - // const hierarchy = this.hierarchy - - // if (!hierarchy.isDerived(it.mentionedInClass, activity.class.ActivityMessage)) { - // return false - // } - - // const message = (await this.client.findOne(it.mentionedInClass, { _id: it.mentionedIn })) as ActivityMessage - - // if (message !== undefined) { - // return await this.createThreadMessage(message, account, text, files) - // } - - // return false - } - - async replyToNotification ( - account: any, + async replyToMessage ( + account: AccountUuid, + socialId: PersonId, record: MessageRecord, text: string, files: TelegramFileInfo[] ): Promise { - // TODO: FIXME - throw new Error('Not implemented') - // const inboxNotification = await this.client.findOne(notification.class.InboxNotification, { - // _id: record.notificationId - // }) + const message = await this.client.findOne(activity.class.ActivityMessage, { _id: record.messageId }) - // if (inboxNotification === undefined) { - // return false - // } - // const hierarchy = this.hierarchy - // if (hierarchy.isDerived(inboxNotification._class, notification.class.ActivityInboxNotification)) { - // return await this.replyToActivityNotification( - // inboxNotification as ActivityInboxNotification, - // account, - // text, - // files - // ) - // } else if (hierarchy.isDerived(inboxNotification._class, notification.class.MentionInboxNotification)) { - // return await this.replyToMention(inboxNotification as MentionInboxNotification, account, text, files) - // } + if (message === undefined) { + return false + } - // return false - } - - async replyToMessage (account: any, record: MessageRecord, text: string, files: TelegramFileInfo[]): Promise { - // TODO: FIXME - throw new Error('Not implemented') - // const message = await this.client.findOne(activity.class.ActivityMessage, { _id: record.messageId }) - - // if (message === undefined) { - // return false - // } - - // return await this.createThreadMessage(message, account, text, files) - } - - public async reply (record: MessageRecord, text: string, files: TelegramFileInfo[]): Promise { - // TODO: FIXME - throw new Error('Not implemented') - // const account = await this.client.getModel().findOne(contact.class.PersonAccount, { email: record.email }) - // if (account === undefined) { - // return false - // } - - // if (record.messageId != null) { - // return await this.replyToMessage(account, record, text, files) - // } - - // if (record.notificationId != null) { - // return await this.replyToNotification(account, record, text, files) - // } - - // return false - } - - async close (): Promise { - await this.client.close() + return await this.createThreadMessage(message, account, socialId, text, files) } async getFiles (_id: Ref): Promise { @@ -336,8 +244,9 @@ export class WorkspaceClient { const res: PlatformFileInfo[] = [] for (const attachment of attachments) { if (attachment.type === 'application/link-preview') continue - const chunks = await this.storageAdapter.read(this.ctx, this.workspace as any, attachment.file) // TODO: FIXME - const buffer = Buffer.concat(chunks) + const chunks: Buffer[] = await this.storage.read(this.ctx, this.workspace as any, attachment.file) + const uint8Chunks: Uint8Array[] = chunks.map((chunk) => new Uint8Array(chunk)) + const buffer = Buffer.concat(uint8Chunks) if (buffer.length > 0) { res.push({ buffer, @@ -349,103 +258,82 @@ export class WorkspaceClient { return res } - async getChannels (email: string, onlyStarred: boolean): Promise { - // TODO: FIXME - throw new Error('Not implemented') - // const account = await this.client.findOne(contact.class.PersonAccount, { email }) - // if (account === undefined) return [] + async getChannels (account: AccountUuid, onlyStarred: boolean): Promise { + if (!onlyStarred) { + return await this.client.findAll(chunter.class.ChunterSpace, { + members: account + }) + } - // if (!onlyStarred) { - // return await this.client.findAll(chunter.class.ChunterSpace, { - // members: account._id - // }) - // } + const contexts = await this.client.findAll(notification.class.DocNotifyContext, { + objectClass: { $in: [chunter.class.Channel, chunter.class.DirectMessage] }, + isPinned: true, + user: account + }) - // const contexts = await this.client.findAll(notification.class.DocNotifyContext, { - // objectClass: { $in: [chunter.class.Channel, chunter.class.DirectMessage] }, - // isPinned: true, - // user: account._id - // }) + if (contexts.length === 0) { + return [] + } - // if (contexts.length === 0) { - // return [] - // } - - // return await this.client.findAll(chunter.class.ChunterSpace, { - // _id: { $in: contexts.map((context) => context.objectId as Ref) }, - // members: account._id - // }) + return await this.client.findAll(chunter.class.ChunterSpace, { + _id: { $in: contexts.map((context) => context.objectId as Ref) }, + members: account + }) } - async getPersons (_ids: AccountUuid[], myEmail: string): Promise { - // TODO: FIXME - throw new Error('Not implemented') - // const me = await this.client.findOne(contact.class.PersonAccount, { email: myEmail }) - // const accounts = this.client.getModel().findAllSync(contact.class.PersonAccount, { _id: { $in: _ids } }) - // const persons = accounts.filter((account) => account.person !== me?.person).map(({ person }) => person) - // return await this.client.findAll(contact.class.Person, { _id: { $in: persons } }) + async getPersons (_ids: AccountUuid[]): Promise { + return (await this.client.findAll(contact.class.Person, { personUuid: { $in: _ids } })) as Person[] } async sendMessage ( channel: ChannelRecord, + account: AccountUuid, + socialId: PersonId, text: Markup, file?: TelegramFileInfo ): Promise | undefined> { - // TODO: FIXME - throw new Error('Not implemented') - // const account = await this.client.getModel().findOne(contact.class.PersonAccount, { email: channel.email }) + const doc = await this.client.findOne(channel._class, { _id: channel._id, members: account }) - // if (account === undefined) { - // return undefined - // } + if (doc === undefined) { + return undefined + } - // const doc = await this.client.findOne(channel.channelClass, { _id: channel.channelId, members: account._id }) + const txFactory = new TxFactory(socialId) + const messageId = generateId() + const attachments = await this.createAttachments( + txFactory, + messageId, + chunter.class.ChatMessage, + channel._id, + file !== undefined ? [file] : [] + ) - // if (doc === undefined) { - // return undefined - // } + if (attachments === 0 && isEmptyMarkup(text)) { + return undefined + } - // const txFactory = new TxFactory(account._id) - // const messageId = generateId() - // const attachments = await this.createAttachments( - // txFactory, - // messageId, - // chunter.class.ChatMessage, - // channel.channelId, - // file !== undefined ? [file] : [] - // ) + const collectionTx = txFactory.createTxCollectionCUD( + channel._class, + channel._id, + channel._id, + 'messages', + txFactory.createTxCreateDoc( + chunter.class.ChatMessage, + channel._id, + { + message: text, + attachments, + attachedTo: channel._id, + attachedToClass: channel._class, + collection: 'messages', + provider: contact.channelProvider.Telegram + }, + messageId + ) + ) - // if (attachments === 0 && isEmptyMarkup(text)) { - // return undefined - // } + await this.client.tx(collectionTx) - // const collectionTx = txFactory.createTxCollectionCUD( - // channel.channelClass, - // channel.channelId, - // channel.channelId, - // 'messages', - // txFactory.createTxCreateDoc( - // chunter.class.ChatMessage, - // channel.channelId, - // { - // message: text, - // attachments, - // attachedTo: channel.channelId, - // attachedToClass: channel.channelClass, - // collection: 'messages', - // provider: contact.channelProvider.Telegram - // }, - // messageId - // ) - // ) - - // await this.client.tx(collectionTx) - - // return messageId + return messageId } } - -async function connectPlatform (token: string): Promise { - const endpoint = await getTransactorEndpoint(token) - return await createClient(endpoint, token) -}