feat: post analytics (#2971)

capJavert · web-flow · commit 4805eb48874d · 2025-08-12T15:34:06.000+02:00
diff --git a/.env b/.env
@@ -85,3 +85,7 @@ OPEN_EXCHANGE_RATES_APP_ID=topsecret
 PERSONALIZED_DIGEST_SECRET=topsecret
 
 GEOIP_PATH=
+
+CLICKHOUSE_URL=http://host.docker.internal:18123
+CLICKHOUSE_USER=default
+CLICKHOUSE_PASSWORD=changeme
diff --git a/.infra/Pulumi.prod.yaml b/.infra/Pulumi.prod.yaml
@@ -160,6 +160,12 @@ config:
       secure: AAABACPC/WLJuAHpqC3oE43YKjEzcFSXYSLfAx69vvRnC5xsOj2Md4tHo3RAAeVfrFaLRwugIQ7bPEZ4VBJnwJPem1NW0Nkm
     briefingFeed:
       secure: AAABAJWDkWAezGM15dk8xXwDdav6cmPwS/2x+tKvSwql/JUzFadVDpmlRRTSOGIh7WA++s+D7C6wdyy35Cg2FQ==
+    clickhouseUser:
+      secure: AAABAGoCA+Y4PvGtDV8m3osmBGGiI2vVoyzbdEDJ3n7W8a40Tk5rk/hbCQ==
+    clickhousePassword:
+      secure: AAABAAysCHiW8/lYE8NP3D9CE2UBE8YJdDZKf4B7aYVygfgyUAj9rk52klyHA26/jRtBW5pzCTTMHTatBm+ABdMTajgGy1GZ3yYkXMwzhhbJo30a+pq18vYGcx0F04MI
+    clickhouseUrl:
+      secure: AAABAOH/zIUW1KRNkCa3XH3Wxrs/GLDTjOeXsHqlDPoGowQC04O958oU8MXxtE0OeiNoHjTm/e6PrEr66nluAeEcwV0BhQLvVAAoP3L150tQNUU=
   api:k8s:
     host: subs.daily.dev
     namespace: daily
diff --git a/.infra/crons.ts b/.infra/crons.ts
@@ -122,4 +122,8 @@ export const crons: Cron[] = [
     name: 'clean-stale-user-transactions',
     schedule: '50 6 * * 0',
   },
+  {
+    name: 'post-analytics-clickhouse',
+    schedule: '*/5 * * * *',
+  },
 ];
