<commit_before>import { AsyncSubject } from 'rxjs/AsyncSubject';

import Dexie from 'dexie';
import { asyncMap } from './promise-extras';
import { User, ChannelBase, Message } from './models/api-shapes';
import { Pair } from './utils';
import { Api } from './models/slack-api';

import './standard-operators';
import 'request-idle-polyfill';
import { MessageKey } from './store';

//const d = require('debug')('trickline:dexie-data-model');
const dn = require('debug')('trickline-noisy:dexie-data-model');

const VERSION = 1;

export interface DeferredPutItem<T, TKey> {
  item: T;
  key: TKey;
  completion: AsyncSubject<void>;
}

export interface DeferredGetItem<T, TKey> {
  key: TKey;
  completion: AsyncSubject<T>;
}

declare module 'dexie' {
  module Dexie {
    interface Table<T, Key> {
      deferredPut: ((item: T, key: Key) => Promise<void>);
      deferredGet: ((key: Key, database: Dexie) => Promise<T & { token: string }>);
      idlePutHandle: NodeJS.Timer | null;
      idleGetHandle: number | null;
      deferredPuts: DeferredPutItem<T & Api, Key>[];
      deferredGets: DeferredGetItem<T, Key>[];
    }
  }
}

function deferredPut<T, Key>(this: Dexie.Table<T, Key>, item: T & Api, key: Key): Promise<void> {
  let newItem = { item, key, completion: new AsyncSubject<void>() };
  newItem.item = Object.assign({}, newItem.item);

  let createIdle = () => window.requestIdleCallback(async deadline => {
    while (deadline.timeRemaining() > 5/*ms*/) {
      // TODO: This would be cooler if it recognized duplicate IDs and threw out the older
      // one (i.e. you update channel.topic and channel.name in the same batch, so we really
      // only need to write the newer one)
      let itemsToAdd = this.deferredPuts.slice(0, 128).map(x => {
        if (x.item.api) {
          x.item.token = x.item.api.token();
          x.item.api = null;
        }

        return x;
      });

      if (itemsToAdd.length < 1) break;

      await this.bulkPut(itemsToAdd.map(x => x.item))
        .then(
          () => itemsToAdd.forEach(x => { dn(`Actually wrote ${JSON.stringify(x.key)}!`); x.completion.next(undefined); x.completion.complete(); }),
          (e) => itemsToAdd.forEach(x => x.completion.error(e)));

      this.deferredPuts.splice(0, itemsToAdd.length);
    }

    if (this.deferredPuts.length > 0) {
      this.idlePutHandle = createIdle();
    } else {
      this.idlePutHandle = null;
    }
  });

  dn(`Queuing new item for write! ${JSON.stringify(newItem.key)}`);
  this.deferredPuts = this.deferredPuts || [];
  this.deferredPuts.push(newItem);

  if (!this.idlePutHandle) {
    this.idlePutHandle = createIdle();
  }

  return newItem.completion.toPromise();
}

