feat: E5 — Engagement ## Goal Social loop that keeps viewers coming ba

src/workers/channel-do.test.ts

@@ -0,0 +1,217 @@
+import { describe, expect, it } from 'vitest';
+import { ChannelSubscriberDO, triggerFanOut } from './channel-do';
+
+// ---------------------------------------------------------------------------
+// Fake DurableObjectState — only blockConcurrencyWhile is needed by fanOut.
+// ---------------------------------------------------------------------------
+function fakeState(): DurableObjectState {
+  return {
+    blockConcurrencyWhile<T>(cb: () => Promise<T>): Promise<T> {
+      return cb();
+    },
+  } as unknown as DurableObjectState;
+}
+
+// ---------------------------------------------------------------------------
+// Fake D1Database
+//
+// The fanOut loop:
+//   1. SELECT subscriber_user_id FROM subscriptions WHERE channel_user_id = ?
+//      AND subscriber_user_id > ?  ORDER BY subscriber_user_id ASC  LIMIT ?
+//   2. DB.batch([INSERT INTO subscription_inbox ... ON CONFLICT DO NOTHING])
+//
+// The fake pages subscribers by cursor and records every batch INSERT.
+// ---------------------------------------------------------------------------
+type InsertTuple = [subscriberId: string, videoId: string, channelUserId: string];
+
+interface FakeStmt {
+  bind: (...values: unknown[]) => FakeStmt;
+  first: <T>() => Promise<T | null>;
+  all: <T>() => Promise<{ results: T[] }>;
+  run: () => Promise<{ success: boolean }>;
+  _sql: string;
+  _bound: unknown[];
+}
+
+function fakeDB(opts: { subscribers?: string[] } = {}): {
+  db: D1Database;
+  inserts: InsertTuple[];
+} {
+  const subscribers = (opts.subscribers ?? []).slice().sort();
+  const inserts: InsertTuple[] = [];
+
+  // Each bind() call must return a NEW stmt with the new bound values so that
+  // `rows.map((r) => stmt.bind(r.id, videoId, channelId))` produces N distinct
+  // objects, not N references to the same mutated object.
+  const makeFakeStmt = (sql: string, bound: unknown[] = []): FakeStmt => {
+    const stmt: FakeStmt = {
+      _sql: sql,
+      _bound: bound,
+      bind: (...values: unknown[]) => makeFakeStmt(sql, values),
+      first: async <T>() => null as T | null,
+      all: async <T>() => {
+        // Subscription page query: (channelUserId, cursor, limit)
+        if (sql.includes('FROM subscriptions') && sql.includes('subscriber_user_id > ?')) {
+          const cursor = bound[1] as string;
+          const limit = bound[2] as number;
+          const page = subscribers.filter((id) => id > cursor).slice(0, limit);
+          return { results: page.map((id) => ({ subscriber_user_id: id })) as unknown as T[] };
+        }
+        return { results: [] as T[] };
+      },
+      run: async () => ({ success: true }),
+    };
+    return stmt;
+  };
+
+  const db: D1Database = {
+    prepare: (sql: string) => makeFakeStmt(sql) as unknown as D1PreparedStatement,
+    batch: async (stmts: D1PreparedStatement[]) => {
+      for (const s of stmts) {
+        const fake = s as unknown as FakeStmt;
+        if (fake._sql?.includes('INSERT INTO subscription_inbox')) {
+          inserts.push(fake._bound as InsertTuple);
+        }
+      }
+      return stmts.map(() => ({
+        results: [],
+        success: true,
+        meta: { duration: 0, last_row_id: 0, changes: 1, size_after: 0, rows_read: 0, rows_written: 0, changed_db: false },
+      }));
+    },
+    dump: async () => new ArrayBuffer(0),
+    exec: async () => ({ count: 0, duration: 0 }),
+  } as unknown as D1Database;
+
+  return { db, inserts };
+}
+
+function makeDO(subscribers: string[] = []): {
+  do: ChannelSubscriberDO;
+  inserts: InsertTuple[];
+} {
+  const { db, inserts } = fakeDB({ subscribers });
+  const doInstance = new ChannelSubscriberDO(fakeState(), { DB: db });
+  return { do: doInstance, inserts };
+}
+
+async function fanOutRequest(
+  doInstance: ChannelSubscriberDO,
+  videoId: string,
+  channelUserId: string,
+): Promise<{ inserted: number }> {
+  const res = await doInstance.fetch(
+    new Request('https://channel-do/fan-out', {
+      method: 'POST',
+      headers: { 'content-type': 'application/json' },
+      body: JSON.stringify({ videoId, channelUserId }),
+    }),
+  );
+  return res.json() as Promise<{ inserted: number }>;
+}
+
+// ---------------------------------------------------------------------------
+// Tests
+// ---------------------------------------------------------------------------
+
+describe('ChannelSubscriberDO.fetch', () => {
+  it('returns 400 for a malformed payload', async () => {
+    const { do: doInstance } = makeDO();
+    const res = await doInstance.fetch(
+      new Request('https://channel-do/fan-out', {
+        method: 'POST',
+        headers: { 'content-type': 'application/json' },
+        body: JSON.stringify({ videoId: 'v1' }), // missing channelUserId
+      }),
+    );
+    expect(res.status).toBe(400);
+  });
+
+  it('returns 404 for unknown paths', async () => {
+    const { do: doInstance } = makeDO();
+    const res = await doInstance.fetch(new Request('https://channel-do/unknown'));
+    expect(res.status).toBe(404);
+  });
+
+  it('inserts 0 rows for a channel with no subscribers', async () => {
+    const { do: doInstance, inserts } = makeDO([]);
+    const body = await fanOutRequest(doInstance, 'v1', 'ch1');
+    expect(body.inserted).toBe(0);
+    expect(inserts).toHaveLength(0);
+  });
+
+  it('inserts one inbox row per subscriber on a single page', async () => {
+    const subs = ['u1', 'u2', 'u3'];
+    const { do: doInstance, inserts } = makeDO(subs);
+    const body = await fanOutRequest(doInstance, 'v1', 'ch1');
+    expect(body.inserted).toBe(subs.length);
+    expect(inserts).toHaveLength(subs.length);
+    // Each insert carries (subscriberId, videoId, channelUserId)
+    const subscriberIds = inserts.map(([sub]) => sub).sort();
+    expect(subscriberIds).toEqual([...subs].sort());
+    for (const [, vid, ch] of inserts) {
+      expect(vid).toBe('v1');
+      expect(ch).toBe('ch1');
+    }
+  });
+
+  it('pages through subscribers when the count exceeds the batch size (200)', async () => {
+    // 201 subscribers forces a second page read.
+    const subs = Array.from({ length: 201 }, (_, i) =>
+      `user-${String(i).padStart(4, '0')}`,
+    );
+    const { do: doInstance, inserts } = makeDO(subs);
+    const body = await fanOutRequest(doInstance, 'v-big', 'ch-big');
+    expect(body.inserted).toBe(201);
+    expect(inserts).toHaveLength(201);
+    const ids = inserts.map(([sub]) => sub).sort();
+    expect(ids).toEqual([...subs].sort());
+  });
+
+  it('stops after an exactly full first page (200 rows)', async () => {
+    const subs = Array.from({ length: 200 }, (_, i) =>
+      `user-${String(i).padStart(4, '0')}`,
+    );
+    const { do: doInstance, inserts } = makeDO(subs);
+    const body = await fanOutRequest(doInstance, 'v-exact', 'ch-exact');
+    expect(body.inserted).toBe(200);
+    expect(inserts).toHaveLength(200);
+  });
+});
+
+describe('triggerFanOut', () => {
+  it('is a no-op when the binding is undefined', async () => {
+    await expect(
+      triggerFanOut(undefined, { videoId: 'v', channelUserId: 'c' }),
+    ).resolves.toBeUndefined();
+  });
+
+  it('calls the DO stub with the payload', async () => {
+    let called = false;
+    const fakeNs = {
+      idFromName: () => 'fake-id',
+      get: () => ({
+        fetch: async () => {
+          called = true;
+          return Response.json({ inserted: 1 });
+        },
+      }),
+    } as unknown as Parameters<typeof triggerFanOut>[0];
+    await triggerFanOut(fakeNs, { videoId: 'v1', channelUserId: 'c1' });
+    expect(called).toBe(true);
+  });
+
+  it('swallows errors so fan-out never blocks the caller', async () => {
+    const fakeNs = {
+      idFromName: () => 'fake-id',
+      get: () => ({
+        fetch: async () => {
+          throw new Error('network failure');
+        },
+      }),
+    } as unknown as Parameters<typeof triggerFanOut>[0];
+    await expect(
+      triggerFanOut(fakeNs, { videoId: 'v1', channelUserId: 'c1' }),
+    ).resolves.toBeUndefined();
+  });
+});

