mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-02 16:58:52 +01:00
fix(external-objects): cover budget resume and scoped deletions
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
50cd007005
commit
25a8a4392c
5 files changed
+79
-13
No files matched your search
@@ -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. |
|
||||
|
||||
@@ -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");
|
||||
|
||||
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | 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);
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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({
|
||||
|
||||
Reference in new issue
Block a user