function deferredGet<T, Key>(this: Dexie.Table<T, Key>, key: Key, database: Dexie): Promise<T> {
  let newItem = { key, completion: new AsyncSubject<T>() };

  let createIdle = () => window.setTimeout(async () => {
    let itemsToGet = this.deferredGets;
    this.deferredGets = [];

    try {
      await database.transaction('r', this, () => {
        return asyncMap(itemsToGet, (x) => {
          // First, search pending writes to see if we're about to save this
          dn(`Attempting to fetch ${x.key}!`);

          let pending = (this.deferredPuts || []).find(y => y.key === x.key);
          if (pending) {
            dn(`Early-completing ${key}!`);

            // NB: We need to do a shallow clone here because otherwise at some point,
            // api will be replaced by 'token' on this pending object, thereby trolling
            // the caller
            x.completion.next(Object.assign({}, pending.item));
            x.completion.complete();
            return Promise.resolve();
          }

          return this.get(x.key).then(result => {
            x.completion.next(result!);
            x.completion.complete();
          }, (e: Error) => x.completion.error(e));
        }, 32);
      });
    } finally {
      this.idleGetHandle = null;
    }

    if (this.deferredGets.length > 0) {
      this.idleGetHandle = createIdle();
    }
  }, 10);

  this.deferredGets = this.deferredGets || [];
  this.deferredGets.push(newItem);<commit_msg>Let's see if this hoses us<commit_after>import { AsyncSubject } from 'rxjs/AsyncSubject';

import Dexie from 'dexie';
import { asyncMap } from './promise-extras';
import { User, ChannelBase, Message } from './models/api-shapes';
import { Pair } from './utils';
import { Api } from './models/slack-api';

import './standard-operators';
import 'request-idle-polyfill';
import { MessageKey } from './store';

//const d = require('debug')('trickline:dexie-data-model');
const dn = require('debug')('trickline-noisy:dexie-data-model');

const VERSION = 1;

export interface DeferredPutItem<T, TKey> {
  item: T;
  key: TKey;
  completion: AsyncSubject<void>;
}

export interface DeferredGetItem<T, TKey> {
  key: TKey;
  completion: AsyncSubject<T>;
}

declare module 'dexie' {
  module Dexie {
    interface Table<T, Key> {
      deferredPut: ((item: T, key: Key) => Promise<void>);
      deferredGet: ((key: Key, database: Dexie) => Promise<T & { token: string }>);
      idlePutHandle: NodeJS.Timer | null;
      idleGetHandle: number | null;
      deferredPuts: DeferredPutItem<T & Api, Key>[];
      deferredGets: DeferredGetItem<T, Key>[];
    }
  }
}

function deferredPut<T, Key>(this: Dexie.Table<T, Key>, item: T & Api, key: Key): Promise<void> {
  let newItem = { item, key, completion: new AsyncSubject<void>() };
  newItem.item = Object.assign({}, newItem.item);

  let createIdle = () => window.requestIdleCallback(async deadline => {
    while (deadline.timeRemaining() > 5/*ms*/) {
      // TODO: This would be cooler if it recognized duplicate IDs and threw out the older
      // one (i.e. you update channel.topic and channel.name in the same batch, so we really
      // only need to write the newer one)
      let itemsToAdd = this.deferredPuts.slice(0, 128).map(x => {
        if (x.item.api) {
          x.item.token = x.item.api.token();
          x.item.api = null;
        }

        return x;
      });

      if (itemsToAdd.length < 1) break;

      await this.bulkPut(itemsToAdd.map(x => x.item))
        .then(
          () => itemsToAdd.forEach(x => { dn(`Actually wrote ${JSON.stringify(x.key)}!`); x.completion.next(undefined); x.completion.complete(); }),
          (e) => itemsToAdd.forEach(x => x.completion.error(e)));

      this.deferredPuts.splice(0, itemsToAdd.length);
    }

    if (this.deferredPuts.length > 0) {
      this.idlePutHandle = createIdle();
    } else {
      this.idlePutHandle = null;
    }
  });

  dn(`Queuing new item for write! ${JSON.stringify(newItem.key)}`);
  this.deferredPuts = this.deferredPuts || [];
  this.deferredPuts.push(newItem);

  if (!this.idlePutHandle) {
    this.idlePutHandle = createIdle();
  }

  return newItem.completion.toPromise();
}

function deferredGet<T, Key>(this: Dexie.Table<T, Key>, key: Key, database: Dexie): Promise<T> {
  let newItem = { key, completion: new AsyncSubject<T>() };

  let createIdle = () => window.requestAnimationFrame(async () => {
    let itemsToGet = this.deferredGets;
    this.deferredGets = [];

    try {
      await database.transaction('r', this, () => {
        return asyncMap(itemsToGet, (x) => {
          // First, search pending writes to see if we're about to save this
          dn(`Attempting to fetch ${x.key}!`);

          let pending = (this.deferredPuts || []).find(y => y.key === x.key);
          if (pending) {
            dn(`Early-completing ${key}!`);

            // NB: We need to do a shallow clone here because otherwise at some point,
            // api will be replaced by 'token' on this pending object, thereby trolling
            // the caller
            x.completion.next(Object.assign({}, pending.item));
            x.completion.complete();
            return Promise.resolve();
          }

          return this.get(x.key).then(result => {
            x.completion.next(result!);
            x.completion.complete();
          }, (e: Error) => x.completion.error(e));
        }, 32);
      });
    } finally {
      this.idleGetHandle = null;
    }

    if (this.deferredGets.length > 0) {
      this.idleGetHandle = createIdle();
    }
  });

  this.deferredGets = this.deferredGets || [];
  this.deferredGets.push(newItem);
{'fraction_non_alphanumeric': 0.15572592237661714, 'fraction_numerical': 0.003114518447532343, 'mean_word_length': 2.786394557823129, 'pattern_counts': {'":': 0, '<': 39, '<?xml version=': 0, '>': 77, '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': '1333177', 'n_tokens_mistral': 2768, 'n_tokens_neox': 2692, 'n_words': 649}