import invariant from 'tiny-invariant'
import { groupBy } from '$lib/functions/common/array'
import { tuple } from '$lib/functions/common/tuple'
import { normalizeCaseIndependentName } from '$lib/functions/felderaRelation'
import type {
  ConnectorHealth,
  ControllerStatus,
  InputEndpointMetrics,
  InputEndpointStatus,
  OutputEndpointMetrics,
  OutputEndpointStatus
} from '$lib/services/manager'
import type { GlobalMetricsTimestamp, TimeSeriesEntry } from '$lib/types/pipelineManager'
import { discreteDerivative } from './common/math'

export type AggregatedMetrics<M, EndpointStatus = {}> = {
  aggregate: { metrics: M }
  connectors: ({
    endpointName: string
    metrics: M
  } & EndpointStatus)[]
}

export type AggregatedInputEndpointMetrics = AggregatedMetrics<
  InputEndpointMetrics,
  Pick<InputEndpointStatus, 'paused' | 'barrier' | 'health' | 'fatal_error'> & {
    io_active: boolean
    transaction_phase?: 'started' | 'committed'
  }
>

export type AggregatedOutputEndpointMetrics = AggregatedMetrics<
  OutputEndpointMetrics,
  Pick<OutputEndpointStatus, 'health' | 'fatal_error'> & { io_active: boolean }
>

export const emptyPipelineMetrics = {
  tables: new Map<string, AggregatedInputEndpointMetrics>(),
  views: new Map<string, AggregatedOutputEndpointMetrics>(),
  inputs: [] as InputEndpointStatus[],
  outputs: [] as OutputEndpointStatus[],
  global: {
    transaction_status: 'NoTransaction',
    transaction_initiators: { initiated_by_connectors: {} }
  } as GlobalMetricsTimestamp
}

export type PipelineMetrics = typeof emptyPipelineMetrics & { lastTimestamp?: number }

const addZeroMetrics = (previous: PipelineMetrics) => ({
  ...previous,
  tables: new Map(),
  views: new Map()
})

