From 6ab497fa12b8a816eae0636906729a38c3700422 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 8 Sep 2026 17:16:54 +0000 Subject: [PATCH] Select the filings to index in the database, not by sifting headers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `limit` bounded the number of filings INDEXED, and an already-indexed one was skipped without counting toward it. So `--limit 5` against a corpus that is already indexed read every converted document and probed the knowledge base once per row before concluding there was nothing to do — and `ask` builds the index implicitly, so that ran on every question. `selectDocumentsToIndex` makes the set difference an anti-join, the shape `selectFilingsToConvert` already uses for the same question about conversion: `LEFT JOIN kb_document ... WHERE doc_id IS NULL ... LIMIT ?`, with the scope filters (`since` included, which used to be applied after the read) pushed into the query. It asks for one row more than the limit, so `truncated` is something the run observed rather than inferred from having filled the limit exactly. The knowledge-base tables are built lazily by the first command that opens the index, and this runs before that, so a database with converted filings and no index has no table to join against: the join is dropped and every document needs work, which is the answer rather than an error. Documents with no sections are excluded. They embed nothing, so they never enter the knowledge base and would be re-selected on every run — and under a selection-time limit they would spend it on work that cannot happen. `skipped` stays exact through one COUNT over the same join, replacing the unfiltered `count` this task took for its progress denominator. That denominator was wrong anyway: it measured a run indexing three filings out of three hundred candidates at one percent, where the honest denominator is the work selected. The repository fallback consults no knowledge base, because it is reached only where there cannot be one: a non-durable document repository is invisible to `getDb()`, so opening the index there would read a real database the caller never wrote to, and on Postgres the index does not exist at all. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01LWp6Z6wvAPDaDCjAFcTSj6 --- src/task/kb/IndexFilingSectionsTask.test.ts | 61 +++-- src/task/kb/IndexFilingSectionsTask.ts | 86 ++---- .../kb/selectDocumentsToIndex.sqlite.test.ts | 151 +++++++++++ src/task/kb/selectDocumentsToIndex.ts | 248 ++++++++++++++++++ 4 files changed, 465 insertions(+), 81 deletions(-) create mode 100644 src/task/kb/selectDocumentsToIndex.sqlite.test.ts create mode 100644 src/task/kb/selectDocumentsToIndex.ts diff --git a/src/task/kb/IndexFilingSectionsTask.test.ts b/src/task/kb/IndexFilingSectionsTask.test.ts index bab61dde..dd6254b3 100644 --- a/src/task/kb/IndexFilingSectionsTask.test.ts +++ b/src/task/kb/IndexFilingSectionsTask.test.ts @@ -5,15 +5,17 @@ */ import { afterEach, describe, expect, it, vi } from "vitest"; -import { globalServiceRegistry } from "workglow"; +import type { DocumentNode } from "workglow"; +import { Document, globalServiceRegistry, NodeKind } from "workglow"; import { withSqliteDb } from "../../config/testing/withSqliteDb"; -import { resetSecKnowledgeBaseForTesting } from "../../kb/secKnowledgeBase"; +import { getSecKnowledgeBase, resetSecKnowledgeBaseForTesting } from "../../kb/secKnowledgeBase"; import { FILING_DOCUMENT_REPOSITORY_TOKEN, type FilingDocument, } from "../../storage/document/FilingDocumentSchema"; import { FILING_SECTION_REPOSITORY_TOKEN } from "../../storage/document/FilingSectionSchema"; import { IndexFilingSectionsTask } from "./IndexFilingSectionsTask"; +import { kbDocIdFor } from "./selectDocumentsToIndex"; const header = (index: number, filingDate: string): FilingDocument => ({ cik: 320193, @@ -54,30 +56,55 @@ describe("IndexFilingSectionsTask selection", () => { } }; - it("streams the converted filings rather than loading every header first", async () => { + /** Mark a filing as already in the knowledge base. */ + const markIndexed = async (index: number): Promise => { + const kb = await getSecKnowledgeBase(); + const title = `Filing ${index}`; + const root = { kind: NodeKind.DOCUMENT, title, children: [] } as unknown as DocumentNode; + const docId = kbDocIdFor(`0000320193-26-${String(index).padStart(6, "0")}`, "primary.htm"); + await kb.upsertDocument(new Document(root, { title } as never, [], docId)); + }; + + it("picks the work in the database rather than reading headers to sift them", async () => { await seed(5); const repo = globalServiceRegistry.get(FILING_DOCUMENT_REPOSITORY_TOKEN); - const getAll = vi.spyOn(repo, "getAll"); + const reads = (["getAll", "records", "query", "queryPage"] as const).map((method) => + vi.spyOn(repo, method) + ); - const out = await new IndexFilingSectionsTask().run({ limit: 0 }); + const out = await new IndexFilingSectionsTask().run({ cik: 320193, limit: 0 }); - // `getAll()` on the unscoped path is the whole converted corpus in memory - // — hundreds of thousands of rows to take the first few of. - expect(getAll).not.toHaveBeenCalled(); - expect(out).toMatchObject({ indexed: 0, truncated: true }); + // Reading headers to decide is the whole converted corpus crossing the + // process boundary to take the first few of it. + for (const read of reads) expect(read).not.toHaveBeenCalled(); + expect(out.truncated).toBe(true); }); - it("streams a scoped selection by page too", async () => { + it("still reports what is already indexed once nothing is left to do", async () => { + // The count `sec index` prints, and what makes it suggest asking a question + // rather than converting filings there are none of. await seed(3); - const repo = globalServiceRegistry.get(FILING_DOCUMENT_REPOSITORY_TOKEN); - const query = vi.spyOn(repo, "query"); - const queryPage = vi.spyOn(repo, "queryPage"); + for (const index of [0, 1, 2]) await markIndexed(index); - const out = await new IndexFilingSectionsTask().run({ cik: 320193, limit: 0 }); + const out = await new IndexFilingSectionsTask().run({ limit: 5 }); - expect(query).not.toHaveBeenCalled(); - expect(queryPage).toHaveBeenCalled(); - expect(out.truncated).toBe(true); + expect(out).toMatchObject({ indexed: 0, sections: 0, skipped: 3, truncated: false }); + }); + + it("spends its limit on filings that need work, not on ones already indexed", async () => { + // The defect: `limit` counted filings INDEXED, and an already-indexed one + // was skipped without counting toward it — so a small limit on an indexed + // corpus read every row and probed the knowledge base once per row. + await seed(4); + await markIndexed(0); + await markIndexed(1); + const kb = await getSecKnowledgeBase(); + const getDocument = vi.spyOn(kb, "getDocument"); + + const out = await new IndexFilingSectionsTask().run({ limit: 2 }); + + expect(getDocument).not.toHaveBeenCalled(); + expect(out).toMatchObject({ skipped: 2, truncated: false }); }); it("reports no truncation when the scope leaves nothing to index", async () => { diff --git a/src/task/kb/IndexFilingSectionsTask.ts b/src/task/kb/IndexFilingSectionsTask.ts index d8039bc0..e31f4051 100644 --- a/src/task/kb/IndexFilingSectionsTask.ts +++ b/src/task/kb/IndexFilingSectionsTask.ts @@ -5,7 +5,7 @@ */ import { Type } from "typebox"; -import type { DocumentNode, IExecuteContext, PageCursor } from "workglow"; +import type { DocumentNode, IExecuteContext } from "workglow"; import { Document, globalServiceRegistry, @@ -15,17 +15,14 @@ import { TaskAbortedError, } from "workglow"; import { getSecKnowledgeBase } from "../../kb/secKnowledgeBase"; -import { - FILING_DOCUMENT_REPOSITORY_TOKEN, - type FilingDocument, - type FilingDocumentRepositoryStorage, -} from "../../storage/document/FilingDocumentSchema"; +import type { FilingDocument } from "../../storage/document/FilingDocumentSchema"; import { FILING_SECTION_REPOSITORY_TOKEN, type FilingSection, } from "../../storage/document/FilingSectionSchema"; import { accessionWithoutDashes } from "../../util/accession"; import type { TaskPorts } from "../taskPorts"; +import { countAlreadyIndexed, kbDocIdFor, selectDocumentsToIndex } from "./selectDocumentsToIndex"; export interface IndexFilingSectionsTaskInput { /** Only this issuer's filings. */ @@ -63,9 +60,6 @@ export interface IndexFilingSectionsTaskOutput { */ export const DEFAULT_ASK_INDEX_LIMIT = 25; -/** Rows per read while walking the converted filings. */ -const HEADER_PAGE_SIZE = 500; - /** The EDGAR URL a citation points at. */ function filingUrl(cik: number, accession: string, docFile: string): string { return ( @@ -117,32 +111,6 @@ function toDocument(header: FilingDocument, sections: readonly FilingSection[]): } as never); } -/** - * The converted filings matching `criteria`, a page at a time. - * - * Streamed rather than collected: the unscoped case is every converted filing - * in the database, and materializing that array to take the first few of it - * costs the whole corpus in memory before any work starts. - */ -async function* streamHeaders( - repo: FilingDocumentRepositoryStorage, - criteria: Record -): AsyncGenerator { - if (Object.keys(criteria).length === 0) { - yield* repo.records(HEADER_PAGE_SIZE); - return; - } - let cursor: PageCursor | undefined; - for (;;) { - const page = await repo.queryPage(criteria as never, { limit: HEADER_PAGE_SIZE, cursor }); - yield* page.items; - // Both conditions: a cursor can be handed back for a page that concurrent - // deletes have since emptied, and looping on it alone would not terminate. - if (page.nextCursor === undefined || page.items.length === 0) return; - cursor = page.nextCursor; - } -} - /** * Embeds converted filing sections into the knowledge base `sec ask` reads. * @@ -187,43 +155,33 @@ export class IndexFilingSectionsTask extends Task< context: IExecuteContext ): Promise> { const kb = await getSecKnowledgeBase(); - const documentRepo = globalServiceRegistry.get(FILING_DOCUMENT_REPOSITORY_TOKEN); const sectionRepo = globalServiceRegistry.get(FILING_SECTION_REPOSITORY_TOKEN); - const criteria: Record = {}; - if (input.cik !== undefined) criteria.cik = input.cik; - if (input.form !== undefined) criteria.form = input.form; - if (input.accession !== undefined) criteria.accession_number = input.accession; + const scope = { + cik: input.cik, + form: input.form, + since: input.since, + accession: input.accession, + }; - // Only a denominator for the progress line, so an over-count from the - // `since` filter (which no backend expresses as a criterion) is harmless. - const candidates = await documentRepo.count( - Object.keys(criteria).length === 0 ? undefined : (criteria as never) - ); + // One document more than asked for, so "there are more" is something this + // run observed rather than inferred from having filled the limit exactly. const limit = input.limit; - const denominator = Math.max(1, limit === undefined ? candidates : Math.min(limit, candidates)); + const candidates = await selectDocumentsToIndex({ + ...scope, + limit: limit === undefined ? undefined : limit + 1, + force: input.force, + }); + const truncated = limit !== undefined && candidates.length > limit; + const work = truncated ? candidates.slice(0, limit) : candidates; + const skipped = input.force === true ? 0 : await countAlreadyIndexed(scope); + const denominator = Math.max(1, work.length); let indexed = 0; let sectionTotal = 0; - let skipped = 0; - let truncated = false; - for await (const header of streamHeaders(documentRepo, criteria)) { + for (const header of work) { if (context.signal?.aborted) throw new TaskAbortedError(); - // Scope first, then the limit, so a run that stops has genuinely left - // candidates behind rather than rows the scope excludes. - if (input.since !== undefined && (header.filing_date ?? "") < input.since) continue; - if (limit !== undefined && indexed >= limit) { - truncated = true; - break; - } - - // The document id is derived from the filing, so "already indexed" is a - // lookup rather than a second table to keep in step with this one. - const docId = `${header.accession_number}:${header.doc_file}`; - if (input.force !== true && (await kb.getDocument(docId)) !== undefined) { - skipped += 1; - continue; - } + const docId = kbDocIdFor(header.accession_number, header.doc_file); const sections = ( (await sectionRepo.query({ diff --git a/src/task/kb/selectDocumentsToIndex.sqlite.test.ts b/src/task/kb/selectDocumentsToIndex.sqlite.test.ts new file mode 100644 index 00000000..2558827b --- /dev/null +++ b/src/task/kb/selectDocumentsToIndex.sqlite.test.ts @@ -0,0 +1,151 @@ +/** + * @license + * Copyright 2026 Steven Roussey + * SPDX-License-Identifier: Apache-2.0 + */ + +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import type { DocumentNode } from "workglow"; +import { Document, globalServiceRegistry, NodeKind } from "workglow"; +import { withSqliteDb } from "../../config/testing/withSqliteDb"; +import { getSecKnowledgeBase, resetSecKnowledgeBaseForTesting } from "../../kb/secKnowledgeBase"; +import { + FILING_DOCUMENT_REPOSITORY_TOKEN, + type FilingDocument, +} from "../../storage/document/FilingDocumentSchema"; +import { countAlreadyIndexed, kbDocIdFor, selectDocumentsToIndex } from "./selectDocumentsToIndex"; + +const doc = (index: number, over: Partial = {}): FilingDocument => ({ + cik: 320193, + accession_number: `0000320193-26-${String(index).padStart(6, "0")}`, + doc_file: "primary.htm", + doc_type: "10-K", + description: null, + sequence: 1, + is_primary: true, + form: "10-K", + filing_date: "2026-03-01", + title: `Filing ${index}`, + section_count: 1, + char_count: 100, + converter_version: "1", + converted_at: "2026-01-01T00:00:00.000Z", + ...over, +}); + +/** + * `limit` used to bound the number of documents INDEXED, and an already-indexed + * one was skipped without counting toward it. So a small `--limit` on a corpus + * that is already indexed read every row and probed the knowledge base once per + * row before concluding there was nothing to do — and `sec ask` builds the + * index implicitly, so that ran on every question. + * + * These cases are about which rows come back and how many, not about embedding. + */ +describe("selectDocumentsToIndex (sqlite)", () => { + withSqliteDb("select_documents_to_index", [FILING_DOCUMENT_REPOSITORY_TOKEN]); + + beforeEach(async () => { + await resetSecKnowledgeBaseForTesting(); + delete process.env.SEC_EMBEDDING_MODEL; + }); + + afterEach(async () => { + await resetSecKnowledgeBaseForTesting(); + }); + + async function seed(count: number, over: (i: number) => Partial = () => ({})) { + const repo = globalServiceRegistry.get(FILING_DOCUMENT_REPOSITORY_TOKEN); + for (let i = 1; i <= count; i += 1) await repo.put(doc(i, over(i)) as never); + } + + /** Mark a document as already in the knowledge base. */ + async function markIndexed(index: number) { + const kb = await getSecKnowledgeBase(); + const title = `Filing ${index}`; + const root = { kind: NodeKind.DOCUMENT, title, children: [] } as unknown as DocumentNode; + const docId = kbDocIdFor(`0000320193-26-${String(index).padStart(6, "0")}`, "primary.htm"); + await kb.upsertDocument(new Document(root, { title } as never, [], docId)); + } + + it("returns nothing once every document is indexed", async () => { + await seed(3); + for (const i of [1, 2, 3]) await markIndexed(i); + + expect(await selectDocumentsToIndex({ limit: 5 })).toEqual([]); + }); + + it("counts the limit in documents that need work, not documents examined", async () => { + // The defect: two of five are already indexed, so a limit of 2 must return + // the two that are NOT — never fewer because the indexed ones consumed it. + await seed(5); + await markIndexed(1); + await markIndexed(2); + + const picked = await selectDocumentsToIndex({ limit: 2 }); + + expect(picked).toHaveLength(2); + expect(picked.map((d) => d.accession_number)).not.toContain("0000320193-26-000001"); + expect(picked.map((d) => d.accession_number)).not.toContain("0000320193-26-000002"); + }); + + it("re-selects everything under force", async () => { + await seed(3); + for (const i of [1, 2, 3]) await markIndexed(i); + + expect(await selectDocumentsToIndex({ limit: 5, force: true })).toHaveLength(3); + }); + + it("applies `since` in the query rather than after it", async () => { + await seed(4, (i) => ({ filing_date: i <= 2 ? "2020-01-01" : "2026-06-01" })); + + const picked = await selectDocumentsToIndex({ limit: 10, since: "2026-01-01" }); + + expect(picked).toHaveLength(2); + expect(picked.every((d) => (d.filing_date ?? "") >= "2026-01-01")).toBe(true); + }); + + it("narrows by cik, form and accession", async () => { + await seed(3, (i) => ({ form: i === 2 ? "8-K" : "10-K" })); + + expect(await selectDocumentsToIndex({ limit: 10, form: "8-K" })).toHaveLength(1); + expect(await selectDocumentsToIndex({ limit: 10, cik: 999 })).toHaveLength(0); + expect( + await selectDocumentsToIndex({ limit: 10, accession: "0000320193-26-000003" }) + ).toHaveLength(1); + }); + + it("returns newest first, so an interrupted backfill covers what people read", async () => { + await seed(3, (i) => ({ filing_date: `2026-0${i}-01` })); + + const picked = await selectDocumentsToIndex({ limit: 3 }); + + expect(picked.map((d) => d.filing_date)).toEqual(["2026-03-01", "2026-02-01", "2026-01-01"]); + }); + + it("leaves out documents with no sections, which can never be indexed", async () => { + // They embed nothing, so they never enter the knowledge base — selecting + // them spends the limit on work that cannot happen, on every run. + await seed(3, (i) => ({ section_count: i === 2 ? 0 : 1 })); + + const picked = await selectDocumentsToIndex({ limit: 10 }); + + expect(picked.map((d) => d.accession_number)).not.toContain("0000320193-26-000002"); + expect(picked).toHaveLength(2); + }); + + it("counts nothing as already indexed before the index exists", async () => { + // The knowledge-base tables are built by the first command that opens the + // index, and this runs before that — a database with converted filings and + // no index has no table to join against. + await seed(2); + + expect(await selectDocumentsToIndex({ limit: 10 })).toHaveLength(2); + expect(await countAlreadyIndexed({})).toBe(0); + }); + + it("returns nothing for a non-positive limit rather than everything", async () => { + await seed(2); + expect(await selectDocumentsToIndex({ limit: 0 })).toEqual([]); + }); +}); diff --git a/src/task/kb/selectDocumentsToIndex.ts b/src/task/kb/selectDocumentsToIndex.ts new file mode 100644 index 00000000..c36da210 --- /dev/null +++ b/src/task/kb/selectDocumentsToIndex.ts @@ -0,0 +1,248 @@ +/** + * @license + * Copyright 2026 Steven Roussey + * SPDX-License-Identifier: Apache-2.0 + */ + +import type { PageCursor } from "workglow"; +import { globalServiceRegistry } from "workglow"; +import { KB_DOCUMENT_TABLE } from "../../kb/secKbTables"; +import { + FILING_DOCUMENT_REPOSITORY_TOKEN, + type FilingDocument, + type FilingDocumentRepositoryStorage, +} from "../../storage/document/FilingDocumentSchema"; +import { getDb } from "../../util/db"; +import { resolveSqlBackend } from "../../util/sqlBackend"; + +export interface SelectDocumentsOptions { + readonly cik?: number | undefined; + readonly form?: string | undefined; + readonly since?: string | undefined; + readonly accession?: string | undefined; + /** Stop after this many documents that still need indexing; all of them when unset. */ + readonly limit?: number | undefined; + /** Re-index documents the knowledge base already holds. */ + readonly force?: boolean | undefined; +} + +/** The `kb_document.doc_id` a converted document is stored under. */ +export function kbDocIdFor(accession: string, docFile: string): string { + return `${accession}:${docFile}`; +} + +function documentRepoIfRegistered(): FilingDocumentRepositoryStorage | undefined { + return globalServiceRegistry.has(FILING_DOCUMENT_REPOSITORY_TOKEN) + ? globalServiceRegistry.get(FILING_DOCUMENT_REPOSITORY_TOKEN) + : undefined; +} + +/** + * Whether there is a table to anti-join against. + * + * The three knowledge-base tables are built lazily by the first command that + * opens the index, and this selector runs before that — so on a database with + * converted filings that has never been indexed they do not exist yet. Nothing + * is indexed in that case, which is the answer, not an error. + */ +function kbDocumentTableExists(): boolean { + return ( + getDb() + .prepare<[string], { name: string }>( + "SELECT name FROM sqlite_master WHERE type = 'table' AND name = ?" + ) + .all(KB_DOCUMENT_TABLE).length > 0 + ); +} + +/** + * Converted documents the knowledge base does not already hold, newest first. + * + * The set difference is an anti-join, so it is raw SQL for the same reason + * `selectFilingsToConvert` is: `ITabularStorage` cannot express one, and the + * alternative is a knowledge-base round trip per document over a table with + * hundreds of thousands of rows. + * + * That alternative is what this replaces, and the round trips were not the + * worst of it. The limit used to count documents actually INDEXED, and an + * already-indexed one was skipped without counting toward it — so `--limit 5` + * against a corpus that was already indexed read every row and probed the + * knowledge base once per row before concluding it had nothing to do. `ask` + * builds the index implicitly, so that ran on every question. + * + * The fast path is SQLite only, and there is no Postgres arm to write: the + * knowledge base lives in the connection `getDb()` owns and + * `getSecKnowledgeBase` refuses every other backend by name, so where this + * query cannot run the index cannot exist either. + */ +export async function selectDocumentsToIndex( + options: SelectDocumentsOptions +): Promise { + if (options.limit !== undefined && options.limit <= 0) return []; + const documentRepo = documentRepoIfRegistered(); + if (documentRepo === undefined) return []; + + if (resolveSqlBackend("read", documentRepo) !== "sqlite") { + return await selectByScan(documentRepo, options); + } + + const antiJoin = options.force !== true && kbDocumentTableExists(); + const params: (string | number)[] = []; + const clauses: string[] = []; + if (options.cik !== undefined) { + clauses.push("d.`cik` = ?"); + params.push(options.cik); + } + if (options.form !== undefined) { + clauses.push("d.`form` = ?"); + params.push(options.form); + } + if (options.accession !== undefined) { + clauses.push("d.`accession_number` = ?"); + params.push(options.accession); + } + // In the query rather than after it. Filtered in the loop, `since` cost a + // full read of everything older than the cutoff to discard it. + if (options.since !== undefined) { + clauses.push("d.`filing_date` >= ?"); + params.push(options.since); + } + // A document with no sections has nothing to embed, so it never enters the + // knowledge base and would be re-selected on every run — spending the limit + // on work that cannot happen. `section_count` is written in the same + // transaction as the section rows, so it is the same answer reading them + // gives. + clauses.push("d.`section_count` > 0"); + if (antiJoin) clauses.push("k.`doc_id` IS NULL"); + // SQLite numbers `?` by position, so the limit binds last because it is + // written last. + if (options.limit !== undefined) params.push(options.limit); + + const join = antiJoin + ? "LEFT JOIN `" + + KB_DOCUMENT_TABLE + + "` k ON k.`doc_id` = d.`accession_number` || ':' || d.`doc_file`" + : ""; + const where = clauses.length === 0 ? "1 = 1" : clauses.join(" AND "); + return getDb() + .prepare<(string | number)[], FilingDocument>( + `SELECT d.* + FROM \`filing_document\` d + ${join} + WHERE ${where} + ORDER BY d.\`filing_date\` DESC, d.\`accession_number\` DESC + ${options.limit === undefined ? "" : "LIMIT ?"}` + ) + .all(...params); +} + +/** + * How many documents in scope the knowledge base already holds. + * + * Its own query rather than a by-product of the selection, because the + * selection stops at the limit and so cannot count what it never reached. One + * COUNT replaces the unfiltered `count` this run used to take for its progress + * denominator, so it costs no extra query — and the denominator it replaces was + * wrong anyway, measuring a run that indexes three filings out of three hundred + * candidates at one percent. + * + * Zero where the anti-join cannot run, which is the same answer + * {@link selectDocumentsToIndex} gives there: no index, nothing already in it. + */ +export async function countAlreadyIndexed( + options: Omit +): Promise { + const documentRepo = documentRepoIfRegistered(); + if (documentRepo === undefined) return 0; + if (resolveSqlBackend("read", documentRepo) !== "sqlite") return 0; + if (!kbDocumentTableExists()) return 0; + + const params: (string | number)[] = []; + const clauses: string[] = []; + if (options.cik !== undefined) { + clauses.push("d.`cik` = ?"); + params.push(options.cik); + } + if (options.form !== undefined) { + clauses.push("d.`form` = ?"); + params.push(options.form); + } + if (options.accession !== undefined) { + clauses.push("d.`accession_number` = ?"); + params.push(options.accession); + } + if (options.since !== undefined) { + clauses.push("d.`filing_date` >= ?"); + params.push(options.since); + } + const where = clauses.length === 0 ? "1 = 1" : clauses.join(" AND "); + const row = getDb() + .prepare<(string | number)[], { n: number }>( + `SELECT COUNT(*) AS n + FROM \`filing_document\` d + JOIN \`${KB_DOCUMENT_TABLE}\` k + ON k.\`doc_id\` = d.\`accession_number\` || ':' || d.\`doc_file\` + WHERE ${where}` + ) + .get(...params); + return row?.n ?? 0; +} + +const HEADER_PAGE_SIZE = 500; + +/** + * The repository path: page the document table, apply the filters, stop at the + * limit. + * + * It consults no knowledge base, because it is reached only where there cannot + * be one to consult. A non-durable document repository is invisible to + * `getDb()`, so opening the index here would read a real database this caller + * never wrote to and report the wrong documents as already indexed; and on + * Postgres the index does not exist at all. Both are better served by naming + * the candidates and letting the caller fail where the knowledge base itself + * refuses. + */ +async function selectByScan( + repo: FilingDocumentRepositoryStorage, + options: SelectDocumentsOptions +): Promise { + const criteria: Record = {}; + if (options.cik !== undefined) criteria.cik = options.cik; + if (options.form !== undefined) criteria.form = options.form; + if (options.accession !== undefined) criteria.accession_number = options.accession; + + const picked: FilingDocument[] = []; + for await (const header of streamHeaders(repo, criteria)) { + if (options.since !== undefined && (header.filing_date ?? "") < options.since) continue; + if (header.section_count <= 0) continue; + picked.push(header); + if (options.limit !== undefined && picked.length >= options.limit) break; + } + return picked; +} + +/** + * The matching documents, a page at a time. + * + * Streamed rather than collected: the unscoped case is every converted filing + * in the database, and materializing that to take the first few of it costs the + * whole corpus in memory before any work starts. + */ +async function* streamHeaders( + repo: FilingDocumentRepositoryStorage, + criteria: Record +): AsyncGenerator { + if (Object.keys(criteria).length === 0) { + yield* repo.records(HEADER_PAGE_SIZE); + return; + } + let cursor: PageCursor | undefined; + for (;;) { + const page = await repo.queryPage(criteria as never, { limit: HEADER_PAGE_SIZE, cursor }); + yield* page.items; + // Both conditions: a cursor can be handed back for a page that concurrent + // deletes have since emptied, and looping on it alone would not terminate. + if (page.nextCursor === undefined || page.items.length === 0) return; + cursor = page.nextCursor; + } +}