Watch-Folder Anchoring in Node.js: chokidar, Queues, and Rate-Limit Backoff

npm install chokidar

import chokidar from 'chokidar';

const watcher = chokidar.watch('./intake', {
  persistent: true,
  ignoreInitial: false,
  awaitWriteFinish: {
    stabilityThreshold: 2000,  // ms of size stability before firing
    pollInterval: 200,
  },
});

watcher.on('add', (filePath) => {
  console.log(`Detected: ${filePath}`);
});

watcher.on('error', (err) => console.error('Watcher error:', err));

import { createHash } from 'crypto';
import { createReadStream } from 'fs';

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

class Queue {
  constructor(concurrency = 3) {
    this.concurrency = concurrency;
    this.running = 0;
    this.pending = [];
  }

  add(task) {
    return new Promise((resolve, reject) => {
      this.pending.push({ task, resolve, reject });
      this.drain();
    });
  }

  drain() {
    while (this.running < this.concurrency && this.pending.length > 0) {
      const { task, resolve, reject } = this.pending.shift();
      this.running++;
      Promise.resolve()
        .then(() => task())
        .then(resolve, reject)
        .finally(() => {
          this.running--;
          this.drain();
        });
    }
  }
}

const queue = new Queue(3);

import path from 'path';

const ANCHOR_ENDPOINT = 'https://proofledger.io/api/v1/proof';
const API_KEY = process.env.ANCHOR_API_KEY;

async function anchor(filePath) {
  const digest = await sha256(filePath);
  const filename = path.basename(filePath);

  const res = await fetch(ANCHOR_ENDPOINT, {
    method: 'POST',
    headers: {
      'Authorization': `Bearer ${API_KEY}`,
      'Content-Type': 'application/json',
    },
    body: JSON.stringify({ sha256: digest, filename }),
  });

  if (res.status === 409) {
    const body = await res.json();
    console.log(`Already anchored: ${filename} (original: ${body.duplicate_of})`);
    return { duplicate: true, original: body.duplicate_of };
  }

  if (!res.ok) {
    const err = new Error(`API error: ${res.status}`);
    err.status = res.status;
    if (res.status === 429) {
      err.retryAfter = res.headers.get('Retry-After') || '5';
    }
    throw err;
  }

  const proof = await res.json();
  console.log(`Anchored: ${filename}`);
  console.log(`  Proof ID: ${proof.id}`);
  console.log(`  Certificate: ${proof.certificate_url}`);
  return proof;
}

async function anchorWithRetry(filePath, maxAttempts = 5) {
  for (let attempt = 1; attempt <= maxAttempts; attempt++) {
    try {
      return await anchor(filePath);
    } catch (err) {
      const status = err.status;

      if (status === 400 || status === 401 || status === 403) {
        console.error(`Non-retryable error ${status}: ${path.basename(filePath)}`);
        throw err;
      }

      if (status === 429) {
        const waitSec = parseInt(err.retryAfter || '5', 10);
        console.warn(`Rate limited. Waiting ${waitSec}s (attempt ${attempt}/${maxAttempts})`);
        await sleep(waitSec * 1000);
        continue;
      }

      // Network errors, 5xx — exponential backoff
      const backoff = Math.min(1000 * 2 ** attempt, 30000);
      console.warn(`Attempt ${attempt} failed: ${err.message}. Retry in ${backoff}ms`);
      await sleep(backoff);
    }
  }

  throw new Error(`All ${maxAttempts} attempts failed: ${path.basename(filePath)}`);
}

function sleep(ms) {
  return new Promise((r) => setTimeout(r, ms));
}

import chokidar from 'chokidar';
import { createHash } from 'crypto';
import { createReadStream } from 'fs';
import path from 'path';

const INTAKE_DIR = process.env.INTAKE_DIR || './intake';
const API_KEY = process.env.ANCHOR_API_KEY;

if (!API_KEY) {
  console.error('ANCHOR_API_KEY is not set');
  process.exit(1);
}

const queue = new Queue(3);

const watcher = chokidar.watch(INTAKE_DIR, {
  persistent: true,
  ignoreInitial: false,
  awaitWriteFinish: {
    stabilityThreshold: 2000,
    pollInterval: 200,
  },
});

watcher.on('add', (filePath) => {
  queue
    .add(() => anchorWithRetry(filePath))
    .catch((err) => {
      console.error(`Final failure for ${path.basename(filePath)}: ${err.message}`);
    });
});

watcher.on('error', (err) => console.error('Watcher error:', err));

console.log(`Watching ${INTAKE_DIR} for incoming files...`);

ANCHOR_API_KEY=sk_YOUR_KEY_HERE node service.mjs