export const accumulatePipelineMetrics =
  (newTimestamp: number) =>
  (
    oldData: PipelineMetrics | undefined,
    { status: newData }: { status: ControllerStatus | null }
  ): PipelineMetrics | undefined => {
    if (!newData) {
      return oldData ? addZeroMetrics(oldData) : oldData
    }
    const globalWithTimestamp = {
      ...newData.global_metrics,
      timeMs: newTimestamp
    }
    const initiatedByConnectors =
      newData.global_metrics.transaction_initiators.initiated_by_connectors
    return {
      lastTimestamp: oldData?.lastTimestamp,
      inputs: newData.inputs,
      outputs: newData.outputs,
      tables: new Map(
        groupBy(newData.inputs, (i) => normalizeCaseIndependentName({ name: i.config.stream })).map(
          ([relationName, endpoints]) => {
            const oldRelation = oldData?.tables.get(relationName)
            const connectors = endpoints.map((cur) => {
              const prev = oldRelation?.connectors.find((c) => c.endpointName === cur.endpoint_name)
              const connectorPhase = ((phase) =>
                phase && (phase === ('started' as const) || phase === ('committed' as const))
                  ? phase
                  : undefined)(initiatedByConnectors[cur.endpoint_name]?.phase?.toLowerCase())
              return {
                endpointName: cur.endpoint_name,
                metrics: cur.metrics,
                paused: cur.paused,
                barrier: cur.barrier,
                io_active:
                  prev !== undefined && cur.metrics.total_records > prev.metrics.total_records,
                transaction_phase: connectorPhase,
                health: cur.health,
                fatal_error: cur.fatal_error
              }
            })
            const metrics: AggregatedInputEndpointMetrics = {
              connectors,
              aggregate: {
                metrics: endpoints.reduce(
                  (acc: InputEndpointMetrics, cur) => {
                    const metrics = cur.metrics
                    return {
                      total_bytes: acc.total_bytes + metrics.total_bytes,
                      total_records: acc.total_records + metrics.total_records,
                      buffered_records: acc.buffered_records + metrics.buffered_records,
                      num_transport_errors: acc.num_transport_errors + metrics.num_transport_errors,
                      num_parse_errors: acc.num_parse_errors + metrics.num_parse_errors,
                      end_of_input: acc.end_of_input && metrics.end_of_input,
                      buffered_bytes: acc.buffered_bytes + metrics.buffered_bytes
                    }
                  },
                  {
                    total_bytes: 0,
                    total_records: 0,
                    buffered_bytes: 0,
                    buffered_records: 0,
                    num_transport_errors: 0,
                    num_parse_errors: 0,
                    end_of_input: true
                  }
                )
              }
            }
            return tuple(relationName, metrics)
          }
        )
      ),
      views: new Map(
        groupBy(newData.outputs, (i) =>
          normalizeCaseIndependentName({ name: i.config.stream })
        ).map(([relationName, endpoints]) => {
          const oldRelation = oldData?.views.get(relationName)
          const metrics: AggregatedOutputEndpointMetrics = {
            connectors: endpoints.map((cur) => {
              const prev = oldRelation?.connectors.find((c) => c.endpointName === cur.endpoint_name)
              return {
                endpointName: cur.endpoint_name,
                metrics: cur.metrics,
                io_active:
                  prev !== undefined &&
                  cur.metrics.transmitted_records > prev.metrics.transmitted_records,
                health: cur.health,
                fatal_error: cur.fatal_error
              }
            }),
            aggregate: {
              metrics: endpoints.reduce(
                (acc: OutputEndpointMetrics, cur) => {
                  const metrics = cur.metrics
                  return {
                    buffered_batches: acc.buffered_batches + metrics.buffered_batches,
                    buffered_records: acc.buffered_records + metrics.buffered_records,
                    num_encode_errors: acc.num_encode_errors + metrics.num_encode_errors,
                    num_transport_errors: acc.num_transport_errors + metrics.num_transport_errors,
                    total_processed_input_records:
                      acc.total_processed_input_records + metrics.total_processed_input_records,
                    transmitted_bytes: acc.transmitted_bytes + metrics.transmitted_bytes,
                    transmitted_records: acc.transmitted_records + metrics.transmitted_records,
                    total_processed_steps:
                      acc.total_processed_steps + metrics.total_processed_steps,
                    queued_batches: acc.queued_batches + metrics.queued_batches,
                    queued_records: acc.queued_records + metrics.queued_records,
                    memory: acc.memory + metrics.memory
                  }
                },
                {
                  buffered_batches: 0,
                  buffered_records: 0,
                  num_encode_errors: 0,
                  num_transport_errors: 0,
                  total_processed_input_records: 0,
                  transmitted_bytes: 0,
                  transmitted_records: 0,
                  total_processed_steps: 0,
                  queued_batches: 0,
                  queued_records: 0,
                  memory: 0
                }
              )
            }
          }
          return tuple(relationName, metrics)
        })
      ),
      global: globalWithTimestamp
    }
  }

/**
 * @returns Time series of throughput with smoothing window over 3 data intervals
 */
export const calcPipelineThroughput = (metrics: TimeSeriesEntry[]) => {
  const series = discreteDerivative(metrics, (n1, n0) => {
    return {
      name: n1.t.toFixed(),
      value: tuple(n1.t.toNumber(), n1.r.minus(n0.r).toNumber())
    }
  })

  const avgN = Math.min(Math.ceil(series.length / 5), 4)
  const valueMax = series.length
    ? series
        .slice()
        .sort((a, b) => a.value[1] - b.value[1])
        .slice(-avgN)
        .reduce((acc, cur) => acc + cur.value[1], 0) / avgN
    : 0
  const yMaxStep = 10 ** Math.ceil(Math.log10(valueMax)) / 5
  const yMax = valueMax !== 0 ? Math.ceil((valueMax * 1.25) / yMaxStep) * yMaxStep : 100
  const yMin = 0
  const current = series.at(-1)?.value?.[1] ?? 0
  const average = series.length
    ? series.reduce((acc, cur) => cur.value[1] + acc, 0) / series.length
    : 0
  return { series, current, average, yMin, yMax }
}
