Node.js and Express source walkthrough

The runnable source below is included from the same files packaged in the notes download. Lockfiles, configuration, and the shared browser UI are included in the download.

Database

A notes request reaches this adapter after input validation. DATABASE_ENGINE chooses PostgreSQL, MariaDB, or MongoDB; DATABASE_URL comes from the app’s attachment. SQL uses parameters rather than interpolating note text, and MongoDB uses driver operations. The initialization process creates a table or unique index. Records live in the database, so replacing a web instance leaves them available to its replacement.

const engine = process.env.DATABASE_ENGINE || 'postgresql';
let db;
async function database() {
  if (db) return db;
  if (!process.env.DATABASE_URL) throw new Error('Attach a database first');
  if (engine === 'postgresql') db = new pg.Pool({ connectionString: process.env.DATABASE_URL, max: 5 });
  else if (engine === 'mariadb') db = mysql.createPool(process.env.DATABASE_URL);
  else if (engine === 'mongodb') {
    const client = new MongoClient(process.env.DATABASE_URL);
    await client.connect();
    db = client.db().collection('notes');
  } else throw new Error('Unsupported DATABASE_ENGINE');
  return db;
}
async function migrate() {
  const db = await database();
  if (engine === 'mongodb') await db.createIndex({ id: 1 }, { unique: true });
  else await db.query('CREATE TABLE IF NOT EXISTS notes (id VARCHAR(36) PRIMARY KEY, body TEXT NOT NULL)');
  console.log('Schema ready');
}
async function notes() {
  const db = await database();
  if (engine === 'mongodb') return db.find({}, { projection: { _id: 0 } }).sort({ id: 1 }).toArray();
  const result = await db.query('SELECT id, body FROM notes ORDER BY id');
  return engine === 'postgresql' ? result.rows : result[0];
}
async function addNote(body) {
  const db = await database(), note = { id: randomUUID(), body };
  if (engine === 'mongodb') await db.insertOne({ ...note });
  else await db.query(engine === 'postgresql'
    ? 'INSERT INTO notes (id, body) VALUES ($1, $2)'
    : 'INSERT INTO notes (id, body) VALUES (?, ?)', [note.id, body]);
  return note;
}

Worker

Creating an export enqueues an ID and returns immediately. This loop reads the notes, writes an export under that ID, records its status, and acknowledges the stream entry. It also reclaims jobs abandoned before acknowledgement. Retrying writes the same export key, so interruption does not create an additional export file; the snapshot can reflect newer notes if they changed before the retry. See worker recovery.

async function worker() {
  const redis = await queue(), consumer = randomUUID();
  try { await redis.xGroupCreate('exports', 'workers', '0', { MKSTREAM: true }); }
  catch (error) { if (!error.message.includes('BUSYGROUP')) throw error; }
  let stopping = false;
  process.on('SIGTERM', () => { stopping = true; });
  process.on('SIGINT', () => { stopping = true; });
  console.log('Worker ready');
  while (!stopping) {
    try {
      const claimed = await redis.xAutoClaim('exports', 'workers', consumer, 30000, '0-0', { COUNT: 1 });
      const messages = claimed.messages.length ? claimed.messages
        : (await redis.xReadGroup('workers', consumer, [{ key: 'exports', id: '>' }], { COUNT: 1, BLOCK: 5000 }))?.[0].messages || [];
      for (const message of messages) {
        const id = message.message.id;
        if (!/^[0-9a-f-]{36}$/.test(id || '')) {
          await redis.xAck('exports', 'workers', message.id);
          continue;
        }
        if (await redis.get(`export:${id}`) !== 'ready') {
          await saveExport(redis, id, JSON.stringify(await notes(), null, 2));
          await redis.set(`export:${id}`, 'ready');
        }
        await redis.xAck('exports', 'workers', message.id);
        await redis.xDel('exports', message.id);
        console.log(`Export ready: ${id}`);
      }
    } catch {
      console.error('Export failed; unacknowledged jobs retry after 30 seconds. Check attachments and schema.');
      await new Promise(resolve => setTimeout(resolve, 1000));
    }
  }
  process.exit(0);
}

Object storage

Without S3 configuration, exports are stored in Redis for the worker checkpoint. With an attached bucket, the worker writes JSON to exports/<id>.json and the web process reads that object for downloads. Both processes need their own bucket attachment. The client uses the supplied endpoint and region with path-style requests. Credentials stay on the server; a private bucket does not add authentication to the example’s public download route.

let s3;
function storage() {
  s3 ||= new S3Client({ endpoint: process.env.S3_ENDPOINT, region: process.env.S3_REGION, forcePathStyle: true });
  return s3;
}
async function saveExport(redis, id, data) {
  if (process.env.S3_BUCKET) {
    await storage().send(new PutObjectCommand({ Bucket: process.env.S3_BUCKET, Key: `exports/${id}.json`, Body: data, ContentType: 'application/json' }));
  } else await redis.set(`export:${id}:data`, data);
}
async function loadExport(redis, id) {
  if (!process.env.S3_BUCKET) return redis.get(`export:${id}:data`);
  const result = await storage().send(new GetObjectCommand({ Bucket: process.env.S3_BUCKET, Key: `exports/${id}.json` }));
  return result.Body.transformToString();
}

Complete application

The complete source and startup commands are in the notes ZIP linked above. The HTTP layer connects the three components: POST /notes validates and saves a record, GET /notes lists records, POST /exports queues work, and the export status and download routes expose the result. Follow the language’s tutorial sequence to exercise each part, then review production preparation.