4ventures-client-portal/src/lib/pipeline-publish-worker.ts

/**
 * Shared worker: publish due creative_pipeline rows to Meta.
 * Used by cron and by schedule-save when the time is already due.
 */
import { serviceHeaders, supabaseUrl } from '@/lib/agency';
import {
  mergePermalinkNotes,
  publishPipelineItem,
  type PipelinePublishRow,
} from '@/lib/meta-publish';
import { normalizeChannels } from '@/lib/pipeline-schedule';
import { sanitizePublishCaption } from '@/lib/publish-caption';

const SELECT =
  'id,tenant_id,title,format,status,scheduled_at,channels,publish_status,publish_attempts,cover_path,cover_mime,slide_paths,caption,notes,meta_ig_media_id,meta_fb_post_id,updated_at';

async function patchItem(
  base: string,
  headers: Record<string, string>,
  id: string,
  body: Record<string, unknown>
) {
  return fetch(`${base}/rest/v1/creative_pipeline?id=eq.${id}`, {
    method: 'PATCH',
    headers: {
      ...headers,
      'Content-Type': 'application/json',
      Prefer: 'return=minimal',
    },
    body: JSON.stringify({ ...body, updated_at: new Date().toISOString() }),
  });
}

/** Recover rows stuck in publishing longer than `staleMin` minutes. */
async function recoverStalePublishing(
  base: string,
  headers: Record<string, string>,
  staleMin = 12
) {
  const cutoff = new Date(Date.now() - staleMin * 60_000).toISOString();
  await fetch(
    `${base}/rest/v1/creative_pipeline?publish_status=eq.publishing&updated_at=lt.${encodeURIComponent(cutoff)}`,
    {
      method: 'PATCH',
      headers: {
        ...headers,
        'Content-Type': 'application/json',
        Prefer: 'return=minimal',
      },
      body: JSON.stringify({
        publish_status: 'failed',
        publish_error:
          'Publicación interrumpida (timeout). Se reintentará en el siguiente ciclo.',
        updated_at: new Date().toISOString(),
      }),
    }
  );
}

export type DuePublishResult = {
  id: string;
  title: string;
  ok: boolean;
  detail: string;
};

/**
 * Fetch and publish due items.
 * Criteria: scheduled_at <= now, cover present, and
 *   - publish_status in (queued, failed), OR
 *   - publish_status=idle AND status=aprobada
 */
export async function processDuePipelinePublishes(opts?: {
  limit?: number;
  tenantId?: string;
  itemId?: string;
  dryRun?: boolean;
}): Promise<{
  ok: boolean;
  due: number;
  results: DuePublishResult[];
  error?: string;
  dry?: boolean;
  items?: unknown[];
}> {
  const headers = serviceHeaders();
  const base = supabaseUrl();
  if (!headers || !base) {
    return { ok: false, due: 0, results: [], error: 'Service role required' };
  }

  await recoverStalePublishing(base, headers);

  const now = new Date().toISOString();
  const limit = Math.min(50, Math.max(1, opts?.limit || 10));

  // Complementary queries (PostgREST OR with nested AND is fragile).
  // Incluye revision_agencia/programada en idle: a veces se programa
  // desde calendario sin pasar por el flujo que pone publish_status=queued.
  const common =
    `scheduled_at=lte.${encodeURIComponent(now)}` +
    `&cover_path=not.is.null` +
    `&published_at=is.null` +
    `&select=${SELECT}&order=scheduled_at.asc&limit=${limit}` +
    (opts?.tenantId
      ? `&tenant_id=eq.${encodeURIComponent(opts.tenantId)}`
      : '') +
    (opts?.itemId ? `&id=eq.${encodeURIComponent(opts.itemId)}` : '');

  const q1 =
    `${base}/rest/v1/creative_pipeline?publish_status=in.(queued,failed)&${common}`;
  const q2 =
    `${base}/rest/v1/creative_pipeline?publish_status=eq.idle&status=eq.aprobada&${common}`;
  const q3 =
    `${base}/rest/v1/creative_pipeline?publish_status=eq.idle&status=in.(revision_agencia,programada)&${common}`;

  const [r1, r2, r3] = await Promise.all([
    fetch(q1, { headers, cache: 'no-store' }),
    fetch(q2, { headers, cache: 'no-store' }),
    fetch(q3, { headers, cache: 'no-store' }),
  ]);
  if (!r1.ok || !r2.ok || !r3.ok) {
    const err = !r1.ok
      ? await r1.text()
      : !r2.ok
        ? await r2.text()
        : await r3.text();
    return { ok: false, due: 0, results: [], error: err.slice(0, 400) };
  }

  const a = (await r1.json()) as (PipelinePublishRow & {
    title: string;
    notes?: string | null;
    updated_at?: string;
  })[];
  const b = (await r2.json()) as typeof a;
  const c = (await r3.json()) as typeof a;
  const seen = new Set<string>();
  const due: typeof a = [];
  for (const row of [...a, ...b, ...c]) {
    if (seen.has(row.id)) continue;
    seen.add(row.id);
    due.push(row);
  }
  due.sort((x, y) =>
    String(
      (x as { scheduled_at?: string }).scheduled_at || ''
    ).localeCompare(
      String((y as { scheduled_at?: string }).scheduled_at || '')
    )
  );
  const slice = due.slice(0, limit);

  if (opts?.dryRun) {
    return {
      ok: true,
      dry: true,
      due: slice.length,
      results: [],
      items: slice.map((d) => ({
        id: d.id,
        title: d.title,
        scheduled_at: (d as { scheduled_at?: string }).scheduled_at,
        publish_status: (d as { publish_status?: string }).publish_status,
        status: (d as { status?: string }).status,
      })),
    };
  }

  const results: DuePublishResult[] = [];

  for (const due of slice) {
    let item: typeof due = due;
    const attempts = (item.publish_attempts || 0) + 1;
    const lock = await patchItem(base, headers, item.id, {
      publish_status: 'publishing',
      publish_attempts: attempts,
      publish_error: null,
      status: 'aprobada',
    });
    if (!lock.ok) {
      results.push({
        id: item.id,
        title: item.title,
        ok: false,
        detail: 'lock failed',
      });
      continue;
    }

    // Sanitize + persist clean caption before hitting Meta.
    const scrub = sanitizePublishCaption(item.caption || item.title || '', {
      maxLen: 6000,
    });
    const cleanCaption = scrub.caption || item.title || '';
    if (scrub.changed && cleanCaption !== (item.caption || '')) {
      await patchItem(base, headers, item.id, { caption: cleanCaption || null });
      item = { ...item, caption: cleanCaption };
    }

    // Publish IG then FB in two steps, persisting each Meta id immediately
    // so a crash / FB failure cannot create a duplicate IG on retry.
    const channels = normalizeChannels(item.channels);
    // Horarios por canal opcionales en notes: __4v_sched__:{"instagram":"...","facebook":"..."}
    let channelSched: { instagram?: string; facebook?: string } = {};
    const schedMatch = String(item.notes || '').match(
      /__4v_sched__:(\{[^}\n]+\})/
    );
    if (schedMatch) {
      try {
        channelSched = JSON.parse(schedMatch[1]) as typeof channelSched;
      } catch {
        channelSched = {};
      }
    }
    const nowMs = Date.now();
    const channelDue = (ch: 'instagram' | 'facebook') => {
      const iso = channelSched[ch];
      if (!iso) return true; // usa scheduled_at global (ya filtrado)
      const t = Date.parse(iso);
      return !Number.isNaN(t) && t <= nowMs;
    };
    const errors: string[] = [];
    let igMediaId: string | null = item.meta_ig_media_id || null;
    let fbPostId: string | null = item.meta_fb_post_id || null;
    let igPermalink: string | null = null;
    let fbPermalink: string | null = null;

    // LinkedIn-only: stub hasta API key (no pasa por Meta)
    if (
      channels.includes('linkedin') &&
      !channels.includes('instagram') &&
      !channels.includes('facebook')
    ) {
      const li = await publishPipelineItem(item);
      await patchItem(base, headers, item.id, {
        publish_status: li.ok ? 'published' : 'failed',
        publish_error: li.error || null,
        published_at: li.ok ? new Date().toISOString() : null,
        publish_attempts: (item.publish_attempts || 0) + 1,
      });
      results.push({
        id: item.id,
        title: item.title,
        ok: Boolean(li.ok),
        detail: li.ok ? 'linkedin' : li.error || 'linkedin_failed',
      });
      continue;
    }

    if (channels.includes('instagram') && !igMediaId && channelDue('instagram')) {
      const ig = await publishPipelineItem(item, { skipFb: true });
      if (ig.sanitizedCaption && ig.sanitizedCaption !== (item.caption || '')) {
        await patchItem(base, headers, item.id, {
          caption: ig.sanitizedCaption,
        });
        item = { ...item, caption: ig.sanitizedCaption };
      }
      if (ig.igMediaId) {
        igMediaId = ig.igMediaId;
        igPermalink = ig.igPermalink || null;
        await patchItem(base, headers, item.id, {
          meta_ig_media_id: igMediaId,
        });
        item = { ...item, meta_ig_media_id: igMediaId };
      }
      if (!ig.ok) {
        errors.push(ig.error || 'IG falló');
      }
    }

    if (
      channels.includes('facebook') &&
      !fbPostId &&
      errors.length === 0 &&
      channelDue('facebook')
    ) {
      const fb = await publishPipelineItem(
        { ...item, meta_ig_media_id: igMediaId },
        { skipIg: true }
      );
      if (fb.fbPostId) {
        fbPostId = fb.fbPostId;
        fbPermalink = fb.fbPermalink || null;
        await patchItem(base, headers, item.id, {
          meta_fb_post_id: fbPostId,
        });
        item = { ...item, meta_fb_post_id: fbPostId };
      }
      if (!fb.ok) {
        errors.push(fb.error || 'FB falló');
      }
    }

    const wantIg = channels.includes('instagram') && channelDue('instagram');
    const wantFb = channels.includes('facebook') && channelDue('facebook');
    // Si un canal aún no toca por horario, no marcar fallo: reintentará el cron.
    const igPending =
      channels.includes('instagram') && !channelDue('instagram') && !igMediaId;
    const fbPending =
      channels.includes('facebook') && !channelDue('facebook') && !fbPostId;
    if ((igPending || fbPending) && errors.length === 0) {
      await patchItem(base, headers, item.id, {
        publish_status: 'queued',
        publish_error: null,
        meta_ig_media_id: igMediaId,
        meta_fb_post_id: fbPostId,
      });
      results.push({
        id: item.id,
        title: item.title,
        ok: true,
        detail: 'partial_wait_channel_schedule',
      });
      continue;
    }
    const igOk = !wantIg || Boolean(igMediaId);
    const fbOk = !wantFb || Boolean(fbPostId);
    const ok = igOk && fbOk && errors.length === 0;

    if (!ok) {
      await patchItem(base, headers, item.id, {
        publish_status: 'failed',
        publish_error: errors.join(' · ') || 'Error Meta',
        meta_ig_media_id: igMediaId,
        meta_fb_post_id: fbPostId,
        caption: cleanCaption || null,
      });
      results.push({
        id: item.id,
        title: item.title,
        ok: false,
        detail: errors.join(' · ') || 'publish failed',
      });
      continue;
    }

    const notes = mergePermalinkNotes(item.notes, igPermalink, fbPermalink);
    await patchItem(base, headers, item.id, {
      publish_status: 'published',
      publish_error: null,
      published_at: new Date().toISOString(),
      status: 'publicada',
      meta_ig_media_id: igMediaId,
      meta_fb_post_id: fbPostId,
      caption: cleanCaption || null,
      notes,
    });
    results.push({
      id: item.id,
      title: item.title,
      ok: true,
      detail: `published ig=${igMediaId || '-'} fb=${fbPostId || '-'}`,
    });
  }

  return { ok: true, due: slice.length, results };
}