Sincronizar contactos con Fluyio

Esta guía separa dos tareas: importar la audiencia inicial y propagar después los cambios de tu sistema. Utiliza exclusivamente los endpoints de suscriptores publicados: POST, GET, PATCH, DELETE, subscribe y unsubscribe.

Necesitas una clave live de la marca con subscribers:read y subscribers:write. Las claves test devuelven 403 test_mode_unavailable; prueba los scripts con datos y una marca aislados.

Identidad y correspondencias

Fluyio genera un ID sub_…. Guarda tu ID estable en external_id para reconciliar registros aunque cambie el email. Un email duplicado al crear en la misma marca devuelve 409 resource_already_exists.

your_db.users.id  ──►  fluyio.subscribers.external_id  ──►  fluyio.subscribers.id
   user_12345                user_12345                          sub_9f1c2b3a…

No existe una operación única de crear o actualizar. Busca por filter[external_id] o filter[email]: si existe, usa PATCH; en caso contrario, POST. Utiliza una clave de idempotencia estable por operación.

Mapeo de datos

{
  "email": "[email protected]",
  "phone": "+447911123456",
  "external_id": "user_12345",
  "first_name": "Ada",
  "last_name": "Lovelace",
  "properties": {
    "plan": "pro",
    "signup_source": "organic",
    "city": "London"
  },
  "consent": {
    "email": "subscribed",
    "sms": "subscribed"
  }
}

• Email: subscribed solo con un alta explícita. Omitir el canal o indicar never_subscribed deja el estado pendiente; unsubscribed representa una baja.
• SMS: subscribed exige consentimiento y teléfono; unsubscribed conserva una baja conocida. Una identidad o permiso desconocidos no deben convertirse en alta.
• WhatsApp: admite consentimiento inicial explícito con teléfono E.164; al crear se registra origen api. Para documentar origen import, usa subscribe con ese origen cuando proceda.
• Listas y etiquetas: solo lectura en la API; se gestionan desde el panel.
• Propiedades: las claves con prefijo __ están reservadas.

Primera fase: importación

Exporta y conserva la evidencia

Obtén email, nombre, apellidos, ID propio, estados por canal y evidencia de origen y fecha. Conserva esta evidencia en tu sistema; no presupongas que el objeto público devuelve un historial completo.

Los ejemplos usan columnas booleanas: true significa alta acreditada, false baja conocida y null desconocido. Adapta esas columnas si tu origen distingue más estados o supresiones.

Calcula el ritmo

No hay endpoint masivo. Cada creación y cada consulta previa consumen una petición. El límite sostenido es de 600 peticiones por minuto y clave; 36.000 por hora es un techo teórico de peticiones, no una garantía de perfiles importados. Respeta también la ventana de 150 cada 10 segundos y Retry-After.

Ejecuta el importador

Los scripts leen PostgreSQL por lotes, limitan la concurrencia y reintentan 429. Ajusta CONCURRENCY a tus tiempos reales y monitoriza resultados. Revisa las dependencias del ejemplo antes de ejecutarlo.

// scripts/import-subscribers.ts
import { Client } from 'pg';
import pLimit from 'p-limit';

const BASE = 'https://fluyio.com/api/v1';
const HEADERS = {
  Authorization: `Bearer ${process.env.FLUYIO_API_KEY}`,
  'Fluyio-Version': '2026-05-15',
  'Content-Type': 'application/json',
};
const BATCH_SIZE = 500;
const CONCURRENCY = 8;

const db = new Client({ connectionString: process.env.DATABASE_URL });
const limit = pLimit(CONCURRENCY);

async function fluyio(method: string, path: string, body?: unknown, extra: Record<string, string> = {}) {
  for (let attempt = 0; attempt < 6; attempt++) {
    const res = await fetch(`${BASE}${path}`, {
      method,
      headers: { ...HEADERS, ...extra },
      body: body === undefined ? undefined : JSON.stringify(body),
    });
    if (res.status === 429) {
      const wait = Number(res.headers.get('retry-after') ?? '1') * 1000;
      await new Promise((r) => setTimeout(r, wait + Math.random() * 500));
      continue;
    }
    const json = await res.json();
    if (!res.ok) throw Object.assign(new Error(json.error?.message), { status: res.status, code: json.error?.code });
    return json;
  }
  throw new Error('rate limited too many times');
}

