import type { Encodable } from "@automate.ax/codec"
import {
  branch,
  closed,
  collect,
  concurrent,
  correlate,
  debounce,
  dependentOn,
  each,
  failed,
  filter,
  funnel,
  gate,
  markSignificant,
  merge,
  onClose,
  onFailure,
  onSuccess,
  outcome,
  once,
  partition,
  race,
  rateLimit,
  serialize,
  succeeded,
  take,
  window,
  type AdmissionOptions,
  type BranchResult,
  type BranchSection,
  type BranchTail,
  type BranchTailReturn,
  type CollectionOptions,
  type ConcurrencyOptions,
  type CorrelationInput,
  type CorrelationOptions,
  type Duration,
  type FunnelOptions,
  type FunnelValue,
  type NonEmptyArray,
  type RateLimitOptions,
  type ValidBranchTail,
  type ValueInheritanceOptions,
  type WindowUntilOptions,
} from "./signal-operators"
import {
  registerSignalMethodImplementations,
  type GlobalSignal,
  type InferSignal,
  type InheritSignalCoordination,
  isSignal,
  type KeyedSignal,
  type Signal,
  type AutomationError,
  type SignalOutcome,
  type Synchronous,
} from "./signal-protocol"
import { correlationId, timestamp } from "./signal-resolution"
import { delay, timeout } from "../components/core/delay"

type AdditionalSignals = Signal | readonly [Signal, ...Signal[]]

type AdditionalSignalValue<TSignals extends AdditionalSignals> =
  TSignals extends Signal
    ? InferSignal<TSignals>
    : TSignals extends readonly Signal[]
      ? InferSignal<TSignals[number]>
      : never

/** Core methods implemented directly by the signal proxy. */
export interface SignalCoreMethods<T> {
  /**
   * Signals cannot be converted directly to strings. Import `t` from
   * `automate.ax` and use it as a tagged template to interpolate them.
   */
  toString(): never

  /** Transforms the receiver after its value materializes. */
  transform<TTransformer extends (value: T) => unknown>(
    transformer: Synchronous<TTransformer> extends never ? never : TTransformer,
  ): Signal<ReturnType<TTransformer>>
  transform<TOther, TTransformer extends (value: T, other: TOther) => unknown>(
    other: Signal<TOther>,
    transformer: Synchronous<TTransformer> extends never ? never : TTransformer,
  ): Signal<ReturnType<TTransformer>>
  transform<
    const TOthers extends readonly [unknown, ...unknown[]],
    TTransformer extends (value: T, ...others: TOthers) => unknown,
  >(
    others: {
      readonly [TIndex in keyof TOthers]: Signal<TOthers[TIndex]>
    },
    transformer: Synchronous<TTransformer> extends never ? never : TTransformer,
  ): Signal<ReturnType<TTransformer>>

  /** Uses one shared partition in each cross-context operator consuming it. */
  globally(): GlobalSignal<T>

  /** Returns an immutable view carrying a pure partition key. */
  keyBy(getKey: (value: T) => Encodable): KeyedSignal<T>
}

/** Source-first utilities adapted into fluent methods. */
export interface SignalMethods {
  /** Traverses a selected section and optional nested else-if sections. */
  branch<TTrueSection extends BranchSection, const TTail extends BranchTail>(
    this: Signal<boolean>,
    whenTrue: TTrueSection,
    ...tail: TTail & ValidBranchTail<ReturnType<TTrueSection>, TTail>
  ): BranchResult<ReturnType<TTrueSection>, BranchTailReturn<TTail>>

  /** Emits whether the receiver closes without a value. */
  closed(this: Signal): Signal<boolean>

  /** Derives a stable keyed identity from the receiver's durable boundary. */
  correlationId(this: Signal, perDeclaration?: boolean): KeyedSignal<string>

