environmentTheme.ts

// @effect-diagnostics nodeBuiltinImport:off - the guarded file read needs open
// flags (O_NOFOLLOW, O_NONBLOCK) the FileSystem service does not expose.
/**
 * EnvironmentTheme - palettes this machine publishes for clients to follow.
 *
 * A desktop that retints its apps when the user switches system theme writes
 * `<stateDir>/themes/<id>.json`; this service watches that directory and
 * streams the published set to connected clients so a theme change lands
 * without a restart. The filename is the theme id: it stays stable while the
 * machine rewrites the colors underneath, so `defaultTheme` and a client\'s
 * selection keep pointing at the same theme across recolors. Theming is
 * cosmetic, so every failure here degrades to "not published" rather than
 * propagating.
 *
 * @module EnvironmentTheme
 */
import * as NodeFS from "node:fs";

import {
  EnvironmentTheme,
  EnvironmentThemeFile,
  EnvironmentThemeId,
  environmentThemeFileHasColors,
} from "@t3tools/contracts";
import { UNPUBLISHABLE_THEME_IDS } from "@t3tools/shared/themePalettes";
import * as Cause from "effect/Cause";
import * as Context from "effect/Context";
import * as Duration from "effect/Duration";
import * as Effect from "effect/Effect";
import * as Equal from "effect/Equal";
import * as Exit from "effect/Exit";
import * as FileSystem from "effect/FileSystem";
import * as Layer from "effect/Layer";
import * as PubSub from "effect/PubSub";
import * as Ref from "effect/Ref";
import * as Schema from "effect/Schema";
import * as Semaphore from "effect/Semaphore";
import * as Scope from "effect/Scope";
import * as Stream from "effect/Stream";

import * as ServerConfig from "./config.ts";

const decodeEnvironmentThemeFileJsonExit = Schema.decodeUnknownExit(
  Schema.fromJsonString(EnvironmentThemeFile),
);
const isEnvironmentThemeId = Schema.is(EnvironmentThemeId);

const THEME_FILE_SUFFIX = ".json";

/**
 * Bounds on what a machine can publish. The directory is local, so this is not
 * a trust boundary -- but an accidental dump of large files there would
 * otherwise be read in full, streamed to every client, and repainted, so the
 * cost of a mistake is capped rather than unbounded.
 */
const MAX_THEME_FILES = 32;
/** Exported so the publish path cannot accept a file the watcher will skip. */
export const MAX_THEME_FILE_BYTES = 32 * 1024;
/**
 * The set travels whole in a websocket event to every subscriber, so the sum
 * matters more than any single file. An exported theme runs a few KB, leaving
 * this far above any real directory while keeping a mistake off the wire.
 */
const MAX_THEME_TOTAL_BYTES = 192 * 1024;

/** The published set with the sequence number it was observed at. */
interface PublishedThemes {
  readonly seq: number;
  readonly themes: ReadonlyArray<EnvironmentTheme>;
}

export class EnvironmentThemeService extends Context.Service<
  EnvironmentThemeService,
  {
    /**
     * The set published right now, read from disk rather than from the
     * watcher\'s last observation: a client connecting must see what the
     * machine actually publishes even if it missed a filesystem event.
     */
    readonly current: Effect.Effect<ReadonlyArray<EnvironmentTheme>>;

    /**
     * The current set followed by every change, with repeats dropped. The
     * subscription is acquired before the current set is read, so a publish
     * landing while a client connects is delivered rather than lost.
     */
    readonly streamChanges: Stream.Stream<ReadonlyArray<EnvironmentTheme>>;
  }
>()("t3/environmentTheme/EnvironmentThemeService") {}

/**
 * Reads a theme file through one opened handle, so every check binds to the
 * file actually read rather than to a path that may have been swapped since:
 * O_NOFOLLOW rejects a symlink outright (a symlinked themes directory stays
 * usable, a symlinked file inside it does not), O_NONBLOCK keeps a FIFO from
 * blocking the open, and the fstat type and size gate examines the open
 * descriptor. Returns null for anything that is not a small regular file.
 */
