Harden scheduled Meta publish and auto-link Drive for pipeline creatives.

Schedule save auto-approves and publishes when due; cron recovers stale jobs every 5m. Drive sync auto-matches client folders under Clientes when not linked.

Co-authored-by: Cursor <cursoragent@cursor.com>

src/lib/pipeline-publish-worker.ts

@@ -0,0 +1,217 @@
+/**
+ * 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';
+
+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=aprobado
+ */
+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));
+
+  // Two complementary queries (PostgREST OR with nested AND is fragile).
+  const common =
+    `scheduled_at=lte.${encodeURIComponent(now)}` +
+    `&cover_path=not.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.aprobado&${common}`;
+
+  const [r1, r2] = await Promise.all([
+    fetch(q1, { headers, cache: 'no-store' }),
+    fetch(q2, { headers, cache: 'no-store' }),
+  ]);
+  if (!r1.ok || !r2.ok) {
+    const err = !r1.ok ? await r1.text() : await r2.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 seen = new Set<string>();
+  const due: typeof a = [];
+  for (const row of [...a, ...b]) {
+    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 item of slice) {
+    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: 'aprobado',
+    });
+    if (!lock.ok) {
+      results.push({
+        id: item.id,
+        title: item.title,
+        ok: false,
+        detail: 'lock failed',
+      });
+      continue;
+    }
+
+    const result = await publishPipelineItem(item);
+    if (!result.ok) {
+      await patchItem(base, headers, item.id, {
+        publish_status: 'failed',
+        publish_error: result.error || 'Error Meta',
+        meta_ig_media_id: result.igMediaId || item.meta_ig_media_id || null,
+        meta_fb_post_id: result.fbPostId || item.meta_fb_post_id || null,
+      });
+      results.push({
+        id: item.id,
+        title: item.title,
+        ok: false,
+        detail: result.error || 'publish failed',
+      });
+      continue;
+    }
+
+    const notes = mergePermalinkNotes(
+      item.notes,
+      result.igPermalink,
+      result.fbPermalink
+    );
+    await patchItem(base, headers, item.id, {
+      publish_status: 'published',
+      publish_error: null,
+      published_at: new Date().toISOString(),
+      status: 'publicado',
+      meta_ig_media_id: result.igMediaId || null,
+      meta_fb_post_id: result.fbPostId || null,
+      notes,
+    });
+    results.push({
+      id: item.id,
+      title: item.title,
+      ok: true,
+      detail: `published ig=${result.igMediaId || '-'} fb=${result.fbPostId || '-'}`,
+    });
+  }
+
+  return { ok: true, due: slice.length, results };
+}