diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 391c6fa..3b060c1 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -67,6 +67,11 @@ jobs: working-directory: backend run: npm test + - name: Run indexing integration tests (PostgreSQL) + working-directory: backend + timeout-minutes: 5 + run: npm run test:integration:indexing + - name: Audit production dependencies working-directory: backend run: npm run audit:security diff --git a/backend/package.json b/backend/package.json index 5bbfc96..6b30979 100644 --- a/backend/package.json +++ b/backend/package.json @@ -18,6 +18,7 @@ "test": "node scripts/test.mjs", "test:integration:s1": "tsx scripts/validate-s1.mts", "test:integration:s105": "tsx scripts/validate-s105.mts", + "test:integration:indexing": "tsx scripts/test-indexing-db.mts", "test:integration:search": "tsx --test src/modules/search/search.repository.db.test.ts" }, "dependencies": { diff --git a/backend/scripts/test-indexing-db.mts b/backend/scripts/test-indexing-db.mts new file mode 100644 index 0000000..7c84860 --- /dev/null +++ b/backend/scripts/test-indexing-db.mts @@ -0,0 +1,68 @@ +import { createServer, type Server } from "node:http"; +import { spawn, spawnSync } from "node:child_process"; +import { randomUUID } from "node:crypto"; +import { Client } from "pg"; +import { applyUntil } from "../src/database/migration-test-utils.js"; + +const name = "sinapse-s203-test-" + randomUUID().slice(0,8); +const password = randomUUID(); +function docker(args: string[]) { + const result = spawnSync("docker", args, { encoding: "utf8" }); + if (result.status !== 0) throw new Error("Não foi possível preparar o PostgreSQL de testes."); + return result.stdout.trim(); +} +let started = false; +let bridge: Server | undefined; +try { + docker(["run","-d","--name",name,"-e","POSTGRES_USER=sinapse","-e","POSTGRES_PASSWORD="+password, + "-e","POSTGRES_DB=sinapse_s203_test","-p","127.0.0.1::5432","pgvector/pgvector:pg16"]); + started = true; + const port = JSON.parse(docker(["inspect",name]))[0].NetworkSettings.Ports["5432/tcp"][0].HostPort; + const url = "postgresql://sinapse:"+password+"@127.0.0.1:"+port+"/sinapse_s203_test"; + let ready = false; + for (let i=0;i<40;i++) { + const result=spawnSync("docker",["exec",name,"pg_isready","-h","127.0.0.1","-U","sinapse","-d","sinapse_s203_test"],{encoding:"utf8"}); + if (result.status===0) { ready=true; break; } + await new Promise(r=>setTimeout(r,500)); + } + if (!ready) throw new Error("PostgreSQL não ficou pronto."); + const client = new Client({connectionString:url}); + await client.connect(); + try { await applyUntil(client,true,()=>false); } finally { await client.end(); } + console.log("Todas as migrações aplicadas em banco isolado com pgvector."); + let realEndpoint = ""; + if (process.argv.includes("--real-ai")) { + bridge = createServer(async (req, res) => { + const buffers = []; + for await (const part of req) buffers.push(part); + const python = [ + "import os,sys,urllib.request", + "payload=sys.stdin.buffer.read()", + "request=urllib.request.Request('http://127.0.0.1:8000/ingest/entity',data=payload,headers={'Content-Type':'application/json','X-Service-Token':os.environ['AI_SERVICE_TOKEN']})", + "response=urllib.request.urlopen(request,timeout=120)", + "sys.stdout.buffer.write(response.read())", + ].join("\n"); + const response = spawnSync("docker", ["exec","-i","sinapse-ai-service","python","-c",python], { + input: Buffer.concat(buffers), maxBuffer: 8 * 1024 * 1024, timeout: 125000, + }); + res.setHeader("Content-Type", "application/json"); + res.statusCode = response.status === 0 ? 200 : 502; + res.end(response.status === 0 ? response.stdout : JSON.stringify({ error: "Serviço IA indisponível." })); + }); + await new Promise(r => bridge!.listen(0, "127.0.0.1", r)); + realEndpoint = "http://127.0.0.1:" + (bridge.address() as { port: number }).port; + } + const exitCode = await new Promise((resolve, reject) => { + const child = spawn(process.execPath,["--import","tsx","--test", + "src/modules/indexing/indexing.db.test.ts","src/modules/indexing/indexing.migration.db.test.ts","src/modules/indexing/indexing.transport.db.test.ts","src/modules/indexing/indexing.state.db.test.ts","src/modules/indexing/indexing.recovery.db.test.ts","src/modules/indexing/indexing.worker.test.ts","src/modules/indexing/indexing.integration.db.test.ts"],{ + env:{...process.env,NODE_ENV:"test",DATABASE_URL:url,INDEX_TEST_DATABASE_URL:url,INDEX_AI_ENDPOINT:realEndpoint}, + stdio:"inherit", + }); + child.once("error", reject); + child.once("exit", code => resolve(code ?? 1)); + }); + process.exitCode=exitCode; +} finally { + if (bridge) { bridge.closeAllConnections(); await new Promise(r => bridge!.close(() => r())); } + if (started) { docker(["rm","-f","-v",name]); console.log("Banco temporário removido."); } +} diff --git a/backend/src/app.ts b/backend/src/app.ts index da8d917..a656d8d 100644 --- a/backend/src/app.ts +++ b/backend/src/app.ts @@ -1,3 +1,4 @@ +import { indexingRouter } from "./modules/indexing/indexing.routes.js"; import { createProjectAccessRouter } from "./middleware/projectAccess.js"; import express, { Request, Response } from "express"; import cors from "cors"; @@ -107,6 +108,8 @@ app.use("/api/v1/projects/:projectId/documents", requireAuth, documentsRouter); // Server-to-server: pipeline de ingestão persistindo chunks indexados do documento. app.use("/api/v1/projects/:projectId/documents/:documentId/chunks", chunksRouter); +app.use("/api/v1/projects/:projectId/index-events", indexingRouter); + // Swagger Documentation app.use('/docs', swaggerUi.serve, swaggerUi.setup(swaggerSpec)); diff --git a/backend/src/index.ts b/backend/src/index.ts index 163cd3e..e99d10e 100644 --- a/backend/src/index.ts +++ b/backend/src/index.ts @@ -1,3 +1,4 @@ +import { startIndexingWorker } from "./modules/indexing/indexing.worker.js"; import app from "./app.js"; import { startRepoAnalysesWorker } from "./modules/repo-analyses/repo-analyses.service.js"; import { env } from "./config/env.js"; @@ -10,6 +11,7 @@ if (env.NODE_ENV !== "test") { // Worker unico pos-merge S2-01: processa documentos pendentes via claim/lease // (processPendingDocuments) + flush de eventos + reconciliacao de arquivos. startDocumentsBackgroundWorker(); + startIndexingWorker(); startRepoAnalysesWorker(); if (!env.DOCUMENT_EVENTS_WEBHOOK_URL?.trim()) { console.warn("[Documents] DOCUMENT_EVENTS_WEBHOOK_URL is not configured; removal events will retry until a consumer is configured."); diff --git a/backend/src/modules/indexing/indexing.db.test.ts b/backend/src/modules/indexing/indexing.db.test.ts new file mode 100644 index 0000000..a9378e6 --- /dev/null +++ b/backend/src/modules/indexing/indexing.db.test.ts @@ -0,0 +1,299 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import { randomUUID } from "node:crypto"; +import { createServer } from "node:http"; +import { Pool } from "pg"; +import { SearchRepository } from "../search/search.repository.js"; +import { validateTarget } from "../../database/seed-lib.js"; +import { IndexingRepository } from "./indexing.repository.js"; +import { IndexingWorker } from "./indexing.worker.js"; + +test("S2-03: eventos transacionais, reindexação, falha, fencing, arquivamento e isolamento", { + skip: !process.env.INDEX_TEST_DATABASE_URL, +}, async () => { + const db = new Pool({ connectionString: validateTarget(process.env.INDEX_TEST_DATABASE_URL, "test") }); + const repository = new IndexingRepository(db); + const [project, other, epic, feature, pbi, technology] = Array.from({ length: 6 }, () => randomUUID()); + let mode: "ok" | "fail" | "scope" = "ok"; + let requests = 0; + const server = createServer(async (req, res) => { + const buffers = []; + for await (const part of req) buffers.push(part); + const request = JSON.parse(Buffer.concat(buffers).toString()); + requests++; + assert.equal(request.data.project_id, project); + res.setHeader("Content-Type", "application/json"); + res.statusCode = mode === "fail" ? 503 : 200; + res.end(JSON.stringify({ + entity_type: request.entity_type, + content: request.data.title + " " + request.data.description, + metadata: { project_id: mode === "scope" ? other : project }, + embedding: Array(1024).fill(0.1), + })); + }); + await new Promise(r => server.listen(0, "127.0.0.1", r)); + const port = (server.address() as { port: number }).port; + const worker = new IndexingWorker(repository, "http://127.0.0.1:" + port); + const event = async (id: string = pbi) => (await db.query("SELECT * FROM indice_evento WHERE projeto_id=$1 AND entidade_id=$2", [project, id])).rows[0]; + const chunks = async (id: string = pbi) => (await db.query("SELECT * FROM chunk WHERE projeto_id=$1 AND entidade_id=$2", [project, id])).rows; + try { + await db.query("INSERT INTO projeto(id,nome,cliente,status) VALUES($1,'Teste S2-03','Sintético','ativo'),($2,'Outro','Sintético','ativo')", [project, other]); + await db.query("INSERT INTO epico(id,projeto_id,titulo,status) VALUES($1,$2,'Épico','rascunho')", [epic, project]); + await db.query("INSERT INTO feature(id,epico_id,titulo,status) VALUES($1,$2,'Feature','rascunho')", [feature, epic]); + await db.query("INSERT INTO pbi(id,feature_id,codigo,titulo,historia_como_um,historia_eu_quero,historia_para_que,status) VALUES($1::uuid,$2,$1::text,'Consultar acervo','PO','consultar','reutilizar','rascunho')", [pbi, feature]); + assert.equal(await event(), undefined); + const client = await db.connect(); + try { + await client.query("BEGIN"); + await client.query("UPDATE pbi SET status='concluido' WHERE id=$1", [pbi]); + assert.equal((await client.query("SELECT count(*)::int n FROM indice_evento WHERE projeto_id=$1", [project])).rows[0].n, 1); + await client.query("ROLLBACK"); + } finally { client.release(); } + assert.equal(await event(), undefined); + + await db.query("UPDATE pbi SET status='concluido' WHERE id=$1", [pbi]); + const published = await event(); + assert.equal(published.payload.project_id, project); + assert.equal(published.payload.level, "pbi"); + assert.equal(published.payload.status, "concluido"); + await worker.tick(project); + assert.equal((await event()).status, "processado"); + assert.equal((await chunks()).length, 1); + await worker.tick(project); + assert.equal(requests, 1); + const originalChunk = (await chunks())[0].id; + + await db.query("SELECT sincronizar_indice_projeto($1)", [project]); + assert.equal((await event()).evento_id, published.evento_id); + await repository.retry(project, published.evento_id); + const queued = await event(); + await repository.retry(project, published.evento_id); + assert.equal((await event()).updated_at.getTime(), queued.updated_at.getTime()); + await worker.tick(project); + assert.equal((await chunks()).length, 1); + assert.equal((await chunks())[0].id, originalChunk); + + await db.query("UPDATE pbi SET titulo='Consultar versão nova' WHERE id=$1", [pbi]); + assert.equal((await chunks()).length, 0); + mode = "fail"; await worker.tick(project); + assert.equal((await event()).status, "falha"); + assert.equal((await event()).erro_codigo, "SERVICE_HTTP_ERROR"); + await repository.retry(project, (await event()).evento_id); + mode = "scope"; await worker.tick(project); + assert.equal((await event()).erro_codigo, "INVALID_SCOPE"); + assert.equal((await chunks()).length, 0); + mode = "ok"; + await repository.retry(project, (await event()).evento_id); + await worker.tick(project); + assert.match((await chunks())[0].texto, /versão nova/); + + await db.query("INSERT INTO tecnologia(id,nome) VALUES($1::uuid,$1::text)", [technology]); + await db.query("INSERT INTO entidade_tecnologia(entidade_tipo,entidade_id,tecnologia_id) VALUES('pbi',$1,$2)", [pbi, technology]); + assert.deepEqual((await event()).payload.tecnologias_ids, [technology]); + await db.query("INSERT INTO criterio_aceitacao(entidade_tipo,entidade_id,nome,dado,quando,entao,ordem) VALUES('pbi',$1,'Busca','Documento indexado','Consultar','Deve encontrar conteúdo',1)", [pbi]); + assert.match((await event()).payload.acceptanceCriteria[0], /encontrar/); + const beforeRename = (await event()).evento_id; + await db.query("UPDATE criterio_aceitacao SET nome='Busca por origem' WHERE entidade_id=$1", [pbi]); + assert.notEqual((await event()).evento_id, beforeRename); + assert.match((await event()).payload.acceptanceCriteria[0], /Busca por origem/); + await worker.tick(project); + + await repository.retry(project, (await event()).evento_id); + const stale = (await repository.claim(project))!; + assert.ok(stale); + assert.equal(await repository.claim(project), null); + await assert.rejects(repository.retry(project, stale.evento_id), /processamento/); + await db.query("UPDATE pbi SET titulo='Consultar terceira versão' WHERE id=$1", [pbi]); + await repository.finish(stale, { content: "Conteúdo atrasado", embedding: Array(1024).fill(0.1) }); + await repository.fail(stale, "PROCESSING_ERROR"); + assert.equal((await event()).status, "pendente"); + assert.equal((await chunks()).length, 0); + await worker.tick(project); + assert.match((await chunks())[0].texto, /terceira versão/); + + await db.query("UPDATE epico SET status='concluido' WHERE id=$1", [epic]); + await db.query("UPDATE feature SET status='concluido' WHERE id=$1", [feature]); + assert.equal((await event(epic)).payload.level, "epico"); + assert.equal((await event(feature)).payload.level, "feature"); + await worker.tick(project); + await worker.tick(project); + assert.equal((await chunks(epic)).length, 1); + assert.equal((await chunks(feature)).length, 1); + const decision = randomUUID(); + await db.query("INSERT INTO decisao(id,entidade_tipo,entidade_id,titulo,contexto,decisao,justificativa) VALUES($1,'feature',$2,'Usar fila','Concorrência','Fila persistida','Recuperação')", [decision, feature]); + assert.equal((await event(decision)).payload.level, "feature"); + await worker.tick(project); + assert.equal((await chunks(decision)).length, 1); + assert.deepEqual((await repository.list(other)).items, []); + await assert.rejects(repository.retry(other, (await event()).evento_id), /não encontrado/); + assert.equal((await db.query("SELECT count(*)::int n FROM chunk WHERE projeto_id=$1", [other])).rows[0].n, 0); + + const search = new SearchRepository(db); + const searchProject = (id: string) => search.hybridSearch({ query: "versão", projectId: id, limit: 20 }, Array(1024).fill(0.1), 0.5); + assert.ok((await searchProject(project)).some(row => row.entity_id === pbi)); + assert.equal((await searchProject(other)).length, 0); + await db.query("UPDATE decisao SET decisao='Fila revisada' WHERE id=$1", [decision]); + assert.match((await event(decision)).payload.decisionsAndRationale, /revisada/); + await worker.tick(project); + const inFlight = (await event()).evento_id; + await repository.retry(project, inFlight); + const beforeArchive = (await repository.claim(project))!; + await db.query("UPDATE epico SET status='arquivado' WHERE id=$1", [epic]); + assert.equal((await chunks()).length, 0); + assert.equal((await chunks(decision)).length, 0); + assert.equal((await searchProject(project)).length, 0); + await repository.finish(beforeArchive, { content: "Arquivado", embedding: Array(1024).fill(0.1) }); + assert.equal((await chunks()).length, 0); + await worker.tick(project); await worker.tick(project); + assert.equal((await event()).operacao, "remover"); + + await db.query("UPDATE epico SET status='rascunho' WHERE id=$1", [epic]); + const expired = (await repository.claim(project))!; + await db.query("UPDATE indice_evento SET tentativas=5,bloqueado_ate=CURRENT_TIMESTAMP-INTERVAL '1 second' WHERE evento_id=$1", [expired.evento_id]); + await repository.claim(project); + assert.equal((await event(expired.entidade_id)).erro_codigo, "ATTEMPTS_EXHAUSTED"); + await repository.retry(project, expired.evento_id); + await db.query("UPDATE projeto SET status='arquivado' WHERE id=$1", [project]); + assert.equal((await db.query("SELECT count(*)::int n FROM chunk WHERE projeto_id=$1", [project])).rows[0].n, 0); + } finally { + server.closeAllConnections(); + await new Promise(r => server.close(() => r())); + await db.query("DELETE FROM projeto WHERE id=ANY($1::uuid[])", [[project, other]]); + await db.query("DELETE FROM tecnologia WHERE id=$1", [technology]); + await db.end(); + } +}); + + +test("S2-03 HTTP: autenticação, escopo, observabilidade e nova tentativa", { + skip: !process.env.INDEX_TEST_DATABASE_URL, +}, async () => { + const { pool } = await import("../../database/db.js"); + const { app } = await import("../../app.js"); + const { sessionService } = await import("../auth/session.service.js"); + const [project, other, epic, admin, developer] = Array.from({ length: 5 }, () => randomUUID()); + const api = app.listen(0, "127.0.0.1"); + await new Promise(r => api.once("listening", r)); + const base = "http://127.0.0.1:" + (api.address() as { port: number }).port; + try { + await pool.query("INSERT INTO projeto(id,nome,cliente,status) VALUES($1,'HTTP','Teste','ativo'),($2,'Outro','Teste','ativo')", [project, other]); + await pool.query("INSERT INTO usuario(id,nome,email,senha_hash,role) VALUES($1::uuid,'Admin',$1::text||'@example.test','inutilizado','admin'),($2::uuid,'Dev',$2::text||'@example.test','inutilizado','dev')", [admin, developer]); + const adminSession = await sessionService.createSession(admin); + const devSession = await sessionService.createSession(developer); + const headers = { Authorization: "Bearer " + adminSession.token }; + const path = "/api/v1/projects/" + project + "/index-events"; + assert.equal((await fetch(base + path)).status, 401); + assert.equal((await fetch(base + path, { headers: { Authorization: "Bearer " + devSession.token } })).status, 404); + await pool.query("INSERT INTO epico(id,projeto_id,titulo,descricao,objetivo,escopo_macro,resultado_esperado,status) VALUES($1,$2,'Consultar índice','Descrição','Objetivo','Escopo','Resultado','rascunho')", [epic, project]); + assert.equal((await pool.query("SELECT count(*)::int n FROM indice_evento WHERE projeto_id=$1", [project])).rows[0].n, 0); + const completePath = base + "/api/v1/epics/" + epic + "/complete"; + assert.equal((await fetch(completePath, { method: "PATCH", headers })).status, 400); + assert.equal((await pool.query("SELECT count(*)::int n FROM indice_evento WHERE projeto_id=$1", [project])).rows[0].n, 0); + const criterion = await fetch(base + "/api/v1/criteria", { + method: "POST", headers: { ...headers, "Content-Type": "application/json" }, + body: JSON.stringify({ entidade_tipo: "epico", entidade_id: epic, texto: "Encontrar origem" }), + }); + assert.equal(criterion.status, 201); + assert.equal((await fetch(completePath, { method: "PATCH", headers })).status, 200); + const listing = await fetch(base + path, { headers }); + assert.equal(listing.status, 200); + const body = await listing.json() as { items: { evento_id: string; status: string; payload?: unknown }[] }; + assert.equal(body.items.length, 1); + assert.equal(body.items[0].status, "pendente"); + assert.equal(body.items[0].payload, undefined); + const developerProfile = randomUUID(); + await pool.query("INSERT INTO desenvolvedor(id,usuario_id) VALUES($1,$2)", [developerProfile, developer]); + await pool.query("INSERT INTO alocacao(desenvolvedor_id,projeto_id,papel) VALUES($1,$2,'Teste')", [developerProfile, project]); + const devHeaders = { Authorization: "Bearer " + devSession.token }; + assert.equal((await fetch(base + path, { headers: devHeaders })).status, 200); + assert.equal((await fetch(base + path + "/" + body.items[0].evento_id + "/reprocess", { method: "POST", headers: devHeaders })).status, 403); + const eventId = body.items[0].evento_id; + assert.equal((await fetch(completePath, { method: "PATCH", headers })).status, 200); + assert.equal((await pool.query("SELECT evento_id FROM indice_evento WHERE projeto_id=$1", [project])).rows[0].evento_id, eventId); + assert.equal((await fetch(base + path + "/" + eventId + "/reprocess", { method: "POST", headers })).status, 202); + assert.equal((await fetch(base + "/api/v1/projects/" + other + "/index-events/" + eventId + "/reprocess", { method: "POST", headers })).status, 404); + assert.equal((await fetch(base + path + "/invalido/reprocess", { method: "POST", headers })).status, 400); + const savedDecision = await fetch(base + "/api/v1/epics/" + epic + "/decisions", { + method: "POST", headers: { ...headers, "Content-Type": "application/json" }, + body: JSON.stringify({ titulo: "Usar fila", contexto: "Concorrência", decisao: "Fila persistida", justificativa: "Recuperação" }), + }); + assert.equal(savedDecision.status, 201); + const saved = await savedDecision.json() as { id: string }; + const decisionEvent = (await pool.query("SELECT payload FROM indice_evento WHERE entidade_tipo='decisao' AND entidade_id=$1", [saved.id])).rows[0]; + assert.equal(decisionEvent.payload.project_id, project); + assert.equal(decisionEvent.payload.level, "epico"); + await pool.query("UPDATE projeto SET status='arquivado' WHERE id=$1", [project]); + const current = (await pool.query("SELECT evento_id FROM indice_evento WHERE projeto_id=$1", [project])).rows[0].evento_id; + assert.equal((await fetch(base + path + "/" + current + "/reprocess", { method: "POST", headers })).status, 409); + } finally { + api.closeAllConnections(); + await new Promise(r => api.close(() => r())); + await pool.query("DELETE FROM projeto WHERE id=ANY($1::uuid[])", [[project, other]]); + await pool.query("DELETE FROM usuario WHERE id=ANY($1::uuid[])", [[admin, developer]]); + await pool.end(); + } +}); + +test("S2-03 regressão: excluir ancestral em rascunho retira decisões órfãs do índice", { + skip: !process.env.INDEX_TEST_DATABASE_URL, +}, async () => { + const db = new Pool({ connectionString: validateTarget(process.env.INDEX_TEST_DATABASE_URL, "test") }); + const repository = new IndexingRepository(db); + const [project, epic, feature, decision] = Array.from({ length: 4 }, () => randomUUID()); + try { + await db.query("INSERT INTO projeto(id,nome,cliente,status) VALUES($1,'Regressão','Teste','ativo')", [project]); + await db.query("INSERT INTO epico(id,projeto_id,titulo,status) VALUES($1,$2,'Épico','rascunho')", [epic, project]); + await db.query("INSERT INTO feature(id,epico_id,titulo,status) VALUES($1,$2,'Feature','rascunho')", [feature, epic]); + await db.query("INSERT INTO decisao(id,entidade_tipo,entidade_id,titulo,contexto,decisao,justificativa) VALUES($1,'feature',$2,'Fila','Contexto','Usar fila','Recuperação')", [decision, feature]); + const job = (await repository.claim(project))!; + await repository.finish(job, { content: "Usar fila", embedding: Array(1024).fill(0.1) }); + assert.equal((await db.query("SELECT count(*)::int n FROM chunk WHERE projeto_id=$1", [project])).rows[0].n, 1); + await db.query("DELETE FROM feature WHERE id=$1", [feature]); + assert.equal((await db.query("SELECT count(*)::int n FROM chunk WHERE projeto_id=$1", [project])).rows[0].n, 0); + assert.equal((await db.query("SELECT operacao FROM indice_evento WHERE projeto_id=$1", [project])).rows[0].operacao, "remover"); + } finally { + await db.query("DELETE FROM projeto WHERE id=$1", [project]); + await db.end(); + } +}); + +test("S2-03 regressão: reserva expirada enquanto aguarda lock não grava resposta", { + skip: !process.env.INDEX_TEST_DATABASE_URL, + timeout: 10000, +}, async () => { + const db = new Pool({ + connectionString: validateTarget(process.env.INDEX_TEST_DATABASE_URL, "test"), + application_name: "s203-lease-regression", + }); + const repository = new IndexingRepository(db); + const [project, epic] = [randomUUID(), randomUUID()]; + const blocker = await db.connect(); + let finishing: Promise | undefined; + try { + await db.query("INSERT INTO projeto(id,nome,cliente,status) VALUES($1,'Reserva','Teste','ativo')", [project]); + await db.query("INSERT INTO epico(id,projeto_id,titulo,status) VALUES($1,$2,'Épico','concluido')", [epic, project]); + const event = (await repository.claim(project))!; + await blocker.query("BEGIN"); + await blocker.query("SELECT id FROM projeto WHERE id=$1 FOR UPDATE", [project]); + finishing = repository.finish(event, { content: "Resposta atrasada", embedding: Array(1024).fill(0.1) }); + let waiting = false; + for (let i = 0; i < 50; i++) { + const activity = await db.query("SELECT 1 FROM pg_stat_activity WHERE application_name='s203-lease-regression' AND wait_event_type='Lock'"); + if (activity.rowCount) { waiting = true; break; } + await new Promise(r => setTimeout(r, 10)); + } + assert.ok(waiting, "A conclusão deve estar esperando o lock do projeto."); + await db.query("SELECT pg_sleep(0.05)"); + await db.query("UPDATE indice_evento SET bloqueado_ate=clock_timestamp()-INTERVAL '1 millisecond' WHERE evento_id=$1", [event.evento_id]); + await blocker.query("COMMIT"); + await finishing; + assert.equal((await db.query("SELECT count(*)::int n FROM chunk WHERE projeto_id=$1", [project])).rows[0].n, 0); + assert.equal((await db.query("SELECT status FROM indice_evento WHERE evento_id=$1", [event.evento_id])).rows[0].status, "processando"); + } finally { + await blocker.query("ROLLBACK"); + blocker.release(); + await finishing; + await db.query("DELETE FROM projeto WHERE id=$1", [project]); + await db.end(); + } +}); diff --git a/backend/src/modules/indexing/indexing.integration.db.test.ts b/backend/src/modules/indexing/indexing.integration.db.test.ts new file mode 100644 index 0000000..1b99f77 --- /dev/null +++ b/backend/src/modules/indexing/indexing.integration.db.test.ts @@ -0,0 +1,45 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import { randomUUID } from "node:crypto"; +import { Pool } from "pg"; +import { validateTarget } from "../../database/seed-lib.js"; +import { IndexingRepository } from "./indexing.repository.js"; +import { IndexingWorker } from "./indexing.worker.js"; +import { SearchRepository } from "../search/search.repository.js"; + +test("S2-03/S2-02: serviço IA real, embedding, substituição e arquivamento", { + skip: !process.env.INDEX_TEST_DATABASE_URL || !process.env.INDEX_AI_ENDPOINT, + timeout: 180000, +}, async () => { + const db = new Pool({ connectionString: validateTarget(process.env.INDEX_TEST_DATABASE_URL, "test") }); + const [project, epic] = [randomUUID(), randomUUID()]; + const worker = new IndexingWorker(new IndexingRepository(db), process.env.INDEX_AI_ENDPOINT); + const search = new SearchRepository(db); + try { + await db.query("INSERT INTO projeto(id,nome,cliente,status) VALUES($1,'Integração S2-03','Sintético','ativo')", [project]); + await db.query("INSERT INTO epico(id,projeto_id,titulo,objetivo,status) VALUES($1,$2,'Consultar acervo sintético','Preservar requisitos de integração','concluido')", [epic, project]); + await worker.tick(project); + const state = async () => (await db.query("SELECT status,erro_codigo FROM indice_evento WHERE projeto_id=$1", [project])).rows[0]; + assert.equal((await state()).status, "processado", JSON.stringify(await state())); + const original = (await db.query("SELECT id,texto,embedding::text FROM chunk WHERE projeto_id=$1", [project])).rows[0]; + assert.match(original.texto, /Preservar requisitos/); + const embedding = JSON.parse(original.embedding) as number[]; + assert.equal(embedding.length, 1024); + assert.ok(embedding.some(value => value !== 0)); + assert.equal((await search.hybridSearch({ projectId: project, query: "acervo", limit: 5 }, embedding, 0.5)).length, 1); + await db.query("UPDATE epico SET objetivo='Consultar conhecimento atualizado de integração' WHERE id=$1", [epic]); + await worker.tick(project); + assert.equal((await state()).status, "processado"); + const revised = (await db.query("SELECT id,texto FROM chunk WHERE projeto_id=$1", [project])).rows; + assert.equal(revised.length, 1); + assert.equal(revised[0].id, original.id); + assert.match(revised[0].texto, /conhecimento atualizado/); + assert.doesNotMatch(revised[0].texto, /Preservar requisitos/); + await db.query("UPDATE epico SET status='arquivado' WHERE id=$1", [epic]); + assert.equal((await search.hybridSearch({ projectId: project, query: "acervo", limit: 5 }, embedding, 0.5)).length, 0); + console.log("IA real: vetor 1024d, busca, substituição sem duplicação e retirada imediata ao arquivar."); + } finally { + await db.query("DELETE FROM projeto WHERE id=$1", [project]); + await db.end(); + } +}); diff --git a/backend/src/modules/indexing/indexing.migration.db.test.ts b/backend/src/modules/indexing/indexing.migration.db.test.ts new file mode 100644 index 0000000..67ac27e --- /dev/null +++ b/backend/src/modules/indexing/indexing.migration.db.test.ts @@ -0,0 +1,36 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import { randomUUID } from "node:crypto"; +import { applyUntil, withDatabase } from "../../database/migration-test-utils.js"; +import { validateTarget } from "../../database/seed-lib.js"; + +test("S2-03: migration 019 atualiza banco existente e não republica ao reaplicar", { + skip: !process.env.INDEX_TEST_DATABASE_URL, +}, async () => { + const adminUrl = validateTarget(process.env.INDEX_TEST_DATABASE_URL, "test"); + await withDatabase(adminUrl, async (client, vectorAvailable) => { + assert.ok(vectorAvailable); + await applyUntil(client, true, file => file.startsWith("019_")); + const [project, archivedProject, epic, archivedEpic, nestedEpic, decision, archivedDecision] = Array.from({ length: 7 }, () => randomUUID()); + await client.query("INSERT INTO projeto(id,nome,cliente,status) VALUES($1::uuid,$1::text,'Migração','ativo'),($2::uuid,$2::text,'Migração','arquivado')", [project, archivedProject]); + await client.query("INSERT INTO epico(id,projeto_id,titulo,status) VALUES($1,$4,'Ativo','concluido'),($2,$4,'Arquivado','arquivado'),($3,$5,'Ancestral arquivado','concluido')", [epic, archivedEpic, nestedEpic, project, archivedProject]); + await client.query("INSERT INTO decisao(id,entidade_tipo,entidade_id,titulo,contexto,decisao,justificativa) VALUES($1,'projeto',$3,'Ativa','Contexto','Fila','Recuperação'),($2,'epico',$4,'Arquivada','Contexto','Fila','Recuperação')", [decision, archivedDecision, project, archivedEpic]); + for (const [kind, id, projectId] of [ + ["epico", epic, project], ["epico", epic, project], + ["epico", archivedEpic, project], ["decisao", decision, project], + ["decisao", archivedDecision, project], ["epico", nestedEpic, archivedProject], + ]) { + await client.query("INSERT INTO chunk(projeto_id,entidade_tipo,entidade_id,texto) VALUES($1,$2,$3,'Legado')", [projectId, kind, id]); + } + await applyUntil(client, true, () => false); + const events = (await client.query("SELECT projeto_id,entidade_id,status,evento_id FROM indice_evento ORDER BY entidade_id")).rows; + assert.equal(events.length, 2); + assert.deepEqual(new Set(events.map(row => row.entidade_id)), new Set([epic, decision])); + assert.ok(events.every(row => row.projeto_id === project && row.status === "pendente")); + assert.equal((await client.query("SELECT count(*)::int n FROM chunk")).rows[0].n, 0); + await applyUntil(client, true, () => false); + assert.deepEqual((await client.query("SELECT projeto_id,entidade_id,status,evento_id FROM indice_evento ORDER BY entidade_id")).rows, events); + assert.equal((await client.query("SELECT count(*)::int n FROM decisao")).rows[0].n, 2); + assert.equal((await client.query("SELECT count(*)::int n FROM epico")).rows[0].n, 3); + }); +}); diff --git a/backend/src/modules/indexing/indexing.recovery.db.test.ts b/backend/src/modules/indexing/indexing.recovery.db.test.ts new file mode 100644 index 0000000..fbe5308 --- /dev/null +++ b/backend/src/modules/indexing/indexing.recovery.db.test.ts @@ -0,0 +1,113 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import { randomUUID } from "node:crypto"; +import { Pool } from "pg"; +import { validateTarget } from "../../database/seed-lib.js"; +import { IndexingRepository } from "./indexing.repository.js"; + +async function fixture() { + const db = new Pool({ connectionString: validateTarget(process.env.INDEX_TEST_DATABASE_URL, "test") }); + const repository = new IndexingRepository(db); + const [project, other, epic, otherEpic] = Array.from({ length: 4 }, () => randomUUID()); + await db.query("INSERT INTO projeto(id,nome,cliente,status) VALUES($1::uuid,'Relógio '||$1::text,'Teste','ativo'),($2::uuid,'Outro '||$2::text,'Teste','ativo')", [project, other]); + await db.query("INSERT INTO epico(id,projeto_id,titulo,status) VALUES($1,$2,'Épico','concluido'),($3,$4,'Outro','concluido')", [epic, project, otherEpic, other]); + async function close() { + await db.query("DELETE FROM projeto WHERE id=ANY($1::uuid[])", [[project, other]]); + await db.end(); + } + return { db, repository, project, other, epic, close }; +} + +test("S2-03: diferença no relógio do backend não libera uma reserva ativa", { + skip: !process.env.INDEX_TEST_DATABASE_URL, +}, async () => { + const { db, repository, project, close } = await fixture(); + const now = Date.now; + try { + const event = (await repository.claim(project))!; + Date.now = () => now() + 10 * 60 * 1000; + await assert.rejects(repository.retry(project, event.evento_id), /em processamento/); + assert.equal((await db.query("SELECT status FROM indice_evento WHERE evento_id=$1", [event.evento_id])).rows[0].status, "processando"); + } finally { + Date.now = now; + await close(); + } +}); + +test("S2-03: claim de um projeto não altera reservas expiradas de outro", { + skip: !process.env.INDEX_TEST_DATABASE_URL, +}, async () => { + const { db, repository, project, other, close } = await fixture(); + try { + const event = (await repository.claim(other))!; + await db.query("UPDATE indice_evento SET tentativas=5,bloqueado_ate=clock_timestamp()-INTERVAL '1 second' WHERE evento_id=$1", [event.evento_id]); + await repository.claim(project); + assert.equal((await db.query("SELECT status FROM indice_evento WHERE evento_id=$1", [event.evento_id])).rows[0].status, "processando"); + await repository.claim(other); + assert.equal((await db.query("SELECT erro_codigo FROM indice_evento WHERE evento_id=$1", [event.evento_id])).rows[0].erro_codigo, "ATTEMPTS_EXHAUSTED"); + } finally { await close(); } +}); + +test("S2-03: mover origem entre projetos remove índice antigo e descarta resposta atrasada", { + skip: !process.env.INDEX_TEST_DATABASE_URL, +}, async () => { + const { db, repository, project, other, epic, close } = await fixture(); + const prepared = { content: "Conteúdo sintético", embedding: Array(1024).fill(0.1) }; + try { + const initial = (await repository.claim(project))!; + await repository.finish(initial, prepared); + await repository.retry(project, initial.evento_id); + const stale = (await repository.claim(project))!; + await db.query("UPDATE epico SET projeto_id=$1 WHERE id=$2", [other, epic]); + assert.equal((await db.query("SELECT count(*)::int n FROM chunk WHERE projeto_id=$1", [project])).rows[0].n, 0); + await repository.finish(stale, prepared); + assert.equal((await db.query("SELECT count(*)::int n FROM chunk WHERE projeto_id=$1", [project])).rows[0].n, 0); + await repository.finish((await repository.claim(other))!, prepared); + await repository.finish((await repository.claim(other))!, prepared); + const moved = (await db.query("SELECT projeto_id,metadados_json FROM chunk WHERE entidade_id=$1", [epic])).rows; + assert.equal(moved.length, 1); + assert.equal(moved[0].projeto_id, other); + assert.equal(moved[0].metadados_json.project_id, other); + } finally { await close(); } +}); + + +test("S2-03: workers concorrentes reservam eventos distintos e não duplicam uma origem", { + skip: !process.env.INDEX_TEST_DATABASE_URL, +}, async () => { + const { db, repository, project, close } = await fixture(); + try { + const firstWave = await Promise.all([repository.claim(project), repository.claim(project)]); + assert.equal(firstWave.filter(Boolean).length, 1); + const first = firstWave.find(event => event !== null)!; + await repository.finish(first, { content: "Concorrência", embedding: Array(1024).fill(0.1) }); + const ids = [randomUUID(), randomUUID()]; + await db.query("INSERT INTO epico(id,projeto_id,titulo,status) VALUES($1,$3,'Segundo','concluido'),($2,$3,'Terceiro','concluido')", [...ids, project]); + const secondWave = await Promise.all([repository.claim(project), repository.claim(project)]); + assert.ok(secondWave[0] && secondWave[1]); + assert.notEqual(secondWave[0].evento_id, secondWave[1].evento_id); + await Promise.all(secondWave.map(event => repository.finish(event!, { content: "Concorrência", embedding: Array(1024).fill(0.1) }))); + assert.equal((await db.query("SELECT count(*)::int n FROM chunk WHERE projeto_id=$1", [project])).rows[0].n, 3); + } finally { await close(); } +}); + +test("S2-03: falha na gravação preserva o índice anterior e permite recuperação", { + skip: !process.env.INDEX_TEST_DATABASE_URL, +}, async () => { + const { db, repository, project, close } = await fixture(); + const prepared = { content: "Versão válida", embedding: Array(1024).fill(0.1) }; + try { + const first = (await repository.claim(project))!; + await repository.finish(first, prepared); + const before = (await db.query("SELECT id,texto FROM chunk WHERE projeto_id=$1", [project])).rows; + await repository.retry(project, first.evento_id); + const retry = (await repository.claim(project))!; + await assert.rejects(repository.finish(retry, { content: "Não deve persistir", embedding: [1] })); + await repository.fail(retry, "PROCESSING_ERROR"); + assert.deepEqual((await db.query("SELECT id,texto FROM chunk WHERE projeto_id=$1", [project])).rows, before); + assert.equal((await db.query("SELECT status FROM indice_evento WHERE evento_id=$1", [first.evento_id])).rows[0].status, "falha"); + await repository.retry(project, first.evento_id); + await repository.finish((await repository.claim(project))!, prepared); + assert.deepEqual((await db.query("SELECT id,texto FROM chunk WHERE projeto_id=$1", [project])).rows, before); + } finally { await close(); } +}); diff --git a/backend/src/modules/indexing/indexing.repository.ts b/backend/src/modules/indexing/indexing.repository.ts new file mode 100644 index 0000000..1fab1b9 --- /dev/null +++ b/backend/src/modules/indexing/indexing.repository.ts @@ -0,0 +1,143 @@ +import { randomUUID } from "node:crypto"; +import { Pool } from "pg"; +import { pool } from "../../database/db.js"; +import { withTransaction } from "../../database/transaction.js"; +import { ConflictError, NotFoundError } from "../../shared/errors.js"; +import { assertWritable, lockHierarchy } from "../projects/hierarchy-archive.js"; + +export interface IndexEvent { + projeto_id: string; + entidade_tipo: "epico" | "feature" | "pbi" | "decisao"; + entidade_id: string; + evento_id: string; + versao: number; + operacao: "indexar" | "remover"; + payload: Record; + lease_id: string; +} + +export interface PreparedEntity { + content: string; + embedding: number[]; +} + +export class IndexingRepository { + constructor(private readonly db: Pool = pool) {} + + async claim(projectId?: string): Promise { + await this.db.query( + `UPDATE indice_evento SET status='falha',erro_codigo='ATTEMPTS_EXHAUSTED', + lease_id=NULL,bloqueado_ate=NULL,updated_at=CURRENT_TIMESTAMP + WHERE status='processando' AND bloqueado_ate<=CURRENT_TIMESTAMP AND tentativas>=5 + AND ($1::uuid IS NULL OR projeto_id=$1)`, + [projectId ?? null], + ); + const result = await this.db.query( + `WITH picked AS ( + SELECT projeto_id,entidade_tipo,entidade_id FROM indice_evento + WHERE status<>'processado' AND tentativas<5 AND proxima_tentativa<=CURRENT_TIMESTAMP + AND (bloqueado_ate IS NULL OR bloqueado_ate<=CURRENT_TIMESTAMP) + AND ($2::uuid IS NULL OR projeto_id=$2) + ORDER BY proxima_tentativa,evento_id LIMIT 1 FOR UPDATE SKIP LOCKED + ) + UPDATE indice_evento ev SET status='processando',tentativas=tentativas+1, + lease_id=$1,bloqueado_ate=CURRENT_TIMESTAMP+INTERVAL '3 minutes',updated_at=CURRENT_TIMESTAMP + FROM picked WHERE ev.projeto_id=picked.projeto_id + AND ev.entidade_tipo=picked.entidade_tipo AND ev.entidade_id=picked.entidade_id + RETURNING ev.*`, + [randomUUID(), projectId ?? null], + ); + return result.rows[0] ?? null; + } + + async finish(event: IndexEvent, prepared?: PreparedEntity): Promise { + await withTransaction(this.db, async client => { + await lockHierarchy(client, "projeto", event.projeto_id); + const current = await client.query( + `SELECT 1 FROM indice_evento WHERE projeto_id=$1 AND entidade_tipo=$2 AND entidade_id=$3 + AND evento_id=$4 AND lease_id=$5 AND status='processando' + AND bloqueado_ate>clock_timestamp() FOR UPDATE`, + [event.projeto_id, event.entidade_tipo, event.entidade_id, event.evento_id, event.lease_id], + ); + if (!current.rowCount) return; + + const source = await client.query( + `SELECT elegivel,payload FROM indice_origem + WHERE projeto_id=$1 AND entidade_tipo=$2 AND entidade_id=$3`, + [event.projeto_id, event.entidade_tipo, event.entidade_id], + ); + if (event.operacao === "indexar" && source.rows[0]?.elegivel) { + // Compara o snapshot mesmo se um escritor externo não publicou o evento. + if (JSON.stringify(source.rows[0].payload) !== JSON.stringify(event.payload)) { + await client.query("SELECT sincronizar_indice_projeto($1)", [event.projeto_id]); + return; + } + if (!prepared) throw new Error("Conteúdo preparado ausente."); + await client.query( + "DELETE FROM chunk WHERE projeto_id=$1 AND entidade_tipo=$2 AND entidade_id=$3", + [event.projeto_id, event.entidade_tipo, event.entidade_id], + ); + await client.query( + `INSERT INTO chunk(id,projeto_id,entidade_tipo,entidade_id,texto,embedding,metadados_json) + VALUES(md5($1::uuid::text||':'||$2::text||':'||$3::uuid::text)::uuid,$1::uuid,$2::text,$3::uuid,$4,$5::vector,$6::jsonb)`, + [event.projeto_id, event.entidade_tipo, event.entidade_id, prepared.content, + JSON.stringify(prepared.embedding), JSON.stringify({ + ...event.payload, event_id: event.evento_id, version: event.versao, chunk_index: 0, + })], + ); + } else { + await client.query( + "DELETE FROM chunk WHERE projeto_id=$1 AND entidade_tipo=$2 AND entidade_id=$3", + [event.projeto_id, event.entidade_tipo, event.entidade_id], + ); + } + await client.query( + `UPDATE indice_evento SET status='processado',lease_id=NULL,bloqueado_ate=NULL, + erro_codigo=NULL,updated_at=CURRENT_TIMESTAMP WHERE evento_id=$1 AND lease_id=$2`, + [event.evento_id, event.lease_id], + ); + }); + } + + async fail(event: IndexEvent, code: string): Promise { + await this.db.query( + `UPDATE indice_evento SET status='falha',erro_codigo=$3,lease_id=NULL,bloqueado_ate=NULL, + proxima_tentativa=CURRENT_TIMESTAMP+make_interval(secs=>LEAST(3600,15*power(2,tentativas))::int), + updated_at=CURRENT_TIMESTAMP WHERE evento_id=$1 AND lease_id=$2 AND projeto_id=$4`, + [event.evento_id, event.lease_id, code, event.projeto_id], + ); + } + + async list(projectId: string) { + const result = await this.db.query( + `SELECT evento_id,projeto_id,entidade_tipo,entidade_id,versao,operacao,status, + tentativas,proxima_tentativa,erro_codigo,updated_at FROM indice_evento + WHERE projeto_id=$1 ORDER BY updated_at DESC,evento_id LIMIT 100`, + [projectId], + ); + return { items: result.rows }; + } + + async retry(projectId: string, eventId: string): Promise { + await withTransaction(this.db, async client => { + await lockHierarchy(client, "projeto", projectId); + await assertWritable(client, "projeto", projectId); + const result = await client.query( + "SELECT status,(bloqueado_ate>clock_timestamp()) AS reserva_ativa FROM indice_evento WHERE projeto_id=$1 AND evento_id=$2 FOR UPDATE", + [projectId, eventId], + ); + const event = result.rows[0]; + if (!event) throw new NotFoundError("Evento não encontrado."); + if (event.status === "pendente") return; + if (event.reserva_ativa) { + throw new ConflictError("Evento em processamento."); + } + await client.query( + `UPDATE indice_evento SET status='pendente',tentativas=0,lease_id=NULL,bloqueado_ate=NULL, + erro_codigo=NULL,proxima_tentativa=CURRENT_TIMESTAMP,updated_at=CURRENT_TIMESTAMP + WHERE projeto_id=$1 AND evento_id=$2`, + [projectId, eventId], + ); + }); + } +} diff --git a/backend/src/modules/indexing/indexing.routes.ts b/backend/src/modules/indexing/indexing.routes.ts new file mode 100644 index 0000000..5bd9290 --- /dev/null +++ b/backend/src/modules/indexing/indexing.routes.ts @@ -0,0 +1,30 @@ +import { Router } from "express"; +import { requireAuth } from "../../middleware/requireAuth.js"; +import { requireRole } from "../../middleware/requireRole.js"; +import { validateUuid } from "../../shared/errors.js"; +import { IndexingRepository } from "./indexing.repository.js"; + +export function createIndexingRouter(repository = new IndexingRepository()) { + const router = Router({ mergeParams: true }); + router.use(requireAuth); + router.get("/", async (req, res, next) => { + try { + const projectId = String((req.params as Record).projectId); + validateUuid(projectId, "Projeto"); + res.json(await repository.list(projectId)); + } catch (error) { next(error); } + }); + router.post("/:eventId/reprocess", requireRole("admin", "po"), async (req, res, next) => { + try { + const projectId = String(req.params.projectId); + const eventId = String(req.params.eventId); + validateUuid(projectId, "Projeto"); + validateUuid(eventId, "Evento"); + await repository.retry(projectId, eventId); + res.status(202).json({ status: "pendente" }); + } catch (error) { next(error); } + }); + return router; +} + +export const indexingRouter = createIndexingRouter(); diff --git a/backend/src/modules/indexing/indexing.state.db.test.ts b/backend/src/modules/indexing/indexing.state.db.test.ts new file mode 100644 index 0000000..4d29641 --- /dev/null +++ b/backend/src/modules/indexing/indexing.state.db.test.ts @@ -0,0 +1,222 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import { randomUUID } from "node:crypto"; +import { Pool } from "pg"; +import { validateTarget } from "../../database/seed-lib.js"; +import { IndexingRepository, type IndexEvent } from "./indexing.repository.js"; + +type Kind = "projeto" | "epico" | "feature" | "pbi" | "decisao"; +interface Source { id: string; kind: Kind; parent?: string; status: string; title: string } +function random(seed: number) { + let state = seed; + return (max: number) => { state = (Math.imul(state, 1664525) + 1013904223) >>> 0; return Math.floor(state / 0x100000000 * max); }; +} + +test("S2-03: sequências combinadas mantêm o índice equivalente às origens", { + skip: !process.env.INDEX_TEST_DATABASE_URL, + timeout: 180000, +}, async t => { + const firstSeed = Number(process.env.INDEX_FIRST_SEED ?? 1); + for (let seed = firstSeed; seed < firstSeed + 8; seed++) { + await t.test("sequência " + seed + ": 80 operações", async () => { + const db = new Pool({ connectionString: validateTarget(process.env.INDEX_TEST_DATABASE_URL, "test") }); + const repository = new IndexingRepository(db); + const sources = new Map(); + const projects: string[] = []; + const flights: IndexEvent[] = []; + const pick = random(seed); + const vector = Array(1024).fill(0.1); + const technology = randomUUID(); + const links = new Set(); + const criteria = new Map(); + let technologyName = "Tecnologia " + technology; + function all(kind?: Kind) { return [...sources.values()].filter(s => !kind || s.kind === kind); } + function choose(values: T[]) { return values[pick(values.length)]; } + function ancestry(source: Source): Source[] { + const chain = [source]; + let cursor = source; + while (cursor.parent) { + const parent = sources.get(cursor.parent); + if (!parent) return []; + chain.push(parent); + cursor = parent; + } + return cursor.kind === "projeto" ? chain : []; + } + function scope(source: Source) { + const chain = ancestry(source); + const active = chain.length > 0 && chain.every(s => s.status !== "arquivado"); + return { + project: chain.at(-1)?.id, + eligible: active && (source.kind === "decisao" || source.status === "concluido"), + level: source.kind === "decisao" ? sources.get(source.parent!)?.kind : source.kind, + }; + } + function prepare(event: IndexEvent) { + return { content: String(event.payload.title) + " " + String(event.payload.description), embedding: vector }; + } + async function claimLocal() { + for (const project of projects) { + const event = await repository.claim(project); + if (event) return event; + } + return null; + } + async function settle() { + for (const flight of flights.splice(0)) await repository.finish(flight, prepare(flight)); + // Simula recuperação após interrupção sem alterar a versão da origem. + await db.query("UPDATE indice_evento SET bloqueado_ate=clock_timestamp()-INTERVAL '1 second' WHERE projeto_id=ANY($1::uuid[]) AND status='processando'", [projects]); + await db.query("UPDATE indice_evento SET proxima_tentativa=clock_timestamp(),tentativas=0 WHERE projeto_id=ANY($1::uuid[]) AND status='falha'", [projects]); + for (let i = 0; i < 100; i++) { + const event = await claimLocal(); + if (!event) return; + await repository.finish(event, prepare(event)); + } + assert.fail("A fila não esvaziou."); + } + async function invariant(drained = false) { + const rows = (await db.query("SELECT entidade_id,entidade_tipo,projeto_id,metadados_json,vector_dims(embedding) AS dimensions FROM chunk WHERE projeto_id=ANY($1::uuid[])", [projects])).rows; + const seen = new Set(); + for (const row of rows) { + const source = sources.get(row.entidade_id); + assert.ok(source, "Origem excluída não pode continuar indexada."); + const expected = scope(source); + assert.ok(expected.eligible, "Origem em rascunho/arquivada não pode estar indexada."); + assert.equal(row.projeto_id, expected.project); + assert.equal(row.metadados_json.project_id, expected.project); + assert.equal(row.metadados_json.level, expected.level); + assert.equal(row.metadados_json.title, source.title); + assert.equal(row.metadados_json.status, source.kind === "decisao" ? sources.get(source.parent!)!.status : source.status); + assert.equal(row.metadados_json.entity_id, source.id); + assert.deepEqual(row.metadados_json.ancestor_ids, ancestry(source).filter(s => s.kind !== "decisao").map(s => s.id).reverse()); + const technologySource = source.kind === "decisao" ? source.parent! : source.id; + assert.deepEqual(row.metadados_json.acceptanceCriteria, criteria.get(technologySource) ?? []); + assert.deepEqual(row.metadados_json.tecnologias_ids, links.has(technologySource) ? [technology] : []); + if (links.has(technologySource)) assert.equal(row.metadados_json.technologies[0].nome, technologyName); + assert.equal(row.dimensions, 1024); + const key = row.projeto_id + ":" + row.entidade_tipo + ":" + row.entidade_id; + assert.ok(!seen.has(key), "Origem duplicada."); + seen.add(key); + } + if (drained) { + const expected = all().filter(s => s.kind !== "projeto" && scope(s).eligible).length; + assert.equal(rows.length, expected, "Toda origem elegível deve estar indexada após esvaziar a fila."); + } + } + async function create(kind: Kind, parent?: string) { + const id = randomUUID(), title = "Sintético " + id; + const source: Source = { id, kind, parent, title, status: kind === "projeto" ? "ativo" : "concluido" }; + if (kind === "projeto") { + projects.push(id); + await db.query("INSERT INTO projeto(id,nome,cliente,status) VALUES($1,$2,'Estado','ativo')", [id, title]); + } else if (kind === "epico") { + await db.query("INSERT INTO epico(id,projeto_id,titulo,status) VALUES($1,$2,$3,'concluido')", [id, parent, title]); + } else if (kind === "feature") { + await db.query("INSERT INTO feature(id,epico_id,titulo,status) VALUES($1,$2,$3,'concluido')", [id, parent, title]); + } else if (kind === "pbi") { + await db.query("INSERT INTO pbi(id,feature_id,codigo,titulo,historia_como_um,historia_eu_quero,historia_para_que,status) VALUES($1::uuid,$2,$1::text,$3,'PO','consultar','reutilizar','concluido')", [id, parent, title]); + } else { + await db.query("INSERT INTO decisao(id,entidade_tipo,entidade_id,titulo,contexto,decisao,justificativa) VALUES($1,$2,$3,$4,'Contexto','Usar fila','Recuperação')", [id, sources.get(parent!)!.kind, parent, title]); + } + sources.set(id, source); + return source; + } + try { + for (let i = 0; i < 2; i++) { + const project = await create("projeto"); + const epic = await create("epico", project.id); + const feature = await create("feature", epic.id); + const pbi = await create("pbi", feature.id); + await create("decisao", project.id); + await create("decisao", feature.id); + await create("decisao", pbi.id); + } + await db.query("INSERT INTO tecnologia(id,nome) VALUES($1,$2)", [technology, technologyName]); + await settle(); await invariant(true); + for (let step = 0; step < 80; step++) { + const operation = pick(14); + const source = choose(all()); + if (operation === 0 && source.kind !== "decisao") { + source.status = choose(source.kind === "projeto" ? ["ativo", "arquivado"] : ["rascunho", "concluido", "arquivado"]); + await db.query("UPDATE " + source.kind + " SET status=$1 WHERE id=$2", [source.status, source.id]); + } else if (operation === 1 && source.kind !== "projeto") { + source.title = "Revisão " + seed + "-" + step; + await db.query("UPDATE " + source.kind + " SET titulo=$1 WHERE id=$2", [source.title, source.id]); + } else if (operation === 2 && source.kind !== "projeto") { + if (source.kind === "decisao") { + const parent = choose(all().filter(s => s.kind !== "decisao")); + await db.query("UPDATE decisao SET entidade_tipo=$1,entidade_id=$2 WHERE id=$3", [parent.kind, parent.id, source.id]); + source.parent = parent.id; + } else { + const parentKind = { epico: "projeto", feature: "epico", pbi: "feature" }[source.kind] as Kind; + const column = { epico: "projeto_id", feature: "epico_id", pbi: "feature_id" }[source.kind]; + const parent = choose(all(parentKind)); + await db.query("UPDATE " + source.kind + " SET " + column + "=$1 WHERE id=$2", [parent.id, source.id]); + source.parent = parent.id; + } + } else if (operation === 3) { + const client = await db.connect(); + try { + await client.query("BEGIN"); + await client.query("UPDATE " + source.kind + " SET " + (source.kind === "projeto" ? "nome" : "titulo") + "='Rollback' WHERE id=$1", [source.id]); + await client.query("ROLLBACK"); + } finally { client.release(); } + } else if (operation === 4) { + const event = await claimLocal(); + if (event) flights.push(event); + } else if (operation === 5 && flights.length) { + const flight = flights.splice(pick(flights.length), 1)[0]; + await repository.finish(flight, prepare(flight)); + await repository.finish(flight, prepare(flight)); + } else if (operation === 6 && flights.length) { + const flight = flights.splice(pick(flights.length), 1)[0]; + await repository.fail(flight, "SERVICE_UNAVAILABLE"); + } else if (operation === 7 && source.kind !== "decisao") { + if (links.has(source.id)) { + await db.query("DELETE FROM entidade_tecnologia WHERE entidade_tipo=$1 AND entidade_id=$2", [source.kind, source.id]); + links.delete(source.id); + } else { + await db.query("INSERT INTO entidade_tecnologia(entidade_tipo,entidade_id,tecnologia_id) VALUES($1,$2,$3)", [source.kind, source.id, technology]); + links.add(source.id); + } + } else if (operation === 8) { + technologyName = "Stack " + technology + "-" + step; + await db.query("UPDATE tecnologia SET nome=$1 WHERE id=$2", [technologyName, technology]); + } else if (operation === 9) { + const decision = choose(all("decisao")); + if (decision) { + await db.query("DELETE FROM decisao WHERE id=$1", [decision.id]); + sources.delete(decision.id); + } else await create("decisao", choose(all("pbi")).id); + } else if (operation === 10 && flights.length) { + const flight = choose(flights); + await db.query("UPDATE indice_evento SET bloqueado_ate=clock_timestamp()-INTERVAL '1 second' WHERE evento_id=$1 AND lease_id=$2", [flight.evento_id, flight.lease_id]); + } else if (operation === 12) { + const pbi = choose(all("pbi")); + if (pbi) { + await db.query("INSERT INTO criterio_aceitacao(entidade_tipo,entidade_id,nome,dado,quando,entao,ordem) VALUES('pbi',$1,$2,'Acervo','Consultar','Encontrar origem',$3)", [pbi.id, "Cenário " + seed + "-" + step, step + 1]); + criteria.set(pbi.id, [...(criteria.get(pbi.id) ?? []), "Cenário " + seed + "-" + step + " Acervo Consultar Encontrar origem"]); + } + } else if (operation === 13) { + await db.query("DELETE FROM tecnologia WHERE id=$1", [technology]); + links.clear(); + await db.query("INSERT INTO tecnologia(id,nome) VALUES($1,$2)", [technology, technologyName]); + } else if (operation === 11) { + const candidate = (await db.query("SELECT projeto_id,evento_id FROM indice_evento WHERE projeto_id=ANY($1::uuid[]) AND status='processado' ORDER BY evento_id LIMIT 1", [projects])).rows[0]; + if (candidate && sources.get(candidate.projeto_id)?.status !== "arquivado") { + await repository.retry(candidate.projeto_id, candidate.evento_id); + await repository.retry(candidate.projeto_id, candidate.evento_id); + } + } + await invariant(); + if (step % 10 === 9) { await settle(); await invariant(true); } + } + await settle(); await invariant(true); + } finally { + await db.query("DELETE FROM projeto WHERE id=ANY($1::uuid[])", [projects]); + await db.query("DELETE FROM tecnologia WHERE id=$1", [technology]); + await db.end(); + } + }); + } +}); diff --git a/backend/src/modules/indexing/indexing.transport.db.test.ts b/backend/src/modules/indexing/indexing.transport.db.test.ts new file mode 100644 index 0000000..a87a9e5 --- /dev/null +++ b/backend/src/modules/indexing/indexing.transport.db.test.ts @@ -0,0 +1,64 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import { randomUUID } from "node:crypto"; +import { createServer } from "node:http"; +import { Pool } from "pg"; +import { validateTarget } from "../../database/seed-lib.js"; +import { IndexingRepository } from "./indexing.repository.js"; +import { IndexingWorker } from "./indexing.worker.js"; + +test("S2-03: respostas inválidas e erros HTTP não gravam trechos e permitem recuperação", { + skip: !process.env.INDEX_TEST_DATABASE_URL, +}, async t => { + const db = new Pool({ connectionString: validateTarget(process.env.INDEX_TEST_DATABASE_URL, "test") }); + const repository = new IndexingRepository(db); + const [project, epic] = [randomUUID(), randomUUID()]; + let mode = "ok"; + const server = createServer((_req, res) => { + res.setHeader("Content-Type", "application/json"); + if (mode.startsWith("http-")) { + res.statusCode = Number(mode.slice(5)); + res.end(JSON.stringify({ message: "Não deve ser persistido: segredo" })); + return; + } + const response = { + entity_type: mode === "wrong-type" ? "pbi" : "epico", + content: mode === "empty" ? "" : "Conteúdo sintético", + metadata: { project_id: mode === "wrong-project" ? randomUUID() : project }, + embedding: mode === "short-vector" ? [1] : Array(1024).fill(mode === "zero-vector" ? 0 : 0.1), + }; + res.end(mode === "invalid-json" ? "{" : JSON.stringify(response)); + }); + await new Promise(r => server.listen(0, "127.0.0.1", r)); + const worker = new IndexingWorker(repository, "http://127.0.0.1:" + (server.address() as { port: number }).port); + const state = async () => (await db.query("SELECT status,erro_codigo,evento_id FROM indice_evento WHERE projeto_id=$1", [project])).rows[0]; + try { + await db.query("INSERT INTO projeto(id,nome,cliente,status) VALUES($1,$1::uuid::text,'Transporte','ativo')", [project]); + await db.query("INSERT INTO epico(id,projeto_id,titulo,status) VALUES($1,$2,'Consultar acervo','concluido')", [epic, project]); + for (const scenario of ["empty", "short-vector", "zero-vector", "invalid-json", "wrong-type", "wrong-project", "http-401", "http-429", "http-503"]) { + await t.test(scenario, async () => { + mode = scenario; + await repository.retry(project, (await state()).evento_id); + await worker.tick(project); + const failed = await state(); + assert.equal(failed.status, "falha"); + const expected = scenario.startsWith("http-") ? "SERVICE_HTTP_ERROR" + : scenario.startsWith("wrong-") ? "INVALID_SCOPE" : "INVALID_RESPONSE"; + assert.equal(failed.erro_codigo, expected); + assert.equal((await db.query("SELECT count(*)::int n FROM chunk WHERE projeto_id=$1", [project])).rows[0].n, 0); + assert.doesNotMatch(JSON.stringify(await repository.list(project)), /segredo/); + }); + } + mode = "ok"; + await repository.retry(project, (await state()).evento_id); + await worker.tick(project); + assert.equal((await state()).status, "processado"); + assert.equal((await state()).erro_codigo, null); + assert.equal((await db.query("SELECT count(*)::int n FROM chunk WHERE projeto_id=$1", [project])).rows[0].n, 1); + } finally { + server.closeAllConnections(); + await new Promise(r => server.close(() => r())); + await db.query("DELETE FROM projeto WHERE id=$1", [project]); + await db.end(); + } +}); diff --git a/backend/src/modules/indexing/indexing.worker.test.ts b/backend/src/modules/indexing/indexing.worker.test.ts new file mode 100644 index 0000000..7fc512b --- /dev/null +++ b/backend/src/modules/indexing/indexing.worker.test.ts @@ -0,0 +1,22 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import { randomUUID } from "node:crypto"; +import { indexingFailureCode, validatePreparedEntity } from "./indexing.worker.js"; + +test("valida vetor, conteúdo e escopo retornados pela IA", () => { + const project = randomUUID(); + const data = { entity_type: "pbi", content: "Requisito", metadata: { project_id: project }, embedding: Array(1024).fill(0.1) }; + assert.equal(validatePreparedEntity(data, "pbi", project).content, "Requisito"); + for (const invalid of [ + { ...data, content: "" }, { ...data, embedding: [1] }, + { ...data, embedding: Array(1024).fill(0) }, { ...data, embedding: Array(1024).fill(Infinity) }, + { ...data, metadata: { project_id: randomUUID() } }, { ...data, entity_type: "decisao" }, + ]) assert.throws(() => validatePreparedEntity(invalid, "pbi", project)); +}); + +test("diagnósticos não expõem mensagens, corpo HTTP ou credenciais", () => { + assert.equal(indexingFailureCode({ isAxiosError: true, code: "ECONNABORTED" }), "TIMEOUT"); + assert.equal(indexingFailureCode({ isAxiosError: true, response: { status: 503, data: "segredo" } }), "SERVICE_HTTP_ERROR"); + assert.equal(indexingFailureCode(new Error("token secreto")), "PROCESSING_ERROR"); + assert.equal(indexingFailureCode(new Error("INVALID_SCOPE")), "INVALID_SCOPE"); +}); diff --git a/backend/src/modules/indexing/indexing.worker.ts b/backend/src/modules/indexing/indexing.worker.ts new file mode 100644 index 0000000..db0a3d7 --- /dev/null +++ b/backend/src/modules/indexing/indexing.worker.ts @@ -0,0 +1,75 @@ +import axios from "axios"; +import { z } from "zod"; +import { env } from "../../config/env.js"; +import { serviceHeaders } from "../../shared/service-auth.js"; +import { IndexingRepository, type PreparedEntity } from "./indexing.repository.js"; + +const responseSchema = z.object({ + entity_type: z.string(), + content: z.string().trim().min(1).max(500000), + metadata: z.object({ project_id: z.string().uuid() }), + embedding: z.array(z.number().finite()).length(1024).refine(vector => vector.some(value => value !== 0)), +}); + +export function validatePreparedEntity(data: unknown, entityType: string, projectId: string): PreparedEntity { + const result = responseSchema.parse(data); + if (result.entity_type !== entityType || result.metadata.project_id !== projectId) { + throw new Error("INVALID_SCOPE"); + } + return { content: result.content, embedding: result.embedding }; +} + +export function indexingFailureCode(error: unknown): string { + if (error instanceof z.ZodError) return "INVALID_RESPONSE"; + if (error instanceof Error && error.message === "INVALID_SCOPE") return "INVALID_SCOPE"; + if (axios.isAxiosError(error)) { + if (error.code === "ECONNABORTED" || error.code === "ETIMEDOUT") return "TIMEOUT"; + return error.response ? "SERVICE_HTTP_ERROR" : "SERVICE_UNAVAILABLE"; + } + return "PROCESSING_ERROR"; +} + +export class IndexingWorker { + constructor( + private readonly repository = new IndexingRepository(), + private readonly endpoint = `${env.AI_SERVICE_URL}/ingest/entity`, + ) {} + + async tick(projectId?: string): Promise { + const event = await this.repository.claim(projectId); + if (!event) return; + try { + if (event.operacao === "remover") { + await this.repository.finish(event); + return; + } + const { data } = await axios.post(this.endpoint, { + entity_type: event.entidade_tipo, + data: event.payload, + }, { headers: serviceHeaders(), timeout: 120000, maxContentLength: 8 * 1024 * 1024 }); + const prepared = validatePreparedEntity(data, event.entidade_tipo, event.projeto_id); + await this.repository.finish(event, prepared); + } catch (error) { + await this.repository.fail(event, indexingFailureCode(error)); + } + } +} + +export function startIndexingWorker(worker = new IndexingWorker()): () => void { + let running = false; + async function tick() { + if (running) return; + running = true; + try { + await worker.tick(); + } catch { + console.warn("[Indexing] Não foi possível processar a fila."); + } finally { + running = false; + } + } + const timer = setInterval(() => void tick(), 15000); + timer.unref(); + void tick(); + return () => clearInterval(timer); +} diff --git a/database/migrations/019_entity_index_events.sql b/database/migrations/019_entity_index_events.sql new file mode 100644 index 0000000..520709b --- /dev/null +++ b/database/migrations/019_entity_index_events.sql @@ -0,0 +1,169 @@ +-- S2-03: outbox transacional por origem, com versão e reserva de processamento. +CREATE TABLE indice_evento ( + projeto_id UUID NOT NULL REFERENCES projeto(id) ON DELETE CASCADE, + entidade_tipo TEXT NOT NULL CHECK (entidade_tipo IN ('epico','feature','pbi','decisao')), + entidade_id UUID NOT NULL, + evento_id UUID NOT NULL DEFAULT uuid_generate_v4(), + versao INTEGER NOT NULL DEFAULT 1, + operacao TEXT NOT NULL CHECK (operacao IN ('indexar','remover')), + payload JSONB NOT NULL, + status TEXT NOT NULL DEFAULT 'pendente' CHECK (status IN ('pendente','processando','processado','falha')), + tentativas INTEGER NOT NULL DEFAULT 0, + proxima_tentativa TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + lease_id UUID, + bloqueado_ate TIMESTAMPTZ, + erro_codigo TEXT, + updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (projeto_id, entidade_tipo, entidade_id), + UNIQUE (evento_id) +); +CREATE INDEX idx_indice_evento_fila ON indice_evento(proxima_tentativa) + WHERE status <> 'processado'; + +-- Inclui origens não elegíveis para que arquivamento e retorno a rascunho gerem remoção. +CREATE VIEW indice_origem AS +WITH itens AS ( + SELECT p.id AS projeto_id, 'projeto'::text AS tipo, p.id, + ARRAY[p.id] AS ancestrais, p.status <> 'arquivado' AS ativo, + p.status, to_jsonb(p) AS dados + FROM projeto p + UNION ALL + SELECT e.projeto_id, 'epico', e.id, ARRAY[e.projeto_id,e.id], + p.status <> 'arquivado' AND e.status <> 'arquivado', e.status, to_jsonb(e) + FROM epico e JOIN projeto p ON p.id=e.projeto_id + UNION ALL + SELECT e.projeto_id, 'feature', f.id, ARRAY[e.projeto_id,e.id,f.id], + p.status <> 'arquivado' AND e.status <> 'arquivado' AND f.status <> 'arquivado', f.status, to_jsonb(f) + FROM feature f JOIN epico e ON e.id=f.epico_id JOIN projeto p ON p.id=e.projeto_id + UNION ALL + SELECT e.projeto_id, 'pbi', b.id, ARRAY[e.projeto_id,e.id,f.id,b.id], + p.status <> 'arquivado' AND e.status <> 'arquivado' AND f.status <> 'arquivado' AND b.status <> 'arquivado', + b.status, to_jsonb(b) + FROM pbi b JOIN feature f ON f.id=b.feature_id JOIN epico e ON e.id=f.epico_id + JOIN projeto p ON p.id=e.projeto_id +), origens AS ( + SELECT i.projeto_id, i.tipo, i.id, i.ancestrais, i.ativo AND i.status='concluido' AS elegivel, + i.tipo AS nivel, i.id AS origem_id, i.status, i.dados + FROM itens i WHERE i.tipo <> 'projeto' + UNION ALL + SELECT i.projeto_id, 'decisao', d.id, i.ancestrais, i.ativo, + i.tipo, i.id, i.status, to_jsonb(d) + FROM decisao d JOIN itens i ON i.tipo=d.entidade_tipo AND i.id=d.entidade_id +) +SELECT o.projeto_id, o.tipo AS entidade_tipo, o.id AS entidade_id, o.ancestrais, o.elegivel, + jsonb_build_object( + 'project_id',o.projeto_id,'entity_id',o.id,'entity_type',o.tipo, + 'level',o.nivel,'source_id',o.origem_id,'status',o.status, + 'ancestor_ids',to_jsonb(o.ancestrais), + 'title',COALESCE(o.dados->>'titulo',''),'source_locator',o.dados->>'codigo', + 'description',concat_ws(E'\n\n', + o.dados->>'descricao',o.dados->>'objetivo',o.dados->>'escopo_macro', + o.dados->>'resultado_esperado',o.dados->>'historia_como_um', + o.dados->>'historia_eu_quero',o.dados->>'historia_para_que', + o.dados->>'regras_observacoes',o.dados->>'contexto'), + 'decisionsAndRationale',concat_ws(E'\n\n',o.dados->>'decisao',o.dados->>'justificativa',o.dados->>'alternativas'), + 'provenance',COALESCE(o.dados->>'provenance','human-authored'), + 'technologies',COALESCE(( + SELECT jsonb_agg(jsonb_build_object('id',t.id,'nome',t.nome) ORDER BY t.id) + FROM entidade_tecnologia et JOIN tecnologia t ON t.id=et.tecnologia_id + WHERE et.entidade_tipo=o.nivel AND et.entidade_id=o.origem_id + ),'[]'::jsonb), + 'tecnologias_ids',COALESCE(( + SELECT jsonb_agg(et.tecnologia_id ORDER BY et.tecnologia_id) + FROM entidade_tecnologia et WHERE et.entidade_tipo=o.nivel AND et.entidade_id=o.origem_id + ),'[]'::jsonb), + 'acceptanceCriteria',COALESCE(( + SELECT jsonb_agg(concat_ws(' ',c.nome,c.texto,c.dado,c.quando,c.entao) ORDER BY c.ordem,c.id) + FROM criterio_aceitacao c WHERE c.entidade_tipo=o.nivel AND c.entidade_id=o.origem_id + ),'[]'::jsonb) + ) AS payload +FROM origens o; + +CREATE FUNCTION sincronizar_indice_projeto(project UUID) RETURNS VOID LANGUAGE plpgsql AS $$ +BEGIN + -- Mesma ordem de locks dos escritores da hierarquia e do consumidor. + PERFORM id FROM projeto WHERE id=project FOR UPDATE; + INSERT INTO indice_evento(projeto_id,entidade_tipo,entidade_id,operacao,payload) + SELECT projeto_id,entidade_tipo,entidade_id, + CASE WHEN elegivel THEN 'indexar' ELSE 'remover' END,payload + FROM indice_origem WHERE projeto_id=project + AND (elegivel OR EXISTS ( + SELECT 1 FROM indice_evento ev WHERE ev.projeto_id=project + AND ev.entidade_tipo=indice_origem.entidade_tipo AND ev.entidade_id=indice_origem.entidade_id)) + ON CONFLICT (projeto_id,entidade_tipo,entidade_id) DO UPDATE SET + evento_id=uuid_generate_v4(),versao=indice_evento.versao+1, + operacao=EXCLUDED.operacao,payload=EXCLUDED.payload, + status='pendente',tentativas=0,proxima_tentativa=CURRENT_TIMESTAMP, + lease_id=NULL,bloqueado_ate=NULL,erro_codigo=NULL,updated_at=CURRENT_TIMESTAMP + WHERE indice_evento.payload IS DISTINCT FROM EXCLUDED.payload OR indice_evento.operacao<>EXCLUDED.operacao; + + UPDATE indice_evento ev SET operacao='remover',evento_id=uuid_generate_v4(), + versao=versao+1,status='pendente',tentativas=0,lease_id=NULL,bloqueado_ate=NULL, + erro_codigo=NULL,proxima_tentativa=CURRENT_TIMESTAMP,updated_at=CURRENT_TIMESTAMP + WHERE ev.projeto_id=project AND ev.operacao<>'remover' AND NOT EXISTS ( + SELECT 1 FROM indice_origem o WHERE o.projeto_id=project + AND o.entidade_tipo=ev.entidade_tipo AND o.entidade_id=ev.entidade_id); + + -- Retira também trechos legados que ainda não tinham evento. + DELETE FROM chunk c WHERE c.projeto_id=project AND ( + EXISTS (SELECT 1 FROM projeto p WHERE p.id=project AND p.status='arquivado') + OR (c.entidade_tipo IN ('epico','feature','pbi','decisao') AND NOT EXISTS ( + SELECT 1 FROM indice_origem o WHERE o.projeto_id=c.projeto_id + AND o.entidade_tipo=c.entidade_tipo AND o.entidade_id=c.entidade_id AND o.elegivel))); + + -- Nenhuma janela de exposição de conteúdo arquivado ou desatualizado. + DELETE FROM chunk c USING indice_evento ev + WHERE c.projeto_id=project AND ev.projeto_id=c.projeto_id + AND ev.entidade_tipo=c.entidade_tipo AND ev.entidade_id=c.entidade_id + AND (ev.operacao='remover' OR ev.status<>'processado'); +END; +$$; + +CREATE FUNCTION publicar_evento_indice() RETURNS TRIGGER LANGUAGE plpgsql AS $$ +DECLARE + row_data JSONB; + kind TEXT; + source_id UUID; + project UUID; +BEGIN + FOR row_data IN SELECT value FROM jsonb_array_elements( + CASE WHEN TG_OP='INSERT' THEN jsonb_build_array(to_jsonb(NEW)) + WHEN TG_OP='DELETE' THEN jsonb_build_array(to_jsonb(OLD)) + ELSE jsonb_build_array(to_jsonb(OLD),to_jsonb(NEW)) END) + LOOP + kind := CASE WHEN TG_TABLE_NAME IN ('entidade_tecnologia','criterio_aceitacao') + THEN row_data->>'entidade_tipo' ELSE TG_TABLE_NAME END; + source_id := CASE WHEN TG_TABLE_NAME IN ('entidade_tecnologia','criterio_aceitacao') + THEN (row_data->>'entidade_id')::uuid ELSE (row_data->>'id')::uuid END; + FOR project IN + SELECT DISTINCT projeto_id FROM indice_origem + WHERE (kind='projeto' AND projeto_id=source_id) + OR (entidade_tipo=kind AND entidade_id=source_id) + OR (kind IN ('epico','feature','pbi') AND source_id=ANY(ancestrais)) + OR (kind='tecnologia' AND payload->'tecnologias_ids' ? source_id::text) + UNION + SELECT projeto_id FROM indice_evento WHERE (entidade_tipo=kind AND entidade_id=source_id) + OR (kind IN ('epico','feature','pbi') AND payload->'ancestor_ids' ? source_id::text) + UNION + SELECT projeto_id FROM indice_evento WHERE kind='projeto' AND projeto_id=source_id + ORDER BY 1 + LOOP + PERFORM sincronizar_indice_projeto(project); + END LOOP; + END LOOP; + RETURN NULL; +END; +$$; + +DO $$ +DECLARE table_name TEXT; +BEGIN + FOREACH table_name IN ARRAY ARRAY['projeto','epico','feature','pbi','decisao','criterio_aceitacao','entidade_tecnologia','tecnologia'] + LOOP + EXECUTE format('CREATE TRIGGER publicar_indice AFTER INSERT OR UPDATE OR DELETE ON %I FOR EACH ROW EXECUTE FUNCTION publicar_evento_indice()',table_name); + END LOOP; +END; +$$; + +-- Instalações existentes também entram na fila. +SELECT sincronizar_indice_projeto(id) FROM projeto ORDER BY id; diff --git a/docs/S2-03-INDEXACAO.md b/docs/S2-03-INDEXACAO.md new file mode 100644 index 0000000..f150a97 --- /dev/null +++ b/docs/S2-03-INDEXACAO.md @@ -0,0 +1,83 @@ +# S2-03 — Eventos e sincronização do índice + +A entrega consolida S2-03 e S2-04 (PBI-02.2.2 e PBI-02.2.3). A base é a main atualizada em 14/10/2026, já com S2-02. O processamento de arquivos continua no pipeline de documentos; itens e decisões usam o contrato estruturado existente da IA, /ingest/entity, que produz um trecho por origem. + +## Publicação e processamento + +A migration 019 cria indice_origem e a outbox indice_evento. Triggers nas tabelas de origem publicam o snapshot na mesma transação da escrita do backend. Se a alteração falhar, o evento também é desfeito. Não há escrita de dados de negócio pela IA. + +Épicos, features e PBIs concluídos são elegíveis quando nenhum ancestral está arquivado. Decisões salvas são elegíveis mesmo com o item em rascunho. O snapshot inclui projeto, tipo, nível, status, conteúdo, código do PBI, critérios, proveniência e tecnologias. Alterar critérios, tecnologias ou o nome de uma tecnologia também atualiza a origem. + +A outbox mantém a última versão por origem e projeto. Mudanças sucessivas antes do consumo são consolidadas; evento_id muda a cada snapshot distinto e versao aumenta. Repetir um snapshot não cria um novo evento. Esta tabela é fila e estado atual, não histórico de todas as versões. + +O worker do Node consulta a fila a cada 15 segundos, usa reserva de três minutos, timeout HTTP de dois minutos e até cinco tentativas com backoff. Valida conteúdo, projeto, tipo e vetor finito não nulo de 1024 dimensões. O Node grava um único trecho por origem, com ID determinístico, em transação. + +A conclusão exige o mesmo evento e a mesma reserva ainda válida. Respostas de eventos substituídos, removidos ou arquivados não podem sobrescrever o índice. Falhas registram somente códigos seguros. O evento permanece disponível para nova tentativa manual após o limite. + +## Atualização e arquivamento + +Ao mudar uma origem, o trecho anterior sai da busca até a nova versão terminar; não há janela com conteúdo antigo apresentado como atual. Ao arquivar a origem, um ancestral ou o projeto, a retirada ocorre na própria transação, sem esperar a IA. O consumidor de remoção não chama o serviço de embeddings. O arquivamento também limpa trechos legados. + +A publicação recalcula os snapshots do projeto afetado para cobrir decisões herdadas e alterações nos ancestrais. Projetos muito grandes podem exigir otimização futura dessa varredura. Não há varredura de todo o acervo a cada ciclo do worker. + +## Operação + +Aplicar as migrações com npm run migrate no backend e reiniciar o backend. O worker inicia junto com o serviço. Usa AI_SERVICE_URL e AI_SERVICE_TOKEN já existentes, sem novas bibliotecas ou segredo adicional. + +- GET /api/v1/projects/{projectId}/index-events lista os últimos 100 eventos, sem payload, com versão, operação, estado, tentativas, próxima tentativa e código da falha. +- POST /api/v1/projects/{projectId}/index-events/{eventId}/reprocess exige admin ou PO. Repetições enquanto pendente preservam a fila; reserva ativa ou projeto arquivado retorna 409. Evento de outro projeto ou substituído retorna 404. +- Ambos os endpoints seguem a autenticação e a guarda de acesso ao projeto. + +O serviço IA mantém o endpoint estruturado já usado pelo projeto. Não foi criado workflow n8n adicional nem alterado o pipeline de documentos da S2-02. + +## Validação + +No backend: + +```text +npm run typecheck +npm test +node --import tsx scripts/test-indexing-db.mts +node --import tsx scripts/test-indexing-db.mts --real-ai +``` + +O roteiro cria PostgreSQL 16/pgvector em contêiner separado, aplica todas as migrações e remove banco e contêiner ao terminar. O modo real-ai exige sinapse-ai-service e Ollama locais ativos; encaminha as requisições ao endpoint real sem imprimir credenciais. + +Os testes cobrem publicação após conclusão, decisões salvas/alteradas, rollback, reexecução, substituição, metadados, tecnologias, critérios, reservas exclusivas, falhas, nova tentativa, respostas atrasadas, limite de tentativas, isolamento entre projetos, autenticação, autorização e retirada da busca ao arquivar. A integração real verifica embedding 1024d, busca híbrida, atualização do texto sem duplicação e arquivamento. + +A revisão humana e a integração por Pull Request continuam etapas posteriores à implementação. + +### Resultado em 14/10/2026 + +Suíte geral: 338 testes aprovados, sem falhas; 29 testes condicionais ignorados nessa execução. Execução específica com PostgreSQL/pgvector e IA real: 7 testes aprovados, sem falhas ou skips, incluindo os cenários de API e busca. Todos os ambientes de banco criados para os testes foram removidos. + +### Revisão de erros em 14/10/2026 + +Três falhas foram reproduzidas antes da correção e cobertas por regressões: + +- Renomear o cenário BDD não alterava o snapshot. O nome agora faz parte do conteúdo indexado. +- Excluir um ancestral em rascunho podia deixar decisões órfãs indexadas. O evento agora preserva os IDs dos ancestrais para localizar o projeto mesmo depois da exclusão. +- Uma reserva podia expirar enquanto a conclusão aguardava um lock. A validade agora usa o relógio atual do PostgreSQL, e não o horário fixado no início da transação. + +Após as correções: 338 testes gerais aprovados (29 condicionais ignorados), 7 testes específicos aprovados com PostgreSQL e IA real (sem skips), e verificação de tipos aprovada. + +### Segunda revisão em 14/10/2026 + +Dois problemas adicionais foram reproduzidos antes da correção: + +- Relógio do backend adiantado podia liberar uma reserva ainda ativa. O reprocessamento agora consulta a validade pelo relógio do PostgreSQL. +- A recuperação de reservas expiradas não respeitava o projeto informado ao claim. O filtro de projeto agora vale também para essa etapa; o worker global continua processando todo o acervo. + +Os testes adicionais também confirmaram transferência entre projetos sem resíduos ou resposta atrasada, reservas exclusivas com workers simultâneos e rollback de falha de gravação preservando o índice anterior. As fixtures usam nomes exclusivos para não colidir em execuções paralelas. + +Execução específica com PostgreSQL/pgvector e IA real: 12 testes aprovados, sem falhas ou skips. Verificação de tipos e compilação aprovadas. + +### Rodadas ampliadas em 14/10/2026 + +Foram concluídas cinco rodadas de oito sequências com 80 operações cada (3.200 operações), incluindo alterações de título e estado, critérios BDD, tecnologias, transferência entre projetos, exclusão de decisões, rollback, falhas e respostas atrasadas. O índice foi comparado após cada operação a um modelo independente das origens; ao esvaziar a fila, toda origem elegível precisava aparecer exatamente uma vez e com metadados atuais. + +A execução final específica aprovou 32 testes/subtestes sem falhas ou skips, incluindo IA real. Também foram testados nove retornos inválidos/erros HTTP, criação de critério e conclusão pela API, decisão salva pela API, permissões de desenvolvedor alocado e migração sobre dados preexistentes com reaplicação sem duplicação. + +Não apareceram novos defeitos no programa nessas rodadas. Foi ajustado o gerador de cenários, que inicialmente repetia a ordem de critérios; o banco rejeitou corretamente esses dados de teste. + +Suíte geral: 338 aprovados, zero falhas e 37 condicionais ignorados. Tipos e compilação aprovados. Os testes específicos de PostgreSQL estão incluídos no CI e podem ser executados com npm run test:integration:indexing; para incluir IA real, usar npm run test:integration:indexing -- --real-ai. O CI não inicia Ollama e, portanto, ignora somente o cenário de IA real nessa etapa. diff --git a/docs/api/openapi.yaml b/docs/api/openapi.yaml index 907e293..36a356f 100644 --- a/docs/api/openapi.yaml +++ b/docs/api/openapi.yaml @@ -12,6 +12,50 @@ servers: description: Serviço de IA local paths: + /api/v1/projects/{projectId}/index-events: + parameters: + - {name: projectId, in: path, required: true, schema: {type: string, format: uuid}} + get: + summary: Lista os últimos 100 eventos de indexação do projeto + operationId: listIndexEvents + security: [{cookieAuth: []}] + responses: + '200': + description: Estados, versões, tentativas e códigos seguros de falha, sem conteúdo da origem. + content: + application/json: + schema: + type: object + properties: + items: + type: array + items: + $ref: '#/components/schemas/IndexEvent' + '401': {$ref: '#/components/responses/Unauthorized'} + '404': {description: Projeto inexistente ou sem acesso} + /api/v1/projects/{projectId}/index-events/{eventId}/reprocess: + parameters: + - {name: projectId, in: path, required: true, schema: {type: string, format: uuid}} + - {name: eventId, in: path, required: true, schema: {type: string, format: uuid}} + post: + summary: Reenfileira um evento de indexação (admin ou PO) + operationId: reprocessIndexEvent + description: Repetir enquanto pendente preserva a fila. Eventos substituídos retornam 404; reserva ativa ou projeto arquivado retorna 409. + security: [{cookieAuth: []}] + responses: + '202': + description: Evento pendente + content: + application/json: + schema: + type: object + properties: + status: {type: string, enum: [pendente]} + '400': {$ref: '#/components/responses/ValidationError'} + '401': {$ref: '#/components/responses/Unauthorized'} + '403': {description: Sem permissão de escrita} + '404': {description: Projeto ou evento inexistente ou sem acesso} + '409': {description: Projeto arquivado ou evento em processamento} /api/v1/projects/{projectId}/epicos: $ref: './epics-compat.yaml#/collection' /api/projects/{projectId}/epicos: @@ -2964,6 +3008,23 @@ components: name: sinapse_session schemas: + IndexEvent: + type: object + properties: + evento_id: {type: string, format: uuid} + projeto_id: {type: string, format: uuid} + entidade_tipo: {type: string, enum: [epico, feature, pbi, decisao]} + entidade_id: {type: string, format: uuid} + versao: {type: integer, minimum: 1} + operacao: {type: string, enum: [indexar, remover]} + status: {type: string, enum: [pendente, processando, processado, falha]} + tentativas: {type: integer, minimum: 0} + proxima_tentativa: {type: string, format: date-time} + erro_codigo: + type: string + nullable: true + enum: [ATTEMPTS_EXHAUSTED, INVALID_RESPONSE, INVALID_SCOPE, TIMEOUT, SERVICE_HTTP_ERROR, SERVICE_UNAVAILABLE, PROCESSING_ERROR, null] + updated_at: {type: string, format: date-time} TechnologyOption: type: object required: [id, nome]