"""Company Atlas — initial schema (spec §6–7, §22, §26, §70–76). Historical-first: nothing here is ever overwritten by the pipeline — observations, snapshots, changes and events are append-only; entities (jobs, people, products, plans, locations) carry first_seen / last_seen / removed_at and a status instead of being deleted. Revision ID: 0001 Revises: """ 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"; -- ============================================================================================================ reference create table if not exists industries ( slug text primary key, name text not null, parent_slug text references industries(slug), description text, keywords text[] not null default '{}', sort_order int not null default 100 ); create table if not exists countries ( code char(2) primary key, name text not null, region text, subregion text, lat double precision, lon double precision ); -- ============================================================================================================ companies create table if not exists companies ( id text primary key, slug text not null unique, legal_name text, display_name text not null, canonical_domain text not null unique, website text not null, description text, industries text[] not null default '{}', industry_primary text references industries(slug), country char(2) references countries(code), hq_city text, hq_region text, founded_year int, company_type text, public_company boolean not null default false, ticker text, exchange text, employees_band text, employees int, wikidata_id text unique, lei text, sec_cik text, logo_url text, status text not null default 'ACTIVE', onboarding_status text not null default 'pending', onboarding_error text, importance real not null default 0.2, tier smallint not null default 4, indexed boolean not null default false, source_meta jsonb not null default '{}'::jsonb, stats jsonb not null default '{}'::jsonb, discovered_at timestamptz not null default now(), first_observed_at timestamptz, last_observed_at timestamptz, last_change_at timestamptz, last_event_at timestamptz, created_at timestamptz not null default now(), updated_at timestamptz not null default now(), search tsvector generated always as ( setweight(to_tsvector('simple', coalesce(display_name, '')), 'A') || setweight(to_tsvector('simple', coalesce(legal_name, '')), 'B') || setweight(to_tsvector('simple', coalesce(canonical_domain, '')), 'B') || setweight(to_tsvector('english', coalesce(description, '')), 'C')) stored ); create index if not exists companies_search_idx on companies using gin (search); create index if not exists companies_name_trgm_idx on companies using gin (display_name gin_trgm_ops); create index if not exists companies_domain_trgm_idx on companies using gin (canonical_domain gin_trgm_ops); create index if not exists companies_country_idx on companies (country); create index if not exists companies_industry_idx on companies using gin (industries); create index if not exists companies_importance_idx on companies (importance desc); create index if not exists companies_onboarding_idx on companies (onboarding_status) where onboarding_status <> 'active'; create index if not exists companies_last_event_idx on companies (last_event_at desc nulls last); create table if not exists company_aliases ( company_id text not null references companies(id) on delete cascade, alias text not null, alias_norm text not null, kind text not null default 'alias', -- alias | legal | brand | former | ticker | native source text, primary key (company_id, alias_norm) ); create index if not exists company_aliases_norm_idx on company_aliases (alias_norm); create table if not exists domains ( id text primary key, company_id text not null references companies(id) on delete cascade, domain text not null, kind text not null default 'primary', -- primary | alias | redirect | subdomain | former first_seen_at timestamptz not null default now(), last_seen_at timestamptz not null default now(), status text not null default 'active', unique (domain, company_id) ); create index if not exists domains_domain_idx on domains (domain); create table if not exists company_relationships ( id text primary key, from_company_id text not null references companies(id) on delete cascade, to_company_id text references companies(id) on delete set null, to_name text, kind text not null, -- PARENT_OF | SUBSIDIARY_OF | ACQUIRED_BY | ACQUIRED | PARTNER_OF | COMPETITOR_OF | INVESTOR_IN | BRAND_OF valid_from date, valid_to date, first_seen_at timestamptz not null default now(), last_seen_at timestamptz not null default now(), source_url text, confidence real not null default 0.5, provenance jsonb not null default '{}'::jsonb ); create index if not exists company_relationships_from_idx on company_relationships (from_company_id, kind); create index if not exists company_relationships_to_idx on company_relationships (to_company_id, kind); -- ============================================================================================================ connectors / sensors create table if not exists connectors ( id text primary key, -- generic-html-v1, greenhouse-v1 … name text not null, version text not null, category text not null, -- surface fetch_mode text not null default 'http', supports_discovery boolean not null default false, supports_incremental boolean not null default true, default_interval_s int not null default 86400, enabled boolean not null default true, created_at timestamptz not null default now(), updated_at timestamptz not null default now(), stats jsonb not null default '{}'::jsonb ); create table if not exists sensors ( id text primary key, company_id text not null references companies(id) on delete cascade, surface text not null, connector_id text not null references connectors(id), url text not null, canonical_url text not null, domain text not null, discovery_confidence real not null default 0.5, discovery_method text, -- nav | sitemap | robots | pattern | ats | feed | manual | event quality_score real not null default 50, status text not null default 'pending', tier char(1) not null default 'D', base_interval_s int not null default 86400, current_interval_s int not null default 86400, next_run_at timestamptz not null default now(), last_run_at timestamptz, last_success_at timestamptz, last_change_at timestamptz, last_meaningful_change_at timestamptz, last_status int, last_failure_class text, last_error text, consecutive_failures int not null default 0, consecutive_unchanged int not null default 0, etag text, last_modified text, last_content_hash text, last_normalized_hash text, last_structural_hash text, last_snapshot_id text, snapshot_count int not null default 0, observation_count int not null default 0, change_count int not null default 0, meaningful_change_count int not null default 0, event_count int not null default 0, config jsonb not null default '{}'::jsonb, priority real not null default 0.5, claimed_by text, claimed_at timestamptz, created_at timestamptz not null default now(), updated_at timestamptz not null default now(), retired_at timestamptz, unique (company_id, canonical_url) ); create index if not exists sensors_due_idx on sensors (next_run_at) where status in ('active', 'failing', 'pending'); create index if not exists sensors_company_idx on sensors (company_id, surface); create index if not exists sensors_domain_idx on sensors (domain); create index if not exists sensors_status_idx on sensors (status); create index if not exists sensors_connector_idx on sensors (connector_id); create index if not exists sensors_claimed_idx on sensors (claimed_at) where claimed_by is not null; create table if not exists domain_budgets ( domain text primary key, max_concurrency int not null default 2, requests_per_minute int not null default 20, daily_budget int not null default 600, browser_budget int not null default 20, used_today int not null default 0, browser_used_today int not null default 0, budget_day date not null default current_date, crawl_delay_s real, blocked_until timestamptz, block_reason text, notes text, updated_at timestamptz not null default now() ); -- ============================================================================================================ observations / snapshots create table if not exists observations ( id text primary key, sensor_id text not null references sensors(id) on delete cascade, company_id text not null references companies(id) on delete cascade, fetched_at timestamptz not null default now(), status_code int, duration_ms int, transport text not null default 'http', final_url text, redirects int not null default 0, not_modified boolean not null default false, changed boolean not null default false, failure_class text, error text, content_hash text, normalized_hash text, structural_hash text, object_key text, size_bytes int, content_type text, connector_version text, collection_method text not null default 'live', worker text ); create index if not exists observations_sensor_idx on observations (sensor_id, fetched_at desc); create index if not exists observations_company_idx on observations (company_id, fetched_at desc); create index if not exists observations_time_idx on observations (fetched_at desc); create index if not exists observations_failure_idx on observations (failure_class, fetched_at desc) where failure_class is not null; create table if not exists snapshots ( id text primary key, sensor_id text not null references sensors(id) on delete cascade, company_id text not null references companies(id) on delete cascade, observation_id text references observations(id) on delete set null, previous_snapshot_id text, version_no int not null default 1, fetched_at timestamptz not null default now(), content_hash text not null, normalized_hash text not null, structural_hash text not null, object_key text, -- raw bytes text_key text, -- normalized text blocks_key text, -- JSON blocks extracted jsonb not null default '{}'::jsonb, -- structured fields (jobs, prices, people, locations, news, products, meta) extracted_summary jsonb not null default '{}'::jsonb, -- small counters for listings (job_count, plan_count …) title text, language text, size_bytes int, text_length int, block_count int, content_type text, connector_version text, collection_method text not null default 'live' ); create index if not exists snapshots_sensor_idx on snapshots (sensor_id, fetched_at desc); create index if not exists snapshots_company_idx on snapshots (company_id, fetched_at desc); create table if not exists changes ( id text primary key, sensor_id text not null references sensors(id) on delete cascade, company_id text not null references companies(id) on delete cascade, surface text not null, snapshot_before text references snapshots(id) on delete set null, snapshot_after text not null references snapshots(id) on delete cascade, detected_at timestamptz not null default now(), significance real not null, kind text not null, -- noise | minor | meaningful | major | critical blocks_added int not null default 0, blocks_removed int not null default 0, blocks_modified int not null default 0, blocks_moved int not null default 0, text_delta_ratio real not null default 0, similarity real, diff jsonb not null default '{}'::jsonb, -- block-level diff (bounded) structured_delta jsonb not null default '{}'::jsonb, -- typed deltas (jobs added/removed, prices, people …) status text not null default 'pending', -- pending | processed | enriched | archived processed_at timestamptz, diff_version text not null default 'diff-v1' ); create index if not exists changes_company_idx on changes (company_id, detected_at desc); create index if not exists changes_sensor_idx on changes (sensor_id, detected_at desc); create index if not exists changes_kind_idx on changes (kind, detected_at desc); create index if not exists changes_pending_idx on changes (status) where status = 'pending'; -- ============================================================================================================ events create table if not exists event_clusters ( id text primary key, company_id text not null references companies(id) on delete cascade, cluster_key text not null unique, event_type text not null, event_subtype text, title text, first_detected_at timestamptz not null default now(), last_detected_at timestamptz not null default now(), source_count int not null default 1, surfaces text[] not null default '{}', confidence real not null default 0.5, canonical_event_id text ); create table if not exists events ( id text primary key, company_id text not null references companies(id) on delete cascade, sensor_id text references sensors(id) on delete set null, change_id text references changes(id) on delete set null, cluster_id text references event_clusters(id) on delete set null, surface text, event_type text not null, event_subtype text not null, importance real not null default 0.5, confidence real not null default 0.7, confidence_label text not null default 'LIKELY', title text not null, summary text, old_value text, new_value text, payload jsonb not null default '{}'::jsonb, entities jsonb not null default '{}'::jsonb, -- {jobs:[…], people:[…], products:[…], locations:[…], amounts:[…]} tags text[] not null default '{}', detected_at timestamptz not null default now(), effective_at timestamptz, published_at timestamptz, source_url text, snapshot_before text, snapshot_after text, language text, origin text not null default 'deterministic', -- deterministic | llm | hybrid | backfill model_provider text, model_name text, model_version text, prompt_version text, schema_version text not null default 'event-v1', status text not null default 'active', retracted_reason text, dedupe_key text unique, created_at timestamptz not null default now(), search tsvector generated always as ( setweight(to_tsvector('english', coalesce(title, '')), 'A') || setweight(to_tsvector('english', coalesce(summary, '')), 'B')) stored ); create index if not exists events_company_idx on events (company_id, detected_at desc); create index if not exists events_time_idx on events (detected_at desc) where status = 'active'; create index if not exists events_type_idx on events (event_type, detected_at desc); create index if not exists events_subtype_idx on events (event_subtype, detected_at desc); create index if not exists events_importance_idx on events (importance desc, detected_at desc); create index if not exists events_search_idx on events using gin (search); create index if not exists events_cluster_idx on events (cluster_id); create index if not exists events_tags_idx on events using gin (tags); create table if not exists event_sources ( event_id text not null references events(id) on delete cascade, sensor_id text references sensors(id) on delete set null, source_url text not null, snapshot_id text, surface text, detected_at timestamptz not null default now(), kind text not null default 'primary', primary key (event_id, source_url) ); -- ============================================================================================================ extracted entities create table if not exists jobs ( id text primary key, company_id text not null references companies(id) on delete cascade, sensor_id text references sensors(id) on delete set null, external_id text, fingerprint text not null, title text not null, department text, team text, location_text text, city text, region text, country char(2), remote boolean, employment_type text, seniority text, skills text[] not null default '{}', salary_min numeric, salary_max numeric, salary_currency text, salary_period text, url text, description_hash text, posted_at timestamptz, first_seen_at timestamptz not null default now(), last_seen_at timestamptz not null default now(), removed_at timestamptz, status text not null default 'open', -- open | no_longer_listed is_ai boolean not null default false, is_engineering boolean not null default false, raw jsonb not null default '{}'::jsonb, unique (company_id, fingerprint) ); create index if not exists jobs_company_idx on jobs (company_id, status); create index if not exists jobs_first_seen_idx on jobs (first_seen_at desc); create index if not exists jobs_country_idx on jobs (country) where status = 'open'; create index if not exists jobs_ai_idx on jobs (company_id) where is_ai and status = 'open'; create index if not exists jobs_title_trgm_idx on jobs using gin (title gin_trgm_ops); create table if not exists people ( id text primary key, company_id text not null references companies(id) on delete cascade, sensor_id text references sensors(id) on delete set null, name text not null, name_norm text not null, title text, role_category text, -- ceo | cfo | cto | coo | founder | president | chair | board | vp | head | other is_executive boolean not null default false, first_seen_at timestamptz not null default now(), last_seen_at timestamptz not null default now(), removed_at timestamptz, status text not null default 'listed', -- listed | no_longer_listed source_url text, unique (company_id, name_norm) ); create index if not exists people_company_idx on people (company_id, status); create table if not exists products ( id text primary key, company_id text not null references companies(id) on delete cascade, sensor_id text references sensors(id) on delete set null, name text not null, name_norm text not null, category text, description text, url text, first_seen_at timestamptz not null default now(), last_seen_at timestamptz not null default now(), removed_at timestamptz, status text not null default 'listed', unique (company_id, name_norm) ); create index if not exists products_company_idx on products (company_id, status); create table if not exists pricing_plans ( id text primary key, company_id text not null references companies(id) on delete cascade, sensor_id text references sensors(id) on delete set null, plan_name text not null, plan_norm text not null, currency text, billing_period text, -- month | year | one_time | usage | contact price numeric, price_text text, unit text, features jsonb not null default '[]'::jsonb, contact_sales boolean not null default false, version_no int not null default 1, valid_from timestamptz not null default now(), valid_to timestamptz, first_seen_at timestamptz not null default now(), last_seen_at timestamptz not null default now(), status text not null default 'current', -- current | superseded | removed source_url text ); create index if not exists pricing_plans_company_idx on pricing_plans (company_id, status); create index if not exists pricing_plans_current_idx on pricing_plans (company_id, plan_norm) where status = 'current'; create table if not exists locations ( id text primary key, company_id text not null references companies(id) on delete cascade, sensor_id text references sensors(id) on delete set null, kind text not null default 'office', -- headquarters | office | store | factory | warehouse | lab | data_center | other name text, name_norm text not null, city text, region text, country char(2), lat double precision, lon double precision, first_seen_at timestamptz not null default now(), last_seen_at timestamptz not null default now(), removed_at timestamptz, status text not null default 'listed', source_url text, unique (company_id, name_norm) ); create index if not exists locations_company_idx on locations (company_id, status); create index if not exists locations_country_idx on locations (country) where status = 'listed'; create table if not exists news_items ( id text primary key, company_id text not null references companies(id) on delete cascade, sensor_id text references sensors(id) on delete set null, url text not null, canonical_url text not null, title text not null, summary text, category text, -- press | blog | changelog | research | ir | other published_at timestamptz, first_seen_at timestamptz not null default now(), language text, entities jsonb not null default '{}'::jsonb, unique (company_id, canonical_url) ); create index if not exists news_items_company_idx on news_items (company_id, coalesce(published_at, first_seen_at) desc); create index if not exists news_items_time_idx on news_items (first_seen_at desc); -- ============================================================================================================ metrics create table if not exists metrics_current ( company_id text not null references companies(id) on delete cascade, metric text not null, value double precision not null, confidence real not null default 0.5, inputs jsonb not null default '{}'::jsonb, formula_version text not null, computed_at timestamptz not null default now(), primary key (company_id, metric) ); create index if not exists metrics_current_metric_idx on metrics_current (metric, value desc); create table if not exists metric_series ( company_id text not null references companies(id) on delete cascade, metric text not null, day date not null, value double precision not null, confidence real not null default 0.5, formula_version text not null, primary key (company_id, metric, day) ); create index if not exists metric_series_metric_day_idx on metric_series (metric, day desc); create table if not exists company_daily ( company_id text not null references companies(id) on delete cascade, day date not null, observations int not null default 0, changes int not null default 0, meaningful_changes int not null default 0, events int not null default 0, events_by_type jsonb not null default '{}'::jsonb, jobs_open int, jobs_new int not null default 0, jobs_removed int not null default 0, jobs_ai_open int, news_items int not null default 0, sensors_active int, primary key (company_id, day) ); create index if not exists company_daily_day_idx on company_daily (day desc); create table if not exists baselines ( company_id text not null references companies(id) on delete cascade, metric text not null, mean double precision not null, stddev double precision not null, samples int not null, window_days int not null, computed_at timestamptz not null default now(), primary key (company_id, metric) ); create table if not exists global_daily ( day date primary key, companies_active int not null default 0, sensors_active int not null default 0, observations int not null default 0, changes int not null default 0, meaningful_changes int not null default 0, events int not null default 0, events_by_type jsonb not null default '{}'::jsonb, jobs_open int, jobs_new int not null default 0, jobs_removed int not null default 0, activity_index double precision, -- coverage-normalised Global Corporate Activity Index (100 = baseline) by_country jsonb not null default '{}'::jsonb, by_industry jsonb not null default '{}'::jsonb, computed_at timestamptz not null default now() ); create table if not exists signals ( id text primary key, company_id text references companies(id) on delete cascade, scope text not null default 'company', -- company | industry | country | global scope_key text, kind text not null, -- hiring_surge | hiring_freeze | launch_buildup | expansion | pricing_migration | developer_push | enterprise_repositioning | ai_acceleration | abnormal_activity strength real not null, confidence real not null, title text not null, explanation text, evidence jsonb not null default '{}'::jsonb, window_days int not null default 30, detected_at timestamptz not null default now(), expires_at timestamptz, status text not null default 'active' ); create index if not exists signals_company_idx on signals (company_id, detected_at desc); create index if not exists signals_scope_idx on signals (scope, scope_key, detected_at desc); create table if not exists trends ( term text not null, day date not null, mentions int not null default 0, companies int not null default 0, primary key (term, day) ); -- ============================================================================================================ operations create table if not exists queue_jobs ( id text primary key, kind text not null, -- discover | run_sensor | enrich_change | recompute_metrics | digest | repair_sensor key text not null unique, -- idempotency (sensor_id + window …) payload jsonb not null default '{}'::jsonb, priority real not null default 0.5, run_at timestamptz not null default now(), locked_at timestamptz, locked_by text, attempts int not null default 0, max_attempts int not null default 3, status text not null default 'pending', -- pending | running | done | failed | dead last_error text, created_at timestamptz not null default now(), finished_at timestamptz ); create index if not exists queue_jobs_due_idx on queue_jobs (kind, priority desc, run_at) where status = 'pending'; create index if not exists queue_jobs_status_idx on queue_jobs (status, kind); create table if not exists llm_jobs ( id text primary key, kind text not null, -- classify_change | summarize_event | extract_entities | classify_industry ref_id text not null, company_id text references companies(id) on delete cascade, model text, prompt_version text, status text not null default 'pending', attempts int not null default 0, request_tokens int, response_tokens int, latency_ms int, result jsonb, error text, created_at timestamptz not null default now(), started_at timestamptz, finished_at timestamptz ); create index if not exists llm_jobs_status_idx on llm_jobs (status, created_at); create index if not exists llm_jobs_ref_idx on llm_jobs (ref_id); create table if not exists failures ( id text primary key, sensor_id text references sensors(id) on delete cascade, company_id text references companies(id) on delete cascade, at timestamptz not null default now(), failure_class text not null, status_code int, message text, url text ); create index if not exists failures_time_idx on failures (at desc); create index if not exists failures_sensor_idx on failures (sensor_id, at desc); create table if not exists crawl_runs ( id text primary key, kind text not null, -- scheduler_tick | onboarding | metrics | daily | backup | repair worker text, started_at timestamptz not null default now(), finished_at timestamptz, stats jsonb not null default '{}'::jsonb, error text ); create index if not exists crawl_runs_kind_idx on crawl_runs (kind, started_at desc); create table if not exists review_queue ( id text primary key, kind text not null, -- company_merge | major_event | low_confidence | sensor_migration | legal_sensitive | unexpected_activity | blocked_source ref_id text, company_id text references companies(id) on delete cascade, payload jsonb not null default '{}'::jsonb, status text not null default 'open', -- open | accepted | rejected | resolved resolution text, created_at timestamptz not null default now(), resolved_at timestamptz ); create index if not exists review_queue_open_idx on review_queue (kind, created_at) where status = 'open'; create table if not exists cost_ledger ( day date not null, dimension text not null, -- fetch | browser | llm | storage_gb | event key text not null default '', units double precision not null default 0, cost_estimate double precision not null default 0, primary key (day, dimension, key) ); -- ============================================================================================================ users-light (no accounts required) create table if not exists owners ( token_hash text primary key, created_at timestamptz not null default now(), last_seen_at timestamptz not null default now(), email text, plan text not null default 'free' ); create table if not exists watchlists ( id text primary key, owner_hash text not null references owners(token_hash) on delete cascade, name text not null default 'My Watchlist', created_at timestamptz not null default now() ); create table if not exists watchlist_items ( watchlist_id text not null references watchlists(id) on delete cascade, company_id text not null references companies(id) on delete cascade, added_at timestamptz not null default now(), primary key (watchlist_id, company_id) ); create table if not exists alerts ( id text primary key, owner_hash text not null references owners(token_hash) on delete cascade, company_id text references companies(id) on delete cascade, name text not null, condition jsonb not null, -- {event_types:[…], min_importance, metrics:{activity_score:{gt:80}}, industries, countries} channel text not null default 'web', -- web | email | webhook target text, enabled boolean not null default true, created_at timestamptz not null default now(), last_fired_at timestamptz ); create table if not exists alert_deliveries ( id text primary key, alert_id text not null references alerts(id) on delete cascade, event_id text references events(id) on delete cascade, delivered_at timestamptz not null default now(), channel text not null, status text not null default 'queued', detail text ); create index if not exists alert_deliveries_alert_idx on alert_deliveries (alert_id, delivered_at desc); create table if not exists api_keys ( id text primary key, key_hash text not null unique, prefix text not null, name text not null, tier text not null default 'authenticated', -- anonymous | authenticated | paid | internal owner_hash text references owners(token_hash) on delete set null, created_at timestamptz not null default now(), last_used_at timestamptz, request_count bigint not null default 0, revoked_at timestamptz ); create table if not exists settings_kv ( key text primary key, value jsonb not null, updated_at timestamptz not null default now() ); insert into settings_kv (key, value) values ('dataset_started_at', to_jsonb(now())) on conflict (key) do nothing; """ def upgrade() -> None: for stmt in _split(SQL): op.execute(stmt) def downgrade() -> None: raise RuntimeError("forward-only migrations: historical data is never dropped") def _split(sql: str) -> list[str]: out: list[str] = [] buf: list[str] = [] for line in sql.splitlines(): buf.append(line) if line.rstrip().endswith(";"): stmt = "\n".join(buf).strip() body = "\n".join(x for x in stmt.splitlines() if not x.strip().startswith("--")).strip() if body and body != ";": out.append(stmt) buf = [] return out