import type pg from 'pg'; import type { InventoryRepository } from '../domain/ports.js'; import type { SetAvailableStockCommand, StockCommand, StockItem } from '../domain/stock.js'; interface StockRow { id: string; variant_id: string; available: number; reserved: number; sold: number; incoming: number; created_at: Date; updated_at: Date; } type InventoryOperation = 'reserve' | 'release' | 'confirm' | 'set_available'; export class PgInventoryRepository implements InventoryRepository { constructor(private readonly pool: pg.Pool) {} async findByVariantId(variantId: string): Promise { const result = await this.pool.query( 'SELECT * FROM inventory_stock WHERE variant_id = $1', [variantId], ); const row = result.rows[0]; return row ? toStockItem(row) : undefined; } async setAvailable(input: SetAvailableStockCommand): Promise { const client = await this.pool.connect(); try { await client.query('BEGIN'); const result = await client.query( `INSERT INTO inventory_stock (variant_id, available) VALUES ($1, $2) ON CONFLICT (variant_id) DO UPDATE SET available = EXCLUDED.available, updated_at = now() RETURNING *`, [input.variantId, input.quantity], ); const item = rowOrThrow(result.rows[0], 'inventory_stock upsert returned no row'); await insertMovement(client, input.variantId, 'set_available', input.quantity); await client.query('COMMIT'); return item; } catch (error) { await client.query('ROLLBACK'); throw error; } finally { client.release(); } } async reserve(input: StockCommand): Promise { return this.applyAtomicOperation(input, 'reserve', (client, command) => client.query( `UPDATE inventory_stock SET available = available - $2, reserved = reserved + $2, updated_at = now() WHERE variant_id = $1 AND available >= $2 RETURNING *`, [command.variantId, command.quantity], ), ); } async release(input: StockCommand): Promise { return this.applyAtomicOperation(input, 'release', (client, command) => client.query( `UPDATE inventory_stock SET reserved = reserved - $2, available = available + $2, updated_at = now() WHERE variant_id = $1 AND reserved >= $2 RETURNING *`, [command.variantId, command.quantity], ), ); } async confirm(input: StockCommand): Promise { return this.applyAtomicOperation(input, 'confirm', (client, command) => client.query( `UPDATE inventory_stock SET reserved = reserved - $2, sold = sold + $2, updated_at = now() WHERE variant_id = $1 AND reserved >= $2 RETURNING *`, [command.variantId, command.quantity], ), ); } private async applyAtomicOperation( input: StockCommand, operation: InventoryOperation, update: (client: pg.PoolClient, input: StockCommand) => Promise>, ): Promise { const client = await this.pool.connect(); try { await client.query('BEGIN'); const result = await update(client, input); const row = result.rows[0]; if (!row) { await client.query('ROLLBACK'); return undefined; } await insertMovement(client, input.variantId, operation, input.quantity); await client.query('COMMIT'); return toStockItem(row); } catch (error) { await client.query('ROLLBACK'); throw error; } finally { client.release(); } } } async function insertMovement( client: pg.PoolClient, variantId: string, operation: InventoryOperation, quantity: number, ): Promise { await client.query( `INSERT INTO inventory_movements (variant_id, operation, quantity) VALUES ($1, $2, $3)`, [variantId, operation, quantity], ); } function rowOrThrow(row: StockRow | undefined, message: string): StockItem { if (!row) { throw new Error(message); } return toStockItem(row); } function toStockItem(row: StockRow): StockItem { return { id: row.id, variantId: row.variant_id, available: row.available, reserved: row.reserved, sold: row.sold, incoming: row.incoming, createdAt: row.created_at, updatedAt: row.updated_at, }; }