Migração Estratégica de Dados: Implementando ETL e Repositório Local com Node.js e MongoDB

Problemas Reais e Implicações Técnicas

• Dependência total da Notion API: Hoje, meu blog consome exatamente a estrutura que o Notion expõe. Se amanhã essa API mudar o formato de resposta, alterar regras de acesso ou limitar endpoints, o conteúdo do blog fica comprometido. Isso significa que parte do meu produto depende diretamente de uma decisão de negócio externa. E isso nunca é um bom sinal a longo prazo.

• Tempo de resposta prejudicado: A cada requisição de conteúdo, o backend precisa buscar os dados no Notion primeiro — um caminho longo, sujeito a latência e instabilidade de rede.
Em um ambiente onde o usuário quer tudo praticamente instantâneo, isso se torna um gargalo para performance e SEO.

• Dados majoritariamente estáticos → falta de cache: Conteúdos de blog raramente mudam. Faz todo sentido que estejam armazenados e servidos de forma otimizada, sem depender de chamadas externas a cada acesso.

• Dificuldade para evoluir com novas funcionalidades: Recursos aparentemente simples, como adicionar um sistema de curtidas, tornam-se complexos quando os dados não estão no seu controle. Por exemplo:
• Escrever as curtidas diretamente no Notion permaneceríamos reféns do problema nº 1
• Criar um banco paralelo só para curtidas? Espalharia responsabilidades demais
Quando não há um modelo de dados próprio, cada feature vira uma gambiarra arquitetural.

Solução Proposta: Pipeline ETL + Banco Local (MongoDB)

O que é o ETL?

Como funciona esse algoritmo?

• Extract (extrair): Coletar dados das fontes externas — no meu caso, da Notion API.

• Transform (transformar) Padronizar os campos, enriquecer o conteúdo, limpar o que não é necessário e adaptar ao modelo interno.

• Load (carregar): Salvar esses dados já tratados em um banco de dados local (no meu caso MongoDB).

Imagine que você tem dados em uma planilha do Google (CSV) e quer exibi-los como JSON no seu site.

O ETL faria a leitura do CSV, organizaria os valores no formato desejado e os armazenaria localmente para consumo rápido.

Implementando o Código

Repository Pattern: O Core do ETL

• Leitura (get/query) – recuperar entidades ou coleções;

• Escrita (set/batch) – criar ou atualizar registros;

• Remoção (remove) – excluir entidades existentes.

export const URI = 'repository'
export type URI = typeof URI

export type RepositoryResult<E> = E | Promise<E>

export interface Repository<E extends Entity = any> extends Tagged<URI> {
  readonly get: (id: string) => RepositoryResult<E | undefined>
  readonly set: (entity: E) => RepositoryResult<E>
  readonly remove: (e: Identifier) => RepositoryResult<void>
  readonly query: (q?: Query<E>) => RepositoryResult<E[]>
  readonly batch: (b: Batch<E>) => RepositoryResult<WriteBatchResult>
}

#Case 1: ETL Post Data - Quando Webhook é Acionado

export const ETLPostData = Service<Request, Env, void>(
  ({ id }) =>
    async ({ posts, props, contents }) => {
    
	    // Extrair dados referentes ao ID informado
      const [properties, content, exists] = await Promise.all([
        props.get(id),
        contents.get(id),
        posts.query(Query.where('external_ref', '==', id)),
      ])

      const [post] = exists
			// Transformar os dados e decidir a intenção da ação
      const intent = decideIntent({ properties, content, post })

      if (intent.kind === 'none') return Left({ message: 'intent.kind none' })
		  // Gravar o efeito no repositório alvo
      else if (intent.kind === 'remove') await posts.remove(intent.data)
      else if (intent.kind === 'upsert') await posts.set(intent.data)

      return Right()
    },
)

• Extração paralela → menos latência e maior rendimento

• Processamento direcionado → o ETL só toca nos dados afetados

• Decisão inteligente de ação → elimina operações desnecessárias

• Independência da estrutura externa → quem dita as regras é o modelo local

• Consistência garantida → o banco local sempre reflete o estado atual da fonte

#Case 2: ETL Reconciler - Sincronização Periódica

