feat(backend): add BullMQ JobManager framework - #1427

packages/shared/src/utils.test.ts (1)

98-112: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick win

Add regression coverage for deprecated-key fallback.

Add tests for experiment_repoDrivenPermissionSyncIntervalMs and experiment_userDrivenPermissionSyncIntervalMs. Also verify that current keys override deprecated keys when both are configured.

The uncovered branches are in packages/shared/src/utils.ts:82-90.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/shared/src/utils.test.ts` around lines 98 - 112, Extend the
resolveConfigSettings tests to cover fallback from
experiment_repoDrivenPermissionSyncIntervalMs and
experiment_userDrivenPermissionSyncIntervalMs to their current keys when only
deprecated values are provided. Add cases confirming each current key takes
precedence when both current and deprecated values are configured, using the
existing DEFAULT_CONFIG_SETTINGS expectations where applicable.

packages/backend/src/repoIndexWorkload.ts (2)

34-56: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Document why this workload sets job state in process instead of onStarted.

Every other workload in this cohort marks the job IN_PROGRESS and updates the parent latest...JobId in onStarted. This workload performs the same writes inside process through prepareRepoIndexJob, so that eligibility checks can skip the job without creating an IN_PROGRESS row. The deviation is reasonable, but it is not stated in the code.

Add a short comment that explains the deviation, so later changes do not move the logic into onStarted and reintroduce spurious IN_PROGRESS rows for skipped jobs.

As per path instructions for packages/backend/**/*.{ts,tsx}: "In onStarted, upsert the workload-specific job row as IN_PROGRESS; update a parent resource's latest...JobId in the same transaction."

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/backend/src/repoIndexWorkload.ts` around lines 34 - 56, Add a brief
comment near the `prepareRepoIndexJob` call in `process` documenting that
job-state and parent `latest...JobId` updates intentionally occur there, rather
than in `onStarted`, so eligibility checks can skip jobs without creating
spurious `IN_PROGRESS` rows.

Source: Path instructions

---

279-289: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value

Remove the duplicate repository-root validation.

Line 279 already awaits isPathAValidGitRepoRoot and enters the block only when the path is not a valid git repository root. Line 280 repeats the same check. The second call only adds signal. Move signal into the first call and drop the second.

♻️ Proposed refactor

-    if (existsSync(repoPath) && !(await isPathAValidGitRepoRoot({ path: repoPath }))) {
-        const isValidGitRepo = await isPathAValidGitRepoRoot({
-            path: repoPath,
-            signal,
-        });
-
-        if (!isValidGitRepo && !isReadOnly) {
-            logger.warn(`${repoPath} is not a valid git repository root. Deleting directory and performing fresh clone.`);
-            await rm(repoPath, { recursive: true, force: true });
-        }
-    }
+    if (
+        existsSync(repoPath) &&
+        !isReadOnly &&
+        !(await isPathAValidGitRepoRoot({ path: repoPath, signal }))
+    ) {
+        logger.warn(`${repoPath} is not a valid git repository root. Deleting directory and performing fresh clone.`);
+        await rm(repoPath, { recursive: true, force: true });
+    }

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/backend/src/repoIndexWorkload.ts` around lines 279 - 289, Update the
initial isPathAValidGitRepoRoot call in the repository validation block to pass
signal, then reuse its result for the invalid-repository branch. Remove the
second isPathAValidGitRepoRoot invocation and preserve the existing isReadOnly
guard, warning, and deletion behavior.

packages/backend/src/attachmentPruneWorkload.ts (1)

133-143: 🚀 Performance & Scalability | 🔵 Trivial | 💤 Low value

Bound the failedIds exclusion list.

failedIds grows for every tombstone whose bytes cannot be deleted. If a storage outage affects many rows, the notIn list can reach thousands of values in one run. Prisma sends each value as a bind parameter, which can hit database parameter limits and slows each query.

Consider stopping the run after a failure threshold, or paginating by a cursor on id instead of excluding failed IDs.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/backend/src/attachmentPruneWorkload.ts` around lines 133 - 143,
Bound the failedIds handling in the attachment-pruning loop so the notIn filter
cannot grow without limit during storage failures. Prefer stopping the run after
a defined failure threshold, or replace failedIds exclusion with cursor-based id
pagination while preserving retries for eligible DELETING attachments; update
the workload’s main pruning function and its findMany query accordingly.

