apps/backend/src/infrastructure/database/migrations/0044_sprint_v_external_integrations.sqlSQL
-- Sprint V increment 1 — PMS / external POS / supplier integrations.
create table if not exists external_system_mappings (
id uuid primary key default gen_random_uuid(),
organization_id uuid not null references organizations(id) on delete cascade,
integration_connection_id uuid not null references integration_connections(id) on delete cascade,
location_id uuid references locations(id) on delete cascade,
mapping_type text not null,
local_entity_type text not null,
local_entity_id uuid,
external_entity_type text not null,
external_id text not null,
external_parent_id text,
direction text not null default 'bidirectional',
status text not null default 'active',
metadata_json jsonb not null default '{}'::jsonb,
created_at timestamptz not null default now(),
updated_at timestamptz not null default now(),
constraint external_mapping_type_check check(
mapping_type in (
'room','reservation','guest','folio','charge_code',
'product','modifier','category','tax','tender',
'supplier','supplier_product','purchase_order','goods_receipt'
)
),
constraint external_mapping_direction_check check(direction in ('inbound','outbound','bidirectional')),
constraint external_mapping_status_check check(status in ('active','disabled','conflict'))
);
create unique index if not exists idx_external_mapping_external_unique
on external_system_mappings(integration_connection_id,external_entity_type,external_id);
create unique index if not exists idx_external_mapping_local_unique
on external_system_mappings(integration_connection_id,local_entity_type,local_entity_id)
where local_entity_id is not null;
create table if not exists external_integration_inbox (
id uuid primary key default gen_random_uuid(),
organization_id uuid not null references organizations(id) on delete cascade,
integration_connection_id uuid not null references integration_connections(id) on delete cascade,
provider text not null,
external_event_id text not null,
event_type text not null,
external_entity_type text,
external_entity_id text,
payload_json jsonb not null default '{}'::jsonb,
payload_hash text,
provider_version text,
provider_updated_at timestamptz,
status text not null default 'received',
attempt_count integer not null default 0,
next_attempt_at timestamptz,
received_at timestamptz not null default now(),
processed_at timestamptz,
last_error text,
constraint external_integration_inbox_status_check check(
status in ('received','processing','processed','ignored','failed','dead_letter')
),
constraint external_integration_inbox_unique unique(integration_connection_id,external_event_id)
);
create index if not exists idx_external_integration_inbox_work
on external_integration_inbox(status,next_attempt_at,received_at);
create table if not exists external_integration_outbox (
id uuid primary key default gen_random_uuid(),
organization_id uuid not null references organizations(id) on delete cascade,
integration_connection_id uuid not null references integration_connections(id) on delete cascade,
provider text not null,
local_entity_type text not null,
local_entity_id uuid not null,
command_type text not null,
idempotency_key text not null,
payload_json jsonb not null default '{}'::jsonb,
status text not null default 'queued',
attempt_count integer not null default 0,
next_attempt_at timestamptz,
provider_reference text,
provider_status text,
created_at timestamptz not null default now(),
completed_at timestamptz,
last_error text,
constraint external_integration_outbox_status_check check(
status in ('queued','processing','completed','failed','dead_letter','unsupported','uncertain')
),
constraint external_integration_outbox_unique unique(integration_connection_id,idempotency_key)
);
create index if not exists idx_external_integration_outbox_work
on external_integration_outbox(status,next_attempt_at,created_at);
create table if not exists external_integration_conflicts (
id uuid primary key default gen_random_uuid(),
organization_id uuid not null references organizations(id) on delete cascade,
location_id uuid references locations(id) on delete set null,
integration_connection_id uuid not null references integration_connections(id) on delete cascade,
conflict_type text not null,
local_entity_type text,
local_entity_id uuid,
external_entity_type text,
external_id text,
status text not null default 'open',
severity text not null default 'high',
local_snapshot_json jsonb not null default '{}'::jsonb,
external_snapshot_json jsonb not null default '{}'::jsonb,
summary text not null,
created_at timestamptz not null default now(),
resolved_at timestamptz,
resolved_by_user_id uuid references user_accounts(id) on delete set null,
constraint external_integration_conflict_type_check check(
conflict_type in (
'mapping_missing','mapping_ambiguous','version_conflict','ownership_conflict',
'unsupported_state','tax_mapping_missing','tender_mapping_missing',
'product_mapping_missing','supplier_mapping_missing','provider_rejected'
)
),
constraint external_integration_conflict_status_check check(status in ('open','resolved','ignored')),
constraint external_integration_conflict_severity_check check(severity in ('medium','high','critical'))
);
create table if not exists external_integration_conflict_resolutions (
id uuid primary key default gen_random_uuid(),
organization_id uuid not null references organizations(id) on delete cascade,
conflict_id uuid not null references external_integration_conflicts(id) on delete cascade,
action text not null,
actor_user_id uuid references user_accounts(id) on delete set null,
note text,
result_json jsonb not null default '{}'::jsonb,
created_at timestamptz not null default now(),
constraint external_integration_conflict_resolution_action_check check(
action in ('map_existing','create_local','use_external','keep_local','ignore')
)
);
create table if not exists external_integration_health_history (
id uuid primary key default gen_random_uuid(),
organization_id uuid not null references organizations(id) on delete cascade,
integration_connection_id uuid not null references integration_connections(id) on delete cascade,
ok boolean not null,
status_code integer,
code text,
latency_ms integer,
checked_at timestamptz not null default now()
);
create table if not exists external_sync_cursors (
id uuid primary key default gen_random_uuid(),
organization_id uuid not null references organizations(id) on delete cascade,
integration_connection_id uuid not null references integration_connections(id) on delete cascade,
stream_key text not null,
cursor_value text,
last_polled_at timestamptz,
last_success_at timestamptz,
updated_at timestamptz not null default now(),
constraint external_sync_cursor_unique unique(integration_connection_id,stream_key)
);
create table if not exists pms_posted_charges (
id uuid primary key default gen_random_uuid(),
organization_id uuid not null references organizations(id) on delete cascade,
location_id uuid not null references locations(id) on delete cascade,
integration_connection_id uuid not null references integration_connections(id) on delete cascade,
service_session_id uuid references service_sessions(id) on delete set null,
check_id uuid references checks(id) on delete set null,
external_folio_id text not null,
external_reservation_id text,
external_guest_id text,
charge_code text not null,
amount_minor bigint not null,
currency_code text not null,
description text not null,
provider_charge_id text,
status text not null default 'queued',
idempotency_key text not null,
created_at timestamptz not null default now(),
completed_at timestamptz,
last_error text,
constraint pms_posted_charge_status_check check(
status in ('queued','processing','completed','failed','uncertain','reversed')
),
constraint pms_posted_charge_unique unique(integration_connection_id,idempotency_key)
);
create table if not exists external_pos_import_batches (
id uuid primary key default gen_random_uuid(),
organization_id uuid not null references organizations(id) on delete cascade,
location_id uuid not null references locations(id) on delete cascade,
integration_connection_id uuid not null references integration_connections(id) on delete cascade,
external_business_date date not null,
provider_batch_id text not null,
status text not null default 'received',
sales_gross_minor bigint not null default 0,
sales_net_minor bigint not null default 0,
tax_minor bigint not null default 0,
payload_json jsonb not null default '{}'::jsonb,
received_at timestamptz not null default now(),
processed_at timestamptz,
constraint external_pos_import_batch_status_check check(
status in ('received','processing','processed','failed','conflict')
),
constraint external_pos_import_batch_unique unique(integration_connection_id,provider_batch_id)
);
create table if not exists supplier_catalog_imports (
id uuid primary key default gen_random_uuid(),
organization_id uuid not null references organizations(id) on delete cascade,
integration_connection_id uuid not null references integration_connections(id) on delete cascade,
supplier_id uuid references suppliers(id) on delete set null,
external_catalog_id text,
status text not null default 'received',
imported_items integer not null default 0,
changed_items integer not null default 0,
rejected_items integer not null default 0,
source_file_asset_id uuid references file_assets(id) on delete set null,
payload_json jsonb not null default '{}'::jsonb,
created_at timestamptz not null default now(),
completed_at timestamptz,
constraint supplier_catalog_import_status_check check(
status in ('received','preview','committing','completed','failed','conflict')
)
);
alter table integration_connections
add column if not exists system_role text,
add column if not exists source_of_truth_json jsonb not null default '{}'::jsonb,
add column if not exists polling_interval_seconds integer,
add column if not exists last_successful_sync_at timestamptz;
alter table integration_connections drop constraint if exists integration_connection_system_role_check;
alter table integration_connections add constraint integration_connection_system_role_check
check(system_role is null or system_role in ('source','mirror','peer'));
insert into permissions(permission_key,description) values
('integrations.external.read','Read PMS/POS/supplier integration state'),
('integrations.external.manage','Manage PMS/POS/supplier integrations'),
('integrations.external.sync','Run external integration sync'),
('integrations.external.conflicts','Resolve external integration conflicts'),
('integrations.pms.charge','Post charges to PMS folios')
on conflict(permission_key) do nothing;