-- Market Atlas — initial schema (PostgreSQL 17)
-- Metadata (instruments, exchanges, sources, connectors, events) + time-series (observations partitioned by day, bars).

create table if not exists schema_migrations (
  version text primary key,
  applied_at timestamptz not null default now()
);

create table if not exists countries (
  code text primary key,
  name text not null,
  region text not null,
  currency text
);

create table if not exists exchanges (
  id text primary key,
  mic text,
  name text not null,
  operator text,
  country text not null references countries(code),
  city text,
  timezone text not null,
  currency text,
  website text,
  sessions jsonb not null default '{"regular":[]}'::jsonb,
  lat double precision,
  lon double precision,
  asset_classes text[] not null default '{}',
  created_at timestamptz not null default now(),
  updated_at timestamptz not null default now()
);
create index if not exists exchanges_country_idx on exchanges(country);

create table if not exists exchange_holidays (
  exchange_id text not null references exchanges(id) on delete cascade,
  date date not null,
  name text not null,
  kind text not null default 'CLOSED',
  close_time text,
  source_id text,
  primary key (exchange_id, date)
);

create table if not exists companies (
  id text primary key,
  name text not null,
  cik text,
  country text,
  sector text,
  industry text,
  website text,
  created_at timestamptz not null default now(),
  updated_at timestamptz not null default now()
);
create index if not exists companies_cik_idx on companies(cik);

create table if not exists instruments (
  id text primary key,
  symbol text not null,
  name text not null,
  asset_class text not null,
  exchange_id text references exchanges(id),
  mic text,
  currency text,
  country text,
  company_id text references companies(id),
  isin text,
  security_type text,
  base text,
  quote text,
  is_active boolean not null default true,
  metadata jsonb not null default '{}'::jsonb,
  search_text text generated always as (lower(symbol || ' ' || name)) stored,
  created_at timestamptz not null default now(),
  updated_at timestamptz not null default now()
);
create index if not exists instruments_symbol_idx on instruments(upper(symbol));
create index if not exists instruments_class_idx on instruments(asset_class);
create index if not exists instruments_exchange_idx on instruments(exchange_id);
create index if not exists instruments_country_idx on instruments(country);
create index if not exists instruments_company_idx on instruments(company_id);
create index if not exists instruments_search_idx on instruments using gin (to_tsvector('simple', search_text));

create table if not exists symbol_aliases (
  alias text not null,
  source_id text not null default '*',
  instrument_id text not null references instruments(id) on delete cascade,
  primary key (alias, source_id)
);
create index if not exists symbol_aliases_instrument_idx on symbol_aliases(instrument_id);

create table if not exists sources (
  id text primary key,
  name text not null,
  organization text,
  source_type text not null,
  homepage text,
  jurisdiction text,
  rights_status text not null,
  realtime_status text not null,
  family text,
  category text not null default 'PUBLIC_MARKET_SOURCE',
  enabled boolean not null default true,
  created_at timestamptz not null default now(),
  updated_at timestamptz not null default now()
);

create table if not exists connectors (
  id text primary key,
  source_id text not null references sources(id),
  name text not null,
  version text not null,
  source_type text not null,
  rights_status text not null,
  realtime_status text not null,
  enabled boolean not null default true,
  paused boolean not null default false,
  metadata jsonb not null default '{}'::jsonb,
  state jsonb not null default '{}'::jsonb,           -- connector-private persisted state (cursors, seen ids)
  schema_fingerprints jsonb not null default '{}'::jsonb, -- kind -> [fingerprints]
  status text not null default 'STARTING',
  last_message_at timestamptz,
  last_success_at timestamptz,
  last_error_at timestamptz,
  last_error text,
  messages_total bigint not null default 0,
  errors_total bigint not null default 0,
  reconnects bigint not null default 0,
  reliability_score double precision,
  created_at timestamptz not null default now(),
  updated_at timestamptz not null default now()
);

create table if not exists connector_versions (
  connector_id text not null references connectors(id) on delete cascade,
  version text not null,
  first_seen_at timestamptz not null default now(),
  metadata jsonb not null default '{}'::jsonb,
  primary key (connector_id, version)
);

create table if not exists connector_schema_changes (
  id bigserial primary key,
  connector_id text not null references connectors(id) on delete cascade,
  kind text not null,
  old_fingerprint text,
  new_fingerprint text not null,
  sample jsonb,
  detected_at timestamptz not null default now(),
  acknowledged boolean not null default false
);
create index if not exists connector_schema_changes_idx on connector_schema_changes(connector_id, detected_at desc);

-- Time series of connector health snapshots (1 per connector per minute).
create table if not exists connector_health (
  ts timestamptz not null,
  connector_id text not null,
  status text not null,
  messages_1m integer not null default 0,
  errors_1m integer not null default 0,
  median_latency_ms integer,
  p95_latency_ms integer,
  parse_success_rate double precision,
  instruments_covered integer,
  reconnects integer not null default 0,
  reliability_score double precision,
  primary key (connector_id, ts)
);
create index if not exists connector_health_ts_idx on connector_health(ts);

-- L3 observations, partitioned by received day. Values are kept at source precision.
create table if not exists observations (
  received_at timestamptz not null,
  source_ts timestamptz,
  instrument_id text not null,
  field text not null,
  value double precision not null,
  currency text,
  source_id text not null,
  connector_id text not null,
  rights_status text not null,
  realtime_status text not null,
  timestamp_trust text not null,
  confidence real,
  latency_ms bigint,
  sequence text,
  fingerprint text not null,
  raw_ref text,
  normalizer_version text not null,
  meta jsonb
) partition by range (received_at);
create index if not exists observations_instr_idx on observations(instrument_id, field, received_at desc);
create index if not exists observations_source_idx on observations(source_id, received_at desc);
create index if not exists observations_fp_idx on observations(fingerprint);

create table if not exists canonical_quotes (
  instrument_id text primary key references instruments(id) on delete cascade,
  symbol text not null,
  price double precision,
  open double precision,
  high double precision,
  low double precision,
  previous_close double precision,
  change double precision,
  change_percent double precision,
  volume double precision,
  bid double precision,
  ask double precision,
  currency text,
  source_count integer not null default 0,
  dispersion_bps double precision,
  confidence double precision not null default 0,
  freshness_ms bigint,
  realtime_status text not null default 'UNKNOWN',
  rights_status text not null default 'UNKNOWN',
  updated_at timestamptz not null,
  source_ts timestamptz,
  session_high double precision,
  session_low double precision,
  contributions jsonb not null default '[]'::jsonb,
  consensus_version text not null
);
create index if not exists canonical_quotes_updated_idx on canonical_quotes(updated_at desc);
create index if not exists canonical_quotes_change_idx on canonical_quotes(change_percent);

create table if not exists bars (
  instrument_id text not null,
  resolution text not null,
  ts timestamptz not null,
  open double precision not null,
  high double precision not null,
  low double precision not null,
  close double precision not null,
  volume double precision,
  source_count integer not null default 1,
  producer text not null,
  version text not null,
  primary key (instrument_id, resolution, ts)
);
create index if not exists bars_res_ts_idx on bars(resolution, ts desc);

create table if not exists market_events (
  id text primary key,
  type text not null,
  ts timestamptz not null,
  instrument_ids text[] not null default '{}',
  entity_ids text[] not null default '{}',
  severity text not null,
  confidence double precision not null,
  source_count integer not null default 1,
  sources text[] not null default '{}',
  title text not null,
  summary text,
  data jsonb not null default '{}'::jsonb,
  fingerprint text not null unique,
  supporting_observations text[] not null default '{}',
  first_seen_at timestamptz not null default now(),
  confirmed_at timestamptz
);
create index if not exists market_events_ts_idx on market_events(ts desc);
create index if not exists market_events_type_idx on market_events(type, ts desc);
create index if not exists market_events_instr_idx on market_events using gin (instrument_ids);

create table if not exists filings (
  id text primary key,
  source_id text not null,
  cik text,
  company_name text not null,
  form_type text not null,
  filed_at timestamptz not null,
  url text not null,
  instrument_ids text[] not null default '{}',
  metadata jsonb not null default '{}'::jsonb,
  created_at timestamptz not null default now()
);
create index if not exists filings_filed_idx on filings(filed_at desc);
create index if not exists filings_cik_idx on filings(cik);
create index if not exists filings_form_idx on filings(form_type);

create table if not exists corporate_actions (
  id text primary key,
  instrument_id text references instruments(id),
  company_id text references companies(id),
  type text not null,
  announced_at timestamptz,
  ex_date date,
  effective_date date,
  payable_date date,
  value double precision,
  currency text,
  ratio text,
  source_id text not null,
  url text,
  metadata jsonb not null default '{}'::jsonb,
  created_at timestamptz not null default now()
);

-- Slowly-changing facts diff log (market cap, shares, profile…).
create table if not exists fact_changes (
  id bigserial primary key,
  entity_type text not null,
  entity_id text not null,
  field text not null,
  old_value jsonb,
  new_value jsonb,
  detected_at timestamptz not null default now(),
  effective_at timestamptz,
  source_id text
);
create index if not exists fact_changes_entity_idx on fact_changes(entity_type, entity_id, detected_at desc);

-- Content fingerprints for change detection on documents/pages.
create table if not exists document_snapshots (
  connector_id text not null,
  url text not null,
  document_hash text not null,
  section_hashes jsonb not null default '{}'::jsonb,
  fetched_at timestamptz not null default now(),
  raw_ref text,
  primary key (connector_id, url)
);

create table if not exists system_counters (
  key text primary key,
  value bigint not null default 0,
  updated_at timestamptz not null default now()
);

create table if not exists observation_archives (
  partition_name text primary key,
  day date not null,
  rows_archived bigint not null,
  path text not null,
  bytes bigint not null,
  archived_at timestamptz not null default now()
);

-- Daily counts used by the homepage telemetry (observations today, etc.).
create table if not exists daily_stats (
  day date primary key,
  observations bigint not null default 0,
  events bigint not null default 0,
  filings bigint not null default 0,
  updated_at timestamptz not null default now()
);