import { HttpException, HttpStatus, Injectable } from '@nestjs/common'; import { InjectRepository } from '@nestjs/typeorm'; import { Repository, Like, Not, ILike, LessThan, Brackets, In, IsNull, MoreThanOrEqual } from 'typeorm'; import { Cron, CronExpression } from '@nestjs/schedule'; import { SocialMedias } from 'src/entities/social-medias.entity'; import { FacebookServices } from 'src/common/services/facebook/facebook.service'; import { InstagramServices } from 'src/common/services/instagram/instagram.service'; import { TiktokServices } from 'src/common/services/tiktok/tiktok.service'; import { ApifyServices } from 'src/common/services/apify/apify.service'; import { YoutubeServices } from 'src/common/services/youtube/youtube.service'; import { getLocalIPv4 } from 'src/common/config/network.utils'; import { runWithSimpleLock } from 'src/common/services/redis/runWithSimpleLock'; import { UsersServices } from 'src/common/services/users/users.service'; import { CounterService } from 'src/common/config/counter.service'; import { CampaignService } from 'src/modules/campaign/campaign.services'; import { AiReportService } from 'src/modules/campaign/ai-report.service'; import { InfluencerRegisterService } from 'src/modules/influencer-register/influencer-register.services'; import { InfluencerStatus } from 'src/entities/campaign_influencer.entity'; //import { runWithSimpleLock } from 'src/common/services/redis/lock.helper'; const moment = require('moment'); @Injectable() export class CronService { constructor( private facebookServices: FacebookServices, private instagramServices: InstagramServices, private tiktokServices: TiktokServices, private apifyService: ApifyServices, private youtubeServices: YoutubeServices, private userService: UsersServices, private campaignService: CampaignService, private aiReportService: AiReportService, private influencerRegisterService: InfluencerRegisterService, @InjectRepository(SocialMedias) private SocialMedias: Repository, ) { } private manualRunCount = 0; async checkEnvironment() { if (this.manualRunCount > 0) { return true; } return process.env.NODE_ENV === 'production'; } runJobs() { this.manualRunCount = 1; this.instagramServices.syncInstagramInsights(); } /** * Refreshes the access token of all social media accounts which have expired * by querying the database for all social media accounts whose access token * expires today. * * This function is a cron job that runs everyday at midnight to refresh * the access tokens of all social media accounts. * @returns {Promise} */ @Cron(CronExpression.EVERY_DAY_AT_1AM) async RefreshAccessToken() { await runWithSimpleLock('refresh:access-token', 60000, async () => { const today = moment.utc().format('YYYY-MM-DD'); const socialMedias = await this.SocialMedias .createQueryBuilder('socialMedia') .where('DATE(socialMedia.access_token_expire_at) <= :today', { today }) .andWhere('socialMedia.is_token_expire = :isTokenExpire', { isTokenExpire: false }) .getMany(); for (const socialMedia of socialMedias) { switch (socialMedia.social_type) { case 'youtube': break; case 'facebook': await this.facebookServices.socialMediaRefreshToken(socialMedia); break; case 'instagram': await this.instagramServices.refreshToken(socialMedia); break; case 'tiktok': await this.tiktokServices.refreshToken(socialMedia); break; default: console.log('WRONG SOCIAL MEDIA TYPE') } } console.log('ACCESS TOKEN REFRESHED'); }); } /** * Runs the instagram Apify Scraper to scrape all instagram token expired accounts which are of type instagram. * This function is a cron job that runs every *****. */ @Cron(CronExpression.EVERY_DAY_AT_1AM) async InstagramApifyScraper() { if (!(await this.checkEnvironment())) return; await runWithSimpleLock('instagram:scraper', 60000, async () => { const page = 0; const perPage = 100; await this.apifyService.instagramScraper(perPage); console.log('INSTAGRAM SCRAPED'); }); } /** * Runs the facebook Apify Scraper to scrape all facebook token expired accounts which are of type facebook. * This function is a cron job that runs every *****. */ @Cron(CronExpression.EVERY_DAY_AT_1AM) async FacebookApifyScraper() { if (!(await this.checkEnvironment())) return; await runWithSimpleLock('facebook:scraper', 60000, async () => { const page = 0; const perPage = 100; await this.apifyService.facebookScraper(perPage); console.log('FACEBOOK SCRAPfED'); }); } /** * Runs the Tiktok Apify Scraper to scrape all tiktok token expired accounts which are of type Tiktok. * This function is a cron job that runs every *****. */ @Cron(CronExpression.EVERY_DAY_AT_1AM) async TiktokApifyScraper() { if (!(await this.checkEnvironment())) return; await runWithSimpleLock('tiktok:scraper', 60000, async () => { const page = 0; const perPage = 100; await this.apifyService.tiktokScraper(perPage); console.log('TIKTOK SCRAPED'); }); } // @Cron(CronExpression.EVERY_10_SECONDS) // async functionOne() { // this.commonFunction(1); // } // @Cron(CronExpression.EVERY_10_SECONDS) // async functionTwo() { // this.commonFunction(2); // } // async commonFunction(number) { // console.log('Common Function is running every 10 seconds: ' + number); // await runWithSimpleLock('function-one', 60000, async () => { // // Your function logic here // console.log('Function executed: ' + number); // await new Promise(resolve => setTimeout(resolve, 40000)); // Simulate a delay // }); // } @Cron(CronExpression.EVERY_DAY_AT_MIDNIGHT) async InstagramApifyScraperForCountry() { if (!(await this.checkEnvironment())) return; /** * @param country the country of the Instagram accounts PASS TWO DIGIT CODE ONLY * @param condition the condition of the followers count, can be "BETWEEN", "GREATER_THAN", "LESS_THAN", or "EQUAL_TO" * @param days the number of days ago that the records were last updated * @param from the from value of the followers count range * @param to the to value of the followers count range default 0 * */ console.log('INSTAGRAM SCRAPED COUNTRY FILTER START'); await runWithSimpleLock('instagram:scraper-country', 60000, async () => { await this.apifyService.instagramScraperCountryFilter("HK", "BETWEEN", 60, 1000, 10000); await this.apifyService.instagramScraperCountryFilter("HK", "GREATER_THAN", 30, 10000); await this.apifyService.instagramScraperCountryFilter("SG", "BETWEEN", 60, 1000, 10000); await this.apifyService.instagramScraperCountryFilter("SG", "GREATER_THAN", 30, 10000); console.log('INSTAGRAM SCRAPED COUNTRY FILTER'); }); console.log('INSTAGRAM SCRAPED COUNTRY FILTER END'); } @Cron(CronExpression.EVERY_QUARTER) async FacebookApifyScraperForCountry() { if (!(await this.checkEnvironment())) return; /** * @param country the country of the Facebook accounts PASS TWO DIGIT CODE ONLY * @param condition the condition of the followers count, can be "BETWEEN", "GREATER_THAN", "LESS_THAN", or "EQUAL_TO" * @param days the number of days ago that the records were last updated * @param from the from value of the followers count range * @param to the to value of the followers count range default 0 * */ await runWithSimpleLock('facebook:scraper-country', 60000, async () => { await this.apifyService.facebookScraperCountryFilter("HK", "BETWEEN", 30, 1000, 30000); await this.apifyService.facebookScraperCountryFilter("HK", "GREATER_THAN", 10, 30000); await this.apifyService.facebookScraperCountryFilter("SG", "GREATER_THAN", 30, 30000); console.log('FACEBOOK SCRAPED COUNTRY FILTER'); }); } @Cron(CronExpression.EVERY_QUARTER) async TikTokApifyScraperForCountry() { if (!(await this.checkEnvironment())) return; /** * @param country the country of the TIKTOK accounts PASS TWO DIGIT CODE ONLY * @param condition the condition of the followers count, can be "BETWEEN", "GREATER_THAN", "LESS_THAN", or "EQUAL_TO" * @param days the number of days ago that the records were last updated * @param from the from value of the followers count range * @param to the to value of the followers count range default 0 * */ await runWithSimpleLock('tiktok:scraper-country', 60000, async () => { await this.apifyService.tikTokScraperCountryFilter("HK", "BETWEEN", 30, 1000, 30000); await this.apifyService.tikTokScraperCountryFilter("HK", "GREATER_THAN", 10, 30000); await this.apifyService.tikTokScraperCountryFilter("SG", "GREATER_THAN", 30, 30000); console.log('TIKTOK SCRAPED COUNTRY FILTER'); }); } // CALL THIS CRON BEFORE YOUTUBE SCRAPER @Cron(CronExpression.EVERY_DAY_AT_11PM) async getYoutubeChannelId() { if (!(await this.checkEnvironment())) return; await runWithSimpleLock('youtube:get-channel-id', 60000, async () => { await this.youtubeServices.getChannelIdFromName(); console.log('YOUTUBE CHANNEL ID UPDATED'); }); } // SYNC/SCRAPE BOTH YOUTUBE CAHANNEL DATA @Cron(CronExpression.EVERY_DAY_AT_MIDNIGHT) async YoutubeScraper() { if (!(await this.checkEnvironment())) return; await runWithSimpleLock('youtube:scraper', 60000, async () => { const page = 0; const perPage = 25; CounterService.reset(); await this.youtubeServices.youtubeScraper(perPage); console.log('YOUTUBE SCRAPED'); }); } // @Cron(CronExpression.EVERY_QUARTER) @Cron(CronExpression.EVERY_DAY_AT_MIDNIGHT) async YoutubeScraperForCountry() { if (!(await this.checkEnvironment())) return; /** * @param country the country of the YOUTUBE accounts PASS TWO DIGIT CODE ONLY * @param condition the condition of the followers count, can be "BETWEEN", "GREATER_THAN", "LESS_THAN", or "EQUAL_TO" * @param days the number of days ago that the records were last updated * @param from the from value of the followers count range * @param to the to value of the followers count range default 0 * */ // RESET COUNTER BEFORE SCRAPING console.log('YOUTUBE SCRAPED COUNTRY FILTER START'); await runWithSimpleLock('youtube:scraper-country', 60000, async () => { CounterService.reset(); await this.youtubeServices.youtubeScraperCountryFilter("HK", "BETWEEN", 30, 1000, 30000); await this.youtubeServices.youtubeScraperCountryFilter("HK", "GREATER_THAN", 15, 30000); await this.youtubeServices.youtubeScraperCountryFilter("SG", "BETWEEN", 30, 1000, 30000); await this.youtubeServices.youtubeScraperCountryFilter("SG", "GREATER_THAN", 15, 30000); console.log('YOUTUBE SCRAPED COUNTRY FILTER'); }); console.log('YOUTUBE SCRAPED COUNTRY FILTER END'); } @Cron(CronExpression.EVERY_DAY_AT_MIDNIGHT) /** * Syncs the Facebook page data of all social media accounts which are of type facebook. * This function is a cron job that runs every *****. */ async syncFacebookPage() { if (!(await this.checkEnvironment())) return; await runWithSimpleLock('facebook:sync', 60000, async () => { const page = 0; const perPage = 25; await this.facebookServices.syncFacebookPage(page); console.log('SYNC FACEBOOK PAGE'); }); } @Cron(CronExpression.EVERY_DAY_AT_MIDNIGHT) /** * Syncs the Instagram page data of all social media accounts which are of type Instagram. * TThis function is a cron job that runs every *****. */ async syncInstagramProfile() { if (!(await this.checkEnvironment())) return; await runWithSimpleLock('instagram:sync', 60000, async () => { const page = 0; const perPage = 25; await this.instagramServices.syncInstagramProfile(perPage); console.log('SYNC INSTAGRAM PAGE'); }); } @Cron('13 13 * * *') /** * Syncs the Instagram page data of all social media accounts which are of type Instagram. * TThis function is a cron job that runs every *****. */ async checkInstagramToken() { if (!(await this.checkEnvironment())) return; await runWithSimpleLock('instagram:sync', 60000, async () => { const page = 0; const perPage = 1000; await this.instagramServices.checkInstagramToken(perPage); console.log('SYNC INSTAGRAM PAGE'); }); } @Cron(CronExpression.EVERY_DAY_AT_3AM) /** * Fetches Instagram engaged audience demographics insights for accounts with more than 100 followers. * Each account is synced at most once every 30 days. */ async syncInstagramInsights() { if (!(await this.checkEnvironment())) return; await runWithSimpleLock('instagram:insights', 60000, async () => { const perPage = 25; await this.instagramServices.syncInstagramInsights(perPage); console.log('SYNC INSTAGRAM INSIGHTS'); }); } @Cron(CronExpression.EVERY_DAY_AT_2AM) /** * Syncs the Instagram page data of all social media accounts which are of type Instagram. * TThis function is a cron job that runs every *****. */ async updateCampaignPostsInsights() { // if (!(await this.checkEnvironment())) return; await runWithSimpleLock('campaign:update-posts-insights', 60000, async () => { await this.influencerRegisterService.updateCampaignPostsInsights(); console.log('UPDATE CAMPAIGN POSTS INSIGHTS'); }); } //@Cron(CronExpression.EVERY_QUARTER) @Cron(CronExpression.EVERY_DAY_AT_MIDNIGHT) /**sdd * Syncs the Instagram page data of all social media accounts which are of type Instagram. * TThis function is a cron job that runs every *****. */ async syncTiktokProfile() { if (!(await this.checkEnvironment())) return; await runWithSimpleLock('tiktok:sync', 60000, async () => { const page = 0; const perPage = 25; await this.tiktokServices.syncTiktokProfile(perPage); console.log('SYNC INSTAGRAM PAGE'); }); } // @Cron(CronExpression.EVERY_DAY_AT_MIDNIGHT) @Cron(CronExpression.EVERY_DAY_AT_MIDNIGHT) async uploadPostImageOnS3Bucket() { if (!(await this.checkEnvironment())) return; console.log('IMAGE UPLOAD SCRAP'); await runWithSimpleLock('image:upload', 60000, async () => { const page = 0; const perPage = 100; await this.apifyService.uploadPostImageOnS3Bucket(perPage); console.log('IMAGE UPLOAD SCRAPED IN S3 BUCKET'); }); } @Cron(CronExpression.EVERY_DAY_AT_MIDNIGHT) async changeUserPermission() { console.log('USER PERMISSION CHANGE'); await runWithSimpleLock('user:permission', 60000, async () => { await this.userService.changeUserPermission(); console.log('USER PERMISSION CHANGED'); }); } @Cron(CronExpression.EVERY_DAY_AT_8AM) /** * Changes the plan of users from paid to free if the current day is equal to the end date of their subscription. * This function is a cron job that runs every day at 8am. */ async convertToFreePlan() { // if (!(await this.checkEnvironment())) return; console.log('USER CHANGE PAID TO FREE PERMISSION CHANGE'); await runWithSimpleLock('user:permission', 60000, async () => { await this.userService.convertToFreePlan(); console.log('USER CHANGE PAID TO FREE PERMISSION CHANGE'); }); }; /** * Sends notifications to influencer if the campaign stayed at "Posting Deliverable" status for more than 24 hours. * TThis function is a cron job that runs every 10 minutes. */ @Cron(CronExpression.EVERY_10_MINUTES) async campaignInfluencerStatusNotification() { console.log('CAMPAIGN INFLUENCER POSTING DELIVERABLES STATUS NOTIFICATION'); await runWithSimpleLock('campaign:influencer-status-notification', 60000, async () => { await this.campaignService.campaignInfluencerStatusNotification(); console.log('CAMPAIGN INFLUENCER POSTING DELIVERABLES STATUS NOTIFICATION DONE'); }); } @Cron(CronExpression.EVERY_DAY_AT_1AM) async campaignBrandStatusNotification() { console.log('CAMPAIGN BRAND COMPLETION NOTIFICATION'); await runWithSimpleLock('campaign:brand-completion-notification', 60000, async () => { //await this.campaignService.campaignBrandStatusNotification(); await this.campaignService.weeklyCompletionRequestNotification(); console.log('CAMPAIGN BRAND COMPLETION NOTIFICATION DONE'); }); } @Cron(CronExpression.EVERY_DAY_AT_2AM) async dailyApplicationSummaryNotification() { console.log('DAILY APPLICATION SUMMARY NOTIFICATION'); await runWithSimpleLock('campaign:daily-application-summary-notification', 60000, async () => { await this.campaignService.dailyApplicationSummaryNotification(); console.log('DAILY APPLICATION SUMMARY NOTIFICATION DONE'); }); } @Cron(CronExpression.EVERY_10_MINUTES) async campaignBrandDraftStatusNotification() { console.log('Brand Draft Status Notification'); await runWithSimpleLock('campaign:brand-draft-status-notification', 60000, async () => { await this.campaignService.campaignBrandStatusNotification(InfluencerStatus.DRAFT_AVAILABLE, 'draft_available_reminder_notification'); console.log('Brand Draft Status Notification Done'); }); } /** * Sends notifications to brands if the campaign is completed. * This function is a cron job that runs every day at midnight. */ @Cron(CronExpression.EVERY_10_MINUTES) async campaignCompleteStatusNotification() { console.log('CAMPAIGN COMPLETE STATUS NOTIFICATION'); await runWithSimpleLock('campaign:status-complete-notification', 60000, async () => { await this.campaignService.campaignCompleteStatusNotification(); console.log('CAMPAIGN COMPLETE STATUS NOTIFICATION DONE'); }); } @Cron(CronExpression.EVERY_DAY_AT_1AM) async processCompletionRequests30Days() { console.log('PROCESS COMPLETION REQUESTS 30 DAYS - START'); await runWithSimpleLock('campaign:completion-request-30-days', 60000, async () => { try { await this.campaignService.processCompletionRequests30Days(); } catch (err) { console.log('Error processing completion requests 30 days:', err); } }); console.log('PROCESS COMPLETION REQUESTS 30 DAYS - END'); } // @Cron(CronExpression.EVERY_MINUTE) // async processPendingAiReports() { // await runWithSimpleLock('campaign:ai-report-worker', 50000, async () => { // try { // await this.aiReportService.processPendingAiReports(2); // } catch (err) { // console.log('Error processing pending AI reports:', err); // } // }); // } }