Fix: discard interrupted SQLite connections and make asset reads stable under churn

internal/infrastructure/db/postgres/asset_repo.go

@@ -125,43 +125,46 @@ func (r *assetRepository) GetAssets(
 	if len(assetIds) == 0 {
 		return nil, nil
 	}
-	var assets []domain.Asset
-	txBody := func(querierWithTx *queries.Queries) error {
-		rows, err := querierWithTx.SelectAssetsByIds(ctx, assetIds)
-		if err != nil {
-			return err
-		}
-		assets = make([]domain.Asset, 0, len(rows))
-		for _, row := range rows {
-			supplyStr, err := querierWithTx.SelectAssetSupply(ctx, row.ID)
-			if err != nil {
-				return fmt.Errorf("failed to compute supply for asset %s: %w", row.ID, err)
-			}
-			supply := new(big.Int)
-			if _, ok := supply.SetString(supplyStr, 10); !ok {
-				return fmt.Errorf("invalid supply value: %s", supplyStr)
-			}
+
+	rows, err := r.querier.SelectAssetsWithUnspentAmountsByIds(ctx, assetIds)
+	if err != nil {
+		return nil, err
+	}
+
+	assets := make([]domain.Asset, 0, len(rows))
+	indexByID := make(map[string]int, len(rows))
+	for _, row := range rows {
+		idx, ok := indexByID[row.ID]
+		if !ok {
 			ast := domain.Asset{
 				Id:             row.ID,
 				ControlAssetId: row.ControlAssetID.String,
-				Supply:         *supply,
+				Supply:         *big.NewInt(0),
 			}
+
 			if row.Metadata.Valid {
 				// Parsing metadata should never fail but if it does we just return an empty list
 				// of metadata and log the error
-				ast.Metadata, err = asset.NewMetadataListFromString(row.Metadata.String)
-				if err != nil {
-					log.WithError(err).Warnf("failed to parse metadata for asset %s", row.ID)
+				metadata, parseErr := asset.NewMetadataListFromString(row.Metadata.String)
+				if parseErr != nil {
+					log.WithError(parseErr).Warnf("failed to parse metadata for asset %s", row.ID)
+				} else {
+					ast.Metadata = metadata
 				}
 			}
 
 			assets = append(assets, ast)
+			idx = len(assets) - 1
+			indexByID[row.ID] = idx
 		}
-		return nil
-	}
-	if err := execTx(ctx, r.db, txBody); err != nil {
-		return nil, err
+
+		amount, ok := new(big.Int).SetString(row.AssetAmount, 10)
+		if !ok {
+			return nil, fmt.Errorf("invalid supply value: %s", row.AssetAmount)
+		}
+		assets[idx].Supply.Add(&assets[idx].Supply, amount)
 	}
+
 	return assets, nil
 }
 

internal/infrastructure/db/postgres/sqlc/query.sql

@@ -441,6 +441,22 @@ VALUES (@asset_id, @txid, @vout, @amount);
 -- name: SelectAssetsByIds :many
 SELECT * FROM asset WHERE asset.id = ANY($1::varchar[]);
 
+-- name: SelectAssetsWithUnspentAmountsByIds :many
+SELECT
+  a.id,
+  a.is_immutable,
+  a.metadata_hash,
+  a.metadata,
+  a.control_asset_id,
+  COALESCE(v.asset_amount, 0)::TEXT AS asset_amount
+FROM asset a
+LEFT JOIN vtxo_vw v
+  ON v.asset_id = a.id
+ AND v.spent = false
+ AND v.asset_amount > 0
+WHERE a.id = ANY($1::varchar[])
+ORDER BY a.id;
+
 -- name: SelectAssetSupply :one
 SELECT (COALESCE(SUM(ap.amount), 0))::TEXT AS supply
 FROM asset_projection ap
@@ -451,4 +467,4 @@ WHERE ap.asset_id = $1 AND v.spent = false;
 SELECT control_asset_id FROM asset WHERE id = $1;
 
 -- name: SelectAssetExists :one
-SELECT 1 FROM asset WHERE id = $1 LIMIT 1;
+SELECT 1 FROM asset WHERE id = $1 LIMIT 1;

internal/infrastructure/db/service.go

@@ -310,12 +310,16 @@ func NewService(config ServiceConfig, txDecoder ports.TxDecoder) (ports.RepoMana
 		}
 
 		dbFile := filepath.Join(baseDir, sqliteDbFile)
-		db, err := sqlitedb.OpenDb(dbFile)
+		db, err := sqlitedb.OpenDb(
+			dbFile,
+			sqlitedb.WithJournalModeWAL(),
+			sqlitedb.WithBusyTimeout(5*time.Second),
+		)
 		if err != nil {
 			return nil, fmt.Errorf("failed to open db: %s", err)
 		}
 
-		driver, err := sqlitemigrate.WithInstance(db, &sqlitemigrate.Config{})
+		driver, err := sqlitemigrate.WithInstance(db.Write(), &sqlitemigrate.Config{})
 		if err != nil {
 			return nil, fmt.Errorf("failed to init driver: %s", err)
 		}
@@ -330,7 +334,7 @@ func NewService(config ServiceConfig, txDecoder ports.TxDecoder) (ports.RepoMana
 			return nil, fmt.Errorf("failed to create migration instance: %s", err)
 		}
 
-		err = handleIntentTxidMigration(m, db, config.DataStoreType)
+		err = handleIntentTxidMigration(m, db.Write(), config.DataStoreType)
 		if err != nil {
 			return nil, fmt.Errorf("failed to handle intent txid migration: %w", err)
 		}

internal/infrastructure/db/service_test.go

@@ -186,6 +186,7 @@ func TestService(t *testing.T) {
 			testOffchainTxRepository(t, svc)
 			testAssetRepository(t, svc)
 			testVtxoRepository(t, svc)
+			testAssetRepositorySpentOnlySupply(t, svc)
 			testScheduledSessionRepository(t, svc)
 			testConvictionRepository(t, svc)
 			testFeeRepository(t, svc)
@@ -1769,6 +1770,53 @@ func testAssetRepository(t *testing.T, svc ports.RepoManager) {
 	})
 }
 
+func testAssetRepositorySpentOnlySupply(t *testing.T, svc ports.RepoManager) {
+	t.Run("test_asset_repository_spent_only_supply", func(t *testing.T) {
+		ctx := t.Context()
+		repo := svc.Assets()
+		vtxoRepo := svc.Vtxos()
+
+		assetID := randomString(16)
+		vtxoTxid := randomString(32)
+		spentBy := randomString(32)
+		arkTxid := randomString(32)
+
+		count, err := repo.AddAssets(ctx, map[string][]domain.Asset{"spentOnlyAssetTx": {
+			{
+				Id:       assetID,
+				Metadata: []asset.Metadata{},
+			},
+		}})
+		require.NoError(t, err)
+		require.Equal(t, 1, count)
+
+		spentOnlyVtxo := domain.Vtxo{
+			Outpoint: domain.Outpoint{
+				Txid: vtxoTxid,
+				VOut: 0,
+			},
+			Amount: 330,
+			Assets: []domain.AssetDenomination{{
+				AssetId: assetID,
+				Amount:  42,
+			}},
+		}
+		err = vtxoRepo.AddVtxos(ctx, []domain.Vtxo{spentOnlyVtxo})
+		require.NoError(t, err)
+
+		err = vtxoRepo.SpendVtxos(ctx, map[domain.Outpoint]string{
+			spentOnlyVtxo.Outpoint: spentBy,
+		}, arkTxid)
+		require.NoError(t, err)
+
+		assets, err := repo.GetAssets(ctx, []string{assetID})
+		require.NoError(t, err)
+		require.Len(t, assets, 1)
+		require.Equal(t, assetID, assets[0].Id)
+		require.Zero(t, assets[0].Supply.Sign())
+	})
+}
+
 func testFeeRepository(t *testing.T, svc ports.RepoManager) {
 	t.Run("test_fee_repository", func(t *testing.T) {
 		ctx := context.Background()