@horizon-js/integrations-core

Core genérico para integrações com APIs externas. Define contratos (Source, Stream, HttpStream), sync modes, state management, pagination, retry, error handling. Agnóstico a domínio.

Inspirado no Airbyte CDK, adaptado pra TypeScript.

Instalação

pnpm add @horizon-js/integrations-core

Uso básico

import {
  HttpStream,
  SyncMode,
  hashObject,
  compareListings,
  SyncError,
  type Source,
  type SourceSpec,
  type ListingEntry,
} from "@horizon-js/integrations-core"

Exports principais

Types

• SyncMode / DestinationSyncMode — enums de sync mode
• SyncState / SourceState — state persistido entre syncs
• ListingEntry — {ref, updatedAt, hash?} pra delta
• SyncResult<T> — resultado agregado de uma sync
• SyncMetadataFields / WithSyncMeta<T> — convenção sync_hash, sync_version flat
• SourceMetadata / SourceType — metadata do vendor/adapter
• SourceCapabilities — paginação, timestamp, webhooks, writes granular, etc.
• KnownIssue — bugs/limitações documentadas do provider externo
• StreamDescriptor / WriterDescriptor — descritores no manifest
• WriteResult<T> / WriterOperation — resultado de operação de escrita
• AuthConfig — union legacy (use AuthStrategy pra novo código)

Errors

• FailureType — CONFIG_ERROR | USER_ERROR | TRANSIENT_ERROR | SYSTEM_ERROR
• SyncError — Error padronizado com helpers (configError, transientError, etc.) e toHttpStatus()
• shouldRetry(failureType) — helper pra decidir se retenta

Interfaces

• Source<Config> — connector concreto (com manifest obrigatório desde v0.2)
• SourceSpec<Config> / OAuthSpec — declaração de credenciais
• SourceManifest<Config> — cartão de identidade declarativo completo
• Stream<T> — contrato mínimo
• HttpStream<T> — APIs HTTP (com timeout/retry/rateLimit embutido)
• XmlStream<T> — feeds XML (NIFB-VRSync pattern)
• Writer<T> — contrato de escrita (create/update/delete/upsert)
• Webhook<Event> / WebhookHandlerResult — APIs com push events
• Client / ClientDescriptor — contrato do SDK HTTP consumido pelos sites
• IncrementalStream<T> / PaginatedStream<T> / PaginationStrategy
• SyncPipeline<T> / DestinationAdapter<T>

Helpers

• hashObject(value, options?) / hashRecords(array, getRef) — SHA-256 determinístico, 64/128/256 bits
• stableStringify(value) — JSON.stringify determinístico
• compareListings(source, dest) — diff {toCreate, toUpdate, toDelete, unchanged}
• retry(fn, options?) / httpRetryPolicy(options?) — exponential backoff + jitter
• RateLimiter — token bucket simples
• fetchWithTimeout(input, init) — fetch com AbortController
• parseResponseDefensive(items, convert, getRef?) — batch parse coletando erros
• validateManifest(m) — validação runtime do manifest
• profileRecords(records) — mapa {campo: valores_únicos} com dot-notation (descoberta pra IA)
• normalizeBoolean / normalizeTimestamp / normalizeNumber / parseCompositeAddress / parseGeoString

Auth Strategies

• AuthStrategy — interface com applyHeaders, applyQueryParams, refresh?
• BearerAuth — Authorization: Bearer <token>
• BearerRawAuth — Authorization: <token> sem prefixo (Arbo)
• NoAuth — fallback pra feeds públicos
• ApiKeyHeaderAuth — chave em header customizado (Smart, Tecimob)
• ApiKeyUrlAuth — chave em query param (Jetimob)
• DualHeaderAuth — 2 headers simultâneos (Imoview)
• OAuth2PasswordGrantAuth — password grant com cache+refresh automático (SI9)

Docs internas

