Feat/fake trends

src/databases/migrations/1765743134688-addHashtagCreatedAt.ts

@@ -0,0 +1,17 @@
+import { MigrationInterface, QueryRunner } from 'typeorm';
+
+export class AddHashtagCreatedAt1765743134688 implements MigrationInterface {
+    name = 'AddHashtagCreatedAt1765743134688';
+
+    public async up(query_runner: QueryRunner): Promise<void> {
+        await query_runner.query(`ALTER TABLE "hashtag" ADD "category" character varying`);
+        await query_runner.query(
+            `ALTER TABLE "tweet_hashtags" ADD "tweet_created_at" TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT now()`
+        );
+    }
+
+    public async down(query_runner: QueryRunner): Promise<void> {
+        await query_runner.query(`ALTER TABLE "tweet_hashtags" DROP COLUMN "tweet_created_at"`);
+        await query_runner.query(`ALTER TABLE "hashtag" DROP COLUMN "category"`);
+    }
+}

src/migrations/1765743134688-addHashtagCreatedAt.ts

@@ -0,0 +1,17 @@
+import { MigrationInterface, QueryRunner } from 'typeorm';
+
+export class AddHashtagCreatedAt1765743134688 implements MigrationInterface {
+    name = 'AddHashtagCreatedAt1765743134688';
+
+    public async up(query_runner: QueryRunner): Promise<void> {
+        await query_runner.query(`ALTER TABLE "hashtag" ADD "category" character varying`);
+        await query_runner.query(
+            `ALTER TABLE "tweet_hashtags" ADD "tweet_created_at" TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT now()`
+        );
+    }
+
+    public async down(query_runner: QueryRunner): Promise<void> {
+        await query_runner.query(`ALTER TABLE "tweet_hashtags" DROP COLUMN "tweet_created_at"`);
+        await query_runner.query(`ALTER TABLE "hashtag" DROP COLUMN "category"`);
+    }
+}

src/trend/fake-trend.service.spec.ts

@@ -1,18 +1,26 @@
 import { Test, TestingModule } from '@nestjs/testing';
 import { FakeTrendService } from './fake-trend.service';
 import { getRepositoryToken } from '@nestjs/typeorm';
-import { Repository } from 'typeorm';
+import { DataSource, Repository } from 'typeorm';
 import { User } from 'src/user/entities/user.entity';
 import { TweetsService } from 'src/tweets/tweets.service';
 import { TrendDataConstants } from 'src/constants/variables';
 import * as bcrypt from 'bcrypt';
