AmberDB

 view release on metacpan or  search on metacpan

lib/AmberDB.pm  view on Meta::CPAN

    $self->match_add( $idx_path, $table_info, \@batch );
    $self->sort_add( $idx_path, $table_info, \@batch );

    # 2. Tiered stream (A: Aktif, B: Pasif/Junk)
    if ( $table_info->{use_junk} ) {
        my ( @active_batch, @junk_batch, @active_rids, @junk_rids );
        for my $rec (@batch) {
            if ( $self->junk_rules( $table_info, @$rec ) ) {
                push @junk_batch, $rec;
                push @junk_rids, $rec->[0];
            }
            else {
                push @active_batch, $rec;
                push @active_rids, $rec->[0];
            }
        }

        if (@active_batch) {
            $self->records_add( $idx_path, $table_info, $tableid, \@active_rids, 'A' );
            $self->search_add( $idx_path, $table_info, $tableid, \@active_batch, 'A' );
            $self->match_add( $idx_path, $table_info, \@active_batch, 'A' );
            $self->sort_add( $idx_path, $table_info, \@active_batch, 'A' ) if $table_info->{sort_block};
            $self->facet_add( $idx_path, $table_info, \@active_batch ) if $table_info->{use_facet};
        }

        if (@junk_batch) {
            $self->records_add( $idx_path, $table_info, $tableid, \@junk_rids, 'B' );
            $self->search_add( $idx_path, $table_info, $tableid, \@junk_batch, 'B' );
            $self->match_add( $idx_path, $table_info, \@junk_batch, 'B' );
            $self->sort_add( $idx_path, $table_info, \@junk_batch, 'B' ) if $table_info->{sort_block};
        }
    }
    elsif ( $table_info->{use_facet} ) {
        $self->facet_add( $idx_path, $table_info, \@batch );
    }

    if ($has_unique) {
        $self->unique_add( $idx_path, $table_info, \@batch );
    }

    $self->slug_add( $idx_path, $table_info, $tableid, \@batch );
    $self->suggest_add( $idx_path, $table_info, $tableid, \@batch );

    unless ($is_async_write) {
        if ( $table_info->{log_owner} ) {
            foreach my $rec (@batch) {
                $self->auth_write( $tableid, $table_path, "add", $rec->[0] );
            }
        }

        unless ( $self->config('no_backup') || $table_info->{no_backup} ) {
            $self->recs_back( "add", $tableid, @batch );
        }
    }

    return \%statu;
}

# Replace the DB record with new data.
# ------------------------------------------------
sub update_id {

    my ( $self, $tableid, $rid, @record ) = @_;

    # Perform the checks.
    $tableid or return;

    if ( ref($rid) eq 'HASH' && !@record ) {
        if ( $self->schema_matches_hash( $tableid, $rid ) ) {
            my $h = { %$rid };
            $rid = $h->{id} // $h->{ID};
            @record = ($h);
        }
        else {
            my $h = { %$rid };
            $rid = delete $h->{id} // delete $h->{ID};
            @record = ($h);
        }
    }

    $rid = $self->id_check( $tableid, $rid );
    $rid or return;

    # Deflate ONLY if table has a schema AND hash matches schema
    if ( @record && ref($record[0]) eq 'HASH' ) {
        if ( $self->schema_matches_hash( $tableid, $record[0] ) ) {
            my $h = { %{ $record[0] } };
            $h->{id} //= $rid;
            my $def = $self->deflate( $tableid, $h );
            if ( ref($def) eq 'ARRAY' ) {
                @record = ( scalar(@$def) > 1 ) ? @{$def}[ 1 .. $#$def ] : ();
            }
        }
    }

    my $table_info = $self->table_info($tableid);
    (undef, @record) = $self->repeat_fields( $table_info, $rid, @record );

    my $table_path = $self->table_path($tableid);
    my $file_path  = "$table_path.$self->{db_ext}";

    # Write authority cancelled.
    $self->config('no_write')
      and do { $self->transact_error( $file_path, "No authority to write to the file" ); return; };

    # Transaction journal & record locking (Lock BEFORE reading/modifying - Strict 2PL)
    my $is_txn = ( $self->{_txn} && $self->{_txn}->{active} ) ? 1 : 0;
    $self->flock_open( $tableid, "write", $rid );
    if ($is_txn) {
        $self->{_txn}->{locks}->{"${tableid}_${rid}"} = 1;
    }

    my $use_ramdisk = $table_info ? ( $table_info->{use_ramdisk} // $table_info->{use_cache} // 0 ) : 0;
    $use_ramdisk = $self->_normalize_ramdisk_tier($use_ramdisk);
    my $ramdisk_path;
    if ($use_ramdisk) {
        $self->ramdisk_ensure($tableid);
        $ramdisk_path = $self->ramdisk_path($tableid);
    }

    my $is_async_write = ( $use_ramdisk == 4 && !$is_txn );

lib/AmberDB.pm  view on Meta::CPAN

    if ($is_async_write) {
        $self->ramdisk_mark_dirty( $file_path, $rid, 2 );
    }
    elsif ( $use_ramdisk == 4 && $is_txn ) {
        $self->ramdisk_unmark_dirty( $file_path, $rid );
    }

    $self->table_close($file_path) unless ( $is_async_write || $use_ramdisk == 3 );
    $self->table_close($ram_file) if $ram_file;
    unless ($is_txn) { $self->flock_close( $tableid, $rid ); }

    # Cache invalidate
    $self->clear_cache($tableid, $rid);

    my @new_rec = ( $rid, @record );

    # text backup record.
    unless ( $is_async_write || $use_ramdisk == 3 ) {
        $self->recs_back( "edit", $tableid, \@new_rec )
          or cluck "[DB_TIE] Backup error (edit). $tableid\n";
    }

    if ( $self->config('use_simple') || ( $table_info && $table_info->{use_simple} ) ) {
        $self->_invalidate_child_ramdisk_tables($tableid);
        return $rid;
    }

    my @old_rec = $old_record ? ( $rid, $self->db_decode($old_record) ) : ($rid);

    # Index update (search, match, facet, sort)
    my @pairs = ( [ $rid, \@old_rec, \@new_rec ] );
    my $idx_path = ( $use_ramdisk && $ramdisk_path ) ? $ramdisk_path : $table_path;

    # 1. Base stream unconditional index updates
    $self->search_modify( $idx_path, $table_info, $tableid, \@pairs );
    $self->match_modify( $idx_path, $table_info, \@pairs );

    # 2. Tiered stream updates
    if ( $table_info->{use_junk} ) {
        $self->junk_transition( $idx_path, $table_info, $tableid, \@pairs );
    }
    elsif ( $table_info->{use_facet} ) {
        $self->facet_modify( $idx_path, $table_info, \@pairs );
    }
    $self->sort_modify( $idx_path, $table_info, \@pairs );
    $self->unique_modify( $idx_path, $table_info, \@pairs );

    $self->slug_modify( $idx_path, $table_info, $tableid, \@pairs );
    $self->suggest_add( $idx_path, $table_info, $tableid, [ \@new_rec ] );

    # Authorization
    $self->auth_write( $tableid, $table_path, "edit", $rid ) unless ( $is_async_write || $use_ramdisk == 3 );

    $self->_invalidate_child_ramdisk_tables($tableid);
    return 1;
}

# List record modify.
# Note: Bulk operations (insert_list, modify_list, delete_list) do NOT use transactions (_txn_log).
# ------------------------------------------------
sub update_list {

    my ( $self, $tableid, @records ) = @_;

    local $self->{_no_txn} = 1;

    $tableid        or return {};
    scalar @records or return {};

    # Deflate ONLY if table has a schema and records match schema
    my $should_deflate = 0;
    if ( $self->has_schema($tableid) ) {
        my $first_rec = ( @records == 1 && ref($records[0]) eq 'ARRAY' && @{$records[0]} && ref($records[0]->[0]) eq 'HASH' )
          ? $records[0]->[0]
          : ( ref($records[0]) eq 'HASH' ? $records[0] : ( ( @records > 1 && ref($records[1]) eq 'HASH' ) ? $records[1] : undef ) );
        if ( $first_rec && $self->schema_matches_hash( $tableid, $first_rec ) ) {
            $should_deflate = 1;
        }
    }

    if ($should_deflate) {
        if ( @records == 1 && ref($records[0]) eq 'ARRAY' && @{$records[0]} && ref($records[0]->[0]) eq 'HASH' ) {
            @records = $self->deflate( $tableid, @{ $records[0] } );
        }
        elsif ( @records == 1 && ref($records[0]) eq 'HASH' ) {
            @records = $self->deflate( $tableid, $records[0] );
        }
        elsif ( grep { ref($_) eq 'HASH' } @records ) {
            @records = $self->deflate( $tableid, @records );
        }
    }

    # Write authority cancelled.
    $self->config('no_write')
      and do { cluck "[DB_TIE] No authority to write to the file.\n"; return; };

    my $table_info = $self->table_info($tableid);
    my $is_simple  = $self->config('use_simple') || ( $table_info && $table_info->{use_simple} );

    my $table_path = $self->table_path($tableid);
    my $file_path  = "$table_path.$self->{db_ext}";

    my $use_ramdisk = $table_info ? ( $table_info->{use_ramdisk} // $table_info->{use_cache} // 0 ) : 0;
    $use_ramdisk = $self->_normalize_ramdisk_tier($use_ramdisk);
    my $ramdisk_path;
    if ($use_ramdisk) {
        $self->ramdisk_ensure($tableid);
        $ramdisk_path = $self->ramdisk_path($tableid);
    }

    my $is_txn         = ( $self->{_txn} && $self->{_txn}->{active} ) ? 1 : 0;
    my $is_async_write = ( $use_ramdisk == 4 && !$is_txn );
    my $ram_file       = $ramdisk_path ? "$ramdisk_path.$self->{db_ext}" : undef;
    my $target_file    = ( $is_async_write && $ram_file ) ? $ram_file : $file_path;

    my $has_unique = ( $table_info->{valid} && grep { /unique/i } values %{ $table_info->{valid} } ) ? 1 : 0;
    my $has_repeat = ( $table_info->{repeat} && %{ $table_info->{repeat} } ) ? 1 : 0;

    my ( @valid_inputs, @input_rids );
    foreach my $record (@records) {
        my ( $rid, @data ) = @$record;

lib/AmberDB.pm  view on Meta::CPAN

            foreach my $rec (@records_to_put) {
                $self->ramdisk_mark_dirty( $file_path, $rec->[0], 2 );
            }
        }
        elsif ( $use_ramdisk == 4 && $is_txn ) {
            foreach my $rec (@records_to_put) {
                $self->ramdisk_unmark_dirty( $file_path, $rec->[0] );
            }
        }
    }
    $self->table_close($file_path);
    $self->table_close($ram_file) if $ram_file;

    return \%statu unless @pairs;
    return \%statu if $is_simple;

    my $idx_path = ( $use_ramdisk && $ramdisk_path ) ? $ramdisk_path : $table_path;

    # Phase 2: bulk index updates
    # 1. Base stream unconditional index updates
    $self->search_modify( $idx_path, $table_info, $tableid, \@pairs );
    $self->match_modify( $idx_path, $table_info, \@pairs );

    # 2. Tiered stream updates
    if ( $table_info->{use_junk} ) {
        $self->junk_transition( $idx_path, $table_info, $tableid, \@pairs );
    }
    elsif ( $table_info->{use_facet} ) {
        $self->facet_modify( $idx_path, $table_info, \@pairs );
    }
    $self->sort_modify( $idx_path, $table_info, \@pairs );
    $self->unique_modify( $idx_path, $table_info, \@pairs );

    $self->slug_modify( $idx_path, $table_info, $tableid, \@pairs );

    if ( $table_info->{log_owner} && !$is_async_write ) {
        foreach my $pair (@pairs) {
            $self->auth_write( $tableid, $table_path, "edit", $pair->[0] );
        }
    }

    unless ( $is_async_write || $self->config('no_backup') || $table_info->{no_backup} ) {
        $self->recs_back( "edit", $tableid, map { $_->[2] } @pairs );
    }

    return \%statu;
}

# Standard SQL/CRUD aliases
sub modify_id   { shift->update_id(@_) }
sub modify_list { shift->update_list(@_) }

# Updates a field value in an existing record.
# Supports updating fixed schema blocks:
#   $adb->update_field($table, $rid, "price", 1750)
#   $adb->update_field($table, $rid, block => "price", 1750)
# Supports updating repeating child items by 'id' or 'pos':
#   $adb->update_field($table, $rid, id => 202, $new_item)
#   $adb->update_field($table, $rid, pos => 0, $new_item)
# ------------------------------------------------
sub update_field {
    my $self    = shift;
    my $tableid = shift;
    my $rid     = shift;

    $tableid or return;
    $rid = $self->id_check( $tableid, $rid );
    $rid or return;

    my ( $target_spec, $new_val );
    if ( @_ == 3 && !ref($_[0]) ) {
        my ( $k, $v, $val ) = @_;
        $target_spec = { $k => $v };
        $new_val     = $val;
    }
    elsif ( @_ == 2 ) {
        ( $target_spec, $new_val ) = @_;
        if ( !ref($target_spec) ) {
            $target_spec = { block => $target_spec };
        }
    }
    else {
        return;
    }

    my $table_info = $self->table_info($tableid);
    my $rep_start  = $table_info->{repeat_start};
    if ( !defined $rep_start && $table_info->{blocks} ) {
        for my $i ( 0 .. $#{ $table_info->{blocks} } ) {
            my $b = $table_info->{blocks}->[$i];
            if ( ref($b) eq 'HASH' && defined $b->{type} && lc( $b->{type} ) eq 'repeat' ) {
                $rep_start = $i;
                last;
            }
        }
    }

    my $table_path = $self->table_path($tableid);
    my $file_path  = "$table_path.$self->{db_ext}";

    $self->config('no_write')
      and do { $self->transact_error( $file_path, "No authority to write to the file" ); return; };

    my $is_txn = ( $self->{_txn} && $self->{_txn}->{active} ) ? 1 : 0;
    $self->flock_open( $tableid, "write", $rid );
    if ($is_txn) {
        $self->{_txn}->{locks}->{"${tableid}_${rid}"} = 1;
    }

    $self->table_write($file_path)
      or do {
          unless ($is_txn) { $self->flock_close( $tableid, $rid ); }
          $self->transact_error( $file_path, "$file_path can't open" );
          return;
      };

    my $old_record = $self->recs_get( $file_path, $rid )->{$rid};
    if ( !$table_info->{force} && !$old_record ) {
        $self->table_close($file_path);
        unless ($is_txn) { $self->flock_close( $tableid, $rid ); }
        $self->transact_error( $file_path, "Record not exist: $rid" );