feat: update connector#4
Conversation
📝 WalkthroughWalkthroughThe PR links external resources to documents, adds connector content-export contracts and Google Drive retrieval, identifies ingestable resources during syncs, and queues document ingestion through storage and Temporal-backed worker activities. ChangesConnector resource ingestion
Estimated code review effort: 4 (Complex) | ~45 minutes Possibly related PRs
🚥 Pre-merge checks | ✅ 3 | ❌ 2❌ Failed checks (1 warning, 1 inconclusive)
✅ Passed checks (3 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Warning There were issues while running some tools. Please review the errors and either fix the tool's configuration or disable the tool if it's a critical failure. 🔧 ESLint
ESLint install failed: private package registry requires authentication. Disable ESLint in CodeRabbit settings or use public packages. 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. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 5
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
apps/api/prisma/migrations/20260715072615_link_external_resource_document/migration.sql (1)
4-6: 🚀 Performance & Scalability | 🟠 Major | 🏗️ Heavy liftAdd the constraint as
NOT VALIDto avoid locking tables.Adding a foreign key constraint to an existing table requires a full table scan and acquires a
SHARE ROW EXCLUSIVElock on both tables, which blocks writes and can cause downtime in production.To safely apply this without blocking writes, add the constraint as
NOT VALIDfirst, which skips the immediate table scan. Then, runVALIDATE CONSTRAINTin a separate transaction.As per static analysis hints, adding a foreign key constraint requires a table scan and a
SHARE ROW EXCLUSIVElock on both tables, which blocks writes to each table. AddNOT VALIDto the constraint in one transaction and then VALIDATE the constraint in a separate transaction.⚡ Proposed migration fix
-- AddForeignKey -ALTER TABLE "external_resources" ADD CONSTRAINT "external_resources_documentId_fkey" FOREIGN KEY ("documentId") REFERENCES "documents"("id") ON DELETE SET NULL ON UPDATE CASCADE; +ALTER TABLE "external_resources" ADD CONSTRAINT "external_resources_documentId_fkey" FOREIGN KEY ("documentId") REFERENCES "documents"("id") ON DELETE SET NULL ON UPDATE CASCADE NOT VALID; + +-- Note: Execute the validation in a separate transaction/migration to completely avoid write locks: +-- ALTER TABLE "external_resources" VALIDATE CONSTRAINT "external_resources_documentId_fkey";🤖 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 `@apps/api/prisma/migrations/20260715072615_link_external_resource_document/migration.sql` around lines 4 - 6, Update the foreign-key statement for external_resources_documentId_fkey to add the constraint as NOT VALID, then validate it with a separate ALTER TABLE ... VALIDATE CONSTRAINT statement in a distinct transaction, preserving the existing delete and update actions.Source: Linters/SAST tools
🧹 Nitpick comments (1)
packages/workflows/src/workflows/connector-sync.workflow.ts (1)
99-113: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueRemove redundant inline type definition.
Since
activities.syncServicePageis already strongly typed by theConnectorActivitiesContractinjected intoproxyActivities, explicitly redefiningSyncPageOutputinline is redundant and adds unnecessary clutter.🧹 Proposed refactor
- const page: { - nextPageCursor: string | null; - resourceCount: number; - ingestableExternalIds: string[]; - } = await activities.syncServicePage({ + const page = await activities.syncServicePage({ connectorId: input.connectorId, service, jobId, pageCursor, });🤖 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/workflows/src/workflows/connector-sync.workflow.ts` around lines 99 - 113, Remove the inline type annotation from the page assignment in the workflow using activities.syncServicePage. Rely on the strongly typed ConnectorActivitiesContract proxy result, while preserving the existing pagination, progress updates, and ingestResources call.
🤖 Prompt for all review comments with 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.
Inline comments:
In `@packages/connectors/google/src/http.ts`:
- Around line 34-40: Update googleGetBinary to accept the download-size limit
and read the response incrementally rather than calling response.arrayBuffer();
track accumulated bytes, abort and reject once the limit is exceeded, and return
a Buffer only after the full response is within the limit. Update
driveFetchContent and any other callers to pass MAX_DOWNLOAD_BYTES through this
helper.
In `@packages/workflows/src/workflows/connector-sync.workflow.ts`:
- Around line 34-52: Update ingestResources to start ingestion.ingestResource
for all externalIds concurrently and await them with Promise.all, while
preserving per-resource progress updates and failure logging. Keep each
resource’s connectorId/externalId context and ensure one rejected activity does
not prevent the remaining resources from completing.
In `@services/connector-worker/src/activities.ts`:
- Around line 608-610: Update the checksum-matching branch in the document
activity so it does not return immediately when the prior workflow start may
have failed. Before treating unchanged content as complete, resume or create the
pending ingestion job for that document version using a deterministic workflow
ID, then retain the no-op result only after the workflow is confirmed started or
recoverable.
In `@services/connector-worker/src/config.ts`:
- Around line 21-24: Update the STORAGE_USE_SSL schema to validate only the
accepted boolean representation instead of transforming every non-`true` string
to false. Preserve the default false behavior when the variable is absent, while
causing any other configured value to fail validation.
In `@services/connector-worker/src/context.ts`:
- Around line 62-66: Update the temporal client initialization in the context
factory so a rejected Connection.connect promise clears the temporal cache
before propagating the failure. Preserve successful TemporalClient memoization
while allowing subsequent calls to retry with a fresh connection.
---
Outside diff comments:
In
`@apps/api/prisma/migrations/20260715072615_link_external_resource_document/migration.sql`:
- Around line 4-6: Update the foreign-key statement for
external_resources_documentId_fkey to add the constraint as NOT VALID, then
validate it with a separate ALTER TABLE ... VALIDATE CONSTRAINT statement in a
distinct transaction, preserving the existing delete and update actions.
---
Nitpick comments:
In `@packages/workflows/src/workflows/connector-sync.workflow.ts`:
- Around line 99-113: Remove the inline type annotation from the page assignment
in the workflow using activities.syncServicePage. Rely on the strongly typed
ConnectorActivitiesContract proxy result, while preserving the existing
pagination, progress updates, and ingestResources call.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 46b62ee3-deb1-4954-8ecd-6a89ba82cc0d
⛔ Files ignored due to path filters (1)
pnpm-lock.yamlis excluded by!**/pnpm-lock.yaml
📒 Files selected for processing (15)
apps/api/prisma/migrations/20260715072615_link_external_resource_document/migration.sqlapps/api/prisma/schema.prismapackages/connectors/core/src/index.tspackages/connectors/core/src/types.tspackages/connectors/google/src/connector.tspackages/connectors/google/src/http.tspackages/connectors/google/src/services/drive.tspackages/workflows/src/connector-contract.tspackages/workflows/src/definitions.tspackages/workflows/src/index.tspackages/workflows/src/workflows/connector-sync.workflow.tsservices/connector-worker/package.jsonservices/connector-worker/src/activities.tsservices/connector-worker/src/config.tsservices/connector-worker/src/context.ts
| export async function googleGetBinary( | ||
| ctx: ConnectorContext, | ||
| url: string, | ||
| params: Record<string, string | number | boolean | undefined> = {}, | ||
| ): Promise<Buffer> { | ||
| const response = await googleFetch(ctx, url, params); | ||
| return Buffer.from(await response.arrayBuffer()); |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Enforce the download limit while reading the response.
arrayBuffer() allocates the entire file before driveFetchContent checks its size. An unexpectedly large Drive file can therefore exhaust worker memory despite MAX_DOWNLOAD_BYTES. Pass a limit into this helper and abort streaming once it is exceeded.
🤖 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/connectors/google/src/http.ts` around lines 34 - 40, Update
googleGetBinary to accept the download-size limit and read the response
incrementally rather than calling response.arrayBuffer(); track accumulated
bytes, abort and reject once the limit is exceeded, and return a Buffer only
after the full response is within the limit. Update driveFetchContent and any
other callers to pass MAX_DOWNLOAD_BYTES through this helper.
| /** Queue ingestion for each synced resource; failures only log. */ | ||
| async function ingestResources( | ||
| connectorId: string, | ||
| externalIds: string[], | ||
| progress: { ingested: number }, | ||
| ): Promise<void> { | ||
| for (const externalId of externalIds) { | ||
| try { | ||
| const result = await ingestion.ingestResource({ connectorId, externalId }); | ||
| if (result.queued) progress.ingested += 1; | ||
| } catch (error) { | ||
| log.warn('resource ingestion failed', { | ||
| connectorId, | ||
| externalId, | ||
| error: error instanceof Error ? error.message : String(error), | ||
| }); | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
🚀 Performance & Scalability | 🟠 Major | ⚡ Quick win
Execute ingestion activities concurrently to improve workflow performance.
Executing Temporal activities sequentially in a for...of loop will create a severe performance bottleneck, extending the workflow runtime considerably for pages with many resources. Refactoring this to use Promise.all allows Temporal to dispatch the activities concurrently, significantly improving sync throughput.
⚡ Proposed fix for concurrent execution
async function ingestResources(
connectorId: string,
externalIds: string[],
progress: { ingested: number },
): Promise<void> {
- for (const externalId of externalIds) {
- try {
- const result = await ingestion.ingestResource({ connectorId, externalId });
- if (result.queued) progress.ingested += 1;
- } catch (error) {
- log.warn('resource ingestion failed', {
- connectorId,
- externalId,
- error: error instanceof Error ? error.message : String(error),
- });
- }
- }
+ await Promise.all(
+ externalIds.map(async (externalId) => {
+ try {
+ const result = await ingestion.ingestResource({ connectorId, externalId });
+ if (result.queued) progress.ingested += 1;
+ } catch (error) {
+ log.warn('resource ingestion failed', {
+ connectorId,
+ externalId,
+ error: error instanceof Error ? error.message : String(error),
+ });
+ }
+ })
+ );
}📝 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.
| /** Queue ingestion for each synced resource; failures only log. */ | |
| async function ingestResources( | |
| connectorId: string, | |
| externalIds: string[], | |
| progress: { ingested: number }, | |
| ): Promise<void> { | |
| for (const externalId of externalIds) { | |
| try { | |
| const result = await ingestion.ingestResource({ connectorId, externalId }); | |
| if (result.queued) progress.ingested += 1; | |
| } catch (error) { | |
| log.warn('resource ingestion failed', { | |
| connectorId, | |
| externalId, | |
| error: error instanceof Error ? error.message : String(error), | |
| }); | |
| } | |
| } | |
| } | |
| /** Queue ingestion for each synced resource; failures only log. */ | |
| async function ingestResources( | |
| connectorId: string, | |
| externalIds: string[], | |
| progress: { ingested: number }, | |
| ): Promise<void> { | |
| await Promise.all( | |
| externalIds.map(async (externalId) => { | |
| try { | |
| const result = await ingestion.ingestResource({ connectorId, externalId }); | |
| if (result.queued) progress.ingested += 1; | |
| } catch (error) { | |
| log.warn('resource ingestion failed', { | |
| connectorId, | |
| externalId, | |
| error: error instanceof Error ? error.message : String(error), | |
| }); | |
| } | |
| }) | |
| ); | |
| } |
🤖 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/workflows/src/workflows/connector-sync.workflow.ts` around lines 34
- 52, Update ingestResources to start ingestion.ingestResource for all
externalIds concurrently and await them with Promise.all, while preserving
per-resource progress updates and failure logging. Keep each resource’s
connectorId/externalId context and ensure one rejected activity does not prevent
the remaining resources from completing.
| if (document && document.checksum === checksum) { | ||
| return { queued: false, documentId: document.id, reason: 'content unchanged' }; | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
Do not let the checksum no-op suppress a failed workflow start.
The document checksum is persisted before Lines 675-699 start the workflow. If Temporal startup fails, the activity retry reaches this return and permanently skips ingestion. Resume/create the pending job for this document version, preferably with a deterministic workflow ID, before treating matching content as complete.
🤖 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 `@services/connector-worker/src/activities.ts` around lines 608 - 610, Update
the checksum-matching branch in the document activity so it does not return
immediately when the prior workflow start may have failed. Before treating
unchanged content as complete, resume or create the pending ingestion job for
that document version using a deterministic workflow ID, then retain the no-op
result only after the workflow is confirmed started or recoverable.
| STORAGE_USE_SSL: z | ||
| .string() | ||
| .default('false') | ||
| .transform((v) => v === 'true'), |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Map the config file structure first
ast-grep outline services/connector-worker/src/config.ts --view expanded || true
# Show the relevant section with line numbers
sed -n '1,120p' services/connector-worker/src/config.ts
# Find where STORAGE_USE_SSL is used
rg -n "STORAGE_USE_SSL|useSsl|ssl" services/connector-worker/src -SRepository: shibadityadeb/Context-Brain
Length of output: 3844
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Find storage client setup and the useSSL wiring
rg -n "useSSL|storage\." services/connector-worker/src -S
# Inspect the likely client setup file(s) if present
fd -a ".*" services/connector-worker/src | sed -n '1,120p'Repository: shibadityadeb/Context-Brain
Length of output: 1397
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Show the storage client implementation and related config consumers
rg -n "useSSL|STORAGE_USE_SSL|storage" services/connector-worker/src -S
# Read the files that wire the storage config if they exist
for f in $(fd -a "storage|minio|s3" services/connector-worker/src); do
echo "### $f"
sed -n '1,220p' "$f"
doneRepository: shibadityadeb/Context-Brain
Length of output: 1956
Reject invalid STORAGE_USE_SSL values. Any typo or non-true string currently falls back to false, which can silently disable TLS for storage traffic. Use a stricter boolean parser or enum so misconfiguration fails fast.
🤖 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 `@services/connector-worker/src/config.ts` around lines 21 - 24, Update the
STORAGE_USE_SSL schema to validate only the accepted boolean representation
instead of transforming every non-`true` string to false. Preserve the default
false behavior when the variable is absent, while causing any other configured
value to fail validation.
| temporalClient: () => { | ||
| temporal ??= Connection.connect({ address: config.temporal.address }).then( | ||
| (connection) => new TemporalClient({ connection, namespace: config.temporal.namespace }), | ||
| ); | ||
| return temporal; |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
set -euo pipefail
git ls-files services/connector-worker/src/context.ts
wc -l services/connector-worker/src/context.ts
cat -n services/connector-worker/src/context.ts | sed -n '1,180p'Repository: shibadityadeb/Context-Brain
Length of output: 3252
Reset the Temporal cache on connection failure. A rejected Connection.connect() promise stays memoized in temporal, so later calls immediately fail until the worker restarts. Clear the cache in a rejection handler so retries can establish a fresh connection.
🤖 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 `@services/connector-worker/src/context.ts` around lines 62 - 66, Update the
temporal client initialization in the context factory so a rejected
Connection.connect promise clears the temporal cache before propagating the
failure. Preserve successful TemporalClient memoization while allowing
subsequent calls to retry with a fresh connection.
1 participant
Summary by CodeRabbit