claudetools/projects/dataforth-dos/datasheet-pipeline/implementation-upload/database/upload-to-api.js

/**
 * Post-import uploader — pushes just-imported records to Dataforth's Hoffman
 * API. Called from import.js after insertBatch, and from the /api/upload
 * endpoint for individual/bulk UI pushes.
 *
 * Datasheet content is rendered in memory from the DB row via
 * render-datasheet.renderContent — no For_Web filesystem dependency.
 *
 * Credentials come from C:\ProgramData\dataforth-uploader\credentials.json
 * (ACL'd to SYSTEM + Administrators + svc_testdatadb).
 *
 * The API is idempotent — already-present records return Unchanged.
 */
const fs = require('fs');
const https = require('https');
const { URL } = require('url');
const db = require('./db');
const { renderContent } = require('./render-datasheet');

const CREDS_PATH = 'C:\\ProgramData\\dataforth-uploader\\credentials.json';
const BATCH = 100;
const TOKEN_LEEWAY_MS = 60 * 1000;
const HTTP_TIMEOUT_MS = 120 * 1000;

const RECORD_COLUMNS = 'id, log_type, model_number, serial_number, test_date, test_station, overall_result, raw_data, source_file';

let _creds = null;
function loadCreds() {
    if (_creds) return _creds;
    if (!fs.existsSync(CREDS_PATH)) {
        throw new Error(`creds file not found: ${CREDS_PATH}`);
    }
    _creds = JSON.parse(fs.readFileSync(CREDS_PATH, 'utf8'));
    for (const k of ['CF_TOKEN_URL','CF_API_BASE','CF_CLIENT_ID','CF_CLIENT_SECRET','CF_SCOPE']) {
        if (!_creds[k]) throw new Error(`${CREDS_PATH} missing field ${k}`);
    }
    return _creds;
}

let _tok = { value: null, expiresAt: 0 };

function httpPost(uri, body, headers) {
    return new Promise((resolve, reject) => {
        const u = new URL(uri);
        const req = https.request({
            hostname: u.hostname, port: u.port || 443, path: u.pathname + u.search,
            method: 'POST',
            headers: Object.assign({}, headers, {'Content-Length': Buffer.byteLength(body)}),
            timeout: HTTP_TIMEOUT_MS,
        }, res => {
            let data = '';
            res.on('data', c => data += c);
            res.on('end', () => {
                try { resolve({status: res.statusCode, body: JSON.parse(data)}); }
                catch (e) { resolve({status: res.statusCode, body: {_raw: data}}); }
            });
        });
        req.on('error', reject);
        req.on('timeout', () => req.destroy(new Error('http timeout')));
        req.write(body);
        req.end();
    });
}

async function getToken(force = false) {
    const c = loadCreds();
    if (!force && _tok.value && Date.now() < _tok.expiresAt - TOKEN_LEEWAY_MS) {
        return _tok.value;
    }
    const form = Object.entries({
        grant_type: 'client_credentials',
        client_id: c.CF_CLIENT_ID,
        client_secret: c.CF_CLIENT_SECRET,
        scope: c.CF_SCOPE,
    }).map(([k, v]) => `${encodeURIComponent(k)}=${encodeURIComponent(v)}`).join('&');
    const r = await httpPost(c.CF_TOKEN_URL, form, {'Content-Type': 'application/x-www-form-urlencoded'});
    if (r.status !== 200 || !r.body.access_token) {
        throw new Error(`token fetch failed: ${r.status} ${JSON.stringify(r.body).slice(0,200)}`);
    }
    _tok.value = r.body.access_token;
    _tok.expiresAt = Date.now() + (r.body.expires_in || 3600) * 1000;
    return _tok.value;
}

async function bulkPost(items) {
    const c = loadCreds();
    for (let attempt = 0; attempt < 2; attempt++) {
        const tok = await getToken(attempt > 0);
        try {
            const r = await httpPost(
                `${c.CF_API_BASE}/api/v1/TestReportDataFiles/bulk`,
                JSON.stringify({Items: items}),
                {'Authorization': `Bearer ${tok}`, 'Content-Type': 'application/json'},
            );
            if (r.status === 401 && attempt === 0) continue;
            return r;
        } catch (e) {
            if (attempt === 0) { await new Promise(r => setTimeout(r, 5000)); continue; }
            return {status: 0, body: {_error: e.message}};
        }
    }
}

