routstrd/src/daemon/index.ts

import { createServer } from "http";
import { existsSync, unlinkSync } from "fs";
import {
  ProviderManager,
  createStorageAdapterFromStore,
  createSdkStore,
} from "@routstr/sdk";
// ModelManager must come from the bun entrypoint so persistent Nostr event
// storage (eventStoreDbPath) gets its SQLite-backed factory. The default
// "@routstr/sdk" export is browser-safe and throws without that factory
// (SDK 0.3.7+ browser-safe entrypoint split).
import { ModelManager } from "@routstr/sdk/bun";
import type { SdkLogger } from "@routstr/sdk";
import {
  CONFIG_DIR,
  DB_PATH,
  SOCKET_PATH,
  PID_FILE,
  REQUEST_RESPONSE_LOGS_DIR,
} from "../utils/config";
import { logger } from "../utils/logger";


function makeSdkLogger(prefix?: string): SdkLogger {
  const tag = prefix ? `[${prefix}]` : undefined;
  const fmt = (...args: unknown[]) => (tag ? [tag, ...args] : args);
  return {
    log: (...args: unknown[]) => logger.log(...fmt(...args)),
    warn: (...args: unknown[]) => logger.warn(...fmt(...args)),
    error: (...args: unknown[]) => logger.error(...fmt(...args)),
    debug: (...args: unknown[]) => logger.debug(...fmt(...args)),
    child: (p: string) => makeSdkLogger(prefix ? `${prefix}:${p}` : p),
  };
}
const daemonSdkLogger: SdkLogger = makeSdkLogger();
const STARTUP_LOG_PREFIX = "[routstrd:start]";

function startupProgress(message: string): void {
  logger.info(message);
  console.log(`${STARTUP_LOG_PREFIX} ${message}`);
}

import { parseArgs } from "./args";
import { ensureDirs, loadDaemonConfig, loadDaemonConfigSync, saveDaemonConfig } from "./config-store";
import {
  createBunSqliteDriver,
  createBunSqliteUsageTrackingDriver,
  createShardedDiscoveryAdapter,
} from "@routstr/sdk/storage/bun";
import { createWalletAdapter } from "./wallet";
import type { AutoRefillConfig } from "./wallet/auto-refill";
import { createModelService } from "./models";
import { createDaemonRequestHandler } from "./http";
import { FileRequestResponseLogSink } from "./request-response-log-sink";
import { refreshModelsAndIntegrations } from "../integrations";
import { RoutstrClient } from "@routstr/sdk";
import { mkdirSync } from "fs";
import { dirname } from "path";
import {
  assertLegacyCocodNotRunning,
  claimLegacyCocodPidFile,
  createCocoClient,
  stopLegacyCocod,
} from "./wallet/coco-client";
import { migrateLegacyWallet } from "./wallet/migration";
import {
  legacyCocodPidPath,
  legacyCocodSocketPath,
} from "./wallet/paths";
import { installGlobalErrorHandlers } from "./fatal-error";

