Commit any set of mutations to PostgreSQL via an RPC as a transaction

commit_transaction.sql

/** @param {CommitTransactionInput} transaction_input */
create or replace function public.commit_transaction(transaction_input jsonb)
returns int
language plpgsql
as $$
declare
  _table_name text;
  /** @type {TableMutationPatch} */
  _table_mutation_patch jsonb;
  _table_info jsonb;
  _action text;
  _action_entities jsonb;
  _column_name text;
  _column_info jsonb;
  _column_value text;
  _sql text := '';
begin

for _table_name, _table_mutation_patch in
  select * from jsonb_each(transaction_input) loop
  /** @type
  {
    [column_name: string]: {
      column_default: string | null;
      typename: string;
      is_generated: boolean;
    }
  }
  */
  _table_info := (select json_object_agg(column_name, jsonb_build_object(
    'column_default', column_default,
    'typename', udt_name,
    'is_generated', is_generated = 'ALWAYS'
  )) as table_info
  from information_schema.columns where table_name = _table_name);

  for _action, _action_entities in
    select * from jsonb_each(_table_mutation_patch) loop
    -- INSERTS
    if _action = 'inserts' then
      _sql := _sql || format('insert into "public"."%s"(', _table_name);

      -- INSERTED COLUMN DEFINITIONS
      for _column_name, _column_info in
        select * from jsonb_each(_table_info) loop
        continue when (_column_info->>'is_generated')::boolean;

        _sql := _sql || format('"%s",', _column_name);
      end loop;

      -- REMOVE LAST COMMA
      _sql := (select substr(_sql, 1, length(_sql) - 1));

      _sql := _sql || ') select';

      -- INSERTED FIELDS
      for _column_name, _column_info in
        select * from jsonb_each(_table_info) loop
        continue when (_column_info->>'is_generated')::boolean;

        _column_value := case _column_info->>'typename'
          when 'text' then format('rec->>''%s''', _column_name)
          when 'uuid' then format('uuid(rec->>''%s'')', _column_name)
          else format(
            '(rec->>''%1$s'')::%2$s',
            _column_name,
            _column_info->>'typename'
          )
        end;

        -- ADD THE DEFAULT VALUE IF REQUIRED
        if _column_info->>'column_default' is not null then
          _column_value := format(
            'coalesce(%1$s, %2$s)',
            _column_value,
            _column_info->>'column_default'
          );
        end if;

        _sql := _sql || format(' %s,', _column_value);
      end loop;

      -- REMOVE LAST COMMA
      _sql := (select substr(_sql, 1, length(_sql) - 1));

      _sql := _sql || format(
        ' from jsonb_array_elements(%s::jsonb) as t(rec);',
        quote_literal(_action_entities::text)
      );

    -- UPDATES
    elsif _action = 'updates' then
      _sql := _sql || format('update "public"."%s" as orig set', _table_name);

      -- UPDATED FIELDS
      for _column_name, _column_info in
        select * from jsonb_each(_table_info) loop
        continue when _column_name = 'id'
          or (_column_info->>'is_generated')::boolean;

        _column_value := case _column_info->>'typename'
          when 'text' then format('rec->>''%s''', _column_name)
          when 'uuid' then format('uuid(rec->>''%s'')', _column_name)
          else format(
            '(rec->>''%1$s'')::%2$s',
            _column_name,
            _column_info->>'typename'
          )
        end;

        _sql := _sql || format(
          ' "%1$s" = case when (rec->''%1$s'')::text is null then "%1$s" else %2$s end,',
          _column_name,
          _column_value
        );
      end loop;

      -- REMOVE LAST COMMA
      _sql := (select substr(_sql, 1, length(_sql) - 1));

      _sql := _sql || format(
        ' from jsonb_array_elements(%s::jsonb) as t(rec) where uuid(rec->>''id'') = orig.id;',
        quote_literal(_action_entities::text)
      );

    -- DELETIONS
    elsif _action = 'deletes' then
      _sql := _sql || format(
        'delete from "public"."%1$s" as rec where rec.id in (select uuid(jsonb_array_elements_text(%2$s)));',
        _table_name,
        quote_literal(_action_entities::text)
      );

    end if;
  end loop;
end loop;

execute _sql;

return 0;
end;
$$;

commitTransaction.ts

import { PostgrestSingleResponse, SupabaseClient } from "@supabase/supabase-js";
// generated from `supabase gen types typescript`
import { Database, TablesInsert, TablesUpdate } from "./types";

export type PublicTableName = keyof Database["public"]["Tables"];

export type TableMutationPatch<TName extends PublicTableName> = Partial<{
  inserts: TablesInsert<TName>[];
  updates: TablesUpdate<TName>[];
  deletes: string[];
}>;

export type CommitTransactionInput = Partial<{
  [TName in PublicTableName]: TableMutationPatch<TName>;
}>;

export async function commitTransaction(
  supabase: SupabaseClient<SupabaseDatabase>,
  transaction: CommitTransactionInput,
): Promise<PostgrestSingleResponse<number>> {
  const res = await supabase.rpc("commit_transaction", {
    transaction_input: transaction,
  });
  return res;
}