feat: add inc cdc filtering (#3452)

Author: capJavert

.infra/application.properties

@@ -15,7 +15,7 @@ debezium.source.plugin.name=pgoutput
 debezium.source.heartbeat.interval.ms=60000
 debezium.source.topic.prefix=api
 debezium.source.tombstones.on.delete=false
-debezium.transforms=Reroute,Notifications,ReadOperationFilter,PostsFilter
+debezium.transforms=Reroute,Notifications,ReadOperationFilter,PostsFilter,UserIncFilter
 debezium.transforms.Reroute.type=io.debezium.transforms.ByLogicalTableRouter
 debezium.transforms.Reroute.topic.regex=^((?!\.notification_v2).)*$
 debezium.transforms.Reroute.topic.replacement=%topic%
@@ -28,4 +28,7 @@ debezium.transforms.ReadOperationFilter.condition=!(valueSchema.field('op') && v
 debezium.transforms.PostsFilter.type=io.debezium.transforms.Filter
 debezium.transforms.PostsFilter.language=jsr223.groovy
 debezium.transforms.PostsFilter.condition=!(valueSchema.field('op') && value.op == 'u' && value.source.table == 'post' && value.before.views != value.after.views) && !(valueSchema.field('op') && value.op == 'u' && value.source.table == 'user' && value.before.cioRegistered != value.after.cioRegistered)
+debezium.transforms.UserIncFilter.type=io.debezium.transforms.Filter
+debezium.transforms.UserIncFilter.language=jsr223.groovy
+debezium.transforms.UserIncFilter.condition=!(valueSchema.field('op') && value.op == 'u' && value.source.table == 'user' && value.before.inc != value.after.inc)
 debezium.sink.type=pubsub

src/entity/user/User.ts

@@ -365,4 +365,9 @@ export class User {
   })
   @Index('IDX_user_locationId')
   location: Promise<DatasetLocation>;
+
+  // used for diffing changes, eg. filters for cdc during updates
+  // due to user table having a lot of depenencies
+  @Column({ type: 'int', default: 0 })
+  inc: number;
 }

src/migration/1768572789493-UserInc.ts

@@ -0,0 +1,15 @@
+import { MigrationInterface, QueryRunner } from 'typeorm';
+
+export class UserInc1768572789493 implements MigrationInterface {
+  name = 'UserInc1768572789493';
+
+  public async up(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(
+      `ALTER TABLE "user" ADD "inc" integer NOT NULL DEFAULT '0'`,
+    );
+  }
+
+  public async down(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(`ALTER TABLE "user" DROP COLUMN "inc"`);
+  }
+}