Merge pull request #34 from goodmind/type-safety

Type safety

most-typings.d.ts

@@ -6,6 +6,7 @@ declare module './lib' {
     token: Token
     updates: Stream<TcombUpdatesState>
     responses: Stream<any>
+    selectResponses<T, R> (query: Partial<{ responseType: t.Type<R>, method: string }>): Stream<T>
     events (eventName: string): Stream<TcombUpdate>
     dispose (): void
   }

rxjs-typings.d.ts

@@ -1,12 +1,14 @@
 import { TcombUpdate, TcombUpdatesState, Token } from './lib'
+import { ComponentSinks, ComponentSources } from './lib/plugins'
 import { Observable } from 'rxjs'
-import { ComponentSinks, ComponentSources } from './lib/plugins';
+import * as t from 'tcomb'
 
 declare module './lib' {
   interface DriverExecution {
     token: Token
     updates: Observable<TcombUpdatesState>
     responses: Observable<any>
+    selectResponses<T, R> (query: Partial<{ responseType: t.Type<R>, method: string }>): Observable<T>
     events (eventName: string): Observable<TcombUpdate>
     dispose (): void
   }

src/interfaces.ts

@@ -1,5 +1,6 @@
 import { Observable } from 'rxjs'
 import { TcombWebhookResponse, TcombRequest, TcombUpdate, TcombUpdatesState } from './runtime-types/types'
+import * as t from 'tcomb'
 
 export type Token = string
 export type GenericStream<T> = any
@@ -34,27 +35,37 @@ export interface DriverExecution {
   token: Token
   updates: GenericStream<TcombUpdatesState>
   responses: GenericStream<any>
+  selectResponses<T, R> (query: Partial<{ responseType: t.Type<R>, method: string }>): GenericStream<T>
   events (eventName: string): GenericStream<TcombUpdate>
   dispose (): void
 }
 
-export interface TelegramAPIRequest {
-  token: Token
-  method: string
-  query: any
-  httpMethod?: string
-}
+export namespace TelegramAPI {
+  export interface Request {
+    token: Token
+    method: string
+    query: any
+    httpMethod?: string
+    returnType?: t.Type<any>
+  }
 
-export interface TelegramAPIError {
-  ok: boolean
-  description: string
-  error_code: number
-}
+  export interface ResponseParameters {
+    migrate_to_chat_id: string
+    retry_after: number
+  }
+
+  export interface Error {
+    ok: boolean
+    description: string
+    error_code: number
+    parameters?: ResponseParameters
+  }
 
-export interface TelegramAPIResponseResult {}
+  export interface ResponseResult {}
 
-export interface TelegramAPIResponse {
-  ok: boolean
-  description?: string
-  result: TelegramAPIResponseResult
+  export interface Response {
+    ok: boolean
+    description?: string
+    result: ResponseResult
+  }
 }

src/runtime-types/types.ts

@@ -335,14 +335,15 @@ export const Request = t.struct<TcombRequest>({
   type: t.enums.of(['sink']),
   multipart: t.maybe(t.Boolean),
   method: t.String,
-  returnType: t.maybe(t.String),
+  // TODO: stricter types
+  returnType: t.maybe((t as any).Type),
   options: t.Object
 })
 export interface TcombRequest {
   type: 'sink',
   multipart?: boolean,
   method: string,
-  returnType?: string
+  returnType?: t.Type<any>
   options: any
 }
 

src/telegram-driver/api-request.ts

@@ -1,10 +1,15 @@
-import { TelegramAPIRequest, TelegramAPIResponse, TelegramAPIResponseResult, TelegramAPIError } from '../interfaces'
+import { TelegramAPI } from '../interfaces'
 
 import { Observable, Observer, Observable as $ } from 'rxjs'
 import * as request from 'superagent'
 import { Request, Response } from 'superagent'
 import { propOr, last, values, pipe, mapObjIndexed, curryN, ifElse } from 'ramda'
 
+export type OriginalResponseStream = Observable<TelegramAPI.ResponseResult | TelegramAPI.Error>
+export type ResponseStream =
+  Observable<TelegramAPI.ResponseResult | TelegramAPI.Error> &
+  { request: TelegramAPI.Request }
+
 let fromSuperagent =
   (request: Request): Observable<any> => $.create((obs: Observer<Response>): () => void => {
     request.end((err, res) => {
@@ -31,24 +36,37 @@ let transformReq = curryN(2, (req: Request, multipart: boolean) => ifElse(
   req.send.bind(req)
 ))
 
-export function makeAPIRequest (
-  {
-    token,
-    method,
-    query,
-    httpMethod = 'POST'
-  }: TelegramAPIRequest,
-  multipart = false
-): Observable<TelegramAPIResponseResult | TelegramAPIError> {
+function createResponse (
+  { token, method, query, httpMethod = 'POST' }: TelegramAPI.Request,
+  multipart: boolean
+): OriginalResponseStream {
   let endpoint = `https://api.telegram.org/bot${token}`
   let url = `${endpoint}/${method}`
   let req = transformReq(request(httpMethod, url).redirects(0), multipart)(query)
 
   return fromSuperagent(req)
     .catch(e => $.throw(e instanceof Error ? e : new Error(e)))
-    .map<any, TelegramAPIResponse>(res => res.body)
+    .map<any, TelegramAPI.Response>(res => res.body)
     .map(body => body.ok
       ? $.of(body.result)
       : $.throw(body))
-    .switch()
+}
+
+function makeRequestToResponse (request: TelegramAPI.Request) {
+  return function requestToResponse (response: OriginalResponseStream): ResponseStream {
+    Object.defineProperty(response, 'request', {
+      value: request,
+      writable: false
+    })
+
+    return (response as ResponseStream)
+  }
+}
+
+export function makeAPIRequest (
+  apiReq: TelegramAPI.Request,
+  multipart = false
+) {
+  return createResponse(apiReq, multipart)
+    .map(makeRequestToResponse(apiReq))
 }

src/telegram-driver/sinks.ts

@@ -1,4 +1,4 @@
-import { Request, WebhookResponse } from '../runtime-types'
+import { Request, Chat, WebhookResponse } from '../runtime-types'
 /* tslint:disable */
 import { TcombRequest, TcombWebhookResponse, TcombUpdate } from '../runtime-types/types'
 /* tslint:enable */
@@ -397,6 +397,7 @@ export let getChat = curryN(2, ({ chat_id }: SinkPayload, update: Update) => {
   return Request({
     type: 'sink',
     method: 'getChat',
+    returnType: Chat,
     options
   })
 })

src/telegram-driver/sources.ts

@@ -29,7 +29,7 @@ let makeUpdatesResolver =
     token,
     method: 'getUpdates',
     query: { offset, timeout: 60000 }
-  }))
+  }).switch())
 
 export function makeUpdates (initialState: TcombUpdatesState, token: Token): Observable<TcombUpdatesState> {
   UpdatesState(initialState)

src/telegram-driver/telegram-driver.ts

@@ -1,5 +1,6 @@
 import { Observable, Subject, Observable as $ } from 'rxjs'
-import { mapObjIndexed } from 'ramda'
+import { T, F, tryCatch, last, always, and, is, identity, mapObjIndexed, both, invoker, cond } from 'ramda'
+import { isType, Type } from 'tcomb'
 
 import {
   DriverOptions,
@@ -35,6 +36,36 @@ let makeEventsSelector =
         'callback_query': callbackQuery.share()
       })[eventName]
 
+function makeResponsesSelector (res: Observable<any>) {
+  const emptyType = Object.assign(identity.bind({}), { is: () => true })
+  const filter = (invoker as any)(1, 'filter')
+  const tryFilter = (x: any) => filter((tryCatch as any)(x, F))
+
+  return function responses<T, R> ({
+    responseType,
+    method
+  }: { responseType?: Type<R>, method?: string } = {}): Observable<T> {
+    const responseFilter = ([_, v]: any) => responseType.is(responseType(v))
+    const methodRFilter = ([{ request: { returnType: Type = emptyType } }, v]: any) => Type.is(Type(v))
+    const requestFilter = ([{ request: { returnType: Type = responseType } }, v]: any) => Type.is(Type(v))
+    const methodFilter = ([{ request: r }]: any) => r.method === method
+    let selectedRes = cond([
+      [
+        always(and(isType(responseType), !method)),
+        tryFilter(responseFilter)],
+      [
+        always(and(is(String, method), !responseType)),
+        tryFilter(both(methodFilter, methodRFilter))],
+      [
+        always(and(isType(responseType), is(String, method))),
+        tryFilter(both(methodFilter, requestFilter))],
+      [T, identity]
+    ])($.zip(res, res.switch())) as Observable<[any, any]>
+
+    return selectedRes.map<any, T>(last)
+  }
+}
+
 let handleWebhook = (
   token: Token,
   request: Observable<Observable<TcombWebhookResponse>>,
@@ -54,8 +85,9 @@ let handleRequest =
     .flatMap(({
       method,
       multipart,
+      returnType,
       options: query
-    }) => makeAPIRequest({token, method, query}, multipart))
+    }) => makeAPIRequest({returnType, token, method, query}, multipart))
 
 export function makeTelegramDriver (
   token: Token,
@@ -68,7 +100,7 @@ export function makeTelegramDriver (
   }
 
   let proxyUpdates = options.skipUpdates ? $.never<TcombUpdatesState>() : makeUpdates(state, token)
-  let proxyWebHook = new Subject<any>()
+  let proxyWebHook = new Subject<TcombUpdate[]>()
 
   if (options.webhook) {
     proxyUpdates = makeWebHook(state, proxyWebHook)
@@ -100,8 +132,9 @@ export function makeTelegramDriver (
       },
       mapObjIndexed(adapt, {
         events: makeEventsSelector(sources),
-        updates,
-        responses
+        selectResponses: makeResponsesSelector(responses),
+        responses: responses.switch(),
+        updates
       })) as DriverExecution
   }