Task S4T11: Snapshot Functions for Waypoints & Relations¶

-- Run snapshots in order
SELECT stats.fn_snapshot_waypoints();
SELECT stats.fn_snapshot_cache_visits();
SELECT stats.fn_snapshot_relations();
SELECT stats.fn_snapshot_relationship_tables();

-- Check how many rows were written
SELECT * FROM stats.job_log
WHERE job_name LIKE 'fn_snapshot_%'
ORDER BY completed_at DESC
LIMIT 10;

-- Verify symmetry after relations snapshot
SELECT COUNT(*) AS asymmetric_pairs
FROM stats.user_related_users a
WHERE NOT EXISTS (
  SELECT 1 FROM stats.user_related_users b
  WHERE b.user_id = a.related_user_id
    AND b.related_user_id = a.user_id
);
-- Must return 0

CREATE OR REPLACE FUNCTION stats.fn_snapshot_waypoints()
  RETURNS BIGINT
  LANGUAGE plpgsql
  SECURITY DEFINER
AS $$
DECLARE
  v_rows BIGINT;
  v_started_at TIMESTAMPTZ := clock_timestamp();
BEGIN
  INSERT INTO stats.waypoints (waypoint_code, source, lat, lon, country, first_seen_at)
  SELECT
    UPPER(m.waypoint) AS waypoint_code,
    'UK' AS source,
    MAX(m.lat) FILTER (WHERE m.lat IS NOT NULL) AS lat,
    MAX(m.lon) FILTER (WHERE m.lon IS NOT NULL) AS lon,
    MAX(UPPER(m.country)) FILTER (WHERE m.country IS NOT NULL AND BTRIM(m.country) <> '') AS country,
    MIN(m.moved_on_datetime) AS first_seen_at
  FROM geokrety.gk_moves m
  WHERE m.waypoint IS NOT NULL
    AND m.move_type <> 2
  GROUP BY UPPER(m.waypoint)
  ON CONFLICT (waypoint_code) DO UPDATE SET
    lat = COALESCE(stats.waypoints.lat, EXCLUDED.lat),
    lon = COALESCE(stats.waypoints.lon, EXCLUDED.lon),
    country = COALESCE(stats.waypoints.country, EXCLUDED.country),
    first_seen_at = LEAST(stats.waypoints.first_seen_at, EXCLUDED.first_seen_at);

  GET DIAGNOSTICS v_rows = ROW_COUNT;

  INSERT INTO stats.job_log (job_name, status, metadata, started_at, completed_at)
  VALUES (
    'fn_snapshot_waypoints',
    'completed',
    jsonb_build_object('rows_affected', v_rows),
    v_started_at,
    clock_timestamp()
  );

  RETURN v_rows;
END;
$$;

CREATE OR REPLACE FUNCTION stats.fn_snapshot_cache_visits()
  RETURNS BIGINT
  LANGUAGE plpgsql
  SECURITY DEFINER
AS $$
DECLARE
  v_rows BIGINT := 0;
  v_partial BIGINT := 0;
  v_started_at TIMESTAMPTZ := clock_timestamp();
BEGIN
  TRUNCATE TABLE stats.gk_cache_visits, stats.user_cache_visits;

  INSERT INTO stats.gk_cache_visits (
    gk_id, waypoint_id, visit_count, first_visited_at, last_visited_at
  )
  SELECT
    m.geokret_id,
    w.id,
    COUNT(*)::BIGINT,
    MIN(m.moved_on_datetime),
    MAX(m.moved_on_datetime)
  FROM geokrety.gk_moves m
  JOIN stats.waypoints w ON w.waypoint_code = UPPER(m.waypoint)
  WHERE m.waypoint IS NOT NULL
    AND m.move_type <> 2
  GROUP BY m.geokret_id, w.id;

  GET DIAGNOSTICS v_partial = ROW_COUNT;
  v_rows := v_rows + v_partial;

  INSERT INTO stats.user_cache_visits (
    user_id, waypoint_id, visit_count, first_visited_at, last_visited_at
  )
  SELECT
    m.author,
    w.id,
    COUNT(*)::BIGINT,
    MIN(m.moved_on_datetime),
    MAX(m.moved_on_datetime)
  FROM geokrety.gk_moves m
  JOIN stats.waypoints w ON w.waypoint_code = UPPER(m.waypoint)
  WHERE m.waypoint IS NOT NULL
    AND m.author IS NOT NULL
    AND m.move_type <> 2
  GROUP BY m.author, w.id;

  GET DIAGNOSTICS v_partial = ROW_COUNT;
  v_rows := v_rows + v_partial;

  INSERT INTO stats.job_log (job_name, status, metadata, started_at, completed_at)
  VALUES (
    'fn_snapshot_cache_visits',
    'completed',
    jsonb_build_object('rows_affected', v_rows),
    v_started_at,
    clock_timestamp()
  );

  RETURN v_rows;
