feat(nlu): nlu cloud configuration

modules/nlu/package.json

@@ -26,7 +26,7 @@
     "rimraf": "^3.0.2"
   },
   "dependencies": {
-    "@botpress/nlu-client": "0.1.2",
+    "@botpress/nlu-client": "0.1.4",
     "axios": "^0.21.1",
     "bluebird": "^3.5.3",
     "bluebird-global": "^1.0.1",
@@ -37,9 +37,11 @@
     "fs-extra": "^7.0.1",
     "immutable": "^4.0.0-rc.12",
     "joi": "^13.6.0",
+    "lock": "^1.1.0",
     "lodash": "^4.17.19",
     "lru-cache": "^5.1.1",
     "mathjs": "^7.5.1",
+    "moment": "^2.29.1",
     "ms": "^2.1.1",
     "nanoid": "^2.0.1",
     "semver": "^7.3.2",

modules/nlu/src/backend/application/bot-factory.ts

@@ -1,12 +1,16 @@
 import * as sdk from 'botpress/sdk'
 
 import _ from 'lodash'
-import { IStanEngine } from '../stan'
+import crypto from 'crypto'
+import { Client } from '@botpress/nlu-client'
+import { LanguageSource } from 'src/config'
+import { IStanEngine, StanEngine } from '../stan'
 import pickSeed from './pick-seed'
 import { Bot, IBot } from './scoped/bot'
 import { ScopedDefinitionsService, IDefinitionsService } from './scoped/definitions-service'
 import { IDefinitionsRepository } from './scoped/infrastructure/definitions-repository'
 import { BotDefinition, BotConfig, I } from './typings'
+import { NLUCloudClient } from '../cloud/client'
 
 export interface ScopedServices {
   bot: IBot
@@ -23,16 +27,28 @@ export type IScopedServicesFactory = I<ScopedServicesFactory>
 
 export class ScopedServicesFactory {
   constructor(
-    private _engine: IStanEngine,
+    private _languageSource: LanguageSource,
     private _logger: sdk.Logger,
     private _makeDefRepo: DefinitionRepositoryFactory
   ) {}
 
+  private makeEngine(botConfig: BotConfig): IStanEngine {
+    const { cloud } = botConfig
+
+    const stanClient = cloud
+      ? new NLUCloudClient({ ...cloud, endpoint: this._languageSource.endpoint })
+      : new Client(this._languageSource.endpoint, this._languageSource.authToken)
+
+    return new StanEngine(stanClient, this._languageSource.authToken ?? '')
+  }
+
   public makeBot = async (botConfig: BotConfig): Promise<ScopedServices> => {
     const { id: botId } = botConfig
 
+    const engine = this.makeEngine(botConfig)
+
     const { defaultLanguage } = botConfig
-    const { languages: engineLanguages } = await this._engine.getInfo()
+    const { languages: engineLanguages } = await engine.getInfo()
     const languages = _.intersection(botConfig.languages, engineLanguages)
     if (botConfig.languages.length !== languages.length) {
       const missingLangMsg = `Bot ${botId} has configured languages that are not supported by language sources. Configure a before incoming hook to call an external NLU provider for those languages.`
@@ -50,7 +66,7 @@ export class ScopedServicesFactory {
 
     const defService = new ScopedDefinitionsService(botDefinition, defRepo)
 
-    const bot = new Bot(botDefinition, this._engine, defService, this._logger)
+    const bot = new Bot(botDefinition, engine, defService, this._logger)
 
     return {
       defService,

modules/nlu/src/backend/application/index.ts

@@ -90,7 +90,7 @@ export class NLUApplication {
     return async (lang: string) => {
       const trainSet = await defService.getTrainSet(lang)
       const trainInput = mapTrainSet(trainSet)
-      const { exists, modelId } = await this._engine.hasModelFor(bot.id, trainInput)
+      const { exists, modelId } = await bot.hasModelFor(trainInput)
       const trainId = { botId, language: lang }
       if (exists) {
         await this.trainRepository.inTransaction(trx => trx.delete(trainId))

modules/nlu/src/backend/application/scoped/bot.ts

@@ -7,6 +7,7 @@ import { Predictor, ProgressCallback, Trainable, I } from '../typings'
 
 import { IDefinitionsService } from './definitions-service'
 import { ScopedPredictionHandler } from './prediction-handler'
+import { TrainInput } from '@botpress/nlu-client'
 
 interface BotDefinition {
   botId: string
@@ -105,4 +106,8 @@ export class Bot implements Trainable, Predictor {
     const { _predictor, _defaultLanguage } = this
     return _predictor.predict(textInput, anticipatedLanguage ?? _defaultLanguage)
   }
+
+  public hasModelFor(trainInput: TrainInput) {
+    return this._engine.hasModelFor(this.id, trainInput)
+  }
 }

modules/nlu/src/backend/application/typings.ts

@@ -9,6 +9,13 @@ export interface BotConfig {
   defaultLanguage: string
   languages: string[]
   nluSeed?: number
+  cloud?: CloudConfig
+}
+
+export interface CloudConfig {
+  oauthUrl: string
+  clientId: string
+  clientSecret: string
 }
 
 export interface BotDefinition {

modules/nlu/src/backend/bootstrap.ts

@@ -41,19 +41,17 @@ export async function bootStrap(bp: typeof sdk): Promise<NonBlockingNluApplicati
     )
   }
 
-  const { endpoint, authToken } = getNLUServerConfig(globalConfig.nluServer)
-  const stanClient = new Client(endpoint, authToken)
-
-  const modelPassword = '' // No need for password as Stan is protected by an auth token
-  const engine = new StanEngine(stanClient, modelPassword)
+  const nluServerConnectionInfo = getNLUServerConfig(globalConfig.nluServer)
+  const stanClient = new Client(nluServerConnectionInfo.endpoint, nluServerConnectionInfo.authToken)
+  const engine = new StanEngine(stanClient, '') // No need for password as Stan is protected by an auth token
 
   const socket = getWebsocket(bp)
 
   const botService = new BotService()
 
   const makeDefRepo = (bot: BotDefinition) => new ScopedDefinitionsRepository(bot, bp)
 
-  const servicesFactory = new ScopedServicesFactory(engine, bp.logger, makeDefRepo)
+  const servicesFactory = new ScopedServicesFactory(nluServerConnectionInfo, bp.logger, makeDefRepo)
 
   const trainRepo = new TrainingRepository(bp.database)
   const trainingQueue = new DistributedTrainingQueue(trainRepo, bp.logger, botService, bp.distributed, socket, {

modules/nlu/src/backend/cloud/cache.ts

@@ -0,0 +1,52 @@
+import ms from 'ms'
+import { Locker } from './lock'
+
+export interface Options<T> {
+  getToken?: (res: T) => string
+  getExpiryInMs?: (res: T) => number
+}
+
+const defaultExpiry = ms('10m') // 10 minutes in ms
+const isTokenActive = (token: string | null, expiration: number | null) =>
+  token && expiration && expiration > Date.now()
+
+export const cache = <T>(authenticate: () => Promise<T>, options?: Options<T>) => {
+  const getToken: (res: any) => string = options?.getToken ?? (res => res)
+  const getExpiry: (res: any) => number = options?.getExpiryInMs ?? (() => defaultExpiry)
+
+  const lock = Locker()
+
+  let token: string | null = null
+  let expiration: number | null = null
+
+  const tokenCache = async () => {
+    if (isTokenActive(token, expiration)) {
+      return token!
+    }
+
+    const unlock = await lock('cache')
+
+    try {
+      if (isTokenActive(token, expiration)) {
+        return token!
+      }
+
+      const res = await authenticate()
+
+      token = getToken(res)
+      expiration = Date.now() + getExpiry(res)
+      return token
+    } catch (e) {
+      throw e
+    } finally {
+      unlock()
+    }
+  }
+
+  tokenCache.reset = () => {
+    token = null
+    expiration = null
+  }
+
+  return tokenCache
+}

modules/nlu/src/backend/cloud/client.ts

@@ -0,0 +1,12 @@
+import { Client } from '@botpress/nlu-client'
+import { AxiosInstance } from 'axios'
+import { createOauthClient, OauthClientProps } from './oauth'
+
+export class NLUCloudClient extends Client {
+  private _client: AxiosInstance
+
+  constructor(options: OauthClientProps) {
+    super(options.endpoint, undefined)
+    this._client = createOauthClient(options)
+  }
+}

modules/nlu/src/backend/cloud/lock.test.ts

@@ -0,0 +1,43 @@
+import { Locker } from './lock'
+
+describe('getLock', () => {
+  it('should never release a lock', async () => {
+    const fn = jest.fn()
+
+    const lock = Locker()
+    lock('key')
+    lock('key').then(() => {
+      fn()
+    })
+
+    await new Promise(r => setTimeout(r, 100))
+
+    expect(fn).not.toHaveBeenCalled()
+  })
+
+  it('should wait for lock before updating value', async done => {
+    const lock = Locker()
+
+    let value = ''
+
+    lock('key').then(unlock =>
+      setTimeout(() => {
+        value = 'unexpected'
+        unlock()
+      }, 100)
+    )
+
+    expect(value).toEqual('')
+
+    await lock('key')
+
+    expect(value).toEqual('unexpected')
+    value = 'expected'
+    expect(value).toEqual('expected')
+
+    setTimeout(() => {
+      expect(value).toEqual('expected')
+      done!()
+    }, 200)
+  })
+})

modules/nlu/src/backend/cloud/lock.ts

@@ -0,0 +1,17 @@
+import { Lock as LockWithCallback } from 'lock'
+
+type unlockFunc = () => void
+
+export const Locker = () => {
+  const lock = LockWithCallback()
+
+  const getLock = async (key: string) =>
+    new Promise<unlockFunc>(resolve => {
+      lock(key, async unlockFn => {
+        const unlock = unlockFn()
+        resolve(unlock)
+      })
+    })
+
+  return getLock
+}

modules/nlu/src/backend/cloud/oauth.ts

@@ -0,0 +1,70 @@
+import axios, { AxiosError, AxiosInstance, AxiosRequestConfig } from 'axios'
+import { cache } from './cache'
+import qs from 'querystring'
+
+interface OauthTokenClientProps {
+  oauthUrl: string
+  clientId: string
+  clientSecret: string
+}
+
+interface OauthResponse {
+  access_token: string
+  expires_in: number
+  scope: string
+  token_type: string
+}
+
+const createOauthTokenClient = (axios: AxiosInstance, oauthTokenClientProps: OauthTokenClientProps) => async () => {
+  const { oauthUrl, clientId, clientSecret } = oauthTokenClientProps
+  const res = await axios.post(
+    oauthUrl,
+    qs.stringify({ client_id: clientId, client_secret: clientSecret, grant_type: 'client_credentials' })
+  )
+
+  return res.data as OauthResponse
+}
+
+const requestInterceptor = (authenticate: () => Promise<string>) => async (config: AxiosRequestConfig) => {
+  const token = await authenticate()
+  config.headers.Authorization = `Bearer ${token}`
+  return config
+}
+
+type ErrorRetrier = AxiosError & { config: { _retry: boolean } }
+
+const errorInterceptor = (instance: AxiosInstance, authenticate: () => Promise<string>) => async (
+  error: ErrorRetrier
+) => {
+  if (error.response?.status === 401 && !error.config._retry) {
+    error.config._retry = true
+    const token = await authenticate()
+    const config = error.config
+    config.headers.Authorization = `Bearer ${token}`
+    return instance.request(config)
+  }
+
+  return Promise.reject(error)
+}
+
+export type OauthClientProps = OauthTokenClientProps & {
+  endpoint: string
+}
+
+export const createOauthClient = ({ clientId, clientSecret, endpoint, oauthUrl }: OauthClientProps) => {
+  const oauthTokenClient = createOauthTokenClient(axios.create(), {
+    oauthUrl,
+    clientId,
+    clientSecret
+  })
+
+  const tokenCache = cache(oauthTokenClient, {
+    getExpiryInMs: res => res.expires_in * 1000,
+    getToken: res => res.access_token
+  })
+
+  const axiosClient = axios.create({ baseURL: endpoint })
+  axiosClient.interceptors.request.use(requestInterceptor(tokenCache))
+  axiosClient.interceptors.response.use(undefined, errorInterceptor(axiosClient, tokenCache))
+  return axiosClient
+}

modules/nlu/yarn.lock

@@ -2,10 +2,10 @@
 # yarn lockfile v1
 
 
-"@botpress/nlu-client@0.1.2":
-  version "0.1.2"
-  resolved "https://registry.yarnpkg.com/@botpress/nlu-client/-/nlu-client-0.1.2.tgz#9efacce79fd5229c75b7f04c65dcbe0416736376"
-  integrity sha512-YX7dzUK0q1TbDPWqIETHxsj/6h6s4y9t758UxrdSasGq9Zx3q81SZXqh0bnx0coi3Xv8OcNZy2ofhW33lVFCmg==
+"@botpress/nlu-client@0.1.4":
+  version "0.1.4"
+  resolved "https://registry.yarnpkg.com/@botpress/nlu-client/-/nlu-client-0.1.4.tgz#9aa8fb4cb3481532fd67f5ea00a969b1a25aadf9"
+  integrity sha512-oNrtZ1irfxv3PxMmfAUQA5yxKQF+S2LFeVGrXH95oxzELjj4CLyJeBOYmxT7aX5vR8PMLiaXiKsjS6wtRzIogg==
   dependencies:
     axios "^0.21.1"
     lodash "^4.17.19"
@@ -545,6 +545,11 @@ jsonfile@^4.0.0:
   optionalDependencies:
     graceful-fs "^4.1.6"
 
+lock@^1.1.0:
+  version "1.1.0"
+  resolved "https://registry.yarnpkg.com/lock/-/lock-1.1.0.tgz#53157499d1653b136ca66451071fca615703fa55"
+  integrity sha1-UxV0mdFlOxNspmRRBx/KYVcD+lU=
+
 lodash@>=4.17.21, lodash@^4.1.1, lodash@^4.17.19, lodash@^4.17.4:
   version "4.17.21"
   resolved "https://registry.npmjs.org/lodash/-/lodash-4.17.21.tgz#679591c564c3bffaae8454cf0b3df370c3d6911c"
@@ -661,6 +666,11 @@ mkdirp@^0.5.0:
   dependencies:
     minimist "^1.2.5"
 
+moment@^2.29.1:
+  version "2.29.1"
+  resolved "https://registry.yarnpkg.com/moment/-/moment-2.29.1.tgz#b2be769fa31940be9eeea6469c075e35006fa3d3"
+  integrity sha512-kHmoybcPV8Sqy59DwNDY3Jefr64lK/by/da0ViFcuA4DH0vQg5Q6Ze5VimxkfQNSC+Mls/Kx53s7TjP1RhFEDQ==
+
 ms@2.0.0:
   version "2.0.0"
   resolved "https://registry.yarnpkg.com/ms/-/ms-2.0.0.tgz#5608aeadfc00be6c2901df5f9861788de0d597c8"

packages/bp/src/sdk/botpress.d.ts

@@ -936,6 +936,14 @@ declare module 'botpress/sdk' {
     qna: {
       disabled: boolean
     }
+
+    cloud?: CloudConfig
+  }
+
+  export interface CloudConfig {
+    oauthUrl: string
+    clientId: string
+    clientSecret: string
   }
 
   export type Pipeline = Stage[]