read_replicate/schema_catalog.ts
Rama main · réplica de lectura del repositorio original.
import { sortJson } from './schema_json.ts'
export const REPLICA_TABLES = [
'orgs',
'stripe_info',
'org_users',
'apps',
'app_versions',
'channels',
'channel_devices',
'delta_files',
'delta_sessions',
'notifications',
'onboarding_demo_data',
'version_files',
] as const
export const REPLICA_TYPES = [
'channel_update_package',
'disable_update',
'manifest_entry',
'stripe_status',
] as const
const REPLICA_SEQUENCES = [
'app_versions_id_seq',
'channel_devices_id_seq',
'channel_id_seq',
'org_users_id_seq',
'stripe_info_id_seq',
] as const
export const REPLICA_FUNCTIONS = ['one_month_ahead'] as const
export const REPLICA_EXCLUDED_INDEXES = [
// replicate_prepare.sh intentionally omits this source-only customer lookup index from the replica DDL.
'idx_stripe_info_customer_id',
// Primary-only reclaim helper for null_migrated_app_version_manifests; not needed on subscribers
// and Cloud SQL server-side import cannot apply CREATE INDEX before primary migrations.
'app_versions_manifest_present_idx',
// Delta storage writes and GC lookups run on the primary; plugin reads join by primary key only.
'delta_files_app_id_idx',
'delta_files_entry_key_idx',
'delta_files_path_key_idx',
'delta_sessions_app_id_session_idx',
'version_files_file_ids_gin_idx',
] as const
export function replicaConfigPattern(values: readonly string[]): string {
return values.map(escapeRegex).join('|')
}
function escapeRegex(value: string): string {
return value.replace(/[.*+?^${}()|[\]\\]/g, String.raw`\$&`)
}
export interface Queryable {
query: (
queryText: string,
values?: unknown[],
) => Promise<{ rows: Record<string, any>[] }>
}
// The release checker reads this catalog before and after schema DDL. The explicit
// CURRENT_TIMESTAMP predicate makes Hyperdrive treat the verification read as
// non-cacheable, so it observes DDL from the same release run.
export const READ_REPLICA_SCHEMA_CATALOG_SQL = `
WITH RECURSIVE replica_tables(table_name) AS (
SELECT unnest($1::text[])
),
replica_sequences(sequence_name) AS (
SELECT unnest($3::text[])
),
replica_functions(function_name) AS (
SELECT unnest($4::text[])
),
replica_excluded_indexes(index_name) AS (
SELECT unnest($5::text[])
),
tables AS (
SELECT
c.oid AS table_oid,
c.relname AS table_name,
c.relkind,
COALESCE((
SELECT jsonb_agg(option ORDER BY option)
FROM unnest(c.reloptions) AS option
), '[]'::jsonb) AS reloptions
FROM pg_class c
JOIN pg_namespace n ON n.oid = c.relnamespace
JOIN replica_tables rt ON rt.table_name = c.relname
WHERE n.nspname = 'public'
AND c.relkind IN ('r', 'p')
),
selected_column_types(type_oid) AS (
SELECT DISTINCT a.atttypid
FROM tables t
JOIN pg_attribute a ON a.attrelid = t.table_oid
WHERE a.attnum > 0
AND NOT a.attisdropped
),
referenced_types(type_oid) AS (
SELECT type_oid
FROM selected_column_types
UNION
SELECT CASE
WHEN typ.typbasetype <> 0 THEN typ.typbasetype
ELSE typ.typelem
END
FROM referenced_types referenced
JOIN pg_type typ ON typ.oid = referenced.type_oid
WHERE typ.typbasetype <> 0
OR typ.typelem <> 0
),
selected_type_names(type_name) AS (
SELECT unnest($2::text[])
UNION
SELECT DISTINCT typ.typname
FROM referenced_types referenced
JOIN pg_type typ ON typ.oid = referenced.type_oid
JOIN pg_namespace type_namespace ON type_namespace.oid = typ.typnamespace
WHERE type_namespace.nspname = 'public'
AND typ.typtype IN ('e', 'c')
),
columns AS (
SELECT
t.table_name,
row_number() OVER (PARTITION BY t.table_name ORDER BY a.attnum)::int AS position,
a.attname AS column_name,
pg_catalog.format_type(a.atttypid, a.atttypmod) AS data_type,
a.attnotnull AS not_null,
a.attidentity AS identity_kind,
a.attgenerated AS generated_kind,
pg_get_expr(d.adbin, d.adrelid) AS default_expr
FROM tables t
JOIN pg_attribute a ON a.attrelid = t.table_oid
LEFT JOIN pg_attrdef d ON d.adrelid = a.attrelid AND d.adnum = a.attnum
WHERE a.attnum > 0
AND NOT a.attisdropped
),
constraints AS (
SELECT
t.table_name,
con.conname AS constraint_name,
con.contype AS constraint_type,
con.convalidated AS is_valid,
pg_get_constraintdef(con.oid, true) AS definition
FROM tables t
JOIN pg_constraint con ON con.conrelid = t.table_oid
WHERE con.contype IN ('p', 'u', 'c')
),
indexes AS (
SELECT
t.table_name,
idx.relname AS index_name,
pg_get_indexdef(idx.oid) AS definition,
ix.indisvalid AS is_valid,
(constraint_owner.oid IS NOT NULL) AS constraint_owned
FROM tables t
JOIN pg_index ix ON ix.indrelid = t.table_oid
JOIN pg_class idx ON idx.oid = ix.indexrelid
LEFT JOIN pg_constraint constraint_owner
ON constraint_owner.conindid = ix.indexrelid
AND constraint_owner.contype IN ('p', 'u', 'x')
WHERE NOT EXISTS (
SELECT 1
FROM replica_excluded_indexes rei
WHERE rei.index_name = idx.relname
)
),
types AS (
SELECT
typ.typname AS type_name,
typ.typtype AS type_kind,
CASE
WHEN typ.typtype = 'e' THEN (
SELECT jsonb_agg(e.enumlabel ORDER BY e.enumsortorder)
FROM pg_enum e
WHERE e.enumtypid = typ.oid
)
WHEN typ.typtype = 'c' THEN (
SELECT jsonb_agg(
jsonb_build_object(
'position', a.attnum,
'name', a.attname,
'type', pg_catalog.format_type(a.atttypid, a.atttypmod)
)
ORDER BY a.attnum
)
FROM pg_attribute a
WHERE a.attrelid = typ.typrelid
AND a.attnum > 0
AND NOT a.attisdropped
)
ELSE NULL
END AS definition
FROM pg_type typ
JOIN pg_namespace n ON n.oid = typ.typnamespace
JOIN selected_type_names rt ON rt.type_name = typ.typname
WHERE n.nspname = 'public'
),
selected_sequence_oids(sequence_oid) AS (
SELECT seq.oid
FROM pg_class seq
JOIN pg_namespace n ON n.oid = seq.relnamespace
JOIN replica_sequences rs ON rs.sequence_name = seq.relname
WHERE n.nspname = 'public'
AND seq.relkind = 'S'
UNION
SELECT dep.objid
FROM pg_depend dep
JOIN tables t ON t.table_oid = dep.refobjid
JOIN pg_class seq ON seq.oid = dep.objid
JOIN pg_namespace n ON n.oid = seq.relnamespace
WHERE dep.deptype IN ('a', 'i')
AND n.nspname = 'public'
AND seq.relkind = 'S'
),
sequences AS (
SELECT
seq.relname AS sequence_name,
pg_catalog.format_type(s.seqtypid, NULL) AS data_type,
s.seqstart::text AS start_value,
s.seqincrement::text AS increment_by,
s.seqmin::text AS min_value,
s.seqmax::text AS max_value,
s.seqcache::text AS cache_size,
s.seqcycle AS cycle,
owned_table.relname AS owned_table,
owned_attr.attname AS owned_column
FROM pg_class seq
JOIN pg_namespace n ON n.oid = seq.relnamespace
JOIN pg_sequence s ON s.seqrelid = seq.oid
JOIN selected_sequence_oids selected_sequence ON selected_sequence.sequence_oid = seq.oid
LEFT JOIN pg_depend dep ON dep.objid = seq.oid AND dep.deptype IN ('a', 'i')
LEFT JOIN pg_class owned_table ON owned_table.oid = dep.refobjid
LEFT JOIN pg_attribute owned_attr ON owned_attr.attrelid = dep.refobjid AND owned_attr.attnum = dep.refobjsubid
WHERE n.nspname = 'public'
AND seq.relkind = 'S'
),
functions AS (
SELECT
p.proname AS function_name,
pg_get_function_identity_arguments(p.oid) AS arguments,
pg_get_functiondef(p.oid) AS definition
FROM pg_proc p
JOIN pg_namespace n ON n.oid = p.pronamespace
JOIN replica_functions rf ON rf.function_name = p.proname
WHERE n.nspname = 'public'
)
SELECT jsonb_build_object(
'version', 1,
'tables', COALESCE((
SELECT jsonb_agg(jsonb_build_object(
'name', table_name,
'kind', relkind,
'reloptions', reloptions
) ORDER BY table_name)
FROM tables
), '[]'::jsonb),
'columns', COALESCE((
SELECT jsonb_agg(jsonb_build_object(
'table', table_name,
'position', position,
'name', column_name,
'type', data_type,
'notNull', not_null,
'default', default_expr,
'identity', identity_kind,
'generated', generated_kind
) ORDER BY table_name, position)
FROM columns
), '[]'::jsonb),
'constraints', COALESCE((
SELECT jsonb_agg(jsonb_build_object(
'table', table_name,
'name', constraint_name,
'type', constraint_type,
'valid', is_valid,
'definition', definition
) ORDER BY table_name, constraint_name)
FROM constraints
), '[]'::jsonb),
'indexes', COALESCE((
SELECT jsonb_agg(jsonb_build_object(
'table', table_name,
'name', index_name,
'definition', definition,
'valid', is_valid,
'constraintOwned', constraint_owned
) ORDER BY table_name, index_name)
FROM indexes
), '[]'::jsonb),
'types', COALESCE((
SELECT jsonb_agg(jsonb_build_object(
'name', type_name,
'kind', type_kind,
'definition', definition
) ORDER BY type_name)
FROM types
), '[]'::jsonb),
'sequences', COALESCE((
SELECT jsonb_agg(jsonb_build_object(
'name', sequence_name,
'type', data_type,
'start', start_value,
'increment', increment_by,
'min', min_value,
'max', max_value,
'cache', cache_size,
'cycle', cycle,
'ownedTable', owned_table,
'ownedColumn', owned_column
) ORDER BY sequence_name)
FROM sequences
), '[]'::jsonb),
'functions', COALESCE((
SELECT jsonb_agg(jsonb_build_object(
'name', function_name,
'arguments', arguments,
'definition', definition
) ORDER BY function_name, arguments)
FROM functions
), '[]'::jsonb)
) AS catalog
FROM (SELECT CURRENT_TIMESTAMP AS checked_at) AS fresh_catalog_read
WHERE fresh_catalog_read.checked_at IS NOT NULL
`
export function stableStringify(value: unknown): string {
return JSON.stringify(sortJson(value), null, 2)
}
export async function readReplicaSchemaCatalog(
client: Queryable,
): Promise<unknown> {
const result = await client.query(READ_REPLICA_SCHEMA_CATALOG_SQL, [
REPLICA_TABLES,
REPLICA_TYPES,
REPLICA_SEQUENCES,
REPLICA_FUNCTIONS,
REPLICA_EXCLUDED_INDEXES,
])
const catalog = result.rows[0]?.catalog
if (!catalog)
throw new Error('Read-replica schema catalog query returned no rows')
return sortJson(catalog)
}