admin/changemaker.lite
1
0
Fork 0
Code Issues Packages Projects Releases 31 Wiki Activity

Add CRM activity enrichment, notification bridging, crash-safe scheduled jobs, and quick wins

Browse Source 
Workstream A — CRM & Notifications:
- Add fire-and-forget CRM activity helper (api/src/utils/crm-activity.ts) hooked into
  campaign email, canvass visit, donation, and purchase write sites
- Add 5 operational NotificationType enum values (shift_signup_confirmed, shift_reminder,
  shift_cancelled, canvass_session_summary, reengagement) via Prisma migration
- Bridge notification email queue to in-app notifications for volunteer-facing events
- Extend TYPE_TO_PREF map and NotificationsPage labels for new types

Workstream B — Quick Wins:
- Extract shared role constants (11 roles) to admin/src/utils/role-constants.ts,
  update 4 consuming pages
- Add Ad Analytics sidebar entry in payments submenu
- Gate 6 calendar routes with enableSocialCalendar feature flag
- Add GET /series/:id/count endpoint and fix hardcoded shiftsCount={0} in ShiftsPage
- Add influenceCampaignId to Order model for donation-campaign attribution,
  wire through Stripe checkout metadata

Workstream C — Crash-Safe Scheduled Jobs:
- Create BullMQ scheduled-jobs queue with 10 repeatable job types replacing
  setInterval blocks in server.ts (dynamic imports, concurrency: 2)
- Keep presenceService (1min) and challengeScoringService (5min) as setInterval

Bunker Admin
This commit is contained in:
bunker-admin 2026-03-09 14:15:30 -06:00
parent c192c04c79
commit 900a0affe5
24 changed files with 438 additions and 110 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
admin/src/App.tsx
View File

