From a40afed5145cdc4c62c6116f862c84771ab556ad Mon Sep 17 00:00:00 2001 From: djgrant <1670902+djgrant@users.noreply.github.com> Date: Sun, 19 Jul 2026 17:49:28 +0100 Subject: [PATCH] Expose reconciler integrations --- packages/cli/src/deploy.ts | 27 ++++- packages/cli/src/destroy.ts | 18 ++- packages/cli/src/index.ts | 13 +- packages/cli/src/plan.ts | 14 +-- packages/cli/src/stdio.ts | 17 +++ packages/core/src/provisioner/index.ts | 1 + .../core/src/provisioner/operations/index.ts | 4 - .../provisioner/operations/operation.base.ts | 27 ----- .../operations/operation.create.ts | 65 ---------- .../operations/operation.delete.ts | 35 ------ .../provisioner/operations/operation.read.ts | 66 ---------- .../operations/operation.update.ts | 44 ------- .../core/src/provisioner/workflows/index.ts | 5 + .../provisioner/workflows/workflow.deploy.ts | 4 +- .../provisioner/workflows/workflow.destroy.ts | 6 +- .../provisioner/workflows/workflow.plan.ts | 4 +- .../provisioner/workflows/workflow.refresh.ts | 4 +- .../provisioner/resource-registry.test.ts | 4 +- packages/dashboard/package.json | 2 +- packages/dashboard/server/server.test.ts | 30 +++++ packages/dashboard/server/server.ts | 114 +++++++++++------- packages/reconciler/src/index.ts | 1 + packages/reconciler/src/plan.ts | 2 +- packages/reconciler/src/protocol.ts | 17 +++ packages/reconciler/test/protocol.test.ts | 25 ++++ .../reconciler/test/reconciler.plan.test.ts | 37 ++++-- packages/resource/src/resource.schema.ts | 45 +++---- packages/resource/src/resource.ts | 13 +- packages/resource/src/types.ts | 15 ++- pnpm-lock.yaml | 6 +- 30 files changed, 303 insertions(+), 362 deletions(-) create mode 100644 packages/cli/src/stdio.ts delete mode 100644 packages/core/src/provisioner/operations/index.ts delete mode 100644 packages/core/src/provisioner/operations/operation.base.ts delete mode 100644 packages/core/src/provisioner/operations/operation.create.ts delete mode 100644 packages/core/src/provisioner/operations/operation.delete.ts delete mode 100644 packages/core/src/provisioner/operations/operation.read.ts delete mode 100644 packages/core/src/provisioner/operations/operation.update.ts create mode 100644 packages/dashboard/server/server.test.ts create mode 100644 packages/reconciler/src/protocol.ts create mode 100644 packages/reconciler/test/protocol.test.ts diff --git a/packages/cli/src/deploy.ts b/packages/cli/src/deploy.ts index f1211a0..b81d5ad 100644 --- a/packages/cli/src/deploy.ts +++ b/packages/cli/src/deploy.ts @@ -1,12 +1,33 @@ -import { deployApp } from "@notation/core"; +import { createNdjsonEventEmitter, deployApp } from "@notation/core"; import { compile } from "./compile"; +import { redirectStdoutToStderr } from "./stdio"; + +export type DeployCommandOptions = { + json?: boolean; +}; + +export async function deploy( + entryPoint: string, + opts: DeployCommandOptions = {}, +) { + // In --json mode console output moves to stderr so stdout carries only the + // NDJSON event stream; capture the real stdout for the emitter first. + const emit = opts.json + ? createNdjsonEventEmitter(redirectStdoutToStderr().write) + : undefined; -export async function deploy(entryPoint: string) { await compile(entryPoint); console.log(`Deploying ${entryPoint}`); try { - await deployApp(entryPoint); + await deployApp( + entryPoint, + undefined, + undefined, + undefined, + undefined, + emit, + ); } catch (err: any) { if (err.name === "CredentialsProviderError") { console.log( diff --git a/packages/cli/src/destroy.ts b/packages/cli/src/destroy.ts index 21c9d95..0186f97 100644 --- a/packages/cli/src/destroy.ts +++ b/packages/cli/src/destroy.ts @@ -1,7 +1,19 @@ -import { destroyApp } from "@notation/core"; +import { createNdjsonEventEmitter, destroyApp } from "@notation/core"; import { compile } from "./compile"; +import { redirectStdoutToStderr } from "./stdio"; + +export type DestroyCommandOptions = { + json?: boolean; +}; + +export async function destroy( + entryPoint: string, + opts: DestroyCommandOptions = {}, +) { + const emit = opts.json + ? createNdjsonEventEmitter(redirectStdoutToStderr().write) + : undefined; -export async function destroy(entryPoint: string) { await compile(entryPoint); - await destroyApp(entryPoint); + await destroyApp(entryPoint, undefined, undefined, emit); } diff --git a/packages/cli/src/index.ts b/packages/cli/src/index.ts index c0ce281..3ef90b7 100644 --- a/packages/cli/src/index.ts +++ b/packages/cli/src/index.ts @@ -7,6 +7,7 @@ import { plan } from "./plan"; import { visualise } from "./visualise"; import { watch } from "./watch"; import { startDashboardServer } from "@notation/dashboard"; +import { createDefaultStateBackend } from "@notation/core"; program .command("compile") @@ -20,23 +21,25 @@ program .command("dashboard") .description("Start Notation Dashboard") .action(async () => { - await startDashboardServer(); + await startDashboardServer({ state: createDefaultStateBackend() }); }); program .command("deploy") .argument("", "entryPoint") .description("Deploy Notation App") - .action(async (entryPoint) => { - await deploy(entryPoint); + .option("--json", "stream reconciler events as NDJSON") + .action(async (entryPoint, options) => { + await deploy(entryPoint, { json: options.json }); }); program .command("destroy") .argument("", "entryPoint") .description("Destroy Notation App") - .action(async (entryPoint) => { - await destroy(entryPoint); + .option("--json", "stream reconciler events as NDJSON") + .action(async (entryPoint, options) => { + await destroy(entryPoint, { json: options.json }); }); program diff --git a/packages/cli/src/plan.ts b/packages/cli/src/plan.ts index 24371cf..90fc49a 100644 --- a/packages/cli/src/plan.ts +++ b/packages/cli/src/plan.ts @@ -1,5 +1,6 @@ import { planApp, type Plan, type PlanNode } from "@notation/core"; import { compile } from "./compile"; +import { redirectStdoutToStderr } from "./stdio"; export type PlanCommandOptions = { json?: boolean; @@ -18,12 +19,12 @@ export async function plan(entryPoint: string, opts: PlanCommandOptions = {}) { try { if (opts.json) { let result: Plan; - const restoreStdout = redirectStdoutToStderr(); + const { restore } = redirectStdoutToStderr(); try { await compile(entryPoint); result = await planApp(entryPoint); } finally { - restoreStdout(); + restore(); } process.stdout.write(`${JSON.stringify(result, null, 2)}\n`); return; @@ -68,12 +69,3 @@ function printPlanSummary(result: Plan) { console.log(`${changedNodes.length > 0 ? "\n" : ""}Plan: ${summary}.`); } - -function redirectStdoutToStderr() { - const originalWrite = process.stdout.write.bind(process.stdout); - process.stdout.write = ((chunk: any, ...args: any[]) => - (process.stderr.write as any)(chunk, ...args)) as typeof process.stdout.write; - return () => { - process.stdout.write = originalWrite; - }; -} diff --git a/packages/cli/src/stdio.ts b/packages/cli/src/stdio.ts new file mode 100644 index 0000000..0102385 --- /dev/null +++ b/packages/cli/src/stdio.ts @@ -0,0 +1,17 @@ +export function redirectStdoutToStderr() { + const originalWrite = process.stdout.write.bind(process.stdout); + const write = (line: string): void => { + originalWrite(line); + }; + process.stdout.write = ((chunk: any, ...args: any[]) => + (process.stderr.write as any)( + chunk, + ...args, + )) as typeof process.stdout.write; + return { + write, + restore: () => { + process.stdout.write = originalWrite; + }, + }; +} diff --git a/packages/core/src/provisioner/index.ts b/packages/core/src/provisioner/index.ts index 4b757c5..89bf6e7 100644 --- a/packages/core/src/provisioner/index.ts +++ b/packages/core/src/provisioner/index.ts @@ -1,2 +1,3 @@ export * from "./workflows"; export * from "./resource-registry"; +export * from "./state-backend"; diff --git a/packages/core/src/provisioner/operations/index.ts b/packages/core/src/provisioner/operations/index.ts deleted file mode 100644 index 218ae0c..0000000 --- a/packages/core/src/provisioner/operations/index.ts +++ /dev/null @@ -1,4 +0,0 @@ -export * from "./operation.create"; -export * from "./operation.delete"; -export * from "./operation.read"; -export * from "./operation.update"; diff --git a/packages/core/src/provisioner/operations/operation.base.ts b/packages/core/src/provisioner/operations/operation.base.ts deleted file mode 100644 index 3f83873..0000000 --- a/packages/core/src/provisioner/operations/operation.base.ts +++ /dev/null @@ -1,27 +0,0 @@ -import { BaseResource } from "src/orchestrator/resource"; - -export const operation = ( - action: string, - operation: (opts: Opts) => Promise, -) => { - return async ( - opts: Opts & { dryRun?: boolean; quiet?: boolean }, - ): Promise => { - const { dryRun, quiet, ...opOpts } = opts; - const message = `${action} ${opOpts.resource.id}`; - - if (dryRun) { - console.log(`[Dry Run]: ${message}`); - return {} as V; - } - - try { - const result = await operation(opOpts as unknown as Opts); - if (!quiet) console.log(`[Success]: ${message}`); - return result; - } catch (err) { - console.error(`[Error]: ${message}`); - throw err; - } - }; -}; diff --git a/packages/core/src/provisioner/operations/operation.create.ts b/packages/core/src/provisioner/operations/operation.create.ts deleted file mode 100644 index a2d0fdd..0000000 --- a/packages/core/src/provisioner/operations/operation.create.ts +++ /dev/null @@ -1,65 +0,0 @@ -import { operation } from "./operation.base"; -import { BaseResource } from "src/orchestrator/resource"; -import type { StateBackend } from "@notation/state"; -import { readResource } from "."; - -export const createResource = operation("Creating", create); - -async function create( - opts: { resource: BaseResource; state: StateBackend }, - backoff = 1000, -) { - const { resource, state } = opts; - - try { - const params = await resource.getParams(); - const maybeComputedPrimaryKey = await resource.create(params); - - resource.setOutput(params); - - if (maybeComputedPrimaryKey) { - resource.setOutput({ ...maybeComputedPrimaryKey, ...resource.output }); - } - - const readResult = await readResource({ resource, state, quiet: true }); - - resource.setOutput({ ...resource.output, ...readResult }); - - await state.update(resource.id, 0, { - id: resource.id, - groupId: resource.groupId, - groupType: resource.groupType, - type: resource.type, - lastOperation: "create", - lastOperationAt: new Date().toISOString(), - config: resource.config, - params: resource.toState(params), - output: resource.toState(resource.output), - }); - } catch (err: any) { - const retryCondition = - resource.retryLaterOnError && - resource.retryLaterOnError.find( - (retry) => retry.name === err.name && retry.message === err.message, - ); - - if (retryCondition) { - console.log(`[Retry]: Creating ${resource.type} ${resource.id}`); - console.log(`[Reason]: ${retryCondition.reason}`); - await new Promise((resolve) => setTimeout(resolve, backoff)); - backoff *= 1.5; - await create(opts, backoff); - } - // else if (err.name === "ConflictException") { - // // todo: provide some means to requisition the resource - // console.log( - // `[Info]: Resource ${resource.type} ${resource.id} already exists but isn't owned by Notation.`, - // ); - // throw err; - // } - else { - console.log(`[Error]: ${err}`); - throw err; - } - } -} diff --git a/packages/core/src/provisioner/operations/operation.delete.ts b/packages/core/src/provisioner/operations/operation.delete.ts deleted file mode 100644 index 7e09040..0000000 --- a/packages/core/src/provisioner/operations/operation.delete.ts +++ /dev/null @@ -1,35 +0,0 @@ -import { operation } from "./operation.base"; -import { BaseResource } from "src/orchestrator/resource"; -import type { StateBackend } from "@notation/state"; - -export const deleteResource = operation("Destroying", delete_); - -async function delete_(opts: { resource: BaseResource; state: StateBackend }) { - const { resource, state } = opts; - const stateNode = await state.get(resource.id); - if (!stateNode) { - throw new Error(`Missing state for ${resource.type} ${resource.id}`); - } - - try { - await resource.delete(resource.key, resource.toState(resource.output)); - } catch (err: any) { - // @todo: declare these in the resource provider - if ( - [ - "NotFoundException", - "ResourceNotFoundException", - "NoSuchEntityException", - ].includes(err.name) || - ["ENOENT"].find((srt) => err.message.includes(srt)) - ) { - console.log( - `Resource ${resource.type} ${resource.id} has already been deleted. \nā†’ Removing from state.\n`, - ); - } else { - throw err; - } - } - - await state.delete(resource.id, stateNode.rev); -} diff --git a/packages/core/src/provisioner/operations/operation.read.ts b/packages/core/src/provisioner/operations/operation.read.ts deleted file mode 100644 index 340beb1..0000000 --- a/packages/core/src/provisioner/operations/operation.read.ts +++ /dev/null @@ -1,66 +0,0 @@ -import { operation } from "./operation.base"; -import { BaseResource } from "src/orchestrator/resource"; -import type { StateBackend } from "@notation/state"; - -export const readResource = operation("Reading", read); - -async function read(opts: { resource: BaseResource; state: StateBackend }) { - const { resource, state } = opts; - - let backoff = 1000; - - if (!resource.read) { - const params = await resource.getParams(); - const stateNode = await state.get(resource.id); - if (!stateNode) return params; - return { ...stateNode.output, ...params }; - } - - async function getSettledReadResult() { - if (!resource.read) return {}; - const readResult = await resource.read(resource.key); - - const needsRetry = resource.retryReadOnCondition?.some((condition) => { - if (!condition) return false; - - const { key, value, reason } = condition; - const msg = `[Info]: ${reason}`; - - if (value && readResult[key] !== value) { - console.log(msg); - return true; - } - - if (!readResult[key]) { - console.log(msg); - return true; - } - - return false; - }); - - if (needsRetry) { - await new Promise((resolve) => setTimeout(resolve, backoff)); - backoff *= 1.2; - return getSettledReadResult(); - } - - return readResult; - } - - try { - const params = await resource.getParams(); - const result = await getSettledReadResult(); - return { ...params, ...result }; - } catch (err: any) { - // todo: normalise not found errors within resource class - // add logic for interpreting error in resource - // if (err.name === "NoSuchEntityException") { - // return null; - // } - console.log( - `[Error]: Reading remote resource ${resource.type} ${resource.id}`, - ); - throw err; - } -} diff --git a/packages/core/src/provisioner/operations/operation.update.ts b/packages/core/src/provisioner/operations/operation.update.ts deleted file mode 100644 index 172f382..0000000 --- a/packages/core/src/provisioner/operations/operation.update.ts +++ /dev/null @@ -1,44 +0,0 @@ -import { operation } from "./operation.base"; -import { BaseResource } from "src/orchestrator/resource"; -import type { StateBackend } from "@notation/state"; -import { readResource } from "."; - -export const updateResource = operation("Updating", update); - -async function update(opts: { - resource: BaseResource; - state: StateBackend; - patch: any; -}): Promise { - const { resource, state, patch } = opts; - const stateNode = await state.get(resource.id); - if (!stateNode) { - throw new Error(`Missing state for ${resource.type} ${resource.id}`); - } - - if (!resource.update) { - throw new Error( - `Update not implemented for ${resource.type} ${resource.id}`, - ); - } - - const params = await resource.getParams(); - - await resource.update( - resource.key, - patch, - params, - resource.toState(resource.output), - ); - resource.setOutput({ ...resource.key, ...params }); - - const result = await readResource({ resource, state, quiet: true }); - resource.setOutput({ ...resource.output, ...result }); - - await state.update(resource.id, stateNode.rev, { - lastOperation: "update", - lastOperationAt: new Date().toISOString(), - params: resource.toState(params), - output: resource.toState(resource.output), - }); -} diff --git a/packages/core/src/provisioner/workflows/index.ts b/packages/core/src/provisioner/workflows/index.ts index 7a29250..e1f075c 100644 --- a/packages/core/src/provisioner/workflows/index.ts +++ b/packages/core/src/provisioner/workflows/index.ts @@ -1,3 +1,8 @@ +export { + createConsoleReconcilerSubscriber, + createNdjsonEventEmitter, + type ReconcilerEventEmitter, +} from "@notation/reconciler"; export * from "./workflow.deploy"; export * from "./workflow.destroy"; export * from "./workflow.plan"; diff --git a/packages/core/src/provisioner/workflows/workflow.deploy.ts b/packages/core/src/provisioner/workflows/workflow.deploy.ts index ead7270..6e005bf 100644 --- a/packages/core/src/provisioner/workflows/workflow.deploy.ts +++ b/packages/core/src/provisioner/workflows/workflow.deploy.ts @@ -1,6 +1,7 @@ import { Reconciler, createConsoleReconcilerSubscriber, + type ReconcilerEventEmitter, type ResourceRegistry, } from "@notation/reconciler"; import type { StateBackend } from "@notation/state"; @@ -13,13 +14,14 @@ export async function deployApp( dryRun = false, registry?: ResourceRegistry, stateBackend?: StateBackend, + emit: ReconcilerEventEmitter = createConsoleReconcilerSubscriber(), ): Promise { const graph = await getResourceGraph(entryPoint); const state = stateBackend ?? createDefaultStateBackend(); const reconciler = new Reconciler({ state, registry, - emit: createConsoleReconcilerSubscriber(), + emit, }); await reconciler.deploy(graph.resources, { diff --git a/packages/core/src/provisioner/workflows/workflow.destroy.ts b/packages/core/src/provisioner/workflows/workflow.destroy.ts index 3e6f61b..071c4fb 100644 --- a/packages/core/src/provisioner/workflows/workflow.destroy.ts +++ b/packages/core/src/provisioner/workflows/workflow.destroy.ts @@ -1,6 +1,7 @@ import { Reconciler, createConsoleReconcilerSubscriber, + type ReconcilerEventEmitter, type ResourceRegistry, } from "@notation/reconciler"; import type { StateBackend } from "@notation/state"; @@ -12,17 +13,18 @@ export async function destroyApp( entryPoint: string, registry?: ResourceRegistry, stateBackend?: StateBackend, + emit: ReconcilerEventEmitter = createConsoleReconcilerSubscriber(), ) { console.log(`Destroying ${entryPoint}\n`); const state = stateBackend ?? createDefaultStateBackend(); - await refreshState(entryPoint, false, registry, state); + await refreshState(entryPoint, false, registry, state, emit); const graph = await getResourceGraph(entryPoint); const reconciler = new Reconciler({ state, - emit: createConsoleReconcilerSubscriber(), + emit, }); await reconciler.destroy(graph.resources); diff --git a/packages/core/src/provisioner/workflows/workflow.plan.ts b/packages/core/src/provisioner/workflows/workflow.plan.ts index 6fde523..13d8e51 100644 --- a/packages/core/src/provisioner/workflows/workflow.plan.ts +++ b/packages/core/src/provisioner/workflows/workflow.plan.ts @@ -2,6 +2,7 @@ import { Reconciler, createConsoleReconcilerSubscriber, type Plan, + type ReconcilerEventEmitter, type ResourceRegistry, } from "@notation/reconciler"; import type { StateBackend } from "@notation/state"; @@ -15,13 +16,14 @@ export async function planApp( driftDetection = true, registry?: ResourceRegistry, stateBackend?: StateBackend, + emit: ReconcilerEventEmitter = createConsoleReconcilerSubscriber(), ): Promise { const graph = await getResourceGraph(entryPoint); const state = stateBackend ?? createDefaultStateBackend(); const reconciler = new Reconciler({ state, registry, - emit: createConsoleReconcilerSubscriber(), + emit, }); return reconciler.plan(graph.resources, { driftDetection }); diff --git a/packages/core/src/provisioner/workflows/workflow.refresh.ts b/packages/core/src/provisioner/workflows/workflow.refresh.ts index b70eb86..d7d5114 100644 --- a/packages/core/src/provisioner/workflows/workflow.refresh.ts +++ b/packages/core/src/provisioner/workflows/workflow.refresh.ts @@ -1,6 +1,7 @@ import { Reconciler, createConsoleReconcilerSubscriber, + type ReconcilerEventEmitter, type ResourceRegistry, } from "@notation/reconciler"; import type { StateBackend } from "@notation/state"; @@ -15,6 +16,7 @@ export async function refreshState( dryRun = false, registry?: ResourceRegistry, stateBackend?: StateBackend, + emit: ReconcilerEventEmitter = createConsoleReconcilerSubscriber(), ): Promise { console.log(`${dryRun ? "[Dry Run]: " : ""}Refreshing ${entryPoint} state\n`); @@ -24,7 +26,7 @@ export async function refreshState( const reconciler = new Reconciler({ state, registry, - emit: createConsoleReconcilerSubscriber(), + emit, }); await reconciler.refresh(graph.resources, { dryRun }); diff --git a/packages/core/test/provisioner/resource-registry.test.ts b/packages/core/test/provisioner/resource-registry.test.ts index c5a7c6d..e01102c 100644 --- a/packages/core/test/provisioner/resource-registry.test.ts +++ b/packages/core/test/provisioner/resource-registry.test.ts @@ -17,7 +17,9 @@ describe("provisioner resource registry", () => { it("returns undefined when a resource type is not registered", () => { const registry = createResourceRegistry([TestResource]); - expect(resolveResourceClass(registry, "test/service/unknown")).toBeUndefined(); + expect( + resolveResourceClass(registry, "test/service/unknown"), + ).toBeUndefined(); }); it("creates a structured warning event for orphan skips", () => { diff --git a/packages/dashboard/package.json b/packages/dashboard/package.json index aa43f1c..53116b3 100644 --- a/packages/dashboard/package.json +++ b/packages/dashboard/package.json @@ -14,7 +14,7 @@ ], "dependencies": { "@fastify/static": "^9.1.3", - "chokidar": "^4.0.3", + "@notation/state": "workspace:*", "fastify": "^5.10.0", "react": "^19.0.0", "react-dom": "^19.0.0" diff --git a/packages/dashboard/server/server.test.ts b/packages/dashboard/server/server.test.ts new file mode 100644 index 0000000..e9cf2b3 --- /dev/null +++ b/packages/dashboard/server/server.test.ts @@ -0,0 +1,30 @@ +import { MemoryStateBackend } from "@notation/state"; +import { describe, expect, it } from "vitest"; +import { readStateSnapshot } from "./server"; + +describe("dashboard state", () => { + it("reads state through the backend contract", async () => { + const state = new MemoryStateBackend(); + await state.update( + "service", + 0, + { + id: "service", + type: "test/service/main", + config: {}, + params: {}, + output: { ready: true }, + lastOperation: "create", + lastOperationAt: "2026-07-18T00:00:00.000Z", + }, + ); + + await expect(readStateSnapshot(state)).resolves.toMatchObject({ + service: { + id: "service", + rev: 1, + output: { ready: true }, + }, + }); + }); +}); diff --git a/packages/dashboard/server/server.ts b/packages/dashboard/server/server.ts index 6629759..bcc5cf7 100644 --- a/packages/dashboard/server/server.ts +++ b/packages/dashboard/server/server.ts @@ -1,58 +1,82 @@ -import Fastify from "fastify"; import fastifyStatic from "@fastify/static"; -import chokidar from "chokidar"; -import fs from "fs/promises"; -import path from "path"; -import { fileURLToPath } from "url"; -import { dirname } from "path"; +import type { StateBackend, StateNode } from "@notation/state"; +import Fastify from "fastify"; +import { dirname, join } from "node:path"; +import { fileURLToPath } from "node:url"; -const __filename = fileURLToPath(import.meta.url); -const __dirname = dirname(__filename); +const serverDirectory = dirname(fileURLToPath(import.meta.url)); -const fastify = Fastify({}); +export type DashboardServerOptions = { + state: StateBackend; + pollInterval?: number; +}; -const defaultStatePath = path.join(process.cwd(), ".notation", "state.json"); -const statePath = process.env.NOTATION_STATE_PATH ?? defaultStatePath; +export type StartDashboardServerOptions = DashboardServerOptions & { + port?: number; +}; -fastify.register(fastifyStatic, { - root: path.join(__dirname, "./"), - prefix: "/", -}); +export async function readStateSnapshot( + state: StateBackend, +): Promise> { + const nodes = await state.values(); + return Object.fromEntries(nodes.map((node) => [node.id, node])); +} -fastify.get("/state", (request, reply) => { - reply.raw.setHeader("Content-Type", "text/event-stream"); - reply.raw.setHeader("Cache-Control", "no-cache"); - reply.raw.setHeader("Connection", "keep-alive"); +export function createDashboardServer({ + state, + pollInterval = 500, +}: DashboardServerOptions) { + const server = Fastify({}); - const sendState = async () => { - try { - const state = await fs.readFile(statePath, "utf8"); - reply.raw.write(`data: ${JSON.stringify(JSON.parse(state))}\n\n`); - } catch (error) { - console.error("Error reading state.json:", error); - } - }; + server.register(fastifyStatic, { + root: join(serverDirectory, "./"), + prefix: "/", + }); - const watcher = chokidar.watch(statePath); + server.get("/state", (request, reply) => { + reply.raw.setHeader("Content-Type", "text/event-stream"); + reply.raw.setHeader("Cache-Control", "no-cache"); + reply.raw.setHeader("Connection", "keep-alive"); - watcher.on("change", sendState); + let lastSnapshot: string | undefined; + let reading = false; + let closed = false; + const sendState = async () => { + if (reading || closed) return; + reading = true; + try { + const snapshot = JSON.stringify(await readStateSnapshot(state)); + if (closed || snapshot === lastSnapshot) return; + lastSnapshot = snapshot; + reply.raw.write(`data: ${snapshot}\n\n`); + } catch (error) { + request.log.error(error, "Unable to read state"); + } finally { + reading = false; + } + }; - sendState(); + const timer = setInterval(sendState, pollInterval); + timer.unref(); + void sendState(); - request.raw.on("close", () => { - watcher.close(); + request.raw.on("close", () => { + closed = true; + clearInterval(timer); + }); + reply.hijack(); }); - reply.hijack(); -}); - -export const startDashboardServer = async (port: number = 6682) => { - try { - await fastify.listen({ port }); - console.log("\nNotation dashboard is running on:\n\n"); - console.log(`āžœ http://localhost:${port}`); - } catch (err) { - console.error(err); - process.exit(1); - } -}; + return server; +} + +export async function startDashboardServer({ + port = 6682, + ...options +}: StartDashboardServerOptions) { + const server = createDashboardServer(options); + await server.listen({ port }); + console.log("\nNotation dashboard is running on:\n\n"); + console.log(`āžœ http://localhost:${port}`); + return server; +} diff --git a/packages/reconciler/src/index.ts b/packages/reconciler/src/index.ts index 712f362..db78141 100644 --- a/packages/reconciler/src/index.ts +++ b/packages/reconciler/src/index.ts @@ -9,3 +9,4 @@ export * from "./dependency-graph"; export * from "./plan"; export * from "./reconciler"; export * from "./console-subscriber"; +export * from "./protocol"; diff --git a/packages/reconciler/src/plan.ts b/packages/reconciler/src/plan.ts index 7f0aff0..87fb621 100644 --- a/packages/reconciler/src/plan.ts +++ b/packages/reconciler/src/plan.ts @@ -133,7 +133,7 @@ export async function resolvePlanParams( const config = resource.config as Record; for (const [key, item] of Object.entries(resource.schema)) { - if (item.propertyType !== "param") continue; + if (item.propertyType === "computed") continue; params[key] = key in config && config[key] !== undefined ? config[key] diff --git a/packages/reconciler/src/protocol.ts b/packages/reconciler/src/protocol.ts new file mode 100644 index 0000000..bf3706e --- /dev/null +++ b/packages/reconciler/src/protocol.ts @@ -0,0 +1,17 @@ +import type { ReconcilerEvent, ReconcilerEventEmitter } from "./reconciler"; + +export const EVENT_STREAM_VERSION = 1 as const; + +export type WireReconcilerEvent = ReconcilerEvent & { + version: typeof EVENT_STREAM_VERSION; +}; + +export function encodeReconcilerEvent(event: ReconcilerEvent): string { + return `${JSON.stringify({ version: EVENT_STREAM_VERSION, ...event })}\n`; +} + +export function createNdjsonEventEmitter( + write: (line: string) => void | Promise, +): ReconcilerEventEmitter { + return (event) => write(encodeReconcilerEvent(event)); +} diff --git a/packages/reconciler/test/protocol.test.ts b/packages/reconciler/test/protocol.test.ts new file mode 100644 index 0000000..00ac6cf --- /dev/null +++ b/packages/reconciler/test/protocol.test.ts @@ -0,0 +1,25 @@ +import { describe, expect, it } from "vitest"; +import { createNdjsonEventEmitter } from "../src"; + +describe("event stream protocol", () => { + it("writes one versioned JSON document per event", async () => { + const lines: string[] = []; + const emit = createNdjsonEventEmitter((line) => { + lines.push(line); + }); + await emit({ + level: "info", + event: "reconciler.deploy.decision", + resourceId: "service", + resourceType: "test/service/main", + decision: "create", + }); + + expect(lines).toHaveLength(1); + expect(lines[0]?.endsWith("\n")).toBe(true); + expect(JSON.parse(lines[0]!)).toMatchObject({ + version: 1, + decision: "create", + }); + }); +}); diff --git a/packages/reconciler/test/reconciler.plan.test.ts b/packages/reconciler/test/reconciler.plan.test.ts index 1d5ce49..88b8c91 100644 --- a/packages/reconciler/test/reconciler.plan.test.ts +++ b/packages/reconciler/test/reconciler.plan.test.ts @@ -21,12 +21,22 @@ function createMemoryState(initial: Record = {}) { delete store[id]; }), values: vi.fn(async () => Object.values(store)), + lease: vi.fn(async (scope: string, ttl: number) => ({ + scope, + expiresAt: new Date(Date.now() + ttl).toISOString(), + renew: vi.fn(async (nextTtl: number) => + new Date(Date.now() + nextTtl).toISOString(), + ), + release: vi.fn(async () => undefined), + })), }; } function createTestResourceClass(opts: { type: `${string}/${string}/${string}`; - create?: (params: Record) => Promise | void>; + create?: ( + params: Record, + ) => Promise | void>; read?: (key: Record) => Promise>; update?: ( key: Record, @@ -34,7 +44,10 @@ function createTestResourceClass(opts: { params: Record, state: Record, ) => Promise; - delete?: (key: Record, state: Record) => Promise; + delete?: ( + key: Record, + state: Record, + ) => Promise; notFoundOnError?: { name: string; reason: string }[]; }) { return resource({ type: opts.type }) @@ -195,7 +208,9 @@ describe("reconciler plan", () => { err.name = "NotFoundException"; throw err; }, - notFoundOnError: [{ name: "NotFoundException", reason: "deleted remotely" }], + notFoundOnError: [ + { name: "NotFoundException", reason: "deleted remotely" }, + ], }); const state = createMemoryState({ @@ -260,8 +275,12 @@ describe("reconciler plan", () => { }); it("populates dependsOn from resource dependencies", async () => { - const AResource = createTestResourceClass({ type: "test/service/plan-dep-a" }); - const BResource = createTestResourceClass({ type: "test/service/plan-dep-b" }); + const AResource = createTestResourceClass({ + type: "test/service/plan-dep-a", + }); + const BResource = createTestResourceClass({ + type: "test/service/plan-dep-b", + }); const resourceA = new AResource({ id: "a", config: { name: "a" } }); const resourceB = new BResource({ @@ -280,8 +299,12 @@ describe("reconciler plan", () => { }); it("marks params derived from uncreated dependencies as unknown after apply", async () => { - const AResource = createTestResourceClass({ type: "test/service/plan-unknown-a" }); - const BResource = createTestResourceClass({ type: "test/service/plan-unknown-b" }) + const AResource = createTestResourceClass({ + type: "test/service/plan-unknown-a", + }); + const BResource = createTestResourceClass({ + type: "test/service/plan-unknown-b", + }) .requireDependencies<{ a: BaseResource }>() .deriveParams(({ deps }) => ({ name: (deps.a.output as { name: string }).name, diff --git a/packages/resource/src/resource.schema.ts b/packages/resource/src/resource.schema.ts index c245314..2810b9a 100644 --- a/packages/resource/src/resource.schema.ts +++ b/packages/resource/src/resource.schema.ts @@ -164,16 +164,16 @@ export type SchemaFromApi< [K in keyof ApiCompoundKey]: SchemaItem & ({ primaryKey: true } | { secondaryKey: true }); } & { - [K in keyof OmitOptional< - Omit - >]: SchemaItem & { + [ + K in keyof OmitOptional> + ]: SchemaItem & { propertyType: "param"; presence: "required"; }; } & { - [K in keyof PickOptional< - Omit - >]: SchemaItem & { + [ + K in keyof PickOptional> + ]: SchemaItem & { propertyType: "param"; presence: "optional"; }; @@ -185,10 +185,9 @@ export type SchemaFromApi< immutable: true; }; } & { - [K in keyof Omit< - ApiReadResult, - keyof ApiCreateParams | keyof ApiCompoundKey - >]: SchemaItem & { + [ + K in keyof Omit + ]: SchemaItem & { propertyType: "computed"; }; }; @@ -205,18 +204,22 @@ export type MapSchema< IncludeKey = any, > = Simplify< { - [K in keyof S as S[K] extends { presence: "optional" } | ExcludeConditions - ? never - : IncludeKey extends keyof S[K] - ? K - : never]: S[K]["valueType"]["_output"]; - } & { - [K in keyof S as S[K] extends ExcludeConditions - ? never - : S[K] extends { presence: "optional" } - ? IncludeKey extends keyof S[K] + [ + K in keyof S as S[K] extends { presence: "optional" } | ExcludeConditions + ? never + : IncludeKey extends keyof S[K] ? K : never - : never]?: S[K]["valueType"]["_output"]; + ]: S[K]["valueType"]["_output"]; + } & { + [ + K in keyof S as S[K] extends ExcludeConditions + ? never + : S[K] extends { presence: "optional" } + ? IncludeKey extends keyof S[K] + ? K + : never + : never + ]?: S[K]["valueType"]["_output"]; } >; diff --git a/packages/resource/src/resource.ts b/packages/resource/src/resource.ts index 94a8c48..b2ed61d 100644 --- a/packages/resource/src/resource.ts +++ b/packages/resource/src/resource.ts @@ -105,8 +105,7 @@ export abstract class Resource< T extends ResourceTypes = ResourceTypes, D extends Record = {}, C extends Record = T["params"], -> implements BaseResource -{ +> implements BaseResource { config: C; id: string; groupId = -1; @@ -263,11 +262,11 @@ export type ResourceBuilder = { defineSchema: < S extends Schema & SchemaFromApi< - ApiSchema["Key"], - ApiSchema["CreateParams"], - Fallback, - Fallback - >, + ApiSchema["Key"], + ApiSchema["CreateParams"], + Fallback, + Fallback + >, >( schema: S, ) => ResourceSchemaBuilder>>; diff --git a/packages/resource/src/types.ts b/packages/resource/src/types.ts index b1e3036..e1c55ac 100644 --- a/packages/resource/src/types.ts +++ b/packages/resource/src/types.ts @@ -1,11 +1,10 @@ -export type IfAllPropertiesOptional = T extends Partial - ? Partial extends T - ? Y - : N - : N; - -export type OptionalIfAllPropertiesOptional = - IfAllPropertiesOptional; +export type IfAllPropertiesOptional = + T extends Partial ? (Partial extends T ? Y : N) : N; + +export type OptionalIfAllPropertiesOptional< + K extends string, + T, +> = IfAllPropertiesOptional; export type Fallback = T extends undefined ? U : T; diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 8a98b39..b8ee0b3 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -255,9 +255,9 @@ importers: '@fastify/static': specifier: ^9.1.3 version: 9.1.3 - chokidar: - specifier: ^4.0.3 - version: 4.0.3 + '@notation/state': + specifier: workspace:* + version: link:../state fastify: specifier: ^5.10.0 version: 5.10.0