  /** Re-emits the receiver after a durable relative delay. */
  delay<TSignal extends Signal>(
    this: TSignal,
    duration: Duration,
  ): InheritSignalCoordination<TSignal, InferSignal<TSignal>>

  /** Preserves the receiver while attaching additional dependencies. */
  dependentOn<TSignal extends Signal>(
    this: TSignal,
    prerequisites: AdditionalSignals,
  ): InheritSignalCoordination<TSignal, InferSignal<TSignal>>

  /** Runs one durable section for every item in an array signal. */
  each<TItem, TResult>(
    this: Signal<readonly TItem[]>,
    section: (item: Signal<TItem>) => Signal<TResult>,
  ): Signal<TResult[]>
  each<TItem, TResult>(
    this: Signal<readonly TItem[]>,
    section: (item: Signal<TItem>) => Signal<TResult> | void,
  ): void

  /** Emits whether the receiver fails. */
  failed(this: Signal): Signal<boolean>

  /** Emits every occurrence from the receiver and additional streams. */
  merge<TSignal extends Signal, const TSignals extends AdditionalSignals>(
    this: TSignal,
    signals: TSignals,
  ): Signal<InferSignal<TSignal> | AdditionalSignalValue<TSignals>>

  /** Emits the receiver only when its value matches a predicate. */
  filter<TSignal extends Signal, TNarrowed extends InferSignal<TSignal>>(
    this: TSignal,
    predicate: (value: InferSignal<TSignal>) => value is TNarrowed,
  ): InheritSignalCoordination<TSignal, TNarrowed>
  filter<TSignal extends Signal>(
    this: TSignal,
    predicate: (value: InferSignal<TSignal>) => boolean,
  ): InheritSignalCoordination<TSignal, InferSignal<TSignal>>

  /** Emits the receiver only when a boolean signal opens its gate. */
  gate<TSignal extends Signal>(
    this: TSignal,
    condition: Signal<boolean>,
  ): InheritSignalCoordination<TSignal, InferSignal<TSignal>>

  /** Marks the containing causal run as significant when the receiver emits. */
  markSignificant<TSignal extends Signal>(this: TSignal): TSignal

  /** Emits null or traverses a section only when the receiver closes. */
  onClose<TSignal extends Signal>(this: TSignal): Signal<null>
  onClose<
    TSignal extends Signal,
    TSection extends (closure: Signal<null>, source: TSignal) => unknown,
  >(
    this: TSignal,
    section: Synchronous<TSection> extends never ? never : TSection,
  ): ReturnType<TSection>

  /** Emits failure details or traverses a section only when the receiver fails. */
  onFailure<TSignal extends Signal>(this: TSignal): Signal<AutomationError>
  onFailure<
    TSignal extends Signal,
    TSection extends (
      failure: Signal<AutomationError>,
      source: TSignal,
    ) => unknown,
  >(
    this: TSignal,
    section: Synchronous<TSection> extends never ? never : TSection,
  ): ReturnType<TSection>

  /** Emits the value or traverses a section only when the receiver succeeds. */
  onSuccess<TSignal extends Signal>(
    this: TSignal,
  ): InheritSignalCoordination<TSignal, InferSignal<TSignal>>
  onSuccess<
    TSignal extends Signal,
    TSection extends (
      value: InheritSignalCoordination<TSignal, InferSignal<TSignal>>,
      source: TSignal,
    ) => unknown,
  >(
    this: TSignal,
    section: Synchronous<TSection> extends never ? never : TSection,
  ): ReturnType<TSection>

  /** Converts every receiver terminal state into a tagged value. */
  outcome<TSignal extends Signal>(
    this: TSignal,
  ): Signal<SignalOutcome<InferSignal<TSignal>>>

