feat: add commands for managing deployments and resources

- Implemented `down` command to delete managed resources, with an option to delete ingress and PVCs.
- Added `exec` command to execute commands inside deployment pods.
- Created `logs` command to show deployment logs with an option to follow the output.
- Introduced `start` command to start deployments without rebuilding images.
- Developed `stop` command to scale managed deployments to zero.

refactor: enhance resource management and Kubernetes interactions

- Refactored resource handling in `lib/apply.ts` to support sorting and applying resources.
- Improved logging functionality with a new logger in `lib/logger.ts`.
- Enhanced build process in `lib/build.ts` to support remote builds and syncing with Git repositories.
- Added functions to manage deployments and pods in `lib/shared.ts`.

chore: update dependencies and TypeScript configuration

- Updated package.json to include `listr2` for task management.
- Modified tsconfig.json to exclude build artifacts and fixtures.

command/down.ts

@@ -0,0 +1,59 @@
+import { defineCommand } from "citty";
+import { Listr } from "listr2";
+import { ctx } from "../lib/context";
+import {
+  deleteResource,
+  listManagedResources,
+  sortResources,
+} from "../lib/apply";
+
+export const down = defineCommand({
+  meta: {
+    name: "down",
+    description: "Delete managed resources except ingress and PVCs",
+  },
+  args: {
+    full: {
+      type: "boolean",
+      alias: "f",
+      description: "Also delete ingress, PVCs, and the namespace",
+    },
+  },
+  async run({ args }) {
+    const { project } = ctx();
+    const resources = sortResources(await listManagedResources(project)).filter(
+      (resource) =>
+        args.full ||
+        (resource.kind !== "Ingress" &&
+          resource.kind !== "PersistentVolumeClaim"),
+    );
+
+    if (args.full) {
+      resources.push({
+        apiVersion: "v1",
+        kind: "Namespace",
+        metadata: { name: project },
+      });
+    }
+
+    await new Listr([
+      {
+        title: "Delete resources",
+        skip: () =>
+          resources.length > 0 ? false : "No managed resources to delete",
+        task: (_ctx, task) => {
+          const ordered = sortResources(resources).reverse();
+
+          task.output = `${ordered.length} resources queued`;
+          return task.newListr(
+            ordered.map((resource) => ({
+              title: `${resource.kind} ${resource.metadata?.name}`,
+              task: () => deleteResource(resource),
+            })),
+            { concurrent: false, exitOnError: true },
+          );
+        },
+      },
+    ]).run();
+  },
+});

command/exec.ts

@@ -0,0 +1,43 @@
+import { defineCommand } from "citty";
+import { getPodContainerName, getPodForDeployment } from "../lib/shared";
+import { execClient } from "../lib/k8s";
+
+export const exec = defineCommand({
+  meta: {
+    name: "exec",
+    description: "Execute a command inside a deployment pod",
+  },
+  async run({ args }) {
+    const [deployment, ...command] = args._;
+    if (!deployment) throw new Error("Deployment name is required");
+    if (command.length === 0) throw new Error("Command is required");
+
+    const pod = await getPodForDeployment(deployment);
+    if (pod.status?.phase !== "Running") {
+      throw new Error(
+        `Deployment ${deployment} has no running pod to exec into (${pod.metadata?.name ?? "unknown pod"} is ${pod.status?.phase ?? "not ready"})`,
+      );
+    }
+
+    const namespace = pod.metadata?.namespace ?? "default";
+    const podName = pod.metadata?.name;
+    if (!podName) throw new Error(`No pod name found for deployment ${deployment}`);
+
+    const socket = await execClient.exec(
+      namespace,
+      podName,
+      getPodContainerName(pod),
+      command,
+      process.stdout,
+      process.stderr,
+      process.stdin,
+      Boolean(process.stdin.isTTY),
+    );
+
+    await new Promise<void>((resolve, reject) => {
+      socket.onclose = () => resolve();
+      socket.onerror = (event: { error?: unknown }) =>
+        reject(event.error ?? new Error("Exec failed"));
+    });
+  },
+});

command/logs.ts

