-- Buduje odchudzona kopie positions obok tabeli produkcyjnej.
-- Ten etap nie zatrzymuje ingestu. Koncowa podmiana jest w osobnym skrypcie.
-- Uruchomienie:
--   psql "$DATABASE_URL" -v positions_rebuild_confirm=YES \
--     -f scripts/rebuild-positions-shadow.sql

\set ON_ERROR_STOP on

\if :{?positions_rebuild_confirm}
\else
\set positions_rebuild_confirm 'NO'
\endif

SELECT upper(:'positions_rebuild_confirm') = 'YES' AS rebuild_confirmed \gset
\if :rebuild_confirmed
\else
\echo 'PRZERWANO: wymagane -v positions_rebuild_confirm=YES.'
\quit
\endif


DO $preflight$
BEGIN
  IF to_regclass('public.positions') IS NULL THEN
    RAISE EXCEPTION 'Brak public.positions';
  END IF;
  IF to_regclass('public.positions_compact_build') IS NOT NULL THEN
    RAISE EXCEPTION 'public.positions_compact_build juz istnieje';
  END IF;
  IF to_regclass('public.positions_uncompacted_backup') IS NOT NULL THEN
    RAISE EXCEPTION 'public.positions_uncompacted_backup juz istnieje';
  END IF;
END
$preflight$;

CREATE TABLE IF NOT EXISTS public.tracker_positions_rebuild_state (
  singleton boolean PRIMARY KEY DEFAULT TRUE CHECK (singleton),
  cutoff_id bigint NOT NULL,
  raw_payload_cutoff timestamptz NOT NULL,
  source_count bigint NOT NULL,
  compact_count bigint,
  status text NOT NULL CHECK (status IN ('building', 'ready', 'complete')),
  started_at timestamptz NOT NULL DEFAULT clock_timestamp(),
  completed_at timestamptz
);

TRUNCATE public.tracker_positions_rebuild_state;

INSERT INTO public.tracker_positions_rebuild_state (
  singleton,
  cutoff_id,
  raw_payload_cutoff,
  source_count,
  status
)
SELECT
  TRUE,
  COALESCE(MAX(id), 0),
  clock_timestamp() - interval '7 days',
  COUNT(*),
  'building'
FROM public.positions;

CREATE TABLE public.positions_compact_build (
  LIKE public.positions
    INCLUDING DEFAULTS
    INCLUDING STORAGE
    INCLUDING COMMENTS
);

ALTER TABLE public.positions_compact_build
  ALTER COLUMN raw_payload DROP NOT NULL;

WITH rebuild AS MATERIALIZED (
  SELECT cutoff_id, raw_payload_cutoff
  FROM public.tracker_positions_rebuild_state
  WHERE singleton
),
ordered AS MATERIALIZED (
  SELECT
    p.*,
    (
      p.ignition IS FALSE
      AND COALESCE(p.speed, 0) <= 0
    ) AS is_stationary,
    LAG(p.ignition IS FALSE AND COALESCE(p.speed, 0) <= 0) OVER timeline
      AS previous_is_stationary,
    LEAD(p.ignition IS FALSE AND COALESCE(p.speed, 0) <= 0) OVER timeline
      AS next_is_stationary,
    LAG(p.geom) OVER timeline AS previous_geom,
    LEAD(p.geom) OVER timeline AS next_geom,
    LAG(p.ignition) OVER timeline AS previous_ignition,
    LEAD(p.ignition) OVER timeline AS next_ignition,
    LAG(p.gps_valid) OVER timeline AS previous_gps_valid,
    LEAD(p.gps_valid) OVER timeline AS next_gps_valid,
    LAG(p.charging) OVER timeline AS previous_charging,
    LEAD(p.charging) OVER timeline AS next_charging,
    LAG(p.defense) OVER timeline AS previous_defense,
    LEAD(p.defense) OVER timeline AS next_defense,
    LAG(p.position_type) OVER timeline AS previous_position_type,
    LEAD(p.position_type) OVER timeline AS next_position_type,
    ROW_NUMBER() OVER (
      PARTITION BY
        p.device_id,
        (p.ignition IS FALSE AND COALESCE(p.speed, 0) <= 0),
        FLOOR(EXTRACT(EPOCH FROM p.server_time) / 900)
      ORDER BY p.server_time, p.id
    ) AS rank_in_15_minute_bucket
  FROM public.positions p
  CROSS JOIN rebuild
  WHERE p.id <= rebuild.cutoff_id
  WINDOW timeline AS (
    PARTITION BY p.device_id
    ORDER BY p.server_time, p.id
  )
),
kept AS MATERIALIZED (
  SELECT ordered.*
  FROM ordered
  WHERE
    NOT is_stationary
    OR NOT COALESCE(previous_is_stationary, FALSE)
    OR NOT COALESCE(next_is_stationary, FALSE)
    OR rank_in_15_minute_bucket = 1
    OR (
      previous_is_stationary
      AND previous_geom IS NOT NULL
      AND ST_Distance(geom, previous_geom) >= 50
    )
    OR (
      next_is_stationary
      AND next_geom IS NOT NULL
      AND ST_Distance(geom, next_geom) >= 50
    )
    OR ignition IS DISTINCT FROM previous_ignition
    OR ignition IS DISTINCT FROM next_ignition
    OR gps_valid IS DISTINCT FROM previous_gps_valid
    OR gps_valid IS DISTINCT FROM next_gps_valid
    OR charging IS DISTINCT FROM previous_charging
    OR charging IS DISTINCT FROM next_charging
    OR defense IS DISTINCT FROM previous_defense
    OR defense IS DISTINCT FROM next_defense
    OR position_type IS DISTINCT FROM previous_position_type
    OR position_type IS DISTINCT FROM next_position_type
)
INSERT INTO public.positions_compact_build (
  id,
  device_id,
  vehicle_id,
  imei,
  protocol_name,
  raw_payload,
  device_time,
  server_time,
  lat,
  lon,
  speed,
  heading,
  ignition,
  battery,
  gsm_signal,
  satellite_count,
  gps_valid,
  charging,
  defense,
  position_type,
  geom
)
SELECT
  id,
  device_id,
  vehicle_id,
  imei,
  protocol_name,
  CASE WHEN server_time >= rebuild.raw_payload_cutoff THEN raw_payload ELSE NULL END,
  device_time,
  server_time,
  lat,
  lon,
  speed,
  heading,
  ignition,
  battery,
  gsm_signal,
  satellite_count,
  gps_valid,
  charging,
  defense,
  position_type,
  geom
