package ApmModule;
# ApmModule.pm v3.4.0
#   APM child-process connector for Perl 5.10+
#   IPC via stdin/stdout binary frames — identical wire protocol to Node.js connector.
#
#   Requires: JSON (cpan install JSON  or  apt install libjson-perl)
#
#   APM injects env vars:
#     APM=1         — confirms process is managed by APM
#     APM_INDEX     — 0-based instance index (if env_index configured)
#
#   Usage:
#     use ApmModule;
#
#     my $apm = ApmModule->new(sub {
#         my ($s) = @_;
#         $s->write("Hello World", { '_status' => 200, 'content-type' => 'text/plain' });
#         $s->close();
#     });
#     $apm->{on_channel} = sub { my ($ch, $data, $reply) = @_; $reply->({ok => 1}) if $reply; };
#     $apm->{on_stream}  = sub { my ($stream) = @_; $stream->accept({name => 'worker1'}); };
#     $apm->run();

use strict;
use warnings;
use JSON qw(encode_json decode_json);
use Scalar::Util qw(looks_like_number);
use Time::HiRes qw(time);

unless (caller()) {
    if (grep { $_ eq '-update' } @ARGV) {
        my $url  = 'https://processmanager.dev/connectors/ApmModule.pm';
        my $dest = __FILE__;
        my $tmp  = "$dest.tmp";
        my $sig  = "$tmp.sig";
        print "Updating $dest ...\n";
        # Every refusal drops both temp files, so a failed update can never
        # leave a half-written connector (or a stray signature) behind.
        my $fail = sub { unlink $tmp, $sig; print STDERR "Update failed: $_[0]"; exit 1; };
        # LWP::UserAgent (not ::Simple) so the body can be checked against
        # Content-Length — a truncated download must never be renamed over
        # the connector. The curl/wget fallbacks exit non-zero on a short
        # transfer, so their output is only trusted on exit status 0.
        # Returns (body, undef) on success, (undef, reason) otherwise.
        my $fetch = sub {
            my ($u) = @_;
            local $@;
            my $d = eval {
                require LWP::UserAgent;
                my $res = LWP::UserAgent->new(timeout => 30)->get($u);
                die "HTTP " . $res->status_line . "\n" unless $res->is_success;
                die "incomplete download\n" if $res->header('Client-Aborted') || $res->header('X-Died');
                my $clen = $res->header('Content-Length');
                die "truncated download (" . length($res->content) . " of $clen bytes)\n"
                    if defined $clen && length($res->content) != $clen;
                $res->decoded_content(charset => 'none') // $res->content;
            };
            # Only fall back to curl/wget when LWP itself is missing — an HTTP
            # error or a short body from LWP is final.
            return (undef, $@) if $@ && $@ !~ /^Can't locate LWP/;
            if (!defined $d) { $d = `curl -fsSL "$u" 2>/dev/null`; undef $d if $?; }
            if (!defined $d) { $d = `wget -qO- "$u" 2>/dev/null`; undef $d if $?; }
            return (defined $d && length $d) ? ($d, undef) : (undef, "could not download\n");
        };
        # Write to a temp file and rename, so a failed write can never
        # leave a half-written connector behind.
        my $write = sub {
            my ($path, $body) = @_;
            open(my $fh, '>', $path) or $fail->("$!\n");
            binmode $fh;
            print $fh $body;
            close $fh;
        };
        my ($data, $err) = $fetch->($url);
        $fail->($err) unless defined $data;
        $write->($tmp, $data);
        # Every published connector has a detached ed25519 signature next to
        # it at <url>.sig; without one there is nothing to verify.
        my ($sigdata) = $fetch->("$url.sig");
        $fail->("no signature published for ApmModule.pm\n") unless defined $sigdata;
        $write->($sig, $sigdata);
        # The verifier is the installed apm CLI (the release key is compiled
        # into it): APM_BIN when set, else PATH, then the two install paths.
        my @cands = length($ENV{APM_BIN} // '') ? ($ENV{APM_BIN})
            : ((map { "$_/apm" } grep { length } split(/:/, $ENV{PATH} // '')), '/usr/sbin/apm', '/usr/local/bin/apm');
        my ($apm) = grep { -f $_ && -x _ } @cands;
        $fail->("cannot verify the download — the apm CLI was not found (install APM, or set APM_BIN to its path)\n")
            unless defined $apm;
        # List form via IPC::Open3 (core) — the paths never pass through a
        # shell; stderr is folded into stdout so the verifier's line is shown.
        require IPC::Open3;
        my ($in, $vh);
        my $pid = eval { IPC::Open3::open3($in, $vh, undef, $apm, 'verify', $tmp, $sig) };
        $fail->("signature check failed — not installing\ncannot run $apm\n") unless $pid;
        close $in;
        my $out = do { local $/; <$vh> } // '';
        close $vh;
        waitpid($pid, 0);
        if ($? != 0) {
            $out =~ s/\s+\z//;
            $fail->("signature check failed — not installing\n$out\n");
        }
        # Refuse anything that does not look like this connector (an error
        # page, a different file) and keep the current one.
        $fail->("downloaded file is not ApmModule.pm\n")
            unless index(substr($data, 0, 200), 'package ApmModule;') >= 0;
        rename($tmp, $dest) or $fail->("$!\n");
        unlink $sig;
        print "Updated.\n";
        exit 0;
    }
    print STDERR "Usage: perl ApmModule.pm -update\n";
    exit 1;
}

# exit 1 (not die's 255) so the guard exit code matches every other connector.
unless ($ENV{APM}) {
    print STDERR "[APM] Must run under APM daemon (APM env var not set)\n";
    exit 1;
}

my $_INDEX = $ENV{APM_INDEX} // '0';
my $_rid_counter = 0;

sub _send_frame {
    my ($json_hdr, $binary) = @_;
    $binary //= '';
    # A UTF8-flagged payload (e.g. decoded text with non-ASCII characters)
    # must be downgraded to octets first: length() would count characters,
    # and concatenating it upgrades the whole frame, mangling the 0x05
    # header bytes ≥ 0x80 on output — a permanent frame desync.
    utf8::encode($binary) if utf8::is_utf8($binary);
    my $json   = encode_json($json_hdr);
    my $length = length($json) + 1 + length($binary);
    my $frame  = "\x05" . pack('N', $length) . $json . "\x03" . $binary;
    binmode STDOUT;
    print STDOUT $frame;
    STDOUT->flush();
}

sub _next_rid {
    $_rid_counter++;
    return "${_INDEX}_${_rid_counter}_pl";
}

# App-supplied headers for write(): plain keys become response headers; of the
# _-prefixed control fields only the ones the daemon reads on a response pass
# through. _status must be a JSON number — the daemon ignores a string ("404"
# → 200), and a scalar once used in string context encodes as one — so it is
# coerced; anything else _-prefixed (_command, _session, _sessionData…) is the
# connector's own and is dropped, so a header map can never redirect or spoof
# the frame. Each dropped key is logged once.
my %_warned_keys;

sub _warn_once {
    my ($key, $msg) = @_;
    return if $_warned_keys{$key}++;
    print STDERR "[APM] write(): $msg\n";
}

sub _resp_headers {
    my ($headers) = @_;
    my %h;
    for my $k (keys %$headers) {
        my $v = $headers->{$k};
        if ($k eq '_status') {
            if (defined $v && !ref $v && looks_like_number($v)) { $h{_status} = int($v) + 0; }
            else { _warn_once($k, 'header "_status" is not numeric (' . (defined $v ? "'$v'" : 'undef') . ') — dropped'); }
            next;
        }
        if ($k =~ /^_/ && $k ne '_headers' && $k ne '_reason') {
            _warn_once($k, "reserved header \"$k\" dropped");
            next;
        }
        $h{$k} = $v;
    }
    return \%h;
}

# ─── Session ──────────────────────────────────────────────────────────────────

package ApmSession;

sub new {
    my ($class) = @_;
    return bless {
        session_id    => '',
        session_type  => 'new',
        session_data  => {},
        protocol      => '',
        method        => undef,
        path          => '/',
        path_array    => [],
        query         => {},
        query_object  => {},
        cookies       => {},
        headers       => {},
        remote_ip     => '',
        instance_id   => $_INDEX,
        _module       => undef,   # set by ApmModule; carries session-scoped requests
        _deleted      => 0,
        _backlog      => [],
        on_data       => undef,
        on_close      => undef,
    }, $class;
}

sub active    { return !$_[0]->{_deleted} }
sub _mark_deleted { $_[0]->{_deleted} = 1 }

sub write {
    my ($self, $data, $headers) = @_;
    $headers //= {};
    my $h = ApmModule::_resp_headers($headers);
    $h->{_command} = 'write';
    $h->{dataType} //= 'text';
    if (ref $data && ref $data eq 'SCALAR') { $data = $$data; }
    $self->_send($h, $data);
}

# Send a binary frame (WS binary opcode; on HTTP the type marker is
# ignored). Perl scalars are byte strings, so the intent must be explicit.
sub write_binary {
    my ($self, $data, $headers) = @_;
    $headers //= {};
    my $h = ApmModule::_resp_headers($headers);
    $h->{_command} = 'write';
    $h->{dataType} = 'binary';
    if (ref $data && ref $data eq 'SCALAR') { $data = $$data; }
    $self->_send($h, $data);
}

sub write_raw {
    my ($self, $data) = @_;
    $self->_send({ _command => 'writeRaw' }, $data);
}

# Close the connection. code/reason are only sent when given, so the daemon
# applies its protocol default (HTTP 200, WS 1000) — the old unconditional
# code=200 is an invalid WebSocket close code.
sub close {
    my ($self, $code, $reason) = @_;
    my %h = ( _command => 'closeConnection' );
    $h{code}    = $code + 0   if defined $code;
    $h{_reason} = "$reason"   if defined $reason;
    $self->_send(\%h, '');
}

sub save_session_data {
    my ($self) = @_;
    $self->_send({ _command => 'saveSessionData', _sessionData => $self->{session_data} }, '');
}

# Liveness policy for this connection (WebSocket): the daemon pings after
# `interval` ms of silence, waits `timeout` ms for the pong, and after `miss`
# unanswered pings closes the client (1001) and calls on_close. A key left out
# (or 0) returns that field to the worker's ping_* configuration;
# set_ping() with no arguments resets all three.
sub set_ping {
    my ($self, %o) = @_;
    my %h = ( _command => 'setPing' );
    $h{interval} = $o{interval} + 0 if defined $o{interval};
    $h{timeout}  = $o{timeout}  + 0 if defined $o{timeout};
    $h{miss}     = $o{miss}     + 0 if defined $o{miss};
    $self->_send(\%h, '');
}

# Ping round-trip statistics the daemon keeps for this connection, from the
# pongs to its liveness pings (ms): a hash with last, lastAt, count, avg5m,
# min5m, max5m, n5m, avg60m, min60m, max60m, n60m, interval, timeout, miss.
# last is undef before the first pong. Returns undef if the daemon does not
# answer within $timeout ms.
sub ping_stats {
    my ($self, $timeout) = @_;
    return undef unless $self->{_module};
    return $self->{_module}->_session_request({ _command => 'pingStats', _session => $self->{session_id} }, $timeout);
}

sub _send {
    my ($self, $json_hdr, $binary) = @_;
    $json_hdr->{_session} = $self->{session_id};
    ApmModule::_send_frame($json_hdr, $binary);
}


# ─── IPC Stream ──────────────────────────────────────────────────────────────

package IpcStream;

sub new {
    my ($class, $id, $header, $role) = @_;
    return bless {
        id        => $id,
        header    => $header // {},
        peer_id   => undef,
        on_data   => undef,   # sub($data, $peer)
        on_attach => undef,   # sub($peer) — mediator only
        on_detach => undef,   # sub($peer) — mediator only
        on_close  => undef,   # sub()
        _closed   => 0,
        _role     => $role,
    }, $class;
}

sub active { return !$_[0]->{_closed} }

sub write {
    my ($self, $data) = @_;
    return if $self->{_closed};
    ApmModule::_send_frame({ _command => 'stream_data', _sid => $self->{id}, dataType => 'text' }, $data);
}

sub write_binary {
    my ($self, $data) = @_;
    return if $self->{_closed};
    ApmModule::_send_frame({ _command => 'stream_data', _sid => $self->{id}, dataType => 'binary' }, $data);
}

sub close {
    my ($self) = @_;
    return if $self->{_closed};
    $self->{_closed} = 1;
    ApmModule::_send_frame({ _command => 'stream_close', _sid => $self->{id} });
}

sub accept {
    my ($self, $header) = @_;
    my %hdr = ( _command => 'stream_accept', _sid => $self->{id} );
    $hdr{header} = $header if $header;
    ApmModule::_send_frame(\%hdr);
    return $self;
}

sub reject {
    my ($self) = @_;
    ApmModule::_send_frame({ _command => 'stream_reject', _sid => $self->{id} });
    $self->{_closed} = 1;
}

sub _mark_closed { $_[0]->{_closed} = 1 }


# ─── Module ───────────────────────────────────────────────────────────────────

package ApmModule;

sub new {
    my ($class, $on_connect) = @_;
    return bless {
        _on_connect => $on_connect,
        _sessions   => {},
        _streams    => {},
        _pending    => {},
        _buf        => '',
        _deferred   => [],      # frames parked while blocked inside request()
        _unknown_warned => {},  # session ids already logged as unknown
        instance_id => $_INDEX,
        on_channel  => undef,   # sub($channel, $data, $reply_fn|undef)
        on_stream   => undef,   # sub($stream)
    }, $class;
}

sub metric {
    my ($self, $name, $value, $type) = @_;
    $type //= 'counter';
    _send_frame({ _command => 'metric', name => $name, value => $value+0, type => $type });
}

sub set_dash_value {
    my ($self, $module_id, $value, $color) = @_;
    $color //= '';
    _send_frame({ _command => 'dash_value', module_id => $module_id+0, value => $value+0, color => $color });
}

# ── IPC: Channel ──────────────────────────────────────────────────────────────

sub send {
    my ($self, $channel, $data) = @_;
    _send_frame({ _command => 'ipc_send', channel => $channel, data => $data });
}

# Blocking request/reply. Returns reply data or undef on timeout.
# Safe to call from inside a callback: while waiting, the connector keeps
# reading frames but only processes the ones that resolve pending requests —
# everything else is deferred and replayed in order after the request
# completes, so no callback ever runs re-entrantly. (The old implementation
# processed every frame in the pump, re-entering on_connect/on_data/
# on_channel while user code was blocked mid-call.)
sub request {
    my ($self, $channel, $data, $timeout) = @_;
    my $rid = _next_rid();
    my %hdr = ( _command => 'ipc_request', channel => $channel, data => $data, _rid => $rid );
    $hdr{timeout} = $timeout if $timeout && $timeout > 0;
    $self->{_pending}{$rid} = { result => undef, done => 0 };
    _send_frame(\%hdr);

    $self->_pump_until($rid, Time::HiRes::time() + (($timeout || 5000) + 1000) / 1000.0);

    my $result = $self->{_pending}{$rid}{result};
    delete $self->{_pending}{$rid};
    return $result;
}

# A frame with a _rid whose ipc_reply resolves it (ping_stats); same pending
# table and nesting rules as request().
sub _session_request {
    my ($self, $hdr, $timeout) = @_;
    my $rid = _next_rid();
    $hdr->{_rid} = $rid;
    $self->{_pending}{$rid} = { result => undef, done => 0 };
    _send_frame($hdr);
    $self->_pump_until($rid, Time::HiRes::time() + (($timeout && $timeout > 0) ? $timeout : 2000) / 1000.0);
    my $result = $self->{_pending}{$rid}{result};
    delete $self->{_pending}{$rid};
    return $result;
}

# ── IPC: Stream ───────────────────────────────────────────────────────────────

sub request_stream {
    my ($self, $channel, $header, $timeout) = @_;
    my $sid = _next_rid();
    my %hdr = ( _command => 'stream_open', channel => $channel, _sid => $sid );
    $hdr{header}  = $header  if $header;
    $hdr{timeout} = $timeout if $timeout && $timeout > 0;

    my $stream = IpcStream->new($sid, $header, 'mediator');
    $self->{_streams}{$sid} = $stream;
    $self->{_pending}{$sid} = { result => undef, done => 0 };
    _send_frame(\%hdr);

    $self->_pump_until($sid, Time::HiRes::time() + (($timeout || 10000) + 1000) / 1000.0);

    my $ok = $self->{_pending}{$sid}{result};
    delete $self->{_pending}{$sid};
    if ($ok) { return $stream; }
    $stream->_mark_closed();
    delete $self->{_streams}{$sid};
    return undef;
}

# ── Event loop ────────────────────────────────────────────────────────────────

sub run {
    my ($self) = @_;
    binmode STDIN;
    binmode STDOUT;
    while (1) {
        $self->_drain_deferred();
        # sysread returns as soon as at least one byte is available —
        # read() had fread semantics and sat blocked until 64 KB of frames
        # accumulated, so a normal small request was never processed
        # before the daemon's 504.
        my $chunk = '';
        my $n = sysread(STDIN, $chunk, 65536);
        last unless defined $n && $n > 0;
        $self->{_buf} .= $chunk;
        foreach my $frame (@{ $self->_extract_frames() }) {
            $self->_process_frame($frame);
        }
    }
    $self->_drain_deferred();
}

sub _extract_frames {
    my ($self) = @_;
    my @frames;
    while (length $self->{_buf}) {
        if (ord(substr($self->{_buf}, 0, 1)) != 0x05) {
            my $pos = index($self->{_buf}, "\x05");
            if ($pos == -1) { $self->{_buf} = ''; last; }
            $self->{_buf} = substr($self->{_buf}, $pos);
        }
        last if length($self->{_buf}) < 5;

        my ($payload_len) = unpack('N', substr($self->{_buf}, 1, 4));
        my $frame_len = $payload_len + 6;

        last if length($self->{_buf}) < $frame_len;

        push @frames, substr($self->{_buf}, 0, $frame_len);
        $self->{_buf} = substr($self->{_buf}, $frame_len);
    }
    return \@frames;
}

# Frames parked while user code was blocked inside request()/request_stream()
# are replayed in arrival order before the next read, so nothing is lost and
# no callback ever runs re-entrantly inside a blocking call.
sub _drain_deferred {
    my ($self) = @_;
    while (@{ $self->{_deferred} }) {
        my $frame = shift @{ $self->{_deferred} };
        $self->_process_frame($frame);
    }
}

# Does this frame resolve a pending request/stream-open? Those frames only
# flip completion flags — no user callbacks — so they are safe to process
# while blocked; everything else is deferred.
sub _frame_resolves_pending {
    my ($self, $frame) = @_;
    my $sep = index($frame, "\x03", 5);
    return 0 if $sep == -1;
    my $header;
    eval { $header = decode_json(substr($frame, 5, $sep - 5)); };
    return 0 if $@ || !$header;
    my $t = $header->{_type} // '';
    if ($t eq 'ipc_reply') {
        return defined $header->{_rid} && exists $self->{_pending}{$header->{_rid}};
    }
    if ($t eq 'stream_opened' || $t eq 'stream_rejected') {
        return defined $header->{_sid} && exists $self->{_pending}{$header->{_sid}};
    }
    return 0;
}

sub _pump_until {
    my ($self, $key, $deadline) = @_;
    my $rin = '';
    vec($rin, fileno(STDIN), 1) = 1;
    while ($self->{_pending}{$key} && !$self->{_pending}{$key}{done}) {
        my $remaining = $deadline - Time::HiRes::time();
        return if $remaining <= 0;
        my $rout = $rin;
        my $n = select($rout, undef, undef, $remaining);
        next unless $n && $n > 0;
        my $chunk = '';
        my $got = sysread(STDIN, $chunk, 65536);
        return unless defined $got && $got > 0;
        $self->{_buf} .= $chunk;
        foreach my $frame (@{ $self->_extract_frames() }) {
            if ($self->_frame_resolves_pending($frame)) {
                $self->_process_frame($frame);
            } else {
                push @{ $self->{_deferred} }, $frame;
            }
        }
    }
}

sub _process_frame {
    my ($self, $frame) = @_;
    my $sep = index($frame, "\x03", 5);
    return if $sep == -1;

    my $header;
    eval { $header = decode_json(substr($frame, 5, $sep - 5)); };
    return if $@ || !$header;

    my $binary_data = substr($frame, $sep + 1);
    my $type = $header->{_type} // '';

    # ── IPC frames ────────────────────────────────────────────────────────────

    if ($type eq 'ipc_message') {
        if ($self->{on_channel}) {
            my $channel = $header->{channel} // '';
            my $data    = $header->{data};
            my $reply   = undef;
            if ($header->{_needReply} && $header->{_rid}) {
                my $rid = $header->{_rid};
                $reply = sub { _send_frame({ _command => 'ipc_reply', _rid => $rid, data => $_[0] }); };
            }
            eval { $self->{on_channel}->($channel, $data, $reply); };
            if ($@) { print STDERR "[APM] on_channel error: $@\n"; }
        }
        return;
    }

    if ($type eq 'ipc_reply') {
        my $rid = $header->{_rid} // '';
        if ($rid && $self->{_pending}{$rid}) {
            $self->{_pending}{$rid}{result} = $header->{data};
            $self->{_pending}{$rid}{done}   = 1;
        }
        return;
    }

    if ($type eq 'stream_request') {
        if ($self->{on_stream}) {
            my $sid    = $header->{_sid} // '';
            my $stream = IpcStream->new($sid, $header->{header}, 'peer');
            $self->{_streams}{$sid} = $stream;
            eval { $self->{on_stream}->($stream); };
            if ($@) { print STDERR "[APM] on_stream error: $@\n"; }
        }
        return;
    }

    if ($type eq 'stream_opened') {
        my $sid = $header->{_sid} // '';
        if ($self->{_pending}{$sid}) {
            $self->{_pending}{$sid}{result} = 1;
            $self->{_pending}{$sid}{done}   = 1;
        }
        return;
    }

    if ($type eq 'stream_rejected') {
        my $sid = $header->{_sid} // '';
        if ($self->{_pending}{$sid}) {
            $self->{_pending}{$sid}{result} = 0;
            $self->{_pending}{$sid}{done}   = 1;
        }
        if ($self->{_streams}{$sid}) {
            $self->{_streams}{$sid}->_mark_closed();
            delete $self->{_streams}{$sid};
        }
        return;
    }

    if ($type eq 'stream_attached') {
        my $sid = $header->{_sid} // '';
        if ($self->{_streams}{$sid}) {
            $self->{_streams}{$sid}{peer_id} = $header->{_peer};
        }
        return;
    }

    if ($type eq 'stream_attach') {
        my $sid = $header->{_sid} // '';
        my $stream = $self->{_streams}{$sid};
        if ($stream && $stream->{on_attach}) {
            eval { $stream->{on_attach}->({ id => ($header->{_peer} // ''), header => ($header->{header} // {}) }); };
            if ($@) { print STDERR "[APM] stream.on_attach error: $@\n"; }
        }
        return;
    }

    if ($type eq 'stream_detach') {
        my $sid = $header->{_sid} // '';
        my $stream = $self->{_streams}{$sid};
        if ($stream && $stream->{on_detach}) {
            eval { $stream->{on_detach}->({ id => ($header->{_peer} // '') }); };
            if ($@) { print STDERR "[APM] stream.on_detach error: $@\n"; }
        }
        return;
    }

    if ($type eq 'stream_data') {
        my $sid = $header->{_sid} // '';
        my $stream = $self->{_streams}{$sid};
        if ($stream && $stream->{on_data}) {
            my $peer = $header->{_peer} ? { id => $header->{_peer} } : undef;
            eval { $stream->{on_data}->($binary_data, $peer); };
            if ($@) { print STDERR "[APM] stream.on_data error: $@\n"; }
        }
        return;
    }

    if ($type eq 'stream_close') {
        my $sid = $header->{_sid} // '';
        my $stream = $self->{_streams}{$sid};
        if ($stream) {
            $stream->_mark_closed();
            delete $self->{_streams}{$sid};
            if ($stream->{on_close}) {
                eval { $stream->{on_close}->(); };
            }
        }
        return;
    }

    # ── Session frames ────────────────────────────────────────────────────────

    return unless $header->{_sessionId};

    my $sid  = $header->{_sessionId};
    my $sess = $self->{_sessions}{$sid};

    if (!defined $sess) {
        # Only a connect frame creates a session. A close event for a session
        # already gone is the normal end of a race; any other stray frame is
        # dropped and logged once per id.
        if ($type ne 'connect') {
            return if $type eq 'event' && ($header->{_event} // '') eq 'connectionClosed';
            $self->_warn_unknown($sid);
            return;
        }

        my $sd = $header->{_sessionData};
        $sess = ApmSession->new();
        $sess->{_module} = $self;
        $sess->{session_id}   = $sid;
        $sess->{session_type} = $header->{_sessionType}  // 'new';
        $sess->{session_data} = (ref $sd eq 'HASH') ? $sd : {};
        $sess->{protocol}     = $header->{protocol}      // '';
        $sess->{method}       = $header->{method}        // undef;
        $sess->{path}         = $header->{path}          // '/';
        $sess->{path_array}   = $header->{path_array}    // [];
        $sess->{query}        = $header->{query}         // {};
        $sess->{query_object} = $header->{query_object}  // {};
        $sess->{cookies}      = $header->{cookies}       // {};
        $sess->{headers}      = $header->{headers}       // {};
        my @parts             = split(/,/, $header->{remoteAddress} // '');
        ($sess->{remote_ip}   = $parts[0] // '') =~ s/^\s+|\s+$//g;

        $self->{_sessions}{$sid} = $sess;

        eval { $self->{_on_connect}->($sess); };
        if ($@) { print STDERR "[APM] on_connect error: $@\n"; }

        foreach my $item (@{ $sess->{_backlog} }) {
            if ($sess->active() && $sess->{on_data}) {
                eval { $sess->{on_data}->($item->{data}, $item->{binary}); };
                if ($@) { print STDERR "[APM] on_data error: $@\n"; }
            }
        }
        $sess->{_backlog} = [];
        return;
    }

    if ($header->{headers})      { %{$sess->{headers}}      = (%{$sess->{headers}},      %{$header->{headers}});  }
    if ($header->{path})         { $sess->{path}         = $header->{path};         }
    if ($header->{path_array})   { $sess->{path_array}   = $header->{path_array};   }
    if ($header->{query})        { $sess->{query}        = $header->{query};        }
    if ($header->{query_object}) { $sess->{query_object} = $header->{query_object}; }
    if ($header->{cookies})      { $sess->{cookies}      = $header->{cookies};      }
    if ($header->{method})       { $sess->{method}       = $header->{method};       }

    return unless $sess->active();

    my $ftype = $header->{_type}  // '';
    my $event = $header->{_event} // '';

    if ($ftype eq 'data' || $ftype eq 'chunk') {
        my $is_binary = (($header->{dataType} // '') eq 'binary') ? 1 : 0;
        my $data      = substr($frame, $sep + 1);
        if ($sess->{on_data}) {
            eval { $sess->{on_data}->($data, $is_binary); };
            if ($@) { print STDERR "[APM] on_data error: $@\n"; }
        } else {
            push @{ $sess->{_backlog} }, { data => $data, binary => $is_binary };
        }
        return;
    }

    if ($ftype eq 'event' && $event eq 'connectionClosed') {
        if ($sess->{on_close}) {
            eval { $sess->{on_close}->(); };
        }
        $sess->_mark_deleted();
        delete $self->{_sessions}{$sid};
    }
}

# One stderr line per unknown session id; the hash is capped so a flood of
# stray ids cannot grow it without bound.
sub _warn_unknown {
    my ($self, $sid) = @_;
    return if $self->{_unknown_warned}{$sid};
    $self->{_unknown_warned} = {} if keys %{ $self->{_unknown_warned} } >= 1000;
    $self->{_unknown_warned}{$sid} = 1;
    print STDERR "[APM] frame for unknown session $sid dropped\n";
}

1;