From 26dc411647ae9c7e05dbd06a586a4f9fad648494 Mon Sep 17 00:00:00 2001 From: 4Ventures <4ventures@MacBook-Air-de-4Ventures.local> Date: Wed, 7 Oct 2026 13:33:41 +0200 Subject: [PATCH] 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 --- scripts/backfill_pipeline_drive.ts | 61 +++++ .../dokploy_ensure_pipeline_publish_cron.py | 167 ++++---------- src/app/api/creative-pipeline/route.ts | 59 ++++- src/app/api/cron/daily-sync/route.ts | 103 +-------- src/app/api/cron/pipeline-publish/route.ts | 180 ++------------- src/components/CreativePipelineBoard.tsx | 10 +- src/lib/google-drive.ts | 3 +- src/lib/pipeline-drive.ts | 146 +++++++++++- src/lib/pipeline-publish-worker.ts | 217 ++++++++++++++++++ 9 files changed, 531 insertions(+), 415 deletions(-) create mode 100644 scripts/backfill_pipeline_drive.ts create mode 100644 src/lib/pipeline-publish-worker.ts diff --git a/scripts/backfill_pipeline_drive.ts b/scripts/backfill_pipeline_drive.ts new file mode 100644 index 0000000..8cdc7c6 --- /dev/null +++ b/scripts/backfill_pipeline_drive.ts @@ -0,0 +1,61 @@ +/** + * Backfill pipeline covers/slides → client Drive folder tree. + * npx tsx scripts/backfill_pipeline_drive.ts [tenantSlug] + */ +import { config } from 'dotenv'; +import { resolve } from 'path'; +config({ path: resolve(process.cwd(), '.env.local') }); + +import { serviceHeaders, supabaseUrl } from '../src/lib/agency'; +import { backfillPipelineItemToDrive } from '../src/lib/pipeline-drive'; + +async function main() { + const slugFilter = process.argv[2]?.trim(); + const headers = serviceHeaders(); + const base = supabaseUrl(); + if (!headers || !base) throw new Error('service role missing'); + + const tRes = await fetch(`${base}/rest/v1/tenants?select=id,name,slug`, { + headers, + cache: 'no-store', + }); + let tenants = (await tRes.json()) as { + id: string; + name: string; + slug: string; + }[]; + if (slugFilter) { + tenants = tenants.filter( + (t) => t.slug === slugFilter || t.name.toLowerCase().includes(slugFilter.toLowerCase()) + ); + } + if (!tenants.length) throw new Error('no tenants'); + + for (const t of tenants) { + const pRes = await fetch( + `${base}/rest/v1/creative_pipeline?tenant_id=eq.${encodeURIComponent( + t.id + )}&cover_path=not.is.null&select=id,title,format,caption,notes,scheduled_at,created_at,cover_path,slide_paths&order=created_at.desc&limit=50`, + { headers, cache: 'no-store' } + ); + const pieces = pRes.ok ? await pRes.json() : []; + console.log(`\n=== ${t.name} (${pieces.length} con portada) ===`); + for (const item of pieces) { + const r = await backfillPipelineItemToDrive({ + tenantId: t.id, + clientLabel: t.slug || t.name, + item, + }); + console.log( + item.title, + r.ok ? 'OK' : 'PARTIAL', + r.results.map((x) => `${x.slot}:${x.detail}`).join(' | ') + ); + } + } +} + +main().catch((e) => { + console.error(e); + process.exit(1); +}); diff --git a/scripts/dokploy_ensure_pipeline_publish_cron.py b/scripts/dokploy_ensure_pipeline_publish_cron.py index bd4439d..a8cf384 100644 --- a/scripts/dokploy_ensure_pipeline_publish_cron.py +++ b/scripts/dokploy_ensure_pipeline_publish_cron.py @@ -1,14 +1,10 @@ #!/usr/bin/env python3 -""" -Ensure Dokploy schedule for pipeline Meta publish (every 10 min). -Uses application.createSchedule / findAll if API available; falls back to UI. -""" +"""Ensure / refresh Dokploy schedule for pipeline Meta publish (every 5 min).""" from __future__ import annotations import json import re import sys -import time from pathlib import Path from playwright.sync_api import sync_playwright @@ -17,18 +13,13 @@ ROOT = Path("/Users/4ventures/4ventures-client-portal") BPS = Path("/Users/4ventures/Projects/bps-facturacion") BASE = "https://dokploy.4ventures.es" APP_ID = "NZS797t9Rnq9Pj4mHiii3" -APP = ( - f"{BASE}/dashboard/project/Z_4M-uDOH1_--itGSTh4l" - f"/environment/bGja70l_a7HQ1dxyXqSuf/services/application/{APP_ID}" -) SCHEDULE_NAME = "pipeline-publish" -CRON_EXPR = "*/10 * * * *" -# Runs inside the app container where CRON_SECRET is already in env. +CRON_EXPR = "*/5 * * * *" +# Public URL + env expansion via bash -lc so CRON_SECRET from app env is used. CMD = ( - 'curl -sS -H "Authorization: Bearer $CRON_SECRET" ' - '"http://127.0.0.1:3000/api/cron/pipeline-publish" || ' - 'curl -sS -H "Authorization: Bearer $CRON_SECRET" ' - '"https://app.4ventures.es/api/cron/pipeline-publish"' + "bash -lc " + "'curl -sS -H \"Authorization: Bearer $CRON_SECRET\" " + "\"https://app.4ventures.es/api/cron/pipeline-publish?limit=20\"'" ) @@ -66,10 +57,9 @@ def main() -> int: page.wait_for_url(re.compile(r"/dashboard"), timeout=90000) print("login ok", flush=True) - # Probe Dokploy schedule APIs - probe = page.evaluate( + result = page.evaluate( """async ({ appId, name, cron, command }) => { - const tryFetch = async (path, body) => { + const post = async (path, body) => { const r = await fetch(path, { method: 'POST', credentials: 'include', @@ -79,43 +69,41 @@ def main() -> int: const text = await r.text(); let json = null; try { json = JSON.parse(text); } catch {} - return { ok: r.ok, status: r.status, path, text: text.slice(0, 500), json }; + return { ok: r.ok, status: r.status, text: text.slice(0, 1200), json }; }; - const listPaths = [ - '/api/trpc/schedule.all', - '/api/application.one', - ]; - const outs = {}; - // application.one to inspect shape - outs.one = await tryFetch('/api/application.one', { applicationId: appId }); - // Dokploy v0. something: schedule.create - const createBodies = [ - { applicationId: appId, name, cron, command, shellType: 'bash' }, - { applicationId: appId, scheduleName: name, cronExpression: cron, command, enabled: true }, - { applicationId: appId, name, cronExpression: cron, command }, - ]; - const createPaths = [ - '/api/schedule.create', - '/api/application.createSchedule', - '/api/trpc/schedule.create', - ]; - outs.creates = []; - for (const path of createPaths) { - for (const body of createBodies) { - const r = await tryFetch(path, body); - outs.creates.push(r); - if (r.ok) return { ok: true, via: path, r }; - } - } - // list schedules - for (const path of [ + + // Delete existing schedules with same name + for (const listPath of [ '/api/schedule.all', - '/api/application.schedules', '/api/schedule.byApplicationId', ]) { - outs['list_'+path] = await tryFetch(path, { applicationId: appId }); + const listed = await post(listPath, { applicationId: appId }); + const rows = Array.isArray(listed.json) + ? listed.json + : listed.json?.data || listed.json?.schedules || []; + if (Array.isArray(rows)) { + for (const row of rows) { + const n = row?.name || row?.scheduleName; + const id = row?.scheduleId || row?.id; + if (n === name && id) { + await post('/api/schedule.delete', { scheduleId: id }); + await post('/api/schedule.remove', { scheduleId: id }); + } + } + } } - return { ok: false, outs }; + + const created = await post('/api/schedule.create', { + applicationId: appId, + name, + cron, + cronExpression: cron, + command, + shellType: 'bash', + enabled: true, + scheduleType: 'application', + }); + return { created }; }""", { "appId": APP_ID, @@ -124,83 +112,10 @@ def main() -> int: "command": CMD, }, ) - print("API_PROBE", json.dumps(probe, indent=2)[:4000], flush=True) - - if probe.get("ok"): - print("SCHEDULE_OK_API", flush=True) - browser.close() - return 0 - - # UI fallback: Schedules tab - page.goto(APP, wait_until="domcontentloaded", timeout=90000) - page.wait_for_timeout(1000) - tab = page.get_by_role("tab", name=re.compile(r"Schedule", re.I)) - if not tab.count(): - # try Advanced / Domains area - for name in ("Schedules", "Cron", "Advanced"): - t = page.get_by_text(re.compile(rf"^{name}$", re.I)) - if t.count(): - t.first.click() - page.wait_for_timeout(600) - break - else: - tab.first.click() - page.wait_for_timeout(800) - - body = page.inner_text("body") - print("UI_BODY_SNIP", body[:1500].replace("\n", " | "), flush=True) - - if SCHEDULE_NAME in body or "pipeline-publish" in body: - print("SCHEDULE_ALREADY_PRESENT", flush=True) - browser.close() - return 0 - - add = page.get_by_role("button", name=re.compile(r"Add|Create|New|Schedule", re.I)) - if add.count(): - add.first.click() - page.wait_for_timeout(700) - # fill fields heuristically - inputs = page.locator("input, textarea") - n = inputs.count() - filled = 0 - for i in range(min(n, 12)): - el = inputs.nth(i) - try: - typ = (el.get_attribute("type") or "").lower() - name = (el.get_attribute("name") or el.get_attribute("placeholder") or "").lower() - if typ == "hidden": - continue - if "name" in name or filled == 0: - el.fill(SCHEDULE_NAME) - filled += 1 - elif "cron" in name or filled == 1: - el.fill(CRON_EXPR) - filled += 1 - elif "command" in name or "cmd" in name or filled == 2: - el.fill(CMD) - filled += 1 - except Exception: - pass - save = page.get_by_role("button", name=re.compile(r"Save|Create|Confirm|Add", re.I)) - if save.count(): - save.first.click() - page.wait_for_timeout(1200) - print("SCHEDULE_UI_SAVED", flush=True) - browser.close() - return 0 - - print("SCHEDULE_NEEDS_MANUAL", flush=True) - # Persist ops note - note = ROOT / ".ops" / "pipeline-publish-cron.txt" - note.write_text( - f"Schedule needed on Dokploy app {APP_ID}:\n" - f" name: {SCHEDULE_NAME}\n" - f" cron: {CRON_EXPR}\n" - f" command:\n{CMD}\n", - encoding="utf-8", - ) + print(json.dumps(result, indent=2)[:4000], flush=True) + ok = bool(result.get("created", {}).get("ok")) browser.close() - return 0 # non-fatal: docs + note written + return 0 if ok else 1 if __name__ == "__main__": diff --git a/src/app/api/creative-pipeline/route.ts b/src/app/api/creative-pipeline/route.ts index 5963e13..d70014e 100644 --- a/src/app/api/creative-pipeline/route.ts +++ b/src/app/api/creative-pipeline/route.ts @@ -20,6 +20,7 @@ import { scheduledAtFromPiece, type CalendarDoc, } from '@/lib/calendar-documents'; +import { processDuePipelinePublishes } from '@/lib/pipeline-publish-worker'; const SELECT = 'id,external_ref,title,format,month,status,cover_path,cover_mime,slide_paths,caption,notes,approved_at,created_at,updated_at,scheduled_at,channels,publish_status,publish_attempts,published_at,meta_ig_media_id,meta_fb_post_id,publish_error,calendar_piece_id'; @@ -381,15 +382,15 @@ export async function PATCH(req: NextRequest) { } clean.scheduled_at = parsed; if (parsed) { - // Re-arm worker if agency reschedules after a failure. - if ( - current === 'aprobado' || - clean.status === 'aprobado' || - !clean.status - ) { - clean.publish_status = 'queued'; - clean.publish_error = null; + // Always arm the worker + ensure editorial status is aprobado. + // (Cron only publishes queued/aprobado due items.) + if (current !== 'publicado' && clean.status !== 'publicado') { + clean.status = 'aprobado'; + clean.approved_by = user.id; + clean.approved_at = new Date().toISOString(); } + clean.publish_status = 'queued'; + clean.publish_error = null; } else { clean.publish_status = 'idle'; clean.publish_error = null; @@ -431,7 +432,47 @@ export async function PATCH(req: NextRequest) { return NextResponse.json({ error: await r.text() }, { status: 500 }); } const data = await r.json(); - return NextResponse.json({ item: Array.isArray(data) ? data[0] : data }); + let item = Array.isArray(data) ? data[0] : data; + + // If schedule is already due (or within 90s), publish in this request + // so we don't depend solely on the Dokploy cron tick. + if ( + agency && + item && + typeof item.scheduled_at === 'string' && + item.publish_status === 'queued' && + item.cover_path + ) { + const dueAt = Date.parse(item.scheduled_at); + if (!Number.isNaN(dueAt) && dueAt <= Date.now() + 90_000) { + const pub = await processDuePipelinePublishes({ + limit: 1, + tenantId: ctx.tenantId, + itemId: id, + }); + const hit = pub.results.find((x) => x.id === item.id); + if (hit) { + // reload row + const again = await fetch( + `${supabaseUrl()}/rest/v1/creative_pipeline?id=eq.${encodeURIComponent( + id + )}&select=${SELECT}&limit=1`, + { headers, cache: 'no-store' } + ); + if (again.ok) { + const rows = await again.json(); + if (rows?.[0]) item = rows[0]; + } + return NextResponse.json({ + item, + publishedNow: hit.ok, + publishDetail: hit.detail, + }); + } + } + } + + return NextResponse.json({ item }); } export async function DELETE(req: NextRequest) { diff --git a/src/app/api/cron/daily-sync/route.ts b/src/app/api/cron/daily-sync/route.ts index 95b1b83..0739515 100644 --- a/src/app/api/cron/daily-sync/route.ts +++ b/src/app/api/cron/daily-sync/route.ts @@ -9,110 +9,11 @@ import { } from '@/lib/superhuman-docs'; import { syncAllClientifyTenants } from '@/lib/clientify-sync'; import { syncAllGhlTenants } from '@/lib/gohighlevel-sync'; -import { serviceHeaders, supabaseUrl } from '@/lib/agency'; -import { - mergePermalinkNotes, - publishPipelineItem, - type PipelinePublishRow, -} from '@/lib/meta-publish'; +import { processDuePipelinePublishes } from '@/lib/pipeline-publish-worker'; export const dynamic = 'force-dynamic'; export const maxDuration = 300; -const PIPELINE_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'; - -async function runDuePipelinePublish(limit = 5): Promise> { - const headers = serviceHeaders(); - const base = supabaseUrl(); - if (!headers || !base) { - return { ok: false, error: 'Service role required' }; - } - const now = new Date().toISOString(); - const url = - `${base}/rest/v1/creative_pipeline?status=eq.aprobado` + - `&scheduled_at=lte.${encodeURIComponent(now)}` + - `&publish_status=in.(idle,queued,failed)` + - `&select=${PIPELINE_SELECT}&order=scheduled_at.asc&limit=${limit}`; - const list = await fetch(url, { headers, cache: 'no-store' }); - if (!list.ok) { - return { ok: false, error: (await list.text()).slice(0, 200) }; - } - const due = (await list.json()) as (PipelinePublishRow & { - title: string; - notes?: string | null; - })[]; - const results: { id: string; ok: boolean; detail: string }[] = []; - for (const item of due) { - if (!item.cover_path) { - results.push({ id: item.id, ok: false, detail: 'missing cover' }); - continue; - } - await fetch(`${base}/rest/v1/creative_pipeline?id=eq.${item.id}`, { - method: 'PATCH', - headers: { - ...headers, - 'Content-Type': 'application/json', - Prefer: 'return=minimal', - }, - body: JSON.stringify({ - publish_status: 'publishing', - publish_attempts: (item.publish_attempts || 0) + 1, - publish_error: null, - updated_at: new Date().toISOString(), - }), - }); - const result = await publishPipelineItem(item); - if (!result.ok) { - await fetch(`${base}/rest/v1/creative_pipeline?id=eq.${item.id}`, { - method: 'PATCH', - headers: { - ...headers, - 'Content-Type': 'application/json', - Prefer: 'return=minimal', - }, - body: JSON.stringify({ - 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, - updated_at: new Date().toISOString(), - }), - }); - results.push({ - id: item.id, - ok: false, - detail: result.error || 'failed', - }); - continue; - } - await fetch(`${base}/rest/v1/creative_pipeline?id=eq.${item.id}`, { - method: 'PATCH', - headers: { - ...headers, - 'Content-Type': 'application/json', - Prefer: 'return=minimal', - }, - body: JSON.stringify({ - 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: mergePermalinkNotes( - item.notes, - result.igPermalink, - result.fbPermalink - ), - updated_at: new Date().toISOString(), - }), - }); - results.push({ id: item.id, ok: true, detail: 'published' }); - } - return { ok: true, due: due.length, results }; -} - /** * Master daily job: organic + ads + superhuman. * Dokploy (daily ~06:00 UTC): @@ -213,7 +114,7 @@ export async function GET(req: NextRequest) { // 5) Pipeline Meta publish (safety net; prefer dedicated */10 cron) try { - out.pipelinePublish = await runDuePipelinePublish(5); + out.pipelinePublish = await processDuePipelinePublishes({ limit: 5 }); } catch (e) { out.pipelinePublish = { ok: false, diff --git a/src/app/api/cron/pipeline-publish/route.ts b/src/app/api/cron/pipeline-publish/route.ts index 3e33662..17b4023 100644 --- a/src/app/api/cron/pipeline-publish/route.ts +++ b/src/app/api/cron/pipeline-publish/route.ts @@ -1,185 +1,37 @@ /** - * Cron: pick creative_pipeline rows that are aprobado + scheduled_at due, - * publish to Meta IG+FB via Content Publishing API. + * Cron: publish due creative_pipeline rows to Meta IG+FB. * * Auth: Authorization: Bearer $CRON_SECRET * curl -sS -H "Authorization: Bearer $CRON_SECRET" \ * "https://app.4ventures.es/api/cron/pipeline-publish" */ import { NextRequest, NextResponse } from 'next/server'; -import { serviceHeaders, supabaseUrl } from '@/lib/agency'; -import { - mergePermalinkNotes, - publishPipelineItem, - type PipelinePublishRow, -} from '@/lib/meta-publish'; +import { assertCronAuth } from '@/lib/cron-auth'; +import { processDuePipelinePublishes } from '@/lib/pipeline-publish-worker'; export const dynamic = 'force-dynamic'; export const runtime = 'nodejs'; export const maxDuration = 120; -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'; - -function authorize(req: NextRequest): boolean { - const secret = process.env.CRON_SECRET?.trim(); - if (!secret) return false; - const auth = req.headers.get('authorization') || ''; - if (auth === `Bearer ${secret}`) return true; - const q = req.nextUrl.searchParams.get('secret'); - return q === secret; -} - -async function patchItem( - base: string, - headers: Record, - id: string, - body: Record -) { - 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() }), - }); -} - export async function GET(req: NextRequest) { - if (!authorize(req)) { - return NextResponse.json({ error: 'Unauthorized' }, { status: 401 }); - } + const denied = assertCronAuth(req); + if (denied) return denied; - const headers = serviceHeaders(); - const base = supabaseUrl(); - if (!headers || !base) { - return NextResponse.json({ error: 'Service role required' }, { status: 503 }); - } - - const now = new Date().toISOString(); - const limit = Math.min( - 50, - Math.max(1, Number(req.nextUrl.searchParams.get('limit') || 10)) - ); + const limit = Number(req.nextUrl.searchParams.get('limit') || 10); const dryRun = req.nextUrl.searchParams.get('dry') === '1'; + const tenantId = + req.nextUrl.searchParams.get('tenantId')?.trim() || undefined; - const url = - `${base}/rest/v1/creative_pipeline?status=eq.aprobado` + - `&scheduled_at=lte.${encodeURIComponent(now)}` + - `&publish_status=in.(idle,queued,failed)` + - `&select=${SELECT}&order=scheduled_at.asc&limit=${limit}`; - - const list = await fetch(url, { headers, cache: 'no-store' }); - if (!list.ok) { + const out = await processDuePipelinePublishes({ limit, dryRun, tenantId }); + if (!out.ok && out.error) { return NextResponse.json( - { error: (await list.text()).slice(0, 400) }, + { error: out.error }, { status: 500 } ); } - const due = (await list.json()) as (PipelinePublishRow & { - title: string; - publish_status: string; - notes?: string | null; - })[]; - - if (dryRun) { - return NextResponse.json({ - ok: true, - dry: true, - due: due.length, - items: due.map((d) => ({ - id: d.id, - title: d.title, - scheduled_at: (d as { scheduled_at?: string }).scheduled_at, - channels: d.channels, - attempts: d.publish_attempts, - })), - }); - } - - const results: { - id: string; - title: string; - ok: boolean; - detail: string; - }[] = []; - - for (const item of due) { - if (!item.cover_path) { - await patchItem(base, headers, item.id, { - publish_status: 'failed', - publish_error: 'Sin portada — no se puede publicar', - publish_attempts: (item.publish_attempts || 0) + 1, - }); - results.push({ - id: item.id, - title: item.title, - ok: false, - detail: 'missing cover', - }); - continue; - } - - const attempts = (item.publish_attempts || 0) + 1; - const lock = await patchItem(base, headers, item.id, { - publish_status: 'publishing', - publish_attempts: attempts, - publish_error: null, - }); - 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 NextResponse.json({ - ok: true, - processed: results.length, - results, - }); + return NextResponse.json(out); +} + +export async function POST(req: NextRequest) { + return GET(req); } diff --git a/src/components/CreativePipelineBoard.tsx b/src/components/CreativePipelineBoard.tsx index 7c370f1..1afd1e2 100644 --- a/src/components/CreativePipelineBoard.tsx +++ b/src/components/CreativePipelineBoard.tsx @@ -1876,9 +1876,13 @@ function DetailModal({

