Giomoret · Oct 22, 2026 · Oct 22, 2026
diff --git a/backend/scripts/test-indexing-db.mts b/backend/scripts/test-indexing-db.mts
@@ -63,7 +63,7 @@ try {
   });
   process.exitCode=exitCode;
   if (exitCode === 0) {
-    const suggestions = spawnSync(process.execPath, ["--import", "tsx", "--test", "src/modules/suggestions/suggestions.db.test.ts"], {
+    const suggestions = spawnSync(process.execPath, ["--import", "tsx", "--test", "src/modules/suggestions/suggestions.db.test.ts", "src/modules/pbis/pbis.relations.db.test.ts", "src/modules/documents/chunks.http.db.test.ts"], {
       env: { ...process.env, NODE_ENV: "test", DATABASE_URL: url, ARCHIVE_TEST_DATABASE_URL: url },
       stdio: "inherit",
     });

diff --git a/backend/src/app.ts b/backend/src/app.ts
@@ -57,6 +57,9 @@ app.get("/health", async (_req: Request, res: Response) => {
   });
 });
 
+// O callback de ingestão usa token de serviço e valida documento/projeto no próprio serviço.
+app.use("/api/v1/projects/:projectId/documents/:documentId/chunks", chunksRouter);
+
 app.use("/api/v1", createProjectAccessRouter());
 app.use("/api", createProjectAccessRouter());
 
@@ -110,8 +113,6 @@ app.use("/api/v1/projects/:projectId/backlog-search", backlogSearchRouter);
 
 // Documentos do projeto (S1-19/S1-20/S1-22)
 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);
 app.use("/api/v1/projects/:projectId/documents", documentTextRouter);

