1// Listeners' content in the site's D1 database (the DB binding; the schema
2// is worker/migrations/0004_listener_content.sql): songs and their
3// revisions, covers' records (the images are in R2), comments, pitches, and
4// content_jobs, the work Jev still owes. content.ts validates everything
5// before it arrives here; content-jobs.ts runs the jobs.
6//
7// The jobs are made and removed by the schema's triggers as the items'
8// `screen` and `art` change, so nothing here writes a job except to claim
9// or reschedule it. A claim is one UPDATE … RETURNING: D1 runs writes one
10// statement at a time on its primary, so two workers never claim the same
11// job, and an expired lease is claimable again.
12import type { Subject, Verdict } from './screen';
13
14export type Kind = 'revision' | 'cover' | 'comment' | 'pitch';
15export type Task = 'screen' | 'score';
16export type Why = 'new' | 'budget' | 'budget-unavailable' | 'unreachable' | 'upstream' | 'unanswered';
17export type Screen = 'pending' | Verdict;
18export type Job = { kind: Kind; item: string; task: Task; userId: string; dueAt: number; attempts: number; why: Why };
19
20export type SongFields = { title: string; description: string; spec: string; code: string };
21export type Art = { art: number; artRuns: number[]; criteria: Record<string, number>; model: string; method: string };
22export type CoverRecord = { id: string; song: string; userId: string; contentType: string; bytes: number; alt: string };
23// A comment is on one of the site's songs or a listener's, by id.
24export type Target = { site: string } | { listener: string };
25
26export type PublicSong = {
27  id: string;
28  author: string;
29  // the author's account: its profile (data.ts's accounts/<id>)
30  authorId: string;
31  rev: number;
32  title: string;
33  description: string;
34  updatedAt: number;
35  art: Art | null;
36  cover: { id: string; alt: string } | null;
37  // the pitch it answers, its author's say (0005_pitch_votes.sql)
38  pitch: string | null;
39};
40export type PublicSongView = PublicSong & {
41  spec: string;
42  code: string;
43  revisions: { rev: number; at: number; art: number | null }[];
44};
45export type PublicText = { id: string; author: string; authorId: string; body: string; at: number };
46export type PitchSort = 'new' | 'votes';
47export type PublicPitch = PublicText & { votes: number; voted: boolean; answers: { id: string; title: string }[] };
48
49// What the author sees of one item: whether it is public, pending (and
50// why, and when Jev is asked again) or held (and Jev's verdict).
51export type Status =
52  | { status: 'public' }
53  | { status: 'pending'; why: Why; next: number; attempts: number }
54  | { status: 'held'; verdict: Exclude<Verdict, 'fine'>; confidence: number | null };
55
56type Row = Record<string, unknown>;
57const art = (r: Row): PublicSong['art'] =>
58  r.art === null || r.art === undefined
59    ? null
60    : {
61        art: r.art as number,
62        artRuns: JSON.parse(r.art_runs as string),
63        criteria: JSON.parse(r.criteria as string),
64        model: r.art_model as string,
65        method: r.art_method as string,
66      };
67
68// The newest fine revision of each song: the song as the public sees it.
69const PUBLIC_REVISION = `
70  SELECT r.* FROM song_revisions r
71  WHERE r.screen = 'fine' AND r.rev = (SELECT max(rev) FROM song_revisions WHERE song = r.song AND screen = 'fine')`;
72
73const publicSong = (r: Row): PublicSong => ({
74  id: r.song as string,
75  author: r.author as string,
76  authorId: r.author_id as string,
77  rev: r.rev as number,
78  title: r.title as string,
79  description: r.description as string,
80  updatedAt: r.created_at as number,
81  art: art(r),
82  cover: r.cover_id ? { id: r.cover_id as string, alt: r.cover_alt as string } : null,
83  pitch: (r.pitch as string | null | undefined) ?? null,
84});
85
86// Of an item's screen and its job (if any), what its author is told.
87export function statusOf(screen: Screen, confidence: number | null, job: Row | null): Status {
88  if (screen === 'fine') return { status: 'public' };
89  if (screen === 'pending') {
90    // the schema's triggers make a pending item's job with it, and mine()
91    // reads both in one transaction, so a pending item without one is a
92    // broken invariant, not a state to show
93    if (!job) throw new Error('a pending item has no job');
94    return { status: 'pending', why: job.why as Why, next: job.due_at as number, attempts: job.attempts as number };
95  }
96  return { status: 'held', verdict: screen, confidence };
97}
98
99export function d1Content(db: D1Database, now: () => number = Date.now) {
100  const TABLE: Record<Kind, string> = {
101    revision: 'song_revisions',
102    cover: 'covers',
103    comment: 'comments',
104    pitch: 'pitches',
105  };
106  const pitchIsPublic = async (pitchId: string): Promise<boolean> =>
107    (await db.prepare(`SELECT 1 AS yes FROM pitches WHERE id = ? AND screen = 'fine'`).bind(pitchId).first('yes')) === 1;
108
109  return {
110    // the store's clock, which the jobs also schedule by
111    now,
112
113    // ── writing content ─────────────────────────────
114    async createSong(userId: string, songId: string, revisionId: string, f: SongFields): Promise<void> {
115      const at = now();
116      await db.batch([
117        db.prepare('INSERT INTO listener_songs (id, user_id, created_at) VALUES (?, ?, ?)').bind(songId, userId, at),
118        db
119          .prepare(
120            `INSERT INTO song_revisions (id, song, rev, title, description, spec, code, created_at, screen)
121             VALUES (?, ?, 1, ?, ?, ?, ?, ?, 'pending')`,
122          )
123          .bind(revisionId, songId, f.title, f.description, f.spec, f.code, at),
124      ]);
125    },
126
127    // The next revision of a song, numbered after the last in the same statement.
128    async addRevision(songId: string, revisionId: string, f: SongFields): Promise<number> {
129      const row = await db
130        .prepare(
131          `INSERT INTO song_revisions (id, song, rev, title, description, spec, code, created_at, screen)
132           SELECT ?, ?, coalesce(max(rev), 0) + 1, ?, ?, ?, ?, ?, 'pending' FROM song_revisions WHERE song = ?
133           RETURNING rev`,
134        )
135        .bind(revisionId, songId, f.title, f.description, f.spec, f.code, now(), songId)
136        .first<{ rev: number }>();
137      return row!.rev;
138    },
139
140    async songOwner(songId: string): Promise<string | null> {
141      return db.prepare('SELECT user_id FROM listener_songs WHERE id = ?').bind(songId).first<string>('user_id');
142    },
143
144    // A new cover for a song, replacing its old one; returns the old one's
145    // id, whose R2 object the caller deletes.
146    async putCover(c: CoverRecord): Promise<string | null> {
147      const old = await db.prepare('SELECT id FROM covers WHERE song = ?').bind(c.song).first<string>('id');
148      await db.batch([
149        db.prepare('DELETE FROM covers WHERE song = ?').bind(c.song),
150        db
151          .prepare(
152            `INSERT INTO covers (id, song, user_id, content_type, bytes, alt, created_at, screen)
153             VALUES (?, ?, ?, ?, ?, ?, ?, 'pending')`,
154          )
155          .bind(c.id, c.song, c.userId, c.contentType, c.bytes, c.alt, now()),
156      ]);
157      return old;
158    },
159
160    async addComment(id: string, userId: string, target: Target, body: string): Promise<void> {
161      await db
162        .prepare(
163          `INSERT INTO comments (id, user_id, site_song, listener_song, body, created_at, screen)
164           VALUES (?, ?, ?, ?, ?, ?, 'pending')`,
165        )
166        .bind(id, userId, 'site' in target ? target.site : null, 'listener' in target ? target.listener : null, body, now())
167        .run();
168    },
169
170    async addPitch(id: string, userId: string, body: string): Promise<void> {
171      await db
172        .prepare(`INSERT INTO pitches (id, user_id, body, created_at, screen) VALUES (?, ?, ?, ?, 'pending')`)
173        .bind(id, userId, body, now())
174        .run();
175    },
176
177    // How many of this user's items wait on a screening.
178    async pendingScreens(userId: string): Promise<number> {
179      const n = await db
180        .prepare(`SELECT count(*) AS n FROM content_jobs WHERE user_id = ? AND task = 'screen'`)
181        .bind(userId)
182        .first<number>('n');
183      return n ?? 0;
184    },
185
186    // ── reading what is public ──────────────────────
187    // Every public song, the most art first (unscored last), as the site's own are listed.
188    async publicSongs(limit = 200): Promise<PublicSong[]> {
189      const { results } = await db
190        .prepare(
191          `SELECT p.*, u.display_name AS author, u.id AS author_id, c.id AS cover_id, c.alt AS cover_alt, s.pitch AS pitch
192           FROM (${PUBLIC_REVISION}) p
193           JOIN listener_songs s ON s.id = p.song
194           JOIN users u ON u.id = s.user_id
195           LEFT JOIN covers c ON c.song = p.song AND c.screen = 'fine'
196           ORDER BY p.art IS NULL, p.art DESC, p.created_at DESC
197           LIMIT ?`,
198        )
199        .bind(limit)
200        .all<Row>();
201      return results.map(publicSong);
202    },
203
204    // One account's public songs, newest first: its profile's.
205    async publicSongsBy(userId: string, limit = 100): Promise<PublicSong[]> {
206      const { results } = await db
207        .prepare(
208          `SELECT p.*, u.display_name AS author, u.id AS author_id, c.id AS cover_id, c.alt AS cover_alt
209           FROM (${PUBLIC_REVISION}) p
210           JOIN listener_songs s ON s.id = p.song
211           JOIN users u ON u.id = s.user_id
212           LEFT JOIN covers c ON c.song = p.song AND c.screen = 'fine'
213           WHERE s.user_id = ?
214           ORDER BY p.created_at DESC
215           LIMIT ?`,
216        )
217        .bind(userId, limit)
218        .all<Row>();
219      return results.map(publicSong);
220    },
221
222    async publicSong(songId: string): Promise<PublicSongView | null> {
223      const r = await db
224        .prepare(
225          `SELECT p.*, u.display_name AS author, u.id AS author_id, c.id AS cover_id, c.alt AS cover_alt, s.pitch AS pitch
226           FROM (${PUBLIC_REVISION}) p
227           JOIN listener_songs s ON s.id = p.song
228           JOIN users u ON u.id = s.user_id
229           LEFT JOIN covers c ON c.song = p.song AND c.screen = 'fine'
230           WHERE p.song = ?`,
231        )
232        .bind(songId)
233        .first<Row>();
234      if (!r) return null;
235      const { results } = await db
236        .prepare(`SELECT rev, created_at AS at, art FROM song_revisions WHERE song = ? AND screen = 'fine' ORDER BY rev`)
237        .bind(songId)
238        .all<{ rev: number; at: number; art: number | null }>();
239      return { ...publicSong(r), spec: r.spec as string, code: r.code as string, revisions: results };
240    },
241
242    // A cover as served: only a fine one whose song is public.
243    async publicCover(coverId: string): Promise<{ contentType: string; bytes: number } | null> {
244      return db
245        .prepare(
246          `SELECT c.content_type AS contentType, c.bytes FROM covers c
247           WHERE c.id = ? AND c.screen = 'fine'
248             AND EXISTS (SELECT 1 FROM song_revisions r WHERE r.song = c.song AND r.screen = 'fine')`,
249        )
250        .bind(coverId)
251        .first<{ contentType: string; bytes: number }>();
252    },
253
254    async publicComments(target: Target, limit = 200): Promise<PublicText[]> {
255      const [column, value] = 'site' in target ? ['site_song', target.site] : ['listener_song', target.listener];
256      const { results } = await db
257        .prepare(
258          `SELECT c.id, u.display_name AS author, u.id AS authorId, c.body, c.created_at AS at FROM comments c
259           JOIN users u ON u.id = c.user_id
260           WHERE c.${column} = ? AND c.screen = 'fine' ORDER BY c.created_at LIMIT ?`,
261        )
262        .bind(value, limit)
263        .all<PublicText>();
264      return results;
265    },
266
267    // Public pitches, newest or most voted first, each with its votes,
268    // whether `viewer` (an account id) voted for it, and the public
269    // listener songs that answer it (a site song says so in its SPEC.md,
270    // which the page reads).
271    async publicPitches(
272      limit = 100,
273      { sort = 'new', viewer = null }: { sort?: PitchSort; viewer?: string | null } = {},
274    ): Promise<PublicPitch[]> {
275      const order = sort === 'votes' ? 'votes DESC, p.created_at DESC' : 'p.created_at DESC';
276      const [pitches, answers] = (await db.batch([
277        db
278          .prepare(
279            `SELECT p.id, u.display_name AS author, u.id AS authorId, p.body, p.created_at AS at,
280               (SELECT count(*) FROM pitch_votes v WHERE v.pitch = p.id) AS votes,
281               EXISTS (SELECT 1 FROM pitch_votes v WHERE v.pitch = p.id AND v.user_id = ?) AS voted
282             FROM pitches p JOIN users u ON u.id = p.user_id
283             WHERE p.screen = 'fine' ORDER BY ${order} LIMIT ?`,
284          )
285          .bind(viewer ?? '', limit),
286        db.prepare(
287          `SELECT s.pitch, p.song AS id, p.title FROM (${PUBLIC_REVISION}) p
288           JOIN listener_songs s ON s.id = p.song WHERE s.pitch IS NOT NULL ORDER BY p.created_at`,
289        ),
290      ])) as D1Result<Row>[];
291      const answering = new Map<string, { id: string; title: string }[]>();
292      for (const a of answers.results) {
293        const list = answering.get(a.pitch as string) ?? [];
294        list.push({ id: a.id as string, title: a.title as string });
295        answering.set(a.pitch as string, list);
296      }
297      return pitches.results.map((p) => ({
298        id: p.id as string,
299        author: p.author as string,
300        authorId: p.authorId as string,
301        body: p.body as string,
302        at: p.at as number,
303        votes: p.votes as number,
304        voted: Boolean(p.voted),
305        answers: answering.get(p.id as string) ?? [],
306      }));
307    },
308
309    pitchIsPublic,
310
311    // An account's vote for a public pitch, cast (`on`) or taken back; the
312    // pitch's votes after, or null when it is not public.
313    async votePitch(pitchId: string, userId: string, on: boolean): Promise<{ votes: number; voted: boolean } | null> {
314      if (!(await pitchIsPublic(pitchId))) return null;
315      const [, count] = (await db.batch([
316        on
317          ? db.prepare('INSERT OR IGNORE INTO pitch_votes (pitch, user_id, created_at) VALUES (?, ?, ?)').bind(pitchId, userId, now())
318          : db.prepare('DELETE FROM pitch_votes WHERE pitch = ? AND user_id = ?').bind(pitchId, userId),
319        db.prepare('SELECT count(*) AS n FROM pitch_votes WHERE pitch = ?').bind(pitchId),
320      ])) as D1Result<Row>[];
321      return { votes: (count.results[0]?.n as number) ?? 0, voted: on };
322    },
323
324    // The pitch a listener song answers (null for none); its author's call.
325    async setSongPitch(songId: string, pitchId: string | null): Promise<void> {
326      await db.prepare('UPDATE listener_songs SET pitch = ? WHERE id = ?').bind(pitchId, songId).run();
327    },
328
329    // ── the author's own ────────────────────────────
330    // Everything this user wrote, newest first, each with its status.
331    async mine(userId: string) {
332      // one batch, one transaction: the items and their jobs as of one moment
333      const [jobRows, revisions, covers, comments, pitches] = (await db.batch([
334        db.prepare('SELECT kind, item, task, due_at, attempts, why FROM content_jobs WHERE user_id = ?').bind(userId),
335        db
336          .prepare(
337            `SELECT r.id, r.song, r.rev, r.title, r.created_at, r.screen, r.screen_confidence, r.art
338             FROM song_revisions r JOIN listener_songs s ON s.id = r.song
339             WHERE s.user_id = ? ORDER BY r.created_at DESC LIMIT 200`,
340          )
341          .bind(userId),
342        db
343          .prepare(
344            'SELECT id, song, alt, created_at, screen, screen_confidence FROM covers WHERE user_id = ? ORDER BY created_at DESC',
345          )
346          .bind(userId),
347        db
348          .prepare(
349            `SELECT id, site_song, listener_song, body, created_at, screen, screen_confidence FROM comments
350             WHERE user_id = ? ORDER BY created_at DESC LIMIT 200`,
351          )
352          .bind(userId),
353        db
354          .prepare(
355            'SELECT id, body, created_at, screen, screen_confidence FROM pitches WHERE user_id = ? ORDER BY created_at DESC LIMIT 200',
356          )
357          .bind(userId),
358      ])) as D1Result<Row>[];
359      const jobs = new Map<string, Row>();
360      for (const j of jobRows.results) jobs.set(`${j.kind}/${j.item}/${j.task}`, j);
361      const status = (kind: Kind, r: Row) =>
362        statusOf(r.screen as Screen, (r.screen_confidence as number | null) ?? null, jobs.get(`${kind}/${r.id}/screen`) ?? null);
363
364      return {
365        revisions: revisions.results.map((r) => {
366          const score = jobs.get(`revision/${r.id}/score`);
367          return {
368            song: r.song as string,
369            rev: r.rev as number,
370            title: r.title as string,
371            at: r.created_at as number,
372            ...status('revision', r),
373            art: (r.art as number | null) ?? null,
374            // a fine revision not scored yet says why, as a pending screening does
375            ...(score
376              ? { score: { why: score.why as Why, next: score.due_at as number, attempts: score.attempts as number } }
377              : {}),
378          };
379        }),
380        covers: covers.results.map((r) => ({
381          id: r.id as string,
382          song: r.song as string,
383          alt: r.alt as string,
384          at: r.created_at as number,
385          ...status('cover', r),
386        })),
387        comments: comments.results.map((r) => ({
388          id: r.id as string,
389          song: (r.site_song as string | null) ?? `listener:${r.listener_song}`,
390          body: r.body as string,
391          at: r.created_at as number,
392          ...status('comment', r),
393        })),
394        pitches: pitches.results.map((r) => ({
395          id: r.id as string,
396          body: r.body as string,
397          at: r.created_at as number,
398          ...status('pitch', r),
399        })),
400      };
401    },
402
403    // One item's status, for the answer to the write that made it.
404    async itemStatus(kind: Kind, item: string): Promise<Status | null> {
405      const [rows, jobRows] = (await db.batch([
406        db.prepare(`SELECT screen, screen_confidence FROM ${TABLE[kind]} WHERE id = ?`).bind(item),
407        db.prepare(`SELECT due_at, attempts, why FROM content_jobs WHERE kind = ? AND item = ? AND task = 'screen'`).bind(kind, item),
408      ])) as D1Result<Row>[];
409      const r = rows.results[0];
410      return r ? statusOf(r.screen as Screen, (r.screen_confidence as number | null) ?? null, jobRows.results[0] ?? null) : null;
411    },
412
413    // ── the jobs ────────────────────────────────────
414    // Claims up to `limit` due jobs (all users', or one user's), oldest due
415    // first, for `leaseMs`.
416    async claimDue(limit: number, leaseMs: number, userId: string | null = null): Promise<Job[]> {
417      const t = now();
418      const { results } = await db
419        .prepare(
420          `UPDATE content_jobs SET lease_until = ?1
421           WHERE rowid IN (
422             SELECT rowid FROM content_jobs
423             WHERE due_at <= ?2 AND (lease_until IS NULL OR lease_until <= ?2) AND (?3 IS NULL OR user_id = ?3)
424             ORDER BY due_at LIMIT ?4)
425           RETURNING kind, item, task, user_id, due_at, attempts, why`,
426        )
427        .bind(t + leaseMs, t, userId, limit)
428        .all<Row>();
429      return results.map((r) => ({
430        kind: r.kind as Kind,
431        item: r.item as string,
432        task: r.task as Task,
433        userId: r.user_id as string,
434        dueAt: r.due_at as number,
435        attempts: r.attempts as number,
436        why: r.why as Why,
437      }));
438    },
439
440    // Not done: asked again at `dueAt`, saying why.
441    async reschedule(job: Job, dueAt: number, why: Why): Promise<void> {
442      await db
443        .prepare(
444          `UPDATE content_jobs SET due_at = ?, why = ?, attempts = attempts + 1, lease_until = NULL
445           WHERE kind = ? AND item = ? AND task = ?`,
446        )
447        .bind(dueAt, why, job.kind, job.item, job.task)
448        .run();
449    },
450
451    // What the screening of an item is asked about, or null if it is gone.
452    async subject(kind: Kind, item: string): Promise<Subject | null> {
453      switch (kind) {
454        case 'revision': {
455          const r = await db
456            .prepare('SELECT title, description, spec, code, rev FROM song_revisions WHERE id = ?')
457            .bind(item)
458            .first<{ title: string; description: string; spec: string; code: string; rev: number }>();
459          return r && { kind, ...r };
460        }
461        case 'cover': {
462          const r = await db
463            .prepare(
464              `SELECT c.alt, (SELECT title FROM song_revisions WHERE song = c.song ORDER BY rev DESC LIMIT 1) AS songTitle
465               FROM covers c WHERE c.id = ?`,
466            )
467            .bind(item)
468            .first<{ alt: string; songTitle: string | null }>();
469          return r && { kind, alt: r.alt, songTitle: r.songTitle };
470        }
471        case 'comment': {
472          const r = await db
473            .prepare(
474              `SELECT c.body, coalesce(c.site_song,
475                 (SELECT title FROM song_revisions WHERE song = c.listener_song AND screen = 'fine' ORDER BY rev DESC LIMIT 1),
476                 'a listener''s song') AS song
477               FROM comments c WHERE c.id = ?`,
478            )
479            .bind(item)
480            .first<{ body: string; song: string }>();
481          return r && { kind, ...r };
482        }
483        case 'pitch': {
484          const r = await db.prepare('SELECT body FROM pitches WHERE id = ?').bind(item).first<{ body: string }>();
485          return r && { kind, body: r.body };
486        }
487      }
488    },
489
490    // A revision as the critic reads it, or null if it is gone.
491    async toScore(item: string): Promise<{ title: string; spec: string; code: string } | null> {
492      return db
493        .prepare(`SELECT title, spec, code FROM song_revisions WHERE id = ? AND screen = 'fine'`)
494        .bind(item)
495        .first<{ title: string; spec: string; code: string }>();
496    },
497
498    // The verdict, once: an item already screened is left as it is (a
499    // second worker whose lease had expired lost the race).
500    async recordScreen(kind: Kind, item: string, verdict: Verdict, confidence: number | null): Promise<boolean> {
501      const { meta } = await db
502        .prepare(
503          `UPDATE ${TABLE[kind]} SET screen = ?, screen_confidence = ?, screened_at = ? WHERE id = ? AND screen = 'pending'`,
504        )
505        .bind(verdict, confidence, now(), item)
506        .run();
507      return meta.changes > 0;
508    },
509
510    async recordScore(item: string, a: Art): Promise<boolean> {
511      const { meta } = await db
512        .prepare(
513          `UPDATE song_revisions SET art = ?, art_runs = ?, criteria = ?, art_model = ?, art_method = ?, scored_at = ?
514           WHERE id = ? AND screen = 'fine' AND art IS NULL`,
515        )
516        .bind(a.art, JSON.stringify(a.artRuns), JSON.stringify(a.criteria), a.model, a.method, now(), item)
517        .run();
518      return meta.changes > 0;
519    },
520
521    // ── what anyone may read of everything (data.ts) ─
522    // Every item, public or not: pending ones with why, held ones with
523    // Jev's verdict and confidence, each with its author's display name.
524    // As text only; data.ts never serves a held item's code or cover to be
525    // played or shown as an image.
526    async songsPage(limit: number, offset: number) {
527      const [total, page] = (await db.batch([
528        db.prepare('SELECT count(*) AS n FROM listener_songs'),
529        db
530          .prepare(
531            `SELECT s.id, u.id AS author_id, u.display_name AS author, s.created_at,
532               r.rev, r.title, r.screen, r.art,
533               (SELECT count(*) FROM song_revisions x WHERE x.song = s.id) AS revisions,
534               EXISTS (SELECT 1 FROM song_revisions x WHERE x.song = s.id AND x.screen = 'fine') AS public
535             FROM listener_songs s JOIN users u ON u.id = s.user_id
536             LEFT JOIN song_revisions r ON r.song = s.id AND r.rev = (SELECT max(rev) FROM song_revisions WHERE song = s.id)
537             ORDER BY s.created_at DESC, s.id LIMIT ? OFFSET ?`,
538          )
539          .bind(limit, offset),
540      ])) as D1Result<Row>[];
541      return {
542        total: (total.results[0]?.n as number) ?? 0,
543        songs: page.results.map((r) => ({
544          id: r.id as string,
545          author: { id: r.author_id as string, name: r.author as string },
546          created: r.created_at as number,
547          newest: { rev: r.rev as number, title: r.title as string, screen: r.screen as Screen, art: (r.art as number | null) ?? null },
548          revisions: r.revisions as number,
549          public: Boolean(r.public),
550        })),
551      };
552    },
553
554    // A song and every revision whole (spec and code as text), whatever
555    // Jev's verdict, with its cover's record.
556    async songRecord(songId: string) {
557      const [songs, revisions, covers] = (await db.batch([
558        db
559          .prepare(
560            `SELECT s.id, s.created_at, u.id AS author_id, u.display_name AS author FROM listener_songs s
561             JOIN users u ON u.id = s.user_id WHERE s.id = ?`,
562          )
563          .bind(songId),
564        db
565          .prepare(
566            `SELECT rev, title, description, spec, code, created_at, screen, screen_confidence, screened_at,
567               art, criteria, art_model, scored_at FROM song_revisions WHERE song = ? ORDER BY rev DESC`,
568          )
569          .bind(songId),
570        db
571          .prepare('SELECT id, content_type, bytes, alt, created_at, screen, screen_confidence FROM covers WHERE song = ?')
572          .bind(songId),
573      ])) as D1Result<Row>[];
574      const s = songs.results[0];
575      if (!s) return null;
576      const c = covers.results[0];
577      return {
578        id: s.id as string,
579        author: { id: s.author_id as string, name: s.author as string },
580        created: s.created_at as number,
581        revisions: revisions.results.map((r) => ({
582          rev: r.rev as number,
583          title: r.title as string,
584          description: r.description as string,
585          spec: r.spec as string,
586          code: r.code as string,
587          created: r.created_at as number,
588          screen: r.screen as Screen,
589          confidence: (r.screen_confidence as number | null) ?? null,
590          screened: (r.screened_at as number | null) ?? null,
591          art: (r.art as number | null) ?? null,
592          criteria: r.criteria ? (JSON.parse(r.criteria as string) as Record<string, number>) : null,
593          artModel: (r.art_model as string | null) ?? null,
594          scored: (r.scored_at as number | null) ?? null,
595        })),
596        cover: c
597          ? {
598              id: c.id as string,
599              contentType: c.content_type as string,
600              bytes: c.bytes as number,
601              alt: c.alt as string,
602              created: c.created_at as number,
603              screen: c.screen as Screen,
604              confidence: (c.screen_confidence as number | null) ?? null,
605            }
606          : null,
607      };
608    },
609
610    async coversPage(limit: number, offset: number) {
611      const [total, page] = (await db.batch([
612        db.prepare('SELECT count(*) AS n, coalesce(sum(bytes), 0) AS bytes FROM covers'),
613        db
614          .prepare(
615            `SELECT c.id, c.song, c.content_type, c.bytes, c.alt, c.created_at, c.screen, c.screen_confidence,
616               u.id AS author_id, u.display_name AS author,
617               EXISTS (SELECT 1 FROM song_revisions r WHERE r.song = c.song AND r.screen = 'fine') AS song_public
618             FROM covers c JOIN users u ON u.id = c.user_id
619             ORDER BY c.created_at DESC, c.id LIMIT ? OFFSET ?`,
620          )
621          .bind(limit, offset),
622      ])) as D1Result<Row>[];
623      return {
624        total: (total.results[0]?.n as number) ?? 0,
625        bytes: (total.results[0]?.bytes as number) ?? 0,
626        covers: page.results.map((c) => ({
627          id: c.id as string,
628          song: c.song as string,
629          author: { id: c.author_id as string, name: c.author as string },
630          contentType: c.content_type as string,
631          bytes: c.bytes as number,
632          alt: c.alt as string,
633          created: c.created_at as number,
634          screen: c.screen as Screen,
635          confidence: (c.screen_confidence as number | null) ?? null,
636          // what the site serves at /jev/listeners/covers/<id> (publicCover)
637          served: c.screen === 'fine' && Boolean(c.song_public),
638        })),
639      };
640    },
641
642    async commentsPage(limit: number, offset: number) {
643      const [total, page] = (await db.batch([
644        db.prepare('SELECT count(*) AS n FROM comments'),
645        db
646          .prepare(
647            `SELECT c.id, c.site_song, c.listener_song, c.body, c.created_at, c.screen, c.screen_confidence,
648               u.id AS author_id, u.display_name AS author
649             FROM comments c JOIN users u ON u.id = c.user_id
650             ORDER BY c.created_at DESC, c.id LIMIT ? OFFSET ?`,
651          )
652          .bind(limit, offset),
653      ])) as D1Result<Row>[];
654      return {
655        total: (total.results[0]?.n as number) ?? 0,
656        comments: page.results.map((c) => ({
657          id: c.id as string,
658          song: (c.site_song as string | null) ?? `listener:${c.listener_song}`,
659          author: { id: c.author_id as string, name: c.author as string },
660          body: c.body as string,
661          created: c.created_at as number,
662          screen: c.screen as Screen,
663          confidence: (c.screen_confidence as number | null) ?? null,
664        })),
665      };
666    },
667
668    // Every pitch, with who voted for it and when (0005_pitch_votes.sql:
669    // votes are public, voters included), and the listener songs whose
670    // authors say they answer it.
671    async pitchesPage(limit: number, offset: number) {
672      const onPage = `SELECT id FROM pitches ORDER BY created_at DESC, id LIMIT ? OFFSET ?`;
673      const [total, page, votes, answers] = (await db.batch([
674        db.prepare('SELECT count(*) AS n FROM pitches'),
675        db
676          .prepare(
677            `SELECT p.id, p.body, p.created_at, p.screen, p.screen_confidence, u.id AS author_id, u.display_name AS author
678             FROM pitches p JOIN users u ON u.id = p.user_id
679             ORDER BY p.created_at DESC, p.id LIMIT ? OFFSET ?`,
680          )
681          .bind(limit, offset),
682        db
683          .prepare(
684            `SELECT v.pitch, v.created_at, u.id AS voter_id, u.display_name AS voter
685             FROM pitch_votes v JOIN users u ON u.id = v.user_id
686             WHERE v.pitch IN (${onPage}) ORDER BY v.created_at`,
687          )
688          .bind(limit, offset),
689        db
690          .prepare(`SELECT s.id, s.pitch FROM listener_songs s WHERE s.pitch IN (${onPage}) ORDER BY s.created_at`)
691          .bind(limit, offset),
692      ])) as D1Result<Row>[];
693      const votersOf = new Map<string, { id: string; name: string; at: number }[]>();
694      for (const v of votes.results) {
695        const list = votersOf.get(v.pitch as string) ?? [];
696        list.push({ id: v.voter_id as string, name: v.voter as string, at: v.created_at as number });
697        votersOf.set(v.pitch as string, list);
698      }
699      const songsOf = new Map<string, string[]>();
700      for (const a of answers.results) songsOf.set(a.pitch as string, [...(songsOf.get(a.pitch as string) ?? []), a.id as string]);
701      return {
702        total: (total.results[0]?.n as number) ?? 0,
703        pitches: page.results.map((p) => ({
704          id: p.id as string,
705          author: { id: p.author_id as string, name: p.author as string },
706          body: p.body as string,
707          created: p.created_at as number,
708          screen: p.screen as Screen,
709          confidence: (p.screen_confidence as number | null) ?? null,
710          votes: votersOf.get(p.id as string) ?? [],
711          answeredBy: songsOf.get(p.id as string) ?? [],
712        })),
713      };
714    },
715
716    // The work Jev owes, oldest due first, with whose budget pays.
717    async jobsPage(limit: number, offset: number) {
718      const [total, page] = (await db.batch([
719        db.prepare('SELECT count(*) AS n FROM content_jobs'),
720        db
721          .prepare(
722            `SELECT j.kind, j.item, j.task, j.due_at, j.attempts, j.why, j.lease_until, u.id AS author_id, u.display_name AS author
723             FROM content_jobs j JOIN users u ON u.id = j.user_id
724             ORDER BY j.due_at, j.kind, j.item LIMIT ? OFFSET ?`,
725          )
726          .bind(limit, offset),
727      ])) as D1Result<Row>[];
728      const t = now();
729      return {
730        total: (total.results[0]?.n as number) ?? 0,
731        jobs: page.results.map((j) => ({
732          kind: j.kind as Kind,
733          item: j.item as string,
734          task: j.task as Task,
735          author: { id: j.author_id as string, name: j.author as string },
736          due: j.due_at as number,
737          attempts: j.attempts as number,
738          why: j.why as Why,
739          running: j.lease_until !== null && (j.lease_until as number) > t,
740        })),
741      };
742    },
743
744    // How many of each kind there are, by Jev's verdict.
745    async contentTotals() {
746      const byScreen = (table: string) => db.prepare(`SELECT screen, count(*) AS n FROM ${table} GROUP BY screen`);
747      const [songs, revisions, covers, comments, pitches, jobs, pitchVotes] = (await db.batch([
748        db.prepare('SELECT count(*) AS n FROM listener_songs'),
749        byScreen('song_revisions'),
750        byScreen('covers'),
751        byScreen('comments'),
752        byScreen('pitches'),
753        db.prepare('SELECT task, why, count(*) AS n FROM content_jobs GROUP BY task, why ORDER BY task, why'),
754        db.prepare('SELECT count(*) AS n FROM pitch_votes'),
755      ])) as D1Result<Row>[];
756      const screens = (r: D1Result<Row>) =>
757        Object.fromEntries(r.results.map((x) => [x.screen as string, x.n as number])) as Partial<Record<Screen, number>>;
758      return {
759        songs: (songs.results[0]?.n as number) ?? 0,
760        revisions: screens(revisions),
761        covers: screens(covers),
762        comments: screens(comments),
763        pitches: screens(pitches),
764        pitchVotes: (pitchVotes.results[0]?.n as number) ?? 0,
765        jobs: jobs.results.map((j) => ({ task: j.task as Task, why: j.why as Why, n: j.n as number })),
766      };
767    },
768  };
769}
770export type ContentStore = ReturnType<typeof d1Content>;