message-auditor.service.ts 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226
  1. import { map, Observable, of, Subject } from "rxjs";
  2. import { ErrorTrigger, MessageAuditorServiceInterface, MessageSynchronisationServiceSetting } from "../type/datatype";
  3. import { MessageLog } from "../dependencies/log/type/datatype";
  4. import * as _ from 'lodash'
  5. import { LoggingService } from "../dependencies/log/interface/export";
  6. import { BaseMessage } from "../dependencies/log/dependencies/msgutil/interface/export";
  7. let processedMsgIds = new Set();
  8. require('dotenv').config();
  9. export class MessageAuditorService implements MessageAuditorServiceInterface {
  10. private settings: MessageSynchronisationServiceSetting
  11. private sourceSrc: LoggingService = new LoggingService()
  12. private targetSrc: LoggingService = new LoggingService()
  13. private missingMessageSubject: Subject<MessageLog> = new Subject()
  14. /* Set up the targets or points of synchronization. This is where it will register the 2 different location of
  15. the data to be synchronized */
  16. public init(settings: MessageSynchronisationServiceSetting): void {
  17. this.settings = settings;
  18. if (settings.filters) {
  19. console.log(`Integrating filters: ${Object.keys(this.settings.filters)} in AuditMessage service`)
  20. }
  21. }
  22. /* This is the main interface of the message sync service. The argument will take in an observable stream of
  23. error notifications, prompting it to perform the subscription of the targeted sources and it's corresponding
  24. target. Essentially, this does not synchronize, but rather it checks against the two sources and compare
  25. and return the missing data, which will then be passed into the targeted subject stream as specified by the
  26. respective client. They can choose how they want to handle the missing messages returned. */
  27. public subscribe(obsTrigger: Observable<ErrorTrigger>): Observable<MessageLog> {
  28. // Subsribe to the errorTrigger obs to listen to any notification.
  29. obsTrigger.subscribe({
  30. next: obsTrigger => {
  31. console.log(obsTrigger.message)// just checking the message
  32. if (!this.settings.filters) {
  33. console.log(`No filters applies`)
  34. } else {
  35. console.log(`Synchronizating with filters: '${Object.keys(this.settings.filters)}': '${Object.values(this.settings.filters)}'`)
  36. }
  37. let missingMsg: Observable<MessageLog> = this.synchronize()
  38. missingMsg.subscribe({
  39. next: element => {
  40. this.missingMessageSubject.next(element)
  41. console.log(`AuditService: Returning missing messages ${element.appData.msgId} ....`)
  42. }
  43. })
  44. }
  45. })
  46. return this.missingMessageSubject
  47. }
  48. /* ________________ Private Functions _________________ */
  49. // Filtering functions to filters out messages
  50. private filterData(filters: any, message: MessageLog): boolean {
  51. let response: boolean = true //Just using this like a statemanagement
  52. let payload: BaseMessage = JSON.parse(message.appData.msgPayload as string) // Extract the payload from the messageLog first
  53. this.checkIfIsInPayloadDataFormat(payload) // Convert stringified nested payload if there's any
  54. // Making a separate function to cater to different multi filters conditions are coded below
  55. if (filters) { // if filters is not null
  56. if (Object.keys(filters).length > 1) {
  57. let totalCount = Object.keys(filters).length
  58. let matchedCount = 0
  59. Object.entries(filters).forEach(([key, value]) => {
  60. let filters = { [key]: value }
  61. // console.log(filters)
  62. if (this.checkValues(payload, filters) == true) matchedCount++
  63. })
  64. if (totalCount == matchedCount) { // check if all the criterias are met
  65. response = true
  66. } else {
  67. response = false
  68. }
  69. } else {
  70. if (this.checkValues(payload, filters) == true) {
  71. response = true
  72. } else {
  73. response = false
  74. }
  75. }
  76. } else { // if not filters is provided. Then the just let response be true so that the data can be further processed
  77. response = true
  78. }
  79. return response
  80. }
  81. /* This is where the 'synching' operation takes place. */
  82. private synchronize(): Subject<MessageLog> {
  83. let subjectOutput: Subject<MessageLog> = new Subject()
  84. // Acquire the data from both location and return them as an array respectively.
  85. this.acquireData().then((data: { arr1: MessageLog[], arr2: MessageLog[] }) => {
  86. // In the case where there are differences in the array length, then extensive comparison
  87. // will be carried out to filters out the differences. Differences are the missing data.
  88. if(process.env.CheckAudit)
  89. {
  90. console.log("Record set 1: ", _.keys(_.countBy(data.arr1,function(data:MessageLog){return data.appData['msgId']})).length);
  91. console.log("Record set 2: ", _.keys(_.countBy(data.arr2,function(data:MessageLog){return data.appData['msgId']})).length);
  92. }
  93. if(process.env.CheckAudit)
  94. {
  95. console.log("[CheckAudit] Record set 1: ", _.keys(_.countBy(data.arr1,function(data:MessageLog){return data.appData['msgId']})).length);
  96. console.log("[CheckAudit] Record set 2: ", _.keys(_.countBy(data.arr2,function(data:MessageLog){return data.appData['msgId']})).length);
  97. }
  98. this.checkArrayDifferences(data).then((data: MessageLog[]) => {
  99. if(process.env.CheckAudit)
  100. {
  101. console.log("Difference: ",data.length);
  102. }
  103. data.forEach(msgElement => {
  104. let refined = JSON.parse(JSON.stringify(msgElement))
  105. // Once the missing data has been weeded out, it is then passed into the Subject
  106. // to be returned for the subscribe method.`
  107. subjectOutput.next(refined)
  108. })
  109. })
  110. }).catch((e) => console.error(e))
  111. return subjectOutput
  112. }
  113. /* This is where the targeted data is queried. The process is pretty straightforward. */
  114. private async acquireData(): Promise<any> {
  115. const promiseQuery: Promise<any> = new Promise((resolve, reject) => {
  116. // declare what to expect.
  117. let allSets: { arr1: MessageLog[], arr2: MessageLog[] } = {
  118. arr1: [],
  119. arr2: []
  120. }
  121. let set1: MessageLog[] = []
  122. let set2: MessageLog[] = []
  123. // Initiate the source to find the location of the targeted data to be synched.
  124. this.sourceSrc.init(this.settings.incomingSource).then(() => {
  125. this.targetSrc.init(this.settings.target).then(() => {
  126. // Filter also carries out the query aspect of the operation, allowing it to acquire all the relevant data.
  127. this.sourceSrc.filter({ msgTag: this.settings.incomingSource.tags[0] }).then((data: MessageLog[]) => {
  128. data.forEach((message: MessageLog) => {
  129. if (this.filterData(this.settings.filters, message)) set1.push(message)
  130. })
  131. }).catch((err) => {
  132. console.error(err.message)
  133. }).then(() => {
  134. this.targetSrc.filter({ msgTag: this.settings.target.tags[0] }).then((data: MessageLog[]) => {
  135. data.forEach(message => {
  136. if (this.filterData(this.settings.filters, message)) set2.push(message)
  137. })
  138. allSets.arr1 = set1
  139. allSets.arr2 = set2
  140. resolve(allSets)
  141. })
  142. })
  143. })
  144. })
  145. })
  146. return promiseQuery
  147. }
  148. // compare results and return differences
  149. private async checkArrayDifferences(args: { arr1: MessageLog[], arr2: MessageLog[] }): Promise<MessageLog[]> {
  150. return new Promise((resolve, reject) => {
  151. let missingMsg: MessageLog[] = []
  152. args.arr1.forEach((msgElement: MessageLog) => {
  153. // In this case, we are just checking if the msgId matches within the given the array.
  154. // Just to save time, there's no need to check the entire message structure unless
  155. // the circumstances necessitates it.
  156. if (args.arr2.some(obj => obj.appData.msgId === msgElement.appData.msgId)) {
  157. console.log(`Item Found!`)
  158. } else {
  159. console.log(`This ${msgElement.appData.msgId} is missing`)
  160. missingMsg.push(msgElement)
  161. resolve(missingMsg)
  162. }
  163. })
  164. })
  165. }
  166. // To be used by the filterData function to check between payload values and filter conditions
  167. private checkValues(payload, filters): boolean { //FYI, all parameters are string
  168. let key = Object.keys(filters)
  169. // console.log(Object.values(filters))
  170. let value = Object.values(filters)[0]
  171. let res = _.get(payload, key[0])
  172. // Check first if the payload has the filtering properties/path
  173. if (_.has(payload, key[0])) {
  174. let strarray: string[]
  175. // check array
  176. if (Array.isArray(value)) {
  177. strarray = value as string[]
  178. }
  179. else {
  180. strarray = [value as string]
  181. }
  182. // compare array with that string
  183. if (strarray.includes(res)) {
  184. return true
  185. } else {
  186. return false
  187. }
  188. } else {
  189. console.log(`${key} does not exists in payload`)
  190. return false
  191. }
  192. }
  193. // Check in the case of notification messages, for the nested data properties
  194. // Notification message may have multiple nested data properties that maybe in string format
  195. private checkIfIsInPayloadDataFormat(payload: BaseMessage | any) {
  196. let parsedData: any
  197. if (payload.data
  198. && payload.data.data
  199. && payload.data.data.data && typeof payload.data.data.data === 'string') {
  200. parsedData = JSON.parse(payload.data.data.data)
  201. // console.log(parsedData)
  202. payload.data.data.data = parsedData
  203. return payload
  204. } else {
  205. return payload
  206. }
  207. }
  208. }