export const ETLReconciler = Service(
  () =>
    async ({ posts, props, contents }: Env) => {
      /*
       * Buscar as informações do repositório do NotionAPI
       * Buscar as informações do repositório de Posts
       */
      const [listProps, listPosts] = await Promise.all([
        props.query(),
        posts.query(),
      ])

      // Fazer uma listagem única de external_ref presente nos dois sistemas (notionAPI e PostRepository)
      const uniqueRefs = [
        ...new Set(
          listProps
            .map(prop => prop.id)
            .concat(listPosts.map(post => post.external_ref)),
        ).values(),
      ]

      // Criar uma lista de intent.kind para cada external_ref único
      const intents = await Promise.all(
        uniqueRefs.map(async external_ref => {
          const post = listPosts.find(p => p.external_ref === external_ref)
          const properties = listProps.find(p => p.id === external_ref)
          const content = properties
            ? await contents.get(external_ref)
            : undefined

          const intent = decideIntent({ properties, content, post })

          return {
            id: external_ref,
            intent,
          }
        }),
      )

      // Mapear a transformação de acordo com cada intent.kind
      const batch = intents
        .filter(intent => intent.intent.kind !== 'none')
        .map(intent => ({
          type: intent.intent.kind,
          data: intent.intent.data,
        })) as Batch<Post>

      if (!batch.length)
        return Left({ message: 'The reconciler is not necessary' })

      // Realizar um batch no repositório de Posts
      const result = await posts.batch(batch)

      if (result.status === 'failed') return Left({ message: 'Batch failed' })

      return Right()
    },
)

• Alta confiabilidade → Nenhuma mudança externa permanece não identificada.

• Correção de inconsistências → Se algo falhar no webhook, o Reconciler corrige depois.

• Execução otimizada em lote → Menos chamadas desnecessárias e maior rendimento do banco.

• Controle total da integridade do modelo → O estado do sistema passa a ser previsível e sempre coerente.

• Escalabilidade da governança dos dados → Quanto mais o blog crescer, mais importante se torna ter dados sob domínio próprio.

A Fase de Transformação

• O dado deve ser criado no banco local?

• O dado já existe e precisa ser atualizado?

• O dado foi removido ou está inválido no Notion e deve ser excluído localmente?

• Ou simplesmente nada mudou?

export function decideIntent({
  properties,
  content,
  post,
}: {
  properties?: ExtractedPostProps
  content?: ExtractedPostContent
  post?: Post
}): IntentResult {
	// Se faltar qualquer tipo de dados externo, pode significar
	// remoção ou dados incompletos
  if (!properties || !content)
    return post ? kind('remove', { id: post.id }) : kind('none')

	// Caso o post não esteja publicado, isto pode significar que
	// o post foi despublicado ou ainda não foi publicado pela primeira vez
  if (!properties.published) {
    return post ? kind('remove', { id: post.id }) : kind('none')
  }

  const converted = convertExtractToPost({ properties, content, post })

	// Se não existir evidência do post no banco, então precisa ser criado
  if (!post) return kind('upsert', converted)

	// Caso haja evidência, deve-se verificar se foi atualizado para
	// então realizar uma atualização
  return isPostUpdated(post, converted)
    ? kind('upsert', converted)
    : kind('none')
}

export function convertExtractToPost({
  properties,
  content,
  post,
}: {
  properties: ExtractedPostProps
  content: ExtractedPostContent
  post?: Post
}): Post {
  const p = {
    slug: properties.slug,
    title: properties.title,
    description: properties.description,
    tags: properties.tags,
    content: content.content,
    estimated_readtime: estimateReadingTime(content.content),
    external_ref: properties.id,
    published: true,
    publish_at: properties.publish_at,
    created_at: properties.created_at,
    updated_at: properties.updated_at,
  }

  return post ? { ...post, ...p } : Post.create(p)
}

• Código único para transformação → aplicado no ETL via webhook e no ETL periódico

• Regras de negócio concentradas em um único ponto

• Segurança de que o modelo local sempre representa a verdade do sistema

• Evolução fácil: basta ajustar a transformação para mudanças futuras no conteúdo

• Independência da estrutura externa do Notion

Implementando MongoDB com Repository Pattern

