import type { Hono } from "hono";
import type { AuthRoutesConfig } from "../api/auth-routes";
import type { JwtHelper } from "../api/jwt";
import { buildServer } from "../api/server";
import { createSseBroker, type SseBroker } from "../api/sse-broker";
import type { PgClient } from "../db/connection";
import { extractTableInfo } from "../db/query";
import { createRegistry } from "../engine/registry";
import type { AppContext, FeatureDefinition, JobRunIn, Registry, TenantId } from "../engine/types";
import { createArchivedStreamsTable, createEventsTable } from "../event-store";
import { createJobRunner, type JobRunner } from "../jobs";
import type { Lifecycle } from "../lifecycle";
import { createNoopProvider, type ObservabilityProvider } from "../observability";
import type { Dispatcher, EventDispatcher } from "../pipeline";
import { createEntityCache, createEventDedup, createIdempotencyGuard } from "../pipeline";
import { createInMemorySearchAdapter } from "../search";
import type { SearchAdapter } from "../search/types";
import { createTestDb } from "./db";
import { createEventCollector, type EventCollector } from "./event-collector";
import { createTestRedis, type TestRedis } from "./redis";
import { createRequestHelper, type RequestHelper } from "./request-helper";
import { unsafePushTables } from "./table-helpers";

export type TestStack = {
  app: Hono;
  jwt: JwtHelper;
  registry: Registry;
  db: import("../db").DbConnection;
  redis: TestRedis;
  search: SearchAdapter;
  events: EventCollector;
  http: RequestHelper;
  observability: ObservabilityProvider;
  // In-memory broker backing the server's SSE routes + access-invalidation
  // channel. Tests subscribe directly (e.g. sseBroker.subscribeAccessInvalidation)
  // to assert a consumer pushed an invalidation without opening a real SSE
  // connection.
  sseBroker: SseBroker;
  // Command-dispatcher behind the HTTP routes — for direct system-writes
  // in tests and dev-server extraRoutes (provider-webhook wiring).
  dispatcher: Dispatcher;
  // The AppContext buildServer handed the request path, incl. the fields it
  // wires itself (_fileProviderResolver). A dev-server that starts its own
  // lane job-runners beside this stack must hand them THIS, not a
  // `{ db, registry }` literal — that's the #1232 drift, and it makes an
  // event-triggered job reaching for ctx.files die where the request path works.
  context: AppContext;
  // Present whenever a system consumer (SSE, Search) or
  // r.multiStreamProjection is wired. Tests drain it via runOnce() for
  // deterministic assertion — no timer-induced flakiness.
  eventDispatcher?: EventDispatcher;
  // Only set when the caller passed `lifecycle` via options. Tests that
  // exercise drain() / /health/ready wire one in; ordinary suites ignore it.
  lifecycle?: Lifecycle;
  // Only set when `options.jobs` is truthy AND ≥1 `r.job(...)` is
  // registered in the mounted features. Lets integration tests call
  // `stack.jobRunner.dispatch(...)` directly, mirroring `ctx.jobRunner`.
  jobRunner?: JobRunner;
  cleanup: () => Promise<void>;
};

