<commit_before>import * as events from 'events';
import * as Promise from 'bluebird';
import {
  DataSourceSpec,
  Dependencies, DependentFetchFn, FetchFnWithDeps,
  TypedEventEmitter, TimingEvent
} from './model';
import mapObject from './map-object';

interface Stats {
  avg: number,
  count: number
}

type NormalizedStateSpec<T> = {
  [K in keyof T]: NormalizedSpec<T, T[K]>
}

type NormalizedSpec<T, V> = {
  paths: Array<NormalizedPath<T, V>>
}

type NormalizedPath<T, V> = {
  dependencies: Array<keyof T>,
  fn: FetchFnWithDeps<T, V>
}

type StatsMap = Map<Function, Stats>;

type OrchestratorEventEmitter<T> = TypedEventEmitter<{ timing: TimingEvent<T> }>;

class Orchestrator<T, U> {
  public readonly emitter: OrchestratorEventEmitter<T>;
  private readonly spec: NormalizedStateSpec<T>;
  private readonly stats: StatsMap;

  constructor(spec: DataSourceSpec<T, U>) {
    this.emitter = <OrchestratorEventEmitter<T>>new events.EventEmitter();
    this.spec = <NormalizedStateSpec<T>>mapObject(spec, (key : keyof T) => this.normalizeSpec(spec[key]));
    this.stats = new Map();
  }

  normalizeSpec(spec) {
    if (typeof spec === 'function') {
      const fn = spec;
      return {
        paths: [
          {
            dependencies: [],
            fn: fn.length ? ((_, cb?) => fn(cb)) : ((_: Dependencies<T>) => fn())
          }
        ]
      };
    }

    if (!(spec[0] instanceof Array)) {
      return {
        paths: [
          this.normalizeDependentFn(spec)
        ]
      };
    }

    return {
      paths: spec.map((s) => this.normalizeDependentFn(s))
    };
  }

  normalizeDependentFn<V>(depFn : DependentFetchFn<T, V>) : NormalizedPath<T, V> {
    const dependencies = <Array<keyof T>>(depFn.slice(0, -1));

    const fn = <FetchFnWithDeps<T, V>>(depFn[depFn.length - 1]);
    if (typeof(fn) !== 'function') {
      throw new Error(`Invalid dependent function specification ${JSON.stringify(depFn)}.`);
    }

    return { dependencies, fn };
  }

  resolve(key, cache, cb?) {
    if (!(key in cache)) {
      if (!(key in this.spec)) {
        return Promise.reject(`${key} is not defined in the data source`);
      }

      cache[key] = this.fetchResult(key, cache).then(result => {
        return cache[key] = result;
      }, err => {
        delete cache[key];
        throw err;
      });
    }

    return Promise.resolve(cache[key]).asCallback(cb);
  }

  fetchResult(key, cache) {
    const costedPaths = this.getPrioritizedPaths(key, cache);
    return costedPaths.slice(1).reduce((chain, { path }) => {
      return chain.catch(() => this.resolvePath(cache, key, path));
    }, this.resolvePath(cache, key, costedPaths[0].path));
  }

  resolvePath(cache, key, { dependencies, fn }) {
    return this.fetchDependentResult(cache, key, dependencies, fn);
  }