diff --git a/backend/src/modules/documents/chunks.http.db.test.ts b/backend/src/modules/documents/chunks.http.db.test.ts
@@ -0,0 +1,117 @@
+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 { DocumentsRepository } from "./documents.repository.js";
+import { AdminRepository } from "../admin/admin.repository.js";
+
+test("Callback de chunks autentica o serviço e preserva as guardas de documentos", { skip: !process.env.ARCHIVE_TEST_DATABASE_URL }, async t => {
+  const db = new Pool({ connectionString: validateTarget(process.env.ARCHIVE_TEST_DATABASE_URL, "test") });
+  const { app } = await import("../../app.js");
+  const { pool } = await import("../../database/db.js");
+  const { env } = await import("../../config/env.js");
+  const server = app.listen(0, "127.0.0.1");
+  await new Promise<void>(resolve => server.once("listening", resolve));
+  const base = "http://127.0.0.1:" + (server.address() as { port: number }).port + "/api/v1/projects/";
+  const [project, otherProject, document, user] = Array.from({ length: 4 }, () => randomUUID());
+  const token = env.DOCUMENT_INGESTION_TOKEN || "sinapse-dev-document-ingestion-token-2026";
+  const payload = { processamento_lease_id: randomUUID() as string, chunks: [{ chunk_index: 0, content: "Texto de teste", embedding: Array(1024).fill(0.1) }] };
+  const documents = new DocumentsRepository(db);
+  const claim = async () => {
+    const claimed = (await documents.claimPendingIngestion(100)).find(row => row.id === document);
+    assert.ok(claimed);
+    return claimed.processamento_lease_id;
+  };
+  const endpoint = (projectId = project) => base + projectId + "/documents/" + document + "/chunks";
+  const send = (url = endpoint(), credential: string | null = token, body: unknown = payload) => fetch(url, {
+    method: "POST",
+    headers: { "Content-Type": "application/json", ...(credential ? { Authorization: "Bearer " + credential } : {}) },
+    body: JSON.stringify(body),
+  });
+  try {
+    await db.query("INSERT INTO usuario(id,nome,email,senha_hash,role) VALUES ($1,'QA',$2,'h','admin')", [user, user + "@chunks.test"]);
+    for (const id of [project, otherProject]) {
+      await db.query("INSERT INTO projeto(id,nome,cliente,status) VALUES ($1,$2,'QA','ativo')", [id, id]);
+    }
+    await db.query("INSERT INTO documento(id,projeto_id,nome,caminho) VALUES ($1,$2,'QA.txt','qa.txt')", [document, project]);
+    payload.processamento_lease_id = await claim();
+    await t.test("recusa token ausente ou inválido sem gravar chunks", async () => {
+      for (const credential of [null, "token-invalido"]) {
+        const response = await send(endpoint(), credential);
+        assert.equal(response.status, 401);
+        assert.equal((await response.json() as { code: string }).code, "INGEST_TOKEN_INVALID");
+      }
+      assert.equal((await db.query("SELECT count(*)::int AS n FROM chunk WHERE entidade_id=$1", [document])).rows[0].n, 0);
+    });
+    await t.test("token sozinho não substitui a reserva de execução", async () => {
+      assert.equal((await send(endpoint(), token, { chunks: payload.chunks })).status, 400);
+      assert.equal((await send(endpoint(), token, { ...payload, processamento_lease_id: "invalida" })).status, 400);
+      assert.equal((await send(endpoint(), token, { ...payload, processamento_lease_id: randomUUID() })).status, 409);
+      assert.equal((await db.query("SELECT count(*)::int AS n FROM chunk WHERE entidade_id=$1", [document])).rows[0].n, 0);
+    });
+    await t.test("token e reserva válidos funcionam sem sessão e repetir o envio não duplica chunks", async () => {
+      for (let i = 0; i < 2; i++) {
+        const response = await send();
+        assert.equal(response.status, 200, await response.text());
+      }
+      assert.equal((await db.query("SELECT count(*)::int AS n FROM chunk WHERE entidade_id=$1", [document])).rows[0].n, 1);
+      assert.equal((await db.query("SELECT status_processamento FROM documento WHERE id=$1", [document])).rows[0].status_processamento, "processado");
+    });
+    await t.test("não permite associar documento ao projeto errado", async () => {
+      const response = await send(endpoint(otherProject));
+      assert.equal(response.status, 404);
+      assert.equal((await db.query("SELECT projeto_id FROM chunk WHERE entidade_id=$1", [document])).rows[0].projeto_id, project);
+    });
+    await t.test("expurgo invalida callback antigo; nova reserva funciona e respostas antigas não sobrescrevem", async () => {
+      const oldLease = payload.processamento_lease_id;
+      await new AdminRepository(db).purgeProjectContext(project!, project!, user!);
+      assert.equal((await send()).status, 409);
+      assert.equal((await db.query("SELECT count(*)::int AS n FROM chunk WHERE entidade_id=$1", [document])).rows[0].n, 0);
+      assert.equal((await db.query("SELECT status_processamento FROM documento WHERE id=$1", [document])).rows[0].status_processamento, "falha");
+      assert.equal(await documents.retryIngestion(project!, document!), true);
+      assert.equal((await send()).status, 409);
+      const expiredLease = await claim();
+      assert.notEqual(expiredLease, oldLease);
+      assert.equal((await send()).status, 409);
+      payload.processamento_lease_id = expiredLease;
+      await db.query("UPDATE documento SET processamento_bloqueado_ate=clock_timestamp()-INTERVAL '1 second' WHERE id=$1", [document]);
+      assert.equal((await send()).status, 409);
+      const freshLease = await claim();
+      assert.notEqual(freshLease, expiredLease);
+      payload.processamento_lease_id = freshLease;
+      payload.chunks[0]!.content = "Conteúdo da nova tentativa";
+      assert.equal((await send()).status, 200);
+      const saved = (await db.query("SELECT id,texto FROM chunk WHERE entidade_id=$1", [document])).rows;
+      payload.processamento_lease_id = oldLease;
+      payload.chunks[0]!.content = "Resposta antiga";
+      assert.equal((await send()).status, 409);
+      payload.processamento_lease_id = expiredLease;
+      assert.equal((await send()).status, 409);
+      assert.deepEqual((await db.query("SELECT id,texto FROM chunk WHERE entidade_id=$1", [document])).rows, saved);
+      payload.processamento_lease_id = freshLease;
+      assert.equal((await send()).status, 200);
+      assert.deepEqual((await db.query("SELECT id,texto FROM chunk WHERE entidade_id=$1", [document])).rows, saved);
+    });
+    await t.test("projeto arquivado continua recusando escrita", async () => {
+      await db.query("UPDATE projeto SET status='arquivado' WHERE id=$1", [project]);
+      const response = await send();
+      assert.equal(response.status, 409);
+      assert.equal((await response.json() as { code: string }).code, "ARCHIVE_CONFLICT");
+    });
+    await t.test("token de serviço não concede acesso às rotas de usuário", async () => {
+      for (const path of [base + project + "/documents", base + project + "/documents/" + document + "/chunks/extra"]) {
+        const response = await fetch(path, { headers: { Authorization: "Bearer " + token } });
+        assert.equal(response.status, 401);
+        assert.equal((await response.json() as { code: string }).code, "UNAUTHORIZED");
+      }
+    });
+  } finally {
+    server.closeAllConnections();
+    await new Promise<void>(resolve => server.close(() => resolve()));
+    await db.query("DELETE FROM projeto WHERE id=ANY($1::uuid[])", [[project, otherProject]]);
+    await db.query("DELETE FROM usuario WHERE id=$1", [user]);
+    await db.end();
+    await pool.end();
+  }
+});
diff --git a/backend/src/modules/documents/chunks.routes.ts b/backend/src/modules/documents/chunks.routes.ts
@@ -34,8 +34,9 @@ export function createChunksRouter(service: ChunksService = chunksService): Rout
    *     summary: Persiste chunks com embeddings (pipeline de ingestão -> backend)
    *     description: >
    *       Chamado pelo pipeline após extrair texto, chunking e embeddings.
