From 25a8a4392c807d4b5eb19cbf713b5eda99c017d3 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 16:35:43 -0500 Subject: [PATCH] fix(external-objects): cover budget resume and scoped deletions Co-Authored-By: Paperclip --- .../2026-10-30-external-objects-sleep.md | 4 +- .../external-object-refresh-work.test.ts | 63 ++++++++++++++++++- .../external-objects-service.test.ts | 4 +- server/src/services/budgets.ts | 3 + server/src/services/external-objects.ts | 18 +++--- 5 files changed, 79 insertions(+), 13 deletions(-) diff --git a/doc/plans/2026-10-30-external-objects-sleep.md b/doc/plans/2026-10-30-external-objects-sleep.md index 639bbaea343..ab566999917 100644 --- a/doc/plans/2026-10-30-external-objects-sleep.md +++ b/doc/plans/2026-10-30-external-objects-sleep.md @@ -47,10 +47,10 @@ and fences automatic result persistence. It does not modify that draft PR. | Mutation | Owner and activation | | --- | --- | | Issue title/description, comment and document references | Existing sync methods register company intent before mention replacement and object upsert. Routes, runtime tools, adapters and plugin paths reuse these methods. | -| Comment/document disappearance | Sync registers reconstruction before deleting obsolete mentions. | +| Comment/document disappearance | Sync derives the retained mention owners and registers only those companies before deleting obsolete mentions. | | Last mention removal and reattachment | Mention replacement invalidates company deadlines. Retained objects stay available for reattachment. | | Issue deletion | Issue service registers its company before cascading mention deletion. | -| Company status/archive/deletion | Company services register company reconstruction before their status or deletion writes. | +| Company status/archive/deletion, including budget pause/resume | Company and budget services register company reconstruction before status or deletion writes. | | Feature enable/disable | Instance settings registers global reconstruction before the experimental-settings write. | | Plugin manifest/status/uninstall | Plugin registry registers reconstruction before lifecycle writes. Installed plugins acquire support through the ordinary ready transition. | | Plugin company configuration and enabled setting | Registry writes register only the affected company. There is no plugin-readiness poll. | diff --git a/server/src/__tests__/external-object-refresh-work.test.ts b/server/src/__tests__/external-object-refresh-work.test.ts index 8d1f5cdc00c..f429c31396b 100644 --- a/server/src/__tests__/external-object-refresh-work.test.ts +++ b/server/src/__tests__/external-object-refresh-work.test.ts @@ -2,10 +2,12 @@ import { randomUUID } from "node:crypto"; import { setTimeout as delay } from "node:timers/promises"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { eq, sql } from "drizzle-orm"; -import { activityLog, companies, createDb, externalObjects, issues, plugins } from "@paperclipai/db"; +import { activityLog, approvals, applyPendingMigrations, budgetIncidents, budgetPolicies, companies, costEvents, createDb, externalObjects, issues, issueComments, plugins } from "@paperclipai/db"; import { externalObjectService, type ExternalObjectResolveResult } from "../services/external-objects.js"; import type { PaperclipPluginManifestV1 } from "@paperclipai/shared"; import type { PluginWorkerManager } from "../services/plugin-worker-manager.js"; +import { documentService } from "../services/documents.js"; +import { budgetService } from "../services/budgets.js"; import { pluginRegistryService } from "../services/plugin-registry.js"; import { companyService } from "../services/companies.js"; import { instanceSettingsService } from "../services/instance-settings.js"; @@ -34,14 +36,19 @@ const active: ExternalObjectResolveResult = { ok: true, snapshot: { const errors = vi.fn(); const companyIds: string[] = []; beforeAll(async () => { + if (url) await applyPendingMigrations(url); if (!url) temp = await startEmbeddedPostgresTestDatabase("external-refresh-work-"); db = createDb(url ?? temp!.connectionString, { maxConnections: 6 }); await db.execute(sql`truncate table ${companies} cascade`); - }); + }, 90_000); afterEach(async () => { await coordinator?.stop(); coordinator = undefined; vi.useRealTimers(); vi.restoreAllMocks(); for (const id of companyIds.splice(0)) { + await db.delete(budgetIncidents).where(eq(budgetIncidents.companyId, id)); + await db.delete(budgetPolicies).where(eq(budgetPolicies.companyId, id)); + await db.delete(approvals).where(eq(approvals.companyId, id)); + await db.delete(costEvents).where(eq(costEvents.companyId, id)); await db.delete(activityLog).where(eq(activityLog.companyId, id)); await db.delete(issues).where(eq(issues.companyId, id)); await db.delete(externalObjects).where(eq(externalObjects.companyId, id)); @@ -256,6 +263,58 @@ const active: ExternalObjectResolveResult = { ok: true, snapshot: { expect(calls).toHaveBeenCalledOnce(); }); + it("reactivates refresh after ordinary budget pause and budget raise", async () => { + clock(); const svc = service(); provider.mockResolvedValue(active); await start(svc); + const f = await fixture(); await svc.syncIssue(f.issueId); + await settle(async () => (await row(f.companyId))?.lastResolvedAt !== null); + await db.insert(costEvents).values({ companyId: f.companyId, usageKind: "decision", provider: "fixture", + biller: "fixture", billingType: "metered_api", model: "simulated", costCents: 200, occurredAt: new Date() }); + const budgets = budgetService(db); + await budgets.upsertPolicy(f.companyId, { scopeType: "company", scopeId: f.companyId, + amount: 100, hardStopEnabled: true }, null); + expect((await db.select().from(companies).where(eq(companies.id, f.companyId)))[0]!.status).toBe("paused"); + await quiet(); + const reads = vi.spyOn(db, "select"); provider.mockClear(); + await vi.advanceTimersByTimeAsync(900_000); expect(reads).not.toHaveBeenCalled(); expect(provider).not.toHaveBeenCalled(); + provider.mockResolvedValue(terminal); + await budgets.upsertPolicy(f.companyId, { scopeType: "company", scopeId: f.companyId, amount: 300 }, null); + await settle(async () => (await row(f.companyId)).isTerminal); await quiet(); + expect(provider).toHaveBeenCalledOnce(); + }); + + it.each(["comment", "document"] as const)("keeps hard-deleted %s activation scoped to its retained mention owner", async kind => { + clock(); const svc = service(); provider.mockResolvedValue(active); await start(svc); + const a = await fixture(); const b = await fixture(); await svc.syncIssue(b.issueId); + let id: string; + if (kind === "comment") { + const [comment] = await db.insert(issueComments).values({ companyId: a.companyId, issueId: a.issueId, + body: `https://example.com/${a.companyId}` }).returning(); + id = comment!.id; await svc.syncComment(id); + } else { + const result = await documentService(db).upsertIssueDocument({ issueId: a.issueId, + key: "notes", format: "markdown", body: `https://example.com/${a.companyId}` }); + id = result.document.id; await svc.syncDocument(id); + } + await settle(async () => (await row(a.companyId))?.lastResolvedAt !== null && (await row(b.companyId))?.lastResolvedAt !== null); + await settle(() => coordinator!.nextWakeAt() !== null); + const reads: Array<{ query: string; params: unknown[] }> = []; + const debug = db.$client.options.debug; + db.$client.options.debug = (_connection, query, params) => { + if (/^select/i.test(query) && query.includes('from "external_objects"') && query.includes('inner join "companies"')) + reads.push({ query, params: [...params] }); + }; + try { + if (kind === "comment") { await db.delete(issueComments).where(eq(issueComments.id, id)); await svc.syncComment(id); } + else { await documentService(db).deleteIssueDocument(a.issueId, "notes"); await svc.syncDocument(id); } + await settle(() => reads.length > 0); + for (const read of reads) { + expect(read.query).toContain('"external_objects"."company_id" ='); + expect(read.params).toContain(a.companyId); expect(read.params).not.toContain(b.companyId); + } + } finally { db.$client.options.debug = debug; } + expect(provider).toHaveBeenCalledTimes(2); + }); + it("keeps warm standby startup SQL-free and resumes a first committed producer", async () => { clock(); admitted = false; reconcile = false; const svc = service(); const selects = vi.spyOn(db, "select"); diff --git a/server/src/__tests__/external-objects-service.test.ts b/server/src/__tests__/external-objects-service.test.ts index e50e298b466..bc268d0ba6a 100644 --- a/server/src/__tests__/external-objects-service.test.ts +++ b/server/src/__tests__/external-objects-service.test.ts @@ -3,6 +3,7 @@ import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest import { eq } from "drizzle-orm"; import { companies, + applyPendingMigrations, createDb, activityLog, externalObjectMentions, @@ -711,9 +712,10 @@ describeEmbeddedPostgres("externalObjectService", () => { let tempDb: Awaited> | null = null; beforeAll(async () => { + if (externalDatabaseUrl) await applyPendingMigrations(externalDatabaseUrl); if (!externalDatabaseUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-external-objects-"); db = createDb(externalDatabaseUrl ?? tempDb!.connectionString, { maxConnections: 6 }); - }, 20_000); + }, 90_000); afterEach(async () => { await db.delete(activityLog); diff --git a/server/src/services/budgets.ts b/server/src/services/budgets.ts index 6777124ee90..88c3255da4e 100644 --- a/server/src/services/budgets.ts +++ b/server/src/services/budgets.ts @@ -1,3 +1,4 @@ +import { notifyExternalObjectRefresh } from "./external-object-refresh-notifications.js"; import { notifyApprovalWork } from "./approval-work-notifications.js"; import { notifyCompanyRetentionWork } from "./retention-work-notifications.js"; import { addCents, compareCents, normalizeCents } from "@paperclipai/shared"; @@ -301,6 +302,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica const scope = await resolveScopeRecord(db, "company", policy.scopeId); if (scope.paused) return; await notifyCompanyRetentionWork(db, policy.companyId); + await notifyExternalObjectRefresh(db, policy.companyId); await notifyDeliveryWork(db, DELIVERY_QUEUES.agentTimers); await notifyDeliveryWork(db, DELIVERY_QUEUES.routineSchedules); await notifyDeliveryWork(db, DELIVERY_QUEUES.issueMonitors); @@ -361,6 +363,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica const scope = await resolveScopeRecord(db, "company", policy.scopeId); if (scope.pauseReason !== "budget") return; await notifyCompanyRetentionWork(db, policy.companyId); + await notifyExternalObjectRefresh(db, policy.companyId); await notifyDeliveryWork(db, DELIVERY_QUEUES.agentTimers); await notifyDeliveryWork(db, DELIVERY_QUEUES.routineSchedules); await notifyDeliveryWork(db, DELIVERY_QUEUES.issueMonitors); diff --git a/server/src/services/external-objects.ts b/server/src/services/external-objects.ts index c49df7ff076..d524b473e1a 100644 --- a/server/src/services/external-objects.ts +++ b/server/src/services/external-objects.ts @@ -571,6 +571,14 @@ export function externalObjectService( return dbOrTx === db ? db.transaction(runSync) : runSync(dbOrTx); } + async function deleteMissingSourceMentions(kind: "comment" | "document", recordId: string, tx: any) { + const source = and(eq(externalObjectMentions.sourceKind, kind), eq(externalObjectMentions.sourceRecordId, recordId)); + const owners: Array<{ companyId: string }> = await tx.selectDistinct({ companyId: externalObjectMentions.companyId }) + .from(externalObjectMentions).where(source); + for (const owner of owners) await notifyExternalObjectRefresh(tx, owner.companyId); + await tx.delete(externalObjectMentions).where(source); + } + async function syncComment(commentId: string, dbOrTx: any = db) { const runSync = async (tx: any) => { if (!(await isEnabled())) return; @@ -585,10 +593,7 @@ export function externalObjectService( .where(eq(issueComments.id, commentId)) .then((rows: Array<{ id: string; companyId: string; issueId: string; body: string }>) => rows[0] ?? null); if (!comment) { - await notifyExternalObjectRefresh(tx); - await tx - .delete(externalObjectMentions) - .where(and(eq(externalObjectMentions.sourceKind, "comment"), eq(externalObjectMentions.sourceRecordId, commentId))); + await deleteMissingSourceMentions("comment", commentId, tx); return; } await replaceSourceMentions({ @@ -620,10 +625,7 @@ export function externalObjectService( .where(eq(documents.id, documentId)) .then((rows: Array<{ documentId: string; companyId: string; issueId: string; key: string; body: string }>) => rows[0] ?? null); if (!document) { - await notifyExternalObjectRefresh(tx); - await tx - .delete(externalObjectMentions) - .where(and(eq(externalObjectMentions.sourceKind, "document"), eq(externalObjectMentions.sourceRecordId, documentId))); + await deleteMissingSourceMentions("document", documentId, tx); return; } await replaceSourceMentions({