export function MongoDBRepository<E extends Entity>({
  database,
  collection,
  converter,
  projection,
  ...rest
}: ClientConfig<E> | Config<E>): MongoDBRepository<E> {
  const client: MongoClient =
    (rest as any).client ?? new MongoClient((rest as any).uri)
  let coll: Collection<Document>
  let status = CONNECTION_STATUS.READY

  const connect: MongoDBRepository<E>['connect'] = async () => {
    if (status !== CONNECTION_STATUS.READY)
      throw new Error('Connection has been initialized')

    if (await isConnected(client)) {
      status = CONNECTION_STATUS.CONNECTED
      coll = client.db(database).collection(collection)

      return
    }

    const connection = await client.connect()

    coll = connection.db(database).collection(collection)
    status = CONNECTION_STATUS.CONNECTED
  }

  const disconnect = verifyConnectionProxy<MongoDBRepository<E>['disconnect']>(
    async () => {
      await client.close()
      status = CONNECTION_STATUS.DISCONNECTED
    },
  )

  const get = verifyConnectionProxy<Repository<E>['get']>(async id => {
    const item = await coll.findOne(
      { _id: ObjectId.createFromHexString(id) },
      { projection },
    )

    if (item === null) return

    return converter.from(item)
  })

  const set = verifyConnectionProxy<Repository<E>['set']>(async (entity: E) => {
    const { _id, ...props } = converter.to(entity)

    const result = await coll.updateOne(
      { _id: _id },
      { $set: props },
      { upsert: true },
    )

    return result.upsertedCount
      ? converter.from(
          await coll.findOne({ _id: result.upsertedId! }, { projection }),
        )
      : entity
  })

  const remove = verifyConnectionProxy<Repository<E>['remove']>(
    async ({ id }) => {
      await coll.deleteOne({ _id: ObjectId.createFromHexString(id) })
    },
  )

  const query = verifyConnectionProxy<Repository<E>['query']>(
    async ({ where, sorts, cursor, limit } = {}) => {
      let find = coll.find(where ? whereAdaptToFindQuery(where) : {})

      if (limit) {
        const skip = cursor ? parseInt(cursor) * limit : 0
        find = find.limit(limit).skip(skip)
      }

      if (sorts) {
        find = find.sort(applySorts(sorts))
      }

      if (projection) find.project(projection)

      return (await find.toArray()).map(converter.from)
    },
  )

  const batch = verifyConnectionProxy<Repository<E>['batch']>(async b => {
    const bulk = b.map(item => {
      if (item.type === 'remove')
        return {
          deleteOne: {
            filter: { _id: ObjectId.createFromHexString(item.data.id) },
          },
        }

      const { _id, ...props } = converter.to(item.data)

      return {
        updateOne: {
          filter: { _id },
          update: { $set: props },
          upsert: true,
        },
      }
    })

    const result = await coll.bulkWrite(bulk, { ordered: false })

    return {
      status: result.isOk() ? 'successful' : 'failed',
      time: new Date(),
    }
  })

  const clear = verifyConnectionProxy<MongoDBRepository<E>['clear']>(
    async () => {
      await coll.deleteMany({})
    },
  )

  function verifyConnectionProxy<F extends (...args: any[]) => any>(
    fn: F,
  ): (...args: Parameters<F>) => ReturnType<F> {
    return (...args) => {
      if (status !== CONNECTION_STATUS.CONNECTED)
        throw new Error('
        connection not established')

      return fn(...args)
    }
  }

  return applyTag('repository')({
    get,
    set,
    remove,
    query,
    batch,
    connect,
    disconnect,
    clear,
  })
}

• Batch operations no Reconciler → atualizações em massa e alta performance

• Converters para isolar o formato interno do formato do Mongo → domínio sem dependência do driver do banco

• Validação de conexão antes das operações → proteção e segurança de execução

• Queries flexíveis → base para futuras funcionalidades (paginadas, ordenadas, filtradas)

Resultados e benefícios da estratégia

• Aumento expressivo na performance das requisições: O backend agora consome dados diretamente do banco local, eliminando chamadas desnecessárias com serviços externos.

• Maior estabilidade e resiliência do sistema: Se a Notion API ficar lenta ou indisponível, o blog continua funcionando normalmente com os dados já sincronizados.

• Governança e flexibilidade sobre o modelo de dados: O formato dos dados agora segue necessidades do blog, e não limitações da API externa. Permite inovações como: curtidas, categorização própria, histórico de edição, KPIs de leitura etc.

• Menor custo de rede e processamento: A dependência de chamadas externas caiu drasticamente, reduzindo latência, riscos de timeout e uso excessivo de rede.

• Melhor SEO e experiência do usuário: Com respostas mais rápidas do backend, mecanismos de busca reconhecem a melhoria — o que impacta diretamente visibilidade e ranking.

• Arquitetura pronta para escalar: O pipeline ETL pode evoluir e absorver outras fontes no futuro sem reescrever a base da aplicação.

Riscos e cuidados

• Risco de desatualização dos dados locais: Se o ETL falhar ou ficar inoperante, o banco pode armazenar versões obsoletas dos conteúdos. Para resolver é necessário monitorar a integridade e validar com o Reconciler.

• Dependência da consistência entre múltiplas fontes: Agora existe uma duplicação de dados — e é preciso garantir alinhamento constante entre elas. Para isto, Testes automazidos, logs detalhados e alertas ajudam muito.

• Tratamento de mudanças no schema externo: APIs externas podem introduzir novos campos ou remover propriedades utilizadas no pipeline. O código deve ser resiliente a variações e falhas parciais.

• Volume de dados pode crescer com o tempo: Mesmo com conteúdo estático, versões acumulam histórico. Para resolver é necessário planejar indexação, arquivamento e limpeza periódica.

• Erros de transformação podem propagar dados inválidos: Uma conversão mal feita não só atrapalha funcionalidades, como pode quebrar a UI. Validação rigorosa na fase de transformação é fundamental.

Conclusão