// Global error handlers — the daemon is spawned detached with stdout/stderr
// redirected to a file, so without these, uncaught async errors would kill
// the process silently. Uncaught exceptions are fatal: the process state can
// no longer be trusted, so the daemon logs and exits for its supervisor to
// restart (see fatal-error.ts).
export async function runDaemon(argv: string[] = process.argv): Promise<void> {
  installGlobalErrorHandlers();
  // Install signal handlers before migration and wallet startup. If a signal
  // arrives before the full shutdown path is wired, process.exit() still runs
  // the synchronous PID-lock exit hooks registered by claimPidFile.
  let shutdownDaemon: () => void = () => process.exit(0);
  const shutdownForSignal = (signal: NodeJS.Signals) => {
    logger.log(`Received ${signal}; shutting down...`);
    shutdownDaemon();
  };
  process.once("SIGINT", shutdownForSignal);
  process.once("SIGTERM", shutdownForSignal);

  startupProgress("Loading configuration...");
  const args = parseArgs(argv);
  const config = await loadDaemonConfig();

  const port = args.port ?? config.port ?? 8008;
  const host = args.host || config.host || "127.0.0.1";
  const provider = args.provider || config.provider;
  const requestResponseLogDir =
    process.env.ROUTSTRD_REQUEST_RESPONSE_LOG_DIR ||
    (config.requestResponseLogging?.enabled
      ? config.requestResponseLogging.dir || REQUEST_RESPONSE_LOGS_DIR
      : undefined);
  const requestResponseLogSink = requestResponseLogDir
    ? new FileRequestResponseLogSink({
        dir: requestResponseLogDir,
        logger: daemonSdkLogger.child("request-response-log"),
      })
    : undefined;

  await ensureDirs();

  const updatedConfig = { ...config, port, host, provider };
  saveDaemonConfig(updatedConfig);

  startupProgress("Opening Routstr databases...");
  const sqliteDriver = await createBunSqliteDriver(DB_PATH, { logger: daemonSdkLogger });
  const { store, hydrate } = createSdkStore({ driver: sqliteDriver });
  await hydrate;
  const usageTrackingDriver = await createBunSqliteUsageTrackingDriver({
    dbPath: DB_PATH,
    legacyStorageDriver: sqliteDriver,
  });

  const discoveryAdapter = await createShardedDiscoveryAdapter({ driver: sqliteDriver });
  const storageAdapter = createStorageAdapterFromStore(store);
  startupProgress("Routstr databases ready.");
  const modelManager = new ModelManager(discoveryAdapter, {
    logger: daemonSdkLogger,
    eventStoreDbPath: `${CONFIG_DIR}/events.db`,
    routstrPubkey: config.routstrPubkey,
    routstrModelsPubkey: config.routstrModelsPubkey,
    nostrRelays: config.relays,
  });
  // Create shared ProviderManager for consistent failure tracking across all requests
  const providerManager = new ProviderManager(discoveryAdapter, store, daemonSdkLogger);
  const { ensureProvidersBootstrapped, getRoutstr21Models, getModelProviders, refreshProvidersAndModels } =
    createModelService(modelManager, providerManager, store);

  // The daemon may be launched directly (or by an older/global CLI), so do
  // not rely on the parent command having stopped the external wallet first.
  await stopLegacyCocod({
    socketPath: legacyCocodSocketPath(),
    pidFilePath: legacyCocodPidPath(),
  });

  const migration = await migrateLegacyWallet({
    assertLegacyStopped: () =>
      assertLegacyCocodNotRunning({
        socketPath: legacyCocodSocketPath(),
      }),
    acquireLegacyLock: () => {
      mkdirSync(dirname(legacyCocodPidPath()), {
        recursive: true,
        mode: 0o700,
      });
      return claimLegacyCocodPidFile({
        pidFilePath: legacyCocodPidPath(),
      });
    },
  });
  if (migration.status === "migrated") {
    startupProgress(`Wallet migrated from ${migration.from} to ${migration.to}.`);
    for (const warning of migration.cleanupWarnings) logger.warn(warning);
  }

  const walletClient = await createCocoClient();

  // ── Auto-refill configuration ────────────────────────────────
  // Uses a getter that reads config from disk each cycle, so
  // CLI changes take effect immediately without a daemon restart.

  const getAutoRefillConfig = (): AutoRefillConfig | undefined => {
    const cfg = loadDaemonConfigSync();
    if (cfg.nwc?.autoRefill?.enabled && cfg.nwc?.connectionString) {
      return {
        threshold: cfg.nwc.autoRefill.threshold,
        amount: cfg.nwc.autoRefill.amount,
        cooldownMs: cfg.nwc.autoRefill.cooldownMs,
      };
    }
    return undefined;
  };

  const walletAdapter = await createWalletAdapter({
    cocodPath: config.cocodPath,
    walletClient,
    getAutoRefillConfig,
    nwcConnectionString: config.nwc?.connectionString,
  });

  const refundClient = new RoutstrClient(
    walletAdapter,
    storageAdapter,
    discoveryAdapter,
    "min",
    "apikeys",
    { logger: daemonSdkLogger },
  );

  const server = createServer();
  server.on(
    "request",
    createDaemonRequestHandler({
      provider,
      server,
      shutdown: () => shutdownDaemon(),
      store,
      walletClient,
      walletAdapter,
      storageAdapter,
      discoveryAdapter,
      modelManager,
      ensureProvidersBootstrapped,
      getRoutstr21Models,
      getModelProviders,
      refreshProvidersAndModels,
      mode: config.mode || "apikeys",
      maxTokens: config.maxTokens ?? 64000,
      routstrPubkey: config.routstrPubkey,
      routstrModelsPubkey: config.routstrModelsPubkey,
      usageTrackingDriver,
      providerManager,
      refundClient,
      requestResponseLogSink,
    }),
  );

  Bun.write(PID_FILE, String(process.pid));

  try {
    if (existsSync(SOCKET_PATH)) {
      unlinkSync(SOCKET_PATH);
    }
  } catch {
    // Ignore
  }

  const DEFAULT_REFRESH_INTERVAL_MS = 21 * 60 * 1000; // 21 mins
  // While the job is disabled we keep a light poll running so `clients
  // --enable-automatic-refresh` takes effect without a daemon restart.
  const DISABLED_REFRESH_POLL_MS = 60 * 1000;

  /**
   * Read autoRefresh from disk on every tick (like the NWC auto-refill
   * getter) so CLI/config changes apply immediately without a restart.
   */
  const readAutoRefreshSettings = (): {
    enabled: boolean;
    intervalMs: number;
  } => {
    const autoRefresh = loadDaemonConfigSync().autoRefresh;
    const intervalMs =
      typeof autoRefresh?.intervalMs === "number" && autoRefresh.intervalMs > 0
        ? autoRefresh.intervalMs
        : DEFAULT_REFRESH_INTERVAL_MS;
    return { enabled: autoRefresh?.enabled !== false, intervalMs };
  };

  // Recurring job to refresh routstr21 models
  let refreshTimer: ReturnType<typeof setTimeout> | null = null;
  let refreshJobActive = false;
  let disabledNoticeLogged = false;

  const runScheduledRefresh = async (): Promise<void> => {
    logger.log("Running scheduled Nostr event refresh...");
    try {
      await modelManager.refreshNostrEvents();
    } catch (error) {
      logger.error("Scheduled Nostr event refresh failed:", error);
    }

    logger.log("Running scheduled model refresh...");
    try {
      await refreshModelsAndIntegrations(getRoutstr21Models, updatedConfig, "Scheduled");
    } catch (error) {
      logger.error("Scheduled model refresh failed:", error);
    }
  };

  const scheduleNextRefresh = (): void => {
    if (!refreshJobActive) return;

    const { enabled, intervalMs } = readAutoRefreshSettings();
    if (!enabled) {
      if (!disabledNoticeLogged) {
        logger.log(
          "Scheduled refresh job is disabled (autoRefresh.enabled=false). Polling for re-enable every 60s.",
        );
        disabledNoticeLogged = true;
      }
      refreshTimer = setTimeout(
        scheduleNextRefresh,
        Math.min(intervalMs, DISABLED_REFRESH_POLL_MS),
      );
      return;
    }

    if (disabledNoticeLogged) {
      logger.log("Scheduled refresh job re-enabled.");
      disabledNoticeLogged = false;
    }

    refreshTimer = setTimeout(() => {
      void runScheduledRefresh()
        .catch((error) => logger.error("Model refresh interval escaped:", error))
        .finally(() => scheduleNextRefresh());
    }, intervalMs);
  };

  const startModelRefreshJob = () => {
    refreshJobActive = true;
    const { enabled, intervalMs } = readAutoRefreshSettings();
    logger.log(
      enabled
        ? `Starting recurring model refresh job (every ${Math.round(intervalMs / 60_000)} minutes)`
        : "Recurring model refresh job is disabled (autoRefresh.enabled=false).",
    );
    scheduleNextRefresh();
  };

  const stopModelRefreshJob = () => {
    refreshJobActive = false;
    if (refreshTimer) {
      clearTimeout(refreshTimer);
      refreshTimer = null;
      logger.log("Stopped recurring model refresh job.");
    }
  };

  // Recurring job to refund pending tokens every 42 minutes
  const REFUND_INTERVAL_MS = 42 * 60 * 1000; // 42 minutes
  let refundInterval: ReturnType<typeof setInterval> | null = null;

  const startRefundJob = async () => {
    logger.log(
      `Starting recurring refund job (every ${REFUND_INTERVAL_MS / 1000 / 60} minutes)`,
    );

    refundInterval = setInterval(() => {
      (async () => {
        logger.log("Running scheduled refund...");
        try {
          const state = store.getState() as any;
          const pendingDistribution = (state.cachedTokens || []).map(
            (t: { baseUrl: string; balance?: number }) => ({
              baseUrl: t.baseUrl,
              amount: t.balance || 0,
            }),
          );
          const apiKeysStored = (state.apiKeys || []).map(
            (k: { baseUrl: string; balance?: number }) => ({
              baseUrl: k.baseUrl,
              amount: k.balance || 0,
            }),
          );

          if (pendingDistribution.length === 0 && apiKeysStored.length === 0) {
            logger.log("No pending tokens to refund.");
            return;
          }

          const mintUrl = walletAdapter.getActiveMintUrl();
          if (!mintUrl) {
            logger.log("No active mint URL for refund.");
            return;
          }

          const spender = refundClient.getCashuSpender();
          const results = await spender.refundProviders(mintUrl);

          const successCount = results.filter(
            (r: { success: boolean }) => r.success,
          ).length;
          logger.log(
            `Scheduled refund completed: ${successCount}/${results.length} providers refunded.`,
          );
        } catch (error) {
          logger.error("Scheduled refund failed:", error);
        }
      })().catch((error) => logger.error("Refund interval escaped:", error));
    }, REFUND_INTERVAL_MS);
  };

  const stopRefundJob = () => {
    if (refundInterval) {
      clearInterval(refundInterval);
      refundInterval = null;
      logger.log("Stopped recurring refund job.");
    }
  };

  let walletDisposePromise: Promise<void> | undefined;
  const disposeWallet = (): Promise<void> => {
    walletDisposePromise ??= walletClient.dispose?.() || Promise.resolve();
    return walletDisposePromise;
  };

  server.on("close", () => {
    stopModelRefreshJob();
    stopRefundJob();
    void disposeWallet().catch((error) => {
      logger.error("Failed to dispose wallet:", error);
    });
  });

  shutdownDaemon = () => {
    server.close(() => {
      void disposeWallet().finally(() => process.exit(0));
    });
  };

  // Without this a listen failure only reaches uncaughtException, which logs and
  // returns, leaving the daemon alive holding the wallet locks with no listener.
  server.on("error", (error) => {
    logger.error("Failed to start Routstr daemon:", error);
    console.error(
      "Failed to start Routstr daemon:",
      error instanceof Error ? error.message : error,
    );
    process.exit(1);
  });

  startupProgress("Starting HTTP server...");
  server.listen(port, host, async () => {
    logger.log(`Routstr daemon listening on http://${host}:${port}/v1`);
    if (requestResponseLogDir) {
      logger.log(`Raw request/response logs: ${requestResponseLogDir}`);
    }

    // Start the recurring model refresh job after initial bootstrap
    void ensureProvidersBootstrapped()
      .then(async () => {
        // Catch up on any Nostr events published since last run
        logger.log("Running initial Nostr event refresh...");
        await modelManager.refreshNostrEvents();

        startModelRefreshJob();
        startRefundJob();
        // Run an immediate refresh to populate models right away. Client
        // integrations are skipped when the scheduled job is disabled, so a
        // restart does not overwrite hand-edited client configs.
        if (readAutoRefreshSettings().enabled) {
          logger.log("Running initial model refresh...");
          await refreshModelsAndIntegrations(getRoutstr21Models, updatedConfig, "Initial");
        } else {
          logger.log("Running initial model refresh (client integrations skipped)...");
          await getRoutstr21Models(true);
        }
      })
      .catch((error) => {
        logger.error("Initial model refresh failed:", error);
        // Still start the jobs even if initial refresh fails
        startModelRefreshJob();
        startRefundJob();
      });
  });
}

if (import.meta.main) {
  runDaemon().catch((error) => {
    logger.error("Failed to start Routstr daemon:", error);
    // Also write to stderr so the spawning CLI can surface the real error
    // (stdout/stderr are redirected to debug.log by start-daemon.ts).
    console.error(
      "Failed to start Routstr daemon:",
      error instanceof Error ? error.message : error,
    );
    process.exit(1);
  });
}