• docs/philosophy.md — princípios arquiteturais
• docs/creating-new-source.md — passo-a-passo pra novo adapter
• docs/manifest-structure.md — como declarar manifest completo
• docs/sync-modes-guide.md — FULL_REFRESH vs INCREMENTAL + delta-via-hash
• docs/sync-metadata-guide.md — convenção sync_hash + source_*
• docs/pagination-guide.md — 4 patterns
• docs/error-handling-guide.md — FailureType + retry + rate limit
• docs/auth-strategies-guide.md — catálogo das 7 strategies com tabela de decisão
• docs/normalizers-guide.md — quando usar cada normalizer
• docs/xml-sources-guide.md — XmlStream pattern
• docs/writer-guide.md — contrato Writer + exemplos
• docs/client-sdk-pattern.md — subpath /client pro SDK
• docs/evolution-patterns.md — SemVer + backward-compat

Exemplo mínimo — criar um Source

import {
  HttpStream,
  SyncMode,
  type Source,
  type SourceSpec,
  type Stream,
} from "@horizon-js/integrations-core"

// 1. Declare credenciais
interface MyApiCredentials {
  apiKey: string
}

// 2. Spec
const mySpec: SourceSpec<MyApiCredentials> = {
  name: "my-api",
  title: "My API",
  configSchema: {
    type: "object",
    required: ["apiKey"],
    properties: {
      apiKey: { type: "string", airbyteSecret: true },
    },
  },
}

// 3. Stream
class MyPropertyStream extends HttpStream<MyProperty> {
  readonly name = "properties"
  readonly supportedSyncModes = [SyncMode.FULL_REFRESH, SyncMode.INCREMENTAL]
  readonly cursorField = "updated_at"
  readonly primaryKey = ["id"]

  constructor(private config: MyApiCredentials) { super() }

  get urlBase() { return "https://api.example.com/v1/" }
  path() { return "properties" }

  requestHeaders() {
    return { Authorization: `Bearer ${this.config.apiKey}` }
  }

  parseResponse(body: unknown): unknown[] {
    return (body as { data: unknown[] }).data
  }

  nextPageToken(response: { body: unknown }): string | undefined {
    return (response.body as { next?: string }).next
  }

  getJsonSchema() {
    return { type: "object", properties: { id: { type: "string" } } }
  }
}

// 4. Source
class MySource implements Source<MyApiCredentials> {
  readonly name = "my-api"
  readonly spec = mySpec

  async check(config: MyApiCredentials) {
    const r = await fetch("https://api.example.com/v1/ping", {
      headers: { Authorization: `Bearer ${config.apiKey}` },
    })
    return r.ok ? { ok: true } : { ok: false, message: `HTTP ${r.status}` }
  }

  streams(config: MyApiCredentials): Stream<unknown>[] {
    return [new MyPropertyStream(config)]
  }
}

Exemplo — delta via listing + hash (API sem updated_at)

import { hashObject, compareListings } from "@horizon-js/integrations-core"

// 1. Source listing com hash (calculado a partir do record convertido)
const sourceListing = await stream.getListing()
// → [{ ref: "438", updatedAt: null, hash: "a3f9..." }, ...]

// 2. Destination listing (do banco do site)
const destListing = await db.query<ListingEntry>(
  "SELECT ref, sync_hash as hash FROM properties"
)

// 3. Compara
const diff = compareListings(sourceListing, destListing)
// → { toCreate: [...], toUpdate: [...], toDelete: [...], unchanged: [...] }

// 4. Aplica
for (const ref of diff.toCreate) {
  const record = await stream.fetchByRef(ref)
  if (record) await db.insert({ ...record, sync_hash: hashObject(record) })
}

for (const ref of diff.toUpdate) {
  const record = await stream.fetchByRef(ref)
  if (record) await db.update(ref, { ...record, sync_hash: hashObject(record) })
}

for (const ref of diff.toDelete) {
  await db.delete(ref)
}

Design

• Zero dependências em runtime — bundle mínimo, sem lock-in
• Inspirado em Airbyte — vocabulário estabelecido, portável
• TypeScript idiomático — async generators, generics, interface + abstract class
• Hexagonal — core = ports, pacotes específicos = adapters

Veja docs/philosophy.md pra detalhes.

License

MIT — Horizon Modules