@ -375,10 +375,10 @@ export default function App() {
<Route path="/volunteer/challenges" element={<ChallengesPage />} />
<Route path="/volunteer/challenges/:id" element={<ChallengeDetailPage />} />
<Route path="/volunteer/tickets" element={<MyTicketsPage />} />
<Route path="/volunteer/calendar/shared/:id" element={<SharedCalendarViewPage />} />
<Route path="/volunteer/calendar/shared" element={<SharedCalendarsPage />} />
<Route path="/volunteer/calendar/friend/:userId" element={<FriendCalendarPage />} />
<Route path="/volunteer/calendar" element={<MyCalendarPage />} />
<Route path="/volunteer/calendar/shared/:id" element={<FeatureGate feature="enableSocialCalendar"><SharedCalendarViewPage /></FeatureGate>} />
<Route path="/volunteer/calendar/shared" element={<FeatureGate feature="enableSocialCalendar"><SharedCalendarsPage /></FeatureGate>} />
<Route path="/volunteer/calendar/friend/:userId" element={<FeatureGate feature="enableSocialCalendar"><FriendCalendarPage /></FeatureGate>} />
<Route path="/volunteer/calendar" element={<FeatureGate feature="enableSocialCalendar"><MyCalendarPage /></FeatureGate>} />
<Route path="/volunteer/*" element={<NotFoundPage />} />
</Route>
@ -807,7 +807,9 @@ export default function App() {
path="scheduling/calendar-views/:id"
element={
<ProtectedRoute requiredRoles={SCHEDULING_ROLES}>
<FeatureGate feature="enableSocialCalendar">
<AdminCalendarViewPage />
</FeatureGate>
</ProtectedRoute>
}
/>
@ -815,7 +817,9 @@ export default function App() {
path="scheduling/calendar"
element={
<ProtectedRoute requiredRoles={SCHEDULING_ROLES}>
<FeatureGate feature="enableSocialCalendar">
<SchedulingCalendarPage />
</FeatureGate>
</ProtectedRoute>
}
/>

1
admin/src/components/AppLayout.tsx
View File

@ -309,6 +309,7 @@ function buildMenuItems(settings: import('@/types/api').SiteSettings | null, use
{ key: '/app/payments/donation-pages', icon: <HeartOutlined />, label: 'Donation Pages' },
{ key: '/app/payments/donations', icon: <DollarOutlined />, label: 'Donation Orders' },
{ key: '/app/payments/ads', icon: <PictureOutlined />, label: 'Gallery Ads' },
{ key: '/app/payments/ads/analytics', icon: <BarChartOutlined />, label: 'Ad Analytics' },
{ key: '/app/payments/settings', icon: <SettingOutlined />, label: 'Settings' },
],
});

17
admin/src/pages/AdminCalendarPage.tsx
View File

@ -17,14 +17,7 @@ import dayjs from 'dayjs';
import { api } from '@/lib/api';
import type { AdminCalendarView } from '@/types/api';
import type { AppOutletContext } from '@/components/AppLayout';
const ROLE_OPTIONS = [
{ label: 'Super Admin', value: 'SUPER_ADMIN' },
{ label: 'Influence Admin', value: 'INFLUENCE_ADMIN' },
{ label: 'Map Admin', value: 'MAP_ADMIN' },
{ label: 'User', value: 'USER' },
{ label: 'Temp', value: 'TEMP' },
];
import { ROLE_COLORS, ROLE_OPTIONS } from '@/utils/role-constants';
const LAYER_TYPE_OPTIONS = [
{ label: 'Shifts', value: 'SHIFTS' },
@ -33,14 +26,6 @@ const LAYER_TYPE_OPTIONS = [
{ label: 'Public Events', value: 'PUBLIC_EVENTS' },
];
const ROLE_COLORS: Record<string, string> = {
SUPER_ADMIN: 'red',
INFLUENCE_ADMIN: 'blue',
MAP_ADMIN: 'green',
USER: 'default',
TEMP: 'orange',
};
export default function AdminCalendarPage() {
const navigate = useNavigate();
const { setPageHeader } = useOutletContext<AppOutletContext>();

9
admin/src/pages/AdminCalendarViewPage.tsx
View File

@ -29,17 +29,10 @@ import type {
AdminCalendarUser,
AdminCalendarItem,
} from '@/types/api';
import { ROLE_COLORS } from '@/utils/role-constants';
const { Title, Text } = Typography;
const ROLE_COLORS: Record<string, string> = {
SUPER_ADMIN: 'red',
INFLUENCE_ADMIN: 'blue',
MAP_ADMIN: 'green',
USER: 'default',
TEMP: 'orange',
};
export default function AdminCalendarViewPage() {
const { id } = useParams<{ id: string }>();
const navigate = useNavigate();

17
admin/src/pages/SchedulingCalendarPage.tsx
View File

@ -24,20 +24,13 @@ import UnifiedCalendar from '@/components/calendar/UnifiedCalendar';
import { api } from '@/lib/api';
import type { UnifiedCalendarItem, AdminCalendarView } from '@/types/api';
import { useNavigate } from 'react-router-dom';
import { ROLE_COLORS, ROLE_OPTIONS } from '@/utils/role-constants';
const { Title, Text } = Typography;
const VIEWS_PANEL_WIDTH = 480;
const FORM_PANEL_WIDTH = 380;
const ROLE_OPTIONS = [
{ label: 'Super Admin', value: 'SUPER_ADMIN' },
{ label: 'Influence Admin', value: 'INFLUENCE_ADMIN' },
{ label: 'Map Admin', value: 'MAP_ADMIN' },
{ label: 'User', value: 'USER' },
{ label: 'Temp', value: 'TEMP' },
];
const LAYER_TYPE_OPTIONS = [
{ label: 'Shifts', value: 'SHIFTS' },
{ label: 'Tickets', value: 'TICKETS' },
@ -45,14 +38,6 @@ const LAYER_TYPE_OPTIONS = [
{ label: 'Public Events', value: 'PUBLIC_EVENTS' },
];
const ROLE_COLORS: Record<string, string> = {
SUPER_ADMIN: 'red',
INFLUENCE_ADMIN: 'blue',
MAP_ADMIN: 'green',
USER: 'default',
TEMP: 'orange',
};
export default function SchedulingCalendarPage() {
const navigate = useNavigate();
const addEventRef = useRef<(() => void) | null>(null);

9
admin/src/pages/ShiftsPage.tsx
View File

@ -121,6 +121,7 @@ export default function ShiftsPage() {
const [activeTab, setActiveTab] = useState<'table' | 'calendar'>('table');
const [editModeModalOpen, setEditModeModalOpen] = useState(false);
const [editingSeriesShift, setEditingSeriesShift] = useState<Shift | null>(null);
const [seriesShiftCount, setSeriesShiftCount] = useState(0);
const [calendarData, setCalendarData] = useState<CalendarData['dates']>({});
const [calendarLoading, setCalendarLoading] = useState(false);
const [currentMonth] = useState(dayjs());
@ -355,6 +356,12 @@ export default function ShiftsPage() {
// Part of a series - show edit mode modal
setEditingSeriesShift(shift);
setEditModeModalOpen(true);
// Fetch series shift count
if (shift.seriesId) {
api.get(`/api/map/shifts/series/${shift.seriesId}/count`)
.then((res) => setSeriesShiftCount(res.data.count ?? 0))
.catch(() => setSeriesShiftCount(0));
}
} else {
// Regular shift or exception - edit normally
openEdit(shift);
@ -1207,7 +1214,7 @@ export default function ShiftsPage() {
}}
onConfirm={handleEditMode}
shiftDate={editingSeriesShift?.date || ''}
shiftsCount={0} // TODO: fetch series shifts count
shiftsCount={seriesShiftCount}
/>
</>
);

19
admin/src/pages/social/SocialGraphPage.tsx
View File

@ -14,25 +14,10 @@ import { ReactFlowProvider } from '@xyflow/react';
import SocialNetworkGraph, { type GraphData } from '@/components/social/SocialNetworkGraph';
import { api } from '@/lib/api';
import type { AppOutletContext } from '@/types/api';
import { ROLE_COLORS, ROLE_FILTER_OPTIONS } from '@/utils/role-constants';
const { Text, Title } = Typography;
const ROLE_COLORS: Record<string, string> = {
SUPER_ADMIN: 'red',
INFLUENCE_ADMIN: 'blue',
MAP_ADMIN: 'green',
USER: 'default',
TEMP: 'orange',
};
const ROLE_OPTIONS = [
{ label: 'All Roles', value: '' },
{ label: 'Super Admin', value: 'SUPER_ADMIN' },
{ label: 'Influence Admin', value: 'INFLUENCE_ADMIN' },
{ label: 'Map Admin', value: 'MAP_ADMIN' },
{ label: 'User', value: 'USER' },
];
type LayoutMode = 'force' | 'radial';
interface SelectedUser {
@ -145,7 +130,7 @@ function GraphPageInner() {
<Select
value={roleFilter}
onChange={setRoleFilter}
options={ROLE_OPTIONS}
options={ROLE_FILTER_OPTIONS}
style={{ width: 150 }}
size="small"
/>

5
admin/src/pages/volunteer/NotificationsPage.tsx
View File

@ -18,6 +18,11 @@ const TYPE_LABELS: Record<string, { label: string; color: string }> = {
upload_rejected: { label: 'Rejected', color: 'red' },
achievement: { label: 'Achievement', color: 'gold' },
system: { label: 'System', color: 'default' },
shift_signup_confirmed: { label: 'Shift Signup', color: 'geekblue' },
shift_reminder: { label: 'Shift Reminder', color: 'purple' },
shift_cancelled: { label: 'Shift Cancelled', color: 'red' },
canvass_session_summary: { label: 'Canvass Summary', color: 'volcano' },
reengagement: { label: 'We Miss You', color: 'magenta' },
};
export default function NotificationsPage() {

37
admin/src/utils/role-constants.ts Normal file
View File

@ -0,0 +1,37 @@
import type { UserRole } from '@/types/api';
/** Tag color for each role (Ant Design Tag color prop values) */
export const ROLE_COLORS: Record<string, string> = {
SUPER_ADMIN: 'red',
INFLUENCE_ADMIN: 'blue',
MAP_ADMIN: 'green',
BROADCAST_ADMIN: 'cyan',
CONTENT_ADMIN: 'geekblue',
MEDIA_ADMIN: 'purple',
PAYMENTS_ADMIN: 'gold',
EVENTS_ADMIN: 'magenta',
SOCIAL_ADMIN: 'volcano',
USER: 'default',
TEMP: 'orange',
};
/** Role options for Select components (no "All" entry) */
export const ROLE_OPTIONS: { label: string; value: UserRole }[] = [
{ label: 'Super Admin', value: 'SUPER_ADMIN' },
{ label: 'Influence Admin', value: 'INFLUENCE_ADMIN' },
{ label: 'Map Admin', value: 'MAP_ADMIN' },
{ label: 'Broadcast Admin', value: 'BROADCAST_ADMIN' },
{ label: 'Content Admin', value: 'CONTENT_ADMIN' },
{ label: 'Media Admin', value: 'MEDIA_ADMIN' },
{ label: 'Payments Admin', value: 'PAYMENTS_ADMIN' },
{ label: 'Events Admin', value: 'EVENTS_ADMIN' },
{ label: 'Social Admin', value: 'SOCIAL_ADMIN' },
{ label: 'User', value: 'USER' },
{ label: 'Temp', value: 'TEMP' },
];
/** Role options with a leading "All Roles" entry for filter dropdowns */
export const ROLE_FILTER_OPTIONS: { label: string; value: string }[] = [
{ label: 'All Roles', value: '' },
...ROLE_OPTIONS,
];

16
api/prisma/migrations/20260309100000_add_notification_types_and_donation_campaign_attribution/migration.sql Normal file
View File

@ -0,0 +1,16 @@
-- AlterEnum
-- Add operational notification types for shift/canvass/reengagement notifications
ALTER TYPE "NotificationType" ADD VALUE 'shift_signup_confirmed';
ALTER TYPE "NotificationType" ADD VALUE 'shift_reminder';
ALTER TYPE "NotificationType" ADD VALUE 'shift_cancelled';
ALTER TYPE "NotificationType" ADD VALUE 'canvass_session_summary';
ALTER TYPE "NotificationType" ADD VALUE 'reengagement';
-- AlterTable: Add campaign attribution to donation orders
ALTER TABLE "orders" ADD COLUMN "influence_campaign_id" TEXT;
-- CreateIndex
CREATE INDEX "idx_orders_influence_campaign" ON "orders"("influence_campaign_id");
-- AddForeignKey
ALTER TABLE "orders" ADD CONSTRAINT "orders_influence_campaign_id_fkey" FOREIGN KEY ("influence_campaign_id") REFERENCES "campaigns"("id") ON DELETE SET NULL ON UPDATE CASCADE;

10
api/prisma/schema.prisma
View File

@ -290,6 +290,7 @@ model Campaign {
smsCampaigns SmsCampaign[] @relation("SmsCampaigns")
stories ImpactStory[] @relation("CampaignStories")
milestones CampaignMilestone[] @relation("CampaignMilestones")
donationOrders Order[] @relation("CampaignDonations")
@@index([moderationStatus])
@@index([isUserGenerated])
@ -1522,6 +1523,12 @@ enum NotificationType {
shared_view_invite
shared_view_accepted
calendar_event_invite
// Operational notification types
shift_signup_confirmed
shift_reminder
shift_cancelled
canvass_session_summary
reengagement
}
// ============================================================================
@ -3472,6 +3479,8 @@ model Order {
product Product? @relation(fields: [productId], references: [id])
donationPageId String? @map("donation_page_id")
donationPage DonationPage? @relation("DonationPageOrders", fields: [donationPageId], references: [id], onDelete: SetNull)
influenceCampaignId String? @map("influence_campaign_id")
influenceCampaign Campaign? @relation("CampaignDonations", fields: [influenceCampaignId], references: [id], onDelete: SetNull)
tickets Ticket[] @relation("TicketOrder")
@@index([userId], map: "idx_orders_user")
@ -3479,6 +3488,7 @@ model Order {
@@index([status], map: "idx_orders_status")
@@index([type], map: "idx_orders_type")
@@index([donationPageId], map: "idx_orders_donation_page")
@@index([influenceCampaignId], map: "idx_orders_influence_campaign")
@@map("orders")
}

9
api/src/modules/influence/campaign-emails/campaign-emails.service.ts
View File

@ -3,6 +3,7 @@ import { prisma } from '../../../config/database';
import { AppError } from '../../../middleware/error-handler';
import { emailQueueService } from '../../../services/email-queue.service';
import { recordCampaignEmail } from '../../../utils/metrics';
import { recordCrmActivity } from '../../../utils/crm-activity';
import { groupService } from '../../social/group.service';
import { achievementsService } from '../../social/achievements.service';
import type { SendCampaignEmailInput, TrackMailtoInput, ListCampaignEmailsInput } from './campaign-emails.schemas';
@ -89,6 +90,14 @@ export const campaignEmailsService = {
recordCampaignEmail(campaign.id);
// CRM activity (fire-and-forget)
recordCrmActivity({
email: data.userEmail,
activityType: 'EMAIL_SENT',
title: `Sent campaign email: ${campaign.title}`,
metadata: { campaignId: campaign.id, campaignSlug: campaign.slug, recipientEmail: data.recipientEmail, emailMethod: data.emailMethod },
}).catch(() => {});
// Social group sync (fire-and-forget)
groupService.syncCampaignTeam(campaign.id).catch(() => {});

16
api/src/modules/map/canvass/canvass.service.ts
View File

@ -5,6 +5,7 @@ import { AppError } from '../../../middleware/error-handler';
import { logger } from '../../../utils/logger';
import { recordLocationQuery } from '../../../utils/metrics';
import { isPointInPolygon, parseGeoJsonPolygon } from '../../../utils/spatial';
import { recordCrmActivity } from '../../../utils/crm-activity';
import { calculateWalkingRoute } from './canvass-route.service';
import { recordCanvassVisit, setActiveCanvassSessions } from '../../../utils/metrics';
import { notificationQueueService } from '../../../services/notification-queue.service';
@ -653,6 +654,21 @@ export const canvassService = {
recordCanvassVisit(data.outcome);
// CRM activity via ContactAddress lookup (fire-and-forget)
prisma.contactAddress.findFirst({
where: { addressId: data.addressId },
select: { contactId: true },
}).then((ca) => {
if (ca) {
recordCrmActivity({
contactId: ca.contactId,
activityType: 'CANVASS_VISIT',
title: `Canvass visit: ${data.outcome}`,
metadata: { addressId: data.addressId, outcome: data.outcome, visitId: visit.id },
}).catch(() => {});
}
}).catch(() => {});
// Achievement check (fire-and-forget)
achievementsService.checkAndUnlock(userId, ['canvass']).catch(() => {});

10
api/src/modules/map/shifts/shift-series.routes.ts
View File

@ -28,6 +28,16 @@ router.post(
}
);
// Get series shift count
router.get('/:id/count', async (req, res, next) => {
try {
const count = await ShiftSeriesService.getShiftCount(req.params.id as string);
res.json({ count });
} catch (error) {
next(error);
}
});
// Get series
router.get('/:id', async (req, res, next) => {
try {

9
api/src/modules/map/shifts/shift-series.service.ts
View File

@ -117,6 +117,15 @@ export class ShiftSeriesService {
};
}
/**
* Get count of non-exception shifts in a series
*/
static async getShiftCount(seriesId: string): Promise<number> {
return prisma.shift.count({
where: { seriesId, isException: false },
});
}
/**
* Get series with all its shifts
*/

3
api/src/modules/payments/donations.service.ts
View File

@ -16,6 +16,7 @@ export const donationsService = {
donationPageId?: string,
donationPageSlug?: string,
donationPageTitle?: string,
campaignId?: string,
) {
const settings = await paymentSettingsService.get();
if (!settings.enableDonations) throw new Error('Donations are currently disabled');
@ -55,6 +56,7 @@ export const donationsService = {
message: message || '',
isAnonymous: isAnonymous ? 'true' : 'false',
donationPageId: donationPageId || '',
campaignId: campaignId || '',
},
});
@ -70,6 +72,7 @@ export const donationsService = {
donorMessage: message || null,
isAnonymous: isAnonymous || false,
donationPageId: donationPageId || null,
influenceCampaignId: campaignId || null,
},
});

6
api/src/modules/payments/payments-public.routes.ts
View File

@ -125,13 +125,17 @@ router.post(
validate(createDonationCheckoutSchema),
async (req: Request, res: Response, next: NextFunction) => {
try {
const { amountCents, email, name, message, isAnonymous } = req.body;
const { amountCents, email, name, message, isAnonymous, campaignId } = req.body;
const result = await donationsService.createDonationCheckout(
amountCents,
email,
name,
message,
isAnonymous,
undefined, // donationPageId
undefined, // donationPageSlug
undefined, // donationPageTitle
campaignId,
);
res.json(result);
} catch (err) {

1
api/src/modules/payments/payments.schemas.ts
View File

@ -89,6 +89,7 @@ export const createDonationCheckoutSchema = z.object({
name: z.string().max(200).optional(),
message: z.string().max(2000).optional(),
isAnonymous: z.boolean().optional(),
campaignId: z.string().optional(),
});
// --- Refund ---

27
api/src/modules/payments/webhook.service.ts
View File

@ -2,6 +2,7 @@ import Stripe from 'stripe';
import { prisma } from '../../config/database';
import { getStripe, getWebhookSecret } from '../../services/stripe.client';
import { logger } from '../../utils/logger';
import { recordCrmActivity } from '../../utils/crm-activity';
import { paymentEmailService } from './payment-email.service';
import { listmonkEventSyncService } from '../../services/listmonk-event-sync.service';
@ -214,6 +215,16 @@ export const webhookService = {
orderId: updatedOrder.id,
}).catch(() => {});
}
// CRM activity (fire-and-forget)
if (updatedOrder.buyerEmail) {
recordCrmActivity({
email: updatedOrder.buyerEmail,
activityType: 'PURCHASE',
title: `Purchased: ${updatedOrder.product?.title || 'Product'}`,
metadata: { orderId: updatedOrder.id, productId: updatedOrder.product ? order.productId : null, amountCents: updatedOrder.amountCAD },
}).catch(() => {});
}
}
},
@ -231,8 +242,9 @@ export const webhookService = {
? session.payment_intent
: (session.payment_intent as { id: string } | null)?.id || null;
// Link to donation page if metadata contains donationPageId (from page-specific checkout)
// Link to donation page and/or campaign if metadata contains them
const donationPageId = session.metadata?.donationPageId || null;
const campaignId = session.metadata?.campaignId || null;
const updateData: Record<string, unknown> = {
status: 'COMPLETED',
stripePaymentIntentId: paymentIntentId,
@ -241,6 +253,9 @@ export const webhookService = {
if (donationPageId && !order.donationPageId) {
updateData.donationPageId = donationPageId;
}
if (campaignId && !order.influenceCampaignId) {
updateData.influenceCampaignId = campaignId;
}
await prisma.order.update({
where: { id: order.id },
@ -274,6 +289,16 @@ export const webhookService = {
orderId: order.id,
}).catch(() => {});
}
// CRM activity (fire-and-forget)
if (order.buyerEmail) {
recordCrmActivity({
email: order.buyerEmail,
activityType: 'DONATION',
title: `Donation: $${(order.amountCAD / 100).toFixed(2)}`,
metadata: { orderId: order.id, amountCents: order.amountCAD },
}).catch(() => {});
}
},
async handleInvoicePaid(invoice: Stripe.Invoice) {

6
api/src/modules/social/notification.service.ts
View File

@ -19,6 +19,12 @@ const TYPE_TO_PREF: Record<string, string> = {
shared_view_invite: 'enableFriendRequests',
shared_view_accepted: 'enableFriendRequests',
calendar_event_invite: 'enableFriendRequests',
// Operational notification types
shift_signup_confirmed: 'enableSystemUpdates',
shift_reminder: 'enableSystemUpdates',
shift_cancelled: 'enableSystemUpdates',
canvass_session_summary: 'enableSystemUpdates',
reengagement: 'enableSystemUpdates',
};
export const notificationService = {

50
api/src/server.ts
View File

@ -62,7 +62,6 @@ import { notificationQueueService } from './services/notification-queue.service'
import { geocodeQueueService } from './services/geocode-queue.service';
import { startProxy, stopProxy } from './services/listmonk-proxy.service';
import { pagesService } from './modules/pages/pages.service';
import { listmonkSyncService } from './services/listmonk-sync.service';
import { canvassService } from './modules/map/canvass/canvass.service';
import { trackingService } from './modules/map/tracking/tracking.service';
import { verificationTokenService } from './services/verification-token.service';
@ -115,6 +114,7 @@ import { presenceService } from './modules/social/presence.service';
import { upgradeService } from './modules/upgrade/upgrade.service';
import { autoUpgradeService } from './services/auto-upgrade.service';
import { calendarFeedQueueService } from './services/calendar-feed-queue.service';
import { scheduledJobsQueueService } from './services/scheduled-jobs-queue.service';
import { WebSocketServer } from 'ws';
import { docsCollabService } from './modules/docs/docs-collab.service';
@ -324,6 +324,7 @@ async function start() {
notificationQueueService.startWorker();
geocodeQueueService.startWorker();
calendarFeedQueueService.startWorker();
scheduledJobsQueueService.startWorker();
startProxy();
// Load SMS config from DB (env fallback for empty fields)
@ -341,47 +342,15 @@ async function start() {
logger.info('SMS integration enabled (Termux API)');
}
// Clean expired verification/reset tokens on startup + hourly
// One-time startup calls (recurring runs handled by scheduled-jobs queue)
verificationTokenService.cleanupExpiredTokens().catch(() => {});
passwordResetTokenService.cleanupExpiredTokens().catch(() => {});
setInterval(() => {
verificationTokenService.cleanupExpiredTokens().catch(() => {});
passwordResetTokenService.cleanupExpiredTokens().catch(() => {});
}, 60 * 60 * 1000);
// Close abandoned canvass sessions on startup + hourly
canvassService.closeAbandonedSessions().catch(() => {});
setInterval(() => {
canvassService.closeAbandonedSessions().catch(() => {});
}, 60 * 60 * 1000);
// Listmonk scheduled full sync (every 6h)
if (env.LISTMONK_SYNC_ENABLED === 'true') {
setInterval(() => {
listmonkSyncService.syncAll().catch(() => {});
}, 6 * 60 * 60 * 1000);
logger.info('Listmonk scheduled full sync enabled (every 6h)');
}
// Clean old tracking data on startup + daily
trackingService.cleanupOldData(30).catch(() => {});
setInterval(() => trackingService.cleanupOldData(30).catch(() => {}), 24 * 60 * 60 * 1000);
// Close stale tracking sessions (no data for 2h) — hourly
trackingService.closeStaleTrackingSessions(120).catch(() => {});
setInterval(() => trackingService.closeStaleTrackingSessions(120).catch(() => {}), 60 * 60 * 1000);
// Clean old docs analytics data on startup + daily (90-day retention)
docsAnalyticsService.cleanupOldData(90).catch(() => {});
setInterval(() => docsAnalyticsService.cleanupOldData(90).catch(() => {}), 24 * 60 * 60 * 1000);
// Volunteer re-engagement scanner — daily
reengagementService.scan().catch(() => {});
setInterval(() => reengagementService.scan().catch(() => {}), 24 * 60 * 60 * 1000);
// Social digest email scanner — daily
socialDigestService.scan().catch(() => {});
setInterval(() => socialDigestService.scan().catch(() => {}), 24 * 60 * 60 * 1000);
// SSE + Presence: mark all users offline on startup, start heartbeat + stale cleanup
presenceService.markAllOffline().catch(() => {});
@ -438,7 +407,7 @@ async function start() {
logger.warn('Startup sync of MkDocs overrides failed:', err);
});
// Validate MkDocs exports on startup
// Validate MkDocs exports on startup (recurring runs handled by scheduled-jobs queue)
pagesService.validateExports()
.then(({ validated, repaired, errors }) => {
if (repaired > 0 || errors.length > 0) {
@ -447,13 +416,6 @@ async function start() {
})
.catch((err) => logger.warn('Validation failed:', err));
// Schedule daily validation
setInterval(() => {
pagesService.validateExports().catch((err) => {
logger.warn('Scheduled validation failed:', err);
});
}, 24 * 60 * 60 * 1000);
const server = app.listen(env.PORT, () => {
logger.info(`API server running on port ${env.PORT} [${env.NODE_ENV}]`);
});
@ -477,9 +439,8 @@ async function start() {
});
});
// Clean stale collab states on startup + daily
// Clean stale collab states on startup (recurring runs handled by scheduled-jobs queue)
docsCollabService.cleanupStaleStates().catch(() => {});
setInterval(() => docsCollabService.cleanupStaleStates().catch(() => {}), 24 * 60 * 60 * 1000);
} catch (err) {
logger.error('Failed to start server:', err);
process.exit(1);
@ -500,6 +461,7 @@ for (const signal of ['SIGTERM', 'SIGINT']) {
await geocodeQueueService.close();
await smsQueueService.close();
await calendarFeedQueueService.close();
await scheduledJobsQueueService.close();
await prisma.$disconnect();
redis.disconnect();
process.exit(0);

40
api/src/services/notification-queue.service.ts
View File

@ -2,6 +2,8 @@ import { Queue, Worker, type Job } from 'bullmq';
import { env } from '../config/env';
import { logger } from '../utils/logger';
import { emailService } from './email.service';
import { prisma } from '../config/database';
import { notificationService } from '../modules/social/notification.service';
// ─── Job Data Types ────────────────────────────────────────────────
@ -117,6 +119,26 @@ type NotificationJobData =
// ─── Queue Service ─────────────────────────────────────────────────
/** Resolve userId from email for in-app notification bridging */
async function resolveUserId(email: string): Promise<string | null> {
const user = await prisma.user.findUnique({ where: { email }, select: { id: true } });
return user?.id ?? null;
}
/** Fire-and-forget in-app notification creation */
function bridgeToInApp(
email: string,
type: 'shift_signup_confirmed' | 'shift_reminder' | 'shift_cancelled' | 'canvass_session_summary' | 'reengagement',
title: string,
message: string,
metadata?: Record<string, unknown>,
) {
resolveUserId(email).then((userId) => {
if (!userId) return;
notificationService.createNotification(userId, type, title, message, metadata);
}).catch((err) => logger.warn('Failed to bridge in-app notification', err));
}
class NotificationQueueService {
private queue: Queue;
private worker: Worker | null = null;
@ -155,9 +177,21 @@ class NotificationQueueService {
break;
case 'volunteer-session-summary':
await emailService.sendVolunteerSessionSummary(data);
bridgeToInApp(
data.volunteerEmail, 'canvass_session_summary',
'Canvass Session Complete',
`You visited ${data.visitCount} addresses in ${data.cutName}`,
{ cutName: data.cutName, visitCount: data.visitCount, durationMinutes: data.durationMinutes },
);
break;
case 'volunteer-cancellation':
await emailService.sendVolunteerCancellationAck(data);
bridgeToInApp(
data.volunteerEmail, 'shift_cancelled',
'Shift Cancelled',
`Your shift "${data.shiftTitle}" on ${data.shiftDate} has been cancelled`,
{ shiftTitle: data.shiftTitle, shiftDate: data.shiftDate },
);
break;
case 'volunteer-shift-reminder':
await emailService.sendShiftDetailsEmail({
@ -173,6 +207,12 @@ class NotificationQueueService {
maxVolunteers: data.maxVolunteers,
shiftStatus: data.shiftStatus,
});
bridgeToInApp(
data.recipientEmail, 'shift_reminder',
'Shift Reminder',
`Reminder: "${data.shiftTitle}" on ${data.shiftDate} at ${data.shiftStartTime}`,
{ shiftTitle: data.shiftTitle, shiftDate: data.shiftDate, shiftLocation: data.shiftLocation },
);
break;
case 'volunteer-shift-thank-you':
await emailService.sendVolunteerShiftThankYou(data);

160
api/src/services/scheduled-jobs-queue.service.ts Normal file
View File

@ -0,0 +1,160 @@
import { Queue, Worker, type Job } from 'bullmq';
import { env } from '../config/env';
import { logger } from '../utils/logger';
const QUEUE_NAME = 'scheduled-jobs';
type ScheduledJobType =
| 'reengagement-scan'
| 'social-digest-scan'
| 'close-abandoned-canvass-sessions'
| 'close-stale-tracking-sessions'
| 'cleanup-tracking-data'
| 'cleanup-docs-analytics'
| 'cleanup-verification-tokens'
| 'listmonk-full-sync'
| 'validate-mkdocs-exports'
| 'cleanup-docs-collab-states';
interface ScheduledJobData {
type: ScheduledJobType;
}
const HOUR = 60 * 60 * 1000;
const JOB_DEFINITIONS: Array<{ type: ScheduledJobType; every: number; conditional?: boolean }> = [
{ type: 'reengagement-scan', every: 24 * HOUR },
{ type: 'social-digest-scan', every: 24 * HOUR },
{ type: 'close-abandoned-canvass-sessions', every: HOUR },
{ type: 'close-stale-tracking-sessions', every: HOUR },
{ type: 'cleanup-tracking-data', every: 24 * HOUR },
{ type: 'cleanup-docs-analytics', every: 24 * HOUR },
{ type: 'cleanup-verification-tokens', every: HOUR },
{ type: 'listmonk-full-sync', every: 6 * HOUR, conditional: true },
{ type: 'validate-mkdocs-exports', every: 24 * HOUR },
{ type: 'cleanup-docs-collab-states', every: 24 * HOUR },
];
async function executeJob(type: ScheduledJobType): Promise<void> {
switch (type) {
case 'reengagement-scan': {
const { reengagementService } = await import('./reengagement.service');
await reengagementService.scan();
break;
}
case 'social-digest-scan': {
const { socialDigestService } = await import('./social-digest.service');
await socialDigestService.scan();
break;
}
case 'close-abandoned-canvass-sessions': {
const { canvassService } = await import('../modules/map/canvass/canvass.service');
await canvassService.closeAbandonedSessions();
break;
}
case 'close-stale-tracking-sessions': {
const { trackingService } = await import('../modules/map/tracking/tracking.service');
await trackingService.closeStaleTrackingSessions(120);
break;
}
case 'cleanup-tracking-data': {
const { trackingService } = await import('../modules/map/tracking/tracking.service');
await trackingService.cleanupOldData(30);
break;
}
case 'cleanup-docs-analytics': {
const { docsAnalyticsService } = await import('../modules/docs-analytics/docs-analytics.service');
await docsAnalyticsService.cleanupOldData(90);
break;
}
case 'cleanup-verification-tokens': {
const { verificationTokenService } = await import('./verification-token.service');
const { passwordResetTokenService } = await import('./password-reset-token.service');
await verificationTokenService.cleanupExpiredTokens();
await passwordResetTokenService.cleanupExpiredTokens();
break;
}
case 'listmonk-full-sync': {
const { listmonkSyncService } = await import('./listmonk-sync.service');
await listmonkSyncService.syncAll();
break;
}
case 'validate-mkdocs-exports': {
const { pagesService } = await import('../modules/pages/pages.service');
await pagesService.validateExports();
break;
}
case 'cleanup-docs-collab-states': {
const { docsCollabService } = await import('../modules/docs/docs-collab.service');
await docsCollabService.cleanupStaleStates();
break;
}
}
}
class ScheduledJobsQueueService {
private queue: Queue;
private worker: Worker | null = null;
constructor() {
this.queue = new Queue(QUEUE_NAME, {
connection: { url: env.REDIS_URL },
defaultJobOptions: {
removeOnComplete: { age: 60 * 60, count: 200 },
removeOnFail: { age: 24 * 60 * 60 },
},
});
}
startWorker() {
// Register repeatable jobs
for (const def of JOB_DEFINITIONS) {
// Skip conditional jobs when their feature is disabled
if (def.type === 'listmonk-full-sync' && env.LISTMONK_SYNC_ENABLED !== 'true') {
continue;
}
this.queue.add(
def.type,
{ type: def.type } satisfies ScheduledJobData,
{
repeat: { every: def.every },
jobId: `scheduled-${def.type}`,
}
);
}
this.worker = new Worker(
QUEUE_NAME,
async (job: Job<ScheduledJobData>) => {
const { type } = job.data;
logger.debug(`Scheduled job starting: ${type}`);
await executeJob(type);
},
{
connection: { url: env.REDIS_URL },
concurrency: 2,
}
);
this.worker.on('completed', (job) => {
logger.debug(`Scheduled job ${job.name} completed`);
});
this.worker.on('failed', (job, err) => {
logger.error(`Scheduled job ${job?.name} failed: ${err.message}`);
});
logger.info('Scheduled jobs queue worker started (10 job types)');
}
async close() {
if (this.worker) {
await this.worker.close();
}
await this.queue.close();
logger.info('Scheduled jobs queue closed');
}
}
export const scheduledJobsQueueService = new ScheduledJobsQueueService();

55
api/src/utils/crm-activity.ts Normal file
View File

@ -0,0 +1,55 @@
import { ContactActivityType } from '@prisma/client';
import { prisma } from '../config/database';
import { logger } from './logger';
interface RecordActivityParams {
userId?: string;
email?: string;
contactId?: string;
activityType: ContactActivityType;
title: string;
description?: string;
metadata?: Record<string, unknown>;
}
/**
* Fire-and-forget CRM activity recorder.
* Resolves a Contact by userId or email, then writes a ContactActivity row.
* Skips silently if no matching Contact is found (anonymous users).
*/
export async function recordCrmActivity(params: RecordActivityParams): Promise<void> {
try {
let contactId = params.contactId;
if (!contactId) {
const conditions: Record<string, unknown>[] = [];
if (params.userId) conditions.push({ userId: params.userId });
if (params.email) conditions.push({ email: params.email });
if (conditions.length === 0) return;
const contact = await prisma.contact.findFirst({
where: {
mergedIntoId: null,
OR: conditions,
},
select: { id: true },
});
if (!contact) return;
contactId = contact.id;
}
await prisma.contactActivity.create({
data: {
contactId,
type: params.activityType,
title: params.title,
description: params.description,
metadata: params.metadata as unknown as import('@prisma/client').Prisma.InputJsonValue,
},
});
} catch (err) {
logger.error('Failed to record CRM activity', { error: err instanceof Error ? err.message : String(err), params: { activityType: params.activityType } });
}
}