feat: add new market data socket with coincap provider for live price updates #1153
Conversation
📝 WalkthroughWalkthroughAdds a websocket-based market-data subsystem (connection handler, CoinCap client), a websocket base package with heartbeat/routing, a new @shapeshiftoss/prometheus package, updates many coinstack apps to import Prometheus from the new package, and wires WS support into the proxy API using COINCAP_API_KEY. Changes
Sequence Diagram(s)sequenceDiagram
autonumber
participant Client as Client WS
participant API as Proxy API (HTTP + WS)
participant Conn as MarketDataConnectionHandler
participant Coincap as CoincapWebsocketClient
participant Prom as Prometheus
Client->>API: WS upgrade to market-data endpoint
API->>Conn: MarketDataConnectionHandler.start(ws, coincapClient, prometheus, logger)
Conn->>Coincap: subscribe(clientId, subscriptionId, assets)
alt first subscription for provider
Coincap->>Coincap: connect(provider URL with API key)
end
Coincap-->>Conn: price update (raw)
Conn->>Client: publish(subscriptionId, price_update filtered)
Conn->>Prom: update websocket metrics (connect/close/message)
Client->>Conn: unsubscribe / close
Conn->>Coincap: unsubscribe(clientId, subscriptionId)
alt no remaining subscriptions
Coincap->>Coincap: close connection
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Poem
✨ Finishing Touches
🧪 Generate unit tests
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
SupportNeed help? Create a ticket on our support page for assistance with any issues or questions. CodeRabbit Commands (Invoked using PR/Issue comments)Type Other keywords and placeholders
CodeRabbit Configuration File (
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 12
📜 Review details
Configuration used: CodeRabbit UI
Review profile: ASSERTIVE
Plan: Pro
💡 Knowledge Base configuration:
- MCP integration is disabled by default for public repositories
- Jira integration is disabled by default for public repositories
- Linear integration is disabled by default for public repositories
You can enable these sources in your CodeRabbit configuration.
📒 Files selected for processing (2)
node/proxy/api/src/app.ts(2 hunks)node/proxy/api/src/marketData.ts(1 hunks)
🧰 Additional context used
🧬 Code Graph Analysis (2)
node/proxy/api/src/marketData.ts (2)
go/pkg/api/middleware.go (1)
Logger(30-63)node/proxy/api/src/app.ts (1)
logger(14-17)
node/proxy/api/src/app.ts (1)
node/proxy/api/src/marketData.ts (1)
MarketDataWebSocket(46-291)
🔇 Additional comments (1)
node/proxy/api/src/marketData.ts (1)
240-263: Parse-assets behavior: empty assets means no updates.This is fine if intentional. If the desired behavior is “send all” when assets is omitted, we can easily switch to that. Confirm product intent.
I can update parseAssetsFromUrl to return a sentinel like ['ALL'] and alter broadcastToClients accordingly if needed.
node/proxy/api/src/app.ts
Outdated
| const marketDataWS = new MarketDataWebSocket(logger, process.env.MARKET_DATA_PROVIDER, process.env.MARKET_DATA_API_KEY) | ||
| marketDataWS.setupWebSocketServer(server) | ||
|
|
There was a problem hiding this comment.
🧹 Nitpick (assertive)
Ensure the ws.Server is closed during shutdown (or let MarketDataWebSocket own it).
Right now you don’t hold the ws server reference, so you can’t call close() on it. Either:
- Accept the class-side change I suggested (MarketDataWebSocket owns and closes the server), or
- Store the ws server here and close it in the signal handlers.
If you choose the app-side approach, apply this diff:
-const marketDataWS = new MarketDataWebSocket(logger, process.env.MARKET_DATA_PROVIDER, process.env.MARKET_DATA_API_KEY)
-marketDataWS.setupWebSocketServer(server)
+const marketDataWS = new MarketDataWebSocket(
+ logger,
+ process.env.MARKET_DATA_PROVIDER,
+ process.env.MARKET_DATA_API_KEY
+)
+const wsServer = marketDataWS.setupWebSocketServer(server) marketDataWS.disconnect()
- server.close(() => {
+ wsServer.close(() => {
+ server.close(() => {
logger.info('Server closed')
process.exit(0)
- })
+ })
+ })And similarly in the SIGINT handler.
Also applies to: 59-66, 68-75
🤖 Prompt for AI Agents
In node/proxy/api/src/app.ts around lines 56-58 (and similarly for 59-66,
68-75), you're not retaining the ws.Server reference so you can't close it on
shutdown; capture the WebSocket server instance when you create/setup it (e.g.,
const marketDataWSServer = marketDataWS.setupWebSocketServer(server) or have
setupWebSocketServer return the ws.Server) and store it in a variable at module
scope, then in your SIGINT/SIGTERM handlers call
marketDataWSServer.close(callback) (handle/ log errors in the callback and
ensure the process exits after close); if you prefer the alternative, move
ownership into MarketDataWebSocket so it internally creates and closes the
ws.Server on shutdown and update signal handlers accordingly — apply the same
change for the other ranges mentioned.
node/proxy/api/src/marketData.ts
Outdated
| @@ -0,0 +1,291 @@ | |||
| import { Logger } from '@shapeshiftoss/logger' | |||
| import WebSocket, { Server } from 'ws' | |||
There was a problem hiding this comment.
💡 Verification agent
🧩 Analysis chain
Use WebSocketServer import for ws v8+ (and align types).
Most modern installs use ws v8+, where the server class is exported as WebSocketServer (not Server). Your current import may fail type-check or at runtime depending on ws version.
Apply this diff:
-import WebSocket, { Server } from 'ws'
+import WebSocket, { WebSocketServer } from 'ws'And update the server setup signature/instantiation:
- setupWebSocketServer(server: HttpServer, path = '/market-data'): Server {
- const wsServer = new Server({ server, path })
+ setupWebSocketServer(server: HttpServer, path = '/market-data'): WebSocketServer {
+ const wsServer = new WebSocketServer({ server, path })Optionally verify the installed ws version (expect >= 8 for WebSocketServer):
Also applies to: 265-267
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Find package.json under node/proxy/api and show the ws version if present
API_PKG=$(fd -a 'package.json' node/proxy/api | head -n1)
if [[ -z "${API_PKG:-}" ]]; then
echo "node/proxy/api/package.json not found"
exit 1
fi
echo "ws version in node/proxy/api/package.json:"
jq -r '.dependencies.ws // .devDependencies.ws // "not declared"' "$API_PKG"Length of output: 406
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "Searching for ws declarations in all package.json files..."
mapfile -t pkgs < <(fd -a 'package.json' .)
for pkg in "${pkgs[@]}"; do
version=$(jq -r '.dependencies.ws // .devDependencies.ws // empty' "$pkg")
if [[ -n "$version" ]]; then
echo "$pkg: $version"
fi
doneLength of output: 297
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "Scanning all package.json files (excluding node_modules) for ws dependency versions..."
find . -type f -name package.json -not -path "*/node_modules/*" | while read -r pkg; do
version=$(jq -r '.dependencies.ws // .devDependencies.ws // empty' "$pkg")
if [[ -n "$version" ]]; then
echo "$pkg: $version"
fi
doneLength of output: 6030
Use WebSocketServer import for ws v8+ (and update setup signature)
Our monorepo consistently uses ws v8+ (see ^8.15.0 in several packages), where the server class is exported as WebSocketServer rather than Server. Continue using v8+ here to avoid type‐ or runtime mismatches.
– In node/proxy/api/src/marketData.ts, change:
-import WebSocket, { Server } from 'ws'
+import WebSocket, { WebSocketServer } from 'ws'– Around lines 265–267, update the method signature and instantiation:
- setupWebSocketServer(server: HttpServer, path = '/market-data'): Server {
- const wsServer = new Server({ server, path })
+ setupWebSocketServer(server: HttpServer, path = '/market-data'): WebSocketServer {
+ const wsServer = new WebSocketServer({ server, path })– Verify (or add) in node/proxy/api/package.json:
"dependencies": {
"ws": "^8.15.0"
}🤖 Prompt for AI Agents
In node/proxy/api/src/marketData.ts around line 2 and around lines 265–267, the
file imports Server from 'ws' and uses the older setup signature; change the
import to use WebSocketServer (import WebSocket, { WebSocketServer } from 'ws')
and update the setup/signature and instantiation at ~265–267 to use
WebSocketServer (and the newer constructor/method signature used by ws v8+).
Also ensure node/proxy/api/package.json lists "ws": "^8.15.0" in dependencies so
the v8+ types/runtime are used.
node/proxy/api/src/marketData.ts
Outdated
| } | ||
|
|
||
| setupWebSocketServer(server: HttpServer, path = '/market-data'): Server { | ||
| const wsServer = new Server({ server, path }) |
There was a problem hiding this comment.
🧹 Nitpick (assertive)
Constrain client message size to mitigate DoS (wsServer option).
By default, ws allows large payloads. Set a reasonable maxPayload to protect your server from oversized messages.
Apply this diff:
- const wsServer = new Server({ server, path })
+ const wsServer = new WebSocketServer({ server, path, maxPayload: 1 * 1024 * 1024 }) // 1 MiBCommittable suggestion skipped: line range outside the PR's diff.
🤖 Prompt for AI Agents
In node/proxy/api/src/marketData.ts around line 266 the WebSocket Server is
created without a maxPayload setting, leaving it vulnerable to oversized client
messages; update the Server instantiation to include a sensible maxPayload
option (for example 64*1024 bytes or another limit appropriate for your message
sizes) so the server rejects messages larger than that, and ensure any related
tests or consumers handle potential payload rejections gracefully.
|
Looks good overall. Some suggestions:
I will push up a draft for reference |
|
@kaladinlight wow this is super extensive thanks for that. Cleanup looks good as well. I'll test this out and try and build on top the asset selection Edit: I've pulled in your PR and implemented the client subscription and asset tracking. How does that look? |
premiumjibles
force-pushed
the
10283_coincap_poc
branch
from
75c16ef to
efe8cbd
Compare
There was a problem hiding this comment.
Actionable comments posted: 8
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
node/coinstacks/common/api/src/middleware.ts (1)
61-68: Reduce Prometheus label cardinality; avoid querystrings and cast statusCode to string.Using
originalUrl/urlincludes query params, exploding series cardinality. Preferreq.pathand pass label values as strings.Apply:
- res.on('finish', () => { - const route = req.originalUrl ?? req.url + res.on('finish', () => { + const route = req.path if (!route.startsWith('/api/v1/') || res.statusCode === 404) return - prometheus.metrics.httpRequestCounter.inc({ method: req.method, route, statusCode: res.statusCode }, 1) - end({ method: req.method, route, statusCode: res.statusCode }) + const labels = { method: req.method, route, statusCode: String(res.statusCode) } + prometheus.metrics.httpRequestCounter.inc(labels, 1) + end(labels) })
🧹 Nitpick comments (8)
node/coinstacks/solana/api/src/app.ts (2)
77-78: Include port in startup log for quicker ops triage.-const server = app.listen(PORT, () => logger.info('Server started')) +const server = app.listen(PORT, () => logger.info({ port: PORT }, 'Server started'))
72-75: Renameheliusto provider-agnosticwsClientfor clarity.The variable name implies a specific provider; the type is generic.
-const helius = new WebsocketClient(WS_URL, { +const wsClient = new WebsocketClient(WS_URL, { apiKey: WS_API_KEY, transactionHandler: registry.onTransaction.bind(registry), }) ... -wsServer.on('connection', (connection) => { - ConnectionHandler.start(connection, registry, helius, prometheus, logger) -}) +wsServer.on('connection', (connection) => { + ConnectionHandler.start(connection, registry, wsClient, prometheus, logger) +})Also applies to: 81-81
node/coinstacks/common/api/src/middleware.ts (3)
52-54: Align logger filter with metrics and use path-based check.Minor inconsistency: logger checks
'/api/v1'vs metrics'/api/v1/'. Usereq.pathand the same prefix to avoid accidental mismatches.-export const requestLogger = morgan('short', { - skip: (req, res) => !req.url?.startsWith('/api/v1') || res.statusCode === 404, -}) +export const requestLogger = morgan('short', { + skip: (req, res) => !req.path?.startsWith('/api/v1/') || res.statusCode === 404, +})
56-71: Option: account for aborted requests (client disconnects).Only listening to
finishmisses aborted responses. If you care about those, handlecloseonce and guard against double-recording.export const metrics = (prometheus: Prometheus) => (req: Request, res: Response, next: NextFunction): void => { const end = prometheus.metrics.httpRequestDurationSeconds.startTimer() - res.on('finish', () => { + let recorded = false + const record = () => { + if (recorded) return + recorded = true const route = req.path if (!route.startsWith('/api/v1/') || res.statusCode === 404) return - prometheus.metrics.httpRequestCounter.inc({ method: req.method, route, statusCode: res.statusCode }, 1) - end({ method: req.method, route, statusCode: res.statusCode }) - }) + const labels = { method: req.method, route, statusCode: String(res.statusCode) } + prometheus.metrics.httpRequestCounter.inc(labels, 1) + end(labels) + } + res.once('finish', record) + res.once('close', record) next() }
21-25: Preferinstanceoffor custom error detection.If
ApiErroris a class,instanceof ApiErroris clearer and safer than comparing constructor names.- if (err.constructor.name === ApiError.prototype.constructor.name) { + if (err instanceof ApiError) { const e = err as ApiError console.error(e) return res.status(e.statusCode ?? 500).json({ message: e.message }) }node/packages/websocket/src/connectionHandler.ts (1)
51-59: Make close path idempotent to prevent double cleanup on error→closeonerror calls close(), and onclose also calls close(). Guard to avoid invoking onClose twice.
export abstract class BaseConnectionHandler { public readonly clientId: string @@ private readonly pingIntervalMs = 10000 + private closed = false @@ private close(interval: NodeJS.Timeout): void { - this.pingTimeout && clearTimeout(this.pingTimeout) - clearInterval(interval) - this.onClose() - this.subscriptionIds.clear() + if (this.closed) return + this.closed = true + if (this.pingTimeout) clearTimeout(this.pingTimeout) + clearInterval(interval) + this.onClose() + this.subscriptionIds.clear() }node/proxy/api/src/coincap.ts (2)
10-12: Preserve apiKey in base args if neededYou’re dropping apiKey from Args when passing to super. If BaseWebsocketClient uses it (e.g., auth headers), forward it.
- super(url, { logger: args.logger }, opts) + super(url, { logger: args.logger, apiKey: args.apiKey }, opts)
30-37: Avoid console.log in library codeUse the injected logger for consistency and to respect log levels.
(Handled in the larger diff above.)
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
💡 Knowledge Base configuration:
- MCP integration is disabled by default for public repositories
- Jira integration is disabled by default for public repositories
- Linear integration is disabled by default for public repositories
You can enable these sources in your CodeRabbit configuration.
📒 Files selected for processing (47)
node/coinstacks/arbitrum-nova/api/package.json(1 hunks)node/coinstacks/arbitrum-nova/api/src/app.ts(1 hunks)node/coinstacks/arbitrum/api/package.json(1 hunks)node/coinstacks/arbitrum/api/src/app.ts(1 hunks)node/coinstacks/avalanche/api/package.json(1 hunks)node/coinstacks/avalanche/api/src/app.ts(1 hunks)node/coinstacks/base/api/package.json(1 hunks)node/coinstacks/base/api/src/app.ts(1 hunks)node/coinstacks/bitcoin/api/package.json(1 hunks)node/coinstacks/bitcoin/api/src/app.ts(1 hunks)node/coinstacks/bitcoincash/api/package.json(1 hunks)node/coinstacks/bitcoincash/api/src/app.ts(1 hunks)node/coinstacks/bnbsmartchain/api/package.json(1 hunks)node/coinstacks/bnbsmartchain/api/src/app.ts(1 hunks)node/coinstacks/common/api/package.json(1 hunks)node/coinstacks/common/api/src/index.ts(0 hunks)node/coinstacks/common/api/src/middleware.ts(1 hunks)node/coinstacks/common/api/src/registry.ts(1 hunks)node/coinstacks/common/api/src/websocket.ts(1 hunks)node/coinstacks/dogecoin/api/package.json(1 hunks)node/coinstacks/dogecoin/api/src/app.ts(1 hunks)node/coinstacks/ethereum/api/package.json(1 hunks)node/coinstacks/ethereum/api/src/app.ts(1 hunks)node/coinstacks/gnosis/api/package.json(1 hunks)node/coinstacks/gnosis/api/src/app.ts(1 hunks)node/coinstacks/litecoin/api/package.json(1 hunks)node/coinstacks/litecoin/api/src/app.ts(1 hunks)node/coinstacks/optimism/api/package.json(1 hunks)node/coinstacks/optimism/api/src/app.ts(1 hunks)node/coinstacks/polygon/api/package.json(1 hunks)node/coinstacks/polygon/api/src/app.ts(1 hunks)node/coinstacks/solana/api/package.json(1 hunks)node/coinstacks/solana/api/src/app.ts(1 hunks)node/coinstacks/solana/api/src/websocket.ts(4 hunks)node/packages/blockbook/src/websocket.ts(4 hunks)node/packages/prometheus/package.json(1 hunks)node/packages/prometheus/src/index.ts(1 hunks)node/packages/prometheus/tsconfig.json(1 hunks)node/packages/websocket/package.json(1 hunks)node/packages/websocket/src/connectionHandler.ts(1 hunks)node/packages/websocket/src/index.ts(1 hunks)node/packages/websocket/src/websocket.ts(3 hunks)node/proxy/api/src/app.ts(2 hunks)node/proxy/api/src/coincap.ts(1 hunks)node/proxy/api/src/marketData.ts(1 hunks)node/proxy/sample.env(1 hunks)package.json(0 hunks)
💤 Files with no reviewable changes (2)
- node/coinstacks/common/api/src/index.ts
- package.json
🚧 Files skipped from review as they are similar to previous changes (38)
- node/coinstacks/bnbsmartchain/api/package.json
- node/coinstacks/ethereum/api/src/app.ts
- node/coinstacks/gnosis/api/package.json
- node/packages/websocket/package.json
- node/coinstacks/bitcoincash/api/package.json
- node/packages/websocket/src/index.ts
- node/coinstacks/bitcoin/api/src/app.ts
- node/coinstacks/avalanche/api/src/app.ts
- node/coinstacks/litecoin/api/package.json
- node/packages/prometheus/tsconfig.json
- node/coinstacks/litecoin/api/src/app.ts
- node/coinstacks/dogecoin/api/package.json
- node/coinstacks/gnosis/api/src/app.ts
- node/coinstacks/ethereum/api/package.json
- node/coinstacks/common/api/src/registry.ts
- node/coinstacks/bnbsmartchain/api/src/app.ts
- node/coinstacks/dogecoin/api/src/app.ts
- node/coinstacks/arbitrum-nova/api/package.json
- node/packages/prometheus/package.json
- node/coinstacks/optimism/api/package.json
- node/coinstacks/polygon/api/package.json
- node/coinstacks/base/api/src/app.ts
- node/coinstacks/common/api/package.json
- node/coinstacks/polygon/api/src/app.ts
- node/coinstacks/optimism/api/src/app.ts
- node/coinstacks/solana/api/package.json
- node/coinstacks/arbitrum/api/package.json
- node/packages/prometheus/src/index.ts
- node/proxy/sample.env
- node/proxy/api/src/marketData.ts
- node/coinstacks/solana/api/src/websocket.ts
- node/coinstacks/bitcoincash/api/src/app.ts
- node/packages/websocket/src/websocket.ts
- node/coinstacks/base/api/package.json
- node/coinstacks/arbitrum-nova/api/src/app.ts
- node/packages/blockbook/src/websocket.ts
- node/proxy/api/src/app.ts
- node/coinstacks/arbitrum/api/src/app.ts
🧰 Additional context used
🧠 Learnings (2)
📚 Learning: 2025-08-20T06:35:01.854Z
Learnt from: premiumjibles
PR: shapeshift/unchained#1153
File: node/proxy/api/src/marketData.ts:33-33
Timestamp: 2025-08-20T06:35:01.854Z
Learning: premiumjibles confirmed that CoinCap v3 uses wss.coincap.io as the host for WebSocket connections, not ws.coincap.io as in previous versions.
Applied to files:
node/proxy/api/src/coincap.ts
📚 Learning: 2025-08-20T06:35:03.599Z
Learnt from: premiumjibles
PR: shapeshift/unchained#1153
File: node/proxy/api/src/marketData.ts:104-107
Timestamp: 2025-08-20T06:35:03.599Z
Learning: For WebSocket integrations with specific financial data APIs like CoinCap, the message payloads are typically JSON strings, so simple .toString() handling is usually sufficient rather than comprehensive type checking for all possible ws message payload types.
Applied to files:
node/proxy/api/src/coincap.ts
🧬 Code graph analysis (3)
node/packages/websocket/src/connectionHandler.ts (1)
node/packages/prometheus/src/prometheus.ts (1)
Prometheus(7-35)
node/proxy/api/src/coincap.ts (2)
node/proxy/api/src/marketData.ts (3)
MarketDataClient(22-30)MarketDataConnectionHandler(32-83)MarketDataMessage(7-12)node/packages/websocket/src/websocket.ts (2)
Args(16-19)Options(21-25)
node/coinstacks/common/api/src/websocket.ts (2)
node/coinstacks/common/api/src/registry.ts (1)
Registry(25-161)node/packages/prometheus/src/prometheus.ts (1)
Prometheus(7-35)
🪛 Biome (2.1.2)
node/packages/websocket/src/connectionHandler.ts
[error] 87-87: The function should not return a value because its return type is void.
The function is here:
'void' signals the absence of value. The returned value is likely to be ignored by the caller.
(lint/correctness/noVoidTypeReturn)
[error] 90-90: The function should not return a value because its return type is void.
The function is here:
'void' signals the absence of value. The returned value is likely to be ignored by the caller.
(lint/correctness/noVoidTypeReturn)
[error] 92-92: The function should not return a value because its return type is void.
The function is here:
'void' signals the absence of value. The returned value is likely to be ignored by the caller.
(lint/correctness/noVoidTypeReturn)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (1)
- GitHub Check: check
🔇 Additional comments (3)
node/coinstacks/solana/api/src/app.ts (1)
3-3: Prometheus import migration approved. Confirmed solana/api depends on @shapeshiftoss/prometheus@^10.0.0, no lingering @shapeshiftoss/common-api Prometheus imports, and the new package exports Prometheus.node/coinstacks/avalanche/api/package.json (1)
16-17: Prometheus migration dependency added — approve; confirm no lingering common-api Prometheus usageChange matches the migration pattern. Ensure code imports reference @shapeshiftoss/prometheus and not @shapeshiftoss/common-api.
Use the verification script provided in the bitcoin-api comment; it covers this package as well.
node/coinstacks/common/api/src/websocket.ts (1)
88-95: LGTM on cleanup flowUnsubscribing all and resubscribing aggregate addresses on close is consistent with the registry model.
| "dependencies": { | ||
| "@shapeshiftoss/blockbook": "^10.0.0", | ||
| "@shapeshiftoss/common-api": "^10.0.0", | ||
| "@shapeshiftoss/prometheus": "^10.0.0", |
There was a problem hiding this comment.
💡 Verification agent
🧩 Analysis chain
Prometheus package addition — looks good; verify imports and lockfile consistency
Dependency aligns with the repo-wide migration and local versioning (^10.0.0). Please ensure all imports moved off @shapeshiftoss/common-api and that no direct prom-client usages remain.
Run to verify consistency across the repo:
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "1) Check TypeScript imports switched to @shapeshiftoss/prometheus"
rg -nP --type=ts -C1 $'from\\s+[\'"]@shapeshiftoss/(common-api|prometheus)[\'"]'
echo; echo "2) Find any remaining direct prom-client usages (should be none)"
rg -nP --type=ts -C2 '\bprom-client\b' || true
echo; echo "3) Ensure all package.jsons depend on @shapeshiftoss/prometheus and not prom-client"
rg -nP --type=json '"@shapeshiftoss/prometheus"\\s*:\\s*"[^^]?10\\.' -n || true
rg -nP --type=json '"prom-client"\\s*:' -n || true
echo; echo "4) Optional: show duplicated versions of @shapeshiftoss/prometheus (should be a single range)"
jq -r '..|objects|select(has("@shapeshiftoss/prometheus"))["@shapeshiftoss/prometheus"]' **/package.json 2>/dev/null | sort | uniq -cLength of output: 11701
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "1) Versions of @shapeshiftoss/prometheus in all package.json under node/"
PKGS=$(find node -type f -name package.json)
jq -r '..|objects|select(has("@shapeshiftoss/prometheus"))["@shapeshiftoss/prometheus"]' $PKGS | sort | uniq -c || true
echo
echo "2) Check lockfiles in repo root for @shapeshiftoss/prometheus entries"
for lock in yarn.lock package-lock.json pnpm-lock.yaml; do
if [ -f "$lock" ]; then
echo "Lockfile: $lock"
grep -n "@shapeshiftoss/prometheus@" "$lock" || echo "No entries found"
echo
fi
doneLength of output: 3778
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "1) List all lockfiles in the repository"
find . -type f \( -name yarn.lock -o -name package-lock.json -o -name pnpm-lock.yaml \)
echo
echo "2) Search each lockfile for @shapeshiftoss/prometheus entries"
find . -type f \( -name yarn.lock -o -name package-lock.json -o -name pnpm-lock.yaml \) | while read -r lock; do
echo "---- $lock ----"
grep -n "@shapeshiftoss/prometheus@" "$lock" || echo "No entries found"
echo
doneLength of output: 888
Update root lockfile with new dependency
No entries for @shapeshiftoss/prometheus found in the root yarn.lock—run yarn install (or your package manager) and commit the updated lockfile.
🤖 Prompt for AI Agents
In node/coinstacks/bitcoin/api/package.json around line 17, the new dependency
"@shapeshiftoss/prometheus" was added but the repository root lockfile wasn't
updated; run your project's package manager (e.g., from repo root run `yarn
install` or the equivalent), verify that the root yarn.lock (or lockfile used by
the repo) now contains entries for @shapeshiftoss/prometheus, and then commit
the updated lockfile alongside this change so CI and other developers use the
same resolved dependency tree.
| @@ -1,10 +1,10 @@ | |||
| import { Prometheus } from '@shapeshiftoss/prometheus' | |||
There was a problem hiding this comment.
💡 Verification agent
🧩 Analysis chain
Shared Prometheus import: looks good; verify dependency wiring.
Confirm that all workspaces/packages importing @shapeshiftoss/prometheus declare it in their package.json and that no stale local ./prometheus imports remain.
Run to verify:
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "Calls importing @shapeshiftoss/prometheus:"
rg -nP --type=ts -C1 "from\s+['\"]@shapeshiftoss/prometheus['\"]"
echo -e "\nResidual local imports (should be none):"
rg -nP --type=ts -C1 "from\s+['\"]\./prometheus['\"]" || true
echo -e "\nChecking dependent package.json files for declared dependency:"
while IFS= read -r pkg; do
dir="$(dirname "$pkg")"
if rg -nP "from\s+['\"]@shapeshiftoss/prometheus['\"]" "$dir"/**/*.ts >/dev/null 2>&1; then
if ! jq -er '(.dependencies["@shapeshiftoss/prometheus"] // .devDependencies["@shapeshiftoss/prometheus"])' "$pkg" >/dev/null; then
echo "Missing @shapeshiftoss/prometheus in $pkg"
fi
fi
done < <(fd -t f package.json)Length of output: 5993
🏁 Script executed:
#!/usr/bin/env bash
set -euo pipefail
echo "Residual local imports (excluding Prometheus package):"
grep -R -nP "from\s+['\"]\./prometheus['\"]" node --include '*.ts' | grep -v "node/packages/prometheus/src/index.ts" || echo "None found"
echo -e "\nChecking package.json for @shapeshiftoss/prometheus declarations:"
for pkg in $(fd -t f package.json); do
dir=$(dirname "$pkg")
if grep -R -nP "from\s+['\"]@shapeshiftoss/prometheus['\"]" "$dir" --include '*.ts' >/dev/null; then
if ! jq -e '.dependencies["@shapeshiftoss/prometheus"] or .devDependencies["@shapeshiftoss/prometheus"]' "$pkg" >/dev/null; then
echo "Missing @shapeshiftoss/prometheus in $pkg"
fi
fi
doneLength of output: 12226
Add missing Prometheus dependency in node/proxy/api
node/proxy/api/src imports { Prometheus } from @shapeshiftoss/prometheus, but node/proxy/api/package.json doesn’t list it in dependencies. No residual ./prometheus imports remain.
🤖 Prompt for AI Agents
In node/coinstacks/common/api/src/middleware.ts at line 1: the file imports {
Prometheus } from '@shapeshiftoss/prometheus' but node/proxy/api/package.json
does not list that package as a dependency; update node/proxy/api/package.json
to add '@shapeshiftoss/prometheus' under "dependencies" with a suitable version
(match workspace root or the version used elsewhere), run the package manager
(yarn/npm install) to update lockfile, and ensure there are no leftover local
'./prometheus' imports anywhere in the proxy/api codebase.
| // eslint-disable-next-line @typescript-eslint/no-explicit-any | ||
| function isTxsTopicData(data: any): data is TxsTopicData { | ||
| return data && 'topic' in data | ||
| } |
There was a problem hiding this comment.
Strengthen the topic type guard
Current check only verifies existence of “topic” and can misroute. Ensure topic === 'txs' and addresses shape is valid.
-// eslint-disable-next-line @typescript-eslint/no-explicit-any
-function isTxsTopicData(data: any): data is TxsTopicData {
- return data && 'topic' in data
-}
+// eslint-disable-next-line @typescript-eslint/no-explicit-any
+function isTxsTopicData(data: any): data is TxsTopicData {
+ return (
+ data &&
+ typeof data === 'object' &&
+ data.topic === 'txs' &&
+ Array.isArray(data.addresses) &&
+ data.addresses.every((a: unknown) => typeof a === 'string')
+ )
+}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| // eslint-disable-next-line @typescript-eslint/no-explicit-any | |
| function isTxsTopicData(data: any): data is TxsTopicData { | |
| return data && 'topic' in data | |
| } | |
| // eslint-disable-next-line @typescript-eslint/no-explicit-any | |
| function isTxsTopicData(data: any): data is TxsTopicData { | |
| return ( | |
| data && | |
| typeof data === 'object' && | |
| data.topic === 'txs' && | |
| Array.isArray(data.addresses) && | |
| data.addresses.every((a: unknown) => typeof a === 'string') | |
| ) | |
| } |
🤖 Prompt for AI Agents
In node/coinstacks/common/api/src/websocket.ts around lines 21 to 24, the
current type guard only checks for existence of "topic" which can misroute
messages; update the guard to first ensure data is non-null/undefined, that
data.topic is a string and equals 'txs', and that data.addresses exists as an
array of strings (use Array.isArray and addresses.every(addr => typeof addr ===
'string') and optionally allow empty/non-empty depending on expected semantics).
Return true only if all those checks pass so the compiler can safely narrow to
TxsTopicData.
| onSubscribe(subscriptionId: string, data?: unknown): void { | ||
| if (!isTxsTopicData(data)) { | ||
| this.sendError(`no topic specified for subscribe`, subscriptionId) | ||
| return | ||
| } | ||
|
|
||
| this.pingTimeout = setTimeout(() => { | ||
| this.logger.debug({ clientId: this.clientId, fn: 'pingTimeout' }, 'heartbeat failed') | ||
| this.websocket.terminate() | ||
| }, this.pingIntervalMs + 1000) | ||
| const callback = this.routes[data.topic].subscribe | ||
| if (callback) { | ||
| callback(subscriptionId, data) | ||
| } else { | ||
| this.sendError(`subscribe method not implemented for topic: ${data.topic}`, subscriptionId) | ||
| } | ||
| } |
There was a problem hiding this comment.
Guard missing/unknown topics to avoid runtime exception
Accessing this.routes[data.topic].subscribe without checking route can throw.
onSubscribe(subscriptionId: string, data?: unknown): void {
- if (!isTxsTopicData(data)) {
+ if (!isTxsTopicData(data)) {
this.sendError(`no topic specified for subscribe`, subscriptionId)
return
}
- const callback = this.routes[data.topic].subscribe
- if (callback) {
- callback(subscriptionId, data)
- } else {
- this.sendError(`subscribe method not implemented for topic: ${data.topic}`, subscriptionId)
- }
+ const route = this.routes[data.topic as Topics]
+ if (!route || typeof route.subscribe !== 'function') {
+ this.sendError(`subscribe method not implemented for topic: ${data.topic}`, subscriptionId)
+ return
+ }
+ route.subscribe(subscriptionId, data)
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| onSubscribe(subscriptionId: string, data?: unknown): void { | |
| if (!isTxsTopicData(data)) { | |
| this.sendError(`no topic specified for subscribe`, subscriptionId) | |
| return | |
| } | |
| this.pingTimeout = setTimeout(() => { | |
| this.logger.debug({ clientId: this.clientId, fn: 'pingTimeout' }, 'heartbeat failed') | |
| this.websocket.terminate() | |
| }, this.pingIntervalMs + 1000) | |
| const callback = this.routes[data.topic].subscribe | |
| if (callback) { | |
| callback(subscriptionId, data) | |
| } else { | |
| this.sendError(`subscribe method not implemented for topic: ${data.topic}`, subscriptionId) | |
| } | |
| } | |
| onSubscribe(subscriptionId: string, data?: unknown): void { | |
| if (!isTxsTopicData(data)) { | |
| this.sendError(`no topic specified for subscribe`, subscriptionId) | |
| return | |
| } | |
| const route = this.routes[data.topic as Topics] | |
| if (!route || typeof route.subscribe !== 'function') { | |
| this.sendError(`subscribe method not implemented for topic: ${data.topic}`, subscriptionId) | |
| return | |
| } | |
| route.subscribe(subscriptionId, data) | |
| } |
🤖 Prompt for AI Agents
In node/coinstacks/common/api/src/websocket.ts around lines 60 to 72, the code
accesses this.routes[data.topic].subscribe without verifying that
this.routes[data.topic] exists; add a guard to fetch the route into a local
variable (e.g. const route = this.routes[data.topic]) and check route before
calling route.subscribe, sending an error via this.sendError when the topic is
unknown or the subscribe handler is missing; ensure you use that guarded
variable for both existence and calling to avoid a runtime exception.
| onUnsubscribe(subscriptionId: string, data?: unknown): void { | ||
| if (!isTxsTopicData(data)) { | ||
| this.sendError(`no topic specified for unsubscribe`, subscriptionId) | ||
| return | ||
| } | ||
|
|
||
| private onMessage(event: WebSocket.MessageEvent): void { | ||
| try { | ||
| const payload: RequestPayload = JSON.parse(event.data.toString()) | ||
|
|
||
| switch (payload.method) { | ||
| // browsers do not support ping/pong frame, handle message instead | ||
| case 'ping': { | ||
| this.websocket.send('pong') | ||
| break | ||
| } | ||
| case 'subscribe': | ||
| case 'unsubscribe': { | ||
| const topic = payload.data?.topic | ||
|
|
||
| if (!topic) { | ||
| this.sendError(`no topic specified for method: ${payload.method}`, payload.subscriptionId) | ||
| break | ||
| } | ||
|
|
||
| const callback = this.routes[topic][payload.method] | ||
| if (callback) { | ||
| callback(payload.subscriptionId, payload.data) | ||
| } else { | ||
| this.sendError(`${payload.method} method not implemented for topic: ${topic}`, payload.subscriptionId) | ||
| } | ||
| } | ||
| } | ||
| } catch (err) { | ||
| this.logger.error(err, { clientId: this.clientId, fn: 'onMessage', event }, 'Error processing message') | ||
| const callback = this.routes[data.topic].unsubscribe | ||
| if (callback) { | ||
| callback(subscriptionId, data) | ||
| } else { | ||
| this.sendError(`unsubscribe method not implemented for topic: ${data.topic}`, subscriptionId) | ||
| } | ||
| } |
There was a problem hiding this comment.
Mirror the same guard for unsubscribe
Same potential throw path on unknown topic.
onUnsubscribe(subscriptionId: string, data?: unknown): void {
- if (!isTxsTopicData(data)) {
+ if (!isTxsTopicData(data)) {
this.sendError(`no topic specified for unsubscribe`, subscriptionId)
return
}
- const callback = this.routes[data.topic].unsubscribe
- if (callback) {
- callback(subscriptionId, data)
- } else {
- this.sendError(`unsubscribe method not implemented for topic: ${data.topic}`, subscriptionId)
- }
+ const route = this.routes[data.topic as Topics]
+ if (!route || typeof route.unsubscribe !== 'function') {
+ this.sendError(`unsubscribe method not implemented for topic: ${data.topic}`, subscriptionId)
+ return
+ }
+ route.unsubscribe(subscriptionId, data)
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| onUnsubscribe(subscriptionId: string, data?: unknown): void { | |
| if (!isTxsTopicData(data)) { | |
| this.sendError(`no topic specified for unsubscribe`, subscriptionId) | |
| return | |
| } | |
| private onMessage(event: WebSocket.MessageEvent): void { | |
| try { | |
| const payload: RequestPayload = JSON.parse(event.data.toString()) | |
| switch (payload.method) { | |
| // browsers do not support ping/pong frame, handle message instead | |
| case 'ping': { | |
| this.websocket.send('pong') | |
| break | |
| } | |
| case 'subscribe': | |
| case 'unsubscribe': { | |
| const topic = payload.data?.topic | |
| if (!topic) { | |
| this.sendError(`no topic specified for method: ${payload.method}`, payload.subscriptionId) | |
| break | |
| } | |
| const callback = this.routes[topic][payload.method] | |
| if (callback) { | |
| callback(payload.subscriptionId, payload.data) | |
| } else { | |
| this.sendError(`${payload.method} method not implemented for topic: ${topic}`, payload.subscriptionId) | |
| } | |
| } | |
| } | |
| } catch (err) { | |
| this.logger.error(err, { clientId: this.clientId, fn: 'onMessage', event }, 'Error processing message') | |
| const callback = this.routes[data.topic].unsubscribe | |
| if (callback) { | |
| callback(subscriptionId, data) | |
| } else { | |
| this.sendError(`unsubscribe method not implemented for topic: ${data.topic}`, subscriptionId) | |
| } | |
| } | |
| onUnsubscribe(subscriptionId: string, data?: unknown): void { | |
| if (!isTxsTopicData(data)) { | |
| this.sendError(`no topic specified for unsubscribe`, subscriptionId) | |
| return | |
| } | |
| const route = this.routes[data.topic as Topics] | |
| if (!route || typeof route.unsubscribe !== 'function') { | |
| this.sendError(`unsubscribe method not implemented for topic: ${data.topic}`, subscriptionId) | |
| return | |
| } | |
| route.unsubscribe(subscriptionId, data) | |
| } |
🤖 Prompt for AI Agents
In node/coinstacks/common/api/src/websocket.ts around lines 74 to 86, the
unsubscribe handler accesses this.routes[data.topic] without verifying the topic
exists which can throw for unknown topics; add the same guard used elsewhere:
after validating isTxsTopicData(data), check that this.routes[data.topic] is
defined and if not call sendError about unknown topic and return, otherwise read
the unsubscribe callback and invoke it or send the existing "method not
implemented" error.
| this.websocket.ping() | ||
|
|
||
| const pingInterval = setInterval(() => { | ||
| this.websocket.ping() | ||
| }, this.pingIntervalMs) | ||
|
|
There was a problem hiding this comment.
🛠️ Refactor suggestion
Start heartbeat immediately to avoid never-armed timeout
Without an initial heartbeat, a dead connection before the first pong won’t terminate. Arm the timeout once at construction.
this.websocket = websocket
- this.websocket.ping()
+ this.websocket.ping()
+ this.heartbeat()
const pingInterval = setInterval(() => {
this.websocket.ping()
}, this.pingIntervalMs)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| this.websocket.ping() | |
| const pingInterval = setInterval(() => { | |
| this.websocket.ping() | |
| }, this.pingIntervalMs) | |
| this.websocket = websocket | |
| this.websocket.ping() | |
| this.heartbeat() | |
| const pingInterval = setInterval(() => { | |
| this.websocket.ping() | |
| }, this.pingIntervalMs) |
🤖 Prompt for AI Agents
In node/packages/websocket/src/connectionHandler.ts around lines 45–50, the
heartbeat logic currently only starts the periodic ping interval which leaves
the timeout unarmed until the first interval tick; call the ping immediately at
construction (or right before installing the setInterval) so the timeout is
armed immediately, then install the setInterval to continue pings; ensure the
same timeout/ponthandling code used for interval pings is applied to this
initial ping (no extra second timer).
| private onMessage(event: WebSocket.MessageEvent): void { | ||
| try { | ||
| const payload: RequestPayload = JSON.parse(event.data.toString()) | ||
|
|
||
| switch (payload.method) { | ||
| // browsers do not support ping/pong frame, handle message instead | ||
| case 'ping': { | ||
| return this.websocket.send('pong') | ||
| } | ||
| case 'subscribe': | ||
| return this.onSubscribe(payload.subscriptionId, payload.data) | ||
| case 'unsubscribe': | ||
| return this.onUnsubscribe(payload.subscriptionId, payload.data) | ||
| } |
There was a problem hiding this comment.
Fix Biome errors: don’t return expressions from void methods
onMessage returns void but currently returns expressions in the switch cases. Replace with side-effect calls plus bare returns.
Apply:
private onMessage(event: WebSocket.MessageEvent): void {
try {
const payload: RequestPayload = JSON.parse(event.data.toString())
switch (payload.method) {
// browsers do not support ping/pong frame, handle message instead
case 'ping': {
- return this.websocket.send('pong')
+ this.websocket.send('pong')
+ return
}
case 'subscribe':
- return this.onSubscribe(payload.subscriptionId, payload.data)
+ this.onSubscribe(payload.subscriptionId, payload.data)
+ return
case 'unsubscribe':
- return this.onUnsubscribe(payload.subscriptionId, payload.data)
+ this.onUnsubscribe(payload.subscriptionId, payload.data)
+ return
}
} catch (err) {
this.logger.error(err, { clientId: this.clientId, fn: 'onMessage', event }, 'Error processing message')
}
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| private onMessage(event: WebSocket.MessageEvent): void { | |
| try { | |
| const payload: RequestPayload = JSON.parse(event.data.toString()) | |
| switch (payload.method) { | |
| // browsers do not support ping/pong frame, handle message instead | |
| case 'ping': { | |
| return this.websocket.send('pong') | |
| } | |
| case 'subscribe': | |
| return this.onSubscribe(payload.subscriptionId, payload.data) | |
| case 'unsubscribe': | |
| return this.onUnsubscribe(payload.subscriptionId, payload.data) | |
| } | |
| private onMessage(event: WebSocket.MessageEvent): void { | |
| try { | |
| const payload: RequestPayload = JSON.parse(event.data.toString()) | |
| switch (payload.method) { | |
| // browsers do not support ping/pong frame, handle message instead | |
| case 'ping': { | |
| this.websocket.send('pong') | |
| return | |
| } | |
| case 'subscribe': | |
| this.onSubscribe(payload.subscriptionId, payload.data) | |
| return | |
| case 'unsubscribe': | |
| this.onUnsubscribe(payload.subscriptionId, payload.data) | |
| return | |
| } | |
| } catch (err) { | |
| this.logger.error(err, { clientId: this.clientId, fn: 'onMessage', event }, 'Error processing message') | |
| } | |
| } |
🧰 Tools
🪛 Biome (2.1.2)
[error] 87-87: The function should not return a value because its return type is void.
The function is here:
'void' signals the absence of value. The returned value is likely to be ignored by the caller.
(lint/correctness/noVoidTypeReturn)
[error] 90-90: The function should not return a value because its return type is void.
The function is here:
'void' signals the absence of value. The returned value is likely to be ignored by the caller.
(lint/correctness/noVoidTypeReturn)
[error] 92-92: The function should not return a value because its return type is void.
The function is here:
'void' signals the absence of value. The returned value is likely to be ignored by the caller.
(lint/correctness/noVoidTypeReturn)
🤖 Prompt for AI Agents
In node/packages/websocket/src/connectionHandler.ts around lines 80 to 93, the
onMessage method is declared to return void but currently uses "return" with
expressions inside the switch cases; change each case to call the side-effect
and then use a bare "return" (e.g., replace "return this.websocket.send('pong')"
with "this.websocket.send('pong'); return;" and similarly replace "return
this.onSubscribe(...)" and "return this.onUnsubscribe(...)" with
"this.onSubscribe(...); return;" and "this.onUnsubscribe(...); return;"
respectively so the method returns void without returning expressions.
| private handleMessage(message: Record<string, string>): void { | ||
| for (const [clientId, client] of this.clients) { | ||
| try { | ||
| const payload: MarketDataMessage = { | ||
| type: 'price_update', | ||
| source: 'coincap', | ||
| data: message, | ||
| timestamp: Date.now(), | ||
| } | ||
|
|
||
| client.publish(clientId, payload) | ||
| } catch (error) { |
There was a problem hiding this comment.
Wrong identifier passed to publish: using clientId instead of subscriptionId
This breaks the client contract; published messages carry an unexpected subscriptionId. Track subscriptions per client and publish per subscription.
-import { BaseWebsocketClient, Args, Options, BaseConnectionHandler } from '@shapeshiftoss/websocket'
+import { BaseWebsocketClient, Args, Options } from '@shapeshiftoss/websocket'
@@
-export class CoincapWebsocketClient extends BaseWebsocketClient implements MarketDataClient {
- clients = new Map<string, BaseConnectionHandler>()
+type Client = {
+ connection: MarketDataConnectionHandler
+ // subscriptionId -> assets
+ subscriptions: Map<string, Array<string>>
+}
+
+export class CoincapWebsocketClient extends BaseWebsocketClient implements MarketDataClient {
+ clients = new Map<string, Client>()
@@
- subscribe(clientId: string, subscriptionId: string, connection: MarketDataConnectionHandler, assets: Array<string>) {
- console.log(this.clients.size, { subscriptionId, assets })
- if (!this.clients.size) this.connect()
- this.clients.set(clientId, connection)
+ subscribe(clientId: string, subscriptionId: string, connection: MarketDataConnectionHandler, assets: Array<string>) {
+ if (!this.clients.size) this.connect()
+ const existing = this.clients.get(clientId)
+ if (existing) {
+ existing.connection = connection
+ existing.subscriptions.set(subscriptionId, assets)
+ } else {
+ this.clients.set(clientId, { connection, subscriptions: new Map([[subscriptionId, assets]]) })
+ }
+ this.logger.debug({ clientId, subscriptionId, assets }, 'coincap subscribe')
}
@@
- unsubscribe(clientId: string, subscriptionId: string, assets: Array<string>) {
- console.log({ subscriptionId, assets })
- if (!this.clients.has(clientId)) return
- this.clients.delete(clientId)
- if (!this.clients.size) this.socket?.close(1000)
+ unsubscribe(clientId: string, subscriptionId: string, assets: Array<string>) {
+ const client = this.clients.get(clientId)
+ if (!client) return
+ if (subscriptionId) {
+ client.subscriptions.delete(subscriptionId)
+ if (!client.subscriptions.size) this.clients.delete(clientId)
+ } else {
+ // Unsubscribe all for this client
+ this.clients.delete(clientId)
+ }
+ if (!this.clients.size) this.socket?.close(1000)
+ this.logger.debug({ clientId, subscriptionId, assets }, 'coincap unsubscribe')
}
@@
- private handleMessage(message: Record<string, string>): void {
- for (const [clientId, client] of this.clients) {
- try {
- const payload: MarketDataMessage = {
- type: 'price_update',
- source: 'coincap',
- data: message,
- timestamp: Date.now(),
- }
-
- client.publish(clientId, payload)
- } catch (error) {
- this.logger.error({ clientId, error }, 'failed to handle message')
- }
- }
- }
+ private handleMessage(message: Record<string, string>): void {
+ for (const [clientId, { connection, subscriptions }] of this.clients) {
+ try {
+ for (const [subscriptionId, assets] of subscriptions) {
+ // Optional server-side filtering; keep ALL to broadcast everything for that sub
+ const filtered =
+ !assets?.length || assets.includes('ALL')
+ ? message
+ : Object.fromEntries(Object.entries(message).filter(([asset]) => assets.includes(asset)))
+ if (!Object.keys(filtered).length) continue
+ const payload: MarketDataMessage = {
+ type: 'price_update',
+ source: 'coincap',
+ data: filtered,
+ timestamp: Date.now(),
+ }
+ connection.publish(subscriptionId, payload)
+ }
+ } catch (error) {
+ this.logger.error({ clientId, error }, 'failed to handle message')
+ }
+ }
+ }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| private handleMessage(message: Record<string, string>): void { | |
| for (const [clientId, client] of this.clients) { | |
| try { | |
| const payload: MarketDataMessage = { | |
| type: 'price_update', | |
| source: 'coincap', | |
| data: message, | |
| timestamp: Date.now(), | |
| } | |
| client.publish(clientId, payload) | |
| } catch (error) { | |
| // At the top of node/proxy/api/src/coincap.ts | |
| import { BaseWebsocketClient, Args, Options } from '@shapeshiftoss/websocket' | |
| // A client holds its connection and a map of subscriptionId → asset filters | |
| type Client = { | |
| connection: MarketDataConnectionHandler | |
| subscriptions: Map<string, string[]> | |
| } | |
| // Within the same file... | |
| export class CoincapWebsocketClient extends BaseWebsocketClient implements MarketDataClient { | |
| clients = new Map<string, Client>() | |
| subscribe( | |
| clientId: string, | |
| subscriptionId: string, | |
| connection: MarketDataConnectionHandler, | |
| assets: string[], | |
| ) { | |
| - console.log(this.clients.size, { subscriptionId, assets }) | |
| - if (!this.clients.size) this.connect() | |
| if (!this.clients.size) this.connect() | |
| const existing = this.clients.get(clientId) | |
| if (existing) { | |
| existing.connection = connection | |
| existing.subscriptions.set(subscriptionId, assets) | |
| } else { | |
| this.clients.set(clientId, { | |
| connection, | |
| subscriptions: new Map([[subscriptionId, assets]]), | |
| }) | |
| } | |
| this.logger.debug({ clientId, subscriptionId, assets }, 'coincap subscribe') | |
| } | |
| unsubscribe( | |
| clientId: string, | |
| subscriptionId: string, | |
| assets: string[], | |
| ) { | |
| - console.log({ subscriptionId, assets }) | |
| - if (!this.clients.has(clientId)) return | |
| - this.clients.delete(clientId) | |
| const client = this.clients.get(clientId) | |
| if (!client) return | |
| if (subscriptionId) { | |
| client.subscriptions.delete(subscriptionId) | |
| if (!client.subscriptions.size) { | |
| this.clients.delete(clientId) | |
| } | |
| } else { | |
| // Remove all subscriptions for this client | |
| this.clients.delete(clientId) | |
| } | |
| if (!this.clients.size) { | |
| this.socket?.close(1000) | |
| } | |
| this.logger.debug({ clientId, subscriptionId, assets }, 'coincap unsubscribe') | |
| } | |
| - private handleMessage(message: Record<string, string>): void { | |
| - for (const [clientId, client] of this.clients) { | |
| - try { | |
| - const payload: MarketDataMessage = { | |
| - type: 'price_update', | |
| - source: 'coincap', | |
| - data: message, | |
| - timestamp: Date.now(), | |
| - } | |
| - | |
| - client.publish(clientId, payload) | |
| - } catch (error) { | |
| - this.logger.error({ clientId, error }, 'failed to handle message') | |
| - } | |
| - } | |
| private handleMessage(message: Record<string, string>): void { | |
| for (const [clientId, { connection, subscriptions }] of this.clients) { | |
| try { | |
| for (const [subscriptionId, assets] of subscriptions) { | |
| // Filter if specific assets requested (or broadcast all if 'ALL') | |
| const filtered = | |
| !assets?.length || assets.includes('ALL') | |
| ? message | |
| : Object.fromEntries( | |
| Object.entries(message).filter( | |
| ([asset]) => assets.includes(asset), | |
| ), | |
| ) | |
| if (!Object.keys(filtered).length) continue | |
| const payload: MarketDataMessage = { | |
| type: 'price_update', | |
| source: 'coincap', | |
| data: filtered, | |
| timestamp: Date.now(), | |
| } | |
| connection.publish(subscriptionId, payload) | |
| } | |
| } catch (error) { | |
| this.logger.error({ clientId, error }, 'failed to handle message') | |
| } | |
| } | |
| } | |
| } |
🤖 Prompt for AI Agents
In node/proxy/api/src/coincap.ts around lines 42 to 53, the publish call is
using clientId as the subscription identifier, which violates the client
contract; update the logic so you iterate the subscriptions for each client and
call client.publish with the correct subscriptionId instead of clientId. Ensure
you maintain/lookup a per-client subscription list (or map) and for each
subscriptionId call client.publish(subscriptionId, payload) inside the try
block; if the current clients map doesn't store subscriptions, extend its value
to include the subscription collection and adjust subscription add/remove code
accordingly.
aec767a to
9cbf0e1
Compare
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (4)
node/proxy/api/src/coincap.ts (3)
41-43: Deduplicate subscription assets at add time.Prevents redundant per-message lookups and needless growth.
- clientSubscriptions.set(subscriptionId, assets) + clientSubscriptions.set(subscriptionId, Array.from(new Set(assets)))
74-93: Support ALL/empty-assets broadcast and avoid needless filtering.Honor “ALL” (or empty) to pass-through the provider payload.
- // Send updates for each subscription - for (const [subscriptionId, assets] of clientSubscriptions) { - // Filter data to only include assets this subscription requested - const filteredData: Record<string, string> = {} - for (const asset of assets) { - if (message[asset] !== undefined) { - filteredData[asset] = message[asset] - } - } + // Send updates for each subscription + for (const [subscriptionId, assets] of clientSubscriptions) { + const wantsAll = !assets?.length || assets.includes('ALL') + const filteredData: Record<string, string> = wantsAll ? message : {} + if (!wantsAll) { + for (const asset of assets) { + if (message[asset] !== undefined) { + filteredData[asset] = message[asset] + } + } + }
7-9: Keep connection and subscriptions in one structure to avoid drift.Optional: store { connection, subscriptions } per clientId to ensure Maps never get out of sync.
node/proxy/api/src/marketData.ts (1)
45-58: Normalize/dedupe assets before delegating to client.Reduces downstream filtering misses due to casing/whitespace and avoids duplicates.
- this.subscriptionIds.add(subscriptionId) - this.client.subscribe(this.clientId, subscriptionId, this, data.assets) + this.subscriptionIds.add(subscriptionId) + const uniqueAssets = Array.from(new Set(data.assets.map((a) => a.trim().toLowerCase()))) + this.client.subscribe(this.clientId, subscriptionId, this, uniqueAssets)
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
💡 Knowledge Base configuration:
- MCP integration is disabled by default for public repositories
- Jira integration is disabled by default for public repositories
- Linear integration is disabled by default for public repositories
You can enable these sources in your CodeRabbit configuration.
📒 Files selected for processing (2)
node/proxy/api/src/coincap.ts(1 hunks)node/proxy/api/src/marketData.ts(1 hunks)
🧰 Additional context used
🧠 Learnings (2)
📚 Learning: 2025-08-20T06:35:01.854Z
Learnt from: premiumjibles
PR: shapeshift/unchained#1153
File: node/proxy/api/src/marketData.ts:33-33
Timestamp: 2025-08-20T06:35:01.854Z
Learning: premiumjibles confirmed that CoinCap v3 uses wss.coincap.io as the host for WebSocket connections, not ws.coincap.io as in previous versions.
Applied to files:
node/proxy/api/src/coincap.tsnode/proxy/api/src/marketData.ts
📚 Learning: 2025-08-20T06:35:03.599Z
Learnt from: premiumjibles
PR: shapeshift/unchained#1153
File: node/proxy/api/src/marketData.ts:104-107
Timestamp: 2025-08-20T06:35:03.599Z
Learning: For WebSocket integrations with specific financial data APIs like CoinCap, the message payloads are typically JSON strings, so simple .toString() handling is usually sufficient rather than comprehensive type checking for all possible ws message payload types.
Applied to files:
node/proxy/api/src/coincap.tsnode/proxy/api/src/marketData.ts
🧬 Code graph analysis (2)
node/proxy/api/src/coincap.ts (2)
node/proxy/api/src/marketData.ts (3)
MarketDataClient(22-30)MarketDataConnectionHandler(32-78)MarketDataMessage(7-12)node/packages/websocket/src/websocket.ts (2)
Args(16-19)Options(21-25)
node/proxy/api/src/marketData.ts (2)
node/packages/websocket/src/connectionHandler.ts (1)
MessageResponse(18-20)node/packages/prometheus/src/prometheus.ts (1)
Prometheus(7-35)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (1)
- GitHub Check: check
🔇 Additional comments (2)
node/proxy/api/src/coincap.ts (1)
68-96: Publish uses subscriptionId (not clientId) — LGTM.This fixes the earlier contract violation noted in review. Good.
node/proxy/api/src/marketData.ts (1)
7-12: Message type definition is clear and specific — good.
| constructor(url: string, args: Args, opts?: Options) { | ||
| super(url, { logger: args.logger }, opts) | ||
| } |
There was a problem hiding this comment.
Pass Args through to BaseWebsocketClient (apiKey currently dropped).
You’re reconstructing Args and discarding apiKey (and any future fields). Pass args directly.
- constructor(url: string, args: Args, opts?: Options) {
- super(url, { logger: args.logger }, opts)
- }
+ constructor(url: string, args: Args, opts?: Options) {
+ super(url, args, opts)
+ }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| constructor(url: string, args: Args, opts?: Options) { | |
| super(url, { logger: args.logger }, opts) | |
| } | |
| constructor(url: string, args: Args, opts?: Options) { | |
| super(url, args, opts) | |
| } |
🤖 Prompt for AI Agents
In node/proxy/api/src/coincap.ts around lines 11 to 13, the constructor is
reconstructing Args as { logger: args.logger } and dropping apiKey and any
future fields; change the super call to pass the original args object through
(i.e., call super(url, args, opts)) so apiKey and other properties are
preserved.
| function isSubscribePayload(data: unknown): data is SubscribePayload { | ||
| return data !== null && typeof data === 'object' && 'assets' in data | ||
| } |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Strengthen SubscribePayload guard.
Validate array shape and non-empty string elements.
-function isSubscribePayload(data: unknown): data is SubscribePayload {
- return data !== null && typeof data === 'object' && 'assets' in data
-}
+function isSubscribePayload(data: unknown): data is SubscribePayload {
+ if (!data || typeof data !== 'object') return false
+ const assets = (data as any).assets
+ return Array.isArray(assets) && assets.length > 0 && assets.every((a) => typeof a === 'string' && a.length > 0)
+}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| function isSubscribePayload(data: unknown): data is SubscribePayload { | |
| return data !== null && typeof data === 'object' && 'assets' in data | |
| } | |
| function isSubscribePayload(data: unknown): data is SubscribePayload { | |
| if (!data || typeof data !== 'object') return false | |
| const assets = (data as any).assets | |
| return Array.isArray(assets) | |
| && assets.length > 0 | |
| && assets.every(a => typeof a === 'string' && a.length > 0) | |
| } |
🤖 Prompt for AI Agents
In node/proxy/api/src/marketData.ts around lines 18 to 20, the current
isSubscribePayload type guard only checks that data is an object and has an
'assets' property; update it to verify that 'assets' is an actual array and that
every element is a non-empty string. Concretely, cast data to any/object, check
Array.isArray(data.assets) and that data.assets.length > 0 (if empty arrays
should be rejected), and use data.assets.every(a => typeof a === 'string' &&
a.trim().length > 0) before returning true so the guard reliably narrows to
SubscribePayload.
node/proxy/api/src/marketData.ts
Outdated
| onUnsubscribe(subscriptionId: string, data?: unknown): void { | ||
| if (!isSubscribePayload(data)) { | ||
| this.sendError(`invalid subscription payload, no assets provided`, subscriptionId) | ||
| return | ||
| } | ||
|
|
||
| if (subscriptionId) { | ||
| this.subscriptionIds.delete(subscriptionId) | ||
| this.client.unsubscribe(this.clientId, subscriptionId) | ||
| } else { | ||
| this.subscriptionIds.clear() | ||
| this.client.unsubscribe(this.clientId) | ||
| } | ||
| } |
There was a problem hiding this comment.
Unsubscribe requires assets payload — breaks client contract.
Unsubscribe should not validate payload; it only needs subscriptionId (or none to clear all).
- onUnsubscribe(subscriptionId: string, data?: unknown): void {
- if (!isSubscribePayload(data)) {
- this.sendError(`invalid subscription payload, no assets provided`, subscriptionId)
- return
- }
+ onUnsubscribe(subscriptionId: string, _data?: unknown): void {📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| onUnsubscribe(subscriptionId: string, data?: unknown): void { | |
| if (!isSubscribePayload(data)) { | |
| this.sendError(`invalid subscription payload, no assets provided`, subscriptionId) | |
| return | |
| } | |
| if (subscriptionId) { | |
| this.subscriptionIds.delete(subscriptionId) | |
| this.client.unsubscribe(this.clientId, subscriptionId) | |
| } else { | |
| this.subscriptionIds.clear() | |
| this.client.unsubscribe(this.clientId) | |
| } | |
| } | |
| onUnsubscribe(subscriptionId: string, _data?: unknown): void { | |
| if (subscriptionId) { | |
| this.subscriptionIds.delete(subscriptionId) | |
| this.client.unsubscribe(this.clientId, subscriptionId) | |
| } else { | |
| this.subscriptionIds.clear() | |
| this.client.unsubscribe(this.clientId) | |
| } | |
| } |
🤖 Prompt for AI Agents
In node/proxy/api/src/marketData.ts around lines 60 to 73, the onUnsubscribe
handler is incorrectly validating the payload and returning an error when assets
are missing; unblock the client contract by removing the isSubscribePayload
check and any sendError return path — simply treat the optional data parameter
as irrelevant, and perform unsubscribe logic based only on subscriptionId: if
subscriptionId present delete it from this.subscriptionIds and call
this.client.unsubscribe(this.clientId, subscriptionId), otherwise clear
this.subscriptionIds and call this.client.unsubscribe(this.clientId).
There was a problem hiding this comment.
Actionable comments posted: 0
♻️ Duplicate comments (3)
node/packages/websocket/src/connectionHandler.ts (2)
45-49: Start and arm heartbeat immediately (and on each ping send).Arm the timeout right after the initial ping and on each interval tick so dead connections without a first pong are terminated.
this.websocket = websocket - this.websocket.ping() + this.websocket.ping() + this.heartbeat() const pingInterval = setInterval(() => { - this.websocket.ping() + this.websocket.ping() + this.heartbeat() }, this.pingIntervalMs)
80-93: Fix Biome error: don't return values from a void method.Remove expression returns in switch cases.
private onMessage(event: WebSocket.MessageEvent): void { try { const payload: RequestPayload = JSON.parse(event.data.toString()) switch (payload.method) { // browsers do not support ping/pong frame, handle message instead case 'ping': { - return this.websocket.send('pong') + this.websocket.send('pong') + return } case 'subscribe': - return this.onSubscribe(payload.subscriptionId, payload.data) + this.onSubscribe(payload.subscriptionId, payload.data) + return case 'unsubscribe': - return this.onUnsubscribe(payload.subscriptionId) + this.onUnsubscribe(payload.subscriptionId) + return }node/proxy/api/src/marketData.ts (1)
18-20: Harden SubscribePayload guard.Validate array shape and element types to protect the boundary.
-function isSubscribePayload(data: unknown): data is SubscribePayload { - return data !== null && typeof data === 'object' && 'assets' in data -} +function isSubscribePayload(data: unknown): data is SubscribePayload { + if (!data || typeof data !== 'object') return false + const assets = (data as any).assets + return Array.isArray(assets) && assets.length > 0 && assets.every((a) => typeof a === 'string' && a.length > 0) +}
🧹 Nitpick comments (3)
node/packages/websocket/src/connectionHandler.ts (2)
51-54: Terminate socket on error; rely on onclose for cleanup.Avoid manual close path on error; let 'close' event handle metrics and teardown.
- this.websocket.onerror = (error) => { - this.logger.error({ clientId: this.clientId, error, fn: 'ws.onerror' }, 'websocket error') - this.close(pingInterval) - } + this.websocket.onerror = (error) => { + this.logger.error({ clientId: this.clientId, error, fn: 'ws.onerror' }, 'websocket error') + // Force shutdown; onclose will run cleanup and metrics dec. + this.websocket.terminate() + }
60-62: Count inbound client pings as liveness too.Update heartbeat on 'ping' from client to avoid false timeouts when client drives heartbeats.
- this.websocket.on('ping', () => this.websocket.pong()) + this.websocket.on('ping', () => { + this.websocket.pong() + this.heartbeat() + })node/proxy/api/src/marketData.ts (1)
45-58: Prevent duplicate subscriptionIds per client (idempotency/clarity).Avoid double wiring the same subscription; either reject or treat as idempotent.
onSubscribe(subscriptionId: string, data?: unknown): void { if (!subscriptionId) { this.sendError('subscriptionId required', subscriptionId) return } if (!isSubscribePayload(data)) { this.sendError(`invalid subscription payload, no assets provided`, subscriptionId) return } - this.subscriptionIds.add(subscriptionId) + if (this.subscriptionIds.has(subscriptionId)) { + this.sendError('subscriptionId already active', subscriptionId) + return + } + this.subscriptionIds.add(subscriptionId) this.client.subscribe(this.clientId, subscriptionId, this, data.assets) }
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
💡 Knowledge Base configuration:
- MCP integration is disabled by default for public repositories
- Jira integration is disabled by default for public repositories
- Linear integration is disabled by default for public repositories
You can enable these sources in your CodeRabbit configuration.
📒 Files selected for processing (2)
node/packages/websocket/src/connectionHandler.ts(1 hunks)node/proxy/api/src/marketData.ts(1 hunks)
🧰 Additional context used
🧠 Learnings (3)
📓 Common learnings
Learnt from: premiumjibles
PR: shapeshift/unchained#1153
File: node/proxy/api/src/marketData.ts:33-33
Timestamp: 2025-08-20T06:35:01.854Z
Learning: premiumjibles confirmed that CoinCap v3 uses wss.coincap.io as the host for WebSocket connections, not ws.coincap.io as in previous versions.
📚 Learning: 2025-08-20T06:35:01.854Z
Learnt from: premiumjibles
PR: shapeshift/unchained#1153
File: node/proxy/api/src/marketData.ts:33-33
Timestamp: 2025-08-20T06:35:01.854Z
Learning: premiumjibles confirmed that CoinCap v3 uses wss.coincap.io as the host for WebSocket connections, not ws.coincap.io as in previous versions.
Applied to files:
node/proxy/api/src/marketData.ts
📚 Learning: 2025-08-20T06:35:03.599Z
Learnt from: premiumjibles
PR: shapeshift/unchained#1153
File: node/proxy/api/src/marketData.ts:104-107
Timestamp: 2025-08-20T06:35:03.599Z
Learning: For WebSocket integrations with specific financial data APIs like CoinCap, the message payloads are typically JSON strings, so simple .toString() handling is usually sufficient rather than comprehensive type checking for all possible ws message payload types.
Applied to files:
node/proxy/api/src/marketData.ts
🧬 Code graph analysis (2)
node/proxy/api/src/marketData.ts (3)
node/packages/websocket/src/connectionHandler.ts (1)
MessageResponse(18-20)node/packages/prometheus/src/prometheus.ts (1)
Prometheus(7-35)node/proxy/api/src/app.ts (1)
logger(18-21)
node/packages/websocket/src/connectionHandler.ts (1)
node/packages/prometheus/src/prometheus.ts (1)
Prometheus(7-35)
🪛 Biome (2.1.2)
node/packages/websocket/src/connectionHandler.ts
[error] 87-87: The function should not return a value because its return type is void.
The function is here:
'void' signals the absence of value. The returned value is likely to be ignored by the caller.
(lint/correctness/noVoidTypeReturn)
[error] 90-90: The function should not return a value because its return type is void.
The function is here:
'void' signals the absence of value. The returned value is likely to be ignored by the caller.
(lint/correctness/noVoidTypeReturn)
[error] 92-92: The function should not return a value because its return type is void.
The function is here:
'void' signals the absence of value. The returned value is likely to be ignored by the caller.
(lint/correctness/noVoidTypeReturn)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (1)
- GitHub Check: check
🔇 Additional comments (2)
node/proxy/api/src/marketData.ts (2)
60-68: Unsubscribe semantics look good.Supports single-subscription and bulk unsubscribe paths cleanly.
70-72: Good: cleanup on connection close.Ensures provider-side teardown when client disconnects.
Summary
As the title says this introduces a new market data socket to our proxy server. This needs to be a proxy as coincap charges per socket connection, so this allows us to have one socket connection and as many subscribers as we want.
Coincap also doesn't seem to charge based on how many tokens you subscribe to. So we basically subscribe to all of them and let the listener pick which token updates they care about.
Only subscribes to coincap if we have clients for the socket
For this to be useful to the frontend we need to upgrade the coincap adapter (shapeshift/web#10312) and see if we can bring in more assets. If even after than it's not enough we'll need to look at different providers
How to test
docker compose up -din the unchained directorydocker compose upinnode/proxy/api{"subscriptionId":"test-123","method":"subscribe","data":{"assets":["bitcoin","ethereum"]}}I imagine we'll need to do some devopsy stuff as well to get this working in prod if we're happy with it.
Summary by CodeRabbit
New Features
Chores
Documentation