feat(geo): switch geocoder from self-host Photon to OpenCage API

logic_geo проксирует OpenCage (suggest/geocode/reverse): env OPENCAGE_API_KEY, OPENCAGE_URL, OPENCAGE_TIMEOUT_MS, OPENCAGE_COUNTRYCODE, OPENCAGE_THROTTLE_RPS (1 rps под free-тариф, 0 - выключить). HTTP 402/403/429 апстрима -> 503 geo_unavailable; source=opencage. Сервис photon и volume photon-data удалены из swarm compose; eunit-фикстуры под формат OpenCage. Refs EventHub/EventHubBack#77
2026-08-18 23:30:15 +03:00
parent 6b70e1336d
commit 73d43e5fdc
5 changed files with 163 additions and 87 deletions
@@ -43,8 +43,7 @@ REMINDER_LEAD_HOURS=24
 # Uploads (avatar/cover) — каталог на volume /app/data (Back#71)
 UPLOAD_DIR=/app/data/uploads
 UPLOAD_MAX_BYTES=2097152
-# Photon (DevOps#16). Пусто = /v1/geo → 503. Включить: PHOTON_REPLICAS=1 и PHOTON_URL=http://photon:2322
-PHOTON_URL=
-PHOTON_TIMEOUT_MS=3000
-PHOTON_REPLICAS=0
+# OpenCage (DevOps#16). ������ ���� = /v1/geo -> 503 (degrade). Free: 2500/�����, 1 rps.
+OPENCAGE_API_KEY=
+OPENCAGE_TIMEOUT_MS=3000

@@ -70,8 +70,8 @@ services:
      - ADMIN_SUPPORT_PASSWORD=${ADMIN_SUPPORT_PASSWORD}
      - CLUSTER_MODE=true
      - DNS_NAME=eventhub-node
-      - PHOTON_URL=${PHOTON_URL:-}
-      - PHOTON_TIMEOUT_MS=${PHOTON_TIMEOUT_MS:-3000}
+      - OPENCAGE_API_KEY=${OPENCAGE_API_KEY:-}
+      - OPENCAGE_TIMEOUT_MS=${OPENCAGE_TIMEOUT_MS:-3000}
    networks:
      eventhub-net:
        aliases:
@@ -235,29 +235,6 @@ services:
        max_attempts: 3
        window: 120s

-  # OSM geocoder (komoot Photon). Internal only — do not publish :2322.
-  # Default replicas 0. Enable: PHOTON_REPLICAS=1 PHOTON_URL=http://photon:2322
-  photon:
-    image: ${PHOTON_IMAGE:-rtuszik/photon-docker:latest}
-    environment:
-      - REGION=${PHOTON_REGION:-russia}
-      - UPDATE_STRATEGY=${PHOTON_UPDATE_STRATEGY:-PARALLEL}
-    volumes:
-      - photon-data:/photon/data
-    networks:
-      eventhub-net:
-        aliases:
-          - photon
-    deploy:
-      replicas: ${PHOTON_REPLICAS:-0}
-      resources:
-        limits:
-          memory: ${PHOTON_MEMORY_LIMIT:-1536M}
-        reservations:
-          memory: 128M
-      restart_policy:
-        condition: any
-
 networks:
  eventhub-net:
    driver: overlay
@@ -268,5 +245,4 @@ volumes:
  prometheus-data:
  grafana-data:
  traefik-logs:
-  loglynx-data:
-  photon-data:
+  loglynx-data:
@@ -15,7 +15,7 @@ trails() ->
    #{
      path => <<"/v1/geo/suggest">>,
      method => <<"GET">>,
