diff --git a/docs/local-dev-loop.md b/docs/local-dev-loop.md index ec0003d6..800c4bed 100644 --- a/docs/local-dev-loop.md +++ b/docs/local-dev-loop.md @@ -6,8 +6,8 @@ Flatbread's local loop has four moving parts: content paths. 2. **Schema rebuild** — `@flatbread/core` turns loaded records and refs into a GraphQL schema after ID/ref validation. -3. **Codegen refresh** — `flatbread codegen --watch` regenerates TypeScript - artifacts when config, content, or GraphQL documents change. +3. **Codegen refresh** — the unified watcher regenerates TypeScript artifacts + when config, content, or GraphQL documents change. 4. **Framework restart / refresh** — `flatbread start -- ` runs the GraphQL server beside your app command. @@ -26,24 +26,21 @@ cd examples/nextjs pnpm exec flatbread codegen --verbose ``` -For development, use two terminals. This path avoids the example package's -HTTPS convenience script and keeps the Flatbread GraphQL endpoint on plain HTTP -port `5057`. +For development, use the unified watcher. This path avoids the example +package's HTTPS convenience script and keeps the Flatbread GraphQL endpoint on +plain HTTP port `5057`. ```bash -# terminal 1 — regenerate TypeScript artifacts -pnpm exec flatbread codegen --watch --verbose -``` - -```bash -# terminal 2 — serve GraphQL + Next.js without HTTPS for headless/dev agents +# serve GraphQL, refresh generated artifacts, and run Next.js without HTTPS pnpm exec flatbread start --watch -- next dev --turbopack ``` Expected behavior: +- One unified watcher owns config/content/document classification, GraphQL + hot-swaps, and generated artifact refreshes. - Editing a `.graphql` document or a content/config file refreshes - `generated/graphql.ts`. + `generated/graphql.ts`; do not run `flatbread codegen --watch` beside it. - The generated content-model types and prototype read API are refreshed by the same codegen command. - The running GraphQL endpoint at `http://localhost:5057/graphql` hot-swaps @@ -51,20 +48,20 @@ Expected behavior: ## Current reload matrix -| Change | Codegen watcher behavior | Running GraphQL server | Framework app | Action required today | -| --------------------------------------- | --------------------------------------- | ------------------------------------------------ | -------------------------------------------------------- | -------------------------------------------------------- | -| Markdown/YAML field value | Refreshes types if watched path matches | Hot-swaps after validation | Keeps rendering whatever the endpoint returns | None; framework refresh remains explicit | -| New/removed content file | Refreshes types if watched path matches | Hot-swaps after validation | Keeps rendering whatever the endpoint returns | None; framework refresh remains explicit | -| `.graphql` document | Regenerates operation types | No restart unless query text used by app changed | Framework dev server normally recompiles importing files | No Flatbread restart unless app code needs it | -| `flatbread.config.*` content/ref change | Reloads config and refreshes types | Rebuilds and hot-swaps after validation | Keeps rendering whatever the endpoint returns | None; framework refresh remains explicit | -| Transformer/source package code | Does not rebuild package code | Keeps previous imported package code | May keep previous imported package code | Rebuild/watch package separately, rerun codegen, restart | -| `generated/graphql.ts` | Output of codegen | No direct effect | Framework dev server recompiles imports | No Flatbread restart | +| Change | Unified watcher behavior | Running GraphQL server | Framework app | Action required today | +| --------------------------------------- | ------------------------------------------------------------------------------ | ------------------------------------------------ | -------------------------------------------------------- | -------------------------------------------------------- | +| Markdown/YAML field value | Atomically reindexes/hot-swaps, then refreshes codegen | Hot-swaps after validation | Keeps rendering whatever the endpoint returns | None; framework refresh remains explicit | +| New/removed content file | Atomically reindexes/hot-swaps, then refreshes codegen | Hot-swaps after validation | Keeps rendering whatever the endpoint returns | None; framework refresh remains explicit | +| `.graphql` document | Refreshes codegen only | No restart unless query text used by app changed | Framework dev server normally recompiles importing files | No Flatbread restart unless app code needs it | +| `flatbread.config.*` content/ref change | Reloads config/matchers, atomically rebuilds/hot-swaps, then refreshes codegen | Rebuilds and hot-swaps after validation | Keeps rendering whatever the endpoint returns | None; framework refresh remains explicit | +| Transformer/source package code | Does not rebuild package code | Keeps previous imported package code | May keep previous imported package code | Rebuild/watch package separately, rerun codegen, restart | +| `generated/graphql.ts` | Output of codegen | No direct effect | Framework dev server recompiles imports | No Flatbread restart | ## Failure semantics today -- If content becomes invalid while `flatbread codegen --watch` is running, the - watcher logs the validation/codegen error and keeps watching. Existing - generated files are left as-is until a later successful regeneration. +- Rejected config, content, or codegen phases are logged and keep the unified + watch loop alive. Existing generated files are left as-is until a later + successful regeneration. - In one-shot mode (`flatbread codegen` without `--watch`), validation or codegen errors exit non-zero and do not prove the live server changed. - If the running GraphQL server was started before the invalid edit, it keeps @@ -72,40 +69,41 @@ Expected behavior: validation error at startup. - In unified watch mode, invalid candidates are rejected atomically: generated artifacts and the live GraphQL server remain on the previous committed graph. + A codegen failure does not undo an already committed GraphQL generation, and + edits received during an in-flight generation are queued for the next + serialized batch. -## Draft unified watch design (implemented) +## Unified watch coordinator contract -The unified loop should eventually make this one command: +The unified loop is started with: ```bash -flatbread start --watch -- next dev --turbopack +pnpm exec flatbread start --watch -- next dev --turbopack ``` -Design contract: - -1. Watch the same content/config/document paths that `flatbread codegen --watch` - already derives from `LoadedFlatbreadConfig`. -2. On content changes, reload records, rerun ID/ref/cardinality validation, - rebuild the schema, refresh generated TypeScript, and swap the GraphQL - server schema only if the new graph validates. If validation fails, keep the - previous schema active and log the failure. -3. On config changes, reload config, rebuild watch globs, rebuild schema, - refresh generated TypeScript, and restart only the Flatbread GraphQL server - boundary if a safe hot swap is not possible. A safe hot swap means replacing - schema/data without losing the child framework process, open port, or - in-flight request handling state. -4. On GraphQL document changes, refresh generated TypeScript only. -5. Keep framework restarts explicit. Flatbread should not assume every - framework can be restarted safely; it should document whether the app command - is left running, restarted, or expected to recompile through its own dev - server. +The coordinator contract is: + +1. A single coordinator classifies config, content, and document events, then + serializes all rebuild and codegen phases. +2. Content changes reindex records and atomically hot-swap the GraphQL schema + before refreshing generated TypeScript. +3. Config changes reload the config and matchers, rebuild and atomically + hot-swap the schema, then refresh generated TypeScript. +4. GraphQL document changes refresh generated TypeScript without reindexing + content. +5. Rejected phases emit an error but do not stop the loop. Committed GraphQL + generations are not rolled back when a later codegen phase fails. +6. Events received during an in-flight generation are queued and processed + serially. +7. Framework restarts remain explicit. Flatbread keeps the framework child + process running and relies on its own dev server to recompile or refresh. ## Known limitations - `flatbread start --watch` hot-swaps valid content/config generations; invalid candidates leave the prior schema active. -- `flatbread codegen --watch` is a long-running process; do not use it in CI or - one-shot scripts. +- Unified watch mode is a long-running process; do not use it in CI or one-shot + scripts. - The Next.js example `pnpm dev` includes `--https` for local convenience, but the Flatbread GraphQL endpoint remains documented as HTTP on `5057`. In headless environments prefer `pnpm exec flatbread start -- next dev --turbopack`. diff --git a/packages/codegen/src/generator.ts b/packages/codegen/src/generator.ts index a41fdffd..274bd22a 100644 --- a/packages/codegen/src/generator.ts +++ b/packages/codegen/src/generator.ts @@ -16,8 +16,12 @@ import { readFile, writeFile } from 'node:fs/promises'; import kleur from 'kleur'; // @ts-ignore - chokidar types will be available after npm install import chokidar from 'chokidar'; -import { generateSchema } from '@flatbread/core'; -import type { LoadedFlatbreadConfig } from '@flatbread/core'; +import { createWatchCoordinator, generateSchema } from '@flatbread/core'; +import type { + LoadedFlatbreadConfig, + WatchCoordinatorResult, + WatchEventType, +} from '@flatbread/core'; import { generateInstallCommand } from '@flatbread/utils'; import type { CodegenOptions, CodegenResult, CodegenCache } from './types.js'; import { DEFAULT_CODEGEN_OPTIONS, PLUGIN_PRESETS } from './types.js'; @@ -40,6 +44,29 @@ function emitMissingDepsWarning(missingDeps: string[]) { ); } +function deriveOptionsFromConfig( + loadedConfig: LoadedFlatbreadConfig, + previous: CodegenOptions +): CodegenOptions { + const cfg = loadedConfig.codegen || {}; + return { + // Always enable in watch + enabled: true, + // Prefer latest config for dynamic fields changed via config + outputDir: cfg.outputDir ?? DEFAULT_CODEGEN_OPTIONS.outputDir, + outputFile: cfg.outputFile ?? DEFAULT_CODEGEN_OPTIONS.outputFile, + documents: cfg.documents ?? [], + plugins: cfg.plugins ?? previous.plugins, + pluginConfig: cfg.pluginConfig ?? previous.pluginConfig, + schema: cfg.schema ?? previous.schema, + codegenConfig: cfg.codegenConfig ?? previous.codegenConfig, + // Preserve runtime flags + watch: true, + cache: previous.cache, + preset: previous.preset, + }; +} + /** * Generate TypeScript types from a GraphQL schema using GraphQL Code Generator */ @@ -599,33 +626,12 @@ export async function watchAndGenerate( // Maintain a mutable set of options that can be refreshed when the config changes let currentOptions: CodegenOptions = { ...options }; - // Helper to derive effective codegen options from the latest config - const deriveOptionsFromConfig = ( - loadedConfig: LoadedFlatbreadConfig, - previous: CodegenOptions - ): CodegenOptions => { - const cfg = loadedConfig.codegen || {}; - return { - // Always enable in watch - enabled: true, - // Prefer latest config for dynamic fields changed via config - outputDir: cfg.outputDir ?? DEFAULT_CODEGEN_OPTIONS.outputDir, - outputFile: cfg.outputFile ?? DEFAULT_CODEGEN_OPTIONS.outputFile, - documents: cfg.documents ?? [], - plugins: cfg.plugins ?? previous.plugins, - pluginConfig: cfg.pluginConfig ?? previous.pluginConfig, - schema: cfg.schema ?? previous.schema, - codegenConfig: cfg.codegenConfig ?? previous.codegenConfig, - // Preserve runtime flags - watch: true, - cache: previous.cache, - preset: previous.preset, - }; - }; - // Initial generation using the current options await generateTypes(schema, config, currentOptions); + let currentConfig = config; + let currentSchema = schema; + // Set up file watchers const patterns = flattenFlatbreadWatchPatterns( deriveFlatbreadWatchPatterns(config, currentOptions) @@ -650,81 +656,76 @@ export async function watchAndGenerate( persistent: true, }); - let regenerating = false; - - const regenerateTypes = async (path: string, event: string) => { - if (regenerating) { - return; // Avoid concurrent regenerations - } - - regenerating = true; - - try { - console.log(kleur.yellow(`\n📝 ${event}: ${path}`)); - console.log(kleur.blue('🔄 Regenerating schema and types...')); - - let currentConfig = config; - - // If a config file changed, reload the configuration - if (path.includes('flatbread.config.')) { - try { - const { loadConfig } = await import('@flatbread/config'); - const configResult = await loadConfig({ cwd: process.cwd() }); - - if (configResult.config) { - currentConfig = configResult.config; - console.log(kleur.dim('🔧 Configuration reloaded')); - // Refresh codegen options from the updated config so changes like outputFile/outputDir/documents are applied - currentOptions = deriveOptionsFromConfig( - currentConfig, - currentOptions - ); - } - } catch (error) { - console.warn( - kleur.yellow( - `⚠️ Failed to reload config, using existing: ${ - error instanceof Error ? error.message : 'Unknown error' - }` - ) - ); - } - } - - // Regenerate the schema first since the source files may have changed - const newSchema = await generateSchema({ config: currentConfig }); - - // Generate types with the new schema + const coordinator = createWatchCoordinator({ + config: currentConfig, + documentPatterns: (cfg) => cfg.codegen?.documents ?? [], + loadConfig: async () => { + const { loadConfig } = await import('@flatbread/config'); + return (await loadConfig({ cwd: process.cwd() })).config!; + }, + applyConfig: async (cfg) => { + currentConfig = cfg; + currentOptions = deriveOptionsFromConfig(cfg, currentOptions); + currentSchema = await generateSchema({ config: cfg }); + return { status: 'committed' }; + }, + reindexContent: async () => { + currentSchema = await generateSchema({ config: currentConfig }); + return { status: 'committed' }; + }, + refreshCodegen: async () => { const result = await generateTypes( - newSchema, + currentSchema, currentConfig, currentOptions ); - - if (result.success) { - console.log(kleur.green('✅ Types regenerated successfully')); - } else { - console.error( - kleur.red('❌ Failed to regenerate types:'), - result.error - ); + if (!result.success) { + throw new Error(result.error ?? 'Codegen failed'); } - } catch (error) { - console.error(kleur.red('❌ Error during regeneration:'), error); - } finally { - regenerating = false; + }, + }); + + const resultMessage = (result: WatchCoordinatorResult): void => { + if (result.status === 'rejected') { + const phase = + result.kind === 'config' + ? 'reload config' + : result.kind === 'content' + ? 'reindex content' + : result.kind === 'documents' + ? 'refresh documents' + : 'regenerate types'; + console.error(kleur.red(`❌ Failed to ${phase}:`), result.error); + return; } + + if (result.kind === 'config') { + console.log(kleur.dim('🔧 Configuration reloaded')); + } else if (result.kind === 'codegen' || result.kind === 'documents') { + console.log(kleur.green('✅ Types regenerated successfully')); + } + }; + + coordinator.subscribe(resultMessage); + + const eventLabels: Record = { + create: 'File added', + update: 'File changed', + delete: 'File removed', + }; + const pushEvent = (path: string, type: WatchEventType) => { + console.log(kleur.yellow(`\n📝 ${eventLabels[type]}: ${path}`)); + console.log(kleur.blue('🔄 Regenerating schema and types...')); + coordinator.push([{ path, type }]); }; // Set up event handlers watcher - .on('add', (path: string) => regenerateTypes(path, 'File added')) - .on('change', (path: string) => regenerateTypes(path, 'File changed')) - .on('unlink', (path: string) => regenerateTypes(path, 'File removed')) - .on('addDir', (path: string) => regenerateTypes(path, 'Directory added')) - .on('unlinkDir', (path: string) => - regenerateTypes(path, 'Directory removed') - ) + .on('add', (path: string) => pushEvent(path, 'create')) + .on('change', (path: string) => pushEvent(path, 'update')) + .on('unlink', (path: string) => pushEvent(path, 'delete')) + .on('addDir', (path: string) => pushEvent(path, 'create')) + .on('unlinkDir', (path: string) => pushEvent(path, 'delete')) .on('error', (error: Error) => console.error(kleur.red('Watcher error:'), error) ) @@ -733,6 +734,7 @@ export async function watchAndGenerate( // Handle graceful shutdown const shutdown = () => { console.log(kleur.yellow('\n🛑 Shutting down watcher...')); + void coordinator.dispose(); watcher.close(); process.exit(0); }; diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 2e4880a1..5ad90ba8 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -22,6 +22,19 @@ export type { PathClassification, RecordsByCollection, } from './records'; +export { + createWatchCoordinator, + type WatchAdapterGeneration, + type WatchContentChange, + type WatchCoordinator, + type WatchCoordinatorOptions, + type WatchCoordinatorResult, + type WatchEvent, + type WatchEventType, + type WatchGenerationKind, + type WatchScheduler, + type WatchTimer, +} from './watch/coordinator'; export * from './types'; export { FlatbreadProvider } from './providers/base'; diff --git a/packages/core/src/watch/coordinator.ts b/packages/core/src/watch/coordinator.ts new file mode 100644 index 00000000..bf62d88c --- /dev/null +++ b/packages/core/src/watch/coordinator.ts @@ -0,0 +1,384 @@ +import { isMatch } from 'matcher'; +import { relative, resolve } from 'node:path'; +import type { LoadedFlatbreadConfig } from '../types'; +import { classifyPath } from '../records'; +import type { PathClassification } from '../records'; + +export type WatchEventType = 'create' | 'update' | 'delete'; +export interface WatchEvent { + readonly path: string; + readonly type: WatchEventType; +} +export interface WatchContentChange extends PathClassification { + readonly path: string; + readonly type: WatchEventType; +} +export type WatchGenerationKind = + | 'config' + | 'content' + | 'documents' + | 'codegen'; +export type WatchAdapterGeneration = + | { readonly status: 'committed'; readonly generation?: number } + | { + readonly status: 'rejected'; + readonly generation?: number; + readonly error: Error; + }; +export type WatchCoordinatorResult = + | { + readonly status: 'committed'; + readonly sequence: number; + readonly kind: WatchGenerationKind; + readonly config: LoadedFlatbreadConfig; + readonly content: readonly WatchContentChange[]; + readonly adapterGeneration?: number; + } + | { + readonly status: 'rejected'; + readonly sequence: number; + readonly kind: WatchGenerationKind; + readonly config: LoadedFlatbreadConfig; + readonly content: readonly WatchContentChange[]; + readonly adapterGeneration?: number; + readonly error: Error; + }; +export interface WatchTimer { + readonly cancel: () => void; +} +export interface WatchScheduler { + readonly schedule: (delayMs: number, callback: () => void) => WatchTimer; +} +export interface WatchCoordinatorOptions { + readonly config: LoadedFlatbreadConfig; + readonly cwd?: string; + readonly debounceMs?: number; + readonly configPatterns?: readonly string[]; + readonly documentPatterns: ( + config: LoadedFlatbreadConfig + ) => readonly string[]; + readonly loadConfig: () => Promise; + readonly applyConfig: ( + config: LoadedFlatbreadConfig + ) => Promise; + readonly reindexContent: ( + changes: readonly WatchContentChange[] + ) => Promise; + readonly refreshCodegen: (context: { + readonly reason: 'config' | 'content' | 'documents'; + readonly config: LoadedFlatbreadConfig; + readonly content: readonly WatchContentChange[]; + }) => Promise; + readonly scheduler?: WatchScheduler; +} +export interface WatchCoordinator { + readonly push: (events: readonly WatchEvent[]) => void; + readonly flush: () => Promise; + readonly drain: () => Promise; + readonly subscribe: ( + listener: (result: WatchCoordinatorResult) => void + ) => () => void; + readonly dispose: () => Promise; +} + +const defaultScheduler: WatchScheduler = { + schedule: (delayMs, callback) => { + const timer = setTimeout(callback, delayMs); + return { cancel: () => clearTimeout(timer) }; + }, +}; + +function expandBraces(pattern: string): string[] { + const match = /\{([^{}]*)\}/.exec(pattern); + if (!match) return [pattern]; + return match[1] + .split(',') + .flatMap((part) => + expandBraces( + `${pattern.slice(0, match.index)}${part}${pattern.slice( + match.index + match[0].length + )}` + ) + ); +} + +function matches(patterns: readonly string[], path: string): boolean { + return patterns + .flatMap(expandBraces) + .some((pattern) => isMatch(path, pattern, { caseSensitive: true })); +} + +function asError(error: unknown): Error { + return error instanceof Error ? error : new Error(String(error)); +} + +export function createWatchCoordinator( + options: WatchCoordinatorOptions +): WatchCoordinator { + const cwd = options.cwd ?? process.cwd(); + if (cwd !== process.cwd()) { + throw new Error( + `Watch coordinator cwd must equal process.cwd(): ${process.cwd()}` + ); + } + const scheduler = options.scheduler ?? defaultScheduler; + const debounceMs = options.debounceMs ?? 150; + const configPatterns = options.configPatterns ?? ['flatbread.config.*']; + let activeConfig = options.config; + let pending = new Map(); + let timer: WatchTimer | undefined; + let processing = false; + let disposed = false; + let sequence = 0; + const listeners = new Set<(result: WatchCoordinatorResult) => void>(); + const idleWaiters: Array<() => void> = []; + let forceNext = false; + + const emit = ( + status: 'committed' | 'rejected', + kind: WatchGenerationKind, + config: LoadedFlatbreadConfig, + content: readonly WatchContentChange[], + generation?: number, + error?: Error + ) => { + const result = { + status, + sequence: ++sequence, + kind, + config, + content, + ...(generation === undefined ? {} : { adapterGeneration: generation }), + ...(error === undefined ? {} : { error }), + } as WatchCoordinatorResult; + for (const listener of listeners) listener(result); + }; + + const isIdle = () => !timer && !processing && pending.size === 0; + const resolveIdle = () => { + if (!isIdle()) return; + while (idleWaiters.length) idleWaiters.shift()!(); + }; + const waitForIdle = () => + isIdle() + ? Promise.resolve() + : new Promise((resolve) => idleWaiters.push(resolve)); + + const start = async (): Promise => { + if (processing || disposed || pending.size === 0) return; + if (timer) { + timer.cancel(); + timer = undefined; + } + processing = true; + const batch = pending; + pending = new Map(); + const raw = [...batch.values()]; + let configChanged = false; + const relativePath = (path: string) => + relative(cwd, resolve(path)).split('\\').join('/'); + try { + configChanged = raw.some((event) => + matches(configPatterns, relativePath(event.path)) + ); + if (configChanged) { + let loaded: LoadedFlatbreadConfig | undefined; + try { + loaded = await options.loadConfig(); + const result = await options.applyConfig(loaded); + if (result.status === 'committed') { + activeConfig = loaded; + emit('committed', 'config', activeConfig, [], result.generation); + try { + await options.refreshCodegen({ + reason: 'config', + config: activeConfig, + content: [], + }); + emit('committed', 'codegen', activeConfig, []); + } catch (error) { + emit( + 'rejected', + 'codegen', + activeConfig, + [], + undefined, + asError(error) + ); + } + } else { + emit( + 'rejected', + 'config', + activeConfig, + [], + result.generation, + result.error + ); + } + } catch (error) { + emit( + 'rejected', + 'config', + activeConfig, + [], + undefined, + asError(error) + ); + } + } + const content: WatchContentChange[] = []; + let documents = false; + for (const event of raw) { + const path = resolve(event.path); + if (matches(configPatterns, relativePath(path))) continue; + const classification = classifyPath(path, activeConfig); + if (classification) { + const prior = content.findIndex((change) => change.path === path); + const change = { ...classification, path, type: event.type }; + if (prior < 0) content.push(change); + else if (content[prior].type !== 'delete') content[prior] = change; + continue; + } + if ( + matches(options.documentPatterns(activeConfig), relativePath(path)) + ) { + documents = true; + } + } + if (content.length) { + try { + const result = await options.reindexContent(content); + if (result.status === 'committed') { + emit( + 'committed', + 'content', + activeConfig, + content, + result.generation + ); + try { + await options.refreshCodegen({ + reason: 'content', + config: activeConfig, + content, + }); + emit('committed', 'codegen', activeConfig, content); + } catch (error) { + emit( + 'rejected', + 'codegen', + activeConfig, + content, + undefined, + asError(error) + ); + } + } else { + emit( + 'rejected', + 'content', + activeConfig, + content, + result.generation, + result.error + ); + } + } catch (error) { + emit( + 'rejected', + 'content', + activeConfig, + content, + undefined, + asError(error) + ); + } + } + if (documents) { + try { + await options.refreshCodegen({ + reason: 'documents', + config: activeConfig, + content: [], + }); + emit('committed', 'documents', activeConfig, []); + } catch (error) { + emit( + 'rejected', + 'documents', + activeConfig, + [], + undefined, + asError(error) + ); + } + } + } finally { + processing = false; + if (!disposed && pending.size) { + if (forceNext) { + forceNext = false; + void start(); + } else { + timer = scheduler.schedule(debounceMs, () => { + timer = undefined; + void start(); + }); + } + } else { + // Nothing follows this batch; a latched flush intent must not leak + // into a later unrelated push and skip its debounce window. + forceNext = false; + } + resolveIdle(); + } + }; + + const push = (events: readonly WatchEvent[]) => { + if (disposed) return; + for (const event of events) { + const path = resolve(event.path); + const previous = pending.get(path); + if (!previous || previous.type !== 'delete') { + pending.set(path, { path, type: event.type }); + } + } + if (!processing && !timer) { + timer = scheduler.schedule(debounceMs, () => { + timer = undefined; + void start(); + }); + } + }; + + return { + push, + flush: async () => { + if (disposed) return; + if (isIdle()) return; + if (timer) { + timer.cancel(); + timer = undefined; + } + forceNext = true; + if (!processing) void start(); + await waitForIdle(); + }, + drain: waitForIdle, + subscribe: (listener) => { + listeners.add(listener); + return () => listeners.delete(listener); + }, + dispose: async () => { + if (disposed) return; + disposed = true; + if (timer) { + timer.cancel(); + timer = undefined; + } + pending.clear(); + await (processing ? waitForIdle() : Promise.resolve()); + }, + }; +} diff --git a/packages/core/src/watch/tests/coordinator.test.ts b/packages/core/src/watch/tests/coordinator.test.ts new file mode 100644 index 00000000..6eaa591a --- /dev/null +++ b/packages/core/src/watch/tests/coordinator.test.ts @@ -0,0 +1,421 @@ +import test from 'ava'; +import { resolve } from 'node:path'; +import type { + WatchAdapterGeneration, + WatchContentChange, + WatchCoordinator, + WatchCoordinatorResult, + WatchScheduler, + WatchTimer, +} from '../coordinator'; +import { createWatchCoordinator } from '../coordinator'; +import type { LoadedFlatbreadConfig } from '../../types'; + +const cwd = process.cwd(); +const contentPath = resolve('watch/content/post.md'); +const secondContentPath = resolve('watch/content/other.md'); + +function config( + documents: string[] = ['src/**/*.graphql'] +): LoadedFlatbreadConfig { + return { + source: { fetch: async () => ({}) }, + transformer: [], + content: [{ path: 'watch/content', collection: 'Post' }], + fieldNameTransform: (field) => field, + loaded: { extensions: ['md'] }, + codegen: { documents }, + }; +} + +class ManualScheduler implements WatchScheduler { + callbacks: Array<{ callback: () => void; cancelled: boolean }> = []; + schedule(_delayMs: number, callback: () => void): WatchTimer { + const entry = { callback, cancelled: false }; + this.callbacks.push(entry); + return { cancel: () => (entry.cancelled = true) }; + } + fire() { + const entries = this.callbacks.splice(0); + for (const entry of entries) if (!entry.cancelled) entry.callback(); + } +} + +function deferred() { + let resolvePromise!: (value: T) => void; + const promise = new Promise((resolve) => (resolvePromise = resolve)); + return { promise, resolve: resolvePromise }; +} + +function make( + overrides: Partial[0]> = {} +): { + coordinator: WatchCoordinator; + scheduler: ManualScheduler; + calls: string[]; + results: WatchCoordinatorResult[]; +} { + const scheduler = new ManualScheduler(); + const calls: string[] = []; + const results: WatchCoordinatorResult[] = []; + const coordinator = createWatchCoordinator({ + config: config(), + scheduler, + documentPatterns: (cfg) => cfg.codegen?.documents ?? [], + loadConfig: async () => config(), + applyConfig: async () => ({ status: 'committed', generation: 1 }), + reindexContent: async (changes) => { + calls.push(`reindex:${changes.map((change) => change.type).join(',')}`); + return { status: 'committed', generation: 1 }; + }, + refreshCodegen: async ({ reason }) => { + calls.push(`codegen:${reason}`); + }, + ...overrides, + }); + coordinator.subscribe((result) => results.push(result)); + return { coordinator, scheduler, calls, results }; +} + +test.serial( + 'classifies config, content collection/captures, documents, and unmatched paths', + async (t) => { + const calls: string[] = []; + const { coordinator, results } = make({ + config: { + ...config(), + content: [{ path: 'watch/[section]/post.md', collection: 'Section' }], + }, + reindexContent: async (changes) => { + calls.push( + `${changes[0].collection}:${changes[0].captures.section}:${changes[0].type}` + ); + return { status: 'committed' }; + }, + refreshCodegen: async ({ reason }) => { + calls.push(reason); + }, + loadConfig: async () => ({ + ...config(), + content: [{ path: 'watch/[section]/post.md', collection: 'Section' }], + }), + }); + coordinator.push([ + { path: 'flatbread.config.ts', type: 'update' }, + { path: 'watch/news/post.md', type: 'update' }, + { path: 'src/queries/a.graphql', type: 'update' }, + { path: 'ignored.txt', type: 'update' }, + ]); + await coordinator.flush(); + t.deepEqual(calls, [ + 'config', + 'Section:news:update', + 'content', + 'documents', + ]); + t.deepEqual( + results.map(({ kind, status }) => `${kind}:${status}`), + [ + 'config:committed', + 'codegen:committed', + 'content:committed', + 'codegen:committed', + 'documents:committed', + ] + ); + } +); + +test.serial( + 'coalesces multiple content events into one reindex and lets delete win over recreate', + async (t) => { + const { coordinator, calls } = make(); + coordinator.push([ + { path: contentPath, type: 'update' }, + { path: contentPath, type: 'update' }, + { path: contentPath, type: 'delete' }, + { path: contentPath, type: 'create' }, + ]); + await coordinator.flush(); + t.deepEqual(calls, ['reindex:delete', 'codegen:content']); + } +); + +test.serial( + 'reloads config before classifying remaining paths and recompiles document matchers', + async (t) => { + const newConfig = { + ...config(['new/*.graphql']), + content: [{ path: 'new/content', collection: 'New' }], + }; + const order: string[] = []; + const { coordinator } = make({ + loadConfig: async () => { + order.push('load'); + return newConfig; + }, + applyConfig: async () => { + order.push('apply'); + return { status: 'committed' }; + }, + reindexContent: async (changes) => { + order.push(`reindex:${changes[0].collection}`); + return { status: 'committed' }; + }, + refreshCodegen: async ({ reason }) => { + order.push(`codegen:${reason}`); + }, + config: { + ...config(['old/**/*.graphql']), + content: [{ path: 'old', collection: 'Old' }], + }, + }); + coordinator.push([ + { path: 'flatbread.config.ts', type: 'update' }, + { path: 'new/content/post.md', type: 'update' }, + { path: 'new/query.graphql', type: 'update' }, + ]); + await coordinator.flush(); + t.deepEqual(order, [ + 'load', + 'apply', + 'codegen:config', + 'reindex:New', + 'codegen:content', + 'codegen:documents', + ]); + } +); + +test.serial( + 'queues events received during an in-flight live-style reindex for the next batch', + async (t) => { + const gate = deferred(); + const calls: string[] = []; + const { coordinator } = make({ + reindexContent: async (changes) => { + calls.push(changes[0].path); + if (calls.length === 1) await gate.promise; + return { status: 'committed' }; + }, + refreshCodegen: async () => {}, + }); + coordinator.push([{ path: contentPath, type: 'update' }]); + const first = coordinator.flush(); + await Promise.resolve(); + coordinator.push([{ path: secondContentPath, type: 'update' }]); + gate.resolve(); + await first; + await coordinator.drain(); + t.deepEqual(calls, [contentPath, secondContentPath]); + } +); + +test.serial( + 'queues events received during an in-flight codegen-style rebuild for the next batch', + async (t) => { + const gate = deferred(); + let refreshes = 0; + const { coordinator } = make({ + reindexContent: async () => ({ status: 'committed' }), + refreshCodegen: async () => { + refreshes++; + if (refreshes === 1) await gate.promise; + }, + }); + coordinator.push([{ path: contentPath, type: 'update' }]); + const first = coordinator.flush(); + await Promise.resolve(); + coordinator.push([{ path: secondContentPath, type: 'update' }]); + gate.resolve(); + await first; + await coordinator.drain(); + t.is(refreshes, 2); + } +); + +test.serial( + 'recovers after rejected content generation and processes the next batch', + async (t) => { + let count = 0; + const { coordinator, results, calls } = make({ + reindexContent: async () => { + count++; + return count === 1 + ? ({ + status: 'rejected', + error: new Error('bad'), + } as WatchAdapterGeneration) + : ({ status: 'committed', generation: 2 } as WatchAdapterGeneration); + }, + }); + coordinator.push([{ path: contentPath, type: 'update' }]); + await coordinator.flush(); + coordinator.push([{ path: secondContentPath, type: 'update' }]); + await coordinator.flush(); + t.is( + results.filter( + (result) => result.kind === 'content' && result.status === 'rejected' + ).length, + 1 + ); + t.deepEqual(calls, ['codegen:content']); + } +); + +test.serial( + 'schedules codegen after committed content and documents without reindex', + async (t) => { + const { coordinator, calls } = make(); + coordinator.push([{ path: contentPath, type: 'update' }]); + await coordinator.flush(); + coordinator.push([{ path: 'src/queries/query.graphql', type: 'update' }]); + await coordinator.flush(); + t.deepEqual(calls, [ + 'reindex:update', + 'codegen:content', + 'codegen:documents', + ]); + } +); + +test.serial( + 'does not schedule codegen after a rejected config replacement', + async (t) => { + const { coordinator, results, calls } = make({ + applyConfig: async () => ({ + status: 'rejected', + error: new Error('invalid'), + }), + }); + coordinator.push([{ path: 'flatbread.config.ts', type: 'update' }]); + await coordinator.flush(); + t.false(calls.includes('codegen:config')); + t.is( + results.find((result) => result.kind === 'config')?.status, + 'rejected' + ); + coordinator.push([{ path: 'src/queries/query.graphql', type: 'update' }]); + await coordinator.flush(); + t.true(calls.includes('codegen:documents')); + } +); + +test.serial( + 'reports codegen rejection without stopping the next batch', + async (t) => { + let count = 0; + const { coordinator, results } = make({ + refreshCodegen: async () => { + count++; + if (count === 1) throw new Error('codegen'); + }, + }); + coordinator.push([{ path: contentPath, type: 'update' }]); + await coordinator.flush(); + coordinator.push([{ path: secondContentPath, type: 'update' }]); + await coordinator.flush(); + t.is( + results.filter( + (result) => result.kind === 'codegen' && result.status === 'rejected' + ).length, + 1 + ); + t.is( + results.filter( + (result) => result.kind === 'content' && result.status === 'committed' + ).length, + 2 + ); + } +); + +test.serial( + 'flush and dispose control deterministic scheduler lifecycle', + async (t) => { + const scheduler = new ManualScheduler(); + const gate = deferred(); + const { coordinator } = make({ + scheduler, + reindexContent: async () => { + await gate.promise; + return { status: 'committed' }; + }, + }); + coordinator.push([{ path: contentPath, type: 'update' }]); + t.is(scheduler.callbacks.length, 1); + const running = coordinator.flush(); + coordinator.push([{ path: secondContentPath, type: 'update' }]); + const disposed = coordinator.dispose(); + gate.resolve(); + await Promise.all([running, disposed]); + t.is(scheduler.callbacks.filter((entry) => !entry.cancelled).length, 0); + t.pass(); + } +); + +test.serial( + 'flush while idle does not latch and skip a later debounce window', + async (t) => { + const scheduler = new ManualScheduler(); + const gate = deferred(); + const reindexed: string[] = []; + let blockFirst = true; + const { coordinator } = make({ + scheduler, + reindexContent: async (changes) => { + reindexed.push(changes[0].path); + if (blockFirst) { + blockFirst = false; + await gate.promise; + } + return { status: 'committed' }; + }, + refreshCodegen: async () => {}, + }); + + // Idle flush: nothing pending, nothing in flight; must be a no-op that + // does not latch the skip-debounce intent. + await coordinator.flush(); + t.is(scheduler.callbacks.length, 0); + + // A later unrelated push must still get its debounce timer. + coordinator.push([{ path: contentPath, type: 'update' }]); + t.is(scheduler.callbacks.length, 1); + t.deepEqual(reindexed, [], 'batch must not run before the timer fires'); + scheduler.fire(); + + // First build is now in flight and blocked; queue a follow-on event. + coordinator.push([{ path: secondContentPath, type: 'update' }]); + gate.resolve(); + + // Let the first batch settle (bounded microtask drain; no real timers). + for (let i = 0; i < 20 && scheduler.callbacks.length === 0; i++) { + await Promise.resolve(); + } + + // The follow-on batch must receive its own debounce window instead of + // running immediately off a leaked forceNext. + t.is(scheduler.callbacks.length, 1); + t.deepEqual(reindexed, [contentPath]); + scheduler.fire(); + await coordinator.drain(); + t.deepEqual(reindexed, [contentPath, secondContentPath]); + } +); + +test.serial( + 'routes a brace-expansion document pattern to the documents phase', + async (t) => { + const calls: string[] = []; + const { coordinator } = make({ + documentPatterns: () => ['src/**/*.{graphql,gql}'], + refreshCodegen: async ({ reason }) => { + calls.push(reason); + }, + }); + coordinator.push([{ path: 'src/queries/thing.gql', type: 'update' }]); + await coordinator.flush(); + t.deepEqual(calls, ['documents']); + } +); diff --git a/packages/flatbread/src/graphql/liveServer.ts b/packages/flatbread/src/graphql/liveServer.ts index 7e3b68ce..f689bca5 100644 --- a/packages/flatbread/src/graphql/liveServer.ts +++ b/packages/flatbread/src/graphql/liveServer.ts @@ -1,23 +1,22 @@ import { subscribe } from '@parcel/watcher'; -import { - deriveFlatbreadWatchPatterns, - generateTypes, -} from '@flatbread/codegen'; +import { generateTypes } from '@flatbread/codegen'; import type { ConfigResult, LoadedFlatbreadConfig, LiveSchemaReloader, SchemaSnapshot, } from '@flatbread/core'; -import { createLiveSchemaReloader } from '@flatbread/core'; +import { + createLiveSchemaReloader, + createWatchCoordinator, + type WatchCoordinator, +} from '@flatbread/core'; import { ApolloServer } from '@apollo/server'; import { expressMiddleware } from '@as-integrations/express5'; import { InMemoryLRUCache } from '@apollo/utils.keyvaluecache'; import cors from 'cors'; import express, { type RequestHandler } from 'express'; import http from 'http'; -import picomatch from 'picomatch'; -import { relative, resolve } from 'node:path'; import { loadFlatbreadConfig } from '../utils/getSchema'; export interface GraphqlServerOptions { @@ -43,6 +42,11 @@ export async function startGraphqlServer( 'Flatbread watch mode requires the configured source to implement fetchPaths(paths).' ); } + if (options.watch && cwd !== process.cwd()) { + throw new Error( + `Flatbread watch mode requires cwd to equal process.cwd() (${process.cwd()}).` + ); + } const app = express(); const httpServer = http.createServer(app); let current: GenerationServer | undefined; @@ -105,30 +109,46 @@ export async function startGraphqlServer( ? address.port : options.port ?? 5050; let subscription: { unsubscribe(): Promise } | undefined; - let timer: ReturnType | undefined; - const pending = new Set(); - const pendingDeleted = new Set(); - let pendingConfig = false; - let pendingDocuments = false; - let matchers = compileMatchers(config); - const refreshCodegen = async ( - snapshot: SchemaSnapshot, - loadedConfig = snapshot.graph.config - ) => { - const cfg = loadedConfig.codegen; - // No codegen section in the config → nothing to refresh; avoids writing - // generated artifacts the user never asked for. - if (!cfg) return; - const result = await generateTypes(snapshot.schema, loadedConfig, { - ...cfg, - enabled: cfg.enabled ?? true, - documents: cfg.documents ?? [], - }); - if (!result.success) { - console.error('Flatbread codegen refresh failed:', result.error); - } - }; + let coordinator: WatchCoordinator | undefined; if (options.watch) { + coordinator = createWatchCoordinator({ + config, + cwd, + documentPatterns: (cfg) => cfg.codegen?.documents ?? [], + loadConfig: async () => (await loadFlatbreadConfig(cwd)).config!, + applyConfig: async (cfg) => reloader.replaceConfig(cfg), + reindexContent: async (changes) => + reloader.notifyChanged({ + paths: changes.map(({ path }) => path), + source: 'watcher', + }), + refreshCodegen: async ({ config: loadedConfig }) => { + const cfg = loadedConfig.codegen; + if (!cfg) return; + const result = await generateTypes( + reloader.getSnapshot().schema, + loadedConfig, + { + ...cfg, + enabled: cfg.enabled ?? true, + documents: cfg.documents ?? [], + } + ); + if (!result.success) { + throw new Error(result.error ?? 'Codegen refresh failed'); + } + }, + }); + coordinator.subscribe((result) => { + if (result.status !== 'rejected') return; + const label = + result.kind === 'config' + ? 'config reload' + : result.kind === 'content' + ? 'content reindex' + : 'codegen refresh'; + console.error(`Flatbread ${label} failed:`, result.error); + }); subscription = await subscribe( cwd, (error, events) => { @@ -136,54 +156,9 @@ export async function startGraphqlServer( console.error('Flatbread watcher error:', error); return; } - for (const event of events) { - const path = resolve(event.path); - const kind = classifyPath(path, matchers, cwd); - if (kind === 'config') pendingConfig = true; - else if (kind === 'document') pendingDocuments = true; - else if (kind === 'content') { - if (event.type === 'delete') { - pendingDeleted.add(path); - pending.add(path); - } else if (!pendingDeleted.has(path)) { - pending.add(path); - } - } - } - if (!timer) { - timer = setTimeout(async () => { - timer = undefined; - const paths = [...pending]; - pending.clear(); - pendingDeleted.clear(); - if (pendingConfig) { - pendingConfig = false; - try { - const loaded = await loadFlatbreadConfig(cwd); - const result = await reloader.replaceConfig(loaded.config!); - if (result.status === 'committed') { - matchers = compileMatchers(loaded.config!); - await refreshCodegen(reloader.getSnapshot(), loaded.config); - } else console.error(result.error); - } catch (error) { - console.error('Flatbread config reload failed:', error); - } - } - if (paths.length > 0) { - const result = await reloader.notifyChanged({ - paths, - source: 'watcher', - }); - if (result.status === 'committed') { - await refreshCodegen(reloader.getSnapshot()); - } else console.error(result.error); - } - if (pendingDocuments) { - pendingDocuments = false; - await refreshCodegen(reloader.getSnapshot()); - } - }, 150); - } + coordinator!.push( + events.map((event) => ({ path: event.path, type: event.type })) + ); }, { ignore: ['**/node_modules/**', '**/.git/**', '**/dist/**'] } ); @@ -194,7 +169,7 @@ export async function startGraphqlServer( async close() { if (closed) return; closed = true; - if (timer) clearTimeout(timer); + await coordinator?.dispose(); await subscription?.unsubscribe(); await current?.stop(); await new Promise((closeResolve) => { @@ -246,26 +221,3 @@ async function startGeneration( }; return { middleware: wrapped, stopWhenDrained, stop }; } -function compileMatchers(config: LoadedFlatbreadConfig) { - const patterns = deriveFlatbreadWatchPatterns(config, { - documents: config.codegen?.documents ?? [], - }); - return { - config: patterns.config.map((pattern) => picomatch(pattern)), - content: patterns.content.map((pattern) => picomatch(pattern)), - documents: patterns.documents.map((pattern) => picomatch(pattern)), - }; -} -function classifyPath( - path: string, - matchers: ReturnType, - cwd: string -): 'config' | 'content' | 'document' | undefined { - const relativePath = relative(cwd, path).split('\\').join('/'); - if (matchers.config.some((matcher) => matcher(relativePath))) return 'config'; - if (matchers.content.some((matcher) => matcher(relativePath))) - return 'content'; - if (matchers.documents.some((matcher) => matcher(relativePath))) - return 'document'; - return undefined; -}