import type { QueryClient } from '@tanstack/react-query'
import type { z } from 'zod'
import { runPipeline } from './pipeline/index'
import { fetchStage } from './pipeline/stages/fetch'
import { PluginBus } from './plugins/bus'
import { createConsoleLogPlugin } from './plugins/console-log'
import { engineRegistry } from './registry'
import { createStatsStore, incrementStat, type StatsStore } from './stats-store'
import { createEngineStore, type EngineStore } from './store'
import type {
  FetchContext,
  InferDescriptorResult,
  NativeCacheConfig,
  SourceDescriptor,
  TriggerOptions,
  TriggerResult,
  WorkflowPlugin,
  ZodValidationMode,
} from './types'

interface EngineConfig {
  queryClient: QueryClient
  /**
   * Required unique identifier for this engine instance.
   * Flows automatically to:
   * - Engine registry key (duplicate names throw at construction time)
   * - DevTools label (unless devtools.name overrides it)
   * - Native cache storeKey prefix: `name:descriptorKey`
   * - Console log prefix: `[name]` (when logging is enabled and prefix is not set)
   */
  name: string
  /** Pass true or an options object to enable Zustand DevTools integration */
  devtools?: boolean | { name?: string }
  /** Pass true to console.log all state transitions. Pass an object for finer control. */
  logging?: boolean | { prefix?: string; logLevel?: 'log' | 'debug' | 'info' }
  /**
   * Engine-level native cache configuration applied to every registered descriptor.
   * Overrides the registry-level default (set via `engineRegistry.setDefaultNativeCache`).
   * Set to `false` to explicitly opt this engine out of the global default.
   * Individual descriptors can still override or disable via their own nativeCache field.
   */
  nativeCache?: NativeCacheConfig | false
  /**
   * Default throttle window in milliseconds applied to every trigger when the
   * descriptor does not set its own throttle value. Defaults to 200 ms.
   * Set to 0 to disable the default throttle engine-wide (useful in tests).
   * Descriptor-level throttle always takes precedence over this value.
   */
  defaultThrottle?: number
  /**
   * Global default for how Zod schema mismatches are handled.
   * Individual descriptors can override via their own zodMode field. Defaults to 'warn'.
   * - 'warn': partial data + per-field warnings + partialSuccess (default)
   * - 'error': pipeline fails at the validate phase
   * - 'passthrough': raw data passes through + warnings surfaced + partialSuccess
   */
  zodMode?: ZodValidationMode
}

interface DebouncePending {
  timer: ReturnType<typeof setTimeout>
  resolvers: Array<{
    resolve: (r: TriggerResult<unknown>) => void
    reject: (e: unknown) => void
  }>
  options: TriggerOptions
}

interface ThrottlePending {
  promise: Promise<TriggerResult<unknown>>
  expiresAt: number
}

export class WorkflowEngine<
  TRegistry extends Record<string, SourceDescriptor<z.ZodTypeAny>> = Record<
    never,
    never
  >,
