Plataforma de Integrações (E-2)

Integrações de entrada/saída configuráveis pelo admin, sem desenvolvimento: um runtime genérico interpreta configuração persistida. Decisão em ADR-023 (.claude/docs/decisions.md) e skill integration-platform. Isolamento por organização (orgId) em toda entidade.

Modelo

Entidade	Papel
AuthConfig	Credencial plugável: NONE / API_KEY / BASIC / BEARER / OAUTH2 / LOGIN_TOKEN. O segredo é cifrado em repouso (AES-256-GCM, INTEGRATION_SECRET_KEY) e nunca devolvido pela API (hasSecret sinaliza presença).
DataSource	Origem/destino lógico: baseUrl, kind (inbound/outbound/bidirectional), auth, headers padrão, timeout.
Connector	Operação REST configurável: method, path com templating {{var}}, query/headers/body, paginação (page/offset/cursor), responseMapping (de-para tipado → canônico), targetEntity.
EntityDefinition	Entidade canônica (metamodelo, ADR-024): key, origin (BUILTIN/CUSTOM), fields[] (name+type). Alvo do de-para; built-in semeadas por org.
IntegrationRecord	Staging bruto (fatia 3): rawPayload retido + mappedPayload + payloadHash + status. Reprocessável sem rechamar a fonte.
EntityRecord	Canônico persistido (fatia 3): naturalKey (indexada) + data + refKeys (GIN, chaves de junção). Navegável/consultável.
IntegrationSchedule	MANUAL ou CRON (expressão + timezone + maxConcurrency).
SyncRun	Execução (histórico/tracking): status, estatísticas, cursor, tentativa.
SyncRecordLog	Log por registro/lote (troubleshooting/auditoria).

Auth LOGIN_TOKEN — login com corpo e sessão configuráveis

Cobre o padrão "logar e reusar a sessão": faz login (config.loginMethod, default POST) em config.loginUrl com o corpo de config.loginBody e reusa o que voltar nas chamadas seguintes.

O corpo é livre: nenhum campo tem nome fixo, porque cada sistema batiza os seus (UserName/ Password no SAP B1, username/password em outros, mais extras como CompanyDB ou grant_type). O admin monta chave a chave na tela. A senha não fica na config — onde ela entra, escreve-se o placeholder {{secret}} (aliases: {{password}}, {{senha}}), substituído pelo segredo cifrado no momento da chamada. Headers extras do login: config.loginHeaders (opcional — um login que só manda corpo deixa vazio). A sessão é cacheada por credencial (config.tokenTtlSec, default 300s) e um 401 da fonte força relogin + repetição da chamada.

Formato legado

Configs anteriores usavam userField/passField + username + extraBody; o runtime ainda as aceita quando não há loginBody, e a tela as converte para loginBody ao abrir a credencial.

Dois modos de captura da sessão (na tela, campo Modo da sessão):

Modo	Config	Como a sessão é capturada e reenviada
Token no corpo (default)	tokenPath	Extrai o token da resposta por dot-path (access_token, data.token) e injeta em headerName (default Authorization) com headerScheme (default Bearer).
Cookie de sessão	sessionCookie: true	Reaproveita os cookies do Set-Cookie do login, filtrados por cookieNames, e monta a string Cookie completa em headerName (default Cookie, sem prefixo).

No modo cookie, sessionTtlPath lê a validade da própria resposta do login (sessionTtlUnit: minutes/seconds, menos sessionTtlMarginSec, default 120s) em vez de um número fixo.

Exemplo — SAP Business One Service Layer (botão Preencher p/ SAP Business One na tela): o POST /b1s/v2/Login recebe { CompanyDB, UserName, Password }, responde { SessionId, SessionTimeout } e seta os cookies B1SESSION (a sessão) e ROUTEID (afinidade de nó num Service Layer balanceado — sem ele a sessão "some", erro -5002, no round-robin).