END;
$$;

CREATE OR REPLACE FUNCTION stats.fn_snapshot_relations()
  RETURNS BIGINT
  LANGUAGE plpgsql
  SECURITY DEFINER
AS $$
DECLARE
  v_rows BIGINT := 0;
  v_partial BIGINT := 0;
  v_started_at TIMESTAMPTZ := clock_timestamp();
BEGIN
  TRUNCATE TABLE stats.gk_related_users, stats.user_related_users;

  INSERT INTO stats.gk_related_users (
    geokrety_id, user_id, interaction_count, first_interaction, last_interaction
  )
  SELECT
    m.geokret_id,
    m.author,
    COUNT(*)::BIGINT,
    MIN(m.moved_on_datetime),
    MAX(m.moved_on_datetime)
  FROM geokrety.gk_moves m
  WHERE m.author IS NOT NULL
    AND m.move_type IN (0, 1, 3, 5)
  GROUP BY m.geokret_id, m.author;

  GET DIAGNOSTICS v_partial = ROW_COUNT;
  v_rows := v_rows + v_partial;

  INSERT INTO stats.user_related_users (
    user_id, related_user_id, shared_geokrety_count, first_seen_at, last_seen_at
  )
  SELECT
    a.user_id,
    b.user_id AS related_user_id,
    COUNT(DISTINCT a.geokrety_id)::BIGINT,
    MIN(LEAST(a.first_interaction, b.first_interaction)),
    MAX(GREATEST(a.last_interaction, b.last_interaction))
  FROM stats.gk_related_users a
  JOIN stats.gk_related_users b
    ON a.geokrety_id = b.geokrety_id
   AND a.user_id <> b.user_id
  GROUP BY a.user_id, b.user_id;

  GET DIAGNOSTICS v_partial = ROW_COUNT;
  v_rows := v_rows + v_partial;

  INSERT INTO stats.job_log (job_name, status, metadata, started_at, completed_at)
  VALUES (
    'fn_snapshot_relations',
    'completed',
    jsonb_build_object('rows_affected', v_rows),
    v_started_at,
    clock_timestamp()
  );

  RETURN v_rows;
END;
$$;

CREATE OR REPLACE FUNCTION stats.fn_snapshot_relationship_tables(p_period daterange DEFAULT NULL)
  RETURNS BIGINT
  LANGUAGE plpgsql
  SECURITY DEFINER
AS $$
DECLARE
  v_rows BIGINT := 0;
  v_started_at TIMESTAMPTZ := clock_timestamp();
BEGIN
  -- Sprint 4 canonizes a thin facade: the wrapper keeps the stable master-spec
  -- entry point while delegating to the promoted concrete helpers.
  v_rows := v_rows + stats.fn_snapshot_waypoints();
  v_rows := v_rows + stats.fn_snapshot_cache_visits();
  v_rows := v_rows + stats.fn_snapshot_relations();

  INSERT INTO stats.job_log (job_name, status, metadata, started_at, completed_at)
  VALUES (
    'fn_snapshot_relationship_tables',
    'completed',
    jsonb_build_object(
      'rows_affected', v_rows,
      'requested_period', p_period,
      'delegated_helpers', jsonb_build_array(
        'fn_snapshot_waypoints',
        'fn_snapshot_cache_visits',
        'fn_snapshot_relations'
      )
    ),
    v_started_at,
    clock_timestamp()
  );

  RETURN v_rows;
END;
$$;