diff --git a/.infra/application.properties b/.infra/application.properties index 7a4868b321..248163d8c3 100644 --- a/.infra/application.properties +++ b/.infra/application.properties @@ -1,6 +1,7 @@ debezium.source.connector.class=io.debezium.connector.postgresql.PostgresConnector debezium.source.slot.name=%slot_name% -debezium.source.publication.name=api_debezium +debezium.source.publication.name=api_debezium_filtered +debezium.source.publication.autocreate.mode=disabled debezium.source.offset.storage.file.filename=data/offsets.dat debezium.source.offset.flush.interval.ms=0 debezium.source.database.hostname=%hostname% @@ -9,14 +10,18 @@ debezium.source.database.user=%database_user% debezium.source.database.password=%database_pass% debezium.source.database.dbname=%database_dbname% debezium.source.database.server.name=api -debezium.source.table.include.list=public.comment,public.user_comment,public.comment_mention,public.source_request,public.post,public.user,public.post_report,public.source_feed,public.settings,public.reputation_event,public.submission,public.user_state,public.notification_v2,public.source_member,public.feature,public.source,public.post_mention,public.content_image,public.comment_report,public.user_post,public.banner,public.post_relation,public.marketing_cta,public.squad_public_request,public.user_streak,public.bookmark,public.bookmark_list,public.user_company,public.source_report,public.user_top_reader,public.source_post_moderation,public.user_report,public.user_transaction,public.content_preference,public.campaign,public.opportunity_match,public.opportunity,public.organization,public.user_candidate_preference,public.user_experience,public.feedback,public.hot_take,public.user_stack,public.quest,public.quest_reward,public.quest_rotation,public.user_quest,public.user_quest_profile,public.highlights_canonical,public.feed,public.opportunity_user,public.user_marketing_cta,public.contribution_submission -debezium.source.column.exclude.list=public.post.tsv,public.post.placeholder,public.source.flags,public.user_top_reader.image +debezium.source.table.include.list=public.comment,public.user_comment,public.comment_mention,public.source_request,public.post,public.user,public.post_report,public.source_feed,public.settings,public.reputation_event,public.submission,public.user_state,public.notification_v2,public.source_member,public.feature,public.source,public.post_mention,public.content_image,public.comment_report,public.user_post,public.banner,public.post_relation,public.marketing_cta,public.squad_public_request,public.user_streak,public.bookmark,public.bookmark_list,public.user_company,public.source_report,public.user_top_reader,public.source_post_moderation,public.user_report,public.user_transaction,public.content_preference,public.campaign,public.opportunity_match,public.opportunity,public.organization,public.user_candidate_preference,public.user_experience,public.feedback,public.hot_take,public.user_stack,public.quest,public.quest_reward,public.quest_rotation,public.user_quest,public.user_quest_profile,public.highlights_canonical,public.feed,public.opportunity_user,public.user_marketing_cta,public.contribution_submission,public.heartbeat +debezium.source.column.exclude.list=public.post.tsv,public.post.placeholder,public.post.slug,public.post.views,public.source.flags,public.user_top_reader.image debezium.source.skip.messages.without.change=true debezium.source.plugin.name=pgoutput debezium.source.heartbeat.interval.ms=60000 +debezium.source.heartbeat.action.query=INSERT INTO heartbeat (id, ts) VALUES (1, NOW()) ON CONFLICT(id) DO UPDATE SET ts=EXCLUDED.ts; debezium.source.topic.prefix=api debezium.source.tombstones.on.delete=false -debezium.transforms=Reroute,Notifications,ReadOperationFilter,DigestPostFilter,PostsFilter,UserIncFilter,UserMarketingCtaFilter +debezium.transforms=HeartbeatFilter,Reroute,Notifications,ReadOperationFilter,DigestPostFilter,PostsFilter,UserIncFilter,UserMarketingCtaFilter +debezium.transforms.HeartbeatFilter.type=io.debezium.transforms.Filter +debezium.transforms.HeartbeatFilter.language=jsr223.groovy +debezium.transforms.HeartbeatFilter.condition=!(valueSchema.field('op') && value.source.table == 'heartbeat') debezium.transforms.Reroute.type=io.debezium.transforms.ByLogicalTableRouter debezium.transforms.Reroute.topic.regex=^((?!\.notification_v2).)*$ debezium.transforms.Reroute.topic.replacement=%topic% diff --git a/.infra/clickhouse-sync.yml b/.infra/clickhouse-sync.yml index 35c08224ab..5aa4a941fa 100644 --- a/.infra/clickhouse-sync.yml +++ b/.infra/clickhouse-sync.yml @@ -51,7 +51,8 @@ database.allowPublicKeyRetrieval: "true" snapshot.mode: "no_data" slot.name: "clickhouse_sync" -publication.name: "clickhouse_sync" +publication.name: "clickhouse_sync_filtered" +publication.autocreate.mode: "disabled" # offset.flush.interval.ms: The number of milliseconds to wait before flushing recent offsets to Kafka. This ensures that offsets are committed within the specified time interval. offset.flush.timeout.ms: 10000 diff --git a/src/entity/DebeziumHeartbeat.ts b/src/entity/DebeziumHeartbeat.ts new file mode 100644 index 0000000000..8b91fd04c2 --- /dev/null +++ b/src/entity/DebeziumHeartbeat.ts @@ -0,0 +1,13 @@ +import { Column, Entity, PrimaryColumn } from 'typeorm'; + +@Entity({ name: 'heartbeat' }) +export class DebeziumHeartbeat { + @PrimaryColumn({ + type: 'integer', + primaryKeyConstraintName: 'PK_heartbeat', + }) + id: number; + + @Column({ type: 'timestamptz' }) + ts: Date; +} diff --git a/src/entity/index.ts b/src/entity/index.ts index 35ff60a63a..d8f7a33e45 100644 --- a/src/entity/index.ts +++ b/src/entity/index.ts @@ -4,6 +4,7 @@ export * from './Alerts'; export * from './Banner'; export * from './Bookmark'; export * from './BookmarkList'; +export * from './DebeziumHeartbeat'; export * from './DevCard'; export * from './Feed'; export * from './FeedAdvancedSettings'; diff --git a/src/migration/1784279440502-CreateDebeziumHeartbeat.ts b/src/migration/1784279440502-CreateDebeziumHeartbeat.ts new file mode 100644 index 0000000000..859d9f671b --- /dev/null +++ b/src/migration/1784279440502-CreateDebeziumHeartbeat.ts @@ -0,0 +1,21 @@ +import type { MigrationInterface, QueryRunner } from 'typeorm'; + +export class CreateDebeziumHeartbeat1784279440502 implements MigrationInterface { + name = 'CreateDebeziumHeartbeat1784279440502'; + + public async up(queryRunner: QueryRunner): Promise { + await queryRunner.query(/* sql */ ` + CREATE TABLE "heartbeat" ( + "id" integer NOT NULL, + "ts" TIMESTAMP WITH TIME ZONE NOT NULL, + CONSTRAINT "PK_heartbeat" PRIMARY KEY ("id") + ) + `); + } + + public async down(queryRunner: QueryRunner): Promise { + await queryRunner.query(/* sql */ ` + DROP TABLE "heartbeat" + `); + } +} diff --git a/src/migration/1784280115875-CreateFilteredPublications.ts b/src/migration/1784280115875-CreateFilteredPublications.ts new file mode 100644 index 0000000000..20e6498a74 --- /dev/null +++ b/src/migration/1784280115875-CreateFilteredPublications.ts @@ -0,0 +1,156 @@ +import type { MigrationInterface, QueryRunner } from 'typeorm'; + +export class CreateFilteredPublications1784280115875 implements MigrationInterface { + name = 'CreateFilteredPublications1784280115875'; + + public async up(queryRunner: QueryRunner): Promise { + await queryRunner.query(/* sql */ ` + CREATE PUBLICATION "api_debezium_filtered" + WITH (publish_generated_columns = stored) + `); + + await queryRunner.query(/* sql */ ` + DO $$ + DECLARE + publication_tables TEXT; + BEGIN + SELECT STRING_AGG( + FORMAT('%I.%I', namespace.nspname, relation.relname), + ', ' ORDER BY desired.ordinality + ) + INTO publication_tables + FROM UNNEST(ARRAY[ + 'comment', + 'user_comment', + 'comment_mention', + 'source_request', + 'post', + 'user', + 'post_report', + 'source_feed', + 'settings', + 'reputation_event', + 'submission', + 'user_state', + 'notification_v2', + 'source_member', + 'feature', + 'source', + 'post_mention', + 'content_image', + 'comment_report', + 'user_post', + 'banner', + 'post_relation', + 'marketing_cta', + 'squad_public_request', + 'user_streak', + 'bookmark', + 'bookmark_list', + 'user_company', + 'source_report', + 'user_top_reader', + 'source_post_moderation', + 'user_report', + 'user_transaction', + 'content_preference', + 'campaign', + 'opportunity_match', + 'opportunity', + 'organization', + 'user_candidate_preference', + 'user_experience', + 'feedback', + 'hot_take', + 'user_stack', + 'quest', + 'quest_reward', + 'quest_rotation', + 'user_quest', + 'user_quest_profile', + 'highlights_canonical', + 'feed', + 'opportunity_user', + 'user_marketing_cta', + 'contribution_submission', + 'heartbeat' + ]) WITH ORDINALITY AS desired(name, ordinality) + INNER JOIN pg_catalog.pg_class AS relation + ON relation.relname = desired.name + AND relation.relkind IN ('r', 'p') + INNER JOIN pg_catalog.pg_namespace AS namespace + ON namespace.oid = relation.relnamespace + AND namespace.nspname = 'public'; + + IF publication_tables IS NOT NULL THEN + EXECUTE FORMAT( + 'ALTER PUBLICATION %I ADD TABLE %s', + 'api_debezium_filtered', + publication_tables + ); + END IF; + END + $$ + `); + + await queryRunner.query(/* sql */ ` + CREATE PUBLICATION "clickhouse_sync_filtered" + WITH (publish_generated_columns = stored) + `); + + await queryRunner.query(/* sql */ ` + DO $$ + DECLARE + publication_tables TEXT; + BEGIN + SELECT STRING_AGG( + FORMAT('%I.%I', namespace.nspname, relation.relname), + ', ' ORDER BY desired.ordinality + ) + INTO publication_tables + FROM UNNEST(ARRAY[ + 'post', + 'source', + 'keyword', + 'niche', + 'keyword_niche', + 'user', + 'content_preference', + 'post_keyword', + 'post_niche', + 'comment', + 'campaign', + 'post_relation', + 'user_personalized_digest', + 'user_company', + 'highlights_canonical' + ]) WITH ORDINALITY AS desired(name, ordinality) + INNER JOIN pg_catalog.pg_class AS relation + ON relation.relname = desired.name + AND relation.relkind IN ('r', 'p') + INNER JOIN pg_catalog.pg_namespace AS namespace + ON namespace.oid = relation.relnamespace + AND namespace.nspname = 'public'; + + IF publication_tables IS NOT NULL THEN + EXECUTE FORMAT( + 'ALTER PUBLICATION %I ADD TABLE %s', + 'clickhouse_sync_filtered', + publication_tables + ); + END IF; + END + $$ + `); + } + + public async down(queryRunner: QueryRunner): Promise { + await queryRunner.query(/* sql */ ` + DROP PUBLICATION IF EXISTS "clickhouse_sync_filtered" + `); + + await queryRunner.query(/* sql */ ` + DROP PUBLICATION IF EXISTS "api_debezium_filtered" + `); + } +}