-      description => <<"Address typeahead (Photon). Public + IP rate-limit.">>,
+      description => <<"Address typeahead (OpenCage). Public + IP rate-limit.">>,
      tags => [<<"Geo">>],
      parameters => [
        #{name => <<"q">>, in => <<"query">>, required => true, schema => #{type => string}},
@@ -31,7 +31,7 @@ trails() ->
    #{
      path => <<"/v1/geo/geocode">>,
      method => <<"POST">>,
-      description => <<"Forward geocode (Photon). Bearer required.">>,
+      description => <<"Forward geocode (OpenCage). Bearer required.">>,
      tags => [<<"Geo">>],
      responses => #{
        200 => #{description => <<"OK">>},
@@ -43,7 +43,7 @@ trails() ->
    #{
      path => <<"/v1/geo/reverse">>,
      method => <<"POST">>,
-      description => <<"Reverse geocode (Photon). Bearer required.">>,
+      description => <<"Reverse geocode (OpenCage). Bearer required.">>,
      tags => [<<"Geo">>],
      responses => #{
        200 => #{description => <<"OK">>},
@@ -1,6 +1,7 @@
 %%%-------------------------------------------------------------------
-%%% @doc Прокси к self-host Photon: suggest / geocode / reverse.
-%%% Кэш и rate-limit в ETS. Без PHOTON_URL — {error, unavailable}.
+%%% @doc Прокси к OpenCage API: suggest / geocode / reverse.
+%%% Кэш и rate-limit в ETS. Без OPENCAGE_API_KEY — {error, unavailable}.
+%%% Глобальный троттлинг апстрима (free-тариф: 1 rps, 2500/сутки).
 %%% @end
 %%%-------------------------------------------------------------------
 -module(logic_geo).
@@ -9,11 +10,13 @@

 -define(CACHE, eventhub_geo_cache).
 -define(RL, eventhub_geo_rl).
+-define(UP, eventhub_geo_upstream).
 -define(MIN_Q, 3).
 -define(CACHE_TTL_SEC, 604800).
 -define(RL_WINDOW_SEC, 60).
 -define(RL_SUGGEST, 40).
 -define(RL_WRITE, 20).
+-define(DEFAULT_URL, <<"https://api.opencagedata.com/geocode/v1/json">>).

 -spec ensure() -> ok.
 ensure() ->
@@ -27,6 +30,11 @@ ensure() ->
      ets:new(?RL, [named_table, public, set, {write_concurrency, true}]);
    _ -> ok
  end,
+  case ets:info(?UP) of
+    undefined ->
+      ets:new(?UP, [named_table, public, set]);
+    _ -> ok
+  end,
  ok.

 -spec suggest(binary(), binary(), binary()) ->
@@ -44,12 +52,12 @@ suggest(Q, Lang, RlKey) ->
          case cache_get(CacheKey) of
            {ok, Hits} -> {ok, Hits};
            miss ->
-              case photon_get(<<"/api">>, [{<<"q">>, Qn}, {<<"lang">>, LangN},
-                                           {<<"limit">>, <<"8">>}]) of
+              case opencage_get(forward_q(Qn, LangN, <<"8">>)) of
                {ok, Body} ->
-                  Hits = features_to_hits(Body),
+                  Hits = results_to_hits(Body),
                  cache_put(CacheKey, Hits),
                  {ok, Hits};
+                {error, rate_limited} -> {error, rate_limited};
                {error, _} -> {error, unavailable}
              end
          end
@@ -71,16 +79,16 @@ geocode(Q, Lang, RlKey) ->
          case cache_get(CacheKey) of
            {ok, Hit} -> {ok, Hit};
            miss ->
-              case photon_get(<<"/api">>, [{<<"q">>, Qn}, {<<"lang">>, LangN},
-                                           {<<"limit">>, <<"1">>}]) of
+              case opencage_get(forward_q(Qn, LangN, <<"1">>)) of
                {ok, Body} ->
-                  case features_to_hits(Body) of
+                  case results_to_hits(Body) of
                    [Hit | _] ->
-                      Out = Hit#{<<"source">> => <<"photon">>},
+                      Out = Hit#{<<"source">> => <<"opencage">>},
                      cache_put(CacheKey, Out),
                      {ok, Out};
                    [] -> {error, not_found}
                  end;
+                {error, rate_limited} -> {error, rate_limited};
                {error, _} -> {error, unavailable}
              end
          end
@@ -105,19 +113,21 @@ reverse(Lat, Lon, Lang, RlKey) when is_number(Lat), is_number(Lon) ->
            miss ->
              LatB = float_to_bin(LatR),
              LonB = float_to_bin(LonR),
-              case photon_get(<<"/reverse">>, [{<<"lat">>, LatB}, {<<"lon">>, LonB},
-                                               {<<"lang">>, LangN}]) of
+              Qs = [{<<"q">>, <<LatB/binary, ",", LonB/binary>>},
+                    {<<"language">>, LangN}],
+              case opencage_get(Qs) of
                {ok, Body} ->
-                  Hit = case features_to_hits(Body) of
-                          [H | _] -> H#{<<"source">> => <<"photon">>};
+                  Hit = case results_to_hits(Body) of
+                          [H | _] -> H#{<<"source">> => <<"opencage">>};
                          [] ->
                            #{<<"address">> => fallback_addr(LatR, LonR),
                              <<"lat">> => LatR,
                              <<"lon">> => LonR,
-                              <<"source">> => <<"photon">>}
+                              <<"source">> => <<"opencage">>}
                        end,
                  cache_put(CacheKey, Hit),
                  {ok, Hit};
