handle_info({cable_transport, ConnPid, Data}, State = #state{event_handler = Handler, connections = Conns}) ->

handle_info(
    {cable_transport, ConnPid, Data}, State = #state{event_handler = Handler, connections = Conns}
) ->

            io:format("[Transport] Received ~p bytes from ~p (conn: ~p)~n", [byte_size(Data), PeerAddr, ConnPid]),

            io:format("[Transport] Received ~p bytes from ~p (conn: ~p)~n", [
                byte_size(Data), PeerAddr, ConnPid
            ]),

            Result = try
                % Set the controlling process to the transport gen_server
                % so it receives {cable_transport, ConnPid, Data} messages
                ok = enoise_cable:controlling_process(ConnPid, TransportPid),
                % Get the peer address from the connection
                {ok, {PeerIP, PeerPort}} = enoise_cable:peername(ConnPid),
                PeerAddr = {PeerIP, PeerPort},
                io:format("[Transport] Peer address: ~p~n", [PeerAddr]),
                TransportPid ! {new_connection, ConnPid, PeerAddr},
                ok
            catch
                Error:ErrorReason:Stacktrace ->
                    io:format(
                        "[Transport] Error setting up connection ~p: ~p:~p~n  Stacktrace: ~p~n",
                        [ConnPid, Error, ErrorReason, Stacktrace]
                    ),
                    % Close the connection on error
                    catch enoise_cable:close(ConnPid),
                    error
            end,

            Result =
                try
                    % Set the controlling process to the transport gen_server
                    % so it receives {cable_transport, ConnPid, Data} messages
                    ok = enoise_cable:controlling_process(ConnPid, TransportPid),
                    % Get the peer address from the connection
                    {ok, {PeerIP, PeerPort}} = enoise_cable:peername(ConnPid),
                    PeerAddr = {PeerIP, PeerPort},
                    io:format("[Transport] Peer address: ~p~n", [PeerAddr]),
                    TransportPid ! {new_connection, ConnPid, PeerAddr},
                    ok
                catch
                    Error:ErrorReason:Stacktrace ->
                        io:format(
                            "[Transport] Error setting up connection ~p: ~p:~p~n  Stacktrace: ~p~n",
                            [ConnPid, Error, ErrorReason, Stacktrace]
                        ),
                        % Close the connection on error
                        catch enoise_cable:close(ConnPid),
                        error
                end,

    Result = case maps:get(Peer, Lists, undefined) of
        undefined -> {ok, []};
        Channels -> {ok, Channels}
    end,

    Result =
        case maps:get(Peer, Lists, undefined) of
            undefined -> {ok, []};
            Channels -> {ok, Channels}
        end,

handle_system_messages(fetchHistory, {Channel, Limit, From}, State = #state{db = Db, peers = Peers}) ->

handle_system_messages(
    fetchHistory, {Channel, Limit, From}, State = #state{db = Db, peers = Peers}
) ->

                    ok % Not a persistent peer, no action needed

                    % Not a persistent peer, no action needed
                    ok

    io:format("[Peer] Tracking new connection ~p to ~p with metadata ~p~n", [ConnPid, NormalizedAddr, PeerMetadata]),

    io:format("[Peer] Tracking new connection ~p to ~p with metadata ~p~n", [
        ConnPid, NormalizedAddr, PeerMetadata
    ]),

            io:format("[Peer] Updated persistent peer ~s: score +10, reset backoff~n", [NormalizedAddr]);

            io:format("[Peer] Updated persistent peer ~s: score +10, reset backoff~n", [
                NormalizedAddr
            ]);

            ok % Not a persistent peer, no action needed

            % Not a persistent peer, no action needed
            ok

            io:format("[Peer] Sent channel list request (~p) to conn ~p~n", [hex:bin_to_hexstr(ReqId), ConnPid]),

            io:format("[Peer] Sent channel list request (~p) to conn ~p~n", [
                hex:bin_to_hexstr(ReqId), ConnPid
            ]),

            io:format("[Peer] Failed to send channel list request to conn ~p: ~p~n", [ConnPid, Reason]),

            io:format("[Peer] Failed to send channel list request to conn ~p: ~p~n", [
                ConnPid, Reason
            ]),

                            {ok, {_, _, Size}} = send_hash_response(State, ConnPid, ReqId, [PostHash]),

                            {ok, {_, _, Size}} = send_hash_response(State, ConnPid, ReqId, [
                                PostHash
                            ]),

                            {_OrigConnPid, OrigMsg} = maps:get(ReqId, ActiveOut, {ConnPid, undefined}),
                            OrigMsgType = case OrigMsg of
                                undefined -> unknown;
                                [OrigHeader, _OrigBody] ->
                                    proplists:get_value(msgType, OrigHeader, unknown);
                                _ -> unknown
                            end,

                            {_OrigConnPid, OrigMsg} = maps:get(
                                ReqId, ActiveOut, {ConnPid, undefined}
                            ),
                            OrigMsgType =
                                case OrigMsg of
                                    undefined ->
                                        unknown;
                                    [OrigHeader, _OrigBody] ->
                                        proplists:get_value(msgType, OrigHeader, unknown);
                                    _ ->
                                        unknown
                                end,

                            {HashesToFetch, HashesToStore, Channel} = case OrigMsgType of
                                5 -> % channel_state_request - fetch all eagerly
                                    io:format("[Peer] Hash response for channel_state_request: fetching all ~p hashes eagerly~n", [length(NewHashes)]),
                                    {NewHashes, [], undefined};
                                4 -> % channel_time_range_request - hybrid strategy
                                    Chan = case OrigMsg of
                                        [_OH, OrigBody] -> proplists:get_value(channel, OrigBody, undefined);
                                        _ -> undefined
                                    end,
                                    EagerCount = min(100, length(NewHashes)),
                                    Eager = lists:sublist(NewHashes, EagerCount),
                                    Lazy = lists:sublist(NewHashes, EagerCount + 1, length(NewHashes)),
                                    io:format("[Peer] Hash response for channel_time_range_request (~s): fetching ~p hashes eagerly, storing ~p as unfetched~n",
                                        [Chan, length(Eager), length(Lazy)]),
                                    {Eager, Lazy, Chan};
                                _ -> % unknown - default to eager
                                    io:format("[Peer] Hash response for unknown request type: fetching all ~p hashes eagerly (fallback)~n", [length(NewHashes)]),
                                    {NewHashes, [], undefined}
                            end,

                            {HashesToFetch, HashesToStore, Channel} =
                                case OrigMsgType of
                                    % channel_state_request - fetch all eagerly
                                    5 ->
                                        io:format(
                                            "[Peer] Hash response for channel_state_request: fetching all ~p hashes eagerly~n",
                                            [length(NewHashes)]
                                        ),
                                        {NewHashes, [], undefined};
                                    % channel_time_range_request - hybrid strategy
                                    4 ->
                                        Chan =
                                            case OrigMsg of
                                                [_OH, OrigBody] ->
                                                    proplists:get_value(
                                                        channel, OrigBody, undefined
                                                    );
                                                _ ->
                                                    undefined
                                            end,
                                        EagerCount = min(100, length(NewHashes)),
                                        Eager = lists:sublist(NewHashes, EagerCount),
                                        Lazy = lists:sublist(
                                            NewHashes, EagerCount + 1, length(NewHashes)
                                        ),
                                        io:format(
                                            "[Peer] Hash response for channel_time_range_request (~s): fetching ~p hashes eagerly, storing ~p as unfetched~n",
                                            [Chan, length(Eager), length(Lazy)]
                                        ),
                                        {Eager, Lazy, Chan};
                                    % unknown - default to eager
                                    _ ->
                                        io:format(
                                            "[Peer] Hash response for unknown request type: fetching all ~p hashes eagerly (fallback)~n",
                                            [length(NewHashes)]
                                        ),
                                        {NewHashes, [], undefined}
                                end,

                                [] -> ok;

                                [] ->
                                    ok;

                                    gen_server:call(Db, {unfetchedAdd, HashesToStore, Channel, "hash_response"});

                                    gen_server:call(
                                        Db, {unfetchedAdd, HashesToStore, Channel, "hash_response"}
                                    );

                                    io:format("[Peer] Warning: Cannot store unfetched hashes without channel info~n"),

                                    io:format(
                                        "[Peer] Warning: Cannot store unfetched hashes without channel info~n"
                                    ),

                                    UpdatedOut = maps:put(PostReqId, {ConnPid, post_request}, ActiveOut),

                                    UpdatedOut = maps:put(
                                        PostReqId, {ConnPid, post_request}, ActiveOut
                                    ),

                        [] -> ok;

                        [] ->
                            ok;

                            io:format("[Peer] Removed ~p fetched hashes from unfetched table~n", [length(FetchedHashes)])

                            io:format("[Peer] Removed ~p fetched hashes from unfetched table~n", [
                                length(FetchedHashes)
                            ])

                    Channels = lists:filtermap(fun(P) ->
                        [_PostHeader, PostBody] = posts:decode(P),
                        case proplists:get_value(channel, PostBody, undefined) of
                            undefined -> false;
                            Chan -> {true, Chan}
                        end
                    end, NewPosts),

                    Channels = lists:filtermap(
                        fun(P) ->
                            [_PostHeader, PostBody] = posts:decode(P),
                            case proplists:get_value(channel, PostBody, undefined) of
                                undefined -> false;
                                Chan -> {true, Chan}
                            end
                        end,
                        NewPosts
                    ),

    BaseDelayMs = 5000,  % 5 seconds
    MaxDelayMs = 300000, % 5 minutes

    % 5 seconds
    BaseDelayMs = 5000,
    % 5 minutes
    MaxDelayMs = 300000,

    Target = case string:split(Address, ":") of
        [HostStr, PortStr] ->
            try
                Port = list_to_integer(PortStr),
                case inet:getaddr(HostStr, inet) of
                    {ok, IP} -> {IP, Port};
                    _ -> undefined
                end
            catch
                _:_ -> undefined
            end;
        _ -> undefined
    end,

    Target =
        case string:split(Address, ":") of
            [HostStr, PortStr] ->
                try
                    Port = list_to_integer(PortStr),
                    case inet:getaddr(HostStr, inet) of
                        {ok, IP} -> {IP, Port};
                        _ -> undefined
                    end
                catch
                    _:_ -> undefined
                end;
            _ ->
                undefined
        end,

        undefined -> false;

        undefined ->
            false;

            io:format("[Peer] Updating event handler ~p interval to ~pms~n", [HandlerPid, IntervalMs]),

            io:format("[Peer] Updating event handler ~p interval to ~pms~n", [
                HandlerPid, IntervalMs
            ]),

            io:format("[Peer] Registered event handler ~p with interval ~pms~n", [HandlerPid, IntervalMs]),

            io:format("[Peer] Registered event handler ~p with interval ~pms~n", [
                HandlerPid, IntervalMs
            ]),

                    ok; % Nothing to notify

                    % Nothing to notify
                    ok;

                    lists:foreach(fun(Chan) ->
                        try
                            HandlerPid ! {channel_event, {new_messages, Chan}}
                        catch
                            _:_ ->
                                io:format("[Peer] Failed to send event to handler ~p~n", [HandlerPid])
                        end
                    end, PendingList),
                    io:format("[Peer] Notified handler ~p of ~p channels~n", [HandlerPid, length(PendingList)])

                    lists:foreach(
                        fun(Chan) ->
                            try
                                HandlerPid ! {channel_event, {new_messages, Chan}}
                            catch
                                _:_ ->
                                    io:format("[Peer] Failed to send event to handler ~p~n", [
                                        HandlerPid
                                    ])
                            end
                        end,
                        PendingList
                    ),
                    io:format("[Peer] Notified handler ~p of ~p channels~n", [
                        HandlerPid, length(PendingList)
                    ])

            State; % No handlers, nothing to do

            % No handlers, nothing to do
            State;

            NewHandlers = maps:map(fun(_Pid, Handler) ->
                Pending = maps:get(pending, Handler),
                NewPending = lists:foldl(fun sets:add_element/2, Pending, Channels),
                Handler#{pending => NewPending}
            end, Handlers),

            NewHandlers = maps:map(
                fun(_Pid, Handler) ->
                    Pending = maps:get(pending, Handler),
                    NewPending = lists:foldl(fun sets:add_element/2, Pending, Channels),
                    Handler#{pending => NewPending}
                end,
                Handlers
            ),

                    ok; % Already connected, skip

                    % Already connected, skip
                    ok;

                            ok % Backoff period not elapsed, skip

                            % Backoff period not elapsed, skip
                            ok

            io:format(user, "[enoise_cable connection ~p] Received tcp_closed for unknown socket~n", [self()]),

            io:format(
                user, "[enoise_cable connection ~p] Received tcp_closed for unknown socket~n", [
                    self()
                ]
            ),

            io:format(user, "[enoise_cable connection ~p] Received tcp_error for unknown socket~n", [self()]),

            io:format(
                user, "[enoise_cable connection ~p] Received tcp_error for unknown socket~n", [
                    self()
                ]
            ),

            ChannelBin = case ChannelData of
                {blob, B} -> B;
                B when is_binary(B) -> B
            end,

            ChannelBin =
                case ChannelData of
                    {blob, B} -> B;
                    B when is_binary(B) -> B
                end,

    Res = case sqlite3:write(Db, peers, Row) of
        {rowid, RowId} -> {ok, RowId};
        {error, 19, _Msg} -> {error, already_exists}
    end,

    Res =
        case sqlite3:write(Db, peers, Row) of
            {rowid, RowId} -> {ok, RowId};
            {error, 19, _Msg} -> {error, already_exists}
        end,

    Res = case sqlite3:update(Db, peers, {address, Address}, Updates) of
        ok -> ok;
        Error -> {error, Error}
    end,

    Res =
        case sqlite3:update(Db, peers, {address, Address}, Updates) of
            ok -> ok;
            Error -> {error, Error}
        end,

                score => case Score of null -> 0; S -> S end,
                last_seen => case LastSeen of null -> 0; LS -> LS end,
                last_attempt => case LastAttempt of null -> 0; LA -> LA end,
                attempt_count => case AttemptCount of null -> 0; AC -> AC end,

                score =>
                    case Score of
                        null -> 0;
                        S -> S
                    end,
                last_seen =>
                    case LastSeen of
                        null -> 0;
                        LS -> LS
                    end,
                last_attempt =>
                    case LastAttempt of
                        null -> 0;
                        LA -> LA
                    end,
                attempt_count =>
                    case AttemptCount of
                        null -> 0;
                        AC -> AC
                    end,

                notes => case Notes of
                    null -> undefined;
                    N -> binary_to_list(N)
                end

                notes =>
                    case Notes of
                        null -> undefined;
                        N -> binary_to_list(N)
                    end

    Result = case sqlite3:read(Db, peers, {address, Address}) of
        [{columns, _Cols}, {rows, [{Id, Addr, Score, LastSeen, LastAttempt, AttemptCount, CreatedAt, Notes}]}] ->
            {ok, #{
                id => Id,
                address => binary_to_list(Addr),
                score => case Score of null -> 0; S -> S end,
                last_seen => case LastSeen of null -> 0; LS -> LS end,
                last_attempt => case LastAttempt of null -> 0; LA -> LA end,
                attempt_count => case AttemptCount of null -> 0; AC -> AC end,
                created_at => CreatedAt,
                notes => case Notes of
                    null -> undefined;
                    N -> binary_to_list(N)
                end
            }};
        _ ->
            not_found
    end,

    Result =
        case sqlite3:read(Db, peers, {address, Address}) of
            [
                {columns, _Cols},
                {rows, [{Id, Addr, Score, LastSeen, LastAttempt, AttemptCount, CreatedAt, Notes}]}
            ] ->
                {ok, #{
                    id => Id,
                    address => binary_to_list(Addr),
                    score =>
                        case Score of
                            null -> 0;
                            S -> S
                        end,
                    last_seen =>
                        case LastSeen of
                            null -> 0;
                            LS -> LS
                        end,
                    last_attempt =>
                        case LastAttempt of
                            null -> 0;
                            LA -> LA
                        end,
                    attempt_count =>
                        case AttemptCount of
                            null -> 0;
                            AC -> AC
                        end,
                    created_at => CreatedAt,
                    notes =>
                        case Notes of
                            null -> undefined;
                            N -> binary_to_list(N)
                        end
                }};
            _ ->
                not_found
        end,

    Query = "SELECT hash FROM unfetched_hashes WHERE channel = ? ORDER BY discovered_at ASC LIMIT ?",

    Query =
        "SELECT hash FROM unfetched_hashes WHERE channel = ? ORDER BY discovered_at ASC LIMIT ?",

    Schema = <<"
