Merge pull request #357 from sgbett/feat/351-postgres-auto-fund

feat(wallet-postgres): implement auto-fund storage methods

Author: sgbett

gem/bsv-wallet-postgres/lib/bsv/wallet_postgres/migrations/004_add_pending_metadata.rb

@@ -0,0 +1,69 @@
+# frozen_string_literal: true
+
+# Add pending lock metadata columns and satoshis to wallet_outputs.
+#
+# Introduces dedicated columns to support the auto-fund UTXO management
+# methods (+find_spendable_outputs+, +lock_utxos+, +update_output_state+,
+# +release_stale_pending!+) without requiring expensive JSONB extraction
+# on every query.
+#
+# === Columns added
+#
+# * +satoshis+          — bigint, nullable. Mirrors the value stored inside
+#                         the JSONB +data+ blob so that +find_spendable_outputs+
+#                         can +ORDER BY satoshis+ without casting. New writes
+#                         populate this column; existing rows leave it NULL
+#                         and queries fall back to +COALESCE(satoshis, (data->>'satoshis')::bigint, 0)+.
+#
+# * +pending_since+     — timestamp (UTC), nullable. Set to +NOW()+ when a
+#                         UTXO is locked via +lock_utxos+. Used by
+#                         +release_stale_pending!+ to identify stale locks.
+#
+# * +pending_reference+ — text, nullable. Caller-supplied label passed to
+#                         +lock_utxos+. Carried through for observability.
+#
+# * +no_send+           — boolean, default +false+. When +true+ the lock is
+#                         exempt from automatic stale recovery by
+#                         +release_stale_pending!+.
+#
+# === Index added
+#
+# A partial index on +(state, basket)+ filtered to rows where
+# +state IS NULL OR state = 'spendable'+ accelerates the hot path of
+# +find_spendable_outputs+: the vast majority of rows in a live wallet
+# are spendable, not pending or spent.
+#
+# === Backward compatibility
+#
+# All new columns are nullable (or carry a safe default). No existing rows
+# are modified. The application layer handles NULL values via +COALESCE+ or
+# explicit IS NULL checks.
+Sequel.migration do
+  up do
+    alter_table(:wallet_outputs) do
+      add_column :satoshis,          :bigint, null: true
+      add_column :pending_since,     :timestamptz, null: true
+      add_column :pending_reference, String,   null: true
+      add_column :no_send,           :boolean, null: false, default: false
+    end
+
+    # Partial index on spendable rows only — keeps the index small and
+    # fast for the typical coin-selection scan.
+    run <<~SQL
+      CREATE INDEX wallet_outputs_spendable_basket_idx
+        ON wallet_outputs (state, basket)
+        WHERE (state IS NULL OR state = 'spendable');
+    SQL
+  end
+
+  down do
+    run 'DROP INDEX IF EXISTS wallet_outputs_spendable_basket_idx;'
+
+    alter_table(:wallet_outputs) do
+      drop_column :no_send
+      drop_column :pending_reference
+      drop_column :pending_since
+      drop_column :satoshis
+    end
+  end
+end

gem/bsv-wallet-postgres/lib/bsv/wallet_postgres/postgres_store.rb

@@ -109,7 +109,14 @@ def store_output(output_data)
         @db[:wallet_outputs]
           .insert_conflict(
             target: :outpoint,
-            update: { basket: row[:basket], tags: row[:tags], spendable: row[:spendable], data: row[:data] }
+            update: {
+              basket: row[:basket],
+              tags: row[:tags],
+              spendable: row[:spendable],
+              state: row[:state],
+              satoshis: row[:satoshis],
+              data: row[:data]
+            }
           )
           .insert(row)
         output_data
@@ -128,6 +135,154 @@ def delete_output(outpoint)
         @db[:wallet_outputs].where(outpoint: outpoint).delete.positive?
       end
 
+      # Returns outputs whose effective state is +:spendable+.
+      #
+      # Legacy rows with +state = NULL+ are treated as spendable when the
+      # +spendable+ boolean is true (or absent), matching MemoryStore's
+      # effective_state logic.
+      #
+      # @param basket [String, nil] restrict to this basket when provided
+      # @param min_satoshis [Integer, nil] exclude outputs below this value
+      # @param sort_order [Symbol] +:asc+ or +:desc+ (default +:desc+, largest first)
+      # @return [Array<Hash>]
+      def find_spendable_outputs(basket: nil, min_satoshis: nil, sort_order: :desc)
+        ds = @db[:wallet_outputs]
+             .where(Sequel.lit('(state = ? OR (state IS NULL AND spendable = TRUE))', 'spendable'))
+        ds = ds.where(basket: basket) if basket
+        if min_satoshis
+          ds = ds.where(
+            Sequel.lit('COALESCE(satoshis, (data->>?)::bigint, 0) >= ?', 'satoshis', min_satoshis)
+          )
+        end
+        satoshis_expr = Sequel.lit('COALESCE(satoshis, (data->>?)::bigint, 0)', 'satoshis')
+        ds = ds.order(sort_order == :asc ? Sequel.asc(satoshis_expr) : Sequel.desc(satoshis_expr))
+        ds.all.map { |r| symbolise_keys(r[:data]) }
+      end
+
+      # Transitions the state of an existing output.
+      #
+      # When +new_state+ is +:pending+, sets +pending_since+, +pending_reference+,
+      # and +no_send+, and merges those values into the JSONB +data+ blob.
+      #
+      # When transitioning away from +:pending+, clears the pending metadata
+      # columns and removes the corresponding keys from the JSONB blob.
+      #
+      # @param outpoint [String] the outpoint identifier
+      # @param new_state [Symbol] +:spendable+, +:pending+, or +:spent+
+      # @param pending_reference [String, nil] caller-supplied label for a pending lock
+      # @param no_send [Boolean, nil] true if the lock belongs to a no_send transaction
+      # @raise [BSV::Wallet::WalletError] if the outpoint is not found
+      # @return [Hash] the updated output hash
+      def update_output_state(outpoint, new_state, pending_reference: nil, no_send: nil)
+        state_str = new_state.to_s
+
+        # Keep legacy spendable boolean in sync so filter_outputs and other
+        # queries that haven't migrated to the state column still work.
+        spendable_bool = new_state == :spendable
+
+        if new_state == :pending
+          updates = {
+            state: state_str,
+            spendable: spendable_bool,
+            pending_since: Sequel.lit('NOW()'),
+            pending_reference: pending_reference,
+            no_send: no_send ? true : false,
+            data: Sequel.lit(
+              "data || jsonb_build_object('state', ?, 'pending_since', NOW()::text, 'pending_reference', ?, 'no_send', ?)",
+              state_str, pending_reference, no_send ? true : false
+            )
+          }
+        else
+          updates = {
+            state: state_str,
+            spendable: spendable_bool,
+            pending_since: nil,
+            pending_reference: nil,
+            no_send: false
+          }
+          # Remove pending keys from JSONB blob, update state
+          updates[:data] = Sequel.lit(
+            "(data - 'pending_since' - 'pending_reference' - 'no_send') || jsonb_build_object('state', ?)",
+            state_str
+          )
+        end
+
+        ds = @db[:wallet_outputs].where(outpoint: outpoint)
+        rows_updated = ds.update(updates)
+        raise WalletError, "Output not found: #{outpoint}" if rows_updated.zero?
+
+        row = ds.first
+        symbolise_keys(row[:data])
+      end
+
+      # Atomically marks a set of outpoints as +:pending+.
+      #
+      # Uses +UPDATE ... WHERE state = 'spendable' ... RETURNING outpoint+ so that
+      # the check-and-set is atomic at the database level. A concurrent caller that
+      # wins the race will have already changed the state to 'pending', so the
+      # second caller's WHERE clause will not match and will return nothing. No
+      # explicit row-level locking is needed — the UPDATE itself takes the lock.
+      #
+      # Legacy rows with +state = NULL AND spendable = TRUE+ are also eligible.
+      #
+      # @param outpoints [Array<String>] outpoint identifiers to lock
+      # @param reference [String] caller-supplied pending reference
+      # @param no_send [Boolean] true if this is a no_send lock
+      # @return [Array<String>] outpoints that were actually locked
+      def lock_utxos(outpoints, reference:, no_send: false)
+        return [] if outpoints.empty?
+
+        rows = @db[:wallet_outputs]
+               .where(outpoint: outpoints)
+               .where(Sequel.lit('(state = ? OR (state IS NULL AND spendable = TRUE))', 'spendable'))
+               .returning(:outpoint)
+               .update(
+                 state: 'pending',
+                 spendable: false,
+                 pending_since: Sequel.lit('NOW()'),
+                 pending_reference: reference,
+                 no_send: no_send ? true : false,
+                 data: Sequel.lit(
+                   "data || jsonb_build_object('state', 'pending', 'pending_since', NOW()::text, " \
+                   "'pending_reference', ?, 'no_send', ?)",
+                   reference, no_send ? true : false
+                 )
+               )
+
+        rows.map { |r| r[:outpoint] }
+      end
+
+      # Releases stale pending locks back to +:spendable+.
+      #
+      # Any output in +:pending+ state whose +pending_since+ is older than
+      # +timeout+ seconds is reset to +spendable+ and its pending metadata is
+      # cleared. Outputs with +no_send = true+ are exempt and remain pending.
+      # Outputs with +pending_since = NULL+ are also skipped — they are treated
+      # as freshly locked (NULL means "just acquired but no timestamp yet").
+      #
+      # @param timeout [Integer] age in seconds before a lock is considered stale (default 300)
+      # @return [Integer] number of outputs released
+      def release_stale_pending!(timeout: 300)
+        rows = @db[:wallet_outputs]
+               .where(state: 'pending')
+               .where(Sequel.lit('no_send IS NOT TRUE'))
+               .where(Sequel.lit('pending_since IS NOT NULL'))
+               .where(Sequel.lit('pending_since < (NOW() - INTERVAL ?)', "#{timeout} seconds"))
+               .returning(:outpoint)
+               .update(
+                 state: 'spendable',
+                 spendable: true,
+                 pending_since: nil,
+                 pending_reference: nil,
+                 no_send: false,
+                 data: Sequel.lit(
+                   "(data - 'pending_since' - 'pending_reference' - 'no_send') || jsonb_build_object('state', 'spendable')"
+                 )
+               )
+
+        rows.length
+      end
+
       # --- Certificates ---
 
       def store_certificate(cert_data)
@@ -207,11 +362,14 @@ def action_row(data)
 
       def output_row(data)
         spendable = data[:spendable] != false # nil treated as spendable, like MemoryStore
+        state = data[:state]&.to_s
         {
           outpoint: data[:outpoint],
           basket: data[:basket],
           tags: Sequel.pg_array(Array(data[:tags]), :text),
           spendable: spendable,
+          state: state,
+          satoshis: data[:satoshis],
           data: Sequel.pg_jsonb(data.to_h)
         }
       end
@@ -236,7 +394,7 @@ def filter_outputs(ds, query)
         ds = ds.where(outpoint: query[:outpoint]) if query[:outpoint]
         ds = ds.where(basket: query[:basket]) if query[:basket]
         ds = apply_array_filter(ds, :tags, query[:tags], query[:tag_query_mode])
-        ds = ds.where(spendable: true) unless query[:include_spent]
+        ds = ds.where(Sequel.lit('(state = ? OR (state IS NULL AND spendable = TRUE))', 'spendable')) unless query[:include_spent]
         ds
       end