+                {error, rate_limited} -> {error, rate_limited};
                {error, _} -> {error, unavailable}
              end
          end
@@ -184,15 +194,37 @@ cache_put(Key, Val) ->
  ets:insert(?CACHE, {Key, Exp, Val}),
  ok.

-photon_url() ->
-  case os:getenv("PHOTON_URL") of
+%% Forward-запрос: q + language + limit (+ countrycode, если задан).
+forward_q(Q, LangN, Limit) ->
+  Qs = [{<<"q">>, Q}, {<<"language">>, LangN}, {<<"limit">>, Limit}],
+  case countrycode() of
+    <<>> -> Qs;
+    CC -> [{<<"countrycode">>, CC} | Qs]
+  end.
+
+api_key() ->
+  case os:getenv("OPENCAGE_API_KEY") of
    false -> <<>>;
    "" -> <<>>;
+    K -> list_to_binary(string:trim(K))
+  end.
+
+base_url() ->
+  case os:getenv("OPENCAGE_URL") of
+    false -> ?DEFAULT_URL;
+    "" -> ?DEFAULT_URL;
    Url -> list_to_binary(string:trim(Url))
  end.

+countrycode() ->
+  case os:getenv("OPENCAGE_COUNTRYCODE") of
+    false -> <<>>;
+    "" -> <<>>;
+    CC -> list_to_binary(string:trim(CC))
+  end.
+
 timeout_ms() ->
-  case os:getenv("PHOTON_TIMEOUT_MS") of
+  case os:getenv("OPENCAGE_TIMEOUT_MS") of
    false -> 3000;
    "" -> 3000;
    S ->
@@ -204,14 +236,47 @@ timeout_ms() ->
      end
  end.

-photon_get(Path, Qs) ->
-  case photon_url() of
+%% 0 — троттлинг выключен (платный тариф); по умолчанию 1 rps (free).
+throttle_rps() ->
+  case os:getenv("OPENCAGE_THROTTLE_RPS") of
+    false -> 1;
+    "" -> 1;
+    S ->
+      try list_to_integer(S) of
+        N when N >= 0 -> N;
+        _ -> 1
+      catch
+        _:_ -> 1
+      end
+  end.
+
+throttle_allow() ->
+  case throttle_rps() of
+    0 -> true;
+    Rps when Rps > 0 ->
+      NowMs = erlang:monotonic_time(millisecond),
+      MinGap = 1000 div Rps,
+      case ets:lookup(?UP, last) of
+        [{last, Prev}] when NowMs - Prev < MinGap -> false;
+        _ ->
+          ets:insert(?UP, {last, NowMs}),
+          true
+      end
+  end.
+
+opencage_get(Qs) ->
+  case api_key() of
    <<>> -> {error, unavailable};
-    Base0 ->
-      Base = strip_slash(Base0),
-      Query = uri_string:compose_query(Qs),
-      Url = binary_to_list(<<Base/binary, Path/binary, $?, Query/binary>>),
-      http_get(Url)
+    Key ->
+      case throttle_allow() of
+        false -> {error, rate_limited};
+        true ->
+          Full = [{<<"key">>, Key}, {<<"no_annotations">>, <<"1">>} | Qs],
+          Query = uri_string:compose_query(Full),
+          Url = binary_to_list(<<(strip_slash(base_url()))/binary,
+                                 $?, Query/binary>>),
+          http_get(Url)
+      end
  end.

 strip_slash(B) ->
