pokemon/scripts/sync-prices.ts

import 'dotenv/config';

import chalk from 'chalk';
import { db, ClosePool } from '../src/db/index.ts';
import { sql, inArray, eq } from 'drizzle-orm';
import { skus, processingSkus, priceHistory } from '../src/db/schema.ts';
import { toSnakeCase } from 'drizzle-orm/casing';
import * as Indexing from './indexing.ts';


function sleep(ms: number) {
  return new Promise(resolve => setTimeout(resolve, ms));
}

async function resetProcessingTable() {
    // Use sql.raw to execute the TRUNCATE TABLE statement
    await db.execute(sql.raw('TRUNCATE TABLE pokemon.processing_skus;'));
    await db.insert(processingSkus).select(db.select({skuId: skus.skuId}).from(skus));
}

async function syncPrices() {
    const batchSize = 1000;
    // const skuIndex = client.collections('skus');

    await resetProcessingTable();
    console.log(chalk.green('Processing table reset and populated with current SKUs.'));

    while (true) {
        const batch = await db.select().from(processingSkus).limit(batchSize);
        if (batch.length === 0) {
            console.log('All SKUs processed.');
            break;
        }

        const skuIds = batch.map(item => item.skuId);
        console.log(`${chalk.blue('Processing SKUs:')} ${chalk.gray(skuIds.join(', '))}`);

        const skuResponse = await fetch('https://mpgateway.tcgplayer.com/v1/pricepoints/marketprice/skus/search', {
            method: 'POST',
            headers: {
            'Content-Type': 'application/json',
            },
            body: JSON.stringify({ skuIds: skuIds }),
        });

        if (!skuResponse.ok) {
            console.error('Error fetching SKU pricing:', skuResponse.statusText);
            process.exit(1);
        }
        

        const skuData = await skuResponse.json();

        if (skuData.length !== batchSize) {
            console.error(chalk.yellow(`Expected ${batchSize} SKUs, got ${skuData.length}`));
        }

        if (skuData.length === 0) {
            console.error(chalk.red('0 SKUs, skipping DB updates.'));
            // remove skus from the 'working' processingSkus table
            await db.delete(processingSkus).where(inArray(processingSkus.skuId, skuIds));
            // be nice to the API and not send too many requests in a short time
            await sleep(200);
            continue;
        }

        const skuUpdates = skuData.map((sku: any) => { return {
            skuId: sku.skuId,
            cardId: 0,
            productId: 0,
            condition: '',
            language: '',
            variant: '',
            calculatedAt: sku.calculatedAt ? new Date(sku.calculatedAt) : null,
            highestPrice: sku.highestPrice,
            lowestPrice: sku.lowestPrice,
            marketPrice: sku.marketPrice,
            priceCount: null,
        }});
        const skuRows = await db.insert(skus).values(skuUpdates).onConflictDoUpdate({
            target: skus.skuId,
            set: {
                calculatedAt: sql.raw(`excluded.${toSnakeCase(skus.calculatedAt.name)}`),
                highestPrice: sql.raw(`excluded.${toSnakeCase(skus.highestPrice.name)}`),
                lowestPrice: sql.raw(`excluded.${toSnakeCase(skus.lowestPrice.name)}`),
                marketPrice: sql.raw(`excluded.${toSnakeCase(skus.marketPrice.name)}`),
            },
            setWhere: sql`skus.market_price is distinct from excluded.market_price`,
        }).returning();

        if (skuRows && skuRows.length > 0) {
            const skuHistory = skuRows.filter(row => row.calculatedAt != null).map(row => { return {
                skuId: row.skuId,
                calculatedAt: new Date(row.calculatedAt?.toISOString().slice(0, 10)||0),
                marketPrice: row.marketPrice,
            }});
            if (skuHistory && skuHistory.length > 0) {
                await db.insert(priceHistory).values(skuHistory).onConflictDoUpdate({
                    target: [priceHistory.skuId,priceHistory.calculatedAt],
                    set: {
                        marketPrice: sql.raw(`excluded.${toSnakeCase(skus.marketPrice.name)}`),
                    }
                });
                console.log(chalk.cyan(`${skuRows.length} history rows added.`));
            }
        }

        // remove skus from the 'working' processingSkus table
        await db.delete(processingSkus).where(inArray(processingSkus.skuId, skuIds));

        // be nice to the API and not send too many requests in a short time
        await sleep(200);
    }

}


const start = Date.now();
await syncPrices();
await Indexing.upsertSkuCollection(db);
await ClosePool();
const end = Date.now();
const duration = (end - start) / 1000;
console.log(chalk.green(`Price sync completed in ${duration.toFixed(2)} seconds.`));

export {};