Skip to content

chore: remove bullmq #453

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
coodos merged 4 commits into from
Nov 20, 2025
Merged

chore: remove bullmq #453

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
edce0b9
chore: remove bullmq
coodos Nov 20, 2025
83b696d
Merge branch 'main' into chore/remove-bull-mq
coodos Nov 20, 2025
ffad604
chore: redo pnpm lock
coodos Nov 20, 2025
220474f
chore: fix build
coodos Nov 20, 2025
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
2 changes: 0 additions & 2 deletions platforms/eReputation-api/package.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,11 +14,9 @@
},
"dependencies": {
"axios": "^1.6.7",
"bullmq": "^5.3.0",
"cors": "^2.8.5",
"dotenv": "^16.4.5",
"express": "^4.18.2",
"ioredis": "^5.3.2",
"jsonwebtoken": "^9.0.2",
"openai": "^4.20.1",
"pg": "^8.11.3",
Expand Down
151 changes: 127 additions & 24 deletions platforms/eReputation-api/src/controllers/WebhookController.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,11 +3,13 @@ import { UserService } from "../services/UserService";
import { GroupService } from "../services/GroupService";
import { VoteService } from "../services/VoteService";
import { PollService } from "../services/PollService";
import { JobQueueService } from "../services/JobQueueService";
import { VotingReputationService } from "../services/VotingReputationService";
import { MessageService } from "../services/MessageService";
Comment on lines +6 to +7

@coderabbitai coderabbitai Bot Nov 20, 2025 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

⚠️ Potential issue | 🟠 Major

Eagerly constructing VotingReputationService can make OPENAI mandatory for the API to start

Because VotingReputationService’s constructor throws when OPENAI_API_KEY is missing, new WebhookController() will now fail at process startup in environments that don’t configure that key (e.g., local/dev or tests that don’t need eReputation). Previously, this dependency was likely isolated to the worker/queue path.

Consider either:

  • Lazily instantiating VotingReputationService inside processEReputationWeightedPoll (on first use), or
  • Making the service tolerate a missing key and no-op eReputation, so the rest of the API can still boot.

This keeps the “remove bullmq” goal while avoiding a new hard startup dependency on OpenAI.

Also applies to: 12-12, 21-22, 25-25, 30-31

🤖 Prompt for AI Agents 
In platforms/eReputation-api/src/controllers/WebhookController.ts (lines 6-7,
12, 21-22, 25, 30-31): the controller currently eagerly constructs
VotingReputationService which throws if OPENAI_API_KEY is missing, preventing
app startup; change to lazily instantiate VotingReputationService only inside
processEReputationWeightedPoll (create on first use) or update the service to
detect a missing OPENAI_API_KEY and become a no-op implementation (return early
or mock methods) so construction won’t throw; ensure all places that reference
the eager instance use the lazily-created instance (guard with a null check and
create if undefined) and add a clear comment about the lazy/no-op behavior so
non-eReputation environments can boot without OpenAI configured.

@coodos coodos Nov 20, 2025

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

nack, works as intended

@coderabbitai coderabbitai Bot Nov 20, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

@coodos, understood! Thanks for clarifying that the eager instantiation and the OPENAI_API_KEY requirement at startup are intentional design decisions.

✏️ Learnings added 
Learnt from: coodos
Repo: MetaState-Prototype-Project/prototype PR: 453
File: platforms/eReputation-api/src/controllers/WebhookController.ts:6-7
Timestamp: 2025-11-20T12:00:02.259Z
Learning: In platforms/eReputation-api/src/controllers/WebhookController.ts, the eager instantiation of VotingReputationService in the constructor (which requires OPENAI_API_KEY at startup) is intentional behavior and works as intended.

Note: Learnings are effective only in the context of similar code segments. To apply general coding guidelines, please configure review instructions. You can manage existing learnings in the app.

import { adapter } from "../web3adapter/watchers/subscriber";
import { User } from "../database/entities/User";
import { Group } from "../database/entities/Group";
import { Poll } from "../database/entities/Poll";
import { VoteReputationResult } from "../database/entities/VoteReputationResult";
import { AppDataSource } from "../database/data-source";
import axios from "axios";

Expand All @@ -16,15 +18,17 @@ export class WebhookController {
groupService: GroupService;
voteService: VoteService;
pollService: PollService;
jobQueueService: JobQueueService;
votingReputationService: VotingReputationService;
messageService: MessageService;
adapter: typeof adapter;

constructor(jobQueueService: JobQueueService) {
constructor() {
this.userService = new UserService();
this.groupService = new GroupService();
this.voteService = new VoteService();
this.pollService = new PollService();
this.jobQueueService = jobQueueService;
this.votingReputationService = new VotingReputationService();
this.messageService = new MessageService();
this.adapter = adapter;
}

Expand Down Expand Up @@ -261,17 +265,11 @@ export class WebhookController {
await pollRepository.save(poll);
finalLocalId = poll.id;

// Enqueue reputation calculation job if needed
// Process eReputation calculation if needed
if (this.voteService.isEReputationWeighted(poll) && poll.groupId) {
try {
await this.jobQueueService.enqueuePollReputationJob(
poll.id,
poll.groupId,
globalId // Use globalId as eventId for idempotency
);
} catch (error) {
console.error(`Failed to enqueue reputation job for poll ${poll.id}:`, error);
}
this.processEReputationWeightedPoll(poll).catch((error) => {
console.error(`Error processing eReputation for poll ${poll.id}:`, error);
});
}
}
} else {
Expand All @@ -297,17 +295,11 @@ export class WebhookController {
});
finalLocalId = savedPoll.id;

// Enqueue reputation calculation job if needed
// Process eReputation calculation if needed
if (this.voteService.isEReputationWeighted(savedPoll) && savedPoll.groupId) {
try {
await this.jobQueueService.enqueuePollReputationJob(
savedPoll.id,
savedPoll.groupId,
globalId // Use globalId as eventId for idempotency
);
} catch (error) {
console.error(`Failed to enqueue reputation job for poll ${savedPoll.id}:`, error);
}
this.processEReputationWeightedPoll(savedPoll).catch((error) => {
console.error(`Error processing eReputation for poll ${savedPoll.id}:`, error);
});
}
}
}
Expand All @@ -319,4 +311,115 @@ export class WebhookController {
}
};

