diff --git a/packages/db/src/migrations/0125_x_mention_poller.sql b/packages/db/src/migrations/0125_x_mention_poller.sql new file mode 100644 index 0000000000..426d648f8b --- /dev/null +++ b/packages/db/src/migrations/0125_x_mention_poller.sql @@ -0,0 +1,88 @@ +CREATE TABLE IF NOT EXISTS "x_mention_sources" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL, + "company_id" uuid NOT NULL REFERENCES "companies"("id") ON DELETE cascade, + "source_key" text NOT NULL, + "account_user_id" text NOT NULL, + "account_handle" text, + "since_id" text, + "monthly_budget_cents" integer DEFAULT 5000 NOT NULL, + "per_run_budget_cents" integer DEFAULT 500 NOT NULL, + "budget_paused_at" timestamp with time zone, + "budget_pause_reason" text, + "rate_limit_reset_at" timestamp with time zone, + "created_at" timestamp with time zone DEFAULT now() NOT NULL, + "updated_at" timestamp with time zone DEFAULT now() NOT NULL +); + +CREATE UNIQUE INDEX IF NOT EXISTS "x_mention_sources_company_source_uq" + ON "x_mention_sources" ("company_id", "source_key"); +CREATE INDEX IF NOT EXISTS "x_mention_sources_company_account_idx" + ON "x_mention_sources" ("company_id", "account_user_id"); +CREATE INDEX IF NOT EXISTS "x_mention_sources_budget_paused_idx" + ON "x_mention_sources" ("company_id", "budget_paused_at"); + +CREATE TABLE IF NOT EXISTS "x_mention_author_allowlist" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL, + "company_id" uuid NOT NULL REFERENCES "companies"("id") ON DELETE cascade, + "x_user_id" text NOT NULL, + "handle" text, + "is_active" boolean DEFAULT true NOT NULL, + "created_at" timestamp with time zone DEFAULT now() NOT NULL, + "updated_at" timestamp with time zone DEFAULT now() NOT NULL +); + +CREATE UNIQUE INDEX IF NOT EXISTS "x_mention_allowlist_company_user_uq" + ON "x_mention_author_allowlist" ("company_id", "x_user_id"); +CREATE INDEX IF NOT EXISTS "x_mention_allowlist_company_active_idx" + ON "x_mention_author_allowlist" ("company_id", "is_active"); + +CREATE TABLE IF NOT EXISTS "x_mentions" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL, + "company_id" uuid NOT NULL REFERENCES "companies"("id") ON DELETE cascade, + "source_id" uuid NOT NULL REFERENCES "x_mention_sources"("id") ON DELETE cascade, + "tweet_id" text NOT NULL, + "author_user_id" text NOT NULL, + "author_handle" text, + "text" text DEFAULT '' NOT NULL, + "mentioned_at" timestamp with time zone, + "raw" jsonb DEFAULT '{}'::jsonb NOT NULL, + "gate_status" text DEFAULT 'stored' NOT NULL, + "hydration_status" text DEFAULT 'none' NOT NULL, + "manual_approved_at" timestamp with time zone, + "queued_at" timestamp with time zone, + "hydrated_at" timestamp with time zone, + "created_at" timestamp with time zone DEFAULT now() NOT NULL, + "updated_at" timestamp with time zone DEFAULT now() NOT NULL +); + +CREATE UNIQUE INDEX IF NOT EXISTS "x_mentions_company_tweet_uq" + ON "x_mentions" ("company_id", "tweet_id"); +CREATE INDEX IF NOT EXISTS "x_mentions_source_tweet_idx" + ON "x_mentions" ("source_id", "tweet_id"); +CREATE INDEX IF NOT EXISTS "x_mentions_company_gate_idx" + ON "x_mentions" ("company_id", "gate_status"); +CREATE INDEX IF NOT EXISTS "x_mentions_company_hydration_idx" + ON "x_mentions" ("company_id", "hydration_status"); +CREATE INDEX IF NOT EXISTS "x_mentions_company_author_idx" + ON "x_mentions" ("company_id", "author_user_id"); + +CREATE TABLE IF NOT EXISTS "x_mention_budget_ledger" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL, + "company_id" uuid NOT NULL REFERENCES "companies"("id") ON DELETE cascade, + "source_id" uuid NOT NULL REFERENCES "x_mention_sources"("id") ON DELETE cascade, + "mention_id" uuid REFERENCES "x_mentions"("id") ON DELETE cascade, + "operation" text NOT NULL, + "estimated_cost_cents" integer NOT NULL, + "actual_cost_cents" integer, + "status" text DEFAULT 'recorded' NOT NULL, + "failure_reason" text, + "occurred_at" timestamp with time zone DEFAULT now() NOT NULL, + "created_at" timestamp with time zone DEFAULT now() NOT NULL +); + +CREATE INDEX IF NOT EXISTS "x_mention_budget_company_occurred_idx" + ON "x_mention_budget_ledger" ("company_id", "occurred_at"); +CREATE INDEX IF NOT EXISTS "x_mention_budget_source_occurred_idx" + ON "x_mention_budget_ledger" ("source_id", "occurred_at"); +CREATE INDEX IF NOT EXISTS "x_mention_budget_operation_idx" + ON "x_mention_budget_ledger" ("company_id", "operation"); diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 6c63f8b5d9..907795d0c1 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -876,6 +876,13 @@ "when": 1782440000000, "tag": "0124_agent_api_key_scope_config", "breakpoints": true + }, + { + "idx": 125, + "version": "7", + "when": 1782440100000, + "tag": "0125_x_mention_poller", + "breakpoints": true } ] } diff --git a/packages/db/src/schema/index.ts b/packages/db/src/schema/index.ts index 8bd5f0bfab..72cf3ea1f0 100644 --- a/packages/db/src/schema/index.ts +++ b/packages/db/src/schema/index.ts @@ -38,6 +38,12 @@ export { issueRecoveryActions } from "./issue_recovery_actions.js"; export { issueReferenceMentions } from "./issue_reference_mentions.js"; export { externalObjects } from "./external_objects.js"; export { externalObjectMentions } from "./external_object_mentions.js"; +export { + xMentionAuthorAllowlist, + xMentionBudgetLedger, + xMentionSources, + xMentions, +} from "./x_mentions.js"; export { issueRelations } from "./issue_relations.js"; export { routines, routineRevisions, routineTriggers, routineRuns } from "./routines.js"; export { pipelines, pipelineStages, pipelineTransitions } from "./pipelines.js"; diff --git a/packages/db/src/schema/x_mentions.ts b/packages/db/src/schema/x_mentions.ts new file mode 100644 index 0000000000..d6e79dc480 --- /dev/null +++ b/packages/db/src/schema/x_mentions.ts @@ -0,0 +1,104 @@ +import { + boolean, + index, + integer, + jsonb, + pgTable, + text, + timestamp, + uniqueIndex, + uuid, +} from "drizzle-orm/pg-core"; +import { companies } from "./companies.js"; + +export const xMentionSources = pgTable( + "x_mention_sources", + { + id: uuid("id").primaryKey().defaultRandom(), + companyId: uuid("company_id").notNull().references(() => companies.id, { onDelete: "cascade" }), + sourceKey: text("source_key").notNull(), + accountUserId: text("account_user_id").notNull(), + accountHandle: text("account_handle"), + sinceId: text("since_id"), + monthlyBudgetCents: integer("monthly_budget_cents").notNull().default(5000), + perRunBudgetCents: integer("per_run_budget_cents").notNull().default(500), + budgetPausedAt: timestamp("budget_paused_at", { withTimezone: true }), + budgetPauseReason: text("budget_pause_reason"), + rateLimitResetAt: timestamp("rate_limit_reset_at", { withTimezone: true }), + createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(), + updatedAt: timestamp("updated_at", { withTimezone: true }).notNull().defaultNow(), + }, + (table) => ({ + companySourceUq: uniqueIndex("x_mention_sources_company_source_uq").on(table.companyId, table.sourceKey), + companyAccountIdx: index("x_mention_sources_company_account_idx").on(table.companyId, table.accountUserId), + budgetPausedIdx: index("x_mention_sources_budget_paused_idx").on(table.companyId, table.budgetPausedAt), + }), +); + +export const xMentionAuthorAllowlist = pgTable( + "x_mention_author_allowlist", + { + id: uuid("id").primaryKey().defaultRandom(), + companyId: uuid("company_id").notNull().references(() => companies.id, { onDelete: "cascade" }), + xUserId: text("x_user_id").notNull(), + handle: text("handle"), + isActive: boolean("is_active").notNull().default(true), + createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(), + updatedAt: timestamp("updated_at", { withTimezone: true }).notNull().defaultNow(), + }, + (table) => ({ + companyUserUq: uniqueIndex("x_mention_allowlist_company_user_uq").on(table.companyId, table.xUserId), + companyActiveIdx: index("x_mention_allowlist_company_active_idx").on(table.companyId, table.isActive), + }), +); + +export const xMentions = pgTable( + "x_mentions", + { + id: uuid("id").primaryKey().defaultRandom(), + companyId: uuid("company_id").notNull().references(() => companies.id, { onDelete: "cascade" }), + sourceId: uuid("source_id").notNull().references(() => xMentionSources.id, { onDelete: "cascade" }), + tweetId: text("tweet_id").notNull(), + authorUserId: text("author_user_id").notNull(), + authorHandle: text("author_handle"), + text: text("text").notNull().default(""), + mentionedAt: timestamp("mentioned_at", { withTimezone: true }), + raw: jsonb("raw").$type>().notNull().default({}), + gateStatus: text("gate_status").notNull().default("stored"), + hydrationStatus: text("hydration_status").notNull().default("none"), + manualApprovedAt: timestamp("manual_approved_at", { withTimezone: true }), + queuedAt: timestamp("queued_at", { withTimezone: true }), + hydratedAt: timestamp("hydrated_at", { withTimezone: true }), + createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(), + updatedAt: timestamp("updated_at", { withTimezone: true }).notNull().defaultNow(), + }, + (table) => ({ + companyTweetUq: uniqueIndex("x_mentions_company_tweet_uq").on(table.companyId, table.tweetId), + sourceTweetIdx: index("x_mentions_source_tweet_idx").on(table.sourceId, table.tweetId), + companyGateIdx: index("x_mentions_company_gate_idx").on(table.companyId, table.gateStatus), + companyHydrationIdx: index("x_mentions_company_hydration_idx").on(table.companyId, table.hydrationStatus), + companyAuthorIdx: index("x_mentions_company_author_idx").on(table.companyId, table.authorUserId), + }), +); + +export const xMentionBudgetLedger = pgTable( + "x_mention_budget_ledger", + { + id: uuid("id").primaryKey().defaultRandom(), + companyId: uuid("company_id").notNull().references(() => companies.id, { onDelete: "cascade" }), + sourceId: uuid("source_id").notNull().references(() => xMentionSources.id, { onDelete: "cascade" }), + mentionId: uuid("mention_id").references(() => xMentions.id, { onDelete: "cascade" }), + operation: text("operation").notNull(), + estimatedCostCents: integer("estimated_cost_cents").notNull(), + actualCostCents: integer("actual_cost_cents"), + status: text("status").notNull().default("recorded"), + failureReason: text("failure_reason"), + occurredAt: timestamp("occurred_at", { withTimezone: true }).notNull().defaultNow(), + createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(), + }, + (table) => ({ + companyOccurredIdx: index("x_mention_budget_company_occurred_idx").on(table.companyId, table.occurredAt), + sourceOccurredIdx: index("x_mention_budget_source_occurred_idx").on(table.sourceId, table.occurredAt), + operationIdx: index("x_mention_budget_operation_idx").on(table.companyId, table.operation), + }), +); diff --git a/server/src/__tests__/x-mention-poller.test.ts b/server/src/__tests__/x-mention-poller.test.ts new file mode 100644 index 0000000000..b65feae491 --- /dev/null +++ b/server/src/__tests__/x-mention-poller.test.ts @@ -0,0 +1,333 @@ +import { describe, expect, it, vi } from "vitest"; +import { + createXMentionPoller, + XRateLimitError, + type StoredXMention, + type XMentionAdapter, + type XMentionInput, + type XMentionSource, + type XMentionStore, +} from "../services/x-mention-poller.js"; + +function createMemoryStore(options: { + allowlistedUserIds?: string[]; + monthlyBudgetCents?: number; + perRunBudgetCents?: number; +} = {}) { + const source: XMentionSource = { + id: "source-1", + companyId: "company-1", + sourceKey: "paperclip", + accountUserId: "999", + accountHandle: "paperclip", + sinceId: null, + monthlyBudgetCents: options.monthlyBudgetCents ?? 5000, + perRunBudgetCents: options.perRunBudgetCents ?? 500, + budgetPausedAt: null, + budgetPauseReason: null, + rateLimitResetAt: null, + }; + const sources = new Map([[`${source.companyId}:${source.sourceKey}`, source]]); + const allowlist = new Set(options.allowlistedUserIds ?? []); + const mentions = new Map }>(); + const ledger: Array<{ + operation: string; + estimatedCostCents: number; + actualCostCents: number | null; + status: string; + failureReason: string | null; + }> = []; + const getSourceById = (sourceId: string) => { + const current = [...sources.values()].find((candidate) => candidate.id === sourceId); + if (!current) throw new Error(`Unknown source ${sourceId}`); + return current; + }; + + const store: XMentionStore = { + async getOrCreateSource(input) { + const key = `${input.companyId}:${input.sourceKey}`; + let current = sources.get(key); + if (!current) { + current = { + ...source, + id: `source-${sources.size + 1}`, + companyId: input.companyId, + sourceKey: input.sourceKey, + sinceId: null, + budgetPausedAt: null, + budgetPauseReason: null, + rateLimitResetAt: null, + }; + sources.set(key, current); + } + current.companyId = input.companyId; + current.sourceKey = input.sourceKey; + current.accountUserId = input.accountUserId; + current.accountHandle = input.accountHandle ?? null; + if (input.monthlyBudgetCents !== undefined) current.monthlyBudgetCents = input.monthlyBudgetCents; + if (input.perRunBudgetCents !== undefined) current.perRunBudgetCents = input.perRunBudgetCents; + return current; + }, + async updateSourceCursor(input) { + getSourceById(input.sourceId).sinceId = input.sinceId; + }, + async pauseSourceForBudget(input) { + const current = getSourceById(input.sourceId); + current.budgetPausedAt = input.now; + current.budgetPauseReason = input.reason; + }, + async markSourceRateLimited(input) { + getSourceById(input.sourceId).rateLimitResetAt = input.resetAt; + }, + async isAuthorAllowlisted(input) { + return allowlist.has(input.xUserId); + }, + async upsertMention(input) { + const existing = mentions.get(input.mention.tweetId); + if (existing) { + const approvedByExistingManualApproval = existing.manualApprovedAt !== null; + existing.sourceId = input.sourceId; + existing.authorUserId = input.mention.authorUserId; + existing.authorHandle = input.mention.authorHandle ?? null; + existing.gateStatus = approvedByExistingManualApproval ? "approved" : input.gateStatus; + existing.hydrationStatus = approvedByExistingManualApproval ? "queued" : input.hydrationStatus; + return { mention: existing, inserted: false }; + } + const mention: StoredXMention = { + id: `mention-${mentions.size + 1}`, + companyId: input.companyId, + sourceId: input.sourceId, + tweetId: input.mention.tweetId, + authorUserId: input.mention.authorUserId, + authorHandle: input.mention.authorHandle ?? null, + gateStatus: input.gateStatus, + hydrationStatus: input.hydrationStatus, + manualApprovedAt: null, + }; + mentions.set(input.mention.tweetId, mention); + return { mention, inserted: true }; + }, + async listQueuedHydration(input) { + return [...mentions.values()] + .filter((mention) => mention.companyId === input.companyId && mention.sourceId === input.sourceId && mention.hydrationStatus === "queued") + .slice(0, input.limit); + }, + async markHydrated(input) { + const mention = [...mentions.values()].find((candidate) => candidate.id === input.mentionId); + if (mention) { + mention.hydrationStatus = "hydrated"; + mention.raw = input.data; + } + }, + async sumBudgetSince() { + return ledger + .filter((row) => row.status === "recorded") + .reduce((sum, row) => sum + (row.actualCostCents ?? 0), 0); + }, + async recordBudget(input) { + ledger.push({ + operation: input.operation, + estimatedCostCents: input.estimatedCostCents, + actualCostCents: input.actualCostCents ?? null, + status: input.status ?? "recorded", + failureReason: input.failureReason ?? null, + }); + }, + }; + + return { store, source, sources, mentions, ledger, allowlist }; +} + +function createAdapter(input: { + mentions?: XMentionInput[]; + estimate?: number | null | Partial>; + fetchError?: Error; +} = {}) { + const estimate = input.estimate ?? 1; + const adapter: XMentionAdapter = { + estimateOperation: vi.fn(({ operation }) => { + if (typeof estimate === "object") return estimate[operation] ?? null; + return estimate; + }), + fetchMentions: vi.fn(async () => { + if (input.fetchError) throw input.fetchError; + return { mentions: input.mentions ?? [], nextSinceId: input.mentions?.at(-1)?.tweetId ?? null }; + }), + hydrateMention: vi.fn(async ({ tweetId }) => ({ + thread: { tweetId }, + replies: { count: 1 }, + media: { count: 0 }, + })), + }; + return adapter; +} + +const pollInput = { + companyId: "company-1", + sourceKey: "paperclip", + accountUserId: "999", + accountHandle: "paperclip", +}; + +describe("x mention poller", () => { + it("stores every mention but only queues allowlisted authors by stable X user id", async () => { + const memory = createMemoryStore({ allowlistedUserIds: ["42"] }); + const adapter = createAdapter({ + mentions: [ + { tweetId: "100", authorUserId: "42", authorHandle: "allowed", text: "@paperclip hi" }, + { tweetId: "101", authorUserId: "77", authorHandle: "blocked", text: "@paperclip nope" }, + ], + }); + const poller = createXMentionPoller({ store: memory.store, adapter, now: () => new Date("2026-06-01T00:00:00Z") }); + + const result = await poller.pollMentions(pollInput); + + expect(result).toMatchObject({ status: "ok", stored: 2, queued: 1, duplicates: 0, cursor: "101" }); + expect(memory.mentions.get("100")).toMatchObject({ gateStatus: "queued", hydrationStatus: "queued" }); + expect(memory.mentions.get("101")).toMatchObject({ gateStatus: "stored", hydrationStatus: "none" }); + }); + + it("uses idempotent upsert and cursoring to prevent duplicate intake rows across retries", async () => { + const memory = createMemoryStore({ allowlistedUserIds: ["42"] }); + const adapter = createAdapter({ + mentions: [{ tweetId: "100", authorUserId: "42", text: "@paperclip first" }], + }); + const poller = createXMentionPoller({ store: memory.store, adapter }); + + await expect(poller.pollMentions(pollInput)).resolves.toMatchObject({ stored: 1, duplicates: 0, cursor: "100" }); + await expect(poller.pollMentions(pollInput)).resolves.toMatchObject({ stored: 0, duplicates: 1, cursor: "100" }); + + expect(memory.mentions).toHaveLength(1); + expect(adapter.fetchMentions).toHaveBeenLastCalledWith(expect.objectContaining({ sinceId: "100" })); + }); + + it("moves duplicate tweets to the latest source so queued hydration can find them", async () => { + const memory = createMemoryStore({ allowlistedUserIds: ["42"] }); + const firstAdapter = createAdapter({ + mentions: [{ tweetId: "100", authorUserId: "42", text: "@paperclip @support please hydrate" }], + }); + const secondAdapter = createAdapter({ + mentions: [{ tweetId: "100", authorUserId: "42", text: "@paperclip @support please hydrate" }], + }); + const firstPoller = createXMentionPoller({ store: memory.store, adapter: firstAdapter }); + const secondPoller = createXMentionPoller({ store: memory.store, adapter: secondAdapter }); + const supportInput = { + ...pollInput, + sourceKey: "support", + accountUserId: "998", + accountHandle: "support", + }; + + await expect(firstPoller.pollMentions(pollInput)).resolves.toMatchObject({ stored: 1, queued: 1 }); + expect(memory.mentions.get("100")).toMatchObject({ sourceId: "source-1", hydrationStatus: "queued" }); + + await expect(secondPoller.pollMentions(supportInput)).resolves.toMatchObject({ stored: 0, duplicates: 1, queued: 1 }); + expect(memory.mentions.get("100")).toMatchObject({ sourceId: "source-2", hydrationStatus: "queued" }); + + await expect(secondPoller.hydrateQueuedMentions(supportInput)).resolves.toMatchObject({ status: "ok", hydrated: 1 }); + expect(secondAdapter.hydrateMention).toHaveBeenCalledWith({ + tweetId: "100", + operations: ["hydrate_thread", "hydrate_replies", "hydrate_media"], + }); + }); + + it("does not advance since_id when a fetch fails so the next run retries the same cursor", async () => { + const memory = createMemoryStore(); + memory.source.sinceId = "99"; + const adapter = createAdapter({ fetchError: new Error("network down") }); + const poller = createXMentionPoller({ store: memory.store, adapter }); + + await expect(poller.pollMentions(pollInput)).rejects.toThrow("network down"); + + expect(memory.source.sinceId).toBe("99"); + }); + + it("records rate-limit backoff without advancing the cursor", async () => { + const memory = createMemoryStore(); + memory.source.sinceId = "99"; + const resetAt = new Date("2026-06-01T00:15:00Z"); + const adapter = createAdapter({ fetchError: new XRateLimitError(resetAt) }); + const poller = createXMentionPoller({ store: memory.store, adapter }); + + const result = await poller.pollMentions(pollInput); + + expect(result).toMatchObject({ status: "rate_limited", cursor: "99", rateLimitResetAt: resetAt }); + expect(memory.source.rateLimitResetAt).toEqual(resetAt); + }); + + it("fails closed and pauses the source when poll cost cannot be estimated", async () => { + const memory = createMemoryStore(); + const adapter = createAdapter({ estimate: { poll: null } }); + const poller = createXMentionPoller({ store: memory.store, adapter }); + + const result = await poller.pollMentions(pollInput); + + expect(result).toMatchObject({ status: "budget_paused", reason: "missing_cost_estimate:poll" }); + expect(adapter.fetchMentions).not.toHaveBeenCalled(); + expect(memory.source.budgetPauseReason).toBe("missing_cost_estimate:poll"); + expect(memory.ledger).toContainEqual(expect.objectContaining({ + operation: "poll", + status: "rejected", + failureReason: "missing_cost_estimate:poll", + })); + }); + + it("treats zero-cost estimates as unsafe so misconfigured adapters cannot bypass budgets", async () => { + const memory = createMemoryStore(); + const adapter = createAdapter({ estimate: { poll: 0 } }); + const poller = createXMentionPoller({ store: memory.store, adapter }); + + const result = await poller.pollMentions(pollInput); + + expect(result).toMatchObject({ status: "budget_paused", reason: "missing_cost_estimate:poll" }); + expect(adapter.fetchMentions).not.toHaveBeenCalled(); + expect(memory.ledger).toContainEqual(expect.objectContaining({ + operation: "poll", + status: "rejected", + estimatedCostCents: 0, + failureReason: "missing_cost_estimate:poll", + })); + }); + + it("does not hydrate while the source is inside an active rate-limit window", async () => { + const memory = createMemoryStore({ allowlistedUserIds: ["42"] }); + const resetAt = new Date("2026-06-01T00:15:00Z"); + memory.source.rateLimitResetAt = resetAt; + const adapter = createAdapter({ + mentions: [{ tweetId: "100", authorUserId: "42", text: "@paperclip hydrate" }], + }); + const poller = createXMentionPoller({ + store: memory.store, + adapter, + now: () => new Date("2026-06-01T00:00:00Z"), + }); + + const result = await poller.hydrateQueuedMentions(pollInput); + + expect(result).toMatchObject({ status: "rate_limited", rateLimitResetAt: resetAt, hydrated: 0 }); + expect(adapter.hydrateMention).not.toHaveBeenCalled(); + }); + + it("applies per-run caps across hydration thread, replies, and media operations", async () => { + const memory = createMemoryStore({ allowlistedUserIds: ["42"], perRunBudgetCents: 3 }); + const adapter = createAdapter({ + estimate: { + poll: 1, + hydrate_thread: 1, + hydrate_replies: 1, + hydrate_media: 2, + }, + mentions: [{ tweetId: "100", authorUserId: "42", text: "@paperclip hydrate" }], + }); + const poller = createXMentionPoller({ store: memory.store, adapter }); + + await poller.pollMentions(pollInput); + const result = await poller.hydrateQueuedMentions(pollInput); + + expect(result).toMatchObject({ status: "budget_paused", reason: "per_run_budget_exceeded:hydrate_media", hydrated: 0 }); + expect(adapter.hydrateMention).not.toHaveBeenCalled(); + expect(memory.ledger.map((row) => row.operation)).toEqual(["poll", "hydrate_media"]); + expect(memory.ledger.filter((row) => row.status === "recorded")).toHaveLength(1); + expect(memory.ledger.at(-1)).toMatchObject({ status: "rejected", failureReason: "per_run_budget_exceeded:hydrate_media" }); + }); +}); diff --git a/server/src/services/x-mention-poller.ts b/server/src/services/x-mention-poller.ts new file mode 100644 index 0000000000..12ef2d707c --- /dev/null +++ b/server/src/services/x-mention-poller.ts @@ -0,0 +1,694 @@ +import { execFile as nodeExecFile } from "node:child_process"; +import { promisify } from "node:util"; +import { and, eq, gte, sql } from "drizzle-orm"; +import type { Db } from "@paperclipai/db"; +import { + xMentionAuthorAllowlist, + xMentionBudgetLedger, + xMentionSources, + xMentions, +} from "@paperclipai/db"; + +const execFileAsync = promisify(nodeExecFile); + +export type XMentionOperation = "poll" | "hydrate_thread" | "hydrate_replies" | "hydrate_media"; +export type XMentionGateStatus = "stored" | "queued" | "approved"; +export type XMentionHydrationStatus = "none" | "queued" | "hydrated"; + +export interface XMentionSource { + id: string; + companyId: string; + sourceKey: string; + accountUserId: string; + accountHandle: string | null; + sinceId: string | null; + monthlyBudgetCents: number; + perRunBudgetCents: number; + budgetPausedAt: Date | null; + budgetPauseReason: string | null; + rateLimitResetAt: Date | null; +} + +export interface XMentionInput { + tweetId: string; + authorUserId: string; + authorHandle?: string | null; + text?: string | null; + mentionedAt?: Date | string | null; + raw?: Record; +} + +export interface StoredXMention { + id: string; + companyId: string; + sourceId: string; + tweetId: string; + authorUserId: string; + authorHandle: string | null; + gateStatus: XMentionGateStatus; + hydrationStatus: XMentionHydrationStatus; + manualApprovedAt: Date | null; +} + +export interface XMentionFetchResult { + mentions: XMentionInput[]; + nextSinceId?: string | null; + rateLimitResetAt?: Date | null; +} + +export interface XHydrationResult { + thread?: Record | null; + replies?: Record | null; + media?: Record | null; +} + +export interface XMentionAdapter { + estimateOperation(input: { + operation: XMentionOperation; + source: XMentionSource; + mention?: StoredXMention | null; + }): Promise | number | null; + fetchMentions(input: { + accountUserId: string; + sinceId: string | null; + limit: number; + }): Promise; + hydrateMention?(input: { + tweetId: string; + operations: Exclude[]; + }): Promise; +} + +export interface XMentionStore { + getOrCreateSource(input: { + companyId: string; + sourceKey: string; + accountUserId: string; + accountHandle?: string | null; + monthlyBudgetCents?: number; + perRunBudgetCents?: number; + }): Promise; + updateSourceCursor(input: { sourceId: string; sinceId: string | null; now: Date }): Promise; + pauseSourceForBudget(input: { sourceId: string; reason: string; now: Date }): Promise; + markSourceRateLimited(input: { sourceId: string; resetAt: Date; now: Date }): Promise; + isAuthorAllowlisted(input: { companyId: string; xUserId: string }): Promise; + upsertMention(input: { + companyId: string; + sourceId: string; + mention: XMentionInput; + gateStatus: XMentionGateStatus; + hydrationStatus: XMentionHydrationStatus; + now: Date; + }): Promise<{ mention: StoredXMention; inserted: boolean }>; + listQueuedHydration(input: { companyId: string; sourceId: string; limit: number }): Promise; + markHydrated(input: { mentionId: string; now: Date; data?: Record }): Promise; + sumBudgetSince(input: { sourceId: string; since: Date }): Promise; + recordBudget(input: { + companyId: string; + sourceId: string; + mentionId?: string | null; + operation: XMentionOperation; + estimatedCostCents: number; + actualCostCents?: number | null; + status?: "recorded" | "rejected"; + failureReason?: string | null; + now: Date; + }): Promise; +} + +export class XRateLimitError extends Error { + readonly resetAt: Date; + + constructor(resetAt: Date, message = "X API rate limit exceeded") { + super(message); + this.name = "XRateLimitError"; + this.resetAt = resetAt; + } +} + +export class XBudgetPausedError extends Error { + readonly reason: string; + + constructor(reason: string) { + super(reason); + this.name = "XBudgetPausedError"; + this.reason = reason; + } +} + +function toDate(value: Date | string | null | undefined) { + if (!value) return null; + return value instanceof Date ? value : new Date(value); +} + +function monthStartUtc(now: Date) { + return new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), 1)); +} + +function maxTweetId(current: string | null, mentions: XMentionInput[]) { + let max = current; + for (const mention of mentions) { + if (!max || compareTweetIds(mention.tweetId, max) > 0) { + max = mention.tweetId; + } + } + return max; +} + +function compareTweetIds(a: string, b: string) { + try { + const left = BigInt(a); + const right = BigInt(b); + return left === right ? 0 : left > right ? 1 : -1; + } catch { + return a.localeCompare(b); + } +} + +function assertSafeEstimate(value: number | null, operation: XMentionOperation) { + if (value === null || !Number.isFinite(value) || value <= 0) { + return `missing_cost_estimate:${operation}`; + } + return null; +} + +export function createXMentionPoller(options: { + store: XMentionStore; + adapter: XMentionAdapter; + now?: () => Date; +}) { + const { store, adapter } = options; + const now = options.now ?? (() => new Date()); + + async function checkBudget(input: { + source: XMentionSource; + operation: XMentionOperation; + mention?: StoredXMention | null; + runSpendCents: number; + unrecordedSpendCents?: number; + }) { + const at = now(); + const estimate = await adapter.estimateOperation({ + operation: input.operation, + source: input.source, + mention: input.mention ?? null, + }); + const unsafeReason = assertSafeEstimate(estimate, input.operation); + if (unsafeReason) { + await store.recordBudget({ + companyId: input.source.companyId, + sourceId: input.source.id, + mentionId: input.mention?.id ?? null, + operation: input.operation, + estimatedCostCents: 0, + status: "rejected", + failureReason: unsafeReason, + now: at, + }); + await store.pauseSourceForBudget({ sourceId: input.source.id, reason: unsafeReason, now: at }); + throw new XBudgetPausedError(unsafeReason); + } + + const estimatedCostCents = Math.ceil(estimate as number); + if (input.runSpendCents + estimatedCostCents > input.source.perRunBudgetCents) { + const reason = `per_run_budget_exceeded:${input.operation}`; + await store.recordBudget({ + companyId: input.source.companyId, + sourceId: input.source.id, + mentionId: input.mention?.id ?? null, + operation: input.operation, + estimatedCostCents, + status: "rejected", + failureReason: reason, + now: at, + }); + await store.pauseSourceForBudget({ sourceId: input.source.id, reason, now: at }); + throw new XBudgetPausedError(reason); + } + + const monthSpend = await store.sumBudgetSince({ + sourceId: input.source.id, + since: monthStartUtc(at), + }); + if (monthSpend + (input.unrecordedSpendCents ?? 0) + estimatedCostCents > input.source.monthlyBudgetCents) { + const reason = `monthly_budget_exceeded:${input.operation}`; + await store.recordBudget({ + companyId: input.source.companyId, + sourceId: input.source.id, + mentionId: input.mention?.id ?? null, + operation: input.operation, + estimatedCostCents, + status: "rejected", + failureReason: reason, + now: at, + }); + await store.pauseSourceForBudget({ sourceId: input.source.id, reason, now: at }); + throw new XBudgetPausedError(reason); + } + + return estimatedCostCents; + } + + async function recordSuccessfulBudget(input: { + source: XMentionSource; + operation: XMentionOperation; + estimatedCostCents: number; + mention?: StoredXMention | null; + }) { + await store.recordBudget({ + companyId: input.source.companyId, + sourceId: input.source.id, + mentionId: input.mention?.id ?? null, + operation: input.operation, + estimatedCostCents: input.estimatedCostCents, + actualCostCents: input.estimatedCostCents, + now: now(), + }); + } + + async function reserveBudget(input: { + source: XMentionSource; + operation: XMentionOperation; + mention?: StoredXMention | null; + runSpendCents: number; + }) { + const estimatedCostCents = await checkBudget(input); + await recordSuccessfulBudget({ ...input, estimatedCostCents }); + return estimatedCostCents; + } + + async function pollMentions(input: { + companyId: string; + sourceKey: string; + accountUserId: string; + accountHandle?: string | null; + limit?: number; + monthlyBudgetCents?: number; + perRunBudgetCents?: number; + }) { + const source = await store.getOrCreateSource(input); + if (source.budgetPausedAt) { + return { + status: "budget_paused" as const, + reason: source.budgetPauseReason ?? "budget_paused", + stored: 0, + queued: 0, + duplicates: 0, + cursor: source.sinceId, + }; + } + if (source.rateLimitResetAt && source.rateLimitResetAt > now()) { + return { + status: "rate_limited" as const, + rateLimitResetAt: source.rateLimitResetAt, + stored: 0, + queued: 0, + duplicates: 0, + cursor: source.sinceId, + }; + } + + let runSpendCents = 0; + try { + runSpendCents += await reserveBudget({ source, operation: "poll", runSpendCents }); + const result = await adapter.fetchMentions({ + accountUserId: source.accountUserId, + sinceId: source.sinceId, + limit: input.limit ?? 100, + }); + let stored = 0; + let queued = 0; + let duplicates = 0; + for (const mention of result.mentions) { + const approved = await store.isAuthorAllowlisted({ + companyId: source.companyId, + xUserId: mention.authorUserId, + }); + const upserted = await store.upsertMention({ + companyId: source.companyId, + sourceId: source.id, + mention, + gateStatus: approved ? "queued" : "stored", + hydrationStatus: approved ? "queued" : "none", + now: now(), + }); + if (upserted.inserted) { + stored += 1; + } else { + duplicates += 1; + } + if (upserted.mention.hydrationStatus === "queued") { + queued += 1; + } + } + const nextSinceId = result.nextSinceId ?? maxTweetId(source.sinceId, result.mentions); + await store.updateSourceCursor({ sourceId: source.id, sinceId: nextSinceId, now: now() }); + if (result.rateLimitResetAt) { + await store.markSourceRateLimited({ sourceId: source.id, resetAt: result.rateLimitResetAt, now: now() }); + } + return { + status: "ok" as const, + stored, + queued, + duplicates, + cursor: nextSinceId, + }; + } catch (err) { + if (err instanceof XBudgetPausedError) { + return { + status: "budget_paused" as const, + reason: err.reason, + stored: 0, + queued: 0, + duplicates: 0, + cursor: source.sinceId, + }; + } + if (err instanceof XRateLimitError) { + await store.markSourceRateLimited({ sourceId: source.id, resetAt: err.resetAt, now: now() }); + return { + status: "rate_limited" as const, + rateLimitResetAt: err.resetAt, + stored: 0, + queued: 0, + duplicates: 0, + cursor: source.sinceId, + }; + } + throw err; + } + } + + async function hydrateQueuedMentions(input: { + companyId: string; + sourceKey: string; + accountUserId: string; + accountHandle?: string | null; + limit?: number; + operations?: Exclude[]; + }) { + const source = await store.getOrCreateSource(input); + if (source.budgetPausedAt) { + return { status: "budget_paused" as const, reason: source.budgetPauseReason ?? "budget_paused", hydrated: 0 }; + } + if (source.rateLimitResetAt && source.rateLimitResetAt > now()) { + return { status: "rate_limited" as const, rateLimitResetAt: source.rateLimitResetAt, hydrated: 0 }; + } + if (!adapter.hydrateMention) { + return { status: "ok" as const, hydrated: 0 }; + } + const operations = input.operations ?? ["hydrate_thread", "hydrate_replies", "hydrate_media"]; + const queued = await store.listQueuedHydration({ + companyId: input.companyId, + sourceId: source.id, + limit: input.limit ?? 25, + }); + let runSpendCents = 0; + let hydrated = 0; + try { + for (const mention of queued) { + const budgets: Array<{ operation: Exclude; estimatedCostCents: number }> = []; + let pendingSpendCents = 0; + for (const operation of operations) { + const estimatedCostCents = await checkBudget({ + source, + operation, + mention, + runSpendCents: runSpendCents + pendingSpendCents, + unrecordedSpendCents: pendingSpendCents, + }); + budgets.push({ operation, estimatedCostCents }); + pendingSpendCents += estimatedCostCents; + } + const data = await adapter.hydrateMention({ tweetId: mention.tweetId, operations }); + for (const budget of budgets) { + await recordSuccessfulBudget({ source, mention, ...budget }); + } + runSpendCents += pendingSpendCents; + await store.markHydrated({ mentionId: mention.id, now: now(), data: data as Record }); + hydrated += 1; + } + return { status: "ok" as const, hydrated }; + } catch (err) { + if (err instanceof XBudgetPausedError) { + return { status: "budget_paused" as const, reason: err.reason, hydrated }; + } + if (err instanceof XRateLimitError) { + await store.markSourceRateLimited({ sourceId: source.id, resetAt: err.resetAt, now: now() }); + return { status: "rate_limited" as const, rateLimitResetAt: err.resetAt, hydrated }; + } + throw err; + } + } + + return { pollMentions, hydrateQueuedMentions }; +} + +export function createDbXMentionStore(db: Db): XMentionStore { + return { + async getOrCreateSource(input) { + const now = new Date(); + const rows = await db + .insert(xMentionSources) + .values({ + companyId: input.companyId, + sourceKey: input.sourceKey, + accountUserId: input.accountUserId, + accountHandle: input.accountHandle ?? null, + monthlyBudgetCents: input.monthlyBudgetCents ?? 5000, + perRunBudgetCents: input.perRunBudgetCents ?? 500, + updatedAt: now, + }) + .onConflictDoUpdate({ + target: [xMentionSources.companyId, xMentionSources.sourceKey], + set: { + accountUserId: input.accountUserId, + accountHandle: input.accountHandle ?? null, + monthlyBudgetCents: input.monthlyBudgetCents ?? sql`${xMentionSources.monthlyBudgetCents}`, + perRunBudgetCents: input.perRunBudgetCents ?? sql`${xMentionSources.perRunBudgetCents}`, + updatedAt: now, + }, + }) + .returning(); + return rows[0] as XMentionSource; + }, + async updateSourceCursor(input) { + await db + .update(xMentionSources) + .set({ sinceId: input.sinceId, updatedAt: input.now }) + .where(eq(xMentionSources.id, input.sourceId)); + }, + async pauseSourceForBudget(input) { + await db + .update(xMentionSources) + .set({ budgetPausedAt: input.now, budgetPauseReason: input.reason, updatedAt: input.now }) + .where(eq(xMentionSources.id, input.sourceId)); + }, + async markSourceRateLimited(input) { + await db + .update(xMentionSources) + .set({ rateLimitResetAt: input.resetAt, updatedAt: input.now }) + .where(eq(xMentionSources.id, input.sourceId)); + }, + async isAuthorAllowlisted(input) { + const rows = await db + .select({ id: xMentionAuthorAllowlist.id }) + .from(xMentionAuthorAllowlist) + .where(and( + eq(xMentionAuthorAllowlist.companyId, input.companyId), + eq(xMentionAuthorAllowlist.xUserId, input.xUserId), + eq(xMentionAuthorAllowlist.isActive, true), + )); + return rows.length > 0; + }, + async upsertMention(input) { + const mentionedAt = toDate(input.mention.mentionedAt); + const rows = await db + .insert(xMentions) + .values({ + companyId: input.companyId, + sourceId: input.sourceId, + tweetId: input.mention.tweetId, + authorUserId: input.mention.authorUserId, + authorHandle: input.mention.authorHandle ?? null, + text: input.mention.text ?? "", + mentionedAt, + raw: input.mention.raw ?? {}, + gateStatus: input.gateStatus, + hydrationStatus: input.hydrationStatus, + queuedAt: input.hydrationStatus === "queued" ? input.now : null, + createdAt: input.now, + updatedAt: input.now, + }) + .onConflictDoUpdate({ + target: [xMentions.companyId, xMentions.tweetId], + set: { + sourceId: input.sourceId, + authorUserId: input.mention.authorUserId, + authorHandle: input.mention.authorHandle ?? null, + text: input.mention.text ?? "", + mentionedAt, + raw: input.mention.raw ?? {}, + gateStatus: sql`case when ${xMentions.manualApprovedAt} is not null then 'approved' else ${input.gateStatus} end`, + hydrationStatus: sql`case when ${xMentions.manualApprovedAt} is not null then 'queued' else ${input.hydrationStatus} end`, + queuedAt: sql`case when ${xMentions.manualApprovedAt} is not null or ${input.hydrationStatus} = 'queued' then coalesce(${xMentions.queuedAt}, ${input.now}) else ${xMentions.queuedAt} end`, + updatedAt: input.now, + }, + }) + .returning(); + const mention = rows[0] as StoredXMention & { createdAt?: Date }; + return { + mention, + inserted: mention.createdAt?.getTime() === input.now.getTime(), + }; + }, + async listQueuedHydration(input) { + const rows = await db + .select() + .from(xMentions) + .where(and( + eq(xMentions.companyId, input.companyId), + eq(xMentions.sourceId, input.sourceId), + eq(xMentions.hydrationStatus, "queued"), + )) + .limit(input.limit); + return rows as StoredXMention[]; + }, + async markHydrated(input) { + await db + .update(xMentions) + .set({ + hydrationStatus: "hydrated", + hydratedAt: input.now, + raw: input.data ?? sql`${xMentions.raw}`, + updatedAt: input.now, + }) + .where(eq(xMentions.id, input.mentionId)); + }, + async sumBudgetSince(input) { + const rows = await db + .select({ total: sql`coalesce(sum(${xMentionBudgetLedger.actualCostCents}), 0)` }) + .from(xMentionBudgetLedger) + .where(and( + eq(xMentionBudgetLedger.sourceId, input.sourceId), + eq(xMentionBudgetLedger.status, "recorded"), + gte(xMentionBudgetLedger.occurredAt, input.since), + )); + return Number(rows[0]?.total ?? 0); + }, + async recordBudget(input) { + await db.insert(xMentionBudgetLedger).values({ + companyId: input.companyId, + sourceId: input.sourceId, + mentionId: input.mentionId ?? null, + operation: input.operation, + estimatedCostCents: input.estimatedCostCents, + actualCostCents: input.actualCostCents ?? null, + status: input.status ?? "recorded", + failureReason: input.failureReason ?? null, + occurredAt: input.now, + }); + }, + }; +} + +interface XApiMentionAdapterOptions { + bearerToken: string; + baseUrl?: string; + fetchImpl?: typeof fetch; + estimates?: Partial>; +} + +export function createXApiV2MentionAdapter(options: XApiMentionAdapterOptions): XMentionAdapter { + const fetchImpl = options.fetchImpl ?? fetch; + const baseUrl = options.baseUrl ?? "https://api.x.com/2"; + const estimates = options.estimates ?? { poll: 1, hydrate_thread: 1, hydrate_replies: 1, hydrate_media: 1 }; + + return { + estimateOperation({ operation }) { + return estimates[operation] ?? null; + }, + async fetchMentions({ accountUserId, sinceId, limit }) { + const url = new URL(`${baseUrl}/users/${encodeURIComponent(accountUserId)}/mentions`); + url.searchParams.set("max_results", String(Math.max(5, Math.min(limit, 100)))); + url.searchParams.set("tweet.fields", "author_id,created_at,conversation_id,referenced_tweets"); + url.searchParams.set("expansions", "author_id"); + url.searchParams.set("user.fields", "username"); + if (sinceId) url.searchParams.set("since_id", sinceId); + const response = await fetchImpl(url, { + headers: { authorization: `Bearer ${options.bearerToken}` }, + }); + if (response.status === 429) { + const resetAt = Number(response.headers.get("x-rate-limit-reset")); + throw new XRateLimitError(Number.isFinite(resetAt) ? new Date(resetAt * 1000) : new Date(Date.now() + 15 * 60_000)); + } + if (!response.ok) { + throw new Error(`X mention fetch failed with HTTP ${response.status}`); + } + const body = await response.json() as { + data?: Array>; + includes?: { users?: Array<{ id?: string; username?: string }> }; + meta?: { newest_id?: string }; + }; + const users = new Map((body.includes?.users ?? []).map((user) => [String(user.id), user.username ?? null])); + return { + mentions: (body.data ?? []).map((tweet) => { + const authorUserId = String(tweet.author_id ?? ""); + return { + tweetId: String(tweet.id), + authorUserId, + authorHandle: users.get(authorUserId) ?? null, + text: typeof tweet.text === "string" ? tweet.text : "", + mentionedAt: typeof tweet.created_at === "string" ? tweet.created_at : null, + raw: tweet, + }; + }).filter((mention) => mention.tweetId && mention.authorUserId), + nextSinceId: body.meta?.newest_id ?? null, + }; + }, + }; +} + +interface XcMentionAdapterOptions { + command?: string; + execFile?: typeof nodeExecFile; + estimates?: Partial>; + timeoutMs?: number; +} + +export function createXcMentionAdapter(options: XcMentionAdapterOptions = {}): XMentionAdapter { + const command = options.command ?? "xc"; + const estimates = options.estimates ?? { poll: 1, hydrate_thread: 1, hydrate_replies: 1, hydrate_media: 1 }; + const run = options.execFile + ? promisify(options.execFile) + : execFileAsync; + + async function execJson(args: string[]) { + const { stdout } = await run(command, args, { timeout: options.timeoutMs ?? 30_000 }); + return JSON.parse(stdout.toString()) as Record; + } + + return { + estimateOperation({ operation }) { + return estimates[operation] ?? null; + }, + async fetchMentions({ accountUserId, sinceId, limit }) { + const args = ["mentions", "list", "--user-id", accountUserId, "--limit", String(limit), "--json"]; + if (sinceId) args.push("--since-id", sinceId); + const body = await execJson(args) as { + mentions?: XMentionInput[]; + nextSinceId?: string | null; + rateLimitResetAt?: string | null; + }; + return { + mentions: body.mentions ?? [], + nextSinceId: body.nextSinceId ?? null, + rateLimitResetAt: toDate(body.rateLimitResetAt ?? null), + }; + }, + async hydrateMention({ tweetId, operations }) { + const args = ["mentions", "hydrate", "--tweet-id", tweetId, "--operations", operations.join(","), "--json"]; + return await execJson(args) as XHydrationResult; + }, + }; +}