async function stampConfirmed(items, errors) {
    const badSns = new Set();
    for (const e of (errors || [])) {
        const matches = String(e).match(/\b\d+-\d+[A-Z]?\b/gi) || [];
        for (const m of matches) badSns.add(m);
    }
    const confirmedSns = items.map(it => it.SerialNumber).filter(sn => !badSns.has(sn));
    if (confirmedSns.length === 0) return;
    try {
        const placeholders = confirmedSns.map((_, j) => `$${j + 1}`).join(',');
        await db.execute(
            `UPDATE test_records SET api_uploaded_at = NOW() WHERE serial_number IN (${placeholders})`,
            confirmedSns,
        );
    } catch (e) {
        console.error(`[API-UPLOAD] stamp failed: ${e.message}`);
    }
}

async function uploadRecords(records, result) {
    const t0 = Date.now();
    for (let i = 0; i < records.length; i += BATCH) {
        const chunk = records.slice(i, i + BATCH);
        const items = [];
        for (const r of chunk) {
            let content;
            try {
                content = renderContent(r);
            } catch (e) {
                console.error(`[API-UPLOAD] render fail ${r.serial_number}: ${e.message}`);
                result.errors++;
                continue;
            }
            if (!content) { result.skipped++; continue; }
            items.push({ SerialNumber: r.serial_number, Content: content });
        }
        if (items.length === 0) continue;

        let resp;
        try { resp = await bulkPost(items); }
        catch (e) {
            console.error(`[API-UPLOAD] batch threw: ${e.message}`);
            result.errors += items.length;
            continue;
        }
        if (resp.status !== 200) {
            console.error(`[API-UPLOAD] HTTP ${resp.status}: ${JSON.stringify(resp.body).slice(0,200)}`);
            result.errors += items.length;
            continue;
        }
        result.created   += resp.body.Created   || 0;
        result.updated   += resp.body.Updated   || 0;
        result.unchanged += resp.body.Unchanged || 0;
        result.errors    += (resp.body.Errors || []).length;

        await stampConfirmed(items, resp.body.Errors);
    }
    const elapsed = ((Date.now() - t0) / 1000).toFixed(1);
    console.log(`[API-UPLOAD] done in ${elapsed}s: created=${result.created} updated=${result.updated} unchanged=${result.unchanged} errors=${result.errors} skipped=${result.skipped}`);
}

/**
 * Post-import upload. Non-throwing — upload failure must not wedge the import flow.
 * @param {string[]} filePaths - source_file values from the just-completed import
 */
async function uploadNewRecords(filePaths) {
    const result = {created: 0, updated: 0, unchanged: 0, errors: 0, skipped: 0};
    try {
        if (!filePaths || filePaths.length === 0) return result;
        if (!fs.existsSync(CREDS_PATH)) {
            console.log(`[API-UPLOAD] credentials not configured (${CREDS_PATH}); skipping`);
            return result;
        }
        const placeholders = filePaths.map((_, i) => `$${i + 1}`).join(',');
        const records = await db.query(
            `SELECT ${RECORD_COLUMNS} FROM test_records WHERE overall_result = 'PASS' AND source_file IN (${placeholders})`,
            filePaths,
        );
        if (records.length === 0) {
            console.log('[API-UPLOAD] no records eligible for upload');
            return result;
        }
        console.log(`[API-UPLOAD] ${records.length} records to upload`);
        await uploadRecords(records, result);
    } catch (e) {
        console.error(`[API-UPLOAD] fatal: ${e.message}`);
    }
    return result;
}

/**
 * Upload records identified by serial numbers. Called by /api/upload for
 * per-record and bulk UI pushes. Throws on hard failures so the endpoint
 * can return 500.
 */
async function uploadBySerialNumbers(sns) {
    const result = {created: 0, updated: 0, unchanged: 0, errors: 0, skipped: 0};
    if (!sns || sns.length === 0) return result;
    if (!fs.existsSync(CREDS_PATH)) {
        throw new Error(`credentials not configured (${CREDS_PATH})`);
    }
    const placeholders = sns.map((_, i) => `$${i + 1}`).join(',');
    const records = await db.query(
        `SELECT ${RECORD_COLUMNS} FROM test_records WHERE serial_number IN (${placeholders}) AND overall_result = 'PASS'`,
        sns,
    );
    if (records.length === 0) return result;
    await uploadRecords(records, result);
    return result;
}

module.exports = { uploadNewRecords, uploadBySerialNumbers };