feat(app-service): refactor trigger getter, publish;

packages/app-service/.env.template

@@ -0,0 +1,8 @@
+DB=laf-app
+DB_URI=mongodb://root:password123@localhost:27017/?authSource=admin&replicaSet=laf&writeConcern=majority
+DB_POOL_LIMIT=10
+SERVER_SALT=abcdefg1234567!@#$%^&sadfqwef&*^*#!@^
+LOG_LEVEL=debug
+ENABLE_CLOUD_FUNCTION_LOG = always
+# FILE_SYSTEM_HTTP_CACHE_CONTROL=max-age=60
+# FILE_SYSTEM_ENABLE_UNAUTHORIZED_UPLOAD = off

packages/app-service/src/api/function-log.ts

@@ -1,7 +1,7 @@
 /*
  * @Author: Maslow<wangfugen@126.com>
  * @Date: 2021-07-30 10:30:29
- * @LastEditTime: 2021-08-18 16:51:13
+ * @LastEditTime: 2021-09-09 23:43:20
  * @Description: 
  */
 import { Constants } from "../constants"
@@ -16,7 +16,6 @@ const db = DatabaseAgent.db
  */
 export async function addFunctionLog(data: any) {
   if (!data) return null
-
   const r = await db.collection(Constants.function_log_collection)
     .add(data)
 

packages/app-service/src/api/rules.ts

@@ -1,34 +1,35 @@
 /*
  * @Author: Maslow<wangfugen@126.com>
  * @Date: 2021-07-30 10:30:29
- * @LastEditTime: 2021-08-18 16:52:27
+ * @LastEditTime: 2021-09-09 23:42:13
  * @Description: 
  */
 
-import * as assert from 'assert'
 import { Constants } from '../constants'
 import { DatabaseAgent } from "../lib/database"
 import { PolicyAgentInstance } from '../lib/policy-agent'
+import { PolicyDataStruct } from '../lib/policy-agent/types'
 
-const db = DatabaseAgent.db
 
 /**
  * Get all access policies
  */
 export async function getPolicyRules() {
-  const r = await db.collection(Constants.policy_collection).get()
-  assert.ok(r.ok)
+  const db = DatabaseAgent.accessor.db
+  const docs = await db.collection(Constants.policy_collection)
+    .find({})
+    .toArray<PolicyDataStruct>()
 
-  return r.data
+  return docs
 }
 
 
 /**
  * Applying access policies' rules
  */
 export async function applyPolicyRules() {
-  PolicyAgentInstance.clear()
   const policies = await getPolicyRules()
+  PolicyAgentInstance.clear()
   for (const policy of policies) {
     await PolicyAgentInstance.set(policy.name, policy)
   }

packages/app-service/src/api/trigger.ts

@@ -1,37 +1,36 @@
 /*
  * @Author: Maslow<wangfugen@126.com>
  * @Date: 2021-07-30 10:30:29
- * @LastEditTime: 2021-08-18 16:53:53
+ * @LastEditTime: 2021-09-09 23:36:39
  * @Description: 
  */
 
+import { Trigger } from "cloud-function-engine"
 import { Constants } from "../constants"
 import { DatabaseAgent } from "../lib/database"
 
-const db = DatabaseAgent.db
-
 /**
  * Get triggers
  * @param status The default value is 1, means will select all enabled triggers
  * @returns 
  */
 export async function getTriggers(status = 1) {
-  const r = await db.collection(Constants.trigger_collection)
-    .where({ status: status })
-    .get()
+  const db = DatabaseAgent.accessor.db
 
-  return r.data
-}
+  const docs = await db.collection(Constants.function_collection)
+    .find({
+      triggers: { $exists: true },
+      'triggers.status': status
+    }, {
+      projection: { triggers: 1 }
+    })
+    .toArray()
 
-/**
- * Get trigger by id
- * @param id 
- * @returns 
- */
-export async function getTriggerById(id: string) {
-  const r = await db.collection(Constants.trigger_collection)
-    .where({ _id: id })
-    .getOne()
+  const triggers = []
+  for (const func of docs) {
+    func.triggers.forEach((tri: any) => tri['func_id'] = func._id)
+    triggers.push(...func.triggers)
+  }
 
-  return r.data
+  return triggers.map(data => Trigger.fromJson(data))
 }

packages/app-service/src/constants.ts

@@ -9,11 +9,6 @@ export const Constants = {
    */
   function_collection: '__published__functions',
 
-  /**
-   * collection name of cloud functions' triggers
-   */
-  trigger_collection: '__published__triggers',
-
   /**
    * collection name of policy which used for authenticating `client-access-database` requests
    */

packages/app-service/src/lib/scheduler/index.ts

@@ -1,12 +1,11 @@
 /*
  * @Author: Maslow<wangfugen@126.com>
  * @Date: 2021-07-30 10:30:29
- * @LastEditTime: 2021-09-06 00:27:44
+ * @LastEditTime: 2021-09-10 00:08:05
  * @Description: 
  */
 
 import { getTriggers } from "../../api/trigger"
-import { Trigger } from "cloud-function-engine"
 import { DatabaseAgent } from "../database"
 import { createLogger } from "../logger"
 import { ChangeStreamDocument } from "mongodb"
@@ -28,11 +27,9 @@ export const SchedulerInstance = new FrameworkScheduler()
  */
 accessor.ready.then(async () => {
   // initialize triggers
-  const data = await getTriggers()
-  logger.trace('loadTriggers: ', data)
-  const triggers = data.map(data => Trigger.fromJson(data))
+  const triggers = await getTriggers()
+  logger.debug('loadTriggers: ', triggers)
   SchedulerInstance.init(triggers)
-  SchedulerInstance.emit('App:ready')
   logger.info('triggers loaded')
 
   // initialize policies
@@ -42,17 +39,29 @@ accessor.ready.then(async () => {
   // watch database operation event through `WatchStream` of mongodb
   const db = accessor.db
   const stream = db.watch([], { fullDocument: 'updateLookup' })
-  stream.on("change", (doc) => {
-    DatabaseChangeEventCallBack(doc)
-  })
+  stream.on("change", (doc) => { DatabaseChangeEventCallBack(doc) })
+
+  // emit `App:ready` event
+  SchedulerInstance.emit('App:ready')
 })
 
-// debounce function `applyPolicyRules`
-const debouncedApplyPolicy = debounce(() => {
+/**
+ * debounce function `applyPolicyRules`
+ */
+const debouncedApplyPolicies = debounce(() => {
   applyPolicyRules()
     .then(() => logger.info('hot update: policy rules applied'))
     .catch(err => logger.error('hot update: policy rules applied failed: ', err))
-}, 2000, { trailing: true, leading: false })
+}, 1000, { trailing: true, leading: false })
+
+/**
+ * debounce function of apply triggers
+ */
+const debouncedApplyTriggers = debounce(() => {
+  getTriggers()
+    .then(data => SchedulerInstance.init(data))
+    .catch(err => logger.error('hot update: triggers applied failed', err))
+}, 1000, { trailing: true, leading: false })
 
 /**
  * Callback function for database change event
@@ -64,24 +73,12 @@ function DatabaseChangeEventCallBack(doc: ChangeStreamDocument) {
 
   // apply policies while policies changed
   if (collection === Constants.policy_collection && ['insert', 'update', 'delete', 'replace'].includes(operationType)) {
-    debouncedApplyPolicy()
+    debouncedApplyPolicies()
   }
 
-  // update triggers while triggers changed
-  if (collection === Constants.trigger_collection) {
-    if (['insert', 'update', 'replace'].includes(operationType)) {
-      const trigger = Trigger.fromJson(doc.fullDocument)
-      SchedulerInstance.updateTrigger(trigger)
-    }
-
-    if (operationType === 'delete') {
-      getTriggers()
-        .then(data => {
-          const triggers = data.map(it => Trigger.fromJson(it))
-          SchedulerInstance.init(triggers)
-        })
-        .catch(err => logger.error(err))
-    }
+  // update triggers while functions changed
+  if (collection === Constants.function_collection) {
+    debouncedApplyTriggers()
   }
 
   // ignore database changes of internal collections
@@ -92,4 +89,5 @@ function DatabaseChangeEventCallBack(doc: ChangeStreamDocument) {
   // emit database change event
   const event = `DatabaseChange:${collection}#${operationType}`
   SchedulerInstance.emit(event, doc)
+  logger.debug(`scheduler emit database change event: ${event}`)
 }

packages/app-service/src/lib/scheduler/scheduler.ts

@@ -1,7 +1,7 @@
 /*
  * @Author: Maslow<wangfugen@126.com>
  * @Date: 2021-07-30 10:30:29
- * @LastEditTime: 2021-08-18 16:17:46
+ * @LastEditTime: 2021-09-09 22:59:48
  * @Description: 
  */
 
@@ -54,6 +54,6 @@ export class FrameworkScheduler extends TriggerScheduler {
    * @param params 
    */
   async log(...params: any[]) {
-    logger.debug(...params)
+    logger.trace(...params)
   }
 }