@@ -0,0 +1,142 @@
+import { defineCommand } from "citty";
+import { finished } from "node:stream/promises";
+import { Writable } from "node:stream";
+import { createLogger } from "../lib/logger";
+import { core, logClient } from "../lib/k8s";
+import {
+  getPodContainerName,
+  listManagedDeployments,
+  listPodsForDeployment,
+} from "../lib/shared";
+
+class LoggerStream extends Writable {
+  #buffer = "";
+
+  constructor(private readonly writeLine: (line: string) => void) {
+    super();
+  }
+
+  override _write(chunk: Buffer | string, _encoding: BufferEncoding, callback: (error?: Error | null) => void) {
+    this.#buffer += chunk.toString();
+
+    let newline = this.#buffer.indexOf("\n");
+    while (newline !== -1) {
+      const line = this.#buffer.slice(0, newline).replace(/\r$/, "");
+      if (line) this.writeLine(line);
+      this.#buffer = this.#buffer.slice(newline + 1);
+      newline = this.#buffer.indexOf("\n");
+    }
+
+    callback();
+  }
+
+  override _final(callback: (error?: Error | null) => void) {
+    const line = this.#buffer.replace(/\r$/, "");
+    if (line) this.writeLine(line);
+    this.#buffer = "";
+    callback();
+  }
+}
+
+function formatError(error: unknown): string {
+  if (error instanceof Error) return error.message;
+  return String(error);
+}
+
+async function logDeployment(name: string, follow: boolean) {
+  const pods = await listPodsForDeployment(name);
+  if (pods.length === 0) throw new Error(`No pods found for deployment ${name}`);
+
+  const logger = createLogger(name);
+
+  if (!follow) {
+    for (const pod of pods) {
+      const podName = pod.metadata?.name;
+      if (!podName) continue;
+
+      try {
+        const text = await core.readNamespacedPodLog({
+          namespace: pod.metadata?.namespace ?? "default",
+          name: podName,
+          container: getPodContainerName(pod),
+        });
+
+        for (const line of text.split(/\r?\n/)) {
+          if (line) logger`${line}`;
+        }
+      } catch (error) {
+        logger.warn`${podName}: ${formatError(error)}`;
+      }
+    }
+    return;
+  }
+
+  const controllers = await Promise.all(
+    pods.map(async (pod) => {
+      const podName = pod.metadata?.name;
+      if (!podName) return;
+
+      const stream = new LoggerStream((line) => logger`${line}`);
+      try {
+        const controller = await logClient.log(
+          pod.metadata?.namespace ?? "default",
+          podName,
+          getPodContainerName(pod),
+          stream,
+          { follow: true },
+        );
+
+        return { controller, stream };
+      } catch (error) {
+        logger.warn`${podName}: ${formatError(error)}`;
+        stream.end();
+      }
+    }),
+  );
+
+  const activeControllers = controllers.filter(
+    (item): item is NonNullable<(typeof controllers)[number]> => Boolean(item),
+  );
+  if (activeControllers.length === 0) return;
+
+  await new Promise<void>((resolve) => {
+    const abort = () => {
+      for (const item of activeControllers) item.controller.abort();
+      resolve();
+    };
+
+    process.once("SIGINT", abort);
+    process.once("SIGTERM", abort);
+  });
+
+  await Promise.all(activeControllers.map((item) => finished(item.stream)));
+}
+
+export const logs = defineCommand({
+  meta: {
+    name: "logs",
+    description: "Show deployment logs",
+  },
+  args: {
+    follow: {
+      type: "boolean",
+      alias: "f",
+      description: "Follow log output",
+    },
+  },
+  async run({ args }) {
+    const [deployment] = args._;
+    const names = deployment
+      ? [deployment]
+      : (await listManagedDeployments())
+          .map((item) => item.metadata?.name)
+          .filter((item): item is string => Boolean(item));
+
+    if (args.follow) {
+      await Promise.all(names.map((name) => logDeployment(name, true)));
+      return;
+    }
+
+    for (const name of names) await logDeployment(name, false);
+  },
+});

command/start.ts

@@ -0,0 +1,12 @@
+import { defineCommand } from "citty";
+import { runUp } from "./up";
+
+export const start = defineCommand({
+  meta: {
+    name: "start",
+    description: "Start deployments without rebuilding images",
+  },
+  async run() {
+    await runUp(false);
+  },
+});

command/stop.ts

