Spaces:

Benaah
/

MicProtocol

Running

MicProtocol / apps /api /src /network /network.processor.ts

Upload folder using huggingface_hub

6abfd9d verified about 21 hours ago

26.3 kB

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<any> {
        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: `
<div style="font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif; max-width: 600px; margin: 0 auto; padding: 20px;">
<h2 style="color: #1a1a2e;">You have a booking request!</h2>
<p style="color: #4a4a6a; line-height: 1.6;">
<strong>${safePodcastTitle}</strong> would like to invite you as a guest.
</p>
<blockquote style="border-left: 3px solid #FF6B35; padding-left: 16px; margin: 20px 0; color: #4a4a6a; font-style: italic;">
${safeMessage}
</blockquote>
<div style="text-align: center; margin: 30px 0;">
<a href="${safeActionUrl}"
style="background: #FF6B35; color: white; padding: 12px 24px; text-decoration: none; border-radius: 6px; font-weight: 600;">
View Request
</a>
</div>
<p style="color: #888; font-size: 12px;">
This request was sent via the MicProtocol Verified Guest Network.
</p>
</div>
`,
                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"
                ? `<p style="color: #4a4a6a;">The episode and booking have been automatically created in your pipeline. Check your dashboard for details.</p>`
                : `<p style="color: #4a4a6a;">You can look for other guests in the Verified Guest Network.</p>`;

            try {
                await this.emailQueue.add("send-email", {
                    to: hostEmails[0], // Primary host
                    subject: `Guest ${statusText}: ${guestName} — ${podcastTitle}`,
                    bodyHtml: `
<div style="font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif; max-width: 600px; margin: 0 auto; padding: 20px;">
<h2 style="color: ${headerColor};">
${headerText}
</h2>
<p style="color: #4a4a6a; line-height: 1.6;">
<strong>${safeGuestName}</strong> has ${statusText} your booking request for
<strong>${safePodcastTitle}</strong>.
</p>
${bodyPara}
<div style="text-align: center; margin: 30px 0;">
<a href="${safeActionUrl}"
style="background: #FF6B35; color: white; padding: 12px 24px; text-decoration: none; border-radius: 6px; font-weight: 600;">
View Details
</a>
</div>
</div>
`,
                    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: `
<div style="font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif; max-width: 600px; margin: 0 auto; padding: 20px;">
<p style="color: #4a4a6a;">
<strong>${safeGuestName}</strong> has ${statusText} the booking request for
<strong>${safePodcastTitle}</strong>.
</p>
<p><a href="${safeActionUrl}">View Details</a></p>
</div>
`,
                        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`);
    }
}

	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<any> {
	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: `
	<div style="font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif; max-width: 600px; margin: 0 auto; padding: 20px;">
	<h2 style="color: #1a1a2e;">You have a booking request!</h2>
	<p style="color: #4a4a6a; line-height: 1.6;">
	<strong>${safePodcastTitle}</strong> would like to invite you as a guest.
	</p>
	<blockquote style="border-left: 3px solid #FF6B35; padding-left: 16px; margin: 20px 0; color: #4a4a6a; font-style: italic;">
	${safeMessage}
	</blockquote>
	<div style="text-align: center; margin: 30px 0;">
	<a href="${safeActionUrl}"
	style="background: #FF6B35; color: white; padding: 12px 24px; text-decoration: none; border-radius: 6px; font-weight: 600;">
	View Request
	</a>
	</div>
	<p style="color: #888; font-size: 12px;">
	This request was sent via the MicProtocol Verified Guest Network.
	</p>
	</div>
	`,
	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"
	? `<p style="color: #4a4a6a;">The episode and booking have been automatically created in your pipeline. Check your dashboard for details.</p>`
	: `<p style="color: #4a4a6a;">You can look for other guests in the Verified Guest Network.</p>`;

	try {
	await this.emailQueue.add("send-email", {
	to: hostEmails[0], // Primary host
	subject: `Guest ${statusText}: ${guestName} — ${podcastTitle}`,
	bodyHtml: `
	<div style="font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif; max-width: 600px; margin: 0 auto; padding: 20px;">
	<h2 style="color: ${headerColor};">
	${headerText}
	</h2>
	<p style="color: #4a4a6a; line-height: 1.6;">
	<strong>${safeGuestName}</strong> has ${statusText} your booking request for
	<strong>${safePodcastTitle}</strong>.
	</p>
	${bodyPara}
	<div style="text-align: center; margin: 30px 0;">
	<a href="${safeActionUrl}"
	style="background: #FF6B35; color: white; padding: 12px 24px; text-decoration: none; border-radius: 6px; font-weight: 600;">
	View Details
	</a>
	</div>
	</div>
	`,
	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: `
	<div style="font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif; max-width: 600px; margin: 0 auto; padding: 20px;">
	<p style="color: #4a4a6a;">
	<strong>${safeGuestName}</strong> has ${statusText} the booking request for
	<strong>${safePodcastTitle}</strong>.
	</p>
	<p><a href="${safeActionUrl}">View Details</a></p>
	</div>
	`,
	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`);
	}
	}