From 453f5fbb78c48968c1d938df2ea83502c3b33364 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sat, 31 Oct 2026 14:09:24 -0500 Subject: [PATCH] perf(external-objects): schedule scoped refresh obligations (#18384) Reviewed and verified at 3e58e12f877d70e5acab272d817ac65b273767c0. Co-Authored-By: Paperclip --- .../2026-10-30-external-objects-sleep.md | 145 ++++ .../external-object-refresh-work.test.ts | 645 ++++++++++++++++++ .../external-objects-service.test.ts | 80 ++- .../github-connection-events.test.ts | 20 +- .../__tests__/scoped-deadline-work.test.ts | 37 + .../server-startup-feedback-export.test.ts | 20 +- server/src/index.ts | 17 +- server/src/services/budgets.ts | 3 + server/src/services/companies.ts | 3 + server/src/services/company-deletion.ts | 2 + .../services/delivery-work-notifications.ts | 1 + .../external-object-refresh-notifications.ts | 8 + server/src/services/external-objects.ts | 306 ++++++--- .../src/services/github-connection-events.ts | 45 +- server/src/services/instance-settings.ts | 2 + server/src/services/issues.ts | 10 + server/src/services/plugin-registry.ts | 9 +- server/src/services/scoped-deadline-work.ts | 10 +- 18 files changed, 1174 insertions(+), 189 deletions(-) create mode 100644 doc/plans/2026-10-30-external-objects-sleep.md create mode 100644 server/src/__tests__/external-object-refresh-work.test.ts create mode 100644 server/src/services/external-object-refresh-notifications.ts diff --git a/doc/plans/2026-10-30-external-objects-sleep.md b/doc/plans/2026-10-30-external-objects-sleep.md new file mode 100644 index 00000000000..659a42c4649 --- /dev/null +++ b/doc/plans/2026-10-30-external-objects-sleep.md @@ -0,0 +1,145 @@ +# 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. + +A plugin result carries its selected plugin ID inside the host owner. Persistence +rechecks that exact plugin's readiness, read capability, object declaration and +same-company enabled setting in the result update. Another available resolver +cannot authorize an obsolete plugin response. A rejected response releases only +its own claim; the due row remains for a fresh allowed resolver. Plugin RPC +payloads and public result shapes do not contain this host selection context. + + +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. Ordinary ingestion uses object IDs +or stable provider/object/external identity hints, including concurrent upserts, +so one mention write does not reconstruct all company history. 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 object intent before mention replacement and stable identity intent before object upsert. Routes, runtime tools, adapters and plugin paths reuse these methods. | +| 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 | The issue service selects distinct mentioned object IDs by company and `sourceIssueId`, then registers only those IDs before the cascade. The source owner includes title, description, comments and documents. | +| 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. | +| 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. + + +## Host cancellation and registered ownership + +The resolver contract accepts an optional host `AbortSignal`. The worker passes +its coordinator signal through refresh to the resolver. This reuses the narrow +provider-owned capability introduced with merge-confirmation work: GitHub combines +host cancellation with a ten-second request budget for real fetch, response-body +reads and visibility probes. Plugin RPC payloads remain unchanged. The host +observes cancellation while retaining handlers for any eventual plugin/custom +result; it releases only its own claim and cannot publish that late result. +Startup does not await provider work. Shutdown closes the active GitHub HTTP +request, and its saved due row can resume immediately after restart. + +The old `refreshDueObjects*` batch helpers are removed. Existing selection, +lease, concurrent manual refresh, terminal-history and not-found recovery tests +now enter through the registered scheduler owner. The removed baseline control +recorded six SQL reads over three empty aggregate ticks; the current quiet-work +acceptance measures zero recurring SQL after settlement. The scheduled check retains its +post-refresh scoped read to revalidate current company, reference and resolver +eligibility and recover an authoritative lease after a competing writer; all such +reads target one object rather than rebuilding a company. + +### Repeated eligibility changes + +Company status and company plugin settings affect every eligible object. Their +company-wide reconstruction hints now coalesce for 30 seconds after a completed +reconstruction. This is a deadline for a newly received dirty hint, not a periodic +scan. Object and provider-identity hints remain immediate; already known object +deadlines still execute on time and re-read current company/provider eligibility. +Reactivation after a second broad change can wait up to 30 seconds to reconstruct +previously removed future obligations. Startup and instance-wide feature changes +retain their existing behavior. + +The shared scoped worker accepts an owner-selected reconstruction interval for +its typed scope; existing numeric callers retain their policy. It does not know +companies or provider identities. A real PostgreSQL regression with 1,100 future +objects reproduces five immediate page reads after a status toggle before the +fix. With the fix, five repeated status writes retain one owed reconstruction, +then execute exactly one company-scoped five-page pass. Twenty quiet minutes add +no reads or provider calls. A scheduler control proves an exact object hint and a +250ms saved deadline still preempt the broad reconstruction. All 76 affected +external-object, route, and scheduler tests and server types pass. diff --git a/server/src/__tests__/external-object-refresh-work.test.ts b/server/src/__tests__/external-object-refresh-work.test.ts new file mode 100644 index 00000000000..b07e4c3cf8c --- /dev/null +++ b/server/src/__tests__/external-object-refresh-work.test.ts @@ -0,0 +1,645 @@ +import { createServer } from "node:http"; +import { randomUUID } from "node:crypto"; +import { setTimeout as delay } from "node:timers/promises"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { and, eq, inArray, sql } from "drizzle-orm"; +import { activityLog, approvals, applyPendingMigrations, databaseWorkPending, budgetIncidents, budgetPolicies, companies, costEvents, createDb, externalObjects, externalObjectMentions, 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 { issueService } from "../services/issues.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"; +import { notifyExternalObjectRefresh } from "../services/external-object-refresh-notifications.js"; +import { DELIVERY_QUEUES } from "../services/delivery-work-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; + let temp: Awaited> | undefined; + let coordinator: ReturnType | undefined; + let admitted = true; + let reconcile = true; + let enabled = true; + const provider = vi.fn(async (): Promise => terminal); + 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)); + 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) { + 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) { + await db.transaction(async tx => { + await notifyExternalObjectRefresh(tx, companyId); + await tx.update(externalObjects).set(patch).where(eq(externalObjects.companyId, companyId)); + }); + } + + 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(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(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.each(["company setting", "status", "uninstall", "manifest", "other company"] as const)( + "revalidates the selected plugin after %s changes even while the built-in fallback remains available", async change => { + clock(); const registry = pluginRegistryService(db); + const manifest: PaperclipPluginManifestV1 = { + id: `paperclip.selected-${randomUUID()}`, apiVersion: 1, version: "1.0.0", displayName: "Selected provider", + description: "Selected resolver fixture", author: "Paperclip", categories: ["connector"], + capabilities: ["external.objects.detect", "external.objects.read"], entrypoints: { worker: "worker.js" }, + objectReferences: [{ providerKey: "github", displayName: "GitHub", objectTypes: ["pull_request"], urlPatterns: ["https://github.com/:owner/:repo/pull/:id"] }], + }; + const plugin = await registry.install({ packageName: "selected-provider-fixture" }, manifest); + await registry.updateStatus(plugin!.id, { status: "ready" }); + let finishPlugin!: (result: ExternalObjectResolveResult) => void; + const pluginResult = new Promise(resolve => { finishPlugin = resolve; }); + let selected = false; + const calls = vi.fn(async (_id: string, method: string) => { + if (method === "detectExternalObjects") return { detections: [] }; + selected = true; return pluginResult; + }); + let finishFallback!: () => void; + const fallbackResult = new Promise(resolve => { finishFallback = resolve; }); + const fallback = vi.fn(async () => { + await fallbackResult; + return new Response(JSON.stringify({ state: "closed", merged: true, title: "Allowed fallback" }), { status: 200 }); + }); + const svc = externalObjectService(db, { pluginWorkerManager: { call: calls } as unknown as PluginWorkerManager, + github: { fetch: fallback, credentialProvider: () => ({ token: "simulated-company-token", source: "company_secret" }) } }); + try { + await start(svc); const f = await fixture(); + await db.update(issues).set({ description: "https://github.com/acme/app/pull/42" }).where(eq(issues.id, f.issueId)); + await svc.syncIssue(f.issueId); await settle(() => selected); + if (change === "company setting") await registry.upsertCompanySettings(plugin!.id, f.companyId, { enabled: false, settingsJson: {} }); + else if (change === "status") await registry.updateStatus(plugin!.id, { status: "error" }); + else if (change === "uninstall") await registry.uninstall(plugin!.id, true); + else if (change === "manifest") await registry.update(plugin!.id, { manifest: { ...manifest, capabilities: [], objectReferences: [] } }); + else { + const other = await fixture(); + await registry.upsertCompanySettings(plugin!.id, other.companyId, { enabled: false, settingsJson: {} }); + } + finishPlugin({ ok: true, snapshot: { statusKey: "plugin_result", statusCategory: "succeeded", statusTone: "success", isTerminal: true } }); + if (change === "other company") { + await settle(async () => (await row(f.companyId))?.statusKey === "plugin_result"); await quiet(); + expect(fallback).not.toHaveBeenCalled(); + } else { + await settle(async () => fallback.mock.calls.length > 0 || Boolean((await row(f.companyId))?.lastResolvedAt)); + const beforeFallback = await row(f.companyId); + expect(beforeFallback.isTerminal).toBe(false); expect(beforeFallback.statusKey).toBeNull(); expect(beforeFallback.lastResolvedAt).toBeNull(); + expect(fallback).toHaveBeenCalledOnce(); + finishFallback(); + await settle(async () => (await row(f.companyId))?.statusKey === "merged"); await quiet(); + expect((await row(f.companyId)).displayTitle).toBe("acme/app#42: Allowed fallback"); + } + } finally { finishPlugin(terminal); finishFallback(); } + }); + + 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 () => { + const left = await row(a.companyId), right = await row(b.companyId); + return Boolean(left?.lastResolvedAt && right?.lastResolvedAt); + }); + const expectedDeadline = Math.min((await row(a.companyId)).nextRefreshAt!.getTime(), (await row(b.companyId)).nextRefreshAt!.getTime()); + await settle(() => coordinator!.nextWakeAt() === expectedDeadline && !databaseWorkPending(db, DELIVERY_QUEUES.externalObjectRefresh)); + 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("coalesces repeated company status writes without rebuilding every future object", async () => { + clock(); const svc = service(); const own = await fixture(), foreign = await fixture(); + const future = new Date(Date.now() + 86_400_000); + for (let offset = 0; offset < 1100; offset += 250) { + const ids = Array.from({ length: Math.min(250, 1100 - offset) }, () => randomUUID()); + await db.insert(externalObjects).values(ids.map(id => ({ id, companyId: own.companyId, + providerKey: "url", objectType: "link", externalId: id, nextRefreshAt: future }))); + await db.insert(externalObjectMentions).values(ids.map(objectId => ({ companyId: own.companyId, + sourceIssueId: own.issueId, sourceKind: "description" as const, objectId }))); + } + await svc.syncIssue(foreign.issueId); + await db.update(externalObjects).set({ nextRefreshAt: future }); + await start(svc); + await settle(() => coordinator!.nextWakeAt() === future.getTime()); + 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 { + const companiesSvc = companyService(db); + await companiesSvc.update(own.companyId, { status: "paused" }); + await settle(() => reads.length === 1 && coordinator!.nextWakeAt() === future.getTime() + && !databaseWorkPending(db, DELIVERY_QUEUES.externalObjectRefresh)); + for (const status of ["active", "paused", "active", "paused", "active"] as const) { + await companiesSvc.update(own.companyId, { status }); + await settle(() => !databaseWorkPending(db, DELIVERY_QUEUES.externalObjectRefresh) + && (coordinator!.nextWakeAt() ?? 0) > Date.now()); + expect(reads).toHaveLength(1); + } + const rebuildAt = coordinator!.nextWakeAt()!; + expect(rebuildAt - Date.now()).toBeGreaterThan(20_000); + expect(rebuildAt - Date.now()).toBeLessThanOrEqual(30_000); + await vi.advanceTimersByTimeAsync(rebuildAt - Date.now()); + await settle(() => coordinator!.nextWakeAt() === future.getTime() && reads.length === 6); + for (const read of reads) { + expect(read.query).toContain('"external_objects"."company_id" ='); + expect(read.params).toContain(own.companyId); + expect(read.params).not.toContain(foreign.companyId); + expect(read.params.at(-1)).toBe(250); + } + const count = reads.length; + await vi.advanceTimersByTimeAsync(20 * 60_000); + expect(reads).toHaveLength(count); expect(provider).not.toHaveBeenCalled(); + expect(coordinator!.nextWakeAt()).toBe(future.getTime()); + expect(errors).not.toHaveBeenCalled(); + } finally { db.$client.options.debug = debug; } + }); + + it("scopes repeated issue deletions to cascaded source objects and cancels their last future mention", async () => { + clock(); const svc = service(); + const own = await fixture(), foreign = await fixture(); + const future = new Date(Date.now() + 86_400_000), removedDeadline = new Date(Date.now() + 1_800_000); + const sharedUrl = "https://example.com/shared-issue-object"; + const [retained] = await db.insert(issues).values({ companyId: own.companyId, title: "Retained", + description: sharedUrl }).returning(); + await svc.syncIssue(retained!.id); + await db.update(issues).set({ description: sharedUrl }).where(eq(issues.id, foreign.issueId)); + await svc.syncIssue(foreign.issueId); + const deletions: Array<{ id: string; affected: string[]; orphaned: string[] }> = []; + for (let index = 0; index < 3; index++) { + const [issue] = await db.insert(issues).values({ companyId: own.companyId, + title: sharedUrl, description: `https://example.com/deleted-${index}` }).returning(); + await svc.syncIssue(issue!.id); + const [comment] = await db.insert(issueComments).values({ companyId: own.companyId, issueId: issue!.id, + body: `https://example.com/comment-${index}` }).returning(); + await svc.syncComment(comment!.id); + const doc = await documentService(db).upsertIssueDocument({ issueId: issue!.id, key: "notes", format: "markdown", + body: `https://example.com/document-${index}` }); + await svc.syncDocument(doc.document.id); + const mentions = await db.select().from(externalObjectMentions).where(and( + eq(externalObjectMentions.companyId, own.companyId), eq(externalObjectMentions.sourceIssueId, issue!.id))); + expect(new Set(mentions.map(mention => mention.sourceKind))).toEqual(new Set(["title", "description", "comment", "document"])); + deletions.push({ id: issue!.id, affected: [...new Set(mentions.map(mention => mention.objectId!))], + orphaned: mentions.filter(mention => mention.sourceKind !== "title").map(mention => mention.objectId!) }); + } + const shared = (await db.select().from(externalObjectMentions).where(eq(externalObjectMentions.sourceIssueId, retained!.id)))[0]!.objectId!; + for (let offset = 0; offset < 1100; offset += 250) { + const ids = Array.from({ length: Math.min(250, 1100 - offset) }, () => randomUUID()); + await db.insert(externalObjects).values(ids.map(id => ({ id, companyId: own.companyId, + providerKey: "url", objectType: "link", externalId: id, nextRefreshAt: future }))); + await db.insert(externalObjectMentions).values(ids.map(objectId => ({ companyId: own.companyId, + sourceIssueId: own.issueId, sourceKind: "description" as const, objectId }))); + } + await db.update(externalObjects).set({ nextRefreshAt: future }); + for (const entry of deletions) for (const id of entry.orphaned) + await db.update(externalObjects).set({ nextRefreshAt: removedDeadline }).where(eq(externalObjects.id, id)); + await start(svc); + await settle(() => coordinator!.nextWakeAt() === removedDeadline.getTime() && !databaseWorkPending(db, DELIVERY_QUEUES.externalObjectRefresh)); + expect(provider).not.toHaveBeenCalled(); + const reads: Array<{ query: string; params: unknown[] }> = []; + const selections: 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] }); + if (/^select distinct/i.test(query) && query.includes('from "external_object_mentions"')) + selections.push({ query, params: [...params] }); + }; + try { + for (let index = 0; index < deletions.length; index++) { + const entry = deletions[index], first = reads.length; + const result = await issueService(db).remove(entry.id); + expect(result?.id).toBe(entry.id); + const next = index === deletions.length - 1 ? future : removedDeadline; + await settle(() => reads.length > first && coordinator!.nextWakeAt() === next.getTime() + && !databaseWorkPending(db, DELIVERY_QUEUES.externalObjectRefresh)); + const deletionReads = reads.slice(first); + expect(deletionReads.length).toBeLessThanOrEqual(entry.affected.length); + for (const read of deletionReads) { + expect(read.query).toContain('"external_objects"."id" ='); + expect(read.params).toContain(own.companyId); expect(read.params).not.toContain(foreign.companyId); + expect(read.params.some(value => entry.affected.includes(String(value)))).toBe(true); + } + } + expect(selections).toHaveLength(deletions.length); + for (let index = 0; index < selections.length; index++) { + expect(selections[index]!.query).toContain('"external_object_mentions"."company_id" ='); + expect(selections[index]!.query).toContain('"external_object_mentions"."source_issue_id" ='); + expect(selections[index]!.params).toContain(own.companyId); expect(selections[index]!.params).toContain(deletions[index]!.id); + } + const lastMentions = await db.select().from(externalObjectMentions).where(eq(externalObjectMentions.objectId, shared)); + expect(lastMentions).toHaveLength(1); expect(lastMentions[0]!.sourceIssueId).toBe(retained!.id); + const orphanIds = deletions.flatMap(entry => entry.orphaned); + const orphanRows = await db.select().from(externalObjects).where(inArray(externalObjects.id, orphanIds)); + expect(orphanRows).toHaveLength(orphanIds.length); + for (const orphan of orphanRows) expect(orphan.nextRefreshAt).toEqual(removedDeadline); + expect(await db.select().from(externalObjectMentions).where(inArray(externalObjectMentions.objectId, orphanIds))).toEqual([]); + const foreignRows = await db.select().from(externalObjects).where(eq(externalObjects.companyId, foreign.companyId)); + expect(foreignRows).toHaveLength(1); expect(foreignRows[0]!.nextRefreshAt).toEqual(future); + const count = reads.length; + await vi.advanceTimersByTimeAsync(removedDeadline.getTime() - Date.now() + 1); + expect(reads).toHaveLength(count); expect(provider).not.toHaveBeenCalled(); + expect(coordinator!.nextWakeAt()).toBe(future.getTime()); + } finally { db.$client.options.debug = debug; } + }); + + 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 cancels the observed result on 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 stop; expect(stopped).toBe(true); coordinator = undefined; + const cancelled = await row(f.companyId); expect(cancelled.refreshToken).toBeNull(); expect(cancelled.isTerminal).toBe(false); + finish(terminal); await delay(10); + expect((await row(f.companyId)).isTerminal).toBe(false); + const selects = vi.spyOn(db, "select"); + await vi.advanceTimersByTimeAsync(600_000); + expect(selects).not.toHaveBeenCalled(); expect(provider).toHaveBeenCalledTimes(2); + }); + + it("cancels an actual stalled GitHub HTTP body, retains due debt and resumes after restart", async () => { + clock(); let stall = true, started = false, closed = false; + const server = createServer((request, response) => { + if (request.url?.includes("/pulls/")) { + if (stall) { + started = true; response.on("close", () => { closed = true; }); + response.writeHead(200, { "content-type": "application/json" }); response.write('{"state":'); + } else response.end(JSON.stringify({ state: "closed", merged: true, title: "Done" })); + } else response.end(JSON.stringify({ private: false })); + }); + await new Promise(resolve => server.listen(0, "127.0.0.1", resolve)); + const address = server.address(); if (!address || typeof address === "string") throw new Error("Missing HTTP test port"); + const svc = externalObjectService(db, { github: { + fetch: (url, init) => fetch(`http://127.0.0.1:${address.port}${new URL(url).pathname}`, init), + credentialProvider: () => ({ token: "simulated-company-token", source: "company_secret" }), + } }); + try { + await start(svc); const f = await fixture(); + await db.update(issues).set({ description: "https://github.com/acme/app/pull/42" }).where(eq(issues.id, f.issueId)); + await svc.syncIssue(f.issueId); await settle(() => started); + await coordinator!.stop(); coordinator = undefined; + await settle(() => closed); + const cancelled = await row(f.companyId); + expect(cancelled.refreshToken).toBeNull(); expect(cancelled.lastResolvedAt).toBeNull(); + expect(cancelled.nextRefreshAt!.getTime()).toBeLessThanOrEqual(Date.now()); + stall = false; await start(svc); + await settle(async () => Boolean((await row(f.companyId))?.isTerminal)); await quiet(); + } finally { await coordinator?.stop(); coordinator = undefined; server.closeAllConnections(); await new Promise(resolve => server.close(() => resolve())); } + }); + + it("loads only the changed identity and objects after ingestion into ten thousand future objects", async () => { + clock(); const svc = service(); const f = await fixture(); + const future = new Date(Date.now() + 86_400_000); + const ids = Array.from({ length: 10_000 }, () => randomUUID()); + for (let offset = 0; offset < ids.length; offset += 500) { + const batch = ids.slice(offset, offset + 500); + await db.insert(externalObjects).values(batch.map(id => ({ id, companyId: f.companyId, + providerKey: "url", objectType: "link", externalId: id, nextRefreshAt: future }))); + await db.insert(externalObjectMentions).values(batch.map(objectId => ({ companyId: f.companyId, + sourceIssueId: f.issueId, sourceKind: "description" as const, objectId }))); + } + await start(svc); await settle(() => coordinator!.nextWakeAt() === future.getTime()); + const [incoming] = await db.insert(issues).values({ companyId: f.companyId, title: "Incoming", + description: "https://example.com/new-object" }).returning(); + 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 { + await svc.syncIssue(incoming!.id); + await settle(() => provider.mock.calls.length === 1); + await settle(() => coordinator!.nextWakeAt() === future.getTime() && !databaseWorkPending(db, DELIVERY_QUEUES.externalObjectRefresh)); + expect(reads.length).toBeGreaterThan(0); expect(reads.length).toBeLessThanOrEqual(6); + for (const read of reads) expect(read.query).toMatch(/"external_objects"\."(?:id|external_id)" =/); + } finally { db.$client.options.debug = debug; } + }); + + 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(); + }); +}); diff --git a/server/src/__tests__/external-objects-service.test.ts b/server/src/__tests__/external-objects-service.test.ts index af3b4667deb..419580a1fbb 100644 --- a/server/src/__tests__/external-objects-service.test.ts +++ b/server/src/__tests__/external-objects-service.test.ts @@ -1,9 +1,10 @@ -import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { randomUUID } from "node:crypto"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { eq } from "drizzle-orm"; import { companies, + applyPendingMigrations, createDb, activityLog, externalObjectMentions, @@ -12,6 +13,7 @@ import { issues, plugins, } from "@paperclipai/db"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase, @@ -27,8 +29,8 @@ import type { PaperclipPluginManifestV1 } from "@paperclipai/shared"; import type { PluginWorkerManager } from "../services/plugin-worker-manager.js"; import { createGitHubExternalObjectProvider } from "../services/github-external-object-provider.js"; -const external = resolveExternalTestDatabaseUrl("PAPERCLIP_EXTERNAL_OBJECTS_TEST_DATABASE_URL"); -const embeddedPostgresSupport = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); +const externalDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_EXTERNAL_OBJECTS_TEST_DATABASE_URL"); +const embeddedPostgresSupport = externalDatabaseUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; if (!embeddedPostgresSupport.supported) { @@ -711,9 +713,10 @@ describeEmbeddedPostgres("externalObjectService", () => { let tempDb: Awaited> | null = null; beforeAll(async () => { - if (!external) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-external-objects-"); - db = createDb(external ?? tempDb!.connectionString, { maxConnections: 6 }); - }, 20_000); + if (externalDatabaseUrl) await applyPendingMigrations(externalDatabaseUrl); + if (!externalDatabaseUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-external-objects-"); + db = createDb(externalDatabaseUrl ?? tempDb!.connectionString, { maxConnections: 6 }); + }, 90_000); afterEach(async () => { await db.delete(activityLog); @@ -731,6 +734,18 @@ describeEmbeddedPostgres("externalObjectService", () => { await tempDb?.cleanup(); }); + async function runRefreshWork(service: ReturnType) { + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, + onError: error => errors.push(error) }); + try { + await service.registerRefreshWork(coordinator, () => true).ready; + expect(errors).toEqual([]); + } finally { + await coordinator.stop(); + } + } + async function createIssue(companyId = randomUUID()) { const issueId = randomUUID(); await db.insert(companies).values({ @@ -880,9 +895,9 @@ describeEmbeddedPostgres("externalObjectService", () => { expect(object.nextRefreshAt).toBeInstanceOf(Date); - const refreshed = await svc.refreshDueObjects(companyId, 50, new Date(Date.now() + 1_000)); + await runRefreshWork(svc); - expect(refreshed).toHaveLength(1); + expect((await db.select().from(externalObjects))[0]?.lastResolvedAt).toBeInstanceOf(Date); expect(resolve).toHaveBeenCalledTimes(1); }); @@ -919,15 +934,13 @@ describeEmbeddedPostgres("externalObjectService", () => { await svc.syncIssue(issueId); const object = await db.select().from(externalObjects).then((rows) => rows[0]!); - const dueRefresh = svc.refreshDueObjects(companyId, 50, new Date(Date.now() + 1_000)); + const dueRefresh = runRefreshWork(svc); await resolverStarted; const manualRefresh = svc.refreshObject(object.id, { companyId, force: true }); finishResolver(); - const [dueResults, manualResult] = await Promise.all([dueRefresh, manualRefresh]); + const [, manualResult] = await Promise.all([dueRefresh, manualRefresh]); - expect(dueResults).toHaveLength(1); - expect(dueResults[0]?.refreshed).toBe(true); expect(manualResult.refreshed).toBe(true); expect(manualResult.object.statusLabel).toBe("Open"); expect(resolve).toHaveBeenCalledTimes(1); @@ -967,14 +980,12 @@ describeEmbeddedPostgres("externalObjectService", () => { await scheduledService.syncIssue(issueId); const object = await db.select().from(externalObjects).then((rows) => rows[0]!); - const dueRefresh = scheduledService.refreshDueObjects(companyId, 50, new Date(Date.now() + 1_000)); + const dueRefresh = runRefreshWork(scheduledService); await resolverStarted; const manualRefresh = await manualService.refreshObject(object.id, { companyId, force: true }); finishResolver(); - const dueResults = await dueRefresh; + await dueRefresh; - expect(dueResults).toHaveLength(1); - expect(dueResults[0]?.refreshed).toBe(true); expect(manualRefresh.refreshed).toBe(false); expect(manualRefresh.reason).toBe("refresh_in_progress"); expect(manualRefresh.object).not.toHaveProperty("refreshToken"); @@ -1030,7 +1041,7 @@ describeEmbeddedPostgres("externalObjectService", () => { const object = await db.select().from(externalObjects).then((rows) => rows[0]!); const claimedAt = new Date(Date.now() + 1_000); - const dueRefresh = scheduledService.refreshDueObjects(companyId, 50, claimedAt); + const dueRefresh = runRefreshWork(scheduledService); await resolverStarted; const replacementResult = await replacementService.refreshObject(object.id, { companyId, @@ -1038,11 +1049,10 @@ describeEmbeddedPostgres("externalObjectService", () => { now: new Date(claimedAt.getTime() + 301_000), }); finishResolver(); - const dueResults = await dueRefresh; + await dueRefresh; const finalObject = await db.select().from(externalObjects).then((rows) => rows[0]!); expect(replacementResult.refreshed).toBe(true); - expect(dueResults[0]).toMatchObject({ refreshed: false, reason: "refresh_superseded" }); expect(finalObject.statusLabel).toBe("Closed"); expect(finalObject.refreshStartedAt).toBeNull(); expect(finalObject.refreshToken).toBeNull(); @@ -1076,9 +1086,7 @@ describeEmbeddedPostgres("externalObjectService", () => { await svc.syncIssue(active.issueId); await svc.syncIssue(paused.issueId); - const result = await svc.refreshDueObjectsForActiveCompanies(50, new Date(Date.now() + 1_000)); - - expect(result).toEqual({ companies: 1, checked: 1, refreshed: 1 }); + await runRefreshWork(svc); expect(resolve).toHaveBeenCalledTimes(1); const rows = await db.select().from(externalObjects); const activeObject = rows.find((row) => row.companyId === active.companyId); @@ -1137,9 +1145,8 @@ describeEmbeddedPostgres("externalObjectService", () => { .set({ nextRefreshAt: new Date(0) }) .where(eq(externalObjects.id, object.id)); - const refreshed = await svc.refreshDueObjects(companyId); + await runRefreshWork(svc); - expect(refreshed).toEqual([]); expect(resolve).toHaveBeenCalledTimes(1); }); @@ -1182,7 +1189,7 @@ describeEmbeddedPostgres("externalObjectService", () => { }); it("re-checks previously frozen GitHub not-found objects so they can self-heal", async () => { - const { companyId } = await createIssue(); + const { companyId, issueId } = await createIssue(); const canonical = canonicalizeExternalObjectUrl("https://github.com/acme/app/pull/42"); if (!canonical) throw new Error("expected canonical url"); const inserted = await db.insert(externalObjects).values({ @@ -1223,9 +1230,10 @@ describeEmbeddedPostgres("externalObjectService", () => { const svc = externalObjectService(db, { resolvers: [resolver], github: false }); expect(object.isTerminal).toBe(true); - const refreshed = await svc.refreshDueObjects(companyId); - expect(refreshed).toHaveLength(1); - expect(refreshed[0]?.refreshed).toBe(true); + const [linked] = await db.select().from(externalObjects).where(eq(externalObjects.companyId, companyId)); + await db.insert(externalObjectMentions).values({ companyId, sourceIssueId: issueId, + sourceKind: "description", objectId: linked!.id }); + await runRefreshWork(svc); const updated = await db.select().from(externalObjects).then((rows) => rows[0]!); expect(updated.statusLabel).toBe("Open"); @@ -1234,7 +1242,7 @@ describeEmbeddedPostgres("externalObjectService", () => { }); it("does not re-check probe-verified GitHub not-found objects (deleted references are terminal)", async () => { - const { companyId } = await createIssue(); + const { companyId, issueId } = await createIssue(); const canonical = canonicalizeExternalObjectUrl("https://github.com/acme/app/pull/42"); if (!canonical) throw new Error("expected canonical url"); await db.insert(externalObjects).values({ @@ -1271,14 +1279,16 @@ describeEmbeddedPostgres("externalObjectService", () => { }; const svc = externalObjectService(db, { resolvers: [resolver], github: false }); - const refreshed = await svc.refreshDueObjects(companyId); + const [linked] = await db.select().from(externalObjects).where(eq(externalObjects.companyId, companyId)); + await db.insert(externalObjectMentions).values({ companyId, sourceIssueId: issueId, + sourceKind: "description", objectId: linked!.id }); + await runRefreshWork(svc); - expect(refreshed).toEqual([]); expect(resolver.resolve).not.toHaveBeenCalled(); }); it("does not re-check terminal objects that are not GitHub not-found verdicts", async () => { - const { companyId } = await createIssue(); + const { companyId, issueId } = await createIssue(); const canonical = canonicalizeExternalObjectUrl("https://github.com/acme/app/pull/42"); if (!canonical) throw new Error("expected canonical url"); await db.insert(externalObjects).values({ @@ -1314,9 +1324,11 @@ describeEmbeddedPostgres("externalObjectService", () => { }; const svc = externalObjectService(db, { resolvers: [resolver], github: false }); - const refreshed = await svc.refreshDueObjects(companyId); + const [linked] = await db.select().from(externalObjects).where(eq(externalObjects.companyId, companyId)); + await db.insert(externalObjectMentions).values({ companyId, sourceIssueId: issueId, + sourceKind: "description", objectId: linked!.id }); + await runRefreshWork(svc); - expect(refreshed).toEqual([]); expect(resolver.resolve).not.toHaveBeenCalled(); }); diff --git a/server/src/__tests__/github-connection-events.test.ts b/server/src/__tests__/github-connection-events.test.ts index 7c7b7025243..0417a434718 100644 --- a/server/src/__tests__/github-connection-events.test.ts +++ b/server/src/__tests__/github-connection-events.test.ts @@ -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(resolve => { markStarted = resolve; }); + const held = new Promise(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) }), }); diff --git a/server/src/__tests__/scoped-deadline-work.test.ts b/server/src/__tests__/scoped-deadline-work.test.ts index 3d838bb6c96..84cd4d77374 100644 --- a/server/src/__tests__/scoped-deadline-work.test.ts +++ b/server/src/__tests__/scoped-deadline-work.test.ts @@ -295,3 +295,40 @@ it("keeps an exact deadline and late reconstruction debt with concurrency enable expect(load).toHaveBeenCalledTimes(3); expect(check).toHaveBeenCalledOnce(); expect(onError).not.toHaveBeenCalled(); }); + +it("coalesces broad reconstruction without delaying an exact object hint or saved deadline", async () => { + vi.useFakeTimers(); + const owner = installDatabaseWorkSignals({ execute: vi.fn(async () => []) }); + coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => true, onError: vi.fn() }); + const now = Date.now(); + let dueAt = now + 60_000; + const load = vi.fn(async () => [{ id: "object", companyId: "company", dueAt }]); + const check = vi.fn(async () => null); + const worker = registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.externalObjectRefresh, + minimumPassIntervalMs: 0, + reconstructionIntervalMs: (scope: string | undefined) => scope === "company" ? 30_000 : 0, + decodeScope: hint => hint === "*" ? undefined : hint, + matches: () => true, + load, + check, + }, () => true); + await worker.ready; + worker.invalidate("company"); await vi.advanceTimersByTimeAsync(0); + expect(load).toHaveBeenCalledTimes(2); + worker.invalidate("company"); await vi.advanceTimersByTimeAsync(0); + expect(load).toHaveBeenCalledTimes(2); + expect(coordinator.nextWakeAt()).toBe(now + 30_000); + dueAt = now + 250; + worker.invalidate("object"); await vi.advanceTimersByTimeAsync(0); + expect(load).toHaveBeenCalledTimes(3); + expect(coordinator.nextWakeAt()).toBe(dueAt); + await vi.advanceTimersByTimeAsync(250); + expect(check).toHaveBeenCalledOnce(); + load.mockResolvedValue([]); + await vi.advanceTimersByTimeAsync(29_750); + expect(load).toHaveBeenCalledTimes(4); + expect(coordinator.nextWakeAt()).toBeNull(); + await vi.advanceTimersByTimeAsync(120_000); + expect(load).toHaveBeenCalledTimes(4); expect(check).toHaveBeenCalledOnce(); +}); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index b78ec44664d..cea50f28f75 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -215,7 +215,8 @@ const { }; const executionWorkspaceServiceFactoryMock = vi.fn(() => executionWorkspaceServiceMock); const externalObjectsServiceMock = { - refreshDueObjectsForActiveCompanies: vi.fn(async () => ({ companies: 0, checked: 0, refreshed: 0 })), + registerRefreshWork: vi.fn((coordinator: { register: typeof deliveryWorkRegisterMock }, _canRun: () => boolean) => + coordinator.register("external-object-refresh", { run: async () => undefined, hasPending: async () => false })), }; const externalObjectsServiceFactoryMock = vi.fn(() => externalObjectsServiceMock); const routineServiceMock = { @@ -1489,7 +1490,9 @@ describe("startServer feedback export wiring", () => { await Promise.resolve(); expect(heartbeatServiceMock.tickAgentTimers).toHaveBeenCalledTimes(1); - expect(externalObjectsServiceMock.refreshDueObjectsForActiveCompanies).toHaveBeenCalledTimes(1); + expect(externalObjectsServiceMock.registerRefreshWork).toHaveBeenCalledTimes(1); + expect(externalObjectsServiceMock.registerRefreshWork).toHaveBeenCalledWith(expect.any(Object), expect.any(Function)); + expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.externalObjectRefresh)).toBe(true); expect(issueThreadInteractionServiceMock.mergedPullRequestConfirmationPage).toHaveBeenCalledTimes(1); expect(issueThreadInteractionServiceMock.sweepMergedPullRequestConfirmations).not.toHaveBeenCalled(); expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.mergedPullRequestConfirmations)).toBe(true); @@ -1514,6 +1517,15 @@ describe("startServer feedback export wiring", () => { } }); + it.each([true, false])("does not wait for external-object provider work before readiness (scheduler=%s)", async enabled => { + loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled })); + externalObjectsServiceMock.registerRefreshWork.mockReturnValueOnce({ + ready: new Promise(() => {}), wake: vi.fn(), stop: vi.fn(async () => {}), + }); + await startServer(); + expect(externalObjectsServiceMock.registerRefreshWork).toHaveBeenCalledOnce(); + }); + it("keeps external object refresh active when heartbeat scheduling is disabled", async () => { loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: false, @@ -1541,7 +1553,9 @@ describe("startServer feedback export wiring", () => { await Promise.resolve(); await Promise.resolve(); - expect(externalObjectsServiceMock.refreshDueObjectsForActiveCompanies).toHaveBeenCalledTimes(1); + expect(externalObjectsServiceMock.registerRefreshWork).toHaveBeenCalledTimes(1); + expect(externalObjectsServiceMock.registerRefreshWork).toHaveBeenCalledWith(expect.any(Object), expect.any(Function)); + expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.externalObjectRefresh)).toBe(true); expect(routineServiceMock.tickScheduledTriggers).not.toHaveBeenCalled(); expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1); expect(environmentCustomImagesServiceMock.cleanupExpiredSetupSessions).not.toHaveBeenCalled(); diff --git a/server/src/index.ts b/server/src/index.ts index 0cd8308747c..a378b376020 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1321,19 +1321,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 @@ -1360,6 +1347,8 @@ async function startServerWithDatabaseTeardown( resolveNativeQuestion: (interaction) => deliverNativeQuestionResponse(db as any, interaction), }); const deliveryWork = app.locals.deliveryWork as ReturnType; + externalObjects.registerRefreshWork(deliveryWork, + () => !heartbeatSchedulerStopped && !isIdleTaskDrainActive() && !isWarmStandby()); if (config.nativeHarnessBackupExpirationEnabled) { nativeHarnessBackupExpiration.registerCheckpointExpiration(deliveryWork); } @@ -1933,7 +1922,6 @@ async function startServerWithDatabaseTeardown( } if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; - scheduleExternalObjectRefreshSweep(new Date()); if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; scheduleGitHubConnectionEventPoll(); @@ -1999,7 +1987,6 @@ async function startServerWithDatabaseTeardown( // as the enabled path. await startEnvironmentCleanup(); startHeartbeatSchedulerInterval(() => { - scheduleExternalObjectRefreshSweep(new Date()); scheduleGitHubConnectionEventPoll(); scheduleGitHubConnectionContinuitySweep(); }); diff --git a/server/src/services/budgets.ts b/server/src/services/budgets.ts index 1b31bcf6994..effbb8577e0 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 { notifyStatusCardIssueChange } from "./status-card-notifications.js"; import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js"; import { notifyRunDispatch } from "./run-dispatch-notifications.js"; @@ -326,6 +327,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica await notifyRunDispatch(db, { companyId: policy.companyId }); await notifyDeferredExecutionEligibility(db, { companyId: policy.companyId }); await notifyCompanyRetentionWork(db, policy.companyId); + await notifyExternalObjectRefresh(db, policy.companyId); await notifyStatusCardIssueChange(db, policy.companyId); await notifyDeliveryWork(db, DELIVERY_QUEUES.agentTimers); await notifyDeliveryWork(db, DELIVERY_QUEUES.routineSchedules); @@ -391,6 +393,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica await notifyRunDispatch(db, { companyId: policy.companyId }); await notifyDeferredExecutionEligibility(db, { companyId: policy.companyId }); await notifyCompanyRetentionWork(db, policy.companyId); + await notifyExternalObjectRefresh(db, policy.companyId); await notifyStatusCardIssueChange(db, policy.companyId); await notifyDeliveryWork(db, DELIVERY_QUEUES.agentTimers); await notifyDeliveryWork(db, DELIVERY_QUEUES.routineSchedules); diff --git a/server/src/services/companies.ts b/server/src/services/companies.ts index a2ea9543607..04552f28fe4 100644 --- a/server/src/services/companies.ts +++ b/server/src/services/companies.ts @@ -3,6 +3,7 @@ import { notifyDeferredExecutionEligibility } from "./deferred-execution-notific import { notifyRunDispatch } from "./run-dispatch-notifications.js"; 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"; @@ -339,6 +340,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) { await notifyDeferredExecutionEligibility(tx, { companyId: id }); 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 notifyStatusCardIssueChange(tx, id); @@ -498,6 +500,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) { await notifyDeferredExecutionEligibility(tx, { companyId: id }); 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 notifyStatusCardIssueChange(tx, id); diff --git a/server/src/services/company-deletion.ts b/server/src/services/company-deletion.ts index 07766b84c86..37894b17113 100644 --- a/server/src/services/company-deletion.ts +++ b/server/src/services/company-deletion.ts @@ -1,3 +1,4 @@ +import { notifyExternalObjectRefresh } from "./external-object-refresh-notifications.js"; import { notifyStatusCardEvaluation } from "./status-card-notifications.js"; import { notifyIssueRecovery } from "./issue-recovery-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js"; @@ -73,6 +74,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 notifyIssueRecovery(tx, { companyId: id }); await notifyChatCompanyConfiguration(tx, id); await chatCompanyDeletion.deleteCompanyData(tx, id); diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 562c9ed4f50..b3f96f0170c 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -18,6 +18,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", diff --git a/server/src/services/external-object-refresh-notifications.ts b/server/src/services/external-object-refresh-notifications.ts new file mode 100644 index 00000000000..e4c381ab3ed --- /dev/null +++ b/server/src/services/external-object-refresh-notifications.ts @@ -0,0 +1,8 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export type ExternalObjectRefreshIdentity = { providerKey: string; objectType: string; externalId: string; ignoreExternalIdCase?: boolean }; + +export async function notifyExternalObjectRefresh(tx: object, companyId?: string, object?: string | ExternalObjectRefreshIdentity) { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.externalObjectRefresh, + companyId ? JSON.stringify({ companyId, ...(typeof object === "string" ? { objectId: object } : object ? { identity: object } : {}) }) : "*"); +} diff --git a/server/src/services/external-objects.ts b/server/src/services/external-objects.ts index 306f043d753..17ca2005adc 100644 --- a/server/src/services/external-objects.ts +++ b/server/src/services/external-objects.ts @@ -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, type ExternalObjectRefreshIdentity } 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"; @@ -96,6 +100,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; identity?: ExternalObjectRefreshIdentity }; +type RefreshDeadline = { id: string; companyId: string; dueAt: number; providerKey: string; objectType: string; externalId: string }; + + +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), @@ -246,7 +262,7 @@ function objectSnapshot(object: ExternalObjectRecord): PluginExternalObjectRecor }; } -async function readyObjectReferencePlugins(db: Db) { +async function readyObjectReferencePlugins(db: Db, companyId?: string) { return db .select({ id: plugins.id, @@ -254,7 +270,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), @@ -268,7 +289,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) { @@ -325,31 +346,35 @@ async function resolveViaPluginProvider( db: Db, pluginWorkerManager: PluginWorkerManager | undefined, object: ExternalObjectRecord, -): Promise { +): Promise<{ pluginId: string; result: 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; if (!manifestProvidesObject(manifest, object)) continue; try { - return await pluginWorkerManager.call(provider.id, "resolveExternalObject", { + const result = await pluginWorkerManager.call(provider.id, "resolveExternalObject", { companyId: object.companyId, providerKey: object.providerKey, objectType: object.objectType, externalId: object.externalId, object: objectSnapshot(object), }); + return { pluginId: provider.id, result }; } catch (err) { logger.warn( { err, pluginId: provider.id, pluginKey: provider.pluginKey, objectId: object.id }, "plugin external object resolver failed", ); return { - ok: false, - liveness: "unreachable", - errorCode: "plugin_resolver_failed", - errorMessage: err instanceof Error ? err.message : String(err), + pluginId: provider.id, + result: { + ok: false, + liveness: "unreachable", + errorCode: "plugin_resolver_failed", + errorMessage: err instanceof Error ? err.message : String(err), + }, }; } } @@ -375,10 +400,43 @@ export function externalObjectService( ...(opts.detectors ?? []), ...(githubProvider ? [githubProvider.detector] : []), ]); - const resolverRegistry = createExternalObjectResolverRegistry([ + const resolvers = [ ...(opts.resolvers ?? []), ...(githubProvider?.resolvers ?? []), - ]); + ]; + const resolverRegistry = createExternalObjectResolverRegistry(resolvers); + + function pluginRefreshWhere(database: Db, selectedPluginId?: string) { + return exists(database.select({ id: plugins.id }).from(plugins).where(and( + selectedPluginId ? eq(plugins.id, selectedPluginId) : undefined, + 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), + )))), + ))); + } + + 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(pluginRefreshWhere(database)); + 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(); @@ -423,6 +481,7 @@ export function externalObjectService( nextRefreshAt: now, updatedAt: now, }; + await notifyExternalObjectRefresh(dbOrTx, companyId, { providerKey: detection.providerKey, objectType: detection.objectType, externalId: detection.externalId }); const inserted = await dbOrTx .insert(externalObjects) .values(values) @@ -452,6 +511,9 @@ export function externalObjectService( dbOrTx: any = db, ) { const urls = extractExternalObjectCanonicalUrls(input.text ?? ""); + const previous: Array<{ objectId: string | null }> = await dbOrTx.selectDistinct({ objectId: externalObjectMentions.objectId }) + .from(externalObjectMentions).where(sourceWhere(input)); + for (const row of previous) if (row.objectId) await notifyExternalObjectRefresh(dbOrTx, input.companyId, row.objectId); await dbOrTx.delete(externalObjectMentions).where(sourceWhere(input)); if (urls.length === 0) return; @@ -470,6 +532,7 @@ export function externalObjectService( if (seen.has(sourceKey)) continue; seen.add(sourceKey); const object = await upsertObjectFromDetection(input.companyId, detection, dbOrTx); + await notifyExternalObjectRefresh(dbOrTx, input.companyId, object.id); values.push({ companyId: input.companyId, sourceIssueId: input.sourceIssueId, @@ -521,6 +584,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; objectId: string | null }> = await tx.selectDistinct({ companyId: externalObjectMentions.companyId, objectId: externalObjectMentions.objectId }) + .from(externalObjectMentions).where(source); + for (const owner of owners) if (owner.objectId) await notifyExternalObjectRefresh(tx, owner.companyId, owner.objectId); + await tx.delete(externalObjectMentions).where(source); + } + async function syncComment(commentId: string, dbOrTx: any = db) { const runSync = async (tx: any) => { if (!(await isEnabled())) return; @@ -535,9 +606,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 tx - .delete(externalObjectMentions) - .where(and(eq(externalObjectMentions.sourceKind, "comment"), eq(externalObjectMentions.sourceRecordId, commentId))); + await deleteMissingSourceMentions("comment", commentId, tx); return; } await replaceSourceMentions({ @@ -569,9 +638,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 tx - .delete(externalObjectMentions) - .where(and(eq(externalObjectMentions.sourceKind, "document"), eq(externalObjectMentions.sourceRecordId, documentId))); + await deleteMissingSourceMentions("document", documentId, tx); return; } await replaceSourceMentions({ @@ -781,6 +848,8 @@ export function externalObjectService( companyId: string; actor?: Pick; force?: boolean; + scheduled?: boolean; + signal?: AbortSignal; now?: Date; }; @@ -802,39 +871,50 @@ export function externalObjectService( return { object: toObjectPayload(latest, now), refreshed: false, reason: "refresh_superseded" as const }; } + async function persistRefresh(object: ExternalObjectRecord, token: string, + patch: Partial, scheduled = false, selectedPluginId?: string) { + 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, + selectedPluginId ? pluginRefreshWhere(tx as unknown as Db, selectedPluginId) : undefined)).returning(); + if ((scheduled || selectedPluginId) && 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, now: Date, refreshToken: string, ) { - 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({ + input.signal?.throwIfAborted(); + const pluginResolution = await observeRefreshCancellation(resolveViaPluginProvider(db, opts.pluginWorkerManager, object), input.signal); + const resolver = pluginResolution ? null : resolverRegistry.find(object); + if (!pluginResolution && !resolver) { + 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 }; } - const result = pluginResult ?? await resolver!.resolve({ companyId: object.companyId, object }); + const result = pluginResolution?.result ?? await observeRefreshCancellation( + resolver!.resolve({ companyId: object.companyId, object, signal: input.signal }), input.signal); + input.signal?.throwIfAborted(); if (!result.ok) { // A failed refresh means the object's status is currently unknown, not // 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, @@ -844,9 +924,7 @@ export function externalObjectService( refreshStartedAt: null, refreshToken: null, updatedAt: now, - }) - .where(refreshOwnerWhere(object, refreshToken)) - .returning(); + }, input.scheduled, pluginResolution?.pluginId); if (!updated) return refreshSupersededResult(object, now); publishLiveEvent({ companyId: object.companyId, @@ -880,14 +958,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, pluginResolution?.pluginId); if (!updated) return refreshSupersededResult(object, now); const next = updated ?? object; if (objectChanged(object, next) && input.actor) { @@ -948,6 +1022,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; @@ -970,6 +1046,16 @@ export function externalObjectService( }; } + function observeRefreshCancellation(work: Promise, signal?: AbortSignal): Promise { + if (!signal) return work; + return new Promise((resolve, reject) => { + const abort = () => reject(signal.reason); + work.then(value => { signal.removeEventListener("abort", abort); resolve(value); }, + error => { signal.removeEventListener("abort", abort); reject(error); }); + if (signal.aborted) abort(); else signal.addEventListener("abort", abort, { once: true }); + }); + } + const objectRefreshesInFlight = new Map>>>(); async function refreshObject( @@ -989,7 +1075,7 @@ export function externalObjectService( const refreshKey = `${object.companyId}:${object.id}`; const existingRefresh = objectRefreshesInFlight.get(refreshKey); - if (existingRefresh) return existingRefresh; + if (existingRefresh) return observeRefreshCancellation(existingRefresh, input.signal); const claimed = await claimObjectRefresh(object, input, now); if (!claimed) { @@ -1006,7 +1092,10 @@ export function externalObjectService( } const stopRenewingRefreshLease = startRefreshLeaseRenewal(object, claimed.refreshToken); - const refresh = resolveObjectRefresh(object, input, now, claimed.refreshToken).finally(() => { + const refresh = resolveObjectRefresh(object, input, now, claimed.refreshToken).catch(async error => { + if (input.signal?.aborted) await persistRefresh(object, claimed.refreshToken!, { refreshStartedAt: null, refreshToken: null }); + throw error; + }).finally(() => { stopRenewingRefreshLease(); objectRefreshesInFlight.delete(refreshKey); }); @@ -1031,73 +1120,69 @@ export function externalObjectService( return results; } - 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 due = await db - .select({ id: externalObjects.id }) - .from(externalObjects) - .where( - and( - eq(externalObjects.companyId, companyId), - refreshable, - lte(externalObjects.nextRefreshAt, now), - or( - isNull(externalObjects.refreshStartedAt), - lte(externalObjects.refreshStartedAt, staleRefreshStartedBefore), - ), - ), - ) - .limit(limit); - const results = []; - for (const row of due) { - results.push(await refreshObject(row.id, { - companyId, - actor: { actorType: "system", actorId: "external-object-resolver", agentId: null, runId: null }, - now, - })); + function registerRefreshWork( + coordinator: ReturnType, + canRun: () => boolean, + ) { + const cursors = new Map(); + 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, + providerKey: externalObjects.providerKey, objectType: externalObjects.objectType, externalId: externalObjects.externalId }).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, + scope?.identity ? and(eq(externalObjects.providerKey, scope.identity.providerKey), + eq(externalObjects.objectType, scope.identity.objectType), scope.identity.ignoreExternalIdCase + ? sql`lower(${externalObjects.externalId}) = lower(${scope.identity.externalId})` + : eq(externalObjects.externalId, scope.identity.externalId)) : 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 }; } - return results; - } - - async function refreshDueObjects(companyId: string, limit = 50, now = new Date()) { - if (!(await isEnabled())) return []; - return refreshDueObjectsUnchecked(companyId, limit, now); - } - - async function refreshDueObjectsForActiveCompanies(limitPerCompany = 50, now = new Date()) { - if (!(await isEnabled())) return { companies: 0, checked: 0, refreshed: 0 }; - const activeCompanies = await db - .select({ id: companies.id }) - .from(companies) - .where(eq(companies.status, "active")); - let checked = 0; - let refreshed = 0; - for (const company of activeCompanies) { - const results = await refreshDueObjectsUnchecked(company.id, limitPerCompany, now); - checked += results.length; - refreshed += results.filter((result) => result.refreshed).length; + function deadline(object: Pick): RefreshDeadline { + return { id: object.id, companyId: object.companyId, providerKey: object.providerKey, objectType: object.objectType, externalId: object.externalId, + dueAt: Math.max(object.nextRefreshAt!.getTime(), object.refreshStartedAt + ? object.refreshStartedAt.getTime() + DEFAULT_REFRESH_LEASE_SECONDS * 1000 : 0) }; } - return { companies: activeCompanies.length, checked, refreshed }; + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.externalObjectRefresh, + batchSize: 50, + minimumPassIntervalMs: 0, + // Eligibility changes affect every object in the company. Coalesce repeated + // rebuilds while object/identity hints and saved deadlines remain immediate. + reconstructionIntervalMs: scope => scope && !scope.objectId && !scope.identity ? 30_000 : 0, + decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint) as RefreshScope, + matches: (entry, scope) => !scope || (entry.companyId === scope.companyId + && (!scope.objectId || entry.id === scope.objectId) + && (!scope.identity || (entry.providerKey === scope.identity.providerKey && entry.objectType === scope.identity.objectType + && (scope.identity.ignoreExternalIdCase ? entry.externalId.toLowerCase() === scope.identity.externalId.toLowerCase() + : entry.externalId === scope.identity.externalId)))), + 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, + signal, + 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 { @@ -1113,7 +1198,6 @@ export function externalObjectService( getProjectSummary, refreshObject, refreshIssueObjects, - refreshDueObjects, - refreshDueObjectsForActiveCompanies, + registerRefreshWork, }; } diff --git a/server/src/services/github-connection-events.ts b/server/src/services/github-connection-events.ts index 47bdac122a2..643bc557b93 100644 --- a/server/src/services/github-connection-events.ts +++ b/server/src/services/github-connection-events.ts @@ -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,31 @@ 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, { providerKey: "github", objectType: "pull_request", + externalId: snapshot.externalId, ignoreExternalIdCase: true }); + 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([{ diff --git a/server/src/services/instance-settings.ts b/server/src/services/instance-settings.ts index 81b95761469..9c833a471f4 100644 --- a/server/src/services/instance-settings.ts +++ b/server/src/services/instance-settings.ts @@ -16,6 +16,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"; @@ -635,6 +636,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 (chatBecameEnabled) await notifyDeliveryWork(tx, DELIVERY_QUEUES.strandedAssignments, "*"); if (patch.enableWorktreeRunExecution !== undefined) await notifyScheduledRetryCancellation(tx); if (patch.enableWorktreeRunExecution !== undefined) { diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index d15ff12cde6..f3e35046ac9 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -1,3 +1,4 @@ +import { notifyExternalObjectRefresh } from "./external-object-refresh-notifications.js"; import { notifyMergeConfirmationWork } from "./merged-pr-confirmation-notifications.js"; import { notifyRoutineExecutionStatus } from "./routine-execution-status-notifications.js"; import { notifyIssueAttentionRetentionWork } from "./retention-work-notifications.js"; @@ -96,6 +97,7 @@ import { heartbeatRuns, routineRuns, executionWorkspaces, + externalObjectMentions, issueApprovals, issueAccessGrants, issueAttachments, @@ -12022,6 +12024,14 @@ export function issueService(db: Db) { if (owner) { await notifyIssueAttentionRetentionWork(tx, { companyId: owner.companyId, issueId: id }); await notifyExternalConversationSettlement(tx, { companyId: owner.companyId }); + const mentionedObjects = await tx.selectDistinct({ objectId: externalObjectMentions.objectId }) + .from(externalObjectMentions) + .where(and( + eq(externalObjectMentions.companyId, owner.companyId), + eq(externalObjectMentions.sourceIssueId, id), + isNotNull(externalObjectMentions.objectId), + )); + for (const mention of mentionedObjects) await notifyExternalObjectRefresh(tx, owner.companyId, mention.objectId!); await notifyApprovalWork(tx, owner.companyId); await notifyStatusCardIssueChange(tx, owner.companyId); } diff --git a/server/src/services/plugin-registry.ts b/server/src/services/plugin-registry.ts index 4be581f4394..68d15d5d9de 100644 --- a/server/src/services/plugin-registry.ts +++ b/server/src/services/plugin-registry.ts @@ -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"; @@ -89,9 +90,9 @@ export function pluginRegistryService(db: Db) { } async function writeWorkspaceProviderChange(reconsiderWorkspaces: boolean, write: (database: Db) => Promise): Promise { - if (!reconsiderWorkspaces) return write(db); return db.transaction(async tx => { - await notifyDeliveryWork(tx, DELIVERY_QUEUES.terminalWorkspaces); + await notifyExternalObjectRefresh(tx); + if (reconsiderWorkspaces) await notifyDeliveryWork(tx, DELIVERY_QUEUES.terminalWorkspaces); return write(tx as unknown as Db); }); } @@ -292,6 +293,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 @@ -355,6 +357,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"); @@ -397,6 +400,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"); @@ -479,6 +483,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"); diff --git a/server/src/services/scoped-deadline-work.ts b/server/src/services/scoped-deadline-work.ts index fd90a9b372e..bfcf13a2e17 100644 --- a/server/src/services/scoped-deadline-work.ts +++ b/server/src/services/scoped-deadline-work.ts @@ -13,7 +13,7 @@ export function registerScopedDeadlineWork( batchSize?: number; minimumPassIntervalMs?: number; concurrentChecks?: { limit: number; key(entry: Entry): string }; - reconstructionIntervalMs?: number; + reconstructionIntervalMs?: number | ((scope: Scope | undefined) => number); beforePass?(): Promise; decodeScope(hint: string): Scope | undefined; matches(entry: Entry, scope?: Scope): boolean; @@ -62,7 +62,7 @@ export function registerScopedDeadlineWork( } }, run: async signal => { if (!canRun() || Date.now() < earliest) return; - if ((options.reconstructionIntervalMs ?? 0) > 0) for (const [hint, until] of reconstructAfter) { + for (const [hint, until] of reconstructAfter) { if (until <= Date.now()) reconstructAfter.delete(hint); } const ready = [...dirty].filter(hint => Math.max(reconstructAfter.get(hint) ?? 0, scopeFailures.get(hint)?.retryAt ?? 0) <= Date.now()); @@ -98,8 +98,10 @@ export function registerScopedDeadlineWork( entries.delete(entry.id); failures.delete(entry.id); } scans.delete(hint); - if ((options.reconstructionIntervalMs ?? 0) > 0) { - reconstructAfter.set(hint, Date.now() + options.reconstructionIntervalMs!); + const reconstructionInterval = typeof options.reconstructionIntervalMs === "function" + ? options.reconstructionIntervalMs(scope) : (options.reconstructionIntervalMs ?? 0); + if (reconstructionInterval > 0) { + reconstructAfter.set(hint, Date.now() + reconstructionInterval); } if ((versions.get(hint) ?? 0) !== scan.version) dirty.add(hint); else versions.delete(hint); if (!scope) scopeFailures.clear(); else scopeFailures.delete(hint);