{
  "loginUrl": "https://sap-b1:50000/b1s/v2/Login",
  "loginBody": {
    "CompanyDB": "SBODEMOBR",
    "UserName": "manager",
    "Password": "{{secret}}"
  },
  "sessionCookie": true,
  "cookieNames": ["B1SESSION", "ROUTEID"],
  "headerName": "Cookie",
  "sessionTtlPath": "SessionTimeout",
  "sessionTtlUnit": "minutes"
}

A senha vai no campo Segredo (cifrado em repouso), nunca na config.

AuthConfig ──< DataSource ──< Connector ──< IntegrationSchedule
                                   │              │
                                   └──< SyncRun >─┘ ──< SyncRecordLog
                                   │
              (execução/ingestão)  └──> IntegrationRecord (staging bruto)
                                        └─(de-para p/ targetEntity)─> EntityRecord (canônico)
EntityDefinition (metamodelo: built-in + custom) ──guia o de-para──┘

Runtime de execução

1. O runtime HTTP resolve DataSource + AuthConfig, monta a requisição por templating, aplica a auth plugável, pagina a fonte e mapeia a resposta para o modelo canônico.
2. O teste de conexão (POST /connectors/:id/test) é um dry-run: uma chamada real, sem persistir, com status/duração + amostra sanitizada. É fiel ao run — envia os mesmos params da 1ª página (page/pageSize/…); se extrair 0 registros de um 200, mostra o corpo cru + os params enviados (revela itemsPath/paginação errados, em vez de esconder o problema).
3. Uma execução (POST /connectors/:id/run, ou via scheduler) cria um SyncRun e roda a ingestão em background: grava o staging bruto (IntegrationRecord) + o canônico (EntityRecord) + SyncRecordLog (amostra). Se ler 0, registra um log diagnóstico com o corpo cru. Reprocessar a execução: POST /sync-runs/:id/reprocess (re-executa o conector).
4. O scheduler (poll por minuto, gated por INTEGRATION_SCHEDULER_ENABLED) dispara os agendamentos CRON cujo minuto casa, respeitando maxConcurrency.
5. A tela de Execuções atualiza ao vivo (SSE + Redis pub/sub): o executor publica cada mudança de status do SyncRun num canal Redis; o endpoint GET /sync-runs/stream faz stream por org.

Leitura paginada e progresso por lote (ADR-051)

A leitura da fonte é paginada pela configuração do conector (pagination: page/offset/ cursor, pageSize, pageParam/sizeParam/cursorParam, itemsPath). Ex.: SAP B1 Service Layer (OData) usa type: "offset", pageParam: "$skip", sizeParam: "$top", itemsPath: "value".

Cada página é gravada assim que chega — não há acúmulo da fonte inteira em memória — e o parcial é publicado no SSE. Na tela, a linha da execução mostra, enquanto roda: a duração correndo, os contadores lidos / gravados / erros subindo e a etapa (lote 12 gravado (50 itens)); o detalhe da linha traz páginas lidas.

Gravação em lote (ADR-052). Os registros são persistidos de N em N (INTEGRATION_BATCH_SIZE, default 50): uma consulta traz os hashes existentes do lote e uma transação grava staging + canônico de todos. Antes eram ~3 idas ao banco por registro — é daí que vinham os minutos numa carga grande. O tamanho da página (quanto o ERP devolve por chamada) e o do lote (quanto se grava por transação) são independentes: a página de 20 do SAP não dita o custo de gravação.

Se um lote falha, ele é refeito registro a registro: o erro fica no registro que o causou e os demais são gravados, em vez de a transação inteira levar 49 registros bons junto.

Registros que a fonte deixou de devolver (ADR-053)

Cada registro gravado leva o carimbo da execução que o viu (lastSeenRunId) — é o "conjunto de códigos vistos", guardado na linha em vez da memória. No fim, quem não tem o carimbo da execução corrente é o que a fonte deixou de devolver. Não é preciso segurar a base inteira para comparar, nem montar um NOT IN do tamanho do catálogo.

Marque "Lê a base completa" no conector (deactivateMissing, desligado por padrão) e, ao fim de uma execução completa, os registros ausentes viram inativos (enabled = false). Quando voltam a vir, o upsert os reativa sozinho. A tela de Dados canônicos mostra a coluna Situação e aceita o filtro enabled=true|false.

