Feature/pav 121 processar lotes

[RuhanFreitas]

Card

• PAV-121

O que foi feito

• Organiza o pós-coleta do scraper-go em lotes limitados: coleta → normalização → deduplicação → classificação → persistência → indexação, com fila limitada (2 × maior lote), backpressure e cancelamento em todas as etapas. Não cria goroutine por vaga e não acumula a coleta inteira antes de persistir.
• Adiciona lotes configuráveis e fail-fast: SCRAPER_CLASSIFICATION_BATCH_SIZE=100 (máx. 1000), SCRAPER_PERSIST_BATCH_SIZE=100 (máx. 1000) e SCRAPER_INDEX_BATCH_SIZE=250 (máx. 2500). Valor ausente usa o default; valor explícito ≤0, vazio ou acima do máximo impede a inicialização. 1 permanece válido para diagnóstico. Compose usa ${VAR-default}.
• O catálogo coletado continua no Valkey (scraper:job:{id}) como fonte da verdade da execução; o PostgreSQL segue para usuários/saved_jobs. A indexação invertida só recebe IDs confirmados por SaveBatch. Persistência em MULTI/EXEC por lote, com upsert pelo ID estável: atualiza descrição, URL, classificação, fontes e keywords; campo vazio não apaga dado existente. Falha no lote faz rollback, não indexa aquele lote e aplica retry limitado só para erros transitórios.
• Indexação incremental com SADD em pipelines de SCRAPER_INDEX_BATCH_SIZE, sem RENAME do set inteiro (isso apagaria lotes anteriores da mesma execução). Se a indexação falhar após o persist, ReindexPersistedJobs relê os documentos no Valkey e reconstrói os índices, sem repetir a coleta externa.
• Deduplicação intra e inter-lote usa as chaves já existentes no domínio (ID externo, title|company|location e URL canônica). Títulos iguais de vagas diferentes são preservados. A classificação em lote mantém o resultado individual, sem novas famílias, keywords ou pontuações. A fórmula da cache key permanece inalterada.

Validação

• go test ./... em scraper-go
• go test -race ./... em scraper-go
• Conferir .env.example, docker-compose.yml e SCRAPER.md com SCRAPER_CLASSIFICATION_BATCH_SIZE=100, - [x]SCRAPER_PERSIST_BATCH_SIZE=100 e SCRAPER_INDEX_BATCH_SIZE=250
• Confirmar que valores inválidos (0, vazio, acima do máximo) impedem a inicialização e que 1 é aceito
• Subir o scraper e validar nos logs scraper batch sizes / scraper runtime configurado e os logs por run_id/lote (recebidos, duplicados, aprovados, inseridos, atualizados, indexados, falhas, duração)
• Executar uma coleta controlada e confirmar: só vagas persistidas são indexadas; falha de persistência não indexa o lote; falha de indexação fica registrada e pode ser reconciliada via ReindexPersistedJobs; GET /jobs/search e healthcheck permanecem compatíveis
• Confirmar ausência de alterações em frontend/, front_admin/, novas famílias/keywords e package-lock.json

[Jovinull]

Revisei o diff todo + o card. Arquitetura tá boa, backpressure/cancelamento ok, só indexa o que persistiu, testes cobrem bem. CI verde. Nada que eu ache bloqueante, mas uns pontos:

1. O ProcessStats que o processIncomingJobs retorna tá sendo descartado no runWithConcurrency (processed, _, processErr). Com isso o OnComplete do cronjob virou (kws, len(jobs), len(jobs), ...) — saved == scraped sempre, e sumiram os new_saved/skipped do log de conclusão. Os números reais (inserted/updated/duplicates/etc) só ficam no slog por lote agora. Quem consome esse callback ou lê new_saved pra métrica vai quebrar sem perceber. Acho que vale propagar o stats (ou pelo menos inserted+updated como saved).

