feat: add X mention poller backend (#8707)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Agents and operators increasingly need external event sources to become durable Paperclip work inputs > - X mentions are one such source, but intake needs to be safe before any downstream automation consumes them > - The backend needs stable source state, idempotent mention storage, author gating, rate-limit handling, and budget accounting > - This pull request adds the database contract and service layer for X mention polling and hydration queueing > - The benefit is that future X-triggered workflows can build on a controlled, test-covered ingestion path instead of calling the X API directly ## Linked Issues or Issue Description - No public GitHub issue exists for this exact backend extraction. - Problem: Paperclip does not yet have a durable, budget-aware backend path for ingesting X mentions as external work inputs. - Proposed solution: add X mention source, mention, allowlist, and budget ledger tables plus a poller service that stores mentions idempotently, queues only allowlisted authors for hydration, tracks cursor state, records spend decisions, and fails closed when cost estimates are unavailable. - Related but not duplicate: #8609, #8199, and #7316 touch internal mention wake behavior rather than X API mention ingestion. ## What Changed - Added X mention poller database tables and schema exports for sources, stored mentions, author allowlists, and budget ledger entries. - Added a server-side X mention poller service with cursoring, idempotent upsert behavior, allowlist gating, hydration queue handling, rate-limit backoff, and budget pause behavior. - Added focused Vitest coverage for intake gating, duplicate retries, cursor safety, rate limits, budget failures, and hydration budget pauses. ## Verification - `pnpm exec vitest run server/src/__tests__/x-mention-poller.test.ts` - `pnpm --filter @paperclipai/db typecheck` - `pnpm --filter @paperclipai/server typecheck` ## Risks - Migration ordering matters because this adds migration `0125_x_mention_poller.sql`; it should merge after the existing `0124` migrations on `master`. - The service is backend-only and adapter-driven in this PR, so product behavior should not change until callers wire it into a runtime path. - Budget accounting intentionally fails closed when estimates are missing, which may pause a source rather than risk unbounded API spend. ## Model Used - OpenAI Codex, GPT-5-based coding agent, tool-enabled local repository and terminal workflow. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Dotta
GitHub
parent
4c6c0c6ad0
commit
44e2ab53fe
|
|
@ -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");
|
||||
|
|
@ -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
|
||||
}
|
||||
]
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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";
|
||||
|
|
|
|||
|
|
@ -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<Record<string, unknown>>().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),
|
||||
}),
|
||||
);
|
||||
|
|
@ -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<string, XMentionSource>([[`${source.companyId}:${source.sourceKey}`, source]]);
|
||||
const allowlist = new Set(options.allowlistedUserIds ?? []);
|
||||
const mentions = new Map<string, StoredXMention & { raw?: Record<string, unknown> }>();
|
||||
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<Record<string, number | null>>;
|
||||
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" });
|
||||
});
|
||||
});
|
||||
|
|
@ -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<string, unknown>;
|
||||
}
|
||||
|
||||
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<string, unknown> | null;
|
||||
replies?: Record<string, unknown> | null;
|
||||
media?: Record<string, unknown> | null;
|
||||
}
|
||||
|
||||
export interface XMentionAdapter {
|
||||
estimateOperation(input: {
|
||||
operation: XMentionOperation;
|
||||
source: XMentionSource;
|
||||
mention?: StoredXMention | null;
|
||||
}): Promise<number | null> | number | null;
|
||||
fetchMentions(input: {
|
||||
accountUserId: string;
|
||||
sinceId: string | null;
|
||||
limit: number;
|
||||
}): Promise<XMentionFetchResult>;
|
||||
hydrateMention?(input: {
|
||||
tweetId: string;
|
||||
operations: Exclude<XMentionOperation, "poll">[];
|
||||
}): Promise<XHydrationResult>;
|
||||
}
|
||||
|
||||
export interface XMentionStore {
|
||||
getOrCreateSource(input: {
|
||||
companyId: string;
|
||||
sourceKey: string;
|
||||
accountUserId: string;
|
||||
accountHandle?: string | null;
|
||||
monthlyBudgetCents?: number;
|
||||
perRunBudgetCents?: number;
|
||||
}): Promise<XMentionSource>;
|
||||
updateSourceCursor(input: { sourceId: string; sinceId: string | null; now: Date }): Promise<void>;
|
||||
pauseSourceForBudget(input: { sourceId: string; reason: string; now: Date }): Promise<void>;
|
||||
markSourceRateLimited(input: { sourceId: string; resetAt: Date; now: Date }): Promise<void>;
|
||||
isAuthorAllowlisted(input: { companyId: string; xUserId: string }): Promise<boolean>;
|
||||
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<StoredXMention[]>;
|
||||
markHydrated(input: { mentionId: string; now: Date; data?: Record<string, unknown> }): Promise<void>;
|
||||
sumBudgetSince(input: { sourceId: string; since: Date }): Promise<number>;
|
||||
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<void>;
|
||||
}
|
||||
|
||||
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<XMentionOperation, "poll">[];
|
||||
}) {
|
||||
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<XMentionOperation, "poll">; 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<string, unknown> });
|
||||
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<number>`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<Record<XMentionOperation, number | null>>;
|
||||
}
|
||||
|
||||
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<Record<string, unknown>>;
|
||||
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<Record<XMentionOperation, number | null>>;
|
||||
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<string, unknown>;
|
||||
}
|
||||
|
||||
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;
|
||||
},
|
||||
};
|
||||
}
|
||||
Loading…
Reference in New Issue