/**
 * Entry point for the realtime server (apps/realtime) — TASK-0091.
 *
 * Wires configuration, logging, and the production seams (Prisma
 * persistence + permission/meta resolvers, Redis update log) into the
 * server built in `server.ts`, and manages the process lifecycle:
 * startup, SIGTERM/SIGINT → the R6 drain (4010 + retryAfterMs to all
 * sockets, flush all dirty rooms, ≤ 10 s, exit 0).
 *
 * Dev mode without the sidecars: `REALTIME_REDIS=0` / `REALTIME_DB=0`
 * (config.ts) fall back to the in-memory update log + doc store.
 *
 * The production seam modules (`@prisma/client`, `ioredis`) load lazily
 * via DYNAMIC imports — this is an ESM package, and a top-level
 * `require()` crashed the process on boot in every mode (recorded
 * TASK-0091 fix). A missing/unbuildable client degrades to the
 * in-memory seam (dev mode) with a warning; any other seam error fails
 * the boot loudly.
 */

import pino from "pino";
import { loadConfig, type RealtimeConfig } from "./config.js";
import {
  createRealtimeServer,
  type AuditSink,
  type EventSink,
  type PermissionResolver,
  type RoomMetaResolver,
} from "./server.js";
import { InMemoryDocStore, type DocStore } from "./rooms.js";
import {
  InMemoryUpdateLog,
  RedisUpdateLog,
  type UpdateLog,
  type UpdateLogBackend,
} from "./update-log.js";
import {
  PrismaDocStore,
  createPrismaMetaResolver,
  createPrismaPermissionResolver,
} from "./prisma-store.js";
import { HttpDocStore } from "./http-store.js";
import { createDomainEventSource, type DomainEventSource } from "./event-source.js";
import { initOtel, shutdownOtel } from "@factory/observability";

async function main(): Promise<void> {
  const config = loadConfig();
  const logger = pino({ level: config.logLevel });

  // TASK-0135: OTel FIRST (same contract as the API/worker — the lazy metric
  // instruments bind to the SDK provider at first touch; disabled → no-op).
  // OTEL_* come straight from process.env (NOT in config.ts's schema).
  const otelEndpoint = process.env.OTEL_EXPORTER_OTLP_ENDPOINT;
  await initOtel({
    serviceName: "factory-realtime",
    environment: process.env.NODE_ENV,
    ...(process.env.OTEL_ENABLED === "true" && otelEndpoint ? { endpoint: otelEndpoint } : {})
  });

  const store = await buildStore(config, logger);
  const log = await buildUpdateLog(config, logger);
  const permissions = await buildPermissions(config, logger);
  const meta = await buildMeta(config, logger);

  const audit: AuditSink = {
    // The audit seam publishes the abuse event; the api's audit
    // consumer (the event bus) writes the row — the Phase-0/1 audit
    // path, no new table (task Backend Changes).
    publish: (event, payload) => {
      logger.warn({ event, ...payload }, "audit: realtime abuse event");
    },
  };

  // The domain-event seam (TASK-0092 R3): the flush emits `page.updated`
  // (the Phase-3 shape, the `blockDelta` additive field). The realtime app
  // has no Redis dependency of its own, so the seam is logged here (the
  // event is reconstructable from the structured log); the event-bus
  // wiring (XADD `factory:events`) is the Phase-8/production seam —
  // recorded in the task file (see the `EventSink` docs in server.ts).
  const events: EventSink = {
    publish: (event) => {
      logger.info(
        { name: event.name, pageId: (event.payload as { pageId?: string }).pageId },
        "event: domain event emitted",
      );
    },
  };

  // The (b) CONTROL producer (TASK-0092 6b, generalized in TASK-0094): poll
  // the API's `factory:events` stream ONCE and route every event —
  // `page.updated` → `block:changed` refetch-hint, `comment.*` →
  // `comment.event` room frames, `notification.created` → `notification.push`
  // control-room push (the server's routing sink does the per-event work).
  // Lazy ioredis, same degradation pattern as the update log — no Redis → no
  // control fan-out (the web overlay still works via its REST path; recorded
  // in the task file).
  const domainEventSourceFactory = await buildDomainEventSource(config, logger);

  const rt = createRealtimeServer(config, {
    logger,
    store,
    log,
    permissions,
    meta,
    audit,
    events,
    ...(domainEventSourceFactory !== null ? { domainEventSource: domainEventSourceFactory } : {}),
  });

  await rt.start();
  logger.info(
    { port: config.port, healthPort: config.healthPort },
    "realtime server started: ws + health listening",
  );

  let shuttingDown = false;
  const shutdown = (signal: string): void => {
    if (shuttingDown) return;
    shuttingDown = true;
    logger.info({ signal }, "shutting down realtime server (drain)");
    rt.drain()
      .catch((err: unknown) => {
        logger.error({ err: String(err), signal }, "error during realtime drain");
      })
      .finally(async () => {
        // TASK-0135: flush the OTel SDK (pending spans + the final metric
        // collection) before the process dies.
        await shutdownOtel().catch(() => undefined);
        process.exit(0);
      });
  };
  process.on("SIGINT", () => shutdown("SIGINT"));
  process.on("SIGTERM", () => shutdown("SIGTERM"));
}

