perf(external-objects): schedule scoped refresh obligations

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-10-30 16:24:47 -05:00
1 parent 198218bf55
commit 50cd007005
14 files changed
+659 -100

No files matched your search

@@ -0,0 +1,92 @@
# External object refresh deadlines
External-object refresh uses the delivery coordinator. The server registers the
service in both heartbeat-scheduler configurations. It no longer calls the
company-wide aggregate sweep on each heartbeat tick.
`external-objects.ts` owns automatic eligibility, resolver selection, refresh
claims, provider results and the next deadline. `index.ts` supplies registration
and admission gates. The existing scoped deadline helper owns wake coalescing,
page reconciliation, fair company selection, retry backoff and shutdown.
## Durable obligations and quiet states
An automatic refresh needs an active company, a retained mention in that company,
a non-null `next_refresh_at` and an available resolver. Injected resolvers,
including custom `url/link` resolvers, remain supported. A plugin resolver needs a
ready plugin with `external.objects.read`, a matching provider/object declaration,
and no company setting that disables it. Built-in resolvers remain available
independently of optional plugin support.
Terminal objects are passive. The existing exception remains: GitHub
`not_found` snapshots without `data.probeVerified=true` need a repository-visibility
probe before their terminal verdict is definitive. A null deadline retains its
existing never-refresh meaning. Registration does not change stored timestamps.
The worker reconstructs eligible deadlines in pages of at most 250 rows. A
company notification reads that company only. A due callback reads only its
object. Pages select scheduling columns and retain compact deadlines in memory.
Dispatch uses fair batches of at most 50 objects. The deadline is the later of
`next_refresh_at` and the current refresh lease expiry. Lease renewals and atomic
refresh tokens continue to exclude another claimant.
Unsupported generic links, orphan objects, disabled plugins, empty configuration
and retained terminal history do not schedule fallback discovery. Manual refresh
retains its existing `no_resolver` result. Actual provider failures retain their
saved retry deadline. Unexpected failures retain scoped coordinator backoff;
restart reconstructs the saved object and any unexpired claim.
The supported-resolver and retained-reference approach follows the related draft
[PR #17298](https://github.com/paperclipai/paperclip/pull/17298) by
[@kaluli123123](https://github.com/kaluli123123). This change adapts that eligibility
pattern to scoped scheduling, adds dynamic plugin/company permission predicates,
and fences automatic result persistence. It does not modify that draft PR.
## Producer and transaction audit
| 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. |
| 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. |
| 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. |
| Manual and automatic resolver result | Result persistence registers object intent in the same transaction. Automatic persistence rechecks reference and resolver eligibility. Lost support releases the worker's own claim without publishing the stale result. |
| Normalized GitHub ingress | The event service registers company intent with the snapshot write. It clears the old refresh token, so an older provider response cannot overwrite that event. |
| Import and restore | No independent object-table importer exists in the source baseline. Supported reference ingestion uses the ordinary sync boundary. Startup reconstructs restored durable objects. |
Notifications are awaited before writes. Caller-owned and nested transactions
retain the outer commit boundary through the existing database-work protocol.
Rollback does not publish the write. Notifications received during a scan or an
active resolver remain dirty until a successful reconciliation.
The supported live writer model is one shared `createDb` signal owner, including
its transactions and child pools. These process-local signals do not cover
independent SQL clients or other replicas. Such writers must invoke the supported
service boundary or arrange explicit reconstruction/restart. Existing normalized
GitHub ingress uses the supported boundary. No permanent database listener or
periodic global safety scan is added.
## Acceptance evidence
The focused suite uses real PostgreSQL with a controlled simulated provider.
Ordinary sync and registry/settings writes create work; the fixture does not
pre-create the tested resolved outcome. Assertions independently read saved
objects. Tests measure owner reads and provider calls across former tick windows.
Coverage includes empty and terminal quiet, unsupported links, orphan removal and
reattachment, feature/company eligibility, plugin readiness and permission loss
during a held provider call, exact and earlier deadlines, durable provider retry,
rollback and outer commit after nested savepoint release, startup and paginated
reconstruction, inactive companies, never-refresh rows, first work during drain,
warm standby, failed refresh after restart, lease renewal, duplicate exclusion,
and shutdown while a provider call remains active. The GitHub ingress suite also
checks a real normalized event superseding a held resolver response.
The conservative external-object host-sleep blocker remains. An in-process
deadline cannot wake stopped compute. This change proves worker quiet and
reactivation in the tested process; it does not establish Cloud alarm coverage,
host/database suspension or deployed savings.
@@ -0,0 +1,334 @@
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 { externalObjectService, type ExternalObjectResolveResult } from "../services/external-objects.js";
import type { PaperclipPluginManifestV1 } from "@paperclipai/shared";
import type { PluginWorkerManager } from "../services/plugin-worker-manager.js";
import { pluginRegistryService } from "../services/plugin-registry.js";
import { companyService } from "../services/companies.js";
import { instanceSettingsService } from "../services/instance-settings.js";
import { notifyExternalObjectRefresh } from "../services/external-object-refresh-notifications.js";
import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
const url = resolveExternalTestDatabaseUrl("PAPERCLIP_EXTERNAL_OBJECT_REFRESH_TEST_DATABASE_URL");
const support = url ? { supported: true } : await getEmbeddedPostgresTestSupport();
const terminal: ExternalObjectResolveResult = { ok: true, snapshot: {
statusKey: "done", statusCategory: "succeeded", statusTone: "success", isTerminal: true,
} };
const active: ExternalObjectResolveResult = { ok: true, snapshot: {
statusKey: "open", statusCategory: "open", statusTone: "neutral", isTerminal: false, ttlSeconds: 60,
} };
(support.supported ? describe : describe.skip)("external object refresh work (real PostgreSQL, simulated provider)", () => {
let db: ReturnType<typeof createDb>;
let temp: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
let admitted = true;
let reconcile = true;
let enabled = true;
const provider = vi.fn(async (): Promise<ExternalObjectResolveResult> => terminal);
const errors = vi.fn();
const companyIds: string[] = [];
beforeAll(async () => {
if (!url) temp = await startEmbeddedPostgresTestDatabase("external-refresh-work-");
db = createDb(url ?? temp!.connectionString, { maxConnections: 6 });
await db.execute(sql`truncate table ${companies} cascade`);
});
afterEach(async () => {
await coordinator?.stop(); coordinator = undefined;
vi.useRealTimers(); vi.restoreAllMocks();
for (const id of companyIds.splice(0)) {
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));
await db.delete(companies).where(eq(companies.id, id));
}
await db.delete(plugins);
admitted = true; reconcile = true; enabled = true; errors.mockClear();
provider.mockReset().mockResolvedValue(terminal);
});
afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temp?.cleanup(); });
function service() {
return externalObjectService(db, { github: false, enabled: () => enabled,
resolvers: [{ providerKey: "url", resolve: provider }] });
}
async function start(svc = service()) {
coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => admitted,
canReconcile: () => reconcile, onError: errors });
const worker = svc.registerRefreshWork(coordinator, () => admitted);
await worker.ready;
return worker;
}
async function fixture(status = "active") {
const id = randomUUID(); companyIds.push(id);
await db.insert(companies).values({ id, name: "Refresh test", status,
issuePrefix: `E${id.replaceAll("-", "").slice(0, 8)}` });
const [issue] = await db.insert(issues).values({ companyId: id, title: "Reference",
description: `https://example.com/${id}`, status: "todo" }).returning();
return { companyId: id, issueId: issue!.id };
}
async function row(companyId: string) {
return (await db.select().from(externalObjects).where(eq(externalObjects.companyId, companyId)))[0]!;
}
async function settle(predicate: () => Promise<boolean> | boolean) {
for (let i = 0; i < 200; i++) {
await vi.advanceTimersByTimeAsync(0); await delay(5);
if (await predicate()) return;
}
throw new Error("refresh worker did not settle");
}
async function quiet() { await settle(() => coordinator!.nextWakeAt() === null); }
function clock() { vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true }); }
async function change(companyId: string, patch: Partial<typeof externalObjects.$inferInsert>) {
await db.transaction(async tx => {
await notifyExternalObjectRefresh(tx, companyId);
await tx.update(externalObjects).set(patch).where(eq(externalObjects.companyId, companyId));
});
}
it("captures the old configured-empty aggregate SQL cost", async () => {
const svc = service(); await fixture();
const selects = vi.spyOn(db, "select");
for (let i = 0; i < 3; i++) await svc.refreshDueObjectsForActiveCompanies();
expect(selects).toHaveBeenCalledTimes(6);
expect(provider).not.toHaveBeenCalled();
});
it("activates saved work when the ordinary feature and company settings enable it", async () => {
clock(); const settings = instanceSettingsService(db);
await settings.updateExperimental({ enableExternalObjects: true });
const svc = externalObjectService(db, { github: false,
enabled: async () => (await settings.getExperimental()).enableExternalObjects,
resolvers: [{ providerKey: "url", resolve: provider }] });
const f = await fixture("paused"); await svc.syncIssue(f.issueId);
await settings.updateExperimental({ enableExternalObjects: false });
await start(svc); await quiet();
await companyService(db).update(f.companyId, { status: "active" }); await quiet();
expect(provider).not.toHaveBeenCalled();
await settings.updateExperimental({ enableExternalObjects: true });
await settle(async () => (await row(f.companyId)).isTerminal); await quiet();
expect(provider).toHaveBeenCalledOnce();
});
it("settles empty, configured-idle and retained terminal history without recurring SQL or provider calls", async () => {
clock();
const svc = service(); await start(svc); await quiet();
const selects = vi.spyOn(db, "select");
await vi.advanceTimersByTimeAsync(15 * 60_000);
expect(selects).not.toHaveBeenCalled(); expect(provider).not.toHaveBeenCalled();
const f = await fixture(); await svc.syncIssue(f.issueId);
await settle(async () => (await row(f.companyId))?.isTerminal === true); await quiet();
expect((await row(f.companyId)).statusKey).toBe("done");
selects.mockClear(); provider.mockClear();
await vi.advanceTimersByTimeAsync(15 * 60_000);
expect(selects).not.toHaveBeenCalled(); expect(provider).not.toHaveBeenCalled();
expect(errors).not.toHaveBeenCalled();
});
it("registers exact future deadlines, preempts them on earlier writes and cancels terminal work", async () => {
clock(); provider.mockResolvedValue(active);
const svc = service(); await start(svc);
const f = await fixture(); await svc.syncIssue(f.issueId);
await settle(() => provider.mock.calls.length === 1);
await settle(async () => (await row(f.companyId)).lastResolvedAt !== null);
const saved = await row(f.companyId);
await settle(() => coordinator!.nextWakeAt() === saved.nextRefreshAt!.getTime());
const earlier = new Date(Date.now() + 1000); await change(f.companyId, { nextRefreshAt: earlier });
await settle(() => coordinator!.nextWakeAt() === earlier.getTime());
await vi.advanceTimersByTimeAsync(999); expect(provider).toHaveBeenCalledTimes(1);
provider.mockResolvedValue(terminal); await vi.advanceTimersByTimeAsync(1);
await settle(async () => (await row(f.companyId)).isTerminal); await quiet();
await vi.advanceTimersByTimeAsync(120_000); expect(provider).toHaveBeenCalledTimes(2);
});
it("keeps disabled and paused company work passive and reconstructs on eligibility mutation", async () => {
clock(); enabled = false;
const svc = service(); await start(svc); await quiet();
const f = await fixture("paused");
enabled = true; await svc.syncIssue(f.issueId); await quiet();
await vi.advanceTimersByTimeAsync(600_000); expect(provider).not.toHaveBeenCalled();
await db.transaction(async tx => {
await notifyExternalObjectRefresh(tx, f.companyId);
await tx.update(companies).set({ status: "active" }).where(eq(companies.id, f.companyId));
});
await settle(async () => (await row(f.companyId)).isTerminal); await quiet();
expect(provider).toHaveBeenCalledOnce();
});
it("waits for the outer commit and leaves rolled-back ingestion quiet", async () => {
clock(); const svc = service(); await start(svc); await quiet();
const f = await fixture();
await expect(db.transaction(async tx => {
await svc.syncIssue(f.issueId, tx); throw new Error("rollback");
})).rejects.toThrow("rollback");
await quiet(); expect(await row(f.companyId)).toBeUndefined(); expect(provider).not.toHaveBeenCalled();
let release!: () => void; const held = new Promise<void>(resolve => { release = resolve; });
let nestedDone = false;
const commit = db.transaction(async tx => {
await tx.transaction(async nested => { await svc.syncIssue(f.issueId, nested); });
nestedDone = true; await held;
});
await settle(() => nestedDone); await vi.advanceTimersByTimeAsync(60_000);
expect(provider).not.toHaveBeenCalled(); release(); await commit;
await settle(async () => (await row(f.companyId))?.isTerminal); await quiet();
expect(provider).toHaveBeenCalledOnce();
});
it("retains first work during drain and recovers a provider failure after restart", async () => {
clock(); const svc = service(); await start(svc); await quiet();
admitted = false; const f = await fixture(); await svc.syncIssue(f.issueId);
const selects = vi.spyOn(db, "select");
await vi.advanceTimersByTimeAsync(90_000); expect(provider).not.toHaveBeenCalled();
selects.mockClear(); await vi.advanceTimersByTimeAsync(90_000); expect(selects).not.toHaveBeenCalled();
admitted = true; provider.mockRejectedValueOnce(new Error("simulated transport failure"));
await vi.advanceTimersByTimeAsync(30_000); await settle(() => errors.mock.calls.length === 1);
const claimed = await row(f.companyId); expect(claimed.refreshToken).not.toBeNull();
await coordinator!.stop(); coordinator = undefined; await start(svc);
const due = claimed.refreshStartedAt!.getTime() + 300_000;
expect(coordinator!.nextWakeAt()).toBe(due);
await vi.advanceTimersByTimeAsync(due - Date.now() - 1); expect(provider).toHaveBeenCalledTimes(1);
await vi.advanceTimersByTimeAsync(1);
await settle(async () => (await row(f.companyId)).isTerminal); await quiet();
expect(provider).toHaveBeenCalledTimes(2);
});
it("leaves unsupported URLs passive and cancels orphan work until ordinary reattachment", async () => {
clock(); const unsupported = externalObjectService(db, { github: false });
await start(unsupported); const f = await fixture(); await unsupported.syncIssue(f.issueId); await quiet();
const selects = vi.spyOn(db, "select");
await vi.advanceTimersByTimeAsync(900_000); expect(selects).not.toHaveBeenCalled();
expect((await row(f.companyId)).lastResolvedAt).toBeNull();
await coordinator!.stop(); coordinator = undefined;
provider.mockResolvedValue(active); const svc = service(); await start(svc);
await settle(async () => (await row(f.companyId)).lastResolvedAt !== null);
await db.update(issues).set({ description: null }).where(eq(issues.id, f.issueId));
await svc.syncIssue(f.issueId); await quiet();
selects.mockClear(); provider.mockClear();
await vi.advanceTimersByTimeAsync(900_000);
expect(selects).not.toHaveBeenCalled(); expect(provider).not.toHaveBeenCalled();
provider.mockResolvedValue(terminal);
await db.update(issues).set({ description: `https://example.com/${f.companyId}` }).where(eq(issues.id, f.issueId));
await svc.syncIssue(f.issueId);
await settle(async () => (await row(f.companyId)).isTerminal); await quiet();
expect(provider).toHaveBeenCalledOnce();
});
it("activates a newly ready plugin and fences a result after company permission is withdrawn", async () => {
clock(); const registry = pluginRegistryService(db);
const manifest: PaperclipPluginManifestV1 = {
id: `paperclip.refresh-${randomUUID()}`, apiVersion: 1, version: "1.0.0", displayName: "Refresh fixture",
description: "Simulated external object provider", author: "Paperclip", categories: ["connector"],
capabilities: ["external.objects.detect", "external.objects.read"], entrypoints: { worker: "worker.js" },
objectReferences: [{ providerKey: "tracker", displayName: "Tracker", objectTypes: ["ticket"], urlPatterns: ["https://example.com/:id"] }],
};
const plugin = await registry.install({ packageName: "refresh-fixture" }, manifest);
let finish!: (result: ExternalObjectResolveResult) => void;
const calls = vi.fn(async (_id: string, method: string) => {
if (method === "detectExternalObjects") return { detections: [] };
if (calls.mock.calls.filter(call => call[1] === "resolveExternalObject").length === 1)
return new Promise<ExternalObjectResolveResult>(resolve => { finish = resolve; });
return terminal;
});
const svc = externalObjectService(db, { github: false,
pluginWorkerManager: { call: calls } as unknown as PluginWorkerManager,
detectors: [{ key: "fixture", detect: ({ urls }) => urls.map(canonical => ({ canonical,
detectorKey: "fixture", providerKey: "tracker", objectType: "ticket", externalId: canonical.canonicalIdentityHash })) }],
});
await start(svc); const f = await fixture(); await svc.syncIssue(f.issueId); await quiet();
const selects = vi.spyOn(db, "select");
await vi.advanceTimersByTimeAsync(900_000); expect(selects).not.toHaveBeenCalled(); expect(calls).not.toHaveBeenCalled();
await registry.updateStatus(plugin!.id, { status: "ready" });
await settle(() => typeof finish === "function");
await registry.upsertCompanySettings(plugin!.id, f.companyId, { enabled: false, settingsJson: {} });
finish(terminal);
await settle(async () => (await row(f.companyId)).refreshToken === null); await quiet();
expect((await row(f.companyId)).isTerminal).toBe(false);
selects.mockClear(); calls.mockClear();
await vi.advanceTimersByTimeAsync(900_000); expect(selects).not.toHaveBeenCalled(); expect(calls).not.toHaveBeenCalled();
calls.mockImplementation(async () => terminal);
await registry.upsertCompanySettings(plugin!.id, f.companyId, { enabled: true, settingsJson: {} });
await settle(async () => (await row(f.companyId)).isTerminal); await quiet();
expect(calls).toHaveBeenCalledOnce();
});
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");
await start(svc); await vi.advanceTimersByTimeAsync(90_000);
expect(selects).not.toHaveBeenCalled();
const f = await fixture(); await svc.syncIssue(f.issueId);
selects.mockClear(); await vi.advanceTimersByTimeAsync(90_000);
expect(selects).not.toHaveBeenCalled(); expect(provider).not.toHaveBeenCalled();
admitted = true; reconcile = true;
await vi.advanceTimersByTimeAsync(30_000);
await settle(async () => (await row(f.companyId))?.isTerminal); await quiet();
expect(provider).toHaveBeenCalledOnce();
});
it("renews an active provider lease so another service cannot duplicate its action", async () => {
clock(); const svc = service(); await start(svc);
let finish!: (result: ExternalObjectResolveResult) => void;
provider.mockImplementationOnce(() => new Promise(resolve => { finish = resolve; }));
const f = await fixture(); await svc.syncIssue(f.issueId);
await settle(() => provider.mock.calls.length === 1);
const claimed = await row(f.companyId);
await vi.advanceTimersByTimeAsync(60_000);
await settle(async () => (await row(f.companyId)).refreshStartedAt!.getTime() > claimed.refreshStartedAt!.getTime());
const competing = await service().refreshObject(claimed.id, { companyId: f.companyId,
force: true, now: new Date(claimed.refreshStartedAt!.getTime() + 301_000) });
expect(competing.reason).toBe("refresh_in_progress"); expect(provider).toHaveBeenCalledOnce();
finish(terminal); await settle(async () => (await row(f.companyId)).isTerminal); await quiet();
});
it("uses durable provider retry deadlines and waits for in-flight shutdown without overlap", async () => {
clock(); const svc = service(); await start(svc);
provider.mockResolvedValueOnce({ ok: false, liveness: "rate_limited", errorCode: "slow_down", retryAfterSeconds: 5 });
const f = await fixture(); await svc.syncIssue(f.issueId);
await settle(async () => (await row(f.companyId))?.lastErrorCode === "slow_down");
const saved = await row(f.companyId);
await settle(() => coordinator!.nextWakeAt() === saved.nextRefreshAt!.getTime());
expect(saved.refreshToken).toBeNull();
await vi.advanceTimersByTimeAsync(saved.nextRefreshAt!.getTime() - Date.now() - 1);
expect(provider).toHaveBeenCalledOnce();
let finish!: (result: ExternalObjectResolveResult) => void;
provider.mockImplementationOnce(() => new Promise(resolve => { finish = resolve; }));
await vi.advanceTimersByTimeAsync(1); await settle(() => provider.mock.calls.length === 2);
await svc.syncIssue(f.issueId); await vi.advanceTimersByTimeAsync(0);
expect(provider).toHaveBeenCalledTimes(2);
let stopped = false; const stop = coordinator!.stop().then(() => { stopped = true; });
await delay(10); expect(stopped).toBe(false);
finish(terminal); await stop; coordinator = undefined;
const selects = vi.spyOn(db, "select");
await vi.advanceTimersByTimeAsync(600_000);
expect(selects).not.toHaveBeenCalled(); expect(provider).toHaveBeenCalledTimes(2);
});
it("reconstructs paginated multi-company work and never invokes inactive or never-refresh objects", async () => {
clock(); const svc = service(); const a = await fixture(); const b = await fixture(); const inactive = await fixture("paused");
for (const companyId of [b.companyId, inactive.companyId]) {
await db.insert(externalObjects).values(Array.from({ length: 255 }, (_, index) => ({
companyId, providerKey: "url", objectType: "link", externalId: String(index),
sanitizedCanonicalUrl: `https://example.com/${companyId}/${index}`, canonicalIdentityHash: `${companyId}-${index}`,
nextRefreshAt: null,
})));
}
await db.update(issues).set({ description: Array.from({ length: 255 }, (_, index) =>
`https://example.com/${a.companyId}/${index}`).join(" ") }).where(eq(issues.id, a.issueId));
await svc.syncIssue(a.issueId);
const future = new Date(Date.now() + 60_000);
await db.update(externalObjects).set({ nextRefreshAt: future }).where(eq(externalObjects.companyId, a.companyId));
await svc.syncIssue(b.issueId);
await start(svc); await settle(() => provider.mock.calls.length === 1);
await settle(() => coordinator!.nextWakeAt() === future.getTime());
expect((await row(a.companyId)).nextRefreshAt).toEqual(future);
const companiesSeen = provider.mock.calls.map(call => (call as unknown as [{ companyId: string }])[0].companyId);
expect(companiesSeen).toEqual([b.companyId]);
await change(a.companyId, { isTerminal: true }); await quiet();
await vi.advanceTimersByTimeAsync(600_000); expect(provider).toHaveBeenCalledOnce();
});
});
@@ -11,6 +11,7 @@ import {
issues,
plugins,
} from "@paperclipai/db";
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
import {
getEmbeddedPostgresTestSupport,
startEmbeddedPostgresTestDatabase,
@@ -27,7 +28,8 @@ import type { PluginWorkerManager } from "../services/plugin-worker-manager.js";
import { createGitHubExternalObjectProvider } from "../services/github-external-object-provider.js";
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
const externalDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_EXTERNAL_OBJECTS_TEST_DATABASE_URL");
const describeEmbeddedPostgres = externalDatabaseUrl || embeddedPostgresSupport.supported ? describe : describe.skip;
if (!embeddedPostgresSupport.supported) {
console.warn(
@@ -709,8 +711,8 @@ describeEmbeddedPostgres("externalObjectService", () => {
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
beforeAll(async () => {
tempDb = await startEmbeddedPostgresTestDatabase("paperclip-external-objects-");
db = createDb(tempDb.connectionString);
if (!externalDatabaseUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-external-objects-");
db = createDb(externalDatabaseUrl ?? tempDb!.connectionString, { maxConnections: 6 });
}, 20_000);
afterEach(async () => {
@@ -725,6 +727,7 @@ describeEmbeddedPostgres("externalObjectService", () => {
});
afterAll(async () => {
await db?.$client.end({ timeout: 1 });
await tempDb?.cleanup();
});
@@ -1,3 +1,4 @@
import { externalObjectService } from "../services/external-objects.js";
import { startToolHealthWorker } from "./helpers/tool-health-worker.js";
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
import { randomUUID } from "node:crypto";
@@ -35,7 +36,7 @@ describeEmbeddedPostgres("GitHub connection event delivery", () => {
beforeAll(async () => {
if (!externalTestDatabaseUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-github-events-");
if (externalTestDatabaseUrl) await applyPendingMigrations(externalTestDatabaseUrl);
db = createDb(externalTestDatabaseUrl ?? tempDb!.connectionString);
db = createDb(externalTestDatabaseUrl ?? tempDb!.connectionString, { maxConnections: 6 });
}, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS);
afterEach(async () => {
@@ -171,12 +172,27 @@ describeEmbeddedPostgres("GitHub connection event delivery", () => {
expect(connector.leaseEvents).toHaveBeenCalledTimes(5);
}
let releaseRefresh!: () => void;
let markStarted!: () => void;
const started = new Promise<void>(resolve => { markStarted = resolve; });
const held = new Promise<void>(resolve => { releaseRefresh = resolve; });
const resolver = externalObjectService(db, { github: false, resolvers: [{
providerKey: "github", resolve: async () => {
markStarted(); await held;
return { ok: true, snapshot: { statusKey: "open", statusCategory: "open", statusTone: "info", isTerminal: false } };
},
}] });
const oldRefresh = resolver.refreshObject(externalObjectId, { companyId, force: true });
await started;
const warning = vi.spyOn(logger, "warn").mockImplementation(() => undefined);
await expect(service.pollOnce({ forSetup: activeSetup })).resolves.toMatchObject({ leased: 2, processed: 1, duplicate: 0, failed: 1 });
expect(warning.mock.calls).toEqual([[{
deliveryId: "undecryptable-delivery", reason: "inner_event_authentication_failed",
}, "GitHub connector event retained for retry"]]);
warning.mockRestore();
releaseRefresh();
await expect(oldRefresh).resolves.toMatchObject({ refreshed: false, reason: "refresh_superseded" });
const [snapshot] = await db.select().from(externalObjects).where(eq(externalObjects.id, externalObjectId));
expect(snapshot).toMatchObject({
statusKey: "merged",
@@ -184,6 +200,8 @@ describeEmbeddedPostgres("GitHub connection event delivery", () => {
statusCategory: "succeeded",
statusTone: "success",
isTerminal: true,
refreshToken: null,
refreshStartedAt: null,
remoteVersion: "2026-09-04T11:59:01.000Z",
data: expect.objectContaining({ marker: "preserved", merged: true, headRef: "feature", baseSha: "b".repeat(40) }),
});
+2 -15
View File
@@ -1323,19 +1323,6 @@ async function startServerWithDatabaseTeardown(
pluginWorkerManager,
enabled: async () => (await instanceSettingsService(db).getExperimental()).enableExternalObjects === true,
});
const scheduleExternalObjectRefreshSweep = (now = new Date()) => {
if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return;
trackHeartbeatSchedulerWork(externalObjects
.refreshDueObjectsForActiveCompanies(50, now)
.then((result) => {
if (result.checked > 0 || result.refreshed > 0) {
logger.info({ ...result }, "external-object scheduler tick refreshed due objects");
}
})
.catch((err) => {
logger.error({ err }, "external-object scheduler tick failed");
}));
};
// The retry backstop for orphan sandboxes. An acquire that rejects a
// foreign-company insert tears the provisioned sandbox down. If that teardown
@@ -1362,6 +1349,8 @@ async function startServerWithDatabaseTeardown(
resolveNativeQuestion: (interaction) => deliverNativeQuestionResponse(db as any, interaction),
});
const deliveryWork = app.locals.deliveryWork as ReturnType<typeof createDeliveryWorkCoordinator>;
externalObjects.registerRefreshWork(deliveryWork,
() => !heartbeatSchedulerStopped && !isIdleTaskDrainActive() && !isWarmStandby());
if (config.nativeHarnessBackupExpirationEnabled) {
nativeHarnessBackupExpiration.registerCheckpointExpiration(deliveryWork);
}
@@ -2007,7 +1996,6 @@ async function startServerWithDatabaseTeardown(
}
if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return;
scheduleExternalObjectRefreshSweep(new Date());
if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return;
scheduleMergedPullRequestConfirmationSweep();
@@ -2133,7 +2121,6 @@ async function startServerWithDatabaseTeardown(
// as the enabled path.
await startEnvironmentCleanup();
startHeartbeatSchedulerInterval(() => {
scheduleExternalObjectRefreshSweep(new Date());
scheduleGitHubConnectionEventPoll();
scheduleGitHubConnectionContinuitySweep();
});
+3
View File
@@ -1,5 +1,6 @@
import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js";
import { notifyCompanyRetentionWork } from "./retention-work-notifications.js";
import { notifyExternalObjectRefresh } from "./external-object-refresh-notifications.js";
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
import { isRunnerApiResponseAsset } from "./asset-content-access.js";
import { deleteCompany } from "./company-deletion.js";
@@ -334,6 +335,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) {
if (companyPatch.status !== undefined && companyPatch.status !== existing.status) {
await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync);
await notifyExternalObjectRefresh(tx, id);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.routineSchedules);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
await notifyCompanyRetentionWork(tx, id);
@@ -490,6 +492,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) {
if (!wasAlreadyArchived) {
await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync);
await notifyExternalObjectRefresh(tx, id);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.routineSchedules);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
await notifyCompanyRetentionWork(tx, id);
+2
View File
@@ -1,3 +1,4 @@
import { notifyExternalObjectRefresh } from "./external-object-refresh-notifications.js";
import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js";
import { notifyCancelledWorkspacePreservation } from "./cancelled-workspace-preservation-notifications.js";
import { notifyInstructionCollection } from "./instruction-collection-notifications.js";
@@ -71,6 +72,7 @@ export async function deleteCompany(db: Db, id: string) {
const [existing] = await tx.select({ id: companies.id }).from(companies)
.where(eq(companies.id, id)).for("no key update");
if (!existing) return null;
await notifyExternalObjectRefresh(tx, id);
await notifyChatCompanyConfiguration(tx, id);
await chatCompanyDeletion.deleteCompanyData(tx, id);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync);
@@ -6,6 +6,7 @@ export const DELIVERY_QUEUES = {
providerTraceRetention: "provider-trace-retention",
nativeRunnerEvidenceRetention: "native-runner-evidence-retention",
skillVersionRetention: "skill-version-retention",
externalObjectRefresh: "external-object-refresh",
databaseBackups: "database-backups",
importSpoolCleanup: "import-spool-cleanup",
externalConversationSettlement: "external-conversation-settlement",
@@ -0,0 +1,7 @@
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
export async function notifyExternalObjectRefresh(tx: object, companyId?: string, objectId?: string) {
await notifyDeliveryWork(tx, DELIVERY_QUEUES.externalObjectRefresh,
companyId ? JSON.stringify({ companyId, ...(objectId ? { objectId } : {}) }) : "*");
}
+137 -46
View File
@@ -1,7 +1,7 @@
import { randomUUID } from "node:crypto";
import { and, asc, eq, inArray, isNull, lte, not, or, sql } from "drizzle-orm";
import { and, asc, eq, exists, gt, inArray, isNotNull, isNull, lte, not, or, sql } from "drizzle-orm";
import type { Db } from "@paperclipai/db";
import { companies, documents, externalObjectMentions, externalObjects, issueComments, issueDocuments, issues, plugins } from "@paperclipai/db";
import { companies, documents, externalObjectMentions, externalObjects, issueComments, issueDocuments, issues, plugins, pluginCompanySettings } from "@paperclipai/db";
import {
formatExternalObjectMentionSourceLabel,
type ExternalObjectCanonicalUrl,
@@ -18,6 +18,10 @@ import { notFound } from "../errors.js";
import { logger } from "../middleware/logger.js";
import { logActivity, type LogActivityInput } from "./activity-log.js";
import { createGitHubExternalObjectProvider, type GitHubExternalObjectProviderOptions } from "./github-external-object-provider.js";
import { notifyExternalObjectRefresh } from "./external-object-refresh-notifications.js";
import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
import { registerScopedDeadlineWork } from "./scoped-deadline-work.js";
import { publishLiveEvent } from "./live-events.js";
import type { PluginWorkerManager } from "./plugin-worker-manager.js";
@@ -95,6 +99,18 @@ const DEFAULT_RETRY_AFTER_SECONDS = 300;
const DEFAULT_REFRESH_LEASE_SECONDS = 300;
const REFRESH_LEASE_RENEW_INTERVAL_MS = 60_000;
type RefreshScope = { companyId: string; objectId?: string };
type RefreshDeadline = { id: string; companyId: string; dueAt: number };
function refreshableObjectWhere() {
return or(eq(externalObjects.isTerminal, false), and(
eq(externalObjects.providerKey, "github"),
eq(externalObjects.statusKey, "not_found"),
not(sql`coalesce(${externalObjects.data}->>'probeVerified', 'false') = 'true'`),
))!;
}
function sourceWhere(input: ExternalObjectSourceContext) {
const conditions = [
eq(externalObjectMentions.companyId, input.companyId),
@@ -245,7 +261,7 @@ function objectSnapshot(object: ExternalObjectRecord): PluginExternalObjectRecor
};
}
async function readyObjectReferencePlugins(db: Db) {
async function readyObjectReferencePlugins(db: Db, companyId?: string) {
return db
.select({
id: plugins.id,
@@ -253,7 +269,12 @@ async function readyObjectReferencePlugins(db: Db) {
manifestJson: plugins.manifestJson,
})
.from(plugins)
.where(eq(plugins.status, "ready"))
.where(and(eq(plugins.status, "ready"), companyId ? not(exists(db.select({ id: pluginCompanySettings.id })
.from(pluginCompanySettings).where(and(
eq(pluginCompanySettings.pluginId, plugins.id),
eq(pluginCompanySettings.companyId, companyId),
eq(pluginCompanySettings.enabled, false),
)))) : undefined))
.orderBy(asc(plugins.installOrder))
.then((rows) =>
rows.filter((row) => (row.manifestJson.objectReferences?.length ?? 0) > 0),
@@ -267,7 +288,7 @@ function createPluginProviderDetector(
return {
key: "plugin-object-reference-providers",
async detect(input) {
const providers = await readyObjectReferencePlugins(db);
const providers = await readyObjectReferencePlugins(db, input.companyId);
const detections: ExternalObjectDetection[] = [];
for (const provider of providers) {
@@ -326,7 +347,7 @@ async function resolveViaPluginProvider(
object: ExternalObjectRecord,
): Promise<PluginExternalObjectResolveResult | null> {
if (!pluginWorkerManager) return null;
const providers = await readyObjectReferencePlugins(db);
const providers = await readyObjectReferencePlugins(db, object.companyId);
for (const provider of providers) {
const manifest = provider.manifestJson;
if (!manifest.capabilities.includes("external.objects.read")) continue;
@@ -374,10 +395,38 @@ export function externalObjectService(
...(opts.detectors ?? []),
...(githubProvider ? [githubProvider.detector] : []),
]);
const resolverRegistry = createExternalObjectResolverRegistry([
const resolvers = [
...(opts.resolvers ?? []),
...(githubProvider?.resolvers ?? []),
]);
];
const resolverRegistry = createExternalObjectResolverRegistry(resolvers);
function automaticRefreshWhere(database: Db = db) {
const supported = resolvers.map(resolver => and(
eq(externalObjects.providerKey, resolver.providerKey),
resolver.objectType ? eq(externalObjects.objectType, resolver.objectType) : undefined,
)!);
if (opts.pluginWorkerManager) supported.push(exists(database.select({ id: plugins.id }).from(plugins).where(and(
eq(plugins.status, "ready"),
sql`${plugins.manifestJson}->'capabilities' ? 'external.objects.read'`,
sql`exists (select 1 from jsonb_array_elements(coalesce(${plugins.manifestJson}->'objectReferences', '[]'::jsonb)) as reference
where reference->>'providerKey' = ${externalObjects.providerKey}
and reference->'objectTypes' ? ${externalObjects.objectType})`,
not(exists(database.select({ id: pluginCompanySettings.id }).from(pluginCompanySettings).where(and(
eq(pluginCompanySettings.pluginId, plugins.id),
eq(pluginCompanySettings.companyId, externalObjects.companyId),
eq(pluginCompanySettings.enabled, false),
)))),
))));
return and(
refreshableObjectWhere(), isNotNull(externalObjects.nextRefreshAt),
supported.length ? or(...supported) : sql`false`,
exists(database.select({ id: externalObjectMentions.id }).from(externalObjectMentions).where(and(
eq(externalObjectMentions.companyId, externalObjects.companyId),
eq(externalObjectMentions.objectId, externalObjects.id),
))),
)!;
}
async function isEnabled() {
if (typeof opts.enabled === "function") return await opts.enabled();
@@ -422,6 +471,7 @@ export function externalObjectService(
nextRefreshAt: now,
updatedAt: now,
};
await notifyExternalObjectRefresh(dbOrTx, companyId);
const inserted = await dbOrTx
.insert(externalObjects)
.values(values)
@@ -451,6 +501,7 @@ export function externalObjectService(
dbOrTx: any = db,
) {
const urls = extractExternalObjectCanonicalUrls(input.text ?? "");
await notifyExternalObjectRefresh(dbOrTx, input.companyId);
await dbOrTx.delete(externalObjectMentions).where(sourceWhere(input));
if (urls.length === 0) return;
@@ -534,6 +585,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)));
@@ -568,6 +620,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)));
@@ -780,6 +833,7 @@ export function externalObjectService(
companyId: string;
actor?: Pick<LogActivityInput, "actorType" | "actorId" | "agentId" | "runId">;
force?: boolean;
scheduled?: boolean;
now?: Date;
};
@@ -801,6 +855,18 @@ export function externalObjectService(
return { object: toObjectPayload(latest, now), refreshed: false, reason: "refresh_superseded" as const };
}
async function persistRefresh(object: ExternalObjectRecord, token: string,
patch: Partial<typeof externalObjects.$inferInsert>, scheduled = false) {
return db.transaction(async tx => {
await notifyExternalObjectRefresh(tx, object.companyId, object.id);
const rows = await tx.update(externalObjects).set(patch)
.where(and(refreshOwnerWhere(object, token), scheduled ? automaticRefreshWhere(tx as unknown as Db) : undefined)).returning();
if (scheduled && rows.length === 0) await tx.update(externalObjects)
.set({ refreshStartedAt: null, refreshToken: null }).where(refreshOwnerWhere(object, token));
return rows;
});
}
async function resolveObjectRefresh(
object: ExternalObjectRecord,
input: RefreshObjectInput,
@@ -810,17 +876,13 @@ export function externalObjectService(
const pluginResult = await resolveViaPluginProvider(db, opts.pluginWorkerManager, object);
const resolver = pluginResult ? null : resolverRegistry.find(object);
if (!pluginResult && !resolver) {
const [updated] = await db
.update(externalObjects)
.set({
const [updated] = await persistRefresh(object, refreshToken, {
liveness: object.isTerminal ? object.liveness : visibleLiveness(object, now) === "fresh" ? "stale" : object.liveness,
nextRefreshAt: addSeconds(now, DEFAULT_RETRY_AFTER_SECONDS),
refreshStartedAt: null,
refreshToken: null,
updatedAt: now,
})
.where(refreshOwnerWhere(object, refreshToken))
.returning();
}, input.scheduled);
if (!updated) return refreshSupersededResult(object, now);
return { object: toObjectPayload(updated ?? object, now), refreshed: false, reason: "no_resolver" as const };
}
@@ -831,9 +893,7 @@ export function externalObjectService(
// terminal. Clear a previous terminal verdict so the object keeps being
// scheduled for automatic refresh and self-heals (e.g. a GitHub "not
// found" that later resolves once repo access improves).
const [updated] = await db
.update(externalObjects)
.set({
const [updated] = await persistRefresh(object, refreshToken, {
liveness: result.liveness,
isTerminal: false,
lastErrorAt: now,
@@ -843,9 +903,7 @@ export function externalObjectService(
refreshStartedAt: null,
refreshToken: null,
updatedAt: now,
})
.where(refreshOwnerWhere(object, refreshToken))
.returning();
}, input.scheduled);
if (!updated) return refreshSupersededResult(object, now);
publishLiveEvent({
companyId: object.companyId,
@@ -879,14 +937,10 @@ export function externalObjectService(
refreshToken: null,
updatedAt: now,
};
const [updated] = await db
.update(externalObjects)
.set({
const [updated] = await persistRefresh(object, refreshToken, {
...patch,
lastChangedAt: objectChanged(object, { ...object, ...patch }) ? now : object.lastChangedAt,
})
.where(refreshOwnerWhere(object, refreshToken))
.returning();
}, input.scheduled);
if (!updated) return refreshSupersededResult(object, now);
const next = updated ?? object;
if (objectChanged(object, next) && input.actor) {
@@ -947,6 +1001,8 @@ export function externalObjectService(
eq(externalObjects.id, object.id),
eq(externalObjects.companyId, object.companyId),
dueOrForced,
input.scheduled ? automaticRefreshWhere() : undefined,
input.scheduled ? sql`exists (select 1 from ${companies} where ${companies.id} = ${externalObjects.companyId} and ${companies.status} = 'active')` : undefined,
))
.returning();
return claimed ?? null;
@@ -1032,26 +1088,7 @@ export function externalObjectService(
async function refreshDueObjectsUnchecked(companyId: string, limit = 50, now = new Date()) {
const staleRefreshStartedBefore = new Date(now.getTime() - DEFAULT_REFRESH_LEASE_SECONDS * 1000);
// Terminal objects are usually skipped forever, but a GitHub "not found"
// verdict is only trustworthy while the repository is actually visible to
// the configured credentials. GitHub hides inaccessible (often private)
// repos behind a 404, so a not-found object can become resolvable later
// when credentials or repo visibility improve. Re-check objects that were
// frozen as "not found" before the repository probe existed so they
// self-heal instead of freezing. Once the provider has probed the repo and
// confirmed it is visible (data.probeVerified = "true"), the not-found
// verdict is definitive, so stop re-polling that object to avoid burning
// the company's GitHub API budget on deleted references forever.
const probeVerifiedNotFound = sql`coalesce(${externalObjects.data}->>'probeVerified', 'false') = 'true'`;
const refreshable = or(
eq(externalObjects.isTerminal, false),
and(
eq(externalObjects.isTerminal, true),
eq(externalObjects.providerKey, "github"),
eq(externalObjects.statusKey, "not_found"),
not(probeVerifiedNotFound),
),
)!;
const refreshable = refreshableObjectWhere();
const due = await db
.select({ id: externalObjects.id })
.from(externalObjects)
@@ -1099,6 +1136,59 @@ export function externalObjectService(
return { companies: activeCompanies.length, checked, refreshed };
}
function registerRefreshWork(
coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
canRun: () => boolean,
) {
const cursors = new Map<string, string>();
const scopeKey = (scope?: RefreshScope) => JSON.stringify(scope ?? null);
async function load(scope?: RefreshScope) {
if (!(await isEnabled())) { cursors.clear(); return []; }
const key = scopeKey(scope);
const cursor = cursors.get(key);
const rows = await db.select({ id: externalObjects.id, companyId: externalObjects.companyId,
nextRefreshAt: externalObjects.nextRefreshAt, refreshStartedAt: externalObjects.refreshStartedAt }).from(externalObjects)
.innerJoin(companies, eq(companies.id, externalObjects.companyId))
.where(and(
eq(companies.status, "active"), automaticRefreshWhere(),
scope ? eq(externalObjects.companyId, scope.companyId) : undefined,
scope?.objectId ? eq(externalObjects.id, scope.objectId) : undefined,
cursor ? gt(externalObjects.id, cursor) : undefined,
)).orderBy(asc(externalObjects.id)).limit(250);
const entries = rows.map(deadline);
if (rows.length === 250) cursors.set(key, rows[rows.length - 1]!.id);
else cursors.delete(key);
return { rows: entries, complete: rows.length < 250 };
}
function deadline(object: Pick<ExternalObjectRecord, "id" | "companyId" | "nextRefreshAt" | "refreshStartedAt">): RefreshDeadline {
return { id: object.id, companyId: object.companyId,
dueAt: Math.max(object.nextRefreshAt!.getTime(), object.refreshStartedAt
? object.refreshStartedAt.getTime() + DEFAULT_REFRESH_LEASE_SECONDS * 1000 : 0) };
}
return registerScopedDeadlineWork<RefreshDeadline, RefreshScope>(coordinator, {
queue: DELIVERY_QUEUES.externalObjectRefresh,
batchSize: 50,
minimumPassIntervalMs: 0,
decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint) as RefreshScope,
matches: (entry, scope) => !scope || (entry.companyId === scope.companyId
&& (!scope.objectId || entry.id === scope.objectId)),
load,
check: async (entry, signal) => {
if (signal?.aborted || !canRun()) return entry;
const current = await load({ companyId: entry.companyId, objectId: entry.id });
const scheduled = Array.isArray(current) ? current[0] : current.rows[0];
if (!scheduled || scheduled.dueAt > Date.now()) return scheduled ?? null;
await refreshObject(entry.id, {
companyId: entry.companyId,
scheduled: true,
actor: { actorType: "system", actorId: "external-object-resolver", agentId: null, runId: null },
});
const updated = await load({ companyId: entry.companyId, objectId: entry.id });
return (Array.isArray(updated) ? updated[0] : updated.rows[0]) ?? null;
},
}, canRun);
}
return {
syncIssue,
syncComment,
@@ -1114,5 +1204,6 @@ export function externalObjectService(
refreshIssueObjects,
refreshDueObjects,
refreshDueObjectsForActiveCompanies,
registerRefreshWork,
};
}
+25 -19
View File
@@ -1,3 +1,4 @@
import { notifyExternalObjectRefresh } from "./external-object-refresh-notifications.js";
import { notifyToolConnectionHealth } from "./tool-connection-health-notifications.js";
import { dispatchGitHubBotCloudEvent } from "./chat-github-cloud-ingress.js";
import { chatEndpoints } from "@paperclipai/db";
@@ -244,25 +245,30 @@ export function githubConnectionEventService(
const snapshot = githubSnapshotUpdate(event.payload);
if (!snapshot) return;
const appliedAt = now();
await db.update(externalObjects).set({
statusKey: snapshot.statusKey,
statusLabel: snapshot.statusLabel,
statusCategory: snapshot.statusCategory,
statusTone: snapshot.statusTone,
statusIconKey: snapshot.statusIconKey,
isTerminal: snapshot.merged || snapshot.state === "closed",
data: sql`${externalObjects.data} || ${JSON.stringify(snapshot.data)}::jsonb`,
remoteVersion: snapshot.remoteVersion,
lastResolvedAt: appliedAt,
lastChangedAt: appliedAt,
nextRefreshAt: appliedAt,
updatedAt: appliedAt,
}).where(and(
eq(externalObjects.companyId, companyId),
eq(externalObjects.providerKey, "github"),
eq(externalObjects.objectType, "pull_request"),
sql`lower(${externalObjects.externalId}) = lower(${snapshot.externalId})`,
));
await db.transaction(async tx => {
await notifyExternalObjectRefresh(tx, companyId);
await tx.update(externalObjects).set({
statusKey: snapshot.statusKey,
statusLabel: snapshot.statusLabel,
statusCategory: snapshot.statusCategory,
statusTone: snapshot.statusTone,
statusIconKey: snapshot.statusIconKey,
isTerminal: snapshot.merged || snapshot.state === "closed",
data: sql`${externalObjects.data} || ${JSON.stringify(snapshot.data)}::jsonb`,
remoteVersion: snapshot.remoteVersion,
lastResolvedAt: appliedAt,
lastChangedAt: appliedAt,
nextRefreshAt: appliedAt,
refreshStartedAt: null,
refreshToken: null,
updatedAt: appliedAt,
}).where(and(
eq(externalObjects.companyId, companyId),
eq(externalObjects.providerKey, "github"),
eq(externalObjects.objectType, "pull_request"),
sql`lower(${externalObjects.externalId}) = lower(${snapshot.externalId})`,
));
});
if (snapshot.merged && event.action === "closed") {
await issueThreadInteractionService(db, { wakeup: options.wakeup })
.sweepMergedPullRequestConfirmations([{
+2
View File
@@ -13,6 +13,7 @@ import { notifyInstructionCommit } from "./instruction-commit-notifications.js";
import { notifyScheduledRetryCancellation } from "./scheduled-retry-cancellation-notifications.js";
import { notifyVoiceChange } from "./voice/voice-recovery-notifications.js";
import { notifyToolConnectionHealth } from "./tool-connection-health-notifications.js";
import { notifyExternalObjectRefresh } from "./external-object-refresh-notifications.js";
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
import type { Db } from "@paperclipai/db";
import { companies, instanceSettings } from "@paperclipai/db";
@@ -628,6 +629,7 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService
await notifyDeliveryWork(tx, DELIVERY_QUEUES.mcpEvents, "*");
await notifyDeliveryWork(tx, DELIVERY_QUEUES.dotEvents, "*");
await notifyDeliveryWork(tx, DELIVERY_QUEUES.grokDispatch);
if (patch.enableExternalObjects !== undefined) await notifyExternalObjectRefresh(tx);
if (patch.enableWorktreeRunExecution !== undefined) await notifyScheduledRetryCancellation(tx);
if (patch.enableWorktreeRunExecution !== undefined) {
await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
+2
View File
@@ -1,3 +1,4 @@
import { notifyExternalObjectRefresh } from "./external-object-refresh-notifications.js";
import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js";
import { notifyNativeRecoveryNotices } from "./native-recovery-notice-notifications.js";
import { notifyNativeRunProjection } from "./native-run-projection-notifications.js";
@@ -11953,6 +11954,7 @@ export function issueService(db: Db) {
const [owner] = await tx.select({ companyId: issues.companyId }).from(issues).where(eq(issues.id, id));
if (owner) {
await notifyExternalConversationSettlement(tx, { companyId: owner.companyId });
await notifyExternalObjectRefresh(tx, owner.companyId);
await notifyApprovalWork(tx, owner.companyId);
}
await notifyDecisionTargetIssueRemoval(tx, id);
+27 -16
View File
@@ -1,3 +1,4 @@
import { notifyExternalObjectRefresh } from "./external-object-refresh-notifications.js";
import { isDeepStrictEqual } from "node:util";
import { writePluginJobChange } from "./plugin-job-work-notifications.js";
import { asc, eq, isNull, ne, sql, and } from "drizzle-orm";
@@ -239,12 +240,15 @@ export function pluginRegistryService(db: Db) {
setClause.categories = data.manifest.categories;
}
return db
.update(plugins)
.set(setClause)
.where(eq(plugins.id, id))
.returning()
.then((rows) => rows[0] ?? null);
return db.transaction(async tx => {
await notifyExternalObjectRefresh(tx);
return tx
.update(plugins)
.set(setClause)
.where(eq(plugins.id, id))
.returning()
.then((rows) => rows[0] ?? null);
});
},
// ----- Status ---------------------------------------------------------
@@ -254,16 +258,19 @@ export function pluginRegistryService(db: Db) {
const plugin = await getById(id);
if (!plugin) throw notFound("Plugin not found");
return db
.update(plugins)
.set({
status: input.status,
lastError: input.lastError ?? null,
updatedAt: new Date(),
})
.where(eq(plugins.id, id))
.returning()
.then((rows) => rows[0] ?? null);
return db.transaction(async tx => {
await notifyExternalObjectRefresh(tx);
return tx
.update(plugins)
.set({
status: input.status,
lastError: input.lastError ?? null,
updatedAt: new Date(),
})
.where(eq(plugins.id, id))
.returning()
.then((rows) => rows[0] ?? null);
});
},
// ----- Uninstall / Remove --------------------------------------------
@@ -280,6 +287,7 @@ export function pluginRegistryService(db: Db) {
if (!plugin) throw notFound("Plugin not found");
return writePluginJobChange(db, async db => {
await notifyExternalObjectRefresh(db);
if (removeData) {
// Hard delete – plugin_config cascades via FK onDelete
return db
@@ -343,6 +351,7 @@ export function pluginRegistryService(db: Db) {
return db.transaction(async tx => {
const db = tx as unknown as Db;
await notifyDeliveryWork(db, DELIVERY_QUEUES.pluginLifecycle);
await notifyExternalObjectRefresh(db, companyId);
const plugin = await getById(pluginId, db);
if (!plugin) throw notFound("Plugin not found");
@@ -385,6 +394,7 @@ export function pluginRegistryService(db: Db) {
return db.transaction(async tx => {
const db = tx as unknown as Db;
await notifyDeliveryWork(db, DELIVERY_QUEUES.pluginLifecycle);
await notifyExternalObjectRefresh(db, companyId);
const plugin = await getById(pluginId, db);
if (!plugin) throw notFound("Plugin not found");
@@ -467,6 +477,7 @@ export function pluginRegistryService(db: Db) {
return db.transaction(async tx => {
const db = tx as unknown as Db;
await notifyDeliveryWork(db, DELIVERY_QUEUES.pluginLifecycle);
await notifyExternalObjectRefresh(db, companyId);
const plugin = await getById(pluginId, db);
if (!plugin) throw notFound("Plugin not found");