import { Processor, WorkerHost, OnQueueEvent, QueueEventsHost } from "@nestjs/bullmq"; import { Logger } from "@nestjs/common"; import { Job } from "bullmq"; import { InjectQueue } from "@nestjs/bullmq"; import { Queue } from "bullmq"; import { Prisma } from "@prisma/client"; import { GuestJourneyEventType } from "@micprotocol/shared-types"; import { PrismaService } from "../prisma/prisma.service"; import { ReputationEngine } from "./reputation.service"; import { SmsService } from "./sms.service"; import { GraphService } from "./graph.service"; import { MeilisearchService } from "./meilisearch.service"; import { escapeHtml } from "../common/escape-html"; /** * Network Queue Job Types */ export interface BookingRequestNotifyJob { requestId: string; guestEmail: string; guestPhone?: string; guestWhatsapp?: string; podcastTitle: string; message: string; } export interface BookingRequestResponseJob { requestId: string; status: "accepted" | "declined"; podcastId: string; } export interface BookingRequestAcceptedJob { requestId: string; podcastId: string; profileId: string; } export interface ReputationRecalculateJob { profileId: string; trigger: "review" | "booking_completed" | "scheduled"; } export interface GraphIngestJob { profileId: string; userId: string; } export interface WeeklyDigestJob { profileId: string; } export interface SearchIndexProfileJob { profileId: string; } export interface SearchRemoveProfileJob { profileId: string; } /** * Network Queue Processor * * Handles background jobs for the Verified Guest Network: * - Booking request notifications (Email + SMS/WhatsApp) * - Reputation score recalculation * - Social graph updates * - Weekly analytics digests */ @Processor("network") export class NetworkProcessor extends WorkerHost { private readonly logger = new Logger(NetworkProcessor.name); constructor( private readonly prisma: PrismaService, private readonly reputationEngine: ReputationEngine, private readonly smsService: SmsService, private readonly graphService: GraphService, private readonly meilisearch: MeilisearchService, @InjectQueue("email") private readonly emailQueue: Queue, ) { super(); } async process(job: Job): Promise { this.logger.log(`Processing network job ${job.id}: ${job.name}`); switch (job.name) { case "booking-request.notify": return this.handleBookingRequestNotify(job.data as BookingRequestNotifyJob); case "booking-request.response-notify": return this.handleBookingResponseNotify(job.data as BookingRequestResponseJob); case "booking-request.accepted": return this.handleBookingAccepted(job.data as BookingRequestAcceptedJob); case "reputation.recalculate": return this.handleReputationRecalculate(job.data as ReputationRecalculateJob); case "graph.ingest-profile": return this.handleGraphIngest(job.data as GraphIngestJob); case "digest.weekly": return this.handleWeeklyDigest(job.data as WeeklyDigestJob); case "search.reindex": return this.handleSearchReindex(); case "search.index-profile": return this.handleSearchIndexProfile(job.data as SearchIndexProfileJob); case "search.remove-profile": return this.handleSearchRemoveProfile(job.data as SearchRemoveProfileJob); default: this.logger.warn(`Unknown job type: ${job.name}`); return { success: false, error: "Unknown job type" }; } } private extractFirstProposedDate(proposedDates: Prisma.JsonValue): Date | null { if (!Array.isArray(proposedDates)) { return null; } for (const candidate of proposedDates) { if (typeof candidate !== "string") { continue; } const parsedDate = new Date(candidate); if (!Number.isNaN(parsedDate.getTime())) { return parsedDate; } } return null; } /** * Handle booking request notification * Sends Email + SMS/WhatsApp to guest */ private async handleBookingRequestNotify( data: BookingRequestNotifyJob ): Promise<{ success: boolean; channels: string[] }> { this.logger.log(`Sending booking request notification to ${data.guestEmail}`); const channels: string[] = []; try { // Get guest profile for action URL const profile = await this.prisma.guestPublicProfile.findFirst({ where: { user: { email: data.guestEmail } }, select: { vanityUrl: true }, }); const actionUrl = profile ? `${process.env.APP_URL || "https://micprotocol.io"}/network/booking/${data.requestId}` : process.env.APP_URL || "https://micprotocol.io"; // Send SMS/WhatsApp if phone available if (data.guestWhatsapp || data.guestPhone) { const phone = data.guestWhatsapp || data.guestPhone!; const lang = this.smsService.detectLanguage(phone); const result = await this.smsService.sendBookingRequest(phone, { podcastName: data.podcastTitle, message: data.message, actionUrl, }, lang); if (result.success) { channels.push(result.channel); } } // Send email notification try { const safePodcastTitle = escapeHtml(data.podcastTitle); const safeMessage = escapeHtml(data.message); const safeActionUrl = encodeURI(actionUrl); await this.emailQueue.add("send-email", { to: data.guestEmail, subject: `Booking request from ${data.podcastTitle}`, bodyHtml: `

You have a booking request!

${safePodcastTitle} would like to invite you as a guest.

${safeMessage}
View Request

This request was sent via the MicProtocol Verified Guest Network.

`, bodyText: `Booking request from ${data.podcastTitle}: ${data.message}\n\nView: ${actionUrl}`, }); channels.push("email"); } catch (emailError) { this.logger.warn(`Failed to queue booking email: ${emailError instanceof Error ? emailError.message : String(emailError)}`); } this.logger.log(`Notification sent via: ${channels.join(", ")}`); return { success: true, channels }; } catch (error) { this.logger.error(`Failed to send booking notification: ${error}`); throw error; // Will be retried } } /** * Handle booking response notification * Notifies host when guest accepts/declines */ private async handleBookingResponseNotify( data: BookingRequestResponseJob ): Promise<{ success: boolean }> { this.logger.log(`Sending booking response notification for ${data.requestId}`); try { const request = await this.prisma.networkBookingRequest.findUnique({ where: { id: data.requestId }, include: { toProfile: { select: { headline: true, phoneNumber: true, whatsappNumber: true, }, }, fromPodcast: { select: { title: true, workspace: { select: { members: { where: { role: "owner" }, select: { user: { select: { email: true }, }, }, }, }, }, }, }, }, }); if (!request) { throw new Error("Booking request not found"); } // Cast to access included relations const req = request as unknown as { toProfile: { headline: string | null; phoneNumber: string | null; whatsappNumber: string | null } | null; fromPodcast: { title: string; workspace: { members: Array<{ user: { email: string } }>; }; }; }; // Get host emails const hostEmails = req.fromPodcast?.workspace?.members ?.map((m) => m.user.email) .filter(Boolean) || []; const guestName = req.toProfile?.headline?.split("|")[0]?.trim() || "Guest"; const podcastTitle = req.fromPodcast?.title || "the podcast"; const statusText = data.status === "accepted" ? "accepted" : "declined"; const actionUrl = `${process.env.APP_URL || "https://micprotocol.io"}/network/requests/${data.requestId}`; // Send email to hosts if (hostEmails.length > 0) { const safeGuestName = escapeHtml(guestName); const safePodcastTitle = escapeHtml(podcastTitle); const safeActionUrl = encodeURI(actionUrl); const headerColor = data.status === "accepted" ? "#00B894" : "#E17055"; const headerText = data.status === "accepted" ? "Booking Accepted" : "Booking Declined"; const bodyPara = data.status === "accepted" ? `

The episode and booking have been automatically created in your pipeline. Check your dashboard for details.

` : `

You can look for other guests in the Verified Guest Network.

`; try { await this.emailQueue.add("send-email", { to: hostEmails[0], // Primary host subject: `Guest ${statusText}: ${guestName} — ${podcastTitle}`, bodyHtml: `

${headerText}

${safeGuestName} has ${statusText} your booking request for ${safePodcastTitle}.

${bodyPara}
View Details
`, bodyText: `Guest ${statusText}: ${guestName} for ${podcastTitle}.\n\nView: ${actionUrl}`, }); // CC additional hosts for (const email of hostEmails.slice(1)) { await this.emailQueue.add("send-email", { to: email, subject: `Guest ${statusText}: ${guestName} — ${podcastTitle}`, bodyHtml: `

${safeGuestName} has ${statusText} the booking request for ${safePodcastTitle}.

View Details

`, bodyText: `Guest ${statusText}: ${guestName}. View: ${actionUrl}`, }); } } catch (emailError) { this.logger.warn(`Failed to send response email: ${emailError instanceof Error ? emailError.message : String(emailError)}`); } } // Send SMS notification to host if configured const hostPhone = req.toProfile?.whatsappNumber || req.toProfile?.phoneNumber; if (hostPhone) { try { const lang = this.smsService.detectLanguage(hostPhone); await this.smsService.sendBookingRequest(hostPhone, { podcastName: podcastTitle, message: `Your booking request was ${statusText} by ${guestName}.`, actionUrl, }, lang); } catch (smsError) { this.logger.warn(`Failed to send response SMS: ${smsError instanceof Error ? smsError.message : String(smsError)}`); } } return { success: true }; } catch (error) { this.logger.error(`Failed to send response notification: ${error}`); throw error; } } /** * Handle booking accepted * Creates guest and episode in host's podcast pipeline */ private async handleBookingAccepted( data: BookingRequestAcceptedJob ): Promise<{ success: boolean; guestId?: string; episodeId?: string }> { this.logger.log(`Creating pipeline entry for accepted booking ${data.requestId}`); try { const [request, profile] = await Promise.all([ this.prisma.networkBookingRequest.findUnique({ where: { id: data.requestId }, include: { fromPodcast: { select: { title: true }, }, }, }), this.prisma.guestPublicProfile.findUnique({ where: { id: data.profileId }, include: { user: { select: { email: true } } }, }), ]); if (!request || !profile) { throw new Error("Request or profile not found"); } const fullName = profile.headline?.split("|")[0]?.trim() || profile.user.email.split("@")[0] || "Network Guest"; const company = profile.headline?.split("|")[1]?.trim() || null; const existingGuest = await this.prisma.guest.findUnique({ where: { podcastId_email: { podcastId: data.podcastId, email: profile.user.email, }, }, }); const guest = existingGuest ? await this.prisma.guest.update({ where: { id: existingGuest.id }, data: { fullName, company, bio: profile.fullBio, socialLinks: profile.socialProof || {}, status: "scheduled", lastContactAt: new Date(), }, }) : await this.prisma.guest.create({ data: { podcastId: data.podcastId, email: profile.user.email, fullName, company, title: "", bio: profile.fullBio, socialLinks: profile.socialProof || {}, status: "scheduled", lastContactAt: new Date(), }, }); if (!existingGuest || existingGuest.status !== "scheduled") { await this.prisma.guestJourneyEvent.create({ data: { guestId: guest.id, eventType: GuestJourneyEventType.SCHEDULED, title: "Guest marked as scheduled", details: "Guest lifecycle moved to scheduled via accepted network booking.", metadata: { source: "network.processor.handleBookingAccepted", requestId: data.requestId, } as Prisma.InputJsonValue, }, }); } const requestMarker = `[network-request:${request.id}]`; const proposedStart = this.extractFirstProposedDate(request.proposedDates as Prisma.JsonValue); let episode = await this.prisma.episode.findFirst({ where: { podcastId: data.podcastId, guestId: guest.id, hostNotes: { contains: requestMarker, }, }, orderBy: { createdAt: "desc" }, }); if (!episode) { episode = await this.prisma.episode.create({ data: { podcastId: data.podcastId, guestId: guest.id, title: `Network booking: ${fullName}`, status: proposedStart ? "scheduled" : "draft", scheduledAt: proposedStart, hostNotes: `${requestMarker}\n\nNetwork booking request accepted`, }, }); } else { await this.prisma.episode.update({ where: { id: episode.id }, data: { status: "scheduled", scheduledAt: episode.scheduledAt || proposedStart, }, }); } if (proposedStart) { const existingBooking = await this.prisma.booking.findFirst({ where: { episodeId: episode.id }, }); if (!existingBooking) { const proposedEnd = new Date(proposedStart.getTime() + 60 * 60 * 1000); await this.prisma.booking.create({ data: { episodeId: episode.id, startTime: proposedStart, endTime: proposedEnd, timezone: profile.timezone || "UTC", meetLink: null, isConfirmed: false, }, }); } } const existingBookingHistory = await this.prisma.bookingHistory.findFirst({ where: { profileId: data.profileId, episodeId: episode.id, }, }); if (!existingBookingHistory) { await this.prisma.bookingHistory.create({ data: { profileId: data.profileId, episodeId: episode.id, podcastName: request.fromPodcast.title, appearedAt: new Date(), isPublic: true, topicsDiscussed: request.proposedTopics as Prisma.InputJsonValue, }, }); } this.logger.log(`Pipeline entry created: guest=${guest.id}, episode=${episode.id}`); return { success: true, guestId: guest.id, episodeId: episode.id, }; } catch (error) { this.logger.error(`Failed to create pipeline entry: ${error}`); throw error; } } /** * Handle reputation recalculation */ private async handleReputationRecalculate( data: ReputationRecalculateJob ): Promise<{ success: boolean; score?: number; tier?: string }> { this.logger.log(`Recalculating reputation for ${data.profileId} (trigger: ${data.trigger})`); try { const score = await this.reputationEngine.calculate(data.profileId); this.logger.log( `Reputation updated: ${data.profileId} = ${score.composite} (${score.tier})` ); // Update search index with new score const profile = await this.prisma.guestPublicProfile.findUnique({ where: { id: data.profileId }, }); if (profile) { await this.meilisearch.updateProfile(profile); } return { success: true, score: score.composite, tier: score.tier, }; } catch (error) { this.logger.error(`Failed to recalculate reputation: ${error}`); throw error; } } /** * Handle graph ingestion */ private async handleGraphIngest( data: GraphIngestJob ): Promise<{ success: boolean }> { this.logger.log(`Ingesting profile ${data.profileId} into graph`); try { await this.graphService.ingestProfile(data.profileId); return { success: true }; } catch (error) { this.logger.error(`Failed to ingest profile: ${error}`); throw error; } } /** * Handle search index profile */ private async handleSearchIndexProfile( data: SearchIndexProfileJob ): Promise<{ success: boolean }> { this.logger.log(`Indexing profile ${data.profileId} in search`); try { const profile = await this.prisma.guestPublicProfile.findUnique({ where: { id: data.profileId }, include: { expertiseTags: true }, }); if (profile) { await this.meilisearch.indexProfile(profile); } return { success: true }; } catch (error) { this.logger.error(`Failed to index profile: ${error}`); throw error; } } /** * Handle search remove profile */ private async handleSearchRemoveProfile( data: SearchRemoveProfileJob ): Promise<{ success: boolean }> { this.logger.log(`Removing profile ${data.profileId} from search`); try { await this.meilisearch.removeProfile(data.profileId); return { success: true }; } catch (error) { this.logger.error(`Failed to remove profile from search: ${error}`); throw error; } } /** * Handle weekly digest * Sends analytics summary to guests */ private async handleSearchReindex(): Promise<{ indexed: number; errors: number }> { this.logger.log("Starting search reindex via background job"); try { return await this.meilisearch.bulkIndex(); } catch (error) { this.logger.error("Search reindex failed", error); throw error; } } private async handleWeeklyDigest( data: WeeklyDigestJob ): Promise<{ success: boolean }> { this.logger.log(`Sending weekly digest for ${data.profileId}`); try { const profile = await this.prisma.guestPublicProfile.findUnique({ where: { id: data.profileId }, include: { user: { select: { email: true } }, }, }); if (!profile || !profile.isPublic) { return { success: false }; } // Get stats const viewsLastWeek = await this.prisma.profileView.count({ where: { profileId: data.profileId, viewedAt: { gte: new Date(Date.now() - 7 * 24 * 60 * 60 * 1000), }, }, }); const pendingRequests = await this.prisma.networkBookingRequest.count({ where: { toProfileId: data.profileId, status: "pending", }, }); // Send notification if significant activity if (viewsLastWeek > 10 || pendingRequests > 0) { if (profile.whatsappNumber || profile.phoneNumber) { await this.smsService.sendProfileViewNotification( profile.whatsappNumber || profile.phoneNumber!, viewsLastWeek ); } } return { success: true }; } catch (error) { this.logger.error(`Failed to send weekly digest: ${error}`); throw error; } } } /** * Network Queue Events Listener */ export class NetworkQueueEvents extends QueueEventsHost { private readonly logger = new Logger(NetworkQueueEvents.name); @OnQueueEvent("completed") onCompleted(job: Job) { this.logger.debug(`Job ${job.id} completed successfully`); } @OnQueueEvent("failed") onFailed(job: Job, error: Error) { this.logger.error(`Job ${job.id} failed: ${error.message}`); } @OnQueueEvent("stalled") onStalled(jobId: string) { this.logger.warn(`Job ${jobId} has stalled`); } }