view release on metacpan or  search on metacpan

lib/DBIx/Class/Async/Schema.pm  view on Meta::CPAN

        $METRICS->set($name, @args);
    }

    return;
}

sub _resolve_source {
    my ($self, $source_name) = @_;

    croak "Missing source name." unless defined $source_name;

    my $schema_class = $self->{_async_db}{_schema_class};

    croak "Schema class not found." unless defined $schema_class;

    # 1. Ask the main DBIC Schema class for the source metadata
    # We call this on the class name, not an instance, to stay "light"
    my $source = eval { $schema_class->source($source_name) };

    if ($@ || !$source) {
        croak "Could not resolve source '$source_name' in $schema_class: $@";
    }

    # 2. Extract only what we need for the Async side
    return {
        result_class  => $source->result_class,
        columns       => [ $source->columns ],
        relationships => {
            # We map relationships to know how to handle joins/prefetch later
            map { $_ => $source->relationship_info($_) } $source->relationships
        },
    };
}

sub _build_inflator_map {
    my ($self, $schema) = @_;

    my $map = {};
    foreach my $source_name ($schema->sources) {
        my $source = $schema->source($source_name);
        foreach my $col ($source->columns) {
            my $info = $source->column_info($col);

            # Extract both inflate and deflate coderefs
            if ($info->{deflate} || $info->{inflate}) {
                $map->{$source_name}{$col} = {
                    deflate => $info->{deflate},
                    inflate => $info->{inflate},
                };
            }
            # Explicit check for JSON Serializer
            elsif ($info->{serializer_class}
                   && $info->{serializer_class} eq 'JSON') {
                require JSON;
                my $json = JSON->new->utf8->allow_nonref;
                $map->{$source_name}{$col} = {
                    inflate => sub {
                        my $val = shift;
                        return $val if !defined $val || ref($val);

                        my $decoded;
                        eval  {
                            $decoded = $json->decode($val); 1;
                        }
                        or do {
                            warn "Failed to inflate JSON in $col: $@ (Value: $val)";
                            return $val;
                        };
                        return $decoded;
                    },
                    deflate => sub {
                        my $val = shift;
                        return $val if !defined $val || !ref($val);

                        return $json->encode($val);
                    },
                };
            }
        }
    }

    return $map;
}

=head1 METADATA & REFLECTION

=over 4

=item B<source( $source_name )>

Returns the L<DBIx::Class::ResultSource> for the given name.

Unlike standard DBIC, this uses a B<persistent metadata provider> (a cached
internal schema) to ensure that ResultSource objects remain stable across
async calls without re-connecting to the database unnecessarily.

=item B<sources()>

Returns a list of all source names available in the schema. This creates a
temporary, light-weight connection to extract the current schema structure
and then immediately disconnects to save resources.

=item B<class( $source_name )>

Returns the Result Class string (e.g., C<MyApp::Schema::Result::User>) for
the given source. Useful for dynamic inspections.

=item B<schema_class()>

Returns the name of the base L<DBIx::Class> schema class being proxied.

=item B<schema_version()>

Returns the version number defined in your DBIC Schema class, if available.

=back

=head1 TRANSACTION MANAGEMENT

=over 4

=item B<txn_begin / txn_commit / txn_rollback>

These methods return L<Future> objects. Because workers are persistent,
calling C<txn_begin> pins your current logic to a specific worker until
C<commit> or C<rollback> is called.

B<Note:> Use these carefully in an async loop to avoid "Worker Starvation"
where all workers are waiting on long-running manual transactions.