File Coverage

blib/lib/PAGI/SSE.pm

Criterion	Covered	Total	%
statement	324	379	85.4
branch	124	178	69.6
condition	49	94	52.1
subroutine	60	66	90.9
pod	44	48	91.6
total	601	765	78.5

package PAGI::SSE;
$PAGI::SSE::VERSION = '0.002001';
use strict;
use warnings;
use Carp qw(croak);
use Hash::MultiValue;
use Future::AsyncAwait;
use Future;
use JSON::MaybeXS ();
use Scalar::Util qw(blessed);
use Encode qw(decode FB_CROAK FB_DEFAULT LEAVE_SRC);


sub new {
    my ($class, $scope, $receive, $send) = @_;

    croak "PAGI::SSE requires scope hashref"
        unless $scope && ref($scope) eq 'HASH';
    croak "PAGI::SSE requires receive coderef"
        unless $receive && ref($receive) eq 'CODE';
    croak "PAGI::SSE requires send coderef"
        unless $send && ref($send) eq 'CODE';
    croak "PAGI::SSE requires scope type 'sse', got '$scope->{type}'"
        unless ($scope->{type} // '') eq 'sse';

    # Return existing SSE object if one was already created for this scope
    # This ensures consistent state (is_started, is_closed, callbacks) if
    # multiple code paths create SSE objects from the same scope.
    return $scope->{'pagi.sse'} if $scope->{'pagi.sse'};

    my $self = bless {
        scope             => $scope,
        receive           => $receive,
        send              => $send,
        _state            => 'pending',  # pending -> started -> closed
        _on_close         => [],
        _on_error         => [],
        _disconnect_reason => undef,     # Set when disconnect received
    }, $class;

    # Cache in scope for reuse (weakened to avoid circular reference leak)
    $scope->{'pagi.sse'} = $self;
    Scalar::Util::weaken($scope->{'pagi.sse'});

    return $self;
}

# Scope property accessors
sub scope        { shift->{scope} }
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} // '' }

# URL decode helper (handles + as space per application/x-www-form-urlencoded)
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;
}

# UTF-8 decode helper with strict/lenient 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);
}

# 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.sse.query.raw' : ($strict ? 'pagi.sse.query.strict' : 'pagi.sse.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);
}

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

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

