feat: transaction grpc server

bats/ci_run.sh

@@ -7,7 +7,7 @@ echo "Running rust builds..."
 buck2 build //core/api-keys:api-keys //core/notifications:notifications
 
 echo "Running api builds..."
-buck2 build //core/api:api //core/api-ws-server:api-ws-server //core/api-trigger:api-trigger //core/api-exporter:api-exporter --verbose 4
+buck2 build //core/api:api //core/api-ws-server:api-ws-server //core/api-trigger:api-trigger //core/api-exporter:api-exporter //core/api-transactions-grpc-stream:api-transactions-grpc-stream --verbose 4
 
 # echo "Running apps builds..."
 # buck2 build //apps/dashboard:dashboard //apps/consent:consent //apps/pay:pay-ci //apps/admin-panel:admin-panel //apps/map:map //apps/voucher:voucher --verbose 4

bats/ci_setup_suite.bash

@@ -2,6 +2,7 @@
 
 export REPO_ROOT=$(git rev-parse --show-toplevel)
 source "${REPO_ROOT}/bats/helpers/_common.bash"
+source "${REPO_ROOT}/bats/helpers/transactions-grpc-stream.bash"
 
 TILT_PID_FILE="${BATS_ROOT_DIR}/.tilt_pid"
 
@@ -11,6 +12,7 @@ setup_suite() {
   await_notifications_is_up
   await_api_keys_is_up
   await_api_is_up
+  await_transactions_grpc_stream_is_up
   await_pay_is_up
 }
 
@@ -55,3 +57,7 @@ await_notifications_is_up() {
 
   retry 360 2 notifications_is_up
 }
+
+await_transactions_grpc_stream_is_up() {
+  retry 360 2 transactions_grpc_stream_is_up
+}

bats/core/api/transactions-grpc-stream.bats

@@ -0,0 +1,70 @@
+#!/usr/bin/env bats
+
+load "../../helpers/_common.bash"
+load "../../helpers/onchain.bash"
+load "../../helpers/transactions-grpc-stream.bash"
+load "../../helpers/user.bash"
+
+ALICE='alice'
+
+setup_file() {
+  clear_cache
+  create_user "$ALICE"
+}
+
+create_settled_onchain_receive_and_get_transaction_id() {
+  local token_name=$1
+  local wallet_id_name="$token_name.btc_wallet_id"
+
+  variables=$(
+    jq -n \
+    --arg wallet_id "$(read_value $wallet_id_name)" \
+    '{input: {walletId: $wallet_id}}'
+  )
+  exec_graphql "$token_name" 'on-chain-address-create' "$variables" >&2
+  address="$(graphql_output '.data.onChainAddressCreate.address')"
+  [[ "${address}" != "null" ]] || exit 1
+
+  bitcoin_cli sendtoaddress "$address" "0.001" >&2
+  bitcoin_cli -generate 2 >&2
+  retry 30 1 check_for_onchain_initiated_settled "$token_name" "$address" >&2
+
+  exec_graphql "$token_name" 'transactions' '{"first": 20}' >&2
+  transaction_id="$(get_from_transaction_by_address "$address" '.id')"
+  [[ -n "${transaction_id}" ]] || exit 1
+  [[ "${transaction_id}" != "null" ]] || exit 1
+
+  echo "$transaction_id"
+}
+
+@test "transactions-grpc-stream: invalid cursor returns invalid argument" {
+  transactions_grpc_stream_request '{"after_transaction_id":"not-an-object-id"}'
+
+  [[ "$status" -ne 0 ]] || exit 1
+  echo "$output" | grep -F "InvalidArgument" || exit 1
+  echo "$output" | grep -F "after_transaction_id must be a valid Mongo ObjectId" || exit 1
+}
+
+@test "transactions-grpc-stream: empty cursor returns invalid argument" {
+  transactions_grpc_stream_request '{"after_transaction_id":""}'
+
+  [[ "$status" -ne 0 ]] || exit 1
+  echo "$output" | grep -F "InvalidArgument" || exit 1
+  echo "$output" | grep -F "after_transaction_id must be a valid Mongo ObjectId" || exit 1
+}
+
+@test "transactions-grpc-stream: replays transactions after cursor" {
+  first_transaction_id="$(create_settled_onchain_receive_and_get_transaction_id "$ALICE")"
+  second_transaction_id="$(create_settled_onchain_receive_and_get_transaction_id "$ALICE")"
+
+  data=$(
+    jq -n \
+    --arg after_transaction_id "$first_transaction_id" \
+    '{after_transaction_id: $after_transaction_id}'
+  )
+  transactions_grpc_stream_request "$data" -max-time 3
+
+  echo "$output" | grep -F "\"ledgerTransactionId\": \"$second_transaction_id\"" || exit 1
+  echo "$output" | grep -F "\"walletId\": \"$(read_value "$ALICE.btc_wallet_id")\"" || exit 1
+  echo "$output" | grep -F "\"accountId\": \"$(read_value "$ALICE.account_id")\"" || exit 1
+}

