db: env-configurable client options; parallelize attention feed queries (#10795)
## Thinking Path
> - Paperclip is the open source app people use to manage AI agents for
work
> - The server stores all state in PostgreSQL through Drizzle and the
postgres.js driver
> - Self-hosted installs run Postgres on localhost, so per-query latency
is near zero; hosted installs often attach Postgres over a network,
sometimes through a transaction-mode pooler
> - The DB client passes no options to the driver, so operators cannot
disable prepared statements or tune the pool without a source edit, and
the deploy docs told them to edit `client.ts`
> - The attention feed also runs its related-data lookups one after
another, so its latency grows as queries × network round trip
> - This pull request adds optional environment configuration for the DB
client and batches the independent attention-feed lookups with
`Promise.all`
> - The benefit is that network-attached deployments get correct pooler
support and a much faster attention feed, while self-hosted behavior
does not change
## Linked Issues or Issue Description
No public issue exists for this; description follows the bug report
template:
**What happened?**
On deployments where PostgreSQL is network-attached (managed providers,
pooled endpoints), the attention feed endpoint is slow:
`attentionService.list()` awaits ~15–20 queries strictly in sequence, so
a 70ms round trip turns into more than one second of pure network wait
per call. Separately, connecting through a transaction-mode pooler
(pgbouncer, Supavisor port 6543, Neon `-pooler` hosts) requires
disabling prepared statements, and the only documented way was to
hand-edit `packages/db/src/client.ts` — which `doc/DATABASE.md` itself
tells operators not to do.
**Expected behavior**
The DB client is configurable from the environment (prepared statements,
pool size, timeouts) with driver defaults when unset, and hot read paths
do not multiply network latency by issuing independent queries
sequentially.
**Steps to reproduce**
1. Run the server with `DATABASE_URL` pointing at a Postgres instance
with ~70ms round-trip latency.
2. Open the attention feed (`GET /companies/:companyId/attention`) and
measure response time — it exceeds one second even with little data.
3. Try to connect through a transaction-mode pooler: there is no
supported configuration to disable prepared statements.
## What Changed
- `packages/db/src/client.ts`: `createDb` accepts a
`DatabaseClientOptions` argument and reads optional env config —
`DATABASE_PREPARED_STATEMENTS`, `DATABASE_POOL_MAX`,
`DATABASE_IDLE_TIMEOUT_SECONDS`, `DATABASE_CONNECT_TIMEOUT_SECONDS`.
When nothing is set, no option is passed to the driver and behavior is
identical to the previous bare `postgres(url)`.
- `packages/db/src/client-options.test.ts` (new): env parsing and
driver-option mapping tests, including malformed-value rejection.
- `server/src/services/attention.ts`: the independent related-data
lookups in each feed section now run under `Promise.all` (issue
summary/image/plan-document maps, decision bundle titles, blocked-issue
maps, the newer-runs scan). Section order, item assembly, and query
shapes are unchanged.
- `doc/DATABASE.md` and `docs/deploy/database.md`: the edit-source
pooling instruction is replaced with the env toggle, plus a short
client-tuning reference.
## Verification
- `pnpm --filter @paperclipai/db exec vitest run
src/client-options.test.ts` — 6 tests pass.
- `pnpm --filter server exec vitest run
src/__tests__/attention-service.test.ts` — 22 tests pass.
- `pnpm --filter server exec vitest run
src/__tests__/decisions-service.test.ts
src/__tests__/decision-training.test.ts` — 45 tests pass; this covers
the call path that runs `attentionService.list()` inside
`db.transaction`, where postgres.js serializes queries on the reserved
connection.
- `tsc` reports no errors in the changed files.
## Risks
- Low risk for self-hosted installs: with no env vars set,
`postgres(url, {})` receives an empty options object, which postgres.js
treats the same as no options — driver defaults throughout.
- The `Promise.all` batches only group queries that had no data
dependency on each other; on the transaction call path the driver still
executes them one at a time on the reserved connection, so transactional
semantics are unchanged.
- Malformed env values now fail fast at startup with a clear message
instead of being silently ignored; this is intentional and only affects
operators who set the new variables.
## Model Used
Claude Fable 5 (`claude-fable-5`), Anthropic — via Claude Code CLI,
extended thinking enabled, tool use (test execution, live latency
measurement against a network-attached Postgres to size the problem).
## Checklist
- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above (searched "prepared statements", "pgbouncer", "pool",
"attention feed", "lockfile" — closest matches are #10573/#10787
lockfile chores, unrelated to this change)
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [ ] All Paperclip CI gates are green
- [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge
This commit is contained in:
Devin Foley
GitHub
parent
2ab797dcbe
commit
cb52f0b750
|
|
@ -113,7 +113,18 @@ DATABASE_URL=postgres://postgres.[PROJECT-REF]:[PASSWORD]@aws-0-[REGION].pooler.
|
|||
DATABASE_MIGRATION_URL=postgres://postgres.[PROJECT-REF]:[PASSWORD]@aws-0-[REGION].pooler.supabase.com:5432/postgres
|
||||
```
|
||||
|
||||
If your hosted database requires transaction-pooling-only connections, use a direct or session-pooled connection for Paperclip until runtime pooling support is documented in this guide. Do not edit database client source files as part of deployment setup.
|
||||
If your hosted database requires transaction-pooling-only connections (pgbouncer transaction mode, Supavisor port 6543, Neon `-pooler` endpoints), set `DATABASE_PREPARED_STATEMENTS=false` so the client does not rely on session-scoped prepared statements, and keep `DATABASE_MIGRATION_URL` on a direct connection. Do not edit database client source files as part of deployment setup.
|
||||
|
||||
### Client tuning (optional)
|
||||
|
||||
All of these are optional; when unset, the driver defaults apply and behavior is unchanged — typical self-hosted setups need none of them:
|
||||
|
||||
```sh
|
||||
DATABASE_PREPARED_STATEMENTS=false # required for transaction-mode poolers; default: enabled
|
||||
DATABASE_POOL_MAX=25 # connection pool size; default: 10
|
||||
DATABASE_IDLE_TIMEOUT_SECONDS=60 # close idle pooled connections; default: keep open
|
||||
DATABASE_CONNECT_TIMEOUT_SECONDS=10 # default: 30
|
||||
```
|
||||
|
||||
### Push the schema
|
||||
|
||||
|
|
|
|||
|
|
@ -56,16 +56,14 @@ For production, use a hosted provider like [Supabase](https://supabase.com/).
|
|||
|
||||
Use the **direct connection** (port 5432) for migrations and the **pooled connection** (port 6543) for the application.
|
||||
|
||||
If using connection pooling, disable prepared statements:
|
||||
If using connection pooling (transaction mode), disable prepared statements via the environment — no source edits needed:
|
||||
|
||||
```ts
|
||||
// packages/db/src/client.ts
|
||||
export function createDb(url: string) {
|
||||
const sql = postgres(url, { prepare: false });
|
||||
return drizzlePg(sql, { schema });
|
||||
}
|
||||
```sh
|
||||
DATABASE_PREPARED_STATEMENTS=false
|
||||
```
|
||||
|
||||
Related optional client tuning (driver defaults apply when unset): `DATABASE_POOL_MAX`, `DATABASE_IDLE_TIMEOUT_SECONDS`, `DATABASE_CONNECT_TIMEOUT_SECONDS`.
|
||||
|
||||
## Switching Between Modes
|
||||
|
||||
| `DATABASE_URL` | Mode |
|
||||
|
|
|
|||
|
|
@ -0,0 +1,57 @@
|
|||
import { describe, expect, it } from "vitest";
|
||||
import { databaseClientOptionsFromEnv, postgresJsOptions } from "./client.js";
|
||||
|
||||
describe("databaseClientOptionsFromEnv", () => {
|
||||
it("returns no options when nothing is set, preserving driver defaults", () => {
|
||||
expect(databaseClientOptionsFromEnv({})).toEqual({});
|
||||
expect(postgresJsOptions(databaseClientOptionsFromEnv({}))).toEqual({});
|
||||
});
|
||||
|
||||
it("ignores empty values", () => {
|
||||
expect(
|
||||
databaseClientOptionsFromEnv({
|
||||
DATABASE_PREPARED_STATEMENTS: "",
|
||||
DATABASE_POOL_MAX: "",
|
||||
}),
|
||||
).toEqual({});
|
||||
});
|
||||
|
||||
it("parses prepared-statement toggles", () => {
|
||||
expect(databaseClientOptionsFromEnv({ DATABASE_PREPARED_STATEMENTS: "false" })).toEqual({ prepare: false });
|
||||
expect(databaseClientOptionsFromEnv({ DATABASE_PREPARED_STATEMENTS: "0" })).toEqual({ prepare: false });
|
||||
expect(databaseClientOptionsFromEnv({ DATABASE_PREPARED_STATEMENTS: "true" })).toEqual({ prepare: true });
|
||||
expect(databaseClientOptionsFromEnv({ DATABASE_PREPARED_STATEMENTS: "TRUE" })).toEqual({ prepare: true });
|
||||
});
|
||||
|
||||
it("parses pool and timeout settings", () => {
|
||||
expect(
|
||||
databaseClientOptionsFromEnv({
|
||||
DATABASE_POOL_MAX: "25",
|
||||
DATABASE_IDLE_TIMEOUT_SECONDS: "60",
|
||||
DATABASE_CONNECT_TIMEOUT_SECONDS: "10",
|
||||
}),
|
||||
).toEqual({ maxConnections: 25, idleTimeoutSeconds: 60, connectTimeoutSeconds: 10 });
|
||||
});
|
||||
|
||||
it("rejects malformed values instead of silently ignoring them", () => {
|
||||
expect(() => databaseClientOptionsFromEnv({ DATABASE_PREPARED_STATEMENTS: "maybe" })).toThrow(
|
||||
/DATABASE_PREPARED_STATEMENTS/,
|
||||
);
|
||||
expect(() => databaseClientOptionsFromEnv({ DATABASE_POOL_MAX: "0" })).toThrow(/DATABASE_POOL_MAX/);
|
||||
expect(() => databaseClientOptionsFromEnv({ DATABASE_POOL_MAX: "-3" })).toThrow(/DATABASE_POOL_MAX/);
|
||||
expect(() => databaseClientOptionsFromEnv({ DATABASE_CONNECT_TIMEOUT_SECONDS: "1.5" })).toThrow(
|
||||
/DATABASE_CONNECT_TIMEOUT_SECONDS/,
|
||||
);
|
||||
});
|
||||
|
||||
it("maps to postgres.js option names", () => {
|
||||
expect(
|
||||
postgresJsOptions({
|
||||
prepare: false,
|
||||
maxConnections: 25,
|
||||
idleTimeoutSeconds: 60,
|
||||
connectTimeoutSeconds: 10,
|
||||
}),
|
||||
).toEqual({ prepare: false, max: 25, idle_timeout: 60, connect_timeout: 10 });
|
||||
});
|
||||
});
|
||||
|
|
@ -45,8 +45,72 @@ export type MigrationState =
|
|||
reason: "no-migration-journal-empty-db" | "no-migration-journal-non-empty-db" | "pending-migrations";
|
||||
};
|
||||
|
||||
export function createDb(url: string) {
|
||||
const sql = postgres(url);
|
||||
export interface DatabaseClientOptions {
|
||||
/**
|
||||
* postgres.js `prepare`. Set false when connecting through a
|
||||
* transaction-mode pooler (pgbouncer / Neon `-pooler` endpoints /
|
||||
* Supabase Supavisor transaction ports) so the client does not rely on
|
||||
* session-scoped prepared statements. Defaults to the driver default
|
||||
* (enabled), preserving existing behavior on direct connections.
|
||||
*/
|
||||
prepare?: boolean;
|
||||
/** postgres.js `max` — connection pool size (driver default: 10). */
|
||||
maxConnections?: number;
|
||||
/** postgres.js `idle_timeout` in seconds (driver default: disabled). */
|
||||
idleTimeoutSeconds?: number;
|
||||
/** postgres.js `connect_timeout` in seconds (driver default: 30). */
|
||||
connectTimeoutSeconds?: number;
|
||||
}
|
||||
|
||||
function envBoolean(env: NodeJS.ProcessEnv, name: string): boolean | undefined {
|
||||
const value = env[name]?.trim().toLowerCase();
|
||||
if (value === undefined || value === "") return undefined;
|
||||
if (value === "true" || value === "1") return true;
|
||||
if (value === "false" || value === "0") return false;
|
||||
throw new Error(`${name} must be "true" or "false", got: ${env[name]}`);
|
||||
}
|
||||
|
||||
function envPositiveInteger(env: NodeJS.ProcessEnv, name: string): number | undefined {
|
||||
const value = env[name]?.trim();
|
||||
if (value === undefined || value === "") return undefined;
|
||||
if (!/^[1-9]\d*$/.test(value)) {
|
||||
throw new Error(`${name} must be a positive integer, got: ${env[name]}`);
|
||||
}
|
||||
return Number.parseInt(value, 10);
|
||||
}
|
||||
|
||||
/**
|
||||
* Database client tuning from the environment, so hosted deployments can
|
||||
* adapt to their connection topology (pooled endpoints, network latency)
|
||||
* without editing source. Every variable is optional; when unset the
|
||||
* driver defaults apply and behavior is identical to a bare
|
||||
* `postgres(url)` — self-hosted setups need none of these.
|
||||
*/
|
||||
export function databaseClientOptionsFromEnv(env: NodeJS.ProcessEnv = process.env): DatabaseClientOptions {
|
||||
const options: DatabaseClientOptions = {};
|
||||
const prepare = envBoolean(env, "DATABASE_PREPARED_STATEMENTS");
|
||||
if (prepare !== undefined) options.prepare = prepare;
|
||||
const maxConnections = envPositiveInteger(env, "DATABASE_POOL_MAX");
|
||||
if (maxConnections !== undefined) options.maxConnections = maxConnections;
|
||||
const idleTimeoutSeconds = envPositiveInteger(env, "DATABASE_IDLE_TIMEOUT_SECONDS");
|
||||
if (idleTimeoutSeconds !== undefined) options.idleTimeoutSeconds = idleTimeoutSeconds;
|
||||
const connectTimeoutSeconds = envPositiveInteger(env, "DATABASE_CONNECT_TIMEOUT_SECONDS");
|
||||
if (connectTimeoutSeconds !== undefined) options.connectTimeoutSeconds = connectTimeoutSeconds;
|
||||
return options;
|
||||
}
|
||||
|
||||
export function postgresJsOptions(options: DatabaseClientOptions): Record<string, unknown> {
|
||||
const driverOptions: Record<string, unknown> = {};
|
||||
if (options.prepare !== undefined) driverOptions.prepare = options.prepare;
|
||||
if (options.maxConnections !== undefined) driverOptions.max = options.maxConnections;
|
||||
if (options.idleTimeoutSeconds !== undefined) driverOptions.idle_timeout = options.idleTimeoutSeconds;
|
||||
if (options.connectTimeoutSeconds !== undefined) driverOptions.connect_timeout = options.connectTimeoutSeconds;
|
||||
return driverOptions;
|
||||
}
|
||||
|
||||
export function createDb(url: string, options?: DatabaseClientOptions) {
|
||||
const resolved = options ?? databaseClientOptionsFromEnv();
|
||||
const sql = postgres(url, postgresJsOptions(resolved));
|
||||
return drizzlePg(sql, { schema });
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -1023,8 +1023,10 @@ export function attentionService(db: Db, serviceOptions: AttentionServiceOptions
|
|||
if (options.all && !options.queue && !options.allowUnscopedAll) {
|
||||
throw badRequest("all requires a queue filter");
|
||||
}
|
||||
const prefix = await companyPrefix(db, companyId);
|
||||
const dismissals = await dismissalByKey(db, companyId, options.userId);
|
||||
const [prefix, dismissals] = await Promise.all([
|
||||
companyPrefix(db, companyId),
|
||||
dismissalByKey(db, companyId, options.userId),
|
||||
]);
|
||||
const includeDismissed = options.includeDismissed === true;
|
||||
const now = serviceOptions.now?.() ?? Date.now();
|
||||
const collected: AttentionItem[] = [];
|
||||
|
|
@ -1126,9 +1128,11 @@ export function attentionService(db: Db, serviceOptions: AttentionServiceOptions
|
|||
))
|
||||
.orderBy(desc(issueThreadInteractions.updatedAt), desc(issueThreadInteractions.id));
|
||||
const visibleInteractionRows = collapsePendingConfirmationsToNewest(interactionRows);
|
||||
const interactionIssueMap = await issueSummaryMap(db, companyId, visibleInteractionRows.map((row) => row.issueId));
|
||||
const interactionImageMap = await issueImageMap(db, companyId, visibleInteractionRows.map((row) => row.issueId));
|
||||
const interactionPlanDocumentMap = await planDocumentMap(db, companyId, visibleInteractionRows.map((row) => row.issueId));
|
||||
const [interactionIssueMap, interactionImageMap, interactionPlanDocumentMap] = await Promise.all([
|
||||
issueSummaryMap(db, companyId, visibleInteractionRows.map((row) => row.issueId)),
|
||||
issueImageMap(db, companyId, visibleInteractionRows.map((row) => row.issueId)),
|
||||
planDocumentMap(db, companyId, visibleInteractionRows.map((row) => row.issueId)),
|
||||
]);
|
||||
|
||||
for (const interaction of visibleInteractionRows) {
|
||||
const issue = interactionIssueMap.get(interaction.issueId) ?? null;
|
||||
|
|
@ -1194,16 +1198,18 @@ export function attentionService(db: Db, serviceOptions: AttentionServiceOptions
|
|||
const openDecisions = options.all
|
||||
? await openDecisionQuery
|
||||
: await openDecisionQuery.limit(openDecisionLimit);
|
||||
const decisionIssueMap = await issueSummaryMap(db, companyId, openDecisions.map((decision) => decision.originIssueId));
|
||||
// Bundle titles let the feed render a single "Agent proposed N decisions"
|
||||
// group header over sibling decisions (v1 still decides each independently).
|
||||
const bundleIds = [...new Set(openDecisions.map((decision) => decision.bundleId).filter((value): value is string => Boolean(value)))];
|
||||
const bundleTitleMap = new Map<string, string>();
|
||||
if (bundleIds.length > 0) {
|
||||
const bundleRows = await db.select({ id: decisionBundles.id, title: decisionBundles.title })
|
||||
.from(decisionBundles).where(and(eq(decisionBundles.companyId, companyId), inArray(decisionBundles.id, bundleIds)));
|
||||
for (const row of bundleRows) bundleTitleMap.set(row.id, row.title);
|
||||
}
|
||||
const [decisionIssueMap, bundleRows] = await Promise.all([
|
||||
issueSummaryMap(db, companyId, openDecisions.map((decision) => decision.originIssueId)),
|
||||
bundleIds.length > 0
|
||||
? db.select({ id: decisionBundles.id, title: decisionBundles.title })
|
||||
.from(decisionBundles).where(and(eq(decisionBundles.companyId, companyId), inArray(decisionBundles.id, bundleIds)))
|
||||
: Promise.resolve([]),
|
||||
]);
|
||||
for (const row of bundleRows) bundleTitleMap.set(row.id, row.title);
|
||||
for (const decision of openDecisions) {
|
||||
const issue = decisionIssueMap.get(decision.originIssueId) ?? null;
|
||||
add(createItem({
|
||||
|
|
@ -1301,12 +1307,14 @@ export function attentionService(db: Db, serviceOptions: AttentionServiceOptions
|
|||
inArray(issueRecoveryActions.ownerType, [...HUMAN_RECOVERY_OWNER_TYPES]),
|
||||
))
|
||||
.orderBy(desc(issueRecoveryActions.updatedAt), desc(issueRecoveryActions.id));
|
||||
const recoveryIssueMap = await issueSummaryMap(
|
||||
db,
|
||||
companyId,
|
||||
recoveryRows.flatMap((row) => [row.sourceIssueId, row.recoveryIssueId]),
|
||||
);
|
||||
const recoveryImageMap = await issueImageMap(db, companyId, recoveryRows.map((row) => row.sourceIssueId));
|
||||
const [recoveryIssueMap, recoveryImageMap] = await Promise.all([
|
||||
issueSummaryMap(
|
||||
db,
|
||||
companyId,
|
||||
recoveryRows.flatMap((row) => [row.sourceIssueId, row.recoveryIssueId]),
|
||||
),
|
||||
issueImageMap(db, companyId, recoveryRows.map((row) => row.sourceIssueId)),
|
||||
]);
|
||||
|
||||
for (const recovery of recoveryRows) {
|
||||
const sourceIssue = recoveryIssueMap.get(recovery.sourceIssueId) ?? null;
|
||||
|
|
@ -1378,9 +1386,11 @@ export function attentionService(db: Db, serviceOptions: AttentionServiceOptions
|
|||
notInArray(issues.status, [...PRODUCTIVITY_REVIEW_TERMINAL_STATUSES]),
|
||||
))
|
||||
.orderBy(desc(issues.updatedAt), desc(issues.id));
|
||||
const productivitySourceMap = await issueSummaryMap(db, companyId, productivityRows.map((row) => row.originId));
|
||||
const productivityReviewMap = await issueSummaryMap(db, companyId, productivityRows.map((row) => row.id));
|
||||
const productivityImageMap = await issueImageMap(db, companyId, productivityRows.map((row) => row.id));
|
||||
const [productivitySourceMap, productivityReviewMap, productivityImageMap] = await Promise.all([
|
||||
issueSummaryMap(db, companyId, productivityRows.map((row) => row.originId)),
|
||||
issueSummaryMap(db, companyId, productivityRows.map((row) => row.id)),
|
||||
issueImageMap(db, companyId, productivityRows.map((row) => row.id)),
|
||||
]);
|
||||
|
||||
for (const review of productivityRows) {
|
||||
const reviewIssue = productivityReviewMap.get(review.id);
|
||||
|
|
@ -1427,14 +1437,16 @@ export function attentionService(db: Db, serviceOptions: AttentionServiceOptions
|
|||
const terminalBlockerIssueIds = typedBlockedIssues
|
||||
.map((issue) => issue.blockerAttention?.terminalBlockerIssueId)
|
||||
.filter((issueId): issueId is string => Boolean(issueId));
|
||||
const blockedIssueSummaries = await issueSummaryMap(db, companyId, blockedIssues.map((issue) => issue.id));
|
||||
const terminalBlockerSummaries = await issueSummaryMap(db, companyId, terminalBlockerIssueIds);
|
||||
const blockerImageMap = await issueImageMap(
|
||||
db,
|
||||
companyId,
|
||||
[...blockedIssues.map((issue) => issue.id), ...terminalBlockerIssueIds],
|
||||
);
|
||||
const blockingIssues = await blockingIssueMap(db, companyId, blockedIssues.map((issue) => issue.id));
|
||||
const [blockedIssueSummaries, terminalBlockerSummaries, blockerImageMap, blockingIssues] = await Promise.all([
|
||||
issueSummaryMap(db, companyId, blockedIssues.map((issue) => issue.id)),
|
||||
issueSummaryMap(db, companyId, terminalBlockerIssueIds),
|
||||
issueImageMap(
|
||||
db,
|
||||
companyId,
|
||||
[...blockedIssues.map((issue) => issue.id), ...terminalBlockerIssueIds],
|
||||
),
|
||||
blockingIssueMap(db, companyId, blockedIssues.map((issue) => issue.id)),
|
||||
]);
|
||||
const terminalCandidates = new Map<string, {
|
||||
issue: BlockedAttentionIssue;
|
||||
issueSummary: IssueSummaryRow | null;
|
||||
|
|
@ -1560,8 +1572,10 @@ export function attentionService(db: Db, serviceOptions: AttentionServiceOptions
|
|||
eq(approvals.status, "pending"),
|
||||
));
|
||||
const pendingApprovalByIssueId = new Map(pendingReviewApprovalRows.map((row) => [row.issueId, row.approvalId]));
|
||||
const reviewIssueMap = await issueSummaryMap(db, companyId, reviewIssueIds);
|
||||
const reviewImageMap = await issueImageMap(db, companyId, reviewIssueIds);
|
||||
const [reviewIssueMap, reviewImageMap] = await Promise.all([
|
||||
issueSummaryMap(db, companyId, reviewIssueIds),
|
||||
issueImageMap(db, companyId, reviewIssueIds),
|
||||
]);
|
||||
|
||||
for (const review of reviewRows) {
|
||||
const state = parseIssueExecutionState(review.executionState);
|
||||
|
|
@ -1634,37 +1648,39 @@ export function attentionService(db: Db, serviceOptions: AttentionServiceOptions
|
|||
}
|
||||
const failedRows = [...latestExhaustedByRunId.values()];
|
||||
const failedIssueIds = failedRows.map((row) => readRunIssueId(row.contextSnapshot));
|
||||
const failedIssueMap = await issueSummaryMap(
|
||||
db,
|
||||
companyId,
|
||||
failedIssueIds,
|
||||
);
|
||||
const failedImageMap = await issueImageMap(db, companyId, failedIssueIds);
|
||||
const failedAgentIds = [...new Set(failedRows.map((row) => row.agentId))];
|
||||
const oldestFailedRunCreatedAt = failedRows.reduce<Date | null>((oldest, row) => {
|
||||
if (!oldest || row.createdAt < oldest) return row.createdAt;
|
||||
return oldest;
|
||||
}, null);
|
||||
const [failedIssueMap, failedImageMap, newerRuns] = await Promise.all([
|
||||
issueSummaryMap(
|
||||
db,
|
||||
companyId,
|
||||
failedIssueIds,
|
||||
),
|
||||
issueImageMap(db, companyId, failedIssueIds),
|
||||
oldestFailedRunCreatedAt && failedAgentIds.length > 0
|
||||
? db
|
||||
.select({
|
||||
agentId: heartbeatRuns.agentId,
|
||||
createdAt: heartbeatRuns.createdAt,
|
||||
contextSnapshot: heartbeatRuns.contextSnapshot,
|
||||
})
|
||||
.from(heartbeatRuns)
|
||||
.where(and(
|
||||
eq(heartbeatRuns.companyId, companyId),
|
||||
inArray(heartbeatRuns.agentId, failedAgentIds),
|
||||
gt(heartbeatRuns.createdAt, oldestFailedRunCreatedAt),
|
||||
))
|
||||
: Promise.resolve([]),
|
||||
]);
|
||||
const latestRunCreatedAtByKey = new Map<string, Date>();
|
||||
if (oldestFailedRunCreatedAt && failedAgentIds.length > 0) {
|
||||
const newerRuns = await db
|
||||
.select({
|
||||
agentId: heartbeatRuns.agentId,
|
||||
createdAt: heartbeatRuns.createdAt,
|
||||
contextSnapshot: heartbeatRuns.contextSnapshot,
|
||||
})
|
||||
.from(heartbeatRuns)
|
||||
.where(and(
|
||||
eq(heartbeatRuns.companyId, companyId),
|
||||
inArray(heartbeatRuns.agentId, failedAgentIds),
|
||||
gt(heartbeatRuns.createdAt, oldestFailedRunCreatedAt),
|
||||
));
|
||||
for (const newerRun of newerRuns) {
|
||||
const newerRunKey = `${newerRun.agentId}:${readRunIssueId(newerRun.contextSnapshot) ?? ""}`;
|
||||
const latestCreatedAt = latestRunCreatedAtByKey.get(newerRunKey);
|
||||
if (!latestCreatedAt || newerRun.createdAt > latestCreatedAt) {
|
||||
latestRunCreatedAtByKey.set(newerRunKey, newerRun.createdAt);
|
||||
}
|
||||
for (const newerRun of newerRuns) {
|
||||
const newerRunKey = `${newerRun.agentId}:${readRunIssueId(newerRun.contextSnapshot) ?? ""}`;
|
||||
const latestCreatedAt = latestRunCreatedAtByKey.get(newerRunKey);
|
||||
if (!latestCreatedAt || newerRun.createdAt > latestCreatedAt) {
|
||||
latestRunCreatedAtByKey.set(newerRunKey, newerRun.createdAt);
|
||||
}
|
||||
}
|
||||
for (const run of failedRows) {
|
||||
|
|
|
|||
Loading…
Reference in New Issue