@@ -0,0 +1,31 @@
+import { defineCommand } from "citty";
+import { Listr } from "listr2";
+import { listManagedDeployments, scaleDeployment } from "../lib/shared";
+
+export const stop = defineCommand({
+  meta: {
+    name: "stop",
+    description: "Scale managed deployments to zero",
+  },
+  async run() {
+    const deployments = await listManagedDeployments();
+
+    await new Listr([
+      {
+        title: "Scale deployments",
+        skip: () =>
+          deployments.length > 0 ? false : "No managed deployments found",
+        task: (_ctx, task) => {
+          task.output = `${deployments.length} deployments queued`;
+          return task.newListr(
+            deployments.map((deployment) => ({
+              title: `Deployment ${deployment.metadata?.name}`,
+              task: () => scaleDeployment(deployment.metadata!.name!, 0),
+            })),
+            { concurrent: false, exitOnError: true },
+          );
+        },
+      },
+    ]).run();
+  },
+});

command/up.ts

@@ -1,6 +1,100 @@
 import { defineCommand } from "citty";
+import { Listr } from "listr2";
+import type { ComposeSpecification } from "../schema/docker.d";
 import { ctx } from "../lib/context";
-import { serviceToDeployment } from "../lib/convert";
+import {
+  applyResource,
+  deleteResource,
+  getStaleResources,
+  sortResources,
+} from "../lib/apply";
+import { buildServices } from "../lib/build";
+import { composeToKubernetes, type KubernetesResource } from "../lib/convert";
+
+type UpContext = {
+  compose?: ComposeSpecification;
+  resources?: KubernetesResource[];
+  staleResources?: KubernetesResource[];
+};
+
+export async function runUp(build: boolean) {
+  const { project, compose, cwd } = ctx();
+
+  await new Listr<UpContext>([
+    {
+      title: "Read compose",
+      task: async (taskCtx, task) => {
+        taskCtx.compose = await compose();
+        task.output = `${Object.keys(taskCtx.compose.services ?? {}).length} services`;
+      },
+    },
+    {
+      title: "Build images",
+      enabled: () => build,
+      skip: (taskCtx) =>
+        Object.values(taskCtx.compose?.services ?? {}).some((service) => service.build)
+          ? false
+          : "No buildable services",
+      task: async (taskCtx, task) => {
+        const built = await buildServices(project, taskCtx.compose!, cwd, (message) => {
+          task.output = message;
+        });
+        task.output = `Built ${built} image${built === 1 ? "" : "s"}`;
+      },
+    },
+    {
+      title: "Render manifests",
+      task: async (taskCtx, task) => {
+        taskCtx.resources = await composeToKubernetes(project, taskCtx.compose!, cwd);
+        task.output = `${taskCtx.resources.length} resources`;
+      },
+    },
+    {
+      title: "Plan reconciliation",
+      task: async (taskCtx, task) => {
+        taskCtx.staleResources = (await getStaleResources(
+          project,
+          taskCtx.resources!,
+        )) as KubernetesResource[];
+        task.output = `${taskCtx.staleResources.length} stale resource${taskCtx.staleResources.length === 1 ? "" : "s"}`;
+      },
+    },
+    {
+      title: "Apply resources",
+      task: (taskCtx, task) => {
+        const resources = sortResources(taskCtx.resources!);
+
+        task.output = `${resources.length} resources queued`;
+        return task.newListr(
+          resources.map((resource) => ({
+            title: `${resource.kind} ${resource.metadata?.name}`,
+            task: () => applyResource(resource),
+          })),
+          { concurrent: false, exitOnError: true },
+        );
+      },
+    },
+    {
+      title: "Delete stale resources",
+      skip: (taskCtx) =>
+        taskCtx.staleResources && taskCtx.staleResources.length > 0
+          ? false
+          : "No stale resources",
+      task: (taskCtx, task) => {
+        const resources = sortResources(taskCtx.staleResources!).reverse();
+
+        task.output = `${resources.length} resources queued`;
+        return task.newListr(
+          resources.map((resource) => ({
+            title: `${resource.kind} ${resource.metadata?.name}`,
+            task: () => deleteResource(resource),
+          })),
+          { concurrent: false, exitOnError: true },
+        );
+      },
+    },
+  ], { rendererOptions: { collapseErrors: false } }).run();
+}
 
 export const up = defineCommand({
   meta: {
@@ -16,11 +110,7 @@ export const up = defineCommand({
       negativeDescription: "Don't build an image, even if it's policy",
     },
   },
-  async run() {
-    const { project, compose } = ctx();
-    const cmp = await compose();
-    console.dir(serviceToDeployment(project, "app", cmp.services!.app!), {
-      depth: 999,
-    });
+  async run({ args }) {
+    await runUp(args.build);
   },
 });