2. Umas docstrings do jobstore.go foram removidas sem relação com a mudança (StableID, GetAll, GetSample, Count) — dá pra voltar.

3. isTransient classificando erro por substring da mensagem é meio frágil pra mudança de driver/idioma, mas pro escopo tá ok — só deixa um comentário marcando como heurística.

4. classifier.ClassifyJobs parece ficar sem uso em prod depois desse PR (o processStageBatch usa Classify em loop). Confirma? Se for, ou tira ou usa.

Perguntas:

• Falha de persist aborta a run inteira. O card §8 deixa em aberto continuar vs interromper — foi decisão consciente? Se sim vale documentar no código.
• Os sets scraper:jobs:keyword:* agora só crescem via SADD. Órfão acumula neles ou o Expire de 9d dá conta? (o índice global tem poda no GetSample, os de keyword não vi)
• Tem os números do benchmark de memória pra colar aqui? A checklist marca como feito mas não tem os valores.

[Benevanio]

Status

❌ REQUER ALTERAÇÕES ANTES DO MERGE

A implementação de processamento em lotes e controle de backpressure está bem encaminhada, porém foram identificados alguns pontos que podem causar regressões funcionais e inconsistências nos dados processados.

Pontos que precisam ser corrigidos

• 🔴 ALTO — scraper-go/internal/pipeline/index.go:99-137 — A indexação deixou de utilizar a reconstrução atômica com RENAME e passou a utilizar SADD de forma incremental, porém as associações antigas não são removidas.

  Com isso, uma vaga que deixe de corresponder a uma keyword, família, senioridade ou tecnologia pode continuar presente nos sets anteriores enquanto esses índices forem atualizados por outras vagas. Consequentemente, o GET /jobs/search pode retornar vagas que não correspondem mais aos filtros informados.

  Solicitação: ajustar a estratégia de indexação para garantir que os índices representem exatamente o estado atual da execução. Pode ser feita a remoção das associações obsoletas ou, preferencialmente, a reconstrução dos índices de forma isolada seguida de uma publicação atômica, sem remover os lotes que já foram processados.

• 🔴 ALTO — scraper-go/internal/pipeline/process.go:171-176 — Duplicatas que chegam após o flush estão sendo descartadas apenas com base em suas chaves, sem que ocorra o merge com o documento que já foi persistido.

  Isso pode causar perda de informações quando a mesma vaga é encontrada por fontes diferentes. Dados complementares como URL, descrição, fontes e keywords podem deixar de ser incorporados dependendo de em qual lote a ocorrência foi processada.

  Solicitação: preservar o comportamento de merge entre lotes, buscando e mesclando o documento já persistido ou mantendo estado suficiente para realizar esse merge quando a duplicata chegar em um lote posterior.

• 🟡 MÉDIO — scraper-go/internal/cronjob/cronjob.go:283 — O callback OnComplete está recebendo saved = len(jobs), independentemente do resultado real da persistência.

  Esse valor pode divergir da quantidade efetivamente salva, considerando inserções, atualizações, rejeições, duplicatas e falhas. Isso pode gerar métricas incorretas e impactar qualquer consumidor que utilize o callback para reportar os resultados da execução.

  Solicitação: propagar o ProcessStats até o callback ou, no mínimo, utilizar os contadores reais de persistência para que OnComplete continue representando corretamente o resultado da execução.

Resumo

Os mecanismos de processamento em lotes e backpressure estão bem encaminhados, porém os pontos acima precisam ser corrigidos antes da integração, principalmente porque podem resultar em:

• resultados de busca contendo associações obsoletas;
• perda de informações durante a deduplicação entre lotes;
• métricas incorretas sobre o resultado do scraper.

Solicito as alterações nesses pontos antes do merge. Após os ajustes, recomendo validar especialmente cenários em que a mesma vaga aparece em lotes diferentes, vagas que deixam de corresponder a determinados filtros e execuções com inserções, atualizações, duplicatas e falhas de persistência.