-   *       Substitui qualquer chunk anterior do mesmo documento (reindexação
-   *       idempotente) e marca o documento como 'processado'. Autenticação por
+   *       Exige processamento_lease_id da reserva vigente emitida pelo backend.
+   *       Entregas antigas ou canceladas retornam 409. Repetições concluídas
+   *       retornam o resultado salvo, sem regravar. Autenticação por
    *       token compartilhado em Authorization: Bearer <DOCUMENT_INGESTION_TOKEN>.
    *     tags: [Documents]
    *     security:
@@ -50,7 +51,7 @@ export function createChunksRouter(service: ChunksService = chunksService): Rout
    *       404:
    *         description: Projeto ou documento não encontrados
    *       409:
-   *         description: Projeto arquivado
+   *         description: Projeto arquivado ou reserva expirada, cancelada ou substituída
    */
   router.post("/", requireIngestToken, async (req: Request, res: Response, next: NextFunction) => {
     try {
@@ -60,6 +61,7 @@ export function createChunksRouter(service: ChunksService = chunksService): Rout
       const result = await service.persist({
         projetoId: paramOf(req.params.projectId),
         documentoId: paramOf(req.params.documentId),
+        leaseId: (req.body as { processamento_lease_id?: string }).processamento_lease_id as string,
         chunks: (req.body as { chunks?: unknown }).chunks as never,
       });
       res.status(200).json(result);

diff --git a/backend/src/modules/documents/chunks.service.ts b/backend/src/modules/documents/chunks.service.ts
@@ -1,6 +1,6 @@
 import type { PoolClient } from "pg";
 import { pool } from "../../database/db.js";
-import { AppError, NotFoundError, ValidationError, validateUuid } from "../../shared/errors.js";
+import { AppError, ConflictError, NotFoundError, ValidationError, validateUuid } from "../../shared/errors.js";
 import { ArchiveConflict } from "../projects/archive.types.js";
 import { lockHierarchy } from "../projects/hierarchy-archive.js";
 import { auditService } from "../audit/audit.service.js";
@@ -15,6 +15,7 @@ export interface IncomingChunk {
 export interface PersistChunksInput {
   projetoId: string;
   documentoId: string;
+  leaseId: string;
   chunks: IncomingChunk[];
 }
 
@@ -57,6 +58,7 @@ export class ChunksService {
   async persist(input: PersistChunksInput): Promise<PersistChunksResult> {
     validateUuid(input.projetoId, "ID do projeto");
     validateUuid(input.documentoId, "ID do documento");
+    validateUuid(input.leaseId, "Reserva de processamento");
     if (!Array.isArray(input.chunks) || input.chunks.length === 0) {
       throw new ValidationError("Envie ao menos um chunk.");
     }
@@ -83,13 +85,32 @@ export class ChunksService {
         throw new ArchiveConflict("Projeto arquivado é somente leitura e não recebe novos chunks.");
       }
 
-      const document = await client.query<{ id: string; projeto_id: string; status_processamento: string }>(
-        "SELECT id, projeto_id, status_processamento FROM documento WHERE id = $1 AND projeto_id = $2 FOR UPDATE",
+      const document = await client.query<{ status_processamento: string; processamento_lease_id: string | null; lease_active: boolean }>(
+        "SELECT status_processamento, processamento_lease_id, processamento_bloqueado_ate > clock_timestamp() AS lease_active FROM documento WHERE id = $1 AND projeto_id = $2 FOR UPDATE",
         [input.documentoId, input.projetoId],
       );
       if (document.rowCount === 0) {
         throw new NotFoundError("Documento não encontrado no projeto informado.");
       }
+      const current = document.rows[0]!;
+      if (current.status_processamento === "processado") {
+        const receipt = await client.query<{ total: number; matching: number }>(
+          `SELECT count(*)::int AS total,
+            count(*) FILTER (WHERE metadados_json->>'ingestion_lease_id'=$3)::int AS matching
+           FROM chunk WHERE projeto_id=$1 AND entidade_tipo='documento' AND entidade_id=$2`,
+          [input.projetoId, input.documentoId, input.leaseId],
+        );
+        const row = receipt.rows[0]!;
+        if (row.total > 0 && row.total === row.matching) {
+          await client.query("COMMIT");
+          return { documento_id: input.documentoId, projeto_id: input.projetoId,
+            total_chunks: row.total, status_processamento: "processado" };
+        }
+      }
+      if (current.status_processamento !== "processando"
+        || current.processamento_lease_id !== input.leaseId || !current.lease_active) {
+        throw new ConflictError("A reserva de ingestão expirou, foi cancelada ou pertence a outra execução.");
+      }
 
       // Reindexação idempotente: substitui chunks anteriores do mesmo documento.
       await client.query(
@@ -105,14 +126,16 @@ export class ChunksService {
             input.projetoId,
             input.documentoId,
             chunk.content,
-            JSON.stringify({ ...(chunk.metadata ?? {}), chunk_index: chunk.chunk_index }),
+            JSON.stringify({ ...(chunk.metadata ?? {}), chunk_index: chunk.chunk_index, ingestion_lease_id: input.leaseId }),
             toVectorLiteral(chunk.embedding),
           ],
         );
       }
 
       await client.query(
-        "UPDATE documento SET status_processamento = 'processado', updated_at = CURRENT_TIMESTAMP WHERE id = $1",
+        `UPDATE documento SET status_processamento='processado', processamento_erro=NULL,
+          processamento_lease_id=NULL, processamento_bloqueado_ate=NULL,
+          processamento_tentativas=processamento_tentativas+1, updated_at=CURRENT_TIMESTAMP WHERE id=$1`,
         [input.documentoId],
       );
 
@@ -139,7 +162,7 @@ export class ChunksService {
       };
     } catch (error) {
       await client.query("ROLLBACK");
-      if (error instanceof AppError) throw error;
+      if (error instanceof AppError || error instanceof ArchiveConflict) throw error;
       throw new AppError("Não foi possível persistir os chunks agora.", 503, "CHUNK_PERSIST_FAILED");
     } finally {
       client.release();

diff --git a/backend/src/modules/pbis/pbis.controller.ts b/backend/src/modules/pbis/pbis.controller.ts
@@ -78,7 +78,7 @@ export class PbisController {
   listRelations = async (req: Request, res: Response, next: NextFunction): Promise<void> => {
     try {
       const id = getParamId(req.params.id);
-      const result = await this.relationsService.list(id);
+      const result = await this.relationsService.list(id, req.auth!.id);
       res.status(200).json(result);
     } catch (error) {
       next(error);