// @flow
import "babel-polyfill";
import { Observable } from 'rxjs/Rx';

type RepoIn<TIn,TOut> = (obj:TIn)=>TOut;
type RepoOut<TIn,TOut> = (obj:TOut)=>TIn;
type ConnectFn = (url:string)=>()=>Promise<Object>;

import {
  Connect as MongoConnect,
  Dispose as MongoDispose
} from './MongoConnect';

export {MongoConnect};
export {MongoDispose};

export const cursorToObservable = (cursor)=>{
  return Observable.create(observer=>cursor
    .once('end',()=>observer.complete())
    .once('error',err=>observer.error(err))
    .on('data',data=>observer.next(data))
  );
}

export const Query = getColumn=>
(params:Object = {},cursor:Object = { }):Observable=>
  Observable
  .fromPromise(getColumn())
  .map(col=>col.find(params))
  .map(queryCursor=>{
    if(cursor.limit != null){
      return queryCursor.limit(cursor.limit);
    }else {
      return queryCursor;
    }
  })
  .flatMap(cursorToObservable);

export const Insert = getColumn=>
(input:Observable):Observable=>Observable
  .fromPromise(getColumn())
  .mergeMap(col=>input
    // Buffer into batches of 1 thousand to avoid freaking mongo out
    .bufferCount(1000)
    // Insert the docs
    .flatMap(items=>{
      return col.insertMany(items)
    })
    .retry(5)
    // Flatten the batches back out into an observable
    .flatMap(res=>Observable.from(res.ops))
  );

const updateOptions = {upsert: false}
export const Update = getColumn =>
  (input: Observable, options: Object = updateOptions): Observable => Observable
    .fromPromise(getColumn())
    .mergeMap(col => input
      .flatMap(item => col
        .findOneAndReplace({_id: item._id}, item, options)
      )
      .map(res => (res.value && res.value._id) || res.lastErrorObject.upserted)
    );

export const Delete = getColumn=>
(input:Observable):Observable=>Observable
  .fromPromise(getColumn())
  .mergeMap(col=>input
    .flatMap(item=>col
      .findOneAndDelete({_id:item._id})
    )
    .map(res=>res.value._id)
  );

export const GetCollection = (collectionName:string,connectFn:ConnectFn)=>async()=>{
  let db = await connectFn();
  let col = db.collection(collectionName);
  return col;
};

export const HandleArrayArgument = (insertFn)=>(input:Array<Object>|Observable)=>{
  if(input instanceof Observable){
    return insertFn(input);
  }else{
    return insertFn(Observable.from(input));
  }
}

export const MapObservableArgument = (mapFn,decoratedFn)=>
(input:Observable)=>
  decoratedFn(input.map(mapFn));

export const MapObservableResult = (mapFn,decoratedFn)=>
(...args)=>decoratedFn(...args).map(mapFn);

export const CRUDRepository = (columnName:string,connectFn:ConnectFn)=>({
  query:Query(GetCollection(columnName,connectFn)),
  insert:HandleArrayArgument(
    Insert(GetCollection(columnName,connectFn))),
  update:Update(GetCollection(columnName,connectFn)),
  delete:Delete(GetCollection(columnName,connectFn))
})

export const MappedRepository = (columnName:string,connectFn:ConnectFn,inMap:RepoIn,outMap:RepoOut)=>({
  query:MapObservableResult(outMap,
    Query(GetCollection(columnName,connectFn))),
  insert:HandleArrayArgument(
    MapObservableArgument(inMap,
      MapObservableResult(outMap,
        Insert(GetCollection(columnName,connectFn))))),
  update:HandleArrayArgument(
    MapObservableArgument(inMap,
      Update(GetCollection(columnName,connectFn)))),
  delete:
  HandleArrayArgument(
    MapObservableArgument(inMap,
      Delete(GetCollection(columnName,connectFn))))
});

@@ -1329,45 +1329,273 @@
 );%0A%0A
+export 
 const 
-u
+U
 pdate
-Options = %7Bupsert: false%7D
+ = getColumn =%3E%0A  (input: Observable): Observable =%3E Observable%0A    .fromPromise(getColumn())%0A    .mergeMap(col =%3E input%0A      .flatMap(item =%3E col%0A        .findOneAndReplace(%7B_id: item._id%7D, item)%0A      )%0A      .map(res =%3E res.value._id)%0A    );%0A
 %0Aexp
@@ -1598,36 +1598,36 @@
 %0Aexport const Up
-date
+sert
  = getColumn =%3E%0A
@@ -1650,41 +1650,8 @@
 able
-, options: Object = updateOptions
 ): O
@@ -1812,102 +1812,115 @@
 em, 
-options)%0A      )%0A      .map(res =%3E (res.value && res.value._id) %7C%7C res.lastErrorObject.upserte
+%7B%0A          upsert: true,%0A          returnOriginal: false%0A        %7D)%0A      )%0A      .map(res =%3E res.value._i
 d)%0A 
@@ -3012,24 +3012,78 @@
 onnectFn)),%0A
+  upsert:Upsert(GetCollection(columnName,connectFn)),%0A
   delete:Del
@@ -3581,24 +3581,140 @@
 nectFn)))),%0A
+  upsert:HandleArrayArgument(%0A    MapObservableArgument(inMap,%0A      Upsert(GetCollection(columnName,connectFn)))),%0A
   delete:%0A  

Change upsert option to its own method