Blame for apps/api/src/network/network.processor.ts

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`);
    }
}