How can I transform various sequential inserts into PostgreSQL into one transaction?

I'm implementing edit tracking for a shipment management system where changes need to be tracked across multiple related tables. Each edit to a shipment might involve changes to multiple tables in a specific order:

• deliveries (main shipment info)
• orders (related order details)
• order_quantities (product quantities in shipment)
• product_serial (serial numbers of products)
• invoices (related invoice information)

I want to capture a complete snapshot of the shipment state including all related data whenever any of these tables change. I've created an edit_history table:

CREATE TABLE edit_history (
id integer PRIMARY KEY DEFAULT nextval('edit_history_id_seq'),
tstamp timestamp without time zone DEFAULT now(),
tabname text,
operation text,
who text DEFAULT CURRENT_USER,
key jsonb,
new_val jsonb,
old_val jsonb,
transaction_id bigint
);


The idea is that edit_history stores a snapshot of data at the time and highlights changes, so they can be easily compared in the front end (using PostgREST).

And I'm using a trigger function to track changes:

CREATE OR REPLACE FUNCTION public.delivery_edit_function()
RETURNS trigger
LANGUAGE plpgsql
SECURITY DEFINER
AS $function$
DECLARE
j_new jsonb := null;
j_old jsonb := null;
j_key jsonb;
v_delivery_id bigint;
v_transaction_id bigint;
v_snapshot jsonb;
v_exists boolean;
v_changes jsonb;
BEGIN
v_transaction_id := txid_current();

-- Get delivery_id
IF TG_TABLE_NAME = 'deliveries' THEN
v_delivery_id := COALESCE(NEW.delivery_id, OLD.delivery_id);
ELSIF TG_TABLE_NAME = 'order_quantities' THEN
v_delivery_id := COALESCE(NEW.fk_delivery_id, OLD.fk_delivery_id);
ELSIF TG_TABLE_NAME = 'product_serial' THEN
SELECT DISTINCT oq.fk_delivery_id INTO v_delivery_id
FROM order_quantities oq
WHERE oq.fk_product_id = COALESCE(NEW.fk_sage_details_id, OLD.fk_sage_details_id);
END IF;

IF v_delivery_id IS NULL THEN
RETURN NULL;
END IF;

-- Check for existing operation in transaction
SELECT EXISTS (
SELECT 1 FROM edit_history
WHERE transaction_id = v_transaction_id
AND tabname = TG_TABLE_NAME
AND operation = TG_OP
AND (key->>'delivery_id')::bigint = v_delivery_id
) INTO v_exists;

IF v_exists THEN
RETURN NULL;
END IF;

-- Get existing changes in transaction
SELECT new_val INTO v_changes
FROM edit_history
WHERE transaction_id = v_transaction_id
AND (key->>'delivery_id')::bigint = v_delivery_id
ORDER BY id DESC
LIMIT 1;

-- Get or create snapshot
IF NOT EXISTS (
SELECT 1 FROM edit_history
WHERE transaction_id = v_transaction_id
AND (key->>'delivery_id')::bigint = v_delivery_id
) THEN
SELECT jsonb_build_object(
'delivery_number', d.delivery_number,
'status', d.status,
'courier', d.courier,
'order_id', o.order_id,
'sage_sales_order_number', COALESCE(o.sage_sales_order_number, 'TBC'),
'po_number', COALESCE(o.po_number, 'TBC'),
'order_comments', o.order_comments,
'invoice_number', COALESCE(i.invoice_number, 'TBC'),
'customer_id', c2.customer_id,
'shipment_date', COALESCE(d.shipment_date::text, 'TBC'),
'tracking_number', COALESCE(d.tracking_number, 'TBC'),
'shipment_origin_id', d.shipment_origin_id,
'shipment_origin', COALESCE(d.shipment_origin, 'TBC'),
'address_nickname', a.address_nickname,
'contact_name', c.contact_name,
'customer_name', COALESCE(c2.customer_name, 'TBC'),
'products', (
SELECT jsonb_agg(
jsonb_build_object(
'product_id', sd.sage_details_id,
'product_code', sd.sage_product_code,
'product_description', sd.sage_product_description,
'quantity', oq.quantity,
'serials', (
SELECT jsonb_agg(DISTINCT ps.device_serial ORDER BY ps.device_serial)
FROM product_serial ps
WHERE ps.fk_sage_details_id = sd.sage_details_id
AND EXISTS (
SELECT 1 FROM order_quantities oq2
WHERE oq2.fk_product_id = ps.fk_sage_details_id
AND oq2.fk_delivery_id = d.delivery_id
)
)
)
ORDER BY sd.sage_details_id
)
FROM order_quantities oq
JOIN sage_details sd ON sd.sage_details_id = oq.fk_product_id
WHERE oq.fk_delivery_id = d.delivery_id
)
)
FROM deliveries d
LEFT JOIN orders o ON o.order_id = d.fk_order_id
LEFT JOIN invoices i ON i.invoice_id = d.fk_invoice_id
LEFT JOIN contacts c ON c.contact_id = o.fk_contact_id
LEFT JOIN customers c2 ON c2.customer_id = c.fk_customer_id
LEFT JOIN addresses a ON a.address_id = c.fk_address_id
WHERE d.delivery_id = v_delivery_id
INTO v_snapshot;
ELSE
SELECT old_val INTO v_snapshot
FROM edit_history
WHERE transaction_id = v_transaction_id
AND (key->>'delivery_id')::bigint = v_delivery_id
ORDER BY id DESC
LIMIT 1;
END IF;

