db-pgsql-migrate.ts

import * as fs from 'fs';
import * as path from 'path';
import * as crypto from 'crypto';
import { PrismaClient } from '@prisma/client';
import { Logger } from '@nestjs/common';

const prisma = new PrismaClient();
const logger = new Logger('PgsqlMigrate');

async function main() {
    const pgsqlDir = process.env.PGSQL_DIR || './pgsql';

    try {
        // Check if directory exists
        if (!fs.existsSync(pgsqlDir)) {
            throw new Error(`Directory ${pgsqlDir} does not exist`);
        }

        const files = fs.readdirSync(pgsqlDir).filter((file) => file.endsWith('.pgsql'));

        if (files.length === 0) {
            logger.warn(`No .pgsql files found in ${pgsqlDir}`);
            return;
        }

        const fileHashes = await Promise.all(files.map(async (file) => {
            const filePath = path.join(pgsqlDir, file);
            return {
                fileName: file,
                hash: await calculateSHA256(filePath),
                content: await fs.promises.readFile(filePath, 'utf-8'),
            };
        }));

        await createMigrationTable();

        const dbEntries = await getExistingMigrations();

        const filesToUpdate = fileHashes.filter((file) => {
            const dbEntry = dbEntries.find((entry) => entry.file_name === file.fileName);
            return !dbEntry || dbEntry.hash !== file.hash;
        });

        if (filesToUpdate.length > 0) {
            await updateFunctions(filesToUpdate);
            logger.log(`Updated ${filesToUpdate.length} PostgreSQL function(s).`);
        } else {
            logger.log('No PostgreSQL functions need updating.');
        }
    } catch (error) {
        logger.error('An error occurred during migration:', error.stack);
        process.exit(1);
    } finally {
        await prisma.$disconnect();
    }
}

async function createMigrationTable() {
    await prisma.$executeRaw`
        CREATE TABLE IF NOT EXISTS _migrate_pgsql (
            file_name TEXT PRIMARY KEY,
            hash TEXT NOT NULL,
            last_sql TEXT NOT NULL,
            updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
        );
    `;
}

async function getExistingMigrations() {
    return prisma.$queryRaw<{ file_name: string; hash: string }[]>`
        SELECT file_name, hash FROM _migrate_pgsql;
    `;
}

async function updateFunctions(filesToUpdate: { fileName: string; hash: string; content: string }[]) {
    await prisma.$transaction(async (tx) => {
        for (const file of filesToUpdate) {
            logger.log(`Updating ${file.fileName}...`);
            try {
                await tx.$executeRawUnsafe(file.content);

                await tx.$executeRaw`
                    INSERT INTO _migrate_pgsql (file_name, hash, last_sql)
                    VALUES (${file.fileName}, ${file.hash}, ${file.content})
                    ON CONFLICT (file_name)
                    DO UPDATE SET hash = EXCLUDED.hash, last_sql = EXCLUDED.last_sql, updated_at = CURRENT_TIMESTAMP;
                `;
            } catch (error) {
                logger.error(`Error updating ${file.fileName}:`, error.message);
                throw error; // Re-throw to trigger transaction rollback
            }
        }
    });
}

async function calculateSHA256(filePath: string): Promise<string> {
    return new Promise((resolve, reject) => {
        const hash = crypto.createHash('sha256');
        const stream = fs.createReadStream(filePath);
        stream.on('error', reject);
        stream.on('data', chunk => hash.update(chunk));
        stream.on('end', () => resolve(hash.digest('hex')));
    });
}

main().catch((e) => {
    logger.error('Unhandled error:', e);
    process.exit(1);
});