feat: add txs websockets

node/coinstacks/common/api/src/websocket.ts

@@ -1,9 +1,9 @@
 import WebSocket from 'ws'
 import { v4 } from 'uuid'
-import { WebsocketClient } from '@shapeshiftoss/blockbook'
 import { Logger } from '@shapeshiftoss/logger'
 import { Registry } from './registry'
 import { Prometheus } from './prometheus'
+import { IWebsocketClient } from '@shapeshiftoss/blockbook'
 
 export interface RequestPayload {
   subscriptionId: string
@@ -42,7 +42,7 @@ export class ConnectionHandler {
 
   private readonly websocket: WebSocket
   private readonly registry: Registry
-  private readonly blockbook: WebsocketClient
+  private readonly blockbook: IWebsocketClient
   private readonly prometheus: Prometheus
   private readonly logger: Logger
   private readonly routes: Record<Topics, Methods>
@@ -54,7 +54,7 @@ export class ConnectionHandler {
   private constructor(
     websocket: WebSocket,
     registry: Registry,
-    blockbook: WebsocketClient,
+    blockbook: IWebsocketClient,
     prometheus: Prometheus,
     logger: Logger
   ) {
@@ -96,7 +96,7 @@ export class ConnectionHandler {
   static start(
     websocket: WebSocket,
     registry: Registry,
-    blockbook: WebsocketClient,
+    blockbook: IWebsocketClient,
     prometheus: Prometheus,
     logger: Logger
   ): void {
@@ -159,7 +159,7 @@ export class ConnectionHandler {
     }
 
     this.subscriptionIds.clear()
-    this.blockbook.subscribeAddresses(this.registry.getAddresses())
+    this.blockbook?.subscribeAddresses(this.registry.getAddresses())
   }
 
   private handleSubscribeTxs(subscriptionId: string, data?: TxsTopicData): void {

node/coinstacks/solana/api/src/app.ts

@@ -1,17 +1,30 @@
 import express from 'express'
 import { join } from 'path'
 import swaggerUi from 'swagger-ui-express'
-import { middleware, Prometheus } from '@shapeshiftoss/common-api'
+import { ConnectionHandler, middleware, Prometheus, Registry, TransactionHandler } from '@shapeshiftoss/common-api'
 import { Logger } from '@shapeshiftoss/logger'
 import { RegisterRoutes } from './routes'
+import { Server } from 'ws'
+import { SolanaWebsocketClient } from './websocket'
+import { Helius } from 'helius-sdk'
+import { GeyserResultTransaction } from './models'
 
 const PORT = process.env.PORT ?? 3000
+const RPC_API_KEY = process.env.RPC_API_KEY
+const WEBSOCKET_URL = process.env.WEBSOCKET_URL
+const WEBSOCKET_API_KEY = process.env.WEBSOCKET_API_KEY
 
 export const logger = new Logger({
   namespace: ['unchained', 'coinstacks', 'solana', 'api'],
   level: process.env.LOG_LEVEL,
 })
 
+if (!WEBSOCKET_URL) throw new Error('WEBSOCKET_URL env var not set')
+if (!WEBSOCKET_API_KEY) throw new Error('WEBSOCKET_API_KEY env var not set')
+if (!RPC_API_KEY) throw new Error('RPC_API_KEY env var not set')
+
+export const heliusSdk = new Helius(RPC_API_KEY)
+
 const prometheus = new Prometheus({ coinstack: 'solana' })
 
 const app = express()
@@ -45,4 +58,30 @@ app.get('/', async (_, res) => {
 
 app.use(middleware.errorHandler, middleware.notFoundHandler)
 
-app.listen(PORT, () => logger.info('Server started'))
+const server = app.listen(PORT, () => logger.info('Server started'))
+
+const transactionHandler: TransactionHandler<GeyserResultTransaction, GeyserResultTransaction> = async (geyserTx) => {
+  const addresses = geyserTx.transaction.message.accountKeys.map((key) => key.pubkey)
+
+  return { addresses, tx: geyserTx }
+}
+
+const registry = new Registry({
+  addressFormatter: (address: string) => address,
+  // @TODO: Make it optional in the registry class
+  blockHandler: async () => {
+    return { txs: [] }
+  },
+  transactionHandler,
+})
+
+const heliusWebsocket = new SolanaWebsocketClient(WEBSOCKET_URL, {
+  apiKey: WEBSOCKET_API_KEY,
+  transactionHandler: registry.onTransaction.bind(registry),
+})
+
+const wsServer = new Server({ server })
+
+wsServer.on('connection', (connection) => {
+  ConnectionHandler.start(connection, registry, heliusWebsocket, prometheus, logger)
+})

node/coinstacks/solana/api/src/controller.ts

@@ -1,4 +1,8 @@
 import { Logger } from '@shapeshiftoss/logger'
+import { validatePageSize } from '@shapeshiftoss/common-api'
+import { VersionedMessage } from '@solana/web3.js'
+import axios from 'axios'
+import { EnrichedTransaction } from 'helius-sdk'
 import { Body, Get, Path, Post, Query, Response, Route, Tags } from 'tsoa'
 import {
   BadRequestError,
@@ -9,11 +13,8 @@ import {
   ValidationError,
   handleError,
 } from '../../../common/api/src' // unable to import models from a module with tsoa
+import { heliusSdk } from './app'
 import { Account, API, EstimateFeesBody, PriorityFees, Tx, TxHistory } from './models'
-import { EnrichedTransaction, Helius } from 'helius-sdk'
-import { VersionedMessage } from '@solana/web3.js'
-import { validatePageSize } from '@shapeshiftoss/common-api'
-import axios from 'axios'
 
 const RPC_URL = process.env.RPC_URL
 const RPC_API_KEY = process.env.RPC_API_KEY
@@ -31,8 +32,6 @@ export const logger = new Logger({
   level: process.env.LOG_LEVEL,
 })
 
-const heliusSdk = new Helius(RPC_API_KEY)
-
 const axiosNoRetry = axios.create({ timeout: 5000, params: { 'api-key': RPC_API_KEY } })
 
 @Route('api/v1')

node/coinstacks/solana/api/src/models.ts

@@ -53,6 +53,144 @@ export interface PriorityFees {
   fast: number
 }
 
+// Geyser Types
+export interface GeyserWebsocketResponse {
+  jsonrpc: string
+  method: string
+  params: GeyserParams
+}
+
+export interface GeyserParams {
+  subscription: number
+  result: GeyserResult
+}
+
+export interface GeyserResult {
+  transaction: GeyserResultTransaction
+  signature: string
+  slot: number
+}
+
+export interface GeyserResultTransaction {
+  transaction: GeyserTransactionDetails
+  meta: GeyserTransactionMeta
+  version: number
+}
+
+export interface GeyserTransactionMeta {
+  err: null
+  status: GeyserTransactionStatus
+  fee: number
+  preBalances: number[]
+  postBalances: number[]
+  innerInstructions: GeyserInnerInstruction[]
+  logMessages: string[]
+  preTokenBalances: GeyserTokenBalance[]
+  postTokenBalances: GeyserTokenBalance[]
+  rewards: unknown[]
+  computeUnitsConsumed: number
+}
+
+export interface GeyserInnerInstruction {
+  index: number
+  instructions: GeyserInnerInstructionDetails[]
+}
+
+export interface GeyserInnerInstructionDetails {
+  program?: string
+  programId: string
+  parsed?: GeyserParsedInstruction
+  stackHeight: number
+  accounts?: string[]
+  data?: string
+}
+
+export interface GeyserParsedInstruction {
+  info: GeyserParsedInstructionInfo
+  type: string
+}
+
+export interface GeyserParsedInstructionInfo {
+  extensionTypes?: string[]
+  mint?: string
+  lamports?: number
+  newAccount?: string
+  owner?: string
+  source?: string
+  space?: number
+  account?: string
+  amount?: string
+  authority?: string
+  destination?: string
+}
+
+export interface GeyserTokenBalance {
+  accountIndex: number
+  mint: string
+  uiTokenAmount: GeyserUITokenAmount
+  owner: string
+  programId: string
+}
+
+export interface GeyserUITokenAmount {
+  uiAmount: number
+  decimals: number
+  amount: string
+  uiAmountString: string
+}
+
+export interface GeyserTransactionStatus {
+  Ok: null
+}
+
+export interface GeyserTransactionDetails {
+  signatures: string[]
+  message: GeyserTransactionMessage
+}
+
+export interface GeyserTransactionMessage {
+  accountKeys: GeyserAccountKey[]
+  recentBlockhash: string
+  instructions: GeyserMessageInstruction[]
+  addressTableLookups: unknown[]
+}
+
+export interface GeyserAccountKey {
+  pubkey: string
+  writable: boolean
+  signer: boolean
+  source: GeyserAccountSource
+}
+
+export enum GeyserAccountSource {
+  Transaction = 'transaction',
+}
+
+export interface GeyserMessageInstruction {
+  programId: string
+  accounts?: string[]
+  data?: string
+  stackHeight: null
+  program?: string
+  parsed?: GeyserParsedMessageInstruction
+}
+
+export interface GeyserParsedMessageInstruction {
+  info: GeyserParsedMessageInstructionInfo
+  type: string
+}
+
+export interface GeyserParsedMessageInstructionInfo {
+  account?: string
+  mint?: string
+  source: string
+  systemProgram?: string
+  tokenProgram?: string
+  wallet?: string
+  destination?: string
+  lamports?: number
+}
+
 /**
  * Contains the base64 encoded transaction message
  */