FROM kept
CROSS JOIN rebuild
ORDER BY id;

ALTER TABLE public.positions_compact_build
  ADD CONSTRAINT positions_compact_build_pkey PRIMARY KEY (id),
  ADD CONSTRAINT positions_compact_build_device_id_fkey
    FOREIGN KEY (device_id) REFERENCES public.devices(id) ON DELETE CASCADE,
  ADD CONSTRAINT positions_compact_build_vehicle_id_fkey
    FOREIGN KEY (vehicle_id) REFERENCES public.vehicles(id) ON DELETE SET NULL;

CREATE INDEX idx_positions_vehicle_time_compact
  ON public.positions_compact_build(vehicle_id, server_time DESC);

CREATE INDEX idx_positions_device_time_compact
  ON public.positions_compact_build(device_id, server_time DESC);

CREATE INDEX idx_positions_raw_payload_retention_compact
  ON public.positions_compact_build(server_time)
  WHERE raw_payload IS NOT NULL;

ANALYZE public.positions_compact_build;

UPDATE public.tracker_positions_rebuild_state
SET
  compact_count = (SELECT COUNT(*) FROM public.positions_compact_build),
  status = 'ready'
WHERE singleton;

DO $validate$
DECLARE
  source_moving bigint;
  compact_moving bigint;
  source_overspeed bigint;
  compact_overspeed bigint;
  missing_boundaries bigint;
  expired_payloads bigint;
BEGIN
  SELECT COUNT(*) INTO source_moving
  FROM public.positions p
  CROSS JOIN public.tracker_positions_rebuild_state state
  WHERE state.singleton
    AND p.id <= state.cutoff_id
    AND NOT (p.ignition IS FALSE AND COALESCE(p.speed, 0) <= 0);

  SELECT COUNT(*) INTO compact_moving
  FROM public.positions_compact_build
  WHERE NOT (ignition IS FALSE AND COALESCE(speed, 0) <= 0);

  IF source_moving <> compact_moving THEN
    RAISE EXCEPTION 'Niezgodna liczba punktow jazdy: source %, compact %',
      source_moving, compact_moving;
  END IF;

  SELECT COUNT(*) INTO source_overspeed
  FROM public.positions p
  CROSS JOIN public.tracker_positions_rebuild_state state
  WHERE state.singleton AND p.id <= state.cutoff_id AND p.speed > 140;

  SELECT COUNT(*) INTO compact_overspeed
  FROM public.positions_compact_build
  WHERE speed > 140;

  IF source_overspeed <> compact_overspeed THEN
    RAISE EXCEPTION 'Niezgodna liczba przekroczen: source %, compact %',
      source_overspeed, compact_overspeed;
  END IF;

  SELECT COUNT(*) INTO missing_boundaries
  FROM (
    SELECT device_id, MIN(server_time) first_time, MAX(server_time) last_time
    FROM public.positions p
    CROSS JOIN public.tracker_positions_rebuild_state state
    WHERE state.singleton AND p.id <= state.cutoff_id
    GROUP BY device_id
    EXCEPT
    SELECT device_id, MIN(server_time), MAX(server_time)
    FROM public.positions_compact_build
    GROUP BY device_id
  ) differences;

  IF missing_boundaries <> 0 THEN
    RAISE EXCEPTION 'Niezgodne granice historii dla % urzadzen', missing_boundaries;
  END IF;

  SELECT COUNT(*) INTO expired_payloads
  FROM public.positions_compact_build p
  CROSS JOIN public.tracker_positions_rebuild_state state
  WHERE state.singleton
    AND p.server_time < state.raw_payload_cutoff
    AND p.raw_payload IS NOT NULL;

  IF expired_payloads <> 0 THEN
    RAISE EXCEPTION 'Pozostaly przeterminowane raw_payload: %', expired_payloads;
  END IF;
END
$validate$;

SELECT
  source_count,
  compact_count,
  ROUND(100.0 * compact_count / NULLIF(source_count, 0), 2) AS retained_percent,
  cutoff_id,
  raw_payload_cutoff,
  status,
  pg_size_pretty(pg_total_relation_size('public.positions')) AS old_total,
  pg_size_pretty(pg_total_relation_size('public.positions_compact_build')) AS compact_total
FROM public.tracker_positions_rebuild_state
WHERE singleton;

