From 5f634204256348ed3d631e472ad31eec056b8e89 Mon Sep 17 00:00:00 2001 From: ttmx Date: Thu, 16 Jul 2026 12:06:07 +0100 Subject: [PATCH 01/39] feat(cloudflare): add AI Search plugin Cloudflare-native AI Search integration: content sync hooks, metadata-based documents with always-on metadata-only retrieval, locale-aware search, a cron-flushed reindex queue, query synonyms, an index-status endpoint, and the settings admin UI. --- packages/cloudflare/package.json | 22 +- .../src/plugins/ai-search-admin.tsx | 510 +++++++ packages/cloudflare/src/plugins/ai-search.ts | 1322 +++++++++++++++++ packages/cloudflare/src/plugins/index.ts | 1 + packages/cloudflare/tsdown.config.ts | 3 + 5 files changed, 1857 insertions(+), 1 deletion(-) create mode 100644 packages/cloudflare/src/plugins/ai-search-admin.tsx create mode 100644 packages/cloudflare/src/plugins/ai-search.ts diff --git a/packages/cloudflare/package.json b/packages/cloudflare/package.json index 3770bc72e0..39728b6c9b 100644 --- a/packages/cloudflare/package.json +++ b/packages/cloudflare/package.json @@ -65,6 +65,15 @@ "types": "./dist/plugins/cloudflare-email.d.mts", "default": "./dist/plugins/cloudflare-email.mjs" }, + "./plugins/ai-search": { + "types": "./dist/plugins/ai-search.d.mts", + "default": "./dist/plugins/ai-search.mjs" + }, + "./plugins/ai-search-admin": { + "types": "./dist/plugins/ai-search-admin.d.mts", + "default": "./dist/plugins/ai-search-admin.mjs" + }, + "./media/images-runtime": { "types": "./dist/media/images-runtime.d.mts", "default": "./dist/media/images-runtime.mjs" @@ -103,13 +112,21 @@ "peerDependencies": { "@astrojs/cloudflare": ">=12.0.0", "@cloudflare/workers-types": ">=4.0.0", + "@phosphor-icons/react": "^2.1.10", "astro": ">=6.0.0-beta.0", "kysely": ">=0.28.17", - "pg": ">=8.16.3" + "pg": ">=8.16.3", + "react": "^18.0.0 || ^19.0.0" }, "peerDependenciesMeta": { "pg": { "optional": true + }, + "@phosphor-icons/react": { + "optional": true + }, + "react": { + "optional": true } }, "devDependencies": { @@ -117,7 +134,10 @@ "@astrojs/cloudflare": "catalog:", "@cloudflare/workers-types": "catalog:", "@types/pg": "^8.16.0", + "@phosphor-icons/react": "catalog:", + "@types/react": "catalog:", "publint": "catalog:", + "react": "catalog:", "tsdown": "catalog:", "typescript": "catalog:", "vitest": "catalog:" diff --git a/packages/cloudflare/src/plugins/ai-search-admin.tsx b/packages/cloudflare/src/plugins/ai-search-admin.tsx new file mode 100644 index 0000000000..68f5a032fa --- /dev/null +++ b/packages/cloudflare/src/plugins/ai-search-admin.tsx @@ -0,0 +1,510 @@ +/** + * AI Search Plugin — Admin Components + * + * Settings page with a "Sync All Content" button that triggers + * a full reindex of all configured collections into AI Search. + */ + +import { Badge, Banner, Button, Combobox, Input, Loader } from "@cloudflare/kumo"; +import { + ArrowRight, + ArrowsClockwise, + CheckCircle, + ListMagnifyingGlass, + Plus, + Trash, +} from "@phosphor-icons/react"; +import type { PluginAdminExports } from "emdash"; +import { apiFetch, parseApiResponse } from "emdash/plugin-utils"; +import * as React from "react"; + +const API_BASE = "/_emdash/api/plugins/ai-search"; +const SCHEMA_API_BASE = "/_emdash/api/schema"; + +const indexedLabel = (n: number) => `${n} item${n !== 1 ? "s" : ""}`; + +// ============================================================================= +// Types +// ============================================================================= + +interface ReindexResult { + jobId: string; + status: "running" | "complete"; + done: boolean; + onlyMissing: boolean; + indexed: number; + errors: number; + skipped?: number; + collections: string[]; +} + +interface CollectionOption { + slug: string; + label: string; +} + +interface MissingItem { + id: string; + slug: string | null; + title: string | null; + status: string; +} + +interface CollectionStatus { + collection: string; + eligible: number; + indexed: number; + missing: MissingItem[]; +} + +interface IndexStatus { + instanceName: string; + binding: string; + hybridSearch: boolean; + totalIndexed: number; + collections: CollectionStatus[]; +} + +interface Synonym { + from: string; + to: string; +} + +// ============================================================================= +// Settings Page +// ============================================================================= + +function SettingsPage() { + const [syncMode, setSyncMode] = React.useState<"full" | "missing" | null>(null); + const isSyncing = syncMode !== null; + const [result, setResult] = React.useState(null); + const [error, setError] = React.useState(null); + const [available, setAvailable] = React.useState([]); + const [loadingCollections, setLoadingCollections] = React.useState(true); + const [collectionsError, setCollectionsError] = React.useState(null); + const [selected, setSelected] = React.useState([]); + + const [status, setStatus] = React.useState(null); + const [isCheckingStatus, setIsCheckingStatus] = React.useState(false); + const [statusError, setStatusError] = React.useState(null); + + const [synonyms, setSynonyms] = React.useState([]); + const [isSavingSynonyms, setIsSavingSynonyms] = React.useState(false); + const [synonymsSaved, setSynonymsSaved] = React.useState(false); + const [synonymsError, setSynonymsError] = React.useState(null); + + React.useEffect(() => { + let cancelled = false; + void (async () => { + try { + const [collectionsRes, configRes] = await Promise.all([ + apiFetch(`${SCHEMA_API_BASE}/collections`), + apiFetch(`${API_BASE}/config`), + ]); + const collectionsData = await parseApiResponse<{ items: CollectionOption[] }>( + collectionsRes, + ); + const configData = await parseApiResponse<{ + collections: string[]; + synonyms?: Synonym[]; + }>(configRes); + if (cancelled) return; + + setSynonyms(configData.synonyms ?? []); + + const options = collectionsData.items.map((c) => ({ slug: c.slug, label: c.label })); + setAvailable(options); + + const valid = new Set(options.map((c) => c.slug)); + const saved = (configData.collections ?? []).filter((slug) => valid.has(slug)); + // Restore the last configured selection; fall back to the common + // content collections when nothing has been configured yet. + const initial = + saved.length > 0 + ? saved + : options.filter((c) => c.slug === "posts" || c.slug === "pages").map((c) => c.slug); + setSelected(initial); + } catch (err) { + if (!cancelled) { + setCollectionsError(err instanceof Error ? err.message : "Failed to load collections"); + } + } finally { + if (!cancelled) setLoadingCollections(false); + } + })(); + return () => { + cancelled = true; + }; + }, []); + + const labelForSlug = React.useCallback( + (slug: string) => available.find((c) => c.slug === slug)?.label ?? slug, + [available], + ); + + // Persist the operator's selection so it is restored on the next visit. + const persistSelection = React.useCallback((next: string[]) => { + void apiFetch(`${API_BASE}/config`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ collections: next }), + }).catch(() => { + // Non-critical: selection still applies for this session. + }); + }, []); + + const handleSelectionChange = React.useCallback( + (value: unknown) => { + const next = Array.isArray(value) ? (value as string[]) : []; + setSelected(next); + persistSelection(next); + }, + [persistSelection], + ); + + const runSync = React.useCallback( + async (body: Record, mode: "full" | "missing") => { + setSyncMode(mode); + setError(null); + try { + const response = await apiFetch(`${API_BASE}/reindex`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify(body), + }); + let data = await parseApiResponse(response); + if ("error" in data) throw new Error(data.error); + setResult(data); + + while (!data.done) { + await new Promise((resolve) => setTimeout(resolve, 5_000)); + const statusResponse = await apiFetch(`${API_BASE}/reindex`); + const jobStatus = await parseApiResponse(statusResponse); + if (!jobStatus || jobStatus.jobId !== data.jobId) { + throw new Error("Reindex job not found"); + } + data = jobStatus; + setResult(data); + } + } catch (err) { + setError(err instanceof Error ? err.message : "Sync failed"); + } finally { + setSyncMode(null); + } + }, + [], + ); + + const handleSync = (onlyMissing = false) => { + setResult(null); + void runSync({ collections: selected, onlyMissing }, onlyMissing ? "missing" : "full"); + }; + + React.useEffect(() => { + void (async () => { + try { + const response = await apiFetch(`${API_BASE}/reindex`); + const job = await parseApiResponse(response); + if (job && !job.done) { + await runSync({ jobId: job.jobId }, job.onlyMissing ? "missing" : "full"); + } + } catch { + // A missing prior job is normal. + } + })(); + }, [runSync]); + + const handleCheckStatus = async () => { + setIsCheckingStatus(true); + setStatus(null); + setStatusError(null); + try { + const query = selected.length > 0 ? `?collections=${selected.join(",")}` : ""; + const response = await apiFetch(`${API_BASE}/status${query}`); + const data = await parseApiResponse(response); + if ("error" in data) { + setStatusError(data.error); + } else { + setStatus(data); + } + } catch (err) { + setStatusError(err instanceof Error ? err.message : "Failed to load status"); + } finally { + setIsCheckingStatus(false); + } + }; + + const updateSynonym = (index: number, patch: Partial) => { + setSynonymsSaved(false); + setSynonyms((prev) => prev.map((s, i) => (i === index ? { ...s, ...patch } : s))); + }; + + const addSynonym = () => { + setSynonymsSaved(false); + setSynonyms((prev) => [...prev, { from: "", to: "" }]); + }; + + const removeSynonym = (index: number) => { + setSynonymsSaved(false); + setSynonyms((prev) => prev.filter((_, i) => i !== index)); + }; + + const handleSaveSynonyms = async () => { + setIsSavingSynonyms(true); + setSynonymsSaved(false); + setSynonymsError(null); + // Drop incomplete rows before persisting. + const cleaned = synonyms + .map((s) => ({ from: s.from.trim(), to: s.to.trim() })) + .filter((s) => s.from && s.to); + try { + const response = await apiFetch(`${API_BASE}/config`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ synonyms: cleaned }), + }); + const data = await parseApiResponse<{ synonyms?: Synonym[]; error?: string }>(response); + if (data.error) { + setSynonymsError(data.error); + } else { + setSynonyms(data.synonyms ?? cleaned); + setSynonymsSaved(true); + } + } catch (err) { + setSynonymsError(err instanceof Error ? err.message : "Failed to save synonyms"); + } finally { + setIsSavingSynonyms(false); + } + }; + + return ( +
+

AI Search

+ +
+

Sync Content

+

+ Re-upload all content to the search index. Content is indexed automatically on save — use + this for a full re-sync after initial setup or to recover from issues. +

+ +
+ {loadingCollections ? ( +
+ + Loading collections… +
+ ) : collectionsError ? ( + + ) : ( + c.slug)} + value={selected} + onValueChange={handleSelectionChange} + className="max-w-sm" + > + ( + + {labelForSlug(slug)} + + )} + /> + + No collections found + + {(slug: string) => ( + + {labelForSlug(slug)} + + )} + + + + )} + +
+ + + + + + + {result && !error && ( + + {indexedLabel(result.indexed)} indexed + + )} +
+ + {result?.done && ( + 0 ? "alert" : "default"} + icon={} + title="Sync complete" + description={ + <> + Indexed {indexedLabel(result.indexed)} across{" "} + {result.collections.join(", ")} + {result.skipped ? ` — ${result.skipped} already indexed` : ""} + {result.errors > 0 && ( + <> + {" "} + — {result.errors} error{result.errors !== 1 ? "s" : ""} + + )} + + } + /> + )} + + {error && } + + {statusError && ( + + )} + + {status && ( +
+
+ Instance {status.instanceName} ( + {status.binding}){" — "} + {indexedLabel(status.totalIndexed)} indexed total + {status.hybridSearch ? " · hybrid search" : ""} +
+ {status.collections.map((c) => { + const complete = c.missing.length === 0; + return ( +
+
+ {labelForSlug(c.collection)} + + {c.indexed} / {c.eligible} indexed + +
+ {!complete && ( +
    + {c.missing.slice(0, 10).map((m) => ( +
  • + {m.title || m.slug || m.id}{" "} + ({m.status}) +
  • + ))} + {c.missing.length > 10 &&
  • …and {c.missing.length - 10} more
  • } +
+ )} +
+ ); + })} +
+ )} +
+
+ +
+