-- Handle changes by table
IF TG_TABLE_NAME = 'deliveries' THEN
IF TG_OP = 'UPDATE' THEN
SELECT jsonb_build_object(
'products', COALESCE(v_changes->'products', '[]'::jsonb)
) || (
SELECT jsonb_object_agg(key, value)
FROM jsonb_each(to_jsonb(NEW))
WHERE to_jsonb(NEW)->key IS DISTINCT FROM to_jsonb(OLD)->key
AND key != 'delivery_id'
) INTO j_new;
ELSIF TG_OP IN ('INSERT', 'DELETE') THEN
j_new := jsonb_build_object(
'products', COALESCE(v_changes->'products', '[]'::jsonb)
) || (to_jsonb(CASE WHEN TG_OP = 'INSERT' THEN NEW ELSE OLD END) - 'delivery_id');
END IF;
ELSIF TG_TABLE_NAME = 'product_serial' AND TG_OP IN ('INSERT', 'UPDATE') THEN
j_new := jsonb_build_object(
'products',
COALESCE(
v_changes->'products' ||
jsonb_build_array(
jsonb_build_object(
'product_id', NEW.fk_sage_details_id,
'serials', jsonb_build_array(NEW.device_serial)
)
),
jsonb_build_array(
jsonb_build_object(
'product_id', NEW.fk_sage_details_id,
'serials', jsonb_build_array(NEW.device_serial)
)
)
)
);
ELSIF TG_TABLE_NAME = 'order_quantities' AND TG_OP IN ('INSERT', 'UPDATE') THEN
j_new := jsonb_build_object(
'products',
COALESCE(
v_changes->'products' ||
jsonb_build_array(
jsonb_build_object(
'product_id', NEW.fk_product_id,
'quantity', NEW.quantity
)
),
jsonb_build_array(
jsonb_build_object(
'product_id', NEW.fk_product_id,
'quantity', NEW.quantity
)
)
)
);
END IF;

-- Insert history record
INSERT INTO edit_history (
transaction_id, tabname, operation, who,
key, new_val, old_val
) VALUES (
v_transaction_id, TG_TABLE_NAME, TG_OP, CURRENT_USER,
jsonb_build_object('delivery_id', v_delivery_id),
j_new,
v_snapshot
);

RETURN NULL;
EXCEPTION WHEN OTHERS THEN
RAISE WARNING 'Error in delivery_edit_function: %', SQLERRM;
RETURN NULL;
END;
$function$;


My issues are:

• Each PostgREST API call is a separate transaction, but they're logically part of one edit session. Since I am using ng-swagger-gen and using the CRUD services it generates from the database structure. Here's an example of front-end operations:

For example creating a delivery:

// 1. Create/find order
this.ordersService.getOrders({
sageSalesOrderNumber: `eq.${sageOrderNumber}`,
}).pipe(
switchMap((existingOrders) => {
if (existingOrders.length > 0) {
return of(existingOrders[0]);
} else {
return this.ordersService.postOrders({
orders: newOrder,
Prefer: 'return=representation',
})
}
}),
// 2. Create invoice using order_id
switchMap((orderResponse) => {
return this.invoicesService.postInvoices({
invoices: {
invoice_number: invoiceNumber,
fk_order_id: orderResponse.order_id
}
})
}),
// 3. Create delivery using order_id and invoice_id
switchMap(({order, invoice}) => {
return this.deliveriesService.postDeliveries({
deliveries: newDelivery
})
}),
// 4. Create product serials and quantities in parallel
switchMap(({order, invoice, delivery}) => {
const devicePromises = filledDevices.map(device => {
// Create serials first if needed
const serialPromises = serialNumbers.map(serialNumber =>
this.productSerialService.postProductSerial({
productSerial: {
device_serial: serialNumber,
fk_sage_details_id: sageDetailsId
}
})
);

return forkJoin(serialPromises).pipe(
switchMap(() => this.orderQuantitiesService.postOrderQuantities({
orderQuantities: {
fk_order_id: order.order_id,
fk_delivery_id: delivery.delivery_id,
fk_product_id: sageDetailsId,
quantity: quantity
}
}))
);
});

return forkJoin(devicePromises);
})
)


And editing a delivery:

// Creating new shipments
private postShipment(...) { /* shown above */ }

// Editing existing shipments
private editShipment(
existingOrder: Orders,
existingDelivery: Deliveries,
filledDevices: AbstractControl[]
): void {
// 1. Update invoice first
const invoiceUpdate$ = this.invoicesService
.getInvoices({
fkOrderId: `eq.${existingOrder.order_id}`,
})
.pipe(
switchMap((existingInvoices) => {
const deliveryInvoice = existingInvoices.find(
(inv) => inv.invoice_id === existingDelivery.fk_invoice_id
);

if (deliveryInvoice) {
return this.invoicesService.patchInvoices({
invoiceId: `eq.${deliveryInvoice.invoice_id}`,
invoices: {
invoice_number: formValues.invoiceNumber || 'TBC',
} as Invoices
});
} else {
return this.invoicesService.postInvoices({
invoices: {
invoice_number: formValues.invoiceNumber || 'TBC',
fk_order_id: existingOrder.order_id,
} as Invoices,
Prefer: 'return=representation',
});
}
})
);

// 2. Update order details
const orderUpdate$ = this.ordersService.patchOrders({
orderId: `eq.${existingOrder.order_id}`,
orders: {
sage_sales_order_number: formValues.salesOrder?.trim(),
po_number: formValues.purchaseOrder?.trim(),
order_comments: formValues.orderComments?.trim(),
fk_contact_id: this.selectedCustomer?.contact_id,
} as Orders
});

// 3. Delete and recreate all products
const quantitiesUpdate$ = this.orderQuantitiesService
.getOrderQuantities({
fkDeliveryId: `eq.${existingDelivery.delivery_id}`,
})
.pipe(
switchMap((quantities) => {
if (quantities.length === 0) return of(null);

const sageDetailsIds = quantities.map((q) => q.fk_product_id);

// Delete all serials first
return this.productSerialService
.getProductSerial({
fkSageDetailsId: `in.(${sageDetailsIds.join(',')})`,
})
.pipe(
switchMap((serials) => {
const deleteSerials$ = serials.length > 0
? this.productSerialService.deleteProductSerial({
productId: `in.(${serials.map((s) => s.product_id).join(',')})`,
})
: of(null);

// Then delete quantities
return deleteSerials$.pipe(
switchMap(() => this.orderQuantitiesService.deleteOrderQuantities({
fkDeliveryId: `eq.${existingDelivery.delivery_id}`,
}))
);
})
);
})
);

// 4. Run all updates in parallel then update delivery
forkJoin({
invoice: invoiceUpdate$,
order: orderUpdate$,
quantity: quantitiesUpdate$,
})
.pipe(
switchMap(() => {
return this.deliveriesService
.patchDeliveries({
deliveryId: `eq.${existingDelivery.delivery_id}`,
deliveries: {
shipment_date: formValues.shipmentDate,
courier: formValues.courier,
tracking_number: formValues.trackingNumber?.trim() || undefined,
status: formValues.deliveryStatus,
} as Deliveries,
});
})
)
.subscribe({...});
}

• Needs to capture complete snapshots including all related data
• Changes happen in a specific order (order -> invoice -> delivery -> products)
• Need to track product serials and quantities that are added in parallel

How can I track these related changes when they come as separate transactions from PostgREST?

Should I create one history record per API call, or try to group them?

Is there a better way to structure the trigger function to handle the parallel product/serial updates?

Would it be better to modify how the frontend makes API calls?

I've tried using transaction IDs and checking for existing operations, but I'm still getting multiple history records for what should be one logical edit session. Any suggestions on improving this approach would be greatly appreciated.

Continue reading...