claudetools/projects/dataforth-dos/datasheet-pipeline/implementation-upload/database/pull-hoffman-inventory.js

/**
 * Pull full Hoffman API inventory, diff against local DB, write three files:
 *   - _hoffman_only_sns.txt   (SNs on Hoffman not in local DB)
 *   - _local_only_sns.txt     (SNs in local DB not on Hoffman)
 *   - _pull_inventory.log     (progress and summary)
 *
 * Writes directly via fs.appendFileSync so progress survives SSH disconnects.
 * Run detached; tail the log file for progress.
 */
const fs = require('fs');
const https = require('https');
const { URL } = require('url');
const db = require('./db');

const LOG = 'C:/Shares/testdatadb/database/_pull_inventory.log';
const OUT_HOFFMAN = 'C:/Shares/testdatadb/database/_hoffman_only_sns.txt';
const OUT_LOCAL = 'C:/Shares/testdatadb/database/_local_only_sns.txt';
const CREDS_PATH = 'C:/ProgramData/dataforth-uploader/credentials.json';
const PAGE_SIZE = 1000;

function log(msg) {
    const line = `[${new Date().toISOString()}] ${msg}\n`;
    fs.appendFileSync(LOG, line);
}

function req(method, uri, headers) {
    return new Promise((resolve, reject) => {
        const u = new URL(uri);
        const r = https.request({
            hostname: u.hostname, port: u.port || 443, path: u.pathname + u.search,
            method, headers, timeout: 45000,
        }, res => {
            let data = '';
            res.on('data', c => data += c);
            res.on('end', () => { clearTimeout(hardTimer); resolve({ status: res.statusCode, body: data }); });
            res.on('error', e => { clearTimeout(hardTimer); reject(e); });
        });
        // Hard deadline — some proxies keep TCP alive but never send data. If we
        // don't hear back in 45s, destroy the request and reject.
        const hardTimer = setTimeout(() => {
            r.destroy(new Error('hard deadline 45s'));
            reject(new Error('hard deadline 45s'));
        }, 45000);
        r.on('error', e => { clearTimeout(hardTimer); reject(e); });
        r.on('timeout', () => r.destroy(new Error('socket timeout')));
        r.end();
    });
}

async function reqRetry(method, uri, headers, tries = 3) {
    let lastErr;
    for (let i = 0; i < tries; i++) {
        try { return await req(method, uri, headers); }
        catch (e) {
            lastErr = e;
            log(`  retry ${i+1}/${tries} after ${e.message}`);
            await new Promise(r => setTimeout(r, 2000 * (i + 1)));
        }
    }
    throw lastErr;
}

async function getToken(creds) {
    const form = 'grant_type=client_credentials' +
        '&client_id=' + encodeURIComponent(creds.CF_CLIENT_ID) +
        '&client_secret=' + encodeURIComponent(creds.CF_CLIENT_SECRET) +
        '&scope=' + encodeURIComponent(creds.CF_SCOPE);
    const r = await new Promise((res, rej) => {
        const u = new URL(creds.CF_TOKEN_URL);
        const rq = https.request({
            hostname: u.hostname, port: u.port || 443, path: u.pathname,
            method: 'POST',
            headers: {
                'Content-Type': 'application/x-www-form-urlencoded',
                'Content-Length': Buffer.byteLength(form),
            },
            timeout: 30000,
        }, resp => {
            let d = '';
            resp.on('data', c => d += c);
            resp.on('end', () => res({ status: resp.statusCode, body: d }));
        });
        rq.on('error', rej);
        rq.write(form);
        rq.end();
    });
    const parsed = JSON.parse(r.body);
    if (!parsed.access_token) throw new Error('token fetch failed: ' + r.status + ' ' + r.body.slice(0, 200));
    return parsed.access_token;
}

(async () => {
    try {
        fs.writeFileSync(LOG, '');
        log('START inventory pull');

        const creds = JSON.parse(fs.readFileSync(CREDS_PATH, 'utf8'));
        const token = await getToken(creds);
        log('token len=' + token.length);

        const allSns = new Set();
        let page = 1;
        let total = null;
        const t0 = Date.now();
        const skippedPages = [];
        let consecutiveFailures = 0;
        while (true) {
            const url = creds.CF_API_BASE + '/api/v1/TestReportDataFiles?page=' + page + '&pageSize=' + PAGE_SIZE;
            let r;
            try {
                r = await reqRetry('GET', url, { 'Authorization': 'Bearer ' + token });
                consecutiveFailures = 0;
            } catch (e) {
                // Skip this page on sustained failure, don't abort.
                log(`page ${page} SKIPPED after retries: ${e.message}`);
                skippedPages.push(page);
                consecutiveFailures++;
                if (consecutiveFailures >= 10) {
                    log(`FATAL: ${consecutiveFailures} consecutive page failures — aborting`);
                    break;
                }
                page++;
                continue;
            }
            if (r.status !== 200) {
                log(`page ${page} HTTP ${r.status}: ${r.body.slice(0, 300)}`);
                break;
            }
            const obj = JSON.parse(r.body);
            total = obj.TotalCount;
            for (const it of obj.Items) allSns.add(it.SerialNumber);
            if (page === 1 || page % 50 === 0 || allSns.size >= total) {
                const rate = allSns.size / Math.max(1, (Date.now() - t0) / 1000);
                const eta = Math.round((total - allSns.size) / Math.max(rate, 1));
                log(`page ${page} collected ${allSns.size}/${total} rate ${rate.toFixed(0)}/s eta ${eta}s skipped=${skippedPages.length}`);
            }
            if (obj.Items.length < PAGE_SIZE || allSns.size >= total) break;
            page++;
        }
        if (skippedPages.length > 0) {
            log(`retrying ${skippedPages.length} skipped pages with longer delay`);
            for (const p of skippedPages) {
                const url = creds.CF_API_BASE + '/api/v1/TestReportDataFiles?page=' + p + '&pageSize=' + PAGE_SIZE;
                try {
                    await new Promise(r => setTimeout(r, 3000));
                    const r2 = await reqRetry('GET', url, { 'Authorization': 'Bearer ' + token });
                    if (r2.status === 200) {
                        const obj = JSON.parse(r2.body);
                        for (const it of obj.Items) allSns.add(it.SerialNumber);
                        log(`  recovered page ${p} (+${obj.Items.length} SNs)`);
                    }
                } catch (e) {
                    log(`  page ${p} still failed: ${e.message}`);
                }
            }
        }
        log(`Hoffman inventory collected: ${allSns.size}`);

        log('querying local DB...');
        const localRows = await db.query('SELECT serial_number FROM test_records');
        const localSns = new Set(localRows.map(r => r.serial_number));
        log(`Local DB unique SNs: ${localSns.size}`);

        const hoffmanOnly = [];
        for (const s of allSns) if (!localSns.has(s)) hoffmanOnly.push(s);
        const localOnly = [];
        for (const s of localSns) if (!allSns.has(s)) localOnly.push(s);

        fs.writeFileSync(OUT_HOFFMAN, hoffmanOnly.join('\n'));
        fs.writeFileSync(OUT_LOCAL, localOnly.join('\n'));
        log(`Hoffman-only (need pull): ${hoffmanOnly.length}  -> ${OUT_HOFFMAN}`);
        log(`Local-only (not on Hoffman): ${localOnly.length}  -> ${OUT_LOCAL}`);
        log('DONE');

        await db.close();
    } catch (e) {
        log('FATAL: ' + e.message + '\n' + (e.stack || ''));
        process.exit(1);
    }
})();