@@ -240,27 +305,50 @@ default_http_get(Url) ->
      {error, Reason}
  end.

-features_to_hits(Body) when is_binary(Body) ->
+results_to_hits(Body) when is_binary(Body) ->
  try jsx:decode(Body, [return_maps]) of
-    Map when is_map(Map) ->
-      Feats = maps:get(<<"features">>, Map, []),
-      lists:filtermap(fun feature_to_hit/1, Feats);
+    #{<<"results">> := Results} when is_list(Results) ->
+      lists:filtermap(fun result_to_hit/1, Results);
    _ -> []
  catch
    _:_ -> []
  end.

-feature_to_hit(#{<<"geometry">> := #{<<"coordinates">> := [Lon, Lat | _]}} = F) ->
-  Props = maps:get(<<"properties">>, F, #{}),
-  Addr = format_address(Props),
+result_to_hit(#{<<"geometry">> := #{<<"lat">> := Lat, <<"lng">> := Lng}} = R) ->
  {true, #{
-    <<"address">> => Addr,
+    <<"address">> => address_of(R),
    <<"lat">> => to_num(Lat),
-    <<"lon">> => to_num(Lon)
+    <<"lon">> => to_num(Lng)
  }};
-feature_to_hit(_) ->
+result_to_hit(_) ->
  false.

+address_of(R) ->
+  case bin(maps:get(<<"formatted">>, R, <<>>)) of
+    <<>> ->
+      C = maps:get(<<"components">>, R, #{}),
+      Name = bin(maps:get(<<"name">>, C, <<>>)),
+      House = bin(maps:get(<<"house_number">>, C, <<>>)),
+      Road = bin(maps:get(<<"road">>, C, <<>>)),
+      StreetLine = case {House, Road} of
+                     {<<>>, Rd} -> Rd;
+                     {H, <<>>} -> H;
+                     {H, Rd} -> <<H/binary, " ", Rd/binary>>
+                   end,
+      City = first_nonempty([
+        maps:get(<<"city">>, C, <<>>),
+        maps:get(<<"town">>, C, <<>>),
+        maps:get(<<"village">>, C, <<>>)
+      ]),
+      Country = bin(maps:get(<<"country">>, C, <<>>)),
+      Parts = [P || P <- [Name, StreetLine, City, Country], P =/= <<>>],
+      case unique_keep(Parts) of
+        [] -> <<"—">>;
+        Ps -> join_comma(Ps)
+      end;
+    Addr -> Addr
+  end.
+
 to_num(N) when is_integer(N) -> float(N);
 to_num(N) when is_float(N) -> N;
 to_num(_) -> 0.0.
@@ -4,16 +4,22 @@
 setup() ->
  catch ets:delete(eventhub_geo_cache),
  catch ets:delete(eventhub_geo_rl),
+  catch ets:delete(eventhub_geo_upstream),
  logic_geo:ensure(),
  application:unset_env(eventhub, geo_http_get),
-  os:putenv("PHOTON_URL", "http://photon.test"),
+  os:putenv("OPENCAGE_API_KEY", "test-key"),
+  os:putenv("OPENCAGE_URL", "http://opencage.test/geocode/v1/json"),
+  os:putenv("OPENCAGE_THROTTLE_RPS", "0"),
  ok.

 cleanup(_) ->
  application:unset_env(eventhub, geo_http_get),
-  os:unsetenv("PHOTON_URL"),
+  os:unsetenv("OPENCAGE_API_KEY"),
+  os:unsetenv("OPENCAGE_URL"),
+  os:unsetenv("OPENCAGE_THROTTLE_RPS"),
  catch ets:delete(eventhub_geo_cache),
  catch ets:delete(eventhub_geo_rl),
+  catch ets:delete(eventhub_geo_upstream),
  ok.

 logic_geo_test_() ->
@@ -22,22 +28,21 @@ logic_geo_test_() ->
    fun cleanup/1,
    [
      {"Suggest too short", fun test_suggest_short/0},
-      {"Suggest parses photon geojson", fun test_suggest_ok/0},
-      {"Geocode empty features is not_found", fun test_geocode_not_found/0},
-      {"Unavailable without photon url", fun test_unavailable/0},
+      {"Suggest parses opencage results", fun test_suggest_ok/0},
+      {"Geocode empty results is not_found", fun test_geocode_not_found/0},
+      {"Unavailable without opencage key", fun test_unavailable/0},
+      {"Upstream quota (402) is unavailable", fun test_quota_exceeded/0},
      {"Reverse fallback address", fun test_reverse_empty/0},
      {"Format address from properties", fun test_format_address/0}
    ]}.

-sample_geojson() ->
+sample_opencage() ->
  <<"{
-    \"features\": [{
-      \"geometry\": {\"coordinates\": [37.62, 55.75]},
-      \"properties\": {
-        \"name\": \"Red Square\",
-        \"city\": \"Moscow\",
-        \"country\": \"Russia\"
-      }
+    \"status\": {\"code\": 200, \"message\": \"OK\"},
+    \"results\": [{
+      \"formatted\": \"Red Square, Moscow, Russia\",
+      \"geometry\": {\"lat\": 55.75, \"lng\": 37.62},
+      \"components\": {\"city\": \"Moscow\", \"country\": \"Russia\"}
    }]
  }">>.