diff --git a/__tests__/cron/postAnalyticsClickhouse.ts b/__tests__/cron/postAnalyticsClickhouse.ts
@@ -0,0 +1,10 @@
+import { crons } from '../../src/cron/index';
+import { postAnalyticsClickhouseCron as cron } from '../../src/cron/postAnalyticsClickhouse';
+
+describe('postAnalyticsClickhouse cron', () => {
+  it('should be registered', () => {
+    const registeredWorker = crons.find((item) => item.name === cron.name);
+
+    expect(registeredWorker).toBeDefined();
+  });
+});
diff --git a/package.json b/package.json
@@ -31,6 +31,7 @@
   "dependencies": {
     "@apple/app-store-server-library": "^1.5.0",
     "@bufbuild/protobuf": "^1.10.0",
+    "@clickhouse/client": "1.12.0",
     "@connectrpc/connect": "^1.6.1",
     "@connectrpc/connect-fastify": "^1.6.1",
     "@connectrpc/connect-node": "^1.6.1",
diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml
diff --git a/src/common/clickhouse.ts b/src/common/clickhouse.ts
@@ -0,0 +1,15 @@
+import { createClient } from '@clickhouse/client';
+
+let client: ReturnType<typeof createClient> | null = null;
+
+export const getClickHouseClient = () => {
+  if (!client) {
+    client = createClient({
+      url: process.env.CLICKHOUSE_URL,
+      username: process.env.CLICKHOUSE_USER,
+      password: process.env.CLICKHOUSE_PASSWORD,
+    });
+  }
+
+  return client;
+};
diff --git a/src/common/schema/postAnalytics.ts b/src/common/schema/postAnalytics.ts
@@ -0,0 +1,16 @@
+import { z } from 'zod';
+
+export const postAnalyticsClickhouseSchema = z
+  .object({
+    id: z.string(),
+    updatedAt: z.coerce.date(),
+    impressions: z.coerce.number().nonnegative(),
+    reach: z.coerce.number().nonnegative(),
+    bookmarks: z.coerce.number().nonnegative(),
+    profileViews: z.coerce.number().nonnegative(),
+    followers: z.coerce.number().nonnegative(),
+    squadJoins: z.coerce.number().nonnegative(),
+    sharesExternal: z.coerce.number().nonnegative(),
+    sharesInternal: z.coerce.number().nonnegative(),
+  })
+  .strict();
diff --git a/src/config.ts b/src/config.ts
@@ -34,6 +34,7 @@ export enum StorageTopic {
   Njord = 'njord',
   Paddle = 'paddle',
   RedisCounter = 'redis_counter',
+  Cron = 'cron',
 }
 
 export enum StorageKey {
diff --git a/src/cron/index.ts b/src/cron/index.ts
@@ -19,6 +19,7 @@ import { cleanZombieUserCompany } from './cleanZombieUserCompany';
 import { calculateTopReaders } from './calculateTopReaders';
 import cleanGiftedPlus from './cleanGiftedPlus';
 import { cleanStaleUserTransactions } from './cleanStaleUserTransactions';
+import { postAnalyticsClickhouseCron } from './postAnalyticsClickhouse';
 
 export const crons: Cron[] = [
   updateViews,
@@ -41,4 +42,5 @@ export const crons: Cron[] = [
   validateActiveUsers,
   cleanGiftedPlus,
   cleanStaleUserTransactions,
+  postAnalyticsClickhouseCron,
 ];
diff --git a/src/cron/postAnalyticsClickhouse.ts b/src/cron/postAnalyticsClickhouse.ts
@@ -0,0 +1,115 @@
+import { format, startOfToday } from 'date-fns';
+import { Cron } from './cron';
+import { getClickHouseClient } from '../common/clickhouse';
+import { postAnalyticsClickhouseSchema } from '../common/schema/postAnalytics';
+import { z } from 'zod';
+import { PostAnalytics } from '../entity/posts/PostAnalytics';
+import { getRedisHash, setRedisHash } from '../redis';
+import { generateStorageKey, StorageTopic } from '../config';
+
+type PostAnalyticsClickhouseCronConfig = Partial<{
+  lastRunAt: string;
+}>;
+
+export const postAnalyticsClickhouseCron: Cron = {
+  name: 'post-analytics-clickhouse',
+  handler: async (con, logger) => {
+    const redisStorageKey = generateStorageKey(
+      StorageTopic.Cron,
+      postAnalyticsClickhouseCron.name,
+      'config',
+    );
+
+    const cronConfig: Partial<PostAnalyticsClickhouseCronConfig> =
+      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: `
+        SELECT
+            post_id AS id,
+            max(created_at) AS "updatedAt",
+            sum(impressions) AS impressions,
+            uniqMerge(reach) AS reach,
+            uniqMerge(bookmarks) AS bookmarks,
+            uniqMerge(profile_views) AS "profileViews",
+            uniqMerge(followers) AS followers,
+            uniqMerge(squad_joins) AS "squadJoins",
+            sum(shares_external) AS "sharesExternal",
+            sum(shares_internal) AS "sharesInternal"
+        FROM api.post_analytics
+        FINAL
+        WHERE created_at >= now() - INTERVAL 45 DAY
+        GROUP BY post_id
+        HAVING "updatedAt" > {lastRunAt: DateTime}
+        ORDER BY "updatedAt" DESC;
+      `,
+      format: 'JSONEachRow',
+      query_params: {
+        lastRunAt: format(lastRunAt, 'yyyy-MM-dd HH:mm:ss'),
+      },
+    });
+
+    const result = z
+      .array(postAnalyticsClickhouseSchema)
+      .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: PostAnalytics[][] = [];
+    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 PostAnalytics);
+    });
+
+    const currentRunAt = new Date();
+
+    await con.transaction(async (entityManager) => {
+      for (const chunk of chunks) {
+        if (chunk.length === 0) {
+          continue;
+        }
+
+        await entityManager
+          .createQueryBuilder()
+          .insert()
+          .into(PostAnalytics)
+          .values(chunk)
+          .orUpdate(Object.keys(chunk[0]), ['id'])
+          .execute();
+      }
+    });
+
+    await setRedisHash<PostAnalyticsClickhouseCronConfig>(redisStorageKey, {
+      lastRunAt: currentRunAt.toISOString(),
+    });
+  },
+};
diff --git a/src/entity/posts/PostAnalytics.ts b/src/entity/posts/PostAnalytics.ts
@@ -0,0 +1,69 @@
+import {
+  Column,
+  CreateDateColumn,
+  Entity,
+  OneToOne,
+  PrimaryColumn,
+  UpdateDateColumn,
+} from 'typeorm';
+import type { Post } from './Post';
+
+@Entity()
+export class PostAnalytics {
+  @PrimaryColumn({ type: 'text' })
+  id: string;
+
+  @CreateDateColumn()
+  createdAt: Date;
+
+  @UpdateDateColumn()
+  updatedAt: Date;
+
+  @Column({ default: 0 })
+  impressions: number;
+
+  @Column({ default: 0 })
+  reach: number;
+
+  @Column({ default: 0 })
+  bookmarks: number;
+
+  @Column({ default: 0 })
+  profileViews: number;
+
+  @Column({ default: 0 })
+  followers: number;
+
+  @Column({ default: 0 })
+  squadJoins: number;
+
+  @Column({ default: 0 })
+  sharesExternal: number;
+
+  @Column({ default: 0 })
+  sharesInternal: number;
+
+  @Column({ default: 0 })
+  reputation: number;
+
+  @Column({ default: 0 })
+  coresEarned: number;
+
+  @Column({ default: 0 })
+  upvotes: number;
+
+  @Column({ default: 0 })
+  downvotes: number;
+
+  @Column({ default: 0 })
+  comments: number;
+
+  @Column({ default: 0 })
+  awards: number;
+
+  @OneToOne('Post', {
+    lazy: true,
+    onDelete: 'CASCADE',
+  })
+  post: Promise<Post>;
+}
diff --git a/src/migration/1754664628189-PostAnalytics.ts b/src/migration/1754664628189-PostAnalytics.ts
@@ -0,0 +1,15 @@
+import { MigrationInterface, QueryRunner } from 'typeorm';
+
+export class PostAnalytics1754664628189 implements MigrationInterface {
+  name = 'PostAnalytics1754664628189';
+
+  public async up(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(
+      `CREATE TABLE "post_analytics" ("id" text NOT NULL, "createdAt" TIMESTAMP NOT NULL DEFAULT now(), "updatedAt" TIMESTAMP NOT NULL DEFAULT now(), "impressions" integer NOT NULL DEFAULT '0', "reach" integer NOT NULL DEFAULT '0', "bookmarks" integer NOT NULL DEFAULT '0', "profileViews" integer NOT NULL DEFAULT '0', "followers" integer NOT NULL DEFAULT '0', "squadJoins" integer NOT NULL DEFAULT '0', "sharesExternal" integer NOT NULL DEFAULT '0', "sharesInternal" integer NOT NULL DEFAULT '0', "reputation" integer NOT NULL DEFAULT '0', "coresEarned" integer NOT NULL DEFAULT '0', "upvotes" integer NOT NULL DEFAULT '0', "downvotes" integer NOT NULL DEFAULT '0', "comments" integer NOT NULL DEFAULT '0', "awards" integer NOT NULL DEFAULT '0', CONSTRAINT "PK_421771f55c623cd4f6103828196" PRIMARY KEY ("id"))`,
+    );
+  }
+
+  public async down(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(`DROP TABLE "post_analytics"`);
+  }
+}
diff --git a/src/redis.ts b/src/redis.ts
@@ -112,3 +112,6 @@ export const getRedisHashField = (key: string, field: string) =>
 
 export const checkRedisObjectExists = (key: string) =>
   ioRedisPool.execute((client) => client.exists(key));
+
+export const setRedisHash = <T extends object>(key: string, value: T) =>
+  ioRedisPool.execute((client) => client.hset(key, value));

Original file line number	Diff line number	Diff line change
`@@ -34,6 +34,7 @@ export enum StorageTopic {`
`34`	`34`	`Njord = 'njord',`
`35`	`35`	`Paddle = 'paddle',`
`36`	`36`	`RedisCounter = 'redis_counter',`
	`37`	`+ Cron = 'cron',`
`37`	`38`	`}`
`38`	`39`
`39`	`40`	`export enum StorageKey {`