+import { TrendService } from './trend.service';
+import { Hashtag } from 'src/tweets/entities/hashtags.entity';
+import { Tweet } from 'src/tweets/entities';
+import { TweetHashtag } from 'src/tweets/entities/tweet-hashtag.entity';
 
 jest.mock('bcrypt');
 
 describe('FakeTrendService', () => {
     let fake_trend_service: FakeTrendService;
     let user_repo: Repository<User>;
     let tweets_service: TweetsService;
+    let trend_service: TrendService;
+    let hashtag_repo: Repository<Hashtag>;
+    let tweet_hashtag_repo: Repository<TweetHashtag>;
+    let data_source: DataSource;
 
     const mock_repo = (): Record<string, jest.Mock> => ({
         create: jest.fn(),
@@ -49,26 +57,43 @@ describe('FakeTrendService', () => {
             buildDefaultHashtagTopics: jest.fn().mockReturnValue({}),
             deleteTweetsByUserId: jest.fn().mockResolvedValue(undefined),
         };
+        const mock_trend_service = {};
+        const mock_hashtag_repo = mock_repo();
+        const mock_tweet_hashtag_repo = mock_repo();
+        const mock_data_source = {};
 
         const module: TestingModule = await Test.createTestingModule({
             providers: [
                 FakeTrendService,
                 { provide: getRepositoryToken(User), useValue: mock_user_repo },
                 { provide: TweetsService, useValue: mock_tweets_service },
+                { provide: TrendService, useValue: { mock_trend_service } },
+                { provide: getRepositoryToken(Hashtag), useValue: mock_repo() },
+                { provide: getRepositoryToken(TweetHashtag), useValue: mock_repo() },
+                { provide: DataSource, useValue: mock_data_source },
             ],
         }).compile();
 
         fake_trend_service = module.get<FakeTrendService>(FakeTrendService);
         user_repo = mock_user_repo as unknown as Repository<User>;
         tweets_service = module.get<TweetsService>(TweetsService);
+        trend_service = module.get<TrendService>(TrendService);
+        hashtag_repo = module.get<Repository<Hashtag>>(getRepositoryToken(Hashtag));
+        tweet_hashtag_repo = module.get<Repository<TweetHashtag>>(getRepositoryToken(TweetHashtag));
+        data_source = module.get<DataSource>(DataSource);
+    });
 
+    afterEach(() => {
         jest.clearAllMocks();
     });
 
     it('should be defined', () => {
         expect(fake_trend_service).toBeDefined();
         expect(user_repo).toBeDefined();
         expect(tweets_service).toBeDefined();
+        expect(trend_service).toBeDefined();
+        expect(hashtag_repo).toBeDefined();
+        expect(tweet_hashtag_repo).toBeDefined();
     });
 
     describe('insertTrendBotIfNotExists', () => {

src/trend/fake-trend.service.ts

@@ -1,11 +1,15 @@
 import { Injectable, Logger } from '@nestjs/common';
 import { Cron } from '@nestjs/schedule';
 import { InjectRepository } from '@nestjs/typeorm';
-import { Repository } from 'typeorm';
+import { DataSource, Repository } from 'typeorm';
 import { TweetsService } from 'src/tweets/tweets.service';
 import { User } from 'src/user/entities/user.entity';
 import { TrendDataConstants } from 'src/constants/variables';
 import * as bcrypt from 'bcrypt';
+import { Hashtag } from 'src/tweets/entities/hashtags.entity';
+import { TweetHashtag } from 'src/tweets/entities/tweet-hashtag.entity';
+import { TrendService } from './trend.service';
+import { HashtagJobDto } from 'src/background-jobs/hashtag/hashtag-job.dto';
 
 interface IFakeTrendHashtags {
     hashtags: string[];
@@ -20,8 +24,15 @@ export class FakeTrendService {
 
     constructor(
         private readonly tweets_service: TweetsService,
+        private readonly trend_service: TrendService,
+
         @InjectRepository(User)
-        private readonly user_repository: Repository<User>
+        private readonly user_repository: Repository<User>,
+        @InjectRepository(Hashtag)
+        private readonly hashtag_repository: Repository<Hashtag>,
+        private readonly data_source: DataSource,
+        @InjectRepository(TweetHashtag)
+        private readonly tweet_hashtags_repository: Repository<TweetHashtag>
     ) {}
 
     // Every 20 minutes
@@ -194,4 +205,131 @@ export class FakeTrendService {
         const random_template = templates[Math.floor(Math.random() * templates.length)];
         return random_template;
     }
+
+    async seedTrend(): Promise<void> {
+        // UPDATE TWEET TIMESTAMP TO LAST 6 HOURS
+        await this.data_source.query(`
+        UPDATE tweets
+        SET created_at = NOW() - (random() * interval '6 hours')
+    `);
+
+        console.log('Updated tweet timestamps to last 6 hours DONE');
+
+        await this.data_source.query(`
+        UPDATE tweet_hashtags
+        SET tweet_created_at = t.created_at
+        FROM tweets t
+        WHERE tweet_hashtags.tweet_id = t.tweet_id
+    `);
+
+        console.log('Updated tweet_hashtags timestamps to match tweets DONE');
+
+        // SELECT TOP HASHTAGS FROM EACH CATEGORY
+        const sports_hashtags = await this.hashtag_repository.find({
+            where: { category: 'Sports' },
+            order: { usage_count: 'DESC' },
+            take: 30,
+        });
+
+        const entertainment_hashtags = await this.hashtag_repository.find({
+            where: { category: 'Entertainment' },
+            order: { usage_count: 'DESC' },
+            take: 30,
+        });
+
+        const news_hashtags = await this.hashtag_repository.find({
+            where: { category: 'News' },
+            order: { usage_count: 'DESC' },
+            take: 20,
+        });
+
+        console.log('Fetched top hashtags from each category DONE');
+
+        const all_hashtags = [
+            ...sports_hashtags.map((h) => ({ ...h, category: 'Sports' })),
+            ...entertainment_hashtags.map((h) => ({ ...h, category: 'Entertainment' })),
+            ...news_hashtags.map((h) => ({ ...h, category: 'News' })),
+        ];
+
+        const hashtag_names = all_hashtags.map((h) => h.name);
+
+        const tweet_hashtag_data = await this.data_source.query(
+            `
+        SELECT 
+            th.hashtag_name,
+            th.tweet_created_at,
+            h.category
+        FROM tweet_hashtags th
+        JOIN hashtag h ON th.hashtag_name = h.name
+        WHERE th.hashtag_name = ANY($1)
+        ORDER BY th.tweet_created_at DESC
+        `,
+            [hashtag_names]
+        );
+
+        console.log('Fetched tweet hashtag timestamp DONE');
+
+        // Group by tweet timestamp
+        const timestamp_map = new Map<number, Map<string, Record<string, number>>>();
+
+        for (const row of tweet_hashtag_data) {
+            const timestamp = new Date(row.tweet_created_at).getTime();
+            const hashtag_name = row.hashtag_name;
+            const category = row.category;
+
+            if (!timestamp_map.has(timestamp)) {
+                timestamp_map.set(timestamp, new Map());
+            }
+
+            const hashtag_map = timestamp_map.get(timestamp);
+
+            if (hashtag_map) {
+                if (!hashtag_map.has(hashtag_name)) {
+                    hashtag_map.set(hashtag_name, {});
+                }
+
+                const categories = hashtag_map.get(hashtag_name);
+                if (categories) {
+                    categories[category] = 100;
+                }
+            }
+        }
+
+        console.log(`Processing ${timestamp_map.size} unique timestamps`);
+
+        const BATCH_SIZE = 50;
+        const timestamps = Array.from(timestamp_map.entries());
+
+        for (let i = 0; i < timestamps.length; i += BATCH_SIZE) {
+            const batch = timestamps.slice(i, i + BATCH_SIZE);
+
+            console.log(
+                `Processing batch ${Math.floor(i / BATCH_SIZE) + 1}/${Math.ceil(timestamps.length / BATCH_SIZE)}`
+            );
+
+            await Promise.all(
+                batch.map(async ([timestamp, hashtag_map]) => {
+                    const hashtags: Record<string, Record<string, number>> = {};
+
+                    for (const [hashtag_name, categories] of hashtag_map.entries()) {
+                        hashtags[hashtag_name] = categories;
+                    }
+
+                    const job_data: HashtagJobDto = {
+                        hashtags,
+                        timestamp,
+                    };
+
+                    // Execute all three operations in parallel for each timestamp
+                    await Promise.all([
+                        this.trend_service.insertCandidateHashtags(job_data),
+                        this.trend_service.updateHashtagCounts(job_data),
+                        this.trend_service.insertCandidateCategories(job_data),
+                    ]);
+                })
+            );
+        }
+
+        console.log(`Seeded trends for ${timestamp_map.size} unique timestamps DONE`);
+    }
 }

src/trend/trend.controller.ts

@@ -35,4 +35,9 @@ export class TrendController {
     async deleteFakeTrends() {
         return await this.fake_trend_service.deleteFakeTrends();
     }
+
+    @Post('/seed-trends')
+    async seedTrends() {
+        return await this.fake_trend_service.seedTrend();
+    }
 }