/** True when a dynamic import failed because `@prisma/client` is not
 *  installed/generated yet (the dev-mode fallback case). */
function isMissingPrismaClient(err: unknown): boolean {
  if (err === null || typeof err !== "object") return false;
  const code = (err as { code?: unknown }).code;
  const msg = err instanceof Error ? err.message : "";
  return (
    code === "ERR_MODULE_NOT_FOUND" || /@prisma\/client|prisma.*not (yet )?generated/i.test(msg)
  );
}

/** Load the Prisma-backed seams (doc store + permission + meta). The
 *  modules are imported lazily so the process boots even when the
 *  generated client is absent (the `REALTIME_DB=0` dev contract). */
async function loadPrismaSeams(): Promise<{
  PrismaClient: new (cfg?: unknown) => import("@prisma/client").PrismaClient;
} | null> {
  try {
    // Both imports must land: prisma-store.js statically imports
    // @prisma/client, so they succeed or fail together.
    const [client] = await Promise.all([import("@prisma/client"), import("./prisma-store.js")]);
    return {
      PrismaClient: client.PrismaClient as new (
        cfg?: unknown,
      ) => import("@prisma/client").PrismaClient,
    };
  } catch (err) {
    if (isMissingPrismaClient(err)) return null;
    throw err; // unexpected seam error — fail the boot, do not degrade silently
  }
}

async function buildStore(config: RealtimeConfig, logger: pino.Logger): Promise<DocStore> {
  if (!config.useDatabase) {
    logger.warn({}, "REALTIME_DB=0 — using the in-memory doc store (dev mode)");
    return new InMemoryDocStore();
  }
  if (config.internalServiceToken.length > 0) {
    // Production seam (TASK-0092 slice 2): the doc store persists through
    // the API's internal endpoints (one Postgres write funnel — the API's
    // audit + event layers). Fail loud when the token is rejected on boot
    // (a misconfigured token would otherwise surface as a flush 401 at
    // first edit, mid-session).
    const store = new HttpDocStore({
      apiUrl: config.apiUrl,
      serviceToken: config.internalServiceToken,
    });
    if (!(await store.ping())) {
      throw new Error(
        "HttpDocStore boot check failed: the API's internal endpoints are unreachable " +
          `(${config.apiUrl}/internal/* — check API_URL + INTERNAL_SERVICE_TOKEN on BOTH sides)`,
      );
    }
    logger.info({ apiUrl: config.apiUrl }, "wired the HTTP doc store (API internal endpoints)");
    return store;
  }
  const seams = await loadPrismaSeams();
  if (seams === null) {
    logger.warn({}, "@prisma/client unavailable — in-memory doc store (dev mode)");
    return new InMemoryDocStore();
  }
  logger.info({}, "wired the Prisma doc store");
  return new PrismaDocStore(new seams.PrismaClient());
}