@@ -45,27 +50,35 @@ test_suggest_short() ->
  ?assertEqual({error, invalid_query}, logic_geo:suggest(<<"ab">>, <<"ru">>, <<"ip">>)).

 test_suggest_ok() ->
-  application:set_env(eventhub, geo_http_get, fun(_) -> {ok, sample_geojson()} end),
+  application:set_env(eventhub, geo_http_get, fun(_) -> {ok, sample_opencage()} end),
  {ok, Hits} = logic_geo:suggest(<<"красная площадь">>, <<"ru">>, <<"ip1">>),
  ?assertMatch([#{<<"lat">> := 55.75, <<"lon">> := 37.62}], Hits),
  [#{<<"address">> := Addr}] = Hits,
-  ?assertNotEqual(<<>>, Addr).
+  ?assertEqual(<<"Red Square, Moscow, Russia">>, Addr).

 test_geocode_not_found() ->
-  application:set_env(eventhub, geo_http_get, fun(_) -> {ok, <<"{\"features\":[]}">>} end),
+  application:set_env(eventhub, geo_http_get,
+    fun(_) -> {ok, <<"{\"status\":{\"code\":200},\"results\":[]}">>} end),
  ?assertEqual({error, not_found},
    logic_geo:geocode(<<"nowherexyz">>, <<"ru">>, <<"u1">>)).

 test_unavailable() ->
-  os:unsetenv("PHOTON_URL"),
+  os:unsetenv("OPENCAGE_API_KEY"),
  application:unset_env(eventhub, geo_http_get),
  ?assertEqual({error, unavailable},
    logic_geo:suggest(<<"москва центр">>, <<"ru">>, <<"ip2">>)).

+test_quota_exceeded() ->
+  application:set_env(eventhub, geo_http_get,
+    fun(_) -> {error, {http_status, 402}} end),
+  ?assertEqual({error, unavailable},
+    logic_geo:suggest(<<"москва центр">>, <<"ru">>, <<"ip3">>)).
+
 test_reverse_empty() ->
-  application:set_env(eventhub, geo_http_get, fun(_) -> {ok, <<"{\"features\":[]}">>} end),
+  application:set_env(eventhub, geo_http_get,
+    fun(_) -> {ok, <<"{\"status\":{\"code\":200},\"results\":[]}">>} end),
  {ok, Hit} = logic_geo:reverse(55.75, 37.62, <<"ru">>, <<"u2">>),
-  ?assertEqual(<<"photon">>, maps:get(<<"source">>, Hit)),
+  ?assertEqual(<<"opencage">>, maps:get(<<"source">>, Hit)),
  ?assertEqual(55.75, maps:get(<<"lat">>, Hit)).

 test_format_address() ->