feat: notify price changes

history/.eslintrc.json

@@ -1,3 +1,4 @@
 {
-  "extends": "../eslintrc.base.json"
+  "extends": "../eslintrc.base.json",
+  "ignorePatterns": ["proto"]
 }

history/package.json

@@ -19,7 +19,8 @@
     "db:migrate": "knex --knexfile ./src/config/database.ts migrate:latest",
     "db:rollback": "knex --knexfile ./src/config/database.ts migrate:rollback",
     "db:create:seed": "knex --knexfile ./src/config/database.ts seed:make -x ts",
-    "db:seed": "knex --knexfile ./src/config/database.ts seed:run"
+    "db:seed": "knex --knexfile ./src/config/database.ts seed:run",
+    "codegen:notifications": "cd src/services/notifications/proto && buf generate"
   },
   "dependencies": {
     "@grpc/grpc-js": "^1.9.12",
@@ -39,12 +40,18 @@
     "ajv": "^8.12.0",
     "ccxt": "^4.1.70",
     "dotenv": "^16.0.3",
+    "google-protobuf": "^3.21.2",
     "grpc-health-check": "^2.0.0",
     "js-yaml": "^4.1.0",
     "knex": "^3.0.1",
     "lodash.merge": "^4.6.2",
     "node-cron": "^3.0.3",
     "pg": "^8.11.0",
     "pino": "^8.16.2"
+  },
+  "devDependencies": {
+    "grpc-tools": "^1.12.4",
+    "grpc_tools_node_protoc_ts": "^5.3.3",
+    "protoc-gen-js": "^3.21.2"
   }
 }

history/src/app/history/index.ts

@@ -1,2 +1,3 @@
 export * from "./get-price-history"
 export * from "./update-price-history"
+export * from "./notify-price-change"

history/src/app/history/index.types.d.ts

@@ -2,3 +2,7 @@ type GetPriceHistoryArgs = {
   currency: string
   range?: string
 }
+
+type NotifyPriceChangeArgs = {
+  range?: PriceRange
+}

history/src/app/history/notify-price-change.ts

@@ -0,0 +1,36 @@
+import { defaultBaseCurrency } from "@config"
+import { PriceRange, PriceRepositoryError } from "@domain/price"
+import { checkedToCurrency } from "@domain/primitives"
+import { PriceRepository } from "@services/database"
+import { NotificationsService } from "@services/notifications"
+
+export const notifyPriceChange = async ({ range }: NotifyPriceChangeArgs) => {
+  const quote = checkedToCurrency("USD")
+
+  if (quote instanceof Error) {
+    return quote
+  }
+
+  const rangeToQuery = range || PriceRange.OneDay
+  const prices = await PriceRepository().listPrices({
+    base: defaultBaseCurrency,
+    quote,
+    range: rangeToQuery,
+  })
+
+  if (prices instanceof PriceRepositoryError) {
+    return prices
+  }
+
+  if (prices.length < 2) {
+    return
+  }
+
+  const notificationsService = NotificationsService()
+
+  return notificationsService.priceChanged({
+    initialPrice: prices[0],
+    finalPrice: prices[prices.length - 1],
+    range: rangeToQuery,
+  })
+}

history/src/config/process.ts

@@ -12,3 +12,5 @@ export const databaseConfig = {
   poolMax: parseInt(process.env.DB_POOL_MAX || "", 10) || 5,
   debug: process.env.DB_DEBUG === "true",
 }
+
+export const notificationsEndpoint = process.env.notificationsEndpoint || "localhost:6685"

history/src/domain/notifications/errors.ts

@@ -0,0 +1,6 @@
+import { ErrorLevel, ServiceError } from "../errors"
+
+export class NotificationsServiceError extends ServiceError {}
+export class UnknownNotificationServiceError extends NotificationsServiceError {
+  level = ErrorLevel.Critical
+}

history/src/domain/notifications/index.ts

@@ -0,0 +1 @@
+export * from "./errors"

history/src/domain/notifications/index.types.d.ts

@@ -0,0 +1,11 @@
+type NotificationsServiceError = import("./errors").NotificationsServiceError
+
+interface INotificationsService {
+  priceChanged(args: PriceChangedArgs): Promise<void | NotificationsServiceError>
+}
+
+type PriceChangedArgs = {
+  range: PriceRange
+  initialPrice: Tick
+  finalPrice: Tick
+}

history/src/servers/history/cron.ts

@@ -8,6 +8,7 @@ dotenv.config()
 
 const startServer = async () => {
   await History.updatePriceHistory()
+  await History.notifyPriceChange({})
   await closeDbConnections()
 }
 

