From 9a1e566fe48360ff5c280a34c4db6a5150c730f6 Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Sun, 8 Nov 2026 14:29:42 +0000 Subject: [PATCH 1/2] feat(ads): free c0mpute-drawn heroes for campaigns that shipped without one The worker sweep (every 5 min, one campaign per run, never overlapping) asks the c0mpute gateway's POST /v1/images (Z-Image-Turbo on CPU workers) for a 768x512 hero built from the house art-direction prompt, uploads it with the existing uploadHero, and sets image_url on that campaign's image-less creatives. The request path is unchanged (copy already takes ~2 min there). 503 backs the sweep off 15 min; a per-campaign failure backs it off 6 h. Off switch: AD_HERO_C0MPUTE=0. Co-Authored-By: Claude Opus 5.5 --- lib/ads/heroBackfill.ts | 175 ++++++++++++++++++++++++++++++++ lib/ads/heroImage.ts | 2 +- tests/ads-hero-backfill.test.ts | 113 +++++++++++++++++++++ worker/index.ts | 17 ++++ 4 files changed, 306 insertions(+), 1 deletion(-) create mode 100644 lib/ads/heroBackfill.ts create mode 100644 tests/ads-hero-backfill.test.ts diff --git a/lib/ads/heroBackfill.ts b/lib/ads/heroBackfill.ts new file mode 100644 index 00000000..f5293aeb --- /dev/null +++ b/lib/ads/heroBackfill.ts @@ -0,0 +1,175 @@ +// Free ad heroes on c0mpute: CPU image generation (Z-Image-Turbo on a +// c0mpute worker, `diffusion.txt2img`) for campaigns that shipped with no hero. +// +// The request path never waits for this. Ad copy alone takes ~2 minutes on +// c0mpute, and an image is ~2 more, so /dashboard/ads/new keeps shipping the +// ad on its accent tint (or the advertiser's og:image) and this worker sweep +// fills the art in afterwards: one campaign per run, every creative of that +// campaign still missing an image gets the same hero, exactly as the request +// path would have set it. +// +// Paid gpt-image heroes stay where they were (AD_GENERATOR_PAID_IMAGES=1); +// this is the free tier that runs when nothing else drew one. + +import type { SupabaseClient } from "@supabase/supabase-js"; +import { buildPrompt, uploadHero } from "./heroImage"; + +type Env = Record; + +// Read through a variable so nothing here can be inlined at build time. +function runtimeEnv(): Env { + const e: Env = process.env; + return e; +} + +export type C0mputeImageTarget = { url: string; token: string }; + +/** The c0mpute gateway's image route, or null when none is configured. */ +export function c0mputeImageTarget(env: Env = runtimeEnv()): C0mputeImageTarget | null { + if (/^(0|off|false|no)$/i.test((env.AD_HERO_C0MPUTE ?? "").trim())) return null; + const base = (env.C0MPUTE_SPEECH_URL ?? "").trim().replace(/\/+$/, ""); + const token = (env.C0MPUTE_SPEECH_TOKEN ?? "").trim(); + if (!base || !token) return null; + return { url: `${base}/v1/images`, token }; +} + +const PNG_MAGIC = Buffer.from([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a]); +const GENERATE_TIMEOUT_MS = 6 * 60 * 1000; + +export class C0mputeImageError extends Error { + constructor(message: string, readonly status: number) { + super(message); + } +} + +/** One 768x512 (3:2, the hero's shape) image from a c0mpute CPU worker. */ +export async function generateC0mputeImage( + prompt: string, + target: C0mputeImageTarget, + fetchImpl: typeof fetch = fetch, +): Promise { + const res = await fetchImpl(target.url, { + method: "POST", + headers: { authorization: `Bearer ${target.token}`, "content-type": "application/json" }, + body: JSON.stringify({ prompt: prompt.slice(0, 2000), width: 768, height: 512 }), + signal: AbortSignal.timeout(GENERATE_TIMEOUT_MS), + }); + if (!res.ok) { + const detail = (await res.text().catch(() => "")).slice(0, 200); + throw new C0mputeImageError(`c0mpute image ${res.status}: ${detail}`, res.status); + } + const bytes = Buffer.from(await res.arrayBuffer()); + if (bytes.length < 1000 || !bytes.subarray(0, 8).equals(PNG_MAGIC)) { + throw new C0mputeImageError("c0mpute image: not a PNG", 502); + } + return bytes; +} + +export function domainOf(url: string | null | undefined): string { + try { + return new URL(String(url)).hostname.replace(/^www\./, ""); + } catch { + return ""; + } +} + +type Candidate = { + campaign_id: string; + headline: string | null; + body: string | null; + bg_color: string | null; + accent_color: string | null; + ad_campaigns: { destination_url: string | null; status: string | null } | null; +}; + +// Per-process memory of what failed, so a bad campaign or an empty network is +// not retried every tick. Lost on restart, which is fine: it only rate-limits. +const campaignRetryAt = new Map(); +let networkRetryAt = 0; +const CAMPAIGN_BACKOFF_MS = 6 * 60 * 60 * 1000; +const NETWORK_BACKOFF_MS = 15 * 60 * 1000; +const LOOKBACK_DAYS = 30; + +export type HeroBackfillResult = { + skipped?: "not_configured" | "network_backoff" | "nothing_to_do"; + campaignId?: string; + heroUrl?: string; + creativesUpdated?: number; + error?: string; +}; + +/** + * Give one campaign that has no hero a c0mpute-drawn one. Never throws. + */ +export async function backfillAdHeroes( + supabase: SupabaseClient, + opts: { env?: Env; now?: number; fetchImpl?: typeof fetch } = {}, +): Promise { + const target = c0mputeImageTarget(opts.env); + if (!target) return { skipped: "not_configured" }; + const now = opts.now ?? Date.now(); + if (now < networkRetryAt) return { skipped: "network_backoff" }; + + const since = new Date(now - LOOKBACK_DAYS * 24 * 60 * 60 * 1000).toISOString(); + const { data, error } = await supabase + .from("ad_creatives") + .select("campaign_id, headline, body, bg_color, accent_color, ad_campaigns!inner(destination_url, status)") + .is("image_url", null) + .eq("status", "ready") + .in("ad_campaigns.status", ["active", "pending_review", "draft"]) + .gte("created_at", since) + .order("created_at", { ascending: false }) + .limit(50); + if (error) return { error: `select: ${error.message}` }; + + const pick = ((data ?? []) as unknown as Candidate[]).find( + (row) => (campaignRetryAt.get(row.campaign_id) ?? 0) <= now, + ); + if (!pick) return { skipped: "nothing_to_do" }; + + const domain = domainOf(pick.ad_campaigns?.destination_url); + const prompt = buildPrompt( + { title: "", description: "", domain } as Parameters[0], + { + headline: pick.headline ?? "", + body: pick.body ?? "", + bgColor: pick.bg_color ?? "#0b0d10", + accentColor: pick.accent_color ?? "#6ee7b7", + }, + ); + + let bytes: Buffer; + try { + bytes = await generateC0mputeImage(prompt, target, opts.fetchImpl); + } catch (err) { + const status = err instanceof C0mputeImageError ? err.status : 0; + // 503 = no worker free / none online: back off the whole sweep, not this campaign. + if (status === 503 || status === 0) networkRetryAt = now + NETWORK_BACKOFF_MS; + else campaignRetryAt.set(pick.campaign_id, now + CAMPAIGN_BACKOFF_MS); + return { campaignId: pick.campaign_id, error: err instanceof Error ? err.message : String(err) }; + } + + const heroUrl = await uploadHero(bytes); + if (!heroUrl) { + campaignRetryAt.set(pick.campaign_id, now + CAMPAIGN_BACKOFF_MS); + return { campaignId: pick.campaign_id, error: "upload failed" }; + } + + const { data: updated, error: updateError } = await supabase + .from("ad_creatives") + .update({ image_url: heroUrl }) + .eq("campaign_id", pick.campaign_id) + .is("image_url", null) + .select("id"); + if (updateError) { + campaignRetryAt.set(pick.campaign_id, now + CAMPAIGN_BACKOFF_MS); + return { campaignId: pick.campaign_id, heroUrl, error: `update: ${updateError.message}` }; + } + return { campaignId: pick.campaign_id, heroUrl, creativesUpdated: updated?.length ?? 0 }; +} + +/** Test hook: forget backoff state. */ +export function resetHeroBackfillState() { + campaignRetryAt.clear(); + networkRetryAt = 0; +} diff --git a/lib/ads/heroImage.ts b/lib/ads/heroImage.ts index acf380d9..5f7a5c11 100644 --- a/lib/ads/heroImage.ts +++ b/lib/ads/heroImage.ts @@ -158,7 +158,7 @@ export function looksLikeIcon(url: string): boolean { // prefix, so a user-scoped client can't write to `heroes/` (it fails with "new // row violates row-level security policy"). Upload with the service-role client, // which bypasses RLS. -async function uploadHero(bytes: Buffer): Promise { +export async function uploadHero(bytes: Buffer): Promise { const svc = serviceClient(); const id = crypto.randomUUID(); const originalPath = `heroes/${id}.png`; diff --git a/tests/ads-hero-backfill.test.ts b/tests/ads-hero-backfill.test.ts new file mode 100644 index 00000000..d272d712 --- /dev/null +++ b/tests/ads-hero-backfill.test.ts @@ -0,0 +1,113 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +vi.mock("@/lib/onion", () => ({ smartFetch: vi.fn() })); +vi.mock("@/lib/supabase/service", () => ({ serviceClient: vi.fn() })); +const uploadHero = vi.fn(); +vi.mock("@/lib/ads/heroImage", async (orig) => ({ + ...(await orig()), + uploadHero: (...a: unknown[]) => uploadHero(...a), +})); + +import { + backfillAdHeroes, + c0mputeImageTarget, + domainOf, + resetHeroBackfillState, +} from "@/lib/ads/heroBackfill"; + +const ENV = { C0MPUTE_SPEECH_URL: "http://gw:8787/", C0MPUTE_SPEECH_TOKEN: "t".repeat(32) }; +const PNG = Buffer.concat([Buffer.from([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a]), Buffer.alloc(2000)]); + +function fakeSupabase(rows: unknown[]) { + const updates: Array<{ values: unknown; campaign: unknown }> = []; + const select = { + is: () => select, + eq: () => select, + in: () => select, + gte: () => select, + order: () => select, + limit: async () => ({ data: rows, error: null }), + }; + const client = { + from: () => ({ + select: () => select, + update: (values: unknown) => { + const u = { values, campaign: null as unknown }; + const chain = { + eq: (_c: string, v: unknown) => ((u.campaign = v), chain), + is: () => chain, + select: async () => { + updates.push(u); + return { data: [{ id: "a" }, { id: "b" }], error: null }; + }, + }; + return chain; + }, + }), + }; + return { client: client as never, updates }; +} + +const row = { + campaign_id: "c1", + headline: "Ship faster", + body: "CI in seconds", + bg_color: "#101010", + accent_color: "#ff6600", + ad_campaigns: { destination_url: "https://www.example.dev/x", status: "active" }, +}; + +describe("ad hero backfill on c0mpute", () => { + beforeEach(() => { + resetHeroBackfillState(); + uploadHero.mockReset(); + }); + + it("targets the gateway's /v1/images and can be switched off", () => { + expect(c0mputeImageTarget(ENV)).toEqual({ url: "http://gw:8787/v1/images", token: ENV.C0MPUTE_SPEECH_TOKEN }); + expect(c0mputeImageTarget({})).toBeNull(); + expect(c0mputeImageTarget({ ...ENV, AD_HERO_C0MPUTE: "0" })).toBeNull(); + expect(domainOf("https://www.example.dev/x")).toBe("example.dev"); + }); + + it("draws one hero and writes it to the campaign's image-less creatives", async () => { + const { client, updates } = fakeSupabase([row]); + uploadHero.mockResolvedValue("https://cdn/heroes/1.webp"); + const fetchImpl = vi.fn(async (_url: string, init: RequestInit) => { + const body = JSON.parse(String(init.body)); + expect(body).toMatchObject({ width: 768, height: 512 }); + expect(body.prompt).toContain("Ship faster"); + expect(body.prompt).toContain("example.dev"); + return new Response(PNG, { status: 200, headers: { "content-type": "image/png" } }); + }); + const r = await backfillAdHeroes(client, { env: ENV, fetchImpl: fetchImpl as never }); + expect(r).toEqual({ campaignId: "c1", heroUrl: "https://cdn/heroes/1.webp", creativesUpdated: 2 }); + expect(updates).toEqual([{ values: { image_url: "https://cdn/heroes/1.webp" }, campaign: "c1" }]); + }); + + it("backs the whole sweep off when no worker is free (503)", async () => { + const { client } = fakeSupabase([row]); + const fetchImpl = vi.fn(async () => new Response("no worker", { status: 503 })); + const now = 1_000_000; + const first = await backfillAdHeroes(client, { env: ENV, fetchImpl: fetchImpl as never, now }); + expect(first.error).toMatch(/503/); + const second = await backfillAdHeroes(client, { env: ENV, fetchImpl: fetchImpl as never, now: now + 60_000 }); + expect(second).toEqual({ skipped: "network_backoff" }); + expect(fetchImpl).toHaveBeenCalledTimes(1); + }); + + it("skips a campaign that failed on its own, and rejects non-PNG bytes", async () => { + const { client } = fakeSupabase([row]); + const fetchImpl = vi.fn(async () => new Response("", { status: 200 })); + const now = 5_000_000; + expect((await backfillAdHeroes(client, { env: ENV, fetchImpl: fetchImpl as never, now })).error).toMatch(/not a PNG/); + expect(await backfillAdHeroes(client, { env: ENV, fetchImpl: fetchImpl as never, now: now + 1 })).toEqual({ + skipped: "nothing_to_do", + }); + }); + + it("does nothing without a gateway", async () => { + const { client } = fakeSupabase([row]); + expect(await backfillAdHeroes(client, { env: {} })).toEqual({ skipped: "not_configured" }); + }); +}); diff --git a/worker/index.ts b/worker/index.ts index 229dbc1d..a009a051 100644 --- a/worker/index.ts +++ b/worker/index.ts @@ -36,6 +36,7 @@ import { deliverArticle } from "../lib/lx/webhookDeliver"; import { getHub } from "../lib/webhooks/hub"; import { repairStuckLxJobs } from "../lib/lx/repair"; import { repairMissingArticleImages } from "../lib/lx/repairImages"; +import { backfillAdHeroes } from "../lib/ads/heroBackfill"; import { processDueSocialFeeds } from "../lib/sp/feedAutopost"; import { processBrowserPost } from "../lib/sp/browserPost"; import { getOrMintInstallationToken } from "../lib/github/installations"; @@ -1432,6 +1433,22 @@ console.log("[worker] webhooks drainer started"); setInterval(() => sweep().catch(() => {}), 60_000); setInterval(() => browserPostSweep().catch((e) => console.error("[worker] browser post sweep", e)), 60_000); setInterval(() => lxSweep().catch((e) => console.error("[worker] lx sweep", e)), 60_000); +// Free ad heroes from c0mpute CPU workers for campaigns that shipped without +// one (lib/ads/heroBackfill). One image takes ~2 minutes, so a run never +// overlaps the next. +let heroBackfillRunning = false; +setInterval(() => { + if (heroBackfillRunning) return; + heroBackfillRunning = true; + backfillAdHeroes(supabase) + .then((r) => { + if (r.heroUrl || r.error) console.log("[worker] ad hero backfill", r); + }) + .catch((e) => console.error("[worker] ad hero backfill crashed", e)) + .finally(() => { + heroBackfillRunning = false; + }); +}, 5 * 60_000); setInterval( () => socialFeedSweep().catch((e) => console.error("[worker] social feed sweep", e)), 60_000, From 4c7266a65b4bc82fc8d250f6726c4ecce1713079 Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Sun, 8 Nov 2026 14:35:44 +0000 Subject: [PATCH 2/2] fix(ads): give a slow CPU worker 14 min per hero Co-Authored-By: Claude Opus 5.5 --- lib/ads/heroBackfill.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/lib/ads/heroBackfill.ts b/lib/ads/heroBackfill.ts index f5293aeb..401d269c 100644 --- a/lib/ads/heroBackfill.ts +++ b/lib/ads/heroBackfill.ts @@ -34,7 +34,9 @@ export function c0mputeImageTarget(env: Env = runtimeEnv()): C0mputeImageTarget } const PNG_MAGIC = Buffer.from([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a]); -const GENERATE_TIMEOUT_MS = 6 * 60 * 1000; +// The gateway runs the auction and retries other workers itself; a 6-core +// worker can need ~5 minutes per image. +const GENERATE_TIMEOUT_MS = 14 * 60 * 1000; export class C0mputeImageError extends Error { constructor(message: string, readonly status: number) {