async function buildPermissions(
  config: RealtimeConfig,
  logger: pino.Logger,
): Promise<PermissionResolver> {
  if (!config.useDatabase) {
    logger.warn({}, "no DB — fail-closed permission resolver (view-only joins)");
  }
  const seams = await loadPrismaSeams();
  if (seams === null) {
    logger.warn({}, "no DB — fail-closed permission resolver (view-only joins)");
    return { resolveLevel: async () => "view" as const };
  }
  return createPrismaPermissionResolver(new seams.PrismaClient());
}

async function buildMeta(config: RealtimeConfig, logger: pino.Logger): Promise<RoomMetaResolver> {
  const seams = await loadPrismaSeams();
  if (seams === null) {
    logger.warn({}, "no DB — fail-closed meta resolver (all rooms 4002)");
    return { resolve: async () => null };
  }
  return createPrismaMetaResolver(new seams.PrismaClient());
}

function buildUpdateLog(config: RealtimeConfig, logger: pino.Logger): Promise<UpdateLog> {
  return (async () => {
    if (!config.useRedis) {
      logger.warn({}, "REALTIME_REDIS=0 — using the in-memory update log (dev mode)");
      return new InMemoryUpdateLog();
    }
    try {
      const { Redis } = await import("ioredis");
      const redis = new Redis(config.redisUrl, { maxRetriesPerRequest: 2 });
      redis.on("error", (err: Error) => {
        logger.warn({ err: err.message }, "redis connection error");
      });
      logger.info({}, "wired the Redis update log");
      const log = new RedisUpdateLog({
        redis: redis as unknown as UpdateLogBackend,
      });
      return {
        append: (k: string, u: string, v: number) => log.append(k, u, v),
        entriesAfter: (k: string, s: number) => log.entriesAfter(k, s),
        maxSeq: (k: string) => log.maxSeq(k),
        trimAfter: (k: string, s: number) => log.trimAfter(k, s),
        clear: (k: string) => log.clear(k),
        ping: async () => redis.status === "ready" || redis.status === "connecting",
      };
    } catch (err) {
      logger.warn({ err: String(err) }, "ioredis unavailable — in-memory update log (dev mode)");
      return new InMemoryUpdateLog();
    }
  })();
}

/** The (b) control producer (TASK-0092 6b, generalized in TASK-0094): a
 *  domain-event source that polls the API's `factory:events` stream (lazy
 *  ioredis, same degradation pattern as the update log). Returns NULL when
 *  there is no Redis (dev mode) — the server then runs no control fan-out
 *  (the web overlay still works via its REST path). */
async function buildDomainEventSource(
  config: RealtimeConfig,
  logger: pino.Logger,
): Promise<
  ((on: (event: import("@factory/shared").DomainEvent) => void) => DomainEventSource) | null
> {
  if (!config.useRedis) {
    logger.warn({}, "REALTIME_REDIS=0 — no domain-event control source (dev mode)");
    return null;
  }
  try {
    const { Redis } = await import("ioredis");
    const redis = new Redis(config.redisUrl, { maxRetriesPerRequest: 2 });
    redis.on("error", (err: Error) => {
      logger.warn({ err: err.message }, "redis connection error (domain-event source)");
    });
    logger.info({}, "wired the domain-event control source (factory:events stream)");
    return (on) =>
      createDomainEventSource(on, {
        client: redis as unknown as import("./event-source.js").DomainEventStreamClient,
        onError: (err) => logger.warn({ err: String(err) }, "domain-event source error"),
      });
  } catch (err) {
    logger.warn(
      { err: String(err) },
      "ioredis unavailable — no domain-event control source (dev mode)",
    );
    return null;
  }
}

main().catch((err: unknown) => {
  console.error("realtime server failed to start", err);
  process.exit(1);
});