  /** Splits the receiver into complementary matching and nonmatching paths. */
  partition<TSignal extends Signal, TNarrowed extends InferSignal<TSignal>>(
    this: TSignal,
    predicate: (value: InferSignal<TSignal>) => value is TNarrowed,
  ): readonly [
    matched: InheritSignalCoordination<TSignal, TNarrowed>,
    unmatched: InheritSignalCoordination<
      TSignal,
      Exclude<InferSignal<TSignal>, TNarrowed>
    >,
  ]
  partition<TSignal extends Signal>(
    this: TSignal,
    predicate: (value: InferSignal<TSignal>) => boolean,
  ): readonly [
    matched: InheritSignalCoordination<TSignal, InferSignal<TSignal>>,
    unmatched: InheritSignalCoordination<TSignal, InferSignal<TSignal>>,
  ]

  /** Selects the first causally related receiver or additional signal. */
  race<TSignal extends Signal, const TSignals extends AdditionalSignals>(
    this: TSignal,
    signals: TSignals,
  ): Signal<InferSignal<TSignal> | AdditionalSignalValue<TSignals>>

  /** Runs one structured work region at a time in each receiver partition. */
  serialize<
    TSignal extends KeyedSignal | GlobalSignal,
    TSection extends () => unknown,
  >(
    this: TSignal,
    section: Synchronous<TSection> extends never ? never : TSection,
  ): ReturnType<TSection>

  /** Runs up to `limit` structured work regions in each receiver partition. */
  concurrent<
    TSignal extends KeyedSignal | GlobalSignal,
    TSection extends () => unknown,
  >(
    this: TSignal,
    options: ConcurrencyOptions,
    section: Synchronous<TSection> extends never ? never : TSection,
  ): ReturnType<TSection>

  /** Emits whether the receiver succeeds with a value. */
  succeeded(this: Signal): Signal<boolean>

  /** Resolves when the receiver's durable occurrence emitted. */
  timestamp(this: Signal): Signal<Date>

  /** Fails unless the receiver settles before a durable deadline. */
  timeout<TSignal extends Signal>(
    this: TSignal,
    duration: Duration,
  ): InheritSignalCoordination<TSignal, InferSignal<TSignal>>

  /** Collects a fixed number of keyed or global receiver occurrences. */
  collect<TSignal extends KeyedSignal | GlobalSignal>(
    this: TSignal,
    count: number | Signal<number>,
    options?: CollectionOptions,
  ): InheritSignalCoordination<TSignal, NonEmptyArray<InferSignal<TSignal>>>

  /** Emits the first receiver occurrence in each partition. */
  once<TSignal extends KeyedSignal | GlobalSignal>(
    this: TSignal,
    options?: AdmissionOptions,
  ): InheritSignalCoordination<TSignal, InferSignal<TSignal>>

  /** Emits only the first `count` receiver occurrences in each partition. */
  take<TSignal extends KeyedSignal | GlobalSignal>(
    this: TSignal,
    count: number,
    options?: AdmissionOptions,
  ): InheritSignalCoordination<TSignal, InferSignal<TSignal>>

  /** Applies a rolling rate limit to each receiver partition. */
  rateLimit<TSignal extends KeyedSignal | GlobalSignal>(
    this: TSignal,
    options: RateLimitOptions,
  ): InheritSignalCoordination<TSignal, InferSignal<TSignal>>

  /** Matches a keyed receiver with one or more other keyed streams. */
  correlate<TSignal extends KeyedSignal>(
    this: TSignal,
    stream: CorrelationInput,
    options?: CorrelationOptions,
  ): Signal<null>
  correlate<TSignal extends KeyedSignal>(
    this: TSignal,
    streams: readonly [CorrelationInput, ...CorrelationInput[]],
    options?: CorrelationOptions,
  ): Signal<null>

  /** Emits the latest receiver occurrence after its partition stays quiet. */
  debounce<TSignal extends KeyedSignal | GlobalSignal>(
    this: TSignal,
    duration: Duration,
  ): InheritSignalCoordination<TSignal, InferSignal<TSignal>>

