feat: Start adding CLI stuff

packages/core/src/actor.ts

@@ -60,7 +60,7 @@ export namespace Actor {
 
   export type Info = z.Infer<typeof Info>;
 
-  export const ctx = Context.create<Info>();
+  export const ctx = Context.create<Info>("Actor");
 
   export function userID() {
     const actor = ctx.use();

packages/core/src/bus/index.ts

@@ -0,0 +1,182 @@
+import { RedisClient } from "bun";
+import { event } from "./event.js";
+import { Actor } from "../actor.js";
+import { Log } from "../utils/log.js";
+import { Flag } from "../flag/index.js";
+import { ZodValidator } from "./validator.js";
+
+export namespace Bus {
+  const log = Log.create({ service: "bus" });
+
+  let pubClient: RedisClient | null = null;
+  let subClient: RedisClient | null = null;
+
+  // keep the listener we registered with Bun so we don't re-register for the same channel
+  const channelListeners = new Map<
+    string,
+    (message: string, channel: string) => void
+  >();
+
+  // local in-process registry of handlers per event type
+  const subscribers = new Map<string, ((payload: any, raw: any) => void)[]>();
+
+  async function connect() {
+    try {
+      if (pubClient && subClient) return { pubClient, subClient };
+
+      const url = Flag.NESTRI_REDIS_URL || "redis://127.0.0.1:6379";
+
+      pubClient = new RedisClient(url);
+      subClient = await pubClient.duplicate();
+
+      //connect to the instance, but we await readiness for clarity
+      await Promise.all([
+        pubClient.connect().catch(() => Promise.resolve()),
+        subClient.connect().catch(() => Promise.resolve()),
+      ]);
+
+      // global dispatcher for incoming messages
+      subClient.subscribe("message", (channel: string, message: string) => {
+        const handlers = subscribers.get(channel);
+        if (!handlers?.length) return;
+        try {
+          const parsed = JSON.parse(message);
+          for (const h of handlers) {
+            // call handlers without awaiting so one slow handler doesn't block the dispatcher
+            try {
+              const r = h(parsed.detail, parsed) as any;
+              if (r instanceof Promise)
+                r.catch((e) => log.error("bus handler error", e));
+            } catch (e: any) {
+              log.error("bus handler error", e);
+            }
+          }
+        } catch (e: any) {
+          log.error("bus: failed to parse message", e);
+        }
+      });
+
+      return { pubClient, subClient };
+    } catch (err: any) {
+      log.error("bus: failed to connect to redis", err);
+      throw err;
+    }
+  }
+
+  export function subscribe<Events extends event.Definition>(
+    _events: Events | Events[],
+    cb: (
+      input: {
+        [K in Events["type"]]: Extract<Events, { type: K }>["$payload"];
+      }[Events["type"]],
+      raw: { type: string; detail: any },
+    ) => Promise<void> | void,
+  ) {
+    const events = Array.isArray(_events) ? _events : [_events];
+
+    (async () => {
+      const { subClient } = await connect();
+      for (const e of events) {
+        const channel = e.type;
+        // ensure subscribers map entry exists
+        if (!subscribers.has(channel)) subscribers.set(channel, []);
+
+        // push the handler
+        subscribers.get(channel)!.push(async (payload, raw) => {
+          await cb(payload, raw);
+        });
+
+        // only register a single Bun listener per channel (first time)
+        if (!channelListeners.has(channel)) {
+          const listener = (message: string, ch: string) => {
+            // message is string, ch is actual channel; some Bun versions pass (message, channel) or (message)
+            const usedChannel = ch ?? channel;
+            const handlers = subscribers.get(usedChannel);
+            if (!handlers?.length) return;
+
+            let parsed: any;
+            try {
+              parsed = JSON.parse(message);
+            } catch (parseErr) {
+              // If the publisher didn't send JSON, treat raw message as detail
+              parsed = { detail: message, type: usedChannel };
+            }
+
+            for (const h of handlers) {
+              try {
+                const r = h(parsed.detail ?? parsed, parsed) as any;
+                if (r instanceof Promise)
+                  r.catch((e) => log.error("bus handler error", e));
+              } catch (e: any) {
+                log.error("bus handler error", e);
+              }
+            }
+          };
+
+          // register listener with Bun's subscribe API
+          try {
+            // Bun's subscribe accepts (channel, listener)
+            await subClient.subscribe(channel, listener);
+            channelListeners.set(channel, listener);
+          } catch (err: any) {
+            log.error("bus: failed to subscribe to", { channel, err });
+            throw err;
+          }
+        }
+      }
+    })();
+  }
+
+  export async function publish<Definition extends event.Definition = any>(
+    name: string | { name: string },
+    def: Definition | string,
+    properties: Definition["$input"],
+    options?: {
+      metadata?: Definition["$metadata"];
+    },
+  ) {
+    const { pubClient } = await connect();
+
+    const evt =
+      typeof def === "string"
+        ? {
+            type: def,
+            properties,
+            metadata: options?.metadata || {},
+          }
+        : await def.create(properties, options?.metadata);
+
+    const payload = {
+      type: evt.type,
+      metadata: evt.metadata,
+      detail: evt.properties,
+      bus: typeof name === "string" ? name : name.name,
+    };
+
+    try {
+      // Bun's publish returns a Promise<number> (number of clients that received it)
+      await pubClient.publish(evt.type, JSON.stringify(payload));
+      return { ok: true, event: evt.type, detail: payload.detail };
+    } catch (e) {
+      throw new PublishError(evt.type, e);
+    }
+  }
+
+  export class PublishError extends Error {
+    constructor(
+      public readonly eventType: string,
+      public readonly cause?: any,
+    ) {
+      super(`Failed to publish event: ${eventType}`);
+    }
+  }
+
+  export const createEvent = event.builder({
+    validator: ZodValidator,
+    metadata() {
+      return {
+        actor: Actor.use(),
+      };
+    },
+  });
+}

packages/core/src/context.ts

@@ -1,15 +1,19 @@
 import { AsyncLocalStorage } from "node:async_hooks";
 
 export namespace Context {
-  export class NotFound extends Error {}
+  export class NotFound extends Error {
+    constructor(public override readonly name: string) {
+      super(`No context found for ${name}`);
+    }
+  }
 
-  export function create<T>() {
+  export function create<T>(name: string) {
     const storage = new AsyncLocalStorage<T>();
     return {
       use() {
         const result = storage.getStore();
         if (!result) {
-          throw new Error("No context available");
+          throw new NotFound(name);
         }
         return result;
       },

packages/core/src/drizzle/index.ts

@@ -1,18 +1,16 @@
 import postgres from "postgres";
 import { Context } from "../context.js";
+import { Flag } from "../flag/index.js";
 import { ExtractTablesWithRelations } from "drizzle-orm";
 import { PgTransaction, PgTransactionConfig } from "drizzle-orm/pg-core";
 import { PostgresJsQueryResultHKT, drizzle } from "drizzle-orm/postgres-js";
 
 export namespace Database {
-  const { DATABASE_URL } = process.env;
-  // if (!DATABASE_URL) {
-  //   throw new Error("DATABASE_URL is not set");
-  // }
-  const dbUrl =
-    DATABASE_URL || "postgres://user:password@localhost:5432/nestri";
+  const url =
+    Flag.NESTRI_POSTGRES_URL ||
+    "postgres://user:password@localhost:5432/nestri";
 
-  const sql = postgres(dbUrl, {
+  const sql = postgres(url, {
     max: 20,
     idle_timeout: 60,
   });
@@ -30,7 +28,7 @@ export namespace Database {
   const TransactionContext = Context.create<{
     tx: TxOrDb;
     effects: (() => void | Promise<void>)[];
-  }>();
+  }>("DatabaseTransactionContext");
 
   export async function use<T>(callback: (trx: TxOrDb) => Promise<T>) {
     try {

packages/core/src/event/index.ts

@@ -1,12 +0,0 @@
-import { event } from "./event.js";
-import { Actor } from "../actor.js";
-import { ZodValidator } from "./validator.js";
-
-export const createEvent = event.builder({
-  validator: ZodValidator,
-  metadata() {
-    return {
-      actor: Actor.use(),
-    };
-  },
-});

packages/core/src/flag/index.ts

@@ -0,0 +1,12 @@
+export namespace Flag {
+  export const NESTRI_FAKE_VCS = process.env["NESTRI_FAKE_VCS"];
+  export const NESTRI_REDIS_URL = process.env["NESTRI_REDIS_URL"];
+  export const NESTRI_POSTGRES_URL = process.env["NESTRI_POSTGRES_URL"];
+  export const NESTRI_SKIP_DB_MIGRATIONS = truthy("NESTRI_SKIP_DB_MIGRATIONS");
+  export const NESTRI_DISABLE_ANALYTICS = truthy("NESTRI_DISABLE_ANALYTICS");
+
+  function truthy(key: string) {
+    const value = process.env[key]?.toLowerCase();
+    return value === "true" || value === "1";
+  }
+}

packages/core/src/installation/index.ts

@@ -0,0 +1,110 @@
+import path from "path";
+import { $ } from "bun";
+import z from "zod";
+import { Bus } from "../bus/index.js";
+import { Log } from "../utils/log.js";
+import { NamedError } from "../utils/error.js";
+
+declare global {
+  const NESTRI_VERSION: string;
+  const NESTRI_CHANNEL: string;
+}
+
+export namespace Installation {
+  const log = Log.create({ service: "installation" });
+
+  export type Method = Awaited<ReturnType<typeof method>>;
+
+  export const Event = {
+    Updated: Bus.createEvent(
+      "installation.updated",
+      z.object({
+        version: z.string(),
+      }),
+    ),
+  };
+
+  export const Info = z
+    .object({
+      version: z.string(),
+      latest: z.string(),
+    })
+    .meta({
+      ref: "InstallationInfo",
+    });
+
+  export type Info = z.infer<typeof Info>;
+
+  export async function info() {
+    return {
+      version: VERSION,
+      latest: await latest(),
+    };
+  }
+
+  export function isPreview() {
+    return CHANNEL !== "latest";
+  }
+
+  export function isLocal() {
+    return CHANNEL === "local";
+  }
+
+  export async function method() {
+    if (process.execPath.includes(path.join(".nestri", "bin"))) return "curl";
+    if (process.execPath.includes(path.join(".local", "bin"))) return "curl";
+
+    return "unknown";
+  }
+
+  export const UpgradeFailedError = NamedError.create(
+    "UpgradeFailedError",
+    z.object({
+      stderr: z.string(),
+    }),
+  );
+
+  export async function upgrade(method: Method, target: string) {
+    let cmd;
+    switch (method) {
+      case "curl":
+        cmd = $`curl -fsSL https://nestri.io/install | bash`.env({
+          ...process.env,
+          VERSION: target,
+        });
+        break;
+      default:
+        throw new Error(`Unknown method: ${method}`);
+    }
+    const result = await cmd.quiet().throws(false);
+    log.info("upgraded", {
+      method,
+      target,
+      stdout: result.stdout.toString(),
+      stderr: result.stderr.toString(),
+    });
+    if (result.exitCode !== 0)
+      throw new UpgradeFailedError({
+        stderr: result.stderr.toString("utf8"),
+      });
+  }
+
+  export const VERSION =
+    typeof NESTRI_VERSION === "string" ? NESTRI_VERSION : "local";
+  export const CHANNEL =
+    typeof NESTRI_CHANNEL === "string" ? NESTRI_CHANNEL : "local";
+  export const USER_AGENT = `nestri/${CHANNEL}/${VERSION}`;
+
+  export async function latest() {
+    const res = await fetch(
+      `https://api.github.com/repos/nestrilabs/nestri/releases/latest`,
+    );
+
+    if (!res.ok) {
+      throw new Error(`GitHub API error: ${res.statusText}`);
+    }
+
+    const data = await res.json();
+    return data.tag_name?.replace(/^v/, ""); // removes leading "v" if present
+  }
+}

packages/core/src/redis/index.ts

@@ -1,8 +1,8 @@
 import { Log } from "../utils/log.js";
-import { Redis, type RedisOptions } from "ioredis";
+import { Redis as IORedis, type RedisOptions } from "ioredis";
 
 export {
-  Redis,
+  Redis as IORedis,
   Cluster,
   type Callback,
   type RedisOptions,
@@ -10,22 +10,22 @@ export {
   type RedisCommander,
 } from "ioredis";
 
-const defaultOptions: Partial<RedisOptions> = {
-  retryStrategy: (times: number) => {
-    const delay = Math.min(times * 50, 1000);
-    return delay;
-  },
-  maxRetriesPerRequest: 20,
-};
+export namespace Redis {
+  const defaultOptions: Partial<RedisOptions> = {
+    retryStrategy: (times: number) => {
+      const delay = Math.min(times * 50, 1000);
+      return delay;
+    },
+    maxRetriesPerRequest: 20,
+  };
 
-const logger = Log.create({ namespace: "redis" });
+  const logger = Log.create({ service: "redis" });
 
-export namespace RedisClient {
   export function create(
     options: RedisOptions,
     handlers?: { onError?: (err: Error) => void },
   ) {
-    const client = new Redis({
+    const client = new IORedis({
       ...defaultOptions,
       ...options,
     });

packages/core/src/utils/error.ts

@@ -0,0 +1,59 @@
+import z from "zod/v4";
+
+export abstract class NamedError extends Error {
+  abstract schema(): z.core.$ZodType;
+  abstract toObject(): { name: string; data: any };
+
+  static create<Name extends string, Data extends z.core.$ZodType>(
+    name: Name,
+    data: Data,
+  ) {
+    const schema = z
+      .object({
+        name: z.literal(name),
+        data,
+      })
+      .meta({
+        ref: name,
+      });
+    const result = class extends NamedError {
+      public static readonly Schema = schema;
+
+      public override readonly name = name as Name;
+
+      constructor(
+        public readonly data: z.input<Data>,
+        options?: ErrorOptions,
+      ) {
+        super(name, options);
+        this.name = name;
+      }
+
+      static isInstance(input: any): input is InstanceType<typeof result> {
+        return (
+          typeof input === "object" && "name" in input && input.name === name
+        );
+      }
+
+      schema() {
+        return schema;
+      }
+
+      toObject() {
+        return {
+          name: name,
+          data: this.data,
+        };
+      }
+    };
+    Object.defineProperty(result, "name", { value: name });
+    return result;
+  }
+
+  public static readonly Unknown = NamedError.create(
+    "UnknownError",
+    z.object({
+      message: z.string(),
+    }),
+  );
+}

packages/core/src/utils/fs.ts

@@ -0,0 +1,73 @@
+import { exists } from "fs/promises";
+import { dirname, join, relative } from "path";
+
+export namespace Filesystem {
+  export function overlaps(a: string, b: string) {
+    const relA = relative(a, b);
+    const relB = relative(b, a);
+    return !relA || !relA.startsWith("..") || !relB || !relB.startsWith("..");
+  }
+
+  export function contains(parent: string, child: string) {
+    return !relative(parent, child).startsWith("..");
+  }
+
+  export async function findUp(target: string, start: string, stop?: string) {
+    let current = start;
+    const result = [];
+    while (true) {
+      const search = join(current, target);
+      if (await exists(search)) result.push(search);
+      if (stop === current) break;
+      const parent = dirname(current);
+      if (parent === current) break;
+      current = parent;
+    }
+    return result;
+  }
+
+  export async function* up(options: {
+    targets: string[];
+    start: string;
+    stop?: string;
+  }) {
+    const { targets, start, stop } = options;
+    let current = start;
+    while (true) {
+      for (const target of targets) {
+        const search = join(current, target);
+        if (await exists(search)) yield search;
+      }
+      if (stop === current) break;
+      const parent = dirname(current);
+      if (parent === current) break;
+      current = parent;
+    }
+  }
+
+  export async function globUp(pattern: string, start: string, stop?: string) {
+    let current = start;
+    const result = [];
+    while (true) {
+      try {
+        const glob = new Bun.Glob(pattern);
+        for await (const match of glob.scan({
+          cwd: current,
+          absolute: true,
+          onlyFiles: true,
+          followSymlinks: true,
+          dot: true,
+        })) {
+          result.push(match);
+        }
+      } catch {
+        // Skip invalid glob patterns
+      }
+      if (stop === current) break;
+      const parent = dirname(current);
+      if (parent === current) break;
+      current = parent;
+    }
+    return result;
+  }
+}

packages/core/tsconfig.base.json

@@ -1,7 +1,7 @@
 {
   "compilerOptions": {
-    "target": "es2022",
-    "lib": ["ES2022", "DOM", "DOM.Iterable", "DOM.AsyncIterable"],
+    "target": "es2023",
+    "lib": ["ES2023", "DOM", "DOM.Iterable", "DOM.AsyncIterable"],
     "module": "NodeNext",
     "moduleResolution": "NodeNext",
     "moduleDetection": "force",