diff --git a/packages/db/src/migrations/0125_x_mention_poller.sql b/packages/db/src/migrations/0125_x_mention_poller.sql deleted file mode 100644 index 426d648f8b..0000000000 --- a/packages/db/src/migrations/0125_x_mention_poller.sql +++ /dev/null @@ -1,88 +0,0 @@ -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 907795d0c1..6c63f8b5d9 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -876,13 +876,6 @@ "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 72cf3ea1f0..8bd5f0bfab 100644 --- a/packages/db/src/schema/index.ts +++ b/packages/db/src/schema/index.ts @@ -38,12 +38,6 @@ 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 deleted file mode 100644 index d6e79dc480..0000000000 --- a/packages/db/src/schema/x_mentions.ts +++ /dev/null @@ -1,104 +0,0 @@ -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 deleted file mode 100644 index b65feae491..0000000000 --- a/server/src/__tests__/x-mention-poller.test.ts +++ /dev/null @@ -1,333 +0,0 @@ -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 deleted file mode 100644 index 12ef2d707c..0000000000 --- a/server/src/services/x-mention-poller.ts +++ /dev/null @@ -1,694 +0,0 @@ -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; - }, - }; -}