fusion/packages/dashboard/src/project-store-resolver.ts

/**
 * Project-scoped TaskStore resolver for the dashboard server.
 *
 * Caches TaskStore instances by projectId so that SSE subscriptions
 * and API route handlers for the same project share a single in-memory
 * EventEmitter. Without this cache, every call to
 * `TaskStore.getOrCreateForProject()` creates an independent TaskStore
 * with its own EventEmitter — mutations on one instance would never
 * reach SSE listeners on another, breaking real-time dashboard updates
 * for project-scoped views.
 *
 * Usage:
 *   import { getOrCreateProjectStore } from "./project-store-resolver.js";
 *   const store = await getOrCreateProjectStore(projectId);
 */

import type { TaskStore } from "@fusion/core";

/**
 * Internal cache: projectId → TaskStore instance.
 * Keyed by projectId (not project path) because the dashboard server
 * routes identify projects by their central-registry ID.
 */
const storeCache = new Map<string, TaskStore>();

/**
 * In-flight creation promises, keyed by projectId.
 * Prevents concurrent requests from creating duplicate store instances
 * before the first creation completes and is added to storeCache.
 */
const pendingCreations = new Map<string, Promise<TaskStore>>();

/**
 * Track which stores have been fully initialized for real-time operation
 * (watcher started). This prevents duplicate watch() calls on repeated
 * lookups.
 */
const initializedProjects = new Set<string>();
const projectRegisteredListeners = new Set<(projectId: string, store: TaskStore) => void>();

/**
 * Optional callback invoked once when a new project store is first created.
 * Used by the dashboard server to lazily start an engine for secondary projects.
 */
let _onProjectFirstCreated: ((projectId: string) => void) | undefined;

/**
 * Register a callback to be called once when a new project is first accessed.
 * The callback fires after the store is cached — exactly once per projectId.
 * Pass `undefined` to clear the callback.
 */
export function setOnProjectFirstCreated(cb: ((projectId: string) => void) | undefined): void {
  _onProjectFirstCreated = cb;
}

/**
 * Get or create a cached TaskStore for the given projectId.
 *
 * - First call for a projectId: creates, inits, and caches the store.
 *   Also starts the SQLite polling watcher so external changes (CLI,
 *   engine agents) are detected and emitted as events.
 * - Subsequent calls: returns the cached instance immediately.
 *
 * Concurrent calls for the same projectId are deduplicated via a pending
 * promise map, preventing the race condition where the SSE endpoint and
 * an API mutation request both miss the cache and create separate store
 * instances with independent EventEmitters.
 *
 * @param projectId - The central-registry project ID
 * @returns A shared TaskStore instance for this project
 */
export async function getOrCreateProjectStore(projectId: string): Promise<TaskStore> {
  const cached = storeCache.get(projectId);
  if (cached) {
    return cached;
  }

  // Deduplicate concurrent creation requests so SSE and API routes always
  // share the same store instance even when both call this before the first
  // creation completes.
  const pending = pendingCreations.get(projectId);
  if (pending) {
    return pending;
  }

  const creation = (async () => {
    const { TaskStore: TaskStoreClass } = await import("@fusion/core");
    const store = await TaskStoreClass.getOrCreateForProject(projectId);

    // Start watching for external changes (CLI, engine agents, etc.)
    // so SSE listeners receive live events even when mutations happen
    // outside this process.
    if (!initializedProjects.has(projectId)) {
      initializedProjects.add(projectId);
      await store.watch();
    }

    storeCache.set(projectId, store);
    pendingCreations.delete(projectId);

    // Notify once that a new project was first accessed
    if (_onProjectFirstCreated) {
      _onProjectFirstCreated(projectId);
    }

    for (const listener of projectRegisteredListeners) {
      listener(projectId, store);
    }

    return store;
  })();

  pendingCreations.set(projectId, creation);
  return creation;
}

/**
 * Remove a cached store and stop its watcher.
 * Useful for cleanup on project removal or server shutdown.
 */
export function evictProjectStore(projectId: string): void {
  pendingCreations.delete(projectId);
  const store = storeCache.get(projectId);
  if (store) {
    store.stopWatching();
    store.close();
    storeCache.delete(projectId);
    initializedProjects.delete(projectId);
  }
}

/**
 * Evict all cached stores. Used during server shutdown.
 */
export function evictAllProjectStores(): void {
  pendingCreations.clear();
  for (const projectId of storeCache.keys()) {
    evictProjectStore(projectId);
  }
}

/**
 * Invalidate the global settings cache in all cached project stores.
 *
 * Each project-specific TaskStore holds its own GlobalSettingsStore with an
 * in-memory cache. When global settings are updated via the main store (e.g.,
 * PUT /settings/global), the file on disk is updated but the per-project
 * caches remain stale. Calling this function forces the next getSettings()
 * call in each project store to re-read from disk.
 */
export function invalidateAllGlobalSettingsCaches(): void {
  for (const store of storeCache.values()) {
    store.getGlobalSettingsStore().invalidateCache();
  }
}

export function listRegisteredProjectStores(): Array<{ projectId: string; store: TaskStore }> {
  return Array.from(storeCache.entries(), ([projectId, store]) => ({ projectId, store }));
}

export function onProjectStoreRegistered(listener: (projectId: string, store: TaskStore) => void): () => void {
  projectRegisteredListeners.add(listener);
  return () => {
    projectRegisteredListeners.delete(listener);
  };
}