fix: invalid post commit volumes format (#829)

Author: gfyrag

internal/storage/bucket/migrations/19-transactions-fill-pcv/up.sql

@@ -8,7 +8,7 @@ do $$
 		drop table if exists moves_view;
 
 		create temp table moves_view as
-		select transactions_id::numeric, public.aggregate_objects(json_build_object(accounts_address, json_build_object(asset, post_commit_volumes))::jsonb) as volumes
+		select transactions_id::numeric, public.aggregate_objects(json_build_object(accounts_address, json_build_object(asset, json_build_object('input', (post_commit_volumes).inputs, 'output', (post_commit_volumes).outputs)))::jsonb) as volumes
 		from (
 			SELECT DISTINCT ON (moves.transactions_id, accounts_address, asset) moves.transactions_id, accounts_address, asset,
 						first_value(post_commit_volumes) OVER (

internal/storage/bucket/migrations/19-transactions-fill-pcv/up_tests_after.sql

@@ -0,0 +1,10 @@
+do $$
+	declare
+		expected varchar = '{"fees": {"USD": {"input": 1, "output": 0}}, "world": {"USD": {"input": 0, "output": 100}}, "orders:0": {"USD": {"input": 100, "output": 100}}, "sellers:0": {"USD": {"input": 99, "output": 0}}}';
+	begin
+		set search_path = '{{.Schema}}';
+		assert (select post_commit_volumes::varchar from transactions where id = 0) = expected,
+			'post_commit_volumes should be equals to ' || expected || ' but was ' || (select to_jsonb(post_commit_volumes) from transactions where id = 0);
+	end;
+$$
+

internal/storage/bucket/migrations/27-fix-invalid-pcv/notes.yaml

@@ -0,0 +1 @@
+name: Fill invalid post_commit_volumes after upgrade existing txs to v2.2

internal/storage/bucket/migrations/27-fix-invalid-pcv/up.sql

@@ -0,0 +1,59 @@
+do $$
+	declare
+		_offset integer := 0;
+		_batch_size integer := 1000;
+	begin
+		set search_path = '{{ .Schema }}';
+
+		drop table if exists moves_view;
+
+		create temp table moves_view as
+		select transactions_id::numeric, public.aggregate_objects(json_build_object(accounts_address, json_build_object(asset, json_build_object('input', (post_commit_volumes).inputs, 'output', (post_commit_volumes).outputs)))::jsonb) as volumes
+		from (
+			SELECT DISTINCT ON (moves.transactions_id, accounts_address, asset)
+				moves.transactions_id,
+				accounts_address,
+				asset,
+				first_value(post_commit_volumes) OVER (
+					PARTITION BY moves.transactions_id, accounts_address, asset
+					ORDER BY seq DESC
+				) AS post_commit_volumes
+			FROM moves
+			where insertion_date < (
+				select tstamp from goose_db_version where version_id = 12
+			)
+		) moves
+		group by transactions_id;
+
+		create index moves_view_idx on moves_view(transactions_id);
+
+		if (select count(*) from moves_view) = 0 then
+			return;
+		end if;
+
+		perform pg_notify('migrations-{{ .Schema }}', 'init: ' || (select count(*) from moves_view));
+
+		loop
+			with data as (
+				select transactions_id, volumes
+				from moves_view
+				-- play better than offset/limit
+				where transactions_id >= _offset and transactions_id < _offset + _batch_size
+			)
+			update transactions
+			set post_commit_volumes = data.volumes
+			from data
+			where transactions.id = data.transactions_id;
+
+			exit when not found;
+
+			_offset = _offset + _batch_size;
+
+			perform pg_notify('migrations-{{ .Schema }}', 'continue: ' || _batch_size);
+
+			commit;
+		end loop;
+
+		drop table if exists moves_view;
+	end
+$$;