history/src/services/notifications/grpc-client.ts

@@ -0,0 +1,38 @@
+import { promisify } from "util"
+
+import { credentials, Metadata } from "@grpc/grpc-js"
+
+import { NotificationsServiceClient } from "./proto/notifications_grpc_pb"
+import {
+  HandleNotificationEventRequest,
+  HandleNotificationEventResponse,
+} from "./proto/notifications_pb"
+
+export type GrpcNotificationsClientConfig = {
+  notificationsApi: string
+}
+
+export const GrpcNotificationsClient = ({
+  notificationsApi,
+}: GrpcNotificationsClientConfig) => {
+  const notificationsClient = new NotificationsServiceClient(
+    notificationsApi,
+    credentials.createInsecure(),
+  )
+  const metadata = new Metadata()
+
+  const handleNotificationEvent = (request: HandleNotificationEventRequest) => {
+    return promisify<
+      HandleNotificationEventRequest,
+      Metadata,
+      HandleNotificationEventResponse
+    >(notificationsClient.handleNotificationEvent.bind(notificationsClient))(
+      request,
+      metadata,
+    )
+  }
+
+  return {
+    handleNotificationEvent,
+  }
+}

history/src/services/notifications/index.ts

@@ -0,0 +1,39 @@
+import { notificationsEndpoint } from "@config"
+import {
+  NotificationsServiceError,
+  UnknownNotificationServiceError,
+} from "@domain/notifications"
+
+import { GrpcNotificationsClient } from "./grpc-client"
+import { createPriceChangedEvent } from "./price-changed-event"
+import {
+  HandleNotificationEventRequest,
+  NotificationEvent,
+} from "./proto/notifications_pb"
+
+export const NotificationsService = (): INotificationsService => {
+  const grpcClient = GrpcNotificationsClient({
+    notificationsApi: notificationsEndpoint,
+  })
+
+  const priceChanged = async (
+    args: PriceChangedArgs,
+  ): Promise<void | NotificationsServiceError> => {
+    try {
+      const req = new HandleNotificationEventRequest()
+      const event = new NotificationEvent()
+      const priceChangedEvent = createPriceChangedEvent(args)
+      event.setPrice(priceChangedEvent)
+      req.setEvent(event)
+      await grpcClient.handleNotificationEvent(req)
+    } catch (err) {
+      if (err instanceof Error) {
+        return new NotificationsServiceError(err.message)
+      }
+      return new UnknownNotificationServiceError(err.toString())
+    }
+  }
+  return {
+    priceChanged,
+  }
+}

history/src/services/notifications/price-changed-event.ts

@@ -0,0 +1,28 @@
+import { Money, PriceChangeDirection, PriceChanged } from "./proto/notifications_pb"
+
+export const createPriceChangedEvent = ({
+  initialPrice,
+  finalPrice,
+}: PriceChangedArgs) => {
+  const priceChange = calculatePriceChangePercentage(initialPrice.price, finalPrice.price)
+
+  const priceOfOneBitcoin = new Money()
+  priceOfOneBitcoin.setMinorUnits(usdMajorUnitToMinorUnit(finalPrice.price))
+  priceOfOneBitcoin.setCurrencyCode("USD")
+
+  const priceChangedEvent = new PriceChanged()
+  priceChangedEvent.setPriceOfOneBitcoin(priceOfOneBitcoin)
+  priceChangedEvent.setDirection(priceChange.direction)
+  priceChangedEvent.setPriceChangePercentage(priceChange.percentage)
+  return priceChangedEvent
+}
+
+const usdMajorUnitToMinorUnit = (usd: number) => usd * 100
+
+const calculatePriceChangePercentage = (initialPrice: Price, finalPrice: Price) => {
+  const percentage = ((finalPrice - initialPrice) / initialPrice) * 100
+
+  const direction =
+    finalPrice > initialPrice ? PriceChangeDirection.UP : PriceChangeDirection.DOWN
+  return { percentage, direction }
+}

history/src/services/notifications/proto/buf.gen.yaml

@@ -0,0 +1,15 @@
+# /proto/buf.gen.yaml
+version: v1
+
+plugins:
+  - name: js
+    out: .
+    opt: import_style=commonjs,binary
+  - name: grpc
+    out: .
+    opt: grpc_js
+    path: grpc_tools_node_protoc_plugin
+  - name: ts
+    out: .
+    opt: grpc_js
+    path: protoc-gen-ts