feat: post analytics history (#2982)

Author: capJavert

.infra/crons.ts

@@ -126,4 +126,8 @@ export const crons: Cron[] = [
     name: 'post-analytics-clickhouse',
     schedule: '*/5 * * * *',
   },
+  {
+    name: 'post-analytics-history-day-clickhouse',
+    schedule: '3-59/5 * * * *',
+  },
 ];

__tests__/cron/postAnalyticsHistoryDayClickhouse.ts

@@ -0,0 +1,10 @@
+import { crons } from '../../src/cron/index';
+import { postAnalyticsHistoryDayClickhouseCron as cron } from '../../src/cron/postAnalyticsHistoryDayClickhouse';
+
+describe('postAnalyticsHistoryDayClickhouse cron', () => {
+  it('should be registered', () => {
+    const registeredWorker = crons.find((item) => item.name === cron.name);
+
+    expect(registeredWorker).toBeDefined();
+  });
+});

src/common/schema/postAnalytics.ts

@@ -1,3 +1,4 @@
+import { format } from 'date-fns';
 import { z } from 'zod';
 
 export const postAnalyticsClickhouseSchema = z
@@ -14,3 +15,12 @@ export const postAnalyticsClickhouseSchema = z
     sharesInternal: z.coerce.number().nonnegative(),
   })
   .strict();
+
+export const postAnalyticsHistoryClickhouseSchema = z
+  .object({
+    id: z.string(),
+    updatedAt: z.coerce.date(),
+    impressions: z.coerce.number().nonnegative(),
+    date: z.coerce.date().transform((date) => format(date, 'yyyy-MM-dd')),
+  })
+  .strict();

src/cron/index.ts

@@ -20,6 +20,7 @@ import { calculateTopReaders } from './calculateTopReaders';
 import cleanGiftedPlus from './cleanGiftedPlus';
 import { cleanStaleUserTransactions } from './cleanStaleUserTransactions';
 import { postAnalyticsClickhouseCron } from './postAnalyticsClickhouse';
