app/lib/social/queue.server.ts

Functions

Imports

It imports app/db/schema.ts, app/lib/people.server.ts, app/lib/social/lint.ts, app/lib/social/platforms.server.ts, app/lib/social/routine.server.ts, app/lib/timed-fetch.server.ts. Imported by app/lib/ai.server.ts, app/lib/content.server.ts, app/lib/health.server.ts, app/routes/social.tsx, workers/app.ts.

Source

1// The social queue (design decision 5). A piece going live is the only thing a post can announce.
2// For each account that announces that project, the post comes from, in order:
3//
4//   1. a draft stored ahead by the session that finished the piece (draftSocialPost);
5//   2. else a Claude Code routine Carrel fires, batching every waiting item into one run (the
6//      routine stores its drafts the same way), with a nightly sweep;
7//   3. else, when the routine fails or its daily cap is reached, or it drafted nothing in time, a
8//      template post, which the health check reports to Dustin.
9//
10// Every post is linted; a finding holds it for Dustin. Then each account's switch decides: approval
11// (Dustin approves each post) or auto (it goes when the caps allow). Nothing here calls an AI. The
12// loop over accounts, the daily cap, the gap between posts and the record of every post are carried
13// over from legacy Recova's social-poster cron.
14
15import { and, asc, desc, eq, gte, inArray, isNotNull, isNull, lte } from "drizzle-orm";
16import { drizzle } from "drizzle-orm/d1";
17
18import { socialAccounts, socialEvents, socialPosts, socialRoutineRuns } from "~/db/schema";
19import type { Viewer } from "~/lib/people.server";
20
21import { lintPost } from "./lint";
22import { credentialsFor, postOriginal, xCostMills } from "./platforms.server";
23import { fireRoutine, routineConfig, routinePrompt } from "./routine.server";
24import { timedFetch } from "~/lib/timed-fetch.server";
25
26type Fetch = typeof fetch;
27
28/** How long published items gather before one routine run takes them all. */
29export const BATCH_DELAY_MS = 30 * 60_000;
30/** How long the routine has to store its drafts before the template is used. */
31export const ROUTINE_WAIT_MS = 3 * 60 * 60_000;
32/** Routine runs a day: one to three, under the subscription's routine cap (design decision 5). */
33const ROUTINE_DAILY_CAP = 3;
34/** The nightly sweep's hour, UTC (3 a.m. Central in daylight time). */
35const NIGHTLY_HOUR_UTC = 8;
36
37const DEFAULT_TEMPLATE = "New: {title}. {summary} {link}";
38
39function requireOwner(viewer: Viewer) {
40  if (!viewer.isOwner) throw new Response("Not found", { status: 404 });
41}
42
43function dayStart(now: Date): string {
44  return `${now.toISOString().slice(0, 10)}T00:00:00.000Z`;
45}
46
47function monthStart(now: Date): string {
48  return `${now.toISOString().slice(0, 7)}-01T00:00:00.000Z`;
49}
50
51// ---------- accounts and switches
52
53export type AccountInput = { key: string; name: string; platform: "bluesky" | "x"; kind: "personal" | "brand"; handle: string; projectId: number | null };
54
55export async function createAccount(db: D1Database, viewer: Viewer, input: AccountInput) {
56  requireOwner(viewer);
57  if (!/^[a-z0-9]+(?:-[a-z0-9]+)*$/.test(input.key)) return { ok: false as const, error: "The key is lower-case letters, digits and hyphens, such as germomics-bluesky." };
58  if (!input.name.trim() || !input.handle.trim()) return { ok: false as const, error: "Give the account a name and its handle." };
59  try {
60    // Every account starts off, in approval mode.
61    await drizzle(db).insert(socialAccounts).values({ ...input, name: input.name.trim(), handle: input.handle.trim().replace(/^@/, ""), enabled: false, mode: "approval" });
62  } catch {
63    return { ok: false as const, error: `An account with the key ${input.key} already exists.` };
64  }
65  return { ok: true as const };
66}
67
68export async function setSwitches(db: D1Database, viewer: Viewer, id: number, input: { enabled?: boolean; mode?: "approval" | "auto"; dailyCap?: number; monthlyBudgetMills?: number; template?: string; voiceGuide?: string }) {
69  requireOwner(viewer);
70  const d = drizzle(db);
71  const account = await d.select().from(socialAccounts).where(eq(socialAccounts.id, id)).get();
72  if (!account) throw new Response("Not found", { status: 404 });
73  // The personal account is approved post by post, always.
74  if (input.mode === "auto" && account.kind === "personal") return { ok: false as const, error: "The personal account is always approval: every post is Dustin's call." };
75  const set: Partial<typeof socialAccounts.$inferInsert> = {};
76  if (input.enabled !== undefined) set.enabled = input.enabled;
77  if (input.mode) set.mode = input.mode;
78  if (input.dailyCap !== undefined) set.dailyCap = Math.max(0, Math.min(20, Math.floor(input.dailyCap)));
79  if (input.monthlyBudgetMills !== undefined) set.monthlyBudgetMills = Math.max(0, Math.floor(input.monthlyBudgetMills));
80  if (input.template !== undefined) set.template = input.template.slice(0, 500);
81  if (input.voiceGuide !== undefined) set.voiceGuide = input.voiceGuide.slice(0, 4000);
82  if (Object.keys(set).length) await d.update(socialAccounts).set(set).where(eq(socialAccounts.id, id));
83  return { ok: true as const };
84}
85
86// ---------- events and drafts
87
88/** The description line in a post's frontmatter: Dustin's one-line summary, for templates and the routine. */
89export function summaryFrom(source: string): string {
90  const fm = /^---\n([\s\S]*?)\n---/.exec(source.replace(/\r\n/g, "\n"))?.[1] ?? "";
91  return /^description:\s*"?(.*?)"?\s*$/m.exec(fm)?.[1]?.trim() ?? "";
92}
93
94/**
95 * A piece went live: every account announcing its project gets an event, which becomes sendable now.
96 * An account that is off when the piece goes live closes its event, so turning it on later does
97 * not post a backlog.
98 */
99export async function recordPublication(db: D1Database, projectId: number, item: { id: string; title: string; url: string; summary: string }, now = new Date()) {
100  const d = drizzle(db);
101  const accounts = await d.select().from(socialAccounts).where(eq(socialAccounts.projectId, projectId)).all();
102  for (const account of accounts) {
103    const values = { title: item.title, url: item.url, summary: item.summary, publishedAt: now.toISOString() };
104    const existing = await d
105      .select()
106      .from(socialEvents)
107      .where(and(eq(socialEvents.accountId, account.id), eq(socialEvents.projectId, projectId), eq(socialEvents.itemId, item.id)))
108      .get();
109    if (existing?.publishedAt) continue; // Announced once, not on every re-publish.
110    const state = account.enabled ? (existing?.state ?? "waiting") : "closed";
111    if (existing) await d.update(socialEvents).set({ ...values, state }).where(eq(socialEvents.id, existing.id));
112    else await d.insert(socialEvents).values({ accountId: account.id, projectId, itemId: item.id, ...values, state });
113  }
114}
115
116/**
117 * Stores a drafted post for an account and a piece, ahead of publication or after the routine was
118 * asked for it. The newest draft replaces any earlier one that has not gone out. This is what the
119 * session that finishes a piece, and the routine, call through MCP.
120 */
121export async function draftSocialPost(db: D1Database, input: { accountKey: string; projectId: number; itemId: string; text: string; createdBy: string }) {
122  const d = drizzle(db);
123  const account = await d.select().from(socialAccounts).where(eq(socialAccounts.key, input.accountKey)).get();
124  if (!account || account.projectId !== input.projectId) return { ok: false as const, error: `No account ${input.accountKey} announces this project.` };
125  let event = await d
126    .select()
127    .from(socialEvents)
128    .where(and(eq(socialEvents.accountId, account.id), eq(socialEvents.projectId, input.projectId), eq(socialEvents.itemId, input.itemId)))
129    .get();
130  if (!event) {
131    [event] = await d.insert(socialEvents).values({ accountId: account.id, projectId: input.projectId, itemId: input.itemId }).returning();
132  }
133  if (event!.state === "templated" || event!.state === "closed") return { ok: false as const, error: "This item's post was already settled (template used, or the account was off)." };
134  await d
135    .update(socialPosts)
136    .set({ status: "rejected", error: "Replaced by a newer draft." })
137    .where(and(eq(socialPosts.eventId, event!.id), inArray(socialPosts.status, ["drafted", "held", "awaiting", "queued"])));
138  const [post] = await d
139    .insert(socialPosts)
140    .values({ accountId: account.id, eventId: event!.id, text: input.text.trim(), source: event!.state === "routine" ? "routine" : "predrafted", status: "drafted", createdBy: input.createdBy })
141    .returning({ id: socialPosts.id });
142  return { ok: true as const, postId: post!.id };
143}
144
145/** The routine is told each item by its event id; this stores its draft for that event. */
146export async function draftForEvent(db: D1Database, input: { eventId: number; text: string; createdBy: string }) {
147  const row = await drizzle(db)
148    .select({ event: socialEvents, key: socialAccounts.key })
149    .from(socialEvents)
150    .innerJoin(socialAccounts, eq(socialAccounts.id, socialEvents.accountId))
151    .where(eq(socialEvents.id, input.eventId))
152    .get();
153  if (!row) return { ok: false as const, error: `No social event ${input.eventId}.` };
154  return draftSocialPost(db, { accountKey: row.key, projectId: row.event.projectId, itemId: row.event.itemId, text: input.text, createdBy: input.createdBy });
155}
156
157function fillTemplate(template: string, event: { title: string; summary: string; url: string }): string {
158  return (template || DEFAULT_TEMPLATE)
159    .replaceAll("{title}", event.title)
160    .replaceAll("{summary}", event.summary)
161    .replaceAll("{link}", event.url)
162    .replace(/\s+/g, " ")
163    .replace(/\s+([.,])/g, "$1")
164    .replace(/\.\./g, ".")
165    .trim();
166}
167
168// ---------- the tick
169
170export type TickResult = { linted: number; routineFired: number; templated: number; sent: number; waitingOnCaps: number; errors: string[] };
171
172type Account = typeof socialAccounts.$inferSelect;
173
174/** Lints a drafted post and routes it by the account's switch. */
175async function route(db: D1Database, post: typeof socialPosts.$inferSelect, account: Account) {
176  const findings = lintPost(post.text, account.platform);
177  const status = findings.length ? "held" : account.mode === "auto" ? "queued" : "awaiting";
178  await drizzle(db).update(socialPosts).set({ status, lint: JSON.stringify(findings) }).where(eq(socialPosts.id, post.id));
179  await drizzle(db).update(socialEvents).set({ state: "covered" }).where(and(eq(socialEvents.id, post.eventId), inArray(socialEvents.state, ["waiting", "routine"])));
180}
181
182async function useTemplate(db: D1Database, event: typeof socialEvents.$inferSelect, account: Account) {
183  const d = drizzle(db);
184  const [post] = await d
185    .insert(socialPosts)
186    .values({ accountId: account.id, eventId: event.id, text: fillTemplate(account.template, event), source: "template", status: "drafted", createdBy: "template" })
187    .returning();
188  await d.update(socialEvents).set({ state: "templated" }).where(eq(socialEvents.id, event.id));
189  await route(db, post!, account);
190}
191
192async function sendQueued(env: Env, account: Account, now: Date, fetcher: Fetch, result: TickResult) {
193  const d = drizzle(env.DB);
194  const queued = await d.select().from(socialPosts).where(and(eq(socialPosts.accountId, account.id), eq(socialPosts.status, "queued"))).orderBy(asc(socialPosts.id)).all();
195  if (queued.length === 0) return;
196  const sent = await d
197    .select({ sentAt: socialPosts.sentAt, cost: socialPosts.costMills })
198    .from(socialPosts)
199    .where(and(eq(socialPosts.accountId, account.id), eq(socialPosts.status, "sent"), gte(socialPosts.sentAt, monthStart(now))))
200    .orderBy(desc(socialPosts.sentAt))
201    .all();
202  let today = sent.filter((s) => s.sentAt! >= dayStart(now)).length;
203  let spent = sent.reduce((n, s) => n + s.cost, 0);
204  let last = sent[0]?.sentAt ? Date.parse(sent[0].sentAt) : 0;
205  const creds = credentialsFor(env, account);
206
207  for (const post of queued) {
208    // Recova's caps, per account: the daily cap and the gap between posts; X's monthly budget too.
209    if (today >= account.dailyCap || now.getTime() - last < account.minGapMinutes * 60_000) {
210      result.waitingOnCaps++;
211      return;
212    }
213    const cost = account.platform === "x" ? xCostMills(post.text) : 0;
214    if (account.platform === "x" && spent + cost > account.monthlyBudgetMills) {
215      await d.update(socialPosts).set({ error: `Held: it would pass this month's X budget ($${(account.monthlyBudgetMills / 1000).toFixed(2)}).` }).where(eq(socialPosts.id, post.id));
216      result.waitingOnCaps++;
217      return;
218    }
219    if ("missing" in creds) {
220      await d.update(socialPosts).set({ error: `Waiting for credentials: ${creds.missing.join(", ")}.` }).where(eq(socialPosts.id, post.id));
221      result.errors.push(`${account.key}: credentials missing`);
222      return;
223    }
224    try {
225      const id = await postOriginal(creds, post.text, fetcher, now.getTime());
226      await d.update(socialPosts).set({ status: "sent", platformPostId: id, costMills: cost, sentAt: now.toISOString(), error: null }).where(eq(socialPosts.id, post.id));
227      today++;
228      spent += cost;
229      last = now.getTime();
230      result.sent++;
231    } catch (error) {
232      const message = error instanceof Error ? error.message : String(error);
233      await d.update(socialPosts).set({ status: "failed", error: message }).where(eq(socialPosts.id, post.id));
234      result.errors.push(`${account.key}: ${message}`);
235    }
236  }
237}
238
239/** One pass of the queue, run by the cron. Every step is idempotent, so a missed run loses nothing. */
240export async function processSocial(env: Env, deps: { fetcher?: Fetch; now?: Date } = {}): Promise<TickResult> {
241  const fetcher = deps.fetcher ?? timedFetch;
242  const now = deps.now ?? new Date();
243  const d = drizzle(env.DB);
244  const result: TickResult = { linted: 0, routineFired: 0, templated: 0, sent: 0, waitingOnCaps: 0, errors: [] };
245  const accounts = new Map((await d.select().from(socialAccounts).where(eq(socialAccounts.enabled, true)).all()).map((a) => [a.id, a]));
246  if (accounts.size === 0) return result;
247  const ids = [...accounts.keys()];
248
249  // 1. Drafts for published pieces, stored ahead or by the routine: lint, then the switch.
250  const drafted = await d
251    .select({ post: socialPosts })
252    .from(socialPosts)
253    .innerJoin(socialEvents, eq(socialEvents.id, socialPosts.eventId))
254    .where(and(eq(socialPosts.status, "drafted"), isNotNull(socialEvents.publishedAt), inArray(socialPosts.accountId, ids)))
255    .all();
256  for (const { post } of drafted) {
257    await route(env.DB, post, accounts.get(post.accountId)!);
258    result.linted++;
259  }
260
261  // 3a. The routine was asked and drafted nothing in time: the template.
262  const stale = await d
263    .select({ event: socialEvents, requestedAt: socialRoutineRuns.requestedAt })
264    .from(socialEvents)
265    .innerJoin(socialRoutineRuns, eq(socialRoutineRuns.id, socialEvents.routineRunId))
266    .where(and(eq(socialEvents.state, "routine"), inArray(socialEvents.accountId, ids), lte(socialRoutineRuns.requestedAt, new Date(now.getTime() - ROUTINE_WAIT_MS).toISOString())))
267    .all();
268  for (const { event } of stale) {
269    await useTemplate(env.DB, event, accounts.get(event.accountId)!);
270    result.templated++;
271  }
272
273  // 2. Published pieces with no draft: one routine run for all of them, after the batch delay, or
274  // at the nightly sweep.
275  const nightly = now.getUTCHours() === NIGHTLY_HOUR_UTC;
276  const waiting = (
277    await d
278      .select()
279      .from(socialEvents)
280      .where(and(eq(socialEvents.state, "waiting"), isNotNull(socialEvents.publishedAt), inArray(socialEvents.accountId, ids)))
281      .all()
282  ).filter((e) => nightly || Date.parse(e.publishedAt!) <= now.getTime() - BATCH_DELAY_MS);
283  if (waiting.length > 0) {
284    const config = routineConfig(env);
285    const runsToday = await d.select({ id: socialRoutineRuns.id }).from(socialRoutineRuns).where(and(eq(socialRoutineRuns.status, "fired"), gte(socialRoutineRuns.requestedAt, dayStart(now)))).all();
286    let templateAll: string | null = null;
287    if ("missing" in config) templateAll = config.missing;
288    else if (runsToday.length >= ROUTINE_DAILY_CAP) templateAll = `the routine's daily cap of ${ROUTINE_DAILY_CAP} runs is reached`;
289    else {
290      const prompt = routinePrompt(
291        waiting.map((e) => {
292          const a = accounts.get(e.accountId)!;
293          return { eventId: e.id, account: `${a.name} (${a.key})`, platform: a.platform, voiceGuide: a.voiceGuide, title: e.title, url: e.url, summary: e.summary };
294        }),
295      );
296      const fired = await fireRoutine(config, prompt, fetcher);
297      const [run] = await d
298        .insert(socialRoutineRuns)
299        .values({
300          kind: nightly ? "nightly" : "batch",
301          requestedAt: now.toISOString(),
302          events: JSON.stringify(waiting.map((e) => e.id)),
303          status: fired.ok ? "fired" : "failed",
304          sessionUrl: fired.ok ? fired.sessionUrl : null,
305          error: fired.ok ? null : fired.error,
306        })
307        .returning({ id: socialRoutineRuns.id });
308      if (fired.ok) {
309        await d.update(socialEvents).set({ state: "routine", routineRunId: run!.id }).where(inArray(socialEvents.id, waiting.map((e) => e.id)));
310        result.routineFired++;
311      } else if (!fired.retry) {
312        templateAll = `the routine refused the run (${fired.error})`;
313      } else {
314        // The hourly limit or a brief outage: the next tick tries again; the items keep waiting.
315        result.errors.push(`routine: ${fired.error}`);
316      }
317    }
318    if (templateAll) {
319      result.errors.push(`template used: ${templateAll}`);
320      for (const event of waiting) {
321        await useTemplate(env.DB, event, accounts.get(event.accountId)!);
322        result.templated++;
323      }
324    }
325  }
326
327  // Posts cleared to go: within each account's caps.
328  for (const account of accounts.values()) await sendQueued(env, account, now, fetcher, result);
329  return result;
330}
331
332// ---------- Dustin's decisions
333
334export async function decidePost(
335  db: D1Database,
336  viewer: Viewer,
337  id: number,
338  decision: { action: "approve" | "reject" | "by-hand" | "acknowledge" } | { action: "edit"; text: string },
339  now = new Date(),
340) {
341  requireOwner(viewer);
342  const d = drizzle(db);
343  const row = await d.select({ post: socialPosts, account: socialAccounts }).from(socialPosts).innerJoin(socialAccounts, eq(socialAccounts.id, socialPosts.accountId)).where(eq(socialPosts.id, id)).get();
344  if (!row) throw new Response("Not found", { status: 404 });
345  const { post, account } = row;
346  const open = ["held", "awaiting"].includes(post.status);
347  switch (decision.action) {
348    case "approve":
349      // A held post is approved only once its text passes the lint (edit it first).
350      if (post.status !== "awaiting") return { ok: false as const, error: post.status === "held" ? "Edit the post until the lint passes, then approve it." : "Only a post awaiting approval can be approved." };
351      await d.update(socialPosts).set({ status: "queued", approvedBy: viewer.id }).where(eq(socialPosts.id, id));
352      return { ok: true as const };
353    case "reject":
354      if (!open && post.status !== "queued") return { ok: false as const, error: "Only a post that has not gone out can be rejected." };
355      await d.update(socialPosts).set({ status: "rejected", approvedBy: viewer.id }).where(eq(socialPosts.id, id));
356      return { ok: true as const };
357    case "by-hand":
358      // The personal account's free path: Dustin copies the text and posts it himself.
359      if (!open) return { ok: false as const, error: "Only a post that has not gone out can be posted by hand." };
360      await d.update(socialPosts).set({ status: "by-hand", approvedBy: viewer.id, sentAt: now.toISOString() }).where(eq(socialPosts.id, id));
361      return { ok: true as const };
362    case "edit": {
363      if (!open) return { ok: false as const, error: "Only a held or awaiting post can be edited." };
364      const findings = lintPost(decision.text, account.platform);
365      await d.update(socialPosts).set({ text: decision.text.trim(), lint: JSON.stringify(findings), status: findings.length ? "held" : "awaiting" }).where(eq(socialPosts.id, id));
366      return { ok: true as const, findings };
367    }
368    case "acknowledge":
369      if (post.source !== "template") return { ok: false as const, error: "Only a template post has an alert to acknowledge." };
370      await d.update(socialPosts).set({ acknowledgedAt: now.toISOString() }).where(eq(socialPosts.id, id));
371      return { ok: true as const };
372  }
373}
374
375// ---------- reading
376
377export async function socialOverview(db: D1Database, viewer: Viewer) {
378  requireOwner(viewer);
379  const d = drizzle(db);
380  const [accounts, posts, runs] = await Promise.all([
381    d.select().from(socialAccounts).orderBy(asc(socialAccounts.key)).all(),
382    d
383      .select({ post: socialPosts, title: socialEvents.title, url: socialEvents.url, accountKey: socialAccounts.key, platform: socialAccounts.platform, kind: socialAccounts.kind })
384      .from(socialPosts)
385      .innerJoin(socialEvents, eq(socialEvents.id, socialPosts.eventId))
386      .innerJoin(socialAccounts, eq(socialAccounts.id, socialPosts.accountId))
387      .orderBy(desc(socialPosts.id))
388      .limit(100)
389      .all(),
390    d.select().from(socialRoutineRuns).orderBy(desc(socialRoutineRuns.id)).limit(10).all(),
391  ]);
392  return { accounts, posts, runs };
393}
394
395/** Template posts that went out and Dustin has not yet marked seen: the health check's alert. */
396export async function unseenTemplates(db: D1Database) {
397  return drizzle(db)
398    .select({ id: socialPosts.id, accountKey: socialAccounts.key, text: socialPosts.text })
399    .from(socialPosts)
400    .innerJoin(socialAccounts, eq(socialAccounts.id, socialPosts.accountId))
401    .where(and(eq(socialPosts.source, "template"), eq(socialPosts.status, "sent"), isNull(socialPosts.acknowledgedAt)))
402    .all();
403}