Synonyms

+

+ Rewrite search queries before they reach the index. When a query contains a term on the + left, it is transparently replaced with the term on the right — e.g. “autorag” → “AI + Search”. Matching is whole-word and case-insensitive. +

+ +
+ {synonyms.length === 0 && ( +

No synonyms configured.

+ )} + + {synonyms.map((syn, index) => ( + // eslint-disable-next-line react/no-array-index-key -- rows are positional; no stable id +
+ ) => + updateSynonym(index, { from: e.target.value }) + } + placeholder="autorag" + className="max-w-[12rem]" + /> + + ) => + updateSynonym(index, { to: e.target.value }) + } + placeholder="AI Search" + className="max-w-[12rem]" + /> +
+ ))} + +
+ + + {synonymsSaved && Saved} +
+ + {synonymsError && ( + + )} +
+
+
+ ); +} + +// ============================================================================= +// Exports +// ============================================================================= + +export const pages: PluginAdminExports["pages"] = { + "/settings": SettingsPage, +}; diff --git a/packages/cloudflare/src/plugins/ai-search.ts b/packages/cloudflare/src/plugins/ai-search.ts new file mode 100644 index 0000000000..3c757c82a5 --- /dev/null +++ b/packages/cloudflare/src/plugins/ai-search.ts @@ -0,0 +1,1322 @@ +/** + * AI Search Plugin + * + * Semantic search using Cloudflare AI Search namespace bindings. + * Indexes content on save, removes on delete, exposes a search route. + * + * Requires only the `ai_search_namespaces` binding in wrangler.jsonc — + * no API tokens, no account IDs, no manual instance creation. + * + * @example + * ```typescript + * // astro.config.mjs + * import { aiSearch } from "@emdash-cms/cloudflare/plugins"; + * + * export default defineConfig({ + * integrations: [ + * emdash({ + * plugins: [aiSearch()], + * }), + * ], + * }); + * ``` + * + * @example + * ```jsonc + * // wrangler.jsonc + * { + * "ai_search_namespaces": [ + * { "binding": "AI_SEARCH", "namespace": "default" } + * ] + * } + * ``` + */ + +import type { + ContentDeleteEvent, + ContentHookEvent, + ContentPublishStateChangeEvent, + PluginContext, + PluginDescriptor, + ResolvedPlugin, + RouteContext, +} from "emdash"; +import { definePlugin, extractPlainText } from "emdash"; + +const MD_EXT = /\.md$/; +const ITEM_PREFIX = /^item:/; + +// ============================================================================= +// Configuration +// ============================================================================= + +export interface AISearchConfig { + /** AI Search instance name. @default "emdash-content" */ + instanceName?: string; + /** Binding name in wrangler.jsonc. @default "AI_SEARCH" */ + binding?: string; + /** Enable hybrid search (vector + keyword). @default true */ + hybridSearch?: boolean; +} + +/** + * KV key holding the collections the operator last configured in the admin + * dashboard. Persisted so the picker can restore the previous selection and + * the content hooks know which collections to index. + */ +const CONFIG_COLLECTIONS_KEY = "config:collections"; + +/** + * KV key holding query synonyms configured in the admin dashboard. Each entry + * maps a term/phrase (`from`) to a replacement (`to`) that is substituted into + * search queries before they reach AI Search, to improve recall. + */ +const CONFIG_SYNONYMS_KEY = "config:synonyms"; + +const REINDEX_JOB_KEY = "reindex:job"; +const REINDEX_CRON_TASK = "reindex"; +const REINDEX_PAGE_SIZE = 50; +const REINDEX_PAGES_PER_TICK = 2; +const REINDEX_HOOK_TIMEOUT_MS = 300_000; + +type ReindexJobStatus = "running" | "complete"; + +interface ReindexJob { + id: string; + status: ReindexJobStatus; + collections: string[]; + collectionIndex: number; + cursor?: string; + onlyMissing: boolean; + indexed: number; + errors: number; + skipped: number; + /** Item keys accepted on the current page, used to resume mid-page safely. */ + completedItemKeys?: string[]; + updatedAt: string; +} + +/** AI Search's maximum length for a text custom-metadata value. */ +const METADATA_TEXT_MAX_LENGTH = 500; + +/** Preferred maximum length of the indexed article-preview description. */ +const DESCRIPTION_MAX_LENGTH = 400; + +/** + * Separator used to pack `title` and `description` into a single metadata + * field (AI Search allows at most 5 custom_metadata fields). The ASCII Unit + * Separator (U+001F) is chosen because it never appears in extracted plain + * text, so it can't collide with title or description content. + */ +const TITLE_DESC_SEP = "\u001F"; + +/** Pack a title and description into one value within AI Search's text limit. */ +export function packTitleDescription(title: string, description: string): string { + if ((title + TITLE_DESC_SEP + description).length <= METADATA_TEXT_MAX_LENGTH) { + return title + TITLE_DESC_SEP + description; + } + + // Real titles fit comfortably inside the limit; retain one character for the + // separator if an unexpectedly long title does not. + const packedTitle = title.slice(0, METADATA_TEXT_MAX_LENGTH - TITLE_DESC_SEP.length); + const descriptionBudget = METADATA_TEXT_MAX_LENGTH - packedTitle.length - TITLE_DESC_SEP.length; + return packedTitle + TITLE_DESC_SEP + truncateDescription(description, descriptionBudget); +} + +/** Unpack a packed `title_desc` value, splitting on the first separator only. */ +export function unpackTitleDescription(value: string): { title: string; description: string } { + const i = value.indexOf(TITLE_DESC_SEP); + if (i < 0) return { title: value, description: "" }; + return { title: value.slice(0, i), description: value.slice(i + 1) }; +} + +/** A single query synonym: replace `from` with `to` in incoming queries. */ +export interface Synonym { + from: string; + to: string; +} + +// ============================================================================= +// Minimal types for the AI Search namespace binding +// +// These mirror the generated types from `wrangler types` (see +// worker-configuration.d.ts). Remove once @cloudflare/workers-types +// ships the AI Search binding types. +// ============================================================================= + +interface AiSearchSearchRequest { + messages: Array<{ role: string; content: string | null }>; + ai_search_options?: { + retrieval?: { + retrieval_type?: "vector" | "keyword" | "hybrid"; + match_threshold?: number; + max_num_results?: number; + filters?: Record; + context_expansion?: number; + /** Return only item metadata, skipping the (slow) full-text chunks. */ + metadata_only?: boolean; + }; + query_rewrite?: { enabled?: boolean; model?: string; rewrite_prompt?: string }; + reranking?: { enabled?: boolean; model?: string; match_threshold?: number }; + }; +} + +interface AiSearchSearchResponse { + search_query: string; + chunks: Array<{ + id: string; + type: string; + score: number; + text: string; + item: { key: string; timestamp?: number; metadata?: Record }; + }>; +} + +interface AiSearchItemInfo { + id: string; + key: string; + status: string; + metadata?: Record; +} + +interface AiSearchConfig { + id: string; + type?: string; + source?: string; + custom_metadata?: Array<{ field_name: string; data_type: "text" | "number" | "boolean" }>; + [key: string]: unknown; +} + +interface AiSearchInstance { + search(params: AiSearchSearchRequest): Promise; + update(config: Partial): Promise; + info(): Promise<{ id: string; [key: string]: unknown }>; + items: { + upload( + name: string, + content: string, + options?: { metadata?: Record }, + ): Promise; + delete(itemId: string): Promise; + }; +} + +interface AiSearchNamespace { + get(name: string): AiSearchInstance; + create(config: AiSearchConfig): Promise; +} + +// ============================================================================= +// Helpers +// ============================================================================= + +/** Get Cloudflare runtime env via cloudflare:workers. */ +async function getCloudflareEnv(): Promise | null> { + try { + const { env } = await import("cloudflare:workers"); + return env as unknown as Record; + } catch { + return null; + } +} + +/** + * Keep the Worker isolate alive for the given promise. + * Uses cloudflare:workers waitUntil — safe to call after the response is sent. + * Silently no-ops outside Workers (e.g. during local dev). + */ +function cfWaitUntil(promise: Promise): void { + import("cloudflare:workers").then(({ waitUntil }) => waitUntil(promise)).catch(() => {}); +} + +const SYSTEM_CONTENT_KEYS = new Set([ + "id", + "slug", + "status", + "authorId", + "author_id", + "primaryBylineId", + "primary_byline_id", + "createdAt", + "created_at", + "updatedAt", + "updated_at", + "publishedAt", + "published_at", + "scheduledAt", + "scheduled_at", + "deletedAt", + "deleted_at", + "version", + "liveRevisionId", + "live_revision_id", + "draftRevisionId", + "draft_revision_id", + "locale", + "translationGroup", + "translation_group", +]); + +function isSystemContentKey(key: string): boolean { + return key.startsWith("_") || SYSTEM_CONTENT_KEYS.has(key); +} + +function extractIndexableText(value: unknown): string { + if (typeof value === "string") return extractPlainText(value); + if (Array.isArray(value)) { + // eslint-disable-next-line @typescript-eslint/no-explicit-any, typescript-eslint(no-unsafe-type-assertion) -- Portable Text arrays are untyped; extractPlainText handles validation + return extractPlainText(value as any); + } + return ""; +} + +function truncateDescription(value: string, maxLength: number = DESCRIPTION_MAX_LENGTH): string { + if (maxLength <= 0) return ""; + const normalized = value.replace(/\s+/g, " ").trim(); + if (normalized.length <= maxLength) return normalized; + if (maxLength === 1) return "\u2026"; + + const truncated = normalized.slice(0, maxLength - 1); + const lastSpace = truncated.lastIndexOf(" "); + return `${lastSpace > 0 ? truncated.slice(0, lastSpace) : truncated}\u2026`; +} + +/** Convert a content entry to Markdown for indexing. */ +function contentToMarkdown(content: Record, collection: string): string { + const parts: string[] = []; + + if (typeof content.title === "string") parts.push(`# ${content.title}`); + parts.push(`Collection: ${collection}`); + + for (const [key, value] of Object.entries(content)) { + if (key === "title" || isSystemContentKey(key)) continue; + const text = extractIndexableText(value); + if (text) parts.push(text); + } + + return parts.join("\n\n"); +} + +/** + * Build a short plain-text description (article preview) from a content entry, + * used as searchable/returnable metadata so metadata-only retrieval can render + * a snippet without fetching the full document. Prefers an explicit excerpt, + * then falls back to other indexable content fields. Excludes system metadata + * and truncates to `DESCRIPTION_MAX_LENGTH` at a word boundary. + */ +function contentToDescription(content: Record): string { + const excerpt = extractIndexableText(content.excerpt); + if (excerpt) return truncateDescription(excerpt); + + const parts: string[] = []; + for (const [key, value] of Object.entries(content)) { + if (key === "title" || key === "excerpt" || isSystemContentKey(key)) continue; + const text = extractIndexableText(value); + if (text) parts.push(text); + } + return truncateDescription(parts.join(" ")); +} + +function imageUrlFromValue(value: unknown): string { + if (!value || typeof value !== "object" || Array.isArray(value)) return ""; + const obj = value as Record; + if (typeof obj.src === "string" && obj.src) return obj.src; + const meta = obj.meta as Record | undefined; + if (typeof meta?.storageKey === "string" && meta.storageKey) { + return `/_emdash/api/media/file/${meta.storageKey}`; + } + return ""; +} + +/** + * Extract a thumbnail URL from a content entry, preferring the conventional + * featured-image field before falling back to the first image-shaped value. + */ +function extractImageUrl(content: Record): string { + const featured = imageUrlFromValue(content.featured_image ?? content.featuredImage); + if (featured) return featured; + + for (const [key, value] of Object.entries(content)) { + if (key.startsWith("_") || key === "featured_image" || key === "featuredImage") continue; + const image = imageUrlFromValue(value); + if (image) return image; + } + return ""; +} + +/** + * Get the `visible_after` timestamp for a content item. + * Returns 0 for published content (immediately visible) or the + * scheduled_at unix timestamp in seconds for scheduled content. + */ +function getVisibleAfter(content: Record): number { + const status = typeof content.status === "string" ? content.status : ""; + // Hook events expose the camelCase `scheduledAt`; reindex merges the raw + // row which may still carry snake_case `scheduled_at`. Accept either. + const scheduledAt = content.scheduledAt ?? content.scheduled_at; + if ( + status === "scheduled" && + (typeof scheduledAt === "string" || typeof scheduledAt === "number") + ) { + const d = new Date(scheduledAt); + if (!isNaN(d.getTime())) return Math.floor(d.getTime() / 1000); + } + return 0; +} + +/** + * Flatten a content-hook record. Content hooks pass the `ContentItem` shape, + * where the editable fields (title, body, images) live under `.data` while the + * system columns (id, slug, status, locale, scheduledAt) sit at the top level. + * Merging `.data` up gives the same flat record the reindex path builds with + * `{ ...item, ...item.data }`, so field extraction behaves identically in both + * paths (without it, the hook path reads an empty title and skips the body). + */ +export function flattenContentRecord(content: Record): Record { + const data = content.data && typeof content.data === "object" ? content.data : {}; + return { ...content, ...data }; +} + +/** Deterministic document key: `{collection}/{id}.md`. */ +function contentKey(collection: string, id: string): string { + return `${collection}/${id}.md`; +} + +async function retry(operation: () => Promise): Promise { + let lastError: unknown; + for (let attempt = 0; attempt < 3; attempt++) { + try { + return await operation(); + } catch (error) { + lastError = error; + if (attempt < 2) await new Promise((resolve) => setTimeout(resolve, 250 * 2 ** attempt)); + } + } + throw lastError; +} + +interface UploadItemOptions { + /** Called as soon as AI Search accepts the upload, before mirror bookkeeping. */ + onUploaded?: (item: AiSearchItemInfo) => Promise; + /** Leave accepted uploads in place when mirror bookkeeping fails. */ + tolerateMirrorFailure?: boolean; +} + +/** Replace a mirrored item before uploading because AI Search keys are unique. */ +async function uploadItem( + instance: AiSearchInstance, + key: string, + markdown: string, + metadata: Record, + ctx: PluginContext, + options: UploadItemOptions = {}, +): Promise { + const mirrorKey = `item:${key}`; + const previousId = await ctx.kv.get(mirrorKey); + let previousRemoved = false; + if (previousId) { + try { + await instance.items.delete(previousId); + previousRemoved = true; + } catch { + // A stale mirror is harmless: upload will either succeed or surface + // the real key conflict to the caller. + } + } + + let item: AiSearchItemInfo; + try { + item = await instance.items.upload(key, markdown, { metadata }); + } catch (error) { + if (previousRemoved) await ctx.kv.delete(mirrorKey); + throw error; + } + // AI Search's non-polling upload call returning without throwing is the + // acceptance boundary. Reindex progress is checkpointed here; indexing may + // continue asynchronously inside AI Search. + await options.onUploaded?.(item); + + if (!item?.id) { + if (options.tolerateMirrorFailure) return; + throw new Error(`upload for ${key} returned no item id`); + } + try { + await ctx.kv.set(mirrorKey, item.id); + } catch (error) { + if (options.tolerateMirrorFailure) { + console.error(`[ai-search] Failed to mirror accepted upload ${key}:`, error); + return; + } + // Hook-driven writes retain their stronger rollback guarantee. Reindexing + // can reconcile an accepted-but-unmirrored upload in a later pass. + try { + await instance.items.delete(item.id); + } catch {} + if (previousRemoved) await ctx.kv.delete(mirrorKey); + throw error; + } +} + +function createReindexJob(collections: string[], onlyMissing: boolean): ReindexJob { + return { + id: crypto.randomUUID(), + status: "running", + collections, + collectionIndex: 0, + onlyMissing, + indexed: 0, + errors: 0, + skipped: 0, + updatedAt: new Date().toISOString(), + }; +} + +function reindexResult(job: ReindexJob) { + return { + jobId: job.id, + status: job.status, + done: job.status === "complete", + onlyMissing: job.onlyMissing, + collections: job.collections, + indexed: job.indexed, + errors: job.errors, + skipped: job.skipped, + }; +} + +/** Parse a content key back into collection + id. */ +function parseContentKey(key: string): { collection: string; id: string } { + const [col, ...rest] = key.split("/"); + return { collection: col ?? "", id: rest.join("/").replace(MD_EXT, "") }; +} + +/** + * Normalize a `collections` request field into a trimmed slug array. + * Accepts a comma-separated string or an array of strings; returns `null` + * when the input is neither (so callers can fall back or error). + */ +function parseCollections(value: unknown): string[] | null { + const raw = + typeof value === "string" + ? value.split(",") + : Array.isArray(value) + ? value.filter((v): v is string => typeof v === "string") + : null; + if (raw === null) return null; + return raw.map((c) => c.trim()).filter(Boolean); +} + +/** + * Normalize a `synonyms` request field into a validated `Synonym[]`. Accepts an + * array of `{ from, to }` objects; trims whitespace and drops entries missing + * either side. Returns `null` when the input is not an array. + */ +function parseSynonyms(value: unknown): Synonym[] | null { + if (!Array.isArray(value)) return null; + const result: Synonym[] = []; + for (const entry of value) { + if (typeof entry !== "object" || entry === null) continue; + const from = (entry as Record).from; + const to = (entry as Record).to; + if (typeof from !== "string" || typeof to !== "string") continue; + const trimmedFrom = from.trim(); + const trimmedTo = to.trim(); + if (!trimmedFrom || !trimmedTo) continue; + result.push({ from: trimmedFrom, to: trimmedTo }); + } + return result; +} + +/** Escape a string for safe use inside a RegExp. */ +function escapeRegex(value: string): string { + return value.replace(/[.*+?^${}()|[\]\\]/g, "\\$&"); +} + +/** + * A synonym rewriter compiled from a synonym set. Holds a single combined + * regex plus the `from` -> `to` lookup so a query can be rewritten in one pass. + */ +interface SynonymRewriter { + re: RegExp | null; + lookup: Map; +} + +/** + * Compile a synonym set into a single reusable regex. Benchmarking showed that + * recompiling a regex per synonym per query is O(synonyms) and dominates the + * cost (~29µs at 100 synonyms), while one combined precompiled alternation is + * flat (~0.2µs) and still supports multi-word phrases — unlike a word-split + * `Map`/`find` lookup. Callers cache the result so compilation happens only + * when the synonym set changes. + */ +function compileSynonyms(synonyms: Synonym[]): SynonymRewriter { + // Longer phrases first so multi-word terms win over their sub-words. + const sorted = [...synonyms].filter((s) => s.from).sort((a, b) => b.from.length - a.from.length); + const lookup = new Map(); + for (const s of sorted) { + const key = s.from.toLowerCase(); + if (!lookup.has(key)) lookup.set(key, s.to); + } + if (sorted.length === 0) return { re: null, lookup }; + const pattern = sorted.map((s) => escapeRegex(s.from)).join("|"); + return { re: new RegExp(`\\b(?:${pattern})\\b`, "gi"), lookup }; +} + +/** + * Transparently rewrite a query by substituting configured synonym terms in + * place, using a precompiled rewriter. Whole-word (case-insensitive) + * occurrences of each `from` anywhere in the query are replaced with `to` — + * e.g. with `autorag` -> `AI Search`, "what is autorag" becomes "what is AI + * Search". + */ +function applySynonyms(query: string, rewriter: SynonymRewriter): string { + if (!rewriter.re) return query; + rewriter.re.lastIndex = 0; + return query.replace(rewriter.re, (match) => rewriter.lookup.get(match.toLowerCase()) ?? match); +} + +// ============================================================================= +// Descriptor (for astro.config.mjs) +// ============================================================================= + +export function aiSearch(config: AISearchConfig = {}): PluginDescriptor { + return { + id: "ai-search", + version: "1.0.0", + entrypoint: "@emdash-cms/cloudflare/plugins/ai-search", + options: config, + capabilities: ["read:content"], + adminEntry: "@emdash-cms/cloudflare/plugins/ai-search-admin", + adminPages: [{ path: "/settings", label: "AI Search", icon: "search" }], + }; +} + +// ============================================================================= +// Plugin implementation (loaded at runtime via entrypoint) +// ============================================================================= + +export function createPlugin(config: AISearchConfig = {}): ResolvedPlugin { + const instanceName = config.instanceName ?? "emdash-content"; + const bindingName = config.binding ?? "AI_SEARCH"; + const hybridSearch = config.hybridSearch ?? true; + + /** + * Read the collections the operator last configured in the dashboard. + * Returns `null` when nothing has been configured yet (never persisted), + * distinct from an explicit empty selection. + */ + async function getConfiguredCollections(ctx: PluginContext): Promise { + const saved = await ctx.kv.get(CONFIG_COLLECTIONS_KEY); + return Array.isArray(saved) ? saved : null; + } + + /** Persist the operator's collection selection from the dashboard. */ + async function saveConfiguredCollections( + ctx: PluginContext, + collections: string[], + ): Promise { + await ctx.kv.set(CONFIG_COLLECTIONS_KEY, collections); + } + + /** Read the query synonyms configured in the dashboard. */ + async function getConfiguredSynonyms(ctx: PluginContext): Promise { + const saved = await ctx.kv.get(CONFIG_SYNONYMS_KEY); + return Array.isArray(saved) ? saved : []; + } + + // Cache the compiled synonym regex across requests in this isolate so we only + // recompile when the configured set actually changes (keyed by its JSON). + let synonymCache: { key: string; rewriter: SynonymRewriter } | null = null; + + /** Read synonyms from KV and return a compiled (cached) rewriter. */ + async function getSynonymRewriter(ctx: PluginContext): Promise { + const synonyms = await getConfiguredSynonyms(ctx); + const key = JSON.stringify(synonyms); + if (synonymCache?.key !== key) { + synonymCache = { key, rewriter: compileSynonyms(synonyms) }; + } + return synonymCache.rewriter; + } + + /** Persist the operator's query synonyms from the dashboard. */ + async function saveConfiguredSynonyms(ctx: PluginContext, synonyms: Synonym[]): Promise { + await ctx.kv.set(CONFIG_SYNONYMS_KEY, synonyms); + } + + /** + * Whether a content hook should act on the given collection. Content is + * synced only for collections the operator selected in the dashboard. When + * nothing has been configured yet, all collections are indexed so the + * plugin works out of the box until the operator narrows the selection. + */ + async function shouldSync(collection: string, ctx: PluginContext): Promise { + const configured = await getConfiguredCollections(ctx); + return configured === null || configured.includes(collection); + } + + async function getBinding(): Promise { + const env = await getCloudflareEnv(); + if (!env?.[bindingName]) return null; + return env[bindingName] as AiSearchNamespace; + } + + async function ensureInstance(ns: AiSearchNamespace): Promise { + const handle = ns.get(instanceName); + try { + await handle.info(); + return handle; + } catch { + return ns.create({ + id: instanceName, + hybrid_search_enabled: hybridSearch, + // AI Search allows at most 5 custom metadata fields, so title and + // description are packed into a single `title_desc` field to make + // room for `locale`. + custom_metadata: [ + { field_name: "visible_after", data_type: "number" }, + { field_name: "title_desc", data_type: "text" }, + { field_name: "slug", data_type: "text" }, + { field_name: "image", data_type: "text" }, + { field_name: "locale", data_type: "text" }, + ], + }); + } + } + + /** + * Index a content item in AI Search. + * + * @param visibleAfter Unix timestamp (seconds) when the content becomes + * visible. Use 0 for already-published content. For scheduled content, + * pass the `scheduled_at` timestamp so the query filter + * `visible_after <= now` excludes it until the scheduled time. + */ + async function indexContent( + content: Record, + collection: string, + ctx: PluginContext, + visibleAfter: number = 0, + ): Promise { + const ns = await getBinding(); + if (!ns) { + console.warn("[ai-search] indexContent: binding not available"); + return; + } + + const key = contentKey(collection, String(content.id)); + try { + const instance = await ensureInstance(ns); + // Hook events nest editable fields under `.data`; flatten so title/body + // extraction matches the reindex path. + const record = flattenContentRecord(content); + const markdown = contentToMarkdown(record, collection); + if (!markdown.trim()) return; + + const slug = typeof record.slug === "string" ? record.slug : ""; + const title = typeof record.title === "string" ? record.title : ""; + const description = contentToDescription(record); + const image = extractImageUrl(record); + const locale = + typeof record.locale === "string" && record.locale + ? record.locale + : (ctx.site?.locale ?? "en"); + + const metadata: Record = { + visible_after: String(visibleAfter), + title_desc: packTitleDescription(title, description), + slug, + locale, + }; + if (image) metadata.image = image; + + await retry(() => uploadItem(instance, key, markdown, metadata, ctx)); + console.log(`[ai-search] Queued ${key}`); + } catch (error) { + console.error("[ai-search] Error indexing content:", error); + } + } + + /** Remove a content item from the AI Search index. */ + async function removeFromIndex( + collection: string, + id: string, + ctx: PluginContext, + ): Promise { + const ns = await getBinding(); + if (!ns) return; + + const key = contentKey(collection, id); + try { + const itemId = await ctx.kv.get(`item:${key}`); + if (!itemId) return; + + const instance = await ensureInstance(ns); + await instance.items.delete(itemId); + // Do not erase a replacement written by a concurrent save/reindex. + if ((await ctx.kv.get(`item:${key}`)) === itemId) { + await ctx.kv.delete(`item:${key}`); + } + console.log(`[ai-search] Removed ${key} (item: ${itemId})`); + } catch (error) { + console.error("[ai-search] Error removing content:", error); + } + } + + /** + * Synchronize one content item with the public search index based on its + * status. Published content is indexed as immediately visible, scheduled + * content is indexed but gated behind its `visible_after` timestamp, and + * anything else (draft, trashed) is removed from the index. + */ + function syncSearchIndex( + content: Record, + collection: string, + ctx: PluginContext, + ): Promise { + const status = typeof content.status === "string" ? content.status : ""; + if (status === "published") { + return indexContent(content, collection, ctx, 0); + } + if (status === "scheduled") { + return indexContent(content, collection, ctx, getVisibleAfter(content)); + } + return removeFromIndex(collection, String(content.id), ctx); + } + + /** Keep the worker alive until the index write settles, then surface it. */ + function waitForSync(work: Promise): Promise { + cfWaitUntil(work); + return work; + } + + /** Process exactly one content page so every request stays bounded. */ + async function processReindexBatch(job: ReindexJob, ctx: PluginContext) { + if (job.status === "complete") return reindexResult(job); + if (!ctx.content) throw new Error("Content access not available"); + const ns = await getBinding(); + if (!ns) throw new Error("AI Search binding not available"); + const instance = await ensureInstance(ns); + const collection = job.collections[job.collectionIndex]; + if (!collection) { + job.status = "complete"; + await ctx.kv.set(REINDEX_JOB_KEY, job); + return reindexResult(job); + } + + const page = await ctx.content.list(collection, { + limit: REINDEX_PAGE_SIZE, + cursor: job.cursor, + }); + const completedItemKeys = new Set(job.completedItemKeys ?? []); + let checkpointWrites = Promise.resolve(); + + const checkpointAcceptedUpload = async (key: string): Promise => { + job.indexed++; + completedItemKeys.add(key); + job.completedItemKeys = [...completedItemKeys]; + job.updatedAt = new Date().toISOString(); + + // Uploads stay concurrent, while checkpoint writes are serialized so an + // older snapshot cannot overwrite a newer completion out of order. + const snapshot: ReindexJob = { ...job, completedItemKeys: [...completedItemKeys] }; + checkpointWrites = checkpointWrites.then(() => ctx.kv.set(REINDEX_JOB_KEY, snapshot)); + await checkpointWrites; + }; + + await Promise.all( + page.items.map(async (item) => { + const key = contentKey(collection, item.id); + if (completedItemKeys.has(key)) return; + try { + if (item.status !== "published" && item.status !== "scheduled") return; + if (job.onlyMissing && (await ctx.kv.get(`item:${key}`))) { + job.skipped++; + return; + } + + const record = { ...item, ...item.data }; + const markdown = contentToMarkdown(record, collection); + if (!markdown.trim()) { + job.skipped++; + return; + } + + const visibleAfter = getVisibleAfter(record); + if (item.status === "scheduled" && visibleAfter === 0) { + throw new Error("Scheduled content is missing its publication time"); + } + const metadata: Record = { + visible_after: String(visibleAfter), + title_desc: packTitleDescription( + typeof item.data.title === "string" ? item.data.title : "", + contentToDescription(record), + ), + slug: typeof item.slug === "string" ? item.slug : "", + locale: + typeof item.locale === "string" && item.locale + ? item.locale + : (ctx.site?.locale ?? "en"), + }; + const image = extractImageUrl(record); + if (image) metadata.image = image; + await retry(() => + uploadItem(instance, key, markdown, metadata, ctx, { + onUploaded: () => checkpointAcceptedUpload(key), + tolerateMirrorFailure: true, + }), + ); + } catch (error) { + console.error(`[ai-search] Failed to index ${collection}/${item.id}:`, error); + job.errors++; + } + }), + ); + await checkpointWrites; + + if (page.cursor) { + job.cursor = page.cursor; + } else { + job.collectionIndex++; + delete job.cursor; + if (job.collectionIndex >= job.collections.length) job.status = "complete"; + } + delete job.completedItemKeys; + job.updatedAt = new Date().toISOString(); + await ctx.kv.set(REINDEX_JOB_KEY, job); + return reindexResult(job); + } + + return definePlugin({ + id: "ai-search", + version: "1.0.0", + capabilities: ["read:content"], + admin: { + entry: "@emdash-cms/cloudflare/plugins/ai-search-admin", + pages: [{ path: "/settings", label: "AI Search", icon: "search" }], + }, + + hooks: { + "plugin:install": { + handler: async (_event: unknown, ctx: PluginContext): Promise => { + const collections = (await getConfiguredCollections(ctx)) ?? ["posts", "pages"]; + await saveConfiguredCollections(ctx, collections); + const job = createReindexJob(collections, false); + await ctx.kv.set(REINDEX_JOB_KEY, job); + await ctx.cron?.schedule(REINDEX_CRON_TASK, { schedule: "* * * * *" }); + }, + }, + + cron: { + // The default five-second plugin hook timeout can expire while a page + // of accepted uploads is still checkpointing. Keep the scheduled event + // alive for the bounded two-page batch; this does not poll for indexing. + timeout: REINDEX_HOOK_TIMEOUT_MS, + handler: async (event, ctx): Promise => { + if (event.name !== REINDEX_CRON_TASK) return; + const job = await ctx.kv.get(REINDEX_JOB_KEY); + if (!job || job.status === "complete") { + await ctx.cron?.cancel(REINDEX_CRON_TASK); + return; + } + + for (let page = 0; page < REINDEX_PAGES_PER_TICK; page++) { + if ((await processReindexBatch(job, ctx)).done) break; + } + if (reindexResult(job).done) await ctx.cron?.cancel(REINDEX_CRON_TASK); + }, + }, + + "content:afterSave": { + handler: async (event: ContentHookEvent, ctx: PluginContext): Promise => { + const { content, collection } = event; + if (!(await shouldSync(collection, ctx))) return; + + // Sync based on the current status: published content is visible + // immediately (visible_after=0), scheduled content is indexed but + // gated until its scheduledAt timestamp, and drafts are removed. + return waitForSync(syncSearchIndex(content, collection, ctx)); + }, + }, + + "content:afterPublish": { + handler: async ( + event: ContentPublishStateChangeEvent, + ctx: PluginContext, + ): Promise => { + const { content, collection } = event; + if (!(await shouldSync(collection, ctx))) return; + + return waitForSync(indexContent(content, collection, ctx)); + }, + }, + + "content:afterUnpublish": { + handler: async ( + event: ContentPublishStateChangeEvent, + ctx: PluginContext, + ): Promise => { + const { content, collection } = event; + if (!(await shouldSync(collection, ctx))) return; + + return waitForSync(removeFromIndex(collection, String(content.id), ctx)); + }, + }, + + "content:afterSchedule": { + handler: async ( + event: ContentPublishStateChangeEvent, + ctx: PluginContext, + ): Promise => { + const { content, collection } = event; + if (!(await shouldSync(collection, ctx))) return; + + // Index the item with its `visible_after` gate so it stays hidden + // from search results until the scheduled time arrives. + return waitForSync(syncSearchIndex(content, collection, ctx)); + }, + }, + + "content:afterUnschedule": { + handler: async ( + event: ContentPublishStateChangeEvent, + ctx: PluginContext, + ): Promise => { + const { content, collection } = event; + if (!(await shouldSync(collection, ctx))) return; + + // Unscheduling returns the item to a draft state — drop it from + // the index. + return waitForSync(removeFromIndex(collection, String(content.id), ctx)); + }, + }, + + "content:afterRestore": { + handler: async ( + event: ContentPublishStateChangeEvent, + ctx: PluginContext, + ): Promise => { + const { content, collection } = event; + if (!(await shouldSync(collection, ctx))) return; + + // Restored content re-enters the index according to its restored + // status (published/scheduled index, otherwise remove). + return waitForSync(syncSearchIndex(content, collection, ctx)); + }, + }, + + "content:afterDelete": { + handler: async (event: ContentDeleteEvent, ctx: PluginContext): Promise => { + const { id, collection } = event; + if (!(await shouldSync(collection, ctx))) return; + + return waitForSync(removeFromIndex(collection, id, ctx)); + }, + }, + }, + + routes: { + query: { + public: true, + handler: async (ctx: RouteContext): Promise => { + const start = Date.now(); + + // Support both JSON body input and URL query params (for GET requests) + const input = ctx.input as Record | undefined; + const url = new URL(ctx.request.url); + const params = url.searchParams; + + const ns = await getBinding(); + if (!ns) { + console.warn("[ai-search] Query failed: binding not available"); + return { error: `${bindingName} binding not available`, results: [] }; + } + + const q = + (typeof input?.q === "string" ? input.q : undefined) ?? params.get("q") ?? undefined; + if (!q) { + return { error: "Query parameter 'q' is required", results: [] }; + } + + const locale = + (typeof input?.locale === "string" ? input.locale : undefined) ?? + params.get("locale") ?? + undefined; + if (!locale) { + return { error: "Query parameter 'locale' is required", results: [] }; + } + + const limit = + (typeof input?.limit === "number" ? input.limit : undefined) ?? + (params.has("limit") ? Number(params.get("limit")) : undefined) ?? + 10; + const collection = + (typeof input?.collection === "string" ? input.collection : undefined) ?? + params.get("collection") ?? + undefined; + + // Transparently substitute configured synonym terms so the query + // sent to AI Search uses the canonical wording that indexes better. + const rewriter = await getSynonymRewriter(ctx); + const effectiveQuery = applySynonyms(q, rewriter); + + console.log( + `[ai-search] Query: q=${JSON.stringify(q)}${ + effectiveQuery === q ? "" : ` -> ${JSON.stringify(effectiveQuery)}` + } limit=${limit} collection=${collection ?? "all"}`, + ); + + try { + const instance = await ensureInstance(ns); + const nowSeconds = Math.floor(Date.now() / 1000); + + // Run a search for a specific locale and return deduped, mapped + // results. Extracted so the locale fallback can reuse it verbatim. + const searchLocale = async (searchLocaleCode: string) => { + const response = await instance.search({ + messages: [{ role: "user", content: effectiveQuery }], + ai_search_options: { + retrieval: { + max_num_results: limit, + filters: { + visible_after: { $lte: nowSeconds }, + locale: { $eq: searchLocaleCode }, + }, + // Metadata-only retrieval is always used: results are + // rendered from the packed title/description metadata, + // skipping the slower full-text chunk retrieval. + metadata_only: true, + }, + }, + }); + + let chunks = response.chunks; + if (collection) { + const cols = collection.split(",").map((c) => c.trim()); + chunks = chunks.filter((c) => cols.some((col) => c.item.key.startsWith(`${col}/`))); + } + + // Deduplicate by item key, keeping the highest-scoring chunk per item + const bestByKey = new Map(); + for (const c of chunks) { + const existing = bestByKey.get(c.item.key); + if (!existing || c.score > existing.score) { + bestByKey.set(c.item.key, c); + } + } + const uniqueChunks = [...bestByKey.values()]; + + // Resolve slug/title/description for each result. Title and + // description are packed into the `title_desc` metadata field + // (present in both normal and metadata-only mode); unpack it. + // The snippet uses the full-text chunk when available, otherwise + // the description (the only text available in metadata-only mode). + const mapped = uniqueChunks.map((c) => { + const parsed = parseContentKey(c.item.key); + const md = c.item.metadata ?? {}; + const slug = typeof md.slug === "string" && md.slug ? md.slug : null; + const packed = typeof md.title_desc === "string" ? md.title_desc : ""; + const { title: rawTitle, description: rawDescription } = + unpackTitleDescription(packed); + const title = rawTitle ? rawTitle : null; + const description = rawDescription ? rawDescription : null; + const image = typeof md.image === "string" && md.image ? md.image : null; + + const snippet = c.text && c.text.trim() ? c.text : (description ?? ""); + return { + ...parsed, + slug, + title, + description, + image, + score: c.score, + snippet, + }; + }); + + return { searchQuery: response.search_query, results: mapped }; + }; + + let { searchQuery, results } = await searchLocale(locale); + + // Fall back to the site default locale when the requested locale + // returns nothing, so untranslated content is still discoverable. + if (results.length === 0 && ctx.site?.locale && locale !== ctx.site.locale) { + ({ searchQuery, results } = await searchLocale(ctx.site.locale)); + } + + const elapsed = Date.now() - start; + console.log( + `[ai-search] Query complete: ${results.length} results in ${elapsed}ms (rewritten: ${JSON.stringify(searchQuery)})`, + ); + return { query: searchQuery, results }; + } catch (error) { + const elapsed = Date.now() - start; + console.error(`[ai-search] Query failed after ${elapsed}ms:`, error); + return { + error: error instanceof Error ? error.message : "Search failed", + results: [], + }; + } + }, + }, + + status: { + handler: async (ctx: RouteContext): Promise => { + if (!ctx.content) { + return { error: "Content access not available" }; + } + + // Build the set of item keys currently present in the index from the + // KV id-map (`item:{collection}/{id}.md` -> AI Search item id). + const itemEntries = await ctx.kv.list("item:"); + const indexedKeys = new Set(); + for (const entry of itemEntries) { + const key = entry.key.replace(ITEM_PREFIX, "").replace(MD_EXT, ""); + indexedKeys.add(key); + } + + const input = ctx.input as Record | undefined; + const params = new URL(ctx.request.url).searchParams; + const requested = + typeof input?.collections === "string" + ? input.collections.split(",") + : Array.isArray(input?.collections) + ? (input.collections as string[]) + : (params.get("collections")?.split(",") ?? []); + const trimmed = requested.map((c) => c.trim()).filter(Boolean); + const collections = + trimmed.length > 0 ? trimmed : ((await getConfiguredCollections(ctx)) ?? []); + + const perCollection: Array<{ + collection: string; + eligible: number; + indexed: number; + missing: Array<{ + id: string; + slug: string | null; + title: string | null; + status: string; + }>; + }> = []; + + for (const collection of collections) { + let cursor: string | undefined; + let eligible = 0; + let indexed = 0; + const missing: Array<{ + id: string; + slug: string | null; + title: string | null; + status: string; + }> = []; + try { + do { + const page = await ctx.content.list(collection, { limit: 50, cursor }); + for (const item of page.items) { + const status = typeof item.status === "string" ? item.status : ""; + if (status !== "published" && status !== "scheduled") continue; + eligible++; + if (indexedKeys.has(`${collection}/${item.id}`)) { + indexed++; + } else { + missing.push({ + id: item.id, + slug: item.slug, + title: typeof item.data.title === "string" ? item.data.title : null, + status, + }); + } + } + cursor = page.cursor; + } while (cursor); + perCollection.push({ collection, eligible, indexed, missing }); + } catch (error) { + console.error(`[ai-search] Status failed for ${collection}:`, error); + perCollection.push({ collection, eligible, indexed, missing }); + } + } + + return { + instanceName, + binding: bindingName, + hybridSearch, + totalIndexed: indexedKeys.size, + collections: perCollection, + }; + }, + }, + + // Read or persist the operator's dashboard configuration (indexed + // collections and query synonyms). GET returns the saved config; POST + // updates whichever fields are provided. + config: { + handler: async (ctx: RouteContext): Promise => { + if (ctx.request.method.toUpperCase() === "GET") { + return { + collections: (await getConfiguredCollections(ctx)) ?? [], + synonyms: await getConfiguredSynonyms(ctx), + }; + } + + const input = ctx.input as Record | undefined; + + if (input?.collections !== undefined) { + const collections = parseCollections(input.collections); + if (!collections) { + return { error: "collections must be an array of collection slugs" }; + } + await saveConfiguredCollections(ctx, collections); + } + + if (input?.synonyms !== undefined) { + const synonyms = parseSynonyms(input.synonyms); + if (!synonyms) { + return { error: "synonyms must be an array of { from, to } objects" }; + } + await saveConfiguredSynonyms(ctx, synonyms); + } + + return { + collections: (await getConfiguredCollections(ctx)) ?? [], + synonyms: await getConfiguredSynonyms(ctx), + }; + }, + }, + + reindex: { + handler: async (ctx: RouteContext): Promise => { + const current = await ctx.kv.get(REINDEX_JOB_KEY); + if (ctx.request.method.toUpperCase() === "GET") { + return current ? reindexResult(current) : null; + } + if (!ctx.cron) return { error: "Cron scheduling is not available" }; + + const input = ctx.input as Record | undefined; + const requestedJobId = typeof input?.jobId === "string" ? input.jobId : undefined; + if (requestedJobId && current?.id !== requestedJobId) { + return { error: "Reindex job not found" }; + } + + let job = current?.status === "running" ? current : null; + if (!job) { + const collections = + parseCollections(input?.collections) ?? (await getConfiguredCollections(ctx)) ?? []; + if (collections.length === 0) { + return { + error: "No collections specified. Select collections in the dashboard first.", + }; + } + await saveConfiguredCollections(ctx, collections); + job = createReindexJob(collections, input?.onlyMissing === true); + await ctx.kv.set(REINDEX_JOB_KEY, job); + } + + await ctx.cron.schedule(REINDEX_CRON_TASK, { schedule: "* * * * *" }); + return reindexResult(job); + }, + }, + }, + }); +} + +export default createPlugin; diff --git a/packages/cloudflare/src/plugins/index.ts b/packages/cloudflare/src/plugins/index.ts index 84dc964428..b60234d0ef 100644 --- a/packages/cloudflare/src/plugins/index.ts +++ b/packages/cloudflare/src/plugins/index.ts @@ -10,3 +10,4 @@ export { createCloudflareEmailDeliver, type CloudflareEmailConfig, } from "./cloudflare-email.js"; +export { aiSearch, type AISearchConfig } from "./ai-search.js"; diff --git a/packages/cloudflare/tsdown.config.ts b/packages/cloudflare/tsdown.config.ts index e6f8419a79..8c80350644 100644 --- a/packages/cloudflare/tsdown.config.ts +++ b/packages/cloudflare/tsdown.config.ts @@ -19,6 +19,9 @@ export default defineConfig({ // as their `entrypoint`, so the astro integration can statically import // `createPlugin` from it (#1721). "src/plugins/cloudflare-email.ts", + "src/plugins/ai-search.ts", + "src/plugins/ai-search-admin.tsx", + // Media provider runtimes "src/media/images-runtime.ts", "src/media/stream-runtime.ts", From 7e9e09392965ddbccdf9c10ef07474ad3bee50c8 Mon Sep 17 00:00:00 2001 From: ttmx Date: Thu, 16 Jul 2026 15:29:15 +0100 Subject: [PATCH 02/39] feat(demo): wire AI Search into the Cloudflare demo Registers the AI_SEARCH namespace binding, adds a cron trigger to flush the reindex queue, documents the aiSearch() options at their defaults, and adds the search UI: Cloudflare's AI Search snippet (search-modal-snippet) opened by a nav button or Cmd/Ctrl+K, backed by a /api/ai-search/search endpoint that queries the binding directly. Replaces the previous custom /search page. --- demos/cloudflare/astro.config.mjs | 9 + demos/cloudflare/package.json | 1 + demos/cloudflare/src/layouts/Base.astro | 190 +- .../src/pages/api/ai-search/search.ts | 161 ++ demos/cloudflare/src/pages/search.astro | 182 -- demos/cloudflare/worker-configuration.d.ts | 2360 +++++++++++++++-- demos/cloudflare/wrangler.jsonc | 10 + pnpm-lock.yaml | 98 +- 8 files changed, 2478 insertions(+), 533 deletions(-) create mode 100644 demos/cloudflare/src/pages/api/ai-search/search.ts delete mode 100644 demos/cloudflare/src/pages/search.astro diff --git a/demos/cloudflare/astro.config.mjs b/demos/cloudflare/astro.config.mjs index a349f89745..69c5b0b6cc 100644 --- a/demos/cloudflare/astro.config.mjs +++ b/demos/cloudflare/astro.config.mjs @@ -10,6 +10,7 @@ import { cloudflareImages, cloudflareStream, } from "@emdash-cms/cloudflare"; +import { aiSearch } from "@emdash-cms/cloudflare/plugins"; import { formsPlugin } from "@emdash-cms/plugin-forms"; import webhookNotifier from "@emdash-cms/plugin-webhook-notifier"; import { defineConfig, fontProviders } from "astro/config"; @@ -72,6 +73,14 @@ export default defineConfig({ plugins: [ // Test plugin that exercises all v2 APIs formsPlugin(), + aiSearch({ + // AI Search instance name (created on first index). Default: "emdash-content". + instanceName: "emdash-content", + // wrangler.jsonc `ai_search_namespaces` binding name. Default: "AI_SEARCH". + binding: "AI_SEARCH", + // Hybrid search (vector + keyword). Default: true. + hybridSearch: true, + }), ], // Sandboxed plugins (run in isolated workers) sandboxed: [webhookNotifier], diff --git a/demos/cloudflare/package.json b/demos/cloudflare/package.json index 0459fa6213..82047e5080 100644 --- a/demos/cloudflare/package.json +++ b/demos/cloudflare/package.json @@ -15,6 +15,7 @@ "dependencies": { "@astrojs/cloudflare": "catalog:", "@astrojs/react": "catalog:", + "@cloudflare/ai-search-snippet": "^0.0.40", "@emdash-cms/cloudflare": "workspace:*", "@emdash-cms/plugin-forms": "workspace:*", "@emdash-cms/plugin-webhook-notifier": "workspace:*", diff --git a/demos/cloudflare/src/layouts/Base.astro b/demos/cloudflare/src/layouts/Base.astro index df078d8c05..042d0fb8f6 100644 --- a/demos/cloudflare/src/layouts/Base.astro +++ b/demos/cloudflare/src/layouts/Base.astro @@ -7,7 +7,6 @@ import { EmDashBodyEnd, } from "emdash/ui"; import { createPublicPageContext } from "emdash/page"; -import LiveSearch from "emdash/ui/search"; import { Font } from "astro:assets"; import { resolveBlogSiteIdentity } from "../utils/site-identity"; import "../styles/theme.css"; @@ -110,14 +109,23 @@ const isLoggedIn = !!Astro.locals.user; }