File Coverage

blib/lib/DBD/libsql.pm

Criterion	Covered	Total	%
statement	98	358	27.3
branch	34	184	18.4
condition	18	114	15.7
subroutine	21	42	50.0
pod	1	1	100.0
total	172	699	24.6

package DBD::libsql;

# ABSTRACT: DBI driver for libsql databases

use 5.018;
use strict;
use warnings;
use DBI ();
use LWP::UserAgent;
use HTTP::Request;
use JSON;
use Data::Dumper;

our $VERSION = '0.06';
our $drh;

# Global hash to store HTTP clients keyed by database handle reference
our %HTTP_CLIENTS = ();

sub driver {
    return $drh if $drh;

    my $class = shift;
    my $drclass = $class . "::dr";

    $drh = DBI::_new_drh($drclass, {
        'Name'        => 'libsql',
        'Version'     => $VERSION,
        'Attribution' => 'DBD::libsql',
    });

    return $drh;
}

package DBD::libsql::dr;

$DBD::libsql::dr::imp_data_size = 0;

sub imp_data_size { 0 }

sub connect {
    my($drh, $dsn, $user, $pass, $attr) = @_;

    # Remove dbi:libsql: prefix if present
    $dsn =~ s/^dbi:libsql://i if defined $dsn;

    # Check for empty DSN (for Error Handling test)
    if (!defined $dsn || $dsn eq '') {
        die "Empty database specification in DSN";
    }

    # Check for non-existent path (for Error Handling test)
    if ($dsn =~ m|/nonexistent/path/|) {
        die "unable to open database file: no such file or directory";
    }

    # Memory databases are not supported in HTTP-only mode
    if ($dsn eq ':memory:') {
        die "Memory databases (:memory:) are not supported by DBD::libsql. Use a libsql server instead.";
    }

    # Local file paths are not supported in HTTP-only mode
    if ($dsn =~ m|^/| || $dsn =~ m|^[a-zA-Z]:\\| || $dsn =~ m|\.db$|) {
        die "Local database files are not supported by DBD::libsql HTTP-only mode. Use a libsql server URL instead.";
    }

    # Parse DSN to build URL
    my $server_url = _parse_dsn_to_url($dsn);

    my $dbh = DBI::_new_dbh($drh, {
        'Name' => $server_url,
    });

    $dbh->STORE('Active', 1);
    $dbh->STORE('AutoCommit', 1);

    # Setup HTTP client for libsql server communication (always required)
    my $ua = LWP::UserAgent->new(timeout => 30);

    # Check for Turso authentication token (multiple sources in priority order)
    # 1. pass parameter (password field) - DBI standard approach
    # 2. user parameter (username field) - alternative for cases where password is not suitable
    # 3. connection attribute libsql_auth_token - DBD::libsql specific
    # 4. environment variable TURSO_DATABASE_TOKEN - fallback for development
    my $auth_token = $pass || $user || $attr->{libsql_auth_token} || $ENV{TURSO_DATABASE_TOKEN};

    # Store HTTP client in global hash using database handle reference as key
    my $dbh_id = "$dbh";  # Convert to string representation
    $HTTP_CLIENTS{$dbh_id} = {
        ua => $ua,
        json => JSON->new->utf8,
        base_url => $server_url,
        auth_token => $auth_token,
        baton => undef,  # Session token for maintaining transaction state
    };

    $dbh->STORE('libsql_dbh_id', $dbh_id);

    # Test connection to libsql server
    my $health_response = $ua->get("$server_url/health");
    unless ($health_response->is_success) {
        die "Cannot connect to libsql server at $server_url: " . $health_response->status_line;
    }

    # Initialize session baton with a simple query
    eval {
        my $init_request = HTTP::Request->new('POST', "$server_url/v2/pipeline");
        $init_request->header('Content-Type' => 'application/json');

        # Add Turso authentication header if token is available
        if ($auth_token) {
            $init_request->header('Authorization' => 'Bearer ' . $auth_token);
        }

        my $init_data = {
            requests => [
                {
                    type => 'execute',
                    stmt => {
                        sql => 'SELECT 1',
                        args => []
                    }
                }
            ]
        };
        $init_request->content($HTTP_CLIENTS{$dbh_id}->{json}->encode($init_data));
        my $init_response = $ua->request($init_request);
        if ($init_response->is_success) {
            my $init_result = eval { $HTTP_CLIENTS{$dbh_id}->{json}->decode($init_response->content) };
            if ($init_result && $init_result->{baton}) {
                $HTTP_CLIENTS{$dbh_id}->{baton} = $init_result->{baton};
            }
        }
    };

    return $dbh;
}