packages/backend/src/connectionWorkload.ts (4)

209-220: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Consider reporting terminal connection-sync failures to Sentry.

onTerminalFailure records the error in the database only. The permission-sync workloads call Sentry.captureException(error, { tags: { jobId, queue } }) in the same hook (see packages/backend/src/ee/repoPermissionSyncWorkload.ts lines 48-207 in the provided context). Add the same reporting here so connection-sync failures stay visible in Sentry.

♻️ Proposed change

     onTerminalFailure: async ({ jobId }, error) => {
+        Sentry.captureException(error, {
+            tags: {
+                jobId,
+                queue: CONNECTION_QUEUE.name,
+            },
+        });
         await db.connectionSyncJob.update({

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/backend/src/connectionWorkload.ts` around lines 209 - 220, Update
the onTerminalFailure hook to report the received error to Sentry with
captureException, including jobId and the connection-sync queue name in the
tags, while preserving the existing database failure update.

---

238-248: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick win

Replace the O(n²) deduplication with a keyed map.

filter plus findIndex compares every repository against every earlier repository. Connection discovery can return thousands of repositories, so this scales quadratically on a hot path. A Map keyed by the same composite key gives the same first-wins result in linear time.

⚡ Proposed change

-const deduplicateRepos = (repos: RepoData[]): RepoData[] =>
-    repos.filter(
-        (repo, index, allRepos) =>
-            index ===
-            allRepos.findIndex(
-                (candidate) =>
-                    candidate.external_id === repo.external_id &&
-                    candidate.external_codeHostUrl ===
-                        repo.external_codeHostUrl,
-            ),
-    );
+const deduplicateRepos = (repos: RepoData[]): RepoData[] => {
+    const uniqueRepos = new Map<string, RepoData>();
+    for (const repo of repos) {
+        const key = `${repo.external_id}\u0000${repo.external_codeHostUrl}`;
+        if (!uniqueRepos.has(key)) {
+            uniqueRepos.set(key, repo);
+        }
+    }
+    return [...uniqueRepos.values()];
+};

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/backend/src/connectionWorkload.ts` around lines 238 - 248, Update
deduplicateRepos to use a Map keyed by the composite external_id and
external_codeHostUrl values, preserving the current first-wins behavior while
reducing deduplication to linear time. Return the Map’s values as the resulting
RepoData array.

---

274-304: 🚀 Performance & Scalability | 🔵 Trivial | 🏗️ Heavy lift

Sequential upserts make discovery latency scale with repository count.

Each discovered repository triggers one awaited db.repo.upsert round trip. A connection with several thousand repositories produces the same number of sequential round trips. Consider batching the upserts with a bounded concurrency helper, or splitting the work into a createMany/updateMany pair where the schema allows it.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/backend/src/connectionWorkload.ts` around lines 274 - 304, Update
the repository persistence loop in the connection workload to avoid awaiting
each db.repo.upsert sequentially. Use the project’s bounded-concurrency helper
to process deduplicateRepos(discoveredRepos) with a controlled number of
concurrent upserts, while preserving the existing upsert payload, selected
fields, and currentRepos collection behavior.

---

365-375: 🚀 Performance & Scalability | 🔵 Trivial | 💤 Low value

Bound the scheduler fan-out.

Promise.all issues one Redis scheduler upsert per current repository at the same time. A large connection creates thousands of concurrent Redis commands in one burst. Consider chunking the calls, for example in batches of 50 to 100, to keep Redis pressure predictable. The same applies to the permission-sync scheduler loop at lines 452-462.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/backend/src/connectionWorkload.ts` around lines 365 - 375, Bound
scheduler fan-out in the current repository indexing loop by replacing the
unbounded Promise.all over currentRepos with sequential batches of roughly
50–100 upsertJobScheduler calls, while preserving each repo’s existing
arguments. Apply the same batching approach to the permission-sync scheduler
loop, ensuring the next batch starts only after the prior batch completes.

packages/backend/src/connectionWorkload.test.ts (2)

26-62: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Prefer importOriginal for the @sourcebot/shared mock.

This mock replaces the whole module and hardcodes CONNECTION_QUEUE and JOB_PRIORITIES. The assertions at lines 314 and 322 then compare against literals that are copies of the real constants. If the real JOB_PRIORITIES values change, these tests keep passing against stale values. The permission-sync tests in this cohort spread importOriginal and override only loadConfig and the queue spec. Use the same pattern here so only the intentionally stubbed exports are replaced.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/backend/src/connectionWorkload.test.ts` around lines 26 - 62, The
`@sourcebot/shared` mock in the connection workload tests should preserve real
exports instead of hardcoding CONNECTION_QUEUE and JOB_PRIORITIES. Update the
mock factory to use importOriginal, retain the actual shared module values, and
override only loadConfig and the intentionally customized queue configuration
while keeping the existing test behavior.

---

341-377: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Add coverage for the swallowed search-context failure.

@sentry/node is mocked at line 22 but no test asserts against it. The process implementation catches syncSearchContexts errors, logs them, and calls Sentry.captureException (lines 143-156 of packages/backend/src/connectionWorkload.ts). That is the one path where a failure does not fail the job. Add a test that rejects mocks.syncSearchContexts and asserts the job still resolves and the error reaches Sentry.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/backend/src/connectionWorkload.test.ts` around lines 341 - 377, Add
a test for connectionWorkload.process that makes mocks.syncSearchContexts
reject, verifies the job still resolves, and asserts the rejected error is
passed to the mocked Sentry.captureException. Reuse the existing successful
workload setup and lifecycle context, targeting the syncSearchContexts
error-handling path rather than the repo reconciliation failure path.

packages/backend/src/ee/accountPermissionSyncWorkload.test.ts (2)

311-327: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Assert the permission write in the success path.

permissionCreateMany and permissionDeleteMany are mocked but this test only checks the client construction and the transaction count. The success path result that users depend on is the set of persisted accountToRepoPermission rows. Add a case that returns repositories from mocks.getReposForAuthenticatedBitbucketServerUser and asserts the resulting createMany payload.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/backend/src/ee/accountPermissionSyncWorkload.test.ts` around lines
311 - 327, The success-path test “syncs the requested account” does not verify
persisted permissions. Configure
mocks.getReposForAuthenticatedBitbucketServerUser to return repositories, then
assert permissionCreateMany was called with the expected accountToRepoPermission
rows and preserve the existing client and transaction assertions.

---

431-459: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Add a stale-job case for onCompleted.

This test covers the case where the job is still the latest. The guard that matters is the opposite case. accountUpdateMany is conditioned on latestPermissionSyncJobId: "job_1", so a superseded job must not clear permissionSyncIssue. Add a test that sets accountUpdateMany.mockResolvedValue({ count: 0 }) and asserts the hook resolves and still updates its own job row.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/backend/src/ee/accountPermissionSyncWorkload.test.ts` around lines
431 - 459, Add a stale-job test alongside the existing latest-job onCompleted
test: configure accountUpdateMany to resolve with { count: 0 }, invoke
createWorkload().onCompleted with the existing lifecycle context, assert the
hook resolves without throwing, and verify permissionSyncJobUpdate still updates
job_1 to COMPLETED while the account issue is not cleared.

packages/backend/src/ee/repoPermissionSyncWorkload.test.ts (1)

39-65: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Several mocked branches are never exercised.

mocks.getProjectMembers, mocks.createGitLabFromPersonalAccessToken, and mocks.getUserPermissionsForServerRepo are registered here but no test drives the GitLab or Bitbucket Server paths. The missing-credentials branch is also uncovered: the implementation throws No credentials found for repo ${id} when getAuthCredentialsForRepo resolves to a falsy value (see packages/backend/src/ee/repoPermissionSyncWorkload.ts lines 48-207 in the provided context). Add cases for at least the missing-credentials branch and one non-GitHub provider.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/backend/src/ee/repoPermissionSyncWorkload.test.ts` around lines 39 -
65, Add tests in the repo permission sync workload suite that cover
getAuthCredentialsForRepo returning a falsy value and assert the expected “No
credentials found for repo ${id}” error, plus a successful non-GitHub provider
path such as GitLab or Bitbucket Server that exercises its registered client and
permission mocks. Ensure the selected provider test drives the corresponding
symbols, including getProjectMembers/createGitLabFromPersonalAccessToken or
getUserPermissionsForServerRepo.

packages/backend/src/jobManager.ts (1)

115-132: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick win

Timed-out worker close leaves work in flight.

stop() races worker.close() against a timer. When the timer wins, stop() continues and closes the BullMQ client while the worker still processes a job. That job then fails with connection errors instead of stalling cleanly. Call worker.close(true) (force) after the grace period so the worker stops before the shared client closes.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/backend/src/jobManager.ts` around lines 115 - 132, Update
JobManager.stop so each worker is force-closed with worker.close(true) when the
graceful shutdown race times out, ensuring all workers have stopped before
bullmqClient.close() runs. Preserve the existing graceful close and timeout
behavior.

packages/shared/src/jobLogger.test.ts (1)

118-130: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Assert the parsed structured entry with a literal object.

Line 120 builds the expected value with parseJobLogEntry, the same production code path under test. A defect in parseJobLogEntry produces a matching expectation, so this assertion cannot fail. Use a literal object for the structured entry, as the test already does for the legacy entry.

♻️ Proposed change

         expect(result).toEqual({
             logs: [
-                parseJobLogEntry(structuredEntry),
+                {
+                    version: 1,
+                    timestamp: "2026-07-28T03:00:00.000Z",
+                    level: "info",
+                    message: "Started",
+                    attempt: 1,
+                },
                 {

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/shared/src/jobLogger.test.ts` around lines 118 - 130, Replace
parseJobLogEntry(structuredEntry) in the expected logs array of the relevant
test with a literal object containing the structured entry’s expected parsed
fields, matching the existing legacy-entry assertion style. Keep the production
call under test unchanged and preserve the count and legacy log expectations.

packages/backend/src/executionLock.ts (2)

109-122: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick win

Guard isLockContention against rejected attempt promises.

error.attempts holds promises. If any promise rejects, Promise.all rejects inside the catch block at Line 192. The rejection then replaces the original Redlock error and hides the acquisition failure cause. Return false when the attempts cannot be inspected.

🛡️ Proposed change

 const isLockContention = async (error: unknown): Promise<boolean> => {
     if (!(error instanceof ExecutionError) || error.attempts.length === 0) {
         return false;
     }
 
-    const attempts = await Promise.all(error.attempts);
+    let attempts;
+    try {
+        attempts = await Promise.all(error.attempts);
+    } catch {
+        return false;
+    }
     return attempts.every(

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/backend/src/executionLock.ts` around lines 109 - 122, Update
isLockContention to handle rejected promises from error.attempts without
propagating the rejection: wrap the Promise.all inspection in error handling and
return false when any attempt cannot be resolved, while preserving the existing
contention checks for successfully resolved attempts.

---

156-205: 🚀 Performance & Scalability | 🔵 Trivial | ⚖️ Poor tradeoff

Consider a bound on contention retries.

The loop retries lock acquisition without a limit. Only shutdown ends the wait. A long-held lock keeps a worker slot occupied for the whole contention period, which reduces effective concurrency for that queue. Add a maximum wait or a maximum retry count, then fail the attempt so BullMQ retry and backoff handle rescheduling. Emit a metric or warning when the wait exceeds a threshold so contention is observable.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/backend/src/executionLock.ts` around lines 156 - 205, Bound the
contention retry loop in the lock-acquisition method surrounding redlock. Track
elapsed wait time or retry attempts, stop retrying after a configured maximum,
and throw the contention error so BullMQ can reschedule with its retry/backoff
policy; also emit the existing warning or metric when the threshold is exceeded.
Preserve immediate propagation for acquired-lock failures and shutdown aborts.

packages/backend/src/jobManager.test.ts (1)

365-397: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Add coverage for the non-terminal failure branch and hook errors.

The suite only exercises the terminal path where attemptsMade equals opts.attempts. Add two cases: a failure with attemptsMade below opts.attempts must not call onTerminalFailure, and a hook that rejects must call Sentry.captureException and still flush the logger. These cases protect the classification logic flagged in packages/backend/src/jobManager.ts.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/backend/src/jobManager.test.ts` around lines 365 - 397, Add tests
alongside the existing terminal-failure test for the non-terminal failed-job
path and rejected lifecycle hooks. Verify a failure with attemptsMade below
opts.attempts does not invoke onTerminalFailure, and verify a rejecting hook
calls Sentry.captureException while the job logger still flushes; use the
existing BullMQJobManager, worker failed handler, and logger mocks.

packages/backend/src/configManager.ts (1)

153-164: 🩺 Stability & Availability | 🔵 Trivial | 💤 Low value

Scheduler removal and connection deletion are not atomic.

removeJobScheduler runs before prisma.connection.delete. If the delete fails, the scheduler is already gone and the connection stops syncing while still present in the database. The reverse order leaves an orphan scheduler that enqueues jobs for a missing connection.

The current order is the safer one. Consider logging a warning when removeJobScheduler returns false, so orphaned schedulers are visible.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/backend/src/configManager.ts` around lines 153 - 164, Update the
deleted-connection loop to capture the boolean result of removeJobScheduler for
each connection and log a warning when it returns false, including the
connection ID or name so orphaned schedulers are identifiable; preserve the
existing removal-before-prisma.connection.delete order.

packages/backend/src/reconcileJobSchedulersAtStartup.ts (2)

48-66: 🚀 Performance & Scalability | 🔵 Trivial | 💤 Low value

Consider bounding scheduler upsert concurrency at startup.

targets contains one entry per repository. Large installations produce thousands of concurrent upsertJobScheduler calls against Redis in one Promise.all. This can saturate the connection pool during startup.

Batch the upserts, for example in chunks of 50 to 100.

The jobOptions branch is also redundant. upsertJobScheduler declares options as optional, so passing undefined is equivalent.

♻️ Suggested simplification of the branch

-        targets.map(({ schedulerId, data }) => {
-            if (jobOptions) {
-                return jobManager.upsertJobScheduler(
-                    workloadName,
-                    schedulerId,
-                    schedule,
-                    data,
-                    jobOptions,
-                );
-            }
-            return jobManager.upsertJobScheduler(
-                workloadName,
-                schedulerId,
-                schedule,
-                data,
-            );
-        }),
+        targets.map(({ schedulerId, data }) =>
+            jobManager.upsertJobScheduler(
+                workloadName,
+                schedulerId,
+                schedule,
+                data,
+                jobOptions,
+            ),
+        ),

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/backend/src/reconcileJobSchedulersAtStartup.ts` around lines 48 -
66, Limit startup scheduler upserts in the targets mapping within
reconcileJobSchedulersAtStartup by processing entries in bounded batches (for
example, 50–100) rather than issuing every upsert through one unbounded
Promise.all, while preserving completion of all targets. Simplify the
jobManager.upsertJobScheduler invocation by removing the redundant jobOptions
branch and passing the optional value consistently.

---

133-172: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Scheduler ID prefixes are duplicated across files. Three of the four workloads build scheduler IDs from inline string literals instead of shared helpers, so the prefix used for cleanup and the template used for creation can drift apart. account-permission-sync already uses ACCOUNT_PERMISSION_SYNC_SCHEDULER_ID_PREFIX and getAccountPermissionSyncSchedulerId. Follow that pattern for the rest.

• packages/backend/src/reconcileJobSchedulersAtStartup.ts#L133-L172: replace the inline connection-sync-v1-, repo-index-v1-, and repo-permission-sync-v1- prefixes and ID templates with exported prefix constants and ID helper functions.
• packages/backend/src/configManager.ts#L18-L19: delete the local getConnectionSyncSchedulerId and import the shared connection-sync ID helper instead.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/backend/src/reconcileJobSchedulersAtStartup.ts` around lines 133 -
172, Replace the inline scheduler prefixes and ID templates in
reconcileJobSchedulersAtStartup with shared exported prefix constants and ID
helper functions for connection-sync, repo-index, and repo-permission-sync,
matching the existing account-permission-sync pattern; update
packages/backend/src/reconcileJobSchedulersAtStartup.ts lines 133-172
accordingly. In packages/backend/src/configManager.ts lines 18-19, remove the
local getConnectionSyncSchedulerId and import the shared connection-sync helper
instead.

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes