"""initial canonical schema

Revision ID: 0001
Revises:
Create Date: 2026-09-11
"""
from __future__ import annotations

from alembic import op

revision = "0001"
down_revision = None
branch_labels = None
depends_on = None

SQL = r"""
create extension if not exists pg_trgm;
create extension if not exists "uuid-ossp";

-- ---------------------------------------------------------------- sources & connector observability
create table sources (
  id text primary key,                          -- e.g. 'celestrak'
  name text not null,
  type text not null,                           -- orbital | catalog | registry | regulatory | company | news | weather
  base_url text,
  official boolean not null default false,
  country_code text,
  authority_type text,                          -- government | intergovernmental | scientific | operator | secondary
  license text,
  attribution_required boolean not null default true,
  attribution_text text,
  update_frequency_seconds integer,
  enabled boolean not null default true,
  priority integer not null default 100,
  created_at timestamptz not null default now(),
  updated_at timestamptz not null default now()
);

create table connectors (
  name text primary key,                        -- e.g. 'celestrak_gp'
  source_id text not null references sources(id),
  description text,
  interval_seconds integer not null,
  enabled boolean not null default true,
  priority integer not null default 100,
  config jsonb not null default '{}'::jsonb,
  consecutive_failures integer not null default 0,
  circuit_open_until timestamptz,
  last_success_at timestamptz,
  last_attempt_at timestamptz,
  last_duration_ms integer,
  next_run_at timestamptz,
  created_at timestamptz not null default now(),
  updated_at timestamptz not null default now()
);

create table connector_runs (
  id text primary key,
  connector_name text not null references connectors(name),
  source_id text not null references sources(id),
  started_at timestamptz not null default now(),
  finished_at timestamptz,
  status text not null default 'running',       -- running | success | unchanged | failed | skipped
  duration_ms integer,
  records_fetched integer not null default 0,
  records_created integer not null default 0,
  records_updated integer not null default 0,
  records_skipped integer not null default 0,
  error text,
  payload_hash text,
  meta jsonb not null default '{}'::jsonb
);
create index connector_runs_name_started_idx on connector_runs (connector_name, started_at desc);

create table connector_errors (
  id bigserial primary key,
  connector_name text not null references connectors(name),
  run_id text references connector_runs(id),
  occurred_at timestamptz not null default now(),
  error_type text,
  message text not null,
  context jsonb
);

create table raw_records (
  id text primary key,
  source_id text not null references sources(id),
  connector_name text not null,
  run_id text references connector_runs(id),
  source_native_id text,                        -- group name, file name, query…
  content_type text not null,
  payload_hash text not null,
  byte_size bigint not null default 0,
  storage_path text,                            -- relative to SI_DATA_DIR/raw
  source_url text,
  fetched_at timestamptz not null default now(),
  processed_at timestamptz,
  processing_status text not null default 'pending',   -- pending | processed | failed | unchanged
  record_count integer,
  error text
);
create index raw_records_source_fetched_idx on raw_records (source_id, fetched_at desc);
create index raw_records_hash_idx on raw_records (payload_hash);

-- ---------------------------------------------------------------- reference: countries, owners, launch sites
create table countries (
  code text primary key,                        -- ISO 3166-1 alpha-2
  iso3 text,
  name text not null,
  slug text not null unique,
  region text,
  flag text
);

create table owner_codes (
  code text primary key,                        -- SATCAT owner code (US, PRC, CIS, SES, ESA…)
  name text not null,
  kind text not null,                           -- country | organization | intergovernmental | consortium | unknown
  country_code text references countries(code),
  organization_id text
);

create table launch_sites (
  code text primary key,                        -- SATCAT launch site code
  name text not null,
  slug text not null unique,
  country_code text references countries(code),
  latitude double precision,
  longitude double precision,
  active boolean not null default true
);

-- ---------------------------------------------------------------- organizations & constellations
create table organizations (
  id text primary key,
  slug text not null unique,
  name text not null,
  normalized_name text not null,
  kind text not null default 'operator',        -- operator | manufacturer | agency | military | launch_provider | consortium
  country_code text references countries(code),
  official_url text,
  description text,
  founded_year integer,
  created_at timestamptz not null default now(),
  updated_at timestamptz not null default now()
);
create index organizations_normalized_idx on organizations (normalized_name);
create index organizations_trgm_idx on organizations using gin (name gin_trgm_ops);

create table organization_aliases (
  id bigserial primary key,
  organization_id text not null references organizations(id) on delete cascade,
  alias text not null,
  normalized text not null,
  source_id text references sources(id),
  unique (organization_id, normalized)
);
create index organization_aliases_normalized_idx on organization_aliases (normalized);

create table constellations (
  id text primary key,
  slug text not null unique,
  name text not null,
  operator_id text references organizations(id),
  country_code text references countries(code),
  service_type text,                            -- communications | earth-observation | navigation | iot | weather | science | military | technology
  orbit_class text,                             -- LEO | MEO | GEO | HEO | MIXED
  lifecycle_stage text not null default 'OPERATIONAL',
  description text,
  official_url text,
  planned_count integer,
  authorized_count integer,
  match_patterns jsonb not null default '[]'::jsonb,
  celestrak_groups jsonb not null default '[]'::jsonb,
  created_at timestamptz not null default now(),
  updated_at timestamptz not null default now()
);

-- ---------------------------------------------------------------- launches
create table launch_vehicle_families (
  id text primary key, slug text not null unique, name text not null, manufacturer_id text references organizations(id), country_code text references countries(code)
);
create table launch_vehicles (
  id text primary key, slug text not null unique, name text not null, family_id text references launch_vehicle_families(id), variant text
);

create table launches (
  id text primary key,
  cospar_launch_id text not null unique,        -- '1998-067'
  launch_date date,
  launch_year integer,
  launch_site_code text references launch_sites(code),
  launch_vehicle_id text references launch_vehicles(id),
  provider_id text references organizations(id),
  owner_codes text[] not null default '{}',
  payload_count integer not null default 0,
  object_count integer not null default 0,
  on_orbit_count integer not null default 0,
  primary_name text,
  first_seen_at timestamptz not null default now(),
  updated_at timestamptz not null default now()
);
create index launches_date_idx on launches (launch_date desc);
create index launches_site_idx on launches (launch_site_code);

-- ---------------------------------------------------------------- satellites (canonical objects)
create table satellites (
  id text primary key,
  slug text not null unique,
  canonical_name text not null,
  normalized_name text not null,
  norad_id integer unique,
  cospar_id text,
  object_type text not null default 'UNKNOWN',  -- PAYLOAD | ROCKET_BODY | DEBRIS | UNKNOWN | STATION | CREWED
  status text not null default 'UNKNOWN',       -- ACTIVE | INACTIVE | DECAYED | LOST | FAILED | UNKNOWN | PLANNED
  ops_status_code text,                         -- SATCAT: + P B S X D ?
  operator_id text references organizations(id),
  owner_code text references owner_codes(code),
  country_code text references countries(code),
  constellation_id text references constellations(id),
  launch_id text references launches(id),
  launch_date date,
  launch_site_code text references launch_sites(code),
  decay_date date,
  mission_type text,                            -- communications | earth-observation | navigation | weather | science | military | technology | station | unknown
  orbit_class text,                             -- LEO | MEO | GEO | HEO | OTHER
  period_minutes double precision,
  inclination_deg double precision,
  apogee_km double precision,
  perigee_km double precision,
  rcs_m2 double precision,
  orbit_center text,
  orbit_type text,
  has_gp boolean not null default false,
  latest_epoch timestamptz,
  mass_kg double precision,
  description text,
  official_url text,
  first_seen_at timestamptz not null default now(),
  last_seen_at timestamptz not null default now(),
  created_at timestamptz not null default now(),
  updated_at timestamptz not null default now()
);
create index satellites_cospar_idx on satellites (cospar_id);
create index satellites_status_idx on satellites (status);
create index satellites_type_status_idx on satellites (object_type, status);
create index satellites_constellation_idx on satellites (constellation_id);
create index satellites_operator_idx on satellites (operator_id);
create index satellites_country_idx on satellites (country_code);
create index satellites_launch_idx on satellites (launch_id);
create index satellites_launch_date_idx on satellites (launch_date desc);
create index satellites_decay_date_idx on satellites (decay_date desc);
create index satellites_orbit_class_idx on satellites (orbit_class);
create index satellites_name_trgm_idx on satellites using gin (canonical_name gin_trgm_ops);
create index satellites_norad_text_idx on satellites ((norad_id::text) text_pattern_ops);

create table satellite_aliases (
  id bigserial primary key,
  satellite_id text not null references satellites(id) on delete cascade,
  alias text not null,
  normalized text not null,
  source_id text references sources(id),
  unique (satellite_id, normalized)
);
create index satellite_aliases_normalized_idx on satellite_aliases (normalized);

create table satellite_slugs (
  slug text primary key,
  satellite_id text not null references satellites(id) on delete cascade,
  created_at timestamptz not null default now()
);

create table satellite_tags (
  satellite_id text not null references satellites(id) on delete cascade,
  tag text not null,                            -- celestrak group name or derived tag
  source_id text references sources(id),
  first_seen_at timestamptz not null default now(),
  last_seen_at timestamptz not null default now(),
  primary key (satellite_id, tag)
);
create index satellite_tags_tag_idx on satellite_tags (tag);

create table satellite_status_history (
  id bigserial primary key,
  satellite_id text not null references satellites(id) on delete cascade,
  field text not null,
  old_value text,
  new_value text,
  source_id text references sources(id),
  changed_at timestamptz not null default now()
);
create index satellite_status_history_sat_idx on satellite_status_history (satellite_id, changed_at desc);
create index satellite_status_history_changed_idx on satellite_status_history (changed_at desc);

create table constellation_memberships (
  id bigserial primary key,
  satellite_id text not null references satellites(id) on delete cascade,
  constellation_id text not null references constellations(id) on delete cascade,
  method text not null,                          -- celestrak_group | name_pattern | manual
  since timestamptz not null default now(),
  until timestamptz
);
create index constellation_memberships_sat_idx on constellation_memberships (satellite_id) where until is null;

-- ---------------------------------------------------------------- orbital history (append only)
create table orbital_elements (
  id bigserial primary key,
  satellite_id text not null references satellites(id) on delete cascade,
  source_id text not null references sources(id),
  epoch timestamptz not null,
  mean_motion double precision not null,
  eccentricity double precision not null,
  inclination double precision not null,
  raan double precision not null,
  arg_of_perigee double precision not null,
  mean_anomaly double precision not null,
  bstar double precision,
  mean_motion_dot double precision,
  mean_motion_ddot double precision,
  element_set_no integer,
  rev_at_epoch integer,
  classification text,
  ephemeris_type integer,
  semi_major_axis_km double precision,
  perigee_km double precision,
  apogee_km double precision,
  period_minutes double precision,
  element_format text not null default 'omm_json',
  raw_omm jsonb,
  received_at timestamptz not null default now(),
  created_at timestamptz not null default now(),
  unique (satellite_id, source_id, epoch)
);
create index orbital_elements_sat_epoch_idx on orbital_elements (satellite_id, epoch desc);
create index orbital_elements_received_idx on orbital_elements (received_at desc);

-- latest element set per satellite (kept in sync by the connector; avoids DISTINCT ON over history)
create table orbital_state (
  satellite_id text primary key references satellites(id) on delete cascade,
  element_id bigint not null references orbital_elements(id),
  source_id text not null references sources(id),
  epoch timestamptz not null,
  mean_motion double precision not null,
  eccentricity double precision not null,
  inclination double precision not null,
  raan double precision not null,
  arg_of_perigee double precision not null,
  mean_anomaly double precision not null,
  bstar double precision,
  mean_motion_dot double precision,
  mean_motion_ddot double precision,
  semi_major_axis_km double precision,
  perigee_km double precision,
  apogee_km double precision,
  period_minutes double precision,
  orbit_class text,
  updated_at timestamptz not null default now()
);
create index orbital_state_epoch_idx on orbital_state (epoch);
create index orbital_state_class_idx on orbital_state (orbit_class);

-- ---------------------------------------------------------------- identifiers, provenance, quality
create table entity_identifiers (
  id bigserial primary key,
  entity_type text not null,
  entity_id text not null,
  source_id text references sources(id),
  identifier_type text not null,                 -- norad | cospar | jcat | un_registration | fcc | itu | source_native_id | launch_id
  identifier_value text not null,
  confidence double precision not null default 1.0,
  first_seen_at timestamptz not null default now(),
  last_seen_at timestamptz not null default now(),
  verified boolean not null default false,
  metadata jsonb,
  unique (entity_type, entity_id, identifier_type, identifier_value)
);
create index entity_identifiers_lookup_idx on entity_identifiers (identifier_type, identifier_value);

create table field_provenance (
  id bigserial primary key,
  entity_type text not null,
  entity_id text not null,
  field_name text not null,
  field_value text,
  source_id text not null references sources(id),
  source_record_id text,
  confidence double precision not null default 1.0,
  observed_at timestamptz not null default now(),
  selected_as_canonical boolean not null default true,
  unique (entity_type, entity_id, field_name, source_id)
);
create index field_provenance_entity_idx on field_provenance (entity_type, entity_id);

create table data_quality_flags (
  id bigserial primary key,
  entity_type text not null,
  entity_id text not null,
  flag text not null,                            -- SOURCE_CONFLICT | MISSING_ID | AMBIGUOUS_ENTITY | STALE_DATA | SUSPECT_ORBIT | UNKNOWN_OPERATOR | UNKNOWN_COUNTRY | DUPLICATE_OBJECT
  detail text,
  created_at timestamptz not null default now(),
  resolved_at timestamptz,
  unique (entity_type, entity_id, flag)
);
create index data_quality_flags_open_idx on data_quality_flags (flag) where resolved_at is null;

create table manual_review_queue (
  id bigserial primary key,
  kind text not null,                            -- possible_duplicate | unknown_owner | conflict
  entity_a_type text, entity_a_id text, entity_b_type text, entity_b_id text,
  confidence double precision,
  detail jsonb,
  status text not null default 'open',           -- open | merged | kept_separate | dismissed
  created_at timestamptz not null default now(),
  resolved_at timestamptz, resolved_by text
);

create table entity_merges (
  id bigserial primary key,
  entity_type text not null, kept_id text not null, merged_id text not null,
  reason text, performed_by text, performed_at timestamptz not null default now(), snapshot jsonb
);

-- ---------------------------------------------------------------- events
create table events (
  id text primary key,
  type text not null,
  title text not null,
  summary text,
  event_time timestamptz not null,
  detected_at timestamptz not null default now(),
  confidence double precision not null default 1.0,
  source_id text references sources(id),
  source_url text,
  dedupe_key text unique,
  metadata jsonb not null default '{}'::jsonb,
  created_at timestamptz not null default now()
);
create index events_time_idx on events (event_time desc);
create index events_type_time_idx on events (type, event_time desc);

create table event_entities (
  event_id text not null references events(id) on delete cascade,
  entity_type text not null,
  entity_id text not null,
  relationship text not null default 'subject',
  primary key (event_id, entity_type, entity_id, relationship)
);
create index event_entities_entity_idx on event_entities (entity_type, entity_id);

-- ---------------------------------------------------------------- search
create table search_index (
  entity_type text not null,
  entity_id text not null,
  slug text not null,
  title text not null,
  subtitle text,
  keywords text not null default '',
  weight double precision not null default 1.0,
  tsv tsvector,
  updated_at timestamptz not null default now(),
  primary key (entity_type, entity_id)
);
create index search_index_tsv_idx on search_index using gin (tsv);
create index search_index_title_trgm_idx on search_index using gin (title gin_trgm_ops);
create index search_index_keywords_trgm_idx on search_index using gin (keywords gin_trgm_ops);

-- ---------------------------------------------------------------- derived metrics (versioned)
create table metric_definitions (
  key text primary key, name text not null, version text not null, methodology text not null, inputs jsonb not null default '[]'::jsonb, updated_at timestamptz not null default now()
);

create table stats_snapshots (
  key text primary key,                           -- 'global' | 'orbit_buckets' | …
  computed_at timestamptz not null default now(),
  payload jsonb not null
);

-- ---------------------------------------------------------------- trending
create table page_views (
  day date not null, entity_type text not null, entity_id text not null, views integer not null default 0,
  primary key (day, entity_type, entity_id)
);

-- ---------------------------------------------------------------- materialized views
create materialized view country_stats as
select c.code, c.name, c.slug,
  count(s.id) filter (where s.object_type in ('PAYLOAD','STATION') and s.status = 'ACTIVE') as active_payloads,
  count(s.id) filter (where s.object_type in ('PAYLOAD','STATION') and s.decay_date is null) as on_orbit_payloads,
  count(s.id) filter (where s.object_type in ('PAYLOAD','STATION')) as total_payloads,
  count(s.id) filter (where s.object_type = 'DEBRIS' and s.decay_date is null) as debris_on_orbit,
  count(s.id) filter (where s.object_type = 'ROCKET_BODY' and s.decay_date is null) as rocket_bodies_on_orbit,
  count(s.id) filter (where s.decay_date is null) as objects_on_orbit,
  count(s.id) as total_objects,
  count(distinct s.launch_id) as launches,
  count(distinct s.operator_id) as operators,
  count(s.id) filter (where s.launch_date >= (current_date - interval '365 days') and s.object_type in ('PAYLOAD','STATION')) as payloads_last_365d
from countries c left join satellites s on s.country_code = c.code
group by c.code, c.name, c.slug;
create unique index country_stats_code_idx on country_stats (code);

create materialized view operator_stats as
select o.id, o.slug, o.name, o.kind, o.country_code,
  count(s.id) filter (where s.status = 'ACTIVE' and s.object_type in ('PAYLOAD','STATION')) as active_payloads,
  count(s.id) filter (where s.object_type in ('PAYLOAD','STATION') and s.decay_date is null) as on_orbit_payloads,
  count(s.id) filter (where s.object_type in ('PAYLOAD','STATION')) as total_payloads,
  count(s.id) filter (where s.decay_date is not null) as decayed,
  count(distinct s.constellation_id) as constellations,
  count(distinct s.launch_id) as launches,
  count(s.id) filter (where s.launch_date >= (current_date - interval '365 days')) as payloads_last_365d,
  min(s.launch_date) as first_launch, max(s.launch_date) as last_launch
from organizations o left join satellites s on s.operator_id = o.id
group by o.id, o.slug, o.name, o.kind, o.country_code;
create unique index operator_stats_id_idx on operator_stats (id);

create materialized view constellation_stats as
select k.id, k.slug, k.name, k.operator_id, k.country_code, k.service_type, k.orbit_class,
  count(s.id) filter (where s.status = 'ACTIVE') as active,
  count(s.id) filter (where s.status = 'INACTIVE') as inactive,
  count(s.id) filter (where s.decay_date is not null) as decayed,
  count(s.id) filter (where s.decay_date is null) as on_orbit,
  count(s.id) as total,
  count(s.id) filter (where s.launch_date >= (current_date - interval '365 days')) as launched_last_365d,
  count(s.id) filter (where s.launch_date >= (current_date - interval '30 days')) as launched_last_30d,
  count(distinct s.launch_id) as launches,
  min(s.launch_date) as first_launch, max(s.launch_date) as last_launch,
  percentile_cont(0.5) within group (order by s.perigee_km) filter (where s.status='ACTIVE') as median_perigee_km,
  percentile_cont(0.5) within group (order by s.inclination_deg) filter (where s.status='ACTIVE') as median_inclination_deg
from constellations k left join satellites s on s.constellation_id = k.id
group by k.id, k.slug, k.name, k.operator_id, k.country_code, k.service_type, k.orbit_class;
create unique index constellation_stats_id_idx on constellation_stats (id);

create materialized view launch_year_stats as
select l.launch_year as year,
  count(*) as launches,
  sum(l.payload_count) as payloads,
  count(*) filter (where l.launch_site_code is not null) as with_site
from launches l where l.launch_year is not null
group by l.launch_year;
create unique index launch_year_stats_year_idx on launch_year_stats (year);

create materialized view orbital_bucket_stats as
with b as (
  select s.id, s.object_type, s.status, s.perigee_km, s.apogee_km, s.orbit_class,
    case
      when s.orbit_class = 'GEO' then 'GEO'
      when s.orbit_class = 'MEO' then 'MEO'
      when s.orbit_class = 'HEO' then 'HEO'
      when s.perigee_km is null then 'UNKNOWN'
      when s.perigee_km < 200 then '0-200'
      when s.perigee_km < 300 then '200-300'
      when s.perigee_km < 400 then '300-400'
      when s.perigee_km < 500 then '400-500'
      when s.perigee_km < 600 then '500-600'
      when s.perigee_km < 800 then '600-800'
      when s.perigee_km < 1000 then '800-1000'
      when s.perigee_km < 2000 then '1000-2000'
      else 'OTHER' end as bucket
  from satellites s where s.decay_date is null and s.orbit_center = 'EA'
)
select bucket,
  count(*) as objects,
  count(*) filter (where object_type in ('PAYLOAD','STATION') and status = 'ACTIVE') as active_payloads,
  count(*) filter (where object_type in ('PAYLOAD','STATION')) as payloads,
  count(*) filter (where object_type = 'DEBRIS') as debris,
  count(*) filter (where object_type = 'ROCKET_BODY') as rocket_bodies
from b group by bucket;
create unique index orbital_bucket_stats_bucket_idx on orbital_bucket_stats (bucket);
"""


def upgrade() -> None:
    for stmt in _split(SQL):
        op.execute(stmt)


def downgrade() -> None:
    op.execute("drop schema public cascade; create schema public;")


def _split(sql: str) -> list[str]:
    """Split on ';' at end of line — the DDL above never contains ';' inside string literals except in comments."""
    out: list[str] = []
    buf: list[str] = []
    for line in sql.splitlines():
        stripped = line.split("--")[0].rstrip() if not line.lstrip().startswith("--") else ""
        if line.lstrip().startswith("--") and not buf:
            continue
        buf.append(line)
        if stripped.endswith(";"):
            out.append("\n".join(buf))
            buf = []
    if "".join(buf).strip():
        out.append("\n".join(buf))
    return out