sub scheme       { shift->{scope}{scheme} // 'http' }
sub http_version { shift->{scope}{http_version} // '1.1' }
sub client       { shift->{scope}{client} }
sub server       { shift->{scope}{server} }


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

# Path parameter accessors - captured from URL path by router
# Stored in scope->{path_params} for router-agnostic access
sub path_params {
    my ($self) = @_;
    return $self->{scope}{path_params} // {};
}

sub path_param {
    my ($self, $name) = @_;
    my $params = $self->{scope}{path_params} // {};
    return $params->{$name};
}

# Connection state accessors
sub connection_state { shift->{_state} }

sub is_started {
    my $self = shift;
    return $self->{_state} eq 'started';
}

sub is_closed {
    my $self = shift;
    return $self->{_state} eq 'closed';
}

# True while the stream is live (started and not yet closed/disconnected).
# Mirrors PAGI::WebSocket->is_connected (state eq 'connected'): an SSE scope
# carries no pagi.connection (spec: N/A for sse), so connection liveness is
# derived from this object's own state machine, not the base Context method.
sub is_connected {
    my $self = shift;
    return $self->{_state} eq 'started';
}

# Disconnect reason - why the connection closed
# Common values: 'client_closed', 'write_error', 'send_timeout', 'idle_timeout'
sub disconnect_reason {
    my $self = shift;
    return $self->{_disconnect_reason};
}

# 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;
}

# Internal state setters
sub _set_state {
    my ($self, $state) = @_;
    $self->{_state} = $state;
}

sub _set_closed {
    my ($self) = @_;
    $self->{_state} = 'closed';

    # Wake a parked run() (which races this future against $receive) so a close
    # from deep in a helper actually ends the stream.
    $self->{_closed_future}->done
        if $self->{_closed_future} && !$self->{_closed_future}->is_ready;
}

# Start the SSE stream
async sub start {
    my ($self, %opts) = @_;

    # Idempotent - don't start twice
    return $self if $self->is_started || $self->is_closed;

    my $event = {
        type   => 'sse.start',
        status => $opts{status} // 200,
    };
    $event->{headers} = $opts{headers} if exists $opts{headers};

    await $self->{send}->($event);
    $self->_set_state('started');

    return $self;
}

# Enable keepalive - sends sse.keepalive event to server
# Server handles the timer; this is loop-agnostic
async sub keepalive {
    my ($self, $interval, $comment) = @_;

    await $self->{send}->({
        type     => 'sse.keepalive',
        interval => $interval // 0,
        comment  => $comment // '',
    });

    return $self;
}

# Single header lookup (case-insensitive, returns last value)
sub header {
    my ($self, $name) = @_;
    $name = lc($name);
    my $value;
    for my $pair (@{$self->{scope}{headers} // []}) {
        if (lc($pair->[0]) eq $name) {
            $value = $pair->[1];
        }
    }
    return $value;
}

# All headers as Hash::MultiValue (cached in scope)
sub headers {
    my $self = shift;
    return $self->{scope}{'pagi.request.headers'} if $self->{scope}{'pagi.request.headers'};

    my @pairs;
    for my $pair (@{$self->{scope}{headers} // []}) {
        push @pairs, lc($pair->[0]), $pair->[1];
    }

    $self->{scope}{'pagi.request.headers'} = Hash::MultiValue->new(@pairs);
    return $self->{scope}{'pagi.request.headers'};
}

# All values for a header
sub header_all {
    my ($self, $name) = @_;
    return $self->headers->get_all(lc($name));
}

# Get Last-Event-ID header from client (for reconnection)
sub last_event_id {
    my ($self) = @_;
    return $self->header('last-event-id');
}

# Send data-only event
async sub send {
    my ($self, $data) = @_;

    croak "Cannot send on closed SSE connection" if $self->is_closed;

    # Auto-start if not started
    await $self->start unless $self->is_started;

    await $self->{send}->({
        type => 'sse.send',
        data => $data,
    });

    return $self;
}

# Send JSON-encoded data
async sub send_json {
    my ($self, $data) = @_;

    croak "Cannot send on closed SSE connection" if $self->is_closed;

    await $self->start unless $self->is_started;

    my $json = JSON::MaybeXS::encode_json($data);

    await $self->{send}->({
        type => 'sse.send',
        data => $json,
    });

    return $self;
}

# Send full SSE event with all fields
async sub send_event {
    my ($self, %opts) = @_;

    croak "Cannot send on closed SSE connection" if $self->is_closed;
    croak "send_event requires 'data' parameter" unless exists $opts{data};

    await $self->start unless $self->is_started;

    # Auto-encode hashref/arrayref data as JSON
    my $data = $opts{data};
    if (ref $data) {
        $data = JSON::MaybeXS::encode_json($data);
    }

    my $event = {
        type => 'sse.send',
        data => $data,
    };

    $event->{event} = $opts{event} if defined $opts{event};
    $event->{id}    = "$opts{id}"  if defined $opts{id};
    $event->{retry} = int($opts{retry}) if defined $opts{retry};

    await $self->{send}->($event);

    return $self;
}

# Safe send - returns bool instead of throwing
async sub try_send {
    my ($self, $data) = @_;
    return 0 if $self->is_closed;

    eval {
        await $self->start unless $self->is_started;
        await $self->{send}->({
            type => 'sse.send',
            data => $data,
        });
    };
    if (my $err = $@) {
        $self->_set_closed;
        await $self->_trigger_error($err);
        return 0;
    }
    return 1;
}

async sub try_send_json {
    my ($self, $data) = @_;
    return 0 if $self->is_closed;

    eval {
        await $self->start unless $self->is_started;
        my $json = JSON::MaybeXS::encode_json($data);
        await $self->{send}->({
            type => 'sse.send',
            data => $json,
        });
    };
    if (my $err = $@) {
        $self->_set_closed;
        await $self->_trigger_error($err);
        return 0;
    }
    return 1;
}

# Send SSE comment (doesn't trigger onmessage in browser)
async sub send_comment {
    my ($self, $comment) = @_;

    croak "Cannot send on closed SSE connection" if $self->is_closed;

    await $self->start unless $self->is_started;

    await $self->{send}->({
        type    => 'sse.comment',
        comment => $comment,
    });

    return $self;
}

async sub try_send_comment {
    my ($self, $comment) = @_;
    return 0 if $self->is_closed;

    eval {
        await $self->start unless $self->is_started;
        await $self->{send}->({
            type    => 'sse.comment',
            comment => $comment,
        });
    };
    if (my $err = $@) {
        $self->_set_closed;
        await $self->_trigger_error($err);
        return 0;
    }
    return 1;
}

async sub try_send_event {
    my ($self, %opts) = @_;
    return 0 if $self->is_closed;

    eval {
        await $self->start unless $self->is_started;

        my $data = $opts{data} // '';
        if (ref $data) {
            $data = JSON::MaybeXS::encode_json($data);
        }

        my $event = {
            type => 'sse.send',
            data => $data,
        };
        $event->{event} = $opts{event} if defined $opts{event};
        $event->{id}    = "$opts{id}"  if defined $opts{id};
        $event->{retry} = int($opts{retry}) if defined $opts{retry};

        await $self->{send}->($event);
    };
    if (my $err = $@) {
        $self->_set_closed;
        await $self->_trigger_error($err);
        return 0;
    }
    return 1;
}

# Internal: trigger error callbacks
async sub _trigger_error {
    my ($self, $error) = @_;

    for my $cb (@{$self->{_on_error}}) {
        eval {
            my $r = $cb->($self, $error);
            if (blessed($r) && $r->isa('Future')) {
                await $r;
            }
        };
        if ($@) {
            warn "PAGI::SSE on_error callback error: $@";
        }
    }

    # If no error handlers registered, warn
    if (!@{$self->{_on_error}}) {
        warn "PAGI::SSE error: $error";
    }
}

# Register close callback
sub on_close {
    my ($self, $callback) = @_;
    push @{$self->{_on_close}}, $callback;
    return $self;
}

# Register error callback
sub on_error {
    my ($self, $callback) = @_;
    push @{$self->{_on_error}}, $callback;
    return $self;
}

# Generic event dispatcher - dispatches to on_close or on_error
sub on {
    my ($self, $event, $callback) = @_;
    if ($event eq 'close') {
        return $self->on_close($callback);
    }
    elsif ($event eq 'error') {
        return $self->on_error($callback);
    }
    else {
        croak "Unknown event type: $event (expected close or error)";
    }
}

# Internal: run all on_close callbacks
async sub _run_close_callbacks {
    my ($self) = @_;

    # Only run once
    return if $self->{_close_callbacks_ran};
    $self->{_close_callbacks_ran} = 1;

    my $reason = $self->{_disconnect_reason} // 'unknown';

    for my $cb (@{$self->{_on_close}}) {
        eval {
            my $r = $cb->($self, $reason);
            if (blessed($r) && $r->isa('Future')) {
                await $r;
            }
        };
        if ($@) {
            warn "PAGI::SSE on_close callback error: $@";
        }
    }

    # Clear all callback arrays to break any closure-based cycles
    $self->{_on_close} = [];
    $self->{_on_error} = [];
}

# Close the connection
async sub close {
    my ($self, %opts) = @_;

    return $self if $self->is_closed;

    # Record the reason so on_close callbacks (and the server access log) see it.
    $self->{_disconnect_reason} //= $opts{reason} // 'app_closed';

    # Tell the server to end the stream now (sse.close). The reason is
    # server-side only and is never written to the wire.
    await $self->{send}->({
        type => 'sse.close',
        (defined $opts{reason} ? (reason => $opts{reason}) : ()),
    });

    $self->_set_closed;
    await $self->_run_close_callbacks;

    return $self;
}

# Wait for disconnect
async sub run {
    my ($self) = @_;

    await $self->start unless $self->is_started;

    # Race incoming events against an explicit close() (which resolves this
    # future via _set_closed), so run() returns when the stream is closed from
    # anywhere -- not only when the client disconnects.
    $self->{_closed_future} //= Future->new;

    while (!$self->is_closed) {
        my $event = eval { await Future->wait_any($self->{receive}->(), $self->{_closed_future}) };
        if (my $err = $@) {
            warn "PAGI::SSE receive error: $err";
            last;
        }

        last if $self->is_closed;   # woken by close()

        my $type = $event->{type} // '';

        if ($type eq 'sse.disconnect') {
            $self->{_disconnect_reason} = $event->{reason} // 'client_closed';
            $self->_set_closed;
            await $self->_run_close_callbacks;
            last;
        }
    }

    return;
}

# Iterate over items and send events
async sub each {
    my ($self, $source, $callback) = @_;

    await $self->start unless $self->is_started;

    my $index = 0;

    # Handle arrayref
    if (ref $source eq 'ARRAY') {
        for my $item (@$source) {
            last if $self->is_closed;

            my $result = await $callback->($item, $index++);

            # If callback returns a hashref, treat as event spec
            if (ref $result eq 'HASH') {
                await $self->send_event(%$result);
            }
        }
    }
    # Handle coderef iterator
    elsif (ref $source eq 'CODE') {
        while (!$self->is_closed) {
            my $item = $source->();
            last unless defined $item;

            my $result = await $callback->($item, $index++);

            if (ref $result eq 'HASH') {
                await $self->send_event(%$result);
            }
        }
    }
    else {
        croak "each() requires arrayref or coderef, got " . ref($source);
    }

    return $self;
}

# Periodic callback execution with interval delay
# Requires Future::IO to be installed
async sub every {
    my ($self, $interval, $callback) = @_;

    croak "every() requires interval" unless defined $interval && $interval > 0;
    croak "every() requires callback coderef" unless ref $callback eq 'CODE';

    # Future::IO is required for every() - fail clearly if not available
    eval { require Future::IO; 1 }
        or croak "every() requires Future::IO to be installed. "
               . "Install it with: cpanm Future::IO";

    # Future::IO must be configured with a backend
    no warnings 'once';
    croak "Future::IO backend not configured. Add to your app.pl:\n"
        . "  use Future::IO::Impl::IOAsync;\n"
        unless $Future::IO::IMPL;

    await $self->start unless $self->is_started;

    # Start background disconnect monitor
    my $disconnect_future = $self->_watch_for_disconnect;

    while (!$self->is_closed) {
        # Execute the callback
        my $ok = eval { await $callback->(); 1 };
        unless ($ok) {
            my $err = $@;
            # Callback failed - connection likely closed or error occurred
            $self->_set_closed;
            await $self->_run_close_callbacks;
            last;
        }

        # Race between sleep and disconnect detection
        my $sleep_future = Future::IO->sleep($interval);
        my $winner = await Future->wait_any($sleep_future, $disconnect_future);

        # If disconnect won, exit the loop
        if ($self->is_closed) {
            $sleep_future->cancel if $sleep_future->can('cancel') && !$sleep_future->is_ready;
            last;
        }
    }

    # Clean up disconnect monitor if still running
    $disconnect_future->cancel if $disconnect_future->can('cancel') && !$disconnect_future->is_ready;

    return $self;
}

# Background future that completes when disconnect is detected
sub _watch_for_disconnect {
    my ($self) = @_;

    my $receive = $self->{receive};
    require Scalar::Util;
    my $weak_self = $self;
    Scalar::Util::weaken($weak_self);

    return (async sub {
        while ($weak_self && !$weak_self->is_closed) {
            my $event = eval { await $receive->() };
            last unless $event;

            my $type = $event->{type} // '';
            if ($type eq 'sse.disconnect') {
                if ($weak_self) {
                    $weak_self->{_disconnect_reason} = $event->{reason} // 'client_closed';
                    $weak_self->_set_closed;
                    await $weak_self->_run_close_callbacks;
                }
                last;
            }
        }
    })->();
}

1;

__END__