Database Table Change Data Capture Streamer

Common errors

• TypeError: Invalid Version: 12.x.x

  cause Attempting to run `af-streams` on an incompatible Node.js version, typically older than v14.
  fix Update your Node.js environment to version 14 or higher. Use a Node Version Manager (nvm) for easy switching: `nvm install 18 && nvm use 18`.

• Error: Missing `timestampField` in stream configuration.

  cause The `timestampField` property, crucial for change detection, was not provided or was null in the `DatabaseStream` constructor.
  fix Add `timestampField: 'YourTimestampColumnName'` to your `DatabaseStream` configuration, pointing to a valid timestamp column in your database table.

• Error: No primary key fields defined. Provide 'primaryKeyFields' in the configuration.

  cause The `primaryKeyFields` array, required for uniquely identifying records, was empty or missing from the `DatabaseStream` configuration.
  fix Specify `primaryKeyFields: ['Column1', 'Column2']` (or a single string for a single-column primary key) in your `DatabaseStream` configuration.

• Cannot find module 'mssql' (or 'pg')

  cause The required database driver for the specified `dbType` was not installed as a peer dependency.
  fix Install the correct database driver for your setup: `npm install mssql` for MSSQL, or `npm install pg` for PostgreSQL.

Warnings

• breaking The package requires Node.js version 14 or higher. Running it on older Node.js versions will result in runtime errors due to unsupported syntax or APIs.
  Fix: Upgrade your Node.js environment to version 14 or newer. Consult your project's `engines` field in `package.json` for specific requirements.
• gotcha Database tables streamed by af-streams must have a timestamp field and a defined set of primary key fields. Incorrect or missing configuration for these fields will prevent the stream from identifying new or updated records correctly.
  Fix: Ensure your database table schema includes a timestamp field (e.g., `UpdatedAt`, `CreatedAt`) and that `timestampField` and `primaryKeyFields` are correctly specified in your `DatabaseStream` configuration.
• breaking Major version 3 introduced potential breaking changes from previous versions (v1 or v2). While specific changes aren't detailed in the provided excerpt, typically new major versions involve API changes or modified behavior that requires adaptation.
  Fix: Review the official changelog or migration guide for `af-streams` v3 on its GitHub repository to understand specific breaking changes and update your code accordingly.
• gotcha Database drivers (e.g., `mssql` or `pg`) are not direct dependencies of `af-streams` itself. You must explicitly install the appropriate driver for your chosen database type (`dbType`) for the library to function.
  Fix: Install the necessary database driver package, e.g., `npm install mssql` for MSSQL or `npm install pg` for PostgreSQL, in your project.

Install

• npm install af-streams npm
• yarn add af-streams yarn
• pnpm add af-streams pnpm

Imports

• DatabaseStream
  wrong:

  const { DatabaseStream } = require('af-streams');

  correct:

  import { DatabaseStream } from 'af-streams';

  Main class for initiating a database stream. ESM is preferred; CommonJS `require` is a common mistake in modern Node.js projects, especially with TypeScript.
• TimeWindow
  wrong:

  import TimeWindow from 'af-streams';

  correct:

  import { TimeWindow } from 'af-streams';

  One of the several named exports for stream windowing logic. Always use named imports for these utilities.
• ConsoleSender
  wrong:

  const ConsoleSender = require('af-streams').ConsoleSender;

  correct:

  import { ConsoleSender } from 'af-streams';

  A utility class for outputting stream data to the console. Follows the named import pattern.

Quickstart

This quickstart demonstrates how to set up and start a data stream from an MSSQL database table using `af-streams`, configured to output to both the console and a custom callback function. It highlights essential configuration for database connection, table specifics, and stream polling, using environment variables for sensitive credentials.

import { DatabaseStream, ConsoleSender } from 'af-streams';
import { type ConnectionOptions as MSSQLConnectionOptions } from 'mssql';

const mssqlConfig: MSSQLConnectionOptions = {
  user: process.env.DB_USER ?? 'sa',
  password: process.env.DB_PASSWORD ?? 'YourStrong@Password',
  server: process.env.DB_HOST ?? 'localhost', 
  database: process.env.DB_NAME ?? 'MyDatabase',
  options: {
    encrypt: process.env.DB_ENCRYPT === 'true', // Use for Azure SQL Database, default: false
    trustServerCertificate: process.env.DB_TRUST_SERVER_CERTIFICATE === 'true' // Change to true for local dev / self-signed certs
  }
};

async function startDatabaseStream() {
  try {
    const stream = new DatabaseStream({
      dbType: 'mssql', // or 'postgres'
      dbConfig: mssqlConfig,
      table: 'YourTableName',
      timestampField: 'UpdatedAt',
      primaryKeyFields: ['Id', 'TenantId'], // Composite primary key
      pollIntervalMs: 5000, // Poll every 5 seconds
      // Additional options like initialTimestamp, batchSize, etc.
    });

    // Example: Using the ConsoleSender to output streamed data
    stream.addSender(new ConsoleSender());

    // Example: Using a callback sender for custom processing
    stream.addSender({
      send: (data) => {
        console.log('Callback Sender received record:', data);
        // Here you could process the data, send to a message queue, etc.
      }
    });

    console.log('Starting database stream...');
    await stream.start();

    // To stop the stream after some time or on a specific event
    // setTimeout(() => {
    //   stream.stop();
    //   console.log('Database stream stopped.');
    // }, 60000); // Stop after 1 minute

  } catch (error) {
    console.error('Failed to start database stream:', error);
    process.exit(1);
  }
}

startDatabaseStream();