feat(api): fill network slots from a pool of distinct carbon ads

gitbook-docs/architecture/ads.md

@@ -68,6 +68,9 @@ link and a CPM bid (1–100). It is stored in `ad_campaigns` and served as `sour
   production) or the author is an admin (`ADMIN_EMAILS`). **Production runs with it off**: ads
   start in `review`, and an advertiser cannot approve their own ad. Admins review through
   `GET /admin/ads?status=review` and `PATCH /admin/ads/:id { status }`; there is no review page yet.
+- **Carbon pool:** Carbon returns one ad per call, so the API keeps a pool of up to 8 distinct
+  ads (3 calls at most once a minute, 30-minute ttl) and fills three of every four network slots
+  from it, rotating; the fourth is a house ad. With an empty pool, house ads fill everything.
 - **Fill in production:** house ads are Distro TV's own promos only (the recognisable dev-tool
   creatives are local-only), and Carbon serves only with a real `CARBON_ZONE_KEY` or
   `CARBON_ENABLED=1` — its sandbox zone must be a deliberate choice for a public audience.

packages/api/src/__tests__/ad-supply.test.ts

@@ -70,3 +70,54 @@ describe("displayLabel", () => {
     expect(displayLabel("not a url", "Acme")).toBe("Acme")
   })
 })
+
+describe("fillSlots", () => {
+  const c = (id: string) => ({ adId: `carbon:${id}` })
+  const h = (i: number) => ({ adId: `house:${i}` })
+  it("fills mostly from the carbon pool, rotating through it, with a house ad every fourth slot", async () => {
+    const { fillSlots } = await import("../services/ad-supply.service.js")
+    const out = fillSlots(8, [c("a"), c("b"), c("c")], h, 0).picked.map((x) => x.adId)
+    expect(out).toEqual([
+      "carbon:a",
+      "carbon:b",
+      "carbon:c",
+      "house:0",
+      "carbon:a",
+      "carbon:b",
+      "carbon:c",
+      "house:1",
+    ])
+  })
+  it("continues the rotation across batches so consecutive batches don't restart on the same ad", async () => {
+    const { fillSlots } = await import("../services/ad-supply.service.js")
+    const first = fillSlots(2, [c("a"), c("b"), c("c")], h, 0)
+    const second = fillSlots(2, [c("a"), c("b"), c("c")], h, first.cursor)
+    expect(first.picked.map((x) => x.adId)).toEqual(["carbon:a", "carbon:b"])
+    expect(second.picked.map((x) => x.adId)).toEqual(["carbon:c", "house:0"])
+  })
+  it("falls back to house ads alone when the pool is empty", async () => {
+    const { fillSlots } = await import("../services/ad-supply.service.js")
+    expect(fillSlots(3, [], h, 0).picked.map((x) => x.adId)).toEqual([
+      "house:0",
+      "house:1",
+      "house:2",
+    ])
+  })
+})
+
+describe("carbon pool", () => {
+  it("keeps distinct ads, newest last, drops stale ones and caps its size", async () => {
+    const { mergeCarbonPool } = await import("../services/carbon-ad.provider.js")
+    const ad = (id: string) => ({ adId: id }) as never
+    let pool = mergeCarbonPool([], [ad("a"), ad("b"), ad("a")], 1_000)
+    expect(pool.map((p) => p.ad.adId)).toEqual(["a", "b"])
+    pool = mergeCarbonPool(pool, [ad("c")], 2_000)
+    expect(pool.map((p) => p.ad.adId)).toEqual(["a", "b", "c"])
+    // 31 minutes later everything older than the 30-minute ttl has aged out
+    pool = mergeCarbonPool(pool, [ad("d")], 1_000 + 31 * 60_000)
+    expect(pool.map((p) => p.ad.adId)).toEqual(["d"])
+    // and it never grows past its cap
+    const many = Array.from({ length: 20 }, (_, i) => ad(`x${i}`))
+    expect(mergeCarbonPool([], many, 5_000).length).toBeLessThanOrEqual(8)
+  })
+})

packages/api/src/services/ad-supply.service.ts

@@ -4,13 +4,13 @@ import { env } from "../config/env.js"
 import { getDb } from "../db/index.js"
 import { adImpressions } from "../db/schema/ad_impressions.js"
 import { logger } from "../lib/logger.js"
-import { displayLabel, fetchCarbonAd } from "./carbon-ad.provider.js"
+import { displayLabel, getCarbonAds } from "./carbon-ad.provider.js"
 import { clickCode } from "./ad-impression.service.js"
 import { listActiveDirectAds, pickDirect } from "./advertiser-ads.service.js"
 import type { NormalizedAd } from "./carbon-ad.provider.js"
 import { houseAd } from "./house-ads.js"
 
-let houseCursor = 0
+let fillCursor = 0
 
 // how much of a batch direct (advertiser-written) ads may take. the rest stays
 // network + house fill so one advertiser can't own every slot.
@@ -32,8 +32,8 @@ export async function nextAds(args: {
 }): Promise<SponsoredPayload[]> {
   if (args.n <= 0) return []
   try {
-    const [carbon, direct] = await Promise.all([
-      fetchCarbonAd(),
+    const [carbonAds, direct] = await Promise.all([
+      getCarbonAds(),
       listActiveDirectAds().catch(() => []),
     ])
     const fillCpm = env.adCpmRate
@@ -57,13 +57,10 @@ export async function nextAds(args: {
       cpmRate: d.bidCpm,
     }))
 
-    // network fill: carbon takes the first free slot, house ads the rest
-    let carbonLeft: NormalizedAd | null = carbon
-    while (picked.length < args.n) {
-      const ad: NormalizedAd = carbonLeft ?? houseAd(houseCursor++)
-      carbonLeft = null
-      picked.push({ ad, deliveryId: randomUUID(), cpmRate: fillCpm })
-    }
+    // network fill: mostly the carbon pool, a house ad every fourth slot
+    const fill = fillSlots(args.n - picked.length, carbonAds, (i) => houseAd(i), fillCursor)
+    fillCursor = fill.cursor
+    for (const ad of fill.picked) picked.push({ ad, deliveryId: randomUUID(), cpmRate: fillCpm })
 
     // interleave so a batch doesn't open with a run of the same kind
     const ordered = interleaveBySource(picked)
@@ -118,3 +115,29 @@ function interleaveBySource(picked: Picked[]): Picked[] {
   }
   return out
 }
+
+const HOUSE_EVERY = 4
+
+// fills n slots from the carbon pool, rotating through it, with a house ad in every
+// fourth position; house ads alone when the pool is empty. `cursor` carries the
+// rotation across batches so consecutive batches don't restart on the same ad.
+export function fillSlots<T>(
+  n: number,
+  carbonPool: T[],
+  house: (index: number) => T,
+  cursor: number
+): { picked: T[]; cursor: number } {
+  const picked: T[] = []
+  let c = cursor
+  for (let i = 0; i < n; i++, c++) {
+    const pos = c % HOUSE_EVERY
+    const round = Math.floor(c / HOUSE_EVERY)
+    if (carbonPool.length === 0) picked.push(house(c))
+    else if (pos === HOUSE_EVERY - 1) picked.push(house(round))
+    else {
+      const ad = carbonPool[(round * (HOUSE_EVERY - 1) + pos) % carbonPool.length]
+      if (ad !== undefined) picked.push(ad)
+    }
+  }
+  return { picked, cursor: c }
+}

packages/api/src/services/carbon-ad.provider.ts

@@ -76,27 +76,60 @@ export function mapCarbonAd(ad: CarbonAdLike): NormalizedAd | null {
   }
 }
 
-let cached: { at: number; ad: NormalizedAd | null } | null = null
+export interface PooledAd {
+  ad: NormalizedAd
+  at: number
+}
 
-// never throws. 60s in-process cache so /me/content/next doesn't hit carbon per request.
-export async function fetchCarbonAd(): Promise<NormalizedAd | null> {
-  if (!env.carbonEnabled) return null
-  if (cached && Date.now() - cached.at < CACHE_TTL_MS) return cached.ad
-  try {
-    const opts: { placement: string; serve?: string } = { placement: env.carbonPlacement }
-    if (env.carbonZoneKey) opts.serve = env.carbonZoneKey
-    const raw = await Promise.race([
-      fetchAd(opts),
-      new Promise<null>((_, reject) =>
-        setTimeout(() => reject(new Error("carbon_fetch_timeout")), FETCH_TIMEOUT_MS)
-      ),
-    ])
-    const ad = raw ? mapCarbonAd(raw as CarbonAdLike) : null
-    cached = { at: Date.now(), ad }
-    return ad
-  } catch (err) {
-    logger.warn({ err }, "carbon ad fetch failed")
-    cached = { at: Date.now(), ad: null }
-    return null
+const POOL_TTL_MS = 30 * 60_000
+const POOL_MAX = 8
+const FETCHES_PER_REFRESH = 3
+
+// distinct ads seen recently, oldest first. carbon returns one ad per call and rotates
+// its inventory, so a few calls a minute build a small, varied pool.
+export function mergeCarbonPool(pool: PooledAd[], fresh: NormalizedAd[], now: number): PooledAd[] {
+  const byId = new Map<string, PooledAd>()
+  for (const p of pool) if (now - p.at < POOL_TTL_MS) byId.set(p.ad.adId, p)
+  for (const ad of fresh) {
+    // a re-seen ad keeps its place but has its clock reset
+    const existing = byId.get(ad.adId)
+    if (existing) existing.at = now
+    else byId.set(ad.adId, { ad, at: now })
+  }
+  return [...byId.values()].slice(-POOL_MAX)
+}
+
+let pool: PooledAd[] = []
+let lastRefreshAt = 0
+
+async function fetchOne(): Promise<NormalizedAd | null> {
+  const opts: { placement: string; serve?: string } = { placement: env.carbonPlacement }
+  if (env.carbonZoneKey) opts.serve = env.carbonZoneKey
+  const raw = await Promise.race([
+    fetchAd(opts),
+    new Promise<null>((_, reject) =>
+      setTimeout(() => reject(new Error("carbon_fetch_timeout")), FETCH_TIMEOUT_MS)
+    ),
+  ])
+  return raw ? mapCarbonAd(raw as CarbonAdLike) : null
+}
+
+// never throws. refreshes at most once a minute, so /me/content/next doesn't hit carbon
+// per request; between refreshes it serves what is already in the pool.
+export async function getCarbonAds(): Promise<NormalizedAd[]> {
+  if (!env.carbonEnabled) return []
+  const now = Date.now()
+  if (now - lastRefreshAt >= CACHE_TTL_MS) {
+    lastRefreshAt = now
+    const results = await Promise.allSettled(
+      Array.from({ length: FETCHES_PER_REFRESH }, () => fetchOne())
+    )
+    const fresh: NormalizedAd[] = []
+    for (const r of results) {
+      if (r.status === "fulfilled" && r.value) fresh.push(r.value)
+      else if (r.status === "rejected") logger.warn({ err: r.reason }, "carbon ad fetch failed")
+    }
+    pool = mergeCarbonPool(pool, fresh, now)
   }
+  return pool.map((p) => p.ad)
 }