export type TestStackOptions = {
  features: readonly FeatureDefinition[];
  /** System hooks to wire up. Default: all (sse, search) */
  systemHooks?: ("sse" | "search")[];
  /** Search config per tenant — defaults to tenant 1 with all text fields */
  searchConfig?: {
    tenantId: TenantId;
    searchableFields: string[];
    rankingFields: string[];
  };
  jwtSecret?: string;
  /** Extra fields merged into the AppContext (e.g. _notifyFactory, configResolver).
   *  Can be a function receiving (registry, db, sseBroker) for late binding. */
  extraContext?:
    | Record<string, unknown>
    | ((deps: {
        registry: Registry;
        db: import("../db").DbConnection;
        sseBroker: import("../api/sse-broker").SseBroker;
        redis: import("ioredis").default;
      }) => Record<string, unknown>);
  /** Wire up auth routes (login, tenant-switch). Leave undefined to skip. */
  authConfig?: AuthRoutesConfig;
  /** Register a file storage provider so uploads via POST /api/files work and
   *  `ctx.files.ref(key)` is available to hooks/MSPs. Omit to skip — tests
   *  without file handling don't need it. */
  files?: { storageProvider: import("../files").FileStorageProvider };
  /** Observability provider — omit for NoopProvider (no spans/metrics).
   *  Pass a ConsoleProvider to see the span tree in stdout, or a custom
   *  provider (e.g. a recording provider for assertions in tests). */
  observability?: ObservabilityProvider;
  /** Inject a process lifecycle so tests can drain() and observe
   *  /health/ready flipping to 503. Omit if the suite doesn't care. */
  lifecycle?: Lifecycle;
  /** Wire L1 (global-IP) and/or L2 (auth-endpoint) rate-limit middleware.
   *  The resolver is auto-built from the test Redis. Mirrors
   *  buildServer's `rateLimit` option 1:1 — see there for shape. */
  rateLimit?: import("../api/server").ServerOptions["rateLimit"];
  /** Inject a MasterKeyProvider for secrets-backed tests. Lands typed in
   *  AppContext — set/delete/get + rotation job pick it up. Omit for
   *  suites that don't touch secrets. */
  masterKeyProvider?: import("../secrets").MasterKeyProvider;
  /** Feature-toggle resolver. When present the dispatcher's feature-gate,
   *  hook-filter, and MSP-filter all consult it; absent = every feature
   *  treated as always-on. Pass the callback from
   *  GlobalFeatureToggleRuntime.effectiveFeatures for real DB-backed
   *  toggles, or a plain `() => new Set<string>(registry.features.keys())`
   *  to force a specific snapshot in a unit-style setup. */
  effectiveFeatures?: (tenantId: TenantId) => ReadonlySet<string>;
  /** Pin the underlying Postgres DB name instead of the default
   *  `kumiko_test_<8chars>`. Forwarded to createTestDb. Primary use
   *  case: dev servers that want persistent storage across restarts —
   *  combine with `persistentDb: true`. */
  dbName?: string;
  /** When true, cleanup() keeps the Postgres DB around — the caller
   *  owns its lifecycle. Default false (test contract). Used by
   *  dev-server wiring to survive hot-reloads. */
  persistentDb?: boolean;
  /** Forwarded to buildServer — when set, requests without a JWT pass
   *  through as anonymous instead of 401. See AnonymousAccessConfig.
   *  Akzeptiert entweder einen statischen Config-Object ODER eine Factory
   *  `({registry, db, sseBroker, redis}) => Config` — gleiches Pattern wie
   *  `extraContext`. Die Factory wird einmal beim Boot aufgerufen, der
   *  TenantResolver darin closure'd typischerweise `db` für Subdomain-
   *  Lookups. */
  anonymousAccess?:
    | import("../api/server").ServerOptions["anonymousAccess"]
    | ((deps: {
        registry: Registry;
        db: import("../db").DbConnection;
        sseBroker: import("../api/sse-broker").SseBroker;
        redis: import("ioredis").default;
      }) => import("../api/server").ServerOptions["anonymousAccess"]);
  /** Optional post-factory enricher (e.g. merge auth-foundation tenant
   *  providers). Keeps framework free of a bundled-features dependency. */
  enrichAnonymousAccess?: (
    base: import("../api/server").ServerOptions["anonymousAccess"] | undefined,
    deps: {
      registry: Registry;
      db: import("../db").DbConnection;
    },
  ) => Promise<import("../api/server").ServerOptions["anonymousAccess"] | undefined>;
  /** Opt-in JobRunner wired into ctx.jobRunner and merged into
   *  dispatcherOptions so event-triggered jobs enqueue on commit — mirrors
   *  the prod entrypoint's `buildJobRunnerWithHook`. Unlike prod (which
   *  always builds one), this only builds a runner when `registry.getAllJobs()`
   *  is non-empty — a deliberate test-stack-only shortcut, not parity.
   *  Default `undefined`. Pass `consumerLane` only when this test IS the
   *  sole consumer (otherwise enqueuer-only, avoiding double-running
   *  `runOnBoot`/cron jobs against a caller-owned consumer). */
  jobs?: {
    consumerLane?: JobRunIn;
    queueNamePrefix?: string;
  };
  /** Override the event dispatcher's polling-timer interval. Default 50ms.
   *  Tests that assert LISTEN/NOTIFY wake-up latency need this pushed far
   *  out (e.g. 60_000) so the polling timer can't land inside the
   *  assertion window and mask a dead subscription — see E.4 (#2042). */
  eventDispatcherPollIntervalMs?: number;
};

const DEFAULT_JWT_SECRET = "test-stack-secret-minimum-32-characters!!";