create table IF NOT EXISTS posts (
  -- these exist for all posts
  --
  id integer primary key AUTOINCREMENT,
  hash blob unique not null check (length(hash) = 32),
  type integer not null check (type >= 0 AND type <= 5 ),
  -- milliseconds since UNIX Epoch
  timestamp integer not null check (timestamp > 0),
  user_id integer not null,
  raw_post blob not null,
  -- depends on postType
  channel text,          -- all but delete have this one
  -- join & leave don't have other fields then channel
  text text,
  topic text,
  deletedHashes blob,    -- concatenated
  infos blob,            -- encoded as (json?) map {k=>v, k=>v}
  -- `post/text`  | a textual chat message, posted to a channel |
  CONSTRAINT is_text check (
    type != 0 OR (channel is not null AND text is not null)
  ),
  -- `post/delete`| the deletion of a previously published post |
  CONSTRAINT is_delete check (
    type != 1 OR (deletedHashes is not null)
  ),
  -- `post/info`  | set or clear informative key/value pairs on a user
  CONSTRAINT is_info check (
    type != 2 OR (infos is not null)
  ),
  -- `post/topic` | set or clear a channel's topic string
  CONSTRAINT is_topic check (
    type != 3 OR (channel is not null AND topic is not null)
  ),
  -- `post/join`  | announce membership to a channel
  CONSTRAINT is_join check (
    type != 4 OR (channel is not null)
  ),
  -- `post/leave` | announce cessation of membership to a channel
  CONSTRAINT is_leave check (
    type != 5 OR (channel is not null)
  ),
  foreign key(user_id) references users(id)
);
create table IF NOT EXISTS links (
    channel text,
    source blob not null,
    parent blob not null
    -- we will get posts for which we don't have the parent's (yet)
    -- foreign key(source) references posts(hash),
    -- foreign key(parent) references posts(hash),
);
CREATE INDEX IF NOT EXISTS parent_idx ON links(parent);
create table IF NOT EXISTS channel_members (
    id integer primary key AUTOINCREMENT,
    channel text not null,
    user_id integer not null,
    last_hash blob not null,
    UNIQUE(channel, user_id)
);
">>,

    Schema =
        <<"\n"
        "create table IF NOT EXISTS posts (\n"
        "  -- these exist for all posts\n"
        "  --\n"
        "  id integer primary key AUTOINCREMENT,\n"
        "  hash blob unique not null check (length(hash) = 32),\n"
        "  type integer not null check (type >= 0 AND type <= 5 ),\n"
        "  -- milliseconds since UNIX Epoch\n"
        "  timestamp integer not null check (timestamp > 0),\n"
        "\n"
        "  user_id integer not null,\n"
        "\n"
        "  raw_post blob not null,\n"
        "\n"
        "\n"
        "  -- depends on postType\n"
        "  channel text,          -- all but delete have this one\n"
        "  -- join & leave don't have other fields then channel\n"
        "  text text,\n"
        "  topic text,\n"
        "  deletedHashes blob,    -- concatenated\n"
        "  infos blob,            -- encoded as (json?) map {k=>v, k=>v}\n"
        "\n"
        "  -- `post/text`  | a textual chat message, posted to a channel |\n"
        "  CONSTRAINT is_text check (\n"
        "    type != 0 OR (channel is not null AND text is not null)\n"
        "  ),\n"
        "  -- `post/delete`| the deletion of a previously published post |\n"
        "  CONSTRAINT is_delete check (\n"
        "    type != 1 OR (deletedHashes is not null)\n"
        "  ),\n"
        "  -- `post/info`  | set or clear informative key/value pairs on a user\n"
        "  CONSTRAINT is_info check (\n"
        "    type != 2 OR (infos is not null)\n"
        "  ),\n"
        "  -- `post/topic` | set or clear a channel's topic string\n"
        "  CONSTRAINT is_topic check (\n"
        "    type != 3 OR (channel is not null AND topic is not null)\n"
        "  ),\n"
        "  -- `post/join`  | announce membership to a channel\n"
        "  CONSTRAINT is_join check (\n"
        "    type != 4 OR (channel is not null)\n"
        "  ),\n"
        "  -- `post/leave` | announce cessation of membership to a channel\n"
        "  CONSTRAINT is_leave check (\n"
        "    type != 5 OR (channel is not null)\n"
        "  ),\n"
        "\n"
        "  foreign key(user_id) references users(id)\n"
        ");\n"
        "\n"
        "create table IF NOT EXISTS links (\n"
        "    channel text,\n"
        "    source blob not null,\n"
        "    parent blob not null\n"
        "\n"
        "    -- we will get posts for which we don't have the parent's (yet)\n"
        "    -- foreign key(source) references posts(hash),\n"
        "    -- foreign key(parent) references posts(hash),\n"
        ");\n"
        "CREATE INDEX IF NOT EXISTS parent_idx ON links(parent);\n"
        "\n"
        "create table IF NOT EXISTS channel_members (\n"
        "    id integer primary key AUTOINCREMENT,\n"
        "    channel text not null,\n"
        "    user_id integer not null,\n"
        "    last_hash blob not null,\n"
        "    UNIQUE(channel, user_id)\n"
        ");\n">>,

        intensity => 3,  % Max 3 restarts
        period => 10     % Within 10 seconds

        % Max 3 restarts
        intensity => 3,
        % Within 10 seconds
        period => 10

            timer:sleep(100)  % Small delay between joins

            % Small delay between joins
            timer:sleep(100)

        0 -> ok;

        0 ->
            ok;

            timer:sleep(100)  % Small delay between connections

            % Small delay between connections
            timer:sleep(100)

        0 -> ok;

        0 ->
            ok;