  /** Applies one durable burst timing and output policy to the receiver. */
  funnel<
    TSignal extends KeyedSignal | GlobalSignal,
    const TOptions extends FunnelOptions,
  >(
    this: TSignal,
    options: TOptions,
  ): InheritSignalCoordination<
    TSignal,
    FunnelValue<InferSignal<TSignal>, TOptions>
  >

  /** Collects receiver occurrences admitted by a fixed-duration window. */
  window<TSignal extends KeyedSignal | GlobalSignal>(
    this: TSignal,
    duration: Duration,
    options?: ValueInheritanceOptions,
  ): InheritSignalCoordination<TSignal, NonEmptyArray<InferSignal<TSignal>>>
  window<TSignal extends KeyedSignal | GlobalSignal>(
    this: TSignal,
    options: WindowUntilOptions,
  ): InheritSignalCoordination<TSignal, NonEmptyArray<InferSignal<TSignal>>>
}

/** Complete property namespace reserved for fluent signal methods. */
export type SignalReservedMethods = SignalCoreMethods<unknown> & SignalMethods

/** Fluent methods installed through the shared source-first adapter. */
export type SignalRegisteredMethods = Omit<
  SignalReservedMethods,
  keyof SignalCoreMethods<unknown>
>

registerSignalMethodImplementations({
  branch,
  closed,
  collect,
  concurrent,
  correlate: correlateFromSignal,
  correlationId,
  debounce,
  delay: delaySignal,
  dependentOn,
  each,
  failed,
  filter,
  funnel,
  gate,
  markSignificant,
  merge: mergeFromSignal,
  onClose,
  onFailure,
  onSuccess,
  outcome,
  once,
  partition,
  race: raceFromSignal,
  rateLimit,
  serialize,
  succeeded,
  take,
  timestamp,
  timeout,
  window,
})

/**
 * Prepends a keyed receiver to a correlation's remaining streams.
 *
 * @param signal - Fluent keyed receiver.
 * @param streams - Remaining keyed streams.
 * @param options - Optional temporal ordering.
 */
function correlateFromSignal(
  signal: KeyedSignal,
  streams:
    | CorrelationInput
    | readonly [CorrelationInput, ...CorrelationInput[]],
  options?: CorrelationOptions,
): Signal<null> {
  const remaining: readonly [CorrelationInput, ...CorrelationInput[]] =
    isCorrelationInputTuple(streams) ? streams : [streams]
  return correlate([signal, ...remaining], options)
}

/**
 * Narrows the fluent correlation's single-or-tuple input.
 *
 * @param value - Remaining fluent correlation input.
 */
function isCorrelationInputTuple(
  value: CorrelationInput | readonly [CorrelationInput, ...CorrelationInput[]],
): value is readonly [CorrelationInput, ...CorrelationInput[]] {
  return Array.isArray(value)
}

/**
 * Reorders the source-first fluent delay form for the duration-first utility.
 *
 * @param signal - Fluent source signal.
 * @param duration - Durable relative delay.
 */
function delaySignal<TSignal extends Signal>(
  signal: TSignal,
  duration: Duration,
): InheritSignalCoordination<TSignal, InferSignal<TSignal>> {
  return delay(duration, signal)
}

/**
 * Prepends a receiver to an independent stream union.
 *
 * @param signal - Fluent source signal.
 * @param signals - Remaining selection inputs.
 */
function mergeFromSignal(signal: Signal, signals: AdditionalSignals): Signal {
  const remaining: readonly [Signal, ...Signal[]] = isSignal(signals)
    ? [signals]
    : signals
  return merge([signal, ...remaining])
}

/**
 * Prepends a receiver to a causal race.
 *
 * @param signal - Fluent source signal.
 * @param signals - Remaining race inputs.
 */
function raceFromSignal(signal: Signal, signals: AdditionalSignals): Signal {
  const remaining: readonly [Signal, ...Signal[]] = isSignal(signals)
    ? [signals]
    : signals
  return race([signal, ...remaining])
}