postgres(minor): async query support via dblink + MongoDB job tracking

[salatv-ai]

Async Query Support via dblink + MongoDB Job Tracking

Overview

Adds a non-blocking async execution path to the PostgreSQL Query component. Long-running queries are dispatched via dblink_send_query and tracked in MongoDB — allowing the Appmixer flow to return immediately and deliver results later without occupying a worker thread.

---

Architecture

Async Execution Flow

asyncMode=true
  → Query.js: insertOne(pgAsyncJobs) + dblink_send_query()
  → return immediately (no blocking)

jobs.js (poll every 5s):
  → findOne({ status: "pending" })
  → dblink_is_busy() → still running? skip
  → done → dblink_get_result() → updateOne({ status: "done", result })
  → triggerComponent(flowId, componentId, _asyncJobId)

Query.js receive(_asyncJobId):
  → findOne(pgAsyncJobs, { _asyncJobId })
  → sendJson(out)

Key Components

File	Purpose
AsyncJobModel.js	MongoDB model for pgAsyncJobs collection; TTL index for automatic cleanup
connections.js	Process-global pg.Client pool for persistent dblink sessions; survives require() cache clears (same pattern as Kafka connector)
jobs.js	Background poller (5s) + cleanup job (10min); distributed locks prevent duplicate processing across cluster nodes
config.js	Configurable polling intervals and TTL thresholds
plugin.js	Plugin entry point — registers background jobs on connector load
Query.js	Unified component: sync path unchanged; async path added as opt-in via asyncMode toggle

Why dblink?

PostgreSQL's dblink extension allows one connection to fire a query and another to poll its status — decoupling execution from result delivery. This avoids long-held connections and worker starvation on slow queries.

Why MongoDB for job state?

Appmixer already uses MongoDB as its operational store. Storing async job state there (vs. Redis or in-memory) gives us:

• Persistence across restarts
• Cluster-safe — any node can pick up a job
• TTL cleanup — old jobs expire automatically

---

Sync Path

Unchanged. asyncMode=false (default) goes through the existing execution path with zero overhead.

[salatv-ai]

Code Review — Postgres Async Query (PR #1007)

✅ What looks good

• Architecture: dblink-based async execution is a solid approach for long-running PG queries. Clean separation: connections.js (in-memory connection tracking), AsyncJobModel.js (Mongo persistence), jobs.js (polling/cleanup), plugin.js (init).
• Resilience: Good recovery pattern — if a node restarts, jobs.js detects orphaned jobs (no local connection) and restarts the query. Stuck job cleanup with configurable threshold is nice.
• Distributed locking: Both poll and cleanup jobs use locks to avoid concurrent execution across cluster nodes.
• TTL index on updatedAt for auto-cleanup of completed/errored jobs.
• Sync mode preserved: Existing sync behavior is untouched (just restructured for readability).
• Bug fix: messages → message typo in emptyResult output — good catch.

⚠️ Issues / Suggestions

1. Security: auth credentials stored in MongoDB
startAsyncQuery() persists context.auth (including dbPassword) directly into the pgAsyncJobs collection. This means plaintext PG passwords sit in Mongo. Consider:

• Encrypting sensitive fields before storage
• Or not persisting auth at all — instead, re-fetch it from the flow/component context during recovery

2. buildConnStr is vulnerable to injection
The buildConnStr() function concatenates auth values directly into a libpq connection string without escaping. If dbHost, dbUser, or dbPassword contain spaces or special characters, the connection string breaks or worse. Consider using pg-connection-string or at minimum quoting the values: '${value}'.

3. Missing query validation in async mode
Async mode wraps the user query in SELECT row_to_json(t)::text AS _row FROM (${query}) t — this is direct string interpolation. While the sync mode uses QueryStream (parameterized), the async path does not. This is a regression from the SQL injection fix in v2.1.0.

4. _asyncJobId and _asyncError in component.json schema
These fields are used internally by triggerComponent() but are not declared in component.json schema. While this works, it means:

• They could be set by a user in the UI (unintended behavior)
• Consider adding them as hidden/private fields or checking the source of the message

5. Plugin requires dblink extension
CREATE EXTENSION IF NOT EXISTS dblink requires superuser or CREATE privilege on the database. The tooltip mentions this, but the error message when it fails would be cryptic. Consider catching and providing a clearer error.

6. process.PG_ASYNC_CONNECTIONS global
Attaching to process object is fragile — it survives require cache clears but is unconventional. Comment mentions Kafka connector does the same, so this might be an established pattern in the codebase. Just flagging it.

7. Test flows
Four test flow files added — great coverage.

🔧 ESLint fix applied

Fixed 3 ESLint errors in connections.js:

• Double quotes → single quotes
• _row → jsonRow (no-underscore-dangle)
• Unnamed catch _ → e

Summary

Overall solid implementation. The main concerns are security (plaintext credentials in Mongo, connection string injection) and SQL injection in async mode. The architecture and recovery patterns are well-thought-out.