355 lines
16 KiB
JavaScript
355 lines
16 KiB
JavaScript
import { backfillManagedReaderSourceConnections } from "./reader-source-scope.mjs";
|
|
|
|
export async function migrate(pool) {
|
|
await pool.query("create extension if not exists timescaledb");
|
|
await pool.query("create extension if not exists postgis");
|
|
|
|
await pool.query(`
|
|
create table if not exists external_data_plane_products (
|
|
id text primary key,
|
|
version text not null,
|
|
ontology_revision text not null,
|
|
delivery_mode text not null,
|
|
semantic_types jsonb not null,
|
|
fields jsonb not null,
|
|
field_contracts jsonb not null default '{}'::jsonb,
|
|
history_policy jsonb not null,
|
|
active boolean not null default true,
|
|
created_at timestamptz not null default now(),
|
|
updated_at timestamptz not null default now(),
|
|
check (jsonb_typeof(semantic_types) = 'array' and jsonb_array_length(semantic_types) > 0),
|
|
check (jsonb_typeof(fields) = 'array' and jsonb_array_length(fields) > 0),
|
|
constraint external_data_plane_products_field_contracts_object_ck
|
|
check (jsonb_typeof(field_contracts) = 'object')
|
|
)
|
|
`);
|
|
await pool.query("alter table external_data_plane_products add column if not exists field_contracts jsonb not null default '{}'::jsonb");
|
|
await pool.query(`
|
|
do $$
|
|
begin
|
|
if not exists (
|
|
select 1 from pg_constraint
|
|
where conrelid = 'external_data_plane_products'::regclass
|
|
and conname = 'external_data_plane_products_field_contracts_object_ck'
|
|
) then
|
|
alter table external_data_plane_products
|
|
add constraint external_data_plane_products_field_contracts_object_ck
|
|
check (jsonb_typeof(field_contracts) = 'object');
|
|
end if;
|
|
end
|
|
$$
|
|
`);
|
|
|
|
await pool.query(`
|
|
create table if not exists external_data_plane_batches (
|
|
id uuid primary key,
|
|
tenant_id text not null,
|
|
connection_id text not null,
|
|
provider_id text not null,
|
|
data_product_id text not null,
|
|
contract_version text not null,
|
|
ontology_revision text not null,
|
|
run_id text not null,
|
|
sequence integer not null,
|
|
idempotency_key text not null,
|
|
received_at timestamptz not null,
|
|
fact_count integer not null default 0,
|
|
inserted_fact_count integer not null default 0,
|
|
created_at timestamptz not null default now(),
|
|
unique (tenant_id, connection_id, provider_id, data_product_id, idempotency_key)
|
|
)
|
|
`);
|
|
await pool.query("create index if not exists external_data_plane_batches_scope_idx on external_data_plane_batches (tenant_id, connection_id, data_product_id, received_at desc)");
|
|
await pool.query("alter table external_data_plane_batches add column if not exists current_updated_count integer not null default 0");
|
|
await pool.query("alter table external_data_plane_batches add column if not exists current_removed_count integer not null default 0");
|
|
await pool.query("alter table external_data_plane_batches add column if not exists history_inserted_count integer not null default 0");
|
|
await pool.query("alter table external_data_plane_batches add column if not exists patch_operation_count integer not null default 0");
|
|
await pool.query("alter table external_data_plane_batches add column if not exists delivery_cursor bigint");
|
|
await pool.query("alter table external_data_plane_batches add column if not exists request_fingerprint text");
|
|
|
|
await pool.query(`
|
|
create table if not exists external_data_plane_raw_envelopes (
|
|
id uuid primary key,
|
|
batch_id uuid not null unique references external_data_plane_batches(id) on delete cascade,
|
|
payload_hash text not null,
|
|
content_type text not null,
|
|
payload jsonb,
|
|
payload_ref text,
|
|
payload_bytes integer,
|
|
received_at timestamptz not null,
|
|
expires_at timestamptz,
|
|
check (payload is not null or payload_ref is not null)
|
|
)
|
|
`);
|
|
await pool.query("create index if not exists external_data_plane_raw_expiry_idx on external_data_plane_raw_envelopes (expires_at)");
|
|
|
|
await pool.query(`
|
|
create table if not exists external_data_plane_writer_bindings (
|
|
id uuid primary key,
|
|
token_hash text not null unique,
|
|
binding_key text,
|
|
request_hash text,
|
|
generation integer not null default 1,
|
|
tenant_id text not null,
|
|
connection_id text not null,
|
|
provider_id text not null,
|
|
allowed_data_product_ids jsonb not null,
|
|
expires_at timestamptz,
|
|
active boolean not null default true,
|
|
created_at timestamptz not null default now(),
|
|
rotated_at timestamptz,
|
|
revoked_at timestamptz,
|
|
check (generation > 0),
|
|
check (
|
|
case when jsonb_typeof(allowed_data_product_ids) = 'array'
|
|
then jsonb_array_length(allowed_data_product_ids) > 0
|
|
else false
|
|
end
|
|
)
|
|
)
|
|
`);
|
|
await pool.query("alter table external_data_plane_writer_bindings add column if not exists binding_key text");
|
|
await pool.query("alter table external_data_plane_writer_bindings add column if not exists request_hash text");
|
|
await pool.query("alter table external_data_plane_writer_bindings add column if not exists generation integer not null default 1");
|
|
await pool.query("alter table external_data_plane_writer_bindings alter column expires_at drop not null");
|
|
await pool.query(`
|
|
do $$
|
|
begin
|
|
if not exists (
|
|
select 1 from pg_constraint
|
|
where conrelid = 'external_data_plane_writer_bindings'::regclass
|
|
and conname = 'external_data_plane_writer_bindings_generation_positive_ck'
|
|
) then
|
|
alter table external_data_plane_writer_bindings
|
|
add constraint external_data_plane_writer_bindings_generation_positive_ck
|
|
check (generation > 0);
|
|
end if;
|
|
if not exists (
|
|
select 1 from pg_constraint
|
|
where conrelid = 'external_data_plane_writer_bindings'::regclass
|
|
and conname = 'external_data_plane_writer_bindings_managed_metadata_ck'
|
|
) then
|
|
alter table external_data_plane_writer_bindings
|
|
add constraint external_data_plane_writer_bindings_managed_metadata_ck
|
|
check (
|
|
(binding_key is null and request_hash is null)
|
|
or (binding_key is not null and request_hash ~ '^[a-f0-9]{64}$')
|
|
);
|
|
end if;
|
|
if not exists (
|
|
select 1 from pg_constraint
|
|
where conrelid = 'external_data_plane_writer_bindings'::regclass
|
|
and conname = 'external_data_plane_writer_bindings_lifecycle_ck'
|
|
) then
|
|
alter table external_data_plane_writer_bindings
|
|
add constraint external_data_plane_writer_bindings_lifecycle_ck
|
|
check (
|
|
(binding_key is null and expires_at is not null)
|
|
or (binding_key is not null and expires_at is null)
|
|
) not valid;
|
|
end if;
|
|
end
|
|
$$
|
|
`);
|
|
await pool.query("create unique index if not exists external_data_plane_writer_bindings_managed_key_idx on external_data_plane_writer_bindings (binding_key, generation) where binding_key is not null");
|
|
await pool.query("create index if not exists external_data_plane_writer_bindings_active_idx on external_data_plane_writer_bindings (active, expires_at)");
|
|
|
|
await pool.query(`
|
|
create table if not exists external_data_plane_reader_bindings (
|
|
id uuid primary key,
|
|
token_hash text not null unique,
|
|
binding_key text,
|
|
request_hash text,
|
|
generation integer not null default 1,
|
|
tenant_id text not null,
|
|
connection_id text not null,
|
|
provider_id text not null,
|
|
allowed_data_product_ids jsonb not null,
|
|
expires_at timestamptz,
|
|
active boolean not null default true,
|
|
created_at timestamptz not null default now(),
|
|
rotated_at timestamptz,
|
|
revoked_at timestamptz,
|
|
check (generation > 0),
|
|
check (
|
|
case when jsonb_typeof(allowed_data_product_ids) = 'array'
|
|
then jsonb_array_length(allowed_data_product_ids) > 0
|
|
else false
|
|
end
|
|
)
|
|
)
|
|
`);
|
|
await pool.query("alter table external_data_plane_reader_bindings add column if not exists binding_key text");
|
|
await pool.query("alter table external_data_plane_reader_bindings add column if not exists request_hash text");
|
|
await pool.query("alter table external_data_plane_reader_bindings add column if not exists generation integer not null default 1");
|
|
await pool.query("alter table external_data_plane_reader_bindings add column if not exists source_connection_id text");
|
|
await pool.query("alter table external_data_plane_reader_bindings alter column expires_at drop not null");
|
|
await pool.query(`
|
|
do $$
|
|
begin
|
|
if not exists (
|
|
select 1 from pg_constraint
|
|
where conrelid = 'external_data_plane_reader_bindings'::regclass
|
|
and conname = 'external_data_plane_reader_bindings_generation_positive_ck'
|
|
) then
|
|
alter table external_data_plane_reader_bindings
|
|
add constraint external_data_plane_reader_bindings_generation_positive_ck
|
|
check (generation > 0);
|
|
end if;
|
|
if not exists (
|
|
select 1 from pg_constraint
|
|
where conrelid = 'external_data_plane_reader_bindings'::regclass
|
|
and conname = 'external_data_plane_reader_bindings_managed_metadata_ck'
|
|
) then
|
|
alter table external_data_plane_reader_bindings
|
|
add constraint external_data_plane_reader_bindings_managed_metadata_ck
|
|
check (
|
|
(binding_key is null and request_hash is null)
|
|
or (binding_key is not null and request_hash ~ '^[a-f0-9]{64}$')
|
|
);
|
|
end if;
|
|
if not exists (
|
|
select 1 from pg_constraint
|
|
where conrelid = 'external_data_plane_reader_bindings'::regclass
|
|
and conname = 'external_data_plane_reader_bindings_lifecycle_ck'
|
|
) then
|
|
alter table external_data_plane_reader_bindings
|
|
add constraint external_data_plane_reader_bindings_lifecycle_ck
|
|
check (
|
|
(binding_key is null and expires_at is not null)
|
|
or (binding_key is not null and expires_at is null)
|
|
);
|
|
end if;
|
|
end
|
|
$$
|
|
`);
|
|
await pool.query("create unique index if not exists external_data_plane_reader_bindings_managed_key_idx on external_data_plane_reader_bindings (binding_key, generation) where binding_key is not null");
|
|
await pool.query("create index if not exists external_data_plane_reader_bindings_active_idx on external_data_plane_reader_bindings (active, expires_at)");
|
|
await pool.query("create index if not exists external_data_plane_reader_bindings_source_idx on external_data_plane_reader_bindings (tenant_id, source_connection_id, provider_id, active)");
|
|
await pool.query(`
|
|
update external_data_plane_reader_bindings
|
|
set source_connection_id = connection_id
|
|
where binding_key is null and source_connection_id is null
|
|
`);
|
|
await backfillManagedReaderSourceConnections(pool);
|
|
|
|
await pool.query(`
|
|
create table if not exists external_data_plane_facts (
|
|
id uuid not null,
|
|
batch_id uuid not null references external_data_plane_batches(id) on delete cascade,
|
|
tenant_id text not null,
|
|
connection_id text not null,
|
|
provider_id text not null,
|
|
data_product_id text not null,
|
|
source_id text not null,
|
|
semantic_type text not null,
|
|
observed_at timestamptz not null,
|
|
received_at timestamptz not null,
|
|
attributes jsonb not null default '{}'::jsonb,
|
|
geometry geography(Geometry, 4326),
|
|
fingerprint text not null,
|
|
primary key (id, observed_at),
|
|
unique (tenant_id, connection_id, provider_id, data_product_id, source_id, semantic_type, observed_at, fingerprint)
|
|
)
|
|
`);
|
|
await pool.query("select create_hypertable('external_data_plane_facts', 'observed_at', if_not_exists => true, migrate_data => true)");
|
|
await pool.query("create index if not exists external_data_plane_facts_source_time_idx on external_data_plane_facts (tenant_id, connection_id, data_product_id, source_id, observed_at desc)");
|
|
await pool.query("create index if not exists external_data_plane_facts_geometry_idx on external_data_plane_facts using gist (geometry)");
|
|
|
|
await pool.query(`
|
|
create table if not exists external_data_plane_current (
|
|
tenant_id text not null,
|
|
connection_id text not null,
|
|
provider_id text not null,
|
|
data_product_id text not null,
|
|
source_id text not null,
|
|
semantic_type text not null,
|
|
observed_at timestamptz not null,
|
|
received_at timestamptz not null,
|
|
attributes jsonb not null default '{}'::jsonb,
|
|
geometry geography(Geometry, 4326),
|
|
fingerprint text not null,
|
|
updated_at timestamptz not null default now(),
|
|
primary key (tenant_id, connection_id, provider_id, data_product_id, source_id, semantic_type)
|
|
)
|
|
`);
|
|
await pool.query(`
|
|
do $$
|
|
begin
|
|
if exists (
|
|
select 1
|
|
from pg_attribute
|
|
where attrelid = 'external_data_plane_current'::regclass
|
|
and attname = 'geometry'
|
|
and not attisdropped
|
|
and format_type(atttypid, atttypmod) = 'geography(Point,4326)'
|
|
) then
|
|
alter table external_data_plane_current
|
|
alter column geometry type geography(Geometry, 4326)
|
|
using geometry::geometry::geography;
|
|
end if;
|
|
end
|
|
$$
|
|
`);
|
|
await pool.query("create index if not exists external_data_plane_current_scope_idx on external_data_plane_current (tenant_id, connection_id, data_product_id, observed_at desc)");
|
|
await pool.query("create index if not exists external_data_plane_current_geometry_idx on external_data_plane_current using gist (geometry)");
|
|
|
|
await pool.query(`
|
|
create table if not exists external_data_plane_delivery_state (
|
|
tenant_id text not null,
|
|
connection_id text not null,
|
|
provider_id text not null,
|
|
data_product_id text not null,
|
|
current_cursor bigint not null default 0,
|
|
current_generation_at timestamptz,
|
|
current_generation_id text,
|
|
updated_at timestamptz not null default now(),
|
|
primary key (tenant_id, connection_id, provider_id, data_product_id)
|
|
)
|
|
`);
|
|
await pool.query("alter table external_data_plane_delivery_state add column if not exists current_generation_at timestamptz");
|
|
await pool.query("alter table external_data_plane_delivery_state add column if not exists current_generation_id text");
|
|
|
|
await pool.query(`
|
|
create table if not exists external_data_plane_patch_outbox (
|
|
tenant_id text not null,
|
|
connection_id text not null,
|
|
provider_id text not null,
|
|
data_product_id text not null,
|
|
cursor bigint not null,
|
|
previous_cursor bigint not null,
|
|
batch_id uuid not null references external_data_plane_batches(id) on delete cascade,
|
|
operations jsonb not null,
|
|
emitted_at timestamptz not null default now(),
|
|
primary key (tenant_id, connection_id, provider_id, data_product_id, cursor),
|
|
check (jsonb_typeof(operations) = 'array' and jsonb_array_length(operations) > 0)
|
|
)
|
|
`);
|
|
await pool.query("create index if not exists external_data_plane_patch_outbox_retention_idx on external_data_plane_patch_outbox (emitted_at)");
|
|
|
|
await pool.query(`
|
|
create table if not exists external_data_plane_history (
|
|
tenant_id text not null,
|
|
connection_id text not null,
|
|
provider_id text not null,
|
|
data_product_id text not null,
|
|
source_id text not null,
|
|
semantic_type text not null,
|
|
bucket_start timestamptz not null,
|
|
observed_at timestamptz not null,
|
|
received_at timestamptz not null,
|
|
attributes jsonb not null default '{}'::jsonb,
|
|
geometry geography(Geometry, 4326),
|
|
fingerprint text not null,
|
|
updated_at timestamptz not null default now(),
|
|
primary key (
|
|
tenant_id, connection_id, provider_id, data_product_id,
|
|
source_id, semantic_type, bucket_start
|
|
)
|
|
)
|
|
`);
|
|
await pool.query("select create_hypertable('external_data_plane_history', 'bucket_start', if_not_exists => true, migrate_data => true)");
|
|
await pool.query("create index if not exists external_data_plane_history_source_time_idx on external_data_plane_history (tenant_id, connection_id, data_product_id, source_id, bucket_start desc)");
|
|
await pool.query("create index if not exists external_data_plane_history_geometry_idx on external_data_plane_history using gist (geometry)");
|
|
}
|