+import { postAnalyticsHistoryDayClickhouseCron } from './postAnalyticsHistoryDayClickhouse';
 
 export const crons: Cron[] = [
   updateViews,
@@ -43,4 +44,5 @@ export const crons: Cron[] = [
   cleanGiftedPlus,
   cleanStaleUserTransactions,
   postAnalyticsClickhouseCron,
+  postAnalyticsHistoryDayClickhouseCron,
 ];

src/cron/postAnalyticsHistoryDayClickhouse.ts

@@ -0,0 +1,112 @@
+import { format, startOfToday } from 'date-fns';
+import { Cron } from './cron';
+import { getClickHouseClient } from '../common/clickhouse';
+import { postAnalyticsHistoryClickhouseSchema } from '../common/schema/postAnalytics';
+import { z } from 'zod';
+import { getRedisHash, setRedisHash } from '../redis';
+import { generateStorageKey, StorageTopic } from '../config';
+import { PostAnalyticsHistory } from '../entity/posts/PostAnalyticsHistory';
+
+type PostAnalyticsHistoryClickhouseCronConfig = Partial<{
+  lastRunAt: string;
+}>;
+
+export const postAnalyticsHistoryDayClickhouseCron: Cron = {
+  name: 'post-analytics-history-day-clickhouse',
+  handler: async (con, logger) => {
+    const redisStorageKey = generateStorageKey(
+      StorageTopic.Cron,
+      postAnalyticsHistoryDayClickhouseCron.name,
+      'config',
+    );
+
+    const cronConfig: Partial<PostAnalyticsHistoryClickhouseCronConfig> =
+      await getRedisHash(redisStorageKey);
+
+    const lastRunAt = cronConfig.lastRunAt
+      ? new Date(cronConfig.lastRunAt)
+      : startOfToday(); // for now use start of today if no last run time is set
+
+    if (Number.isNaN(lastRunAt.getTime())) {
+      throw new Error('Invalid last run time');
+    }
+
+    const clickhouseClient = getClickHouseClient();
+
+    const response = await clickhouseClient.query({
+      query: /* sql */ `
+        SELECT
+          post_id AS id,
+          date,
+          max(created_at) AS "updatedAt",
+          sum(impressions) AS impressions
+        FROM api.post_analytics_history
+        FINAL
+        WHERE date = {date: Date}
+        AND created_at > {lastRunAt: DateTime}
+        GROUP BY date, post_id
+      `,
+      format: 'JSONEachRow',
+      query_params: {
+        lastRunAt: format(lastRunAt, 'yyyy-MM-dd HH:mm:ss'),
+        date: format(new Date(), 'yyyy-MM-dd'),
+      },
+    });
+
+    const result = z
+      .array(postAnalyticsHistoryClickhouseSchema)
+      .safeParse(await response.json());
+
+    if (!result.success) {
+      logger.error(
+        {
+          schemaError: result.error.errors[0],
+        },
+        'Invalid post analytics data',
+      );
+
+      throw new Error('Invalid post analytics data');
+    }
+
+    const { data } = result;
+
+    const chunks: PostAnalyticsHistory[][] = [];
+    const chunkSize = 500;
+
+    data.forEach((item) => {
+      if (
+        chunks.length === 0 ||
+        chunks[chunks.length - 1].length === chunkSize
+      ) {
+        chunks.push([]);
+      }
+
+      chunks[chunks.length - 1].push(item as PostAnalyticsHistory);
+    });
+
+    const currentRunAt = new Date();
+
+    await con.transaction(async (entityManager) => {
+      for (const chunk of chunks) {
+        if (chunk.length === 0) {
+          continue;
+        }
+
+        await entityManager
+          .createQueryBuilder()
+          .insert()
+          .into(PostAnalyticsHistory)
+          .values(chunk)
+          .orUpdate(Object.keys(chunk[0]), ['id'])
+          .execute();
+      }
+    });
+
+    await setRedisHash<PostAnalyticsHistoryClickhouseCronConfig>(
+      redisStorageKey,
+      {
+        lastRunAt: currentRunAt.toISOString(),
+      },
+    );
+  },
+};

src/entity/posts/PostAnalytics.ts

@@ -2,6 +2,7 @@ import {
   Column,
   CreateDateColumn,
   Entity,
+  JoinColumn,
   OneToOne,
   PrimaryColumn,
   UpdateDateColumn,
@@ -61,9 +62,11 @@ export class PostAnalytics {
   @Column({ default: 0 })
   awards: number;
 
+  //  not added to migration because raw events data has some invalid post ids
   @OneToOne('Post', {
     lazy: true,
     onDelete: 'CASCADE',
   })
+  @JoinColumn({ name: 'id' })
   post: Promise<Post>;
 }

src/entity/posts/PostAnalyticsHistory.ts

@@ -0,0 +1,36 @@
+import {
+  Column,
+  CreateDateColumn,
+  Entity,
+  JoinColumn,
+  OneToOne,
+  PrimaryColumn,
+  UpdateDateColumn,
+} from 'typeorm';
+import type { Post } from './Post';
+
+@Entity()
+export class PostAnalyticsHistory {
+  @PrimaryColumn({ type: 'text' })
+  id: string;
+
+  @CreateDateColumn()
+  createdAt: Date;
+
+  @UpdateDateColumn()
+  updatedAt: Date;
+
+  @Column()
+  date: string;
+
+  @Column({ default: 0 })
+  impressions: number;
+
+  // not added to migration because raw events data has some invalid post ids
+  @OneToOne('Post', {
+    lazy: true,
+    onDelete: 'CASCADE',
+  })
+  @JoinColumn({ name: 'id' })
+  post: Promise<Post>;
+}

src/migration/1755090244241-PostAnalyticsHistory.ts

@@ -0,0 +1,15 @@
+import { MigrationInterface, QueryRunner } from 'typeorm';
+
+export class PostAnalyticsHistory1755090244241 implements MigrationInterface {
+  name = 'PostAnalyticsHistory1755090244241';
+
+  public async up(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(
+      `CREATE TABLE "post_analytics_history" ("id" text NOT NULL, "createdAt" TIMESTAMP NOT NULL DEFAULT now(), "updatedAt" TIMESTAMP NOT NULL DEFAULT now(), "date" character varying NOT NULL, "impressions" integer NOT NULL DEFAULT '0', CONSTRAINT "PK_6bb82232882b9a8bfa54a034c5f" PRIMARY KEY ("id"))`,
+    );
+  }
+
+  public async down(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(`DROP TABLE "post_analytics_history"`);
+  }
+}