Exemplo: a base tem 9.000 clientes; na execução seguinte o ERP devolve 8.999 e CLI005401 não vem. Os 8.999 recebem o carimbo da execução; CLI005401 fica sem e é marcado inativo — sem ser apagado, então histórico e pedidos que o referenciam continuam intactos. Se semanas depois ele voltar a vir, o upsert o reativa.

"Ignorados" não é "não visto"

Um registro que veio igual ao que já estava conta como Ignorado nos números — mas ele foi gravado e recebeu o carimbo da execução. Por isso uma sincronização em que nada mudou (167 lidos / 0 gravados / 167 ignorados) não desativa ninguém: os 167 continuam "vistos".

É também por isso que o canônico é regravado mesmo sem mudança. Parece desperdício e não é: pular a gravação do inalterado faria a execução seguinte tratar a base inteira como ausente.

Não ligue em conector incremental

Se o conector lê só o que mudou (filtro por data/delta), "não veio" significa "não mudou" — e a desativação esvaziaria o catálogo. Por segurança, a desativação nunca roda quando a leitura foi interrompida, truncada no teto de páginas ou teve erro; nesses casos a execução registra um log deactivate explicando por que não desativou.

O que decide é a leitura, não a quantidade (ADR-061). A pergunta é sempre a mesma: a fonte parou de mandar por erro, ou respondeu 200 dizendo que não tem mais?

Como a execução terminou	O que acontece
Erro, timeout, truncada no teto de páginas	Não desativa nada — "não veio" aqui significa "a leitura parou"
Leitura completa, menos itens que antes	Desativa os ausentes, sem teto de proporção
Leitura completa, zero itens (200 com lista vazia)	Desativa tudo — a fonte foi vista inteira e não tem mais nada
A fonte mandou itens, mas o de-para extraiu zero	Não desativa nada — é itemsPath errado ou filtro de ingestão descartando tudo, não baixa de cadastro

A última linha é a que separa "o ERP não tem mais" de "nós não soubemos ler": sem ela, um itemsPath digitado errado zeraria o catálogo com ar de resposta legítima.

Filtro errado na origem some com o catálogo

Um $filter/$where mal escrito faz o ERP responder 200 com lista vazia — e isso é indistinguível de "não existe mais nada lá". A execução vai desativar tudo, registrando no log deactivate que a fonte respondeu sem itens.

Não é perda de dado: os registros ficam enabled = false, com histórico e pedidos intactos, e a execução seguinte que trouxer os registros os reativa sozinha. Mas o catálogo fica vazio na tela até lá — então, ao mexer na query de um conector com "Lê a base completa" ligado, confira o número de registros lidos na primeira execução antes de ir embora.

Teto de páginas e pageSize

A paginação para no fim da fonte ou num teto de maxPages (default 1000). Ao parar pelo teto, a execução termina PARTIAL, grava stats.truncated e mostra um aviso — antes, ler 20.000 de 50.000 registros terminava como "sucesso".

Cuidado ao aumentar o pageSize: se o servidor limita a página por conta própria (no SAP B1 é o header Prefer: odata.maxpagesize=N do Data Source), pedir $top=200 com limite de 20 devolve 20 — e como a regra de parada é "página incompleta = última", a leitura pararia na primeira página. Suba os dois juntos.

Sink canônico

A execução persiste o canônico genérico (EntityRecord) + o staging bruto (IntegrationRecord) — ver "Dados persistidos" abaixo; written = registros aplicados. O que ainda depende do módulo Comercial (E-4) é o upsert nas tabelas fixas de Clientes/Produtos (dados fiscais, contatos, endereços) — sem mudar este contrato.

Entidade alvo + de-para tipado (ADR-024)

A entidade alvo (targetEntity) é a entidade canônica que a operação alimenta. Ela deixou de ser texto livre: agora vem do metamodelo EntityDefinition (GET /integrations/entities), que unifica built-in (semeadas por org — client, product, priceTable, paymentCondition, paymentMethod, seller, order, orderStatus, receivable) e custom (definidas pelo admin — UI na fatia 2). Cada entidade traz seus campos (name + type), que guiam o de-para. Operações sem destino canônico (geocodificação, PDF, consulta de pagamento) podem ficar sem entidade.

