diff --git a/.env.example b/.env.example index e2caf20d0..01d386b8a 100755 --- a/.env.example +++ b/.env.example @@ -203,8 +203,9 @@ MIXPANEL_TOKEN= # Referee Finder ML_REFEREE_GET_PRESIGNED_URL= -ML_REFEREE_TRIGGER_CID= +ML_REFEREE_TRIGGER_START= ML_REFEREE_FINDER_RESULT= +ML_REFEREE_FILE_PREFIX= # Temp var for metering comms between nextjs api backend/nodes backend INTERNAL_SERVICE_SECRET=psssssssssst diff --git a/desci-server/kubernetes/deployment_dev.yaml b/desci-server/kubernetes/deployment_dev.yaml index 04108e122..54f87956f 100644 --- a/desci-server/kubernetes/deployment_dev.yaml +++ b/desci-server/kubernetes/deployment_dev.yaml @@ -114,8 +114,9 @@ spec: export AWS_SQS_DATA_MIGRATION_QUEUE_URL="{{ .Data.AWS_SQS_DATA_MIGRATION_QUEUE_URL }}" export AWS_SQS_ML_TOOL_QUEUE_URL="{{ .Data.AWS_SQS_ML_TOOL_QUEUE_URL }}" export ML_REFEREE_GET_PRESIGNED_URL="{{ .Data.ML_REFEREE_GET_PRESIGNED_URL }}" - export ML_REFEREE_TRIGGER_CID="{{ .Data.ML_REFEREE_TRIGGER_CID }}" + export ML_REFEREE_TRIGGER_START="{{ .Data.ML_REFEREE_TRIGGER_START }}" export ML_REFEREE_FINDER_RESULT="{{ .Data.ML_REFEREE_FINDER_RESULT }}" + export ML_REFEREE_FILE_PREFIX="{{ .Data.ML_REFEREE_FILE_PREFIX }}" export ENABLE_GUEST_MODE="{{ .Data.ENABLE_GUEST_MODE }}" export MIXPANEL_TOKEN="{{ .Data.MIXPANEL_TOKEN }}" export SCIWEAVE_GOOGLE_CLIENT_ID="{{ .Data.SCIWEAVE_GOOGLE_CLIENT_ID }}" diff --git a/desci-server/kubernetes/deployment_prod.yaml b/desci-server/kubernetes/deployment_prod.yaml index d46c3523c..40af09f67 100755 --- a/desci-server/kubernetes/deployment_prod.yaml +++ b/desci-server/kubernetes/deployment_prod.yaml @@ -113,8 +113,9 @@ spec: export AWS_SQS_DATA_MIGRATION_QUEUE_URL="{{ .Data.AWS_SQS_DATA_MIGRATION_QUEUE_URL }}" export AWS_SQS_ML_TOOL_QUEUE_URL="{{ .Data.AWS_SQS_ML_TOOL_QUEUE_URL }}" export ML_REFEREE_GET_PRESIGNED_URL="{{ .Data.ML_REFEREE_GET_PRESIGNED_URL }}" - export ML_REFEREE_TRIGGER_CID="{{ .Data.ML_REFEREE_TRIGGER_CID }}" + export ML_REFEREE_TRIGGER_START="{{ .Data.ML_REFEREE_TRIGGER_START }}" export ML_REFEREE_FINDER_RESULT="{{ .Data.ML_REFEREE_FINDER_RESULT }}" + export ML_REFEREE_FILE_PREFIX="{{ .Data.ML_REFEREE_FILE_PREFIX }}" export ENABLE_GUEST_MODE="{{ .Data.ENABLE_GUEST_MODE }}" export MIXPANEL_TOKEN="{{ .Data.MIXPANEL_TOKEN }}" export SCIWEAVE_GOOGLE_CLIENT_ID="{{ .Data.SCIWEAVE_GOOGLE_CLIENT_ID }}" diff --git a/desci-server/kubernetes/deployment_staging.yaml b/desci-server/kubernetes/deployment_staging.yaml index b87db2684..117a98001 100644 --- a/desci-server/kubernetes/deployment_staging.yaml +++ b/desci-server/kubernetes/deployment_staging.yaml @@ -120,8 +120,9 @@ spec: export AWS_SQS_DATA_MIGRATION_QUEUE_URL="{{ .Data.AWS_SQS_DATA_MIGRATION_QUEUE_URL }}" export AWS_SQS_ML_TOOL_QUEUE_URL="{{ .Data.AWS_SQS_ML_TOOL_QUEUE_URL }}" export ML_REFEREE_GET_PRESIGNED_URL="{{ .Data.ML_REFEREE_GET_PRESIGNED_URL }}" - export ML_REFEREE_TRIGGER_CID="{{ .Data.ML_REFEREE_TRIGGER_CID }}" + export ML_REFEREE_TRIGGER_START="{{ .Data.ML_REFEREE_TRIGGER_START }}" export ML_REFEREE_FINDER_RESULT="{{ .Data.ML_REFEREE_FINDER_RESULT }}" + export ML_REFEREE_FILE_PREFIX="{{ .Data.ML_REFEREE_FILE_PREFIX }}" export ENABLE_GUEST_MODE="{{ .Data.ENABLE_GUEST_MODE }}" export MIXPANEL_TOKEN="{{ .Data.MIXPANEL_TOKEN }}" export SCIWEAVE_GOOGLE_CLIENT_ID="{{ .Data.SCIWEAVE_GOOGLE_CLIENT_ID }}" diff --git a/desci-server/prisma/migrations/20250818192731_add_querying_data_external_api_table/migration.sql b/desci-server/prisma/migrations/20250818192731_add_querying_data_external_api_table/migration.sql new file mode 100644 index 000000000..702ea6be7 --- /dev/null +++ b/desci-server/prisma/migrations/20250818192731_add_querying_data_external_api_table/migration.sql @@ -0,0 +1,2 @@ +-- AlterTable +ALTER TABLE "ExternalApiUsage" ADD COLUMN "queryingData" JSONB; diff --git a/desci-server/prisma/migrations/20250818192759_add_gin_index_external_api_querying_data/migration.sql b/desci-server/prisma/migrations/20250818192759_add_gin_index_external_api_querying_data/migration.sql new file mode 100644 index 000000000..9ecece804 --- /dev/null +++ b/desci-server/prisma/migrations/20250818192759_add_gin_index_external_api_querying_data/migration.sql @@ -0,0 +1,6 @@ +-- CreateIndex: Add GIN index for ExternalApiUsage.queryingData field +-- This allows efficient JSON containment queries on the queryingData field + +CREATE INDEX CONCURRENTLY IF NOT EXISTS "idx_external_api_usage_querying_data_gin" +ON "ExternalApiUsage" USING GIN ("queryingData"); + diff --git a/desci-server/prisma/migrations/20250818194118_gin_index_ext_api_querying_data/migration.sql b/desci-server/prisma/migrations/20250818194118_gin_index_ext_api_querying_data/migration.sql new file mode 100644 index 000000000..2c8d514aa --- /dev/null +++ b/desci-server/prisma/migrations/20250818194118_gin_index_ext_api_querying_data/migration.sql @@ -0,0 +1,2 @@ +-- DropIndex +DROP INDEX "idx_external_api_usage_querying_data_gin"; diff --git a/desci-server/prisma/schema.prisma b/desci-server/prisma/schema.prisma index d803afb0b..fbb690b13 100755 --- a/desci-server/prisma/schema.prisma +++ b/desci-server/prisma/schema.prisma @@ -1435,12 +1435,13 @@ model JournalFormResponse { } model ExternalApiUsage { - id Int @id @default(autoincrement()) - userId Int - apiType ExternalApi - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt - data Json? + id Int @id @default(autoincrement()) + userId Int + apiType ExternalApi + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + data Json? // Can dump additional metadata here + queryingData Json? // GIN indexed, important data thats used for querying goes here, avoid dumping random metadata. // Relationships user User @relation(fields: [userId], references: [id]) diff --git a/desci-server/src/controllers/externalApi/RefereeRecommender/getResults.ts b/desci-server/src/controllers/externalApi/RefereeRecommender/getResults.ts index 6b97d201a..2ffad8007 100644 --- a/desci-server/src/controllers/externalApi/RefereeRecommender/getResults.ts +++ b/desci-server/src/controllers/externalApi/RefereeRecommender/getResults.ts @@ -1,6 +1,8 @@ +import { ExternalApi } from '@prisma/client'; import { Response } from 'express'; import { z } from 'zod'; +import { prisma } from '../../../client.js'; import { sendSuccess, sendError } from '../../../core/api.js'; import { AuthenticatedRequest } from '../../../core/types.js'; import { logger as parentLogger } from '../../../logger.js'; @@ -29,32 +31,33 @@ export const getResults = async (req: AuthenticatedRequest, res: Response) => { 'Fetching referee recommendation results', ); - // Verify user has access to this filename by checking session - const sessionResult = await RefereeRecommenderService.getSession(UploadedFileName, user.id); - if (sessionResult.isErr()) { - logger.warn( - { - userId: user.id, - uploadedFileName: UploadedFileName, + // Check ExternalApiUsage table for completed results + const existingResult = await prisma.externalApiUsage.findFirst({ + where: { + userId: user.id, + apiType: ExternalApi.REFEREE_FINDER, + queryingData: { + path: ['fileName'], + equals: UploadedFileName, }, - 'Session not found or expired', - ); - return sendError(res, 'Results not found or access denied', 404); - } + }, + orderBy: { + createdAt: 'desc', // Get the most recent result + }, + }); - const session = sessionResult.value; - if (session.userId !== user.id) { + if (!existingResult) { logger.warn( { userId: user.id, - sessionUserId: session.userId, uploadedFileName: UploadedFileName, }, - 'User does not have access to this session', + 'No results found for this filename and user', ); - return sendError(res, 'Results not found or access denied', 404); + return sendError(res, 'Results not found', 404); } + // User has access to this filename, now fetch actual results from lambda API const result = await RefereeRecommenderService.getRefereeResults(UploadedFileName); if (result.isErr()) { @@ -63,8 +66,9 @@ export const getResults = async (req: AuthenticatedRequest, res: Response) => { error: result.error, userId: user.id, uploadedFileName: UploadedFileName, + resultId: existingResult.id, }, - 'Failed to fetch referee results', + 'Failed to fetch referee results from lambda API', ); return sendError(res, result.error.message, 500); } @@ -75,9 +79,10 @@ export const getResults = async (req: AuthenticatedRequest, res: Response) => { { userId: user.id, uploadedFileName: UploadedFileName, - status: responseData.status, + resultId: existingResult.id, + createdAt: existingResult.createdAt, }, - 'Successfully fetched referee recommendation results', + 'Successfully fetched referee recommendation results from database', ); return sendSuccess(res, responseData); diff --git a/desci-server/src/controllers/externalApi/RefereeRecommender/issuePresignedUrl.ts b/desci-server/src/controllers/externalApi/RefereeRecommender/issuePresignedUrl.ts index 3df2ab314..3e7ae8949 100644 --- a/desci-server/src/controllers/externalApi/RefereeRecommender/issuePresignedUrl.ts +++ b/desci-server/src/controllers/externalApi/RefereeRecommender/issuePresignedUrl.ts @@ -1,6 +1,8 @@ +import { ExternalApi } from '@prisma/client'; import { Response } from 'express'; import { z } from 'zod'; +import { prisma } from '../../../client.js'; import { sendSuccess, sendError } from '../../../core/api.js'; import { AuthenticatedRequest } from '../../../core/types.js'; import { logger as parentLogger } from '../../../logger.js'; @@ -10,6 +12,7 @@ const logger = parentLogger.child({ module: 'RefereeRecommender::PresignedUrlCon const generatePresignedUrlSchema = z.object({ body: z.object({ + fileHash: z.string().min(1, 'fileHash is required'), fileName: z.string().min(1, 'fileName is required'), }), }); @@ -24,17 +27,51 @@ export const generatePresignedUrl = async (req: AuthenticatedRequest, res: Respo } const { body } = generatePresignedUrlSchema.parse(req); - const { fileName } = body; + const { fileHash, fileName } = body; logger.info( { userId: user.id, + fileHash, fileName, }, - 'Generating presigned URL for referee recommender', + 'Checking cache and potentially generating presigned URL for referee recommender', ); - // Validate file extension (assuming PDFs for now) + // First, check if we have cached results for this fileHash + const cachedResult = await prisma.externalApiUsage.findFirst({ + where: { + userId: user.id, + apiType: ExternalApi.REFEREE_FINDER, + queryingData: { + path: ['fileHash'], + equals: fileHash, + }, + }, + orderBy: { + createdAt: 'desc', // Get the most recent result + }, + }); + + if (cachedResult) { + logger.info( + { + userId: user.id, + fileHash, + cachedResultId: cachedResult.id, + }, + 'Found cached results for fileHash', + ); + + return sendSuccess(res, { + cached: true, + message: 'Results for this file are already available', + resultKey: RefereeRecommenderService.prepareFormattedFileName(fileHash), + }); + } + + // No cached results found, proceed with generating presigned URL + // Validate file extension if (!fileName.toLowerCase().endsWith('.pdf')) { return sendError(res, 'Only PDF files are supported', 400); } @@ -49,7 +86,7 @@ export const generatePresignedUrl = async (req: AuthenticatedRequest, res: Respo return sendError(res, 'Failed to generate presigned URL', 500); } - const { presignedUrl, fileName: generatedFileName } = result.value; + const { presignedUrl, downloadUrl, fileName: generatedFileName } = result.value; logger.info( { @@ -60,9 +97,11 @@ export const generatePresignedUrl = async (req: AuthenticatedRequest, res: Respo ); return sendSuccess(res, { + cached: false, presignedUrl, + downloadUrl, fileName: generatedFileName, - expiresIn: 3600, // 1 hour + fileHash, }); } catch (error) { if (error instanceof z.ZodError) { diff --git a/desci-server/src/controllers/externalApi/RefereeRecommender/triggerRecommendation.ts b/desci-server/src/controllers/externalApi/RefereeRecommender/triggerRecommendation.ts index c69a901a2..f5a0d96a5 100644 --- a/desci-server/src/controllers/externalApi/RefereeRecommender/triggerRecommendation.ts +++ b/desci-server/src/controllers/externalApi/RefereeRecommender/triggerRecommendation.ts @@ -1,6 +1,9 @@ +import { ExternalApi } from '@prisma/client'; import { Response } from 'express'; +import _ from 'lodash'; import { z } from 'zod'; +import { prisma } from '../../../client.js'; import { sendSuccess, sendError } from '../../../core/api.js'; import { AuthenticatedRequest } from '../../../core/types.js'; import { logger as parentLogger } from '../../../logger.js'; @@ -19,18 +22,69 @@ export const triggerRecommendation = async (req: AuthenticatedRequest, res: Resp } const { body } = triggerRefereeRecommendationSchema.parse(req); + const { fileUrl, fileHash } = body; logger.info( { userId: user.id, - cid: body.cid, - external: body.external, + fileUrl, + fileHash, }, - 'Triggering referee recommendation', + 'Checking cache and potentially triggering referee recommendation', ); + // Check if we have cached results for this fileHash (if provided) + let cachedResult = null; + + if (fileHash) { + cachedResult = await prisma.externalApiUsage.findFirst({ + where: { + userId: user.id, + apiType: ExternalApi.REFEREE_FINDER, + queryingData: { + path: ['fileHash'], + equals: fileHash, + }, + }, + orderBy: { + createdAt: 'desc', // Get the most recent result + }, + }); + + if (cachedResult) { + logger.info( + { + userId: user.id, + fileHash, + cachedResultId: cachedResult.id, + }, + 'Found cached results for fileHash, returning cached data', + ); + + return sendSuccess(res, { + cached: true, + message: 'Results for this file are already available', + createdAt: cachedResult.createdAt, + resultKey: RefereeRecommenderService.prepareFormattedFileName(fileHash), + }); + } + } + const result = await RefereeRecommenderService.triggerRefereeRecommendation( - body as typeof body & { cid: string }, + { + // hash_value, + // hash_verified, + file_url: body.fileUrl, + top_n_closely_matching: body.top_n_closely_matching, + number_referees: body.number_referees, + force_run: body.force_run, + classify: body.classify, + coi_filter: body.coi_filter, + meta_data_only: body.meta_data_only, + exclude_fields: body.exclude_fields, + exclude_works: body.exclude_works, + exclude_authors: body.exclude_authors, + }, user.id, ); @@ -39,7 +93,6 @@ export const triggerRecommendation = async (req: AuthenticatedRequest, res: Resp { error: result.error, userId: user.id, - cid: body.cid, }, 'Failed to trigger referee recommendation', ); @@ -54,13 +107,23 @@ export const triggerRecommendation = async (req: AuthenticatedRequest, res: Resp logger.info( { userId: user.id, - uploadedFileName: responseData.uploaded_file_name, + uploadedFileName: responseData.UploadedFileName, info: responseData.info, }, 'Successfully triggered referee recommendation', ); - return sendSuccess(res, responseData); + const formattedResponse = _.omit(responseData, 'execution_arn'); + + // Check if results are already available based on info message + const isResultReady = + responseData.info?.toLowerCase().includes('already exists') && + responseData.info?.toLowerCase().includes('ready to be polled'); + + return sendSuccess(res, { + ...formattedResponse, + cached: isResultReady, + }); } catch (error) { if (error instanceof z.ZodError) { logger.warn({ error: error.errors }, 'Invalid request parameters'); diff --git a/desci-server/src/docs/aiServices.ts b/desci-server/src/docs/aiServices.ts new file mode 100644 index 000000000..95e7e0d15 --- /dev/null +++ b/desci-server/src/docs/aiServices.ts @@ -0,0 +1,339 @@ +import { z } from 'zod'; +import { ZodOpenApiOperationObject, ZodOpenApiPathsObject } from 'zod-openapi'; + +// Schemas +const FileHashSchema = z.object({ + fileHash: z.string().min(1).describe('SHA-256 hash of the file content'), + fileName: z.string().min(1).describe('Original filename'), +}); + +const PresignedUrlResponseSchema = z.object({ + cached: z.boolean().describe('Whether results are cached'), + presignedUrl: z.string().optional().describe('S3 presigned upload URL'), + downloadUrl: z.string().optional().describe('S3 download URL'), + fileName: z.string().optional().describe('Generated S3 filename'), + fileHash: z.string().optional().describe('File hash'), + resultKey: z.string().optional().describe('Formatted cache key for results'), + message: z.string().optional().describe('Cache message'), +}); + +const TriggerRefereeRequestSchema = z.object({ + fileUrl: z.string().url().describe('URL to the PDF file'), + fileHash: z.string().optional().describe('SHA-256 hash of file (for caching)'), + top_n_closely_matching: z + .number() + .int() + .min(1) + .max(50) + .optional() + .default(5) + .describe('Number of closely matching papers to analyze'), + number_referees: z + .number() + .int() + .min(1) + .max(50) + .optional() + .default(10) + .describe('Number of referee recommendations to return'), + force_run: z.boolean().optional().default(false).describe('Force re-processing even if cached results exist'), + classify: z.boolean().optional().describe('Enable classification features'), + coi_filter: z + .object({ + co_author_overlap: z.boolean().optional(), + institution_overlap: z.boolean().optional(), + supervisor_supervisee_check: z.boolean().optional(), + }) + .optional() + .describe('Conflict of interest filtering options'), + meta_data_only: z.boolean().optional().describe('Return only metadata'), + exclude_fields: z.array(z.string()).optional().describe('Research fields to exclude'), + exclude_works: z.array(z.string()).optional().describe('Specific works to exclude'), + exclude_authors: z.array(z.string()).optional().describe('Specific authors to exclude'), +}); + +const TriggerRefereeResponseSchema = z.object({ + cached: z.boolean().describe('Whether results were cached'), + UploadedFileName: z.string().optional().describe('Backend filename for processing'), + api_version: z.string().optional().describe('API version used'), + info: z.string().optional().describe('Processing information'), + resultKey: z.string().optional().describe('Formatted cache key'), + fileHash: z.string().optional().describe('File hash'), + createdAt: z.string().optional().describe('Creation timestamp'), + data: z.any().optional().describe('Cached result data'), +}); + +const UsageStatusSchema = z.object({ + totalLimit: z.number().nullable().describe('Total monthly limit (null for unlimited)'), + totalUsed: z.number().describe('Total requests used this month'), + totalRemaining: z.number().nullable().describe('Remaining requests (null for unlimited)'), + planCodename: z.string().describe('User plan identifier'), + isWithinLimit: z.boolean().describe('Whether user is within usage limits'), +}); + +const RefereeResultsSchema = z.object({ + ok: z.boolean().describe('Request success status'), + data: z + .object({ + paper_data: z.object({ + abstract: z.string(), + title: z.string(), + author_info: z.array( + z.object({ + display_name: z.string(), + }), + ), + context_novelty: z.number(), + content_novelty: z.number(), + }), + focal_authors_data: z.array( + z.object({ + display_name: z.string(), + id: z.string(), + orcid: z.string(), + }), + ), + referees: z.object({ + recommended: z.array( + z.object({ + display_name: z.string(), + id: z.string(), + orcid: z.string(), + cited_by_count: z.number(), + affiliations: z.string(), + h_index: z.number(), + tags: z.array(z.string()), + works: z.array( + z.object({ + work_id: z.string(), + title: z.string(), + doi: z.string(), + similarity_score: z.number(), + }), + ), + work_similarity_score: z.number(), + topic_similarity_score: z.number(), + total_direct_citations: z.number().nullable(), + earliest_work: z.number(), + }), + ), + }), + }) + .describe('Referee recommendation results'), +}); + +const ErrorResponseSchema = z.object({ + ok: z.boolean().default(false), + message: z.string().describe('Error message'), +}); + +// Operations +export const generatePresignedUrlOperation: ZodOpenApiOperationObject = { + operationId: 'generatePresignedUrl', + tags: ['AI Services'], + summary: 'Generate presigned URL for referee recommender file upload', + description: + 'Get a presigned S3 URL for uploading a PDF file for referee recommendation analysis. Checks for cached results first.', + security: [{ bearerAuth: [] }], + requestBody: { + content: { + 'application/json': { + schema: FileHashSchema, + }, + }, + }, + responses: { + '200': { + description: 'Presigned URL generated successfully or cached results found', + content: { + 'application/json': { + schema: PresignedUrlResponseSchema, + }, + }, + }, + '400': { + description: 'Invalid request parameters', + content: { + 'application/json': { + schema: ErrorResponseSchema, + }, + }, + }, + '401': { + description: 'Authentication required', + content: { + 'application/json': { + schema: ErrorResponseSchema, + }, + }, + }, + '500': { + description: 'Internal server error', + content: { + 'application/json': { + schema: ErrorResponseSchema, + }, + }, + }, + }, +}; + +export const triggerRefereeRecommendationOperation: ZodOpenApiOperationObject = { + operationId: 'triggerRefereeRecommendation', + tags: ['AI Services'], + summary: 'Trigger referee recommendation analysis', + description: + 'Start processing a PDF file for referee recommendations. Supports caching and various filtering options.', + security: [{ bearerAuth: [] }], + requestBody: { + content: { + 'application/json': { + schema: TriggerRefereeRequestSchema, + }, + }, + }, + responses: { + '200': { + description: 'Processing triggered successfully or cached results returned', + content: { + 'application/json': { + schema: TriggerRefereeResponseSchema, + }, + }, + }, + '400': { + description: 'Invalid request parameters', + content: { + 'application/json': { + schema: ErrorResponseSchema, + }, + }, + }, + '401': { + description: 'Authentication required', + content: { + 'application/json': { + schema: ErrorResponseSchema, + }, + }, + }, + '429': { + description: 'Rate limit exceeded', + content: { + 'application/json': { + schema: ErrorResponseSchema, + }, + }, + }, + '500': { + description: 'Internal server error', + content: { + 'application/json': { + schema: ErrorResponseSchema, + }, + }, + }, + }, +}; + +export const getRefereeResultsOperation: ZodOpenApiOperationObject = { + operationId: 'getRefereeResults', + tags: ['AI Services'], + summary: 'Get referee recommendation results', + description: 'Retrieve the results of referee recommendation analysis using the uploaded filename.', + security: [{ bearerAuth: [] }], + parameters: [ + { + name: 'UploadedFileName', + in: 'query', + required: true, + description: 'Backend filename from processing', + schema: { + type: 'string', + }, + }, + ], + responses: { + '200': { + description: 'Results retrieved successfully', + content: { + 'application/json': { + schema: RefereeResultsSchema, + }, + }, + }, + '404': { + description: 'Results not found', + content: { + 'application/json': { + schema: ErrorResponseSchema, + }, + }, + }, + '401': { + description: 'Authentication required', + content: { + 'application/json': { + schema: ErrorResponseSchema, + }, + }, + }, + '500': { + description: 'Internal server error', + content: { + 'application/json': { + schema: ErrorResponseSchema, + }, + }, + }, + }, +}; + +export const getRefereeUsageOperation: ZodOpenApiOperationObject = { + operationId: 'getRefereeUsage', + tags: ['AI Services'], + summary: 'Get referee recommender usage status', + description: 'Check current usage limits and remaining quota for referee recommender service.', + security: [{ bearerAuth: [] }], + responses: { + '200': { + description: 'Usage status retrieved successfully', + content: { + 'application/json': { + schema: UsageStatusSchema, + }, + }, + }, + '401': { + description: 'Authentication required', + content: { + 'application/json': { + schema: ErrorResponseSchema, + }, + }, + }, + '500': { + description: 'Internal server error', + content: { + 'application/json': { + schema: ErrorResponseSchema, + }, + }, + }, + }, +}; + +export const aiServicesPaths: ZodOpenApiPathsObject = { + '/services/ai/referee-recommender/presigned-url': { + post: generatePresignedUrlOperation, + }, + '/services/ai/referee-recommender/trigger': { + post: triggerRefereeRecommendationOperation, + }, + '/services/ai/referee-recommender/results': { + get: getRefereeResultsOperation, + }, + '/services/ai/referee-recommender/usage': { + get: getRefereeUsageOperation, + }, +}; diff --git a/desci-server/src/docs/openai.ts b/desci-server/src/docs/openai.ts index 5ada02851..1eb230827 100644 --- a/desci-server/src/docs/openai.ts +++ b/desci-server/src/docs/openai.ts @@ -31,6 +31,7 @@ import { updateSubmissionStatusOperation, } from './submissions.js'; import { userPaths } from './users.js'; +import { aiServicesPaths } from './aiServices.js'; export const analyticsPaths = { '/v1/admin/analytics': { @@ -96,6 +97,7 @@ export const openaiDocumentation = createDocument({ { name: 'OpenAlex', description: 'OpenAlex related operations' }, { name: 'Authors', description: 'Authors-related operations' }, { name: 'Search', description: 'Elastic search api operations' }, + { name: 'AI Services', description: 'AI-powered research tools and services' }, ], components: { securitySchemes: { @@ -140,6 +142,7 @@ export const openaiDocumentation = createDocument({ ...journalPaths, ...adminNodesPaths, ...metricsPaths, + ...aiServicesPaths, }, }); diff --git a/desci-server/src/schemas/externalApi.schema.ts b/desci-server/src/schemas/externalApi.schema.ts index d2ea6282e..7126986c4 100644 --- a/desci-server/src/schemas/externalApi.schema.ts +++ b/desci-server/src/schemas/externalApi.schema.ts @@ -14,8 +14,8 @@ export const generatePresignedUrlSchema = z.object({ // Trigger Referee Recommendation Request export const triggerRefereeRecommendationSchema = z.object({ body: z.object({ - cid: z.string().min(1, 'CID is required'), - external: z.boolean().optional().default(false), + fileUrl: z.string().min(1, 'fileUrl is required'), + fileHash: z.string().min(1, 'fileHash is required').optional(), top_n_closely_matching: z.number().int().min(1).max(50).optional().default(5), number_referees: z.number().int().min(1).max(50).optional().default(10), force_run: z.boolean().optional().default(false), diff --git a/desci-server/src/services/externalApi/RefereeRecommenderService.ts b/desci-server/src/services/externalApi/RefereeRecommenderService.ts index 59dc58c54..11719186a 100644 --- a/desci-server/src/services/externalApi/RefereeRecommenderService.ts +++ b/desci-server/src/services/externalApi/RefereeRecommenderService.ts @@ -1,7 +1,5 @@ -import { randomUUID } from 'crypto'; +import { randomUUID, createHash } from 'crypto'; -import { GetObjectCommand, PutObjectCommand, S3Client } from '@aws-sdk/client-s3'; -import { getSignedUrl } from '@aws-sdk/s3-request-presigner'; import { ExternalApi, Feature } from '@prisma/client'; import axios from 'axios'; import { ok, err, Result } from 'neverthrow'; @@ -15,21 +13,14 @@ const logger = parentLogger.child({ module: 'RefereeRecommender::Service' }); export const SESSION_TTL_SECONDS = 86400; // 24 hours -// Initialize S3 client and config at module level -const s3Client = new S3Client({ - region: process.env.AWS_REGION || 'us-east-2', - credentials: { - accessKeyId: process.env.AWS_ACCESS_KEY_ID || '', - secretAccessKey: process.env.AWS_SECRET_ACCESS_KEY || '', - }, -}); - -const BUCKET_NAME = process.env.REFEREE_RECOMMENDER_S3_BUCKET || ''; const API_VERSION = process.env.REFEREE_RECOMMENDER_VERSION || '0.1.3'; // ML Referee API endpoints -const ML_REFEREE_TRIGGER_URL = process.env.ML_REFEREE_TRIGGER_CID; +const ML_REFEREE_TRIGGER_URL = process.env.ML_REFEREE_TRIGGER_START; const ML_REFEREE_RESULTS_URL = process.env.ML_REFEREE_FINDER_RESULT; +const ML_REFEREE_GET_PRESIGNED_URL = process.env.ML_REFEREE_GET_PRESIGNED_URL; + +const ML_REFEREE_FILE_PREFIX = process.env.ML_REFEREE_FILE_PREFIX; interface PresignedUrlRequest { userId: number; @@ -38,18 +29,20 @@ interface PresignedUrlRequest { interface RefereeRecommenderSession { userId: number; - originalFileName: string; + hashedFileUrl: string; createdAt: number; } interface PresignedUrlResponse { presignedUrl: string; + downloadUrl: string; fileName: string; } interface TriggerRefereeRequest { - cid: string; - external?: boolean; + file_url: string; + hash_value?: string; + hash_verified?: boolean; top_n_closely_matching?: number; number_referees?: number; force_run?: boolean; @@ -66,10 +59,10 @@ interface TriggerRefereeRequest { } interface TriggerRefereeResponse { - execution_arn?: string; - uploaded_file_name: string; + UploadedFileName: string; api_version: string; info: string; + cached: boolean; } interface GetRefereeResultsResponse { @@ -78,6 +71,32 @@ interface GetRefereeResultsResponse { result?: any; } +/** + * Hashes a file URL with SHA256, and encodes it in Base64 + * This is for a short cache key in redis for URLs, S3 file URLs can be around ~1500 chars otherwise. + * @param fileUrl - The file URL to hash + * @returns The Base64-encoded hash of the file URL + */ +function prepareSessionCacheKey(fileUrl: string) { + // We use the file URLs here instead of the file hash, because it's the only + // deterministic link that we have available early on for every upload method (S3, CID, External pdf url) + // for us to be able to retrieve a result back. + // string -> SHA256 -> Base64 + return createHash('sha256').update(fileUrl).digest('base64'); // ~44 chars with '=' padding +} + +/** + * Results are stored in this format: {API}_{API_VERSION}_{SHA256FILE_HASH} + * @param fileHash - The SHA256 hash of the file URL + */ +function prepareFormattedFileName(fileHash: string) { + if (!ML_REFEREE_FILE_PREFIX) { + console.error('ML_REFEREE_FILE_PREFIX is not set'); + throw new Error('ML_REFEREE_FILE_PREFIX env is not set'); + } + return ML_REFEREE_FILE_PREFIX + fileHash; +} + async function generatePresignedUploadUrl(request: PresignedUrlRequest): Promise> { try { // Check feature limits first @@ -120,25 +139,32 @@ async function generatePresignedUploadUrl(request: PresignedUrlRequest): Promise 'Feature limit check passed for presigned URL generation', ); - // Generate filename with convention: referee_rec_v0.1.3_{original_file.pdf} - // original file name is salted with a UUID atm, will change to a hash in the future. + // Call external endpoint to get presigned URL const fileName = generateFileName(request.originalFileName); - // Create presigned URL for upload - const command = new PutObjectCommand({ - Bucket: BUCKET_NAME, - Key: fileName, - ContentType: 'application/pdf', - }); + if (!ML_REFEREE_GET_PRESIGNED_URL) { + return err(new Error('ML_REFEREE_GET_PRESIGNED_URL environment variable not configured')); + } - const presignedUrl = await getSignedUrl(s3Client, command, { - expiresIn: 3600, // 1 hour + const presignedResponse = await axios.get(ML_REFEREE_GET_PRESIGNED_URL, { + params: { + file_name: fileName, + }, }); - // Store session in Redis - const storeResult = await storeSession(fileName, { + if (presignedResponse.status !== 200) { + return err(new Error(`Presigned URL service returned status ${presignedResponse.status}`)); + } + + const { upload_url, download_url, s3_file_name } = presignedResponse.data; + + // Use download_url for cache key and store hashedFileUrl in session + const hashedFileUrl = prepareSessionCacheKey(download_url); + + // Store session in Redis using the hashed download URL + const storeResult = await storeSession(download_url, { userId: request.userId, - originalFileName: request.originalFileName, + hashedFileUrl, createdAt: Date.now(), }); @@ -146,17 +172,19 @@ async function generatePresignedUploadUrl(request: PresignedUrlRequest): Promise return err(storeResult.error); } - await markSessionAsActive(request.userId, fileName); + await markSessionAsActive(request.userId, download_url); logger.info( { userId: request.userId, - fileName, + originalFileName: request.originalFileName, + s3FileName: s3_file_name, + hashedFileUrl, }, - 'Generated presigned URL for referee recommender', + 'Generated presigned URL for referee recommender via external endpoint', ); - return ok({ presignedUrl, fileName }); + return ok({ presignedUrl: upload_url, downloadUrl: download_url, fileName: s3_file_name }); } catch (error) { logger.error({ error, request }, 'Failed to generate presigned URL'); return err(error instanceof Error ? error : new Error('Failed to generate presigned URL')); @@ -176,13 +204,15 @@ function generateFileName(originalFileName: string): string { return `referee_rec_v${API_VERSION}_${uuid}_${sanitizedBaseName}.${fileExtension}`; } -async function storeSession(fileName: string, session: RefereeRecommenderSession): Promise> { +async function storeSession(fileUrl: string, session: RefereeRecommenderSession): Promise> { try { - const userCacheKey = `referee-recommender-session:${session.userId}:${fileName}`; + const hashedFileUrl = prepareSessionCacheKey(fileUrl); + + const userCacheKey = `referee-recommender-session:${session.userId}:${hashedFileUrl}`; await setToCache(userCacheKey, session, SESSION_TTL_SECONDS); // Store/append to filename-based lookup for SQS handler - const filenameCacheKey = `referee-recommender-filename:${fileName}`; + const filenameCacheKey = `referee-recommender-file-url:${hashedFileUrl}`; const existingSessions = (await getFromCache(filenameCacheKey)) || []; const now = Date.now(); @@ -200,7 +230,7 @@ async function storeSession(fileName: string, session: RefereeRecommenderSession logger.debug( { - fileName, + fileUrl, userId: session.userId, sessionCount: validSessions.length, }, @@ -208,32 +238,33 @@ async function storeSession(fileName: string, session: RefereeRecommenderSession ); return ok(undefined); } catch (error) { - logger.error({ error, fileName }, 'Failed to store session in Redis'); + logger.error({ error, fileUrl }, 'Failed to store session in Redis'); return err(error instanceof Error ? error : new Error('Failed to store session in Redis')); } } -async function getSession(fileName: string, userId: number): Promise> { +async function getSession(fileUrlHash: string, userId: number): Promise> { try { - const cacheKey = `referee-recommender-session:${userId}:${fileName}`; + const cacheKey = `referee-recommender-session:${userId}:${fileUrlHash}`; const session = await getFromCache(cacheKey); if (session) { - logger.debug({ fileName, userId: session.userId }, 'Found referee recommender session'); + logger.debug({ fileUrlHash, userId: session.userId }, 'Found referee recommender session'); return ok(session); } - logger.debug({ fileName, userId }, 'Session not found in Redis'); + logger.debug({ fileUrlHash, userId }, 'Session not found in Redis'); return err(new Error('Session not found')); } catch (error) { - logger.error({ error, fileName, userId }, 'Failed to get session from Redis'); + logger.error({ error, fileUrlHash, userId }, 'Failed to get session from Redis'); return err(error instanceof Error ? error : new Error('Failed to get session from Redis')); } } -async function getSessionsByFileName(fileName: string): Promise> { +async function getSessionsByFileUrl(fileUrl: string): Promise> { try { - const filenameCacheKey = `referee-recommender-filename:${fileName}`; + const hashedFileUrl = prepareSessionCacheKey(fileUrl); + const filenameCacheKey = `referee-recommender-file-url:${hashedFileUrl}`; const sessions = await getFromCache(filenameCacheKey); if (sessions && sessions.length > 0) { @@ -254,7 +285,7 @@ async function getSessionsByFileName(fileName: string): Promise { +async function markSessionAsActive(userId: number, fileUrl: string): Promise { try { + const hashedFileUrl = prepareSessionCacheKey(fileUrl); const activeSessionsKey = `referee-recommender-active:${userId}`; const activeSessions = (await getFromCache(activeSessionsKey)) || []; // Add this filename to active sessions if not already present - if (!activeSessions.includes(fileName)) { - activeSessions.push(fileName); + if (!activeSessions.includes(hashedFileUrl)) { + activeSessions.push(hashedFileUrl); await setToCache(activeSessionsKey, activeSessions, SESSION_TTL_SECONDS); } } catch (error) { - logger.error({ error, userId, fileName }, 'Failed to mark session as active'); + logger.error({ error, userId, fileUrl }, 'Failed to mark session as active'); } } -async function markSessionAsCompleted(userId: number, fileName: string): Promise { +async function markSessionAsCompleted(userId: number, fileUrl: string): Promise { try { + const hashedFileUrl = prepareSessionCacheKey(fileUrl); const activeSessionsKey = `referee-recommender-active:${userId}`; const activeSessions = (await getFromCache(activeSessionsKey)) || []; // Remove this filename from active sessions - const updatedSessions = activeSessions.filter((session) => session !== fileName); + const updatedSessions = activeSessions.filter((session) => session !== hashedFileUrl); if (updatedSessions.length > 0) { await setToCache(activeSessionsKey, updatedSessions, SESSION_TTL_SECONDS); @@ -550,7 +601,7 @@ async function markSessionAsCompleted(userId: number, fileName: string): Promise await setToCache(activeSessionsKey, [], 1); // Short TTL to effectively delete } } catch (error) { - logger.error({ error, userId, fileName }, 'Failed to mark session as completed'); + logger.error({ error, userId, fileUrl }, 'Failed to mark session as completed'); } } @@ -651,7 +702,7 @@ async function cleanupExpiredSessions(userId: number): Promise { export const RefereeRecommenderService = { generatePresignedUploadUrl, getSession, - getSessionsByFileName, + getSessionsByFileUrl, triggerRefereeRecommendation, getRefereeResults, getUserUsageCount, @@ -661,4 +712,6 @@ export const RefereeRecommenderService = { markSessionAsCompleted, handleProcessingFailure, cleanupExpiredSessions, + prepareFormattedFileName, + prepareSessionCacheKey, }; diff --git a/desci-server/src/services/externalApi/RefereeRecommenderSqsHandler.ts b/desci-server/src/services/externalApi/RefereeRecommenderSqsHandler.ts index 07556aacb..a83068a8f 100644 --- a/desci-server/src/services/externalApi/RefereeRecommenderSqsHandler.ts +++ b/desci-server/src/services/externalApi/RefereeRecommenderSqsHandler.ts @@ -1,4 +1,4 @@ -import { ExternalApi } from '@prisma/client'; +import { ExternalApi, Prisma } from '@prisma/client'; import { prisma } from '../../client.js'; import { logger as parentLogger } from '../../logger.js'; @@ -10,9 +10,14 @@ import { RefereeRecommenderService, SESSION_TTL_SECONDS } from './RefereeRecomme const logger = parentLogger.child({ module: 'RefereeRecommender::SqsHandler' }); +interface RefereeRecommenderQueryingData { + fileName: string; + eventType: string; + fileHash?: string; +} + interface RefereeRecommenderUsageData { eventType: string; - fileName: string; originalFileName: string; timestamp: string; sessionCreatedAt: number; @@ -75,8 +80,11 @@ export class RefereeRecommenderSqsHandler { '[SQS] Processing referee recommender event', ); + // debugger; + const fileUrl = eventData.data.file_url; + // Look up user sessions from Redis using filename - const sessionsResult = await RefereeRecommenderService.getSessionsByFileName(eventData.file_name); + const sessionsResult = await RefereeRecommenderService.getSessionsByFileUrl(fileUrl); if (sessionsResult.isErr()) { logger.warn( { fileName: eventData.file_name, error: sessionsResult.error.message }, @@ -113,6 +121,11 @@ export class RefereeRecommenderSqsHandler { try { let eventType: WebSocketEventType; let eventPayload: any; + + // Compute fileUrlHashed using our consistent hashing method + // This ensures it matches what the frontend computed + const fileUrlHashed = RefereeRecommenderService.prepareSessionCacheKey(eventData.data.file_url); + switch (eventData.eventType) { case 'PROCESSING_COMPLETED': eventType = WebSocketEventType.REFEREE_REC_PROCESSING_COMPLETED; @@ -121,6 +134,8 @@ export class RefereeRecommenderSqsHandler { originalFileName: session.originalFileName, status: 'completed', timestamp: new Date().toISOString(), + fileUrlHashed, + paperTitle: eventData.data?.paper_title, }; break; @@ -132,6 +147,8 @@ export class RefereeRecommenderSqsHandler { status: 'failed', error: eventData.data?.error || 'Processing failed', timestamp: new Date().toISOString(), + fileUrlHashed, + paperTitle: eventData.data?.paper_title, }; break; @@ -142,6 +159,8 @@ export class RefereeRecommenderSqsHandler { originalFileName: session.originalFileName, status: 'started', timestamp: new Date().toISOString(), + fileUrlHashed, + paperTitle: eventData.data?.paper_title, }; break; @@ -151,9 +170,11 @@ export class RefereeRecommenderSqsHandler { fileName: eventData.file_name, originalFileName: session.originalFileName, status: 'processing', - progress: eventData.data?.progress_percent || 0, + progress_percent: eventData.data?.progress_percent || 0, // Keep original field name message: eventData.data?.message || '', timestamp: new Date().toISOString(), + fileUrlHashed, + paperTitle: eventData.data?.paper_title, }; break; @@ -198,7 +219,7 @@ export class RefereeRecommenderSqsHandler { where: { userId: session.userId, apiType: ExternalApi.REFEREE_FINDER, - data: { + queryingData: { path: ['fileName'], equals: eventData.file_name, }, @@ -240,9 +261,15 @@ export class RefereeRecommenderSqsHandler { } // Create new record for this session - const usageData: RefereeRecommenderUsageData = { - eventType: eventData.eventType, + // Separate queryable data (GIN indexed) from metadata + const queryingData: RefereeRecommenderQueryingData = { fileName: eventData.file_name, + eventType: eventData.eventType, + ...(eventData.file_hash && { fileHash: eventData.file_hash }), + }; + + const metadataData: RefereeRecommenderUsageData = { + eventType: eventData.eventType, originalFileName: session.originalFileName, timestamp: new Date().toISOString(), sessionCreatedAt: session.createdAt, @@ -253,7 +280,8 @@ export class RefereeRecommenderSqsHandler { data: { userId: session.userId, apiType: ExternalApi.REFEREE_FINDER, - data: usageData, + queryingData: queryingData as unknown as Prisma.JsonValue, // Fast queries on this field + data: metadataData, // Full metadata storage }, }); diff --git a/desci-server/src/services/sqs/SqsMessageTypes.ts b/desci-server/src/services/sqs/SqsMessageTypes.ts index ef07f21cb..3d5d8fc6e 100644 --- a/desci-server/src/services/sqs/SqsMessageTypes.ts +++ b/desci-server/src/services/sqs/SqsMessageTypes.ts @@ -28,5 +28,17 @@ export interface ExternalApiSqsMessage extends BaseSqsMessage { messageType: SqsMessageType.EXTERNAL_REFEREE_RECOMMENDER_API; eventType: 'PROCESSING_STARTED' | 'PROCESSING_COMPLETED' | 'PROCESSING_FAILED' | 'PROCESSING_PROGRESS'; file_name: string; // e.g., "referee_rec_v0.1.3_{original_file.pdf}" // Move to hashes later for security. - data?: any; + file_hash?: string; // SHA-256 hash of the file content + file_url: string; // Source URL + data?: ExternalApiSqsMessageData | any; + fileUrlHashed: string; } + +export type ExternalApiSqsMessageData = { + file_url: string; + hash_value?: string; + hash_verified?: boolean; + paper_title?: string; + api_version?: string; + completed_at?: number; // Unix TS +};