bats/helpers/_common.bash

@@ -220,6 +220,7 @@ grpcurl_request() {
   local address="$3"
   local service_method="$4"
   local data="${5:-""}"
+  shift 5
 
   echo "gRPCurl request - import-path ${import_path} - proto: ${proto_file} - address: ${address} - service/method: ${service_method} - data: ${data}"
 
@@ -228,7 +229,7 @@ grpcurl_request() {
     run_cmd="run"
   fi
 
-  cmd=(${run_cmd} grpcurl -plaintext -import-path ${import_path} -proto "${proto_file}")
+  cmd=(${run_cmd} grpcurl -plaintext "$@" -import-path ${import_path} -proto "${proto_file}")
 
   if [[ -n "$data" ]]; then
     cmd+=(-d "${data}")

bats/helpers/transactions-grpc-stream.bash

@@ -0,0 +1,25 @@
+CURRENT_FILE=${BASH_SOURCE:-bats/helpers/.}
+source "$(dirname "$CURRENT_FILE")/_common.bash"
+
+export TRANSACTIONS_GRPC_STREAM_ADDRESS="${TRANSACTIONS_GRPC_STREAM_ADDRESS:-localhost:50053}"
+export TRANSACTIONS_GRPC_STREAM_HEALTH_URL="${TRANSACTIONS_GRPC_STREAM_HEALTH_URL:-http://localhost:8889/healthz}"
+export TRANSACTIONS_GRPC_STREAM_PROTO_IMPORT_PATH="${REPO_ROOT}/core/api/src/servers/transactions-grpc-stream/proto"
+export TRANSACTIONS_GRPC_STREAM_PROTO_FILE="transactions.proto"
+export TRANSACTIONS_GRPC_STREAM_SERVICE_METHOD="services.transactions.v1.TransactionsStream/SubscribeTransactions"
+
+transactions_grpc_stream_request() {
+  local data="${1:-""}"
+  shift || true
+
+  grpcurl_request \
+    "${TRANSACTIONS_GRPC_STREAM_PROTO_IMPORT_PATH}" \
+    "${TRANSACTIONS_GRPC_STREAM_PROTO_FILE}" \
+    "${TRANSACTIONS_GRPC_STREAM_ADDRESS}" \
+    "${TRANSACTIONS_GRPC_STREAM_SERVICE_METHOD}" \
+    "${data}" \
+    "$@"
+}
+
+transactions_grpc_stream_is_up() {
+  curl -fsS "${TRANSACTIONS_GRPC_STREAM_HEALTH_URL}" > /dev/null
+}

ci/core/pipeline.yml

@@ -250,11 +250,11 @@ jobs:
           - get: #@ component_src_resource_name("api")
             trigger: true
             passed:
-              #@ for component in ["api", "api-migrate", "api-ws-server", "api-trigger"]:
+              #@ for component in ["api", "api-migrate", "api-ws-server", "api-trigger", "api-transactions-grpc-stream"]:
               -  #@ build_edge_image_name(component)
           #@ end
           - get: pipeline-tasks
-          #@ for component in ["api", "api-migrate", "api-ws-server", "api-trigger", "notifications"]:
+          #@ for component in ["api", "api-migrate", "api-ws-server", "api-trigger", "api-transactions-grpc-stream", "notifications"]:
           - get: #@ edge_image_resource_name(component)
             passed:
               -  #@ build_edge_image_name(component)
@@ -272,6 +272,7 @@ jobs:
             - name: api-edge-image
             - name: api-migrate-edge-image
             - name: api-trigger-edge-image
+            - name: api-transactions-grpc-stream-edge-image
             - name: notifications-edge-image
             - name: pipeline-tasks
           outputs:

ci/core/tasks/open-core-bundle-charts-pr.sh

@@ -33,7 +33,7 @@ git remote set-url origin ${github_url}
 git config remote.origin.fetch "+refs/heads/*:refs/remotes/origin/*"
 
 git checkout ${ref}
-app_src_files=($(buck2 uquery 'inputs(deps("//core/api:")) + inputs(deps("//core/api-ws-server:")) + inputs(deps("//core/api-trigger:")) + inputs(deps("//core/api-exporter:")) + inputs(deps("//core/api-cron:"))' 2>/dev/null))
+app_src_files=($(buck2 uquery 'inputs(deps("//core/api:")) + inputs(deps("//core/api-ws-server:")) + inputs(deps("//core/api-trigger:")) + inputs(deps("//core/api-transactions-grpc-stream:")) + inputs(deps("//core/api-exporter:")) + inputs(deps("//core/api-cron:"))' 2>/dev/null))
 
 declare -A relevant_commits
 relevant_commits=()

ci/core/tasks/prep-quickstart.sh

@@ -2,6 +2,7 @@
 
 export api_digest=$(cat ./api-edge-image/digest | sed 's/:/@/')
 export trigger_digest=$(cat ./api-trigger-edge-image/digest | sed 's/:/@/')
+export transactions_grpc_stream_digest=$(cat ./api-transactions-grpc-stream-edge-image/digest | sed 's/:/@/')
 export notifications_digest=$(cat ./notifications-edge-image/digest | sed 's/:/@/')
 export migrate_digest=$(cat ./api-migrate-edge-image/digest | sed 's/:/@/')
 
@@ -11,6 +12,7 @@ pushd repo/quickstart || exit 1
 
 ./bin/bump-galoy-image-digest.sh "api" "$api_digest"
 ./bin/bump-galoy-image-digest.sh "trigger" "$trigger_digest"
+./bin/bump-galoy-image-digest.sh "transactions_grpc_stream" "$transactions_grpc_stream_digest"
 ./bin/bump-galoy-image-digest.sh "notifications" "$notifications_digest"
 ./bin/bump-mongodb-migrate-image-digest.sh "$migrate_digest"
 ./bin/re-render.sh

ci/core/template.lib.yml

@@ -1,6 +1,6 @@
 #@ load("@ytt:data", "data")
 
-#@ core_bundle_components = ["api", "api-cron", "api-trigger", "api-ws-server", "api-exporter"]
+#@ core_bundle_components = ["api", "api-cron", "api-trigger", "api-transactions-grpc-stream", "api-ws-server", "api-exporter"]
 
 #@ def galoy_dev_image():
 #@   return data.values.docker_registry + "/galoy-dev"

core/api-transactions-grpc-stream/BUCK

@@ -0,0 +1,20 @@
+load(
+  "@toolchains//workspace-pnpm:macros.bzl",
+  "prod_tsc_build_bin",
+)
+
+alias(
+    name = "prod_build",
+    actual = "//core/api:prod_build",
+)
+
+prod_tsc_build_bin(
+  name = "api-transactions-grpc-stream",
+  preload_file = "services/tracing.js",
+  run_file = "servers/transactions-grpc-stream-server.js",
+)
+
+alias(
+    name = "dev",
+    actual = "//core/api:dev-transactions-grpc-stream",
+)

core/api-transactions-grpc-stream/Dockerfile

@@ -0,0 +1,32 @@
+FROM nixos/nix:latest AS builder
+ARG APP=api-transactions-grpc-stream
+
+COPY . /workdir
+WORKDIR /workdir
+
+RUN set -eux; \
+    nix \
+    --extra-experimental-features "nix-command flakes impure-derivations ca-derivations" \
+    --option filter-syscalls false \
+    build \
+    ".#$APP";
+
+RUN mkdir -p /tmp/nix-store-closure /tmp/local-bin
+RUN cp -R $(nix-store --query --requisites result/) /tmp/nix-store-closure
+RUN ln -snf $(nix-store --query result/)/bin/* /tmp/local-bin/
+
+FROM gcr.io/distroless/static-debian11 AS final
+ARG APP=api-transactions-grpc-stream
+
+WORKDIR /app/$APP
+COPY --from=builder /tmp/nix-store-closure /nix/store
+COPY --from=builder /tmp/local-bin/* /usr/local/bin/
+
+USER 1000
+
+ARG COMMITHASH
+ENV COMMITHASH ${COMMITHASH}
+
+CMD [ \
+    "/usr/local/bin/run" \
+    ]

core/api/BUCK

@@ -63,7 +63,8 @@ filegroup(
   name = "protos",
   srcs = glob([
     "src/services/**/protos/**",
-    "src/services/**/proto/**"
+    "src/services/**/proto/**",
+    "src/servers/**/proto/**",
   ])
 )
 
@@ -287,6 +288,14 @@ dev_pnpm_task_binary(
   visibility = ["PUBLIC"],
 )
 
+dev_pnpm_task_binary(
+  name = "dev-transactions-grpc-stream",
+  command = "dev:api-transactions-grpc-stream",
+  srcs = [":src"],
+  deps = ["//:node_modules"],
+  visibility = ["PUBLIC"],
+)
+
 dev_pnpm_task_binary(
   name = "dev-cron",
   command = "dev:cron",

core/api/package.json

@@ -5,7 +5,7 @@
     "eslint-check": "eslint src test --ext .ts",
     "eslint-fix": "eslint src test --ext .ts --fix",
     "circular-deps-check": "madge --circular --extensions ts src",
-    "build": "tsc -p tsconfig-build.json && cp -R src/services/price/protos dist/services/price/ && cp -R src/services/dealer-price/proto dist/services/dealer-price/ && cp -R src/services/bria/proto dist/services/bria/ && cp -R src/services/notifications/proto dist/services/notifications/ && cp -R src/services/api-keys/proto dist/services/api-keys/ && tscpaths --silent -p tsconfig.json -s ./src -o ./dist",
+    "build": "tsc -p tsconfig-build.json && cp -R src/services/price/protos dist/services/price/ && cp -R src/services/dealer-price/proto dist/services/dealer-price/ && cp -R src/services/bria/proto dist/services/bria/ && cp -R src/services/notifications/proto dist/services/notifications/ && cp -R src/services/api-keys/proto dist/services/api-keys/ && cp -R src/servers/transactions-grpc-stream/proto dist/servers/transactions-grpc-stream/ && tscpaths --silent -p tsconfig.json -s ./src -o ./dist",
     "trigger": "pnpm run build && node dist/servers/trigger.js | pino-pretty -c -l",
     "ws": "pnpm run build && node dist/servers/ws-server.js | pino-pretty -c -l",
     "watch": "nodemon -V -e ts,graphql -w ./src -x pnpm run start",
@@ -26,14 +26,16 @@
     "dev:api-trigger": "tsx src/servers/trigger.ts",
     "dev:api-exporter": "tsx src/servers/exporter.ts",
     "dev:api-ws-server": "tsx src/servers/ws-server.ts",
+    "dev:api-transactions-grpc-stream": "tsx src/servers/transactions-grpc-stream-server.ts",
     "dev:cron": "tsx src/servers/cron.ts",
     "migrate:create": "migrate-mongo create $MIGRATION_NAME -f src/migrations/migrate-mongo-config.js",
     "migrate:status": "migrate-mongo status -f src/migrations/migrate-mongo-config.js",
     "migrate:up": "migrate-mongo up -f src/migrations/migrate-mongo-config.js",
     "migrate:down": "migrate-mongo down -f src/migrations/migrate-mongo-config.js",
     "mongodb-migrate": "pnpm run migrate:status && pnpm run migrate:up && pnpm run migrate:status",
     "codegen:notifications": "cd ./src/services/notifications/proto && buf generate",
-    "codegen:api-keys": "cd ./src/services/api-keys/proto && buf generate"
+    "codegen:api-keys": "cd ./src/services/api-keys/proto && buf generate",
+    "codegen:transactions-stream": "cd ./src/servers/transactions-grpc-stream/proto && buf generate"
   },
   "engines": {
     "node": "20"