export async function setupTestStack(options: TestStackOptions): Promise<TestStack> {
  const jwtSecret = options.jwtSecret ?? DEFAULT_JWT_SECRET;
  const enabledHooks = options.systemHooks ?? ["sse", "search"];

  // Temporal-Polyfill installieren bevor Feature-Code läuft. Idempotent —
  // Production-Server-Boot ruft das gleich. Auf Runtimes mit nativem
  // Temporal ein No-Op.
  const { ensureTemporalPolyfill } = await import("../time/polyfill");
  await ensureTemporalPolyfill();

  // Forward db-name/persistent-flag through to createTestDb. The
  // defaults (undefined dbName, persistent:false) keep the legacy
  // test contract: fresh kumiko_test_<random> DB per setup, dropped
  // on cleanup.
  const [testDb, testRedis] = await Promise.all([
    createTestDb({
      ...(options.dbName !== undefined && { dbName: options.dbName }),
      ...(options.persistentDb !== undefined && { persistent: options.persistentDb }),
    }),
    createTestRedis(),
  ]);

  // Every ES-entity writes events via createEventStoreExecutor in the
  // feature's write handlers. Auto-create the events table so every
  // setupTestStack call is ready for writes without needing a manual
  // createEventsTable().
  await createEventsTable(testDb.db);
  // Archive-stream metadata — needed by ctx.appendEvent's archive guard and
  // loadAggregate's default-skip. Idempotent, so production boot running
  // the same call is fine.
  await createArchivedStreamsTable(testDb.db);

  // Framework state for projection rebuild/status + event-consumer cursors.
  // Idempotent — production boot flows run the same calls.
  const { createProjectionStateTable, createEventConsumerStateTable } = await import("../pipeline");
  await createProjectionStateTable(testDb.db);
  await createEventConsumerStateTable(testDb.db);

  // Files support: when a provider is registered, the fileRefs table must
  // exist before the first upload. Skipped when no provider — the table
  // stays off tenant test DBs that never touch files.
  if (options.files) {
    const { fileRefsTable } = await import("../files");
    await unsafePushTables(testDb.db, { fileRefsTable });
  }

  // Projection-/MSP-/raw-tables: the executor (or async dispatcher) writes
  // into them as soon as the first matching event flows, so the DDL must
  // exist before setupTestStack returns. The source list is shared with
  // collectTableMetas (`kumiko schema generate`) — divergence between the
  // two was exactly the #255 prod-crash. Two registrations backed by the
  // same physical table (e.g. an alternative apply-shape for the same
  // read-model in a test feature) are deduped by table reference so we
  // emit only one CREATE TABLE per physical table.
  const { enumerateFeatureTableSources } = await import("../db/feature-table-sources");
  const projectionTables: Record<string, unknown> = {};
  // Dedup by NAME, matching collectTableMetas — by-reference alone let two
  // distinct table objects with the same name slip through as a double
  // CREATE TABLE while schema-generate emitted only one meta (silent
  // test-vs-schema divergence).
  const seenTableNames = new Set<string>();
  for (const feature of options.features) {
    for (const { table, origin } of enumerateFeatureTableSources(feature)) {
      const name = extractTableInfo(table).name;
      if (seenTableNames.has(name)) continue;
      seenTableNames.add(name);
      projectionTables[origin] = table;
    }
  }
  if (Object.keys(projectionTables).length > 0) {
    // unsafePushTables emits raw CREATE TABLE — fine for ephemeral test DBs but
    // collides on re-boot against a persistent DB whose projection tables
    // were created during a previous run. Filter out the ones that already
    // exist so the re-boot doesn't fail on duplicate CREATE TABLE.
    const { tableExists } = await import("../db/schema-inspection");
    const missing: Record<string, unknown> = {};
    for (const [key, tbl] of Object.entries(projectionTables)) {
      const physical = extractTableInfo(tbl).name;
      if (await tableExists(testDb.db, `public.${physical}`)) continue;
      missing[key] = tbl;
    }
    if (Object.keys(missing).length > 0) {
      await unsafePushTables(testDb.db, missing);
    }
  }

  const searchAdapter = createInMemorySearchAdapter();
  const events = createEventCollector();
  const registry = createRegistry([...options.features]);

  // Wire SSE broker with event collector — built early (not just before
  // buildServer) because extraContext + the job context below both need it.
  const sseBroker = createSseBroker();
  sseBroker.addClient(
    "tenant:00000000-0000-4000-8000-000000000001",
    (event) => events.sse.push(event),
    () => {},
  );

  const entityCache = createEntityCache(testRedis.redis, { ttlSeconds: 60 });

  // A static `files.storageProvider` is wired as the per-tenant resolver — the
  // framework test seam that doesn't require mounting config + file-foundation.
  // (Bundled GDPR tests mount the real provider features instead.)
  let fileProviderResolver: import("../files").FileProviderResolver | undefined;
  if (options.files) {
    const provider = options.files.storageProvider;
    fileProviderResolver = () => Promise.resolve(provider);
  }

  const observability = options.observability ?? createNoopProvider();

  // Same AppContext buildServer() gets below (db/redis/searchAdapter/
  // entityCache/masterKeyProvider/fileProviderResolver/extraContext) —
  // shared so the job context can't drift from the request-path context
  // (kumiko-framework#1232: a reduced `{ db, registry }` literal let jobs
  // pass in tests while reaching for fields only prod's context has).
  //
  // effectiveFeatures deliberately stays OUT of appContext — prod never puts
  // it there either (only dispatcherOptions.effectiveFeatures, consumed by
  // the command-dispatcher — see buildJobRunnerWithHook in entrypoint/
  // index.ts). Putting it here would reintroduce the exact test-vs-prod
  // drift #1232 fixed: a job/handler reading `ctx.effectiveFeatures` would
  // pass in tests and break in prod (kumiko-framework#1255).
  const appContext = {
    db: testDb.db,
    redis: testRedis.redis,
    searchAdapter,
    entityCache,
    registry,
    ...(options.masterKeyProvider ? { masterKeyProvider: options.masterKeyProvider } : {}),
    ...(fileProviderResolver ? { _fileProviderResolver: fileProviderResolver } : {}),
    ...(typeof options.extraContext === "function"
      ? options.extraContext({ registry, db: testDb.db, sseBroker, redis: testRedis.redis })
      : options.extraContext),
  };

  // Built before buildServer() so it can be merged into dispatcherOptions
  // (write handlers' `ctx.jobRunner.dispatch(...)` and the afterCommit
  // event-trigger hook both need it present at dispatcher-construction
  // time, same as the prod entrypoint's buildJobRunnerWithHook). Uses the
  // test-stack's own ephemeral redis — no separate `redisUrl` seam needed.
  //
  // context = appContext + tracer/meter, mirroring the prod entrypoint's
  // `contextWithObservability(options.context, observability)`.
  let jobRunner: JobRunner | undefined;
  if (options.jobs && registry.getAllJobs().size > 0) {
    jobRunner = createJobRunner({
      registry,
      context: { ...appContext, tracer: observability.tracer, meter: observability.meter },
      // The real REDIS_URL, not one rebuilt from `.options` — that lost
      // password/username/tls/path (pr-review kumiko-framework #1036/2).
      redisUrl: testRedis.redisUrl,
      ...(options.jobs.consumerLane !== undefined && { consumerLane: options.jobs.consumerLane }),
      ...(options.jobs.queueNamePrefix !== undefined && {
        queueNamePrefix: options.jobs.queueNamePrefix,
      }),
    });
  }

  // From here on, any throw must stop a jobRunner that was already created
  // above (createJobRunner() itself opens live BullMQ Queue/lock Redis
  // connections, .start() adds a live worker) — otherwise a failing
  // buildServer()/search-config/etc. leaks that connection and the test
  // process hangs on exit (pr-review kumiko-framework #1036/1).
  try {
    if (jobRunner) await jobRunner.start();

    // Auto-configure search for tenant 1 based on registry
    if (enabledHooks.includes("search")) {
      const searchableFields: string[] = [];
      for (const feature of options.features) {
        for (const [, entity] of Object.entries(feature.entities ?? {})) {
          for (const [fieldName, field] of Object.entries(entity.fields)) {
            if (field.type === "text" && field.searchable) {
              searchableFields.push(fieldName);
            }
            if (field.type === "embedded") {
              for (const [subName, subField] of Object.entries(field.schema)) {
                if (subField.searchable) {
                  searchableFields.push(`${fieldName}_${subName}`);
                }
              }
            }
          }
        }
      }

      if (options.searchConfig) {
        await searchAdapter.configure(options.searchConfig.tenantId, {
          searchableFields: options.searchConfig.searchableFields,
          rankingFields: options.searchConfig.rankingFields,
        });
      } else if (searchableFields.length > 0) {
        await searchAdapter.configure("00000000-0000-4000-8000-000000000001", {
          searchableFields,
          rankingFields: searchableFields,
        });
      }
    }

    const idempotency = createIdempotencyGuard(testRedis.redis, { ttlSeconds: 60 });
    const eventDedup = createEventDedup(testRedis.redis, { ttlSeconds: 60 });

    const server = buildServer({
      registry,
      context: appContext,
      jwtSecret,
      ...(options.observability ? { observability: options.observability } : {}),
      dispatcherOptions: {
        idempotency,
        ...(options.effectiveFeatures && { effectiveFeatures: options.effectiveFeatures }),
        ...(jobRunner && { jobRunner }),
      },
      eventDedup,
      sseBroker,
      // Tests drive the dispatcher via stack.eventDispatcher.runOnce() for
      // deterministic drains — no timer-induced flakiness. pollIntervalMs
      // stays short anyway in case a test opts into `.start()`. pgClient
      // plumbs through the LISTEN wake-up for tests that want to measure
      // post-commit latency (Sprint E.4).
      eventDispatcher: {
        pollIntervalMs: options.eventDispatcherPollIntervalMs ?? 50,
        pgClient: testDb.client as PgClient | undefined,
        systemConsumers: {
          sse: enabledHooks.includes("sse"),
          search: enabledHooks.includes("search"),
          accessInvalidation: enabledHooks.includes("sse"),
        },
      },
      // Default tests to no login rate-limiter so existing suites that loop
      // over logins don't hit a 429 after 10 attempts. Suites specifically
      // testing the limiter can override via authConfig.loginRateLimit.
      ...(options.authConfig
        ? {
            auth: {
              ...options.authConfig,
              ...(options.authConfig.loginRateLimit === undefined ? { loginRateLimit: null } : {}),
            },
          }
        : {}),
      ...(options.lifecycle ? { lifecycle: options.lifecycle } : {}),
      ...(options.rateLimit ? { rateLimit: options.rateLimit } : {}),
      ...(await (async () => {
        const baseAnon =
          typeof options.anonymousAccess === "function"
            ? options.anonymousAccess({
                registry,
                db: testDb.db,
                sseBroker,
                redis: testRedis.redis,
              })
            : options.anonymousAccess;
        const resolvedAnon = options.enrichAnonymousAccess
          ? await options.enrichAnonymousAccess(baseAnon, {
              registry,
              db: testDb.db,
            })
          : baseAnon;
        return resolvedAnon ? { anonymousAccess: resolvedAnon } : {};
      })()),
    });

    const eventDispatcher: EventDispatcher | undefined = server.eventDispatcher;

    // Pre-register consumer state rows so tests can call runOnce() directly
    // without a preceding explicit start(). Timer fires at pollIntervalMs=50
    // but passInFlight serialises concurrent passes — tests that drain via
    // runOnce() remain deterministic. Tests that specifically exercise the
    // timer loop call start() again (idempotent) after setup.
    if (eventDispatcher) await eventDispatcher.ensureRegistered();

    const http = createRequestHelper(server.app, server.jwt, {
      ...(options.authConfig?.sessionCreator !== undefined && {
        sessionCreator: options.authConfig.sessionCreator,
      }),
    });

    return {
      app: server.app,
      jwt: server.jwt,
      registry,
      db: testDb.db,
      redis: testRedis,
      search: searchAdapter,
      events,
      http,
      observability: server.observability,
      sseBroker,
      dispatcher: server.dispatcher,
      context: server.context,
      ...(eventDispatcher ? { eventDispatcher } : {}),
      ...(server.lifecycle ? { lifecycle: server.lifecycle } : {}),
      ...(jobRunner ? { jobRunner } : {}),
      cleanup: async () => {
        if (jobRunner) await jobRunner.stop();
        if (eventDispatcher) await eventDispatcher.stop();
        await server.observability.shutdown();
        await Promise.all([testDb.cleanup(), testRedis.cleanup()]);
      },
    };
  } catch (error) {
    // Best-effort — a broken jobRunner.stop()/cleanup() must never mask the
    // real setup failure below. testDb/testRedis are created before this
    // try even starts, so a throw anywhere in setup (buildServer,
    // search-config, ...) would otherwise leak their open pg/ioredis
    // sockets — the caller never gets the stack object back to call
    // cleanup() itself, which is the same "hangs on exit" failure class
    // this function exists to prevent.
    if (jobRunner) {
      try {
        await jobRunner.stop();
      } catch {
        // ignore — `error` is the one that matters
      }
    }
    try {
      await testDb.cleanup();
    } catch {
      // ignore — `error` is the one that matters
    }
    try {
      await testRedis.cleanup();
    } catch {
      // ignore — `error` is the one that matters
    }
    throw error;
  }
}