通过 Node.js 使用 ClickHouse Managed Postgres

DATABASE_URL=postgresql://postgres:<PASSWORD>@your-instance.pg.clickhouse.cloud:5432/guide_nodejs
DATABASE_CA_CERT=./ca-certificate.pem

import fs from 'node:fs';
import pg from 'pg';

export const pool = new pg.Pool({
  connectionString: process.env.DATABASE_URL,
  ssl: {
    // Trust only your instance's CA; the hostname is verified too (verify-full).
    ca: fs.readFileSync(process.env.DATABASE_CA_CERT, 'utf8'),
  },
  max: 10,
  idleTimeoutMillis: 30_000,
});

// An idle client can be disconnected by the server (for example, during a failover).
// Without this handler, the error would crash the process.
pool.on('error', (err) => {
  console.error('Idle client error:', err.message);
});

import { pool } from './db.mjs';

await pool.query(`
  CREATE TABLE IF NOT EXISTS accounts (
    id      integer GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    owner   text NOT NULL UNIQUE,
    balance numeric(12, 2) NOT NULL CHECK (balance >= 0)
  );
  CREATE TABLE IF NOT EXISTS transfers (
    id         integer GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    from_id    integer NOT NULL REFERENCES accounts (id),
    to_id      integer NOT NULL REFERENCES accounts (id),
    amount     numeric(12, 2) NOT NULL CHECK (amount > 0),
    created_at timestamptz NOT NULL DEFAULT now()
  );
`);

const accounts = [
  ['alice', 100],
  ['bob', 50],
];
for (const [owner, balance] of accounts) {
  await pool.query(
    'INSERT INTO accounts (owner, balance) VALUES ($1, $2) ON CONFLICT (owner) DO NOTHING',
    [owner, balance],
  );
}

const { rows } = await pool.query(
  'SELECT id, owner, balance FROM accounts WHERE owner = ANY($1) ORDER BY id',
  [['alice', 'bob']],
);
console.table(rows);

await pool.end();

import { pool } from './db.mjs';

// Moves money between two accounts atomically: either every statement commits, or none do.
export async function transfer(fromOwner, toOwner, amount) {
  const client = await pool.connect();
  try {
    await client.query('BEGIN');
    const debit = await client.query(
      'UPDATE accounts SET balance = balance - $1 WHERE owner = $2 AND balance >= $1 RETURNING id',
      [amount, fromOwner],
    );
    if (debit.rowCount === 0) {
      throw new Error(`Insufficient funds or unknown account: ${fromOwner}`);
    }
    const credit = await client.query(
      'UPDATE accounts SET balance = balance + $1 WHERE owner = $2 RETURNING id',
      [amount, toOwner],
    );
    if (credit.rowCount === 0) {
      throw new Error(`Unknown account: ${toOwner}`);
    }
    const { rows } = await client.query(
      'INSERT INTO transfers (from_id, to_id, amount) VALUES ($1, $2, $3) RETURNING *',
      [debit.rows[0].id, credit.rows[0].id, amount],
    );
    await client.query('COMMIT');
    return rows[0];
  } catch (err) {
    await client.query('ROLLBACK');
    throw err;
  } finally {
    client.release();
  }
}

import http from 'node:http';
import { pool } from './db.mjs';
import { transfer } from './transfer.mjs';

const port = Number(process.env.PORT ?? 3107);

function sendJson(res, status, body) {
  res.writeHead(status, { 'content-type': 'application/json' });
  res.end(`${JSON.stringify(body, null, 2)}\n`);
}

async function readJson(req) {
  let body = '';
  for await (const chunk of req) body += chunk;
  return JSON.parse(body);
}

const server = http.createServer(async (req, res) => {
  const url = new URL(req.url, `http://${req.headers.host}`);
  try {
    if (req.method === 'GET' && url.pathname === '/accounts') {
      const { rows } = await pool.query('SELECT id, owner, balance FROM accounts ORDER BY id');
      return sendJson(res, 200, rows);
    }
    if (req.method === 'GET' && url.pathname === '/transfers') {
      const limit = Number(url.searchParams.get('limit') ?? 10);
      const { rows } = await pool.query(
        `SELECT t.id, f.owner AS "from", o.owner AS "to", t.amount, t.created_at
           FROM transfers t
           JOIN accounts f ON f.id = t.from_id
           JOIN accounts o ON o.id = t.to_id
          ORDER BY t.id DESC
          LIMIT $1`,
        [limit],
      );
      return sendJson(res, 200, rows);
    }
    if (req.method === 'POST' && url.pathname === '/transfers') {
      const { from, to, amount } = await readJson(req);
      return sendJson(res, 201, await transfer(from, to, amount));
    }
    sendJson(res, 404, { error: 'Not found' });
  } catch (err) {
    sendJson(res, 400, { error: err.message });
  }
});

server.listen(port, () => console.log(`Listening on http://localhost:${port}`));

// Stop accepting requests, let in-flight ones finish, then close all pool connections.
function shutdown(signal) {
  console.log(`${signal} received, shutting down`);
  server.close(async () => {
    await pool.end();
    console.log('Pool closed');
  });
}
process.on('SIGINT', shutdown);
process.on('SIGTERM', shutdown);