From 9bde981c44e9bdfb459d8e1a031c5462ee563daa Mon Sep 17 00:00:00 2001
From: Ilya Averyanov <av@rubybox.dev>
Date: Fri, 28 Jun 2024 20:30:26 +0300
Subject: [PATCH] feat(queue): fix static check issues

---
 apps/emqx/priv/bpapi.versions                                | 1 +
 .../emqx_persistent_session_ds_shared_subs.erl               | 2 +-
 .../emqx_persistent_session_ds_shared_subs_null_agent.erl    | 4 ++--
 apps/emqx_ds_shared_sub/src/emqx_ds_shared_sub_group_sm.erl  | 2 +-
 apps/emqx_ds_shared_sub/src/emqx_ds_shared_sub_leader.erl    | 2 +-
 apps/emqx_ds_shared_sub/src/emqx_ds_shared_sub_proto.erl     | 5 +++--
 .../src/proto/emqx_ds_shared_sub_proto_v1.erl                | 3 ++-
 7 files changed, 11 insertions(+), 8 deletions(-)

diff --git a/apps/emqx/priv/bpapi.versions b/apps/emqx/priv/bpapi.versions
index bc3e4f1a2..02dd84f03 100644
--- a/apps/emqx/priv/bpapi.versions
+++ b/apps/emqx/priv/bpapi.versions
@@ -27,6 +27,7 @@
 {emqx_ds,2}.
 {emqx_ds,3}.
 {emqx_ds,4}.
+{emqx_ds_shared_sub,1}.
 {emqx_eviction_agent,1}.
 {emqx_eviction_agent,2}.
 {emqx_eviction_agent,3}.
