Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
64 changes: 48 additions & 16 deletions lib/ads/heroBackfill.ts
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,8 @@
// this is the free tier that runs when nothing else drew one.

import type { SupabaseClient } from "@supabase/supabase-js";
import { request as httpRequest } from "node:http";
import { request as httpsRequest } from "node:https";
import { buildPrompt, uploadHero } from "./heroImage";

type Env = Record<string, string | undefined>;
Expand Down Expand Up @@ -44,27 +46,57 @@ export class C0mputeImageError extends Error {
}
}

type Poster = (
url: string,
token: string,
body: string,
timeoutMs: number,
) => Promise<{ status: number; body: Buffer }>;

/**
* POST over node:http. Not fetch: Bun 1.4's fetch gave up on a c0mpute image
* after ~6 minutes even with `timeout: false` (seen live 2026-11-08), and
* Node's fetch stops waiting for headers at 300 s. This waits `timeoutMs`.
*/
export const postLong: Poster = (url, token, body, timeoutMs) =>
new Promise((resolve, reject) => {
const u = new URL(url);
const send = u.protocol === "https:" ? httpsRequest : httpRequest;
const req = send(
u,
{
method: "POST",
headers: {
authorization: `Bearer ${token}`,
"content-type": "application/json",
"content-length": Buffer.byteLength(body),
},
},
(res) => {
const chunks: Buffer[] = [];
res.on("data", (c: Buffer) => chunks.push(c));
res.on("end", () => resolve({ status: res.statusCode ?? 0, body: Buffer.concat(chunks) }));
res.on("error", reject);
},
);
req.setTimeout(timeoutMs, () => req.destroy(new Error(`c0mpute image: no answer in ${Math.round(timeoutMs / 1000)}s`)));
req.on("error", reject);
req.end(body);
});

/** 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,
post: Poster = postLong,
): Promise<Buffer> {
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),
// The worker runs on Bun, whose fetch has its own default timeout; ours
// above is the one that should apply (a CPU image can pass 5 minutes).
// Node's fetch ignores this key.
timeout: false,
} as RequestInit);
if (!res.ok) {
const detail = (await res.text().catch(() => "")).slice(0, 200);
const payload = JSON.stringify({ prompt: prompt.slice(0, 2000), width: 768, height: 512 });
const res = await post(target.url, target.token, payload, GENERATE_TIMEOUT_MS);
if (res.status !== 200) {
const detail = res.body.toString("utf8").slice(0, 200);
throw new C0mputeImageError(`c0mpute image ${res.status}: ${detail}`, res.status);
}
const bytes = Buffer.from(await res.arrayBuffer());
const bytes = res.body;
if (bytes.length < 1000 || !bytes.subarray(0, 8).equals(PNG_MAGIC)) {
throw new C0mputeImageError("c0mpute image: not a PNG", 502);
}
Expand Down Expand Up @@ -109,7 +141,7 @@ export type HeroBackfillResult = {
*/
export async function backfillAdHeroes(
supabase: SupabaseClient,
opts: { env?: Env; now?: number; fetchImpl?: typeof fetch } = {},
opts: { env?: Env; now?: number; post?: Poster } = {},
): Promise<HeroBackfillResult> {
const target = c0mputeImageTarget(opts.env);
if (!target) return { skipped: "not_configured" };
Expand Down Expand Up @@ -146,7 +178,7 @@ export async function backfillAdHeroes(

let bytes: Buffer;
try {
bytes = await generateC0mputeImage(prompt, target, opts.fetchImpl);
bytes = await generateC0mputeImage(prompt, target, opts.post);
} catch (err) {
const status = err instanceof C0mputeImageError ? err.status : 0;
// 503 = no worker free / none online: back off the whole sweep, not this campaign.
Expand Down
50 changes: 39 additions & 11 deletions tests/ads-hero-backfill.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -73,35 +73,38 @@ describe("ad hero backfill on c0mpute", () => {
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));
const post = vi.fn(async (url: string, token: string, raw: string, timeoutMs: number) => {
expect(url).toBe("http://gw:8787/v1/images");
expect(token).toBe(ENV.C0MPUTE_SPEECH_TOKEN);
expect(timeoutMs).toBe(14 * 60 * 1000);
const body = JSON.parse(raw);
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" } });
return { status: 200, body: PNG };
});
const r = await backfillAdHeroes(client, { env: ENV, fetchImpl: fetchImpl as never });
const r = await backfillAdHeroes(client, { env: ENV, post });
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 post = vi.fn(async () => ({ status: 503, body: Buffer.from("no worker") }));
const now = 1_000_000;
const first = await backfillAdHeroes(client, { env: ENV, fetchImpl: fetchImpl as never, now });
const first = await backfillAdHeroes(client, { env: ENV, post, now });
expect(first.error).toMatch(/503/);
const second = await backfillAdHeroes(client, { env: ENV, fetchImpl: fetchImpl as never, now: now + 60_000 });
const second = await backfillAdHeroes(client, { env: ENV, post, now: now + 60_000 });
expect(second).toEqual({ skipped: "network_backoff" });
expect(fetchImpl).toHaveBeenCalledTimes(1);
expect(post).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("<html>", { status: 200 }));
const post = vi.fn(async () => ({ status: 200, body: Buffer.from("<html>") }));
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({
expect((await backfillAdHeroes(client, { env: ENV, post, now })).error).toMatch(/not a PNG/);
expect(await backfillAdHeroes(client, { env: ENV, post, now: now + 1 })).toEqual({
skipped: "nothing_to_do",
});
});
Expand All @@ -111,3 +114,28 @@ describe("ad hero backfill on c0mpute", () => {
expect(await backfillAdHeroes(client, { env: {} })).toEqual({ skipped: "not_configured" });
});
});

import { createServer } from "node:http";
import { postLong } from "@/lib/ads/heroBackfill";

describe("postLong", () => {
it("POSTs to a real server and returns the body", async () => {
const server = createServer((req, res) => {
let b = "";
req.on("data", (c) => (b += c));
req.on("end", () => {
res.writeHead(200, { "x-auth": String(req.headers.authorization) });
res.end(`ok:${b}`);
});
});
await new Promise<void>((r) => server.listen(0, "127.0.0.1", () => r()));
const { port } = server.address() as { port: number };
try {
const r = await postLong(`http://127.0.0.1:${port}/v1/images`, "tok", '{"a":1}', 5000);
expect(r.status).toBe(200);
expect(r.body.toString()).toBe('ok:{"a":1}');
} finally {
server.close();
}
});
});
Loading