Skip to content
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion apps/web/app/(ee)/api/cron/partner-platforms/route.ts
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,7 @@ export const POST = withCron(async ({ rawBody }) => {
lastCheckedAt: null,
},
],
// only check partners that are approved or trusted
// only check partners approved/trusted in the network
partner: {
networkStatus: {
in: ["approved", "trusted"],
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,27 @@
import * as z from "zod/v4";

const domainRatingSchema = z.object({
domain_rating: z.object({
domain_rating: z.number(),
}),
});

export async function getDomainRating(target: string) {
const response = await fetch(
`https://api.ahrefs.com/v3/public/domain-rating-free?target=${encodeURIComponent(target)}&output=json`,
{
headers: {
Accept: "application/json",
},
},
);
Comment thread
steven-tey marked this conversation as resolved.

if (!response.ok) {
throw new Error(
`Failed to fetch domain rating for ${target}: ${response.status} ${await response.text()}`,
);
}

const data = await response.json();
return Math.round(domainRatingSchema.parse(data).domain_rating.domain_rating);
}
118 changes: 118 additions & 0 deletions apps/web/app/(ee)/api/cron/partner-platforms/website/route.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,118 @@
import { qstash } from "@/lib/cron";
import { withCron } from "@/lib/cron/with-cron";
import { prisma } from "@dub/prisma";
import { PlatformType } from "@dub/prisma/client";
import { APP_DOMAIN_WITH_NGROK, chunk, getDomainWithoutWWW } from "@dub/utils";
import * as z from "zod/v4";
import { logAndRespond } from "../../utils";
import { getDomainRating } from "./get-domain-rating";

export const dynamic = "force-dynamic";

const BATCH_SIZE = 500;
const CONCURRENCY = 10;

const schema = z.object({
startingAfter: z.string().optional(),
});

/**
* This route is used to update domain rating (DR) for verified website partners using the Ahrefs free API
* Runs once a day at 05:00 AM UTC (cron expression: 0 5 * * *)
* POST /api/cron/partner-platforms/website
*/
export const POST = withCron(async ({ rawBody }) => {
let { startingAfter } = schema.parse(
rawBody ? JSON.parse(rawBody) : { startingAfter: undefined },
);

const websites = await prisma.partnerPlatform.findMany({
where: {
type: PlatformType.website,
verifiedAt: {
not: null,
},
},
take: BATCH_SIZE,
...(startingAfter && {
cursor: {
id: startingAfter,
},
skip: 1,
}),
orderBy: {
id: "asc",
},
});

if (websites.length === 0) {
return logAndRespond(
"No more website platforms found. Finished updating website domain ratings.",
);
}

const websiteChunks = chunk(websites, CONCURRENCY);

for (const websiteChunk of websiteChunks) {
await Promise.allSettled(
websiteChunk.map(async (website) => {
const target =
getDomainWithoutWWW(website.identifier) ?? website.identifier;

if (!target) {
console.error(
`Invalid website identifier for partner platform ${website.id}: ${website.identifier}`,
);
return;
}

try {
const domainRating = await getDomainRating(target);

if (website.subscribers === BigInt(domainRating)) {
console.log(
`No changes to update for ${target} (DR: ${domainRating}), skipping...`,
);
return;
}

await prisma.partnerPlatform.update({
where: {
id: website.id,
},
data: {
subscribers: domainRating,
lastCheckedAt: new Date(),
},
});

console.log(`Updated domain rating for ${target}`, {
domainRating,
});
} catch (error) {
console.error(`Error updating domain rating for ${target}:`, error);
}
}),
);
}

if (websites.length === BATCH_SIZE) {
startingAfter = websites[websites.length - 1].id;

await qstash.publishJSON({
url: `${APP_DOMAIN_WITH_NGROK}/api/cron/partner-platforms/website`,
method: "POST",
body: {
startingAfter,
},
});

return logAndRespond(
`Processed ${BATCH_SIZE} websites. Scheduled next batch (startingAfter: ${startingAfter}).`,
);
}

return logAndRespond(
`Finished updating domain ratings for ${websites.length} websites.`,
);
});
65 changes: 59 additions & 6 deletions apps/web/app/(ee)/api/cron/partner-platforms/youtube/route.ts
Original file line number Diff line number Diff line change
@@ -1,23 +1,35 @@
import { qstash } from "@/lib/cron";
import { withCron } from "@/lib/cron/with-cron";
import { prisma } from "@dub/prisma";
import { PlatformType } from "@dub/prisma/client";
import { chunk } from "@dub/utils";
import { APP_DOMAIN_WITH_NGROK, chunk } from "@dub/utils";
import * as z from "zod/v4";
import { logAndRespond } from "../../utils";
import { youtubeChannelSchema } from "./youtube-channel-schema";

export const dynamic = "force-dynamic";

const BATCH_SIZE = 1000;
const YOUTUBE_API_CHUNK_SIZE = 50;

const schema = z.object({
startingAfter: z.string().optional(),
});

/**
* This route is used to update stats for YouTube verified partners using the YouTube API
* Runs once a day at 06:00 AM UTC (cron expression: 0 6 * * *)
* POST /api/cron/partner-platforms/youtube
*/
export const POST = withCron(async () => {
export const POST = withCron(async ({ rawBody }) => {
if (!process.env.YOUTUBE_API_KEY) {
throw new Error("YOUTUBE_API_KEY is not defined");
}

let { startingAfter } = schema.parse(
rawBody ? JSON.parse(rawBody) : { startingAfter: undefined },
);

const youtubeChannels = await prisma.partnerPlatform.findMany({
where: {
type: PlatformType.youtube,
Expand All @@ -28,15 +40,25 @@ export const POST = withCron(async () => {
not: null,
},
},
take: BATCH_SIZE,
...(startingAfter && {
cursor: {
id: startingAfter,
},
skip: 1,
}),
orderBy: {
id: "asc",
},
});

if (youtubeChannels.length === 0) {
return logAndRespond(
"No YouTube platforms found. Skipping YouTube stats update.",
"No more YouTube platforms found. Finished updating YouTube stats.",
);
}

const channelChunks = chunk(youtubeChannels, 50);
const channelChunks = chunk(youtubeChannels, YOUTUBE_API_CHUNK_SIZE);

for (const channelChunk of channelChunks) {
const channelIds = channelChunk.map((channel) => channel.platformId);
Expand Down Expand Up @@ -79,12 +101,27 @@ export const POST = withCron(async () => {
subscribers: channel.statistics.subscriberCount,
posts: channel.statistics.videoCount,
views: channel.statistics.viewCount,
avatarUrl: channel.snippet?.thumbnails?.default?.url,
avatarUrl: channel.snippet?.thumbnails?.default?.url ?? null,
...(channel.snippet?.customUrl && {
identifier: channel.snippet.customUrl.replace("@", ""),
}),
};

const hasChanges =
partnerPlatform.subscribers !== BigInt(newStats.subscribers) ||
partnerPlatform.posts !== BigInt(newStats.posts) ||
partnerPlatform.views !== BigInt(newStats.views) ||
partnerPlatform.avatarUrl !== newStats.avatarUrl ||
("identifier" in newStats &&
partnerPlatform.identifier !== newStats.identifier);

if (!hasChanges) {
console.log(
`No changes to update for @${partnerPlatform.identifier}, skipping...`,
);
return;
}

await prisma.partnerPlatform.update({
where: {
id: partnerPlatform.id,
Expand All @@ -104,7 +141,23 @@ export const POST = withCron(async () => {
}
}

if (youtubeChannels.length === BATCH_SIZE) {
startingAfter = youtubeChannels[youtubeChannels.length - 1].id;

await qstash.publishJSON({
url: `${APP_DOMAIN_WITH_NGROK}/api/cron/partner-platforms/youtube`,
method: "POST",
body: {
startingAfter,
},
});

return logAndRespond(
`Processed ${BATCH_SIZE} YouTube channels. Scheduled next batch (startingAfter: ${startingAfter}).`,
);
}

return logAndRespond(
`YouTube stats updated for ${youtubeChannels.length} partners`,
`Finished updating YouTube stats for ${youtubeChannels.length} partners.`,
);
});
Loading
Loading