diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 5fa4626..da0ff56 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -49,6 +49,16 @@ jobs: - run: bun install --frozen-lockfile - run: bun run test:ingress + test-ingest: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: oven-sh/setup-bun@v2 + with: + bun-version: 1.3.14 + - run: bun install --frozen-lockfile + - run: bun run test:ingest + build: runs-on: ubuntu-latest steps: diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 1c147c5..ef6485b 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -1,17 +1,23 @@ # Architecture -## Two faces, one dependency direction +## Three faces, one dependency direction -This package has two independent faces, each with its own source directory: +This package has three independent faces, each with its own source directory: - `src/tools/` — the Granola API client and the tools an agent calls (fetch a note, search notes). Published as the package root, `@corbits/granola`. - `src/ingress/` — the extension that receives Granola webhooks, verifies signatures, and dispatches notes to handlers. Published as the subpath export `@corbits/granola/ingress`. - -`src/ingress` depends on `src/tools` for the client. `src/tools` must never depend on -`src/ingress`, or on any hub, mounting, extension or webhook machinery at all — the +- `src/ingest/` — the host-agnostic pipeline that turns an acked webhook event into a + persisted transcript and a dispatched bucket handler: fetch note, resolve bucket, + persist, capture knowledge, hand off. Published as the subpath export + `@corbits/granola/ingest`. + +`src/ingress` depends on `src/tools` for the client. `src/ingest` depends on both — +`src/tools` for the client and note shapes, `src/ingress` for the binding store and +webhook payload it processes. `src/tools` must never depend on `src/ingress` or +`src/ingest`, or on any hub, mounting, extension or webhook machinery at all — the operator requirement is that someone can import the tools and grant them to any agent like any other plain Interchange tool, and nothing hub-shaped comes along for the ride. diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 93e5fd7..ac5fd2c 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -18,13 +18,14 @@ extension depends on the tools, not the other way around. Checked by - **Red first.** A bug fix starts with a test that fails for the reason you believe, and you should watch it fail. - Assert **behavior a consumer can observe** over internal call shapes. -- Tools tests live under `src/tools`; ingress tests live under `src/ingress`. Keep that - split — it is what lets `test:tools` and `test:ingress` run and fail independently in - CI. +- Tools tests live under `src/tools`; ingress tests live under `src/ingress`; ingest + pipeline tests live under `src/ingest`. Keep that split — it is what lets + `test:tools`, `test:ingress`, and `test:ingest` run and fail independently in CI. ## Pull requests - Keep commits focused, and keep the diff to the change you are describing. - Explain *why* in the commit message; the code already says what. -- CI must be green: dependency check, typecheck, tools tests, ingress tests, and build. +- CI must be green: dependency check, typecheck, tools tests, ingress tests, ingest + tests, and build. - Contributions are accepted under the repository's LGPL-2.1-only licence. diff --git a/package.json b/package.json index bd1c56f..102dc35 100644 --- a/package.json +++ b/package.json @@ -37,6 +37,11 @@ "bun": "./src/ingress/index.ts", "types": "./dist/ingress/index.d.ts", "default": "./dist/ingress/index.js" + }, + "./ingest": { + "bun": "./src/ingest/index.ts", + "types": "./dist/ingest/index.d.ts", + "default": "./dist/ingest/index.js" } }, "files": [ @@ -54,6 +59,7 @@ "check-deps": "bun run scripts/check-deps.ts", "test:tools": "bun test src/tools", "test:ingress": "bun test src/ingress", + "test:ingest": "bun test src/ingest", "test": "bun test src", "test:coverage": "bun test --coverage src" }, diff --git a/src/ingest/bucket-handler.ts b/src/ingest/bucket-handler.ts new file mode 100644 index 0000000..78b9265 --- /dev/null +++ b/src/ingest/bucket-handler.ts @@ -0,0 +1,47 @@ +/** + * The contract a Granola bucket-type handler must satisfy. + * + * Declared by the pipeline module, same inversion as the lifecycle + * contract: the pipeline is the generic caller, so the shape it dispatches + * through belongs to whoever implements it. Hosts register concrete + * handlers per bucket type (e.g. Scout's diligence and internal handlers). + * + * `TRef` is whatever the host's transcript store returns from `persist` — + * opaque to the pipeline, handed through to the handler unexamined. + */ +import type { GranolaBucket, GranolaNote } from "../tools/types.js"; + +/** Where the ack message for a note's processing run lives; every later post about that note threads under it. */ +export type GranolaThreadAnchor = { channel: string; ts: string }; + +/** + * Context handed to a bucket-type handler once a note's transcript has been + * persisted and captured. `threadAnchor` is populated for every real + * ingest-triggered dispatch (the ack posted before this handler runs) — it + * is optional only so handlers keep working standalone in tests/call sites + * that construct a context with no ack thread to reply into. + * + * `TAnchor` mirrors the pipeline's own anchor type (established by + * `GranolaIngestLifecycle.onProcessingStarted`) — defaulted to + * `GranolaThreadAnchor` so existing single-type-argument call sites keep + * compiling unchanged. + */ +export type GranolaBucketHandlerContext< + TRef = unknown, + TAnchor = GranolaThreadAnchor, +> = { + note: GranolaNote; + transcriptText: string; + bucket: GranolaBucket; + artifactRef: TRef; + threadAnchor?: TAnchor; + /** + * Set by a host's re-entry affordance (e.g. Scout's ambiguous-ask card): + * the handler pins these companies on the run, skipping extraction. + */ + pinnedCompanies?: string[]; +}; + +export type GranolaBucketHandler = ( + context: GranolaBucketHandlerContext, +) => Promise; diff --git a/src/ingest/index.ts b/src/ingest/index.ts new file mode 100644 index 0000000..21b0eb5 --- /dev/null +++ b/src/ingest/index.ts @@ -0,0 +1,20 @@ +/** + * Public surface of the Granola ingest pipeline: the chat- and + * host-agnostic machinery downstream of an acked webhook event. Hosts + * provide the lifecycle (rendering), transcript store (persistence), + * knowledge capture (enrichment), and bucket handlers (behavior). + */ +export { createGranolaIngest } from "./pipeline.js"; +export type { + CreateGranolaIngestOptions, + GranolaIngest, + GranolaIngestHandlers, + GranolaKnowledgeCapture, + GranolaTranscriptStore, +} from "./pipeline.js"; +export type { GranolaIngestLifecycle } from "./lifecycle.js"; +export type { + GranolaBucketHandler, + GranolaBucketHandlerContext, + GranolaThreadAnchor, +} from "./bucket-handler.js"; diff --git a/src/ingest/lifecycle.ts b/src/ingest/lifecycle.ts new file mode 100644 index 0000000..0b92fee --- /dev/null +++ b/src/ingest/lifecycle.ts @@ -0,0 +1,115 @@ +/** + * The lifecycle-hooks CONTRACT the Granola ingest pipeline reports through. + * + * Declared by the pipeline module because the pipeline is the generic + * caller: the shape it reports through belongs to whoever implements it. + * A host provides a concrete implementation (e.g. Scout's Slack-rendering + * lifecycle) and hands it to `createGranolaIngest`. + */ +import type { GranolaBucketType } from "../tools/types.js"; +import type { GranolaThreadAnchor } from "./bucket-handler.js"; + +/** + * Every human-visible event the pipeline reports, in the order a single + * note's run can produce them. `TAnchor` is whatever the host's + * `onProcessingStarted` returns — opaque to the pipeline, threaded back + * unexamined through every later hook for that note's run. + * + * One hook per event, not one generic "post text" hook: each carries + * exactly the structured data a host needs to render its own copy (or + * choose not to post at all), rather than the pipeline deciding wording. + */ +export type GranolaIngestLifecycle = { + /** + * A second event for a note already being processed arrived — usually + * two webhook deliveries racing (e.g. `note.generated` and `note.edited` + * close together). `anchor` is set once processing reached + * `onProcessingStarted`; `bucketChannel` is set once a bucket resolved + * even if processing hasn't reached that point yet. Both undefined means + * there is nowhere known to notify — the host may no-op. + */ + onDuplicateEvent(args: { + noteId: string; + anchor: TAnchor | undefined; + bucketChannel: string | undefined; + }): Promise; + + /** + * The note couldn't be fetched from the Granola API at all — no bucket is + * known yet (folder membership only comes back from the fetch that just + * failed), so `candidateChannels` is every channel any current binding + * points at, deduped. The host decides how to broadcast (or not) across + * them. + */ + onFetchFailed(args: { + noteId: string; + error: string; + candidateChannels: string[]; + }): Promise; + + /** + * A `note.access_granted` event arrived for a note whose AI summary hasn't + * been generated yet — Granola's API 404s on ungenerated notes and offers + * no way to trigger generation. Recovery is automatic: a human clicking + * "Generate Notes" in Granola fires `note.generated`, which re-enters the + * pipeline. Like `onFetchFailed`, no bucket is known yet, so + * `candidateChannels` is every bound channel. + */ + onNoteNotGenerated(args: { + noteId: string; + candidateChannels: string[]; + }): Promise; + + /** + * A re-added note already has a persisted transcript — the durable + * human-in-the-loop gate: a human decides whether to reprocess rather + * than the pipeline silently skipping or blindly rerunning. + */ + onAlreadyProcessed(args: { + noteId: string; + noteTitle: string; + bucketChannel: string; + }): Promise; + + /** + * Fresh processing is starting for a note whose bucket just resolved. + * Returns the anchor this note's run threads every later hook through — + * e.g. a Slack host posts an initial message here and returns its + * channel+ts. + */ + onProcessingStarted(args: { + noteId: string; + noteTitle: string; + bucketChannel: string; + }): Promise; + + /** + * The note has no transcript/summary yet. `isFinalAttempt: false` means a + * retry was just scheduled `retryDelayMs` out; `true` means the retry + * already ran and the pipeline is giving up on the note until another + * webhook event arrives for it. + */ + onNoteNotReady(args: { + anchor: TAnchor; + isFinalAttempt: boolean; + retryDelayMs: number; + }): Promise; + + /** Transcript/summary text is in hand and about to be persisted. */ + onTranscriptReady(args: { + anchor: TAnchor; + segmentCount: number | undefined; + }): Promise; + + /** Persisting the transcript artifact failed; processing stops here. */ + onPersistFailed(args: { anchor: TAnchor; error: string }): Promise; + + /** About to dispatch to the bucket-type handler — the last stage before handler-specific behavior takes over. */ + onHandlerDispatching(args: { + anchor: TAnchor; + bucketType: GranolaBucketType; + }): Promise; + + /** The bucket-type handler threw. */ + onHandlerFailed(args: { anchor: TAnchor; error: string }): Promise; +}; diff --git a/src/ingest/pipeline.test.ts b/src/ingest/pipeline.test.ts new file mode 100644 index 0000000..ad835bb --- /dev/null +++ b/src/ingest/pipeline.test.ts @@ -0,0 +1,456 @@ +import { describe, expect, test } from "bun:test"; + +import { GranolaApiError, type GranolaClient, type GranolaNote } from "../tools/client.js"; +import type { GranolaBucket } from "../tools/types.js"; +import type { GranolaBindingStore, GranolaWebhookPayload } from "../ingress/index.js"; +import { createGranolaIngest, type GranolaTranscriptStore } from "./pipeline.js"; +import type { GranolaIngestLifecycle } from "./lifecycle.js"; +import type { GranolaThreadAnchor } from "./bucket-handler.js"; + +const BUCKETS: GranolaBucket[] = [ + { folderId: "fol_diligence", type: "diligence", channel: "C_DILIGENCE" }, + { folderId: "fol_internal", type: "internal", channel: "C_INTERNAL" }, +]; + +function samplePayload( + overrides: Partial = {}, +): GranolaWebhookPayload { + return { + event_id: "evt_1", + event_type: "note.generated", + note_id: "note_1", + occurred_at: "2026-07-31T00:00:00Z", + ...overrides, + }; +} + +function noteWith(overrides: Partial = {}): GranolaNote { + return { + id: "note_1", + title: "Weekly sync", + folder_membership: { folderId: "fol_diligence" }, + transcript: [ + { speaker: { source: "microphone", attribution: "Alice" }, text: "Hello." }, + ], + ...overrides, + }; +} + +function fakeClient(notesByCall: GranolaNote[]): GranolaClient { + let index = 0; + return { + async getNote() { + const note = notesByCall[Math.min(index, notesByCall.length - 1)]; + index += 1; + if (note === undefined) throw new Error("no note configured for this call"); + return note; + }, + async listNotes() { + throw new Error("not exercised in this suite"); + }, + async listFolders() { + throw new Error("not exercised in this suite"); + }, + }; +} + +function failingClient(error: Error): GranolaClient { + return { + async getNote() { + throw error; + }, + async listNotes() { + throw new Error("not exercised in this suite"); + }, + async listFolders() { + throw new Error("not exercised in this suite"); + }, + }; +} + +function fakeBindingStore(buckets: GranolaBucket[] = BUCKETS): GranolaBindingStore { + return { + async list() { + return buckets; + }, + async replaceAll() { + throw new Error("not exercised in this suite"); + }, + }; +} + +type TestRef = { artifactId: string }; + +function fakeTranscripts(existingNoteIds: string[] = []): { + transcripts: GranolaTranscriptStore; + persisted: { granolaNoteId: string; text: string; bucketType: string }[]; +} { + const persisted: { granolaNoteId: string; text: string; bucketType: string }[] = []; + const existing = new Set(existingNoteIds); + return { + persisted, + transcripts: { + async hasTranscript(granolaNoteId) { + return existing.has(granolaNoteId); + }, + async persist(args) { + persisted.push({ + granolaNoteId: args.granolaNoteId, + text: args.text, + bucketType: args.bucketType, + }); + return { artifactId: `art_${args.granolaNoteId}` }; + }, + }, + }; +} + +type LifecycleEvent = { hook: string; args: unknown }; + +function recordingLifecycle(): { + lifecycle: GranolaIngestLifecycle; + events: LifecycleEvent[]; +} { + const events: LifecycleEvent[] = []; + const record = + (hook: string) => + async (args: unknown): Promise => { + events.push({ hook, args }); + }; + return { + events, + lifecycle: { + onDuplicateEvent: record("onDuplicateEvent"), + onFetchFailed: record("onFetchFailed"), + onNoteNotGenerated: record("onNoteNotGenerated"), + onAlreadyProcessed: record("onAlreadyProcessed"), + async onProcessingStarted(args) { + events.push({ hook: "onProcessingStarted", args }); + return { channel: args.bucketChannel, ts: "1700000000.000100" }; + }, + onNoteNotReady: record("onNoteNotReady"), + onTranscriptReady: record("onTranscriptReady"), + onPersistFailed: record("onPersistFailed"), + onHandlerDispatching: record("onHandlerDispatching"), + onHandlerFailed: record("onHandlerFailed"), + }, + }; +} + +function hooks(events: LifecycleEvent[]): string[] { + return events.map((event) => event.hook); +} + +describe("createGranolaIngest", () => { + test("a ready note persists, captures knowledge, and dispatches to its bucket handler", async () => { + const { transcripts, persisted } = fakeTranscripts(); + const { lifecycle, events } = recordingLifecycle(); + const captured: { artifactId: string; title: string }[] = []; + const handled: string[] = []; + + const onEvent = createGranolaIngest({ + client: fakeClient([noteWith()]), + bindingStore: fakeBindingStore(), + transcripts, + captureKnowledge: async ({ artifactRef, noteTitle }) => { + captured.push({ artifactId: artifactRef.artifactId, title: noteTitle }); + }, + lifecycle, + handlers: { + diligence: async (context) => { + handled.push(context.artifactRef.artifactId); + expect(context.threadAnchor).toEqual({ + channel: "C_DILIGENCE", + ts: "1700000000.000100", + }); + }, + }, + }); + + await onEvent(samplePayload()); + + expect(persisted).toHaveLength(1); + expect(persisted[0]?.bucketType).toBe("diligence"); + expect(persisted[0]?.text).toContain("Alice: Hello."); + expect(captured).toEqual([{ artifactId: "art_note_1", title: "Weekly sync" }]); + expect(handled).toEqual(["art_note_1"]); + expect(hooks(events)).toEqual([ + "onProcessingStarted", + "onTranscriptReady", + "onHandlerDispatching", + ]); + }); + + test("a summary-only note persists the summary markdown, never an empty artifact", async () => { + const summary = "## Summary\nDiscussed the pilot."; + const { transcripts, persisted } = fakeTranscripts(); + const { lifecycle } = recordingLifecycle(); + + const onEvent = createGranolaIngest({ + client: fakeClient([ + noteWith({ transcript: undefined, summary_markdown: summary }), + ]), + bindingStore: fakeBindingStore(), + transcripts, + captureKnowledge: async () => {}, + lifecycle, + }); + + await onEvent(samplePayload()); + + expect(persisted[0]?.text).toBe(summary); + }); + + test("a note with no bound folder is ignored without lifecycle noise", async () => { + const { transcripts, persisted } = fakeTranscripts(); + const { lifecycle, events } = recordingLifecycle(); + + const onEvent = createGranolaIngest({ + client: fakeClient([noteWith({ folder_membership: { folderId: "fol_unbound" } })]), + bindingStore: fakeBindingStore(), + transcripts, + captureKnowledge: async () => {}, + lifecycle, + }); + + await onEvent(samplePayload()); + + expect(persisted).toHaveLength(0); + expect(events).toHaveLength(0); + }); + + test("a fetch failure reports onFetchFailed with every bound channel", async () => { + const { transcripts, persisted } = fakeTranscripts(); + const { lifecycle, events } = recordingLifecycle(); + + const onEvent = createGranolaIngest({ + client: failingClient(new Error("granola 500")), + bindingStore: fakeBindingStore(), + transcripts, + captureKnowledge: async () => {}, + lifecycle, + }); + + await onEvent(samplePayload()); + + expect(persisted).toHaveLength(0); + expect(hooks(events)).toEqual(["onFetchFailed"]); + expect(events[0]?.args).toEqual({ + noteId: "note_1", + error: "granola 500", + candidateChannels: ["C_DILIGENCE", "C_INTERNAL"], + }); + }); + + test("a 404 on note.access_granted reports onNoteNotGenerated instead of onFetchFailed", async () => { + const { transcripts } = fakeTranscripts(); + const { lifecycle, events } = recordingLifecycle(); + + const onEvent = createGranolaIngest({ + client: failingClient(new GranolaApiError(404, "Not Found", "")), + bindingStore: fakeBindingStore(), + transcripts, + captureKnowledge: async () => {}, + lifecycle, + }); + + await onEvent(samplePayload({ event_type: "note.access_granted" })); + + expect(hooks(events)).toEqual(["onNoteNotGenerated"]); + expect(events[0]?.args).toEqual({ + noteId: "note_1", + candidateChannels: ["C_DILIGENCE", "C_INTERNAL"], + }); + }); + + test("a 404 on note.generated stays on the onFetchFailed path", async () => { + const { transcripts } = fakeTranscripts(); + const { lifecycle, events } = recordingLifecycle(); + + const onEvent = createGranolaIngest({ + client: failingClient(new GranolaApiError(404, "Not Found", "")), + bindingStore: fakeBindingStore(), + transcripts, + captureKnowledge: async () => {}, + lifecycle, + }); + + await onEvent(samplePayload({ event_type: "note.generated" })); + + expect(hooks(events)).toEqual(["onFetchFailed"]); + }); + + test("an already-persisted note gates on onAlreadyProcessed instead of reprocessing", async () => { + const { transcripts, persisted } = fakeTranscripts(["note_1"]); + const { lifecycle, events } = recordingLifecycle(); + + const onEvent = createGranolaIngest({ + client: fakeClient([noteWith()]), + bindingStore: fakeBindingStore(), + transcripts, + captureKnowledge: async () => {}, + lifecycle, + }); + + await onEvent(samplePayload()); + + expect(persisted).toHaveLength(0); + expect(hooks(events)).toEqual(["onAlreadyProcessed"]); + }); + + test("reprocess bypasses the already-processed gate", async () => { + const { transcripts, persisted } = fakeTranscripts(["note_1"]); + const { lifecycle } = recordingLifecycle(); + + const onEvent = createGranolaIngest({ + client: fakeClient([noteWith()]), + bindingStore: fakeBindingStore(), + transcripts, + captureKnowledge: async () => {}, + lifecycle, + }); + + await onEvent.reprocess("note_1"); + + expect(persisted).toHaveLength(1); + }); + + test("reprocessPinned threads pinnedCompanies through to the handler context", async () => { + const { transcripts } = fakeTranscripts(); + const { lifecycle } = recordingLifecycle(); + const pinned: (string[] | undefined)[] = []; + + const onEvent = createGranolaIngest({ + client: fakeClient([noteWith()]), + bindingStore: fakeBindingStore(), + transcripts, + captureKnowledge: async () => {}, + lifecycle, + handlers: { + diligence: async (context) => { + pinned.push(context.pinnedCompanies); + }, + }, + }); + + await onEvent.reprocessPinned("note_1", ["Acme", "Globex"]); + + expect(pinned).toEqual([["Acme", "Globex"]]); + }); + + test("a not-ready note schedules one retry and gives up after it", async () => { + const notReady = noteWith({ transcript: undefined }); + const { transcripts, persisted } = fakeTranscripts(); + const { lifecycle, events } = recordingLifecycle(); + + const onEvent = createGranolaIngest({ + client: fakeClient([notReady, notReady]), + bindingStore: fakeBindingStore(), + transcripts, + captureKnowledge: async () => {}, + lifecycle, + retryDelayMs: 5, + }); + + await onEvent(samplePayload()); + await new Promise((resolve) => setTimeout(resolve, 30)); + + expect(persisted).toHaveLength(0); + expect(hooks(events)).toEqual([ + "onProcessingStarted", + "onNoteNotReady", + "onNoteNotReady", + ]); + expect(events[2]?.args).toMatchObject({ isFinalAttempt: true }); + }); + + test("a concurrent event for an in-flight note reports onDuplicateEvent", async () => { + const { transcripts } = fakeTranscripts(); + const { lifecycle, events } = recordingLifecycle(); + let releaseHandler: () => void = () => {}; + const handlerGate = new Promise((resolve) => { + releaseHandler = resolve; + }); + + const onEvent = createGranolaIngest({ + client: fakeClient([noteWith()]), + bindingStore: fakeBindingStore(), + transcripts, + captureKnowledge: async () => {}, + lifecycle, + handlers: { + diligence: async () => { + await handlerGate; + }, + }, + }); + + const first = onEvent(samplePayload()); + await new Promise((resolve) => setTimeout(resolve, 10)); + await onEvent(samplePayload({ event_id: "evt_2", event_type: "note.edited" })); + releaseHandler(); + await first; + + expect(hooks(events)).toContain("onDuplicateEvent"); + }); + + test("a knowledge-capture failure does not block handler dispatch", async () => { + const { transcripts } = fakeTranscripts(); + const { lifecycle, events } = recordingLifecycle(); + let dispatched = false; + + const onEvent = createGranolaIngest({ + client: fakeClient([noteWith()]), + bindingStore: fakeBindingStore(), + transcripts, + captureKnowledge: async () => { + throw new Error("embedding service down"); + }, + lifecycle, + handlers: { + diligence: async () => { + dispatched = true; + }, + }, + }); + + await onEvent(samplePayload()); + + expect(dispatched).toBe(true); + expect(hooks(events)).toContain("onHandlerDispatching"); + }); + + test("a persist failure reports onPersistFailed and stops", async () => { + const { lifecycle, events } = recordingLifecycle(); + let dispatched = false; + + const onEvent = createGranolaIngest({ + client: fakeClient([noteWith()]), + bindingStore: fakeBindingStore(), + transcripts: { + async hasTranscript() { + return false; + }, + async persist() { + throw new Error("disk full"); + }, + }, + captureKnowledge: async () => {}, + lifecycle, + handlers: { + diligence: async () => { + dispatched = true; + }, + }, + }); + + await onEvent(samplePayload()); + + expect(dispatched).toBe(false); + expect(hooks(events)).toEqual([ + "onProcessingStarted", + "onTranscriptReady", + "onPersistFailed", + ]); + }); +}); diff --git a/src/ingest/pipeline.ts b/src/ingest/pipeline.ts new file mode 100644 index 0000000..12eb133 --- /dev/null +++ b/src/ingest/pipeline.ts @@ -0,0 +1,613 @@ +/** + * Granola ingestion pipeline: webhook event -> fetch note -> resolve bucket + * -> persist transcript -> knowledge capture -> bucket-type handler. + * + * `createGranolaIngest` returns the `onEvent` function `mountGranolaWebhook` + * calls after it has already verified the signature, deduped by `event_id`, + * and acked the webhook — this module owns everything downstream of that. + * + * CHAT-AGNOSTIC AND HOST-AGNOSTIC BY DESIGN: this module never imports a + * chat client, never builds UI, and never touches a host's storage schema. + * Every human-visible event goes through the injected + * `lifecycle: GranolaIngestLifecycle`; transcript persistence goes + * through the injected `transcripts: GranolaTranscriptStore`; and + * knowledge capture goes through the injected `captureKnowledge` port. + * `TAnchor` (established by `onProcessingStarted`) and `TRef` (returned by + * `transcripts.persist`) are opaque here — threaded through unexamined. + */ +import { getLogger } from "@intx/log"; + +import { + GranolaApiError, + transcriptText, + type GranolaClient, + type GranolaNote, +} from "../tools/client.js"; +import type { GranolaBucket, GranolaBucketType } from "../tools/types.js"; +import type { GranolaBindingStore } from "../ingress/binding-store.js"; +import type { GranolaWebhookPayload } from "../ingress/webhook.js"; +import type { GranolaIngestLifecycle } from "./lifecycle.js"; +import type { + GranolaBucketHandler, + GranolaThreadAnchor, +} from "./bucket-handler.js"; + +type Logger = ReturnType; + +const log = getLogger(["granola", "ingest"]); + +/** How long to wait before the one retry for a note whose transcript isn't ready yet. */ +const DEFAULT_RETRY_DELAY_MS = 60_000; + +export type GranolaIngestHandlers< + TRef = unknown, + TAnchor = GranolaThreadAnchor, +> = Partial>>; + +/** + * The host's durable transcript storage. `hasTranscript` backs the + * already-processed human-in-the-loop gate; `persist` stores the note's + * text and returns whatever ref the host's handlers and knowledge capture + * need back (`TRef` is opaque to the pipeline). The host closes over its + * own tenancy/attribution — the pipeline never sees a principal. + */ +export type GranolaTranscriptStore = { + hasTranscript(granolaNoteId: string): Promise; + persist(args: { + granolaNoteId: string; + noteTitle: string; + bucketType: string; + bucketChannel: string; + text: string; + }): Promise; +}; + +/** + * Knowledge capture port — enrichment, not a gate: the pipeline calls it + * after the transcript artifact is persisted and swallows (logs) failures, + * so a capture error never costs the host its downstream processing. + */ +export type GranolaKnowledgeCapture = (args: { + artifactRef: TRef; + noteTitle: string; + text: string; +}) => Promise; + +export type CreateGranolaIngestOptions< + TAnchor = GranolaThreadAnchor, + TRef = unknown, +> = { + client: GranolaClient; + /** Resolves a note's folder ids to a bucket per event — see `GranolaBindingStore`. */ + bindingStore: GranolaBindingStore; + transcripts: GranolaTranscriptStore; + captureKnowledge: GranolaKnowledgeCapture; + /** Reports every human-visible event — see `GranolaIngestLifecycle`'s own doc comment. */ + lifecycle: GranolaIngestLifecycle; + /** + * Per-bucket-type handlers, injected by the caller. A bucket type with no + * handler here falls back to a no-op that only logs. + */ + handlers?: GranolaIngestHandlers; + log?: Logger; + /** Delay before the one retry for a not-yet-ready note. Defaults to 60s; overridable for tests. */ + retryDelayMs?: number; +}; + +function defaultHandler( + bucketType: GranolaBucketType, + logger: Logger, +): GranolaBucketHandler { + return async (context) => { + logger.info( + "No handler registered for bucket type {bucketType} — note {noteId} persisted but not otherwise processed", + { bucketType, noteId: context.note.id }, + ); + }; +} + +/** + * Extracts folder ids from a note's `folder_membership` field, whose shape + * the client deliberately leaves as `unknown` (the Granola API docs don't + * pin it down precisely). Tolerates the shapes a folder membership + * plausibly takes — a single id, a list of ids, or a list of + * `{folderId}`/`{id}` objects — rather than throwing on an unexpected + * shape, since an ingestion pipeline should degrade to "no bucket matched" + * instead of crashing on a note whose membership just looks different than + * expected. + */ +function extractFolderIds(membership: unknown): string[] { + if (typeof membership === "string") return [membership]; + + if (Array.isArray(membership)) { + return membership.flatMap((entry) => { + if (typeof entry === "string") return [entry]; + if (typeof entry === "object" && entry !== null) { + const record = entry as Record; + const id = record["folderId"] ?? record["id"]; + return typeof id === "string" ? [id] : []; + } + return []; + }); + } + + if (typeof membership === "object" && membership !== null) { + const record = membership as Record; + if (typeof record["folderId"] === "string") return [record["folderId"]]; + if (Array.isArray(record["folderIds"])) { + return (record["folderIds"] as unknown[]).filter( + (id): id is string => typeof id === "string", + ); + } + } + + return []; +} + +async function resolveBucket( + bindingStore: GranolaBindingStore, + note: GranolaNote, + logger: Logger, +): Promise { + const folderIds = extractFolderIds(note.folder_membership); + if (folderIds.length === 0) return undefined; + + const buckets = await bindingStore.list(); + const matches = buckets.filter((bucket) => + folderIds.includes(bucket.folderId), + ); + if (matches.length > 1) { + logger.info( + "Granola note {noteId} folders match {count} configured buckets — using the first match ({folderId})", + { + noteId: note.id, + count: matches.length, + folderId: matches[0]?.folderId, + }, + ); + } + return matches[0]; +} + +/** + * A note is "not ready" when Granola delivered the webhook before it + * finished generating a summary/transcript — the note exists but has + * nothing worth persisting yet. + */ +function isNoteReady(note: GranolaNote): boolean { + return noteText(note).length > 0; +} + +/** + * The text worth persisting: the transcript when Granola delivered one, + * otherwise the summary. Never returns whitespace-only text, so a ready + * note can never persist an empty artifact. + */ +function noteText(note: GranolaNote): string { + const transcript = transcriptText(note).trim(); + if (transcript.length > 0) return transcript; + const markdown = note.summary_markdown?.trim() ?? ""; + if (markdown.length > 0) return markdown; + return note.summary_text?.trim() ?? ""; +} + +/** + * Per-note state threaded through one processing run (including its + * not-ready retry): the lifecycle anchor once established, the resolved + * bucket channel once known, and the re-entry flags a host's Reprocess + * affordances set. + */ +type InFlightNote = { + anchor?: TAnchor; + bucketChannel?: string; + forceReprocess?: boolean; + pinnedCompanies?: string[]; +}; + +/** The webhook entry plus the host's forced re-entry affordances. */ +export type GranolaIngest = (( + payload: GranolaWebhookPayload, +) => Promise) & { + /** Re-enters the pipeline bypassing the already-processed gate — the human just answered it. */ + reprocess(noteId: string): Promise; + /** + * Forced reprocess with `pinnedCompanies` pinned through to the bucket + * handler's context, so the handler skips extraction and fans out exactly + * these companies. + */ + reprocessPinned(noteId: string, companies: string[]): Promise; +}; + +export function createGranolaIngest< + TAnchor = GranolaThreadAnchor, + TRef = unknown, +>(options: CreateGranolaIngestOptions): GranolaIngest { + const { client, bindingStore, transcripts, captureKnowledge, lifecycle } = + options; + const logger = options.log ?? log; + const retryDelayMs = options.retryDelayMs ?? DEFAULT_RETRY_DELAY_MS; + const handlers = options.handlers ?? {}; + + // Bounded by nature: at most one pending retry per in-flight not-ready + // note. Non-persistent — a host restart within the retry window loses the + // scheduled retry and the note simply waits for its next webhook event + // (regenerated/edited) to be reprocessed from scratch. + const pendingRetries = new Set(); + + // Guards against two concurrent events for the same note_id (e.g. + // note.generated and note.edited delivered close together) both running + // the pipeline for the same note at once. Held only for the duration of + // one processing run (including its retry) — a fresh event for a note + // that is NOT currently in flight always reprocesses from scratch, whether + // the note was never processed or was already completed. A concurrent + // duplicate is dropped with a notice rather than queued (see `processNote`). + const inFlight = new Map>(); + + function handlerFor( + bucketType: GranolaBucketType, + ): GranolaBucketHandler { + return ( + handlers[bucketType] ?? defaultHandler(bucketType, logger) + ); + } + + async function notifyDuplicate( + noteId: string, + state: InFlightNote, + ): Promise { + await lifecycle + .onDuplicateEvent({ + noteId, + anchor: state.anchor, + bucketChannel: state.bucketChannel, + }) + .catch((cause: unknown) => { + logger.error("Granola duplicate-notice hook failed: {error}", { + error: cause instanceof Error ? cause.message : String(cause), + }); + }); + } + + async function processNote( + noteId: string, + eventType: string, + isRetry: boolean, + carriedState?: InFlightNote, + ): Promise { + const existing = inFlight.get(noteId); + if (existing !== undefined) { + logger.info( + "Granola note {noteId} is already being processed — ignoring concurrent {eventType}", + { noteId, eventType }, + ); + await notifyDuplicate(noteId, existing); + return; + } + const state = carriedState ?? {}; + inFlight.set(noteId, state); + try { + await processNoteBody(noteId, eventType, isRetry, state); + } finally { + inFlight.delete(noteId); + } + } + + async function processNoteBody( + noteId: string, + eventType: string, + isRetry: boolean, + state: InFlightNote, + ): Promise { + // Bucket resolution needs the note's folder membership, which only + // comes back from the note fetch itself — so this one fetch happens + // before processing is reported as started. `onProcessingStarted` still + // fires before anything else in the pipeline (the not-ready retry wait, + // persistence, capture, handler dispatch). + // + // A fetch failure can't know WHICH bound channel owns the note (the + // folder membership is inside the fetch that just failed), so + // `onFetchFailed`/`onNoteNotGenerated` carry every bound channel — the + // no-silent-exits rule prefers a couple of channels seeing "couldn't + // fetch call X" over an operator re-dropping a call into total silence. + let note: GranolaNote; + try { + note = await client.getNote(noteId, { includeTranscript: true }); + } catch (cause) { + const message = cause instanceof Error ? cause.message : String(cause); + // A 404 on note.access_granted is documented Granola behavior, not a + // failure: an ungenerated call was added to a folder, and the note only + // becomes fetchable once a human clicks "Generate Notes" (which fires + // note.generated). Any other 404 — e.g. on note.generated itself — is + // genuinely anomalous and stays on the failure path. + const notGeneratedYet = + cause instanceof GranolaApiError && + cause.status === 404 && + eventType === "note.access_granted"; + if (notGeneratedYet) { + logger.info( + "Granola note {noteId} isn't generated yet — waiting for note.generated", + { noteId }, + ); + } else { + logger.error("Granola note fetch failed for {noteId}: {error}", { + noteId, + error: message, + }); + } + let candidateChannels: string[] = []; + try { + const bindings = await bindingStore.list(); + candidateChannels = [...new Set(bindings.map((bucket) => bucket.channel))]; + } catch { + // No bindings readable — logging above is all that's possible. + } + await (notGeneratedYet + ? lifecycle.onNoteNotGenerated({ noteId, candidateChannels }) + : lifecycle.onFetchFailed({ noteId, error: message, candidateChannels }) + ).catch((hookCause: unknown) => { + logger.error("Granola fetch-failure hook failed: {error}", { + error: + hookCause instanceof Error + ? hookCause.message + : String(hookCause), + }); + }); + return; + } + + const bucket = await resolveBucket(bindingStore, note, logger); + if (bucket === undefined) { + // No bound channel exists to notify — nothing to over-communicate to. + logger.info( + "Granola note {noteId} has no folder bound to a bucket — ignoring", + { noteId }, + ); + return; + } + state.bucketChannel = bucket.channel; + + // Human-in-the-loop gate (operator design rule): the durable record — a + // persisted transcript — is the source of truth for "already + // processed". A re-added note doesn't silently skip (invisible) or + // blindly rerun (wasteful); the human decides via whatever affordance + // the host's `onAlreadyProcessed` renders. `ingest.reprocess`/ + // `ingest.reprocessPinned` (below) are that affordance's re-entry. + if (state.forceReprocess !== true && !isRetry) { + let alreadyProcessed = false; + try { + alreadyProcessed = await transcripts.hasTranscript(noteId); + } catch (cause) { + // The check is an optimization for the human, never a gate on the + // pipeline: if it fails, process as if new. + logger.warn( + "Already-processed lookup failed for note {noteId} — treating as new: {error}", + { + noteId, + error: cause instanceof Error ? cause.message : String(cause), + }, + ); + } + if (alreadyProcessed) { + await lifecycle.onAlreadyProcessed({ + noteId, + noteTitle: note.title, + bucketChannel: bucket.channel, + }); + return; + } + } + + if (state.anchor === undefined) { + try { + state.anchor = await lifecycle.onProcessingStarted({ + noteId, + noteTitle: note.title, + bucketChannel: bucket.channel, + }); + } catch (cause) { + logger.error( + "Granola processing-started hook failed for note {noteId}: {error}", + { + noteId, + error: cause instanceof Error ? cause.message : String(cause), + }, + ); + return; + } + } + const anchor = state.anchor; + + if (!isNoteReady(note)) { + if (isRetry) { + logger.info( + "Granola note {noteId} still has no transcript/summary after retry — giving up", + { noteId }, + ); + await lifecycle + .onNoteNotReady({ anchor, isFinalAttempt: true, retryDelayMs }) + .catch((cause: unknown) => { + logger.error( + "Granola note-not-ready (final) hook failed for note {noteId}: {error}", + { + noteId, + error: cause instanceof Error ? cause.message : String(cause), + }, + ); + }); + return; + } + if (pendingRetries.has(noteId)) { + logger.info( + "Granola note {noteId} already has a pending retry — not scheduling another", + { + noteId, + }, + ); + return; + } + pendingRetries.add(noteId); + logger.info( + "Granola note {noteId} has no transcript/summary yet — retrying once in {retryDelayMs}ms", + { noteId, retryDelayMs }, + ); + await lifecycle + .onNoteNotReady({ anchor, isFinalAttempt: false, retryDelayMs }) + .catch((cause: unknown) => { + logger.error( + "Granola note-not-ready hook failed for note {noteId}: {error}", + { + noteId, + error: cause instanceof Error ? cause.message : String(cause), + }, + ); + }); + setTimeout(() => { + pendingRetries.delete(noteId); + processNote(noteId, eventType, true, state).catch( + (cause: unknown) => { + logger.error("Granola retry for note {noteId} failed: {error}", { + noteId, + error: cause instanceof Error ? cause.message : String(cause), + }); + }, + ); + }, retryDelayMs); + return; + } + + const segmentCount = note.transcript?.length; + await lifecycle + .onTranscriptReady({ anchor, segmentCount }) + .catch((cause: unknown) => { + logger.error( + "Granola transcript-ready hook failed for note {noteId}: {error}", + { + noteId, + error: cause instanceof Error ? cause.message : String(cause), + }, + ); + }); + + // Re-drops are the documented retry path: a note that was already + // processed reprocesses from scratch on any fresh webhook event, rather + // than being permanently gated by a prior `hasTranscript` hit. Guarding + // against overwriting a newer artifact with a stale re-fetch is the + // host store's job, not this module's. + const text = noteText(note); + let artifactRef: TRef; + try { + artifactRef = await transcripts.persist({ + granolaNoteId: noteId, + noteTitle: note.title, + bucketType: bucket.type, + bucketChannel: bucket.channel, + text, + }); + } catch (cause) { + const message = cause instanceof Error ? cause.message : String(cause); + logger.error( + "Persisting the call transcript failed for note {noteId}: {error}", + { noteId, error: message }, + ); + await lifecycle + .onPersistFailed({ anchor, error: message }) + .catch((hookCause: unknown) => { + logger.error( + "Granola persist-failed hook failed for note {noteId}: {error}", + { + noteId, + error: + hookCause instanceof Error + ? hookCause.message + : String(hookCause), + }, + ); + }); + return; + } + + // Knowledge capture is enrichment, not a gate: the artifact is already + // persisted, and the mount has already acked (so nothing external + // retries). Failing here must not cost the host its downstream work. + try { + await captureKnowledge({ artifactRef, noteTitle: note.title, text }); + } catch (cause) { + logger.error( + "Knowledge capture failed for Granola note {noteId} — continuing to handler dispatch: {error}", + { + noteId, + error: cause instanceof Error ? cause.message : String(cause), + }, + ); + } + + await lifecycle + .onHandlerDispatching({ anchor, bucketType: bucket.type }) + .catch((cause: unknown) => { + logger.error( + "Granola handler-dispatching hook failed for note {noteId}: {error}", + { + noteId, + error: cause instanceof Error ? cause.message : String(cause), + }, + ); + }); + + try { + await handlerFor(bucket.type)({ + note, + transcriptText: text, + bucket, + artifactRef, + threadAnchor: anchor, + ...(state.pinnedCompanies !== undefined && { + pinnedCompanies: state.pinnedCompanies, + }), + }); + } catch (cause) { + const message = cause instanceof Error ? cause.message : String(cause); + logger.error("Granola handler failed for note {noteId}: {error}", { + noteId, + error: message, + }); + await lifecycle + .onHandlerFailed({ anchor, error: message }) + .catch((hookCause: unknown) => { + logger.error( + "Granola handler-failed hook failed for note {noteId}: {error}", + { + noteId, + error: + hookCause instanceof Error + ? hookCause.message + : String(hookCause), + }, + ); + }); + } + } + + const ingest = async (payload: GranolaWebhookPayload): Promise => { + await processNote(payload.note_id, payload.event_type, false); + }; + // Companion entry for a host's Reprocess affordance: re-enters the + // pipeline bypassing the already-processed gate — the human just + // answered it. + ingest.reprocess = async (noteId: string): Promise => { + await processNote(noteId, "operator.reprocess", false, { + forceReprocess: true, + }); + }; + // A host's ambiguity-resolution re-entry: same forced reprocess, with the + // chosen companies pinned through to the bucket handler's context. + ingest.reprocessPinned = async ( + noteId: string, + companies: string[], + ): Promise => { + await processNote(noteId, "operator.pinned-diligence", false, { + forceReprocess: true, + pinnedCompanies: companies, + }); + }; + return ingest; +}