export const readThemeFileGuarded = (filePath: string, maxBytes: number): string | null => {
  let fd: number;
  try {
    fd = NodeFS.openSync(
      filePath,
      NodeFS.constants.O_RDONLY | NodeFS.constants.O_NOFOLLOW | NodeFS.constants.O_NONBLOCK,
    );
  } catch {
    return null;
  }
  try {
    const info = NodeFS.fstatSync(fd);
    if (!info.isFile() || info.size > maxBytes) return null;
    const contents = Buffer.alloc(info.size);
    let offset = 0;
    while (offset < contents.length) {
      const read = NodeFS.readSync(fd, contents, offset, contents.length - offset, offset);
      if (read <= 0) break;
      offset += read;
    }
    return contents.subarray(0, offset).toString("utf8");
  } catch {
    return null;
  } finally {
    NodeFS.closeSync(fd);
  }
};

/**
 * Every theme the directory actually publishes. A file that is missing,
 * unreadable, malformed, colorless, or misnamed is simply skipped; the rest of
 * the set is unaffected. The one place that decides what "published" means, so
 * a caller validating an id cannot disagree with the watcher serving it.
 */
export const readPublishedThemes = Effect.fn(function* (themesDir: string) {
  const fs = yield* FileSystem.FileSystem;
  const entries = yield* fs
    .readDirectory(themesDir)
    .pipe(Effect.orElseSucceed((): Array<string> => []));

  const themes: Array<EnvironmentTheme> = [];
  let examined = 0;
  let totalBytes = 0;
  for (const entry of entries.toSorted()) {
    if (!entry.endsWith(THEME_FILE_SUFFIX)) continue;
    const id = entry.slice(0, -THEME_FILE_SUFFIX.length);
    // A reserved id is either shadowed by a built-in on the client or captures
    // clients that never chose it, so it is not publishable.
    if (!isEnvironmentThemeId(id) || UNPUBLISHABLE_THEME_IDS.has(id)) continue;

    // Counts files examined, not themes accepted: capping the output would
    // let a directory of malformed files be opened, read, and decoded in full
    // on every refresh and every client connect.
    examined += 1;
    if (examined > MAX_THEME_FILES) {
      yield* Effect.logWarning("ignoring environment theme files past the limit", {
        path: themesDir,
        limit: MAX_THEME_FILES,
      });
      break;
    }

    const filePath = `${themesDir}/${entry}`;
    const raw = readThemeFileGuarded(filePath, MAX_THEME_FILE_BYTES);
    if (raw === null) {
      yield* Effect.logWarning("ignoring unusable environment theme file", {
        path: filePath,
        limit: MAX_THEME_FILE_BYTES,
      });
      continue;
    }
    if (raw.trim().length === 0) continue;

    const decoded = decodeEnvironmentThemeFileJsonExit(raw);
    if (decoded._tag === "Failure") {
      yield* Effect.logWarning("ignoring invalid environment theme", {
        path: filePath,
        detail: Cause.pretty(decoded.cause),
      });
      continue;
    }
    const file = decoded.value;
    if (!environmentThemeFileHasColors(file)) {
      yield* Effect.logWarning("ignoring environment theme without colors", { path: filePath });
      continue;
    }

    // Counted only once accepted: the cap bounds what travels to clients, so
    // a skipped file must not eat the budget of valid themes sorted after it.
    // Bytes, not string length -- the cap describes wire weight.
    totalBytes += Buffer.byteLength(raw);
    if (totalBytes > MAX_THEME_TOTAL_BYTES) {
      yield* Effect.logWarning("ignoring environment themes past the total size limit", {
        path: themesDir,
        limit: MAX_THEME_TOTAL_BYTES,
      });
      break;
    }

    themes.push({ id, ...file });
  }
  return themes;
});