src/workers/og-meta.test.ts

@@ -1,5 +1,41 @@
 import { describe, expect, it } from 'vitest';
-import { buildOgMetaTags, clampForMeta, isPublicViewable } from './og-meta';
+import { buildOgMetaTags, clampForMeta, isPublicViewable, ogImageUrl } from './og-meta';
+
+describe('ogImageUrl', () => {
+  it('appends resize params to videodelivery.net thumbnails', () => {
+    const url = ogImageUrl(
+      'https://videodelivery.net/abc123/thumbnails/thumbnail.jpg',
+      'https://spooool.com',
+    );
+    expect(url).toBe(
+      'https://videodelivery.net/abc123/thumbnails/thumbnail.jpg?width=1200&height=630&fit=crop',
+    );
+  });
+
+  it('appends resize params to customer cloudflarestream.com subdomains', () => {
+    const url = ogImageUrl(
+      'https://customer-abc.cloudflarestream.com/abc123/thumbnails/thumbnail.jpg',
+      'https://spooool.com',
+    );
+    expect(url).toBe(
+      'https://customer-abc.cloudflarestream.com/abc123/thumbnails/thumbnail.jpg?width=1200&height=630&fit=crop',
+    );
+  });
+
+  it('passes non-Stream URLs through unchanged', () => {
+    const raw = 'https://r2.example.com/thumbs/vid.jpg';
+    expect(ogImageUrl(raw, 'https://spooool.com')).toBe(raw);
+  });
+
+  it('falls back to /icon.png when thumbnail is null', () => {
+    expect(ogImageUrl(null, 'https://x.test')).toBe('https://x.test/icon.png');
+  });
+
+  it('returns the raw value when the URL is not parseable', () => {
+    const bad = 'not a url';
+    expect(ogImageUrl(bad, 'https://x.test')).toBe(bad);
+  });
+});
 
 describe('clampForMeta', () => {
   it('returns the empty string for null/undefined/empty input', () => {

src/workers/og-meta.ts

@@ -32,6 +32,26 @@ interface VideoMetaRow {
 const TITLE_MAX = 70;
 const DESCRIPTION_MAX = 200;
 
+// Cloudflare Stream CDN hostnames that accept thumbnail resize parameters.
+// Requesting 1200×630 ensures social crawlers get a properly sized OG card
+// instead of whatever resolution the Stream encoder chose for the player.
+const CF_STREAM_HOSTS = ['videodelivery.net', 'cloudflarestream.com'];
+
+export function ogImageUrl(thumbnailUrl: string | null, origin: string): string {
+  if (!thumbnailUrl) return `${origin}/icon.png`;
+  try {
+    const u = new URL(thumbnailUrl);
+    const isStream = CF_STREAM_HOSTS.some((h) => u.hostname === h || u.hostname.endsWith(`.${h}`));
+    if (!isStream) return thumbnailUrl;
+    u.searchParams.set('width', '1200');
+    u.searchParams.set('height', '630');
+    u.searchParams.set('fit', 'crop');
+    return u.toString();
+  } catch {
+    return thumbnailUrl;
+  }
+}
+
 export function clampForMeta(value: string | null | undefined, max: number): string {
   if (!value) return '';
   // Iterate code points so emoji at the boundary aren't split into lone
@@ -55,7 +75,7 @@ export function buildOgMetaTags(args: {
     video.description ?? `Watch on Spooool${video.channel_name ? ` — ${video.channel_name}` : ''}`,
     DESCRIPTION_MAX,
   );
-  const image = video.thumbnail_url ?? `${origin}/icon.png`;
+  const image = ogImageUrl(video.thumbnail_url, origin);
 
   const escape = (v: string): string =>
     v