diff --git a/apps/emqx/src/emqx_persistent_session_ds/emqx_persistent_session_ds_shared_subs.erl b/apps/emqx/src/emqx_persistent_session_ds/emqx_persistent_session_ds_shared_subs.erl
index bf0798e1a..ad00fadbd 100644
--- a/apps/emqx/src/emqx_persistent_session_ds/emqx_persistent_session_ds_shared_subs.erl
+++ b/apps/emqx/src/emqx_persistent_session_ds/emqx_persistent_session_ds_shared_subs.erl
@@ -364,7 +364,7 @@ agent_opts(#{session_id := SessionId}) ->
 now_ms() ->
     erlang:system_time(millisecond).
 
-is_use_finished(S, #srs{unsubscribed = Unsubscribed}) ->
+is_use_finished(_S, #srs{unsubscribed = Unsubscribed}) ->
     Unsubscribed.
 
 is_stream_fully_acked(S, SRS) ->
diff --git a/apps/emqx/src/emqx_persistent_session_ds/emqx_persistent_session_ds_shared_subs_null_agent.erl b/apps/emqx/src/emqx_persistent_session_ds/emqx_persistent_session_ds_shared_subs_null_agent.erl
index 5bdae08da..d984194a8 100644
--- a/apps/emqx/src/emqx_persistent_session_ds/emqx_persistent_session_ds_shared_subs_null_agent.erl
+++ b/apps/emqx/src/emqx_persistent_session_ds/emqx_persistent_session_ds_shared_subs_null_agent.erl
@@ -14,7 +14,7 @@
     on_unsubscribe/2,
     on_stream_progress/2,
     on_info/2,
-    on_disconnect/1,
+    on_disconnect/2,
 
     renew_streams/1
 ]).
@@ -37,7 +37,7 @@ on_subscribe(_Agent, _TopicFilter, _SubOpts) ->
 on_unsubscribe(Agent, _TopicFilter) ->
     Agent.
 
-on_disconnect(Agent) ->
+on_disconnect(Agent, _) ->
     Agent.
 
 renew_streams(Agent) ->
diff --git a/apps/emqx_ds_shared_sub/src/emqx_ds_shared_sub_group_sm.erl b/apps/emqx_ds_shared_sub/src/emqx_ds_shared_sub_group_sm.erl
index 3932aa6ce..aab47802b 100644
--- a/apps/emqx_ds_shared_sub/src/emqx_ds_shared_sub_group_sm.erl
+++ b/apps/emqx_ds_shared_sub/src/emqx_ds_shared_sub_group_sm.erl
@@ -372,7 +372,7 @@ handle_leader_renew_stream_lease(GSM, VersionOld, VersionNew) ->
     %% Unexpected versions or state
     transition(GSM, ?connecting, #{}).
 
--spec handle_stream_progress(group_sm(), emqx_ds_shared_sub_proto:agent_stream_progress()) ->
+-spec handle_stream_progress(group_sm(), list(emqx_ds_shared_sub_proto:agent_stream_progress())) ->
     group_sm().
 handle_stream_progress(#{state := ?connecting} = GSM, _StreamProgresses) ->
     GSM;
diff --git a/apps/emqx_ds_shared_sub/src/emqx_ds_shared_sub_leader.erl b/apps/emqx_ds_shared_sub/src/emqx_ds_shared_sub_leader.erl
index ecd06846c..ce38a72f9 100644
--- a/apps/emqx_ds_shared_sub/src/emqx_ds_shared_sub_leader.erl
+++ b/apps/emqx_ds_shared_sub/src/emqx_ds_shared_sub_leader.erl
@@ -40,7 +40,7 @@
 -type agent_state() :: #{
     %% Our view of group gm's status
     %% it lags the actual state
-    state := emqx_ds_shared_sub_agent:status(),
+    state := ?waiting_replaying | ?replaying | ?waiting_updating | ?updating,
     prev_version := emqx_maybe:t(emqx_ds_shared_sub_proto:version()),
     version := emqx_ds_shared_sub_proto:version(),
     agent_metadata := emqx_ds_shared_sub_proto:agent_metadata(),
diff --git a/apps/emqx_ds_shared_sub/src/emqx_ds_shared_sub_proto.erl b/apps/emqx_ds_shared_sub/src/emqx_ds_shared_sub_proto.erl
index 53a6693b2..0b1770f3c 100644
--- a/apps/emqx_ds_shared_sub/src/emqx_ds_shared_sub_proto.erl
+++ b/apps/emqx_ds_shared_sub/src/emqx_ds_shared_sub_proto.erl
@@ -244,8 +244,9 @@ leader_invalidate(ToAgent, OfGroup) ->
 %% Internal API
 %%--------------------------------------------------------------------
 
-agent(_Id, Pid) ->
-    ?agent(_Id, Pid).
+agent(Id, Pid) ->
+    _ = Id,
+    ?agent(Id, Pid).
 
 format_streams(Streams) ->
     lists:map(
diff --git a/apps/emqx_ds_shared_sub/src/proto/emqx_ds_shared_sub_proto_v1.erl b/apps/emqx_ds_shared_sub/src/proto/emqx_ds_shared_sub_proto_v1.erl
index 117b34e98..2dfc8be65 100644
--- a/apps/emqx_ds_shared_sub/src/proto/emqx_ds_shared_sub_proto_v1.erl
+++ b/apps/emqx_ds_shared_sub/src/proto/emqx_ds_shared_sub_proto_v1.erl
@@ -14,6 +14,7 @@
     agent_connect_leader/5,
     agent_update_stream_states/5,
     agent_update_stream_states/6,
+    agent_disconnect/5,
 
     leader_lease_streams/6,
     leader_renew_stream_lease/4,
@@ -30,7 +31,7 @@ introduced_in() ->
     emqx_ds_shared_sub_proto:leader(),
     emqx_ds_shared_sub_proto:agent(),
     emqx_ds_shared_sub_proto:agent_metadata(),
-    emqx_ds_shared_sub_proto:topic_filter()
+    emqx_persistent_session_ds:share_topic_filter()
 ) -> ok.
 agent_connect_leader(Node, ToLeader, FromAgent, AgentMetadata, TopicFilter) ->
     erpc:cast(Node, emqx_ds_shared_sub_proto, agent_connect_leader, [