O de-para (responseMapping) liga a resposta ao canônico. Além do legado fields (rename por dot-path), há rules[] tipadas: cada regra { targetField, source, type?, transforms? } mapeia um campo canônico ← um caminho na origem, com coerção de tipo e transforms (trim/lower/upper/cast_number/cast_bool/parse_date/default/split_list). O targetField é validado contra os campos da entidade. (JSONata + sink + mapping versionado ficam para fatias futuras.)

➡️ De-para (referência) — a página completa: template e filtros por campo, valor padrão, ordem dos transforms, flatten (1→N) e filter, com os exemplos do SAP B1.

O dry-run (POST /connectors/:id/dry-run) roda o de-para sobre uma amostra (viva ou enviada) e devolve, por registro, input→output + erros por campo, sem persistir — é o "ver como vai salvar" antes de executar. Na UI: botão Testar mapeamento no editor do conector.

No editor, o de-para já nasce preenchido com o canônico inteiro da entidade alvo (campos obrigatórios marcados com *); basta apontar a origem de cada um. O botão Descobrir campos (POST /connectors/:id/discover-schema) faz uma chamada viva e alimenta o autocomplete da origem com os caminhos (dot-path) reais da fonte. Preencher do canônico readiciona campos faltantes.

Campo de lista: o coringa * (ADR-069)

Quando a fonte devolve o dado em array de objetos e o campo canônico é uma lista (tipo json), use * para pegar o campo de todos os itens:

BPPaymentMethods.*.PaymentMethodCode   →  ["CR-BOL-BRA", "CR-DIN-CONV", "CR-TRANSF-CONV"]
BPPaymentMethods.0.PaymentMethodCode   →  "CR-BOL-BRA"   (só o primeiro)

Sem o coringa a única forma era o índice fixo — e a diferença é silenciosa: num cliente com 6 formas de pagamento, cinco ficavam de fora e nada no de-para dizia isso. O painel Schema da fonte oferece as duas variantes lado a lado (a de lista aparece como "lista de 3 (1º: …)"), para o coringa não depender de quem leu a documentação.

Regras, todas pensadas para ser previsíveis em vez de espertas:

Situação	Resultado
* sobre array	lista com o campo de cada item
* sobre objeto ou escalar	undefined — quem escreveu * espera lista; devolver um item só esconderia o erro de caminho
array vazio ([])	[] — "não tem nenhum" é diferente de "o campo não existe"
item sem o campo, ou com null	descartado da lista
item com 0, false ou ""	mantido — são respostas legítimas da fonte
* aninhado (a.*.b.*.c)	lista de listas; achatar fundiria dois níveis sem o caminho dizer isso

Mantenha o split_list na regra

Ele devolve array inalterado, e é ele que transforma "a fonte não trouxe a chave do array" em [] em vez de null.

Busca pontual: recorte por parâmetro e leitura completa (ADR-071)

O filter do de-para aceita {{params.x}} no valor, resolvido com os parâmetros da chamada:

{ "filter": [{ "path": "PriceList", "op": "eq", "value": "{{params.externalCode}}" }] }

Serve para recortar uma lista achatada quando a fonte não sabe filtrar: no SAP B1, o preço vem dentro de /Items em ItemPrices[], e o $filter do OData filtra o item, não os filhos — então o recorte só pode acontecer depois do flatten. Só o value é resolvido: path e op ficam literais, porque um path vindo de parâmetro deixaria quem chama escolher qual campo é comparado.

Parâmetro ausente devolve zero, não tudo

{{params.x}} sem valor vira string vazia, nada casa, e a resposta vem vazia. É a direção certa de falhar: devolver a base inteira para quem pediu um recorte faria o consumidor achar que a lista cresceu.

A resposta da busca pontual diz se a leitura terminou, e isso é parte do contrato:

{ "items": [...], "count": 167, "pages": 1, "statusCode": 200, "complete": true }

maxPages tem padrão 1. Quem usa esta rota para decidir ausência — "a lista veio sem o item X, então desative X" — só pode agir com complete: true. Com false, a resposta é um pedaço, e agir sobre ela desativa dado válido. É a mesma regra do freio de desativação na carga: a proteção é sobre a leitura, não sobre a quantidade.

Mudou o de-para? A próxima execução aplica (ADR-070)

Quando um registro chega idêntico ao que já está gravado, a execução não reescreve o payload — é o que faz uma sincronização sem novidade ser rápida. Mas "idêntico" é medido por duas coisas, não uma:

1. o conteúdo que a fonte mandou; e
2. o resultado que o de-para produz.

A segunda existe porque a saída é f(fonte, de-para). Sem ela, alterar o de-para e executar de novo não mudava nada: a fonte tinha mandado o mesmo, o registro era pulado, e o canônico ficava com o mapeamento antigo — com a execução terminando em SUCCESS, lendo tudo e gravando zero. Era o tipo de falha que só aparecia dias depois, quando alguém notava um campo vazio no consumidor.

Então, na prática: mexeu no de-para, execute o conector. Os registros cujo resultado mudou são regravados; os demais continuam no caminho rápido. E o contador de "gravados" conta exatamente isso — regravação de verdade, não leitura.

Ações em massa e limpeza de dados (ADR-067)

A lista de conectores tem seleção por linha e um menu de ações em massa: Sincronizar agora, Habilitar, Desabilitar e Limpar dados. Marcar o cabeçalho seleciona os visíveis — com busca ativa, o que está fora da tela não entra. As ações rodam uma a uma, não em paralelo: o ERP não aguentaria vinte sincronizações simultâneas, e o backend recusa execução concorrente do mesmo conector.

Limpar dados apaga o que o conector trouxe — o staging bruto e o canônico —, mantendo o conector, os agendamentos e (por padrão) o histórico de execuções. É operação própria porque não há cascata: os registros guardam o id do conector como texto, sem chave estrangeira, já que o canônico sobrevive ao conector que o alimentou. Sem isso, apagar o conector deixaria os dados órfãos no banco.

A confirmação mostra quantos registros serão apagados, por conector, e exige a palavra limpar digitada.

Para um conector só, não é preciso selecionar nada: o menu de três pontinhos (⋮) no fim da linha traz as duas ações destrutivas, escritas — Limpar dados e Apagar conector. Elas ficam ali, e não como ícones soltos, porque a diferença entre apagar os dados e apagar o conector não cabe num desenho: limpar deixa o conector e o de-para de pé, prontos para reintegrar; apagar não deixa nada.

Apagar o conector: o que vai e o que fica (ADR-068)

Excluir um conector não é o mesmo que apagar os dados dele. O modal de exclusão mostra as três coisas, com número:

	O quê	Por quê
vai junto	agendamentos e todo o histórico de execuções	caem por cascata da chave estrangeira
vai junto	os registros de staging	IntegrationRecord é chaveado por (orgId, connectorId, externalId) — pertence ao conector. Sem ele, ficariam inalcançáveis: fora da tela, impossíveis de reprocessar e sem nada que os referencie
fica	os registros canônicos	EntityRecord é chaveado por (orgId, entityKey, naturalKey) — pertence à entidade, não a quem a alimentou

O canônico sobreviver é decisão, não esquecimento: dá para apagar o sap-product e criar um sap-product-v2 apontando para outro endpoint sem evaporar o catálogo que outros sistemas já consomem. Quem quer o canônico fora também usa Limpar dados antes de excluir — depois de remover o conector não há mais por onde alcançá-lo, e o próprio modal avisa isso.

Apagar um data source segue a mesma regra. Os conectores dele caem por cascata do banco — que passa longe do código do conector —, então o staging de todos eles é apagado junto na mesma transação. Sem isso, a ação de raio maior órfanava mais dados do que a de raio menor.

O que a limpeza não desfaz

