Skip to content

Added Email Integration and complete worker #50

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Vamsi-o merged 2 commits into from
Jan 5, 2026
Merged

Added Email Integration and complete worker #50

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ce631bf
Added Email Integration and complete worker
Vamsi-o Jan 5, 2026
3945fe7
Merge branch 'main' into feature/worker
Vamsi-o Jan 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -488,11 +488,11 @@ See [CONTRIBUTING.md](./CONTRIBUTING.md) for guidelines on:

## 📄 License

[Add your license here]
[MIT license]

## 👥 Authors

- [Your Name/Team]
- [Vamsi , Teja]

---

Expand Down
28 changes: 6 additions & 22 deletions apps/http-backend/src/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,9 +1,4 @@
// import {prismaClient} from '@repo/db/index.js'
// import axios from "axios";
import { prismaClient } from "@repo/db/client";
import cookieParser from 'cookie-parser'


import { NodeRegistry } from "@repo/nodes/nodeClient";
import express from "express";
import { userRouter } from "./routes/userRoutes/userRoutes.js";
Expand All @@ -22,30 +17,19 @@ app.use(cors({

app.use(express.json())
app.use(cookieParser());
// const main = async () => {
// try {
// const users = await prismaClient.user.findMany();
// console.log("Users from DB:", users);
// } catch (err) {
// console.error("Error fetching users:", err);
// }
// };

// main().then(() => {
// console.log("This log is from http Backend");
// });

app.use("/user" , userRouter)
app.use('/node', sheetRouter)
app.use('/google', googleAuth)
app.use('/auth/google', googleAuth) // ← CHANGED THIS LINE!

const PORT= 3002

async function startServer() {
await NodeRegistry.registerAll()
tokenScheduler.start();
app.listen(PORT, () => {
console.log(`Server running on port ${PORT}`);
})
console.log(`Server running on port ${PORT}`);
})
}


startServer()
startServer()
75 changes: 54 additions & 21 deletions apps/http-backend/src/routes/google_callback.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,23 +1,56 @@
import { google } from 'googleapis';
import { GoogleOAuthService } from '@repo/nodes';
import type { OAuthTokens } from '@repo/nodes';
import { userMiddleware } from './userRoutes/userMiddleware.js';
import { Router, Request, Response } from 'express';
import { google } from "googleapis";
import { GoogleOAuthService } from "@repo/nodes";
import type { OAuthTokens } from "@repo/nodes";
import { AuthRequest, userMiddleware } from "./userRoutes/userMiddleware.js";
import { Router, Request, Response } from "express";

export const googleAuth: Router = Router()
export const googleAuth: Router = Router();
googleAuth.get(
"/initiate",
userMiddleware,
async (req: AuthRequest, res: Response) => {
// const userId = req.user?.id || "test_user"; // Get from auth middleware
const userId = req.user?.id;

googleAuth.get('/callback', async(req: Request, res: Response)=>{
const oauth2 = new google.auth.OAuth2(
process.env.GOOGLE_CLIENT_ID,
process.env.GOOGLE_CLIENT_SECRET,
process.env.GOOGLE_REDIRECT_URI ||
"http://localhost:3002/auth/google/callback"
);

const authUrl = oauth2.generateAuthUrl({
access_type: "offline",
scope: [
"https://www.googleapis.com/auth/spreadsheets",
"https://www.googleapis.com/auth/drive.readonly",
"https://www.googleapis.com/auth/gmail.send",
"https://www.googleapis.com/auth/gmail.readonly",
],
state: userId,
prompt: "consent",
});

return res.redirect(authUrl);
}
);
Comment on lines +8 to +36

@coderabbitai coderabbitai Bot Jan 5, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Add explicit check for authenticated user.

While userMiddleware is applied, there's no explicit check if req.user exists before using req.user?.id as the OAuth state parameter (line 30). If req.user is undefined, the state will be undefined, which could cause the callback flow to fail when associating tokens with the user.

🔎 Proposed fix to add explicit authentication check 
   "/initiate",
   userMiddleware,
   async (req: AuthRequest, res: Response) => {
-    // const userId = req.user?.id || "test_user"; // Get from auth middleware
-    const userId = req.user?.id;
+    if (!req.user?.id) {
+      return res.status(401).json({
+        message: "User authentication required"
+      });
+    }
+    const userId = req.user.id;

     const oauth2 = new google.auth.OAuth2(
📝 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.

Suggested change
googleAuth.get(
"/initiate",
userMiddleware,
async (req: AuthRequest, res: Response) => {
// const userId = req.user?.id || "test_user"; // Get from auth middleware
const userId = req.user?.id;
googleAuth.get('/callback', async(req: Request, res: Response)=>{
const oauth2 = new google.auth.OAuth2(
process.env.GOOGLE_CLIENT_ID,
process.env.GOOGLE_CLIENT_SECRET,
process.env.GOOGLE_REDIRECT_URI ||
"http://localhost:3002/auth/google/callback"
);
const authUrl = oauth2.generateAuthUrl({
access_type: "offline",
scope: [
"https://www.googleapis.com/auth/spreadsheets",
"https://www.googleapis.com/auth/drive.readonly",
"https://www.googleapis.com/auth/gmail.send",
"https://www.googleapis.com/auth/gmail.readonly",
],
state: userId,
prompt: "consent",
});
return res.redirect(authUrl);
}
);
googleAuth.get(
"/initiate",
userMiddleware,
async (req: AuthRequest, res: Response) => {
if (!req.user?.id) {
return res.status(401).json({
message: "User authentication required"
});
}
const userId = req.user.id;
const oauth2 = new google.auth.OAuth2(
process.env.GOOGLE_CLIENT_ID,
process.env.GOOGLE_CLIENT_SECRET,
process.env.GOOGLE_REDIRECT_URI ||
"http://localhost:3002/auth/google/callback"
);
const authUrl = oauth2.generateAuthUrl({
access_type: "offline",
scope: [
"https://www.googleapis.com/auth/spreadsheets",
"https://www.googleapis.com/auth/drive.readonly",
"https://www.googleapis.com/auth/gmail.send",
"https://www.googleapis.com/auth/gmail.readonly",
],
state: userId,
prompt: "consent",
});
return res.redirect(authUrl);
}
);
🤖 Prompt for AI Agents 
In @apps/http-backend/src/routes/google_callback.ts around lines 8-36, The route
handler uses req.user?.id as state without verifying authentication; add an
explicit check after userMiddleware to ensure req.user exists (e.g., validate in
the googleAuth.get handler) and return an appropriate 401/403 response if not
authenticated so state is never undefined. Locate the googleAuth.get handler,
the userMiddleware usage, and the userId variable assignment in this file and
guard the call to oauth2.generateAuthUrl (and subsequent res.redirect) with that
check.


googleAuth.get(
"/callback",
userMiddleware,
async (req: Request, res: Response) => {
const code = req.query.code;
const state = req.query.state; // userId
const Oauth = new GoogleOAuthService;
if (!code || typeof code !== 'string') {
return res.json({ error: 'Missing or invalid authorization code' });
const state = req.query.state;
const Oauth = new GoogleOAuthService();
if (!code || typeof code !== "string") {
return res.json({ error: "Missing or invalid authorization code" });
}

const oauth2 = new google.auth.OAuth2(
process.env.GOOGLE_CLIENT_ID,
process.env.GOOGLE_CLIENT_SECRET,
process.env.GOOGLE_REDIRECT_URI
process.env.GOOGLE_CLIENT_ID,
process.env.GOOGLE_CLIENT_SECRET,
process.env.GOOGLE_REDIRECT_URI ||
"http://localhost:3002/auth/google/callback"
);

try {
Expand All @@ -43,13 +76,13 @@ googleAuth.get('/callback', async(req: Request, res: Response)=>{
console.log(' ✅ Tokens saved to database');
}

// Redirect to success page
return res.redirect('http://localhost:3000/workflow');
}
catch (err: any) {
console.error('Google token exchange error:', err);
return res.redirect(
`http://localhost:3000/workflow?google=error&msg=${encodeURIComponent(err?.message ?? 'Token exchange failed')}`);
// Redirect to success page
return res.redirect("http://localhost:3000/workflow");
} catch (err: any) {
console.error("Google token exchange error:", err);
return res.redirect(
`http://localhost:3000/workflow?google=error&msg=${encodeURIComponent(err?.message ?? "Token exchange failed")}`
);
}
})

Expand Down
124 changes: 89 additions & 35 deletions apps/http-backend/src/routes/userRoutes/userRoutes.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -153,51 +153,105 @@ router.get("/getAvailableTriggers",
}
);

// //------------------------------ GET CREDENTIALS -----------------------------

// router.get('/getCredentials/:type',
// userMiddleware,
// async (req: AuthRequest, res) =>{
// try{
// console.log("user from getcredentials: ",req.user)
// if(!req.user){
// return res.status(statusCodes.BAD_REQUEST).json({
// message: "User is not Loggedin"
// })
// }
// const userId = req.user.sub;
// const type = req.params.type
// console.log(userId," -userid")
// if(!type || !userId){
// return res.status(statusCodes.BAD_REQUEST).json({
// message: "Incorrect type Input",
// });
// }
// const executor = new GoogleSheetsNodeExecutor()
// const response = await executor.getAllCredentials(userId,type)
// // console.log( typeof(response));
// // console.log("response: ",response)
// const authUrl = typeof response === 'string' ? response : null
// // console.log(authUrl);

// const credentials = response instanceof Object ? response : null
// // console.log(credentials)
// if(authUrl){
// return res.status(statusCodes.OK).json({
// message: "Credentials not found create credentials using this auth url",
// Data: authUrl,
// });
// }
// else return res.status(statusCodes.OK).json({
// message: "Credentials Fetched succesfully",
// Data: credentials,
// });
// }
// catch(e){
// console.log("Error Fetching the credentials ", e instanceof Error ? e.message : "Unkown reason");
// return res
// .status(statusCodes.INTERNAL_SERVER_ERROR)
// .json({ message: "Internal server from fetching the credentials" });
// }
// }
// );

//------------------------------ GET CREDENTIALS -----------------------------

router.get('/getCredentials/:type',
userMiddleware,
async (req: AuthRequest, res) =>{
try{
console.log("user from getcredentials: ",req.user)
if(!req.user){
return res.status(statusCodes.BAD_REQUEST).json({
message: "User is not Loggedin"
})
}
const userId = req.user.sub;
const type = req.params.type
console.log(userId," -userid")
if(!type || !userId){
return res.status(statusCodes.BAD_REQUEST).json({
message: "Incorrect type Input",
});
async (req: AuthRequest, res) => {
try {
console.log("user from getcredentials: ", req.user)
if (!req.user) {
return res.status(statusCodes.BAD_REQUEST).json({
message: "User is not Loggedin"
})
}
const userId = req.user.sub;

@coderabbitai coderabbitai Bot Jan 5, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🔴 Critical

Inconsistent userId extraction may cause authorization issues.

This route uses req.user.sub while all other routes in this file consistently use req.user.id (lines 81, 297, 346, 373, 404). This inconsistency could prevent users from retrieving their own credentials if the auth middleware populates req.user.id but not req.user.sub, or vice versa.

🔎 Proposed fix to use consistent userId extraction 
-      const userId = req.user.sub;
+      const userId = req.user.id;
📝 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.

Suggested change
const userId = req.user.sub;
const userId = req.user.id;
🤖 Prompt for AI Agents 
In @apps/http-backend/src/routes/userRoutes/userRoutes.ts around line 217,
Replace the inconsistent extraction of the authenticated user's id by changing
the assignment that reads const userId = req.user.sub to use req.user.id instead
so it matches the other handlers in this module (keep the const userId variable
name and use req.user.id consistently across this file).

const type = req.params.type
console.log(userId, " -userid")

if (!type || !userId) {
return res.status(statusCodes.BAD_REQUEST).json({
message: "Incorrect type Input",
});
}

// Check if credentials exist in database
const credentials = await prismaClient.credential.findMany({
where: {
userId: userId,
type : type
}
const executor = new GoogleSheetsNodeExecutor()
const response = await executor.getAllCredentials(userId,type)
// console.log( typeof(response));
// console.log("response: ",response)
const authUrl = typeof response === 'string' ? response : null
// console.log(authUrl);

const credentials = response instanceof Object ? response : null
// console.log(credentials)
if(authUrl){
return res.status(statusCodes.OK).json({
});

if (credentials.length === 0) {
// No credentials found - return the correct auth URL
const authUrl = `${process.env.BACKEND_URL || 'http://localhost:3002'}/auth/google/initiate`;
return res.status(statusCodes.OK).json({
message: "Credentials not found create credentials using this auth url",
Data: authUrl,
});
}
else return res.status(statusCodes.OK).json({
message: "Credentials Fetched succesfully",
Data: credentials,
});
}
catch(e){
console.log("Error Fetching the credentials ", e instanceof Error ? e.message : "Unkown reason");
}

// Credentials found - return them
return res.status(statusCodes.OK).json({
message: "Credentials Fetched successfully",
Data: credentials,
});

} catch (e) {
console.log("Error Fetching the credentials ", e instanceof Error ? e.message : "Unknown reason");
return res
.status(statusCodes.INTERNAL_SERVER_ERROR)
.json({ message: "Internal server from fetching the credentials" });
.json({ message: "Internal server error from fetching the credentials" });
}
}
);
Expand Down
2 changes: 2 additions & 0 deletions apps/worker/package.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,12 +5,14 @@
"main": "index.js",
"scripts": {
"test": "echo \"Error: no test specified\" && exit 1",
"build": "tsc -b",
"dev": "tsc -b && node dist/index.js"
},
"type": "module",
"keywords": [],
"devDependencies": {
"@repo/db": "workspace:*",
"@repo/nodes": "workspace:*",
"@workspace/typescript-config": "workspace:*"
},
"author": "",
Expand Down
72 changes: 72 additions & 0 deletions apps/worker/src/engine/executor.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,72 @@
import { prismaClient } from "@repo/db/client";
import { register } from "./registory.js";
export async function executeWorkflow(
workflowExecutionId: string
): Promise<void> {
console.log(`workflowExecutionId is ${workflowExecutionId}`);
const data = await prismaClient.workflowExecution.findUnique({
where: { id: workflowExecutionId },
include: {
workflow: {
include: {
nodes: {
include: {
AvailableNode: true,
credentials: true,
},
orderBy: { position: "asc" },
},
},
},
},
});
let currentInputData = data?.metadata;
if (!data) {
console.log(`No workflow execution found for id ${workflowExecutionId}`);
return;
}

const update = await prismaClient.workflowExecution.update({

Copilot AI Jan 5, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Unused variable update.

Suggested change
const update = await prismaClient.workflowExecution.update({
await prismaClient.workflowExecution.update({

Copilot uses AI. Check for mistakes.
where: {
id: workflowExecutionId,
},
data: {
status: "InProgress",
},
});

const nodes = data?.workflow.nodes;

for (const node of nodes) {
const nodeType = node.AvailableNode.type;
const context = {
nodeId: node.id,
userId: data.workflow.userId,
credentialId: node.credentials[0]?.id,

Copilot AI Jan 5, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Potential null pointer issue. If no credentials are found for a node (credentials array is empty), this will fail when trying to access 'credentials[0]'. The code should check if credentials exist before accessing them.

Suggested change
credentialId: node.credentials[0]?.id,
credentialId: node.credentials?.[0]?.id,

Copilot uses AI. Check for mistakes.

Copilot AI Jan 5, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Property name mismatch. The context object uses 'credentialId' but the ExecutionContext interface in types.ts defines it as 'credentialsId' (plural). This inconsistency will cause type errors and runtime issues.

Suggested change
credentialId: node.credentials[0]?.id,
credentialsId: node.credentials[0]?.id,

Copilot uses AI. Check for mistakes.
config: node.config as Record<string, any>,
inputData: currentInputData,
};
Comment on lines +42 to +48

@coderabbitai coderabbitai Bot Jan 5, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🔴 Critical

Critical: Field name mismatch - credentialId should be credentialsId.

Line 45 uses credentialId (singular), but the ExecutionContext interface in apps/worker/src/types.ts line 4 defines the field as credentialsId (plural). This breaks the type contract and causes runtime issues when executors try to access the credentials.

🔎 Proposed fix 
 const context = {
   nodeId: node.id,
   userId: data.workflow.userId,
-  credentialId: node.credentials[0]?.id,
+  credentialsId: node.credentials[0]?.id,
   config: node.config as Record<string, any>,
   inputData: currentInputData,
 };
📝 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.

Suggested change
const context = {
nodeId: node.id,
userId: data.workflow.userId,
credentialId: node.credentials[0]?.id,
config: node.config as Record<string, any>,
inputData: currentInputData,
};
const context = {
nodeId: node.id,
userId: data.workflow.userId,
credentialsId: node.credentials[0]?.id,
config: node.config as Record<string, any>,
inputData: currentInputData,
};
🤖 Prompt for AI Agents 
In @apps/worker/src/engine/executor.ts around lines 42-48, The context object in
executor.ts uses the wrong field name `credentialId` (singular) which doesn't
match the `ExecutionContext` interface's `credentialsId` (plural); update the
context literal in the executor (the `context` variable) to use `credentialsId`
and populate it with the same value currently assigned to `credentialId` (e.g.,
node.credentials[0]?.id) so the shape matches the `ExecutionContext` defined in
apps/worker/src/types.ts and executors can access credentials correctly.


const execute = await register.execute(nodeType, context);
if (!execute.success) {
await prismaClient.workflowExecution.update({
where: { id: workflowExecutionId },
data: {
status: "Failed",
error: execute.error,
completedAt: new Date(),
},
});
return;
}
currentInputData = execute.output;
}
const updatedStatus = await prismaClient.workflowExecution.update({
where: { id: workflowExecutionId },
data: {
status: "Completed",
completedAt: new Date(),
},
});
console.log(updatedStatus);
}
Loading