Elixir ATProtocol ingestion and sync library.
feat: validate message sequence
+23
-11
+23
-11
lib/drinkup.ex
+23
-11
lib/drinkup.ex
···
22
22
|> URI.append_query(URI.encode_query(%{cursor: cursor}))
23
23
|> URI.to_string()
24
24
25
-
WebSockex.start_link(url, __MODULE__, %{cursor: cursor})
25
+
WebSockex.start_link(url, __MODULE__, 0)
26
26
end
27
27
28
28
def handle_connect(conn, state) do
···
34
34
with {:ok, header, next} <- CAR.DagCbor.decode(msg),
35
35
{:ok, payload, _} <- CAR.DagCbor.decode(next),
36
36
{%{"op" => @op_regular, "t" => type}, _} <- {header, payload},
37
-
message <- from_payload(type, payload) do
37
+
true <- type == "#info" || valid_seq?(state, payload["seq"]),
38
+
message <-
39
+
from_payload(type, payload) do
38
40
case message do
39
41
%Firehose.Commit{} = commit ->
40
42
IO.inspect(commit.ops, label: commit.repo)
···
42
44
msg ->
43
45
IO.inspect(msg)
44
46
end
47
+
48
+
{:ok, payload["seq"] || state}
45
49
else
50
+
false ->
51
+
Logger.error("Got out of sequence or invalid `seq` from Firehose")
52
+
{:ok, state}
53
+
46
54
{%{"op" => @op_error, "t" => type}, payload} ->
47
55
Logger.error("Got error from Firehose: #{inspect({type, payload})}")
56
+
{:ok, state}
48
57
49
58
{:error, reason} ->
50
59
Logger.warning("Failed to decode frame from Firehose: #{inspect(reason)}")
60
+
{:ok, state}
51
61
end
52
-
53
-
{:ok, state}
54
62
end
55
63
56
64
def handle_frame({:text, msg}, state) do
57
-
Logger.warning("Got unexpected text frame from Firehose #{inspect(msg)}")
65
+
Logger.warning("Got unexpected text frame from Firehose: #{inspect(msg)}")
58
66
{:ok, state}
59
67
end
68
+
69
+
@spec valid_seq?(integer(), any()) :: boolean()
70
+
defp valid_seq?(last_seq, seq) when is_integer(seq), do: seq > last_seq
71
+
defp valid_seq?(_last_seq, _seq), do: false
60
72
61
73
@spec from_payload(String.t(), map()) ::
62
74
Firehose.Commit.t()
···
65
77
| Firehose.Account.t()
66
78
| Firehose.Info.t()
67
79
| nil
68
-
def from_payload("#commit", payload), do: Firehose.Commit.from(payload)
69
-
def from_payload("#sync", payload), do: Firehose.Sync.from(payload)
70
-
def from_payload("#identity", payload), do: Firehose.Identity.from(payload)
71
-
def from_payload("#account", payload), do: Firehose.Account.from(payload)
72
-
def from_payload("#info", payload), do: Firehose.Info.from(payload)
73
-
def from_payload(_type, _payload), do: nil
80
+
defp from_payload("#commit", payload), do: Firehose.Commit.from(payload)
81
+
defp from_payload("#sync", payload), do: Firehose.Sync.from(payload)
82
+
defp from_payload("#identity", payload), do: Firehose.Identity.from(payload)
83
+
defp from_payload("#account", payload), do: Firehose.Account.from(payload)
84
+
defp from_payload("#info", payload), do: Firehose.Info.from(payload)
85
+
defp from_payload(_type, _payload), do: nil
74
86
end