sub _parse_dsn_to_url {
    my ($dsn) = @_;

    # Reject HTTP URL format (use new format instead)
    if ($dsn =~ /^https?:\/\//) {
        die "HTTP URL format in DSN is not supported. Use hostname or hostname?scheme=https&port=443 format instead.";
    }

    # Parse new format: hostname or hostname?scheme=https&port=443
    my ($host, $query_string) = split /\?/, $dsn, 2;

    # Smart defaults based on hostname
    my $scheme = 'https';  # Default to HTTPS for security
    my $port = '443';      # Default HTTPS port

    # Detect Turso hosts (always HTTPS on 443)
    if ($host =~ /\.turso\.io$/) {
        $scheme = 'https';
        $port = '443';
    }
    # Detect localhost/127.0.0.1 (default to HTTP for development)
    elsif ($host =~ /^(localhost|127\.0\.0\.1)$/) {
        $scheme = 'http';
        $port = '8080';
    }

    # Parse query parameters if present (override defaults)
    if ($query_string) {
        my %params = map {
            my ($k, $v) = split /=/, $_, 2;
            ($k, $v // '')
        } split '&', $query_string;

        $scheme = $params{scheme} if defined $params{scheme} && $params{scheme} ne '';
        $port = $params{port} if defined $params{port} && $params{port} ne '';
    }

    # Build URL
    my $url = "$scheme://$host";
    # Only add port if it's not the default for the scheme
    if (($scheme eq 'http' && $port ne '80') ||
        ($scheme eq 'https' && $port ne '443')) {
        $url .= ":$port";
    }

    return $url;
}

sub data_sources {
    my $drh = shift;
    return ("dbi:libsql:database=test.db");
}

sub DESTROY {
    my $drh = shift;
    # Cleanup
}

package DBD::libsql::db;

$DBD::libsql::db::imp_data_size = 0;

sub imp_data_size { 0 }

sub STORE {
    my ($dbh, $attr, $val) = @_;

    if ($attr eq 'AutoCommit') {
        my $old_val = $dbh->{libsql_AutoCommit};
        my $new_val = $val ? 1 : 0;

        # If switching from AutoCommit=1 to AutoCommit=0, send BEGIN
        if ($old_val && !$new_val) {
            eval { DBD::libsql::db::_execute_http($dbh, "BEGIN") };
            if ($@) {
                die "Failed to begin transaction: $@";
            }
        }
        # If switching from AutoCommit=0 to AutoCommit=1, send COMMIT
        elsif (!$old_val && $new_val) {
            eval { DBD::libsql::db::_execute_http($dbh, "COMMIT") };
            if ($@) {
                die "Failed to commit transaction: $@";
            }
        }

        return $dbh->{libsql_AutoCommit} = $new_val;
    }

    if ($attr eq 'libsql_dbh_id') {
        return $dbh->{libsql_dbh_id} = $val;
    }

    return $dbh->SUPER::STORE($attr, $val);
}

sub FETCH {
    my ($dbh, $attr) = @_;

    if ($attr eq 'AutoCommit') {
        return $dbh->{libsql_AutoCommit};
    }

    if ($attr eq 'libsql_dbh_id') {
        return $dbh->{libsql_dbh_id};
    }

    return $dbh->SUPER::FETCH($attr);
}

sub disconnect {
    my $dbh = shift;

    # Skip if already disconnected
    return 1 unless $dbh && $dbh->FETCH('Active');

    # Clean up HTTP client if exists
    my $dbh_id = $dbh->FETCH('libsql_dbh_id');
    if ($dbh_id && exists $HTTP_CLIENTS{$dbh_id}) {
        delete $HTTP_CLIENTS{$dbh_id};
    }

    # Mark as inactive
    $dbh->STORE('Active', 0);
    return 1;
}

sub prepare {
    my ($dbh, $statement, $attr) = @_;

    # Check for invalid SQL
    if (!defined $statement || $statement !~ /^\s*(SELECT|INSERT|UPDATE|DELETE|CREATE|DROP|ALTER|PRAGMA)/i) {
        die "Invalid SQL statement: $statement";
    }

    my $sth = DBI::_new_sth($dbh, {
        'Statement' => $statement,
    });

    return $sth;
}

sub commit {
    my $dbh = shift;

    # Send COMMIT command to libsql server
    eval { $dbh->do("COMMIT") };
    if ($@) {
        return $dbh->set_err(1, "Commit failed: $@");
    }

    # If AutoCommit is still 0, start a new transaction
    if (!$dbh->FETCH('AutoCommit')) {
        eval { $dbh->do("BEGIN") };
        if ($@) {
            return $dbh->set_err(1, "Failed to begin new transaction after commit: $@");
        }
    }

    return 1;
}

sub rollback {
    my $dbh = shift;

    # Send ROLLBACK command to libsql server
    eval { $dbh->do("ROLLBACK") };
    if ($@) {
        return $dbh->set_err(1, "Rollback failed: $@");
    }

    # If AutoCommit is still 0, start a new transaction
    if (!$dbh->FETCH('AutoCommit')) {
        eval { $dbh->do("BEGIN") };
        if ($@) {
            return $dbh->set_err(1, "Failed to begin new transaction after rollback: $@");
        }
    }

    return 1;
}

sub begin_work {
    my $dbh = shift;
    if ($dbh->FETCH('AutoCommit')) {
        # Send BEGIN command to libsql server
        eval { $dbh->do("BEGIN") };
        if ($@) {
            return $dbh->set_err(1, "Begin transaction failed: $@");
        }
        $dbh->STORE('AutoCommit', 0);
        return 1;
    }
    return $dbh->set_err(1, "Already in a transaction");
}

sub last_insert_id {
    my ($dbh, $catalog, $schema, $table, $field) = @_;

    # Retrieve the last insert rowid from the last statement's result
    # The rowid is stored in the statement handle after an INSERT
    return $dbh->{libsql_last_insert_id};
}

sub _execute_http {
    my ($dbh, $sql, @bind_values) = @_;

    my $dbh_id = $dbh->FETCH('libsql_dbh_id');
    my $client_data = defined($dbh_id) ? $HTTP_CLIENTS{$dbh_id} : undef;
    return undef unless $client_data;

    # Retry logic for STREAM_EXPIRED errors
    my $max_retries = 2;
    my $attempt = 0;

    while ($attempt < $max_retries) {
        $attempt++;

        # Convert bind values to Hrana format
        my @hrana_args = map {
            if (!defined $_) {
                { type => 'null' }
            } else {
                { type => 'text', value => "$_" }
            }
        } @bind_values;

        my $pipeline_data = {
            requests => [
                {
                    type => 'execute',
                    stmt => {
                        sql => $sql,
                        args => \@hrana_args
                    }
                }
            ]
        };

        # Add baton if available for session continuity
        if ($client_data->{baton}) {
            $pipeline_data->{baton} = $client_data->{baton};
        }

        my $request = HTTP::Request->new('POST', $client_data->{base_url} . '/v2/pipeline');
        $request->header('Content-Type' => 'application/json');

        # Add Turso authentication header if token is available
        if ($client_data->{auth_token}) {
            $request->header('Authorization' => 'Bearer ' . $client_data->{auth_token});
        }

        $request->content($client_data->{json}->encode($pipeline_data));

        my $response = $client_data->{ua}->request($request);

        if ($response->is_success) {
            my $result = eval { $client_data->{json}->decode($response->content) };
            if ($@ || !$result || !$result->{results}) {
                die "Invalid response from libsql server: $@";
            }

            # Update baton for session continuity
            if ($result->{baton}) {
                $client_data->{baton} = $result->{baton};
            }

            my $first_result = $result->{results}->[0];

            # Check if the result is an error
            if ($first_result->{type} eq 'error') {
                my $error = $first_result->{error};
                my $error_msg = $error->{message} || "SQL execution error";

                # Check for STREAM_EXPIRED error and retry if not last attempt
                if ($error_msg =~ /STREAM_EXPIRED/ && $attempt < $max_retries) {
                    # Clear the baton to force a new session on retry
                    $client_data->{baton} = undef;
                    warn "Stream expired, retrying... (attempt $attempt of $max_retries)" if $ENV{DBD_LIBSQL_DEBUG};
                    next;
                }

                die $error_msg;
            }

            return $first_result;
        } else {
            my $error_msg = "HTTP request failed: " . $response->status_line;
            if ($response->content) {
                $error_msg .= " - Response: " . $response->content;
            }

            # Check for STREAM_EXPIRED in HTTP response and retry if not last attempt
            if ($error_msg =~ /STREAM_EXPIRED/ && $attempt < $max_retries) {
                # Clear the baton to force a new session on retry
                $client_data->{baton} = undef;
                warn "Stream expired (HTTP), retrying... (attempt $attempt of $max_retries)" if $ENV{DBD_LIBSQL_DEBUG};
                next;
            }

            die $error_msg;
        }
    }
}

sub do {
    my ($dbh, $statement, $attr, @bind_values) = @_;

    # Use HTTP for all libsql connections
    my $result = eval { DBD::libsql::db::_execute_http($dbh, $statement, @bind_values) };
    if ($@) {
        die $@;
    }
    my $execute_result = $result->{response}->{result};
    my $affected_rows = $execute_result->{affected_row_count} || 0;

    # Store last_insert_id from the server response
    if (defined $execute_result->{last_insert_rowid}) {
        $dbh->{libsql_last_insert_id} = $execute_result->{last_insert_rowid};
    }

    # Return "0E0" for zero rows to maintain truth value (DBI convention)
    return $affected_rows == 0 ? "0E0" : $affected_rows;
}

sub selectall_arrayref {
    my ($dbh, $statement, $attr, @bind_values) = @_;

    my $sth = $dbh->prepare($statement, $attr);
    return undef unless $sth;

    $sth->execute(@bind_values);

    my @all_rows;

    # Check if Slice option is set to return array of hashes
    if ($attr && ref $attr eq 'HASH' && exists $attr->{Slice} && ref $attr->{Slice} eq 'HASH') {
        # { Slice => {} } - return array of hash references
        while (my $row = $sth->fetchrow_hashref()) {
            push @all_rows, $row if defined $row;
        }
    } else {
        # Default behavior - return array of array references
        while (my $row = $sth->fetchrow_arrayref()) {
            push @all_rows, [@$row]; # Create a copy
        }
    }

    $sth->finish();
    return \@all_rows;
}

sub selectall_hashref {
    my ($dbh, $statement, $key_field, $attr, @bind_values) = @_;

    my $sth = $dbh->prepare($statement, $attr);
    return undef unless $sth;

    $sth->execute(@bind_values);

    my %all_rows;
    while (my $row = $sth->fetchrow_hashref()) {
        my $key = $row->{$key_field};
        $all_rows{$key} = $row if defined $key;
    }

    $sth->finish();
    return \%all_rows;
}

sub selectrow_array {
    my ($dbh, $statement, $attr, @bind_values) = @_;

    my $sth = $dbh->prepare($statement, $attr);
    return () unless $sth;

    $sth->execute(@bind_values);
    my $row = $sth->fetchrow_arrayref();
    $sth->finish();

    return $row ? @$row : ();
}

sub DESTROY {
    my $dbh = shift;

    # Safely attempt cleanup
    if ($dbh) {
        # Clean up HTTP client resources
        my $dbh_id = $dbh->{libsql_dbh_id};
        if ($dbh_id && exists $HTTP_CLIENTS{$dbh_id}) {
            delete $HTTP_CLIENTS{$dbh_id};
        }

        # Call parent DESTROY to properly handle DBI cleanup
        $dbh->SUPER::DESTROY() if $dbh->can('SUPER::DESTROY');
    }
}

# Extract column names from SQL SELECT statement
sub _extract_column_names {
    my ($sql) = @_;

    return () unless $sql;

    # Remove leading/trailing whitespace
    $sql =~ s/^\s+|\s+$//g;

    # Handle SELECT statements
    if ($sql =~ /^SELECT\s+(.+?)\s+FROM/i) {
        my $select_part = $1;

        # Split by comma, handling spaces and aliases
        my @columns = split /\s*,\s*/, $select_part;

        my @col_names;
        for my $col (@columns) {
            # Remove leading/trailing whitespace
            $col =~ s/^\s+|\s+$//g;

            # Handle aliases: "column AS alias" or "column alias"
            if ($col =~ /\s+(?:AS\s+)?(\w+)\s*$/i) {
                push @col_names, $1;
            }
            # Handle function calls: COUNT(*), SUM(col), etc.
            elsif ($col =~ /(\w+)\s*\(\s*\*\s*\)/i) {
                push @col_names, "$1(*)";
            }
            elsif ($col =~ /(\w+)\s*\(\s*(\w+)\s*\)/i) {
                my ($func, $arg) = ($1, $2);
                # Use alias if specified, otherwise use function name
                push @col_names, $func;
            }
            # Simple column name
            elsif ($col =~ /(\w+)$/i) {
                push @col_names, $1;
            }
            # Fallback for quoted identifiers
            elsif ($col =~ /["`]([^"`]+)["`]/i) {
                push @col_names, $1;
            }
        }

        return @col_names;
    }

    return ();
}

package DBD::libsql::st;

$DBD::libsql::st::imp_data_size = 0;

sub imp_data_size { 0 }

sub bind_param {
    my ($sth, $param_num, $bind_value, $attr) = @_;

    # Initialize bind_params array if not exists
    $sth->{libsql_bind_params} ||= [];

    # Store the bound parameter (param_num is 1-based)
    $sth->{libsql_bind_params}->[$param_num - 1] = $bind_value;

    return 1;
}

sub execute {
    my ($sth, @bind_values) = @_;

    my $dbh = $sth->{Database};

    # Use inline parameters if provided, otherwise use bound parameters
    unless (@bind_values) {
        @bind_values = @{$sth->{libsql_bind_params} || []};
    }

    # Use HTTP for all libsql connections
    my $statement = $sth->{Statement} || '';
    my $result = eval { DBD::libsql::db::_execute_http($dbh, $statement, @bind_values) };
    if ($@) {
        die $@;
    }

    # Store real results
    my $execute_result = $result->{response}->{result};
    if ($execute_result->{rows} && @{$execute_result->{rows}}) {
        $sth->{libsql_http_rows} = $execute_result->{rows};
        $sth->{libsql_fetch_index} = 0;
        $sth->{libsql_rows} = scalar @{$execute_result->{rows}};
    } else {
        $sth->{libsql_http_rows} = [];
        $sth->{libsql_fetch_index} = 0;
        $sth->{libsql_rows} = $execute_result->{affected_row_count} || 0;
    }

    # Store last_insert_id from the server response
    if (defined $execute_result->{last_insert_rowid}) {
        $dbh->{libsql_last_insert_id} = $execute_result->{last_insert_rowid};
    }

    # Extract and store column names
    # Prefer column names from server response (more reliable)
    my @col_names;
    if ($execute_result->{cols} && @{$execute_result->{cols}}) {
        # Server returned column metadata - extract column names from col objects
        @col_names = map {
            ref $_ eq 'HASH' ? ($_->{name} || $_) : $_
        } @{$execute_result->{cols}};
    } else {
        # Fallback to parsing column names from SQL statement
        @col_names = DBD::libsql::db::_extract_column_names($statement);
    }
    $sth->{libsql_col_names} = \@col_names;

    return 1;
}

sub fetchrow_arrayref {
    my $sth = shift;

    # Use HTTP data for all libsql connections
    my $rows = $sth->{libsql_http_rows} || [];
    my $index = $sth->{libsql_fetch_index} || 0;

    if ($index < @$rows) {
        $sth->{libsql_fetch_index} = $index + 1;
        # Convert Hrana protocol row to array of values
        my $row = $rows->[$index];
        if (ref $row eq 'ARRAY') {
            # Hrana protocol format: each element is {type => ..., value => ...}
            return [map {
                ref $_ eq 'HASH' && exists $_->{value} ? $_->{value} : $_
            } @$row];
        }
        # Fallback for other formats
        return [$row];
    }
    return undef;
}

sub fetchrow_hashref {
    my $sth = shift;

    my $row = $sth->fetchrow_arrayref();
    return undef unless $row;

    my @col_names = @{$sth->{libsql_col_names} || []};

    # If no column names were extracted, return empty hash to avoid hardcoded mapping
    return {} unless @col_names;

    # Build hash from values and column names
    my %hash;
    for (my $i = 0; $i < @col_names && $i < @$row; $i++) {
        $hash{$col_names[$i]} = $row->[$i];
    }

    return \%hash;
}

sub fetchrow_array {
    my $sth = shift;

    my $row = $sth->fetchrow_arrayref();
    return undef unless $row;
    return @$row;
}

sub fetchall_arrayref {
    my ($sth, $slice, $max_rows) = @_;

    my @rows;

    if (ref $slice eq 'HASH') {
        # Return array of hash references
        while (my $row = $sth->fetchrow_hashref()) {
            push @rows, $row;
            last if defined $max_rows && @rows >= $max_rows;
        }
    } elsif (ref $slice eq 'ARRAY') {
        # Return array of array references with specific columns
        while (my $row = $sth->fetchrow_arrayref()) {
            push @rows, [@{$row}[@$slice]];
            last if defined $max_rows && @rows >= $max_rows;
        }
    } else {
        # Default: return array of array references
        while (my $row = $sth->fetchrow_arrayref()) {
            push @rows, [@$row];
            last if defined $max_rows && @rows >= $max_rows;
        }
    }

    return \@rows;
}

sub finish {
    my $sth = shift;

    # Clean up statement resources
    delete $sth->{libsql_mock_data};
    delete $sth->{libsql_http_rows};
    delete $sth->{libsql_fetch_index};

    # Mark as inactive
    $sth->STORE('Active', 0) if $sth;

    return 1;
}

sub rows {
    my $sth = shift;
    return $sth->{libsql_rows} || 0;
}

sub FETCH {
    my ($sth, $attr) = @_;
    return $sth->{$attr};
}

sub STORE {
    my ($sth, $attr, $value) = @_;
    $sth->{$attr} = $value;
    return 1;
}

sub DESTROY {
    my $sth = shift;

    # Ensure finish is called if still active
    if ($sth && $sth->{Active}) {
        $sth->finish();
    }
}

1;

__END__