diff --git a/.env.example b/.env.example index 73a3b40d35..b0c5cdd094 100644 --- a/.env.example +++ b/.env.example @@ -50,6 +50,10 @@ EVENT_EMITTER_MAX_LISTENERS=50 # If you don't even want an expiration, enter the value false DEL_INSTANCE=false +# Maximum number of saved instances restored at the same time on startup. +# Prevents a reconnect storm when the service restarts with many accounts. +INSTANCE_LOAD_CONCURRENCY=10 + # Provider: postgresql | mysql | psql_bouncer DATABASE_PROVIDER=postgresql DATABASE_CONNECTION_URI='postgresql://user:pass@postgres:5432/evolution_db?schema=evolution_api' diff --git a/.github/workflows/check_code_quality.yml b/.github/workflows/check_code_quality.yml index df156f2117..da297a9bc8 100644 --- a/.github/workflows/check_code_quality.yml +++ b/.github/workflows/check_code_quality.yml @@ -2,9 +2,9 @@ name: Check Code Quality on: pull_request: - branches: [ main, develop ] + branches: [ main, homolog, develop ] push: - branches: [ main, develop ] + branches: [ main, homolog, develop ] jobs: check-lint-and-build: @@ -39,4 +39,4 @@ jobs: run: npm run db:generate - name: Check build - run: npm run build \ No newline at end of file + run: npm run build diff --git a/.github/workflows/publish_docker_image.yml b/.github/workflows/publish_docker_image.yml index c5a3996edb..58615e2504 100644 --- a/.github/workflows/publish_docker_image.yml +++ b/.github/workflows/publish_docker_image.yml @@ -2,9 +2,17 @@ name: Build Docker image on: push: + branches: + - main + - develop + - homolog tags: - "*.*.*" +env: + OCIR_REGISTRY: sa-saopaulo-1.ocir.io + IMAGE_NAME: sa-saopaulo-1.ocir.io/gr6emtyps1q4/evolution-api-v2 + jobs: build_deploy: name: Build and Deploy @@ -22,29 +30,31 @@ jobs: id: meta uses: docker/metadata-action@v5 with: - images: evoapicloud/evolution-api - tags: type=semver,pattern=v{{version}} - - - name: Set up QEMU - uses: docker/setup-qemu-action@v3 + images: ${{ env.IMAGE_NAME }} + tags: | + type=semver,pattern={{version}} + type=raw,value=stable,enable=${{ github.ref == 'refs/heads/main' || startsWith(github.ref, 'refs/tags/') }} + type=raw,value=hlg,enable=${{ github.ref == 'refs/heads/homolog' }} + type=raw,value=dev,enable=${{ github.ref == 'refs/heads/develop' }} - name: Set up Docker Buildx uses: docker/setup-buildx-action@v3 - - name: Login to GitHub Container Registry + - name: Login to OCIR uses: docker/login-action@v3 with: - username: ${{ secrets.DOCKER_USERNAME }} - password: ${{ secrets.DOCKER_PASSWORD }} + registry: ${{ env.OCIR_REGISTRY }} + username: ${{ secrets.OCI_TENANCY_NAME }}/${{ secrets.OCI_REGISTRY_USERNAME }} + password: ${{ secrets.OCI_AUTH_TOKEN }} - name: Build and push id: docker_build uses: docker/build-push-action@v6 with: - platforms: linux/amd64,linux/arm64 + platforms: linux/amd64 push: true tags: ${{ steps.meta.outputs.tags }} labels: ${{ steps.meta.outputs.labels }} - name: Image digest - run: echo ${{ steps.docker_build.outputs.digest }} \ No newline at end of file + run: echo ${{ steps.docker_build.outputs.digest }} diff --git a/.github/workflows/publish_docker_image_homolog.yml b/.github/workflows/publish_docker_image_homolog.yml deleted file mode 100644 index a76e1008be..0000000000 --- a/.github/workflows/publish_docker_image_homolog.yml +++ /dev/null @@ -1,50 +0,0 @@ -name: Build Docker image - -on: - push: - branches: - - develop - -jobs: - build_deploy: - name: Build and Deploy - runs-on: ubuntu-latest - permissions: - contents: read - packages: write - steps: - - name: Checkout - uses: actions/checkout@v5 - with: - submodules: recursive - - - name: Docker meta - id: meta - uses: docker/metadata-action@v5 - with: - images: evoapicloud/evolution-api - tags: homolog - - - name: Set up QEMU - uses: docker/setup-qemu-action@v3 - - - name: Set up Docker Buildx - uses: docker/setup-buildx-action@v3 - - - name: Login to Docker Hub - uses: docker/login-action@v3 - with: - username: ${{ secrets.DOCKER_USERNAME }} - password: ${{ secrets.DOCKER_PASSWORD }} - - - name: Build and push - id: docker_build - uses: docker/build-push-action@v6 - with: - platforms: linux/amd64,linux/arm64 - push: true - tags: ${{ steps.meta.outputs.tags }} - labels: ${{ steps.meta.outputs.labels }} - - - name: Image digest - run: echo ${{ steps.docker_build.outputs.digest }} diff --git a/.github/workflows/publish_docker_image_latest.yml b/.github/workflows/publish_docker_image_latest.yml deleted file mode 100644 index f73fe80e4e..0000000000 --- a/.github/workflows/publish_docker_image_latest.yml +++ /dev/null @@ -1,50 +0,0 @@ -name: Build Docker image - -on: - push: - branches: - - main - -jobs: - build_deploy: - name: Build and Deploy - runs-on: ubuntu-latest - permissions: - contents: read - packages: write - steps: - - name: Checkout - uses: actions/checkout@v5 - with: - submodules: recursive - - - name: Docker meta - id: meta - uses: docker/metadata-action@v5 - with: - images: evoapicloud/evolution-api - tags: latest - - - name: Set up QEMU - uses: docker/setup-qemu-action@v3 - - - name: Set up Docker Buildx - uses: docker/setup-buildx-action@v3 - - - name: Login to Docker Hub - uses: docker/login-action@v3 - with: - username: ${{ secrets.DOCKER_USERNAME }} - password: ${{ secrets.DOCKER_PASSWORD }} - - - name: Build and push - id: docker_build - uses: docker/build-push-action@v6 - with: - platforms: linux/amd64,linux/arm64 - push: true - tags: ${{ steps.meta.outputs.tags }} - labels: ${{ steps.meta.outputs.labels }} - - - name: Image digest - run: echo ${{ steps.docker_build.outputs.digest }} diff --git a/.github/workflows/security.yml b/.github/workflows/security.yml index b83f0b5d98..354b25570d 100644 --- a/.github/workflows/security.yml +++ b/.github/workflows/security.yml @@ -2,9 +2,9 @@ name: Security Scan on: push: - branches: [ main, develop ] + branches: [ main, homolog, develop ] pull_request: - branches: [ main, develop ] + branches: [ main, homolog, develop ] schedule: - cron: '0 0 * * 1' # Weekly on Mondays diff --git a/.gitignore b/.gitignore index 768d8afa41..f0d962c2dd 100644 --- a/.gitignore +++ b/.gitignore @@ -38,6 +38,7 @@ lerna-debug.log* /src/env.yml /store *.env +!deploy/**/.env.example /temp/* diff --git a/README.md b/README.md index 4586e4028a..92f28becd4 100644 --- a/README.md +++ b/README.md @@ -27,6 +27,13 @@ --- +## Meets deployment + +The Meets integration publishes the provider image from `develop`, `homolog`, +and `main`. Infrastructure provisioning is intentionally managed outside this +repository so the provider does not introduce dedicated databases, caches, +proxies, or Kubernetes resources without an approved environment plan. + ## About **Evolution API** is a powerful, production-ready REST API for WhatsApp and multi-channel messaging. Originally focused on WhatsApp, it has grown into a comprehensive platform supporting multiple messaging providers and integrations. diff --git a/chart/Chart.yaml b/chart/Chart.yaml new file mode 100644 index 0000000000..cc6c0422f7 --- /dev/null +++ b/chart/Chart.yaml @@ -0,0 +1,6 @@ +apiVersion: v2 +name: meets-evolution-api +description: Minimal Evolution API deployment for Meets environments +type: application +version: 0.1.0 +appVersion: '2' diff --git a/chart/templates/deployment.yaml b/chart/templates/deployment.yaml new file mode 100644 index 0000000000..337d6d2098 --- /dev/null +++ b/chart/templates/deployment.yaml @@ -0,0 +1,60 @@ +apiVersion: apps/v1 +kind: Deployment +metadata: + name: {{ .Release.Name }} + labels: + app.kubernetes.io/name: {{ .Chart.Name }} + app.kubernetes.io/instance: {{ .Release.Name }} +spec: + replicas: {{ .Values.replicaCount }} + selector: + matchLabels: + app.kubernetes.io/name: {{ .Chart.Name }} + app.kubernetes.io/instance: {{ .Release.Name }} + template: + metadata: + labels: + app.kubernetes.io/name: {{ .Chart.Name }} + app.kubernetes.io/instance: {{ .Release.Name }} + spec: + containers: + - name: evolution-api + image: '{{ .Values.image.repository }}:{{ .Values.image.tag }}' + imagePullPolicy: {{ .Values.image.pullPolicy }} + ports: + - name: http + containerPort: {{ .Values.service.port }} + envFrom: + - secretRef: + name: {{ .Values.secretName }} + env: + - name: ENVIRONMENT + value: {{ .Values.environment | quote }} +{{- with .Values.env }} +{{ toYaml . | nindent 12 }} +{{- end }} +{{- with .Values.nodeEnv }} +{{ toYaml . | nindent 12 }} +{{- end }} + resources: +{{ toYaml .Values.resources | nindent 12 }} + startupProbe: + httpGet: + path: / + port: http + failureThreshold: 30 + periodSeconds: 10 + readinessProbe: + httpGet: + path: / + port: http + initialDelaySeconds: 20 + periodSeconds: 10 + failureThreshold: 6 + livenessProbe: + httpGet: + path: / + port: http + initialDelaySeconds: 60 + periodSeconds: 30 + failureThreshold: 5 diff --git a/chart/templates/service.yaml b/chart/templates/service.yaml new file mode 100644 index 0000000000..a97da79c3d --- /dev/null +++ b/chart/templates/service.yaml @@ -0,0 +1,16 @@ +apiVersion: v1 +kind: Service +metadata: + name: {{ .Release.Name }} + labels: + app.kubernetes.io/name: {{ .Chart.Name }} + app.kubernetes.io/instance: {{ .Release.Name }} +spec: + type: ClusterIP + selector: + app.kubernetes.io/name: {{ .Chart.Name }} + app.kubernetes.io/instance: {{ .Release.Name }} + ports: + - name: http + port: {{ .Values.service.port }} + targetPort: http diff --git a/chart/values-dev.yaml b/chart/values-dev.yaml new file mode 100644 index 0000000000..ed8fba71cd --- /dev/null +++ b/chart/values-dev.yaml @@ -0,0 +1,35 @@ +environment: dev +image: + tag: dev +secretName: evolution-api-dev-secrets +env: + - name: SERVER_TYPE + value: http + - name: SERVER_PORT + value: '8080' + - name: DATABASE_ENABLED + value: 'true' + - name: DATABASE_PROVIDER + value: postgresql + - name: DATABASE_CONNECTION_CLIENT_NAME + value: evolution_dev + # Required by the gateway's paginated history/media synchronization. + # Data remains isolated in the existing DEV PostgreSQL instance. + - name: DATABASE_SAVE_DATA_NEW_MESSAGE + value: 'true' + - name: DATABASE_SAVE_MESSAGE_UPDATE + value: 'true' + - name: DATABASE_SAVE_DATA_HISTORIC + value: 'true' + - name: CACHE_REDIS_ENABLED + value: 'true' + - name: CACHE_REDIS_PREFIX_KEY + value: 'evolution:dev' + - name: CACHE_REDIS_SAVE_INSTANCES + value: 'false' + - name: CACHE_LOCAL_ENABLED + value: 'false' + - name: EVOLUTION_OWNERSHIP_ENABLED + value: 'false' + - name: TELEMETRY_ENABLED + value: 'false' diff --git a/chart/values-hlg-a.yaml b/chart/values-hlg-a.yaml new file mode 100644 index 0000000000..2a10dd4598 --- /dev/null +++ b/chart/values-hlg-a.yaml @@ -0,0 +1,5 @@ +nodeEnv: + - name: EVOLUTION_NODE_ID + value: evolution-hlg-a + - name: EVOLUTION_NODE_BASE_URL + value: http://evolution-api-hlg-a.hlg.svc.cluster.local:8080 diff --git a/chart/values-hlg-b.yaml b/chart/values-hlg-b.yaml new file mode 100644 index 0000000000..8f55f00e10 --- /dev/null +++ b/chart/values-hlg-b.yaml @@ -0,0 +1,5 @@ +nodeEnv: + - name: EVOLUTION_NODE_ID + value: evolution-hlg-b + - name: EVOLUTION_NODE_BASE_URL + value: http://evolution-api-hlg-b.hlg.svc.cluster.local:8080 diff --git a/chart/values-hlg.yaml b/chart/values-hlg.yaml new file mode 100644 index 0000000000..40098e3492 --- /dev/null +++ b/chart/values-hlg.yaml @@ -0,0 +1,31 @@ +environment: hlg +image: + tag: hlg +secretName: evolution-api-hlg-secrets +env: + - name: SERVER_TYPE + value: http + - name: SERVER_PORT + value: '8080' + - name: DATABASE_ENABLED + value: 'true' + - name: DATABASE_PROVIDER + value: postgresql + - name: DATABASE_CONNECTION_CLIENT_NAME + value: evolution_hlg + - name: CACHE_REDIS_ENABLED + value: 'true' + - name: CACHE_REDIS_PREFIX_KEY + value: 'evolution:hlg' + - name: CACHE_REDIS_SAVE_INSTANCES + value: 'false' + - name: CACHE_LOCAL_ENABLED + value: 'false' + - name: EVOLUTION_OWNERSHIP_ENABLED + value: 'true' + - name: EVOLUTION_OWNERSHIP_LEASE_TTL_MS + value: '90000' + - name: EVOLUTION_OWNERSHIP_RENEW_INTERVAL_MS + value: '30000' + - name: TELEMETRY_ENABLED + value: 'false' diff --git a/chart/values-prod-a.yaml b/chart/values-prod-a.yaml new file mode 100644 index 0000000000..cc528c4ccc --- /dev/null +++ b/chart/values-prod-a.yaml @@ -0,0 +1,5 @@ +nodeEnv: + - name: EVOLUTION_NODE_ID + value: evolution-prod-a + - name: EVOLUTION_NODE_BASE_URL + value: http://evolution-api-prod-a.prod.svc.cluster.local:8080 diff --git a/chart/values-prod-b.yaml b/chart/values-prod-b.yaml new file mode 100644 index 0000000000..09867c591a --- /dev/null +++ b/chart/values-prod-b.yaml @@ -0,0 +1,5 @@ +nodeEnv: + - name: EVOLUTION_NODE_ID + value: evolution-prod-b + - name: EVOLUTION_NODE_BASE_URL + value: http://evolution-api-prod-b.prod.svc.cluster.local:8080 diff --git a/chart/values-prod.yaml b/chart/values-prod.yaml new file mode 100644 index 0000000000..b72cee385a --- /dev/null +++ b/chart/values-prod.yaml @@ -0,0 +1,38 @@ +environment: prod +image: + tag: stable +secretName: evolution-api-prod-secrets +resources: + requests: + cpu: 250m + memory: 768Mi + limits: + cpu: '1' + memory: 1536Mi +env: + - name: SERVER_TYPE + value: http + - name: SERVER_PORT + value: '8080' + - name: DATABASE_ENABLED + value: 'true' + - name: DATABASE_PROVIDER + value: postgresql + - name: DATABASE_CONNECTION_CLIENT_NAME + value: evolution_prod + - name: CACHE_REDIS_ENABLED + value: 'true' + - name: CACHE_REDIS_PREFIX_KEY + value: 'evolution:prod' + - name: CACHE_REDIS_SAVE_INSTANCES + value: 'false' + - name: CACHE_LOCAL_ENABLED + value: 'false' + - name: EVOLUTION_OWNERSHIP_ENABLED + value: 'true' + - name: EVOLUTION_OWNERSHIP_LEASE_TTL_MS + value: '90000' + - name: EVOLUTION_OWNERSHIP_RENEW_INTERVAL_MS + value: '30000' + - name: TELEMETRY_ENABLED + value: 'false' diff --git a/chart/values.yaml b/chart/values.yaml new file mode 100644 index 0000000000..0a2682be8c --- /dev/null +++ b/chart/values.yaml @@ -0,0 +1,22 @@ +replicaCount: 1 + +image: + repository: evoapicloud/evolution-api + tag: dev + pullPolicy: Always + +service: + port: 8080 + +secretName: evolution-api-secrets + +resources: + requests: + cpu: 100m + memory: 512Mi + limits: + cpu: 500m + memory: 1Gi + +env: [] +nodeEnv: [] diff --git a/deploy/README.md b/deploy/README.md new file mode 100644 index 0000000000..8eb2683358 --- /dev/null +++ b/deploy/README.md @@ -0,0 +1,58 @@ +# Meets deployment + +This chart deploys an internal Evolution provider node. It creates only a +`Deployment` and a `ClusterIP` Service. PostgreSQL, Redis, volumes, +Ingress, HPA, namespaces and Secrets are intentionally external resources. + +## Prerequisites + +For each environment, Infrastructure must provision a dedicated PostgreSQL +database and a dedicated Redis service for Evolution. Create the Kubernetes +Secret from the approved secret manager; do not commit its values. + +The Secret must contain these keys: + +```text +AUTHENTICATION_API_KEY +DATABASE_CONNECTION_URI +CACHE_REDIS_URI +``` + +For example, after loading the values into a protected local file: + +```bash +kubectl -n dev create secret generic evolution-api-dev-secrets \ + --from-env-file=/secure/path/evolution-dev.env +``` + +The database URI must select PostgreSQL and the environment database, for +example `postgresql://:@:5432/evolution_dev`. + +## Deploy + +```bash +helm upgrade --install evolution-api-dev chart/ \ + --namespace dev \ + --values chart/values-dev.yaml \ + --wait --timeout 10m +``` + +DEV uses one provider with ownership disabled. HLG and production use two +separate releases, each with its own stable Service identity: + +```bash +helm upgrade --install evolution-api-hlg-a chart/ --namespace hlg \ + --values chart/values-hlg.yaml --values chart/values-hlg-a.yaml +helm upgrade --install evolution-api-hlg-b chart/ --namespace hlg \ + --values chart/values-hlg.yaml --values chart/values-hlg-b.yaml +``` + +The equivalent production files are `values-prod.yaml`, `values-prod-a.yaml` +and `values-prod-b.yaml`. They enable ownership with a 90-second lease renewed +every 30 seconds. Do not deploy the production pair until the HLG ownership +acceptance test has passed. The public gateway remains a separate service; this +provider has no Ingress and is reached through its internal Kubernetes Service. + +Before the first rollout in an environment, run the Prisma migration using the +same Secret-backed variables and then validate QR, reconnect, text, media and +webhook delivery with one WhatsApp account. diff --git a/src/api/controllers/chat.controller.ts b/src/api/controllers/chat.controller.ts index 22e90b9faa..27a84d48a9 100644 --- a/src/api/controllers/chat.controller.ts +++ b/src/api/controllers/chat.controller.ts @@ -2,6 +2,7 @@ import { ArchiveChatDto, BlockUserDto, DeleteMessage, + FetchMessageHistoryDto, getBase64FromMediaMessageDto, MarkChatUnreadDto, NumberDto, @@ -10,6 +11,7 @@ import { ProfilePictureDto, ProfileStatusDto, ReadMessageDto, + ResolveLidDto, SendPresenceDto, UpdateMessageDto, WhatsAppNumberDto, @@ -26,6 +28,14 @@ export class ChatController { return await this.waMonitor.waInstances[instanceName].whatsappNumber(data); } + public async resolveLid({ instanceName }: InstanceDto, data: ResolveLidDto) { + const instance = this.waMonitor.waInstances[instanceName]; + if (typeof instance.resolveLid !== 'function') { + return { lid: data.lid, phone: null }; + } + return await instance.resolveLid(data); + } + public async readMessage({ instanceName }: InstanceDto, data: ReadMessageDto) { return await this.waMonitor.waInstances[instanceName].markMessageAsRead(data); } @@ -62,6 +72,10 @@ export class ChatController { return await this.waMonitor.waInstances[instanceName].fetchMessages(query); } + public async fetchMessageHistory({ instanceName }: InstanceDto, data: FetchMessageHistoryDto) { + return await this.waMonitor.waInstances[instanceName].fetchMessageHistory(data); + } + public async fetchStatusMessage({ instanceName }: InstanceDto, query: Query) { return await this.waMonitor.waInstances[instanceName].fetchStatusMessage(query); } diff --git a/src/api/dto/chat.dto.ts b/src/api/dto/chat.dto.ts index b11f32b054..46d5e2e3f6 100644 --- a/src/api/dto/chat.dto.ts +++ b/src/api/dto/chat.dto.ts @@ -22,10 +22,20 @@ export class getBase64FromMediaMessageDto { convertToMp4?: boolean; } +export class FetchMessageHistoryDto { + count?: number; + key: proto.IMessageKey; + messageTimestamp: number; +} + export class WhatsAppNumberDto { numbers: string[]; } +export class ResolveLidDto { + lid: string; +} + export class NumberDto { number: string; } diff --git a/src/api/integrations/channel/whatsapp/whatsapp.baileys.service.ts b/src/api/integrations/channel/whatsapp/whatsapp.baileys.service.ts index 60e857fcc1..60716e2269 100644 --- a/src/api/integrations/channel/whatsapp/whatsapp.baileys.service.ts +++ b/src/api/integrations/channel/whatsapp/whatsapp.baileys.service.ts @@ -11,6 +11,7 @@ import { OnWhatsAppDto, PrivacySettingDto, ReadMessageDto, + ResolveLidDto, SendPresenceDto, UpdateMessageDto, WhatsAppNumberDto, @@ -739,6 +740,13 @@ export class BaileysStartupService extends ChannelStartupService { } } + public async restart(): Promise { + if (!this.client) { + throw new BadRequestException('WhatsApp connection is not available'); + } + this.client.ws?.close(); + } + public async reloadConnection(): Promise { try { return await this.createClient(this.phoneNumber); @@ -1030,7 +1038,7 @@ export class BaileysStartupService extends ChannelStartupService { } } - if (messagesRepository?.has(m.key.id)) { + if (syncType !== proto.HistorySync.HistorySyncType.ON_DEMAND && messagesRepository?.has(m.key.id)) { continue; } @@ -3673,6 +3681,14 @@ export class BaileysStartupService extends ChannelStartupService { return onWhatsapp; } + public async resolveLid(data: ResolveLidDto) { + const phone = await this.client.signalRepository.lidMapping.getPNForLID(data.lid); + return { + lid: data.lid, + phone: phone ? jidNormalizedUser(phone) : null, + }; + } + public async markMessageAsRead(data: ReadMessageDto) { try { const keys: proto.IMessageKey[] = []; @@ -5119,4 +5135,25 @@ export class BaileysStartupService extends ChannelStartupService { }, }; } + + public async fetchMessageHistory(data: { count?: number; key: proto.IMessageKey; messageTimestamp: number }) { + if (!this.client) { + throw new BadRequestException('WhatsApp connection is not available'); + } + const count = Math.min(Math.max(Number(data.count) || 50, 1), 100); + const suppliedTimestamp = Number(data.messageTimestamp); + if (!Number.isFinite(suppliedTimestamp)) { + throw new BadRequestException('messageTimestamp is required to fetch message history'); + } + const messageTimestamp = + suppliedTimestamp >= 100_000_000_000 ? Math.floor(suppliedTimestamp / 1_000) : Math.floor(suppliedTimestamp); + const requestId = await this.client.fetchMessageHistory(count, data.key, Math.floor(messageTimestamp)); + + return { + status: 'requested', + requestId, + count, + remoteJid: data.key.remoteJid, + }; + } } diff --git a/src/api/routes/chat.router.ts b/src/api/routes/chat.router.ts index 158947ed22..0a5806e8e1 100644 --- a/src/api/routes/chat.router.ts +++ b/src/api/routes/chat.router.ts @@ -3,6 +3,7 @@ import { ArchiveChatDto, BlockUserDto, DeleteMessage, + FetchMessageHistoryDto, getBase64FromMediaMessageDto, MarkChatUnreadDto, NumberDto, @@ -11,6 +12,7 @@ import { ProfilePictureDto, ProfileStatusDto, ReadMessageDto, + ResolveLidDto, SendPresenceDto, UpdateMessageDto, WhatsAppNumberDto, @@ -24,6 +26,7 @@ import { blockUserSchema, contactValidateSchema, deleteMessageSchema, + fetchMessageHistorySchema, markChatUnreadSchema, messageUpSchema, messageValidateSchema, @@ -34,6 +37,7 @@ import { profileSchema, profileStatusSchema, readMessageSchema, + resolveLidSchema, updateMessageSchema, whatsappNumberSchema, } from '@validate/validate.schema'; @@ -60,6 +64,16 @@ export class ChatRouter extends RouterBroker { return res.status(HttpStatus.BAD_REQUEST).json(error); } }) + .post(this.routerPath('resolveLid'), ...guards, async (req, res) => { + const response = await this.dataValidate({ + request: req, + schema: resolveLidSchema, + ClassRef: ResolveLidDto, + execute: (instance, data) => chatController.resolveLid(instance, data), + }); + + return res.status(HttpStatus.OK).json(response); + }) .post(this.routerPath('markMessageAsRead'), ...guards, async (req, res) => { const response = await this.dataValidate({ request: req, @@ -171,6 +185,16 @@ export class ChatRouter extends RouterBroker { return res.status(HttpStatus.OK).json(response); }) + .post(this.routerPath('fetchMessageHistory'), ...guards, async (req, res) => { + const response = await this.dataValidate({ + request: req, + schema: fetchMessageHistorySchema, + ClassRef: FetchMessageHistoryDto, + execute: (instance, data) => chatController.fetchMessageHistory(instance, data), + }); + + return res.status(HttpStatus.ACCEPTED).json(response); + }) .post(this.routerPath('findStatusMessage'), ...guards, async (req, res) => { const response = await this.dataValidate>({ request: req, diff --git a/src/api/routes/index.router.ts b/src/api/routes/index.router.ts index 45c43fca5b..8be238902b 100644 --- a/src/api/routes/index.router.ts +++ b/src/api/routes/index.router.ts @@ -27,6 +27,7 @@ import { ViewsRouter } from './view.router'; enum HttpStatus { OK = 200, + ACCEPTED = 202, CREATED = 201, NOT_FOUND = 404, FORBIDDEN = 403, diff --git a/src/api/server.module.ts b/src/api/server.module.ts index 668b9e2722..3409911fde 100644 --- a/src/api/server.module.ts +++ b/src/api/server.module.ts @@ -40,6 +40,7 @@ import { S3Service } from './integrations/storage/s3/services/s3.service'; import { ProviderFiles } from './provider/sessions'; import { PrismaRepository } from './repository/repository.service'; import { CacheService } from './services/cache.service'; +import { InstanceOwnershipService } from './services/instance-ownership.service'; import { WAMonitoringService } from './services/monitor.service'; import { ProxyService } from './services/proxy.service'; import { SettingsService } from './services/settings.service'; @@ -54,6 +55,7 @@ if (configService.get('CHATWOOT').ENABLED) { export const cache = new CacheService(new CacheEngine(configService, 'instance').getEngine()); const baileysCache = new CacheService(new CacheEngine(configService, 'baileys').getEngine()); +const instanceOwnership = new InstanceOwnershipService(configService); let providerFiles: ProviderFiles = null; if (configService.get('PROVIDER').ENABLED) { @@ -70,6 +72,7 @@ export const waMonitor = new WAMonitoringService( cache, chatwootCache, baileysCache, + instanceOwnership, ); const s3Service = new S3Service(prismaRepository); diff --git a/src/api/services/instance-ownership.service.ts b/src/api/services/instance-ownership.service.ts new file mode 100644 index 0000000000..fdb5cf1d5c --- /dev/null +++ b/src/api/services/instance-ownership.service.ts @@ -0,0 +1,101 @@ +import { redisClient } from '@cache/rediscache.client'; +import { ConfigService, InstanceOwnership } from '@config/env.config'; +import { Logger } from '@config/logger.config'; +import { randomUUID } from 'crypto'; + +type OwnerLease = { + nodeId: string; + baseUrl: string; + leaseVersion: string; +}; + +export class InstanceOwnershipService { + private readonly logger = new Logger('InstanceOwnershipService'); + private readonly leases = new Map(); + private readonly renewals = new Map(); + + constructor(private readonly config: ConfigService) {} + + async acquire(instanceName: string): Promise { + const settings = this.settings(); + if (!settings.ENABLED) return true; + if (!settings.NODE_ID || !settings.NODE_BASE_URL) { + throw new Error('Evolution ownership requires EVOLUTION_NODE_ID and EVOLUTION_NODE_BASE_URL'); + } + + const client = await this.connection(); + const lease: OwnerLease = { + nodeId: settings.NODE_ID, + baseUrl: settings.NODE_BASE_URL, + leaseVersion: randomUUID(), + }; + const acquired = await client.set(this.key(instanceName), JSON.stringify(lease), { + NX: true, + PX: settings.LEASE_TTL_MS, + }); + if (acquired !== 'OK') return false; + this.leases.set(instanceName, lease); + return true; + } + + startRenewal(instanceName: string, onLost: () => Promise): void { + const settings = this.settings(); + if (!settings.ENABLED || this.renewals.has(instanceName)) return; + const timer = setInterval(async () => { + if (await this.renew(instanceName)) return; + this.stopRenewal(instanceName); + await onLost(); + }, settings.RENEW_INTERVAL_MS); + this.renewals.set(instanceName, timer); + } + + async release(instanceName: string): Promise { + this.stopRenewal(instanceName); + const settings = this.settings(); + const lease = this.leases.get(instanceName); + this.leases.delete(instanceName); + if (!settings.ENABLED || !lease) return; + const client = await this.connection(); + await client.eval( + "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end", + { keys: [this.key(instanceName)], arguments: [JSON.stringify(lease)] }, + ); + } + + private async renew(instanceName: string): Promise { + const settings = this.settings(); + const lease = this.leases.get(instanceName); + if (!lease) return false; + try { + const client = await this.connection(); + const result = await client.eval( + "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('pexpire', KEYS[1], ARGV[2]) else return 0 end", + { keys: [this.key(instanceName)], arguments: [JSON.stringify(lease), String(settings.LEASE_TTL_MS)] }, + ); + return result === 1; + } catch (error) { + this.logger.error({ action: 'ownership.renew', instanceName, error }); + return false; + } + } + + private stopRenewal(instanceName: string): void { + const timer = this.renewals.get(instanceName); + if (timer) clearInterval(timer); + this.renewals.delete(instanceName); + } + + private settings(): InstanceOwnership { + return this.config.get('OWNERSHIP'); + } + + private key(instanceName: string): string { + return `evolution:owner:${instanceName}`; + } + + private async connection() { + const client = redisClient.getConnection(); + if (!client.isOpen) await client.connect(); + return client; + } +} diff --git a/src/api/services/monitor.service.ts b/src/api/services/monitor.service.ts index 438530b57e..158b96d15c 100644 --- a/src/api/services/monitor.service.ts +++ b/src/api/services/monitor.service.ts @@ -3,7 +3,15 @@ import { ProviderFiles } from '@api/provider/sessions'; import { PrismaRepository } from '@api/repository/repository.service'; import { channelController } from '@api/server.module'; import { Events, Integration } from '@api/types/wa.types'; -import { CacheConf, Chatwoot, ConfigService, Database, DelInstance, ProviderSession } from '@config/env.config'; +import { + CacheConf, + Chatwoot, + ConfigService, + Database, + DelInstance, + ProviderSession, + Startup, +} from '@config/env.config'; import { Logger } from '@config/logger.config'; import { INSTANCE_DIR, STORE_DIR } from '@config/path.config'; import { NotFoundException } from '@exceptions'; @@ -13,6 +21,7 @@ import { rmSync } from 'fs'; import { join } from 'path'; import { CacheService } from './cache.service'; +import { InstanceOwnershipService } from './instance-ownership.service'; export class WAMonitoringService { constructor( @@ -23,6 +32,7 @@ export class WAMonitoringService { private readonly cache: CacheService, private readonly chatwootCache: CacheService, private readonly baileysCache: CacheService, + private readonly ownership: InstanceOwnershipService, ) { this.removeInstance(); this.noConnection(); @@ -271,6 +281,10 @@ export class WAMonitoringService { } private async setInstance(instanceData: InstanceDto) { + if (!(await this.ownership.acquire(instanceData.instanceName))) { + this.logger.warn(`Skipping instance "${instanceData.instanceName}" because another node owns it`); + return; + } const instance = channelController.init(instanceData, { configService: this.configService, eventEmitter: this.eventEmitter, @@ -281,7 +295,10 @@ export class WAMonitoringService { providerFiles: this.providerFiles, }); - if (!instance) return; + if (!instance) { + await this.ownership.release(instanceData.instanceName); + return; + } instance.setInstance({ instanceId: instanceData.instanceId, @@ -305,35 +322,38 @@ export class WAMonitoringService { } this.waInstances[instanceData.instanceName] = instance; + this.ownership.startRenewal(instanceData.instanceName, async () => { + this.logger.error(`Lost ownership of instance "${instanceData.instanceName}"`); + await this.waInstances[instanceData.instanceName]?.client?.logout('Lost Evolution ownership'); + delete this.waInstances[instanceData.instanceName]; + }); } private async loadInstancesFromRedis() { const keys = await this.cache.keys(); if (keys?.length > 0) { - await Promise.all( - keys.map(async (k) => { - const instanceData = await this.prismaRepository.instance.findUnique({ - where: { id: k.split(':')[1] }, - }); - - if (!instanceData) { - return; - } + await this.loadInBatches(keys, async (k) => { + const instanceData = await this.prismaRepository.instance.findUnique({ + where: { id: k.split(':')[1] }, + }); - const instance = { - instanceId: k.split(':')[1], - instanceName: k.split(':')[2], - integration: instanceData.integration, - token: instanceData.token, - number: instanceData.number, - businessId: instanceData.businessId, - connectionStatus: instanceData.connectionStatus as any, // Pass connection status - }; - - this.setInstance(instance); - }), - ); + if (!instanceData) { + return; + } + + const instance = { + instanceId: k.split(':')[1], + instanceName: k.split(':')[2], + integration: instanceData.integration, + token: instanceData.token, + number: instanceData.number, + businessId: instanceData.businessId, + connectionStatus: instanceData.connectionStatus as any, // Pass connection status + }; + + await this.setInstance(instance); + }); } } @@ -348,20 +368,18 @@ export class WAMonitoringService { return; } - await Promise.all( - instances.map(async (instance) => { - this.setInstance({ - instanceId: instance.id, - instanceName: instance.name, - integration: instance.integration, - token: instance.token, - number: instance.number, - businessId: instance.businessId, - ownerJid: instance.ownerJid, - connectionStatus: instance.connectionStatus as any, // Pass connection status - }); - }), - ); + await this.loadInBatches(instances, async (instance) => { + await this.setInstance({ + instanceId: instance.id, + instanceName: instance.name, + integration: instance.integration, + token: instance.token, + number: instance.number, + businessId: instance.businessId, + ownerJid: instance.ownerJid, + connectionStatus: instance.connectionStatus as any, // Pass connection status + }); + }); } private async loadInstancesFromProvider() { @@ -371,22 +389,28 @@ export class WAMonitoringService { return; } - await Promise.all( - instances?.data?.map(async (instanceId: string) => { - const instance = await this.prismaRepository.instance.findUnique({ - where: { id: instanceId }, - }); + await this.loadInBatches(instances.data, async (instanceId: string) => { + const instance = await this.prismaRepository.instance.findUnique({ + where: { id: instanceId }, + }); - this.setInstance({ - instanceId: instance.id, - instanceName: instance.name, - integration: instance.integration, - token: instance.token, - businessId: instance.businessId, - connectionStatus: instance.connectionStatus as any, // Pass connection status - }); - }), - ); + await this.setInstance({ + instanceId: instance.id, + instanceName: instance.name, + integration: instance.integration, + token: instance.token, + businessId: instance.businessId, + connectionStatus: instance.connectionStatus as any, // Pass connection status + }); + }); + } + + private async loadInBatches(items: T[], load: (item: T) => Promise): Promise { + const { INSTANCE_LOAD_CONCURRENCY: concurrency } = this.configService.get('STARTUP'); + + for (let index = 0; index < items.length; index += concurrency) { + await Promise.all(items.slice(index, index + concurrency).map(load)); + } } private removeInstance() { @@ -404,6 +428,7 @@ export class WAMonitoringService { try { delete this.waInstances[instanceName]; + await this.ownership.release(instanceName); } catch (error) { this.logger.error(error); } diff --git a/src/config/env.config.ts b/src/config/env.config.ts index 7c4e382e7e..4f7d79034e 100644 --- a/src/config/env.config.ts +++ b/src/config/env.config.ts @@ -389,6 +389,18 @@ export type EventEmitter = { MAX_LISTENERS: number; }; +export type Startup = { + INSTANCE_LOAD_CONCURRENCY: number; +}; + +export type InstanceOwnership = { + ENABLED: boolean; + NODE_ID: string; + NODE_BASE_URL: string; + LEASE_TTL_MS: number; + RENEW_INTERVAL_MS: number; +}; + export type Production = boolean; export interface Env { @@ -406,6 +418,7 @@ export interface Env { LOG: Log; DEL_INSTANCE: DelInstance; DEL_TEMP_INSTANCES: boolean; + OWNERSHIP: InstanceOwnership; LANGUAGE: Language; WEBHOOK: Webhook; PUSHER: Pusher; @@ -428,6 +441,7 @@ export interface Env { FACEBOOK: Facebook; SENTRY: Sentry; EVENT_EMITTER: EventEmitter; + STARTUP: Startup; PRODUCTION?: Production; } @@ -743,6 +757,19 @@ export class ConfigService { DEL_TEMP_INSTANCES: isBooleanString(process.env?.DEL_TEMP_INSTANCES) ? process.env.DEL_TEMP_INSTANCES === 'true' : true, + STARTUP: { + INSTANCE_LOAD_CONCURRENCY: (() => { + const value = Number.parseInt(process.env.INSTANCE_LOAD_CONCURRENCY || '10', 10); + return Number.isFinite(value) ? Math.max(1, value) : 10; + })(), + }, + OWNERSHIP: { + ENABLED: process.env?.EVOLUTION_OWNERSHIP_ENABLED === 'true', + NODE_ID: process.env?.EVOLUTION_NODE_ID || '', + NODE_BASE_URL: process.env?.EVOLUTION_NODE_BASE_URL || '', + LEASE_TTL_MS: Number.parseInt(process.env?.EVOLUTION_OWNERSHIP_LEASE_TTL_MS || '90000', 10), + RENEW_INTERVAL_MS: Number.parseInt(process.env?.EVOLUTION_OWNERSHIP_RENEW_INTERVAL_MS || '30000', 10), + }, LANGUAGE: process.env?.LANGUAGE || 'en', WEBHOOK: { GLOBAL: { diff --git a/src/validate/chat.schema.ts b/src/validate/chat.schema.ts index 7dae44539b..0b1bc1b018 100644 --- a/src/validate/chat.schema.ts +++ b/src/validate/chat.schema.ts @@ -41,6 +41,18 @@ export const whatsappNumberSchema: JSONSchema7 = { }, }; +export const resolveLidSchema: JSONSchema7 = { + $id: v4(), + type: 'object', + properties: { + lid: { + type: 'string', + pattern: '^[0-9]+@lid$', + }, + }, + required: ['lid'], +}; + export const readMessageSchema: JSONSchema7 = { $id: v4(), type: 'object', @@ -241,6 +253,26 @@ export const messageValidateSchema: JSONSchema7 = { }, }; +export const fetchMessageHistorySchema: JSONSchema7 = { + $id: v4(), + type: 'object', + required: ['key', 'messageTimestamp'], + properties: { + count: { type: 'integer', minimum: 1, maximum: 100 }, + messageTimestamp: { type: 'integer', minimum: 1 }, + key: { + type: 'object', + required: ['id', 'remoteJid'], + properties: { + id: { type: 'string', minLength: 1 }, + remoteJid: { type: 'string', minLength: 1 }, + fromMe: { type: 'boolean' }, + participant: { type: 'string' }, + }, + }, + }, +}; + export const messageUpSchema: JSONSchema7 = { $id: v4(), type: 'object',