feat(core-api): add verification pipeline with vertex attire adapter
This commit is contained in:
zouantchaw
@@ -35,14 +35,34 @@ function toTextFromCandidate(candidate) {
|
||||
.trim();
|
||||
}
|
||||
|
||||
export async function invokeVertexModel({ prompt, responseJsonSchema, fileUrls = [] }) {
|
||||
const { project, location } = buildVertexConfig();
|
||||
const model = process.env.LLM_MODEL || 'gemini-2.0-flash-001';
|
||||
const timeoutMs = Number.parseInt(process.env.LLM_TIMEOUT_MS || '20000', 10);
|
||||
function withJsonSchemaInstruction(prompt, responseJsonSchema) {
|
||||
const schemaText = JSON.stringify(responseJsonSchema);
|
||||
const fileContext = fileUrls.length > 0 ? `\nFiles:\n${fileUrls.join('\n')}` : '';
|
||||
const instruction = `Respond with strict JSON only. Follow this schema exactly:\n${schemaText}`;
|
||||
const textPrompt = `${prompt}\n\n${instruction}${fileContext}`;
|
||||
return `${prompt}\n\nRespond with strict JSON only. Follow this schema exactly:\n${schemaText}`;
|
||||
}
|
||||
|
||||
function guessMimeTypeFromUri(fileUri) {
|
||||
const path = fileUri.split('?')[0].toLowerCase();
|
||||
if (path.endsWith('.jpg') || path.endsWith('.jpeg')) return 'image/jpeg';
|
||||
if (path.endsWith('.png')) return 'image/png';
|
||||
if (path.endsWith('.pdf')) return 'application/pdf';
|
||||
return 'application/octet-stream';
|
||||
}
|
||||
|
||||
function buildMultimodalParts(prompt, fileUris = []) {
|
||||
const parts = [{ text: prompt }];
|
||||
for (const fileUri of fileUris) {
|
||||
parts.push({
|
||||
fileData: {
|
||||
fileUri,
|
||||
mimeType: guessMimeTypeFromUri(fileUri),
|
||||
},
|
||||
});
|
||||
}
|
||||
return parts;
|
||||
}
|
||||
|
||||
async function callVertexJsonModel({ model, timeoutMs, parts }) {
|
||||
const { project, location } = buildVertexConfig();
|
||||
const url = `https://${location}-aiplatform.googleapis.com/v1/projects/${project}/locations/${location}/publishers/google/models/${model}:generateContent`;
|
||||
const auth = new GoogleAuth({
|
||||
scopes: ['https://www.googleapis.com/auth/cloud-platform'],
|
||||
@@ -56,7 +76,7 @@ export async function invokeVertexModel({ prompt, responseJsonSchema, fileUrls =
|
||||
url,
|
||||
method: 'POST',
|
||||
data: {
|
||||
contents: [{ role: 'user', parts: [{ text: textPrompt }] }],
|
||||
contents: [{ role: 'user', parts }],
|
||||
generationConfig: {
|
||||
temperature: 0.2,
|
||||
responseMimeType: 'application/json',
|
||||
@@ -91,3 +111,35 @@ export async function invokeVertexModel({ prompt, responseJsonSchema, fileUrls =
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
export async function invokeVertexModel({ prompt, responseJsonSchema, fileUrls = [] }) {
|
||||
const model = process.env.LLM_MODEL || 'gemini-2.0-flash-001';
|
||||
const timeoutMs = Number.parseInt(process.env.LLM_TIMEOUT_MS || '20000', 10);
|
||||
const promptWithSchema = withJsonSchemaInstruction(prompt, responseJsonSchema);
|
||||
const fileContext = fileUrls.length > 0 ? `\nFiles:\n${fileUrls.join('\n')}` : '';
|
||||
return callVertexJsonModel({
|
||||
model,
|
||||
timeoutMs,
|
||||
parts: [{ text: `${promptWithSchema}${fileContext}` }],
|
||||
});
|
||||
}
|
||||
|
||||
export async function invokeVertexMultimodalModel({
|
||||
prompt,
|
||||
responseJsonSchema,
|
||||
fileUris = [],
|
||||
model,
|
||||
timeoutMs,
|
||||
}) {
|
||||
const resolvedModel = model || process.env.LLM_MODEL || 'gemini-2.0-flash-001';
|
||||
const resolvedTimeoutMs = Number.parseInt(
|
||||
`${timeoutMs || process.env.LLM_TIMEOUT_MS || '20000'}`,
|
||||
10
|
||||
);
|
||||
const promptWithSchema = withJsonSchemaInstruction(prompt, responseJsonSchema);
|
||||
return callVertexJsonModel({
|
||||
model: resolvedModel,
|
||||
timeoutMs: resolvedTimeoutMs,
|
||||
parts: buildMultimodalParts(promptWithSchema, fileUris),
|
||||
});
|
||||
}
|
||||
|
||||
@@ -72,3 +72,12 @@ export async function generateReadSignedUrl({ fileUri, actorUid, expiresInSecond
|
||||
expiresAt: new Date(expiresAtMs).toISOString(),
|
||||
};
|
||||
}
|
||||
|
||||
export async function ensureFileExistsForActor({ fileUri, actorUid }) {
|
||||
const { bucket, path } = validateFileUriAccess({ fileUri, actorUid });
|
||||
const file = storage.bucket(bucket).file(path);
|
||||
const [exists] = await file.exists();
|
||||
if (!exists) {
|
||||
throw new AppError('NOT_FOUND', 'Evidence file not found', 404, { fileUri });
|
||||
}
|
||||
}
|
||||
|
||||
510
backend/core-api/src/services/verification-jobs.js
Normal file
510
backend/core-api/src/services/verification-jobs.js
Normal file
@@ -0,0 +1,510 @@
|
||||
import crypto from 'node:crypto';
|
||||
import { AppError } from '../lib/errors.js';
|
||||
import { invokeVertexMultimodalModel } from './llm.js';
|
||||
|
||||
const jobs = new Map();
|
||||
|
||||
export const VerificationStatus = Object.freeze({
|
||||
PENDING: 'PENDING',
|
||||
PROCESSING: 'PROCESSING',
|
||||
AUTO_PASS: 'AUTO_PASS',
|
||||
AUTO_FAIL: 'AUTO_FAIL',
|
||||
NEEDS_REVIEW: 'NEEDS_REVIEW',
|
||||
APPROVED: 'APPROVED',
|
||||
REJECTED: 'REJECTED',
|
||||
ERROR: 'ERROR',
|
||||
});
|
||||
|
||||
const MACHINE_TERMINAL_STATUSES = new Set([
|
||||
VerificationStatus.AUTO_PASS,
|
||||
VerificationStatus.AUTO_FAIL,
|
||||
VerificationStatus.NEEDS_REVIEW,
|
||||
VerificationStatus.ERROR,
|
||||
]);
|
||||
|
||||
const HUMAN_TERMINAL_STATUSES = new Set([
|
||||
VerificationStatus.APPROVED,
|
||||
VerificationStatus.REJECTED,
|
||||
]);
|
||||
|
||||
function nowIso() {
|
||||
return new Date().toISOString();
|
||||
}
|
||||
|
||||
function accessMode() {
|
||||
return process.env.VERIFICATION_ACCESS_MODE || 'authenticated';
|
||||
}
|
||||
|
||||
function eventRecord({ fromStatus, toStatus, actorType, actorId, details = {} }) {
|
||||
return {
|
||||
id: crypto.randomUUID(),
|
||||
fromStatus,
|
||||
toStatus,
|
||||
actorType,
|
||||
actorId,
|
||||
details,
|
||||
createdAt: nowIso(),
|
||||
};
|
||||
}
|
||||
|
||||
function toPublicJob(job) {
|
||||
return {
|
||||
verificationId: job.id,
|
||||
type: job.type,
|
||||
subjectType: job.subjectType,
|
||||
subjectId: job.subjectId,
|
||||
fileUri: job.fileUri,
|
||||
status: job.status,
|
||||
confidence: job.confidence,
|
||||
reasons: job.reasons,
|
||||
extracted: job.extracted,
|
||||
provider: job.provider,
|
||||
review: job.review,
|
||||
createdAt: job.createdAt,
|
||||
updatedAt: job.updatedAt,
|
||||
};
|
||||
}
|
||||
|
||||
function assertAccess(job, actorUid) {
|
||||
if (accessMode() === 'authenticated') {
|
||||
return;
|
||||
}
|
||||
if (job.ownerUid !== actorUid) {
|
||||
throw new AppError('FORBIDDEN', 'Not allowed to access this verification', 403);
|
||||
}
|
||||
}
|
||||
|
||||
function requireJob(id) {
|
||||
const job = jobs.get(id);
|
||||
if (!job) {
|
||||
throw new AppError('NOT_FOUND', 'Verification not found', 404, { verificationId: id });
|
||||
}
|
||||
return job;
|
||||
}
|
||||
|
||||
function normalizeMachineStatus(status) {
|
||||
if (
|
||||
status === VerificationStatus.AUTO_PASS
|
||||
|| status === VerificationStatus.AUTO_FAIL
|
||||
|| status === VerificationStatus.NEEDS_REVIEW
|
||||
) {
|
||||
return status;
|
||||
}
|
||||
return VerificationStatus.NEEDS_REVIEW;
|
||||
}
|
||||
|
||||
function clampConfidence(value, fallback = 0.5) {
|
||||
const parsed = Number(value);
|
||||
if (!Number.isFinite(parsed)) return fallback;
|
||||
if (parsed < 0) return 0;
|
||||
if (parsed > 1) return 1;
|
||||
return parsed;
|
||||
}
|
||||
|
||||
function asReasonList(reasons, fallback) {
|
||||
if (Array.isArray(reasons) && reasons.length > 0) {
|
||||
return reasons.map((item) => `${item}`);
|
||||
}
|
||||
return [fallback];
|
||||
}
|
||||
|
||||
function providerTimeoutMs() {
|
||||
return Number.parseInt(process.env.VERIFICATION_PROVIDER_TIMEOUT_MS || '8000', 10);
|
||||
}
|
||||
|
||||
function attireModel() {
|
||||
return process.env.VERIFICATION_ATTIRE_MODEL || 'gemini-2.0-flash-lite-001';
|
||||
}
|
||||
|
||||
async function runAttireChecks(job) {
|
||||
if (process.env.VERIFICATION_ATTIRE_AUTOPASS === 'true') {
|
||||
return {
|
||||
status: VerificationStatus.AUTO_PASS,
|
||||
confidence: 0.8,
|
||||
reasons: ['Auto-pass mode enabled for attire in dev'],
|
||||
extracted: {
|
||||
expected: job.rules,
|
||||
},
|
||||
provider: {
|
||||
name: 'attire-auto-pass',
|
||||
reference: null,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
const attireProvider = process.env.VERIFICATION_ATTIRE_PROVIDER || 'vertex';
|
||||
if (attireProvider !== 'vertex') {
|
||||
return {
|
||||
status: VerificationStatus.NEEDS_REVIEW,
|
||||
confidence: 0.45,
|
||||
reasons: [`Attire provider '${attireProvider}' is not supported`],
|
||||
extracted: {
|
||||
expected: job.rules,
|
||||
},
|
||||
provider: {
|
||||
name: attireProvider,
|
||||
reference: null,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
try {
|
||||
const prompt = [
|
||||
'You are validating worker attire evidence.',
|
||||
`Rules: ${JSON.stringify(job.rules || {})}`,
|
||||
'Return AUTO_PASS only when the image clearly matches required attire.',
|
||||
'Return AUTO_FAIL when the image clearly violates required attire.',
|
||||
'Return NEEDS_REVIEW when uncertain.',
|
||||
].join('\n');
|
||||
|
||||
const schema = {
|
||||
type: 'object',
|
||||
properties: {
|
||||
status: { type: 'string' },
|
||||
confidence: { type: 'number' },
|
||||
reasons: {
|
||||
type: 'array',
|
||||
items: { type: 'string' },
|
||||
},
|
||||
extracted: {
|
||||
type: 'object',
|
||||
additionalProperties: true,
|
||||
},
|
||||
},
|
||||
required: ['status', 'confidence', 'reasons'],
|
||||
};
|
||||
|
||||
const modelOutput = await invokeVertexMultimodalModel({
|
||||
prompt,
|
||||
responseJsonSchema: schema,
|
||||
fileUris: [job.fileUri],
|
||||
model: attireModel(),
|
||||
timeoutMs: providerTimeoutMs(),
|
||||
});
|
||||
|
||||
const result = modelOutput?.result || {};
|
||||
return {
|
||||
status: normalizeMachineStatus(result.status),
|
||||
confidence: clampConfidence(result.confidence, 0.6),
|
||||
reasons: asReasonList(result.reasons, 'Attire check completed'),
|
||||
extracted: result.extracted || {},
|
||||
provider: {
|
||||
name: 'vertex-attire',
|
||||
reference: modelOutput?.model || attireModel(),
|
||||
},
|
||||
};
|
||||
} catch (error) {
|
||||
return {
|
||||
status: VerificationStatus.NEEDS_REVIEW,
|
||||
confidence: 0.35,
|
||||
reasons: ['Automatic attire check unavailable, manual review required'],
|
||||
extracted: {},
|
||||
provider: {
|
||||
name: 'vertex-attire',
|
||||
reference: `error:${error?.code || 'unknown'}`,
|
||||
},
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
function getProviderConfig(type) {
|
||||
if (type === 'government_id') {
|
||||
return {
|
||||
name: 'government-id-provider',
|
||||
url: process.env.VERIFICATION_GOV_ID_PROVIDER_URL,
|
||||
token: process.env.VERIFICATION_GOV_ID_PROVIDER_TOKEN,
|
||||
};
|
||||
}
|
||||
return {
|
||||
name: 'certification-provider',
|
||||
url: process.env.VERIFICATION_CERT_PROVIDER_URL,
|
||||
token: process.env.VERIFICATION_CERT_PROVIDER_TOKEN,
|
||||
};
|
||||
}
|
||||
|
||||
async function runThirdPartyChecks(job, type) {
|
||||
const provider = getProviderConfig(type);
|
||||
if (!provider.url) {
|
||||
return {
|
||||
status: VerificationStatus.NEEDS_REVIEW,
|
||||
confidence: 0.4,
|
||||
reasons: [`${provider.name} is not configured`],
|
||||
extracted: {},
|
||||
provider: {
|
||||
name: provider.name,
|
||||
reference: null,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
const controller = new AbortController();
|
||||
const timeout = setTimeout(() => controller.abort(), providerTimeoutMs());
|
||||
|
||||
try {
|
||||
const response = await fetch(provider.url, {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
'Content-Type': 'application/json',
|
||||
...(provider.token ? { Authorization: `Bearer ${provider.token}` } : {}),
|
||||
},
|
||||
body: JSON.stringify({
|
||||
type,
|
||||
subjectType: job.subjectType,
|
||||
subjectId: job.subjectId,
|
||||
fileUri: job.fileUri,
|
||||
rules: job.rules,
|
||||
metadata: job.metadata,
|
||||
}),
|
||||
signal: controller.signal,
|
||||
});
|
||||
|
||||
const bodyText = await response.text();
|
||||
let body = {};
|
||||
try {
|
||||
body = bodyText ? JSON.parse(bodyText) : {};
|
||||
} catch {
|
||||
body = {};
|
||||
}
|
||||
|
||||
if (!response.ok) {
|
||||
return {
|
||||
status: VerificationStatus.NEEDS_REVIEW,
|
||||
confidence: 0.35,
|
||||
reasons: [`${provider.name} returned ${response.status}`],
|
||||
extracted: {},
|
||||
provider: {
|
||||
name: provider.name,
|
||||
reference: body?.reference || null,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
return {
|
||||
status: normalizeMachineStatus(body.status),
|
||||
confidence: clampConfidence(body.confidence, 0.6),
|
||||
reasons: asReasonList(body.reasons, `${provider.name} completed check`),
|
||||
extracted: body.extracted || {},
|
||||
provider: {
|
||||
name: provider.name,
|
||||
reference: body.reference || null,
|
||||
},
|
||||
};
|
||||
} catch (error) {
|
||||
const isAbort = error?.name === 'AbortError';
|
||||
return {
|
||||
status: VerificationStatus.NEEDS_REVIEW,
|
||||
confidence: 0.3,
|
||||
reasons: [
|
||||
isAbort
|
||||
? `${provider.name} timeout, manual review required`
|
||||
: `${provider.name} unavailable, manual review required`,
|
||||
],
|
||||
extracted: {},
|
||||
provider: {
|
||||
name: provider.name,
|
||||
reference: null,
|
||||
},
|
||||
};
|
||||
} finally {
|
||||
clearTimeout(timeout);
|
||||
}
|
||||
}
|
||||
|
||||
async function runMachineChecks(job) {
|
||||
if (job.type === 'attire') {
|
||||
return runAttireChecks(job);
|
||||
}
|
||||
|
||||
if (job.type === 'government_id') {
|
||||
return runThirdPartyChecks(job, 'government_id');
|
||||
}
|
||||
|
||||
return runThirdPartyChecks(job, 'certification');
|
||||
}
|
||||
|
||||
async function processVerificationJob(id) {
|
||||
const job = requireJob(id);
|
||||
if (job.status !== VerificationStatus.PENDING) {
|
||||
return;
|
||||
}
|
||||
|
||||
const beforeProcessing = job.status;
|
||||
job.status = VerificationStatus.PROCESSING;
|
||||
job.updatedAt = nowIso();
|
||||
job.events.push(
|
||||
eventRecord({
|
||||
fromStatus: beforeProcessing,
|
||||
toStatus: VerificationStatus.PROCESSING,
|
||||
actorType: 'system',
|
||||
actorId: 'verification-worker',
|
||||
})
|
||||
);
|
||||
|
||||
try {
|
||||
const outcome = await runMachineChecks(job);
|
||||
if (!MACHINE_TERMINAL_STATUSES.has(outcome.status)) {
|
||||
throw new Error(`Invalid machine outcome status: ${outcome.status}`);
|
||||
}
|
||||
const fromStatus = job.status;
|
||||
job.status = outcome.status;
|
||||
job.confidence = outcome.confidence;
|
||||
job.reasons = outcome.reasons;
|
||||
job.extracted = outcome.extracted;
|
||||
job.provider = outcome.provider;
|
||||
job.updatedAt = nowIso();
|
||||
job.events.push(
|
||||
eventRecord({
|
||||
fromStatus,
|
||||
toStatus: job.status,
|
||||
actorType: 'system',
|
||||
actorId: 'verification-worker',
|
||||
details: {
|
||||
confidence: job.confidence,
|
||||
reasons: job.reasons,
|
||||
provider: job.provider,
|
||||
},
|
||||
})
|
||||
);
|
||||
} catch (error) {
|
||||
const fromStatus = job.status;
|
||||
job.status = VerificationStatus.ERROR;
|
||||
job.confidence = null;
|
||||
job.reasons = [error?.message || 'Verification processing failed'];
|
||||
job.extracted = {};
|
||||
job.provider = {
|
||||
name: 'verification-worker',
|
||||
reference: null,
|
||||
};
|
||||
job.updatedAt = nowIso();
|
||||
job.events.push(
|
||||
eventRecord({
|
||||
fromStatus,
|
||||
toStatus: VerificationStatus.ERROR,
|
||||
actorType: 'system',
|
||||
actorId: 'verification-worker',
|
||||
details: {
|
||||
error: error?.message || 'Verification processing failed',
|
||||
},
|
||||
})
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
function queueVerificationProcessing(id) {
|
||||
setTimeout(() => {
|
||||
processVerificationJob(id).catch(() => {});
|
||||
}, 0);
|
||||
}
|
||||
|
||||
export function createVerificationJob({ actorUid, payload }) {
|
||||
const now = nowIso();
|
||||
const id = `ver_${crypto.randomUUID()}`;
|
||||
const job = {
|
||||
id,
|
||||
type: payload.type,
|
||||
subjectType: payload.subjectType || null,
|
||||
subjectId: payload.subjectId || null,
|
||||
ownerUid: actorUid,
|
||||
fileUri: payload.fileUri,
|
||||
rules: payload.rules || {},
|
||||
metadata: payload.metadata || {},
|
||||
status: VerificationStatus.PENDING,
|
||||
confidence: null,
|
||||
reasons: [],
|
||||
extracted: {},
|
||||
provider: null,
|
||||
review: null,
|
||||
createdAt: now,
|
||||
updatedAt: now,
|
||||
events: [
|
||||
eventRecord({
|
||||
fromStatus: null,
|
||||
toStatus: VerificationStatus.PENDING,
|
||||
actorType: 'system',
|
||||
actorId: actorUid,
|
||||
}),
|
||||
],
|
||||
};
|
||||
jobs.set(id, job);
|
||||
queueVerificationProcessing(id);
|
||||
return toPublicJob(job);
|
||||
}
|
||||
|
||||
export function getVerificationJob(verificationId, actorUid) {
|
||||
const job = requireJob(verificationId);
|
||||
assertAccess(job, actorUid);
|
||||
return toPublicJob(job);
|
||||
}
|
||||
|
||||
export function reviewVerificationJob(verificationId, actorUid, review) {
|
||||
const job = requireJob(verificationId);
|
||||
assertAccess(job, actorUid);
|
||||
|
||||
if (HUMAN_TERMINAL_STATUSES.has(job.status)) {
|
||||
throw new AppError('CONFLICT', 'Verification already finalized', 409, {
|
||||
verificationId,
|
||||
status: job.status,
|
||||
});
|
||||
}
|
||||
|
||||
const fromStatus = job.status;
|
||||
job.status = review.decision;
|
||||
job.review = {
|
||||
decision: review.decision,
|
||||
reviewedBy: actorUid,
|
||||
reviewedAt: nowIso(),
|
||||
note: review.note || '',
|
||||
reasonCode: review.reasonCode || 'MANUAL_REVIEW',
|
||||
};
|
||||
job.updatedAt = nowIso();
|
||||
job.events.push(
|
||||
eventRecord({
|
||||
fromStatus,
|
||||
toStatus: job.status,
|
||||
actorType: 'reviewer',
|
||||
actorId: actorUid,
|
||||
details: {
|
||||
reasonCode: job.review.reasonCode,
|
||||
},
|
||||
})
|
||||
);
|
||||
|
||||
return toPublicJob(job);
|
||||
}
|
||||
|
||||
export function retryVerificationJob(verificationId, actorUid) {
|
||||
const job = requireJob(verificationId);
|
||||
assertAccess(job, actorUid);
|
||||
|
||||
if (job.status === VerificationStatus.PROCESSING) {
|
||||
throw new AppError('CONFLICT', 'Cannot retry while verification is processing', 409, {
|
||||
verificationId,
|
||||
});
|
||||
}
|
||||
|
||||
const fromStatus = job.status;
|
||||
job.status = VerificationStatus.PENDING;
|
||||
job.confidence = null;
|
||||
job.reasons = [];
|
||||
job.extracted = {};
|
||||
job.provider = null;
|
||||
job.review = null;
|
||||
job.updatedAt = nowIso();
|
||||
job.events.push(
|
||||
eventRecord({
|
||||
fromStatus,
|
||||
toStatus: VerificationStatus.PENDING,
|
||||
actorType: 'reviewer',
|
||||
actorId: actorUid,
|
||||
details: {
|
||||
retried: true,
|
||||
},
|
||||
})
|
||||
);
|
||||
queueVerificationProcessing(verificationId);
|
||||
return toPublicJob(job);
|
||||
}
|
||||
|
||||
export function __resetVerificationJobsForTests() {
|
||||
jobs.clear();
|
||||
}
|
||||
Reference in New Issue
Block a user