> {
  readonly store: EngineStore
  readonly statsStore: StatsStore
  readonly name: string
  private readonly queryClient: QueryClient
  private readonly registry = new Map<string, SourceDescriptor<z.ZodTypeAny>>()
  private readonly bus = new PluginBus()
  private readonly debouncePending = new Map<string, DebouncePending>()
  private readonly throttlePending = new Map<string, ThrottlePending>()
  private readonly inFlight = new Map<string, Promise<TriggerResult<unknown>>>()
  private readonly nativeCache: NativeCacheConfig | undefined
  private readonly zodMode: ZodValidationMode
  private readonly defaultThrottle: number
  // Per-section params fingerprints for split descriptors.
  // Maps descriptorKey → (splitKey → JSON.stringify(params[splitKey])).
  // Used by the auto-derived shouldFetch to detect per-section payload changes.
  private readonly splitSectionParams = new Map<string, Map<string, string>>()

  constructor(config: EngineConfig) {
    this.queryClient = config.queryClient
    this.name = config.name
    // Resolve effective native cache:
    // false → explicit opt-out (no cache even if registry has a default)
    // NativeCacheConfig → engine-level override (takes precedence over registry default)
    // undefined → inherit from registry default if one is set
    this.nativeCache =
      config.nativeCache === false
        ? undefined
        : (config.nativeCache ??
          engineRegistry._resolveDefaultNativeCache(config.name))
    this.zodMode = config.zodMode ?? 'warn'
    this.defaultThrottle = config.defaultThrottle ?? 200
    this.statsStore = createStatsStore()
    this.store = createEngineStore({
      devtools: config.devtools,
      name: config.name,
    })

    if (config.logging) {
      const loggingOpts =
        typeof config.logging === 'object' ? config.logging : {}
      const prefix =
        loggingOpts.prefix ?? (config.name ? `[${config.name}]` : undefined)
      this.bus.addGlobal(createConsoleLogPlugin({ ...loggingOpts, prefix }))
    }

    engineRegistry._register(this)
  }

  use(plugin: WorkflowPlugin): this {
    this.bus.addGlobal(plugin)
    return this
  }

  /**
   * When split is configured and shouldFetch is not provided, auto-derive
   * shouldFetch from the split slice states. The convention: the top-level keys
   * of options.params that match split record keys are the requested sections.
   * Fetch is skipped if every requested section already has data in its slice.
   * Explicit user-supplied shouldFetch always takes precedence.
   */
  private applyDefaults(
    descriptor: SourceDescriptor<z.ZodTypeAny>,
  ): SourceDescriptor<z.ZodTypeAny> {
    if (!descriptor.split || descriptor.shouldFetch) return descriptor
    const splitKeys = new Set(Object.keys(descriptor.split))
    const descriptorKey = descriptor.key
    return {
      ...descriptor,
      shouldFetch: (opts) => {
        const requested = Object.keys(opts.params ?? {}).filter((k) =>
          splitKeys.has(k),
        )
        if (requested.length === 0) return true
        const sectionMap = this.splitSectionParams.get(descriptorKey)
        return requested.some((k) => {
          // Section not yet loaded → fetch
          if ((this.store.getState().slices[k]?.data ?? null) == null)
            return true
          // Section never fetched with tracked params → fetch
          if (!sectionMap?.has(k)) return true
          // Section payload changed → fetch
          const currentParams = JSON.stringify(opts.params?.[k] ?? null)
          return sectionMap.get(k) !== currentParams
        })
      },
    }
  }

  register<K extends string, S extends z.ZodTypeAny>(
    descriptor: SourceDescriptor<S> & { key: K },
  ): WorkflowEngine<TRegistry & Record<K, SourceDescriptor<S>>> {
    const processed = this.applyDefaults(
      descriptor as SourceDescriptor<z.ZodTypeAny>,
    )
    this.registry.set(descriptor.key, processed)
    this.store.getState().actions.initSlice(descriptor.key)
    this.statsStore.setState((s) => ({
      [descriptor.key]: s[descriptor.key] ?? {
        triggerCount: 0,
        throttledCount: 0,
        dedupedCount: 0,
        shouldFetchSkipCount: 0,
        cacheHitCount: 0,
        apiFetchCount: 0,
        errorCount: 0,
      },
    }))

    if (descriptor.split) {
      for (const splitKey of Object.keys(descriptor.split)) {
        this.store.getState().actions.initSlice(splitKey)
      }
    }

    return this as unknown as WorkflowEngine<
      TRegistry & Record<K, SourceDescriptor<S>>
    >
  }

  async trigger<K extends keyof TRegistry & string>(
    key: K,
    options?: TriggerOptions,
  ): Promise<TriggerResult<InferDescriptorResult<TRegistry[K]>>>
  async trigger(
    key: string,
    options?: TriggerOptions,
  ): Promise<TriggerResult<unknown>>
  async trigger(
    key: string,
    options: TriggerOptions = {},
  ): Promise<TriggerResult<unknown>> {
    const descriptor = this.registry.get(key)
    if (!descriptor)
      throw new Error(`No descriptor registered for key "${key}"`)

    incrementStat(this.statsStore, key, 'triggerCount')

    if (descriptor.debounce) {
      return this.triggerDebounced<T>(key, options, descriptor.debounce)
    }

    // Default throttle: 200 ms per key+params pair (engine-level defaultThrottle).
    // Descriptor-level throttle overrides; throttle: 0 disables entirely.
    // force:true always bypasses throttle — explicit refresh must always execute.
    const throttleMs = descriptor.throttle ?? this.defaultThrottle
    if (throttleMs > 0 && !options.force) {
      return this.triggerThrottled<T>(key, options, throttleMs)
    }

    return this.executeTrigger<T>(key, options)
  }

  private triggerDebounced<T>(
    key: string,
    options: TriggerOptions,
    delayMs: number,
  ): Promise<TriggerResult<T>> {
    return new Promise((resolve, reject) => {
      const existing = this.debouncePending.get(key)
      if (existing) {
        clearTimeout(existing.timer)
        existing.resolvers.push({
          resolve: resolve as (r: TriggerResult<unknown>) => void,
          reject,
        })
        existing.options = options
      } else {
        const pending: DebouncePending = {
          timer: undefined as unknown as ReturnType<typeof setTimeout>,
          resolvers: [
            {
              resolve: resolve as (r: TriggerResult<unknown>) => void,
              reject,
            },
          ],
          options,
        }
        this.debouncePending.set(key, pending)
        pending.timer = setTimeout(async () => {
          const p = this.debouncePending.get(key)
          this.debouncePending.delete(key)
          if (!p) return
          try {
            const result = await this.executeTrigger(key, p.options)
            for (const { resolve: res } of p.resolvers) res(result)
          } catch (err) {
            for (const { reject: rej } of p.resolvers) rej(err)
          }
        }, delayMs)
      }
    })
  }

  private triggerThrottled<T>(
    key: string,
    options: TriggerOptions,
    windowMs: number,
  ): Promise<TriggerResult<T>> {
    // Key by key+params so different payloads are never throttled against each
    // other — only identical payload calls share the throttle window.
    const throttleKey = this.inFlightKey(key, options)
    const existing = this.throttlePending.get(throttleKey)
    if (existing && Date.now() < existing.expiresAt) {
      incrementStat(this.statsStore, key, 'throttledCount')
      return existing.promise as Promise<TriggerResult<T>>
    }

    const promise = this.executeTrigger<T>(key, options)
    this.throttlePending.set(throttleKey, {
      promise: promise as Promise<TriggerResult<unknown>>,
      expiresAt: Date.now() + windowMs,
    })
    return promise
  }

  private resolveNativeCache(descriptor: SourceDescriptor<z.ZodTypeAny>): {
    retrieve: (key: string) => Promise<unknown>
    upsert: (key: string, data: unknown) => Promise<void>
    storeKey: string
  } | null {
    if (descriptor.nativeCache === false) return null
    const global = this.nativeCache
    const local = descriptor.nativeCache
    if (local?.enabled === false) return null
    const retrieve = local?.retrieve ?? global?.retrieve
    const upsert = local?.upsert ?? global?.upsert
    if (!retrieve || !upsert) return null
    const baseKey = local?.storeKey ?? descriptor.key
    const storeKey = this.name ? `${this.name}:${baseKey}` : baseKey
    return { retrieve, upsert, storeKey }
  }

  // Stable sorted-JSON fingerprint of options.params. Empty string = no params.
  private paramsKey(options: TriggerOptions): string {
    const p = options.params
    if (!p || Object.keys(p).length === 0) return ''
    return JSON.stringify(
      Object.fromEntries(
        Object.entries(p).sort(([a], [b]) => a.localeCompare(b)),
      ),
    )
  }

  // Build a stable in-flight key that includes params so different-payload
  // triggers are NOT deduplicated against each other, but identical concurrent
  // calls (same key + same params) still share one pipeline execution.
  private inFlightKey(key: string, options: TriggerOptions): string {
    const pk = this.paramsKey(options)
    return pk ? `${key}:${pk}` : key
  }

  // Sync entry point — returns the same Promise to every concurrent caller for
  // the same key+params so the pipeline only runs once per in-flight execution.
  private executeTrigger<T>(
    key: string,
    options: TriggerOptions,
  ): Promise<TriggerResult<T>> {
    const ifKey = this.inFlightKey(key, options)
    if (!options.force) {
      const existing = this.inFlight.get(ifKey)
      if (existing) {
        incrementStat(this.statsStore, key, 'dedupedCount')
        return existing as Promise<TriggerResult<T>>
      }
    }

    const promise = this._executeCore<T>(key, options).finally(() => {
      this.inFlight.delete(ifKey)
    })
    this.inFlight.set(ifKey, promise as Promise<TriggerResult<unknown>>)
    return promise
  }

  private async _executeCore<T>(
    key: string,
    options: TriggerOptions,
  ): Promise<TriggerResult<T>> {
    const descriptor = this.registry.get(key)
    if (!descriptor)
      throw new Error(`No descriptor registered for key "${key}"`)

    // shouldFetch predicate — short-circuit before loading state and cache/API.
    // Bypassed by force:true so an explicit refresh always gets through.
    if (descriptor.shouldFetch && !options.force) {
      const existing = this.store.getState().slices[key]?.data ?? null
      if (!descriptor.shouldFetch(options, existing)) {
        incrementStat(this.statsStore, key, 'shouldFetchSkipCount')
        // For split descriptors, return the merged split slice data.
        // For regular descriptors, return the slice data directly.
        const data = descriptor.split
          ? Object.fromEntries(
              Object.keys(descriptor.split).map((k) => [
                k,
                this.store.getState().slices[k]?.data ?? null,
              ]),
            )
          : existing
        return { status: 'success', data: data as T, warnings: [] }
      }
    }

    this.store
      .getState()
      .actions.setSlice(key, { isLoading: true, isFetching: true })

    const perKey = descriptor.plugins ?? []

    // Native cache restore: serve from cache unless force:true or the params changed.
    // "Params changed" means the caller is asking for a different payload than what
    // was last written to the cache (e.g. a new page in a paginated list).
    const nc = this.resolveNativeCache(descriptor)
    const currentParamsKey = this.paramsKey(options)
    if (nc && !options.force) {
      const slice = this.store.getState().slices[key]
      const paramsUnchanged =
        !slice ||
        slice.lastParamsKey === null ||
        slice.lastParamsKey === currentParamsKey
      if (paramsUnchanged) {
        try {
          const cached = await nc.retrieve(nc.storeKey)
          if (cached != null) {
            incrementStat(this.statsStore, key, 'cacheHitCount')
            const now = Date.now()
            if (descriptor.split) {
              for (const splitKey of Object.keys(descriptor.split)) {
                // The cached value is the parent slice data, which is always
                // keyed by split record key (from combinedData in runPipeline).
                // splitEntry.path is the extraction path from the *API response*,
                // not from the cached object — always use splitKey here.
                const splitData = (cached as Record<string, unknown>)?.[
                  splitKey
                ]
                this.store.getState().actions.setSlice(splitKey, {
                  data: splitData ?? null,
                  status: 'success',
                  isLoading: false,
                  isFetching: false,
                  isError: false,
                  error: null,
                  warnings: [],
                  lastUpdated: now,
                  lastParamsKey: currentParamsKey,
                })
                await this.bus.runObserver('afterWrite', perKey, {
                  key: splitKey,
                  data: splitData ?? null,
                  warnings: [],
                  status: 'success' as const,
                  syncToRedux: descriptor.syncToRedux,
                  source: 'native-cache' as const,
                })
              }
              // Populate per-section fingerprints so the auto-derived shouldFetch
              // can give an instant skip on subsequent same-session calls, without
              // going through the native cache again.
              if (options.params) {
                if (!this.splitSectionParams.has(key)) {
                  this.splitSectionParams.set(key, new Map())
                }
                const sectionMap = this.splitSectionParams.get(key)!
                for (const sectionKey of Object.keys(descriptor.split)) {
                  if (sectionKey in options.params) {
                    sectionMap.set(
                      sectionKey,
                      JSON.stringify(options.params[sectionKey] ?? null),
                    )
                  }
                }
              }
            }
            this.store.getState().actions.setSlice(key, {
              data: cached,
              status: 'success',
              isLoading: false,
              isFetching: false,
              isError: false,
              error: null,
              warnings: [],
              lastUpdated: now,
              lastParamsKey: currentParamsKey,
            })
            await this.bus.runObserver('afterWrite', perKey, {
              key,
              data: cached,
              warnings: [],
              status: 'success' as const,
              syncToRedux: descriptor.syncToRedux,
              source: 'native-cache' as const,
            })
            return { status: 'success', data: cached as T, warnings: [] }
          }
        } catch {
          // native cache error — fall through to API
        }
      }
    }

    let fetchCtx: FetchContext = {
      key,
      options,
      queryClient: this.queryClient,
      meta: {},
    }
    fetchCtx = await this.bus.run('beforeFetch', perKey, fetchCtx)

    incrementStat(this.statsStore, key, 'apiFetchCount')

    let responseCtx: Awaited<ReturnType<typeof fetchStage>>
    try {
      responseCtx = await fetchStage(
        descriptor,
        fetchCtx.options,
        this.queryClient,
      )
    } catch (err) {
      const cause = err instanceof Error ? err : new Error(String(err))
      const triggerError = { phase: 'fetch' as const, cause }
      incrementStat(this.statsStore, key, 'errorCount')
      this.store.getState().actions.setSlice(key, {
        isLoading: false,
        isFetching: false,
        isError: true,
        error: triggerError,
        status: 'error',
      })
      await this.bus.runObserver('onError', perKey, {
        key,
        error: triggerError,
      })
      return {
        status: 'error',
        data: null,
        error: triggerError,
      } as TriggerResult<T>
    }

    const afterFetchCtx = await this.bus.run('afterFetch', perKey, responseCtx)

    const result = await (runPipeline(
      descriptor,
      afterFetchCtx.response,
      this.store,
      this.bus,
      this.zodMode,
      currentParamsKey,
    ) as Promise<TriggerResult<T>>)

    if (result.status === 'error') {
      incrementStat(this.statsStore, key, 'errorCount')
    }

    // Record per-section params fingerprints for split descriptors so that
    // the auto-derived shouldFetch can detect per-section payload changes.
    if (descriptor.split && result.status !== 'error' && options.params) {
      if (!this.splitSectionParams.has(key)) {
        this.splitSectionParams.set(key, new Map())
      }
      const sectionMap = this.splitSectionParams.get(key)!
      for (const sectionKey of Object.keys(descriptor.split)) {
        if (sectionKey in options.params) {
          sectionMap.set(
            sectionKey,
            JSON.stringify(options.params[sectionKey] ?? null),
          )
        }
      }
    }

    // Upsert final post-merge Zustand data to native cache after successful pipeline
    if (nc && result.status !== 'error') {
      const finalData = this.store.getState().slices[key]?.data
      if (finalData != null) {
        nc.upsert(nc.storeKey, finalData) // fire-and-forget
      }
    }

    return result
  }

  async wrapQuery<K extends keyof TRegistry & string>(
    key: K,
    data: unknown,
  ): Promise<TriggerResult<InferDescriptorResult<TRegistry[K]>>>
  async wrapQuery(key: string, data: unknown): Promise<TriggerResult<unknown>>
  async wrapQuery(key: string, data: unknown): Promise<TriggerResult<unknown>> {
    const descriptor = this.registry.get(key)
    if (!descriptor)
      throw new Error(`No descriptor registered for key "${key}"`)

    return runPipeline(
      descriptor,
      data,
      this.store,
      this.bus,
      this.zodMode,
    ) as Promise<TriggerResult<T>>
  }

  async updateSlice(
    key: string,
    updater: (prev: unknown) => unknown,
  ): Promise<void> {
    this.store.getState().actions.patchSlice(key, (slice) => {
      slice.data = updater(slice.data)
      slice.lastUpdated = Date.now()
    })
    const descriptor = this.registry.get(key)
    if (descriptor) {
      const data = this.store.getState().slices[key]?.data ?? null
      const perKey = descriptor.plugins ?? []
      await this.bus.runObserver('afterWrite', perKey, {
        key,
        data,
        warnings: [],
        status: 'success' as const,
        syncToRedux: descriptor.syncToRedux,
      })
    }
  }

  async batchUpdate(
    updates: Partial<
      Record<keyof TRegistry & string, (prev: unknown) => unknown>
    >,
  ): Promise<void> {
    await Promise.all(
      Object.entries(updates).map(([key, updater]) =>
        updater
          ? this.updateSlice(key as keyof TRegistry & string, updater)
          : Promise.resolve(),
      ),
    )
  }
}