fp/packages/link2cid/index.js

'use strict';

require('dotenv').config();
const express = require('express');
const bodyParser = require('body-parser');
const cors = require('cors');
const fs = require('fs');
const fsp = require('fs/promises');
const { openAsBlob } = require('node:fs');
const { rm, stat } = require('fs/promises');
const os = require('os');
const path = require('path');
const SseStream = require('ssestream').default;
const { Transform, Readable } = require('node:stream');
const { pipeline } = require('node:stream/promises');
const { differenceInSeconds } = require('date-fns');

const cidRegex = /Qm[1-9A-HJ-NP-Za-km-z]{44,}|b[A-Za-z2-7]{58,}|B[A-Z2-7]{58,}|z[1-9A-HJ-NP-Za-km-z]{48,}|F[0-9A-F]{50,}/;
const app = express();
app.use(cors());
app.use(bodyParser.json());
app.use(bodyParser.urlencoded({ extended: false }));

// environment variables
const port = process.env.PORT || 3000;
const ipfsUrl = process.env.IPFS_URL || 'http://localhost:5001';
if (!process.env.API_KEY) throw new Error('API_KEY was missing in env');
if (!process.env.PORT) throw new Error('PORT is missing in env');

async function ipfsHealthCheck() {
    const url = ipfsUrl+'/version'
    // console.log(`ipfsHealthCheck at url=${url}`)
    try {
        const res = await fetch(url, {
            method: 'GET'
        })
        const body = await res.text()
        if (!body.includes('Version')) throw new Error('response from ipfs did not contain Version')
    } catch (e) {
        console.error('failure while checking IPFS connection.')
        console.error(e)
    }
}

// greetz https://stackoverflow.com/a/51302466/1004931
async function downloadFile(url, filePath, sse) {
    console.log(`downloading url=${url} to filePath=${filePath}`);
    const res = await fetch(url);
    const fileSize = res.headers.get('content-length');
    const fileStream = fs.createWriteStream(filePath, { flags: 'wx' });

    let downloadedBytes = 0;
    const logInterval = 1 * 1024 * 1024; // 1MB in bytes

    const progressLogger = new Transform({
        transform(chunk, encoding, callback) {
            downloadedBytes += chunk.length;

            if (downloadedBytes % logInterval < chunk.length) {
                console.log(`${downloadedBytes / (1024 * 1024)} MB processed`);
                const progress = (downloadedBytes / fileSize) * 100;
                console.log(`Download Progress: ${progress.toFixed(2)}%`);
                sse.write({
                    event: 'dlProgress',
                    data: `${Math.floor(progress)}`
                });
            }

            this.push(chunk);
            callback();
        }
    });


    await pipeline(
        res.body,
        progressLogger,
        fileStream
    )

    console.log('download finished');
    
    // verify the file
    // If we don't, we get text error messages sent to kubo which gets added and it's a bad time.
    console.log(`fileSize=${fileSize}. downloadedBytes=${downloadedBytes}`);
    if (fileSize != downloadedBytes) throw new Error('downloadedBytes did not match fileSize');
}


async function healthRes(_, res) {
    const version = await getPackageVersion();
    res.json({ error: false, message: `*link2cid ${version} pisses on the floor*` });
}

async function getPackageVersion() {
    const packageJsonFile = await fsp.readFile(path.join(__dirname, 'package.json'), { encoding: 'utf-8' });
    const json = JSON.parse(packageJsonFile);
    return json.version;
}

/**
 * 
 * We use this to upload files and get progress notifications
 * 
 */
async function streamingPostFetch(
    url,
    formData,
    basename,
    sse,
    filesize
) {
    console.log(`streamingPostFetch with url=${url}, formData=${formData.get('file')}, basename=${basename}, sse=${sse}, filesize=${filesize}`);

    try {
        const res = await fetch(url, {
            method: 'POST',
            body: formData
        });

        if (!res.ok) {
            throw new Error(`HTTP error! Status-- ${res.status}`);
        }

        const reader = res.body?.getReader();
        if (!reader) {
            throw new Error('Failed to get reader from response body');
        }


        while (true) {
            const { done, value } = await reader.read();

            const chunk = new TextDecoder().decode(value);
            const lines = chunk.split('\n');
            for (const line of lines) {
                const trimmedLine = line.trim()
                if (!!trimmedLine) {
                    console.log(trimmedLine);
                    const json = JSON.parse(trimmedLine);
                    // console.log(`comparing json.Name=${json.Name} with basename=${basename}`);
                    sse.write({
                        event: 'addProgress',
                        data: `${Math.floor(json?.Size / filesize * 100)}`
                    })
                    if (json.Name === basename && json.Hash && json.Size) {
                        // this is the last chunk
                        return json;
                    }
                }
            }

            if (done) {
                throw new Error('Response reader finished before receiving a CID which indicates a failiure.');
            }

        }
    } catch (error) {
        console.error('An error occurred:', error);
        throw error;
    }
}


