From 78f80d1d06cc5a8baa46f7d45aeea7d0e3af0eee Mon Sep 17 00:00:00 2001 From: Jonas Jesus Date: Tue, 27 Oct 2026 17:17:10 -0300 Subject: [PATCH] =?UTF-8?q?refactor(grain):=20live=20Grain=20API=20only=20?= =?UTF-8?q?=E2=80=94=20the=20index=20moved=20to=20grain-indexer?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The webhook-fed index (grain_meeting_recordings in Supabase), its search tool, the trigger storage and the backfill script are gone. They now live, with full transcripts, privacy rules and signed webhooks, in decocms/grain-indexer — and the old tables were dropped there. What stays: LIST_RECORDINGS, GET_RECORDING, GET_TRANSCRIPT, GET_SUMMARY, live against Grain with the connection's own key. No state, no webhooks created on anyone's Grain account, no Supabase dependency. Co-Authored-By: Claude Opus 5.5 (1M context) --- bun.lock | 1 - grain/README.md | 124 ++-------- grain/package.json | 1 - grain/scripts/backfill.ts | 119 ---------- grain/server/constants.ts | 10 - grain/server/db/queries.ts | 122 ---------- grain/server/db/schema.sql | 111 --------- grain/server/lib/grain-client.ts | 33 --- grain/server/lib/supabase-client.ts | 105 --------- grain/server/lib/types.ts | 58 ----- grain/server/main.ts | 215 +----------------- grain/server/tools/index.ts | 4 - .../server/tools/search-indexed-recordings.ts | 105 --------- grain/server/triggers/publisher.ts | 31 --- grain/server/triggers/store.ts | 66 ------ grain/server/types/env.ts | 10 +- 16 files changed, 29 insertions(+), 1086 deletions(-) delete mode 100644 grain/scripts/backfill.ts delete mode 100644 grain/server/db/queries.ts delete mode 100644 grain/server/db/schema.sql delete mode 100644 grain/server/lib/supabase-client.ts delete mode 100644 grain/server/tools/search-indexed-recordings.ts delete mode 100644 grain/server/triggers/publisher.ts delete mode 100644 grain/server/triggers/store.ts diff --git a/bun.lock b/bun.lock index 27b37f56..25e0dccb 100644 --- a/bun.lock +++ b/bun.lock @@ -651,7 +651,6 @@ "dependencies": { "@decocms/bindings": "1.0.1-alpha.23", "@decocms/runtime": "1.3.1", - "@supabase/supabase-js": "^2.57.4", "zod": "^4.0.0", }, "devDependencies": { diff --git a/grain/README.md b/grain/README.md index 238be1d1..1d499a8f 100644 --- a/grain/README.md +++ b/grain/README.md @@ -1,115 +1,35 @@ # Grain MCP -Grain MCP is maintained by Deco and built on top of Grain public APIs. It gives AI agents structured access to meeting recordings, transcripts, and AI summaries, and keeps a searchable Supabase index updated by webhooks. +Grain MCP is maintained by deco and built on Grain's public API (v2). It gives +AI agents live access to the meeting recordings **your Grain API key can see**: +list them, get one with its summary, participants and highlights, or fetch just +the transcript or just the summary (fewer tokens). -## What This MCP Does +It keeps no state: no index, no webhooks, no database. Each call goes to +Grain with the key of the connection that made it. -- Lists recordings directly from Grain -- Retrieves full recording details by ID -- Retrieves transcript-only output to reduce token usage -- Retrieves summary-only output to reduce token usage -- Indexes webhook events into Supabase for fast local search +> **Searching what was said across deco's calls, transcripts in a database, +> call events and the account map** live in +> [decocms/grain-indexer](https://github.com/decocms/grain-indexer) — the +> index this MCP used to keep (a `grain_meeting_recordings` table fed by +> webhooks) moved there, with full transcripts, privacy rules and signed +> webhooks. ## Authentication -This MCP uses a Grain API key only (no OAuth flow). +1. Generate an API key at [Grain → Integrations → API](https://grain.com/app/settings/integrations?tab=api). +2. In the connection settings, paste it in the **Token** field. -1. Generate your API key at [Grain API Integrations](https://grain.com/app/settings/integrations?tab=api) -2. In Deco Mesh connection settings, paste the API key in the **Token** field -3. Save the connection - -The MCP reads this value from `MESH_REQUEST_CONTEXT.authorization`. - -## Webhook URL - -The webhook is created automatically by API when the connection configuration changes. -This is done in `onChange` in `server/main.ts`: - -- The MCP calls Grain webhooks endpoints (`listWebhooks` and `createWebhook`) -- It checks if the webhook already exists -- It creates the webhook only when missing -- The create payload includes `hook_type: "recording_added"` and `hook_url` - -If you need to configure it manually in Grain, use this URL pattern: - -`https://sites-grain.deco.site/webhooks/grain/{connectionId}` - -- Replace `{connectionId}` with your Deco Mesh connection ID -- This is a dedicated public webhook route (no `/mcp` in the external URL) -- Internally, the route forwards the payload to the `MESH_PUBLIC_GRAIN_WEBHOOK` tool -- The MCP also exposes a readonly `WEBHOOK_URL` template in configuration - -In runtime, the MCP automatically builds this URL from Mesh context: - -`{meshUrl}/webhooks/grain/{connectionId}` - -If `DEVELOPMENT_MODE=true`, the MCP uses this fixed base URL instead of `meshUrl`: - -`https://localhost-c056dce8.deco.host/webhooks/grain/{connectionId}` - -Optional development override: - -- `DEVELOPMENT_WEBHOOK_URL` (full URL) -- When set together with `DEVELOPMENT_MODE=true`, this value is used as-is without URL reconstruction - -In Grain, webhook management is available in the same Integrations/API area where the API key is generated: -[Grain API Integrations](https://grain.com/app/settings/integrations?tab=api) - -## Supabase Indexing - -The MCP listens to Grain webhook events and stores indexed data in Supabase table `grain_meeting_recordings`. - -- Webhook setup is automatic during configuration -- New or updated recordings are upserted in Supabase -- `SEARCH_INDEXED_RECORDINGS` queries this local index for fast filtering - -Create both tables manually in Supabase using [`server/db/schema.sql`](server/db/schema.sql): -- `grain_meeting_recordings` — indexed recording data -- `grain_trigger_credentials` — persists Mesh trigger callbacks across restarts - -## Triggers - -Two Mesh triggers are available: - -| Trigger | Fires when | -| --- | --- | -| `grain.recording.added` | A new meeting recording is indexed | -| `grain.recording.updated` | An existing recording is updated (e.g. AI notes ready) | - -Both triggers support the same filter params: - -| Param | Description | -| --- | --- | -| `owner_email` | Only fire for recordings owned by this email | -| `tag` | Only fire when the recording has this exact tag | -| `participant_email` | Only fire when this participant was in the meeting | - -Trigger payload fields: `recording_id`, `title`, `url`, `start_datetime`, `end_datetime`, `owners`, `tags`, `participants` (`[{name, email, scope}]`), `has_summary`, `indexed_at`. +The MCP reads it from `MESH_REQUEST_CONTEXT.authorization`. ## Tools | Tool | Purpose | | --- | --- | -| `LIST_RECORDINGS` | List and filter recordings from Grain API | -| `GET_RECORDING` | Get full details for one recording, with optional selective output (`summary`, `transcript`, `highlights`) | -| `GET_TRANSCRIPT` | Get transcript URL and optional transcript content (`json`, `txt`, `srt`, `vtt`) | -| `GET_SUMMARY` | Get only summary fields (`summary`, `summary_points`, `intelligence_notes_md`) | -| `SEARCH_INDEXED_RECORDINGS` | Search Supabase-indexed recordings by query and/or date filters | - -## Environment Variables - -Supabase access is required for indexed search: - -- `SUPABASE_URL` -- `SUPABASE_ANON_KEY` - -## Event Flow - -```text -Grain webhook -> Deco Mesh event -> Grain MCP event handler -> Supabase upsert -``` - -The MCP also emits `grain.recording_indexed` after a successful index operation when Mesh org context is available. +| `LIST_RECORDINGS` | List and filter recordings from the Grain API | +| `GET_RECORDING` | One recording, with optional selective output (`summary`, `transcript`, `highlights`) | +| `GET_TRANSCRIPT` | Transcript URL and optional content (`json`, `txt`, `srt`, `vtt`) | +| `GET_SUMMARY` | Only the summary fields (`summary`, `summary_points`, `intelligence_notes_md`) | ## Development @@ -120,8 +40,4 @@ bun run check bun run build ``` -## Notes - -- Grain API base URL: `https://api.grain.com` -- This project is part of the Deco MCP monorepo - +Grain API base URL: `https://api.grain.com/_/public-api/v2`. diff --git a/grain/package.json b/grain/package.json index 60359cbd..39475e2e 100644 --- a/grain/package.json +++ b/grain/package.json @@ -15,7 +15,6 @@ "dependencies": { "@decocms/bindings": "1.0.1-alpha.23", "@decocms/runtime": "1.3.1", - "@supabase/supabase-js": "^2.57.4", "zod": "^4.0.0" }, "devDependencies": { diff --git a/grain/scripts/backfill.ts b/grain/scripts/backfill.ts deleted file mode 100644 index d5ef2d16..00000000 --- a/grain/scripts/backfill.ts +++ /dev/null @@ -1,119 +0,0 @@ -#!/usr/bin/env bun -/** - * Backfill — indexes Grain recordings into grain_meeting_recordings (v2 API), - * with participants and the AI summary, the same row the webhook writes. - * Existing rows are updated. - * - * Usage: GRAIN_TOKEN=… SUPABASE_URL=… SUPABASE_SERVICE_ROLE_KEY=… \ - * bun run scripts/backfill.ts [--since 2026-09-30] - */ -import { createClient } from "@supabase/supabase-js"; -import { - GRAIN_API_VERSION, - GRAIN_BASE_URL, - INDEX_INCLUDE, -} from "../server/constants.ts"; -import { type GrainRecordingV2, toDetails } from "../server/lib/types.ts"; - -const GRAIN_TOKEN = process.env.GRAIN_TOKEN; -const SUPABASE_URL = process.env.SUPABASE_URL; -const SUPABASE_KEY = - process.env.SUPABASE_SERVICE_ROLE_KEY ?? process.env.SUPABASE_ANON_KEY; - -if (!GRAIN_TOKEN || !SUPABASE_URL || !SUPABASE_KEY) { - console.error( - "Missing env vars: GRAIN_TOKEN, SUPABASE_URL, SUPABASE_SERVICE_ROLE_KEY", - ); - process.exit(1); -} - -const sinceArg = process.argv.indexOf("--since"); -const since = sinceArg > 0 ? process.argv[sinceArg + 1] : undefined; - -const supabase = createClient(SUPABASE_URL, SUPABASE_KEY, { - auth: { persistSession: false, autoRefreshToken: false }, -}); - -async function fetchPage( - cursor?: string, -): Promise<{ recordings: GrainRecordingV2[]; cursor: string | null }> { - const res = await fetch(`${GRAIN_BASE_URL}/recordings`, { - method: "POST", - headers: { - Authorization: `Bearer ${GRAIN_TOKEN}`, - "Public-Api-Version": GRAIN_API_VERSION, - "Content-Type": "application/json", - }, - body: JSON.stringify({ - ...(cursor ? { cursor } : {}), - filter: since ? { after_datetime: since } : {}, - include: INDEX_INCLUDE, - }), - }); - if (!res.ok) - throw new Error(`Grain API error: ${res.status} ${await res.text()}`); - return res.json(); -} - -async function upsertBatch(recordings: GrainRecordingV2[], userId: string) { - const now = new Date().toISOString(); - const rows = recordings.map((r) => { - const d = toDetails(r); - const participants = d.participants ?? []; - return { - id: d.id, - title: d.title ?? null, - url: d.url ?? null, - start_datetime: d.start_datetime ?? null, - end_datetime: d.end_datetime ?? null, - public_thumbnail_url: d.public_thumbnail_url ?? null, - owners: d.owners ?? [], - tags: d.tags ?? [], - participants, - participants_text: participants - .map((p) => [p.name, p.email].filter(Boolean).join(" ")) - .join(", "), - intelligence_notes_md: d.intelligence_notes_md ?? null, - user_id: userId, - webhook_type: "backfill", - raw_payload: { type: "backfill", user_id: userId, data: r }, - indexed_at: now, - updated_at: now, - }; - }); - const { error } = await supabase - .from("grain_meeting_recordings") - .upsert(rows, { onConflict: "id" }); - if (error) throw new Error(`Supabase upsert error: ${error.message}`); -} - -async function main() { - const who = await fetch(`${GRAIN_BASE_URL}/token-info`, { - headers: { - Authorization: `Bearer ${GRAIN_TOKEN}`, - "Public-Api-Version": GRAIN_API_VERSION, - }, - }).then((r) => r.json() as Promise<{ user_id: string; name: string }>); - console.log(`Backfilling as ${who.name}${since ? ` since ${since}` : ""}…`); - - let cursor: string | undefined; - let total = 0; - for (let page = 1; ; page++) { - const { recordings, cursor: next } = await fetchPage(cursor); - if (recordings.length === 0) break; - await upsertBatch(recordings, who.user_id); - total += recordings.length; - console.log( - `Page ${page}: ${recordings.length} [${recordings.at(-1)?.start_datetime} → ${recordings[0]?.start_datetime}]`, - ); - if (!next) break; - cursor = next; - await new Promise((r) => setTimeout(r, 200)); - } - console.log(`Done. Indexed ${total} recordings.`); -} - -main().catch((err) => { - console.error("Backfill failed:", err); - process.exit(1); -}); diff --git a/grain/server/constants.ts b/grain/server/constants.ts index 276eacee..e29161bd 100644 --- a/grain/server/constants.ts +++ b/grain/server/constants.ts @@ -7,16 +7,6 @@ export const GRAIN_BASE_URL = "https://api.grain.com/_/public-api/v2"; export const GRAIN_API_VERSION = "2026-10-01"; export const GRAIN_RECORDINGS_ENDPOINT = "/recordings"; -export const GRAIN_HOOKS_ENDPOINT = "/hooks"; - -/** Hook types the index listens to. */ -export const INDEXED_HOOK_TYPES = [ - "recording_added", - "recording_updated", -] as const; - -/** What every indexed recording is fetched with. */ -export const INDEX_INCLUDE = { participants: true, ai_summary: true } as const; export const DEFAULT_PAGE_SIZE = 50; export const MAX_PAGE_SIZE = 100; diff --git a/grain/server/db/queries.ts b/grain/server/db/queries.ts deleted file mode 100644 index dd8e99f7..00000000 --- a/grain/server/db/queries.ts +++ /dev/null @@ -1,122 +0,0 @@ -import type { PostgrestError } from "@supabase/supabase-js"; -import { getSupabaseClient } from "../lib/supabase-client.ts"; -import type { - WebhookPayload, - RecordingDetails, - RecordingParticipant, -} from "../lib/types.ts"; - -export interface GrainRecordingRow { - id: string; - title: string | null; - url: string | null; - start_datetime: string | null; - end_datetime: string | null; - public_thumbnail_url: string | null; - owners: string[]; - tags: string[]; - participants: RecordingParticipant[]; - participants_text: string | null; - intelligence_notes_md: string | null; - user_id: string | null; - webhook_type: string; - raw_payload: WebhookPayload; - indexed_at: string; - updated_at: string; -} - -function throwIfError(error: PostgrestError | null, action: string): void { - if (error) { - throw new Error(`Failed to ${action}: ${error.message}`); - } -} - -function buildParticipantsText(participants: RecordingParticipant[]): string { - return participants - .map((p) => [p.name, p.email].filter(Boolean).join(" ")) - .join(", "); -} - -export interface EnrichedWebhookData { - payload: WebhookPayload; - /** Fetched from the Grain API — never taken from the (unsigned) webhook body. */ - details: RecordingDetails; -} - -function toDbRow({ payload, details }: EnrichedWebhookData): GrainRecordingRow { - const participants = details.participants ?? []; - const now = new Date().toISOString(); - return { - id: details.id, - title: details.title ?? null, - url: details.url ?? null, - start_datetime: details.start_datetime ?? null, - end_datetime: details.end_datetime ?? null, - public_thumbnail_url: details.public_thumbnail_url ?? null, - owners: details.owners ?? [], - tags: details.tags ?? [], - participants, - participants_text: buildParticipantsText(participants), - intelligence_notes_md: details.intelligence_notes_md ?? null, - user_id: payload.user_id ?? null, - webhook_type: payload.type, - raw_payload: payload, - indexed_at: now, - updated_at: now, - }; -} - -export async function indexRecording(data: EnrichedWebhookData): Promise { - const client = getSupabaseClient(); - const row = toDbRow(data); - - const { error } = await client - .from("grain_meeting_recordings") - .upsert(row, { onConflict: "id" }); - - throwIfError(error, "upsert indexed recording"); -} - -export interface IndexedRecordingsSearchFilters { - query?: string; - startDate?: string; - endDate?: string; - owner?: string; - tag?: string; - limit: number; -} - -export async function searchIndexedRecordings( - filters: IndexedRecordingsSearchFilters, -): Promise { - const client = getSupabaseClient(); - let qb = client - .from("grain_meeting_recordings") - .select("*") - .order("start_datetime", { ascending: false }) - .limit(filters.limit); - - if (filters.startDate) { - qb = qb.gte("start_datetime", filters.startDate); - } - if (filters.endDate) { - qb = qb.lte("start_datetime", filters.endDate); - } - if (filters.tag) { - qb = qb.contains("tags", [filters.tag]); - } - if (filters.owner) { - qb = qb.contains("owners", [filters.owner]); - } - if (filters.query && filters.query.trim().length > 0) { - // Remove characters that have special meaning in PostgREST filter syntax - const q = filters.query.trim().replace(/[,.*()\\%]/g, ""); - if (q.length > 0) { - qb = qb.or(`title.ilike.%${q}%,participants_text.ilike.%${q}%`); - } - } - - const { data, error } = await qb; - throwIfError(error, "search indexed recordings"); - return (data ?? []) as GrainRecordingRow[]; -} diff --git a/grain/server/db/schema.sql b/grain/server/db/schema.sql deleted file mode 100644 index ef3c1edf..00000000 --- a/grain/server/db/schema.sql +++ /dev/null @@ -1,111 +0,0 @@ --- Grain Meeting Recordings index table --- Run this manually in Supabase SQL Editor --- To migrate an existing table without data loss: --- ALTER TABLE grain_recordings RENAME TO grain_meeting_recordings; --- ALTER TABLE grain_meeting_recordings ALTER COLUMN user_id DROP NOT NULL; - -CREATE EXTENSION IF NOT EXISTS pg_trgm; - --- ============================================================================ --- Recording index table --- ============================================================================ - -DROP TABLE IF EXISTS grain_meeting_recordings; - -CREATE TABLE grain_meeting_recordings ( - id TEXT PRIMARY KEY, - title TEXT, - url TEXT, - start_datetime TIMESTAMPTZ, - end_datetime TIMESTAMPTZ, - public_thumbnail_url TEXT, - owners TEXT[] DEFAULT '{}', - tags TEXT[] DEFAULT '{}', - participants JSONB DEFAULT '[]', - participants_text TEXT, - intelligence_notes_md TEXT, - user_id TEXT, - webhook_type TEXT NOT NULL, - raw_payload JSONB NOT NULL, - indexed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), - updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() -); - --- Fast date-range filtering -CREATE INDEX idx_grain_meeting_rec_start - ON grain_meeting_recordings (start_datetime DESC); - --- Fuzzy title search -CREATE INDEX idx_grain_meeting_rec_title_trgm - ON grain_meeting_recordings USING gin (title gin_trgm_ops); - --- Search inside participants text (names + emails flattened) -CREATE INDEX idx_grain_meeting_rec_participants_trgm - ON grain_meeting_recordings USING gin (participants_text gin_trgm_ops); - --- Search inside AI notes -CREATE INDEX idx_grain_meeting_rec_notes_trgm - ON grain_meeting_recordings USING gin (intelligence_notes_md gin_trgm_ops); - --- Filter by tag -CREATE INDEX idx_grain_meeting_rec_tags - ON grain_meeting_recordings USING gin (tags); - --- Filter by owner -CREATE INDEX idx_grain_meeting_rec_owners - ON grain_meeting_recordings USING gin (owners); - -ALTER TABLE grain_meeting_recordings ENABLE ROW LEVEL SECURITY; - -CREATE POLICY "grain_meeting_anon_select" - ON grain_meeting_recordings FOR SELECT - TO anon - USING (true); - -CREATE POLICY "grain_meeting_anon_insert" - ON grain_meeting_recordings FOR INSERT - TO anon - WITH CHECK (true); - -CREATE POLICY "grain_meeting_anon_update" - ON grain_meeting_recordings FOR UPDATE - TO anon - USING (true) - WITH CHECK (true); - --- ============================================================================ --- Trigger credentials table (persists TRIGGER_CONFIGURE state across restarts) --- ============================================================================ - -DROP TABLE IF EXISTS grain_trigger_credentials; - -CREATE TABLE grain_trigger_credentials ( - connection_id TEXT PRIMARY KEY, - callback_url TEXT NOT NULL, - callback_token TEXT NOT NULL, - active_trigger_types TEXT[] NOT NULL DEFAULT '{}', - updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() -); - -ALTER TABLE grain_trigger_credentials ENABLE ROW LEVEL SECURITY; - -CREATE POLICY "grain_trigger_anon_select" - ON grain_trigger_credentials FOR SELECT - TO anon - USING (true); - -CREATE POLICY "grain_trigger_anon_insert" - ON grain_trigger_credentials FOR INSERT - TO anon - WITH CHECK (true); - -CREATE POLICY "grain_trigger_anon_update" - ON grain_trigger_credentials FOR UPDATE - TO anon - USING (true) - WITH CHECK (true); - -CREATE POLICY "grain_trigger_anon_delete" - ON grain_trigger_credentials FOR DELETE - TO anon - USING (true); diff --git a/grain/server/lib/grain-client.ts b/grain/server/lib/grain-client.ts index db301288..52c99596 100644 --- a/grain/server/lib/grain-client.ts +++ b/grain/server/lib/grain-client.ts @@ -1,14 +1,10 @@ import { GRAIN_API_VERSION, GRAIN_BASE_URL, - GRAIN_HOOKS_ENDPOINT, GRAIN_RECORDINGS_ENDPOINT, } from "../constants.ts"; import { type GrainRecordingV2, - type Hook, - type HookType, - type ListHooksResponse, type ListRecordingsParams, type ListRecordingsResponse, type RecordingDetails, @@ -179,33 +175,4 @@ export class GrainClient { end_time: s.end, })); } - - /** POST /v2/hooks (list). */ - async listHooks(): Promise { - return this.request("POST", GRAIN_HOOKS_ENDPOINT, {}); - } - - /** - * POST /v2/hooks/create. Grain sends a reachability POST to `hookUrl` - * first; it must answer 2xx or the hook is not created. - */ - async createHook( - hookUrl: string, - hookType: HookType, - include?: RecordingsInclude, - ): Promise { - return this.request("POST", `${GRAIN_HOOKS_ENDPOINT}/create`, { - hook_url: hookUrl, - hook_type: hookType, - ...(include ? { include } : {}), - }); - } - - /** DELETE /v2/hooks/:id */ - async deleteHook(hookId: string): Promise { - await this.request<{ success: boolean }>( - "DELETE", - `${GRAIN_HOOKS_ENDPOINT}/${hookId}`, - ); - } } diff --git a/grain/server/lib/supabase-client.ts b/grain/server/lib/supabase-client.ts deleted file mode 100644 index 3cc9c39b..00000000 --- a/grain/server/lib/supabase-client.ts +++ /dev/null @@ -1,105 +0,0 @@ -import { createClient, type SupabaseClient } from "@supabase/supabase-js"; - -let supabaseClient: SupabaseClient | null = null; - -export function getSupabaseClient(): SupabaseClient { - if (supabaseClient) { - return supabaseClient; - } - - const supabaseUrl = process.env.SUPABASE_URL; - // Server-side key: the tables have RLS on and no anon access. The anon key - // stays as a fallback only for local setups that still use it. - const supabaseKey = - process.env.SUPABASE_SERVICE_ROLE_KEY ?? process.env.SUPABASE_ANON_KEY; - - if (!supabaseUrl || !supabaseKey) { - throw new Error( - "SUPABASE_URL and SUPABASE_SERVICE_ROLE_KEY are required to use indexed recording search.", - ); - } - - supabaseClient = createClient(supabaseUrl, supabaseKey, { - auth: { - persistSession: false, - autoRefreshToken: false, - }, - }); - - return supabaseClient; -} - -// ============================================================================ -// TRIGGER CREDENTIALS (grain_trigger_credentials table) -// ============================================================================ - -interface TriggerCredentialsRow { - connection_id: string; - callback_url: string; - callback_token: string; - active_trigger_types: string[]; - updated_at: string; -} - -export interface TriggerState { - credentials: { callbackUrl: string; callbackToken: string }; - activeTriggerTypes: string[]; -} - -export async function saveTriggerCredentials( - connectionId: string, - state: TriggerState, -): Promise { - const client = getSupabaseClient(); - const { error } = await client.from("grain_trigger_credentials").upsert( - { - connection_id: connectionId, - callback_url: state.credentials.callbackUrl, - callback_token: state.credentials.callbackToken, - active_trigger_types: state.activeTriggerTypes, - updated_at: new Date().toISOString(), - } as TriggerCredentialsRow, - { onConflict: "connection_id" }, - ); - if (error) { - throw new Error(`Failed to save trigger credentials: ${error.message}`); - } -} - -export async function loadTriggerCredentials( - connectionId: string, -): Promise { - const client = getSupabaseClient(); - const { data, error } = await client - .from("grain_trigger_credentials") - .select("*") - .eq("connection_id", connectionId) - .single(); - - if (error) { - if (error.code === "PGRST116") return null; - throw new Error(`Failed to load trigger credentials: ${error.message}`); - } - - const row = data as TriggerCredentialsRow; - return { - credentials: { - callbackUrl: row.callback_url, - callbackToken: row.callback_token, - }, - activeTriggerTypes: row.active_trigger_types, - }; -} - -export async function deleteTriggerCredentials( - connectionId: string, -): Promise { - const client = getSupabaseClient(); - const { error } = await client - .from("grain_trigger_credentials") - .delete() - .eq("connection_id", connectionId); - if (error) { - throw new Error(`Failed to delete trigger credentials: ${error.message}`); - } -} diff --git a/grain/server/lib/types.ts b/grain/server/lib/types.ts index a6aca2ae..b45ff8fc 100644 --- a/grain/server/lib/types.ts +++ b/grain/server/lib/types.ts @@ -1,5 +1,3 @@ -import { z } from "zod"; - /** * Grain Public API v2 types * @see https://developers.grain.com/endpoints.html @@ -130,59 +128,3 @@ export function toDetails(r: GrainRecordingV2): RecordingDetails { })), }; } - -// -- Hooks (v2) -- - -export type HookType = - | "recording_added" - | "recording_updated" - | "recording_deleted" - | "highlight_added" - | "highlight_updated" - | "highlight_deleted" - | "story_added" - | "story_updated" - | "story_deleted" - | "upload_added" - | "upload_updated"; - -export interface Hook { - id: string; - enabled: boolean; - hook_url: string; - hook_type: HookType; - include: Record; - inserted_at: string; -} - -export interface ListHooksResponse { - hooks: Hook[]; -} - -// -- Webhook payloads (inbound from Grain, v2 envelope) -- - -export type WebhookEventType = HookType; - -export interface WebhookPayload { - type: WebhookEventType; - user_id?: string; - data: { id: string; [key: string]: unknown }; -} - -export const WebhookPayloadSchema = z.object({ - type: z.enum([ - "recording_added", - "recording_updated", - "recording_deleted", - "highlight_added", - "highlight_updated", - "highlight_deleted", - "story_added", - "story_updated", - "story_deleted", - "upload_added", - "upload_updated", - ]), - user_id: z.string().optional(), - data: z.object({ id: z.string() }).passthrough(), -}); diff --git a/grain/server/main.ts b/grain/server/main.ts index 04d93b3f..b4ed3d7b 100644 --- a/grain/server/main.ts +++ b/grain/server/main.ts @@ -2,217 +2,18 @@ import { withRuntime } from "@decocms/runtime"; import { serve } from "@decocms/mcps-shared/serve"; import { tools } from "./tools/index.ts"; -import { GrainClient } from "./lib/grain-client.ts"; -import { INDEX_INCLUDE, INDEXED_HOOK_TYPES } from "./constants.ts"; -import { getGrainApiKey } from "./lib/env.ts"; -import { WebhookPayloadSchema } from "./lib/types.ts"; -import type { RecordingDetails } from "./lib/types.ts"; -import { indexRecording } from "./db/queries.ts"; -import { publishRecordingTrigger } from "./triggers/publisher.ts"; import type { Env } from "./types/env.ts"; import { StateSchema } from "./types/env.ts"; -const DEVELOPMENT_WEBHOOK_BASE_URL = "https://localhost-c056dce8.deco.host"; -const WEBHOOK_PUBLIC_PATH = "/webhooks/grain"; - -let cachedGrainClient: GrainClient | null = null; - -function buildWebhookUrl(baseUrl: string, connectionId: string): string { - return `${baseUrl.replace(/\/+$/, "")}${WEBHOOK_PUBLIC_PATH}/${connectionId}`; -} - +/** + * Grain's public API, live, with the caller's own key: list, get, transcript, + * summary. No index, no webhooks, no state. Searching what was said across + * deco's calls, transcripts in a database and call events live in + * decocms/grain-indexer. + */ const runtime = withRuntime({ - configuration: { - onChange: async (env) => { - try { - const grainToken = getGrainApiKey(env); - const meshUrl = env.MESH_REQUEST_CONTEXT?.meshUrl; - const connectionId = env.MESH_REQUEST_CONTEXT?.connectionId; - - cachedGrainClient = new GrainClient({ apiKey: grainToken }); - - if (!meshUrl || !connectionId) { - console.error("[GRAIN_MCP] Missing meshUrl or connectionId"); - return; - } - - const isDev = process.env.DEVELOPMENT_MODE === "true"; - const devOverride = process.env.DEVELOPMENT_WEBHOOK_URL; - const webhookUrl = - isDev && devOverride - ? devOverride - : buildWebhookUrl( - isDev ? DEVELOPMENT_WEBHOOK_BASE_URL : meshUrl, - connectionId, - ); - - console.log("[GRAIN_MCP] Setting up webhook", { - webhookUrl, - isDevelopmentMode: isDev, - }); - - const { hooks } = await cachedGrainClient.listHooks(); - - // Our hooks pointing somewhere else (old mesh URL, dev tunnel): drop them. - const staleHooks = hooks.filter( - (h) => - h.hook_url.includes(WEBHOOK_PUBLIC_PATH) && - h.hook_url !== webhookUrl, - ); - await Promise.all( - staleHooks.map((stale) => - cachedGrainClient!.deleteHook(stale.id).then(() => - console.log("[GRAIN_MCP] Deleted stale webhook", { - hookId: stale.id, - url: stale.hook_url, - }), - ), - ), - ); - - // v2 hooks are one per type; make sure each indexed type has one. - for (const hookType of INDEXED_HOOK_TYPES) { - if ( - hooks.some( - (h) => - h.hook_url === webhookUrl && - h.hook_type === hookType && - h.enabled, - ) - ) - continue; - const hook = await cachedGrainClient.createHook( - webhookUrl, - hookType, - INDEX_INCLUDE, - ); - console.log("[GRAIN_MCP] Hook created", { - hookId: hook.id, - hookType, - }); - } - } catch (error) { - console.error("[GRAIN_MCP] Error setting up webhook", error); - } - }, - state: StateSchema, - }, + configuration: { state: StateSchema }, tools, }); -/** - * The client for webhook deliveries: the one from the last configuration, or - * GRAIN_API_KEY after a restart (onChange only runs when the connection is - * saved, so an in-memory client alone loses every webhook after a redeploy). - */ -function webhookClient(): GrainClient | null { - if (cachedGrainClient) return cachedGrainClient; - const key = process.env.GRAIN_API_KEY; - return key ? (cachedGrainClient = new GrainClient({ apiKey: key })) : null; -} - -/** - * Re-reads the recording from Grain. Webhooks are not signed, so the payload - * only names *which* recording changed; what gets indexed comes from the API. - */ -async function fetchRecording( - recordingId: string, -): Promise { - const client = webhookClient(); - if (!client) { - console.warn("[GRAIN_MCP] No Grain client (set GRAIN_API_KEY), skipping"); - return null; - } - try { - return await client.getRecording(recordingId, INDEX_INCLUDE); - } catch (err) { - console.warn("[GRAIN_MCP] Failed to fetch recording, not indexing", { - recordingId, - error: err instanceof Error ? err.message : String(err), - }); - return null; - } -} - -async function handleWebhookPost( - body: string, - connectionId: string, -): Promise { - if (!body || body.trim() === "" || body.trim() === "{}") { - return new Response("ok", { status: 200 }); - } - - let json: unknown; - try { - json = JSON.parse(body); - } catch { - return new Response("ok", { status: 200 }); - } - - const parsed = WebhookPayloadSchema.safeParse(json); - if (!parsed.success) { - console.warn("[GRAIN_MCP] Webhook validation failed", { - error: parsed.error.format(), - type: (json as Record)?.type, - }); - return new Response("ok", { status: 200 }); - } - - const payload = parsed.data; - console.log("[GRAIN_MCP] Received webhook", { - type: payload.type, - recordingId: payload.data.id, - }); - - if ( - payload.type === "recording_added" || - payload.type === "recording_updated" - ) { - try { - const details = await fetchRecording(payload.data.id); - if (!details) return new Response("ok", { status: 200 }); - - await indexRecording({ payload, details }); - - console.log("[GRAIN_MCP] Recording indexed", { - id: payload.data.id, - participants: details?.participants?.length ?? 0, - hasTags: (details?.tags?.length ?? 0) > 0, - hasNotes: !!details?.intelligence_notes_md, - }); - - publishRecordingTrigger(connectionId, payload.type, details, payload); - } catch (indexError) { - console.error("[GRAIN_MCP] Failed to index recording", indexError); - } - } - - return new Response(JSON.stringify({ success: true }), { - status: 200, - headers: { "content-type": "application/json" }, - }); -} - -serve(async (request, env, ctx) => { - const url = new URL(request.url); - - if (url.pathname.startsWith(`${WEBHOOK_PUBLIC_PATH}/`)) { - const connectionId = url.pathname.slice(`${WEBHOOK_PUBLIC_PATH}/`.length); - if (!connectionId) { - return new Response("Missing connection id", { status: 400 }); - } - - if (request.method === "GET" || request.method === "HEAD") { - return new Response("ok", { status: 200 }); - } - - if (request.method === "POST") { - const body = await request.text(); - return handleWebhookPost(body, connectionId); - } - - return new Response("Method not allowed", { status: 405 }); - } - - return runtime.fetch(request, env, ctx); -}); +serve(runtime.fetch); diff --git a/grain/server/tools/index.ts b/grain/server/tools/index.ts index 35328eac..5600caba 100644 --- a/grain/server/tools/index.ts +++ b/grain/server/tools/index.ts @@ -1,15 +1,11 @@ import { createListRecordingsTool } from "./list-recordings.ts"; import { createGetRecordingTool } from "./get-recording.ts"; -import { createSearchIndexedRecordingsTool } from "./search-indexed-recordings.ts"; import { createGetTranscriptTool } from "./get-transcript.ts"; import { createGetSummaryTool } from "./get-summary.ts"; -import { triggers } from "../triggers/store.ts"; export const tools = [ createListRecordingsTool, createGetRecordingTool, createGetTranscriptTool, createGetSummaryTool, - createSearchIndexedRecordingsTool, - () => triggers.tools(), ]; diff --git a/grain/server/tools/search-indexed-recordings.ts b/grain/server/tools/search-indexed-recordings.ts deleted file mode 100644 index 011edd95..00000000 --- a/grain/server/tools/search-indexed-recordings.ts +++ /dev/null @@ -1,105 +0,0 @@ -import { createPrivateTool } from "@decocms/runtime/tools"; -import { z } from "zod"; -import { searchIndexedRecordings } from "../db/queries.ts"; -import type { Env } from "../types/env.ts"; - -export const createSearchIndexedRecordingsTool = (env: Env) => - createPrivateTool({ - id: "SEARCH_INDEXED_RECORDINGS", - description: - "Search recordings indexed in Supabase via webhooks. " + - "Text search matches title, participant names/emails, and AI notes. " + - "Can also filter by date range, tag, or owner email.", - inputSchema: z.object({ - query: z - .string() - .optional() - .describe( - "Free-text search across title, participants, and intelligence notes", - ), - start_date: z - .string() - .optional() - .describe("Recordings on or after this date (ISO 8601)"), - end_date: z - .string() - .optional() - .describe("Recordings on or before this date (ISO 8601)"), - tag: z.string().optional().describe("Filter by exact tag"), - owner: z - .string() - .optional() - .describe("Filter by owner email (exact match)"), - limit: z - .number() - .int() - .min(1) - .max(100) - .default(10) - .optional() - .describe("Max results (1-100, default 10)"), - }), - outputSchema: z.object({ - recordings: z.array( - z.object({ - id: z.string(), - title: z.string().nullable(), - url: z.string().nullable(), - start_datetime: z.string().nullable(), - end_datetime: z.string().nullable(), - owners: z.array(z.string()), - tags: z.array(z.string()), - participants: z.array( - z - .object({ - email: z.string().nullable(), - name: z.string().nullable(), - scope: z.string().nullable(), - }) - .passthrough() - .catchall(z.unknown()), - ), - intelligence_notes_md: z.string().nullable(), - indexed_at: z.string(), - }), - ), - count: z.number(), - }), - execute: async ({ context }) => { - void env; - const { query, start_date, end_date, tag, owner, limit = 10 } = context; - - const trimmedQuery = query?.trim() || undefined; - if (!trimmedQuery && !start_date && !end_date && !tag && !owner) { - throw new Error( - "Provide at least one filter: query, start_date, end_date, tag, or owner.", - ); - } - - const recordings = await searchIndexedRecordings({ - query: trimmedQuery, - startDate: start_date, - endDate: end_date, - tag, - owner, - limit, - }); - - return { - recordings: recordings.map((r) => ({ - id: r.id, - title: r.title, - url: r.url, - start_datetime: r.start_datetime, - end_datetime: r.end_datetime, - owners: r.owners, - tags: r.tags, - // eslint-disable-next-line @typescript-eslint/no-explicit-any - participants: r.participants as any, - intelligence_notes_md: r.intelligence_notes_md, - indexed_at: r.indexed_at, - })), - count: recordings.length, - }; - }, - }); diff --git a/grain/server/triggers/publisher.ts b/grain/server/triggers/publisher.ts deleted file mode 100644 index 98a26458..00000000 --- a/grain/server/triggers/publisher.ts +++ /dev/null @@ -1,31 +0,0 @@ -import { triggers } from "./store.ts"; -import type { RecordingDetails, WebhookPayload } from "../lib/types.ts"; - -export function publishRecordingTrigger( - connectionId: string, - webhookType: "recording_added" | "recording_updated", - details: RecordingDetails, - payload: WebhookPayload, -): void { - const triggerType = - webhookType === "recording_added" - ? "grain.recording.added" - : "grain.recording.updated"; - - triggers.notify(connectionId, triggerType, { - recording_id: payload.data.id, - title: details.title ?? null, - url: details.url ?? null, - start_datetime: details.start_datetime ?? null, - end_datetime: details.end_datetime ?? null, - owners: details.owners ?? [], - tags: details.tags ?? [], - participants: (details.participants ?? []).map((p) => ({ - name: p.name, - email: p.email, - scope: p.scope, - })), - has_summary: !!details.intelligence_notes_md, - indexed_at: new Date().toISOString(), - }); -} diff --git a/grain/server/triggers/store.ts b/grain/server/triggers/store.ts deleted file mode 100644 index ac2f7b4b..00000000 --- a/grain/server/triggers/store.ts +++ /dev/null @@ -1,66 +0,0 @@ -import { createTriggers, type TriggerStorage } from "@decocms/runtime/triggers"; -import { - saveTriggerCredentials, - loadTriggerCredentials, - deleteTriggerCredentials, - type TriggerState, -} from "../lib/supabase-client.ts"; -import { z } from "zod"; - -class SupabaseTriggerStorage implements TriggerStorage { - async get(connectionId: string) { - return loadTriggerCredentials(connectionId); - } - - async set(connectionId: string, state: TriggerState) { - await saveTriggerCredentials(connectionId, state); - } - - async delete(connectionId: string) { - try { - await deleteTriggerCredentials(connectionId); - } catch (error) { - console.error( - `[GRAIN_MCP] Failed to delete trigger credentials for ${connectionId}:`, - error instanceof Error ? error.message : String(error), - ); - } - } -} - -const recordingParams = z.object({ - owner_email: z - .string() - .optional() - .describe("Only trigger for recordings owned by this email address"), - tag: z - .string() - .optional() - .describe("Only trigger when the recording has this exact tag"), - participant_email: z - .string() - .optional() - .describe("Only trigger when this participant email was in the meeting"), -}); - -export const triggers = createTriggers({ - definitions: [ - { - type: "grain.recording.added", - description: - "Triggered when a new Grain meeting recording is indexed. " + - "Payload: `recording_id`, `title`, `url`, `start_datetime`, `end_datetime`, " + - "`owners` (email list), `tags`, `participants` (array of {name, email}), " + - "`has_summary` (AI notes available), `indexed_at`.", - params: recordingParams, - }, - { - type: "grain.recording.updated", - description: - "Triggered when an existing Grain meeting recording is updated (e.g. AI notes ready). " + - "Same payload as grain.recording.added.", - params: recordingParams, - }, - ], - storage: new SupabaseTriggerStorage(), -}); diff --git a/grain/server/types/env.ts b/grain/server/types/env.ts index 3c1ac382..a23c0a7b 100644 --- a/grain/server/types/env.ts +++ b/grain/server/types/env.ts @@ -1,14 +1,6 @@ import { type DefaultEnv } from "@decocms/runtime"; import z from "zod"; -export const StateSchema = z.object({ - WEBHOOK_URL: z - .string() - .default("https://sites-grain.decocache.com/webhooks/grain/{connectionId}") - .readonly() - .describe( - "Webhook URL template used by Grain. Replace {connectionId} with your connection ID from the Mesh URL.", - ), -}); +export const StateSchema = z.object({}); export type Env = DefaultEnv;