  fetchDependentResult(cache, name, dependencies, fn) {
    const requestStart = Date.now();
    let fetchStart = requestStart;

    const task = dependencies.length ?
      Promise.all(dependencies.map(dep => this.resolve(dep, cache))).then((values) => {
        const depsObj = dependencies.reduce((result, dep, i) => Object.assign(result, { [dep]: values[i] }), {});

        fetchStart = Date.now();
<commit_msg>Do not resolve infinite/unreachable paths<commit_after>import * as events from 'events';
import * as Promise from 'bluebird';
import {
  DataSourceSpec,
  Dependencies, DependentFetchFn, FetchFnWithDeps,
  TypedEventEmitter, TimingEvent
} from './model';
import mapObject from './map-object';

interface Stats {
  avg: number,
  count: number
}

type NormalizedStateSpec<T> = {
  [K in keyof T]: NormalizedSpec<T, T[K]>
}

type NormalizedSpec<T, V> = {
  paths: Array<NormalizedPath<T, V>>
}

type NormalizedPath<T, V> = {
  dependencies: Array<keyof T>,
  fn: FetchFnWithDeps<T, V>
}

type StatsMap = Map<Function, Stats>;

type OrchestratorEventEmitter<T> = TypedEventEmitter<{ timing: TimingEvent<T> }>;

class Orchestrator<T, U> {
  public readonly emitter: OrchestratorEventEmitter<T>;
  private readonly spec: NormalizedStateSpec<T>;
  private readonly stats: StatsMap;

  constructor(spec: DataSourceSpec<T, U>) {
    this.emitter = <OrchestratorEventEmitter<T>>new events.EventEmitter();
    this.spec = <NormalizedStateSpec<T>>mapObject(spec, (key : keyof T) => this.normalizeSpec(spec[key]));
    this.stats = new Map();
  }

  normalizeSpec(spec) {
    if (typeof spec === 'function') {
      const fn = spec;
      return {
        paths: [
          {
            dependencies: [],
            fn: fn.length ? ((_, cb?) => fn(cb)) : ((_: Dependencies<T>) => fn())
          }
        ]
      };
    }

    if (!(spec[0] instanceof Array)) {
      return {
        paths: [
          this.normalizeDependentFn(spec)
        ]
      };
    }

    return {
      paths: spec.map((s) => this.normalizeDependentFn(s))
    };
  }

  normalizeDependentFn<V>(depFn : DependentFetchFn<T, V>) : NormalizedPath<T, V> {
    const dependencies = <Array<keyof T>>(depFn.slice(0, -1));

    const fn = <FetchFnWithDeps<T, V>>(depFn[depFn.length - 1]);
    if (typeof(fn) !== 'function') {
      throw new Error(`Invalid dependent function specification ${JSON.stringify(depFn)}.`);
    }

    return { dependencies, fn };
  }

  resolve(key, cache, cb?) {
    if (!(key in cache)) {
      if (!(key in this.spec)) {
        return Promise.reject(`${key} is not defined in the data source`);
      }

      cache[key] = this.fetchResult(key, cache).then(result => {
        return cache[key] = result;
      }, err => {
        delete cache[key];
        throw err;
      });
    }

    return Promise.resolve(cache[key]).asCallback(cb);
  }

  fetchResult(key, cache) {
    return this.getPrioritizedPaths(key, cache).reduce((chain, { cost, path }) => {
      const execution = cost === Number.POSITIVE_INFINITY
        ? () => Promise.reject(`Cannot resolve ${key}`)
        : () => this.resolvePath(cache, key, path);

      if (chain) {
        return (chain as Promise).catch(execution);
      } else {
        return execution();
      }
    }, null);
  }

  resolvePath(cache, key, { dependencies, fn }) {
    return this.fetchDependentResult(cache, key, dependencies, fn);
  }

  fetchDependentResult(cache, name, dependencies, fn) {
    const requestStart = Date.now();
    let fetchStart = requestStart;

    const task = dependencies.length ?
      Promise.all(dependencies.map(dep => this.resolve(dep, cache))).then((values) => {
        const depsObj = dependencies.reduce((result, dep, i) => Object.assign(result, { [dep]: values[i] }), {});

        fetchStart = Date.now();

{'fraction_non_alphanumeric': 0.15279878971255673, 'fraction_numerical': 0.0015128593040847202, 'mean_word_length': 3.111318407960199, 'pattern_counts': {'":': 0, '<': 59, '<?xml version=': 0, '>': 82, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4968437', 'n_tokens_mistral': 2155, 'n_tokens_neox': 2106, 'n_words': 478}