import { getClickhouseAdminClient } from "@/lib/clickhouse"; import { CLICKHOUSE_COLUMN_NORMALIZERS } from "@/lib/external-db-sync"; import type { Tenancy } from "@/lib/tenancies"; import { getPrismaClientForTenancy } from "@/prisma-client"; import { DEFAULT_DB_SYNC_MAPPINGS } from "@hexclave/shared/dist/config/db-sync-mappings"; import { HexclaveAssertionError } from "@hexclave/shared/dist/utils/errors"; import { deindent } from "@hexclave/shared/dist/utils/strings"; import type { RecurseFunction } from "./recurse"; // Sort key columns for each mapping (after project_id, branch_id), matching ClickHouse ORDER BY const SORT_KEYS = { users: ["id"], contact_channels: ["id"], teams: ["id"], team_member_profiles: ["team_id", "user_id"], team_permissions: ["team_id", "user_id", "id"], team_invitations: ["id"], email_outboxes: ["id"], project_permissions: ["user_id", "permission_id"], notification_preferences: ["user_id", "notification_category_id"], refresh_tokens: ["id"], connected_accounts: ["user_id", "provider", "provider_account_id"], } satisfies Record; const SYNC_COLUMNS_TO_STRIP = ["sync_sequence_id", "sync_is_deleted", "sync_created_at", "tenancyId"]; function compareRows(a: Record, b: Record, sortKeys: string[]): number { for (const key of sortKeys) { const aVal = String(a[key] ?? ""); const bVal = String(b[key] ?? ""); if (aVal < bVal) return -1; if (aVal > bVal) return 1; } return 0; } function normalizePostgresValue(value: unknown, columnType: string | undefined): unknown { if (value === null || value === undefined) return null; if (columnType === "json") { // Postgres returns parsed JS values for jsonb columns; always stringify for consistent comparison return JSON.stringify(value); } if (columnType === "boolean") { if (typeof value === "boolean") return value ? 1 : 0; return value; } if (columnType === "nullable_boolean") { if (typeof value === "boolean") return value ? 1 : 0; return value; } if (columnType === "bigint") { return Number(value); } // For dates, normalize to ms epoch if (value instanceof Date) { return value.getTime(); } return value; } function normalizeClickhouseValue(value: unknown, columnType: string | undefined): unknown { if (value === null || value === undefined) return null; if (columnType === "json") { // ClickHouse stores null JSON as the literal string "null" if (value === "null") return null; return typeof value === "string" ? value : JSON.stringify(value); } // For dates (ClickHouse returns as string like "2024-01-01 00:00:00.000" in UTC) if (typeof value === "string" && /^\d{4}-\d{2}-\d{2}[ T]\d{2}:\d{2}:\d{2}/.test(value)) { // Append 'Z' to force UTC interpretation if no timezone indicator present const dateStr = value.includes("Z") || value.includes("+") ? value : value.replace(" ", "T") + "Z"; return new Date(dateStr).getTime(); } return value; } function normalizeRow( row: Record, normalizers: Record, side: "postgres" | "clickhouse", ): Record { const result: Record = {}; for (const [key, value] of Object.entries(row)) { const columnType = normalizers[key]; result[key] = side === "postgres" ? normalizePostgresValue(value, columnType) : normalizeClickhouseValue(value, columnType); } return result; } // Strip null values and empty objects from nested structures. // ClickHouse's native JSON type omits these, so we need to normalize before comparing. function stripNullsAndEmpties(value: unknown): unknown { if (value === null || value === undefined) return undefined; if (typeof value !== "object") return value; if (Array.isArray(value)) return value.map(stripNullsAndEmpties); const obj = value as Record; const result: Record = {}; for (const [k, v] of Object.entries(obj)) { const stripped = stripNullsAndEmpties(v); if (stripped === undefined) continue; if (typeof stripped === "object" && stripped !== null && !Array.isArray(stripped) && Object.keys(stripped).length === 0) continue; result[k] = stripped; } return result; } function deepEqual(a: unknown, b: unknown): boolean { if (a === b) return true; if (a === null || b === null) return a === b; if (typeof a !== typeof b) return false; if (typeof a === "object" && typeof b === "object") { // Normalize both sides to handle ClickHouse JSON stripping nulls/empties const aNorm = stripNullsAndEmpties(a) as Record; const bNorm = stripNullsAndEmpties(b) as Record; const aKeys = Object.keys(aNorm); const bKeys = Object.keys(bNorm); if (aKeys.length !== bKeys.length) return false; return aKeys.every(key => deepEqual(aNorm[key], bNorm[key])); } return false; } function findDifferences( pgRow: Record, chRow: Record, ): string[] { const diffs: string[] = []; const allKeys = new Set([...Object.keys(pgRow), ...Object.keys(chRow)]); for (const key of allKeys) { if (!deepEqual(pgRow[key], chRow[key])) { diffs.push(`${key}: pg=${JSON.stringify(pgRow[key])} ch=${JSON.stringify(chRow[key])}`); } } return diffs; } export async function verifyClickhouseSync(options: { tenancy: Tenancy, projectId: string, branchId: string, recurse: RecurseFunction, }) { const { tenancy, projectId, branchId, recurse } = options; const clickhouseClient = getClickhouseAdminClient(); const prisma = await getPrismaClientForTenancy(tenancy); for (const [mappingName, mapping] of Object.entries(DEFAULT_DB_SYNC_MAPPINGS)) { await recurse(`[${mappingName}]`, async () => { const fetchQuery = mapping.internalDbFetchQueries.clickhouse; if (!fetchQuery) return; if (!(mappingName in SORT_KEYS)) { throw new HexclaveAssertionError(`No sort keys defined for mapping ${mappingName}`); } const sortKeys = SORT_KEYS[mappingName as keyof typeof SORT_KEYS]; const normalizers = CLICKHOUSE_COLUMN_NORMALIZERS[mappingName] ?? {}; // Fetch all non-deleted rows from Postgres using the same query the sync uses const pgRows: Record[] = []; let lastSequenceId = -1; const BATCH_LIMIT = 1000; // eslint-disable-next-line no-constant-condition while (true) { const batch = await prisma.$queryRawUnsafe[]>( fetchQuery, tenancy.id, lastSequenceId, ); if (batch.length === 0) break; for (const row of batch) { const syncIsDeleted = row.sync_is_deleted; if (syncIsDeleted === true || syncIsDeleted === "true") continue; const stripped: Record = {}; for (const [key, value] of Object.entries(row)) { if (!SYNC_COLUMNS_TO_STRIP.includes(key)) { stripped[key] = value; } } pgRows.push(stripped); } // Find max sequence ID in batch for cursor let maxSeq = lastSequenceId; for (const row of batch) { const seq = Number(row.sync_sequence_id); if (Number.isFinite(seq) && seq > maxSeq) { maxSeq = seq; } } lastSequenceId = maxSeq; if (batch.length < BATCH_LIMIT) break; } // Fetch all rows from ClickHouse view (already FINAL + sync_is_deleted = 0) const chResult = await clickhouseClient.query({ query: `SELECT * FROM default.${mapping.targetTable} WHERE project_id = {project_id:String} AND branch_id = {branch_id:String}`, query_params: { project_id: projectId, branch_id: branchId }, format: "JSONEachRow", }); const chRows = await chResult.json>(); // Compare row counts if (pgRows.length !== chRows.length) { throw new HexclaveAssertionError(deindent` ClickHouse sync row count mismatch for ${mappingName}. Postgres: ${pgRows.length} rows, ClickHouse: ${chRows.length} rows. `); } if (pgRows.length === 0) return; // Sort both by primary key columns const fullSortKeys = ["project_id", "branch_id", ...sortKeys]; pgRows.sort((a, b) => compareRows(a, b, fullSortKeys)); chRows.sort((a, b) => compareRows(a, b, fullSortKeys)); // Compare row by row for (let i = 0; i < pgRows.length; i++) { const normalizedPg = normalizeRow(pgRows[i], normalizers, "postgres"); const normalizedCh = normalizeRow(chRows[i], normalizers, "clickhouse"); if (!deepEqual(normalizedPg, normalizedCh)) { const diffs = findDifferences(normalizedPg, normalizedCh); const keyValues = fullSortKeys.map(k => `${k}=${pgRows[i][k]}`).join(", "); throw new HexclaveAssertionError(deindent` ClickHouse sync data mismatch for ${mappingName} at row ${keyValues}. Differences: ${diffs.join("; ")} `); } } }); } }