diff --git a/apps/dumili/api/.env b/apps/dumili/api/.env index 7f2e9383c..3d1f6c552 100644 --- a/apps/dumili/api/.env +++ b/apps/dumili/api/.env @@ -11,3 +11,5 @@ MYSQL_ROOT_PASSWORD=changeme SENTRY_DSN= TOKEN_SECRET=3543c30fe79047b4f73cfb61aa1eb52cb3173de4b3941e0fc4ec1b127bbeed6019695a1a453a81c33c2eea964ccc577e69c7df994124bd2751e262a311ea23a1 + +REDIS_URL=redis://localhost:6379 diff --git a/apps/dumili/api/Dockerfile.worker b/apps/dumili/api/Dockerfile.worker new file mode 100644 index 000000000..5082e9d18 --- /dev/null +++ b/apps/dumili/api/Dockerfile.worker @@ -0,0 +1,11 @@ +FROM node:26-slim +LABEL org.opencontainers.image.authors="Bruno Perel" +WORKDIR /app + +RUN apt-get update && apt-get install -y openssl && apt-get clean && rm -rf /var/lib/apt/lists/* + +COPY apps/dumili/api/.env /app/ +COPY apps/dumili/api/dist/worker.mjs /app/worker.mjs +COPY apps/dumili/api/dist/instrument.mjs /app/instrument.mjs + +CMD ["node", "--import", "./instrument.mjs", "worker.mjs"] diff --git a/apps/dumili/api/docker-compose-dev.yml b/apps/dumili/api/docker-compose-dev.yml index 4e0d31f5f..887a42783 100644 --- a/apps/dumili/api/docker-compose-dev.yml +++ b/apps/dumili/api/docker-compose-dev.yml @@ -22,6 +22,15 @@ services: build: context: ../../.. dockerfile: apps/dumili/api/paddleocr/Dockerfile + redis: + container_name: dumili-redis + restart: always + image: redis:7-alpine + command: redis-server --appendonly yes + ports: + - "6379:6379" + volumes: + - redis-data:/data db: container_name: dumili-db restart: always @@ -35,3 +44,6 @@ services: environment: MYSQL_DATABASE: dumili MARIADB_ROOT_PASSWORD: $MYSQL_ROOT_PASSWORD + +volumes: + redis-data: diff --git a/apps/dumili/api/docker-compose.yml b/apps/dumili/api/docker-compose.yml index e1e38dfcc..8b7832ce6 100644 --- a/apps/dumili/api/docker-compose.yml +++ b/apps/dumili/api/docker-compose.yml @@ -23,10 +23,29 @@ services: container_name: dumili-api image: ghcr.io/bperel/dumili-api restart: always + environment: + REDIS_URL: redis://redis:6379 networks: - public-network - dumili-network - dm-api-network + + dumili-worker: + container_name: dumili-worker + image: ghcr.io/bperel/dumili-worker + restart: always + environment: + REDIS_URL: redis://redis:6379 + networks: + - dumili-network + - dm-api-network + + redis: + container_name: dumili-redis + image: redis:7-alpine + restart: always + networks: + - dumili-network db: container_name: dumili-db restart: always diff --git a/apps/dumili/api/index.ts b/apps/dumili/api/index.ts index b11917956..43ba95537 100644 --- a/apps/dumili/api/index.ts +++ b/apps/dumili/api/index.ts @@ -8,12 +8,14 @@ dotenv.config({ override: true, }); +import { createAdapter } from "@socket.io/redis-adapter"; import { v2 as cloudinary } from "cloudinary"; import { createServer, type IncomingMessage, type ServerResponse } from "http"; import { Server } from "socket.io"; import type { SessionUser } from "~dm-types/SessionUser"; +import { createRedisClient } from "./queue/connection"; import { authenticateUser } from "./services/_auth"; import type { FullIndexation } from "./services/indexation"; import { handleHttpFileUpload, server as indexation } from "./services/indexation"; @@ -40,6 +42,9 @@ const io = new Server(httpServer, { maxHttpBufferSize: 100 * 1024 * 1024, }); +const pubClient = createRedisClient(); +io.adapter(createAdapter(pubClient, pubClient.duplicate())); + indexations(io); indexation(io); diff --git a/apps/dumili/api/package.json b/apps/dumili/api/package.json index 20e395167..87afbb8d1 100644 --- a/apps/dumili/api/package.json +++ b/apps/dumili/api/package.json @@ -7,15 +7,18 @@ }, "scripts": { "clean": "rm -rf dist", - "build": "prisma generate && for source in index.ts instrument.ts; do bun build --entry-naming [dir]/[name].mjs --sourcemap=linked --target node --external @napi-rs/canvas $source --outdir dist; done && pnpm run copy-wasm && pnpm sentry:sourcemaps", + "build": "prisma generate && for source in index.ts instrument.ts worker.ts; do bun build --entry-naming [dir]/[name].mjs --sourcemap=linked --target node --external @napi-rs/canvas $source --outdir dist; done && pnpm run copy-wasm && pnpm sentry:sourcemaps", "copy-wasm": "cp node_modules/node-unrar-js/dist/js/unrar.wasm dist/ && cp node_modules/pdfjs-dist/build/pdf.worker.mjs dist/", "prisma-pull-generate": "prisma db pull && prisma generate", "prisma-generate": "prisma generate", "lint": "eslint --fix .", "dev": "docker compose -f docker-compose-dev.yml up --force-recreate -d && bash ../../../packages/prisma-schemas/wait-until-db-ready.bash && prisma generate && concurrently --kill-others-on-fail -n docker-compose,typecheck \"docker compose -f docker-compose-dev.yml logs -f\" \"tsc --noEmit\"", "dev:bun": "bun --inspect run index.ts", + "dev:worker": "bun run worker.ts", + "dev:all": "concurrently --kill-others-on-fail -n api,worker \"bun --inspect run index.ts\" \"bun run worker.ts\"", "prod:deploy": "DIR=apps/dumili/api pnpm -F '~ci' prod:docker-compose-up", "prod-build-docker-api": "REPO_NAME=ghcr.io/bperel/dumili-api pnpm -F '~ci' prod:build-docker -f apps/dumili/api/Dockerfile", + "prod-build-docker-worker": "REPO_NAME=ghcr.io/bperel/dumili-worker pnpm -F '~ci' prod:build-docker -f apps/dumili/api/Dockerfile.worker", "prod-build-docker-kumiko": "REPO_NAME=ghcr.io/bperel/kumiko pnpm -F '~ci' prod:build-docker -f apps/dumili/api/kumiko/Dockerfile", "prod-build-docker-paddleocr": "REPO_NAME=ghcr.io/bperel/paddleocr pnpm -F '~ci' prod:build-docker -f apps/dumili/api/paddleocr/Dockerfile", "prod:build-docker": "pnpm run '/^prod-build-docker-/'", @@ -29,18 +32,22 @@ "@prisma/client": "^7.8.0", "@sentry/node": "^10.62.0", "@sentry/profiling-node": "^10.62.0", + "@socket.io/redis-adapter": "^8.3.0", + "@socket.io/redis-emitter": "^5.1.0", "axios": "^1.18.1", "axios-cache-interceptor": "^1.12.0", + "bullmq": "^5.79.2", "busboy": "^1.6.0", "cloudinary": "^1.41.3", "dotenv": "^17.4.2", "fflate": "^0.8.3", + "ioredis": "^5.11.1", "jsonwebtoken": "^9.0.3", "node-unrar-js": "^2.0.2", "pdfjs-dist": "^6.1.200", "sharp": "^0.33.5", "socket-call-client": "^0.7.7", - "socket-call-server": "^0.7.7", + "socket-call-server": "^0.8.0", "socket.io": "^4.8.3", "unpdf": "^1.6.2", "~api": "workspace:*", diff --git a/apps/dumili/api/pnpm-lock.yaml b/apps/dumili/api/pnpm-lock.yaml index 6d6101e37..e332c452f 100644 --- a/apps/dumili/api/pnpm-lock.yaml +++ b/apps/dumili/api/pnpm-lock.yaml @@ -23,12 +23,21 @@ importers: '@sentry/profiling-node': specifier: ^10.62.0 version: 10.62.0(@opentelemetry/core@2.8.0(@opentelemetry/api@1.9.1)) + '@socket.io/redis-adapter': + specifier: ^8.3.0 + version: 8.3.0(socket.io-adapter@2.5.8) + '@socket.io/redis-emitter': + specifier: ^5.1.0 + version: 5.1.0 axios: specifier: ^1.18.1 version: 1.18.1 axios-cache-interceptor: specifier: ^1.12.0 version: 1.12.0(axios@1.18.1) + bullmq: + specifier: ^5.79.2 + version: 5.79.2 busboy: specifier: ^1.6.0 version: 1.6.0 @@ -41,6 +50,9 @@ importers: fflate: specifier: ^0.8.3 version: 0.8.3 + ioredis: + specifier: ^5.11.1 + version: 5.11.1 jsonwebtoken: specifier: ^9.0.3 version: 9.0.3 @@ -57,8 +69,8 @@ importers: specifier: ^0.7.7 version: 0.7.7(axios-cache-interceptor@1.12.0(axios@1.18.1))(typescript@6.0.3)(vue@3.5.22(typescript@6.0.3)) socket-call-server: - specifier: ^0.7.7 - version: 0.7.7 + specifier: ^0.8.0 + version: 0.8.0 socket.io: specifier: ^4.8.3 version: 4.8.3 @@ -336,12 +348,48 @@ packages: cpu: [x64] os: [win32] + '@ioredis/commands@1.10.0': + resolution: {integrity: sha512-UmeW7z4LfctwoQ5wkhVzgq8tXkreED2xZGpX+Bg+zA+WJFZCT6c062AfCK/Dfk81xZnnwdhJCUMkitihRaoC2Q==} + + '@ioredis/commands@1.5.1': + resolution: {integrity: sha512-JH8ZL/ywcJyR9MmJ5BNqZllXNZQqQbnVZOqpPQqE1vHiFgAw4NHbvE0FOduNU8IX9babitBT46571OnPTT0Zcw==} + '@jridgewell/sourcemap-codec@1.5.5': resolution: {integrity: sha512-cYQ9310grqxueWbl+WuIUIaiUaDcj7WOq5fVhEljNVgRfOUhY9fy2zTvfoqWsnebh8Sl70VScFbICvJnLKB0Og==} '@kurkle/color@0.3.4': resolution: {integrity: sha512-M5UknZPHRu3DEDWoipU6sE8PdkZ6Z/S+v4dD+Ke8IaNlpdSQah50lz1KtcFBa2vsdOnwbbnxJwVM4wty6udA5w==} + '@msgpackr-extract/msgpackr-extract-darwin-arm64@3.0.4': + resolution: {integrity: sha512-LCkGo6JDfaBhgST7UpPWgNgLINpcpabaHfyz5OBx75nUYxBsaEPxjnyNjWpeb/xBup/682QnBfRBy2/LvPutZQ==} + cpu: [arm64] + os: [darwin] + + '@msgpackr-extract/msgpackr-extract-darwin-x64@3.0.4': + resolution: {integrity: sha512-zExlW9zUJKZH/tOtVMttwjKa4Xm/3KcNjnE3dPN92uCktwavMxpgCA3MoJK/DOnTWsQgo224OaST27/mPNAf+w==} + cpu: [x64] + os: [darwin] + + '@msgpackr-extract/msgpackr-extract-linux-arm64@3.0.4': + resolution: {integrity: sha512-dgX0P/9wGPJeHFBG+ZmhgE6bmtMt7NP5CRBGyyktpopdk/mW4POnrpQsSLtKI1dwpc+pPLuXHDh6vvskyQE/sw==} + cpu: [arm64] + os: [linux] + + '@msgpackr-extract/msgpackr-extract-linux-arm@3.0.4': + resolution: {integrity: sha512-Tg3yX65f5GbtXLkrYEHE5oibZG9epyYWas7FogTTEJeDEF9JlXJzKgXaNhT3UXlTOeA+AfZpYZYZ0uPj7Cfquw==} + cpu: [arm] + os: [linux] + + '@msgpackr-extract/msgpackr-extract-linux-x64@3.0.4': + resolution: {integrity: sha512-8TNXMEjJc3QEy7R/x1INhgiU+XakDAFUzBhaz7+Rbrs8NH5UQeHQxxmzsSBJGyV6I1jW79undiQm8tOI+D+8FQ==} + cpu: [x64] + os: [linux] + + '@msgpackr-extract/msgpackr-extract-win32-x64@3.0.4': + resolution: {integrity: sha512-CmCXPQrkbwExx3j946/PtHWHbYJiCRBRDl4BlkRQcJB/YOwQxJRTpoo7aTsortjgoJ1x7opzTSxn7C+ASSLVjQ==} + cpu: [x64] + os: [win32] + '@napi-rs/canvas-android-arm64@1.0.1': resolution: {integrity: sha512-d7ZCwJsgH4QNG50C7HQeVRsRG1gRDa1UeDUb1jEcqgLuiEJp6GVbGiZkFXPlmt0dEs2QHRQCPJoOv+bOkSQR/w==} engines: {node: '>= 10'} @@ -709,6 +757,15 @@ packages: '@socket.io/component-emitter@3.1.2': resolution: {integrity: sha512-9BCxFwvbGg/RsZK9tjXd8s4UcwR0MWeFQ1XEKIQVVvAGJyINdrqKMcTRyLoK8Rse1GjzLV9cwjWV1olXRWEXVA==} + '@socket.io/redis-adapter@8.3.0': + resolution: {integrity: sha512-ly0cra+48hDmChxmIpnESKrc94LjRL80TEmZVscuQ/WWkRP81nNj8W8cCGMqbI4L6NCuAaPRSzZF1a9GlAxxnA==} + engines: {node: '>=10.0.0'} + peerDependencies: + socket.io-adapter: ^2.5.4 + + '@socket.io/redis-emitter@5.1.0': + resolution: {integrity: sha512-QQUFPBq6JX7JIuM/X1811ymKlAfwufnQ8w6G2/59Jaqp09hdF1GJ/+e8eo/XdcmT0TqkvcSa2TT98ggTXa5QYw==} + '@standard-schema/spec@1.1.0': resolution: {integrity: sha512-l2aFy5jALhniG5HgqrD6jXLi/rUWrKvqN/qJx6yoJsgKhblVd+iqqU4RCXavm/jPityDo5TCvKMnpjKnOriy0w==} @@ -926,6 +983,15 @@ packages: buffer-equal-constant-time@1.0.1: resolution: {integrity: sha512-zRpUiDwd/xk6ADqPMATG8vc9VPrkck7T07OIx0gnjmJAnHnTVXNQG3vfvWNuiZIkwu9KrKdA1iJKfsfTVxE6NA==} + bullmq@5.79.2: + resolution: {integrity: sha512-FebD+8XCZl/hnS1R4to24L4EAN70XSndKZO0776M36vGRk5MKVhKlNFM8/34zXLXKyYB4QaeIPFhVXSYYGTHpQ==} + engines: {node: '>=12.22.0'} + peerDependencies: + redis: '>=5.0.0' + peerDependenciesMeta: + redis: + optional: true + busboy@1.6.0: resolution: {integrity: sha512-8SFQbg/0hQ9xy3UNTB0YEnsNBbWfhf7RtnzpL7TkBiTBRfrQ9Fxcnz7VJsleJpyp6rVLvXiuORqjlHi5q+PYuA==} engines: {node: '>=10.16.0'} @@ -980,6 +1046,14 @@ packages: resolution: {integrity: sha512-4o84y+E7dbif3lMns+p3UW6w6hLHEifbX/7zBJvaih1E9QNMZITENQ14GPYJC4JmhygYXsuuBb9bRA3xWEoOfg==} engines: {node: '>=0.6'} + cluster-key-slot@1.1.1: + resolution: {integrity: sha512-rwHwUfXL40Chm1r08yrhU3qpUvdVlgkKNeyeGPOxnW8/SyVDvgRaed/Uz54AqWNaTCAThlj6QAs3TZcKI0xDEw==} + engines: {node: '>=0.10.0'} + + cluster-key-slot@1.1.2: + resolution: {integrity: sha512-RMr0FhtfXemyinomL4hrWcYJxmX6deFdCxpJzhDttxgO1+bcCnkk+9drydLVDmAMG7NE6aN/fl4F7ucU/90gAA==} + engines: {node: '>=0.10.0'} + color-convert@2.0.1: resolution: {integrity: sha512-RRECPsj7iu/xb5oKYcsFHSppFNnsj/52OVTRKb4zP5onXwVF3zVmmToNcOfGC+CRDpfK/U584fMg38ZHCaElKQ==} engines: {node: '>=7.0.0'} @@ -1020,6 +1094,10 @@ packages: resolution: {integrity: sha512-tJtZBBHA6vjIAaF6EnIaq6laBBP9aq/Y3ouVJjEfoHbRBcHBAHYcMh/w8LDrk2PvIMMq8gmopa5D4V8RmbrxGw==} engines: {node: '>= 0.10'} + cron-parser@4.9.0: + resolution: {integrity: sha512-p0SaNjrHOnQeR8/VnfGbmg9te2kfyYSQ7Sc/j/6DtPL3JQvKxmjO9TSjNFpujqV3vEYYBvNNvXSxzyksBWAx1Q==} + engines: {node: '>=12.0.0'} + cross-spawn@7.0.6: resolution: {integrity: sha512-uV2QOWP2nWzsy2aMp8aRibhi9dlzF5Hgh5SHaB9OiTGEyDTiJJyx0uy51QXdyWbtAHNua4XJzUKca3OzKUd3vA==} engines: {node: '>= 8'} @@ -1365,6 +1443,14 @@ packages: resolution: {integrity: sha512-JmXMZ6wuvDmLiHEml9ykzqO6lwFbof0GG4IkcGaENdCRDDmMVnny7s5HsIgHCbaq0w2MyPhDqkhTUgS2LU2PHA==} engines: {node: '>=0.8.19'} + ioredis@5.10.1: + resolution: {integrity: sha512-HuEDBTI70aYdx1v6U97SbNx9F1+svQKBDo30o0b9fw055LMepzpOOd0Ccg9Q6tbqmBSJaMuY0fB7yw9/vjBYCA==} + engines: {node: '>=12.22.0'} + + ioredis@5.11.1: + resolution: {integrity: sha512-ehuGcf94bQXhfagULNXrJdfnWO38v070jxSx/qE87Kjzmu2fU7ro5EFAb+OPituLqgfyuQaym5DlrNydW2sJ9A==} + engines: {node: '>=12.22.0'} + is-arrayish@0.3.4: resolution: {integrity: sha512-m6UrgzFVUYawGBh1dUsWR5M2Clqic9RVXC/9f8ceNlv2IcO9j9J/z8UoCLPqtsPBFNzEpfR3xftohbfqDx8EQA==} @@ -1427,9 +1513,15 @@ packages: resolution: {integrity: sha512-iPZK6eYjbxRu3uB4/WZ3EsEIMJFMqAoopl3R+zuq0UjcAm/MO6KCweDgPfP3elTztoKP3KtnVHxTn2NHBSDVUw==} engines: {node: '>=10'} + lodash.defaults@4.2.0: + resolution: {integrity: sha512-qjxPLHd3r5DnsdGacqOMU6pb/avJzdh9tFX2ymgoZE27BmjXrNy/y4LoaiTeAb+O3gL8AfpJGtqfX/ae2leYYQ==} + lodash.includes@4.3.0: resolution: {integrity: sha512-W3Bx6mdkRTGtlJISOvVD/lbqjTlPPUDTMnlXZFnVwi9NKJ6tiAk6LVdlhZMm17VZisqhKcgzpO5Wz91PCt5b0w==} + lodash.isarguments@3.1.0: + resolution: {integrity: sha512-chi4NHZlZqZD18a0imDHnZPrDeBbTtVN7GXMwuGdRH9qotxAjYs3aVLKc7zNOG9eddR5Ksd8rvFEBc9SsggPpg==} + lodash.isboolean@3.0.3: resolution: {integrity: sha512-Bz5mupy2SVbPHURB98VAcw+aHh4vRV5IPNhILUCsOzRmsTmSQ17jIuqopAentWoehktxGd9e/hbIXq980/1QJg==} @@ -1458,6 +1550,10 @@ packages: resolution: {integrity: sha512-DqC6n3QQ77zdFpCMASA1a3Jlb64Hv2N2DciFGkO/4L9+q/IpIAuRlKOvCXabtRW6cQf8usbmM6BE/TOPysCdIA==} engines: {bun: '>=1.0.0', deno: '>=1.30.0', node: '>=8.0.0'} + luxon@3.7.2: + resolution: {integrity: sha512-vtEhXh/gNjI9Yg1u4jX/0YVPMvxzHuGgCm6tC5kZyb08yjGWGnqAjGJvcXbqQR2P3MyMEFnRbpcdFS6PBcLqew==} + engines: {node: '>=12'} + magic-string@0.30.21: resolution: {integrity: sha512-vd2F4YUyEXKGcLHoq+TEyCjxueSeHnFxyyjNp80yg0XV4vUhnDer/lvvlqM/arB5bXQN5K2/3oinyCRyx8T2CQ==} @@ -1490,6 +1586,13 @@ packages: ms@2.1.3: resolution: {integrity: sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA==} + msgpackr-extract@3.0.4: + resolution: {integrity: sha512-4kmO/MdyUIkLIvTPr8VHLil4AtoKIoniWPIEk5+CDy0xnWC84azhSFmuJ7PxZdsYtiP5kEeQsORAVIeMgxT+Hw==} + hasBin: true + + msgpackr@2.0.4: + resolution: {integrity: sha512-o1C5KRmuRt+apqMr1HuGSqWStZoRBUpEsCsl15uM9VdAF1qHLtvMOU2En747EnTyEl6c4pzPewRMFF31s1CNbA==} + mysql2@3.15.3: resolution: {integrity: sha512-FBrGau0IXmuqg4haEZRBfHNWB5mUARw6hNwPDXXGg0XzVJ50mr/9hb267lvpVMnhZ1FON3qNd4Xfcez1rbFwSg==} engines: {node: '>= 8.0'} @@ -1514,10 +1617,20 @@ packages: resolution: {integrity: sha512-KdHvFWZjEKDf0cakgFjebl371GPsISX2oZHcuyKqM7DtogIsHrqKeLTo8wBHxaXRAQlY2PsPlZmfo+9ZCxEREQ==} engines: {node: '>=10'} + node-abort-controller@3.1.1: + resolution: {integrity: sha512-AGK2yQKIjRuqnc6VkX2Xj5d+QW8xZ87pa1UK6yA6ouUyuxfHuMP6umE5QK7UmTeOAymo+Zx1Fxiuw9rVx8taHQ==} + + node-gyp-build-optional-packages@5.2.2: + resolution: {integrity: sha512-s+w+rBWnpTMwSFbaE0UXsRlg7hU4FjekKU4eyAih5T8nJuNZT1nNsskXpxmeqSK9UzkBl6UgRlnKc8hz8IEqOw==} + hasBin: true + node-unrar-js@2.0.2: resolution: {integrity: sha512-hLNmoJzqaKJnod8yiTVGe9hnlNRHotUi0CreSv/8HtfRi/3JnRC8DvsmKfeGGguRjTEulhZK6zXX5PXoVuDZ2w==} engines: {node: '>=10.0.0'} + notepack.io@3.0.1: + resolution: {integrity: sha512-TKC/8zH5pXIAMVQio2TvVDTtPRX+DJPHDqjRbxogtFiByHyzKmy96RA0JtCQJ+WouyyL4A10xomQzgbUT+1jCg==} + object-assign@4.1.1: resolution: {integrity: sha512-rJgTQnkUnH1sFw8yT6VSU3zD3sWmu6sZhIseY8VX+GRu3P6F7Fu+JNDoXfklElbLJSnc3FUQHVe4cU5hj+BcUg==} engines: {node: '>=0.10.0'} @@ -1645,6 +1758,14 @@ packages: resolution: {integrity: sha512-9u/XQ1pvrQtYyMpZe7DXKv2p5CNvyVwzUB6uhLAnQwHMSgKMBR62lc7AHljaeteeHXn11XTAaLLUVZYVZyuRBQ==} engines: {node: '>= 20.19.0'} + redis-errors@1.2.0: + resolution: {integrity: sha512-1qny3OExCf0UvUV/5wpYKf2YwPcOqXzkwKKSmKHiE6ZMQs5heeE/c8eXK+PNllPvmjgAbfnsbpkGZWy8cBpn9w==} + engines: {node: '>=4'} + + redis-parser@3.0.0: + resolution: {integrity: sha512-DJnGAeenTdpMEH6uAJRK/uiyEIH9WVsUmoLwzudwGJUwZPp80PDBWPHXSAGNPwNvIXAbe7MSUB1zQFugFml66A==} + engines: {node: '>=4'} + remeda@2.33.4: resolution: {integrity: sha512-ygHswjlc/opg2VrtiYvUOPLjxjtdKvjGz1/plDhkG66hjNjFr1xmfrs2ClNFo/E6TyUFiwYNh53bKV26oBoMGQ==} @@ -1736,9 +1857,9 @@ packages: axios-cache-interceptor: optional: true - socket-call-server@0.7.7: - resolution: {integrity: sha512-znkJ4/guRtgQhn4+jjJn3yZCBUni5453CunWc7ON/qkSLKws+rPT7SU+JaOdEXJFf/wqd8iJUHPurKIX02KZ8A==} - engines: {node: '>=22.0.0 <23.0.0'} + socket-call-server@0.8.0: + resolution: {integrity: sha512-tfrJkmKpU09kVwg1AGkWWiSoCp9IlBxivIscPKldBuqsN19zJI2tIohr2TJNTV5Tz4nrj2Ql2KO23PYz3BWuDA==} + engines: {node: '>=22.0.0'} socket.io-adapter@2.5.8: resolution: {integrity: sha512-6Oy52pbg+kvdCVvjcN+FnY7BvxZ7cIHNScbvztT/It5d0vbwoJoVZmF2gjJmnV0/4WlXRfG15zc45ySk9Ah8bw==} @@ -1767,6 +1888,9 @@ packages: resolution: {integrity: sha512-qC9iz2FlN7DQl3+wjwn3802RTyjCx7sDvfQEXchwa6CWOx07/WVfh91gBmQ9fahw8snwGEWU3xGzOt4tFyHLxg==} engines: {node: '>= 0.6'} + standard-as-callback@2.1.0: + resolution: {integrity: sha512-qoRRSyROncaz1z0mvYqIE4lCd9p2R90i6GxW3uZv5ucSu8tU7B5HXUP1gG8pVZsYNVaXjk8ClXHPttLyxAL48A==} + std-env@3.10.0: resolution: {integrity: sha512-5GS12FdOZNliM5mAOxFRg7Ir0pWz8MdpYm6AY6VPkGpbA7ZzmbzNcBJQ0GPvvyWgcY7QAhCgf9Uy89I03faLkg==} @@ -1830,6 +1954,10 @@ packages: engines: {node: '>=14.17'} hasBin: true + uid2@1.0.0: + resolution: {integrity: sha512-+I6aJUv63YAcY9n4mQreLUt0d4lvwkkopDNmpomkAUz0fAkEMV9pRWxN0EjhW1YfRhcuyHg2v3mwddCDW1+LFQ==} + engines: {node: '>= 4.0.0'} + undici-types@7.24.6: resolution: {integrity: sha512-WRNW+sJgj5OBN4/0JpHFqtqzhpbnV0GuB+OozA9gCL7a993SmU+1JBZCzLNxYsbMfIeDL+lTsphD5jN5N+n0zg==} @@ -2113,10 +2241,32 @@ snapshots: '@img/sharp-win32-x64@0.33.5': optional: true + '@ioredis/commands@1.10.0': {} + + '@ioredis/commands@1.5.1': {} + '@jridgewell/sourcemap-codec@1.5.5': {} '@kurkle/color@0.3.4': {} + '@msgpackr-extract/msgpackr-extract-darwin-arm64@3.0.4': + optional: true + + '@msgpackr-extract/msgpackr-extract-darwin-x64@3.0.4': + optional: true + + '@msgpackr-extract/msgpackr-extract-linux-arm64@3.0.4': + optional: true + + '@msgpackr-extract/msgpackr-extract-linux-arm@3.0.4': + optional: true + + '@msgpackr-extract/msgpackr-extract-linux-x64@3.0.4': + optional: true + + '@msgpackr-extract/msgpackr-extract-win32-x64@3.0.4': + optional: true + '@napi-rs/canvas-android-arm64@1.0.1': optional: true @@ -2480,6 +2630,23 @@ snapshots: '@socket.io/component-emitter@3.1.2': {} + '@socket.io/redis-adapter@8.3.0(socket.io-adapter@2.5.8)': + dependencies: + debug: 4.3.7 + notepack.io: 3.0.1 + socket.io-adapter: 2.5.8 + uid2: 1.0.0 + transitivePeerDependencies: + - supports-color + + '@socket.io/redis-emitter@5.1.0': + dependencies: + debug: 4.3.7 + notepack.io: 3.0.1 + socket.io-parser: 4.2.6 + transitivePeerDependencies: + - supports-color + '@standard-schema/spec@1.1.0': {} '@types/bcryptjs@2.4.6': {} @@ -2753,6 +2920,17 @@ snapshots: buffer-equal-constant-time@1.0.1: {} + bullmq@5.79.2: + dependencies: + cron-parser: 4.9.0 + ioredis: 5.10.1 + msgpackr: 2.0.4 + node-abort-controller: 3.1.1 + semver: 7.8.5 + tslib: 2.8.1 + transitivePeerDependencies: + - supports-color + busboy@1.6.0: dependencies: streamsearch: 1.1.0 @@ -2812,6 +2990,10 @@ snapshots: lodash: 4.18.1 q: 1.5.1 + cluster-key-slot@1.1.1: {} + + cluster-key-slot@1.1.2: {} + color-convert@2.0.1: dependencies: color-name: 1.1.4 @@ -2854,6 +3036,10 @@ snapshots: object-assign: 4.1.1 vary: 1.1.2 + cron-parser@4.9.0: + dependencies: + luxon: 3.7.2 + cross-spawn@7.0.6: dependencies: path-key: 3.1.1 @@ -3192,6 +3378,32 @@ snapshots: imurmurhash@0.1.4: {} + ioredis@5.10.1: + dependencies: + '@ioredis/commands': 1.5.1 + cluster-key-slot: 1.1.2 + debug: 4.4.3 + denque: 2.1.0 + lodash.defaults: 4.2.0 + lodash.isarguments: 3.1.0 + redis-errors: 1.2.0 + redis-parser: 3.0.0 + standard-as-callback: 2.1.0 + transitivePeerDependencies: + - supports-color + + ioredis@5.11.1: + dependencies: + '@ioredis/commands': 1.10.0 + cluster-key-slot: 1.1.1 + debug: 4.4.3 + denque: 2.1.0 + redis-errors: 1.2.0 + redis-parser: 3.0.0 + standard-as-callback: 2.1.0 + transitivePeerDependencies: + - supports-color + is-arrayish@0.3.4: {} is-core-module@2.16.2: @@ -3259,8 +3471,12 @@ snapshots: dependencies: p-locate: 5.0.0 + lodash.defaults@4.2.0: {} + lodash.includes@4.3.0: {} + lodash.isarguments@3.1.0: {} + lodash.isboolean@3.0.3: {} lodash.isinteger@4.0.4: {} @@ -3279,6 +3495,8 @@ snapshots: lru.min@1.1.4: {} + luxon@3.7.2: {} + magic-string@0.30.21: dependencies: '@jridgewell/sourcemap-codec': 1.5.5 @@ -3305,6 +3523,22 @@ snapshots: ms@2.1.3: {} + msgpackr-extract@3.0.4: + dependencies: + node-gyp-build-optional-packages: 5.2.2 + optionalDependencies: + '@msgpackr-extract/msgpackr-extract-darwin-arm64': 3.0.4 + '@msgpackr-extract/msgpackr-extract-darwin-x64': 3.0.4 + '@msgpackr-extract/msgpackr-extract-linux-arm': 3.0.4 + '@msgpackr-extract/msgpackr-extract-linux-arm64': 3.0.4 + '@msgpackr-extract/msgpackr-extract-linux-x64': 3.0.4 + '@msgpackr-extract/msgpackr-extract-win32-x64': 3.0.4 + optional: true + + msgpackr@2.0.4: + optionalDependencies: + msgpackr-extract: 3.0.4 + mysql2@3.15.3: dependencies: aws-ssl-profiles: 1.1.2 @@ -3331,8 +3565,17 @@ snapshots: dependencies: semver: 7.8.5 + node-abort-controller@3.1.1: {} + + node-gyp-build-optional-packages@5.2.2: + dependencies: + detect-libc: 2.1.2 + optional: true + node-unrar-js@2.0.2: {} + notepack.io@3.0.1: {} + object-assign@4.1.1: {} object-code@2.0.0: {} @@ -3443,6 +3686,12 @@ snapshots: readdirp@5.0.0: {} + redis-errors@1.2.0: {} + + redis-parser@3.0.0: + dependencies: + redis-errors: 1.2.0 + remeda@2.33.4: {} require-from-string@2.0.2: {} @@ -3545,7 +3794,7 @@ snapshots: - supports-color - utf-8-validate - socket-call-server@0.7.7: + socket-call-server@0.8.0: dependencies: socket.io: 4.8.3 transitivePeerDependencies: @@ -3600,6 +3849,8 @@ snapshots: sqlstring@2.3.3: {} + standard-as-callback@2.1.0: {} + std-env@3.10.0: {} streamsearch@1.1.0: {} @@ -3652,6 +3903,8 @@ snapshots: typescript@6.0.3: {} + uid2@1.0.0: {} + undici-types@7.24.6: {} undici@6.27.0: {} diff --git a/apps/dumili/api/queue/connection.ts b/apps/dumili/api/queue/connection.ts new file mode 100644 index 000000000..96ba9456e --- /dev/null +++ b/apps/dumili/api/queue/connection.ts @@ -0,0 +1,16 @@ +import type { ConnectionOptions } from "bullmq"; +import { Redis, type RedisOptions } from "ioredis"; + +export const REDIS_URL = process.env.REDIS_URL || "redis://localhost:6379"; +const redisUrl = new URL(REDIS_URL); + +export const createRedisClient = (options?: RedisOptions) => + options ? new Redis(REDIS_URL, options) : new Redis(REDIS_URL); + +export const bullConnection: ConnectionOptions = (() => ({ + host: redisUrl.hostname, + port: redisUrl.port ? Number(redisUrl.port) : 6379, + username: redisUrl.username || undefined, + password: redisUrl.password || undefined, + maxRetriesPerRequest: null, +}))(); diff --git a/apps/dumili/api/queue/indexation-ai.queue.ts b/apps/dumili/api/queue/indexation-ai.queue.ts new file mode 100644 index 000000000..e84a5eacd --- /dev/null +++ b/apps/dumili/api/queue/indexation-ai.queue.ts @@ -0,0 +1,50 @@ +import { Queue } from "bullmq"; + +import { bullConnection, createRedisClient } from "./connection"; + +export const INDEXATION_AI_QUEUE = "indexation-ai"; + +export type IndexationAiJobData = { + indexationId: string; + userId: number; +}; + +const jobOptions = { + jobId: undefined, + removeOnComplete: true, + removeOnFail: true, +} as const; + +const dirtyKey = (indexationId: string) => `indexation-ai:dirty:${indexationId}`; + +const redis = createRedisClient(); + +export const indexationAiQueue = new Queue( + INDEXATION_AI_QUEUE, + { connection: bullConnection }, +); + +const addJob = (indexationId: string, userId: number) => + indexationAiQueue.add( + "process", + { indexationId, userId }, + { ...jobOptions, jobId: indexationId }, + ); + +export const enqueueIndexationAi = async ( + indexationId: string, + userId: number, +) => { + await redis.set(dirtyKey(indexationId), "1"); + await addJob(indexationId, userId); +}; + +export const markProcessingStarted = (indexationId: string) => + redis.getdel(dirtyKey(indexationId)); + +export const requeueIfDirty = async (indexationId: string, userId: number) => { + const dirty = await redis.getdel(dirtyKey(indexationId)); + if (dirty) { + await addJob(indexationId, userId); + } +}; diff --git a/apps/dumili/api/services/indexation/ai-pipeline/index.ts b/apps/dumili/api/services/indexation/ai-pipeline/index.ts new file mode 100644 index 000000000..9d491288e --- /dev/null +++ b/apps/dumili/api/services/indexation/ai-pipeline/index.ts @@ -0,0 +1,13 @@ +import type { IndexationAiContext } from "../context"; +import { runKumikoOnPages } from "../kumiko"; +import { setInferredEntriesStoryKinds } from "./story-kinds"; +import { createAiStorySuggestions } from "./story-suggestions"; + +export const runIndexationAi = async (ctx: IndexationAiContext) => { + await runKumikoOnPages(ctx); + await setInferredEntriesStoryKinds(ctx); + await createAiStorySuggestions(ctx); +}; + +export { setInferredEntriesStoryKinds } from "./story-kinds"; +export { createAiStorySuggestions } from "./story-suggestions"; diff --git a/apps/dumili/api/services/indexation/ai-pipeline/story-kinds.ts b/apps/dumili/api/services/indexation/ai-pipeline/story-kinds.ts new file mode 100644 index 000000000..f52200e50 --- /dev/null +++ b/apps/dumili/api/services/indexation/ai-pipeline/story-kinds.ts @@ -0,0 +1,109 @@ +import { getEntryPages } from "~dumili-utils/entryPages"; +import prisma from "~prisma/client"; +import type { aiKumikoResult } from "~prisma/client_dumili/client"; + +import type { IndexationAiContext } from "../context"; + +export const setInferredEntriesStoryKinds = async ( + { indexation, events }: IndexationAiContext +) => { + for (const entry of indexation.entries) { + if ( + entry.storyKindSuggestions.some( + ({ aiKumikoResultId }) => aiKumikoResultId, + ) + ) { + console.log( + `Entry starting at page ${entry.position}: already has an inferred story kind`, + ); + continue; + } + + events.reportSetInferredEntryStoryKind(entry.id); + const pagesInferredStoryKinds = await prisma.image.findMany({ + include: { + aiKumikoResult: true, + }, + where: { + id: { + in: getEntryPages(indexation, entry.id) + .filter(({ imageId }) => !!imageId) + .map(({ imageId }) => imageId!), + }, + }, + }); + + if (!pagesInferredStoryKinds.length) { + console.log( + `Entry starting at page ${entry.position}: No pages with inferred story kinds found`, + ); + } else { + const mostInferredStoryKind = Object.entries( + ( + pagesInferredStoryKinds.filter( + ({ aiKumikoResult }) => aiKumikoResult !== null, + ) as { aiKumikoResult: aiKumikoResult }[] + ) + .map(({ aiKumikoResult: { id, inferredStoryKindRowsStr } }) => ({ + id, + inferredStoryKindRowsStr, + })) + .groupBy("inferredStoryKindRowsStr", "id[]"), + ).sort((a, b) => b[1].length - a[1].length)[0][0]; + + const entryIdx = indexation.entries.findIndex( + ({ id }) => id === entry.id, + ); + console.log( + `Kumiko: entry #${entryIdx}: inferred story kind and number of rows are ${mostInferredStoryKind}`, + ); + + if (mostInferredStoryKind) { + const suggestion = indexation.entries[ + entryIdx + ].storyKindSuggestions.find( + ({ storyKindRowsStr }) => storyKindRowsStr === mostInferredStoryKind, + ); + if (suggestion) { + await prisma.entry.update({ + data: { + storyKindSuggestions: { + update: { + data: { + aiKumikoResultId: pagesInferredStoryKinds.find( + ({ aiKumikoResult }) => + aiKumikoResult?.inferredStoryKindRowsStr === + mostInferredStoryKind, + )?.aiKumikoResultId, + }, + where: { + id: suggestion.id, + }, + }, + }, + }, + where: { + id: entry.id, + }, + }); + if (!entry.acceptedStoryKindSuggestionId) { + await prisma.entry.update({ + data: { + acceptedStoryKindSuggestionId: suggestion.id, + }, + where: { + id: entry.id, + }, + }); + } + } else { + console.warn( + `Entry starting at page ${entry.position}: no inferred story kind and number of rows found`, + ); + } + } + } + + events.reportSetInferredEntryStoryKindEnd(entry.id); + } +}; diff --git a/apps/dumili/api/services/indexation/ai-pipeline/story-suggestions.ts b/apps/dumili/api/services/indexation/ai-pipeline/story-suggestions.ts new file mode 100644 index 000000000..8219d8ddf --- /dev/null +++ b/apps/dumili/api/services/indexation/ai-pipeline/story-suggestions.ts @@ -0,0 +1,265 @@ +import { COVER, STORY } from "~dumili-types/storyKinds"; +import { getEntryPages } from "~dumili-utils/entryPages"; +import prisma from "~prisma/client"; + +import type { + FullEntry, + FullIndexation, + IndexationAiContext, +} from "../context"; +import { runOcrOnImage } from "../ocr"; +import { + getFullStoriesFromKeywords, + getPublicationLanguagecode, + getStoriesFromImage, +} from "../story-search"; + +type EntryImage = NonNullable; + +// The two strategies are tried in order; the first one that produces matches wins. +const STRATEGIES = [ + { + name: "image-based story search", + field: "aiStorySearchResult", + storyField: "aiStorySearchPossibleStory", + }, + { + name: "OCR-based story search", + field: "aiOcrResult", + storyField: "aiOcrPossibleStory", + }, +] as const; + +type StorySearchStrategy = typeof STRATEGIES[number] + +const runStorySearch = async ( + indexationEvents: IndexationAiContext['events'], + entry: FullEntry, + image: EntryImage, + strategy: StorySearchStrategy, + isCover: boolean, + languagecode: string, +) => + strategy.field === "aiStorySearchResult" + ? getStoriesFromImage(image, isCover) + : getFullStoriesFromKeywords( + ( + await runOcrOnImage(indexationEvents, entry.position, image, languagecode) + ).map(({ text }) => text), + ); + +const persistStorySuggestions = async ({ + entry, + image, + strategy, + results, + currentlyAcceptedStorycode, +}: { + entry: FullEntry; + image: EntryImage; + strategy: StorySearchStrategy; + results: { stories: { storycode: string; score: number }[] }; + currentlyAcceptedStorycode: string | undefined; +}) => { + const { name, field, storyField } = strategy; + + let aiResultId = ( + await prisma.image.findUnique({ where: { id: image.id } }) + )?.[`${field}Id`]; + + if (aiResultId) { + if (field === "aiOcrResult") { + await prisma.storySuggestion.deleteMany({ + where: { + aiStorySuggestion: { aiOcrPossibleStory: { resultId: aiResultId } }, + }, + }); + await prisma.aiOcrPossibleStory.deleteMany({ + where: { resultId: aiResultId }, + }); + } else { + await prisma.storySuggestion.deleteMany({ + where: { + aiStorySuggestion: { + aiStorySearchPossibleStory: { resultId: aiResultId }, + }, + }, + }); + await prisma.aiStorySearchPossibleStory.deleteMany({ + where: { resultId: aiResultId }, + }); + } + } else { + aiResultId = + field === "aiOcrResult" + ? (await prisma.aiOcrResult.create({ data: {} })).id + : (await prisma.aiStorySearchResult.create({ data: {} })).id; + } + + await prisma.image.update({ + where: { id: image.id }, + data: { [`${field}Id`]: aiResultId }, + }); + + if (!results.stories.length) { + console.info( + `Entry starting at page ${entry.position}: No ${name} results found`, + ); + return false; + } + + console.log( + `Entry starting at page ${entry.position}: ${results.stories.length} ${name} matches found`, + ); + await prisma.storySuggestion.deleteMany({ + where: { + aiStorySuggestion: { + [storyField]: { + // aiOcrPossibleStory or aiStorySearchPossibleStory + isNot: null, + }, + }, + entryId: entry.id, + }, + }); + + const storiesWithScores = results.stories.groupBy("storycode", "score[]"); + + for (const storycode of Object.keys(storiesWithScores)) { + console.log("Creating story suggestion for storycode", storycode); + const data = { + aiStorySuggestion: { + create: { + [storyField]: { + // aiOcrPossibleStory or aiStorySearchPossibleStory + create: { + [field]: { + // aiOcrResult or aiStorySearchResult + connect: { + id: aiResultId, + }, + }, + score: storiesWithScores[storycode].sort((a, b) => b - a)[0], + }, + }, + }, + }, + storycode, + entry: { + connect: { + id: entry.id, + }, + }, + }; + await prisma.storySuggestion.upsert({ + where: { + entryId_storycode: { + entryId: entry.id, + storycode, + }, + }, + create: data, + update: data, + }); + } + + const newEntry = (await prisma.entry.findUnique({ + include: { + storySuggestions: { + include: { + aiStorySuggestion: { + include: { + [storyField]: true, + }, + }, + }, + }, + }, + where: { + id: entry.id, + }, + }))!; + const acceptedStorySuggestionId = newEntry.storySuggestions.find( + ({ storycode }) => storycode === currentlyAcceptedStorycode, + )?.id; + // If no story is currently accepted, we accept the first story suggestion + if (acceptedStorySuggestionId) { + await prisma.entry.update({ + where: { + id: entry.id, + }, + data: { + acceptedStory: { + connect: { + id: acceptedStorySuggestionId, + }, + }, + }, + }); + } + return true; +}; + +const createEntryStorySuggestions = async ( + {indexation, events}: IndexationAiContext, + entry: FullEntry, + languagecode: string, +) => { + const currentlyAcceptedStorycode = entry.acceptedStory?.storycode; + const firstPageOfEntry = getEntryPages(indexation, entry.id)[0]; + if (!firstPageOfEntry.image) { + return; + } + const image = firstPageOfEntry.image; + const isCover = entry.acceptedStoryKind?.storyKindRows?.kind === COVER; + + events.reportCreateAiStorySuggestions(entry.id); + + for (const strategy of STRATEGIES) { + const results = await runStorySearch( + events, + entry, + image, + strategy, + isCover, + languagecode, + ); + if ("error" in results) { + console.error(results.error); + continue; + } + const created = await persistStorySuggestions({ + entry, + image, + strategy, + results, + currentlyAcceptedStorycode, + }); + if (created) { + break; + } + } + + events.reportCreateAiStorySuggestionsEnd(entry.id); +}; + +export const createAiStorySuggestions = async (ctx: IndexationAiContext) => { + const { indexation } = ctx; + const languagecode = indexation.acceptedIssueSuggestion?.publicationcode + ? await getPublicationLanguagecode( + indexation.acceptedIssueSuggestion.publicationcode, + ) + : "en"; + + for (const entry of indexation.entries) { + if ( + [STORY, COVER].includes(entry.acceptedStoryKind?.storyKindRows?.kind ?? "") + ) { + await createEntryStorySuggestions(ctx, entry, languagecode); + } else { + console.log( + `Entry starting at page ${entry.position}: This entry is not a story or a cover`, + ); + } + } +}; diff --git a/apps/dumili/api/services/indexation/context.ts b/apps/dumili/api/services/indexation/context.ts new file mode 100644 index 000000000..0fbca738f --- /dev/null +++ b/apps/dumili/api/services/indexation/context.ts @@ -0,0 +1,145 @@ +import type { Socket } from "socket.io"; +import type { NamespaceProxyTarget } from "socket-call-server"; +import { type ServerSentStartEndEvents } from "socket-call-server"; + +import prisma from "~prisma/client"; +import type { Prisma } from "~prisma/client_dumili/client"; + +import type { SessionDataWithIndexation } from "../../index"; + +export const indexationPayloadInclude = { + user: true, + pages: { + orderBy: { + pageNumber: "asc", + }, + include: { + image: { + include: { + aiKumikoResult: { + include: { + detectedPanels: true, + inferredStoryKindRows: true, + }, + }, + aiOcrResult: { + include: { + matches: true, + stories: { + include: { + aiStorySuggestion: true, + }, + }, + }, + }, + aiStorySearchResult: { + include: { + stories: { + include: { + aiStorySuggestion: true, + }, + }, + }, + }, + }, + }, + }, + }, + acceptedIssueSuggestion: true, + issueSuggestions: true, + entries: { + include: { + acceptedStory: true, + acceptedStoryKind: { include: { storyKindRows: true } }, + storyKindSuggestions: { include: { storyKindRows: true } }, + storySuggestions: { + include: { + aiStorySuggestion: { + include: { + aiStorySearchPossibleStory: true, + }, + }, + }, + }, + }, + }, +} as const; + +export type FullIndexation = Prisma.indexationGetPayload<{ + include: typeof indexationPayloadInclude; +}>; + +export type FullEntry = FullIndexation["entries"][number]; + +export type IndexationServerSentStartEvents = { + reportSetKumikoInferredPageStoryKinds: (pageId: number) => void; + reportSetInferredEntryStoryKind: (entryId: number) => void; + reportCreateAiStorySuggestions: (entryId: number) => void; + reportRunOcrOnImage: (imageId: number) => void; + reportRunStorySearchOnImage: (imageId: number) => void; + reportDocumentAnalyzed: (pageNumbers: number[]) => void; + reportDocumentPageUploaded: (pageNumber: number) => void; +}; + +export type IndexationServerSentStartEndEvents = + ServerSentStartEndEvents & { + indexationUpdated: (indexation: FullIndexation) => void; + }; + +// The "start" events whose handler receives a single numeric id (i.e. every one +// except reportDocumentAnalyzed, which receives a number[]). +export type IndexationNumberIdEvent = { + [K in keyof IndexationServerSentStartEvents]: IndexationServerSentStartEvents[K] extends ( + id: number, + ) => void + ? K + : never; +}[keyof IndexationServerSentStartEvents]; + +export type IndexationSocket = Socket< + object, + IndexationServerSentStartEndEvents, + object, + SessionDataWithIndexation +>; + +export type IndexationServices = NamespaceProxyTarget< + IndexationSocket, + IndexationServerSentStartEndEvents +>; + +export type IndexationEvents = IndexationServerSentStartEndEvents; + +export type UserId = SessionDataWithIndexation["user"]["id"] + +export type IndexationAiContext = { + events: IndexationEvents; + userId: UserId; + indexation: FullIndexation; +}; + +export const fetchFullIndexation = async ( + userId: UserId, + id: string, +) => { + const indexation = await prisma.indexation.findUnique({ + where: { id, dmUserId: userId }, + include: indexationPayloadInclude, + }); + if (indexation) { + indexation.entries = indexation.entries.sort( + (a, b) => a.position - b.position, + ); + } + return indexation; +}; + +export const refreshIndexation = async ( + events: IndexationEvents, + userId: UserId, + indexationId: string, +) => { + const indexation = (await fetchFullIndexation(userId, indexationId))!; + events.indexationUpdated(indexation); + return indexation; +}; diff --git a/apps/dumili/api/services/indexation/index.ts b/apps/dumili/api/services/indexation/index.ts index 7c8cf0304..6ae8cf777 100644 --- a/apps/dumili/api/services/indexation/index.ts +++ b/apps/dumili/api/services/indexation/index.ts @@ -1,18 +1,9 @@ -import "~group-by"; - import { v2 as cloudinary } from "cloudinary"; -import type { Server, Socket } from "socket.io"; -import type { NamespaceProxyTarget } from "socket-call-server"; -import { - type ServerSentStartEndEvents, - useSocketEvents, -} from "socket-call-server"; +import type { Server } from "socket.io"; +import { getServerSentEvents, useSocketEvents } from "socket-call-server"; -import { COVER, STORY } from "~dumili-types/storyKinds"; -import { getEntryPages } from "~dumili-utils/entryPages"; import prisma from "~prisma/client"; import type { - aiKumikoResult, entry, indexation, issueSuggestion, @@ -21,30 +12,30 @@ import type { storyKindSuggestion, storySuggestion, } from "~prisma/client_dumili/client"; -import type { ClientEvents as CoaEvents } from "~dm-services/coa"; -import dmNamespaces from "~dm-services/namespaces"; import type { SessionDataWithIndexation } from "../../index"; +import { enqueueIndexationAi } from "../../queue/indexation-ai.queue"; import { RequiredAuthMiddleware } from "../_auth"; import namespaces from "../namespaces"; -import { runKumikoOnPages } from "./kumiko"; -import { runOcrOnImage } from "./ocr"; import { - getStoriesFromImage, - getFullStoriesFromKeywords, -} from "./story-search"; -import { SocketClient } from "socket-call-client"; + fetchFullIndexation, + type FullIndexation, + type IndexationEvents, + type IndexationServerSentStartEndEvents, + type IndexationServices, + refreshIndexation, +} from "./context"; import { readFileSync, existsSync } from "fs"; import { fileURLToPath } from "url"; import { dirname, join } from "path"; -import { definePDFJSModule, getDocumentProxy, renderPageAsImage } from 'unpdf' +import { definePDFJSModule, getDocumentProxy, renderPageAsImage } from "unpdf"; import { createExtractorFromData } from "node-unrar-js"; import { unzipSync } from "fflate"; -await definePDFJSModule(() => import('pdfjs-dist')) +await definePDFJSModule(() => import("pdfjs-dist")); // In the production bundle, bun bakes in the CI build path for unrar.wasm. // We override it by loading the file explicitly when it's present next to the bundle. @@ -54,10 +45,7 @@ const unrarWasmBinary = existsSync(bundleSideWasm) ? (() => { const buf = readFileSync(bundleSideWasm); return buf.buffer.slice(buf.byteOffset, buf.byteOffset + buf.byteLength); })() : undefined; -const canvasImport = () => import('@napi-rs/canvas'); - -const socket = new SocketClient(process.env.DM_SOCKET_URL!); -const coaEvents = socket.addNamespace(dmNamespaces.COA); +const canvasImport = () => import("@napi-rs/canvas"); const MAX_DOCUMENT_FILE_SIZE = 50 * 1024 * 1024; const MAX_IMAGE_FILE_SIZE = 5 * 1024 * 1024; @@ -75,7 +63,7 @@ const ALLOWED_IMAGE_MIME_TYPES = new Set([ const ALLOWED_MIME_TYPES = new Set([...ALLOWED_DOCUMENT_MIME_TYPES, ...ALLOWED_IMAGE_MIME_TYPES]); const inferMimeType = (value?: string) => { - const extension = value?.toLowerCase()?.split('.')?.pop(); + const extension = value?.toLowerCase()?.split(".")?.pop(); if (!extension) { return undefined; } @@ -95,8 +83,10 @@ const inferMimeType = (value?: string) => { return undefined; }; -const uploadToCloudinary = async ({ services, folder, context, page, ...params }: { - services: IndexationServices; +const uploadToCloudinary = async ({ events, userId, indexation, folder, context, page, ...params }: { + events: IndexationEvents; + userId: SessionDataWithIndexation["user"]["id"]; + indexation: FullIndexation; folder: string; context: Record; page: page; @@ -123,11 +113,8 @@ const uploadToCloudinary = async ({ services, folder, context, page, ...params } else { console.log(`Uploaded page ${page.pageNumber} to Cloudinary:`, result.secure_url); - await setPageUrl( - services, - page.id, - result.secure_url, - ); + await setPageUrl(indexation, page.id, result.secure_url); + await refreshIndexation(events, userId, indexation.id); resolve(); } }, @@ -136,595 +123,74 @@ const uploadToCloudinary = async ({ services, folder, context, page, ...params } }, ); -const indexationPayloadInclude = { - user: true, - pages: { - orderBy: { - pageNumber: "asc", - }, - include: { - image: { - include: { - aiKumikoResult: { - include: { - detectedPanels: true, - inferredStoryKindRows: true, - }, - }, - aiOcrResult: { - include: { - matches: true, - stories: { - include: { - aiStorySuggestion: true, - }, - }, - }, - }, - aiStorySearchResult: { - include: { - stories: { - include: { - aiStorySuggestion: true, - }, - }, - }, - }, - }, - }, - }, - }, - acceptedIssueSuggestion: true, - issueSuggestions: true, - entries: { - include: { - acceptedStory: true, - acceptedStoryKind: { include: { storyKindRows: true } }, - storyKindSuggestions: { include: { storyKindRows: true } }, - storySuggestions: { - include: { - aiStorySuggestion: { - include: { - aiStorySearchPossibleStory: true - } - } - } - }, - }, - }, -} as const; - -const setPageUrl = async (services: IndexationServices, id: number, url: string | null) => { - if ( - !services._socket.data.indexation.pages.some( - ({ id: pageId }) => pageId === id, - ) - ) { +const setPageUrl = async ( + indexation: FullIndexation, + id: number, + url: string | null, +) => { + if (!indexation.pages.some(({ id: pageId }) => pageId === id)) { return { error: "This indexation does not have any page with this ID", }; } - return prisma.page - .update({ - data: { - image: url - ? { - connectOrCreate: { - create: { - url, - }, - where: { - url, - }, + await prisma.page.update({ + data: { + image: url + ? { + connectOrCreate: { + create: { + url, }, - } - : { disconnect: true }, - }, - where: { - id, - }, - }) - .then(async () => { - await refreshIndexation(services); - return "OK" as const; - }); -}; - -export type FullIndexation = Prisma.indexationGetPayload<{ - include: typeof indexationPayloadInclude; -}>; - -export type FullEntry = FullIndexation["entries"][number]; - -export type IndexationServerSentStartEvents = { - reportSetKumikoInferredPageStoryKinds: (pageId: number) => void; - reportSetInferredEntryStoryKind: (entryId: number) => void; - reportCreateAiStorySuggestions: (entryId: number) => void; - reportRunOcrOnImage: (imageId: number) => void; - reportRunStorySearchOnImage: (imageId: number) => void; - reportDocumentAnalyzed: (pageNumbers: number[]) => void; - reportDocumentPageUploaded: (pageNumber: number) => void; + where: { + url, + }, + }, + } + : { disconnect: true }, + }, + where: { + id, + }, + }); + return "OK" as const; }; -export type IndexationServerSentStartEndEvents = - ServerSentStartEndEvents & { - indexationUpdated: (indexation: FullIndexation) => void; - }; - -export type IndexationSocket = Socket< - object, +export type { + FullIndexation, + FullEntry, + IndexationServerSentStartEvents, IndexationServerSentStartEndEvents, - object, - SessionDataWithIndexation ->; + IndexationNumberIdEvent, +} from "./context"; -const isAiRunning: Record = {}; -export const getFullIndexation = ( - services: IndexationServices, - indexationId: string, - runAi = true, -) => - prisma.indexation - .findUnique({ - where: { id: indexationId, dmUserId: services._socket.data.user.id }, - include: indexationPayloadInclude, - }) - .then((indexation) => { - if (indexation) { - indexation.entries = indexation.entries.sort( - (a, b) => a.position - b.position, - ); - if (runAi && !(indexationId in isAiRunning)) { - isAiRunning[indexationId] = true; - runKumikoOnPages(services, indexation) - .then(() => - setInferredEntriesStoryKinds(services, indexation.entries), - ) - .then(() => createAiStorySuggestions(services, indexation)) - .finally(() => { - delete isAiRunning[indexationId]; - refreshIndexation(services, false, indexationId); - }); - } - } - return indexation; - }); +const enqueueAi = (services: IndexationServices) => + enqueueIndexationAi( + services._socket.data.indexation.id, + services._socket.data.user.id, + ); -export const refreshIndexation = async ( - services: IndexationServices, - runAi = true, - id = services._socket.data.indexation.id, -) => { - services._socket.data.indexation = (await getFullIndexation( +const refreshConnection = async (services: IndexationServices) => { + services._socket.data.indexation = await refreshIndexation( services, - id, - runAi, - ))!; - services.indexationUpdated(services._socket.data.indexation); + services._socket.data.user.id, + services._socket.data.indexation.id, + ); }; -const createAiStorySuggestions = async ( - services: IndexationServices, +const uploadPages = async ( + events: IndexationEvents, + user: SessionDataWithIndexation["user"], indexation: FullIndexation, -) => { - const languagecode = indexation.acceptedIssueSuggestion?.publicationcode - ? (await coaEvents.getPublicationLanguagecode( - indexation.acceptedIssueSuggestion.publicationcode, - )) || "en" - : "en"; - - for (const entry of indexation.entries) { - if ( - [STORY, COVER].includes( - entry.acceptedStoryKind?.storyKindRows?.kind ?? "", - ) - ) { - const currentlyAcceptedStorycode = entry.acceptedStory?.storycode; - - const firstPageOfEntry = getEntryPages(indexation, entry.id)[0]; - - if (!firstPageOfEntry.image) { - continue; - } - - services.reportCreateAiStorySuggestions(entry.id); - - for (const { name, field, storyField, _storySuggestionRelationship } of [ - { - name: "image-based story search", - field: "aiStorySearchResult", - storyField: "aiStorySearchPossibleStory", - _storySuggestionRelationship: "storySearchDetails", - }, - { - name: "OCR-based story search", - field: "aiOcrResult", - storyField: "aiOcrPossibleStory", - _storySuggestionRelationship: "ocrDetails", - }, - ] as const) { - const cachedResults: - | { - type: typeof _storySuggestionRelationship; - storycode: string; - }[] - | undefined = undefined; /*firstPageOfEntry.image[field]?.stories - .filter( - ( - story - ): story is typeof story & { - aiStorySuggestion: { storycode: string }; - } => !!story.aiStorySuggestion - ) - .map( - ({ - aiStorySuggestion: { - storySuggestion: { storycode }, - }, - }) => ({ - type: storySuggestionRelationship, - storycode, - }) - );*/ - - if (cachedResults) { - // if (cachedResults.length) { - // console.log( - // `Entry starting at page ${entry.position}: ${cachedResults.length} ${name} matches found (cached)`, - // ); - // break; - // } else { - // console.log( - // `Entry starting at page ${entry.position}: No ${name} matches found (cached)`, - // ); - // continue; - // } - } else { - const results = - field === "aiStorySearchResult" - ? await getStoriesFromImage( - firstPageOfEntry.image, - entry.acceptedStoryKind?.storyKindRows?.kind === COVER, - ) - : await getFullStoriesFromKeywords( - ( - await runOcrOnImage( - services, - entry.position, - firstPageOfEntry.image, - languagecode, - ) - ).map(({ text }) => text), - ); - if ("error" in results) { - console.error(results.error); - } else { - let aiResultId = ( - await prisma.image.findUnique({ - where: { - id: firstPageOfEntry.image.id, - }, - }) - )?.[`${field}Id`]; - - if (aiResultId) { - if (field === "aiOcrResult") { - await prisma.storySuggestion.deleteMany({ - where: { - aiStorySuggestion: { - aiOcrPossibleStory: { - resultId: aiResultId, - }, - }, - }, - }); - await prisma.aiOcrPossibleStory.deleteMany({ - where: { - resultId: aiResultId, - }, - }); - } else { - await prisma.storySuggestion.deleteMany({ - where: { - aiStorySuggestion: { - aiStorySearchPossibleStory: { - resultId: aiResultId, - }, - }, - }, - }); - await prisma.aiStorySearchPossibleStory.deleteMany({ - where: { - resultId: aiResultId, - }, - }); - } - } else { - if (field === "aiOcrResult") { - aiResultId = ( - await prisma.aiOcrResult.create({ - data: {}, - }) - ).id; - } else { - aiResultId = ( - await prisma.aiStorySearchResult.create({ - data: {}, - }) - ).id; - } - } - - await prisma.image.update({ - where: { - id: firstPageOfEntry.image.id, - }, - data: { - [`${field}Id`]: aiResultId, - }, - }); - if (!results.stories.length) { - console.info( - `Entry starting at page ${entry.position}: No ${name} results found`, - ); - } else { - console.log( - `Entry starting at page ${entry.position}: ${results.stories.length} ${name} matches found`, - ); - await prisma.storySuggestion.deleteMany({ - where: { - aiStorySuggestion: { - [storyField]: { - // aiOcrPossibleStory or aiStorySearchPossibleStory - isNot: null, - }, - }, - entryId: entry.id, - }, - }); - - const storiesWithScores = results.stories.groupBy( - "storycode", - "score[]", - ); - - for (const storycode of Object.keys(storiesWithScores)) { - console.log( - "Creating story suggestion for storycode", - storycode, - ); - const data = { - aiStorySuggestion: { - create: { - [storyField]: { - // aiOcrPossibleStory or aiStorySearchPossibleStory - create: { - [field]: { - // aiOcrResult or aiStorySearchResult - connect: { - id: aiResultId, - }, - }, - score: storiesWithScores[storycode].sort( - (a, b) => b - a, - )[0], - }, - }, - }, - }, - storycode, - entry: { - connect: { - id: entry.id, - }, - }, - }; - await prisma.storySuggestion.upsert({ - where: { - entryId_storycode: { - entryId: entry.id, - storycode, - }, - }, - create: data, - update: data, - }); - } - - const newEntry = (await prisma.entry.findUnique({ - include: { - storySuggestions: { - include: { - aiStorySuggestion: { - include: { - [storyField]: true, // ocrDetails or storySearchDetails - }, - }, - }, - }, - }, - where: { - id: entry.id, - }, - }))!; - const acceptedStorySuggestionId = newEntry.storySuggestions.find( - ({ storycode }) => storycode === currentlyAcceptedStorycode, - )?.id; - // If no story is currently accepted, we accept the first story suggestion - if (acceptedStorySuggestionId) { - await prisma.entry.update({ - where: { - id: entry.id, - }, - data: { - acceptedStory: { - connect: { - id: acceptedStorySuggestionId, - }, - }, - }, - }); - } - break; - } - } - } - } - - services.reportCreateAiStorySuggestionsEnd(entry.id); - } else { - console.log( - `Entry starting at page ${entry.position}: This entry is not a story or a cover`, - ); - } - } -}; - -const setInferredEntriesStoryKinds = async ( - services: IndexationServices, - entries: FullIndexation["entries"], - force?: boolean, -) => { - for (const entry of entries) { - if ( - entry.storyKindSuggestions.some( - ({ aiKumikoResultId }) => aiKumikoResultId, - ) && - !force - ) { - console.log( - `Entry starting at page ${entry.position}: already has an inferred story kind`, - ); - continue; - } - - services.reportSetInferredEntryStoryKind(entry.id); - const { indexation } = services._socket.data; - const pagesInferredStoryKinds = await prisma.image.findMany({ - include: { - aiKumikoResult: true, - }, - where: { - id: { - in: getEntryPages(indexation, entry.id) - .filter(({ imageId }) => !!imageId) - .map(({ imageId }) => imageId!), - }, - }, - }); - - if (!pagesInferredStoryKinds.length) { - console.log( - `Entry starting at page ${entry.position}: No pages with inferred story kinds found`, - ); - } else { - const mostInferredStoryKind = Object.entries( - ( - pagesInferredStoryKinds.filter( - ({ aiKumikoResult }) => aiKumikoResult !== null, - ) as { aiKumikoResult: aiKumikoResult }[] - ) - .map(({ aiKumikoResult: { id, inferredStoryKindRowsStr } }) => ({ - id, - inferredStoryKindRowsStr, - })) - .groupBy("inferredStoryKindRowsStr", "id[]"), - ).sort((a, b) => b[1].length - a[1].length)[0][0]; - - const entryIdx = services._socket.data.indexation.entries.findIndex( - ({ id }) => id === entry.id, - ); - console.log( - `Kumiko: entry #${entryIdx}: inferred story kind and number of rows are ${mostInferredStoryKind}`, - ); - - if (mostInferredStoryKind) { - const suggestion = indexation.entries[ - entryIdx - ].storyKindSuggestions.find( - ({ storyKindRowsStr }) => storyKindRowsStr === mostInferredStoryKind, - ); - if (suggestion) { - await prisma.entry.update({ - data: { - storyKindSuggestions: { - update: { - data: { - aiKumikoResultId: pagesInferredStoryKinds.find( - ({ aiKumikoResult }) => - aiKumikoResult?.inferredStoryKindRowsStr === - mostInferredStoryKind, - )?.aiKumikoResultId, - }, - where: { - id: suggestion.id, - }, - }, - }, - }, - where: { - id: entry.id, - }, - }); - if (!entry.acceptedStoryKindSuggestionId) { - await prisma.entry.update({ - data: { - acceptedStoryKindSuggestionId: suggestion.id, - }, - where: { - id: entry.id, - }, - }); - } - } else { - console.warn( - `Entry starting at page ${entry.position}: no inferred story kind and number of rows found`, - ); - } - } - } - - services.reportSetInferredEntryStoryKindEnd(entry.id); - } -}; - -export type IndexationServices = NamespaceProxyTarget< - Socket< - typeof listenEvents, - IndexationServerSentStartEndEvents, - object, - SessionDataWithIndexation - >, - IndexationServerSentStartEndEvents ->; - -const makeServicesProxy = (io: Server, indexationId: string, user: SessionDataWithIndexation["user"], initialIndexation: FullIndexation): IndexationServices => { - const data = { user, indexation: initialIndexation } as const; - const ns = `/indexation/${indexationId}`; - const socket = { - data, - nsp: { name: ns }, - emit: (event: string, ...args: unknown[]) => io.of(ns).emit(event, ...args), - }; - return new Proxy({} as IndexationServices, { - get(_, prop: string) { - if (prop === '_socket') return socket; - return (...args: unknown[]) => socket.emit(prop, ...args); - } - }); -}; - -const uploadPages = async (services: IndexationServices, { - buffer, fileName, mimeType, firstPageNumber, firstOutOfRangePageNumber, -}: { - buffer: Buffer; - fileName: string; - mimeType: string; - firstPageNumber: number; - firstOutOfRangePageNumber: number; -}) => { + { + buffer, fileName, mimeType, firstPageNumber, firstOutOfRangePageNumber, + }: { + buffer: Buffer; + fileName: string; + mimeType: string; + firstPageNumber: number; + firstOutOfRangePageNumber: number; + }) => { try { - const { indexation, user } = services._socket.data; const effectiveMimeType = mimeType || inferMimeType(fileName); if (!effectiveMimeType || !ALLOWED_MIME_TYPES.has(effectiveMimeType)) { return { error: "Unsupported file type" }; @@ -754,12 +220,12 @@ const uploadPages = async (services: IndexationServices, { throw new Error("No valid pages found in the document"); } const pagesToOverwrite = pagesToPotentiallyOverwrite.slice(0, items.length); - services.reportDocumentAnalyzed(pagesToOverwrite.map(({ pageNumber }) => pageNumber)); + events.reportDocumentAnalyzed(pagesToOverwrite.map(({ pageNumber }) => pageNumber)); for (const [idx, page] of pagesToOverwrite.entries()) { const { name, getData } = items[idx]; console.info(`Uploading page ${page.pageNumber} from file ${name}...`); - await uploadToCloudinary({ services, buffer: await getData(), page, folder, context }); - services.reportDocumentPageUploaded(page.pageNumber); + await uploadToCloudinary({ events, userId: user.id, indexation, buffer: await getData(), page, folder, context }); + events.reportDocumentPageUploaded(page.pageNumber); } }; @@ -770,7 +236,7 @@ const uploadPages = async (services: IndexationServices, { const files = [...extracted.files] .filter(f => f.extraction?.length && - f.fileHeader.name.split('/').length === 1 && + f.fileHeader.name.split("/").length === 1 && inferMimeType(f.fileHeader.name) && ALLOWED_IMAGE_MIME_TYPES.has(inferMimeType(f.fileHeader.name)!) ) @@ -782,7 +248,7 @@ const uploadPages = async (services: IndexationServices, { case "application/zip": { const files = Object.entries(unzipSync(new Uint8Array(buffer))) .filter(([name]) => - name.split('/').length === 1 && + name.split("/").length === 1 && inferMimeType(name) && ALLOWED_IMAGE_MIME_TYPES.has(inferMimeType(name)!) ) @@ -807,8 +273,8 @@ const uploadPages = async (services: IndexationServices, { } } else { const page = indexation.pages.find(({ pageNumber }) => pageNumber === firstPageNumber)!; - await uploadToCloudinary({ services, buffer, page, folder, context }); - services.reportDocumentPageUploaded(page.pageNumber); + await uploadToCloudinary({ events, userId: user.id, indexation, buffer, page, folder, context }); + events.reportDocumentPageUploaded(page.pageNumber); } return { status: "OK" }; @@ -834,18 +300,27 @@ export const handleHttpFileUpload = async ( firstOutOfRangePageNumber: number; }, ) => { - const indexation = await prisma.indexation.findUnique({ - where: { id: indexationId, dmUserId: user.id }, - include: indexationPayloadInclude, - }); + const indexation = await fetchFullIndexation(user.id, indexationId); if (!indexation) return { error: "Indexation not found" }; - indexation.entries = indexation.entries.sort((a, b) => a.position - b.position); - const services = makeServicesProxy(io, indexationId, user, indexation); - return uploadPages(services, { buffer, fileName, mimeType, firstPageNumber, firstOutOfRangePageNumber }); + const events = getServerSentEvents( + io.of(`/indexation/${indexationId}`), + ); + const result = await uploadPages(events, user, indexation, { buffer, fileName, mimeType, firstPageNumber, firstOutOfRangePageNumber }); + if (!("error" in result)) { + await enqueueIndexationAi(indexationId, user.id); + } + return result; }; const listenEvents = (services: IndexationServices) => ({ - setPageUrl: async (id: number, url: string | null) => setPageUrl(services, id, url), + setPageUrl: async (id: number, url: string | null) => { + const result = await setPageUrl(services._socket.data.indexation, id, url); + if (result === "OK") { + await refreshConnection(services); + await enqueueAi(services); + } + return result; + }, deleteIndexation: async () => { const { id: indexationId } = services._socket.data.indexation; @@ -880,7 +355,8 @@ const listenEvents = (services: IndexationServices) => ({ }, }); - await refreshIndexation(services); + await refreshConnection(services); + await enqueueAi(services); return { status: "OK" }; }, @@ -949,7 +425,8 @@ const listenEvents = (services: IndexationServices) => ({ }), ) .then(async () => { - await refreshIndexation(services); + await refreshConnection(services); + await enqueueAi(services); return { status: "OK" as const, }; @@ -983,7 +460,8 @@ const listenEvents = (services: IndexationServices) => ({ }, }) .then(async () => { - await refreshIndexation(services); + await refreshConnection(services); + await enqueueAi(services); return { status: "OK", }; @@ -998,7 +476,7 @@ const listenEvents = (services: IndexationServices) => ({ data: suggestion, }) .then(async (createdStorySuggestion) => { - await refreshIndexation(services); + await refreshConnection(services); return { createdStorySuggestion, }; @@ -1043,7 +521,8 @@ const listenEvents = (services: IndexationServices) => ({ }), ) .then(async () => { - await refreshIndexation(services); + await refreshConnection(services); + await enqueueAi(services); return createdIssueSuggestion; }), ), @@ -1102,7 +581,8 @@ const listenEvents = (services: IndexationServices) => ({ id: services._socket.data.indexation.id, }, }) - .then(() => refreshIndexation(services)) + .then(() => refreshConnection(services)) + .then(() => enqueueAi(services)) .then(() => ({ status: "OK", })); @@ -1133,7 +613,7 @@ const listenEvents = (services: IndexationServices) => ({ }, }); - await refreshIndexation(services); + await refreshConnection(services); return { status: "OK" }; }, @@ -1167,7 +647,8 @@ const listenEvents = (services: IndexationServices) => ({ }, }); - await refreshIndexation(services); + await refreshConnection(services); + await enqueueAi(services); return { status: "OK" }; }, @@ -1200,14 +681,16 @@ const listenEvents = (services: IndexationServices) => ({ }, }); - await refreshIndexation(services); + await refreshConnection(services); + await enqueueAi(services); return { status: "OK" }; }, createEntry: async (position: number) => createEntry(services._socket.data.indexation.id, position) - .then(() => refreshIndexation(services)) + .then(() => refreshConnection(services)) + .then(() => enqueueAi(services)) .then(() => ({ status: "OK" })), }); @@ -1227,7 +710,15 @@ export const { client, server } = useSocketEvents< return; } - await refreshIndexation(services, true, indexationId); + const { user } = services._socket.data; + const indexation = await fetchFullIndexation(user.id, indexationId); + if (!indexation) { + next(new Error("Indexation not found")); + return; + } + services._socket.data.indexation = indexation; + services.indexationUpdated(indexation); + await enqueueIndexationAi(indexationId, user.id); next(); }, ], diff --git a/apps/dumili/api/services/indexation/kumiko.ts b/apps/dumili/api/services/indexation/kumiko.ts index 1f19e0173..2d10bb0b3 100644 --- a/apps/dumili/api/services/indexation/kumiko.ts +++ b/apps/dumili/api/services/indexation/kumiko.ts @@ -5,8 +5,7 @@ import { getEntryFromPage } from "~dumili-utils/entryPages"; import prisma from "~prisma/client"; import type { aiKumikoResultPanel, Prisma } from "~prisma/client_dumili/client"; -import type { IndexationServices } from "."; -import { type FullIndexation, refreshIndexation } from "."; +import { type IndexationAiContext, refreshIndexation } from "./context"; type KumikoResult = { filename: string; @@ -29,7 +28,7 @@ const inferStoryKindFromAiResults = ( const getPanelRows = ( panels: Awaited>[number], -): number => { +) => { if (!panels.length) { return 0; } @@ -53,7 +52,7 @@ const getPanelRows = ( }; const runKumikoOnPage = async ( - indexationServices: IndexationServices, + ctx: IndexationAiContext, page: Prisma.pageGetPayload<{ include: { image: { include: { aiKumikoResult: true } } }; }>, @@ -64,7 +63,7 @@ const runKumikoOnPage = async ( } else if (page.image.aiKumikoResult && !force) { console.info(`Kumiko: page ${page.pageNumber}: already inferred`); } else { - indexationServices.reportSetKumikoInferredPageStoryKinds(page.id); + ctx.events.reportSetKumikoInferredPageStoryKinds(page.id); const panelsPerPage = await runKumiko([page.image.url]); const panelsOfPage = panelsPerPage[0] || []; console.info( @@ -127,29 +126,32 @@ const runKumikoOnPage = async ( }, }); - await refreshIndexation(indexationServices); + ctx.indexation = await refreshIndexation( + ctx.events, + ctx.userId, + ctx.indexation.id, + ); - indexationServices.reportSetKumikoInferredPageStoryKindsEnd(page.id); + ctx.events.reportSetKumikoInferredPageStoryKindsEnd(page.id); return true; } return false; }; export const runKumikoOnPages = async ( - services: IndexationServices, - indexation: FullIndexation, + ctx: IndexationAiContext, force = false, ) => { const updatedImageIds = []; - for (const page of indexation.pages) { - if (await runKumikoOnPage(services, page, force)) { + for (const page of ctx.indexation.pages) { + if (await runKumikoOnPage(ctx, page, force)) { updatedImageIds.push(page.id); } } const outdatedEntryIds = new Set( updatedImageIds - .map((id) => getEntryFromPage(indexation, id)?.id) + .map((id) => getEntryFromPage(ctx.indexation, id)?.id) .filter((id) => !!id) .map((id) => id!), ); @@ -159,9 +161,7 @@ export const runKumikoOnPages = async ( } }; -export const runKumiko = async ( - urls: (string | null)[], -): Promise => +export const runKumiko = async (urls: (string | null)[]) => axios .get( `${process.env.KUMIKO_HOST}?i=${urls.filter((url) => !!url).join(",")}`, diff --git a/apps/dumili/api/services/indexation/ocr.ts b/apps/dumili/api/services/indexation/ocr.ts index 3b818e9c9..54ddeefe4 100644 --- a/apps/dumili/api/services/indexation/ocr.ts +++ b/apps/dumili/api/services/indexation/ocr.ts @@ -3,7 +3,7 @@ import axios from "axios"; import prisma from "~prisma/client"; import type { aiKumikoResultPanel } from "~prisma/client_dumili/client"; -import { type FullIndexation, type IndexationServices } from "."; +import { type FullIndexation, type IndexationAiContext } from "./context"; type OcrResult = { box: [number /* x1 */, number /* y1 */, number /* x2 */, number /* y2 */]; @@ -12,7 +12,7 @@ type OcrResult = { }; export const runOcrOnImage = async ( - services: IndexationServices, + indexationEvents: IndexationAiContext['events'], pageNumber: number, image: NonNullable, languagecode: string, @@ -26,7 +26,7 @@ export const runOcrOnImage = async ( console.log(`Page ${pageNumber}: This page already has OCR results`); return image.aiOcrResult.matches; } - services.reportRunOcrOnImage(image.id); + indexationEvents.reportRunOcrOnImage(image.id); const firstPanelUrl = image.url.replace( "/pg_", `/c_crop,h_${firstPanel.height},w_${firstPanel.width},x_${firstPanel.x},y_${firstPanel.y},pg_`, @@ -69,7 +69,7 @@ export const runOcrOnImage = async ( }, }); - services.reportRunOcrOnImageEnd(image.id); + indexationEvents.reportRunOcrOnImageEnd(image.id); return matches; }; /* Adding a bit of extra in case the storycode is just outside the panel */ diff --git a/apps/dumili/api/services/indexation/story-search.ts b/apps/dumili/api/services/indexation/story-search.ts index 5fd8d8945..7cae007ef 100644 --- a/apps/dumili/api/services/indexation/story-search.ts +++ b/apps/dumili/api/services/indexation/story-search.ts @@ -16,6 +16,11 @@ const storySearchEvents = storySearchSocket.addNamespace( dmNamespaces.STORY_SEARCH, ); +export const getPublicationLanguagecode = (publicationcode: string) => + coaEvents + .getPublicationLanguagecode(publicationcode) + .then((languagecode) => languagecode || "en"); + export const getFullStoriesFromKeywords = async (keywords: string[]) => { const response = await coaEvents.getFullStoriesFromKeywords(keywords); if ("error" in response) { diff --git a/apps/dumili/api/tsconfig.json b/apps/dumili/api/tsconfig.json index e58f03b7b..67935b449 100644 --- a/apps/dumili/api/tsconfig.json +++ b/apps/dumili/api/tsconfig.json @@ -53,9 +53,11 @@ "eslint.config.mjs", "index.ts", "instrument.ts", + "worker.ts", "prisma.config.ts", "prisma/client_dumili/**/*.ts", "prisma/client.ts", + "queue/**/*.ts", "scripts/**/*.mjs", "services/**/*.ts", ] diff --git a/apps/dumili/api/worker.ts b/apps/dumili/api/worker.ts new file mode 100644 index 000000000..97c2d2c56 --- /dev/null +++ b/apps/dumili/api/worker.ts @@ -0,0 +1,71 @@ +import dotenv from "dotenv"; +dotenv.config({ + path: ".env", +}); +dotenv.config({ + path: ".env.local", + override: true, +}); + +import "~group-by"; + +import { Emitter } from "@socket.io/redis-emitter"; +import { Worker } from "bullmq"; +import { getServerSentEvents } from "socket-call-server"; + +import { bullConnection, createRedisClient } from "./queue/connection"; +import { + INDEXATION_AI_QUEUE, + type IndexationAiJobData, + markProcessingStarted, + requeueIfDirty, +} from "./queue/indexation-ai.queue"; +import { runIndexationAi } from "./services/indexation/ai-pipeline"; +import { + fetchFullIndexation, + type IndexationServerSentStartEndEvents, +} from "./services/indexation/context"; + +const concurrency = Number(process.env.INDEXATION_AI_CONCURRENCY) || 2; + +// Publishes to the same Redis channels the API server's adapter subscribes to, +// so events reach clients connected to the API process. +const emitter = new Emitter(createRedisClient()); + +const worker = new Worker( + INDEXATION_AI_QUEUE, + async (job) => { + const { indexationId, userId } = job.data; + await markProcessingStarted(indexationId); + + const indexation = await fetchFullIndexation(userId, indexationId); + if (!indexation) { + console.warn( + `Indexation ${indexationId} not found (user ${userId}); skipping AI run`, + ); + return; + } + + const events = getServerSentEvents( + emitter.of(`/indexation/${indexationId}`), + ); + + console.log(`Running AI pipeline for indexation ${indexationId}`); + await runIndexationAi({ events, userId, indexation }); + }, + { connection: bullConnection, concurrency }, +); + +// Coalescing: if a request arrived while this run was in flight, run once more +// against the now current DB state (the job id is free after removal). +worker.on("completed", (job) => { + requeueIfDirty(job.data.indexationId, job.data.userId).catch((error) => + console.error("Failed to re-enqueue indexation AI job", error), + ); +}); + +worker.on("failed", (job, error) => { + console.error(`Indexation AI job failed (${job?.data.indexationId})`, error); +}); + +console.log(`Dumili indexation AI worker started (concurrency ${concurrency})`); diff --git a/apps/dumili/src/components/AiTooltip.vue b/apps/dumili/src/components/AiTooltip.vue index 04edd0318..208fae691 100644 --- a/apps/dumili/src/components/AiTooltip.vue +++ b/apps/dumili/src/components/AiTooltip.vue @@ -21,9 +21,9 @@ /> -