mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-30 13:20:54 +01:00
fix(server): index the finite voice report window
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
250fca6cff
commit
94c60683b8
9 files changed
+234
-12
No files matched your search
@@ -3,7 +3,8 @@
|
|||||||
Voice report reconciliation uses the shared delivery coordinator and scheduler.
|
Voice report reconciliation uses the shared delivery coordinator and scheduler.
|
||||||
It no longer runs from the chat maintenance interval. Existing
|
It no longer runs from the chat maintenance interval. Existing
|
||||||
`chat_voice_sessions` and `chat_voice_reports` rows own its durable state; this
|
`chat_voice_sessions` and `chat_voice_reports` rows own its durable state; this
|
||||||
change adds no queue table, schema, provider endpoint, or independent timer.
|
change adds no queue table, provider endpoint, or independent timer. A concurrent
|
||||||
|
partial index supports bounded reconstruction of the existing report window.
|
||||||
|
|
||||||
## Trigger and deadline
|
## Trigger and deadline
|
||||||
|
|
||||||
@@ -20,7 +21,9 @@ result does the same. Healthy active-session observations do not activate report
|
|||||||
work. Nested transactions wait for the outer commit; rollback does not publish
|
work. Nested transactions wait for the outer commit; rollback does not publish
|
||||||
new work. The existing work-signals protocol owns uncertain outcomes.
|
new work. The existing work-signals protocol owns uncertain outcomes.
|
||||||
|
|
||||||
Startup reconstructs eligible sessions in pages. Committed hints reload their
|
Startup reconstructs eligible sessions in pages ordered by `(ended_at, id)`.
|
||||||
|
A partial index restricts the search to terminal sessions with provider IDs and
|
||||||
|
the current 24-hour window. The cursor retains PostgreSQL timestamp precision. Committed hints reload their
|
||||||
session, endpoint, connection, or company scope. Each cached entry carries the
|
session, endpoint, connection, or company scope. Each cached entry carries the
|
||||||
saved report deadline or the first millisecond after the refresh window ends,
|
saved report deadline or the first millisecond after the refresh window ends,
|
||||||
whichever comes first. Submillisecond PostgreSQL timestamps round in the safe
|
whichever comes first. Submillisecond PostgreSQL timestamps round in the safe
|
||||||
@@ -57,9 +60,12 @@ work. Restart reconstructs the persisted retry instead of immediately repeating
|
|||||||
the provider GET. The explicit one-shot voice reconciliation method remains for
|
the provider GET. The explicit one-shot voice reconciliation method remains for
|
||||||
diagnostics; production maintenance does not call it periodically.
|
diagnostics; production maintenance does not call it periodically.
|
||||||
|
|
||||||
Signals cover supported writes through the owning database client in this
|
The supported deployment has one active server process writing voice session
|
||||||
process. Independent SQL writers and another process need their own notification
|
and report state through its owning database client. Active-active servers and
|
||||||
contract. This is worker quiet, not a claim of physical host sleep. Cloud alarm
|
out-of-band SQL writers are not supported by this worker. Process-local signals
|
||||||
|
do not reach another process; such a deployment needs a durable cross-process
|
||||||
|
wake contract before removing its reconciliation backstop. This PR neither adds
|
||||||
|
that transport nor asserts that startup reconstruction covers concurrent writers. This is worker quiet, not a claim of physical host sleep. Cloud alarm
|
||||||
coverage and conservative sleep eligibility are unchanged; a local deadline
|
coverage and conservative sleep eligibility are unchanged; a local deadline
|
||||||
cannot wake stopped compute.
|
cannot wake stopped compute.
|
||||||
|
|
||||||
|
|||||||
@@ -150,3 +150,45 @@ After both merges, all 138 full voice/app cases pass on a fresh migrated databas
|
|||||||
those worker implementations are unchanged by the parent repairs. Remote checks
|
those worker implementations are unchanged by the parent repairs. Remote checks
|
||||||
and reviews are required for the final pushed head. Parent #17580's company
|
and reviews are required for the final pushed head. Parent #17580's company
|
||||||
deletion finding remains unresolved; denial latency finding is resolved.
|
deletion finding remains unresolved; denial latency finding is resolved.
|
||||||
|
|
||||||
|
|
||||||
|
## Paperclip review: report window index and writer scope
|
||||||
|
|
||||||
|
Review findings 4370594150 and 4370594357 identified deployment-contract clarity
|
||||||
|
and expensive report reconstruction. The supported writer model is now explicit:
|
||||||
|
one active server process owns voice writes through its DB client. Active-active
|
||||||
|
servers and independent SQL mutation are unsupported without a separate durable
|
||||||
|
wake contract. This is a documented scope restriction, not enforcement added by
|
||||||
|
this PR, and not a claim that startup recovers concurrent foreign writers.
|
||||||
|
|
||||||
|
The query finding reproduces on fresh PostgreSQL with 20,001 old sessions and
|
||||||
|
205 eligible sessions: an actual reconstruction query filters 20,001 historical
|
||||||
|
rows. The repair adds partial `(ended_at, id)` index 0389 and uses that order and
|
||||||
|
keyset cursor. The cursor retains PostgreSQL text timestamp precision, so sessions
|
||||||
|
100 microseconds apart do not repeat or disappear across a page. All 14 report
|
||||||
|
lifecycle cases pass after the migration; the query-plan regression checks all
|
||||||
|
three pages and each stays below 1,000 rows removed by filtering. Report scope,
|
||||||
|
24-hour policy, and provider behavior are unchanged.
|
||||||
|
|
||||||
|
The generated migration is concurrent and idempotent and registered with the
|
||||||
|
application's top-level migration runner. It applies and a second application is
|
||||||
|
a no-op. Migration numbering/safety checks pass. The generator retained exactly
|
||||||
|
five snapshots, 0385 through 0389, pruning 0384 per repository policy.
|
||||||
|
|
||||||
|
Retained failures: the first probe reused a database with two unrelated eligible
|
||||||
|
fixture sessions and failed its exact fixture count. It is not performance proof.
|
||||||
|
A fresh database reproduced the 20,001-row filtering problem. The first migration
|
||||||
|
invocation used unsupported top-level await in tsx eval and did not apply; its
|
||||||
|
subsequent test still failed without the index. The corrected chained invocation
|
||||||
|
applied/replayed successfully and all 14 cases passed. No assertion was relaxed.
|
||||||
|
Logs: /tmp/sleep25-report-index-before.log (fixture contamination),
|
||||||
|
/tmp/sleep25-report-index-clean-before.log (cost reproduction),
|
||||||
|
/tmp/sleep25-report-index-apply.log (invalid invocation),
|
||||||
|
/tmp/sleep25-report-index-after.log (unapplied index),
|
||||||
|
/tmp/sleep25-report-index-apply-fixed.log (application/replay),
|
||||||
|
/tmp/sleep25-report-index-fixed-tests.log (14 pass).
|
||||||
|
Workspace typecheck and build pass. Fresh full voice integration and app wiring
|
||||||
|
pass all 138 cases after the index repair. Logs:
|
||||||
|
/tmp/sleep25-report-review-{types,build,acceptance}.log.
|
||||||
|
Publish and propagate through Teams #17590/#17592 after acceptance. Fresh remote
|
||||||
|
review/checks and all prior full-stack requirements remain open.
|
||||||
@@ -514,6 +514,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([
|
|||||||
"0386_omniscient_luke_cage.sql",
|
"0386_omniscient_luke_cage.sql",
|
||||||
"0387_huge_skrulls.sql",
|
"0387_huge_skrulls.sql",
|
||||||
"0388_lying_trish_tilby.sql",
|
"0388_lying_trish_tilby.sql",
|
||||||
|
"0389_workable_spencer_smythe.sql",
|
||||||
"0380_funny_adam_warlock.sql",
|
"0380_funny_adam_warlock.sql",
|
||||||
]);
|
]);
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1 @@
|
|||||||
|
CREATE INDEX CONCURRENTLY IF NOT EXISTS "chat_voice_sessions_report_window_idx" ON "chat_voice_sessions" USING btree ("ended_at","id") WHERE "chat_voice_sessions"."state" in ('ended', 'failed', 'expired') and "chat_voice_sessions"."provider_session_id" is not null;
|
||||||
+124
-2
@@ -1,6 +1,6 @@
|
|||||||
{
|
{
|
||||||
"id": "45d3a35e-2d6d-4fc1-8445-c35eef1ea173",
|
"id": "8bd7f279-1063-4ba6-83bb-b85cba334eb3",
|
||||||
"prevId": "75e7a97b-132d-4326-b683-003c1a2b7048",
|
"prevId": "f83c3343-8665-48bc-919b-bc9fe7513b87",
|
||||||
"version": "7",
|
"version": "7",
|
||||||
"dialect": "postgresql",
|
"dialect": "postgresql",
|
||||||
"tables": {
|
"tables": {
|
||||||
@@ -5721,6 +5721,28 @@
|
|||||||
}
|
}
|
||||||
},
|
},
|
||||||
"indexes": {
|
"indexes": {
|
||||||
|
"approvals_pending_retention_idx": {
|
||||||
|
"name": "approvals_pending_retention_idx",
|
||||||
|
"columns": [
|
||||||
|
{
|
||||||
|
"expression": "updated_at",
|
||||||
|
"isExpression": false,
|
||||||
|
"asc": true,
|
||||||
|
"nulls": "last"
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"expression": "id",
|
||||||
|
"isExpression": false,
|
||||||
|
"asc": true,
|
||||||
|
"nulls": "last"
|
||||||
|
}
|
||||||
|
],
|
||||||
|
"isUnique": false,
|
||||||
|
"where": "\"approvals\".\"status\" = 'pending'",
|
||||||
|
"concurrently": false,
|
||||||
|
"method": "btree",
|
||||||
|
"with": {}
|
||||||
|
},
|
||||||
"approvals_company_status_type_idx": {
|
"approvals_company_status_type_idx": {
|
||||||
"name": "approvals_company_status_type_idx",
|
"name": "approvals_company_status_type_idx",
|
||||||
"columns": [
|
"columns": [
|
||||||
@@ -14138,6 +14160,24 @@
|
|||||||
"primaryKey": false,
|
"primaryKey": false,
|
||||||
"notNull": false
|
"notNull": false
|
||||||
},
|
},
|
||||||
|
"provider_account_id": {
|
||||||
|
"name": "provider_account_id",
|
||||||
|
"type": "text",
|
||||||
|
"primaryKey": false,
|
||||||
|
"notNull": false
|
||||||
|
},
|
||||||
|
"cleanup_confirmed_at": {
|
||||||
|
"name": "cleanup_confirmed_at",
|
||||||
|
"type": "timestamp with time zone",
|
||||||
|
"primaryKey": false,
|
||||||
|
"notNull": false
|
||||||
|
},
|
||||||
|
"cleanup_next_attempt_at": {
|
||||||
|
"name": "cleanup_next_attempt_at",
|
||||||
|
"type": "timestamp with time zone",
|
||||||
|
"primaryKey": false,
|
||||||
|
"notNull": false
|
||||||
|
},
|
||||||
"expires_at": {
|
"expires_at": {
|
||||||
"name": "expires_at",
|
"name": "expires_at",
|
||||||
"type": "timestamp with time zone",
|
"type": "timestamp with time zone",
|
||||||
@@ -14160,6 +14200,22 @@
|
|||||||
}
|
}
|
||||||
},
|
},
|
||||||
"indexes": {
|
"indexes": {
|
||||||
|
"chat_voice_inbound_cleanup_idx": {
|
||||||
|
"name": "chat_voice_inbound_cleanup_idx",
|
||||||
|
"columns": [
|
||||||
|
{
|
||||||
|
"expression": "id",
|
||||||
|
"isExpression": false,
|
||||||
|
"asc": true,
|
||||||
|
"nulls": "last"
|
||||||
|
}
|
||||||
|
],
|
||||||
|
"isUnique": false,
|
||||||
|
"where": "\"chat_voice_inbound_calls\".\"cleanup_confirmed_at\" is null",
|
||||||
|
"concurrently": false,
|
||||||
|
"method": "btree",
|
||||||
|
"with": {}
|
||||||
|
},
|
||||||
"chat_voice_inbound_expiry_idx": {
|
"chat_voice_inbound_expiry_idx": {
|
||||||
"name": "chat_voice_inbound_expiry_idx",
|
"name": "chat_voice_inbound_expiry_idx",
|
||||||
"columns": [
|
"columns": [
|
||||||
@@ -14912,6 +14968,28 @@
|
|||||||
"concurrently": false,
|
"concurrently": false,
|
||||||
"method": "btree",
|
"method": "btree",
|
||||||
"with": {}
|
"with": {}
|
||||||
|
},
|
||||||
|
"chat_voice_sessions_report_window_idx": {
|
||||||
|
"name": "chat_voice_sessions_report_window_idx",
|
||||||
|
"columns": [
|
||||||
|
{
|
||||||
|
"expression": "ended_at",
|
||||||
|
"isExpression": false,
|
||||||
|
"asc": true,
|
||||||
|
"nulls": "last"
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"expression": "id",
|
||||||
|
"isExpression": false,
|
||||||
|
"asc": true,
|
||||||
|
"nulls": "last"
|
||||||
|
}
|
||||||
|
],
|
||||||
|
"isUnique": false,
|
||||||
|
"where": "\"chat_voice_sessions\".\"state\" in ('ended', 'failed', 'expired') and \"chat_voice_sessions\".\"provider_session_id\" is not null",
|
||||||
|
"concurrently": false,
|
||||||
|
"method": "btree",
|
||||||
|
"with": {}
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
"foreignKeys": {
|
"foreignKeys": {
|
||||||
@@ -32079,6 +32157,28 @@
|
|||||||
"method": "btree",
|
"method": "btree",
|
||||||
"with": {}
|
"with": {}
|
||||||
},
|
},
|
||||||
|
"grok_dispatch_prepared_idx": {
|
||||||
|
"name": "grok_dispatch_prepared_idx",
|
||||||
|
"columns": [
|
||||||
|
{
|
||||||
|
"expression": "created_at",
|
||||||
|
"isExpression": false,
|
||||||
|
"asc": true,
|
||||||
|
"nulls": "last"
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"expression": "id",
|
||||||
|
"isExpression": false,
|
||||||
|
"asc": true,
|
||||||
|
"nulls": "last"
|
||||||
|
}
|
||||||
|
],
|
||||||
|
"isUnique": false,
|
||||||
|
"where": "\"grok_bot_dispatches\".\"status\" = 'prepared'",
|
||||||
|
"concurrently": false,
|
||||||
|
"method": "btree",
|
||||||
|
"with": {}
|
||||||
|
},
|
||||||
"grok_dispatch_company_status_idx": {
|
"grok_dispatch_company_status_idx": {
|
||||||
"name": "grok_dispatch_company_status_idx",
|
"name": "grok_dispatch_company_status_idx",
|
||||||
"columns": [
|
"columns": [
|
||||||
@@ -41873,6 +41973,28 @@
|
|||||||
}
|
}
|
||||||
},
|
},
|
||||||
"indexes": {
|
"indexes": {
|
||||||
|
"join_requests_pending_retention_idx": {
|
||||||
|
"name": "join_requests_pending_retention_idx",
|
||||||
|
"columns": [
|
||||||
|
{
|
||||||
|
"expression": "updated_at",
|
||||||
|
"isExpression": false,
|
||||||
|
"asc": true,
|
||||||
|
"nulls": "last"
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"expression": "id",
|
||||||
|
"isExpression": false,
|
||||||
|
"asc": true,
|
||||||
|
"nulls": "last"
|
||||||
|
}
|
||||||
|
],
|
||||||
|
"isUnique": false,
|
||||||
|
"where": "\"join_requests\".\"status\" = 'pending_approval'",
|
||||||
|
"concurrently": false,
|
||||||
|
"method": "btree",
|
||||||
|
"with": {}
|
||||||
|
},
|
||||||
"join_requests_invite_unique_idx": {
|
"join_requests_invite_unique_idx": {
|
||||||
"name": "join_requests_invite_unique_idx",
|
"name": "join_requests_invite_unique_idx",
|
||||||
"columns": [
|
"columns": [
|
||||||
+7
@@ -2710,6 +2710,13 @@
|
|||||||
"when": 1792921047529,
|
"when": 1792921047529,
|
||||||
"tag": "0388_lying_trish_tilby",
|
"tag": "0388_lying_trish_tilby",
|
||||||
"breakpoints": true
|
"breakpoints": true
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"idx": 389,
|
||||||
|
"version": "7",
|
||||||
|
"when": 1792926867183,
|
||||||
|
"tag": "0389_workable_spencer_smythe",
|
||||||
|
"breakpoints": true
|
||||||
}
|
}
|
||||||
]
|
]
|
||||||
}
|
}
|
||||||
@@ -36,6 +36,8 @@ export const chatVoiceSessions = pgTable("chat_voice_sessions", {
|
|||||||
uniqueIndex("chat_voice_sessions_request_uq").on(t.companyId, t.callerId, t.idempotencyKey),
|
uniqueIndex("chat_voice_sessions_request_uq").on(t.companyId, t.callerId, t.idempotencyKey),
|
||||||
uniqueIndex("chat_voice_sessions_live_caller_task_uq").on(t.companyId, t.callerId, t.issueId).where(sql`${t.state} not in ('ended', 'failed', 'expired')`),
|
uniqueIndex("chat_voice_sessions_live_caller_task_uq").on(t.companyId, t.callerId, t.issueId).where(sql`${t.state} not in ('ended', 'failed', 'expired')`),
|
||||||
index("chat_voice_sessions_expiry_idx").on(t.state, t.expiresAt),
|
index("chat_voice_sessions_expiry_idx").on(t.state, t.expiresAt),
|
||||||
|
index("chat_voice_sessions_report_window_idx").on(t.endedAt, t.id)
|
||||||
|
.where(sql`${t.state} in ('ended', 'failed', 'expired') and ${t.providerSessionId} is not null`),
|
||||||
foreignKey({ columns: [t.companyId, t.endpointId], foreignColumns: [chatEndpoints.companyId, chatEndpoints.id], name: "chat_voice_sessions_endpoint_fk" }),
|
foreignKey({ columns: [t.companyId, t.endpointId], foreignColumns: [chatEndpoints.companyId, chatEndpoints.id], name: "chat_voice_sessions_endpoint_fk" }),
|
||||||
foreignKey({ columns: [t.companyId, t.conversationId], foreignColumns: [chatConversations.companyId, chatConversations.id], name: "chat_voice_sessions_conversation_fk" }),
|
foreignKey({ columns: [t.companyId, t.conversationId], foreignColumns: [chatConversations.companyId, chatConversations.id], name: "chat_voice_sessions_conversation_fk" }),
|
||||||
foreignKey({ columns: [t.companyId, t.issueId], foreignColumns: [issues.companyId, issues.id], name: "chat_voice_sessions_issue_fk" }),
|
foreignKey({ columns: [t.companyId, t.issueId], foreignColumns: [issues.companyId, issues.id], name: "chat_voice_sessions_issue_fk" }),
|
||||||
|
|||||||
@@ -195,6 +195,44 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
|
|||||||
await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(fetchReport).toHaveBeenCalledTimes(105); }, { timeout: 5000 });
|
await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(fetchReport).toHaveBeenCalledTimes(105); }, { timeout: 5000 });
|
||||||
expect((await report(broken.id)).attempts).toBe(1); await flush(); await quiet();
|
expect((await report(broken.id)).attempts).toBe(1); await flush(); await quiet();
|
||||||
});
|
});
|
||||||
|
it("reconstructs recent reports without scanning old history and preserves submillisecond page boundaries", async () => {
|
||||||
|
const seed = await session({ endedAt: new Date(Date.now() - 2 * day) });
|
||||||
|
await db.execute(sql`insert into chat_voice_sessions
|
||||||
|
(company_id, endpoint_id, conversation_id, issue_id, assigned_agent_id, caller_id,
|
||||||
|
caller_authority, mode, state, generation, credential_fingerprint, provider_session_id,
|
||||||
|
tool_token_hash, idempotency_key, request_fingerprint, expires_at, ended_at, created_at)
|
||||||
|
select company_id, endpoint_id, conversation_id, issue_id, assigned_agent_id, gen_random_uuid()::text,
|
||||||
|
caller_authority, mode, state, generation, credential_fingerprint, gen_random_uuid()::text,
|
||||||
|
tool_token_hash, gen_random_uuid()::text, request_fingerprint, expires_at, ended_at, created_at
|
||||||
|
from chat_voice_sessions cross join generate_series(1, 20000) where id = ${seed.id}`);
|
||||||
|
const stamp = new Date(Date.now() - 60_000).toISOString();
|
||||||
|
const expected: string[] = [];
|
||||||
|
for (let i = 0; i < 205; i++) {
|
||||||
|
const row = await session(); expected.push(row.id);
|
||||||
|
await db.execute(sql`update chat_voice_sessions set ended_at = ${stamp}::timestamptz + (${i % 3} * interval '0.0001 second') where id = ${row.id}`);
|
||||||
|
}
|
||||||
|
await db.execute(sql`analyze chat_voice_sessions`);
|
||||||
|
const queries: Array<{ text: string; parameters: unknown[] }> = [];
|
||||||
|
const original = db.$client.unsafe.bind(db.$client);
|
||||||
|
const capture = vi.spyOn(db.$client, "unsafe").mockImplementation(((text: string, parameters: unknown[] = [], ...rest: unknown[]) => {
|
||||||
|
if (text.includes('from "chat_voice_sessions"') && text.includes('left join "chat_voice_reports"')) queries.push({ text, parameters });
|
||||||
|
return original(text, parameters as never, ...rest as []);
|
||||||
|
}) as typeof db.$client.unsafe);
|
||||||
|
let loaded: Awaited<ReturnType<ReturnType<typeof voiceCallReports>["load"]>>;
|
||||||
|
try { loaded = await voiceCallReports(db, fetchReport).load(); }
|
||||||
|
finally { capture.mockRestore(); }
|
||||||
|
expect(loaded!.map(row => row.id).sort()).toEqual(expected.sort());
|
||||||
|
expect(queries).toHaveLength(3);
|
||||||
|
function removed(node: Record<string, unknown>): number {
|
||||||
|
return Math.max(Number(node["Rows Removed by Filter"] ?? 0),
|
||||||
|
...(Array.isArray(node.Plans) ? node.Plans.map(child => removed(child)) : []));
|
||||||
|
}
|
||||||
|
for (const query of queries) {
|
||||||
|
const [row] = await db.$client.unsafe(`explain (analyze, buffers, format json) ${query.text}`, query.parameters as never);
|
||||||
|
expect(removed(row["QUERY PLAN"][0].Plan)).toBeLessThan(1000);
|
||||||
|
}
|
||||||
|
expect(fetchReport).not.toHaveBeenCalled();
|
||||||
|
}, 30_000);
|
||||||
it("joins an in-flight report at shutdown and resumes its saved retry", async () => {
|
it("joins an in-flight report at shutdown and resumes its saved retry", async () => {
|
||||||
const call = await session(); let release!: (value: SpekoCallReport) => void;
|
const call = await session(); let release!: (value: SpekoCallReport) => void;
|
||||||
fetchReport.mockImplementationOnce(() => new Promise(resolve => { release = resolve; }));
|
fetchReport.mockImplementationOnce(() => new Promise(resolve => { release = resolve; }));
|
||||||
|
|||||||
@@ -1,4 +1,4 @@
|
|||||||
import { and, eq, gt, gte, inArray, isNotNull, lte, ne, or, sql } from "drizzle-orm";
|
import { and, eq, gte, inArray, isNotNull, lte, ne, or, sql } from "drizzle-orm";
|
||||||
import { activityLog, chatEndpoints, chatVoiceReports, chatVoiceSessions, issues, type Db } from "@paperclipai/db";
|
import { activityLog, chatEndpoints, chatVoiceReports, chatVoiceSessions, issues, type Db } from "@paperclipai/db";
|
||||||
import { logger } from "../../middleware/logger.js";
|
import { logger } from "../../middleware/logger.js";
|
||||||
import { documentService } from "../documents.js";
|
import { documentService } from "../documents.js";
|
||||||
@@ -21,11 +21,12 @@ const eligible = (now: number) => and(inArray(chatVoiceSessions.state, ["ended",
|
|||||||
* state, not a promise of immutable provider data. Keep that finite obligation. */
|
* state, not a promise of immutable provider data. Keep that finite obligation. */
|
||||||
export function voiceCallReports(db: Db, fetchReport: (session: VoiceSessionRow) => Promise<Report>) {
|
export function voiceCallReports(db: Db, fetchReport: (session: VoiceSessionRow) => Promise<Report>) {
|
||||||
async function load(scope?: VoiceRecoveryScope, signal?: AbortSignal): Promise<Entry[]> {
|
async function load(scope?: VoiceRecoveryScope, signal?: AbortSignal): Promise<Entry[]> {
|
||||||
const entries: Entry[] = []; let cursor: string | undefined;
|
const entries: Entry[] = []; let cursor: { id: string; endedAt: string } | undefined;
|
||||||
for (;;) {
|
for (;;) {
|
||||||
signal?.throwIfAborted();
|
signal?.throwIfAborted();
|
||||||
const rows = await db.select({ id: chatVoiceSessions.id, companyId: chatVoiceSessions.companyId,
|
const rows = await db.select({ id: chatVoiceSessions.id, companyId: chatVoiceSessions.companyId,
|
||||||
endpointId: chatVoiceSessions.endpointId, connectionId: chatEndpoints.connectionId,
|
endpointId: chatVoiceSessions.endpointId, connectionId: chatEndpoints.connectionId,
|
||||||
|
endedAt: sql<string>`${chatVoiceSessions.endedAt}::text`,
|
||||||
dueAt: sql<string>`least(coalesce(ceil(extract(epoch from ${chatVoiceReports.nextCheckAt}) * 1000), 0), floor(extract(epoch from ${chatVoiceSessions.endedAt}) * 1000) + ${WINDOW_MS} + 1)`,
|
dueAt: sql<string>`least(coalesce(ceil(extract(epoch from ${chatVoiceReports.nextCheckAt}) * 1000), 0), floor(extract(epoch from ${chatVoiceSessions.endedAt}) * 1000) + ${WINDOW_MS} + 1)`,
|
||||||
}).from(chatVoiceSessions)
|
}).from(chatVoiceSessions)
|
||||||
.innerJoin(chatEndpoints, and(eq(chatEndpoints.id, chatVoiceSessions.endpointId), eq(chatEndpoints.companyId, chatVoiceSessions.companyId)))
|
.innerJoin(chatEndpoints, and(eq(chatEndpoints.id, chatVoiceSessions.endpointId), eq(chatEndpoints.companyId, chatVoiceSessions.companyId)))
|
||||||
@@ -35,10 +36,12 @@ export function voiceCallReports(db: Db, fetchReport: (session: VoiceSessionRow)
|
|||||||
scope?.kind === "endpoint" ? eq(chatVoiceSessions.endpointId, scope.id) : undefined,
|
scope?.kind === "endpoint" ? eq(chatVoiceSessions.endpointId, scope.id) : undefined,
|
||||||
scope?.kind === "connection" ? eq(chatEndpoints.connectionId, scope.id) : undefined,
|
scope?.kind === "connection" ? eq(chatEndpoints.connectionId, scope.id) : undefined,
|
||||||
scope?.kind === "company" ? eq(chatVoiceSessions.companyId, scope.id) : undefined,
|
scope?.kind === "company" ? eq(chatVoiceSessions.companyId, scope.id) : undefined,
|
||||||
cursor ? gt(chatVoiceSessions.id, cursor) : undefined)).orderBy(chatVoiceSessions.id).limit(100);
|
cursor ? sql`(${chatVoiceSessions.endedAt}, ${chatVoiceSessions.id}) > (${cursor.endedAt}::timestamptz, ${cursor.id}::uuid)` : undefined))
|
||||||
entries.push(...rows.map(row => ({ ...row, dueAt: Number(row.dueAt) })));
|
.orderBy(chatVoiceSessions.endedAt, chatVoiceSessions.id).limit(100);
|
||||||
|
entries.push(...rows.map(({ endedAt: _cursor, ...row }) => ({ ...row, dueAt: Number(row.dueAt) })));
|
||||||
if (rows.length < 100) return entries;
|
if (rows.length < 100) return entries;
|
||||||
cursor = rows[rows.length - 1].id;
|
// Keep PostgreSQL precision; JS Date truncation can repeat a page.
|
||||||
|
cursor = rows[rows.length - 1];
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
async function projectTranscript(session: VoiceSessionRow) {
|
async function projectTranscript(session: VoiceSessionRow) {
|
||||||
|
|||||||
Reference in new issue
Block a user