Implement Gmail push notifications with workflows#1318
Conversation
WalkthroughThis update introduces a comprehensive subscription and workflow management system for handling email provider integrations (Google and Outlook) in a mail server application. It adds new subscription factory classes, workflow pipelines for processing Gmail data, utility functions, and updates configuration files to support queues and workflows. Type definitions, prompt templates, and notification mechanisms are also enhanced. Changes
Sequence Diagram(s)sequenceDiagram
participant Client
participant ServerAPI
participant SubscriptionFactory
participant ProviderAPI
participant Workflow
Client->>ServerAPI: Enable brain function (mutation)
ServerAPI->>SubscriptionFactory: getSubscriptionFactory(provider)
ServerAPI->>SubscriptionFactory: subscribe({connectionId, providerId})
SubscriptionFactory->>ProviderAPI: Setup subscription (e.g., Gmail watch)
ProviderAPI-->>SubscriptionFactory: Subscription confirmation
SubscriptionFactory-->>ServerAPI: Subscription result
Note over ServerAPI,Workflow: On notification from provider
ProviderAPI->>ServerAPI: POST /a8n/notify/:providerId
ServerAPI->>Workflow: Trigger MainWorkflow(historyId, subscriptionName)
Workflow->>Workflow: Orchestrate ZeroWorkflow and ThreadWorkflow for processing
Possibly related PRs
Poem
✨ Finishing Touches
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. Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
MrgSub
changed the title
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 16
🔭 Outside diff range comments (1)
apps/server/wrangler.jsonc (1)
95-105: Add missing secrets & replace the placeholder service-account
setupGmailWatch()relies onenv.GOOGLE_CLIENT_IDandenv.GOOGLE_CLIENT_SECRET, but these variables are not present in any environment block.
Likewise, keeping"GOOGLE_S_ACCOUNT": "{}"will parse, yet every required field (client_email,private_key, …) is empty and will make JWT signing fail.• Inject the real service-account JSON through Wrangler secrets or
.dev.vars.
• DeclareGOOGLE_CLIENT_ID/GOOGLE_CLIENT_SECRETin eachvarssection (again via secrets in prod).Failing to do so will make the whole Gmail subscription flow break at runtime.
Also applies to: 223-230, 345-352
🧹 Nitpick comments (10)
apps/server/package.json (1)
6-8:--show-interactive-dev-session=truecan leak env/secrets in CIInteractive Dev Sessions surface request/response logs in a public URL.
If you ever runnpm run devin CI or a shared environment those logs may contain cookies or bearer tokens set inwrangler.toml / wrangler.jsonc.Consider guarding the flag with an env check:
- "dev": "wrangler dev --show-interactive-dev-session=true --experimental-vectorize-bind-to-prod --env local", + "dev": "wrangler dev $([[ $CI == \"true\" ]] && echo \"\" || echo \"--show-interactive-dev-session=true\") --experimental-vectorize-bind-to-prod --env local",apps/server/src/types.ts (1)
13-44: Duplicate hard-codeddefaultLabels– consider centralising
An almost identicaldefaultLabelsarray already lives inapps/mail/components/mail/mail.tsx. Maintaining two sources is error-prone and violates DRY. Export the array from a shared module (e.g.shared/labels.ts) and import it in both client & server to guarantee they stay in lock-step.apps/server/src/lib/server-utils.ts (2)
53-59: Avoid banned{}type – give the payload an explicit shape
Biome flags this line. You already define distinct payload types below,
so the umbrella type can be a union instead of bare{}:-type ListThreadsNotification = { - type: 'listThreads'; - payload: {}; -}; +type ListThreadsNotification = { + type: 'listThreads'; + payload: Record<string, never>; +};or simply drop the redundant wrapper and use
voidwhere no data is
expected.
116-130:verifyTokenhits Google endpoint on every call – cache or use certs
Round-tripping tohttps://oauth2.googleapis.com/tokeninfoper request
adds latency and may run into quota limits. Google-issued tokens are JWTs
that can be verified locally with the certs from
https://www.googleapis.com/oauth2/v3/certs. Consider switching to
offline verification with periodic cert refresh.apps/server/src/lib/factories/outlook-subscription.factory.ts (2)
8-15: Method stubs throw generic errors – return501 Not ImplementedResponse
These factories sit behind an HTTP-style interface. Throwing raw Errors
bubbles up as 500s and pollutes logs. Return an explicitResponsewith
status = 501 to make the intent clear until implementation lands.- throw new Error('Outlook subscription not implemented yet'); + return new Response('Outlook subscription not implemented', { status: 501 });
24-28: Token verification placeholder
Be aware that Microsoft Graph token validation requires hitting
https://login.microsoftonline.com/common/v2.0/.well-known/openid-configuration
to fetch keys, similar to Google. Plan for local JWT validation as with
Google to avoid per-request latency.apps/server/src/lib/utils.ts (1)
13-23: Move test-onlycstub out of production codeThe exported mock
cleaks into runtime bundles and risks accidental use in real handlers. Relocate it to a dedicated test helper and exclude it from the worker build.apps/server/src/lib/factories/base-subscription.factory.ts (1)
26-36: Reuse database handle instead of reconnecting per call
createDbspawns a fresh Postgres connection each timegetConnectionFromDbruns, which can overwhelm connection limits under load. Pass a shared DB instance to the factory or cache it at module scope.apps/server/src/trpc/routes/brain.ts (1)
40-43: Null-safety guard for metadata
result.connectionis accessed without checking that the metadata actually contains theconnectionkey. Add a nullish check to avoidundefinedcomparisons that would erroneously pass the ownership test.apps/server/src/pipelines.ts (1)
240-255: Serial 4 s sleep limits throughputSleeping 4 seconds per thread (
threadsToProcess.forEach) caps you at 900 threads/hour even when Gmail quota allows more.Consider:
• Using a concurrency counter &
Promise.allSettledup to N parallel threads.
• Back-off only on 429 / rate-limit responses.This dramatically improves latency while remaining quota-safe.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
⛔ Files ignored due to path filters (1)
pnpm-lock.yamlis excluded by!**/pnpm-lock.yaml
📒 Files selected for processing (18)
apps/mail/components/party.tsx(1 hunks)apps/server/package.json(3 hunks)apps/server/src/lib/auth.ts(3 hunks)apps/server/src/lib/brain.fallback.prompts.ts(2 hunks)apps/server/src/lib/brain.ts(1 hunks)apps/server/src/lib/driver/google.ts(1 hunks)apps/server/src/lib/driver/types.ts(1 hunks)apps/server/src/lib/factories/base-subscription.factory.ts(1 hunks)apps/server/src/lib/factories/google-subscription.factory.ts(1 hunks)apps/server/src/lib/factories/outlook-subscription.factory.ts(1 hunks)apps/server/src/lib/factories/subscription-factory.registry.ts(1 hunks)apps/server/src/lib/server-utils.ts(2 hunks)apps/server/src/lib/utils.ts(1 hunks)apps/server/src/main.ts(4 hunks)apps/server/src/pipelines.ts(1 hunks)apps/server/src/trpc/routes/brain.ts(3 hunks)apps/server/src/types.ts(1 hunks)apps/server/wrangler.jsonc(9 hunks)
🧰 Additional context used
🧬 Code Graph Analysis (8)
apps/server/src/lib/brain.fallback.prompts.ts (1)
apps/server/src/types.ts (1)
defaultLabels(13-44)
apps/server/src/lib/factories/outlook-subscription.factory.ts (1)
apps/server/src/lib/factories/base-subscription.factory.ts (2)
SubscriptionData(6-10)UnsubscriptionData(12-15)
apps/server/src/types.ts (1)
apps/mail/components/mail/mail.tsx (1)
defaultLabels(74-105)
apps/server/src/lib/brain.ts (2)
apps/server/src/db/schema.ts (1)
connection(87-106)apps/server/src/lib/factories/subscription-factory.registry.ts (1)
getSubscriptionFactory(20-26)
apps/server/src/lib/factories/base-subscription.factory.ts (3)
apps/server/src/db/index.ts (1)
createDb(5-9)apps/server/src/db/schema.ts (1)
connection(87-106)apps/server/src/types.ts (1)
defaultLabels(13-44)
apps/server/src/lib/factories/subscription-factory.registry.ts (1)
apps/server/src/lib/factories/outlook-subscription.factory.ts (1)
OutlookSubscriptionFactory(8-29)
apps/server/src/lib/utils.ts (1)
apps/server/src/types.ts (1)
AppContext(211-211)
apps/server/src/trpc/routes/brain.ts (5)
apps/server/src/trpc/trpc.ts (1)
activeConnectionProcedure(31-42)apps/server/src/db/schema.ts (1)
connection(87-106)apps/server/src/lib/utils.ts (1)
setSubscribedState(33-41)apps/server/src/types.ts (1)
ISubscribeBatch(8-11)apps/server/src/lib/brain.ts (1)
disableBrainFunction(11-16)
🪛 Biome (1.9.4)
apps/server/src/lib/server-utils.ts
[error] 57-57: Don't use '{}' as a type.
Prefer explicitly define the object shape. '{}' means "any non-nullable value".
(lint/complexity/noBannedTypes)
🔇 Additional comments (6)
apps/server/package.json (1)
44-48: Pinningcheeriowithout a caret may block security patchesMost prod deps are locked to patch versions (
1.2.9,12.0.0, …) so this may be deliberate.
If not, switch to a caret so Dependabot/Snyk can surface CVEs automatically.- "cheerio": "1.1.0", + "cheerio": "^1.1.0",apps/server/src/lib/driver/types.ts (1)
70-73: Interface change may break non-Google drivers
listHistorywas added toMailManager, but only the Google implementation is provided in this PR.
Any Outlook or future providers that compile against this interface will now fail.Add a stub (throwing) implementation to the other drivers or make the method optional:
- listHistory<T>(historyId: string): Promise<{ history: T[]; historyId: string }>; + listHistory?<T = unknown>(historyId: string): Promise<{ history: T[]; historyId: string }>;apps/server/src/lib/auth.ts (1)
121-125: Enum cast may hide unsupported providers
disableBrainFunction({ providerId: connection.providerId as EProviders })
assumes every stored string is a valid enum value. If the DB still holds
'microsoft'while the enum expects'outlook', this will throw. Handle
unknown providers explicitly or normalise the stored values during
migration.apps/server/src/main.ts (1)
203-240: Verify KV key used for expiry trackingThe scheduler reads last-subscribe timestamps from
env.gmail_sub_age, but the only writer shown issetSubscribedState, which writes toenv.subscribed_accounts. If nothing populatesgmail_sub_age, every key appears expired. Confirm the correct namespace or update writer/reader to match.apps/server/wrangler.jsonc (1)
38-57: Double-check queue binding namesThe bindings are lowercase (
thread_queue,subscribe_queue) while Workers runtime usually exposes them as the exact identifier (env.thread_queue).
If your code (e.g. main.ts) importsenv.THREAD_QUEUE/SUBSCRIBE_QUEUEthe mismatch will throw “binding not found” at deploy time.Verify the names match the identifiers used in the handlers or rename here for consistency.
Also applies to: 166-185, 315-334
apps/server/src/lib/factories/google-subscription.factory.ts (1)
200-221:google-auth-libraryAPI usage may be outdated
auth.refreshAccessToken()was removed in v9; current API isauth.getAccessToken()orauth.refreshAccessTokenAsync().
If the project is on the latestgoogle-auth-library, this call will throw.Confirm library version and update the call accordingly.
| public async listHistory<T>(historyId: string): Promise<{ history: T[]; historyId: string }> { | ||
| return this.withErrorHandler( | ||
| 'listHistory', | ||
| async () => { | ||
| const response = await this.gmail.users.history.list({ | ||
| userId: 'me', | ||
| startHistoryId: historyId, | ||
| }); | ||
|
|
||
| const history = response.data.history || []; | ||
| const nextHistoryId = response.data.historyId || historyId; | ||
|
|
||
| return { history: history as T[], historyId: nextHistoryId }; | ||
| }, | ||
| { historyId }, | ||
| ); | ||
| } |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Typed cast hides the actual Gmail Schema$History
history is force-cast to T[], losing type safety and IntelliSense.
If you want a generic return, expose the concrete type and let callers up-cast:
- public async listHistory<T>(historyId: string): Promise<{ history: T[]; historyId: string }> {
+ public async listHistory(
+ historyId: string,
+ ): Promise<{ history: gmail_v1.Schema$History[]; historyId: string }> {
…
- const history = response.data.history || [];
+ const history = response.data.history ?? [];Call sites that care about sub-shapes can as-cast, but the default is now correct.
Also consider handling the “History cursor is too old” 404 explicitly and marking the connection as stale so a new watch can be issued.
Committable suggestion skipped: line range outside the PR's diff.
🤖 Prompt for AI Agents
In apps/server/src/lib/driver/google.ts lines 47 to 63, the method listHistory
force-casts the Gmail API response history to a generic type T[], which hides
the actual Gmail Schema$History type and reduces type safety. To fix this,
change the method to return the concrete Gmail Schema$History[] type by default
and let callers cast to more specific types if needed. Additionally, add
explicit handling for the "History cursor is too old" 404 error from the Gmail
API by catching this error, marking the connection as stale, and triggering a
new watch to refresh the history cursor.
MrgSub
force-pushed
the
06-16-background_pipelines
branch
from
151dbb1 to
306f5ed
Compare
There was a problem hiding this comment.
Actionable comments posted: 2
♻️ Duplicate comments (2)
apps/server/src/lib/brain.fallback.prompts.ts (1)
217-224: Fix malformed XML tags in label template.The XML tags are incorrectly formatted:
- Line 220:
<name/>should be</name>- Line 221: Missing
</usecase>closing tag${labels .map( (label) => `<item> - <name>${label.name}<name/> - <usecase>${defaultLabels.find((e) => e.name === label.name)?.usecase || ''}</usecase> + <name>${label.name}</name> + <usecase>${defaultLabels.find((e) => e.name === label.name)?.usecase || ''}</usecase> </item>`, ) .join('\n')}apps/server/src/lib/factories/google-subscription.factory.ts (1)
270-277: Move KV write after successful Gmail watch setup to prevent leaks.The
gmail_sub_age.put()operation occurs beforesetupGmailWatch(). If the watch setup fails, the KV entry remains, incorrectly marking the connection as subscribed.console.log(`[SUBSCRIPTION] Setting up Gmail watch for connection: ${connectionData.id}`); await this.setupGmailWatch(connectionData, pubSubName); + // Only mark as subscribed after successful watch setup await env.gmail_sub_age.put( `${connectionId}__${EProviders.google}`, new Date().toISOString(), );
🧹 Nitpick comments (2)
apps/server/src/lib/server-utils.ts (1)
117-131: Strengthen token validation beyond presence check.The current implementation only verifies that the tokeninfo endpoint returns some data. Consider validating specific fields like
aud(audience),iss(issuer), oremail_verifiedto ensure the token is not just valid but also appropriate for your application.export const verifyToken = async (token: string) => { const response = await fetch(`https://oauth2.googleapis.com/tokeninfo?id_token=${token}`, { method: 'GET', headers: { 'Content-Type': 'application/json', }, }); if (!response.ok) { throw new Error(`Failed to verify token: ${await response.text()}`); } const data = (await response.json()) as any; - return !!data; + // Validate essential token fields + return !!(data && data.email_verified && data.aud === env.GOOGLE_CLIENT_ID); };apps/server/src/lib/factories/google-subscription.factory.ts (1)
326-339: Consolidate duplicate token verification logic.This
verifyTokenmethod duplicates the logic fromserver-utils.ts. Consider:
- Using the shared
verifyTokenfrom server-utils- Or if provider-specific validation is needed, extend the shared implementation
+ import { verifyToken as verifyGoogleToken } from '../../lib/server-utils'; + public async verifyToken(token: string): Promise<boolean> { - try { - const response = await fetch(`https://oauth2.googleapis.com/tokeninfo?id_token=${token}`); - - if (!response.ok) { - return false; - } - - const data = await response.json(); - return !!data; - } catch { - return false; - } + return verifyGoogleToken(token); }
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
⛔ Files ignored due to path filters (1)
pnpm-lock.yamlis excluded by!**/pnpm-lock.yaml
📒 Files selected for processing (19)
apps/mail/components/party.tsx(1 hunks)apps/mail/components/ui/ai-sidebar.tsx(2 hunks)apps/server/package.json(2 hunks)apps/server/src/lib/auth.ts(4 hunks)apps/server/src/lib/brain.fallback.prompts.ts(4 hunks)apps/server/src/lib/brain.ts(1 hunks)apps/server/src/lib/driver/google.ts(1 hunks)apps/server/src/lib/driver/types.ts(1 hunks)apps/server/src/lib/factories/base-subscription.factory.ts(1 hunks)apps/server/src/lib/factories/google-subscription.factory.ts(1 hunks)apps/server/src/lib/factories/outlook-subscription.factory.ts(1 hunks)apps/server/src/lib/factories/subscription-factory.registry.ts(1 hunks)apps/server/src/lib/server-utils.ts(2 hunks)apps/server/src/lib/utils.ts(1 hunks)apps/server/src/main.ts(4 hunks)apps/server/src/pipelines.ts(1 hunks)apps/server/src/trpc/routes/brain.ts(3 hunks)apps/server/src/types.ts(1 hunks)apps/server/wrangler.jsonc(9 hunks)
✅ Files skipped from review due to trivial changes (1)
- apps/mail/components/ui/ai-sidebar.tsx
🚧 Files skipped from review as they are similar to previous changes (13)
- apps/server/package.json
- apps/server/src/lib/driver/types.ts
- apps/mail/components/party.tsx
- apps/server/src/lib/auth.ts
- apps/server/src/lib/driver/google.ts
- apps/server/src/lib/factories/outlook-subscription.factory.ts
- apps/server/src/lib/brain.ts
- apps/server/src/main.ts
- apps/server/src/lib/utils.ts
- apps/server/src/trpc/routes/brain.ts
- apps/server/src/lib/factories/base-subscription.factory.ts
- apps/server/wrangler.jsonc
- apps/server/src/pipelines.ts
🧰 Additional context used
🧬 Code Graph Analysis (2)
apps/server/src/types.ts (1)
apps/mail/components/mail/mail.tsx (1)
defaultLabels(74-105)
apps/server/src/lib/brain.fallback.prompts.ts (1)
apps/server/src/types.ts (1)
defaultLabels(13-44)
🪛 Biome (1.9.4)
apps/server/src/lib/server-utils.ts
[error] 57-57: Don't use '{}' as a type.
Prefer explicitly define the object shape. '{}' means "any non-nullable value".
(lint/complexity/noBannedTypes)
🔇 Additional comments (1)
apps/server/src/lib/factories/subscription-factory.registry.ts (1)
1-27: Well-structured factory registry implementation.Good job addressing the previous concern by commenting out the unfinished OutlookSubscriptionFactory. The registry pattern is cleanly implemented with proper error handling and accessor methods.
| export const defaultLabels = [ | ||
| { | ||
| name: 'to respond', | ||
| usecase: 'emails you need to respond to. NOT sales, marketing, or promotions.', | ||
| }, | ||
| { | ||
| name: 'FYI', | ||
| usecase: | ||
| 'emails that are not important, but you should know about. NOT sales, marketing, or promotions.', | ||
| }, | ||
| { | ||
| name: 'comment', | ||
| usecase: | ||
| 'Team chats in tools like Google Docs, Slack, etc. NOT marketing, sales, or promotions.', | ||
| }, | ||
| { | ||
| name: 'notification', | ||
| usecase: 'Automated updates from services you use. NOT sales, marketing, or promotions.', | ||
| }, | ||
| { | ||
| name: 'promotion', | ||
| usecase: 'Sales, marketing, cold emails, special offers or promotions. NOT to respond to.', | ||
| }, | ||
| { | ||
| name: 'meeting', | ||
| usecase: 'Calendar events, invites, etc. NOT sales, marketing, or promotions.', | ||
| }, | ||
| { | ||
| name: 'billing', | ||
| usecase: 'Billing notifications. NOT sales, marketing, or promotions.', | ||
| }, | ||
| ]; |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Eliminate code duplication by centralizing label definitions.
The defaultLabels array is an exact duplicate of the one defined in apps/mail/components/mail/mail.tsx. This violates the DRY principle and creates maintenance risks.
Consider moving the label definitions to a shared location (e.g., a dedicated constants file) and importing them in both places.
🤖 Prompt for AI Agents
In apps/server/src/types.ts lines 13 to 44, the defaultLabels array duplicates
the same array in apps/mail/components/mail/mail.tsx, violating DRY principles.
To fix this, create a new shared constants file (e.g.,
apps/shared/constants/labels.ts) to hold the defaultLabels array, then import
and use this shared constant in both apps/server/src/types.ts and
apps/mail/components/mail/mail.tsx, removing the duplicated definitions from
both files.
| } catch (error) { | ||
| console.error('[SUBSCRIPTION] Setup failed:', error); | ||
|
|
||
| // Clean up on failure using base class method | ||
| // await this.cleanupOnFailure(connectionId, env); | ||
|
|
||
| if (error instanceof Error && error.message.includes('Already Exists')) { | ||
| console.log('Resource already exists, continuing...'); | ||
| return c.json({}); | ||
| } | ||
|
|
||
| throw error; | ||
| } | ||
| } catch (error) { | ||
| console.error('[SUBSCRIPTION] Error:', error); | ||
|
|
||
| // Clean up on error using base class method | ||
| // await this.cleanupOnFailure(connectionId, env); | ||
|
|
||
| return c.json({ error: 'Internal server error' }, { status: 500 }); | ||
| } | ||
| } |
There was a problem hiding this comment.
Enable cleanup on failure and manage subscription state.
The error handling has commented-out cleanup calls and doesn't set the subscription state. This can lead to:
- Resource leaks on failure
- Inconsistent subscription state
} catch (error) {
console.error('[SUBSCRIPTION] Setup failed:', error);
// Clean up on failure using base class method
- // await this.cleanupOnFailure(connectionId, env);
+ await this.cleanupOnFailure(connectionId, env);
if (error instanceof Error && error.message.includes('Already Exists')) {
console.log('Resource already exists, continuing...');
+ await this.setSubscribedState(connectionId, EProviders.google, 'active', env);
return c.json({});
}
throw error;
}
} catch (error) {
console.error('[SUBSCRIPTION] Error:', error);
// Clean up on error using base class method
- // await this.cleanupOnFailure(connectionId, env);
+ await this.cleanupOnFailure(connectionId, env);
return c.json({ error: 'Internal server error' }, { status: 500 });
}Additionally, add a call to set the subscription state to 'active' after successful setup:
console.log(`[SUBSCRIPTION] Initializing labels for connection: ${connectionId}`);
await this.initializeConnectionLabels(connectionId);
+
+ // Mark subscription as active
+ await this.setSubscribedState(connectionId, EProviders.google, 'active', env);
console.log(`[SUBSCRIPTION] Setup completed successfully for connection: ${connectionId}`);📝 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.
| } catch (error) { | |
| console.error('[SUBSCRIPTION] Setup failed:', error); | |
| // Clean up on failure using base class method | |
| // await this.cleanupOnFailure(connectionId, env); | |
| if (error instanceof Error && error.message.includes('Already Exists')) { | |
| console.log('Resource already exists, continuing...'); | |
| return c.json({}); | |
| } | |
| throw error; | |
| } | |
| } catch (error) { | |
| console.error('[SUBSCRIPTION] Error:', error); | |
| // Clean up on error using base class method | |
| // await this.cleanupOnFailure(connectionId, env); | |
| return c.json({ error: 'Internal server error' }, { status: 500 }); | |
| } | |
| } | |
| } catch (error) { | |
| console.error('[SUBSCRIPTION] Setup failed:', error); | |
| // Clean up on failure using base class method | |
| await this.cleanupOnFailure(connectionId, env); | |
| if (error instanceof Error && error.message.includes('Already Exists')) { | |
| console.log('Resource already exists, continuing...'); | |
| await this.setSubscribedState(connectionId, EProviders.google, 'active', env); | |
| return c.json({}); | |
| } | |
| throw error; | |
| } | |
| } catch (error) { | |
| console.error('[SUBSCRIPTION] Error:', error); | |
| // Clean up on error using base class method | |
| await this.cleanupOnFailure(connectionId, env); | |
| return c.json({ error: 'Internal server error' }, { status: 500 }); | |
| } |
| } catch (error) { | |
| console.error('[SUBSCRIPTION] Setup failed:', error); | |
| // Clean up on failure using base class method | |
| // await this.cleanupOnFailure(connectionId, env); | |
| if (error instanceof Error && error.message.includes('Already Exists')) { | |
| console.log('Resource already exists, continuing...'); | |
| return c.json({}); | |
| } | |
| throw error; | |
| } | |
| } catch (error) { | |
| console.error('[SUBSCRIPTION] Error:', error); | |
| // Clean up on error using base class method | |
| // await this.cleanupOnFailure(connectionId, env); | |
| return c.json({ error: 'Internal server error' }, { status: 500 }); | |
| } | |
| } | |
| console.log(`[SUBSCRIPTION] Initializing labels for connection: ${connectionId}`); | |
| await this.initializeConnectionLabels(connectionId); | |
| // Mark subscription as active | |
| await this.setSubscribedState(connectionId, EProviders.google, 'active', env); | |
| console.log(`[SUBSCRIPTION] Setup completed successfully for connection: ${connectionId}`); |
🤖 Prompt for AI Agents
In apps/server/src/lib/factories/google-subscription.factory.ts between lines
283 and 304, uncomment and enable the cleanupOnFailure calls in both catch
blocks to prevent resource leaks on failure. Also, after a successful
subscription setup, add a call to set the subscription state to 'active' to
ensure consistent subscription state management.
1 participant
Implement Gmail Push Notifications with Workflows
This PR implements a comprehensive Gmail push notification system using Cloudflare Workflows. It creates a subscription factory pattern to handle different email providers, with Google implementation completed. The system processes Gmail history updates, vectorizes messages, generates thread summaries, and applies AI-generated labels.
Type of Change
Areas Affected
Testing Done
Checklist
Additional Notes
Key changes:
The system now processes Gmail notifications in real-time, extracts meaningful data, and provides AI-generated labels for better email organization.
Summary by CodeRabbit
New Features
Improvements
Bug Fixes
Chores
Documentation