- Publicar ahora sale al momento. Con fecha futura + - estado aprobado, el cron publica IG+FB al unísono. - Si ya está en redes y está mal: bórralo aquí y vuelve a publicar. + Publicar ahora sale al momento.{' '} + Guardar programación marca la pieza como + aprobada y la encola: si la hora ya pasó (o es en menos de 2 + min) publica al guardar; si es futura, el cron la saca cada 10 + min. + Si ya está en redes y está mal: bórralo aquí y vuelve a + publicar.

) : null} diff --git a/src/lib/google-drive.ts b/src/lib/google-drive.ts index 6f95511..bea34a8 100644 --- a/src/lib/google-drive.ts +++ b/src/lib/google-drive.ts @@ -663,9 +663,10 @@ export async function ensureChildFolder( const params = new URLSearchParams({ q, fields: 'files(id,name)', - pageSize: '1', + pageSize: '5', supportsAllDrives: 'true', includeItemsFromAllDrives: 'true', + corpora: 'allDrives', }); const find = await fetch( `https://www.googleapis.com/drive/v3/files?${params}`, diff --git a/src/lib/pipeline-drive.ts b/src/lib/pipeline-drive.ts index 93e14b3..37e7115 100644 --- a/src/lib/pipeline-drive.ts +++ b/src/lib/pipeline-drive.ts @@ -9,14 +9,18 @@ */ import { slugify } from '@/lib/calendar-import'; import { + applyMatches, driveConfigured, ensureChildFolder, getTenantDriveFolderId, + listClientFoldersUnderClientes, listFilesInFolder, + matchFoldersToTenants, uploadToDrive, getValidAccessToken, } from '@/lib/google-drive'; import { serviceHeaders, supabaseUrl } from '@/lib/agency'; +import { ASSETS_BUCKET, serviceSupabase } from '@/lib/assets'; export type PipelineDriveItem = { id: string; @@ -26,6 +30,8 @@ export type PipelineDriveItem = { notes?: string | null; scheduled_at?: string | null; created_at?: string | null; + cover_path?: string | null; + slide_paths?: unknown; }; function pad2(n: number): string { @@ -56,7 +62,6 @@ export function hookLabel(item: PipelineDriveItem): string { export function formatDriveFileName(opts: { item: PipelineDriveItem; ext: string; - /** 1-based slide number for carrusel; omit for single creatives */ slideNumber?: number; }): string { const d = pieceLocalDate(opts.item); @@ -122,7 +127,10 @@ async function uploadOrReplace(opts: { return { ...created, replaced: false }; } -async function loadTenantSlug(tenantId: string): Promise { +async function loadTenant(tenantId: string): Promise<{ + slug: string; + name: string; +} | null> { const headers = serviceHeaders(); if (!headers) return null; const r = await fetch( @@ -133,12 +141,46 @@ async function loadTenantSlug(tenantId: string): Promise { ); if (!r.ok) return null; const rows = (await r.json()) as { slug?: string; name?: string }[]; - return rows[0]?.slug || rows[0]?.name || null; + if (!rows[0]) return null; + return { + slug: rows[0].slug || rows[0].name || 'cliente', + name: rows[0].name || rows[0].slug || 'cliente', + }; +} + +/** + * Resolve client Drive folder: linked integration, or auto-match under «Clientes». + */ +export async function resolveOrLinkTenantDriveFolder( + tenantId: string, + clientLabel: string +): Promise<{ folderId: string; folderName?: string; autoLinked?: boolean } | null> { + const linked = await getTenantDriveFolderId(tenantId); + if (linked?.folderId) return linked; + + const tenant = await loadTenant(tenantId); + if (!tenant) return null; + + const { folders } = await listClientFoldersUnderClientes(); + if (!folders.length) return null; + + const matches = await matchFoldersToTenants( + [{ id: tenantId, name: tenant.name || clientLabel, slug: tenant.slug }], + folders + ); + const hit = matches.find((m) => m.status === 'matched' && m.folderId); + if (!hit?.folderId) return null; + + await applyMatches([hit]); + return { + folderId: hit.folderId, + folderName: hit.folderName || undefined, + autoLinked: true, + }; } /** * Ensure month folder under client's Drive root and upload the creative. - * Non-fatal if Drive is not linked — returns warning. */ export async function syncPipelineAssetToDrive(opts: { tenantId: string; @@ -156,6 +198,7 @@ export async function syncPipelineAssetToDrive(opts: { driveFileId?: string; drivePath?: string; fileName?: string; + autoLinked?: boolean; }> { if (!driveConfigured()) { return { @@ -165,20 +208,40 @@ export async function syncPipelineAssetToDrive(opts: { }; } - const linked = await getTenantDriveFolderId(opts.tenantId); + let linked: { + folderId: string; + folderName?: string; + autoLinked?: boolean; + } | null = null; + try { + linked = await resolveOrLinkTenantDriveFolder( + opts.tenantId, + opts.clientLabel + ); + } catch (e) { + return { + ok: false, + warning: + e instanceof Error + ? `Drive auto-link: ${e.message}` + : 'Drive auto-link falló', + }; + } + if (!linked?.folderId) { return { ok: false, skipped: true, warning: - 'Cliente sin carpeta Drive asociada. Empareja Drive en Agencia / ficha.', + 'Cliente sin carpeta Drive. Empareja Drive en Agencia (o crea carpeta bajo 03-Clientes).', }; } try { - const slug = - (await loadTenantSlug(opts.tenantId)) || opts.clientLabel || 'cliente'; - const rootName = pipelineRootFolderName(slug); + const tenant = await loadTenant(opts.tenantId); + const rootName = pipelineRootFolderName( + tenant?.slug || opts.clientLabel || 'cliente' + ); const d = pieceLocalDate(opts.item); const year = String(d.getFullYear()); const month = pad2(d.getMonth() + 1); @@ -190,7 +253,6 @@ export async function syncPipelineAssetToDrive(opts: { const isCarousel = (opts.item.format || '').toLowerCase() === 'carrusel'; let slideNumber: number | undefined; if (isCarousel) { - // cover → 1; slide index 0 → 2, etc. slideNumber = opts.slot === 'cover' ? 1 : Math.max(2, opts.slideIndex + 2); } @@ -212,7 +274,8 @@ export async function syncPipelineAssetToDrive(opts: { ok: true, driveFileId: up.id, fileName, - drivePath: `${rootName}/${year}/${month}/${fileName}`, + drivePath: `${linked.folderName || 'cliente'}/${rootName}/${year}/${month}/${fileName}`, + autoLinked: linked.autoLinked, }; } catch (e) { return { @@ -221,3 +284,64 @@ export async function syncPipelineAssetToDrive(opts: { }; } } + +/** Backfill: push an already-stored cover/slide from Storage → Drive. */ +export async function backfillPipelineItemToDrive(opts: { + tenantId: string; + clientLabel: string; + item: PipelineDriveItem; +}): Promise<{ ok: boolean; results: { slot: string; ok: boolean; detail: string }[] }> { + const svc = serviceSupabase(); + const results: { slot: string; ok: boolean; detail: string }[] = []; + if (!svc) { + return { ok: false, results: [{ slot: '-', ok: false, detail: 'no service' }] }; + } + + const jobs: { path: string; slot: 'cover' | 'slide'; index: number }[] = []; + if (opts.item.cover_path) { + jobs.push({ path: opts.item.cover_path, slot: 'cover', index: 0 }); + } + const slides = Array.isArray(opts.item.slide_paths) + ? (opts.item.slide_paths as string[]) + : []; + slides.forEach((p, i) => { + if (p) jobs.push({ path: p, slot: 'slide', index: i }); + }); + + for (const job of jobs) { + const dl = await svc.storage.from(ASSETS_BUCKET).download(job.path); + if (dl.error || !dl.data) { + results.push({ + slot: job.slot, + ok: false, + detail: dl.error?.message || 'download fail', + }); + continue; + } + const ab = await dl.data.arrayBuffer(); + const buf = Buffer.from(ab); + const ext = + job.path.split('.').pop()?.toLowerCase() || + (dl.data.type.includes('png') ? 'png' : 'jpg'); + const mime = dl.data.type || 'image/jpeg'; + const sync = await syncPipelineAssetToDrive({ + tenantId: opts.tenantId, + clientLabel: opts.clientLabel, + item: opts.item, + buffer: buf, + mimeType: mime, + ext, + slot: job.slot, + slideIndex: job.index, + }); + results.push({ + slot: `${job.slot}${job.slot === 'slide' ? job.index + 1 : ''}`, + ok: sync.ok, + detail: sync.ok + ? sync.drivePath || 'ok' + : sync.warning || 'fail', + }); + } + + return { ok: results.every((r) => r.ok), results }; +} diff --git a/src/lib/pipeline-publish-worker.ts b/src/lib/pipeline-publish-worker.ts new file mode 100644 index 0000000..af06415 --- /dev/null +++ b/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, + id: string, + body: Record +) { + 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, + 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(); + 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 }; +}