"use strict";
var __decorate = (this && this.__decorate) || function (decorators, target, key, desc) {
var c = arguments.length, r = c < 3 ? target : desc === null ? desc = Object.getOwnPropertyDescriptor(target, key) : desc, d;
if (typeof Reflect === "object" && typeof Reflect.decorate === "function") r = Reflect.decorate(decorators, target, key, desc);
else for (var i = decorators.length - 1; i >= 0; i--) if (d = decorators[i]) r = (c < 3 ? d(r) : c > 3 ? d(target, key, r) : d(target, key)) || r;
return c > 3 && r && Object.defineProperty(target, key, r), r;
};
var __metadata = (this && this.__metadata) || function (k, v) {
if (typeof Reflect === "object" && typeof Reflect.metadata === "function") return Reflect.metadata(k, v);
};
var __param = (this && this.__param) || function (paramIndex, decorator) {
return function (target, key) { decorator(target, key, paramIndex); }
};
var NetworkProcessor_1;
Object.defineProperty(exports, "__esModule", { value: true });
exports.NetworkQueueEvents = exports.NetworkProcessor = void 0;
const bullmq_1 = require("@nestjs/bullmq");
const common_1 = require("@nestjs/common");
const bullmq_2 = require("bullmq");
const bullmq_3 = require("@nestjs/bullmq");
const bullmq_4 = require("bullmq");
const shared_types_1 = require("@micprotocol/shared-types");
const prisma_service_1 = require("../prisma/prisma.service");
const reputation_service_1 = require("./reputation.service");
const sms_service_1 = require("./sms.service");
const graph_service_1 = require("./graph.service");
const meilisearch_service_1 = require("./meilisearch.service");
const escape_html_1 = require("../common/escape-html");
let NetworkProcessor = NetworkProcessor_1 = class NetworkProcessor extends bullmq_1.WorkerHost {
prisma;
reputationEngine;
smsService;
graphService;
meilisearch;
emailQueue;
logger = new common_1.Logger(NetworkProcessor_1.name);
constructor(prisma, reputationEngine, smsService, graphService, meilisearch, emailQueue) {
super();
this.prisma = prisma;
this.reputationEngine = reputationEngine;
this.smsService = smsService;
this.graphService = graphService;
this.meilisearch = meilisearch;
this.emailQueue = emailQueue;
}
async process(job) {
this.logger.log(`Processing network job ${job.id}: ${job.name}`);
switch (job.name) {
case "booking-request.notify":
return this.handleBookingRequestNotify(job.data);
case "booking-request.response-notify":
return this.handleBookingResponseNotify(job.data);
case "booking-request.accepted":
return this.handleBookingAccepted(job.data);
case "reputation.recalculate":
return this.handleReputationRecalculate(job.data);
case "graph.ingest-profile":
return this.handleGraphIngest(job.data);
case "digest.weekly":
return this.handleWeeklyDigest(job.data);
case "search.reindex":
return this.handleSearchReindex();
case "search.index-profile":
return this.handleSearchIndexProfile(job.data);
case "search.remove-profile":
return this.handleSearchRemoveProfile(job.data);
default:
this.logger.warn(`Unknown job type: ${job.name}`);
return { success: false, error: "Unknown job type" };
}
}
extractFirstProposedDate(proposedDates) {
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;
}
async handleBookingRequestNotify(data) {
this.logger.log(`Sending booking request notification to ${data.guestEmail}`);
const channels = [];
try {
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";
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);
}
}
try {
const safePodcastTitle = (0, escape_html_1.escapeHtml)(data.podcastTitle);
const safeMessage = (0, escape_html_1.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}
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;
}
}
async handleBookingResponseNotify(data) {
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");
}
const req = request;
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}`;
if (hostEmails.length > 0) {
const safeGuestName = (0, escape_html_1.escapeHtml)(guestName);
const safePodcastTitle = (0, escape_html_1.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],
subject: `Guest ${statusText}: ${guestName} — ${podcastTitle}`,
bodyHtml: `
${headerText}
${safeGuestName} has ${statusText} your booking request for
${safePodcastTitle}.
${bodyPara}
`,
bodyText: `Guest ${statusText}: ${guestName} for ${podcastTitle}.\n\nView: ${actionUrl}`,
});
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)}`);
}
}
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;
}
}
async handleBookingAccepted(data) {
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: shared_types_1.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,
},
},
});
}
const requestMarker = `[network-request:${request.id}]`;
const proposedStart = this.extractFirstProposedDate(request.proposedDates);
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,
},
});
}
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;
}
}
async handleReputationRecalculate(data) {
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})`);
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;
}
}
async handleGraphIngest(data) {
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;
}
}
async handleSearchIndexProfile(data) {
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;
}
}
async handleSearchRemoveProfile(data) {
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;
}
}
async handleSearchReindex() {
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;
}
}
async handleWeeklyDigest(data) {
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 };
}
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",
},
});
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;
}
}
};
exports.NetworkProcessor = NetworkProcessor;
exports.NetworkProcessor = NetworkProcessor = NetworkProcessor_1 = __decorate([
(0, bullmq_1.Processor)("network"),
__param(5, (0, bullmq_3.InjectQueue)("email")),
__metadata("design:paramtypes", [prisma_service_1.PrismaService,
reputation_service_1.ReputationEngine,
sms_service_1.SmsService,
graph_service_1.GraphService,
meilisearch_service_1.MeilisearchService,
bullmq_4.Queue])
], NetworkProcessor);
class NetworkQueueEvents extends bullmq_1.QueueEventsHost {
logger = new common_1.Logger(NetworkQueueEvents.name);
onCompleted(job) {
this.logger.debug(`Job ${job.id} completed successfully`);
}
onFailed(job, error) {
this.logger.error(`Job ${job.id} failed: ${error.message}`);
}
onStalled(jobId) {
this.logger.warn(`Job ${jobId} has stalled`);
}
}
exports.NetworkQueueEvents = NetworkQueueEvents;
__decorate([
(0, bullmq_1.OnQueueEvent)("completed"),
__metadata("design:type", Function),
__metadata("design:paramtypes", [bullmq_2.Job]),
__metadata("design:returntype", void 0)
], NetworkQueueEvents.prototype, "onCompleted", null);
__decorate([
(0, bullmq_1.OnQueueEvent)("failed"),
__metadata("design:type", Function),
__metadata("design:paramtypes", [bullmq_2.Job, Error]),
__metadata("design:returntype", void 0)
], NetworkQueueEvents.prototype, "onFailed", null);
__decorate([
(0, bullmq_1.OnQueueEvent)("stalled"),
__metadata("design:type", Function),
__metadata("design:paramtypes", [String]),
__metadata("design:returntype", void 0)
], NetworkQueueEvents.prototype, "onStalled", null);
//# sourceMappingURL=network.processor.js.map