Primeros pasos con Fetcher

Dos caminos hacia una primera extracción con Fetcher: el arnés en memoria del Engine sin infraestructura, o el stack completo de Manager y Worker con Docker Compose.

Hay dos formas de ver a Fetcher funcionar. Elige una.

Camino	Lo que cuesta	Lo que demuestra
A — Engine embebido	Un go get. Sin infraestructura.	Reglas de extracción, planificación, límites y el resultado en modo directo.
B — Servicios independientes	Docker Compose. MongoDB, RabbitMQ, almacenamiento, Valkey.	La API REST, los jobs asíncronos y los resultados almacenados.

El camino A es la ruta más corta a un primer éxito. Empieza ahí si solo quieres evaluar el producto.

​

Camino A — Ejecuta el Engine sin infraestructura

---

El Engine incluye un arnés en memoria en pkg/engine/memory. Cubre los puertos de almacenamiento: el registro de conectores, el connection store, la caché de esquemas, el result sink y el execution store. No necesitas MongoDB, ni RabbitMQ, ni almacenamiento de objetos. No incluye un CredentialProtector, así que si activas la persistencia cifrada debes aportar el tuyo.

1. Agrega el módulo

   go get github.com/LerianStudio/fetcher/pkg/engine
   

   El Engine es un módulo separado de los servicios. No tiene dependencias de terceros, así que esta importación no arrastra nada más.

2. Construye, planifica, ejecuta

   package main
   
   import (
   	"context"
   	"fmt"
   	"log"
   
   	"github.com/LerianStudio/fetcher/pkg/engine"
   	"github.com/LerianStudio/fetcher/pkg/engine/memory"
   )
   
   func main() {
   	ctx := context.Background()
   
   	store := memory.NewConnectionStore()
   	registry := memory.NewConnectorRegistry()
   
   	// WithConnectorRegistry is the only required option.
   	eng, err := engine.New(
   		engine.WithConnectorRegistry(registry),
   		engine.WithConnectionStore(store),
   	)
   	if err != nil {
   		log.Fatal(err)
   	}
   
   	// Every operation is scoped to a tenant.
   	tenant, err := engine.NewTenantContext("tenant-123")
   	if err != nil {
   		log.Fatal(err)
   	}
   
   	conn := memory.NewTemplateConnector(memory.ConnectorBehavior{
   		Schema: engine.SchemaSnapshot{
   			ConfigName: "pg-main",
   			Tables:     []engine.TableSnapshot{{Name: "public.users", Fields: []string{"id", "email"}}},
   		},
   		Rows: map[string][]map[string]any{
   			"public.users": {{"id": 1, "email": "[e-mail]"}},
   		},
   	})
   	registry.Register("postgres", memory.NewConnectorFactory(conn))
   
   	if _, err = eng.CreateConnection(ctx, tenant, engine.NewConnectionInput(engine.ConnectionInputParams{
   		ConfigName: "pg-main",
   		Type:       "postgres",
   		Host:       "localhost",
   		Port:       5432,
   	})); err != nil {
   		log.Fatal(err)
   	}
   
   // Plan validates the request against a cache-first schema snapshot and enforces limits.
   	plan, err := eng.PlanExtraction(ctx, tenant, engine.ExtractionRequest{
   		MappedFields: map[string]engine.FieldSelection{
   			"pg-main": {"public.users": {"id", "email"}},
   		},
   	})
   	if err != nil {
   		log.Fatal(err)
   	}
   
   	result, err := eng.ExecuteExtraction(ctx, plan)
   	if err != nil {
   		log.Fatal(err)
   	}
   
   	fmt.Printf("rows=%d bytes=%d\n", result.Direct.RowCount, len(result.Direct.Data))
   }
   
3. Lee el resultado

   Aquí no hay ningún sink de resultados conectado, así que el Engine corre en modo directo. Devuelve las filas en línea como JSON indentado, más un digest SHA-256 sobre esos bytes exactos. Los bytes son deterministas: la misma entrada siempre produce el mismo digest.

Para pasar a producción, cambia el arnés en memoria por tus propios adaptadores. El Manager y el Worker de Fetcher son la implementación de referencia.

​

Camino B — Ejecuta los servicios independientes

---

Este camino te da la API REST y los jobs asíncronos. Todo corre localmente con Docker Compose.

​

Prerrequisitos

• Docker y Docker Compose
• Make
• Go, solo para desarrollo. La versión del toolchain está en el go.mod del repositorio.

​

Configura y ejecuta

1. Clona el repositorio

   git clone https://github.com/LerianStudio/fetcher.git
   cd fetcher
   
2. Crea los archivos de entorno

   make set-env
   

   Esto copia el .env.example de cada componente a .env.

3. Genera la clave maestra de cifrado

   make generate-master-key
   

   Copia la clave en APP_ENC_KEY en ambos components/manager/.env y components/worker/.env. Los dos servicios necesitan el mismo valor. El Worker la usa para descifrar credenciales y para verificar firmas de mensajes.

   Reemplaza el placeholder antes de arrancar: usa una clave válida de 32 bytes codificada en Base64. El placeholder creado por make set-env falla al decodificar una clave maestra Base64 inválida. El mensaje master key too short: got 0 bytes, minimum 32 required corresponde a un valor vacío o corto que sí se decodifica. Fetcher no tiene modo alternativo en texto plano.

4. Levanta todo

   make up
   
5. Verifica que la API responde

   • API REST: http://localhost:4006
   • Referencia de API en Scalar, cuando SWAGGER_ENABLED=true: http://localhost:4006/swagger/docs
   • Gestión de RabbitMQ: http://localhost:3008

​

Ejecuta tu primera extracción

Una extracción tiene tres movimientos. Registra una conexión, crea un job y después consulta el job.

​

1. Registra una conexión de base de datos

curl -X POST http://localhost:4006/v1/management/connections \
  -H "Content-Type: application/json" \
  -H "X-Product-Name: quickstart" \
  -d '{
    "configName": "my_postgres",
    "type": "POSTGRESQL",
    "host": "host.docker.internal",
    "port": 5432,
    "databaseName": "mydb",
    "userName": "postgres",
    "password": "postgres"
  }'

El header X-Product-Name nombra el producto dueño de la conexión. Usa el mismo valor en metadata.source del job en el paso 2, porque Fetcher compara los dos.

Fetcher cifra la contraseña antes de guardar el registro. Prueba la conexión antes de usarla:

curl -X POST http://localhost:4006/v1/management/connections/{id}/test

​

2. Crea un job de extracción

Nombra los campos que quieres, por tabla y por datasource:

curl -X POST http://localhost:4006/v1/fetcher \
  -H "Content-Type: application/json" \
  -d '{
    "dataRequest": {
      "mappedFields": {
        "my_postgres": {
          "accounts": ["id", "email", "created_at"]
        }
      }
    },
    "metadata": {
      "source": "quickstart"
    }
  }'

La API responde 202 Accepted con un ID de job. Si envías la misma petición dos veces dentro de 5 minutos, recibes 200 OK con el primer job en lugar de un segundo. Un job fallido no bloquea un reintento.

​

3. Consulta el job

curl http://localhost:4006/v1/fetcher/{id}

Un job termina en uno de dos estados terminales: completed o failed. Al completarse, el Worker ya cifró el resultado en el almacenamiento de objetos y publicó un evento job.completed. Los dos estados anteriores son pending y processing. Jobs de extracción trae el ciclo de vida completo de cuatro estados.

Activa la autenticación con PLUGIN_AUTH_ENABLED=true. Las peticiones llevan entonces un header Authorization: Bearer <token>. Este quickstart corre con la autenticación desactivada.

​

Próximos pasos

---

Conceptos centrales

Conexiones, descubrimiento de esquema, jobs, filtros y resultados.

Configuración

Todas las variables de entorno, por componente.

Despliegue

Dependencias, colas, escalado y las verificaciones de arranque que fallan cerrado.

Seguridad

Derivación de claves, rotación, firma de mensajes y validación de host.