/**
 * Reads the directory and folds it into the sequenced state, publishing only
 * a genuine change. Every reader goes through here, so the snapshot a client
 * connects on and the events it then receives come from one ordered source
 * rather than from disk and the queue independently.
 */

const make = Effect.gen(function* () {
  const { environmentThemesDir } = yield* ServerConfig.ServerConfig;
  const fs = yield* FileSystem.FileSystem;
  /**
   * Sliding with capacity 1: every update carries the complete set, so a
   * subscriber that stops consuming holds at most the newest set rather than
   * an unbounded backlog. Every observed set carries a sequence number, so a
   * subscriber can drop queued events that predate the snapshot it started
   * from. Without it a publish landing between subscribing and reading
   * replays after the newer value and walks clients backwards onto stale
   * colors.
   */
  const changes = yield* PubSub.sliding<PublishedThemes>(1);
  const published = yield* Ref.make<PublishedThemes>({ seq: 0, themes: [] });
  /**
   * Guards the whole read/compare/publish, not just the state update. The
   * directory read is async, so two concurrent refreshes can finish out of
   * order and a slower read of an older set would publish under a higher
   * sequence -- which the subscriber filter, ordering publications rather than
   * observations, could not then drop.
   */
  const refreshSemaphore = yield* Semaphore.make(1);
  const watcherScope = yield* Scope.make("sequential");
  yield* Effect.addFinalizer(() => Scope.close(watcherScope, Exit.void));

  const refresh = refreshSemaphore.withPermits(1)(
    Effect.gen(function* () {
      const themes = yield* readPublishedThemes(environmentThemesDir).pipe(
        Effect.provideService(FileSystem.FileSystem, fs),
      );
      // Structural equality over the whole decoded value: a hand-rolled field
      // list here silently drops republishes for any field it forgets.
      const [changed, next] = yield* Ref.modify(
        published,
        (previous): readonly [readonly [boolean, PublishedThemes], PublishedThemes] => {
          if (Equal.equals(previous.themes, themes)) return [[false, previous], previous];
          const updated: PublishedThemes = { seq: previous.seq + 1, themes };
          return [[true, updated], updated];
        },
      );
      if (changed) yield* PubSub.publish(changes, next).pipe(Effect.asVoid);
      return next;
    }),
  );

  // The directory is created up front so the watcher has something to attach
  // to before the first publisher writes into it.
  yield* fs
    .makeDirectory(environmentThemesDir, { recursive: true })
    .pipe(Effect.ignoreCause({ log: true }));

  // Debounced for the same reason settings watching is: a theme script emits
  // several events per save and `fs.watch` can fire before the content is
  // flushed. Every event triggers a full re-read, so no event needs filtering.
  const watchEvents = fs.watch(environmentThemesDir).pipe(Stream.debounce(Duration.millis(100)));

  // Seeds the dedupe so a watch event that reports no actual change (a touch,
  // a rewrite with identical contents) does not retint every client.
  yield* refresh;
  yield* Stream.runForEach(watchEvents, () => refresh.pipe(Effect.ignoreCause({ log: true }))).pipe(
    Effect.ignoreCause({ log: true }),
    Effect.forkIn(watcherScope),
    Effect.asVoid,
  );

  return {
    current: Effect.map(refresh, (state) => state.themes),
    get streamChanges() {
      return Stream.unwrap(
        Effect.gen(function* () {
          // Subscribe first so nothing published during the read is missed,
          // then drop anything the snapshot already accounts for.
          const subscription = yield* PubSub.subscribe(changes);
          const snapshot = yield* refresh;
          return Stream.concat(
            Stream.make(snapshot.themes),
            Stream.fromSubscription(subscription).pipe(
              Stream.filter((update) => update.seq > snapshot.seq),
              Stream.map((update) => update.themes),
            ),
          );
        }),
      );
    },
  } satisfies EnvironmentThemeService["Service"];
});

export const layer = Layer.effect(EnvironmentThemeService, make);