Registro que o ERP já não devolve não volta numa nova sincronização — se existia só aqui, se perde. E os dados que outros sistemas já consumiram (a base de quem lê o canônico) não são afetados: o SFA é a origem, não o destino.

Gerir as entidades canônicas

A aba Entidades (/dashboard/integrations) lista built-in + custom e permite criar/editar o canônico: cada campo tem name, type e as flags obrigatório (o de-para exige) e indexado (busca futura). Built-in têm key/origem fixas mas os campos são editáveis (evoluem com o negócio); custom podem ser removidas. RBAC: integration:configure para escrever.

Persistência do canônico

O dry-run não grava (é só conferência). A execução do conector, sim: persiste o staging bruto (IntegrationRecord) + o canônico (EntityRecord) — ver abaixo. O upsert nas tabelas fixas do Comercial (Clients/Products) segue como follow-up do E-4.

Dados persistidos: staging + canônico navegável

Ao executar um conector, a ingestão grava dois níveis (ADR-024 fatia 3):

• IntegrationRecord (staging bruto) — o payload original retido + o resultado do de-para + payloadHash (idempotência) + status (RECEIVED/MAPPED/APPLIED/FAILED). Reprocessar (POST /records/reprocess) re-aplica o mapeamento sobre o cru sem rechamar a fonte — ideal depois de ajustar o de-para. Na UI: ícone Dados no conector (ver bruto/canônico + reprocessar).
• EntityRecord (canônico) — o dado projetado, navegável e consultável por GET /entities/:key/records. Chaveado por naturalKey (coluna indexada, derivada de externalCode ou do 1º campo indexed); as chaves de junção (refKeys = campos marcados indexed) ganham índice GIN — é o que o Data Views (ADR-025) usa para join/filtro escalável. Na UI: ícone Dados na aba Entidades.

Segurança de segredos

• Segredos nunca em texto plano: cifrados em repouso (AES-256-GCM) e omitidos das respostas.
• Export/import (GET/POST /integrations/{export,import}) produz um IntegrationBundle portável sem segredos — cada credencial referencia uma env var por secretRef, resolvida e cifrada na importação do destino. Assim a config viaja versionada, a credencial não.
• Templating ({{params.x}}) não recebe segredos; a auth é aplicada por fora (headers/query).

Endpoints (Scalar)

Todas exigem organização ativa + RBAC (integration:read / integration:configure / integration:run). Recursos sob /api/integrations/*:

Recurso	Rotas
Credenciais	GET/POST /auth-configs, GET/PATCH/DELETE /auth-configs/:id
Data sources	GET/POST /data-sources, GET/PATCH/DELETE /data-sources/:id
Conectores	GET/POST /connectors, GET/PATCH/DELETE /connectors/:id, POST /connectors/:id/test, POST /connectors/:id/dry-run, POST /connectors/:id/discover-schema, POST /connectors/:id/run
Entidades	GET/POST /entities, GET/PATCH/DELETE /entities/:key (canônico built-in + custom)
Registros	GET /entities/:key/records (canônico), GET /records (staging), POST /records/reprocess
Agendamentos	GET/POST /schedules, GET/PATCH/DELETE /schedules/:id
Execuções	GET /sync-runs, GET /sync-runs/:id, GET /sync-runs/:id/records, GET /sync-runs/stream (SSE, ao vivo), POST /sync-runs/:id/reprocess
Export/Import	GET /integrations/export, POST /integrations/import

Pipelines também são portáveis: GET /pipelines/export e POST /pipelines/import.

Schemas completos na Referência da API (Scalar), tags Integrações · ….

Integrações de entrada/saída relacionadas

O runtime acima cobre a sincronização (puxar/empurrar dados via conector). Fluxos dirigidos por evento têm páginas próprias:

• Webhooks de entrada — recebem POSTs de fora (autenticados por API key ou HMAC) e disparam um pipeline, que pode acionar conectores (nós run-connector/call-connector) e chamar outras APIs (ADR-027/028).
• Webhooks de saída — o SFA notifica URLs externas quando um gatilho ocorre (ex.: sincronização concluída de um conector), com assinatura HMAC + retry + histórico.