GITGUD

neiam neiam /apollos_crib

It's Apollo's Crib

public · Issues · Pulls 1 · Labels · Forks · Compare · Actions failure · Badges · Packages

⭐

Log in to mark this repository.

Credo: the other source workers and Hermes in smaller pieces

ca307c7 · Gabe Morell · 2026-11-04 20:59

38 files +488 -507
Message 
{commit_body(@commit)}

Files changed

modified apps/room_air_quality/lib/room_air_quality/aqi_worker.ex
+230 −225
@@ -99,236 +99,241 @@ defmodule RoomAirQuality.Worker do
99 99 datetime = NaiveDateTime.local_now()
100 100 Logger.info("AQI::#{id} writing bundle #{type}")
101 101
102 − case type do
103 − :hourly ->
104 − result.body
105 − |> String.split(~r/\r?\n/)
106 − |> Enum.filter(fn x -> x != "" end)
107 − |> Enum.map(&(&1 <> "\n"))
108 − |> CSV.decode(
109 − headers: [
110 − :valid_date,
111 − :valid_time,
112 − :aqsid,
113 − :site_name,
114 − :gmt_offset,
115 − :parameter_name,
116 − :reporting_units,
117 − :value,
118 − :data_source
119 − ],
120 − separator: ?|
102 + write_bundle(type, result, id, datetime)
103 +
104 + Logger.info("AQI::#{id} completed writing bundle")
105 + end
106 +
107 + defp write_bundle(:hourly, result, id, datetime) do
108 + result.body
109 + |> String.split(~r/\r?\n/)
110 + |> Enum.filter(fn x -> x != "" end)
111 + |> Enum.map(&(&1 <> "\n"))
112 + |> CSV.decode(
113 + headers: [
114 + :valid_date,
115 + :valid_time,
116 + :aqsid,
117 + :site_name,
118 + :gmt_offset,
119 + :parameter_name,
120 + :reporting_units,
121 + :value,
122 + :data_source
123 + ],
124 + separator: ?|
125 + )
126 + |> Stream.map(fn {:ok, data} ->
127 + RoomSanctum.Storage.change_hourly_data(
128 + %RoomSanctum.Storage.AirNow.HourlyData{},
129 + data
130 + |> Map.put(:source_id, id)
131 + |> Map.put(:valid_date, data.valid_date |> Timex.parse!("%m/%d/%y", :strftime))
132 + )
133 + |> Map.put(:inserted_at, datetime)
134 + |> Map.put(:updated_at, datetime)
135 + end)
136 + |> Enum.each(fn x ->
137 + Repo.insert(
138 + x,
139 + on_conflict: {:replace_all_except, [:id]},
140 + conflict_target: [:source_id, :site_name, :parameter_name]
141 + )
142 + end)
143 + end
144 +
145 + defp write_bundle(:sites, result, id, datetime) do
146 + result.body
147 + |> String.split(~r/\r?\n/)
148 + |> List.delete_at(0)
149 + |> Enum.filter(fn x -> x != "" end)
150 + # The :csv library expects each stream element to end with a newline;
151 + # without it, consecutive lines are interpreted as quote-escaped
152 + # continuations and the entire file collapses into a single row.
153 + |> Enum.map(&(&1 <> "\n"))
154 + |> CSV.decode(
155 + headers: [
156 + :station_id,
157 + :aqsid,
158 + :full_aqsid,
159 + :parameter,
160 + :monitor_type,
161 + :site_code,
162 + :site_name,
163 + :status,
164 + :agency_id,
165 + :agency_name,
166 + :epa_region,
167 + :latitude,
168 + :longitude,
169 + :elevation,
170 + :gmt_offset,
171 + :country_fips,
172 + :cbsa_id,
173 + :cbsa_name,
174 + :state_aqs_code,
175 + :state_abbreviation,
176 + :county_aqs_code,
177 + :county_name
178 + ],
179 + separator: ?|
180 + )
181 + |> Enum.to_list()
182 + |> Enum.map(fn {:ok, x} -> x end)
183 + |> Enum.group_by(fn x -> x.aqsid end)
184 + |> Enum.map(fn {_k, v} ->
185 + p = v |> Enum.map(fn q -> q.parameter end)
186 + f = v |> List.first()
187 + f |> Map.put(:parameters, p)
188 + end)
189 + |> Enum.map(fn data ->
190 + # {lon, lat}, as PostGIS expects -- st_distance against a foci
191 + # compares the two directly.
192 + point = %Geo.Point{
193 + coordinates: {data.longitude |> String.to_float(), data.latitude |> String.to_float()},
194 + srid: 4326
195 + }
196 +
197 + RoomSanctum.Storage.change_monitoring_site(
198 + %RoomSanctum.Storage.AirNow.MonitoringSite{},
199 + data
200 + |> Map.put(:source_id, id)
201 + |> Map.put(:point, point)
202 + |> Map.put(:gmt_offset, data.gmt_offset |> String.to_float() |> Kernel.trunc())
203 + )
204 + |> Map.put(:inserted_at, datetime)
205 + |> Map.put(:updated_at, datetime)
206 + end)
207 + |> Enum.each(fn x ->
208 + Repo.insert(
209 + x,
210 + on_conflict: {:replace_all_except, [:id]},
211 + conflict_target: [:source_id, :station_id]
212 + )
213 + end)
214 + end
215 +
216 + defp write_bundle(:hourly_obs, result, id, datetime) do
217 + # Column order matches AirNow's current HourlyAQObs CSV header (verified live).
218 + # Positions 23-34 reorder: PM25, OZONE, NO2, CO, SO2, PM10 (each followed by its _Unit).
219 + result.body
220 + |> String.split(~r/\r?\n/)
221 + |> List.delete_at(0)
222 + |> Enum.filter(fn x -> x != "" end)
223 + # The :csv library expects each stream element to end with a newline;
224 + # without it, consecutive lines are interpreted as quote-escaped
225 + # continuations and the entire file collapses into a single row.
226 + |> Enum.map(&(&1 <> "\n"))
227 + |> CSV.decode(
228 + headers: [
229 + :aqsid,
230 + :site_name,
231 + :status,
232 + :epa_region,
233 + :lat,
234 + :lon,
235 + :elevation,
236 + :gmt_offset,
237 + :country_code,
238 + :state_name,
239 + :valid_date,
240 + :valid_time,
241 + :data_source,
242 + :reporting_areas,
243 + :ozone_aqi,
244 + :pm10_aqi,
245 + :pm25_aqi,
246 + :no2_aqi,
247 + :ozone_measured,
248 + :pm10_measured,
249 + :pm25_measured,
250 + :no2_measured,
251 + :pm25,
252 + :pm25_unit,
253 + :ozone,
254 + :ozone_unit,
255 + :no2,
256 + :no2_unit,
257 + :co,
258 + :co_unit,
259 + :so2,
260 + :so2_unit,
261 + :pm10,
262 + :pm10_unit
263 + ]
264 + )
265 + |> Enum.to_list()
266 + |> Enum.map(fn {:ok, x} -> x end)
267 + |> Enum.map(&normalize_blanks/1)
268 + |> Enum.map(fn data ->
269 + lat = parse_float(data.lat)
270 + lon = parse_float(data.lon)
271 +
272 + point =
273 + if lat && lon do
274 + %Geo.Point{coordinates: {lon, lat}, srid: 4326}
275 + end
276 +
277 + RoomSanctum.Storage.change_hourly_obs_data(
278 + %RoomSanctum.Storage.AirNow.HourlyObsData{},
279 + data
280 + |> Map.put(:source_id, id)
281 + |> Map.put(:point, point)
282 + |> Map.put(:lat, lat)
283 + |> Map.put(:lon, lon)
284 + |> Map.put(:elevation, parse_float(data[:elevation]))
285 + |> Map.put(:gmt_offset, parse_float(data[:gmt_offset]))
286 + |> Map.put(
287 + :reporting_areas,
288 + (data[:reporting_areas] || "") |> String.split("|")
121 289 )
122 − |> Stream.map(fn {:ok, data} ->
123 − RoomSanctum.Storage.change_hourly_data(
124 − %RoomSanctum.Storage.AirNow.HourlyData{},
125 − data
126 − |> Map.put(:source_id, id)
127 − |> Map.put(:valid_date, data.valid_date |> Timex.parse!("%m/%d/%y", :strftime))
128 − )
129 − |> Map.put(:inserted_at, datetime)
130 − |> Map.put(:updated_at, datetime)
131 − end)
132 − |> Enum.each(fn x ->
133 − Repo.insert(
134 − x,
135 − on_conflict: {:replace_all_except, [:id]},
136 − conflict_target: [:source_id, :site_name, :parameter_name]
137 − )
138 − end)
139 −
140 − :sites ->
141 − result.body
142 − |> String.split(~r/\r?\n/)
143 − |> List.delete_at(0)
144 − |> Enum.filter(fn x -> x != "" end)
145 − # The :csv library expects each stream element to end with a newline;
146 − # without it, consecutive lines are interpreted as quote-escaped
147 − # continuations and the entire file collapses into a single row.
148 − |> Enum.map(&(&1 <> "\n"))
149 − |> CSV.decode(
150 − headers: [
151 − :station_id,
152 − :aqsid,
153 − :full_aqsid,
154 − :parameter,
155 − :monitor_type,
156 − :site_code,
157 − :site_name,
158 − :status,
159 − :agency_id,
160 − :agency_name,
161 − :epa_region,
162 − :latitude,
163 − :longitude,
164 − :elevation,
165 − :gmt_offset,
166 − :country_fips,
167 − :cbsa_id,
168 − :cbsa_name,
169 − :state_aqs_code,
170 − :state_abbreviation,
171 − :county_aqs_code,
172 − :county_name
173 − ],
174 − separator: ?|
290 + |> Map.put(
291 + :valid_date,
292 + data.valid_date |> Timex.parse!("%m/%d/%Y", :strftime) |> NaiveDateTime.to_date()
175 293 )
176 − |> Enum.to_list()
177 − |> Enum.map(fn {:ok, x} -> x end)
178 − |> Enum.group_by(fn x -> x.aqsid end)
179 − |> Enum.map(fn {_k, v} ->
180 − p = v |> Enum.map(fn q -> q.parameter end)
181 − f = v |> List.first()
182 − f |> Map.put(:parameters, p)
183 − end)
184 − |> Enum.map(fn data ->
185 − # {lon, lat}, as PostGIS expects -- st_distance against a foci
186 − # compares the two directly.
187 − point = %Geo.Point{
188 − coordinates:
189 − {data.longitude |> String.to_float(), data.latitude |> String.to_float()},
190 − srid: 4326
191 − }
192 −
193 − RoomSanctum.Storage.change_monitoring_site(
194 − %RoomSanctum.Storage.AirNow.MonitoringSite{},
195 − data
196 − |> Map.put(:source_id, id)
197 − |> Map.put(:point, point)
198 − |> Map.put(:gmt_offset, data.gmt_offset |> String.to_float() |> Kernel.trunc())
199 − )
200 − |> Map.put(:inserted_at, datetime)
201 − |> Map.put(:updated_at, datetime)
202 − end)
203 − |> Enum.each(fn x ->
204 − Repo.insert(
205 − x,
206 − on_conflict: {:replace_all_except, [:id]},
207 − conflict_target: [:source_id, :station_id]
208 − )
209 − end)
210 −
211 − :hourly_obs ->
212 − # Column order matches AirNow's current HourlyAQObs CSV header (verified live).
213 − # Positions 23-34 reorder: PM25, OZONE, NO2, CO, SO2, PM10 (each followed by its _Unit).
214 − result.body
215 − |> String.split(~r/\r?\n/)
216 − |> List.delete_at(0)
217 − |> Enum.filter(fn x -> x != "" end)
218 − # The :csv library expects each stream element to end with a newline;
219 − # without it, consecutive lines are interpreted as quote-escaped
220 − # continuations and the entire file collapses into a single row.
221 − |> Enum.map(&(&1 <> "\n"))
222 − |> CSV.decode(
223 − headers: [
224 − :aqsid,
225 − :site_name,
226 − :status,
227 − :epa_region,
228 − :lat,
229 − :lon,
230 − :elevation,
231 − :gmt_offset,
232 − :country_code,
233 − :state_name,
234 − :valid_date,
235 − :valid_time,
236 − :data_source,
237 − :reporting_areas,
238 − :ozone_aqi,
239 − :pm10_aqi,
240 − :pm25_aqi,
241 − :no2_aqi,
242 − :ozone_measured,
243 − :pm10_measured,
244 − :pm25_measured,
245 − :no2_measured,
246 − :pm25,
247 − :pm25_unit,
248 − :ozone,
249 − :ozone_unit,
250 − :no2,
251 − :no2_unit,
252 − :co,
253 − :co_unit,
254 − :so2,
255 − :so2_unit,
256 − :pm10,
257 − :pm10_unit
258 − ]
294 + |> Map.put(
295 + :valid_time,
296 + data.valid_time |> Timex.parse!("%H:%M", :strftime) |> NaiveDateTime.to_time()
297 + )
298 + )
299 + |> Map.put(:inserted_at, datetime)
300 + |> Map.put(:updated_at, datetime)
301 + |> Map.put(:ozone_aqi, data.ozone_aqi |> intify)
302 + |> Map.put(:pm10_aqi, data.pm10_aqi |> intify)
303 + |> Map.put(:pm25_aqi, data.pm25_aqi |> intify)
304 + |> Map.put(:no2_aqi, data.no2_aqi |> intify)
305 + end)
306 + |> Enum.reduce({0, 0, []}, &insert_obs/2)
307 + |> case do
308 + {ok, 0, _} ->
309 + Logger.info("AQI::#{id} inserted #{ok} rows")
310 +
311 + {ok, fail, sample} ->
312 + Logger.warning(
313 + "AQI::#{id} inserted #{ok} rows, #{fail} failed. First failures: #{inspect(sample)}"
259 314 )
260 − |> Enum.to_list()
261 − |> Enum.map(fn {:ok, x} -> x end)
262 − |> Enum.map(&normalize_blanks/1)
263 − |> Enum.map(fn data ->
264 − lat = parse_float(data.lat)
265 − lon = parse_float(data.lon)
266 −
267 − point =
268 − if lat && lon do
269 − %Geo.Point{coordinates: {lon, lat}, srid: 4326}
270 − end
271 −
272 − RoomSanctum.Storage.change_hourly_obs_data(
273 − %RoomSanctum.Storage.AirNow.HourlyObsData{},
274 − data
275 − |> Map.put(:source_id, id)
276 − |> Map.put(:point, point)
277 − |> Map.put(:lat, lat)
278 − |> Map.put(:lon, lon)
279 − |> Map.put(:elevation, parse_float(data[:elevation]))
280 − |> Map.put(:gmt_offset, parse_float(data[:gmt_offset]))
281 − |> Map.put(
282 − :reporting_areas,
283 − (data[:reporting_areas] || "") |> String.split("|")
284 − )
285 − |> Map.put(
286 − :valid_date,
287 − data.valid_date |> Timex.parse!("%m/%d/%Y", :strftime) |> NaiveDateTime.to_date()
288 − )
289 − |> Map.put(
290 − :valid_time,
291 − data.valid_time |> Timex.parse!("%H:%M", :strftime) |> NaiveDateTime.to_time()
292 − )
293 − )
294 − |> Map.put(:inserted_at, datetime)
295 − |> Map.put(:updated_at, datetime)
296 − |> Map.put(:ozone_aqi, data.ozone_aqi |> intify)
297 − |> Map.put(:pm10_aqi, data.pm10_aqi |> intify)
298 − |> Map.put(:pm25_aqi, data.pm25_aqi |> intify)
299 − |> Map.put(:no2_aqi, data.no2_aqi |> intify)
300 − end)
301 − |> Enum.reduce({0, 0, []}, fn cs, {ok, fail, sample_errs} ->
302 − case Repo.insert(cs,
303 − on_conflict: {:replace_all_except, [:id]},
304 − conflict_target: [:source_id, :aqsid]
305 − ) do
306 − {:ok, _} ->
307 − {ok + 1, fail, sample_errs}
308 −
309 − {:error, bad} ->
310 − new_sample =
311 − if length(sample_errs) < 3 do
312 − [{cs.changes[:aqsid] || cs.data.aqsid, bad.errors} | sample_errs]
313 − else
314 − sample_errs
315 − end
316 −
317 − {ok, fail + 1, new_sample}
318 − end
319 − end)
320 − |> case do
321 − {ok, 0, _} ->
322 − Logger.info("AQI::#{id} inserted #{ok} rows")
323 −
324 − {ok, fail, sample} ->
325 − Logger.warning(
326 − "AQI::#{id} inserted #{ok} rows, #{fail} failed. First failures: #{inspect(sample)}"
327 − )
328 − end
329 315 end
316 + end
330 317
331 − Logger.info("AQI::#{id} completed writing bundle")
318 + # Counts inserts and failures, keeping the first three failures to report.
319 + defp insert_obs(cs, {ok, fail, sample_errs}) do
320 + case Repo.insert(cs,
321 + on_conflict: {:replace_all_except, [:id]},
322 + conflict_target: [:source_id, :aqsid]
323 + ) do
324 + {:ok, _} ->
325 + {ok + 1, fail, sample_errs}
326 +
327 + {:error, bad} ->
328 + new_sample =
329 + if length(sample_errs) < 3 do
330 + [{cs.changes[:aqsid] || cs.data.aqsid, bad.errors} | sample_errs]
331 + else
332 + sample_errs
333 + end
334 +
335 + {ok, fail + 1, new_sample}
336 + end
332 337 end
333 338
334 339 # defp build_todays_url(ts \\ DateTime.utc_now()) do
modified apps/room_bourse/lib/room_bourse/worker.ex
+13 −4
@@ -159,10 +159,8 @@ defmodule RoomBourse.Worker do
159 159 defp present(meta, symbol) do
160 160 price = meta["regularMarketPrice"]
161 161 previous = meta["chartPreviousClose"] || meta["previousClose"]
162 − change = if is_number(price) and is_number(previous), do: price - previous
163 −
164 − change_pct =
165 − if is_number(change) and is_number(previous) and previous != 0, do: change / previous * 100
162 + change = change(price, previous)
163 + change_pct = change_pct(change, previous)
166 164
167 165 %{
168 166 symbol: meta["symbol"] || String.upcase(symbol),
@@ -183,6 +181,17 @@ defmodule RoomBourse.Worker do
183 181 }
184 182 end
185 183
184 + defp change(price, previous) when is_number(price) and is_number(previous),
185 + do: price - previous
186 +
187 + defp change(_price, _previous), do: nil
188 +
189 + defp change_pct(change, previous)
190 + when is_number(change) and is_number(previous) and previous != 0,
191 + do: change / previous * 100
192 +
193 + defp change_pct(_change, _previous), do: nil
194 +
186 195 defp direction(change) when is_number(change) and change > 0, do: :up
187 196 defp direction(change) when is_number(change) and change < 0, do: :down
188 197 defp direction(change) when is_number(change), do: :flat
modified apps/room_drought/lib/room_drought/worker.ex
+29 −29
@@ -93,40 +93,40 @@ defmodule RoomDrought.Worker do
93 93 def handle_call({:read, query}, _from, state) do
94 94 product = Map.get(query, :product) || Map.get(query, "product") || "usdm"
95 95
96 − result =
97 − case product do
98 − "usdm" ->
99 − case scope_from_query(query) do
100 − nil ->
101 − []
102 −
103 − target ->
104 − # Make sure we're tracking it for next refresh
105 − GenServer.cast(self(), {:add_target, target})
106 −
107 − case Map.get(state.usdm, target) do
108 − nil ->
109 − case fetch_usdm(target) do
110 − {:ok, record} -> [record]
111 − :error -> []
112 − end
113 −
114 − record ->
115 − [record]
116 − end
117 − end
96 + {:reply, read_product(product, query, state), state}
97 + end
118 98
119 − product when product in ["csi", "cpc", "gpcc"] ->
120 − [%{"product" => product, "status" => "not_yet_implemented"}]
99 + def handle_call(_msg, _from, state), do: {:reply, :ok, state}
121 100
122 − _ ->
123 − []
124 − end
101 + defp read_product("usdm", query, state) do
102 + case scope_from_query(query) do
103 + nil ->
104 + []
125 105
126 − {:reply, result, state}
106 + target ->
107 + # Make sure we're tracking it for next refresh
108 + GenServer.cast(self(), {:add_target, target})
109 + read_usdm(target, state)
110 + end
127 111 end
128 112
129 − def handle_call(_msg, _from, state), do: {:reply, :ok, state}
113 + defp read_product(product, _query, _state) when product in ["csi", "cpc", "gpcc"],
114 + do: [%{"product" => product, "status" => "not_yet_implemented"}]
115 +
116 + defp read_product(_product, _query, _state), do: []
117 +
118 + defp read_usdm(target, state) do
119 + case Map.get(state.usdm, target) do
120 + nil ->
121 + case fetch_usdm(target) do
122 + {:ok, record} -> [record]
123 + :error -> []
124 + end
125 +
126 + record ->
127 + [record]
128 + end
129 + end
130 130
131 131 # Helpers
132 132 defp via(name), do: via_tuple(name)
modified apps/room_github/lib/room_github/worker.ex
+48 −45
@@ -231,66 +231,69 @@ defmodule RoomGithub.Worker do
231 231 end
232 232
233 233 # HTTPoison async stream handlers
234 − def handle_info(msg, state) do
235 − case msg do
236 − %HTTPoison.Error{reason: {:closed, :timeout}} ->
237 − {:noreply, state}
238 −
239 − %HTTPoison.AsyncStatus{code: 200, id: _id} ->
240 − {:noreply, state}
234 + def handle_info(%HTTPoison.Error{reason: {:closed, :timeout}}, state) do
235 + {:noreply, state}
236 + end
241 237
242 − %HTTPoison.AsyncStatus{code: code, id: rid} ->
243 − Logger.info("GH::#{state.inst.id} non-200: #{code}")
238 + def handle_info(%HTTPoison.AsyncStatus{code: 200, id: _id}, state) do
239 + {:noreply, state}
240 + end
244 241
245 − ref = state.refs[rid]
246 − ref = ref && Map.put(ref, :write, false)
247 − refs = if ref, do: Map.put(state.refs, rid, ref), else: state.refs
248 − {:noreply, state |> Map.put(:refs, refs)}
242 + def handle_info(%HTTPoison.AsyncStatus{code: code, id: rid}, state) do
243 + Logger.info("GH::#{state.inst.id} non-200: #{code}")
249 244
250 − %HTTPoison.AsyncHeaders{headers: headers, id: rid} ->
251 − case state.refs[rid] do
252 − nil ->
253 − {:noreply, state}
245 + ref = state.refs[rid]
246 + ref = ref && Map.put(ref, :write, false)
247 + refs = if ref, do: Map.put(state.refs, rid, ref), else: state.refs
248 + {:noreply, state |> Map.put(:refs, refs)}
249 + end
254 250
255 − ref ->
256 − ref = Map.put(ref, :next_url, parse_next_link(headers))
257 − {:noreply, state |> Map.put(:refs, Map.put(state.refs, rid, ref))}
258 − end
251 + def handle_info(%HTTPoison.AsyncHeaders{headers: headers, id: rid}, state) do
252 + case state.refs[rid] do
253 + nil ->
254 + {:noreply, state}
259 255
260 − %HTTPoison.AsyncChunk{chunk: chunk, id: rid} ->
261 − case state.refs[rid] do
262 − nil -> :ok
263 − %{fd: fd} -> IO.binwrite(fd, chunk)
264 − end
256 + ref ->
257 + ref = Map.put(ref, :next_url, parse_next_link(headers))
258 + {:noreply, state |> Map.put(:refs, Map.put(state.refs, rid, ref))}
259 + end
260 + end
265 261
266 − {:noreply, state}
262 + def handle_info(%HTTPoison.AsyncChunk{chunk: chunk, id: rid}, state) do
263 + case state.refs[rid] do
264 + nil -> :ok
265 + %{fd: fd} -> IO.binwrite(fd, chunk)
266 + end
267 267
268 − %HTTPoison.AsyncEnd{id: rid} ->
269 − case state.refs[rid] do
270 − nil ->
271 − {:noreply, state}
268 + {:noreply, state}
269 + end
272 270
273 − %{write: false} = ref ->
274 − close_buffer(ref.fd)
275 − {:noreply, state |> Map.put(:refs, Map.delete(state.refs, rid))}
271 + def handle_info(%HTTPoison.AsyncEnd{id: rid}, state) do
272 + case state.refs[rid] do
273 + nil ->
274 + {:noreply, state}
276 275
277 − ref ->
278 − {_, body} = StringIO.contents(ref.fd)
279 − close_buffer(ref.fd)
276 + %{write: false} = ref ->
277 + close_buffer(ref.fd)
278 + {:noreply, state |> Map.put(:refs, Map.delete(state.refs, rid))}
280 279
281 − new_state =
282 − state
283 − |> ingest_response(ref, body)
284 − |> follow_pagination(ref)
280 + ref ->
281 + {_, body} = StringIO.contents(ref.fd)
282 + close_buffer(ref.fd)
285 283
286 − {:noreply, new_state |> Map.put(:refs, Map.delete(new_state.refs, rid))}
287 − end
284 + new_state =
285 + state
286 + |> ingest_response(ref, body)
287 + |> follow_pagination(ref)
288 288
289 − _ ->
290 − {:noreply, state}
289 + {:noreply, new_state |> Map.put(:refs, Map.delete(new_state.refs, rid))}
291 290 end
292 291 end
293 292
293 + def handle_info(_msg, state) do
294 + {:noreply, state}
295 + end
296 +
294 297 # Helpers
295 298 defp open_buffer do
296 299 {:ok, fd} = StringIO.open("")
modified apps/room_gitlab/lib/room_gitlab/gitlab_worker.ex
+54 −94
@@ -167,51 +167,13 @@ defmodule RoomGitlab.Worker do
167 167
168 168 defp filter_watched(_projects, _inst), do: []
169 169
170 − defp do_gitlab_req(uri, token, lookup, id \\ nil) do
171 − {path, params} =
172 − case lookup do
173 − # , %{updated_after: DateTime.utc_now |> DateTime.add(-3600 * 7) |> DateTime.truncate(:second) |> DateTime.to_iso8601, order_by: "updated_at"}}
174 − :projects -> {"/api/v4/projects", %{simple: true, per_page: 100}}
175 − :commits -> {"/api/v4/projects/#{id}/repository/commits/master", %{}}
176 − :jobs -> {"/api/v4/projects/#{id}/jobs", %{}}
177 − end
178 −
179 − case HTTPoison.get(
180 − uri <> path,
181 − ["PRIVATE-TOKEN": "#{token}", Accept: "Application/json; Charset=utf-8"],
182 − params: params,
183 − follow_redirect: true
184 − ) do
185 − {:ok, response} ->
186 − case response.status_code do
187 − 500 ->
188 − Logger.info("Failed to retrieve data from Gitlab due to #{response.body}")
189 − {:error, :five_hundred}
190 −
191 − 200 ->
192 − case response.body |> Jason.decode() do
193 − {:ok, decoded} -> {:ok, decoded}
194 − {:error, error} -> {:error, :decode_json}
195 − end
196 −
197 − 404 ->
198 − {:error, :four_o_four}
199 − end
200 −
201 − {:error, error} ->
202 − Logger.info("Failed to retrieve data from Gitlab due to #{error.reason}")
203 − {:error, :http}
204 − end
205 − end
170 + # , %{updated_after: DateTime.utc_now |> DateTime.add(-3600 * 7) |> DateTime.truncate(:second) |> DateTime.to_iso8601, order_by: "updated_at"}}
171 + defp gitlab_path(:projects, _id), do: {"/api/v4/projects", %{simple: true, per_page: 100}}
172 + defp gitlab_path(:commits, id), do: {"/api/v4/projects/#{id}/repository/commits/master", %{}}
173 + defp gitlab_path(:jobs, id), do: {"/api/v4/projects/#{id}/jobs", %{}}
206 174
207 175 defp do_gitlab_reqa(uri, token, lookup, id \\ nil) do
208 − {path, params} =
209 − case lookup do
210 − # , %{updated_after: DateTime.utc_now |> DateTime.add(-3600 * 7) |> DateTime.truncate(:second) |> DateTime.to_iso8601, order_by: "updated_at"}}
211 − :projects -> {"/api/v4/projects", %{simple: true, per_page: 100}}
212 − :commits -> {"/api/v4/projects/#{id}/repository/commits/master", %{}}
213 − :jobs -> {"/api/v4/projects/#{id}/jobs", %{}}
214 − end
176 + {path, params} = gitlab_path(lookup, id)
215 177
216 178 HTTPoison.get(
217 179 uri <> path,
@@ -375,68 +337,66 @@ defmodule RoomGitlab.Worker do
375 337 {:reply, [:ok], state}
376 338 end
377 339
378 − def handle_info(msg, state) do
379 − # IO.inspect({:handle_info, msg})
380 − case msg do
381 − %HTTPoison.Error{reason: {:closed, :timeout}, id: resp_id} ->
382 − {:noreply, state}
340 + # IO.inspect({:handle_info, msg})
341 + def handle_info(%HTTPoison.Error{reason: {:closed, :timeout}, id: resp_id}, state) do
342 + {:noreply, state}
343 + end
383 344
384 − %HTTPoison.AsyncStatus{code: status_code, id: resp_id} ->
385 − # IO.inspect(status_code)
386 −
387 − if status_code == 200 do
388 − {:noreply, state}
389 − else
390 − refs = state[:refs]
391 − r = state[:refs][resp_id]
392 − r = r |> Map.put(:write, false)
393 − refs = refs |> Map.put(resp_id, r)
394 − {:noreply, state |> Map.put(:refs, refs)}
395 − end
345 + def handle_info(%HTTPoison.AsyncStatus{code: status_code, id: resp_id}, state) do
346 + # IO.inspect(status_code)
396 347
397 − %HTTPoison.AsyncHeaders{headers: headers, id: resp_id} ->
398 − # IO.inspect(headers)
399 − {:noreply, state}
348 + if status_code == 200 do
349 + {:noreply, state}
350 + else
351 + refs = state[:refs]
352 + r = state[:refs][resp_id]
353 + r = r |> Map.put(:write, false)
354 + refs = refs |> Map.put(resp_id, r)
355 + {:noreply, state |> Map.put(:refs, refs)}
356 + end
357 + end
400 358
401 − %HTTPoison.AsyncChunk{chunk: chunk, id: resp_id} ->
402 − IO.binwrite(state[:refs][resp_id][:fd], chunk)
403 − {:noreply, state}
359 + def handle_info(%HTTPoison.AsyncHeaders{headers: headers, id: resp_id}, state) do
360 + # IO.inspect(headers)
361 + {:noreply, state}
362 + end
404 363
405 − %HTTPoison.AsyncEnd{id: resp_id} ->
406 − {_, str} = StringIO.contents(state[:refs][resp_id][:fd])
407 − StringIO.close(state[:refs][resp_id][:fd])
408 − t = state[:refs][resp_id][:type]
364 + def handle_info(%HTTPoison.AsyncChunk{chunk: chunk, id: resp_id}, state) do
365 + IO.binwrite(state[:refs][resp_id][:fd], chunk)
366 + {:noreply, state}
367 + end
409 368
410 − case Jason.decode(str) do
411 − {:ok, decoded} ->
412 − case state[:refs][resp_id] |> Map.get(:write) do
413 − false ->
414 − {:noreply, state}
369 + def handle_info(%HTTPoison.AsyncEnd{id: resp_id}, state) do
370 + {_, str} = StringIO.contents(state[:refs][resp_id][:fd])
371 + StringIO.close(state[:refs][resp_id][:fd])
372 + t = state[:refs][resp_id][:type]
415 373
416 − nil ->
417 − case t do
418 − :projects ->
419 − filtered = filter_watched(decoded, state.inst)
374 + case Jason.decode(str) do
375 + {:ok, decoded} ->
376 + case state[:refs][resp_id] |> Map.get(:write) do
377 + false ->
378 + {:noreply, state}
420 379
421 − {:noreply,
422 − state
423 − |> Map.put(:available_projects, decoded)
424 − |> Map.put(:projects, filtered)}
380 + nil ->
381 + {:noreply, store_decoded(state, t, state[:refs][resp_id][:id], decoded)}
382 + end
425 383
426 − x when x in [:commits, :jobs] ->
427 − c = state |> Map.get(t)
384 + {:error, error} ->
385 + {:noreply, state}
386 + end
387 + end
428 388
429 − i = state[:refs][resp_id][:id]
430 − c = c |> Map.put(i, decoded)
389 + defp store_decoded(state, :projects, _id, decoded) do
390 + filtered = filter_watched(decoded, state.inst)
431 391
432 − {:noreply, state |> Map.put(t, c)}
433 − end
434 − end
392 + state
393 + |> Map.put(:available_projects, decoded)
394 + |> Map.put(:projects, filtered)
395 + end
435 396
436 − {:error, error} ->
437 − {:noreply, state}
438 − end
439 − end
397 + defp store_decoded(state, type, id, decoded) when type in [:commits, :jobs] do
398 + c = state |> Map.get(type) |> Map.put(id, decoded)
399 + state |> Map.put(type, c)
440 400 end
441 401
442 402 defp via_tuple(name), do: {:via, Registry, {@registry, name}}
modified apps/room_hermes/lib/room_hermes/mail/imap.ex
+29 −23
@@ -127,17 +127,7 @@ defmodule RoomHermes.Mail.IMAP do
127 127 uids =
128 128 lines
129 129 |> Enum.filter(&String.match?(&1, ~r/^\* SEARCH/i))
130 − |> Enum.flat_map(fn line ->
131 − line
132 − |> String.replace(~r/^\* SEARCH/i, "")
133 − |> String.split(~r/\s+/, trim: true)
134 − |> Enum.flat_map(fn tok ->
135 − case Integer.parse(tok) do
136 − {n, ""} -> [n]
137 − _ -> []
138 − end
139 − end)
140 − end)
130 + |> Enum.flat_map(&search_uids/1)
141 131 |> Enum.sort()
142 132
143 133 {:ok, conn, uids}
@@ -147,6 +137,18 @@ defmodule RoomHermes.Mail.IMAP do
147 137 end
148 138 end
149 139
140 + defp search_uids(line) do
141 + line
142 + |> String.replace(~r/^\* SEARCH/i, "")
143 + |> String.split(~r/\s+/, trim: true)
144 + |> Enum.flat_map(fn tok ->
145 + case Integer.parse(tok) do
146 + {n, ""} -> [n]
147 + _ -> []
148 + end
149 + end)
150 + end
151 +
150 152 @doc """
151 153 Fetch one message's raw source without marking it read.
152 154 """
@@ -371,19 +373,23 @@ defmodule RoomHermes.Mail.IMAP do
371 373 {:error, String.trim(trimmed)}
372 374
373 375 true ->
374 − # A line ending in {N} announces exactly N bytes of raw data that
375 − # are not line-delimited and may contain anything, including CRLFs.
376 − case Regex.run(~r/\{(\d+)\}$/, trimmed) do
377 − [_, count] ->
378 − case read_literal(conn, String.to_integer(count)) do
379 − {:ok, literal} -> collect(conn, tag, [{:literal, literal}, trimmed | acc])
380 − {:error, reason} -> {:error, reason}
381 − end
382 −
383 − nil ->
384 − collect(conn, tag, [trimmed | acc])
385 − end
376 + collect_untagged(conn, tag, trimmed, acc)
377 + end
378 + end
379 + end
380 +
381 + # A line ending in {N} announces exactly N bytes of raw data that
382 + # are not line-delimited and may contain anything, including CRLFs.
383 + defp collect_untagged(conn, tag, trimmed, acc) do
384 + case Regex.run(~r/\{(\d+)\}$/, trimmed) do
385 + [_, count] ->
386 + case read_literal(conn, String.to_integer(count)) do
387 + {:ok, literal} -> collect(conn, tag, [{:literal, literal}, trimmed | acc])
388 + {:error, reason} -> {:error, reason}
386 389 end
390 +
391 + nil ->
392 + collect(conn, tag, [trimmed | acc])
387 393 end
388 394 end
389 395
modified apps/room_hermes/lib/room_hermes/mail/imap_worker.ex
+21 −17
@@ -137,23 +137,7 @@ defmodule RoomHermes.Mail.ImapWorker do
137 137 {:ok, 0}
138 138
139 139 {:ok, conn, uids} ->
140 − selected = Enum.take(uids, @max_per_cycle)
141 −
142 − if length(uids) > @max_per_cycle do
143 − Logger.info(
144 − "Mailbox::#{source.id} #{length(uids)} unread, handling #{@max_per_cycle} this cycle"
145 − )
146 − end
147 −
148 − {_conn, n} =
149 − Enum.reduce(selected, {conn, 0}, fn uid, {conn, count} ->
150 − case handle_uid(conn, uid, source, consumers) do
151 − {:ok, conn} -> {conn, count + 1}
152 − {:skip, conn} -> {conn, count}
153 − end
154 − end)
155 −
156 − {:ok, n}
140 + handle_unseen(conn, uids, source, consumers)
157 141
158 142 {:error, reason} ->
159 143 {:error, reason}
@@ -165,6 +149,26 @@ defmodule RoomHermes.Mail.ImapWorker do
165 149 end
166 150 end
167 151
152 + defp handle_unseen(conn, uids, source, consumers) do
153 + selected = Enum.take(uids, @max_per_cycle)
154 +
155 + if length(uids) > @max_per_cycle do
156 + Logger.info(
157 + "Mailbox::#{source.id} #{length(uids)} unread, handling #{@max_per_cycle} this cycle"
158 + )
159 + end
160 +
161 + {_conn, n} =
162 + Enum.reduce(selected, {conn, 0}, fn uid, {conn, count} ->
163 + case handle_uid(conn, uid, source, consumers) do
164 + {:ok, conn} -> {conn, count + 1}
165 + {:skip, conn} -> {conn, count}
166 + end
167 + end)
168 +
169 + {:ok, n}
170 + end
171 +
168 172 @doc """
169 173 Routing designators belonging to sources that reference this mailbox.
170 174
modified apps/room_hermes/lib/room_hermes_web/controllers/resource_controller.ex
+17 −27
@@ -13,35 +13,25 @@ defmodule RoomHermesWeb.ResourceController do
13 13 "name" => name,
14 14 "permission" => permission
15 15 }) do
16 − case {vhost, resource} do
17 − {"/", "queue"} ->
18 − IO.puts("qqq")
19 −
20 − case name |> String.starts_with?("mqtt-subscription") do
21 − true -> send_resp(conn, :ok, "allow")
22 − false -> send_resp(conn, :ok, "deny")
23 − end
24 −
25 − {"/", "exchange"} ->
26 − case name do
27 − # Write is allowed at the exchange so a client can talk back --
28 − # a phone reporting where it is, for a foci that travels with it.
29 − # What it may write is not decided here: the topic check that
30 − # follows confines a client to its own uplink prefix, and this
31 − # would be far too broad on its own.
32 − "amq.topic" ->
33 − case permission do
34 − "read" -> send_resp(conn, :ok, "allow")
35 − "write" -> send_resp(conn, :ok, "allow")
36 − _otherwise -> send_resp(conn, :ok, "deny")
37 − end
16 + send_resp(conn, :ok, decide(vhost, resource, name, permission))
17 + end
38 18
39 − _otherwise ->
40 − send_resp(conn, :ok, "deny")
41 − end
19 + defp decide("/", "queue", name, _permission) do
20 + IO.puts("qqq")
42 21
43 − _otherwise ->
44 − send_resp(conn, :ok, "deny")
22 + case name |> String.starts_with?("mqtt-subscription") do
23 + true -> "allow"
24 + false -> "deny"
45 25 end
46 26 end
27 +
28 + # Write is allowed at the exchange so a client can talk back --
29 + # a phone reporting where it is, for a foci that travels with it.
30 + # What it may write is not decided here: the topic check that
31 + # follows confines a client to its own uplink prefix, and this
32 + # would be far too broad on its own.
33 + defp decide("/", "exchange", "amq.topic", permission) when permission in ["read", "write"],
34 + do: "allow"
35 +
36 + defp decide(_vhost, _resource, _name, _permission), do: "deny"
47 37 end
modified apps/room_packages/lib/room_packages/package_worker.ex
+26 −22
@@ -59,28 +59,32 @@ defmodule RoomPackages.Worker do
59 59
60 60 pending
61 61 |> Enum.chunk_every(RoomPackages.USPS.max_batch())
62 − |> Enum.each(fn chunk ->
63 − numbers = Enum.map(chunk, & &1.number)
64 −
65 − case RoomPackages.USPS.track_many(source, numbers) do
66 − {:ok, summaries} ->
67 − # Re-read per write: update_source_meta_tracking/3 builds the new
68 − # list from the struct it is handed.
69 − Enum.reduce(summaries, source, fn summary, src ->
70 − existing = Enum.find(src.meta.tracking, &(&1.number == summary.number))
71 −
72 − if existing && changed?(existing, summary) do
73 − Configuration.update_source_meta_tracking(src, summary.number, summary)
74 − Configuration.get_source!(src.id)
75 − else
76 − src
77 − end
78 − end)
79 −
80 − {:error, reason} ->
81 − Logger.debug("USPS batch of #{length(numbers)}: #{inspect(reason)}")
82 − end
83 − end)
62 + |> Enum.each(&refresh_chunk(source, &1))
63 + end
64 + end
65 +
66 + defp refresh_chunk(source, chunk) do
67 + numbers = Enum.map(chunk, & &1.number)
68 +
69 + case RoomPackages.USPS.track_many(source, numbers) do
70 + {:ok, summaries} ->
71 + # Re-read per write: update_source_meta_tracking/3 builds the new
72 + # list from the struct it is handed.
73 + Enum.reduce(summaries, source, &record_summary/2)
74 +
75 + {:error, reason} ->
76 + Logger.debug("USPS batch of #{length(numbers)}: #{inspect(reason)}")
77 + end
78 + end
79 +
80 + defp record_summary(summary, src) do
81 + existing = Enum.find(src.meta.tracking, &(&1.number == summary.number))
82 +
83 + if existing && changed?(existing, summary) do
84 + Configuration.update_source_meta_tracking(src, summary.number, summary)
85 + Configuration.get_source!(src.id)
86 + else
87 + src
84 88 end
85 89 end
86 90
modified apps/room_pollen/lib/room_pollen/worker.ex
+21 −21
@@ -95,21 +95,8 @@ defmodule RoomPollen.Worker do
95 95 GenServer.cast(self(), {:add_target, {lat, lng, days}})
96 96
97 97 case Map.get(state.forecasts, {lat, lng, days}) do
98 − nil ->
99 − # Lazy fetch on first read
100 − case creds(state.inst) do
101 − {:ok, key} ->
102 − case fetch_forecast(key, lat, lng, days) do
103 − {:ok, daily_info} -> daily_info
104 − :error -> []
105 − end
106 −
107 − :no_creds ->
108 − []
109 − end
110 −
111 − cached ->
112 − cached
98 + nil -> fetch_on_read(state.inst, lat, lng, days)
99 + cached -> cached
113 100 end
114 101 end
115 102
@@ -118,6 +105,20 @@ defmodule RoomPollen.Worker do
118 105
119 106 def handle_call(_msg, _from, state), do: {:reply, :ok, state}
120 107
108 + # Lazy fetch on first read
109 + defp fetch_on_read(inst, lat, lng, days) do
110 + case creds(inst) do
111 + {:ok, key} ->
112 + case fetch_forecast(key, lat, lng, days) do
113 + {:ok, daily_info} -> daily_info
114 + :error -> []
115 + end
116 +
117 + :no_creds ->
118 + []
119 + end
120 + end
121 +
121 122 # Helpers
122 123 defp via(name), do: via_tuple(name)
123 124 defp via_tuple(name), do: {:via, Registry, {@registry, name}}
@@ -134,12 +135,7 @@ defmodule RoomPollen.Worker do
134 135 RoomSanctum.Configuration.place_for!(query) ||
135 136 Map.get(query, :foci_id) || Map.get(query, "foci_id")
136 137
137 − days =
138 − case Map.get(query, :days) || Map.get(query, "days") do
139 − n when is_integer(n) and n in 1..5 -> n
140 − n when is_binary(n) -> String.to_integer(n) |> max(1) |> min(5)
141 − _ -> 3
142 − end
138 + days = days_from_query(Map.get(query, :days) || Map.get(query, "days"))
143 139
144 140 case resolve_foci(anchor) do
145 141 {lat, lng} -> {lat, lng, days}
@@ -147,6 +143,10 @@ defmodule RoomPollen.Worker do
147 143 end
148 144 end
149 145
146 + defp days_from_query(n) when is_integer(n) and n in 1..5, do: n
147 + defp days_from_query(n) when is_binary(n), do: String.to_integer(n) |> max(1) |> min(5)
148 + defp days_from_query(_), do: 3
149 +
150 150 defp resolve_foci(nil), do: nil
151 151
152 152 # A Plani asks from wherever its client is, which is not a foci and has no

Parents: 68a501b