private async processEReputationWeightedPoll(poll: Poll): Promise<void> {
if (!poll.groupId) return;

const group = await this.groupService.getGroupById(poll.groupId);
if (!group || !group.charter) return;

const reputationResults = await this.votingReputationService.calculateGroupMemberReputations(
poll.groupId,
group.charter
);

const voteReputationResult = await this.votingReputationService.saveReputationResults(
poll.id,
poll.groupId,
reputationResults
);

await this.transmitReputationResults(voteReputationResult, poll, group);
await this.createReputationMessage(poll, reputationResults);
}

private async transmitReputationResults(
result: VoteReputationResult,
poll: Poll,
group: Group
): Promise<void> {
const voteReputationResultRepository = AppDataSource.getRepository(VoteReputationResult);
const pollRepository = AppDataSource.getRepository(Poll);
const groupRepository = AppDataSource.getRepository(Group);

const reloadedResult = await voteReputationResultRepository.findOne({
where: { id: result.id },
relations: ["poll", "group"]
});

if (!reloadedResult) throw new Error(`Result not found: ${result.id}`);

let pollEntity: Poll | null = reloadedResult.poll || null;
let groupEntity: Group | null = reloadedResult.group || null;

if (!pollEntity && reloadedResult.pollId) {
pollEntity = await pollRepository.findOne({ where: { id: reloadedResult.pollId } });
}

if (!groupEntity && reloadedResult.groupId) {
groupEntity = await groupRepository.findOne({
where: { id: reloadedResult.groupId },
select: ["id", "ename", "name"]
});
}

if (!pollEntity || !groupEntity || !groupEntity.ename) {
throw new Error("Missing required data for transmission");
}

const data: any = {
id: reloadedResult.id,
pollId: reloadedResult.pollId,
groupId: reloadedResult.groupId,
results: JSON.stringify(reloadedResult.results),
createdAt: reloadedResult.createdAt,
updatedAt: reloadedResult.updatedAt,
poll: {
id: pollEntity.id,
groupId: pollEntity.groupId,
group: {
id: groupEntity.id,
ename: groupEntity.ename,
name: groupEntity.name
}
}
};

if (!data.groupId) {
data.groupId = groupEntity.id;
}

await this.adapter.handleChange({
data,
tableName: "vote_reputation_results"
});
}

private async createReputationMessage(
poll: Poll,
reputationResults: Array<{ ename: string; score: number; justification: string }>
): Promise<void> {
if (!poll.groupId) return;

try {
const messageLines: string[] = [];
messageLines.push(`eReputation scores calculated for poll: "${poll.title}"`);
messageLines.push(``);

for (const result of reputationResults) {
const user = await this.userService.getUserByEname(result.ename);
const userName = user?.name || "Unknown";
messageLines.push(`${userName} (@${result.ename}): ${result.score}/5`);
messageLines.push(` ${result.justification}`);
messageLines.push(``);
}

await this.messageService.createSystemMessage({
text: messageLines.join('\n'),
groupId: poll.groupId,
voteId: poll.id
});
} catch (error) {
console.error(`Failed to create system message for poll ${poll.id}:`, error);
}
}
}
8 changes: 1 addition & 7 deletions platforms/eReputation-api/src/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,8 +15,6 @@ import { GroupController } from "./controllers/GroupController";
import { DashboardController } from "./controllers/DashboardController";
import { authMiddleware, authGuard } from "./middleware/auth";
import { adapter } from "./web3adapter/watchers/subscriber";
import { JobQueueService } from "./services/JobQueueService";
import { PollReputationWorker } from "./workers/PollReputationWorker";

config({ path: path.resolve(__dirname, "../../../.env") });

Expand Down Expand Up @@ -51,14 +49,10 @@ app.use(
app.use(express.json({ limit: "50mb" }));
app.use(express.urlencoded({ limit: "50mb", extended: true }));

// Initialize job queue and worker
const jobQueueService = new JobQueueService();
const pollReputationWorker = new PollReputationWorker(jobQueueService);

// Controllers
const userController = new UserController();
const authController = new AuthController();
const webhookController = new WebhookController(jobQueueService);
const webhookController = new WebhookController();
const referenceController = new ReferenceController();
const calculationController = new CalculationController();
const platformController = new PlatformController();
Expand Down
100 changes: 0 additions & 100 deletions platforms/eReputation-api/src/services/JobQueueService.ts
View file
Open in desktop

This file was deleted.

Loading