function authenticate(req, res, next) {
    const apiKey = req.query?.token;
    if (!apiKey) {
        const msg = `authorization 'token' was missing from query`;
        console.error(msg);
        return res.status(401).json({ error: true, message: msg });
    }
    if (apiKey !== process.env.API_KEY) {
        const msg = 'INCORRECT API_KEY (wrong token)';
        console.error(msg);
        return res.status(403).json({ error: true, message: msg });
    } else {
        next();
    }
}


async function getFormStuff(filePath) {
    const url = `${ipfsUrl}/api/v0/add?progress=false&cid-version=1&pin=true`;
    const blob = await openAsBlob(filePath);
    const basename = path.basename(filePath);
    const filesize = (await stat(filePath)).size;
    const formData = new FormData();
    return {
        url,
        blob,
        basename,
        filesize,
        formData
    }
}


/**
 * Add a file from URL to IPFS.
 * 
 * uses SSE to send progress reports as the script
 * downloads the file to disk and then does `ipfs add`
 * finally returning a CID
 * 
 * events:
 *   - heartbeat
 *   - dlProgress
 *   - addProgress
 *   - end
 */
async function addHandler(req, res) {
    console.log(`/add`)
    let url;
    const urlStr = req.query.url;
    if (!urlStr) return res.status(400).json({
        error: 'url was missing from query'
    });
    try {
        url = new URL(urlStr);
    } catch (e) {
        return res.status(400).json({
            error: e?.message
        })
    }
    const timestamp = new Date().valueOf();
    const fileName = `${timestamp}-${url.pathname.split('/').at(-1)}`;
    const destinationFilePath = path.join(os.tmpdir(), fileName);

    console.log(`fileName=${fileName}, destinationFilePath=${destinationFilePath}`);

    const sse = new SseStream(req);
    sse.pipe(res);

    let hbStartTime = new Date();
    const heartbeat = setInterval(() => {
        sse.write({
            event: 'heartbeat',
            data: `${differenceInSeconds(new Date(), hbStartTime)}`
        });
    }, 15000);

    res.on('close', () => {
        console.log('Connection closed.');
        clearTimeout(heartbeat);
        sse.unpipe(res);
    });

    console.log(`Downloading '${urlStr}' to destinationFilePath=${destinationFilePath}`);
    await downloadFile(urlStr, destinationFilePath, sse);


    sse.write({
        event: 'dlProgress',
        data: '100'
    })

    console.log(`'ipfs add' the file ${destinationFilePath}`);
    const { url: kuboUrl, blob, basename, filesize, formData } = await getFormStuff(destinationFilePath);

    formData.append('file', blob, basename);

    let cid;
    try {
        const output = await streamingPostFetch(kuboUrl, formData, basename, sse, filesize);
        console.log(`streamingPostFetch output as follows.`);
        console.log(output);
        if (!output?.Hash) throw new Error('No CID was received from remote IPFS node.');
        if (!output?.Size) throw new Error(`'ipfs add' was missing Size in its output.`);
        // if (output.Size !== filesize) throw new Error(`input and output sizes did not match. Expected output.Size ${output.Size} to equal ${filesize}.`);
        // console.log(`filesize=${filesize} output.Size=${output.Size}`);
        cid = output.Hash;

        console.log('cleanup');
        await rm(destinationFilePath);

        console.log('end SSE');
        clearTimeout(heartbeat);
    } catch (e) {
        return sse.end({
            event: 'end',
            error: true,
            message: e
        })
    }
    return sse.end({
        event: 'end',
        data: cid
    })
}

app.get('/', authenticate, healthRes);
app.get('/health', healthRes);
app.get('/add', authenticate, addHandler);

app.listen(port, async () => {
    setInterval(async () => {
        await ipfsHealthCheck()
    }, 10000)
    const version = await getPackageVersion();
    console.log(`link2cid ${version} listening on port ${port}`);
});