File Coverage

blib/lib/PAGI/Request.pm

Criterion	Covered	Total	%
statement	297	349	85.1
branch	89	128	69.5
condition	57	91	62.6
subroutine	73	88	82.9
pod	64	69	92.7
total	580	725	80.0

package PAGI::Request;
$PAGI::Request::VERSION = '0.002000';
use strict;
use warnings;
use Hash::MultiValue;
use PAGI::Headers ();
use Encode qw(decode FB_CROAK FB_DEFAULT LEAVE_SRC);
use Cookie::Baker qw(crush_cookie);
use MIME::Base64 qw(decode_base64);
use Future::AsyncAwait;
use JSON::MaybeXS qw(decode_json);
use Carp qw(croak carp);
use PAGI::Request::MultiPartHandler;
use PAGI::Request::Upload;
use PAGI::Request::Negotiate;
use PAGI::Request::BodyStream;

sub new {
    my ($class, $scope, $receive) = @_;
    return bless {
        scope   => $scope,
        receive => $receive,
    }, $class;
}

# Basic properties from scope
sub method       { shift->{scope}{method} }
sub path         { shift->{scope}{path} }
sub raw_path     { my $s = shift; $s->{scope}{raw_path} // $s->{scope}{path} }
sub query_string { shift->{scope}{query_string} // '' }
sub scheme       { shift->{scope}{scheme} // 'http' }
sub http_version { shift->{scope}{http_version} // '1.1' }
sub client       { shift->{scope}{client} }
sub raw          { shift->{scope} }

# Internal: URL decode a string (handles + as space)
sub _url_decode {
    my ($str) = @_;
    return '' unless defined $str;
    $str =~ s/\+/ /g;
    $str =~ s/%([0-9A-Fa-f]{2})/chr(hex($1))/ge;
    return $str;
}

# Internal: Decode UTF-8 with replacement or croak in strict mode
sub _decode_utf8 {
    my ($str, $strict) = @_;
    return '' unless defined $str;
    my $flag = $strict ? FB_CROAK : FB_DEFAULT;
    $flag |= LEAVE_SRC;
    return decode('UTF-8', $str, $flag);
}

# Host from headers
sub host {
    my $self = shift;
    return $self->header('host');
}

# Content-Type shortcut
sub content_type {
    my $self = shift;
    my $ct = $self->header('content-type') // '';
    # Strip parameters like charset
    $ct =~ s/;.*//;
    return $ct;
}

# Content-Length shortcut
sub content_length {
    my $self = shift;
    return $self->header('content-length');
}

# Private cache: the source of truth for header lookups (built once per request).
sub _header_snapshot {
    my $self = shift;
    return $self->{scope}{'pagi.request.headers'}
        //= PAGI::Headers->new($self->{scope}{headers} // []);
}

# Public: a PAGI::Headers snapshot of the inbound headers. Returns an independent
# CLONE so mutating it cannot poison later header()/content_type()/cookie lookups.
sub headers { return $_[0]->_header_snapshot->clone }

# Single header lookup (case-insensitive, last value)
sub header     { return $_[0]->_header_snapshot->get($_[1]) }

# All values for a header
sub header_all { return $_[0]->_header_snapshot->get_all($_[1]) }

# Query params as Hash::MultiValue (cached in scope)
# Options: strict => 1 (croak on invalid UTF-8), raw => 1 (skip UTF-8 decoding)
sub query_params {
    my ($self, %opts) = @_;
    my $strict = delete $opts{strict} // 0;
    my $raw    = delete $opts{raw}    // 0;
    croak("Unknown options to query_params: " . join(', ', keys %opts)) if %opts;

    my $cache_key = $raw ? 'pagi.request.query.raw' : ($strict ? 'pagi.request.query.strict' : 'pagi.request.query');
    return $self->{scope}{$cache_key} if $self->{scope}{$cache_key};

    my $qs = $self->query_string;
    my @pairs;

    for my $part (split /[&;]/, $qs) {
        next unless length $part;
        my ($key, $val) = split /=/, $part, 2;
        $key //= '';
        $val //= '';

        # URL decode (handles + as space)
        my $key_decoded = _url_decode($key);
        my $val_decoded = _url_decode($val);

        # UTF-8 decode unless raw mode
        my $key_final = $raw ? $key_decoded : _decode_utf8($key_decoded, $strict);
        my $val_final = $raw ? $val_decoded : _decode_utf8($val_decoded, $strict);

        push @pairs, $key_final, $val_final;
    }

    $self->{scope}{$cache_key} = Hash::MultiValue->new(@pairs);
    return $self->{scope}{$cache_key};
}

# Raw query params (no UTF-8 decoding)
sub raw_query_params {
    my $self = shift;
    return $self->query_params(raw => 1);
}

# Shortcut for single query param
sub query_param {
    my ($self, $name, %opts) = @_;
    return $self->query_params(%opts)->get($name);
}

# DEPRECATED: Alias with warning
sub query {
    my $self = shift;
    carp "query() is deprecated; use query_param() instead";
    return $self->query_param(@_);
}

# Raw single query param
sub raw_query_param {
    my ($self, $name) = @_;
    return $self->query_param($name, raw => 1);
}

# DEPRECATED: Alias with warning
sub raw_query {
    my $self = shift;
    carp "raw_query() is deprecated; use raw_query_param() instead";
    return $self->raw_query_param(@_);
}

# All cookies as hashref (cached in scope)
sub cookies {
    my $self = shift;
    return $self->{scope}{'pagi.request.cookies'} if exists $self->{scope}{'pagi.request.cookies'};

    my $cookie_header = $self->header('cookie') // '';
    $self->{scope}{'pagi.request.cookies'} = crush_cookie($cookie_header);
    return $self->{scope}{'pagi.request.cookies'};
}

# Single cookie value
sub cookie {
    my ($self, $name) = @_;
    return $self->cookies->{$name};
}

# Method predicates
sub is_get     { uc(shift->method // '') eq 'GET' }
sub is_post    { uc(shift->method // '') eq 'POST' }
sub is_put     { uc(shift->method // '') eq 'PUT' }
sub is_patch   { uc(shift->method // '') eq 'PATCH' }
sub is_delete  { uc(shift->method // '') eq 'DELETE' }
sub is_head    { uc(shift->method // '') eq 'HEAD' }
sub is_options { uc(shift->method // '') eq 'OPTIONS' }

# =============================================================================
# Connection State Methods (PAGI spec 0.3)
#
# These methods provide non-destructive disconnect detection via the
# pagi.connection scope key, which is a PAGI::Server::ConnectionState object.
# =============================================================================

# Get the connection state object
sub connection {
    my $self = shift;
    return $self->{scope}{'pagi.connection'};
}

# Check if client is still connected (synchronous, non-destructive)
sub is_connected {
    my $self = shift;
    my $conn = $self->connection;
    return 0 unless $conn;
    return $conn->is_connected;
}

# Check if client has disconnected (synchronous, non-destructive)
# This is the inverse of is_connected - preferred for new code
sub is_disconnected {
    my $self = shift;
    return !$self->is_connected;
}

# Get the disconnect reason string, or undef if still connected
sub disconnect_reason {
    my $self = shift;
    my $conn = $self->connection;
    return undef unless $conn;
    return $conn->disconnect_reason;
}

# Register a callback to be invoked on an abnormal disconnect (not on a clean
# finish). The counterpart to on_complete; exactly one of the two fires.
sub on_disconnect {
    my ($self, $cb) = @_;
    my $conn = $self->connection;
    $conn->on_disconnect($cb) if $conn;
    return $self;
}

# Register a callback to be invoked only when the request completes successfully.
# The counterpart to on_disconnect; exactly one of the two fires.
sub on_complete {
    my ($self, $cb) = @_;
    my $conn = $self->connection;
    $conn->on_complete($cb) if $conn;
    return $self;
}

# Get a Future that resolves when the client disconnects
sub disconnect_future {
    my $self = shift;
    my $conn = $self->connection;
    return undef unless $conn;
    return $conn->disconnect_future;
}

# Outbound flow-control introspection (delegates to the pagi.transport handle)
sub buffered_amount {
    my $self = shift;
    my $t = $self->{scope}{'pagi.transport'};
    return 0 unless $t;
    return $t->buffered_amount;
}

sub high_water_mark {
    my $self = shift;
    my $t = $self->{scope}{'pagi.transport'};
    return undef unless $t;
    return $t->high_water_mark;
}

sub low_water_mark {
    my $self = shift;
    my $t = $self->{scope}{'pagi.transport'};
    return undef unless $t;
    return $t->low_water_mark;
}

sub on_high_water {
    my ($self, $cb) = @_;
    my $t = $self->{scope}{'pagi.transport'};
    $t->on_high_water($cb) if $t && $t->can('on_high_water');
    return $self;
}

sub on_drain {
    my ($self, $cb) = @_;
    my $t = $self->{scope}{'pagi.transport'};
    $t->on_drain($cb) if $t && $t->can('on_drain');
    return $self;
}

sub is_writable {
    my $self = shift;
    my $t = $self->{scope}{'pagi.transport'};
    return 1 unless $t;
    my $high = $t->high_water_mark;
    return 1 unless defined $high;
    return $t->buffered_amount < $high ? 1 : 0;
}

# Content-type predicates
sub is_json {
    my $self = shift;
    my $ct = $self->content_type;
    return $ct eq 'application/json';
}

sub is_form {
    my $self = shift;
    my $ct = $self->content_type;
    return $ct eq 'application/x-www-form-urlencoded'
        || $ct =~ m{^multipart/form-data};
}

sub is_multipart {
    my $self = shift;
    my $ct = $self->content_type;
    return $ct =~ m{^multipart/form-data};
}

# Accept header check using Negotiate module
# Combines multiple Accept headers per RFC 7230 Section 3.2.2
sub accepts {
    my ($self, $mime_type) = @_;
    my @accept_values = $self->header_all('accept');
    my $accept = join(', ', @accept_values);
    return PAGI::Request::Negotiate->accepts_type($accept, $mime_type);
}

# Find best matching content type from supported list
# Combines multiple Accept headers per RFC 7230 Section 3.2.2
sub preferred_type {
    my ($self, @types) = @_;
    my @accept_values = $self->header_all('accept');
    my $accept = join(', ', @accept_values);
    return PAGI::Request::Negotiate->best_match(\@types, $accept);
}

# Extract Bearer token from Authorization header
sub bearer_token {
    my $self = shift;
    my $auth = $self->header('authorization') // '';
    if ($auth =~ /^Bearer\s+(.+)$/i) {
        return $1;
    }
    return undef;
}

# Extract Basic auth credentials
sub basic_auth {
    my $self = shift;
    my $auth = $self->header('authorization') // '';
    if ($auth =~ /^Basic\s+(.+)$/i) {
        my $decoded = decode_base64($1);
        my ($user, $pass) = split /:/, $decoded, 2;
        return ($user, $pass);
    }
    return (undef, undef);
}

# Path parameters - captured from URL path by router
# Stored in scope->{path_params} for router-agnostic access
sub path_params {
    my ($self, %opts) = @_;
    my $strict = delete $opts{strict};
    croak("Unknown options to path_params: " . join(', ', keys %opts)) if %opts;

    my $params = $self->{scope}{path_params};
    if (!defined $params && $strict) {
        croak "path_params not set in scope (no router configured?). "
            . "Pass strict => 0 to allow this.";
    }
    return $params // {};
}

sub _default_path_param_strict_opt { return 1 }

sub path_param {
    my ($self, $name, %opts) = @_;
    my $strict = exists $opts{strict} ? delete $opts{strict} : $self->_default_path_param_strict_opt;
    croak("Unknown options to path_param: " . join(', ', keys %opts)) if %opts;

    my $params = $self->path_params;

    if ($strict && !exists $params->{$name}) {
        my @available = keys %$params;
        croak "path_param '$name' not found. "
            . (@available ? "Available: " . join(', ', sort @available) : "No path params set (no router?)");
    }

    return $params->{$name};
}

sub scope { shift->{scope} }

# Vend a detached response bound to this request's scope (the raw-app analog
# of $ctx->response). It is a value, not a connection; call ->respond($send)
# to send it.
sub response {
    my $self = shift;
    require PAGI::Response;
    return PAGI::Response->new($self->{scope});
}


# Application state (injected by PAGI::Lifespan, read-only)
sub state {
    my $self = shift;
    return $self->{scope}{state} // {};
}

# Body streaming - mutually exclusive with buffered body methods
sub body_stream {
    my ($self, %opts) = @_;

    croak "Body already consumed; streaming not available" if $self->{scope}{'pagi.request.body.read'};
    croak "Body streaming already started" if $self->{scope}{'pagi.request.body.stream.created'};

    $self->{scope}{'pagi.request.body.stream.created'} = 1;

    my $max_bytes = $opts{max_bytes};
    my $limit_name = defined $max_bytes ? 'max_bytes' : undef;
    if (!defined $max_bytes) {
        my $cl = $self->content_length;
        if (defined $cl) {
            $max_bytes = $cl;
            $limit_name = 'content-length';
        }
    }

    return PAGI::Request::BodyStream->new(
        receive    => $self->{receive},
        max_bytes  => $max_bytes,
        limit_name => $limit_name,
        decode     => $opts{decode},
        strict     => $opts{strict},
    );
}

# Streaming multipart - mutually exclusive with buffered body methods
sub multipart_stream {
    my ($self, %opts) = @_;
    croak "Body already consumed; multipart_stream() not available"
        if $self->{scope}{'pagi.request.body.read'}
        || $self->{scope}{'pagi.request.body.stream.created'};
    croak "multipart_stream() requires a multipart/form-data request" unless $self->is_multipart;

    my $ct = $self->header('content-type') // '';
    my ($boundary) = $ct =~ /boundary=([^;\s]+)/;
    $boundary =~ s/^["']|["']$//g if defined $boundary;  # Strip quotes
    croak "No boundary found in Content-Type" unless defined $boundary && length $boundary;

    $self->{scope}{'pagi.request.body.stream.created'} = 1;  # latch: lock out buffered readers

    require PAGI::Request::MultipartStream;
    return PAGI::Request::MultipartStream->new(
        receive  => $self->{receive},
        boundary => $boundary,
        map { defined $opts{$_} ? ($_ => $opts{$_}) : () }
            qw(max_files max_fields max_field_size max_file_size max_request_body),
    );
}

# Read raw body bytes (async, cached in scope)
async sub body {
    my $self = shift;

    croak "Body streaming already started; buffered helpers unavailable"
        if $self->{scope}{'pagi.request.body.stream.created'};

    # Return cached body if already read
    return $self->{scope}{'pagi.request.body'} if $self->{scope}{'pagi.request.body.read'};

    my $receive = $self->{receive};
    die "No receive callback provided" unless $receive;

    my $body = '';
    while (1) {
        my $message = await $receive->();
        last unless $message && $message->{type};
        last if $message->{type} eq 'http.disconnect';

        $body .= $message->{body} // '';
        last unless $message->{more};
    }

    $self->{scope}{'pagi.request.body'} = $body;
    $self->{scope}{'pagi.request.body.read'} = 1;
    return $body;
}

# Read body as decoded UTF-8 text (async)
# Options: strict => 1 (croak on invalid UTF-8)
async sub text {
    my ($self, %opts) = @_;
    my $strict = delete $opts{strict} // 0;
    croak("Unknown options to text: " . join(', ', keys %opts)) if %opts;

    my $body = await $self->body;
    return _decode_utf8($body, $strict);
}

# Parse body as JSON (async, dies on error)
async sub json {
    my $self = shift;
    my $body = await $self->body;
    return decode_json($body);
}

# Parse URL-encoded form body (async, returns Hash::MultiValue, cached in scope)
# Options: strict => 1 (croak on invalid UTF-8), raw => 1 (skip UTF-8 decoding)
async sub form_params {
    my ($self, %opts) = @_;
    my $strict = delete $opts{strict} // 0;
    my $raw    = delete $opts{raw}    // 0;

    # Extract multipart options before checking for unknown opts
    my %multipart_opts;
    for my $key (qw(max_field_size max_file_size spool_threshold max_files max_fields temp_dir)) {
        $multipart_opts{$key} = delete $opts{$key} if exists $opts{$key};
    }
    croak("Unknown options to form_params: " . join(', ', keys %opts)) if %opts;

    my $cache_key = $raw ? 'pagi.request.form.raw' : ($strict ? 'pagi.request.form.strict' : 'pagi.request.form');

    # Return cached if available
    return $self->{scope}{$cache_key} if $self->{scope}{$cache_key};

    # For multipart, delegate to uploads handling
    if ($self->is_multipart) {
        # Multipart always parses to default cache, then copy
        my $form = await $self->_parse_multipart_form(%multipart_opts);
        $self->{scope}{$cache_key} = $form;
        return $form;
    }

    # URL-encoded form
    my $body = await $self->body;
    my @pairs;

    for my $part (split /[&;]/, $body) {
        next unless length $part;
        my ($key, $val) = split /=/, $part, 2;
        $key //= '';
        $val //= '';

        # URL decode (handles + as space)
        my $key_decoded = _url_decode($key);
        my $val_decoded = _url_decode($val);

        # UTF-8 decode unless raw mode
        my $key_final = $raw ? $key_decoded : _decode_utf8($key_decoded, $strict);
        my $val_final = $raw ? $val_decoded : _decode_utf8($val_decoded, $strict);

        push @pairs, $key_final, $val_final;
    }

    $self->{scope}{$cache_key} = Hash::MultiValue->new(@pairs);
    return $self->{scope}{$cache_key};
}

# DEPRECATED: Alias with warning
async sub form {
    my $self = shift;
    carp "form() is deprecated; use form_params() instead";
    return await $self->form_params(@_);
}

# Singular accessor for form params
async sub form_param {
    my ($self, $name, %opts) = @_;
    my $form = await $self->form_params(%opts);
    return $form->get($name);
}

# Raw form params (no UTF-8 decoding)
async sub raw_form_params {
    my ($self, %opts) = @_;
    return await $self->form_params(%opts, raw => 1);
}

# DEPRECATED: Alias with warning
async sub raw_form {
    my $self = shift;
    carp "raw_form() is deprecated; use raw_form_params() instead";
    return await $self->raw_form_params(@_);
}

# Raw singular accessor
async sub raw_form_param {
    my ($self, $name) = @_;
    return await $self->form_param($name, raw => 1);
}

# Parse multipart form (internal, cached in scope)
async sub _parse_multipart_form {
    my ($self, %opts) = @_;

    croak "Body streaming already started; buffered helpers unavailable"
        if $self->{scope}{'pagi.request.body.stream.created'};

    # Already parsed?
    return $self->{scope}{'pagi.request.form'}
        if $self->{scope}{'pagi.request.form'} && $self->{scope}{'pagi.request.uploads'};

    # Extract boundary from content-type
    my $ct = $self->header('content-type') // '';
    my ($boundary) = $ct =~ /boundary=([^;\s]+)/;
    $boundary =~ s/^["']|["']$//g if $boundary;  # Strip quotes

    die "No boundary found in Content-Type" unless $boundary;

    my $handler = PAGI::Request::MultiPartHandler->new(
        boundary        => $boundary,
        receive         => $self->{receive},
        max_field_size  => $opts{max_field_size},
        max_file_size   => $opts{max_file_size},
        spool_threshold => $opts{spool_threshold},
        max_files       => $opts{max_files},
        max_fields      => $opts{max_fields},
        temp_dir        => $opts{temp_dir},
    );

    my ($form, $uploads) = await $handler->parse;

    $self->{scope}{'pagi.request.form'} = $form;
    $self->{scope}{'pagi.request.uploads'} = $uploads;
    $self->{scope}{'pagi.request.body.read'} = 1;  # Body has been consumed

    return $form;
}

# Get all uploads as Hash::MultiValue (cached in scope)
async sub uploads {
    my ($self, %opts) = @_;

    return $self->{scope}{'pagi.request.uploads'} if $self->{scope}{'pagi.request.uploads'};

    if ($self->is_multipart) {
        await $self->_parse_multipart_form(%opts);
        return $self->{scope}{'pagi.request.uploads'};
    }

    # Not multipart - return empty
    $self->{scope}{'pagi.request.uploads'} = Hash::MultiValue->new();
    return $self->{scope}{'pagi.request.uploads'};
}

# Get single upload by field name
async sub upload {
    my ($self, $name, %opts) = @_;
    my $uploads = await $self->uploads(%opts);
    return $uploads->get($name);
}

# Get all uploads for a field name
async sub upload_all {
    my ($self, $name, %opts) = @_;
    my $uploads = await $self->uploads(%opts);
    return $uploads->get_all($name);
}

1;

__END__