infrastructure/plunk
2
0
Fork 0
Code Pull Requests Activity

chore: replace standard logging with signale

Browse Source
This commit is contained in:
Dries Augustyns
2025-12-10 14:21:42 +01:00
parent 3225c5005b
commit 1430f31e2d
18 changed files with 85 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/src/jobs/campaign-processor.ts
+6 -5
View File
@@ -4,6 +4,7 @@
*/
import {type Job, Worker} from 'bullmq';
import signale from 'signale';
import {CampaignService} from '../services/CampaignService.js';
import {type CampaignBatchJobData, campaignQueue} from '../services/QueueService.js';
@@ -14,11 +15,11 @@ export function createCampaignWorker() {
async (job: Job<CampaignBatchJobData>) => {
const {campaignId, batchNumber, offset, limit, cursor} = job.data;
console.log(`[CAMPAIGN-PROCESSOR] Processing batch ${batchNumber} for campaign ${campaignId}`);
signale.info(`[CAMPAIGN-PROCESSOR] Processing batch ${batchNumber} for campaign ${campaignId}`);
await CampaignService.processBatch(campaignId, batchNumber, offset, limit, cursor);
console.log(`[CAMPAIGN-PROCESSOR] Completed batch ${batchNumber} for campaign ${campaignId}`);
signale.info(`[CAMPAIGN-PROCESSOR] Completed batch ${batchNumber} for campaign ${campaignId}`);
},
{
connection: campaignQueue.opts.connection,
@@ -27,15 +28,15 @@ export function createCampaignWorker() {
);
worker.on('completed', job => {
console.log(`[CAMPAIGN-PROCESSOR] Job ${job.id} completed`);
signale.info(`[CAMPAIGN-PROCESSOR] Job ${job.id} completed`);
});
worker.on('failed', (job, err) => {
console.error(`[CAMPAIGN-PROCESSOR] Job ${job?.id} failed:`, err.message);
signale.error(`[CAMPAIGN-PROCESSOR] Job ${job?.id} failed:`, err.message);
});
worker.on('error', err => {
console.error('[CAMPAIGN-PROCESSOR] Worker error:', err);
signale.error('[CAMPAIGN-PROCESSOR] Worker error:', err);
});
return worker;
apps/api/src/jobs/email-processor.ts
+10 -9
View File
@@ -5,6 +5,7 @@
import {EmailSourceType, EmailStatus} from '@plunk/db';
import {type Job, Worker} from 'bullmq';
import signale from 'signale';
import {prisma} from '../database/prisma.js';
import {EmailService} from '../services/EmailService.js';
@@ -23,23 +24,23 @@ async function getEmailRateLimit(): Promise<number> {
// If env variable is set, use it (override)
if (EMAIL_RATE_LIMIT_PER_SECOND !== undefined) {
console.log(`[EMAIL-PROCESSOR] Using rate limit from environment: ${EMAIL_RATE_LIMIT_PER_SECOND} emails/second`);
signale.info(`[EMAIL-PROCESSOR] Using rate limit from environment: ${EMAIL_RATE_LIMIT_PER_SECOND} emails/second`);
return EMAIL_RATE_LIMIT_PER_SECOND;
}
// Try to fetch from AWS SES
console.log('[EMAIL-PROCESSOR] Fetching rate limit from AWS SES...');
signale.info('[EMAIL-PROCESSOR] Fetching rate limit from AWS SES...');
const quota = await getSendingQuota();
if (quota) {
console.log(
signale.info(
`[EMAIL-PROCESSOR] AWS SES quota: ${quota.maxSendRate} emails/second (${quota.sentLast24Hours}/${quota.max24HourSend} emails sent today)`,
);
return quota.maxSendRate;
}
// Fallback to safe default
console.warn(`[EMAIL-PROCESSOR] Failed to fetch AWS quota, using safe default: ${DEFAULT_RATE_LIMIT} emails/second`);
signale.warn(`[EMAIL-PROCESSOR] Failed to fetch AWS quota, using safe default: ${DEFAULT_RATE_LIMIT} emails/second`);
return DEFAULT_RATE_LIMIT;
}
@@ -69,7 +70,7 @@ export async function createEmailWorker() {
// Check if project is disabled
if (email.project.disabled) {
console.warn(`[EMAIL-PROCESSOR] Project ${email.projectId} is disabled, cancelling email ${emailId}`);
signale.warn(`[EMAIL-PROCESSOR] Project ${email.projectId} is disabled, cancelling email ${emailId}`);
await prisma.email.update({
where: {id: emailId},
data: {
@@ -170,7 +171,7 @@ export async function createEmailWorker() {
sentAt: new Date().toISOString(),
});
} catch (error) {
console.error(`[EMAIL-PROCESSOR] Failed to send email ${emailId}:`, error);
signale.error(`[EMAIL-PROCESSOR] Failed to send email ${emailId}:`, error);
// Mark as failed
await prisma.email.update({
@@ -195,15 +196,15 @@ export async function createEmailWorker() {
);
worker.on('completed', job => {
console.log(`[EMAIL-PROCESSOR] Job ${job.id} completed`);
signale.info(`[EMAIL-PROCESSOR] Job ${job.id} completed`);
});
worker.on('failed', (job, err) => {
console.error(`[EMAIL-PROCESSOR] Job ${job?.id} failed:`, err.message);
signale.error(`[EMAIL-PROCESSOR] Job ${job?.id} failed:`, err.message);
});
worker.on('error', err => {
console.error('[EMAIL-PROCESSOR] Worker error:', err);
signale.error('[EMAIL-PROCESSOR] Worker error:', err);
});
return worker;
apps/api/src/jobs/import-processor.ts
+8 -7
View File
@@ -5,6 +5,7 @@
import {type Job, Worker} from 'bullmq';
import {parse} from 'csv-parse/sync';
import signale from 'signale';
import {prisma} from '../database/prisma.js';
import {ContactService} from '../services/ContactService.js';
@@ -28,7 +29,7 @@ export function createImportWorker() {
async (job: Job<ContactImportJobData>) => {
const {projectId, csvData, filename} = job.data;
console.log(`[IMPORT-PROCESSOR] Processing import for project ${projectId} (${filename})`);
signale.info(`[IMPORT-PROCESSOR] Processing import for project ${projectId} (${filename})`);
// Fetch project information for notifications
const project = await prisma.project.findUnique({
@@ -66,7 +67,7 @@ export function createImportWorker() {
throw new Error('CSV file is empty');
}
console.log(`[IMPORT-PROCESSOR] Parsed ${records.length} rows from CSV`);
signale.info(`[IMPORT-PROCESSOR] Parsed ${records.length} rows from CSV`);
// Notify that import has started
await NtfyService.notifyContactImportStarted(projectName, projectId, filename, result.totalRows);
@@ -151,7 +152,7 @@ export function createImportWorker() {
await job.updateProgress(progress);
}
console.log(
signale.info(
`[IMPORT-PROCESSOR] Import completed: ${result.createdCount} created, ${result.updatedCount} updated, ${result.failureCount} failed`,
);
@@ -168,7 +169,7 @@ export function createImportWorker() {
return result;
} catch (error) {
console.error(`[IMPORT-PROCESSOR] Failed to process import:`, error);
signale.error(`[IMPORT-PROCESSOR] Failed to process import:`, error);
// Notify that import has failed
const errorMessage = error instanceof Error ? error.message : 'Unknown error';
@@ -191,15 +192,15 @@ export function createImportWorker() {
);
worker.on('completed', job => {
console.log(`[IMPORT-PROCESSOR] Job ${job.id} completed`);
signale.info(`[IMPORT-PROCESSOR] Job ${job.id} completed`);
});
worker.on('failed', (job, err) => {
console.error(`[IMPORT-PROCESSOR] Job ${job?.id} failed:`, err.message);
signale.error(`[IMPORT-PROCESSOR] Job ${job?.id} failed:`, err.message);
});
worker.on('error', err => {
console.error('[IMPORT-PROCESSOR] Worker error:', err);
signale.error('[IMPORT-PROCESSOR] Worker error:', err);
});
return worker;
apps/api/src/jobs/scheduled-processor.ts
+9 -8
View File
@@ -5,6 +5,7 @@
import {CampaignStatus} from '@plunk/db';
import {type Job, Worker} from 'bullmq';
import signale from 'signale';
import {prisma} from '../database/prisma.js';
import {CampaignService} from '../services/CampaignService.js';
@@ -16,7 +17,7 @@ export function createScheduledCampaignWorker() {
async (job: Job<ScheduledCampaignJobData>) => {
const {campaignId} = job.data;
console.log(`[SCHEDULED-PROCESSOR] Processing scheduled campaign ${campaignId}`);
signale.info(`[SCHEDULED-PROCESSOR] Processing scheduled campaign ${campaignId}`);
// Get campaign with project
const campaign = await prisma.campaign.findUnique({
@@ -29,13 +30,13 @@ export function createScheduledCampaignWorker() {
});
if (!campaign) {
console.warn(`[SCHEDULED-PROCESSOR] Campaign ${campaignId} not found, skipping`);
signale.warn(`[SCHEDULED-PROCESSOR] Campaign ${campaignId} not found, skipping`);
return;
}
// Check if project is disabled
if (campaign.project.disabled) {
console.warn(
signale.warn(
`[SCHEDULED-PROCESSOR] Project ${campaign.projectId} (${campaign.project.name}) is disabled, cancelling campaign ${campaignId}`,
);
await prisma.campaign.update({
@@ -47,7 +48,7 @@ export function createScheduledCampaignWorker() {
// Verify campaign is still in SCHEDULED status
if (campaign.status !== CampaignStatus.SCHEDULED) {
console.warn(
signale.warn(
`[SCHEDULED-PROCESSOR] Campaign ${campaignId} is not in SCHEDULED status (${campaign.status}), skipping`,
);
return;
@@ -56,7 +57,7 @@ export function createScheduledCampaignWorker() {
// Start sending the campaign
await CampaignService.startSending(campaign.projectId, campaignId);
console.log(`[SCHEDULED-PROCESSOR] Started sending campaign ${campaignId}`);
signale.info(`[SCHEDULED-PROCESSOR] Started sending campaign ${campaignId}`);
},
{
connection: scheduledQueue.opts.connection,
@@ -65,15 +66,15 @@ export function createScheduledCampaignWorker() {
);
worker.on('completed', job => {
console.log(`[SCHEDULED-PROCESSOR] Job ${job.id} completed`);
signale.info(`[SCHEDULED-PROCESSOR] Job ${job.id} completed`);
});
worker.on('failed', (job, err) => {
console.error(`[SCHEDULED-PROCESSOR] Job ${job?.id} failed:`, err.message);
signale.error(`[SCHEDULED-PROCESSOR] Job ${job?.id} failed:`, err.message);
});
worker.on('error', err => {
console.error('[SCHEDULED-PROCESSOR] Worker error:', err);
signale.error('[SCHEDULED-PROCESSOR] Worker error:', err);
});
return worker;
apps/api/src/jobs/workflow-processor-queue.ts
+4 -3
View File
@@ -4,6 +4,7 @@
*/
import {type Job, Worker} from 'bullmq';
import signale from 'signale';
import {workflowQueue, type WorkflowStepJobData} from '../services/QueueService.js';
import {WorkflowExecutionService} from '../services/WorkflowExecutionService.js';
@@ -33,15 +34,15 @@ export function createWorkflowWorker() {
);
worker.on('completed', job => {
console.log(`[WORKFLOW-PROCESSOR] Job ${job.id} completed`);
signale.info(`[WORKFLOW-PROCESSOR] Job ${job.id} completed`);
});
worker.on('failed', (job, err) => {
console.error(`[WORKFLOW-PROCESSOR] Job ${job?.id} failed:`, err.message);
signale.error(`[WORKFLOW-PROCESSOR] Job ${job?.id} failed:`, err.message);
});
worker.on('error', err => {
console.error('[WORKFLOW-PROCESSOR] Worker error:', err);
signale.error('[WORKFLOW-PROCESSOR] Worker error:', err);
});
return worker;