feat: async opportunity parsing

[capJavert]

Parse opportunity in the worker similar how we do preview. Mostly targeted for anon flow where we parse while user does registration so they instantly see their results.

[capJavert]

@claude review

[claude]

Claude finished @capJavert's task in 1m 40s —— View job

---

Code Review

Tasks

• Read and analyze changed files
• Review implementation approach and architecture
• Check for bugs, security, and performance issues
• Verify test coverage
• Provide comprehensive feedback

---

Review Summary

This PR implements async opportunity parsing using a worker, similar to the preview flow. The implementation is well-structured and follows existing patterns in the codebase. I've identified several issues and suggestions below.

Critical Issues

1. Missing Error State Reset (src/schema/opportunity.ts:2772)

When triggering async parsing, the code doesn't clear any existing parseError flag before transitioning to PARSING state. If a previous parse attempt failed, the error message will remain in the flags.

Recommendation: Clear the parseError flag when starting a new parse:

flags: updateFlagsStatement<OpportunityJob>({
  file: fileData,
  parseError: null, // Clear any previous parse errors
}),

2. Inconsistent Organization Association Logic

In createOpportunityFromParsedData (src/common/opportunity/parse.ts:303-320), the code only associates an opportunity with an organization if ctx.userId is present. However, in the async worker flow for anonymous users, userId is stored in flags.file.userId but may not be in ctx.userId when the worker runs.

Impact: Anonymous users who complete registration won't have their parsed opportunity automatically associated with their organization even if they had previous opportunities.

Recommendation: Extract userId from flags.file.userId if ctx.userId is not available in the worker context.

3. Race Condition in State Validation (src/workers/opportunity/parseOpportunity.ts:28-30)

The worker returns early if state is not PARSING, but there's no logging or handling for this case. If the worker processes a message after the opportunity was already parsed (e.g., duplicate message delivery), it silently exits.

Recommendation: Add logging to track when workers skip processing:

if (opportunity.state !== OpportunityState.PARSING) {
  logger.info({ opportunityId, state: opportunity.state }, 'Skipping: opportunity not in PARSING state');
  return;
}

Major Issues

4. File Cleanup on Early Returns

The worker downloads the file from GCS (line 51-54) but only deletes it on successful completion (line 68). If the worker returns early (e.g., missing opportunity, wrong state, missing file data), the file remains in GCS indefinitely.

Recommendation: Use try-finally to ensure cleanup, or track which cases need cleanup:

let shouldCleanupFile = false;
try {
  const fileData = opportunity.flags?.file;
  if (fileData) {
    shouldCleanupFile = true;
    // ... download and process
  }
  // ... success path
  shouldCleanupFile = false; // Already cleaned up
} finally {
  if (shouldCleanupFile && fileData) {
    await storage.bucket(fileData.bucketName).file(fileData.blobName).delete().catch(err => 
      logger.warn({ err, blobName: fileData.blobName }, 'Failed to cleanup file')
    );
  }
}

5. Missing Recruiter Assignment in Async Flow

When parseOpportunity creates an opportunity with userId (authenticated users), it creates an OpportunityUserRecruiter entry (src/schema/opportunity.ts:2761-2766). However, when the worker later calls createOpportunityFromParsedData with the existing opportunityId (line 62-66), this function doesn't check if a recruiter needs to be assigned for cases where the userId became available after initial creation.

Impact: Edge case where a user starts parsing anonymously, logs in before parsing completes, but doesn't get assigned as recruiter.

6. Potential Memory Issue with Large Files

The worker loads the entire file into memory as a buffer (line 51-54) before passing to Brokkr. For large PDF files (e.g., 10MB+), this could cause memory pressure if multiple workers process large files concurrently.

Recommendation: Consider adding file size validation or streaming if Brokkr supports it.

Minor Issues / Suggestions

7. Inconsistent MIME Type Validation

In parseOpportunity mutation (src/schema/opportunity.ts:2728), MIME validation happens before uploading to GCS. However, the worker downloads the file and doesn't re-validate before sending to Brokkr. If the uploaded file was somehow corrupted or modified in GCS, Brokkr might receive invalid data.

Recommendation: Add basic validation after download or rely on Brokkr's error handling.

8. Missing Performance Metrics

The worker tracks duration but only logs on success/failure. Consider adding intermediate timing for debugging slow operations:

• GCS download time
• Brokkr parsing time
• Database transaction time

9. Test Coverage Gap

The test file __tests__/workers/opportunity/parseOpportunity.ts is comprehensive, but I noticed the tests mock all external services (Brokkr, GCS). Consider adding integration tests that verify:

• The pubsub message flow from mutation to worker
• End-to-end anonymous user flow (parse → register → see results)

10. Documentation

The AGENTS.md update mentions the async flow but could benefit from a sequence diagram showing:

1. Anonymous user uploads file
2. Mutation creates PARSING opportunity and publishes event
3. Worker downloads, parses, updates to DRAFT
4. User registers and sees instant results

Code Quality

Strengths:

• ✅ Follows existing patterns (similar to preview worker)
• ✅ Good error handling with proper state transitions
• ✅ Type safety with Zod validation
• ✅ Performance timing instrumentation
• ✅ Comprehensive test coverage
• ✅ Proper use of transactions for data consistency
• ✅ Infrastructure properly configured (Pulumi, worker registration)

Areas for Improvement:

• Function complexity in createOpportunityFromParsedData is growing - consider extracting sub-functions
• Consider adding JSDoc comments for the worker handler to document the async flow

Security Considerations

• ✅ File type validation before processing
• ✅ Proper error sanitization (Zod error prettification)
• ✅ No raw SQL queries
• ⚠️ Consider rate limiting on the pubsub topic to prevent abuse

Architecture Alignment

The implementation correctly follows the codebase patterns:

• ✅ Uses TypedWorker pattern with proper typing
• ✅ Follows pubsub event-driven architecture
• ✅ Proper separation between GraphQL mutation and worker logic
• ✅ Reuses existing parse logic from src/common/opportunity/parse.ts

Verdict

Overall Assessment: Good implementation with solid architecture. The critical issues around error flag management and file cleanup should be addressed before merging. The worker pattern is correctly implemented and aligns well with existing code.

Recommendation: Request changes for critical issues #1 and #4. Issues #2, #3, and #5 should be considered for follow-up improvements.

---

[pulumi]

🍹 The Update (preview) for dailydotdev/api/prod (at a72024c) was successful.

Resource Changes

    Name                                                   Type                                  Operation
~   vpc-native-update-tags-str-cron                        kubernetes:batch/v1:CronJob           update
~   vpc-native-post-analytics-clickhouse-cron              kubernetes:batch/v1:CronJob           update
~   vpc-native-deployment                                  kubernetes:apps/v1:Deployment         update
-   vpc-native-api-db-migration-f5819835                   kubernetes:batch/v1:Job               delete
~   vpc-native-post-analytics-history-day-clickhouse-cron  kubernetes:batch/v1:CronJob           update
~   vpc-native-update-trending-cron                        kubernetes:batch/v1:CronJob           update
~   vpc-native-private-deployment                          kubernetes:apps/v1:Deployment         update
-   vpc-native-api-clickhouse-migration-f5819835           kubernetes:batch/v1:Job               delete
+   api-sub-api.opportunity-parse                          gcp:pubsub/subscription:Subscription  create
~   vpc-native-generic-referral-reminder-cron              kubernetes:batch/v1:CronJob           update
~   vpc-native-clean-zombie-images-cron                    kubernetes:batch/v1:CronJob           update
~   vpc-native-update-current-streak-cron                  kubernetes:batch/v1:CronJob           update
+   vpc-native-api-db-migration-6e5d1b1d                   kubernetes:batch/v1:Job               create
~   vpc-native-update-views-cron                           kubernetes:batch/v1:CronJob           update
~   vpc-native-validate-active-users-cron                  kubernetes:batch/v1:CronJob           update
~   vpc-native-clean-gifted-plus-cron                      kubernetes:batch/v1:CronJob           update
+   vpc-native-api-clickhouse-migration-6e5d1b1d           kubernetes:batch/v1:Job               create
~   vpc-native-personalized-digest-cron                    kubernetes:batch/v1:CronJob           update
~   vpc-native-daily-digest-cron                           kubernetes:batch/v1:CronJob           update
~   vpc-native-bg-deployment                               kubernetes:apps/v1:Deployment         update
~   vpc-native-clean-stale-user-transactions-cron          kubernetes:batch/v1:CronJob           update
~   vpc-native-ws-deployment                               kubernetes:apps/v1:Deployment         update
~   vpc-native-update-source-public-threshold-cron         kubernetes:batch/v1:CronJob           update
~   vpc-native-clean-zombie-opportunities-cron             kubernetes:batch/v1:CronJob           update
~   vpc-native-check-analytics-report-cron                 kubernetes:batch/v1:CronJob           update
~   vpc-native-hourly-notification-cron                    kubernetes:batch/v1:CronJob           update
~   vpc-native-generate-search-invites-cron                kubernetes:batch/v1:CronJob           update
~   vpc-native-update-source-tag-view-cron                 kubernetes:batch/v1:CronJob           update
~   vpc-native-temporal-deployment                         kubernetes:apps/v1:Deployment         update
~   vpc-native-user-profile-updated-sync-cron              kubernetes:batch/v1:CronJob           update
~   vpc-native-clean-zombie-user-companies-cron            kubernetes:batch/v1:CronJob           update
~   vpc-native-personalized-digest-deployment              kubernetes:apps/v1:Deployment         update
~   vpc-native-clean-zombie-users-cron                     kubernetes:batch/v1:CronJob           update
~   vpc-native-sync-subscription-with-cio-cron             kubernetes:batch/v1:CronJob           update
~   vpc-native-calculate-top-readers-cron                  kubernetes:batch/v1:CronJob           update
~   vpc-native-update-highlighted-views-cron               kubernetes:batch/v1:CronJob           update
~   vpc-native-update-tag-recommendations-cron             kubernetes:batch/v1:CronJob           update