function toFluyio(row: any) {
  return {
    email: row.email.toLowerCase().trim(),
    phone: row.phone || undefined,
    external_id: `user_${row.id}`,
    first_name: row.first_name,
    last_name: row.last_name,
    properties: { plan: row.plan, city: row.city },
    consent: {
      email: row.email_opt_in === true ? 'subscribed' : row.email_opt_in === false ? 'unsubscribed' : 'never_subscribed',
      sms: row.phone && row.sms_opt_in === true ? 'subscribed' : row.sms_opt_in === false ? 'unsubscribed' : 'never_subscribed',
    },
  };
}

async function importOne(row: any) {
  const body = toFluyio(row);
  try {
    const sub = await fluyio('POST', '/subscribers', body, { 'Idempotency-Key': `import-${row.id}` });
    return { created: sub.id };
  } catch (err: any) {
    if (err.code === 'resource_already_exists') {
      const page = await fluyio('GET', `/subscribers?filter[email]=${encodeURIComponent(body.email)}&limit=1`);
      const existing = page.data[0];
      const { consent, ...patch } = body; // consent is not patchable
      await fluyio('PATCH', `/subscribers/${existing.id}`, patch);
      return { updated: existing.id };
    }
    if (err.code === 'profile_limit') throw err; // stop: plan limit reached
    console.error(`user ${row.id}: ${err.code} ${err.message}`);
    return { failed: row.id };
  }
}

async function* batchUsers() {
  await db.connect();
  let cursor = 0;
  while (true) {
    const { rows } = await db.query(
      `SELECT id, email, phone, first_name, last_name, plan, city, email_opt_in, sms_opt_in
         FROM users WHERE id > $1 ORDER BY id LIMIT $2`,
      [cursor, BATCH_SIZE],
    );
    if (rows.length === 0) break;
    cursor = rows[rows.length - 1].id;
    yield rows;
  }
  await db.end();
}

async function main() {
  let created = 0, updated = 0, failed = 0;
  for await (const batch of batchUsers()) {
    const results = await Promise.all(batch.map((row) => limit(() => importOne(row))));
    for (const r of results) {
      if ('created' in r) created++;
      else if ('updated' in r) updated++;
      else failed++;
    }
    console.log(`[${new Date().toISOString()}] created ${created}, updated ${updated}, failed ${failed}`);
  }
}

main().catch((err) => {
  console.error(err);
  process.exit(1);
});

DATABASE_URL=postgres://... FLUYIO_API_KEY=fly_live_... npx tsx scripts/import-subscribers.ts

Consentimiento de WhatsApp

Si necesitas registrar un alta acreditada con origen importación, utiliza POST /v1/subscribers/:id/subscribe con { "channel": "whatsapp", "source": "import" } para el número actual. No repitas altas a partir de una copia antigua que pueda ignorar bajas posteriores. El canal de envío sigue sujeto a beta privada.

Verifica el resultado

Revisa en el panel los perfiles creados, actualizados y fallidos. Resuelve duplicados y errores antes de dar por terminada la migración.

Segunda fase: sincronización de perfiles

Publica un trabajo en una cola cuando cambie un perfil y conserva el ID y la versión de origen para evitar trabajos duplicados.

Crear o actualizar

Estos ejemplos sincronizan identidad y propiedades. La creación incluye el estado acreditado del origen; para un perfil existente no se reactiva automáticamente un consentimiento que Fluyio ya retiró.

import { fluyio } from './fluyio-client'; // the helper from the import script

export async function upsertSubscriber(user: User) {
  const body = {
    email: user.email.toLowerCase().trim(),
    external_id: `user_${user.id}`,
    first_name: user.firstName,
    last_name: user.lastName,
    properties: { plan: user.plan, lifetime_value_cents: user.ltvCents },
  };

  const page = await fluyio('GET', `/subscribers?filter[external_id]=user_${user.id}&limit=1`);
  const existing = page.data[0];

  let subscriber;
  if (existing) {
    subscriber = await fluyio('PATCH', `/subscribers/${existing.id}`, body);
  } else {
    subscriber = await fluyio('POST', '/subscribers', {
      ...body,
      consent: { email: user.emailOptIn ? 'subscribed' : 'unsubscribed' },
    }, { 'Idempotency-Key': `upsert-${user.id}-${user.updatedAt.getTime()}` });
  }

  // Profile sync does not subscribe an existing profile automatically.
  // Process a new, verified consent event separately.
  return subscriber;
}

Integra la cola en tu aplicación

Una cola como BullMQ, Sidekiq, Celery o SQS permite regular el ritmo y separar errores de la petición del usuario. Agrupa trabajos por ID y versión del perfil.

// src/routes/users.ts
router.put('/users/:id', async (req, res) => {
  const user = await db.users.update(req.params.id, req.body);
  await queue.add('subscriber.sync', { userId: user.id }, {
    jobId: `sync-${user.id}-${user.updatedAt.getTime()}`, // dedupe in the queue
  });
  res.json(user);
});

// src/workers/subscriber-sync.ts
new Worker('subscriber.sync', async (job) => {
  const user = await db.users.findById(job.data.userId);
  if (!user) return;
  await upsertSubscriber(user);
}, { concurrency: 5 });

Cambios explícitos de consentimiento

Procesa las bajas como eventos propios mediante unsubscribe. Solo un evento nuevo y verificado de consentimiento debe invocar subscribe; compara fechas y origen para que una copia antigua no reactive una baja posterior.

Al cambiar email, el alta del email anterior no pasa al nuevo. Al cambiar teléfono tampoco se trasladan las altas SMS/WhatsApp. Las bajas y supresiones permanecen; intentar reactivar un rebote, una queja o un bloqueo devuelve 409 channel_unavailable.

De Fluyio a tu base de datos

Una baja desde el pie de un email puede ocurrir después de tu última copia local. Los webhooks salientes están pendientes; reconcilia mediante consultas periódicas:

// Nightly: mirror email consent back into your users table
for await (const sub of allSubscribers()) {          // the paginator from the pagination page
  if (!sub.external_id?.startsWith('user_')) continue;
  await db.users.updateById(sub.external_id.slice(5), {
    email_opt_in: sub.consent.email === 'subscribed',
  });
}

Con páginas de 100 registros, 100.000 perfiles requieren al menos 1.000 consultas. Planifica el tiempo con los límites, la latencia y los errores reales.

Preguntas frecuentes

¿Cómo tratar un cambio de email?

Busca por filter[external_id] y utiliza PATCH. No confíes en que la actualización use el mismo control previo de duplicados que la creación; resuelve primero la identidad en tu origen.

¿Cómo se normalizan los datos?

Los emails se recortan y convierten a minúsculas. Normaliza teléfonos a E.164, especialmente si registrarás consentimiento de WhatsApp.

¿Cuándo reintentar?

Ante 429 espera Retry-After. Ante 5xx o un resultado incierto de escritura, comprueba el estado. Si la operación ya fue reservada, la reserva de idempotencia dura 24 horas y no permite reejecución automática. Mantén la misma clave para la misma operación; no generes una nueva para forzar el reintento.

Un 503 idempotency_unavailable indica que no se inició la operación; no garantiza una reserva creada. Reintenta la misma operación con la misma clave, sin generar otra para eludir la protección.

¿Qué clave de idempotencia utilizar?

Usa una identificación estable de operación y versión, formada solo por caracteres permitidos y dentro de 8–128 caracteres. Evita email u otros valores con @, espacios o puntos. Los ejemplos import-ID suponen una ejecución con contenido estable; para otra versión del mismo perfil, define otra operación.

Referencias

• Suscriptores.
• Paginación.
• Límites de peticiones.
• Idempotencia.
• Webhooks pendientes.